diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs index 3a5e4f37ec..d1b0c8bb2d 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/serving.rs @@ -189,6 +189,7 @@ pub struct CodeIndexBuildProgressSlotStateV1 { generation_id: Option, pub(super) owner_epoch: u64, progress_epoch: u64, + last_progress_micros: Option, snapshot: Option>, } @@ -198,27 +199,61 @@ impl CodeIndexBuildProgressSlotStateV1 { self.progress_epoch = self.progress_epoch.saturating_add(1).max(1); self.generation_id = Some(generation_id); self.snapshot = None; + self.last_progress_micros = None; self.owner_epoch } + /// A discarded or replaced staging file starts a different committed + /// sequence, even if the sealed generation and scheduler owner survive. + /// Only the head-opening singleflight may reset it, after it has verified + /// that a new staging builder was actually created. The next boundary + /// keeps the monotonic publication epoch and reports its real counters. + pub(super) fn begin_staging_attempt( + &mut self, + generation_id: &CodeGenerationId, + owner_epoch: u64, + ) -> bool { + if self.generation_id.as_ref() != Some(generation_id) || self.owner_epoch != owner_epoch { + return false; + } + self.snapshot = None; + true + } + pub(super) fn publish( &mut self, generation_id: &CodeGenerationId, owner_epoch: u64, mut snapshot: CodeIndexBuildProgressV1, ) -> bool { - if self.generation_id.as_ref() != Some(generation_id) || self.owner_epoch != owner_epoch { + if self.generation_id.as_ref() != Some(generation_id) + || self.owner_epoch != owner_epoch + || snapshot.generation_id != generation_id.as_str() + { return false; } - // A progress producer can timestamp its snapshot before another - // producer publishes. Preserve the most recent observed timestamp - // even when the later lock acquisition delivers an older sample. - // Replacing the generation clears the snapshot and starts a new owner. if let Some(current) = self.snapshot.as_ref() { - snapshot.last_progress_micros = snapshot - .last_progress_micros - .max(current.last_progress_micros); + // Concurrent phase/report producers may complete out of order. + // A delayed report must not undo already committed boundaries or + // withdraw a ready text owner from the same generation. + if snapshot.committed_pages < current.committed_pages + || snapshot.committed_chunks < current.committed_chunks + || snapshot.committed_imports < current.committed_imports + || snapshot.committed_payload_bytes < current.committed_payload_bytes + || (current.phase == CodeIndexBuildPhaseV1::Ready + && snapshot.phase != CodeIndexBuildPhaseV1::Ready) + { + return false; + } } + // The next producer may have assembled its report before an earlier + // producer acquired the lock. Keep the owner's latest observed time + // even across a verified staging restart (which clears the snapshot). + // Replacing the owner clears this timestamp authority. + if let Some(latest) = self.last_progress_micros { + snapshot.last_progress_micros = snapshot.last_progress_micros.max(latest); + } + self.last_progress_micros = Some(snapshot.last_progress_micros); self.progress_epoch = self.progress_epoch.saturating_add(1).max(1); snapshot.progress_epoch = self.progress_epoch; @@ -233,6 +268,9 @@ impl CodeIndexBuildProgressSlotStateV1 { else { return; }; + if snapshot.blocked_reason == Some(reason) { + return; + } let mut snapshot = snapshot.as_ref().clone(); snapshot.blocked_reason = Some(reason); let _ = self.publish(&generation_id, self.owner_epoch, snapshot); @@ -252,6 +290,7 @@ impl CodeIndexBuildProgressSlotStateV1 { return false; }; if current.phase == CodeIndexBuildPhaseV1::Ready + || current.blocked_reason == Some(reason) || expected_progress_epoch.is_some_and(|epoch| current.progress_epoch != epoch) { return false; @@ -2693,6 +2732,17 @@ impl LatestCodeTextGenerationV1 { let Some(current) = slot.snapshot() else { return; }; + // A retry that re-enters an unchanged phase has not made + // progress. Keep the last committed observation epoch and + // elapsed sample until a new phase or batch boundary arrives. + if current.phase == phase + && current.current_batch_pages == current_batch_pages + && current.current_batch_payload_bytes == current_batch_payload_bytes + && !(phase == CodeIndexBuildPhaseV1::Ready + && current.blocked_reason.is_some()) + { + return; + } let mut snapshot = current.as_ref().clone(); snapshot.phase = phase; if !phase_has_file_rate_estimate(phase) { @@ -3220,6 +3270,10 @@ impl LatestCodeTextGenerationV1 { } else { None }; + // A missing/incompatible staging file is a new durable attempt for + // this sealed generation. Its committed counters legitimately start + // at zero; the old attempt must not fence that fresh boundary. + let mut fresh_staging_attempt = resumed.is_none(); let mut builder = match resumed { Some(builder) => builder, None => match self.carry_parent_text_artifact( @@ -3262,6 +3316,7 @@ impl LatestCodeTextGenerationV1 { Err(VerifiedSealedLexicalCursorRestoreErrorV1::IncompatiblePosition) => { drop(builder); store.discard_incompatible_staging(&staging_path, control)?; + fresh_staging_attempt = true; builder = CodeLexicalArtifactBuilderV1::create_with_memory_budget( &staging_path, metadata, @@ -3288,6 +3343,15 @@ impl LatestCodeTextGenerationV1 { staging_path, _build_reservation: build_reservation, }; + if fresh_staging_attempt + && !self + .text_progress_slot + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .begin_staging_attempt(&generation_id, self.text_progress_owner_epoch) + { + return Err(RetrievalPortError::GenerationMismatch); + } self.publish_text_progress_boundary( &initialized, &progress, diff --git a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs index 6a48bb325e..b33a45fe79 100644 --- a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs +++ b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs @@ -3041,6 +3041,248 @@ fn progress_timestamp_never_regresses_within_one_owner() { assert_eq!(slot.snapshot().unwrap().last_progress_micros, 42); } +#[test] +fn delayed_progress_cannot_undo_committed_boundaries_or_ready() { + let generation = CodeGenerationId::new("generation.stale-progress").expect("generation"); + let mut slot = super::super::CodeIndexBuildProgressSlotStateV1::default(); + let owner = slot.replace_generation(generation.clone()); + + let mut committed = progress_snapshot_for_generation(&generation, 6); + committed.committed_imports = 3; + committed.last_progress_micros = 600; + assert!(slot.publish(&generation, owner, committed)); + let accepted = slot.snapshot().expect("accepted committed progress"); + + for counter in ["pages", "chunks", "imports", "payload"] { + let mut delayed = accepted.as_ref().clone(); + delayed.last_progress_micros = 700; + match counter { + "pages" => delayed.committed_pages -= 1, + "chunks" => delayed.committed_chunks -= 1, + "imports" => delayed.committed_imports -= 1, + "payload" => delayed.committed_payload_bytes -= 1, + _ => unreachable!(), + } + assert!( + !slot.publish(&generation, owner, delayed), + "delayed {counter} cannot overwrite committed progress" + ); + assert_eq!( + slot.snapshot().expect("progress after stale report").as_ref(), + accepted.as_ref(), + "refusing stale {counter} must preserve the epoch and last progress time" + ); + } + + let mut wrong_generation = accepted.as_ref().clone(); + wrong_generation.generation_id = "generation.foreign".to_owned(); + assert!(!slot.publish(&generation, owner, wrong_generation)); + assert_eq!( + slot.snapshot().expect("correct generation still owns progress").as_ref(), + accepted.as_ref() + ); + + let mut ready = progress_snapshot_for_generation(&generation, 7); + ready.committed_imports = 3; + ready.phase = CodeIndexBuildPhaseV1::Ready; + ready.last_progress_micros = 650; + assert!(slot.publish(&generation, owner, ready)); + let settled = slot.snapshot().expect("ready progress"); + + let mut late_build = settled.as_ref().clone(); + late_build.phase = CodeIndexBuildPhaseV1::BulkCommit; + late_build.last_progress_micros = 700; + assert!(!slot.publish(&generation, owner, late_build)); + assert_eq!( + slot.snapshot().expect("ready progress remains").as_ref(), + settled.as_ref(), + "a delayed build phase must not withdraw ready" + ); + + // A remounted owner starts a new committed sequence even when its + // generation identity is the same. The old owner must stay fenced off. + let replacement_owner = slot.replace_generation(generation.clone()); + let fresh = progress_snapshot_for_generation(&generation, 1); + assert!(slot.publish(&generation, replacement_owner, fresh)); + assert_eq!(slot.snapshot().expect("new owner").committed_pages, 1); + assert!(!slot.publish( + &generation, + owner, + progress_snapshot_for_generation(&generation, 99) + )); +} + +#[test] +fn a_new_staging_attempt_can_truthfully_restart_committed_progress() { + let generation = CodeGenerationId::new("generation.staging-retry").expect("generation"); + let mut slot = super::super::CodeIndexBuildProgressSlotStateV1::default(); + let owner = slot.replace_generation(generation.clone()); + + let mut old = progress_snapshot_for_generation(&generation, 12); + old.last_progress_micros = 500; + assert!(slot.publish(&generation, owner, old)); + let old_epoch = slot.snapshot().expect("first staging attempt").progress_epoch; + + // Missing/incompatible SQLite staging is a verified new build attempt, + // not a delayed report from an older commit boundary. Without resetting + // the slot, rejecting counter regressions would leave status frozen. + assert!(slot.begin_staging_attempt(&generation, owner)); + let mut restarted = progress_snapshot_for_generation(&generation, 0); + // A wall-clock adjustment or delayed sample must not rewind the + // timestamp even though a verified new staging attempt reset its counts. + restarted.last_progress_micros = 400; + restarted.phase = CodeIndexBuildPhaseV1::SourceScan; + assert!(slot.publish(&generation, owner, restarted)); + let restarted = slot.snapshot().expect("new staging attempt"); + assert_eq!(restarted.committed_pages, 0); + assert_eq!(restarted.phase, CodeIndexBuildPhaseV1::SourceScan); + assert_eq!( + restarted.last_progress_micros, 500, + "staging restart must not rewind the same owner's last progress timestamp" + ); + assert!( + restarted.progress_epoch > old_epoch, + "a new verified staging attempt must produce a newer report epoch" + ); + + let mut resumed = progress_snapshot_for_generation(&generation, 2); + resumed.last_progress_micros = 700; + assert!(slot.publish(&generation, owner, resumed)); + assert_eq!(slot.snapshot().expect("resumed commits").committed_pages, 2); + + let foreign = CodeGenerationId::new("generation.foreign-stage").expect("foreign"); + assert!(!slot.begin_staging_attempt(&foreign, owner)); + assert!(!slot.begin_staging_attempt(&generation, owner + 1)); + assert_eq!(slot.snapshot().expect("unchanged latest").committed_pages, 2); +} + +#[test] +fn retrying_an_unchanged_phase_does_not_fabricate_status_progress() { + let fixture = GitFixture::new(&[( + "src/lib.rs", + "pub fn caller() { callee(); }\npub fn callee() {}\n", + )]); + let store = TempDir::new().expect("store root"); + let mut scheduler = scheduler( + &fixture, + store.path().to_path_buf(), + Arc::new(SharedCodeIndexBytePoolV1::default()), + ); + published(scheduler.reconcile_now().expect("publish")); + let latest = scheduler.latest_complete().expect("latest generation"); + assert!( + !latest.advance_text_serving(0).expect("open text projection"), + "a zero-work admission must leave the projection pending" + ); + let before = build_progress_snapshot(&scheduler); + assert_ne!(before.phase, CodeIndexBuildPhaseV1::Ready); + + latest.text.publish_text_progress_phase( + before.phase, + before.current_batch_pages, + before.current_batch_payload_bytes, + ); + let repeated = build_progress_snapshot(&scheduler); + assert_eq!( + repeated.progress_epoch, before.progress_epoch, + "a retry reporting no new phase or batch boundary must not feign progress" + ); + assert_eq!(repeated.last_progress_micros, before.last_progress_micros); + assert_eq!(repeated.committed_pages, before.committed_pages); + + let next_phase = if before.phase == CodeIndexBuildPhaseV1::Verification { + CodeIndexBuildPhaseV1::GraphPublication + } else { + CodeIndexBuildPhaseV1::Verification + }; + latest.text.publish_text_progress_phase( + next_phase, + before.current_batch_pages, + before.current_batch_payload_bytes, + ); + let transitioned = build_progress_snapshot(&scheduler); + assert_eq!(transitioned.phase, next_phase); + assert!(transitioned.progress_epoch > before.progress_epoch); + latest.text.publish_text_progress_phase( + next_phase, + before.current_batch_pages, + before.current_batch_payload_bytes, + ); + let unchanged = build_progress_snapshot(&scheduler); + assert_eq!(unchanged.progress_epoch, transitioned.progress_epoch); + assert_eq!(unchanged.last_progress_micros, transitioned.last_progress_micros); + + let mut advances = 0; + while !latest.advance_text_serving(8).expect("resume durable text build") { + advances += 1; + assert!(advances < 1_000, "text build failed to finish"); + } + let ready = build_progress_snapshot(&scheduler); + assert_eq!(ready.phase, CodeIndexBuildPhaseV1::Ready); + assert!(ready.progress_epoch > transitioned.progress_epoch); + + // Ready is the one idempotent-looking phase that may still be a real + // state change: success must clear a previously published refusal. + scheduler + .build_progress_slot() + .write() + .expect("progress slot") + .block_current(CodeIndexBuildBlockedReasonV1::ResidentMemory); + let blocked = build_progress_snapshot(&scheduler); + assert_eq!( + blocked.blocked_reason, + Some(CodeIndexBuildBlockedReasonV1::ResidentMemory) + ); + latest.text.publish_text_progress_phase( + CodeIndexBuildPhaseV1::Ready, + blocked.current_batch_pages, + blocked.current_batch_payload_bytes, + ); + let cleared = build_progress_snapshot(&scheduler); + assert_eq!(cleared.blocked_reason, None); + assert!(cleared.progress_epoch > blocked.progress_epoch); +} + +#[test] +fn repeated_blocked_reason_is_not_reported_as_new_progress() { + let generation = CodeGenerationId::new("generation.blocked-report").expect("generation"); + let mut slot = super::super::CodeIndexBuildProgressSlotStateV1::default(); + let owner = slot.replace_generation(generation.clone()); + assert!(slot.publish( + &generation, + owner, + progress_snapshot_for_generation(&generation, 2) + )); + slot.block_current(CodeIndexBuildBlockedReasonV1::ResidentMemory); + let first = slot.snapshot().expect("first refusal"); + assert_eq!( + first.blocked_reason, + Some(CodeIndexBuildBlockedReasonV1::ResidentMemory) + ); + + slot.block_current(CodeIndexBuildBlockedReasonV1::ResidentMemory); + assert!(!slot.block_owned_text( + &generation, + owner, + CodeIndexBuildBlockedReasonV1::ResidentMemory, + Some(first.progress_epoch) + )); + assert_eq!( + slot.snapshot().expect("repeated refusal").progress_epoch, + first.progress_epoch, + "retries without a new refusal or committed work must not inflate the observation epoch" + ); + + let mut committed = progress_snapshot_for_generation(&generation, 3); + committed.last_progress_micros = first.last_progress_micros + 1; + assert!(slot.publish(&generation, owner, committed)); + assert_eq!( + slot.snapshot().expect("committed progress").blocked_reason, + None, + "durable progress clears the previous blocked verdict" + ); +} + #[test] fn progress_owner_epoch_rejects_original_a_after_a_b_a_replacement() { let generation_a = CodeGenerationId::new("generation.progress-a").expect("generation A");