From 848721a70dc5758f9111abfc312e6c959df42ce8 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Mon, 19 Oct 2026 14:16:40 -0700 Subject: [PATCH] fix(store): boot counter, durable releases, clock marker --- tines-ros2/src/node.rs | 15 +-- tines-store/src/log.rs | 195 ++++++++++++++++++++----------- tines-store/tests/persistence.rs | 137 ++++++++++++++++++---- tines/src/main.rs | 10 +- tines/tests/ipc.rs | 8 +- 5 files changed, 256 insertions(+), 109 deletions(-) diff --git a/tines-ros2/src/node.rs b/tines-ros2/src/node.rs index ec571a24..7f0043c6 100644 --- a/tines-ros2/src/node.rs +++ b/tines-ros2/src/node.rs @@ -122,15 +122,16 @@ impl TinesNode { /// /// * `log_path` — where the backing [`LogStore`] lives (created if missing). /// * `peer` / `radio` — the BLE connection this node exchanges over. - /// * `first_seq` — the next tines observation sequence to use. /// * `topic_cap` — the maximum number of ROS2 topics to manage at once. /// - /// The ROS2 context is created internally via [`rclrs::Context::default_from_env`]; - /// the store's anchor tick is `0`; ages then advance under the `Tick` - /// values passed to [`TinesNode::poll`]. + /// The tines sequence numbers are seeded from the LogStore's boot counter + /// (`store.first_seq()`), so a resumed node never reuses a sequence + /// number. The ROS2 context is created internally via + /// [`rclrs::Context::default_from_env`]; the store's anchor tick is `0`; + /// ages then advance under the `Tick` values passed to + /// [`TinesNode::poll`]. pub fn new( me: NodeId, - first_seq: u64, log_path: &std::path::Path, peer: NodeId, radio: impl Radio + 'static, @@ -144,9 +145,9 @@ impl TinesNode { let executor = context.create_basic_executor(); let node = executor.create_node("tines")?; - let store = LogStore::new(log_path, 0, first_seq)?; + let store = LogStore::open(log_path, 0)?; let transport = BtTransport::new(me, peer, radio); - let tines = TinesNodeCore::new(me, first_seq, store, transport); + let tines = TinesNodeCore::new(me, store.first_seq(), store, transport); let receipt_pub = node.create_publisher::(RECEIPTS_TOPIC)?; let (inbound_tx, inbound_rx) = channel::(); diff --git a/tines-store/src/log.rs b/tines-store/src/log.rs index c4f2d278..667af3ee 100644 --- a/tines-store/src/log.rs +++ b/tines-store/src/log.rs @@ -12,14 +12,23 @@ //! inserted at (`stored_at`). Deduped on `id` at load. //! * `Record::Trim` — the id of an observation removed by `trim`; the last //! event for an id wins, so a re-inserted (new `Obs` record) id survives. -//! * `Record::Meta` — the current `next_seq` and the full set of released ids. -//! The last `Meta` on disk wins, so it always reflects the latest flushed -//! state. +//! * `Record::Release` — the id of an observation that was released; the last +//! event for an id wins, so a re-inserted (new `Obs` record) id becomes +//! un-released again. Appended immediately by [`Store::release`], so a +//! release survives a kill without a flush. +//! * `Record::Boot` — the boot counter, written by [`LogStore::open`] and +//! incremented on every open. `first_seq() = boot_count << 32`, which is +//! core's documented contract; a resumed node therefore never reuses a +//! sequence number. +//! * `Record::Clock` — a tick marker written by [`LogStore::flush`]. On load +//! each live observation's age is folded up to the last clock marker so +//! pre-reboot elapsed time is recovered (see [`LogStore::open`]). //! //! On load the index is rebuilt by replaying the log, then every age is //! rebased to load time: each entry's `age_ms` is folded by the time elapsed //! since its own `stored_at` (the tick it was inserted, carried on the `Obs` -//! record). Anchoring the fold to each entry's insert tick rather than to a +//! record) up to the last persisted clock marker, and then from the marker to +//! load time. Anchoring the fold to each entry's insert tick rather than to a //! single persisted clock marker makes the reopen fold exact for idle stores //! and for observations inserted before the last flush — ages continue across a //! reboot instead of freezing at (and under-counting from) their pre-shutdown @@ -29,7 +38,7 @@ //! //! * Torn trailing records. The log is append-only, so a crash mid-append can //! leave a partial trailing record — a length prefix whose body ran past -//! end-of-file — or a bit-roted/corrupt record. [`LogStore::new`] recovers by +//! end-of-file — or a bit-roted/corrupt record. [`LogStore::open`] recovers by //! truncating the file back to the last *complete* record boundary and //! loading what survived: the partial tail is dropped, complete records are //! preserved. A corrupt length prefix that claims an absurd body size is @@ -37,12 +46,14 @@ //! same way, so a malformed prefix cannot exhaust memory or make the store //! unloadable. //! -//! * Durability contract. `flush()` appends a `Record::Meta` and -//! `sync_all`s, making the released set, `next_seq`, and the appended -//! observation records durable across a power loss. `insert`, `release`, and -//! `trim` update the in-memory index and (for `insert`/`trim`) append -//! records, but those records reach disk only on the next `flush()`. Hosts -//! that need crash durability must `flush()` before shutdown. +//! * Durability contract. `flush()` appends a `Record::Clock` and `sync_all`s, +//! making the appended observation and release records durable across a power +//! loss. `insert`, `release`, and `trim` update the in-memory index and (for +//! `insert`/`trim`) append records, but those records reach disk only on the +//! next `flush()`. Hosts that need crash durability must `flush()` before +//! shutdown. `release`, by contrast, appends its `Release` record immediately +//! (best-effort: an append error is logged), so a single release is durable +//! across a kill-without-flush without waiting for a flush. use std::collections::{BTreeMap, BTreeSet, HashSet}; use std::fs::{File, OpenOptions}; @@ -78,9 +89,14 @@ enum Record { Trim { id: ObservationId, }, - Meta { - next_seq: u64, - released: Vec, + Release { + id: ObservationId, + }, + Boot { + count: u64, + }, + Clock { + tick: Tick, }, } @@ -94,26 +110,33 @@ struct Entry { /// A persistent, append-only [`Store`] backed by a single observation log. /// /// See the [module documentation](crate) for the on-disk record format and the -/// rebase-on-load behavior. +/// boot counter, durable releases, and rebase-on-load behavior. pub struct LogStore { file: File, index: BTreeMap, - next_seq: u64, + boot_count: u64, now: Tick, } impl LogStore { /// Open (creating if necessary) the log at `path` and rebuild the index. /// - /// `now` is the current tick and is used both as the base for rebasing ages - /// and as the store's current clock. `first_seq` is the next sequence - /// number to hand out; the persisted `next_seq` (if any) is honored when it - /// is larger, so a resumed run never reuses a sequence. - pub fn new(path: &Path, now: Tick, first_seq: u64) -> Result { + /// `now` is the current tick: it is the base for rebasing ages and the + /// store's current clock. On open the log's boot counter is read, + /// incremented, and durably written (`fsync`) before returning, so + /// [`Self::first_seq`] strictly increases across reboots and a resumed node + /// never reuses a sequence number. + /// + /// The boot counter is the number of times this log has been opened (a + /// fresh log starts at boot 1). It is exposed as + /// `first_seq() = boot_count << 32`, core's documented contract: a boot can + /// hand out up to 2³² sequence numbers before the high word would roll + /// over, which is more than any robot needs. + pub fn open(path: &Path, now: Tick) -> Result { // Open for reads (to load) and appends (to write later). Append mode is // correct because we load by reading the whole file below, and every // mutation appends to the end of the log. - let file = OpenOptions::new() + let mut file = OpenOptions::new() .read(true) .create(true) .append(true) @@ -128,7 +151,6 @@ impl LogStore { let mut index: BTreeMap = BTreeMap::new(); let mut dead: HashSet = HashSet::new(); - let mut next_seq = first_seq; let mut records_loaded: usize = 0; // Byte offset just past the last *complete* record replayed so far. A @@ -139,6 +161,11 @@ impl LogStore { let mut pos = 0usize; let mut good_end = 0usize; + // The boot counter read off the last `Record::Boot` on disk, and the + // tick of the last `Record::Clock` (the load-time age fold's anchor). + let mut last_boot_count: Option = None; + let mut last_clock_tick: Option = None; + loop { match parse_record(&data, pos) { Ok(Some((new_pos, rec))) => { @@ -146,14 +173,13 @@ impl LogStore { Record::Obs { id, obs, stored_at } => { // A preceding Trim for this id is cancelled by the // re-insert, so revive it; otherwise dedupe to the - // first (oldest) occurrence. Fold in the time - // elapsed since the record's own insert tick so the - // age is exact regardless of when the last flush - // happened. + // first (oldest) occurrence. The age is folded after + // the loop (once the last clock marker is known), so + // it starts from its stored, insert-time age. dead.remove(&id); index.entry(id).or_insert_with(|| Entry { - obs: obs.aged(now.saturating_sub(stored_at)), - stored_at: now, + obs: *obs, + stored_at, released: false, }); } @@ -161,17 +187,19 @@ impl LogStore { dead.insert(id); index.remove(&id); } - Record::Meta { - next_seq: ns, - released, - } => { - next_seq = ns.max(next_seq); - for rid in released { - if let Some(entry) = index.get_mut(&rid) { - entry.released = true; - } + Record::Release { id } => { + // Mark released if the id is still live; a later + // re-insert (a fresh `Obs`) makes it un-released. + if let Some(entry) = index.get_mut(&id) { + entry.released = true; } } + Record::Boot { count } => { + last_boot_count = Some(count); + } + Record::Clock { tick } => { + last_clock_tick = Some(tick); + } } records_loaded += 1; // Advance the last-complete-record boundary now that this @@ -211,33 +239,63 @@ impl LogStore { ); } + // Increment the boot counter and durably persist it before returning, so + // a later open reads a strictly higher value and first_seq() climbs. + let boot_count = last_boot_count.map(|c| c + 1).unwrap_or(1); + let record = Record::Boot { count: boot_count }; + write_record(&mut file, &record)?; + file.sync_all().map_err(|_| StoreError::Backend)?; + + // Fold each live observation's age up to the last clock marker and then + // to load time. Anchoring to each entry's insert tick keeps the fold + // exact even for observations inserted before the last flush (see the + // module docs). + for entry in index.values_mut() { + entry.obs.age_ms = match last_clock_tick { + Some(marker) if marker >= entry.stored_at => entry + .obs + .age_ms + .saturating_add(marker.saturating_sub(entry.stored_at)) + .saturating_add(now.saturating_sub(marker)), + _ => entry + .obs + .age_ms + .saturating_add(now.saturating_sub(entry.stored_at)), + }; + entry.stored_at = now; + } + Ok(LogStore { file, index, - next_seq, + boot_count, now, }) } - /// The next sequence number this store will hand out. - pub fn next_seq(&self) -> u64 { - self.next_seq + /// The sequence number the next `Node::observe` should use for this boot. + /// + /// Equals `boot_count << 32` (see [`LogStore::open`]). It strictly increases + /// across reboots, so a resumed node's sequence numbers can't collide + /// (content-id dedupe) with the previous boot's observations. + pub fn first_seq(&self) -> u64 { + self.boot_count << 32 } - /// Persist the current index state (released flags, `next_seq`, and the - /// observation log) to disk. + /// Persist the current index state to disk. /// - /// This is the durability boundary: `flush()` appends a `Record::Meta` - /// and calls `sync_all`, after which the released set, `next_seq`, and the - /// appended observation records are durable across a power loss. Because - /// `insert`, `release`, and `trim` only reach disk via a subsequent - /// `flush()`, hosts that need crash durability must call `flush()` before - /// shutdown (see the durability contract in the module docs). + /// This is the durability boundary: `flush()` advances each entry's age to + /// the current tick, appends a `Record::Clock` tick marker, and calls + /// `sync_all`, after which the appended observation and release records are + /// durable across a power loss. Because `insert` only reaches disk via a + /// subsequent `flush()`, hosts that need crash durability must call + /// `flush()` before shutdown (see the durability contract in the module + /// docs). pub fn flush(&mut self) -> Result<(), StoreError> { // Advance every entry's age up to this tick in memory, so each // `entry.obs.age_ms` is the true current age and `stored_at` marks when // that age was computed. Ages are recomputed from each entry's own - // `stored_at` on the next load (see [`LogStore::new`]), so this keeps + // `stored_at` on the next load (see [`LogStore::open`]), so this keeps // the in-memory representation coherent within this run. for entry in self.index.values_mut() { entry.obs.age_ms = entry @@ -247,18 +305,9 @@ impl LogStore { entry.stored_at = self.now; } - let mut released: Vec = self - .index - .iter() - .filter(|(_, entry)| entry.released) - .map(|(id, _)| *id) - .collect(); - released.sort_unstable(); - - let record = Record::Meta { - next_seq: self.next_seq, - released, - }; + // Record a clock marker so the load-time age fold has a durable anchor + // for the time elapsed up to this flush. + let record = Record::Clock { tick: self.now }; write_record(&mut self.file, &record)?; self.file.sync_all().map_err(|_| StoreError::Backend)?; Ok(()) @@ -284,7 +333,7 @@ fn write_record(file: &mut File, record: &Record) -> Result<(), StoreError> { /// /// This is a pure slice parser (no file I/O), so a torn or corrupt trailing /// record is detectable purely by byte offsets, which is exactly what -/// [`LogStore::new`] needs to truncate to the last complete boundary. +/// [`LogStore::open`] needs to truncate to the last complete boundary. /// /// * `Ok(None)` — clean end-of-file: `pos` is exactly at the end of `data`. /// * `Ok(Some((new_pos, rec)))` — a complete record was parsed; `new_pos` is @@ -294,7 +343,7 @@ fn write_record(file: &mut File, record: &Record) -> Result<(), StoreError> { /// declared body length exceeds [`MAX_RECORD_BYTES`] (rejected *before* any /// allocation, so a corrupt `0xFFFFFFFF` prefix cannot exhaust memory), the /// body runs past `data.len()`, or the body fails to decode. -/// [`LogStore::new`] treats this as a torn tail and truncates the file. +/// [`LogStore::open`] treats this as a torn tail and truncates the file. fn parse_record(data: &[u8], pos: usize) -> Result, StoreError> { // Clean EOF: nothing left. if pos == data.len() { @@ -349,7 +398,6 @@ impl Store for LogStore { released: false, }, ); - self.next_seq += 1; Ok(true) } @@ -399,13 +447,21 @@ impl Store for LogStore { /// Mark the observation with `id` as released: it has been sent and must /// not be offered for exchange again. /// - /// The mark is set in memory immediately, but is only persisted to disk on - /// the next [`Self::flush`]; without a flush the release is lost across a - /// power loss (see the durability contract in the module docs). + /// The mark is set in memory immediately and a `Record::Release` is + /// appended to the log at the same time (best-effort: an append error is + /// logged, not raised, keeping [`Store::release`] infallible). The release + /// is therefore durable across a kill-without-flush — unlike a flush-only + /// release, it does not wait for a later [`Self::flush`]. fn release(&mut self, id: &ObservationId) { if let Some(entry) = self.index.get_mut(id) { entry.released = true; } + // Best-effort durable append: `release` is infallible, so a backend + // failure is logged rather than raised (the worst case is a duplicate + // custodian after a reboot). + if let Err(e) = self.append(&Record::Release { id: *id }) { + eprintln!("tines-store: failed to append release record for {id:?}: {e:?}"); + } } fn is_released(&self, id: &ObservationId) -> bool { @@ -451,6 +507,7 @@ impl Store for LogStore { continue; } let mut ids = member_ids.clone(); + ids.sort_unstable(); ids.sort_by(|&a, &b| self.age_cmp(&a, &b, now)); for id in ids.into_iter().skip(keep) { removed.insert(id); diff --git a/tines-store/tests/persistence.rs b/tines-store/tests/persistence.rs index c537891a..18404b7e 100644 --- a/tines-store/tests/persistence.rs +++ b/tines-store/tests/persistence.rs @@ -47,7 +47,7 @@ fn reopen_preserves_store() { let path = store_path("reopen"); { - let mut s = LogStore::new(&path, 0, 1).expect("open fresh store"); + let mut s = LogStore::open(&path, 0).expect("open fresh store"); let a = obs([1; 16], 1, 0, Some(10)); let b = obs([2; 16], 2, 0, Some(20)); let ib = b.id(); @@ -57,19 +57,17 @@ fn reopen_preserves_store() { assert!(!s.is_released(&ib)); s.release(&ib); assert!(s.is_released(&ib)); - assert_eq!(s.next_seq(), 3); s.flush().expect("flush before drop"); } { - let s = LogStore::new(&path, 2000, 3).expect("reopen store"); + let s = LogStore::open(&path, 2000).expect("reopen store"); let a = obs([1; 16], 1, 0, Some(10)); let b = obs([2; 16], 2, 0, Some(20)); assert!(s.contains(&a.id())); assert!(s.contains(&b.id())); assert!(!s.is_released(&a.id())); assert!(s.is_released(&b.id())); - assert_eq!(s.next_seq(), 3); } } @@ -78,14 +76,14 @@ fn ages_rebase_on_load() { let path = store_path("ages"); { - let mut s = LogStore::new(&path, 1000, 1).unwrap(); + let mut s = LogStore::open(&path, 1000).unwrap(); let o = obs([1; 16], 1, 500, Some(10)); assert!(s.insert(o, 1000).unwrap()); s.flush().unwrap(); } { - let s = LogStore::new(&path, 5000, 1).unwrap(); + let s = LogStore::open(&path, 5000).unwrap(); let o = obs([1; 16], 1, 500, Some(10)); let id = o.id(); // 500ms original age plus the 4000ms that elapsed while the store was @@ -104,7 +102,7 @@ fn ages_rebase_after_nonidle_flush() { let path = store_path("ages_nonidle"); { - let mut s = LogStore::new(&path, 1000, 1).unwrap(); + let mut s = LogStore::open(&path, 1000).unwrap(); let a = obs([1; 16], 1, 100, Some(10)); assert!(s.insert(a, 1000).unwrap()); // Advance the store clock via a second, distinct insert so the flush @@ -120,7 +118,7 @@ fn ages_rebase_after_nonidle_flush() { // = 8100. A flush that fails to catch `age_ms` up folds the reopen // against the flush clock (3000) and under-counts by the 2000ms // between insert and flush, reporting only 6100. - let s = LogStore::new(&path, 9000, 1).unwrap(); + let s = LogStore::open(&path, 9000).unwrap(); let a = obs([1; 16], 1, 100, Some(10)); let g = s.get(&a.id(), 9000).unwrap(); assert_eq!(g.age_ms, 8100); @@ -128,22 +126,112 @@ fn ages_rebase_after_nonidle_flush() { } #[test] -fn next_seq_persists() { - let path = store_path("seq"); +fn ages_include_time_to_clock_marker() { + // A flush records a Clock marker; on load the age fold has to recover the + // elapsed time from each observation's insert tick up to that marker as + // well as from the marker to now, so an observation idle across the flush + // keeps its true age rather than freezing at the flush tick. + let path = store_path("clock_age"); { - let mut s = LogStore::new(&path, 0, 1).unwrap(); - for i in 0..3u64 { - let o = obs([1; 16], 1 + i, 0, Some(10)); - assert!(s.insert(o, 1 + i).unwrap()); - } - assert_eq!(s.next_seq(), 4); + let mut s = LogStore::open(&path, 1000).unwrap(); + let o = obs([1; 16], 1, 500, Some(10)); + assert!(s.insert(o, 1000).unwrap()); + // A second insert advances the store clock to 3000; the flush below + // records a Clock marker at that tick even though `o` was inserted + // earlier (at 1000) and sits idle until the store is dropped. + let d = obs([2; 16], 2, 0, Some(20)); + assert!(s.insert(d, 3000).unwrap()); s.flush().unwrap(); } { - let s = LogStore::new(&path, 100, 1).unwrap(); - assert_eq!(s.next_seq(), 4); + // Reopen at 9000. `o`'s true age is 500 (insert-time) + (9000 - 1000) + // = 8500; the marker at 3000 must fold both the pre-flush 2000ms and + // the post-flush 6000ms. + let s = LogStore::open(&path, 9000).unwrap(); + let o = obs([1; 16], 1, 500, Some(10)); + let g = s.get(&o.id(), 9000).unwrap(); + assert_eq!(g.age_ms, 8500); + } +} + +#[test] +fn boot_counter_first_seq_increases_and_no_collision() { + // A boot counter makes `first_seq` strictly increase across reboots, so a + // resumed node's sequence numbers land in a disjoint range and cannot + // collide (content-id dedupe) with the previous boot's observations. + let path = store_path("boot"); + + let first1; + let first2; + + // Boot 1: first_seq is 1 << 32. + { + let mut s = LogStore::open(&path, 0).expect("open boot 1"); + first1 = s.first_seq(); + assert_eq!(first1, 1u64 << 32); + // A node seeded here observes first1, first1 + 1, ... + let a = obs([1; 16], first1, 0, Some(10)); + let a2 = obs([1; 16], first1 + 1, 0, Some(10)); + assert!(s.insert(a, 10).unwrap()); + assert!(s.insert(a2, 11).unwrap()); + drop(s); + } + + // Boot 2: first_seq strictly increased. A fresh observation at the new + // first_seq is a *new* id (Ok(true)), not deduped against boot 1's ids + // that would happen if the seq range were reused. + { + let mut s = LogStore::open(&path, 100).expect("open boot 2"); + first2 = s.first_seq(); + assert!(first2 > first1); + let c = obs([1; 16], first2, 0, Some(20)); + assert!( + s.insert(c, 101).unwrap(), + "first_seq must advance so boot-2 ids don't collide with boot-1 ids" + ); + // Boot 1's observations are still present (content-addressed). + assert!(s.contains(&obs([1; 16], first1, 0, Some(10)).id())); + drop(s); + } + + // Boot 3: still strictly increasing. + { + let s = LogStore::open(&path, 200).expect("open boot 3"); + assert!(s.first_seq() > first2); + } +} + +#[test] +fn release_survives_kill_without_flush() { + // release appends a durable Release record immediately, so a release that + // is never followed by a flush (a kill without graceful shutdown) still + // survives a reopen. + let path = store_path("release"); + + { + let mut s = LogStore::open(&path, 0).expect("open release store"); + let a = obs([1; 16], 1, 0, Some(10)); + let b = obs([2; 16], 2, 0, Some(20)); + assert!(s.insert(a, 100).unwrap()); + assert!(s.insert(b.clone(), 101).unwrap()); + let ib = b.id(); + s.release(&ib); + assert!(s.is_released(&ib)); + // Drop WITHOUT flush: the release record must already be on disk + // (appended immediately), not waiting for a flush to persist it. + drop(s); + } + + { + let s = LogStore::open(&path, 200).expect("reopen after kill w/o flush"); + let b = obs([2; 16], 2, 0, Some(20)); + assert!(s.contains(&b.id())); + assert!( + s.is_released(&b.id()), + "a release must survive a kill without a flush" + ); } } @@ -155,7 +243,7 @@ fn reopen_recovers_from_torn_trailing_record() { let path = store_path("torn"); { - let mut s = LogStore::new(&path, 0, 1).expect("open fresh store"); + let mut s = LogStore::open(&path, 0).expect("open fresh store"); let a = obs([1; 16], 1, 0, Some(10)); let b = obs([2; 16], 2, 0, Some(20)); assert!(s.insert(a, 1000).unwrap()); @@ -177,11 +265,12 @@ fn reopen_recovers_from_torn_trailing_record() { } // Reopen: the torn tail is truncated, the two complete records survive. + // The boot counter has advanced one boot (this is the store's second open). { - let s = LogStore::new(&path, 3000, 1).expect("reopen after torn tail"); + let s = LogStore::open(&path, 3000).expect("reopen after torn tail"); assert!(s.contains(&obs([1; 16], 1, 0, Some(10)).id())); assert!(s.contains(&obs([2; 16], 2, 0, Some(20)).id())); - assert_eq!(s.next_seq(), 3); + assert_eq!(s.first_seq(), 2u64 << 32); } } @@ -193,14 +282,14 @@ fn reopen_recovers_from_oversized_record_length() { let path = store_path("oversize"); { - let mut s = LogStore::new(&path, 0, 1).expect("open fresh store"); + let mut s = LogStore::open(&path, 0).expect("open fresh store"); let a = obs([1; 16], 1, 0, Some(10)); assert!(s.insert(a, 1000).unwrap()); s.flush().expect("flush before drop"); } { - // Declare a body of u32::MAX bytes with no actual body. read_record + // Declare a body of u32::MAX bytes with no actual body. parse_record // must reject it before allocating any buffer. let mut f = std::fs::OpenOptions::new() .append(true) @@ -212,7 +301,7 @@ fn reopen_recovers_from_oversized_record_length() { } { - let s = LogStore::new(&path, 3000, 1).expect("reopen after oversized prefix"); + let s = LogStore::open(&path, 3000).expect("reopen after oversized prefix"); assert!(s.contains(&obs([1; 16], 1, 0, Some(10)).id())); } } diff --git a/tines/src/main.rs b/tines/src/main.rs index 0acb4323..1f6c3f83 100644 --- a/tines/src/main.rs +++ b/tines/src/main.rs @@ -5,8 +5,8 @@ //! One instance per device — the typical deployment on robots without ROS2. //! //! The node/peer ids are 32 lowercase hex chars (the 16-byte `NodeId`), and -//! `first_seq` is the next tines sequence number to use (persisted across -//! reboots, starting from `tines`'s 1-based numbering). +//! The tines sequence numbers are seeded from the LogStore's boot counter, so +//! they persist across reboots without a separate config field. use std::error::Error; use std::path::Path; @@ -23,8 +23,6 @@ struct Config { node_id: String, /// The peer's advertised `NodeId`, 32 lowercase hex chars. peer: String, - /// Next tines sequence number to use (persisted across reboots). - first_seq: u64, /// Absolute path to the LogStore append-only log. log_path: String, /// Unix-domain socket path the IPC listens on. @@ -42,11 +40,11 @@ fn main() -> Result<(), Box> { // LogStore is the Store (S); BtTransport wraps it as the Transport (T). let clock = HostClock::start(); - let store = LogStore::new(Path::new(&config.log_path), clock.now(), config.first_seq) + let store = LogStore::open(Path::new(&config.log_path), clock.now()) .map_err(|e| format!("failed to open log store: {e:?}"))?; let radio = BluerRadio::new(me, peer); let transport = BtTransport::new(me, peer, radio); - let node = Node::new(me, config.first_seq, store, transport); + let node = Node::new(me, store.first_seq(), store, transport); println!( "tines: listening on {} (node {})", diff --git a/tines/tests/ipc.rs b/tines/tests/ipc.rs index 1cc6f813..907e09cc 100644 --- a/tines/tests/ipc.rs +++ b/tines/tests/ipc.rs @@ -127,8 +127,8 @@ fn unique_dir(tag: &str) -> PathBuf { fn build_node(peer: NodeId, dir: &Path) -> Node { let store_path = dir.join("store.log"); - let store = LogStore::new(&store_path, 0, 1).expect("store open"); - Node::new(peer, 1, store, MockTransport::new(peer)) + let store = LogStore::open(&store_path, 0).expect("store open"); + Node::new(peer, store.first_seq(), store, MockTransport::new(peer)) } fn connect(path: &Path) -> UnixStream { @@ -174,7 +174,9 @@ fn handle_status_and_belief_direct() { stored, } => { assert_eq!(id, hex_encode(&node.id())); - assert_eq!(next_seq, 2); + // Seeded from the boot counter (first_seq = boot_count << 32 = 1<<32 + // on a fresh store); after one observe next_seq is that + 1. + assert_eq!(next_seq, (1u64 << 32) + 1); assert_eq!(stored, 1); } other => panic!("expected status, got {other:?}"),