Struct ReactiveEngine 

pub struct ReactiveEngine<S, N: Network = Ethereum> { /* private fields */ }

Binds a ReactiveRuntime to an EventSubscriber for the common subscribe-ingest lifecycle.

The engine treats the runtime registry as the single source of truth for handler lifecycle: register_handler and unregister_handler update runtime routing and subscriber interests as one operation, keyed by the handler’s stable HandlerId. Registration is continuity-safe by default — once the runtime has journaled canonical block N, a newly registered handler is live-adopted, replayed owner-only at N, and then caught up globally with every handler from N + 1 through activation. A factory-discovered pool therefore misses none of its own logs without making later history owner-local and unrollbackable. The subscriber must absorb overlap that crosses batch boundaries; the runtime validates and merges duplicate representations only within one ReactiveInputBatch.

Registration methods by intent:

Method	Backfill
register_handler	coordinated owner replay at the last retained block plus global catch-up above it (live-only on a fresh runtime)
register_handler_with_backfill	exactly one hash-certified block still retained by the rollback journal
register_handler_live_only	none — future logs only

Unregistering a handler stops future subscription routing and runtime decode for that handler; it deliberately does not evict EvmCache state or undo runtime side effects. See unregister_handler for the complete teardown recipe.

The runtime and subscriber stay independently accessible through runtime_mut / subscriber_mut for advanced use. One caution: avoid calling EventSubscriber::register_interests (the full-replacement setup API) on an engine-managed subscriber — implementations may clear owner-scoped bookkeeping, after which per-handler unregistration no longer releases the handler’s transport subscriptions. To bootstrap the subscriber from a runtime that already has handlers, use sync_handler_interests, which registers one owner per handler instead of one unowned blob.

Implementations

impl<S, N> ReactiveEngine<S, N>

where N: Network, S: EventSubscriber<N>,

pub fn new(runtime: ReactiveRuntime<N>, subscriber: S) -> Self

Bind a runtime and subscriber.

pub fn into_parts(self) -> Result<(ReactiveRuntime<N>, S), Box<Self>>

Split the engine into its runtime and subscriber parts when no commit is pending.

A failed delivery acknowledgement or durable checkpoint commit remains live protocol state: dropping it would allow the caller to lose the already-applied report/token pair and poll past an uncommitted batch. In that case this returns the intact engine so the caller can repair the dependency and retry through the normal ingestion method.

Errors

Returns the intact boxed engine when an acknowledgement or checkpoint commit is pending.

pub fn preview_durable_resume_position( &self, metadata: &DurableCheckpointMetadata, ) -> Result<SubscriberResumePosition, ReactiveCheckpointRestoreError>

Preview the exact subscriber position a durable restore will install.

This read-only step exists for durable subscribers that must complete asynchronous source or transport preparation before the engine invokes the synchronous EventSubscriber::restore_position hook. It decodes and validates the core runtime checkpoint, applies this runtime’s configured journal retention to the preview, and returns the same SubscriberResumePosition that resume_from_durable_checkpoint will later pass to the subscriber.

Call this on the same fresh engine that will perform the restore. After subscriber preparation completes, pass the identical metadata to resume_from_durable_checkpoint (or restore the same loaded checkpoint through restore_durable_checkpoint) without mutating engine runtime or checkpoint state in between. The checkpoint identity and, for non-finalized state, its canonical block must still be validated by the caller before external preparation.

This method does not mutate the runtime, subscriber, or checkpoint bookkeeping.

Errors

Returns ReactiveCheckpointRestoreError when the subscriber is not durable, its chain identity conflicts with the checkpoint, the engine is not fresh, or the stored runtime checkpoint is malformed, unsupported, or internally inconsistent.

pub fn resume_from_durable_checkpoint( &mut self, metadata: &DurableCheckpointMetadata, ) -> Result<(), ReactiveCheckpointRestoreError>

Resume delivery bookkeeping and canonical continuity from a cache checkpoint that has already been identity- and hash-validated and restored into EvmCache.

Call this on a fresh engine. The anchor has no rollback effects of its own: it represents the state baseline embodied by the checkpoint, while newly ingested blocks are journaled normally above it. The subscriber must advertise SubscriberCapability::DurableReplay; restoring an ephemeral stream would claim a restart guarantee it cannot uphold and is rejected before cache or runtime mutation.

Prefer restore_durable_checkpoint when the cache has not yet been restored: that helper rolls the cache back as well if runtime or subscriber activation fails.

Errors

Returns ReactiveCheckpointRestoreError when the subscriber is not durable, chain identity conflicts, the runtime is not pristine, stored runtime state is invalid, or the subscriber rejects the restored position. Runtime state is restored on subscriber failure.

pub fn restore_durable_checkpoint( &mut self, cache: &mut EvmCache, loaded: LoadedDurableCheckpoint, expected: &DurableCheckpointIdentity, ) -> Result<DurableCheckpointMetadata, ReactiveCheckpointRestoreError>

Atomically restore cache, runtime, and subscriber position from one validated durable checkpoint.

Inspect LoadedDurableCheckpoint::metadata and validate its canonical block against an authoritative RPC source before calling this method when the block is not finalized. Identity, cache-chain, runtime-state, and subscriber failures leave the cache and engine runtime unchanged. The subscriber follows EventSubscriber::restore_position’s retry contract. It must advertise SubscriberCapability::DurableReplay.

Errors

Returns ReactiveCheckpointRestoreError for checkpoint identity, cache-chain, runtime-state, subscriber-capability, subscriber-chain, or position-restore failures. Cache and runtime state remain unchanged.

pub fn runtime(&self) -> &ReactiveRuntime<N>

Borrow the runtime.

pub fn runtime_mut(&mut self) -> &mut ReactiveRuntime<N>

Mutably borrow the runtime.

pub fn subscriber(&self) -> &S

Borrow the subscriber.

pub fn subscriber_mut(&mut self) -> &mut S

Mutably borrow the subscriber.

pub fn adopt_canonical_baseline( &mut self, cache: &EvmCache, baseline: ReactiveCanonicalBaseline, ) -> Result<(), ReactiveEngineError>

Adopt a hash-pinned RPC cache snapshot as the runtime’s canonical cold-start baseline.

The cache must use the exact canonical hash selector and block-number context named by baseline; when the baseline includes a timestamp, the cache timestamp must match too. Cache, baseline, and any already-resolved subscriber identity must name the same chain. No delivery or checkpoint commit may be pending. After this succeeds, call sync_handler_interests_with_backfill before polling: it exact-replaces subscriber owners and begins event catch-up at C + 1.

Errors

Returns ReactiveEngineError when commit state is pending, the runtime is active or already has a conflicting baseline, cache/subscriber chain identity differs, or the cache is not pinned to the exact baseline.

pub fn next_batch( &mut self, cache: &EvmCache, ) -> Result<SubscriberNextBatch<'_, N>, ReactiveEngineError>

Poll the subscriber for the next batch without ingesting it.

This low-level escape hatch is unavailable while the engine owes an acknowledgement or checkpoint commit. Callers that use it must return any subscriber-owned delivery metadata through a combined next_ingest helper; raw ingestion deliberately rejects that metadata so it cannot be discarded accidentally.

Errors

Returns ReactiveEngineError when an acknowledgement/checkpoint commit is pending or subscriber and cache chain identities conflict.

pub fn ingest_batch( &mut self, cache: &mut EvmCache, batch: ReactiveInputBatch<N>, ) -> Result<ReactiveBatchReport<N>, ReactiveEngineError>

Ingest one already-polled batch through the runtime (direct effects only; surfaced resync requests are reported, not executed).

Errors

Returns ReactiveEngineError when commit state is pending, the batch carries subscriber-owned commit metadata, chain identity conflicts, or runtime ingestion fails.

pub fn ingest_batch_with_resync( &mut self, cache: &mut EvmCache, batch: ReactiveInputBatch<N>, ) -> Result<ReactiveBatchReport<N>, ReactiveEngineError>

Ingest one already-polled batch and execute the storage/account resyncs it surfaces, exactly like ReactiveRuntime::ingest_batch_with_resync.

Errors

Returns ReactiveEngineError when commit state is pending, the batch carries subscriber-owned commit metadata, chain identity conflicts, or runtime ingestion fails.

pub async fn next_ingest( &mut self, cache: &mut EvmCache, ) -> Result<Option<ReactiveBatchReport<N>>, ReactiveEngineError>

Poll the subscriber once and ingest the returned batch when present (direct effects only).

Errors

Returns ReactiveEngineError for subscriber/cache chain mismatch, pending checkpoint state, subscriber polling, runtime ingestion, or delivery-acknowledgement failure. A failed acknowledgement remains pending and is retried before polling again.

pub async fn next_ingest_with_resync( &mut self, cache: &mut EvmCache, ) -> Result<Option<ReactiveBatchReport<N>>, ReactiveEngineError>

Poll the subscriber once and ingest the returned batch with resync execution — the loop shape for consumers that rely on coverage-gap repair (root-gate resyncs, handler-requested re-reads).

Errors

Returns ReactiveEngineError for subscriber/cache chain mismatch, pending checkpoint state, subscriber polling, runtime ingestion, or delivery-acknowledgement failure. A failed acknowledgement remains pending and is retried before polling again.

pub async fn next_ingest_checkpointed( &mut self, cache: &mut EvmCache, store: &DurableCheckpointStore, identity: &DurableCheckpointIdentity, ) -> Result<Option<CheckpointedIngest<N>>, ReactiveEngineError>

Poll, ingest, atomically checkpoint, then acknowledge one batch.

The ordering is strict: subscriber acknowledgement is never attempted until the complete cache checkpoint is synced. If checkpointing or acknowledgement fails, the in-memory pending commit is retried before any later batch is polled, so a transient disk failure cannot cause the already-applied batch to execute twice in the same process. Across a process restart, resume_from_durable_checkpoint uses the stored delivery token and delivery witness to recognize and acknowledge an identical replay without re-ingestion. Reusing a token for different input or cursor state fails closed. Mutating the cache while a commit is pending also fails closed rather than binding newer state to older delivery metadata. Any explicit, implicit, or removed-log reorg that cannot be proven from the retained effect journal is rejected before mutation/save/ACK; configure ReactiveConfig::journal_depth to cover the subscriber’s reorg horizon. Hooks are dispatched only after checkpoint staging succeeds, but remain in-process observers rather than a durable outbox; see ReactiveHook. The subscriber must advertise SubscriberCapability::DurableReplay; ephemeral subscribers are rejected before polling.

Errors

Returns ReactiveEngineError when the subscriber lacks durable replay, identities or replay witnesses conflict, a checkpoint/ACK is already in an incompatible state, polling or ingestion fails, complete rollback proof is unavailable, the cache changes after staging, persistence fails, or delivery acknowledgement fails. Pending checkpoint/ACK work is retained for retry before another poll.

pub async fn next_ingest_with_resync_checkpointed( &mut self, cache: &mut EvmCache, store: &DurableCheckpointStore, identity: &DurableCheckpointIdentity, ) -> Result<Option<CheckpointedIngest<N>>, ReactiveEngineError>

Checkpointed counterpart to next_ingest_with_resync. Requires SubscriberCapability::DurableReplay and rejects an ephemeral subscriber before polling.

Errors

Returns ReactiveEngineError for the same durability, identity, rollback-proof, replay-witness, polling, ingestion, persistence, mutation-fence, and acknowledgement failures as next_ingest_checkpointed.

impl<S, N> ReactiveEngine<S, N>

where N: Network, S: InterestOwnerSubscriber<N>,

pub async fn register_handler( &mut self, handler: Arc<dyn ReactiveHandler<N>>, ) -> Result<(), ReactiveEngineRegisterError>

Register a handler with both the runtime and subscriber, backfilling its log interests from the runtime’s last canonical block.

This is the continuity-safe default for mid-lifecycle registration. The subscriber adopts the live desired state first, delivers the new owner’s matching records at retained block C as owner catch-up, then delivers C + 1 through activation as global canonical catch-up over the complete handler union. No discovery gap opens, and every effect after C enters the ordinary global rollback journal. On a runtime that has not journaled any canonical block yet (fresh start, or journal_depth 0) registration is live-only, matching pre-ingestion bootstrap. Use register_handler_with_backfill for an explicit replay of one retained block or register_handler_live_only to opt out of backfill entirely.

Subscriber registration commits before runtime routing is installed. If the subscriber operation fails or is cancelled, the runtime remains unchanged.

Errors

Returns ReactiveEngineRegisterError when the handler id is already registered or the subscriber rejects/does not support the required owner update or coordinated catch-up.

pub async fn register_handler_with_backfill( &mut self, handler: Arc<dyn ReactiveHandler<N>>, backfill: SubscriberBackfill, ) -> Result<(), ReactiveEngineRegisterError>

Register a handler and replay its matching logs at one exact retained canonical block.

Owner-only effects are appended to that block’s existing rollback journal entry. Consequently this method accepts only a bounded SubscriberBackfill whose start, end, and hash-certified retained anchor all identify the same journaled block. Wider/deeper recovery must use ordinary global canonical ingestion (for example startup catch-up), where every handler sees the records and the runtime advances coverage.

If subscriber registration fails or is cancelled, the runtime remains unchanged.

Errors

Returns ReactiveEngineRegisterError when the handler id is already registered, the requested backfill is not exactly one hash-certified retained journal block, or the subscriber update fails.

pub async fn register_handler_live_only( &mut self, handler: Arc<dyn ReactiveHandler<N>>, ) -> Result<(), ReactiveEngineRegisterError>

Register a handler without any log backfill — only logs delivered after its live subscription starts are routed to it.

If subscriber registration fails or is cancelled, the runtime remains unchanged.

Errors

Returns ReactiveEngineRegisterError when the handler id is already registered or the subscriber cannot commit the owner update.

pub async fn sync_handler_interests(&mut self) -> Result<(), SubscriberError>

Register every handler currently in the runtime registry as a subscriber interest owner.

This is the no-history bootstrap path for a fresh runtime/subscriber pair before ingestion starts, or for reattaching an already-aligned durable subscriber whose exact owner state was restored independently. Each handler becomes its own owner through one exact bulk replacement; crash-stale owners and unowned/base interests are removed.

No backfill is requested. It is therefore not the restart-recovery path for a new or potentially stale subscriber after the runtime has processed canonical state: use sync_handler_interests_with_backfill, which exact-replaces the owner set and closes continuity from the restored runtime position.

The complete exact set commits through one subscriber operation; an error or cancellation leaves the previously committed topology authoritative.

Errors

Returns SubscriberError when the subscriber cannot atomically replace the complete owner topology.

pub async fn sync_handler_interests_with_backfill( &mut self, ) -> Result<(), SubscriberError>

Rebuild subscriber owner state from a runtime that already embodies a canonical checkpoint.

The runtime registry is authoritative: the subscriber must atomically replace its complete owner set, removing crash-stale owners as well as adding the current ones. Log catch-up is routed globally through normal canonical ingestion and begins strictly at C + 1, where C is ReactiveRuntime::last_canonical_block, because the restored cache already contains every effect through C. The exact number/hash identity of C remains attached as a retained baseline and must be validated by the subscriber before it exposes post-baseline records. Global routing is essential: startup catch-up effects enter the ordinary canonical journal and can be rolled back if the certified branch later reorganizes; owner-only catch-up is reserved for a true mid-lifecycle handler addition.

A runtime without a canonical position must use sync_handler_interests instead. Block u64::MAX is rejected rather than wrapping or replaying the baseline. The replacement is one subscriber commit boundary: errors and cancellation leave the previous topology authoritative.

Errors

Returns SubscriberError::InvalidConfig when no canonical baseline exists or no exclusive successor can be represented, and otherwise propagates subscriber validation, transport, or atomic-commit failures.

pub async fn unregister_handler( &mut self, id: &HandlerId, ) -> Result<Option<Arc<dyn ReactiveHandler<N>>>, SubscriberError>

Unregister a handler from both the subscriber and runtime.

Subscriber interests are removed first so no new live records are routed to a handler after it has left the runtime registry. Returns the removed handler when the id was registered. If subscriber removal fails or is cancelled, runtime routing remains installed.

This is the routing/transport half of dropping an adapter. State the handler accumulated is deliberately left in place; the complete teardown for a pool or adapter that will not return is:

engine.unregister_handler(&id).await?;
for request_id in handler_request_ids {
    // Drop only this handler generation's queued repair work.
    engine.runtime_mut().cancel_pending_resync(&request_id);
}
for address in exclusively_owned_addresses {
    // Shared accounts require caller-side owner reference counting.
    engine.runtime_mut().untrack_account(address);
}
// optional: evict cached state via StateUpdate::purge / cache purge APIs

Health, metrics, the reorg journal, hooks, and freshness stamps are runtime-global and are never touched by handler removal.

Errors

Returns SubscriberError when the subscriber cannot commit owner removal. In that case runtime routing remains installed.