cohort_algo’s battle-tested on-chain OrderFilled decoder + resilient WSS feed was extracted into a new lean, watchlist-parameterized reusable git crate pmv2-onchain-watch so position_manager can run its own self-wallet (Balu/Bandi) instant-alert feed by depending on it (git dep, exactly like pmv2-contracts) instead of copying the code. cohort dogfoods the crate — its watch/ is now a thin run_feed consumer. Built subagent-driven with per-task reviews; a real gap-recovery watermark bug was caught in review and fixed (in the new crate and in the original cohort code it inherited the flaw from) before publish.
For Agents — state at a glance
New crate repo: private github.com/wowjeeez/pmv2-onchain-watch, main @ pin 719dc32 (latest; was 1b8a0f9 post-cleanup, 5142d98 original publish). Lean: alloy-primitives + serde_json + tokio + reqwest (rustls) + tokio-tungstenite + anyhow/tracing/futures — NO sqlx, NO gamma, NO data-api, NO NATS.unsafe_code = "forbid", edition 2024, rust 1.85, clippy pedantic+nursery. 37 tests (was 1701 lines after the 2026-06-23 review pass; the 719dc32WatchedFill identity-fields change is additive — see WatchedFill identity fields (PM #430)).
Standalone source on disk:/Users/levander/coding/pmv2/pmv2-onchain-watch/ (src/{lib,decode,feed}.rs + README + Cargo.lock).
Two modules:decode (pure on-chain decode primitives, moved out of polymarket-data/src/onchain.rs, renamed watchlist-neutral Cohort*→Watched*) and feed (raw WSS loop moved + a NEW generalized run_feed runner).
cohort dogfoods it (babylon #402):cohort_algomain @ 4a3f8b2 (was 0fdfe8a ← 69a293a) — re-vendored the 719dc32 crate (additive WatchedFill identity fields, no call-site churn). Dropped polymarket-data/src/onchain.rs + watch/backfill.rs, vendors the crate as a workspace path-dep (crates/pmv2-onchain-watch, like pmv2-contracts), and watch/ is now a thin run_feed consumer. Behavior-preserving. Full suite green, clippy -D warnings, fmt clean, dormant-safe.
Consumer surface UNCHANGED across the cleanup — run_feed signature + the decode keep-list are byte-stable, so the −378-line review pass was a safe in-place cleanup (re-vendored at cohort 0fdfe8a). See Review pass (2026-06-23).
Why: PM wanted instant on-chain entry/exit alerts for the self-wallets (position-manager-interface-design “Plan 2 — self-wallet watcher”). Rather than duplicate cohort’s decoder + feed, Andras directed exposing it as a shared lib (operator-directed, babylon #402, 2026-06-23).
This is library extraction + dogfood, not a behavior change. cohort’s handle_cohort / first-mover / exit / gating logic is byte-identical from markets.resolve onward — only the input type changed (OrderFilledEvent/CohortAction → WatchedFill).
Why a shared crate (the reuse decision)
position_manager (PM, @deploy’s execution crate — see position-manager-interface-design) wants instant on-chain entry/exit alerts for the self-wallets (Balu / Bandi). PM’s interface design already sequenced this as “Plan 2 — self-wallet watcher”. The naive path is to copy cohort’s decoder + WSS feed into PM. But that decoder (the verified OrderFilled side/price/outcome logic — see On-Chain Detection and the on-chain decode rule in memory) and that feed (reconnect + backoff + stall watchdog + gap-backfill, see watcher-onchain-gap-backfill) are battle-tested; duplicating them means two copies drifting and two places to fix the next on-chain bug.
So Andras directed (operator-directed, babylon #402) extracting the watch substrate into a shared git crate that both services depend on — the same pattern already used for [[cohort-algo-v2-wire-contract-cutover|pmv2-contracts]] (the shared wire schema). cohort consumes it; PM will consume the same run_feed with its own self-wallet watchlist + its own sink. One feed, two callers, zero copy.
Lean by design — it pulls only what a decode+feed substrate needs. The deps are alloy-primitives (Address / U256 / log types), serde_json (JSON-RPC framing), tokio (sync/time/rt), reqwest (rustls, for eth_getLogs backfill over HTTP-RPC), tokio-tungstenite (rustls, the WSS substrate), plus anyhow/tracing/futures. Deliberately absent:sqlx, the gamma client, the data-api client, NATS — those belong to the callers, not the watch substrate. #![forbid(unsafe_code)].
decode module
Pure on-chain decode primitives — moved out of polymarket-data/src/onchain.rs and renamed watchlist-neutral (Cohort* → Watched*):
Constants:POLYGON_WSS_ENV, EXCHANGE_ADDRESSES (the CTF V2 exchange addresses), ORDER_FILLED_TOPIC0, CONDITION_RESOLUTION_TOPIC0.
WatchedFill { party, side, token_id, usdc_micros, token_micros, … } with price() / notional_usd() / shares(); plus the Side enum (decode.rs:71, WatchedFill at decode.rs:77).
decode_watched_fill(event, watchlist: &HashSet<Address, S>) -> Option<WatchedFill> — yields the watched party’s fill (maker or taker leg), or None if neither side is in the watchlist or the leg has no USDC collateral. Generic over the hasher S — nothing is hardcoded to any one watchlist (this genericity is what made the extraction near-free, see Lesson 2).
async fn run_feed(cfg: FeedConfig, watchlist: tokio::sync::watch::Receiver<HashSet<Address>>, sink: tokio::sync::mpsc::Sender<WatchedFill>) — subscribes to OrderFilled filtered to the caller’s current watchlist (maker + taker slots), decodes + dedups, and delivers WatchedFills to the caller’s sink. Internalizes the entire resilience story so callers don’t re-implement it:
reconnect with backoff (respawn_backoff),
stall watchdog — force-reconnect if no frames arrive within watchdog (600s),
live watchlist updates — resubscribes when the watch::Receiver changes (callers just send_replace a new set),
backfill-on-reconnect — eth_getLogs from a last_seen_block watermark over the gap (the watcher-onchain-gap-backfill mechanic, now generalized + fixed — see Lesson 1).
bounded sink — drops-with-warn when the mpsc is full, so the feed can never block on a slow consumer.
Caller surface is two channels
let (wl_tx, wl_rx) = tokio::sync::watch::channel(my_wallets); // HashSet<Address>let (fill_tx, mut fill_rx) = tokio::sync::mpsc::channel::<WatchedFill>(4096);tokio::spawn(run_feed(FeedConfig { wss_url, rpc_url, ..Default::default() }, wl_rx, fill_tx));while let Some(fill) = fill_rx.recv().await { /* react to fill.party BUY/SELL */ }wl_tx.send_replace(new_wallets); // update the watchlist any time
Env: POLYGON_WSS_URL (live feed, dRPC recommended — see polygon-wss-provider-drpc) + an RPC URL (FeedConfig.rpc_url) for eth_getLogs backfill.
Commit story (the build order)
Commit
What
33c87e7
decode module — moved from polymarket-data/onchain, watchlist-neutral rename
b6910e2
feed module — raw WSS loop + watchlist-parameterized run_feed (reconnect/backfill/dedup)
d94d7c5
fix(feed): backfill_once advances watermark to scanned to — prevents gap-recovery stall / live-signal miss (Lesson 1)
5142d98
docs: README — crate surface + run_feed usage (the original publish pin)
cohort_algo immediately consumes its own extracted crate — proof the abstraction is genuinely reusable, not a speculative carve. Shipped on cohort main @ 69a293a (“refactor(watch): consume pmv2-onchain-watch crate; drop polymarket-data/onchain + backfill.rs; watch is a thin run_feed consumer (#402)”). Current cohort pin is 0fdfe8a — it re-vendored the cleaned crate after the Review pass (2026-06-23); the dogfood description below is unchanged by that (consumer surface stable), only the test count moved 306→295.
polymarket-data slimmed: dropped src/onchain.rs (verified REMOVED on disk) + its re-exports.
Vendored as a workspace path-dep — Cargo.toml workspace members now includes crates/pmv2-onchain-watch, [workspace.dependencies] pmv2-onchain-watch = { path = "crates/pmv2-onchain-watch" }, and crates/cohort_algo/Cargo.toml takes pmv2-onchain-watch = { workspace = true }. Exactly the pmv2-contracts vendoring pattern.
watch/ is now a thin run_feed consumer (watch/mod.rs): a spawn_cohort_refresh(pool, wl_tx) task owns a watch::Sender<HashSet<Address>> that re-derives the cohort set from pmv2_traders on a 60sREFRESH_INTERVAL → feeds run_feed(FeedConfig { … }, wl_rx, fill_tx) → the mpsc sink drains WatchedFills into handle_cohort.
watch/backfill.rs DELETED — its eth_getLogs sweep is now run_feed’s internal backfill-on-reconnect. (watch/ retains only mod.rs, cohort.rs, process.rs.)
Behavior-preserving — only the input type changed
handle_cohort / first-mover / exit / gating logic is byte-identical from markets.resolve onward. The ONLY change is the input type: the watcher used to hand cohort an OrderFilledEvent + CohortAction; it now hands a WatchedFill. cohort.rs confirms: use pmv2_onchain_watch::decode::{Side, WatchedFill}; and handle_cohort(action: &WatchedFill, …) calls markets.resolve(&deps.gamma, action.token_id) — same resolve, same downstream. Gate: 306 tests, clippy -D warnings, fmt clean, dormant-safe (no panic when POLYGON_WSS_URL / NATS_URL unset). Mode unchanged (cohort still OFF).
This supersedes the watcher internals described in watcher-onchain-gap-backfill and Layout (which still show watch/backfill.rs + the polymarket-dataonchain.rs primitives) — those are now historical for cohort; the feed substrate lives in the shared crate.
Review pass (2026-06-23)
After the crate was published at 5142d98, a DRY / SOLID / bug review pass (two parallel adversarial reviewers, all findings verified against disk before acting) led to a substantial cleanup that is now the canonical pin: wowjeeez/pmv2-onchain-watch @ 1b8a0f9 (was 5142d98). cohort re-vendored the cleaned crate at cohort_algomain @ 0fdfe8a — 295 tests, clippy, fmt all green.
The run_feed signature and the decode keep-list (WatchedFill, decode_watched_fill, Side, FeedConfig, the constants cohort imports) are byte-stable across the cleanup. Everything deleted was unreachable from the one real entry point (run_feed). That’s why a −378-line refactor could re-vendor into cohort with zero call-site churn and stay behavior-preserving. The pin moved; the API did not.
What changed (−378 lines, 2079 → 1701)
Dead code deleted (24 items). All carried over verbatim from cohort’s original onchain.rs and unreachable from the single real entry point run_feed:
The entire newHeads / Detection / DetectionSink / run_detector block-latency path — a second consumer surface nothing in the new crate ever calls.
The SubCommand / dynamic-resubscribe path: run_log_feed_static hardwired a dummy command channel, so the SubCommand select-arm was dead code (live watchlist updates already flow through the watch::Receiver, not a command channel).
A write-only ack registry (written, never read).
Several unused pub helpers: order_filled_topic0, CONDITION_RESOLUTION_TOPIC0, watchlist_address_topics.
DRY extraction (single source of truth for two duplicated patterns):
topic_hex(B256) — the 0x + hex-encode pattern was duplicated ~5×; now one helper.
watchlist_topic_slots(...) — single source for the maker = slot 2 / taker = slot 3 topic-filter invariant, feeding both the subscribe builder and the getLogs builder (which had each duplicated it). One place to get the invariant right.
Robustness fixes:
Proxied-hang trap closed. The eth_getLogs / eth_block_number reqwest client now sets both timeout AND connect_timeout and is reused via OnceLock. This is the documented proxied-reqwest hang (a missing connect_timeout can hang ~75 min with zero warnings — the standing reference_proxied-client-connect-timeout rule).
Keepalive ping moved onto its own timer. The WSS ping was coupled to the 30s read-timeout select-arm, so the 15s PING_INTERVAL was dead under load (a busy socket never hit the read-timeout arm, so the ping never fired) and stall detection drifted to ~90s. Now a dedicated tokio::time::interval(PING_INTERVAL) select-arm restores the intended ~45s liveness cadence.
Boot watermark no longer silently seeds to 0. A failing boot eth_block_number used to silently seed the backfill watermark to 0 — which disabled backfill until the first live fill arrived. It now retries 3× and warns on failure instead.
Behavior preserved — only observability added
The decode_watched_fillprice ≥ 1 (usdc ≥ token) reject and the both-watched → maker-first pick are inherited cohort semantics that cohort relies on. The review pass did not touch that logic — it only added debug! observability around those branches. No decode-logic change.
Reusable lessons (from the review pass)
Lesson 4 — verbatim module extraction carries dead weight; sweep against the real consumer surface
Moving a module wholesale (onchain.rs → crate) brought along two entire entry points (run_detector block-latency path + the SubCommand resubscribe path) plus helpers that the new single consumer (run_feed) never touches. A post-extraction dead-code sweep against the actual consumer surface shed ~⅓ of the crate (−378 lines). Takeaway: “extract verbatim, then prune” is a fine workflow — but the prune step is not optional. What was reachable in the source’s multi-entry world is often dead in the destination’s single-entry world.
Lesson 5 — a reviewer's severity can be wrong; verify it against disk too
A reviewer flagged the ping-cadence issue as “Critical (600s detection + frame loss).” Tracing the actual connect_once loop on disk showed the read-timeout select-arm DOES fire on silence, so stalls were still being detected (~90s) — the ping bug degraded cadence, it did not blind the watchdog. Downgraded Critical → Important (a cadence fix, not an emergency). Takeaway: the “verify subagent output against disk” rule applies to a finding’s severity/impact claim, not just its existence. A real bug can come with an inflated blast radius.
Lesson 6 — keepalive ping cadence must live on its own timer
A keepalive ping gated behind a read-timeout arm only fires when the socket is idle long enough to hit that timeout — so under load the configured PING_INTERVAL is silently ineffective. Give the ping a dedicated tokio::time::interval select-arm so its cadence is independent of traffic. (This is the concrete fix behind robustness fix #2 above — and the generalizable rule.)
WatchedFill identity fields (PM #430, 2026-06-24)
WatchedFill now carries the on-chain identity triple (tx_hash, log_index, block_number) so consumers can build a stable, restart-safe dedup key per fill. This was added to fix a permanent one-alert-per-wallet suppression in position_manager’s self-wallet watch — the consumer this crate exists to serve (see position-manager-interface-design “Plan 2 — self-wallet watcher”).
The bug it fixes — PM's self-wallet dedup collapsed to per-wallet
PM’s self-wallet watch (the Balu/Bandi instant-alert feed) dedups on the key (wallet, transaction_hash, outcome_index), persisting a row in pm_self_trade_alerts so it doesn’t re-alert across restarts. But WatchedFill previously did NOT carry tx_hash, so PM was forced to set transaction_hash="" and outcome_index=0 for every fill. After the first fill of a given wallet was recorded, the dedup check is_alerted returned true for all subsequent fills of that wallet → permanent one-alert-per-wallet suppression (adversarial review on PM’s side confirmed this).
Why the fix had to live in the crate: no sound PM-side interim key existed — token_id + amounts isn’t unique, and now() breaks idempotency across restarts. The only durable per-fill identity is the on-chain (tx_hash, log_index), which only the decoder has. So the crate had to expose it.
The fix
Three new public fields on WatchedFill:pub tx_hash: B256, pub log_index: u64, pub block_number: u64. (block_number was a bonus — PM derives trade_time from it.)
decode_watched_fill now sets all three from the OrderFilledEvent already in scope — the event already carried them at decode (decode.rs:53-64), so this is a pure plumb-through, no new extraction.
run_feed emits the decoded WatchedFill straight through to the sink, so the emit path carries the new fields with no extra change.
The cohort consumer needed NO change — it uses decode_watched_fill and has zero WatchedFill { … } struct literals, so adding fields can’t break a call site. (This is the same single-entry-surface property that made the Review pass (2026-06-23) cleanup re-vendor cleanly.)
For Agents — PM's mapping (consumer side)
transaction_hash = format!("{:#x}", fill.tx_hash);outcome_index = fill.log_index; // log_index is the per-fill discriminator within a tx
So PM’s dedup key becomes (wallet, "{:#x}"(tx_hash), log_index) — unique per on-chain fill, stable across restarts.
Deferred — explicitly out of the lean crate's scope
block_time — would need a block→timestamp RPC; PM derives it from block_number or uses ingest time instead.
condition_id / a trueoutcome_index — gamma token→market enrichment stays consumer-side per the crate’s out-of-scope boundary (Lesson 3 — keep the gamma resolver out). PM’s outcome_index = log_index dedup key is unaffected by this deferral.
Gates green both sides: crate = 37 tests + clippy --all-targets + fmt; cohort = full suite (no failures) + clippy + fmt.
babylon: task #430 resolved, status #435 posted to @positionmanager.
Lesson 7 — a shared struct can be extended without breaking consumers when nobody constructs it directly
Adding three fields to WatchedFill broke zero consumers because every consumer (cohort and PM) obtains WatchedFillonly via decode_watched_fill / run_feed — there are no WatchedFill { … } struct literals outside the crate. Takeaway: when a library type is produced solely by the library’s own constructors (no public literal construction at call sites), you can grow its fields additively with no downstream churn. Inverting the I/O edges (Lesson 2) also bought this: callers receive the type, they don’t build it.
Reusable lessons
Lesson 1 — a gap-recovery watermark must advance on the SCAN, not only on delivery
Bug (caught in review, fixed before publish — commit d94d7c5): the new run_feed runner first advanced last_seen_block only on a delivered (watchlist-matched) fill. Combined with the 2000-block backfill cap (max_backfill_blocks), a long quiet stretch containing only non-deliverable watched OrderFilled events could stall the watermark — and then on a later reconnect the [watermark, latest] span would exceed the 2000-block cap and permanently miss a real fill in the gap.
Fix:backfill_once now advances the watermark to the scanned tounconditionally after each successful scan, via a const fn next_watermark(current, scanned_to) -> u64 (feed.rs:638 — returns scanned_to when greater, else current; a monotonic max that never rewinds on out-of-order delivery). Called unconditionally at feed.rs:634 (backfill) and feed.rs:846 (live path); the per-fill deliver() still bumps on delivered blocks (feed.rs:583) as a belt-and-suspenders. This also fixed an inherited flaw in the original cohort backfill.rs — the extraction surfaced a latent bug in the source.
Takeaway: a gap-recovery watermark tracks what you’ve examined, not what you’ve acted on. Advancing it only on delivery means quiet periods silently shrink your recovery window until a real event falls off the edge.
Lesson 2 — watchlist-parameterization turned a cohort-specific watcher into a generic library with near-zero logic change
The decode function was already generic over the watchlist (decode_cohort_action / now decode_watched_fill took &HashSet<Address, S> — generic over the hasher). So the whole generalization was: (1) the rename Cohort* → Watched*, (2) a caller-supplied tokio::sync::watch::Receiver<HashSet<Address>> for the watchlist (instead of a cohort-baked set), and (3) a caller-supplied tokio::sync::mpsc::Sender<WatchedFill> sink (instead of calling handle_cohort directly). No decode-logic change. Takeaway: when a “specific” component is already parameterized over its varying input, the path to a reusable library is mostly naming + inverting the I/O edges (caller supplies the watchlist + the sink), not a rewrite.
Lesson 3 — keep the gamma resolver OUT of the lean crate
The decode produces a token_id; turning that into a market is the caller’s concern. cohort resolves via its GammaClient (markets.resolve); PM will enrich from its own ledger + markets_by_clob_token. Keeping gamma out of pmv2-onchain-watch keeps PM’s dependency footprint light — PM has no GammaClient (this is exactly the constraint that makes producer-side gating load-bearing — see cohort-pre-live-signal-gating) and doesn’t gain one just to watch its own wallets. Takeaway: a shared substrate should stop at the boundary where callers diverge; resolution/enrichment lives caller-side.
How it was built
Subagent-driven with per-task reviews (the standard pmv2 cadence). The per-task review is what caught Lesson 1’s watermark bug before the crate was published at 5142d98 — consistent with the broader pattern that pmv2’s review rounds repeatedly catch real capital-or-correctness blockers (autotrade-3c2-round2-validation, cohort-algo-v2-wire-contract-cutover).
Documentation-only — not yet wired into PM, not deployed
This note records the crate extraction + cohort dogfood + the 2026-06-23 review-pass cleanup. PM’s self-wallet consumer (position-manager-interface-design Plan 2) is the next step that will actually use run_feed with the Balu/Bandi watchlist — it is not built here. Crate canonical pin @ 1b8a0f9; cohort dogfooding @ 0fdfe8a; mode unchanged (OFF). No prod deploy.
Related
cohort-algo-component — the producer that the watcher feeds; its watch/ Layout + polymarket-data onchain primitives are now historical (substrate moved here)
cohort-pre-live-signal-gating — why PM has no GammaClient (FAIL-CLOSED producer-side gating); the constraint behind Lesson 3
watcher-onchain-gap-backfill — the original eth_getLogs gap-backfill sweep, now generalized into run_feed (and its inherited watermark flaw fixed per Lesson 1)