From 83fad8730204a9036991ded335bb92ebfb142236 Mon Sep 17 00:00:00 2001 From: Kevin Wang Date: Sat, 22 Aug 2026 21:27:58 -0700 Subject: [PATCH] fix(gateway): collect the tombstones the cluster has finished with MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Nothing in this repository ever called wavekv's tombstone collector, so every deleted KV record — one `inst/` tombstone per deregistered CVM, plus one per operator override the delete withdraws — stayed on disk for the life of the deployment, riding in every sync digest and every full re-exchange on every node. Run `collect_tombstone_garbage` on both stores, beside the periodic persist and WAL-sync tasks. Safety comes from wavekv's ack watermark, not from this trigger: a tombstone goes only once every known peer has acknowledged the delete, at which point no peer holds the live record and nothing can resurrect it. (v1's TTL API was removed in 2.0 for exactly that resurrection.) The trigger still has to land in the same window cluster-wide: the state digest covers tombstones — it must, or it could not catch a replica that lost one — so a node that has collected and one that has not report different digests while both are correct, and wavekv's divergence detector answers with a full re-exchange that reinstates the tombstone for the next cycle to drop again. Free-running or per-node timers pay that forever. So collect once every `tombstone_gc_writes` replicated writes (default 10000, zero disables), per store. The trigger is a pure function of the replicated write count — the sum of every origin's ack watermark — which converges on every node within a sync round, so the cluster crosses each boundary together without any node reading a clock; a wall-clock-aligned tick would hold only while every host's clock stayed stepped-together. The cadence also follows write volume, which is what produces tombstones in the first place. The pace must be one number cluster-wide, so an operator override lives in the KV itself (`SetTombstoneGcConfig`/`GetTombstoneGcConfig` admin RPCs, `global/tombstone_gc_config`): it replicates to every node and takes precedence over the per-node config-file default. A corrupt override fails closed — the round is skipped rather than collected on the local default, which would be exactly the phase drift the shared pace exists to prevent. A node that has not collected since the task started treats the next check as due, so a single-node gateway — which has nobody to resurrect from, and where nothing else would ever collect — and a store that never writes again both shed the backlog they already hold. --- CHANGELOG.md | 1 + dstack/gateway/gateway.toml | 9 + dstack/gateway/rpc/proto/gateway_rpc.proto | 25 ++ dstack/gateway/src/admin_service.rs | 37 +- dstack/gateway/src/config.rs | 21 + dstack/gateway/src/kv/mod.rs | 489 ++++++++++++++++++++- dstack/gateway/src/main_service.rs | 141 +++++- dstack/gateway/src/main_service/tests.rs | 36 ++ 8 files changed, 750 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f24ca6508..3d346f714 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: 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 - sdk: the Go compose-hash helper HTML-escaped `<`, `>` and `&`, so an app-compose carrying an `os_version` bound such as `">=0.6.1"` hashed differently in Go than in Rust, Python and JavaScript. Any digest Go produced for such a manifest was wrong, and that digest is what gets whitelisted on chain diff --git a/dstack/gateway/gateway.toml b/dstack/gateway/gateway.toml index f367e52fd..4f963b4ff 100644 --- a/dstack/gateway/gateway.toml +++ b/dstack/gateway/gateway.toml @@ -215,3 +215,12 @@ wal_sync_window = "5s" sync_connections_enabled = true # Interval for syncing instance connections to KV store sync_connections_interval = "30s" +# Collect tombstones every peer has already acknowledged once every this many +# replicated writes, per store. Zero keeps every delete on disk for the life of +# the deployment. The trigger counts replicated writes rather than reading a +# clock so that every node in a cluster collects in the same window -- the +# state digest covers tombstones, so nodes collecting on their own phase +# report a mismatch and pay a full re-exchange. An operator override stored in +# the KV itself (SetTombstoneGcConfig) replicates everywhere and takes +# precedence over this per-node default. +tombstone_gc_writes = 10000 diff --git a/dstack/gateway/rpc/proto/gateway_rpc.proto b/dstack/gateway/rpc/proto/gateway_rpc.proto index 2de2c6db5..22fd5f766 100644 --- a/dstack/gateway/rpc/proto/gateway_rpc.proto +++ b/dstack/gateway/rpc/proto/gateway_rpc.proto @@ -547,6 +547,14 @@ service Admin { // to the new account; it does not deactivate the old ACME account at the CA. rpc RotateAcmeCredentials(google.protobuf.Empty) returns (RotateAcmeCredentialsResponse) {} + // ==================== Tombstone GC Configuration ==================== + // Get the effective tombstone GC pacing on this node + rpc GetTombstoneGcConfig(google.protobuf.Empty) returns (TombstoneGcConfigResponse) {} + // Store the tombstone GC pacing in the replicated KV, overriding every + // node's config-file default so the whole cluster collects on the same + // write-count boundaries. Zero disables collection cluster-wide. + rpc SetTombstoneGcConfig(SetTombstoneGcConfigRequest) returns (google.protobuf.Empty) {} + // ==================== Per-Instance Port Policy Override ==================== // Set an admin override for an instance's port policy. Takes precedence // over the policy reported by the instance itself, and survives app @@ -834,6 +842,23 @@ message SetCertbotConfigRequest { optional string acme_url = 4; } +// ==================== Tombstone GC Configuration Messages ==================== + +// Tombstone GC configuration response +message TombstoneGcConfigResponse { + // Collect tombstones once every this many replicated writes (0 = disabled) + uint64 writes_per_collection = 1; + // True when the value is the replicated KV override; false when it is this + // node's config-file default (which other nodes may not share) + bool stored_in_kv = 2; +} + +// Set tombstone GC configuration request +message SetTombstoneGcConfigRequest { + // Collect tombstones once every this many replicated writes (0 = disabled) + uint64 writes_per_collection = 1; +} + // ==================== Per-Instance Port Policy Override Messages ==================== // Set an admin override for an instance. diff --git a/dstack/gateway/src/admin_service.rs b/dstack/gateway/src/admin_service.rs index 4cf23dfad..0d22428a2 100644 --- a/dstack/gateway/src/admin_service.rs +++ b/dstack/gateway/src/admin_service.rs @@ -21,9 +21,9 @@ use dstack_gateway_rpc::{ RemoveCvmResponse, RemoveNodeRequest, RemoveNodeResponse, RenewCertResponse, RenewZtDomainCertRequest, RenewZtDomainCertResponse, RotateAcmeCredentialsResponse, SetCertbotConfigRequest, SetDefaultDnsCredentialRequest, SetInstancePortPolicyRequest, - SetInstanceReadyRequest, SetNodeStatusRequest, SetNodeUrlRequest, StatusResponse, - StoreSyncStatus, UpdateDnsCredentialRequest, WaveKvStatusResponse, ZtDomainCertStatus, - ZtDomainConfig as ProtoZtDomainConfig, ZtDomainInfo, + SetInstanceReadyRequest, SetNodeStatusRequest, SetNodeUrlRequest, SetTombstoneGcConfigRequest, + StatusResponse, StoreSyncStatus, TombstoneGcConfigResponse, UpdateDnsCredentialRequest, + WaveKvStatusResponse, ZtDomainCertStatus, ZtDomainConfig as ProtoZtDomainConfig, ZtDomainInfo, }; use ra_rpc::{CallContext, RpcCall}; use tracing::{info, warn}; @@ -31,8 +31,8 @@ use wavekv::node::NodeStatus as WaveKvNodeStatus; use crate::{ kv::{ - import::Rejection, DnsCredential, DnsProvider, GlobalCertbotConfig, NodeStatus, PortFlags, - PortPolicy, ZtDomainConfig, + import::Rejection, DnsCredential, DnsProvider, GlobalCertbotConfig, + GlobalTombstoneGcConfig, NodeStatus, PortFlags, PortPolicy, ZtDomainConfig, }, main_service::Proxy, models::PortPolicyView, @@ -718,6 +718,33 @@ impl AdminRpc for AdminRpcHandler { Ok(()) } + // ==================== Tombstone GC Configuration ==================== + + async fn get_tombstone_gc_config(self) -> Result { + Ok(match self.state.kv_store().get_tombstone_gc_config()? { + Some(stored) => TombstoneGcConfigResponse { + writes_per_collection: stored.writes_per_collection, + stored_in_kv: true, + }, + None => TombstoneGcConfigResponse { + writes_per_collection: self.state.config.sync.tombstone_gc_writes, + stored_in_kv: false, + }, + }) + } + + async fn set_tombstone_gc_config(self, request: SetTombstoneGcConfigRequest) -> Result<()> { + let config = GlobalTombstoneGcConfig { + writes_per_collection: request.writes_per_collection, + }; + self.state.kv_store().set_tombstone_gc_config(&config)?; + info!( + "updated tombstone GC config: writes_per_collection={}", + config.writes_per_collection + ); + Ok(()) + } + async fn set_instance_port_policy(self, request: SetInstancePortPolicyRequest) -> Result<()> { let proto = request.policy.context("port policy is required")?; let policy = port_policy_from_proto(proto)?; diff --git a/dstack/gateway/src/config.rs b/dstack/gateway/src/config.rs index e80e46bd0..fdab616ac 100644 --- a/dstack/gateway/src/config.rs +++ b/dstack/gateway/src/config.rs @@ -149,6 +149,15 @@ fn default_true() -> bool { true } +/// Roughly hourly on a busy ephemeral store and rarer on the quiet persistent +/// one -- the cadence follows write volume, which is also what produces +/// tombstones. Any value works for correctness: the ack watermark decides what +/// may go, not this, so it trades disk against how often every node stops to +/// scan its data map. +fn default_tombstone_gc_writes() -> u64 { + 10_000 +} + fn default_handshake_stale() -> Duration { Duration::from_secs(30 * 60) } @@ -590,6 +599,18 @@ pub struct SyncConfig { /// zero turns the work off -- which is why this one is a window. #[serde(with = "serde_duration")] pub wal_sync_window: Duration, + /// Collect tombstones once every this many replicated writes, per store. + /// Zero disables collection, which keeps every delete on disk for the life + /// of the deployment. + /// + /// The trigger is a count of replicated writes rather than a clock so that + /// every node in a cluster collects in the same window without agreeing on + /// the time: see `start_tombstone_gc_task`. This value is only the default + /// -- an operator override stored in the KV itself (`SetTombstoneGcConfig`) + /// replicates to every node and takes precedence, which is also the only + /// way to keep the boundaries identical when config files drift. + #[serde(default = "default_tombstone_gc_writes")] + pub tombstone_gc_writes: u64, /// Enable periodic sync of instance connections to KV store pub sync_connections_enabled: bool, /// Interval for syncing instance connections to KV store diff --git a/dstack/gateway/src/kv/mod.rs b/dstack/gateway/src/kv/mod.rs index 0d7433456..cef6585a1 100644 --- a/dstack/gateway/src/kv/mod.rs +++ b/dstack/gateway/src/kv/mod.rs @@ -49,7 +49,12 @@ pub use https_client::{AppIdValidator, HttpsClientConfig}; pub use sync_service::{fetch_peers_from_bootnode, PersistentWriteNotifier, WaveKvSyncService}; use tracing::{error, warn}; -use std::{collections::BTreeMap, net::Ipv4Addr, path::Path, time::Duration}; +use std::{ + collections::{BTreeMap, BTreeSet}, + net::Ipv4Addr, + path::Path, + time::Duration, +}; use anyhow::{Context, Result}; @@ -191,6 +196,37 @@ impl From<&InstanceInfo> for InstanceRecord { } } +/// What one round of tombstone collection freed, per store. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct CollectedTombstones { + pub persistent: usize, + pub ephemeral: usize, +} + +impl CollectedTombstones { + pub fn total(&self) -> usize { + self.persistent + self.ephemeral + } +} + +/// How many replicated writes this node covers, per store: the sum over every +/// origin of that origin's ack watermark. +/// +/// This is a function of replicated state, so every node's reading converges +/// within a sync round -- which is what lets the tombstone GC trigger on it +/// and land in the same window cluster-wide without any node reading a clock. +/// It is monotone in steady state, and steps back in two ways with different +/// lifetimes. Digest repair lowers an ack until retransmission restores it -- +/// a dip the GC trigger simply waits out. Removing a peer drops that origin's +/// watermark for good; the GC task answers by resetting its baseline (see +/// `start_tombstone_gc_task`) instead of waiting for the cluster to re-earn +/// writes it no longer remembers. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReplicatedWrites { + pub persistent: u64, + pub ephemeral: u64, +} + /// The `inst/` records currently in the KV store, split by readability. /// /// A key that is absent or tombstoned does not appear here at all — that is the @@ -371,6 +407,20 @@ impl Default for GlobalCertbotConfig { } } +/// Tombstone GC pacing (stored in KV, synced across nodes). +/// +/// The pace must be one number cluster-wide: nodes collecting on different +/// boundaries are back to collecting on their own phase, which the state +/// digest reads as divergence. Storing the override in the replicated KV is +/// what makes it one number; the per-node config file only supplies the +/// default for when this record is absent. +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +pub struct GlobalTombstoneGcConfig { + /// Collect once every this many replicated writes, per store. Zero + /// disables collection cluster-wide. + pub writes_per_collection: u64, +} + // Key prefixes and builders pub mod keys { use super::NodeId; @@ -394,6 +444,7 @@ pub mod keys { /// Shared by the `GLOBAL_*` keys below; not itself a key. pub const GLOBAL_PREFIX: &str = "global/"; pub const GLOBAL_CERTBOT_CONFIG: &str = "global/certbot_config"; + pub const GLOBAL_TOMBSTONE_GC_CONFIG: &str = "global/tombstone_gc_config"; pub const GLOBAL_ACME_CREDENTIALS: &str = "global/acme_credentials"; pub const GLOBAL_ACME_ATTESTATION: &str = "global/acme_attestation"; pub const GLOBAL_ACME_ROTATION_LOCK: &str = "global/acme_rotation_lock"; @@ -1483,6 +1534,95 @@ impl KvStore { // ==================== Persistence ==================== + /// Drop tombstones every known peer has already covered. + /// + /// wavekv gates this on an ack watermark rather than a clock: a tombstone + /// may go only once every peer reports having seen the delete that wrote + /// it. v1 took a TTL instead and 2.0 removed that API, because under any + /// state-shipping scheme an uncoordinated clock-based collection lets a + /// lagging replica resurrect the key -- here, a deregistered CVM + /// reappearing in every node's WireGuard config. + /// + /// Both stores, because both accumulate. A `conn/` or `handshake/` + /// tombstone is only in memory and goes on restart; an `inst/` or `admin/` + /// one is on disk and does not, so without this it is kept for the life of + /// the deployment. + /// + /// Both are attempted even if the first fails, for the same reason the + /// deletes are: stopping halfway leaves the stores disagreeing about what + /// has been collected, which is the state the digest reports as divergence. + /// And when both fail, the error names both -- the second failure is a + /// separate fact, not a detail of the first. + pub fn collect_tombstone_garbage(&self) -> Result { + let persistent = self + .persistent + .write() + .collect_tombstone_garbage() + .context("failed to collect tombstones from the persistent store"); + let ephemeral = self + .ephemeral + .write() + .collect_tombstone_garbage() + .context("failed to collect tombstones from the ephemeral store"); + match (persistent, ephemeral) { + (Ok(persistent), Ok(ephemeral)) => Ok(CollectedTombstones { + persistent, + ephemeral, + }), + (Err(err), Ok(_)) | (Ok(_), Err(err)) => Err(err), + (Err(persistent), Err(ephemeral)) => { + Err(persistent.context(format!("the ephemeral store failed too: {ephemeral:#}"))) + } + } + } + + /// How many replicated writes this node covers, per store. + /// + /// The tombstone GC trigger compares this against a boundary; see + /// `ReplicatedWrites` for why it is a count and not a clock. + pub fn replicated_writes(&self) -> ReplicatedWrites { + let sum = |node: &Node| { + node.read() + .acks_snapshot() + .values() + .fold(0u64, |acc, seq| acc.saturating_add(*seq)) + }; + ReplicatedWrites { + persistent: sum(&self.persistent), + ephemeral: sum(&self.ephemeral), + } + } + + /// The sync peer set (both stores share membership). + /// + /// The tombstone GC task watches this for shrinkage: a removed peer takes + /// its ack watermark out of `replicated_writes` permanently, which must + /// not be read as "the cluster has stopped writing". + pub fn peer_ids(&self) -> BTreeSet { + self.persistent.read().get_peers().into_iter().collect() + } + + /// Get the tombstone GC pacing override, if an operator has stored one. + /// + /// `None` means "use the config-file default". Fails closed on a corrupt + /// record: falling back to either the default or `None` would silently put + /// this node on a different collection boundary from its peers. + pub fn get_tombstone_gc_config(&self) -> Result> { + self.persistent + .read() + .decode_strict(keys::GLOBAL_TOMBSTONE_GC_CONFIG) + } + + /// Store the tombstone GC pacing override, replicating it to every node. + pub fn set_tombstone_gc_config(&self, config: &GlobalTombstoneGcConfig) -> Result<()> { + self.persistent.write().put_encoded( + keys::GLOBAL_TOMBSTONE_GC_CONFIG.to_string(), + config, + true, + )?; + Ok(()) + } + pub fn persist_if_dirty(&self) -> Result { self.persistent.persist_if_dirty() } @@ -2660,11 +2800,11 @@ mod decompression_tests { mod corruption_tests { use super::*; - fn test_kv(data_dir: &std::path::Path) -> KvStore { + pub(super) fn test_kv(data_dir: &std::path::Path) -> KvStore { KvStore::new(1, vec![], data_dir, None).expect("failed to create kv store") } - fn put_raw(kv: &KvStore, key: &str, value: &[u8]) { + pub(super) fn put_raw(kv: &KvStore, key: &str, value: &[u8]) { kv.persistent .write() .put(key.to_string(), value.to_vec()) @@ -3315,3 +3455,346 @@ mod key_schema_tests { assert_eq!(keys::parse_node_info_key("node/info/not-a-number"), None); } } + +/// Tombstone GC: what may be collected (the ack watermark), when collection +/// triggers (the replicated write count), and how the pace is shared (the KV +/// override). Split from `corruption_tests` because these are a feature's +/// tests, not corruption drills -- they only borrow its raw-write helpers. +#[cfg(test)] +mod tombstone_gc_tests { + use super::corruption_tests::{put_raw, test_kv}; + use super::*; + + fn seed_instance(kv: &KvStore, id: &str) { + kv.sync_instance( + id, + &InstanceRecord { + app_id: "app".to_string(), + ip: "10.0.0.20".parse().unwrap(), + public_key: format!("key-{id}"), + reg_time: 1, + port_policy: None, + port_policy_hash: String::new(), + health_check: None, + }, + ) + .expect("seed"); + } + + /// Whether the key is still held at all, tombstone included. + fn key_is_held(kv: &KvStore, key: &str) -> bool { + kv.persistent.read().get_including_tombstones(key).is_some() + } + + /// A delete leaves a tombstone, and nothing collected them: they are + /// replicated state, on disk, kept for the life of the deployment. A + /// single-node gateway has no peer to resurrect from, which is exactly the + /// case wavekv collects unconditionally. + #[test] + fn a_single_node_collects_the_tombstones_it_has_finished_with() { + let dir = tempfile::tempdir().expect("temp dir"); + let kv = test_kv(dir.path()); + seed_instance(&kv, "cvm"); + kv.sync_delete_instance("cvm").expect("delete"); + assert!( + key_is_held(&kv, &keys::inst("cvm")), + "a delete leaves a tombstone behind" + ); + assert!( + kv.load_all_instances().decoded.is_empty(), + "which is not the same as the record still being live" + ); + + let collected = kv.collect_tombstone_garbage().expect("collect"); + // The delete tombstones the admin override keys alongside `inst/` + // (recycled ids must not inherit an operator's gate), so the count is + // "at least the instance record", not exactly one. + assert!(collected.persistent >= 1); + assert!(!key_is_held(&kv, &keys::inst("cvm"))); + } + + /// The watermark, not a clock, is what makes this safe. A peer that has not + /// acknowledged the delete may still be holding the live record, so + /// dropping the tombstone would let it push that record back -- a + /// deregistered CVM reappearing in every node's WireGuard config. v1 took a + /// TTL here and 2.0 removed the API for this reason. + #[test] + fn a_tombstone_a_peer_has_not_acknowledged_is_kept() { + let dir = tempfile::tempdir().expect("temp dir"); + let kv = test_kv(dir.path()); + seed_instance(&kv, "cvm"); + kv.sync_delete_instance("cvm").expect("delete"); + kv.add_peer(2).expect("add peer"); + + let collected = kv.collect_tombstone_garbage().expect("collect"); + assert_eq!(collected.persistent, 0); + assert!( + key_is_held(&kv, &keys::inst("cvm")), + "a peer that has not seen the delete can still resurrect the record" + ); + } + + /// Exchange sync envelopes until a full round moves no entries in either + /// direction, then stop. + /// + /// The real path, not a hand-set ack: what makes a tombstone collectable is + /// the peer reporting that it covers the delete, and only a round trip + /// produces that report. The quiet round's envelopes are still applied -- + /// an empty delta still carries the sender's ack map, which is exactly the + /// report the collector is waiting on. + fn sync_until_quiet(left: &KvStore, right: &KvStore) { + for _ in 0..8 { + let mut moved = 0; + for (from, to) in [(left, right), (right, left)] { + let request = from + .persistent + .read() + .prepare_sync(to.my_node_id(), vec![0u8; 16]); + moved += request.entries.len(); + to.persistent + .write() + .apply_envelope(request) + .expect("peer applies our envelope"); + let reply = to + .persistent + .read() + .prepare_sync(from.my_node_id(), vec![0u8; 16]); + moved += reply.entries.len(); + from.persistent + .write() + .apply_envelope(reply) + .expect("we apply the peer's envelope"); + } + if moved == 0 { + return; + } + } + panic!("the stores were still exchanging entries after 8 rounds"); + } + + /// The property the ack watermark buys, and the one a TTL cannot: age plays + /// no part. A tombstone written a moment ago is collectable the instant + /// every peer covers the delete -- and until then it is kept no matter how + /// long that takes, where a TTL would drop it and let the peer push the + /// record it still holds back as a live value. + #[test] + fn a_tombstone_is_collected_on_coverage_not_on_age() { + let left_dir = tempfile::tempdir().expect("temp dir"); + let right_dir = tempfile::tempdir().expect("temp dir"); + let left = KvStore::new(1, vec![2], left_dir.path(), None).expect("left"); + let right = KvStore::new(2, vec![1], right_dir.path(), None).expect("right"); + + seed_instance(&left, "cvm"); + sync_until_quiet(&left, &right); + assert!( + right.load_all_instances().decoded.contains_key("cvm"), + "the peer must hold the live record, or there is nothing to resurrect" + ); + + left.sync_delete_instance("cvm").expect("delete"); + // Freshly written, and the peer has not been told. Age is irrelevant + // here; coverage is not. + assert_eq!( + left.collect_tombstone_garbage() + .expect("collect") + .persistent, + 0 + ); + assert!(key_is_held(&left, &keys::inst("cvm"))); + + sync_until_quiet(&left, &right); + assert!( + right.load_all_instances().decoded.is_empty(), + "the peer must have applied the delete" + ); + assert!( + left.collect_tombstone_garbage() + .expect("collect") + .persistent + >= 1, + "once every peer covers the delete, nothing can push the record back" + ); + assert!(!key_is_held(&left, &keys::inst("cvm"))); + + // And the peer cannot reintroduce it afterwards. + sync_until_quiet(&left, &right); + assert!(left.load_all_instances().decoded.is_empty()); + assert!(right.load_all_instances().decoded.is_empty()); + } + + /// The GC trigger counts replicated writes instead of reading a clock, so + /// what it counts must behave like replicated state: advance with writes, + /// and read the same on every converged node. + #[test] + fn the_replicated_write_count_advances_with_writes_and_converges_between_peers() { + let left_dir = tempfile::tempdir().expect("temp dir"); + let right_dir = tempfile::tempdir().expect("temp dir"); + let left = KvStore::new(1, vec![2], left_dir.path(), None).expect("left"); + let right = KvStore::new(2, vec![1], right_dir.path(), None).expect("right"); + + let before = left.replicated_writes(); + seed_instance(&left, "cvm"); + left.sync_delete_instance("cvm").expect("delete"); + let after = left.replicated_writes(); + assert!( + after.persistent > before.persistent, + "a write and a delete both advance the count" + ); + + assert!( + right.replicated_writes().persistent < after.persistent, + "the peer has not covered the writes yet" + ); + sync_until_quiet(&left, &right); + assert_eq!( + left.replicated_writes().persistent, + right.replicated_writes().persistent, + "converged nodes read the same count, which is what lets them share GC boundaries" + ); + } + + /// `RemovePeer` drops the removed origin's ack watermark, so the write + /// count steps back for good -- the regression the GC task's baseline + /// reset exists for: without it, a boundary earned before the removal + /// gates collection until the cluster re-earns writes it no longer + /// remembers. + #[test] + fn removing_a_peer_permanently_lowers_the_replicated_write_count() { + let left_dir = tempfile::tempdir().expect("temp dir"); + let right_dir = tempfile::tempdir().expect("temp dir"); + let left = KvStore::new(1, vec![2], left_dir.path(), None).expect("left"); + let right = KvStore::new(2, vec![1], right_dir.path(), None).expect("right"); + + // Writes authored by the peer, so the count left holds for origin 2 + // is exactly what the removal deletes. + seed_instance(&right, "cvm"); + sync_until_quiet(&left, &right); + let before = left.replicated_writes().persistent; + + assert!(left.peer_ids().contains(&2)); + left.remove_peer(2).expect("remove peer"); + assert!( + !left.peer_ids().contains(&2), + "the shrinkage the GC task watches for" + ); + assert!( + left.replicated_writes().persistent < before, + "the removed origin's watermark is gone from the count, permanently" + ); + } + + /// The resurrection the watermark cannot prevent, pinned as a known + /// limitation. + /// + /// Removing a peer vacates the watermark, collection then drops the + /// tombstone, and the removed node returning with its old data directory + /// still holds the record live. Ordinary rounds do not leak it back: the + /// returning node's request re-teaches the responder coverage of its + /// origin, so nothing it authored is re-sent. What breaks it is wavekv's + /// own divergence repair -- the digests disagree for as long as the + /// returning node holds the zombie, and after `digest_check_rounds` the + /// repair (`reset_peer_coverage`) makes it re-send everything, tombstoned + /// keys included, with nothing left to beat them under LWW. + /// + /// Until a removed node is locked out at the sync boundary, removal must + /// mean decommission: the removed node's data directory must never come + /// back. + #[test] + fn a_removed_node_returning_with_old_state_resurrects_collected_deletes() { + let left_dir = tempfile::tempdir().expect("temp dir"); + let right_dir = tempfile::tempdir().expect("temp dir"); + let left = KvStore::new(1, vec![2], left_dir.path(), None).expect("left"); + let right = KvStore::new(2, vec![1], right_dir.path(), None).expect("right"); + + // The record is authored by the node that will be removed. + seed_instance(&right, "cvm"); + sync_until_quiet(&left, &right); + + // right goes dark; left deletes the record, retires right, and -- now + // peerless -- collects the tombstone. + left.sync_delete_instance("cvm").expect("delete"); + left.remove_peer(2).expect("retire"); + assert!( + left.collect_tombstone_garbage() + .expect("collect") + .persistent + >= 1 + ); + assert!(left.load_all_instances().decoded.is_empty()); + assert!( + right.load_all_instances().decoded.contains_key("cvm"), + "the removed node still holds the record live -- the zombie" + ); + + // right comes back and initiates rounds -- the real topology: left + // pruned it, so left never initiates. Ordinary rounds do not + // resurrect: right's request carries its ack map, left re-adopts + // coverage of origin 2, and right's zombie stays filtered out. + let ordinary_round = || { + let request = right.persistent.read().prepare_sync(1, vec![0u8; 16]); + let reply = left + .persistent + .write() + .handle_envelope(request, vec![0u8; 16]) + .expect("left answers the returning node"); + right + .persistent + .write() + .apply_envelope(reply) + .expect("right applies the reply"); + }; + ordinary_round(); + ordinary_round(); + assert!( + left.load_all_instances().decoded.is_empty(), + "the plain delta path does not leak the zombie back" + ); + + // But the digests now disagree for good -- left lacks a key right + // holds, with no tombstone and no ack filter to reconcile them -- so + // after `digest_check_rounds` quiescent mismatches the divergence + // repair fires on the returning node, and its next request is a full + // dump. + right.persistent.write().reset_peer_coverage(1); + ordinary_round(); + assert!( + left.load_all_instances().decoded.contains_key("cvm"), + "the deregistered CVM is back, live, in every node's WireGuard config" + ); + } + + /// The pace override lives in the KV so that it is one number cluster-wide; + /// absent means "use the config-file default", and the record replicates + /// like any other write. + #[test] + fn a_tombstone_gc_override_is_absent_by_default_and_replicates() { + let left_dir = tempfile::tempdir().expect("temp dir"); + let right_dir = tempfile::tempdir().expect("temp dir"); + let left = KvStore::new(1, vec![2], left_dir.path(), None).expect("left"); + let right = KvStore::new(2, vec![1], right_dir.path(), None).expect("right"); + + assert_eq!(left.get_tombstone_gc_config().expect("read"), None); + + let config = GlobalTombstoneGcConfig { + writes_per_collection: 5000, + }; + left.set_tombstone_gc_config(&config).expect("store"); + sync_until_quiet(&left, &right); + assert_eq!( + right.get_tombstone_gc_config().expect("read"), + Some(config), + "every node reads the operator's pace, not its own default" + ); + } + + /// Fails closed like the certbot config: reading a corrupt override as the + /// per-node default would silently put this node on a different collection + /// boundary from its peers. + #[test] + fn a_corrupt_tombstone_gc_override_does_not_read_as_the_default() { + let dir = tempfile::tempdir().expect("temp dir"); + let kv = test_kv(dir.path()); + put_raw(&kv, keys::GLOBAL_TOMBSTONE_GC_CONFIG, b"not-messagepack"); + assert!(kv.get_tombstone_gc_config().is_err()); + } +} diff --git a/dstack/gateway/src/main_service.rs b/dstack/gateway/src/main_service.rs index 82e934de6..5288a6c30 100644 --- a/dstack/gateway/src/main_service.rs +++ b/dstack/gateway/src/main_service.rs @@ -41,7 +41,7 @@ use crate::{ kv::{ fetch_peers_from_bootnode, import, AppIdValidator, CertData, HttpsClientConfig, InstanceRecord, KvStore, LegacyOverrides, LoadedInstances, NodeData, NodeStatus, - PortPolicy, PortPolicyOverride, WaveKvSyncService, + PortPolicy, PortPolicyOverride, ReplicatedWrites, WaveKvSyncService, }, models::{InstanceInfo, PortPolicyView, ReportedCapabilities, WgConf, WgPeer}, proxy::{create_acceptor_with_cert_resolver, AddressGroup, AddressInfo, AppAddressResolver}, @@ -1116,6 +1116,143 @@ async fn start_wavekv_sync_task(proxy: Proxy, wavekv_sync: Arc, now: ReplicatedWrites, n: u64) -> bool { + let Some(last) = last else { + return true; + }; + now.persistent / n > last.persistent / n || now.ephemeral / n > last.ephemeral / n +} + +/// Drop tombstones the cluster has finished with. +/// +/// Beside the persist and WAL tasks because it belongs to the same set: work +/// the store needs on a schedule whether or not this node has peers. A +/// single-node gateway has nobody to resurrect from, so wavekv collects every +/// tombstone it holds -- and it is the deployment where nothing else ever +/// would. +/// +/// The pace is read every period rather than once at startup because the +/// operator override lives in the KV itself and can change under a running +/// node. A corrupt override skips the round instead of falling back to the +/// config-file default: the default is per-node, and collecting on it while +/// peers honour the override is exactly the phase drift the shared pace +/// exists to prevent. +/// +/// Collection runs on the blocking pool for the same reason persist and WAL +/// sync do: it takes the store's write lock and walks the whole data map, and +/// a proxy's event loop cannot afford a worker parked on that. +fn start_tombstone_gc_task(proxy: &Proxy) { + let default_pace = proxy.config.sync.tombstone_gc_writes; + let kv_store = proxy.kv_store.clone(); + tokio::spawn(async move { + let mut last: Option = None; + let mut peers = kv_store.peer_ids(); + let mut override_unreadable = false; + loop { + tokio::time::sleep(TOMBSTONE_GC_CHECK_PERIOD).await; + let pace = match kv_store.get_tombstone_gc_config() { + Ok(stored) => { + if override_unreadable { + info!("WaveKV: the tombstone GC override is readable again"); + override_unreadable = false; + } + stored.map_or(default_pace, |c| c.writes_per_collection) + } + Err(err) => { + // Corruption is deterministic until an operator overwrites + // the record, so report the pause once, not every period. + if !override_unreadable { + error!( + "WaveKV: the tombstone GC override is unreadable, \ + collection is paused until it is overwritten: {err:?}" + ); + override_unreadable = true; + } + continue; + } + }; + if pace == 0 { + // Re-enabling starts from the backlog case, deliberately. + last = None; + continue; + } + // A removed peer takes its lifetime of writes out of the count for + // good (`RemovePeer` drops `acks[removed]`). A baseline above the + // shrunken count would gate collection on the cluster re-earning + // writes it no longer remembers -- on a quiet store, indefinitely. + // Removal replicates, so every node observes it within a sync + // round of the others and their resets land clustered. + let now_peers = kv_store.peer_ids(); + if !peers.is_subset(&now_peers) { + last = None; + } + peers = now_peers; + let now = kv_store.replicated_writes(); + if !tombstone_collection_due(last, now, pace) { + continue; + } + let kv = kv_store.clone(); + match tokio::task::spawn_blocking(move || kv.collect_tombstone_garbage()).await { + Ok(Ok(collected)) => { + // On failure `last` keeps its value and the next period + // retries; only success moves the boundary. + last = Some(now); + if collected.total() > 0 { + info!( + "WaveKV: collected {} tombstones ({} persistent, {} ephemeral)", + collected.total(), + collected.persistent, + collected.ephemeral + ); + } + } + Ok(Err(err)) => error!("WaveKV: tombstone collection failed: {err:?}"), + Err(err) => error!("WaveKV: the tombstone collection task did not finish: {err}"), + } + } + }); + if default_pace == 0 { + info!("WaveKV: tombstone collection disabled by default; a stored override can enable it"); + } else { + info!( + "WaveKV: tombstone collection enabled (default pace: every {default_pace} replicated writes)" + ); + } +} + fn start_wavekv_watch_task(proxy: Proxy) -> Result<()> { let kv_store = proxy.kv_store.clone(); @@ -1256,6 +1393,8 @@ fn start_wavekv_watch_task(proxy: Proxy) -> Result<()> { info!("WaveKV: deferred WAL sync enabled (window: {wal_sync_window:?})"); } + start_tombstone_gc_task(&proxy); + // Start periodic connection sync task if proxy.config.sync.sync_connections_enabled { let sync_interval = proxy.config.sync.sync_connections_interval; diff --git a/dstack/gateway/src/main_service/tests.rs b/dstack/gateway/src/main_service/tests.rs index 2244ea3bb..5214495b2 100644 --- a/dstack/gateway/src/main_service/tests.rs +++ b/dstack/gateway/src/main_service/tests.rs @@ -2627,3 +2627,39 @@ async fn a_stuck_bad_record_is_reported_once_and_again_when_it_recovers() { reload_instances_from_kv_store(&state.proxy, &state.kv_store).unwrap(); assert!(state.lock().reported_rejections.is_empty()); } + +/// The GC trigger is a pure function of the replicated write count and the +/// shared pace, so every node in a cluster crosses each collection boundary +/// together without reading a clock. The digest covers tombstones, so a node +/// that has collected and one that has not look diverged to wavekv's detector, +/// which forces a full re-exchange; nodes collecting on their own phase would +/// pay that every cycle. +#[test] +fn tombstone_collection_triggers_on_write_count_boundaries_not_on_time() { + let w = |persistent, ephemeral| ReplicatedWrites { + persistent, + ephemeral, + }; + + // Never collected: due regardless of the count — the backlog case, where a + // store that never writes again would otherwise never shed the tombstones + // it already holds. + assert!(tombstone_collection_due(None, w(0, 0), 100)); + + // Inside the boundary nothing is due, no matter how much time has passed. + assert!(!tombstone_collection_due(Some(w(50, 0)), w(99, 0), 100)); + // Crossing it is what makes collection due, on either store. + assert!(tombstone_collection_due(Some(w(50, 0)), w(100, 0), 100)); + assert!(tombstone_collection_due(Some(w(50, 0)), w(50, 100), 100)); + + // Two nodes whose views of the same state are a beat apart agree on every + // boundary: whichever sees the crossing later still sees it. + let earlier = w(99, 0); + let later = w(101, 0); + assert!(!tombstone_collection_due(Some(w(50, 0)), earlier, 100)); + assert!(tombstone_collection_due(Some(w(50, 0)), later, 100)); + + // Digest repair can lower an ack; a count that stepped back is not due + // until it crosses the next boundary again. + assert!(!tombstone_collection_due(Some(w(100, 0)), w(90, 0), 100)); +}