dendro 0.2.1

//! When to seal: the policy that decides how long a stream's rows accumulate
//! in the WAL before they become a segment.
//!
//! [`SealPolicy`](crate::seal::SealPolicy) answers "is this open segment due?" and [`SegmentAccount`](crate::seal::SegmentAccount)
//! maintains the byte and row counts that question is asked against. Both are
//! properties of *segmenting a source* rather than of the container it
//! lands in, so a plain source and a rolling buffer use exactly the same
//! ones.
//!
//! The caller decides when to act on the answer; dendro never seals behind
//! your back. Nothing here writes; it only accounts and advises.

use std::time::{Duration, Instant};

/// Granularity of the first-seal stagger. A stream's first segment closes at
/// `max_rows - (max_rows / (2 * STAGGER_BUCKETS)) * bucket` for a `bucket` in
/// `[0, STAGGER_BUCKETS)`, that is, somewhere in `[max_rows / 2, max_rows]`. 64
/// buckets is ample spread for a dozen streams, and capping the reduction at
/// 50% bounds the startup cost to one short segment per stream.
pub const STAGGER_BUCKETS: u64 = 64;

/// The stagger identity of a writer that can only ever hold one source.
///
/// With one source there is never a second one to desync against, so every
/// stream can share an identity and the stagger reduces to spreading streams
/// within it. Named rather than spelled `""` at each call site so the reason
/// travels with the value.
#[cfg_attr(not(test), allow(dead_code))]
#[doc(hidden)]
pub const SINGLE_SOURCE_KEY: &str = "";

/// When an open segment is due to be sealed. Byte-first: the byte cap is the
/// one that bounds both the builder's memory footprint and the encoder's input,
/// and it is maintained O(1) per entry by the caller as it stages a row.
///
/// **The age bound exists for the kill-loss window, not finalize cost.** The
/// byte and row caps alone bound finalize time and memory — a slow stream's
/// open segment is naturally tiny — so age sealing only bounds how much data an
/// unclean kill loses. It is also what drives segment count, so the trade (loss
/// window vs segments and read-time merge width) is deliberate.
pub struct SealPolicy {
    /// Seal once the open segment's approximate encoded size reaches this.
    /// The cap that splits the WIDE streams; see [`SealPolicy::default`].
    pub max_bytes: usize,
    /// Seal once the open segment holds this many rows. The cap that splits
    /// the THIN streams, which take a long time to reach any byte threshold.
    pub max_rows: usize,
    /// Seal once the open segment is this old, whatever its size. Bounds how
    /// much an unclean kill loses, not seal cost.
    pub max_age: Duration,
    /// Seal on a wall-clock boundary as well: no segment spans two multiples
    /// of this, in **the same unit as your row timestamps** (dendro does not
    /// mandate one; nanoseconds is the convention). `None`, the default,
    /// leaves segment edges wherever the caps put them.
    ///
    /// Block-oriented stores align because it makes range pruning
    /// predictable — a query for one bucket touches one segment — and it
    /// makes two archives of the same window comparable segment for segment,
    /// which the caps alone never give you, since they depend on how much
    /// happened to arrive.
    ///
    /// Nothing here seals; dendro never does. This is advice, and there are
    /// two ways to take it, which differ by one row: ask
    /// [`SegmentAccount::starts_new_bucket`] before adding a row and seal
    /// first, for exact edges; or just let [`SegmentAccount::is_due`] notice
    /// afterwards, which is one call site instead of two and leaves the
    /// boundary row at the end of the older segment.
    ///
    /// A non-positive value means no alignment, the same as `None`.
    pub align: Option<i64>,
}

/// The two caps and the age bound.
///
/// **Seal policy is not a CPU knob.** Sealing is a minority of what the
/// writer burns — the caller's own per-append work dominates — so
/// moving these caps trades finalize latency, peak memory and the kill-loss
/// window against each other, and barely touches CPU. Tune them for those
/// three, not for throughput.
///
/// The two caps are not redundant: they bind on disjoint sets of tables.
/// `max_rows` splits the *thin* tables, which would otherwise take a long time
/// to reach any byte threshold; `max_bytes` splits the *wide* ones, which reach
/// it almost immediately. Each therefore costs close to nothing on the tables
/// the other one reaches.
///
/// `max_bytes` bounds finalize wall-clock, which is what the streaming writer
/// exists to protect — a container gets on the order of ten seconds between
/// SIGTERM and SIGKILL, and an unsealed tail has to fit in it. A larger cap is
/// tempting because it produces fewer, denser segments, which shrinks the
/// archive and speeds queries (read cost tracks segment count); that trade
/// belongs to the offline compactor, which can have it without charging the
/// producer for it.
///
/// Going smaller is worse than it looks. Segments are the encoder's unit of
/// compression, so starving them re-pays per-column-chunk footer metadata on
/// every split and denies the RLE and dictionary encoders anything to amortize
/// over; well below this the archive inflates several-fold. 8 MiB is where that
/// curve has flattened and finalize has not yet climbed.
///
/// `max_rows` is what bounds the finalize tail on thin tables, and it is nearly
/// free precisely because it does not reach the wide ones.
///
/// **These numbers were measured on one workload**: a telemetry agent
/// sealing a few dozen streams at a ~46 ms append cadence, on a production
/// fleet — and are a starting point, not a property of the container. If you
/// append at a very different cadence, or with rows of a very different size,
/// tune them for your own finalize latency, peak memory and kill-loss
/// window.
impl Default for SealPolicy {
    fn default() -> Self {
        Self {
            max_bytes: 8 * 1024 * 1024,
            max_rows: 900,
            // Not a free variable like the two caps: this bounds how much an
            // unclean kill loses, not seal cost. Trade it against segment count.
            max_age: Duration::from_secs(300),
            // Off: it costs a short segment at every boundary, which is only
            // worth paying when something downstream reads by bucket.
            align: None,
        }
    }
}

/// Everything the seal decision reads about an open segment, and none of the
/// rows.
///
/// **Separate from the caller's own buffering because only one of the two keeps
/// the rows.** A buffering writer encodes the builder it has been filling; this one
/// writes each row to the WAL and rebuilds the table from it at seal time, so
/// it has nothing to ask `rows()` or `approx_bytes()` of. Both must still seal
/// at the same row from the same input, which they do by both deciding here —
/// the alternative is two copies of a four-term predicate drifting apart in a
/// way that only shows up as differently-shaped archives.
pub struct SegmentAccount {
    /// The aligned bucket the open segment's first row fell in, and the
    /// bucket width — `None` when the caller asked for no alignment, or
    /// before the first row. See [`SealPolicy::align`].
    align: Option<i64>,
    first_bucket: Option<i64>,
    last_bucket: Option<i64>,
    rows: usize,
    approx_bytes: usize,
    /// Instant the current segment was opened (the age bound's origin).
    opened_at: Instant,
    max_bytes: usize,
    max_rows: usize,
    max_age: Duration,
}

impl SegmentAccount {
    /// Open a stream's **first** segment, with byte, row and age targets
    /// reduced by a deterministic per-stream fraction of up to 50%.
    ///
    /// **All three caps, not just rows and age.** The byte cap is the one that
    /// splits the *wide* tables (see `SealPolicy`), so leaving it unstaggered
    /// left exactly those tables with no phase offset at all. Within one
    /// source that was survivable — different streams fill at different
    /// rates, so they drift anyway — but two sources of the same producer
    /// carry identical data, so a byte-bound table reached the cap on the same
    /// row in both and they sealed in permanent lockstep. Measured before the
    /// fix: `cpu_usage` 49/49 segment boundaries coincident across two
    /// sources, against 1/6 for the row-bound tables.
    ///
    /// This is a *phase offset*, not a period change. Every row-capped table
    /// otherwise advances exactly one row per tick starting from row 0, so they
    /// all reach `max_rows` in permanent lockstep and seal as one large batch
    /// forever. Co-seals, not large individual segments, are what put a seal
    /// over the tick budget. Shortening only the first segment desyncs the
    /// tables for the life of the source while leaving steady-state segment
    /// size and count untouched — `rotate` restores the full policy.
    pub fn open_first(stream: &str, source_key: &str, policy: &SealPolicy) -> Self {
        let bucket = stagger_bucket(stream, source_key);
        // Divide before multiplying: `max_rows` is `usize::MAX` in several
        // callers, and `max_rows * bucket` would overflow.
        let row_offset = (policy.max_rows / (2 * STAGGER_BUCKETS as usize)) * bucket as usize;
        let age_offset = (policy.max_age / (2 * STAGGER_BUCKETS as u32)) * bucket as u32;
        let byte_offset = (policy.max_bytes / (2 * STAGGER_BUCKETS as usize)) * bucket as usize;
        Self {
            rows: 0,
            approx_bytes: 0,
            opened_at: Instant::now(),
            // `max(1)` for the same reason as the row target: a zero cap would
            // seal an empty segment every tick forever.
            max_bytes: policy.max_bytes.saturating_sub(byte_offset).max(1),
            // `max(1)` so a small policy can never yield a zero row target,
            // which would seal a one-row segment every tick forever.
            max_rows: policy.max_rows.saturating_sub(row_offset).max(1),
            max_age: policy.max_age.saturating_sub(age_offset),
            // Alignment is not staggered: its purpose is for every
            // stream cuts at the same wall-clock instant, so offsetting it
            // per stream would defeat it. The caps are staggered to spread
            // the seal WORK; alignment is about where the edges land.
            align: policy.align,
            first_bucket: None,
            last_bucket: None,
        }
    }

    /// Account one appended row. `bytes` is roughly the encoded size of that
    /// row, which is exactly what the caller's own row accounting would have
    /// charged; `ts` is its timestamp, which is
    /// what [`SealPolicy::align`] is measured against. A caller that wants no
    /// alignment may pass any timestamp; it is only read when `align` is set.
    pub fn add_row(&mut self, bytes: usize, ts: i64) {
        self.rows += 1;
        self.approx_bytes += bytes;
        if let Some(bucket) = self.bucket_of(ts) {
            self.first_bucket.get_or_insert(bucket);
            self.last_bucket = Some(bucket);
        }
    }

    /// Which aligned bucket a timestamp falls in, or `None` when the caller
    /// asked for no alignment.
    ///
    /// `div_euclid`, not `/`: a timestamp can be negative (it is an `i64`,
    /// and zero is 1970, not the bottom of the range), and truncating
    /// division rounds toward zero, which would put `-1` and `1` in the same
    /// bucket either side of the epoch.
    fn bucket_of(&self, ts: i64) -> Option<i64> {
        self.align.filter(|a| *a > 0).map(|a| ts.div_euclid(a))
    }

    /// Would a row at `ts` belong to a different aligned bucket than the ones
    /// already in this open segment?
    ///
    /// Ask before adding it, and seal first, for segments whose edges land
    /// exactly on the boundary. Always false when no alignment is set, and
    /// for the first row of a segment, which starts whatever bucket it is in.
    pub fn starts_new_bucket(&self, ts: i64) -> bool {
        match (self.bucket_of(ts), self.first_bucket) {
            (Some(next), Some(first)) => next != first,
            _ => false,
        }
    }

    /// Whether this open segment is past any seal threshold. An empty segment
    /// never is.
    ///
    /// Takes no policy: the account carries its own copy of all three caps,
    /// because all three are staggered on the first segment and `rotate`
    /// restores them. Reading `policy.max_bytes` here instead is what let the
    /// byte cap escape the stagger.
    pub fn is_due(&self, now: Instant) -> bool {
        self.rows > 0
            && (self.approx_bytes >= self.max_bytes
                || self.rows >= self.max_rows
                || now.duration_since(self.opened_at) >= self.max_age
                // Already holding two buckets — the caller did not ask
                // `starts_new_bucket` first, so the boundary row is in here.
                // Sealing now still bounds the segment to two buckets rather
                // than letting it run to a cap.
                || self.first_bucket != self.last_bucket)
    }

    /// Reset onto a fresh segment after a seal, dropping the startup stagger:
    /// every segment after the first uses the full policy.
    pub fn rotate(&mut self, policy: &SealPolicy, now: Instant) {
        self.rows = 0;
        self.approx_bytes = 0;
        self.opened_at = now;
        self.align = policy.align;
        self.first_bucket = None;
        self.last_bucket = None;
        self.max_bytes = policy.max_bytes;
        self.max_rows = policy.max_rows;
        self.max_age = policy.max_age;
    }

    /// Rows in the open segment.
    pub fn rows(&self) -> usize {
        self.rows
    }

    /// The row and age targets the *current* open segment seals at.
    #[cfg(any(test, feature = "test-support"))]
    pub fn targets(&self) -> (usize, Duration) {
        (self.max_rows, self.max_age)
    }

    /// This segment's byte cap, after the first-segment stagger.
    #[cfg(test)]
    pub fn byte_target(&self) -> usize {
        self.max_bytes
    }
}

/// FNV-1a over the stream name AND the source's identity, reduced to a
/// stagger bucket.
///
/// Hand-written rather than `DefaultHasher` on purpose: the offset must be
/// identical across runs, builds and Rust versions, and `DefaultHasher` is
/// SipHash with an explicitly unstable algorithm and no seed guarantee.
/// Randomizing the initial deadline would desync just as well, but a stable
/// offset keeps a source's segment boundaries reproducible.
///
/// **`source_key` is why this is not just the stream name.** An archive can
/// hold several sources, and two producers have *identical* stream
/// sets — so keying on the stream alone would give every table in source B
/// the same bucket as its namesake in A. The two sources would then seal in
/// permanent lockstep, doubling the co-seal batch size exactly when the archive
/// holds twice the tables: the stagger still working within a source and
/// silently defeated across them.
///
/// The key is the source's canonical label set, not its `sources` row id.
/// An autoincrement id would make the bucket — and so where every segment
/// boundary falls — depend on the order endpoints were listed on the command
/// line, and the same two producers recorded with the flags swapped would segment
/// differently for no reason. Labels are stable across runs and across flag
/// order, and they are what actually distinguishes two arms: `host` separates a
/// multi-host archive, and an A/B on a *single* host separates only on `arm`,
/// which a node name alone would miss.
///
/// Two sources with genuinely identical label sets still collide. That is
/// the degenerate case — the operator gave two endpoints nothing to tell them
/// apart — and the answer is to warn rather than to fold in the id and
/// reintroduce order-dependence.
pub fn stagger_bucket(stream: &str, source_key: &str) -> u64 {
    const PRIME: u64 = 0x0000_0100_0000_01b3; // FNV-1a 64-bit prime

    // Absorb one byte, twice: the byte itself, then the two bits the final
    // `% STAGGER_BUCKETS` would otherwise discard.
    //
    // Plain FNV-1a reduced mod 64 depends only on `byte & 0x3f`, because the
    // prime is odd and so `x -> ((x ^ b) * PRIME) mod 64` is a bijection on
    // Z64 keyed on the low six bits. Two keys that agree byte-for-byte modulo
    // 0x40 therefore draw the SAME bucket for every stream — total lockstep,
    // the exact failure this stagger exists to prevent. The aliasing pairs are
    // ordinary in hostnames: `-` with `m`, `.` with `n`, digits with `p`-`y`.
    // `host=web-01` and `host=webm01` collided on all of them.
    //
    // Folding the high bits in as their own absorbed value breaks that
    // identity for bits 6-7 while leaving the low-bit structure — which
    // measures *better* than a well-avalanched finalizer here — intact.
    //
    // It does NOT close bit 5, and that is now a measured decision rather than
    // a deferral. The same algebra applies one bit lower: `x ^ 0x20` is
    // `x + 32 (mod 64)` and `51 * 32 == 32 (mod 64)`, so flipping bit 5 of an
    // absorbed byte XORs 0x20 through the whole chain and a second flip
    // cancels it. Two label sets differing by an EVEN number of bit-5 flips
    // share every bucket. In printable ASCII bit 5 is the case bit, so this
    // needs two sources whose labels differ only by capitalization.
    //
    // **Closing it costs more than it buys, because the spread and the alias
    // are the same property.** The low-bit structure that makes this hash
    // spread a real stream set PERFECTLY is exactly the affine structure the
    // alias exploits. Measured over 500 source keys, as colliding
    // stream-pairs normalized by what a uniform random assignment would give
    // (0 = perfect, 1.0 = random):
    //
    //     candidate                12 streams   26 streams   alias
    //     this hash                0.000         0.394         total lockstep
    //     + absorb `b >> 5`        1.939         1.378         8/26 (not closed!)
    //     + fold `b>>5 ^ b>>6`     1.939         1.182         1/26
    //     reduce from top bits     0.981         1.002         closed
    //
    // Every candidate that closes the alias lands at or WORSE than random,
    // and the `b >> 5` fold the backlog proposed does not even close it. The
    // alias needs an operator to type two labels differing only in case, so it
    // is detected and warned about instead — see `staggers_identically`, which
    // states the condition exactly rather than approximating it with a
    // case-insensitive compare.
    let absorb = |h: &mut u64, b: u64| {
        *h ^= b;
        *h = h.wrapping_mul(PRIME);
        *h ^= b >> 6;
        *h = h.wrapping_mul(PRIME);
    };

    let mut h: u64 = 0xcbf2_9ce4_8422_2325; // FNV-1a 64-bit offset basis
    for b in stream.as_bytes() {
        absorb(&mut h, *b as u64);
    }
    // A separator no label byte can supply, so `a=1,b=2` and `a=1,b=2` reached
    // from different splits cannot alias.
    absorb(&mut h, 0xff);
    for b in source_key.as_bytes() {
        absorb(&mut h, *b as u64);
    }
    h % STAGGER_BUCKETS
}

/// Whether two source keys draw the same stagger bucket for every stream.
///
/// Not a string comparison — an exact statement about
/// [`stagger_bucket`]'s algebra. The absorb is affine in each
/// byte modulo `STAGGER_BUCKETS`, and `x ^ 0x20` is `x + 32 (mod 64)` with
/// `51 * 32 == 32 (mod 64)`, so flipping bit 5 of one absorbed byte XORs 0x20
/// through the whole chain and flipping it in a second byte cancels that out.
/// Two keys therefore share every bucket exactly when they differ only in
/// bit 5, in an EVEN number of positions.
///
/// In printable ASCII bit 5 is the case bit, so among realistic label values
/// this is: the same labels typed with different capitalization, in an even
/// number of letters. `arm=valkey`/`arm=VALKEY` (six letters) shares all 26
/// buckets; `arm=redis`/`arm=REDIS` (five) shares none.
///
/// This exists so the recorder can WARN about the one situation the aliasing
/// can be reached in, which is far cheaper than hashing around it — see the
/// note on `stagger_bucket` for the measurements behind that choice.
pub fn staggers_identically(a: &str, b: &str) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut bit5_flips = 0usize;
    for (x, y) in a.bytes().zip(b.bytes()) {
        match x ^ y {
            0 => {}
            0x20 => bit5_flips += 1,
            // Any other difference reaches bits the absorb does not cancel.
            _ => return false,
        }
    }
    // `%` rather than `is_multiple_of`, which needs Rust 1.87; the crate
    // supports 1.85, the floor its dependencies set.
    bit5_flips % 2 == 0
}

/// A source's stagger identity: its label set, canonically rendered.
///
/// `BTreeMap` already fixes the order, so this is just a rendering — but it is
/// done in one place so the writer and any test agree on the exact bytes.
pub fn source_stagger_key(labels: &std::collections::BTreeMap<String, String>) -> String {
    let mut out = String::new();
    for (k, v) in labels {
        if !out.is_empty() {
            out.push('\u{1}');
        }
        out.push_str(k);
        out.push('=');
        out.push_str(v);
    }
    out
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The offset must be reproducible across runs, builds and Rust versions,
    /// which is why it is a hand-written FNV-1a and not `DefaultHasher`. The
    /// literals pin the constants: if the hash changes, every source's
    /// segment boundaries move.
    /// The realistic stream set, for spread and aliasing checks alike.
    const STREAMS: [&str; 26] = [
        "cpu_usage",
        "cpu_perf",
        "cpu_bandwidth",
        "cpu_migrations",
        "cpu_tlb_flush",
        "cpu_frequency",
        "scheduler",
        "blockio_latency",
        "blockio_requests",
        "network_interfaces",
        "network_traffic",
        "syscall_counts",
        "syscall_latency",
        "tcp_connect_latency",
        "tcp_packet_latency",
        "tcp_receive",
        "tcp_retransmit",
        "tcp_traffic",
        "memory",
        "page_cache",
        "gpu_nvidia",
        "softirq",
        "weather",
        "cgroup_cpu",
        "cgroup_memory",
        "filesystem",
    ];

    /// `staggers_identically` must agree with the hash it claims to describe.
    ///
    /// It is an algebraic shortcut — "differ only in bit 5, an even number of
    /// times" — standing in for "draws the same bucket for every stream". A
    /// shortcut that drifted from the function it describes would make the
    /// recorder warn about safe pairs, or stay silent on lockstep, and nothing
    /// else would notice. So it is checked against `stagger_bucket` itself.
    /// Alignment is about where the edges land, and it is advice: nothing
    /// here seals. A caller asking before it appends gets exact boundaries.
    #[test]
    fn alignment_cuts_on_the_bucket_a_row_belongs_to() {
        let policy = SealPolicy {
            max_bytes: usize::MAX,
            max_rows: usize::MAX,
            max_age: Duration::from_secs(3600),
            align: Some(100),
        };
        let mut a = SegmentAccount::open_first("s", SINGLE_SOURCE_KEY, &policy);
        // The first row starts whatever bucket it is in — never a boundary.
        assert!(!a.starts_new_bucket(250));
        a.add_row(1, 250);
        assert!(!a.starts_new_bucket(299), "same bucket");
        a.add_row(1, 299);
        assert!(!a.is_due(Instant::now()), "one bucket, no cap reached");
        assert!(a.starts_new_bucket(300), "300 opens the next bucket");

        // A caller that asks first seals here and rotates: clean edges.
        a.rotate(&policy, Instant::now());
        a.add_row(1, 300);
        assert!(!a.is_due(Instant::now()));

        // A caller that does not ask lets the boundary row in, and `is_due`
        // says so afterwards rather than letting the segment run to a cap.
        let mut b = SegmentAccount::open_first("s", SINGLE_SOURCE_KEY, &policy);
        b.add_row(1, 299);
        b.add_row(1, 300);
        assert!(b.is_due(Instant::now()), "two buckets in one segment");
    }

    /// Timestamps are signed, and truncating division would put the two rows
    /// either side of the epoch in one bucket.
    #[test]
    fn alignment_buckets_negative_timestamps_correctly() {
        let policy = SealPolicy {
            max_bytes: usize::MAX,
            max_rows: usize::MAX,
            max_age: Duration::from_secs(3600),
            align: Some(100),
        };
        let mut a = SegmentAccount::open_first("s", SINGLE_SOURCE_KEY, &policy);
        a.add_row(1, -1);
        assert!(a.starts_new_bucket(1), "-1 and 1 are not the same bucket");
        assert!(!a.starts_new_bucket(-100), "-100..-1 is one bucket");
    }

    /// No alignment asked for: nothing about the caps changes, and the
    /// timestamp passed to `add_row` is not consulted at all.
    #[test]
    fn without_alignment_timestamps_do_not_affect_sealing() {
        let policy = SealPolicy {
            max_bytes: usize::MAX,
            max_rows: 4,
            max_age: Duration::from_secs(3600),
            align: None,
        };
        let mut a = SegmentAccount::open_first("s", SINGLE_SOURCE_KEY, &policy);
        for ts in [0, 1_000_000, -5, 7] {
            assert!(!a.starts_new_bucket(ts));
            a.add_row(1, ts);
        }
        assert!(a.is_due(Instant::now()), "the row cap still applies");
    }

    #[test]
    fn staggers_identically_agrees_with_the_hash() {
        let cases = [
            // Even bit-5 flips: every stream collides.
            ("arm=valkey", "arm=VALKEY"),
            ("arm=Ab", "arm=aB"),
            ("host=Web-01", "host=weB-01"),
            // Odd: none do.
            ("arm=redis", "arm=REDIS"),
            ("arm=ab", "arm=aB"),
            // Not a case difference at all.
            ("host=web-01", "host=webm01"),
            ("host=web-01", "host=web-02"),
            ("arm=a", "arm=ab"),
            // Identical.
            ("arm=a", "arm=a"),
        ];
        for (a, b) in cases {
            let predicted = staggers_identically(a, b);
            let shares_all = STREAMS
                .iter()
                .all(|s| stagger_bucket(s, a) == stagger_bucket(s, b));
            assert_eq!(
                predicted, shares_all,
                "staggers_identically({a:?}, {b:?}) said {predicted}, but the hash \
                 shares-all is {shares_all}"
            );
        }
    }

    /// The property the stagger exists for, stated as a number so a future
    /// change to the hash has to answer for it.
    ///
    /// Colliding stream-pairs, normalized by what a uniform random assignment
    /// would give: 0 = every table its own bucket, 1.0 = random. This hash
    /// spreads a 12-stream source PERFECTLY and a 26-stream one better
    /// than random — which is the reason bit-5 aliasing is warned about rather
    /// than hashed away, since every candidate that closed it measured at or
    /// worse than random here.
    #[test]
    fn the_stagger_spreads_a_real_stream_set_better_than_random() {
        fn clumping(key: &str, n: usize) -> f64 {
            let mut counts = [0u32; STAGGER_BUCKETS as usize];
            for s in &STREAMS[..n] {
                counts[stagger_bucket(s, key) as usize] += 1;
            }
            let pairs: f64 = counts
                .iter()
                .map(|&c| f64::from(c) * (f64::from(c) - 1.0) / 2.0)
                .sum();
            let expected = (n as f64) * (n as f64 - 1.0) / 2.0 / STAGGER_BUCKETS as f64;
            pairs / expected
        }

        // Over many sources, not one: the bucket a stream draws depends on
        // both, and a hash that spread well for a single key would say nothing.
        for i in 0..200 {
            let key = format!("host=web-{i:03}\u{1}source=weather");
            assert_eq!(
                clumping(&key, 12),
                0.0,
                "12 streams must each get their own bucket ({key})"
            );
            assert!(
                clumping(&key, 26) < 1.0,
                "26 streams must clump less than random ({key})"
            );
        }
    }

    #[test]
    fn stagger_is_deterministic() {
        assert_eq!(stagger_bucket("cpu_usage", SINGLE_SOURCE_KEY), 32);
        assert_eq!(stagger_bucket("scheduler", SINGLE_SOURCE_KEY), 19);
        assert!(
            (0..STAGGER_BUCKETS).contains(&stagger_bucket("anything_at_all", SINGLE_SOURCE_KEY))
        );
    }

    /// Keys that differ only in the bits `% STAGGER_BUCKETS` discards must
    /// still separate.
    ///
    /// Plain FNV-1a reduced mod 64 depends only on each byte's low six bits,
    /// so two hostnames agreeing byte-for-byte modulo 0x40 drew the same
    /// bucket for every stream — complete lockstep between two sources
    /// that look nothing alike. The pairs are ordinary: `-`/`m`, `.`/`n`,
    /// digits against `p`-`y`.
    #[test]
    fn hosts_that_alias_in_the_low_bits_still_desync() {
        let key = |host: &str| {
            source_stagger_key(
                &[
                    ("host".to_string(), host.to_string()),
                    ("source".to_string(), "weather".to_string()),
                ]
                .into_iter()
                .collect(),
            )
        };
        let streams = [
            "cpu_usage",
            "scheduler",
            "blockio_latency",
            "tcp_traffic",
            "syscall_latency",
            "cpu_bandwidth",
        ];
        // Each pair differs only in bits 6-7 of one byte.
        for (a, b) in [
            ("web-01", "webm01"),
            ("node1", "nodeq"),
            ("web.01", "webn01"),
        ] {
            let (ka, kb) = (key(a), key(b));
            let collisions = streams
                .iter()
                .filter(|s| stagger_bucket(s, &ka) == stagger_bucket(s, &kb))
                .count();
            assert_eq!(
                collisions,
                0,
                "{a} and {b} share {collisions} of {} buckets — the reduction is \
                 discarding the bits that separate them. `collisions < len` would be \
                 too weak a bar here: 5 of 6 coincident is still the lockstep this \
                 test exists to catch",
                streams.len()
            );
        }
    }

    /// The reason the key widened past the stream name.
    ///
    /// Two producers have identical stream sets. Keyed on the stream
    /// alone, every table in one source drew its namesake's bucket in the
    /// other, so both sealed in permanent lockstep — doubling the co-seal batch
    /// exactly when the archive holds twice the tables. This is the assertion
    /// that fails if the source key is ever dropped from the hash.
    #[test]
    fn two_sources_do_not_share_a_streams_bucket() {
        let a = source_stagger_key(
            &[("host".to_string(), "alpha".to_string())]
                .into_iter()
                .collect(),
        );
        let b = source_stagger_key(
            &[("host".to_string(), "beta".to_string())]
                .into_iter()
                .collect(),
        );

        // Every stream the two hosts share must land somewhere different.
        let shared = ["cpu_usage", "scheduler", "blockio_latency", "tcp_traffic"];
        let collisions = shared
            .iter()
            .filter(|s| stagger_bucket(s, &a) == stagger_bucket(s, &b))
            .count();
        assert_eq!(
            collisions, 0,
            "identical stream sets must not draw identical buckets across sources"
        );
    }

    /// The byte cap must be staggered too, not just rows and age.
    ///
    /// This is the one the measurement caught. The byte cap splits the *wide*
    /// tables, so leaving it on the shared policy left exactly those tables
    /// with no phase offset: two sources of one producer carry identical data,
    /// reach the cap on the same row, and seal together forever. Measured
    /// before the fix, `cpu_usage` had 49 of 49 segment boundaries coincident
    /// across two sources; after, 1 of 5.
    #[test]
    fn the_byte_cap_is_staggered_across_sources() {
        let policy = SealPolicy {
            max_bytes: 8 * 1024 * 1024,
            max_rows: 900,
            max_age: Duration::from_secs(300),
            align: None,
        };
        let key = |host: &str| {
            source_stagger_key(
                &[("host".to_string(), host.to_string())]
                    .into_iter()
                    .collect(),
            )
        };

        // Drive the seal, don't just compare the field. The bug was never a
        // missing field — it was `is_due` reading `policy.max_bytes` instead
        // of the account's own copy, which a field comparison cannot see.
        let mut a = SegmentAccount::open_first("cpu_usage", &key("alpha"), &policy);
        let mut b = SegmentAccount::open_first("cpu_usage", &key("beta"), &policy);
        let now = Instant::now();
        // Rows wide enough that the byte cap is what fires, well before
        // `max_rows`: 8 MiB / 900 rows is ~9 KiB per row, so 64 KiB rows are
        // byte-bound by construction.
        let mut split = None;
        for row in 1..=policy.max_rows {
            a.add_row(64 * 1024, 0);
            b.add_row(64 * 1024, 0);
            if a.is_due(now) != b.is_due(now) {
                split = Some(row);
                break;
            }
            // Checked only once they agree, so an unstaggered byte cap falls
            // through to the `split.is_some()` assertion below with its
            // accurate message rather than tripping this one first.
            assert!(
                row < policy.max_rows,
                "the byte cap must fire before the row cap, or this tests the wrong cap"
            );
        }
        assert!(
            split.is_some(),
            "both sources' byte-bound tables sealed on the same row — the byte \
             cap escaped the stagger"
        );

        // And the stagger stays inside its documented bound: at most 50% off,
        // never zero.
        for host in ["alpha", "beta", "gamma"] {
            let acct = SegmentAccount::open_first("cpu_usage", &key(host), &policy);
            assert!(acct.byte_target() > policy.max_bytes / 2 - 1);
            assert!(acct.byte_target() <= policy.max_bytes);
        }
    }

    /// An A/B on one host separates only on `arm` — which is why the key is the
    /// whole label set and not the node name.
    #[test]
    fn same_host_different_arms_still_desync() {
        let base = |arm: &str| {
            source_stagger_key(
                &[
                    ("host".to_string(), "alpha".to_string()),
                    ("arm".to_string(), arm.to_string()),
                ]
                .into_iter()
                .collect(),
            )
        };
        let (a, b) = (base("redis"), base("valkey"));
        assert_ne!(a, b, "the arm label must reach the key");
        assert_ne!(
            stagger_bucket("cpu_usage", &a),
            stagger_bucket("cpu_usage", &b)
        );
    }

    /// The bucket follows a source's labels, not its position.
    ///
    /// This is deliberately not written as "compute the pair in both orders
    /// and compare". `stagger_bucket` takes two `&str` and no index, and
    /// `source_stagger_key` takes a `BTreeMap` that is sorted before it is
    /// called — so any such assertion reduces to `[f(a), f(b)] == [f(a),
    /// f(b))]`, two calls to a pure function compared with themselves. It
    /// cannot fail for any implementation, which is exactly what was wrong
    /// with the version this replaces.
    ///
    /// Order-independence is real, but it is a property of the *layer above*:
    /// the source id — the thing that would have made segmentation depend
    /// on endpoint order — never reaches this function, which is the design
    /// decision itself. `stagger_key_follows_the_labels_not_the_open_order`
    /// in `rez_v3_writer` pins it where the id exists. What is left to assert
    /// here is the content: two arms must land in different buckets.
    #[test]
    fn the_two_arms_of_an_ab_land_in_different_buckets() {
        let key = |arm: &str| {
            source_stagger_key(
                &[
                    ("host".to_string(), "alpha".to_string()),
                    ("arm".to_string(), arm.to_string()),
                ]
                .into_iter()
                .collect(),
            )
        };
        assert_ne!(
            stagger_bucket("cpu_usage", &key("redis")),
            stagger_bucket("cpu_usage", &key("valkey"))
        );
    }
}