ito_core/
event_forwarder.rs

1//! Client-side forwarding of locally produced audit events to the backend.
2//!
3//! The forwarder reads new events from the local JSONL audit log, batches
4//! them, and submits each batch to the backend event ingest endpoint with
5//! an idempotency key so retries are safe. A checkpoint file under
6//! `.ito/.state/` tracks the last forwarded line offset to avoid
7//! re-sending the entire log on each invocation.
8
9use std::path::Path;
10use std::thread;
11use std::time::Duration;
12
13use ito_domain::audit::event::AuditEvent;
14use ito_domain::backend::{BackendError, BackendEventIngestClient, EventBatch};
15
16use crate::backend_client::{idempotency_key, is_retriable_status};
17use crate::errors::{CoreError, CoreResult};
18
19/// Maximum events per batch submission.
20const DEFAULT_BATCH_SIZE: usize = 100;
21
22/// Checkpoint file name within `.ito/.state/`.
23const CHECKPOINT_FILE: &str = "event-forward-offset";
24
25/// Result of a forwarding run, used for CLI diagnostics.
26#[derive(Debug, Clone, PartialEq, Eq)]
27pub struct ForwardResult {
28    /// Total events forwarded in this run.
29    pub forwarded: usize,
30    /// Total events skipped as duplicates by the backend.
31    pub duplicates: usize,
32    /// Number of batches that failed after exhausting retries.
33    pub failed_batches: usize,
34    /// Total events in the local log.
35    pub total_local: usize,
36    /// Offset after forwarding (line count forwarded so far).
37    pub new_offset: usize,
38}
39
40/// Configuration for the event forwarder.
41#[derive(Debug, Clone)]
42pub struct ForwarderConfig {
43    /// Maximum events per batch.
44    pub batch_size: usize,
45    /// Maximum retry attempts per batch on transient failure.
46    pub max_retries: u32,
47    /// Base delay between retries (doubles on each attempt).
48    pub retry_base_delay: Duration,
49}
50
51impl Default for ForwarderConfig {
52    fn default() -> Self {
53        Self {
54            batch_size: DEFAULT_BATCH_SIZE,
55            max_retries: 3,
56            retry_base_delay: Duration::from_millis(500),
57        }
58    }
59}
60
61/// Forward new local audit events to the backend.
62///
63/// Reads the audit log, skips events already forwarded (tracked by offset),
64/// batches new events, and submits each batch with an idempotency key.
65/// Updates the checkpoint file on success.
66///
67/// This function is designed to be called best-effort from the CLI after
68/// command completion in backend mode.
69pub fn forward_events(
70    ingest_client: &dyn BackendEventIngestClient,
71    ito_path: &Path,
72    config: &ForwarderConfig,
73) -> CoreResult<ForwardResult> {
74    let all_events = read_all_events(ito_path);
75    let total_local = all_events.len();
76
77    let mut current_offset = read_checkpoint(ito_path);
78    if current_offset > total_local {
79        current_offset = total_local;
80    }
81
82    if current_offset >= total_local {
83        return Ok(ForwardResult {
84            forwarded: 0,
85            duplicates: 0,
86            failed_batches: 0,
87            total_local,
88            new_offset: current_offset,
89        });
90    }
91
92    let batch_size = config.batch_size.max(1);
93    let new_events = &all_events[current_offset..];
94    let mut forwarded = 0usize;
95    let mut duplicates = 0usize;
96    let mut failed_batches = 0usize;
97    let mut offset = current_offset;
98
99    for chunk in new_events.chunks(batch_size) {
100        let batch = EventBatch {
101            events: chunk.to_vec(),
102            idempotency_key: idempotency_key("event-forward"),
103        };
104
105        match submit_with_retry(ingest_client, &batch, config) {
106            Ok(result) => {
107                forwarded += result.accepted;
108                duplicates += result.duplicates;
109                offset += chunk.len();
110                // Persist offset after each successful batch
111                if let Err(e) = write_checkpoint(ito_path, offset) {
112                    tracing::warn!("failed to write forwarding checkpoint: {e}");
113                }
114            }
115            Err(e) => {
116                tracing::warn!("event forwarding batch failed: {e}");
117                failed_batches += 1;
118                // Stop forwarding on failure to preserve ordering
119                break;
120            }
121        }
122    }
123
124    Ok(ForwardResult {
125        forwarded,
126        duplicates,
127        failed_batches,
128        total_local,
129        new_offset: offset,
130    })
131}
132
133/// Submit a batch with bounded retries on transient failures.
134fn submit_with_retry(
135    client: &dyn BackendEventIngestClient,
136    batch: &EventBatch,
137    config: &ForwarderConfig,
138) -> CoreResult<ito_domain::backend::EventIngestResult> {
139    let mut attempts = 0u32;
140    loop {
141        match client.ingest(batch) {
142            Ok(result) => return Ok(result),
143            Err(err) => {
144                attempts += 1;
145                if !is_retriable_backend_error(&err) || attempts > config.max_retries {
146                    return Err(backend_ingest_error_to_core(err));
147                }
148                // Exponential backoff
149                let delay = config.retry_base_delay * 2u32.saturating_pow(attempts - 1);
150                thread::sleep(delay);
151            }
152        }
153    }
154}
155
156/// Check if a backend error is transient and worth retrying.
157fn is_retriable_backend_error(err: &BackendError) -> bool {
158    match err {
159        BackendError::Unavailable(_) => true,
160        BackendError::Other(msg) => {
161            // Check for HTTP status codes embedded in the message
162            if let Some(code_str) = msg.strip_prefix("HTTP ")
163                && let Ok(code) = code_str
164                    .chars()
165                    .take_while(|c| c.is_ascii_digit())
166                    .collect::<String>()
167                    .parse::<u16>()
168            {
169                return is_retriable_status(code);
170            }
171            false
172        }
173        BackendError::Unauthorized(_) => false,
174        BackendError::NotFound(_) => false,
175        BackendError::LeaseConflict(_) => false,
176        BackendError::RevisionConflict(_) => false,
177    }
178}
179
180/// Convert a backend ingest error to a `CoreError`.
181fn backend_ingest_error_to_core(err: BackendError) -> CoreError {
182    match err {
183        BackendError::Unavailable(msg) => CoreError::process(format!(
184            "Backend unavailable during event forwarding: {msg}"
185        )),
186        BackendError::Unauthorized(msg) => CoreError::validation(format!(
187            "Backend auth failed during event forwarding: {msg}"
188        )),
189        BackendError::NotFound(msg) => {
190            CoreError::not_found(format!("Backend ingest endpoint not found: {msg}"))
191        }
192        BackendError::Other(msg) => {
193            CoreError::process(format!("Backend error during event forwarding: {msg}"))
194        }
195        BackendError::LeaseConflict(c) => CoreError::process(format!(
196            "Unexpected lease conflict during event forwarding: {}",
197            c.change_id
198        )),
199        BackendError::RevisionConflict(c) => CoreError::process(format!(
200            "Unexpected revision conflict during event forwarding: {}",
201            c.change_id
202        )),
203    }
204}
205
206// ── Checkpoint persistence ─────────────────────────────────────────
207
208/// Path to the forwarding checkpoint file.
209fn checkpoint_path(ito_path: &Path) -> std::path::PathBuf {
210    ito_path.join(".state").join(CHECKPOINT_FILE)
211}
212
213/// Read the current forwarding offset from the checkpoint file.
214///
215/// Returns 0 if the file does not exist or is malformed.
216fn read_checkpoint(ito_path: &Path) -> usize {
217    let path = checkpoint_path(ito_path);
218    let Ok(content) = std::fs::read_to_string(&path) else {
219        return 0;
220    };
221    content.trim().parse::<usize>().unwrap_or(0)
222}
223
224/// Write the forwarding offset to the checkpoint file.
225fn write_checkpoint(ito_path: &Path, offset: usize) -> CoreResult<()> {
226    let path = checkpoint_path(ito_path);
227    if let Some(parent) = path.parent() {
228        std::fs::create_dir_all(parent)
229            .map_err(|e| CoreError::io("creating checkpoint directory", e))?;
230    }
231    std::fs::write(&path, offset.to_string())
232        .map_err(|e| CoreError::io("writing forwarding checkpoint", e))
233}
234
235// ── Event reading ──────────────────────────────────────────────────
236
237/// Read all audit events from the routed local audit store.
238fn read_all_events(ito_path: &Path) -> Vec<AuditEvent> {
239    crate::audit::default_audit_store(ito_path).read_all()
240}
241
242#[cfg(test)]
243#[path = "event_forwarder_tests.rs"]
244mod event_forwarder_tests;
ito_core/event_forwarder.rs

ito_core/
event_forwarder.rs
