From 7724ba74261fd7444aad8d4dbdee6e0625badc9f Mon Sep 17 00:00:00 2001 From: moul <94029+moul@users.noreply.github.com> Date: Sun, 9 Aug 2026 14:14:50 +0200 Subject: [PATCH] perf: compute gas stats from stored transactions --- api.go | 151 +++++---------------------------------------- db.go | 178 +++++++++++++++++++++++++++++++++++++++++++++++++++++ db_test.go | 99 +++++++++++++++++++++++++++++ syncer.go | 69 +++++++++++++++++++++ 4 files changed, 362 insertions(+), 135 deletions(-) diff --git a/api.go b/api.go index 986f6df..f923dd6 100644 --- a/api.go +++ b/api.go @@ -1000,151 +1000,32 @@ func (a *API) HandleStorage(w http.ResponseWriter, r *http.Request) { func (a *API) HandleGas(w http.ResponseWriter, r *http.Request) { network := a.networkParam(r) - client := a.clientFor(network) - if client == nil { - jsonError(w, "no client available", 500) - return - } - txs, err := client.GetRecentTransactions(r.Context(), 0) + + // Computed from stored transactions rather than by downloading the chain: + // the numbers here are presented as all-time totals, so they cannot be + // approximated from a recent window. + stats, err := a.db.GetGasStats(network, 20) if err != nil { jsonError(w, err.Error(), 500) return } - var totalGasUsed, totalGasWanted, totalFees int - var successCount, failCount int - type RealmGas struct { - Path string `json:"path"` - Gas int `json:"gas"` - Fees int `json:"fees"` - TxCount int `json:"tx_count"` - } - type TopTx struct { - Hash string `json:"hash"` - BlockHeight int `json:"block_height"` - GasUsed int `json:"gas_used"` - GasWanted int `json:"gas_wanted"` - Fee int `json:"fee"` - Type string `json:"type"` - Detail string `json:"detail"` - Success bool `json:"success"` - } - realmMap := make(map[string]*RealmGas) - - for _, tx := range txs { - totalGasUsed += tx.GasUsed - totalGasWanted += tx.GasWanted - if tx.GasFee != nil { - totalFees += tx.GasFee.Amount - } - if tx.Success { - successCount++ - } else { - failCount++ - } - for _, m := range tx.Messages { - path := m.Value.PkgPath - if path == "" && m.Value.Package != nil { - path = m.Value.Package.Path - } - // Ephemeral packages (MsgRun): aggregate by caller address - if strings.Contains(path, "/e/") { - caller := m.Value.Caller - if caller == "" && m.Value.Creator != "" { - caller = m.Value.Creator - } - if caller != "" { - path = "MsgRun by " + caller - } - } - if path != "" { - rg, ok := realmMap[path] - if !ok { - rg = &RealmGas{Path: path} - realmMap[path] = rg - } - rg.Gas += tx.GasUsed - rg.TxCount++ - if tx.GasFee != nil { - rg.Fees += tx.GasFee.Amount - } - } - } - } - - // Sort realms by gas - var topRealms []RealmGas - for _, rg := range realmMap { - topRealms = append(topRealms, *rg) - } - sort.Slice(topRealms, func(i, j int) bool { return topRealms[i].Gas > topRealms[j].Gas }) - if len(topRealms) > 20 { - topRealms = topRealms[:20] - } - - // Top txs by gas - sorted := make([]Transaction, len(txs)) - copy(sorted, txs) - sort.Slice(sorted, func(i, j int) bool { return sorted[i].GasUsed > sorted[j].GasUsed }) - var topTxs []TopTx - for _, tx := range sorted { - if len(topTxs) >= 20 { - break - } - typ := "" - detail := "" - for _, m := range tx.Messages { - typ = m.Value.Typename - if m.Value.PkgPath != "" { - detail = m.Value.PkgPath - if m.Value.Func != "" { - detail += "::" + m.Value.Func - } - } else if m.Value.Package != nil { - p := m.Value.Package.Path - if strings.Contains(p, "/e/") { - // Ephemeral: show caller instead - caller := m.Value.Caller - if caller != "" { - detail = "MsgRun by " + caller - } else { - detail = p - } - } else { - detail = p - } - } - } - fee := 0 - if tx.GasFee != nil { - fee = tx.GasFee.Amount - } - topTxs = append(topTxs, TopTx{ - Hash: tx.Hash, BlockHeight: tx.BlockHeight, - GasUsed: tx.GasUsed, GasWanted: tx.GasWanted, - Fee: fee, Type: typ, Detail: detail, Success: tx.Success, - }) - } - - // Total source bytes from DB - totalStorageBytes := a.db.TotalSourceBytes(network) - avgGasPerTx := 0 - if len(txs) > 0 { - avgGasPerTx = totalGasUsed / len(txs) + if stats.TotalTxs > 0 { + avgGasPerTx = stats.TotalGasUsed / stats.TotalTxs } jsonResponse(w, map[string]any{ - "total_txs": len(txs), - "total_gas_used": totalGasUsed, - "total_gas_wanted": totalGasWanted, - "total_fees": totalFees, + "total_txs": stats.TotalTxs, + "total_gas_used": stats.TotalGasUsed, + "total_gas_wanted": stats.TotalGasWanted, + "total_fees": stats.TotalFees, "avg_gas_per_tx": avgGasPerTx, - "success_count": successCount, - "fail_count": failCount, - "total_source_bytes": totalStorageBytes, - "top_realms": topRealms, - "top_txs": topTxs, + "success_count": stats.SuccessCount, + "fail_count": stats.FailCount, + "total_source_bytes": a.db.TotalSourceBytes(network), + "top_realms": stats.TopRealms, + "top_txs": stats.TopTxs, }) } diff --git a/db.go b/db.go index a31bd33..78014fa 100644 --- a/db.go +++ b/db.go @@ -508,6 +508,50 @@ func (d *DB) HeightsMissingBlockTime(network string, limit int) ([]int, error) { return heights, rows.Err() } +// HeightsMissingTransactions returns block heights that have event rows with no +// corresponding entry in the transactions table, newest first, capped at limit. +// +// The transactions table was added after the event tables, and incremental sync +// only writes it going forward, so history synced by an older build has calls +// and transfers recorded with no transaction row carrying their gas. +func (d *DB) HeightsMissingTransactions(network string, limit int) ([]int, error) { + d.mu.RLock() + defer d.mu.RUnlock() + + var parts []string + for _, t := range []string{"packages", "calls", "msg_runs", "bank_sends"} { + parts = append(parts, fmt.Sprintf(` + SELECT DISTINCT e.block_height FROM %s e + WHERE e.network = ? + AND NOT EXISTS ( + SELECT 1 FROM transactions t + WHERE t.network = e.network AND t.tx_hash = e.tx_hash)`, t)) + } + query := strings.Join(parts, " UNION ") + " ORDER BY block_height DESC LIMIT ?" + + args := make([]any, 0, 5) + for i := 0; i < 4; i++ { + args = append(args, network) + } + args = append(args, limit) + + rows, err := d.db.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var heights []int + for rows.Next() { + var h int + if err := rows.Scan(&h); err != nil { + return nil, err + } + heights = append(heights, h) + } + return heights, rows.Err() +} + // SetBlockTimes fills in block_time for rows at the given heights that lack it. // Existing values are left alone: this repairs history, it does not rewrite it. func (d *DB) SetBlockTimes(network string, times map[int]string) (int64, error) { @@ -551,6 +595,140 @@ func (d *DB) SetBlockTimes(network string, times map[int]string) (int64, error) return updated, nil } +// GasRealm is per-realm gas consumption. +type GasRealm struct { + Path string `json:"path"` + Gas int `json:"gas"` + Fees int `json:"fees"` + TxCount int `json:"tx_count"` +} + +// GasTx is a single expensive transaction. +type GasTx struct { + Hash string `json:"hash"` + BlockHeight int `json:"block_height"` + GasUsed int `json:"gas_used"` + GasWanted int `json:"gas_wanted"` + Fee int `json:"fee"` + Type string `json:"type"` + Detail string `json:"detail"` + Success bool `json:"success"` +} + +// GasStats aggregates gas usage for a network. +type GasStats struct { + TotalTxs int + TotalGasUsed int + TotalGasWanted int + TotalFees int + SuccessCount int + FailCount int + TopRealms []GasRealm + TopTxs []GasTx +} + +// GetGasStats computes gas aggregates from stored transactions. +// +// Previously this was derived by downloading every transaction on the chain from +// the indexer on each request. The transactions table already carries gas_used, +// gas_wanted, gas_fee and success per network, so this is a handful of aggregates +// over local data instead. +func (d *DB) GetGasStats(network string, topN int) (*GasStats, error) { + d.mu.RLock() + defer d.mu.RUnlock() + + where, args := "", []any{} + if network != "" { + where = " WHERE network = ?" + args = append(args, network) + } + + out := &GasStats{} + err := d.db.QueryRow(` + SELECT COUNT(*), + COALESCE(SUM(gas_used), 0), + COALESCE(SUM(gas_wanted), 0), + COALESCE(SUM(gas_fee), 0), + COALESCE(SUM(CASE WHEN success THEN 1 ELSE 0 END), 0) + FROM transactions`+where, args...). + Scan(&out.TotalTxs, &out.TotalGasUsed, &out.TotalGasWanted, &out.TotalFees, &out.SuccessCount) + if err != nil { + return nil, fmt.Errorf("gas totals: %w", err) + } + out.FailCount = out.TotalTxs - out.SuccessCount + + // Attribute each transaction's gas to what it touched. A transaction is + // joined to at most one realm here; calls and deployments are the two paths + // that carry a package path, and MsgRun is grouped under its caller because + // its ephemeral path is unique per run and would otherwise be one row each. + realmWhere, realmArgs := "", []any{} + if network != "" { + realmWhere = " AND t.network = ?" + realmArgs = append(realmArgs, network, network, network) + } + rows, err := d.db.Query(` + SELECT path, SUM(gas_used), SUM(gas_fee), COUNT(*) FROM ( + SELECT c.pkg_path AS path, t.gas_used, t.gas_fee, t.tx_hash + FROM calls c JOIN transactions t + ON t.network = c.network AND t.tx_hash = c.tx_hash`+realmWhere+` + UNION ALL + SELECT p.path AS path, t.gas_used, t.gas_fee, t.tx_hash + FROM packages p JOIN transactions t + ON t.network = p.network AND t.tx_hash = p.tx_hash`+realmWhere+` + UNION ALL + SELECT 'MsgRun by ' || m.caller AS path, t.gas_used, t.gas_fee, t.tx_hash + FROM msg_runs m JOIN transactions t + ON t.network = m.network AND t.tx_hash = m.tx_hash`+realmWhere+` + ) GROUP BY path ORDER BY SUM(gas_used) DESC LIMIT ?`, + append(realmArgs, topN)...) + if err != nil { + return nil, fmt.Errorf("gas by realm: %w", err) + } + for rows.Next() { + var r GasRealm + if err := rows.Scan(&r.Path, &r.Gas, &r.Fees, &r.TxCount); err != nil { + rows.Close() + return nil, err + } + out.TopRealms = append(out.TopRealms, r) + } + rows.Close() + if err := rows.Err(); err != nil { + return nil, err + } + + // Most expensive transactions, with the type and target resolved from + // whichever table recorded the message. + txRows, err := d.db.Query(` + SELECT t.tx_hash, t.block_height, t.gas_used, t.gas_wanted, t.gas_fee, t.success, + COALESCE( + (SELECT 'MsgCall' FROM calls c WHERE c.network = t.network AND c.tx_hash = t.tx_hash LIMIT 1), + (SELECT 'MsgAddPackage' FROM packages p WHERE p.network = t.network AND p.tx_hash = t.tx_hash LIMIT 1), + (SELECT 'MsgRun' FROM msg_runs m WHERE m.network = t.network AND m.tx_hash = t.tx_hash LIMIT 1), + (SELECT 'BankMsgSend' FROM bank_sends b WHERE b.network = t.network AND b.tx_hash = t.tx_hash LIMIT 1), + ''), + COALESCE( + (SELECT c.pkg_path || '::' || c.func_name FROM calls c WHERE c.network = t.network AND c.tx_hash = t.tx_hash LIMIT 1), + (SELECT p.path FROM packages p WHERE p.network = t.network AND p.tx_hash = t.tx_hash LIMIT 1), + (SELECT 'MsgRun by ' || m.caller FROM msg_runs m WHERE m.network = t.network AND m.tx_hash = t.tx_hash LIMIT 1), + '') + FROM transactions t`+where+` + ORDER BY t.gas_used DESC LIMIT ?`, append(args, topN)...) + if err != nil { + return nil, fmt.Errorf("top gas transactions: %w", err) + } + defer txRows.Close() + for txRows.Next() { + var t GasTx + if err := txRows.Scan(&t.Hash, &t.BlockHeight, &t.GasUsed, &t.GasWanted, + &t.Fee, &t.Success, &t.Type, &t.Detail); err != nil { + return nil, err + } + out.TopTxs = append(out.TopTxs, t) + } + return out, txRows.Err() +} + // MaxBlockHeight returns the highest block height stored for a network, or 0 if // the network has no data yet. func (d *DB) MaxBlockHeight(network string) (int, error) { diff --git a/db_test.go b/db_test.go index 08bc16a..10542a3 100644 --- a/db_test.go +++ b/db_test.go @@ -281,3 +281,102 @@ func TestBackfillBlockTimes(t *testing.T) { t.Errorf("topaz block_time = %q, want it left untouched", topaz) } } + +func TestHeightsMissingTransactions(t *testing.T) { + db, err := NewDB(filepath.Join(t.TempDir(), "txgap.db")) + if err != nil { + t.Fatalf("open db: %v", err) + } + defer db.Close() + + // An event recorded with no transaction row behind it — the shape history + // synced by a build predating the transactions table is left in. + if err := db.InsertBankSend("gnoland1", "ORPHAN", 500, "", "g1a", "g1b", "1ugnot", true); err != nil { + t.Fatalf("seed orphan: %v", err) + } + // And one that is properly paired. + if err := db.InsertCall("gnoland1", "PAIRED", 600, "", "g1c", "gno.land/r/x", "F", true); err != nil { + t.Fatalf("seed call: %v", err) + } + if err := db.UpsertTransaction("gnoland1", "PAIRED", 600, "", 10, 20, 1, true); err != nil { + t.Fatalf("seed transaction: %v", err) + } + // Another network must not appear in the result. + if err := db.InsertBankSend("topaz", "OTHER", 700, "", "g1a", "g1b", "1ugnot", true); err != nil { + t.Fatalf("seed topaz: %v", err) + } + + heights, err := db.HeightsMissingTransactions("gnoland1", 100) + if err != nil { + t.Fatalf("find heights: %v", err) + } + if len(heights) != 1 || heights[0] != 500 { + t.Fatalf("heights = %v, want [500] (only the unpaired event)", heights) + } + + // Once the transaction row lands, the gap closes. + if err := db.UpsertTransaction("gnoland1", "ORPHAN", 500, "", 5, 6, 7, true); err != nil { + t.Fatalf("backfill transaction: %v", err) + } + heights, err = db.HeightsMissingTransactions("gnoland1", 100) + if err != nil { + t.Fatalf("find heights after backfill: %v", err) + } + if len(heights) != 0 { + t.Errorf("heights = %v after backfill, want none", heights) + } +} + +func TestGetGasStatsUsesStoredTransactions(t *testing.T) { + db, err := NewDB(filepath.Join(t.TempDir(), "gas.db")) + if err != nil { + t.Fatalf("open db: %v", err) + } + defer db.Close() + + if err := db.InsertCall("topaz", "T1", 10, "", "g1c", "gno.land/r/demo/hot", "Run", true); err != nil { + t.Fatalf("seed call: %v", err) + } + if err := db.UpsertTransaction("topaz", "T1", 10, "", 1000, 2000, 30, true); err != nil { + t.Fatalf("seed tx: %v", err) + } + if err := db.InsertCall("topaz", "T2", 11, "", "g1c", "gno.land/r/demo/hot", "Run", true); err != nil { + t.Fatalf("seed call: %v", err) + } + if err := db.UpsertTransaction("topaz", "T2", 11, "", 500, 900, 10, false); err != nil { + t.Fatalf("seed tx: %v", err) + } + // A different network's gas must not leak into the totals. + if err := db.UpsertTransaction("gnoland1", "OTHER", 12, "", 99999, 99999, 99999, true); err != nil { + t.Fatalf("seed other network: %v", err) + } + + stats, err := db.GetGasStats("topaz", 20) + if err != nil { + t.Fatalf("gas stats: %v", err) + } + if stats.TotalTxs != 2 { + t.Errorf("total txs = %d, want 2", stats.TotalTxs) + } + if stats.TotalGasUsed != 1500 { + t.Errorf("gas used = %d, want 1500", stats.TotalGasUsed) + } + if stats.TotalFees != 40 { + t.Errorf("fees = %d, want 40", stats.TotalFees) + } + if stats.SuccessCount != 1 || stats.FailCount != 1 { + t.Errorf("success/fail = %d/%d, want 1/1", stats.SuccessCount, stats.FailCount) + } + if len(stats.TopRealms) != 1 || stats.TopRealms[0].Path != "gno.land/r/demo/hot" { + t.Fatalf("top realms = %+v, want the one called realm", stats.TopRealms) + } + if stats.TopRealms[0].Gas != 1500 || stats.TopRealms[0].TxCount != 2 { + t.Errorf("realm gas/txs = %d/%d, want 1500/2", stats.TopRealms[0].Gas, stats.TopRealms[0].TxCount) + } + if len(stats.TopTxs) == 0 || stats.TopTxs[0].Hash != "T1" { + t.Errorf("top txs = %+v, want the most expensive first", stats.TopTxs) + } + if stats.TopTxs[0].Type != "MsgCall" { + t.Errorf("type = %q, want MsgCall resolved from the call row", stats.TopTxs[0].Type) + } +} diff --git a/syncer.go b/syncer.go index 69dddc4..cdb09a8 100644 --- a/syncer.go +++ b/syncer.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "log" + "sync" ) type Syncer struct { @@ -29,6 +30,7 @@ func (s *Syncer) SyncAll(ctx context.Context) error { } s.warnOnHeightRegression(ctx) s.backfillBlockTimes(ctx) + s.backfillTransactions(ctx) if err := s.syncPackages(ctx); err != nil { return err } @@ -110,6 +112,12 @@ func (s *Syncer) syncPackages(ctx context.Context) error { // cannot stall startup or hammer a public indexer. const backfillBatch = 200 +// Transaction repair is one indexer request per block, so it moves in smaller steps. +const ( + backfillTxBatch = 100 + backfillConcurrency = 10 +) + // backfillBlockTimes fills in block_time for rows written before that column // existed. // @@ -148,6 +156,67 @@ func (s *Syncer) backfillBlockTimes(ctx context.Context) { log.Printf("[%s] backfilled block_time on %d rows across %d blocks", s.networkID, updated, len(times)) } +// backfillTransactions fills in transaction rows for history that predates the +// transactions table. +// +// Event tables record what happened, but gas and fee figures live only on the +// transaction row. Without this, all-time gas totals computed from local storage +// silently under-report — on a live instance, 37 transactions out of 2738. +// +// Gas cannot be reconstructed from what is already stored, so the rows have to +// come back from the indexer. Bounded per pass, newest first, same as the +// block_time repair. +func (s *Syncer) backfillTransactions(ctx context.Context) { + heights, err := s.db.HeightsMissingTransactions(s.networkID, backfillTxBatch) + if err != nil { + log.Printf("[%s] transaction backfill: %v", s.networkID, err) + return + } + if len(heights) == 0 { + return + } + + // One request per block, run concurrently: sequential fetches made a large + // gap take the better part of an hour to close. + type blockTxs struct { + txs []Transaction + err error + } + results := make([]blockTxs, len(heights)) + var wg sync.WaitGroup + sem := make(chan struct{}, backfillConcurrency) + for i, h := range heights { + wg.Add(1) + go func(i, h int) { + defer wg.Done() + sem <- struct{}{} + defer func() { <-sem }() + txs, err := s.client.GetTransactionsByBlock(ctx, h) + results[i] = blockTxs{txs: txs, err: err} + }(i, h) + } + wg.Wait() + + var all []Transaction + for _, r := range results { + if r.err != nil { + // An unhealthy indexer: keep what we have and retry the rest next pass. + log.Printf("[%s] transaction backfill: %v", s.networkID, r.err) + break + } + all = append(all, r.txs...) + } + if len(all) == 0 { + return + } + + blockTimes := s.fetchBlockTimes(ctx, all) + for _, tx := range all { + s.upsertTx(tx, blockTimes[tx.BlockHeight]) + } + log.Printf("[%s] backfilled %d transactions across %d blocks", s.networkID, len(all), len(heights)) +} + // chainFingerprint identifies a specific chain instance by its first block. // The chain ID alone is not enough: a reset network keeps its chain ID and comes // back with a different block 1, which is exactly what portal-loop and staging