diff --git a/src/library/scanner/mod.rs b/src/library/scanner/mod.rs index b212b6ba..59d017ee 100644 --- a/src/library/scanner/mod.rs +++ b/src/library/scanner/mod.rs @@ -282,6 +282,10 @@ impl ScanManager { message: Some(message), }); } + // The job is over, whichever way: dropping its sender closes the + // channel, so every watcher reads what was sent and then sees the + // end, and a watcher who comes later finds no channel to wait on. + manager.events.remove(&scan_id); }); } diff --git a/src/services/playlists/edit.rs b/src/services/playlists/edit.rs index 2fd71cba..7a8a312e 100644 --- a/src/services/playlists/edit.rs +++ b/src/services/playlists/edit.rs @@ -234,7 +234,13 @@ impl DomainServices { &serde_json::json!({ "id": id, "name": name.map(str::trim).unwrap_or(¤t.name), - "comment": comment.or(current.comment.as_deref()), + // What the statement above stored: a clear wins over any + // comment sent beside it. + "comment": if clear.comment { + None + } else { + comment.or(current.comment.as_deref()) + }, "public": public.unwrap_or(current.public), "track_ids": ids, }), diff --git a/src/services/smart_playlists/tests.rs b/src/services/smart_playlists/tests.rs index 67763318..3db4d966 100644 --- a/src/services/smart_playlists/tests.rs +++ b/src/services/smart_playlists/tests.rs @@ -134,13 +134,33 @@ fn the_sort_codes_match_the_statement() { "s.duration_ms", "s.bitrate", ]; + // The direction too: the term after `END`, up to the next comma, says + // `ASC` for an odd code and `DESC` for the even one after it. Two terms + // swapped would keep every column in place and reverse every sort. + let direction = |code: usize, column: &str| { + let head = format!("p.sort_key = {code} THEN {column} END"); + let at = EVALUATE + .find(&head) + .unwrap_or_else(|| panic!("code {code} does not sort by {column}")); + let tail = &EVALUATE[at + head.len()..]; + let term = &tail[..tail.find(',').unwrap_or(tail.len())]; + if term.contains(" DESC ") { + "DESC" + } else if term.contains(" ASC ") { + "ASC" + } else { + panic!("code {code} names no direction: {term:?}") + } + }; for (index, column) in columns.iter().enumerate() { let ascending = index * 2 + 1; - assert!( - EVALUATE.contains(&format!("p.sort_key = {ascending} THEN {column} END")), - "code {ascending} does not sort by {column}" + assert_eq!(direction(ascending, column), "ASC", "code {ascending}"); + assert_eq!( + direction(ascending + 1, column), + "DESC", + "code {}", + ascending + 1 ); - assert!(EVALUATE.contains(&format!("p.sort_key = {} THEN {column} END", ascending + 1))); } let random = parse(json!({"sort": "random", "order": "desc"})).unwrap(); assert_eq!(random.sort_code(), SORT_RANDOM); diff --git a/src/services/uploads/mod.rs b/src/services/uploads/mod.rs index d1397da1..0b2122bd 100644 --- a/src/services/uploads/mod.rs +++ b/src/services/uploads/mod.rs @@ -19,6 +19,35 @@ mod transfer; /// problems that are never posed rather than problems that are solved. pub(crate) const MANAGED_DIR: &str = ".waveflow-managed"; +/// Where a session's fragments accumulate: the destination directory itself, +/// under a name carrying no audio extension. +/// +/// Both halves matter. Same directory means same filesystem, so the rename +/// at commit is atomic rather than a copy with a window during which a +/// truncated file exists. And the walk skips it not because it is hidden — +/// nothing about the walk skips hidden directories — but because it filters +/// on extension, which is a rule that already exists and cannot be broken +/// by someone rewriting the traversal. +/// +/// One function for the transfer that writes there and the sweep that cleans +/// up after it: two spellings of the name could drift, and a sweep looking +/// under the wrong one drops the session and leaves the file. +fn staging_path(root: &std::path::Path, session_id: Uuid) -> std::path::PathBuf { + root.join(MANAGED_DIR).join(format!("{session_id}.part")) +} + +/// The name a file keeps, once the bytes are known to be what was promised. +/// +/// By hash rather than by tags: tags move — a correction, a retag, a later +/// re-identification — and a path built from them would leave the server +/// choosing between moving the file and keeping a name that lies. The +/// project already answered that question in the other direction, where a +/// correction never rewrites the file. It keeps its extension because the +/// walk recognises a file by nothing else. +fn relative_final(hash: &str, extension: &str) -> String { + format!("{MANAGED_DIR}/{hash}.{extension}") +} + impl DomainServices { /// How many sessions this process is currently coordinating. /// diff --git a/src/services/uploads/sweep.rs b/src/services/uploads/sweep.rs index 41ea803f..fc27d32f 100644 --- a/src/services/uploads/sweep.rs +++ b/src/services/uploads/sweep.rs @@ -90,9 +90,7 @@ impl DomainServices { let declared_hash: String = row.try_get("declared_hash")?; let extension: String = row.try_get("extension")?; let root: String = row.try_get("root_path")?; - let path = std::path::Path::new(&root) - .join(MANAGED_DIR) - .join(format!("{session_id}.part")); + let path = staging_path(std::path::Path::new(&root), session_id); let lock = self.session_lock(session_id); let held = lock.lock().await; @@ -155,7 +153,7 @@ impl DomainServices { root: &str, session_id: Uuid, ) -> Result { - let relative = format!("{MANAGED_DIR}/{declared_hash}.{extension}"); + let relative = relative_final(declared_hash, extension); let path = std::path::Path::new(root).join(&relative); match tokio::fs::try_exists(&path).await { Ok(false) => return Ok(true), @@ -173,23 +171,24 @@ impl DomainServices { return Ok(false); } } - let claimed: i64 = sqlx::query_scalar( - "SELECT EXISTS(SELECT 1 FROM track WHERE library_id=? AND relative_path=?)", + // Both guards in one statement, so they read one state of the + // database. Asked one after the other, a commit landing between them — + // its track inserted and its session deleted in a single transaction — + // answered "no track yet" to the first and "no session any more" to the + // second, and the sweep deleted the file of the track just committed. + let others: i64 = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM track WHERE library_id=? AND relative_path=?) \ + OR EXISTS(SELECT 1 FROM upload_session \ + WHERE library_id=? AND declared_hash=? AND id<>?)", ) .bind(library_id) .bind(&relative) - .fetch_one(self.db.pool()) - .await?; - let contested: i64 = sqlx::query_scalar( - "SELECT EXISTS(SELECT 1 FROM upload_session \ - WHERE library_id=? AND declared_hash=? AND id<>?)", - ) .bind(library_id) .bind(declared_hash) .bind(session_id.to_string()) .fetch_one(self.db.pool()) .await?; - if claimed != 0 || contested != 0 { + if others != 0 { // Somebody else's, either already or possibly. The row goes anyway: // keeping it would hold a reservation against a file this session // no longer owns. diff --git a/src/services/uploads/transfer.rs b/src/services/uploads/transfer.rs index ef85af13..3f59afbe 100644 --- a/src/services/uploads/transfer.rs +++ b/src/services/uploads/transfer.rs @@ -17,32 +17,14 @@ struct SessionTarget { } impl SessionTarget { - /// Where the fragments accumulate: the destination directory itself, under - /// a name carrying no audio extension. - /// - /// Both halves matter. Same directory means same filesystem, so the rename - /// at commit is atomic rather than a copy with a window during which a - /// truncated file exists. And the walk skips it not because it is hidden — - /// nothing about the walk skips hidden directories — but because it filters - /// on extension, which is a rule that already exists and cannot be broken - /// by someone rewriting the traversal. + /// Where this session's fragments accumulate: see [`staging_path`]. fn staging_path(&self, session_id: Uuid) -> std::path::PathBuf { - self.root - .join(MANAGED_DIR) - .join(format!("{session_id}.part")) + staging_path(&self.root, session_id) } - /// The name the file keeps, once the bytes are known to be what was - /// promised. - /// - /// By hash rather than by tags: tags move — a correction, a retag, a later - /// re-identification — and a path built from them would leave the server - /// choosing between moving the file and keeping a name that lies. The - /// project already answered that question in the other direction, where a - /// correction never rewrites the file. It keeps its extension because the - /// walk recognises a file by nothing else. + /// The name this session's file keeps: see [`relative_final`]. fn relative_final(&self, hash: &str) -> String { - format!("{MANAGED_DIR}/{hash}.{}", self.extension) + relative_final(hash, &self.extension) } } diff --git a/src/storage/catalog/apply.rs b/src/storage/catalog/apply.rs index d1ae4577..7a21d080 100644 --- a/src/storage/catalog/apply.rs +++ b/src/storage/catalog/apply.rs @@ -448,9 +448,12 @@ impl Database { .bind(track_id.to_string()) .execute(&mut **tx) .await?; + // A file may name one genre twice, or in two spellings canonicalisation + // folds together; the pair is the key, so the repeat is not an error. for genre_id in &genre_ids { sqlx::query( - "INSERT INTO track_genre (track_id, genre_id, library_id) VALUES (?, ?, ?)", + "INSERT INTO track_genre (track_id, genre_id, library_id) VALUES (?, ?, ?) \ + ON CONFLICT (track_id, genre_id) DO NOTHING", ) .bind(track_id.to_string()) .bind(genre_id.to_string()) diff --git a/src/storage/catalog/overrides.rs b/src/storage/catalog/overrides.rs index 32838ad3..3bff4305 100644 --- a/src/storage/catalog/overrides.rs +++ b/src/storage/catalog/overrides.rs @@ -108,8 +108,10 @@ pub(crate) async fn apply_track_override_lists( .await?; for name in names { let genre_id = upsert_genre(tx, library_id, name, now).await?; + // A correction may repeat a genre as the tags may: one row. sqlx::query( - "INSERT INTO track_genre (track_id, genre_id, library_id) VALUES (?, ?, ?)", + "INSERT INTO track_genre (track_id, genre_id, library_id) VALUES (?, ?, ?) \ + ON CONFLICT (track_id, genre_id) DO NOTHING", ) .bind(track_id.to_string()) .bind(genre_id.to_string()) diff --git a/tests/catalog/tags.rs b/tests/catalog/tags.rs index 40eb6f94..514b88e1 100644 --- a/tests/catalog/tags.rs +++ b/tests/catalog/tags.rs @@ -632,3 +632,88 @@ async fn genre_matching_is_canonical_on_every_surface() { assert_eq!(album["song"][0]["displayAlbumArtist"], "Nine Mics"); assert_eq!(album["song"][0]["albumArtists"][0]["name"], "Nine Mics"); } + +/// A file that names one genre twice — `Rock; Rock`, or `Rock; rock`, which +/// canonicalisation folds into one — is one genre on its track. +/// +/// `track_genre` keys on the pair, and the scan used to insert the repeat and +/// fail on that key. The failure rolled back the whole batch it was applied in, +/// so a single odd tag kept every other file of the batch out of the library. +#[tokio::test] +async fn a_repeated_genre_is_one_genre_and_spoils_no_batch() { + let (_temp, config, state) = test_app().await; + let (owner, _) = admin_with_api_token(&state, "repeat-owner").await; + let music = config.data_dir.join("repeat-music"); + std::fs::create_dir_all(&music).unwrap(); + let library = state + .db + .create_library( + owner, + "Repeats", + &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(); + + let applies = [ + ("Echo", "Rock; Rock"), + ("Fold", "Rock; rock"), + ("Plain", "Pop"), + ] + .into_iter() + .enumerate() + .map(|(index, (title, genre))| { + let mut input = browse_input( + 600 + index, + title, + "Repeater", + "Twice Told", + Some(index as i64 + 1), + Some(1), + ); + input.relative_path = format!("repeat-{index}.flac"); + input.quick_hash = format!("{:064x}", index + 61_000); + input.full_hash = format!("{:064x}", index + 62_000); + input.genre = Some(genre.into()); + waveflow_server::storage::catalog::CatalogApply { + input, + existing_id: None, + moved: false, + origin_device_id: None, + } + }) + .collect::>(); + state + .db + .apply_catalog_tracks(library, scan, &applies) + .await + .expect("a repeated genre must not fail the batch"); + state.db.finish_scan_job(scan, 0).await.unwrap(); + + let rows: Vec<(String, i64)> = sqlx::query_as( + "SELECT t.title, COUNT(tg.genre_id) FROM track t \ + LEFT JOIN track_genre tg ON tg.track_id = t.id \ + WHERE t.library_id = ? GROUP BY t.id ORDER BY t.title", + ) + .bind(library.to_string()) + .fetch_all(state.db.pool()) + .await + .unwrap(); + assert_eq!( + rows, + vec![ + ("Echo".to_owned(), 1), + ("Fold".to_owned(), 1), + ("Plain".to_owned(), 1), + ], + "every file of the batch is in, each with its one genre" + ); +} diff --git a/tests/native_api/corrections.rs b/tests/native_api/corrections.rs index 748dc78b..1dc7cd96 100644 --- a/tests/native_api/corrections.rs +++ b/tests/native_api/corrections.rs @@ -480,6 +480,20 @@ async fn a_corrected_artist_agrees_with_itself_everywhere() { ); assert_eq!(other_dropped[0].genres, vec!["Ambient".to_owned()]); + // A genre named twice, in any case, is one genre. `track_genre` keys on the + // pair, and the correction used to insert the second spelling and fail on + // that key, refusing the whole edit. + patch_body(serde_json::json!({ "genres": ["Jazz", "Jazz", "jazz"] })).await; + let repeated = state + .services + .songs_by_ids(owner, &[track_id]) + .await + .unwrap(); + // One genre; which of its spellings it is shown under is the genre's own + // business, not this correction's. + assert_eq!(repeated[0].genres.len(), 1); + assert!(repeated[0].genres[0].eq_ignore_ascii_case("jazz")); + // Removing the correction hands the track back to its file. The rows the // correction wrote replaced what the tags said, so this can only be right // by re-reading them — and it must not wait for a scan. Removed by an diff --git a/tests/native_api/playlists.rs b/tests/native_api/playlists.rs index d885a0a6..e0b29f99 100644 --- a/tests/native_api/playlists.rs +++ b/tests/native_api/playlists.rs @@ -536,3 +536,80 @@ async fn a_playlist_entry_moves_to_the_position_it_was_shown_at() { .await; assert_eq!(refused.status(), StatusCode::NOT_FOUND); } + +/// A comment cleared on one device is cleared on the others. +/// +/// The statement writes `NULL` when `clear.comment` is set, whatever comment +/// came with it; the journal entry a device mirrors from used to carry the old +/// comment instead (or the new one sent alongside the clear), so every other +/// device kept a comment the server no longer had. +#[tokio::test] +async fn a_cleared_comment_reaches_the_journal_as_cleared() { + let (_temp, config, state) = test_app().await; + let (_owner, token) = admin_with_api_token(&state, "comment-owner").await; + let router = waveflow_server::app(&config, state.clone()); + let created = json_body( + send_json( + &router, + "POST", + "/api/v2/playlists", + &token, + Some(serde_json::json!({ "name": "Notes", "track_ids": [] })), + ) + .await, + ) + .await; + let uri = format!("/api/v2/playlists/{}", created["id"].as_str().unwrap()); + // Through the route a device uses, so the `clear` list's mapping onto + // the service is under test as well. + let update = |comment: Option<&'static str>, clear: bool| { + let router = router.clone(); + let token = token.clone(); + let uri = uri.clone(); + async move { + let mut body = serde_json::json!({ "comment": comment }); + if clear { + body["clear"] = serde_json::json!(["comment"]); + } + let response = send_json(&router, "PATCH", &uri, &token, Some(body)).await; + assert_eq!(response.status(), StatusCode::OK); + json_body(response).await + } + }; + let journalled = || { + let state = state.clone(); + async move { + let payload: String = sqlx::query_scalar( + "SELECT payload_json FROM sync_event WHERE entity_type='playlist' \ + ORDER BY cursor DESC LIMIT 1", + ) + .fetch_one(state.db.pool()) + .await + .unwrap(); + serde_json::from_str::(&payload).unwrap()["comment"].clone() + } + }; + + update(Some("late night"), false).await; + assert_eq!(journalled().await, "late night"); + + // Cleared on its own. + let cleared = update(None, true).await; + assert!(cleared["comment"].is_null(), "{cleared}"); + assert_eq!( + journalled().await, + serde_json::Value::Null, + "the clear is mirrored" + ); + + // Cleared with a comment beside it: the store keeps neither, so neither + // may the journal. + update(Some("again"), false).await; + let cleared = update(Some("ignored"), true).await; + assert!(cleared["comment"].is_null(), "{cleared}"); + assert_eq!( + journalled().await, + serde_json::Value::Null, + "the journal agrees with the store" + ); +} diff --git a/tests/scanner/scheduling.rs b/tests/scanner/scheduling.rs index 75b6811c..eda7e164 100644 --- a/tests/scanner/scheduling.rs +++ b/tests/scanner/scheduling.rs @@ -403,3 +403,53 @@ async fn a_pending_full_scan_is_not_spent_by_a_scoped_one() { "the request was answered by a full walk" ); } + +/// A scan's event stream ends once the scan has. The channel behind it used to +/// outlive the job — one per scan, never dropped — so a watcher that came after +/// the last progress got the snapshot and then nothing, kept open by +/// keep-alives until it hung up, and the server held every finished scan's +/// channel for good. +#[tokio::test] +async fn a_finished_scan_s_event_stream_ends() { + let (_temp, config, state) = test_app().await; + let (owner, token) = admin_with_api_token(&state, "stream-owner").await; + let music = config.data_dir.join("stream-music"); + std::fs::create_dir_all(&music).unwrap(); + write_test_wav(&music.join("One.wav")); + let root = std::fs::canonicalize(&music).unwrap(); + let library_id = state + .db + .create_library(owner, "Stream", &root, LibraryVisibility::Private, now_ms()) + .await + .unwrap(); + let library = LibraryRecord { + id: library_id, + name: "Stream".into(), + root_path: root, + }; + let scan = scan_once(&state, owner, library).await; + let router = waveflow_server::app(&config, state.clone()); + + let response = tower::ServiceExt::oneshot( + router, + axum::http::Request::get(format!("/api/v2/scans/{scan}/events")) + .header("authorization", format!("Bearer {token}")) + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), axum::http::StatusCode::OK); + // The whole body, which only exists if the stream ends: a stream that + // stays open never yields it, and the deadline says so. + let body = tokio::time::timeout( + std::time::Duration::from_secs(10), + axum::body::to_bytes(response.into_body(), usize::MAX), + ) + .await + .expect("the event stream of a finished scan must end") + .unwrap(); + let text = String::from_utf8(body.to_vec()).unwrap(); + assert!(text.contains("event: snapshot"), "{text}"); + assert!(text.contains("\"completed\""), "{text}"); +}