diff --git a/cmd/integration/commands/state_history.go b/cmd/integration/commands/state_history.go index b100ce2056f..46b91817c9a 100644 --- a/cmd/integration/commands/state_history.go +++ b/cmd/integration/commands/state_history.go @@ -17,8 +17,12 @@ package commands import ( + "bytes" "context" + "encoding/binary" + "errors" "fmt" + "math" "sort" "github.com/spf13/cobra" @@ -26,6 +30,7 @@ import ( "github.com/erigontech/erigon/common" "github.com/erigontech/erigon/common/log/v3" "github.com/erigontech/erigon/db/datadir" + "github.com/erigontech/erigon/db/etl" "github.com/erigontech/erigon/db/kv" "github.com/erigontech/erigon/db/state" "github.com/erigontech/erigon/db/state/statecfg" @@ -44,8 +49,15 @@ func init() { withDataDir2(distributionCmd) withHistoryDomain(distributionCmd) + duplicatesCmd.Flags().Uint64Var(&fromStep, "from", 0, "step from which to scan history") + duplicatesCmd.Flags().Uint64Var(&toStep, "to", 1e18, "step up to which to scan history") + duplicatesCmd.Flags().StringVar(&historyDomain, "domain", "", "restrict scan to one domain (accounts, storage, code, commitment, receipt, rcache); default: all present") + duplicatesCmd.Flags().IntVar(&dupSamples, "samples", 3, "number of example keys with duplicates to print per domain") + withDataDir2(duplicatesCmd) + historyCmd.AddCommand(printCmd) historyCmd.AddCommand(distributionCmd) + historyCmd.AddCommand(duplicatesCmd) rootCmd.AddCommand(historyCmd) } @@ -64,6 +76,7 @@ var ( toStep uint64 historyKey string historyDomain string + dupSamples int ) var historyCmd = &cobra.Command{ @@ -89,10 +102,29 @@ func openHistory(ctx context.Context, dirs datadir.Dirs, domainName string, scan if err != nil { return nil, nil, fmt.Errorf("init history: %w", err) } - history.Scan(ctx, scanToStep*settings.StepSize) + scanToTxNum, err := stepToTxNum(scanToStep, settings.StepSize) + if err != nil { + return nil, nil, err + } + if err := history.Scan(ctx, scanToTxNum); err != nil { + return nil, nil, fmt.Errorf("scan history files: %w", err) + } return history, settings, nil } +// stepToTxNum converts a step bound to a txNum bound, saturating at MaxUint64 +// rather than wrapping: --to defaults to 1e18, which overflows for every real +// step size. +func stepToTxNum(step, stepSize uint64) (uint64, error) { + if stepSize == 0 { + return 0, errors.New("invalid stepSize=0") + } + if step > math.MaxUint64/stepSize { + return math.MaxUint64, nil + } + return step * stepSize, nil +} + var printCmd = &cobra.Command{ Use: "print", Run: func(cmd *cobra.Command, args []string) { @@ -110,7 +142,11 @@ var printCmd = &cobra.Command{ logger.Error("Failed to open history", "error", err) return } - stepSize := settings.StepSize + dumpFrom, dumpTo, err := stepDumpBounds(settings.StepSize) + if err != nil { + logger.Error("Invalid step range", "error", err) + return + } roTx := history.BeginFilesRoForDebug() defer roTx.Close() @@ -123,11 +159,12 @@ var printCmd = &cobra.Command{ } err = roTx.HistoryDump( - int(fromStep)*int(stepSize), - int(toStep)*int(stepSize), + dumpFrom, + dumpTo, keyToDump, - func(key []byte, txNum uint64, val []byte) { + func(key []byte, txNum uint64, val []byte) error { fmt.Printf("key: %x, txn: %d, val: %x\n", key, txNum, val) + return nil }, ) if err != nil { @@ -154,7 +191,11 @@ var distributionCmd = &cobra.Command{ logger.Error("Failed to open history", "error", err) return } - stepSize := settings.StepSize + dumpFrom, dumpTo, err := stepDumpBounds(settings.StepSize) + if err != nil { + logger.Error("Invalid step range", "error", err) + return + } roTx := history.BeginFilesRoForDebug() defer roTx.Close() @@ -163,14 +204,13 @@ var distributionCmd = &cobra.Command{ uniqueEntries := 0 err = roTx.HistoryDump( - int(fromStep)*int(stepSize), - int(toStep)*int(stepSize), + dumpFrom, + dumpTo, nil, - func(key []byte, txNum uint64, val []byte) { + func(key []byte, txNum uint64, val []byte) error { keysEntries[string(key)] += 1 uniqueEntries++ - - //fmt.Printf("key: %x, txn: %d, val: %x\n", key, txNum, val) + return nil }, ) if err != nil { @@ -223,3 +263,240 @@ var distributionCmd = &cobra.Command{ } }, } + +// histDupScan counts, per domain, how many history entries repeat the previous +// value for the same key. HistoryDump yields entries grouped by key and ordered +// by txNum, so a consecutive equal value is a redundant row (an as-of read +// collapses it away). Pure and stateless w.r.t. storage — fed one entry at a time. +type histDupScan struct { + sampleLimit int + + prevKey []byte + prevVal []byte + havePrev bool + curDup bool + + Entries uint64 + DistinctKeys uint64 + KeysWithDup uint64 + DupPairs uint64 + SampleKeys [][]byte +} + +func (s *histDupScan) observe(key, val []byte) { + s.Entries++ + if s.havePrev && bytes.Equal(key, s.prevKey) { + if bytes.Equal(val, s.prevVal) { + s.DupPairs++ + if !s.curDup { + s.curDup = true + if len(s.SampleKeys) < s.sampleLimit { + s.SampleKeys = append(s.SampleKeys, bytes.Clone(key)) + } + } + } + } else { + s.closeKey() + s.DistinctKeys++ + s.curDup = false + } + s.prevKey = append(s.prevKey[:0], key...) + s.prevVal = append(s.prevVal[:0], val...) + s.havePrev = true +} + +func (s *histDupScan) closeKey() { + if s.curDup { + s.KeysWithDup++ + } +} + +func (s *histDupScan) finish() { s.closeKey() } + +func historyDomainNames() []string { + names := make([]string, 0, kv.DomainLen) + for d := range kv.DomainLen { + names = append(names, d.String()) + } + return names +} + +// errHistoryNotInFiles marks a domain whose history has not been collated into +// files yet. HistoryDump reads frozen .ef/.v only, so that domain's DB-resident +// history is not covered and the run must not report it as clean. +var errHistoryNotInFiles = errors.New("history not collated into files yet, so it was not scanned") + +// stepDumpBounds resolves the --from/--to step flags to HistoryDump's arguments. +func stepDumpBounds(stepSize uint64) (int, int, error) { + fromTxNum, err := stepToTxNum(fromStep, stepSize) + if err != nil { + return 0, 0, err + } + toTxNum, err := stepToTxNum(toStep, stepSize) + if err != nil { + return 0, 0, err + } + from, to := dumpBounds(fromTxNum, toTxNum) + return from, to, nil +} + +// dumpBounds converts txNum bounds to HistoryDump's int arguments, using its -1 +// "unbounded" for anything that does not fit. HistoryDump filters whole files +// only, so these are a coarse pre-filter; the exact bound is applied per entry. +func dumpBounds(fromTxNum, toTxNum uint64) (int, int) { + maxInt := uint64(^uint(0) >> 1) + from, to := -1, -1 + if fromTxNum <= maxInt { + from = int(fromTxNum) + } + if toTxNum <= maxInt { + to = int(toTxNum) + } + return from, to +} + +// histDupSorter replays entries sorted by key||txNum. HistoryDump walks files +// outer and keys inner, so the same key reappears once per .ef file with every +// other key in between; sorting is what makes "the previous entry for this key" +// mean the previous entry chain-wide rather than within one file. +type histDupSorter struct { + collector *etl.Collector + txNumBuf [8]byte +} + +func newHistDupSorter(logPrefix, tmpdir string, logger log.Logger) *histDupSorter { + return &histDupSorter{collector: etl.NewCollectorWithAllocator(logPrefix, tmpdir, etl.SmallSortableBuffers, logger)} +} + +func (s *histDupSorter) Close() { s.collector.Close() } + +func (s *histDupSorter) add(key []byte, txNum uint64, val []byte) error { + binary.BigEndian.PutUint64(s.txNumBuf[:], txNum) + return s.collector.Collect(append(bytes.Clone(key), s.txNumBuf[:]...), val) +} + +func (s *histDupSorter) scan(ctx context.Context, sampleLimit int) (*histDupScan, error) { + scan := &histDupScan{sampleLimit: sampleLimit} + // bucket "" with a nil tx: ETL is a sort scratch-pad here, and that pair + // also keeps an empty value (a deletion marker) from being dropped. + if err := s.collector.Load(nil, "", func(k, v []byte, _ etl.CurrentTableReader, next etl.LoadNextFunc) error { + if err := ctx.Err(); err != nil { + return err + } + scan.observe(k[:len(k)-8], v) + return nil + }, etl.TransformArgs{Quit: ctx.Done()}); err != nil { + return nil, err + } + scan.finish() + return scan, nil +} + +func scanDomainDuplicates(ctx context.Context, dirs datadir.Dirs, name string, logger log.Logger) (*histDupScan, error) { + history, settings, err := openHistory(ctx, dirs, name, toStep, logger) + if err != nil { + return nil, err + } + defer history.Close() + + roTx := history.BeginFilesRoForDebug() + defer roTx.Close() + if len(roTx.Files()) == 0 { + return nil, errHistoryNotInFiles + } + + fromTxNum, err := stepToTxNum(fromStep, settings.StepSize) + if err != nil { + return nil, err + } + toTxNum, err := stepToTxNum(toStep, settings.StepSize) + if err != nil { + return nil, err + } + dumpFrom, dumpTo := dumpBounds(fromTxNum, toTxNum) + + sorter := newHistDupSorter(name+" history duplicates", dirs.Tmp, logger) + defer sorter.Close() + + if err := roTx.HistoryDump(dumpFrom, dumpTo, nil, func(key []byte, txNum uint64, val []byte) error { + if err := ctx.Err(); err != nil { + return err + } + // HistoryDump's own from/to filtering is per file, so a partially + // overlapping file yields entries outside the requested range. + if txNum < fromTxNum || txNum >= toTxNum { + return nil + } + return sorter.add(key, txNum, val) + }); err != nil { + return nil, err + } + return sorter.scan(ctx, dupSamples) +} + +var duplicatesCmd = &cobra.Command{ + Use: "duplicates", + Short: "Report keys whose history has consecutive duplicate (redundant) values, per domain", + RunE: func(cmd *cobra.Command, args []string) error { + logger := debug.SetupCobra(cmd, "integration") + + if dupSamples < 0 { + return fmt.Errorf("--samples must be >= 0, got %d", dupSamples) + } + if toStep < fromStep { + return fmt.Errorf("--to (%d) must be >= --from (%d)", toStep, fromStep) + } + + dirs, l, err := datadir.New(datadirCli).MustFlock() + if err != nil { + return fmt.Errorf("opening datadir: %w", err) + } + defer l.Unlock() + + names := historyDomainNames() + if historyDomain != "" { + if _, err := kv.String2Domain(historyDomain); err != nil { + return fmt.Errorf("--domain: %w", err) + } + names = []string{historyDomain} + } + + ctx := cmd.Context() + var withDup, unscanned []string + for _, name := range names { + scan, err := scanDomainDuplicates(ctx, dirs, name, logger) + switch { + case errors.Is(err, errHistoryNotInFiles): + fmt.Printf("domain=%-11s %s\n", name, err) + unscanned = append(unscanned, name) + continue + case err != nil: + // Every domain must be accounted for: a partial scan reported as + // clean is worse than no scan at all. + return fmt.Errorf("scan domain %s: %w", name, err) + } + if scan.Entries == 0 { + fmt.Printf("domain=%-11s no history entries in the requested range\n", name) + continue + } + pct := float64(scan.DupPairs) * 100 / float64(scan.Entries) + fmt.Printf("domain=%-11s entries=%-12d distinctKeys=%-12d keysWithDup=%-10d dupPairs=%-10d (%.2f%% of entries)\n", + name, scan.Entries, scan.DistinctKeys, scan.KeysWithDup, scan.DupPairs, pct) + if scan.DupPairs > 0 { + withDup = append(withDup, name) + for _, k := range scan.SampleKeys { + fmt.Printf(" example key with duplicates: %x\n", k) + } + } + } + if len(withDup) > 0 { + fmt.Printf("domains with duplicate history values: %v\n", withDup) + return nil + } + if len(unscanned) > 0 { + return fmt.Errorf("scan incomplete: %v have history only in the DB", unscanned) + } + fmt.Println("no consecutive duplicate history values found") + return nil + }, +} diff --git a/cmd/integration/commands/state_history_test.go b/cmd/integration/commands/state_history_test.go new file mode 100644 index 00000000000..3b96876549c --- /dev/null +++ b/cmd/integration/commands/state_history_test.go @@ -0,0 +1,168 @@ +// Copyright 2024 The Erigon Authors +// This file is part of Erigon. +// +// Erigon is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// Erigon is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with Erigon. If not, see . + +package commands + +import ( + "math" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/erigontech/erigon/common/log/v3" +) + +func TestHistDupScan(t *testing.T) { + t.Parallel() + + feed := func(s *histDupScan, entries [][2]string) { + for _, e := range entries { + s.observe([]byte(e[0]), []byte(e[1])) + } + s.finish() + } + + t.Run("mixed", func(t *testing.T) { + t.Parallel() + s := &histDupScan{sampleLimit: 10} + feed(s, [][2]string{ + {"A", "v1"}, {"A", "v1"}, {"A", "v2"}, {"A", "v2"}, {"A", "v2"}, // 3 dup pairs + {"B", "v1"}, {"B", "v2"}, // no dup + {"C", "v1"}, {"C", "v1"}, // 1 dup pair + }) + require.Equal(t, uint64(9), s.Entries) + require.Equal(t, uint64(3), s.DistinctKeys) + require.Equal(t, uint64(2), s.KeysWithDup) // A, C + require.Equal(t, uint64(4), s.DupPairs) // A:3 + C:1 + require.Equal(t, [][]byte{[]byte("A"), []byte("C")}, s.SampleKeys) + }) + + t.Run("no duplicates", func(t *testing.T) { + t.Parallel() + s := &histDupScan{sampleLimit: 10} + feed(s, [][2]string{{"A", "v1"}, {"A", "v2"}, {"B", "v1"}}) + require.Equal(t, uint64(3), s.Entries) + require.Equal(t, uint64(2), s.DistinctKeys) + require.Zero(t, s.KeysWithDup) + require.Zero(t, s.DupPairs) + require.Empty(t, s.SampleKeys) + }) + + t.Run("empty", func(t *testing.T) { + t.Parallel() + s := &histDupScan{sampleLimit: 10} + feed(s, nil) + require.Zero(t, s.Entries) + require.Zero(t, s.DistinctKeys) + require.Zero(t, s.KeysWithDup) + require.Zero(t, s.DupPairs) + }) + + t.Run("sample limit respected", func(t *testing.T) { + t.Parallel() + s := &histDupScan{sampleLimit: 1} + feed(s, [][2]string{{"A", "v"}, {"A", "v"}, {"B", "v"}, {"B", "v"}}) + require.Equal(t, uint64(2), s.KeysWithDup) + require.Len(t, s.SampleKeys, 1) // capped at sampleLimit + }) +} + +// TestHistDupSorter_FileOrderIndependent pins the reason the scan sorts at all: +// HistoryDump yields entries file-major, so a key's chain arrives split across +// files with unrelated keys in between. Counting on that raw order misses a +// duplicate pair straddling a file boundary and counts one key many times. +func TestHistDupSorter_FileOrderIndependent(t *testing.T) { + t.Parallel() + + type entry struct { + key string + txNum uint64 + val string + } + // Key A: v1@1, v1@3 — a duplicate pair straddling the file boundary. + // Key B: v1@2, v2@4 — no duplicate, and it separates A's two entries. + fileMajor := []entry{ + {"A", 1, "v1"}, {"B", 2, "v1"}, // .ef file 1 + {"A", 3, "v1"}, {"B", 4, "v2"}, // .ef file 2 + } + + run := func(t *testing.T, entries []entry) *histDupScan { + t.Helper() + sorter := newHistDupSorter(t.Name(), t.TempDir(), log.New()) + t.Cleanup(sorter.Close) + for _, e := range entries { + require.NoError(t, sorter.add([]byte(e.key), e.txNum, []byte(e.val))) + } + scan, err := sorter.scan(t.Context(), 10) + require.NoError(t, err) + return scan + } + + got := run(t, fileMajor) + require.Equal(t, uint64(4), got.Entries) + require.Equal(t, uint64(2), got.DistinctKeys, "a key spanning two files is still one key") + require.Equal(t, uint64(1), got.DupPairs, "the pair straddling the file boundary must be counted") + require.Equal(t, uint64(1), got.KeysWithDup) + require.Equal(t, [][]byte{[]byte("A")}, got.SampleKeys) + + // Same entries handed over in a different order must produce the same report. + shuffled := []entry{fileMajor[3], fileMajor[0], fileMajor[2], fileMajor[1]} + require.Equal(t, got, run(t, shuffled)) +} + +func TestHistDupSorter_KeepsEmptyValues(t *testing.T) { + t.Parallel() + + sorter := newHistDupSorter(t.Name(), t.TempDir(), log.New()) + t.Cleanup(sorter.Close) + // An empty value is a deletion marker, and two in a row are as redundant as + // any other repeat — ETL must not drop them. + require.NoError(t, sorter.add([]byte("A"), 1, nil)) + require.NoError(t, sorter.add([]byte("A"), 2, []byte{})) + + scan, err := sorter.scan(t.Context(), 10) + require.NoError(t, err) + require.Equal(t, uint64(2), scan.Entries) + require.Equal(t, uint64(1), scan.DupPairs) +} + +func TestStepToTxNum_SaturatesInsteadOfWrapping(t *testing.T) { + t.Parallel() + + // The --to default: 1e18 steps times any real step size overflows uint64. + got, err := stepToTxNum(1e18, 1_562_500) + require.NoError(t, err) + require.Equal(t, uint64(math.MaxUint64), got, "an out-of-range bound must saturate, not wrap") + + got, err = stepToTxNum(4, 1_562_500) + require.NoError(t, err) + require.Equal(t, uint64(6_250_000), got) + + _, err = stepToTxNum(1, 0) + require.Error(t, err) +} + +func TestDumpBounds_UnboundedWhenOutOfIntRange(t *testing.T) { + t.Parallel() + + from, to := dumpBounds(10, math.MaxUint64) + require.Equal(t, 10, from) + require.Equal(t, -1, to, "HistoryDump reads -1 as unbounded") + + from, to = dumpBounds(0, 100) + require.Equal(t, 0, from) + require.Equal(t, 100, to) +} diff --git a/db/state/history.go b/db/state/history.go index 85334b34777..c29238c62bf 100644 --- a/db/state/history.go +++ b/db/state/history.go @@ -1420,7 +1420,7 @@ func (ht *HistoryRoTx) HistoryKeyTxNumRange(fromTxNum, toTxNum int, asc order.By // HistoryDump walks every value in the visible files and hands each to dumpTo. // val is only valid until dumpTo returns: it points into a page buffer the next // value decodes over. -func (ht *HistoryRoTx) HistoryDump(fromTxNum, toTxNum int, keyToDump *[]byte, dumpTo func(key []byte, txNum uint64, val []byte)) error { +func (ht *HistoryRoTx) HistoryDump(fromTxNum, toTxNum int, keyToDump *[]byte, dumpTo func(key []byte, txNum uint64, val []byte) error) error { if len(ht.iit.files) == 0 { return nil } @@ -1486,7 +1486,9 @@ func (ht *HistoryRoTx) HistoryDump(fromTxNum, toTxNum int, keyToDump *[]byte, du val, pageBuf = seg.GetFromPage(histKeyBuf, val, pageBuf, true) } - dumpTo(key, txNum, val) + if err := dumpTo(key, txNum, val); err != nil { + return err + } } } }