Keep reports from all completed inputs

internal/advancer/advancer_test.go

@@ -519,7 +519,7 @@ func (s *AdvancerSuite) TestProcess() {
 			terminal := randomAdvanceResult(1)
 			terminal.Status = status
 			terminal.Outputs = nil
-			terminal.Reports = nil
+			terminal.Reports = [][]byte{[]byte("terminal report")}
 			if status == InputCompletionStatus_Exception {
 				terminal.ExceptionData = []byte("guest exception")
 			}
@@ -537,6 +537,7 @@ func (s *AdvancerSuite) TestProcess() {
 			require.True(stopped)
 			require.Len(env.repo.StoredResults, 2)
 			require.Equal(status, env.repo.StoredResults[1].Status)
+			require.Equal(terminal.Reports, env.repo.StoredResults[1].Reports)
 			require.Equal([]byte("must not execute"), inputs[2].RawData)
 		})
 	}
@@ -547,7 +548,7 @@ func (s *AdvancerSuite) TestProcess() {
 		rejected := randomAdvanceResult(0)
 		rejected.Status = InputCompletionStatus_Rejected
 		rejected.Outputs = nil
-		rejected.Reports = nil
+		rejected.Reports = [][]byte{[]byte("rejection report")}
 		accepted := randomAdvanceResult(1)
 
 		processed, stopped, err := env.service.processInputs(
@@ -560,6 +561,7 @@ func (s *AdvancerSuite) TestProcess() {
 		require.Equal(uint64(2), processed)
 		require.False(stopped)
 		require.Len(env.repo.StoredResults, 2)
+		require.Equal(rejected.Reports, env.repo.StoredResults[0].Reports)
 	})
 
 	s.Run("Noop", func() {
@@ -1581,7 +1583,7 @@ func (s *AdvancerSuite) TestTerminalInputStopsEpochAndFutureTicks() {
 	terminal := randomAdvanceResult(1)
 	terminal.Status = InputCompletionStatus_MachineHalted
 	terminal.Outputs = nil
-	terminal.Reports = nil
+	terminal.Reports = [][]byte{[]byte("terminal report")}
 	address := env.app.Application.IApplicationAddress
 	env.repo.GetEpochsReturn = map[common.Address][]*Epoch{
 		address: {{
@@ -1603,6 +1605,7 @@ func (s *AdvancerSuite) TestTerminalInputStopsEpochAndFutureTicks() {
 	require.NoError(err)
 	require.False(hadWork)
 	require.Len(env.repo.StoredResults, 2)
+	require.Equal(terminal.Reports, env.repo.StoredResults[1].Reports)
 	require.Zero(env.repo.EpochInputsProcessedCount,
 		"a terminal epoch must remain closed because it has no accepted terminal state")
 

internal/advancer/determinism_test.go

@@ -355,17 +355,16 @@ func requireDeterminismTarget(
 	require.Len(t, target.PeriodicStateHashes, 2, "a PRT result must retain its periodic state hashes")
 	require.Equal(t, machine.InputEntryCapacity-uint64(len(target.PeriodicStateHashes)), target.PaddingRepetitions)
 
+	require.Equal(t, [][]byte{append([]byte("report:"), targetPayload...)}, target.Reports)
 	if wantStatus == model.InputCompletionStatus_Accepted {
 		require.True(t, target.IsComplete())
 		require.Equal(t, [][]byte{append([]byte("output:"), targetPayload...)}, target.Outputs)
-		require.Equal(t, [][]byte{append([]byte("report:"), targetPayload...)}, target.Reports)
 		require.NotEqual(t, prefix.MachineHash, target.MachineHash)
 		require.NotEqual(t, prefix.TxBufferDataBlock, target.TxBufferDataBlock)
 		return
 	}
 
-	require.Empty(t, target.Outputs, "effects are canonical only for accepted inputs")
-	require.Empty(t, target.Reports, "effects are canonical only for accepted inputs")
+	require.Empty(t, target.Outputs, "outputs are canonical only for accepted inputs")
 	if wantStatus.IsTerminal() {
 		require.True(t, target.IsComplete(),
 			"a terminal result must preserve its actual post-run proof")
@@ -823,12 +822,12 @@ func (m *determinismRuntime) Advance(
 	hashes := []machine.Hash{firstHash, finalHash}
 	response := &machine.AdvanceResponse{
 		Status:              status,
+		Reports:             []machine.Report{report},
 		PeriodicStateHashes: hashes,
 		PaddingRepetitions:  machine.InputEntryCapacity - uint64(len(hashes)),
 	}
 	if status == machine.CompletionStatusAccepted {
 		response.Outputs = []machine.Output{output}
-		response.Reports = []machine.Report{report}
 	} else if status == machine.CompletionStatusException {
 		response.ExceptionData = append([]byte{}, input...)
 	}

internal/advancer/invalid_outputs_root_test.go

@@ -16,7 +16,7 @@ func (s *AdvancerSuite) TestInvalidOutputsRootStopsClaimPreparation() {
 	terminal := randomAdvanceResult(0)
 	terminal.Status = model.InputCompletionStatus_InvalidOutputsRoot
 	terminal.Outputs = nil
-	terminal.Reports = nil
+	terminal.Reports = [][]byte{[]byte("invalid outputs root report")}
 	address := env.app.Application.IApplicationAddress
 	env.repo.GetEpochsReturn = map[common.Address][]*model.Epoch{
 		address: {{
@@ -36,6 +36,7 @@ func (s *AdvancerSuite) TestInvalidOutputsRootStopsClaimPreparation() {
 	s.False(hadWork)
 	s.Require().Len(env.repo.StoredResults, 1)
 	s.Equal(model.InputCompletionStatus_InvalidOutputsRoot, env.repo.StoredResults[0].Status)
+	s.Equal(terminal.Reports, env.repo.StoredResults[0].Reports)
 	s.Zero(env.repo.EpochInputsProcessedCount, "the epoch must not become claim work")
 	s.Zero(env.repo.ApplicationStatusUpdates, "the status belongs to the atomic input-result write")
 
@@ -51,7 +52,7 @@ func (s *AdvancerSuite) TestInvalidOutputsRootStoreFailureStopsService() {
 	terminal := randomAdvanceResult(0)
 	terminal.Status = model.InputCompletionStatus_InvalidOutputsRoot
 	terminal.Outputs = nil
-	terminal.Reports = nil
+	terminal.Reports = [][]byte{[]byte("invalid outputs root report")}
 	pending := newInput(env.app.Application.ID, 0, 0, marshal(terminal))
 	env.repo.StoreAdvanceError = errors.New("terminal result write failed")
 

internal/manager/instance.go

@@ -281,6 +281,7 @@ func (m *MachineInstanceImpl) Advance(
 		EpochIndex:          epochIndex,
 		InputIndex:          index,
 		Status:              status,
+		Reports:             advanceResp.Reports,
 		ExceptionData:       advanceResp.ExceptionData,
 		PeriodicStateHashes: advanceResp.PeriodicStateHashes,
 		PaddingRepetitions:  advanceResp.PaddingRepetitions,
@@ -306,10 +307,10 @@ func (m *MachineInstanceImpl) Advance(
 		}
 		result.StateProof = *postProof
 		result.Outputs = advanceResp.Outputs
-		result.Reports = advanceResp.Reports
 		adoptFork = true
 	case model.InputCompletionStatus_Rejected:
-		// Rejected execution has no canonical state transition or effects.
+		// Rejection keeps the predecessor state and discards outputs. Reports
+		// remain available as diagnostics from the completed execution.
 		result.StateProof = *prevProof
 	case model.InputCompletionStatus_Exception,
 		model.InputCompletionStatus_MachineHalted,

internal/manager/instance_test.go

@@ -404,9 +404,14 @@ func (s *MachineInstanceSuite) TestAdvance() {
 			require.NotNil(res)
 
 			require.Same(inner, instance.runtime)
+			preProof, proofErr := stateProofFromMachine(fork.StateProofReturn)
+			require.NoError(proofErr)
+			require.Equal(*preProof, res.StateProof)
+			require.Equal(int64(1), fork.CloseCalls.Load())
+			require.Zero(inner.CloseCalls.Load())
 			require.Equal(model.InputCompletionStatus_Rejected, res.Status)
 			require.Empty(res.Outputs)
-			require.Empty(res.Reports)
+			require.Equal(expectedReports1, res.Reports)
 			require.Equal(newHash(1), res.TxBufferDataBlock)
 			require.Equal(newHash(2), res.MachineHash)
 			require.True(res.IsComplete())
@@ -444,8 +449,11 @@ func (s *MachineInstanceSuite) TestAdvance() {
 				require.Nil(instance.runtime)
 				require.Equal(inputStatus, res.Status)
 				require.Equal(exceptionData, res.ExceptionData)
+				wantProof, proofErr := stateProofFromMachine(postProof)
+				require.NoError(proofErr)
+				require.Equal(*wantProof, res.StateProof)
 				require.Empty(res.Outputs)
-				require.Empty(res.Reports)
+				require.Equal(expectedReports1, res.Reports)
 				require.Equal(newHash(1), res.TxBufferDataBlock)
 				require.Equal(newHash(2), res.MachineHash)
 				require.True(res.IsComplete())

internal/manager/invalid_outputs_root_test.go

@@ -57,7 +57,7 @@ func (s *MachineInstanceSuite) TestInvalidOutputsRootPreservesFinalEvidence() {
 			s.Equal(fork.AdvanceRemainingReturn, result.PaddingRepetitions)
 			s.Equal(collect, result.IsDaveConsensus)
 			s.Empty(result.Outputs)
-			s.Empty(result.Reports)
+			s.Equal(expectedReports1, result.Reports)
 			s.Nil(result.ExceptionData)
 			s.Equal(2, proofCalls)
 			s.Equal(uint64(6), instance.ProcessedInputs())

internal/replay/compare.go

@@ -100,17 +100,14 @@ func compareRecord(
 		if err := compareBytes("outputs", record.Outputs, actual.Outputs, contradiction); err != nil {
 			return err
 		}
-		if err := compareBytes("reports", record.Reports, actual.Reports, contradiction); err != nil {
-			return err
-		}
 	} else {
-		// Effects of nonaccepted executions are diagnostics, not canonical.
+		// Outputs from nonaccepted executions must not be persisted.
 		if len(record.Outputs) != 0 {
 			return contradiction("outputs.count", 0, len(record.Outputs))
 		}
-		if len(record.Reports) != 0 {
-			return contradiction("reports.count", 0, len(record.Reports))
-		}
+	}
+	if err := compareBytes("reports", record.Reports, actual.Reports, contradiction); err != nil {
+		return err
 	}
 
 	if !isPRT {

internal/replay/compare_test.go

@@ -141,6 +141,8 @@ func TestCompareReplayRecordCompletionMatrix(t *testing.T) {
 		model.InputCompletionStatus_Rejected,
 		model.InputCompletionStatus_Exception,
 		model.InputCompletionStatus_MachineHalted,
+		model.InputCompletionStatus_Overflow,
+		model.InputCompletionStatus_UnexpectedYield,
 		model.InputCompletionStatus_InvalidOutputsRoot,
 	}
 	consensuses := []model.Consensus{
@@ -154,7 +156,8 @@ func TestCompareReplayRecordCompletionMatrix(t *testing.T) {
 				app, record, actual := replayFixture(status, consensus)
 				if status != model.InputCompletionStatus_Accepted {
 					actual.Outputs = [][]byte{[]byte("noncanonical diagnostic output")}
-					actual.Reports = [][]byte{[]byte("noncanonical diagnostic report")}
+					record.Reports = [][]byte{[]byte("diagnostic report")}
+					actual.Reports = [][]byte{[]byte("diagnostic report")}
 				}
 				if consensus == model.Consensus_PRT {
 					checkpoint := common.HexToHash("0x33")
@@ -271,26 +274,22 @@ func TestCompareReplayRecordPersistedRecordValidation(t *testing.T) {
 			ErrContradiction,
 		)
 	})
-	for _, effect := range []string{"output", "report"} {
-		t.Run("nonaccepted persisted "+effect, func(t *testing.T) {
-			app, record, actual := replayFixture(model.InputCompletionStatus_Rejected, model.Consensus_Authority)
-			if effect == "output" {
-				record.Outputs = [][]byte{[]byte("illegal")}
-			} else {
-				record.Reports = [][]byte{[]byte("illegal")}
-			}
-			// Replay diagnostics are ignored, but persisted effects are corruption.
-			actual.Outputs = [][]byte{[]byte("diagnostic")}
-			actual.Reports = [][]byte{[]byte("diagnostic")}
-			require.ErrorIs(t,
-				compareRecord(app.Name, app.ID, app.IsDaveConsensus(), repository.ReplayVerificationFull, record, actual),
-				ErrContradiction,
-			)
-		})
-	}
-	t.Run("nonaccepted replay diagnostics ignored", func(t *testing.T) {
+	t.Run("nonaccepted persisted output", func(t *testing.T) {
+		app, record, actual := replayFixture(model.InputCompletionStatus_Rejected, model.Consensus_Authority)
+		record.Outputs = [][]byte{[]byte("illegal")}
+		actual.Outputs = [][]byte{[]byte("diagnostic")}
+		record.Reports = [][]byte{[]byte("diagnostic")}
+		actual.Reports = [][]byte{[]byte("diagnostic")}
+		err := compareRecord(app.Name, app.ID, app.IsDaveConsensus(), repository.ReplayVerificationFull, record, actual)
+		require.ErrorIs(t, err, ErrContradiction)
+		var detail *ContradictionError
+		require.ErrorAs(t, err, &detail)
+		require.Equal(t, "outputs.count", detail.Field)
+	})
+	t.Run("nonaccepted replay outputs ignored", func(t *testing.T) {
 		app, record, actual := replayFixture(model.InputCompletionStatus_Exception, model.Consensus_Authority)
 		actual.Outputs = [][]byte{[]byte("diagnostic")}
+		record.Reports = [][]byte{[]byte("diagnostic")}
 		actual.Reports = [][]byte{[]byte("diagnostic")}
 		require.NoError(t, compareRecord(app.Name, app.ID, app.IsDaveConsensus(), repository.ReplayVerificationFull, record, actual))
 	})

internal/replay/report_test.go

@@ -0,0 +1,96 @@
+// (c) Cartesi and individual authors (see AUTHORS)
+// SPDX-License-Identifier: Apache-2.0 (see LICENSE)
+
+package replay
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cartesi/rollups-node/internal/model"
+	"github.com/cartesi/rollups-node/internal/repository"
+)
+
+func TestCompareRecordNonacceptedReports(t *testing.T) {
+	t.Parallel()
+	for _, status := range model.InputCompletionStatusAllValues {
+		if status == model.InputCompletionStatus_Accepted || !status.IsCompleted() {
+			continue
+		}
+		t.Run(status.String(), func(t *testing.T) {
+			for _, test := range []struct {
+				name    string
+				reports [][]byte
+				field   string
+			}{
+				{"matching", [][]byte{[]byte("first"), []byte("second")}, ""},
+				{"payload", [][]byte{[]byte("changed"), []byte("second")}, "reports[0]"},
+				{"count", [][]byte{[]byte("first")}, "reports.count"},
+				{"order", [][]byte{[]byte("second"), []byte("first")}, "reports[0]"},
+			} {
+				t.Run(test.name, func(t *testing.T) {
+					app, record, actual := replayFixture(status, model.Consensus_Authority)
+					record.Reports = [][]byte{[]byte("first"), []byte("second")}
+					actual.Reports = test.reports
+					err := compareRecord(app.Name, app.ID, false, repository.ReplayVerificationFull, record, actual)
+					if test.field == "" {
+						require.NoError(t, err)
+						return
+					}
+					require.ErrorIs(t, err, ErrContradiction)
+					var detail *ContradictionError
+					require.ErrorAs(t, err, &detail)
+					require.Equal(t, test.field, detail.Field)
+				})
+			}
+		})
+	}
+}
+
+func TestCompareRecordLegacyReports(t *testing.T) {
+	t.Parallel()
+	app, record, actual := replayFixture(model.InputCompletionStatus_Rejected, model.Consensus_Authority)
+	actual.Reports = [][]byte{[]byte("historically discarded")}
+	require.NoError(t, compareRecord(app.Name, app.ID, false, repository.ReplayVerificationCanonical, record, actual))
+	var detail *ContradictionError
+	err := compareRecord(app.Name, app.ID, false, repository.ReplayVerificationFull, record, actual)
+	require.ErrorAs(t, err, &detail)
+	require.Equal(t, "reports.count", detail.Field)
+	require.Equal(t, "0", detail.Expected)
+	require.Equal(t, "1", detail.Actual)
+}
+
+func TestRunLegacyReports(t *testing.T) {
+	t.Parallel()
+	for _, level := range []repository.ReplayVerificationLevel{
+		repository.ReplayVerificationCanonical, repository.ReplayVerificationFull,
+	} {
+		t.Run(level.String(), func(t *testing.T) {
+			record := replayRecords(1)[0]
+			record.Input.Status = model.InputCompletionStatus_Rejected
+			source := &fakeSource{
+				summary: model.ReplaySummary{ApplicationID: 7, ProcessedInputs: 1, Consensus: model.Consensus_Authority},
+				records: []*model.ReplayRecord{record},
+			}
+			executor := &fakeExecutor{
+				statuses: map[uint64]model.InputCompletionStatus{0: model.InputCompletionStatus_Rejected},
+				reports:  map[uint64][][]byte{0: {[]byte("historically discarded")}},
+			}
+			opts := replayOptions(model.Consensus_Authority, 0, 1)
+			opts.Verification = level
+			result, err := Run(context.Background(), source, executor, opts)
+			if level == repository.ReplayVerificationCanonical {
+				require.NoError(t, err)
+				require.Equal(t, uint64(1), result.ReplayedInputs)
+			} else {
+				require.ErrorIs(t, err, ErrContradiction)
+				var detail *ContradictionError
+				require.ErrorAs(t, err, &detail)
+				require.Equal(t, "reports.count", detail.Field)
+			}
+			require.Nil(t, record.Reports, "replay must leave the source record unchanged")
+		})
+	}
+}

internal/replay/run_test.go

@@ -68,6 +68,7 @@ type fakeExecutor struct {
 	wrongResultPos bool
 	fullPRTResult  bool
 	statuses       map[uint64]model.InputCompletionStatus
+	reports        map[uint64][][]byte
 }
 
 func (executor *fakeExecutor) ProcessedInputs() uint64 { return executor.processed }
@@ -97,6 +98,7 @@ func (executor *fakeExecutor) Advance(
 		EpochIndex: epochIndex,
 		InputIndex: resultIndex,
 		Status:     model.InputCompletionStatus_Accepted,
+		Reports:    executor.reports[inputIndex],
 		StateProof: model.StateProof{
 			MachineHash:       common.BigToHash(newBig(inputIndex + 1)),
 			TxBufferDataBlock: common.BigToHash(newBig(inputIndex + 100)),