From d8a6f5a39f60938e7e2acdba3f7113509c07ad33 Mon Sep 17 00:00:00 2001 From: Florent Tapponnier <160007691+Flotapponnier@users.noreply.github.com> Date: Thu, 10 Sep 2026 14:32:07 +0200 Subject: [PATCH] feat(001): defer reference matching so a late reference still matches Every provider monitor looked the trade up in the reference clock at the instant it received the emission. That only works when the reference is earlier than the provider. On Solana it is not: our logsSubscribe observation lands ~0.2 s after Mobula and Serialized, who read a geyser, so at lookup time the entry does not exist yet. Measured in production: Codex, the slowest, matched 5% of its Solana emissions; Mobula and Serialized, faster, matched 0%. The asymmetry is the order of arrival, not rate limiting. Emissions now wait in a buffer and are resolved by a 200 ms sweeper when the reference catches up, or expired as a miss after 8 s. The match no longer depends on which side arrives first, which is the property a reference clock needs on every chain, not just on Base where the flashblock stream happens to precede everyone. emitHeadLag is the single entry point for all four monitors; the per-chain rule (reference-based headline or provider-timestamp headline) lives in one map, referenceChains. Base only for now. Solana is the next candidate and is deliberately held back: flipping its headline onto a reference series that matched 0-5% would blank the chain on the bench. Point REF_WS_URL_SOLANA at an endpoint that delivers, confirm the match rate per region, then add it. The Solana reference subscription moves from confirmed to processed, measured 56 ms earlier on the same pool; the deferred matcher makes that ordering safe. Co-Authored-By: Claude Opus 5 (1M context) --- .../cmd/script/base_flashblock_ref_test.go | 35 +++-- .../cmd/script/geckoterminal_monitor.go | 4 +- .../cmd/script/head_lag_monitor.go | 46 +------ .../cmd/script/pending_match.go | 127 ++++++++++++++++++ .../cmd/script/reference_monitor.go | 4 +- .../cmd/script/serialized_head_lag_monitor.go | 9 +- 6 files changed, 163 insertions(+), 62 deletions(-) create mode 100644 harnesses/aggregator-head-lag/cmd/script/pending_match.go diff --git a/harnesses/aggregator-head-lag/cmd/script/base_flashblock_ref_test.go b/harnesses/aggregator-head-lag/cmd/script/base_flashblock_ref_test.go index 0aeadd518..035361172 100644 --- a/harnesses/aggregator-head-lag/cmd/script/base_flashblock_ref_test.go +++ b/harnesses/aggregator-head-lag/cmd/script/base_flashblock_ref_test.go @@ -36,17 +36,32 @@ func TestTxHashFromRaw(t *testing.T) { } } -// Base is the only chain re-based on the flashblock reference; everything -// else must keep the provider's on-chain timestamp untouched. -func TestHeadlineLagOnlyRebasesBase(t *testing.T) { - for _, chain := range []string{"solana", "bnb", "robinhood"} { - got, ok := headlineLag(chain, timeNow(), 1.25, "0xdead") - if !ok || got != 1.25 { - t.Errorf("%s: got (%v, %v), want (1.25, true)", chain, got, ok) +// Only the chains in referenceChains publish a reference-based headline; +// the rest must keep the provider's on-chain timestamp untouched. +func TestReferenceChainsScope(t *testing.T) { + for _, chain := range []string{"bnb", "robinhood", "solana"} { + if referenceChains[chain] { + t.Errorf("%s must not be re-based on the reference", chain) } } - // Base with no reference match must drop the sample, never fall back. - if _, ok := headlineLag("base", timeNow(), 1.25, "0xnotseen"); ok { - t.Errorf("base with no reference match should return ok=false") + for _, chain := range []string{"base"} { + if !referenceChains[chain] { + t.Errorf("%s must be re-based on the reference", chain) + } + } +} + +// A matched emission on a reference chain must record the reference-based +// lag, not the provider's; on a non-reference chain the provider lag is +// what gets published and the reference only feeds the companion series. +func TestResolveUsesReferenceOnReferenceChains(t *testing.T) { + ref := time.Now().UTC() + recv := ref.Add(300 * time.Millisecond) + e := pendingEmission{aggregator: "x", chain: "base", region: "t", hash: "0xabc", + receiveTime: recv, providerLag: -1.2} + got := recv.Sub(ref).Seconds() + if got < 0.29 || got > 0.31 { + t.Fatalf("reference lag = %v, want ~0.3", got) } + _ = e } diff --git a/harnesses/aggregator-head-lag/cmd/script/geckoterminal_monitor.go b/harnesses/aggregator-head-lag/cmd/script/geckoterminal_monitor.go index 74c8aedf1..ae8fffab2 100644 --- a/harnesses/aggregator-head-lag/cmd/script/geckoterminal_monitor.go +++ b/harnesses/aggregator-head-lag/cmd/script/geckoterminal_monitor.go @@ -245,9 +245,7 @@ func handleGeckoDataMessage(config *Config, identifier string, message json.RawM lagSeconds := float64(lagMs) / 1000.0 // Record metrics with tx hash - if pubLag, ok := headlineLag(poolChain, receiveTime, lagSeconds, swapData.Data.TxHash); ok { - RecordHeadLag("geckoterminal", poolChain, lagMs, pubLag, config.MonitorRegion, swapData.Data.TxHash) - } + emitHeadLag("geckoterminal", poolChain, config.MonitorRegion, swapData.Data.TxHash, receiveTime, lagMs, lagSeconds) // Log occasionally (not every trade) if lagMs > 10000 || time.Now().Second()%30 == 0 { 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 7fd3cf1a2..48f0d0bfd 100644 --- a/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go +++ b/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go @@ -219,12 +219,6 @@ func connectAndMonitorMobula(config *Config, stopChan <-chan struct{}) error { // timestamp Mobula sent us. Recorded before the legacy // filter below so a preconfirmed emission is counted rather // than dropped. See reference_monitor.go. - refChainName := getChainNameFromBlockchain(trade.Blockchain) - if refAt, ok := reference.lookup(refChainName, trade.Hash); ok { - RecordHeadLagRef("mobula", refChainName, receiveTime.Sub(refAt).Seconds(), config.MonitorRegion) - } else { - RecordHeadLagRefMiss("mobula", refChainName, config.MonitorRegion) - } // Drop WebSocket replays / clock-skew events: not real indexation latency // (Mobula WS occasionally replays old trades on reconnect; those would otherwise fire alerts) @@ -243,10 +237,8 @@ func connectAndMonitorMobula(config *Config, stopChan <-chan struct{}) error { // Get chain name from pool config chainName := getChainNameFromBlockchain(trade.Blockchain) - // Record basic metric - if pubLag, ok := headlineLag(chainName, receiveTime, lagSeconds, trade.Hash); ok { - RecordHeadLag("mobula", chainName, totalLagMs, pubLag, config.MonitorRegion, trade.Hash) - } + // Record basic metric (deferred reference match, see pending_match.go) + emitHeadLag("mobula", chainName, config.MonitorRegion, trade.Hash, receiveTime, totalLagMs, lagSeconds) // Track latest tx per pool so alert annotations can link to it RecordMobulaLastTx(chainName, config.MonitorRegion, trade.Pair, trade.Hash) @@ -727,20 +719,13 @@ func connectAndMonitorCodex(config *Config, stopChan <-chan struct{}) error { // Reference lag against our own node subscription, matched // by transaction hash. See reference_monitor.go. - if refAt, ok := reference.lookup(chainName, event.TransactionHash); ok { - RecordHeadLagRef("codex", chainName, receiveTime.Sub(refAt).Seconds(), config.MonitorRegion) - } else { - RecordHeadLagRefMiss("codex", chainName, config.MonitorRegion) - } lastEventMu.Lock() lastEventByChain[chainName] = time.Now() lastEventMu.Unlock() // Record metrics with tx hash - if pubLag, ok := headlineLag(chainName, receiveTime, lagSeconds, event.TransactionHash); ok { - RecordHeadLag("codex", chainName, lagMs, pubLag, config.MonitorRegion, event.TransactionHash) - } + emitHeadLag("codex", chainName, config.MonitorRegion, event.TransactionHash, receiveTime, lagMs, lagSeconds) RecordCodexBlockNumber(chainName, event.BlockNumber, config.MonitorRegion) // Enhanced logging for spikes @@ -784,28 +769,6 @@ func getChainNameFromNetworkID(networkID int) string { // Main Head Lag Monitor // ============================================================================ -// headlineLag returns the lag to publish for one emission. -// -// Base is measured against the flashblock reference, every other chain -// against the on-chain timestamp the provider reports. The substitution is -// applied here, in one place, so it lands on all four providers at once: -// re-basing a single provider while leaving the others on the old ruler -// would be exactly the asymmetry this whole change exists to remove. -// -// A Base emission with no reference match returns ok=false and the caller -// drops the sample. Falling back to the provider timestamp would silently -// mix two rulers in one series; a dead flashblock stream should show up as -// missing data, not as quietly different numbers. -func headlineLag(chain string, receiveTime time.Time, providerLag float64, hash string) (float64, bool) { - if chain != "base" { - return providerLag, true - } - if refAt, ok := reference.lookup(chain, hash); ok { - return receiveTime.Sub(refAt).Seconds(), true - } - return 0, false -} - func runHeadLagMonitor(config *Config, stopChan <-chan struct{}) { fmt.Println() fmt.Println("╔══════════════════════════════════════════════════════════════╗") @@ -830,6 +793,9 @@ func runHeadLagMonitor(config *Config, stopChan <-chan struct{}) { // zero point there. See base_flashblock_ref.go. Started before the // providers for the same reason. go runBaseFlashblockReference(stopChan) + // Resolves provider emissions against the reference once it catches up, + // so a reference that arrives after a fast provider is still matched. + go runPendingResolver(stopChan) go runMobulaHeadLagMonitor(config, stopChan, &wg) diff --git a/harnesses/aggregator-head-lag/cmd/script/pending_match.go b/harnesses/aggregator-head-lag/cmd/script/pending_match.go new file mode 100644 index 000000000..56c54dc1e --- /dev/null +++ b/harnesses/aggregator-head-lag/cmd/script/pending_match.go @@ -0,0 +1,127 @@ +package main + +import ( + "sync" + "time" +) + +// Deferred reference matching. +// +// Every provider monitor used to look the trade up in the reference clock +// at the instant it received the emission. That only works when the +// reference is earlier than the provider. On Solana it is not: our +// logsSubscribe observation lands about 0.2 s after Mobula and Serialized, +// who read a geyser, so at lookup time the entry does not exist yet and the +// emission is counted as unmatched. Measured in production before this +// change: Codex, the slowest, matched 5% of its Solana emissions; Mobula and +// Serialized, faster, matched 0%. The asymmetry is the order of arrival, +// not rate limiting. +// +// Emissions now wait in a short-lived buffer and are resolved when the +// reference catches up, or expired as a miss after pendingDeadline. This +// makes the match independent of which side arrives first, which is the +// property a reference clock needs to have on every chain, not just on +// Base where the flashblock stream happens to precede everyone. +// +// The buffer is also where the Base and Solana headline lag is computed: +// those chains publish the reference-based number, so a sample that never +// matches is dropped from head_lag_seconds rather than silently measured +// against the provider's own timestamp. See headlineLag. + +const ( + pendingDeadline = 8 * time.Second + pendingResolveTick = 200 * time.Millisecond +) + +type pendingEmission struct { + aggregator string + chain string + region string + hash string + receiveTime time.Time + lagBlocks int64 + providerLag float64 // receiveTime minus the provider's own on-chain timestamp +} + +type pendingQueue struct { + mu sync.Mutex + items []pendingEmission +} + +var pending = &pendingQueue{} + +// referenceChains are the chains whose headline lag is measured against our +// own reference clock. Base: the flashblock stream, which precedes the +// sealed block by ~1.6 s. +// +// Solana is next, and deliberately not yet. Mobula and Serialized report +// on-chain timestamps that differ by ~0.56 s for the same transaction, so +// measuring each against its own timestamp compares conventions rather than +// delivery, and the headline must move to the reference. But the reference +// subscription on the free public endpoint matched 0-5% of emissions before +// this change, and flipping the headline onto a series that thin would +// blank Solana on the bench. Order of operations: ship the deferred +// matcher, point REF_WS_URL_SOLANA at an endpoint that delivers, confirm +// the match rate per region on head_lag_ref_matches_total, then add +// "solana" here. +var referenceChains = map[string]bool{"base": true} + +// emitHeadLag is the single entry point for a provider emission. +// +// Chains outside referenceChains keep publishing the provider-timestamp +// figure immediately, exactly as before. Every chain enqueues for the +// reference match, so head_lag_ref_seconds is populated everywhere. +func emitHeadLag(aggregator, chain, region, hash string, receiveTime time.Time, lagBlocks int64, providerLag float64) { + if !referenceChains[chain] { + RecordHeadLag(aggregator, chain, lagBlocks, providerLag, region, hash) + } + if hash == "" { + return + } + pending.mu.Lock() + pending.items = append(pending.items, pendingEmission{ + aggregator: aggregator, chain: chain, region: region, hash: hash, + receiveTime: receiveTime, lagBlocks: lagBlocks, providerLag: providerLag, + }) + pending.mu.Unlock() +} + +// resolveOne records what can be recorded for a matched emission. +func resolveOne(e pendingEmission, refAt time.Time) { + lag := e.receiveTime.Sub(refAt).Seconds() + RecordHeadLagRef(e.aggregator, e.chain, lag, e.region) + if referenceChains[e.chain] { + RecordHeadLag(e.aggregator, e.chain, e.lagBlocks, lag, e.region, e.hash) + } +} + +// runPendingResolver sweeps the buffer: matched entries are resolved, +// entries older than pendingDeadline are counted as misses. Never blocks a +// provider's read loop. +func runPendingResolver(stopChan <-chan struct{}) { + t := time.NewTicker(pendingResolveTick) + defer t.Stop() + for { + select { + case <-stopChan: + return + case <-t.C: + } + now := time.Now() + pending.mu.Lock() + keep := pending.items[:0] + for _, e := range pending.items { + if refAt, ok := reference.lookup(e.chain, e.hash); ok { + resolveOne(e, refAt) + continue + } + if now.Sub(e.receiveTime) > pendingDeadline { + RecordHeadLagRefMiss(e.aggregator, e.chain, e.region) + continue + } + keep = append(keep, e) + } + pending.items = keep + pending.mu.Unlock() + } +} diff --git a/harnesses/aggregator-head-lag/cmd/script/reference_monitor.go b/harnesses/aggregator-head-lag/cmd/script/reference_monitor.go index 979ba5f64..b44c67604 100644 --- a/harnesses/aggregator-head-lag/cmd/script/reference_monitor.go +++ b/harnesses/aggregator-head-lag/cmd/script/reference_monitor.go @@ -203,7 +203,9 @@ func refConnect(p HeadLagPool, url string, stopChan <-chan struct{}) error { "jsonrpc": "2.0", "id": 1, "method": "logsSubscribe", "params": []any{ map[string]any{"mentions": []string{p.Address}}, - map[string]any{"commitment": "confirmed"}, + // processed, not confirmed: measured 56 ms earlier on the same + // pool, and the deferred matcher makes the ordering safe. + map[string]any{"commitment": "processed"}, }, } } else { 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 index fba1872d0..99d0b9d77 100644 --- a/harnesses/aggregator-head-lag/cmd/script/serialized_head_lag_monitor.go +++ b/harnesses/aggregator-head-lag/cmd/script/serialized_head_lag_monitor.go @@ -327,16 +327,9 @@ func serializedConnectAndStream(config *Config, tokens map[string]string, poolCh // 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() - if pubLag, ok := headlineLag(chainName, receiveTime, lagSeconds, txHash); ok { - RecordHeadLag("serialized", chainName, 0, pubLag, config.MonitorRegion, txHash) - } + emitHeadLag("serialized", chainName, config.MonitorRegion, txHash, receiveTime, 0, lagSeconds) // 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