diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 2dda1ad49..5848268bd 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -2148,6 +2148,7 @@ async fn controller_loop( | Err(WorkerError::WorkerStore(_)) | Err(WorkerError::InvalidState(_)) | Err(WorkerError::SegmentActivationIncomplete { .. }) + | Err(WorkerError::CompactionCleanupPending { .. }) ) { set_controller_state( &shared_state, @@ -2566,6 +2567,19 @@ where (WorkerStatus::Paused, shutdown_requested, false) } Err(e) => { + let status = if matches!( + &e, + WorkerError::SegmentActivationIncomplete { .. } + | WorkerError::CompactionCleanupPending { .. } + ) { + // Metadata has already committed the replacement, but + // one live authority failed to publish. Keep the + // controller visibly busy/fail-closed; publishing Idle + // would admit another turn against inconsistent state. + WorkerStatus::Running + } else { + WorkerStatus::Idle + }; let code = worker_error_code(&e); let message = e.to_string(); let _ = working_event_tx.send(Event::Error { @@ -2581,7 +2595,7 @@ where }, ); } - (WorkerStatus::Idle, shutdown_requested, false) + (status, shutdown_requested, false) } }; } @@ -3319,6 +3333,74 @@ mod tests { } } + #[tokio::test] + async fn segment_activation_failure_from_run_never_publishes_idle() { + let mut env = make_env().await; + let worker_future = async { + Err::(WorkerError::SegmentActivationIncomplete { + source: crate::runtime::worker_allocation::ScopeLockError::UnknownWorker( + "worker".into(), + ), + }) + }; + + let (status, shutdown, may_drain_pending) = drive_turn( + worker_future, + &mut env.method_rx, + &env.working_event_tx, + &env.cancel_tx, + &env.pause_tx, + &env.shared_state, + &env.runtime_dir, + None, + &env.notify_buffer, + &env.pending_submissions, + None, + "worker", + &env.spawned_registry, + false, + ) + .await; + + assert_eq!(status, WorkerStatus::Running); + assert!(!shutdown); + assert!(!may_drain_pending); + } + + #[tokio::test] + async fn pending_compaction_cleanup_from_run_never_publishes_idle() { + let mut env = make_env().await; + let worker_future = async { + Err::(WorkerError::CompactionCleanupPending { + source: crate::runtime::worker_allocation::ScopeLockError::UnknownWorker( + "worker".into(), + ), + }) + }; + + let (status, shutdown, may_drain_pending) = drive_turn( + worker_future, + &mut env.method_rx, + &env.working_event_tx, + &env.cancel_tx, + &env.pause_tx, + &env.shared_state, + &env.runtime_dir, + None, + &env.notify_buffer, + &env.pending_submissions, + None, + "worker", + &env.spawned_registry, + false, + ) + .await; + + assert_eq!(status, WorkerStatus::Running); + assert!(!shutdown); + assert!(!may_drain_pending); + } + #[tokio::test] async fn pause_waits_for_run_boundary_and_uses_safe_pause_channel() { let mut env = make_env().await; diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 7fc65102c..eec9f1077 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -4780,7 +4780,11 @@ impl Worker { completed, "automatic compaction must complete a claimed attempt" ); - if matches!(e, WorkerError::SegmentActivationIncomplete { .. }) { + if matches!( + &e, + WorkerError::SegmentActivationIncomplete { .. } + | WorkerError::CompactionCleanupPending { .. } + ) { Err(e) } else { Ok(()) @@ -5070,7 +5074,7 @@ impl Worker { mut cancel: Option>, trigger: CompactionTrigger, ) -> Result { - self.release_pending_compaction_service().await; + self.release_pending_compaction_service().await?; let _rewrite_guard = self .prepare_session_rewrite(SessionRewriteKind::Compact) .await?; @@ -5143,6 +5147,14 @@ impl Worker { Ok(new_segment_id) } Err(error) => { + let error = if matches!(&error, WorkerError::CompactionCleanupPending { .. }) { + error + } else { + match self.release_compaction_service(&mut lifecycle).await { + Ok(()) => error, + Err(cleanup_error) => cleanup_error, + } + }; let observed_segment_id = self.segment_state.location().segment_id; for metric in attempt.failure_metrics( observed_segment_id, @@ -5152,68 +5164,70 @@ impl Worker { self.try_record_metric(&metric); } lifecycle.revision = lifecycle.revision.saturating_add(1); - lifecycle.state = if matches!(error, WorkerError::CompactCancelled) { + lifecycle.state = if matches!(&error, WorkerError::CompactCancelled) { CompactionLifecycleState::Interrupted } else { CompactionLifecycleState::Failed }; lifecycle.ended_at_ms = Some(segment_log::now_millis()); - self.release_compaction_service(&mut lifecycle).await; - self.set_compaction_progress(None); + if !matches!(&error, WorkerError::CompactionCleanupPending { .. }) { + self.set_compaction_progress(None); + } Err(error) } } } - async fn release_pending_compaction_service(&self) { + async fn release_pending_compaction_service(&self) -> Result<(), WorkerError> { let cleanup = self .pending_compaction_cleanup .lock() .expect("pending compaction cleanup mutex poisoned") .clone(); let Some(cleanup) = cleanup else { - return; + return Ok(()); }; let Some(registry) = &self.internal_worker_registry else { - return; + return Err(WorkerError::CompactionCleanupPending { + source: ScopeLockError::Io(std::io::Error::other( + "compaction registry unavailable during cleanup", + )), + }); }; - loop { - match registry.stop_service(&cleanup.session_id).await { - Ok(_) => { - let mut pending = self - .pending_compaction_cleanup - .lock() - .expect("pending compaction cleanup mutex poisoned"); - if pending - .as_ref() - .is_some_and(|pending| pending.session_id == cleanup.session_id) - { - *pending = None; - } - return; - } - Err(error) => { - warn!( - compaction_id = %cleanup.compaction_id, - error = %error, - "compaction service cleanup failed; retaining cleanup authority and retrying" - ); - tokio::time::sleep(Duration::from_millis(50)).await; + match registry.stop_service(&cleanup.session_id).await { + Ok(_) => { + let mut pending = self + .pending_compaction_cleanup + .lock() + .expect("pending compaction cleanup mutex poisoned"); + if pending + .as_ref() + .is_some_and(|pending| pending.session_id == cleanup.session_id) + { + *pending = None; } + Ok(()) + } + Err(source) => { + warn!( + compaction_id = %cleanup.compaction_id, + error = %source, + "compaction service cleanup failed; retaining cleanup authority" + ); + Err(WorkerError::CompactionCleanupPending { + source: ScopeLockError::Io(source), + }) } } } - async fn release_compaction_service(&self, lifecycle: &mut CompactionLifecycle) { - self.release_pending_compaction_service().await; - if self - .pending_compaction_cleanup - .lock() - .expect("pending compaction cleanup mutex poisoned") - .is_none() - { - lifecycle.internal_worker = None; - } + async fn release_compaction_service( + &self, + lifecycle: &mut CompactionLifecycle, + ) -> Result<(), WorkerError> { + self.release_pending_compaction_service().await?; + lifecycle.internal_worker = None; + Ok(()) } async fn compact_impl( @@ -5761,7 +5775,7 @@ impl Worker { // activation so no internal process or registry authority survives into // the replacement's live lifetime. Cleanup failures retain this future // (and Busy state) until a retry succeeds. - self.release_compaction_service(lifecycle).await; + self.release_compaction_service(lifecycle).await?; self.set_compaction_progress(Some(InFlightCompaction { phase: CompactionPhase::Committing, @@ -7433,6 +7447,12 @@ pub enum WorkerError { source: ScopeLockError, }, + #[error("compaction cleanup is pending; retry or shut down the Worker")] + CompactionCleanupPending { + #[source] + source: ScopeLockError, + }, + #[error(transparent)] PromptCatalog(#[from] CatalogError), diff --git a/docs/design/compaction.md b/docs/design/compaction.md index 01abe4fd7..3a46434a8 100644 --- a/docs/design/compaction.md +++ b/docs/design/compaction.md @@ -59,8 +59,10 @@ closed and restore admission cannot register a competing writer until process teardown, after which restore converges from the replacement named by metadata. The Worker must likewise remain non-idle while terminal compaction-service cleanup is pending. Cleanup authority is stored on the Worker (not only on one compaction -future), retried until the registry record is gone, and cleared exactly once before -activation or a terminal result. +future). A failed stop returns a typed cleanup-pending outcome without clearing that +authority; the next compaction boundary retries it. The authority and lifecycle +reference are cleared exactly once after the registry record is gone, before +activation or a terminal idle-capable result. ## Metrics and comparison procedure