fix(code-index): bound cold builds and resume refused publication

crates/tracedecay-code-index-retention/src/code_index_generations.rs

@@ -33,6 +33,10 @@ use tracedecay_domain::{
     CodeGenerationId, ManifestDigest, UtcMicros, canonical_sha256, sha256_hex_suffix,
 };
 
+/// Temporary publication directories are protected by the shared project
+/// segment lease. Only the exclusive segment sweep may reclaim abandoned ones.
+pub const CODE_INDEX_PREPARATION_DIRECTORY_PREFIX_V1: &str = ".code-index-preparation-";
+
 mod file_quarantine;
 #[cfg(any(test, feature = "test-helpers"))]
 pub mod fixture;
@@ -1487,6 +1491,55 @@ fn sweep_unreferenced_generation_segments(
         let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
             continue;
         };
+        if file_name.starts_with(CODE_INDEX_PREPARATION_DIRECTORY_PREFIX_V1) {
+            let metadata = path.symlink_metadata().map_err(deferred_if_absent)?;
+            if !metadata.file_type().is_dir() {
+                return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!(
+                    "generation preparation '{}' is not a regular directory",
+                    path.display()
+                )));
+            }
+            found = true;
+            if !apply {
+                return Ok((true, false));
+            }
+            // The caller holds the exclusive project segment lock, which
+            // excludes every live preparation's shared lease. Runtime staging
+            // is flat; refusing other entry types avoids following any links.
+            let mut complete = true;
+            for staged in std::fs::read_dir(&path).map_err(storage)? {
+                if is_cancelled() {
+                    return Err(CodeGenerationRetentionErrorV1::Cancelled);
+                }
+                charge_entry_budget(&mut remaining_entries)?;
+                let staged = staged.map_err(storage)?;
+                let staged_path = staged.path();
+                let metadata = staged_path.symlink_metadata().map_err(deferred_if_absent)?;
+                if !metadata.file_type().is_file() {
+                    return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!(
+                        "generation preparation entry '{}' is not a regular file",
+                        staged_path.display()
+                    )));
+                }
+                std::fs::remove_file(&staged_path).map_err(storage)?;
+                reclaimed = reclaimed.saturating_add(metadata.len());
+                reclaimed_segments += 1;
+                if reclaimed_segments == MAX_CODE_GENERATION_RETENTION_BATCH_V1 {
+                    complete = false;
+                    break;
+                }
+            }
+            if complete {
+                std::fs::remove_dir(&path).map_err(storage)?;
+                reclaimed_segments += 1;
+            } else {
+                sync_directory(&path)?;
+            }
+            if reclaimed_segments >= MAX_CODE_GENERATION_RETENTION_BATCH_V1 {
+                break;
+            }
+            continue;
+        }
         let Some(digest) = file_name
             .strip_prefix("segment-")
             .and_then(|name| name.strip_suffix(".json"))
@@ -1517,12 +1570,12 @@ fn sweep_unreferenced_generation_segments(
             break;
         }
     }
-    if reclaimed > 0 {
+    if reclaimed > 0 || reclaimed_segments > 0 {
         sync_directory(&segments_root)?;
     }
     Ok((
         found,
-        reclaimed_segments == MAX_CODE_GENERATION_RETENTION_BATCH_V1,
+        reclaimed_segments >= MAX_CODE_GENERATION_RETENTION_BATCH_V1,
     ))
 }
 

crates/tracedecay-code-index-retention/src/code_index_generations/tests.rs

@@ -3790,3 +3790,212 @@ fn store_release_wait_outlasts_a_reader() {
             .is_some()
     );
 }
+
+#[test]
+fn abandoned_preparations_are_collected_only_after_shared_project_publication_leases_end() {
+    let project = tempfile::TempDir::new().expect("project store");
+    let scope = project
+        .path()
+        .join(code_index_scope_hash(Path::new("/fixture/main")));
+    let peer = project
+        .path()
+        .join(code_index_scope_hash(Path::new("/fixture/peer")));
+    std::fs::create_dir_all(scope.join(GENERATIONS_DIRECTORY)).expect("main scope");
+    std::fs::create_dir_all(peer.join(GENERATIONS_DIRECTORY)).expect("peer scope");
+    let segments = code_generation_segments_root(&scope);
+    std::fs::create_dir_all(&segments).expect("segments root");
+    // A process crash leaves the directory after its TempDir owner and shared
+    // lease disappear. Keep the fixture directory to exercise that filesystem
+    // state through the real planner and executor.
+    let abandoned = tempfile::Builder::new()
+        .prefix(CODE_INDEX_PREPARATION_DIRECTORY_PREFIX_V1)
+        .tempdir_in(&segments)
+        .expect("publication staging")
+        .keep();
+    std::fs::write(
+        abandoned.join("generation-evidence.tmp"),
+        b"interrupted publication",
+    )
+    .expect("abandoned pack");
+    let unrelated = segments.join("unrelated-directory");
+    std::fs::create_dir(&unrelated).expect("unrelated directory");
+    std::fs::write(unrelated.join("payload"), b"preserve").expect("unrelated payload");
+    let lease = acquire_generation_segments_publication_lock(&peer, &|| false)
+        .expect("live peer publication lease");
+    let plan =
+        plan_code_generation_retention(&scope, &BTreeSet::new()).expect("orphan-only census");
+    assert!(plan.has_collectable_work());
+    assert!(plan.collectable_generations.is_empty());
+    let result = execute_code_generation_retention_cancellable(
+        &scope,
+        plan,
+        CodeGenerationRetentionModeV1::Apply,
+        UtcMicros(1),
+        None,
+        &|| false,
+    );
+    assert!(matches!(
+        result,
+        Err(CodeGenerationRetentionErrorV1::GenerationStoreBusy)
+    ));
+    assert!(abandoned.join("generation-evidence.tmp").is_file());
+
+    let plan =
+        plan_code_generation_retention(&scope, &BTreeSet::new()).expect("protected orphan census");
+    let prior_parks = locking::store_lock_park_count();
+    let result = execute_code_generation_retention_cancellable(
+        &scope,
+        plan,
+        CodeGenerationRetentionModeV1::Apply,
+        UtcMicros(1),
+        None,
+        &|| locking::store_lock_park_count() > prior_parks,
+    );
+    assert!(matches!(
+        result,
+        Err(CodeGenerationRetentionErrorV1::Cancelled)
+    ));
+    assert!(
+        locking::store_lock_park_count() > prior_parks,
+        "the canonical cancellation callback observes a contended lock wait"
+    );
+    assert!(abandoned.join("generation-evidence.tmp").is_file());
+    drop(lease);
+
+    let plan = prepare_next_code_generation_retention_cancellable(
+        &scope,
+        &BTreeSet::new(),
+        &|| false,
+        None,
+    )
+    .expect("ordinary maintenance finds stranded staging");
+    assert!(plan.has_collectable_work());
+    let report = execute_code_generation_retention(
+        &scope,
+        plan,
+        CodeGenerationRetentionModeV1::Apply,
+        UtcMicros(2),
+        None,
+    )
+    .expect("reclaim abandoned staging");
+    assert!(report.deleted_generations.is_empty());
+    assert!(!abandoned.exists());
+    assert_eq!(
+        std::fs::read(unrelated.join("payload")).expect("unrelated bytes"),
+        b"preserve"
+    );
+}
+
+#[test]
+fn preparation_directory_collection_observes_cancellation_entry_and_batch_bounds() {
+    let store = tempfile::TempDir::new().expect("store");
+    let segments = code_generation_segments_root(store.path());
+    std::fs::create_dir_all(&segments).expect("segments root");
+    let abandoned = tempfile::Builder::new()
+        .prefix(CODE_INDEX_PREPARATION_DIRECTORY_PREFIX_V1)
+        .tempdir_in(&segments)
+        .expect("staging")
+        .keep();
+    for index in 0..MAX_CODE_GENERATION_RETENTION_BATCH_V1 + 2 {
+        std::fs::write(abandoned.join(format!("segment-{index}.json")), b"").expect("staged file");
+    }
+    let mut budget = 0;
+    assert!(matches!(
+        has_unreferenced_generation_segments_budgeted(store.path(), None, &|| false, &mut budget),
+        Err(CodeGenerationRetentionErrorV1::EntryBudgetExceeded)
+    ));
+    assert!(matches!(
+        has_unreferenced_generation_segments(store.path(), None, &|| true),
+        Err(CodeGenerationRetentionErrorV1::Cancelled)
+    ));
+    assert_eq!(
+        std::fs::read_dir(&abandoned).expect("staged files").count(),
+        MAX_CODE_GENERATION_RETENTION_BATCH_V1 + 2
+    );
+    let first = run_code_generation_retention(
+        store.path(),
+        &BTreeSet::new(),
+        CodeGenerationRetentionModeV1::Apply,
+        UtcMicros(1),
+        None,
+    )
+    .expect("first bounded collection");
+    assert!(first.generation_segment_batch_exhausted);
+    assert_eq!(
+        std::fs::read_dir(&abandoned)
+            .expect("remaining staging")
+            .count(),
+        2
+    );
+    run_code_generation_retention(
+        store.path(),
+        &BTreeSet::new(),
+        CodeGenerationRetentionModeV1::Apply,
+        UtcMicros(2),
+        None,
+    )
+    .expect("finish bounded collection");
+    assert!(!abandoned.exists());
+    let empty = tempfile::Builder::new()
+        .prefix(CODE_INDEX_PREPARATION_DIRECTORY_PREFIX_V1)
+        .tempdir_in(&segments)
+        .expect("empty abandoned staging")
+        .keep();
+    let plan = prepare_next_code_generation_retention_cancellable(
+        store.path(),
+        &BTreeSet::new(),
+        &|| false,
+        None,
+    )
+    .expect("empty directory is collectable work");
+    assert!(plan.has_collectable_work());
+    execute_code_generation_retention(
+        store.path(),
+        plan,
+        CodeGenerationRetentionModeV1::Apply,
+        UtcMicros(3),
+        None,
+    )
+    .expect("collect empty staging");
+    assert!(!empty.exists());
+}
+
+#[cfg(unix)]
+#[test]
+fn preparation_collection_refuses_symbolic_links_without_following_them() {
+    let store = tempfile::TempDir::new().expect("store");
+    let segments = code_generation_segments_root(store.path());
+    std::fs::create_dir_all(&segments).expect("segments root");
+    let outside = tempfile::TempDir::new().expect("outside staging authority");
+    let payload = outside.path().join("payload");
+    std::fs::write(&payload, b"preserve").expect("outside bytes");
+    let link = segments.join(format!("{CODE_INDEX_PREPARATION_DIRECTORY_PREFIX_V1}link"));
+    std::os::unix::fs::symlink(outside.path(), &link).expect("directory symlink");
+    assert!(matches!(
+        plan_code_generation_retention(store.path(), &BTreeSet::new()),
+        Err(CodeGenerationRetentionErrorV1::UnsafeState(_))
+    ));
+    std::fs::remove_file(&link).expect("remove symbolic directory entry");
+    let staging = tempfile::Builder::new()
+        .prefix(CODE_INDEX_PREPARATION_DIRECTORY_PREFIX_V1)
+        .tempdir_in(&segments)
+        .expect("staging directory");
+    std::os::unix::fs::symlink(&payload, staging.path().join("segment.json"))
+        .expect("file symlink");
+    let plan =
+        plan_code_generation_retention(store.path(), &BTreeSet::new()).expect("stage census");
+    assert!(matches!(
+        execute_code_generation_retention(
+            store.path(),
+            plan,
+            CodeGenerationRetentionModeV1::Apply,
+            UtcMicros(1),
+            None
+        ),
+        Err(CodeGenerationRetentionErrorV1::UnsafeState(_))
+    ));
+    assert_eq!(
+        std::fs::read(&payload).expect("outside payload survives"),
+        b"preserve"
+    );
+}

crates/tracedecay-code-index-runtime/src/code_index_scheduler/git_tree_capture.rs

@@ -656,6 +656,7 @@ impl CodeIndexWorktreeSchedulerV1 {
             retained_bytes: _retained_bytes,
             retained_reservations: _retained_reservations,
         } = captured;
+        drop(_retained_bytes);
         let requested_scope = CodeIndexGenerationScopeV1::for_snapshot(&snapshot);
         // Retained-history generations live inside the active publication
         // pointer, so they need an active generation to ride on. A store with
@@ -698,6 +699,7 @@ impl CodeIndexWorktreeSchedulerV1 {
         let generation = owner
             .build_and_publish(
                 CodeIndexBuildRequestV1 {
+                    resident_memory: Arc::clone(&self.resident_memory),
                     snapshot,
                     captured_files,
                     changed_files: changed_paths,

crates/tracedecay-code-index-runtime/src/code_index_scheduler/graph_activation.rs

@@ -633,7 +633,7 @@ impl CodeGraphActivationAuthorityV1 {
                         ),
                     ));
                 }
-                native_runtime::spawn_blocking(move || latest.prewarm_serving_derivations())
+                native_runtime::spawn_blocking(move || latest.prewarm_graph_derivations())
                     .await
                     .map_err(|error| {
                         CodeIndexSchedulerErrorV1::GraphActivation(format!(

crates/tracedecay-code-index-runtime/src/code_index_scheduler/ignored_dependencies.rs

@@ -3,6 +3,7 @@
 use std::collections::BTreeSet;
 use std::io::Read;
 use std::path::{Component, Path, PathBuf};
+use std::sync::Arc;
 
 use gix::bstr::ByteSlice;
 use serde_json::Value;
@@ -256,7 +257,8 @@ impl CodeIndexWorktreeSchedulerV1 {
         previous_roster: Vec<CodeIndexIgnoredSourceAdmissionV1>,
         control: &dyn CodeIndexExecutionControlV1,
     ) -> Result<CodeIndexIgnoredDependencyBuildV1, CodeIndexSchedulerErrorV1> {
-        let began_under = self.epoch.load(std::sync::atomic::Ordering::Acquire);
+        let source_control = self.reconcile_control();
+        let began_under = source_control.expected_epoch();
         let _workers = self.ensure_worker_plan()?;
         let _worker_memory = match self.reserve_worker_memory() {
             Ok(reservation) => reservation,
@@ -305,8 +307,14 @@ impl CodeIndexWorktreeSchedulerV1 {
                 return Err(error);
             }
         };
+        drop(retained_bytes);
+        *self.owner.publication_mut() = self
+            .publication
+            .clone()
+            .with_reconcile_publication_fence(Arc::clone(&self.hints), source_control.clone());
         let generation = match self.owner.build_and_publish(
             CodeIndexBuildRequestV1 {
+                resident_memory: Arc::clone(&self.resident_memory),
                 snapshot: snapshot.clone(),
                 captured_files,
                 changed_files: changed_paths,
@@ -320,12 +328,24 @@ impl CodeIndexWorktreeSchedulerV1 {
         ) {
             Ok(generation) => generation,
             Err(error) => {
+                let requirement = self.owner.take_reconstruction_requirement();
+                // Refused entrypoint admission restores its prior roster;
+                // no unpublished generation may later resurrect that entry.
+                drop(self.owner.take_unpublished());
+                let error = super::reconcile::discarded_construction_refusal(
+                    error,
+                    &source_control,
+                    requirement,
+                    false,
+                    std::iter::once(&_worker_memory).chain(&retained_reservations),
+                    self.resident_memory.admission_watermark_bytes(),
+                );
                 self.ignored_source_admissions = previous_roster;
-                return Err(error.into());
+                return Err(error);
             }
         };
         let generation_id = generation.manifest().generation_id.clone();
-        drop((retained_bytes, retained_reservations));
+        drop(retained_reservations);
         self.latest_content_identity = Some(generation.snapshot().content_identity.clone());
         self.mark_reconciled(
             SourceContentManifestV1::for_snapshot(generation.snapshot()),