feat(backup): per-tenant write-HLC high-water and restore staleness gate

Restoring an older backup into a cluster that has already committed
newer writes for the same tenant silently rolls those writes back.
This change introduces a per-tenant monotonic high-water mark on
SharedState that records the HLC wall-ns of every successful dispatch
and uses it in RESTORE to reject envelopes whose snapshot_watermark
predates the destination's observed write timeline.

SharedState gains a `tenant_write_hlc` map and an
`advance_tenant_write_hlc` helper that is called from every successful
dispatch path: gateway core, dispatch_utils, pgwire sync_dispatch, and
pgwire handler dispatch.

The backup orchestrator captures the envelope's `snapshot_watermark`
from `hlc_clock.now()` AFTER completing the per-node fan-out, so the
stamped watermark is guaranteed to be >= any tenant_write_hlc advanced
by the orchestrator's own dispatches. Envelopes with watermark == 0
(pre-feature backups) are passed through for compatibility.

Author: farhan-syah

nodedb/src/control/backup/orchestrator.rs

@@ -36,20 +36,38 @@ pub async fn backup_tenant(state: &Arc<SharedState>, tenant_id: u32) -> Result<B
     let nodes = unique_origin_nodes(state);
     let snapshot_plan = PhysicalPlan::Meta(MetaOp::CreateTenantSnapshot { tenant_id });
 
-    let meta = EnvelopeMeta {
-        tenant_id,
-        source_vshard_count: VSHARD_COUNT,
-        hash_seed: 0,          // VSHARD_COUNT-derived hash; no seed today
-        snapshot_watermark: 0, // global watermark capture is a future enhancement
-    };
-    let mut writer = EnvelopeWriter::new(meta);
-
+    // Collect per-node sections first. The orchestrator's own
+    // dispatches advance the tenant write-HLC high-water via
+    // `dispatch_async`; capturing the envelope watermark AFTER the
+    // fan-out guarantees `envelope.watermark ≥ tenant_write_hlc`
+    // at backup time, so a subsequent restore of this envelope into
+    // the same (unchanged) cluster passes the staleness gate.
+    let mut sections = Vec::with_capacity(nodes.len());
     for node_id in nodes {
         let body = if is_self(state, node_id) {
             snapshot_self(state, tenant_id, &snapshot_plan).await?
         } else {
             snapshot_remote(state, node_id, tenant_id, &snapshot_plan).await?
         };
+        sections.push((node_id, body));
+    }
+
+    // Capture a cluster-wide logical instant for the envelope via the
+    // HLC. `hlc_clock.now()` advances past any previously observed
+    // local or remote HLC — the wall-ns component is the scalar
+    // watermark we stamp into the header. Restore compares this
+    // against the destination's `tenant_write_hlc` to detect stale
+    // envelopes.
+    let snapshot_watermark = state.hlc_clock.now().wall_ns;
+    let meta = EnvelopeMeta {
+        tenant_id,
+        source_vshard_count: VSHARD_COUNT,
+        hash_seed: 0, // VSHARD_COUNT-derived hash; no seed today
+        snapshot_watermark,
+    };
+    let mut writer = EnvelopeWriter::new(meta);
+
+    for (node_id, body) in sections {
         writer
             .push_section(node_id, body)
             .map_err(|e| Error::Internal {

nodedb/src/control/backup/restore.rs

@@ -81,6 +81,34 @@ pub async fn restore_tenant(
         });
     }
 
+    // Staleness gate. An envelope whose captured watermark is older
+    // than any dispatch this destination cluster has already observed
+    // for the same tenant would silently roll back newer committed
+    // writes. Reject unconditionally today — a `FORCE` override is
+    // the operator's escape hatch and is a separate DDL decision.
+    // Envelopes with `snapshot_watermark == 0` come from a source
+    // that did not capture a watermark (pre-gate envelopes); let them
+    // through for compatibility — the gate only rejects real
+    // ordered-against-ordered comparisons.
+    if !dry_run && env.meta.snapshot_watermark != 0 {
+        let current_high_water = state
+            .tenant_write_hlc
+            .lock()
+            .ok()
+            .and_then(|map| map.get(&tenant_id).copied())
+            .unwrap_or(0);
+        if env.meta.snapshot_watermark < current_high_water {
+            return Err(Error::Internal {
+                detail: format!(
+                    "restore refused: envelope watermark {} is older than the \
+                     destination cluster's last observed write-HLC {} for tenant \
+                     {} — newer writes would be silently overwritten",
+                    env.meta.snapshot_watermark, current_high_water, tenant_id
+                ),
+            });
+        }
+    }
+
     let mut stats = RestoreStats {
         tenant_id,
         dry_run,

nodedb/src/control/gateway/core.rs

@@ -104,6 +104,18 @@ impl Gateway {
             0,
             result.is_ok(),
         );
+
+        // Advance per-tenant observed write-HLC high-water on any
+        // successful cluster dispatch (local or remote). Used by
+        // RESTORE staleness gate. Tracking on success of every
+        // gateway.execute is intentional: backup captures its
+        // envelope watermark AFTER its own fan-out, so a fresh
+        // backup's watermark always dominates the tenant_wm it
+        // itself advanced.
+        if result.is_ok() {
+            self.shared.advance_tenant_write_hlc(ctx.tenant_id.as_u32());
+        }
+
         result
     }
 

nodedb/src/control/server/dispatch_utils.rs

@@ -245,6 +245,15 @@ pub async fn dispatch_to_data_plane_with_source(
         }
     }
 
+    // Advance the tenant's observed write-HLC high-water on any
+    // successful dispatch. Used by RESTORE staleness gate. Advance
+    // on every success (not just writes) is intentionally
+    // conservative — envelope.watermark is captured AFTER fan-out so
+    // it always dominates the tenant_wm of a fresh backup.
+    if response.status == crate::bridge::envelope::Status::Ok {
+        shared.advance_tenant_write_hlc(tenant_id.as_u32());
+    }
+
     observe(shared);
     Ok(response)
 }

nodedb/src/control/server/pgwire/ddl/sync_dispatch.rs

@@ -98,5 +98,16 @@ pub async fn dispatch_async_with_source(
         return Err(crate::Error::Internal { detail });
     }
 
+    // Advance the tenant's observed write-HLC high-water. Used by
+    // RESTORE to reject stale envelopes. Tracking on every dispatch
+    // (not just known-write ops) is intentional: advance is
+    // monotonic, and capturing the backup envelope's watermark AFTER
+    // its own fan-out ensures envelope.wm ≥ tenant_wm on a fresh
+    // backup (so a same-cluster roundtrip passes the staleness gate).
+    // Reached only after the `resp.status != Ok` early-return above, so
+    // this point is the "success" branch per the advance_tenant_write_hlc
+    // contract.
+    state.advance_tenant_write_hlc(tenant_id.as_u32());
+
     Ok(resp.payload.to_vec())
 }

nodedb/src/control/server/pgwire/handler/dispatch.rs

@@ -16,6 +16,22 @@ impl NodeDbPgHandler {
     /// In cluster mode, write operations are proposed to Raft first and only
     /// executed on the Data Plane after quorum commit. Reads bypass Raft.
     pub(super) async fn dispatch_task(&self, task: PhysicalTask) -> crate::Result<Response> {
+        let tenant_id = task.tenant_id;
+        let result = self.dispatch_task_inner(task).await;
+        // Advance per-tenant observed write-HLC high-water on any
+        // successful dispatch (local, raft-replicated, or broadcast).
+        // Used by RESTORE's staleness gate. Backup captures envelope
+        // watermark AFTER its own fan-out, so envelope.wm dominates
+        // tenant_wm on a fresh backup.
+        if let Ok(ref resp) = result
+            && resp.status == crate::bridge::envelope::Status::Ok
+        {
+            self.state.advance_tenant_write_hlc(tenant_id.as_u32());
+        }
+        result
+    }
+
+    async fn dispatch_task_inner(&self, task: PhysicalTask) -> crate::Result<Response> {
         if matches!(
             task.plan,
             crate::bridge::envelope::PhysicalPlan::Document(

nodedb/src/control/state/fields.rs

@@ -352,6 +352,14 @@ pub struct SharedState {
     /// before every stamp so cross-node causality is preserved.
     pub hlc_clock: Arc<nodedb_types::HlcClock>,
 
+    /// Per-tenant monotonic high-water HLC wall-ns observed for any
+    /// dispatched plan. Used by RESTORE to reject envelopes whose
+    /// captured watermark is older than the destination cluster's
+    /// most recent observed dispatch for the same tenant —
+    /// silently overwriting newer committed writes would otherwise
+    /// be a correctness bug.
+    pub tenant_write_hlc: Arc<std::sync::Mutex<std::collections::HashMap<u32, u64>>>,
+
     /// Replicated descriptor lease drain state.
     /// Written by the metadata applier on `DescriptorDrainStart`
     /// / `DescriptorDrainEnd` raft entries (and implicitly

nodedb/src/control/state/init.rs

@@ -185,6 +185,7 @@ impl SharedState {
             ts_partition_registries: Some(Mutex::new(std::collections::HashMap::new())),
             cold_storage: None,
             hlc_clock: Arc::new(nodedb_types::HlcClock::new()),
+            tenant_write_hlc: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
             lease_drain: Arc::new(crate::control::lease::DescriptorDrainTracker::new()),
             lease_refcount: Arc::new(crate::control::lease::LeaseRefCount::new()),
             tuning: TuningConfig::default(),
@@ -443,6 +444,7 @@ impl SharedState {
             ts_partition_registries: Some(Mutex::new(std::collections::HashMap::new())),
             cold_storage: None,
             hlc_clock: Arc::new(nodedb_types::HlcClock::new()),
+            tenant_write_hlc: Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
             lease_drain: Arc::new(crate::control::lease::DescriptorDrainTracker::new()),
             lease_refcount: Arc::new(crate::control::lease::LeaseRefCount::new()),
             tuning,

nodedb/src/control/state/methods.rs

@@ -10,6 +10,23 @@ use crate::types::TenantId;
 use super::SharedState;
 
 impl SharedState {
+    /// Advance the per-tenant observed write-HLC high-water to the current
+    /// HLC wall time. Idempotent and monotonic: no-op if a larger value is
+    /// already recorded. Callers MUST invoke this only after a successful
+    /// dispatch; "success" is defined as `Response.status == Status::Ok`
+    /// (and, for `Result<Response>` callers, `Result::Ok` as well). A
+    /// poisoned lock is silently ignored — the high-water is best-effort
+    /// and the RESTORE staleness gate treats missing entries as zero.
+    pub fn advance_tenant_write_hlc(&self, tenant_id: u32) {
+        let wall = self.hlc_clock.now().wall_ns;
+        if let Ok(mut map) = self.tenant_write_hlc.lock() {
+            let entry = map.entry(tenant_id).or_insert(0);
+            if wall > *entry {
+                *entry = wall;
+            }
+        }
+    }
+
     /// Shared HTTP client reused by every outbound emitter. Cloning the
     /// Arc is cheap — the client itself owns a connection pool, DNS
     /// resolver, and TLS session cache that every caller benefits from.