fix(gateway): keep a removed node out until an operator re-admits it

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

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 ====================

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(())

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<bool> {
+        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::<PeerRemovalRecord>(&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));
+    }
+}

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