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/api.rs b/src/api.rs index ac1e6c2fd..afa66e967 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 { @@ -467,6 +473,7 @@ async fn submit_patch( None, None, None, + None, ) .await { @@ -527,6 +534,7 @@ async fn submit_patch( None, None, None, + None, ) .await { @@ -614,6 +622,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 +635,7 @@ async fn fetch_and_inject_thread( baseline: None, skip_subjects: None, only_subjects: None, + submitted_at, }; sender.send(event).await?; @@ -1236,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/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/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(); 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/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/main.rs b/src/main.rs index 171fa8c73..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?; @@ -553,6 +565,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 +604,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, @@ -681,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"); @@ -699,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, } @@ -1652,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, @@ -1986,14 +2009,19 @@ 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(), &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, @@ -2004,6 +2032,7 @@ async fn process_parsed_article( strict_author, skip_filters.as_ref(), only_filters.as_ref(), + priority, ) .await { 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/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 ( diff --git a/src/settings.rs b/src/settings.rs index eb21ccc59..b582e1a65 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -403,7 +403,181 @@ 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>( + 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 +/// `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)] @@ -411,12 +585,40 @@ pub struct CustomRemoteSettings { #[allow(unused)] pub struct GitSettings { pub repository_path: String, + #[serde(default, deserialize_with = "deserialize_custom_remotes")] 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 +632,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 +853,159 @@ 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()); + } + + #[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); + } } 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