degenbot.arbitrage.engine_registry

The canonical bot-startup orchestrator (Plan 102, slice 3).

EngineRegistry is the one correct way to start a ArbitrageEngine operator: it runs the pre-pump startup ritual (subscribe → stream snapshots → backfill → verify config) and stops before resume(), so the caller can attach its result consumer before any batches flow. It also registers pools and paths.

It is a thin adapter: every decision it used to own now belongs to the Rust core. Pool identity is derived, not mirrored — a pool’s engine pool_id comes from the shared BotState through ArbitrageEngine.pool_id_for_pool / pool_id_for_v4_pool (or straight off the pool’s own core handle), so this module holds no address → id map that could disagree with the state owner. The at-most-once verify claim is likewise core-owned: run_v*_registration_lifecycle enters the session’s VerifyClaims table inside the driver (ADR-022), so concurrent registration workers share one lifecycle run without a Python claim table.

The name stays EngineRegistry because it remains the public registration interface for operators; the registration state it used to own is gone.

Lifted from examples/eth_backrun_v2_v3_v4_rust.py — engine-operation machinery only. Deployment policy (the main loop, dispatcher, simulation overrides) stays example-side (B-mid scope).

Module Contents

class degenbot.arbitrage.engine_registry.EngineRegistry(bot: degenbot.Bot | None = None, *, engine: degenbot.arbitrage.ArbitrageEngine | None = None, path_predicate: degenbot.arbitrage.policy.PathCompositionPredicate | None = None)

Thin adapter over the Rust engine: the public registration interface.

Registration is three questions, all answered by the core:

• which id does this pool have? — the shared BotState, either off the pool’s own core handle or, for a caller holding only an identity, via ArbitrageEngine.pool_id_for_pool() / ArbitrageEngine.pool_id_for_v4_pool().

• has this pool been verified? — the core’s own registration verify lifecycle, run at most once per live claim window and skipped entirely once it has completed (ADR-022 D1). The durable fact lives on the shared driver, never here.

• what is this path’s id? — the engine’s signature dedup, via register_and_solve_path().

The first pool in each path provides the flash borrow: V2 via uniswapV2Call, V3 via swapCallback, V4 via unlockCallback. V4 pools are keyed by the (PoolManager, pool_id) pair — the engine receives the pool_manager address at registration time.

path_predicate: degenbot.arbitrage.policy.PathCompositionPredicate

start(node_http: str, node_ws: str, *, v3_snapshot: degenbot.uniswap.v3_snapshot.UniswapV3LiquiditySnapshot | None = None, v4_snapshot: degenbot.uniswap.v4_snapshot.UniswapV4LiquiditySnapshot | None = None, verify_state_view: str | None = None) → int

Run the pre-pump startup ritual and stop BEFORE resume().

Delegates the core ordering to the engine’s one-call startup ritual (the Rust EngineDriver::start mirror — subscribe(ws) then verify-config); the Python side keeps only the snapshot seed S resolution and the non-DB set_snapshot_seed_block call. Stops at the snapshot-loaded phase so the caller can attach its result consumer before batches begin to flow (resume() is the single gate after which the pump emits one ResultBatch per block into the fire-and-forget channel — attaching the consumer after resume risks unbounded backlog and stale-batch dispatch; resume() also runs the auto-backfill that closes the snapshot→WS gap).

DB snapshot (Shape 2): the V3+V4 DB snapshot is eagerly loaded into the core BotState at Bot.__init__ time via Bot.load_snapshot_from_db — so the DB path needs NO snapshot kwargs here. The snapshot seed block S stays on the shared BotState; the per-pool two-step verify (step-1) reads the stashed _verify_snapshot_block (set below from the same source), and the snapshot→WS backfill runs automatically inside resume() (BlockPump::resume_from_subscribe) using the pump’s own HTTP provider.

Non-DB snapshots (file/memory): pass v3_snapshot/v4_snapshot kwargs; each is converted to a single Python dict and handed to the engine via load_v3_snapshot_from_py / load_v4_snapshot_from_py (ONE PyO3 crossing per family — the per-pool insert_*_pool_snapshot crossings), then snapshot_block = min(s.newest_block) stashes the per-pool step-1 verify seed. These two kwargs are non-DB-only — the DB path constructs the Bot with config.database.path and passes no snapshots here.

Returns:

The first observed WS block from subscribe (the resume live-loop anchor).

static register_v2_pool(pool: degenbot.UniswapV2Pool) → int

Return the shared-core pool_id of a V2 pool.

ADR-006 slice 9: with the engine sharing the bot’s BotState, the V2 pool is ALREADY registered there by bot.build_pool (the V2 builder calls py_bot.register_v2_pool + hands back the Pool handle), and re-registering via engine.register_v2_pool would panic on the duplicate address. The id is read off the pool’s own core handle — the state owner’s answer, not a Python cache — and orientation is decided at register_path time (no fwd_key + 1 shim).

Returns:

The pool’s engine pool_id.

static register_aerodrome_pool(pool: degenbot.aerodrome.pools.AerodromeV2Pool) → int

Return the shared-core pool_id of an Aerodrome V2 pool.

The V2 twin of register_v2_pool() — the pool is already registered in the shared BotState by the delegated build_aerodrome_v2 path’s call to py_bot.register_aerodrome_pool, and the engine derives the Solidly hop family from that BotState identity at register_path time, so no engine-side pre-registration carries a family tag.

Returns:

The registered pool’s engine pool_id.

async register_v3_pool(pool: degenbot.uniswap.v3_liquidity_pool.UniswapV3Pool) → int

Run a V3 pool’s core-owned verify lifecycle and return its pool_id.

Tick data is resolved by the Rust engine from the loaded snapshot (fed via load_v3_snapshot_from_py or the DB path’s load_snapshot_from_db). The engine applies buffered events on top of stale snapshot data.

ADR-006 slice 9 / D1: the engine shares the Bot’s BotState, so the V3 pool is ALREADY registered there by bot.build_pool (the V3 builder calls py_bot.register_v3_pool + hands back the Pool handle). Re-registering via engine.register_v3_pool would PANIC the Rust core on the duplicate address — taking the process down — so this reads the shared-core pool_id off the handle and runs the lifecycle.

The at-most-once policy and the durable verify-once fact both live in the CORE: the lifecycle call enters the driver’s session claim table (ADR-022 D1), so N concurrent workers registering one pool run the choreography once and each receive its outcome, and a completed lifecycle is recorded on the driver so a later registration of the same identity is a no-op. There is deliberately no Python claim table or verified-pool set here — a caller that raced on its own map would have re-run the verify.

Returns:

The registered pool’s engine pool_id.

async register_v4_pool(pool: degenbot.uniswap.v4_liquidity_pool.UniswapV4Pool) → int

Run a V4 pool’s core-owned verify lifecycle and return its pool_id.

Tick data is resolved by the Rust engine from the loaded snapshot (fed via load_v4_snapshot_from_py or the DB path’s load_snapshot_from_db). The engine applies buffered events on top of stale snapshot data.

Pool admission (amount-modifying hooks / dynamic fees) is enforced by the Rust core as a correctness floor — the solver’s V3-CL math assumes no hook intervention + a fixed fee. A rejection surfaces as a typed HookedPoolRejectedError / DynamicFeePoolRejectedError (both subclass ValueError); build_paths classifies by type.

ADR-006 slice 9 / D1: the pool is ALREADY registered in the shared BotState by bot.build_managed_pool; re-registering would raise ValueError("V4 pool already registered") for every V4 hop in every discovered path. V4 hook/dynamic-fee admission is enforced at bot.build_managed_pool time — BEFORE this method is ever called — so it surfaces from the builder, not here. As with V3, the at-most-once verify claim is the driver’s (ADR-022 D1), keyed by the (PoolManager, pool_id) pair.

Returns:

The registered pool’s engine pool_id.

pool_id(pool: degenbot.UniswapV2Pool | degenbot.aerodrome.pools.AerodromeV2Pool | degenbot.uniswap.v3_liquidity_pool.UniswapV3Pool | degenbot.uniswap.v4_liquidity_pool.UniswapV4Pool) → int

Resolve the engine pool_id of a pool from its canonical identity.

The read the retired per-family key maps used to serve, now asked of the shared BotState: a family tag plus the pool’s own address for the address-keyed families, and the (PoolManager, pool_id) pair for V4. Identity is the key, so a pool object and the core can never disagree about which id a hop means.

Returns:

The pool’s engine pool_id.

Raises:

ValueError – If no pool with that identity is registered in the shared BotState.

knows_pool(address: str) → bool

Return whether a V2 or V3 pool with address is registered.

The core-derived question (“does the shared BotState hold a pool with this address in this family?”) — no Python map answers it.

Returns:

True if registered.

run_v3_verify_lifecycle_sync_with_retry(address: str, policy: degenbot.arbitrage.RetryPolicy) → None

V3 seat-thread verify under the core-owned bounded retry dance.

The retry classification (transient RPC/provider vs fatal mismatch) and the backoff dance are the core’s; this adapter injects the driver’s resolved policy and passes the stashed snapshot seed block. A transient failure releases the lifecycle claim, so a retry re-runs the whole choreography.

run_v4_verify_lifecycle_sync_with_retry(pool_manager: str, pool_id_hex: str, policy: degenbot.arbitrage.RetryPolicy) → None

V4 twin of run_v3_verify_lifecycle_sync_with_retry().

register_crawl_path(engine_hops: collections.abc.Sequence[tuple[int, bool]]) → tuple[int, bool]

Register a resolved hop list straight into the engine (PRG-5).

The seat-thread pathRegistration used by the crawl units: unlike register_path() it takes ALREADY-RESOLVED (pool_id, zero_for_one) hops (the unit gets the ids off the build handles). The path-composition predicate is evaluated by the caller over the concrete pools BEFORE hop building. Returns (path_id, created) with the same semantics as register_path() (dedup by construction in the engine, PRG-4; the cap refusal surfaces as typed PathRegistryFullError).

Returns:

(path_id, created) — created is False when the engine’s signature dedup answered (PRG-4).

register_path(pools_and_zfos: collections.abc.Sequence[tuple[degenbot.UniswapV2Pool | degenbot.uniswap.v3_liquidity_pool.UniswapV3Pool | degenbot.uniswap.v4_liquidity_pool.UniswapV4Pool, bool]]) → tuple[int, bool]

Register a path from concrete pool objects + per-hop directions.

Each pool’s engine key is DERIVED from the shared BotState by canonical identity (family + address, or the V4 (PoolManager, pool_id) pair) and dispatched as a (key, zero_for_one) tuple to the engine’s register_and_solve_path (eager solve — the path is immediately included in the next result batch). The Python PathInfo relay is retired — DispatchCandidate resolves the encoder’s composers::PathInfo from the returned path_id via PyArbitrageEngine.path_info_for_core (no Python hop build, no stored copy).

A hop whose pool is not registered in the shared BotState is refused by pool_id() with a ValueError before the engine is reached.

Returns:

(path_id, created) — created is False when the engine’s own signature dedup answered with an existing path_id (PRG-4: dedup is by construction core-side; the Python dedup set retired).

A path-composition policy rejection (when a predicate is injected) surfaces as a typed PathRejectedError subtype (e.g. TokenDenylistedError, HopCountExceededError, DuplicatePoolError) BEFORE hop building + engine dispatch — distinct from the Rust core’s pool-admission floor (HookedPoolRejectedError / DynamicFeePoolRejectedError).