diff --git a/docs/rfcs/RFC-010-external-scrobbling.md b/docs/rfcs/RFC-010-external-scrobbling.md index 67d1868d..9e99ecfa 100644 --- a/docs/rfcs/RFC-010-external-scrobbling.md +++ b/docs/rfcs/RFC-010-external-scrobbling.md @@ -18,6 +18,12 @@ un serveur que nul n'a encore ouvert dans un navigateur ; répondre d'une écoute incertaine reste une décision que son auteur prend, et n'existe que sur la surface HTTP. + Puis [#199](https://github.com/InstaZDLL/waveflow-server/pull/199), **le + correctif de la décision 13** : le joker se compte sur `retried_at`, une + colonne que l'entrée porte, et non sur la survie de la reprise qui l'a + consommé — que la rétention emporte. Les quatre lecteurs qui lisaient la + jointure lisent la colonne, et les bases déjà en service la reçoivent + remplie. Maloja puis Last.fm ensuite, par la décision 11. Le champ *Statut* ci-dessus ne bascule pas — il ne bascule jamais dans ce projet. - **Date** : 2026-09-13 @@ -877,15 +883,19 @@ même transaction que lui. placerait deux sources de vérité sur le même fait. Seule la reprise laisse l'entrée en `uncertain`, et seule elle a besoin d'être notée. -**Et tout ce qui lisait la jointure lit désormais la colonne.** Trois endroits -demandaient « cette entrée a-t-elle déjà servi son joker » en cherchant une -ligne qui la désigne : la liste des incertaines, le compteur du lien, et le -refus d'une seconde reprise. Les trois passent à `retried_at IS NULL`. Le -compteur surtout : `link_health` rend `degraded` dès qu'une incertaine est -comptée, donc en oubliant un seul de ces trois endroits on obtient un lien -définitivement en peine à cause d'une écoute à laquelle sa personne a déjà -répondu — la panne que la décision 12 veut rendre visible, retournée en fausse -alerte permanente. +**Et tout ce qui lisait la jointure lit désormais la colonne.** Ce paragraphe a +d'abord dit *trois* endroits — la liste des incertaines, le compteur du lien et +le refus d'une seconde reprise. Ils sont **quatre** : le refus d'un *rejet* +aussi, `discard_uncertain_scrobble`, qui porte la même exclusion depuis +[#193](https://github.com/InstaZDLL/waveflow-server/pull/193). Le compte venait +d'une relecture de la décision plutôt que du fichier ; c'est le fichier qui a +raison. Les quatre passent à `retried_at IS NULL`. Le compteur surtout : +`link_health` rend `degraded` dès qu'une incertaine est comptée, donc en +oubliant un seul de ces quatre endroits on obtient un lien définitivement en +peine à cause d'une écoute à laquelle sa personne a déjà répondu — la panne que +la décision 12 veut rendre visible, retournée en fausse alerte permanente. +Chacun des quatre est éprouvé à part, par une inversion qui ne remet que +celui-là sur la jointure et fait tomber sa seule assertion. ## La rétention de la file diff --git a/migrations-v2/20260914000000_scrobble_outbox_retried_at.sql b/migrations-v2/20260914000000_scrobble_outbox_retried_at.sql new file mode 100644 index 00000000..81979b48 --- /dev/null +++ b/migrations-v2/20260914000000_scrobble_outbox_retried_at.sql @@ -0,0 +1,37 @@ +-- The joker decision 13 grants is counted on the entry, not on its descendant. +-- +-- Until now "this ambiguous entry has already been retried" was read as a join: +-- an entry stayed answerable for as long as no other row named it through +-- `retry_of`. That holds only while rows are immortal, and RFC-010's retention +-- makes them mortal — a retry ends `sent`, so it becomes purgeable, and the +-- original then turns answerable a second time. Measured rather than feared: +-- with the retry row gone the entry reappears in `uncertain_scrobbles`, +-- `retry_uncertain_scrobble` accepts it again, and the link's `uncertain` +-- counter puts it back to `degraded` for a listen already answered. +-- +-- So the fact moves onto the entry itself: null until it is not. +ALTER TABLE scrobble_outbox ADD COLUMN retried_at INTEGER; + +-- And it is filled on databases already in service. +-- +-- Retried entries exist since #191, recognisable by exactly the join this +-- migration abandons. Adding the column empty would make every one of them +-- answerable a second time: on the only servers that have any, the migration +-- would reintroduce the very defect it exists to correct. So it reads +-- `retry_of` one last time. +-- +-- The instant taken is the retry row's `created_at`, which is when the joker +-- was spent. `scrobble_outbox_one_retry_idx` is unique over `retry_of`, so the +-- correlated subquery names at most one row and the value is not a choice +-- among several. +UPDATE scrobble_outbox + SET retried_at = ( + SELECT r.created_at + FROM scrobble_outbox r + WHERE r.retry_of = scrobble_outbox.id + ) + WHERE EXISTS ( + SELECT 1 + FROM scrobble_outbox r + WHERE r.retry_of = scrobble_outbox.id + ); diff --git a/src/api/scrobbling.rs b/src/api/scrobbling.rs index 7923eb61..1584d2fd 100644 --- a/src/api/scrobbling.rs +++ b/src/api/scrobbling.rs @@ -184,7 +184,7 @@ pub async fn discard_uncertain_scrobble( // generates a client from. // // This route used to declare a `409` as well. Nothing could produce one: the -// service excludes anything already retried with `NOT EXISTS`, so a spent entry +// service refuses anything already retried on `retried_at`, so a spent entry // stops being findable and answers 404 — and `db_error` maps every sqlx failure // to 503 rather than to a conflict, so even the unique index could not surface // as one. Removing the declaration was not enough on its own, because diff --git a/src/database.rs b/src/database.rs index 0cea8f33..8a34e40c 100644 --- a/src/database.rs +++ b/src/database.rs @@ -1606,4 +1606,158 @@ mod tests { "unexpected error: {error:#}" ); } + + /// A retry queued before `retried_at` existed still counts as the joker + /// spent, once the column arrives. + /// + /// Retried entries exist on every database linked since #191, and until + /// `20260914000000` the only trace of one was a second row naming it + /// through `retry_of`. Adding the column empty would leave every one of + /// those originals unanswered again — the migration would reintroduce, on + /// the only servers that have anything to reintroduce it to, exactly the + /// defect it exists to correct. Reading it on a database populated at the + /// schema version before the column is what says otherwise; a rereading of + /// the SQL is not. + #[tokio::test] + async fn a_retry_queued_before_the_column_counts_as_the_joker_spent() { + use sqlx::migrate::Migrate; + + /// The last migration before `scrobble_outbox.retried_at`. + const BEFORE_THE_COLUMN: i64 = 20260913000000; + + /// When the ambiguous entry was last written, and when its retry was + /// queued. Distinct on purpose: the backfill takes the instant the + /// joker was spent, which is the retry's, and a test where the two + /// agreed could not tell one from the other. + const SETTLED_AT: i64 = 1_700_000_111_000; + const RETRIED_AT: i64 = 1_700_000_555_000; + + let temp = tempfile::tempdir().expect("temporary directory"); + let database = Database::open(&Config::for_data_dir(temp.path().join("data"))) + .await + .expect("open database"); + + let mut connection = database.pool.acquire().await.expect("connection"); + connection + .ensure_migrations_table(&MIGRATOR.table_name) + .await + .expect("migrations table"); + for migration in MIGRATOR + .iter() + .filter(|migration| migration.version <= BEFORE_THE_COLUMN) + { + connection + .apply(&MIGRATOR.table_name, migration) + .await + .expect("earlier migration"); + } + + // An account and one authorisation, inserted as rows rather than + // through the services: this test is about what the migration does to + // a shape, and `hash_password` on a literal is what the repository's + // code scanning refuses. + const ACCOUNT: &str = "6f1d1b6e-0f1a-4a3d-9c21-000000000011"; + const LINK: &str = "6f1d1b6e-0f1a-4a3d-9c21-000000000012"; + for (statement, binds) in [ + ( + "INSERT INTO account (id, username, password_hash, role, created_at, updated_at) \ + VALUES (?, 'listener', 'this-account-never-authenticates', 'user', 0, 0)", + vec![ACCOUNT], + ), + ( + "INSERT INTO scrobble_link (id, user_id, provider, status, credential_nonce, \ + credential_ciphertext, created_at, updated_at) \ + VALUES (?, ?, 'listenbrainz', 'active', x'000000000000000000000000', \ + x'00', 0, 0)", + vec![LINK, ACCOUNT], + ), + ] { + let mut query = sqlx::query(statement); + for bind in binds { + query = query.bind(bind.to_owned()); + } + query.execute(&mut *connection).await.expect("seed row"); + } + + // Three rows in the shape #191 leaves behind: an ambiguous entry whose + // joker was spent, the retry that spent it, and a second ambiguous + // entry nobody has answered. + const ANSWERED: &str = "6f1d1b6e-0f1a-4a3d-9c21-000000000013"; + const RETRY: &str = "6f1d1b6e-0f1a-4a3d-9c21-000000000014"; + const UNANSWERED: &str = "6f1d1b6e-0f1a-4a3d-9c21-000000000015"; + for (public_id, state, created_at, updated_at) in [ + (ANSWERED, "uncertain", SETTLED_AT, SETTLED_AT), + (RETRY, "sent", RETRIED_AT, RETRIED_AT), + (UNANSWERED, "uncertain", SETTLED_AT, SETTLED_AT), + ] { + sqlx::query( + "INSERT INTO scrobble_outbox (public_id, link_id, played_at, title, \ + artists_json, state, attempts, next_attempt_at, created_at, updated_at) \ + VALUES (?, ?, 0, 'Zenith', '[\"Nova Kern\"]', ?, 1, 0, ?, ?)", + ) + .bind(public_id) + .bind(LINK) + .bind(state) + .bind(created_at) + .bind(updated_at) + .execute(&mut *connection) + .await + .expect("seed outbox row"); + } + // The link between them is written by lookup rather than by assuming + // the first row took rowid 1: `retry_of` speaks in rowids, and a + // fixture that guessed one would pass for a reason unrelated to the + // migration. + sqlx::query( + "UPDATE scrobble_outbox \ + SET retry_of = (SELECT id FROM scrobble_outbox WHERE public_id = ?) \ + WHERE public_id = ?", + ) + .bind(ANSWERED) + .bind(RETRY) + .execute(&mut *connection) + .await + .expect("point the retry at what it retried"); + drop(connection); + + database + .migrate() + .await + .expect("a database with a queue migrates"); + + let mut connection = database.pool.acquire().await.expect("connection"); + let answered: Option = + sqlx::query_scalar("SELECT retried_at FROM scrobble_outbox WHERE public_id = ?") + .bind(ANSWERED) + .fetch_one(&mut *connection) + .await + .expect("the answered entry"); + assert_eq!( + answered, + Some(RETRIED_AT), + "an entry retried before the column must carry the instant its retry was queued" + ); + + let unanswered: Option = + sqlx::query_scalar("SELECT retried_at FROM scrobble_outbox WHERE public_id = ?") + .bind(UNANSWERED) + .fetch_one(&mut *connection) + .await + .expect("the unanswered entry"); + assert_eq!( + unanswered, None, + "an entry nobody answered must still be asking" + ); + + let retry: Option = + sqlx::query_scalar("SELECT retried_at FROM scrobble_outbox WHERE public_id = ?") + .bind(RETRY) + .fetch_one(&mut *connection) + .await + .expect("the retry"); + assert_eq!( + retry, None, + "the retry spent nobody's joker; it is the one that was spent" + ); + } } diff --git a/src/services/scrobbling.rs b/src/services/scrobbling.rs index 4b2f9867..a556901e 100644 --- a/src/services/scrobbling.rs +++ b/src/services/scrobbling.rs @@ -537,20 +537,27 @@ impl DomainServices { for row in rows { let link_id: String = row.try_get("id")?; let provider = ScrobbleProvider::from_str(row.try_get("provider")?)?; - // `SUM(CASE …)` rather than `COUNT(*) FILTER`: the aggregate filter - // needs a SQLite newer than the floor this crate builds against, - // and one query answering four counts is the point either way. + // `SUM(CASE …)` rather than `COUNT(*) FILTER`, because one query + // answering four counts is the point. An earlier note here gave a + // second reason — that the aggregate filter wanted a SQLite newer + // than this crate builds against — and it was simply untrue: the + // library is bundled, at 3.51, and the schema has asked for 3.37 + // since the first `STRICT` table. // // The uncertain count excludes an entry a person has already // retried. It is still true and still readable; it has simply // stopped asking for a decision, and a counter that kept naming it // would ask for the same one forever. + // + // Read on `retried_at` rather than on a row naming this one through + // `retry_of`: the retry ends `sent`, so retention will eventually + // take it, and a count deduced from its survival would put this + // link back to `degraded` for a listen already answered. let counts = sqlx::query( "SELECT \ SUM(CASE WHEN state='pending' AND attempts=0 THEN 1 ELSE 0 END) AS pending, \ SUM(CASE WHEN state='pending' AND attempts>0 THEN 1 ELSE 0 END) AS retrying, \ - SUM(CASE WHEN state='uncertain' AND id NOT IN \ - (SELECT retry_of FROM scrobble_outbox WHERE retry_of IS NOT NULL) \ + SUM(CASE WHEN state='uncertain' AND retried_at IS NULL \ THEN 1 ELSE 0 END) AS uncertain, \ MIN(CASE WHEN state='pending' THEN created_at END) AS oldest_pending_at \ FROM scrobble_outbox WHERE link_id=?", @@ -628,8 +635,7 @@ impl DomainServices { o.last_failure, o.updated_at \ FROM scrobble_outbox o JOIN scrobble_link l ON l.id = o.link_id \ WHERE l.user_id=? AND l.status <> 'unlinked' AND o.state='uncertain' \ - AND o.id NOT IN \ - (SELECT retry_of FROM scrobble_outbox WHERE retry_of IS NOT NULL) \ + AND o.retried_at IS NULL \ ORDER BY o.played_at DESC, o.id DESC", ) .bind(user_id.to_string()) @@ -658,6 +664,10 @@ impl DomainServices { /// Named by its `public_id`, never by its rowid: the sequential one would /// tell anyone holding a single entry of their own how many listens this /// whole server has queued. + /// + /// An entry already retried is refused here too, and on the same column the + /// other three readers use: it has stopped asking, so there is nothing left + /// to prefer a gap to. pub async fn discard_uncertain_scrobble( &self, user_id: Uuid, @@ -666,8 +676,7 @@ impl DomainServices { let _writer = self.db.writer_guard().await; let changed = sqlx::query( "UPDATE scrobble_outbox SET state='discarded', updated_at=? \ - WHERE public_id=? AND state='uncertain' \ - AND id NOT IN (SELECT retry_of FROM scrobble_outbox WHERE retry_of IS NOT NULL) \ + WHERE public_id=? AND state='uncertain' AND retried_at IS NULL \ AND link_id IN \ (SELECT id FROM scrobble_link WHERE user_id=? AND status <> 'unlinked')", ) @@ -692,6 +701,12 @@ impl DomainServices { /// /// The original stops being counted as uncertain, because it no longer asks /// the person for anything; it has not stopped being true. + /// + /// **The joker is spent on the entry, not on its descendant.** Whether this + /// gesture has already been made is `retried_at`, a fact the row carries, + /// and not the survival of a row naming it through `retry_of`. A retry ends + /// `sent`, so retention will take it, and every reader deducing the answer + /// from a join would then hand this listen a second joker. pub async fn retry_uncertain_scrobble( &self, user_id: Uuid, @@ -700,31 +715,36 @@ impl DomainServices { let now = now_ms(); let _writer = self.db.writer_guard().await; let mut tx = self.db.pool().begin().await?; - // The link has to be live: a retry under a withdrawn authorisation - // would submit to whichever profile happens to be linked now, which is - // the very substitution generations exist to prevent. + // The conditional UPDATE comes first because it is the arbitration: + // spending the joker and refusing a second one are the same write, so + // two simultaneous calls cannot both find the entry unanswered. // - // And the entry must not already have been retried. The original stays - // `uncertain` for good — erasing it would falsify the only trace - // explaining why a duplicate exists — so nothing in the row itself says - // it has been answered, and without this clause a second call would - // queue a second copy, and a third a third. Decision 13 gives that - // acceptance once, for one listen, deliberately. The unique index on - // `retry_of` refuses the insert as well; this clause is what turns the - // refusal into an ordinary 404 instead of a constraint error. - // The rowid comes back from this lookup rather than from the caller: - // `retry_of`, the ordering and the jitter all speak in rowids, and the - // public name is resolved to one exactly here, once. - let rowid = sqlx::query_scalar::<_, i64>( - "SELECT o.id FROM scrobble_outbox o JOIN scrobble_link l ON l.id=o.link_id \ - WHERE o.public_id=? AND o.state='uncertain' AND l.user_id=? AND l.status='active' \ - AND NOT EXISTS (SELECT 1 FROM scrobble_outbox r WHERE r.retry_of=o.id)", + // The link has to be live in the same breath: a retry under a withdrawn + // authorisation would submit to whichever profile happens to be linked + // now, which is the very substitution generations exist to prevent. + // + // `updated_at` is deliberately left alone. It is what + // `uncertain_scrobbles` publishes as the moment this listen became + // ambiguous, and answering it did not make it ambiguous again. + // + // `RETURNING` rather than a second lookup by `public_id`: `retry_of`, + // the ordering and the jitter all speak in rowids, and this hands back + // the rowid of the row the UPDATE itself took. A `SELECT` afterwards + // would name the same row today and would be a separate claim about + // which row that is. + let claimed = sqlx::query_scalar::<_, i64>( + "UPDATE scrobble_outbox SET retried_at=? \ + WHERE public_id=? AND state='uncertain' AND retried_at IS NULL \ + AND link_id IN \ + (SELECT id FROM scrobble_link WHERE user_id=? AND status='active') \ + RETURNING id", ) + .bind(now) .bind(entry.to_string()) .bind(user_id.to_string()) .fetch_optional(&mut *tx) .await?; - let Some(rowid) = rowid else { + let Some(rowid) = claimed else { return Err(ServiceError::NotFound); }; let public_id = Uuid::new_v4(); diff --git a/tests/scrobbling.rs b/tests/scrobbling.rs index c4fec1f7..a01f9f25 100644 --- a/tests/scrobbling.rs +++ b/tests/scrobbling.rs @@ -661,6 +661,90 @@ async fn an_ambiguous_entry_may_be_retried_once_and_not_twice() { assert_eq!(rows(&state).await.len(), 2); } +/// The joker stays spent once the retry that spent it is gone. +/// +/// Until `retried_at`, all four readers deduced "already answered" from a +/// second row naming this one through `retry_of`. That holds only while rows +/// are immortal, and RFC-010's retention makes them mortal: a retry ends +/// `sent`, so it is the first thing a purge takes. The `DELETE` below is that +/// purge, played by hand because the purge itself belongs to the next slice — +/// what it does to this entry does not wait for it. +/// +/// Four assertions rather than one because there were four readers, and the +/// counter is the one that matters most: `link_health` answers `degraded` for a +/// single counted `uncertain`, so a link would sit in permanent false alarm +/// over a listen whose owner answered weeks ago. +#[tokio::test] +async fn a_spent_joker_is_not_returned_when_the_retry_is_purged() { + let (_temp, config, state) = test_app().await; + let fixture = fixture(&config, &state, "purged-retry-listener").await; + state + .services + .link_scrobble(fixture.owner, ScrobbleProvider::ListenBrainz, "lb-secret") + .await + .unwrap(); + let uncertain_id = one_uncertain_entry(&state, &fixture).await; + state + .services + .retry_uncertain_scrobble(fixture.owner, uncertain_id) + .await + .unwrap(); + + // Retention, played by hand: the retry row leaves, the answered original + // stays. `uncertain` is never purged, so this is the shape a real purge + // leaves behind and not an invented one. + sqlx::query("DELETE FROM scrobble_outbox WHERE retry_of IS NOT NULL") + .execute(state.db.pool()) + .await + .unwrap(); + assert_eq!( + rows(&state).await.len(), + 1, + "the purge must have taken the retry and left the original" + ); + + assert!( + state + .services + .uncertain_scrobbles(fixture.owner) + .await + .unwrap() + .is_empty(), + "an answered entry must not come back asking" + ); + let links = state.services.scrobble_links(fixture.owner).await.unwrap(); + assert_eq!( + links[0].uncertain, 0, + "an answered entry must not be counted again" + ); + assert_eq!( + links[0].health, "healthy", + "and the link must not be degraded by a listen already answered" + ); + assert!( + matches!( + state + .services + .retry_uncertain_scrobble(fixture.owner, uncertain_id) + .await + .unwrap_err(), + ServiceError::NotFound + ), + "the joker was spent once and does not come back" + ); + assert!( + matches!( + state + .services + .discard_uncertain_scrobble(fixture.owner, uncertain_id) + .await + .unwrap_err(), + ServiceError::NotFound + ), + "nor does the gesture it replaced" + ); +} + #[tokio::test] async fn a_refused_listen_is_not_offered_to_the_destination_again() { let (_temp, config, state) = test_app().await;