degenbot.runner.build_paths

Path discovery + registration for the settlement-arbitrage BotRunner.

Extracted from examples/eth_backrun_v2_v3_v4_rust.py. Owns build_paths and its registration machinery: ConstructionContext (registration-owned construction resources kept out of the main-loop trim), PathRegistrationPipeline (the reusable, pump-concurrent per-path registration / verify / dedup), and the bounded producer/consumer helper that drives it.

The driver is Python-companion orchestration (stays-python): it registers paths with the Rust-owned engine (EngineRegistry) but owns no pool state.

Module Contents

degenbot.runner.build_paths.REG_INTAKE_WINDOW = 32

class degenbot.runner.build_paths.RegistrationUnitOutcome

The per-path unit outcome, reported back to the driver.

The units run on fleet seats (plain threads, possibly concurrent), so they NEVER touch the pipeline counters — they return one of these and the single-loop driver folds it into the summary counters exactly as the retired inline _consume did (counter parity is the PRG-3/5 bar).

kind: str

tag: str | None = None

created: bool = False

v4_hops: int = 0

counts_as_skip: bool = True

detail: str | None = None

degenbot.runner.build_paths.resolve_directions(pools: list[degenbot.UniswapV2Pool | degenbot.UniswapV3Pool | degenbot.UniswapV4Pool], input_token_address: str) → list[bool]

Determine zero_for_one for each hop so the cycle closes.

The cycle: input_token → hop_0 → intermediate → hop_1 → … → input_token. Returns a list of zfo values (one per hop).

The mechanic lives in the core (degenbot_pathfinding::directions::resolve_directions); this adapter flattens the constructed pools into the typed seam’s hop tuples and re-raises the core’s refusal as DirectionResolutionError. Resolution is kind-blind. V4 pools use NATIVE_CURRENCY_ADDRESS (address(0)) for ETH, which the core treats as equivalent to WETH — the profit token is always WETH. The core’s refusal surfaces as DirectionResolutionError — a hop carries neither tracked token, or the cycle does not close: an invariant violation (wrong pool built, stale subgraph, or a builder bug), never a skip.

Returns:

One zero-for-one value per hop, in hop order.

class degenbot.runner.build_paths.ConstructionContext

Registration-owned construction resources, kept out of run()’s trim.

Bundles everything build_paths needs to construct and register pools, so the registration task owns them as a single self-contained context for its lifetime. BotRunner.run() trims main-loop state (release_python_state() + self.bot = None); the context is a separate identity that a background registration task holds and that the trim never severs — the decoupling seam for Sub-B (background registration on the pump runtime).

The context holds RESOLVED POLICY VALUES only: the construction route (CONTEXT.md, Construction route — the ordered factory rungs + the generic builder rung) and the WETH token, built once here. The core route entry (pool_builder::route) owns the walk — route order, the DB two-step identity, get-or-register into session state — and classifies every failure on the build-refusal taxonomy, so no tracker or snapshot object lives here (the retired three-tracker fallback chain was the bare except-and-continue bug this replaces).

bot: degenbot.Bot

chain_id: int

database_path: pathlib.Path

construction_route: degenbot.builders.request.ConstructionRoute

weth: Any

classmethod for_bot(bot: degenbot.Bot) → ConstructionContext

Build the construction context for a bot.

Resolves the route policy (the mainnet V3 fork factories in policy order, generic builder rung armed) and builds WETH once. The core route entry owns everything else about construction.

Returns:

The construction context.

class degenbot.runner.build_paths.PathRegistrationPipeline(*, context: ConstructionContext, engine_registry: degenbot.arbitrage.engine_registry.EngineRegistry, retry_policy: degenbot.arbitrage.RetryPolicy | None = None, max_paths: int, discovery_batch_size: int, progress_interval_secs: float | None = None)

Reusable, pump-concurrent registration pipeline (D1c).

Owns the per-path registration work that build_paths previously ran inline: construction (through the retained ConstructionContext — the Rust PoolBuilder), engine registration + verification, direction resolution, registered-path dedup, per-path release, and the summary counters.

It is LONG-LIVED by design: it keeps the ConstructionContext AND the engine_registry for the session’s lifetime, so an operator can add a specific path (enqueue_path) or trigger a bounded on-demand discovery (trigger_discovery) at ANY time — including after run() trims the main-loop bot. The context survives the trim (Sub-A seam), so these methods never need the dropped Python bot. The pipeline never awaits the pump, so adds/discovery cannot block update/solve/dispatch.

max_paths is the registered-path cap (0 = uncapped) and is REQUIRED, because it is a configuration value: the caller that resolved it (max_registered_paths in production) states it, and a default here would be a second authority that no config layer can reach. Registration stops accepting new paths once the engine path registry reaches the cap; the engine then reaches steady state with a bounded path universe, so solve performance is observable without ongoing registration load. The pipeline announces the cap it was given at startup, since the code default and the running environment may differ.

The fail-fast tripwire is preserved: a fatal VerificationMismatchError / VerificationRpcError is NOT swallowed here — it propagates out of the worker and must abort the pipeline loudly.

constr_ctx

constr_bot

constr_chain_id

constr_database_path

construction_route

weth

engine_registry

retry_policy_obj

discovery_batch_size

pool_types: list[degenbot.pathfinding.PoolKind] = []

pool_type_per_depth: list[set[degenbot.pathfinding.PoolKind] | None] | None = None

path_count = 0

cap_skip_count = 0

skip_count = 0

token_filter_count = 0

engine_reject_count = 0

dup_count = 0

register_fail_count = 0

v4_pool_count = 0

v4_hook_rejected = 0

v4_dynamic_fee_rejected = 0

other_exc_count = 0

capped = False

emit_registration_progress(*, force: bool = False) → None

Log the registration counters + top skip-reason breakdown.

The legacy [build_paths] Progress line only fires when path_count reaches a multiple of 1000. During a discovery-heavy crawl that registers few paths it never fires, so the skip/dup/reject counts (and their reasons) stay invisible. This is the same summary emitted on force (a wall-clock cadence) so the cause is always observable mid-crawl.

async run_registration(*, producer: collections.abc.AsyncIterable[object]) → None

Run the crawl: submit each discovered path as ONE fleet unit.

PRG-5: the bounded producer/consumer queue retired with the crawl shell — discovery iterates directly and every path leaves as a single PoolStateUpdater intake unit (build + verify lifecycles + path registration inside the Rust core). The concurrency is the fleet’s (duty-counted seats + its own bounded queue), and the driver-side backpressure is the submission window: at most REG_INTAKE_WINDOW receipts are outstanding, so discovery can never outrun registration by more than the window.

Units are resolved in FIFO submission order (the retired workers’ ordering guarantee), so the Progress summary’s counter drift and the 1000-boundary log lines keep their retired shapes exactly.

Returns with EVERY submitted receipt resolved (the completion clause that replaced the retired executor-drain: all cloned Arc<SnapshotDb> handles acquired inside units are dropped before build_paths returns, keeping the close_snapshot_tx() Arc::try_unwrap canary quiet). The unit’s fatal exception (VerificationMismatchError / VerificationRpcError / DirectionResolutionError) propagates through the receipt and aborts the crawl loudly — the “shut down” contract, unchanged. On a fatal the crawl stops submitting immediately (outstanding units still finish — fleet units are never cancelled, the Deferrable cordon class).

async enqueue_path(path_steps: Any, directions: list[bool] | None = None) → None

Add ONE specific path at any time (D1c operator surface).

async trigger_discovery(*, bound: int | None = None) → int

Trigger a bounded one-shot discovery sweep (D1c).

A sweep that runs to NATURAL completion (not bound-truncated, not capped) latches the structural graph edition; a later trigger over the same edition stops immediately and returns 0 — the unchanged structure can only re-yield paths the pipeline already processed. The latch re-arms itself when the edition changes (pool added or removed), when the probe is unavailable, and after any truncated sweep.

Returns:

The number of paths processed by the sweep.

discovery_sweep(*, find_paths_async: collections.abc.Callable[..., collections.abc.AsyncGenerator[object, None]] = find_paths_async) → collections.abc.AsyncGenerator[object, None]

Run a single discovery sweep over the DB subgraph (V2/V3/V4 DFS).

find_paths_async is the discovery producer seam (tests inject a recording producer to observe the forwarded batch size); the default is the production adapter.

Returns:

The async generator of discovered candidate paths.

class degenbot.runner.build_paths.BuildPathsOptions

The knobs build_paths() takes, one options object.

Bundling the construction/registration inputs keeps the build_paths call site to the two required resources plus one options object; each field mirrors the former keyword parameter.

max_registered_paths and discovery_batch_size are the non-defaulted fields, and they are required for the same reason the pipeline’s max_paths and discovery_batch_size are: both are configuration values, so the caller states what it resolved and no code path invents one. A caller that supplies pipeline already carries them on the pipeline it passes.

The snapshot fields are retired: per-pool tick data resolves core-side at construction (the DB arm of the Rust builder, or the Chain arm), so a driver-side snapshot object has no consumer in the construction route.

max_registered_paths: int

discovery_batch_size: int

retry_policy: degenbot.arbitrage.RetryPolicy | None = None

context: ConstructionContext | None = None

pipeline: PathRegistrationPipeline | None = None

permutation_filter: frozenset[str] | None = None

async degenbot.runner.build_paths.build_paths(*, bot: degenbot.Bot, engine_registry: degenbot.arbitrage.engine_registry.EngineRegistry, options: BuildPathsOptions) → None

Discover V2/V3/V4 arb paths, build Python pools, register with Rust engine.

V4 pools are discovered via find_paths_async and built through bot.build_managed_pool(). V4 pool admission (amount-modifying hooks / dynamic fees) is enforced by the Rust core at registration time, surfacing as typed HookedPoolRejectedError / DynamicFeePoolRejectedError. Each per-pool verify lifecycle runs through the core-owned bounded retry dance with the resolved policy injected (transient VerificationRpcError is retried; VerificationMismatchError is never retried and crashes loudly).

Discovery is a single pass over the DB subgraph driven through a reusable PathRegistrationPipeline; after it completes the orphan sweep releases Tracked pools whose path was skipped before register_vN_pool.

options is required because it carries the registered-path cap: the caller that resolved the cap states it, and this function never invents one for a pipeline it builds itself.