src/promote/appdata.rs
//! **The seam** — what an application must be able to say about its own data for
//! the ship to promote it between slots.
//!
//! Skíðblaðnir calls these methods and never inspects the payload. It does not
//! know what a table is, which column carries a change timestamp, which rows must
//! be neutralised after a cross-slot copy, or where the objects live. All of that
//! is the application's, and it stays the application's.
//!
//! # Two rules this trait exists to enforce
//!
//! **1. No silent partial success.** The tool this generalises had a backup that
//! degraded to database-only when the object-store credentials were missing — and
//! still exited 0. The resulting bundle looked exactly like a good one. Here,
//! [`DumpReport`] cannot say "complete" unless every part the application
//! *declared* reports [`CaptureState::Complete`], and an empty report is
//! [`Completeness::Unknown`], never complete. The envelope stamps that verdict
//! into the bundle manifest, so a later `verify` reaches the same conclusion
//! without re-running the dump.
//!
//! **2. The ship never guesses the payload.** If the application cannot answer
//! [`AppData::watermark`], that is `None` = **unknown** —
//! [`Watermark::would_discard`] turns it into [`Discard::Unknown`], which blocks
//! a load exactly as a known conflict does. "We could not tell" is never
//! promoted to "there is nothing newer".
//!
//! # Why the trait is async without an async dependency
//!
//! Real implementations talk to a database, an object store and the Kubernetes
//! API. `async fn` in a trait is not object-safe, and the ship holds `&dyn
//! AppData`, so each method returns a [`Fut`] — a plain `Pin<Box<dyn Future>>`
//! from `std`. No runtime, no `async-trait` crate, no dependency at all. A
//! synchronous implementation returns `Box::pin(async move { … })` and awaits
//! nothing.
use std::collections::BTreeMap;
use std::path::Path;
use anyhow::Result;
use serde::{Deserialize, Serialize};
use super::Fut;
use super::schema::Slot;
/// What happened to one declared part of a capture or a load.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum CaptureState {
/// Everything this part covers is present and accounted for.
Complete,
/// Some of it made it. Both numbers are carried because "1 of 4000" and
/// "3999 of 4000" are very different facts to an operator.
Partial {
captured: u64,
expected: Option<u64>,
},
/// It was not attempted — credentials absent, feature off, caller asked for a
/// narrower scope. **This is the state that used to be an exit code of 0.**
Skipped,
/// It was attempted and it failed.
Failed,
/// It was attempted and the result could not be determined. Distinct from
/// `Skipped` and from `Failed`, and never collapsed into either.
Unknown,
}
impl CaptureState {
pub fn is_complete(&self) -> bool {
matches!(self, CaptureState::Complete)
}
pub fn as_str(&self) -> &'static str {
match self {
CaptureState::Complete => "complete",
CaptureState::Partial { .. } => "partial",
CaptureState::Skipped => "skipped",
CaptureState::Failed => "failed",
CaptureState::Unknown => "unknown",
}
}
}
/// One declared part of a payload — "database", "objects", "configuration".
///
/// The **application** declares which parts exist. That is what makes an omission
/// visible: a part that is never declared cannot be silently absent, because the
/// application said it would be there.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CapturedPart {
/// A stable machine-readable name. The ship does not interpret it; it prints
/// it and stores it.
pub name: String,
/// Is this part required for the capture to be considered whole? A part
/// declared optional may be `Skipped` without spoiling the verdict —
/// but it must still be *declared*, and its state still shows.
pub required: bool,
pub state: CaptureState,
/// How many items this part covers, when the application knows.
/// **`None` means unknown, never zero.**
#[serde(default, skip_serializing_if = "Option::is_none")]
pub items: Option<u64>,
/// Why, in the application's own words. Required for anything that is not
/// `Complete` — see [`CapturedPart::is_well_formed`].
#[serde(default, skip_serializing_if = "Option::is_none")]
pub detail: Option<String>,
}
impl CapturedPart {
/// A part that came through whole.
pub fn complete(name: &str, items: Option<u64>) -> Self {
CapturedPart {
name: name.to_string(),
required: true,
state: CaptureState::Complete,
items,
detail: None,
}
}
/// A part that came through partly. `detail` is not optional: an operator
/// looking at "partial" needs to know what is missing.
pub fn partial(name: &str, captured: u64, expected: Option<u64>, detail: &str) -> Self {
CapturedPart {
name: name.to_string(),
required: true,
state: CaptureState::Partial { captured, expected },
items: Some(captured),
detail: Some(detail.to_string()),
}
}
/// A part that was not attempted. **This is the honest form of the old
/// "degrade quietly and exit 0".**
pub fn skipped(name: &str, detail: &str) -> Self {
CapturedPart {
name: name.to_string(),
required: true,
state: CaptureState::Skipped,
items: None,
detail: Some(detail.to_string()),
}
}
/// A part that was attempted and failed.
pub fn failed(name: &str, detail: &str) -> Self {
CapturedPart {
name: name.to_string(),
required: true,
state: CaptureState::Failed,
items: None,
detail: Some(detail.to_string()),
}
}
/// A part whose outcome could not be determined.
pub fn unknown(name: &str, detail: &str) -> Self {
CapturedPart {
name: name.to_string(),
required: true,
state: CaptureState::Unknown,
items: None,
detail: Some(detail.to_string()),
}
}
/// Mark this part optional — its absence does not spoil the verdict.
pub fn optional(mut self) -> Self {
self.required = false;
self
}
/// Does this part carry the explanation a non-complete state owes?
pub fn is_well_formed(&self) -> bool {
self.state.is_complete() || self.detail.as_deref().is_some_and(|d| !d.trim().is_empty())
}
/// One line for an operator. No colour, no decoration.
pub fn line(&self) -> String {
let mut s = format!("{}: {}", self.name, self.state.as_str());
if let CaptureState::Partial { captured, expected } = &self.state {
s.push_str(&match expected {
Some(e) => format!(" ({captured}/{e})"),
None => format!(" ({captured} of an unknown total)"),
});
} else if let Some(n) = self.items {
s.push_str(&format!(" ({n})"));
}
if !self.required {
s.push_str(" [optional]");
}
if let Some(d) = &self.detail {
s.push_str(&format!(" — {d}"));
}
s
}
}
/// The verdict on a whole capture or load.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum Completeness {
/// Every required part is `Complete`, and at least one part was declared.
Complete,
/// At least one required part is not `Complete`. The offenders are named.
Incomplete { parts: Vec<String> },
/// Nothing was declared, so there is nothing to be complete about.
/// **Deliberately not `Complete`** — an empty report is the shape a
/// short-circuited dump has, and it must not be mistaken for a good one.
Unknown,
}
impl Completeness {
pub fn is_complete(&self) -> bool {
matches!(self, Completeness::Complete)
}
pub fn as_str(&self) -> &'static str {
match self {
Completeness::Complete => "complete",
Completeness::Incomplete { .. } => "incomplete",
Completeness::Unknown => "unknown",
}
}
}
/// The newest state a slot holds, as two independent clocks.
///
/// One clock is not enough, and the second one is not decoration. A change
/// timestamp only ever moves **forward** when rows are written — a *deletion*
/// does not move it, it removes it. So a slot where somebody deleted two hundred
/// rows since the bundle was sealed reads as "nothing happened", and a switch
/// destroys those deletions silently. A magnitude (row counts, object counts —
/// whatever the application counts) is the cheap quantity that actually goes
/// *down*, and it is what makes the check honest in both directions.
///
/// Both fields are `Option`, and `None` is **unknown**, never "zero" and never
/// "nothing changed".
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct Watermark {
/// An orderable mark — typically a timestamp, compared as a string, produced
/// by the application in one consistent format so both sides of a comparison
/// are the same quantity.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub mark: Option<String>,
/// Counts per named bucket. `None` = the application could not produce them
/// (an incremental dump, for instance, whose counts describe a delta and not
/// the whole). A half-true count is worse than no count: it would make the
/// comparison look whole.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub magnitude: Option<BTreeMap<String, i64>>,
}
/// What a load would do to the target's newer state.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case", tag = "discard")]
pub enum Discard {
/// The target holds nothing the bundle does not. Safe.
Nothing,
/// The target has been written to since the bundle was sealed.
NewerWrites {
target_mark: String,
bundle_mark: String,
},
/// The target's counts differ from the bundle's in a way that means rows were
/// removed there — invisible to the mark alone.
RemovedRows { buckets: Vec<String> },
/// It could not be determined. **Blocking**, exactly like a known conflict:
/// this is the case the tool this generalises got wrong by treating a missing
/// watermark as "nothing newer".
Unknown { because: String },
}
impl Discard {
/// Must a human confirm before the load proceeds?
///
/// True for everything except [`Discard::Nothing`] — including
/// [`Discard::Unknown`].
pub fn requires_confirmation(&self) -> bool {
!matches!(self, Discard::Nothing)
}
/// One line for an operator.
pub fn line(&self) -> String {
match self {
Discard::Nothing => "the target holds nothing newer than this bundle".into(),
Discard::NewerWrites {
target_mark,
bundle_mark,
} => format!(
"the target has been written to since this bundle was sealed \
(target {target_mark} > bundle {bundle_mark}) — loading DISCARDS those changes"
),
Discard::RemovedRows { buckets } => format!(
"the target has fewer rows than this bundle in {} — loading would resurrect \
data that was deleted there",
buckets.join(", ")
),
Discard::Unknown { because } => format!(
"UNKNOWN whether the target holds newer data: {because}. \
Unknown is treated as 'may discard' — it is not a green light"
),
}
}
}
impl Watermark {
/// A watermark with only a mark.
pub fn at(mark: &str) -> Self {
Watermark {
mark: Some(mark.to_string()),
magnitude: None,
}
}
/// Add counts.
pub fn with_magnitude(mut self, m: BTreeMap<String, i64>) -> Self {
self.magnitude = Some(m);
self
}
/// **Would loading this bundle over `target` discard anything?**
///
/// `self` is the bundle's watermark; `target` is the slot's, `None` when the
/// application could not answer. Every path that cannot prove safety returns
/// [`Discard::Unknown`].
pub fn would_discard(&self, target: Option<&Watermark>) -> Discard {
let Some(t) = target else {
return Discard::Unknown {
because: "the target slot did not report a watermark".into(),
};
};
// Clock one: has the target been written to since the seal?
match (&self.mark, &t.mark) {
(Some(b), Some(tm)) if tm.as_str() > b.as_str() => {
return Discard::NewerWrites {
target_mark: tm.clone(),
bundle_mark: b.clone(),
};
}
(Some(_), Some(_)) => {}
_ => {
return Discard::Unknown {
because: "one side has no mark to compare".into(),
};
}
}
// Clock two: has anything been removed there? A mark cannot see this.
match (&self.magnitude, &t.magnitude) {
(Some(bm), Some(tm)) => {
let mut shrunk: Vec<String> = bm
.iter()
.filter(|(k, bc)| tm.get(*k).is_some_and(|tc| tc < bc))
.map(|(k, _)| k.clone())
.collect();
// A bucket the target does not report at all is not "zero"; it is
// one more thing we cannot compare.
if bm.keys().any(|k| !tm.contains_key(k)) {
return Discard::Unknown {
because: "the target did not report counts for every bucket in the bundle"
.into(),
};
}
shrunk.sort();
if shrunk.is_empty() {
Discard::Nothing
} else {
Discard::RemovedRows { buckets: shrunk }
}
}
_ => Discard::Unknown {
because: "no comparable counts — a deletion moves no mark, so a mark alone \
cannot prove nothing was removed"
.into(),
},
}
}
}
/// What a dump captured. **The type that makes an incomplete capture impossible
/// to miss.**
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct DumpReport {
/// Every part the application declared, in its own order.
pub parts: Vec<CapturedPart>,
/// The source slot's watermark at seal time. `None` = unknown.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub watermark: Option<Watermark>,
/// Anything else worth carrying into the manifest. **Opaque to the ship** — it
/// stores and round-trips this and never reads a key out of it. Schema
/// version, table list, object keys: all fit here.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub app_meta: Option<serde_json::Value>,
}
impl DumpReport {
/// An empty report. Its verdict is [`Completeness::Unknown`] — see the type
/// doc for why that is not `Complete`.
pub fn new() -> Self {
Self::default()
}
/// Declare a part.
pub fn part(mut self, p: CapturedPart) -> Self {
self.parts.push(p);
self
}
/// Record the source watermark. Passing `None` is legitimate and means
/// *unknown*.
pub fn watermark(mut self, w: Option<Watermark>) -> Self {
self.watermark = w;
self
}
/// Attach payload knowledge the ship will carry and never interpret.
pub fn app_meta(mut self, v: serde_json::Value) -> Self {
self.app_meta = Some(v);
self
}
/// The verdict.
pub fn completeness(&self) -> Completeness {
if self.parts.is_empty() {
return Completeness::Unknown;
}
let bad: Vec<String> = self
.parts
.iter()
.filter(|p| p.required && !p.state.is_complete())
.map(|p| p.line())
.collect();
if bad.is_empty() {
Completeness::Complete
} else {
Completeness::Incomplete { parts: bad }
}
}
/// Refuse unless the capture is whole. The error names every offending part,
/// so the operator does not have to go looking for what is missing.
pub fn require_complete(&self) -> Result<()> {
match self.completeness() {
Completeness::Complete => Ok(()),
Completeness::Incomplete { parts } => anyhow::bail!(
"this capture is INCOMPLETE — it must not be treated as a good bundle:\n {}\n \
Fix what is missing and capture again, or seal it explicitly as incomplete \
(the manifest will say so, and loading it will require saying so again).",
parts.join("\n ")
),
Completeness::Unknown => anyhow::bail!(
"this capture declared NO parts, so there is nothing to be complete about.\n \
An empty report is what a dump that never ran looks like; it is not a \
successful one. The application must declare each part it captures."
),
}
}
/// Every part carries the explanation a non-complete state owes.
pub fn is_well_formed(&self) -> bool {
self.parts.iter().all(CapturedPart::is_well_formed)
}
/// Lines for an operator, one per part.
pub fn lines(&self) -> Vec<String> {
self.parts.iter().map(CapturedPart::line).collect()
}
}
/// What a load actually did. Same shape as [`DumpReport`], same rules — a load
/// that half-worked must be as visible as a capture that half-worked.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct LoadReport {
pub parts: Vec<CapturedPart>,
/// The target's watermark **after** the load, when the application can say.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub watermark: Option<Watermark>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub app_meta: Option<serde_json::Value>,
}
impl LoadReport {
pub fn new() -> Self {
Self::default()
}
pub fn part(mut self, p: CapturedPart) -> Self {
self.parts.push(p);
self
}
pub fn watermark(mut self, w: Option<Watermark>) -> Self {
self.watermark = w;
self
}
pub fn app_meta(mut self, v: serde_json::Value) -> Self {
self.app_meta = Some(v);
self
}
pub fn completeness(&self) -> Completeness {
DumpReport {
parts: self.parts.clone(),
watermark: None,
app_meta: None,
}
.completeness()
}
pub fn require_complete(&self) -> Result<()> {
DumpReport {
parts: self.parts.clone(),
watermark: None,
app_meta: None,
}
.require_complete()
}
pub fn lines(&self) -> Vec<String> {
self.parts.iter().map(CapturedPart::line).collect()
}
}
/// Proof that writers are stopped.
///
/// Returned by [`AppData::quiesce`] and required by [`AppData::load`], so the
/// order is a signature rather than a convention. It is deliberately **not**
/// `Drop`-based: an implicit resume on unwind is how an environment comes back up
/// halfway through a load. Resuming is the caller's explicit step.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Quiesced {
/// The slot's name, so a token cannot be spent on another slot.
pub slot: String,
/// What the application actually stopped, for the operator's log.
pub stopped: Vec<String>,
/// Anything the application needs back in order to resume — replica counts,
/// for instance. Opaque to the ship.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resume_token: Option<serde_json::Value>,
}
impl Quiesced {
pub fn new(slot: &str, stopped: Vec<String>) -> Self {
Quiesced {
slot: slot.to_string(),
stopped,
resume_token: None,
}
}
pub fn with_resume_token(mut self, v: serde_json::Value) -> Self {
self.resume_token = Some(v);
self
}
/// Is this token for `slot`?
pub fn is_for(&self, slot: &str) -> bool {
self.slot.eq_ignore_ascii_case(slot)
}
}
/// What an application must be able to say about its own data.
///
/// Implement this once, in the application's own repository, next to the code
/// that knows the schema. Nothing in Skíðblaðnir reads the payload; it calls
/// these and sequences them.
pub trait AppData: Send + Sync {
/// A stable name for the payload this implementation speaks for. Stamped into
/// the bundle manifest so a bundle knows what produced it.
fn describe(&self) -> &str;
/// Stop writers before a load — typically scale the web workload to zero and
/// wait for it. Returns the token [`AppData::load`] requires.
fn quiesce<'a>(&'a self, slot: &'a Slot) -> Fut<'a, Result<Quiesced>>;
/// Let writers back in. Called by the caller explicitly, including on the
/// failure path: a failed load must never leave the slot scaled to zero.
fn resume<'a>(&'a self, slot: &'a Slot, quiesced: &'a Quiesced) -> Fut<'a, Result<()>>;
/// Dump this slot's data into `staging`. Returns **what was captured** —
/// see [`DumpReport`], and note that returning an empty report is not a way to
/// report success.
fn dump<'a>(&'a self, slot: &'a Slot, staging: &'a Path) -> Fut<'a, Result<DumpReport>>;
/// Load a staged dump into `slot`. The ship has already unsealed and verified
/// the bundle; `quiesced` is the proof that writers are stopped.
fn load<'a>(
&'a self,
slot: &'a Slot,
staging: &'a Path,
quiesced: &'a Quiesced,
) -> Fut<'a, Result<LoadReport>>;
/// The newest state this slot holds. **`Ok(None)` means unknown** — the ship
/// treats that as "may discard newer data", never as "there is nothing
/// newer". An application that cannot compute one should return `Ok(None)`
/// rather than a fabricated value.
fn watermark<'a>(&'a self, slot: &'a Slot) -> Fut<'a, Result<Option<Watermark>>>;
/// Fix up state that must not survive a cross-slot copy — a queue of
/// unprocessed outbound work, a scheduler's claim table, anything whose rows
/// would make the target act on the source's behalf. Called after
/// [`AppData::load`] and before writers are let back in.
///
/// Returns a description of what it changed, for the operator's log. An
/// application with nothing to neutralise returns an empty vector — which is
/// a statement, not a silence, because the method still had to be called.
fn neutralise_after_load<'a>(
&'a self,
slot: &'a Slot,
quiesced: &'a Quiesced,
) -> Fut<'a, Result<Vec<String>>>;
}
#[cfg(test)]
mod tests {
use super::*;
fn counts(pairs: &[(&str, i64)]) -> BTreeMap<String, i64> {
pairs.iter().map(|(k, v)| (k.to_string(), *v)).collect()
}
// ── no silent partial success ────────────────────────────────────────────
/// The exact bug: object storage was unreachable, so the dump captured the
/// database only — and reported success.
#[test]
fn a_dump_that_skipped_a_required_part_cannot_report_success() {
let r = DumpReport::new()
.part(CapturedPart::complete("database", Some(41)))
.part(CapturedPart::skipped(
"objects",
"S3 credentials were not configured",
));
assert!(!r.completeness().is_complete());
let err = r
.require_complete()
.expect_err("a skipped required part is not success");
let m = format!("{err:#}");
assert!(m.contains("INCOMPLETE"), "{m}");
assert!(
m.contains("objects: skipped"),
"the offender must be named: {m}"
);
assert!(m.contains("S3 credentials"), "…with the reason: {m}");
}
/// An empty report is what a dump that never ran looks like.
#[test]
fn an_empty_report_is_unknown_not_complete() {
let r = DumpReport::new();
assert_eq!(r.completeness(), Completeness::Unknown);
assert!(!r.completeness().is_complete());
let err = r.require_complete().unwrap_err();
assert!(format!("{err:#}").contains("declared NO parts"), "{err:#}");
}
#[test]
fn partial_failed_and_unknown_all_spoil_the_verdict() {
for part in [
CapturedPart::partial("objects", 3, Some(4000), "3 of 4000 fetched before a 403"),
CapturedPart::failed("objects", "bucket does not exist"),
CapturedPart::unknown("objects", "the store timed out; count unavailable"),
] {
let r = DumpReport::new()
.part(CapturedPart::complete("database", Some(41)))
.part(part);
assert!(!r.completeness().is_complete(), "{:?}", r.parts);
assert!(r.require_complete().is_err());
}
}
#[test]
fn an_optional_part_may_be_skipped_without_spoiling_the_verdict_but_still_shows() {
let r = DumpReport::new()
.part(CapturedPart::complete("database", Some(41)))
.part(CapturedPart::skipped("images", "image bytes not requested").optional());
assert!(r.completeness().is_complete());
assert!(
r.lines()
.iter()
.any(|l| l.contains("images: skipped") && l.contains("[optional]"))
);
}
#[test]
fn every_non_complete_part_carries_its_reason() {
let r = DumpReport::new()
.part(CapturedPart::skipped("objects", "no credentials"))
.part(CapturedPart::failed("config", "forbidden"));
assert!(r.is_well_formed());
// A forged wire form without a reason is caught.
let forged: CapturedPart =
serde_json::from_str(r#"{"name":"x","required":true,"state":"skipped"}"#).unwrap();
assert!(!forged.is_well_formed());
}
#[test]
fn a_complete_capture_is_complete() {
let r = DumpReport::new()
.part(CapturedPart::complete("database", Some(41)))
.part(CapturedPart::complete("objects", Some(4000)));
assert_eq!(r.completeness(), Completeness::Complete);
r.require_complete().unwrap();
}
#[test]
fn a_load_report_is_held_to_the_same_standard() {
let r = LoadReport::new()
.part(CapturedPart::complete("database", Some(41)))
.part(CapturedPart::partial(
"objects",
10,
Some(4000),
"upload failures",
));
assert!(r.require_complete().is_err());
assert!(
LoadReport::new().require_complete().is_err(),
"an empty load is not a load"
);
}
#[test]
fn app_meta_is_carried_and_round_trips() {
let r = DumpReport::new()
.part(CapturedPart::complete("database", Some(1)))
.app_meta(serde_json::json!({"schema_version": 51, "tables": ["a", "b"]}));
let back: DumpReport = serde_json::from_str(&serde_json::to_string(&r).unwrap()).unwrap();
assert_eq!(back, r);
assert_eq!(back.app_meta.unwrap()["schema_version"], 51);
}
// ── the ship never guesses the payload ───────────────────────────────────
/// The rule, stated as a test: no answer is not a green light.
#[test]
fn an_unknown_target_watermark_blocks_rather_than_passes() {
let bundle = Watermark::at("2026-07-20T10:00:00Z").with_magnitude(counts(&[("t", 10)]));
let d = bundle.would_discard(None);
assert!(matches!(d, Discard::Unknown { .. }), "{d:?}");
assert!(
d.requires_confirmation(),
"unknown must block exactly as a conflict does"
);
assert!(d.line().contains("not a green light"));
}
#[test]
fn a_target_written_to_since_the_seal_is_a_discard() {
let bundle = Watermark::at("2026-07-20T10:00:00Z");
let target = Watermark::at("2026-07-21T08:00:00Z");
let d = bundle.would_discard(Some(&target));
assert!(matches!(d, Discard::NewerWrites { .. }), "{d:?}");
assert!(d.requires_confirmation());
}
/// A deletion does not move a mark — it removes one. Counts are the only
/// cheap quantity that goes down.
#[test]
fn deletions_are_invisible_to_the_mark_and_caught_by_the_counts() {
let bundle = Watermark::at("2026-07-20T10:00:00Z")
.with_magnitude(counts(&[("docs", 4000), ("t", 5)]));
// Same mark — nothing was written — but 200 documents were deleted.
let target = Watermark::at("2026-07-20T10:00:00Z")
.with_magnitude(counts(&[("docs", 3800), ("t", 5)]));
match bundle.would_discard(Some(&target)) {
Discard::RemovedRows { buckets } => assert_eq!(buckets, vec!["docs".to_string()]),
other => panic!("a mark-only check would have called this safe: {other:?}"),
}
}
#[test]
fn missing_counts_are_unknown_not_safe() {
let bundle = Watermark::at("2026-07-20T10:00:00Z");
let target = Watermark::at("2026-07-20T10:00:00Z");
// Same mark, no counts on either side: a deletion would be invisible.
assert!(matches!(
bundle.would_discard(Some(&target)),
Discard::Unknown { .. }
));
// Counts present but the target does not report every bucket.
let bundle = bundle.with_magnitude(counts(&[("a", 1), ("b", 2)]));
let target = target.with_magnitude(counts(&[("a", 1)]));
assert!(matches!(
bundle.would_discard(Some(&target)),
Discard::Unknown { .. }
));
}
#[test]
fn a_missing_mark_on_either_side_is_unknown() {
let with = Watermark::at("2026-07-20T10:00:00Z");
let without = Watermark::default();
assert!(matches!(
with.would_discard(Some(&without)),
Discard::Unknown { .. }
));
assert!(matches!(
without.would_discard(Some(&with)),
Discard::Unknown { .. }
));
}
#[test]
fn a_clean_comparison_is_the_only_thing_that_needs_no_confirmation() {
let bundle = Watermark::at("2026-07-20T10:00:00Z").with_magnitude(counts(&[("t", 5)]));
let target = Watermark::at("2026-07-20T10:00:00Z").with_magnitude(counts(&[("t", 5)]));
let d = bundle.would_discard(Some(&target));
assert_eq!(d, Discard::Nothing);
assert!(!d.requires_confirmation());
// A target with MORE rows and the same mark is not a discard either.
let target = Watermark::at("2026-07-20T10:00:00Z").with_magnitude(counts(&[("t", 9)]));
assert_eq!(bundle.would_discard(Some(&target)), Discard::Nothing);
}
#[test]
fn a_quiesce_token_belongs_to_one_slot() {
let q = Quiesced::new("test1", vec!["deployment/web".into()]);
assert!(q.is_for("test1") && q.is_for("TEST1"));
assert!(!q.is_for("njord"));
}
}