Fence concurrent import observations

Author: luca-ctx

crates/ctx-cli/src/commands/import.rs

@@ -316,30 +316,67 @@ pub(crate) struct InventoryTotals {
 pub(crate) enum SourcePreinventory {
     #[default]
     None,
-    CodexSessionCatalog(CatalogSummary),
-    SourceImportFiles(Vec<SourceImportFile>),
-    SourceRoot(SourceImportFile),
+    CodexSessionCatalog {
+        summary: CatalogSummary,
+        inventory_generation: u64,
+    },
+    SourceImportFiles {
+        files: Vec<SourceImportFile>,
+        inventory_generation: u64,
+    },
+    SourceRoot {
+        file: SourceImportFile,
+        inventory_generation: u64,
+    },
 }
 
 impl SourcePreinventory {
     pub(crate) fn codex_session_catalog(&self) -> Option<&CatalogSummary> {
         match self {
-            Self::CodexSessionCatalog(summary) => Some(summary),
-            Self::None | Self::SourceImportFiles(_) | Self::SourceRoot(_) => None,
+            Self::CodexSessionCatalog { summary, .. } => Some(summary),
+            Self::None | Self::SourceImportFiles { .. } | Self::SourceRoot { .. } => None,
         }
     }
 
     pub(crate) fn source_import_files(&self) -> Option<&[SourceImportFile]> {
         match self {
-            Self::SourceImportFiles(files) => Some(files),
-            Self::None | Self::CodexSessionCatalog(_) | Self::SourceRoot(_) => None,
+            Self::SourceImportFiles { files, .. } => Some(files),
+            Self::None | Self::CodexSessionCatalog { .. } | Self::SourceRoot { .. } => None,
         }
     }
 
     pub(crate) fn source_root_file(&self) -> Option<&SourceImportFile> {
         match self {
-            Self::SourceRoot(file) => Some(file),
-            Self::None | Self::CodexSessionCatalog(_) | Self::SourceImportFiles(_) => None,
+            Self::SourceRoot { file, .. } => Some(file),
+            Self::None | Self::CodexSessionCatalog { .. } | Self::SourceImportFiles { .. } => None,
+        }
+    }
+
+    pub(crate) fn source_root_observation(&self) -> Option<(&SourceImportFile, u64)> {
+        match self {
+            Self::SourceRoot {
+                file,
+                inventory_generation,
+            } => Some((file, *inventory_generation)),
+            Self::None | Self::CodexSessionCatalog { .. } | Self::SourceImportFiles { .. } => None,
+        }
+    }
+
+    pub(crate) fn inventory_generation(&self) -> Option<u64> {
+        match self {
+            Self::None => None,
+            Self::CodexSessionCatalog {
+                inventory_generation,
+                ..
+            }
+            | Self::SourceImportFiles {
+                inventory_generation,
+                ..
+            }
+            | Self::SourceRoot {
+                inventory_generation,
+                ..
+            } => Some(*inventory_generation),
         }
     }
 }

crates/ctx-cli/src/commands/import/catalog.rs

@@ -14,9 +14,16 @@ pub(crate) fn import_incremental_codex_session_tree(
     record_id: Uuid,
     progress: Option<CodexSessionImportProgressCallback>,
     preinventory_catalog: Option<&CatalogSummary>,
+    preinventory_generation: Option<u64>,
     force_selection: bool,
 ) -> Result<ProviderImportSummary> {
     let source_root = source.path.display().to_string();
+    let inventory_generation = match preinventory_generation {
+        Some(generation) => generation,
+        None => {
+            store.allocate_catalog_inventory_generation(CaptureProvider::Codex, &source_root)?
+        }
+    };
     let mut summary = ProviderImportSummary::default();
     if let Some(catalog) = preinventory_catalog {
         summary.failed += catalog.failed_sessions;
@@ -27,6 +34,7 @@ pub(crate) fn import_incremental_codex_session_tree(
             store,
             CodexSessionCatalogOptions {
                 source_root: Some(source.path.clone()),
+                observation_generation: Some(inventory_generation),
                 ..CodexSessionCatalogOptions::default()
             },
         )
@@ -76,6 +84,7 @@ pub(crate) fn import_incremental_codex_session_tree(
                         std::slice::from_ref(session),
                         &error,
                         catalog_import_error_status(&err),
+                        inventory_generation,
                     )?;
                     return Err(err);
                 }
@@ -104,6 +113,7 @@ pub(crate) fn import_incremental_codex_session_tree(
                         std::slice::from_ref(session),
                         &err.to_string(),
                         catalog_import_error_status(&err),
+                        inventory_generation,
                     )?;
                     return Err(err);
                 }
@@ -129,6 +139,7 @@ pub(crate) fn import_incremental_codex_session_tree(
                 utc_now().timestamp_millis(),
                 status,
                 error.as_deref(),
+                inventory_generation,
             )?;
             summary.merge_from(tail_summary);
         } else {
@@ -160,6 +171,7 @@ pub(crate) fn import_incremental_codex_session_tree(
                         std::slice::from_ref(session),
                         &error,
                         catalog_import_error_status(&err),
+                        inventory_generation,
                     )?;
                     if failure_scope == ImportFailureScope::System {
                         return Err(err);
@@ -171,7 +183,12 @@ pub(crate) fn import_incremental_codex_session_tree(
                     continue;
                 }
             };
-            mark_catalog_sessions_result(store, std::slice::from_ref(session), &file_summary)?;
+            mark_catalog_sessions_result(
+                store,
+                std::slice::from_ref(session),
+                &file_summary,
+                inventory_generation,
+            )?;
             summary.merge_from(file_summary);
         }
     }
@@ -196,6 +213,7 @@ pub(crate) fn mark_catalog_sessions_result(
     store: &Store,
     sessions: &[CatalogSession],
     summary: &ProviderImportSummary,
+    inventory_generation: u64,
 ) -> Result<()> {
     let indexed_at_ms = utc_now().timestamp_millis();
     let event_count = if sessions.len() == 1 {
@@ -217,6 +235,7 @@ pub(crate) fn mark_catalog_sessions_result(
             indexed_at_ms,
             status,
             error.as_deref(),
+            inventory_generation,
         )?;
     }
     Ok(())
@@ -229,6 +248,7 @@ pub(crate) fn mark_catalog_session_result(
     indexed_at_ms: i64,
     status: CatalogIndexedStatus,
     error: Option<&str>,
+    inventory_generation: u64,
 ) -> Result<()> {
     let file_sha256 = if status == CatalogIndexedStatus::Indexed {
         let hash = sha256_file_prefix_hex(Path::new(&session.source_path), session.file_size_bytes)
@@ -242,6 +262,7 @@ pub(crate) fn mark_catalog_session_result(
                     std::slice::from_ref(session),
                     &durable_error,
                     catalog_import_error_status(&err),
+                    inventory_generation,
                 )?;
                 return Err(err);
             }
@@ -257,6 +278,7 @@ pub(crate) fn mark_catalog_session_result(
             file_size_bytes: session.file_size_bytes,
             file_modified_at_ms: session.file_modified_at_ms,
             import_revision: session.import_revision,
+            inventory_generation,
             file_sha256: file_sha256.as_deref(),
             event_count,
             indexed_at_ms,
@@ -304,6 +326,7 @@ pub(crate) fn mark_catalog_sessions_error(
     sessions: &[CatalogSession],
     error: &str,
     status: CatalogIndexedStatus,
+    inventory_generation: u64,
 ) -> Result<()> {
     let indexed_at_ms = utc_now().timestamp_millis();
     for session in sessions {
@@ -315,6 +338,7 @@ pub(crate) fn mark_catalog_sessions_error(
                 file_size_bytes: session.file_size_bytes,
                 file_modified_at_ms: session.file_modified_at_ms,
                 import_revision: session.import_revision,
+                inventory_generation,
                 file_sha256: None,
                 event_count: None,
                 indexed_at_ms,

crates/ctx-cli/src/commands/import/inventory.rs

@@ -63,13 +63,13 @@ pub(crate) fn inventory_import_sources(
             .source_bytes
             .saturating_add(plan.stats.bytes);
         match &plan.preinventory {
-            SourcePreinventory::SourceImportFiles(files) => {
+            SourcePreinventory::SourceImportFiles { files, .. } => {
                 inventory.totals.source_import_files += files.len();
             }
-            SourcePreinventory::SourceRoot(_) => {
+            SourcePreinventory::SourceRoot { .. } => {
                 inventory.totals.source_import_files += 1;
             }
-            SourcePreinventory::None | SourcePreinventory::CodexSessionCatalog(_) => {}
+            SourcePreinventory::None | SourcePreinventory::CodexSessionCatalog { .. } => {}
         }
         if let Some((summary, source_json)) = cataloged {
             inventory.catalog.add(&summary);
@@ -103,11 +103,15 @@ fn inventory_import_source(
     source: SourceInfo,
 ) -> Result<(PlannedImportSource, Option<(CatalogSummary, Value)>)> {
     if is_incremental_codex_session_tree(&source) {
+        let source_root = source.path.display().to_string();
+        let inventory_generation =
+            store.allocate_catalog_inventory_generation(CaptureProvider::Codex, &source_root)?;
         let summary = catalog_codex_session_tree(
             &source.path,
             store,
             CodexSessionCatalogOptions {
                 source_root: Some(source.path.clone()),
+                observation_generation: Some(inventory_generation),
                 ..CodexSessionCatalogOptions::default()
             },
         )
@@ -120,7 +124,10 @@ fn inventory_import_source(
         let plan = PlannedImportSource {
             source,
             stats,
-            preinventory: SourcePreinventory::CodexSessionCatalog(summary.clone()),
+            preinventory: SourcePreinventory::CodexSessionCatalog {
+                summary: summary.clone(),
+                inventory_generation,
+            },
         };
         let source_json = json!({
             "provider": plan.source.provider.as_str(),
@@ -138,29 +145,46 @@ fn inventory_import_source(
     }
 
     if source_uses_import_file_manifest(&source) {
+        let source_root = source.path.display().to_string();
+        let inventory_generation =
+            store.allocate_source_import_inventory_generation(source.provider, &source_root)?;
         let files = collect_source_import_files(&source)
             .with_context(|| format!("inventory import files from {}", source.path.display()))?;
-        persist_source_import_files(store, &source, &files)?;
+        persist_source_import_files(store, &source, inventory_generation, &files)?;
         let stats = source_stats_from_import_files(&files);
         return Ok((
             PlannedImportSource {
                 source,
                 stats,
-                preinventory: SourcePreinventory::SourceImportFiles(files),
+                preinventory: SourcePreinventory::SourceImportFiles {
+                    files,
+                    inventory_generation,
+                },
             },
             None,
         ));
     }
 
+    let source_root = source.path.display().to_string();
+    let inventory_generation =
+        store.allocate_source_import_inventory_generation(source.provider, &source_root)?;
     let stats = source_stats(&source.path)
         .with_context(|| format!("inventory import source {}", source.path.display()))?;
     let root_file = source_root_import_file(&source, stats)?;
-    persist_source_import_files(store, &source, std::slice::from_ref(&root_file))?;
+    persist_source_import_files(
+        store,
+        &source,
+        inventory_generation,
+        std::slice::from_ref(&root_file),
+    )?;
     Ok((
         PlannedImportSource {
             source,
             stats,
-            preinventory: SourcePreinventory::SourceRoot(root_file),
+            preinventory: SourcePreinventory::SourceRoot {
+                file: root_file,
+                inventory_generation,
+            },
         },
         None,
     ))

crates/ctx-cli/src/commands/import/manifest.rs

@@ -6,6 +6,7 @@ use crate::commands::import::catalog::system_time_ms;
 pub(crate) fn persist_source_import_files(
     store: &Store,
     source: &SourceInfo,
+    inventory_generation: u64,
     files: &[SourceImportFile],
 ) -> Result<()> {
     let source_root = source.path.display().to_string();
@@ -16,12 +17,13 @@ pub(crate) fn persist_source_import_files(
     let observed_at_ms = utc_now().timestamp_millis();
     store.begin_immediate_batch()?;
     let persist = (|| -> Result<()> {
-        store.upsert_source_import_files(files)?;
+        store.upsert_source_import_files(inventory_generation, files)?;
         store.mark_source_import_missing_paths_stale(
             source.provider,
             &source_root,
             &current_paths,
             observed_at_ms,
+            inventory_generation,
         )?;
         Ok(())
     })();