Struct ExecutionStore 

pub struct ExecutionStore { /* private fields */ }

The executions of one process.

Implementations

impl ExecutionStore

pub fn new(limits: Limits) -> Self

A store with these limits.

pub fn limits(&self) -> Limits

The limits this store was built with.

pub fn create( &self, id: ExecutionId, capability: &str, title: &str, input: Value, cancellable: bool, actor: Actor, repository: RepositoryRef, ) -> Arc<AtomicBool>

Record a new execution and publish its first event. Returns its cancellation token.

The execution is created queued: what runs it decides when it starts.

pub fn adopt( &self, id: ExecutionId, seed: Seed<'_>, pid: u32, ) -> Result<(), AdoptError>

Hold an execution another process runs and reports here, so that this process’s listing, snapshot, history and live channel show it as they show their own.

Nothing runs here: what the record says is what the reporter says, through Self::report. Adopting an id already held as a reported execution changes nothing, so a reporter may send its description with every report. An id this process runs itself is refused.

use majordomus_cli::execution::{Actor, EventPayload, ExecutionId, ExecutionState};
use majordomus_cli::execution::{ExecutionStore, Limits, RepositoryRef, Seed};
let store = ExecutionStore::new(Limits::default());
let seed = || Seed {
    capability: "executions.demonstrate",
    title: "Demonstrate an execution",
    input: serde_json::json!({ "steps": 1 }),
    cancellable: true,
    actor: Actor::reported(4242),
    repository: RepositoryRef { name: "r".into(), id: "id".into(), branch: None },
};
let id = ExecutionId::fresh();
store.adopt(id.clone(), seed(), 4242).unwrap();
store.adopt(id.clone(), seed(), 4242).unwrap(); // again: nothing changes
assert!(store.is_reported(&id));
assert_eq!(store.get(&id).unwrap().event_count, 1, "one `created` event");

pub fn report( &self, id: &ExecutionId, events: impl IntoIterator<Item = (u64, EventPayload)>, ) -> Option<ReportOutcome>

Apply what the process running an execution reports, in its own sequence: each event at most once and in order, so a report sent twice or cut short is harmless.

An event after a gap is not applied; the answer’s through says where the reporter is to resume. An event this store’s state machine refuses is passed over rather than believed, as one from a handler of this process would be. None when the id is not a reported execution of this store.

use majordomus_cli::execution::{Actor, EventPayload, ExecutionId, ExecutionState};
use majordomus_cli::execution::{ExecutionStore, Limits, RepositoryRef, Seed};
let store = ExecutionStore::new(Limits::default());
let seed = || Seed {
    capability: "executions.demonstrate",
    title: "Demonstrate an execution",
    input: serde_json::json!({ "steps": 1 }),
    cancellable: true,
    actor: Actor::reported(4242),
    repository: RepositoryRef { name: "r".into(), id: "id".into(), branch: None },
};
let id = ExecutionId::fresh();
store.adopt(id.clone(), seed(), 4242).unwrap();
let started = store.report(&id, vec![(2, EventPayload::Started)]).unwrap();
assert_eq!((started.applied, started.through), (1, 2));
// sent twice: nothing more is applied
assert_eq!(store.report(&id, vec![(2, EventPayload::Started)]).unwrap().applied, 0);
// after a gap: nothing is applied, and the answer says where to resume
let done = EventPayload::Completed { output: serde_json::json!(1) };
assert_eq!(store.report(&id, vec![(4, done)]).unwrap().through, 2);
assert_eq!(store.get(&id).unwrap().state, ExecutionState::Running);
assert!(store.report(&ExecutionId::fresh(), Vec::new()).is_none(), "not a reported one");

pub fn abandon_silent(&self, silence: Duration) -> Vec<ExecutionId>

Fail every reported execution that is not finished and has not been heard from for longer than silence, and answer which. The process that ran it was killed or lost its way to this one; what it did afterwards is unknown here, which the error says.

use majordomus_cli::execution::{Actor, EventPayload, ExecutionId, ExecutionState};
use majordomus_cli::execution::{ExecutionStore, Limits, RepositoryRef, Seed};
let store = ExecutionStore::new(Limits::default());
let seed = || Seed {
    capability: "executions.demonstrate",
    title: "Demonstrate an execution",
    input: serde_json::json!({ "steps": 1 }),
    cancellable: true,
    actor: Actor::reported(4242),
    repository: RepositoryRef { name: "r".into(), id: "id".into(), branch: None },
};
use std::time::Duration;
let id = ExecutionId::fresh();
store.adopt(id.clone(), seed(), 4242).unwrap();
store.report(&id, vec![(2, EventPayload::Started)]).unwrap();
assert!(store.abandon_silent(Duration::from_secs(30)).is_empty(), "heard just now");
std::thread::sleep(Duration::from_millis(20));
assert_eq!(store.abandon_silent(Duration::from_millis(5)), vec![id.clone()]);
let failed = store.get(&id).unwrap();
assert_eq!(failed.state, ExecutionState::Failed);
assert_eq!(failed.error.unwrap().code, "unavailable");

pub fn is_reported(&self, id: &ExecutionId) -> bool

Is this an execution another process runs and reports here?

use majordomus_cli::execution::{Actor, EventPayload, ExecutionId, ExecutionState};
use majordomus_cli::execution::{ExecutionStore, Limits, RepositoryRef, Seed};
let store = ExecutionStore::new(Limits::default());
let seed = || Seed {
    capability: "executions.demonstrate",
    title: "Demonstrate an execution",
    input: serde_json::json!({ "steps": 1 }),
    cancellable: true,
    actor: Actor::reported(4242),
    repository: RepositoryRef { name: "r".into(), id: "id".into(), branch: None },
};
let reported = ExecutionId::fresh();
store.adopt(reported.clone(), seed(), 4242).unwrap();
assert!(store.is_reported(&reported));
assert!(!store.is_reported(&ExecutionId::fresh()), "nothing is held under that id");

pub fn publish(&self, id: &ExecutionId, payload: EventPayload) -> Option<u64>

Apply an event to an execution: move its state, update its snapshot, retain it, and hand it to every subscriber that wants it.

Returns the sequence the event was given, or None when there is no such execution or the payload asks for a transition the state machine does not allow — which is how a second completed after a cancelled is refused rather than believed.

pub fn get(&self, id: &ExecutionId) -> Option<Execution>

The snapshot of one execution.

pub fn list( &self, state: Option<ExecutionState>, capability: Option<&str>, limit: usize, ) -> Vec<Execution>

Every execution this process remembers, newest first, optionally narrowed.

pub fn len(&self) -> usize

How many executions this store remembers.

pub fn is_empty(&self) -> bool

Does it remember none?

pub fn count(&self, state: ExecutionState) -> usize

How many executions are in a state right now.

pub fn active(&self) -> usize

How many are not finished: running, starting, cancelling or queued.

pub fn events( &self, id: &ExecutionId, after: u64, limit: usize, ) -> Option<EventPage>

A page of one execution’s retained history, after a sequence.

pub fn token(&self, id: &ExecutionId) -> Option<Arc<AtomicBool>>

The cancellation token of an execution, for the worker that runs it.

pub fn request_cancel(&self, id: &ExecutionId, by: &str) -> CancelOutcome

Ask an execution to stop. Publishes execution.cancelling and sets the token; the handler decides when it stops, and a handler that never looks at its token will finish normally, which the final state will say.

Both happen under one hold of the lock. A worker reads the token without the lock, but a handler that stops publishes execution.cancelled, which needs the lock that cancelling is recorded under, so a handler that sees the token cannot publish until cancelling is recorded, whichever of the two is done first inside. Recording the state before the token is extra ordering on top of that: on its own it would hold even were the token set after the lock is let go. What neither may be is the token set in one hold and cancelling published in the next: a handler looking in between publishes cancelled from running — a move the state machine refuses — and the cancelling that follows is the last thing the execution ever says.

pub fn subscribe(&self, filter: Filter) -> (u64, Receiver<Delivery>)

Listen. The receiver is dropped to unsubscribe; the store forgets a subscriber the first time a send finds it gone, so nothing has to be cleaned up by hand.

pub fn resubscribe(&self, subscriber: u64, filter: Filter) -> bool

Change what a subscriber wants, without reconnecting.

pub fn unsubscribe(&self, subscriber: u64)

Stop listening.

pub fn subscribers(&self) -> usize

How many subscribers this store is feeding.