1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
//! Titan pAMM price level stream integration.
//!
//! Titan Builder exposes a WebSocket stream of per-pair quote ladders (simulated price levels)
//! for the pAMMs it builds blocks with (see
//! <https://docs.titanbuilder.xyz/propamms/takers#pamm-price-level>). This module turns those
//! frames directly into [`Update`](crate::protocol::models::Update)s ready for consumption.
//!
//! # Freshness contract
//!
//! Frames are best effort, not complete snapshots: a venue or a pair can be absent from one
//! frame and present in the next, so absence is never read as retirement. Instead the stream
//! serves a component only while an accepted frame carried it within the last `stale_after`
//! (default 24 s, two slots, see [`stale_after`](stream::PriceLevelStreamBuilder::stale_after)):
//!
//! - A frame is accepted only if its wire `timestamp` is younger than `stale_after`, not more than
//! one slot in the future, not older than the newest accepted frame (equal is fine: Titan
//! re-emits within a build round), and its block neither regresses nor jumps more than one block
//! per elapsed slot plus 2. Rejected frames change nothing.
//! - A component no accepted frame has carried for `stale_after` turns stale and is emitted in
//! `removed_pairs`, together with every other component that turned stale at the same instant.
//! The next accepted frame carrying it re-adds it in `new_pairs`. Silence, disconnects,
//! keepalive-only or unparsable traffic, and replayed frames all end in this stale removal.
//! - Every emitted state also refuses to quote once its frame's `timestamp` is one slot
//! ([`QUOTE_TTL`](state::QUOTE_TTL)) old, so a ladder cannot be quoted past the block it targeted
//! even before the removal arrives. The deadline is an `Instant` of this process and is never
//! serialized. A consumer that quotes a state more than one slot after it arrived by design can
//! opt out with [`without_quote_guard`](stream::PriceLevelStreamBuilder::without_quote_guard);
//! the removal after `stale_after` still applies.
//!
//! Recovery is per component: a frame carrying a pair re-adds that pair, nothing more, and a
//! frame carrying one direction re-adds the pair with the other direction unquotable.
//! Consumers cannot tell a stale removal from a retired venue; both mean the component must not
//! be routed until it reappears in `new_pairs`. Removal and re-add are always separate updates.
//! Updates are stamped with the block their frame targets (a removal with the newest accepted
//! block); the number never decreases while something is served, but once every component has
//! turned stale the next accepted frame is judged as a first frame and may carry a lower block,
//! which is how the stream recovers from a frame with an implausible block.
//!
//! Quotes target the block currently being built, so every emitted update is marked as partial
//! and supersedes the previous one for the pairs it contains.
//!
//! # Identity and families
//!
//! Components are identified as `pricelevelstream:{pamm}`, where `{pamm}` is the configured
//! venue name (e.g. `pricelevelstream:fermiswap`) or, for auto-detected venues, the venue
//! address (e.g. `pricelevelstream:0x5979…`). The prefix keeps these components distinct from
//! those any other integration path may produce for the same venue (e.g. `vm:fermiswap`).
//!
//! By default components are emitted under `fallback:{pamm}` instead: tycho-execution routes
//! their swaps through `TychoFallbackRouter`, which retries a reverted pAMM swap on the fallback
//! pool the solver names.
//! [`without_fallback_router`](stream::PriceLevelStreamBuilder::without_fallback_router) keeps
//! every venue on the direct `pricelevelstream:` path.
//!
//! Distinct identifiers do not imply distinct liquidity, though: a venue served here may also be
//! integrated through another path, in which case the components of both paths price the same
//! underlying inventory. Consumers subscribing to multiple paths must expect such overlaps and
//! deduplicate by venue — e.g. via the
//! [`PAMM_ADDRESS_ATTRIBUTE`](stream::PAMM_ADDRESS_ATTRIBUTE) — wherever double-counting
//! matters, such as routing over the combined liquidity.
//!
//! # Observability
//!
//! The stream emits `price_level_stream_*` metrics through the `metrics` facade (frames
//! accepted and rejected by reason, the age of every accepted frame, last seen timestamp and
//! served components per registered venue, stale removals, serving state, reconnects); a
//! consumer that installs a `metrics` recorder receives them with no further setup. Per-venue
//! series start at zero for every registered venue, and no label ever carries a value from the
//! wire, except the venue address itself when a pAMM is served under auto-detection.
//!
//! Label values and gauge encodings, for dashboards and alerts:
//! - `price_level_stream_frame_age_seconds`: a histogram of the wall-clock age of every accepted
//! frame at acceptance, negative for a frame stamped ahead of the local clock. Titan's lag plus
//! delivery delay and clock skew; above one slot the frame's states refuse to quote.
//! - `price_level_stream_frames_rejected_total{reason}`: `parse_error`, `too_old`, `in_future`,
//! `out_of_order`, `block_regression`, `block_jump`.
//! - `price_level_stream_reconnects_total{reason}`: `idle_timeout`, `ended`, `closed`,
//! `read_error`, `connect_failed`, `connect_timeout`.
//! - `price_level_stream_serving_state`: 0 = unserved, 1 = serving.
//! - `venue` on `price_level_stream_last_seen_timestamp_seconds`,
//! `price_level_stream_served_components` and `price_level_stream_stale_removals_total`: the
//! registered venue name, or the address of an auto-detected venue.
//!
//! Entry point: [`PriceLevelStreamBuilder`](stream::PriceLevelStreamBuilder). Register the pAMMs
//! to serve — the known venues via
//! [`with_known_pamms`](stream::PriceLevelStreamBuilder::with_known_pamms), individual
//! [`PriceLevelStreamConfig`](config::PriceLevelStreamConfig)s via
//! [`add_pamm`](stream::PriceLevelStreamBuilder::add_pamm), or any streamed venue via
//! auto-detection — provide token metadata, and consume the resulting stream of
//! [`Update`](crate::protocol::models::Update)s.
use Duration;
/// The post-merge Ethereum slot. Every freshness window of the stream is a multiple of it.
const SLOT: Duration = from_secs;