From 9309f67885db4f4a361e66fa260f84e37e439de6 Mon Sep 17 00:00:00 2001 From: InstaZDLL Date: Sun, 30 Aug 2026 18:45:39 +0200 Subject: [PATCH 1/5] feat(api): record how far a device has read a library's feed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RFC-007 decision 8, built. `library_event_ack`, keyed `(library_id, device_id)` — `sync_ack` minus its account column, because the journal is keyed per account and this feed is keyed per library. A device belongs to exactly one account, so an account column here would be a third value derivable from the other two. **Two checks rather than one, and both in the statement.** The device must be this account's and unrevoked, and the account must be a member of the library. `sync_ack` needs only the first: the journal has no second scope to escape into, and this does. A cursor beyond what the feed has written is refused too — otherwise a client marks itself caught up with events that do not exist yet and is silently behind when they arrive. All three refusals answer 422 alike. Telling them apart would say whether a library exists to somebody who may not know. The stored cursor is never lowered. Two of a client's own requests racing must not let the older win, and the server is not the place to decide which of them is the truth. **And the table does not hold the purge back**, which is the whole of what decision 8 says it is for. A device that never returns would pin a feed forever, and a shared library would lose retention entirely the moment one phone was thrown away. What the acknowledgement buys instead is a number: the purge now reports, per library, how many devices its watermark has just overtaken and sent back to the catalogue. The ack informs; decision 7 decides. ## The tenancy case was tested wrong, and the removal found it Three guards, and only two fell on the first pass. The membership join survived being deleted — because the test's "a library this account is not a member of" case had sent the *owner's* device with the stranger's token, which is refused a step earlier by the device check and says nothing about tenancy. The case that exercises the join is the stranger's own device against a library they cannot see. With that written, deleting the join fails the test: 204 where it wants 422. The other two already fell — an out-of-range cursor is accepted, and an older acknowledgement overwrites a newer one. A test that passes for the wrong reason is worth less than no test, because it also stops anyone looking. Claude-Session: https://claude.ai/code/session_01GtAmdsaBCg8Cs2rvrdLD7Z Signed-off-by: InstaZDLL --- docs/rfcs/RFC-007-library-event-stream.md | 8 +- .../20260830020000_library_event_ack.sql | 28 +++ src/api/libraries.rs | 48 ++++++ src/api/mod.rs | 4 + src/lib.rs | 2 + src/services/library_events.rs | 97 +++++++++++ tests/catalog.rs | 161 ++++++++++++++++++ 7 files changed, 344 insertions(+), 4 deletions(-) create mode 100644 migrations-v2/20260830020000_library_event_ack.sql diff --git a/docs/rfcs/RFC-007-library-event-stream.md b/docs/rfcs/RFC-007-library-event-stream.md index 22af20c..fe7cd43 100644 --- a/docs/rfcs/RFC-007-library-event-stream.md +++ b/docs/rfcs/RFC-007-library-event-stream.md @@ -293,7 +293,7 @@ sautant le trou. C'est exactement la panne que la décision 4 refuse. ## Décision 8 — l'acquittement, et ce qu'il ne décide pas -> **Décidée le 2026-08-30, pas encore construite.** +> **Construite.** Voir la ligne *Implémentée par* de l'en-tête. Une table `library_event_ack`, clé primaire `(library_id, device_id)`, portant le curseur et sa date. C'est `sync_ack` moins sa colonne de compte : là-bas la @@ -325,6 +325,6 @@ Plus rien de cette RFC n'est ouvert. Les trois questions qu'elle portait ont pour les deux. - ~~La forme exacte de l'acquittement.~~ Décision 8. -La décision 7 est construite ; la **décision 8 est décidée et pas encore -construite**. La ligne *Implémentée par* de l'en-tête ne nomme que ce qui -tourne, et c'est elle qu'il faut lire — pas cette section. +Les décisions 7 et 8 sont construites, et cette RFC n'a plus rien en attente. +La ligne *Implémentée par* de l'en-tête ne nomme que ce qui tourne, et c'est +elle qu'il faut lire — pas cette section. diff --git a/migrations-v2/20260830020000_library_event_ack.sql b/migrations-v2/20260830020000_library_event_ack.sql new file mode 100644 index 0000000..e61e62e --- /dev/null +++ b/migrations-v2/20260830020000_library_event_ack.sql @@ -0,0 +1,28 @@ +-- How far a device has read one library's feed. RFC-007 decision 8. +-- +-- `sync_ack` minus its account column, and the difference is the whole reason +-- this is a second table rather than a wider first one: the user journal is +-- keyed per account, this feed is keyed per library. A device belongs to +-- exactly one account (`device.user_id`), so an account column here would be a +-- third value derivable from the other two — a second truth to keep in +-- agreement with the first, and the sort that goes stale in one place only. +-- +-- The account is therefore re-read from the device and checked against +-- `library_member` at write time, which is the rule every other read in this +-- server follows: tenancy lives in the query. +-- +-- Both foreign keys cascade. A revoked device and a deleted library each leave +-- nothing behind, and neither needs a sweeper to notice. +-- +-- What this table deliberately does not do is hold back the purge. A device +-- that never comes back would pin a feed forever, and a shared library would +-- lose retention entirely the moment one phone was thrown away. Retention is +-- decided by RFC-007 decision 7 and reported against these rows; it is not +-- bounded by them. +CREATE TABLE library_event_ack ( + library_id TEXT NOT NULL REFERENCES library(id) ON DELETE CASCADE, + device_id TEXT NOT NULL REFERENCES device(id) ON DELETE CASCADE, + cursor INTEGER NOT NULL CHECK (cursor >= 0), + acknowledged_at INTEGER NOT NULL, + PRIMARY KEY (library_id, device_id) +) STRICT; diff --git a/src/api/libraries.rs b/src/api/libraries.rs index 0e74aa5..fab4bbd 100644 --- a/src/api/libraries.rs +++ b/src/api/libraries.rs @@ -223,6 +223,54 @@ pub async fn scan_events( /// a different sequence and advances for different reasons, so it is a separate /// route with a separate cursor rather than a widening of the other — a rescan /// must not move a client's position in its own user journal. +/// How far a device has read one library's feed. +/// +/// A body rather than a header for the device, exactly like `/api/v2/sync/ack`: +/// the acknowledgement *is* about that device, so it is the request rather than +/// a note attached to it. The two are refused identically — 422 — for an +/// unknown or revoked device, a library the account cannot see, and a cursor +/// beyond what the feed has written. One answer for all three, because telling +/// them apart would say whether a library exists to somebody who may not know. +#[utoipa::path( + put, + path = "/api/v2/libraries/{library_id}/events/ack", + tag = "libraries", + params(("library_id" = Uuid, Path)), + request_body = LibraryEventAckRequest, + responses( + (status = 204), + (status = 401, body = ErrorResponse), + (status = 422, body = ErrorResponse) + ) +)] +pub async fn library_events_ack( + State(state): State, + Path(library_id): Path, + headers: HeaderMap, + Json(request): Json, +) -> Result { + let user = authenticated(&state, &headers, Access::Write).await?; + let acknowledged = state + .services + .acknowledge_library_events(user.id, library_id, request.device_id, request.cursor) + .await + .map_err(service_error)?; + if !acknowledged { + return Err(ApiError::Validation); + } + Ok(StatusCode::NO_CONTENT) +} + +/// What a device says it has read. +#[derive(Debug, serde::Deserialize, utoipa::ToSchema)] +pub struct LibraryEventAckRequest { + pub device_id: Uuid, + /// The highest cursor this device has processed. Never lowered by the + /// server: a client that acknowledges an older cursor after a newer one has + /// raced its own two requests. + pub cursor: i64, +} + #[utoipa::path( get, path = "/api/v2/libraries/{library_id}/events", diff --git a/src/api/mod.rs b/src/api/mod.rs index 1d87200..2e230c2 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -103,6 +103,10 @@ pub fn router(state: AppState) -> Router { .route("/api/v2/scans/{scan_id}/events", get(scan_events)) .route("/api/v2/libraries/{library_id}/tracks", get(list_tracks)) .route("/api/v2/libraries/{library_id}/events", get(library_events)) + .route( + "/api/v2/libraries/{library_id}/events/ack", + put(library_events_ack), + ) // Its own body ceiling, and only its own. Raising the router's would // hand every route on the server a surface none of them asked for. .route( diff --git a/src/lib.rs b/src/lib.rs index 8f599f1..35f6cf7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -127,6 +127,7 @@ fn chunk_body_limit(limits: &config::UploadLimits) -> usize { api::scan_status, api::scan_events, api::library_events, + api::library_events_ack, api::negotiate_uploads, api::upload_session, api::upload_chunk, @@ -265,6 +266,7 @@ fn chunk_body_limit(limits: &config::UploadLimits) -> usize { sync::SyncPage, media::StreamTicketResponse, media::CanvasResponse, + api::LibraryEventAckRequest, scanner::ScanProgress )), modifiers(&SecurityAddon), diff --git a/src/services/library_events.rs b/src/services/library_events.rs index 8eba965..ef73110 100644 --- a/src/services/library_events.rs +++ b/src/services/library_events.rs @@ -20,6 +20,13 @@ pub struct EventPurge { pub events_removed: u64, /// Libraries that lost at least one. pub libraries_trimmed: usize, + /// Devices whose acknowledged cursor now sits below the watermark. + /// + /// They have been sent back to the catalogue snapshot. Counted rather than + /// prevented — RFC-007 decision 8 says the acknowledgement informs and does + /// not decide, or one forgotten phone would stop a shared library from ever + /// being trimmed. + pub devices_stranded: usize, } impl DomainServices { @@ -89,11 +96,37 @@ impl DomainServices { if removed > 0 { purged.events_removed += removed; purged.libraries_trimmed += 1; + // What the cut cost, and to whom. This is the whole of what + // decision 8's table is for: the ack does not hold the purge + // back, so the only useful thing it can do is say which devices + // the purge has just sent back to the snapshot. + let stranded = self.devices_left_behind(&library_id).await?; + if stranded > 0 { + tracing::info!( + library = %library_id, + devices = stranded, + "trimming this feed sent devices back to the catalogue" + ); + purged.devices_stranded += stranded; + } } } Ok(purged) } + /// How many devices this library's watermark has just overtaken. + async fn devices_left_behind(&self, library_id: &str) -> Result { + let stranded: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM library_event_ack a \ + JOIN library l ON l.id=a.library_id \ + WHERE a.library_id=? AND a.cursor < l.events_purged_through", + ) + .bind(library_id) + .fetch_one(self.db.pool()) + .await?; + Ok(usize::try_from(stranded).unwrap_or(0)) + } + /// Cuts one library's feed and moves its watermark with it. /// /// The delete and the watermark are one transaction, and that is the whole @@ -159,6 +192,70 @@ impl DomainServices { Ok(u64::try_from(removed).unwrap_or(0)) } + /// Records how far a device has read one library's feed. + /// + /// RFC-007 decision 8. Two checks rather than one, and both are in the + /// statement: the device must be this account's and unrevoked, and the + /// account must be a member of the library. `sync_ack` needs only the + /// first, because the journal is keyed per account and there is no second + /// scope to escape into; here there is. + /// + /// `false` for anything refused — an unknown or revoked device, a library + /// this account cannot see, a cursor beyond what the feed has written. A + /// caller who is not a member learns nothing about whether the library + /// exists, which is the rule everywhere else in this API. + pub async fn acknowledge_library_events( + &self, + user_id: Uuid, + library_id: Uuid, + device_id: Uuid, + cursor: i64, + ) -> Result { + if cursor < 0 { + return Ok(false); + } + let _writer = self.db.writer_guard().await; + let mut tx = self.db.pool().begin().await?; + + // A cursor beyond what the feed has written would let a client mark + // itself caught up with events that do not exist yet, and then be + // silently behind when they arrive. + let latest: i64 = sqlx::query_scalar( + "SELECT COALESCE(MAX(cursor), 0) FROM library_event WHERE library_id=?", + ) + .bind(library_id.to_string()) + .fetch_one(&mut *tx) + .await?; + if cursor > latest { + return Ok(false); + } + + let result = sqlx::query( + "INSERT INTO library_event_ack (library_id, device_id, cursor, acknowledged_at) \ + SELECT ?, ?, ?, ? WHERE EXISTS ( \ + SELECT 1 FROM device d \ + JOIN library_member m ON m.user_id=d.user_id \ + WHERE d.id=? AND d.user_id=? AND d.revoked_at IS NULL AND m.library_id=? \ + ) ON CONFLICT (library_id, device_id) DO UPDATE SET \ + cursor=MAX(library_event_ack.cursor, excluded.cursor), \ + acknowledged_at=excluded.acknowledged_at", + ) + .bind(library_id.to_string()) + .bind(device_id.to_string()) + .bind(cursor) + .bind(now_ms()) + .bind(device_id.to_string()) + .bind(user_id.to_string()) + .bind(library_id.to_string()) + .execute(&mut *tx) + .await?; + tx.commit().await?; + // Never lowered: a client that acknowledges an older cursor after a + // newer one has raced its own two requests, and the server is not the + // place to decide which of them is the truth. + Ok(result.rows_affected() == 1) + } + /// One page of a library's changes, for a caller entitled to that library. /// /// A caller who is not a member gets `NotFound`, not `Forbidden`: a feed diff --git a/tests/catalog.rs b/tests/catalog.rs index 9a0caeb..3796fd5 100644 --- a/tests/catalog.rs +++ b/tests/catalog.rs @@ -1919,3 +1919,164 @@ async fn retention_cuts_by_age_never_below_the_floor_and_never_at_the_bound() { "the floor kept a usable tail rather than an empty feed" ); } + +/// RFC-007 decision 8. The acknowledgement says how far a device has read; what +/// it deliberately does not do is hold the purge back. +#[tokio::test] +async fn an_acknowledgement_records_a_device_without_holding_the_purge_back() { + let temp = tempfile::tempdir().unwrap(); + let mut config = waveflow_server::Config::for_data_dir(temp.path().join("data")); + config.library_event_retention.min_events = 1; + config.library_event_retention.days = 30; + let state = waveflow_server::initialize(&config).await.unwrap(); + + let password = security::generate_token("test-password-"); + let hash = security::hash_password(&password).unwrap(); + let owner = state + .db + .create_account("ack", &hash, AccountRole::Admin, now_ms()) + .await + .unwrap(); + let music = config.data_dir.join("ack-music"); + std::fs::create_dir_all(&music).unwrap(); + let library = state + .db + .create_library( + owner, + "Ack", + &std::fs::canonicalize(&music).unwrap(), + LibraryVisibility::Private, + now_ms(), + ) + .await + .unwrap(); + + let scan = state + .db + .create_scan_job(library, Some(owner), "manual") + .await + .unwrap(); + state.db.start_scan_job(scan, 3, false).await.unwrap(); + for index in 0..3usize { + let mut input = catalog_input(index, "Nova Kern"); + input.title = format!("Track {index}"); + input.album = None; + input.album_artist = None; + state + .db + .apply_catalog_track(library, scan, &input, None, false) + .await + .unwrap(); + } + state.db.finish_scan_job(scan, 0).await.unwrap(); + + let router = waveflow_server::app(&config, state.clone()); + let (token, device) = login_session(&router, "ack", &password).await; + let cursors: Vec = + sqlx::query_scalar("SELECT cursor FROM library_event WHERE library_id=? ORDER BY cursor") + .bind(library.to_string()) + .fetch_all(state.db.pool()) + .await + .unwrap(); + + let ack = |cursor: i64, device: String, token: String| { + let router = router.clone(); + async move { + router + .oneshot( + Request::put(format!("/api/v2/libraries/{library}/events/ack")) + .header("authorization", format!("Bearer {token}")) + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({ "device_id": device, "cursor": cursor }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap() + .status() + } + }; + + assert_eq!( + ack(cursors[0], device.clone(), token.clone()).await, + StatusCode::NO_CONTENT + ); + // A cursor beyond what the feed has written would let a client mark itself + // caught up with events that do not exist yet, and be silently behind when + // they arrive. + assert_eq!( + ack(cursors[2] + 1, device.clone(), token.clone()).await, + StatusCode::UNPROCESSABLE_ENTITY + ); + // Somebody else's device, refused the same way a library one cannot see is + // — one answer for all of them, or the refusal tells them apart. + let stranger_hash = security::hash_password("correct horse battery staple").unwrap(); + state + .db + .create_account("ack-stranger", &stranger_hash, AccountRole::Admin, now_ms()) + .await + .unwrap(); + let (stranger_token, stranger_device) = + login_session(&router, "ack-stranger", "correct horse battery staple").await; + assert_eq!( + ack(cursors[0], stranger_device.clone(), token.clone()).await, + StatusCode::UNPROCESSABLE_ENTITY, + "a device that is not this account's" + ); + // Their own device, and a library they are not a member of. This is the + // case the membership join answers — the one above is refused a step + // earlier, by the device check, so it says nothing about tenancy. + assert_eq!( + ack(cursors[0], stranger_device, stranger_token).await, + StatusCode::UNPROCESSABLE_ENTITY, + "a library this account is not a member of" + ); + + // Never lowered: two of a client's own requests racing must not let the + // older one win. + assert_eq!( + ack(cursors[2], device.clone(), token.clone()).await, + StatusCode::NO_CONTENT + ); + assert_eq!( + ack(cursors[0], device.clone(), token).await, + StatusCode::NO_CONTENT + ); + let stored: i64 = sqlx::query_scalar( + "SELECT cursor FROM library_event_ack WHERE library_id=? AND device_id=?", + ) + .bind(library.to_string()) + .bind(&device) + .fetch_one(state.db.pool()) + .await + .unwrap(); + assert_eq!(stored, cursors[2], "the highest acknowledged cursor stands"); + + // And the point of the whole table: it does not hold the purge back. This + // device has acknowledged everything, then the feed is aged out anyway — + // one forgotten phone must not stop a shared library being trimmed. + sqlx::query("UPDATE library_event_ack SET cursor=? WHERE library_id=?") + .bind(cursors[0]) + .bind(library.to_string()) + .execute(state.db.pool()) + .await + .unwrap(); + let now = now_ms(); + sqlx::query("UPDATE library_event SET changed_at=? WHERE library_id=?") + .bind(now - 400 * 24 * 60 * 60 * 1000i64) + .bind(library.to_string()) + .execute(state.db.pool()) + .await + .unwrap(); + let purged = state.services.purge_library_events(now).await.unwrap(); + assert_eq!( + purged.events_removed, 2, + "the floor of one keeps the newest" + ); + assert_eq!( + purged.devices_stranded, 1, + "and the cut is reported against the device it sent back" + ); +} From 5fab7f548f0f00bd6e377666539af5ae7c9b9dad Mon Sep 17 00:00:00 2001 From: InstaZDLL Date: Sun, 30 Aug 2026 18:46:09 +0200 Subject: [PATCH 2/5] docs(rfc): name the pull request that built decision 8 #167 joins the line, and with it RFC-007 has nothing left decided but unbuilt. Added once the number existed rather than guessed, which is the only way a line whose whole value is that it can be checked is worth writing. Claude-Session: https://claude.ai/code/session_01GtAmdsaBCg8Cs2rvrdLD7Z Signed-off-by: InstaZDLL --- docs/rfcs/RFC-007-library-event-stream.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/rfcs/RFC-007-library-event-stream.md b/docs/rfcs/RFC-007-library-event-stream.md index fe7cd43..0c5f037 100644 --- a/docs/rfcs/RFC-007-library-event-stream.md +++ b/docs/rfcs/RFC-007-library-event-stream.md @@ -6,7 +6,9 @@ (l'appareil d'origine), [#159](https://github.com/InstaZDLL/waveflow-server/pull/159) (les événements d'album, décision 6), [#166](https://github.com/InstaZDLL/waveflow-server/pull/166) (la rétention, - décision 7). Le champ *Statut* ci-dessus ne bascule jamais dans ce + décision 7), + [#167](https://github.com/InstaZDLL/waveflow-server/pull/167) (l'acquittement, + décision 8). Le champ *Statut* ci-dessus ne bascule jamais dans ce projet : c'est cette ligne qui dit ce qui tourne, et elle se vérifie — une PR se lit, un mot de statut ne s'audite pas. - **Date** : 2026-08-25 From d7de13855dce13dcbb1049776493905dafbe5af6 Mon Sep 17 00:00:00 2001 From: InstaZDLL Date: Sun, 30 Aug 2026 19:20:07 +0200 Subject: [PATCH 3/5] fix(api): document the 403, and bill each purge for its own cut MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Both findings were real. **403 is reachable and the document did not say so.** The route asks for `Access::Write`, and `authenticated` answers `Forbidden` when the token's scopes do not grant it — a read-scoped API token gets 403, not 422. A client generated from this document would have had no branch for it. **And `devices_stranded` counted what it had inherited.** It asked how many acknowledgements sat below the watermark, which is the standing total rather than this pass's cost: a device sent back to the catalogue in August was counted again by every trim for as long as it stayed behind. The number is logged as *"trimming this feed sent devices back"*, present tense, so it has to answer for the pass that prints it. It is now counted between the two watermarks — at or above the old one, below the new — and inside the transaction, because that is the only place both are known. Read afterwards, the old one is already gone. The test grows a second trim with the acknowledgement untouched: the device is still behind, and this pass reports nought. Removing the lower bound fails it, 1 where it wants 0. One thing found while checking the first and deliberately not fixed here: `/api/v2/sync/ack` has the identical omission — same `Access::Write`, same missing 403, same `ErrorResponse` body. It is one line in another feature's file, so it is recorded here rather than folded into a change about the library feed. Claude-Session: https://claude.ai/code/session_01GtAmdsaBCg8Cs2rvrdLD7Z Signed-off-by: InstaZDLL --- src/api/libraries.rs | 1 + src/services/library_events.rs | 57 +++++++++++++++++++++------------- tests/catalog.rs | 34 ++++++++++++++++++++ 3 files changed, 70 insertions(+), 22 deletions(-) diff --git a/src/api/libraries.rs b/src/api/libraries.rs index fab4bbd..744b8eb 100644 --- a/src/api/libraries.rs +++ b/src/api/libraries.rs @@ -240,6 +240,7 @@ pub async fn scan_events( responses( (status = 204), (status = 401, body = ErrorResponse), + (status = 403, body = ErrorResponse), (status = 422, body = ErrorResponse) ) )] diff --git a/src/services/library_events.rs b/src/services/library_events.rs index ef73110..8f9b194 100644 --- a/src/services/library_events.rs +++ b/src/services/library_events.rs @@ -92,15 +92,10 @@ impl DomainServices { // writer gate is process-wide, and holding it across every feed on // a server with fifty libraries would stall every other mutation // for the whole pass. - let removed = self.purge_one_library(&library_id, cutoff).await?; + let (removed, stranded) = self.purge_one_library(&library_id, cutoff).await?; if removed > 0 { purged.events_removed += removed; purged.libraries_trimmed += 1; - // What the cut cost, and to whom. This is the whole of what - // decision 8's table is for: the ack does not hold the purge - // back, so the only useful thing it can do is say which devices - // the purge has just sent back to the snapshot. - let stranded = self.devices_left_behind(&library_id).await?; if stranded > 0 { tracing::info!( library = %library_id, @@ -114,19 +109,6 @@ impl DomainServices { Ok(purged) } - /// How many devices this library's watermark has just overtaken. - async fn devices_left_behind(&self, library_id: &str) -> Result { - let stranded: i64 = sqlx::query_scalar( - "SELECT COUNT(*) FROM library_event_ack a \ - JOIN library l ON l.id=a.library_id \ - WHERE a.library_id=? AND a.cursor < l.events_purged_through", - ) - .bind(library_id) - .fetch_one(self.db.pool()) - .await?; - Ok(usize::try_from(stranded).unwrap_or(0)) - } - /// Cuts one library's feed and moves its watermark with it. /// /// The delete and the watermark are one transaction, and that is the whole @@ -134,7 +116,12 @@ impl DomainServices { /// window where the watermark claims less than has gone — and a client /// reading into it is handed a catch-up that looks complete while silently /// skipping the gap, which is the failure decision 4 exists to refuse. - async fn purge_one_library(&self, library_id: &str, cutoff: i64) -> Result { + /// Answers what was cut and how many devices the cut has *newly* stranded. + async fn purge_one_library( + &self, + library_id: &str, + cutoff: i64, + ) -> Result<(u64, usize), ServiceError> { let _writer = self.db.writer_guard().await; let mut tx = self.db.pool().begin().await?; @@ -162,7 +149,7 @@ impl DomainServices { let removed: i64 = eligible.try_get("n")?; let highest: Option = eligible.try_get("highest")?; let (Some(highest), true) = (highest, removed > 0) else { - return Ok(0); + return Ok((0, 0)); }; sqlx::query( @@ -178,6 +165,29 @@ impl DomainServices { .execute(&mut *tx) .await?; + // Counted between the two watermarks rather than below the new one: + // this pass is answering for what *it* cost. A device already below the + // old watermark was sent back to the catalogue by an earlier pass, and + // counting it again every time the feed is trimmed would report a bill + // that only ever grows. + // + // Inside the transaction because both watermarks are known here and + // nowhere else — read afterwards, the old one is already gone. + let previous: i64 = + sqlx::query_scalar("SELECT events_purged_through FROM library WHERE id=?") + .bind(library_id) + .fetch_one(&mut *tx) + .await?; + let stranded: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM library_event_ack \ + WHERE library_id=? AND cursor >= ? AND cursor < ?", + ) + .bind(library_id) + .bind(previous) + .bind(highest) + .fetch_one(&mut *tx) + .await?; + // `MAX` never decreases: a pass that cut an older tail must not lower a // watermark an earlier one raised, and two passes racing must not // either. @@ -189,7 +199,10 @@ impl DomainServices { .execute(&mut *tx) .await?; tx.commit().await?; - Ok(u64::try_from(removed).unwrap_or(0)) + Ok(( + u64::try_from(removed).unwrap_or(0), + usize::try_from(stranded).unwrap_or(0), + )) } /// Records how far a device has read one library's feed. diff --git a/tests/catalog.rs b/tests/catalog.rs index 3796fd5..ad20271 100644 --- a/tests/catalog.rs +++ b/tests/catalog.rs @@ -2079,4 +2079,38 @@ async fn an_acknowledgement_records_a_device_without_holding_the_purge_back() { purged.devices_stranded, 1, "and the cut is reported against the device it sent back" ); + + // A pass answers for what it cost, not for what it inherits. Write two more + // events, age them, and trim again: the device is still below the + // watermark, but this pass is not what put it there. + let scan = state + .db + .create_scan_job(library, Some(owner), "manual") + .await + .unwrap(); + state.db.start_scan_job(scan, 2, false).await.unwrap(); + for index in 3..5usize { + let mut input = catalog_input(index, "Nova Kern"); + input.title = format!("Track {index}"); + input.album = None; + input.album_artist = None; + state + .db + .apply_catalog_track(library, scan, &input, None, false) + .await + .unwrap(); + } + state.db.finish_scan_job(scan, 0).await.unwrap(); + sqlx::query("UPDATE library_event SET changed_at=? WHERE library_id=?") + .bind(now - 400 * 24 * 60 * 60 * 1000i64) + .bind(library.to_string()) + .execute(state.db.pool()) + .await + .unwrap(); + let purged = state.services.purge_library_events(now).await.unwrap(); + assert!(purged.events_removed > 0, "there was something left to cut"); + assert_eq!( + purged.devices_stranded, 0, + "already stranded is not stranded again — a bill that only grows is not a bill" + ); } From 6af1f1d1862657e3a33c21d318f0f4a8d5d73f30 Mon Sep 17 00:00:00 2001 From: InstaZDLL Date: Sun, 30 Aug 2026 19:29:54 +0200 Subject: [PATCH 4/5] fix(api): scope the acknowledgement's cursor read by membership too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Valid, though not for the reason it might read as. There is no leak today: both ways out of this function answer 422, so a caller cannot tell "beyond the feed" from "not your library" whichever way the read is written. What makes it worth changing is the project's own rule — tenancy lives in the query, never in a check the query trusts. `library_changes` keeps a redundant membership predicate and says so in a comment, precisely so that removing one guard cannot quietly widen another. This read had only `library_id`. The INSERT's `EXISTS` stays where it is. It is what covers a membership revoked between the two statements, and it was not the redundant half. ## The first version of this fix was dead code, and SQL said so Written as an aggregate over the join — `SELECT COALESCE(MAX(e.cursor), 0) FROM library_member m JOIN ...` — it compiled, passed every test, and enforced nothing. An aggregate with no `GROUP BY` returns **one row** when the `WHERE` matches nothing, holding NULL, which `COALESCE` turns into `Some(0)`: a library the caller cannot see would have answered "latest is 0" and the `else` branch beneath it would never have run. Checked against sqlite3 rather than reasoned about: the aggregate form returns one row for an empty membership, the subquery form returns none. So the maximum is a subquery and the outer statement is non-aggregate over `library_member`, where zero matching rows means zero rows. No test distinguishes the two, and none can — both still answer 422. This one was caught by running the SQL, which is the only thing that could have caught it. Claude-Session: https://claude.ai/code/session_01GtAmdsaBCg8Cs2rvrdLD7Z Signed-off-by: InstaZDLL --- src/services/library_events.rs | 28 +++++++++++++++++++++++++--- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/src/services/library_events.rs b/src/services/library_events.rs index 8f9b194..4117f30 100644 --- a/src/services/library_events.rs +++ b/src/services/library_events.rs @@ -233,12 +233,34 @@ impl DomainServices { // A cursor beyond what the feed has written would let a client mark // itself caught up with events that do not exist yet, and then be // silently behind when they arrive. - let latest: i64 = sqlx::query_scalar( - "SELECT COALESCE(MAX(cursor), 0) FROM library_event WHERE library_id=?", + // + // Scoped by membership as well, and that is the project's rule rather + // than a leak being closed: today both paths out of this function + // answer 422, so a caller cannot tell "beyond the feed" from "not your + // library" whichever way this reads. But tenancy lives in the query + // here, never in a check the query trusts — `library_changes` keeps a + // redundant membership predicate for the same reason, so that removing + // one guard cannot quietly widen another. + // The maximum is a *subquery* rather than an aggregate over the join, + // and that is not style. `SELECT MAX(...) FROM ... WHERE ` + // returns one row holding NULL, so `COALESCE(..., 0)` would hand back + // `Some(0)` for a library the caller cannot see and the predicate above + // would decide nothing. Non-aggregate over `library_member`, zero + // matching rows means zero rows returned, which is the answer wanted. + let latest: Option = sqlx::query_scalar( + "SELECT COALESCE( \ + (SELECT MAX(e.cursor) FROM library_event e WHERE e.library_id=m.library_id), 0) \ + FROM library_member m JOIN device d ON d.user_id=m.user_id \ + WHERE m.library_id=? AND m.user_id=? AND d.id=? AND d.revoked_at IS NULL", ) .bind(library_id.to_string()) - .fetch_one(&mut *tx) + .bind(user_id.to_string()) + .bind(device_id.to_string()) + .fetch_optional(&mut *tx) .await?; + let Some(latest) = latest else { + return Ok(false); + }; if cursor > latest { return Ok(false); } From c7de11faa9d70a98beedd5f32268c0c3ffe9564f Mon Sep 17 00:00:00 2001 From: InstaZDLL Date: Sun, 30 Aug 2026 19:41:59 +0200 Subject: [PATCH 5/5] fix(api): let a device acknowledge the watermark itself MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Valid. The ceiling asked the surviving rows how far a client could have read, and a purge does not move that backwards — so a feed trimmed to nothing answered 0 and refused an acknowledgement at a cursor `library_changes` happily accepts to read *from*. Two parts of the same feature disagreeing about the same number. It is `MAX(events_purged_through, MAX(cursor))` now, zero when neither exists. **The state is unreachable in production, and the test says so out loud.** `parse_positive_env` refuses a floor of zero, and a floor of one always leaves a row whose cursor is at or above the watermark — so the rows alone happen to be right, thanks to a bound enforced three files away in the configuration parser. That is the kind of correctness that stops being correct when somebody changes the other file, which is why the expression should not lean on it. The test therefore builds a configuration the server would refuse to start with, a floor of zero, and says in as many words that this is what it is doing. `Config::for_data_dir` does not validate, which is the only reason it can be written — and a guard no test can reach is a guard nobody can check. Removing the watermark from the ceiling fails it: 422 where it wants 204. Claude-Session: https://claude.ai/code/session_01GtAmdsaBCg8Cs2rvrdLD7Z Signed-off-by: InstaZDLL --- src/services/library_events.rs | 21 ++++++- tests/catalog.rs | 106 +++++++++++++++++++++++++++++++++ 2 files changed, 124 insertions(+), 3 deletions(-) diff --git a/src/services/library_events.rs b/src/services/library_events.rs index 4117f30..9a47922 100644 --- a/src/services/library_events.rs +++ b/src/services/library_events.rs @@ -247,10 +247,25 @@ impl DomainServices { // `Some(0)` for a library the caller cannot see and the predicate above // would decide nothing. Non-aggregate over `library_member`, zero // matching rows means zero rows returned, which is the answer wanted. + // The watermark counts as well as the surviving rows, because the + // question is "the furthest position a client could legitimately have + // reached" and a purge does not move that backwards. Read from the rows + // alone, a feed trimmed to nothing would answer 0 and refuse an + // acknowledgement at a cursor `library_changes` accepts to read from — + // the two would disagree about the same number. + // + // Unreachable while `min_events` must be positive, since a floor of one + // leaves a row whose cursor is at or above the watermark. That is an + // invariant enforced three files away in `parse_positive_env`, and this + // is the expression that does not depend on it. let latest: Option = sqlx::query_scalar( - "SELECT COALESCE( \ - (SELECT MAX(e.cursor) FROM library_event e WHERE e.library_id=m.library_id), 0) \ - FROM library_member m JOIN device d ON d.user_id=m.user_id \ + "SELECT MAX( \ + l.events_purged_through, \ + COALESCE((SELECT MAX(e.cursor) FROM library_event e \ + WHERE e.library_id=m.library_id), 0)) \ + FROM library_member m \ + JOIN device d ON d.user_id=m.user_id \ + JOIN library l ON l.id=m.library_id \ WHERE m.library_id=? AND m.user_id=? AND d.id=? AND d.revoked_at IS NULL", ) .bind(library_id.to_string()) diff --git a/tests/catalog.rs b/tests/catalog.rs index ad20271..0a373e4 100644 --- a/tests/catalog.rs +++ b/tests/catalog.rs @@ -2114,3 +2114,109 @@ async fn an_acknowledgement_records_a_device_without_holding_the_purge_back() { "already stranded is not stranded again — a bill that only grows is not a bill" ); } + +/// The acknowledgement's ceiling counts the watermark, not only the rows. +/// +/// **The configuration here is one the server would refuse to start with.** +/// `parse_positive_env` rejects a floor of zero, so a feed can never actually +/// be trimmed to nothing in production, and this state is unreachable. The +/// guard exists anyway because "the furthest position a client could have +/// reached" should not be an expression that happens to be right thanks to a +/// bound enforced three files away — and a test that never reaches the branch +/// is a guard nobody can check. `Config::for_data_dir` does not validate, which +/// is what lets this be written at all. +#[tokio::test] +async fn a_cursor_at_the_watermark_is_acknowledged_even_with_the_feed_emptied() { + let temp = tempfile::tempdir().unwrap(); + let mut config = waveflow_server::Config::for_data_dir(temp.path().join("data")); + config.library_event_retention.min_events = 0; + config.library_event_retention.days = 30; + let state = waveflow_server::initialize(&config).await.unwrap(); + + let password = security::generate_token("test-password-"); + let hash = security::hash_password(&password).unwrap(); + let owner = state + .db + .create_account("watermark-ack", &hash, AccountRole::Admin, now_ms()) + .await + .unwrap(); + let music = config.data_dir.join("watermark-ack-music"); + std::fs::create_dir_all(&music).unwrap(); + let library = state + .db + .create_library( + owner, + "Watermark", + &std::fs::canonicalize(&music).unwrap(), + LibraryVisibility::Private, + now_ms(), + ) + .await + .unwrap(); + + let scan = state + .db + .create_scan_job(library, Some(owner), "manual") + .await + .unwrap(); + state.db.start_scan_job(scan, 2, false).await.unwrap(); + for index in 0..2usize { + let mut input = catalog_input(index, "Nova Kern"); + input.title = format!("Track {index}"); + input.album = None; + input.album_artist = None; + state + .db + .apply_catalog_track(library, scan, &input, None, false) + .await + .unwrap(); + } + state.db.finish_scan_job(scan, 0).await.unwrap(); + + let now = now_ms(); + sqlx::query("UPDATE library_event SET changed_at=? WHERE library_id=?") + .bind(now - 400 * 24 * 60 * 60 * 1000i64) + .bind(library.to_string()) + .execute(state.db.pool()) + .await + .unwrap(); + state.services.purge_library_events(now).await.unwrap(); + + let remaining: i64 = + sqlx::query_scalar("SELECT COUNT(*) FROM library_event WHERE library_id=?") + .bind(library.to_string()) + .fetch_one(state.db.pool()) + .await + .unwrap(); + assert_eq!(remaining, 0, "the floor of zero let the feed empty"); + let watermark: i64 = sqlx::query_scalar("SELECT events_purged_through FROM library WHERE id=?") + .bind(library.to_string()) + .fetch_one(state.db.pool()) + .await + .unwrap(); + assert!(watermark > 0); + + // `library_changes` accepts reading from the watermark, so the + // acknowledgement has to accept recording it — the two must not disagree + // about the same number. + let router = waveflow_server::app(&config, state.clone()); + let (token, device) = login_session(&router, "watermark-ack", &password).await; + assert!(state + .services + .library_changes(owner, library, watermark, 500) + .await + .is_ok()); + let response = router + .oneshot( + Request::put(format!("/api/v2/libraries/{library}/events/ack")) + .header("authorization", format!("Bearer {token}")) + .header("content-type", "application/json") + .body(Body::from( + serde_json::json!({ "device_id": device, "cursor": watermark }).to_string(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NO_CONTENT); +}