From b7fbfcb7090a6ddce57ec14c9d56b6f5ca4a3214 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier <160007691+Flotapponnier@users.noreply.github.com> Date: Sun, 6 Sep 2026 01:08:23 +0200 Subject: [PATCH] feat(001): add Serialized to aggregator-head-lag Their trades stream is keyed by token with an optional pools filter, while the bench is keyed by pool. Subscribing by the pool's native side (SOL, WETH, WBNB) acknowledges and delivers nothing; subscribing by the side their own /v1/pool reports as `token` (USDC, BUSD, USDG) with pools= delivers exactly that market. Verified on all four bench pools before shipping (events with txHash: solana 4, base 10, bnb 2, robinhood 79). One connection per process, four pools multiplexed: their key allows 5 concurrent connections and this harness runs in three regions. Records both series: head_lag_seconds from their own `at`, same treatment as the incumbents, and head_lag_ref_seconds against the node reference clock matched by txHash. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01CpArutAtXuBb1BVNUDXoYA --- benchmarks/aggregator-head-lag.yml | 28 +- .../serialized-onboarding-audit.md | 34 +- .../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, 409 insertions(+), 3 deletions(-) create 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 1d53d822b..927d526a2 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." + - "Aggregators measured: Mobula, Codex, GeckoTerminal, Serialized." - "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: 3 aggregators × 4 chains × 3 regions = 36 active series." + - "Cardinality: 4 aggregators × 4 chains × 3 regions = 48 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,6 +217,30 @@ 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 9fb1d9be9..6715b5476 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.2, 2026-09-05 (§8 corrected in v1.1; §16.3 root cause corrected and §17 added in v1.2). Author: internal. Key used: tenant `OpenChainBench`, +> **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`, > plan `starter`, keyId `d5511a080aaa`, issued 2026-09-04. --- @@ -501,3 +501,35 @@ Layer 2: coverage against canonical fields we define, with the per-vendor mappin Layer 3, the real bench: snapshot every provider's verdict at mint, resolve on-chain at T+7d (liquidity below 5% of peak, or LP pulled), publish recall and false-positive rate per provider. Snapshot `top10HoldersPct` and `bundlersHoldingsPct` as the primary signals per §17.5. + + +## 18. Bench 001: Serialized is in (v1.3, 2026-09-06) + +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? + +Their trades stream is keyed by **token** with an optional `pools` filter, while the bench is keyed +by **pool**. Subscribing by the pool's native side (SOL, WETH, WBNB) acknowledges and delivers +nothing, consistent with their REST 404 on `So111...112`: the chain native is a quote asset to them, +never a token. Their own `GET /v1/pool` names the other side under `token` (USDC on Solana and Base, +BUSD on BNB, USDG on Robinhood). Subscribing by that address with `pools=` delivers the +tape for exactly that market. + +| Chain | pool | subscribe by | events with `txHash` | +|---|---|---|---| +| solana | 7qbRF6... | USDC `EPjF...` | 4 in 75 s (1,694 token-wide) | +| base | 0xd0b5... | USDC `0x8335...` | 10 in 90 s (173 token-wide) | +| bnb | 0x58f8... | BUSD `0xe9e7...` | 2 in 90 s | +| robinhood | 0x69bf... | USDG `0x5fc5...` | 79 in 90 s (3,888 token-wide) | + +One constraint from their official docs shaped the implementation: **5 concurrent connections per +key**. The harness runs in three regions off one key, so the monitor opens one connection per +process and multiplexes the four pools as subscriptions. A first test that opened eight connections +was refused with close code 1008 ("connection limit (5 per key)"), which is also why an earlier +Solana attempt looked like a failure. + +`harnesses/aggregator-head-lag/cmd/script/serialized_head_lag_monitor.go` records both series: +`head_lag_seconds` from their own `at` (same treatment as Mobula and Codex, same negative filter), and +`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. diff --git a/harnesses/aggregator-head-lag/cmd/script/config.go b/harnesses/aggregator-head-lag/cmd/script/config.go index a3fd51143..57fe834e2 100644 --- a/harnesses/aggregator-head-lag/cmd/script/config.go +++ b/harnesses/aggregator-head-lag/cmd/script/config.go @@ -10,6 +10,7 @@ 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) @@ -21,6 +22,7 @@ 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 31cbb69f3..e17f06bf6 100644 --- a/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go +++ b/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go @@ -811,6 +811,9 @@ 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 new file mode 100644 index 000000000..29a350a58 --- /dev/null +++ b/harnesses/aggregator-head-lag/cmd/script/serialized_head_lag_monitor.go @@ -0,0 +1,345 @@ +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) + } + } +}