From 0f84674957733b1f2f34a1d54c83dec8dcaf8b2c Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Sat, 3 Oct 2026 20:19:44 -0400 Subject: [PATCH 1/2] Core: the Account owns its requests and responses; the reducer only carries what follows The Account's session, pending request and status are private; it answers each account intent and response itself and returns an AccountChange. The listening ledger sends and settles its own push. Request ids come from one RequestIds source in core state. apply_account_change is the one place the write order is decided: listening before account, now for sign-out and 401 too, pinned by a new account test. ADR 0001 decision 6 records the order. Closes #69 Co-Authored-By: Claude Opus 5.5 --- crates/cascade-core/src/account.rs | 283 ++++++++++++++++++++++--- crates/cascade-core/src/listening.rs | 178 +++++++++++----- crates/cascade-core/src/server.rs | 27 ++- crates/cascade-core/src/snapshot.rs | 15 +- crates/cascade-core/src/state.rs | 298 ++++++++------------------- crates/cascade-core/tests/account.rs | 41 ++++ docs/adr/0001-listening-time-crdt.md | 2 +- 7 files changed, 526 insertions(+), 318 deletions(-) diff --git a/crates/cascade-core/src/account.rs b/crates/cascade-core/src/account.rs index 6497203..ec2c74f 100644 --- a/crates/cascade-core/src/account.rs +++ b/crates/cascade-core/src/account.rs @@ -5,9 +5,17 @@ //! request may start, what the user is told, and what success, failure and a //! 401 do. A shell only carries the requests it is handed (each one generic //! request, settled with one generic response) and stores the blob. +//! +//! The [`Account`] answers each account intent and each of its responses +//! itself, and tells the reducer only what follows for listening and storage, +//! as one [`AccountChange`]. use serde::{Deserialize, Serialize}; +use crate::effect::Effect; +use crate::server::{parse_verified, Request, RequestIds, StatusClass}; +use crate::snapshot::AccountSnapshot; + /// Storage-schema version for the persisted account blob. pub const ACCOUNT_VERSION: u32 = 1; @@ -22,14 +30,14 @@ pub struct Session { /// The one account request the shell is carrying, if any, and the id its /// response will carry. Only one runs at a time. #[derive(Debug, Clone, PartialEq, Eq)] -pub struct Pending { - pub id: u64, - pub request: PendingRequest, +struct Pending { + id: u64, + request: PendingRequest, } /// What a pending account request is for, and what its success needs. #[derive(Debug, Clone, PartialEq, Eq)] -pub enum PendingRequest { +enum PendingRequest { SignInLink { email: String, }, @@ -46,7 +54,7 @@ pub enum PendingRequest { /// What the user was last told about their account. [`Self::label`] holds the /// copy, so every shell shows the same words. #[derive(Debug, Clone, PartialEq, Eq)] -pub enum AccountStatus { +enum AccountStatus { EnterEmail, LinkSent { email: String }, LinkFailed, @@ -62,7 +70,7 @@ pub enum AccountStatus { } impl AccountStatus { - pub fn label(&self) -> String { + fn label(&self) -> String { match self { Self::EnterEmail => "Enter your email address.".into(), Self::LinkSent { email } => format!("Check {email} for a sign-in link."), @@ -80,33 +88,57 @@ impl AccountStatus { } } -/// The account as the core holds it. Held inside [`crate::State`]. +/// What an account transition means for listening and storage. The reducer +/// carries it through to the listening ledger and the persisted blobs; an +/// outcome that only changes what the user is told is not a change. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum AccountChange { + /// A session was verified. `replaced` when it signed in over another + /// account, whose cross-device total must not show. + SignedIn { replaced: bool }, + /// The session is gone: signed out, or a 401. + SignedOut, + /// The server deleted this device's listening data: rotate the slot. + ListeningDeleted { new_device_id: String }, + /// The server deleted the account: rotate the slot, and the session is + /// gone. + AccountDeleted { new_device_id: String }, +} + +/// The account as the core holds it: the session, the one pending request, +/// and what the user was last told. Held inside [`crate::State`]; every +/// account rule lives here. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct Account { - pub session: Option, - pub pending: Option, - pub status: Option, + session: Option, + pending: Option, + status: Option, } impl Account { - /// An account request is out; every account command but sign-out waits. - pub fn busy(&self) -> bool { - self.pending.is_some() - } - - /// Take the pending request if response `id` settles it. A response for - /// anything else (nothing pending, or a dropped request) changes nothing. - pub fn settle(&mut self, id: u64) -> Option { - if self.pending.as_ref()?.id != id { - return None; - } - self.pending.take().map(|p| p.request) + /// Take the session from a stored blob: the versioned shape, or the + /// version-less one web and Apple stored before the core held the + /// account. Empty, garbage, an unknown version or an empty token mean no + /// account. + pub fn restore(&mut self, json: &str) { + self.session = session_from_json(json); } pub fn session_token(&self) -> Option { self.session.as_ref().map(|s| s.session_token.clone()) } + /// The account view for the UI. Never carries the session token. + pub fn snapshot(&self) -> AccountSnapshot { + let email = self.session.as_ref().map(|s| s.email.clone()); + AccountSnapshot { + signed_in_label: email.as_ref().map(|e| format!("Syncing · {e}")), + email, + status_label: self.status.as_ref().map(AccountStatus::label), + busy: self.busy(), + } + } + /// The blob for [`crate::Effect::PersistAccount`]: empty when signed out, /// meaning "delete the stored account". pub fn to_json(&self) -> String { @@ -123,19 +155,202 @@ impl Account { .unwrap_or_default() } - /// Read a stored blob: the versioned shape, or the version-less one web - /// and Apple stored before the core held the account. Empty, garbage, an - /// unknown version or an empty token mean no account. - pub fn session_from_json(json: &str) -> Option { - let stored: PersistedAccount = serde_json::from_str(json).ok()?; - if stored.version.is_some_and(|v| v != ACCOUNT_VERSION) || stored.session_token.is_empty() { - return None; + /// Ask the server to email a sign-in link to `email`, trimmed. A blank + /// email asks for one instead. + pub fn request_sign_in_link( + &mut self, + email: &str, + platform: Option<&str>, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + if self.busy() { + return; + } + let email = email.trim().to_string(); + if email.is_empty() { + self.status = Some(AccountStatus::EnterEmail); + return; + } + let request = Request::sign_in_link(&email, platform); + self.begin( + PendingRequest::SignInLink { email }, + None, + request, + ids, + effects, + ); + } + + /// Redeem the sign-in token found in what the user pasted. + pub fn submit_sign_in_link( + &mut self, + input: &str, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + if self.busy() { + return; + } + match parse_sign_in_token(input) { + Some(token) => self.begin( + PendingRequest::Verify, + Some(AccountStatus::SigningIn), + Request::verify(&token), + ids, + effects, + ), + None => self.status = Some(AccountStatus::PasteFullLink), + } + } + + /// Ask the server to delete this account's listening data. The slot + /// rotates to `new_device_id` only once the server confirms. + pub fn delete_listening( + &mut self, + new_device_id: String, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + if self.busy() { + return; + } + if let Some(session_token) = self.session_token() { + self.begin( + PendingRequest::DeleteListening { new_device_id }, + None, + Request::delete_listening(session_token), + ids, + effects, + ); + } + } + + /// Ask the server to delete the account. The slot rotates to + /// `new_device_id` only once the server confirms. + pub fn delete_account( + &mut self, + new_device_id: String, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + if self.busy() { + return; + } + if let Some(session_token) = self.session_token() { + self.begin( + PendingRequest::DeleteAccount { new_device_id }, + None, + Request::delete_account(session_token), + ids, + effects, + ); } - Some(Session { - session_token: stored.session_token, - email: stored.email, - }) } + + /// Sign out, cutting in on any pending request. Revokes the session if + /// there is one; nothing waits for the revoke's response. + pub fn sign_out( + &mut self, + ids: &mut RequestIds, + effects: &mut Vec, + ) -> Option { + self.pending = None; + self.status = None; + let session = self.session.take()?; + ids.send(Request::revoke(session.session_token), effects); + Some(AccountChange::SignedOut) + } + + /// Whether the pending request is request `id`. + pub fn awaits(&self, id: u64) -> bool { + self.pending.as_ref().is_some_and(|p| p.id == id) + } + + /// Settle the pending request with its response. The caller has already + /// matched the response's id with [`Self::awaits`]. A 401 while signed in + /// expires the session; any other failure only changes what the user is + /// told. + pub fn settle(&mut self, class: StatusClass, body: &str) -> Option { + let request = self.pending.take()?.request; + if class == StatusClass::Unauthorized && self.session.is_some() { + return Some(self.expire()); + } + let succeeded = class == StatusClass::Success; + let (status, change) = match request { + PendingRequest::SignInLink { email } if succeeded => { + (AccountStatus::LinkSent { email }, None) + } + PendingRequest::Verify if succeeded => match parse_verified(body) { + Some(session) => { + let status = AccountStatus::SignedIn { + email: session.email.clone(), + }; + let replaced = self.session.replace(session).is_some(); + (status, Some(AccountChange::SignedIn { replaced })) + } + None => (AccountStatus::LinkInvalid, None), + }, + PendingRequest::DeleteListening { new_device_id } if succeeded => ( + AccountStatus::ListeningDeleted, + Some(AccountChange::ListeningDeleted { new_device_id }), + ), + PendingRequest::DeleteAccount { new_device_id } if succeeded => { + self.session = None; + ( + AccountStatus::AccountDeleted, + Some(AccountChange::AccountDeleted { new_device_id }), + ) + } + PendingRequest::SignInLink { .. } => (AccountStatus::LinkFailed, None), + PendingRequest::Verify => (AccountStatus::LinkInvalid, None), + PendingRequest::DeleteListening { .. } => (AccountStatus::ListeningDeleteFailed, None), + PendingRequest::DeleteAccount { .. } => (AccountStatus::AccountDeleteFailed, None), + }; + self.status = Some(status); + change + } + + /// The server rejected the session (a 401 on any request, including a + /// listening push): sign out and tell the user to sign in again. No + /// revoke, since the session is already gone server-side. + pub fn expire(&mut self) -> AccountChange { + self.pending = None; + self.status = Some(AccountStatus::SessionExpired); + self.session = None; + AccountChange::SignedOut + } + + /// An account request is out; every account intent but sign-out waits. + fn busy(&self) -> bool { + self.pending.is_some() + } + + /// Send the one account request, remembering what it is for. + fn begin( + &mut self, + request: PendingRequest, + status: Option, + sent: Request, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + let id = ids.send(sent, effects); + self.pending = Some(Pending { id, request }); + self.status = status; + } +} + +/// Read a stored account blob into a session; see [`Account::restore`]. +fn session_from_json(json: &str) -> Option { + let stored: PersistedAccount = serde_json::from_str(json).ok()?; + if stored.version.is_some_and(|v| v != ACCOUNT_VERSION) || stored.session_token.is_empty() { + return None; + } + Some(Session { + session_token: stored.session_token, + email: stored.email, + }) } /// The persisted account blob. `version` is optional on read only, so the @@ -151,7 +366,7 @@ pub struct PersistedAccount { /// Find the sign-in token in what the user pasted: a link carrying a query /// parameter named exactly `token`, or the bare token itself. -pub fn parse_sign_in_token(input: &str) -> Option { +fn parse_sign_in_token(input: &str) -> Option { let input = input.trim(); if let Some(value) = token_query_value(input) { let token = percent_decode(value); diff --git a/crates/cascade-core/src/listening.rs b/crates/cascade-core/src/listening.rs index 43d3a12..d044a3e 100644 --- a/crates/cascade-core/src/listening.rs +++ b/crates/cascade-core/src/listening.rs @@ -17,6 +17,9 @@ use serde::{Deserialize, Serialize}; +use crate::effect::Effect; +use crate::server::{parse_server_total, Request, RequestIds, StatusClass}; + /// Storage-schema version for the persisted listening blob. Separate from /// `SETTINGS_VERSION` because listening data lives in its own blob /// (`cascade.listening.v1`), distinct from user settings. @@ -104,6 +107,19 @@ impl InFlight { } } +/// How the PUT in flight settled. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Settled { + /// The server accepted it: the ledger moved forward and needs persisting. + Synced, + /// Not sent, no response, or any failure but a 401. + Failed, + /// The server rejected the session. + Unauthorized, + /// It was superseded, so its response was dropped. + Dropped, +} + impl Default for ListeningLedger { fn default() -> Self { Self { @@ -159,79 +175,95 @@ impl ListeningLedger { self.device_total_ms = self.device_total_ms.saturating_add(delta); } - /// Decide whether to sync for a shell's `reason`, and if so what to send: - /// `(device_id, device_total_ms)`. Marks that total as in flight as - /// request `id`, so a second call returns `None` until - /// [`Self::sync_succeeded`] or [`Self::sync_failed`] settles it — the one - /// re-entrancy guard every shell shares. - pub fn begin_sync(&mut self, reason: SyncReason, id: u64) -> Option<(String, u64)> { + /// Sync for a shell's `reason` if there is anything to say: push this + /// slot's total under `session_token`, with a fresh id from `ids`, and + /// mark it in flight, so nothing more is pushed until [`Self::settle`] + /// settles it — the one re-entrancy guard every shell shares. + pub fn begin_sync( + &mut self, + reason: SyncReason, + session_token: String, + ids: &mut RequestIds, + effects: &mut Vec, + ) { let worth_sending = match reason { SyncReason::Flush => self.unsynced_ms() > 0, SyncReason::Refresh => true, }; - self.begin_sync_if(worth_sending, id) + self.begin_sync_if(worth_sending, session_token, ids, effects); } /// The core's own routine sync, checked on every tick that counts - /// listening: send only once at least [`LISTENING_SYNC_THRESHOLD_MS`] is - /// unsynced, counted past any failed PUT. Same result and in-flight guard + /// listening: push only once at least [`LISTENING_SYNC_THRESHOLD_MS`] is + /// unsynced, counted past any failed PUT. Same push and in-flight guard /// as [`Self::begin_sync`]. - pub fn begin_threshold_sync(&mut self, id: u64) -> Option<(String, u64)> { + pub fn begin_threshold_sync( + &mut self, + session_token: String, + ids: &mut RequestIds, + effects: &mut Vec, + ) { let worth_sending = self.unsent_ms() >= LISTENING_SYNC_THRESHOLD_MS; - self.begin_sync_if(worth_sending, id) - } - - fn begin_sync_if(&mut self, worth_sending: bool, id: u64) -> Option<(String, u64)> { - if self.in_flight.is_some() { - return None; - } - let device_id = self.device_id.clone()?; - if !worth_sending { - return None; + self.begin_sync_if(worth_sending, session_token, ids, effects); + } + + fn begin_sync_if( + &mut self, + worth_sending: bool, + session_token: String, + ids: &mut RequestIds, + effects: &mut Vec, + ) { + if self.in_flight.is_some() || !worth_sending { + return; } + let Some(device_id) = self.device_id.as_deref() else { + return; + }; + let request = Request::push_listening(device_id, self.device_total_ms, session_token); + let id = ids.send(request, effects); self.in_flight = Some(InFlight::Sent { id, device_total_ms: self.device_total_ms, }); - Some((device_id, self.device_total_ms)) - } - - /// The in-flight PUT succeeded. The server has accepted exactly what was - /// sent — not whatever accrued while the request was out — and reports - /// `server_total_ms` as the cross-device aggregate. This only moves the - /// display baseline forward; it never touches `device_total_ms`, and is - /// monotonic so an out-of-order ack can't rewind the high-water mark. - /// Returns `false` (and changes nothing else) if no sync for the current - /// slot was in flight. The caller has already matched the response's id - /// to [`Self::in_flight`]. - pub fn sync_succeeded(&mut self, server_total_ms: u64) -> bool { + } + + /// Settle the PUT in flight with its response. The caller has already + /// matched the response's id with [`Self::awaits`]. + /// + /// On success the server has accepted exactly what was sent — not + /// whatever accrued while the request was out — and reports the + /// cross-device aggregate. That only moves the display baseline forward; + /// it never touches `device_total_ms`, and is monotonic so an + /// out-of-order ack can't rewind the high-water mark. On failure nothing + /// was acknowledged: a flush or refresh may try again at once, and a + /// threshold sync once another threshold has accrued past what was sent. + /// A 2xx whose body does not parse is a failure. A superseded PUT only + /// frees the slot: its outcome refers to a slot or session already + /// replaced. + pub fn settle(&mut self, class: StatusClass, body: &str) -> Settled { let Some(InFlight::Sent { device_total_ms: sent, .. }) = self.in_flight.take() else { - return false; + return Settled::Dropped; + }; + let server_total_ms = match class { + StatusClass::Success => parse_server_total(body), + StatusClass::Unauthorized | StatusClass::Failed => None, + }; + let Some(server_total_ms) = server_total_ms else { + self.failed_through_ms = Some(sent); + return match class { + StatusClass::Unauthorized => Settled::Unauthorized, + StatusClass::Success | StatusClass::Failed => Settled::Failed, + }; }; self.synced_through_ms = self.synced_through_ms.max(sent); self.server_total_ms = Some(server_total_ms); self.failed_through_ms = None; - true - } - - /// The in-flight PUT failed. Nothing was acknowledged; a flush or refresh - /// may try again at once, and a threshold sync once another threshold has - /// accrued past what was sent. Returns `false` if no sync for the current - /// slot was in flight, so a superseded request's failure means nothing. - pub fn sync_failed(&mut self) -> bool { - let Some(InFlight::Sent { - device_total_ms: sent, - .. - }) = self.in_flight.take() - else { - return false; - }; - self.failed_through_ms = Some(sent); - true + Settled::Synced } /// Take `fallback` as this device's id if it doesn't have one yet. @@ -351,12 +383,20 @@ pub fn format_listening_total(total_ms: u64) -> String { mod tests { use super::*; + /// Start a sync for `reason`, returning the effects it pushed. + fn begin(l: &mut ListeningLedger, reason: SyncReason, ids: &mut RequestIds) -> Vec { + let mut effects = Vec::new(); + l.begin_sync(reason, "token".into(), ids, &mut effects); + effects + } + /// Send whatever is unsynced and have the server accept it. fn sync(l: &mut ListeningLedger, server_total_ms: u64) { l.device_id.get_or_insert_with(|| "device".into()); - l.begin_sync(SyncReason::Refresh, 1) - .expect("nothing in flight"); - assert!(l.sync_succeeded(server_total_ms)); + let pushed = begin(l, SyncReason::Refresh, &mut RequestIds::default()); + assert_eq!(pushed.len(), 1, "nothing in flight"); + let body = format!(r#"{{"serverTotalMs":{server_total_ms}}}"#); + assert_eq!(l.settle(StatusClass::Success, &body), Settled::Synced); } #[test] @@ -424,16 +464,42 @@ mod tests { l.accrue(4_000); sync(&mut l, 9_000); l.accrue(1_000); - l.begin_sync(SyncReason::Flush, 2) - .expect("nothing in flight"); + let pushed = begin(&mut l, SyncReason::Flush, &mut RequestIds::default()); + assert_eq!(pushed.len(), 1, "nothing in flight"); l.forget_server(); assert_eq!(l.server_total_ms, None); - assert_eq!(l.in_flight, Some(InFlight::Superseded { id: 2 })); + assert_eq!(l.in_flight, Some(InFlight::Superseded { id: 1 })); assert_eq!(l.device_total_ms, 5_000); assert_eq!(l.synced_through_ms, 4_000); assert_eq!(l.displayed_total_ms(), 5_000); } + #[test] + fn settle_names_each_outcome() { + let mut ids = RequestIds::default(); + let mut l = ListeningLedger { + device_id: Some("device".into()), + ..ListeningLedger::default() + }; + l.accrue(1_000); + for (class, body, outcome) in [ + (StatusClass::Failed, "", Settled::Failed), + (StatusClass::Success, "not json", Settled::Failed), + (StatusClass::Unauthorized, "", Settled::Unauthorized), + ] { + begin(&mut l, SyncReason::Refresh, &mut ids); + assert_eq!(l.settle(class, body), outcome, "{class:?} {body}"); + assert_eq!(l.failed_through_ms, Some(1_000)); + assert_eq!(l.in_flight, None); + } + begin(&mut l, SyncReason::Refresh, &mut ids); + l.forget_server(); + let ok = r#"{"serverTotalMs":9000}"#; + assert_eq!(l.settle(StatusClass::Success, ok), Settled::Dropped); + assert_eq!(l.server_total_ms, None); + assert_eq!(l.in_flight, None); + } + #[test] fn restore_never_regresses_the_lifetime_total() { let mut live = ListeningLedger::default(); diff --git a/crates/cascade-core/src/server.rs b/crates/cascade-core/src/server.rs index 01e31ff..8fad101 100644 --- a/crates/cascade-core/src/server.rs +++ b/crates/cascade-core/src/server.rs @@ -109,7 +109,7 @@ impl Request { } /// The effect that asks the shell to carry this request as `id`. - pub fn into_effect(self, id: u64) -> Effect { + fn into_effect(self, id: u64) -> Effect { Effect::ServerRequest { id, method: self.method, @@ -120,6 +120,31 @@ impl Request { } } +/// The one source of request ids, held in [`crate::State`] and shared by every +/// module that sends. Starts at 1, is never persisted (a restart forgets every +/// request in flight along with the shell process that was carrying it), and +/// never reuses an id. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RequestIds { + next: u64, +} + +impl Default for RequestIds { + fn default() -> Self { + Self { next: 1 } + } +} + +impl RequestIds { + /// Hand the shell `request` under a fresh id, and return that id. + pub fn send(&mut self, request: Request, effects: &mut Vec) -> u64 { + let id = self.next; + self.next += 1; + effects.push(request.into_effect(id)); + id + } +} + fn to_json(body: &impl Serialize) -> Option { serde_json::to_string(body).ok() } diff --git a/crates/cascade-core/src/snapshot.rs b/crates/cascade-core/src/snapshot.rs index af8c8bf..82959d9 100644 --- a/crates/cascade-core/src/snapshot.rs +++ b/crates/cascade-core/src/snapshot.rs @@ -114,19 +114,6 @@ pub struct AccountSnapshot { pub busy: bool, } -impl AccountSnapshot { - fn from_state(state: &State) -> Self { - let account = &state.account; - let email = account.session.as_ref().map(|s| s.email.clone()); - Self { - signed_in_label: email.as_ref().map(|e| format!("Syncing · {e}")), - email, - status_label: account.status.as_ref().map(|s| s.label()), - busy: account.busy(), - } - } -} - #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct Snapshot { @@ -238,7 +225,7 @@ impl Snapshot { unsynced_ms: state.listening.unsynced_ms(), total_label: format_listening_total(state.listening.displayed_total_ms()), }, - account: AccountSnapshot::from_state(state), + account: state.account.snapshot(), tick_interval_ms: tick_interval_ms(state), } } diff --git a/crates/cascade-core/src/state.rs b/crates/cascade-core/src/state.rs index 762e944..5cb2e6b 100644 --- a/crates/cascade-core/src/state.rs +++ b/crates/cascade-core/src/state.rs @@ -6,13 +6,11 @@ use serde::{Deserialize, Serialize}; -use crate::account::{ - parse_sign_in_token, Account, AccountStatus, Pending, PendingRequest, Session, -}; +use crate::account::{Account, AccountChange}; use crate::command::Command; use crate::effect::Effect; -use crate::listening::{ListeningLedger, PersistedListening, SyncReason}; -use crate::server::{parse_server_total, parse_verified, Request, StatusClass}; +use crate::listening::{ListeningLedger, PersistedListening, Settled, SyncReason}; +use crate::server::{RequestIds, StatusClass}; use crate::settings::{PersistedSettings, SETTINGS_VERSION}; use crate::timer::{clamp_minutes, ActiveTimer, TimerKind, DEFAULT_TIMER_MINUTES}; @@ -66,10 +64,8 @@ pub struct State { /// The optional sign-in. Restored from its own blob via /// [`Command::Restore`]'s `account_json`. pub account: Account, - /// The id the next [`Effect::ServerRequest`] gets. Starts at 1 and only - /// grows; never persisted, since a restart forgets every request in - /// flight along with the shell process that was carrying it. - next_request_id: u64, + /// Where every [`Effect::ServerRequest`] gets its id. + request_ids: RequestIds, /// Whether [`Command::Restore`] has already run. Only the first restore /// takes effect, so a replayed stale blob can't undo a later reset. restored: bool, @@ -89,7 +85,7 @@ impl Default for State { audio_confirmed_playing: false, listening: ListeningLedger::default(), account: Account::default(), - next_request_id: 1, + request_ids: RequestIds::default(), restored: false, } } @@ -159,57 +155,48 @@ fn push_persist_account(state: &State, effects: &mut Vec) { }); } -/// Hand the shell `request` under a fresh id, and return that id. -fn send(state: &mut State, request: Request, effects: &mut Vec) -> u64 { - let id = state.next_request_id; - state.next_request_id += 1; - effects.push(request.into_effect(id)); - id -} - -/// Start a listening sync if `begin` finds anything to send and the account -/// allows it, answering with one push. `begin` is handed the id the push -/// will carry, so the ledger can remember it. -fn begin_listening_sync( - state: &mut State, - effects: &mut Vec, - begin: impl FnOnce(&mut ListeningLedger, u64) -> Option<(String, u64)>, -) { - let Some(session_token) = state.account.session_token() else { - return; - }; - let id = state.next_request_id; - if let Some((device_id, device_total_ms)) = begin(&mut state.listening, id) { - let request = Request::push_listening(&device_id, device_total_ms, session_token); - let sent = send(state, request, effects); - debug_assert_eq!(sent, id, "the push must carry the id the ledger remembers"); +/// Carry an account transition through to listening and storage. The one +/// place the order of those writes is decided: within one update the +/// listening blob is written before the account blob, so a crash between the +/// two never leaves a deleted account's slot with no session to answer for it +/// (ADR 0001, decision 6). +fn apply_account_change(state: &mut State, change: AccountChange, effects: &mut Vec) { + match change { + AccountChange::SignedIn { replaced } => { + // Signing in over another account: nothing of its total may show. + if replaced { + state.listening.forget_server(); + push_persist_listening(state, effects); + } + push_persist_account(state, effects); + begin_listening_sync(state, effects, SyncReason::Refresh); + } + AccountChange::SignedOut => { + state.listening.forget_server(); + push_persist_listening(state, effects); + push_persist_account(state, effects); + } + AccountChange::ListeningDeleted { new_device_id } => { + state.listening.reset(new_device_id); + push_persist_listening(state, effects); + } + AccountChange::AccountDeleted { new_device_id } => { + // The reset forgets the server total and supersedes any in-flight + // sync too, so the slot and the sign-out share one listening write. + state.listening.reset(new_device_id); + push_persist_listening(state, effects); + push_persist_account(state, effects); + } } } -/// Start the one account request, answering with `request`. The caller has -/// already checked that none is pending. -fn begin_account_request( - state: &mut State, - pending: PendingRequest, - status: Option, - request: Request, - effects: &mut Vec, -) { - let id = send(state, request, effects); - state.account.pending = Some(Pending { - id, - request: pending, - }); - state.account.status = status; -} - -/// Drop the session and everything learned through it: the stored blob, the -/// cross-device total, and any sync still in flight for it. -fn forget_session(state: &mut State, effects: &mut Vec) { - state.account.session = None; - push_persist_account(state, effects); - state.listening.forget_server(); - push_persist_listening(state, effects); +/// Offer a listening sync for `reason` if signed in. +fn begin_listening_sync(state: &mut State, effects: &mut Vec, reason: SyncReason) { + if let Some(session_token) = state.account.session_token() { + state + .listening + .begin_sync(reason, session_token, &mut state.request_ids, effects); + } } /// Whether a tick should count as listening: tracking on, audio *confirmed* @@ -332,7 +319,13 @@ pub fn reduce(state: &mut State, command: Command, effects: &mut Vec) { // it is also the one place the routine sync is offered. if was_accruing { push_persist_listening(state, effects); - begin_listening_sync(state, effects, ListeningLedger::begin_threshold_sync); + if let Some(session_token) = state.account.session_token() { + state.listening.begin_threshold_sync( + session_token, + &mut state.request_ids, + effects, + ); + } } } Command::PlatformPlaybackStarted => { @@ -385,10 +378,11 @@ pub fn reduce(state: &mut State, command: Command, effects: &mut Vec) { // Persist a newly adopted id right away, so the next launch sees // the same slot rather than another fallback. let mut persist_listening = state.listening.adopt_device_id(fallback_device_id); - state.account.session = Account::session_from_json(&account_json); + state.account.restore(&account_json); // A cross-device total with no account behind it is left over // from a sign-out before the core held the account. - if state.account.session.is_none() && state.listening.server_total_ms.is_some() { + if state.account.session_token().is_none() && state.listening.server_total_ms.is_some() + { state.listening.forget_server(); persist_listening = true; } @@ -396,178 +390,58 @@ pub fn reduce(state: &mut State, command: Command, effects: &mut Vec) { push_persist_listening(state, effects); } } - Command::BeginListeningSync { reason } => { - begin_listening_sync(state, effects, |l, id| l.begin_sync(reason, id)) - } + Command::BeginListeningSync { reason } => begin_listening_sync(state, effects, reason), Command::ResetListeningData { new_device_id } => { state.listening.reset(new_device_id); push_persist_listening(state, effects); } - // Every account request waits for the pending one; only sign-out - // cuts in. - Command::RequestSignInLink { .. } - | Command::SubmitSignInLink { .. } - | Command::DeleteListeningData { .. } - | Command::DeleteAccount { .. } - if state.account.busy() => {} - Command::RequestSignInLink { email, platform } => { - let email = email.trim().to_string(); - if email.is_empty() { - state.account.status = Some(AccountStatus::EnterEmail); - return; - } - let request = Request::sign_in_link(&email, platform.as_deref()); - begin_account_request( - state, - PendingRequest::SignInLink { email }, - None, - request, - effects, - ); + Command::RequestSignInLink { email, platform } => state.account.request_sign_in_link( + &email, + platform.as_deref(), + &mut state.request_ids, + effects, + ), + Command::SubmitSignInLink { input } => { + state + .account + .submit_sign_in_link(&input, &mut state.request_ids, effects) } - Command::SubmitSignInLink { input } => match parse_sign_in_token(&input) { - Some(token) => begin_account_request( - state, - PendingRequest::Verify, - Some(AccountStatus::SigningIn), - Request::verify(&token), - effects, - ), - None => state.account.status = Some(AccountStatus::PasteFullLink), - }, Command::DeleteListeningData { new_device_id } => { - if let Some(session_token) = state.account.session_token() { - begin_account_request( - state, - PendingRequest::DeleteListening { new_device_id }, - None, - Request::delete_listening(session_token), - effects, - ); - } + state + .account + .delete_listening(new_device_id, &mut state.request_ids, effects) } Command::DeleteAccount { new_device_id } => { - if let Some(session_token) = state.account.session_token() { - begin_account_request( - state, - PendingRequest::DeleteAccount { new_device_id }, - None, - Request::delete_account(session_token), - effects, - ); - } + state + .account + .delete_account(new_device_id, &mut state.request_ids, effects) } Command::SignOut => { - state.account.pending = None; - state.account.status = None; - if let Some(session_token) = state.account.session_token() { - // Remembered nowhere, so its response is ignored. - send(state, Request::revoke(session_token), effects); - forget_session(state, effects); + if let Some(change) = state.account.sign_out(&mut state.request_ids, effects) { + apply_account_change(state, change, effects); } } Command::ServerResponse { id, status, body } => { let class = StatusClass::of(status); - if let Some(request) = state.account.settle(id) { - account_response(state, request, class, &body, effects); + if state.account.awaits(id) { + if let Some(change) = state.account.settle(class, &body) { + apply_account_change(state, change, effects); + } } else if state.listening.awaits(id) { - sync_response(state, class, &body, effects); + match state.listening.settle(class, &body) { + Settled::Synced => push_persist_listening(state, effects), + Settled::Unauthorized => { + let change = state.account.expire(); + apply_account_change(state, change, effects); + } + Settled::Failed | Settled::Dropped => {} + } } } } } -/// Settle the in-flight listening push. A superseded push only frees the -/// slot: its success or 401 refers to a slot or session already replaced. -fn sync_response(state: &mut State, class: StatusClass, body: &str, effects: &mut Vec) { - let server_total_ms = match class { - StatusClass::Success => parse_server_total(body), - StatusClass::Unauthorized | StatusClass::Failed => None, - }; - if let Some(server_total_ms) = server_total_ms { - if state.listening.sync_succeeded(server_total_ms) { - push_persist_listening(state, effects); - } - } else if state.listening.sync_failed() && class == StatusClass::Unauthorized { - session_expired(state, effects); - } -} - -/// Settle the pending account `request`, already taken off the account. -fn account_response( - state: &mut State, - request: PendingRequest, - class: StatusClass, - body: &str, - effects: &mut Vec, -) { - if class == StatusClass::Unauthorized && state.account.session.is_some() { - session_expired(state, effects); - return; - } - let succeeded = class == StatusClass::Success; - match request { - PendingRequest::SignInLink { email } if succeeded => { - state.account.status = Some(AccountStatus::LinkSent { email }); - } - PendingRequest::Verify if succeeded => match parse_verified(body) { - Some(session) => signed_in(state, session, effects), - None => state.account.status = Some(AccountStatus::LinkInvalid), - }, - PendingRequest::DeleteListening { new_device_id } if succeeded => { - state.account.status = Some(AccountStatus::ListeningDeleted); - state.listening.reset(new_device_id); - push_persist_listening(state, effects); - } - PendingRequest::DeleteAccount { new_device_id } if succeeded => { - state.account.status = Some(AccountStatus::AccountDeleted); - // The reset forgets the server total and supersedes any in-flight - // sync too, so the slot and the sign-out share one listening write. - state.listening.reset(new_device_id); - push_persist_listening(state, effects); - state.account.session = None; - push_persist_account(state, effects); - } - PendingRequest::SignInLink { .. } => { - state.account.status = Some(AccountStatus::LinkFailed); - } - PendingRequest::Verify => state.account.status = Some(AccountStatus::LinkInvalid), - PendingRequest::DeleteListening { .. } => { - state.account.status = Some(AccountStatus::ListeningDeleteFailed); - } - PendingRequest::DeleteAccount { .. } => { - state.account.status = Some(AccountStatus::AccountDeleteFailed); - } - } -} - -/// The sign-in token was redeemed for a session: persist the account and -/// start a refresh sync for the cross-device total. -fn signed_in(state: &mut State, session: Session, effects: &mut Vec) { - // Signing in over another account: nothing of its total may show. - if state.account.session.is_some() { - state.listening.forget_server(); - push_persist_listening(state, effects); - } - state.account.status = Some(AccountStatus::SignedIn { - email: session.email.clone(), - }); - state.account.session = Some(session); - push_persist_account(state, effects); - begin_listening_sync(state, effects, |l, id| { - l.begin_sync(SyncReason::Refresh, id) - }); -} - -/// The server rejected the session (a 401 on any request): sign out and tell -/// the user to sign in again. Listening carries on locally. -fn session_expired(state: &mut State, effects: &mut Vec) { - state.account.pending = None; - state.account.status = Some(AccountStatus::SessionExpired); - forget_session(state, effects); -} - fn start_playback(state: &mut State, effects: &mut Vec) { if state.intent.is_playing() { return; diff --git a/crates/cascade-core/tests/account.rs b/crates/cascade-core/tests/account.rs index 73570a9..abd84f0 100644 --- a/crates/cascade-core/tests/account.rs +++ b/crates/cascade-core/tests/account.rs @@ -1008,6 +1008,47 @@ fn a_401_while_signed_out_is_an_ordinary_failure() { ); } +// ---- write order ----------------------------------------------------------- +// +// Within one update the listening blob is written before the account blob, so +// a crash between the two never leaves a deleted account's slot with no +// session to answer for it (ADR 0001, decision 6). + +/// The persist effects in `effects`, in order, by blob. +fn persist_order(effects: &[Effect]) -> Vec<&'static str> { + effects + .iter() + .filter_map(|e| match e { + Effect::PersistListening { .. } => Some("listening"), + Effect::PersistAccount { .. } => Some("account"), + _ => None, + }) + .collect() +} + +#[test] +fn an_account_delete_writes_listening_before_the_account() { + let mut core = signed_in(); + listen(&mut core, 2_000); + let id = start(&mut core, delete_account()); + let effects = core.dispatch(bare_response(id, 204)).effects; + assert_eq!(persist_order(&effects), ["listening", "account"]); +} + +#[test] +fn a_401_writes_listening_before_the_account() { + let mut core = signed_in(); + listen(&mut core, 2_000); + let id = request_id(&begin(&mut core, SyncReason::Refresh)); + let effects = core.dispatch(bare_response(id, 401)).effects; + assert_eq!(persist_order(&effects), ["listening", "account"], "sync"); + + let mut core = signed_in(); + let id = start(&mut core, delete_listening()); + let effects = core.dispatch(bare_response(id, 401)).effects; + assert_eq!(persist_order(&effects), ["listening", "account"], "request"); +} + // ---- wire shape ------------------------------------------------------------ // // Every shell hand-writes these JSON shapes. Lock them. diff --git a/docs/adr/0001-listening-time-crdt.md b/docs/adr/0001-listening-time-crdt.md index b7b088e..bf77599 100644 --- a/docs/adr/0001-listening-time-crdt.md +++ b/docs/adr/0001-listening-time-crdt.md @@ -69,7 +69,7 @@ The table and its routing are tested through `Core::dispatch` in `crates/cascade Auth is email magic-link (no passwords) with **opaque server-side session tokens** (not JWT), so logout / delete-account revoke instantly with one `DELETE`. Tokens are stored only as SHA-256 hashes. "Delete my data" rotates the client's `device_id`, closing the one loophole inherent to grow-only counters: a forgotten offline device can't later resurrect a deleted total by pushing a stale higher counter — it lands in a fresh slot. -*Amended 2026-09-27 (issue #18).* The core owns `device_id`, inside the persisted listening blob. `ResetListeningData { newDeviceId }` zeroes the slot and rotates the id in the same state change, so both reach disk in one `PersistListening` write: a crash can no longer leave a fresh id holding the old total (which the next sync would write into a new server slot — exactly the resurrection this decision exists to prevent). The server cannot enforce this — deletes are by `user_id` only — so the client-side atomicity is the whole guarantee. The core has no randomness, so shells supply ids: a fresh UUID on reset, and `restore`'s `fallbackDeviceId` that the core adopts only when the listening blob has none (shells pass the id they used to store themselves; the core takes it only on the first launch after upgrading, so existing server slots carry over). +*Amended 2026-09-27 (issue #18).* The core owns `device_id`, inside the persisted listening blob. `ResetListeningData { newDeviceId }` zeroes the slot and rotates the id in the same state change, so both reach disk in one `PersistListening` write: a crash can no longer leave a fresh id holding the old total (which the next sync would write into a new server slot — exactly the resurrection this decision exists to prevent). The server cannot enforce this — deletes are by `user_id` only — so the client-side atomicity is the whole guarantee. Within one update, the listening blob is written before the account blob, so a crash between the two writes never leaves a deleted account's slot with no session to answer for it. The core has no randomness, so shells supply ids: a fresh UUID on reset, and `restore`'s `fallbackDeviceId` that the core adopts only when the listening blob has none (shells pass the id they used to store themselves; the core takes it only on the first launch after upgrading, so existing server slots carry over). ## Consequences From 0db916e273fc42e8080488cc63cdfaba4d76c5b8 Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Sat, 3 Oct 2026 20:21:56 -0400 Subject: [PATCH 2/2] Address review: one session guard for every sync, clearer names Co-Authored-By: Claude Opus 5.5 --- crates/cascade-core/src/account.rs | 13 ++++++---- crates/cascade-core/src/listening.rs | 3 ++- crates/cascade-core/src/state.rs | 36 +++++++++++++++++----------- 3 files changed, 32 insertions(+), 20 deletions(-) diff --git a/crates/cascade-core/src/account.rs b/crates/cascade-core/src/account.rs index ec2c74f..2bdea37 100644 --- a/crates/cascade-core/src/account.rs +++ b/crates/cascade-core/src/account.rs @@ -326,17 +326,20 @@ impl Account { self.pending.is_some() } - /// Send the one account request, remembering what it is for. + /// Send the one account request, remembering its `purpose`. fn begin( &mut self, - request: PendingRequest, + purpose: PendingRequest, status: Option, - sent: Request, + request: Request, ids: &mut RequestIds, effects: &mut Vec, ) { - let id = ids.send(sent, effects); - self.pending = Some(Pending { id, request }); + let id = ids.send(request, effects); + self.pending = Some(Pending { + id, + request: purpose, + }); self.status = status; } } diff --git a/crates/cascade-core/src/listening.rs b/crates/cascade-core/src/listening.rs index d044a3e..e37e207 100644 --- a/crates/cascade-core/src/listening.rs +++ b/crates/cascade-core/src/listening.rs @@ -116,7 +116,8 @@ pub enum Settled { Failed, /// The server rejected the session. Unauthorized, - /// It was superseded, so its response was dropped. + /// It was superseded (or nothing was in flight), so its response was + /// dropped. Dropped, } diff --git a/crates/cascade-core/src/state.rs b/crates/cascade-core/src/state.rs index 5cb2e6b..04ec37a 100644 --- a/crates/cascade-core/src/state.rs +++ b/crates/cascade-core/src/state.rs @@ -169,7 +169,9 @@ fn apply_account_change(state: &mut State, change: AccountChange, effects: &mut push_persist_listening(state, effects); } push_persist_account(state, effects); - begin_listening_sync(state, effects, SyncReason::Refresh); + begin_listening_sync(state, effects, |l, token, ids, effects| { + l.begin_sync(SyncReason::Refresh, token, ids, effects) + }); } AccountChange::SignedOut => { state.listening.forget_server(); @@ -190,12 +192,20 @@ fn apply_account_change(state: &mut State, change: AccountChange, effects: &mut } } -/// Offer a listening sync for `reason` if signed in. -fn begin_listening_sync(state: &mut State, effects: &mut Vec, reason: SyncReason) { +/// Offer a listening sync if signed in: `begin` is handed the session token +/// and the id source, and the ledger decides whether to push. +fn begin_listening_sync( + state: &mut State, + effects: &mut Vec, + begin: impl FnOnce(&mut ListeningLedger, String, &mut RequestIds, &mut Vec), +) { if let Some(session_token) = state.account.session_token() { - state - .listening - .begin_sync(reason, session_token, &mut state.request_ids, effects); + begin( + &mut state.listening, + session_token, + &mut state.request_ids, + effects, + ); } } @@ -319,13 +329,7 @@ pub fn reduce(state: &mut State, command: Command, effects: &mut Vec) { // it is also the one place the routine sync is offered. if was_accruing { push_persist_listening(state, effects); - if let Some(session_token) = state.account.session_token() { - state.listening.begin_threshold_sync( - session_token, - &mut state.request_ids, - effects, - ); - } + begin_listening_sync(state, effects, ListeningLedger::begin_threshold_sync); } } Command::PlatformPlaybackStarted => { @@ -390,7 +394,11 @@ pub fn reduce(state: &mut State, command: Command, effects: &mut Vec) { push_persist_listening(state, effects); } } - Command::BeginListeningSync { reason } => begin_listening_sync(state, effects, reason), + Command::BeginListeningSync { reason } => { + begin_listening_sync(state, effects, |l, token, ids, effects| { + l.begin_sync(reason, token, ids, effects) + }) + } Command::ResetListeningData { new_device_id } => { state.listening.reset(new_device_id); push_persist_listening(state, effects);