Skip to content
4 changes: 4 additions & 0 deletions src/library/scanner/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
});
}

Expand Down
8 changes: 7 additions & 1 deletion src/services/playlists/edit.rs
Original file line number Diff line number Diff line change
Expand Up @@ -234,7 +234,13 @@ impl DomainServices {
&serde_json::json!({
"id": id,
"name": name.map(str::trim).unwrap_or(&current.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,
}),
Expand Down
28 changes: 24 additions & 4 deletions src/services/smart_playlists/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
29 changes: 29 additions & 0 deletions src/services/uploads/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
///
Expand Down
25 changes: 12 additions & 13 deletions src/services/uploads/sweep.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -155,7 +153,7 @@ impl DomainServices {
root: &str,
session_id: Uuid,
) -> Result<bool, ServiceError> {
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),
Expand All @@ -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.
Expand Down
26 changes: 4 additions & 22 deletions src/services/uploads/transfer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}

Expand Down
5 changes: 4 additions & 1 deletion src/storage/catalog/apply.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down
4 changes: 3 additions & 1 deletion src/storage/catalog/overrides.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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())
Expand Down
85 changes: 85 additions & 0 deletions tests/catalog/tags.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::<Vec<_>>();
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"
);
}
14 changes: 14 additions & 0 deletions tests/native_api/corrections.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading
Loading