From 5a8be19b4cdd7d5158abaa7c1770d06017107a33 Mon Sep 17 00:00:00 2001 From: Valentin Haudiquet Date: Thu, 5 Nov 2026 19:06:41 +0100 Subject: [PATCH] fix(calendar): fed slots stay fed across a restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reminder source skipped an already-fed slot only while the slot sat in the dispatcher's pending inbox; a claw restart drained that inbox, and every future lead of the same occurrence re-fed — a reminder fired two or three times per calendar event. The source keeps its own fed-slot ledger in the calendar store now (keyed by calendar, occurrence, and lead), records each slot right after a successful ingest, and prunes rows whose occurrence is an hour past every lead, so a pruned row can never resurrect a reminder. The store-backed source's ledger survives a restart; the static-list constructor keeps an in-memory one for the process lifetime, said so in the docs that previously claimed dedup was someone else's problem. Store failures surface loudly (a silently lost row is a re-fired reminder); the prune warns and continues. Offline tests cover the once-only rule, the restart via a tempdir reopen, and the prune. --- .../rhema-calendar-integration/Cargo.toml | 3 + .../rhema-calendar-integration/src/error.rs | 5 + .../rhema-calendar-integration/src/lib.rs | 5 +- .../src/reminder.rs | 172 +++++++++++-- .../rhema-calendar-integration/src/store.rs | 226 ++++++++++++++++-- 5 files changed, 371 insertions(+), 40 deletions(-) diff --git a/integrations/rhema-calendar-integration/Cargo.toml b/integrations/rhema-calendar-integration/Cargo.toml index 65126e4..a5dd0e8 100644 --- a/integrations/rhema-calendar-integration/Cargo.toml +++ b/integrations/rhema-calendar-integration/Cargo.toml @@ -42,6 +42,9 @@ time = { version = "0.3", default-features = false, features = [ tokio = { workspace = true, features = ["rt-multi-thread", "macros", "sync", "time"] } # In-process HTTP mocks for offline tests wiremock = "0.6" +# File-backed stores in offline tests (the fed-slot ledger's restart +# behavior) +tempfile = "3" [lints] workspace = true diff --git a/integrations/rhema-calendar-integration/src/error.rs b/integrations/rhema-calendar-integration/src/error.rs index 8ab4e47..d31ef48 100644 --- a/integrations/rhema-calendar-integration/src/error.rs +++ b/integrations/rhema-calendar-integration/src/error.rs @@ -38,6 +38,11 @@ pub enum CalendarError { /// than poll an under-specified feed. #[error("calendar config: {0}")] Config(String), + /// The calendar store failed to open, migrate, or answer: a lost + /// fed-slot ledger row is a re-fired reminder, so this surfaces + /// instead of shrinking the ledger silently. + #[error("calendar store: {0}")] + Store(String), } /// The longest body snippet an [`CalendarError::Api`] keeps: enough for diff --git a/integrations/rhema-calendar-integration/src/lib.rs b/integrations/rhema-calendar-integration/src/lib.rs index 52e81a2..ee6866a 100644 --- a/integrations/rhema-calendar-integration/src/lib.rs +++ b/integrations/rhema-calendar-integration/src/lib.rs @@ -14,8 +14,9 @@ //! its own; every [`crate::CalendarEvent`] carries an id that stays //! the same across polls of the same occurrence, so any //! implementation re-polling on its own cadence never re-fires a -//! reminder it already fed. The repetition is the consumer's -//! dispatcher's dedup, not this crate's problem. +//! reminder it already fed. The default source keeps that promise in +//! its own fed-slot ledger; a custom source leans on the stable ids +//! and its dispatcher's dedup. //! //! Recurrence is expanded before a poll answers, so a consumer never //! sees a rule, only starts it can compare against its clock: diff --git a/integrations/rhema-calendar-integration/src/reminder.rs b/integrations/rhema-calendar-integration/src/reminder.rs index 5144302..23f5381 100644 --- a/integrations/rhema-calendar-integration/src/reminder.rs +++ b/integrations/rhema-calendar-integration/src/reminder.rs @@ -6,10 +6,15 @@ //! calendar for the occurrences starting within its horizon and //! ingests one event per due reminder slot, a (occurrence, lead time) //! pair whose moment has come. The dedup id carries all three facts -//! (`{calendar}:{occurrence id}:{lead}`), so a slot feeds once and its -//! repeats dedup away in the dispatcher, and the next lead's slot is a -//! different id firing later. A slot whose start has passed is never -//! fed: a reminder never fires after the fact. All-day occurrences +//! (`{calendar}:{occurrence id}:{lead}`), and the next lead's slot is +//! a different id firing later. A slot feeds once: the dispatcher +//! dedups a fed id only while the slot sits in its pending inbox, so +//! the source keeps its own ledger of fed slots in the calendar store +//! and skips anything already there. The store-backed source's ledger +//! survives a restart; a static-list source keeps its ledger in +//! memory, so its slots dedup for the process's lifetime. A slot +//! whose start has passed is never fed: a reminder never fires after +//! the fact. All-day occurrences //! never fire: a lead in minutes against a midnight start is noise, //! and a date-shaped nudge is what the task tools are for. //! @@ -35,6 +40,7 @@ use serde_json::{Value, json}; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; +use crate::store::FedSlot; use crate::{ CaldavCalendar, Calendar, CalendarError, CalendarEvent, GoogleCalendar, IcsCalendar, Window, }; @@ -245,6 +251,14 @@ pub struct CalendarSource { /// callers that hold the specs) or the calendar table, read fresh /// on every detect pass so a runtime add is live next tick. feeds: FeedSource, + /// The fed-slot ledger, the durable half of the reminder dedup: + /// the dispatcher forgets a fed id once its turn drains the + /// inbox, and without this the next tick would re-ingest the + /// identical id and re-dispatch the same reminder for the whole + /// lead window. The store-backed source shares the calendar + /// table's database; a static-list source keeps its ledger in + /// memory. + ledger: Arc, } enum FeedSource { @@ -271,20 +285,30 @@ impl CalendarSource { horizon_hours: calendar.horizon(), }); } + // A static-list source carries no database file, so its + // ledger lives in memory: slots dedup for the process's + // lifetime, and a restart re-feeds them. + let ledger = Arc::new( + crate::store::CalendarStore::in_memory() + .map_err(|err| CalendarError::Store(format!("fed-slot ledger: {err}")))?, + ); dispatcher.register_source(CALENDARS_SOURCE, reminder_key_fn()); Ok(Self { dispatcher: dispatcher.clone(), clock, feeds: FeedSource::Static(feeds), + ledger, }) } /// Builds from the calendar table instead of a static list: the /// source reads it fresh every pass, so a calendar added over the /// HTTP route (or a chat tool) is live without a restart, and a - /// removed one stops polling on the next tick. Config validation - /// ran when the row landed; a calendar that still fails to resolve - /// here names itself and costs its tick, not the daemon. + /// removed one stops polling on the next tick. The same store + /// backs the fed-slot ledger, so dedup outlives a restart. Config + /// validation ran when the row landed; a calendar that still fails + /// to resolve here names itself and costs its tick, not the + /// daemon. pub fn from_store( dispatcher: &Arc, clock: Arc, @@ -294,7 +318,8 @@ impl CalendarSource { Self { dispatcher: dispatcher.clone(), clock, - feeds: FeedSource::Store(store), + feeds: FeedSource::Store(store.clone()), + ledger: store, } } @@ -306,19 +331,23 @@ impl CalendarSource { } } - /// A source over ready-made feeds. Tests use this; production - /// builds from the config with [`CalendarSource::new`]. + /// A source over ready-made feeds, with the ledger passed in (a + /// fresh in-memory store per test, or a reopened one for the + /// restart tests). Tests use this; production builds from the + /// config with [`CalendarSource::new`] or [`CalendarSource::from_store`]. #[cfg(test)] pub(crate) fn from_feeds( dispatcher: &Arc, clock: Arc, feeds: Vec, + ledger: Arc, ) -> Self { dispatcher.register_source(CALENDARS_SOURCE, reminder_key_fn()); Self { dispatcher: dispatcher.clone(), clock, feeds: FeedSource::Static(feeds), + ledger, } } } @@ -359,6 +388,12 @@ impl Source for CalendarSource { let now_ms = self.clock.now_millis() as i64; let now = OffsetDateTime::from_unix_timestamp_nanos(now_ms as i128 * 1_000_000) .map_err(|err| SourceError::Backend(format!("clock: {err}")))?; + // Housekeeping rides the pass it already needs: the prune + // never removes a slot that could still fire, and failing it + // costs table space, not correctness. + if let Err(err) = self.ledger.prune_fed_slots(now.unix_timestamp()) { + eprintln!("[calendar] warning: fed-slot prune failed: {err}"); + } let mut fed = 0; let feeds = self.feeds_for_pass().map_err(SourceError::Backend)?; for feed in &feeds { @@ -377,6 +412,19 @@ impl Source for CalendarSource { } }; for (event, lead) in due_slots(&events, &feed.leads, now) { + // A slot the ledger already fed is skipped, not + // re-ingested: the dispatcher's dedup only holds while + // the slot sits pending, so the ledger is what keeps + // the tick from re-firing it. A ledger read or write + // that fails fails loudly: a silently lost row is a + // re-fired reminder. + let already_fed = self + .ledger + .slot_is_fed(&feed.name, &event.id, lead) + .map_err(|err| SourceError::Backend(format!("fed-slot ledger: {err}")))?; + if already_fed { + continue; + } self.dispatcher.ingest(reminder_event( &feed.name, event, @@ -384,6 +432,15 @@ impl Source for CalendarSource { now, now_ms as u64, ))?; + let slot = FedSlot { + calendar: feed.name.clone(), + occurrence: event.id.clone(), + lead, + starts_at: event.start.unix_timestamp(), + }; + self.ledger + .mark_slot_fed(&slot, now.unix_timestamp()) + .map_err(|err| SourceError::Backend(format!("fed-slot ledger: {err}")))?; fed += 1; } } @@ -395,6 +452,7 @@ impl Source for CalendarSource { mod tests { use super::*; use crate::CalendarError; + use crate::store::CalendarStore; use rhema_rsd::{DispatcherConfig, FixedClock, ResourceKey, SystemClock, VolatileStore}; /// 2026-10-20T00:00:00Z, in milliseconds. @@ -439,6 +497,11 @@ mod tests { } } + /// A fresh in-memory fed-slot ledger for one test's source. + fn ledger() -> Arc { + Arc::new(CalendarStore::in_memory().expect("ledger opens")) + } + fn source( dispatcher: &Arc, clock: Arc, @@ -453,6 +516,7 @@ mod tests { leads: vec![60, 15], horizon_hours: 24, }], + ledger(), ) } @@ -482,8 +546,10 @@ mod tests { assert_eq!(source.detect().await.expect("detect runs"), 1); assert_eq!(inbox_depth(&dispatcher), 1); - // The next pass replays the same dedup id: fed, deduped away. - assert_eq!(source.detect().await.expect("detect runs"), 1); + // The next pass finds the slot in the ledger and skips it: + // nothing re-fed, however many ticks remain in the lead + // window, and nothing re-dispatched once the inbox drains. + assert_eq!(source.detect().await.expect("detect runs"), 0); assert_eq!(inbox_depth(&dispatcher), 1); } @@ -499,11 +565,11 @@ mod tests { clock.advance(30 * 60 * 1000); assert_eq!(source.detect().await.expect("detect runs"), 1); - // At 15 minutes out the second slot fires under its own id. - // The count includes the first slot's replay, which dedups: - // the inbox holds one event per lead, never more. + // At 15 minutes out the second slot fires under its own id; + // the first slot's ledger row keeps its replay out of the + // pass. One event per lead, never more. clock.advance(45 * 60 * 1000); - assert_eq!(source.detect().await.expect("detect runs"), 2); + assert_eq!(source.detect().await.expect("detect runs"), 1); assert_eq!(inbox_depth(&dispatcher), 2); // Past the start: nothing more, ever. @@ -518,13 +584,82 @@ mod tests { let dispatcher = dispatcher(); // The daemon only meets this event 15 minutes before it // starts: the 60 slot's moment came and went unseen, the 15 - // slot is due. Both are fed on that first pass, then dedup. + // slot is due. Both are fed on that first pass; the second + // finds both rows in the ledger and feeds nothing. let source = source(&dispatcher, clock.clone(), vec![event("late", "late", 15)]); assert_eq!(source.detect().await.expect("detect runs"), 2); - assert_eq!(source.detect().await.expect("detect runs"), 2); + assert_eq!(source.detect().await.expect("detect runs"), 0); assert_eq!(inbox_depth(&dispatcher), 2); } + #[tokio::test] + async fn a_reopened_store_still_suppresses_fed_slots() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("calendars.db"); + let clock = Arc::new(FixedClock::new(NOW_MS as u64)); + let feeds = || { + vec![Feed { + name: "home".to_owned(), + client: Arc::new(Fixed(vec![event("standup", "standup", 90)])), + leads: vec![60], + horizon_hours: 24, + }] + }; + + // Sixty minutes out the slot is due and feeds, its row + // landing in the file-backed ledger. + clock.advance(30 * 60 * 1000); + { + let store = Arc::new(CalendarStore::open(&path).expect("store opens")); + let dispatcher = dispatcher(); + let source = CalendarSource::from_feeds(&dispatcher, clock.clone(), feeds(), store); + assert_eq!(source.detect().await.expect("detect runs"), 1); + } + + // A restart: the source, its dispatcher, and its pending + // inbox are all gone, and the slot is still due. The ledger + // is what survives, and it is what keeps the reminder from + // firing a second time. + clock.advance(30 * 60 * 1000); + let store = Arc::new(CalendarStore::open(&path).expect("store reopens")); + let dispatcher = dispatcher(); + let source = CalendarSource::from_feeds(&dispatcher, clock.clone(), feeds(), store); + assert_eq!(source.detect().await.expect("detect runs"), 0); + assert_eq!(inbox_depth(&dispatcher), 0); + } + + #[tokio::test] + async fn a_fed_slot_is_not_refed_when_the_inbox_has_drained() { + // The dispatcher dedups by (source, id) against pending items + // only, so a fresh dispatcher is a drained inbox: whatever it + // holds, the old behavior re-fed. The ledger, not the + // dispatcher, is the reason this stays quiet. + let clock = Arc::new(FixedClock::new(NOW_MS as u64)); + let ledger = ledger(); + let feeds = vec![Feed { + name: "home".to_owned(), + client: Arc::new(Fixed(vec![event("standup", "standup", 90)])), + leads: vec![60, 15], + horizon_hours: 24, + }]; + + // Sixty minutes out, the 60 slot feeds once. + clock.advance(30 * 60 * 1000); + let first = dispatcher(); + let source = + CalendarSource::from_feeds(&first, clock.clone(), feeds.clone(), ledger.clone()); + assert_eq!(source.detect().await.expect("detect runs"), 1); + + // A second source over the same store and the same feed, with + // an empty dispatcher: the 60 slot is in the ledger, the 15 + // slot is not yet due, so nothing feeds. + clock.advance(30 * 60 * 1000); + let second = dispatcher(); + let source = CalendarSource::from_feeds(&second, clock.clone(), feeds, ledger); + assert_eq!(source.detect().await.expect("detect runs"), 0); + assert_eq!(inbox_depth(&second), 0); + } + #[tokio::test] async fn a_started_or_all_day_event_never_fires() { let clock = Arc::new(FixedClock::new(NOW_MS as u64)); @@ -567,6 +702,7 @@ mod tests { horizon_hours: 24, }, ], + ledger(), ); assert_eq!(source.detect().await.expect("detect runs"), 1); assert_eq!( diff --git a/integrations/rhema-calendar-integration/src/store.rs b/integrations/rhema-calendar-integration/src/store.rs index e1a98a7..f2aacba 100644 --- a/integrations/rhema-calendar-integration/src/store.rs +++ b/integrations/rhema-calendar-integration/src/store.rs @@ -1,15 +1,50 @@ -//! The calendar table: agent-managed calendars in SQLite, the same +//! The calendar store: agent-managed calendars in SQLite, the same //! database file the dispatcher's store uses — the pattern the claw's //! watch table established. The config stopped carrying calendars: //! the table is the one truth, and the source reads it fresh on every //! detect pass, so a row added over `/v1/calendars` (or a chat tool) -//! is live without a restart. +//! is live without a restart. The same store owns the reminder +//! source's fed-slot ledger: the durable half of the reminder dedup, +//! so a slot does not re-fire once its turn has drained the inbox. use crate::reminder::CalendarWatch; -use rusqlite::Connection; +use rusqlite::{Connection, OptionalExtension}; use std::path::Path; use std::sync::Mutex; +/// How long past its last possible due moment a fed slot's row stays +/// in the ledger. A slot only fires while its occurrence is still +/// future, so the prune rule reads `start + lead + this margin`, and +/// the margin covers a clock stepped backwards or a stalled pass +/// landing after the prune. The rows are tiny; the margin errs long. +const FED_SLOT_GRACE_SECS: i64 = 60 * 60; + +/// Seconds in a minute, the unit the ledger's `lead` column carries. +const SECS_PER_MINUTE: i64 = 60; + +/// The tables this store owns, created idempotently at open. The +/// dispatcher's store shares this database file, so these land beside +/// its tables, never over them. +const SCHEMA: &str = " +CREATE TABLE IF NOT EXISTS calendars ( + name TEXT PRIMARY KEY, + spec TEXT NOT NULL, + created_at INTEGER NOT NULL +); +CREATE TABLE IF NOT EXISTS fed_slots ( + calendar TEXT NOT NULL, + occurrence TEXT NOT NULL, + lead INTEGER NOT NULL, + starts_at INTEGER NOT NULL, + fed_at INTEGER NOT NULL, + PRIMARY KEY (calendar, occurrence, lead) +);"; + +/// Runs the store's idempotent migration over one connection. +fn migrate(conn: &Connection) -> Result<(), std::io::Error> { + conn.execute_batch(SCHEMA).map_err(std::io::Error::other) +} + /// An open handle on the calendar table. Its own rusqlite connection /// to the shared database file — WAL and a busy timeout, so the /// dispatcher's connection and this one interleave through SQLite's @@ -28,30 +63,17 @@ impl CalendarStore { let conn = Connection::open(path).map_err(std::io::Error::other)?; conn.busy_timeout(std::time::Duration::from_millis(5_000)) .map_err(std::io::Error::other)?; - conn.execute_batch( - "CREATE TABLE IF NOT EXISTS calendars ( - name TEXT PRIMARY KEY, - spec TEXT NOT NULL, - created_at INTEGER NOT NULL - );", - ) - .map_err(std::io::Error::other)?; + migrate(&conn)?; Ok(Self { conn: Mutex::new(conn), }) } - /// An in-memory table; tests use this. + /// An in-memory store; tests and the static-list source's ledger + /// use this. pub fn in_memory() -> Result { let conn = Connection::open_in_memory().map_err(std::io::Error::other)?; - conn.execute_batch( - "CREATE TABLE IF NOT EXISTS calendars ( - name TEXT PRIMARY KEY, - spec TEXT NOT NULL, - created_at INTEGER NOT NULL - );", - ) - .map_err(std::io::Error::other)?; + migrate(&conn)?; Ok(Self { conn: Mutex::new(conn), }) @@ -121,6 +143,92 @@ impl CalendarStore { .map(|spec| serde_json::from_str(spec).map_err(std::io::Error::other)) .collect() } + + /// Whether this slot has already fed. The dispatcher dedups a fed + /// id only while the slot sits in its inbox; this ledger is what + /// keeps the next tick from re-ingesting a slot whose turn has + /// drained. + pub(crate) fn slot_is_fed( + &self, + calendar: &str, + occurrence: &str, + lead: u64, + ) -> Result { + // rusqlite binds no unsigned ints; the column is INTEGER. + let lead = i64::try_from(lead).map_err(std::io::Error::other)?; + let conn = self + .conn + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let fed: Option<()> = conn + .query_row( + "SELECT 1 FROM fed_slots + WHERE calendar = ?1 AND occurrence = ?2 AND lead = ?3", + rusqlite::params![calendar, occurrence, lead], + |_| Ok(()), + ) + .optional() + .map_err(std::io::Error::other)?; + Ok(fed.is_some()) + } + + /// Records a slot as fed; `false` when the ledger already held it. + /// The dedup id stays `{calendar}:{occurrence}:{lead}` downstream; + /// this table is the source's own memory of what it has fed. + pub(crate) fn mark_slot_fed( + &self, + slot: &FedSlot, + fed_at: i64, + ) -> Result { + // rusqlite binds no unsigned ints; the column is INTEGER. + let lead = i64::try_from(slot.lead).map_err(std::io::Error::other)?; + let conn = self + .conn + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let inserted = conn + .execute( + "INSERT INTO fed_slots (calendar, occurrence, lead, starts_at, fed_at) + VALUES (?1, ?2, ?3, ?4, ?5) + ON CONFLICT(calendar, occurrence, lead) DO NOTHING", + rusqlite::params![slot.calendar, slot.occurrence, lead, slot.starts_at, fed_at], + ) + .map_err(std::io::Error::other)?; + Ok(inserted == 1) + } + + /// Deletes the ledger's spent rows: slots whose occurrence start, + /// plus the lead, plus [`FED_SLOT_GRACE_SECS`] is already past + /// `now`. Such a slot can never be due again — a started event is + /// never fed — so pruning keeps the table bounded without ever + /// resurrecting a reminder. Returns how many rows went. + pub(crate) fn prune_fed_slots(&self, now: i64) -> Result { + let conn = self + .conn + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + conn.execute( + "DELETE FROM fed_slots WHERE starts_at + lead * ?1 + ?2 < ?3", + rusqlite::params![SECS_PER_MINUTE, FED_SLOT_GRACE_SECS, now], + ) + .map_err(std::io::Error::other) + } +} + +/// One reminder slot the source has fed, as the ledger holds it: the +/// dedup key (calendar, occurrence, lead) plus the times the prune +/// rule needs. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct FedSlot { + /// The calendar's name, as the dedup id spells it. + pub(crate) calendar: String, + /// The occurrence's stable vocabulary id. + pub(crate) occurrence: String, + /// The slot's lead, in minutes. + pub(crate) lead: u64, + /// When the occurrence starts, epoch seconds: what the prune rule + /// measures the slot's lifetime against. + pub(crate) starts_at: i64, } #[cfg(test)] @@ -128,6 +236,9 @@ mod tests { use super::*; use crate::CalendarKind; + /// 2026-10-20T00:00:00Z, in seconds. + const NOW_SECS: i64 = 1_792_444_800; + fn ics(name: &str) -> CalendarWatch { serde_json::from_str(&format!( "{{\"name\": \"{name}\", \"kind\": \"ics\", \"url\": \"https://example.net/f.ics\"}}" @@ -163,4 +274,79 @@ mod tests { let err = store.add(&no_url).expect_err("an urlless ics is refused"); assert!(err.to_string().contains("url"), "got: {err}"); } + + #[test] + fn a_fed_slot_is_recorded_once_per_key() { + let store = CalendarStore::in_memory().expect("store opens"); + let slot = FedSlot { + calendar: "home".to_owned(), + occurrence: "ics:standup@1792445400".to_owned(), + lead: 60, + starts_at: NOW_SECS + 15 * SECS_PER_MINUTE, + }; + assert!( + store.mark_slot_fed(&slot, NOW_SECS).expect("mark lands"), + "the first insert is new" + ); + assert!( + !store.mark_slot_fed(&slot, NOW_SECS).expect("mark again"), + "the key dedups: the row is never rewritten" + ); + assert!( + store + .slot_is_fed("home", "ics:standup@1792445400", 60) + .expect("read lands"), + ); + assert!( + !store + .slot_is_fed("home", "ics:standup@1792445400", 15) + .expect("read lands"), + "another lead is its own slot" + ); + assert!( + !store + .slot_is_fed("work", "ics:standup@1792445400", 60) + .expect("read lands"), + "another calendar is its own slot" + ); + } + + #[test] + fn the_ledger_prunes_slots_whose_occurrence_is_long_past() { + let store = CalendarStore::in_memory().expect("store opens"); + // Fed two days ago, for an occurrence that started shortly + // after: every lead's moment is long gone. + let spent = FedSlot { + calendar: "home".to_owned(), + occurrence: "ics:old@1792272600".to_owned(), + lead: 15, + starts_at: NOW_SECS - 47 * 3600, + }; + store + .mark_slot_fed(&spent, NOW_SECS - 47 * 3600) + .expect("mark lands"); + // Still ahead of itself: the slot's last due moment is + // `starts_at + lead + grace`, and that is in the future. + let live = FedSlot { + calendar: "home".to_owned(), + occurrence: "ics:soon@1792483200".to_owned(), + lead: 15, + starts_at: NOW_SECS + 3600, + }; + store.mark_slot_fed(&live, NOW_SECS).expect("mark lands"); + + let pruned = store.prune_fed_slots(NOW_SECS).expect("prune lands"); + assert_eq!(pruned, 1, "the spent row goes, and only it"); + assert!( + !store + .slot_is_fed("home", "ics:old@1792272600", 15) + .expect("read lands"), + ); + assert!( + store + .slot_is_fed("home", "ics:soon@1792483200", 15) + .expect("read lands"), + "a slot that can still be due survives the prune" + ); + } }