From 698d6330ae49c08d2051b42f3993054a841359ca Mon Sep 17 00:00:00 2001 From: danielntmd Date: Sun, 13 Sep 2026 22:23:48 -0700 Subject: [PATCH 1/2] perf(rpc): resume shared trace prefixes in RPC workflows --- rpc/shared_trace_test.go | 115 ++++++++++- rpc/v10/progressive_trace_test.go | 331 ++++++++++++++++++++++++++++++ rpc/v10/trace.go | 40 +++- rpc/v10/trace_test.go | 4 +- rpc/v8/trace.go | 69 ++++++- rpc/v9/trace.go | 45 +++- 6 files changed, 572 insertions(+), 32 deletions(-) create mode 100644 rpc/v10/progressive_trace_test.go diff --git a/rpc/shared_trace_test.go b/rpc/shared_trace_test.go index beee3fcd71..06cb790e78 100644 --- a/rpc/shared_trace_test.go +++ b/rpc/shared_trace_test.go @@ -58,11 +58,11 @@ func TestSharedTraceCacheInitialReads(t *testing.T) { (*felt.TransactionHash)(transactions[i].Hash()), ).Return(header.Number, uint64(i), nil).AnyTimes() } - reader.EXPECT().TransactionsByBlockNumber(header.Number).Return(transactions, nil) + reader.EXPECT().TransactionsByBlockNumber(header.Number).Return(transactions, nil).Times(2) reader.EXPECT().StateAtBlockHash(header.ParentHash). - Return(state, func() error { return nil }, nil).Times(2) - reader.EXPECT().HeadState().Return(state, func() error { return nil }, nil).Times(2) - // v8 executes the full block; other versions reuse it, then initial reads replay once. + Return(state, func() error { return nil }, nil).Times(3) + reader.EXPECT().HeadState().Return(state, func() error { return nil }, nil).Times(3) + // v8 and v9 extend the prefix; initial reads replay the complete block. calls := 0 runner.EXPECT().Trace( gomock.Any(), @@ -80,8 +80,12 @@ func TestSharedTraceCacheInitialReads(t *testing.T) { opts vm.TraceOptions, ) (vm.ExecutionResults, error) { calls++ - require.Equal(t, transactions, txs) - require.Equal(t, calls == 2, opts.ReturnInitialReads) + if calls <= 2 { + require.Equal(t, transactions[calls-1:calls], txs) + } else { + require.Equal(t, transactions, txs) + } + require.Equal(t, calls == 3, opts.ReturnInitialReads) result := vm.ExecutionResults{ Traces: make([]vm.TransactionTrace, len(txs)), GasConsumed: make([]core.GasConsumed, len(txs)), @@ -98,7 +102,7 @@ func TestSharedTraceCacheInitialReads(t *testing.T) { result.InitialReads = &vm.InitialReads{} } return result, nil - }).Times(2) + }).Times(3) h := rpc.New(reader, nil, runner, "test", log.NewNopZapLogger(), &networks.Mainnet) methods := registeredTraceMethods(t, h) first, steps, err := methods.transaction8(t.Context(), *transactions[0].Hash()) @@ -109,7 +113,7 @@ func TestSharedTraceCacheInitialReads(t *testing.T) { t.Context(), (*felt.TransactionHash)(transactions[1].Hash()), ) require.Nil(t, err) - require.Equal(t, "0", steps.Get(rpcv9.ExecutionStepsHeader)) + require.Equal(t, "11", steps.Get(rpcv9.ExecutionStepsHeader)) require.Equal(t, uint64(2), second.ExecutionResources.L1Gas) id10 := rpcv10.BlockIDFromHash(header.Hash) withReads, _, err := methods.block10( @@ -125,7 +129,7 @@ func TestSharedTraceCacheInitialReads(t *testing.T) { _, steps, err = methods.transaction8(t.Context(), *transactions[0].Hash()) require.Nil(t, err) require.Equal(t, "0", steps.Get(rpcv8.ExecutionStepsHeader)) - require.Equal(t, 2, calls) + require.Equal(t, 3, calls) } func TestSharedFeederTraceCachePreservesVersionShapes(t *testing.T) { @@ -472,6 +476,93 @@ func TestSharedTraceCacheEachProducer(t *testing.T) { } } +func TestProgressiveTraceEachVersionExtends(t *testing.T) { + for first := range 3 { + t.Run(fmt.Sprintf("first=%d", first), func(t *testing.T) { + ctrl := gomock.NewController(t) + reader := mocks.NewMockReader(ctrl) + runner := mocks.NewMockVM(ctrl) + state := mocks.NewMockStateReader(ctrl) + header := &core.Header{ + Hash: felt.NewFromUint64[felt.Felt](100), + ParentHash: felt.NewFromUint64[felt.Felt](99), + ProtocolVersion: "0.14.0", + } + txs := make([]core.Transaction, 3) + for i := range txs { + txs[i] = &core.InvokeTransaction{TransactionHash: felt.NewFromUint64[felt.Felt](uint64(i + 1))} + reader.EXPECT().Receipt(txs[i].Hash()).Return(nil, header.Hash, header.Number, nil).AnyTimes() + reader.EXPECT().BlockNumberAndIndexByTxHash( + (*felt.TransactionHash)(txs[i].Hash()), + ).Return(header.Number, uint64(i), nil).AnyTimes() + } + reader.EXPECT().BlockByHash(header.Hash). + Return(&core.Block{Header: header, Transactions: txs}, nil).AnyTimes() + reader.EXPECT().BlockHeaderByNumber(header.Number).Return(header, nil).AnyTimes() + reader.EXPECT().Network().Return(&networks.Mainnet).AnyTimes() + reader.EXPECT().TransactionsByBlockNumber(header.Number).Return(txs, nil).Times(2) + reader.EXPECT().StateAtBlockHash(header.ParentHash). + Return(state, func() error { return nil }, nil).Times(3) + reader.EXPECT().HeadState().Return(state, func() error { return nil }, nil).Times(3) + calls := 0 + runner.EXPECT().Trace( + gomock.Any(), + gomock.Any(), + gomock.Any(), + gomock.Any(), + gomock.Any(), + vm.TraceOptions{}, + ).DoAndReturn(func( + actual []core.Transaction, + _ []core.ClassDefinition, + _ []*felt.Felt, + _ *vm.BlockInfo, + _ core.StateReader, + _ vm.TraceOptions, + ) (vm.ExecutionResults, error) { + require.Equal(t, txs[calls:calls+1], actual) + calls++ + return vm.ExecutionResults{ + Traces: []vm.TransactionTrace{{Type: vm.TxnInvoke, StateDiff: &vm.StateDiff{}}}, + GasConsumed: []core.GasConsumed{{}}, + NumSteps: 1, + }, nil + }).Times(3) + h := rpc.New(reader, nil, runner, "test", log.NewNopZapLogger(), &networks.Mainnet) + methods := registeredTraceMethods(t, h) + request := func(version, index int) { + var steps http.Header + var rpcErr *jsonrpc.Error + switch version { + case 0: + _, steps, rpcErr = methods.transaction8(t.Context(), *txs[index].Hash()) + case 1: + _, steps, rpcErr = methods.transaction9( + t.Context(), (*felt.TransactionHash)(txs[index].Hash()), + ) + case 2: + _, steps, rpcErr = methods.transaction10( + t.Context(), (*felt.TransactionHash)(txs[index].Hash()), + ) + } + require.Nil(t, rpcErr) + expected := "1" + if index == 0 && calls == 3 { + expected = "0" + } + require.Equal(t, expected, steps.Get(rpcv10.ExecutionStepsHeader)) + } + for i := range 3 { + request((first+i)%3, i) + } + for version := range 3 { + request(version, 0) + } + require.Equal(t, 3, calls) + }) + } +} + type traceMethods struct { block8 func(context.Context, *rpcv8.BlockID) ( []rpcv8.TracedBlockTransaction, @@ -498,6 +589,11 @@ type traceMethods struct { http.Header, *jsonrpc.Error, ) + transaction10 func(context.Context, *felt.TransactionHash) ( + rpcv10.TransactionTrace, + http.Header, + *jsonrpc.Error, + ) } func registeredTraceMethods(t *testing.T, h *rpc.Handler) traceMethods { @@ -511,6 +607,7 @@ func registeredTraceMethods(t *testing.T, h *rpc.Handler) traceMethods { bindTraceMethod(t, v10, "starknet_traceBlockTransactions", &methods.block10) bindTraceMethod(t, v8, "starknet_traceTransaction", &methods.transaction8) bindTraceMethod(t, v9, "starknet_traceTransaction", &methods.transaction9) + bindTraceMethod(t, v10, "starknet_traceTransaction", &methods.transaction10) return methods } diff --git a/rpc/v10/progressive_trace_test.go b/rpc/v10/progressive_trace_test.go new file mode 100644 index 0000000000..2cdc8a9ddf --- /dev/null +++ b/rpc/v10/progressive_trace_test.go @@ -0,0 +1,331 @@ +package rpcv10 + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync/atomic" + "testing" + + "github.com/NethermindEth/juno/blockchain/networks" + "github.com/NethermindEth/juno/core" + "github.com/NethermindEth/juno/core/felt" + "github.com/NethermindEth/juno/jsonrpc" + "github.com/NethermindEth/juno/mocks" + "github.com/NethermindEth/juno/rpc/rpccore" + "github.com/NethermindEth/juno/rpc/tracecache" + "github.com/NethermindEth/juno/utils/log" + "github.com/NethermindEth/juno/vm" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" +) + +type progressiveVM struct { + vm.VM + run func([]core.Transaction, core.StateReader, vm.TraceOptions) (vm.ExecutionResults, error) +} + +func (v *progressiveVM) Trace( + txs []core.Transaction, + _ []core.ClassDefinition, + _ []*felt.Felt, + _ *vm.BlockInfo, + state core.StateReader, + opts vm.TraceOptions, +) (vm.ExecutionResults, error) { + return v.run(txs, state, opts) +} + +func progressiveResult(txs []core.Transaction) vm.ExecutionResults { + result := vm.ExecutionResults{ + Traces: make([]vm.TransactionTrace, len(txs)), + GasConsumed: make([]core.GasConsumed, len(txs)), + NumSteps: uint64(len(txs)), + } + for i, tx := range txs { + result.Traces[i] = vm.TransactionTrace{Type: vm.TxnInvoke, StateDiff: &vm.StateDiff{}} + result.GasConsumed[i].L1Gas = tx.Hash().Uint64() + } + return result +} + +func progressiveHandler( + t *testing.T, runner vm.VM, +) (*Handler, *core.Header, []core.Transaction, *mocks.MockStateReader) { + t.Helper() + ctrl := gomock.NewController(t) + reader := mocks.NewMockReader(ctrl) + parent := mocks.NewMockStateReader(ctrl) + head := mocks.NewMockStateReader(ctrl) + header := &core.Header{ + Hash: felt.NewFromUint64[felt.Felt](100), + ParentHash: felt.NewFromUint64[felt.Felt](99), + ProtocolVersion: "0.14.0", + } + txs := make([]core.Transaction, 3) + for i := range txs { + txs[i] = &core.InvokeTransaction{TransactionHash: felt.NewFromUint64[felt.Felt](uint64(i + 1))} + } + reader.EXPECT().Network().Return(&networks.Mainnet).AnyTimes() + reader.EXPECT().TransactionsByBlockNumber(header.Number).Return(txs, nil).AnyTimes() + reader.EXPECT().StateAtBlockHash(header.ParentHash). + Return(parent, func() error { return nil }, nil).AnyTimes() + reader.EXPECT().HeadState().Return(head, func() error { return nil }, nil).AnyTimes() + return New(reader, nil, runner, log.NewNopZapLogger()), header, txs, head +} + +func TestProgressiveFailureRetainsPrefixAndOffsetsError(t *testing.T) { + var calls atomic.Uint64 + runner := &progressiveVM{run: func( + txs []core.Transaction, _ core.StateReader, _ vm.TraceOptions, + ) (vm.ExecutionResults, error) { + if calls.Add(1) == 2 { + return vm.ExecutionResults{NumSteps: 9}, fmt.Errorf( + "wrapped: %w", vm.TransactionExecutionError{Index: 0, Cause: json.RawMessage(`"failed"`)}, + ) + } + require.Len(t, txs, 1) + return progressiveResult(txs), nil + }} + h, header, txs, _ := progressiveHandler(t, runner) + first := &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()} + second := &tracecache.TransactionTarget{Index: 1, Hash: txs[1].Hash()} + prefix, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, first) + require.Nil(t, rpcErr) + _, steps, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, second) + require.NotNil(t, rpcErr) + require.Contains(t, rpcErr.Data, "transaction #1") + require.Equal(t, "9", steps.Get(ExecutionStepsHeader)) + hit, steps, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, first) + require.Nil(t, rpcErr) + require.Equal(t, prefix, hit) + require.Equal(t, "0", steps.Get(ExecutionStepsHeader)) + next, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, second) + require.Nil(t, rpcErr) + require.Len(t, next.Traces, 2) + require.Equal(t, uint64(3), calls.Load()) +} + +func TestProgressiveWaiterCancellation(t *testing.T) { + entered, release := make(chan struct{}), make(chan struct{}) + var calls atomic.Uint64 + runner := &progressiveVM{run: func( + txs []core.Transaction, _ core.StateReader, _ vm.TraceOptions, + ) (vm.ExecutionResults, error) { + calls.Add(1) + close(entered) + <-release + return progressiveResult(txs), nil + }} + h, header, txs, _ := progressiveHandler(t, runner) + target := &tracecache.TransactionTarget{Index: 1, Hash: txs[1].Hash()} + done := make(chan *jsonrpc.Error, 1) + go func() { + _, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + done <- rpcErr + }() + <-entered + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, _, rpcErr := h.traceFinalisedBlock(ctx, header, false, target) + require.NotNil(t, rpcErr) + require.Contains(t, rpcErr.Data, context.Canceled.Error()) + close(release) + require.Nil(t, <-done) + result, steps, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + require.Nil(t, rpcErr) + require.Len(t, result.Traces, 2) + require.Equal(t, "0", steps.Get(ExecutionStepsHeader)) + require.Equal(t, uint64(1), calls.Load()) +} + +func TestProgressivePanicAndMalformedResultsDoNotPublish(t *testing.T) { + for _, failure := range []string{"panic", "traces", "gas", "state diff"} { + t.Run(failure, func(t *testing.T) { + calls := 0 + runner := &progressiveVM{run: func( + txs []core.Transaction, _ core.StateReader, _ vm.TraceOptions, + ) (vm.ExecutionResults, error) { + calls++ + result := progressiveResult(txs) + if calls == 1 { + switch failure { + case "panic": + panic("execution panic") + case "traces": + result.Traces = nil + case "gas": + result.GasConsumed = nil + case "state diff": + result.Traces[0].StateDiff = nil + } + } + return result, nil + }} + h, header, txs, _ := progressiveHandler(t, runner) + target := &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()} + request := func() { + _, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + require.NotNil(t, rpcErr) + } + if failure == "panic" { + require.PanicsWithValue(t, "execution panic", request) + } else { + request() + } + result, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + require.Nil(t, rpcErr) + require.Len(t, result.Traces, 1) + require.Equal(t, 2, calls) + }) + } +} + +func TestProgressiveInitialReadsReplayPreservesPrefix(t *testing.T) { + entered, release := make(chan struct{}), make(chan struct{}) + var calls atomic.Uint64 + runner := &progressiveVM{run: func( + txs []core.Transaction, _ core.StateReader, opts vm.TraceOptions, + ) (vm.ExecutionResults, error) { + call := calls.Add(1) + result := progressiveResult(txs) + if call == 1 { + require.Len(t, txs, 1) + require.False(t, opts.ReturnInitialReads) + } else { + require.Len(t, txs, 3) + require.True(t, opts.ReturnInitialReads) + if call == 2 { + close(entered) + <-release + } else { + result.InitialReads = &vm.InitialReads{} + } + } + return result, nil + }} + h, header, txs, _ := progressiveHandler(t, runner) + target := &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()} + prefix, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + require.Nil(t, rpcErr) + done := make(chan *jsonrpc.Error, 1) + go func() { + _, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, true, nil) + done <- rpcErr + }() + <-entered + hit, steps, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + require.Nil(t, rpcErr) + require.Equal(t, prefix, hit) + require.Equal(t, "0", steps.Get(ExecutionStepsHeader)) + close(release) + rpcErr = <-done + require.NotNil(t, rpcErr) + require.Contains(t, rpcErr.Data, "VM omitted initial reads") + hit, _, rpcErr = h.traceFinalisedBlock(t.Context(), header, false, target) + require.Nil(t, rpcErr) + require.Equal(t, prefix, hit) + full, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, true, nil) + require.Nil(t, rpcErr) + require.Len(t, full.Traces, 3) + require.NotNil(t, full.InitialReads) + _, steps, rpcErr = h.traceFinalisedBlock(t.Context(), header, true, nil) + require.Nil(t, rpcErr) + require.Equal(t, "0", steps.Get(ExecutionStepsHeader)) + require.Equal(t, uint64(3), calls.Load()) +} + +func TestProgressiveCheckpointDeclarationsAndStorage(t *testing.T) { + classHash := felt.FromUint64[felt.Felt](44) + address := felt.FromUint64[felt.Felt](45) + key := felt.FromUint64[felt.Felt](46) + definition := &core.DeprecatedCairoClass{} + calls := 0 + runner := &progressiveVM{run: func( + txs []core.Transaction, state core.StateReader, _ vm.TraceOptions, + ) (vm.ExecutionResults, error) { + calls++ + result := progressiveResult(txs) + if calls == 1 { + result.Traces[0].StateDiff.DeprecatedDeclaredClasses = []*felt.Felt{&classHash} + result.Traces[0].StateDiff.StorageDiffs = []vm.StorageDiff{{ + Address: address, + StorageEntries: []vm.Entry{{Key: key, Value: felt.One}}, + }} + } else { + declared, err := state.Class(&classHash) + require.NoError(t, err) + require.Same(t, definition, declared.Class) + value, err := state.ContractStorage(&address, &key) + require.NoError(t, err) + require.Equal(t, felt.One, value) + } + return result, nil + }} + h, header, txs, head := progressiveHandler(t, runner) + head.EXPECT().Class(&classHash).Return(&core.DeclaredClassDefinition{Class: definition}, nil) + _, _, rpcErr := h.traceFinalisedBlock( + t.Context(), header, false, &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()}, + ) + require.Nil(t, rpcErr) + _, _, rpcErr = h.traceFinalisedBlock( + t.Context(), header, false, &tracecache.TransactionTarget{Index: 1, Hash: txs[1].Hash()}, + ) + require.Nil(t, rpcErr) + require.Equal(t, 2, calls) +} + +func TestProgressiveTargetIdentityBeforeExecutionAndOnHit(t *testing.T) { + calls := 0 + runner := &progressiveVM{run: func( + txs []core.Transaction, _ core.StateReader, _ vm.TraceOptions, + ) (vm.ExecutionResults, error) { + calls++ + return progressiveResult(txs), nil + }} + h, header, txs, _ := progressiveHandler(t, runner) + bad := &tracecache.TransactionTarget{Index: 0, Hash: felt.NewFromUint64[felt.Felt](999)} + _, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, bad) + require.Equal(t, rpccore.ErrTxnHashNotFound, rpcErr) + require.Zero(t, calls) + _, _, rpcErr = h.traceFinalisedBlock( + t.Context(), header, false, &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()}, + ) + require.Nil(t, rpcErr) + _, _, rpcErr = h.traceFinalisedBlock(t.Context(), header, false, bad) + require.Equal(t, rpccore.ErrTxnHashNotFound, rpcErr) + require.Equal(t, 1, calls) + _, _, rpcErr = h.traceFinalisedBlock( + t.Context(), header, false, &tracecache.TransactionTarget{Index: 3, Hash: txs[0].Hash()}, + ) + require.Equal(t, rpccore.ErrTxnHashNotFound, rpcErr) + require.Equal(t, 1, calls) +} + +func TestProgressiveCheckpointReadFailurePreservesPrefix(t *testing.T) { + classHash := felt.FromUint64[felt.Felt](44) + calls := 0 + runner := &progressiveVM{run: func( + txs []core.Transaction, _ core.StateReader, _ vm.TraceOptions, + ) (vm.ExecutionResults, error) { + calls++ + result := progressiveResult(txs) + result.Traces[0].StateDiff.DeprecatedDeclaredClasses = []*felt.Felt{&classHash} + return result, nil + }} + h, header, txs, head := progressiveHandler(t, runner) + head.EXPECT().Class(&classHash).Return(nil, errors.New("class read failed")) + target := &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()} + _, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + require.Nil(t, rpcErr) + _, _, rpcErr = h.traceFinalisedBlock( + t.Context(), header, false, &tracecache.TransactionTarget{Index: 1, Hash: txs[1].Hash()}, + ) + require.NotNil(t, rpcErr) + require.Equal(t, jsonrpc.InternalError, rpcErr.Code) + _, steps, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + require.Nil(t, rpcErr) + require.Equal(t, "0", steps.Get(ExecutionStepsHeader)) + require.Equal(t, 1, calls) +} diff --git a/rpc/v10/trace.go b/rpc/v10/trace.go index 3e155dbd00..fa47d67a34 100644 --- a/rpc/v10/trace.go +++ b/rpc/v10/trace.go @@ -144,7 +144,7 @@ func (h *Handler) TraceBlockTransactions( } returnInitialReads := slices.Contains(traceFlags, TraceReturnInitialReadsFlag) - traces, httpHeader, rpcErr := h.traceFinalisedBlock(ctx, header, returnInitialReads) + traces, httpHeader, rpcErr := h.traceFinalisedBlock(ctx, header, returnInitialReads, nil) if rpcErr != nil { return TraceBlockTransactionsResponse{}, httpHeader, rpcErr } @@ -178,6 +178,7 @@ func traceTransactionsWithState( classLookupState core.StateReader, blockInfo *vm.BlockInfo, returnInitialReads bool, + errorIndexOffset uint64, ) (*tracecache.BlockTrace, http.Header, *jsonrpc.Error) { httpHeader := defaultExecutionHeader() @@ -198,6 +199,7 @@ func traceTransactionsWithState( vm.TraceOptions{ReturnInitialReads: returnInitialReads}, ) + vmErr = tracecache.OffsetExecutionError(vmErr, errorIndexOffset) httpHeader.Set(ExecutionStepsHeader, strconv.FormatUint(executionResult.NumSteps, 10)) if vmErr != nil { @@ -263,7 +265,9 @@ func (h *Handler) findAndTraceFinalisedTransaction( return TransactionTrace{}, nil, rpccore.ErrInternal.CloneWithData(err) } - blockTraces, httpHeader, rpcErr := h.traceFinalisedBlock(ctx, header, false) + blockTraces, httpHeader, rpcErr := h.traceFinalisedBlock( + ctx, header, false, &tracecache.TransactionTarget{Index: txIndex, Hash: (*felt.Felt)(hash)}, + ) if rpcErr != nil { return TransactionTrace{}, nil, rpcErr } @@ -325,6 +329,7 @@ func (h *Handler) findAndTraceInPreConfirmed( state, // class lookup state (same for preconfirmed) &blockInfo, false, // returnInitialReads + 0, ) if rpcErr != nil { return TransactionTrace{}, httpHeader, rpcErr @@ -346,18 +351,22 @@ func (h *Handler) traceFinalisedBlock( ctx context.Context, header *core.Header, returnInitialReads bool, + target *tracecache.TransactionTarget, ) (*tracecache.BlockTrace, http.Header, *jsonrpc.Error) { cached, lease, err := h.blockTraceCache.AcquireWithCondition( ctx, header.Hash, func(b *tracecache.BlockTrace) bool { - return b.Covers(returnInitialReads) + return b.CoversTarget(target, returnInitialReads) }, ) if err != nil { return nil, defaultExecutionHeader(), rpccore.ErrUnexpectedError.CloneWithData(err.Error()) } if lease == nil { + if cached.ValidateTarget(target) != nil { + return nil, defaultExecutionHeader(), rpccore.ErrTxnHashNotFound + } return cached, defaultExecutionHeader(), nil } defer lease.Release() @@ -375,6 +384,9 @@ func (h *Handler) traceFinalisedBlock( return nil, defaultExecutionHeader(), rpcErr } + if traces.ValidateTarget(target) != nil { + return nil, defaultExecutionHeader(), rpccore.ErrTxnHashNotFound + } lease.Publish(traces) return traces, defaultExecutionHeader(), nil } @@ -390,11 +402,22 @@ func (h *Handler) traceFinalisedBlock( rpccore.ErrInternal.CloneWithData(err) } - response, httpHeader, rpcErr := h.traceBlockWithVM(header, transactions, returnInitialReads) + plan, planErr := tracecache.PlanRange(cached, transactions, target, returnInitialReads) + if planErr != nil { + if errors.Is(planErr, tracecache.ErrTargetNotFound) { + return nil, defaultExecutionHeader(), rpccore.ErrTxnHashNotFound + } + return nil, defaultExecutionHeader(), rpccore.ErrUnexpectedError.CloneWithData(planErr.Error()) + } + response, httpHeader, rpcErr := h.traceBlockWithVM(header, transactions, returnInitialReads, plan) if rpcErr != nil { return nil, httpHeader, rpcErr } - response.Complete = true + combined, combineErr := plan.Combine(response) + if combineErr != nil { + return nil, httpHeader, rpccore.ErrUnexpectedError.CloneWithData(combineErr.Error()) + } + response = combined lease.Publish(response) return response, httpHeader, nil @@ -405,7 +428,9 @@ func (h *Handler) traceBlockWithVM( header *core.Header, transactions []core.Transaction, returnInitialReads bool, + plan *tracecache.Range, ) (*tracecache.BlockTrace, http.Header, *jsonrpc.Error) { + transactions = transactions[plan.Start:plan.End] // Prepare execution state state, closer, err := h.bcReader.StateAtBlockHash(header.ParentHash) if err != nil { @@ -433,6 +458,10 @@ func (h *Handler) traceBlockWithVM( } defer h.callAndLogErr(headStateCloser, "Failed to close head state in traceBlockTransactions") + state, err = plan.ResumeState(state, headState, header.Number) + if err != nil { + return nil, defaultExecutionHeader(), jsonrpc.Err(jsonrpc.InternalError, err.Error()) + } // Create block info blockInfo, rpcErr := h.buildBlockInfo(header) if rpcErr != nil { @@ -446,6 +475,7 @@ func (h *Handler) traceBlockWithVM( headState, &blockInfo, returnInitialReads, + plan.Start, ) } diff --git a/rpc/v10/trace_test.go b/rpc/v10/trace_test.go index 59f7ba44bc..06d9fca29a 100644 --- a/rpc/v10/trace_test.go +++ b/rpc/v10/trace_test.go @@ -1410,7 +1410,7 @@ func TestTraceBlockTransactionsWithReturnInitialReads(t *testing.T) { OverallFees: []*felt.Felt{&felt.Zero}, DataAvailability: []core.DataAvailability{{L1Gas: 0}}, GasConsumed: []core.GasConsumed{{L1Gas: 0, L1DataGas: 0, L2Gas: 0}}, - Traces: []vm.TransactionTrace{{}}, + Traces: []vm.TransactionTrace{{StateDiff: &vm.StateDiff{}}}, NumSteps: 100, InitialReads: test.initialReads, }, nil) @@ -1496,7 +1496,7 @@ func TestTraceBlockTransactionsInitialReadsCacheCoherence(t *testing.T) { OverallFees: []*felt.Felt{&felt.Zero}, DataAvailability: []core.DataAvailability{{L1Gas: 0}}, GasConsumed: []core.GasConsumed{{L1Gas: 0, L1DataGas: 0, L2Gas: 0}}, - Traces: []vm.TransactionTrace{{}}, + Traces: []vm.TransactionTrace{{StateDiff: &vm.StateDiff{}}}, NumSteps: 100, InitialReads: reads, } diff --git a/rpc/v8/trace.go b/rpc/v8/trace.go index 312c00e51b..6b9c19c72a 100644 --- a/rpc/v8/trace.go +++ b/rpc/v8/trace.go @@ -122,7 +122,11 @@ func (h *Handler) TraceTransaction(ctx context.Context, hash felt.Felt) (*Transa return nil, httpHeader, rpccore.ErrTxnHashNotFound } - traceResults, header, traceBlockErr := h.traceBlockTransactions(ctx, block) + var target *tracecache.TransactionTarget + if !isPendingBlock { + target = &tracecache.TransactionTarget{Index: uint64(txIndex), Hash: &hash} + } + traceResults, header, traceBlockErr := h.traceBlockTransactions(ctx, block, target) if traceBlockErr != nil { return nil, header, traceBlockErr } @@ -140,7 +144,7 @@ func (h *Handler) TraceBlockTransactions( return nil, defaultExecutionHeader(), rpcErr } - traces, httpHeader, rpcErr := h.traceBlockTransactions(ctx, block) + traces, httpHeader, rpcErr := h.traceBlockTransactions(ctx, block, nil) if rpcErr != nil { return nil, httpHeader, rpcErr } @@ -148,17 +152,31 @@ func (h *Handler) TraceBlockTransactions( } // traceBlockTransactions caches local or feeder traces; pending blocks bypass the cache. +// +//nolint:gocyclo // Preserve v8 pending/finalized orchestration and error ordering in one place. func (h *Handler) traceBlockTransactions( - ctx context.Context, block *core.Block, + ctx context.Context, block *core.Block, target *tracecache.TransactionTarget, ) (*tracecache.BlockTrace, http.Header, *jsonrpc.Error) { isPending := block.Hash == nil var lease *tracecache.Lease[felt.Felt, *tracecache.BlockTrace] + var cached *tracecache.BlockTrace + var plan *tracecache.Range if !isPending { - cached, acquiredLease, err := h.blockTraceCache.Acquire(ctx, block.Hash) + previous, acquiredLease, err := h.blockTraceCache.AcquireWithCondition( + ctx, + block.Hash, + func(b *tracecache.BlockTrace) bool { + return b.CoversTarget(target, false) + }, + ) + cached = previous if err != nil { return nil, defaultExecutionHeader(), rpccore.ErrUnexpectedError.CloneWithData(err.Error()) } if acquiredLease == nil { + if cached.ValidateTarget(target) != nil { + return nil, defaultExecutionHeader(), rpccore.ErrTxnHashNotFound + } return cached, defaultExecutionHeader(), nil } lease = acquiredLease @@ -186,26 +204,47 @@ func (h *Handler) traceBlockTransactions( if err != nil { return nil, defaultExecutionHeader(), err } + if traces.ValidateTarget(target) != nil { + return nil, defaultExecutionHeader(), rpccore.ErrTxnHashNotFound + } lease.Publish(traces) return traces, defaultExecutionHeader(), nil } } - traces, httpHeader, rpcErr := h.traceBlockTransactionWithVM(block) + if !isPending { + var planErr error + plan, planErr = tracecache.PlanRange(cached, block.Transactions, target, false) + if planErr != nil { + if errors.Is(planErr, tracecache.ErrTargetNotFound) { + return nil, defaultExecutionHeader(), rpccore.ErrTxnHashNotFound + } + return nil, defaultExecutionHeader(), rpccore.ErrUnexpectedError.CloneWithData(planErr.Error()) + } + } + traces, httpHeader, rpcErr := h.traceBlockTransactionWithVM(block, plan) if rpcErr != nil { return nil, httpHeader, rpcErr } if lease != nil { - traces.Complete = true + combined, combineErr := plan.Combine(traces) + if combineErr != nil { + return nil, httpHeader, rpccore.ErrUnexpectedError.CloneWithData(combineErr.Error()) + } + traces = combined lease.Publish(traces) } return traces, httpHeader, nil } -func (h *Handler) traceBlockTransactionWithVM(block *core.Block) ( +func (h *Handler) traceBlockTransactionWithVM(block *core.Block, plan *tracecache.Range) ( *tracecache.BlockTrace, http.Header, *jsonrpc.Error, ) { httpHeader := defaultExecutionHeader() + transactions := block.Transactions + if plan != nil { + transactions = transactions[plan.Start:plan.End] + } state, closer, err := h.bcReader.StateAtBlockHash(block.ParentHash) if err != nil { return nil, httpHeader, rpccore.ErrBlockNotFound @@ -223,10 +262,17 @@ func (h *Handler) traceBlockTransactionWithVM(block *core.Block) ( } defer h.callAndLogErr(headStateCloser, "Failed to close head state in traceBlockTransactions") + if plan != nil { + state, err = plan.ResumeState(state, headState, block.Number) + if err != nil { + return nil, httpHeader, jsonrpc.Err(jsonrpc.InternalError, err.Error()) + } + } + var classes []core.ClassDefinition paidFeesOnL1 := []*felt.Felt{} - for _, transaction := range block.Transactions { + for _, transaction := range transactions { switch tx := transaction.(type) { case *core.DeclareTransaction: class, stateErr := headState.Class(tx.ClassHash) @@ -251,9 +297,12 @@ func (h *Handler) traceBlockTransactionWithVM(block *core.Block) ( BlockHashToBeRevealed: blockHashToBeRevealed, } - executionResult, err := h.vm.Trace(block.Transactions, classes, paidFeesOnL1, + executionResult, err := h.vm.Trace(transactions, classes, paidFeesOnL1, &blockInfo, state, vm.TraceOptions{}) + if plan != nil { + err = tracecache.OffsetExecutionError(err, plan.Start) + } httpHeader.Set(ExecutionStepsHeader, strconv.FormatUint(executionResult.NumSteps, 10)) if err != nil { @@ -265,7 +314,7 @@ func (h *Handler) traceBlockTransactionWithVM(block *core.Block) ( return nil, httpHeader, rpccore.ErrUnexpectedError.CloneWithData(err.Error()) } - result, packErr := tracecache.FromVM(block.Transactions, &executionResult, false) + result, packErr := tracecache.FromVM(transactions, &executionResult, false) if packErr != nil { return nil, httpHeader, rpccore.ErrUnexpectedError.CloneWithData(packErr.Error()) } diff --git a/rpc/v9/trace.go b/rpc/v9/trace.go index a9c0ee973b..6a7e90dd0b 100644 --- a/rpc/v9/trace.go +++ b/rpc/v9/trace.go @@ -123,7 +123,7 @@ func (h *Handler) TraceBlockTransactions( return nil, defaultExecutionHeader(), rpcErr } - traces, httpHeader, rpcErr := h.traceFinalisedBlock(ctx, header) + traces, httpHeader, rpcErr := h.traceFinalisedBlock(ctx, header, nil) if rpcErr != nil { return nil, httpHeader, rpcErr } @@ -214,6 +214,7 @@ func traceTransactionsWithState( executionState core.StateReader, classLookupState core.StateReader, blockInfo *vm.BlockInfo, + errorIndexOffset uint64, ) (*tracecache.BlockTrace, http.Header, *jsonrpc.Error) { httpHeader := defaultExecutionHeader() @@ -234,6 +235,7 @@ func traceTransactionsWithState( vm.TraceOptions{}, ) + vmErr = tracecache.OffsetExecutionError(vmErr, errorIndexOffset) httpHeader.Set(ExecutionStepsHeader, strconv.FormatUint(executionResult.NumSteps, 10)) if vmErr != nil { @@ -299,7 +301,9 @@ func (h *Handler) findAndTraceFinalisedTransaction( return TransactionTrace{}, nil, rpccore.ErrInternal.CloneWithData(err) } - blockTraces, httpHeader, rpcErr := h.traceFinalisedBlock(ctx, header) + blockTraces, httpHeader, rpcErr := h.traceFinalisedBlock( + ctx, header, &tracecache.TransactionTarget{Index: txIndex, Hash: (*felt.Felt)(hash)}, + ) if rpcErr != nil { return TransactionTrace{}, nil, rpcErr } @@ -360,6 +364,7 @@ func (h *Handler) findAndTraceInPreConfirmed( state, // execution state state, // class lookup state (same for preconfirmed) &blockInfo, + 0, ) if rpcErr != nil { return TransactionTrace{}, httpHeader, rpcErr @@ -378,13 +383,20 @@ func (h *Handler) findAndTraceInPreConfirmed( // traceFinalisedBlock caches local or feeder traces by block hash. // See shouldFetchTracesFromFeederGateway for feeder trace edge cases. func (h *Handler) traceFinalisedBlock( - ctx context.Context, header *core.Header, + ctx context.Context, header *core.Header, target *tracecache.TransactionTarget, ) (*tracecache.BlockTrace, http.Header, *jsonrpc.Error) { - cached, lease, err := h.blockTraceCache.Acquire(ctx, header.Hash) + cached, lease, err := h.blockTraceCache.AcquireWithCondition( + ctx, header.Hash, func(b *tracecache.BlockTrace) bool { + return b.CoversTarget(target, false) + }, + ) if err != nil { return nil, defaultExecutionHeader(), rpccore.ErrUnexpectedError.CloneWithData(err.Error()) } if lease == nil { + if cached.ValidateTarget(target) != nil { + return nil, defaultExecutionHeader(), rpccore.ErrTxnHashNotFound + } return cached, defaultExecutionHeader(), nil } defer lease.Release() @@ -409,15 +421,29 @@ func (h *Handler) traceFinalisedBlock( } return nil, httpHeader, rpccore.ErrInternal.CloneWithData(txErr) } - traces, httpHeader, rpcErr = h.traceBlockWithVM(header, transactions) + plan, planErr := tracecache.PlanRange(cached, transactions, target, false) + if planErr != nil { + if errors.Is(planErr, tracecache.ErrTargetNotFound) { + return nil, defaultExecutionHeader(), rpccore.ErrTxnHashNotFound + } + return nil, defaultExecutionHeader(), rpccore.ErrUnexpectedError.CloneWithData(planErr.Error()) + } + traces, httpHeader, rpcErr = h.traceBlockWithVM(header, transactions, plan) if rpcErr == nil { - traces.Complete = true + combined, combineErr := plan.Combine(traces) + if combineErr != nil { + return nil, httpHeader, rpccore.ErrUnexpectedError.CloneWithData(combineErr.Error()) + } + traces = combined } } if rpcErr != nil { return nil, httpHeader, rpcErr } + if traces.ValidateTarget(target) != nil { + return nil, httpHeader, rpccore.ErrTxnHashNotFound + } lease.Publish(traces) return traces, httpHeader, nil @@ -427,7 +453,9 @@ func (h *Handler) traceFinalisedBlock( func (h *Handler) traceBlockWithVM( header *core.Header, transactions []core.Transaction, + plan *tracecache.Range, ) (*tracecache.BlockTrace, http.Header, *jsonrpc.Error) { + transactions = transactions[plan.Start:plan.End] // Prepare execution state state, closer, err := h.bcReader.StateAtBlockHash(header.ParentHash) if err != nil { @@ -450,6 +478,10 @@ func (h *Handler) traceBlockWithVM( } defer h.callAndLogErr(headStateCloser, "Failed to close head state in traceBlockTransactions") + state, err = plan.ResumeState(state, headState, header.Number) + if err != nil { + return nil, defaultExecutionHeader(), jsonrpc.Err(jsonrpc.InternalError, err.Error()) + } // Create block info blockInfo, rpcErr := h.buildBlockInfo(header) if rpcErr != nil { @@ -462,6 +494,7 @@ func (h *Handler) traceBlockWithVM( state, headState, &blockInfo, + plan.Start, ) } From a557964a2118bed202e2e5ece5224b5e84132e41 Mon Sep 17 00:00:00 2001 From: danielntmd Date: Tue, 15 Sep 2026 15:40:46 -0700 Subject: [PATCH 2/2] test(rpc): exercise progressive tracing through public handlers --- rpc/v10/progressive_trace_test.go | 290 ++++++++++++++++++------------ 1 file changed, 171 insertions(+), 119 deletions(-) diff --git a/rpc/v10/progressive_trace_test.go b/rpc/v10/progressive_trace_test.go index 2cdc8a9ddf..61d8397094 100644 --- a/rpc/v10/progressive_trace_test.go +++ b/rpc/v10/progressive_trace_test.go @@ -1,20 +1,24 @@ -package rpcv10 +package rpcv10_test import ( "context" "encoding/json" "errors" "fmt" + "sync" "sync/atomic" "testing" + "testing/synctest" "github.com/NethermindEth/juno/blockchain/networks" "github.com/NethermindEth/juno/core" "github.com/NethermindEth/juno/core/felt" + "github.com/NethermindEth/juno/db" "github.com/NethermindEth/juno/jsonrpc" "github.com/NethermindEth/juno/mocks" "github.com/NethermindEth/juno/rpc/rpccore" - "github.com/NethermindEth/juno/rpc/tracecache" + rpcv10 "github.com/NethermindEth/juno/rpc/v10" + "github.com/NethermindEth/juno/sync/preconfirmed" "github.com/NethermindEth/juno/utils/log" "github.com/NethermindEth/juno/vm" "github.com/stretchr/testify/require" @@ -50,12 +54,19 @@ func progressiveResult(txs []core.Transaction) vm.ExecutionResults { return result } -func progressiveHandler( - t *testing.T, runner vm.VM, -) (*Handler, *core.Header, []core.Transaction, *mocks.MockStateReader) { +type progressiveFixture struct { + handler *rpcv10.Handler + blockID rpcv10.BlockID + txs []core.Transaction + head *mocks.MockStateReader + reader *mocks.MockReader +} + +func progressiveHandler(t *testing.T, runner vm.VM) progressiveFixture { t.Helper() ctrl := gomock.NewController(t) reader := mocks.NewMockReader(ctrl) + syncReader := mocks.NewMockSyncReader(ctrl) parent := mocks.NewMockStateReader(ctrl) head := mocks.NewMockStateReader(ctrl) header := &core.Header{ @@ -66,13 +77,25 @@ func progressiveHandler( txs := make([]core.Transaction, 3) for i := range txs { txs[i] = &core.InvokeTransaction{TransactionHash: felt.NewFromUint64[felt.Felt](uint64(i + 1))} + reader.EXPECT().BlockNumberAndIndexByTxHash( + (*felt.TransactionHash)(txs[i].Hash()), + ).Return(header.Number, uint64(i), nil).AnyTimes() } + reader.EXPECT().BlockHeaderByNumber(header.Number).Return(header, nil).AnyTimes() reader.EXPECT().Network().Return(&networks.Mainnet).AnyTimes() reader.EXPECT().TransactionsByBlockNumber(header.Number).Return(txs, nil).AnyTimes() reader.EXPECT().StateAtBlockHash(header.ParentHash). Return(parent, func() error { return nil }, nil).AnyTimes() reader.EXPECT().HeadState().Return(head, func() error { return nil }, nil).AnyTimes() - return New(reader, nil, runner, log.NewNopZapLogger()), header, txs, head + syncReader.EXPECT().PreConfirmedChain(). + Return(preconfirmed.ChainReader{}, db.ErrKeyNotFound).AnyTimes() + return progressiveFixture{ + handler: rpcv10.New(reader, syncReader, runner, log.NewNopZapLogger()), + blockID: rpcv10.BlockIDFromNumber(header.Number), + txs: txs, + head: head, + reader: reader, + } } func TestProgressiveFailureRetainsPrefixAndOffsetsError(t *testing.T) { @@ -88,56 +111,73 @@ func TestProgressiveFailureRetainsPrefixAndOffsetsError(t *testing.T) { require.Len(t, txs, 1) return progressiveResult(txs), nil }} - h, header, txs, _ := progressiveHandler(t, runner) - first := &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()} - second := &tracecache.TransactionTarget{Index: 1, Hash: txs[1].Hash()} - prefix, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, first) + f := progressiveHandler(t, runner) + first := (*felt.TransactionHash)(f.txs[0].Hash()) + second := (*felt.TransactionHash)(f.txs[1].Hash()) + prefix, _, rpcErr := f.handler.TraceTransaction(t.Context(), first) require.Nil(t, rpcErr) - _, steps, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, second) + _, steps, rpcErr := f.handler.TraceBlockTransactions(t.Context(), &f.blockID, nil) require.NotNil(t, rpcErr) require.Contains(t, rpcErr.Data, "transaction #1") - require.Equal(t, "9", steps.Get(ExecutionStepsHeader)) - hit, steps, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, first) + require.Equal(t, "9", steps.Get(rpcv10.ExecutionStepsHeader)) + hit, steps, rpcErr := f.handler.TraceTransaction(t.Context(), first) require.Nil(t, rpcErr) require.Equal(t, prefix, hit) - require.Equal(t, "0", steps.Get(ExecutionStepsHeader)) - next, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, second) + require.Equal(t, "0", steps.Get(rpcv10.ExecutionStepsHeader)) + next, _, rpcErr := f.handler.TraceTransaction(t.Context(), second) require.Nil(t, rpcErr) - require.Len(t, next.Traces, 2) + require.Equal(t, uint64(2), next.ExecutionResources.L1Gas) require.Equal(t, uint64(3), calls.Load()) } func TestProgressiveWaiterCancellation(t *testing.T) { - entered, release := make(chan struct{}), make(chan struct{}) - var calls atomic.Uint64 - runner := &progressiveVM{run: func( - txs []core.Transaction, _ core.StateReader, _ vm.TraceOptions, - ) (vm.ExecutionResults, error) { - calls.Add(1) - close(entered) - <-release - return progressiveResult(txs), nil - }} - h, header, txs, _ := progressiveHandler(t, runner) - target := &tracecache.TransactionTarget{Index: 1, Hash: txs[1].Hash()} - done := make(chan *jsonrpc.Error, 1) - go func() { - _, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) - done <- rpcErr - }() - <-entered - ctx, cancel := context.WithCancel(t.Context()) - cancel() - _, _, rpcErr := h.traceFinalisedBlock(ctx, header, false, target) - require.NotNil(t, rpcErr) - require.Contains(t, rpcErr.Data, context.Canceled.Error()) - close(release) - require.Nil(t, <-done) - result, steps, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) - require.Nil(t, rpcErr) - require.Len(t, result.Traces, 2) - require.Equal(t, "0", steps.Get(ExecutionStepsHeader)) - require.Equal(t, uint64(1), calls.Load()) + synctest.Test(t, func(t *testing.T) { + entered, release := make(chan struct{}), make(chan struct{}) + releaseProducer := sync.OnceFunc(func() { close(release) }) + defer releaseProducer() + var calls atomic.Uint64 + runner := &progressiveVM{run: func( + txs []core.Transaction, _ core.StateReader, _ vm.TraceOptions, + ) (vm.ExecutionResults, error) { + if calls.Add(1) == 1 { + close(entered) + } + <-release + return progressiveResult(txs), nil + }} + f := progressiveHandler(t, runner) + target := (*felt.TransactionHash)(f.txs[1].Hash()) + owner := make(chan *jsonrpc.Error, 1) + go func() { + _, _, rpcErr := f.handler.TraceTransaction(t.Context(), target) + owner <- rpcErr + }() + <-entered + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + waiter := make(chan *jsonrpc.Error, 1) + go func() { + _, _, rpcErr := f.handler.TraceTransaction(ctx, target) + waiter <- rpcErr + }() + synctest.Wait() + require.Empty(t, waiter, "request must wait for the producer") + require.Equal(t, uint64(1), calls.Load()) + cancel() + synctest.Wait() + require.Len(t, waiter, 1, "cancellation must wake the waiter") + rpcErr := <-waiter + require.NotNil(t, rpcErr) + require.Contains(t, rpcErr.Data, context.Canceled.Error()) + require.Empty(t, owner, "cancellation must leave the producer running") + releaseProducer() + require.Nil(t, <-owner) + result, steps, rpcErr := f.handler.TraceTransaction(t.Context(), target) + require.Nil(t, rpcErr) + require.Equal(t, uint64(2), result.ExecutionResources.L1Gas) + require.Equal(t, "0", steps.Get(rpcv10.ExecutionStepsHeader)) + require.Equal(t, uint64(1), calls.Load()) + }) } func TestProgressivePanicAndMalformedResultsDoNotPublish(t *testing.T) { @@ -163,10 +203,10 @@ func TestProgressivePanicAndMalformedResultsDoNotPublish(t *testing.T) { } return result, nil }} - h, header, txs, _ := progressiveHandler(t, runner) - target := &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()} + f := progressiveHandler(t, runner) + target := (*felt.TransactionHash)(f.txs[0].Hash()) request := func() { - _, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + _, _, rpcErr := f.handler.TraceTransaction(t.Context(), target) require.NotNil(t, rpcErr) } if failure == "panic" { @@ -174,66 +214,76 @@ func TestProgressivePanicAndMalformedResultsDoNotPublish(t *testing.T) { } else { request() } - result, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + result, _, rpcErr := f.handler.TraceTransaction(t.Context(), target) require.Nil(t, rpcErr) - require.Len(t, result.Traces, 1) + require.Equal(t, uint64(1), result.ExecutionResources.L1Gas) require.Equal(t, 2, calls) }) } } func TestProgressiveInitialReadsReplayPreservesPrefix(t *testing.T) { - entered, release := make(chan struct{}), make(chan struct{}) - var calls atomic.Uint64 - runner := &progressiveVM{run: func( - txs []core.Transaction, _ core.StateReader, opts vm.TraceOptions, - ) (vm.ExecutionResults, error) { - call := calls.Add(1) - result := progressiveResult(txs) - if call == 1 { - require.Len(t, txs, 1) - require.False(t, opts.ReturnInitialReads) - } else { - require.Len(t, txs, 3) - require.True(t, opts.ReturnInitialReads) + synctest.Test(t, func(t *testing.T) { + entered, release := make(chan struct{}), make(chan struct{}) + releaseProducer := sync.OnceFunc(func() { close(release) }) + defer releaseProducer() + var calls atomic.Uint64 + type execution struct { + transactions []core.Transaction + initialReads bool + } + observed := make(chan execution, 3) + runner := &progressiveVM{run: func( + txs []core.Transaction, _ core.StateReader, opts vm.TraceOptions, + ) (vm.ExecutionResults, error) { + call := calls.Add(1) + observed <- execution{transactions: txs, initialReads: opts.ReturnInitialReads} + result := progressiveResult(txs) if call == 2 { close(entered) <-release - } else { + } else if call > 2 { result.InitialReads = &vm.InitialReads{} } + return result, nil + }} + f := progressiveHandler(t, runner) + target := (*felt.TransactionHash)(f.txs[0].Hash()) + prefix, _, rpcErr := f.handler.TraceTransaction(t.Context(), target) + require.Nil(t, rpcErr) + require.Equal(t, execution{transactions: f.txs[:1]}, <-observed) + flags := []rpcv10.TraceFlag{rpcv10.TraceReturnInitialReadsFlag} + done := make(chan *jsonrpc.Error, 1) + go func() { + _, _, rpcErr := f.handler.TraceBlockTransactions(t.Context(), &f.blockID, flags) + done <- rpcErr + }() + <-entered + require.Equal(t, execution{transactions: f.txs, initialReads: true}, <-observed) + hit, steps, rpcErr := f.handler.TraceTransaction(t.Context(), target) + require.Nil(t, rpcErr) + require.Equal(t, prefix, hit) + require.Equal(t, "0", steps.Get(rpcv10.ExecutionStepsHeader)) + releaseProducer() + rpcErr = <-done + require.NotNil(t, rpcErr) + require.Contains(t, rpcErr.Data, "VM omitted initial reads") + hit, _, rpcErr = f.handler.TraceTransaction(t.Context(), target) + require.Nil(t, rpcErr) + require.Equal(t, prefix, hit) + full, _, rpcErr := f.handler.TraceBlockTransactions(t.Context(), &f.blockID, flags) + require.Nil(t, rpcErr) + require.Equal(t, execution{transactions: f.txs, initialReads: true}, <-observed) + require.Len(t, full.Traces, len(f.txs)) + require.NotNil(t, full.InitialReads) + for i := range full.Traces { + require.Equal(t, f.txs[i].Hash(), full.Traces[i].TransactionHash) } - return result, nil - }} - h, header, txs, _ := progressiveHandler(t, runner) - target := &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()} - prefix, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) - require.Nil(t, rpcErr) - done := make(chan *jsonrpc.Error, 1) - go func() { - _, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, true, nil) - done <- rpcErr - }() - <-entered - hit, steps, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) - require.Nil(t, rpcErr) - require.Equal(t, prefix, hit) - require.Equal(t, "0", steps.Get(ExecutionStepsHeader)) - close(release) - rpcErr = <-done - require.NotNil(t, rpcErr) - require.Contains(t, rpcErr.Data, "VM omitted initial reads") - hit, _, rpcErr = h.traceFinalisedBlock(t.Context(), header, false, target) - require.Nil(t, rpcErr) - require.Equal(t, prefix, hit) - full, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, true, nil) - require.Nil(t, rpcErr) - require.Len(t, full.Traces, 3) - require.NotNil(t, full.InitialReads) - _, steps, rpcErr = h.traceFinalisedBlock(t.Context(), header, true, nil) - require.Nil(t, rpcErr) - require.Equal(t, "0", steps.Get(ExecutionStepsHeader)) - require.Equal(t, uint64(3), calls.Load()) + _, steps, rpcErr = f.handler.TraceBlockTransactions(t.Context(), &f.blockID, flags) + require.Nil(t, rpcErr) + require.Equal(t, "0", steps.Get(rpcv10.ExecutionStepsHeader)) + require.Equal(t, uint64(3), calls.Load()) + }) } func TestProgressiveCheckpointDeclarationsAndStorage(t *testing.T) { @@ -263,14 +313,14 @@ func TestProgressiveCheckpointDeclarationsAndStorage(t *testing.T) { } return result, nil }} - h, header, txs, head := progressiveHandler(t, runner) - head.EXPECT().Class(&classHash).Return(&core.DeclaredClassDefinition{Class: definition}, nil) - _, _, rpcErr := h.traceFinalisedBlock( - t.Context(), header, false, &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()}, + f := progressiveHandler(t, runner) + f.head.EXPECT().Class(&classHash).Return(&core.DeclaredClassDefinition{Class: definition}, nil) + _, _, rpcErr := f.handler.TraceTransaction( + t.Context(), (*felt.TransactionHash)(f.txs[0].Hash()), ) require.Nil(t, rpcErr) - _, _, rpcErr = h.traceFinalisedBlock( - t.Context(), header, false, &tracecache.TransactionTarget{Index: 1, Hash: txs[1].Hash()}, + _, _, rpcErr = f.handler.TraceTransaction( + t.Context(), (*felt.TransactionHash)(f.txs[1].Hash()), ) require.Nil(t, rpcErr) require.Equal(t, 2, calls) @@ -284,21 +334,23 @@ func TestProgressiveTargetIdentityBeforeExecutionAndOnHit(t *testing.T) { calls++ return progressiveResult(txs), nil }} - h, header, txs, _ := progressiveHandler(t, runner) - bad := &tracecache.TransactionTarget{Index: 0, Hash: felt.NewFromUint64[felt.Felt](999)} - _, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, bad) + f := progressiveHandler(t, runner) + bad := felt.NewFromUint64[felt.TransactionHash](999) + f.reader.EXPECT().BlockNumberAndIndexByTxHash(bad). + Return(f.blockID.Number(), uint64(0), nil).Times(2) + _, _, rpcErr := f.handler.TraceTransaction(t.Context(), bad) require.Equal(t, rpccore.ErrTxnHashNotFound, rpcErr) require.Zero(t, calls) - _, _, rpcErr = h.traceFinalisedBlock( - t.Context(), header, false, &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()}, + _, _, rpcErr = f.handler.TraceTransaction( + t.Context(), (*felt.TransactionHash)(f.txs[0].Hash()), ) require.Nil(t, rpcErr) - _, _, rpcErr = h.traceFinalisedBlock(t.Context(), header, false, bad) + _, _, rpcErr = f.handler.TraceTransaction(t.Context(), bad) require.Equal(t, rpccore.ErrTxnHashNotFound, rpcErr) require.Equal(t, 1, calls) - _, _, rpcErr = h.traceFinalisedBlock( - t.Context(), header, false, &tracecache.TransactionTarget{Index: 3, Hash: txs[0].Hash()}, - ) + f.reader.EXPECT().BlockNumberAndIndexByTxHash(bad). + Return(f.blockID.Number(), uint64(len(f.txs)), nil) + _, _, rpcErr = f.handler.TraceTransaction(t.Context(), bad) require.Equal(t, rpccore.ErrTxnHashNotFound, rpcErr) require.Equal(t, 1, calls) } @@ -314,18 +366,18 @@ func TestProgressiveCheckpointReadFailurePreservesPrefix(t *testing.T) { result.Traces[0].StateDiff.DeprecatedDeclaredClasses = []*felt.Felt{&classHash} return result, nil }} - h, header, txs, head := progressiveHandler(t, runner) - head.EXPECT().Class(&classHash).Return(nil, errors.New("class read failed")) - target := &tracecache.TransactionTarget{Index: 0, Hash: txs[0].Hash()} - _, _, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + f := progressiveHandler(t, runner) + f.head.EXPECT().Class(&classHash).Return(nil, errors.New("class read failed")) + target := (*felt.TransactionHash)(f.txs[0].Hash()) + _, _, rpcErr := f.handler.TraceTransaction(t.Context(), target) require.Nil(t, rpcErr) - _, _, rpcErr = h.traceFinalisedBlock( - t.Context(), header, false, &tracecache.TransactionTarget{Index: 1, Hash: txs[1].Hash()}, + _, _, rpcErr = f.handler.TraceTransaction( + t.Context(), (*felt.TransactionHash)(f.txs[1].Hash()), ) require.NotNil(t, rpcErr) require.Equal(t, jsonrpc.InternalError, rpcErr.Code) - _, steps, rpcErr := h.traceFinalisedBlock(t.Context(), header, false, target) + _, steps, rpcErr := f.handler.TraceTransaction(t.Context(), target) require.Nil(t, rpcErr) - require.Equal(t, "0", steps.Get(ExecutionStepsHeader)) + require.Equal(t, "0", steps.Get(rpcv10.ExecutionStepsHeader)) require.Equal(t, 1, calls) }