From fde3338a82503e2d441f5efe6bbfe32970c940b9 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 23 Aug 2026 00:45:28 -0700 Subject: [PATCH 1/3] feat(gateway): observability for removal refusals and node liveness The removed-node lockout refuses with a 403 and a warn line, which observes nothing: the audience who needs to notice is not reading either node's logs. That audience is the refused node itself. A mistakenly removed node is exactly the machine whose monitoring is still attached -- and it cannot learn of its removal from local state, because the refusals are what keep its copy of the removal marker from ever replicating to it. The signal has to be read off the door slamming. So the sync client now keeps a non-success HTTP status as a typed error in the chain (`HttpStatusError`) -- the sender must tell the lockout's 403 apart from a peer that is down or broken, and string-matching a Display line is not a contract -- and the sync network counts `dstack_gateway_sync_rejected_as_removed_total` and logs the two ways out (SetNodeUrl re-admission on a surviving gateway, or decommission) whenever a peer answers 403. Nonzero has exactly one meaning: this node has been removed and is still trying to sync. The refusing side stays a warn line, deliberately: the counter that sat there answered no question the victim-side one does not answer better, and a correctly decommissioned box that is still powered shows up in the new liveness gauge's staleness anyway. Every gateway also exposes `dstack_gateway_node_last_seen_timestamp_seconds{node=...}` -- the latest time any gateway observed each known node, max across observers, future-dated reports dropped. Until now a long-offline gateway was visible only as an ack watermark that stopped advancing, which takes a PromQL puzzle to alert on; this is `time() - metric > threshold`. It also prices what an absent node costs the cluster: while a member is unreachable, every tombstone written since its last report is pinned. Tests pin the two properties that make the victim-side counter trustworthy -- only a 403 reads as a removal refusal (a peer that is down, overloaded or broken must not claim this node was removed), and the typed status survives a real TLS request's error chain -- plus the exposition line of the liveness gauge. --- CHANGELOG.md | 2 +- dstack/gateway/src/kv/https_client.rs | 39 +++++++++++- dstack/gateway/src/kv/sync_service.rs | 80 ++++++++++++++++++++++-- dstack/gateway/src/metrics.rs | 43 +++++++++++++ dstack/gateway/src/web_routes/metrics.rs | 15 +++++ 5 files changed, 171 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 527ebdef5..30f47b37e 100644 --- a/CHANGELOG.md +++ b/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 the refusals it receives (`dstack_gateway_sync_rejected_as_removed_total`) — the signal that matters, because a mistakenly removed node is the machine whose monitoring is still attached, and it cannot learn of its removal from local state: the refusals are what keep the marker from ever replicating to it. 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 diff --git a/dstack/gateway/src/kv/https_client.rs b/dstack/gateway/src/kv/https_client.rs index 7e185425a..2d345fe91 100644 --- a/dstack/gateway/src/kv/https_client.rs +++ b/dstack/gateway/src/kv/https_client.rs @@ -156,6 +156,22 @@ 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 "a peer refuses +/// this node as removed" (403) apart from a peer that is down or broken -- +/// and a removed node cannot learn that from local state, because the very +/// refusals are what keep its copy of the removal marker from ever arriving. +#[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 +271,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 +287,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 +298,7 @@ impl HttpsClient { pub async fn post_bytes_no_response(&self, url: &str, body: Vec) -> 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 +530,23 @@ mod transport_tests { .await } + /// The status code survives into the error chain as a typed value: the + /// sync path tells "a peer refuses this node as removed" (403) 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::()) + .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() { diff --git a/dstack/gateway/src/kv/sync_service.rs b/dstack/gateway/src/kv/sync_service.rs index 6b621a6af..17e34df19 100644 --- a/dstack/gateway/src/kv/sync_service.rs +++ b/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,31 @@ pub struct HttpSyncNetwork { } impl HttpSyncNetwork { + /// Whether a send failed because the peer refuses this node as removed. + fn refused_as_removed(err: &anyhow::Error) -> bool { + err.chain() + .filter_map(|cause| cause.downcast_ref::()) + .any(|status| status.0 == 403) + } + + /// Surface a 403 as what it is: this node's only line of sight on its own + /// removal. The marker replicates among the surviving members, and the + /// refusals it drives are exactly what keeps this node from ever pulling + /// a copy -- so the signal cannot come from local state, and has to be + /// read off the door slamming. Counted for the node's own monitoring + /// (`dstack_gateway_sync_rejected_as_removed_total`) and logged with the + /// two ways out. + fn note_if_refused_as_removed(err: &anyhow::Error, peer: NodeId) { + if !Self::refused_as_removed(err) { + return; + } + crate::metrics::record_sync_rejected_as_removed(); + error!( + "peer {peer} refuses this node's sync envelopes as removed (HTTP 403); \ + this node has been removed from the cluster -- re-admit it via SetNodeUrl \ + on a surviving gateway, or decommission it" + ); + } /// `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 +130,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_refused_as_removed(err, peer))?; self.kv_store.update_peer_last_seen(peer); Ok(Some(SyncEnvelope::decode(&body)?)) @@ -118,7 +144,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_refused_as_removed(err, peer))?; Ok(()) } } @@ -309,3 +336,48 @@ pub async fn fetch_peers_from_bootnode( Ok(()) } + +#[cfg(test)] +mod removal_refusal_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", + ) + } + + /// 403 is the lockout answer and nothing else on this path returns it; a + /// peer that is down, overloaded or broken must not read as "this node + /// was removed". + #[test] + fn only_a_403_reads_as_refused_as_removed() { + assert!(HttpSyncNetwork::refused_as_removed(&wrapped(403))); + for status in [400u16, 401, 404, 500, 503] { + assert!( + !HttpSyncNetwork::refused_as_removed(&wrapped(status)), + "status {status} must not read as a removal refusal" + ); + } + assert!(!HttpSyncNetwork::refused_as_removed(&anyhow::anyhow!( + "connection refused" + ))); + } + + /// The refusal must reach the node's own monitoring: a mistakenly removed + /// node is exactly the machine whose metrics are still being scraped, and + /// it cannot learn of its removal any other way. + #[test] + fn a_refusal_is_counted_for_the_victims_own_monitoring() { + // Process-wide static: assert on the delta, never the absolute value. + let before = crate::metrics::sync_rejected_as_removed_count(); + HttpSyncNetwork::note_if_refused_as_removed(&wrapped(500), 2); + assert_eq!( + crate::metrics::sync_rejected_as_removed_count(), + before, + "an ordinary failure must not claim this node was removed" + ); + HttpSyncNetwork::note_if_refused_as_removed(&wrapped(403), 2); + assert!(crate::metrics::sync_rejected_as_removed_count() > before); + } +} diff --git a/dstack/gateway/src/metrics.rs b/dstack/gateway/src/metrics.rs index 89bb0f571..beb392aae 100644 --- a/dstack/gateway/src/metrics.rs +++ b/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_AS_REMOVED: AtomicU64 = AtomicU64::new(0); thread_local! { /// Set while a scrape is sampling live state. @@ -128,6 +129,21 @@ 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 refused with HTTP 403 as +/// removed. The victim-side signal, and the primary one: a mistakenly +/// removed node is exactly the machine whose monitoring is still attached, +/// and it cannot learn of its removal from local state -- the refusals are +/// what keep the marker from ever replicating to it. +pub(crate) fn record_sync_rejected_as_removed() { + SYNC_REJECTED_AS_REMOVED.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_as_removed_count() -> u64 { + SYNC_REJECTED_AS_REMOVED.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 +179,9 @@ pub(crate) struct Snapshot { pub stores: Vec, /// 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 +381,28 @@ pub(crate) fn render(snapshot: &Snapshot) -> String { "", KV_WAL_SYNC_FAILURES.load(Ordering::Relaxed), ); + counter( + &mut out, + "dstack_gateway_sync_rejected_as_removed_total", + "Sync envelopes this node sent that a peer refused with HTTP 403 as removed. Nonzero means THIS node has been removed from the cluster and is still trying to sync: re-admit it via SetNodeUrl on a surviving gateway, or decommission it.", + "", + SYNC_REJECTED_AS_REMOVED.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_as_removed_total on the machine itself is 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 +531,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 +571,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}"); } diff --git a/dstack/gateway/src/web_routes/metrics.rs b/dstack/gateway/src/web_routes/metrics.rs index 6a6aa132e..bdad9cbcb 100644 --- a/dstack/gateway/src/web_routes/metrics.rs +++ b/dstack/gateway/src/web_routes/metrics.rs @@ -49,6 +49,20 @@ fn sample(state: &State) -> 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) -> Snapshot { accel, stores, cert_not_after, + node_last_seen, } } From a116c51c45ff28564e9ee98e602eb58f172a64a9 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 23 Aug 2026 03:28:51 -0700 Subject: [PATCH 2/3] fix(gateway): generalize sync rejection metric --- CHANGELOG.md | 2 +- dstack/gateway/src/kv/sync_service.rs | 60 ++++++++++++--------------- dstack/gateway/src/metrics.rs | 24 +++++------ 3 files changed, 37 insertions(+), 49 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 30f47b37e..1905685c7 100644 --- a/CHANGELOG.md +++ b/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`. The refused node counts the refusals it receives (`dstack_gateway_sync_rejected_as_removed_total`) — the signal that matters, because a mistakenly removed node is the machine whose monitoring is still attached, and it cannot learn of its removal from local state: the refusals are what keep the marker from ever replicating to it. 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: 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 diff --git a/dstack/gateway/src/kv/sync_service.rs b/dstack/gateway/src/kv/sync_service.rs index 17e34df19..f6ac66d9a 100644 --- a/dstack/gateway/src/kv/sync_service.rs +++ b/dstack/gateway/src/kv/sync_service.rs @@ -40,29 +40,24 @@ pub struct HttpSyncNetwork { } impl HttpSyncNetwork { - /// Whether a send failed because the peer refuses this node as removed. - fn refused_as_removed(err: &anyhow::Error) -> bool { + /// Whether a peer rejected a send with HTTP 403. + fn was_rejected(err: &anyhow::Error) -> bool { err.chain() .filter_map(|cause| cause.downcast_ref::()) .any(|status| status.0 == 403) } - /// Surface a 403 as what it is: this node's only line of sight on its own - /// removal. The marker replicates among the surviving members, and the - /// refusals it drives are exactly what keeps this node from ever pulling - /// a copy -- so the signal cannot come from local state, and has to be - /// read off the door slamming. Counted for the node's own monitoring - /// (`dstack_gateway_sync_rejected_as_removed_total`) and logged with the - /// two ways out. - fn note_if_refused_as_removed(err: &anyhow::Error, peer: NodeId) { - if !Self::refused_as_removed(err) { + /// 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_as_removed(); + crate::metrics::record_sync_rejected(); error!( - "peer {peer} refuses this node's sync envelopes as removed (HTTP 403); \ - this node has been removed from the cluster -- re-admit it via SetNodeUrl \ - on a surviving gateway, or decommission it" + "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. @@ -131,7 +126,7 @@ impl ExchangeInterface for HttpSyncNetwork { .post_bytes_response(&sync_url, env.encode()?) .await .with_context(|| format!("failed to sync to peer {peer} at {sync_url}")) - .inspect_err(|err| Self::note_if_refused_as_removed(err, peer))?; + .inspect_err(|err| Self::note_if_rejected(err, peer))?; self.kv_store.update_peer_last_seen(peer); Ok(Some(SyncEnvelope::decode(&body)?)) @@ -145,7 +140,7 @@ impl ExchangeInterface for HttpSyncNetwork { .post_bytes_no_response(&push_url, env.encode()?) .await .with_context(|| format!("failed to push to peer {peer} at {push_url}")) - .inspect_err(|err| Self::note_if_refused_as_removed(err, peer))?; + .inspect_err(|err| Self::note_if_rejected(err, peer))?; Ok(()) } } @@ -347,37 +342,34 @@ mod removal_refusal_tests { ) } - /// 403 is the lockout answer and nothing else on this path returns it; a - /// peer that is down, overloaded or broken must not read as "this node - /// was removed". + /// 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_refused_as_removed() { - assert!(HttpSyncNetwork::refused_as_removed(&wrapped(403))); + fn only_a_403_reads_as_rejected() { + assert!(HttpSyncNetwork::was_rejected(&wrapped(403))); for status in [400u16, 401, 404, 500, 503] { assert!( - !HttpSyncNetwork::refused_as_removed(&wrapped(status)), - "status {status} must not read as a removal refusal" + !HttpSyncNetwork::was_rejected(&wrapped(status)), + "status {status} must not read as a rejection" ); } - assert!(!HttpSyncNetwork::refused_as_removed(&anyhow::anyhow!( + assert!(!HttpSyncNetwork::was_rejected(&anyhow::anyhow!( "connection refused" ))); } - /// The refusal must reach the node's own monitoring: a mistakenly removed - /// node is exactly the machine whose metrics are still being scraped, and - /// it cannot learn of its removal any other way. + /// The rejection must reach the node's own monitoring. #[test] fn a_refusal_is_counted_for_the_victims_own_monitoring() { // Process-wide static: assert on the delta, never the absolute value. - let before = crate::metrics::sync_rejected_as_removed_count(); - HttpSyncNetwork::note_if_refused_as_removed(&wrapped(500), 2); + let before = crate::metrics::sync_rejected_count(); + HttpSyncNetwork::note_if_rejected(&wrapped(500), 2); assert_eq!( - crate::metrics::sync_rejected_as_removed_count(), + crate::metrics::sync_rejected_count(), before, - "an ordinary failure must not claim this node was removed" + "an ordinary failure must not count as a rejection" ); - HttpSyncNetwork::note_if_refused_as_removed(&wrapped(403), 2); - assert!(crate::metrics::sync_rejected_as_removed_count() > before); + HttpSyncNetwork::note_if_rejected(&wrapped(403), 2); + assert!(crate::metrics::sync_rejected_count() > before); } } diff --git a/dstack/gateway/src/metrics.rs b/dstack/gateway/src/metrics.rs index beb392aae..bf0421902 100644 --- a/dstack/gateway/src/metrics.rs +++ b/dstack/gateway/src/metrics.rs @@ -56,7 +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_AS_REMOVED: AtomicU64 = AtomicU64::new(0); +static SYNC_REJECTED: AtomicU64 = AtomicU64::new(0); thread_local! { /// Set while a scrape is sampling live state. @@ -129,19 +129,15 @@ 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 refused with HTTP 403 as -/// removed. The victim-side signal, and the primary one: a mistakenly -/// removed node is exactly the machine whose monitoring is still attached, -/// and it cannot learn of its removal from local state -- the refusals are -/// what keep the marker from ever replicating to it. -pub(crate) fn record_sync_rejected_as_removed() { - SYNC_REJECTED_AS_REMOVED.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_as_removed_count() -> u64 { - SYNC_REJECTED_AS_REMOVED.load(Ordering::Relaxed) +pub(crate) fn sync_rejected_count() -> u64 { + SYNC_REJECTED.load(Ordering::Relaxed) } /// Index of the longest prefix in `prefixes` that `key` starts with. @@ -383,16 +379,16 @@ pub(crate) fn render(snapshot: &Snapshot) -> String { ); counter( &mut out, - "dstack_gateway_sync_rejected_as_removed_total", - "Sync envelopes this node sent that a peer refused with HTTP 403 as removed. Nonzero means THIS node has been removed from the cluster and is still trying to sync: re-admit it via SetNodeUrl on a surviving gateway, or decommission it.", + "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_AS_REMOVED.load(Ordering::Relaxed), + 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_as_removed_total on the machine itself is the removed-by-mistake picture.", + "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 { From 51f0c399d16268144e21c7e1a789ca6590e2a08e Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sun, 23 Aug 2026 04:20:55 -0700 Subject: [PATCH 3/3] docs(gateway): describe generic sync rejections --- dstack/gateway/src/kv/https_client.rs | 12 +++++------- dstack/gateway/src/kv/sync_service.rs | 4 ++-- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/dstack/gateway/src/kv/https_client.rs b/dstack/gateway/src/kv/https_client.rs index 2d345fe91..871c3e861 100644 --- a/dstack/gateway/src/kv/https_client.rs +++ b/dstack/gateway/src/kv/https_client.rs @@ -157,10 +157,9 @@ pub struct HttpsClient { } /// 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 "a peer refuses -/// this node as removed" (403) apart from a peer that is down or broken -- -/// and a removed node cannot learn that from local state, because the very -/// refusals are what keep its copy of the removal marker from ever arriving. +/// 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); @@ -531,9 +530,8 @@ mod transport_tests { } /// The status code survives into the error chain as a typed value: the - /// sync path tells "a peer refuses this node as removed" (403) apart from - /// a peer that is down or broken, and string-matching a Display line is - /// not a contract. + /// 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()) diff --git a/dstack/gateway/src/kv/sync_service.rs b/dstack/gateway/src/kv/sync_service.rs index f6ac66d9a..a999a4fb0 100644 --- a/dstack/gateway/src/kv/sync_service.rs +++ b/dstack/gateway/src/kv/sync_service.rs @@ -333,7 +333,7 @@ pub async fn fetch_peers_from_bootnode( } #[cfg(test)] -mod removal_refusal_tests { +mod sync_rejection_tests { use super::*; fn wrapped(status: u16) -> anyhow::Error { @@ -360,7 +360,7 @@ mod removal_refusal_tests { /// The rejection must reach the node's own monitoring. #[test] - fn a_refusal_is_counted_for_the_victims_own_monitoring() { + 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);