Fetcher improvements

Cargo.toml

@@ -40,7 +40,7 @@ tempfile = "3.24.0"
 termcolor = "1.4.1"
 thiserror = "2.0.17"
 tiktoken-rs = "0.6.0"
-tokio = { version = "1.48.0", features = ["full"] }
+tokio = { version = "1.48.0", features = ["full", "test-util"] }
 tokio-util = { version = "0.7.17", features = ["codec", "io"] }
 toml = "0.8"
 tower-http = { version = "0.6.8", features = ["cors", "fs", "trace"] }

src/api.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use crate::db::Database;
-use crate::events::Event;
+use crate::events::{Event, MessageSource};
 use crate::fetcher::FetchRequest;
 use crate::settings::ServerSettings;
 use axum::{
@@ -305,7 +305,7 @@ fn generate_synthetic_id(prefix: &str) -> String {
         .expect("Time went backwards");
     // e.g. sashiko-local-1715890000-12345
     format!(
-        "sashiko-{}-{}-{}",
+        "sashiko-{}-{}-{}@sashiko.local",
         prefix,
         since_the_epoch.as_secs(),
         fastrand::u32(..)
@@ -346,6 +346,8 @@ async fn submit_patch(
 
             let event = Event::RawMboxSubmitted {
                 raw,
+                submission_id: id.clone(),
+                source: MessageSource::ApiInject,
                 group: "api-submit".to_string(),
                 baseline: base_commit,
                 skip_subjects,
@@ -381,11 +383,30 @@ async fn submit_patch(
                 sha, repo_display
             );
 
+            // Optimistic check: If we already have this patchset in the DB,
+            // skip creating placeholder and skip fetch queue entirely.
+            match state.db.has_patchset_by_msgid(&id).await {
+                Ok(true) => {
+                    info!(
+                        "Remote fetch request for already ingested SHA {}, skipping placeholder and fetch",
+                        id
+                    );
+                    return Ok(Json(SubmitResponse {
+                        status: "accepted".to_string(),
+                        id,
+                    }));
+                }
+                Err(e) => {
+                    error!("Failed to check if patchset exists: {}", e);
+                }
+                _ => {}
+            }
+
             // Create a placeholder record in the DB so the user can track status
             if let Err(e) = state
                 .db
                 .create_fetching_patchset(
-                    &id,
+                    &format!("{}@sashiko.local", id),
                     &format!("Fetching {} from {}...", &sha, repo_display),
                     skip_subjects.as_ref(),
                     only_subjects.as_ref(),
@@ -444,6 +465,7 @@ async fn submit_patch(
                         .send(Event::IngestionFailed {
                             article_id: msgid_clone.clone(),
                             error: format!("Failed to fetch thread: {}", e),
+                            source: MessageSource::ApiFetchThread,
                         })
                         .await;
                 }
@@ -487,6 +509,8 @@ async fn fetch_and_inject_thread(
 
     let event = Event::RawMboxSubmitted {
         raw,
+        submission_id: msgid.to_string(),
+        source: MessageSource::ApiFetchThread,
         group: "api-submit".to_string(),
         baseline: None,
         skip_subjects: None,
@@ -1008,3 +1032,15 @@ async fn rerun_patch(
 
     Ok(Json(serde_json::json!({ "status": "accepted" })))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_generate_synthetic_id_format() {
+        let id = generate_synthetic_id("test");
+        assert!(id.starts_with("sashiko-test-"));
+        assert!(id.ends_with("@sashiko.local"));
+    }
+}

src/db.rs

@@ -3294,9 +3294,20 @@ impl Database {
         self.rerun_patchset(patchset_id).await
     }
 
+    pub async fn has_patchset_by_msgid(&self, msgid: &str) -> Result<bool> {
+        let mut rows = self
+            .conn
+            .query(
+                "SELECT 1 FROM patchsets WHERE cover_letter_message_id = ? OR cover_letter_message_id = ?",
+                libsql::params![msgid, format!("<{}>", msgid)],
+            )
+            .await?;
+        Ok(rows.next().await.ok().flatten().is_some())
+    }
+
     pub async fn create_fetching_patchset(
         &self,
-        article_id: &str,
+        root_msg_id: &str,
         subject: &str,
         skip_filters: Option<&Vec<String>>,
         only_filters: Option<&Vec<String>>,
@@ -3305,13 +3316,10 @@ impl Database {
             .duration_since(std::time::UNIX_EPOCH)?
             .as_secs() as i64;
 
-        let root_msg_id = if article_id.contains('@') {
-            article_id.to_string()
-        } else {
-            format!("{}@sashiko.local", article_id)
-        };
-
-        let clid_candidates = vec![article_id.to_string(), root_msg_id.clone()];
+        let mut clid_candidates = vec![root_msg_id.to_string()];
+        if let Some(sha) = root_msg_id.strip_suffix("@sashiko.local") {
+            clid_candidates.push(sha.to_string());
+        }
 
         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());
@@ -3332,7 +3340,7 @@ impl Database {
 
                 // Only reset to Fetching if it failed or is currently fetching.
                 // We don't want to reset if it is already Incomplete, Pending, or Reviewed.
-                if status == "Failed" || status == "Fetching" {
+                if status == "Failed" || status == "Fetching" || status == "Cancelled" {
                     self.conn.execute(
                         "UPDATE patchsets SET status = 'Fetching', failed_reason = NULL, skip_filters = ?, only_filters = ? WHERE id = ?",
                         libsql::params![skip_filters_json.clone(), only_filters_json.clone(), id]
@@ -3343,7 +3351,7 @@ impl Database {
         }
 
         // 2. Ensure a placeholder thread and message exist to satisfy Foreign Key constraints
-        let thread_id = self.ensure_thread_for_message(&root_msg_id, now).await?;
+        let thread_id = self.ensure_thread_for_message(root_msg_id, now).await?;
 
         // 3. Create the fetching patchset
         let mut rows = self.conn
@@ -3360,12 +3368,7 @@ impl Database {
             Err(anyhow::anyhow!("Failed to get patchset ID"))
         }
     }
-    pub async fn update_patchset_error(&self, article_id: &str, error: &str) -> Result<()> {
-        let root_msg_id = if article_id.contains('@') {
-            article_id.to_string()
-        } else {
-            format!("{}@sashiko.local", article_id)
-        };
+    pub async fn update_patchset_error(&self, root_msg_id: &str, error: &str) -> Result<()> {
         self.conn
             .execute(
                 "UPDATE patchsets SET status = 'Failed', failed_reason = ? WHERE cover_letter_message_id = ?",
@@ -3438,6 +3441,37 @@ impl Database {
         Ok(count_ps + count_rev)
     }
 
+    pub async fn get_stuck_fetches(&self) -> Result<Vec<(String, Option<String>)>> {
+        let mut rows = self
+            .conn
+            .query(
+                "SELECT p.cover_letter_message_id, b.repo_url
+                 FROM patchsets p 
+                 LEFT JOIN baselines b ON p.baseline_id = b.id 
+                 WHERE p.status = 'Fetching'",
+                (),
+            )
+            .await?;
+
+        let mut stuck = Vec::new();
+        while let Ok(Some(row)) = rows.next().await {
+            let msgid: String = row.get(0)?;
+            let repo: Option<String> = row.get(1).ok().flatten();
+            stuck.push((msgid, repo));
+        }
+        Ok(stuck)
+    }
+
+    pub async fn reset_stuck_fetches(&self) -> Result<()> {
+        self.conn
+            .execute(
+                "UPDATE patchsets SET status = 'Failed', failed_reason = 'Stuck after reboot' WHERE status = 'Fetching'",
+                (),
+            )
+            .await?;
+        Ok(())
+    }
+
     pub async fn get_patchset_counts_by_status(
         &self,
     ) -> Result<std::collections::HashMap<String, usize>> {

src/events.rs

@@ -14,6 +14,16 @@
 
 use crate::patch::{Patch, PatchsetMetadata};
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum MessageSource {
+    Nntp,
+    ApiInject,
+    ApiFetchThread,
+    GitFetch,
+    GitImport,
+    GitArchive,
+}
+
 #[derive(Debug)]
 #[allow(dead_code)]
 pub enum Event {
@@ -39,6 +49,8 @@ pub enum Event {
     },
     RawMboxSubmitted {
         raw: String,
+        submission_id: String,
+        source: MessageSource,
         group: String,
         baseline: Option<String>,
         skip_subjects: Option<Vec<String>>,
@@ -47,13 +59,15 @@ pub enum Event {
     IngestionFailed {
         article_id: String,
         error: String,
+        source: MessageSource,
     },
 }
 
 #[derive(Debug)]
 pub struct ParsedArticle {
     pub group: String,
     pub article_id: String,
+    pub source: MessageSource,
     pub metadata: Option<PatchsetMetadata>,
     pub patch: Option<Patch>,
     pub baseline: Option<String>,