feat(gateway): upgrade to wavekv 2.0 delta-state sync with v1 dual-stack

dstack/Cargo.toml

@@ -119,7 +119,8 @@ serde-duration = { path = "serde-duration" }
 dstack-mr = { path = "dstack-mr" }
 dstack-verifier = { path = "verifier", default-features = false }
 size-parser = { path = "size-parser" }
-wavekv = "1.0.0"
+# TODO: repoint to `wavekv = "2.0"` once Phala-Network/wavekv#3 is released to crates.io.
+wavekv = { git = "https://github.com/Phala-Network/wavekv", branch = "feat/delta-state-sync" }
 
 # Core dependencies
 anyhow = { version = "1.0.97", default-features = false }

dstack/gateway/rpc/proto/gateway_rpc.proto

@@ -339,9 +339,22 @@ message PeerSyncStatus {
   uint32 id = 1;
   uint64 local_ack = 2;
   uint64 peer_ack = 3;
-  uint64 buffered_logs = 4;
+  // Always 0 since wavekv 2.0, which replicates state instead of operation logs and
+  // keeps no per-peer log buffers. Retained so existing clients keep decoding.
+  uint64 buffered_logs = 4 [deprecated = true];
   // Last seen timestamps: [(observer_node_id, timestamp), ...]
   repeated LastSeenEntry last_seen = 5;
+  // Whether this peer has ever reported an ack map.
+  bool heard_from = 6;
+  // Sync protocol last negotiated with this peer: "v1" or "v2".
+  string protocol = 7;
+  // Consecutive quiescent rounds whose state digests disagreed. Non-zero means the
+  // replicas have silently diverged; wavekv 1.x could not detect this at all.
+  uint32 digest_mismatches = 8;
+  // Consecutive sync rounds that failed outright. Only a definitive 404/405 demotes a
+  // peer to "v1"; a 5xx or a timeout leaves `protocol` untouched by design, so this is
+  // the only field that moves when a peer is failing every round.
+  uint32 consecutive_failures = 9;
 }
 
 message LastSeenEntry {
@@ -358,6 +371,13 @@ message StoreSyncStatus {
   bool dirty = 5;
   bool wal_enabled = 6;
   repeated PeerSyncStatus peers = 7;
+  // Hex SHA-256 over the replicated state. Two converged replicas produce equal
+  // digests by construction, so comparing this across the cluster is the promotion
+  // gate for the wavekv v2 rollout and the standing divergence check afterwards.
+  string digest = 8;
+  uint64 entries_merged = 9;
+  // Entries refused by the admission policy or the ingest quotas.
+  uint64 entries_rejected = 10;
 }
 
 // WaveKV sync status response

dstack/gateway/src/admin_service.rs

@@ -208,16 +208,33 @@ impl AdminRpc for AdminRpcHandler {
                 .collect()
         };
 
+        // Per-peer protocol/digest telemetry lives on the sync manager, not the store.
+        let links = self
+            .state
+            .wavekv_sync
+            .as_ref()
+            .map(|s| s.link_status())
+            .unwrap_or_default();
+        let links_for = |name: &str| -> Vec<wavekv::sync::PeerLinkStatus> {
+            links
+                .iter()
+                .find(|(store, _)| *store == name)
+                .map(|(_, l)| l.clone())
+                .unwrap_or_default()
+        };
+
         Ok(WaveKvStatusResponse {
             enabled: self.state.config.sync.enabled,
             persistent: Some(build_store_status(
                 "persistent",
                 persistent_status,
+                &links_for("persistent"),
                 &get_peer_last_seen,
             )),
             ephemeral: Some(build_store_status(
                 "ephemeral",
                 ephemeral_status,
+                &links_for("ephemeral"),
                 &get_peer_last_seen,
             )),
         })
@@ -718,6 +735,7 @@ fn port_policy_view_to_proto(view: PortPolicyView) -> GetInstancePortPolicyRespo
 fn build_store_status(
     name: &str,
     status: WaveKvNodeStatus,
+    links: &[wavekv::sync::PeerLinkStatus],
     get_peer_last_seen: &impl Fn(u32) -> Vec<(u32, u64)>,
 ) -> StoreSyncStatus {
     StoreSyncStatus {
@@ -727,6 +745,9 @@ fn build_store_status(
         next_seq: status.next_seq,
         dirty: status.dirty,
         wal_enabled: status.wal,
+        digest: status.digest,
+        entries_merged: status.entries_merged,
+        entries_rejected: status.entries_rejected,
         peers: status
             .peers
             .into_iter()
@@ -735,12 +756,19 @@ fn build_store_status(
                     .into_iter()
                     .map(|(node_id, timestamp)| LastSeenEntry { node_id, timestamp })
                     .collect();
+                let link = links.iter().find(|l| l.id == p.id);
+                #[allow(deprecated)]
                 ProtoPeerSyncStatus {
                     id: p.id,
                     local_ack: p.ack,
-                    peer_ack: p.pack,
-                    buffered_logs: p.logs as u64,
+                    peer_ack: p.peer_ack,
+                    // wavekv 2.0 keeps no per-peer log buffers.
+                    buffered_logs: 0,
                     last_seen,
+                    heard_from: p.heard_from,
+                    protocol: link.map(|l| l.protocol).unwrap_or_default().to_string(),
+                    digest_mismatches: link.map(|l| l.digest_mismatches).unwrap_or(0),
+                    consecutive_failures: link.map(|l| l.consecutive_failures).unwrap_or(0),
                 }
             })
             .collect(),