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