Skip to content
13 changes: 11 additions & 2 deletions db/integrity/commitment_integrity.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"encoding/hex"
"errors"
"fmt"
"os"
"path/filepath"
"runtime"
"sort"
Expand Down Expand Up @@ -214,7 +215,11 @@ func checkCommitmentRootViaSd(ctx context.Context, tx kv.TemporalTx, f state.Vis
if err != nil {
return nil, err
}
sd.GetCommitmentCtx().SetTrace(logger.Enabled(ctx, log.LvlTrace))
if logger.Enabled(ctx, log.LvlTrace) {
sd.GetCommitmentCtx().SetTraceWriter(os.Stderr)
} else {
sd.GetCommitmentCtx().SetTraceWriter(nil)
}
sd.GetCommitmentCtx().SetStateReader(commitmentdb.NewFilesOnlyStateReader(tx, maxTxNum))
latestTxNum, _, err := sd.SeekCommitment(ctx, tx) // seek commitment again to use the new state reader instead
if err != nil {
Expand Down Expand Up @@ -1038,7 +1043,11 @@ func checkCommitmentHistAtBlkWithIdx(ctx context.Context, tx kv.TemporalTx, sd *
// plain state data view: as of end of the block
splitStateReader := commitmentdb.NewSplitHistoryReader(tx, commitmentAsOf, toTxNum, true /* withHistory */)
sd.GetCommitmentCtx().SetStateReader(splitStateReader)
sd.GetCommitmentCtx().SetTrace(logger.Enabled(ctx, log.LvlTrace))
if logger.Enabled(ctx, log.LvlTrace) {
sd.GetCommitmentCtx().SetTraceWriter(os.Stderr)
} else {
sd.GetCommitmentCtx().SetTraceWriter(nil)
}
latestTxNum, latestBlockNum, err := sd.SeekCommitment(ctx, tx) // seek commitment again with new history state reader
if err != nil {
return err
Expand Down
23 changes: 3 additions & 20 deletions db/state/execctx/domain_shared.go
Original file line number Diff line number Diff line change
Expand Up @@ -101,10 +101,8 @@ type SharedDomains struct {

logger log.Logger

txNum uint64
currentStep kv.Step
trace bool //nolint
commitmentCapture bool
txNum uint64
currentStep kv.Step
// disableInlineTouchKey when true, DomainPut skips the TouchKey call.
// Used when the commitment calculator goroutine owns the Updates buffer
// and feeds touches via TouchPlainKeyDirect from the fan-out channel.
Expand Down Expand Up @@ -191,8 +189,7 @@ func NewSharedDomains(ctx context.Context, tx kv.TemporalTx, logger log.Logger,
trieCfg := o.trieCfg

sd := &SharedDomains{
logger: logger,
//trace: true,
logger: logger,
metrics: kvmetrics.DomainMetrics{Domains: map[kv.Domain]*kvmetrics.DomainIOMetrics{}},
stepSize: tx.Debug().StepSize(),
}
Expand Down Expand Up @@ -672,14 +669,6 @@ func (sd *SharedDomains) Unwind(txNumUnwindTo uint64, changeset *[kv.DomainLen][
}
}

func (sd *SharedDomains) Trace() bool {
return sd.trace
}

func (sd *SharedDomains) CommitmentCapture() bool {
return sd.commitmentCapture
}

func (sd *SharedDomains) GetMemBatch() kv.TemporalMemBatch { return sd.mem }
func (sd *SharedDomains) SetInMemHistoryReads(v bool) { sd.mem.SetInMemHistoryReads(v) }
func (sd *SharedDomains) InMemHistoryReads() bool { return sd.mem.InMemHistoryReads() }
Expand Down Expand Up @@ -836,12 +825,6 @@ func (sd *SharedDomains) InlineTouchKeyDisabled() bool {
return sd.disableInlineTouchKey
}

func (sd *SharedDomains) SetTrace(b, capture bool) []string {
sd.trace = b
sd.commitmentCapture = capture
return sd.sdCtx.GetCapture(true)
}

func (sd *SharedDomains) HasPrefix(domain kv.Domain, prefix []byte, roTx kv.Tx) ([]byte, []byte, bool, error) {
return sd.mem.HasPrefix(domain, prefix, roTx)
}
Expand Down
1 change: 0 additions & 1 deletion db/test/aggregator_ext_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -812,7 +812,6 @@ func generateSharedDomainsUpdates(t *testing.T, domains *execctx.SharedDomains,
usedKeys[k] = struct{}{}
}
if txNum%commitEvery == 0 {
// domains.SetTrace(true)
stateRootHash, err := domains.ComputeCommitment(t.Context(), tx, true, txNum/commitEvery, txNum, "", nil)
require.NoErrorf(t, err, "txNum=%d", txNum)
_ = stateRootHash
Expand Down
33 changes: 32 additions & 1 deletion db/test/domains_restart_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -480,7 +480,6 @@ func TestCommit(t *testing.T) {
require.NoError(t, err)
}

domains.SetTrace(false, false)
domainsHash, err := domains.ComputeCommitment(ctx, tx, true, blockNum, txNum, "", nil)
require.NoError(t, err)
err = domains.Flush(ctx, tx)
Expand All @@ -489,3 +488,35 @@ func TestCommit(t *testing.T) {
require.Equal(t, common.BytesToHash(common.FromHex("0xfe81cd91357cd915cae7c02b5a4771e903c16b29dec582818076954be3741030")), common.BytesToHash(domainsHash))

}

// The commitment context routes both the trie trace and the branch read/write
// trace through one io.Writer; branch writes surface as [SDC] lines.
func TestCommitmentContextTraceWriter(t *testing.T) {
ctx := t.Context()
db, _, _ := testDbAndAggregatorv3(t, t.TempDir(), uint64(100))
tx, err := db.BeginTemporalRw(ctx)
require.NoError(t, err)
defer tx.Rollback()

domains, err := execctx.NewSharedDomains(ctx, tx, log.New())
require.NoError(t, err)
defer domains.Close()

acc := accounts.Account{Balance: u256.U64(7), CodeHash: accounts.EmptyCodeHash, Incarnation: 1}
val := accounts.SerialiseV3(&acc)
addr := common.Hex2Bytes("8e5476fc5990638a4fb0b5fd3f61bb4b5c5f395e")
for i := 1; i < 12; i++ { // diverging first nibbles force a branch node
addr[0] = byte(i * 16)
require.NoError(t, domains.DomainPut(kv.AccountsDomain, tx, common.Copy(addr), val, 0, nil))
}

var trace strings.Builder
domains.GetCommitmentCtx().SetTraceWriter(&trace)
_, err = domains.ComputeCommitment(ctx, tx, true, 0, 0, "", nil)
require.NoError(t, err)
domains.GetCommitmentCtx().SetTraceWriter(nil)

out := trace.String()
require.Contains(t, out, "[SDC] PutBranch", "branch writes must trace through the writer")
require.Contains(t, out, "[proc]", "trie must trace processed keys through the same writer")
}
6 changes: 5 additions & 1 deletion execution/commitment/backtester/backtester.go
Original file line number Diff line number Diff line change
Expand Up @@ -217,7 +217,11 @@ func (bt Backtester) backtestBlock(ctx context.Context, tx kv.TemporalTx, block
// - commitment data as-of the beginning of the block
// - account/storage/code data as-of the end of the block
sd.GetCommitmentCtx().SetStateReader(commitmentdb.NewSplitHistoryReader(tx, fromTxNum, toTxNum /* withHistory */, false))
sd.GetCommitmentCtx().SetTrace(bt.logger.Enabled(ctx, log.LvlTrace))
if bt.logger.Enabled(ctx, log.LvlTrace) {
sd.GetCommitmentCtx().SetTraceWriter(os.Stderr)
} else {
sd.GetCommitmentCtx().SetTraceWriter(nil)
}
latestTxNum, latestBlockNum, err := sd.SeekCommitment(ctx, tx)
if err != nil {
return err
Expand Down
9 changes: 3 additions & 6 deletions execution/commitment/commitment.go
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ import (
"encoding/binary"
"errors"
"fmt"
"io"
"math/bits"
"slices"
"strings"
Expand Down Expand Up @@ -92,12 +93,8 @@ type Trie interface {
// RootHash produces root hash of the trie
RootHash() (hash []byte, err error)

// Makes trie more verbose
SetTrace(bool)
// Trace domain writes only (no filding etc)
SetTraceDomain(bool)
SetCapture(capture []string)
GetCapture(truncate bool) []string
// SetTraceWriter sets the trace writer; nil disables tracing
SetTraceWriter(io.Writer)
EnableCsvMetrics(filePathPrefix string)

// Variant returns commitment trie variant
Expand Down
2 changes: 1 addition & 1 deletion execution/commitment/commitment_bench_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ func BenchmarkHexPatriciaHashedFold(b *testing.B) {
ctx := context.Background()
ms := NewMockState(b)
hph := NewHexPatriciaHashed(1, ms, DefaultTrieConfig())
hph.SetTrace(false)
hph.SetTraceWriter(nil)

// Build a trie with accounts and storage to exercise all fold paths
plainKeys, updates := NewUpdateBuilder().
Expand Down
49 changes: 21 additions & 28 deletions execution/commitment/commitmentdb/commitment_context.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import (
"encoding/hex"
"errors"
"fmt"
"io"
"runtime"
"sync"
"sync/atomic"
Expand Down Expand Up @@ -43,9 +44,6 @@ type sd interface {
// read), tagged with source.
MergeMetrics(source kvmetrics.Source, wm *kvmetrics.DomainMetrics)
StepSize() uint64
Trace() bool
CommitmentCapture() bool

// ProbeReadLayers samples sd.mem, parent.mem and tx-direct (MDBX) for one
// key — BranchCache divergence-detection probe. Read-only.
ProbeReadLayers(domain kv.Domain, tx kv.TemporalTx, key []byte) (mem, parentMem, mdbx []byte, memOk, parentOk bool)
Expand All @@ -64,7 +62,7 @@ type SharedDomainsCommitmentContext struct {
patriciaTrie commitment.Trie
variant commitment.TrieVariant // selected trie engine, for the [commitment] log (updates.Mode() is ModeParallel for both parallel and streaming)
justRestored atomic.Bool // set to true when commitment trie was just restored from snapshot
trace bool
traceW io.Writer
stateReader StateReader
paraTrieDB kv.TemporalRoDB // DB used for para trie and/or parallel trie warmup
// warmupBase holds the construction-time portion of the per-call WarmupConfig.
Expand Down Expand Up @@ -170,14 +168,11 @@ func (sdc *SharedDomainsCommitmentContext) SetCustomHistoryStateReader(stateRead
sdc.SetStateReader(stateReader)
}

func (sdc *SharedDomainsCommitmentContext) SetTrace(trace bool) {
sdc.trace = trace
sdc.patriciaTrie.SetTrace(trace)
}

// SetCapture enables/disables trie operation capture for diagnosis.
func (sdc *SharedDomainsCommitmentContext) SetCapture(capture []string) {
sdc.patriciaTrie.SetCapture(capture)
func (sdc *SharedDomainsCommitmentContext) SetTraceWriter(w io.Writer) {
// Wrap once so the main and per-worker TrieContexts share one mutex-guarded
// writer: concurrent workers trace branch reads/writes without racing.
sdc.traceW = commitment.NewSyncWriter(w)
sdc.patriciaTrie.SetTraceWriter(sdc.traceW)
}

// GetUpdates returns the current updates buffer. Used by the commitment
Expand Down Expand Up @@ -225,6 +220,7 @@ func (sdc *SharedDomainsCommitmentContext) trieContext(tx kv.TemporalTx, blockNu
stepSize: sdc.sharedDomains.StepSize(),
txNum: txNum,
blockNum: blockNum,
traceW: sdc.traceW,
}
if sdc.stateReader != nil {
mainTtx.stateReader = sdc.stateReader.CloneForWorker(readCtx, tx)
Expand All @@ -246,10 +242,6 @@ func (sdc *SharedDomainsCommitmentContext) Reset() {
}
}

func (sdc *SharedDomainsCommitmentContext) GetCapture(truncate bool) []string {
return sdc.patriciaTrie.GetCapture(truncate)
}

func (sdc *SharedDomainsCommitmentContext) ClearRam() {
sdc.updates.Reset()
sdc.Reset()
Expand Down Expand Up @@ -421,21 +413,17 @@ func (sdc *SharedDomainsCommitmentContext) ComputeCommitment(ctx context.Context
}
log.Debug("[commitment] processed", "block", blockNum, "txNum", txNum, "keys", common.PrettyCounter(updateCount), "keys/s", common.PrettyCounter(keysPerSec), "mode", sdc.variant, "bufmode", sdc.updates.Mode(), "spent", took, "rootHash", hex.EncodeToString(rootHash))
}()
// Re-apply the trace writer before any trie operations (including the early-return
// RootHash below); GenerateWitness clears it, so it must be restored on each call.
sdc.patriciaTrie.SetTraceWriter(sdc.traceW)

if updateCount == 0 {
rootHash, err = sdc.patriciaTrie.RootHash()
return rootHash, err
}

// data accessing functions should be set when domain is opened/shared context updated

sdc.patriciaTrie.SetTrace(sdc.trace)
sdc.patriciaTrie.SetTraceDomain(sdc.sharedDomains.Trace())
if sdc.sharedDomains.CommitmentCapture() {
if sdc.patriciaTrie.GetCapture(false) == nil {
sdc.patriciaTrie.SetCapture([]string{})
}
}

// Per-ComputeCommitment metrics accumulator for the main (root-fold) reads.
// The fold is single-goroutine, so this lock-free accumulator is owned
// exclusively here; merged into the shared metrics when commitment finishes.
Expand Down Expand Up @@ -626,6 +614,7 @@ func (sdc *SharedDomainsCommitmentContext) trieContextFactory(ctx context.Contex
putter: sdc.sharedDomains.AsPutDel(roTx),
stepSize: stepSize,
txNum: txNum,
traceW: sdc.traceW,
}
if sdc.stateReader != nil {
warmupCtx.stateReader = sdc.stateReader.CloneForWorker(workerCtx, roTx)
Expand Down Expand Up @@ -672,6 +661,7 @@ func (sdc *SharedDomainsCommitmentContext) concurrentTrieContextFactory(ctx cont
stepSize: stepSize,
txNum: txNum,
localCollector: collector,
traceW: sdc.traceW,
}
if sdc.stateReader != nil {
warmupCtx.stateReader = sdc.stateReader.CloneForWorker(workerCtx, roTx)
Expand Down Expand Up @@ -887,7 +877,7 @@ func (sdc *SharedDomainsCommitmentContext) restorePatriciaState(value []byte) (u
return 0, 0, fmt.Errorf("failed restore state : %w", err)
}
sdc.justRestored.Store(true) // to prevent double reset
if sdc.trace {
if sdc.traceW != nil {
rootHash, err := hext.RootHash()
if err != nil {
return 0, 0, fmt.Errorf("failed to get root hash after state restore: %w", err)
Expand All @@ -904,7 +894,7 @@ type TrieContext struct {
blockNum uint64

stepSize uint64
trace bool
traceW io.Writer // nil = disabled; traces branch reads/writes (see [SDC] lines)
stateReader StateReader
localCollector *etl.Collector // per-goroutine collector for concurrent PutBranch
}
Expand All @@ -925,15 +915,18 @@ func (sdc *TrieContext) Branch(pref []byte) ([]byte, kv.Step, error) {
// underlying state cache / getter aliases shared storage that another goroutine
// (concurrent commitment workers) can recycle. Own the bytes at the trie-context
// boundary so all downstream consumers are safe.
if sdc.traceW != nil {
fmt.Fprintf(sdc.traceW, "[SDC] Branch read %x => %x\n", pref, enc)
}
return common.Copy(enc), step, nil
}

func (sdc *TrieContext) PutBranch(prefix []byte, data []byte, prevData []byte) error {
if sdc.stateReader.WithHistory() { // do not store branches if explicitly operate on history
return nil
}
if sdc.trace {
fmt.Printf("[SDC] PutBranch: %x: %x\n", prefix, data)
if sdc.traceW != nil {
fmt.Fprintf(sdc.traceW, "[SDC] PutBranch %x: %x\n", prefix, data)
}
if sdc.localCollector != nil {
return sdc.localCollector.Collect(prefix, data)
Expand Down
Loading
Loading