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, );