diff --git a/.sqlx/query-100e5e9297a9a917eb107c673ac492efe0df0123d9def212d84b88e57aaa0484.json b/.sqlx/query-100e5e9297a9a917eb107c673ac492efe0df0123d9def212d84b88e57aaa0484.json new file mode 100644 index 0000000..8d96b35 --- /dev/null +++ b/.sqlx/query-100e5e9297a9a917eb107c673ac492efe0df0123d9def212d84b88e57aaa0484.json @@ -0,0 +1,98 @@ +{ + "db_name": "SQLite", + "query": "SELECT required_audio AS \"required_audio!: String\",\n dub_blacklist AS \"dub_blacklist!: String\",\n hdr_rules AS \"hdr_rules!: String\",\n size_bands AS \"size_bands!: String\",\n resolution_pref AS \"resolution_pref!: String\",\n source_weights AS \"source_weights!: String\",\n score_weights AS \"score_weights!: String\"\n FROM policies WHERE id = ?", + "describe": { + "columns": [ + { + "name": "required_audio!: String", + "ordinal": 0, + "type_info": "Text", + "origin": { + "Table": { + "table": "policies", + "name": "required_audio" + } + } + }, + { + "name": "dub_blacklist!: String", + "ordinal": 1, + "type_info": "Text", + "origin": { + "Table": { + "table": "policies", + "name": "dub_blacklist" + } + } + }, + { + "name": "hdr_rules!: String", + "ordinal": 2, + "type_info": "Text", + "origin": { + "Table": { + "table": "policies", + "name": "hdr_rules" + } + } + }, + { + "name": "size_bands!: String", + "ordinal": 3, + "type_info": "Text", + "origin": { + "Table": { + "table": "policies", + "name": "size_bands" + } + } + }, + { + "name": "resolution_pref!: String", + "ordinal": 4, + "type_info": "Text", + "origin": { + "Table": { + "table": "policies", + "name": "resolution_pref" + } + } + }, + { + "name": "source_weights!: String", + "ordinal": 5, + "type_info": "Text", + "origin": { + "Table": { + "table": "policies", + "name": "source_weights" + } + } + }, + { + "name": "score_weights!: String", + "ordinal": 6, + "type_info": "Text", + "origin": { + "Table": { + "table": "policies", + "name": "score_weights" + } + } + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false, + false, + false, + false, + false, + false, + false + ] + }, + "hash": "100e5e9297a9a917eb107c673ac492efe0df0123d9def212d84b88e57aaa0484" +} diff --git a/.sqlx/query-a28caf97dfb6a7e6b69543df0693e15c746702c36a0d4068fa70d28b950e7562.json b/.sqlx/query-a28caf97dfb6a7e6b69543df0693e15c746702c36a0d4068fa70d28b950e7562.json new file mode 100644 index 0000000..5bdcdb9 --- /dev/null +++ b/.sqlx/query-a28caf97dfb6a7e6b69543df0693e15c746702c36a0d4068fa70d28b950e7562.json @@ -0,0 +1,26 @@ +{ + "db_name": "SQLite", + "query": "SELECT id AS \"id!: i64\" FROM roots WHERE policy_id = ?", + "describe": { + "columns": [ + { + "name": "id!: i64", + "ordinal": 0, + "type_info": "Integer", + "origin": { + "Table": { + "table": "roots", + "name": "id" + } + } + } + ], + "parameters": { + "Right": 1 + }, + "nullable": [ + false + ] + }, + "hash": "a28caf97dfb6a7e6b69543df0693e15c746702c36a0d4068fa70d28b950e7562" +} diff --git a/DESIGN.md b/DESIGN.md index b730786..fa9f0bd 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -201,14 +201,23 @@ both directions — `only_4k` tightens, `allow_english_audio` loosens. **Stored verdicts follow the effective policy.** A release's verdict is stamped by the search that found it, and both §9.3's deck and the daemon's -manual-grab gate read that stored column. So the three operator actions that +manual-grab gate read that stored column. So the four operator actions that change a title's effective policy — editing its overrides, moving it to a -root with a different policy, and pointing a root at a different policy — -re-derive the stored verdicts of everything they touch, inline in the same -request. The operator is never left reading a verdict computed under a -policy that no longer applies. Inline is affordable even root-wide: -re-evaluation is pure and in-memory, and a row whose verdict does not move -is not rewritten, so a library-sized repoint costs reads, not writes. +root with a different policy, pointing a root at a different policy, and +editing the contents of a policy some root points at — re-derive the stored +verdicts of everything they touch, inline in the same request. The operator +is never left reading a verdict computed under a policy that no longer +applies. + +The fourth is the widest: a repoint moves one library, a policy edit moves +every library sharing the policy. Inline still holds there. Re-evaluation is +pure and in-memory, a row whose verdict does not move is not rewritten, and +the ceiling is the whole database rather than something that grows with the +number of roots — roots partition titles, and a title has exactly one root. +Measured over 2000 titles and 10 000 stored releases split across two roots +sharing one policy: 0.36 s when the edit moves no verdict, 2.7 s in the +pathological case where it moves all 10 000. A rename changes no rule and +re-derives nothing. ### 5.2 Language diff --git a/crates/arr-api/src/policies.rs b/crates/arr-api/src/policies.rs index a7a07d0..6ebb77c 100644 --- a/crates/arr-api/src/policies.rs +++ b/crates/arr-api/src/policies.rs @@ -173,6 +173,33 @@ struct PolicyColumns { score_weights: String, } +/// Everything in a policy row that a verdict depends on — the name is the +/// one column that does not. +#[derive(PartialEq, Eq)] +struct PolicyRules { + required_audio: String, + dub_blacklist: String, + hdr_rules: String, + size_bands: String, + resolution_pref: String, + source_weights: String, + score_weights: String, +} + +impl PolicyColumns { + fn rules(&self) -> PolicyRules { + PolicyRules { + required_audio: self.required_audio.clone(), + dub_blacklist: self.dub_blacklist.clone(), + hdr_rules: self.hdr_rules.clone(), + size_bands: self.size_bands.clone(), + resolution_pref: self.resolution_pref.clone(), + source_weights: self.source_weights.clone(), + score_weights: self.score_weights.clone(), + } + } +} + fn column( column: &'static str, value: &str, @@ -376,6 +403,8 @@ pub async fn update( input.validate().map_err(ApiError::Invalid)?; let name = input.name.trim().to_owned(); let columns = input.into_columns(name)?; + let after = columns.rules(); + let before = stored_rules(&state, id).await?; let result = sqlx::query!( r#"UPDATE policies SET name = ?, required_audio = ?, dub_blacklist = ?, hdr_rules = ?, @@ -405,9 +434,44 @@ pub async fn update( if result.rows_affected() == 0 { return Err(ApiError::PolicyNotFound); } + // Every verdict stored under every root pointing here was reached under + // the rules this write just replaced; §9.3's deck and the grab gate both + // read them (`reclassify`). A rename leaves the rules alone, so it walks + // nothing. + if before.is_none_or(|before| before != after) { + crate::reclassify::policy(&state, id).await?; + } Ok(Json(load_policy(&state, id).await?)) } +/// The rule columns of a policy as they stand, or `None` when there is no +/// such row. Compared against what the write is about to store, so an edit +/// that only moves the name does not re-derive a library. +async fn stored_rules(state: &AppState, id: i64) -> Result, ApiError> { + let row = sqlx::query!( + r#"SELECT required_audio AS "required_audio!: String", + dub_blacklist AS "dub_blacklist!: String", + hdr_rules AS "hdr_rules!: String", + size_bands AS "size_bands!: String", + resolution_pref AS "resolution_pref!: String", + source_weights AS "source_weights!: String", + score_weights AS "score_weights!: String" + FROM policies WHERE id = ?"#, + id + ) + .fetch_optional(pool(state)?) + .await?; + Ok(row.map(|row| PolicyRules { + required_audio: row.required_audio, + dub_blacklist: row.dub_blacklist, + hdr_rules: row.hdr_rules, + size_bands: row.size_bands, + resolution_pref: row.resolution_pref, + source_weights: row.source_weights, + score_weights: row.score_weights, + })) +} + #[utoipa::path( delete, path = "/api/policies/{policy_id}", tag = "policies", params(("policy_id" = i64, Path, description = "Policy row id")), @@ -450,7 +514,7 @@ mod tests { use crate::{router, Upstreams}; use axum::http::StatusCode; - async fn application() -> (tempfile::TempDir, String) { + async fn application() -> (tempfile::TempDir, AppState, String) { let dir = tempfile::tempdir().expect("tempdir"); let database = arr_db::Db::connect(dir.path().join("arr.db")) .await @@ -466,9 +530,9 @@ mod tests { .await .expect("bind"); let address = listener.local_addr().expect("address"); - let app = router(state); + let app = router(state.clone()); tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") }); - (dir, format!("http://{address}")) + (dir, state, format!("http://{address}")) } fn valid_input(name: &str) -> serde_json::Value { @@ -501,7 +565,7 @@ mod tests { #[tokio::test] async fn crud_round_trips_a_policy() { - let (_dir, base) = application().await; + let (_dir, _state, base) = application().await; let created: serde_json::Value = create(&base, valid_input("test policy")) .await @@ -557,7 +621,7 @@ mod tests { #[tokio::test] async fn a_referenced_policy_refuses_to_die() { - let (_dir, base) = application().await; + let (_dir, _state, base) = application().await; let roots: Vec = reqwest::get(format!("{base}/api/roots")) .await .expect("roots") @@ -582,7 +646,7 @@ mod tests { #[tokio::test] async fn an_unknown_resolution_is_a_422_naming_the_field() { - let (_dir, base) = application().await; + let (_dir, _state, base) = application().await; let mut payload = valid_input("bad bands"); payload["size_bands"]["1440p"] = serde_json::json!({ "floor_gib": 2, "target_gib": 6, "penalty_points_per_gib_over": 60 }); @@ -596,7 +660,7 @@ mod tests { #[tokio::test] async fn every_field_validates_by_name() { - let (_dir, base) = application().await; + let (_dir, _state, base) = application().await; let with = |patch: &dyn Fn(&mut serde_json::Value)| { let mut payload = valid_input("validation probe"); patch(&mut payload); @@ -645,7 +709,7 @@ mod tests { #[tokio::test] async fn malformed_json_is_422_not_400_or_500() { - let (_dir, base) = application().await; + let (_dir, _state, base) = application().await; let client = reqwest::Client::new(); let response = client .post(format!("{base}/api/policies")) @@ -667,8 +731,267 @@ mod tests { #[tokio::test] async fn a_duplicate_name_conflicts() { - let (_dir, base) = application().await; + let (_dir, _state, base) = application().await; let response = create(&base, valid_input("Movies — main")).await; assert_eq!(response.status(), StatusCode::CONFLICT); } + + async fn policy_id_named(base: &str, name: &str) -> i64 { + let policies: Vec = reqwest::get(format!("{base}/api/policies")) + .await + .expect("policies") + .json() + .await + .expect("policies json"); + policies + .iter() + .find(|policy| policy["name"] == name) + .and_then(|policy| policy["id"].as_i64()) + .expect("policy id") + } + + /// A movie under `root_id`, with one English 1080p release stamped + /// `verdict` as a search would have stamped it. + async fn movie_with_release( + state: &AppState, + base: &str, + root_id: i64, + tmdb_id: i64, + verdict: &str, + ) -> i64 { + let movie: serde_json::Value = reqwest::Client::new() + .post(format!("{base}/api/movies")) + .json(&serde_json::json!({ + "tmdb_id": tmdb_id, "title": format!("Title {tmdb_id}"), "year": 2024, + "original_language": "en", "root_id": root_id, "overrides": {} + })) + .send() + .await + .expect("create movie") + .json() + .await + .expect("movie json"); + let movie_id = movie["id"].as_i64().expect("movie id"); + let name = format!("Title {tmdb_id} 2024 1080p WEB-DL ENGLISH x264-GROUP"); + let release_id = stamped_release(state, &name, 6_i64 * (1 << 30), verdict).await; + sqlx::query("INSERT INTO movie_releases (movie_id, release_id) VALUES (?, ?)") + .bind(movie_id) + .bind(release_id) + .execute(state.database().expect("database").pool()) + .await + .expect("movie association"); + movie_id + } + + /// One release row, verdict stamped by hand. `waived` carries no rule, + /// which the `releases` CHECK allows — only `rejected` needs one. + async fn stamped_release(state: &AppState, name: &str, size: i64, verdict: &str) -> i64 { + let parsed = arr_parse::parse(name); + sqlx::query_scalar::<_, i64>( + "INSERT INTO releases (indexer_id, guid, name, size, seeders, download_url, parsed, score, verdict) + VALUES (7, ?, ?, ?, 40, 'url', ?, 0, ?) RETURNING id", + ) + .bind(name) + .bind(name) + .bind(size) + .bind(serde_json::to_string(&parsed).expect("parsed json")) + .bind(verdict) + .fetch_one(state.database().expect("database").pool()) + .await + .expect("release") + } + + /// A series under `root_id` with a ten-episode season 9 pack stamped + /// `waived`. Under its own policy the pack is eligible, so the stamp + /// only survives if nothing re-derived it. + async fn series_with_waived_pack(state: &AppState, base: &str, root_id: i64) -> i64 { + let client = reqwest::Client::new(); + let series: serde_json::Value = client + .post(format!("{base}/api/series")) + .json(&serde_json::json!({ + "tmdb_id": 82_728, "title": "Bluey", "year": 2018, + "original_language": "en", "root_id": root_id, + "auto_track": false + })) + .send() + .await + .expect("create series") + .json() + .await + .expect("series json"); + let series_id = series["id"].as_i64().expect("series id"); + let episodes: Vec = (1..=10) + .map(|number| { + serde_json::json!({ + "number": number, "title": format!("Episode {number}"), + "air_date": "2025-01-01" + }) + }) + .collect(); + let season: serde_json::Value = client + .post(format!("{base}/api/series/{series_id}/seasons")) + .json(&serde_json::json!({"number": 9, "episodes": episodes})) + .send() + .await + .expect("create season") + .json() + .await + .expect("season json"); + let season_id = season["id"].as_i64().expect("season id"); + // Ten episodes in 15 GiB: 1.5 GiB each, inside the 1080p band. + let release_id = stamped_release( + state, + "Bluey S09 1080p WEB-DL ENGLISH x264-GROUP", + 15_i64 * (1 << 30), + "waived", + ) + .await; + sqlx::query("INSERT INTO season_releases (season_id, release_id) VALUES (?, ?)") + .bind(season_id) + .bind(release_id) + .execute(state.database().expect("database").pool()) + .await + .expect("season association"); + series_id + } + + async fn verdict_at(url: String) -> serde_json::Value { + let releases: Vec = reqwest::get(url) + .await + .expect("releases") + .json() + .await + .expect("releases json"); + releases[0]["verdict"].clone() + } + + async fn movie_verdict(base: &str, movie_id: i64) -> serde_json::Value { + verdict_at(format!("{base}/api/movies/{movie_id}/releases")).await + } + + async fn root_id_of(state: &AppState, kind: &str, audience: &str) -> i64 { + sqlx::query_scalar("SELECT id FROM roots WHERE kind = ? AND audience = ?") + .bind(kind) + .bind(audience) + .fetch_one(state.database().expect("database").pool()) + .await + .expect("root id") + } + + /// The same policy document with a different required-audio rule, so an + /// English release that was eligible becomes a waiver. + fn requires_portuguese(name: &str) -> serde_json::Value { + let mut payload = valid_input(name); + payload["required_audio"] = serde_json::json!({ "require": "any_of", "langs": ["pt-PT"] }); + payload + } + + /// #246: editing a policy re-derives the stored verdicts of every title + /// under every root pointing at it (§5.1) — both libraries when two + /// roots share it, and nothing under a root that does not. + #[tokio::test] + async fn editing_a_policy_rederives_every_root_that_points_at_it() { + let (_dir, state, base) = application().await; + let client = reqwest::Client::new(); + let shared = policy_id_named(&base, "Movies — main").await; + let main_root = root_id_of(&state, "movie", "main").await; + let kids_root = root_id_of(&state, "movie", "kids").await; + + // Two roots on one policy: the edit has two libraries to reach + // rather than one. The path stays, so nothing on disk moves. + let response = client + .put(format!("{base}/api/roots/{kids_root}")) + .json(&serde_json::json!({ + "kind": "movie", "audience": "kids", + "path": "/mnt/media/movies/kids", "policy_id": shared, + })) + .send() + .await + .expect("repoint kids root"); + assert_eq!( + response.status(), + StatusCode::OK, + "{:?}", + response.text().await + ); + + let here = movie_with_release(&state, &base, main_root, 693_134, "eligible").await; + let there = movie_with_release(&state, &base, kids_root, 27_205, "eligible").await; + // On the TV main root, a different policy, so the edit must not + // reach it. Stamped against its own policy's answer, so a walk that + // did reach it would show. + let tv_root = root_id_of(&state, "tv", "main").await; + let series_id = series_with_waived_pack(&state, &base, tv_root).await; + + let response = client + .put(format!("{base}/api/policies/{shared}")) + .json(&requires_portuguese("Movies — main")) + .send() + .await + .expect("edit policy"); + assert_eq!( + response.status(), + StatusCode::OK, + "{:?}", + response.text().await + ); + + assert_eq!( + movie_verdict(&base, here).await, + "waived", + "the first root's library re-derives" + ); + assert_eq!( + movie_verdict(&base, there).await, + "waived", + "and so does the second root's, sharing the policy" + ); + assert_eq!( + verdict_at(format!("{base}/api/series/{series_id}/seasons/9/releases")).await, + "waived", + "a root on another policy keeps the verdict it was stamped with" + ); + } + + /// Renaming a policy changes no rule, so it re-derives nothing — the + /// stamped verdict survives even though the rules would not produce it. + #[tokio::test] + async fn renaming_a_policy_leaves_verdicts_alone() { + let (_dir, state, base) = application().await; + let client = reqwest::Client::new(); + let kids = policy_id_named(&base, "Movies — kids").await; + let kids_root = root_id_of(&state, "movie", "kids").await; + + // Write the rules through the API once, so the rename that follows + // stores byte-identical rule columns and the only change is the name. + let response = client + .put(format!("{base}/api/policies/{kids}")) + .json(&valid_input("Movies — kids")) + .send() + .await + .expect("normalise policy"); + assert_eq!(response.status(), StatusCode::OK); + + // Deliberately the wrong answer: these rules make an English release + // a waiver, so a re-derivation would move this row. + let movie = movie_with_release(&state, &base, kids_root, 157_336, "eligible").await; + + let response = client + .put(format!("{base}/api/policies/{kids}")) + .json(&valid_input("Movies — children")) + .send() + .await + .expect("rename policy"); + assert_eq!( + response.status(), + StatusCode::OK, + "{:?}", + response.text().await + ); + assert_eq!( + movie_verdict(&base, movie).await, + "eligible", + "a rename touches no rule, so it walks nothing" + ); + } } diff --git a/crates/arr-api/src/reclassify.rs b/crates/arr-api/src/reclassify.rs index e95d265..d4fe79a 100644 --- a/crates/arr-api/src/reclassify.rs +++ b/crates/arr-api/src/reclassify.rs @@ -1,6 +1,7 @@ //! Stored verdicts, re-derived when a title's effective policy changes: -//! an overrides edit (§9.3), a move to a root with a different policy, or a -//! root pointed at a different policy (§5.1). +//! an overrides edit (§9.3), a move to a root with a different policy, a +//! root pointed at a different policy, or an edit to the contents of a +//! policy some root points at (§5.1). //! //! A release's verdict is stamped once, by the search that found it. Both the //! deck and the daemon's manual-grab gate read that stored column, so an @@ -9,7 +10,8 @@ //! reading `rejected` and the grab would be refused. A root move and a //! policy repoint invalidate the column the same way, just wider: nothing //! else re-reads it, so the write that changed the effective policy is the -//! only place the correction can happen. +//! only place the correction can happen. A policy edit invalidates it wider +//! still: every root pointing at that policy, not just one. //! //! So the rules run again here, over the releases already attached to the //! title. This is the same correction the daemon makes when a grab turns out @@ -176,6 +178,30 @@ pub(crate) async fn root(state: &AppState, root_id: i64) -> Result<(), ApiError> Ok(()) } +/// Re-evaluate every release of every title under every root that points at +/// one policy, for a `PUT /api/policies/{id}` that changed its rules (§5.1). +/// +/// Root by root through [`root`], which is title by title through [`movie`] +/// and [`series`]: one walker, one set of skip rules, one place that decides +/// a row does not need rewriting. +/// +/// This is the widest of the four re-derivations — a root repoint moves one +/// library, a policy edit moves every library sharing the policy — but the +/// ceiling is the whole database rather than something that grows with it, +/// since roots partition titles and a title has exactly one root. +pub(crate) async fn policy(state: &AppState, policy_id: i64) -> Result<(), ApiError> { + let root_ids = sqlx::query_scalar!( + r#"SELECT id AS "id!: i64" FROM roots WHERE policy_id = ?"#, + policy_id + ) + .fetch_all(pool(state)?) + .await?; + for root_id in root_ids { + root(state, root_id).await?; + } + Ok(()) +} + #[allow(clippy::too_many_arguments)] async fn apply( state: &AppState,