From 62271e51823e9be89fac858870f3db72c59b75ad Mon Sep 17 00:00:00 2001 From: Nils Lehnen <30603423+iderex@users.noreply.github.com> Date: Fri, 4 Sep 2026 16:22:06 +0200 Subject: [PATCH] Hold 0047's two moments per entry, and answer the age they anchor (#47) 0047 says an entry stores two moments when it is enqueued - the server's own last stated time and the device's wall reading at that instant - and that its age is computed the way 0043 computes a cache entry's, on 0102's anchor, with the same correction and the same two guards. The queue that landed carried neither, and its own module documentation said so. What that costs is the failure 0047 names for a restored queue: one that treats every entry as freshly enqueued keeps its order and loses every age, so a client can say only that something is pending rather than that an action has been waiting three weeks. An entry now carries `WrittenAt` and answers `Entry::age_at`. Both are `crate::cache::freshness`'s own types rather than a second arithmetic here, which is what naming 0043 asks for and what stops the two drifting. Every moment is handed in, so nothing in the module reads a clock. A replacement keeps the earlier entry's two moments, as it keeps its position in the order, which 0047 does not say. Taking the later action's moments reports every actively touched entry as freshly enqueued, so a queue undelivered for a month says seconds for the items somebody kept scrubbing - the restored-queue failure arriving through the coalescing door, landing on the person coalescing at enqueue exists to protect. The age acts on nothing, which is 0047's rule. That absence used to be held by there being no age at all; it is held by a case now, because a guarantee that rests on a thing not existing ends the moment it does. `Reporting::report`, `Reporting::observe`, `Reporting::report_after_a_renewal` and `mid_playback::the_renewal_ended` take the pair and pass it through, for the reason they already take an `ElapsedInstant`: nothing under `src/` reads a platform clock. Neither of #47's two conditions is met by this. Both restart the core and restore a server, and nothing in this tree does either. Signed-off-by: Nils Lehnen <30603423+iderex@users.noreply.github.com> --- src/playback/report.rs | 79 ++++++++-- src/server/write_queue.rs | 290 +++++++++++++++++++++++++++++++++++- src/session/mid_playback.rs | 29 +++- 3 files changed, 377 insertions(+), 21 deletions(-) diff --git a/src/playback/report.rs b/src/playback/report.rs index 31c3b18..7c68678 100644 --- a/src/playback/report.rs +++ b/src/playback/report.rs @@ -67,6 +67,7 @@ use super::AdmittedPosition; use super::Ticks; use super::cadence::{ReportsWithoutWaiting, TheInterval}; +use crate::cache::freshness::WrittenAt; use crate::clock::ElapsedInstant; use crate::server::write_queue::{Target, WhatIsAsserted, WhatTheEnqueueDid, WriteQueue}; @@ -209,11 +210,17 @@ impl Reporting { /// /// What comes back is what the queue did, and it is the only thing that /// comes back: a report that was not asked of the queue does not exist. + /// + /// `enqueued_at` is the pair 0047 stores with every entry, and it is handed + /// in rather than read here for the reason `at` is: nothing under `src/` + /// reads a platform clock, which is 0102's rule and the `no-platform-clock` + /// rule in `.github/invariants/rules`. pub fn report( &mut self, event: ReportsWithoutWaiting, position: AdmittedPosition, at: ElapsedInstant, + enqueued_at: WrittenAt, queue: &mut WriteQueue, ) -> WhatTheEnqueueDid { let what_the_queue_did = queue.enqueue( @@ -223,6 +230,7 @@ impl Reporting { position: position.position(), reported_on: ReportedOn::Event(event), }, + enqueued_at, ); self.interval = self.interval.after(event, at); what_the_queue_did @@ -240,10 +248,16 @@ impl Reporting { /// A report made here moves the interval on from `now`, so the next one is /// due ten seconds after this one rather than ten seconds after the last /// event. + /// + /// `enqueued_at` is asked for on every call and used only where a report is + /// due, which is most calls making none. It is the caller's reading either + /// way, and a signature that took it only when it was needed would be a + /// second entry point to the same cadence. pub fn observe( &mut self, position: AdmittedPosition, now: ElapsedInstant, + enqueued_at: WrittenAt, queue: &mut WriteQueue, ) -> WhatObservingDid { if !self.interval.is_running() { @@ -259,6 +273,7 @@ impl Reporting { position: position.position(), reported_on: ReportedOn::TheInterval, }, + enqueued_at, ); self.interval = self.interval.after_a_report_at(now); WhatObservingDid::Reported(what_the_queue_did) @@ -279,9 +294,16 @@ impl Reporting { /// the cadence changed, and a report that moved it would make the next /// interval report due ten seconds after the renewal rather than ten /// seconds after the last report the cadence made. + /// + /// `enqueued_at` reaches the queue and, where the rejection left an entry + /// for this item at the head, is not what the entry keeps: 0047's + /// coalescing keeps the earlier entry's moments along with its position, so + /// the age says how long the item's position has been undelivered rather + /// than how long ago the renewal ended. pub fn report_after_a_renewal( &mut self, position: AdmittedPosition, + enqueued_at: WrittenAt, queue: &mut WriteQueue, ) -> WhatTheEnqueueDid { queue.enqueue( @@ -291,6 +313,7 @@ impl Reporting { position: position.position(), reported_on: ReportedOn::AfterARenewal, }, + enqueued_at, ) } } @@ -305,7 +328,8 @@ mod tests { //! the queue that holds every report is not durable. use super::{PositionReport, ReportedOn, Reporting, WhatObservingDid}; - use crate::clock::ElapsedInstant; + use crate::cache::freshness::WrittenAt; + use crate::clock::{ElapsedInstant, WallMoment}; use crate::playback::cadence::ReportsWithoutWaiting; use crate::playback::{AdmittedPosition, Ticks}; use crate::server::write_queue::{ @@ -326,11 +350,27 @@ mod tests { Target::item(identifier.to_string()) } + /// The pair 0047 stores with a queue entry. + /// + /// Every case here is about the cadence and about what the queue did with a + /// report, and none of them asks an entry its age, so one agreeing pair is + /// what they hand in. `crate::server::write_queue` is where the arithmetic + /// over that pair is asked anything. + fn enqueued() -> WrittenAt { + WrittenAt::at(WallMoment::from_epoch(0, 0), WallMoment::from_epoch(0, 0)) + } + /// An item started at second zero, with its one report already on the /// queue. fn playing(identifier: &str, queue: &mut WriteQueue) -> Reporting { let mut reporting = Reporting::for_item(item(identifier), at(0)); - let did = reporting.report(ReportsWithoutWaiting::Started, played_to(0), at(0), queue); + let did = reporting.report( + ReportsWithoutWaiting::Started, + played_to(0), + at(0), + enqueued(), + queue, + ); assert_eq!(did, WhatTheEnqueueDid::Added); reporting } @@ -370,6 +410,7 @@ mod tests { ReportsWithoutWaiting::Seeked, position, moment, + enqueued(), &mut queue, )); } @@ -405,7 +446,7 @@ mod tests { "the interval was already due, so this case proves nothing" ); - let did = reporting.report(*event, played_to(1), at(1), &mut queue); + let did = reporting.report(*event, played_to(1), at(1), enqueued(), &mut queue); assert_eq!( did, @@ -433,6 +474,7 @@ mod tests { let did = reporting.observe( played_to(second), at(u64::try_from(second).expect("nine is a small number")), + enqueued(), &mut queue, ); assert_eq!( @@ -467,6 +509,7 @@ mod tests { match reporting.observe( played_to(second), at(u64::try_from(second).expect("a viewing fits in a day")), + enqueued(), queue, ) { WhatObservingDid::Reported(did) => reports.push(did), @@ -481,6 +524,7 @@ mod tests { ReportsWithoutWaiting::Paused, played_to(25), at(25), + enqueued(), &mut queue, )); observed(&mut reporting, &mut queue, 3600, &mut reports); @@ -488,12 +532,14 @@ mod tests { ReportsWithoutWaiting::Resumed, played_to(25), at(3600), + enqueued(), &mut queue, )); reports.push(reporting.report( ReportsWithoutWaiting::Seeked, played_to(30), at(3602), + enqueued(), &mut queue, )); observed(&mut reporting, &mut queue, 3610, &mut reports); @@ -501,6 +547,7 @@ mod tests { ReportsWithoutWaiting::Stopped, played_to(45), at(3620), + enqueued(), &mut queue, )); @@ -543,6 +590,7 @@ mod tests { ReportsWithoutWaiting::Started, played_to(0), at(1), + enqueued(), &mut queue, ); assert_eq!(did, WhatTheEnqueueDid::Added); @@ -550,12 +598,14 @@ mod tests { ReportsWithoutWaiting::Seeked, played_to(90), at(2), + enqueued(), &mut queue, ); second.report( ReportsWithoutWaiting::Seeked, played_to(15), at(3), + enqueued(), &mut queue, ); @@ -579,11 +629,11 @@ mod tests { let mut reporting = playing("the-film", &mut queue); assert_eq!( - reporting.observe(played_to(9), at(9), &mut queue), + reporting.observe(played_to(9), at(9), enqueued(), &mut queue), WhatObservingDid::NotDueYet ); assert_eq!( - reporting.observe(played_to(10), at(10), &mut queue), + reporting.observe(played_to(10), at(10), enqueued(), &mut queue), WhatObservingDid::Reported(WhatTheEnqueueDid::ReplacedInPlace) ); assert_eq!( @@ -591,11 +641,11 @@ mod tests { ReportedOn::TheInterval ); assert_eq!( - reporting.observe(played_to(19), at(19), &mut queue), + reporting.observe(played_to(19), at(19), enqueued(), &mut queue), WhatObservingDid::NotDueYet ); assert_eq!( - reporting.observe(played_to(20), at(20), &mut queue), + reporting.observe(played_to(20), at(20), enqueued(), &mut queue), WhatObservingDid::Reported(WhatTheEnqueueDid::ReplacedInPlace) ); assert_eq!( @@ -616,12 +666,13 @@ mod tests { ReportsWithoutWaiting::Seeked, played_to(second * 100), at(u64::try_from(second).expect("nine is a small number")), + enqueued(), &mut queue, ); } assert_eq!( - reporting.observe(played_to(901), at(10), &mut queue), + reporting.observe(played_to(901), at(10), enqueued(), &mut queue), WhatObservingDid::Reported(WhatTheEnqueueDid::ReplacedInPlace) ); } @@ -637,12 +688,13 @@ mod tests { ReportsWithoutWaiting::Paused, played_to(4), at(4), + enqueued(), &mut queue, ); for now in [4_u64, 5, 14, 3600, 86_400] { assert_eq!( - reporting.observe(played_to(4), at(now), &mut queue), + reporting.observe(played_to(4), at(now), enqueued(), &mut queue), WhatObservingDid::NothingIsPlaying, "a report was made {now} second(s) in while paused" ); @@ -666,22 +718,24 @@ mod tests { ReportsWithoutWaiting::Paused, played_to(4), at(4), + enqueued(), &mut queue, ); let did = reporting.report( ReportsWithoutWaiting::Resumed, played_to(4), at(3600), + enqueued(), &mut queue, ); assert_eq!(did, WhatTheEnqueueDid::ReplacedInPlace); assert_eq!( - reporting.observe(played_to(13), at(3609), &mut queue), + reporting.observe(played_to(13), at(3609), enqueued(), &mut queue), WhatObservingDid::NotDueYet ); assert_eq!( - reporting.observe(played_to(14), at(3610), &mut queue), + reporting.observe(played_to(14), at(3610), enqueued(), &mut queue), WhatObservingDid::Reported(WhatTheEnqueueDid::ReplacedInPlace) ); } @@ -694,7 +748,7 @@ mod tests { let mut reporting = Reporting::for_item(item("the-film"), at(0)); assert_eq!( - reporting.observe(played_to(0), at(30), &mut queue), + reporting.observe(played_to(0), at(30), enqueued(), &mut queue), WhatObservingDid::NothingIsPlaying ); assert!(queue.is_empty()); @@ -712,6 +766,7 @@ mod tests { ReportsWithoutWaiting::Started, played_to(0), at(0), + enqueued(), &mut queue, ); diff --git a/src/server/write_queue.rs b/src/server/write_queue.rs index f6e887e..a8da7f3 100644 --- a/src/server/write_queue.rs +++ b/src/server/write_queue.rs @@ -41,13 +41,24 @@ //! counter it carries is what a restore would restore rather than something that //! survives one today. //! -//! WHAT IS ALSO NOT HERE IS THE AGE. 0047 stores two moments per entry and -//! computes an age the way 0043 computes a cache entry's, for reporting and -//! never to act. That reading belongs with the two guards -//! [`crate::cache::freshness`] already carries, it acts on nothing here, and no -//! function below takes a clock reading at all - which is the same statement as -//! 0047's rule that an entry is never expired by age, in the form that cannot be -//! got wrong later. +//! THIS PARAGRAPH SAID THE AGE WAS NOT HERE. It said 0047 stores two moments per +//! entry and computes an age the way 0043 computes a cache entry's, that the +//! reading belongs with the two guards [`crate::cache::freshness`] already +//! carries, and that no function below takes a clock reading at all. The first +//! half is built: an entry carries the pair 0047 names and answers +//! [`Entry::age_at`] from it. The second half is what building it had to keep, +//! and it is kept by borrowing rather than by discipline - the pair, the +//! correction and the two guards are [`crate::cache::freshness`]'s own types, so +//! there is no second arithmetic here to drift from the one 0043 fixed. The third +//! is unchanged: every moment below arrives as an argument, nothing here reads a +//! clock, and an age is computed only where somebody asks for one. +//! +//! WHAT THE AGE DOES IS NOTHING, AND THAT IS 0047'S RULE RATHER THAN AN +//! UNFINISHED HALF. It is carried for reporting: not a reason to drop an entry, +//! not an input to the bound, and not a threshold. The absence used to be held by +//! there being no age at all, which is a guarantee that ends the moment one +//! arrives, so what holds it now is a case: an entry whose age is unreadable and +//! an entry a year old are both answered by the head and both delivered. //! //! # The number here is chosen and not measured //! @@ -56,6 +67,9 @@ //! than a thousand actions taken. #65 is the harness that would replace it with //! a measured number. +use crate::cache::freshness::{Age, Skew, WrittenAt}; +use crate::clock::WallMoment; + /// The entries one session's queue holds before an overflow drops something. /// /// From 0047, chosen rather than measured. What defends it is the coalescing @@ -172,6 +186,7 @@ pub struct Entry { target: Target, asserted_about: WhatIsAsserted, assertion: A, + enqueued_at: WrittenAt, } /// Written out by hand so an entry cannot carry its target's identifier into an @@ -215,6 +230,42 @@ impl Entry { pub const fn assertion(&self) -> &A { &self.assertion } + + /// The two moments this entry was enqueued with. + /// + /// 0047 names them: the server's own last stated time, and the device's wall + /// reading at the instant the action was queued. They are what an age + /// survives a restart on, and 0047 spends its argument on the alternative - + /// a restored queue that treated every entry as freshly enqueued would keep + /// its order and lose every age, so a client could say only that something is + /// pending. + /// + /// It is [`WrittenAt`] rather than a pair of this module's own because 0047 + /// says the age is computed the way 0043 computes a cache entry's, and one + /// type is how that stays true of a second arithmetic nobody wrote. + #[must_use] + pub const fn enqueued_at(&self) -> WrittenAt { + self.enqueued_at + } + + /// How long this entry has been waiting, on 0102's anchor. + /// + /// The whole of the computation is [`Age::at_read`]'s, with the same + /// correction and the same two guards, which is what 0047 asks for by naming + /// 0043 rather than describing an arithmetic of its own. `the_skew_now` is + /// `None` where there is no current measurement to correct against, which is + /// the ordinary case for a queue: the server the entry is waiting for is the + /// server that has not been reachable. + /// + /// WHAT THIS ANSWER IS FOR IS A SENTENCE A CLIENT SAYS, AND NOTHING HERE + /// READS IT. 0047 makes the age reporting and never a threshold: an action + /// somebody took is not less true because their device was off, and expiring + /// one would be the silent discard that record exists against arriving + /// through a mechanism that looks like hygiene. + #[must_use] + pub fn age_at(&self, the_devices_wall_now: WallMoment, the_skew_now: Option) -> Age { + Age::at_read(self.enqueued_at, the_devices_wall_now, the_skew_now) + } } /// What an entry that was dropped at the bound was about. @@ -344,11 +395,22 @@ impl WriteQueue { /// AT THE BOUND THE OLDEST GOES AND NOT THE NEWEST. The alternative refuses /// to record what somebody just did while holding something from three weeks /// ago, which is the version of the failure they are in front of. + /// + /// A REPLACEMENT KEEPS THE EARLIER ENTRY'S TWO MOMENTS, as it keeps its + /// position in the order, and 0047 says only the second of those. Taking the + /// later action's moments is the shape that reads as obvious - the statement + /// standing is the new one - and it reports every actively touched entry as + /// freshly enqueued, so a queue undelivered for a month says seconds for the + /// items somebody kept scrubbing. That is the failure 0047 names for a + /// restored queue arriving through the coalescing door instead, and it lands + /// on the person who used the application most, which is the same person + /// coalescing at enqueue exists to protect. pub fn enqueue( &mut self, target: Target, asserted_about: WhatIsAsserted, assertion: A, + enqueued_at: WrittenAt, ) -> WhatTheEnqueueDid { if let Some(held) = self .entries @@ -375,6 +437,7 @@ impl WriteQueue { target, asserted_about, assertion, + enqueued_at, }); self.next_order = self.next_order.saturating_add(1); what_it_did @@ -421,6 +484,14 @@ mod tests { use super::{ A_SESSIONS_QUEUE_HOLDS_AT_MOST, Target, WhatIsAsserted, WhatTheEnqueueDid, WriteQueue, }; + use crate::cache::freshness::{Age, Skew, WhyTheAgeIsUnreadable, WrittenAt}; + use crate::clock::WallMoment; + use core::time::Duration; + + /// Seconds in a day, for the moments below, so that a case saying "a year" + /// says it in the units 0043's bound is written in rather than in a numeral + /// nobody can read. + const A_DAY: i64 = 24 * 60 * 60; fn item(identifier: &str) -> Target { Target::item(identifier.to_string()) @@ -430,8 +501,21 @@ mod tests { WriteQueue::empty() } + /// The pair 0047 stores at enqueue, with the two moments agreeing. + /// + /// Both are handed in, because nothing in this module reads a clock. A + /// server and a device that agree put the skew at write at zero, which is + /// what lets the cases below read as the arithmetic they are about; the one + /// case that is about a correction builds its own pair. + fn moments(seconds: i64) -> WrittenAt { + WrittenAt::at( + WallMoment::from_epoch(seconds, 0), + WallMoment::from_epoch(seconds, 0), + ) + } + fn put(queue: &mut WriteQueue, id: &str, kind: WhatIsAsserted, said: &str) { - queue.enqueue(item(id), kind, said.to_string()); + queue.enqueue(item(id), kind, said.to_string(), moments(0)); } /// The order is a counter increased once per entry, which is what a clock @@ -460,6 +544,7 @@ mod tests { item("a"), WhatIsAsserted::PlaybackPosition, "at 30".to_string(), + moments(0), ); assert_eq!(what_it_did, WhatTheEnqueueDid::ReplacedInPlace); @@ -480,6 +565,7 @@ mod tests { item("a"), WhatIsAsserted::PlaybackPosition, format!("at {position}"), + moments(0), ); } @@ -534,6 +620,7 @@ mod tests { item(&format!("item-{target}")), WhatIsAsserted::Watched, "yes".to_string(), + moments(0), ); } assert_eq!(queue.len(), A_SESSIONS_QUEUE_HOLDS_AT_MOST); @@ -543,6 +630,7 @@ mod tests { item("one-too-many"), WhatIsAsserted::Watched, "yes".to_string(), + moments(0), ); let WhatTheEnqueueDid::DroppedTheOldest(dropped) = what_it_did else { @@ -573,6 +661,7 @@ mod tests { item(&format!("item-{target}")), WhatIsAsserted::Watched, "yes".to_string(), + moments(0), ); } @@ -581,6 +670,7 @@ mod tests { item("item-7"), WhatIsAsserted::Watched, format!("still yes {again}"), + moments(0), ); assert_eq!(what_it_did, WhatTheEnqueueDid::ReplacedInPlace); } @@ -636,6 +726,7 @@ mod tests { item(&format!("item-{target}")), WhatIsAsserted::Watched, "yes".to_string(), + moments(0), ); } assert_eq!(queue.dropped(), 1); @@ -700,4 +791,187 @@ mod tests { assert!(queue.after_it_was_delivered().is_none()); assert_eq!(queue.dropped(), 0); } + + /// The two moments 0047 stores are the two the entry hands back, so a restore + /// has something to compute an age from rather than a queue that can say only + /// that something is pending. + #[test] + fn an_entry_carries_the_two_moments_it_was_enqueued_with() { + let mut queue = a_queue(); + queue.enqueue( + item("a"), + WhatIsAsserted::Watched, + "yes".to_string(), + WrittenAt::at( + WallMoment::from_epoch(1_700_000_000, 0), + WallMoment::from_epoch(1_700_000_040, 0), + ), + ); + + let enqueued_at = queue.entries()[0].enqueued_at(); + assert_eq!( + enqueued_at + .the_servers_stated_moment() + .seconds_from_the_epoch(), + 1_700_000_000 + ); + assert_eq!( + enqueued_at + .the_devices_wall_moment() + .seconds_from_the_epoch(), + 1_700_000_040 + ); + } + + /// The age is the device difference, and it is the whole of 0043's + /// arithmetic rather than a second one: a minute on the device is a minute. + #[test] + fn an_entry_reports_how_long_it_has_been_waiting() { + let mut queue = a_queue(); + put(&mut queue, "a", WhatIsAsserted::Watched, "yes"); + + assert_eq!( + queue.entries()[0].age_at(WallMoment::from_epoch(60, 0), None), + Age::Of(Duration::from_secs(60)) + ); + } + + /// A device clock that moved between the enqueue and the reading moved both + /// device readings and neither server moment, and the correction removes + /// exactly that movement. This is 0043's correction reaching the queue, + /// which is what 0047 asks for by naming that record rather than describing + /// an arithmetic of its own. + #[test] + fn a_device_clock_that_jumped_forward_is_corrected_out_of_the_age() { + let mut queue = a_queue(); + queue.enqueue( + item("a"), + WhatIsAsserted::Watched, + "yes".to_string(), + moments(1000), + ); + + // Sixty seconds passed and the device also jumped forty seconds ahead of + // the server, so its own reading is a hundred seconds on. + let age = queue.entries()[0].age_at( + WallMoment::from_epoch(1100, 0), + Some(Skew::between( + WallMoment::from_epoch(1060, 0), + WallMoment::from_epoch(1100, 0), + )), + ); + + assert_eq!(age, Age::Of(Duration::from_secs(60))); + } + + /// The first of 0043's two guards, reaching a queued action: a device that + /// came up believing it is earlier than when the action was taken. + #[test] + fn a_device_clock_that_moved_backwards_leaves_the_age_unreadable() { + let mut queue = a_queue(); + queue.enqueue( + item("a"), + WhatIsAsserted::Watched, + "yes".to_string(), + moments(1000), + ); + + assert_eq!( + queue.entries()[0].age_at(WallMoment::from_epoch(940, 0), None), + Age::Unreadable(WhyTheAgeIsUnreadable::ItComputedAsNegative) + ); + } + + /// The second guard: a device that jumped forward past the bound beyond + /// which a computed age is not believed. + #[test] + fn an_age_past_the_sanity_bound_is_unreadable() { + let mut queue = a_queue(); + put(&mut queue, "a", WhatIsAsserted::Watched, "yes"); + + assert_eq!( + queue.entries()[0].age_at(WallMoment::from_epoch(400 * A_DAY, 0), None), + Age::Unreadable(WhyTheAgeIsUnreadable::ItPassedTheSanityBound) + ); + } + + /// A replacement keeps the earlier entry's moments, as it keeps its position + /// in the order. Taking the later action's moments reports every actively + /// touched entry as freshly enqueued, which is 0047's restored-queue failure + /// arriving through the coalescing door, and it lands on the person who used + /// the application most. + #[test] + fn a_replacement_keeps_the_earlier_entrys_moments() { + let mut queue = a_queue(); + queue.enqueue( + item("a"), + WhatIsAsserted::PlaybackPosition, + "at 10".to_string(), + moments(0), + ); + + let what_it_did = queue.enqueue( + item("a"), + WhatIsAsserted::PlaybackPosition, + "at 90".to_string(), + moments(20 * A_DAY), + ); + + assert_eq!(what_it_did, WhatTheEnqueueDid::ReplacedInPlace); + assert_eq!(queue.entries()[0].assertion(), "at 90"); + assert_eq!( + queue.entries()[0] + .enqueued_at() + .the_devices_wall_moment() + .seconds_from_the_epoch(), + 0, + "the replacement took the later action's moments" + ); + assert_eq!( + queue.entries()[0].age_at(WallMoment::from_epoch(21 * A_DAY, 0), None), + Age::Of(Duration::from_hours(21 * 24)), + "the entry reported the age of the last thing somebody said" + ); + } + + /// 0047's rule that nothing is ever expired by age, held by a case rather + /// than by there being no age to expire on. An entry a year old and an entry + /// whose age is unreadable are both at the head in order and both delivered. + #[test] + fn an_entry_is_never_expired_by_age() { + let mut queue = a_queue(); + queue.enqueue( + item("a-year-ago"), + WhatIsAsserted::Watched, + "yes".to_string(), + moments(0), + ); + queue.enqueue( + item("after-a-clock-that-moved"), + WhatIsAsserted::Watched, + "yes".to_string(), + moments(500 * A_DAY), + ); + + let now = WallMoment::from_epoch(366 * A_DAY, 0); + assert!(matches!( + queue.entries()[0].age_at(now, None), + Age::Unreadable(WhyTheAgeIsUnreadable::ItPassedTheSanityBound) + )); + assert!(matches!( + queue.entries()[1].age_at(now, None), + Age::Unreadable(WhyTheAgeIsUnreadable::ItComputedAsNegative) + )); + + assert_eq!(queue.len(), 2); + assert_eq!(queue.next_to_deliver().map(super::Entry::order), Some(1)); + + let mut delivered = Vec::new(); + while let Some(taken) = queue.after_it_was_delivered() { + delivered.push(taken.target().as_str().to_string()); + } + + assert_eq!(delivered, ["a-year-ago", "after-a-clock-that-moved"]); + assert_eq!(queue.dropped(), 0); + } } diff --git a/src/session/mid_playback.rs b/src/session/mid_playback.rs index 973490d..2e6de27 100644 --- a/src/session/mid_playback.rs +++ b/src/session/mid_playback.rs @@ -64,6 +64,7 @@ //! moment, which is what 0057 makes every position, and this module claims //! nothing about the gap. +use crate::cache::freshness::WrittenAt; use crate::diagnostics::redaction::FieldName; use crate::diagnostics::{Diagnostics, EventName, Field, FieldValue, Severity}; use crate::playback::AdmittedPosition; @@ -243,17 +244,25 @@ pub fn a_report_was_rejected( /// The current position is the client's reading at this moment, taken the way /// 0057 takes every position, and the report leaves the cadence interval where /// it was. +/// +/// `enqueued_at` is the pair 0047 stores with a queue entry, handed in for the +/// reason every moment in this core is. It reaches the queue on the success +/// branch alone, and the entry the rejection left at the head keeps its own +/// moments under 0047's coalescing, so what a client can say afterwards is how +/// long this item's position has been undelivered rather than how long ago the +/// token was renewed. pub fn the_renewal_ended( renewals: &mut Renewals, how: HowTheRenewalEnded, reporting: &mut Reporting, current: AdmittedPosition, + enqueued_at: WrittenAt, queue: &mut WriteQueue, diagnostics: &Diagnostics<'_>, ) -> WhatTheOutcomeDoesToPlayback { match renewals.ended(how) { WhatTheOutcomeDoes::RetryTheWaitingCallsOnce => { - let what_the_queue_did = reporting.report_after_a_renewal(current, queue); + let what_the_queue_did = reporting.report_after_a_renewal(current, enqueued_at, queue); WhatTheOutcomeDoesToPlayback::CurrentPositionReportedAndTheDrainResumes( what_the_queue_did, ) @@ -283,6 +292,7 @@ mod tests { POSITIONS_HELD, WhatARejectedReportDoes, WhatTheOutcomeDoesToPlayback, a_report_was_rejected, the_renewal_ended, }; + use crate::cache::freshness::WrittenAt; use crate::clock::{Clocks, ElapsedInstant, SteadyInstant, WallMoment}; use crate::diagnostics::redaction::CorrelatorSalt; use crate::diagnostics::{Diagnostics, DiagnosticsSink, Event, FieldValue, Severity}; @@ -390,6 +400,7 @@ mod tests { ReportsWithoutWaiting::Started, played_to(0), at(0), + enqueued(), &mut queue ), WhatTheEnqueueDid::Added @@ -399,6 +410,7 @@ mod tests { ReportsWithoutWaiting::Seeked, played_to(10), at(1), + enqueued(), &mut queue ), WhatTheEnqueueDid::ReplacedInPlace @@ -409,6 +421,7 @@ mod tests { ReportsWithoutWaiting::Started, played_to(0), at(2), + enqueued(), &mut queue ), WhatTheEnqueueDid::Added @@ -417,6 +430,15 @@ mod tests { (queue, film) } + /// The pair 0047 stores with a queue entry. + /// + /// The cases here are about what each renewal outcome does to the queue and + /// to the one report the success branch makes, and none of them asks an + /// entry its age, so one agreeing pair is what they hand in. + fn enqueued() -> WrittenAt { + WrittenAt::at(WallMoment::from_epoch(0, 0), WallMoment::from_epoch(0, 0)) + } + fn rejected_under(renewals: &Renewals) -> Rejection { Rejection { went_out_under: renewals.generation(), @@ -494,6 +516,7 @@ mod tests { HowTheRenewalEnded::AFreshToken, &mut film, played_to(12), + enqueued(), &mut queue, &diagnostics, ); @@ -576,6 +599,7 @@ mod tests { HowTheRenewalEnded::AFreshToken, &mut film, played_to(25), + enqueued(), &mut queue, &diagnostics, ); @@ -621,6 +645,7 @@ mod tests { HowTheRenewalEnded::TheServerRefusedIt, &mut film, played_to(25), + enqueued(), &mut queue, &diagnostics, ); @@ -661,6 +686,7 @@ mod tests { HowTheRenewalEnded::NothingAnswered, &mut film, played_to(25), + enqueued(), &mut queue, &diagnostics, ); @@ -692,6 +718,7 @@ mod tests { HowTheRenewalEnded::AFreshToken, &mut film, played_to(25), + enqueued(), &mut queue, &diagnostics, );