feat(api): re-derive verdicts on a policy edit

PUT /api/policies/{id} changed the rule every title under every root
pointing at the policy is judged by, and re-derived nothing, so §9.3's
deck and the daemon's grab gate kept reading verdicts computed under
rules that no longer existed.

Drives #241's walker from a policy id: root by root through
reclassify::root, so the skip rules and the leave-unchanged-rows-alone
rule stay in one place. A rename touches no rule and walks nothing.

Inline still holds at this width. Measured on a release build over 2000
titles and 10 000 stored releases across two roots sharing one policy:
0.36 s when no verdict moves, 2.7 s when all 10 000 do. DESIGN.md §5.1
now names four actions and carries those numbers.

just ci passed through the gate.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

.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"
+}

.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"
+}

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 —
+root with a different policy, pointing a root at a different policy, and
-re-derive the stored verdicts of everything they touch, inline in the same
+editing the contents of a policy some root points at — re-derive the stored
-request. The operator is never left reading a verdict computed under a
+verdicts of everything they touch, inline in the same request. The operator
-policy that no longer applies. Inline is affordable even root-wide:
+is never left reading a verdict computed under a policy that no longer
-re-evaluation is pure and in-memory, and a row whose verdict does not move
+applies.
-is not rewritten, so a library-sized repoint costs reads, not writes.
+
+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
 

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<T: serde::de::DeserializeOwned>(
     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<Option<PolicyRules>, 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<serde_json::Value> = 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<serde_json::Value> = 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<serde_json::Value> = (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<serde_json::Value> = 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"
+        );
+    }
 }

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
+//! an overrides edit (§9.3), a move to a root with a different policy, a
-//! root pointed at a different policy (§5.1).
+//! 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,