nodedb_lite/sync/array/
ack_sender.rs

1//! [`ArrayAckSender`] — periodic ack sender from Lite to Origin.
2//!
3//! A Tokio task (spawned on sync-session start, shut down on disconnect)
4//! that runs every `interval` seconds and sends `ArrayAckMsg` for each known
5//! array to Origin. Origin merges these into its `ArrayAckRegistry` to advance
6//! the GC frontier.
7//!
8//! # What gets acked
9//!
10//! For each registered array the ack carries `last_applied_hlc` — the highest
11//! HLC successfully applied by `LiteApplyEngine`. This is the correct value
12//! because it represents "I have all ops up to this HLC" from Origin's
13//! perspective.
14//!
15//! # Transport
16//!
17//! Frames are sent via a `tokio::sync::mpsc::Sender<Vec<u8>>` that the
18//! session's WebSocket write loop drains. This matches the pattern used by
19//! other Lite sync subsystems.
20
21use std::sync::Arc;
22use std::time::Duration;
23
24use nodedb_array::sync::replica_id::ReplicaId;
25use nodedb_types::sync::wire::SyncMessageType;
26use nodedb_types::sync::wire::array::ArrayAckMsg;
27use tokio::sync::mpsc;
28use tokio::task::JoinHandle;
29use tracing::warn;
30
31use crate::storage::engine::StorageEngine;
32
33use super::inbound::apply::LiteApplyEngine;
34use super::schema_registry::SchemaRegistry;
35
36/// Default ack interval: 30 seconds.
37pub const DEFAULT_ACK_INTERVAL: Duration = Duration::from_secs(30);
38
39/// Periodic array ack sender task.
40///
41/// Spawned on session connect and cancelled on disconnect. The returned
42/// `JoinHandle` should be stored by the session and aborted (via
43/// `handle.abort()`) on session teardown.
44pub fn spawn<S: StorageEngine + 'static>(
45    schemas: Arc<SchemaRegistry<S>>,
46    engine: Arc<LiteApplyEngine<S>>,
47    replica_id: ReplicaId,
48    tx: mpsc::Sender<Vec<u8>>,
49    interval: Duration,
50    mut stop: tokio::sync::oneshot::Receiver<()>,
51) -> JoinHandle<()> {
52    tokio::spawn(async move {
53        let mut ticker = tokio::time::interval(interval);
54        // Skip the immediate first tick so we don't ack before applying anything.
55        ticker.tick().await;
56
57        loop {
58            tokio::select! {
59                _ = ticker.tick() => {
60                    send_acks(&schemas, &engine, replica_id, &tx).await;
61                }
62                _ = &mut stop => {
63                    return;
64                }
65            }
66        }
67    })
68}
69
70/// Build and send `ArrayAckMsg` frames for all known arrays.
71async fn send_acks<S: StorageEngine>(
72    schemas: &SchemaRegistry<S>,
73    engine: &LiteApplyEngine<S>,
74    replica_id: ReplicaId,
75    tx: &mpsc::Sender<Vec<u8>>,
76) {
77    let arrays = schemas.list_arrays();
78    for array in &arrays {
79        let Some(ack_hlc) = engine.last_applied_hlc(array) else {
80            // No ops applied for this array yet — nothing to ack.
81            continue;
82        };
83
84        let msg = ArrayAckMsg {
85            array: array.clone(),
86            replica_id: replica_id.as_u64(),
87            ack_hlc_bytes: ack_hlc.to_bytes(),
88            applied_seq: 0,
89            status: nodedb_types::sync::wire::AckStatus::Applied,
90        };
91
92        let frame = match nodedb_types::sync::wire::SyncFrame::try_encode(
93            SyncMessageType::ArrayAck,
94            &msg,
95        ) {
96            Some(f) => f.to_bytes(),
97            None => {
98                warn!(
99                    array = %array,
100                    "ack_sender: SyncFrame encode failed for ArrayAck"
101                );
102                continue;
103            }
104        };
105
106        if tx.send(frame).await.is_err() {
107            // Transport closed — session is shutting down.
108            return;
109        }
110    }
111}
nodedb_lite/sync/array/ack_sender.rs

nodedb_lite/sync/array/
ack_sender.rs
