From 5951281d9c99c41ecf46015bf593b056515e56a6 Mon Sep 17 00:00:00 2001 From: Elkin Cruz Date: Tue, 28 Jul 2026 23:00:40 +0000 Subject: [PATCH 1/7] api: use server timestamp for API-injected mbox submissions Add a submitted_at field to Event::RawMboxSubmitted, stamped with the server's current time when a raw mbox is submitted via POST /api/submit (Inject variant). This timestamp is propagated through metadata.received_date and used as the patchset date instead of the email's Date: header. This prevents stale mbox timestamps (which can be arbitrarily far in the past) from skewing queue ordering. The original email date is preserved in the messages table for display purposes. The Thread and Remote submission paths are unaffected as they already use server-side timestamps via create_fetching_patchset(). Signed-off-by: Elkin Cruz --- src/api.rs | 12 ++++++++++++ src/events.rs | 4 ++++ src/main.rs | 14 ++++++++++++-- 3 files changed, 28 insertions(+), 2 deletions(-) diff --git a/src/api.rs b/src/api.rs index ac1e6c2fd..6176f3b74 100644 --- a/src/api.rs +++ b/src/api.rs @@ -397,6 +397,11 @@ async fn submit_patch( let id = generate_synthetic_id("inject"); info!("Received raw mbox injection: {} (len: {})", id, raw.len()); + let submitted_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .ok(); + let event = Event::RawMboxSubmitted { raw, submission_id: id.clone(), @@ -405,6 +410,7 @@ async fn submit_patch( baseline: base_commit, skip_subjects, only_subjects, + submitted_at, }; if let Err(e) = state.sender.send(event).await { @@ -614,6 +620,11 @@ async fn fetch_and_inject_thread( }) .await??; + let submitted_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .ok(); + let event = Event::RawMboxSubmitted { raw, submission_id: msgid.to_string(), @@ -622,6 +633,7 @@ async fn fetch_and_inject_thread( baseline: None, skip_subjects: None, only_subjects: None, + submitted_at, }; sender.send(event).await?; diff --git a/src/events.rs b/src/events.rs index 82eaec086..e9cc27008 100644 --- a/src/events.rs +++ b/src/events.rs @@ -58,6 +58,10 @@ pub enum Event { baseline: Option, skip_subjects: Option>, only_subjects: Option>, + /// Server-side timestamp for when the submission was received. + /// Used instead of the email's Date: header for patchset ordering + /// to prevent stale mbox timestamps from skewing the queue. + submitted_at: Option, }, IngestionFailed { article_id: String, diff --git a/src/main.rs b/src/main.rs index 171fa8c73..45c4be433 100644 --- a/src/main.rs +++ b/src/main.rs @@ -553,6 +553,7 @@ async fn main() -> Result<(), Box> { baseline, skip_subjects, only_subjects, + submitted_at, } => { let messages = sashiko::ingestor::split_mbox(raw.as_bytes()); let count = messages.len(); @@ -591,7 +592,13 @@ async fn main() -> Result<(), Box> { group: effective_group, article_id: submission_id.clone(), source, - metadata: Some(metadata), + metadata: { + let mut m = metadata; + if submitted_at.is_some() { + m.received_date = submitted_at; + } + Some(m) + }, patch: patch_opt, baseline: baseline_clone, failed_error: None, @@ -1993,7 +2000,10 @@ async fn process_parsed_article( metadata.message_id.as_str(), &subject, &author, - metadata.date, + // Use server-side timestamp (received_date) when available, + // falling back to the email's Date: header. This prevents + // stale mbox timestamps from skewing queue ordering. + metadata.received_date.unwrap_or(metadata.date), total_parts, PARSER_VERSION, &metadata.to, From 78fec789e29acdf3fa630762fb372b72068ff71d Mon Sep 17 00:00:00 2001 From: Elkin Cruz Date: Sat, 8 Aug 2026 00:07:14 +0000 Subject: [PATCH 2/7] settings: add priority rule configuration types Define PriorityRule and CompiledPriorityRule structs for regex-based patchset priority classification. Add a custom serde deserializer (deserialize_indexed_vec) to handle both TOML array and env-var indexed-map representations. Add the priority_rules field to ReviewSettings with serde(default) so existing configs are unaffected. Signed-off-by: Elkin Cruz --- Settings.toml | 10 +++ src/settings.rs | 164 ++++++++++++++++++++++++++++++++++++++++++++++ third_party/linux | 2 +- 3 files changed, 175 insertions(+), 1 deletion(-) diff --git a/Settings.toml b/Settings.toml index 506572309..fbe53a820 100644 --- a/Settings.toml +++ b/Settings.toml @@ -148,3 +148,13 @@ worktree_dir = "review_trees" timeout_seconds = 7200 max_retries = 3 ignore_files = ["MAINTAINERS", ".mailmap", ".gitignore", "LICENSES/"] + +# Priority rules for patchset queue ordering (evaluated in order; last match wins). +# Range is 1 to 999. Default priority is 500. Values > 500 are elevated; values < 500 are deprioritized. +# [[review.priority_rules]] +# regex = "(?i)^PRODKERNEL:" +# priority = 750 +# +# [[review.priority_rules]] +# regex = "(?i)security" +# priority = 999 diff --git a/src/settings.rs b/src/settings.rs index eb21ccc59..b8fb567fd 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -406,6 +406,53 @@ pub struct CustomRemoteSettings { pub only_branches: Option>, } +/// Deserialize a `Vec` that may arrive as a sequence (from TOML `[[...]]`) +/// or as a map of numeric-string indices to values (from env vars via the +/// `config` crate, e.g. `..._RULES__0__REGEX`). Mirrors +/// `deserialize_custom_remotes` for the non-optional list case. +fn deserialize_indexed_vec<'de, D, T>(deserializer: D) -> Result, D::Error> +where + D: serde::Deserializer<'de>, + T: serde::Deserialize<'de>, +{ + struct IndexedVecVisitor(std::marker::PhantomData); + + impl<'de, T: serde::Deserialize<'de>> serde::de::Visitor<'de> for IndexedVecVisitor { + type Value = Vec; + + fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + f.write_str("a sequence or a map of indices to values") + } + + fn visit_seq(self, mut seq: S) -> Result + where + S: serde::de::SeqAccess<'de>, + { + let mut vec = Vec::new(); + while let Some(elem) = seq.next_element()? { + vec.push(elem); + } + Ok(vec) + } + + fn visit_map(self, mut map: M) -> Result + where + M: serde::de::MapAccess<'de>, + { + use std::collections::BTreeMap; + let mut btree = BTreeMap::new(); + while let Some(key) = map.next_key::()? { + let value = map.next_value::()?; + let idx: usize = key.parse().map_err(serde::de::Error::custom)?; + btree.insert(idx, value); + } + Ok(btree.into_values().collect()) + } + } + + deserializer.deserialize_any(IndexedVecVisitor(std::marker::PhantomData)) +} + #[derive(Debug, Deserialize, Clone)] #[serde(deny_unknown_fields)] #[allow(unused)] @@ -414,9 +461,36 @@ pub struct GitSettings { pub custom_remotes: Option>, } +pub const MIN_PRIORITY: i32 = 1; +pub const MAX_PRIORITY: i32 = 999; +pub const DEFAULT_PRIORITY: i32 = 500; + #[derive(Debug, Deserialize, Clone)] #[serde(deny_unknown_fields)] #[allow(unused)] +pub struct PriorityRule { + pub regex: String, + pub priority: i32, +} + +#[derive(Clone, Debug)] +pub struct CompiledPriorityRule { + pub regex: regex::Regex, + pub priority: i32, +} + +impl PriorityRule { + pub fn compile(&self) -> Result { + let re = regex::Regex::new(&self.regex)?; + Ok(CompiledPriorityRule { + regex: re, + priority: self.priority.clamp(MIN_PRIORITY, MAX_PRIORITY), + }) + } +} + +#[derive(Debug, Deserialize, Clone)] +#[allow(unused)] pub struct ReviewSettings { pub concurrency: usize, pub worktree_dir: String, @@ -430,6 +504,8 @@ pub struct ReviewSettings { pub max_files_touched: usize, #[serde(default)] pub ignore_files: Vec, + #[serde(default, deserialize_with = "deserialize_indexed_vec")] + pub priority_rules: Vec, #[serde(default = "default_email_policy_path")] pub email_policy_path: String, /// Maximum cumulative non-cached tokens (uncached input + output) across all turns in a @@ -649,4 +725,92 @@ mod tests { } } } + + #[test] + fn test_priority_rules_deserialize_toml() { + let toml = r#" + concurrency = 4 + worktree_dir = "/tmp/test" + + [[priority_rules]] + regex = "^PRODKERNEL:" + priority = 750 + + [[priority_rules]] + regex = "security" + priority = 999 + "#; + + let settings: ReviewSettings = toml::from_str(toml).expect("parse review settings"); + assert_eq!(settings.priority_rules.len(), 2); + assert_eq!(settings.priority_rules[0].regex, "^PRODKERNEL:"); + assert_eq!(settings.priority_rules[0].priority, 750); + assert_eq!(settings.priority_rules[1].regex, "security"); + assert_eq!(settings.priority_rules[1].priority, 999); + + let compiled = settings.priority_rules[0].compile().expect("compile rule"); + assert!(compiled.regex.is_match("PRODKERNEL: perf fix")); + assert!(!compiled.regex.is_match("staging: driver")); + assert_eq!(compiled.priority, 750); + + // Verify clamping of out-of-range priority values + let clamped_high = PriorityRule { + regex: "test".to_string(), + priority: 5000, + } + .compile() + .expect("compile rule"); + assert_eq!(clamped_high.priority, MAX_PRIORITY); + + let clamped_low = PriorityRule { + regex: "test".to_string(), + priority: -100, + } + .compile() + .expect("compile rule"); + assert_eq!(clamped_low.priority, MIN_PRIORITY); + } + + #[test] + fn test_priority_rules_deserialize_env_map() { + let toml_base = r#" +[review] +concurrency = 2 +worktree_dir = "/tmp/test" +"#; + let cfg = config::Config::builder() + .add_source(config::File::from_str(toml_base, config::FileFormat::Toml)) + .set_override("review.priority_rules.0.regex", "(?i)^PRODKERNEL:") + .expect("set override") + .set_override("review.priority_rules.0.priority", 750) + .expect("set override") + .set_override("review.priority_rules.1.regex", "security") + .expect("set override") + .set_override("review.priority_rules.1.priority", 999) + .expect("set override") + .build() + .expect("build config"); + + #[derive(Debug, Deserialize)] + struct TestWrapper { + review: ReviewSettings, + } + + let wrapper: TestWrapper = cfg.try_deserialize().expect("deserialize config"); + assert_eq!(wrapper.review.priority_rules.len(), 2); + assert_eq!(wrapper.review.priority_rules[0].regex, "(?i)^PRODKERNEL:"); + assert_eq!(wrapper.review.priority_rules[0].priority, 750); + assert_eq!(wrapper.review.priority_rules[1].regex, "security"); + assert_eq!(wrapper.review.priority_rules[1].priority, 999); + } + + #[test] + fn test_priority_rules_default_empty() { + let toml = r#" + concurrency = 2 + worktree_dir = "/tmp/test" + "#; + let settings: ReviewSettings = toml::from_str(toml).expect("parse review settings"); + assert!(settings.priority_rules.is_empty()); + } } diff --git a/third_party/linux b/third_party/linux index 3609fa95f..6bea6b1f8 160000 --- a/third_party/linux +++ b/third_party/linux @@ -1 +1 @@ -Subproject commit 3609fa95fb0f2c1b099e69e56634edb8fc03f87c +Subproject commit 6bea6b1f81acb0ffbd1da38c8c795c59050dc304 From 253a7ed15cebefa8e004a9a7ee8dfcf7c4fa0319 Mon Sep 17 00:00:00 2001 From: Elkin Cruz Date: Sat, 8 Aug 2026 00:06:42 +0000 Subject: [PATCH 3/7] db: add priority column to patchsets Add a priority INTEGER DEFAULT 500 column to the patchsets table and a composite index idx_patchsets_status_priority_date on (status, priority DESC, date ASC) for efficient priority-ordered queries. The column defaults to 500 so existing patchsets are unaffected. Signed-off-by: Elkin Cruz --- src/schema.sql | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/schema.sql b/src/schema.sql index 6e8c66f54..06e0a3aba 100644 --- a/src/schema.sql +++ b/src/schema.sql @@ -86,12 +86,16 @@ CREATE TABLE IF NOT EXISTS patchsets ( embargo_until INTEGER, embargo_release_started_at INTEGER, slug TEXT, -- URL-friendly slug like "reponame-725" (repo-mrnum) + base_priority INTEGER DEFAULT 500, + priority_cap INTEGER, + priority INTEGER DEFAULT 500, FOREIGN KEY(thread_id) REFERENCES threads(id), FOREIGN KEY(cover_letter_message_id) REFERENCES messages(message_id), FOREIGN KEY(baseline_id) REFERENCES baselines(id) ); CREATE INDEX IF NOT EXISTS idx_patchsets_status ON patchsets(status); +CREATE INDEX IF NOT EXISTS idx_patchsets_status_priority_date ON patchsets(status, priority DESC, date ASC); CREATE TABLE IF NOT EXISTS patches ( From 5e6dc6da49d5700a2a5ba77d64a865cdc881022e Mon Sep 17 00:00:00 2001 From: Elkin Cruz Date: Thu, 20 Aug 2026 23:55:27 +0000 Subject: [PATCH 4/7] db: add priority-aware patchset methods and migration Add migration to create the priority column and composite index. Introduce create_patchset_with_priority() which threads an explicit priority through all INSERT/UPDATE paths, with MIN(priority, ?) semantics to preserve manual deprioritization. Refactor create_patchset() to delegate with default None. Change get_pending_patchsets() ordering to priority DESC, date ASC. Add calculate_priority() for evaluating compiled regex rules against subjects (last match wins). Signed-off-by: Elkin Cruz --- src/api.rs | 3 + src/db.rs | 1186 +++++++++++++++++++++++++++++++++++++++++++--------- 2 files changed, 987 insertions(+), 202 deletions(-) diff --git a/src/api.rs b/src/api.rs index 6176f3b74..afa66e967 100644 --- a/src/api.rs +++ b/src/api.rs @@ -473,6 +473,7 @@ async fn submit_patch( None, None, None, + None, ) .await { @@ -533,6 +534,7 @@ async fn submit_patch( None, None, None, + None, ) .await { @@ -1248,6 +1250,7 @@ async fn forge_webhook( Some(subject), Some(metadata.pr_number), slug.as_deref(), + None, ) .await .map_err(|e| { diff --git a/src/db.rs b/src/db.rs index e48a525ce..a5892808e 100644 --- a/src/db.rs +++ b/src/db.rs @@ -531,6 +531,7 @@ impl Database { } self.migrate_patches_unique_constraint_if_needed().await?; + self.migrate_priority_columns_if_needed().await?; Ok(()) } @@ -542,6 +543,22 @@ impl Database { // let _ = self.conn.execute("UPDATE reviews SET status = 'In Review' WHERE status = 'Applying'", ()).await; // Manual migrations for existing tables + let _ = self + .try_add_column("patchsets", "base_priority", "INTEGER DEFAULT 500") + .await; + let _ = self + .try_add_column("patchsets", "priority_cap", "INTEGER") + .await; + let _ = self + .try_add_column("patchsets", "priority", "INTEGER DEFAULT 500") + .await; + let _ = self + .try_create_index( + "idx_patchsets_status_priority_date", + "patchsets", + "status, priority DESC, date ASC", + ) + .await; let _ = self .try_add_column("messages", "to_recipients", "TEXT") .await; @@ -810,6 +827,26 @@ impl Database { Ok(()) } + async fn migrate_priority_columns_if_needed(&self) -> Result<()> { + let _ = self + .try_add_column("patchsets", "base_priority", "INTEGER DEFAULT 500") + .await; + let _ = self + .try_add_column("patchsets", "priority_cap", "INTEGER") + .await; + let _ = self + .try_add_column("patchsets", "priority", "INTEGER DEFAULT 500") + .await; + let _ = self + .try_create_index( + "idx_patchsets_status_priority_date", + "patchsets", + "status, priority DESC, date ASC", + ) + .await; + Ok(()) + } + pub async fn get_mailing_list_id_by_name(&self, name: &str) -> Result> { let mut rows = self .conn @@ -1873,6 +1910,49 @@ impl Database { strict_author: bool, skip_filters: Option<&Vec>, only_filters: Option<&Vec>, + ) -> Result> { + self.create_patchset_with_priority( + thread_id, + cover_letter_message_id, + message_id, + subject, + author, + date, + total_parts, + parser_version, + to, + cc, + version, + part_index, + baseline_id, + strict_author, + skip_filters, + only_filters, + None, + ) + .await + } + + #[allow(clippy::too_many_arguments)] + pub async fn create_patchset_with_priority( + &self, + thread_id: i64, + cover_letter_message_id: Option<&str>, + message_id: &str, + subject: &str, + author: &str, + date: i64, + total_parts: u32, + parser_version: i32, + to: &str, + cc: &str, + version: Option, + part_index: u32, + baseline_id: Option, + strict_author: bool, + skip_filters: Option<&Vec>, + only_filters: Option<&Vec>, + priority: Option, ) -> Result> { let skip_filters_json = skip_filters.map(|f| serde_json::to_string(f).unwrap_or_default()); let only_filters_json = only_filters.map(|f| serde_json::to_string(f).unwrap_or_default()); @@ -1971,12 +2051,49 @@ impl Database { .await?; } - // Update subject if this is a better index (e.g. going from placeholder to real subject) if part_index < subject_index { + if is_placeholder { + // A fetch/API placeholder is created at default priority (500) + // before its real subject is known. Apply the rule-based priority + // authoritatively here if provided, or leave at default. + if let Some(prio) = priority { + self.conn + .execute( + "UPDATE patchsets SET subject = ?, subject_index = ?, base_priority = ?, priority = CASE WHEN priority_cap IS NOT NULL THEN MIN(priority_cap, ?) ELSE ? END WHERE id = ?", + libsql::params![subject, part_index, prio, prio, prio, id], + ) + .await?; + } else { + self.conn + .execute( + "UPDATE patchsets SET subject = ?, subject_index = ? WHERE id = ?", + libsql::params![subject, part_index, id], + ) + .await?; + } + } else if let Some(prio) = priority { + // Back-filling an earlier part of an existing series: + // Update subject/subject_index and elevate base_priority. + self.conn + .execute( + "UPDATE patchsets SET subject = ?, subject_index = ?, base_priority = MAX(base_priority, ?), priority = CASE WHEN priority_cap IS NOT NULL THEN MIN(priority_cap, MAX(base_priority, ?)) ELSE MAX(base_priority, ?) END WHERE id = ?", + libsql::params![subject, part_index, prio, prio, prio, id], + ) + .await?; + } else { + self.conn + .execute( + "UPDATE patchsets SET subject = ?, subject_index = ? WHERE id = ?", + libsql::params![subject, part_index, id], + ) + .await?; + } + } else if let Some(prio) = priority { + // Even if part_index >= subject_index, elevate base_priority if this part has a higher rule priority. self.conn .execute( - "UPDATE patchsets SET subject = ?, subject_index = ? WHERE id = ?", - libsql::params![subject, part_index, id], + "UPDATE patchsets SET base_priority = MAX(base_priority, ?), priority = CASE WHEN priority_cap IS NOT NULL THEN MIN(priority_cap, MAX(base_priority, ?)) ELSE MAX(base_priority, ?) END WHERE id = ?", + libsql::params![prio, prio, prio, id], ) .await?; } @@ -2196,6 +2313,14 @@ impl Database { let merge_from_id = *merge_from_id; info!("Merging patchset {} into {}", merge_from_id, target_id); + // Inherit higher base_priority from merged patchset + self.conn + .execute( + "UPDATE patchsets SET base_priority = MAX(base_priority, (SELECT COALESCE(base_priority, 500) FROM patchsets WHERE id = ?)), priority = CASE WHEN priority_cap IS NOT NULL THEN MIN(priority_cap, MAX(base_priority, (SELECT COALESCE(base_priority, 500) FROM patchsets WHERE id = ?))) ELSE MAX(base_priority, (SELECT COALESCE(base_priority, 500) FROM patchsets WHERE id = ?)) END WHERE id = ?", + libsql::params![merge_from_id, merge_from_id, merge_from_id, target_id], + ) + .await?; + // Reassign patches self.conn .execute( @@ -2267,26 +2392,29 @@ impl Database { .await?; } - // Conditionally update subject - // Note: We check against the best index found among all merged sets OR the new part_index if part_index < current_subject_index { + if let Some(prio) = priority { + self.conn + .execute( + "UPDATE patchsets SET subject = ?, subject_index = ?, base_priority = MAX(base_priority, ?), priority = CASE WHEN priority_cap IS NOT NULL THEN MIN(priority_cap, MAX(base_priority, ?)) ELSE MAX(base_priority, ?) END WHERE id = ?", + libsql::params![subject, part_index, prio, prio, prio, target_id], + ) + .await?; + } else { + self.conn + .execute( + "UPDATE patchsets SET subject = ?, subject_index = ? WHERE id = ?", + libsql::params![subject, part_index, target_id], + ) + .await?; + } + } else if let Some(prio) = priority { self.conn .execute( - "UPDATE patchsets SET subject = ?, subject_index = ? WHERE id = ?", - libsql::params![subject, part_index, target_id], + "UPDATE patchsets SET base_priority = MAX(base_priority, ?), priority = CASE WHEN priority_cap IS NOT NULL THEN MIN(priority_cap, MAX(base_priority, ?)) ELSE MAX(base_priority, ?) END WHERE id = ?", + libsql::params![prio, prio, prio, target_id], ) .await?; - } else if matches.len() > 1 { - // If we merged, we might need to update the subject index of the target to the best one we found. - // But we don't have the subject string from the merged one easily available here. - // However, the existing target subject is likely fine unless part_index is better. - // Update subject_index to be correct if a better one was merged. - // Actually, if matches[i].1 was better, we should have used its subject. - // But that's complicated. Assuming the target (oldest) usually has the cover letter or we eventually find it. - // Simplification: We only update if CURRENT patch is better. - // If we merged a patchset that HAD the cover letter, we ideally want that subject. - // But we lost it. - // TODO: Optimize merge subject selection. For now, this is better than duplicates. } if let Some(clid) = cover_letter_message_id { @@ -2320,11 +2448,12 @@ impl Database { } // No match found, create new patchset + let eff_priority = priority.unwrap_or(crate::settings::DEFAULT_PRIORITY); let mut rows = self.conn .query( - "INSERT INTO patchsets (thread_id, cover_letter_message_id, subject, author, date, total_parts, received_parts, status, parser_version, to_recipients, cc_recipients, subject_index, baseline_id, baseline_part_index, skip_filters, only_filters) - VALUES (?, ?, ?, ?, ?, ?, 0, 'Incomplete', ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id", - libsql::params![thread_id, cover_letter_message_id, subject, author, date, total_parts, parser_version, to, cc, part_index, baseline_id, baseline_id.map(|_| part_index), skip_filters_json.clone(), only_filters_json.clone()], + "INSERT INTO patchsets (thread_id, cover_letter_message_id, subject, author, date, total_parts, received_parts, status, parser_version, to_recipients, cc_recipients, subject_index, baseline_id, baseline_part_index, skip_filters, only_filters, base_priority, priority) + VALUES (?, ?, ?, ?, ?, ?, 0, 'Incomplete', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id", + libsql::params![thread_id, cover_letter_message_id, subject, author, date, total_parts, parser_version, to, cc, part_index, baseline_id, baseline_id.map(|_| part_index), skip_filters_json.clone(), only_filters_json.clone(), eff_priority, eff_priority], ) .await?; @@ -3441,7 +3570,7 @@ impl Database { pub async fn get_pending_patchsets(&self, limit: usize) -> Result> { let mut rows = self.conn.query( "SELECT id, subject, status, thread_id, author, date, cover_letter_message_id, total_parts, received_parts, baseline_id, failed_reason, target_review_count, skip_filters, only_filters, embargo_until, slug - FROM patchsets WHERE status = 'Pending' ORDER BY date ASC LIMIT ?", + FROM patchsets WHERE status = 'Pending' ORDER BY priority DESC, date ASC LIMIT ?", libsql::params![limit as i64], ).await?; @@ -3900,6 +4029,7 @@ impl Database { mr_title: Option<&str>, mr_number: Option, slug: Option<&str>, + priority: Option, ) -> Result { let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH)? @@ -3932,10 +4062,17 @@ impl Database { || status == "Failed To Apply" || status == "FailedToApply" { - self.conn.execute( - "UPDATE patchsets SET status = 'Fetching', failed_reason = NULL, skip_filters = ?, only_filters = ?, mr_url = ?, mr_title = ?, mr_number = ?, slug = ? WHERE id = ?", - libsql::params![skip_filters_json.clone(), only_filters_json.clone(), mr_url, mr_title, mr_number, slug, id] - ).await?; + if let Some(prio) = priority { + self.conn.execute( + "UPDATE patchsets SET status = 'Fetching', failed_reason = NULL, skip_filters = ?, only_filters = ?, mr_url = ?, mr_title = ?, mr_number = ?, slug = ?, base_priority = ?, priority = ? WHERE id = ?", + libsql::params![skip_filters_json.clone(), only_filters_json.clone(), mr_url, mr_title, mr_number, slug, prio, prio, id] + ).await?; + } else { + self.conn.execute( + "UPDATE patchsets SET status = 'Fetching', failed_reason = NULL, skip_filters = ?, only_filters = ?, mr_url = ?, mr_title = ?, mr_number = ?, slug = ? WHERE id = ?", + libsql::params![skip_filters_json.clone(), only_filters_json.clone(), mr_url, mr_title, mr_number, slug, id] + ).await?; + } } return Ok(id); } @@ -3945,11 +4082,12 @@ impl Database { let thread_id = self.ensure_thread_for_message(root_msg_id, now).await?; // 3. Create the fetching patchset + let eff_priority = priority.unwrap_or(crate::settings::DEFAULT_PRIORITY); let mut rows = self.conn .query( - "INSERT INTO patchsets (thread_id, cover_letter_message_id, subject, status, date, skip_filters, only_filters, mr_url, mr_title, mr_number, slug) - VALUES (?, ?, ?, 'Fetching', ?, ?, ?, ?, ?, ?, ?) RETURNING id", - libsql::params![thread_id, root_msg_id, subject, now, skip_filters_json, only_filters_json, mr_url, mr_title, mr_number, slug], + "INSERT INTO patchsets (thread_id, cover_letter_message_id, subject, status, date, skip_filters, only_filters, mr_url, mr_title, mr_number, slug, base_priority, priority) + VALUES (?, ?, ?, 'Fetching', ?, ?, ?, ?, ?, ?, ?, ?, ?) RETURNING id", + libsql::params![thread_id, root_msg_id, subject, now, skip_filters_json, only_filters_json, mr_url, mr_title, mr_number, slug, eff_priority, eff_priority], ) .await?; @@ -3976,6 +4114,22 @@ impl Database { Ok(()) } + pub fn calculate_priority( + subject: &str, + rules: &[crate::settings::CompiledPriorityRule], + ) -> Option { + let mut priority = None; + for rule in rules { + if rule.regex.is_match(subject) { + priority = Some( + rule.priority + .clamp(crate::settings::MIN_PRIORITY, crate::settings::MAX_PRIORITY), + ); + } + } + priority + } + pub async fn update_patchset_baseline_info( &self, id: i64, @@ -4188,198 +4342,812 @@ impl Database { let mut rows = self .conn .query( - "SELECT 1 FROM patchwork_outbox - WHERE patch_msg_id = ? AND api_url = ? AND context = ?", - libsql::params![patch_msg_id, api_url, context], + "SELECT 1 FROM patchwork_outbox + WHERE patch_msg_id = ? AND api_url = ? AND context = ?", + libsql::params![patch_msg_id, api_url, context], + ) + .await?; + if rows.next().await?.is_some() { + return Ok(()); + } + + let created_at = chrono::Utc::now().timestamp(); + self.conn + .execute( + "INSERT INTO patchwork_outbox (patch_msg_id, api_url, check_state, description, target_url, context, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)", + libsql::params![ + patch_msg_id, + api_url, + check_state, + description, + target_url, + context, + created_at, + ], + ) + .await?; + Ok(()) + } + + pub async fn lock_pending_patchwork(&self) -> Result> { + let now = chrono::Utc::now().timestamp(); + let mut rows = self + .conn + .query( + "UPDATE patchwork_outbox + SET status = 'Sending', locked_at = ? + WHERE id = ( + SELECT id FROM patchwork_outbox + WHERE status = 'Pending' + AND (next_retry_at IS NULL OR next_retry_at <= ?) + LIMIT 1 + ) + RETURNING id, patch_msg_id, api_url, check_state, description, target_url, context, status, retry_count, next_retry_at, locked_at, error_log, created_at", + libsql::params![now, now], + ) + .await?; + + if let Ok(Some(row)) = rows.next().await { + let id: i64 = row.get(0)?; + let patch_msg_id: String = row.get(1)?; + let api_url: String = row.get(2)?; + let check_state: String = row.get(3)?; + let description: String = row.get(4)?; + let target_url: String = row.get(5)?; + let context: String = row.get(6)?; + let status: String = row.get(7)?; + let retry_count: i64 = row.get(8)?; + let next_retry_at: Option = row.get::(9).ok(); + let locked_at: Option = row.get::(10).ok(); + let error_log: Option = row.get::(11).ok(); + let created_at: i64 = row.get(12)?; + + Ok(Some(PatchworkOutboxRow { + id, + patch_msg_id, + api_url, + check_state, + description, + target_url, + context, + status, + retry_count, + next_retry_at, + locked_at, + error_log, + created_at, + })) + } else { + Ok(None) + } + } + + pub async fn mark_patchwork_sent(&self, id: i64) -> Result<()> { + self.conn + .execute( + "UPDATE patchwork_outbox SET status = 'Sent', locked_at = NULL WHERE id = ?", + libsql::params![id], + ) + .await?; + Ok(()) + } + + pub async fn mark_patchwork_failed(&self, id: i64, error_log: &str) -> Result<()> { + self.conn + .execute( + "UPDATE patchwork_outbox SET status = 'Failed', error_log = ?, locked_at = NULL WHERE id = ?", + libsql::params![error_log.to_string(), id], + ) + .await?; + Ok(()) + } + + /// Mark a patchwork outbox entry for retry at a future timestamp. + /// Increments retry_count, sets next_retry_at, and returns to + /// Pending status so the worker loop continues without blocking. + pub async fn set_patchwork_retry_at(&self, id: i64, next_retry_at: i64) -> Result<()> { + self.conn + .execute( + "UPDATE patchwork_outbox SET status = 'Pending', retry_count = retry_count + 1, next_retry_at = ?, locked_at = NULL WHERE id = ?", + libsql::params![next_retry_at, id], + ) + .await?; + Ok(()) + } + + pub async fn sweep_ghost_patchwork(&self) -> Result { + let ten_mins_ago = chrono::Utc::now().timestamp() - 600; + let count = self.conn + .execute( + "UPDATE patchwork_outbox SET status = 'Pending', locked_at = NULL WHERE status = 'Sending' AND locked_at < ?", + libsql::params![ten_mins_ago], + ) + .await?; + Ok(count) + } + + /// Insert a patchwork notification email into the email outbox. + /// + /// Uses patch_id = NULL to avoid colliding with the per-patch dedup + /// guard in insert_email_outbox(). The EmailWorker processes these + /// rows normally since it picks up any row with status = 'Pending'. + pub async fn insert_patchwork_notification( + &self, + status: &str, + to_address: &str, + subject: &str, + in_reply_to: &str, + references_hdr: &str, + body: &str, + ) -> Result<()> { + let mut rows = self + .conn + .query( + "SELECT 1 FROM email_outbox + WHERE patch_id IS NULL AND to_addresses = ? AND subject = ? AND in_reply_to = ?", + libsql::params![ + serde_json::to_string(&[to_address]) + .map_err(|e| libsql::Error::Misuse(e.to_string()))?, + subject, + in_reply_to + ], + ) + .await?; + if rows.next().await?.is_some() { + return Ok(()); + } + + let created_at = chrono::Utc::now().timestamp(); + let to_json = serde_json::to_string(&[to_address]) + .map_err(|e| libsql::Error::Misuse(e.to_string()))?; + self.conn + .execute( + "INSERT INTO email_outbox (patch_id, status, to_addresses, cc_addresses, subject, in_reply_to, references_hdr, body, created_at) + VALUES (NULL, ?, ?, '[]', ?, ?, ?, ?, ?)", + libsql::params![ + status, + to_json, + subject, + in_reply_to, + references_hdr, + body, + created_at, + ], + ) + .await?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::settings::DatabaseSettings; + use std::sync::Arc; + + async fn setup_db() -> Arc { + let settings = DatabaseSettings { + url: ":memory:".to_string(), + token: String::new(), + }; + let db = Database::new(&settings).await.unwrap(); + db.migrate().await.unwrap(); + Arc::new(db) + } + + #[tokio::test] + async fn test_patchset_priority_routing() { + let db = setup_db().await; + + let thread_id = db.create_thread("root", "Test Thread", 1000).await.unwrap(); + + let rules_config = [ + crate::settings::PriorityRule { + regex: "(?i)^PRODKERNEL:".to_string(), + priority: 750, + }, + crate::settings::PriorityRule { + regex: "security".to_string(), + priority: 999, + }, + ]; + let compiled_rules: Vec<_> = rules_config.iter().map(|r| r.compile().unwrap()).collect(); + + assert_eq!( + Database::calculate_priority("PRODKERNEL: perf updates", &compiled_rules), + Some(750) + ); + assert_eq!( + Database::calculate_priority("PRODKERNEL: security leak fix", &compiled_rules), + Some(999) + ); + assert_eq!( + Database::calculate_priority("staging: standard driver change", &compiled_rules), + None + ); + + let ps_low = db + .create_patchset_with_priority( + thread_id, + None, + "msg_low", + "staging: driver update", + "Author", + 1000, + 1, + 1, + "", + "", + None, + 1, + None, + true, + None, + None, + None, + ) + .await + .unwrap() + .unwrap(); + + let ps_high = db + .create_patchset_with_priority( + thread_id, + None, + "msg_high", + "PRODKERNEL: fix memory corruption", + "Author", + 2000, + 1, + 1, + "", + "", + None, + 1, + None, + true, + None, + None, + Some(750), + ) + .await + .unwrap() + .unwrap(); + + db.conn + .execute( + "UPDATE patchsets SET status = 'Pending', received_parts = 1 WHERE id IN (?, ?)", + libsql::params![ps_low, ps_high], + ) + .await + .unwrap(); + + let pending = db.get_pending_patchsets(10).await.unwrap(); + assert_eq!(pending.len(), 2); + assert_eq!(pending[0].id, ps_high); + assert_eq!(pending[1].id, ps_low); + } + + #[tokio::test] + async fn test_patchset_priority_two_column_and_series_elevation() { + let db = setup_db().await; + let thread_id = db.create_thread("root", "Test Thread", 1000).await.unwrap(); + + db.create_message( + "msg_part1", + thread_id, + None, + "Author", + "PRODKERNEL: memory fix", + 1000, + "", + "", + "", + None, + None, + ) + .await + .unwrap(); + db.create_message( + "msg_cover", + thread_id, + None, + "Author", + "[PATCH 0/2] series cover", + 1000, + "", + "", + "", + None, + None, + ) + .await + .unwrap(); + db.create_message( + "msg_part2", + thread_id, + None, + "Author", + "[PATCH 2/2] minor tweak", + 1000, + "", + "", + "", + None, + None, + ) + .await + .unwrap(); + + // Scenario 1: Patch 1 arrives first (PRODKERNEL: -> 750), Cover 0 arrives second (None) + let ps1 = db + .create_patchset_with_priority( + thread_id, + None, + "msg_part1", + "PRODKERNEL: memory fix", + "Author", + 1000, + 2, + 1, + "", + "", + None, + 1, + None, + true, + None, + None, + Some(750), + ) + .await + .unwrap() + .unwrap(); + + // Cover letter arrives second (unmatched -> None) + let ps_cover = db + .create_patchset_with_priority( + thread_id, + Some("msg_cover"), + "msg_cover", + "[PATCH 0/2] series cover", + "Author", + 1000, + 2, + 1, + "", + "", + None, + 0, + None, + true, + None, + None, + None, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(ps1, ps_cover); + + let mut rows = db + .conn + .query( + "SELECT base_priority, priority, priority_cap FROM patchsets WHERE id = ?", + libsql::params![ps1], + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + let base_prio: i32 = row.get(0).unwrap(); + let eff_prio: i32 = row.get(1).unwrap(); + let prio_cap: Option = row.get(2).ok(); + assert_eq!(base_prio, 750); + assert_eq!(eff_prio, 750); + assert!(prio_cap.is_none()); + + // Test priority_cap (e.g. from batch deprioritization) + db.conn + .execute( + "UPDATE patchsets SET priority_cap = 200, priority = 200 WHERE id = ?", + libsql::params![ps1], + ) + .await + .unwrap(); + + // Patch 2 arrives (unmatched -> None). base_priority must remain 750, priority remains capped at 200. + db.create_patchset_with_priority( + thread_id, + Some("msg_cover"), + "msg_part2", + "[PATCH 2/2] minor tweak", + "Author", + 1000, + 2, + 1, + "", + "", + None, + 2, + None, + true, + None, + None, + None, + ) + .await + .unwrap() + .unwrap(); + + let mut rows = db + .conn + .query( + "SELECT base_priority, priority, priority_cap FROM patchsets WHERE id = ?", + libsql::params![ps1], + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + let base_prio: i32 = row.get(0).unwrap(); + let eff_prio: i32 = row.get(1).unwrap(); + let prio_cap: Option = row.get(2).ok(); + assert_eq!(base_prio, 750); + assert_eq!(eff_prio, 200); + assert_eq!(prio_cap, Some(200)); + } + + #[tokio::test] + async fn test_patchset_priority_series_deprioritization() { + let db = setup_db().await; + let thread_id = db.create_thread("root", "Test Thread", 1000).await.unwrap(); + + db.create_message( + "msg_doc_cover", + thread_id, + None, + "Author", + "[PATCH 0/2] doc updates", + 1000, + "", + "", + "", + None, + None, + ) + .await + .unwrap(); + db.create_message( + "msg_doc_part1", + thread_id, + None, + "Author", + "[PATCH 1/2] doc: fix spelling", + 1000, + "", + "", + "", + None, + None, + ) + .await + .unwrap(); + db.create_message( + "msg_doc_part2", + thread_id, + None, + "Author", + "[PATCH 2/2] minor grammar", + 1000, + "", + "", + "", + None, + None, + ) + .await + .unwrap(); + + // Cover letter matches low-priority rule (100) + let ps = db + .create_patchset_with_priority( + thread_id, + Some("msg_doc_cover"), + "msg_doc_cover", + "[PATCH 0/2] doc updates", + "Author", + 1000, + 2, + 1, + "", + "", + None, + 0, + None, + true, + None, + None, + Some(100), + ) + .await + .unwrap() + .unwrap(); + + // Patch 1 arrives (also matches 100) + db.create_patchset_with_priority( + thread_id, + Some("msg_doc_cover"), + "msg_doc_part1", + "[PATCH 1/2] doc: fix spelling", + "Author", + 1000, + 2, + 1, + "", + "", + None, + 1, + None, + true, + None, + None, + Some(100), + ) + .await + .unwrap() + .unwrap(); + + // Patch 2 arrives (unmatched -> None). Must NOT elevate series back to 500! + db.create_patchset_with_priority( + thread_id, + Some("msg_doc_cover"), + "msg_doc_part2", + "[PATCH 2/2] minor grammar", + "Author", + 1000, + 2, + 1, + "", + "", + None, + 2, + None, + true, + None, + None, + None, + ) + .await + .unwrap() + .unwrap(); + + let mut rows = db + .conn + .query( + "SELECT base_priority, priority FROM patchsets WHERE id = ?", + libsql::params![ps], + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + let base_prio: i32 = row.get(0).unwrap(); + let eff_prio: i32 = row.get(1).unwrap(); + assert_eq!(base_prio, 100, "Deprioritized series must stay at 100"); + assert_eq!(eff_prio, 100, "Effective priority must stay at 100"); + } + + #[tokio::test] + async fn test_patchset_priority_merge_elevation() { + let db = setup_db().await; + let thread_id = db.create_thread("root", "Test Thread", 1000).await.unwrap(); + + db.create_message( + "msg_part1", + thread_id, + None, + "Author", + "[PATCH 1/2] minor tweak", + 1000, + "", + "", + "", + None, + None, + ) + .await + .unwrap(); + db.create_message( + "msg_part2", + thread_id, + None, + "Author", + "[PATCH 2/2] security fix", + 1000, + "", + "", + "", + None, + None, + ) + .await + .unwrap(); + db.create_message( + "msg_cover", + thread_id, + None, + "Author", + "[PATCH 0/2] series cover", + 1000, + "", + "", + "", + None, + None, + ) + .await + .unwrap(); + + // Patch 1 arrives first (standard priority 500) + let ps1 = db + .create_patchset_with_priority( + thread_id, + None, + "msg_part1", + "[PATCH 1/2] minor tweak", + "Author", + 1000, + 2, + 1, + "", + "", + None, + 1, + None, + true, + None, + None, + None, + ) + .await + .unwrap() + .unwrap(); + + // Patch 2 exists as an independent patchset in the same thread (elevated priority 999) + let mut rows = db + .conn + .query( + "INSERT INTO patchsets (thread_id, cover_letter_message_id, subject, author, date, total_parts, received_parts, status, parser_version, to_recipients, cc_recipients, subject_index, base_priority, priority) + VALUES (?, NULL, '[PATCH 2/2] security fix', 'Author', 1000, 2, 0, 'Incomplete', 1, '', '', 2, 999, 999) RETURNING id", + libsql::params![thread_id], ) - .await?; - if rows.next().await?.is_some() { - return Ok(()); - } + .await + .unwrap(); + let ps2: i64 = rows.next().await.unwrap().unwrap().get(0).unwrap(); + assert_ne!(ps1, ps2); - let created_at = chrono::Utc::now().timestamp(); - self.conn - .execute( - "INSERT INTO patchwork_outbox (patch_msg_id, api_url, check_state, description, target_url, context, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?)", - libsql::params![ - patch_msg_id, - api_url, - check_state, - description, - target_url, - context, - created_at, - ], + // Cover letter arrives (unmatched -> None) in thread_id, merging both patchsets + let ps_merged = db + .create_patchset_with_priority( + thread_id, + Some("msg_cover"), + "msg_cover", + "[PATCH 0/2] series cover", + "Author", + 1000, + 2, + 1, + "", + "", + None, + 0, + None, + true, + None, + None, + None, ) - .await?; - Ok(()) - } + .await + .unwrap() + .unwrap(); - pub async fn lock_pending_patchwork(&self) -> Result> { - let now = chrono::Utc::now().timestamp(); - let mut rows = self + assert_eq!(ps_merged, ps1); + + let mut rows = db .conn .query( - "UPDATE patchwork_outbox - SET status = 'Sending', locked_at = ? - WHERE id = ( - SELECT id FROM patchwork_outbox - WHERE status = 'Pending' - AND (next_retry_at IS NULL OR next_retry_at <= ?) - LIMIT 1 - ) - RETURNING id, patch_msg_id, api_url, check_state, description, target_url, context, status, retry_count, next_retry_at, locked_at, error_log, created_at", - libsql::params![now, now], + "SELECT base_priority, priority FROM patchsets WHERE id = ?", + libsql::params![ps1], ) - .await?; - - if let Ok(Some(row)) = rows.next().await { - let id: i64 = row.get(0)?; - let patch_msg_id: String = row.get(1)?; - let api_url: String = row.get(2)?; - let check_state: String = row.get(3)?; - let description: String = row.get(4)?; - let target_url: String = row.get(5)?; - let context: String = row.get(6)?; - let status: String = row.get(7)?; - let retry_count: i64 = row.get(8)?; - let next_retry_at: Option = row.get::(9).ok(); - let locked_at: Option = row.get::(10).ok(); - let error_log: Option = row.get::(11).ok(); - let created_at: i64 = row.get(12)?; - - Ok(Some(PatchworkOutboxRow { - id, - patch_msg_id, - api_url, - check_state, - description, - target_url, - context, - status, - retry_count, - next_retry_at, - locked_at, - error_log, - created_at, - })) - } else { - Ok(None) - } - } + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + let base_prio: i32 = row.get(0).unwrap(); + let eff_prio: i32 = row.get(1).unwrap(); + assert_eq!( + base_prio, 999, + "Merged patchset must inherit elevated base_priority from merged series" + ); + assert_eq!( + eff_prio, 999, + "Merged patchset must inherit elevated effective priority from merged series" + ); - pub async fn mark_patchwork_sent(&self, id: i64) -> Result<()> { - self.conn - .execute( - "UPDATE patchwork_outbox SET status = 'Sent', locked_at = NULL WHERE id = ?", - libsql::params![id], + // Verify that ps2 was deleted during merge + let mut rows = db + .conn + .query( + "SELECT count(*) FROM patchsets WHERE id = ?", + libsql::params![ps2], ) - .await?; - Ok(()) + .await + .unwrap(); + let count: i64 = rows.next().await.unwrap().unwrap().get(0).unwrap(); + assert_eq!(count, 0, "ps2 must be deleted after merge"); } - pub async fn mark_patchwork_failed(&self, id: i64, error_log: &str) -> Result<()> { - self.conn - .execute( - "UPDATE patchwork_outbox SET status = 'Failed', error_log = ?, locked_at = NULL WHERE id = ?", - libsql::params![error_log.to_string(), id], - ) - .await?; - Ok(()) - } + #[tokio::test] + async fn test_migration_v1_to_priority_columns() { + let db_settings = crate::settings::DatabaseSettings { + url: ":memory:".to_string(), + token: String::new(), + }; + let db = Database::new(&db_settings).await.unwrap(); - /// Mark a patchwork outbox entry for retry at a future timestamp. - /// Increments retry_count, sets next_retry_at, and returns to - /// Pending status so the worker loop continues without blocking. - pub async fn set_patchwork_retry_at(&self, id: i64, next_retry_at: i64) -> Result<()> { - self.conn - .execute( - "UPDATE patchwork_outbox SET status = 'Pending', retry_count = retry_count + 1, next_retry_at = ?, locked_at = NULL WHERE id = ?", - libsql::params![next_retry_at, id], + // Manually set up schema version 1 WITHOUT priority columns + db.conn + .execute_batch( + "CREATE TABLE threads (id INTEGER PRIMARY KEY, root_message_id TEXT, subject TEXT, last_updated INTEGER); + CREATE TABLE messages (id INTEGER PRIMARY KEY, message_id TEXT NOT NULL UNIQUE, thread_id INTEGER, in_reply_to TEXT, author TEXT, subject TEXT, date INTEGER, body TEXT, to_recipients TEXT, cc_recipients TEXT, git_blob_hash TEXT, mailing_list TEXT, references_hdr TEXT); + CREATE TABLE patchsets (id INTEGER PRIMARY KEY, thread_id INTEGER, cover_letter_message_id TEXT, subject TEXT, author TEXT, date INTEGER, total_parts INTEGER, received_parts INTEGER, status TEXT, parser_version INTEGER, to_recipients TEXT, cc_recipients TEXT, subject_index INTEGER, baseline_id INTEGER, failed_reason TEXT, skip_filters TEXT, only_filters TEXT, target_review_count INTEGER DEFAULT 1, model_name TEXT, prompts_git_hash TEXT, baseline_logs TEXT, mr_url TEXT, mr_title TEXT, mr_number TEXT, slug TEXT, provider TEXT, embargo_until INTEGER, embargo_release_started_at INTEGER); + CREATE TABLE patches (id INTEGER PRIMARY KEY, patchset_id INTEGER NOT NULL, message_id TEXT NOT NULL, part_index INTEGER, diff TEXT, status TEXT, apply_error TEXT, FOREIGN KEY(patchset_id) REFERENCES patchsets(id), FOREIGN KEY(message_id) REFERENCES messages(message_id), UNIQUE(patchset_id, message_id)); + PRAGMA user_version = 1;", ) - .await?; - Ok(()) - } + .await + .unwrap(); - pub async fn sweep_ghost_patchwork(&self) -> Result { - let ten_mins_ago = chrono::Utc::now().timestamp() - 600; - let count = self.conn - .execute( - "UPDATE patchwork_outbox SET status = 'Pending', locked_at = NULL WHERE status = 'Sending' AND locked_at < ?", - libsql::params![ten_mins_ago], - ) - .await?; - Ok(count) - } + // Run migrate() on existing version 1 DB + db.migrate().await.unwrap(); - /// Insert a patchwork notification email into the email outbox. - /// - /// Uses patch_id = NULL to avoid colliding with the per-patch dedup - /// guard in insert_email_outbox(). The EmailWorker processes these - /// rows normally since it picks up any row with status = 'Pending'. - pub async fn insert_patchwork_notification( - &self, - status: &str, - to_address: &str, - subject: &str, - in_reply_to: &str, - references_hdr: &str, - body: &str, - ) -> Result<()> { - let mut rows = self - .conn - .query( - "SELECT 1 FROM email_outbox - WHERE patch_id IS NULL AND to_addresses = ? AND subject = ? AND in_reply_to = ?", - libsql::params![ - serde_json::to_string(&[to_address]) - .map_err(|e| libsql::Error::Misuse(e.to_string()))?, - subject, - in_reply_to - ], + // Must now be able to query and insert priority-aware patchsets without column errors + let thread_id = db.create_thread("root", "Test Thread", 1000).await.unwrap(); + let ps_id = db + .create_patchset_with_priority( + thread_id, + None, + "msg_m", + "Migrated Patch", + "Author", + 1000, + 1, + 1, + "", + "", + None, + 1, + None, + true, + None, + None, + Some(750), ) - .await?; - if rows.next().await?.is_some() { - return Ok(()); - } + .await + .unwrap() + .unwrap(); - let created_at = chrono::Utc::now().timestamp(); - let to_json = serde_json::to_string(&[to_address]) - .map_err(|e| libsql::Error::Misuse(e.to_string()))?; - self.conn + let pending = db.get_pending_patchsets(10).await.unwrap(); + assert_eq!(pending.len(), 0); + + db.conn .execute( - "INSERT INTO email_outbox (patch_id, status, to_addresses, cc_addresses, subject, in_reply_to, references_hdr, body, created_at) - VALUES (NULL, ?, ?, '[]', ?, ?, ?, ?, ?)", - libsql::params![ - status, - to_json, - subject, - in_reply_to, - references_hdr, - body, - created_at, - ], + "UPDATE patchsets SET status = 'Pending', received_parts = 1 WHERE id = ?", + libsql::params![ps_id], ) - .await?; - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::settings::DatabaseSettings; - use std::sync::Arc; + .await + .unwrap(); - async fn setup_db() -> Arc { - let settings = DatabaseSettings { - url: ":memory:".to_string(), - token: String::new(), - }; - let db = Database::new(&settings).await.unwrap(); - db.migrate().await.unwrap(); - Arc::new(db) + let pending = db.get_pending_patchsets(10).await.unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].id, ps_id); } #[tokio::test] @@ -8668,6 +9436,7 @@ mod tests { None, None, None, + None, ) .await .unwrap(); @@ -8739,6 +9508,7 @@ mod tests { None, None, None, + None, ) .await .unwrap(); @@ -8816,12 +9586,20 @@ mod tests { db.has_patchset_by_msgid(sha_patch).await.unwrap(), "has_patchset_by_msgid should return true for SHA existing in patches table" ); + + // 3. Non-existent SHA must return false + assert!( + !db.has_patchset_by_msgid("0000000000000000000000000000000000000000") + .await + .unwrap(), + "has_patchset_by_msgid should return false for unknown SHA" + ); } /// Verify that has_patchset_by_msgid returns false for Failed and Cancelled /// patchsets so that retry submissions can be re-fetched. #[tokio::test] - async fn test_has_patchset_by_msgid_excludes_failed_and_cancelled() { + async fn test_has_patchset_by_msgid_ignores_failed_and_cancelled() { let db = setup_db().await; let sha_failed = "f00f00f001234567890abcdef1234567890abcdef"; let synthetic_failed = format!("{}@sashiko.local", sha_failed); @@ -8837,6 +9615,7 @@ mod tests { None, None, None, + None, ) .await .unwrap(); @@ -8863,6 +9642,7 @@ mod tests { None, None, None, + None, ) .await .unwrap(); @@ -9111,6 +9891,7 @@ mod tests { None, None, None, + None, ) .await .unwrap(); @@ -9184,6 +9965,7 @@ mod tests { None, None, None, + None, ) .await .unwrap(); From 690cf049396212f923a24988d208725f9a2d8977 Mon Sep 17 00:00:00 2001 From: Elkin Cruz Date: Thu, 20 Aug 2026 23:55:36 +0000 Subject: [PATCH 5/7] review: wire priority routing into patchset processing Compile priority_rules from settings at startup and thread them through the DB worker into process_parsed_article(). Use calculate_priority() to compute priority from the patchset subject before calling create_patchset_with_priority(). API callers pass None for priority to create_fetching_patchset(). Signed-off-by: Elkin Cruz --- src/main.rs | 25 ++++++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/src/main.rs b/src/main.rs index 45c4be433..a59d76c22 100644 --- a/src/main.rs +++ b/src/main.rs @@ -369,7 +369,19 @@ async fn main() -> Result<(), Box> { settings.review.stages = Some(stages.clone()); info!("Selected stages via --stages flag: {:?}", stages); } - + let mut compiled_rules = Vec::new(); + for rule in &settings.review.priority_rules { + match rule.compile() { + Ok(r) => compiled_rules.push(r), + Err(e) => { + error!( + "Invalid priority rule regex '{}': {}. Skipping rule.", + rule.regex, e + ); + } + } + } + let compiled_rules = Arc::new(compiled_rules); // Initialize Database let db = Arc::new(Database::new(&settings.database).await?); db.migrate().await?; @@ -688,6 +700,7 @@ async fn main() -> Result<(), Box> { // DB Worker (Transactional Batching) let worker_db = db.clone(); let mapping = settings.subsystems.mapping.clone(); + let db_rules = compiled_rules.clone(); let _db_worker_handle = tokio::spawn(async move { info!("DB Worker started"); @@ -706,7 +719,9 @@ async fn main() -> Result<(), Box> { } for article in buffer.drain(..) { - match process_parsed_article(&worker_db, article, &policy, &mapping).await { + match process_parsed_article(&worker_db, article, &policy, &mapping, &db_rules) + .await + { ProcessStatus::Ingested => total_ingested += 1, ProcessStatus::Error => total_errors += 1, } @@ -1659,6 +1674,7 @@ async fn process_parsed_article( article: ParsedArticle, policy: &sashiko::email_policy::EmailPolicyConfig, subsystem_mapping: &[sashiko::settings::SubsystemMapping], + priority_rules: &[sashiko::settings::CompiledPriorityRule], ) -> ProcessStatus { let ParsedArticle { group, @@ -1993,8 +2009,10 @@ async fn process_parsed_article( None }; + let priority = sashiko::db::Database::calculate_priority(&subject, priority_rules); + match worker_db - .create_patchset( + .create_patchset_with_priority( thread_id, cover_letter_id, metadata.message_id.as_str(), @@ -2014,6 +2032,7 @@ async fn process_parsed_article( strict_author, skip_filters.as_ref(), only_filters.as_ref(), + priority, ) .await { From f6ba7bc370abbf2fdbc4ed71cfbf9886e016ab9b Mon Sep 17 00:00:00 2001 From: Elkin Cruz Date: Tue, 30 Jun 2026 18:30:05 +0000 Subject: [PATCH 6/7] remote: support deserializing custom_remotes list from map format The config crate deserializes custom_remotes environment variables SASHIKO__GIT__CUSTOM_REMOTES__* as a map with numeric string keys instead of a sequence. This caused a deserialization failure when attempting to parse into a Vec structure. Replace the untagged enum deserialization strategy with a custom Visitor pattern in settings.rs. This Visitor successfully handles both sequential TOML/JSON arrays and indexed map formats, preserving type coercion for nested fields. Signed-off-by: Elkin Cruz --- src/settings.rs | 134 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 134 insertions(+) diff --git a/src/settings.rs b/src/settings.rs index b8fb567fd..cb095d5c1 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -406,6 +406,72 @@ pub struct CustomRemoteSettings { pub only_branches: Option>, } +fn deserialize_custom_remotes<'de, D>( + deserializer: D, +) -> Result>, D::Error> +where + D: serde::Deserializer<'de>, +{ + struct CustomRemotesVisitor; + + impl<'de> serde::de::Visitor<'de> for CustomRemotesVisitor { + type Value = Option>; + + fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result { + formatter.write_str( + "a sequence of CustomRemoteSettings or a map of indices to CustomRemoteSettings", + ) + } + + fn visit_none(self) -> Result + where + E: serde::de::Error, + { + Ok(None) + } + + fn visit_some(self, deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + deserializer.deserialize_any(self) + } + + fn visit_seq(self, mut seq: S) -> Result + where + S: serde::de::SeqAccess<'de>, + { + let mut vec = Vec::new(); + while let Some(elem) = seq.next_element()? { + vec.push(elem); + } + Ok(Some(vec)) + } + + fn visit_map(self, mut map: M) -> Result + where + M: serde::de::MapAccess<'de>, + { + use std::collections::BTreeMap; + let mut btree = BTreeMap::new(); + while let Some(key) = map.next_key::()? { + let value = map.next_value::()?; + if let Ok(idx) = key.parse::() { + btree.insert(idx, value); + } else { + return Err(serde::de::Error::custom(format!( + "invalid index in custom_remotes map: {}", + key + ))); + } + } + Ok(Some(btree.into_values().collect())) + } + } + + deserializer.deserialize_option(CustomRemotesVisitor) +} + /// Deserialize a `Vec` that may arrive as a sequence (from TOML `[[...]]`) /// or as a map of numeric-string indices to values (from env vars via the /// `config` crate, e.g. `..._RULES__0__REGEX`). Mirrors @@ -458,6 +524,7 @@ where #[allow(unused)] pub struct GitSettings { pub repository_path: String, + #[serde(default, deserialize_with = "deserialize_custom_remotes")] pub custom_remotes: Option>, } @@ -813,4 +880,71 @@ worktree_dir = "/tmp/test" let settings: ReviewSettings = toml::from_str(toml).expect("parse review settings"); assert!(settings.priority_rules.is_empty()); } + + #[test] + fn test_deserialize_custom_remotes_seq() { + let toml_str = r#" + repository_path = "path/to/repo" + [[custom_remotes]] + name = "remote1" + url = "url1" + check_all_branches = true + "#; + let settings: GitSettings = toml::from_str(toml_str).unwrap(); + let remotes = settings.custom_remotes.unwrap(); + assert_eq!(remotes.len(), 1); + assert_eq!(remotes[0].name, "remote1"); + assert_eq!(remotes[0].url, "url1"); + assert!(remotes[0].check_all_branches); + } + + #[test] + fn test_deserialize_custom_remotes_map() { + let json_str = r#"{ + "repository_path": "path/to/repo", + "custom_remotes": { + "1": { + "name": "remote1", + "url": "url1", + "check_all_branches": false + }, + "0": { + "name": "remote0", + "url": "url0", + "check_all_branches": true + } + } + }"#; + let settings: GitSettings = serde_json::from_str(json_str).unwrap(); + let remotes = settings.custom_remotes.unwrap(); + assert_eq!(remotes.len(), 2); + assert_eq!(remotes[0].name, "remote0"); + assert_eq!(remotes[0].url, "url0"); + assert!(remotes[0].check_all_branches); + assert_eq!(remotes[1].name, "remote1"); + assert_eq!(remotes[1].url, "url1"); + assert!(!remotes[1].check_all_branches); + } + + #[test] + fn test_deserialize_custom_remotes_config_rs() { + let s = Config::builder() + .set_default("repository_path", "path/to/repo") + .unwrap() + .set_default("custom_remotes.0.name", "prodkernel") + .unwrap() + .set_default("custom_remotes.0.url", "sso://prodkernel/kernel/icebreaker") + .unwrap() + .set_default("custom_remotes.0.check_all_branches", "true") + .unwrap() + .build() + .unwrap(); + + let settings: GitSettings = s.try_deserialize().unwrap(); + let remotes = settings.custom_remotes.unwrap(); + assert_eq!(remotes.len(), 1); + assert_eq!(remotes[0].name, "prodkernel"); + assert_eq!(remotes[0].url, "sso://prodkernel/kernel/icebreaker"); + assert!(remotes[0].check_all_branches); + } } From 0906d97dbc8fe0364f8bfcfdcd38b7064bcf7688 Mon Sep 17 00:00:00 2001 From: Elkin Cruz Date: Wed, 29 Jul 2026 08:55:27 +0000 Subject: [PATCH 7/7] remote: add branch_patterns for selective remote fetching When a custom remote has thousands of branches (e.g. icebreaker with 4,467 branches), fetching all refs downloads millions of objects and saturates the pod for hours. Add a branch_patterns field to CustomRemoteSettings that accepts glob patterns (e.g. "*/6.18", "14*"). When configured: 1. ensure_remote builds refspecs from patterns so git fetch only downloads matching refs instead of all branches 2. check_all_branches filters the branch list through the same patterns before adding baseline candidates Implement a simple glob matcher supporting "*" (any substring) and "?" (any single character) in git_ops, avoiding an external dependency for this small use case. Signed-off-by: Elkin Cruz --- src/baseline.rs | 39 ++++++++++++++--- src/git_ops.rs | 109 +++++++++++++++++++++++++++++++++++++++++++++++- src/reviewer.rs | 43 ++++++++++++------- src/settings.rs | 61 +++++++++++++++++++++++++++ 4 files changed, 232 insertions(+), 20 deletions(-) diff --git a/src/baseline.rs b/src/baseline.rs index cf6b16234..02bdd4820 100644 --- a/src/baseline.rs +++ b/src/baseline.rs @@ -291,10 +291,28 @@ impl BaselineRegistry { // with the topic branch once the maintainer rebases. if let Some(custom_remotes) = &self.custom_remotes { for remote in custom_remotes { - // Fetch to ensure we have the latest branches (Issue 1) - if let Err(e) = - crate::git_ops::ensure_remote(&self.repo_path, &remote.name, &remote.url, false) - .await + // Build refspecs from branch_patterns to limit what we fetch. + // Each pattern like "*/6.18" becomes a refspec: + // +refs/heads/*/6.18:refs/remotes//*/6.18 + let refspecs: Option> = + remote.branch_patterns.as_ref().map(|patterns| { + patterns + .iter() + .map(|p| { + format!("+refs/heads/{}:refs/remotes/{}/{}", p, remote.name, p) + }) + .collect() + }); + + // Fetch with refspecs to limit downloaded objects + if let Err(e) = crate::git_ops::ensure_remote_with_refspecs( + &self.repo_path, + &remote.name, + &remote.url, + false, + refspecs.as_deref(), + ) + .await { warn!( "Failed to ensure custom remote {}: {}. Using local branches.", @@ -305,7 +323,16 @@ impl BaselineRegistry { if remote.check_all_branches { match crate::git_ops::get_remote_branches(&self.repo_path, &remote.name).await { Ok(branches) => { - for branch in branches { + // Filter by branch_patterns if configured + let filtered = if let Some(ref patterns) = remote.branch_patterns { + branches + .into_iter() + .filter(|b| crate::git_ops::matches_branch_pattern(b, patterns)) + .collect() + } else { + branches + }; + for branch in filtered { candidates.push(BaselineResolution::RemoteTarget { url: remote.url.clone(), name: remote.name.clone(), @@ -915,6 +942,7 @@ F: patterns/ url: dummy_url, check_all_branches: false, only_branches: Some(vec!["master".to_string()]), + branch_patterns: None, }]), repo_path: repo_path.to_path_buf(), mainline_remote: None, @@ -954,6 +982,7 @@ F: patterns/ url: dummy_url, check_all_branches: false, only_branches: Some(vec!["topic-next".to_string()]), + branch_patterns: None, }]), repo_path: repo_path.to_path_buf(), mainline_remote: None, diff --git a/src/git_ops.rs b/src/git_ops.rs index 115318836..05347e7b2 100644 --- a/src/git_ops.rs +++ b/src/git_ops.rs @@ -510,11 +510,70 @@ fn get_worktree_lock() -> Arc> { .clone() } +/// Checks whether a branch name matches any of the given glob patterns. +/// Supports simple globs: "*" matches any sequence of characters, +/// "?" matches any single character. Matching is case-sensitive. +pub fn matches_branch_pattern(branch: &str, patterns: &[String]) -> bool { + for pattern in patterns { + if glob_match(pattern, branch) { + return true; + } + } + false +} + +/// Simple glob matching: "*" matches any substring (including path +/// separators), "?" matches exactly one character. +fn glob_match(pattern: &str, text: &str) -> bool { + let pat: Vec = pattern.chars().collect(); + let txt: Vec = text.chars().collect(); + let (plen, tlen) = (pat.len(), txt.len()); + let (mut pi, mut ti) = (0usize, 0usize); + let (mut star_pi, mut star_ti) = (usize::MAX, 0usize); + + while ti < tlen { + if pi < plen && pat[pi] == '?' { + pi += 1; + ti += 1; + } else if pi < plen && pat[pi] == '*' { + star_pi = pi; + star_ti = ti; + pi += 1; + } else if pi < plen && pat[pi] == txt[ti] { + pi += 1; + ti += 1; + } else if star_pi != usize::MAX { + pi = star_pi + 1; + star_ti += 1; + ti = star_ti; + } else { + return false; + } + } + while pi < plen && pat[pi] == '*' { + pi += 1; + } + pi == plen +} + pub async fn ensure_remote( repo_path: &Path, name: &str, url: &str, force_fetch: bool, +) -> Result<()> { + ensure_remote_with_refspecs(repo_path, name, url, force_fetch, None).await +} + +/// Like ensure_remote, but optionally limits the fetch to specific +/// refspecs. When refspecs is Some, only the listed refs are fetched +/// instead of all remote branches. +pub async fn ensure_remote_with_refspecs( + repo_path: &Path, + name: &str, + url: &str, + force_fetch: bool, + refspecs: Option<&[String]>, ) -> Result<()> { // 1. Validate repo_path to prevent git from traversing up to parent repos if !repo_path.join(".git").exists() && !repo_path.join("HEAD").exists() { @@ -658,6 +717,11 @@ pub async fn ensure_remote( fetch_args.push("--no-tags"); } fetch_args.push(name); + if let Some(specs) = refspecs { + for spec in specs { + fetch_args.push(spec); + } + } let fetch_future = Command::new("git") .current_dir(repo_path) @@ -690,7 +754,7 @@ pub async fn ensure_remote( } Err(_) => { error_msg = format!( - "Git fetch for {} timed out after {} seconds", + "Timed out fetching remote {} after {}s", name, timeout_duration.as_secs() ); @@ -1480,4 +1544,47 @@ mod tests { Ok(()) } + + #[test] + fn test_glob_match() { + assert!(super::glob_match("*/6.18", "core/6.18")); + assert!(super::glob_match("*/6.18", "net/tcp/6.18")); + assert!(super::glob_match("*/6.18", "ovss/platforms/arm64/6.18")); + assert!(!super::glob_match("*/6.18", "core/6.12")); + assert!(!super::glob_match("*/6.18", "core/6.189")); + + assert!(super::glob_match("14*", "1400")); + assert!(super::glob_match("14*", "1400_136")); + assert!(super::glob_match("14*", "1404_881")); + assert!(!super::glob_match("14*", "1500")); + + assert!(super::glob_match("15*", "1500")); + assert!(super::glob_match("15*", "1501_159")); + assert!(!super::glob_match("15*", "1400")); + + assert!(super::glob_match("linux/master", "linux/master")); + assert!(!super::glob_match("linux/master", "linux/stable")); + + assert!(super::glob_match("linux/stable/*", "linux/stable/6.12")); + assert!(super::glob_match("linux/stable/*", "linux/stable/6.18")); + } + + #[test] + fn test_matches_branch_pattern() { + let patterns = vec![ + "*/6.18".to_string(), + "*/6.12".to_string(), + "14*".to_string(), + "15*".to_string(), + "linux/master".to_string(), + ]; + assert!(super::matches_branch_pattern("core/6.18", &patterns)); + assert!(super::matches_branch_pattern("kvm/mmu/6.12", &patterns)); + assert!(super::matches_branch_pattern("1400_136", &patterns)); + assert!(super::matches_branch_pattern("1500", &patterns)); + assert!(super::matches_branch_pattern("linux/master", &patterns)); + assert!(!super::matches_branch_pattern("core/6.6", &patterns)); + assert!(!super::matches_branch_pattern("1300_448", &patterns)); + assert!(!super::matches_branch_pattern("sched/ghost/6.1", &patterns)); + } } diff --git a/src/reviewer.rs b/src/reviewer.rs index a9ca19b3a..ed7fd02e1 100644 --- a/src/reviewer.rs +++ b/src/reviewer.rs @@ -24,7 +24,6 @@ use crate::email_policy::EmailPolicyConfig; use crate::email_router::{Action as EmailAction, EmailRouter}; use crate::git_ops::{GitWorktree, ensure_remote, get_commit_hash}; use crate::settings::Settings; -use crate::utils::redact_secret; use crate::worker::prompts::ReviewError; use anyhow::Result; use serde::Serialize; @@ -822,19 +821,35 @@ impl Reviewer { let mut current_log = format!("Trying baseline: {}\n", baseline_ref); let mut current_status = "Failed".to_string(); - // Check remote - if let BaselineResolution::RemoteTarget { url, name, .. } = candidate - && let Err(e) = ensure_remote(&repo_path, name, url, false).await - { - let msg = format!("Failed to fetch remote {}: {}\n", redact_secret(url), e); - current_log.push_str(&msg); - error!("{}", msg.trim()); - attempts.push(BaselineAttempt { - baseline: baseline_ref.clone(), - status: current_status, - log: current_log, - }); - continue; + // Only block on ensure_remote when the remote does not exist + // yet (first boot). Once configured, the GitSyncWorker keeps + // it fresh hourly so we never stall reviews on a fetch. + if let BaselineResolution::RemoteTarget { url, name, .. } = candidate { + let exists = tokio::process::Command::new("git") + .current_dir(&repo_path) + .args(["remote", "get-url", name]) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status() + .await + .map(|s| s.success()) + .unwrap_or(false); + if !exists { + // First time: add the remote and fetch (blocking is OK + // because we literally have zero branches otherwise). + if let Err(e) = ensure_remote(&repo_path, name, url, false).await { + let msg = format!("Failed to set up remote {}: {}\n", name, e); + current_log.push_str(&msg); + error!("{}", msg.trim()); + attempts.push(BaselineAttempt { + baseline: baseline_ref.clone(), + status: current_status, + log: current_log, + }); + continue; + } + } + // Remote exists: proceed with whatever branches are available. } // Resolve SHA diff --git a/src/settings.rs b/src/settings.rs index cb095d5c1..b582e1a65 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -403,7 +403,68 @@ pub struct CustomRemoteSettings { pub name: String, pub url: String, pub check_all_branches: bool, + #[serde(default, deserialize_with = "deserialize_optional_string_vec")] pub only_branches: Option>, + /// Glob patterns for branch filtering (e.g., "*/6.18", "14*"). + /// When set, ensure_remote fetches only matching refs and + /// check_all_branches filters to matching branches. + #[serde(default, deserialize_with = "deserialize_optional_string_vec")] + pub branch_patterns: Option>, +} + +/// Deserializes an `Option>` that may arrive as a JSON array +/// (from TOML) or as a map with numeric string keys (from env vars via +/// the `config` crate, e.g. `BRANCH_PATTERNS__0=...`). +fn deserialize_optional_string_vec<'de, D>(deserializer: D) -> Result>, D::Error> +where + D: serde::Deserializer<'de>, +{ + struct OptVecVisitor; + + impl<'de> serde::de::Visitor<'de> for OptVecVisitor { + type Value = Option>; + + fn expecting(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result { + f.write_str("a sequence of strings or a map of indices to strings") + } + + fn visit_none(self) -> Result { + Ok(None) + } + + fn visit_some>( + self, + deserializer: D, + ) -> Result { + deserializer.deserialize_any(self) + } + + fn visit_seq>( + self, + mut seq: S, + ) -> Result { + let mut vec = Vec::new(); + while let Some(elem) = seq.next_element()? { + vec.push(elem); + } + Ok(Some(vec)) + } + + fn visit_map>( + self, + mut map: M, + ) -> Result { + let mut entries = Vec::new(); + while let Some((key, value)) = map.next_entry::()? { + let idx: usize = key.parse().map_err(serde::de::Error::custom)?; + entries.push((idx, value)); + } + entries.sort_by_key(|(idx, _)| *idx); + Ok(Some(entries.into_iter().map(|(_, v)| v).collect())) + } + } + + deserializer.deserialize_option(OptVecVisitor) } fn deserialize_custom_remotes<'de, D>(