diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d346f714..527ebdef5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +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: 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/rpc/proto/gateway_rpc.proto b/dstack/gateway/rpc/proto/gateway_rpc.proto index 22fd5f766..ac3ee856b 100644 --- a/dstack/gateway/rpc/proto/gateway_rpc.proto +++ b/dstack/gateway/rpc/proto/gateway_rpc.proto @@ -474,7 +474,12 @@ service Admin { rpc SetCaa(google.protobuf.Empty) returns (google.protobuf.Empty) {} // Summary API for inspect. rpc GetMeta(google.protobuf.Empty) returns (GetMetaResponse) {} - // Set a node's sync URL - used for dynamic peer management + // Set a node's sync URL - used for dynamic peer management. Writing a + // node's address is also the explicit re-admission decision: it clears the + // durable removal marker RemoveNode leaves, so a node removed by mistake is + // brought back with this call. Re-admit a node with its old data directory + // only if it was never absent across a collected deletion; when in doubt, + // wipe the data directory first. rpc SetNodeUrl(SetNodeUrlRequest) returns (google.protobuf.Empty) {} // Set a node's status (up/down) rpc SetNodeStatus(SetNodeStatusRequest) returns (google.protobuf.Empty) {} @@ -495,7 +500,13 @@ service Admin { rpc ListRejectedInstances(google.protobuf.Empty) returns (ListRejectedInstancesResponse) {} // Remove a decommissioned gateway node from WaveKV and this node's sync peer // set. Idempotent operator recovery action; other gateways prune the node - // from their own peer sets when the removal replicates to them. + // from their own peer sets when the removal replicates to them. Also writes + // a durable removal marker that every gateway's sync endpoints enforce: the + // removed node's envelopes are refused, so it cannot re-register itself by + // restarting, and a stale copy of its data directory cannot resurrect + // records whose deletes the cluster has already collected. Removal is meant + // to be permanent; the escape hatch for a mistake is SetNodeUrl, which + // clears the marker. rpc RemoveNode(RemoveNodeRequest) returns (RemoveNodeResponse) {} // ==================== DNS Credential Management ==================== diff --git a/dstack/gateway/src/admin_service.rs b/dstack/gateway/src/admin_service.rs index 0d22428a2..c36567df9 100644 --- a/dstack/gateway/src/admin_service.rs +++ b/dstack/gateway/src/admin_service.rs @@ -189,6 +189,11 @@ impl AdminRpc for AdminRpcHandler { async fn set_node_url(self, request: SetNodeUrlRequest) -> Result<()> { let kv_store = self.state.kv_store(); + // Writing a node's address back is the explicit re-admission + // decision, so it clears the removal marker the sync lockout reads. + if kv_store.clear_peer_removed(request.id)? { + info!("cleared the removal marker for node {}", request.id); + } kv_store.register_peer_url(request.id, &request.url)?; info!("Updated peer URL: node {} -> {}", request.id, request.url); Ok(()) diff --git a/dstack/gateway/src/kv/mod.rs b/dstack/gateway/src/kv/mod.rs index cef6585a1..e0b3cea28 100644 --- a/dstack/gateway/src/kv/mod.rs +++ b/dstack/gateway/src/kv/mod.rs @@ -407,6 +407,23 @@ impl Default for GlobalCertbotConfig { } } +/// Durable record that an operator removed a node from the cluster (stored in +/// KV, synced across nodes). +/// +/// A **live** record, deliberately. The `__peer_addr` tombstone the removal +/// also writes is food for the tombstone GC, and a removal marker the +/// collector eventually eats reads as "never registered" at precisely the +/// moment the lockout matters -- after collection, when a returning node's +/// stale records have nothing left to beat them under LWW. A fact that must +/// outlive every tombstone cannot be expressed as a deletion. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +pub struct PeerRemovalRecord { + /// Unix seconds when the operator removed the node. + pub removed_at: u64, + /// The node that executed the removal. + pub removed_by: NodeId, +} + /// Tombstone GC pacing (stored in KV, synced across nodes). /// /// The pace must be one number cluster-wide: nodes collecting on different @@ -438,6 +455,7 @@ pub mod keys { pub const HANDSHAKE_PREFIX: &str = "handshake/"; pub const LAST_SEEN_NODE_PREFIX: &str = "last_seen/node/"; pub const PEER_ADDR_PREFIX: &str = "__peer_addr/"; + pub const PEER_REMOVED_PREFIX: &str = "__peer_removed/"; pub const CERT_PREFIX: &str = "cert/"; pub const DNS_CRED_PREFIX: &str = "dns_cred/"; pub const DNS_CRED_DEFAULT: &str = "dns_cred_default"; @@ -505,6 +523,10 @@ pub mod keys { format!("{PEER_ADDR_PREFIX}{node_id}") } + pub fn peer_removed(node_id: NodeId) -> String { + format!("{PEER_REMOVED_PREFIX}{node_id}") + } + // ==================== DNS Credential keys ==================== /// Key for a DNS credential @@ -1653,10 +1675,68 @@ impl KvStore { Ok(removed) } - /// Drop peers whose sync address has been explicitly deleted. + /// Mark a node as removed by an operator. + /// + /// A live record rather than a tombstone -- see [`PeerRemovalRecord`] for + /// why the durable fact cannot be the `__peer_addr` deletion itself. + pub fn mark_peer_removed(&self, node_id: NodeId) -> Result<()> { + let record = PeerRemovalRecord { + removed_at: now_secs(), + removed_by: self.my_node_id, + }; + self.persistent + .write() + .put_encoded(keys::peer_removed(node_id), &record, true)?; + Ok(()) + } + + /// Clear a node's removal marker: the explicit re-admission decision. + /// + /// Returns whether a marker was there to clear. The deletion this writes + /// is an ordinary tombstone -- once a node is re-admitted there is + /// nothing left that must be remembered forever. + pub fn clear_peer_removed(&self, node_id: NodeId) -> Result { + let previous = self + .persistent + .write() + .delete(keys::peer_removed(node_id))?; + Ok(previous.is_some_and(|entry| !entry.is_deleted())) + } + + /// Whether an operator has removed this node and nobody has re-admitted it. /// - /// A tombstoned `__peer_addr/{id}` record is the replicated signal that - /// an operator removed the node (see [`Self::sync_remove_node`]). An + /// Fails closed on a corrupt marker: refusing sync from a node whose + /// marker cannot be read is recoverable -- the operator overwrites or + /// clears it -- while admitting a removed node's full dump can resurrect + /// every record whose delete the cluster has already collected. + pub fn is_peer_removed(&self, node_id: NodeId) -> bool { + match self + .persistent + .read() + .decode_strict::(&keys::peer_removed(node_id)) + { + Ok(record) => record.is_some(), + Err(err) => { + warn!( + "the removal marker for node {node_id} is unreadable, \ + treating the node as removed: {err:#}" + ); + true + } + } + } + + /// Watch for changes to replicated removal markers + pub fn watch_peer_removed(&self) -> watch::Receiver<()> { + self.persistent.watch_prefix(keys::PEER_REMOVED_PREFIX) + } + + /// Drop peers an operator has removed. + /// + /// The durable signal is the live `__peer_removed/{id}` marker (see + /// [`Self::mark_peer_removed`]). A tombstoned `__peer_addr/{id}` record + /// counts too: it is the only signal a removal performed by an older + /// binary leaves, and it works until the tombstone GC collects it. An /// address that was never written does not count: bootstrap can add a /// peer before its address record has synced in, and such a peer must /// not be dropped for being early. @@ -1677,7 +1757,7 @@ impl KvStore { .read() .get_including_tombstones(&keys::peer_addr(peer_id)) .is_some_and(|entry| entry.is_deleted()); - if !tombstoned { + if !tombstoned && !self.is_peer_removed(peer_id) { continue; } warn!("dropping removed node {peer_id} from the sync peer set"); @@ -3798,3 +3878,94 @@ mod tombstone_gc_tests { assert!(kv.get_tombstone_gc_config().is_err()); } } + +/// Node removal markers: the durable "this identity was retired" fact that +/// the sync lockout and peer pruning read. Live records, so the tombstone GC +/// can never collect the signal out from under either. +#[cfg(test)] +mod peer_removal_tests { + use super::corruption_tests::{put_raw, test_kv}; + use super::*; + + /// The property the marker exists for: the `__peer_addr` tombstone a + /// removal writes is collected like any other, but a fact that must + /// outlive every tombstone is stored as a live record, and collection + /// cannot touch it. + #[test] + fn a_removal_marker_survives_the_collector() { + let dir = tempfile::tempdir().expect("temp dir"); + let kv = test_kv(dir.path()); + kv.mark_peer_removed(2).expect("mark"); + // Simulate the rest of the removal: the address delete leaves a + // tombstone, and a peerless store collects everything collectable. + kv.register_peer_url(2, "https://gw2.example.com:9202") + .expect("register"); + kv.remove_peer(2).expect("drop peer"); + kv.sync_remove_node(2).expect("delete records"); + kv.collect_tombstone_garbage().expect("collect"); + + assert!( + kv.persistent + .read() + .get_including_tombstones(&keys::peer_addr(2)) + .is_none(), + "the address tombstone is gone -- the state a returning node meets" + ); + assert!( + kv.is_peer_removed(2), + "the marker is what remains to say the node was removed" + ); + } + + /// The pruning judge accepts either signal: the live marker (durable), + /// or the address tombstone (what a removal by an older binary leaves, + /// until the collector eats it). A peer with neither -- early bootstrap + /// -- is left alone. + #[test] + fn a_marked_peer_is_pruned_even_when_the_address_tombstone_is_long_gone() { + let dir = tempfile::tempdir().expect("temp dir"); + let kv = KvStore::new(1, vec![2], dir.path(), None).expect("kv"); + + // No address record, no marker: an early-bootstrap peer, kept. + kv.prune_removed_peers(); + assert!(kv.peer_ids().contains(&2)); + + // The marker alone -- the post-collection state -- prunes. + kv.mark_peer_removed(2).expect("mark"); + kv.prune_removed_peers(); + assert!( + !kv.peer_ids().contains(&2), + "the marker prunes without any tombstone to read" + ); + } + + /// Re-admission is one explicit call: clearing the marker, whose own + /// deletion is an ordinary tombstone -- once a node is welcome again + /// there is nothing that must be remembered forever. + #[test] + fn clearing_the_marker_re_admits_the_node() { + let dir = tempfile::tempdir().expect("temp dir"); + let kv = test_kv(dir.path()); + + kv.mark_peer_removed(2).expect("mark"); + assert!(kv.is_peer_removed(2)); + + assert!(kv.clear_peer_removed(2).expect("clear")); + assert!(!kv.is_peer_removed(2)); + assert!( + !kv.clear_peer_removed(2).expect("clear again"), + "idempotent, and the retry reports there was nothing to clear" + ); + } + + /// Fails closed: refusing sync from a node whose marker is unreadable is + /// recoverable by overwriting the record; admitting a removed node's full + /// dump can resurrect every collected delete. + #[test] + fn a_corrupt_removal_marker_reads_as_removed() { + let dir = tempfile::tempdir().expect("temp dir"); + let kv = test_kv(dir.path()); + put_raw(&kv, &keys::peer_removed(2), b"not-messagepack"); + assert!(kv.is_peer_removed(2)); + } +} diff --git a/dstack/gateway/src/main_service.rs b/dstack/gateway/src/main_service.rs index 5288a6c30..427405c59 100644 --- a/dstack/gateway/src/main_service.rs +++ b/dstack/gateway/src/main_service.rs @@ -242,11 +242,24 @@ impl Proxy { node_id != self.config.sync.node_id, "a node cannot remove itself" ); - // Drop the peer before publishing the tombstone: the __peer_addr - // deletion wakes the peer-address watcher, whose prune would - // otherwise race this call and make the reported membership depend - // on scheduling. - let removed_from_peer_set = self.kv_store.remove_peer(node_id)?; + // Capture membership before the first write. Every replicated write + // this function makes wakes a watcher whose prune races it -- the + // removal marker wakes the marker watcher just as surely as the + // __peer_addr tombstone wakes the address watcher -- so the answer + // must come from a read taken while no such signal exists yet, or the + // reported membership depends on scheduling. A guard test drives both + // watchers at full speed against this. + let removed_from_peer_set = self.kv_store.peer_ids().contains(&node_id); + // The durable half first: the live marker is what keeps the removed + // node out after the __peer_addr tombstone below has been collected. + self.kv_store + .mark_peer_removed(node_id) + .with_context(|| format!("failed to write the removal marker for node {node_id}"))?; + // Drop the peer here rather than leaving it to the watchers, so the + // removal is complete when this call returns. Idempotent when a + // watcher's prune wins the race, which is why the return value above + // does not come from this call. + self.kv_store.remove_peer(node_id)?; let record_existed = self .kv_store .sync_remove_node(node_id) @@ -313,6 +326,21 @@ impl ProxyInner { ); // Load state from WaveKV + if kv_store.is_peer_removed(config.sync.node_id) { + // Best-effort: a removed node usually never receives its own + // marker -- the refusals the marker drives are what keep it from + // replicating here -- so this only fires when the marker slipped + // in before the lockout took effect. The reliable signal is the + // sender side reading 403 off its own sync attempts. Only a + // warning, because this copy may also be stale: re-admission + // clears the marker on the peers, which is where the lockout is + // enforced, and a node whose marker really is current gets every + // envelope refused and can do no harm either way. + warn!( + "this node's own data directory says an operator removed it from the cluster; \ + peers will refuse to sync until it is re-admitted via SetNodeUrl" + ); + } let instances = kv_store.load_all_instances(); let nodes = kv_store.load_all_nodes(); info!( @@ -1323,6 +1351,22 @@ fn start_wavekv_watch_task(proxy: Proxy) -> Result<()> { } }); + // The removal marker prunes too. It usually replicates in the same round + // as the address tombstone, but a node that was offline for the removal + // can receive the marker alone -- the tombstone may already have been + // collected everywhere else, and a live record is the one signal that + // cannot be. + let mut rx = kv_store.watch_peer_removed(); + let kv_for_removed = kv_store.clone(); + tokio::spawn(async move { + loop { + if rx.changed().await.is_err() { + break; + } + kv_for_removed.prune_removed_peers(); + } + }); + // Start periodic persistence task. // // Both this and the WAL sync below run on the blocking pool. Each ends in a diff --git a/dstack/gateway/src/main_service/tests.rs b/dstack/gateway/src/main_service/tests.rs index 5214495b2..ca5524499 100644 --- a/dstack/gateway/src/main_service/tests.rs +++ b/dstack/gateway/src/main_service/tests.rs @@ -2467,6 +2467,10 @@ async fn an_operator_can_remove_a_decommissioned_node() { assert!(removal.record_existed); assert!(removal.removed_from_peer_set); assert!(kv.get_peer_url(7).is_none()); + assert!( + kv.is_peer_removed(7), + "removal leaves the durable marker the sync lockout reads" + ); assert!(!kv.load_all_nodes().contains_key(&7)); let peers = kv.persistent().read().status().peers; assert!(!peers.iter().any(|peer| peer.id == 7)); @@ -2522,15 +2526,26 @@ async fn remove_node_reports_peer_membership_despite_a_racing_watcher() { let state = create_test_state().await; let kv = state.kv_store.clone(); - // The production watch task prunes the peer set as soon as the - // __peer_addr tombstone lands. remove_node must capture membership - // before publishing the tombstone, or the answer it returns would - // depend on which of the two gets there first. - let mut rx = kv.watch_peer_addrs(); - let kv_for_watch = kv.clone(); - let watcher = tokio::spawn(async move { - while rx.changed().await.is_ok() { - kv_for_watch.prune_removed_peers(); + // The production watch tasks prune the peer set as soon as either signal + // lands: the __peer_addr tombstone wakes the address watcher, and the + // removal marker -- written first -- wakes the marker watcher. + // remove_node must capture membership before writing anything, or the + // answer it returns would depend on which of the racers gets there + // first. Both watchers are driven here at full speed; the marker one is + // the sharper race, because the marker is remove_node's very first + // write. + let mut addr_rx = kv.watch_peer_addrs(); + let kv_for_addrs = kv.clone(); + let addr_watcher = tokio::spawn(async move { + while addr_rx.changed().await.is_ok() { + kv_for_addrs.prune_removed_peers(); + } + }); + let mut marker_rx = kv.watch_peer_removed(); + let kv_for_markers = kv.clone(); + let marker_watcher = tokio::spawn(async move { + while marker_rx.changed().await.is_ok() { + kv_for_markers.prune_removed_peers(); } }); @@ -2541,13 +2556,14 @@ async fn remove_node_reports_peer_membership_despite_a_racing_watcher() { assert!(removal.record_existed); assert!( removal.removed_from_peer_set, - "node {node_id}: membership must be captured before the tombstone publishes" + "node {node_id}: membership must be captured before the first write lands" ); let peers = kv.persistent().read().status().peers; assert!(!peers.iter().any(|peer| peer.id == node_id)); } - watcher.abort(); + addr_watcher.abort(); + marker_watcher.abort(); } #[tokio::test] diff --git a/dstack/gateway/src/web_routes/wavekv_sync.rs b/dstack/gateway/src/web_routes/wavekv_sync.rs index 368831a44..b1a00bf51 100644 --- a/dstack/gateway/src/web_routes/wavekv_sync.rs +++ b/dstack/gateway/src/web_routes/wavekv_sync.rs @@ -149,6 +149,7 @@ pub async fn sync_store( warn!("rejected sync from invalid node_id 0"); return Err(Status::BadRequest); } + refuse_removed_sender(state, env.sender_id)?; let Some(result) = wavekv_sync.handle_envelope(store, env) else { return Err(Status::NotFound); @@ -168,6 +169,23 @@ pub async fn sync_store( )) } +/// Refuse an envelope from a node an operator has removed. +/// +/// The app-identity check above proves the sender is *a* gateway of this app; +/// it cannot prove the sender is still a *member*. A removed node returning +/// with its old data directory diverges from every digest, and wavekv's +/// repair would answer with a full re-exchange that resurrects every record +/// whose delete the cluster has already collected -- so the door, not the +/// merge, is where a removed sender has to stop. Re-admission is an explicit +/// operator decision: SetNodeUrl clears the marker. +fn refuse_removed_sender(state: &Proxy, sender_id: u32) -> Result<(), Status> { + if state.kv_store().is_peer_removed(sender_id) { + warn!("refused an envelope from removed node {sender_id}"); + return Err(Status::Forbidden); + } + Ok(()) +} + /// Opportunistic push endpoint (wavekv RFC 0001 section 3.9). /// /// Entries only: the receiver merges data but never moves its ack coverage from this @@ -191,6 +209,7 @@ pub async fn push_store( warn!("rejected push from invalid node_id 0"); return Err(Status::BadRequest); } + refuse_removed_sender(state, env.sender_id)?; let Some(result) = wavekv_sync.handle_push(store, env) else { return Err(Status::NotFound); @@ -660,6 +679,43 @@ mod tests { assert_eq!(response.status(), Status::NotFound); } + /// The lockout at the door. The app-identity check proves the sender is + /// *a* gateway of this app; only the removal marker says whether it is + /// still a *member*. A removed node's envelopes are refused on both + /// routes before anything merges, and clearing the marker -- the + /// SetNodeUrl re-admission path -- opens the door again. + #[tokio::test] + async fn a_removed_nodes_envelopes_are_refused_at_the_door() { + let (client, proxy, _tmp) = serving_gateway(true).await; + register_peer(&proxy); + proxy.kv_store().mark_peer_removed(PEER).expect("mark"); + + for path in ["/wavekv/sync/persistent", "/wavekv/push/persistent"] { + let response = client + .post(path) + .body(body(&SyncEnvelope::new(PEER, peer_uuid()))) + .dispatch() + .await; + assert_eq!( + response.status(), + Status::Forbidden, + "{path} must refuse a removed sender" + ); + } + + proxy.kv_store().clear_peer_removed(PEER).expect("clear"); + let response = client + .post("/wavekv/sync/persistent") + .body(body(&SyncEnvelope::new(PEER, peer_uuid()))) + .dispatch() + .await; + assert_eq!( + response.status(), + Status::Ok, + "re-admission opens the door again" + ); + } + /// A node with synchronization disabled reports that the service is unavailable. #[tokio::test] async fn a_sync_disabled_node_answers_503() {