From 80882f0aa7fd3048618ca99c508f37e153849ca4 Mon Sep 17 00:00:00 2001 From: Nils Lehnen <30603423+iderex@users.noreply.github.com> Date: Fri, 4 Sep 2026 16:38:21 +0200 Subject: [PATCH] Report a dropped queue entry as it happens, with the correlator and not the item (#47) 0047 says every drop at the bound is reported at the moment it happens, as an event through 0100, carrying the kind of action and the correlator for its target under 0071 rather than the identifier, and that the standing count is kept beside it for a client that was not listening. The count was here and the event was not. The enqueue makes the report itself, and this replaces handing the drop back for somebody else to make it. That shape left the record's "every drop" resting on every caller remembering, and a caller that ignores the answer produces exactly the silent discard 0047 exists against. The value is still handed back, because a caller may want to say something of its own about what was lost. At `failure` rather than `notice`, which is 0105's sentence about a dropped queue entry and its reason: a cache entry can be fetched again and a person's own action cannot, so a client filtering `notice` out would stop seeing the one thing this record exists to prevent being silent about. The target leaves as `for-target`, reduced, so the boundary turns it into 0071's correlator and the identifier reaches no sink. The kind leaves as `asserted-about`, carried whole, because it is one of a fixed set this module declares and two people running the same build cannot hold different values for it. Both are in `lifecycle::every_field_name_the_core_emits`, which is 0071's rule as data, and the suite refuses a name declared and not stated. The order the dropped entry stood in is on the value handed back and not on the event: 0047 names two things the report carries, and a counter meaningful only inside one queue is not one of them. 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/lifecycle/mod.rs | 3 + src/playback/report.rs | 149 +++++++++++- src/server/write_queue.rs | 250 +++++++++++++++++++- src/session/mid_playback.rs | 23 +- tests/the_rule_as_data_names_every_field.rs | 7 +- 5 files changed, 407 insertions(+), 25 deletions(-) diff --git a/src/lifecycle/mod.rs b/src/lifecycle/mod.rs index 72ed851..cb70aa8 100644 --- a/src/lifecycle/mod.rs +++ b/src/lifecycle/mod.rs @@ -59,6 +59,7 @@ use crate::cache::envelope; use crate::diagnostics::DiagnosticsSink; use crate::diagnostics::redaction::FieldName; use crate::failure::Failure; +use crate::server::write_queue; use crate::session::SecretStore; use crate::session::mid_playback; @@ -509,6 +510,8 @@ pub const fn every_field_name_the_core_emits() -> &'static [FieldName] { envelope::CHECK, envelope::VERSION_FOUND, mid_playback::POSITIONS_HELD, + write_queue::FOR_TARGET, + write_queue::ASSERTED_ABOUT, ] } diff --git a/src/playback/report.rs b/src/playback/report.rs index 7c68678..6485c17 100644 --- a/src/playback/report.rs +++ b/src/playback/report.rs @@ -69,6 +69,7 @@ use super::Ticks; use super::cadence::{ReportsWithoutWaiting, TheInterval}; use crate::cache::freshness::WrittenAt; use crate::clock::ElapsedInstant; +use crate::diagnostics::Diagnostics; use crate::server::write_queue::{Target, WhatIsAsserted, WhatTheEnqueueDid, WriteQueue}; /// What occasioned a report. @@ -222,6 +223,7 @@ impl Reporting { at: ElapsedInstant, enqueued_at: WrittenAt, queue: &mut WriteQueue, + diagnostics: &Diagnostics<'_>, ) -> WhatTheEnqueueDid { let what_the_queue_did = queue.enqueue( self.target.clone(), @@ -231,6 +233,7 @@ impl Reporting { reported_on: ReportedOn::Event(event), }, enqueued_at, + diagnostics, ); self.interval = self.interval.after(event, at); what_the_queue_did @@ -259,6 +262,7 @@ impl Reporting { now: ElapsedInstant, enqueued_at: WrittenAt, queue: &mut WriteQueue, + diagnostics: &Diagnostics<'_>, ) -> WhatObservingDid { if !self.interval.is_running() { return WhatObservingDid::NothingIsPlaying; @@ -274,6 +278,7 @@ impl Reporting { reported_on: ReportedOn::TheInterval, }, enqueued_at, + diagnostics, ); self.interval = self.interval.after_a_report_at(now); WhatObservingDid::Reported(what_the_queue_did) @@ -305,6 +310,7 @@ impl Reporting { position: AdmittedPosition, enqueued_at: WrittenAt, queue: &mut WriteQueue, + diagnostics: &Diagnostics<'_>, ) -> WhatTheEnqueueDid { queue.enqueue( self.target.clone(), @@ -314,6 +320,7 @@ impl Reporting { reported_on: ReportedOn::AfterARenewal, }, enqueued_at, + diagnostics, ) } } @@ -329,7 +336,10 @@ mod tests { use super::{PositionReport, ReportedOn, Reporting, WhatObservingDid}; use crate::cache::freshness::WrittenAt; - use crate::clock::{ElapsedInstant, WallMoment}; + use crate::clock::{Clocks, ElapsedInstant, SteadyInstant, WallMoment}; + use crate::diagnostics::Diagnostics; + use crate::diagnostics::Severity; + use crate::diagnostics::redaction::CorrelatorSalt; use crate::playback::cadence::ReportsWithoutWaiting; use crate::playback::{AdmittedPosition, Ticks}; use crate::server::write_queue::{ @@ -360,6 +370,46 @@ mod tests { WrittenAt::at(WallMoment::from_epoch(0, 0), WallMoment::from_epoch(0, 0)) } + /// A clock that does not move, because no case here reads one. + /// + /// The facility below stamps an event with the wall moment, and no case here + /// emits one: 0047 reports a drop, a drop happens at the bound, and this + /// module's cases hold two items at most. + #[derive(Debug, Default)] + struct Still; + + impl Clocks for Still { + fn steady(&self) -> SteadyInstant { + SteadyInstant::from_nanos(0) + } + + fn elapsed(&self) -> ElapsedInstant { + ElapsedInstant::from_nanos(0) + } + + fn wall(&self) -> WallMoment { + WallMoment::from_epoch(0, 0) + } + } + + /// A facility with no sink, which is what a report costs when nobody is + /// listening. + /// + /// It is here because the queue reports its own drops now, so an enqueue + /// takes one. `crate::server::write_queue` is where the event itself is + /// asked anything, and a collector here would be a second place watching + /// one subsystem's report. + static STILL: Still = Still; + + fn nobody_listening() -> Diagnostics<'static> { + Diagnostics::new( + &STILL, + None, + Severity::Detail, + CorrelatorSalt::from_bytes([0x5a; CorrelatorSalt::WIDTH]), + ) + } + /// An item started at second zero, with its one report already on the /// queue. fn playing(identifier: &str, queue: &mut WriteQueue) -> Reporting { @@ -370,6 +420,7 @@ mod tests { at(0), enqueued(), queue, + &nobody_listening(), ); assert_eq!(did, WhatTheEnqueueDid::Added); reporting @@ -412,6 +463,7 @@ mod tests { moment, enqueued(), &mut queue, + &nobody_listening(), )); } @@ -446,7 +498,14 @@ mod tests { "the interval was already due, so this case proves nothing" ); - let did = reporting.report(*event, played_to(1), at(1), enqueued(), &mut queue); + let did = reporting.report( + *event, + played_to(1), + at(1), + enqueued(), + &mut queue, + &nobody_listening(), + ); assert_eq!( did, @@ -476,6 +535,7 @@ mod tests { at(u64::try_from(second).expect("nine is a small number")), enqueued(), &mut queue, + &nobody_listening(), ); assert_eq!( did, @@ -511,6 +571,7 @@ mod tests { at(u64::try_from(second).expect("a viewing fits in a day")), enqueued(), queue, + &nobody_listening(), ) { WhatObservingDid::Reported(did) => reports.push(did), WhatObservingDid::NotDueYet | WhatObservingDid::NothingIsPlaying => {} @@ -526,6 +587,7 @@ mod tests { at(25), enqueued(), &mut queue, + &nobody_listening(), )); observed(&mut reporting, &mut queue, 3600, &mut reports); reports.push(reporting.report( @@ -534,6 +596,7 @@ mod tests { at(3600), enqueued(), &mut queue, + &nobody_listening(), )); reports.push(reporting.report( ReportsWithoutWaiting::Seeked, @@ -541,6 +604,7 @@ mod tests { at(3602), enqueued(), &mut queue, + &nobody_listening(), )); observed(&mut reporting, &mut queue, 3610, &mut reports); reports.push(reporting.report( @@ -549,6 +613,7 @@ mod tests { at(3620), enqueued(), &mut queue, + &nobody_listening(), )); assert_eq!( @@ -592,6 +657,7 @@ mod tests { at(1), enqueued(), &mut queue, + &nobody_listening(), ); assert_eq!(did, WhatTheEnqueueDid::Added); first.report( @@ -600,6 +666,7 @@ mod tests { at(2), enqueued(), &mut queue, + &nobody_listening(), ); second.report( ReportsWithoutWaiting::Seeked, @@ -607,6 +674,7 @@ mod tests { at(3), enqueued(), &mut queue, + &nobody_listening(), ); assert_eq!(queue.len(), 2); @@ -629,11 +697,23 @@ mod tests { let mut reporting = playing("the-film", &mut queue); assert_eq!( - reporting.observe(played_to(9), at(9), enqueued(), &mut queue), + reporting.observe( + played_to(9), + at(9), + enqueued(), + &mut queue, + &nobody_listening() + ), WhatObservingDid::NotDueYet ); assert_eq!( - reporting.observe(played_to(10), at(10), enqueued(), &mut queue), + reporting.observe( + played_to(10), + at(10), + enqueued(), + &mut queue, + &nobody_listening() + ), WhatObservingDid::Reported(WhatTheEnqueueDid::ReplacedInPlace) ); assert_eq!( @@ -641,11 +721,23 @@ mod tests { ReportedOn::TheInterval ); assert_eq!( - reporting.observe(played_to(19), at(19), enqueued(), &mut queue), + reporting.observe( + played_to(19), + at(19), + enqueued(), + &mut queue, + &nobody_listening() + ), WhatObservingDid::NotDueYet ); assert_eq!( - reporting.observe(played_to(20), at(20), enqueued(), &mut queue), + reporting.observe( + played_to(20), + at(20), + enqueued(), + &mut queue, + &nobody_listening() + ), WhatObservingDid::Reported(WhatTheEnqueueDid::ReplacedInPlace) ); assert_eq!( @@ -668,11 +760,18 @@ mod tests { at(u64::try_from(second).expect("nine is a small number")), enqueued(), &mut queue, + &nobody_listening(), ); } assert_eq!( - reporting.observe(played_to(901), at(10), enqueued(), &mut queue), + reporting.observe( + played_to(901), + at(10), + enqueued(), + &mut queue, + &nobody_listening() + ), WhatObservingDid::Reported(WhatTheEnqueueDid::ReplacedInPlace) ); } @@ -690,11 +789,18 @@ mod tests { at(4), enqueued(), &mut queue, + &nobody_listening(), ); for now in [4_u64, 5, 14, 3600, 86_400] { assert_eq!( - reporting.observe(played_to(4), at(now), enqueued(), &mut queue), + reporting.observe( + played_to(4), + at(now), + enqueued(), + &mut queue, + &nobody_listening() + ), WhatObservingDid::NothingIsPlaying, "a report was made {now} second(s) in while paused" ); @@ -720,6 +826,7 @@ mod tests { at(4), enqueued(), &mut queue, + &nobody_listening(), ); let did = reporting.report( ReportsWithoutWaiting::Resumed, @@ -727,15 +834,28 @@ mod tests { at(3600), enqueued(), &mut queue, + &nobody_listening(), ); assert_eq!(did, WhatTheEnqueueDid::ReplacedInPlace); assert_eq!( - reporting.observe(played_to(13), at(3609), enqueued(), &mut queue), + reporting.observe( + played_to(13), + at(3609), + enqueued(), + &mut queue, + &nobody_listening() + ), WhatObservingDid::NotDueYet ); assert_eq!( - reporting.observe(played_to(14), at(3610), enqueued(), &mut queue), + reporting.observe( + played_to(14), + at(3610), + enqueued(), + &mut queue, + &nobody_listening() + ), WhatObservingDid::Reported(WhatTheEnqueueDid::ReplacedInPlace) ); } @@ -748,7 +868,13 @@ mod tests { let mut reporting = Reporting::for_item(item("the-film"), at(0)); assert_eq!( - reporting.observe(played_to(0), at(30), enqueued(), &mut queue), + reporting.observe( + played_to(0), + at(30), + enqueued(), + &mut queue, + &nobody_listening() + ), WhatObservingDid::NothingIsPlaying ); assert!(queue.is_empty()); @@ -768,6 +894,7 @@ mod tests { at(0), enqueued(), &mut queue, + &nobody_listening(), ); let entry = the_entry_for("the-film", &queue); diff --git a/src/server/write_queue.rs b/src/server/write_queue.rs index a8da7f3..8713c99 100644 --- a/src/server/write_queue.rs +++ b/src/server/write_queue.rs @@ -60,6 +60,14 @@ //! 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 DROP REPORT IS MADE HERE AND THE STANDING COUNT IS BESIDE IT. 0047 asks +//! for both, because the two answer different people: an event reaches a client +//! that was listening at the moment it happened, and a count reaches one that +//! was not. The event carries the kind and the correlator 0071 defines for the +//! target, never the identifier, and [`Diagnostics`] is what turns the second +//! into the first, so the reduction is decided at the boundary rather than by +//! this module remembering to do it. +//! //! # The number here is chosen and not measured //! //! 0047 says so of its bound, and says what makes a thousand defensible: with @@ -69,6 +77,31 @@ use crate::cache::freshness::{Age, Skew, WrittenAt}; use crate::clock::WallMoment; +use crate::diagnostics::redaction::FieldName; +use crate::diagnostics::{Diagnostics, EventName, Field, FieldValue, Severity}; + +/// The event 0047 owes at the moment an entry is dropped at the bound. +/// +/// At `failure` rather than `notice`, which is 0105's sentence about a dropped +/// queue entry and its reason: a cache entry can be fetched again and a person's +/// own action cannot, so what did not happen here is something somebody did +/// reaching the server. A client filtering `notice` out is a client that would +/// stop seeing the one thing 0047 exists to prevent being silent about. +const AN_ENTRY_WAS_DROPPED: EventName = EventName::declared("write-queue.entry-dropped"); + +/// Which item the dropped entry was about. +/// +/// Reduced, so it leaves as the correlator 0071 defines and never as the +/// identifier. That is 0047's own requirement for this report, and it is what +/// makes two drops for one item readable as one item rather than as two. +pub(crate) const FOR_TARGET: FieldName = FieldName::reduced("for-target"); + +/// Which statement about that item was dropped. +/// +/// Carried whole: it is one of a fixed set this module declares, so two people +/// running the same build against the same server cannot hold different values +/// for it, which is 0068's question and the one 0071's first treatment is for. +pub(crate) const ASSERTED_ABOUT: FieldName = FieldName::carried_whole("asserted-about"); /// The entries one session's queue holds before an overflow drops something. /// @@ -272,9 +305,12 @@ impl Entry { /// /// 0047 requires a drop to be reported at the moment it happens, through the /// interface in 0100, carrying the kind of action and the correlator for its -/// target rather than the identifier. This is what the enqueue hands back so -/// that the report is made by whoever is holding a diagnostics facility, and it -/// carries the target so the correlator can be derived from it there. +/// target rather than the identifier. THE ENQUEUE MAKES THAT REPORT ITSELF AND +/// THIS PARAGRAPH SAID IT WAS HANDED BACK FOR SOMEBODY ELSE TO MAKE. Handing it +/// back left the record's "every drop" resting on every caller remembering, and +/// a caller that ignores the answer produces exactly the silent discard the +/// whole record exists against. What is still handed back is this value, because +/// a caller may want to say something of its own about what was lost. /// /// IT CARRIES NO ASSERTION. What was dropped is gone, and holding the value /// would put a person's own action into the type whose whole purpose is to say @@ -396,6 +432,17 @@ impl WriteQueue { /// 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 DROP IS REPORTED HERE AND NOT BY THE CALLER. 0047 says every drop is + /// reported at the moment it happens, and the moment it happens is inside + /// this call. The diagnostics facility is a parameter rather than something + /// the queue holds, for the reason every clock reading is one: this module + /// owns no facility of the client's and reaches for nothing. + /// + /// The order the dropped entry stood in is on the value handed back and not + /// on the event. 0047 names two things the report carries, the kind and the + /// correlator, and a counter meaningful only inside one queue is not one of + /// them. + /// /// 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 @@ -411,6 +458,7 @@ impl WriteQueue { asserted_about: WhatIsAsserted, assertion: A, enqueued_at: WrittenAt, + diagnostics: &Diagnostics<'_>, ) -> WhatTheEnqueueDid { if let Some(held) = self .entries @@ -425,11 +473,23 @@ impl WriteQueue { if self.entries.len() >= A_SESSIONS_QUEUE_HOLDS_AT_MOST { let oldest = self.entries.remove(0); self.dropped = self.dropped.saturating_add(1); - what_it_did = WhatTheEnqueueDid::DroppedTheOldest(Dropped { + let dropped = Dropped { target: oldest.target, asserted_about: oldest.asserted_about, order: oldest.order, - }); + }; + diagnostics.emit( + Severity::Failure, + AN_ENTRY_WAS_DROPPED, + &[ + Field::new(FOR_TARGET, FieldValue::Text(dropped.target.as_str())), + Field::new( + ASSERTED_ABOUT, + FieldValue::Text(dropped.asserted_about.as_str()), + ), + ], + ); + what_it_did = WhatTheEnqueueDid::DroppedTheOldest(dropped); } self.entries.push(Entry { @@ -485,8 +545,92 @@ mod tests { A_SESSIONS_QUEUE_HOLDS_AT_MOST, Target, WhatIsAsserted, WhatTheEnqueueDid, WriteQueue, }; use crate::cache::freshness::{Age, Skew, WhyTheAgeIsUnreadable, WrittenAt}; - use crate::clock::WallMoment; + use crate::clock::{Clocks, ElapsedInstant, SteadyInstant, WallMoment}; + use crate::diagnostics::redaction::CorrelatorSalt; + use crate::diagnostics::{Diagnostics, DiagnosticsSink, Event, FieldValue, Severity}; use core::time::Duration; + use std::sync::Mutex; + + /// A clock that does not move. + /// + /// 0100 stamps an event with the wall moment and nothing here reads it back, + /// which that record says of the field itself: it is for lining an event up + /// against a server's own log, and no core behaviour depends on it. + #[derive(Debug, Default)] + struct Still; + + impl Clocks for Still { + fn steady(&self) -> SteadyInstant { + SteadyInstant::from_nanos(0) + } + + fn elapsed(&self) -> ElapsedInstant { + ElapsedInstant::from_nanos(0) + } + + fn wall(&self) -> WallMoment { + WallMoment::from_epoch(0, 0) + } + } + + static STILL: Still = Still; + + fn a_salt() -> CorrelatorSalt { + CorrelatorSalt::from_bytes([0x5a; CorrelatorSalt::WIDTH]) + } + + /// The facility with nobody listening, which is what most cases here want: + /// they are about the order, the coalescing and the bound, and a report + /// nobody receives changes none of the three. + fn nobody_listening() -> Diagnostics<'static> { + Diagnostics::new(&STILL, None, Severity::Detail, a_salt()) + } + + /// One event as a case reads it: how much attention it is worth, what it is + /// called, and its fields as name and text. + type Told = (Severity, &'static str, Vec<(&'static str, String)>); + + /// Keeps each event's severity, name and fields as text, so a case reads + /// what the client received rather than what the queue was asked to send. + #[derive(Debug, Default)] + struct Collector { + told: Mutex>, + } + + impl Collector { + fn told(&self) -> Vec { + self.told + .lock() + .expect("the fixture holds no poisoned lock") + .clone() + } + } + + impl DiagnosticsSink for Collector { + fn event(&self, event: &Event<'_>) { + let fields = event + .fields() + .iter() + .map(|field| { + let value = match field.value() { + FieldValue::Text(text) => text.to_owned(), + FieldValue::Count(count) => count.to_string(), + FieldValue::Interval(interval) => format!("{interval:?}"), + FieldValue::Truth(truth) => truth.to_string(), + }; + (field.name().as_str(), value) + }) + .collect(); + self.told + .lock() + .expect("the fixture holds no poisoned lock") + .push((event.severity(), event.name().as_str(), fields)); + } + } + + fn listening(collector: &Collector) -> Diagnostics<'_> { + Diagnostics::new(&STILL, Some(collector), Severity::Detail, a_salt()) + } /// 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 @@ -515,7 +659,13 @@ mod tests { } fn put(queue: &mut WriteQueue, id: &str, kind: WhatIsAsserted, said: &str) { - queue.enqueue(item(id), kind, said.to_string(), moments(0)); + queue.enqueue( + item(id), + kind, + said.to_string(), + moments(0), + &nobody_listening(), + ); } /// The order is a counter increased once per entry, which is what a clock @@ -545,6 +695,7 @@ mod tests { WhatIsAsserted::PlaybackPosition, "at 30".to_string(), moments(0), + &nobody_listening(), ); assert_eq!(what_it_did, WhatTheEnqueueDid::ReplacedInPlace); @@ -566,6 +717,7 @@ mod tests { WhatIsAsserted::PlaybackPosition, format!("at {position}"), moments(0), + &nobody_listening(), ); } @@ -621,6 +773,7 @@ mod tests { WhatIsAsserted::Watched, "yes".to_string(), moments(0), + &nobody_listening(), ); } assert_eq!(queue.len(), A_SESSIONS_QUEUE_HOLDS_AT_MOST); @@ -631,6 +784,7 @@ mod tests { WhatIsAsserted::Watched, "yes".to_string(), moments(0), + &nobody_listening(), ); let WhatTheEnqueueDid::DroppedTheOldest(dropped) = what_it_did else { @@ -662,6 +816,7 @@ mod tests { WhatIsAsserted::Watched, "yes".to_string(), moments(0), + &nobody_listening(), ); } @@ -671,6 +826,7 @@ mod tests { WhatIsAsserted::Watched, format!("still yes {again}"), moments(0), + &nobody_listening(), ); assert_eq!(what_it_did, WhatTheEnqueueDid::ReplacedInPlace); } @@ -727,6 +883,7 @@ mod tests { WhatIsAsserted::Watched, "yes".to_string(), moments(0), + &nobody_listening(), ); } assert_eq!(queue.dropped(), 1); @@ -806,6 +963,7 @@ mod tests { WallMoment::from_epoch(1_700_000_000, 0), WallMoment::from_epoch(1_700_000_040, 0), ), + &nobody_listening(), ); let enqueued_at = queue.entries()[0].enqueued_at(); @@ -849,6 +1007,7 @@ mod tests { WhatIsAsserted::Watched, "yes".to_string(), moments(1000), + &nobody_listening(), ); // Sixty seconds passed and the device also jumped forty seconds ahead of @@ -874,6 +1033,7 @@ mod tests { WhatIsAsserted::Watched, "yes".to_string(), moments(1000), + &nobody_listening(), ); assert_eq!( @@ -908,6 +1068,7 @@ mod tests { WhatIsAsserted::PlaybackPosition, "at 10".to_string(), moments(0), + &nobody_listening(), ); let what_it_did = queue.enqueue( @@ -915,6 +1076,7 @@ mod tests { WhatIsAsserted::PlaybackPosition, "at 90".to_string(), moments(20 * A_DAY), + &nobody_listening(), ); assert_eq!(what_it_did, WhatTheEnqueueDid::ReplacedInPlace); @@ -934,6 +1096,78 @@ mod tests { ); } + /// 0047's report, at the moment it happens: the kind of action and the + /// correlator for the target, at `failure` rather than `notice` because a + /// person's own action cannot be fetched again. + #[test] + fn a_drop_is_reported_as_it_happens_with_the_kind_and_a_correlator() { + let collector = Collector::default(); + let diagnostics = listening(&collector); + let mut queue: WriteQueue = WriteQueue::empty(); + + for target in 0..A_SESSIONS_QUEUE_HOLDS_AT_MOST { + queue.enqueue( + item(&format!("item-{target}")), + WhatIsAsserted::Watched, + "yes".to_string(), + moments(0), + &diagnostics, + ); + } + assert!( + collector.told().is_empty(), + "a queue under its bound reported something" + ); + + queue.enqueue( + item("one-too-many"), + WhatIsAsserted::Watched, + "yes".to_string(), + moments(0), + &diagnostics, + ); + + let told = collector.told(); + assert_eq!(told.len(), 1, "the drop was reported {} times", told.len()); + let (severity, name, fields) = &told[0]; + assert_eq!(*severity, Severity::Failure); + assert_eq!(*name, "write-queue.entry-dropped"); + assert_eq!(fields.len(), 2, "the event carried {fields:?}"); + assert_eq!(fields[1], ("asserted-about", "watched".to_string())); + } + + /// The half of that report 0047 states as a refusal: the correlator names + /// the target and the identifier reaches nobody. + #[test] + fn a_drop_report_carries_a_correlator_and_never_the_identifier() { + let collector = Collector::default(); + let diagnostics = listening(&collector); + let mut queue: WriteQueue = WriteQueue::empty(); + + for target in 0..=A_SESSIONS_QUEUE_HOLDS_AT_MOST { + queue.enqueue( + item(&format!("an-item-nobody-should-read-{target}")), + WhatIsAsserted::PlaybackPosition, + "at 10".to_string(), + moments(0), + &diagnostics, + ); + } + + let told = collector.told(); + assert_eq!(told.len(), 1); + let (name, value) = &told[0].2[0]; + assert_eq!(*name, "for-target"); + assert!( + !value.contains("an-item-nobody-should-read"), + "the identifier reached the sink as {value}" + ); + assert!( + !value.is_empty() && value.chars().all(|character| character.is_ascii_hexdigit()), + "the target left as something other than a correlator: {value}" + ); + } + /// 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. @@ -945,12 +1179,14 @@ mod tests { WhatIsAsserted::Watched, "yes".to_string(), moments(0), + &nobody_listening(), ); queue.enqueue( item("after-a-clock-that-moved"), WhatIsAsserted::Watched, "yes".to_string(), moments(500 * A_DAY), + &nobody_listening(), ); let now = WallMoment::from_epoch(366 * A_DAY, 0); diff --git a/src/session/mid_playback.rs b/src/session/mid_playback.rs index 2e6de27..62bf0e0 100644 --- a/src/session/mid_playback.rs +++ b/src/session/mid_playback.rs @@ -262,7 +262,8 @@ pub fn the_renewal_ended( ) -> WhatTheOutcomeDoesToPlayback { match renewals.ended(how) { WhatTheOutcomeDoes::RetryTheWaitingCallsOnce => { - let what_the_queue_did = reporting.report_after_a_renewal(current, enqueued_at, queue); + let what_the_queue_did = + reporting.report_after_a_renewal(current, enqueued_at, queue, diagnostics); WhatTheOutcomeDoesToPlayback::CurrentPositionReportedAndTheDrainResumes( what_the_queue_did, ) @@ -308,6 +309,8 @@ mod tests { #[derive(Debug, Default)] struct Still; + static STILL: Still = Still; + impl Clocks for Still { fn steady(&self) -> SteadyInstant { SteadyInstant::from_nanos(0) @@ -356,6 +359,15 @@ mod tests { } } + /// The facility with nobody listening, for the enqueues this fixture makes. + /// + /// 0047 reports a drop and this queue never reaches its bound, so nothing + /// here is a report anybody would receive. The cases that DO watch the sink + /// build their own with a collector. + fn nobody_listening() -> Diagnostics<'static> { + Diagnostics::new(&STILL, None, Severity::Detail, a_salt()) + } + fn a_salt() -> CorrelatorSalt { CorrelatorSalt::from_bytes([0x5a; CorrelatorSalt::WIDTH]) } @@ -401,7 +413,8 @@ mod tests { played_to(0), at(0), enqueued(), - &mut queue + &mut queue, + &nobody_listening(), ), WhatTheEnqueueDid::Added ); @@ -411,7 +424,8 @@ mod tests { played_to(10), at(1), enqueued(), - &mut queue + &mut queue, + &nobody_listening(), ), WhatTheEnqueueDid::ReplacedInPlace ); @@ -422,7 +436,8 @@ mod tests { played_to(0), at(2), enqueued(), - &mut queue + &mut queue, + &nobody_listening(), ), WhatTheEnqueueDid::Added ); diff --git a/tests/the_rule_as_data_names_every_field.rs b/tests/the_rule_as_data_names_every_field.rs index fbf74ef..7f97566 100644 --- a/tests/the_rule_as_data_names_every_field.rs +++ b/tests/the_rule_as_data_names_every_field.rs @@ -237,8 +237,8 @@ fn no_source_file_carries_more_than_one_test_module() { /// a reader of a bundle would meet, and it is a negative disclosure as much as a /// positive one: NO FIELD THIS BUILD EMITS IS EXCLUDED. 0071 puts the session /// token and anything derived from it under that treatment, and nothing in this -/// tree emits one yet, so a bundle assembled today says the reduced field is the -/// only one it holds a correlator for. +/// tree emits one yet, so a bundle assembled today says the two reduced fields +/// are the only ones it holds a correlator for. #[test] fn the_statement_is_the_set_this_build_actually_carries() { let stated = stated(); @@ -249,7 +249,7 @@ fn the_statement_is_the_set_this_build_actually_carries() { .map(|(name, _)| name.as_str()) .collect(); reduced.sort_unstable(); - assert_eq!(reduced, vec!["entry"]); + assert_eq!(reduced, vec!["entry", "for-target"]); let excluded: Vec<&str> = stated .iter() @@ -270,6 +270,7 @@ fn the_statement_is_the_set_this_build_actually_carries() { assert_eq!( whole, vec![ + "asserted-about", "check", "consecutive-refusals", "entry-kind",