Struct DefaultPipelineRunRegistry 

pub struct DefaultPipelineRunRegistry { /* private fields */ }

Concrete in-memory PipelineRunRegistry backed by a PipelineRunRepository for durable persistence.

Uses:

• tokio::sync::broadcast for per-run fan-out event channels.
• tokio::sync::watch for cheap current-phase snapshots.
• A retention task that evicts finished runs every 60 seconds.

Implementations

impl DefaultPipelineRunRegistry

pub fn new( repo: Arc<dyn PipelineRunRepository>, cfg: RegistryConfig, ) -> Arc<DefaultPipelineRunRegistry>

Create a new registry. Starts a background retention task.

pub async fn new_with_orphan_reset( repo: Arc<dyn PipelineRunRepository>, cfg: RegistryConfig, ) -> Result<Arc<DefaultPipelineRunRegistry>, RegistryError>

Create a new registry and reset orphan rows on startup.

Calls repo.reset_orphans("server_restart_orphan") once before returning, per §12 of the spec (crash & restart recovery).

pub async fn get_payload( &self, run_id: Uuid, ) -> Result<Map<String, Value>, DatabaseError>

Fetch the accumulated payload for a run. Returns an empty map if the run has no payload events; returns Err only on DB failure.

pub async fn publish_already_completed(&self, run_id: Uuid) -> bool

Publish a synthetic RunEventKind::AlreadyCompleted event into the per-run broadcast channel for run_id.

Used by the HTTP cognify boxed-future path (and any analogous caller) to surface the check_pipeline_run_qualification short-circuit to WebSocket subscribers — see gap 08-08 §4.5 and locked decision 13. The short-circuit happens inside the convenience cognify(...) before pipeline::execute runs, so the executor’s watcher events do not fire; this method bridges that gap by publishing directly to the slot.

Returns true if the run slot exists and the event was sent (or dropped because no subscribers were attached); false if the slot is unknown.

pub async fn watcher_for(&self, run_id: Uuid) -> Option<Arc<ScopedRunWatcher>>

Construct a ScopedRunWatcher for the given run id, capturing the run’s event channel sink. Returns None if the run slot does not exist.

Trait Implementations

impl PipelineRunRegistry for DefaultPipelineRunRegistry

fn register_inline<'life0, 'async_trait>( &'life0 self, spec: RunSpec, work: Pin<Box<dyn Future<Output = Result<(), Box<dyn Error + Sync + Send>>> + Send>>, ) -> Pin<Box<dyn Future<Output = Result<RunOutcome, RegistryError>> + Send + 'async_trait>>

where 'life0: 'async_trait, DefaultPipelineRunRegistry: 'async_trait,

Register a new run and run its work future inline — the caller .awaits to completion. Returns a RunOutcome describing whether the run succeeded or errored.

fn register_background<'life0, 'async_trait>( &'life0 self, spec: RunSpec, work: Pin<Box<dyn Future<Output = Result<(), Box<dyn Error + Sync + Send>>> + Send>>, ) -> Pin<Box<dyn Future<Output = Result<RunHandle, RegistryError>> + Send + 'async_trait>>

where 'life0: 'async_trait, DefaultPipelineRunRegistry: 'async_trait,

Register a new run and spawn work on the Tokio runtime. Returns immediately with the handle; use subscribe(handle.run_id) to tail live events.

fn subscribe( &self, run_id: Uuid, ) -> Pin<Box<dyn Stream<Item = RunEvent> + Send>>

Subscribe to the event stream for a run.

fn snapshot_status(&self, run_id: Uuid) -> Option<RunPhase>

Snapshot the current high-level phase of a run. Returns None for unknown run ids.

fn abort<'life0, 'async_trait>( &'life0 self, run_id: Uuid, ) -> Pin<Box<dyn Future<Output = Result<(), RegistryError>> + Send + 'async_trait>>

where 'life0: 'async_trait, DefaultPipelineRunRegistry: 'async_trait,

Abort an in-flight background run. If the run has an abort handle, it is dropped immediately. When cfg.abort_writes_errored_row = true (the default), writes a DATASET_PROCESSING_ERRORED row and publishes a final Errored event so subscribers get a terminal frame.

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

where 'life0: 'async_trait, DefaultPipelineRunRegistry: 'async_trait,

Graceful shutdown: abort every in-flight run and drain all channels.