52ad46ecb0
CONTRACT-01 landed its crate, so the local stand-in module goes away and the domain scalars, method payloads, their validation, the canonical digests and the trace format all come from the contract. `src/types.rs` is a facade over that crate plus the few things a coordinator needs that are not part of the type contract: a session-side DomainError, the synthetic composition's schema and event-id derivations, and the coordinator-local ControllerIntent, PortBinding and AgentOutcome that never cross the bus. Consequences worth naming: - Payloads are read and written through DomainType::from_json / to_json instead of serde derives, so a misspelled required field fails where the contract says it should. serde, sha2 and ryu-js leave this crate's dependencies with them. - The step-v1 section 8 trace is the contract's TransitionTrace, with behaviour and operational metadata already separated; the dispatch-order comparison now runs over the contract's own behaviour encoding. - Two things the migration found. IN_PROGRESS is raised strictly before any mutation, so its certainty is "none", not "unknown"; the local module had it wrong. WorldObservation::validate_against checks the views a result carries but does not require every declared view to be present, so requiring them is made explicit in the coordinator's phase C check, where step-v1 puts it. - The lost-Advance-result injection now waits for the worker to report the operation before abandoning the call, so the case it injects really is a loss after dispatch rather than a cancellation before it. Gates: cargo test -p fly-session (61 tests, both transports), cargo clippy --all-targets clean, and the runnable example produces the same behaviour trace over both transports.
446 lines
17 KiB
Rust
446 lines
17 KiB
Rust
//! Domain deduplication and the result cache of `ipc-v1` section 5.
|
|
//!
|
|
//! The operation key for a step mutation is `(sessionId, epoch, step, method, workerId)`, and
|
|
//! there is at most one Prepare, Commit or Advance for it. The cache decides, *before* any
|
|
//! phase check or artifact dereference, whether an arriving request is the original, a safe
|
|
//! replay, a conflicting change, a duplicate of something still running, or a retry whose
|
|
//! result is gone.
|
|
//!
|
|
//! A cached reply owns its artifacts through explicit holds, so a replay is still valid after
|
|
//! the first caller consumed its delivery. Dropping the record drops those holds.
|
|
|
|
use std::collections::{BTreeMap, VecDeque};
|
|
|
|
// `crate::types` is this crate's facade over the shared `fly-session-types` crate; the
|
|
// glob keeps the contract's own names in sight instead of restating them.
|
|
use crate::types::*;
|
|
|
|
/// `ipc-v1` section 5: unacknowledged lifecycle replies are bounded at 16, then BUSY.
|
|
pub const MAX_UNACKNOWLEDGED: usize = 16;
|
|
/// `ipc-v1` section 5: Status and Acknowledge keep a cache of their last 16 replies.
|
|
pub const MAX_READONLY_REPLIES: usize = 16;
|
|
/// `ipc-v1` section 5: keep the current and the immediately previous step's records.
|
|
pub const RETAINED_STEPS: u64 = 2;
|
|
|
|
/// Which retention rule an operation follows.
|
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
|
pub enum OpClass {
|
|
/// Prepare, Commit, Advance: keyed by scope and retained for two steps.
|
|
StepMutation,
|
|
/// Initialize and capture: retained until `Worker.Acknowledge`.
|
|
Lifecycle,
|
|
/// Status and Acknowledge: a small last-16 reply cache, no mutation key.
|
|
ReadOnly,
|
|
}
|
|
|
|
/// A step mutation's identity. Not a bus call id and not a batch id.
|
|
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
|
|
pub struct OperationKey {
|
|
pub session_id: Id,
|
|
pub epoch: Id,
|
|
pub step: u64,
|
|
pub method: String,
|
|
pub worker_id: Id,
|
|
}
|
|
|
|
/// A terminal domain reply plus the artifact holds that keep its attachments readable.
|
|
#[derive(Clone)]
|
|
pub struct CachedReply {
|
|
pub outcome: SessionRpcOutcome,
|
|
pub artifacts: Vec<(String, flybus::Artifact)>,
|
|
}
|
|
|
|
impl CachedReply {
|
|
pub fn new(outcome: SessionRpcOutcome) -> CachedReply {
|
|
CachedReply { outcome, artifacts: Vec::new() }
|
|
}
|
|
|
|
pub fn with_artifacts(
|
|
outcome: SessionRpcOutcome,
|
|
artifacts: Vec<(String, flybus::Artifact)>,
|
|
) -> CachedReply {
|
|
CachedReply { outcome, artifacts }
|
|
}
|
|
|
|
/// The attachment list for a fresh `rpc.reply`, over the same immutable bytes.
|
|
pub fn attachments(&self) -> Vec<(&str, &flybus::Artifact)> {
|
|
self.artifacts.iter().map(|(n, a)| (n.as_str(), a)).collect()
|
|
}
|
|
}
|
|
|
|
impl std::fmt::Debug for CachedReply {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("CachedReply")
|
|
.field("outcome", &self.outcome)
|
|
.field("artifacts", &self.artifacts.len())
|
|
.finish()
|
|
}
|
|
}
|
|
|
|
/// What the cache decided about an arriving request.
|
|
#[derive(Debug)]
|
|
pub enum Admission {
|
|
/// The original: run it, then `record` its reply.
|
|
Execute,
|
|
/// A safe replay: return this reply again, with fresh delivery ownership.
|
|
Replay(CachedReply),
|
|
/// A terminal domain error for this bus call, with no second mutation started.
|
|
Refuse(DomainError),
|
|
}
|
|
|
|
#[derive(Clone, Debug)]
|
|
struct Record {
|
|
request_id: DomainRequestId,
|
|
body: Digest,
|
|
reply: CachedReply,
|
|
}
|
|
|
|
/// One worker's domain request cache.
|
|
pub struct ResultCache {
|
|
steps: BTreeMap<OperationKey, Record>,
|
|
active: BTreeMap<OperationKey, (DomainRequestId, Digest)>,
|
|
lifecycle: BTreeMap<String, Record>,
|
|
lifecycle_order: VecDeque<String>,
|
|
readonly: VecDeque<(String, CachedReply)>,
|
|
highest_serial: Option<u64>,
|
|
step_watermark: Option<u64>,
|
|
/// Serials retired by `Worker.Acknowledge`; reuse below this is refused without keeping a
|
|
/// tombstone per request.
|
|
acknowledged_watermark: Option<u64>,
|
|
}
|
|
|
|
impl Default for ResultCache {
|
|
fn default() -> ResultCache {
|
|
ResultCache::new()
|
|
}
|
|
}
|
|
|
|
impl ResultCache {
|
|
pub fn new() -> ResultCache {
|
|
ResultCache {
|
|
steps: BTreeMap::new(),
|
|
active: BTreeMap::new(),
|
|
lifecycle: BTreeMap::new(),
|
|
lifecycle_order: VecDeque::new(),
|
|
readonly: VecDeque::new(),
|
|
highest_serial: None,
|
|
step_watermark: None,
|
|
acknowledged_watermark: None,
|
|
}
|
|
}
|
|
|
|
/// Decides what to do with an arriving request. Identity is checked before phase.
|
|
pub fn admit(
|
|
&mut self,
|
|
class: OpClass,
|
|
key: &OperationKey,
|
|
request: DomainRequestId,
|
|
body: &Digest,
|
|
) -> Admission {
|
|
match class {
|
|
OpClass::StepMutation => self.admit_step(key, request, body),
|
|
OpClass::Lifecycle => self.admit_lifecycle(request, body),
|
|
OpClass::ReadOnly => Admission::Execute,
|
|
}
|
|
}
|
|
|
|
fn admit_step(&mut self, key: &OperationKey, request: DomainRequestId, body: &Digest) -> Admission {
|
|
if let Some(record) = self.steps.get(key) {
|
|
if record.request_id == request && record.body == *body {
|
|
return Admission::Replay(record.reply.clone());
|
|
}
|
|
return Admission::Refuse(DomainError::new(
|
|
ErrorCode::Conflict,
|
|
"this operation key already holds a different request or body",
|
|
// The earlier result stands; refusing the duplicate undoes nothing.
|
|
MutationCertainty::None,
|
|
));
|
|
}
|
|
if !self.active.is_empty() && !self.active.contains_key(key) {
|
|
// One mutation executes at a time. A second, different one is refused before
|
|
// admission rather than queued behind the first.
|
|
return Admission::Refuse(DomainError::before(
|
|
ErrorCode::Busy,
|
|
"another step mutation is already executing on this worker",
|
|
));
|
|
}
|
|
if let Some((active_request, active_body)) = self.active.get(key) {
|
|
if *active_request == request && *active_body == *body {
|
|
// The duplicate bus call started no work at all, so its certainty is none;
|
|
// it must not be mistaken for the original operation's failure.
|
|
return Admission::Refuse(DomainError::before(
|
|
ErrorCode::InProgress,
|
|
"the original operation is still executing; this duplicate started no work",
|
|
));
|
|
}
|
|
return Admission::Refuse(DomainError::new(
|
|
ErrorCode::Conflict,
|
|
"an operation with a different body is active for this key",
|
|
MutationCertainty::None,
|
|
));
|
|
}
|
|
// No record and nothing active. Eviction must never re-enable execution, so a step
|
|
// below the retention window is refused on the serial and step watermarks.
|
|
if let Some(watermark) = self.step_watermark
|
|
&& key.step + RETAINED_STEPS <= watermark
|
|
{
|
|
let issued = self.highest_serial.is_some_and(|h| request.serial() <= h);
|
|
return Admission::Refuse(if issued {
|
|
DomainError::new(
|
|
ErrorCode::ResultExpired,
|
|
"the retained result for this request is gone; it is never recomputed",
|
|
MutationCertainty::Unknown,
|
|
)
|
|
} else {
|
|
DomainError::before(
|
|
ErrorCode::StaleStep,
|
|
"a newly issued request cannot name a step the worker has left behind",
|
|
)
|
|
});
|
|
}
|
|
if let Some(watermark) = self.acknowledged_watermark
|
|
&& request.serial() <= watermark
|
|
{
|
|
return Admission::Refuse(DomainError::new(
|
|
ErrorCode::ResultExpired,
|
|
"this request serial was acknowledged and cannot be reused",
|
|
MutationCertainty::Unknown,
|
|
));
|
|
}
|
|
self.begin(key.clone(), request, body.clone());
|
|
Admission::Execute
|
|
}
|
|
|
|
fn admit_lifecycle(&mut self, request: DomainRequestId, body: &Digest) -> Admission {
|
|
let id = request.as_str().to_owned();
|
|
if let Some(record) = self.lifecycle.get(&id) {
|
|
if record.body == *body {
|
|
return Admission::Replay(record.reply.clone());
|
|
}
|
|
return Admission::Refuse(DomainError::before(
|
|
ErrorCode::Conflict,
|
|
"this lifecycle request id already holds a different body",
|
|
));
|
|
}
|
|
if let Some(watermark) = self.acknowledged_watermark
|
|
&& request.serial() <= watermark
|
|
{
|
|
return Admission::Refuse(DomainError::new(
|
|
ErrorCode::ResultExpired,
|
|
"this request serial was acknowledged and cannot be reused",
|
|
MutationCertainty::Unknown,
|
|
));
|
|
}
|
|
if self.lifecycle.len() >= MAX_UNACKNOWLEDGED {
|
|
return Admission::Refuse(DomainError::before(
|
|
ErrorCode::Busy,
|
|
"16 lifecycle replies are unacknowledged; acknowledge some before sending more",
|
|
));
|
|
}
|
|
Admission::Execute
|
|
}
|
|
|
|
fn begin(&mut self, key: OperationKey, request: DomainRequestId, body: Digest) {
|
|
self.highest_serial = Some(self.highest_serial.map_or(request.serial(), |h| h.max(request.serial())));
|
|
self.step_watermark = Some(self.step_watermark.map_or(key.step, |w| w.max(key.step)));
|
|
self.active.insert(key, (request, body));
|
|
}
|
|
|
|
/// Stores a terminal reply for a step mutation and prunes what has aged out.
|
|
pub fn record(
|
|
&mut self,
|
|
key: OperationKey,
|
|
request: DomainRequestId,
|
|
body: Digest,
|
|
reply: CachedReply,
|
|
) {
|
|
self.active.remove(&key);
|
|
let step = key.step;
|
|
self.steps.insert(key, Record { request_id: request, body, reply });
|
|
self.step_watermark = Some(self.step_watermark.map_or(step, |w| w.max(step)));
|
|
self.prune();
|
|
}
|
|
|
|
/// Stores a lifecycle reply, retained until `Worker.Acknowledge`.
|
|
pub fn record_lifecycle(&mut self, request: DomainRequestId, body: Digest, reply: CachedReply) {
|
|
let id = request.as_str().to_owned();
|
|
self.highest_serial = Some(self.highest_serial.map_or(request.serial(), |h| h.max(request.serial())));
|
|
if self.lifecycle.insert(id.clone(), Record { request_id: request, body, reply }).is_none()
|
|
{
|
|
self.lifecycle_order.push_back(id);
|
|
}
|
|
}
|
|
|
|
/// Stores a read-only reply in the last-16 cache.
|
|
pub fn record_readonly(&mut self, request: DomainRequestId, reply: CachedReply) {
|
|
let id = request.as_str().to_owned();
|
|
self.readonly.retain(|(existing, _)| *existing != id);
|
|
self.readonly.push_back((id, reply));
|
|
while self.readonly.len() > MAX_READONLY_REPLIES {
|
|
self.readonly.pop_front();
|
|
}
|
|
}
|
|
|
|
/// Releases an operation that did not mutate anything, so the key stays free.
|
|
pub fn abandon(&mut self, key: &OperationKey) {
|
|
self.active.remove(key);
|
|
}
|
|
|
|
/// `Worker.Acknowledge`: drops those lifecycle records, ignoring unknown ids, and raises
|
|
/// the serial watermark so an acknowledged id cannot be reused.
|
|
pub fn acknowledge(&mut self, ids: &[DomainRequestId]) -> Vec<DomainRequestId> {
|
|
let mut out = Vec::new();
|
|
for id in ids {
|
|
if let Some(record) = self.lifecycle.remove(id.as_str()) {
|
|
self.lifecycle_order.retain(|existing| existing != id.as_str());
|
|
self.acknowledged_watermark = Some(
|
|
self.acknowledged_watermark
|
|
.map_or(record.request_id.serial(), |w| w.max(record.request_id.serial())),
|
|
);
|
|
out.push(id.clone());
|
|
}
|
|
}
|
|
out
|
|
}
|
|
|
|
pub fn unacknowledged(&self) -> usize {
|
|
self.lifecycle.len()
|
|
}
|
|
|
|
pub fn retained_steps(&self) -> usize {
|
|
self.steps.len()
|
|
}
|
|
|
|
pub fn has_active(&self) -> bool {
|
|
!self.active.is_empty()
|
|
}
|
|
|
|
/// Deliberately drops a step record, so a retry meets RESULT_EXPIRED instead of a replay.
|
|
pub fn expire_step(&mut self, key: &OperationKey) -> bool {
|
|
self.steps.remove(key).is_some()
|
|
}
|
|
|
|
fn prune(&mut self) {
|
|
let Some(watermark) = self.step_watermark else { return };
|
|
let floor = watermark.saturating_sub(RETAINED_STEPS - 1);
|
|
self.steps.retain(|key, _| key.step >= floor);
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use serde_json::Value;
|
|
|
|
use super::*;
|
|
|
|
fn key(step: u64, method: &str) -> OperationKey {
|
|
OperationKey {
|
|
session_id: id("demo"),
|
|
epoch: id("e1"),
|
|
step,
|
|
method: method.to_owned(),
|
|
worker_id: id("fly-a"),
|
|
}
|
|
}
|
|
|
|
fn reply(tag: &str) -> CachedReply {
|
|
let mut result = serde_json::Map::new();
|
|
result.insert("tag".into(), tag.into());
|
|
CachedReply::new(SessionRpcOutcome::Success(SessionRpcSuccess {
|
|
request_id: DomainRequestId::from_serial(1),
|
|
worker_id: id("fly-a"),
|
|
incarnation_id: id("inc-1"),
|
|
scope: Some(scope_at("demo", "e1", 0)),
|
|
result: Value::Object(result),
|
|
}))
|
|
}
|
|
|
|
fn body(s: &str) -> Digest {
|
|
digest_of_bytes(s.as_bytes())
|
|
}
|
|
|
|
#[test]
|
|
fn the_same_key_request_and_body_replays() {
|
|
let mut c = ResultCache::new();
|
|
let k = key(0, "Agent.Prepare");
|
|
assert!(matches!(c.admit(OpClass::StepMutation, &k, DomainRequestId::from_serial(1), &body("a")), Admission::Execute));
|
|
c.record(k.clone(), DomainRequestId::from_serial(1), body("a"), reply("first"));
|
|
match c.admit(OpClass::StepMutation, &k, DomainRequestId::from_serial(1), &body("a")) {
|
|
Admission::Replay(r) => {
|
|
assert_eq!(r.outcome.result().unwrap()["tag"], "first");
|
|
}
|
|
other => panic!("wanted a replay, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_changed_body_for_a_recorded_key_is_a_conflict() {
|
|
let mut c = ResultCache::new();
|
|
let k = key(0, "Environment.Advance");
|
|
c.admit(OpClass::StepMutation, &k, DomainRequestId::from_serial(1), &body("a"));
|
|
c.record(k.clone(), DomainRequestId::from_serial(1), body("a"), reply("first"));
|
|
match c.admit(OpClass::StepMutation, &k, DomainRequestId::from_serial(1), &body("b")) {
|
|
Admission::Refuse(e) => assert_eq!(e.code, ErrorCode::Conflict),
|
|
other => panic!("wanted CONFLICT, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_duplicate_while_the_original_runs_is_in_progress() {
|
|
let mut c = ResultCache::new();
|
|
let k = key(0, "Agent.Prepare");
|
|
c.admit(OpClass::StepMutation, &k, DomainRequestId::from_serial(1), &body("a"));
|
|
match c.admit(OpClass::StepMutation, &k, DomainRequestId::from_serial(1), &body("a")) {
|
|
Admission::Refuse(e) => assert_eq!(e.code, ErrorCode::InProgress),
|
|
other => panic!("wanted IN_PROGRESS, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn an_evicted_step_gives_result_expired_and_a_fresh_serial_gives_stale_step() {
|
|
let mut c = ResultCache::new();
|
|
for step in 0..4u64 {
|
|
let k = key(step, "Agent.Prepare");
|
|
c.admit(OpClass::StepMutation, &k, DomainRequestId::from_serial(step + 1), &body("a"));
|
|
c.record(k, DomainRequestId::from_serial(step + 1), body("a"), reply("x"));
|
|
}
|
|
assert_eq!(c.retained_steps(), 2);
|
|
// req-1 named step 0, whose record is long gone.
|
|
match c.admit(OpClass::StepMutation, &key(0, "Agent.Prepare"), DomainRequestId::from_serial(1), &body("a")) {
|
|
Admission::Refuse(e) => assert_eq!(e.code, ErrorCode::ResultExpired),
|
|
other => panic!("wanted RESULT_EXPIRED, got {other:?}"),
|
|
}
|
|
// A serial above the highest issued is a new operation naming an old step.
|
|
match c.admit(OpClass::StepMutation, &key(0, "Agent.Prepare"), DomainRequestId::from_serial(99), &body("a")) {
|
|
Admission::Refuse(e) => assert_eq!(e.code, ErrorCode::StaleStep),
|
|
other => panic!("wanted STALE_STEP, got {other:?}"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn lifecycle_replies_are_bounded_and_released_by_acknowledge() {
|
|
let mut c = ResultCache::new();
|
|
for serial in 1..=MAX_UNACKNOWLEDGED as u64 {
|
|
assert!(matches!(
|
|
c.admit(OpClass::Lifecycle, &key(0, "Agent.Initialize"), DomainRequestId::from_serial(serial), &body("a")),
|
|
Admission::Execute
|
|
));
|
|
c.record_lifecycle(DomainRequestId::from_serial(serial), body("a"), reply("init"));
|
|
}
|
|
match c.admit(OpClass::Lifecycle, &key(0, "Agent.Initialize"), DomainRequestId::from_serial(99), &body("a")) {
|
|
Admission::Refuse(e) => assert_eq!(e.code, ErrorCode::Busy),
|
|
other => panic!("wanted BUSY, got {other:?}"),
|
|
}
|
|
let dropped = c.acknowledge(&[
|
|
DomainRequestId::from_serial(1),
|
|
DomainRequestId::from_serial(404),
|
|
]);
|
|
assert_eq!(dropped, vec![DomainRequestId::from_serial(1)]);
|
|
assert_eq!(c.unacknowledged(), MAX_UNACKNOWLEDGED - 1);
|
|
// The acknowledged serial cannot come back.
|
|
match c.admit(OpClass::Lifecycle, &key(0, "Agent.Initialize"), DomainRequestId::from_serial(1), &body("a")) {
|
|
Admission::Refuse(e) => assert_eq!(e.code, ErrorCode::ResultExpired),
|
|
other => panic!("wanted RESULT_EXPIRED, got {other:?}"),
|
|
}
|
|
}
|
|
}
|