feat(eth-indexer): periodic stale-refresh sweep

[raymondjacobson]

Summary

Adds a background sweep inside the existing eth-indexer pod that re-reads the K oldest rows in eth_wallet_balances by updated_at every N seconds and upserts the fresh values. Complements the live WS subscription, which only learns about wallets that move AUDIO — the sweep handles drift correction, missed events during WS disconnects, and any wallet that's been silent on-chain (e.g. multi-wallet backfill placeholders).

How it works

• New goroutine ScheduleStaleRefresh launched from Start alongside runSubscriptionLoop.
• Each tick: SELECT wallet FROM eth_wallet_balances ORDER BY updated_at ASC LIMIT batchSize (uses eth_wallet_balances_updated_at_idx from migration 0203), then fan-out totalAudioBalance via the existing 8-worker pool, then one INSERT … ON CONFLICT … DO UPDATE.
• Reuses the event path's fan-out by extracting processLogs's tail into refreshAddresses(ctx, addrs, blockByAddr). blockByAddr is the event-block map for live events, nil for stale-refresh.

blocknumber semantics fix

The original ON CONFLICT … GREATEST(eth_wallet_balances.blocknumber, EXCLUDED.blocknumber) would write 0 over an existing block when the stale-refresh path didn't have one. Fixed:

NULLIF(unnest(@blocks::bigint[]), 0)   -- pass NULL when stale-refresh
…
ON CONFLICT … DO UPDATE SET
  blocknumber = CASE
    WHEN EXCLUDED.blocknumber IS NULL THEN eth_wallet_balances.blocknumber
    ELSE GREATEST(COALESCE(eth_wallet_balances.blocknumber, 0), EXCLUDED.blocknumber)
  END,

Smoke-checked against the local DB: real block (12345) preserved across both a NULL-block stale refresh and a lower-block event upsert.

Config

env var	default	effect
ethStaleRefreshIntervalSecs	30	Tick interval
ethStaleRefreshBatchSize	50	Wallets per tick

At defaults: ~1.7 wallets/sec, ~5 RPC calls/sec (each wallet = balanceOf + totalStakedFor + getTotalDelegatorStake in parallel). For ~3.15M tracked wallets, a full sweep takes ~22 days. Tune via env vars if you want faster freshness.

The ticker drops a tick if the previous run is still in flight, so a slow upstream can't pile up work. A deferred recover() guards against an unexpected panic taking down the WS subscription with it.

Smoke test (local)

1. Brought up local Postgres, applied migration 0203
2. Seeded two rows in eth_wallet_balances with updated_at='1970-01-01' to force them to the top of the sweep
3. Ran ethStaleRefreshIntervalSecs=5 ethStaleRefreshBatchSize=10 go run main.go eth-indexer against the live Alchemy URL
4. First tick log: stale refresh: tick complete requested:2 updated:2
5. Both rows' updated_at advanced; balance values matched independent eth_call checks (one was on-chain 0, one had moved AUDIO out since an earlier test — indexer correctly reflected both)

Test plan

• go build ./... clean
• go vet ./eth/... clean
• Local end-to-end: stale rows refreshed on first tick, no errors across multiple ticks
• SQL block-preserve verified (real block survives NULL stale-refresh upsert)
• After deploy: kubectl -n api logs deploy/eth-indexer | grep "stale refresh: tick complete" shows ticks at the configured interval
• After ~24h: SELECT MIN(updated_at) FROM eth_wallet_balances is no more than a few days old (proves the sweep is making progress)

Coordinating with the one-shot backfill SQL

Before this lands you'll want to run the backfill SQL from my prior message so eth_wallet_balances has rows. Otherwise the sweep just runs on an empty table and finds nothing to refresh until the live WS path discovers wallets organically. Specifically the third query (multi-wallet placeholders with updated_at='1970-01-01') is what the sweep will chew through first.

🤖 Generated with Claude Code