Struct MemoryHistory 

pub struct MemoryHistory { /* private fields */ }

Available on crate feature serve only.

Implementations

impl MemoryHistory

pub fn new(idem_retention: Duration) -> Self

Trait Implementations

impl RunHistory for MemoryHistory

fn claim_idempotency<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, key: &'life1 str, fingerprint: &'life2 str, run_id: &'life3 str, window: Duration, ) -> Pin<Box<dyn Future<Output = Result<Claim, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Atomically claim key for run_id (or report a replay/conflict). A prior claim older than window is treated as expired and re-claimable.

fn upsert<'life0, 'life1, 'async_trait>( &'life0 self, rec: &'life1 RunRecord, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Insert or replace a run record.

fn get<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<RunRecord>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn list<'life0, 'life1, 'async_trait>( &'life0 self, filter: &'life1 ListFilter, ) -> Pin<Box<dyn Future<Output = Result<ListPage, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

fn delete<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<DeleteOutcome, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Delete a terminal run. Non-terminal → StillRunning (caller maps to 409).

fn purge_expired<'life0, 'async_trait>( &'life0 self, retain_for: Duration, ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Drop terminal records finished longer than retain_for ago. Returns the number removed.

fn record_audit<'life0, 'life1, 'async_trait>( &'life0 self, entry: &'life1 AuditEntry, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Append one audit record. Best-effort but visible: the caller logs a warning on failure (audit writes must never silently vanish, and must never fail the underlying action). Default: no-op — overridden by the memory + SQL backends.

fn list_audit<'life0, 'life1, 'async_trait>( &'life0 self, filter: &'life1 AuditFilter, ) -> Pin<Box<dyn Future<Output = Result<Vec<AuditEntry>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Most-recent audit records matching filter, newest first. Default: empty.

fn recover_orphans<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Mark non-terminal records whose owning instance’s lease has expired as failed (instance-fenced orphan recovery — never touches a live peer’s heartbeated runs, #146 H7). Returns the number recovered. The memory backend has nothing to recover (returns 0).

fn cancel_pending<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<bool, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Cancel a still-Pending (unclaimed) run directly. Returns true if it was pending and is now Cancelled; false if it had already been claimed (the caller should fall back to request_cancel). Default: false.

fn catalog_record<'life0, 'life1, 'async_trait>( &'life0 self, update: &'life1 CatalogUpdate, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Fold one run’s catalog update (two dataset observations + the lineage edge between them) into the store. Idempotent-ish last-write-wins per dataset/edge, so concurrent cluster instances converge. Default: no-op.

fn catalog_list_datasets<'life0, 'life1, 'async_trait>( &'life0 self, filter: &'life1 CatalogListFilter, ) -> Pin<Box<dyn Future<Output = Result<CatalogDatasetPage, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

List catalogued datasets, filtered + keyset-paginated (last_seen DESC, id DESC). Default: empty.

fn catalog_get_dataset<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<CatalogDatasetDetail>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

One dataset’s full detail: current schema, schema timeline, recent volume points, and upstream/downstream edges. Default: None.

fn catalog_lineage<'life0, 'life1, 'async_trait>( &'life0 self, root: Option<&'life1 str>, depth: u32, ) -> Pin<Box<dyn Future<Output = Result<Vec<CatalogLineageEdge>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The lineage edge graph — everything, or a depth-bounded slice around root (a dataset id). Default: empty.

fn catalog_record_config_snapshot<'life0, 'life1, 'async_trait>( &'life0 self, snapshot: &'life1 ConfigSnapshot, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Record the latest resolved+expanded config snapshot for a pipeline (#374). Latest-wins per pipeline (upsert). Best-effort at the call site — recording never fails a run. Default: no-op.

fn catalog_last_config_snapshot<'life0, 'life1, 'async_trait>( &'life0 self, pipeline: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<ConfigSnapshot>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The most recently recorded config snapshot for pipeline, or None if nothing has been recorded yet (a first faucet plan --diff). Default: None.

fn template_register<'life0, 'life1, 'async_trait>( &'life0 self, draft: &'life1 TemplateDraft, ) -> Pin<Box<dyn Future<Output = Result<TemplateRecord, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Append a new version of a template, returning the stored record with the assigned version. Versioning is atomic per id: two concurrent registers produce two distinct versions, never a lost write. Default: unsupported.

fn template_get<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, version: Option<u32>, ) -> Pin<Box<dyn Future<Output = Result<Option<TemplateRecord>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

One template version — the latest when version is None. Default: none.

fn template_list<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<TemplateSummary>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

The latest version of every registered template, newest-registered first. Default: empty.

fn template_versions<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<u32>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Every stored version number for one id, newest first. Default: empty.

fn template_delete<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, version: Option<u32>, ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Delete one version (Some) or every version (None) of a template. Returns how many rows were removed. Implementations must also drop any named-channel pointer aimed at a deleted version, so a channel never dangles. Default: 0.

fn template_set_tag<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, id: &'life1 str, tag: &'life2 str, version: u32, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Point a named channel (dev, prod, …) at an existing version, moving it if it was already set. latest is derived from the version list and never stored, so callers reject it before reaching here. Default: unsupported.

fn template_tags<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<BTreeMap<String, u32>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Every stored channel pointer for a template ({tag: version}), excluding the derived latest. Default: empty.

fn template_delete_tag<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, id: &'life1 str, tag: &'life2 str, ) -> Pin<Box<dyn Future<Output = Result<bool, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Remove one channel pointer. Returns whether it existed. Default: false.

fn template_launch<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, id: &'life1 str, version: u32, launched_by: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<Option<u32>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Append a launch to the template’s log, making version the new stable. Returns the assigned sequence number, or None when version is already stable (a re-launch is a no-op, which keeps previous meaningful rather than letting it degrade into a duplicate of stable). Default: unsupported.

fn template_launches<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Vec<LaunchRecord>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The template’s launch log, newest first. Drives stable / previous, the derived template status, and the launch audit trail. Default: empty.

fn template_set_deprecation<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, id: &'life1 str, record: Option<&'life2 DeprecationRecord>, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Set (Some) or clear (None) the template’s deprecation marker — the only stored part of the lifecycle status. Default: unsupported.

fn template_deprecation<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<DeprecationRecord>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The template’s deprecation marker, if it is deprecated. Default: None.

fn degraded(&self) -> bool

True when the backend is in fallback mode (drives /readyz). Always false for memory.

fn release_idempotency<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Release the idempotency claim(s) pointing at run_id. Called on the submit path when the run-record write that should immediately follow a Fresh claim fails — without it, a fallible (SQL) backend would orphan the claim, so every replay of the key 404s until the claim self-expires within the retention window (F21). Scoped by run_id, so a newer run that re-claimed the same key keeps its claim. Best-effort. Default: no-op — the in-memory backend’s upsert is infallible, so a Fresh claim is always paired with a record.

fn renew_leases<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Heartbeat: extend the lease of this instance’s own non-terminal runs so a peer’s recover_orphans won’t reclaim them. Returns the number of leases renewed. The memory backend (single-process, unshared) is a no-op returning 0.

fn claim_pending<'life0, 'async_trait>( &'life0 self, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<RunRecord>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Atomically claim up to limit oldest Pending runs for this instance, moving them Pending → Running with a fresh lease, and return the claimed records (with config_body) for the caller to execute. Exclusive: a run claimed by one caller is never returned to another. Default: none (memory is single-process and never writes Pending).

fn reclaim_orphans<'life0, 'async_trait>( &'life0 self, max_attempts: u32, ) -> Pin<Box<dyn Future<Output = Result<ReclaimReport, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Cluster failover: expired-lease Running runs whose attempt < max_attempts go back to Pending (owner/lease cleared, attempt++); the rest are Failed (poison). Returns the counts. Default: nothing to reclaim.

fn finalize_owned<'life0, 'life1, 'async_trait>( &'life0 self, rec: &'life1 RunRecord, ) -> Pin<Box<dyn Future<Output = Result<bool, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Owner-fenced terminal write: persist rec only if this instance still owns the run. Returns true if the write landed, false if another instance reclaimed it (the caller should discard its result). Default: delegate to upsert (memory/single-process always owns its runs).

fn finalize_sharded_parent<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, status: RunStatus, finished_at: DateTime<Utc>, error: Option<String>, ) -> Pin<Box<dyn Future<Output = Result<bool, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Status-fenced finalize of a Sharded parent run: set the terminal status / finished_at / error only while the run is still Sharded, and — crucially — WITHOUT re-stamping owner / lease_expires_at. A terminal record must not re-arm a lease, and two shards finishing on two instances at once must not last-writer-wins overwrite each other via the owner-stamping upsert (F45). Returns true if this call performed the transition (the first finalizer wins; a concurrent second call is a no-op). Default: read-guard-write via upsert — correct for the single-process in-memory backend, which has no cross-instance race and no lease columns. The SQL backends override this with one conditional UPDATE.

fn request_cancel<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Flag a Running run for cross-instance cancellation; its owning instance fires the local cancel on its next claim-loop tick. Default: no-op.

fn pending_cancellations<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

This instance’s own Running runs that have a pending cancel request. Default: none.

fn heartbeat_instance<'life0, 'life1, 'async_trait>( &'life0 self, beat: &'life1 InstanceHeartbeat, ) -> Pin<Box<dyn Future<Output = Result<(), HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Membership heartbeat: upsert this instance’s liveness row. Default: no-op.

fn live_instances<'life0, 'async_trait>( &'life0 self, ttl: Duration, ) -> Pin<Box<dyn Future<Output = Result<Vec<InstanceRecord>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Live cluster members (last heartbeat within ttl). Default: none.

fn insert_shards<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, run_id: &'life1 str, shards: &'life2 [ShardInsert], ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Idempotently insert the shard set for run_id (INSERT … ON CONFLICT (run_id, shard_id) DO NOTHING), so concurrent coordinators converge on the same set without a leader. Returns the number of rows newly inserted. Default: no-op.

fn claim_shards<'life0, 'async_trait>( &'life0 self, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<ClaimedShard>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Atomically claim up to limit pending shards for this instance (pending → running with a fresh lease), largest-estimated-size first for skew-aware balancing, returning each with its parent run record. Exclusive, like claim_pending. Default: none.

fn renew_shard_leases<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Heartbeat: extend the lease of this instance’s own running shards so a peer’s reclaim_shards won’t reassign them. Returns the number renewed. Default: no-op.

fn reclaim_shards<'life0, 'async_trait>( &'life0 self, max_attempts: u32, ) -> Pin<Box<dyn Future<Output = Result<ReclaimReport, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Rebalance: expired-lease running shards whose attempt < max_attempts go back to pending (owner cleared, attempt++) for another worker to claim; the rest are failed (poison). Returns the counts. Default: none.

fn finalize_shard<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, run_id: &'life1 str, shard_id: &'life2 str, success: bool, ) -> Pin<Box<dyn Future<Output = Result<bool, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Owner-fenced terminal write for one shard (running → completed/failed), only if this instance still owns it. Returns true if the write landed. Default: false.

fn shard_progress<'life0, 'life1, 'async_trait>( &'life0 self, run_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<ShardProgress, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Aggregate shard status counts for a run (drives parent-run finalization). Default: empty.

fn pending_shard_cancellations<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<Vec<String>, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Distinct run_ids for which THIS instance owns a running shard whose parent run has a pending cancellation request (F10). The claim loop fires each returned run’s local shard tokens via Registry::cancel_run_shards. Default: none (single-process / memory owns no cross-instance shards).

fn finalize_completed_sharded_parents<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<usize, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Sweep sharded parent runs whose shards are ALL terminal and finalize each to Completed (no failures) or Failed, stamping finished_at (F11). Recovers a parent that no shard task finalized inline (e.g. the coordinator crashed after the last shard completed elsewhere). Returns the number finalized. Status-fenced, so a concurrent inline finalize is a benign no-op and the run-finished metric is never double-counted. Default: nothing to finalize.

fn template_state<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<TemplateState, HistoryError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

The template’s full release state: versions, launch-derived stable / previous / newest, channel pointers, and the derived status.