fix(cluster): stop blocking surrogate hwm flush on raft propose

propose_surrogate_hwm blocks for up to DEFAULT_PROPOSE_TIMEOUT (5s)
waiting for the entry to commit, but surrogate assignment also runs on
the Raft apply loop, so a flush triggered from that path parked the
very loop that had to commit the entry, stalling until the timeout
fired. That stall was what made Lite's sync deltas time out before
Origin could ack them.

Dispatch the propose on the current Tokio handle instead of awaiting
it inline. The local write is already durable via the catalog and WAL
before the propose runs, and apply_surrogate_alloc's restore_hwm is
idempotent and monotonic, so out-of-order or duplicate delivery of the
proposed watermark is safe.

Author: farhan-syah

nodedb/src/control/surrogate/assign/core/flush.rs

@@ -3,6 +3,8 @@
 //! Local flush-trigger + durable checkpoint persistence for
 //! [`super::SurrogateAssigner`].
 
+use std::sync::Arc;
+
 use super::super::super::persist::SurrogateHwmPersist;
 use super::super::super::registry::SurrogateRegistry;
 use super::super::super::wal_appender::SurrogateWalAppender;
@@ -47,11 +49,10 @@ impl SurrogateAssigner {
             return Ok(());
         }
         if registry.should_flush() {
-            let raft_shared = self.shared.get().and_then(|w| w.upgrade());
             let combined = CombinedPersist {
                 catalog,
                 wal_appender: self.wal_appender.as_ref(),
-                raft_shared: raft_shared.as_deref(),
+                raft_shared: self.shared.get().and_then(|w| w.upgrade()),
             };
             registry.flush(&combined)?;
         }
@@ -67,23 +68,34 @@ struct CombinedPersist<'a> {
     catalog: &'a SystemCatalog,
     wal_appender: &'a dyn SurrogateWalAppender,
     /// Present when the Raft cluster is active; drives the Raft propose.
-    raft_shared: Option<&'a SharedState>,
+    raft_shared: Option<Arc<SharedState>>,
 }
 
 impl SurrogateHwmPersist for CombinedPersist<'_> {
     fn checkpoint(&self, hwm: u32) -> crate::Result<()> {
         self.catalog.put_surrogate_hwm(hwm)?;
         self.wal_appender.record_alloc_to_wal(hwm)?;
-        // Propose to Raft when in cluster mode so followers advance
-        // their in-memory HWM. Failure is non-fatal for the local
-        // write (which is already durable via the catalog and WAL);
-        // the follower will catch up on the next flush cycle or via
-        // snapshot. We log at warn so operators can detect systemic
-        // issues without breaking the local write path.
-        if let Some(shared) = self.raft_shared
-            && let Err(e) = crate::control::metadata_proposer::propose_surrogate_hwm(shared, hwm)
-        {
-            tracing::warn!(hwm, error = %e, "surrogate hwm raft propose failed; followers may lag");
+        // Propose to Raft when in cluster mode so followers advance their
+        // in-memory HWM. This is dispatched off the caller's thread and
+        // never awaited here: the local write is ALREADY durable via the
+        // catalog and WAL above, so the propose carries no correctness
+        // weight for it — it only advances peers' (and a future joiner's)
+        // view of the watermark.
+        //
+        // Awaiting it inline was a liveness bug. `propose_surrogate_hwm`
+        // blocks for `DEFAULT_PROPOSE_TIMEOUT` (5s) waiting for the entry
+        // to commit, and surrogate assignment runs on the Raft apply loop
+        // as well as on the coordinator — so a flush triggered from the
+        // apply path parked the very loop that had to commit the entry,
+        // and only unwound when the timeout fired. Every such write ate a
+        // 5s stall, which is what made Lite's sync deltas time out before
+        // Origin could ack them.
+        //
+        // Out-of-order or duplicate delivery is safe: `apply_surrogate_alloc`
+        // advances the watermark through `restore_hwm`, which is idempotent
+        // and monotonic and never moves the counter backwards.
+        if let Some(shared) = &self.raft_shared {
+            spawn_hwm_propose(Arc::clone(shared), hwm);
         }
         Ok(())
     }
@@ -92,3 +104,33 @@ impl SurrogateHwmPersist for CombinedPersist<'_> {
         self.catalog.get_surrogate_hwm()
     }
 }
+
+/// Dispatch the `SurrogateAlloc { hwm }` metadata propose without blocking
+/// the caller.
+///
+/// Spawned as a normal runtime task rather than via `spawn_blocking` because
+/// `propose_surrogate_hwm` uses `block_in_place` internally, which is only
+/// legal on a multi-threaded runtime worker.
+///
+/// A missing reactor means this checkpoint ran outside a Tokio context, where
+/// the propose could not have been issued at all. That is not silent data
+/// loss — the hwm is already durable in the catalog and WAL, and the next
+/// flush that does run under a reactor re-proposes the (higher) watermark —
+/// but it is logged so a node that never advances peer watermarks is
+/// diagnosable rather than invisible.
+fn spawn_hwm_propose(shared: Arc<SharedState>, hwm: u32) {
+    let Ok(handle) = tokio::runtime::Handle::try_current() else {
+        tracing::debug!(
+            hwm,
+            "surrogate hwm checkpoint ran outside a Tokio runtime; \
+             skipping the metadata propose (hwm is durable locally; \
+             the next flush under a reactor re-proposes it)"
+        );
+        return;
+    };
+    handle.spawn(async move {
+        if let Err(e) = crate::control::metadata_proposer::propose_surrogate_hwm(&shared, hwm) {
+            tracing::warn!(hwm, error = %e, "surrogate hwm raft propose failed; followers may lag");
+        }
+    });
+}