feat(gateway): removal-refusal signal for the victim, and a node liveness gauge

CHANGELOG.md

@@ -19,7 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - http-client: a caller can bound the response body (`http_request_bounded`, `PrpcClient::with_max_response_bytes`). Nothing is bounded by default — `dstack vmm logs --lines 100000` is a legitimate multi-megabyte fetch — but every client that talks to a guest agent opts in, in the gateway and in the VMM, because a CVM is untrusted and one of them polls on a timer against the whole fleet
 
 ### Fixed
-- gateway: a node removed via `RemoveNode` silently rejoined the cluster the next time it started, because every node re-registers its own sync address on boot. Once tombstone GC is collecting, that comeback is worse than an annoyance: a stale data directory diverges from every digest, and the divergence repair's full re-exchange resurrects records whose deletes the cluster already collected. Removal now writes a durable marker — a live record, so the GC can never eat it — that every gateway's sync endpoints enforce; a removed node's envelopes are refused until an operator re-admits it with `SetNodeUrl`
+- gateway: a node removed via `RemoveNode` silently rejoined the cluster the next time it started, because every node re-registers its own sync address on boot. Once tombstone GC is collecting, that comeback is worse than an annoyance: a stale data directory diverges from every digest, and the divergence repair's full re-exchange resurrects records whose deletes the cluster already collected. Removal now writes a durable marker — a live record, so the GC can never eat it — that every gateway's sync endpoints enforce; a removed node's envelopes are refused until an operator re-admits it with `SetNodeUrl`. The refused node counts HTTP 403 sync rejections (`dstack_gateway_sync_rejected_total`), including removal lockouts and app-identity mismatches. Every gateway also exposes `dstack_gateway_node_last_seen_timestamp_seconds` per known node, so a long-offline gateway is a one-line alert instead of an ack-watermark puzzle
 - gateway: deleted KV records left a tombstone that nothing ever collected, so every deregistered CVM stayed on disk for the life of the deployment. Tombstones every peer has acknowledged are now dropped once every `tombstone_gc_writes` replicated writes (default 10000, zero disables); the trigger counts replicated writes rather than reading a clock, so nodes in a cluster collect in the same window without depending on time synchronization. A `SetTombstoneGcConfig` admin RPC stores an operator override in the KV itself, replicating one pace to every node
 - gateway: `Admin.RemoveCvm` now reports the outcome of the `inst/` tombstone alone. A failure to delete associated override or telemetry records is logged instead of failing the call, so a removal that did take effect is no longer reported as failed — which also aborted the local routing cleanup that follows it. Re-issuing a removal still sweeps up override and telemetry records orphaned by an earlier partial failure
 - sdk: `get_compose_hash` in the Python SDK mutated the dictionary it was given, stripping `docker_config` and `requirements` from it, so hashing the same manifest twice returned two different digests -- the second one for a manifest missing those blocks

dstack/gateway/src/kv/https_client.rs

@@ -156,6 +156,21 @@ pub struct HttpsClient {
     client: HyperClient,
 }
 
+/// A non-success HTTP status, kept as a typed error in the chain so a caller
+/// can react to a specific code. The sync path needs to tell an HTTP 403
+/// rejection apart from a peer that is down or broken without string-matching
+/// a formatted error message.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct HttpStatusError(pub u16);
+
+impl std::fmt::Display for HttpStatusError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "request failed with HTTP status {}", self.0)
+    }
+}
+
+impl std::error::Error for HttpStatusError {}
+
 impl HttpsClient {
     async fn post_gzipped(
         &self,
@@ -255,7 +270,7 @@ impl HttpsClient {
             .with_context(|| format!("failed to send request to {url}"))?;
 
         if !response.status().is_success() {
-            anyhow::bail!("request failed: {}", response.status());
+            return Err(HttpStatusError(response.status().as_u16()).into());
         }
 
         // Bounded like every other response: this is the bootnode GetPeers path, and
@@ -271,7 +286,7 @@ impl HttpsClient {
 
         let status = response.status();
         if !status.is_success() {
-            anyhow::bail!("request failed: {status}");
+            return Err(HttpStatusError(status.as_u16()).into());
         }
 
         let body = read_body_bounded(response.into_body()).await?;
@@ -282,7 +297,7 @@ impl HttpsClient {
     pub async fn post_bytes_no_response(&self, url: &str, body: Vec<u8>) -> Result<()> {
         let response = self.post_gzipped(url, body).await?;
         if !response.status().is_success() {
-            anyhow::bail!("request failed: {}", response.status());
+            return Err(HttpStatusError(response.status().as_u16()).into());
         }
         Ok(())
     }
@@ -514,6 +529,22 @@ mod transport_tests {
             .await
     }
 
+    /// The status code survives into the error chain as a typed value: the
+    /// sync path tells an HTTP 403 rejection apart from a peer that is down or
+    /// broken, and string-matching a Display line is not a contract.
+    #[tokio::test]
+    async fn a_refusal_status_is_readable_from_the_error_chain() {
+        let err = request(StatusCode::FORBIDDEN, Vec::new())
+            .await
+            .expect_err("403 must not read as success");
+        let status = err
+            .chain()
+            .filter_map(|cause| cause.downcast_ref::<HttpStatusError>())
+            .next()
+            .expect("the status must be present as a typed error");
+        assert_eq!(*status, HttpStatusError(403));
+    }
+
     /// A non-success status must never be decoded as a successful sync response.
     #[tokio::test]
     async fn a_server_error_is_rejected() {

dstack/gateway/src/kv/sync_service.rs

@@ -11,7 +11,7 @@ use std::sync::Arc;
 
 use anyhow::{Context, Result};
 use dstack_gateway_rpc::GetPeersResponse;
-use tracing::{info, warn};
+use tracing::{error, info, warn};
 use wavekv::{
     sync::{
         ExchangeInterface, PeerLinkStatus, SyncConfig as KvSyncConfig, SyncEnvelope, SyncManager,
@@ -23,7 +23,7 @@ use wavekv::{
 
 use crate::config::SyncConfig as GwSyncConfig;
 
-use super::https_client::{HttpsClient, HttpsClientConfig};
+use super::https_client::{HttpStatusError, HttpsClient, HttpsClientConfig};
 use super::KvStore;
 
 /// HTTP-based network transport for WaveKV sync.
@@ -40,6 +40,26 @@ pub struct HttpSyncNetwork {
 }
 
 impl HttpSyncNetwork {
+    /// Whether a peer rejected a send with HTTP 403.
+    fn was_rejected(err: &anyhow::Error) -> bool {
+        err.chain()
+            .filter_map(|cause| cause.downcast_ref::<HttpStatusError>())
+            .any(|status| status.0 == 403)
+    }
+
+    /// Count a 403 for this node's own monitoring. Sync endpoints use 403 for
+    /// both removal lockouts and app-identity mismatches, so the metric and log
+    /// deliberately report a rejection without claiming which condition caused it.
+    fn note_if_rejected(err: &anyhow::Error, peer: NodeId) {
+        if !Self::was_rejected(err) {
+            return;
+        }
+        crate::metrics::record_sync_rejected();
+        error!(
+            "peer {peer} rejected this node's sync envelope (HTTP 403); \
+             this can indicate a removal lockout or an app-identity mismatch"
+        );
+    }
     /// `my_uuid` is passed in rather than read back out of the store.
     ///
     /// Our own uuid is local configuration, not replicated state, and sourcing
@@ -105,7 +125,8 @@ impl ExchangeInterface for HttpSyncNetwork {
             .client
             .post_bytes_response(&sync_url, env.encode()?)
             .await
-            .with_context(|| format!("failed to sync to peer {peer} at {sync_url}"))?;
+            .with_context(|| format!("failed to sync to peer {peer} at {sync_url}"))
+            .inspect_err(|err| Self::note_if_rejected(err, peer))?;
 
         self.kv_store.update_peer_last_seen(peer);
         Ok(Some(SyncEnvelope::decode(&body)?))
@@ -118,7 +139,8 @@ impl ExchangeInterface for HttpSyncNetwork {
         self.client
             .post_bytes_no_response(&push_url, env.encode()?)
             .await
-            .with_context(|| format!("failed to push to peer {peer} at {push_url}"))?;
+            .with_context(|| format!("failed to push to peer {peer} at {push_url}"))
+            .inspect_err(|err| Self::note_if_rejected(err, peer))?;
         Ok(())
     }
 }
@@ -309,3 +331,45 @@ pub async fn fetch_peers_from_bootnode(
 
     Ok(())
 }
+
+#[cfg(test)]
+mod sync_rejection_tests {
+    use super::*;
+
+    fn wrapped(status: u16) -> anyhow::Error {
+        anyhow::Error::new(HttpStatusError(status)).context(
+            "failed to sync to peer 2 at https://gw2.example.com:9202/wavekv/sync/persistent",
+        )
+    }
+
+    /// Only HTTP 403 is a rejection; transport errors and other HTTP failures
+    /// must not increment the rejection counter.
+    #[test]
+    fn only_a_403_reads_as_rejected() {
+        assert!(HttpSyncNetwork::was_rejected(&wrapped(403)));
+        for status in [400u16, 401, 404, 500, 503] {
+            assert!(
+                !HttpSyncNetwork::was_rejected(&wrapped(status)),
+                "status {status} must not read as a rejection"
+            );
+        }
+        assert!(!HttpSyncNetwork::was_rejected(&anyhow::anyhow!(
+            "connection refused"
+        )));
+    }
+
+    /// The rejection must reach the node's own monitoring.
+    #[test]
+    fn a_rejection_is_counted_for_the_senders_own_monitoring() {
+        // Process-wide static: assert on the delta, never the absolute value.
+        let before = crate::metrics::sync_rejected_count();
+        HttpSyncNetwork::note_if_rejected(&wrapped(500), 2);
+        assert_eq!(
+            crate::metrics::sync_rejected_count(),
+            before,
+            "an ordinary failure must not count as a rejection"
+        );
+        HttpSyncNetwork::note_if_rejected(&wrapped(403), 2);
+        assert!(crate::metrics::sync_rejected_count() > before);
+    }
+}

dstack/gateway/src/metrics.rs

@@ -56,6 +56,7 @@ static WG_RECONFIGURE_TOTAL: AtomicU64 = AtomicU64::new(0);
 static WG_RECONFIGURE_FAILURES: AtomicU64 = AtomicU64::new(0);
 static KV_PERSIST_FAILURES: AtomicU64 = AtomicU64::new(0);
 static KV_WAL_SYNC_FAILURES: AtomicU64 = AtomicU64::new(0);
+static SYNC_REJECTED: AtomicU64 = AtomicU64::new(0);
 
 thread_local! {
     /// Set while a scrape is sampling live state.
@@ -128,6 +129,17 @@ pub(crate) fn record_kv_wal_sync_failure() {
     KV_WAL_SYNC_FAILURES.fetch_add(1, Ordering::Relaxed);
 }
 
+/// Record a sync envelope of ours that a peer rejected with HTTP 403.
+pub(crate) fn record_sync_rejected() {
+    SYNC_REJECTED.fetch_add(1, Ordering::Relaxed);
+}
+
+/// Test-only read: the counter is process-wide, so tests assert on deltas.
+#[cfg(test)]
+pub(crate) fn sync_rejected_count() -> u64 {
+    SYNC_REJECTED.load(Ordering::Relaxed)
+}
+
 /// Index of the longest prefix in `prefixes` that `key` starts with.
 ///
 /// Longest rather than first so that adding a narrower prefix later (say
@@ -163,6 +175,9 @@ pub(crate) struct Snapshot {
     pub stores: Vec<StoreSnapshot>,
     /// domain -> certificate `notAfter`, in seconds since the epoch.
     pub cert_not_after: Vec<(String, u64)>,
+    /// node id -> the latest time any gateway observed the node, in seconds
+    /// since the epoch (max across observers, future-dated reports dropped).
+    pub node_last_seen: Vec<(u32, u64)>,
 }
 
 /// One WaveKV store (`persistent` or `ephemeral`).
@@ -362,6 +377,28 @@ pub(crate) fn render(snapshot: &Snapshot) -> String {
         "",
         KV_WAL_SYNC_FAILURES.load(Ordering::Relaxed),
     );
+    counter(
+        &mut out,
+        "dstack_gateway_sync_rejected_total",
+        "Sync envelopes this node sent that a peer rejected with HTTP 403. This can indicate a removal lockout or an app-identity mismatch.",
+        "",
+        SYNC_REJECTED.load(Ordering::Relaxed),
+    );
+
+    header(
+        &mut out,
+        "dstack_gateway_node_last_seen_timestamp_seconds",
+        "Latest time any gateway observed the node, unix seconds. Replicated (max across observers): every node reports roughly the same value, so aggregate with max(), and alert on time() minus this. A node an operator removed leaves the series entirely -- absence here plus dstack_gateway_sync_rejected_total on the machine itself can indicate the removed-by-mistake picture.",
+        "gauge",
+    );
+    for (node_id, last_seen) in &snapshot.node_last_seen {
+        line(
+            &mut out,
+            "dstack_gateway_node_last_seen_timestamp_seconds",
+            &format!("{{node=\"{node_id}\"}}"),
+            *last_seen,
+        );
+    }
 
     gauge(
         &mut out,
@@ -490,6 +527,7 @@ mod tests {
                 }],
             }],
             cert_not_after: vec![("app.example.com".to_string(), 1_800_000_000)],
+            node_last_seen: vec![(2, 1_777_000_000)],
         }
     }
 
@@ -529,6 +567,7 @@ mod tests {
             "dstack_gateway_cluster_kv_keys{store=\"persistent\"} 42",
             "dstack_gateway_kv_dirty{store=\"persistent\"} 1",
             "dstack_gateway_cluster_cert_not_after_seconds{domain=\"app.example.com\"} 1800000000",
+            "dstack_gateway_node_last_seen_timestamp_seconds{node=\"2\"} 1777000000",
         ] {
             assert!(rendered.contains(expected), "missing sample: {expected}");
         }

dstack/gateway/src/web_routes/metrics.rs

@@ -49,6 +49,20 @@ fn sample(state: &State<Proxy>) -> Snapshot {
         .map(|(domain, data)| (domain, data.not_after))
         .collect();
 
+    // One pass over the node table plus one ephemeral read per node -- the
+    // cost the counting fast-path above avoids, paid here deliberately,
+    // because a per-node timestamp is the point: a node whose value stops
+    // advancing is offline, however the operator's alerting words it.
+    let node_last_seen = kv_store
+        .load_all_nodes()
+        .into_keys()
+        .filter_map(|id| {
+            kv_store
+                .get_node_latest_last_seen(id)
+                .map(|last_seen| (id, last_seen))
+        })
+        .collect();
+
     Snapshot {
         version: crate::app_version(),
         node_id: kv_store.my_node_id(),
@@ -59,6 +73,7 @@ fn sample(state: &State<Proxy>) -> Snapshot {
         accel,
         stores,
         cert_not_after,
+        node_last_seen,
     }
 }