From a7a019fbd0f9f50d5e1e0f79e1f35a613f455a12 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier <160007691+Flotapponnier@users.noreply.github.com> Date: Tue, 8 Sep 2026 15:32:15 +0200 Subject: [PATCH] revert(001): drop Serialized from aggregator-head-lag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reverts #2268 before it reaches main. Measured 2026-09-08: their `at` is the block timestamp exactly, so the ruler matches, but on Base every sampled trade arrives before its own block exists (11/11 received while the node head was still target-1, median 1.04 s ahead of publication). They stream sequencer preconfirmations; Mobula, Codex and GeckoTerminal read sealed blocks. That is a latency/finality trade-off, not a speed gap, and one ranking would present it as superiority. RecordHeadLag also drops negatives, so on the legacy series they would read as no data on Base while being the fastest. Audit §18.1 records the measurements, the two open questions for them (preconfirmed=false on all lead events, 0 Robinhood events in 240 s) and the conditions to reopen. Serialized stays on benches 004, 005, 008, 090. Claude-Session: https://claude.ai/code/session_01LtBjM1FhTKxiZhwzb3Y61c Co-Authored-By: Claude Opus 5 (1M context) --- benchmarks/aggregator-head-lag.yml | 28 +- .../serialized-onboarding-audit.md | 51 ++- .../aggregator-head-lag/cmd/script/config.go | 2 - .../cmd/script/head_lag_monitor.go | 3 - .../cmd/script/serialized_head_lag_monitor.go | 345 ------------------ 5 files changed, 51 insertions(+), 378 deletions(-) delete mode 100644 harnesses/aggregator-head-lag/cmd/script/serialized_head_lag_monitor.go diff --git a/benchmarks/aggregator-head-lag.yml b/benchmarks/aggregator-head-lag.yml index 927d526a2..1d53d822b 100644 --- a/benchmarks/aggregator-head-lag.yml +++ b/benchmarks/aggregator-head-lag.yml @@ -61,13 +61,13 @@ abstract: | same event. methodology: - - "Aggregators measured: Mobula, Codex, GeckoTerminal, Serialized." + - "Aggregators measured: Mobula, Codex, GeckoTerminal." - "Chains: Base, BNB Chain, Solana, Robinhood Chain." - "Regions: us-east, eu-west, sgp. Cross-region median reported in the headline." - "Reference: archive nodes per chain, validated against block hashes." - "Metric: gauge `head_lag_seconds`, sampled every 15 seconds. Aggregated over the 24-hour window using `quantile_over_time`." - "Success rate: presence ratio. Share of expected sampling slots where a value was actually emitted (5,760 expected per provider per day at the 15s cadence). 100% means the aggregator's feed was reachable for the full window." - - "Cardinality: 4 aggregators × 4 chains × 3 regions = 48 active series." + - "Cardinality: 3 aggregators × 4 chains × 3 regions = 36 active series." findings: - "Cross-chain average puts {{best_name}} on top at {{best_p50}} (p50, 24h) across {{count}} providers. Per-chain leaders differ: Base {{best_name:chain:base}} at {{best_p50:chain:base}}, Solana {{best_name:chain:solana}} at {{best_p50:chain:solana}}, BNB {{best_name:chain:bnb}} at {{best_p50:chain:bnb}}, Robinhood {{best_name:chain:robinhood}} at {{best_p50:chain:robinhood}}. Chains differ in block cadence (Solana 400ms vs Base/BNB 2-3s), which shifts the floor independently of provider speed." @@ -217,30 +217,6 @@ providers: p50: avg by (aggregator) (quantile_over_time(0.50, head_lag_seconds{aggregator="codex", region="sgp"}[24h])) * 1000 series: avg_over_time(head_lag_seconds{aggregator="codex", region="sgp"}[1h]) * 1000 unless (changes(head_lag_seconds{aggregator="codex", region="sgp"}[15m]) == 0 and count_over_time(head_lag_seconds{aggregator="codex", region="sgp"}[15m]) > 5 and on(chain, region) sum by (chain, region) (changes(head_lag_seconds{region="sgp"}[15m])) > 5 and on(aggregator, region) avg_over_time(ws_connected{aggregator="codex", region="sgp", chain=""}[15m]) > 0.8) - - slug: serialized - name: Serialized - tag: WebSocket trades stream - formula: "Median seconds between an on-chain swap on the bench pools and the same trade arriving on Serialized's WebSocket trades stream (subscribed per pool by its token side), sampled every 15s over 24h." - queries: - p50: avg by (aggregator) (quantile_over_time(0.50, head_lag_seconds{aggregator="serialized"}[24h])) * 1000 - p90: avg by (aggregator) (quantile_over_time(0.90, head_lag_seconds{aggregator="serialized"}[24h])) * 1000 - p99: avg by (aggregator) (quantile_over_time(0.99, head_lag_seconds{aggregator="serialized"}[24h])) * 1000 - mean: avg by (aggregator) (avg_over_time(head_lag_seconds{aggregator="serialized"}[24h])) * 1000 - success: clamp_max(avg by (aggregator) (count_over_time(head_lag_seconds{aggregator="serialized"}[24h]) / 5760), 1) - sample_size: sum(count_over_time(head_lag_seconds{aggregator="serialized"}[24h])) - series: avg_over_time(head_lag_seconds{aggregator="serialized"}[1h]) * 1000 unless (changes(head_lag_seconds{aggregator="serialized"}[15m]) == 0 and count_over_time(head_lag_seconds{aggregator="serialized"}[15m]) > 5 and on(chain, region) sum by (chain, region) (changes(head_lag_seconds{}[15m])) > 5 and on(aggregator, region) avg_over_time(ws_connected{aggregator="serialized", chain=""}[15m]) > 0.8) - live_activity: sum(changes(head_lag_seconds{aggregator="serialized"}[15m])) - regions: - - region: us-east - p50: avg by (aggregator) (quantile_over_time(0.50, head_lag_seconds{aggregator="serialized", region="us-east"}[24h])) * 1000 - series: avg_over_time(head_lag_seconds{aggregator="serialized", region="us-east"}[1h]) * 1000 unless (changes(head_lag_seconds{aggregator="serialized", region="us-east"}[15m]) == 0 and count_over_time(head_lag_seconds{aggregator="serialized", region="us-east"}[15m]) > 5 and on(chain, region) sum by (chain, region) (changes(head_lag_seconds{region="us-east"}[15m])) > 5 and on(aggregator, region) avg_over_time(ws_connected{aggregator="serialized", region="us-east", chain=""}[15m]) > 0.8) - - region: eu-west - p50: avg by (aggregator) (quantile_over_time(0.50, head_lag_seconds{aggregator="serialized", region="eu-west"}[24h])) * 1000 - series: avg_over_time(head_lag_seconds{aggregator="serialized", region="eu-west"}[1h]) * 1000 unless (changes(head_lag_seconds{aggregator="serialized", region="eu-west"}[15m]) == 0 and count_over_time(head_lag_seconds{aggregator="serialized", region="eu-west"}[15m]) > 5 and on(chain, region) sum by (chain, region) (changes(head_lag_seconds{region="eu-west"}[15m])) > 5 and on(aggregator, region) avg_over_time(ws_connected{aggregator="serialized", region="eu-west", chain=""}[15m]) > 0.8) - - region: ap-southeast - p50: avg by (aggregator) (quantile_over_time(0.50, head_lag_seconds{aggregator="serialized", region="sgp"}[24h])) * 1000 - series: avg_over_time(head_lag_seconds{aggregator="serialized", region="sgp"}[1h]) * 1000 unless (changes(head_lag_seconds{aggregator="serialized", region="sgp"}[15m]) == 0 and count_over_time(head_lag_seconds{aggregator="serialized", region="sgp"}[15m]) > 5 and on(chain, region) sum by (chain, region) (changes(head_lag_seconds{region="sgp"}[15m])) > 5 and on(aggregator, region) avg_over_time(ws_connected{aggregator="serialized", region="sgp", chain=""}[15m]) > 0.8) - - slug: geckoterminal name: GeckoTerminal tag: REST feed diff --git a/docs/methodology/serialized-onboarding-audit.md b/docs/methodology/serialized-onboarding-audit.md index 6715b5476..81d040389 100644 --- a/docs/methodology/serialized-onboarding-audit.md +++ b/docs/methodology/serialized-onboarding-audit.md @@ -3,7 +3,7 @@ > **Pre-onboarding evaluation.** Run before Serialized is wired into any live harness, so the > decision to include or exclude them on each bench is documented and reproducible. > -> **Version:** v1.3, 2026-09-06 (§8 corrected in v1.1; §16.3 root cause corrected and §17 added in v1.2; §18 added in v1.3: Serialized wired into bench 001). Author: internal. Key used: tenant `OpenChainBench`, +> **Version:** v1.4, 2026-09-08 (§8 corrected in v1.1; §16.3 root cause corrected and §17 added in v1.2; §18 added in v1.3: Serialized wired into bench 001; §18.1 added in v1.4: that wiring reverted after measurement). Author: internal. Key used: tenant `OpenChainBench`, > plan `starter`, keyId `d5511a080aaa`, issued 2026-09-04. --- @@ -503,7 +503,7 @@ Layer 3, the real bench: snapshot every provider's verdict at mint, resolve on-c Snapshot `top10HoldersPct` and `bundlersHoldingsPct` as the primary signals per §17.5. -## 18. Bench 001: Serialized is in (v1.3, 2026-09-06) +## 18. Bench 001: Serialized wired, then pulled (v1.4, 2026-09-08) Earlier sections called 001 "blocked on a policy decision". That was the wrong framing, and it hid a practical question nobody had tested: does Serialized's stream cover the four bench pools at all? @@ -533,3 +533,50 @@ Solana attempt looked like a failure. `head_lag_ref_seconds` against the node reference clock matched by `txHash`, which is the one that can rank providers. Every event carries `txHash`, `block`, `poolAddress` and a `preconfirmed` boolean per their docs, so Base flashblocks are visible rather than inferred. + +### 18.1 Why it was reverted + +Measured 2026-09-08 before promoting the wiring to `main`. One WS connection, the four bench pools, +a 240 s capture, plus a targeted 150 s run that asked a public Base node whether the block existed +at the moment each trade arrived. Single vantage (workstation, NTP offset +0.078 s), so absolute +latencies are not comparable to harness numbers; the signs and the orders of magnitude are. + +**The ruler matches.** Their `at` is the block timestamp exactly, so `head_lag_seconds` would measure +the same quantity as Mobula's `trade.Date` and Codex's `event.Timestamp`: + +| chain | n | median(`at` − block timestamp) | +|---|---|---| +| base | 105 | +0.000 s | +| bnb | 1 | +0.000 s | +| solana | 28 | +0.473 s (artefact: `getBlockTime` returns whole seconds) | + +**But they read a different thing.** On Base, every sampled trade arrived before its own block: + +| check | result | +|---|---| +| median(receipt − block timestamp), base | −0.911 s | +| share of base events with negative lag | 105/105 (100%) | +| trades received before the block existed on a public node | 11/11 | +| median lead over block publication | 1.04 s | + +In all eleven cases the node head at receipt was exactly `target block − 1`: the block was not late in +propagation, it did not exist yet. The spread (−0.21 s to −1.80 s) is the width of one Base slot. +Serialized streams sequencer preconfirmations; Mobula, Codex and GeckoTerminal read sealed blocks. + +**Decision: excluded from bench 001.** Not because the measurement is wrong, but because the two +emissions are different products. A preconfirmation carries no finality guarantee and can be +reordered, so the ~1.5 s lead is a latency/finality trade-off, not pure speed. Publishing both in one +ranking would present that trade-off as superiority. `RecordHeadLag` also drops negatives outright +(`metrics.go`), so on the legacy series they would show as no data on Base while being the fastest — +and any rare positive sample would stick on the gauge and become their published p50. + +Two findings worth raising with them: + +- `preconfirmed` was `false` on all eleven events that preceded block publication. A consumer cannot + tell the two regimes apart from the payload. +- Robinhood delivered **0 events in 240 s** despite an acknowledged subscription, against 79 in 90 s + recorded on 2026-09-06 (§18 table). Re-verify before any future wiring. + +**Reopen if** the bench gains an emission-regime dimension (sealed vs preconfirmed), or a landing-rate +companion series shows their preconfirmations reach sealed blocks 1:1. Serialized stays on benches +004, 005, 008 and 090, where the comparison is like for like. diff --git a/harnesses/aggregator-head-lag/cmd/script/config.go b/harnesses/aggregator-head-lag/cmd/script/config.go index 57fe834e2..a3fd51143 100644 --- a/harnesses/aggregator-head-lag/cmd/script/config.go +++ b/harnesses/aggregator-head-lag/cmd/script/config.go @@ -10,7 +10,6 @@ import ( type Config struct { CoinGeckoAPIKey string MobulaAPIKey string - SerializedAPIKey string DefinedSessionCookie string MonitorRegion string // Deployment region: us-west, us-east, singapore, etc. MobulaWSURL string // Mobula fast-trade WebSocket endpoint (allows staging to use EU-specific cluster) @@ -22,7 +21,6 @@ func loadEnv() (*Config, error) { // First, try to load from environment variables (for production/Railway) config.CoinGeckoAPIKey = strings.TrimSpace(os.Getenv("COINGECKO_API_KEY")) config.MobulaAPIKey = strings.TrimSpace(os.Getenv("MOBULA_API_KEY")) - config.SerializedAPIKey = strings.TrimSpace(os.Getenv("SERIALIZED_API_KEY")) config.DefinedSessionCookie = strings.TrimSpace(os.Getenv("DEFINED_SESSION_COOKIE")) config.MonitorRegion = strings.TrimSpace(os.Getenv("MONITOR_REGION")) config.MobulaWSURL = strings.TrimSpace(os.Getenv("MOBULA_WS_URL")) diff --git a/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go b/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go index e17f06bf6..31cbb69f3 100644 --- a/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go +++ b/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go @@ -811,9 +811,6 @@ func runHeadLagMonitor(config *Config, stopChan <-chan struct{}) { wg.Add(1) go runGeckoTerminalHeadLagMonitor(config, stopChan, &wg) - wg.Add(1) - go runSerializedHeadLagMonitor(config, stopChan, &wg) - // Wait for all to finish wg.Wait() fmt.Println("[HEAD-LAG] All monitors stopped") diff --git a/harnesses/aggregator-head-lag/cmd/script/serialized_head_lag_monitor.go b/harnesses/aggregator-head-lag/cmd/script/serialized_head_lag_monitor.go deleted file mode 100644 index 29a350a58..000000000 --- a/harnesses/aggregator-head-lag/cmd/script/serialized_head_lag_monitor.go +++ /dev/null @@ -1,345 +0,0 @@ -package main - -import ( - "encoding/json" - "fmt" - "io" - "log" - "net/http" - "strings" - "sync" - "time" - - "github.com/gorilla/websocket" -) - -// Serialized head-lag monitor. -// -// Serialized streams trades over one WebSocket, keyed by TOKEN with an -// optional pool filter. That shape matters for this bench, which is keyed -// by POOL: subscribing by the pool's native/wrapped side (SOL, WETH, WBNB) -// acknowledges and then delivers nothing, because Serialized treats the -// chain native as a quote asset and never as a token (their REST surface -// 404s on So111...112 the same way). Subscribing by the pool's `token` -// side as reported by their own GET /v1/pool, with `pools=`, -// delivers the tape for exactly that market. Verified before shipping on -// all four bench pools: base 10, bnb 2, robinhood 79 events with txHash -// in 90 s. So the token address is resolved from /v1/pool at startup and -// falls back to a pinned map only if that call fails. -// -// One connection per process, all pools multiplexed as subscriptions: -// Serialized caps a key at 5 concurrent connections, and this harness -// runs in three regions off one key. -// -// Lag is recorded twice, on purpose. `head_lag_seconds` uses the -// provider's own `at`, exactly like the Mobula and Codex paths, so -// Serialized gets the same treatment as the incumbents on the legacy -// series (including its filter that drops negatives). `head_lag_ref_ -// seconds` uses the node reference clock matched by txHash, which is the -// series that can actually compare providers. See reference_monitor.go. -const serializedStreamURL = "wss://api.serialized.xyz/v1/stream" - -// serializedPinnedToken is the fallback when /v1/pool is unreachable at -// boot. Values are the `token` side of each bench pool as Serialized -// reports it (2026-09-06). -var serializedPinnedToken = map[string]string{ - "solana": "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v", // USDC - "base": "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913", // USDC - "bnb": "0xe9e7cea3dedca5984780bafc599bd69add087d56", // BUSD - "robinhood": "0x5fc5360d040013d5cba0d1de2a9c7e6c4c16b83c", // USDG (best effort; runtime resolve preferred) -} - -type serializedStreamEvent struct { - Op string `json:"op"` - ID string `json:"id"` - Channel string `json:"channel"` - Error *struct { - Code string `json:"code"` - Message string `json:"message"` - } `json:"error,omitempty"` - Data *struct { - TxHash string `json:"txHash"` - ID string `json:"id"` - At int64 `json:"at"` - Block int64 `json:"block"` - Preconfirmed bool `json:"preconfirmed"` - PoolAddress string `json:"poolAddress"` - } `json:"data,omitempty"` -} - -// serializedResolveToken asks Serialized which side of the pool it treats -// as the token. Their /v1/pool is free (0 credits). -func serializedResolveToken(apiKey string, pool HeadLagPool) string { - client := &http.Client{Timeout: 10 * time.Second} - req, err := http.NewRequest("GET", - fmt.Sprintf("https://api.serialized.xyz/v1/pool?chain=%s&address=%s", pool.Blockchain, pool.Address), nil) - if err == nil { - req.Header.Set("Authorization", apiKey) - req.Header.Set("Accept", "application/json") - if resp, err := client.Do(req); err == nil { - defer resp.Body.Close() - body, _ := io.ReadAll(resp.Body) - var out struct { - Data struct { - Token struct { - Address string `json:"address"` - } `json:"token"` - } `json:"data"` - } - if resp.StatusCode == 200 && json.Unmarshal(body, &out) == nil && out.Data.Token.Address != "" { - return out.Data.Token.Address - } - } - } - return serializedPinnedToken[pool.ChainName] -} - -func runSerializedHeadLagMonitor(config *Config, stopChan <-chan struct{}, wg *sync.WaitGroup) { - defer wg.Done() - - if config.SerializedAPIKey == "" { - fmt.Println("[HEAD-LAG][SERIALIZED] SERIALIZED_API_KEY not set — monitor disabled") - RecordWSConnected("serialized", config.MonitorRegion, false) - return - } - fmt.Println("[HEAD-LAG][SERIALIZED] Starting WebSocket monitor...") - - // pool address (lowercased) -> chain name, for events that carry a - // poolAddress we did not subscribe to (should not happen with the - // pools filter, but a misrouted event must never be scored on the - // wrong chain). - poolChain := map[string]string{} - tokens := map[string]string{} - for _, p := range headLagPools { - poolChain[strings.ToLower(p.Address)] = p.ChainName - tokens[p.ChainName] = serializedResolveToken(config.SerializedAPIKey, p) - fmt.Printf("[HEAD-LAG][SERIALIZED] %s: token side %s for pool %s\n", p.ChainName, tokens[p.ChainName], p.Address) - } - - const baseDelay = 5 * time.Second - const maxDelay = 60 * time.Second - delay := baseDelay - attempt := 0 - - for { - select { - case <-stopChan: - return - default: - } - attempt++ - err := serializedConnectAndStream(config, tokens, poolChain, stopChan) - RecordWSConnected("serialized", config.MonitorRegion, false) - if err != nil { - RecordWSReconnect("serialized", config.MonitorRegion) - log.Printf("[HEAD-LAG][SERIALIZED] ❌ attempt #%d ended: %v — reconnect in %v", attempt, err, delay) - msg := err.Error() - switch { - case strings.Contains(msg, "4401"): - log.Printf("[HEAD-LAG][SERIALIZED] 🔑 auth rejected (4401): key missing, invalid or revoked") - delay = maxDelay - case strings.Contains(msg, "4402"): - log.Printf("[HEAD-LAG][SERIALIZED] 💳 monthly quota exhausted (4402)") - delay = maxDelay - case strings.Contains(msg, "connection limit"): - log.Printf("[HEAD-LAG][SERIALIZED] 🚦 5-connections-per-key cap hit: another process is holding sockets on this key") - delay = maxDelay - default: - delay *= 2 - if delay > maxDelay { - delay = maxDelay - } - } - } else { - delay = baseDelay - } - select { - case <-stopChan: - return - case <-time.After(delay): - } - } -} - -func serializedConnectAndStream(config *Config, tokens map[string]string, poolChain map[string]string, stopChan <-chan struct{}) error { - // Plain dialer, not getProxyDialer: the scraping proxy is only for - // Defined.fi and would add its own latency to this feed. - dialer := &websocket.Dialer{HandshakeTimeout: 15 * time.Second} - conn, _, err := dialer.Dial(serializedStreamURL, nil) - if err != nil { - return fmt.Errorf("dial: %w", err) - } - defer conn.Close() - - var writeMu sync.Mutex - send := func(v any) error { - writeMu.Lock() - defer writeMu.Unlock() - return conn.WriteJSON(v) - } - - // Auth must be the first frame within 10 s. - if err := send(map[string]string{"op": "auth", "apiKey": config.SerializedAPIKey}); err != nil { - return fmt.Errorf("auth send: %w", err) - } - _ = conn.SetReadDeadline(time.Now().Add(15 * time.Second)) - _, raw, err := conn.ReadMessage() - if err != nil { - return fmt.Errorf("auth read: %w", err) - } - var ack serializedStreamEvent - if json.Unmarshal(raw, &ack) != nil || ack.Op != "auth.ok" { - return fmt.Errorf("auth not acknowledged: %s", strings.TrimSpace(string(raw))) - } - - subscribed := 0 - for _, p := range headLagPools { - tok := tokens[p.ChainName] - if tok == "" { - log.Printf("[HEAD-LAG][SERIALIZED] %s: no token side known for pool %s — skipped", p.ChainName, p.Address) - continue - } - if err := send(map[string]any{ - "op": "subscribe", "channel": "trades", "id": p.ChainName, - "params": map[string]string{"chain": p.Blockchain, "address": tok, "pools": p.Address}, - }); err != nil { - return fmt.Errorf("subscribe %s: %w", p.ChainName, err) - } - subscribed++ - } - if subscribed == 0 { - return fmt.Errorf("no pool could be subscribed") - } - RecordWSConnected("serialized", config.MonitorRegion, true) - fmt.Printf("[HEAD-LAG][SERIALIZED] ✅ connected, %d pool subscriptions sent\n", subscribed) - - // Keepalive: {"op":"ping"} every 25 s; server closes idle sockets at 60 s. - done := make(chan struct{}) - defer close(done) - go func() { - t := time.NewTicker(25 * time.Second) - defer t.Stop() - for { - select { - case <-done: - return - case <-stopChan: - return - case <-t.C: - if err := send(map[string]string{"op": "ping"}); err != nil { - return - } - } - } - }() - - // Per-chain flow watchdog, same policy as the Codex path: a bench pool - // silent for 10 min means the subscription is dead even if pongs keep - // the socket alive. Purge the gauge so the page never shows a frozen - // value, then force a redial. - var lastMu sync.Mutex - lastEvent := map[string]time.Time{} - for _, p := range headLagPools { - lastEvent[p.ChainName] = time.Now() - } - const flowSilence = 10 * time.Minute - go func() { - t := time.NewTicker(time.Minute) - defer t.Stop() - for { - select { - case <-done: - return - case <-t.C: - lastMu.Lock() - tripped := "" - for chain, last := range lastEvent { - if time.Since(last) > flowSilence { - tripped = chain - break - } - } - lastMu.Unlock() - if tripped != "" { - log.Printf("[HEAD-LAG][SERIALIZED] 🪦 %s silent for >%s — purging gauge and forcing reconnect", tripped, flowSilence) - DeleteHeadLagSeries("serialized", tripped, config.MonitorRegion) - _ = conn.Close() - return - } - } - } - }() - - for { - select { - case <-stopChan: - return nil - default: - } - _ = conn.SetReadDeadline(time.Now().Add(90 * time.Second)) - _, raw, err := conn.ReadMessage() - if err != nil { - return fmt.Errorf("read: %w", err) - } - receiveTime := time.Now().UTC() - - var ev serializedStreamEvent - if json.Unmarshal(raw, &ev) != nil { - continue - } - switch ev.Op { - case "pong", "subscribed", "auth.ok": - continue - case "error": - if ev.Error != nil { - log.Printf("[HEAD-LAG][SERIALIZED] ⚠️ stream error on %q: %s %s", ev.ID, ev.Error.Code, ev.Error.Message) - } - continue - case "event": - default: - continue - } - if ev.Data == nil { - continue - } - d := ev.Data - txHash := d.TxHash - if txHash == "" && d.ID != "" { - // Older events carry the hash only inside id as ":". - txHash = strings.SplitN(d.ID, ":", 2)[0] - } - if txHash == "" || d.At == 0 { - continue - } - - chainName := ev.ID - if c, ok := poolChain[strings.ToLower(d.PoolAddress)]; ok && d.PoolAddress != "" { - chainName = c - } - if chainName == "" { - continue - } - - lastMu.Lock() - lastEvent[chainName] = time.Now() - lastMu.Unlock() - - // Legacy series: provider's own clock, identical treatment to the - // incumbents (RecordHeadLag drops negatives and >120 s itself). - lagSeconds := receiveTime.Sub(time.UnixMilli(d.At)).Seconds() - RecordHeadLag("serialized", chainName, 0, lagSeconds, config.MonitorRegion, txHash) - - // Reference series: our node clock, matched by hash. - if refAt, ok := reference.lookup(chainName, txHash); ok { - RecordHeadLagRef("serialized", chainName, receiveTime.Sub(refAt).Seconds(), config.MonitorRegion) - } else { - RecordHeadLagRefMiss("serialized", chainName, config.MonitorRegion) - } - if d.Preconfirmed { - // Flashblocks preconfirmation on Base. Kept visible in the log; - // the ref series counts it under ahead_of_reference when it lands - // before our node sees the block. - log.Printf("[HEAD-LAG][SERIALIZED] ⚡ preconfirmed trade %s on %s (lag %.3fs vs own clock)", txHash[:10], chainName, lagSeconds) - } - } -}