worker: keep cleanup failures non-idle

crates/worker/src/controller.rs

@@ -2148,6 +2148,7 @@ async fn controller_loop<C, St>(
                         | 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::<WorkerRunResult, _>(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::<WorkerRunResult, _>(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;

crates/worker/src/worker.rs

@@ -4780,7 +4780,11 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
                     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<C: LlmClient + 'static, St: Store> Worker<C, St> {
         mut cancel: Option<tokio::sync::watch::Receiver<bool>>,
         trigger: CompactionTrigger,
     ) -> Result<SegmentId, WorkerError> {
-        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<C: LlmClient + 'static, St: Store> Worker<C, St> {
                 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<C: LlmClient + 'static, St: Store> Worker<C, St> {
                     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;
+                if !matches!(&error, WorkerError::CompactionCleanupPending { .. }) {
-                self.set_compaction_progress(None);
+                    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 {
-            match registry.stop_service(&cleanup.session_id).await {
+            Ok(_) => {
-                Ok(_) => {
+                let mut pending = self
-                    let mut pending = self
+                    .pending_compaction_cleanup
-                        .pending_compaction_cleanup
+                    .lock()
-                        .lock()
+                    .expect("pending compaction cleanup mutex poisoned");
-                        .expect("pending compaction cleanup mutex poisoned");
+                if pending
-                    if pending
+                    .as_ref()
-                        .as_ref()
+                    .is_some_and(|pending| pending.session_id == cleanup.session_id)
-                        .is_some_and(|pending| pending.session_id == cleanup.session_id)
+                {
-                    {
+                    *pending = None;
-                        *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;
                 }
+                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) {
+    async fn release_compaction_service(
-        self.release_pending_compaction_service().await;
+        &self,
-        if self
+        lifecycle: &mut CompactionLifecycle,
-            .pending_compaction_cleanup
+    ) -> Result<(), WorkerError> {
-            .lock()
+        self.release_pending_compaction_service().await?;
-            .expect("pending compaction cleanup mutex poisoned")
+        lifecycle.internal_worker = None;
-            .is_none()
+        Ok(())
-        {
-            lifecycle.internal_worker = None;
-        }
     }
 
     async fn compact_impl(
@@ -5761,7 +5775,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
         // 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),
 

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
+future). A failed stop returns a typed cleanup-pending outcome without clearing that
-activation or a terminal result.
+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