feat(sync): add producer fencing, token refresh backoff, and stale-timeout metrics

SyncClient / FlowController enhancements:
- Add `fenced` flag to SyncClient; set on `AckStatus::Fenced`. The
  push loop halts all outbound frames when fenced and waits for
  reconnect (which clears the flag).
- Add `producer_id` and `accepted_epoch` fields populated from the
  handshake ack; stamped on every outbound frame for idempotent-producer
  gating by Origin.
- Add per-array-ack coalescing: `pending_array_ack` is now keyed by
  array name so the highest-HLC ack for each array is retained and
  multiple arrays can be acked in one push tick.
- Add token refresh exponential backoff: `token_last_attempt_ms` and
  `token_refresh_backoff_ms` (initial 5 s, max 5 min) prevent hammering
  a failing token provider; `is_refresh_backoff_elapsed` gates each
  attempt.
- Add 30-second timeout on the token provider call so a hung provider
  cannot block the sync task.
- Add `stale_timeouts` counter to `SyncMetrics` and
  `FlowController::cleanup_stale_and_record` to evict timed-out
  in-flight entries and increment the metric in one step.

Author: farhan-syah

nodedb-lite/src/sync/client/delta.rs

@@ -34,6 +34,10 @@ impl SyncClient {
                 peer_id: self.peer_id,
                 mutation_id: delta.mutation_id,
                 device_valid_time_ms: Some(device_valid_time_ms),
+                // producer_id, epoch, and seq are overwritten with real producer/epoch/stable-seq in push_crdt_deltas.
+                producer_id: 0,
+                epoch: 0,
+                seq: 0,
             })
             .collect()
     }
@@ -133,12 +137,14 @@ mod tests {
                 collection: "orders".into(),
                 document_id: "o1".into(),
                 delta_bytes: vec![1, 2, 3],
+                seq: 0,
             },
             PendingDelta {
                 mutation_id: 2,
                 collection: "users".into(),
                 document_id: "u1".into(),
                 delta_bytes: vec![4, 5, 6],
+                seq: 0,
             },
         ];
 
@@ -159,6 +165,8 @@ mod tests {
                 mutation_id: 1,
                 lsn: 42,
                 clock_skew_warning_ms: None,
+                applied_seq: 0,
+                status: nodedb_types::sync::wire::AckStatus::Applied,
             })
             .await;
 
@@ -202,6 +210,7 @@ mod tests {
             collection: "test".into(),
             document_id: "d1".into(),
             delta_bytes,
+            seq: 0,
         }];
         let msgs = client.build_delta_pushes(&pending).await;
         assert_eq!(msgs[0].checksum, expected_crc);
@@ -225,18 +234,21 @@ mod tests {
                 collection: "a".into(),
                 document_id: "d1".into(),
                 delta_bytes: vec![1],
+                seq: 0,
             },
             PendingDelta {
                 mutation_id: 2,
                 collection: "a".into(),
                 document_id: "d2".into(),
                 delta_bytes: vec![2],
+                seq: 0,
             },
             PendingDelta {
                 mutation_id: 3,
                 collection: "a".into(),
                 document_id: "d3".into(),
                 delta_bytes: vec![3],
+                seq: 0,
             },
         ];
 
@@ -253,6 +265,8 @@ mod tests {
                 mutation_id: 1,
                 lsn: 10,
                 clock_skew_warning_ms: None,
+                applied_seq: 0,
+                status: nodedb_types::sync::wire::AckStatus::Applied,
             })
             .await;
         let msgs = client.build_delta_pushes(&pending).await;

nodedb-lite/src/sync/client/handshake.rs

@@ -47,9 +47,18 @@ impl SyncClient {
                 clock.advance(peer_id, counter);
             }
         }
+        drop(clock);
+
+        self.set_producer_id(ack.producer_id).await;
+        self.set_accepted_epoch(ack.accepted_epoch).await;
 
         *self.state.lock().await = SyncState::Connected;
-        tracing::info!(session = %ack.session_id, "sync handshake accepted");
+        tracing::info!(
+            session = %ack.session_id,
+            producer_id = ack.producer_id,
+            accepted_epoch = ack.accepted_epoch,
+            "sync handshake accepted"
+        );
         true
     }
 }
@@ -96,6 +105,8 @@ mod tests {
             error: None,
             fork_detected: false,
             server_wire_version: 1,
+            producer_id: 0,
+            accepted_epoch: 0,
         };
 
         assert!(client.handle_handshake_ack(&ack).await);
@@ -112,6 +123,8 @@ mod tests {
             error: Some("invalid token".into()),
             fork_detected: false,
             server_wire_version: 1,
+            producer_id: 0,
+            accepted_epoch: 0,
         };
 
         assert!(!client.handle_handshake_ack(&ack).await);

nodedb-lite/src/sync/client/receive.rs

@@ -9,9 +9,27 @@ use super::state::SyncClient;
 
 impl SyncClient {
     /// Process a ShapeSnapshot from Origin.
+    ///
+    /// Marks the shape as snapshot-loaded and re-bases the sequence tracker
+    /// so that gap detection restarts from the snapshot LSN rather than any
+    /// stale watermark from before the resync. Also clears the resync gate so
+    /// a future gap on this or another shape can trigger a new request.
     pub async fn handle_shape_snapshot(&self, msg: &ShapeSnapshotMsg) {
         let mut shapes = self.shapes.lock().await;
         shapes.mark_snapshot_loaded(&msg.shape_id, msg.snapshot_lsn);
+        drop(shapes);
+
+        // Re-base the per-shape LSN tracker at the snapshot watermark so the
+        // gap detector does not immediately fire again after the resync.
+        self.last_seen_lsn
+            .lock()
+            .await
+            .insert(msg.shape_id.clone(), msg.snapshot_lsn);
+
+        // Clear the resync gate so future gaps can trigger new requests.
+        *self.resync_requested.lock().await = false;
+        *self.pending_resync.lock().await = None;
+
         tracing::info!(
             shape_id = %msg.shape_id,
             lsn = msg.snapshot_lsn,
@@ -74,6 +92,7 @@ impl SyncClient {
                 },
                 from_mutation_id: last_lsn + 1,
                 collection: String::new(),
+                shape_id: shape_id.to_string(),
             });
         }
         tracker.insert(shape_id.to_string(), lsn);
@@ -97,17 +116,33 @@ impl SyncClient {
         self.pending_resync.lock().await.take()
     }
 
-    /// Store a pending `ArrayAck` (set by the dispatch path on successful apply).
+    /// Merge an `ArrayAck` into the pending-ack map.
     ///
-    /// A newer ack overwrites an older one: sending the highest applied HLC is
-    /// sufficient to advance Origin's GC frontier.
+    /// Per-array, only the ack with the highest HLC is kept — `ack_hlc_bytes` is
+    /// stored in the same 18-byte big-endian layout used by `nodedb_array::sync::Hlc`,
+    /// so byte-wise comparison gives the correct temporal ordering. This means no
+    /// ack is silently lost: if two acks arrive for the same array before the push
+    /// loop drains them, the one with the higher frontier is retained, which is
+    /// exactly what Origin needs to advance its GC cursor.
     pub async fn set_pending_array_ack(&self, msg: ArrayAckMsg) {
-        *self.pending_array_ack.lock().await = Some(msg);
+        let mut map = self.pending_array_ack.lock().await;
+        let entry = map.entry(msg.array.clone()).or_insert_with(|| msg.clone());
+        if msg.ack_hlc_bytes > entry.ack_hlc_bytes {
+            *entry = msg;
+        }
     }
 
-    /// Take the pending `ArrayAck` (consumed by the push loop).
-    pub async fn take_pending_array_ack(&self) -> Option<ArrayAckMsg> {
-        self.pending_array_ack.lock().await.take()
+    /// Drain all pending `ArrayAck`s (consumed by the push loop).
+    ///
+    /// Returns all per-array acks accumulated since the last drain.  The map is
+    /// cleared so new acks can accumulate for the next tick.
+    pub async fn drain_pending_array_acks(&self) -> Vec<ArrayAckMsg> {
+        let mut map = self.pending_array_ack.lock().await;
+        if map.is_empty() {
+            return Vec::new();
+        }
+        let drained: Vec<ArrayAckMsg> = map.drain().map(|(_, v)| v).collect();
+        drained
     }
 }
 
@@ -185,6 +220,98 @@ mod tests {
         assert!(client.check_sequence_gap("s1", 20).await.is_none());
     }
 
+    #[tokio::test]
+    async fn array_ack_merge_keeps_highest_hlc() {
+        let client = SyncClient::new(make_config(), 1);
+
+        // Lower HLC bytes first.
+        let lower = ArrayAckMsg {
+            array: "arr1".into(),
+            replica_id: 1,
+            ack_hlc_bytes: [0x00; 18],
+            applied_seq: 0,
+            status: nodedb_types::sync::wire::AckStatus::Applied,
+        };
+        let higher = ArrayAckMsg {
+            array: "arr1".into(),
+            replica_id: 1,
+            ack_hlc_bytes: [0xFF; 18],
+            applied_seq: 0,
+            status: nodedb_types::sync::wire::AckStatus::Applied,
+        };
+
+        client.set_pending_array_ack(lower).await;
+        client.set_pending_array_ack(higher.clone()).await;
+
+        let drained = client.drain_pending_array_acks().await;
+        assert_eq!(drained.len(), 1);
+        assert_eq!(drained[0].ack_hlc_bytes, higher.ack_hlc_bytes);
+    }
+
+    #[tokio::test]
+    async fn array_ack_merge_keeps_higher_over_lower() {
+        let client = SyncClient::new(make_config(), 1);
+
+        // Insert higher first, then lower — should still keep higher.
+        let mut higher_bytes = [0x00u8; 18];
+        higher_bytes[0] = 0x10;
+        let mut lower_bytes = [0x00u8; 18];
+        lower_bytes[0] = 0x01;
+
+        client
+            .set_pending_array_ack(ArrayAckMsg {
+                array: "arr2".into(),
+                replica_id: 1,
+                ack_hlc_bytes: higher_bytes,
+                applied_seq: 0,
+                status: nodedb_types::sync::wire::AckStatus::Applied,
+            })
+            .await;
+        client
+            .set_pending_array_ack(ArrayAckMsg {
+                array: "arr2".into(),
+                replica_id: 1,
+                ack_hlc_bytes: lower_bytes,
+                applied_seq: 0,
+                status: nodedb_types::sync::wire::AckStatus::Applied,
+            })
+            .await;
+
+        let drained = client.drain_pending_array_acks().await;
+        assert_eq!(drained.len(), 1);
+        assert_eq!(drained[0].ack_hlc_bytes, higher_bytes);
+    }
+
+    #[tokio::test]
+    async fn array_ack_merge_separate_arrays_both_drained() {
+        let client = SyncClient::new(make_config(), 1);
+
+        client
+            .set_pending_array_ack(ArrayAckMsg {
+                array: "arr_a".into(),
+                replica_id: 1,
+                ack_hlc_bytes: [0x01; 18],
+                applied_seq: 0,
+                status: nodedb_types::sync::wire::AckStatus::Applied,
+            })
+            .await;
+        client
+            .set_pending_array_ack(ArrayAckMsg {
+                array: "arr_b".into(),
+                replica_id: 1,
+                ack_hlc_bytes: [0x02; 18],
+                applied_seq: 0,
+                status: nodedb_types::sync::wire::AckStatus::Applied,
+            })
+            .await;
+
+        let mut drained = client.drain_pending_array_acks().await;
+        drained.sort_by(|a, b| a.array.cmp(&b.array));
+        assert_eq!(drained.len(), 2);
+        assert_eq!(drained[0].array, "arr_a");
+        assert_eq!(drained[1].array, "arr_b");
+    }
+
     #[tokio::test]
     async fn reset_sequence_tracking_clears_state() {
         let client = SyncClient::new(make_config(), 1);