From 103ef40f57da6d0752f281a1f52e92c0b4cb6ed1 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 18 Sep 2026 06:55:23 +0900 Subject: [PATCH] worker: fence controller during repair --- crates/worker/src/controller.rs | 138 +++++++++++++++++++++++++++++++- crates/worker/src/worker.rs | 18 ++++- docs/design/compaction.md | 5 +- 3 files changed, 157 insertions(+), 4 deletions(-) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 5848268bd..a137134b2 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1489,6 +1489,44 @@ where Ok(workdir_for_view) } +fn reject_method_while_attention_locked( + method: &Method, + working_event_tx: &broadcast::Sender, + message: &str, +) -> bool { + match method { + Method::Shutdown { .. } | Method::ListPendingSubmissions => false, + Method::Submit { + submission_request_id, + .. + } + | Method::SubmitTracked { + submission_request_id, + .. + } => { + let _ = working_event_tx.send(Event::SubmissionRejected { + submission_request_id: submission_request_id.clone(), + message: message.to_owned(), + }); + true + } + Method::Notify { .. } | Method::NotifyTracked { .. } => { + let _ = working_event_tx.send(Event::Error { + code: ErrorCode::InvalidRequest, + message: message.to_owned(), + }); + true + } + _ => { + let _ = working_event_tx.send(Event::Error { + code: ErrorCode::InvalidRequest, + message: message.to_owned(), + }); + true + } + } +} + /// Idle/Paused event loop. Each iteration either fires a staged /// `PendingRun` (delegating to [`drive_turn`] for the Running phase) or /// waits for the next `Method`. Method handlers stop at "update state + @@ -1551,6 +1589,12 @@ async fn controller_loop( let mut deferred_methods = VecDeque::new(); 'controller: loop { + if worker.has_failed_segment_activation() || worker.has_pending_compaction_cleanup() { + // A queued activation may have been accepted while the failing turn + // was still running. Preserve its durable queue entry, but never + // start it while repair/cleanup attention is required. + pending = None; + } // here so the status flip → drive_turn → finish sequence lives // in one place, regardless of which Method caused it. if let Some(run) = pending.take() { @@ -1700,7 +1744,15 @@ async fn controller_loop( tokio::select! { method = method_rx.recv() => match method { Some(method) => method, - None => break, + None => { + while worker.has_pending_compaction_cleanup() { + if let Err(error) = worker.retry_pending_compaction_cleanup().await { + tracing::warn!(error = %error, "shutdown waiting for compaction cleanup"); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + break 'controller; + } }, _ = notify_buffer.notified() => continue, _ = tokio::time::sleep(remaining) => { @@ -1726,12 +1778,63 @@ async fn controller_loop( tokio::select! { method = method_rx.recv() => match method { Some(method) => method, - None => break, + None => { + while worker.has_pending_compaction_cleanup() { + if let Err(error) = worker.retry_pending_compaction_cleanup().await { + tracing::warn!(error = %error, "shutdown waiting for compaction cleanup"); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + break 'controller; + } }, _ = notify_buffer.notified() => continue, } }; + if worker.has_pending_compaction_cleanup() { + match worker.retry_pending_compaction_cleanup().await { + Ok(()) => { + set_controller_state( + &shared_state, + &runtime_dir, + &working_event_tx, + WorkerState::Idle, + ) + .await; + } + Err(error) => { + let message = error.to_string(); + let _ = working_event_tx.send(Event::Error { + code: worker_error_code(&error), + message: message.clone(), + }); + if reject_method_while_attention_locked( + &method, + &working_event_tx, + &message, + ) { + continue; + } + // Shutdown is an explicit cleanup barrier: it is not allowed + // to complete while the service remains unconfirmed. + if matches!(&method, Method::Shutdown { .. }) { + continue; + } + } + } + } + + if worker.has_failed_segment_activation() + && reject_method_while_attention_locked( + &method, + &working_event_tx, + "Segment activation is incomplete; restore or shut down the Worker", + ) + { + continue; + } + match method { Method::Submit { submission_request_id, @@ -2165,6 +2268,15 @@ async fn controller_loop( }); } if shutdown_after_compaction { + while worker.has_pending_compaction_cleanup() { + if let Err(error) = worker.retry_pending_compaction_cleanup().await { + let _ = working_event_tx.send(Event::Error { + code: worker_error_code(&error), + message: error.to_string(), + }); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } let _ = working_event_tx.send(Event::Shutdown); break 'controller; } @@ -3333,6 +3445,28 @@ mod tests { } } + #[test] + fn attention_gate_rejects_submit_before_dispatch() { + let (event_tx, mut event_rx) = broadcast::channel(4); + let method = Method::Submit { + submission_request_id: "request-1".into(), + input: Vec::new(), + }; + + assert!(reject_method_while_attention_locked( + &method, + &event_tx, + "repair required", + )); + assert!(matches!( + event_rx.try_recv().unwrap(), + Event::SubmissionRejected { + submission_request_id, + message, + } if submission_request_id == "request-1" && message == "repair required" + )); + } + #[tokio::test] async fn segment_activation_failure_from_run_never_publishes_idle() { let mut env = make_env().await; diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index eec9f1077..f0b9dce9e 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -5178,6 +5178,21 @@ impl Worker { } } + pub(crate) fn has_pending_compaction_cleanup(&self) -> bool { + self.pending_compaction_cleanup + .lock() + .expect("pending compaction cleanup mutex poisoned") + .is_some() + } + + pub(crate) fn has_failed_segment_activation(&self) -> bool { + self.segment_state.activation_failed.load(Ordering::Acquire) + } + + pub(crate) async fn retry_pending_compaction_cleanup(&self) -> Result<(), WorkerError> { + self.release_pending_compaction_service().await + } + async fn release_pending_compaction_service(&self) -> Result<(), WorkerError> { let cleanup = self .pending_compaction_cleanup @@ -5480,7 +5495,8 @@ impl Worker { })); if let Err(error) = handle.send(summary_input.text).await { - let _ = registry.remove_service(&handle.session_id_string()); + // Keep the registry record and Worker-owned cleanup authority. The + // outer compaction boundary performs the single stop attempt. return Err(WorkerError::InvalidState(error.to_string())); } match handle.wait_until_idle().await { diff --git a/docs/design/compaction.md b/docs/design/compaction.md index 3a46434a8..4a9e26894 100644 --- a/docs/design/compaction.md +++ b/docs/design/compaction.md @@ -62,7 +62,10 @@ is pending. Cleanup authority is stored on the Worker (not only on one compactio 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. +activation or a terminal idle-capable result. While either activation repair or +cleanup attention is outstanding, the controller fences Submit/Notify dispatch and +does not start queued work. Shutdown is itself a cleanup barrier: it is not emitted +until the retained service record has been stopped successfully. ## Metrics and comparison procedure