appcore_sync/sync/
client.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: client.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/06/02 13:08:16 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/07/23 23:50:45 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Follower push client with bounded queue and retry behavior.
12
13use crate::sync::error::{SyncError, SyncResult};
14use crate::sync::outbox::{FileSyncOutbox, InMemorySyncOutbox, SyncOutbox};
15use crate::sync::retry::{SyncPushMetrics, SyncRetryPolicy};
16use crate::sync::transport::HttpSyncTransport;
17use crate::sync::types::SyncMessage;
18use parking_lot::Mutex;
19use std::path::PathBuf;
20use std::sync::Arc;
21use std::time::Duration;
22
23/// Pushes leader events to a follower over transport.
24#[derive(Clone)]
25pub struct FollowerSyncClient {
26    transport: HttpSyncTransport,
27    retry_policy: SyncRetryPolicy,
28    outbox: Arc<dyn SyncOutbox>,
29    flush_lock: Arc<Mutex<()>>,
30    metrics: Arc<Mutex<SyncPushMetrics>>,
31}
32
33impl std::fmt::Debug for FollowerSyncClient {
34    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
35        formatter
36            .debug_struct("FollowerSyncClient")
37            .field("retry_policy", &self.retry_policy)
38            .field("pending_len", &self.pending_len())
39            .field("metrics", &self.metrics())
40            .finish()
41    }
42}
43
44impl FollowerSyncClient {
45    /// Creates a follower client with bounded in-memory buffering and default retries.
46    pub fn new(transport: HttpSyncTransport) -> Self {
47        Self {
48            transport,
49            retry_policy: SyncRetryPolicy::default(),
50            outbox: Arc::new(InMemorySyncOutbox::new()),
51            flush_lock: Arc::new(Mutex::new(())),
52            metrics: Arc::new(Mutex::new(SyncPushMetrics::default())),
53        }
54    }
55
56    /// Replaces retry, backoff, and queue limits.
57    pub fn with_retry_policy(mut self, retry_policy: SyncRetryPolicy) -> Self {
58        self.retry_policy = retry_policy;
59        self
60    }
61
62    /// Returns the configured retry policy.
63    pub fn retry_policy(&self) -> SyncRetryPolicy {
64        self.retry_policy
65    }
66
67    /// Replaces the pending-message outbox implementation.
68    pub fn with_outbox(mut self, outbox: Arc<dyn SyncOutbox>) -> Self {
69        self.outbox = outbox;
70        self
71    }
72
73    /// Configures a durable file outbox at `file_path`.
74    pub fn with_file_outbox(self, file_path: impl Into<PathBuf>) -> SyncResult<Self> {
75        Ok(self.with_outbox(Arc::new(FileSyncOutbox::new(file_path)?)))
76    }
77
78    /// Returns a snapshot of cumulative push counters.
79    pub fn metrics(&self) -> SyncPushMetrics {
80        *self.metrics.lock()
81    }
82
83    /// Returns the pending count, or zero if the outbox cannot be read.
84    pub fn pending_len(&self) -> usize {
85        self.outbox.len().unwrap_or(0)
86    }
87
88    /// Returns pending batches in delivery order.
89    pub fn pending_messages(&self) -> SyncResult<Vec<SyncMessage>> {
90        self.outbox.messages()
91    }
92
93    /// Cancels active transport I/O and retry waits.
94    pub fn cancel(&self) {
95        self.transport.cancel();
96    }
97
98    /// Reports whether this client has been cancelled.
99    pub fn is_cancelled(&self) -> bool {
100        self.transport.is_cancelled()
101    }
102
103    /// Attempts delivery of all currently queued batches.
104    pub fn flush_pending(&self) -> SyncResult<()> {
105        self.flush_queue()
106    }
107
108    /// Enqueues a batch durably before attempting ordered delivery.
109    pub fn push_events(&self, message: &SyncMessage) -> SyncResult<()> {
110        if !self
111            .outbox
112            .try_enqueue(message.clone(), self.retry_policy.max_queue_len)?
113        {
114            let mut metrics = self.metrics.lock();
115            metrics.push_dropped += 1;
116            return Err(SyncError::TransportFailed("sync queue full".to_string()));
117        }
118        self.flush_queue()
119    }
120
121    fn flush_queue(&self) -> SyncResult<()> {
122        let _flush_guard = self.flush_lock.lock();
123        loop {
124            let message = match self.outbox.front()? {
125                Some(message) => message,
126                None => return Ok(()),
127            };
128            if self.try_send_with_retry(&message).is_ok() {
129                self.outbox.acknowledge_front(&message.batch_id)?;
130                let mut metrics = self.metrics.lock();
131                metrics.push_success += 1;
132                continue;
133            }
134            let mut metrics = self.metrics.lock();
135            metrics.push_failed += 1;
136            return Err(SyncError::TransportFailed(
137                "sync push retry exhausted".to_string(),
138            ));
139        }
140    }
141
142    fn try_send_with_retry(&self, message: &SyncMessage) -> SyncResult<()> {
143        let max_attempts = self.retry_policy.max_attempts.max(1);
144        for attempt in 1..=max_attempts {
145            if self.transport.is_cancelled() {
146                return Err(SyncError::TransportFailed(
147                    "sync push cancelled".to_string(),
148                ));
149            }
150            let mut metrics = self.metrics.lock();
151            metrics.push_attempt += 1;
152            drop(metrics);
153            if self.transport.post_sync_events(message).is_ok() {
154                return Ok(());
155            }
156            if attempt < max_attempts
157                && self.retry_policy.backoff_ms > 0
158                && self
159                    .transport
160                    .cancellation_token()
161                    .wait_timeout(Duration::from_millis(self.retry_policy.backoff_ms))
162            {
163                return Err(SyncError::TransportFailed(
164                    "sync push cancelled".to_string(),
165                ));
166            }
167        }
168        Err(SyncError::TransportFailed(
169            "sync push retry exhausted".to_string(),
170        ))
171    }
172}
appcore_sync/sync/client.rs

appcore_sync/sync/
client.rs
