feat(gateway): add rejected-record listing and node removal recovery APIs

dstack/gateway/rpc/proto/gateway_rpc.proto

@@ -437,6 +437,14 @@ service Admin {
   // Remove a CVM from WaveKV and the local data plane. This is an idempotent
   // operator recovery action and also works when the stored record is unreadable.
   rpc RemoveCvm(RemoveCvmRequest) returns (RemoveCvmResponse) {}
+  // List the instance records this node currently refuses to import, with the
+  // reason for each. Pairs with RemoveCvm: list the bad records, then remove
+  // the ones that should not survive.
+  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.
+  rpc RemoveNode(RemoveNodeRequest) returns (RemoveNodeResponse) {}
 
   // ==================== DNS Credential Management ====================
   // List all DNS credentials
@@ -517,6 +525,41 @@ message RemoveCvmResponse {
   bool removed_locally = 2;
 }
 
+// One instance record this node refuses to import.
+message RejectedInstanceInfo {
+  string instance_id = 1;
+  // Why the record is refused.
+  string reason = 2;
+  // "unusable": the record fails validation or its bytes no longer decode.
+  // "lost_conflict": the record lost an IP or key conflict to an older
+  // registration.
+  string rejection = 3;
+  // Whether the instance still holds state in this node's data plane. An
+  // unusable record keeps whatever the data plane already had, so removing
+  // an active instance also drops its routing.
+  bool active_locally = 4;
+}
+
+message ListRejectedInstancesResponse {
+  repeated RejectedInstanceInfo rejected = 1;
+}
+
+// Emergency operator request to remove a decommissioned gateway node.
+message RemoveNodeRequest {
+  uint32 node_id = 1;
+}
+
+// Outcome of RemoveNode. Both fields are false when the node was never known
+// (or the removal already completed), so a mistyped node_id is visible to
+// the operator instead of silently reporting success.
+message RemoveNodeResponse {
+  // Whether any of the node's records (info, status, or sync address) was
+  // live in WaveKV before the tombstones were written.
+  bool record_existed = 1;
+  // Whether the node was still in this gateway's sync peer set.
+  bool removed_from_peer_set = 2;
+}
+
 // ==================== DNS Credential Messages ====================
 
 // DNS credential information

dstack/gateway/src/admin_service.rs

@@ -15,9 +15,10 @@ use dstack_gateway_rpc::{
     GetInstanceHandshakesResponse, GetInstancePortPolicyRequest, GetInstancePortPolicyResponse,
     GetMetaResponse, GetNodeStatusesResponse, GetZtDomainRequest, GlobalConnectionsStats,
     HandshakeEntry, HostInfo, LastSeenEntry, ListCertAttestationsRequest,
-    ListCertAttestationsResponse, ListDnsCredentialsResponse, ListZtDomainsResponse,
-    NodeStatusEntry, PeerSyncStatus as ProtoPeerSyncStatus, PortAttrs as RpcPortAttrs,
-    PortPolicy as RpcPortPolicy, RemoveCvmRequest, RemoveCvmResponse, RenewCertResponse,
+    ListCertAttestationsResponse, ListDnsCredentialsResponse, ListRejectedInstancesResponse,
+    ListZtDomainsResponse, NodeStatusEntry, PeerSyncStatus as ProtoPeerSyncStatus,
+    PortAttrs as RpcPortAttrs, PortPolicy as RpcPortPolicy, RejectedInstanceInfo, RemoveCvmRequest,
+    RemoveCvmResponse, RemoveNodeRequest, RemoveNodeResponse, RenewCertResponse,
     RenewZtDomainCertRequest, RenewZtDomainCertResponse, RotateAcmeCredentialsResponse,
     SetCertbotConfigRequest, SetDefaultDnsCredentialRequest, SetInstancePortPolicyRequest,
     SetNodeStatusRequest, SetNodeUrlRequest, StatusResponse, StoreSyncStatus,
@@ -30,8 +31,8 @@ use wavekv::node::NodeStatus as WaveKvNodeStatus;
 
 use crate::{
     kv::{
-        DnsCredential, DnsProvider, GlobalCertbotConfig, NodeStatus, PortFlags, PortPolicy,
-        ZtDomainConfig,
+        import::Rejection, DnsCredential, DnsProvider, GlobalCertbotConfig, NodeStatus, PortFlags,
+        PortPolicy, ZtDomainConfig,
     },
     main_service::Proxy,
     models::PortPolicyView,
@@ -325,6 +326,37 @@ impl AdminRpc for AdminRpcHandler {
         })
     }
 
+    async fn list_rejected_instances(self) -> Result<ListRejectedInstancesResponse> {
+        let rejected = self
+            .state
+            .rejected_instances()
+            .into_iter()
+            .map(|report| RejectedInstanceInfo {
+                instance_id: report.rejected.instance_id,
+                reason: format!("{:#}", report.rejected.reason),
+                rejection: match report.rejected.rejection {
+                    Rejection::Unusable => "unusable".to_string(),
+                    Rejection::LostConflict => "lost_conflict".to_string(),
+                },
+                active_locally: report.active_locally,
+            })
+            .collect();
+        Ok(ListRejectedInstancesResponse { rejected })
+    }
+
+    async fn remove_node(self, request: RemoveNodeRequest) -> Result<RemoveNodeResponse> {
+        let removal = self.state.remove_node(request.node_id)?;
+        warn!(
+            "admin removed node {} from WaveKV and the sync peer set \
+             (record existed: {}, was a sync peer: {})",
+            request.node_id, removal.record_existed, removal.removed_from_peer_set
+        );
+        Ok(RemoveNodeResponse {
+            record_existed: removal.record_existed,
+            removed_from_peer_set: removal.removed_from_peer_set,
+        })
+    }
+
     // ==================== DNS Credential Management ====================
 
     async fn list_dns_credentials(self) -> Result<ListDnsCredentialsResponse> {
@@ -556,9 +588,13 @@ impl AdminRpc for AdminRpcHandler {
         let kv_store = self.state.kv_store();
 
         let domain = normalize_zt_domain(&request.domain)?;
-        kv_store
-            .get_zt_domain_config(&domain)
-            .context("ZT-Domain config not found")?;
+        // A corrupt config must still be deletable, so check for the record
+        // itself: get_zt_domain_config cannot tell missing from unreadable,
+        // and refusing would leave a corrupt record permanently stuck.
+        ensure!(
+            kv_store.zt_domain_config_exists(&domain),
+            "ZT-Domain config not found"
+        );
 
         // Delete config (cert data, acme, attestations are kept for historical purposes)
         kv_store.delete_zt_domain_config(&domain)?;

dstack/gateway/src/kv/import.rs

@@ -200,9 +200,9 @@ fn accept_instances_at(wg: &WgConfig, loaded: LoadedInstances, now: u64) -> Acce
     let mut instances = BTreeMap::new();
     let mut rejected: Vec<RejectedInstance> = undecodable
         .into_iter()
-        .map(|instance_id| RejectedInstance {
+        .map(|(instance_id, reason)| RejectedInstance {
             instance_id,
-            reason: anyhow::anyhow!("record does not decode"),
+            reason: anyhow::anyhow!(reason),
             rejection: Rejection::Unusable,
         })
         .collect();
@@ -250,7 +250,6 @@ fn accept_instances_at(wg: &WgConfig, loaded: LoadedInstances, now: u64) -> Acce
 mod tests {
     use super::*;
     use ipnet::Ipv4Net;
-    use std::collections::BTreeSet;
 
     /// Wall clock the tests validate against; every fixture `reg_time` below is
     /// well under it unless the test is about the future-timestamp horizon.
@@ -292,7 +291,7 @@ mod tests {
                 .into_iter()
                 .map(|(id, data)| (id.to_string(), data))
                 .collect(),
-            undecodable: BTreeSet::new(),
+            undecodable: BTreeMap::new(),
         }
     }
 
@@ -503,7 +502,9 @@ mod tests {
                 decoded: [("good".to_string(), instance("10.0.0.20", &key(1), 100))]
                     .into_iter()
                     .collect(),
-                undecodable: ["corrupt".to_string()].into_iter().collect(),
+                undecodable: [("corrupt".to_string(), "does not decode".to_string())]
+                    .into_iter()
+                    .collect(),
             },
             NOW,
         );

dstack/gateway/src/kv/mod.rs

@@ -36,12 +36,7 @@ pub use https_client::{AppIdValidator, HttpsClientConfig};
 pub use sync_service::{fetch_peers_from_bootnode, WaveKvSyncService};
 use tracing::{error, warn};
 
-use std::{
-    collections::{BTreeMap, BTreeSet},
-    net::Ipv4Addr,
-    path::Path,
-    time::Duration,
-};
+use std::{collections::BTreeMap, net::Ipv4Addr, path::Path, time::Duration};
 
 use anyhow::{Context, Result};
 
@@ -109,8 +104,11 @@ pub struct InstanceData {
 pub struct LoadedInstances {
     /// Records that decoded successfully, keyed by instance ID.
     pub decoded: BTreeMap<String, InstanceData>,
-    /// Instance IDs whose stored bytes are present but no longer decode.
-    pub undecodable: BTreeSet<String>,
+    /// Instance IDs whose stored bytes are present but no longer decode,
+    /// mapped to the decode error. Loading does not log these: the reload
+    /// path reports them on transitions, and read-only listings must stay
+    /// quiet no matter how often an operator runs them.
+    pub undecodable: BTreeMap<String, String>,
 }
 
 /// Gateway node status (stored separately for independent updates)
@@ -758,8 +756,9 @@ impl KvStore {
                     loaded.decoded.insert(instance_id.into(), data);
                 }
                 Err(err) => {
-                    error!("{err:#}");
-                    loaded.undecodable.insert(instance_id.into());
+                    loaded
+                        .undecodable
+                        .insert(instance_id.into(), format!("{err:#}"));
                 }
             }
         }
@@ -787,6 +786,35 @@ impl KvStore {
             .collect()
     }
 
+    /// Remove a gateway node from replicated state.
+    ///
+    /// Writes tombstones for the node's info, status, and sync address, and
+    /// drops this node's own last_seen observation of it. The `__peer_addr`
+    /// tombstone doubles as the cluster-wide removal signal: every gateway
+    /// prunes its sync peer set when it observes the deletion (see
+    /// [`Self::prune_removed_peers`]).
+    ///
+    /// Returns whether any of the node's persistent records was live before
+    /// the tombstones were written, so a node known only by its sync address
+    /// (registered via `SetNodeUrl` but never booted) still reports as
+    /// existing.
+    pub fn sync_remove_node(&self, node_id: NodeId) -> Result<bool> {
+        let previous = {
+            let mut persistent = self.persistent.write();
+            [
+                persistent.delete(keys::node_info(node_id))?,
+                persistent.delete(keys::node_status(node_id))?,
+                persistent.delete(keys::peer_addr(node_id))?,
+            ]
+        };
+        self.ephemeral
+            .write()
+            .delete(keys::last_seen_node(node_id, self.my_node_id))?;
+        Ok(previous
+            .into_iter()
+            .any(|entry| entry.is_some_and(|entry| !entry.is_deleted())))
+    }
+
     // ==================== Node Status Sync ====================
 
     /// Set node status (stored separately from NodeData)
@@ -946,6 +974,11 @@ impl KvStore {
         self.persistent.watch_prefix(keys::NODE_PREFIX)
     }
 
+    /// Watch for changes to replicated peer sync addresses
+    pub fn watch_peer_addrs(&self) -> watch::Receiver<()> {
+        self.persistent.watch_prefix(keys::PEER_ADDR_PREFIX)
+    }
+
     // ==================== Persistence ====================
 
     pub fn persist_if_dirty(&self) -> Result<bool> {
@@ -960,6 +993,49 @@ impl KvStore {
         Ok(())
     }
 
+    /// Drop a node from the sync peer set of both stores.
+    ///
+    /// Returns whether the persistent store still had it as a peer.
+    pub fn remove_peer(&self, peer_id: NodeId) -> Result<bool> {
+        let removed = self.persistent.write().remove_peer(peer_id)?;
+        self.ephemeral.write().remove_peer(peer_id)?;
+        Ok(removed)
+    }
+
+    /// Drop peers whose sync address has been explicitly deleted.
+    ///
+    /// A tombstoned `__peer_addr/{id}` record is the replicated signal that
+    /// an operator removed the node (see [`Self::sync_remove_node`]). 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.
+    pub fn prune_removed_peers(&self) {
+        let peer_ids: Vec<NodeId> = self
+            .persistent
+            .read()
+            .status()
+            .peers
+            .iter()
+            .map(|peer| peer.id)
+            .collect();
+        for peer_id in peer_ids {
+            // `get` filters tombstones out, so the deletion signal is only
+            // visible through the tombstone-inclusive accessor.
+            let tombstoned = self
+                .persistent
+                .read()
+                .get_including_tombstones(&keys::peer_addr(peer_id))
+                .is_some_and(|entry| entry.is_deleted());
+            if !tombstoned {
+                continue;
+            }
+            warn!("dropping removed node {peer_id} from the sync peer set");
+            if let Err(err) = self.remove_peer(peer_id) {
+                warn!("failed to remove peer {peer_id}: {err:#}");
+            }
+        }
+    }
+
     // ==================== Peer Address (in DB) ====================
 
     /// Register a node's sync URL in DB and add to peer list for sync
@@ -1100,6 +1176,19 @@ impl KvStore {
             .decode(&keys::zt_domain_config(domain))
     }
 
+    /// Whether any record — readable or not — exists for the domain's config.
+    ///
+    /// [`Self::get_zt_domain_config`] cannot distinguish a missing record
+    /// from a corrupt one; deletion must, or a corrupt record could never be
+    /// removed.
+    pub fn zt_domain_config_exists(&self, domain: &str) -> bool {
+        // `get` already excludes tombstones, so Some means a live record.
+        self.persistent
+            .read()
+            .get(&keys::zt_domain_config(domain))
+            .is_some()
+    }
+
     /// Save ZT-Domain configuration
     pub fn save_zt_domain_config(&self, config: &ZtDomainConfig) -> Result<()> {
         self.persistent