Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: separate worker state from runtime lifecycle

Browse Source
This commit is contained in:
Hare
2026-09-06 10:05:54 +09:00
parent 282a8d31b5
commit 2d1956b653
21 changed files with 602 additions and 188 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/controller.rs
+112 -26
View File
@@ -17,7 +17,7 @@ use crate::ipc::notify_buffer::NotifyBuffer;
use crate::ipc::server::SocketServer;
use crate::runtime::dir::RuntimeDir;
use crate::segment_log_sink::SegmentLogSink;
use crate::shared_state::WorkerSharedState;
use crate::shared_state::{WorkerCommandAdmission, WorkerSharedState};
use crate::shutdown_after_idle::{
ShutdownAfterIdleRequest, TicketIntakeReadyShutdownHook, is_ticket_intake_role,
take_shutdown_request_after_status,
@@ -181,21 +181,40 @@ impl WorkerHandle {
}
}
fn command_admission_disposition(
admission: WorkerCommandAdmission,
) -> Result<(), WorkerCommandDisposition> {
match admission {
WorkerCommandAdmission::Accepted => Ok(()),
WorkerCommandAdmission::Retry | WorkerCommandAdmission::StaleCommandId => {
Err(WorkerCommandDisposition::StaleCommandId)
}
WorkerCommandAdmission::Conflict => Err(WorkerCommandDisposition::Conflict),
WorkerCommandAdmission::ExecutionGenerationMismatch => {
Err(WorkerCommandDisposition::StaleExecutionGeneration)
}
WorkerCommandAdmission::StateRevisionMismatch => {
Err(WorkerCommandDisposition::StaleWorkerStateRevision)
}
}
}
fn validate_command(
envelope: WorkerCommandEnvelope,
kind: WorkerCommandKind,
shared_state: &WorkerSharedState,
) -> Result<(), WorkerCommandDisposition> {
command_admission_disposition(shared_state.admit_command(envelope, kind, true))
}
fn validate_shutdown_command(
envelope: WorkerCommandEnvelope,
shared_state: &WorkerSharedState,
) -> Result<(), WorkerCommandDisposition> {
let snapshot = shared_state.snapshot();
if envelope.expected_execution_generation != snapshot.execution_generation {
return Err(WorkerCommandDisposition::StaleExecutionGeneration);
match shared_state.admit_command(envelope, WorkerCommandKind::Shutdown, false) {
WorkerCommandAdmission::Accepted | WorkerCommandAdmission::Retry => Ok(()),
admission => command_admission_disposition(admission),
}
if envelope.expected_worker_state_revision != snapshot.revision {
return Err(WorkerCommandDisposition::StaleWorkerStateRevision);
}
if !shared_state.accept_command_id(envelope.command_id) {
return Err(WorkerCommandDisposition::StaleCommandId);
}
Ok(())
}
fn acknowledge_command(
@@ -205,6 +224,7 @@ fn acknowledge_command(
command: WorkerCommandKind,
disposition: WorkerCommandDisposition,
) {
shared_state.complete_command(command_id, command, disposition);
let _ = working_event_tx.send(Event::CommandAcknowledged {
acknowledgement: WorkerCommandAcknowledgement {
command_id,
@@ -1913,7 +1933,9 @@ async fn controller_loop<C, St>(
}
}
Method::Resume { command } => {
if let Err(disposition) = validate_command(command, &shared_state) {
if let Err(disposition) =
validate_command(command, WorkerCommandKind::Resume, &shared_state)
{
acknowledge_command(
&working_event_tx,
&shared_state,
@@ -1953,7 +1975,9 @@ async fn controller_loop<C, St>(
}
Method::Cancel { command } => {
if let Err(disposition) = validate_command(command, &shared_state) {
if let Err(disposition) =
validate_command(command, WorkerCommandKind::Cancel, &shared_state)
{
acknowledge_command(
&working_event_tx,
&shared_state,
@@ -2017,7 +2041,9 @@ async fn controller_loop<C, St>(
}
Method::Pause { command } => {
if let Err(disposition) = validate_command(command, &shared_state) {
if let Err(disposition) =
validate_command(command, WorkerCommandKind::Pause, &shared_state)
{
acknowledge_command(
&working_event_tx,
&shared_state,
@@ -2036,7 +2062,9 @@ async fn controller_loop<C, St>(
}
Method::Compact { command } => {
if let Err(disposition) = validate_command(command, &shared_state) {
if let Err(disposition) =
validate_command(command, WorkerCommandKind::Compact, &shared_state)
{
acknowledge_command(
&working_event_tx,
&shared_state,
@@ -2081,7 +2109,11 @@ async fn controller_loop<C, St>(
method = method_rx.recv() => {
match method {
Some(Method::Cancel { command }) => {
if let Err(disposition) = validate_command(command, &shared_state) {
if let Err(disposition) = validate_command(
command,
WorkerCommandKind::Cancel,
&shared_state,
) {
acknowledge_command(
&working_event_tx,
&shared_state,
@@ -2101,7 +2133,18 @@ async fn controller_loop<C, St>(
let _ = cancel_tx.send(true);
}
Some(Method::Shutdown { command }) => {
shared_state.accept_command_id(command.command_id);
if let Err(disposition) =
validate_shutdown_command(command, &shared_state)
{
acknowledge_command(
&working_event_tx,
&shared_state,
command.command_id,
WorkerCommandKind::Shutdown,
disposition,
);
continue;
}
shutdown_after_compaction = true;
acknowledge_command(
&working_event_tx,
@@ -2196,9 +2239,18 @@ async fn controller_loop<C, St>(
},
Method::Shutdown { command } => {
// Shutdown remains unconditional/retryable even when the caller's
// live-state fence is stale.
shared_state.accept_command_id(command.command_id);
// Shutdown ignores the state-revision fence but remains bound to the
// current execution generation and command payload identity.
if let Err(disposition) = validate_shutdown_command(command, &shared_state) {
acknowledge_command(
&working_event_tx,
&shared_state,
command.command_id,
WorkerCommandKind::Shutdown,
disposition,
);
continue;
}
acknowledge_command(
&working_event_tx,
&shared_state,
@@ -2544,7 +2596,9 @@ where
method = method_rx.recv(), if input_commit.is_none() => {
match method {
Some(Method::Cancel { command }) => {
if let Err(disposition) = validate_command(command, shared_state) {
if let Err(disposition) =
validate_command(command, WorkerCommandKind::Cancel, shared_state)
{
acknowledge_command(
working_event_tx,
shared_state,
@@ -2583,7 +2637,9 @@ where
let _ = cancel_tx.try_send(());
}
Some(Method::Pause { command }) => {
if let Err(disposition) = validate_command(command, shared_state) {
if let Err(disposition) =
validate_command(command, WorkerCommandKind::Pause, shared_state)
{
acknowledge_command(
working_event_tx,
shared_state,
@@ -2623,7 +2679,16 @@ where
let _ = pause_tx.try_send(());
}
Some(Method::Shutdown { command }) => {
shared_state.accept_command_id(command.command_id);
if let Err(disposition) = validate_shutdown_command(command, shared_state) {
acknowledge_command(
working_event_tx,
shared_state,
command.command_id,
WorkerCommandKind::Shutdown,
disposition,
);
continue;
}
shutdown_requested = true;
set_controller_state(
shared_state,
@@ -2705,7 +2770,9 @@ where
}
}
Some(Method::Resume { command }) => {
if let Err(disposition) = validate_command(command, shared_state) {
if let Err(disposition) =
validate_command(command, WorkerCommandKind::Resume, shared_state)
{
acknowledge_command(
working_event_tx,
shared_state,
@@ -2763,7 +2830,9 @@ where
}
}
Some(Method::Compact { command }) => {
if let Err(disposition) = validate_command(command, shared_state) {
if let Err(disposition) =
validate_command(command, WorkerCommandKind::Compact, shared_state)
{
acknowledge_command(
working_event_tx,
shared_state,
@@ -3677,6 +3746,7 @@ mod tests {
expected_execution_generation: 8,
expected_worker_state_revision: 0,
},
WorkerCommandKind::Pause,
&shared,
),
Err(WorkerCommandDisposition::StaleExecutionGeneration)
@@ -3688,6 +3758,7 @@ mod tests {
expected_execution_generation: 9,
expected_worker_state_revision: 1,
},
WorkerCommandKind::Pause,
&shared,
),
Err(WorkerCommandDisposition::StaleWorkerStateRevision)
@@ -3699,6 +3770,7 @@ mod tests {
expected_execution_generation: 9,
expected_worker_state_revision: 0,
},
WorkerCommandKind::Pause,
&shared,
)
.is_ok()
@@ -3708,12 +3780,25 @@ mod tests {
WorkerCommandEnvelope {
command_id: 1,
expected_execution_generation: 9,
expected_worker_state_revision: 1,
expected_worker_state_revision: 0,
},
WorkerCommandKind::Pause,
&shared,
),
Err(WorkerCommandDisposition::StaleCommandId)
);
assert_eq!(
validate_command(
WorkerCommandEnvelope {
command_id: 1,
expected_execution_generation: 9,
expected_worker_state_revision: 0,
},
WorkerCommandKind::Cancel,
&shared,
),
Err(WorkerCommandDisposition::Conflict)
);
assert!(
validate_command(
WorkerCommandEnvelope {
@@ -3721,6 +3806,7 @@ mod tests {
expected_execution_generation: 9,
expected_worker_state_revision: 1,
},
WorkerCommandKind::Pause,
&shared,
)
.is_ok()
crates/worker/src/shared_state.rs
+139 -10
View File
@@ -1,17 +1,37 @@
use std::collections::VecDeque;
use std::sync::{
OnceLock, RwLock,
atomic::{AtomicBool, Ordering},
};
use protocol::{
WorkerBusyState, WorkerMaintenanceState, WorkerRunState, WorkerState, WorkerStateSnapshot,
WorkerStatus,
WorkerBusyState, WorkerCommandDisposition, WorkerCommandEnvelope, WorkerCommandKind,
WorkerMaintenanceState, WorkerRunState, WorkerState, WorkerStateSnapshot, WorkerStatus,
};
use serde_json::json;
use session_store::SegmentId;
use crate::fs_view::WorkerFsView;
const COMPLETED_COMMAND_RETENTION: usize = 256;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct AcceptedWorkerCommand {
envelope: WorkerCommandEnvelope,
kind: WorkerCommandKind,
disposition: Option<WorkerCommandDisposition>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum WorkerCommandAdmission {
Accepted,
Retry,
Conflict,
StaleCommandId,
ExecutionGenerationMismatch,
StateRevisionMismatch,
}
/// Shared state between WorkerController and runtime directory.
///
/// `WorkerStateSnapshot` is the sole live execution-state authority. Runtime
@@ -23,6 +43,7 @@ pub struct WorkerSharedState {
pub manifest_toml: String,
pub greeting: protocol::Greeting,
state: RwLock<WorkerStateSnapshot>,
accepted_commands: RwLock<VecDeque<AcceptedWorkerCommand>>,
/// Worker-from-the-inside view of the filesystem. Set once in
/// `WorkerController::start` after the local WorkdirSession provider is
/// materialised, and read from the IPC server layer to answer
@@ -55,6 +76,7 @@ impl WorkerSharedState {
manifest_toml,
greeting,
state: RwLock::new(WorkerStateSnapshot::initial(execution_generation)),
accepted_commands: RwLock::new(VecDeque::new()),
fs_view: OnceLock::new(),
flow_transition_enabled: AtomicBool::new(false),
}
@@ -92,17 +114,99 @@ impl WorkerSharedState {
snapshot.clone()
}
pub fn accept_command_id(&self, command_id: u64) -> bool {
pub(crate) fn admit_command(
&self,
envelope: WorkerCommandEnvelope,
kind: WorkerCommandKind,
require_state_revision: bool,
) -> WorkerCommandAdmission {
let mut snapshot = self
.state
.write()
.expect("worker state lock poisoned; refusing command admission");
if command_id <= snapshot.last_command_id {
return false;
let mut accepted = self
.accepted_commands
.write()
.expect("worker command ledger lock poisoned; refusing command admission");
if let Some(existing) = accepted
.iter()
.find(|accepted| accepted.envelope.command_id == envelope.command_id)
{
return if existing.envelope == envelope && existing.kind == kind {
WorkerCommandAdmission::Retry
} else {
WorkerCommandAdmission::Conflict
};
}
snapshot.last_command_id = command_id;
if envelope.expected_execution_generation != snapshot.execution_generation {
return WorkerCommandAdmission::ExecutionGenerationMismatch;
}
if require_state_revision && envelope.expected_worker_state_revision != snapshot.revision {
return WorkerCommandAdmission::StateRevisionMismatch;
}
if envelope.command_id <= snapshot.last_command_id {
return WorkerCommandAdmission::StaleCommandId;
}
snapshot.last_command_id = envelope.command_id;
snapshot.revision = snapshot.revision.saturating_add(1);
true
accepted.push_back(AcceptedWorkerCommand {
envelope,
kind,
disposition: None,
});
WorkerCommandAdmission::Accepted
}
pub(crate) fn complete_command(
&self,
command_id: u64,
kind: WorkerCommandKind,
disposition: WorkerCommandDisposition,
) {
if !matches!(
disposition,
WorkerCommandDisposition::Accepted | WorkerCommandDisposition::InvalidState
) {
return;
}
let mut accepted = self
.accepted_commands
.write()
.expect("worker command ledger lock poisoned; refusing command completion");
if let Some(command) = accepted
.iter_mut()
.find(|command| command.envelope.command_id == command_id && command.kind == kind)
{
command.disposition.get_or_insert(disposition);
}
while accepted
.iter()
.filter(|command| command.disposition.is_some())
.count()
> COMPLETED_COMMAND_RETENTION
{
let Some(index) = accepted
.iter()
.position(|command| command.disposition.is_some())
else {
break;
};
accepted.remove(index);
}
}
#[cfg(test)]
pub(crate) fn command_result(
&self,
command_id: u64,
) -> Option<Option<WorkerCommandDisposition>> {
self.accepted_commands
.read()
.expect("worker command ledger lock poisoned")
.iter()
.find(|command| command.envelope.command_id == command_id)
.map(|command| command.disposition)
}
pub fn snapshot(&self) -> WorkerStateSnapshot {
@@ -190,9 +294,17 @@ mod tests {
}
#[test]
fn accepted_command_id_advances_the_snapshot_revision_atomically() {
fn accepted_command_identity_advances_revision_and_detects_reuse_conflicts() {
let state = test_state();
assert!(state.accept_command_id(9));
let envelope = WorkerCommandEnvelope {
command_id: 9,
expected_execution_generation: 7,
expected_worker_state_revision: 0,
};
assert_eq!(
state.admit_command(envelope, WorkerCommandKind::Pause, true),
WorkerCommandAdmission::Accepted
);
assert_eq!(
state.snapshot(),
WorkerStateSnapshot {
@@ -202,7 +314,24 @@ mod tests {
state: WorkerState::Idle,
}
);
assert!(!state.accept_command_id(9));
assert_eq!(state.command_result(9), Some(None));
state.complete_command(
9,
WorkerCommandKind::Pause,
WorkerCommandDisposition::Accepted,
);
assert_eq!(
state.command_result(9),
Some(Some(WorkerCommandDisposition::Accepted))
);
assert_eq!(
state.admit_command(envelope, WorkerCommandKind::Pause, true),
WorkerCommandAdmission::Retry
);
assert_eq!(
state.admit_command(envelope, WorkerCommandKind::Cancel, true),
WorkerCommandAdmission::Conflict
);
assert_eq!(state.snapshot().revision, 1);
}