use crate::codec::Codec;
use crate::ctx::{Ctx, Handlers};
use crate::env::{Env, Notify};
use crate::error::Error;
use crate::limits::{Acquire, LimitBackend, LocalLimits, Priority};
use crate::object::{ObjectDef, ObjectDetail, Objects};
use fusee_core::record::{
Lsn, Name, Origin, PAUSED, ParentRef, Payload, Record, RecordBuf, RecordRef, RunId, Seq,
StartMeta, StartUpdate, Status, StepTime, decode_canonical_all, decode_canonical_into,
encode_canonical, origin, policy,
};
use fusee_core::run::{
Arm, AwaitStart, Completion, Next, Outcome, RunInfo, RunMachine, StepStart, StreamOpen, empty,
};
use fusee_core::segment::{LogWriter, Progress, ScanError, Scanned, SegmentScanner, Tail};
use fusee_core::shard::{BlobRef, CANCEL_SIGNAL, Event, Found, Mode, Output, ShardCore, TimerKind};
use fusee_core::store::{
Checkpoint, CheckpointRun, EntryRef, MAX_CHAIN, StoreEntry, chain_journal, encode_entry,
encode_entry_into, read_entry,
};
use fusee_core::wire::hash_id;
use rustc_hash::FxHashMap;
use std::cell::{Cell, RefCell};
use std::cmp::Reverse;
use std::collections::{BTreeMap, BinaryHeap, VecDeque};
use std::future::Future;
use std::io;
use std::pin::Pin;
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use std::task::{Context, Poll, Wake, Waker};
mod objects;
pub type RunOutcome = Result<Payload, (Status, Payload)>;
pub type LocalFuture = Pin<Box<dyn Future<Output = RunOutcome>>>;
pub type WorkflowFn = Arc<dyn Fn(Ctx) -> LocalFuture + Send + Sync>;
#[derive(Default)]
pub struct Registry {
types: FxHashMap<String, BTreeMap<u32, WorkflowFn>>,
objects: FxHashMap<String, Arc<ObjectDef>>,
}
impl Registry {
pub fn register_object(&mut self, name: &str, def: ObjectDef) {
self.objects.insert(name.to_owned(), Arc::new(def));
REGISTRY_EPOCH.fetch_add(1, Ordering::Release);
}
pub fn object(&self, name: &str) -> Option<Arc<ObjectDef>> {
self.objects.get(name).cloned()
}
pub fn register(&mut self, name: &str, version: u32, f: WorkflowFn) {
self.types
.entry(name.to_owned())
.or_default()
.insert(version, f);
REGISTRY_EPOCH.fetch_add(1, Ordering::Release);
}
pub fn get(&self, name: &str, version: u32) -> Option<WorkflowFn> {
self.types.get(name)?.get(&version).cloned()
}
pub fn latest(&self, name: &str) -> Option<u32> {
self.types.get(name)?.keys().next_back().copied()
}
pub fn names(&self) -> Vec<String> {
self.types.keys().cloned().collect()
}
pub fn versions(&self, name: &str) -> Vec<u32> {
self.types
.get(name)
.map(|v| v.keys().copied().collect())
.unwrap_or_default()
}
pub fn is_object(&self, name: &str) -> bool {
self.objects.contains_key(name)
}
}
pub type SharedRegistry = Arc<RwLock<Registry>>;
static REGISTRY_EPOCH: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
#[derive(Default)]
struct CodeCache {
epoch: u64,
fns: FxHashMap<String, Vec<(u32, WorkflowFn)>>,
}
impl CodeCache {
fn get(&mut self, registry: &SharedRegistry, wtype: &str, version: u32) -> Option<&WorkflowFn> {
let epoch = REGISTRY_EPOCH.load(Ordering::Acquire);
if self.epoch != epoch {
self.fns.clear();
self.epoch = epoch;
}
let known = self
.fns
.get(wtype)
.is_some_and(|v| v.iter().any(|(ver, _)| *ver == version));
if !known {
let f = registry.read().ok()?.get(wtype, version)?;
self.fns
.entry(wtype.to_owned())
.or_default()
.push((version, f));
}
self.cached(wtype, version)
}
fn cached(&self, wtype: &str, version: u32) -> Option<&WorkflowFn> {
self.fns
.get(wtype)?
.iter()
.find(|(ver, _)| *ver == version)
.map(|(_, f)| f)
}
}
#[derive(Clone, Debug)]
pub struct RuntimeConfig {
pub shard: u32,
pub shards: u32,
pub retention_ms: u64,
pub park_after_ms: u64,
pub memory_budget: u64,
pub checkpoint_interval_ms: u64,
pub max_loaded: usize,
pub segment_bytes: u64,
pub inline_flush: bool,
pub inline_max_us: u64,
pub keep_history: bool,
pub on_exhausted: crate::ctx::Exhausted,
pub archive_history: bool,
pub step_times: bool,
#[cfg(feature = "opentelemetry")]
pub telemetry: Option<Arc<crate::telemetry::Telemetry>>,
pub replay_only: bool,
pub checkpoint_on_close: bool,
pub index_chunk_records: usize,
pub index_rebase_deltas: usize,
#[doc(hidden)]
pub sabotage: Sabotage,
}
#[derive(Clone, Debug)]
pub struct Services {
pub codec: Option<Arc<dyn Codec>>,
pub limits: Arc<dyn LimitBackend>,
}
impl Default for Services {
fn default() -> Self {
Self {
codec: None,
limits: Arc::new(LocalLimits::new()),
}
}
}
#[doc(hidden)]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum Sabotage {
#[default]
None,
SkipGate,
EarlyRelease,
DuplicateUpdates,
IgnoreLimits,
NoAttach,
EarlyAnswers,
EarlyChunks,
UnlockedObjects,
}
impl Default for RuntimeConfig {
fn default() -> Self {
Self {
shard: 0,
shards: 1,
retention_ms: 24 * 3_600_000,
park_after_ms: 1_000,
memory_budget: 0,
checkpoint_interval_ms: 30_000,
max_loaded: 1_000_000,
segment_bytes: fusee_core::segment::DEFAULT_SEGMENT_BYTES,
inline_flush: true,
inline_max_us: 200,
keep_history: false,
on_exhausted: crate::ctx::Exhausted::Fail,
archive_history: false,
step_times: false,
#[cfg(feature = "opentelemetry")]
telemetry: None,
replay_only: false,
checkpoint_on_close: true,
index_chunk_records: INDEX_CHUNK_RECORDS,
index_rebase_deltas: 64,
sabotage: Sabotage::None,
}
}
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize)]
pub struct Metrics {
pub records: u64,
pub log_bytes: u64,
pub flushes: u64,
pub inline_flushes: u64,
pub flushed_records: u64,
pub flush_us_total: u64,
pub steps_run: u64,
pub steps_replayed: u64,
pub runs_started: u64,
pub runs_finished: u64,
pub runs_parked: u64,
pub parked_entries: u64,
pub run_table_bytes: u64,
pub loaded_bytes: u64,
pub open_store_frames: u64,
pub open_store_bytes: u64,
pub open_us: u64,
pub budget_parks: u64,
pub object_compactions: u64,
pub journal_frames_read: u64,
pub journal_bytes_read: u64,
pub runs_loaded: u64,
pub checkpoints: u64,
pub index_bases_built: u64,
pub determinism_errors: u64,
pub busy: u64,
pub gate_waits: u64,
pub gate_wait_us_total: u64,
pub stalls: u64,
pub max_tick_us: u64,
pub max_flush_us: u64,
pub volatile_timers: u64,
pub log_writes: u64,
pub log_device_bytes: u64,
pub log_rolls: u64,
pub log_owner: u64,
}
fn done_summary(journal: &[u8]) -> Result<Vec<u8>, Error> {
let records = decode_canonical_all(journal).map_err(|e| Error::Io(e.to_string()))?;
let mut out = Vec::with_capacity(journal.len().min(4096));
for r in &records {
match r {
Record::Start { .. } => {
let mut start = r.clone();
if let Record::Start { input, update, .. } = &mut start {
*input = fusee_core::run::empty();
*update = None;
}
encode_canonical(&start, &mut out);
}
Record::Attributes { .. }
| Record::Chunk { .. }
| Record::StreamEnd { .. }
| Record::Done { .. } => encode_canonical(r, &mut out),
_ => {}
}
}
Ok(out)
}
const INLINE_DELIVERIES: u32 = 64;
static STALL_US: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(u64::MAX);
static STALL_TRACE: Mutex<Option<std::fs::File>> = Mutex::new(None);
pub fn set_stall_threshold_us(us: u64) {
STALL_US.store(us, Ordering::Relaxed);
}
fn stall_threshold_us() -> u64 {
let v = STALL_US.load(Ordering::Relaxed);
if v != u64::MAX {
return v;
}
let v = std::env::var("FUSEE_STALL_US")
.ok()
.and_then(|s| s.parse().ok())
.unwrap_or(0);
if v != 0
&& let Ok(path) = std::env::var("FUSEE_STALL_TRACE")
&& let Ok(file) = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(path)
&& let Ok(mut t) = STALL_TRACE.lock()
{
*t = Some(file);
}
STALL_US.store(v, Ordering::Relaxed);
v
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum StallKind {
Tick,
InlineFlush,
Flush,
}
#[derive(Clone, Debug, serde::Serialize)]
pub struct Stall {
pub shard: u32,
pub kind: StallKind,
pub us: u64,
pub at_ms: u64,
pub records: u32,
pub bytes: u64,
pub preparing_segment: bool,
pub checkpointing: bool,
pub parked: u32,
pub maintenance_us: u64,
pub loaded: usize,
pub slots: usize,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default, serde::Deserialize, serde::Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum OnRunning {
#[default]
Fail,
UseExisting,
TerminateExisting,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default, serde::Deserialize, serde::Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum OnFinished {
#[default]
Allow,
AllowIfFailed,
Refuse,
UseExisting,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default)]
pub enum ParentClose {
#[default]
Terminate,
RequestCancel,
Abandon,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default, serde::Deserialize, serde::Serialize)]
#[serde(default)]
pub struct IdPolicy {
pub running: OnRunning,
pub finished: OnFinished,
}
impl IdPolicy {
pub const ATTACH: Self = Self {
running: OnRunning::UseExisting,
finished: OnFinished::UseExisting,
};
pub fn bits(self) -> u8 {
let r = match self.running {
OnRunning::Fail => policy::RUNNING_FAIL,
OnRunning::UseExisting => policy::RUNNING_USE_EXISTING,
OnRunning::TerminateExisting => policy::RUNNING_TERMINATE,
};
let f = match self.finished {
OnFinished::Allow => policy::FINISHED_ALLOW,
OnFinished::AllowIfFailed => policy::FINISHED_IF_FAILED,
OnFinished::Refuse => policy::FINISHED_REFUSE,
OnFinished::UseExisting => policy::FINISHED_USE_EXISTING,
};
r | f
}
pub fn from_bits(b: u8) -> Self {
Self {
running: match b & policy::RUNNING_MASK {
policy::RUNNING_USE_EXISTING => OnRunning::UseExisting,
policy::RUNNING_TERMINATE => OnRunning::TerminateExisting,
_ => OnRunning::Fail,
},
finished: match b & policy::FINISHED_MASK {
policy::FINISHED_IF_FAILED => OnFinished::AllowIfFailed,
policy::FINISHED_REFUSE => OnFinished::Refuse,
policy::FINISHED_USE_EXISTING => OnFinished::UseExisting,
_ => OnFinished::Allow,
},
}
}
}
impl ParentClose {
pub fn bits(self) -> u8 {
match self {
Self::Terminate => policy::CLOSE_TERMINATE,
Self::RequestCancel => policy::CLOSE_CANCEL,
Self::Abandon => policy::CLOSE_ABANDON,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct UpdateRequest {
pub name: String,
pub id: String,
pub payload: Payload,
pub token: Option<u64>,
}
#[derive(Clone, Debug)]
pub struct StartRequest {
pub id: String,
pub wtype: String,
pub version: Option<u32>,
pub input: Payload,
pub codec: u8,
pub deadline_ms: u64,
pub parent: Option<ParentRef>,
pub policy: u8,
pub update: Option<UpdateRequest>,
pub sealed: bool,
pub token: Option<u64>,
pub trace: Option<Payload>,
}
impl StartRequest {
pub fn new(id: &str, wtype: &str, input: Payload, ids: IdPolicy) -> Self {
Self {
id: id.to_owned(),
wtype: wtype.to_owned(),
version: None,
input,
codec: 1,
deadline_ms: 0,
parent: None,
policy: ids.bits(),
update: None,
sealed: false,
token: None,
trace: None,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum StartOutcome {
Created { run: RunId },
Existing { run: RunId, done: bool },
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ResultState {
NotFound,
Running(RunId),
Done {
run: RunId,
status: Status,
payload: Payload,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RunSummary {
pub shard: u32,
pub run: RunId,
pub id: String,
pub wtype: String,
pub version: u32,
pub status: String,
pub started_ms: u64,
pub finished_ms: Option<u64>,
pub error: Option<String>,
pub attributes: String,
pub object: bool,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RunDetail {
pub summary: RunSummary,
pub journal: Vec<RecordBuf>,
pub object: Option<Box<ObjectDetail>>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ForeignEvent {
Start { run: RunId, info: RunInfo },
Wake { run: RunId },
Stuck { run: RunId, message: String },
Ended { run: RunId },
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Message {
ChildStart {
parent: ParentRef,
wtype: String,
id: String,
flags: u8,
deadline_ms: u64,
input: Payload,
trace: Option<Payload>,
},
ChildResult {
parent: ParentRef,
status: Status,
payload: Payload,
},
Signal {
target: String,
name: String,
payload: Payload,
dedup: u64,
},
Close {
target: String,
wtype: String,
parent: Option<ParentRef>,
terminate: bool,
reason: Payload,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Delivery {
Resolve {
seq: Seq,
outcome: Outcome,
},
CancelAll {
seqs: Vec<Seq>,
reason: Payload,
},
Update {
ordinal: u64,
name: String,
id: String,
payload: Payload,
token: Option<u64>,
},
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Outgoing {
pub id: u64,
pub target: u32,
pub msg: Message,
sent_ms: u64,
}
const MSG_TOKEN: u64 = 1 << 63;
const RESEND_MS: u64 = 1_000;
fn encode_message(msg: &Message, out: &mut Vec<u8>) {
use fusee_core::wire::{put_bytes, put_varint};
let parent = |out: &mut Vec<u8>, p: &ParentRef| {
put_varint(out, u64::from(p.shard));
put_varint(out, p.run);
put_varint(out, u64::from(p.seq));
};
match msg {
Message::ChildStart {
parent: p,
wtype,
id,
flags,
deadline_ms,
input,
trace,
} => {
out.push(if trace.is_some() { 5 } else { 1 });
parent(out, p);
put_bytes(out, wtype.as_bytes());
put_bytes(out, id.as_bytes());
out.push(*flags);
put_varint(out, *deadline_ms);
put_bytes(out, input);
if let Some(t) = trace {
put_bytes(out, t);
}
}
Message::ChildResult {
parent: p,
status,
payload,
} => {
out.push(2);
parent(out, p);
out.push(*status as u8);
put_bytes(out, payload);
}
Message::Signal {
target,
name,
payload,
dedup,
} => {
out.push(3);
put_bytes(out, target.as_bytes());
put_bytes(out, name.as_bytes());
put_bytes(out, payload);
put_varint(out, *dedup);
}
Message::Close {
target,
wtype,
parent: p,
terminate,
reason,
} => {
out.push(4);
put_bytes(out, target.as_bytes());
put_bytes(out, wtype.as_bytes());
match p {
None => out.push(0),
Some(p) => {
out.push(1);
parent(out, p);
}
}
out.push(u8::from(*terminate));
put_bytes(out, reason);
}
}
}
fn decode_message(r: &mut fusee_core::wire::Reader<'_>) -> fusee_core::wire::DecodeResult<Message> {
use fusee_core::wire::DecodeError;
let parent =
|r: &mut fusee_core::wire::Reader<'_>| -> fusee_core::wire::DecodeResult<ParentRef> {
Ok(ParentRef {
shard: r.varint_u32()?,
run: r.varint()?,
seq: r.varint_u32()?,
})
};
Ok(match r.u8()? {
k @ (1 | 5) => Message::ChildStart {
parent: parent(r)?,
wtype: r.str()?.to_owned(),
id: r.str()?.to_owned(),
flags: r.u8()?,
deadline_ms: r.varint()?,
input: Arc::from(r.bytes()?),
trace: if k == 5 {
Some(Arc::from(r.bytes()?))
} else {
None
},
},
2 => Message::ChildResult {
parent: parent(r)?,
status: Status::from_u8(r.u8()?)?,
payload: Arc::from(r.bytes()?),
},
3 => Message::Signal {
target: r.str()?.to_owned(),
name: r.str()?.to_owned(),
payload: Arc::from(r.bytes()?),
dedup: r.varint()?,
},
4 => {
let target = r.str()?.to_owned();
let wtype = r.str()?.to_owned();
let p = match r.u8()? {
0 => None,
1 => Some(parent(r)?),
_ => return Err(DecodeError("bad parent flag")),
};
Message::Close {
target,
wtype,
parent: p,
terminate: r.u8()? != 0,
reason: Arc::from(r.bytes()?),
}
}
_ => return Err(DecodeError("unknown message kind")),
})
}
pub fn encode_outbox(entries: &[Outgoing]) -> Vec<u8> {
use fusee_core::wire::put_varint;
let mut out = Vec::new();
put_varint(&mut out, entries.len() as u64);
for e in entries {
put_varint(&mut out, e.id);
put_varint(&mut out, u64::from(e.target));
encode_message(&e.msg, &mut out);
}
out
}
pub fn decode_outbox(bytes: &[u8]) -> fusee_core::wire::DecodeResult<Vec<Outgoing>> {
if bytes.is_empty() {
return Ok(Vec::new());
}
let mut r = fusee_core::wire::Reader::new(bytes);
let n = r.varint()?;
let mut out = Vec::new();
for _ in 0..n {
let id = r.varint()?;
let target = r.varint_u32()?;
let msg = decode_message(&mut r)?;
out.push(Outgoing {
id,
target,
msg,
sent_ms: 0,
});
}
Ok(out)
}
pub struct ReadyQueue {
queue: Mutex<Vec<RunId>>,
queued: AtomicBool,
wake: Notify,
}
impl ReadyQueue {
fn push(&self, run: RunId) {
if let Ok(mut q) = self.queue.lock() {
q.push(run);
}
self.queued.store(true, Ordering::Release);
(self.wake)();
}
}
struct LocalReady {
queue: Arc<ReadyQueue>,
runs: Vec<RunId>,
}
thread_local! {
static LOCAL_READY: RefCell<Option<LocalReady>> = const { RefCell::new(None) };
}
fn push_local(queue: &Arc<ReadyQueue>, run: RunId) -> bool {
LOCAL_READY
.try_with(|l| match l.try_borrow_mut().as_deref_mut() {
Ok(Some(l)) if Arc::ptr_eq(&l.queue, queue) => {
l.runs.push(run);
true
}
_ => false,
})
.unwrap_or(false)
}
fn take_local(queue: &Arc<ReadyQueue>, out: Option<&mut Vec<RunId>>) -> bool {
LOCAL_READY
.try_with(|l| match l.try_borrow_mut().as_deref_mut() {
Ok(Some(l)) if Arc::ptr_eq(&l.queue, queue) => {
let any = !l.runs.is_empty();
if let Some(out) = out {
out.append(&mut l.runs);
}
any
}
_ => false,
})
.unwrap_or(false)
}
struct LocalTick<'a> {
spare: &'a RefCell<Vec<RunId>>,
outer: Option<LocalReady>,
}
impl<'a> LocalTick<'a> {
fn enter(queue: Arc<ReadyQueue>, spare: &'a RefCell<Vec<RunId>>) -> Self {
let runs = std::mem::take(&mut *spare.borrow_mut());
let outer = LOCAL_READY
.try_with(|l| {
l.try_borrow_mut()
.ok()
.and_then(|mut l| l.replace(LocalReady { queue, runs }))
})
.ok()
.flatten();
Self { spare, outer }
}
}
impl Drop for LocalTick<'_> {
fn drop(&mut self) {
let outer = self.outer.take();
let mine = LOCAL_READY
.try_with(|l| {
l.try_borrow_mut()
.ok()
.and_then(|mut l| std::mem::replace(&mut *l, outer))
})
.ok()
.flatten();
if let Some(LocalReady { queue, mut runs }) = mine {
if !runs.is_empty() {
if let Ok(mut q) = queue.queue.lock() {
q.append(&mut runs);
}
queue.queued.store(true, Ordering::Release);
(queue.wake)();
}
*self.spare.borrow_mut() = runs;
}
}
}
struct RunWaker {
run: RunId,
queued: AtomicBool,
woken: AtomicBool,
ready: Arc<ReadyQueue>,
}
impl RunWaker {
fn wake_local(&self) {
self.woken.store(true, Ordering::Relaxed);
if !self.queued.load(Ordering::Relaxed) {
self.queued.store(true, Ordering::Relaxed);
if !push_local(&self.ready, self.run) {
self.ready.push(self.run);
}
}
}
}
impl Wake for RunWaker {
fn wake(self: Arc<Self>) {
self.wake_by_ref();
}
fn wake_by_ref(self: &Arc<Self>) {
self.woken.store(true, Ordering::Release);
if !self.queued.swap(true, Ordering::AcqRel) && !push_local(&self.ready, self.run) {
self.ready.push(self.run);
}
}
}
#[derive(Default)]
struct Gates {
fifo: VecDeque<(Lsn, RunId, u64)>,
heap: BinaryHeap<Reverse<(Lsn, RunId, u64)>>,
}
impl Gates {
fn push(&mut self, g: (Lsn, RunId, u64)) {
if self.fifo.back().is_none_or(|b| b.0 <= g.0) {
self.fifo.push_back(g);
} else {
self.heap.push(Reverse(g));
}
}
fn peek(&self) -> Option<(Lsn, RunId, u64)> {
match (self.fifo.front(), self.heap.peek()) {
(Some(&f), Some(&Reverse(h))) => Some(if h.0 < f.0 { h } else { f }),
(Some(&f), None) => Some(f),
(None, Some(&Reverse(h))) => Some(h),
(None, None) => None,
}
}
fn pop(&mut self) {
match (self.fifo.front(), self.heap.peek()) {
(Some(f), Some(Reverse(h))) if h.0 < f.0 => {
self.heap.pop();
}
(Some(_), _) => {
self.fifo.pop_front();
}
(None, _) => {
self.heap.pop();
}
}
}
}
enum JournalIn<'a> {
Borrowed(&'a [RecordBuf]),
Decoded(&'a mut Vec<RecordBuf>),
}
fn run_waker(handle: &Waker) -> &RunWaker {
unsafe { &*handle.data().cast::<RunWaker>() }
}
struct Code {
fut: LocalFuture,
handle: Waker,
}
impl Code {
fn new(fut: LocalFuture, waker: &Arc<RunWaker>) -> Self {
Self {
fut,
handle: Waker::from(waker.clone()),
}
}
fn waker(&self) -> &RunWaker {
run_waker(&self.handle)
}
fn polled_with(&self, waker: &Arc<RunWaker>) -> bool {
std::ptr::eq(self.handle.data().cast::<RunWaker>(), Arc::as_ptr(waker))
}
}
struct Slot {
machine: RunMachine,
waker: Arc<RunWaker>,
foreign: bool,
suspended_since: Option<u64>,
dirty: bool,
broadcasts: u64,
wake_queued: bool,
gate_at: Option<Lsn>,
undelivered: bool,
resolved: Option<(Seq, Outcome)>,
}
struct Removed {
foreign: bool,
}
const SLOT_SPARE: usize = 2048;
const REAP_EARLY: usize = 256;
const SLOT_SPARE_IDLE: usize = 64;
enum Resolution {
Ready(Outcome),
Waiting(Waker),
}
#[derive(Default)]
struct Slots {
entries: Vec<Option<Box<Slot>>>,
index: FxHashMap<RunId, u32>,
free: std::collections::BTreeSet<u32>,
last: Cell<Option<(RunId, u32)>>,
}
impl Slots {
#[inline]
fn idx(&self, run: RunId) -> Option<u32> {
if let Some((r, i)) = self.last.get()
&& r == run
{
return Some(i);
}
self.idx_slow(run)
}
#[cold]
#[inline(never)]
fn idx_slow(&self, run: RunId) -> Option<u32> {
let i = *self.index.get(&run)?;
self.last.set(Some((run, i)));
Some(i)
}
#[inline]
fn get(&self, run: &RunId) -> Option<&Slot> {
let i = self.idx(*run)?;
self.entries.get(i as usize)?.as_deref()
}
#[inline]
fn get_mut(&mut self, run: &RunId) -> Option<&mut Slot> {
let i = self.idx(*run)?;
self.entries.get_mut(i as usize)?.as_deref_mut()
}
#[inline]
fn contains_key(&self, run: &RunId) -> bool {
self.idx(*run).is_some()
}
fn insert(&mut self, run: RunId, slot: Box<Slot>) {
if let Some(i) = self.index.get(&run) {
self.entries[*i as usize] = Some(slot);
return;
}
let i = match self.free.pop_first() {
Some(i) => i,
None => {
self.entries.push(None);
(self.entries.len() - 1) as u32
}
};
self.entries[i as usize] = Some(slot);
self.index.insert(run, i);
self.last.set(Some((run, i)));
}
fn remove(&mut self, run: &RunId) -> Option<(u32, Box<Slot>)> {
let i = self.index.remove(run)?;
if self.last.get().is_some_and(|(r, _)| r == *run) {
self.last.set(None);
}
self.entries[i as usize].take().map(|s| (i, s))
}
fn release(&mut self, i: u32) {
if self.entries.get(i as usize).is_none_or(Option::is_some) {
return;
}
self.free.insert(i);
while self.entries.last().is_some_and(Option::is_none)
&& self
.free
.last()
.is_some_and(|&top| top as usize + 1 == self.entries.len())
{
self.free.pop_last();
self.entries.pop();
}
}
fn len(&self) -> usize {
self.index.len()
}
fn is_empty(&self) -> bool {
self.index.is_empty()
}
fn shrink(&mut self) {
let n = self.index.len();
if self.index.capacity() > 1024 && self.index.capacity() > 4 * n {
self.index.shrink_to(n * 2);
}
let n = self.entries.len();
if self.entries.capacity() > 1024 && self.entries.capacity() > 4 * n {
self.entries.shrink_to(n * 2);
}
}
}
#[derive(Clone, Debug)]
enum Stashed {
Complete(Seq, Completion),
Wake,
}
pub struct Inner<E: Env> {
pub env: E,
pub core: ShardCore,
writer: ShardLog,
cfg: RuntimeConfig,
slots: Slots,
resolved: FxHashMap<(RunId, Seq), Resolution>,
gates: Gates,
round_us: u64,
flush_started_us: u64,
timers: BinaryHeap<Reverse<(u64, u64)>>,
timer_wakers: FxHashMap<u64, Waker>,
timer_seq: u64,
durable: Lsn,
commit: Lsn,
load_room: usize,
commit_dirty: bool,
inline_deliveries: u32,
load_buf: Vec<RunId>,
load_at: Vec<(BlobRef, RunId)>,
in_flight: Option<Lsn>,
submitted: VecDeque<(Lsn, u64)>,
ready: Arc<ReadyQueue>,
registry: SharedRegistry,
foreign_host: bool,
codec: Option<Arc<dyn Codec>>,
pub limits: Arc<dyn LimitBackend>,
update_waiters: FxHashMap<(RunId, Name), Vec<u64>>,
obj: Objects,
offered: FxHashMap<u64, (RunId, Name, Name, Payload)>,
next_update_id: u64,
completion_acks: FxHashMap<(RunId, Seq), Vec<u64>>,
heartbeats: BTreeMap<(RunId, Seq), RecordBuf>,
step_began: FxHashMap<(RunId, Seq), u64>,
released: Vec<Output>,
outbox: BTreeMap<u64, Outgoing>,
next_out: u64,
msg_tokens: FxHashMap<u64, (u32, u64)>,
next_msg_token: u64,
child_acks: FxHashMap<(RunId, Seq), Vec<(u32, u64)>>,
message_acks: Vec<(u32, u64)>,
foreign: VecDeque<ForeignEvent>,
stash: FxHashMap<RunId, Vec<Stashed>>,
awaiting: FxHashMap<String, Vec<RunId>>,
instantiate: Vec<RunId>,
cancel_tokens: FxHashMap<RunId, Vec<u64>>,
by: Option<Payload>,
subscribers: FxHashMap<RunId, Vec<ParentRef>>,
stuck: FxHashMap<RunId, String>,
failed: Option<String>,
metrics: Metrics,
ckpt_pending: Option<(Checkpoint, Option<u64>)>,
ckpt_running: bool,
last_ckpt_ms: u64,
last_ckpt_lsn: Lsn,
last_ckpt_store_len: u64,
obsolete_store_bytes: u64,
last_ckpt_segment: u64,
last_sweep_ms: u64,
current_step: FxHashMap<RunId, Seq>,
pub replicator: Option<Box<dyn Replicator>>,
stall_us: u64,
stalls: VecDeque<Stall>,
dirty: Vec<RunId>,
park_queue: VecDeque<(u64, RunId)>,
removed: Vec<(RunId, u32)>,
undelivered: Vec<RunId>,
tick_parked: u32,
tick_maintenance_us: u64,
flush_records: u32,
flush_bytes: u64,
clock: TickClock,
gate_flushes: u32,
drive_backlog: usize,
host_queue: Option<Arc<std::sync::atomic::AtomicUsize>>,
host_never: bool,
held_loads: VecDeque<RunId>,
reaping: Vec<RunId>,
store_v3: bool,
no_compaction: bool,
index: Option<Vec<BlobRef>>,
index_gen: u64,
index_base: (usize, u64),
index_delta_bytes: u64,
index_build: Option<IndexBuild>,
slow_flushes: u32,
tail: StoreTail,
enc: Vec<u8>,
timer_scratch: Vec<(Seq, TimerKind, u64)>,
decoded: Vec<RecordBuf>,
names: fusee_core::record::InlineDecoder,
#[allow(clippy::vec_box)]
slot_spare: Vec<Box<Slot>>,
streams: FxHashMap<RunId, RunStreams>,
stream_bytes: usize,
stream_waiters: FxHashMap<RunId, Vec<StreamWaiter>>,
stream_due: BinaryHeap<Reverse<(Lsn, RunId)>>,
stream_wait_until: u64,
trim_peak: u64,
spare_mark: u64,
pools_shrunk: bool,
}
#[derive(Default)]
struct TickClock {
on: bool,
ms: Cell<Option<u64>>,
ran: Cell<bool>,
us: Cell<Option<u64>>,
drives: u32,
}
const CLOCK_DRIVES: u32 = 64;
impl TickClock {
fn renew(&self) {
self.ms.set(None);
self.ran.set(false);
self.us.set(None);
}
}
struct Stored {
records: Vec<RecordBuf>,
last_lsn: Lsn,
chain: u64,
depth: u8,
}
struct Chain {
run: RunId,
journal: Vec<u8>,
last_lsn: Lsn,
id_hash: u64,
timers: Vec<(Seq, TimerKind, u64)>,
chain: u64,
depth: u8,
}
#[derive(Default)]
struct StoreTail {
at: u64,
buf: Vec<u8>,
}
const STORE_TAIL_MAX: usize = 256 << 10;
impl StoreTail {
fn put(&mut self, store_len: u64, entry: EntryRef<'_>) -> Result<BlobRef, Error> {
if self.buf.is_empty() {
self.at = store_len;
}
let from = self.buf.len();
encode_entry_into(entry, &mut self.buf);
let len = u32::try_from(self.buf.len() - from).map_err(|_| {
self.buf.truncate(from);
Error::Io("run store frame exceeds 4 GiB".into())
})?;
Ok(BlobRef {
offset: self.at + from as u64,
len,
})
}
fn full(&self) -> bool {
self.buf.len() >= STORE_TAIL_MAX
}
fn get(&self, offset: u64, len: u32) -> Option<&[u8]> {
let from = usize::try_from(offset.checked_sub(self.at)?).ok()?;
if self.buf.is_empty() {
return None;
}
self.buf.get(from..from.checked_add(len as usize)?)
}
}
fn read_store<E: Env>(env: &mut E, tail: &StoreTail, offset: u64, len: u32) -> io::Result<Vec<u8>> {
match tail.get(offset, len) {
Some(bytes) => Ok(bytes.to_vec()),
None => env.store_read(offset, len),
}
}
struct StreamLog {
name: Name,
first: u64,
chunks: VecDeque<(Lsn, Payload)>,
bytes: usize,
len: u64,
end: Option<(Lsn, Status)>,
}
#[derive(Default)]
struct RunStreams {
logs: Vec<StreamLog>,
done: Option<(Lsn, Status)>,
id: Option<Name>,
read_at: u64,
}
struct StreamWaiter {
name: Name,
from: u64,
token: u64,
deadline_ms: u64,
}
struct StreamRead {
run: RunId,
from: u64,
chunks: Vec<Payload>,
end: Option<crate::stream::StreamEnd>,
wait: Option<Lsn>,
id: Option<Name>,
}
struct StreamScan {
chunks: Vec<Payload>,
more: bool,
len: u64,
end: Option<Status>,
done: Option<Status>,
last_lsn: Lsn,
id: Option<Name>,
}
const STREAM_LOG_BYTES: usize = 64 << 10;
const STREAM_LOG_CHUNKS: usize = 1024;
const STREAM_CACHE_BYTES: usize = 8 << 20;
const STREAM_READ_KEEP_MS: u64 = 10_000;
fn end_of(status: Status, closed: bool) -> crate::stream::StreamEnd {
crate::stream::StreamEnd {
status: status.as_str().to_owned(),
closed,
}
}
const INDEX_FRAME_RECORDS: usize = 1 << 15;
const SLOW_DISK_FLUSHES: u32 = 16;
pub const INDEX_CHUNK_RECORDS: usize = 1 << 13;
const INDEX_BUILD_SLICE_US: u64 = 300;
const INDEX_BUILD_SLICE_FRAMES: usize = 4;
struct IndexBuild {
cursor: Option<fusee_core::table::Slot>,
frames: Vec<BlobRef>,
bytes: u64,
generation: u64,
deltas_from: usize,
}
pub const LOADED_RUN_BYTES: u64 = 5632;
const BUDGET_PARKS: usize = 16 * 1024;
enum ShardLog {
Segments(LogWriter),
Sections(fusee_core::wal::SectionWriter),
}
impl ShardLog {
fn append<S: AsRef<str>, B: AsRef<[u8]>>(
&mut self,
rec: &Record<S, B>,
now_ms: u64,
) -> Result<Lsn, fusee_core::segment::TooLarge> {
match self {
Self::Segments(w) => w.append(rec, now_ms),
Self::Sections(w) => w.append(rec),
}
}
fn take(&mut self) -> Option<fusee_core::segment::Batch> {
match self {
Self::Segments(w) => w.take(),
Self::Sections(w) => w.take_up_to(fusee_core::wal::MAX_SECTION),
}
}
fn has_pending(&self) -> bool {
match self {
Self::Segments(w) => w.has_pending(),
Self::Sections(w) => w.has_pending(),
}
}
fn last_lsn(&self) -> Lsn {
match self {
Self::Segments(w) => w.last_lsn(),
Self::Sections(w) => w.last_lsn(),
}
}
fn last_encoded(&self) -> Option<&[u8]> {
match self {
Self::Segments(_) => None,
Self::Sections(w) => Some(w.last_encoded()),
}
}
}
pub trait Replicator {
fn appended(&mut self, lsn: Lsn, canonical: &[u8]);
fn durable(&mut self, lsn: Lsn);
fn commit(&self) -> Lsn;
}
impl<E: Env> Inner<E> {
fn trace_of(&self, run: RunId) -> Option<Payload> {
self.slots
.get(&run)
.and_then(|s| s.machine.info().trace.clone())
}
fn log_segment(&self) -> u64 {
match &self.writer {
ShardLog::Segments(w) => w.segment(),
ShardLog::Sections(_) if !self.env.recovers() => {
let framed = self.metrics.log_bytes
+ self.metrics.flushed_records * fusee_core::segment::FRAME_HEADER as u64;
1 + framed / self.cfg.segment_bytes.max(1)
}
ShardLog::Sections(_) => self.env.log_segment(),
}
}
fn trace_stall(&mut self, kind: StallKind, us: u64) {
match kind {
StallKind::Tick => self.metrics.max_tick_us = self.metrics.max_tick_us.max(us),
_ => self.metrics.max_flush_us = self.metrics.max_flush_us.max(us),
}
if us <= self.stall_us {
return;
}
self.metrics.stalls += 1;
let flush = kind != StallKind::Tick;
let stall = Stall {
shard: self.cfg.shard,
kind,
us,
at_ms: self.env.now_ms(),
records: if flush { self.flush_records } else { 0 },
bytes: if flush { self.flush_bytes } else { 0 },
preparing_segment: self.env.preparing_segment(),
checkpointing: self.ckpt_running,
parked: self.tick_parked,
maintenance_us: self.tick_maintenance_us,
loaded: self.core.loaded_len(),
slots: self.slots.len(),
};
if let Ok(mut t) = STALL_TRACE.lock()
&& let Some(file) = t.as_mut()
&& let Ok(mut line) = serde_json::to_vec(&stall)
{
use std::io::Write;
line.push(b'\n');
let _ = file.write_all(&line);
}
if self.stalls.len() >= 4096 {
self.stalls.pop_front();
}
self.stalls.push_back(stall);
}
#[inline]
fn step_time(&mut self, run: RunId, seq: Seq) -> Option<StepTime> {
if !self.cfg.step_times {
return None;
}
let began_ms = self.step_began.remove(&(run, seq))?;
Some(StepTime {
began_ms,
took_ms: self.now().saturating_sub(began_ms),
})
}
fn now(&self) -> u64 {
match self.clock.ms.get() {
Some(t) if self.clock.on && !self.clock.ran.get() => t,
_ => self.now_exact(),
}
}
fn tick_now(&self) -> u64 {
match self.clock.ms.get() {
Some(t) if self.clock.on => t,
_ => self.now_exact(),
}
}
fn now_exact(&self) -> u64 {
let t = self.env.now_ms();
if self.clock.on {
self.clock.ms.set(Some(t));
self.clock.ran.set(false);
}
t
}
fn metric_us(&self) -> u64 {
if !self.clock.on || self.stall_us != 0 || self.env.last_flush_us() > 0 {
return self.env.monotonic_us();
}
match self.clock.us.get() {
Some(t) => t,
None => {
let t = self.env.monotonic_us();
self.clock.us.set(Some(t));
t
}
}
}
fn set_ticking(&mut self, on: bool) {
self.clock.on = on;
self.clock.drives = 0;
self.clock.renew();
}
fn count_drive(&mut self) {
self.clock.drives += 1;
if self.clock.drives >= CLOCK_DRIVES {
self.clock.drives = 0;
self.clock.renew();
}
}
pub fn seal(&self, p: Payload) -> Result<Payload, Error> {
match &self.codec {
Some(c) if !p.is_empty() => Ok(Arc::from(c.encode(&p).map_err(Error::Codec)?)),
_ => Ok(p),
}
}
pub fn open(&self, p: Payload) -> Result<Payload, Error> {
match &self.codec {
Some(c) if !p.is_empty() => Ok(Arc::from(c.decode(&p).map_err(Error::Codec)?)),
_ => Ok(p),
}
}
fn open_outcome(&self, o: Outcome) -> Result<Outcome, Error> {
if o.status != Status::Ok {
return Ok(o);
}
Ok(Outcome {
status: o.status,
payload: self.open(o.payload)?,
})
}
fn native_code(&self, wtype: &str, version: u32) -> bool {
self.registry
.read()
.is_ok_and(|r| r.get(wtype, version).is_some())
}
fn runs_foreign(&self, wtype: &str, version: u32) -> bool {
self.foreign_host && !self.native_code(wtype, version)
}
pub fn durable(&self) -> Lsn {
self.durable
}
pub fn commit(&self) -> Lsn {
self.commit
}
pub fn last_lsn(&self) -> Lsn {
self.writer.last_lsn()
}
fn loaded_bytes(&self) -> u64 {
self.core.loaded_len() as u64 * LOADED_RUN_BYTES
+ self.core.journal_bytes() as u64
+ self.stream_bytes as u64
}
pub fn metrics(&self) -> Metrics {
let log = self.env.log_stats();
Metrics {
log_writes: log.writes,
log_device_bytes: log.device_bytes,
log_rolls: log.rolls,
log_owner: u64::from(log.owner),
parked_entries: self.core.parked_len() as u64,
run_table_bytes: self.core.table_bytes() as u64,
loaded_bytes: self.loaded_bytes(),
volatile_timers: self.timers.len() as u64,
..self.metrics
}
}
pub fn failed(&self) -> Option<&str> {
self.failed.as_deref()
}
fn check_alive(&self) -> Result<(), Error> {
match &self.failed {
Some(m) => Err(Error::Io(m.clone())),
None => Ok(()),
}
}
fn append(&mut self, rec: &RecordBuf) -> Result<Lsn, Error> {
self.append_with(rec, Some(rec))
}
fn append_with<S: fusee_core::record::ToName, B: fusee_core::record::ToPayload>(
&mut self,
rec: &Record<S, B>,
owned: Option<&RecordBuf>,
) -> Result<Lsn, Error> {
self.check_alive()?;
let now = self.tick_now();
let lsn = self.writer.append(rec, now).map_err(|_| {
Error::Codec(
"the record is longer than the log holds in one piece (a segment of a \
per-shard log, 64 MiB in the shared log): keep data of that size outside \
the journal and record a reference to it"
.into(),
)
})?;
match self.writer.last_encoded() {
Some(canonical) => self.core.apply_encoded(lsn, rec, Mode::LIVE, canonical),
None => self.core.apply(lsn, rec, Mode::LIVE),
}
let run = rec.run();
match rec {
Record::Heartbeat { .. } => {}
Record::Chunk { .. } | Record::StreamEnd { .. } => {
if let Some(rec) = owned {
self.stream_appended(lsn, rec);
}
}
_ => {
if let Record::Done { status, .. } = rec
&& (!self.streams.is_empty() || !self.stream_waiters.is_empty())
{
self.stream_run_done(run, lsn, *status);
}
if let Some(slot) = self.slots.get_mut(&run) {
slot.machine.gate = lsn;
}
}
}
self.metrics.records += 1;
#[cfg(feature = "opentelemetry")]
if let Some(telemetry) = &self.cfg.telemetry {
telemetry.append(self.cfg.shard);
}
if self.replicator.is_none()
&& self.cfg.sabotage == Sabotage::None
&& self.env.durable_on_append()
&& let ShardLog::Sections(w) = &mut self.writer
&& let Some((records, bytes, last)) = w.discard()
{
self.metrics.flushed_records += u64::from(records);
self.metrics.log_bytes += bytes;
self.durable = self.durable.max(last);
self.commit = self.commit.max(last);
self.commit_dirty = true;
}
if let Some(r) = self.replicator.as_mut() {
match self.writer.last_encoded() {
Some(buf) => r.appended(lsn, buf),
None => {
let mut buf = Vec::with_capacity(64);
encode_canonical(rec, &mut buf);
r.appended(lsn, &buf);
}
}
}
Ok(lsn)
}
fn stream_appended(&mut self, lsn: Lsn, rec: &RecordBuf) {
let run = rec.run();
let commit = self.commit;
let (name, at) = match rec {
Record::Chunk { stream, offset, .. } => (stream, *offset),
Record::StreamEnd { stream, chunks, .. } => (stream, *chunks),
_ => return,
};
let rs = self.streams.entry(run).or_default();
if rs.id.is_none() {
rs.id = self.core.get(run).map(|l| l.id.clone());
}
let i = match rs.logs.iter().position(|l| *l.name == **name) {
Some(i) => i,
None => {
rs.logs.push(StreamLog {
name: name.clone(),
first: at,
chunks: VecDeque::new(),
bytes: 0,
len: at,
end: None,
});
rs.logs.len() - 1
}
};
let log = &mut rs.logs[i];
match rec {
Record::Chunk {
offset, payload, ..
} => {
if *offset != log.len {
self.stream_bytes = self.stream_bytes.saturating_sub(log.bytes);
log.chunks.clear();
log.bytes = 0;
log.first = *offset;
}
log.chunks.push_back((lsn, payload.clone()));
log.bytes += payload.len();
self.stream_bytes += payload.len();
log.len = offset + 1;
while (log.bytes > STREAM_LOG_BYTES || log.chunks.len() > STREAM_LOG_CHUNKS)
&& log.chunks.front().is_some_and(|(l, _)| *l <= commit)
{
if let Some((_, p)) = log.chunks.pop_front() {
log.bytes -= p.len();
self.stream_bytes = self.stream_bytes.saturating_sub(p.len());
log.first += 1;
}
}
}
Record::StreamEnd { status, chunks, .. } => {
log.end = Some((lsn, *status));
log.len = log.len.max(*chunks);
}
_ => {}
}
if self.stream_waiters.contains_key(&run) {
self.stream_due.push(Reverse((lsn, run)));
}
}
fn stream_run_done(&mut self, run: RunId, lsn: Lsn, status: Status) {
let waiting = self.stream_waiters.contains_key(&run);
if waiting || self.streams.contains_key(&run) {
self.streams.entry(run).or_default().done = Some((lsn, status));
}
if waiting {
self.stream_due.push(Reverse((lsn, run)));
}
}
fn scan_stream(
&mut self,
run: RunId,
name: &str,
from: u64,
until: u64,
cap: usize,
) -> Result<StreamScan, Error> {
let (records, last_lsn, decoded) = match self.core.get(run) {
Some(l) => {
let last = l.last_lsn;
(self.journal_of(run)?, last, false)
}
None => match self.core.parked(run) {
Some(p) => {
let s = self.read_stored(p.blob())?;
(s.records, s.last_lsn, true)
}
None => return Err(Error::NotFound(format!("run {run}"))),
},
};
let mut scan = StreamScan {
chunks: Vec::new(),
more: false,
len: 0,
end: None,
done: None,
last_lsn,
id: None,
};
let mut bytes = 0;
for r in &records {
match r {
Record::Start { id, .. } if scan.id.is_none() => scan.id = Some(id.clone()),
Record::Chunk {
stream,
offset,
payload,
..
} if &**stream == name => {
scan.len = scan.len.max(offset + 1);
if *offset < from || *offset >= until || scan.more {
continue;
}
if !scan.chunks.is_empty() && bytes + payload.len() > cap {
scan.more = true;
continue;
}
if *offset == from + scan.chunks.len() as u64 {
bytes += payload.len();
scan.chunks.push(payload.clone());
}
}
Record::StreamEnd {
stream,
status,
chunks,
..
} if &**stream == name => {
scan.end = Some(*status);
scan.len = scan.len.max(*chunks);
}
Record::Done { status, .. } => scan.done = Some(*status),
_ => {}
}
}
if decoded {
self.reuse_decoded(records);
}
Ok(scan)
}
fn fill_stream_log(
&mut self,
run: RunId,
name: &str,
from: u64,
first: Option<u64>,
) -> Result<Option<StreamRead>, Error> {
let mut s = self.scan_stream(
run,
name,
from,
first.unwrap_or(u64::MAX),
STREAM_CACHE_BYTES,
)?;
let commit = self.commit;
let rs = self.streams.entry(run).or_default();
if rs.id.is_none() {
rs.id = s.id.clone();
}
if rs.done.is_none()
&& let Some(status) = s.done
{
rs.done = Some((if s.last_lsn <= commit { 0 } else { s.last_lsn }, status));
}
let bytes: usize = s.chunks.iter().map(|c| c.len()).sum();
let fits = !s.more && first.is_none_or(|f| from + s.chunks.len() as u64 == f);
if !fits {
let mut taken = 0;
let mut n = 0;
for c in &s.chunks {
if n > 0
&& (taken + c.len() > crate::stream::BATCH_BYTES
|| n >= crate::stream::BATCH_CHUNKS)
{
break;
}
taken += c.len();
n += 1;
}
s.chunks.truncate(n);
return Ok(Some(StreamRead {
run,
from,
chunks: s.chunks,
end: None,
wait: None,
id: s.id,
}));
}
match rs.logs.iter_mut().find(|l| &*l.name == name) {
Some(log) => {
for c in s.chunks.into_iter().rev() {
log.chunks.push_front((0, c));
}
log.first = from.min(log.first);
log.bytes += bytes;
}
None => rs.logs.push(StreamLog {
name: Arc::from(name),
first: from.min(s.len),
chunks: s.chunks.into_iter().map(|c| (0, c)).collect(),
bytes,
len: s.len,
end: s.end.map(|status| (0, status)),
}),
}
self.stream_bytes += bytes;
Ok(None)
}
fn read_stream(&mut self, run: RunId, name: &str, from: u64) -> Result<StreamRead, Error> {
let commit = if self.cfg.sabotage == Sabotage::EarlyChunks {
Lsn::MAX
} else {
self.commit
};
let mut out = StreamRead {
run,
from,
chunks: Vec::new(),
end: None,
wait: None,
id: None,
};
let first = self
.streams
.get(&run)
.and_then(|rs| rs.logs.iter().find(|l| &*l.name == name))
.map(|l| l.first);
if first.is_none_or(|f| from < f)
&& let Some(direct) = self.fill_stream_log(run, name, from, first)?
{
return Ok(direct);
}
let now = self.tick_now();
let Some(rs) = self.streams.get_mut(&run) else {
return Ok(out);
};
rs.read_at = now;
out.id = rs.id.clone();
let done = rs.done;
let Some(log) = rs.logs.iter().find(|l| &*l.name == name) else {
return Ok(out);
};
if from < log.first {
return Ok(out);
}
let mut bytes = 0;
let skip = usize::try_from(from - log.first).unwrap_or(usize::MAX);
if skip < log.chunks.len() {
for (lsn, p) in log.chunks.range(skip..) {
if *lsn > commit {
out.wait = Some(*lsn);
return Ok(out);
}
if !out.chunks.is_empty()
&& (bytes + p.len() > crate::stream::BATCH_BYTES
|| out.chunks.len() >= crate::stream::BATCH_CHUNKS)
{
return Ok(out);
}
bytes += p.len();
out.chunks.push(p.clone());
}
}
if from + out.chunks.len() as u64 >= log.len {
let closed = log.end.is_some();
match log.end.or(done) {
Some((lsn, status)) if lsn <= commit => out.end = Some(end_of(status, closed)),
Some((lsn, _)) => out.wait = Some(lsn),
None => {}
}
}
Ok(out)
}
fn stream_answer(&self, r: &StreamRead) -> Result<Payload, Error> {
let mut chunks = Vec::with_capacity(r.chunks.len());
for c in &r.chunks {
chunks.push(self.open(c.clone())?);
}
Ok(Arc::from(crate::stream::encode_batch(
r.run,
r.from,
&chunks,
r.end.as_ref(),
)))
}
fn answer_readers(&mut self, run: RunId, now: Option<u64>) {
let Some(waiters) = self.stream_waiters.remove(&run) else {
return;
};
let mut keep = Vec::new();
for w in waiters {
let r = self.read_stream(run, &w.name, w.from);
let answer = match r {
Ok(r)
if !r.chunks.is_empty()
|| r.end.is_some()
|| now.is_some_and(|t| t >= w.deadline_ms) =>
{
self.stream_answer(&r)
}
Ok(r) => {
if let Some(lsn) = r.wait {
self.stream_due.push(Reverse((lsn, run)));
}
keep.push(w);
continue;
}
Err(e) => Err(e),
};
self.released.push(match answer {
Ok(payload) => Output::Reply {
token: w.token,
status: Status::Ok,
payload,
},
Err(e) => Output::Reply {
token: w.token,
status: Status::Failed,
payload: Arc::from(e.to_string().as_bytes()),
},
});
}
if !keep.is_empty() {
self.stream_waiters.insert(run, keep);
}
}
fn wake_stream_readers(&mut self) {
while let Some(Reverse((lsn, run))) = self.stream_due.peek().copied() {
if lsn > self.commit && self.cfg.sabotage != Sabotage::EarlyChunks {
break;
}
self.stream_due.pop();
self.answer_readers(run, None);
}
}
fn expire_stream_reads(&mut self) {
if self.stream_waiters.is_empty() {
self.stream_wait_until = u64::MAX;
return;
}
let now = self.now();
if now < self.stream_wait_until {
return;
}
let runs: Vec<RunId> = self.stream_waiters.keys().copied().collect();
for run in runs {
self.answer_readers(run, Some(now));
}
self.stream_wait_until = self
.stream_waiters
.values()
.flatten()
.map(|w| w.deadline_ms)
.min()
.unwrap_or(u64::MAX);
}
fn drop_stream_logs(&mut self) {
if self.streams.is_empty() {
return;
}
let commit = self.commit;
let now = self.tick_now();
let core = &self.core;
let waiters = &self.stream_waiters;
let mut freed = 0;
self.streams.retain(|run, rs| {
let durable = |at: Option<Lsn>| at.is_none_or(|l| l <= commit);
let settled = durable(rs.done.map(|d| d.0))
&& rs
.logs
.iter()
.all(|l| durable(l.chunks.back().map(|c| c.0)) && durable(l.end.map(|e| e.0)));
let keep = !settled
|| waiters.contains_key(run)
|| now.saturating_sub(rs.read_at) < STREAM_READ_KEEP_MS
|| core.get(*run).is_some_and(|l| !l.is_done());
if !keep {
freed += rs.logs.iter().map(|l| l.bytes).sum::<usize>();
}
keep
});
self.stream_bytes = self.stream_bytes.saturating_sub(freed);
if self.streams.capacity() > 1024 && self.streams.capacity() > 4 * self.streams.len() {
self.streams.shrink_to(self.streams.len() * 2);
}
}
fn stream_run(&mut self, id: &str, run: Option<RunId>) -> Result<RunId, Error> {
let not_found = || Error::NotFound(id.to_owned());
let run = match run {
Some(r) => r,
None => match self.core.find(id) {
Some(Found::Loaded(r) | Found::Parked(r)) => r,
None => return Err(not_found()),
},
};
match self.core.get(run) {
Some(l) if *l.id == *id => Ok(run),
Some(_) => Err(not_found()),
None if self.core.parked(run).is_some() => Ok(run),
None => Err(not_found()),
}
}
fn wake(&mut self, run: RunId) {
if let Some(slot) = self.slots.get_mut(&run) {
slot.broadcasts += 1;
}
self.schedule(run);
}
fn schedule(&mut self, run: RunId) {
self.mark_active(run);
if let Some(slot) = self.slots.get_mut(&run) {
if slot.foreign && !slot.wake_queued {
slot.wake_queued = true;
self.foreign.push_back(ForeignEvent::Wake { run });
}
slot.waker.wake_local();
} else if self.core.get(run).is_some() || self.core.is_parked(run) {
self.stash.entry(run).or_default().push(Stashed::Wake);
}
}
fn complete(&mut self, run: RunId, seq: Seq, c: Completion) {
match self.slots.get_mut(&run) {
Some(slot) => {
if slot.machine.complete(seq, c) {
self.note_undelivered(run);
self.schedule(run);
}
}
None => self
.stash
.entry(run)
.or_default()
.push(Stashed::Complete(seq, c)),
}
}
fn mark_active(&mut self, run: RunId) {
if let Some(slot) = self.slots.get_mut(&run) {
slot.suspended_since = None;
if !slot.dirty {
slot.dirty = true;
self.dirty.push(run);
}
}
}
fn note_undelivered(&mut self, run: RunId) {
if let Some(slot) = self.slots.get_mut(&run)
&& slot.foreign
&& !slot.undelivered
{
slot.undelivered = true;
self.undelivered.push(run);
}
}
fn has_undelivered(&mut self) -> bool {
let slots = &mut self.slots;
self.undelivered.retain(|run| match slots.get_mut(run) {
Some(slot) if slot.machine.has_queued_deliveries() => true,
Some(slot) => {
slot.undelivered = false;
false
}
None => false,
});
!self.undelivered.is_empty()
}
fn remove_slot(&mut self, run: RunId) -> Option<Removed> {
let (idx, mut slot) = self.slots.remove(&run)?;
self.removed.push((run, idx));
if !self.step_began.is_empty() {
self.step_began.retain(|(r, _), _| *r != run);
}
let foreign = slot.foreign;
if self.slot_spare.len() < SLOT_SPARE && slot.machine.reusable() {
slot.machine.release();
slot.resolved = None;
self.slot_spare.push(slot);
}
Some(Removed { foreign })
}
fn new_slot(
&mut self,
run: RunId,
journal: JournalIn<'_>,
gate: Lsn,
foreign: bool,
) -> Result<(), fusee_core::run::DeterminismError> {
let slot = match self.slot_spare.pop() {
Some(mut slot) => {
match journal {
JournalIn::Borrowed(j) => slot.machine.rebuild(j, gate)?,
JournalIn::Decoded(j) => slot.machine.rebuild_drain(j, gate)?,
}
match Arc::get_mut(&mut slot.waker) {
Some(w) => {
w.run = run;
*w.queued.get_mut() = false;
*w.woken.get_mut() = false;
}
None => slot.waker = self.new_waker(run),
}
let Slot {
machine: _,
waker: _,
foreign: f,
suspended_since,
dirty,
broadcasts,
wake_queued,
gate_at,
undelivered,
resolved,
} = &mut *slot;
*f = foreign;
*suspended_since = None;
*dirty = true;
*broadcasts = 0;
*wake_queued = false;
*gate_at = None;
*undelivered = false;
*resolved = None;
slot
}
None => Box::new(Slot {
machine: match journal {
JournalIn::Borrowed(j) => RunMachine::new(j, gate)?,
JournalIn::Decoded(j) => {
let m = RunMachine::new(j, gate)?;
j.clear();
m
}
},
waker: self.new_waker(run),
foreign,
suspended_since: None,
dirty: true,
broadcasts: 0,
wake_queued: false,
gate_at: None,
undelivered: false,
resolved: None,
}),
};
let object = (slot.machine.info().policy & fusee_core::record::policy::OBJECT != 0)
.then(|| slot.machine.info().input.clone());
self.slots.insert(run, slot);
self.dirty.push(run);
if let Some(input) = object {
self.object_on_slot(run, input);
}
Ok(())
}
fn new_waker(&self, run: RunId) -> Arc<RunWaker> {
Arc::new(RunWaker {
run,
queued: AtomicBool::new(false),
woken: AtomicBool::new(false),
ready: self.ready.clone(),
})
}
fn prepare(&mut self, run: RunId, journal: &mut Vec<RecordBuf>) {
let Some(l) = self.core.get(run) else {
return;
};
if l.is_done() || self.slots.contains_key(&run) {
return;
}
let gate = l.last_lsn;
let foreign = self.runs_foreign(&l.wtype, l.version);
match self.new_slot(run, JournalIn::Decoded(journal), gate, foreign) {
Ok(()) => self.instantiate.push(run),
Err(e) => {
self.metrics.determinism_errors += 1;
self.stuck.insert(run, e.to_string());
}
}
}
fn journal_of(&mut self, run: RunId) -> Result<Vec<RecordBuf>, Error> {
let (blob, suffix) = match self.core.get(run) {
Some(l) => (l.blob, l.journal.clone()),
None => match self.core.parked(run) {
Some(p) => (Some(p.blob()), Vec::new()),
None => return Ok(Vec::new()),
},
};
let mut out = Vec::new();
if let Some(b) = blob {
out.extend(self.blob_records(b)?);
}
out.extend(decode_canonical_all(&suffix).map_err(|e| Error::Io(e.to_string()))?);
Ok(out)
}
fn blob_records(&mut self, b: BlobRef) -> Result<Vec<RecordBuf>, Error> {
Ok(self.read_stored(b)?.records)
}
fn read_chain(&mut self, b: BlobRef) -> Result<Chain, Error> {
let frame = read_store(&mut self.env, &self.tail, b.offset, b.len)?;
self.chain_from(b, &frame)
}
fn chain_from(&mut self, b: BlobRef, frame: &[u8]) -> Result<Chain, Error> {
let mut newest = read_entry(frame).map_err(io_err)?;
let run = newest.run();
let (last_lsn, id_hash, timers) = match &mut newest {
StoreEntry::Park {
last_lsn,
id_hash,
timers,
..
}
| StoreEntry::Delta {
last_lsn,
id_hash,
timers,
..
} => (*last_lsn, *id_hash, std::mem::take(timers)),
StoreEntry::Done {
last_lsn, id_hash, ..
} => (*last_lsn, *id_hash, Vec::new()),
StoreEntry::Tombstone { .. } | StoreEntry::Index { .. } => (0, 0, Vec::new()),
};
let env = &mut self.env;
let tail = &self.tail;
let mut failed = None;
let read = chain_journal(newest, b.len, &mut |at| match read_store(
env, tail, at.offset, at.len,
) {
Ok(f) => read_entry(&f),
Err(e) => {
failed = Some(e);
Err(fusee_core::wire::DecodeError(
"unreadable frame in a delta chain",
))
}
});
if let Some(e) = failed {
return Err(e.into());
}
let (journal, chain, depth) = read.map_err(io_err)?;
self.metrics.journal_frames_read += 1 + u64::from(depth);
self.metrics.journal_bytes_read += chain;
Ok(Chain {
run,
journal,
last_lsn,
id_hash,
timers,
chain,
depth,
})
}
fn read_stored(&mut self, b: BlobRef) -> Result<Stored, Error> {
self.stored_from(b, None)
}
fn stored_from(&mut self, b: BlobRef, frame: Option<&[u8]>) -> Result<Stored, Error> {
let read;
let frame = match frame {
Some(f) => f,
None => {
read = read_store(&mut self.env, &self.tail, b.offset, b.len)?;
&read
}
};
let mut records = std::mem::take(&mut self.decoded);
if let Some((_, last_lsn, journal)) =
fusee_core::store::read_whole_journal(frame).map_err(io_err)?
{
self.metrics.journal_frames_read += 1;
self.metrics.journal_bytes_read += u64::from(b.len);
decode_canonical_into(journal, &mut self.names, &mut records).map_err(io_err)?;
return Ok(Stored {
records,
last_lsn,
chain: u64::from(b.len),
depth: 0,
});
}
let c = self.chain_from(b, frame)?;
decode_canonical_into(&c.journal, &mut self.names, &mut records).map_err(io_err)?;
Ok(Stored {
records,
last_lsn: c.last_lsn,
chain: c.chain,
depth: c.depth,
})
}
fn reuse_decoded(&mut self, mut records: Vec<RecordBuf>) {
if records.capacity() <= 1024 {
records.clear();
self.decoded = records;
}
}
fn store_flush(&mut self) -> Result<(), Error> {
if self.tail.buf.is_empty() {
return Ok(());
}
let written = self.env.store_append(&self.tail.buf);
let expected = self.tail.at;
self.tail.buf.clear();
if self.tail.buf.capacity() > 4 * STORE_TAIL_MAX {
self.tail.buf = Vec::new();
}
if written? != expected {
return Err(Error::Io(
"run store frames were written at an unexpected offset".into(),
));
}
Ok(())
}
fn store_flush_or_fail(&mut self) {
if let Err(e) = self.store_flush() {
self.fail(format!("run store write failed: {e}"));
}
}
fn store_len(&self) -> u64 {
if self.tail.buf.is_empty() {
self.env.store_len()
} else {
self.tail.at + self.tail.buf.len() as u64
}
}
fn unpark_stored(&mut self, run: RunId) -> Result<Option<Stored>, Error> {
self.unpark_from(run, None)
}
fn unpark_from(
&mut self,
run: RunId,
frame: Option<(BlobRef, &[u8])>,
) -> Result<Option<Stored>, Error> {
let Some(p) = self.core.parked(run) else {
return Ok(None);
};
let frame = frame.filter(|(b, _)| *b == p.blob()).map(|(_, f)| f);
let s = self.stored_from(p.blob(), frame)?;
let now = self.now();
if !self.core.unpark(run, s.last_lsn, &s.records, now) {
return Err(Error::Io(format!(
"run {run}: stored journal is unreadable"
)));
}
self.core.spilled(run, p.blob(), s.chain, s.depth);
Ok(Some(s))
}
fn load(&mut self, run: RunId) -> Result<(), Error> {
self.load_from(run, None)
}
fn load_from(&mut self, run: RunId, frame: Option<(BlobRef, &[u8])>) -> Result<(), Error> {
let Some(mut s) = self.unpark_from(run, frame)? else {
return Ok(());
};
self.metrics.runs_loaded += 1;
if self.core.get(run).is_some_and(|l| !l.is_done()) {
self.prepare(run, &mut s.records);
self.core.reoffer(run);
}
self.reuse_decoded(s.records);
Ok(())
}
fn ensure_loaded(&mut self, run: RunId) -> Result<(), Error> {
if self.core.is_parked(run) {
self.load(run)?;
}
Ok(())
}
fn park_done(&mut self, run: RunId) -> Result<(), Error> {
let Some(l) = self.core.get(run) else {
return Ok(());
};
if l.done.is_none() {
return Ok(());
}
if !self.cfg.keep_history
&& !self.cfg.archive_history
&& self.store_v3
&& let Some(prev) = l.blob
&& l.blob_depth < MAX_CHAIN
&& !l.journal.is_empty()
{
let entry = EntryRef::Delta {
run,
last_lsn: l.last_lsn,
id_hash: hash_id(&l.id),
timers: &[],
prev,
depth: l.blob_depth + 1,
chain: l.blob_chain,
journal: &l.journal,
};
let blob = self.tail.put(self.env.store_len(), entry)?;
self.core.park(run, blob, true);
if self.tail.full() {
self.store_flush()?;
}
return Ok(());
}
let streamed = l.streams;
let unsuccessful = l.done.is_some_and(|(s, _)| {
matches!(
s,
Status::Failed | Status::Cancelled | Status::TimedOut | Status::Terminated
)
});
let history = if self.cfg.keep_history || unsuccessful {
Some(self.journal_of(run)?)
} else {
None
};
let streams: Vec<RecordBuf> = if history.is_none() && streamed {
self.journal_of(run)?
.into_iter()
.filter(|r| matches!(r, Record::Chunk { .. } | Record::StreamEnd { .. }))
.collect()
} else {
Vec::new()
};
let Some(l) = self.core.get(run) else {
return Ok(());
};
let Some((status, finished_ms)) = l.done else {
return Ok(());
};
let (last_lsn, id_hash) = (l.last_lsn, hash_id(&l.id));
let mut journal = std::mem::take(&mut self.enc);
journal.clear();
match history {
Some(records) => {
for r in records {
encode_canonical(&r, &mut journal);
}
}
None => {
let start: RecordRef<'_> = Record::Start {
run,
id: &l.id,
wtype: &l.wtype,
version: l.version,
codec: l.codec,
started_ms: l.started_ms,
deadline_ms: l.deadline_ms,
parent: l.parent,
policy: l.policy,
from: l.from,
copies: 0,
update: None,
input: &[],
meta: None,
};
encode_canonical(&start, &mut journal);
if !l.attrs.is_empty() {
let attrs = merge_attributes(&l.attrs);
if attrs != "{}" {
let a: RecordRef<'_> = Record::Attributes {
run,
seq: 0,
payload: attrs.as_bytes(),
};
encode_canonical(&a, &mut journal);
}
}
for r in &streams {
encode_canonical(r, &mut journal);
}
let done: RecordRef<'_> = Record::Done {
run,
status,
finished_ms,
payload: l.result.as_deref().unwrap_or(&[]),
by: None,
};
encode_canonical(&done, &mut journal);
}
}
let entry = EntryRef::Done {
run,
last_lsn,
id_hash,
status,
finished_ms,
journal: &journal,
};
let blob = if self.cfg.archive_history {
let mut frame = Vec::new();
encode_entry_into(entry, &mut frame);
self.store_flush()?;
let len = u32::try_from(frame.len())
.map_err(|_| Error::Io("run store frame exceeds 4 GiB".into()))?;
let offset = self.env.store_append_history(&frame, run, last_lsn)?;
BlobRef { offset, len }
} else {
self.tail.put(self.env.store_len(), entry)?
};
self.enc = journal;
if let Some(l) = self.core.get(run)
&& l.blob.is_some()
{
self.obsolete_store_bytes += l.blob_chain;
}
self.core.park(run, blob, false);
if self.tail.full() {
self.store_flush()?;
}
Ok(())
}
fn store_active(&mut self, run: RunId) -> Result<(BlobRef, u64, u8), Error> {
let l = self
.core
.get(run)
.ok_or_else(|| Error::Io("missing active run".into()))?;
let old = l.blob;
let (old_chain, old_depth) = (l.blob_chain, l.blob_depth);
if let Some(b) = old
&& l.journal.is_empty()
{
return Ok((b, old_chain, old_depth));
}
let delta = old.filter(|_| self.store_v3 && old_depth < MAX_CHAIN);
let mut whole = Vec::new();
if let (None, Some(b)) = (delta, old) {
let c = self.read_chain(b)?;
if c.run != run || c.journal.is_empty() {
return Err(Error::Io("wrong active journal in run store".into()));
}
whole = c.journal;
}
let l = self
.core
.get(run)
.ok_or_else(|| Error::Io("missing active run".into()))?;
let (last_lsn, id_hash) = (l.last_lsn, hash_id(&l.id));
let mut timers = std::mem::take(&mut self.timer_scratch);
timers.clear();
timers.extend(l.armed_timers());
let store_len = self.env.store_len();
let put = match delta {
Some(prev) => self
.tail
.put(
store_len,
EntryRef::Delta {
run,
last_lsn,
id_hash,
timers: &timers,
prev,
depth: old_depth + 1,
chain: old_chain,
journal: &l.journal,
},
)
.map(|blob| (blob, old_chain + u64::from(blob.len), old_depth + 1)),
None => {
whole.extend_from_slice(&l.journal);
self.tail
.put(
store_len,
EntryRef::Park {
run,
last_lsn,
id_hash,
timers: &timers,
journal: &whole,
},
)
.map(|blob| (blob, u64::from(blob.len), 0))
}
};
self.timer_scratch = timers;
let stored = put?;
if stored.2 == 0 && old.is_some() {
self.obsolete_store_bytes += old_chain;
}
if self.tail.full() {
self.store_flush()?;
}
Ok(stored)
}
fn park(&mut self, run: RunId) -> Result<bool, Error> {
if self.core.get(run).is_none_or(|l| l.is_done()) {
return Ok(false);
}
if self.object_idle(run)
&& self
.slots
.get(&run)
.is_some_and(|s| s.machine.journal_size() > 0)
{
if let Some(new) = self.object_compact(run)? {
self.object_release(new)?;
}
return Ok(false);
}
if self.core.get(run).is_some_and(|l| l.has_queued_events()) {
self.mark_active(run);
return Ok(false);
}
let (blob, _, depth) = self.store_active(run)?;
self.core.park(run, blob, depth > 0);
self.object_parked(run);
if self.remove_slot(run).is_some_and(|slot| slot.foreign) {
self.foreign.push_back(ForeignEvent::Ended { run });
}
self.current_step.remove(&run);
self.metrics.runs_parked += 1;
Ok(true)
}
fn end_run(&mut self, run: RunId, status: Status, message: &str) -> Result<Option<Lsn>, Error> {
self.ensure_loaded(run)?;
if self.core.get(run).is_none_or(|l| l.is_done()) {
return Ok(None);
}
let now = self.now();
let rec = Record::Done {
run,
status,
finished_ms: now,
payload: Arc::from(message.as_bytes()),
by: self.by.take(),
};
let lsn = self.append(&rec)?;
if let Some(slot) = self.remove_slot(run)
&& slot.foreign
{
self.foreign.push_back(ForeignEvent::Ended { run });
}
self.release_waiting_requests(run);
Ok(Some(lsn))
}
fn process_events(&mut self) -> bool {
const LOAD_CHUNK: usize = 1024;
const OVER_BUDGET: usize = 32;
let share = |i: &Self| {
let share = if i.over_budget() {
OVER_BUDGET
} else {
LOAD_CHUNK
};
share.min(i.load_room.saturating_sub(i.foreign.len()))
};
let mut any = false;
let mut loads = 0;
while loads < share(self) && !self.held_loads.is_empty() {
let n = (share(self) - loads).min(self.held_loads.len());
let batch: Vec<RunId> = self.held_loads.drain(..n).collect();
loads += n;
any = true;
self.load_batch(batch);
}
while let Some(ev) = self.core.take_event() {
any = true;
if let Event::NeedLoad { run } = ev {
let mut batch = std::mem::take(&mut self.load_buf);
batch.clear();
batch.push(run);
while batch.len() + loads < LOAD_CHUNK
&& matches!(self.core.peek_event(), Some(Event::NeedLoad { .. }))
{
if let Some(Event::NeedLoad { run }) = self.core.take_event() {
batch.push(run);
}
}
if loads + batch.len() > share(self) || !self.held_loads.is_empty() {
self.held_loads.extend(batch.drain(..));
self.load_buf = batch;
continue;
}
loads += batch.len();
self.load_batch(batch);
continue;
}
match ev {
Event::Started { .. } => {}
Event::Timer { run, seq, kind } => match kind {
TimerKind::Sleep => self.complete(run, seq, Completion::Fired),
TimerKind::WaitTimeout => self.complete(run, seq, Completion::WaitTimedOut),
TimerKind::Retry => self.wake(run),
TimerKind::Deadline => {
if let Err(e) =
self.end_run(run, Status::TimedOut, "workflow deadline exceeded")
{
self.fail(e.to_string());
}
}
},
Event::SignalReady {
run,
seq,
ordinal,
payload,
} => self.complete(run, seq, Completion::Signal { ordinal, payload }),
Event::Finished { run } => {
self.metrics.runs_finished += 1;
self.release_waiting_requests(run);
if let Some(slot) = self.remove_slot(run)
&& slot.foreign
{
self.foreign.push_back(ForeignEvent::Ended { run });
}
self.current_step.remove(&run);
self.stash.remove(&run);
if let Err(e) = self.park_done(run) {
self.fail(e.to_string());
}
}
Event::NeedLoad { .. } => {}
}
}
self.store_flush_or_fail();
any
}
fn load_batch(&mut self, mut batch: Vec<RunId>) {
const SPAN: u64 = 256 << 10;
const GAP: u64 = 4 << 10;
let mut at = std::mem::take(&mut self.load_at);
at.clear();
at.extend(
batch
.drain(..)
.filter_map(|r| self.core.parked(r).map(|p| (p.blob(), r))),
);
self.load_buf = batch;
at.sort_unstable_by_key(|(b, r)| (b.offset, *r));
let written = self.env.store_len();
let mut span = Vec::new();
let mut i = 0;
while i < at.len() {
let start = at[i].0.offset;
let mut end = start + u64::from(at[i].0.len);
let mut j = i + 1;
while j < at.len() {
let (b, _) = at[j];
let b_end = b.offset + u64::from(b.len);
if b.offset > end + GAP || b_end - start > SPAN || b_end > written {
break;
}
end = end.max(b_end);
j += 1;
}
let read = if j - i > 1 {
match self.env.store_read(start, (end - start) as u32) {
Ok(bytes) => {
span = bytes;
true
}
Err(e) => {
self.fail(e.to_string());
return;
}
}
} else {
false
};
for &(b, run) in &at[i..j] {
let frame = read.then(|| {
let from = (b.offset - start) as usize;
(b, &span[from..from + b.len as usize])
});
if let Err(e) = self.load_from(run, frame) {
self.fail(e.to_string());
}
}
i = j;
}
self.load_at = at;
}
fn over_budget(&self) -> bool {
self.cfg.memory_budget != 0 && self.loaded_bytes() > self.cfg.memory_budget
}
fn fail(&mut self, message: String) {
if self.failed.is_none() {
self.failed = Some(message);
}
}
fn send_out(&mut self, target: u32, msg: Message) {
let id = self.next_out;
self.next_out += 1;
self.outbox.insert(
id,
Outgoing {
id,
target,
msg,
sent_ms: 0,
},
);
}
fn msg_token(&mut self, from: (u32, u64)) -> u64 {
let t = self.next_msg_token;
self.next_msg_token += 1;
self.msg_tokens.insert(t, from);
t
}
fn ack_at_tail(&mut self, from: (u32, u64)) {
let t = self.msg_token(from);
let tail = self.writer.last_lsn();
self.core.add_output(tail, Output::Ack(t));
}
fn route_output(&mut self, o: Output) {
let shards = self.cfg.shards;
match o {
Output::Ack(t) if t >= MSG_TOKEN => {
if let Some(from) = self.msg_tokens.remove(&t) {
self.message_acks.push(from);
}
}
Output::Reject(t) if t >= MSG_TOKEN => {
if let Some(from) = self.msg_tokens.remove(&t) {
self.message_acks.push(from);
}
}
Output::Ack(_) | Output::Reject(_) | Output::Reply { .. } | Output::Refuse { .. } => {
self.released.push(o)
}
Output::Result {
run,
status,
ref payload,
} => {
if !status.is_handover() {
for parent in self.subscribers.remove(&run).unwrap_or_default() {
self.send_out(
parent.shard,
Message::ChildResult {
parent,
status,
payload: payload.clone(),
},
);
}
}
let payload = if status == Status::Ok {
match self.open(payload.clone()) {
Ok(p) => p,
Err(e) => {
self.released.push(Output::Result {
run,
status: Status::Failed,
payload: Arc::from(e.to_string().as_bytes()),
});
return;
}
}
} else {
payload.clone()
};
self.released.push(Output::Result {
run,
status,
payload,
});
}
Output::UpdateResult {
run,
id,
status,
payload,
..
} => {
let tokens = self.update_waiters.remove(&(run, id)).unwrap_or_default();
if tokens.is_empty() {
return;
}
let (status, payload) = match (status, self.open(payload.clone())) {
(Status::Ok, Ok(p)) => (Status::Ok, p),
(Status::Ok, Err(e)) => (Status::Failed, Arc::from(e.to_string().as_bytes())),
(s, _) => (s, payload),
};
for token in tokens {
self.released.push(Output::Reply {
token,
status,
payload: payload.clone(),
});
}
}
Output::Close {
target,
wtype,
parent,
terminate,
reason,
} => {
let shard = fusee_core::shard::shard_of(&target, shards);
self.send_out(
shard,
Message::Close {
target: target.to_string(),
wtype: wtype.to_string(),
parent,
terminate,
reason,
},
);
}
Output::ChildStart {
parent,
wtype,
id,
flags,
deadline_ms,
input,
} => {
let target = fusee_core::shard::shard_of(&id, shards);
let trace = self.trace_of(parent.run);
self.send_out(
target,
Message::ChildStart {
parent,
wtype: wtype.to_string(),
id: id.to_string(),
flags,
deadline_ms,
input,
trace,
},
);
}
Output::ChildResult {
parent,
status,
payload,
} => self.send_out(
parent.shard,
Message::ChildResult {
parent,
status,
payload,
},
),
Output::Signal {
target,
name,
payload,
dedup,
} => {
let shard = fusee_core::shard::shard_of(&target, shards);
self.send_out(
shard,
Message::Signal {
target: target.to_string(),
name: name.to_string(),
payload,
dedup,
},
);
}
}
}
fn on_commit(&mut self) -> bool {
if self.failed.is_some() {
return false;
}
let mut any = false;
let mut out = Vec::new();
let upto = if self.cfg.sabotage == Sabotage::EarlyRelease {
self.writer.last_lsn()
} else {
self.commit
};
self.core.release(upto, &mut out);
any |= !out.is_empty();
for o in out {
self.route_output(o);
}
let mut now_us = None;
while let Some((lsn, run, began)) = self.gates.peek() {
if lsn > self.commit {
break;
}
self.gates.pop();
if let Some(slot) = self.slots.get_mut(&run)
&& slot.gate_at.is_some_and(|g| g <= self.commit)
{
slot.gate_at = None;
}
let now_us = *now_us.get_or_insert_with(|| self.metric_us());
let elapsed = now_us.saturating_sub(began);
self.metrics.gate_wait_us_total += elapsed;
#[cfg(feature = "opentelemetry")]
if let Some(telemetry) = &self.cfg.telemetry {
telemetry.gate(self.cfg.shard, elapsed);
}
self.wake(run);
any = true;
}
if let Some((cp, _)) = &self.ckpt_pending
&& cp.lsn <= self.commit
&& !self.ckpt_running
&& let Some((cp, del)) = self.ckpt_pending.take()
{
self.ckpt_running = true;
self.env.checkpoint(cp, del);
}
if !self.stream_due.is_empty() {
self.wake_stream_readers();
}
any
}
fn set_durable(&mut self, lsn: Lsn, kind: StallKind) {
let elapsed = self.metric_us().saturating_sub(self.flush_started_us);
self.slow_flushes = if self.env.last_flush_us() > self.cfg.inline_max_us {
self.slow_flushes.saturating_add(1)
} else {
0
};
self.metrics.flush_us_total += elapsed;
#[cfg(feature = "opentelemetry")]
if let Some(telemetry) = &self.cfg.telemetry {
telemetry.flush(self.cfg.shard, elapsed);
}
if self.stall_us != 0 {
self.trace_stall(kind, elapsed);
}
self.durable = self.durable.max(lsn);
match self.replicator.as_mut() {
Some(r) => {
r.durable(self.durable);
self.commit = self.commit.max(r.commit());
}
None => self.commit = self.durable,
}
}
fn poll_io(&mut self) -> bool {
let mut any = false;
if let Some(res) = self.env.poll_flushed() {
any = true;
match res {
Ok(lsn) => {
let mut started = None;
while let Some(&(last, at)) = self.submitted.front() {
if last > lsn {
break;
}
started.get_or_insert(at);
self.submitted.pop_front();
}
if let Some(at) = started {
self.flush_started_us = at;
}
self.in_flight = self.submitted.back().map(|&(last, _)| last);
self.set_durable(lsn, StallKind::Flush);
}
Err(e) => {
self.in_flight = None;
self.submitted.clear();
self.fail(format!("log flush failed: {e}; the shard has stopped"));
}
}
}
if let Some(r) = self.replicator.as_ref() {
let c = r.commit();
if c > self.commit {
self.commit = c;
any = true;
}
}
if let Some(res) = self.env.poll_checkpoint() {
any = true;
self.ckpt_running = false;
match res {
Ok(lsn) => {
self.last_ckpt_lsn = lsn;
self.metrics.checkpoints += 1;
}
Err(e) => self.fail(format!("checkpoint failed: {e}")),
}
}
if any {
self.on_commit();
}
any
}
fn poll_timers(&mut self) -> bool {
if self.cfg.replay_only {
return false;
}
let now = self.tick_now();
let mut any = false;
if self.core.next_timer().is_some_and(|t| t <= now) {
self.core.poll_timers(now);
any = self.core.has_events() || self.core.next_timer().is_some_and(|t| t <= now);
}
while let Some(Reverse((at, id))) = self.timers.peek().copied() {
if at > now {
break;
}
self.timers.pop();
if let Some(w) = self.timer_wakers.remove(&id) {
w.wake();
any = true;
}
}
any
}
fn flush_at_gate(&mut self) -> bool {
if self.gate_flushes == 0
|| self.drive_backlog > 0
|| self.in_flight.is_some()
|| self.failed.is_some()
|| !self.foreign.is_empty()
|| !self.instantiate.is_empty()
|| self.core.has_events()
|| self.ready.queued.load(Ordering::Acquire)
|| take_local(&self.ready, None)
|| match &self.host_queue {
Some(q) => q.load(Ordering::Acquire) > 0,
None => !self.host_never,
}
{
return false;
}
self.gate_flushes -= 1;
self.flush_pending(true, true)
}
fn maybe_flush(&mut self, idle: bool) -> bool {
self.flush_pending(idle, false)
}
fn flush_pending(&mut self, idle: bool, lone: bool) -> bool {
if self.failed.is_some()
|| (self.in_flight.is_some() && self.submitted.len() >= self.env.max_in_flight())
{
return false;
}
if !self.heartbeats.is_empty() {
for (_, rec) in std::mem::take(&mut self.heartbeats) {
let run = rec.run();
let live = match (&rec, self.slots.get(&run)) {
(Record::Heartbeat { seq, .. }, Some(slot)) => {
slot.machine.is_live_pending(*seq)
}
_ => false,
};
if live && let Err(e) = self.append(&rec) {
self.fail(e.to_string());
}
}
}
if self.commit_dirty {
self.commit_dirty = false;
self.on_commit();
return true;
}
if !self.writer.has_pending() {
return false;
}
let Some(batch) = self.writer.take() else {
return false;
};
let submitted_at = self.metric_us();
if self.in_flight.is_none() {
self.flush_started_us = submitted_at;
}
#[cfg(feature = "opentelemetry")]
if let Some(telemetry) = &self.cfg.telemetry {
telemetry.batch(self.cfg.shard, batch.records);
}
self.metrics.flushes += 1;
self.metrics.flushed_records += u64::from(batch.records);
self.metrics.log_bytes += batch.bytes;
self.flush_records = batch.records;
self.flush_bytes = batch.bytes;
let slow_disk = self.slow_flushes >= SLOW_DISK_FLUSHES;
let fast = self.env.last_flush_us() <= self.cfg.inline_max_us
|| (slow_disk && (lone || (idle && crate::spin::few_busy())));
if idle
&& fast
&& self.in_flight.is_none()
&& self.cfg.inline_flush
&& self.env.supports_inline()
&& !self.has_undelivered()
{
self.metrics.inline_flushes += 1;
let flushed = self.env.flush_inline(batch);
if self.env.last_flush_us() > 0 {
self.clock.renew();
}
match flushed {
Ok(lsn) => {
self.set_durable(lsn, StallKind::InlineFlush);
self.on_commit();
}
Err(e) => self.fail(format!("log flush failed: {e}; the shard has stopped")),
}
} else {
self.in_flight = Some(batch.last_lsn);
self.submitted.push_back((batch.last_lsn, submitted_at));
self.env.submit(batch);
}
true
}
fn maintenance(&mut self) {
if self.failed.is_some() {
return;
}
let started = (self.stall_us != 0).then(|| self.env.monotonic_us());
if self.stream_wait_until != u64::MAX {
self.expire_stream_reads();
}
self.trim_peak = self.trim_peak.max(self.loaded_bytes());
let swept = self.last_sweep_ms;
self.maintain();
let activity = self.metrics.runs_loaded + self.metrics.runs_started;
if activity == self.spare_mark {
let freed = self.slot_spare.len().saturating_sub(SLOT_SPARE_IDLE);
if freed > 0 {
self.slot_spare.truncate(SLOT_SPARE_IDLE);
self.slot_spare.shrink_to(SLOT_SPARE_IDLE);
}
let parts = self.core.shrink_spare(fusee_core::shard::SPARE_RUNS_IDLE);
self.pools_shrunk = true;
crate::trim::note_freed(((freed + parts) as u64) << 10);
}
self.spare_mark = activity;
let loaded = self.loaded_bytes();
if loaded.saturating_mul(2) < self.trim_peak && self.trim_peak - loaded >= 8 << 20 {
crate::trim::note_freed(self.trim_peak - loaded);
self.trim_peak = loaded;
}
if self.last_sweep_ms != swept {
self.advance_index_build();
}
self.store_flush_or_fail();
if let Some(t) = started {
self.tick_maintenance_us = self.env.monotonic_us().saturating_sub(t);
}
}
pub fn due(&self) -> Due {
let busy = self.ready.queued.load(Ordering::Acquire)
|| self.core.has_events()
|| !self.instantiate.is_empty()
|| !self.foreign.is_empty()
|| self.undelivered.iter().any(|r| {
self.slots
.get(r)
.is_some_and(|s| s.machine.has_queued_deliveries())
})
|| !self.held_loads.is_empty()
|| !self.heartbeats.is_empty()
|| !self.offered.is_empty()
|| self.in_flight.is_some()
|| self.writer.has_pending()
|| self.commit_dirty
|| self.ckpt_running
|| self.ckpt_pending.is_some();
let mut running: Vec<String> = Vec::new();
for slot in self.slots.entries.iter().flatten() {
if !slot.machine.is_suspended() {
let t: &str = &slot.machine.info().wtype;
if !running.iter().any(|r| r == t) {
running.push(t.to_owned());
}
}
}
let mut at = self.core.next_timer();
if let Some(Reverse((t, _))) = self.timers.peek() {
at = Some(at.map_or(*t, |a| a.min(*t)));
}
if let Some(sent) = self.outbox.values().map(|o| o.sent_ms).min() {
let t = sent.saturating_add(RESEND_MS);
at = Some(at.map_or(t, |a| a.min(t)));
}
Due {
busy,
running,
awaiting: self.awaiting.keys().cloned().collect(),
at_ms: at,
steps: self.metrics.steps_run,
open_us: self.metrics.open_us,
unacked: self.outbox.len(),
}
}
fn maintain(&mut self) {
let now = self.now();
if now.saturating_sub(self.last_sweep_ms) >= 100 {
self.last_sweep_ms = now;
let park_after = self.cfg.park_after_ms;
let mut to_park = Vec::new();
for run in std::mem::take(&mut self.dirty) {
let parkable = self.parkable(run);
let Some(slot) = self.slots.get_mut(&run) else {
continue;
};
if park_after == u64::MAX {
slot.dirty = false;
} else if parkable {
slot.dirty = false;
slot.suspended_since = Some(now);
self.park_queue.push_back((now, run));
} else {
self.dirty.push(run);
}
}
while let Some(&(since, run)) = self.park_queue.front() {
if now.saturating_sub(since) < park_after {
break;
}
self.park_queue.pop_front();
let Some(slot) = self.slots.get(&run) else {
continue;
};
if slot.suspended_since != Some(since) {
continue;
}
if self.parkable(run) {
to_park.push(run);
} else {
self.mark_active(run);
}
}
{
self.tick_parked = to_park.len() as u32;
for run in to_park {
if let Err(e) = self.park(run) {
self.fail(e.to_string());
}
}
}
self.park_over_budget();
self.core.shrink();
fn fit<K: std::hash::Hash + Eq, V>(m: &mut FxHashMap<K, V>) {
if m.capacity() > 1024 && m.capacity() > 4 * m.len() {
m.shrink_to(m.len() * 2);
}
}
self.slots.shrink();
self.drop_stream_logs();
fit(&mut self.resolved);
fit(&mut self.stash);
fit(&mut self.current_step);
if let Err(e) = self.expire_retained(now) {
self.fail(e.to_string());
}
}
if !self.ckpt_running && self.ckpt_pending.is_none() {
let rolled = self.log_segment() != self.last_ckpt_segment
&& self.checkpoint_needed()
&& self.env.checkpoint_on_roll();
let active = self.checkpoint_needed();
let due = now.saturating_sub(self.last_ckpt_ms) >= self.cfg.checkpoint_interval_ms;
let wanted = due && self.env.checkpoint_wanted();
if (active && (rolled || due)) || wanted {
self.start_checkpoint(now);
}
}
}
fn park_over_budget(&mut self) {
let budget = self.cfg.memory_budget;
if budget == 0 || self.cfg.park_after_ms == u64::MAX || self.loaded_bytes() <= budget {
return;
}
let target = budget - budget / 10;
let mut parked = 0;
while parked < BUDGET_PARKS && self.loaded_bytes() > target {
let Some((since, run)) = self.park_queue.pop_front() else {
break;
};
let Some(slot) = self.slots.get(&run) else {
continue;
};
if slot.suspended_since != Some(since) {
continue;
}
if !self.parkable(run) {
self.mark_active(run);
continue;
}
match self.park(run) {
Ok(true) => {
parked += 1;
self.metrics.budget_parks += 1;
}
Ok(false) => {}
Err(e) => {
self.fail(e.to_string());
return;
}
}
}
self.tick_parked += parked as u32;
}
fn expire_retained(&mut self, now: u64) -> Result<(), Error> {
let mut expired = Vec::new();
self.obsolete_store_bytes += self.core.expire(now, &mut expired);
for run in expired {
let blob = self
.tail
.put(self.env.store_len(), EntryRef::Tombstone { run })?;
self.obsolete_store_bytes += u64::from(blob.len);
if self.tail.full() {
self.store_flush()?;
}
}
self.store_flush()
}
fn should_compact(&self) -> bool {
!self.no_compaction
&& self.obsolete_store_bytes >= 8 << 20
&& self.obsolete_store_bytes >= self.store_len() / 2
}
fn checkpoint_needed(&self) -> bool {
self.writer.last_lsn() > self.last_ckpt_lsn
|| self.store_len() != self.last_ckpt_store_len
|| self.should_compact()
}
fn compact_store_if_needed(&mut self) -> Result<(), Error> {
let upgrade = !self.store_v3 && !self.no_compaction && self.env.recovers();
if !self.should_compact() && !upgrade {
return Ok(());
}
let chained: Vec<_> = self.core.chained_runs().collect();
for (run, blob) in chained {
let c = self.read_chain(blob)?;
let done = self.core.parked(run).and_then(|p| p.done());
let frame = match done.zip(fusee_core::store::journal_done(&c.journal)) {
Some((_, (status, finished_ms))) => encode_entry(&StoreEntry::Done {
run,
last_lsn: c.last_lsn,
id_hash: c.id_hash,
status,
finished_ms,
journal: if status == Status::Ok && !self.cfg.keep_history {
done_summary(&c.journal)?
} else {
c.journal
},
}),
None => encode_entry(&StoreEntry::Park {
run,
last_lsn: c.last_lsn,
id_hash: c.id_hash,
timers: c.timers,
journal: c.journal,
}),
};
let len = u32::try_from(frame.len())
.map_err(|_| Error::Io("run store frame exceeds 4 GiB".into()))?;
let offset = self.env.store_append(&frame)?;
self.obsolete_store_bytes += c.chain;
self.core.reparked(run, BlobRef { offset, len });
}
let loaded: Vec<_> = self
.core
.loaded_runs()
.filter(|(_, l)| l.blob_depth > 0)
.filter_map(|(r, l)| l.blob.map(|b| (r, b)))
.collect();
for (run, blob) in loaded {
let c = self.read_chain(blob)?;
let frame = encode_entry(&StoreEntry::Park {
run,
last_lsn: c.last_lsn,
id_hash: c.id_hash,
timers: c.timers,
journal: c.journal,
});
let len = u32::try_from(frame.len())
.map_err(|_| Error::Io("run store frame exceeds 4 GiB".into()))?;
let offset = self.env.store_append(&frame)?;
self.obsolete_store_bytes += c.chain;
self.core
.set_blob(run, BlobRef { offset, len }, u64::from(len), 0);
}
let mut live: Vec<_> = self
.core
.parked_runs()
.map(|(r, p)| (r, p.blob()))
.chain(
self.core
.loaded_runs()
.filter_map(|(r, l)| l.blob.map(|b| (r, b))),
)
.collect();
live.sort_unstable_by_key(|(_, b)| b.offset);
if self.env.compact_store(&live)? {
let mut offset = fusee_core::store::STORE_HEADER;
for (run, blob) in live {
self.core.relocate_blob(run, offset);
offset += u64::from(blob.len);
}
if offset != self.env.store_len() {
return Err(Error::Io("unexpected length of rewritten run store".into()));
}
self.store_v3 = self.store_takes_v3();
self.index = None;
self.obsolete_store_bytes = 0;
} else {
self.no_compaction = true;
}
Ok(())
}
fn store_takes_v3(&mut self) -> bool {
self.env
.store_read(0, fusee_core::store::STORE_HEADER as u32)
.ok()
.and_then(|h| fusee_core::store::store_format(&h).ok())
.is_some_and(|f| f >= 3)
}
fn write_index(&mut self) -> Result<Vec<BlobRef>, Error> {
let generation = self.env.store_generation();
let whole =
self.index.is_none() || self.index_gen != generation || self.core.index_needs_base();
const FRAME: usize = INDEX_FRAME_RECORDS * fusee_core::table::RECORD;
let mut records = Vec::new();
if whole {
self.drop_index_build();
if let Some(old) = self.index.take()
&& self.index_gen == generation
{
self.obsolete_store_bytes += old.iter().map(|b| u64::from(b.len)).sum::<u64>();
}
self.core.begin_index_snapshot();
let mut frames = Vec::new();
let mut from = Some(0);
while let Some(at) = from {
records.clear();
from = self
.core
.index_snapshot(at, INDEX_FRAME_RECORDS, &mut records);
if !records.is_empty() {
frames.push(self.append_index(&records)?);
}
}
self.index_base = (frames.len(), frames.iter().map(|b| u64::from(b.len)).sum());
self.index_delta_bytes = 0;
self.index_gen = generation;
self.index = Some(frames);
} else {
self.adopt_index_build();
self.core.index_delta(&mut records);
let mut frames = Vec::new();
for chunk in records.chunks(FRAME) {
frames.push(self.append_index(chunk)?);
}
self.index_delta_bytes += frames.iter().map(|b| u64::from(b.len)).sum::<u64>();
if let Some(index) = self.index.as_mut() {
index.extend(frames);
}
let deltas = self.index.as_ref().map_or(0, Vec::len) - self.index_base.0;
let chunks =
self.index_base.1 / (INDEX_CHUNK_RECORDS * fusee_core::table::RECORD) as u64;
if self.index_build.is_none()
&& (self.index_delta_bytes > self.index_base.1 / 2 + (1 << 20)
|| deltas as u64 >= (self.cfg.index_rebase_deltas as u64).max(8 * chunks))
{
self.index_build = Some(IndexBuild {
cursor: Some(0),
frames: Vec::new(),
bytes: 0,
generation,
deltas_from: self.index.as_ref().map_or(0, Vec::len),
});
self.core.hold_slots(true);
}
}
Ok(self.index.clone().unwrap_or_default())
}
fn advance_index_build(&mut self) {
let Some(mut build) = self.index_build.take() else {
return;
};
if build.generation != self.env.store_generation() {
self.index_build = Some(build);
self.drop_index_build();
return;
}
if let Err(e) = self.store_flush() {
self.index_build = Some(build);
self.fail(format!("run store write failed: {e}"));
return;
}
let started = self.env.monotonic_us();
let mut records = Vec::new();
let mut written = 0;
while let Some(at) = build.cursor {
records.clear();
build.cursor =
self.core
.index_snapshot(at, self.cfg.index_chunk_records.max(1), &mut records);
if !records.is_empty() {
match self.append_index(&records) {
Ok(frame) => {
build.bytes += u64::from(frame.len);
build.frames.push(frame);
}
Err(e) => {
self.index_build = Some(build);
self.drop_index_build();
self.fail(format!("run index write failed: {e}"));
return;
}
}
}
written += 1;
if written >= INDEX_BUILD_SLICE_FRAMES
|| self.env.monotonic_us().saturating_sub(started) >= INDEX_BUILD_SLICE_US
{
break;
}
}
if build.cursor.is_none() {
self.core.hold_slots(false);
}
self.index_build = Some(build);
}
fn adopt_index_build(&mut self) {
let Some(build) = self.index_build.take_if(|b| b.cursor.is_none()) else {
return;
};
let old = self.index.take().unwrap_or_default();
let from = build.deltas_from.min(old.len());
self.obsolete_store_bytes += old[..from].iter().map(|b| u64::from(b.len)).sum::<u64>();
let carried = &old[from..];
self.index_base = (build.frames.len(), build.bytes);
self.index_delta_bytes = carried.iter().map(|b| u64::from(b.len)).sum();
let mut frames = build.frames;
frames.extend_from_slice(carried);
self.metrics.index_bases_built += 1;
self.index = Some(frames);
}
fn drop_index_build(&mut self) {
if let Some(build) = self.index_build.take() {
if build.generation == self.env.store_generation() {
self.obsolete_store_bytes += build.bytes;
}
self.core.hold_slots(false);
}
}
fn append_index(&mut self, records: &[u8]) -> Result<BlobRef, Error> {
let frame = encode_entry(&StoreEntry::Index {
records: records.to_vec(),
});
let len = u32::try_from(frame.len())
.map_err(|_| Error::Io("run index frame exceeds 4 GiB".into()))?;
let offset = self.env.store_append(&frame)?;
Ok(BlobRef { offset, len })
}
fn start_checkpoint(&mut self, now: u64) {
let changed: Vec<_> = self
.core
.loaded_runs()
.filter(|(_, l)| !l.is_done() && !l.journal.is_empty())
.map(|(run, _)| run)
.collect();
for run in changed {
match self.store_active(run) {
Ok((blob, chain, depth)) => self.core.spilled(run, blob, chain, depth),
Err(e) => {
self.fail(format!("active journal checkpoint failed: {e}"));
return;
}
}
}
if let Err(e) = self.store_flush() {
self.fail(format!("active journal checkpoint failed: {e}"));
return;
}
if let Err(e) = self.compact_store_if_needed() {
self.fail(format!("run-store compaction failed: {e}"));
return;
}
let index = if self.store_v3 && self.env.recovers() {
match self.write_index() {
Ok(frames) => Some(frames),
Err(e) => {
self.fail(format!("run index checkpoint failed: {e}"));
return;
}
}
} else {
None
};
self.last_ckpt_store_len = self.env.store_len();
let lsn = self.writer.last_lsn();
let runs = self
.core
.loaded_runs()
.map(|(run, l)| CheckpointRun {
run,
last_lsn: l.last_lsn,
blob: l.blob,
journal: l.journal.clone(),
})
.collect();
let cp = Checkpoint {
log_position: self.env.log_position(),
shard: self.cfg.shard,
shards: self.cfg.shards,
lsn,
next_run: self.core.next_run_id(),
store_generation: self.env.store_generation(),
store_len: self.env.store_len(),
created_ms: now,
index,
obsolete: self.obsolete_store_bytes,
index_runs: self.core.parked_len() as u64,
runs,
outbox: {
let mut all: Vec<Outgoing> = self.outbox.values().cloned().collect();
let shards = self.cfg.shards;
for o in self.core.unreleased() {
let (target, msg) = match o {
Output::ChildStart {
parent,
wtype,
id,
flags,
deadline_ms,
input,
} => (
fusee_core::shard::shard_of(id, shards),
Message::ChildStart {
parent: *parent,
wtype: wtype.to_string(),
id: id.to_string(),
flags: *flags,
deadline_ms: *deadline_ms,
input: input.clone(),
trace: self.trace_of(parent.run),
},
),
Output::ChildResult {
parent,
status,
payload,
} => (
parent.shard,
Message::ChildResult {
parent: *parent,
status: *status,
payload: payload.clone(),
},
),
Output::Signal {
target,
name,
payload,
dedup,
} => (
fusee_core::shard::shard_of(target, shards),
Message::Signal {
target: target.to_string(),
name: name.to_string(),
payload: payload.clone(),
dedup: *dedup,
},
),
Output::Close {
target,
wtype,
parent,
terminate,
reason,
} => (
fusee_core::shard::shard_of(target, shards),
Message::Close {
target: target.to_string(),
wtype: wtype.to_string(),
parent: *parent,
terminate: *terminate,
reason: reason.clone(),
},
),
_ => continue,
};
all.push(Outgoing {
id: 0,
target,
msg,
sent_ms: 0,
});
}
encode_outbox(&all)
},
};
let seg = self.log_segment();
let delete = (seg > 1 && matches!(self.writer, ShardLog::Segments(_))).then(|| seg - 1);
self.last_ckpt_ms = now;
self.last_ckpt_segment = seg;
self.ckpt_pending = Some((cp, delete));
self.on_commit();
}
fn deliver(&mut self, run: RunId) -> Result<Option<Delivery>, Error> {
self.check_alive()?;
let Some(slot) = self.slots.get_mut(&run) else {
return Ok(None);
};
match slot.machine.next() {
Next::Deliver {
seq,
outcome,
record,
} => {
if let Some(r) = record {
if let Record::Step { status, .. } | Record::Completed { status, .. } = &r
&& let Some(slot) = self.slots.get_mut(&run)
&& slot.machine.owns_streams(seq)
{
for e in slot.machine.stream_ends(seq, *status) {
self.append(&e)?;
}
}
let lsn = self.append(&r)?;
if matches!(r, Record::ChildDone { .. })
&& let Some(froms) = self.child_acks.remove(&(run, seq))
{
for from in froms {
let t = self.msg_token(from);
self.core.add_output(lsn, Output::Ack(t));
}
}
if matches!(r, Record::Completed { .. })
&& let Some(tokens) = self.completion_acks.remove(&(run, seq))
{
for t in tokens {
self.core.add_output(lsn, Output::Ack(t));
}
}
} else {
self.metrics.steps_replayed += 1;
}
match self.open_outcome(outcome) {
Ok(outcome) => Ok(Some(Delivery::Resolve { seq, outcome })),
Err(e) => {
self.make_stuck(run, format!("a recorded payload cannot be read: {e}"));
Ok(None)
}
}
}
Next::DeliverLive {
seq,
outcome,
record,
} => {
if let Some(slot) = self.slots.get_mut(&run)
&& slot.machine.owns_streams(seq)
{
for e in slot.machine.stream_ends(seq, outcome.status) {
self.append(&e)?;
}
}
let time = self.step_time(run, seq);
let lsn = self.append_with(&record.record(run, seq, &outcome, time), None)?;
match record {
fusee_core::run::LiveRecord::Completed { .. } => {
if let Some(tokens) = self.completion_acks.remove(&(run, seq)) {
for t in tokens {
self.core.add_output(lsn, Output::Ack(t));
}
}
}
fusee_core::run::LiveRecord::Step { name, .. } => {
if let Some(slot) = self.slots.get_mut(&run) {
slot.machine.return_name(name);
}
}
}
match self.open_outcome(outcome) {
Ok(outcome) => Ok(Some(Delivery::Resolve { seq, outcome })),
Err(e) => {
self.make_stuck(run, format!("a recorded payload cannot be read: {e}"));
Ok(None)
}
}
}
Next::Update {
ordinal,
name,
id,
payload,
live,
} => match self.open(payload) {
Ok(payload) => {
if live.is_none() {
self.object_started(run, ordinal, &name);
}
Ok(Some(Delivery::Update {
ordinal,
name: name.to_string(),
id: id.to_string(),
payload,
token: live,
}))
}
Err(e) => {
if let Some(t) = live {
self.decide_update(t, Err(e.to_string()))?;
return Ok(None);
}
self.make_stuck(run, format!("a recorded update cannot be read: {e}"));
Ok(None)
}
},
Next::CancelAll {
seqs,
reason,
record,
} => {
if let Some(r) = record {
let mut ends = Vec::new();
if let Some(slot) = self.slots.get_mut(&run) {
for s in &seqs {
if slot.machine.owns_streams(*s) {
ends.extend(slot.machine.stream_ends(*s, Status::Cancelled));
}
}
}
for e in &ends {
self.append(e)?;
}
let lsn = self.append(&r)?;
if let Some(tokens) = self.cancel_tokens.remove(&run) {
for t in tokens {
self.core.add_output(lsn, Output::Ack(t));
}
}
}
Ok(Some(Delivery::CancelAll { seqs, reason }))
}
Next::Stalled => match slot.machine.stall_error() {
Some(e) => {
self.make_stuck(run, e.to_string());
Ok(None)
}
None => Ok(None),
},
Next::Idle => Ok(None),
}
}
fn report_pause(&mut self, run: RunId, seq: Seq, error: &[u8]) {
if self.cfg.replay_only {
return;
}
let Some(slot) = self.slots.get(&run) else {
return;
};
let info = slot.machine.info();
let step = slot.machine.pending_name(seq).unwrap_or_default();
let failure = crate::failure::Failure::from_payload(error);
let line = format!(
"workflow '{}' ({} v{}, run {run}) paused: step '{step}' ran out of retries: {}. \
Fix the cause, then resume it: fusee runs resume {}",
info.id, info.wtype, info.version, failure.message, info.id
);
crate::log::emit(&crate::log::Record {
workflow_id: Some(&info.id),
workflow_type: Some(&info.wtype),
run: Some(run),
..crate::log::Record::engine(crate::log::Level::Warn, &line)
});
}
fn make_stuck(&mut self, run: RunId, message: String) {
self.metrics.determinism_errors += 1;
if !self.cfg.replay_only {
let info = self.slots.get(&run).map(|s| s.machine.info());
let who = info
.map(|i| {
format!(
"workflow '{}' ({} v{}, run {run})",
i.id, i.wtype, i.version
)
})
.unwrap_or_else(|| format!("run {run}"));
let line = format!(
"{who} is stuck and will not progress: {message}. Fix: deploy code that matches \
its journal, guard the change with patched(), or reset the run to before that \
step. See {}",
crate::error::docs_link("Determinism")
);
crate::log::emit(&crate::log::Record {
workflow_id: info.map(|i| &*i.id),
workflow_type: info.map(|i| &*i.wtype),
run: Some(run),
..crate::log::Record::engine(crate::log::Level::Error, &line)
});
}
self.release_waiting_requests(run);
if let Some(slot) = self.remove_slot(run)
&& slot.foreign
{
self.foreign.push_back(ForeignEvent::Stuck {
run,
message: message.clone(),
});
}
self.stuck.insert(run, message);
}
fn store_delivery(&mut self, run: RunId, d: Delivery) {
match d {
Delivery::Update { .. } => {}
Delivery::Resolve { seq, outcome } => self.resolve(run, seq, outcome),
Delivery::CancelAll { seqs, reason } => {
for s in seqs {
self.resolve(
run,
s,
Outcome {
status: Status::Cancelled,
payload: reason.clone(),
},
);
}
}
}
}
fn resolve(&mut self, run: RunId, seq: Seq, outcome: Outcome) {
if self.resolved.is_empty()
&& let Some(slot) = self.slots.get_mut(&run)
&& slot.resolved.is_none()
{
slot.resolved = Some((seq, outcome));
return;
}
if let Some(Resolution::Waiting(w)) =
self.resolved.insert((run, seq), Resolution::Ready(outcome))
{
w.wake();
}
}
fn cancel_timer(&mut self, id: u64) {
if self.timer_wakers.remove(&id).is_none() {
return;
}
while let Some(Reverse((_, top))) = self.timers.peek()
&& !self.timer_wakers.contains_key(top)
{
self.timers.pop();
}
if self.timers.len() > 64 && self.timers.len() > 2 * self.timer_wakers.len() {
self.timers
.retain(|Reverse((_, id))| self.timer_wakers.contains_key(id));
}
}
fn release_waiting_requests(&mut self, run: RunId) {
self.release_child_acks(run);
self.object_ended(run, "the object's run ended before the call started");
let tail = self.writer.last_lsn();
if let Some(tokens) = self.cancel_tokens.remove(&run) {
for t in tokens {
self.core.add_output(tail, Output::Ack(t));
}
}
let keys: Vec<(RunId, Name)> = if self.update_waiters.is_empty() {
Vec::new()
} else {
self.update_waiters
.keys()
.filter(|k| k.0 == run)
.cloned()
.collect()
};
if !keys.is_empty() {
let ended: Payload = Arc::from(&b"the workflow ended before the update finished"[..]);
for k in keys {
for t in self.update_waiters.remove(&k).unwrap_or_default() {
self.core.add_output(
tail,
Output::Reply {
token: t,
status: Status::Failed,
payload: ended.clone(),
},
);
}
}
}
if !self.offered.is_empty() {
self.offered.retain(|_, v| v.0 != run);
}
if let Some(slot) = self.slots.get_mut(&run) {
slot.machine.take_offered_updates();
}
if self.completion_acks.is_empty() {
return;
}
let keys: Vec<(RunId, Seq)> = self
.completion_acks
.keys()
.filter(|k| k.0 == run)
.copied()
.collect();
for k in keys {
for t in self.completion_acks.remove(&k).unwrap_or_default() {
self.core.add_output(
tail,
Output::Refuse {
token: t,
message: Arc::from(&b"the step is no longer waiting for a completion"[..]),
},
);
}
}
}
fn start_closed(
&mut self,
id: &str,
wtype: &str,
parent: ParentRef,
status: Status,
reason: &[u8],
) -> Result<Lsn, Error> {
let run = self.core.allocate_run();
let now = self.now();
let version = self
.registry
.read()
.ok()
.and_then(|r| r.latest(wtype))
.unwrap_or(1);
let start: RecordBuf = Record::Start {
run,
id: Arc::from(id),
wtype: Arc::from(wtype),
version,
codec: 1,
started_ms: now,
deadline_ms: 0,
parent: Some(parent),
policy: policy::ATTACH | policy::DETACHED,
from: None,
copies: 0,
update: None,
input: empty(),
meta: None,
};
self.append(&start)?;
self.metrics.runs_started += 1;
self.append(&Record::Done {
run,
status,
finished_ms: now,
payload: Arc::from(reason),
by: None,
})
}
fn fresh_update_id(&mut self) -> String {
let n = self.next_update_id;
self.next_update_id += 1;
format!("auto-{}-{n}", self.now())
}
fn offer_update(
&mut self,
run: RunId,
name: &str,
id: &str,
payload: Payload,
token: Option<u64>,
) -> Result<(), Error> {
if self.object_answered(run, id, token) {
return Ok(());
}
let Some(l) = self.core.get(run) else {
return Err(err_slot(run));
};
let key: (RunId, Name) = (run, Name::from(id));
let known = if self.cfg.sabotage == Sabotage::DuplicateUpdates {
None
} else {
l.update(id)
};
let object = l.policy & policy::OBJECT != 0 && !l.is_done();
if let Some((_, slot)) = known {
let result = slot.result.clone();
match (result, token) {
(Some((status, p, lsn)), Some(t)) => {
let (status, payload) = match (status, self.open(p.clone())) {
(Status::Ok, Ok(p)) => (Status::Ok, p),
(Status::Ok, Err(e)) => {
(Status::Failed, Arc::from(e.to_string().as_bytes()))
}
(s, _) => (s, p),
};
self.core.add_output(
lsn,
Output::Reply {
token: t,
status,
payload,
},
);
}
(None, Some(t)) if l.is_done() => {
let at = l.last_lsn.max(self.writer.last_lsn());
self.core.add_output(
at,
Output::Reply {
token: t,
status: Status::Failed,
payload: Arc::from(
&b"the workflow ended before the update finished"[..],
),
},
);
}
(None, Some(t)) => self.update_waiters.entry(key).or_default().push(t),
_ => {}
}
return Ok(());
}
if l.is_done() {
if let Some(t) = token {
let at = l.last_lsn.max(self.writer.last_lsn());
self.core.add_output(
at,
Output::Refuse {
token: t,
message: Arc::from(&b"the workflow has already finished"[..]),
},
);
}
return Ok(());
}
let in_flight = self
.offered
.values()
.any(|(r, _, i, _)| *r == run && **i == *id);
if let Some(t) = token {
self.update_waiters.entry(key).or_default().push(t);
}
if in_flight {
return Ok(());
}
if object && self.object_hold(run, name, id, &payload) {
return Ok(());
}
self.offer_now(run, name, id, payload)
}
fn offer_now(
&mut self,
run: RunId,
name: &str,
id: &str,
payload: Payload,
) -> Result<(), Error> {
if !self.slots.contains_key(&run) {
const NO_CODE: &str =
"the workflow is not running code (it is stuck or its code is not registered)";
let tail = self.writer.last_lsn();
for t in self
.update_waiters
.remove(&(run, Name::from(id)))
.unwrap_or_default()
{
self.core.add_output(
tail,
Output::Refuse {
token: t,
message: Arc::from(NO_CODE.as_bytes()),
},
);
}
if let Some(o) = self.obj.runs.get_mut(&run)
&& o.offered.as_deref() == Some(id)
{
o.offered = None;
}
self.object_fail_caller(run, id, NO_CODE);
return Ok(());
}
let sealed = self.seal(payload)?;
let internal = self.next_update_id;
self.next_update_id += 1;
let (name, id): (Name, Name) = (Name::from(name), Name::from(id));
self.offered
.insert(internal, (run, name.clone(), id.clone(), sealed.clone()));
if let Some(slot) = self.slots.get_mut(&run) {
slot.machine.offer_update(internal, name, id, sealed);
}
self.wake(run);
Ok(())
}
pub fn decide_update(
&mut self,
token: u64,
decision: Result<(), String>,
) -> Result<Option<u64>, Error> {
let Some((run, name, id, payload)) = self.offered.remove(&token) else {
return Ok(None);
};
let object = self.obj.runs.get_mut(&run).map(|o| {
if o.offered.as_ref() == Some(&id) {
o.offered = None;
}
});
match decision {
Ok(()) => {
let Some(slot) = self.slots.get_mut(&run) else {
return Ok(None);
};
let (ordinal, rec) = slot
.machine
.accept_update(name.clone(), id.clone(), payload);
if object.is_some() {
self.object_accept(run, ordinal, &name, &id, rec)?;
} else {
self.append(&rec)?;
}
Ok(Some(ordinal))
}
Err(message) => {
let at = self
.core
.get(run)
.map(|l| l.last_lsn)
.unwrap_or(0)
.max(self.commit);
let text = message;
let message: Payload = Arc::from(text.as_bytes());
for t in self
.update_waiters
.remove(&(run, id.clone()))
.unwrap_or_default()
{
self.core.add_output(
at,
Output::Refuse {
token: t,
message: message.clone(),
},
);
}
if object.is_some() {
self.object_fail_caller(run, &id, &text);
self.object_settle(run);
}
self.on_commit();
Ok(None)
}
}
}
fn update_done(&mut self, run: RunId, ordinal: u64, outcome: RunOutcome) -> Result<(), Error> {
let (status, plain) = match outcome {
Ok(p) => (Status::Ok, p),
Err((s, p)) => (s, p),
};
let sealed = if status == Status::Ok {
self.seal(plain.clone())?
} else {
plain.clone()
};
let object = self.obj.runs.contains_key(&run);
let state = self.object_commit(run, ordinal, status)?;
let slot = self.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let r = slot.machine.update_done(ordinal, status, sealed);
let (_, mut rec) = stop_on(self, run, r)?;
if let Some(Record::UpdateDone { state: s, .. }) = &mut rec {
*s = state;
}
let lsn = match rec {
Some(rec) => self.append(&rec)?,
None => self.writer.last_lsn(),
};
let id = self
.core
.get(run)
.and_then(|l| l.updates.get(ordinal as usize))
.map(|u| u.id.clone());
if let Some(id) = id {
for t in self.update_waiters.remove(&(run, id)).unwrap_or_default() {
self.core.add_output(
lsn,
Output::Reply {
token: t,
status,
payload: plain.clone(),
},
);
}
}
if object {
self.object_settle(run);
}
Ok(())
}
fn continue_as_new(
&mut self,
run: RunId,
input: Payload,
version: Option<u32>,
) -> Result<RunId, Error> {
let slot = self.slots.get(&run).ok_or_else(|| err_slot(run))?;
let r = slot.machine.check_end();
stop_on(self, run, r)?;
let l = self.core.get(run).ok_or_else(|| err_slot(run))?;
let (id, wtype, parent, pol, deadline, old_version) = (
l.id.clone(),
l.wtype.clone(),
l.parent,
l.policy,
l.deadline_ms,
l.version,
);
let version = version
.or_else(|| self.registry.read().ok()?.latest(&wtype))
.unwrap_or(old_version);
let input = self.seal(input)?;
let new = self.core.allocate_run();
let now = self.now();
let start: RecordBuf = Record::Start {
run: new,
id,
wtype: wtype.clone(),
version,
codec: 1,
started_ms: now,
deadline_ms: deadline,
parent,
policy: pol,
from: Some(Origin {
run,
cause: origin::CONTINUED,
}),
copies: 0,
update: None,
input,
meta: StartMeta::of(None, self.trace_of(run)),
};
let mut payload = Vec::new();
encode_canonical(&start, &mut payload);
let done: RecordBuf = Record::Done {
run,
status: Status::ContinuedAsNew,
finished_ms: now,
payload: Arc::from(payload),
by: None,
};
let lsn = self.append(&done)?;
if let Some(slot) = self.remove_slot(run)
&& slot.foreign
{
self.foreign.push_back(ForeignEvent::Ended { run });
}
if let Some(subs) = self.subscribers.remove(&run) {
self.subscribers.insert(new, subs);
}
self.release_waiting_requests(run);
self.metrics.runs_started += 1;
let foreign = self.runs_foreign(&wtype, version);
self.new_slot(
new,
JournalIn::Borrowed(std::slice::from_ref(&start)),
lsn,
foreign,
)?;
self.instantiate.push(new);
Ok(new)
}
fn release_child_acks(&mut self, run: RunId) {
if self.child_acks.is_empty() {
return;
}
let keys: Vec<(RunId, Seq)> = self
.child_acks
.keys()
.filter(|k| k.0 == run)
.copied()
.collect();
for k in keys {
for from in self.child_acks.remove(&k).unwrap_or_default() {
self.ack_at_tail(from);
}
}
}
fn finish(&mut self, run: RunId, outcome: RunOutcome) {
let (status, payload) = match outcome {
Ok(p) => match self.seal(p) {
Ok(p) => (Status::Ok, p),
Err(e) => (Status::Failed, Arc::from(e.to_string().as_bytes())),
},
Err((s, p)) => (s, p),
};
let now = self.now();
let Some(slot) = self.slots.get_mut(&run) else {
return;
};
match slot.machine.finish(status, payload, now) {
Ok(rec) => {
if let Err(e) = self.append(&rec) {
self.fail(e.to_string());
}
}
Err(e) => self.make_stuck(run, e.to_string()),
}
if !self.resolved.is_empty() {
self.resolved.retain(|(r, _), _| *r != run);
}
self.release_waiting_requests(run);
}
}
pub trait Host {
fn begin_step(&self, run: RunId, name: &str) -> Result<(Seq, StepStart, Lsn), Error>;
fn may_run(&self, run: RunId, gate: Lsn, not_before_ms: u64, waker: &Waker) -> bool;
fn set_current_step(&self, run: RunId, seq: Option<Seq>);
fn current_step(&self, run: RunId) -> Option<Seq>;
fn step_done(&self, run: RunId, seq: Seq, status: Status, payload: Payload);
fn step_done_inline(&self, _run: RunId, _seq: Seq, _payload: &[u8]) -> bool {
false
}
fn hold_inline(&self) -> u32 {
0
}
fn resume_inline(&self, _held: u32) {}
fn step_retry(
&self,
run: RunId,
seq: Seq,
retry_at_ms: u64,
error: Payload,
) -> Result<Lsn, Error>;
fn step_attempt(&self, run: RunId, seq: Seq) -> u32;
fn begin_sleep(&self, run: RunId, name: &str, fire_at_ms: u64) -> Result<Seq, Error>;
fn begin_wait(&self, run: RunId, name: &str, deadline_ms: u64) -> Result<Seq, Error>;
fn begin_child(
&self,
run: RunId,
wtype: &str,
id: &str,
flags: u8,
deadline_ms: u64,
input: Payload,
) -> Result<Option<Seq>, Error>;
fn value(
&self,
run: RunId,
kind: u8,
make: &mut dyn FnMut(u64, u64) -> Payload,
) -> Result<Payload, Error>;
fn send(&self, run: RunId, target: &str, name: &str, payload: Payload) -> Result<(), Error>;
fn take_resolution(&self, run: RunId, seq: Seq, waker: &Waker) -> Option<Outcome>;
fn forget(&self, run: RunId, seq: Seq);
fn now_ms(&self) -> u64;
fn step_paused(&self, _run: RunId, _seq: Seq) -> bool {
false
}
fn step_pending(&self, _run: RunId, _seq: Seq) -> bool {
true
}
fn on_exhausted(&self) -> crate::ctx::Exhausted {
crate::ctx::Exhausted::Fail
}
fn key(&self, run: RunId, seq: Seq) -> String;
fn info(&self, run: RunId) -> Option<RunInfo>;
fn replaying(&self, _run: RunId) -> bool {
false
}
fn op_name(&self, _run: RunId, _seq: Seq) -> Option<String> {
None
}
fn wake_at(&self, at_ms: u64, waker: &Waker) -> u64;
fn cancel_timer(&self, id: u64);
fn broadcasts(&self, run: RunId) -> u64;
fn wake_run_at(&self, run: RunId, at_ms: u64);
fn begin_await(&self, run: RunId, name: &str) -> Result<(Seq, AwaitStart, Lsn), Error>;
fn completion_token(&self, run: RunId, seq: Seq) -> Result<String, Error>;
fn awaiting(&self, run: RunId, seq: Seq) -> Result<(), Error>;
fn heartbeat(&self, run: RunId, seq: Seq, progress: Payload) -> Result<(), Error>;
fn progress(&self, run: RunId, seq: Seq) -> Option<Payload>;
fn patched(&self, run: RunId, id: &str, deprecated: bool) -> Result<bool, Error>;
fn set_attributes(&self, run: RunId, attributes: Payload) -> Result<(), Error>;
fn object_get(&self, run: RunId, key: &str, exclusive: bool) -> Result<Option<Payload>, Error>;
fn object_set(&self, run: RunId, key: &str, value: Option<Payload>) -> Result<(), Error>;
fn object_keys(&self, run: RunId, exclusive: bool) -> Result<Vec<String>, Error>;
fn object_state(&self, run: RunId) -> Result<Payload, Error>;
fn object_edit(&self, run: RunId, edit: &[u8]) -> Result<Payload, Error>;
fn begin_call(
&self,
run: RunId,
wtype: &str,
key: &str,
handler: &str,
arg: Payload,
detached: bool,
) -> Result<Option<Seq>, Error>;
fn update_done(&self, run: RunId, ordinal: u64, outcome: RunOutcome) -> Result<(), Error>;
fn continue_as_new(
&self,
run: RunId,
input: Payload,
version: Option<u32>,
) -> Result<(), Error>;
fn journal_size(&self, run: RunId) -> u32;
fn cancel_workflow(&self, run: RunId, target: &str, reason: &str) -> Result<(), Error>;
fn limit_ticket(&self) -> u64;
fn acquire_in(
&self,
run: RunId,
limit: &str,
ticket: u64,
priority: &Priority,
waker: Option<&Waker>,
) -> Acquire;
fn acquire(&self, run: RunId, limit: &str, ticket: u64, waker: Option<&Waker>) -> Acquire {
self.acquire_in(run, limit, ticket, &Priority::default(), waker)
}
fn release(&self, limit: &str, ticket: u64);
fn stream_open(&self, run: RunId, name: &str, step: Option<Seq>) -> Result<StreamOpen, Error>;
fn stream_append(
&self,
run: RunId,
name: &str,
step: Option<Seq>,
payload: Payload,
) -> Result<(), Error>;
fn stream_close(&self, run: RunId, name: &str, step: Option<Seq>) -> Result<(), Error>;
fn stream_recorded(
&self,
run: RunId,
name: &str,
step: Seq,
limit: u64,
) -> Result<Vec<Payload>, Error>;
}
pub fn format_token(id: &str, run: RunId, seq: Seq, attempt: u32) -> String {
let hex: String = id.bytes().map(|b| format!("{b:02x}")).collect();
format!("fusee:{run}:{seq}:{attempt}:{hex}")
}
pub fn parse_token(token: &str) -> Option<(String, RunId, Seq, u32)> {
let mut parts = token.strip_prefix("fusee:")?.splitn(4, ':');
let run = parts.next()?.parse().ok()?;
let seq = parts.next()?.parse().ok()?;
let attempt = parts.next()?.parse().ok()?;
let hex = parts.next()?;
if hex.len() % 2 != 0 {
return None;
}
let bytes: Option<Vec<u8>> = (0..hex.len())
.step_by(2)
.map(|i| u8::from_str_radix(hex.get(i..i + 2)?, 16).ok())
.collect();
Some((String::from_utf8(bytes?).ok()?, run, seq, attempt))
}
fn journal_paused(recs: &[RecordBuf]) -> bool {
let mut paused: Vec<Seq> = Vec::new();
for r in recs {
match r {
Record::Attempt {
seq, retry_at_ms, ..
} => {
paused.retain(|s| s != seq);
if *retry_at_ms == PAUSED {
paused.push(*seq);
}
}
Record::Resume { seq, .. } | Record::Step { seq, .. } => paused.retain(|s| s != seq),
Record::Cancel { .. } => paused.clear(),
_ => {}
}
}
!paused.is_empty()
}
pub fn merge_attributes(objects: &[Payload]) -> String {
let mut merged = serde_json::Map::new();
for o in objects {
if let Ok(serde_json::Value::Object(m)) = serde_json::from_slice::<serde_json::Value>(o) {
for (k, v) in m {
if v.is_null() {
merged.remove(&k);
} else {
merged.insert(k, v);
}
}
}
}
serde_json::Value::Object(merged).to_string()
}
fn err_slot(run: RunId) -> Error {
Error::Rejected(format!("run {run} is not active on this shard"))
}
pub const KEY_OUTSIDE_STEP: &str = "key() was called outside a step's function, where there is no \
step to key. Call it inside the step's function (in TypeScript and Python, use the step context's \
s.key). See https://github.com/bitsinfra/fusee/blob/main/docs/errors.md#key-outside-a-step";
fn stop_on<E: Env, T>(
i: &mut Inner<E>,
run: RunId,
r: Result<T, fusee_core::run::DeterminismError>,
) -> Result<T, Error> {
r.map_err(|e| {
let msg = e.to_string();
i.make_stuck(run, msg.clone());
Error::Determinism(msg)
})
}
impl<E: Env> Host for RefCell<Inner<E>> {
fn begin_step(&self, run: RunId, name: &str) -> Result<(Seq, StepStart, Lsn), Error> {
let mut i = self.borrow_mut();
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let gate = slot.machine.gate;
let r = slot.machine.begin_step(name);
let (seq, start) = stop_on(&mut i, run, r)?;
if i.cfg.step_times && matches!(start, StepStart::Run { .. }) {
let now = i.now();
i.step_began.entry((run, seq)).or_insert(now);
}
Ok((seq, start, gate))
}
fn may_run(&self, run: RunId, gate: Lsn, not_before_ms: u64, waker: &Waker) -> bool {
let mut i = self.borrow_mut();
if i.cfg.replay_only || i.failed.is_some() {
return false;
}
let mut durable = i.commit >= gate || i.cfg.sabotage == Sabotage::SkipGate;
let time = not_before_ms == 0 || i.now() >= not_before_ms || i.now_exact() >= not_before_ms;
if !durable && time && i.flush_at_gate() {
durable = i.commit >= gate;
}
if durable && time {
return true;
}
if !durable {
i.metrics.gate_waits += 1;
let began = if i.clock.on {
i.round_us
} else {
i.metric_us()
};
let register = match i.slots.get_mut(&run) {
Some(slot) if slot.gate_at.is_some_and(|g| g <= gate) => false,
Some(slot) => {
slot.gate_at = Some(gate);
true
}
None => true,
};
if register {
i.gates.push((gate, run, began));
}
}
if !time && !waker.will_wake(Waker::noop()) {
let id = i.timer_seq;
i.timer_seq += 1;
i.timers.push(Reverse((not_before_ms, id)));
i.timer_wakers.insert(id, waker.clone());
}
false
}
fn set_current_step(&self, run: RunId, seq: Option<Seq>) {
let mut i = self.borrow_mut();
match seq {
Some(s) => {
i.current_step.insert(run, s);
}
None => {
i.current_step.remove(&run);
}
}
}
fn current_step(&self, run: RunId) -> Option<Seq> {
self.borrow().current_step.get(&run).copied()
}
fn step_done_inline(&self, run: RunId, seq: Seq, payload: &[u8]) -> bool {
let mut i = self.borrow_mut();
if i.codec.is_some() || i.inline_deliveries == 0 {
return false;
}
let Some(slot) = i.slots.get_mut(&run) else {
return false;
};
if slot.foreign || slot.resolved.is_some() {
return false;
}
let Some(record) = slot.machine.take_live_step(seq) else {
return false;
};
let fusee_core::run::LiveRecord::Step { name, attempt } = record else {
return false;
};
i.inline_deliveries -= 1;
i.clock.ran.set(true);
i.metrics.steps_run += 1;
#[cfg(feature = "opentelemetry")]
if let Some(telemetry) = &i.cfg.telemetry {
telemetry.step(i.cfg.shard);
}
if !i.heartbeats.is_empty() {
i.heartbeats.remove(&(run, seq));
}
let time = i.step_time(run, seq);
let rec: fusee_core::record::RecordRef<'_> = Record::Step {
run,
seq,
name: &name,
attempt,
status: Status::Ok,
payload,
time,
};
let appended = i.append_with(&rec, None);
if let Some(slot) = i.slots.get_mut(&run) {
slot.machine.return_name(name);
}
if let Err(e) = appended {
i.fail(e.to_string());
}
true
}
fn hold_inline(&self) -> u32 {
std::mem::take(&mut self.borrow_mut().inline_deliveries)
}
fn resume_inline(&self, held: u32) {
self.borrow_mut().inline_deliveries = held;
}
fn step_done(&self, run: RunId, seq: Seq, status: Status, payload: Payload) {
let mut i = self.borrow_mut();
i.clock.ran.set(true);
i.metrics.steps_run += 1;
#[cfg(feature = "opentelemetry")]
if let Some(telemetry) = &i.cfg.telemetry {
telemetry.step(i.cfg.shard);
}
if !i.heartbeats.is_empty() {
i.heartbeats.remove(&(run, seq));
}
let (status, payload) = if status == Status::Ok {
match i.seal(payload) {
Ok(p) => (status, p),
Err(e) => (Status::Failed, Arc::from(e.to_string().as_bytes())),
}
} else {
(status, payload)
};
if i.inline_deliveries > 0
&& let Some(slot) = i.slots.get_mut(&run)
&& !slot.foreign
&& slot.resolved.is_none()
&& let Some(record) = slot.machine.take_live_step(seq)
{
i.inline_deliveries -= 1;
let outcome = Outcome { status, payload };
let time = i.step_time(run, seq);
match i.append_with(&record.record(run, seq, &outcome, time), None) {
Ok(_) => match i.open_outcome(outcome) {
Ok(o) => i.resolve(run, seq, o),
Err(e) => {
i.make_stuck(run, format!("a recorded payload cannot be read: {e}"));
}
},
Err(e) => i.fail(e.to_string()),
}
return;
}
if let Some(slot) = i.slots.get_mut(&run)
&& slot
.machine
.complete(seq, Completion::Step { status, payload })
{
i.note_undelivered(run);
if i.inline_deliveries > 0
&& let Some(slot) = i.slots.get(&run)
&& !slot.foreign
&& slot.resolved.is_none()
&& slot.machine.delivers_only(seq)
{
i.inline_deliveries -= 1;
match i.deliver(run) {
Ok(Some(d @ Delivery::Resolve { .. })) => i.store_delivery(run, d),
Ok(_) => {}
Err(e) => i.fail(e.to_string()),
}
}
}
}
fn step_retry(
&self,
run: RunId,
seq: Seq,
retry_at_ms: u64,
error: Payload,
) -> Result<Lsn, Error> {
let mut i = self.borrow_mut();
i.clock.ran.set(true);
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let paused = (retry_at_ms == PAUSED).then(|| error.clone());
match slot.machine.step_retry(seq, retry_at_ms, error) {
Some(rec) => {
let lsn = i.append(&rec)?;
if let Some(error) = paused {
i.report_pause(run, seq, &error);
}
Ok(lsn)
}
None => Ok(i.commit),
}
}
fn step_paused(&self, run: RunId, seq: Seq) -> bool {
self.borrow()
.slots
.get(&run)
.is_some_and(|s| s.machine.step_paused(seq))
}
fn step_pending(&self, run: RunId, seq: Seq) -> bool {
self.borrow()
.slots
.get(&run)
.is_some_and(|s| s.machine.step_pending(seq))
}
fn on_exhausted(&self) -> crate::ctx::Exhausted {
self.borrow().cfg.on_exhausted
}
fn step_attempt(&self, run: RunId, seq: Seq) -> u32 {
self.borrow()
.slots
.get(&run)
.and_then(|s| s.machine.step_attempt(seq))
.unwrap_or(1)
}
fn begin_sleep(&self, run: RunId, name: &str, fire_at_ms: u64) -> Result<Seq, Error> {
let mut i = self.borrow_mut();
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let r = slot.machine.begin_sleep(name, fire_at_ms);
let (seq, arm) = stop_on(&mut i, run, r)?;
if let Arm::Append(rec) = arm {
i.append(&rec)?;
}
Ok(seq)
}
fn begin_wait(&self, run: RunId, name: &str, deadline_ms: u64) -> Result<Seq, Error> {
let mut i = self.borrow_mut();
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let r = slot.machine.begin_wait(name, deadline_ms);
let (seq, arm) = stop_on(&mut i, run, r)?;
if let Arm::Append(rec) = arm {
i.append(&rec)?;
}
Ok(seq)
}
fn begin_child(
&self,
run: RunId,
wtype: &str,
id: &str,
flags: u8,
deadline_ms: u64,
input: Payload,
) -> Result<Option<Seq>, Error> {
let detached = flags & policy::DETACHED != 0;
let mut i = self.borrow_mut();
let input = i.seal(input)?;
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let r = slot
.machine
.begin_child(wtype, id, flags, deadline_ms, input);
let (seq, arm) = stop_on(&mut i, run, r)?;
if let Arm::Append(rec) = arm {
i.append(&rec)?;
}
Ok((!detached).then_some(seq))
}
fn value(
&self,
run: RunId,
kind: u8,
make: &mut dyn FnMut(u64, u64) -> Payload,
) -> Result<Payload, Error> {
let mut i = self.borrow_mut();
let now = i.now_exact();
let rnd = i.env.random();
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let r = slot.machine.value(kind, || make(now, rnd));
let (_, v, rec) = stop_on(&mut i, run, r)?;
if let Some(rec) = rec {
i.append(&rec)?;
}
Ok(v)
}
fn send(&self, run: RunId, target: &str, name: &str, payload: Payload) -> Result<(), Error> {
let mut i = self.borrow_mut();
let payload = i.seal(payload)?;
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let r = slot.machine.send(target, name, payload);
let (_, rec) = stop_on(&mut i, run, r)?;
if let Some(rec) = rec {
i.append(&rec)?;
}
Ok(())
}
fn take_resolution(&self, run: RunId, seq: Seq, waker: &Waker) -> Option<Outcome> {
use std::collections::hash_map::Entry;
let mut guard = self.borrow_mut();
let i = &mut *guard;
let own = match i.slots.get_mut(&run) {
Some(slot) => {
if slot.resolved.as_ref().is_some_and(|(s, _)| *s == seq) {
return slot.resolved.take().map(|(_, o)| o);
}
std::ptr::eq(waker.data(), Arc::as_ptr(&slot.waker).cast())
}
None => false,
};
if own && i.resolved.is_empty() {
return None;
}
match i.resolved.entry((run, seq)) {
Entry::Occupied(mut e) => match e.get_mut() {
Resolution::Ready(_) => match e.remove() {
Resolution::Ready(o) => Some(o),
Resolution::Waiting(_) => None,
},
Resolution::Waiting(w) => {
if !w.will_wake(waker) {
*w = waker.clone();
}
None
}
},
Entry::Vacant(e) => {
if !own {
e.insert(Resolution::Waiting(waker.clone()));
}
None
}
}
}
fn forget(&self, run: RunId, seq: Seq) {
if let Ok(mut i) = self.try_borrow_mut() {
if i.reaping.contains(&run) {
return;
}
let abandon = i
.slots
.get_mut(&run)
.and_then(|slot| slot.machine.forget(seq));
if let Some(rec) = abandon
&& let Err(e) = i.append(&rec)
{
i.fail(e.to_string());
}
if let Some(slot) = i.slots.get_mut(&run)
&& slot.resolved.as_ref().is_some_and(|(s, _)| *s == seq)
{
slot.resolved = None;
}
if !i.resolved.is_empty() {
i.resolved.remove(&(run, seq));
}
i.heartbeats.remove(&(run, seq));
if let Some(tokens) = i.completion_acks.remove(&(run, seq)) {
let tail = i.writer.last_lsn();
for t in tokens {
i.core.add_output(
tail,
Output::Refuse {
token: t,
message: Arc::from(&b"the step is no longer waiting"[..]),
},
);
}
}
}
}
fn now_ms(&self) -> u64 {
self.borrow().now_exact()
}
fn replaying(&self, run: RunId) -> bool {
self.borrow()
.slots
.get(&run)
.is_some_and(|s| s.machine.is_replaying())
}
fn op_name(&self, run: RunId, seq: Seq) -> Option<String> {
self.borrow()
.slots
.get(&run)
.and_then(|s| s.machine.pending_name(seq).map(str::to_owned))
}
fn key(&self, run: RunId, seq: Seq) -> String {
if seq == Seq::MAX {
let mut i = self.borrow_mut();
if i.slots.contains_key(&run) {
i.make_stuck(run, KEY_OUTSIDE_STEP.to_owned());
}
return String::new();
}
self.borrow()
.slots
.get(&run)
.map(|s| s.machine.key(seq))
.unwrap_or_default()
}
fn info(&self, run: RunId) -> Option<RunInfo> {
self.borrow()
.slots
.get(&run)
.map(|s| s.machine.info().clone())
}
fn wake_at(&self, at_ms: u64, waker: &Waker) -> u64 {
let mut i = self.borrow_mut();
let id = i.timer_seq;
i.timer_seq += 1;
i.timers.push(Reverse((at_ms, id)));
i.timer_wakers.insert(id, waker.clone());
id
}
fn cancel_timer(&self, id: u64) {
if let Ok(mut i) = self.try_borrow_mut() {
i.cancel_timer(id);
}
}
fn broadcasts(&self, run: RunId) -> u64 {
self.borrow().slots.get(&run).map_or(0, |s| s.broadcasts)
}
fn wake_run_at(&self, run: RunId, at_ms: u64) {
let w = self
.borrow()
.slots
.get(&run)
.map(|s| Waker::from(s.waker.clone()));
if let Some(w) = w {
self.wake_at(at_ms, &w);
}
}
fn begin_await(&self, run: RunId, name: &str) -> Result<(Seq, AwaitStart, Lsn), Error> {
let mut i = self.borrow_mut();
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let gate = slot.machine.gate;
let r = slot.machine.begin_await(name);
let (seq, start) = stop_on(&mut i, run, r)?;
Ok((seq, start, gate))
}
fn completion_token(&self, run: RunId, seq: Seq) -> Result<String, Error> {
let i = self.borrow();
let slot = i.slots.get(&run).ok_or_else(|| err_slot(run))?;
let attempt = slot
.machine
.await_attempt(seq)
.ok_or_else(|| Error::Rejected("the step is not waiting for a completion".into()))?;
Ok(format_token(&slot.machine.info().id, run, seq, attempt))
}
fn awaiting(&self, run: RunId, seq: Seq) -> Result<(), Error> {
let mut i = self.borrow_mut();
i.clock.ran.set(true);
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
if let Some(rec) = slot.machine.awaiting(seq) {
i.append(&rec)?;
}
Ok(())
}
fn heartbeat(&self, run: RunId, seq: Seq, progress: Payload) -> Result<(), Error> {
let mut i = self.borrow_mut();
let progress = i.seal(progress)?;
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
if let Some(rec) = slot.machine.heartbeat(seq, progress) {
i.heartbeats.insert((run, seq), rec);
}
Ok(())
}
fn progress(&self, run: RunId, seq: Seq) -> Option<Payload> {
let i = self.borrow();
let p = i.slots.get(&run)?.machine.progress(seq)?;
i.open(p).ok()
}
fn patched(&self, run: RunId, id: &str, deprecated: bool) -> Result<bool, Error> {
let mut i = self.borrow_mut();
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let (on, rec) = slot.machine.patched(id);
if let Some(rec) = rec {
i.append(&rec)?;
}
Ok(on || deprecated)
}
fn set_attributes(&self, run: RunId, attributes: Payload) -> Result<(), Error> {
let mut i = self.borrow_mut();
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let r = slot.machine.set_attributes(attributes);
let (_, rec) = stop_on(&mut i, run, r)?;
if let Some(rec) = rec {
i.append(&rec)?;
}
Ok(())
}
fn object_get(&self, run: RunId, key: &str, exclusive: bool) -> Result<Option<Payload>, Error> {
Ok(self.borrow().object_run(run)?.get(key, exclusive))
}
fn object_set(&self, run: RunId, key: &str, value: Option<Payload>) -> Result<(), Error> {
self.borrow_mut().object_write(run, key, value)
}
fn object_keys(&self, run: RunId, exclusive: bool) -> Result<Vec<String>, Error> {
Ok(self.borrow().object_run(run)?.keys(exclusive))
}
fn object_state(&self, run: RunId) -> Result<Payload, Error> {
Ok(crate::object::state_json(
&self.borrow().object_run(run)?.state,
))
}
fn object_edit(&self, run: RunId, edit: &[u8]) -> Result<Payload, Error> {
self.borrow_mut().object_edit(run, edit)
}
fn begin_call(
&self,
run: RunId,
wtype: &str,
key: &str,
handler: &str,
arg: Payload,
detached: bool,
) -> Result<Option<Seq>, Error> {
let (id, flags, input) = self
.borrow()
.object_call(run, wtype, key, handler, &arg, detached)?;
self.begin_child(run, wtype, &id, flags, 0, input)
}
fn update_done(&self, run: RunId, ordinal: u64, outcome: RunOutcome) -> Result<(), Error> {
let mut i = self.borrow_mut();
let r = i.update_done(run, ordinal, outcome);
i.on_commit();
r
}
fn continue_as_new(
&self,
run: RunId,
input: Payload,
version: Option<u32>,
) -> Result<(), Error> {
self.borrow_mut()
.continue_as_new(run, input, version)
.map(|_| ())
}
fn journal_size(&self, run: RunId) -> u32 {
self.borrow()
.slots
.get(&run)
.map(|s| s.machine.journal_size())
.unwrap_or(0)
}
fn cancel_workflow(&self, run: RunId, target: &str, reason: &str) -> Result<(), Error> {
let mut i = self.borrow_mut();
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
let r = slot
.machine
.send(target, CANCEL_SIGNAL, Arc::from(reason.as_bytes()));
let (_, rec) = stop_on(&mut i, run, r)?;
if let Some(rec) = rec {
i.append(&rec)?;
}
Ok(())
}
fn limit_ticket(&self) -> u64 {
self.borrow().limits.ticket()
}
fn acquire_in(
&self,
run: RunId,
limit: &str,
ticket: u64,
priority: &Priority,
waker: Option<&Waker>,
) -> Acquire {
let (limits, now, own) = {
let i = self.borrow();
let own = match waker {
Some(_) => None,
None => i.slots.get(&run).map(|s| Waker::from(s.waker.clone())),
};
(i.limits.clone(), i.now(), own)
};
let w = waker.or(own.as_ref());
if self.borrow().cfg.sabotage == Sabotage::IgnoreLimits {
limits.define(
limit,
crate::limits::LimitSpec {
concurrent: 1_000,
per_second: 0.0,
},
);
}
let a = limits.acquire_in(limit, ticket, priority, now, w);
if let (
Acquire::Wait {
retry_at_ms: Some(at),
},
Some(w),
) = (a, w)
{
self.wake_at(at, w);
}
a
}
fn release(&self, limit: &str, ticket: u64) {
let limits = self.borrow().limits.clone();
limits.release(limit, ticket);
}
fn stream_open(&self, run: RunId, name: &str, step: Option<Seq>) -> Result<StreamOpen, Error> {
let mut i = self.borrow_mut();
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
slot.machine
.stream_open(name, step)
.map_err(Error::Rejected)
}
fn stream_append(
&self,
run: RunId,
name: &str,
step: Option<Seq>,
payload: Payload,
) -> Result<(), Error> {
let mut i = self.borrow_mut();
let payload = i.seal(payload)?;
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
if let Some(rec) = slot
.machine
.stream_append(name, step, payload)
.map_err(Error::Rejected)?
{
i.append(&rec)?;
}
Ok(())
}
fn stream_close(&self, run: RunId, name: &str, step: Option<Seq>) -> Result<(), Error> {
let mut i = self.borrow_mut();
let slot = i.slots.get_mut(&run).ok_or_else(|| err_slot(run))?;
if let Some(rec) = slot
.machine
.stream_close(name, step)
.map_err(Error::Rejected)?
{
i.append(&rec)?;
}
Ok(())
}
fn stream_recorded(
&self,
run: RunId,
name: &str,
step: Seq,
limit: u64,
) -> Result<Vec<Payload>, Error> {
let mut i = self.borrow_mut();
let mut out = Vec::new();
for r in i.journal_of(run)? {
if out.len() as u64 >= limit {
break;
}
if let Record::Chunk {
seq,
stream,
payload,
..
} = r
&& seq == step
&& *stream == *name
{
out.push(i.open(payload)?);
}
}
Ok(out)
}
}
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Due {
pub busy: bool,
pub running: Vec<String>,
pub awaiting: Vec<String>,
pub at_ms: Option<u64>,
pub steps: u64,
pub open_us: u64,
pub unacked: usize,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct Tick {
pub next_wake_ms: Option<u64>,
pub busy: bool,
}
pub struct ShardRuntime<E: Env + 'static> {
inner: Rc<RefCell<Inner<E>>>,
host: Rc<dyn Host>,
futures: RefCell<Vec<Option<Code>>>,
update_futs: RefCell<FxHashMap<RunId, Vec<(u64, LocalFuture)>>>,
handlers: RefCell<FxHashMap<RunId, Rc<Handlers>>>,
spare_handlers: RefCell<Vec<Rc<Handlers>>>,
deferred: RefCell<Vec<(RunId, String, Payload, u64)>>,
outside: Rc<Cell<u32>>,
spare: RefCell<Vec<RunId>>,
last_drives: Cell<usize>,
local: RefCell<Vec<RunId>>,
code: RefCell<CodeCache>,
}
impl<E: Env + 'static> Drop for ShardRuntime<E> {
fn drop(&mut self) {
self.update_futs.get_mut().clear();
self.futures.get_mut().clear();
for h in self.handlers.get_mut().values() {
h.clear();
}
self.handlers.get_mut().clear();
}
}
pub fn builtin_registry() -> SharedRegistry {
let mut r = Registry::default();
crate::schedule::register_builtins(&mut r);
Arc::new(RwLock::new(r))
}
fn io_err(e: impl std::fmt::Display) -> Error {
Error::Io(e.to_string())
}
#[allow(clippy::type_complexity)]
fn read_valid_prefix<E: Env>(
env: &mut E,
seg: u64,
after: Lsn,
records: &mut Vec<(Lsn, RecordBuf)>,
) -> Result<Result<Option<Scanned>, (ScanError, Vec<u8>)>, Error> {
const FIRST_READ: usize = 256 << 10;
const MAX_READ: usize = 16 << 20;
let (mut bytes, mut eof) = env.read_segment_from(seg, 0, FIRST_READ)?;
let more = |env: &mut E, bytes: &mut Vec<u8>, len: usize| -> Result<bool, Error> {
let (next, end) = env.read_segment_from(seg, bytes.len() as u64, len)?;
if next.is_empty() && !end {
return Err(io_err(format!(
"reading log segment {seg} made no progress"
)));
}
bytes.extend_from_slice(&next);
Ok(end)
};
let mut scanner = match SegmentScanner::start(&bytes, eof) {
Ok(Some(s)) => s,
Ok(None) => return Ok(Ok(None)),
Err(e) => {
while !eof {
eof = more(env, &mut bytes, MAX_READ)?;
}
return Ok(Err((e, bytes)));
}
};
let mut chunk = FIRST_READ;
loop {
let progress = scanner.advance(&bytes, eof, &mut |lsn, rec| {
if lsn > after {
records.push((lsn, rec));
}
});
match progress {
Progress::Done(s) => return Ok(Ok(Some(s))),
Progress::NeedMore => {
eof = more(env, &mut bytes, chunk)?;
chunk = (chunk * 2).min(MAX_READ);
}
}
}
}
impl<E: Env + 'static> ShardRuntime<E> {
pub fn open(
env: E,
cfg: RuntimeConfig,
registry: Option<SharedRegistry>,
wake: Notify,
) -> Result<Self, Error> {
let (registry, foreign_host) = match registry {
Some(r) => (r, false),
None => (builtin_registry(), true),
};
Self::open_with(env, cfg, registry, foreign_host, Services::default(), wake)
}
pub fn open_with(
mut env: E,
cfg: RuntimeConfig,
registry: SharedRegistry,
foreign_host: bool,
services: Services,
wake: Notify,
) -> Result<Self, Error> {
crate::trim::tune();
env.set_notify(wake.clone());
let now = env.now_ms();
let opened_us = env.monotonic_us();
let mut core = ShardCore::new(cfg.shard, cfg.shards, cfg.retention_ms, now);
let cp = env.load_checkpoint()?;
if let Some(cp) = &cp
&& (cp.shards != cfg.shards || cp.shard != cfg.shard)
{
return Err(Error::Io(format!(
"data directory was created with {} shards (this is shard {} of it); \
open it with the same shard count",
cp.shards, cp.shard
)));
}
let (generation, len) = cp
.as_ref()
.map(|c| (c.store_generation, c.store_len))
.unwrap_or((0, 0));
let saved_index = cp.as_ref().and_then(|c| c.index.clone());
let mut scanned: FxHashMap<RunId, (u64, u32, StoreEntry)> = FxHashMap::default();
let mut done_deltas: FxHashMap<RunId, (Status, u64)> = FxHashMap::default();
let mut visited = 0u64;
env.recover_store(generation, len, &mut |off, len, e| {
visited += 1;
match e {
StoreEntry::Tombstone { run } => {
scanned.remove(&run);
}
StoreEntry::Index { .. } => {}
mut e => {
if let StoreEntry::Delta { run, journal, .. } = &e {
match fusee_core::store::journal_done(journal) {
Some(done) => done_deltas.insert(*run, done),
None => done_deltas.remove(run),
};
} else {
done_deltas.remove(&e.run());
}
match &mut e {
StoreEntry::Park { journal, .. }
| StoreEntry::Done { journal, .. }
| StoreEntry::Delta { journal, .. } => *journal = Vec::new(),
StoreEntry::Tombstone { .. } | StoreEntry::Index { .. } => {}
}
scanned.insert(e.run(), (off, len, e));
}
}
})?;
let indexed = visited == 0 && saved_index.is_some();
let mut store_read = 0u64;
if indexed {
let frames = saved_index.as_deref().unwrap_or_default();
core.reserve_runs(cp.as_ref().map_or(0, |c| c.index_runs as usize));
for b in frames {
let frame = env.store_read(b.offset, b.len)?;
store_read += u64::from(b.len);
match read_entry(&frame).map_err(io_err)? {
StoreEntry::Index { records } => core.load_index(&records).map_err(io_err)?,
_ => {
return Err(Error::Io(format!(
"run store frame at {} is not part of the run index",
b.offset
)));
}
}
}
core.index_loaded();
} else {
for (run, (offset, len, e)) in scanned {
let blob = BlobRef { offset, len };
let earliest = |timers: &[(Seq, TimerKind, u64)]| {
timers.iter().map(|t| t.2).min().unwrap_or(0)
};
match e {
StoreEntry::Park {
id_hash, timers, ..
} => core.insert_parked(run, blob, None, false, id_hash, earliest(&timers)),
StoreEntry::Delta {
id_hash, timers, ..
} => match done_deltas.get(&run) {
Some(&(status, finished_ms)) => {
core.insert_parked(run, blob, Some(status), true, id_hash, finished_ms)
}
None => {
core.insert_parked(run, blob, None, true, id_hash, earliest(&timers))
}
},
StoreEntry::Done {
id_hash,
status,
finished_ms,
..
} => core.insert_parked(run, blob, Some(status), false, id_hash, finished_ms),
StoreEntry::Tombstone { .. } | StoreEntry::Index { .. } => {}
}
}
}
let store_v3 = env
.store_read(0, fusee_core::store::STORE_HEADER as u32)
.ok()
.and_then(|h| fusee_core::store::store_format(&h).ok())
.is_some_and(|f| f >= 3);
let after = cp.as_ref().map(|c| c.lsn).unwrap_or(0);
let ready = Arc::new(ReadyQueue {
queue: Mutex::new(Vec::new()),
queued: AtomicBool::new(false),
wake: wake.clone(),
});
let mut inner = Inner {
env,
core,
writer: ShardLog::Segments(LogWriter::new(cfg.shard, 1, 1, cfg.segment_bytes, now)),
cfg: cfg.clone(),
slots: Slots::default(),
resolved: FxHashMap::default(),
gates: Gates::default(),
round_us: 0,
flush_started_us: 0,
timers: BinaryHeap::new(),
timer_wakers: FxHashMap::default(),
timer_seq: 0,
durable: after,
commit: after,
load_room: usize::MAX,
commit_dirty: false,
inline_deliveries: 0,
load_buf: Vec::new(),
load_at: Vec::new(),
in_flight: None,
submitted: VecDeque::new(),
ready,
registry,
foreign_host,
codec: services.codec,
limits: services.limits,
update_waiters: FxHashMap::default(),
obj: Objects::default(),
offered: FxHashMap::default(),
next_update_id: 1,
completion_acks: FxHashMap::default(),
heartbeats: BTreeMap::new(),
step_began: FxHashMap::default(),
released: Vec::new(),
outbox: BTreeMap::new(),
next_out: 1,
msg_tokens: FxHashMap::default(),
next_msg_token: MSG_TOKEN,
child_acks: FxHashMap::default(),
message_acks: Vec::new(),
foreign: VecDeque::new(),
stash: FxHashMap::default(),
awaiting: FxHashMap::default(),
instantiate: Vec::new(),
cancel_tokens: FxHashMap::default(),
by: None,
subscribers: FxHashMap::default(),
stuck: FxHashMap::default(),
failed: None,
metrics: Metrics::default(),
ckpt_pending: None,
ckpt_running: false,
last_ckpt_ms: now,
last_ckpt_lsn: after,
last_ckpt_store_len: len.max(fusee_core::store::STORE_HEADER),
obsolete_store_bytes: 0,
last_ckpt_segment: 0,
last_sweep_ms: now,
current_step: FxHashMap::default(),
replicator: None,
stall_us: stall_threshold_us(),
stalls: VecDeque::new(),
dirty: Vec::new(),
park_queue: VecDeque::new(),
removed: Vec::new(),
undelivered: Vec::new(),
tick_parked: 0,
tick_maintenance_us: 0,
flush_records: 0,
flush_bytes: 0,
clock: TickClock::default(),
held_loads: VecDeque::new(),
reaping: Vec::new(),
store_v3,
no_compaction: false,
index_gen: generation,
index_base: saved_index.as_ref().map_or((0, 0), |f| {
(f.len(), f.iter().map(|b| u64::from(b.len)).sum())
}),
index: if indexed { saved_index.clone() } else { None },
index_delta_bytes: 0,
index_build: None,
slow_flushes: 0,
tail: StoreTail::default(),
gate_flushes: 0,
drive_backlog: 0,
host_queue: None,
host_never: false,
enc: Vec::new(),
timer_scratch: Vec::new(),
decoded: Vec::new(),
names: Default::default(),
slot_spare: Vec::new(),
streams: FxHashMap::default(),
stream_bytes: 0,
stream_waiters: FxHashMap::default(),
stream_due: BinaryHeap::new(),
stream_wait_until: u64::MAX,
trim_peak: 0,
spare_mark: 0,
pools_shrunk: false,
};
inner.metrics.open_store_frames = if indexed {
saved_index.as_ref().map_or(0, |f| f.len() as u64)
} else {
visited
};
inner.metrics.open_store_bytes = if indexed { store_read } else { len };
if let Some(cp) = &cp {
for mut o in decode_outbox(&cp.outbox).map_err(io_err)? {
o.id = inner.next_out;
inner.next_out += 1;
inner.outbox.insert(o.id, o);
}
inner.core.set_next_run_id(cp.next_run);
for r in &cp.runs {
let stored = match r.blob {
Some(b) => Some((b, inner.read_stored(b)?)),
None => None,
};
if let Some((_, s)) = &stored {
for rec in &s.records {
inner.core.apply(r.last_lsn, rec, Mode::BLOB);
}
}
if let Some((b, s)) = &stored {
inner.core.spilled(r.run, *b, s.chain, s.depth);
}
let suffix = decode_canonical_all(&r.journal).map_err(io_err)?;
for rec in &suffix {
inner.core.apply(r.last_lsn, rec, Mode::SUFFIX);
}
}
}
let mut last = after;
let shared = inner.env.shared_log();
let mut next_segment = 0;
let early_checkpoint;
if shared {
let (records, tail) = inner.env.recover_log(after)?;
early_checkpoint = !records.is_empty();
for (lsn, rec) in records {
let run = rec.run();
if inner.core.is_parked(run) && !matches!(rec, Record::Start { .. }) {
inner.unpark_stored(run)?;
}
inner.core.apply(lsn, &rec, Mode::RECOVER);
}
last = tail.max(after);
} else {
let segments = inner.env.segments()?;
let last_segment = segments.last().copied();
let max_segment = last_segment.unwrap_or(0);
let mut begin = 0;
if after > 0 {
for (index, seg) in segments.iter().enumerate().rev() {
let head = inner.env.read_segment_header(*seg)?;
if let Ok(Some(h)) = fusee_core::segment::SegmentHeader::decode(&head)
&& h.segment == *seg
&& h.base_lsn <= after + 1
{
begin = index;
break;
}
}
}
let mut expected_next: Option<Lsn> = None;
let mut first = true;
let exclusive = inner.env.exclusive_log();
let mut empty_tail: Vec<u64> = Vec::new();
let mut free_from: Option<u64> = None;
let mut tail_segments = 0usize;
let mut tail_bytes = 0u64;
for &seg in &segments[begin..] {
let mut records: Vec<(Lsn, RecordBuf)> = Vec::new();
let scanned: Option<Scanned> = match read_valid_prefix(
&mut inner.env,
seg,
after,
&mut records,
)? {
Ok(s) => s,
Err((fusee_core::segment::ScanError::BadHeader, bytes))
if Some(seg) == last_segment =>
{
let base = expected_next.unwrap_or(after + 1);
if fusee_core::segment::headerless_was_flushed(&bytes, base) {
return Err(Error::Io(format!(
"log segment {seg} has a damaged header but holds flushed records: \
acknowledged records were lost. Restore from a replica or backup."
)));
}
inner.env.discard_segment(seg)?;
free_from = Some(seg);
None
}
Err((e, _)) => return Err(Error::Io(format!("log segment {seg}: {e}"))),
};
let Some(s) = scanned else {
if free_from != Some(seg) && exclusive {
empty_tail.push(seg);
}
continue;
};
if s.header.shard != cfg.shard {
return Err(Error::Io(format!(
"segment {seg} belongs to shard {}, not {}",
s.header.shard, cfg.shard
)));
}
if first {
first = false;
if s.header.base_lsn > after + 1 {
return Err(Error::Io(format!(
"log records {}..{} are missing (segment {seg} starts at {})",
after + 1,
s.header.base_lsn - 1,
s.header.base_lsn
)));
}
} else if let Some(n) = expected_next {
if s.header.base_lsn > n && s.max_batch_start <= n {
inner.env.discard_segment(seg)?;
let later: Vec<u64> = inner
.env
.segments()?
.into_iter()
.filter(|x| *x > seg)
.collect();
for l in later {
inner.env.discard_segment(l)?;
}
free_from = Some(seg);
break;
}
if s.header.base_lsn != n {
return Err(Error::Io(format!(
"log segment {seg} starts at record {} but the previous segment ends at {}: \
durable records were lost",
s.header.base_lsn,
n - 1
)));
}
}
if let Tail::Corrupt { at, later_lsn } = s.tail {
return Err(Error::Io(format!(
"log segment {seg} is damaged at offset {at} (record {}), but record {later_lsn} \
after it is intact: acknowledged records were lost. Restore from a replica or \
backup.",
s.last_lsn + 1
)));
}
if s.records == 0 && exclusive {
empty_tail.push(seg);
} else {
empty_tail.clear();
}
if !records.is_empty() {
tail_segments += 1;
tail_bytes += s.end as u64;
}
for (lsn, rec) in records {
let run = rec.run();
if inner.core.is_parked(run) && !matches!(rec, Record::Start { .. }) {
inner.unpark_stored(run)?;
}
inner.core.apply(lsn, &rec, Mode::RECOVER);
}
last = last.max(s.last_lsn.max(after));
expected_next = Some(s.last_lsn + 1);
}
for &seg in &empty_tail {
inner.env.discard_segment(seg)?;
}
next_segment = empty_tail
.first()
.copied()
.or(free_from)
.unwrap_or(max_segment + 1);
let present = inner.env.segments()?.len();
early_checkpoint =
tail_segments >= 2 || present >= 3 || tail_bytes >= cfg.segment_bytes / 4;
}
inner.obsolete_store_bytes = match &cp {
Some(c) if c.index.is_some() => c.obsolete,
_ => {
let live_len = inner
.core
.parked_runs()
.map(|(_, p)| u64::from(p.blob().len))
.chain(inner.core.loaded_runs().filter_map(|(_, l)| {
l.blob.map(|b| {
if l.blob_depth > 0 {
l.blob_chain
} else {
u64::from(b.len)
}
})
}))
.sum::<u64>();
inner
.env
.store_len()
.saturating_sub(fusee_core::store::STORE_HEADER + live_len)
}
};
inner.env.sync_recovered()?;
while inner.core.take_event().is_some() {}
if shared || !inner.env.recovers() {
inner.writer = ShardLog::Sections(fusee_core::wal::SectionWriter::new(last + 1));
inner.last_ckpt_segment = inner.log_segment();
} else {
let segment = next_segment;
inner.env.open_log(segment)?;
let mut writer = LogWriter::new(cfg.shard, segment, last + 1, cfg.segment_bytes, now);
writer.limit_first_segment(inner.env.first_segment_bytes(cfg.segment_bytes));
inner.writer = ShardLog::Segments(writer);
inner.last_ckpt_segment = segment;
}
if early_checkpoint {
inner.last_ckpt_ms = 0;
}
inner.durable = last;
inner.commit = last;
inner.on_commit();
let loaded: Vec<(RunId, bool)> = inner
.core
.loaded_runs()
.map(|(r, l)| (r, l.is_done()))
.collect();
for (run, done) in loaded {
if done {
inner.park_done(run)?;
} else {
let mut journal = inner.journal_of(run)?;
inner.prepare(run, &mut journal);
inner.core.reoffer(run);
}
}
inner.expire_retained(now)?;
inner.metrics.open_us = inner.env.monotonic_us().saturating_sub(opened_us);
let inner = Rc::new(RefCell::new(inner));
let host: Rc<dyn Host> = inner.clone();
let rt = Self {
inner,
host,
futures: RefCell::new(Vec::new()),
update_futs: RefCell::new(FxHashMap::default()),
handlers: RefCell::new(FxHashMap::default()),
spare_handlers: RefCell::new(Vec::new()),
code: RefCell::new(CodeCache::default()),
deferred: RefCell::new(Vec::new()),
outside: Rc::new(Cell::new(0)),
spare: RefCell::new(Vec::new()),
last_drives: Cell::new(0),
local: RefCell::new(Vec::new()),
};
rt.instantiate_pending();
Ok(rt)
}
pub fn with<R>(&self, f: impl FnOnce(&mut Inner<E>) -> R) -> R {
f(&mut self.inner.borrow_mut())
}
pub fn host(&self) -> Rc<dyn Host> {
self.host.clone()
}
pub fn metrics(&self) -> Metrics {
self.inner.borrow().metrics()
}
pub fn waits_outside(&self) -> bool {
self.outside.get() > 0
}
fn instantiate_pending(&self) -> bool {
if self.inner.borrow().instantiate.is_empty() {
return false;
}
let pending = std::mem::take(&mut self.inner.borrow_mut().instantiate);
if pending.is_empty() {
return false;
}
for run in pending {
let (info, foreign, stash) = {
let mut i = self.inner.borrow_mut();
let Some(slot) = i.slots.get(&run) else {
continue;
};
let info = slot.machine.info().clone();
let foreign = slot.foreign;
let stash = i.stash.remove(&run).unwrap_or_default();
(info, foreign, stash)
};
let info = {
let i = self.inner.borrow();
let mut info = info;
match i.open(info.input.clone()) {
Ok(p) => info.input = p,
Err(e) => {
drop(i);
self.inner
.borrow_mut()
.make_stuck(run, format!("the workflow input cannot be read: {e}"));
continue;
}
}
info
};
if foreign {
let mut i = self.inner.borrow_mut();
i.foreign.push_back(ForeignEvent::Start { run, info });
for s in stash {
match s {
Stashed::Complete(seq, c) => i.complete(run, seq, c),
Stashed::Wake => i.wake(run),
}
}
continue;
}
let known = {
let i = self.inner.borrow();
self.code
.borrow_mut()
.get(&i.registry, &info.wtype, info.version)
.is_some()
};
if !known {
let mut i = self.inner.borrow_mut();
i.awaiting
.entry(info.wtype.to_string())
.or_default()
.push(run);
i.stash.insert(run, stash);
continue;
}
let handlers = self
.spare_handlers
.borrow_mut()
.pop()
.unwrap_or_else(|| Rc::new(Handlers::default()));
self.handlers.borrow_mut().insert(run, handlers.clone());
let fut = {
let code = self.code.borrow();
let Some(f) = code.cached(&info.wtype, info.version) else {
continue;
};
let ctx = Ctx::new(self.host.clone(), run, info, handlers, self.outside.clone());
f(ctx)
};
let mut i = self.inner.borrow_mut();
if let Some(idx) = i.slots.idx(run)
&& let Some(slot) = i.slots.entries[idx as usize].as_deref()
{
let mut futures = self.futures.borrow_mut();
let idx = idx as usize;
if futures.len() <= idx {
futures.resize_with(idx + 1, || None);
}
futures[idx] = Some(Code::new(fut, &slot.waker));
}
for s in stash {
match s {
Stashed::Complete(seq, c) => i.complete(run, seq, c),
Stashed::Wake => {}
}
}
i.wake(run);
}
true
}
pub fn registered(&self, wtype: &str) {
let runs = self
.inner
.borrow_mut()
.awaiting
.remove(wtype)
.unwrap_or_default();
if runs.is_empty() {
return;
}
self.inner.borrow_mut().instantiate.extend(runs);
self.instantiate_pending();
}
fn drive(&self, run: RunId) {
{
let mut i = self.inner.borrow_mut();
i.count_drive();
i.mark_active(run);
}
for _ in 0..10_000 {
let (idx, mut code) = {
let i = self.inner.borrow();
let Some(idx) = i.slots.idx(run) else {
return;
};
let idx = idx as usize;
let Some(s) = i.slots.entries[idx].as_deref() else {
return;
};
let taken = self
.futures
.borrow_mut()
.get_mut(idx)
.and_then(Option::take);
let Some(mut code) = taken else {
s.waker.queued.store(false, Ordering::Release);
return;
};
if !code.polled_with(&s.waker) {
code = Code::new(code.fut, &s.waker);
}
(idx, code)
};
code.waker().queued.store(false, Ordering::Release);
code.waker().woken.store(false, Ordering::Release);
let Code { fut, handle } = &mut code;
let mut cx = Context::from_waker(handle);
self.inner.borrow_mut().inline_deliveries = INLINE_DELIVERIES;
let polled = fut.as_mut().poll(&mut cx);
self.inner.borrow_mut().inline_deliveries = 0;
match polled {
Poll::Ready(outcome) => {
drop(code);
self.inner.borrow_mut().finish(run, outcome);
return;
}
Poll::Pending => {
if self.inner.borrow().slots.idx(run) != Some(idx as u32) {
return;
}
}
}
if !self.poll_updates(run, &mut cx) {
return;
}
if run_waker(handle).woken.load(Ordering::Acquire) {
self.put_code(idx, code);
continue;
}
let next = {
let mut i = self.inner.borrow_mut();
match i.deliver(run) {
Ok(d) => d,
Err(e) => {
i.fail(e.to_string());
None
}
}
};
let delivered = match next {
Some(Delivery::Update {
ordinal,
name,
id,
payload,
token,
}) => {
self.start_update(run, ordinal, &name, &id, payload, token);
true
}
Some(d) => {
run_waker(handle).queued.store(true, Ordering::Release);
self.inner.borrow_mut().store_delivery(run, d);
true
}
None => false,
};
if self.inner.borrow().slots.contains_key(&run) {
self.put_code(idx, code);
}
if !delivered {
return;
}
}
if let Some(slot) = self.inner.borrow().slots.get(&run) {
slot.waker.queued.store(false, Ordering::Release);
slot.waker.wake_by_ref();
}
}
fn put_code(&self, idx: usize, code: Code) {
if let Some(place) = self.futures.borrow_mut().get_mut(idx) {
*place = Some(code);
}
}
fn poll_updates(&self, run: RunId, cx: &mut Context<'_>) -> bool {
let Some(mut list) = ({
let mut u = self.update_futs.borrow_mut();
if u.is_empty() { None } else { u.remove(&run) }
}) else {
return true;
};
let mut k = 0;
while k < list.len() {
match list[k].1.as_mut().poll(cx) {
Poll::Ready(outcome) => {
let (ordinal, f) = list.remove(k);
drop(f);
let _ = self.host.update_done(run, ordinal, outcome);
}
Poll::Pending => k += 1,
}
if !self.inner.borrow().slots.contains_key(&run) {
return false;
}
}
if !list.is_empty() {
self.update_futs.borrow_mut().insert(run, list);
}
true
}
fn start_update(
&self,
run: RunId,
ordinal: u64,
name: &str,
id: &str,
payload: Payload,
token: Option<u64>,
) {
let entry = self
.handlers
.borrow()
.get(&run)
.and_then(|h| h.update(name));
let ordinal = match (token, entry.as_ref()) {
(Some(t), None) => {
let r = self
.inner
.borrow_mut()
.decide_update(t, Err(format!("no handler for update '{name}'")));
if let Err(e) = r {
self.inner.borrow_mut().fail(e.to_string());
}
return;
}
(Some(t), Some((validate, _))) => {
let decision = validate(&payload);
let accepted = decision.is_ok();
let decided = self.inner.borrow_mut().decide_update(t, decision);
match decided {
Ok(Some(o)) if accepted => o,
Ok(_) => return,
Err(e) => {
self.inner.borrow_mut().fail(e.to_string());
return;
}
}
}
(None, None) => {
self.inner.borrow_mut().make_stuck(
run,
format!(
"determinism error: the journal delivers update '{name}' (id '{id}') but \
the code registered no handler for it"
),
);
return;
}
(None, Some(_)) => ordinal,
};
if let Some((_, handle)) = entry {
let fut = handle(payload);
self.update_futs
.borrow_mut()
.entry(run)
.or_default()
.push((ordinal, fut));
}
}
fn drive_ready(&self) -> bool {
let mut runs = std::mem::take(&mut *self.spare.borrow_mut());
{
let i = self.inner.borrow();
if i.ready.queued.swap(false, Ordering::Acquire)
&& let Ok(mut q) = i.ready.queue.lock()
{
std::mem::swap(&mut *q, &mut runs);
}
take_local(&i.ready, Some(&mut runs));
}
self.last_drives.set(runs.len());
if runs.is_empty() {
*self.spare.borrow_mut() = runs;
return false;
}
{
let mut i = self.inner.borrow_mut();
i.gate_flushes = if i.env.last_flush_us() == 0 {
64
} else if crate::spin::few_busy() {
1
} else {
0
};
}
let n = runs.len();
let every = drive_flush_every();
for (k, &run) in runs.iter().enumerate() {
let foreign = {
let mut i = self.inner.borrow_mut();
i.drive_backlog = n - k - 1;
match i.slots.get(&run) {
Some(s) => s.foreign,
None => continue,
}
};
if foreign {
let mut i = self.inner.borrow_mut();
if let Some(s) = i.slots.get(&run) {
s.waker.queued.store(false, Ordering::Release);
}
i.mark_active(run);
if let Some(slot) = i.slots.get_mut(&run)
&& !slot.wake_queued
{
slot.wake_queued = true;
i.foreign.push_back(ForeignEvent::Wake { run });
}
continue;
}
self.drive(run);
if every > 0 && (k + 1) % every == 0 && k + 1 < n {
let mut i = self.inner.borrow_mut();
i.poll_io();
i.maybe_flush(false);
}
}
{
let mut i = self.inner.borrow_mut();
i.gate_flushes = 0;
i.drive_backlog = 0;
}
runs.clear();
*self.spare.borrow_mut() = runs;
true
}
pub fn behind(&self) -> bool {
let limit = admit_round();
limit != 0 && self.last_drives.get() > limit
}
fn has_ready(&self) -> bool {
let i = self.inner.borrow();
i.ready.queued.load(Ordering::Acquire)
|| take_local(&i.ready, None)
|| i.core.has_events()
|| !i.instantiate.is_empty()
}
pub(crate) fn needs_foreign_poll(&self, deadline: Option<u64>) -> bool {
let i = self.inner.borrow();
Rc::strong_count(&self.host) > 2
|| i.in_flight.is_some()
|| i.ckpt_running
|| !i.outbox.is_empty()
|| deadline.is_some_and(|d| d <= i.now())
}
pub fn watch_host_queue(&self, queued: Arc<std::sync::atomic::AtomicUsize>) {
self.inner.borrow_mut().host_queue = Some(queued);
}
pub fn tick(&self) -> Tick {
self.inner.borrow_mut().host_never = true;
let tick = self.tick_with(&|| false);
self.inner.borrow_mut().host_never = false;
tick
}
pub fn tick_with(&self, external: &dyn Fn() -> bool) -> Tick {
let traced = {
let mut i = self.inner.borrow_mut();
i.tick_parked = 0;
i.tick_maintenance_us = 0;
i.set_ticking(true);
(i.stall_us != 0).then(|| i.env.monotonic_us())
};
let local = LocalTick::enter(self.inner.borrow().ready.clone(), &self.local);
let tick = self.tick_rounds(external);
drop(local);
self.inner.borrow_mut().set_ticking(false);
if let Some(t) = traced {
let mut i = self.inner.borrow_mut();
let us = i.env.monotonic_us().saturating_sub(t);
i.trace_stall(StallKind::Tick, us);
}
tick
}
pub fn take_stalls(&self) -> Vec<Stall> {
self.inner.borrow_mut().stalls.drain(..).collect()
}
fn tick_rounds(&self, external: &dyn Fn() -> bool) -> Tick {
let mut busy = false;
for round in 0.. {
if round >= 256 {
busy = true;
break;
}
let mut progressed = false;
{
let mut i = self.inner.borrow_mut();
i.round_us = i.metric_us();
}
progressed |= self.inner.borrow_mut().poll_io();
if self.inner.borrow().failed.is_some() {
return Tick {
next_wake_ms: None,
busy: false,
};
}
let host_full = {
let i = self.inner.borrow();
i.load_room != usize::MAX && i.foreign.len() >= i.load_room
};
if !host_full {
progressed |= self.inner.borrow_mut().poll_timers();
}
progressed |= self.inner.borrow_mut().process_events();
progressed |= self.instantiate_pending();
progressed |= self.drive_ready();
progressed |= self.compact_objects();
progressed |= self.retry_deferred() && self.has_ready();
progressed |= self.inner.borrow_mut().process_events();
let waiting = external();
let idle = !waiting && !self.has_ready() && {
let i = self.inner.borrow();
i.foreign.is_empty()
|| (i.foreign.len() == 1
&& matches!(i.foreign.front(), Some(ForeignEvent::Start { .. })))
};
let flushed = self.inner.borrow_mut().maybe_flush(idle);
progressed |= flushed;
if self.inner.borrow().removed.len() >= REAP_EARLY {
self.reap();
}
if !progressed {
break;
}
if waiting {
busy = true;
break;
}
}
if !self.inner.borrow().held_loads.is_empty() {
busy = true;
}
self.reap();
let mut i = self.inner.borrow_mut();
i.maintenance();
if std::mem::take(&mut i.pools_shrunk) {
let mut spare = self.spare_handlers.borrow_mut();
spare.truncate(SLOT_SPARE_IDLE);
spare.shrink_to(SLOT_SPARE_IDLE);
}
let mut next = i.core.next_timer();
if let Some(Reverse((at, _))) = i.timers.peek() {
next = Some(next.map_or(*at, |n| n.min(*at)));
}
let sweep = i.last_sweep_ms + 100;
if i.stream_wait_until != u64::MAX {
let at = i.stream_wait_until;
next = Some(next.map_or(at, |n| n.min(at)));
}
let building = i.index_build.as_ref().is_some_and(|b| b.cursor.is_some());
if !i.slots.is_empty() || i.core.loaded_len() > 0 || !i.streams.is_empty() || building {
next = Some(next.map_or(sweep, |n| n.min(sweep)));
}
if let Some(expiry) = i.core.next_expiry() {
let at = expiry.max(sweep);
next = Some(next.map_or(at, |n| n.min(at)));
}
if !i.ckpt_running && i.ckpt_pending.is_none() && i.checkpoint_needed() {
let c = i.last_ckpt_ms + i.cfg.checkpoint_interval_ms;
next = Some(next.map_or(c, |n| n.min(c)));
}
if i.env.due_wanted() {
let mut due = i.due();
due.busy |= busy;
i.env.report_due(due, false);
}
Tick {
next_wake_ms: next,
busy,
}
}
fn reap(&self) {
if self.inner.borrow().removed.is_empty() {
return;
}
let removed = std::mem::take(&mut self.inner.borrow_mut().removed);
let mut dropped: Vec<Code> = Vec::new();
let mut updates = Vec::new();
let mut handlers = Vec::new();
let mut reloaded = Vec::new();
{
let i = self.inner.borrow();
let mut f = self.futures.borrow_mut();
let mut u = self.update_futs.borrow_mut();
let mut h = self.handlers.borrow_mut();
for &(run, idx) in &removed {
dropped.extend(f.get_mut(idx as usize).and_then(Option::take));
if i.slots.contains_key(&run) {
reloaded.push(run);
continue;
}
updates.extend(u.remove(&run));
handlers.extend(h.remove(&run));
}
}
self.inner.borrow_mut().reaping = reloaded;
drop(dropped);
drop(updates);
for h in &handlers {
h.clear();
}
{
let mut spare = self.spare_handlers.borrow_mut();
for h in handlers {
if spare.len() < SLOT_SPARE && Rc::strong_count(&h) == 1 && Rc::weak_count(&h) == 0
{
spare.push(h);
}
}
}
self.inner.borrow_mut().reaping.clear();
let mut i = self.inner.borrow_mut();
for (_, idx) in removed {
i.slots.release(idx);
}
let mut f = self.futures.borrow_mut();
f.truncate(i.slots.entries.len());
let n = f.len();
if f.capacity() > 1024 && f.capacity() > 4 * n {
f.shrink_to(n * 2);
}
}
pub fn take_released(&self) -> Vec<Output> {
std::mem::take(&mut self.inner.borrow_mut().released)
}
pub fn take_foreign(&self) -> Vec<ForeignEvent> {
let mut i = self.inner.borrow_mut();
let events: Vec<ForeignEvent> = i.foreign.drain(..).collect();
for e in &events {
if let ForeignEvent::Wake { run } = e
&& let Some(slot) = i.slots.get_mut(run)
{
slot.wake_queued = false;
}
}
events
}
pub fn start(&self, req: StartRequest) -> Result<StartOutcome, Error> {
let outcome = self.start_locked(req)?;
self.instantiate_pending();
Ok(outcome)
}
fn start_locked(&self, mut req: StartRequest) -> Result<StartOutcome, Error> {
let mut i = self.inner.borrow_mut();
i.check_alive()?;
let object = i.object_start(&mut req)?;
let version = match req.version {
Some(v) => v,
None => {
let registry = i
.registry
.read()
.map_err(|_| Error::Io("registry poisoned".into()))?;
match registry.latest(&req.wtype) {
Some(v) => v,
None if i.foreign_host => 1,
None => {
return Err(Error::UnknownWorkflow(crate::error::unknown_workflow(
&req.wtype,
®istry.names(),
)));
}
}
}
};
let ids = IdPolicy::from_bits(req.policy);
if let Some(found) = i.core.find(&req.id) {
let run = match found {
Found::Loaded(r) | Found::Parked(r) => r,
};
let done = match found {
Found::Loaded(r) => i.core.get(r).and_then(|l| l.done.map(|d| d.0)),
Found::Parked(r) => i.core.parked(r).and_then(|p| p.done()),
};
if object && done.is_none() {
i.object_check_held(run, &req)?;
}
#[derive(PartialEq)]
enum Act {
Use,
New,
Refuse,
Replace,
}
let act = match done {
None => match ids.running {
OnRunning::Fail => Act::Refuse,
OnRunning::UseExisting => Act::Use,
OnRunning::TerminateExisting => Act::Replace,
},
Some(status) => match ids.finished {
OnFinished::Allow => Act::New,
OnFinished::AllowIfFailed if status != Status::Ok => Act::New,
OnFinished::AllowIfFailed | OnFinished::Refuse => Act::Refuse,
OnFinished::UseExisting => Act::Use,
},
};
let done = done.is_some();
match act {
Act::Refuse => {
return match req.token {
Some(t) => {
let tail = i.writer.last_lsn();
i.core.add_output(tail, Output::Reject(t));
i.on_commit();
Ok(StartOutcome::Existing { run, done })
}
None => Err(Error::AlreadyExists(req.id)),
};
}
Act::Use => {
let own_parent = i.core.get(run).and_then(|l| l.parent);
let detached = req.policy & policy::DETACHED != 0;
let mut ack = true;
if let (Some(parent), false) = (req.parent, detached) {
if done {
let (status, payload) = match self.result_sealed(&mut i, run)? {
Some(r) => r,
None => (Status::Failed, empty()),
};
i.send_out(
parent.shard,
Message::ChildResult {
parent,
status,
payload,
},
);
} else if own_parent != Some(parent) {
let subs = i.subscribers.entry(run).or_default();
if !subs.contains(&parent) {
subs.push(parent);
}
ack = false;
}
}
if let Some(u) = req.update.take() {
i.ensure_loaded(run)?;
let uid = if u.id.is_empty() {
i.fresh_update_id()
} else {
u.id
};
i.offer_update(run, &u.name, &uid, u.payload, u.token)?;
}
if let (Some(t), true) = (req.token, ack) {
let tail = i.writer.last_lsn();
i.core.add_output(tail, Output::Ack(t));
}
i.on_commit();
return Ok(StartOutcome::Existing { run, done });
}
Act::Replace => {
i.end_run(
run,
Status::Terminated,
"terminated by a new start with the same workflow ID",
)?;
}
Act::New => {}
}
}
if i.core.loaded_len() >= i.cfg.max_loaded {
i.metrics.busy += 1;
return Err(Error::Busy);
}
let run = i.core.allocate_run();
let now = i.now();
let input = if req.sealed {
req.input
} else {
i.seal(req.input)?
};
let mut accepted = None;
let update = match req.update.take() {
Some(u) => {
let uid = if u.id.is_empty() {
i.fresh_update_id()
} else {
u.id
};
let payload = i.seal(u.payload)?;
match u.token {
Some(t) if object && crate::object::is_send(&uid) => accepted = Some(t),
Some(t) => i
.update_waiters
.entry((run, Name::from(uid.as_str())))
.or_default()
.push(t),
None => {}
}
Some(Box::new(StartUpdate {
name: Name::from(u.name.as_str()),
id: Name::from(uid.as_str()),
payload,
}))
}
None => None,
};
let start: RecordBuf = Record::Start {
run,
id: Arc::from(req.id.as_str()),
wtype: Arc::from(req.wtype.as_str()),
version,
codec: req.codec,
started_ms: now,
deadline_ms: req.deadline_ms,
parent: req.parent,
policy: req.policy,
from: None,
copies: 0,
update,
input,
meta: StartMeta::of(None, req.trace),
};
let lsn = i.append(&start)?;
if let Some(t) = req.token {
i.core.add_output(lsn, Output::Ack(t));
}
if let Some(t) = accepted {
i.core.add_output(
lsn,
Output::Reply {
token: t,
status: Status::Ok,
payload: Arc::from(&b"null"[..]),
},
);
}
i.metrics.runs_started += 1;
let foreign = i.runs_foreign(&req.wtype, version);
i.new_slot(
run,
JournalIn::Borrowed(std::slice::from_ref(&start)),
lsn,
foreign,
)?;
i.instantiate.push(run);
Ok(StartOutcome::Created { run })
}
fn result_sealed(
&self,
i: &mut Inner<E>,
run: RunId,
) -> Result<Option<(Status, Payload)>, Error> {
if let Some(l) = i.core.get(run) {
return Ok(l
.done
.map(|(s, _)| (s, l.result.clone().unwrap_or_else(empty))));
}
let Some(p) = i.core.parked(run) else {
return Ok(None);
};
if p.done().is_none() {
return Ok(None);
}
for rec in i.blob_records(p.blob())? {
if let Record::Done {
status, payload, ..
} = rec
{
return Ok(Some((status, payload)));
}
}
Ok(None)
}
fn result_locked(&self, i: &mut Inner<E>, run: RunId) -> Result<ResultState, Error> {
let (status, payload, id) = if let Some(l) = i.core.get(run) {
match l.done {
Some(_) if l.last_lsn > i.commit => return Ok(ResultState::Running(run)),
Some((status, _)) => (status, l.result.clone().unwrap_or_else(empty), l.id.clone()),
None => return Ok(ResultState::Running(run)),
}
} else {
let Some(p) = i.core.parked(run) else {
return Ok(ResultState::NotFound);
};
if p.done().is_none() {
return Ok(ResultState::Running(run));
}
let b = p.blob();
let frame = read_store(&mut i.env, &i.tail, b.offset, b.len)?;
if let Ok(StoreEntry::Delta {
last_lsn, journal, ..
}) = read_entry(&frame)
&& let Some(Record::Done {
status, payload, ..
}) = decode_canonical_all(&journal)
.ok()
.and_then(|mut v| v.pop())
&& !status.is_handover()
{
if last_lsn > i.commit {
return Ok(ResultState::Running(run));
}
return self.finished_result(i, run, status, payload);
}
let stored = i.stored_from(b, Some(&frame))?;
if stored.last_lsn > i.commit {
return Ok(ResultState::Running(run));
}
let recs = stored.records;
let id = match recs.first() {
Some(Record::Start { id, .. }) => id.clone(),
_ => return Ok(ResultState::NotFound),
};
match recs.into_iter().find_map(|r| match r {
Record::Done {
status, payload, ..
} => Some((status, payload)),
_ => None,
}) {
Some((status, payload)) => (status, payload, id),
None => return Ok(ResultState::NotFound),
}
};
if status.is_handover() {
return match i.core.find(&id) {
Some(Found::Loaded(r) | Found::Parked(r)) if r > run => self.result_locked(i, r),
_ => Ok(ResultState::Running(run)),
};
}
self.finished_result(i, run, status, payload)
}
fn finished_result(
&self,
i: &mut Inner<E>,
run: RunId,
status: Status,
payload: Payload,
) -> Result<ResultState, Error> {
let payload = if status == Status::Ok {
match i.open(payload) {
Ok(p) => p,
Err(e) => {
return Ok(ResultState::Done {
run,
status: Status::Failed,
payload: Arc::from(e.to_string().as_bytes()),
});
}
}
} else {
payload
};
Ok(ResultState::Done {
run,
status,
payload,
})
}
pub fn result(&self, id: &str) -> Result<ResultState, Error> {
let mut i = self.inner.borrow_mut();
let run = match i.core.find(id) {
Some(Found::Loaded(r) | Found::Parked(r)) => r,
None => return Ok(ResultState::NotFound),
};
self.result_locked(&mut i, run)
}
pub fn find(&self, id: &str) -> Option<RunId> {
match self.inner.borrow().core.find(id)? {
Found::Loaded(r) | Found::Parked(r) => Some(r),
}
}
pub fn with_by<T>(&self, by: Option<Payload>, f: impl FnOnce() -> T) -> T {
if by.is_none() {
return f();
}
self.inner.borrow_mut().by = by;
let out = f();
self.inner.borrow_mut().by = None;
out
}
pub fn signal(
&self,
id: &str,
name: &str,
payload: Payload,
dedup: u64,
token: Option<u64>,
) -> Result<(), Error> {
let payload = self.inner.borrow().seal(payload)?;
self.signal_sealed(id, name, payload, dedup, token)
}
fn signal_sealed(
&self,
id: &str,
name: &str,
payload: Payload,
dedup: u64,
token: Option<u64>,
) -> Result<(), Error> {
{
let mut i = self.inner.borrow_mut();
let by = i.by.take();
i.check_alive()?;
let run = match i.core.find(id) {
Some(Found::Loaded(r) | Found::Parked(r)) => r,
None => return Err(Error::NotFound(id.to_owned())),
};
i.ensure_loaded(run)?;
if let Some(l) = i.core.get(run).filter(|l| l.is_done()) {
let at = l.last_lsn.max(i.writer.last_lsn());
return match token {
Some(t) => {
i.core.add_output(at, Output::Reject(t));
i.on_commit();
Ok(())
}
None => Err(Error::Rejected(format!(
"workflow '{id}' has already finished"
))),
};
}
let lsn = if i.core.seen_signal(run, dedup) {
i.writer.last_lsn()
} else {
let rec: RecordBuf = Record::Signal {
run,
name: Arc::from(name),
dedup,
payload,
by,
};
i.append(&rec)?
};
if let Some(t) = token {
i.core.add_output(lsn, Output::Ack(t));
}
i.on_commit();
}
self.instantiate_pending();
Ok(())
}
pub fn cancel(&self, id: &str, reason: &str, token: Option<u64>) -> Result<(), Error> {
{
let mut i = self.inner.borrow_mut();
let by = i.by.take();
i.check_alive()?;
let run = match i.core.find(id) {
Some(Found::Loaded(r) | Found::Parked(r)) => r,
None => return Err(Error::NotFound(id.to_owned())),
};
i.ensure_loaded(run)?;
if i.core.get(run).is_some_and(|l| l.is_done()) {
if let Some(t) = token {
let tail = i.writer.last_lsn();
i.core.add_output(tail, Output::Ack(t));
i.on_commit();
}
return Ok(());
}
if i.slots
.get(&run)
.is_some_and(|slot| slot.machine.cancellation_covers_current_operations())
{
if let Some(t) = token {
let tail = i.writer.last_lsn();
i.core.add_output(tail, Output::Ack(t));
i.on_commit();
}
return Ok(());
}
match i.slots.get_mut(&run) {
Some(slot) => {
slot.machine
.request_cancel_by(Arc::from(reason.as_bytes()), by);
if let Some(t) = token {
i.cancel_tokens.entry(run).or_default().push(t);
}
i.wake(run);
}
None => {
i.by = by;
let lsn = i.end_run(run, Status::Cancelled, reason);
i.by = None;
let lsn = lsn?;
if let (Some(t), Some(lsn)) = (token, lsn) {
i.core.add_output(lsn, Output::Ack(t));
}
}
}
}
self.instantiate_pending();
Ok(())
}
pub fn resume(&self, id: &str, token: Option<u64>) -> Result<(), Error> {
{
let mut i = self.inner.borrow_mut();
let by = i.by.take();
i.check_alive()?;
let run = match i.core.find(id) {
Some(Found::Loaded(r) | Found::Parked(r)) => r,
None => return Err(Error::NotFound(id.to_owned())),
};
i.ensure_loaded(run)?;
if i.core.get(run).is_some_and(|l| l.is_done()) {
return Err(Error::Rejected(format!(
"workflow '{id}' has finished: nothing to resume"
)));
}
let Some(slot) = i.slots.get_mut(&run) else {
return Err(Error::Rejected(format!(
"workflow '{id}' is not paused (no code for it runs here)"
)));
};
let records = slot.machine.resume(by);
if records.is_empty() {
return Err(Error::Rejected(format!(
"workflow '{id}' is not paused: no step of it ran out of retries"
)));
}
let mut lsn = i.commit;
for rec in &records {
lsn = i.append(rec)?;
}
if let Some(t) = token {
i.core.add_output(lsn, Output::Ack(t));
}
i.wake(run);
}
self.instantiate_pending();
Ok(())
}
pub fn terminate(&self, id: &str, reason: &str, token: Option<u64>) -> Result<(), Error> {
let mut i = self.inner.borrow_mut();
let by = i.by.take();
i.check_alive()?;
let run = match i.core.find(id) {
Some(Found::Loaded(r) | Found::Parked(r)) => r,
None => return Err(Error::NotFound(id.to_owned())),
};
let idx = i.slots.idx(run);
i.by = by;
let lsn = i.end_run(run, Status::Terminated, reason);
i.by = None;
let lsn = lsn?;
let ended = idx.filter(|_| !i.slots.contains_key(&run));
drop(i);
if let Some(idx) = ended {
let fut = self
.futures
.borrow_mut()
.get_mut(idx as usize)
.and_then(Option::take);
drop(fut);
}
let mut i = self.inner.borrow_mut();
let lsn = lsn.unwrap_or_else(|| i.writer.last_lsn());
if let Some(t) = token {
i.core.add_output(lsn, Output::Ack(t));
}
i.on_commit();
Ok(())
}
pub fn message(&self, msg: Message, from: Option<(u32, u64)>) -> Result<(), Error> {
self.inner.borrow().check_alive()?;
let token = from.map(|f| self.inner.borrow_mut().msg_token(f));
match msg {
Message::ChildStart {
parent,
wtype,
id,
flags,
input,
..
} if flags & policy::OBJECT != 0 => {
self.object_call_message(parent, wtype, id, flags, input, token, from)
}
Message::ChildStart {
parent,
wtype,
id,
flags,
deadline_ms,
input,
trace,
} => {
let detached = flags & policy::DETACHED != 0;
let req = StartRequest {
id,
wtype,
version: None,
input,
codec: 1,
deadline_ms,
parent: Some(parent),
policy: if self.inner.borrow().cfg.sabotage == Sabotage::NoAttach {
policy::RUNNING_USE_EXISTING
} else {
policy::ATTACH
} | (flags & (policy::CLOSE_MASK | policy::DETACHED)),
update: None,
sealed: true,
token,
trace,
};
match self.start(req) {
Ok(_) => Ok(()),
Err(e @ Error::UnknownWorkflow(_)) => {
let mut i = self.inner.borrow_mut();
if !detached {
i.send_out(
parent.shard,
Message::ChildResult {
parent,
status: Status::Failed,
payload: e.to_failure().to_payload(),
},
);
}
if let Some(f) = from {
i.ack_at_tail(f);
}
i.on_commit();
Ok(())
}
Err(e) => Err(e),
}
}
Message::ChildResult {
parent,
status,
payload,
} => {
{
let mut i = self.inner.borrow_mut();
i.ensure_loaded(parent.run)?;
i.outbox.retain(|_, o| {
!matches!(&o.msg, Message::ChildStart { parent: p, .. } if *p == parent)
});
let run = parent.run;
let alive = i.core.get(run).is_some_and(|l| !l.is_done());
let resolved = match i.slots.get(&run) {
Some(slot) => {
parent.seq < slot.machine.next_seq()
&& !slot.machine.is_live_pending(parent.seq)
}
None => false,
};
if alive && !resolved {
i.complete(run, parent.seq, Completion::Child { status, payload });
if let Some(f) = from {
i.child_acks.entry((run, parent.seq)).or_default().push(f);
}
} else if let Some(f) = from {
i.ack_at_tail(f);
}
if let Some(t) = token {
i.msg_tokens.remove(&t);
}
i.on_commit();
}
self.instantiate_pending();
Ok(())
}
Message::Signal {
target,
name,
payload,
dedup,
} => match self.signal_sealed(&target, &name, payload, dedup, token) {
Ok(()) => Ok(()),
Err(Error::NotFound(_) | Error::Failed(_)) => {
let mut i = self.inner.borrow_mut();
if let Some(t) = token {
let tail = i.writer.last_lsn();
i.core.add_output(tail, Output::Ack(t));
}
i.on_commit();
Ok(())
}
Err(e) => Err(e),
},
Message::Close {
target,
wtype,
parent,
terminate,
reason,
} => {
let applies = {
let mut i = self.inner.borrow_mut();
match i.core.find(&target) {
Some(Found::Loaded(r) | Found::Parked(r)) => {
i.ensure_loaded(r)?;
i.core.get(r).is_some_and(|l| {
!l.is_done() && (parent.is_none() || l.parent == parent)
})
}
None => {
if let Some(p) = parent.filter(|_| !wtype.is_empty()) {
let status = if terminate {
Status::Terminated
} else {
Status::Cancelled
};
let lsn = i.start_closed(&target, &wtype, p, status, &reason)?;
if let Some(t) = token {
i.core.add_output(lsn, Output::Ack(t));
}
i.on_commit();
return Ok(());
}
false
}
}
};
let reason = String::from_utf8_lossy(&reason).into_owned();
if applies {
if terminate {
self.terminate(&target, &reason, token)
} else {
self.cancel(&target, &reason, token)
}
} else {
let mut i = self.inner.borrow_mut();
if let Some(t) = token {
let tail = i.writer.last_lsn();
i.core.add_output(tail, Output::Ack(t));
}
i.on_commit();
Ok(())
}
}
}
}
pub fn take_outgoing(&self) -> Vec<Outgoing> {
let mut i = self.inner.borrow_mut();
if i.failed.is_some() {
return Vec::new();
}
if i.cfg.replay_only {
i.outbox.clear();
return Vec::new();
}
let now = i.now();
let mut out = Vec::new();
for o in i.outbox.values_mut() {
if o.sent_ms == 0 || now.saturating_sub(o.sent_ms) >= RESEND_MS {
o.sent_ms = now.max(1);
out.push(o.clone());
}
}
out
}
pub fn take_message_acks(&self) -> Vec<(u32, u64)> {
std::mem::take(&mut self.inner.borrow_mut().message_acks)
}
pub fn outbox_ack(&self, id: u64) {
self.inner.borrow_mut().outbox.remove(&id);
}
pub fn outbox_len(&self) -> usize {
self.inner.borrow().outbox.len()
}
pub fn summary(&self, run: RunId) -> Option<RunSummary> {
let mut i = self.inner.borrow_mut();
let shard = i.cfg.shard;
if let Some(l) = i.core.get(run) {
let status = if let Some((s, _)) = l.done {
s.as_str().to_owned()
} else if i.stuck.contains_key(&run) {
"stuck".into()
} else if let Some(slot) = i.slots.get(&run) {
if slot.machine.is_paused() {
"paused".into()
} else if i.parkable(run) {
"suspended".into()
} else {
"running".into()
}
} else {
"waiting_for_code".into()
};
return Some(RunSummary {
shard,
run,
id: l.id.to_string(),
wtype: l.wtype.to_string(),
version: l.version,
status,
started_ms: l.started_ms,
finished_ms: l.done.map(|d| d.1),
error: i.stuck.get(&run).cloned().or_else(|| {
l.done
.filter(|d| d.0 != Status::Ok && !d.0.is_handover())
.and(l.result.as_ref())
.map(|p| String::from_utf8_lossy(p).into_owned())
}),
attributes: merge_attributes(&l.attrs),
object: l.policy & policy::OBJECT != 0,
});
}
let p = i.core.parked(run)?;
let recs = i.blob_records(p.blob()).ok()?;
let (id, wtype, version, started_ms, object) = match recs.first()? {
Record::Start {
id,
wtype,
version,
started_ms,
policy: pol,
..
} => (
id.to_string(),
wtype.to_string(),
*version,
*started_ms,
pol & policy::OBJECT != 0,
),
_ => return None,
};
let finished = recs.iter().find_map(|r| match r {
Record::Done { finished_ms, .. } => Some(*finished_ms),
_ => None,
});
let error = recs.iter().find_map(|r| match r {
Record::Done {
status, payload, ..
} if *status != Status::Ok && !status.is_handover() => {
Some(String::from_utf8_lossy(payload).into_owned())
}
_ => None,
});
let attrs: Vec<Payload> = recs
.iter()
.filter_map(|r| match r {
Record::Attributes { payload, .. } => Some(payload.clone()),
_ => None,
})
.collect();
Some(RunSummary {
shard,
run,
id,
wtype,
version,
status: p.done().map(|s| s.as_str().to_owned()).unwrap_or_else(|| {
if journal_paused(&recs) {
"paused".into()
} else {
"parked".into()
}
}),
started_ms,
finished_ms: p.done().and(finished),
error,
attributes: merge_attributes(&attrs),
object,
})
}
pub fn list(&self, limit: usize) -> Vec<RunSummary> {
self.newest_matching(limit, |_| true)
}
fn newest_matching(&self, limit: usize, keep: impl Fn(&RunSummary) -> bool) -> Vec<RunSummary> {
let mut out = Vec::new();
if limit == 0 {
return out;
}
let mut cursor = self.inner.borrow().core.newest_first();
loop {
let want = (limit - out.len()).clamp(64, 4096);
let batch = cursor.next_batch(&self.inner.borrow().core, want);
if batch.is_empty() {
return out;
}
for r in batch {
if let Some(s) = self.summary(r)
&& keep(&s)
{
out.push(s);
if out.len() >= limit {
return out;
}
}
}
}
}
pub fn inspect(&self, id: &str) -> Result<Option<RunDetail>, Error> {
let Some(run) = self.find(id) else {
return Ok(None);
};
let Some(summary) = self.summary(run) else {
return Ok(None);
};
let mut i = self.inner.borrow_mut();
let journal = i.journal_of(run)?;
let object = i.object_detail(run, &journal).map(Box::new);
Ok(Some(RunDetail {
summary,
journal,
object,
}))
}
pub fn list_where(
&self,
filter: &crate::visibility::ListFilter,
limit: usize,
) -> Vec<RunSummary> {
if limit == 0 {
return Vec::new();
}
if !filter.is_empty() {
let window = limit.saturating_add(64).min(4096);
let candidates = {
self.inner
.borrow_mut()
.env
.visibility_candidates(filter, window)
};
if let Some(c) = candidates {
let current = {
let i = self.inner.borrow();
if c.generation != i.env.store_generation() {
None
} else {
let mut runs: Vec<_> = i
.core
.loaded_runs()
.map(|(r, _)| r)
.chain(
i.core
.parked_runs()
.filter(|(_, p)| p.blob().offset >= c.length)
.map(|(r, _)| r),
)
.collect();
runs.extend(c.runs);
runs.sort_unstable_by(|a, b| b.cmp(a));
runs.dedup();
Some(runs)
}
};
if let Some(runs) = current {
let result: Vec<_> = runs
.into_iter()
.filter_map(|r| self.summary(r))
.filter(|s| filter.matches(s))
.take(limit)
.collect();
if !c.truncated || (c.newest && result.len() == limit) {
return result;
}
}
}
}
self.newest_matching(limit, |s| filter.matches(s))
}
pub fn update(
&self,
id: &str,
name: &str,
update_id: &str,
payload: Payload,
token: Option<u64>,
) -> Result<(), Error> {
{
let mut i = self.inner.borrow_mut();
i.check_alive()?;
let run = match i.core.find(id) {
Some(Found::Loaded(r) | Found::Parked(r)) => r,
None => return Err(Error::NotFound(id.to_owned())),
};
i.ensure_loaded(run)?;
let uid = if update_id.is_empty() {
i.fresh_update_id()
} else {
update_id.to_owned()
};
i.offer_update(run, name, &uid, payload, token)?;
i.on_commit();
}
self.instantiate_pending();
Ok(())
}
pub fn decide_update(
&self,
token: u64,
decision: Result<(), String>,
) -> Result<Option<u64>, Error> {
let mut i = self.inner.borrow_mut();
let r = i.decide_update(token, decision);
i.on_commit();
r
}
pub fn query(&self, id: &str, name: &str, arg: Payload, token: u64) -> Result<(), Error> {
let run = {
let mut i = self.inner.borrow_mut();
i.check_alive()?;
let run = match i.core.find(id) {
Some(Found::Loaded(r) | Found::Parked(r)) => r,
None => return Err(Error::NotFound(id.to_owned())),
};
i.ensure_loaded(run)?;
run
};
self.instantiate_pending();
self.drive(run);
self.try_query(run, name, arg, token);
Ok(())
}
fn try_query(&self, run: RunId, name: &str, arg: Payload, token: u64) {
let refuse = |i: &mut Inner<E>, at: Lsn, msg: &str| {
i.core.add_output(
at,
Output::Refuse {
token,
message: Arc::from(msg.as_bytes()),
},
);
i.on_commit();
};
let lsn = {
let mut i = self.inner.borrow_mut();
let Some(l) = i.core.get(run) else {
let tail = i.writer.last_lsn();
refuse(&mut i, tail, "the workflow is no longer loaded");
return;
};
let lsn = l.last_lsn;
if l.is_done() {
refuse(&mut i, lsn, "the workflow has already finished");
return;
}
match i.slots.get(&run) {
None => {
refuse(
&mut i,
lsn,
"the workflow is not running code (it is stuck or its code is not registered)",
);
return;
}
Some(s) if s.machine.is_replaying() => {
drop(i);
self.deferred
.borrow_mut()
.push((run, name.to_owned(), arg, token));
return;
}
Some(_) => lsn,
}
};
let handler = self.handlers.borrow().get(&run).and_then(|h| h.query(name));
let answer = match handler {
None => Err(format!("no handler for query '{name}'")),
Some(f) => f(arg),
};
let mut i = self.inner.borrow_mut();
if let (Sabotage::EarlyAnswers, Ok(v)) = (i.cfg.sabotage, &answer) {
i.released.push(Output::Reply {
token,
status: Status::Ok,
payload: v.clone(),
});
return;
}
match answer {
Ok(v) => {
i.core.add_output(
lsn,
Output::Reply {
token,
status: Status::Ok,
payload: v,
},
);
i.on_commit();
}
Err(m) => refuse(&mut i, lsn, &m),
}
}
fn retry_deferred(&self) -> bool {
if self.deferred.borrow().is_empty() {
return false;
}
let pending = std::mem::take(&mut *self.deferred.borrow_mut());
if pending.is_empty() {
return false;
}
for (run, name, arg, token) in pending {
self.try_query(run, &name, arg, token);
}
true
}
pub fn stream_read(
&self,
id: &str,
run: Option<RunId>,
name: &str,
from: u64,
wait_ms: u64,
token: u64,
) -> Result<(), Error> {
let mut i = self.inner.borrow_mut();
i.check_alive()?;
let run = i.stream_run(id, run)?;
let mut r = i.read_stream(run, name, from)?;
if r.id.is_none() {
r.id = i.scan_stream(run, name, 0, 0, 0)?.id;
if let Some(rs) = i.streams.get_mut(&run) {
rs.id = r.id.clone();
}
}
if r.id.as_deref().is_some_and(|x| x != id) {
return Err(Error::NotFound(id.to_owned()));
}
if !r.chunks.is_empty() || r.end.is_some() || wait_ms == 0 {
let payload = i.stream_answer(&r)?;
i.released.push(Output::Reply {
token,
status: Status::Ok,
payload,
});
return Ok(());
}
if let Some(lsn) = r.wait {
i.stream_due.push(Reverse((lsn, run)));
}
let deadline_ms = i.now().saturating_add(wait_ms);
i.stream_wait_until = i.stream_wait_until.min(deadline_ms);
i.stream_waiters.entry(run).or_default().push(StreamWaiter {
name: Arc::from(name),
from,
token,
deadline_ms,
});
Ok(())
}
pub fn stream_readers(&self) -> usize {
self.inner
.borrow()
.stream_waiters
.values()
.map(Vec::len)
.sum()
}
pub fn query_point(&self, id: &str) -> Result<(RunId, Lsn), Error> {
let mut i = self.inner.borrow_mut();
let run = match i.core.find(id) {
Some(Found::Loaded(r) | Found::Parked(r)) => r,
None => return Err(Error::NotFound(id.to_owned())),
};
i.ensure_loaded(run)?;
match i.core.get(run) {
Some(l) if l.is_done() => {
Err(Error::Rejected("the workflow has already finished".into()))
}
Some(l) => Ok((run, l.last_lsn)),
None => Err(Error::NotFound(id.to_owned())),
}
}
pub fn ack_at(&self, lsn: Lsn, token: u64) {
let mut i = self.inner.borrow_mut();
i.core.add_output(lsn, Output::Ack(token));
i.on_commit();
}
pub fn complete(
&self,
token: &str,
status: Status,
payload: Payload,
reply: Option<u64>,
) -> Result<(), Error> {
{
let mut i = self.inner.borrow_mut();
i.check_alive()?;
let refuse = |i: &mut Inner<E>, msg: &str| {
if let Some(t) = reply {
let tail = i.writer.last_lsn();
i.core.add_output(
tail,
Output::Refuse {
token: t,
message: Arc::from(msg.as_bytes()),
},
);
i.on_commit();
}
};
let Some((id, run, seq, attempt)) = parse_token(token) else {
return Err(Error::Rejected("not a completion token".into()));
};
match i.core.find(&id) {
Some(Found::Loaded(r) | Found::Parked(r)) if r == run => {}
_ => {
refuse(&mut i, "the workflow run this token belongs to is gone");
return Ok(());
}
}
i.ensure_loaded(run)?;
let marker = i
.core
.get(run)
.filter(|l| !l.is_done())
.and_then(|l| l.awaiting.get(&seq).copied());
let live = i.slots.get(&run).and_then(|s| s.machine.await_attempt(seq));
if marker != Some(attempt) && live != Some(attempt) {
refuse(
&mut i,
"stale completion token: the step finished or moved on to another attempt",
);
return Ok(());
}
let payload = if status == Status::Ok {
i.seal(payload)?
} else {
payload
};
if let Some(t) = reply {
i.completion_acks.entry((run, seq)).or_default().push(t);
}
i.complete(
run,
seq,
Completion::Completed {
attempt,
status,
payload,
},
);
i.on_commit();
}
self.instantiate_pending();
Ok(())
}
pub fn reset(
&self,
id: &str,
before_step: &str,
reapply_signals: bool,
version: Option<u32>,
token: Option<u64>,
) -> Result<RunId, Error> {
let new = {
let mut i = self.inner.borrow_mut();
let by = i.by.take();
i.check_alive()?;
let run = match i.core.find(id) {
Some(Found::Loaded(r) | Found::Parked(r)) => r,
None => return Err(Error::NotFound(id.to_owned())),
};
i.ensure_loaded(run)?;
let journal = i.journal_of(run)?;
let target = journal.iter().find_map(|r| match r {
Record::Step { seq, name, .. }
| Record::Attempt { seq, name, .. }
| Record::Awaiting { seq, name, .. }
if &**name == before_step =>
{
Some(*seq)
}
_ => None,
});
let Some(target) = target else {
let thin = journal.iter().any(|r| matches!(r, Record::Done { .. }))
&& journal.iter().all(|r| {
matches!(
r,
Record::Start { .. }
| Record::Done { .. }
| Record::Attributes { .. }
| Record::Chunk { .. }
| Record::StreamEnd { .. }
)
});
return Err(Error::Rejected(if thin {
format!(
"run of '{id}' finished and only its result was kept; open the engine \
with keep_history to reset finished runs"
)
} else {
format!("the run of '{id}' has no step named '{before_step}'")
}));
};
let cut = journal
.iter()
.skip(1)
.position(|r| r.seq() == Some(target))
.map(|p| p + 1)
.unwrap_or(journal.len());
let Some(Record::Start {
id: rid,
wtype,
version: old_version,
codec,
started_ms,
deadline_ms,
parent,
policy: pol,
update,
input,
meta,
..
}) = journal.first().cloned()
else {
return Err(Error::Io(format!("run {run}: journal has no start")));
};
let mut copies: Vec<RecordBuf> = journal[1..cut]
.iter()
.filter(|r| !matches!(r, Record::Done { .. }))
.cloned()
.collect();
if reapply_signals {
copies.extend(
journal[cut..]
.iter()
.filter(|r| matches!(r, Record::Signal { .. }))
.cloned(),
);
}
let new = i.core.allocate_run();
let start: RecordBuf = Record::Start {
run: new,
id: rid,
wtype,
version: version.unwrap_or(old_version),
codec,
started_ms,
deadline_ms,
parent,
policy: pol,
from: Some(Origin {
run,
cause: origin::RESET,
}),
copies: copies.len() as u32,
update,
input,
meta: StartMeta::of(by.clone(), meta.and_then(|m| m.trace)),
};
let running = i.core.get(run).is_some_and(|l| !l.is_done());
let now = i.now();
let mut lsn = if running {
let mut payload = Vec::new();
encode_canonical(&start, &mut payload);
i.append(&Record::Done {
run,
status: Status::Reset,
finished_ms: now,
payload: Arc::from(payload),
by,
})?
} else {
i.append(&start)?
};
if let Some(slot) = i.remove_slot(run)
&& slot.foreign
{
i.foreign.push_back(ForeignEvent::Ended { run });
}
i.release_waiting_requests(run);
for c in &copies {
lsn = i.append(&c.with_run(new))?;
}
let mut journal = i.journal_of(new)?;
i.prepare(new, &mut journal);
i.core.reoffer(new);
if let Some(t) = token {
i.core.add_output(lsn, Output::Ack(t));
}
i.metrics.runs_started += 1;
i.on_commit();
new
};
self.instantiate_pending();
Ok(new)
}
pub fn next_timer_ms(&self) -> Option<u64> {
let i = self.inner.borrow();
let mut next = i.core.next_timer();
if let Some(Reverse((at, _))) = i.timers.peek() {
next = Some(next.map_or(*at, |n| n.min(*at)));
}
next
}
pub fn import(&self, journal: &[RecordBuf]) -> Result<RunId, Error> {
let new = {
let mut i = self.inner.borrow_mut();
let Some(Record::Start { .. }) = journal.first() else {
return Err(Error::Codec(
"the journal does not begin with a start".into(),
));
};
let copies: Vec<RecordBuf> = journal[1..]
.iter()
.filter(|r| !matches!(r, Record::Done { .. }))
.cloned()
.collect();
let new = i.core.allocate_run();
let mut start = journal[0].with_run(new);
if let Record::Start {
copies: c, from, ..
} = &mut start
{
*c = copies.len() as u32;
*from = None;
}
i.append(&start)?;
for c in &copies {
i.append(&c.with_run(new))?;
}
let mut journal = i.journal_of(new)?;
i.prepare(new, &mut journal);
i.core.reoffer(new);
new
};
self.instantiate_pending();
Ok(new)
}
pub fn stuck_reason(&self, run: RunId) -> Option<String> {
self.inner.borrow().stuck.get(&run).cloned()
}
pub fn code_state(&self, run: RunId) -> Option<(bool, bool)> {
let i = self.inner.borrow();
let s = i.slots.get(&run)?;
Some((s.machine.is_replaying(), s.machine.has_live_pending()))
}
pub(crate) fn is_foreign_slot(&self, run: RunId) -> bool {
self.inner
.borrow()
.slots
.get(&run)
.is_some_and(|slot| slot.foreign)
}
pub fn quiescent(&self, run: RunId) -> Result<Option<Delivery>, Error> {
self.inner.borrow_mut().deliver(run)
}
pub fn foreign_finish(&self, run: RunId, outcome: RunOutcome) {
self.inner.borrow_mut().finish(run, outcome);
}
pub fn foreign_stuck(&self, run: RunId, message: String) {
self.inner.borrow_mut().make_stuck(run, message);
}
pub(crate) fn flush_solo_gate(
&self,
run: RunId,
gate: Lsn,
not_before: u64,
) -> Result<bool, Error> {
let mut i = self.inner.borrow_mut();
i.check_alive()?;
if (i.slots.len() != 1 && (i.has_undelivered() || !i.foreign.is_empty()))
|| !i.slots.get(&run).is_some_and(|s| s.foreign)
|| i.cfg.replay_only
|| !i.cfg.inline_flush
|| i.in_flight.is_some()
|| i.env.last_flush_us() > i.cfg.inline_max_us
|| !i.env.supports_inline()
|| gate <= i.commit
|| gate > i.writer.last_lsn()
|| not_before > i.now()
{
return Ok(false);
}
i.maybe_flush(true);
i.check_alive()?;
Ok(i.commit >= gate)
}
pub fn is_idle(&self) -> bool {
let i = self.inner.borrow();
i.slots.is_empty() && i.in_flight.is_none() && !i.writer.has_pending()
}
pub(crate) fn submit_flush(&self) -> Option<Lsn> {
let mut i = self.inner.borrow_mut();
i.maybe_flush(false);
i.in_flight
}
pub(crate) fn poll_flush(&self) -> bool {
let mut i = self.inner.borrow_mut();
i.in_flight.is_some() && i.poll_io()
}
pub(crate) fn last_flush_us(&self) -> u64 {
self.inner.borrow().env.last_flush_us()
}
pub(crate) fn set_inline_flush(&self, on: bool) {
self.inner.borrow_mut().cfg.inline_flush = on;
}
pub(crate) fn set_load_room(&self, room: usize) {
self.inner.borrow_mut().load_room = room;
}
pub(crate) fn foreign_len(&self) -> usize {
self.inner.borrow().foreign.len()
}
pub(crate) fn quiet(&self, writes: bool) -> bool {
let mut i = self.inner.borrow_mut();
!i.has_undelivered()
&& i.foreign.is_empty()
&& i.instantiate.is_empty()
&& !i.core.has_events()
&& (!writes || (i.in_flight.is_none() && !i.writer.has_pending()))
}
pub fn loaded(&self) -> usize {
self.inner.borrow().core.loaded_len()
}
pub fn parked(&self) -> usize {
self.inner.borrow().core.parked_len()
}
pub fn close(&self) -> Result<(), Error> {
self.flush_all()?;
{
let mut i = self.inner.borrow_mut();
if !i.cfg.replay_only
&& i.cfg.checkpoint_on_close
&& !i.ckpt_running
&& i.ckpt_pending.is_none()
&& (i.checkpoint_needed() || i.env.checkpoint_wanted_on_close())
{
let now = i.now();
i.start_checkpoint(now);
}
}
loop {
self.tick();
let done = {
let i = self.inner.borrow();
i.check_alive()?;
!i.ckpt_running
&& i.ckpt_pending.is_none()
&& !i.writer.has_pending()
&& i.in_flight.is_none()
};
if done {
let mut i = self.inner.borrow_mut();
let due = i.due();
i.env.report_due(due, true);
return Ok(());
}
std::thread::sleep(std::time::Duration::from_micros(100));
}
}
pub fn flush_all(&self) -> Result<(), Error> {
loop {
self.tick();
let (pending, in_flight) = {
let i = self.inner.borrow();
i.check_alive()?;
(i.writer.has_pending(), i.in_flight.is_some())
};
if !pending && !in_flight {
return Ok(());
}
std::thread::sleep(std::time::Duration::from_micros(50));
}
}
}
fn drive_flush_every() -> usize {
static EVERY: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
*EVERY.get_or_init(|| {
std::env::var("FUSEE_DRIVE_FLUSH_EVERY")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(0)
})
}
fn admit_round() -> usize {
static LIMIT: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
*LIMIT.get_or_init(|| {
std::env::var("FUSEE_ADMIT_ROUND")
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(2048)
})
}