graph: re-genesis hysteresis + preserve factory TVL/volume + rich-market merge

Red MEDIUM (FIX C): maybeResetForRegenesis rewound to StartBlock on the FIRST poll where head << cursor, so an idle chain (eth_blockNumber=0x0, 'Building block') or a briefly-lagging RPC backend triggered a full re-index from genesis (DoS amplification). Add hysteresis: require the low head to persist across regenesisConfirmations consecutive polls; a recovered/normal head clears the streak. A genuine relaunch holds low for many polls; a transient dip recovers in one or two. Red MEDIUM: bumpFactory wrote SeedFactoryData with empty TVL/volume, and SeedFactory is INSERT OR REPLACE, so every pool-create/swap clobbered totalVolumeUSD/totalValueLockedUSD (emptying the landing-page header). Carry both USD aggregates through the read-modify-write (asString helper). Red LOW: handleInitializeV4 now MERGES with any volume stub a prior fill accumulated (don't reset volume24h/tradeCount to 0 when enriching with the token pair/fee). decodeUint256 guards the SetString ok bool (malformed hex -> clean 0, not a half-parsed amount). Tests: regenesis requires sustained low head; idle-low-head-that-recovers does NOT wipe; shallow reorg no reset; Initialize builds a rich Market (pair+fee), merges stub volume; bumpFactory preserves TVL/volume; decodeInt256 signed. New engine end-to-end: the dex schema resolves the FE markets/fills query shapes against indexed entities, amm does NOT (the exchange-api routing rationale, mirroring Red's live finding). Full graph suite green (CGO_ENABLED=1).
2026-09-24 00:26:07 +00:00 · 2026-06-21 13:15:52 -07:00
parent ed9e132af8
commit 305636ac07
3 changed files with 329 additions and 33 deletions
@@ -0,0 +1,107 @@
+package engine
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"github.com/luxfi/graph/storage"
+)
+
+// dex_route_test.go is the end-to-end proof for the exchange-api FIX D routing: the
+// FE's market/fill (CLOB) query shapes resolve against the `dex` schema and NOT the
+// `amm` schema. Red proved live that `amm` returns "unknown field: markets"; this
+// pins that markets/fills ARE dex-schema fields that resolve to the entities the
+// 0x9999 indexer writes (Market from the Initialize log, Fill from DEXFill), so the
+// exchange-api routing target (/v1/graph/cchain/dex/graphql) is correct.
+
+func newDexEngine(t *testing.T) (*Engine, *storage.Store) {
+	t.Helper()
+	s, err := storage.New(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Init(nil); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { s.Close() })
+	eng := New(s, &Config{MaxQueryDepth: 10, MaxResultSize: 1 << 20, QueryTimeoutMs: 30000})
+	if err := eng.LoadBuiltin("dex"); err != nil {
+		t.Fatalf("LoadBuiltin(dex): %v", err)
+	}
+	return eng, s
+}
+
+// TestDexSchema_ResolvesMarketsAndFills proves the FE's CLOB query shapes resolve on
+// the dex schema against the exact entities the native 0x9999 indexer produces.
+func TestDexSchema_ResolvesMarketsAndFills(t *testing.T) {
+	eng, s := newDexEngine(t)
+
+	// Seed exactly what the indexer writes: a rich Market (from the 0x9999 Initialize
+	// log) and a Fill (from a DEXFill at 0x9999).
+	poolID := "0x" + strings.Repeat("ab", 32)
+	s.SetEntity("Market", poolID, map[string]interface{}{
+		"id": poolID, "symbol": poolID, "baseToken": "0x11", "quoteToken": "0x22",
+		"feeTier": int64(3000), "volume24h": "400", "tradeCount": int64(1), "lastPrice": "0",
+	})
+	s.SetEntity("Fill", "0xtx#0", map[string]interface{}{
+		"id": "0xtx#0", "market": poolID, "taker": "0xaa", "amountOut": "400", "timestamp": int64(9),
+	})
+
+	// The FE markets query shape.
+	resp := eng.Execute(context.Background(), &Request{
+		Query: "{ markets(first: 5){ id baseToken quoteToken feeTier volume24h } }",
+	})
+	if len(resp.Errors) != 0 {
+		t.Fatalf("markets must resolve on dex schema, got errors: %+v", resp.Errors)
+	}
+	// Round-trip through JSON to assert the field actually carries the seeded data.
+	raw, _ := json.Marshal(resp.Data)
+	got := string(raw)
+	for _, want := range []string{`"markets"`, poolID, `"feeTier"`, `400`} {
+		if !strings.Contains(got, want) {
+			t.Errorf("markets response missing %q; got %s", want, got)
+		}
+	}
+
+	// The FE fills query shape.
+	resp = eng.Execute(context.Background(), &Request{
+		Query: "{ fills(first: 5){ id market taker amountOut } }",
+	})
+	if len(resp.Errors) != 0 {
+		t.Fatalf("fills must resolve on dex schema, got errors: %+v", resp.Errors)
+	}
+	raw, _ = json.Marshal(resp.Data)
+	if !strings.Contains(string(raw), `"fills"`) || !strings.Contains(string(raw), "0xtx#0") {
+		t.Errorf("fills response missing seeded fill; got %s", string(raw))
+	}
+}
+
+// TestAmmSchema_DoesNotResolveMarkets pins the negative half: the amm schema does NOT
+// know `markets` (it returns an unknown-field error) — which is WHY the exchange-api
+// must route CLOB queries to the dex endpoint. This mirrors Red's live finding.
+func TestAmmSchema_DoesNotResolveMarkets(t *testing.T) {
+	s, err := storage.New(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Init(nil); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { s.Close() })
+	eng := New(s, &Config{MaxQueryDepth: 10, MaxResultSize: 1 << 20, QueryTimeoutMs: 30000})
+	if err := eng.LoadBuiltin("amm"); err != nil {
+		t.Fatalf("LoadBuiltin(amm): %v", err)
+	}
+
+	resp := eng.Execute(context.Background(), &Request{Query: "{ markets(first: 1){ id } }"})
+	if len(resp.Errors) == 0 {
+		t.Fatal("amm schema must NOT resolve `markets` (it is a dex-schema field) — routing depends on this")
+	}
+	// And amm DOES resolve its own root (pools), confirming the engine is live.
+	resp = eng.Execute(context.Background(), &Request{Query: "{ pools(first: 1){ id } }"})
+	if len(resp.Errors) != 0 {
+		t.Fatalf("amm schema must resolve `pools`, got: %+v", resp.Errors)
+	}
+}
@@ -137,8 +137,25 @@ func TestHandleDEXFill_ScopedToPoolManager(t *testing.T) {
 	}
 }

+// initV4Log builds an Initialize log in the EXACT shape the 0x9999 precompile's
+// emitInitializeEvent emits (verified by the producer test
+// TestInitializeEvent_EmittedAt9999WithPairAndFee): topics [sig, poolId, currency0,
+// currency1]; data = fee(uint24) | tickSpacing(int24) | hooks(address) |
+// sqrtPriceX96(uint160) | tick(int24). NOT a fabricated layout — this mirrors the
+// real producer so the consumer test exercises the genuine wire format.
+func initV4Log(poolID, cur0, cur1 string, fee, tickSpacing, sqrtPriceX96, tick int64, blockHex string) *logEntry {
+	return &logEntry{
+		Address: LXSettleAddress,
+		Topics:  []string{SigInitializeV4, poolID, addrTopic(cur0), addrTopic(cur1)},
+		Data: "0x" + word(big.NewInt(fee)) + word(big.NewInt(tickSpacing)) +
+			word(new(big.Int)) /*hooks=0*/ + word(big.NewInt(sqrtPriceX96)) + word(big.NewInt(tick)),
+		BlockNumber: blockHex,
+	}
+}
+
 // Initialize at 0x9999 creates an AMM pool, bumps the factory aggregate, and
-// creates a DEX Market.
+// creates a RICH DEX Market carrying the token pair + fee tier (FIX B(a)) — the
+// fields the native producer (emitInitializeEvent at 0x9999) now supplies.
 func TestHandleInitializeV4_PoolMarketFactory(t *testing.T) {
 	s := newMemSQLiteStore(t)
 	idx := NewWithConfig(Config{RPC: "http://unused"}, s)
@@ -146,13 +163,7 @@ func TestHandleInitializeV4_PoolMarketFactory(t *testing.T) {
 	poolID := topic32("d00d")
 	cur0 := "0x0000000000000000000000000000000000000011"
 	cur1 := "0x0000000000000000000000000000000000000022"
-	l := &logEntry{
-		Address:     LXSettleAddress,
-		Topics:      []string{SigInitializeV4, poolID, addrTopic(cur0), addrTopic(cur1)},
-		Data:        "0x" + word(big.NewInt(3000)) + word(big.NewInt(60)) + word(big.NewInt(0)) + word(big.NewInt(0)) + word(big.NewInt(0)),
-		BlockNumber: "0x1",
-	}
-	idx.processLog(l)
+	idx.processLog(initV4Log(poolID, cur0, cur1, 3000, 60, 0, 0, "0x1"))

 	if p, _ := s.GetPool(nil, poolID); p == nil {
 		t.Fatal("expected pool from InitializeV4")
@@ -161,13 +172,86 @@ func TestHandleInitializeV4_PoolMarketFactory(t *testing.T) {
 	if f == nil || asInt64(f.(map[string]interface{})["poolCount"]) != 1 {
 		t.Fatalf("expected factory poolCount=1, got %v", f)
 	}
-	if mk, _ := s.GetByType("Market", poolID); mk == nil {
+	mk, _ := s.GetByType("Market", poolID)
+	if mk == nil {
 		t.Fatal("expected DEX Market from InitializeV4 at 0x9999")
 	}
+	mm := mk.(map[string]interface{})
+	// The Market is RICH, not a poolID-only stub: token pair + fee tier present.
+	if fmt.Sprint(mm["baseToken"]) != cur0 {
+		t.Errorf("Market baseToken = %v, want %s", mm["baseToken"], cur0)
+	}
+	if fmt.Sprint(mm["quoteToken"]) != cur1 {
+		t.Errorf("Market quoteToken = %v, want %s", mm["quoteToken"], cur1)
+	}
+	if asInt64(mm["feeTier"]) != 3000 {
+		t.Errorf("Market feeTier = %v, want 3000", mm["feeTier"])
+	}
+}
+
+// A Fill that arrives BEFORE its market's Initialize creates a volume stub; the
+// later Initialize must MERGE — adding the rich token pair + fee tier WITHOUT
+// resetting the accumulated volume/tradeCount to zero (LOW: stub-vs-rich merge).
+func TestHandleInitializeV4_MergesStubVolume(t *testing.T) {
+	s := newMemSQLiteStore(t)
+	idx := NewWithConfig(Config{RPC: "http://unused"}, s)
+
+	poolID := topic32("beef")
+	taker := "0x00000000000000000000000000000000000000aa"
+	// 1) A DEXFill lands first → writeFill creates a Market stub with volume=400.
+	idx.processLog(&logEntry{
+		Address: LXSettleAddress, Topics: []string{SigDEXFill, poolID, addrTopic(taker)},
+		Data: "0x" + word(big.NewInt(400)) + word(big.NewInt(7)), BlockNumber: "0x5", TransactionHash: "0xaa", LogIndex: "0x0",
+	})
+	// 2) Initialize arrives later → must enrich, not wipe.
+	cur0 := "0x0000000000000000000000000000000000000011"
+	cur1 := "0x0000000000000000000000000000000000000022"
+	idx.processLog(initV4Log(poolID, cur0, cur1, 500, 10, 0, 0, "0x6"))
+
+	mk, _ := s.GetByType("Market", poolID)
+	mm := mk.(map[string]interface{})
+	if mm["volume24h"] != "400" {
+		t.Errorf("Initialize must preserve accumulated volume, got %v want 400", mm["volume24h"])
+	}
+	if asInt64(mm["tradeCount"]) != 1 {
+		t.Errorf("Initialize must preserve tradeCount, got %v want 1", mm["tradeCount"])
+	}
+	if fmt.Sprint(mm["baseToken"]) != cur0 || asInt64(mm["feeTier"]) != 500 {
+		t.Errorf("Initialize must add rich fields: baseToken=%v feeTier=%v", mm["baseToken"], mm["feeTier"])
+	}
+}
+
+// bumpFactory must PRESERVE the USD aggregates across its read-modify-write — the
+// landing-page header's TVL/volume. SeedFactory is INSERT OR REPLACE, so writing
+// them empty on every pool-create/swap would clobber them (MEDIUM).
+func TestBumpFactory_PreservesTVLAndVolume(t *testing.T) {
+	s := newMemSQLiteStore(t)
+	idx := NewWithConfig(Config{RPC: "http://unused"}, s)
+
+	// Seed the factory with USD aggregates (as the price-oracle path would).
+	s.SeedFactory("1", &storage.SeedFactoryData{
+		PoolCount: 2, TxCount: 5, TotalValueLockedUSD: "1000000", TotalVolumeUSD: "5000000",
+	})
+
+	// A pool-create bumps counts — must not zero the USD aggregates.
+	idx.bumpFactory(1)
+
+	f, _ := s.GetFactory(nil, "1")
+	fm := f.(map[string]interface{})
+	if asInt64(fm["poolCount"]) != 3 {
+		t.Errorf("poolCount = %v, want 3", fm["poolCount"])
+	}
+	if fmt.Sprint(fm["totalValueLockedUSD"]) != "1000000" {
+		t.Errorf("totalValueLockedUSD clobbered: got %v want 1000000", fm["totalValueLockedUSD"])
+	}
+	if fmt.Sprint(fm["totalVolumeUSD"]) != "5000000" {
+		t.Errorf("totalVolumeUSD clobbered: got %v want 5000000", fm["totalVolumeUSD"])
+	}
 }

 // The re-genesis self-heal: a persisted cursor far above a freshly-relaunched
-// chain's head must rewind to StartBlock so the new chain indexes from genesis.
+// chain's head must rewind to StartBlock so the new chain indexes from genesis —
+// but only after the low head is CONFIRMED across regenesisConfirmations polls.
 func TestPoll_RegenesisSelfHeal(t *testing.T) {
 	s := newMemSQLiteStore(t)
 	s.SetLastBlock(5_000_000) // stale cursor from the previous chain
@@ -176,10 +260,17 @@ func TestPoll_RegenesisSelfHeal(t *testing.T) {
 		t.Fatalf("precondition: lastBlock=%d", idx.lastBlock)
 	}

-	// Simulate the head as observed on a fresh chain (head=3, far below cursor).
-	idx.maybeResetForRegenesis(3)
+	// A fresh chain reports head=3, far below cursor — sustained across the
+	// confirmation window. Only the FINAL confirming poll rewinds.
+	for i := 0; i < regenesisConfirmations-1; i++ {
+		idx.maybeResetForRegenesis(3)
+		if idx.lastBlock != 5_000_000 {
+			t.Fatalf("must not rewind before %d confirmations (poll %d), lastBlock=%d", regenesisConfirmations, i+1, idx.lastBlock)
+		}
+	}
+	idx.maybeResetForRegenesis(3) // the confirming poll
 	if idx.lastBlock != 0 {
-		t.Fatalf("expected rewind to StartBlock 0, got %d", idx.lastBlock)
+		t.Fatalf("expected rewind to StartBlock 0 after %d confirmations, got %d", regenesisConfirmations, idx.lastBlock)
 	}
 	if s.GetLastBlock() != 0 {
 		t.Fatalf("expected persisted cursor reset to 0, got %d", s.GetLastBlock())
@@ -197,3 +288,33 @@ func TestPoll_ShallowReorgNoReset(t *testing.T) {
 		t.Fatalf("shallow reorg must not reset; lastBlock=%d", idx.lastBlock)
 	}
 }
+
+// An idle/lagging RPC that momentarily reports a low head and then RECOVERS must
+// NOT wipe the cursor (re-genesis hysteresis — the DoS-amplification case Red
+// flagged: a single low reading must not trigger a full re-index from genesis).
+func TestPoll_IdleLowHeadRecoversNoWipe(t *testing.T) {
+	s := newMemSQLiteStore(t)
+	s.SetLastBlock(5_000_000)
+	idx := NewWithConfig(Config{RPC: "http://unused", StartBlock: 0}, s)
+
+	// A couple of transient low readings (idle "Building block" / lag) — fewer than
+	// the confirmation window — must not rewind.
+	idx.maybeResetForRegenesis(0) // idle: eth_blockNumber=0x0
+	idx.maybeResetForRegenesis(1)
+	if idx.lastBlock != 5_000_000 {
+		t.Fatalf("transient low head must not wipe; lastBlock=%d", idx.lastBlock)
+	}
+	// RPC recovers to the true height — the streak must clear, no wipe ever.
+	idx.maybeResetForRegenesis(5_000_001)
+	if idx.lastBlock != 5_000_000 {
+		t.Fatalf("recovered head must not wipe; lastBlock=%d", idx.lastBlock)
+	}
+	if idx.lowHeadStreak != 0 {
+		t.Fatalf("recovery must clear the low-head streak, got %d", idx.lowHeadStreak)
+	}
+	// And a subsequent single low blip still must not wipe (streak restarts at 1).
+	idx.maybeResetForRegenesis(2)
+	if idx.lastBlock != 5_000_000 {
+		t.Fatalf("post-recovery single blip must not wipe; lastBlock=%d", idx.lastBlock)
+	}
+}
@@ -47,6 +47,16 @@ type Config struct {
 // while still self-healing across a clean relaunch.
 const reorgDepth = 128

+// regenesisConfirmations is how many CONSECUTIVE polls must observe a head far
+// below our cursor before we treat it as a re-genesis and rewind. A single low
+// reading is NOT enough: an idle chain ("Building block", head=0x0) or a briefly
+// lagging RPC backend can momentarily report a head below our cursor, and a
+// single-poll rewind would wipe progress and re-index from genesis on every blip
+// (DoS amplification). A genuine relaunch keeps the head low for many polls as the
+// fresh chain rebuilds; a transient dip recovers within one or two. Requiring a
+// sustained streak distinguishes the two without an extra RPC in the common case.
+const regenesisConfirmations = 3
+
 // Indexer watches an EVM RPC and writes events to storage.
 type Indexer struct {
 	rpc         string
@@ -55,8 +65,9 @@ type Indexer struct {
 	store       *storage.Store
 	client      *http.Client

-	lastBlock uint64
-	status    Status
+	lastBlock     uint64
+	lowHeadStreak int // consecutive polls observing a head far below the cursor (re-genesis hysteresis)
+	status        Status
 }

 // New creates an indexer connected to the given RPC endpoint with default
@@ -97,20 +108,35 @@ func (idx *Indexer) Status() Status {
 	return idx.status
 }

-// maybeResetForRegenesis rewinds the cursor to StartBlock when the observed
-// chain head is far enough below our persisted cursor to indicate a clean
-// relaunch (fresh genesis) rather than a shallow reorg. A re-genesised chain
-// reports a head near zero while our store still holds the old chain's height;
-// without this rewind, `latest <= lastBlock` would be true forever and the new
-// chain would never be indexed. A normal reorg is shallow (< reorgDepth) and is
-// ignored here so transient head dips don't wipe progress.
+// maybeResetForRegenesis rewinds the cursor to StartBlock when the observed chain
+// head stays far below our persisted cursor for regenesisConfirmations CONSECUTIVE
+// polls — the signature of a clean relaunch (fresh genesis), not a shallow reorg or
+// a transient dip. A re-genesised chain reports a head near zero while our store
+// still holds the old chain's height; without this rewind `latest <= lastBlock`
+// would hold forever and the new chain would never index.
+//
+// The hysteresis is the fix for an idle/lagging RPC: an idle chain ("Building block",
+// head=0x0) or a backend that briefly reports a stale low head would, under a
+// single-poll rewind, wipe progress and re-index from genesis on every blip (DoS
+// amplification). A genuine relaunch holds the low head across many polls; a transient
+// dip recovers within one or two, clearing the streak before any wipe.
 func (idx *Indexer) maybeResetForRegenesis(latest uint64) {
-	if idx.lastBlock > idx.startBlock+reorgDepth && latest+reorgDepth < idx.lastBlock {
-		log.Printf("[indexer] head %d << cursor %d (>%d behind) — chain reset detected, rewinding to %d",
-			latest, idx.lastBlock, reorgDepth, idx.startBlock)
-		idx.lastBlock = idx.startBlock
-		idx.store.SetLastBlock(idx.startBlock)
+	farBelow := idx.lastBlock > idx.startBlock+reorgDepth && latest+reorgDepth < idx.lastBlock
+	if !farBelow {
+		idx.lowHeadStreak = 0 // head recovered (or normal) — abandon any suspicion.
+		return
 	}
+	idx.lowHeadStreak++
+	if idx.lowHeadStreak < regenesisConfirmations {
+		log.Printf("[indexer] head %d << cursor %d (>%d behind) — suspected chain reset, awaiting confirmation %d/%d",
+			latest, idx.lastBlock, reorgDepth, idx.lowHeadStreak, regenesisConfirmations)
+		return
+	}
+	log.Printf("[indexer] head %d << cursor %d for %d consecutive polls — chain reset confirmed, rewinding to %d",
+		latest, idx.lastBlock, idx.lowHeadStreak, idx.startBlock)
+	idx.lastBlock = idx.startBlock
+	idx.store.SetLastBlock(idx.startBlock)
+	idx.lowHeadStreak = 0
 }

 // Run starts the indexer loop. Blocks until ctx is cancelled.
@@ -375,6 +401,9 @@ func (idx *Indexer) processLog(l *logEntry) {
 }

 // decodeUint256 reads a 32-byte hex word from data at the given word index.
+// Malformed hex (a non-hex digit in the word) yields 0 rather than a silently
+// half-parsed value: SetString reports ok=false and leaves n unmodified, so we
+// reset to a clean zero. A bad word must not masquerade as a real (partial) amount.
 func decodeUint256(data string, wordIndex int) *big.Int {
 	data = strings.TrimPrefix(data, "0x")
 	start := wordIndex * 64
@@ -382,7 +411,9 @@ func decodeUint256(data string, wordIndex int) *big.Int {
 		return new(big.Int)
 	}
 	n := new(big.Int)
-	n.SetString(data[start:start+64], 16)
+	if _, ok := n.SetString(data[start:start+64], 16); !ok {
+		return new(big.Int)
+	}
 	return n
 }

@@ -517,16 +548,38 @@ func decodeInt256(data string, wordIndex int) *big.Int {
 func (idx *Indexer) bumpFactory(deltaPools int64) {
 	f, _ := idx.store.GetFactory(nil, "1")
 	var pc, tc int64
+	var tvl, vol string
 	if m, ok := f.(map[string]interface{}); ok {
 		pc = asInt64(m["poolCount"])
 		tc = asInt64(m["txCount"])
+		// Carry the accumulated USD aggregates through the read-modify-write.
+		// SeedFactory is INSERT OR REPLACE (full overwrite); writing them empty
+		// here would clobber the landing-page header's TVL/volume on every
+		// pool-create / swap. They are accumulated elsewhere (price oracle / day
+		// data); bumpFactory only owns the pool + tx counts and must preserve the rest.
+		tvl = asString(m["totalValueLockedUSD"])
+		vol = asString(m["totalVolumeUSD"])
 	}
 	idx.store.SeedFactory("1", &storage.SeedFactoryData{
-		PoolCount: pc + deltaPools,
-		TxCount:   tc + 1,
+		PoolCount:           pc + deltaPools,
+		TxCount:             tc + 1,
+		TotalValueLockedUSD: tvl,
+		TotalVolumeUSD:      vol,
 	})
 }

+// asString coerces a JSON-decoded value to its string form, returning "" for nil
+// so a missing aggregate carries through a read-modify-write as empty (not "<nil>").
+func asString(v interface{}) string {
+	if v == nil {
+		return ""
+	}
+	if s, ok := v.(string); ok {
+		return s
+	}
+	return fmt.Sprint(v)
+}
+
 // asInt64 best-effort coerces a JSON-decoded numeric (float64/json.Number/int)
 // to int64 for aggregate accumulation.
 func asInt64(v interface{}) int64 {
@@ -567,10 +620,25 @@ func (idx *Indexer) handleInitializeV4(l *logEntry) {
 	idx.bumpFactory(1)

 	if idx.isPoolManager(l.Address) {
-		idx.store.SetEntity("Market", poolID, map[string]interface{}{
+		// MERGE with any stub a prior fill already accumulated (writeFill creates a
+		// volume24h/tradeCount stub when a Fill arrives before its Initialize). The
+		// Initialize supplies the RICH fields (token pair, fee tier); it must NOT
+		// reset the accumulated trade aggregates to zero. Defaults apply only when
+		// the field is absent (no prior stub).
+		mk := map[string]interface{}{
 			"id": poolID, "symbol": poolID, "baseToken": token0, "quoteToken": token1,
-			"feeTier": fee, "volume24h": "0", "tradeCount": 0, "lastPrice": "0",
-		})
+			"feeTier": fee, "volume24h": "0", "tradeCount": int64(0), "lastPrice": "0",
+		}
+		if existing, _ := idx.store.GetByType("Market", poolID); existing != nil {
+			if em, ok := existing.(map[string]interface{}); ok {
+				for _, k := range []string{"volume24h", "tradeCount", "lastPrice", "lastUpdate"} {
+					if v, present := em[k]; present {
+						mk[k] = v
+					}
+				}
+			}
+		}
+		idx.store.SetEntity("Market", poolID, mk)
 	}
 }