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 a canonical block, a newly registered handler is backfilled from that block, so a pool discovered in block N (say via a factory PoolCreated event) misses none of its own logs from N onward even though its live subscription starts later. Overlap between backfill and live delivery is absorbed by subscriber and runtime dedup.

Registration methods by intent:

Method	Backfill
register_handler	from the runtime’s last canonical block (live-only on a fresh runtime)
register_handler_with_backfill	explicit range or anchor (deep history)
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) -> (ReactiveRuntime<N>, S)

Split the engine into its runtime and subscriber parts.

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 next_batch(&mut self) -> SubscriberNextBatch<'_, N>

Poll the subscriber for the next batch.

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

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

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

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

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).

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).

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

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

pub 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 runtime already knows how far it has processed the chain, so the new handler’s logs are fetched from that block forward and no discovery gap opens between “we decided to track this pool” and “its live subscription started”. 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 deeper history or register_handler_live_only to opt out of backfill entirely.

If subscriber registration fails, the runtime registration is rolled back before the error is returned.

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

Register a handler and request an explicit owner-scoped log backfill for its interests (deep history / custom anchors).

If subscriber registration fails, the runtime registration is rolled back before the error is returned.

pub 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, the runtime registration is rolled back before the error is returned.

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

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

This is the bootstrap path for an engine built around a pre-populated runtime: each handler becomes its own owner (upsert semantics, so rerunning is safe and already-registered owners are refreshed in place). No backfill is requested — bootstrap happens before ingestion starts, so there is no processed position to be continuous with; use register_handler_with_backfill for handlers that need history. Owners are not removed by this call: use unregister_handler for lifecycle removal rather than mutating the runtime registry directly.

On error, owners already synced stay registered (upserts are independent); the call can simply be retried.

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

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.

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);
for address in handler_addresses {
    // stop root-gate eth_getProof probes for the account
    engine.runtime_mut().untrack_account(address);
    // drop its queued (unexecuted) repair work from the pending ledger
    engine.runtime_mut().cancel_pending_resyncs(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.