use crate::envelope::Actor;
use crate::fsm::{AttemptState, CommandState, MessageState, StateMachine, TaskState};
use crate::ids::{IdempotencyKey, ReceiptId};
pub const LIVENESS_PRODUCER_KIND: &str = "liveness_producer";
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Cursor<S> {
pub state: S,
pub version: u32,
pub applied_idempotency_key: Option<IdempotencyKey>,
pub applied_by: Option<Actor>,
}
#[derive(Debug, Clone, Copy)]
pub struct TransitionRequest<'a, S> {
pub to: S,
pub expected_version: u32,
pub actor: &'a Actor,
pub idempotency_key: Option<&'a IdempotencyKey>,
pub receipt_id: Option<&'a ReceiptId>,
}
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, specta::Type)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum TransitionResult<S> {
Applied { state: S, version: u32 },
IllegalEdge { from: S, to: S },
StaleVersion { actual: u32, expected: u32 },
UnauthorizedActor { reason: String },
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum GuardViolation {
WrongActor { required_kind: &'static str },
MissingReceipt,
}
impl std::fmt::Display for GuardViolation {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::WrongActor { required_kind } => {
write!(f, "edge requires actor kind {required_kind}")
}
Self::MissingReceipt => f.write_str("edge requires a receipt"),
}
}
}
#[derive(Debug, Clone, Copy)]
pub struct GuardCtx<'a> {
pub actor: &'a Actor,
pub receipt_id: Option<&'a ReceiptId>,
}
pub trait TransitionGuard: StateMachine {
fn guard(_from: Self, _to: Self, _ctx: &GuardCtx<'_>) -> Result<(), GuardViolation> {
Ok(())
}
}
impl TransitionGuard for TaskState {}
impl TransitionGuard for MessageState {}
impl TransitionGuard for CommandState {}
impl TransitionGuard for AttemptState {
fn guard(from: Self, to: Self, ctx: &GuardCtx<'_>) -> Result<(), GuardViolation> {
let is_flip = matches!(
(from, to),
(Self::Running, Self::Blocked) | (Self::Blocked, Self::Running)
);
if is_flip {
if ctx.actor.kind != LIVENESS_PRODUCER_KIND {
return Err(GuardViolation::WrongActor {
required_kind: LIVENESS_PRODUCER_KIND,
});
}
if ctx.receipt_id.is_none_or(|r| r.as_str().is_empty()) {
return Err(GuardViolation::MissingReceipt);
}
}
Ok(())
}
}
pub fn apply<S: TransitionGuard>(
cursor: &Cursor<S>,
req: &TransitionRequest<'_, S>,
) -> TransitionResult<S> {
let ctx = GuardCtx {
actor: req.actor,
receipt_id: req.receipt_id,
};
if let Err(violation) = S::guard(cursor.state, req.to, &ctx) {
return TransitionResult::UnauthorizedActor {
reason: violation.to_string(),
};
}
if let (Some(key), Some(applied)) =
(req.idempotency_key, cursor.applied_idempotency_key.as_ref())
&& key == applied
&& cursor.state == req.to
&& cursor.applied_by.as_ref() == Some(req.actor)
{
return TransitionResult::Applied {
state: cursor.state,
version: cursor.version,
};
}
if !S::can_transition(cursor.state, req.to) {
return TransitionResult::IllegalEdge {
from: cursor.state,
to: req.to,
};
}
if cursor.version != req.expected_version {
return TransitionResult::StaleVersion {
actual: cursor.version,
expected: req.expected_version,
};
}
let Some(next_version) = req.expected_version.checked_add(1) else {
return TransitionResult::StaleVersion {
actual: cursor.version,
expected: req.expected_version,
};
};
TransitionResult::Applied {
state: req.to,
version: next_version,
}
}
#[cfg(test)]
mod tests {
use super::*;
fn kernel() -> Actor {
Actor {
kind: "kernel".into(),
id: None,
}
}
fn liveness() -> Actor {
Actor {
kind: LIVENESS_PRODUCER_KIND.into(),
id: Some("lp-1".into()),
}
}
fn cursor(state: AttemptState, version: u32) -> Cursor<AttemptState> {
Cursor {
state,
version,
applied_idempotency_key: None,
applied_by: None,
}
}
#[test]
fn applied_bumps_version_by_exactly_one() {
let actor = kernel();
let result = apply(
&cursor(AttemptState::Queued, 3),
&TransitionRequest {
to: AttemptState::Leased,
expected_version: 3,
actor: &actor,
idempotency_key: None,
receipt_id: None,
},
);
assert_eq!(
result,
TransitionResult::Applied {
state: AttemptState::Leased,
version: 4
}
);
}
#[test]
fn illegal_edge_is_refused_before_version() {
let actor = kernel();
let result = apply(
&cursor(AttemptState::Queued, 3),
&TransitionRequest {
to: AttemptState::Succeeded,
expected_version: 99,
actor: &actor,
idempotency_key: None,
receipt_id: None,
},
);
assert_eq!(
result,
TransitionResult::IllegalEdge {
from: AttemptState::Queued,
to: AttemptState::Succeeded
}
);
}
#[test]
fn stale_version_reports_actual() {
let actor = kernel();
let result = apply(
&cursor(AttemptState::Running, 7),
&TransitionRequest {
to: AttemptState::Succeeded,
expected_version: 6,
actor: &actor,
idempotency_key: None,
receipt_id: None,
},
);
assert_eq!(
result,
TransitionResult::StaleVersion {
actual: 7,
expected: 6
}
);
}
#[test]
fn version_ceiling_is_refused_not_overflowed() {
let actor = kernel();
let result = apply(
&Cursor {
state: TaskState::Submitted,
version: u32::MAX,
applied_idempotency_key: None,
applied_by: None,
},
&TransitionRequest {
to: TaskState::Working,
expected_version: u32::MAX,
actor: &actor,
idempotency_key: None,
receipt_id: None,
},
);
assert_eq!(
result,
TransitionResult::StaleVersion {
actual: u32::MAX,
expected: u32::MAX,
}
);
}
#[test]
fn blocked_flip_demands_the_liveness_producer() {
let wrong = kernel();
let result = apply(
&cursor(AttemptState::Running, 2),
&TransitionRequest {
to: AttemptState::Blocked,
expected_version: 2,
actor: &wrong,
idempotency_key: None,
receipt_id: None,
},
);
assert!(matches!(result, TransitionResult::UnauthorizedActor { .. }));
}
#[test]
fn blocked_flip_demands_a_receipt() {
let actor = liveness();
let no_receipt = apply(
&cursor(AttemptState::Blocked, 5),
&TransitionRequest {
to: AttemptState::Running,
expected_version: 5,
actor: &actor,
idempotency_key: None,
receipt_id: None,
},
);
assert!(matches!(
no_receipt,
TransitionResult::UnauthorizedActor { .. }
));
let receipt = ReceiptId::new("r-1");
let with_receipt = apply(
&cursor(AttemptState::Blocked, 5),
&TransitionRequest {
to: AttemptState::Running,
expected_version: 5,
actor: &actor,
idempotency_key: None,
receipt_id: Some(&receipt),
},
);
assert_eq!(
with_receipt,
TransitionResult::Applied {
state: AttemptState::Running,
version: 6
}
);
}
#[test]
fn blocked_flip_rejects_a_present_but_empty_receipt() {
let actor = liveness();
let empty = ReceiptId::new("");
let result = apply(
&cursor(AttemptState::Blocked, 5),
&TransitionRequest {
to: AttemptState::Running,
expected_version: 5,
actor: &actor,
idempotency_key: None,
receipt_id: Some(&empty),
},
);
assert!(matches!(result, TransitionResult::UnauthorizedActor { .. }));
}
#[test]
fn blocked_terminal_paths_need_no_liveness_actor() {
let actor = kernel();
for to in [
AttemptState::Canceling,
AttemptState::Failed,
AttemptState::Unknown,
] {
let result = apply(
&cursor(AttemptState::Blocked, 1),
&TransitionRequest {
to,
expected_version: 1,
actor: &actor,
idempotency_key: None,
receipt_id: None,
},
);
assert_eq!(
result,
TransitionResult::Applied {
state: to,
version: 2
},
"blocked -> {to:?} must not demand the liveness producer"
);
}
}
#[test]
fn same_key_retry_is_stable_even_after_version_advanced() {
let actor = kernel();
let key = IdempotencyKey::new("once");
let already_applied = Cursor {
state: TaskState::Working,
version: 2,
applied_idempotency_key: Some(key.clone()),
applied_by: Some(kernel()),
};
let result = apply(
&already_applied,
&TransitionRequest {
to: TaskState::Working,
expected_version: 1,
actor: &actor,
idempotency_key: Some(&key),
receipt_id: None,
},
);
assert_eq!(
result,
TransitionResult::Applied {
state: TaskState::Working,
version: 2
}
);
let other = IdempotencyKey::new("twice");
let result = apply(
&already_applied,
&TransitionRequest {
to: TaskState::Completed,
expected_version: 1,
actor: &actor,
idempotency_key: Some(&other),
receipt_id: None,
},
);
assert_eq!(
result,
TransitionResult::StaleVersion {
actual: 2,
expected: 1
}
);
}
#[test]
fn guarded_flip_refuses_the_wrong_actor_before_disclosing_the_version() {
let engine = Actor {
kind: "engine".into(),
id: Some("e-1".into()),
};
let key = IdempotencyKey::new("flip-once");
let receipt = ReceiptId::new("r-forged");
let result = apply(
&Cursor {
state: AttemptState::Running,
version: 5,
applied_idempotency_key: Some(key.clone()),
applied_by: None,
},
&TransitionRequest {
to: AttemptState::Blocked,
expected_version: 999,
actor: &engine,
idempotency_key: Some(&key),
receipt_id: Some(&receipt),
},
);
assert!(matches!(result, TransitionResult::UnauthorizedActor { .. }));
}
#[test]
fn keyed_replay_requires_the_recording_actor_not_just_the_key() {
let lp = liveness();
let key = IdempotencyKey::new("flip-once");
let cur = Cursor {
state: AttemptState::Blocked,
version: 6,
applied_idempotency_key: Some(key.clone()),
applied_by: Some(lp.clone()),
};
let stable = apply(
&cur,
&TransitionRequest {
to: AttemptState::Blocked,
expected_version: 6,
actor: &lp,
idempotency_key: Some(&key),
receipt_id: None,
},
);
assert_eq!(
stable,
TransitionResult::Applied {
state: AttemptState::Blocked,
version: 6
}
);
let engine = Actor {
kind: "engine".into(),
id: Some("e-1".into()),
};
let probe = apply(
&cur,
&TransitionRequest {
to: AttemptState::Blocked,
expected_version: 6,
actor: &engine,
idempotency_key: Some(&key),
receipt_id: None,
},
);
assert_eq!(
probe,
TransitionResult::IllegalEdge {
from: AttemptState::Blocked,
to: AttemptState::Blocked
}
);
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Gated {
Shut,
}
impl StateMachine for Gated {
const STATES: &'static [Self] = &[Self::Shut];
const EDGES: &'static [(Self, Self)] = &[];
const ESCAPE: &'static [Self] = &[];
}
impl TransitionGuard for Gated {
fn guard(_from: Self, _to: Self, ctx: &GuardCtx<'_>) -> Result<(), GuardViolation> {
if ctx.actor.kind != "gatekeeper" {
return Err(GuardViolation::WrongActor {
required_kind: "gatekeeper",
});
}
Ok(())
}
}
#[test]
fn guard_refusal_precedes_idempotency_edge_and_version() {
let intruder = kernel();
let key = IdempotencyKey::new("harvested");
let result = apply(
&Cursor {
state: Gated::Shut,
version: 4,
applied_idempotency_key: Some(key.clone()),
applied_by: None,
},
&TransitionRequest {
to: Gated::Shut,
expected_version: 999,
actor: &intruder,
idempotency_key: Some(&key),
receipt_id: None,
},
);
assert!(matches!(result, TransitionResult::UnauthorizedActor { .. }));
}
#[test]
fn replayed_key_with_a_different_target_state_falls_through_to_refusal() {
let engine = Actor {
kind: "engine".into(),
id: Some("e-1".into()),
};
let key = IdempotencyKey::new("flip-once");
let cur = Cursor {
state: AttemptState::Running,
version: 5,
applied_idempotency_key: Some(key.clone()),
applied_by: None,
};
let result = apply(
&cur,
&TransitionRequest {
to: AttemptState::Succeeded,
expected_version: 999,
actor: &engine,
idempotency_key: Some(&key),
receipt_id: None,
},
);
assert_eq!(
result,
TransitionResult::StaleVersion {
actual: 5,
expected: 999
}
);
let result = apply(
&cur,
&TransitionRequest {
to: AttemptState::Leased,
expected_version: 999,
actor: &engine,
idempotency_key: Some(&key),
receipt_id: None,
},
);
assert_eq!(
result,
TransitionResult::IllegalEdge {
from: AttemptState::Running,
to: AttemptState::Leased
}
);
}
#[test]
fn transition_result_wire_shape_is_tagged_snake_case() {
let applied: TransitionResult<TaskState> = TransitionResult::Applied {
state: TaskState::Working,
version: 2,
};
assert_eq!(
serde_json::to_value(&applied).expect("serialize"),
serde_json::json!({ "kind": "applied", "state": "working", "version": 2 })
);
let refused: TransitionResult<TaskState> = TransitionResult::UnauthorizedActor {
reason: "edge requires a receipt".into(),
};
assert_eq!(
serde_json::to_value(&refused).expect("serialize"),
serde_json::json!({ "kind": "unauthorized_actor", "reason": "edge requires a receipt" })
);
}
}