From 540972595a9acd7dba56b51376ec339e75a6f834 Mon Sep 17 00:00:00 2001 From: Nils Lehnen <30603423+iderex@users.noreply.github.com> Date: Thu, 3 Sep 2026 09:56:42 +0200 Subject: [PATCH] Hold 0005's mid-playback sequence as what the queue and the session do (#35) 0005 fixes what happens when a token dies while something is playing: the stream is left alone, the report that was due is not lost and not retried at once, the queue is not touched by an authentication failure, one renewal is attempted, and each of its outcomes does one thing to the queue and tells the client one thing or nothing. 0034's counter, 0047's queue and 0057's report were all in the tree as values and nothing joined them, so a rejection during playback had no answer and #35's sequence existed only as a sentence. src/session/mid_playback.rs is the join. A rejected position report is answered off 0034's counter and holds the report where it is in every case, which the shared reference to the queue is what holds: the path cannot remove an entry. A renewal's outcome is read for playback: a fresh token reports the current position through the queue, so 0047's coalescing replaces the entry the rejection left at the head and the drain resumes from the same place with one report rather than two; a refusal keeps every report, answers the forced sign-out 0114 fixes and tells the client once, as an event carrying how many positions are held; a silence keeps every report and moves nothing. Neither answer carries a value of 0004's vocabulary, so nothing here can fail a call the client did not make. src/playback/report.rs gains the third occasion a report is made on, which is 0005's rather than 0057's, and the one call that makes it. The field name the event carries is registered in 0071's statement, and the test that holds that statement against the tree names it. Four guards were watched failing. Answering the current position without asking the queue reddens the success case. Emptying the queue on a refusal reddens the refusal case. Telling the client on a silence reddens two cases. Ending a session at the first rejection without telling the client reddens the no-route case. Nothing here sends a byte, runs a drain or holds a session, and the queue is not durable. #35's condition, a run through a token death against the fake server, has no subject in this tree and the issue stays open on it. Signed-off-by: Nils Lehnen <30603423+iderex@users.noreply.github.com> --- .github/coverage/pinned-surface | 1 + src/lib.rs | 2 + src/lifecycle/mod.rs | 2 + src/playback/report.rs | 45 +- src/session/mid_playback.rs | 706 ++++++++++++++++++++ src/session/mod.rs | 8 + tests/the_rule_as_data_names_every_field.rs | 1 + tests/thread_statements.rs | 7 + 8 files changed, 769 insertions(+), 3 deletions(-) create mode 100644 src/session/mid_playback.rs diff --git a/.github/coverage/pinned-surface b/.github/coverage/pinned-surface index 4711070..98c907a 100644 --- a/.github/coverage/pinned-surface +++ b/.github/coverage/pinned-surface @@ -74,6 +74,7 @@ module src/session/delegated.rs The value that ties a delegated sign-in atte module src/session/renewal.rs The generation a rejection is answered against and the moment a renewal is due. A defect here is twenty renewals where 0034 fixes one, or a session signed out because a network dropped rather than because a server refused it. module src/session/password.rs 0030's password: the one reading that spends it, the account name kept as it was typed, and the closure that refuses an answer missing one of the three facts 0005 says a session holds. A defect here is a credential readable twice or printable, or a session built around a token a server never sent. module src/session/quick_connect.rs 0031's cadence for asking about a Quick Connect exchange, its four endings, and which of the two values the server issued crosses to the client. A defect here is a backoff nobody decided on a route where the answer arrives from a person, a denial reported as a failure, or the value the core presents handed out beside the code. +module src/session/mid_playback.rs 0005's sequence for a token that dies mid-playback: what a rejected position report does, and what each renewal outcome does to the queue and to the one report the success branch makes. A defect here discards a person's positions because their token expired, sends the position reached during a renewal twice, or signs a session out because a connection dropped rather than because a server refused it. module src/session/sign_out.rs 0114's two acts: what each of them takes away, the order that puts the local half of a sign-out before the request to the server, how work in flight ends, and what a removal that could not be completed reports. A defect here leaves a token in memory on a device somebody handed over, empties a library on an ordinary evening act, or tells an operator their data is gone when it is not. module src/session/device.rs The device identity, whose identifier is one part of the cache key 0041 derives and one half of the key a server puts a live session under, so a defect here reaches two other surfaces on this list. module src/lifecycle/mod.rs 0115's creation, its capability answer, its stop bound and the lifetime a call is judged against. diff --git a/src/lib.rs b/src/lib.rs index 5e22fcd..6e51b03 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -147,6 +147,8 @@ const _: () = { any_thread::(); any_thread::(); any_thread::(); + any_thread::(); + any_thread::(); any_thread::>(); any_thread::>(); any_thread::(); diff --git a/src/lifecycle/mod.rs b/src/lifecycle/mod.rs index 8461a13..72ed851 100644 --- a/src/lifecycle/mod.rs +++ b/src/lifecycle/mod.rs @@ -60,6 +60,7 @@ use crate::diagnostics::DiagnosticsSink; use crate::diagnostics::redaction::FieldName; use crate::failure::Failure; use crate::session::SecretStore; +use crate::session::mid_playback; /// The two seconds a stop is bounded by where a client sets nothing. /// @@ -507,6 +508,7 @@ pub const fn every_field_name_the_core_emits() -> &'static [FieldName] { envelope::ENTRY_KIND, envelope::CHECK, envelope::VERSION_FOUND, + mid_playback::POSITIONS_HELD, ] } diff --git a/src/playback/report.rs b/src/playback/report.rs index a28e46a..31c3b18 100644 --- a/src/playback/report.rs +++ b/src/playback/report.rs @@ -72,9 +72,13 @@ use crate::server::write_queue::{Target, WhatIsAsserted, WhatTheEnqueueDid, Writ /// What occasioned a report. /// -/// Two occasions and no third. 0057 fixes that a report is made on each of the -/// five events the moment it happens, and on the interval while something is -/// playing, and it names nothing else that produces one. +/// Three occasions and no fourth. 0057 fixes that a report is made on each of +/// the five events the moment it happens, and on the interval while something +/// is playing, and it names nothing else that produces one. THE THIRD IS +/// 0005'S AND NOT 0057'S: the success branch of a renewal mid-playback reports +/// the current position, which is one more occasion for the same assertion +/// rather than a change to the cadence, and `crate::session::mid_playback` is +/// where it is made. /// /// It is carried on the report rather than discarded so that whatever delivers /// the report can tell a stop from a tick, which #10's table separates by path. @@ -88,6 +92,11 @@ pub enum ReportedOn { Event(ReportsWithoutWaiting), /// The interval said a report was due. TheInterval, + /// A renewal succeeded mid-playback, and 0005 says the current position is + /// reported then. It is the third occasion because 0005 adds it, in the + /// section that names #35, rather than 0057; what it asserts and how the + /// queue coalesces it are unchanged. + AfterARenewal, } /// What one report asserts: where playback of one item is. @@ -254,6 +263,36 @@ impl Reporting { self.interval = self.interval.after_a_report_at(now); WhatObservingDid::Reported(what_the_queue_did) } + + /// A renewal succeeded mid-playback: report where playback is now. + /// + /// 0005 says the success branch drains the queue in order and reports the + /// current position, and this is that report, through the queue like every + /// other. That is what makes it one report rather than two: the entry the + /// rejection left at the head is replaced in place by 0047's coalescing, + /// keeping its place in the order, so the drain resumes exactly where it + /// stopped and the position reached during the renewal arrives once. + /// `crate::session::mid_playback` is the caller, and it is the whole of + /// #35's sequence rather than this one call. + /// + /// The interval is left where it was, as it is for a seek: nothing about + /// 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. + pub fn report_after_a_renewal( + &mut self, + position: AdmittedPosition, + queue: &mut WriteQueue, + ) -> WhatTheEnqueueDid { + queue.enqueue( + self.target.clone(), + WhatIsAsserted::PlaybackPosition, + PositionReport { + position: position.position(), + reported_on: ReportedOn::AfterARenewal, + }, + ) + } } #[cfg(test)] diff --git a/src/session/mid_playback.rs b/src/session/mid_playback.rs new file mode 100644 index 0000000..973490d --- /dev/null +++ b/src/session/mid_playback.rs @@ -0,0 +1,706 @@ +//! A token that dies mid-playback: 0005's sequence, as what the queue and the +//! session do at each step of it. +//! +//! `docs/decisions/0005-the-session-model.md` fixes the sequence under its +//! mid-playback section and names #35 as where it is implemented. Three landed +//! records already hold the values it is made of: 0034 fixes what one renewal +//! is and what each of its outcomes does, which is [`super::renewal`]; 0047 +//! holds the queue every report is on, which is [`crate::server::write_queue`]; +//! and 0057 makes the report, which is [`crate::playback::report`]. This module +//! is the join of the three, and it decides nothing any of them decides. +//! +//! # The guarantee, and how a type holds it rather than a sentence +//! +//! 0005: playback already in flight is not interrupted. The stream is read by +//! the platform's player against the address #111 hands over, the core has no +//! way to stop it and would be wrong to use one, so a background report failing +//! reaches nothing a person is watching. A case cannot observe a stream the +//! core never holds, and the comment of 2026-08-11 on #35 says why a case that +//! asserted one would be a guard that cannot fail. What the core can be asked +//! about is its own output while the token dies, and each of the four things +//! 0005 promises about that is held by a type rather than by a rule somebody +//! remembers: +//! +//! - No cancellation is delivered and no outcome arrives on a call the client +//! did not make. [`WhatARejectedReportDoes`] and +//! [`WhatTheOutcomeDoesToPlayback`] carry no value of 0004's vocabulary, so +//! there is nothing here that could fail a caller with anything. +//! - Nothing already in the queue is discarded, because an authentication +//! failure says nothing about whether the positions are correct. +//! [`a_report_was_rejected`] takes the queue by shared reference and cannot +//! remove an entry from it. The rejected report was the head, it stays the +//! head, and the drain stops there, which is 0047's own rule for an entry +//! that could not be delivered. +//! - On success the queue drains in order and the current position is +//! reported. The report goes through the queue like every other, where 0047's +//! coalescing replaces the held one in place, so a renewal never sends one +//! report and queues a second for the same item. +//! - On failure the session is signed out, the queue is kept, and the client +//! is told once, through 0100, that this session can no longer report and +//! that positions are being held. Not as an error on any call, because it +//! made none. +//! +//! # What is here, and what is deliberately not +//! +//! WHAT IS NOT HERE IS THE DRAIN AND THE RENEWAL. Both are requests, the +//! rejection this module answers arrives from a server, and the transport is +//! #27 and is not built. Nothing here sends or receives a byte, nothing here +//! stops a drain because nothing runs one, and the queue this tree holds is not +//! durable, which `crate::server::write_queue` says of itself. So #35's +//! condition, a run through a token death against the fake server, has no +//! subject in this tree, and what is proven below is what the values answer at +//! each step of the sequence. +//! +//! WHAT IS ALSO NOT HERE IS THE SESSION. [`Renewals`] is the session's counter +//! and is handed in rather than held; the signed-out state is +//! [`LocalHalf`], which is answered rather than performed; and there is no +//! token in this tree to drop from memory. +//! +//! WHAT IS ALSO NOT HERE IS THE POSITION REACHED AFTER THE TOKEN DIED. 0005 +//! guarantees the last position observed before the rejection and nothing +//! after it, bounded by 0057's cadence, and says a core that claimed the exact +//! stopping point would be claiming something it never observed. The current +//! position handed to [`the_renewal_ended`] is the client's reading at that +//! moment, which is what 0057 makes every position, and this module claims +//! nothing about the gap. + +use crate::diagnostics::redaction::FieldName; +use crate::diagnostics::{Diagnostics, EventName, Field, FieldValue, Severity}; +use crate::playback::AdmittedPosition; +use crate::playback::report::{PositionReport, Reporting}; +use crate::server::write_queue::{WhatIsAsserted, WhatTheEnqueueDid, WriteQueue}; +use crate::session::renewal::{ + HowTheRenewalEnded, Rejection, Renewals, WhatARejectedCallDoes, WhatTheOutcomeDoes, +}; +use crate::session::sign_out::{LocalHalf, WhySigningOut}; + +/// The event 0005 owes the client at the moment a renewal fails during +/// playback: this session can no longer report, and positions are being held. +/// +/// At `failure` rather than `notice`, for the reason 0105 gives about a dropped +/// queue entry: the thing that did not happen is somebody's own position +/// reaching the server, and a client reporting what was lost reads this level. +/// It is the one thing the client is told, and it is told through the sink +/// rather than on a call, because it made none. +const REPORTING_SUSPENDED: EventName = EventName::declared("session.reporting-suspended"); + +/// How many position reports the queue is holding for the session at that +/// moment. +/// +/// Carried whole: it is a count, which 0071 lists among the values that cannot +/// differ between two people running the same build against the same server. +/// The item each position is about is not on the event at all, in either +/// treatment, because 0047 reports a queue's contents by correlator and this +/// event is about the session rather than about any one entry. +pub(crate) const POSITIONS_HELD: FieldName = FieldName::carried_whole("positions-held"); + +/// What a position report the server rejected does. +/// +/// Five answers and never nothing, one per answer 0034 gives a rejected call, +/// read for a report rather than for a call a client made. Every one of them +/// holds the report where it is: the queue is not touched on this path, which +/// the signature of [`a_report_was_rejected`] is what holds. +/// +/// Thread safety, from 0009: a plain value, safe from any thread. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum WhatARejectedReportDoes { + /// The report stays at the head, the drain stops there, and the session's + /// one renewal starts. This is the case 0005's sequence is written for. + HeldAndTheRenewalStarts, + /// The report stays at the head and a renewal against this same token is + /// already running. The drain waits for its outcome, and starts nothing. + HeldWhileTheRenewalAlreadyRuns, + /// The report went out under a token the session has since replaced. The + /// drain delivers the head again, once, under the token the session holds + /// now, which is the one retry 0034 allows a rejected call. + DeliveredAgainUnderTheCurrentToken, + /// The report was already the retry and was rejected in turn. It stays at + /// the head, the drain stops, and no second renewal is started, for 0034's + /// reason: a token issued seconds ago and immediately refused is not a + /// token a third one would fix. + HeldAndNothingStarts, + /// The server offers no renewal route, so the first rejection ends the + /// session. Every report is kept, the client is told through the sink, and + /// this is the local half of the sign-out 0114 fixes. + HeldAndTheSessionSignsOut(LocalHalf), +} + +/// What a renewal's outcome does to playback. +/// +/// Three answers, one per outcome 0034 fixes, each read for what it does to +/// the queue and to the one report 0005 says the success branch makes. +/// +/// Thread safety, from 0009: a plain value, safe from any thread. It is not +/// `Copy` because the queue's answer may carry what it dropped, and that holds +/// the dropped entry's target. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WhatTheOutcomeDoesToPlayback { + /// A fresh token came back. The current position was reported through the + /// queue, this is what the queue did with it, and the drain resumes from + /// the head in order. Nothing is told to the client, because nothing + /// happened that a person needs to know about. + CurrentPositionReportedAndTheDrainResumes(WhatTheEnqueueDid), + /// The server refused the renewal. Every report is kept, the session is + /// signed out, and the client was told once through the sink. The core + /// stops attempting to report and does not retry on a schedule, because + /// every attempt would fail for the same reason; what restarts reporting + /// is a person signing in again to the same account on the same server. + HeldAndTheSessionSignsOut(LocalHalf), + /// Nothing answered the renewal, or nothing was running. The session is + /// exactly as it was, the token is not discarded, every report is kept, + /// and nothing is told to anybody: the core learned nothing about the + /// token, and the drain resumes when 0045 next reports the server + /// reachable. + HeldAndTheSessionLeftAsItWas, +} + +/// The number of position reports the queue holds. +fn positions_held(queue: &WriteQueue) -> u64 { + let held = queue + .entries() + .iter() + .filter(|entry| entry.asserted_about() == WhatIsAsserted::PlaybackPosition) + .count(); + u64::try_from(held).unwrap_or(u64::MAX) +} + +/// Tells the client, once, that this session can no longer report. +fn say_positions_are_held(queue: &WriteQueue, diagnostics: &Diagnostics<'_>) { + diagnostics.emit( + Severity::Failure, + REPORTING_SUSPENDED, + &[Field::new( + POSITIONS_HELD, + FieldValue::Count(positions_held(queue)), + )], + ); +} + +/// The sign-out a renewal that produced no token forces. +/// +/// Forced rather than asked for, so it is always the plain act and never a +/// forget, which is [`LocalHalf::completed`]'s own reading of +/// [`WhySigningOut::ARenewalWasRefused`]. A session with a renewal running or a +/// rejection arriving is a session that was signed in, so the caller's reading +/// of it is that it was not already signed out. +/// +/// 0114 names one reason for a forced sign-out and 0034 names two cases that +/// force one, a renewal refused and a renewal that was not possible. Both take +/// the same reason here, because 0114's answer is the same for both and a +/// second variant would carry a distinction nothing downstream reads. +const fn forced_sign_out() -> LocalHalf { + LocalHalf::completed(WhySigningOut::ARenewalWasRefused, false) +} + +/// Answers a position report the server rejected with a token presented. +/// +/// This is 0034's [`Renewals::rejected`] read for the head of the queue rather +/// than for a call a client made, and the difference is the whole of what this +/// function adds: a call fails with a kind from 0004 and a report does not, +/// because nobody is holding a call for it to fail. The report stays where it +/// is in every case, which the shared reference to the queue is what holds. +/// The drain stops at it, because 0047 stops at the first entry that could not +/// be delivered, and what happens next is the renewal's outcome arriving at +/// [`the_renewal_ended`]. +/// +/// The one case that reaches the client is a server with no renewal route, +/// where 0034 ends the session at the first rejection. That is 0005's failure +/// branch arriving without a renewal having been attempted, and it is told the +/// same way. +pub fn a_report_was_rejected( + renewals: &mut Renewals, + rejection: Rejection, + queue: &WriteQueue, + diagnostics: &Diagnostics<'_>, +) -> WhatARejectedReportDoes { + match renewals.rejected(rejection) { + WhatARejectedCallDoes::StartTheRenewal => WhatARejectedReportDoes::HeldAndTheRenewalStarts, + WhatARejectedCallDoes::WaitForTheRenewalAlreadyRunning => { + WhatARejectedReportDoes::HeldWhileTheRenewalAlreadyRuns + } + WhatARejectedCallDoes::RetryAgainstTheCurrentToken => { + WhatARejectedReportDoes::DeliveredAgainUnderTheCurrentToken + } + WhatARejectedCallDoes::FailAndStartNothing => WhatARejectedReportDoes::HeldAndNothingStarts, + WhatARejectedCallDoes::SignTheSessionOut => { + say_positions_are_held(queue, diagnostics); + WhatARejectedReportDoes::HeldAndTheSessionSignsOut(forced_sign_out()) + } + } +} + +/// Applies the outcome of the renewal a rejection during playback started. +/// +/// 0034's [`Renewals::ended`] decides what the outcome does to the session, +/// and this reads each answer for what it does to playback. A fresh token +/// makes the one report 0005's success branch owes, the position now, through +/// the queue: the entry the rejection left at the head is replaced in place, +/// and the drain then resumes from that head in order, so the position a +/// person reached during the renewal arrives at the server once. A refusal +/// keeps every report, signs the session out and tells the client once. A +/// silence keeps every report and moves nothing. +/// +/// 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. +pub fn the_renewal_ended( + renewals: &mut Renewals, + how: HowTheRenewalEnded, + reporting: &mut Reporting, + current: AdmittedPosition, + queue: &mut WriteQueue, + diagnostics: &Diagnostics<'_>, +) -> WhatTheOutcomeDoesToPlayback { + match renewals.ended(how) { + WhatTheOutcomeDoes::RetryTheWaitingCallsOnce => { + let what_the_queue_did = reporting.report_after_a_renewal(current, queue); + WhatTheOutcomeDoesToPlayback::CurrentPositionReportedAndTheDrainResumes( + what_the_queue_did, + ) + } + WhatTheOutcomeDoes::SignTheSessionOut => { + say_positions_are_held(queue, diagnostics); + WhatTheOutcomeDoesToPlayback::HeldAndTheSessionSignsOut(forced_sign_out()) + } + WhatTheOutcomeDoes::LeaveTheSessionExactlyAsItWas => { + WhatTheOutcomeDoesToPlayback::HeldAndTheSessionLeftAsItWas + } + } +} + +#[cfg(test)] +mod tests { + //! 0005's mid-playback sequence, asked of the values at each step. + //! + //! What these cannot ask is #35's condition: a run through a token death + //! against the fake server, with the stream, the drain and the renewal + //! request all in flight. Nothing here sends a byte, and the queue is not + //! durable. + + use std::sync::Mutex; + + use super::{ + POSITIONS_HELD, WhatARejectedReportDoes, WhatTheOutcomeDoesToPlayback, + a_report_was_rejected, the_renewal_ended, + }; + use crate::clock::{Clocks, ElapsedInstant, SteadyInstant, WallMoment}; + use crate::diagnostics::redaction::CorrelatorSalt; + use crate::diagnostics::{Diagnostics, DiagnosticsSink, Event, FieldValue, Severity}; + use crate::playback::cadence::ReportsWithoutWaiting; + use crate::playback::report::{PositionReport, ReportedOn, Reporting}; + use crate::playback::{AdmittedPosition, Ticks}; + use crate::server::write_queue::{Target, WhatTheEnqueueDid, WriteQueue}; + use crate::session::renewal::{HowTheRenewalEnded, Rejection, RenewalRoute, Renewals}; + use crate::session::sign_out::{LocalHalf, WhySigningOut}; + + /// A clock source that does not move. Nothing here reads a clock for a + /// decision; the facility needs one for the moment it stamps on an event. + #[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) + } + } + + /// Keeps every event's name and the count it carried under + /// `positions-held`, so a case can say how many times the client was told + /// and what it was told. + #[derive(Debug, Default)] + struct Collector { + told: Mutex)>>, + } + + impl Collector { + fn told(&self) -> Vec<(&'static str, Option)> { + self.told + .lock() + .expect("the fixture holds no poisoned lock") + .clone() + } + } + + impl DiagnosticsSink for Collector { + fn event(&self, event: &Event<'_>) { + let held = event + .fields() + .iter() + .find(|field| field.name() == POSITIONS_HELD) + .and_then(|field| match field.value() { + FieldValue::Count(count) => Some(count), + _ => None, + }); + self.told + .lock() + .expect("the fixture holds no poisoned lock") + .push((event.name().as_str(), held)); + } + } + + fn a_salt() -> CorrelatorSalt { + CorrelatorSalt::from_bytes([0x5a; CorrelatorSalt::WIDTH]) + } + + fn at(seconds: u64) -> ElapsedInstant { + ElapsedInstant::from_nanos(seconds * 1_000_000_000) + } + + fn played_to(seconds: i64) -> AdmittedPosition { + AdmittedPosition::of(Ticks::from_seconds(seconds).as_ticks(), None) + } + + fn item(identifier: &str) -> Target { + Target::item(identifier.to_string()) + } + + /// What a queue holds, in order, as something a case can compare before + /// and after: the order number, the item, and the position. + fn contents(queue: &WriteQueue) -> Vec<(u64, String, Ticks)> { + queue + .entries() + .iter() + .map(|entry| { + ( + entry.order(), + entry.target().as_str().to_string(), + entry.assertion().position(), + ) + }) + .collect() + } + + /// The queue 0005's sequence is written against: the film at the head, + /// with the report the rejection is about, and an episode behind it. Two + /// items, two positions, and the film's reporting handed back so the + /// success branch can report through it. + fn a_queue_with_the_film_at_the_head() -> (WriteQueue, Reporting) { + let mut queue = WriteQueue::empty(); + let mut film = Reporting::for_item(item("the-film"), at(0)); + assert_eq!( + film.report( + ReportsWithoutWaiting::Started, + played_to(0), + at(0), + &mut queue + ), + WhatTheEnqueueDid::Added + ); + assert_eq!( + film.report( + ReportsWithoutWaiting::Seeked, + played_to(10), + at(1), + &mut queue + ), + WhatTheEnqueueDid::ReplacedInPlace + ); + let mut episode = Reporting::for_item(item("the-episode"), at(2)); + assert_eq!( + episode.report( + ReportsWithoutWaiting::Started, + played_to(0), + at(2), + &mut queue + ), + WhatTheEnqueueDid::Added + ); + assert_eq!(queue.len(), 2); + (queue, film) + } + + fn rejected_under(renewals: &Renewals) -> Rejection { + Rejection { + went_out_under: renewals.generation(), + is_the_retry: false, + } + } + + /// The head of the queue was rejected under the token the session holds. + fn reject_the_head( + renewals: &mut Renewals, + queue: &WriteQueue, + diagnostics: &Diagnostics<'_>, + ) -> WhatARejectedReportDoes { + let rejection = rejected_under(renewals); + a_report_was_rejected(renewals, rejection, queue, diagnostics) + } + + /// 0005's first step. The report that was due when the rejection arrived + /// is not lost and is not retried immediately: it is at the head, it stays + /// at the head, nothing behind it moves, the renewal starts, and the + /// client hears nothing. + #[test] + fn a_rejected_report_stays_at_the_head_and_the_renewal_starts() { + let clocks = Still; + let collector = Collector::default(); + let diagnostics = Diagnostics::new(&clocks, Some(&collector), Severity::Detail, a_salt()); + let (queue, _) = a_queue_with_the_film_at_the_head(); + let before = contents(&queue); + let mut renewals = Renewals::acquired(RenewalRoute::Offered); + + let did = reject_the_head(&mut renewals, &queue, &diagnostics); + + assert_eq!(did, WhatARejectedReportDoes::HeldAndTheRenewalStarts); + assert_eq!(contents(&queue), before); + assert_eq!(queue.dropped(), 0); + assert_eq!(renewals.running_against(), Some(renewals.generation())); + assert!(collector.told().is_empty(), "the client was told something"); + } + + /// The nineteen of twenty, for a report: a second rejection under the same + /// token joins the renewal already running and holds. + #[test] + fn a_second_rejection_under_the_same_token_holds_and_waits() { + let clocks = Still; + let diagnostics = Diagnostics::new(&clocks, None, Severity::Detail, a_salt()); + let (queue, _) = a_queue_with_the_film_at_the_head(); + let before = contents(&queue); + let mut renewals = Renewals::acquired(RenewalRoute::Offered); + let first = reject_the_head(&mut renewals, &queue, &diagnostics); + assert_eq!(first, WhatARejectedReportDoes::HeldAndTheRenewalStarts); + + let second = reject_the_head(&mut renewals, &queue, &diagnostics); + + assert_eq!( + second, + WhatARejectedReportDoes::HeldWhileTheRenewalAlreadyRuns + ); + assert_eq!(contents(&queue), before); + } + + /// A report that went out under a token the session has since replaced + /// takes its one retry, and a retry rejected in turn holds and starts no + /// second renewal. + #[test] + fn a_report_under_a_replaced_token_is_delivered_again_once_and_no_more() { + let clocks = Still; + let collector = Collector::default(); + let diagnostics = Diagnostics::new(&clocks, Some(&collector), Severity::Detail, a_salt()); + let (mut queue, mut film) = a_queue_with_the_film_at_the_head(); + let mut renewals = Renewals::acquired(RenewalRoute::Offered); + let old = renewals.generation(); + reject_the_head(&mut renewals, &queue, &diagnostics); + the_renewal_ended( + &mut renewals, + HowTheRenewalEnded::AFreshToken, + &mut film, + played_to(12), + &mut queue, + &diagnostics, + ); + assert_ne!(renewals.generation(), old); + let before = contents(&queue); + + let again = a_report_was_rejected( + &mut renewals, + Rejection { + went_out_under: old, + is_the_retry: false, + }, + &queue, + &diagnostics, + ); + let the_retry = Rejection { + went_out_under: renewals.generation(), + is_the_retry: true, + }; + let and_again = a_report_was_rejected(&mut renewals, the_retry, &queue, &diagnostics); + + assert_eq!( + again, + WhatARejectedReportDoes::DeliveredAgainUnderTheCurrentToken + ); + assert_eq!(and_again, WhatARejectedReportDoes::HeldAndNothingStarts); + assert_eq!(renewals.running_against(), None); + assert_eq!(contents(&queue), before); + assert!(collector.told().is_empty(), "the client was told something"); + } + + /// 0034 ends the session at the first rejection where the server offers no + /// renewal route. That is 0005's failure branch without a renewal having + /// been attempted: every report is kept, the sign-out is the forced one, + /// and the client is told once, with how many positions are held. + #[test] + fn a_server_with_no_renewal_route_signs_out_at_the_first_rejection_and_holds_every_report() { + let clocks = Still; + let collector = Collector::default(); + let diagnostics = Diagnostics::new(&clocks, Some(&collector), Severity::Detail, a_salt()); + let (queue, _) = a_queue_with_the_film_at_the_head(); + let before = contents(&queue); + let mut renewals = Renewals::acquired(RenewalRoute::NotOffered); + + let did = reject_the_head(&mut renewals, &queue, &diagnostics); + + assert_eq!( + did, + WhatARejectedReportDoes::HeldAndTheSessionSignsOut(LocalHalf::completed( + WhySigningOut::ARenewalWasRefused, + false + )) + ); + assert_eq!(contents(&queue), before); + assert_eq!( + collector.told(), + vec![("session.reporting-suspended", Some(2))] + ); + } + + /// 0005's success branch. The current position is reported through the + /// queue, so 0047's coalescing replaces the entry the rejection left at the + /// head rather than adding a second one for the same item; the head keeps + /// its place in the order, so the drain resumes exactly where it stopped; + /// and the client hears nothing, because nothing happened that a person + /// needs to know about. + #[test] + fn a_fresh_token_reports_the_current_position_in_place_and_the_drain_resumes() { + let clocks = Still; + let collector = Collector::default(); + let diagnostics = Diagnostics::new(&clocks, Some(&collector), Severity::Detail, a_salt()); + let (mut queue, mut film) = a_queue_with_the_film_at_the_head(); + let before = contents(&queue); + let mut renewals = Renewals::acquired(RenewalRoute::Offered); + let old = renewals.generation(); + reject_the_head(&mut renewals, &queue, &diagnostics); + + let did = the_renewal_ended( + &mut renewals, + HowTheRenewalEnded::AFreshToken, + &mut film, + played_to(25), + &mut queue, + &diagnostics, + ); + + assert_eq!( + did, + WhatTheOutcomeDoesToPlayback::CurrentPositionReportedAndTheDrainResumes( + WhatTheEnqueueDid::ReplacedInPlace + ) + ); + let after = contents(&queue); + assert_eq!(after.len(), before.len()); + assert_eq!( + after[0].0, before[0].0, + "the head lost its place in the order" + ); + assert_eq!(after[0].1, "the-film"); + assert_eq!(after[0].2, Ticks::from_seconds(25)); + assert_eq!(after[1], before[1]); + let head = queue.next_to_deliver().expect("the queue holds the film"); + assert_eq!(head.assertion().reported_on(), ReportedOn::AfterARenewal); + assert_ne!(renewals.generation(), old); + assert_eq!(renewals.running_against(), None); + assert!(collector.told().is_empty(), "the client was told something"); + } + + /// 0005's failure branch. The queue is kept rather than dropped, because it + /// belongs to the session and a person who signs in again gets those + /// positions reported in order before anything else; the session is signed + /// out by the forced act; and the client is told exactly once. + #[test] + fn a_refused_renewal_keeps_every_report_and_says_so_once() { + let clocks = Still; + let collector = Collector::default(); + let diagnostics = Diagnostics::new(&clocks, Some(&collector), Severity::Detail, a_salt()); + let (mut queue, mut film) = a_queue_with_the_film_at_the_head(); + let before = contents(&queue); + let mut renewals = Renewals::acquired(RenewalRoute::Offered); + reject_the_head(&mut renewals, &queue, &diagnostics); + + let did = the_renewal_ended( + &mut renewals, + HowTheRenewalEnded::TheServerRefusedIt, + &mut film, + played_to(25), + &mut queue, + &diagnostics, + ); + + assert_eq!( + did, + WhatTheOutcomeDoesToPlayback::HeldAndTheSessionSignsOut(LocalHalf::completed( + WhySigningOut::ARenewalWasRefused, + false + )) + ); + assert_eq!(contents(&queue), before); + assert_eq!(queue.dropped(), 0); + assert_eq!(renewals.running_against(), None); + assert_eq!( + collector.told(), + vec![("session.reporting-suspended", Some(2))] + ); + } + + /// 0034's silence. The core learned nothing about the token, so the + /// session is exactly as it was, the generation has not moved, every + /// report is kept, and the client is told nothing: a session emptied + /// because a connection dropped is the failure that record names. + #[test] + fn a_silent_renewal_leaves_the_session_and_the_queue_exactly_as_they_were() { + let clocks = Still; + let collector = Collector::default(); + let diagnostics = Diagnostics::new(&clocks, Some(&collector), Severity::Detail, a_salt()); + let (mut queue, mut film) = a_queue_with_the_film_at_the_head(); + let before = contents(&queue); + let mut renewals = Renewals::acquired(RenewalRoute::Offered); + reject_the_head(&mut renewals, &queue, &diagnostics); + let generation = renewals.generation(); + + let did = the_renewal_ended( + &mut renewals, + HowTheRenewalEnded::NothingAnswered, + &mut film, + played_to(25), + &mut queue, + &diagnostics, + ); + + assert_eq!( + did, + WhatTheOutcomeDoesToPlayback::HeldAndTheSessionLeftAsItWas + ); + assert_eq!(contents(&queue), before); + assert_eq!(renewals.generation(), generation); + assert_eq!(renewals.running_against(), None); + assert!(collector.told().is_empty(), "the client was told something"); + } + + /// An outcome arriving when no renewal was running is 0034's safe reading + /// of a case it does not describe: nothing moves, nothing is reported, and + /// no position is put on the queue for a renewal that did not happen. + #[test] + fn an_outcome_when_nothing_was_running_changes_nothing() { + let clocks = Still; + let collector = Collector::default(); + let diagnostics = Diagnostics::new(&clocks, Some(&collector), Severity::Detail, a_salt()); + let (mut queue, mut film) = a_queue_with_the_film_at_the_head(); + let before = contents(&queue); + let mut renewals = Renewals::acquired(RenewalRoute::Offered); + + let did = the_renewal_ended( + &mut renewals, + HowTheRenewalEnded::AFreshToken, + &mut film, + played_to(25), + &mut queue, + &diagnostics, + ); + + assert_eq!( + did, + WhatTheOutcomeDoesToPlayback::HeldAndTheSessionLeftAsItWas + ); + assert_eq!(contents(&queue), before); + assert!(collector.told().is_empty(), "the client was told something"); + } +} diff --git a/src/session/mod.rs b/src/session/mod.rs index e7d349c..cbfb55e 100644 --- a/src/session/mod.rs +++ b/src/session/mod.rs @@ -42,9 +42,17 @@ //! session's one renewal and which joins it, what each renewal outcome does, and //! when a renewal ahead of a rejection is due. It performs no renewal and holds //! no session, so what it answers with is what the session then does. +//! +//! [`mid_playback`] holds the part of 0005's mid-playback sequence the values +//! above settle together with the queue in 0047 and the report in 0057: what a +//! position report the server rejected does, and what each renewal outcome then +//! does to the queue and to the one report the success branch makes. It holds +//! no session and runs no drain, for the same reason [`renewal`] performs no +//! renewal. pub mod delegated; pub mod device; +pub mod mid_playback; pub mod password; pub mod quick_connect; pub mod renewal; diff --git a/tests/the_rule_as_data_names_every_field.rs b/tests/the_rule_as_data_names_every_field.rs index cb11536..fbf74ef 100644 --- a/tests/the_rule_as_data_names_every_field.rs +++ b/tests/the_rule_as_data_names_every_field.rs @@ -274,6 +274,7 @@ fn the_statement_is_the_set_this_build_actually_carries() { "consecutive-refusals", "entry-kind", "for-tier", + "positions-held", "released-bytes", "released-entries", "suspended-for", diff --git a/tests/thread_statements.rs b/tests/thread_statements.rs index 3277f59..006f817 100644 --- a/tests/thread_statements.rs +++ b/tests/thread_statements.rs @@ -84,6 +84,7 @@ use flowfin_core::session::delegated::{ NoAttemptMatched, OpenAttempts, Relayable, TieValue, ValueAlreadyOpen, ValueNotUsable, }; use flowfin_core::session::device::{Capabilities, DeviceIdentity, PartNotUsable}; +use flowfin_core::session::mid_playback::{WhatARejectedReportDoes, WhatTheOutcomeDoesToPlayback}; use flowfin_core::session::password::{ AccountName, AnswerRead, FactNotCarried, FactsASessionNeeds, NoSession, Password, }; @@ -568,6 +569,12 @@ fn how_a_renewal_ended_is_safe_from_any_thread() { const _: () = any_thread::(); } +#[test] +fn what_a_token_that_dies_mid_playback_does_is_safe_from_any_thread() { + const _: () = any_thread::(); + const _: () = any_thread::(); +} + #[test] fn what_a_renewal_outcome_does_is_safe_from_any_thread() { const _: () = any_thread::();