feat(gateway): add admin CVM removal API

Dstack-TEE · kvinwang · Aug 12, 2026 · Aug 12, 2026 · Aug 12, 2026 · Aug 12, 2026
commit f58d42f78e648732853d4840702cb2e6163f2e19
diff --git a/dstack/gateway/rpc/proto/gateway_rpc.proto b/dstack/gateway/rpc/proto/gateway_rpc.proto
@@ -434,6 +434,9 @@ service Admin {
   rpc GetGlobalConnections(google.protobuf.Empty) returns (GlobalConnectionsStats) {}
   // Get all node statuses
   rpc GetNodeStatuses(google.protobuf.Empty) returns (GetNodeStatusesResponse) {}
+  // 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 (google.protobuf.Empty) {}
 
   // ==================== DNS Credential Management ====================
   // List all DNS credentials
@@ -499,6 +502,11 @@ service Admin {
 
 // ==================== DNS Credential Messages ====================
 
+// Emergency operator request to remove one CVM's instance record.
+message RemoveCvmRequest {
+  string instance_id = 1;
+}
+
 // DNS credential information
 message DnsCredentialInfo {
   string id = 1;

diff --git a/dstack/gateway/src/admin_service.rs b/dstack/gateway/src/admin_service.rs
@@ -17,7 +17,7 @@ use dstack_gateway_rpc::{
     HandshakeEntry, HostInfo, LastSeenEntry, ListCertAttestationsRequest,
     ListCertAttestationsResponse, ListDnsCredentialsResponse, ListZtDomainsResponse,
     NodeStatusEntry, PeerSyncStatus as ProtoPeerSyncStatus, PortAttrs as RpcPortAttrs,
-    PortPolicy as RpcPortPolicy, RenewCertResponse, RenewZtDomainCertRequest,
+    PortPolicy as RpcPortPolicy, RemoveCvmRequest, RenewCertResponse, RenewZtDomainCertRequest,
     RenewZtDomainCertResponse, RotateAcmeCredentialsResponse, SetCertbotConfigRequest,
     SetDefaultDnsCredentialRequest, SetInstancePortPolicyRequest, SetNodeStatusRequest,
     SetNodeUrlRequest, StatusResponse, StoreSyncStatus, UpdateDnsCredentialRequest,
@@ -305,6 +305,22 @@ impl AdminRpc for AdminRpcHandler {
         Ok(GetNodeStatusesResponse { statuses: entries })
     }
 
+    async fn remove_cvm(self, request: RemoveCvmRequest) -> Result<()> {
+        let instance_id = request.instance_id.trim();
+        ensure!(!instance_id.is_empty(), "instance_id is required");
+        ensure!(
+            instance_id == request.instance_id,
+            "instance_id must not have leading or trailing whitespace"
+        );
+
+        let removed_locally = self.state.remove_cvm(instance_id)?;
+        warn!(
+            "Admin removed CVM {instance_id} from WaveKV and the local data plane \
+             (present locally: {removed_locally})"
+        );
+        Ok(())
+    }
+
     // ==================== DNS Credential Management ====================
 
     async fn list_dns_credentials(self) -> Result<ListDnsCredentialsResponse> {

diff --git a/dstack/gateway/src/main_service.rs b/dstack/gateway/src/main_service.rs
@@ -129,6 +129,26 @@ pub struct ProxyOptions {
 }
 
 impl Proxy {
+    /// Remove one CVM by explicit operator request.
+    ///
+    /// The tombstone is written even when this node cannot decode the stored
+    /// record or no longer has the CVM in memory. This makes the operation an
+    /// idempotent recovery path for bad replicated instance records without
+    /// exposing arbitrary raw-KV deletion.
+    pub fn remove_cvm(&self, instance_id: &str) -> Result<bool> {
+        let mut state = self.lock();
+        state
+            .kv_store
+            .sync_delete_instance(instance_id)
+            .with_context(|| format!("failed to delete CVM {instance_id} from WaveKV"))?;
+
+        let removed = state.forget_instance(instance_id).is_some();
+        if removed {
+            state.reconfigure()?;
+        }
+        Ok(removed)
+    }
+
     pub async fn new(options: ProxyOptions) -> Result<Self> {
         let (port_policy_tx, port_policy_rx) = unbounded_channel();
         let inner = ProxyInner::new(options, port_policy_tx).await?;

diff --git a/dstack/gateway/src/main_service/tests.rs b/dstack/gateway/src/main_service/tests.rs
@@ -623,6 +623,32 @@ async fn an_undecodable_record_keeps_the_instance_it_describes() {
     assert!(state.lock().state.instances.contains_key("peer-instance"));
 }
 
+#[tokio::test]
+async fn an_operator_can_remove_a_cvm_whose_kv_record_is_unreadable() {
+    let state = create_test_state().await;
+    sync_from_peer(&state, "peer-instance", "10.0.0.40", &test_pubkey("good"));
+    reload_instances_from_kv_store(&state.proxy, &state.kv_store).unwrap();
+
+    state
+        .kv_store
+        .persistent()
+        .write()
+        .put(
+            crate::kv::keys::inst("peer-instance"),
+            b"not-messagepack".to_vec(),
+        )
+        .unwrap();
+
+    assert!(state.proxy.remove_cvm("peer-instance").unwrap());
+    assert!(!state.lock().state.instances.contains_key("peer-instance"));
+    let loaded = state.kv_store.load_all_instances();
+    assert!(!loaded.decoded.contains_key("peer-instance"));
+    assert!(!loaded.undecodable.contains("peer-instance"));
+
+    // The recovery operation is safe to retry after a timeout or lost reply.
+    assert!(!state.proxy.remove_cvm("peer-instance").unwrap());
+}
+
 #[tokio::test]
 async fn an_instance_that_lost_an_ip_conflict_stops_being_routable() {
     let state = create_test_state().await;