pgsync_core/
progress.rs

1//! Human-facing progress reporting, decoupled from logging.
2//!
3//! The engine and the bootstrap flow call these hooks; the CLI renders them for a
4//! terminal. The default implementation does nothing, so library users pay nothing.
5
6use crate::lsn::Lsn;
7use std::sync::Arc;
8use std::time::Duration;
9
10/// One applied source transaction, summarised with bounded memory.
11#[derive(Debug, Clone, Default)]
12pub struct TxnSummary {
13    pub xid: u32,
14    pub end_lsn: Lsn,
15    /// Per (table, operation) counts, in first-seen order.
16    pub ops: Vec<OpCount>,
17    /// Up to `MAX_KEYS` `table key=value` samples, only when key reporting is enabled.
18    pub keys: Vec<String>,
19    pub rows: u64,
20    /// Time from BEGIN to COMMIT on the target.
21    pub apply_time: Duration,
22}
23
24#[derive(Debug, Clone)]
25pub struct OpCount {
26    pub table: String,
27    pub kind: &'static str,
28    pub count: u64,
29}
30
31pub const MAX_KEYS: usize = 8;
32pub const MAX_OP_GROUPS: usize = 64;
33
34impl TxnSummary {
35    pub fn record(&mut self, table: &str, kind: &'static str) {
36        self.rows += 1;
37        if let Some(c) = self
38            .ops
39            .iter_mut()
40            .find(|c| c.table == table && c.kind == kind)
41        {
42            c.count += 1;
43        } else if self.ops.len() < MAX_OP_GROUPS {
44            self.ops.push(OpCount {
45                table: table.to_owned(),
46                kind,
47                count: 1,
48            });
49        }
50    }
51
52    pub fn record_key(&mut self, sample: String) {
53        if self.keys.len() < MAX_KEYS {
54            self.keys.push(sample);
55        }
56    }
57}
58
59pub trait Progress: Send + Sync {
60    /// A completed setup step (`✓ ...`).
61    fn step(&self, _message: &str) {}
62    /// A table copy started.
63    fn copy_started(&self, _table: &str) {}
64    /// A table copy finished.
65    fn copy_done(&self, _table: &str, _rows: u64, _bytes: u64, _elapsed: Duration) {}
66    /// Streaming started at `from`.
67    fn streaming(&self, _from: Lsn) {}
68    /// A transaction was committed on the target.
69    fn txn(&self, _summary: &TxnSummary) {}
70    /// A redelivered transaction was skipped.
71    fn skipped(&self, _xid: u32, _end_lsn: Lsn) {}
72    /// Whether key samples should be collected (costs allocations per row).
73    fn wants_keys(&self) -> bool {
74        false
75    }
76}
77
78pub struct NoProgress;
79impl Progress for NoProgress {}
80
81pub type ProgressHandle = Arc<dyn Progress>;
82
83pub fn none() -> ProgressHandle {
84    Arc::new(NoProgress)
85}
pgsync_core/progress.rs

pgsync_core/
progress.rs
