Repository navigation
Conversation
cablehead
force-pushed
the
ttl/count
branch
from
September 29, 2026 17:29
3de11a0 to
7837d92
Compare
Owner
Author
|
Parking the adversarial review's probe tests here before the box they live on goes away. They are the only copy. What they areEleven tests reproducing the findings of a review run against Where they stand nowI re-ran all eleven against
A 200k-frame topic, debug build. The one-off count scan runs under the lock that every writer takes, so it blocks writes to every topic in the store, not just that one. Bounded by topic size, paid once per topic per store.
The review called this inherited from
It trips the first assertion, before reaching what it was written to catch: After importing the id under a second topic, it is still indexed under both, and the stream row survives the trim the probe expects to remove it. This branch does carry the orphan fix from #161 and has its own CaveatsMeasured at The patch (apply to a scratch worktree)diff --git a/src/store/tests.rs b/src/store/tests.rs
index 113ff056..203840c0 100644
--- a/src/store/tests.rs
+++ b/src/store/tests.rs
@@ -3216,3 +3216,457 @@ mod tests_last_count {
}
}
}
+
+// ---------------------------------------------------------------------------
+// Adversarial probes. Not part of the branch.
+// ---------------------------------------------------------------------------
+mod probe_count {
+ use super::*;
+
+ use tempfile::TempDir;
+ use tokio::time::sleep;
+
+ fn store_at(path: std::path::PathBuf) -> Store {
+ let options = StoreOptions::builder()
+ .ttl_sweep(Duration::from_secs(600))
+ .build();
+ Store::open(path, options).unwrap()
+ }
+
+ fn store_without_tick() -> Store {
+ store_at(TempDir::new().unwrap().keep())
+ }
+
+ fn topic_len(store: &Store, topic: &str) -> usize {
+ let options = ReadOptions::builder().topic(topic.to_string()).build();
+ store.read_sync(options).count()
+ }
+
+ fn index_len(store: &Store, topic: &str) -> u64 {
+ let (low, high) = idx_topic_range(idx_topic_key_prefix(topic), None);
+ store.idx_topic.range((low, high)).count() as u64
+ }
+
+ // --- A. dotted topics -------------------------------------------------
+
+ #[tokio::test]
+ async fn probe_a_dotted_topic_counts_and_trims() {
+ let store = store_without_tick();
+ for _ in 0..20 {
+ store
+ .append(Frame::builder("a.b.c").ttl(TTL::Last(5)).build())
+ .unwrap();
+ }
+ store.wait_for_gc().await;
+ assert_eq!(topic_len(&store, "a.b.c"), 5, "dotted topic held");
+ assert_eq!(
+ store.count_of("a.b.c"),
+ Some(index_len(&store, "a.b.c")),
+ "dotted topic count vs index"
+ );
+ // Prefix index entries must be cleaned up with the frames.
+ let (low, high) = idx_topic_range(idx_topic_key_prefix_probe("a."), None);
+ assert_eq!(store.idx_topic.range((low, high)).count(), 5, "a. prefix");
+ let (low, high) = idx_topic_range(idx_topic_key_prefix_probe("a.b."), None);
+ assert_eq!(store.idx_topic.range((low, high)).count(), 5, "a.b. prefix");
+ }
+
+ // The prefix keys use the raw prefix (with its trailing dot) + NUL.
+ fn idx_topic_key_prefix_probe(prefix: &str) -> Vec<u8> {
+ let mut v = prefix.as_bytes().to_vec();
+ v.push(0);
+ v
+ }
+
+ // --- B. a drain that sweeps and trims the same topic ------------------
+
+ /// A `last:n` topic over its cap, with one expired frame on it that the
+ /// same gc drain is sweeping. The trim subtracts that frame from the
+ /// count and then scans it back in, so the drain removes one fewer frame
+ /// than it decided to, and nothing re-queues.
+ #[tokio::test]
+ async fn probe_b_sweep_and_trim_in_one_drain_stops_short() {
+ let store = store_without_tick();
+
+ // Five plain frames, then one that expires, then five more.
+ for _ in 0..5 {
+ store.append(Frame::builder("test").build()).unwrap();
+ }
+ store
+ .append(
+ Frame::builder("test")
+ .ttl(TTL::Time(Duration::from_millis(1)))
+ .build(),
+ )
+ .unwrap();
+ for _ in 0..5 {
+ store.append(Frame::builder("test").build()).unwrap();
+ }
+ assert_eq!(index_len(&store, "test"), 11);
+
+ // First last:n on the topic: counts 11, removes one inline, hands the
+ // rest to the gc worker.
+ store
+ .append(Frame::builder("test").ttl(TTL::Last(5)).build())
+ .unwrap();
+ assert_eq!(store.count_of("test"), Some(11));
+
+ // Let the middle frame expire and queue a sweep, so the drain that
+ // runs the trim also sweeps this topic.
+ sleep(Duration::from_millis(20)).await;
+ store.sweep();
+ store.wait_for_gc().await;
+ store.wait_for_gc().await;
+
+ assert_eq!(
+ index_len(&store, "test"),
+ 5,
+ "the topic is still over its cap after the drain settled"
+ );
+ assert_eq!(topic_len(&store, "test"), 5);
+ }
+
+ // --- C. the durable count row -----------------------------------------
+
+ /// A `last:n` landing on a topic that already holds exactly `n` frames
+ /// never writes the count row, so the one-off scan is repaid on every
+ /// restart.
+ #[tokio::test]
+ async fn probe_c_count_row_is_written_for_a_late_last_n() {
+ let dir = TempDir::new().unwrap().keep();
+ let store = store_at(dir.clone());
+ for _ in 0..5 {
+ store.append(Frame::builder("test").build()).unwrap();
+ }
+ store
+ .append(Frame::builder("test").ttl(TTL::Last(5)).build())
+ .unwrap();
+ store.wait_for_gc().await;
+ assert_eq!(store.count_of("test"), Some(5));
+ assert_eq!(index_len(&store, "test"), 5);
+
+ println!(
+ "PROBE-C meta row for test: {:?}",
+ store.meta.get(count_key("test")).unwrap().is_some()
+ );
+
+ store.flush().unwrap();
+ drop(store);
+
+ let store = store_at(dir);
+ let before = store.stats.trim_scanned.load(Ordering::Relaxed);
+ store
+ .append(Frame::builder("test").ttl(TTL::Last(5)).build())
+ .unwrap();
+ assert_eq!(
+ store.stats.trim_scanned.load(Ordering::Relaxed) - before,
+ 1,
+ "the reopened store rescanned the whole topic"
+ );
+ }
+
+ // --- D. sweeper racing the inline trim --------------------------------
+
+ /// The gc worker reads the expiry index outside `write_lock` and removes
+ /// what it found inside it. An append that trims one of those frames in
+ /// between leaves the drain decrementing the count for a frame that is
+ /// already gone.
+ #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+ async fn probe_d_sweeper_racing_the_inline_trim() {
+ let store = store_without_tick();
+ store
+ .append(Frame::builder("test").ttl(TTL::Last(5)).build())
+ .unwrap();
+
+ // A backlog of expiring frames on the same counted topic.
+ for _ in 0..4000 {
+ store
+ .append(
+ Frame::builder("test")
+ .ttl(TTL::Time(Duration::from_millis(1)))
+ .build(),
+ )
+ .unwrap();
+ }
+ sleep(Duration::from_millis(30)).await;
+
+ let sweeper = {
+ let store = store.clone();
+ std::thread::spawn(move || {
+ for _ in 0..400 {
+ store.sweep();
+ std::thread::yield_now();
+ }
+ })
+ };
+ let writer = {
+ let store = store.clone();
+ std::thread::spawn(move || {
+ for _ in 0..2000 {
+ store
+ .append(Frame::builder("test").ttl(TTL::Last(5)).build())
+ .unwrap();
+ }
+ })
+ };
+ sweeper.join().unwrap();
+ writer.join().unwrap();
+
+ store.sweep();
+ store.wait_for_gc().await;
+ store.wait_for_gc().await;
+
+ let claimed = store.count_of("test");
+ let actual = index_len(&store, "test");
+ assert_eq!(claimed, Some(actual), "count {claimed:?} vs index {actual}");
+ }
+
+ // --- E. import with a reused id ---------------------------------------
+
+ /// `POST /import` takes the frame's id and topic from the request body.
+ /// Sending an id that is already in the stream under a different topic
+ /// adds an index entry to the new topic without counting it.
+ #[tokio::test]
+ async fn probe_e_import_reusing_an_id_under_another_topic() {
+ let store = store_without_tick();
+ let keep = store
+ .append(Frame::builder("b").ttl(TTL::Last(3)).build())
+ .unwrap();
+ let a = store.append(Frame::builder("a").build()).unwrap();
+ assert_eq!(store.count_of("b"), Some(1));
+
+ // Same id, different topic, as an import would deliver it.
+ let mut moved = a.clone();
+ moved.topic = "b".to_string();
+ moved.ttl = Some(TTL::Last(3));
+ store.insert_frame(&moved).unwrap();
+
+ assert_eq!(
+ store.count_of("b"),
+ Some(index_len(&store, "b")),
+ "count for b disagrees with its index",
+ );
+ let _ = keep;
+ }
+
+ // --- F. the race in D, expressed serially -----------------------------
+
+ /// The gc worker reads the expiry index before it takes `write_lock` and
+ /// removes what it read after. An append that lands in that window and
+ /// trims the same frame inline leaves `remove_many` decrementing the
+ /// count for a frame that is already gone.
+ #[tokio::test]
+ async fn probe_f_sweeper_removals_applied_after_an_inline_trim() {
+ let store = store_without_tick();
+ store
+ .append(
+ Frame::builder("test")
+ .ttl(TTL::Time(Duration::from_millis(1)))
+ .build(),
+ )
+ .unwrap();
+ for _ in 0..4 {
+ store
+ .append(Frame::builder("test").ttl(TTL::Last(5)).build())
+ .unwrap();
+ }
+ store.wait_for_gc().await;
+ assert_eq!(store.count_of("test"), Some(5));
+ assert_eq!(index_len(&store, "test"), 5);
+ sleep(Duration::from_millis(20)).await;
+
+ // Exactly what the gc drain reads, before it takes write_lock.
+ let expired = store.expired_at(now_ms(), 100);
+ assert_eq!(expired.len(), 1);
+
+ // An append lands in that window and trims the same frame inline.
+ store
+ .append(Frame::builder("test").ttl(TTL::Last(5)).build())
+ .unwrap();
+ assert_eq!(index_len(&store, "test"), 5);
+ assert_eq!(store.count_of("test"), Some(5));
+
+ // The drain now removes what it read.
+ store.remove_many(expired).unwrap();
+
+ assert_eq!(
+ store.count_of("test"),
+ Some(index_len(&store, "test")),
+ "the drain decremented for a frame the append had already removed",
+ );
+ }
+
+ // --- G. a plain frame on a counted topic ------------------------------
+
+ /// Frames with no `last:n` still take a keep slot, but nothing schedules
+ /// a trim for them, so the topic sits over its cap until some other
+ /// `last:n` append happens to arrive.
+ #[tokio::test]
+ async fn probe_g_a_plain_frame_leaves_the_topic_over_its_cap() {
+ let store = store_without_tick();
+ for _ in 0..3 {
+ store
+ .append(Frame::builder("test").ttl(TTL::Last(3)).build())
+ .unwrap();
+ }
+ store.wait_for_gc().await;
+ for _ in 0..5 {
+ store.append(Frame::builder("test").build()).unwrap();
+ }
+ store.wait_for_gc().await;
+ store.wait_for_gc().await;
+ assert_eq!(index_len(&store, "test"), 3, "topic is over its cap");
+ }
+
+ // --- H. append stalled by a gc drain ----------------------------------
+
+ /// The gc worker holds `write_lock` for the whole of a drain, so an
+ /// append waits behind a trim of a large backlog.
+ #[tokio::test]
+ async fn probe_h_append_stalls_behind_a_drain() {
+ let store = store_without_tick();
+ for _ in 0..40_000 {
+ store.append(Frame::builder("test").build()).unwrap();
+ }
+ // The first last:n counts the topic and hands the backlog to gc.
+ store
+ .append(Frame::builder("test").ttl(TTL::Last(5)).build())
+ .unwrap();
+
+ let started = std::time::Instant::now();
+ store
+ .append(Frame::builder("test").ttl(TTL::Last(5)).build())
+ .unwrap();
+ let waited = started.elapsed();
+ store.wait_for_gc().await;
+ assert!(
+ waited < Duration::from_millis(5),
+ "an append waited {waited:?} behind the gc drain",
+ );
+ }
+
+ // --- I. a count that reads HIGH, from the import endpoint -------------
+
+ /// `insert_frame` decides whether a frame is new by looking in `stream`,
+ /// but the count tracks `idx_topic`. An import that reuses an id under a
+ /// second topic leaves an index entry under that topic which `stream` no
+ /// longer backs. Re-importing it then counts a frame that adds no index
+ /// entry, and the count reads high: the next trim deletes a live frame.
+ #[tokio::test]
+ async fn probe_i_import_can_push_the_count_above_the_index() {
+ let dir = TempDir::new().unwrap().keep();
+
+ let doomed;
+ {
+ let store = store_at(dir.clone());
+ // A counted topic "a" with a cap of one, to trim the shared frame.
+ let a = store
+ .append(Frame::builder("a").ttl(TTL::Last(1)).build())
+ .unwrap();
+
+ // Import the same id under topic "b": a new index entry under b,
+ // uncounted, because `stream` already holds the id.
+ let mut moved = a.clone();
+ moved.topic = "b".to_string();
+ moved.ttl = Some(TTL::Last(3));
+ store.insert_frame(&moved).unwrap();
+
+ // The next append to "a" trims the shared frame out of `stream`,
+ // leaving b's index entry with nothing behind it.
+ store
+ .append(Frame::builder("a").ttl(TTL::Last(1)).build())
+ .unwrap();
+ store.wait_for_gc().await;
+ assert_eq!(store.get(&a.id), None, "the stream entry is gone");
+ assert_eq!(index_len(&store, "b"), 1, "b still indexes it");
+
+ store.flush().unwrap();
+ doomed = a.id;
+ }
+
+ // Reopen: b has no meta row, so its count is re-derived by scanning
+ // its index, which includes the entry with nothing behind it.
+ let store = store_at(dir);
+ let keeper = store
+ .append(Frame::builder("b").ttl(TTL::Last(3)).build())
+ .unwrap();
+ let second = store
+ .append(Frame::builder("b").ttl(TTL::Last(3)).build())
+ .unwrap();
+ store.wait_for_gc().await;
+ assert_eq!(store.count_of("b"), Some(index_len(&store, "b")));
+
+ // Re-import the frame b still indexes. `stream` does not hold it, so
+ // it counts as new, but the index entry is already there.
+ let mut moved = Frame::builder("b").ttl(TTL::Last(3)).build();
+ moved.id = doomed;
+ store.insert_frame(&moved).unwrap();
+ store.wait_for_gc().await;
+
+ let claimed = store.count_of("b").unwrap();
+ let actual = index_len(&store, "b");
+ println!("PROBE-I count for b: {claimed} claimed, {actual} in the index");
+
+ // The cap is three and the topic holds two, so the next append must
+ // not remove anything. The count says otherwise.
+ let third = store
+ .append(Frame::builder("b").ttl(TTL::Last(3)).build())
+ .unwrap();
+ store.wait_for_gc().await;
+
+ let held: Vec<Scru128Id> = store
+ .read_sync(ReadOptions::builder().topic("b".to_string()).build())
+ .map(|f| f.id)
+ .collect();
+ println!("PROBE-I held on b: {} frames", held.len());
+ assert!(
+ held.contains(&keeper.id),
+ "a live frame was deleted from a topic holding fewer than its cap",
+ );
+ let _ = (second, third);
+ }
+ // --- J. the one-off scan, under the write lock ------------------------
+
+ #[tokio::test]
+ async fn probe_j_first_last_n_scan_cost() {
+ let store = store_without_tick();
+ for _ in 0..200_000 {
+ store.append(Frame::builder("test").build()).unwrap();
+ }
+ let started = std::time::Instant::now();
+ store
+ .append(Frame::builder("test").ttl(TTL::Last(5)).build())
+ .unwrap();
+ let waited = started.elapsed();
+ println!("PROBE-J first last:n append on a 200k topic took {waited:?}");
+ store.wait_for_gc().await;
+ assert!(waited < Duration::from_millis(1), "took {waited:?}");
+ }
+
+ // --- K. a corrupt record now poisons the store's only write lock ------
+
+ /// `remove_many` did not take a lock before this branch. It now takes
+ /// `write_lock`, and it point-reads every `Removal::Id` through `get`,
+ /// which panics on a record it cannot deserialize. The panic poisons the
+ /// lock every write and the gc worker take, so one bad record stops the
+ /// whole store rather than one call.
+ #[tokio::test]
+ async fn probe_k_a_corrupt_record_poisons_write_lock() {
+ let store = store_without_tick();
+ let frame = store
+ .append(Frame::builder("test").ttl(TTL::Last(5)).build())
+ .unwrap();
+ // A record `get` cannot deserialize.
+ store.stream.insert(frame.id.as_bytes(), b"not json").unwrap();
+
+ let store2 = store.clone();
+ let id = frame.id;
+ let hit = std::thread::spawn(move || store2.remove(&id)).join();
+ assert!(hit.is_err(), "expected the deserialize panic");
+
+ // Everything else still works.
+ store
+ .append(Frame::builder("other").ttl(TTL::Last(5)).build())
+ .unwrap();
+ }
+}
|
This branch has not been deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
last:nkeeps the newestnframes on a topic. To find where to cut, the trim reverse-scanned the topic's index counting backnentries, on every append that carried alast:n.That is
nindex entries read to delete one frame. At a five million frame topic it is 759 ms of gc worker time per append, and the worker cannot keep up: sustained appends leave the topic permanently over its limit, with no path back down.Change
Keep a live frame count per topic, so the trim knows how many are over without looking.
metakeyspace holds a format version and one count row per counted topic, read with a singleiter()at openlast:nlands on it, by counting what is there once, under the write lockappend_lockbecomeswrite_lockand now covers every path that changes a topic's frame population:append,insert_frame,remove_many, and the gc worker's drainOne entry per append at every size: the scan is gone, not divided by a constant. Append latency goes down as well, because the gc worker is no longer saturating a core and contending for the database.
Breaking change
Stores written before this have no counts and no way to produce them, so
Store::openrefuses them withStoreError::Version, naming the version found and the version needed and pointing at Discord. There is no in-place upgrade.Reviewed, and what that found
An adversarial review of the first two commits found five defects, two of which could lose data. Each is fixed here with a test that fails without the fix.
last:5topic reached 274 frames with a count of 5.a_trim_sharing_a_drain_with_a_sweep_accounts_for_it_oncefails with either half alone: without the subtraction it over-trims and deletes frames the topic was told to keep; without the skip it under-trims and does not retry.last:nlanded on a topic already holding exactlyn, so the count was re-derived on every restart rather than once in the store's life. The count was going into memory before anything reached disk, which the rest of the design does not do.remove_manynow takeswrite_lock, so a corrupt frame found during a removal panicked while holding it, and every later write failed on the poisoned lock with an error that said nothing about corruption. It still panics, but the lock is taken past poisoning: a batch commits all or nothing and counts move only after a commit, so nothing it guards can be left half-changed. The panic message also names the frame now. Before, the error path read the record's key as UTF-8, and astreamkey is raw bytes, so it panicked on that first.insert_frameasked whether the id was instreamwhen it needed to ask whether the write adds an entry to this topic. Different answers once fix: move index entries when a restore changes a frame's topic or ttl #161 let a re-import move a frame between topics. That PR supplies the stored frame, so this is now a comparison rather than a guess, and moving a frame between two counted topics moves both counts.Also
last:ntrims on the append that carries it, not continuously, so a topic grows freely between such appends and the next one collapses it. That was undocumented and the docs claimed otherwise. It is a useful property, so it is written down and pinned by a test rather than changed.The
metricsbenchandcountbenchharnesses move fromsrc/bintoexamples/, wherecargo installdoes not pick them up as binaries besidexs.Tests
252 lib, 12 integration, 12 doc.
./scripts/check.shgreen,astro checkclean.