feat(sync): implement delete-on-ack transport with idempotent frame stamping

Transport layer changes for the durable outbound queue model:

- Add `dispatch_acks` module: each ack type is handled in its own
  function that applies `AckStatus` (Applied/Duplicate → delete durable
  entry; Fenced → halt push; Gap → warn) and advances the stream seq
  frontier via `record_ack`.
- `dispatch_frame`: route all engine ack types to `dispatch_acks`;
  handle `AckStatus` on `DeltaAck` (fenced halt, gap warning).
- `connect_and_run`: reload persisted producer state before reconnect,
  clear engine in-flight maps so durable entries are eligible for
  re-drain, persist server-assigned producer state after handshake.
- Push loop: stamp every outbound frame with `producer_id`, `epoch`,
  and a stable per-collection `seq` (assigned once, reused on
  re-sends). Halt all push when fenced. Run stale in-flight cleanup
  with AIMD multiplicative decrease every tick.
- `SyncDelegate` trait: all pending/ack methods are now async; add
  `next_stream_seq`, `set_pending_delta_seq`, `persist_*_seq` for
  durable seq stamping; add `clear_engine_in_flight`, `load/persist
  _producer_state`.

Author: farhan-syah

nodedb-lite/src/sync/transport/connect.rs

@@ -25,9 +25,29 @@ pub(super) async fn connect_and_run(
     client: &Arc<SyncClient>,
     delegate: &Arc<dyn SyncDelegate>,
 ) -> Result<(), LiteError> {
-    // Reset state for a fresh connection.
+    // Reload durable producer identity so the outbound path has a valid
+    // producer_id/accepted_epoch before the handshake is built. This is a
+    // no-op on the very first connect (returns 0, 0) and a restore on
+    // reconnect (returns the last values Origin assigned).
+    let (persisted_producer_id, persisted_accepted_epoch) = delegate.load_producer_state().await;
+    client
+        .load_producer_state(persisted_producer_id, persisted_accepted_epoch)
+        .await;
+
+    // Reset per-connection inbound state (shape LSN gaps, flow control).
+    // The per-stream seq frontier (StreamSeqTracker) is NOT reset here —
+    // it is loaded once from storage at startup and never cleared on reconnect
+    // so outbound frame numbering resumes from the durable last_assigned.
+    // The fenced flag is cleared so this attempt can push; if Origin still
+    // fences the producer the flag will be set again on the first ack.
     client.reset_sequence_tracking().await;
     client.reset_flow_control().await;
+    client.clear_fenced();
+
+    // Clear in-flight maps for all engine outbound queues. The durable entries
+    // are still in storage; clearing in-flight makes them eligible for
+    // re-drain → re-send → Origin deduplicates via its idempotent gate.
+    delegate.clear_engine_in_flight().await;
 
     // ── Connect ──
     let (ws_stream, _response) = tokio_tungstenite::connect_async(&client.config().url)
@@ -93,6 +113,12 @@ pub(super) async fn connect_and_run(
         });
     }
 
+    // Durably persist the server-assigned producer identity so it survives
+    // process restart and is available on the next reconnect.
+    delegate
+        .persist_producer_state(ack.producer_id, ack.accepted_epoch)
+        .await;
+
     // ── Message loop ──
     let sink = Arc::new(Mutex::new(sink));
 

nodedb-lite/src/sync/transport/delegate.rs

@@ -28,6 +28,11 @@ use crate::sync::outbound::vector::{PendingVectorDelete, PendingVectorInsert};
 pub trait SyncDelegate: Send + Sync + 'static {
     /// Get all pending CRDT deltas to push to Origin.
     fn pending_deltas(&self) -> Vec<PendingDelta>;
+    /// Assign a stable stream seq to a pending CRDT delta (first-send only).
+    ///
+    /// No-op if the delta already has a non-zero seq, so the same seq is
+    /// reused on reconnect re-sends and Origin can deduplicate.
+    async fn set_pending_delta_seq(&self, mutation_id: u64, seq: u64);
     /// Acknowledge deltas up to the given mutation_id.
     fn acknowledge(&self, mutation_id: u64);
     /// Reject a specific delta (rollback optimistic state).
@@ -73,40 +78,189 @@ pub trait SyncDelegate: Send + Sync + 'static {
     fn handle_array_reject(&self, msg: &nodedb_types::sync::wire::ArrayRejectMsg);
 
     // ── Columnar ─────────────────────────────────────────────────────────────
-    fn pending_columnar_batches(&self) -> Vec<PendingColumnarBatch>;
-    fn acknowledge_columnar_batch(&self, batch_id: u64);
-    fn reject_columnar_batch(&self, batch: PendingColumnarBatch);
+    /// Drain up to `PUSH_DRAIN_LIMIT` pending batches from durable storage,
+    /// skipping any currently in-flight entries.
+    ///
+    /// Returns `(durable_key, batch)` pairs. On successful send, call
+    /// `mark_columnar_batch_in_flight`. The durable entry is deleted only when
+    /// Origin's ack arrives via `ack_columnar_batch_in_flight`.
+    async fn pending_columnar_batches(&self) -> Vec<(Vec<u8>, PendingColumnarBatch)>;
+    /// Record that a columnar batch has been sent and is awaiting Origin ack.
+    async fn mark_columnar_batch_in_flight(&self, batch_id: u64, durable_key: Vec<u8>);
+    /// On Origin ack: remove in-flight record and delete the durable entry.
+    async fn ack_columnar_batch_in_flight(&self, batch_id: u64);
+    /// Delete the durable entry for a confirmed batch directly (used for
+    /// un-encodable entries that must be discarded at send time).
+    async fn acknowledge_columnar_batch(&self, durable_key: Vec<u8>);
 
     // ── Vector ───────────────────────────────────────────────────────────────
-    fn pending_vector_inserts(&self) -> Vec<PendingVectorInsert>;
-    fn acknowledge_vector_insert(&self, batch_id: u64);
-    fn reject_vector_insert(&self, entry: PendingVectorInsert);
+    /// Drain up to `PUSH_DRAIN_LIMIT` pending insert entries, skipping in-flight.
+    async fn pending_vector_inserts(&self) -> Vec<(Vec<u8>, PendingVectorInsert)>;
+    /// Record that a vector insert has been sent and is awaiting Origin ack.
+    async fn mark_vector_insert_in_flight(&self, batch_id: u64, durable_key: Vec<u8>);
+    /// On Origin ack: remove in-flight record and delete the durable entry.
+    async fn ack_vector_insert_in_flight(&self, batch_id: u64);
+    /// Delete the durable insert entry directly (for un-encodable entries).
+    async fn acknowledge_vector_insert(&self, durable_key: Vec<u8>);
 
-    fn pending_vector_deletes(&self) -> Vec<PendingVectorDelete>;
-    fn acknowledge_vector_delete(&self, batch_id: u64);
-    fn reject_vector_delete(&self, entry: PendingVectorDelete);
+    /// Drain up to `PUSH_DRAIN_LIMIT` pending delete entries, skipping in-flight.
+    async fn pending_vector_deletes(&self) -> Vec<(Vec<u8>, PendingVectorDelete)>;
+    /// Record that a vector delete has been sent and is awaiting Origin ack.
+    async fn mark_vector_delete_in_flight(&self, batch_id: u64, durable_key: Vec<u8>);
+    /// On Origin ack: remove in-flight record and delete the durable entry.
+    async fn ack_vector_delete_in_flight(&self, batch_id: u64);
+    /// Delete the durable delete entry directly (for un-encodable entries).
+    async fn acknowledge_vector_delete(&self, durable_key: Vec<u8>);
 
     // ── FTS ──────────────────────────────────────────────────────────────────
-    fn pending_fts_indexes(&self) -> Vec<PendingFtsIndex>;
-    fn acknowledge_fts_index(&self, batch_id: u64);
-    fn reject_fts_index(&self, entry: PendingFtsIndex);
+    /// Drain up to `PUSH_DRAIN_LIMIT` pending index entries, skipping in-flight.
+    async fn pending_fts_indexes(&self) -> Vec<(Vec<u8>, PendingFtsIndex)>;
+    /// Record that an FTS index entry has been sent and is awaiting Origin ack.
+    async fn mark_fts_index_in_flight(&self, batch_id: u64, durable_key: Vec<u8>);
+    /// On Origin ack: remove in-flight record and delete the durable entry.
+    async fn ack_fts_index_in_flight(&self, batch_id: u64);
+    /// Delete the durable index entry directly (for un-encodable entries).
+    async fn acknowledge_fts_index(&self, durable_key: Vec<u8>);
 
-    fn pending_fts_deletes(&self) -> Vec<PendingFtsDelete>;
-    fn acknowledge_fts_delete(&self, batch_id: u64);
-    fn reject_fts_delete(&self, entry: PendingFtsDelete);
+    /// Drain up to `PUSH_DRAIN_LIMIT` pending FTS delete entries, skipping in-flight.
+    async fn pending_fts_deletes(&self) -> Vec<(Vec<u8>, PendingFtsDelete)>;
+    /// Record that an FTS delete entry has been sent and is awaiting Origin ack.
+    async fn mark_fts_delete_in_flight(&self, batch_id: u64, durable_key: Vec<u8>);
+    /// On Origin ack: remove in-flight record and delete the durable entry.
+    async fn ack_fts_delete_in_flight(&self, batch_id: u64);
+    /// Delete the durable delete entry directly (for un-encodable entries).
+    async fn acknowledge_fts_delete(&self, durable_key: Vec<u8>);
 
     // ── Spatial ──────────────────────────────────────────────────────────────
-    fn pending_spatial_inserts(&self) -> Vec<PendingSpatialInsert>;
-    fn acknowledge_spatial_insert(&self, batch_id: u64);
-    fn reject_spatial_insert(&self, entry: PendingSpatialInsert);
+    /// Drain up to `PUSH_DRAIN_LIMIT` pending insert entries, skipping in-flight.
+    async fn pending_spatial_inserts(&self) -> Vec<(Vec<u8>, PendingSpatialInsert)>;
+    /// Record that a spatial insert has been sent and is awaiting Origin ack.
+    async fn mark_spatial_insert_in_flight(&self, batch_id: u64, durable_key: Vec<u8>);
+    /// On Origin ack: remove in-flight record and delete the durable entry.
+    async fn ack_spatial_insert_in_flight(&self, batch_id: u64);
+    /// Delete the durable insert entry directly (for un-encodable entries).
+    async fn acknowledge_spatial_insert(&self, durable_key: Vec<u8>);
 
-    fn pending_spatial_deletes(&self) -> Vec<PendingSpatialDelete>;
-    fn acknowledge_spatial_delete(&self, batch_id: u64);
-    fn reject_spatial_delete(&self, entry: PendingSpatialDelete);
+    /// Drain up to `PUSH_DRAIN_LIMIT` pending spatial delete entries, skipping in-flight.
+    async fn pending_spatial_deletes(&self) -> Vec<(Vec<u8>, PendingSpatialDelete)>;
+    /// Record that a spatial delete has been sent and is awaiting Origin ack.
+    async fn mark_spatial_delete_in_flight(&self, batch_id: u64, durable_key: Vec<u8>);
+    /// On Origin ack: remove in-flight record and delete the durable entry.
+    async fn ack_spatial_delete_in_flight(&self, batch_id: u64);
+    /// Delete the durable delete entry directly (for un-encodable entries).
+    async fn acknowledge_spatial_delete(&self, durable_key: Vec<u8>);
 
     // ── Timeseries ───────────────────────────────────────────────────────────
-    fn pending_timeseries_batches(&self) -> Vec<PendingTimeseriesBatch>;
-    /// Acknowledge all pending batches for a collection (Origin confirmed receipt).
-    fn acknowledge_timeseries_collection(&self, collection: &str);
-    fn reject_timeseries_batch(&self, batch: PendingTimeseriesBatch);
+    /// Drain up to `PUSH_DRAIN_LIMIT` pending batches, skipping in-flight entries.
+    async fn pending_timeseries_batches(&self) -> Vec<(Vec<u8>, PendingTimeseriesBatch)>;
+    /// Record that a timeseries batch has been sent and is awaiting Origin ack.
+    ///
+    /// Keyed by `stream_seq` because `TimeseriesAckMsg` echoes `applied_seq`
+    /// but not `batch_id`.
+    async fn mark_timeseries_batch_in_flight(&self, stream_seq: u64, durable_key: Vec<u8>);
+    /// On Origin ack: delete all durable entries whose seq ≤ `applied_seq`.
+    async fn ack_timeseries_batches_through_seq(&self, applied_seq: u64);
+    /// Delete the durable entry directly (for empty/un-encodable batches).
+    async fn acknowledge_timeseries_batch(&self, durable_key: Vec<u8>);
+
+    // ── Stable seq persistence ────────────────────────────────────────────────
+
+    /// Persist an assigned stream seq into the durable columnar entry at `key`.
+    ///
+    /// Must be called before sending the frame. If this returns an error the
+    /// caller must NOT send — it should retain the entry for the next drain tick.
+    async fn persist_columnar_seq(
+        &self,
+        key: &[u8],
+        batch: &PendingColumnarBatch,
+    ) -> Result<(), crate::error::LiteError>;
+
+    /// Persist an assigned stream seq into the durable timeseries entry at `key`.
+    async fn persist_timeseries_seq(
+        &self,
+        key: &[u8],
+        batch: &PendingTimeseriesBatch,
+    ) -> Result<(), crate::error::LiteError>;
+
+    /// Persist an assigned stream seq into the durable vector insert entry at `key`.
+    async fn persist_vector_insert_seq(
+        &self,
+        key: &[u8],
+        insert: &PendingVectorInsert,
+    ) -> Result<(), crate::error::LiteError>;
+
+    /// Persist an assigned stream seq into the durable vector delete entry at `key`.
+    async fn persist_vector_delete_seq(
+        &self,
+        key: &[u8],
+        delete: &PendingVectorDelete,
+    ) -> Result<(), crate::error::LiteError>;
+
+    /// Persist an assigned stream seq into the durable FTS index entry at `key`.
+    async fn persist_fts_index_seq(
+        &self,
+        key: &[u8],
+        entry: &PendingFtsIndex,
+    ) -> Result<(), crate::error::LiteError>;
+
+    /// Persist an assigned stream seq into the durable FTS delete entry at `key`.
+    async fn persist_fts_delete_seq(
+        &self,
+        key: &[u8],
+        entry: &PendingFtsDelete,
+    ) -> Result<(), crate::error::LiteError>;
+
+    /// Persist an assigned stream seq into the durable spatial insert entry at `key`.
+    async fn persist_spatial_insert_seq(
+        &self,
+        key: &[u8],
+        insert: &PendingSpatialInsert,
+    ) -> Result<(), crate::error::LiteError>;
+
+    /// Persist an assigned stream seq into the durable spatial delete entry at `key`.
+    async fn persist_spatial_delete_seq(
+        &self,
+        key: &[u8],
+        delete: &PendingSpatialDelete,
+    ) -> Result<(), crate::error::LiteError>;
+
+    // ── Reconnect ────────────────────────────────────────────────────────────
+
+    /// Clear all engine in-flight maps on reconnect.
+    ///
+    /// The durable entries are still in storage and will be re-drained on the
+    /// next push tick. Origin's idempotent gate deduplicates re-sent batches.
+    async fn clear_engine_in_flight(&self);
+
+    // ── Producer state ───────────────────────────────────────────────────────
+
+    /// Durably persist the server-assigned `producer_id` and `accepted_epoch`
+    /// so they survive process restart and can be reloaded on reconnect.
+    ///
+    /// Called after every successful handshake acknowledgement.
+    async fn persist_producer_state(&self, producer_id: u64, accepted_epoch: u64);
+
+    /// Load the last-persisted `producer_id` and `accepted_epoch`.
+    ///
+    /// Returns `(0, 0)` if no state has been persisted yet (first run).
+    /// Called at the start of each connection attempt so the client has its
+    /// identity available before the outbound handshake is built.
+    async fn load_producer_state(&self) -> (u64, u64);
+
+    // ── Stream sequence ──────────────────────────────────────────────────────
+
+    /// Assign the next monotonic sequence number for the given `stream_id`.
+    ///
+    /// Delegates to `StreamSeqTracker::next_seq` which persists before
+    /// returning (persist-before-send invariant). On storage error, logs a
+    /// warning and returns `0` — a zero seq applies unconditionally on Origin,
+    /// so this is safe degradation with no data loss.
+    async fn next_stream_seq(&self, stream_id: u64) -> u64;
+
+    /// Record that Origin has applied `applied_seq` for `stream_id`.
+    ///
+    /// Delegates to `StreamSeqTracker::record_ack`. Errors are logged and
+    /// ignored — the last_assigned frontier already prevents re-sending
+    /// un-acked seqs, so this is a refinement of the acknowledged frontier.
+    async fn record_stream_ack(&self, stream_id: u64, applied_seq: u64);
 }