feat(indexer): enrich AMM token symbol/name/decimals from ERC20

Token entities were seeded with placeholder metadata (symbol=addr[:8],
name=full address, decimals=18) — no contract read happened, so the AMM
graph returned every token as its own address. Add erc20.go: read
symbol()/name()/decimals() via the existing eth_call RPC (canonical ABI
string + legacy bytes32 fallback, garbage-decimals guard), memoised once
per token (revert→placeholder cached too, never re-probed). Route all four
seed sites through a cache-first seedToken with a cold-start no-downgrade
guard. BackfillTokens (opt-in BACKFILL_TOKENS=1) enriches existing
placeholder rows in place — ≤3 eth_calls per token, no re-sync.
This commit is contained in:
2026-06-21 23:52:42 -07:00
parent 705e38c45e
commit 8c24d384da
5 changed files with 684 additions and 41 deletions
+12
View File
@@ -103,6 +103,18 @@ func main() {
FactoryV3: os.Getenv("FACTORY_V3"),
PoolManager: os.Getenv("POOL_MANAGER"),
}, store)
// One-shot enrichment of Token rows persisted by an older build with the
// address placeholder (symbol == shortAddr). Opt-in via BACKFILL_TOKENS=1 so
// a normal start is unchanged; runs before Run so live indexing is not racing
// the same SeedToken writes. Cheap: ≤3 eth_calls per placeholder token, no
// re-sync. New tokens are enriched on first sight regardless of this flag.
if os.Getenv("BACKFILL_TOKENS") == "1" {
if n, err := idx.BackfillTokens(ctx); err != nil {
slog.Warn("token backfill", "error", err, "enriched", n)
} else {
slog.Info("token backfill", "enriched", n)
}
}
go func() {
if err := idx.Run(ctx); err != nil && ctx.Err() == nil {
slog.Error("indexer", "error", err)
+19 -19
View File
@@ -90,7 +90,7 @@ func TestHandleSwapV4_WritesAmmSwapOnly(t *testing.T) {
TransactionHash: "0xdeadbeef",
LogIndex: "0x0",
}
idx.processLog(l)
idx.processLog(context.Background(), l)
// AMM swap present with correctly signed legs.
sw, err := s.GetSwaps(nil, 10, "timestamp", "desc", nil)
@@ -131,9 +131,9 @@ func TestHandleDEXFill_ScopedToPoolManager(t *testing.T) {
}
}
// Legit fill at 0x9999.
idx.processLog(mkFill(LXSettleAddress, "0xaa", 400))
idx.processLog(context.Background(), mkFill(LXSettleAddress, "0xaa", 400))
// Spoofed fill from a rogue address — must be dropped.
idx.processLog(mkFill("0x0000000000000000000000000000000000009998", "0xbb", 999999))
idx.processLog(context.Background(), mkFill("0x0000000000000000000000000000000000009998", "0xbb", 999999))
fills, _ := s.ListByType("Fill", 10)
fl := fills.([]interface{})
@@ -173,7 +173,7 @@ func TestHandleInitializeV4_PoolMarketFactory(t *testing.T) {
poolID := topic32("d00d")
cur0 := "0x0000000000000000000000000000000000000011"
cur1 := "0x0000000000000000000000000000000000000022"
idx.processLog(initV4Log(poolID, cur0, cur1, 3000, 60, 0, 0, "0x1"))
idx.processLog(context.Background(), initV4Log(poolID, cur0, cur1, 3000, 60, 0, 0, "0x1"))
if p, _ := s.GetPool(nil, poolID); p == nil {
t.Fatal("expected pool from InitializeV4")
@@ -209,14 +209,14 @@ func TestHandleInitializeV4_MergesStubVolume(t *testing.T) {
poolID := topic32("beef")
taker := "0x00000000000000000000000000000000000000aa"
// 1) A DEXFill lands first → writeFill creates a Market stub with volume=400.
idx.processLog(&logEntry{
idx.processLog(context.Background(), &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"))
idx.processLog(context.Background(), initV4Log(poolID, cur0, cur1, 500, 10, 0, 0, "0x6"))
mk, _ := s.GetByType("Market", poolID)
mm := mk.(map[string]interface{})
@@ -446,7 +446,7 @@ func TestHandleV4_RejectsNonPoolManager(t *testing.T) {
poolID := topic32("1111")
spoof := initV4Log(poolID, cur0, cur1, 3000, 60, 0, 0, "0x1")
spoof.Address = rogue // same wire shape, wrong emitter
idx.processLog(spoof)
idx.processLog(context.Background(), spoof)
if p, _ := s.GetPool(nil, poolID); p != nil {
t.Error("rogue Initialize must NOT seed a pool")
@@ -462,7 +462,7 @@ func TestHandleV4_RejectsNonPoolManager(t *testing.T) {
}
// The SAME event from the real 0x9999 IS honored (gate is the emitter, not the shape).
idx.processLog(initV4Log(poolID, cur0, cur1, 3000, 60, 0, 0, "0x1"))
idx.processLog(context.Background(), initV4Log(poolID, cur0, cur1, 3000, 60, 0, 0, "0x1"))
if p, _ := s.GetPool(nil, poolID); p == nil {
t.Fatal("canonical 0x9999 Initialize must seed a pool")
}
@@ -484,13 +484,13 @@ func TestHandleV4_RejectsNonPoolManager(t *testing.T) {
BlockNumber: "0x10", TransactionHash: tx, LogIndex: "0x0",
}
}
idx.processLog(mkSwap(rogue, "0xspoof")) // rogue — dropped
idx.processLog(context.Background(), mkSwap(rogue, "0xspoof")) // rogue — dropped
sw, _ := s.GetSwaps(nil, 10, "timestamp", "desc", nil)
if sw != nil && len(sw.([]interface{})) != 0 {
t.Fatalf("rogue Swap must NOT create an AMM swap, got %d", len(sw.([]interface{})))
}
idx.processLog(mkSwap(LXSettleAddress, "0xreal")) // canonical — honored
idx.processLog(context.Background(), mkSwap(LXSettleAddress, "0xreal")) // canonical — honored
sw, _ = s.GetSwaps(nil, 10, "timestamp", "desc", nil)
if sw == nil || len(sw.([]interface{})) != 1 {
t.Fatalf("canonical 0x9999 Swap must create exactly 1 AMM swap")
@@ -529,7 +529,7 @@ func TestPoll_MalformedLogDoesNotCrash(t *testing.T) {
t.Fatalf("malformed log %d panicked in processLog (must be guarded): %v", i, r)
}
}()
idx.processLog(l)
idx.processLog(context.Background(), l)
}()
}
@@ -643,7 +643,7 @@ func TestHandleV2_RejectsNonFactory(t *testing.T) {
idx := NewWithConfig(Config{RPC: "http://unused", FactoryV2: testFactoryV2, FactoryV3: testFactoryV3}, s)
// --- Rogue PairCreated: nothing seeded, factory not inflated. ---
idx.processLog(pairCreatedLog(rogue, token0, token1, rt))
idx.processLog(context.Background(), pairCreatedLog(rogue, token0, token1, rt))
if p, _ := s.GetPool(nil, rt); p != nil {
t.Error("rogue PairCreated must NOT seed a pair")
}
@@ -655,20 +655,20 @@ func TestHandleV2_RejectsNonFactory(t *testing.T) {
}
// --- A Swap from the rogue (never-registered) pair: no swap recorded. ---
idx.processLog(swapV2Log(rt, token0, "0xspoof", 1000, 2500))
idx.processLog(context.Background(), swapV2Log(rt, token0, "0xspoof", 1000, 2500))
if sw, _ := s.GetSwaps(nil, 10, "timestamp", "desc", nil); sw != nil && len(sw.([]interface{})) != 0 {
t.Fatalf("Swap from a rogue V2 pair must NOT be recorded, got %d", len(sw.([]interface{})))
}
// --- Canonical PairCreated IS honored; its pair's Swap IS recorded. ---
idx.processLog(pairCreatedLog(testFactoryV2, token0, token1, ct))
idx.processLog(context.Background(), pairCreatedLog(testFactoryV2, token0, token1, ct))
if p, _ := s.GetPool(nil, ct); p == nil {
t.Fatal("canonical PairCreated must seed the pair")
}
if f, _ := s.GetFactory(nil, "1"); f == nil || asInt64(f.(map[string]interface{})["poolCount"]) != 1 {
t.Fatalf("canonical PairCreated must bump factory poolCount to 1, got %v", f)
}
idx.processLog(swapV2Log(ct, token0, "0xreal", 1000, 2500))
idx.processLog(context.Background(), swapV2Log(ct, token0, "0xreal", 1000, 2500))
sw, _ := s.GetSwaps(nil, 10, "timestamp", "desc", nil)
if sw == nil || len(sw.([]interface{})) != 1 {
t.Fatalf("Swap from the canonical V2 pair must be recorded exactly once, got %v", sw)
@@ -686,7 +686,7 @@ func TestHandleV3_RejectsNonFactory(t *testing.T) {
idx := NewWithConfig(Config{RPC: "http://unused", FactoryV2: testFactoryV2, FactoryV3: testFactoryV3}, s)
// --- Rogue PoolCreated: nothing seeded, factory not inflated. ---
idx.processLog(poolCreatedLog(rogue, token0, token1, 3000, rt))
idx.processLog(context.Background(), poolCreatedLog(rogue, token0, token1, 3000, rt))
if p, _ := s.GetPool(nil, rt); p != nil {
t.Error("rogue PoolCreated must NOT seed a pool")
}
@@ -695,20 +695,20 @@ func TestHandleV3_RejectsNonFactory(t *testing.T) {
}
// --- A Swap from the rogue (never-registered) pool: no swap recorded. ---
idx.processLog(swapV3Log(rt, token0, "0xspoof", 1000, -2500))
idx.processLog(context.Background(), swapV3Log(rt, token0, "0xspoof", 1000, -2500))
if sw, _ := s.GetSwaps(nil, 10, "timestamp", "desc", nil); sw != nil && len(sw.([]interface{})) != 0 {
t.Fatalf("Swap from a rogue V3 pool must NOT be recorded, got %d", len(sw.([]interface{})))
}
// --- Canonical PoolCreated IS honored; its pool's Swap IS recorded with signed legs. ---
idx.processLog(poolCreatedLog(testFactoryV3, token0, token1, 3000, ct))
idx.processLog(context.Background(), poolCreatedLog(testFactoryV3, token0, token1, 3000, ct))
if p, _ := s.GetPool(nil, ct); p == nil {
t.Fatal("canonical PoolCreated must seed the pool")
}
if f, _ := s.GetFactory(nil, "1"); f == nil || asInt64(f.(map[string]interface{})["poolCount"]) != 1 {
t.Fatalf("canonical PoolCreated must bump factory poolCount to 1, got %v", f)
}
idx.processLog(swapV3Log(ct, token0, "0xreal", 1000, 2500))
idx.processLog(context.Background(), swapV3Log(ct, token0, "0xreal", 1000, 2500))
sw, _ := s.GetSwaps(nil, 10, "timestamp", "desc", nil)
if sw == nil || len(sw.([]interface{})) != 1 {
t.Fatalf("Swap from the canonical V3 pool must be recorded exactly once, got %v", sw)
+216
View File
@@ -0,0 +1,216 @@
package indexer
import (
"context"
"encoding/json"
"strings"
"github.com/luxfi/graph/storage"
)
// ERC20 metadata selectors (first 4 bytes of keccak256 of the signature).
//
// symbol() => 0x95d89b41
// name() => 0x06fdde03
// decimals() => 0x313ce567
//
// AMM Token entities are token *contracts* (an ERC20 underlying, or the
// pair/LP token itself, which is also an ERC20 — e.g. "UNI-V2"). The pool
// creation logs only carry the address; the human symbol/name/decimals live
// on the contract and must be read with eth_call. (This is distinct from the
// no-eth_getCode re-genesis rule, which forbids eth_getCode of the *cursor*
// block for chain-identity probing — an eth_call of a token contract at head
// is a normal, cheap read and is cached one-per-token below.)
const (
selSymbol = "0x95d89b41"
selName = "0x06fdde03"
selDecimals = "0x313ce567"
)
// defaultDecimals is the ERC20 default when decimals() is absent or reverts.
const defaultDecimals = 18
// tokenMeta resolves (and memoises) the ERC20 metadata for a token address.
// It is the single source of truth for symbol/name/decimals enrichment: every
// pool/pair/transfer handler routes its token writes through seedToken, which
// calls this once per address for the lifetime of the process.
//
// Caching is mandatory, not an optimisation: Transfer fires on every transfer
// and PairCreated/PoolCreated re-reference the same tokens, so an un-cached
// read would issue three eth_calls per token *per log* — a per-block RPC storm
// against the chain. The cache collapses that to three eth_calls the first
// time a token is seen and zero thereafter; reverting/non-compliant tokens are
// cached as a placeholder so they are likewise probed only once.
func (idx *Indexer) tokenMeta(ctx context.Context, addr string) *storage.SeedTokenData {
key := strings.ToLower(addr)
if v, ok := idx.tokenCache.Load(key); ok {
return v.(*storage.SeedTokenData)
}
meta := idx.readERC20(ctx, addr)
// LoadOrStore so two goroutines racing the same fresh token converge on one
// cached value (handlers run on a single poll goroutine today, but the cache
// must not depend on that).
actual, _ := idx.tokenCache.LoadOrStore(key, meta)
return actual.(*storage.SeedTokenData)
}
// readERC20 performs the three metadata eth_calls and decodes them, applying
// graceful fallbacks so a non-compliant or reverting token still yields a sane
// Token entity (placeholder symbol/name = address, decimals = 18) rather than
// dropping the token or failing the poll.
func (idx *Indexer) readERC20(ctx context.Context, addr string) *storage.SeedTokenData {
out := &storage.SeedTokenData{
Symbol: shortAddr(addr), // fallbacks, overwritten on a successful read
Name: addr,
Decimals: defaultDecimals,
}
if raw, err := idx.erc20Call(ctx, addr, selSymbol); err == nil {
if s := decodeERC20String(raw); s != "" {
out.Symbol = s
}
}
if raw, err := idx.erc20Call(ctx, addr, selName); err == nil {
if s := decodeERC20String(raw); s != "" {
out.Name = s
}
}
if raw, err := idx.erc20Call(ctx, addr, selDecimals); err == nil {
if d, ok := decodeERC20Decimals(raw); ok {
out.Decimals = d
}
}
return out
}
// erc20Call issues `eth_call {to: addr, data: selector}` at the latest block
// and returns the 0x-prefixed hex result. Reuses the indexer's existing
// JSON-RPC client and endpoint — the same transport eth_getLogs rides on.
func (idx *Indexer) erc20Call(ctx context.Context, addr, selector string) (string, error) {
raw, err := idx.rpcCall(ctx, "eth_call", []interface{}{
map[string]string{"to": addr, "data": selector},
"latest",
})
if err != nil {
return "", err
}
var hex string
if err := json.Unmarshal(raw, &hex); err != nil {
return "", err
}
return hex, nil
}
// decodeERC20String decodes an ERC20 string return value. It accepts BOTH
// shapes seen in the wild:
//
// - Canonical ABI `string`: word0 = data offset (0x20), word1 = byte length,
// then the UTF-8 bytes right-padded to a 32-byte boundary.
// - Legacy `bytes32` (e.g. MKR, original DAI): a single 32-byte word holding
// the symbol/name as right-padded NUL bytes, with NO offset/length header.
//
// Returns "" when the value is empty, all-zero, or malformed so the caller
// keeps its address placeholder.
func decodeERC20String(hexStr string) string {
b := hexToBytes(hexStr)
if len(b) == 0 {
return ""
}
// Canonical ABI string: at least offset(32) + length(32); offset must point
// at 0x20 and the declared length must fit in the payload.
if len(b) >= 64 {
offset := beUint(b[:32])
if offset == 32 && len(b) >= 64 {
length := beUint(b[32:64])
if length > 0 && 64+int(length) <= len(b) {
return sanitizeUTF8(string(b[64 : 64+int(length)]))
}
// length == 0 is a legitimately empty ABI string.
if length == 0 {
return ""
}
}
}
// bytes32 fallback: exactly one word, interpreted as right-padded NUL bytes.
if len(b) == 32 {
return sanitizeUTF8(strings.TrimRight(string(b), "\x00"))
}
return ""
}
// decodeERC20Decimals reads a uint8 decimals() return (a 32-byte word). Returns
// ok=false on an empty/short result so the caller keeps the ERC20 default (18).
func decodeERC20Decimals(hexStr string) (int64, bool) {
b := hexToBytes(hexStr)
if len(b) < 32 {
return 0, false
}
// decimals is a uint8; read the low byte of the first word. Guard against an
// absurd value (a non-compliant contract returning garbage) by rejecting
// anything that does not fit a uint8 — the upper 31 bytes must be zero.
for _, x := range b[:31] {
if x != 0 {
return 0, false
}
}
return int64(b[31]), true
}
// --- byte helpers (pure, table-tested) ---
// hexToBytes decodes a 0x-prefixed hex string, tolerating odd length and
// non-hex input by returning nil (callers treat nil as "no data").
func hexToBytes(s string) []byte {
s = strings.TrimPrefix(strings.TrimSpace(s), "0x")
if len(s) == 0 || len(s)%2 != 0 {
return nil
}
out := make([]byte, len(s)/2)
for i := 0; i < len(out); i++ {
hi, ok1 := hexNibble(s[i*2])
lo, ok2 := hexNibble(s[i*2+1])
if !ok1 || !ok2 {
return nil
}
out[i] = hi<<4 | lo
}
return out
}
func hexNibble(c byte) (byte, bool) {
switch {
case c >= '0' && c <= '9':
return c - '0', true
case c >= 'a' && c <= 'f':
return c - 'a' + 10, true
case c >= 'A' && c <= 'F':
return c - 'A' + 10, true
}
return 0, false
}
// beUint reads a big-endian unsigned integer from up to 8 trailing bytes of a
// word. ABI offsets/lengths far exceed 2^64 only in adversarial input, which
// the length-fits-payload check above rejects anyway.
func beUint(word []byte) int {
var n uint64
start := 0
if len(word) > 8 {
start = len(word) - 8 // keep the low 8 bytes; high bytes are zero for sane lengths
}
for _, x := range word[start:] {
n = n<<8 | uint64(x)
}
return int(n)
}
// sanitizeUTF8 trims surrounding whitespace/NULs and drops any embedded NULs so
// a malformed bytes32 padding never leaks control bytes into the GraphQL JSON.
func sanitizeUTF8(s string) string {
s = strings.ReplaceAll(s, "\x00", "")
return strings.TrimSpace(s)
}
+298
View File
@@ -0,0 +1,298 @@
package indexer
import (
"context"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"github.com/luxfi/graph/storage"
)
// TestDecodeERC20String covers both on-wire shapes plus the malformed/empty
// fallbacks. The "real" cases are the exact bytes Lux C-Chain mainnet returns
// for 0xf07b65… (symbol "UNI-V2", name "Uniswap V2") — the token the bug
// reported as symbol="0xf07b65".
func TestDecodeERC20String(t *testing.T) {
cases := []struct {
name string
hex string
want string
}{
{
name: "abi string UNI-V2 (real on-chain symbol)",
hex: "0x" +
"0000000000000000000000000000000000000000000000000000000000000020" + // offset
"0000000000000000000000000000000000000000000000000000000000000006" + // len 6
"554e492d56320000000000000000000000000000000000000000000000000000", // "UNI-V2"
want: "UNI-V2",
},
{
name: "abi string Uniswap V2 (real on-chain name)",
hex: "0x" +
"0000000000000000000000000000000000000000000000000000000000000020" +
"000000000000000000000000000000000000000000000000000000000000000a" + // len 10
"556e697377617020563200000000000000000000000000000000000000000000", // "Uniswap V2"
want: "Uniswap V2",
},
{
name: "bytes32 symbol (MKR-style, no offset/length header)",
// "MKR" right-padded in a single 32-byte word.
hex: "0x4d4b520000000000000000000000000000000000000000000000000000000000",
want: "MKR",
},
{
name: "empty abi string (length 0)",
hex: "0x" +
"0000000000000000000000000000000000000000000000000000000000000020" +
"0000000000000000000000000000000000000000000000000000000000000000",
want: "",
},
{name: "empty result", hex: "0x", want: ""},
{name: "0x prefix only with no data", hex: "", want: ""},
{name: "all-zero bytes32", hex: "0x" + strings.Repeat("0", 64), want: ""},
{
name: "abi string with declared length past payload (malformed) -> empty",
hex: "0x" +
"0000000000000000000000000000000000000000000000000000000000000020" +
"00000000000000000000000000000000000000000000000000000000000000ff", // claims 255 bytes, none present
want: "",
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := decodeERC20String(tc.hex); got != tc.want {
t.Errorf("decodeERC20String(%s) = %q, want %q", tc.hex, got, tc.want)
}
})
}
}
func TestDecodeERC20Decimals(t *testing.T) {
cases := []struct {
name string
hex string
want int64
wantOK bool
}{
{"18 decimals (real on-chain)", "0x" + "00000000000000000000000000000000000000000000000000000000000000" + "12", 18, true},
{"6 decimals (USDC-style)", "0x" + strings.Repeat("0", 62) + "06", 6, true},
{"0 decimals", "0x" + strings.Repeat("0", 64), 0, true},
{"empty -> not ok", "0x", 0, false},
{"short word -> not ok", "0x12", 0, false},
{
name: "garbage high bytes (non-uint8) -> not ok",
hex: "0x" + "ff" + strings.Repeat("0", 60) + "12",
want: 0,
wantOK: false,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got, ok := decodeERC20Decimals(tc.hex)
if ok != tc.wantOK || (ok && got != tc.want) {
t.Errorf("decodeERC20Decimals(%s) = (%d,%v), want (%d,%v)", tc.hex, got, ok, tc.want, tc.wantOK)
}
})
}
}
// fakeERC20RPC is an httptest server that answers eth_call for the three
// metadata selectors with canned ABI-encoded values, and counts the calls so a
// test can assert the per-token cache (one call set per token, never per-block).
type fakeERC20RPC struct {
symbolHex string
nameHex string
decimalsHex string
calls int32 // total eth_call requests served
}
func (f *fakeERC20RPC) handler() http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
body, _ := io.ReadAll(r.Body)
var req struct {
Method string `json:"method"`
Params []interface{} `json:"params"`
}
_ = json.Unmarshal(body, &req)
result := "0x"
if req.Method == "eth_call" {
atomic.AddInt32(&f.calls, 1)
if call, ok := req.Params[0].(map[string]interface{}); ok {
switch call["data"] {
case selSymbol:
result = f.symbolHex
case selName:
result = f.nameHex
case selDecimals:
result = f.decimalsHex
}
}
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"result":"` + result + `"}`))
}
}
// TestSeedToken_EnrichesAndCaches proves the fix end to end: a token is enriched
// with the real ERC20 symbol/name/decimals from the chain, persisted, and read
// at most once regardless of how many times the handlers re-reference it.
func TestSeedToken_EnrichesAndCaches(t *testing.T) {
fake := &fakeERC20RPC{
symbolHex: "0x" +
"0000000000000000000000000000000000000000000000000000000000000020" +
"0000000000000000000000000000000000000000000000000000000000000006" +
"554e492d56320000000000000000000000000000000000000000000000000000", // UNI-V2
nameHex: "0x" +
"0000000000000000000000000000000000000000000000000000000000000020" +
"000000000000000000000000000000000000000000000000000000000000000a" +
"556e697377617020563200000000000000000000000000000000000000000000", // Uniswap V2
decimalsHex: "0x" + strings.Repeat("0", 62) + "12", // 18
}
srv := httptest.NewServer(fake.handler())
defer srv.Close()
s := newMemSQLiteStore(t)
idx := NewWithConfig(Config{RPC: srv.URL}, s)
const addr = "0xf07b65def8cbe9f2645157bf69e3e5212d3ced9d"
// First sight: reads the chain and persists the real metadata.
idx.seedToken(context.Background(), addr)
tok, err := s.GetToken(nil, addr)
if err != nil || tok == nil {
t.Fatalf("token not persisted: %v", err)
}
m := tok.(map[string]interface{})
if m["symbol"] != "UNI-V2" {
t.Errorf("symbol = %v, want UNI-V2", m["symbol"])
}
if m["name"] != "Uniswap V2" {
t.Errorf("name = %v, want Uniswap V2", m["name"])
}
if asInt64(m["decimals"]) != 18 {
t.Errorf("decimals = %v, want 18", m["decimals"])
}
callsAfterFirst := atomic.LoadInt32(&fake.calls)
if callsAfterFirst != 3 {
t.Fatalf("first sight must issue exactly 3 eth_calls (symbol/name/decimals), got %d", callsAfterFirst)
}
// Re-reference the same token many times (as Transfer/Swap handlers do every
// block): the cache must serve it with ZERO additional RPC.
for i := 0; i < 50; i++ {
idx.seedToken(context.Background(), addr)
}
if got := atomic.LoadInt32(&fake.calls); got != callsAfterFirst {
t.Fatalf("cache breach: %d eth_calls after 50 re-references, want %d (one read per token, never per-block)", got, callsAfterFirst)
}
}
// TestSeedToken_FallbackOnRevert proves a non-compliant/reverting token still
// yields a sane Token entity (address placeholder, 18 decimals) rather than
// dropping it or failing the poll.
func TestSeedToken_FallbackOnRevert(t *testing.T) {
// Server that errors every eth_call (simulates a revert / non-ERC20 contract).
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"execution reverted"}}`))
}))
defer srv.Close()
s := newMemSQLiteStore(t)
idx := NewWithConfig(Config{RPC: srv.URL}, s)
const addr = "0x00000000000000000000000000000000000000ff"
idx.seedToken(context.Background(), addr)
tok, _ := s.GetToken(nil, addr)
if tok == nil {
t.Fatal("token must still be seeded on revert (with placeholder)")
}
m := tok.(map[string]interface{})
if m["symbol"] != shortAddr(addr) {
t.Errorf("symbol = %v, want placeholder %v", m["symbol"], shortAddr(addr))
}
if asInt64(m["decimals"]) != defaultDecimals {
t.Errorf("decimals = %v, want default %d", m["decimals"], defaultDecimals)
}
}
// TestBackfillTokens enriches placeholder rows (from an older build) without a
// re-sync, skips rows that already have a real symbol, and leaves a reverting
// token as a (cached) placeholder.
func TestBackfillTokens(t *testing.T) {
fake := &fakeERC20RPC{
symbolHex: "0x" +
"0000000000000000000000000000000000000000000000000000000000000020" +
"0000000000000000000000000000000000000000000000000000000000000006" +
"554e492d56320000000000000000000000000000000000000000000000000000", // UNI-V2
nameHex: "0x" +
"0000000000000000000000000000000000000000000000000000000000000020" +
"000000000000000000000000000000000000000000000000000000000000000a" +
"556e697377617020563200000000000000000000000000000000000000000000", // Uniswap V2
decimalsHex: "0x" + strings.Repeat("0", 62) + "12",
}
srv := httptest.NewServer(fake.handler())
defer srv.Close()
s := newMemSQLiteStore(t)
idx := NewWithConfig(Config{RPC: srv.URL}, s)
const placeholder = "0xf07b65def8cbe9f2645157bf69e3e5212d3ced9d"
const alreadyGood = "0x1111111111111111111111111111111111111111"
// A placeholder row exactly as the buggy build wrote it.
s.SeedToken(placeholder, &storage.SeedTokenData{Symbol: shortAddr(placeholder), Name: placeholder, Decimals: 18})
// A row already enriched by the fixed build — must be left untouched (0 RPC).
s.SeedToken(alreadyGood, &storage.SeedTokenData{Symbol: "WLUX", Name: "Wrapped LUX", Decimals: 18})
n, err := idx.BackfillTokens(context.Background())
if err != nil {
t.Fatal(err)
}
if n != 1 {
t.Fatalf("backfill enriched %d, want 1 (only the placeholder)", n)
}
// The already-good row must not have been probed.
if got := atomic.LoadInt32(&fake.calls); got != 3 {
t.Fatalf("backfill issued %d eth_calls, want 3 (one placeholder token only)", got)
}
tok, _ := s.GetToken(nil, placeholder)
if m := tok.(map[string]interface{}); m["symbol"] != "UNI-V2" || m["name"] != "Uniswap V2" {
t.Errorf("placeholder not enriched: %v", m)
}
good, _ := s.GetToken(nil, alreadyGood)
if m := good.(map[string]interface{}); m["symbol"] != "WLUX" {
t.Errorf("already-good row was clobbered: %v", m)
}
}
// TestSeedToken_NoDowngradeOnColdStartBlip proves the cold-start clobber guard:
// an already-enriched stored token is NOT overwritten with the address
// placeholder when a transient RPC failure occurs after a restart (empty cache).
func TestSeedToken_NoDowngradeOnColdStartBlip(t *testing.T) {
s := newMemSQLiteStore(t)
const addr = "0xf07b65def8cbe9f2645157bf69e3e5212d3ced9d"
// Simulate a prior run having persisted real metadata.
s.SeedToken(addr, &storage.SeedTokenData{Symbol: "UNI-V2", Name: "Uniswap V2", Decimals: 18})
// New process, empty cache, RPC unreachable (blip).
idx := NewWithConfig(Config{RPC: "http://127.0.0.1:0"}, s)
idx.seedToken(context.Background(), addr)
tok, _ := s.GetToken(nil, addr)
m := tok.(map[string]interface{})
if m["symbol"] != "UNI-V2" {
t.Errorf("cold-start blip downgraded symbol to %v; must keep stored UNI-V2", m["symbol"])
}
}
+139 -22
View File
@@ -15,6 +15,7 @@ import (
"net/http"
"strconv"
"strings"
"sync"
"time"
"github.com/luxfi/graph/storage"
@@ -91,6 +92,12 @@ type Indexer struct {
genesisHash string // chain-identity fingerprint; a change ⇒ true re-genesis (lazily captured)
status Status
// tokenCache memoises ERC20 metadata (symbol/name/decimals) per token
// address for the process lifetime so each token is read from the chain at
// most once — never per-block. Keyed lower-cased address → *SeedTokenData.
// See erc20.go (tokenMeta/seedToken).
tokenCache sync.Map
// genesisHashFn fetches the chain's genesis (block 0) hash. It is a field so
// tests can drive the canonicality decision deterministically without a live
// RPC; in production it is fetchGenesisHash (eth_getBlockByNumber 0x0). Honors
@@ -419,7 +426,7 @@ func (idx *Indexer) poll(ctx context.Context) error {
// 3. Process each log
for i := range logs {
idx.processLog(&logs[i])
idx.processLog(ctx, &logs[i])
idx.status.IndexedEvents++
}
@@ -434,8 +441,10 @@ func (idx *Indexer) poll(ctx context.Context) error {
return nil
}
// processLog matches a log entry's topic0 and writes to storage.
func (idx *Indexer) processLog(l *logEntry) {
// processLog matches a log entry's topic0 and writes to storage. ctx is
// threaded through so the token-seeding handlers can read ERC20 metadata via
// eth_call on the same RPC the poll loop already holds a context for.
func (idx *Indexer) processLog(ctx context.Context, l *logEntry) {
if len(l.Topics) == 0 {
return
}
@@ -450,13 +459,13 @@ func (idx *Indexer) processLog(l *logEntry) {
case SigSwapV3:
idx.handleSwapV3(l, blockNum, txHash, logIdx)
case SigPairCreated:
idx.handlePairCreated(l)
idx.handlePairCreated(ctx, l)
case SigPoolCreated:
idx.handlePoolCreated(l)
idx.handlePoolCreated(ctx, l)
case SigTransfer:
idx.handleTransfer(l, txHash, logIdx)
idx.handleTransfer(ctx, l, txHash, logIdx)
case SigInitializeV4:
idx.handleInitializeV4(l)
idx.handleInitializeV4(ctx, l)
case SigSwapV4:
idx.handleSwapV4(l, blockNum, txHash, logIdx)
case SigDEXFill:
@@ -614,7 +623,7 @@ func (idx *Indexer) handleSwapV3(l *logEntry, blockNum uint64, txHash, logIdx st
})
}
func (idx *Indexer) handlePairCreated(l *logEntry) {
func (idx *Indexer) handlePairCreated(ctx context.Context, l *logEntry) {
// Scope to the canonical V2 factory (the trust root), exactly as the V4
// handlers scope to 0x9999. A PairCreated from any other contract is rejected:
// honoring it would let anyone seed phantom pairs/tokens and inflate the
@@ -639,12 +648,12 @@ func (idx *Indexer) handlePairCreated(l *logEntry) {
Token1: token1,
FeeTier: 3000,
})
idx.store.SeedToken(token0, &storage.SeedTokenData{Symbol: shortAddr(token0), Name: token0, Decimals: 18})
idx.store.SeedToken(token1, &storage.SeedTokenData{Symbol: shortAddr(token1), Name: token1, Decimals: 18})
idx.seedToken(ctx, token0)
idx.seedToken(ctx, token1)
idx.bumpFactory(1)
}
func (idx *Indexer) handlePoolCreated(l *logEntry) {
func (idx *Indexer) handlePoolCreated(ctx context.Context, l *logEntry) {
// Scope to the canonical V3 factory (the trust root), mirroring handlePairCreated
// and the V4 0x9999 gate. A PoolCreated from a rogue contract is rejected so it
// can neither inflate the Factory/TVL aggregates nor register a rogue pool whose
@@ -670,17 +679,17 @@ func (idx *Indexer) handlePoolCreated(l *logEntry) {
Token1: token1,
FeeTier: fee,
})
idx.store.SeedToken(token0, &storage.SeedTokenData{Symbol: shortAddr(token0), Name: token0, Decimals: 18})
idx.store.SeedToken(token1, &storage.SeedTokenData{Symbol: shortAddr(token1), Name: token1, Decimals: 18})
idx.seedToken(ctx, token0)
idx.seedToken(ctx, token1)
idx.bumpFactory(1)
}
func (idx *Indexer) handleTransfer(l *logEntry, txHash, logIdx string) {
// ERC20 Transfer — just record the token if we see it
func (idx *Indexer) handleTransfer(ctx context.Context, l *logEntry, txHash, logIdx string) {
// ERC20 Transfer — record the token if we see it. seedToken reads the real
// symbol/name/decimals from the chain once and caches, so the per-transfer
// firing of this handler does NOT translate into per-block RPC.
if len(l.Topics) >= 3 {
idx.store.SeedToken(l.Address, &storage.SeedTokenData{
Symbol: shortAddr(l.Address), Name: l.Address, Decimals: 18,
})
idx.seedToken(ctx, l.Address)
}
}
@@ -758,7 +767,7 @@ func asInt64(v interface{}) int64 {
// pool id is the indexed bytes32 `id`; token0/token1 are the indexed
// currency0/currency1. Fee is the first data word. Pools created by the 0x9999
// settlement manager are ALSO surfaced as DEX (CLOB) Market entities.
func (idx *Indexer) handleInitializeV4(l *logEntry) {
func (idx *Indexer) handleInitializeV4(ctx context.Context, l *logEntry) {
// Scope to the canonical settlement manager (0x9999), exactly like
// handleDEXFill. Any contract can emit a lookalike Initialize; honoring them
// would let anyone seed phantom pools/tokens and inflate the explorer's public
@@ -776,8 +785,8 @@ func (idx *Indexer) handleInitializeV4(l *logEntry) {
Token1: token1,
FeeTier: fee,
})
idx.store.SeedToken(token0, &storage.SeedTokenData{Symbol: shortAddr(token0), Name: token0, Decimals: 18})
idx.store.SeedToken(token1, &storage.SeedTokenData{Symbol: shortAddr(token1), Name: token1, Decimals: 18})
idx.seedToken(ctx, token0)
idx.seedToken(ctx, token1)
idx.bumpFactory(1)
// We only reach here for the 0x9999 PoolManager (gated above), so every
@@ -895,10 +904,118 @@ func (idx *Indexer) isKnownPool(addr string) bool {
return p != nil
}
// shortAddr returns a stable short symbol for an unknown token address.
// shortAddr returns a stable short symbol for an unknown token address. It is
// the fallback when the chain has no ERC20 symbol()/name() to offer (a revert,
// a non-compliant contract, or an unreachable RPC) — see readERC20.
func shortAddr(addr string) string {
if len(addr) >= 8 {
return addr[:8]
}
return addr
}
// seedToken records a Token entity with ERC20 metadata read from the chain.
// It is the ONE place every pool/pair/transfer handler routes token writes
// through; it owns both the read-once cache (tokenMeta) and the persistence,
// so the four handlers stay in their lane and never duplicate enrichment.
//
// Cold-start clobber guard: SeedToken is INSERT OR REPLACE, and the in-memory
// cache is empty after a restart. If a token was already enriched in a prior
// run (its stored symbol differs from the address placeholder), the cheap
// metadata read might transiently fail (RPC blip) and we must NOT overwrite a
// good stored name with the address fallback. So on a cache miss we consult the
// store: an already-enriched row short-circuits the read entirely, and a freshly
// read placeholder never downgrades an existing real symbol.
func (idx *Indexer) seedToken(ctx context.Context, addr string) {
key := strings.ToLower(addr)
// Already settled this process (resolved metadata or a cached placeholder):
// the value is in the store from the first sight, so the hot Transfer path
// costs zero RPC AND zero DB writes on every subsequent block.
if _, cached := idx.tokenCache.Load(key); cached {
return
}
// First sight this process. If the store already holds a real (non-placeholder)
// symbol from a previous run, adopt it as the cached value and skip the RPC —
// the chain metadata is immutable, so a re-read would only cost RPC and risk a
// blip-induced downgrade.
if existing := idx.storedToken(addr); existing != nil && !isPlaceholderSymbol(existing.Symbol, addr) {
idx.tokenCache.Store(key, existing)
return
}
meta := idx.tokenMeta(ctx, addr)
idx.store.SeedToken(addr, meta)
}
// BackfillTokens enriches Token rows that were persisted with the address
// placeholder (symbol == shortAddr(addr)) by an earlier build that did not read
// ERC20 metadata. It is the cheap, idempotent alternative to a full re-sync:
// it issues at most three eth_calls per *placeholder* token (already-enriched
// rows are skipped) and rewrites only those rows — no block re-scan, no cursor
// rewind, no effect on swaps/pools/factory aggregates.
//
// Safe to call at startup before Run, or as a one-shot maintenance pass. Returns
// the number of tokens enriched. Honors ctx cancellation between tokens.
func (idx *Indexer) BackfillTokens(ctx context.Context) (int, error) {
const pageSize = 1 << 20 // GetTokens caps internally; this asks for "all"
raw, err := idx.store.GetTokens(ctx, pageSize, "", "", nil)
if err != nil {
return 0, err
}
rows, ok := raw.([]interface{})
if !ok {
return 0, nil
}
enriched := 0
for _, r := range rows {
if ctx.Err() != nil {
return enriched, ctx.Err()
}
m, ok := r.(map[string]interface{})
if !ok {
continue
}
addr, _ := m["id"].(string)
symbol, _ := m["symbol"].(string)
if addr == "" || !isPlaceholderSymbol(symbol, addr) {
continue // already has a real symbol — nothing to do
}
before := symbol
idx.seedToken(ctx, addr)
// Count only rows that actually changed (the read may have reverted and
// kept the placeholder — that token is now cached and won't be retried).
if after := idx.storedToken(addr); after != nil && !isPlaceholderSymbol(after.Symbol, addr) && after.Symbol != before {
enriched++
}
}
log.Printf("[indexer] token backfill complete — %d/%d enriched", enriched, len(rows))
return enriched, nil
}
// storedToken loads a token's persisted metadata, or nil if absent. Used only
// by the cold-start guard in seedToken.
func (idx *Indexer) storedToken(addr string) *storage.SeedTokenData {
t, err := idx.store.GetToken(nil, addr)
if err != nil || t == nil {
return nil
}
m, ok := t.(map[string]interface{})
if !ok {
return nil
}
out := &storage.SeedTokenData{Decimals: defaultDecimals}
if s, ok := m["symbol"].(string); ok {
out.Symbol = s
}
if s, ok := m["name"].(string); ok {
out.Name = s
}
out.Decimals = asInt64(m["decimals"])
return out
}
// isPlaceholderSymbol reports whether a stored symbol is the address-derived
// fallback (shortAddr) rather than a real ERC20 symbol — i.e. a row that still
// needs enriching.
func isPlaceholderSymbol(symbol, addr string) bool {
return symbol == "" || strings.EqualFold(symbol, shortAddr(addr))
}