feat(001): defer reference matching so a late reference still matches

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
 }

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 {

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)
 

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()
+	}
+}

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 {

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