diff --git a/benchmarks/aggregator-head-lag.yml b/benchmarks/aggregator-head-lag.yml index 1d53d822..927d526a 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 9fb1d9be..6715b547 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 a3fd5114..57fe834e 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 31cbb69f..e17f06bf 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 00000000..29a350a5 --- /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) + } + } +}