kevy_embedded/metric.rs
1//! Optional push-style metric callback. In-process embed mode has no metrics
2//! endpoint, so persistence events (AOF replay on startup, AOF rewrite/
3//! compaction) are pushed to a caller-supplied sink — wire it to Prometheus,
4//! a log line, a counter, whatever. Wire it via [`crate::Config::with_metric_sink`].
5
6use std::path::PathBuf;
7
8/// What `Store::open` restored — and what it could not. The pull-style
9/// twin of [`KevyMetric::Replay`] (`Store::open_report()`), so a host can
10/// turn "the AOF lost bytes at boot" into a health-check verdict without
11/// wiring a metric sink or scraping stderr.
12///
13/// ```
14/// let s = kevy_embedded::Store::open(kevy_embedded::Config::default())?;
15/// let r = s.open_report();
16/// assert_eq!((r.replayed_commands, r.dropped_bytes, r.corrupt), (0, 0, false));
17/// # Ok::<(), kevy_embedded::KevyError>(())
18/// ```
19#[derive(Debug, Clone, Default)]
20#[non_exhaustive]
21pub struct OpenReport {
22 /// Commands replayed from the AOF(s), summed across shards.
23 ///
24 /// ```
25 /// # use kevy_embedded::{AppendFsync, Config, Store};
26 /// # let dir = kevy_tmpdir::TmpDir::new("open-report-doc");
27 /// # let cfg = || Config::default().with_persist(dir.path()).with_appendfsync(AppendFsync::Always);
28 /// # let s = Store::open(cfg())?;
29 /// # s.set(b"a", b"1")?;
30 /// # s.set(b"b", b"2")?;
31 /// # drop(s);
32 /// # let s = Store::open(cfg())?;
33 /// # let r = s.open_report();
34 /// assert_eq!(r.replayed_commands, 2); // the two SETs came back from the log
35 /// # Ok::<(), Box<dyn std::error::Error>>(())
36 /// ```
37 pub replayed_commands: u64,
38 /// Bytes actually replayed (the valid prefixes).
39 ///
40 /// ```
41 /// # use kevy_embedded::{AppendFsync, Config, Store};
42 /// # let dir = kevy_tmpdir::TmpDir::new("open-report-doc");
43 /// # let cfg = || Config::default().with_persist(dir.path()).with_appendfsync(AppendFsync::Always);
44 /// # let s = Store::open(cfg())?;
45 /// # s.set(b"a", b"1")?;
46 /// # s.set(b"b", b"2")?;
47 /// # drop(s);
48 /// # let s = Store::open(cfg())?;
49 /// # let r = s.open_report();
50 /// let log = std::fs::metadata(dir.path().join("aof-0.aof"))?.len();
51 /// assert!(r.replayed_bytes > 0 && r.replayed_bytes <= log);
52 /// # Ok::<(), Box<dyn std::error::Error>>(())
53 /// ```
54 pub replayed_bytes: u64,
55 /// Wall-clock time of the whole startup replay, in milliseconds.
56 ///
57 /// ```
58 /// # use kevy_embedded::{AppendFsync, Config, Store};
59 /// # let dir = kevy_tmpdir::TmpDir::new("open-report-doc");
60 /// # let cfg = || Config::default().with_persist(dir.path()).with_appendfsync(AppendFsync::Always);
61 /// # let s = Store::open(cfg())?;
62 /// # s.set(b"a", b"1")?;
63 /// # s.set(b"b", b"2")?;
64 /// # drop(s);
65 /// # let s = Store::open(cfg())?;
66 /// # let r = s.open_report();
67 /// assert!(r.elapsed_ms < 60_000); // a two-record log replays at once
68 /// # Ok::<(), Box<dyn std::error::Error>>(())
69 /// ```
70 pub elapsed_ms: u64,
71 /// Bytes dropped past the last replayable frame, summed across shards.
72 /// Non-zero = the store recovered less than the files held.
73 ///
74 /// ```
75 /// # use kevy_embedded::{AppendFsync, Config, Store};
76 /// # let dir = kevy_tmpdir::TmpDir::new("open-report-damaged-doc");
77 /// # let cfg = || Config::default().with_persist(dir.path()).with_appendfsync(AppendFsync::Always);
78 /// # Store::open(cfg())?.set(b"a", b"1")?;
79 /// // a record of length 4 whose checksum is wrong, then bytes behind it
80 /// let mut log = std::fs::OpenOptions::new().append(true).open(dir.path().join("aof-0.aof"))?;
81 /// std::io::Write::write_all(&mut log, &[4, 0, 0, 0, 0, 0, 0, 0, b'a', b'b', b'c', b'd', b'!'])?;
82 /// drop(log);
83 /// let s = Store::open(cfg())?;
84 /// let r = s.open_report();
85 /// assert_eq!(r.dropped_bytes, 13); // everything from the bad record on
86 /// assert_eq!(s.get(b"a")?.as_deref(), Some(&b"1"[..])); // the good prefix survives
87 /// # Ok::<(), Box<dyn std::error::Error>>(())
88 /// ```
89 pub dropped_bytes: u64,
90 /// True when any shard's replay stopped at a corrupt frame.
91 ///
92 /// ```
93 /// # use kevy_embedded::{AppendFsync, Config, Store};
94 /// # let dir = kevy_tmpdir::TmpDir::new("open-report-damaged-doc");
95 /// # let cfg = || Config::default().with_persist(dir.path()).with_appendfsync(AppendFsync::Always);
96 /// # Store::open(cfg())?.set(b"a", b"1")?;
97 /// // a record of length 4 whose checksum is wrong, then bytes behind it
98 /// let mut log = std::fs::OpenOptions::new().append(true).open(dir.path().join("aof-0.aof"))?;
99 /// std::io::Write::write_all(&mut log, &[4, 0, 0, 0, 0, 0, 0, 0, b'a', b'b', b'c', b'd', b'!'])?;
100 /// drop(log);
101 /// let s = Store::open(cfg())?;
102 /// let r = s.open_report();
103 /// assert!(r.corrupt); // stopped at a bad record, not at a torn final one
104 /// # Ok::<(), Box<dyn std::error::Error>>(())
105 /// ```
106 pub corrupt: bool,
107 /// Quarantine files written while repairing dropped tails (one per
108 /// affected shard).
109 ///
110 /// ```
111 /// # use kevy_embedded::{AppendFsync, Config, Store};
112 /// # let dir = kevy_tmpdir::TmpDir::new("open-report-damaged-doc");
113 /// # let cfg = || Config::default().with_persist(dir.path()).with_appendfsync(AppendFsync::Always);
114 /// # Store::open(cfg())?.set(b"a", b"1")?;
115 /// // a record of length 4 whose checksum is wrong, then bytes behind it
116 /// let mut log = std::fs::OpenOptions::new().append(true).open(dir.path().join("aof-0.aof"))?;
117 /// std::io::Write::write_all(&mut log, &[4, 0, 0, 0, 0, 0, 0, 0, b'a', b'b', b'c', b'd', b'!'])?;
118 /// drop(log);
119 /// let s = Store::open(cfg())?;
120 /// let r = s.open_report();
121 /// assert_eq!(r.quarantine_paths.len(), 1);
122 /// assert_eq!(std::fs::metadata(&r.quarantine_paths[0])?.len(), 13); // what was cut off
123 /// # Ok::<(), Box<dyn std::error::Error>>(())
124 /// ```
125 pub quarantine_paths: Vec<PathBuf>,
126 /// Bytes the resync replay hopped over (corrupt regions between valid
127 /// records, summed across shards). Zero under strict replay.
128 ///
129 /// ```
130 /// use kevy_embedded::{AppendFsync, Config, ReplayMode, Store};
131 ///
132 /// let dir = kevy_tmpdir::TmpDir::new("open-report-resync-doc");
133 /// let aof = dir.path().join("aof-0.aof");
134 /// let cfg = || Config::default().with_persist(dir.path()).with_appendfsync(AppendFsync::Always);
135 /// let s = Store::open(cfg())?;
136 /// s.set(b"a", b"1")?;
137 /// let head = std::fs::read(&aof)?;
138 /// s.set(b"b", b"2")?;
139 /// drop(s);
140 /// let all = std::fs::read(&aof)?;
141 /// // a record whose checksum lies, between the two good ones
142 /// let bad = [4, 0, 0, 0, 0, 0, 0, 0, b'a', b'b', b'c', b'd'];
143 /// std::fs::write(&aof, [&head[..], &bad[..], &all[head.len()..]].concat())?;
144 /// let s = Store::open(cfg().with_replay_mode(ReplayMode::Resync))?;
145 /// assert_eq!(s.open_report().resynced_bytes, bad.len() as u64);
146 /// assert_eq!(s.get(b"b")?.as_deref(), Some(&b"2"[..])); // recovered behind it
147 /// # Ok::<(), Box<dyn std::error::Error>>(())
148 /// ```
149 pub resynced_bytes: u64,
150 /// Writes the staging rings held that the AOFs did not — appended
151 /// before the last process was killed and not yet drained — replayed
152 /// and appended to the logs at this open, summed across shards.
153 ///
154 /// ```
155 /// # use kevy_embedded::{AppendFsync, Config, Store};
156 /// # let dir = kevy_tmpdir::TmpDir::new("open-report-killed-doc");
157 /// # let cfg = || {
158 /// # Config::default()
159 /// # .with_persist(dir.path())
160 /// # .with_appendfsync(AppendFsync::EverySec)
161 /// # .with_ttl_reaper_manual()
162 /// # .with_mapped_aof(false)
163 /// # .with_stage_ring(64 * 1024)
164 /// # };
165 /// # let s = Store::open(cfg())?;
166 /// # s.set(b"k", b"v")?;
167 /// // stands in for a killed process: no close runs, the lock goes with it
168 /// std::mem::forget(s);
169 /// std::fs::remove_file(dir.path().join("LOCK"))?;
170 /// let s = Store::open(cfg())?;
171 /// let r = s.open_report();
172 /// assert!(r.stage_recovered > 0); // the ring held the write the log had not
173 /// assert_eq!(s.get(b"k")?.as_deref(), Some(&b"v"[..]));
174 /// # Ok::<(), Box<dyn std::error::Error>>(())
175 /// ```
176 pub stage_recovered: u64,
177 /// Staging rings set aside because they could not show where they
178 /// continue their log; only a power loss or a log changed by hand
179 /// leaves one.
180 ///
181 /// ```
182 /// # use kevy_embedded::{AppendFsync, Config, Store};
183 /// # let dir = kevy_tmpdir::TmpDir::new("open-report-killed-doc");
184 /// # let cfg = || {
185 /// # Config::default()
186 /// # .with_persist(dir.path())
187 /// # .with_appendfsync(AppendFsync::EverySec)
188 /// # .with_ttl_reaper_manual()
189 /// # .with_mapped_aof(false)
190 /// # .with_stage_ring(64 * 1024)
191 /// # };
192 /// # let s = Store::open(cfg())?;
193 /// # s.set(b"k", b"v")?;
194 /// // stands in for a killed process: no close runs, the lock goes with it
195 /// std::mem::forget(s);
196 /// std::fs::remove_file(dir.path().join("LOCK"))?;
197 /// let s = Store::open(cfg())?;
198 /// let r = s.open_report();
199 /// // a kill alone leaves a ring that shows where it continues its log
200 /// assert_eq!((r.stage_recovered > 0, r.stage_discarded), (true, 0));
201 /// # Ok::<(), Box<dyn std::error::Error>>(())
202 /// ```
203 pub stage_discarded: u64,
204}
205
206#[cfg(feature = "persist")]
207#[path = "metric_event.rs"]
208mod event;
209#[cfg(feature = "persist")]
210pub use event::KevyMetric;
211#[cfg(feature = "persist")]
212pub(crate) use event::MetricSink;