fix(indexer): valuation must be O(pools) writes, and V3 amounts are signed

Deployed v1.2.10 to a six-chain explorer and watched one chain produce
aggregates while two others produced nothing at all — no result line, no error
line. Two distinct defects, both found by that silence.

1. The pass wrote O(swaps) rows per interval.

   valueSwaps rewrote every swap's own amountUSD, one serialized INSERT OR
   REPLACE each, every 30s. A chain with a real trade history never finished a
   pass, so it never wrote its pool/factory aggregates either — while a quiet
   chain beside it looked perfectly healthy. Volume is a per-pool rollup and
   belongs on the pool: valueSwaps now READS the swap rows and writes none.
   (Per-swap amountUSD stays "0". It is a separate gap, and the honest fix
   values a swap at the price of its OWN block, not at today's.)

2. A V3 Swap emits int256, and the handler read it unsigned.

   One leg of every V3 swap is negative — the token leaving the pool. Reading
   it unsigned persisted ~2^256 in place of a small negative number. Nothing
   had ever added those amounts up, so the corruption was invisible until the
   volume rollup landed and reported $4.07e72 of trading on Zoo.

   handleSwapV3 now uses decodeInt256, the same decoder the V4 handler already
   used for the same reason. Rows the old build already wrote are healed on
   read rather than by a re-index: a stored leg at or above 2^255 IS the
   two's-complement encoding of a negative int256, and the V2 handler's
   difference of two uint112 reserves can never reach that magnitude, so the
   boundary is unambiguous.

Both were invisible for the same reason, so the third change is observability:

  - Config.Label names the indexer ("cchain/amm") and every indexer log line
    now carries it. One process runs seven of these; a line that does not say
    whose it is cannot be acted on.
  - Every revalue exit logs, including "read no balances", and the result line
    reports valued/total pools and elapsed time. An indexer that reports
    nothing was indistinguishable from one that was never scheduled.
  - Each pass runs under a 60s budget, so a pass that cannot finish is
    cancelled and says so instead of wedging its chain forever.

Tests pin all three: the wrapped-negative leg valuing identically to the signed
one, handleSwapV3 persisting "-2500", and revalue leaving swap rows untouched.

Co-authored-by: Hanzo Dev <[email protected]>

indexer/indexer.go

@@ -10,7 +10,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
-	"log"
 	"math/big"
 	"net/http"
 	"strconv"
@@ -49,6 +48,11 @@ type Config struct {
 	// StartBlock is the genesis-relative block to (re)index from. On a clean
 	// chain relaunch the indexer rewinds to this value (see poll's reorg guard).
 	StartBlock uint64
+	// Label names this indexer in its log lines — e.g. "cchain/amm". A process
+	// that runs one indexer per (chain, subgraph) is otherwise unable to say
+	// WHICH indexer emitted a line, and an indexer that never logs is then
+	// indistinguishable from one that is wedged. Empty => unprefixed.
+	Label string
 	// DexRPC, when non-empty, is the native D-Chain (dexvm) CLOB read-RPC root —
 	// e.g. http://node:9650/ext/bc/D/dex. Setting it runs an ADDITIONAL, orthogonal
 	// CLOBSource (clob.go) that polls dex_get_{markets,trades,orders} and writes
@@ -90,6 +94,7 @@ type Indexer struct {
 	poolManager string // lower-cased 0x9999 settlement address
 	factoryV2   string // lower-cased canonical V2 factory (PairCreated trust root)
 	factoryV3   string // lower-cased canonical V3 factory (PoolCreated trust root)
+	label       string // log prefix identifying this indexer (see Config.Label)
 	startBlock  uint64
 	store       *storage.Store
 	client      *http.Client
@@ -141,6 +146,7 @@ func NewWithConfig(cfg Config, store *storage.Store) *Indexer {
 		poolManager: strings.ToLower(pm),
 		factoryV2:   strings.ToLower(fv2),
 		factoryV3:   strings.ToLower(fv3),
+		label:       cfg.Label,
 		startBlock:  cfg.StartBlock,
 		store:       store,
 		client: &http.Client{
@@ -196,7 +202,7 @@ func (idx *Indexer) ensureGenesisHash(ctx context.Context) {
 	}
 	h, err := idx.genesisHashFn(ctx)
 	if err != nil {
-		log.Printf("[indexer] genesis hash baseline unavailable (will retry): %v", err)
+		idx.logf("[indexer] genesis hash baseline unavailable (will retry): %v", err)
 		return
 	}
 	idx.genesisHash = h
@@ -240,7 +246,7 @@ func (idx *Indexer) maybeResetForRegenesis(ctx context.Context, latest uint64) {
 	}
 	idx.lowHeadStreak++
 	if idx.lowHeadStreak < regenesisConfirmations {
-		log.Printf("[indexer] head %d << cursor %d (>%d behind) — suspected chain reset, awaiting confirmation %d/%d",
+		idx.logf("[indexer] head %d << cursor %d (>%d behind) — suspected chain reset, awaiting confirmation %d/%d",
 			latest, idx.lastBlock, reorgDepth, idx.lowHeadStreak, regenesisConfirmations)
 		return
 	}
@@ -248,27 +254,27 @@ func (idx *Indexer) maybeResetForRegenesis(ctx context.Context, latest uint64) {
 	// Streak confirmed — spend one RPC for the deterministic check.
 	current, err := idx.genesisHashFn(ctx)
 	if err != nil {
-		log.Printf("[indexer] head %d << cursor %d for %d polls but genesis-hash probe failed (%v) — "+
+		idx.logf("[indexer] head %d << cursor %d for %d polls but genesis-hash probe failed (%v) — "+
 			"NOT wiping; cannot confirm a reset without a canonical signal", latest, idx.lastBlock, idx.lowHeadStreak, err)
 		return
 	}
 	if idx.genesisHash == "" {
 		// No baseline yet — adopt the current hash and decline to wipe this round.
 		idx.genesisHash = current
-		log.Printf("[indexer] head %d << cursor %d but no genesis baseline yet — captured %s, NOT wiping",
+		idx.logf("[indexer] head %d << cursor %d but no genesis baseline yet — captured %s, NOT wiping",
 			latest, idx.lastBlock, current)
 		return
 	}
 	if current == idx.genesisHash {
 		// Same chain, just deeply behind/idle. The poll-count gate alone would have
 		// wiped here; the genesis hash proves it canonical, so we hold the cursor.
-		log.Printf("[indexer] head %d << cursor %d for %d polls but genesis hash unchanged (%s) — "+
+		idx.logf("[indexer] head %d << cursor %d for %d polls but genesis hash unchanged (%s) — "+
 			"canonical chain behind a stale backend, NOT wiping", latest, idx.lastBlock, idx.lowHeadStreak, current)
 		idx.lowHeadStreak = 0
 		return
 	}
 
-	log.Printf("[indexer] genesis hash changed %s -> %s (head %d << cursor %d) — re-genesis confirmed, rewinding to %d",
+	idx.logf("[indexer] genesis hash changed %s -> %s (head %d << cursor %d) — re-genesis confirmed, rewinding to %d",
 		idx.genesisHash, current, latest, idx.lastBlock, idx.startBlock)
 	idx.lastBlock = idx.startBlock
 	idx.store.SetLastBlock(idx.startBlock)
@@ -282,12 +288,12 @@ func (idx *Indexer) maybeResetForRegenesis(ctx context.Context, latest uint64) {
 // only the store. A native-DEX chain has its trading state on the D-Chain, not on
 // the EVM RPC, so both sources run to populate the AMM (EVM) and DEX (CLOB) views.
 func (idx *Indexer) Run(ctx context.Context) error {
-	log.Printf("[indexer] starting — rpc=%s lastBlock=%d", idx.rpc, idx.lastBlock)
+	idx.logf("[indexer] starting — rpc=%s lastBlock=%d", idx.rpc, idx.lastBlock)
 
 	if idx.clob != nil {
 		go func() {
 			if err := idx.clob.Run(ctx); err != nil && ctx.Err() == nil {
-				log.Printf("[indexer] clob source: %v", err)
+				idx.logf("[indexer] clob source: %v", err)
 			}
 		}()
 	}
@@ -307,7 +313,7 @@ func (idx *Indexer) Run(ctx context.Context) error {
 			return nil
 		case <-ticker.C:
 			if err := idx.pollGuarded(ctx); err != nil {
-				log.Printf("[indexer] poll error: %v", err)
+				idx.logf("[indexer] poll error: %v", err)
 			}
 		}
 	}
@@ -468,7 +474,7 @@ func (idx *Indexer) poll(ctx context.Context) error {
 	idx.store.SetLastBlock(toBlock)
 
 	if len(logs) > 0 {
-		log.Printf("[indexer] blocks %d..%d — %d events", fromBlock, toBlock, len(logs))
+		idx.logf("[indexer] blocks %d..%d — %d events", fromBlock, toBlock, len(logs))
 	}
 	return nil
 }
@@ -646,8 +652,12 @@ func (idx *Indexer) handleSwapV3(l *logEntry, blockNum uint64, txHash, logIdx st
 	if len(l.Topics) > 1 {
 		sender = topicAddr(l.Topics[1])
 	}
-	amount0 := decodeUint256(l.Data, 0)
-	amount1 := decodeUint256(l.Data, 1)
+	// A V3 Swap emits `int256 amount0, int256 amount1` — one leg of every swap
+	// is NEGATIVE (the token leaving the pool). Reading them unsigned turned
+	// that leg into ~2^256, which is invisible until something actually adds the
+	// amounts up: the volume rollup then reported astronomical figures.
+	amount0 := decodeInt256(l.Data, 0)
+	amount1 := decodeInt256(l.Data, 1)
 
 	idx.store.SeedSwap(id, &storage.SeedSwapData{
 		Timestamp: int64(blockNum),
@@ -1064,7 +1074,7 @@ func (idx *Indexer) BackfillTokens(ctx context.Context) (int, error) {
 			enriched++
 		}
 	}
-	log.Printf("[indexer] token backfill complete — %d/%d enriched", enriched, len(rows))
+	idx.logf("[indexer] token backfill complete — %d/%d enriched", enriched, len(rows))
 	return enriched, nil
 }
 

indexer/valuation.go

@@ -82,6 +82,12 @@ type valuedPool struct {
 	held0, held1 bool    // whether the balance read succeeded
 }
 
+// valuationBudget bounds one pass. A pass that cannot finish inside it is
+// cancelled and says so. Nothing here may run unboundedly: this loop shares a
+// process with six other chains' indexers, and a pass that quietly never
+// returns looks exactly like a pass that was never started.
+const valuationBudget = 60 * time.Second
+
 // runValuation drives the valuation pass on its own ticker until ctx ends. It
 // is started by Run alongside the log poller: orthogonal concerns, one store.
 func (idx *Indexer) runValuation(ctx context.Context) {
@@ -100,24 +106,43 @@ func (idx *Indexer) runValuation(ctx context.Context) {
 	}
 }
 
+// logf prefixes a message with the indexer's label so a multi-chain process can
+// tell whose pass emitted it. An unlabeled indexer (a test, a single-chain
+// binary) prints the bare message.
+func (idx *Indexer) logf(format string, args ...interface{}) {
+	if idx.label == "" {
+		log.Printf(format, args...)
+		return
+	}
+	log.Printf("["+idx.label+"] "+format, args...)
+}
+
 // revalue recomputes every derived USD aggregate from on-chain balances and
 // writes them back. Read-modify-write throughout: this pass owns the USD and
 // price fields and must preserve everything the log handlers own.
-func (idx *Indexer) revalue(ctx context.Context) {
+//
+// Every exit is logged. A pass that reports nothing is indistinguishable from
+// one that was never scheduled, and that ambiguity costs more to debug than the
+// log lines cost to emit.
+func (idx *Indexer) revalue(parent context.Context) {
 	defer func() {
 		if r := recover(); r != nil {
-			log.Printf("[valuation] recovered: %v", r)
+			idx.logf("[valuation] recovered: %v", r)
 		}
 	}()
+	ctx, cancel := context.WithTimeout(parent, valuationBudget)
+	defer cancel()
+	started := time.Now()
 
 	pools := idx.store.PoolsRaw()
 	if len(pools) == 0 {
-		return
+		return // nothing indexed yet for this subgraph; silence is correct here
 	}
 	tokens := idx.store.TokensRaw()
 
 	vps := idx.readBalances(ctx, pools, tokens)
 	if len(vps) == 0 {
+		idx.logf("[valuation] %d pools but no balances read — skipping (aggregates left as-is)", len(pools))
 		return
 	}
 	prices := priceTokens(vps, tokens)
@@ -200,8 +225,9 @@ func (idx *Indexer) revalue(ctx context.Context) {
 		TotalVolumeUSD:      fmtUSD(totalVol),
 	})
 
-	log.Printf("[valuation] %d pools, %d tokens priced — TVL $%s, volume $%s",
-		len(vps), len(prices), fmtUSD(totalTVL), fmtUSD(totalVol))
+	idx.logf("[valuation] %d/%d pools valued, %d tokens priced — TVL $%s, volume $%s (%s)",
+		len(vps), len(pools), len(prices), fmtUSD(totalTVL), fmtUSD(totalVol),
+		time.Since(started).Round(time.Millisecond))
 }
 
 // readBalances asks each pool's two tokens what the pool holds. One eth_call
@@ -316,9 +342,15 @@ func priceTokens(vps []valuedPool, tokens map[string]*storage.SeedTokenData) map
 }
 
 // valueSwaps prices the stored swap history and returns per-pool volume,
-// accumulating per-token volume into tokenVol. A swap whose USD value changed
-// is written back so the trade history carries the same number the aggregates
-// were built from — one valuation, used everywhere.
+// accumulating per-token volume into tokenVol.
+//
+// It READS the swap rows and writes nothing. An earlier cut also wrote each
+// swap's own amountUSD back, which turned one pass into O(swaps) serialized
+// INSERT OR REPLACEs every interval — on a chain with a real trade history the
+// pass never finished, so its chain silently produced no aggregates at all
+// while a quiet chain beside it looked fine. Volume is a per-pool rollup; it
+// belongs on the pool. (Per-swap amountUSD remains "0" — a separate gap, and
+// one that wants the price at the swap's own block, not today's.)
 func (idx *Indexer) valueSwaps(vps []valuedPool, tokens map[string]*storage.SeedTokenData, prices map[string]float64, tokenVol map[string]float64) map[string]float64 {
 	byPool := map[string]*valuedPool{}
 	for i := range vps {
@@ -328,40 +360,49 @@ func (idx *Indexer) valueSwaps(vps []valuedPool, tokens map[string]*storage.Seed
 	for _, vp := range vps {
 		out[vp.id] = 0
 	}
-	for id, sw := range idx.store.SwapsRaw() {
+	for _, sw := range idx.store.SwapsRaw() {
 		vp := byPool[strings.ToLower(sw.Pool)]
 		if vp == nil {
 			continue
 		}
 		v0, ok0 := swapLegUSD(sw.Amount0, vp.t0, tokens, prices)
 		v1, ok1 := swapLegUSD(sw.Amount1, vp.t1, tokens, prices)
-		var usd float64
 		switch {
 		case ok0 && ok1:
-			usd = (v0 + v1) / 2 // both legs priced: the mid of the two quotes
+			out[vp.id] += (v0 + v1) / 2 // both legs priced: the mid of the two quotes
 		case ok0:
-			usd = v0
+			out[vp.id] += v0
 		case ok1:
-			usd = v1
+			out[vp.id] += v1
 		default:
 			continue
 		}
-		out[vp.id] += usd
 		if ok0 {
 			tokenVol[vp.t0] += v0
 		}
 		if ok1 {
 			tokenVol[vp.t1] += v1
 		}
-		if s := fmtUSD(usd); s != sw.AmountUSD {
-			sw.AmountUSD = s
-			idx.store.SeedSwap(id, sw)
-		}
 	}
 	return out
 }
 
-// swapLegUSD converts one signed raw swap leg to its absolute USD value.
+// twoPow255 is the sign boundary of a 256-bit two's-complement word.
+var twoPow255 = new(big.Int).Lsh(big.NewInt(1), 255)
+
+// twoPow256 is 2^256, subtracted to recover the negative value of a word whose
+// high bit is set.
+var twoPow256 = new(big.Int).Lsh(big.NewInt(1), 256)
+
+// swapLegUSD converts one raw swap leg to its absolute USD value.
+//
+// A stored leg at or above 2^255 is the two's-complement encoding of a NEGATIVE
+// int256 that was persisted through an unsigned decode (the V3 handler's old
+// bug — a V3 Swap emits int256 amounts, and one leg of every swap is negative).
+// Reinterpreting it here is exact, not a heuristic: the V2 handler stores a
+// difference of two uint112 reserves, which can never reach that magnitude, so
+// the boundary is unambiguous and no re-index is needed to heal rows the old
+// build wrote.
 func swapLegUSD(raw, token string, tokens map[string]*storage.SeedTokenData, prices map[string]float64) (float64, bool) {
 	price, ok := prices[token]
 	if !ok || raw == "" {
@@ -371,7 +412,10 @@ func swapLegUSD(raw, token string, tokens map[string]*storage.SeedTokenData, pri
 	if !ok {
 		return 0, false
 	}
-	amt := normalize(new(big.Int).Abs(n), decimalsOf(token, tokens))
+	if n.Cmp(twoPow255) >= 0 {
+		n.Sub(n, twoPow256)
+	}
+	amt := normalize(n.Abs(n), decimalsOf(token, tokens))
 	return amt * price, true
 }
 

indexer/valuation_test.go

@@ -216,3 +216,97 @@ func mustFloat(t *testing.T, v interface{}) float64 {
 	}
 	return f
 }
+
+// TestSwapLegUSD_HealsUnsignedNegativeLeg pins the exact defect that made the
+// volume rollup report ~$4e72: a V3 Swap emits int256 amounts and one leg of
+// every swap is negative, but the handler decoded them unsigned, persisting
+// ~2^256 instead of a small negative number.
+func TestSwapLegUSD_HealsUnsignedNegativeLeg(t *testing.T) {
+	const token = "0x0000000000000000000000000000000000000001"
+	tokens := map[string]*storage.SeedTokenData{token: {Symbol: "USDC", Decimals: 18}}
+	prices := map[string]float64{token: 1}
+
+	// -5e18 stored through an unsigned decode: 2^256 - 5e18.
+	neg := new(big.Int).Mul(big.NewInt(-5), new(big.Int).Exp(big.NewInt(10), big.NewInt(18), nil))
+	wrapped := new(big.Int).Add(new(big.Int).Lsh(big.NewInt(1), 256), neg)
+
+	got, ok := swapLegUSD(wrapped.String(), token, tokens, prices)
+	if !ok {
+		t.Fatal("leg must be valued")
+	}
+	if math.Abs(got-5) > 1e-9 {
+		t.Errorf("wrapped negative leg = $%v, want $5 (got the 2^256 reading?)", got)
+	}
+
+	// A plainly-signed leg (what the fixed V3 handler and the V2 handler write)
+	// must value identically.
+	got2, _ := swapLegUSD(neg.String(), token, tokens, prices)
+	if math.Abs(got2-5) > 1e-9 {
+		t.Errorf("signed leg = $%v, want $5", got2)
+	}
+}
+
+// TestHandleSwapV3_DecodesSignedAmounts proves the handler now persists the
+// negative leg as a negative number rather than its 2^256 complement.
+func TestHandleSwapV3_DecodesSignedAmounts(t *testing.T) {
+	const (
+		token0 = "0x0000000000000000000000000000000000000011"
+		token1 = "0x0000000000000000000000000000000000000022"
+		pool   = "0x00000000000000000000000000000000000000c3"
+	)
+	s := newMemSQLiteStore(t)
+	idx := NewWithConfig(Config{RPC: "http://unused", FactoryV2: testFactoryV2, FactoryV3: testFactoryV3}, s)
+	idx.processLog(context.Background(), poolCreatedLog(testFactoryV3, token0, token1, 3000, pool))
+	// amount0 = +1000 in, amount1 = -2500 out.
+	idx.processLog(context.Background(), swapV3Log(pool, token0, "0xreal", 1000, -2500))
+
+	raw := s.SwapsRaw()
+	if len(raw) != 1 {
+		t.Fatalf("expected exactly one swap, got %d", len(raw))
+	}
+	for _, sw := range raw {
+		if sw.Amount1 != "-2500" {
+			t.Errorf("amount1 = %q, want \"-2500\" (an unsigned read gives a ~78-digit number)", sw.Amount1)
+		}
+		if sw.Amount0 != "1000" {
+			t.Errorf("amount0 = %q, want \"1000\"", sw.Amount0)
+		}
+	}
+}
+
+// TestRevalue_DoesNotRewriteSwapRows pins the performance defect that wedged a
+// whole chain: valuing must be O(pools) writes, never O(swaps) writes.
+func TestRevalue_DoesNotRewriteSwapRows(t *testing.T) {
+	const (
+		usdc = "0x0000000000000000000000000000000000000001"
+		wlux = "0x0000000000000000000000000000000000000003"
+		pool = "0x00000000000000000000000000000000000000d1"
+	)
+	s := newMemSQLiteStore(t)
+	s.SeedToken(usdc, &storage.SeedTokenData{Symbol: "USDC", Decimals: 18})
+	s.SeedToken(wlux, &storage.SeedTokenData{Symbol: "WLUX", Decimals: 18})
+	s.SeedPool(pool, &storage.SeedPoolData{Token0: wlux, Token1: usdc, FeeTier: 3000})
+	s.SeedSwap("swap-1", &storage.SeedSwapData{
+		Timestamp: 1, Pool: pool, Amount0: "1000000000000000000", Amount1: "-25000000000000000000",
+		AmountUSD: "0", Sender: "0xabc",
+	})
+
+	srv := balanceRPC(t, map[string]string{
+		wlux + "@" + pool: hexWord(4000),
+		usdc + "@" + pool: hexWord(100),
+	})
+	idx := NewWithConfig(Config{RPC: srv.URL}, s)
+	idx.revalue(context.Background())
+
+	// 1 WLUX @ $0.025 = $0.025; 25 USDC = $25; mid = $12.5125.
+	p, _ := s.GetPool(nil, pool)
+	if got := fmt.Sprint(p.(map[string]interface{})["volumeUSD"]); got != "12.51" {
+		t.Errorf("pool volumeUSD = %q, want 12.51", got)
+	}
+	// The swap row itself is untouched — valuing must not write O(swaps) rows.
+	for _, sw := range s.SwapsRaw() {
+		if sw.AmountUSD != "0" {
+			t.Errorf("valuation must not rewrite swap rows; amountUSD became %q", sw.AmountUSD)
+		}
+	}
+}