diff --git a/Makefile b/Makefile index f872cd90a..8621dbfb4 100644 --- a/Makefile +++ b/Makefile @@ -681,7 +681,7 @@ check-license: ## Verify license headers on Go source files INTEGRATION_SHARDS := basic quorum prt replay restart withdrawal awskms INTEGRATION_SHARD_basic := ^Test(EchoAuthority|RejectException|TerminalMachineStates|MultiApp|EchoAuthorityStaging|IntegrationCLI|IntegrationLogScanner|TransactionFeeReplacement)$$ -INTEGRATION_SHARD_quorum := ^Test(EchoQuorum|SameBlockInputs)$$ +INTEGRATION_SHARD_quorum := ^Test(EchoQuorum|SameBlockInputs|ProcessRSS)$$ INTEGRATION_SHARD_prt := ^Test(EchoPrt|RejectExceptionPrt|ForeclosePrt|PrtPassiveDisputeObserver|PrtRecoveryFixtureConfig|ReaderConfigClearsSignerAndCanRestoreExactSubmittingRecord|RecoveryFixturePreservesFinalizedProfileAndParentLimits|SparseDisputeCommitment(MatchesDenseTrees|CanonicalGeometry|RejectsOutOfRangeRequests)|PassiveObserver(CleanupPreservesRestoreOrder|Config(PreservesOriginalAndUnrelatedFields|RejectsMissingFieldsAndInvalidJSON))|PrtCommitmentMatchesMachineCLI)$$ INTEGRATION_SHARD_replay := ^Test(Foreclose|ForecloseReplay|DivergentClaim|IndependentClaimWatch|IndependentClaimObserverCorrectClaims)$$ INTEGRATION_SHARD_restart := ^Test(Restart|SnapshotPolicy|NodeSubprocess|ManagedNodeDeadlineStopsChildrenBeforeProcessExit)$$ diff --git a/cmd/cartesi-rollups-cli/root/contract/epoch.go b/cmd/cartesi-rollups-cli/root/contract/epoch.go index 719023f97..ba4435bce 100644 --- a/cmd/cartesi-rollups-cli/root/contract/epoch.go +++ b/cmd/cartesi-rollups-cli/root/contract/epoch.go @@ -136,7 +136,7 @@ func (c *chainClient) epochHistoryDave(consensusAddr common.Address) (*EpochHist var epochs []EpochEvent limit := epochLimitFlag - onHit := func(block uint64) error { + onHit := func(block uint64, _, _ *big.Int) error { if limit > 0 && len(epochs) >= limit { return errLimitReached } @@ -207,8 +207,8 @@ func (c *chainClient) epochHistoryDave(consensusAddr common.Address) (*EpochHist prevValue = val } - _, err = ethutil.FindTransitions( - c.callOpts.Context, fromBlock, toBlock, prevValue, oracle, onHit, + _, _, err = ethutil.FindTransitions( + c.callOpts.Context, fromBlock, toBlock, prevValue, nil, oracle, onHit, ) if err != nil && !errors.Is(err, errLimitReached) { return nil, fmt.Errorf("find sealed epoch transitions: %w", err) @@ -261,7 +261,7 @@ func (c *chainClient) epochHistoryAuthority( var claims []ClaimEvent limit := epochLimitFlag - onHit := func(block uint64) error { + onHit := func(block uint64, _, _ *big.Int) error { if limit > 0 && len(claims) >= limit { return errLimitReached } @@ -360,8 +360,8 @@ func (c *chainClient) epochHistoryAuthority( prevValue = val } - _, err = ethutil.FindTransitions( - c.callOpts.Context, fromBlock, toBlock, prevValue, oracle, onHit, + _, _, err = ethutil.FindTransitions( + c.callOpts.Context, fromBlock, toBlock, prevValue, nil, oracle, onHit, ) if err != nil && !errors.Is(err, errLimitReached) { return nil, fmt.Errorf("find claim transitions: %w", err) @@ -428,7 +428,7 @@ func (c *chainClient) epochHistoryQuorum( return consensusCaller.GetNumberOfAcceptedClaims(opts, c.appAddr) } - onHit := func(block uint64) error { + onHit := func(block uint64, _, _ *big.Int) error { if limit > 0 && len(claims) >= limit { return errLimitReached } @@ -489,8 +489,8 @@ func (c *chainClient) epochHistoryQuorum( prevValue = val } - _, err = ethutil.FindTransitions( - c.callOpts.Context, fromBlock, toBlock, prevValue, oracle, onHit, + _, _, err = ethutil.FindTransitions( + c.callOpts.Context, fromBlock, toBlock, prevValue, nil, oracle, onHit, ) if err != nil && !errors.Is(err, errLimitReached) { return nil, fmt.Errorf("find claim accepted transitions: %w", err) diff --git a/cmd/cartesi-rollups-cli/root/contract/match.go b/cmd/cartesi-rollups-cli/root/contract/match.go index 419e863b3..fb0c7cb0f 100644 --- a/cmd/cartesi-rollups-cli/root/contract/match.go +++ b/cmd/cartesi-rollups-cli/root/contract/match.go @@ -261,7 +261,7 @@ func (c *chainClient) findMatchCommitments( return caller.GetMatchCreatedCount(opts) } - onHit := func(block uint64) error { + onHit := func(block uint64, _, _ *big.Int) error { q, qErr := buildEventFilterQuery( tournamentAddr, "MatchCreated", itournament.ITournamentMetaData, block, block, @@ -291,8 +291,8 @@ func (c *chainClient) findMatchCommitments( return nil } - _, err = ethutil.FindTransitions( - c.callOpts.Context, deployBlock, c.blockNum, big.NewInt(0), oracle, onHit, + _, _, err = ethutil.FindTransitions( + c.callOpts.Context, deployBlock, c.blockNum, big.NewInt(0), nil, oracle, onHit, ) if err != nil && !errors.Is(err, errFound) { return [32]byte{}, [32]byte{}, fmt.Errorf("find match commitments: %w", err) diff --git a/cmd/cartesi-rollups-cli/root/contract/tournament.go b/cmd/cartesi-rollups-cli/root/contract/tournament.go index 1fa36caf6..1d1647a73 100644 --- a/cmd/cartesi-rollups-cli/root/contract/tournament.go +++ b/cmd/cartesi-rollups-cli/root/contract/tournament.go @@ -245,7 +245,7 @@ func (c *chainClient) findTournamentForEpoch( var result common.Address - onHit := func(block uint64) error { + onHit := func(block uint64, _, _ *big.Int) error { q, qErr := buildEventFilterQuery( consensusAddr, "EpochSealed", idaveconsensus.IDaveConsensusMetaData, block, block, @@ -278,8 +278,8 @@ func (c *chainClient) findTournamentForEpoch( } // Use -1 as prevValue so epoch 0 is detected as a transition from -1→0. - _, err = ethutil.FindTransitions( - c.callOpts.Context, deployBlock, c.blockNum, big.NewInt(-1), oracle, onHit, + _, _, err = ethutil.FindTransitions( + c.callOpts.Context, deployBlock, c.blockNum, big.NewInt(-1), nil, oracle, onHit, ) if err != nil && !errors.Is(err, errFound) { return common.Address{}, fmt.Errorf("find epoch %d tournament: %w", targetEpoch, err) @@ -738,7 +738,7 @@ func (c *chainClient) findAndCollectEvents( oracle ethutil.TransitionQueryFn, process func(*types.Log) (done bool, err error), ) error { - onHit := func(block uint64) error { + onHit := func(block uint64, _, _ *big.Int) error { q, qErr := buildEventFilterQuery( addr, eventName, itournament.ITournamentMetaData, block, block, ) @@ -774,8 +774,8 @@ func (c *chainClient) findAndCollectEvents( return val, oErr } - _, err := ethutil.FindTransitions( - c.callOpts.Context, fromBlock, c.blockNum, big.NewInt(0), safeOracle, onHit, + _, _, err := ethutil.FindTransitions( + c.callOpts.Context, fromBlock, c.blockNum, big.NewInt(0), nil, safeOracle, onHit, ) if err != nil && !errors.Is(err, errLimitReached) { return fmt.Errorf("find %s transitions: %w", eventName, err) @@ -904,7 +904,7 @@ func (c *chainClient) discoverChildTournaments( return caller.GetNewInnerTournamentCount(opts) } - onHit := func(block uint64) error { + onHit := func(block uint64, _, _ *big.Int) error { q, qErr := buildEventFilterQuery( tournamentAddr, "NewInnerTournament", itournament.ITournamentMetaData, block, block, @@ -929,8 +929,8 @@ func (c *chainClient) discoverChildTournaments( return nil } - _, err = ethutil.FindTransitions( - c.callOpts.Context, fromBlock, c.blockNum, big.NewInt(0), oracle, onHit, + _, _, err = ethutil.FindTransitions( + c.callOpts.Context, fromBlock, c.blockNum, big.NewInt(0), nil, oracle, onHit, ) if err != nil { return nil, fmt.Errorf("find child tournament transitions: %w", err) diff --git a/internal/claimer/blockchain.go b/internal/claimer/blockchain.go index 09b603ada..7938df0e8 100644 --- a/internal/claimer/blockchain.go +++ b/internal/claimer/blockchain.go @@ -177,8 +177,8 @@ func newOnHit[IT eventIterator]( filter func(*bind.FilterOpts, []common.Address, []common.Address) (IT, error), oracle func(context.Context, uint64) (*big.Int, error), onEvent func(IT) error, -) func(block uint64) error { - return func(block uint64) (err error) { +) ethutil.OnHitFn { + return func(block uint64, _, _ *big.Int) (err error) { before := new(big.Int) if block > 0 { before, err = oracle(ctx, block-1) @@ -265,7 +265,7 @@ func (cb *claimerBlockchain) findClaimSubmittedEventAndSucc( return nil, nil, fmt.Errorf("querying number of submitted claims for epoch %v (%v) before block %d: %w", epoch.Index, epoch.VirtualIndex, fromBlock, err) } - _, err = ethutil.FindTransitions(ctx, fromBlock, toBlock, prevValue, oracle, onHit) + _, _, err = ethutil.FindTransitions(ctx, fromBlock, toBlock, prevValue, nil, oracle, onHit) if err != nil { return nil, nil, fmt.Errorf("walking ClaimSubmitted transitions for application: %v, epoch %v (%v): %w", application.IApplicationAddress, epoch.Index, epoch.VirtualIndex, err) @@ -327,7 +327,7 @@ func (cb *claimerBlockchain) findClaimStagedEventAndSucc( if err != nil { return nil, nil, nil, fmt.Errorf("querying number of staged claims before block %d: %w", fromBlock, err) } - _, err = ethutil.FindTransitions(ctx, fromBlock, toBlock, prevValue, oracle, onHit) + _, _, err = ethutil.FindTransitions(ctx, fromBlock, toBlock, prevValue, nil, oracle, onHit) if err != nil { return nil, nil, nil, fmt.Errorf("walking ClaimStaged transitions for application: %v, epoch %v (%v): %w", application.IApplicationAddress, epoch.Index, epoch.VirtualIndex, err) @@ -397,7 +397,7 @@ func (cb *claimerBlockchain) findClaimAcceptedEventAndSucc( if err != nil { return nil, nil, nil, fmt.Errorf("querying number of accepted claims before block %d: %w", fromBlock, err) } - _, err = ethutil.FindTransitions(ctx, fromBlock, toBlock, prevValue, oracle, onHit) + _, _, err = ethutil.FindTransitions(ctx, fromBlock, toBlock, prevValue, nil, oracle, onHit) if err != nil { return nil, nil, nil, fmt.Errorf("walking ClaimAccepted transitions for application: %v, epoch %v (%v): %w", application.IApplicationAddress, epoch.Index, epoch.VirtualIndex, err) diff --git a/internal/claimer/observation_test.go b/internal/claimer/observation_test.go index 20ba664ce..41635ccba 100644 --- a/internal/claimer/observation_test.go +++ b/internal/claimer/observation_test.go @@ -97,7 +97,7 @@ func TestObservedEventWindowRetainsIteratorCloseFailures(t *testing.T) { }, func(_ context.Context, block uint64) (*big.Int, error) { return new(big.Int).SetUint64(block), nil }, func(*observationIterator) error { return failure.callback }) - err := onHit(50) + err := onHit(50, nil, nil) require.ErrorIs(t, err, closeErr) if failure.callback != nil { require.ErrorIs(t, err, failure.callback) @@ -127,9 +127,9 @@ func TestCounterWindowRejectsIncompleteLogs(t *testing.T) { return big.NewInt(9), nil }, func(*observationIterator) error { return nil }) if count == 2 { - require.NoError(t, onHit(50)) + require.NoError(t, onHit(50, nil, nil)) } else { - require.ErrorIs(t, onHit(50), txsender.ErrInconclusive) + require.ErrorIs(t, onHit(50, nil, nil), txsender.ErrInconclusive) } }) } diff --git a/internal/claimer/prior_counter_test.go b/internal/claimer/prior_counter_test.go index b4fc8ccb5..98678eccb 100644 --- a/internal/claimer/prior_counter_test.go +++ b/internal/claimer/prior_counter_test.go @@ -142,13 +142,13 @@ func TestFindTransitions_PrevValueRegression(t *testing.T) { "sanity: oracle(currEpoch.LastBlock=90) must observe both transitions") hits := []uint64{} - onHit := func(block uint64) error { + onHit := func(block uint64, _, _ *big.Int) error { hits = append(hits, block) return nil } - _, err = ethutil.FindTransitions(ctx, fromBlock, currEpochLastBlk, - buggyPrevValue, oracle, onHit) + _, _, err = ethutil.FindTransitions(ctx, fromBlock, currEpochLastBlk, + buggyPrevValue, nil, oracle, onHit) require.Error(t, err, "the buggy prevValue MUST trip the monotonic-assumption check") assert.Contains(t, err.Error(), "monotonic assumption violated", "the specific error string is the reason this bug went undetected for so long; pin it") @@ -166,13 +166,13 @@ func TestFindTransitions_PrevValueRegression(t *testing.T) { "sanity: priorCounter at fromBlock=70 must read oracle(69)=0 (no transitions yet)") hits := []uint64{} - onHit := func(block uint64) error { + onHit := func(block uint64, _, _ *big.Int) error { hits = append(hits, block) return nil } - count, err := ethutil.FindTransitions(ctx, fromBlock, currEpochLastBlk, - fixedPrevValue, oracle, onHit) + count, _, err := ethutil.FindTransitions(ctx, fromBlock, currEpochLastBlk, + fixedPrevValue, nil, oracle, onHit) require.NoError(t, err, "priorCounter MUST satisfy FindTransitions' monotonic invariant") assert.Equal(t, uint64(len(transitions)), count) assert.Equal(t, transitions, hits, diff --git a/internal/config/generate/Config.toml b/internal/config/generate/Config.toml index 4442abe02..462d50259 100644 --- a/internal/config/generate/Config.toml +++ b/internal/config/generate/Config.toml @@ -112,6 +112,18 @@ Maximum number of inputs fetched per database query during advance processing an synchronization. Bounds memory usage for applications with large backlogs of unprocessed inputs.""" used-by = ["advancer", "node"] +[rollups.CARTESI_EVM_READER_MAX_INPUTS_PER_STEP] +default = "500" +go-type = "uint64" +description = """ +Maximum number of new input indices requested by the EVM reader in one service tick, +shared across applications and sealed and open epochs. Failed input queries also consume +their requested allowance. Inputs in the same block can be split across ticks. +Applications take turns receiving priority across ticks. Must be between 1 and 1000, +inclusive, to keep index topic lists within Geth's default limit of 1000 alternatives +per topic position.""" +used-by = ["evmreader", "node"] + [rollups.CARTESI_EVM_READER_POLLING_INTERVAL] default = "12" go-type = "Duration" diff --git a/internal/config/generated.go b/internal/config/generated.go index 80b60d17d..3b2c0aaeb 100644 --- a/internal/config/generated.go +++ b/internal/config/generated.go @@ -81,6 +81,7 @@ const ( BLOCKCHAIN_HTTP_RETRY_MIN_WAIT = "CARTESI_BLOCKCHAIN_HTTP_RETRY_MIN_WAIT" BLOCKCHAIN_MAX_BLOCK_RANGE = "CARTESI_BLOCKCHAIN_MAX_BLOCK_RANGE" CLAIMER_POLLING_INTERVAL = "CARTESI_CLAIMER_POLLING_INTERVAL" + EVM_READER_MAX_INPUTS_PER_STEP = "CARTESI_EVM_READER_MAX_INPUTS_PER_STEP" EVM_READER_POLLING_INTERVAL = "CARTESI_EVM_READER_POLLING_INTERVAL" EVM_READER_READY_MAX_STALENESS = "CARTESI_EVM_READER_READY_MAX_STALENESS" MAX_STARTUP_TIME = "CARTESI_MAX_STARTUP_TIME" @@ -228,6 +229,8 @@ func DeclaredDefault(name string) (string, bool) { return "2000", true case CLAIMER_POLLING_INTERVAL: return "3", true + case EVM_READER_MAX_INPUTS_PER_STEP: + return "500", true case EVM_READER_POLLING_INTERVAL: return "12", true case EVM_READER_READY_MAX_STALENESS: @@ -368,6 +371,8 @@ func SetDefaults() { viper.SetDefault(CLAIMER_POLLING_INTERVAL, "3") + viper.SetDefault(EVM_READER_MAX_INPUTS_PER_STEP, "500") + viper.SetDefault(EVM_READER_POLLING_INTERVAL, "12") viper.SetDefault(EVM_READER_READY_MAX_STALENESS, "0") @@ -832,6 +837,14 @@ type EvmreaderConfig struct { // Maximum number of blocks in a single query to the provider. Queries with larger ranges will be broken into multiple smaller queries. Zero for unlimited. BlockchainMaxBlockRange uint64 `mapstructure:"CARTESI_BLOCKCHAIN_MAX_BLOCK_RANGE"` + // Maximum number of new input indices requested by the EVM reader in one service tick, + // shared across applications and sealed and open epochs. Failed input queries also consume + // their requested allowance. Inputs in the same block can be split across ticks. + // Applications take turns receiving priority across ticks. Must be between 1 and 1000, + // inclusive, to keep index topic lists within Geth's default limit of 1000 alternatives + // per topic position. + EvmReaderMaxInputsPerStep uint64 `mapstructure:"CARTESI_EVM_READER_MAX_INPUTS_PER_STEP"` + // Time in seconds to wait before checking for a new block header. The default (12s) is tuned for mainnet. Reduce for faster chains or devnets. EvmReaderPollingInterval Duration `mapstructure:"CARTESI_EVM_READER_POLLING_INTERVAL"` @@ -944,6 +957,13 @@ func LoadEvmreaderConfig() (*EvmreaderConfig, error) { return nil, fmt.Errorf("CARTESI_BLOCKCHAIN_MAX_BLOCK_RANGE is required for the evmreader service: %w", err) } + cfg.EvmReaderMaxInputsPerStep, err = GetEvmReaderMaxInputsPerStep() + if err != nil && err != ErrNotDefined { + return nil, fmt.Errorf("failed to get CARTESI_EVM_READER_MAX_INPUTS_PER_STEP: %w", err) + } else if err == ErrNotDefined { + return nil, fmt.Errorf("CARTESI_EVM_READER_MAX_INPUTS_PER_STEP is required for the evmreader service: %w", err) + } + cfg.EvmReaderPollingInterval, err = GetEvmReaderPollingInterval() if err != nil && err != ErrNotDefined { return nil, fmt.Errorf("failed to get CARTESI_EVM_READER_POLLING_INTERVAL: %w", err) @@ -1208,6 +1228,14 @@ type NodeConfig struct { // How many seconds the node will wait before querying the database for new claims. ClaimerPollingInterval Duration `mapstructure:"CARTESI_CLAIMER_POLLING_INTERVAL"` + // Maximum number of new input indices requested by the EVM reader in one service tick, + // shared across applications and sealed and open epochs. Failed input queries also consume + // their requested allowance. Inputs in the same block can be split across ticks. + // Applications take turns receiving priority across ticks. Must be between 1 and 1000, + // inclusive, to keep index topic lists within Geth's default limit of 1000 alternatives + // per topic position. + EvmReaderMaxInputsPerStep uint64 `mapstructure:"CARTESI_EVM_READER_MAX_INPUTS_PER_STEP"` + // Time in seconds to wait before checking for a new block header. The default (12s) is tuned for mainnet. Reduce for faster chains or devnets. EvmReaderPollingInterval Duration `mapstructure:"CARTESI_EVM_READER_POLLING_INTERVAL"` @@ -1441,6 +1469,13 @@ func LoadNodeConfig() (*NodeConfig, error) { return nil, fmt.Errorf("CARTESI_CLAIMER_POLLING_INTERVAL is required for the node service: %w", err) } + cfg.EvmReaderMaxInputsPerStep, err = GetEvmReaderMaxInputsPerStep() + if err != nil && err != ErrNotDefined { + return nil, fmt.Errorf("failed to get CARTESI_EVM_READER_MAX_INPUTS_PER_STEP: %w", err) + } else if err == ErrNotDefined { + return nil, fmt.Errorf("CARTESI_EVM_READER_MAX_INPUTS_PER_STEP is required for the node service: %w", err) + } + cfg.EvmReaderPollingInterval, err = GetEvmReaderPollingInterval() if err != nil && err != ErrNotDefined { return nil, fmt.Errorf("failed to get CARTESI_EVM_READER_POLLING_INTERVAL: %w", err) @@ -1856,6 +1891,7 @@ func (c *NodeConfig) ToEvmreaderConfig() *EvmreaderConfig { BlockchainHttpRetryMaxWait: c.BlockchainHttpRetryMaxWait, BlockchainHttpRetryMinWait: c.BlockchainHttpRetryMinWait, BlockchainMaxBlockRange: c.BlockchainMaxBlockRange, + EvmReaderMaxInputsPerStep: c.EvmReaderMaxInputsPerStep, EvmReaderPollingInterval: c.EvmReaderPollingInterval, EvmReaderReadyMaxStaleness: c.EvmReaderReadyMaxStaleness, MaxStartupTime: c.MaxStartupTime, @@ -2745,6 +2781,19 @@ func GetClaimerPollingInterval() (Duration, error) { return notDefinedDuration(), fmt.Errorf("%s: %w", CLAIMER_POLLING_INTERVAL, ErrNotDefined) } +// GetEvmReaderMaxInputsPerStep returns the value for the environment variable CARTESI_EVM_READER_MAX_INPUTS_PER_STEP. +func GetEvmReaderMaxInputsPerStep() (uint64, error) { + s := viper.GetString(EVM_READER_MAX_INPUTS_PER_STEP) + if s != "" { + v, err := toUint64(s) + if err != nil { + return v, fmt.Errorf("failed to parse %s: %w", EVM_READER_MAX_INPUTS_PER_STEP, err) + } + return v, nil + } + return notDefineduint64(), fmt.Errorf("%s: %w", EVM_READER_MAX_INPUTS_PER_STEP, ErrNotDefined) +} + // GetEvmReaderPollingInterval returns the value for the environment variable CARTESI_EVM_READER_POLLING_INTERVAL. func GetEvmReaderPollingInterval() (Duration, error) { s := viper.GetString(EVM_READER_POLLING_INTERVAL) diff --git a/internal/evmreader/edge_cases_test.go b/internal/evmreader/edge_cases_test.go index b8dbfdaa2..f5b294038 100644 --- a/internal/evmreader/edge_cases_test.go +++ b/internal/evmreader/edge_cases_test.go @@ -6,7 +6,6 @@ package evmreader import ( "errors" "fmt" - "math" "math/big" . "github.com/cartesi/rollups-node/internal/model" @@ -60,7 +59,7 @@ func (s *EvmReaderSuite) TestFetchApplicationInputsDuplicateAndOutOfOrderEvents( }, nil) prevValue := new(big.Int).SetUint64(0) // repo returns 0 inputs - inputs, _, err := s.evmReader.fetchInputs(s.ctx, app, 10, 200, prevValue, 0, math.MaxUint64) + inputs, _, _, err := s.evmReader.fetchInputs(s.ctx, app, 10, 200, prevValue, 0, 300) s.Require().NoError(err) // 3 unique inputs, sorted by index @@ -101,8 +100,8 @@ func (s *EvmReaderSuite) TestFetchInputsSameTransactionHashPreservesLogIndex() { mock.Anything, mock.Anything, ).Return([]iinputbox.IInputBoxInputAdded{input0, input1}, nil) - inputs, _, err := s.evmReader.fetchInputs( - s.ctx, app, 10, 200, new(big.Int).SetUint64(0), 0, math.MaxUint64) + inputs, _, _, err := s.evmReader.fetchInputs( + s.ctx, app, 10, 200, new(big.Int).SetUint64(0), 0, 300) s.Require().NoError(err) s.Require().Len(inputs, 2) s.Equal(txHash, inputs[0].TransactionHash) @@ -213,9 +212,7 @@ func (s *EvmReaderSuite) TestInputLogIdentityConflictMarksApplicationCorrupted() } // --- Bounds filtering: inputs outside [lowerBound, upperBound) are excluded --- -// When RetrieveInputs returns events with indices outside the requested bounds, -// fetchInputs must exclude them. This exercises the bounds filtering added in the -// unified fetchInputs (C2 consolidation). +// The RPC topic filter must contain only indices inside the requested bounds. func (s *EvmReaderSuite) TestFetchInputsBoundsFiltering() { addr := common.HexToAddress("0x3333333333333333333333333333333333333333") @@ -236,20 +233,17 @@ func (s *EvmReaderSuite) TestFetchInputsBoundsFiltering() { inputSrc.On("GetNumberOfInputs", blockFrom(100), mock.Anything). Return(new(big.Int).SetUint64(5), nil) - // RetrieveInputs returns indices 0..4, but we only want [2, 4) + // Only indices 2 and 3 may be requested and returned. inputSrc.On("RetrieveInputs", mock.MatchedBy(func(opts *bind.FilterOpts) bool { return opts.Start == 100 }), mock.Anything, mock.Anything, ).Return([]iinputbox.IInputBoxInputAdded{ - makeInputEvent(addr, 0, 100), // below lowerBound → excluded - makeInputEvent(addr, 1, 100), // below lowerBound → excluded makeInputEvent(addr, 2, 100), // in bounds → included makeInputEvent(addr, 3, 100), // in bounds → included - makeInputEvent(addr, 4, 100), // >= upperBound → excluded }, nil) prevValue := new(big.Int).SetUint64(0) - inputs, _, err := s.evmReader.fetchInputs(s.ctx, app, 10, 200, prevValue, 2, 4) + inputs, _, _, err := s.evmReader.fetchInputs(s.ctx, app, 10, 200, prevValue, 2, 4) s.Require().NoError(err) // Only indices 2 and 3 should be included @@ -281,19 +275,12 @@ func (s *EvmReaderSuite) TestFetchInputsEmptyBoundsRange() { inputSrc.On("GetNumberOfInputs", blockFrom(100), mock.Anything). Return(new(big.Int).SetUint64(2), nil) - inputSrc.On("RetrieveInputs", - mock.MatchedBy(func(opts *bind.FilterOpts) bool { return opts.Start == 100 }), - mock.Anything, mock.Anything, - ).Return([]iinputbox.IInputBoxInputAdded{ - makeInputEvent(addr, 0, 100), - makeInputEvent(addr, 1, 100), - }, nil) - // lowerBound == upperBound == 5 → empty range prevValue := new(big.Int).SetUint64(0) - inputs, _, err := s.evmReader.fetchInputs(s.ctx, app, 10, 200, prevValue, 5, 5) + inputs, _, _, err := s.evmReader.fetchInputs(s.ctx, app, 10, 200, prevValue, 5, 5) s.Require().NoError(err) s.Require().Empty(inputs) + inputSrc.AssertNotCalled(s.T(), "RetrieveInputs", mock.Anything, mock.Anything, mock.Anything) } // --- Bounds filtering: boundary-exact inclusion/exclusion --- @@ -323,14 +310,12 @@ func (s *EvmReaderSuite) TestFetchInputsBoundaryExactness() { mock.MatchedBy(func(opts *bind.FilterOpts) bool { return opts.Start == 100 }), mock.Anything, mock.Anything, ).Return([]iinputbox.IInputBoxInputAdded{ - makeInputEvent(addr, 0, 100), makeInputEvent(addr, 1, 100), - makeInputEvent(addr, 2, 100), }, nil) // Range [1, 2): only index 1 should be included prevValue := new(big.Int).SetUint64(0) - inputs, _, err := s.evmReader.fetchInputs(s.ctx, app, 10, 200, prevValue, 1, 2) + inputs, _, _, err := s.evmReader.fetchInputs(s.ctx, app, 10, 200, prevValue, 1, 2) s.Require().NoError(err) s.Require().Len(inputs, 1) s.Require().Equal(uint64(1), inputs[0].Index) @@ -442,7 +427,7 @@ func (s *SealedEpochsSuite) TestFetchSealedEpochInputsRetrieveFailure() { ).Return(([]iinputbox.IInputBoxInputAdded)(nil), errors.New("RPC timeout")) prevValue := new(big.Int).SetUint64(epoch.InputIndexLowerBound) - _, _, err := s.evmReader.fetchInputs(s.ctx, app, + _, _, _, err := s.evmReader.fetchInputs(s.ctx, app, epoch.FirstBlock, epoch.LastBlock, prevValue, epoch.InputIndexLowerBound, epoch.InputIndexUpperBound) diff --git a/internal/evmreader/epoch_partial_block_test.go b/internal/evmreader/epoch_partial_block_test.go new file mode 100644 index 000000000..5d45899fa --- /dev/null +++ b/internal/evmreader/epoch_partial_block_test.go @@ -0,0 +1,114 @@ +// (c) Cartesi and individual authors (see AUTHORS) +// SPDX-License-Identifier: Apache-2.0 (see LICENSE) + +package evmreader + +import ( + "fmt" + "math/big" + "testing" + + . "github.com/cartesi/rollups-node/internal/model" + "github.com/cartesi/rollups-node/pkg/ethutil" + "github.com/ethereum/go-ethereum/accounts/abi/bind" + "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +func TestDavePartialBlockPersistsValidEpochAndResumes(t *testing.T) { + for _, mode := range []struct { + name string + firstSealed, lastSealed bool + }{ + {name: "open"}, + {name: "sealed", firstSealed: true, lastSealed: true}, + {name: "open then sealed", lastSealed: true}, + } { + for _, inputBlock := range []uint64{100, 101} { + t.Run(fmt.Sprintf("%s/input_block_%d", mode.name, inputBlock), func(t *testing.T) { + repo := newMockRepository() + source := newMockInputBox() + previous := &Epoch{Index: 0, FirstBlock: 10, LastBlock: 100, Status: EpochStatus_Closed} + event := makeSealedEpochEvent(1, 0, 1000, 102, common.HexToAddress("0xaaaa")) + source.On("GetNumberOfInputs", blockRange(100, inputBlock), app1Addr). + Return(big.NewInt(0), nil).Maybe() + source.On("GetNumberOfInputs", blockFrom(inputBlock), app1Addr).Return(big.NewInt(1000), nil) + for _, first := range []uint64{0, 500} { + source.On("RetrieveInputs", mock.MatchedBy(func(opts *bind.FilterOpts) bool { + return opts.Start == inputBlock && opts.End != nil && *opts.End == inputBlock + }), []common.Address{app1Addr}, mock.Anything). + Return(batchInputEvents(first, first+500, inputBlock), nil).Once() + } + + var persisted *Epoch + cursor := uint64(99) + stored := uint64(0) + repo.On("CreateEpochsAndInputs", t.Context(), app1Addr.Hex(), mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { + epochs := args.Get(2).(map[*Epoch][]*Input) + require.Len(t, epochs, 1) + for epoch, inputs := range epochs { + // PostgreSQL enforces these bounds on every write. + require.Equal(t, uint64(100), epoch.FirstBlock) + require.GreaterOrEqual(t, epoch.LastBlock, epoch.FirstBlock) + require.Len(t, inputs, 500) + for i, input := range inputs { + require.Equal(t, stored+uint64(i), input.Index) + require.Equal(t, inputBlock, input.BlockNumber) + } + stored += uint64(len(inputs)) + require.Equal(t, stored, epoch.InputIndexUpperBound) + copy := *epoch + persisted = © + } + cursor = args.Get(3).(uint64) + }).Return(nil).Twice() + + for batch, sealed := range []bool{mode.firstSealed, mode.lastSealed} { + // Reconstruct the reader and epoch row between batches to ensure + // resumption depends only on persisted state. + reader := &Service{repository: repo, maxInputsPerStep: 500} + reader.Logger = testLogger(t) + app := newDaveAppContracts(source, nil) + app.application.LastInputCheckBlock = cursor + repo.On("GetEpoch", t.Context(), app1Addr.Hex(), uint64(1)).Return(persisted, nil).Once() + var err error + if sealed { + repo.On("GetEpoch", t.Context(), app1Addr.Hex(), uint64(0)).Return(previous, nil).Once() + repo.On("UpdateEpochClaimTransactionHash", t.Context(), app1Addr.Hex(), previous).Return(nil).Once() + err = reader.processSealedEpochEvent(t.Context(), app, event) + } else { + repo.On("GetLastNonOpenEpoch", t.Context(), app1Addr.Hex()).Return(previous, nil).Once() + repo.On("GetEventLastCheckBlock", t.Context(), app.application.ID, MonitoredEvent_InputAdded). + Return(cursor, nil).Once() + repo.On("GetNumberOfInputs", t.Context(), app1Addr.Hex()).Return(stored, nil).Once() + err = reader.processApplicationOpenEpoch(t.Context(), app, 102) + } + if batch == 0 && sealed { + require.ErrorIs(t, err, ethutil.ErrIncompleteProcessing) + } else { + require.NoError(t, err) + } + if batch == 0 { + require.Equal(t, inputBlock, persisted.LastBlock) + require.Equal(t, inputBlock-1, cursor) + require.Equal(t, EpochStatus_Open, persisted.Status) + } + } + require.Equal(t, uint64(1000), stored) + if mode.lastSealed { + require.Equal(t, EpochStatus_Closed, persisted.Status) + require.Equal(t, uint64(102), persisted.LastBlock) + require.Equal(t, uint64(101), cursor) + } else { + require.Equal(t, EpochStatus_Open, persisted.Status) + require.Equal(t, inputBlock, persisted.LastBlock) + require.Equal(t, inputBlock, cursor) + } + repo.AssertExpectations(t) + source.AssertExpectations(t) + }) + } + } +} diff --git a/internal/evmreader/evmreader.go b/internal/evmreader/evmreader.go index 07046f017..3d50de20e 100644 --- a/internal/evmreader/evmreader.go +++ b/internal/evmreader/evmreader.go @@ -205,6 +205,15 @@ func (r *Service) runBlockScanners( if ctx.Err() != nil { return false } + // Every input scanner in this tick draws from the same allowance. Restore + // the previous scope on every exit, including cancellation and failures. + previousBudget := r.inputBudget + r.inputBudget = nil + if r.maxInputsPerStep != 0 { + remaining := r.maxInputsPerStep + r.inputBudget = &remaining + } + defer func() { r.inputBudget = previousBudget }() // Detect foreclosure first so later scanners use the marker observed in this same tick. success := r.checkForForeclosure(ctx, apps, blockNumber) if ctx.Err() != nil { @@ -213,11 +222,7 @@ func (r *Service) runBlockScanners( plan := buildBlockScanPlan(apps) - success = r.scanDaveConsensusEpochsAndInputs(ctx, plan.daveEpochTargets, blockNumber) && success - if ctx.Err() != nil { - return false - } - success = r.scanIConsensusInputs(ctx, plan.iConsensusInputTargets, blockNumber) && success + success = r.scanApplicationInputs(ctx, plan, blockNumber) && success if ctx.Err() != nil { return false } diff --git a/internal/evmreader/evmreader_test.go b/internal/evmreader/evmreader_test.go index 9b45193b5..db82f6592 100644 --- a/internal/evmreader/evmreader_test.go +++ b/internal/evmreader/evmreader_test.go @@ -65,6 +65,7 @@ func (s *EvmReaderSuite) SetupTest() { BlockchainDefaultBlock: DefaultBlock_Latest, BlockchainId: 1, EvmReaderPollingInterval: 100 * time.Millisecond, + EvmReaderMaxInputsPerStep: 500, }, Repository: s.repository, EthClient: s.client, diff --git a/internal/evmreader/input.go b/internal/evmreader/input.go index afe19ba59..a788a1ad8 100644 --- a/internal/evmreader/input.go +++ b/internal/evmreader/input.go @@ -188,6 +188,9 @@ func (r *Service) scanIConsensusInputUnit( if ctx.Err() != nil { return false } + if r.inputBudgetExhausted() { + return success + } err := r.readAndStoreApplicationInputs(ctx, unit.lastInputCheckBlock, unit.endBlock, app) if err != nil { success = false @@ -346,7 +349,7 @@ func (r *Service) readAndStoreApplicationInputs( return r.recordInputCorruption(ctx, app.application, "Application has epoch length of zero") } - inputs, err := r.readApplicationInputs(ctx, app, lastProcessedBlock+1, mostRecentBlockNumber) + inputs, lastInputBlock, err := r.readApplicationInputs(ctx, app, lastProcessedBlock+1, mostRecentBlockNumber) if err != nil { return err } @@ -354,7 +357,7 @@ func (r *Service) readAndStoreApplicationInputs( if err != nil { return fmt.Errorf("retrieve current epoch: %w", err) } - epochInputMap, err := indexInputsIntoEpochs(epochLength, currentEpoch, inputs, mostRecentBlockNumber) + epochInputMap, err := indexInputsIntoEpochs(epochLength, currentEpoch, inputs, lastInputBlock) if err != nil { if errors.Is(err, ErrInputForNonOpenEpoch) { return r.recordInputCorruption(ctx, app.application, "Should never happen. %v", err) @@ -366,7 +369,7 @@ func (r *Service) readAndStoreApplicationInputs( // No epoch needs to be stored or closed. Only this application's // successfully processed range is eligible for the cursor-only update. err = r.repository.UpdateEventLastCheckBlock( - ctx, []int64{app.application.ID}, MonitoredEvent_InputAdded, mostRecentBlockNumber) + ctx, []int64{app.application.ID}, MonitoredEvent_InputAdded, lastInputBlock) if err != nil { return fmt.Errorf("update input cursor: %w", err) } @@ -375,7 +378,7 @@ func (r *Service) readAndStoreApplicationInputs( // This transaction stores inputs, closes epochs, and advances the cursor. // An epoch closure with no new inputs must use the same atomic write. - err = r.repository.CreateEpochsAndInputs(ctx, address.String(), epochInputMap, mostRecentBlockNumber) + err = r.repository.CreateEpochsAndInputs(ctx, address.String(), epochInputMap, lastInputBlock) if err != nil { if errors.Is(err, repository.ErrInputLogIdentityConflict) { return r.recordInputCorruption(ctx, app.application, @@ -414,53 +417,76 @@ func (r *Service) readAndStoreApplicationInputs( return nil } +func (r *Service) inputBudgetExhausted() bool { + return r.inputBudget != nil && *r.inputBudget == 0 +} + +// inputBatchUpperBound uses the remaining tick budget when a tick is active, +// or the configured limit for standalone scans. Create requires a limit in +// [1, 1000]; direct internal use with a zero-valued limit remains unlimited. +// An exhausted tick budget yields an empty range. Addition saturates on overflow. +func (r *Service) inputBatchUpperBound(lower uint64) uint64 { + limit := r.maxInputsPerStep + if r.inputBudget != nil { + limit = *r.inputBudget + if limit == 0 { + return lower + } + } + if limit == 0 || limit > math.MaxUint64-lower { + return math.MaxUint64 + } + return lower + limit +} + // readApplicationInputs preserves the failure cause for the scan caller. It // validates the fetched logs against the counter observed in the same walk. func (r *Service) readApplicationInputs( ctx context.Context, app appContracts, startBlock, endBlock uint64, -) ([]*Input, error) { +) ([]*Input, uint64, error) { inputCount, err := r.repository.GetNumberOfInputs(ctx, app.application.IApplicationAddress.String()) if err != nil { - return nil, fmt.Errorf("get stored input count: %w", err) + return nil, startBlock - 1, fmt.Errorf("get stored input count: %w", err) } - inputs, endCount, err := r.fetchInputs( - ctx, app, startBlock, endBlock, new(big.Int).SetUint64(inputCount), 0, math.MaxUint64) + inputs, lastInputBlock, endCount, err := r.fetchInputs( + ctx, app, startBlock, endBlock, new(big.Int).SetUint64(inputCount), inputCount, r.inputBatchUpperBound(inputCount)) if err != nil { - return nil, fmt.Errorf("fetch inputs: %w", err) + return nil, startBlock - 1, fmt.Errorf("fetch inputs: %w", err) } expectedNew := endCount - inputCount if uint64(len(inputs)) != expectedNew { - return nil, fmt.Errorf("%w: input count mismatch: stored %d, on-chain %d, expected %d new inputs, got %d", + return nil, startBlock - 1, fmt.Errorf("%w: input count mismatch: stored %d, on-chain %d, expected %d new inputs, got %d", errInconclusiveInputView, inputCount, endCount, expectedNew, len(inputs)) } - return inputs, nil + return inputs, lastInputBlock, nil } -// fetchInputs locates blocks where new inputs were added via FindTransitions -// and accumulates them. The prevValue, lowerBound, and upperBound are caller- -// determined: IConsensus passes prevValue from the DB input count with full -// bounds [0, MaxUint64), while DaveConsensus sealed epochs use the epoch's -// InputIndexLowerBound as prevValue and the epoch's own index range as bounds. +// fetchInputs locates input transitions and fetches events with indices in +// [lowerBound, upperBound). IConsensus seeds prevValue from the stored input +// count and bounds the next batch. Dave sealed scans seed it from the epoch's +// next unstored index and also cap the batch at the seal's exclusive input bound. // -// The second return value is the input counter observed at endBlock during the -// same transition walk, letting callers validate the on-chain count delta -// without a second eth_call pinned to endBlock. +// The second return value is the last completely processed block. If a batch +// ends partway through a block, that block must be revisited with an updated +// prevValue. The third return value is the counter through the fetched batch, +// capped at upperBound; it includes any inputs fetched from a partial block. +// Callers can validate the count delta without another counter query. func (r *Service) fetchInputs( ctx context.Context, app appContracts, startBlock, endBlock uint64, prevValue *big.Int, lowerBound, upperBound uint64, -) ([]*Input, uint64, error) { +) ([]*Input, uint64, uint64, error) { prevCount := uint64(0) if prevValue != nil { prevCount = prevValue.Uint64() } if startBlock > endBlock { // Empty range: no new inputs, and the counter is unchanged from prevValue. - return nil, prevCount, nil + return nil, endBlock, prevCount, nil } r.Logger.Debug("Fetching inputs", @@ -491,12 +517,23 @@ func (r *Service) fetchInputs( } var sortedInputs []*Input - onHit := func(block uint64) error { + onHit := func(block uint64, prevValue, currValue *big.Int) error { filterOpts := &bind.FilterOpts{ Context: ctx, Start: block, End: &block, } + firstInputIdx := max(lowerBound, prevValue.Uint64()) + lastInputIdx := min(upperBound, currValue.Uint64()) + observedEndValue = new(big.Int).SetUint64(lastInputIdx) + if firstInputIdx >= lastInputIdx { + return nil + } + if r.inputBudget != nil { + // Reserve before allocating topics or calling RPC. Failed queries + // still consumed this tick's allowance and are retried next tick. + *r.inputBudget -= lastInputIdx - firstInputIdx + } inputEvents, err := app.inputSource.RetrieveInputs( filterOpts, []common.Address{app.application.IApplicationAddress}, @@ -531,17 +568,21 @@ func (r *Service) fetchInputs( ) } } + if lastInputIdx < currValue.Uint64() { + return ethutil.ErrIncompleteProcessing + } return nil } - _, err := ethutil.FindTransitions( - ctx, startBlock, endBlock, prevValue, oracle, onHit) - if err != nil { - return nil, 0, fmt.Errorf( + _, lastProcessedBlock, err := ethutil.FindTransitions( + ctx, startBlock, endBlock, prevValue, new(big.Int).SetUint64(upperBound), oracle, onHit) + if err != nil && !errors.Is(err, ethutil.ErrIncompleteProcessing) { + return nil, startBlock - 1, 0, fmt.Errorf( "failed to walk input transitions: %w", err) } if observedEndValue == nil { - return nil, 0, fmt.Errorf("failed to observe input count at end block %d", endBlock) + return nil, startBlock - 1, 0, fmt.Errorf( + "failed to observe input count at end block %d", endBlock) } r.Logger.Debug("Fetched inputs", @@ -552,5 +593,5 @@ func (r *Service) fetchInputs( "end_input_count", observedEndValue.Uint64(), "new_inputs", len(sortedInputs), ) - return sortedInputs, observedEndValue.Uint64(), nil + return sortedInputs, lastProcessedBlock, observedEndValue.Uint64(), nil } diff --git a/internal/evmreader/input_batches_test.go b/internal/evmreader/input_batches_test.go new file mode 100644 index 000000000..b96a9dc1e --- /dev/null +++ b/internal/evmreader/input_batches_test.go @@ -0,0 +1,323 @@ +// (c) Cartesi and individual authors (see AUTHORS) +// SPDX-License-Identifier: Apache-2.0 (see LICENSE) + +package evmreader + +import ( + "errors" + "math" + "math/big" + "testing" + + . "github.com/cartesi/rollups-node/internal/model" + "github.com/cartesi/rollups-node/pkg/contracts/idaveconsensus" + "github.com/cartesi/rollups-node/pkg/contracts/iinputbox" + "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +func TestInputBatchWithoutLimit(t *testing.T) { + for _, test := range []struct { + name string + limit, stored, added uint64 + }{ + {name: "zero limit from first input", added: 1000}, + {name: "zero limit after stored inputs", stored: 500, added: 1000}, + {name: "zero limit without new inputs", stored: 500}, + {name: "large limit does not wrap", limit: math.MaxUint64, stored: 500, added: 1000}, + } { + t.Run(test.name, func(t *testing.T) { + repo := newMockRepository() + source := newMockInputBox() + reader := &Service{repository: repo, maxInputsPerStep: test.limit} + reader.Logger = testLogger(t) + app := newDaveAppContracts(source, nil) + end := test.stored + test.added + repo.On("GetNumberOfInputs", t.Context(), app1Addr.Hex()).Return(test.stored, nil).Once() + source.On("GetNumberOfInputs", blockRange(100, 105), app1Addr). + Return(new(big.Int).SetUint64(test.stored), nil) + source.On("GetNumberOfInputs", blockFrom(105), app1Addr). + Return(new(big.Int).SetUint64(end), nil) + if test.added > 0 { + source.On("RetrieveInputs", mock.Anything, []common.Address{app1Addr}, mock.Anything). + Return(batchInputEvents(test.stored, end, 105), nil).Once() + } + + inputs, last, err := reader.readApplicationInputs(t.Context(), app, 100, 110) + require.NoError(t, err) + require.Len(t, inputs, int(test.added)) + require.Equal(t, uint64(110), last) + for i, input := range inputs { + require.Equal(t, test.stored+uint64(i), input.Index) + } + if test.added == 0 { + source.AssertNotCalled(t, "RetrieveInputs", mock.Anything, mock.Anything, mock.Anything) + } + repo.AssertExpectations(t) + source.AssertExpectations(t) + }) + } +} + +func TestEmptyInputBoundsDoNotQueryWildcardTopics(t *testing.T) { + source := newMockInputBox() + reader := &Service{} + reader.Logger = testLogger(t) + app := newDaveAppContracts(source, nil) + source.On("GetNumberOfInputs", mock.Anything, app1Addr).Return(big.NewInt(10), nil).Once() + + inputs, _, _, err := reader.fetchInputs(t.Context(), app, 100, 100, big.NewInt(0), 5, 5) + require.NoError(t, err) + require.Empty(t, inputs) + // An empty index topic list would match every input, rather than none. + source.AssertNotCalled(t, "RetrieveInputs", mock.Anything, mock.Anything, mock.Anything) + source.AssertExpectations(t) +} + +func batchInputEvents(first, end, block uint64) []iinputbox.IInputBoxInputAdded { + events := make([]iinputbox.IInputBoxInputAdded, 0, end-first) + for index := first; index < end; index++ { + events = append(events, makeInputEvent(app1Addr, index, block)) + } + return events +} + +func TestInputBatchResumesAtPartialBlockAfterGap(t *testing.T) { + repo := newMockRepository() + source := newMockInputBox() + reader := &Service{repository: repo, maxInputsPerStep: 500} + reader.Logger = testLogger(t) + app := newDaveAppContracts(source, nil) + source.On("GetNumberOfInputs", blockRange(100, 105), app1Addr).Return(big.NewInt(0), nil) + source.On("GetNumberOfInputs", blockFrom(105), app1Addr).Return(big.NewInt(1000), nil) + source.On("RetrieveInputs", mock.Anything, []common.Address{app1Addr}, mock.Anything). + Return(batchInputEvents(0, 500, 105), nil).Once() + source.On("RetrieveInputs", mock.Anything, []common.Address{app1Addr}, mock.Anything). + Return(batchInputEvents(500, 1000, 105), nil).Once() + repo.On("GetNumberOfInputs", t.Context(), app1Addr.Hex()).Return(uint64(0), nil).Once() + + inputs, last, err := reader.readApplicationInputs(t.Context(), app, 100, 110) + require.NoError(t, err) + require.Len(t, inputs, 500) + require.Equal(t, uint64(104), last) + + // Model the first batch having been committed before the next tick. + repo.On("GetNumberOfInputs", t.Context(), app1Addr.Hex()).Return(uint64(500), nil).Once() + inputs, last, err = reader.readApplicationInputs(t.Context(), app, last+1, 110) + require.NoError(t, err) + require.Len(t, inputs, 500) + require.Equal(t, uint64(500), inputs[0].Index) + require.Equal(t, uint64(999), inputs[len(inputs)-1].Index) + require.GreaterOrEqual(t, last, uint64(105)) + repo.AssertExpectations(t) + source.AssertExpectations(t) +} + +func TestDavePartialSealDefersOpenScanUntilNextTick(t *testing.T) { + repo := newMockRepository() + source := newMockInputBox() + dave := newMockDaveConsensus() + reader := &Service{repository: repo, maxInputsPerStep: 500} + reader.Logger = testLogger(t) + app := newDaveAppContracts(source, dave) + app.application.LastInputCheckBlock = 99 + app.application.LastEpochCheckBlock = 99 + app.application.Status = ApplicationStatus_OK + app.application.Enabled = true + app.application.ConsensusType = Consensus_PRT + app.application.LastOutputCheckBlock = 101 + app.application.LastForecloseCheckBlock = 101 + previous := &Epoch{Index: 0, LastBlock: 50, Status: EpochStatus_Closed} + open := &Epoch{Index: 1, FirstBlock: 50, LastBlock: 99, Status: EpochStatus_Open} + tournament := common.HexToAddress("0xaaaa") + event := makeSealedEpochEvent(1, 0, 600, 101, tournament) + + // Block 100 contains 500 inputs. Block 101 has 100 more before the seal + // and 100 after it, belonging to epoch 2. + dave.On("GetCurrentSealedEpoch", blockRange(99, 101)). + Return(makeSealedEpochResult(0, 0, 0, common.Address{}), nil) + dave.On("GetCurrentSealedEpoch", blockFrom(101)). + Return(makeSealedEpochResult(1, 0, 600, tournament), nil) + dave.On("RetrieveSealedEpochs", mock.Anything). + Return([]*idaveconsensus.IDaveConsensusEpochSealed{event}, nil).Twice() + repo.On("GetEpoch", t.Context(), app1Addr.Hex(), uint64(0)).Return(previous, nil).Twice() + repo.On("UpdateEpochClaimTransactionHash", t.Context(), app1Addr.Hex(), previous).Return(nil).Twice() + repo.On("GetEpoch", t.Context(), app1Addr.Hex(), uint64(1)).Return(open, nil).Twice() + source.On("GetNumberOfInputs", blockRange(99, 100), app1Addr).Return(big.NewInt(0), nil) + source.On("GetNumberOfInputs", blockRange(100, 101), app1Addr).Return(big.NewInt(500), nil) + source.On("GetNumberOfInputs", blockFrom(101), app1Addr).Return(big.NewInt(700), nil) + for _, batch := range []struct{ first, end, block uint64 }{ + {0, 500, 100}, {500, 600, 101}, {600, 700, 101}, + } { + source.On("RetrieveInputs", mock.Anything, []common.Address{app1Addr}, mock.Anything). + Return(batchInputEvents(batch.first, batch.end, batch.block), nil).Once() + } + + assignments := make(map[uint64]uint64) + for _, cursor := range []uint64{99, 100, 101} { + repo.On("CreateEpochsAndInputs", t.Context(), app1Addr.Hex(), mock.Anything, cursor). + Run(func(args mock.Arguments) { + for epoch, inputs := range args.Get(2).(map[*Epoch][]*Input) { + for _, input := range inputs { + _, duplicate := assignments[input.Index] + require.False(t, duplicate) + assignments[input.Index] = epoch.Index + } + } + app.application.LastInputCheckBlock = max(app.application.LastInputCheckBlock, args.Get(3).(uint64)) + }).Return(nil).Once() + } + for _, cursor := range []uint64{100, 101} { + repo.On("UpdateEventLastCheckBlock", t.Context(), []int64{app.application.ID}, MonitoredEvent_EpochSealed, cursor). + Run(func(args mock.Arguments) { app.application.LastEpochCheckBlock = args.Get(3).(uint64) }).Return(nil).Once() + } + // These expectations can only be reached after epoch 1 has closed. + repo.On("GetLastNonOpenEpoch", t.Context(), app1Addr.Hex()). + Run(func(mock.Arguments) { require.Equal(t, EpochStatus_Closed, open.Status) }).Return(open, nil).Once() + repo.On("GetEpoch", t.Context(), app1Addr.Hex(), uint64(2)).Return((*Epoch)(nil), nil).Once() + repo.On("GetEventLastCheckBlock", t.Context(), app.application.ID, MonitoredEvent_InputAdded).Return(uint64(100), nil).Once() + repo.On("GetNumberOfInputs", t.Context(), app1Addr.Hex()).Return(uint64(600), nil).Once() + // The same budget also covers a different consensus application's inputs. + otherSource := newMockInputBox() + other := appContracts{application: &Application{ + ID: 2, Enabled: true, Status: ApplicationStatus_OK, ConsensusType: Consensus_Authority, + IApplicationAddress: app2Addr, IInputBoxAddress: inputBoxAddr, IInputBoxBlock: 10, + EpochLength: 1000, LastInputCheckBlock: 100, LastOutputCheckBlock: 101, LastForecloseCheckBlock: 101, + }, inputSource: otherSource} + otherSource.On("GetNumberOfInputs", mock.Anything, app2Addr).Return(big.NewInt(300), nil) + otherEvents := make([]iinputbox.IInputBoxInputAdded, 300) + for i := range otherEvents { + otherEvents[i] = makeInputEvent(app2Addr, uint64(i), 101) + } + otherSource.On("RetrieveInputs", mock.Anything, []common.Address{app2Addr}, mock.Anything). + Return(otherEvents, nil).Once() + repo.On("GetNumberOfInputs", t.Context(), app2Addr.Hex()).Return(uint64(0), nil).Once() + repo.On("GetEpoch", t.Context(), app2Addr.Hex(), uint64(0)).Return((*Epoch)(nil), nil).Once() + otherStored := 0 + repo.On("CreateEpochsAndInputs", t.Context(), app2Addr.Hex(), mock.Anything, uint64(101)). + Run(func(args mock.Arguments) { + for _, inputs := range args.Get(2).(map[*Epoch][]*Input) { + otherStored += len(inputs) + } + }).Return(nil).Once() + apps := []appContracts{app, other} + + require.True(t, reader.runBlockScanners(t.Context(), apps, 101)) + require.Len(t, assignments, 500) + require.Zero(t, otherStored) + require.Nil(t, reader.inputBudget) + otherSource.AssertNotCalled(t, "GetNumberOfInputs", mock.Anything, mock.Anything) + require.Equal(t, EpochStatus_Open, open.Status) + require.Equal(t, uint64(500), open.InputIndexUpperBound) + require.Equal(t, uint64(100), app.application.LastEpochCheckBlock) + repo.AssertNotCalled(t, "GetLastNonOpenEpoch", mock.Anything, mock.Anything) + + require.True(t, reader.runBlockScanners(t.Context(), apps, 101)) + require.Len(t, assignments, 700) + require.Equal(t, 300, otherStored) + require.Nil(t, reader.inputBudget) + for index := uint64(0); index < 700; index++ { + wantEpoch := uint64(1) + if index >= 600 { + wantEpoch = 2 + } + require.Equal(t, wantEpoch, assignments[index], "input %d", index) + } + require.Equal(t, EpochStatus_Closed, open.Status) + require.Equal(t, uint64(600), open.InputIndexUpperBound) + require.Equal(t, uint64(101), app.application.LastEpochCheckBlock) + require.Equal(t, uint64(101), app.application.LastInputCheckBlock) + repo.AssertExpectations(t) + source.AssertExpectations(t) + dave.AssertExpectations(t) + otherSource.AssertExpectations(t) +} + +func TestInputBudgetSharedAcrossApplications(t *testing.T) { + for _, test := range []struct { + name string + limit uint64 + failFirst bool + }{ + {name: "shared and reset", limit: 500}, + {name: "zero is unlimited"}, + {name: "failed request consumes allowance", limit: 500, failFirst: true}, + } { + t.Run(test.name, func(t *testing.T) { + repo := newMockRepository() + source := newMockInputBox() + reader := &Service{repository: repo, maxInputsPerStep: test.limit} + reader.Logger = testLogger(t) + var apps []appContracts + stored := map[common.Address]int{} + for i, addr := range []common.Address{app1Addr, app2Addr} { + application := &Application{ + ID: int64(i + 1), Enabled: true, Status: ApplicationStatus_OK, ConsensusType: Consensus_Authority, + IApplicationAddress: addr, IInputBoxAddress: inputBoxAddr, IInputBoxBlock: 10, EpochLength: 1000, + LastInputCheckBlock: 99, LastOutputCheckBlock: 100, LastForecloseCheckBlock: 100, + } + apps = append(apps, appContracts{application: application, inputSource: source}) + source.On("GetNumberOfInputs", mock.Anything, addr).Return(big.NewInt(300), nil) + repo.On("GetNumberOfInputs", t.Context(), addr.Hex()).Return(uint64(0), nil).Once() + end := uint64(300) + if i == 1 && test.limit != 0 { + end = 200 + } + events := make([]iinputbox.IInputBoxInputAdded, end) + for index := range events { + events[index] = makeInputEvent(addr, uint64(index), 100) + } + logs := source.On("RetrieveInputs", mock.Anything, []common.Address{addr}, mock.Anything).Once() + if i == 0 && test.failFirst { + logs.Return([]iinputbox.IInputBoxInputAdded(nil), errors.New("RPC failure")) + continue + } + logs.Return(events, nil) + repo.On("GetEpoch", t.Context(), addr.Hex(), uint64(0)).Return((*Epoch)(nil), nil) + cursor := uint64(100) + if end < 300 { + cursor = 99 + } + repo.On("CreateEpochsAndInputs", t.Context(), addr.Hex(), mock.Anything, cursor). + Run(func(args mock.Arguments) { + for _, inputs := range args.Get(2).(map[*Epoch][]*Input) { + stored[addr] += len(inputs) + } + application.LastInputCheckBlock = args.Get(3).(uint64) + }).Return(nil).Once() + } + require.Equal(t, !test.failFirst, reader.runBlockScanners(t.Context(), apps, 100)) + require.Nil(t, reader.inputBudget) + if test.limit == 0 { + require.Equal(t, 600, stored[app1Addr]+stored[app2Addr]) + } else { + require.Equal(t, 200, stored[app2Addr]) + require.Equal(t, uint64(99), apps[1].application.LastInputCheckBlock) + if !test.failFirst { + require.Equal(t, 500, stored[app1Addr]+stored[app2Addr]) + repo.On("GetNumberOfInputs", t.Context(), app2Addr.Hex()).Return(uint64(200), nil).Once() + source.On("RetrieveInputs", mock.Anything, []common.Address{app2Addr}, mock.Anything). + Return(func() []iinputbox.IInputBoxInputAdded { + events := make([]iinputbox.IInputBoxInputAdded, 100) + for i := range events { + events[i] = makeInputEvent(app2Addr, uint64(i)+200, 100) + } + return events + }(), nil).Once() + repo.On("CreateEpochsAndInputs", t.Context(), app2Addr.Hex(), mock.Anything, uint64(100)). + Run(func(args mock.Arguments) { + for _, inputs := range args.Get(2).(map[*Epoch][]*Input) { + stored[app2Addr] += len(inputs) + } + }).Return(nil).Once() + require.True(t, reader.runBlockScanners(t.Context(), apps, 100)) + require.Equal(t, 600, stored[app1Addr]+stored[app2Addr]) + require.Nil(t, reader.inputBudget) + } + } + repo.AssertExpectations(t) + source.AssertExpectations(t) + }) + } +} diff --git a/internal/evmreader/input_deployment_test.go b/internal/evmreader/input_deployment_test.go index 3afbe0bf4..cbdb50921 100644 --- a/internal/evmreader/input_deployment_test.go +++ b/internal/evmreader/input_deployment_test.go @@ -70,7 +70,7 @@ func TestIConsensusPredeploymentObservationDoesNotCorrupt(t *testing.T) { repo := newMockRepository() inputBox := newMockInputBox() applicationContract := newMockApplicationContract() - r := &Service{repository: repo} + r := &Service{repository: repo, maxInputsPerStep: 500} r.Logger = testLogger(t) app := appContracts{ application: &model.Application{ diff --git a/internal/evmreader/input_scan_test.go b/internal/evmreader/input_scan_test.go index fed23db24..6cc762712 100644 --- a/internal/evmreader/input_scan_test.go +++ b/internal/evmreader/input_scan_test.go @@ -108,7 +108,7 @@ func inputScanFixture(t *testing.T, operation string, failure error, during func input := newMockInputBox() app.inputSource = input output := new(bytes.Buffer) - reader := &Service{repository: repo, defaultBlock: model.DefaultBlock_Latest} + reader := &Service{repository: repo, maxInputsPerStep: 500, defaultBlock: model.DefaultBlock_Latest} reader.Logger = slog.New(slog.NewJSONHandler(output, &slog.HandlerOptions{Level: slog.LevelDebug})) t.Cleanup(func() { repo.AssertExpectations(t) diff --git a/internal/evmreader/input_scheduling.go b/internal/evmreader/input_scheduling.go new file mode 100644 index 000000000..c6810c7fe --- /dev/null +++ b/internal/evmreader/input_scheduling.go @@ -0,0 +1,52 @@ +// (c) Cartesi and individual authors (see AUTHORS) +// SPDX-License-Identifier: Apache-2.0 (see LICENSE) + +package evmreader + +import ( + "context" + "sort" +) + +// scanApplicationInputs shares a rotating application order across consensus +// types. The next tick starts after the last attempted application, so a busy +// or failing application cannot repeatedly take the entire budget first. +func (r *Service) scanApplicationInputs(ctx context.Context, plan blockScanPlan, blockNumber uint64) bool { + targets := make([]appContracts, 0, len(plan.daveEpochTargets)+len(plan.iConsensusInputTargets)) + targets = append(targets, plan.daveEpochTargets...) + targets = append(targets, plan.iConsensusInputTargets...) + if len(targets) == 0 { + return true + } + // IDs provide a stable order even when applications are renamed, added, + // removed, or returned in a different order by the repository. + sort.SliceStable(targets, func(i, j int) bool { + return targets[i].application.ID < targets[j].application.ID + }) + start := sort.Search(len(targets), func(i int) bool { + return targets[i].application.ID > r.lastInputApplicationID + }) + if start == len(targets) { + start = 0 + } + + success := true + for offset := range len(targets) { + if ctx.Err() != nil { + return false + } + if r.inputBudgetExhausted() { + break + } + app := targets[(start+offset)%len(targets)] + if app.application.IsDaveConsensus() { + success = r.scanDaveConsensusEpochsAndInputs(ctx, []appContracts{app}, blockNumber) && success + } else { + // Dispatch one application at a time so InputBox/cursor grouping + // cannot reorder applications ahead of the rotating schedule. + success = r.scanIConsensusInputs(ctx, []appContracts{app}, blockNumber) && success + } + r.lastInputApplicationID = app.application.ID + } + return success +} diff --git a/internal/evmreader/input_scheduling_test.go b/internal/evmreader/input_scheduling_test.go new file mode 100644 index 000000000..1051969fc --- /dev/null +++ b/internal/evmreader/input_scheduling_test.go @@ -0,0 +1,173 @@ +// (c) Cartesi and individual authors (see AUTHORS) +// SPDX-License-Identifier: Apache-2.0 (see LICENSE) + +package evmreader + +import ( + "errors" + "math/big" + "testing" + + . "github.com/cartesi/rollups-node/internal/model" + "github.com/cartesi/rollups-node/pkg/contracts/idaveconsensus" + "github.com/cartesi/rollups-node/pkg/contracts/iinputbox" + "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +func schedulingApp(id int64, consensus Consensus, source InputSourceAdapter) appContracts { + return appContracts{application: &Application{ + ID: id, Enabled: true, Status: ApplicationStatus_OK, ConsensusType: consensus, + IApplicationAddress: common.BigToAddress(big.NewInt(id)), IConsensusAddress: consensusAddr, + IInputBoxAddress: inputBoxAddr, IInputBoxBlock: 10, EpochLength: 1000, + LastInputCheckBlock: 99, LastOutputCheckBlock: 101, LastForecloseCheckBlock: 101, + }, inputSource: source} +} + +func schedulingEvents(addr common.Address, first uint64) []iinputbox.IInputBoxInputAdded { + return []iinputbox.IInputBoxInputAdded{ + makeInputEvent(addr, first, 100), makeInputEvent(addr, first+1, 100), + } +} + +func TestInputSchedulingRotatesAcrossConsensusTypes(t *testing.T) { + for _, failFirst := range []bool{false, true} { + name := "busy application" + if failFirst { + name = "failing application" + } + t.Run(name, func(t *testing.T) { + repo := newMockRepository() + source := newMockInputBox() + dave := newMockDaveConsensus() + reader := &Service{repository: repo, maxInputsPerStep: 2} + reader.Logger = testLogger(t) + daveApp := schedulingApp(10, Consensus_PRT, source) + daveApp.daveConsensus = dave + daveApp.application.LastEpochCheckBlock = 99 + authority := schedulingApp(20, Consensus_Authority, source) + quorum := schedulingApp(30, Consensus_Quorum, source) + // An eight-input backlog exceeds every tick's allowance. No app + // finishes during this test, so progress requires rotating priority. + previous := &Epoch{Index: 0, FirstBlock: 10, LastBlock: 90, Status: EpochStatus_Closed} + open := &Epoch{Index: 1, FirstBlock: 90, LastBlock: 99, Status: EpochStatus_Open} + tournament := common.HexToAddress("0xaaaa") + event := makeSealedEpochEvent(1, 0, 8, 101, tournament) + attempts := 2 + if failFirst { + attempts = 1 + } + dave.On("GetCurrentSealedEpoch", blockRange(99, 101)). + Return(makeSealedEpochResult(0, 0, 0, common.Address{}), nil) + dave.On("GetCurrentSealedEpoch", blockFrom(101)). + Return(makeSealedEpochResult(1, 0, 8, tournament), nil) + dave.On("RetrieveSealedEpochs", mock.Anything). + Return([]*idaveconsensus.IDaveConsensusEpochSealed{event}, nil).Times(attempts) + address := daveApp.application.IApplicationAddress.Hex() + repo.On("GetEpoch", t.Context(), address, uint64(0)).Return(previous, nil).Times(attempts) + repo.On("UpdateEpochClaimTransactionHash", t.Context(), address, previous).Return(nil).Times(attempts) + repo.On("GetEpoch", t.Context(), address, uint64(1)).Return(open, nil).Times(attempts) + if !failFirst { + repo.On("UpdateEventLastCheckBlock", t.Context(), []int64{10}, MonitoredEvent_EpochSealed, uint64(100)). + Run(func(mock.Arguments) { daveApp.application.LastEpochCheckBlock = 100 }).Return(nil).Twice() + } + + var order []int64 + stored := make(map[string]int) + for _, app := range []appContracts{daveApp, authority, quorum} { + addr := app.application.IApplicationAddress + id := app.application.ID + source.On("GetNumberOfInputs", blockRange(99, 100), addr).Return(big.NewInt(0), nil).Maybe() + source.On("GetNumberOfInputs", blockFrom(100), addr).Return(big.NewInt(8), nil) + logs := source.On("RetrieveInputs", mock.Anything, []common.Address{addr}, mock.Anything). + Run(func(mock.Arguments) { + order = append(order, id) + require.NotNil(t, reader.inputBudget) + require.Zero(t, *reader.inputBudget) + }).Once() + if id == 10 && failFirst { + logs.Return([]iinputbox.IInputBoxInputAdded(nil), errors.New("RPC unavailable")) + } else { + logs.Return(schedulingEvents(addr, 0), nil) + } + if id != 10 { + repo.On("GetNumberOfInputs", t.Context(), addr.Hex()).Return(uint64(0), nil).Once() + repo.On("GetEpoch", t.Context(), addr.Hex(), uint64(0)).Return((*Epoch)(nil), nil).Once() + } + } + if !failFirst { + addr := daveApp.application.IApplicationAddress + source.On("RetrieveInputs", mock.Anything, []common.Address{addr}, mock.Anything). + Run(func(mock.Arguments) { order = append(order, 10) }). + Return(schedulingEvents(addr, 2), nil).Once() + } + writes := 4 + if failFirst { + writes = 2 + } + repo.On("CreateEpochsAndInputs", t.Context(), mock.Anything, mock.Anything, uint64(99)). + Run(func(args mock.Arguments) { + addr := args.String(1) + for _, inputs := range args.Get(2).(map[*Epoch][]*Input) { + require.Len(t, inputs, 2) + require.Equal(t, uint64(stored[addr]), inputs[0].Index) + stored[addr] += len(inputs) + } + }).Return(nil).Times(writes) + + want := []int64{10, 20, 30, 10} + if failFirst { + want = want[:3] + } + for tick := range want { + // Repository order changes must not give the busy Dave app + // priority over the other consensus types every tick. + apps := []appContracts{quorum, daveApp, authority} + if tick%2 != 0 { + apps = []appContracts{authority, quorum, daveApp} + } + require.Equal(t, !(failFirst && tick == 0), reader.runBlockScanners(t.Context(), apps, 101)) + require.Equal(t, want[:tick+1], order) + require.Nil(t, reader.inputBudget) + } + require.Equal(t, 2, stored[authority.application.IApplicationAddress.Hex()]) + require.Equal(t, 2, stored[quorum.application.IApplicationAddress.Hex()]) + repo.AssertExpectations(t) + source.AssertExpectations(t) + dave.AssertExpectations(t) + }) + } +} + +func TestInputSchedulingHandlesApplicationMembershipChanges(t *testing.T) { + repo := newMockRepository() + source := newMockInputBox() + reader := &Service{repository: repo, maxInputsPerStep: 2} + reader.Logger = testLogger(t) + apps := make(map[int64]appContracts) + var order []int64 + for _, id := range []int64{5, 10, 20, 30, 40} { + app := schedulingApp(id, Consensus_Authority, source) + apps[id] = app + addr := app.application.IApplicationAddress + source.On("GetNumberOfInputs", mock.Anything, addr).Return(big.NewInt(8), nil) + source.On("RetrieveInputs", mock.Anything, []common.Address{addr}, mock.Anything). + Run(func(mock.Arguments) { order = append(order, id) }).Return(schedulingEvents(addr, 0), nil).Once() + repo.On("GetNumberOfInputs", t.Context(), addr.Hex()).Return(uint64(0), nil).Once() + repo.On("GetEpoch", t.Context(), addr.Hex(), uint64(0)).Return((*Epoch)(nil), nil).Once() + repo.On("CreateEpochsAndInputs", t.Context(), addr.Hex(), mock.Anything, uint64(99)).Return(nil).Once() + } + for tick, ids := range [][]int64{ + {30, 10, 20}, {40, 30, 20}, {40, 30}, {5, 40}, {5}, + } { + var targets []appContracts + for _, id := range ids { + targets = append(targets, apps[id]) + } + require.True(t, reader.runBlockScanners(t.Context(), targets, 101)) + require.Equal(t, []int64{10, 20, 30, 40, 5}[:tick+1], order) + } + repo.AssertExpectations(t) + source.AssertExpectations(t) +} diff --git a/internal/evmreader/input_test.go b/internal/evmreader/input_test.go index 48073b816..e28e2883f 100644 --- a/internal/evmreader/input_test.go +++ b/internal/evmreader/input_test.go @@ -120,13 +120,14 @@ func (s *EvmReaderSuite) TestItReadsMultipleInputsFromSingleNewBlock() { mock.Anything, ).Return(new(big.Int).SetUint64(0), nil).Once() + first, second := inputAddedEvent2, inputAddedEvent3 + first.Index, second.Index = big.NewInt(0), big.NewInt(1) // Prepare sequence of inputs s.inputBox.Unset("RetrieveInputs") s.inputBox.On("RetrieveInputs", mock.MatchedBy(func(opts *bind.FilterOpts) bool { return opts.Start == 0x13 }), - mock.Anything, - mock.Anything, - ).Return([]iinputbox.IInputBoxInputAdded{inputAddedEvent2, inputAddedEvent3}, nil) + mock.Anything, mock.Anything, + ).Return([]iinputbox.IInputBoxInputAdded{first, second}, nil) s.inputBox.Unset("GetNumberOfInputs") s.inputBox.On("GetNumberOfInputs", @@ -382,8 +383,7 @@ func (s *EvmReaderSuite) TestCatchUpForeclosedInputsStoresSameBlockInput() { opts.End != nil && *opts.End == app.ForecloseBlock }), - []common.Address{app.IApplicationAddress}, - mock.Anything, + []common.Address{app.IApplicationAddress}, mock.Anything, ).Return([]iinputbox.IInputBoxInputAdded{sameBlockInput}, nil).Once() s.repository.Unset("GetEpoch") diff --git a/internal/evmreader/mocks_test.go b/internal/evmreader/mocks_test.go index 52043dd3c..8e00fc685 100644 --- a/internal/evmreader/mocks_test.go +++ b/internal/evmreader/mocks_test.go @@ -125,20 +125,17 @@ func (m *MockInputBox) SetupDefaultBehavior() *MockInputBox { // RetrieveInputs: matched by Start block, not call order. m.On("RetrieveInputs", mock.MatchedBy(func(opts *bind.FilterOpts) bool { return opts.Start == 0x11 }), - []common.Address{app1Addr}, - mock.Anything, + []common.Address{app1Addr}, []*big.Int(nil), ).Return([]iinputbox.IInputBoxInputAdded{inputAddedEvent0}, nil) m.On("RetrieveInputs", mock.MatchedBy(func(opts *bind.FilterOpts) bool { return opts.Start == 0x12 }), - []common.Address{app1Addr}, - mock.Anything, + []common.Address{app1Addr}, []*big.Int(nil), ).Return([]iinputbox.IInputBoxInputAdded{inputAddedEvent1}, nil) m.On("RetrieveInputs", mock.MatchedBy(func(opts *bind.FilterOpts) bool { return opts.Start == 0x13 }), - []common.Address{app1Addr}, - mock.Anything, + []common.Address{app1Addr}, []*big.Int(nil), ).Return([]iinputbox.IInputBoxInputAdded{inputAddedEvent2, inputAddedEvent3}, nil) // GetNumberOfInputs: block-based matching models the on-chain state. diff --git a/internal/evmreader/openepoch_integrity_test.go b/internal/evmreader/openepoch_integrity_test.go index d2fdf6120..6945daeeb 100644 --- a/internal/evmreader/openepoch_integrity_test.go +++ b/internal/evmreader/openepoch_integrity_test.go @@ -30,6 +30,7 @@ func TestDaveOpenEpochChecksInputCountBeforePublication(t *testing.T) { logIndices []uint64 wantIndices []uint64 mismatch bool + unlimited bool }{ {name: "missing first log", endCount: 2, logIndices: []uint64{1}, mismatch: true}, {name: "missing all logs", endCount: 1, mismatch: true}, @@ -40,11 +41,16 @@ func TestDaveOpenEpochChecksInputCountBeforePublication(t *testing.T) { {name: "no new inputs", storedCount: 2, lowerBound: 2, endCount: 2}, {name: "existing open inputs", storedCount: 4, lowerBound: 2, endCount: 5, logIndices: []uint64{4}, wantIndices: []uint64{4}}, + {name: "rescanned block includes stored open inputs", storedCount: 4, lowerBound: 2, endCount: 5, + logIndices: []uint64{2, 3, 4}, wantIndices: []uint64{4}}, } { t.Run(name+"/"+test.name, func(t *testing.T) { repo := newMockRepository() inputBox := newMockInputBox() - r := &Service{repository: repo} + r := &Service{repository: repo, maxInputsPerStep: 500} + if test.unlimited { + r.maxInputsPerStep = 0 + } r.Logger = testLogger(t) app := newDaveAppContracts(inputBox, nil) app.application.Status = model.ApplicationStatus_OK @@ -74,6 +80,9 @@ func TestDaveOpenEpochChecksInputCountBeforePublication(t *testing.T) { if test.endCount > test.storedCount { events := make([]iinputbox.IInputBoxInputAdded, 0, len(test.logIndices)) for _, index := range test.logIndices { + if index < test.storedCount || index >= test.endCount { + continue + } events = append(events, makeInputEvent(app1Addr, index, head)) } inputBox.On("RetrieveInputs", mock.MatchedBy(func(opts *bind.FilterOpts) bool { diff --git a/internal/evmreader/output.go b/internal/evmreader/output.go index 488cf539c..91de79810 100644 --- a/internal/evmreader/output.go +++ b/internal/evmreader/output.go @@ -346,7 +346,7 @@ func (r *Service) readOutputExecutionsFromBlockChain( var executedOutputs []*iapplication.IApplicationOutputExecuted // Define onHit function that accumulates executed outputs at transition blocks - onHit := func(block uint64) error { + onHit := func(block uint64, _, _ *big.Int) error { filterOpts := &bind.FilterOpts{ Context: ctx, Start: block, @@ -368,7 +368,7 @@ func (r *Service) readOutputExecutionsFromBlockChain( prevValue.SetUint64(execCount) // Use FindTransitions to find blocks where outputs were executed - _, err = ethutil.FindTransitions(ctx, startBlock, endBlock, prevValue, oracle, onHit) + _, _, err = ethutil.FindTransitions(ctx, startBlock, endBlock, prevValue, nil, oracle, onHit) if err != nil { return nil, fmt.Errorf("failed to walk output execution transitions: %w", err) } diff --git a/internal/evmreader/output_test.go b/internal/evmreader/output_test.go index b200c6159..f1d4bdab8 100644 --- a/internal/evmreader/output_test.go +++ b/internal/evmreader/output_test.go @@ -733,11 +733,12 @@ func (s *EvmReaderSuite) setupOutputMismatchTest() { s.contractFactory = newMockAdapterFactory() s.evmReader = &Service{ - client: s.client, - repository: s.repository, - defaultBlock: DefaultBlock_Latest, - adapterFactory: s.contractFactory, - hasEnabledApps: true, + client: s.client, + repository: s.repository, + defaultBlock: DefaultBlock_Latest, + adapterFactory: s.contractFactory, + hasEnabledApps: true, + maxInputsPerStep: 500, } logLevel, err := config.GetLogLevel() @@ -866,8 +867,7 @@ func (s *EvmReaderSuite) setupOutputMismatchTest() { s.inputBox.On("RetrieveInputs", mock.MatchedBy(func(opts *bind.FilterOpts) bool { return opts.Start == 0x11 }), - []common.Address{app1Addr}, - mock.Anything, + []common.Address{app1Addr}, mock.Anything, ).Return([]iinputbox.IInputBoxInputAdded{inputAddedEvent0}, nil) // On-chain: 0 inputs before block 0x11, 1 from block 0x11 diff --git a/internal/evmreader/post_foreclosure_withdrawal.go b/internal/evmreader/post_foreclosure_withdrawal.go index dea448c0d..41bde767f 100644 --- a/internal/evmreader/post_foreclosure_withdrawal.go +++ b/internal/evmreader/post_foreclosure_withdrawal.go @@ -67,7 +67,7 @@ func (r *Service) checkForPostForeclosureWithdrawals( var withdrawals []*Withdrawal - onHit := func(block uint64) error { + onHit := func(block uint64, _, _ *big.Int) error { blockWithdrawals, err := r.withdrawalsAtBlock(ctx, app, block) if err != nil { return err @@ -89,11 +89,12 @@ func (r *Service) checkForPostForeclosureWithdrawals( return false } - _, err = ethutil.FindTransitions( + _, _, err = ethutil.FindTransitions( ctx, startBlock, mostRecentBlockNumber, prevValue, + nil, query, onHit, ) diff --git a/internal/evmreader/readiness_test.go b/internal/evmreader/readiness_test.go index abb9b0080..542d84883 100644 --- a/internal/evmreader/readiness_test.go +++ b/internal/evmreader/readiness_test.go @@ -111,7 +111,7 @@ func TestReadinessTracksScanFailures(t *testing.T) { repo.On("UpdateApplicationLastForecloseCheckBlock", mock.Anything, mock.Anything, mock.Anything).Return(nil) } r := &Service{client: client, repository: repo, - defaultBlock: DefaultBlock_Latest, readyMaxStaleness: time.Hour} + defaultBlock: DefaultBlock_Latest, readyMaxStaleness: time.Hour, maxInputsPerStep: 500} r.Logger = slog.Default() r.resolver = newApplicationAdapterResolver(r.Logger, newMockAdapterFactory().SetupDefaultBehaviorSingleApp(contract, input)) require.False(t, r.Ready()) @@ -194,7 +194,7 @@ func TestReadinessSeparatesInputCorruptionFromSharedFailures(t *testing.T) { client := newMockEthClient() client.On("HeaderByNumber", mock.Anything, mock.Anything).Return(&types.Header{Number: big.NewInt(110)}, nil) r := &Service{client: client, repository: repo, - defaultBlock: DefaultBlock_Latest, readyMaxStaleness: time.Hour} + defaultBlock: DefaultBlock_Latest, readyMaxStaleness: time.Hour, maxInputsPerStep: 500} r.Logger = slog.Default() r.resolver = newApplicationAdapterResolver(r.Logger, newMockAdapterFactory().SetupDefaultBehaviorSingleApp(newMockApplicationContract(), input)) diff --git a/internal/evmreader/sealedepochs.go b/internal/evmreader/sealedepochs.go index 51fd69c38..b1e0f2d42 100644 --- a/internal/evmreader/sealedepochs.go +++ b/internal/evmreader/sealedepochs.go @@ -7,7 +7,6 @@ import ( "context" "errors" "fmt" - "math" "math/big" . "github.com/cartesi/rollups-node/internal/model" @@ -72,6 +71,9 @@ func (r *Service) scanDaveConsensusEpochsAndInputs( // Process each application individually since each has its own DaveConsensus contract for _, app := range applications { + if r.inputBudgetExhausted() { + break + } if app.inputSource == nil { success = false r.Logger.Error("Cannot scan DaveConsensus epochs: InputBox adapter is missing", @@ -91,6 +93,11 @@ func (r *Service) scanDaveConsensusEpochsAndInputs( if errors.Is(err, context.Canceled) { return false // shutting down } + if errors.Is(err, ethutil.ErrIncompleteProcessing) { + // This epoch has a known seal boundary. Finish its batches before + // the open scan can ingest inputs belonging to a later epoch. + continue + } if errors.Is(err, errContractNotDeployedAtBlock) { // Registration can precede deployment visibility at the configured // block. Neither sealed nor open epochs can be observed yet. @@ -223,26 +230,28 @@ func (r *Service) processApplicationSealedEpochs( return err } } - previousTransitionValue := prevValue - onHit := func(block uint64) error { - value, err := r.processEpochTransition(ctx, app, block, previousTransitionValue) - if err == nil { - previousTransitionValue = value - } - return err + onHit := func(block uint64, prevValue, _ *big.Int) error { + return r.processEpochTransition(ctx, app, block, prevValue) } // Use FindTransitions to find epoch transitions - _, err := ethutil.FindTransitions(ctx, nextSearchBlock, mostRecentBlockNumber, prevValue, oracle, onHit) - if err != nil { + _, lastProcessedBlock, err := ethutil.FindTransitions(ctx, nextSearchBlock, mostRecentBlockNumber, prevValue, nil, oracle, onHit) + var incomplete bool + switch { + case errors.Is(err, ethutil.ErrIncompleteProcessing): + incomplete = true + case err != nil: return fmt.Errorf("failed to walk epoch transitions: %w", err) } // Update the last check block for this application - err = r.repository.UpdateEventLastCheckBlock(ctx, []int64{app.application.ID}, MonitoredEvent_EpochSealed, mostRecentBlockNumber) + err = r.repository.UpdateEventLastCheckBlock(ctx, []int64{app.application.ID}, MonitoredEvent_EpochSealed, lastProcessedBlock) if err != nil { return fmt.Errorf("failed to update last epoch check block: %w", err) } + if incomplete { + return ethutil.ErrIncompleteProcessing + } if initializing { r.Logger.Info("Application sealed epoch sync initialized", "application", app.application.Name, "consensus_address", app.application.IConsensusAddress, @@ -260,7 +269,7 @@ func (r *Service) processEpochTransition( app appContracts, transitionBlock uint64, previousEpochNumber *big.Int, -) (*big.Int, error) { +) error { r.Logger.Debug("Processing epoch transition", "application", app.application.Name, "block", transitionBlock) // Get the sealed epoch information at this block @@ -271,7 +280,7 @@ func (r *Service) processEpochTransition( sealedEpoch, err := app.daveConsensus.GetCurrentSealedEpoch(callOpts) if err != nil { - return nil, fmt.Errorf("failed to get sealed epoch at transition block %d: %w", transitionBlock, err) + return fmt.Errorf("failed to get sealed epoch at transition block %d: %w", transitionBlock, err) } // Retrieve the actual EpochSealed events for this transition @@ -283,17 +292,17 @@ func (r *Service) processEpochTransition( sealedEvents, err := app.daveConsensus.RetrieveSealedEpochs(filterOpts) if err != nil { - return nil, fmt.Errorf("failed to retrieve sealed epoch events at block %d: %w", transitionBlock, err) + return fmt.Errorf("failed to retrieve sealed epoch events at block %d: %w", transitionBlock, err) } // Validate the whole block before storing any event. Several epochs can seal // in one block, so a nonempty page can still omit part of the transition. if sealedEpoch.EpochNumber == nil || !sealedEpoch.EpochNumber.IsUint64() { - return nil, fmt.Errorf("invalid sealed epoch number at block %d: %v", transitionBlock, sealedEpoch.EpochNumber) + return fmt.Errorf("invalid sealed epoch number at block %d: %v", transitionBlock, sealedEpoch.EpochNumber) } expectedCount := new(big.Int).Sub(sealedEpoch.EpochNumber, previousEpochNumber) if expectedCount.Sign() <= 0 || !expectedCount.IsUint64() || expectedCount.Uint64() != uint64(len(sealedEvents)) { - return nil, fmt.Errorf("%w: sealed epoch event count mismatch at block %d: expected %s, got %d", + return fmt.Errorf("%w: sealed epoch event count mismatch at block %d: expected %s, got %d", errInconclusiveEpochView, transitionBlock, expectedCount, len(sealedEvents)) } expectedEpoch := new(big.Int).Set(previousEpochNumber) @@ -301,14 +310,14 @@ func (r *Service) processEpochTransition( expectedEpoch.Add(expectedEpoch, big.NewInt(1)) if event == nil || event.EpochNumber == nil || event.EpochNumber.Cmp(expectedEpoch) != 0 || event.Raw.BlockNumber != transitionBlock { - return nil, fmt.Errorf("%w: invalid sealed epoch event sequence at block %d: expected epoch %s", + return fmt.Errorf("%w: invalid sealed epoch event sequence at block %d: expected epoch %s", errInconclusiveEpochView, transitionBlock, expectedEpoch) } } lastEvent := sealedEvents[len(sealedEvents)-1] if lastEvent.InputIndexLowerBound.Cmp(sealedEpoch.InputIndexLowerBound) != 0 || lastEvent.InputIndexUpperBound.Cmp(sealedEpoch.InputIndexUpperBound) != 0 || lastEvent.Tournament != sealedEpoch.Tournament { - return nil, fmt.Errorf("%w: last sealed epoch event does not match current sealed epoch at block %d", + return fmt.Errorf("%w: last sealed epoch event does not match current sealed epoch at block %d", errInconclusiveEpochView, transitionBlock) } r.Logger.Info("Found sealed epoch events", @@ -319,11 +328,11 @@ func (r *Service) processEpochTransition( for _, event := range sealedEvents { err := r.processSealedEpochEvent(ctx, app, event) if err != nil { - return nil, fmt.Errorf("failed to process sealed epoch event at block %d: %w", transitionBlock, err) + return fmt.Errorf("failed to process sealed epoch event at block %d: %w", transitionBlock, err) } } - return sealedEpoch.EpochNumber, nil + return nil } func (r *Service) processSealedEpochEvent( @@ -331,6 +340,9 @@ func (r *Service) processSealedEpochEvent( app appContracts, event *idaveconsensus.IDaveConsensusEpochSealed, ) error { + if r.inputBudgetExhausted() { + return ethutil.ErrIncompleteProcessing + } r.Logger.Debug("Processing sealed epoch event", "epoch_number", event.EpochNumber, "input_lower_bound", event.InputIndexLowerBound, @@ -377,17 +389,23 @@ func (r *Service) processSealedEpochEvent( return fmt.Errorf("failed to fetch epoch %d: %w", epochNumber, err) } + var unprocessedFirstBlock uint64 + var unprocessedIndexLowerBound uint64 if epoch == nil { // Create new epoch from sealed event epoch = &Epoch{ Index: event.EpochNumber.Uint64(), - FirstBlock: firstBlock, // Will be calculated based on epoch length + FirstBlock: firstBlock, LastBlock: event.Raw.BlockNumber, InputIndexLowerBound: event.InputIndexLowerBound.Uint64(), InputIndexUpperBound: event.InputIndexUpperBound.Uint64(), TournamentAddress: &event.Tournament, - Status: EpochStatus_Closed, // Sealed epochs are closed + Status: EpochStatus_Open, // Sealed epochs that are not fully processed are still open } + + // setup the entire epoch as unprocessed + unprocessedFirstBlock = epoch.FirstBlock + unprocessedIndexLowerBound = epoch.InputIndexLowerBound } else { if epoch.FirstBlock != firstBlock || epoch.InputIndexLowerBound != event.InputIndexLowerBound.Uint64() { return fmt.Errorf("epoch %d data mismatch with sealed event", epoch.Index) @@ -401,40 +419,57 @@ func (r *Service) processSealedEpochEvent( // claim/proof progress when replaying a partially completed scan. return nil } + + // LastBlock may contain only a prefix of its inputs. Revisit it with + // the saved index bound, which excludes inputs already stored. + unprocessedFirstBlock = epoch.LastBlock + unprocessedIndexLowerBound = epoch.InputIndexUpperBound + epoch.LastBlock = event.Raw.BlockNumber epoch.InputIndexUpperBound = event.InputIndexUpperBound.Uint64() epoch.TournamentAddress = &event.Tournament - epoch.Status = EpochStatus_Closed // Sealed epochs are closed } + unprocessedIndexUpperBound := min(epoch.InputIndexUpperBound, r.inputBatchUpperBound(unprocessedIndexLowerBound)) + // Fetch inputs for this epoch from the InputBox. - // Always search from epoch.FirstBlock (not lastInputCheckBlock+1) because with + // Always search from unprocessedFirstBlock (not lastInputCheckBlock+1) because with // PRT's overlapping block boundaries (sealed epoch's LastBlock = next epoch's // FirstBlock), inputs for this epoch may exist at the overlap block — added in // a later transaction than the previous epoch's seal within the same block. - // Use InputIndexLowerBound as prevValue rather than the DB input count, since - // the open epoch may have already stored some of these inputs ahead of us. var inputs []*Input - if epoch.InputIndexUpperBound > epoch.InputIndexLowerBound { + if unprocessedIndexLowerBound < unprocessedIndexUpperBound { var err error - prevValue := new(big.Int).SetUint64(epoch.InputIndexLowerBound) - inputs, _, err = r.fetchInputs(ctx, app, - epoch.FirstBlock, epoch.LastBlock, - prevValue, - epoch.InputIndexLowerBound, epoch.InputIndexUpperBound) - if err != nil { + var lastInputBlock uint64 + inputs, lastInputBlock, _, err = r.fetchInputs(ctx, app, + unprocessedFirstBlock, epoch.LastBlock, + new(big.Int).SetUint64(unprocessedIndexLowerBound), + unprocessedIndexLowerBound, unprocessedIndexUpperBound) + if err != nil && !errors.Is(err, ethutil.ErrIncompleteProcessing) { return fmt.Errorf("failed to fetch inputs for epoch %d: %w", epoch.Index, err) } - expectedCount := epoch.InputIndexUpperBound - epoch.InputIndexLowerBound - if uint64(len(inputs)) != expectedCount { + actualCount := uint64(len(inputs)) + expectedCount := unprocessedIndexUpperBound - unprocessedIndexLowerBound + if actualCount != expectedCount { return fmt.Errorf( "%w: epoch %d input count mismatch: expected %d (indices %d to %d), got %d", errInconclusiveEpochView, epoch.Index, expectedCount, - epoch.InputIndexLowerBound, epoch.InputIndexUpperBound, - len(inputs)) + unprocessedIndexLowerBound, epoch.InputIndexUpperBound, + actualCount) } + + // Epoch bounds include observed inputs, even in a partial block. The + // application input cursor separately covers only complete blocks. + epoch.LastBlock = inputObservationLastBlock(epoch.FirstBlock, lastInputBlock, inputs) + epoch.InputIndexUpperBound = unprocessedIndexUpperBound } + + if unprocessedIndexUpperBound == event.InputIndexUpperBound.Uint64() { + epoch.LastBlock = event.Raw.BlockNumber + epoch.Status = EpochStatus_Closed // Fully processed sealed epochs are closed + } + // The seal covers every input before its transaction, but later inputs in // the same block belong to the next epoch. Publish only complete blocks; // the open-epoch scan must finish this boundary block before drain is safe. @@ -446,7 +481,7 @@ func (r *Service) processSealedEpochEvent( ctx, app.application.IApplicationAddress.String(), epochInputMap, - event.Raw.BlockNumber-1, + epoch.LastBlock-1, // last block must be rescanned for inputs for the next epoch ) if err != nil { if errors.Is(err, repository.ErrInputLogIdentityConflict) { @@ -456,6 +491,10 @@ func (r *Service) processSealedEpochEvent( return fmt.Errorf("failed to store epoch and inputs: %w", err) } + if epoch.Status != EpochStatus_Closed { + return ethutil.ErrIncompleteProcessing + } + r.Logger.Debug("Stored sealed epoch and inputs", "application", app.application.Name, "epoch_number", epoch.Index, @@ -470,6 +509,9 @@ func (r *Service) processApplicationOpenEpoch( app appContracts, mostRecentBlockNumber uint64, ) error { + if r.inputBudgetExhausted() { + return nil + } // The input cursor covers complete blocks. A sealed batch in this tick // can advance it only to the block before its seal. if mostRecentBlockNumber < app.application.LastInputCheckBlock { @@ -509,7 +551,6 @@ func (r *Service) processApplicationOpenEpoch( return fmt.Errorf("failed to fetch epoch %d: %w", nextEpochNumber, err) } if openEpoch == nil { - // Create epoch from sealed event openEpoch = &Epoch{ Index: nextEpochNumber, FirstBlock: lastEpoch.LastBlock, @@ -535,11 +576,10 @@ func (r *Service) processApplicationOpenEpoch( return fmt.Errorf( "failed to get number of inputs from repository: %w", err) } - prevValue := new(big.Int).SetUint64(inputCount) - inputs, endCount, err := r.fetchInputs(ctx, app, + inputs, lastInputBlock, endCount, err := r.fetchInputs(ctx, app, lastInputCheckBlock+1, mostRecentBlockNumber, - prevValue, - openEpoch.InputIndexLowerBound, math.MaxUint64) + new(big.Int).SetUint64(inputCount), + openEpoch.InputIndexUpperBound, r.inputBatchUpperBound(openEpoch.InputIndexUpperBound)) if err != nil { return fmt.Errorf("failed to fetch inputs for epoch %d: %w", openEpoch.Index, err) } @@ -548,9 +588,9 @@ func (r *Service) processApplicationOpenEpoch( errInconclusiveInputView, openEpoch.Index, inputCount, endCount, len(inputs)) } - // increase the upper bound according to the number of fetched inputs + // increase the upper bound according to the number of fetched inputs. openEpoch.InputIndexUpperBound += uint64(len(inputs)) - openEpoch.LastBlock = mostRecentBlockNumber + openEpoch.LastBlock = inputObservationLastBlock(openEpoch.FirstBlock, lastInputBlock, inputs) r.Logger.Debug("Storing open epoch", "application", app.application.Name, @@ -564,7 +604,7 @@ func (r *Service) processApplicationOpenEpoch( ctx, app.application.IApplicationAddress.String(), epochInputMap, - mostRecentBlockNumber, + lastInputBlock, ) if err != nil { @@ -583,3 +623,13 @@ func (r *Service) processApplicationOpenEpoch( return nil } + +// inputObservationLastBlock keeps epoch bounds valid when the first observed +// block is only partially ingested and lastCompleteBlock precedes firstBlock. +func inputObservationLastBlock(firstBlock, lastCompleteBlock uint64, inputs []*Input) uint64 { + lastBlock := max(firstBlock, lastCompleteBlock) + for _, input := range inputs { + lastBlock = max(lastBlock, input.BlockNumber) + } + return lastBlock +} diff --git a/internal/evmreader/sealedepochs_checkpoint_test.go b/internal/evmreader/sealedepochs_checkpoint_test.go index 38a4472df..82f17d494 100644 --- a/internal/evmreader/sealedepochs_checkpoint_test.go +++ b/internal/evmreader/sealedepochs_checkpoint_test.go @@ -93,10 +93,16 @@ func TestDaveInputCheckpointAfterFailedOpenScan(t *testing.T) { inputBlock := mock.MatchedBy(func(opts *bind.FilterOpts) bool { return opts.Start == forecloseBlock && opts.End != nil && *opts.End == forecloseBlock }) - inputBox.On("RetrieveInputs", inputBlock, []common.Address{app.IApplicationAddress}, mock.Anything). - Return(inputEvents, nil) + for _, event := range events { + lower, upper := event.InputIndexLowerBound.Uint64(), event.InputIndexUpperBound.Uint64() + if lower == upper { + continue + } + inputBox.On("RetrieveInputs", inputBlock, []common.Address{app.IApplicationAddress}, mock.Anything). + Return(inputEvents[lower:upper], nil).Once() + } contracts := appContracts{application: app, inputSource: inputBox, daveConsensus: dave} - reader := &Service{repository: repo} + reader := &Service{repository: repo, maxInputsPerStep: 500} reader.Logger = slog.New(slog.NewTextHandler(io.Discard, nil)) require.NoError(t, reader.processApplicationSealedEpochs(ctx, contracts, forecloseBlock)) @@ -107,7 +113,7 @@ func TestDaveInputCheckpointAfterFailedOpenScan(t *testing.T) { t.Cleanup(cancel) failedRead := inputBox.On("RetrieveInputs", inputBlock, []common.Address{app.IApplicationAddress}, mock.Anything).Once() if test.cancelWrite { - failedRead.Run(func(mock.Arguments) { cancel() }).Return(inputEvents, nil) + failedRead.Run(func(mock.Arguments) { cancel() }).Return(inputEvents[sealedInputs:], nil) } else { failedRead.Return([]iinputbox.IInputBoxInputAdded{}, context.Canceled) } @@ -135,9 +141,9 @@ func TestDaveInputCheckpointAfterFailedOpenScan(t *testing.T) { inputBox.Unset("RetrieveInputs") inputBox.On("RetrieveInputs", inputBlock, []common.Address{app.IApplicationAddress}, mock.Anything). - Return(inputEvents, nil).Once() + Return(inputEvents[sealedInputs:], nil).Once() contracts.application = app - reader = &Service{repository: repo} + reader = &Service{repository: repo, maxInputsPerStep: 500} reader.Logger = slog.New(slog.NewTextHandler(io.Discard, nil)) reader.scanDaveConsensusEpochsAndInputs(ctx, []appContracts{contracts}, forecloseBlock+10) diff --git a/internal/evmreader/sealedepochs_cursor_test.go b/internal/evmreader/sealedepochs_cursor_test.go index 70f11773e..c79bdd4b6 100644 --- a/internal/evmreader/sealedepochs_cursor_test.go +++ b/internal/evmreader/sealedepochs_cursor_test.go @@ -59,7 +59,7 @@ func TestForeclosureInputDrainRunsBelowSealedCursor(t *testing.T) { repo := newMockRepository() inputBox := newMockInputBox() dave := newMockDaveConsensus() - r := &Service{repository: repo} + r := &Service{repository: repo, maxInputsPerStep: 500} r.Logger = testLogger(t) app := newDaveAppContracts(inputBox, dave) app.application.Status = model.ApplicationStatus_MachineHalted diff --git a/internal/evmreader/sealedepochs_identity_test.go b/internal/evmreader/sealedepochs_identity_test.go index b2239fcad..0c21bdc9f 100644 --- a/internal/evmreader/sealedepochs_identity_test.go +++ b/internal/evmreader/sealedepochs_identity_test.go @@ -43,7 +43,7 @@ func TestDaveInputStoreErrorClassification(t *testing.T) { repo := newMockRepository() inputBox := newMockInputBox() dave := newMockDaveConsensus() - r := &Service{repository: repo} + r := &Service{repository: repo, maxInputsPerStep: 500} r.Logger = testLogger(t) app := appContracts{ application: &model.Application{ diff --git a/internal/evmreader/sealedepochs_incremental_test.go b/internal/evmreader/sealedepochs_incremental_test.go new file mode 100644 index 000000000..91e974e3b --- /dev/null +++ b/internal/evmreader/sealedepochs_incremental_test.go @@ -0,0 +1,106 @@ +// (c) Cartesi and individual authors (see AUTHORS) +// SPDX-License-Identifier: Apache-2.0 (see LICENSE) + +package evmreader + +import ( + "math/big" + "testing" + + . "github.com/cartesi/rollups-node/internal/model" + "github.com/cartesi/rollups-node/pkg/contracts/iinputbox" + "github.com/ethereum/go-ethereum/accounts/abi/bind" + "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +func TestSealedEpochFetchesOnlyUnstoredInputs(t *testing.T) { + const sealBlock uint64 = 101 + for _, test := range []struct { + name string + upper uint64 + indices []uint64 + mismatch bool + unlimited bool + }{ + {name: "first unstored index is included", upper: 6, indices: []uint64{5}}, + {name: "several new inputs and next epoch input", upper: 8, indices: []uint64{5, 6, 7, 8}}, + {name: "all inputs already stored", upper: 5}, + {name: "missing input prevents sealing", upper: 7, indices: []uint64{6}, mismatch: true}, + {name: "zero limit includes all unstored inputs", upper: 8, indices: []uint64{5, 6, 7, 8}, unlimited: true}, + } { + t.Run(test.name, func(t *testing.T) { + repo := newMockRepository() + inputBox := newMockInputBox() + reader := &Service{repository: repo, maxInputsPerStep: 500} + if test.unlimited { + reader.maxInputsPerStep = 0 + } + reader.Logger = testLogger(t) + app := newDaveAppContracts(inputBox, nil) + previous := &Epoch{Index: 0, LastBlock: 50, InputIndexUpperBound: 3, Status: EpochStatus_Closed} + // Inputs 3 and 4 were stored while this epoch was open. Rechecking + // the last observed block must not fetch those inputs again. + open := &Epoch{Index: 1, FirstBlock: 50, LastBlock: 100, + InputIndexLowerBound: 3, InputIndexUpperBound: 5, Status: EpochStatus_Open} + event := makeSealedEpochEvent(1, 3, test.upper, sealBlock, common.HexToAddress("0xaaaa")) + repo.On("GetEpoch", t.Context(), app1Addr.Hex(), uint64(0)).Return(previous, nil).Once() + repo.On("UpdateEpochClaimTransactionHash", t.Context(), app1Addr.Hex(), previous).Return(nil).Once() + repo.On("GetEpoch", t.Context(), app1Addr.Hex(), uint64(1)).Return(open, nil).Once() + + if test.upper > 5 { + inputBox.On("GetNumberOfInputs", mock.MatchedBy(func(opts *bind.CallOpts) bool { + return opts.BlockNumber.Uint64() == 100 + }), app1Addr).Return(big.NewInt(5), nil).Twice() + count := test.upper + for _, index := range test.indices { + count = max(count, index+1) + } + // Only the last observed block and the unscanned block may be + // queried. Input logs must be fetched only from the latter. + inputBox.On("GetNumberOfInputs", mock.MatchedBy(func(opts *bind.CallOpts) bool { + return opts.BlockNumber.Uint64() == sealBlock + }), app1Addr).Return(new(big.Int).SetUint64(count), nil).Once() + events := make([]iinputbox.IInputBoxInputAdded, 0, len(test.indices)) + for _, index := range test.indices { + if index < 5 || index >= test.upper { + continue + } + events = append(events, makeInputEvent(app1Addr, index, sealBlock)) + } + inputBox.On("RetrieveInputs", mock.MatchedBy(func(opts *bind.FilterOpts) bool { + return opts.Start == sealBlock && opts.End != nil && *opts.End == sealBlock + }), []common.Address{app1Addr}, mock.Anything).Return(events, nil).Once() + } + if !test.mismatch { + repo.On("CreateEpochsAndInputs", t.Context(), app1Addr.Hex(), mock.Anything, sealBlock-1). + Run(func(args mock.Arguments) { + epochs := args.Get(2).(map[*Epoch][]*Input) + require.Len(t, epochs, 1) + for epoch, inputs := range epochs { + require.Equal(t, EpochStatus_Closed, epoch.Status) + require.Equal(t, uint64(50), epoch.FirstBlock) + require.Equal(t, sealBlock, epoch.LastBlock) + require.Equal(t, uint64(3), epoch.InputIndexLowerBound) + require.Equal(t, test.upper, epoch.InputIndexUpperBound) + require.Equal(t, event.Tournament, *epoch.TournamentAddress) + require.Len(t, inputs, int(test.upper-5)) + for i, input := range inputs { + require.Equal(t, uint64(5+i), input.Index) + } + } + }).Return(nil).Once() + } + err := reader.processSealedEpochEvent(t.Context(), app, event) + if test.mismatch { + require.ErrorContains(t, err, "input count mismatch") + repo.AssertNotCalled(t, "CreateEpochsAndInputs", mock.Anything, mock.Anything, mock.Anything, mock.Anything) + } else { + require.NoError(t, err) + } + repo.AssertExpectations(t) + inputBox.AssertExpectations(t) + }) + } +} diff --git a/internal/evmreader/sealedepochs_test.go b/internal/evmreader/sealedepochs_test.go index 8818e5be8..0c08e5843 100644 --- a/internal/evmreader/sealedepochs_test.go +++ b/internal/evmreader/sealedepochs_test.go @@ -51,9 +51,10 @@ func (s *SealedEpochsSuite) SetupTest() { s.dave = newMockDaveConsensus() s.evmReader = &Service{ - repository: s.repository, - defaultBlock: DefaultBlock_Latest, - hasEnabledApps: true, + repository: s.repository, + defaultBlock: DefaultBlock_Latest, + hasEnabledApps: true, + maxInputsPerStep: 500, } logLevel, err := config.GetLogLevel() @@ -141,19 +142,9 @@ func (s *SealedEpochsSuite) TestProcessSealedEpochFindsInputAtOverlapBlock() { mock.MatchedBy(func(opts *bind.FilterOpts) bool { return opts.Start == sealBlock0 }), - mock.Anything, - mock.Anything, + mock.Anything, mock.Anything, ).Return([]iinputbox.IInputBoxInputAdded{overlapInput}, nil) - // No inputs at other blocks in the range. - s.inputBox.On("RetrieveInputs", - mock.MatchedBy(func(opts *bind.FilterOpts) bool { - return opts.Start != sealBlock0 - }), - mock.Anything, - mock.Anything, - ).Return([]iinputbox.IInputBoxInputAdded{}, nil) - // CreateEpochsAndInputs captures what was stored. var storedInputs []*Input s.repository.On("CreateEpochsAndInputs", diff --git a/internal/evmreader/service.go b/internal/evmreader/service.go index 54f1dde0c..f2da2a12c 100644 --- a/internal/evmreader/service.go +++ b/internal/evmreader/service.go @@ -5,6 +5,7 @@ package evmreader import ( "context" + "errors" "fmt" "log/slog" "math/big" @@ -40,6 +41,11 @@ type Service struct { lastSuccessfulPoll atomic.Pointer[time.Time] consecutiveScanFailures atomic.Uint32 readyMaxStaleness time.Duration + maxInputsPerStep uint64 + // inputBudget is scoped to runBlockScanners; nil means no shared budget. + inputBudget *uint64 + // The database ID survives changes in application order and membership. + lastInputApplicationID int64 } const EvmReaderConfigKey = "evm-reader" @@ -52,6 +58,13 @@ func Create(ctx context.Context, c *CreateInfo) (service.SupervisedService, erro return nil, err // This returns context.Canceled or context.DeadlineExceeded. } + if c.Config.EvmReaderMaxInputsPerStep <= 0 { + return nil, errors.New("max input per step must be at least 1") + } + if c.Config.EvmReaderMaxInputsPerStep > 1000 { + return nil, errors.New("max input per step must be at most 1000") + } + readyMaxStaleness, err := readinessBudget(&c.Config) if err != nil { return nil, err @@ -102,7 +115,7 @@ func Create(ctx context.Context, c *CreateInfo) (service.SupervisedService, erro s.repository = c.Repository if s.repository == nil { - return nil, fmt.Errorf("repository on evmreader service Create is nil") + return nil, errors.New("repository on evmreader service Create is nil") } nodeConfig, err := s.setupPersistentConfig(ctx, &c.Config) @@ -124,7 +137,7 @@ func Create(ctx context.Context, c *CreateInfo) (service.SupervisedService, erro } else { fullEthClient, ok := ethClient.(*ethclient.Client) if !ok { - return nil, fmt.Errorf("EthClient must be *ethclient.Client when AdapterFactory is not provided") + return nil, errors.New("EthClient must be *ethclient.Client when AdapterFactory is not provided") } s.adapterFactory = &DefaultAdapterFactory{ Client: fullEthClient, @@ -137,6 +150,7 @@ func Create(ctx context.Context, c *CreateInfo) (service.SupervisedService, erro s.resolver = newApplicationAdapterResolver(s.Logger, s.adapterFactory) s.readyMaxStaleness = readyMaxStaleness + s.maxInputsPerStep = c.Config.EvmReaderMaxInputsPerStep s.lastSuccessfulPoll.Store(&time.Time{}) s.Logger.Info("Created", "config", c.Config) @@ -163,7 +177,7 @@ func (s *Service) setupPersistentConfig( // readinessBudget is independent of the HTTP retry-backoff cap. func readinessBudget(c *config.EvmreaderConfig) (time.Duration, error) { if c.EvmReaderReadyMaxStaleness < 0 { - return 0, fmt.Errorf("CARTESI_EVM_READER_READY_MAX_STALENESS must be non-negative") + return 0, errors.New("CARTESI_EVM_READER_READY_MAX_STALENESS must be non-negative") } if c.EvmReaderReadyMaxStaleness > 0 { return c.EvmReaderReadyMaxStaleness, nil diff --git a/internal/evmreader/service_config_test.go b/internal/evmreader/service_config_test.go index 083c89d44..417424803 100644 --- a/internal/evmreader/service_config_test.go +++ b/internal/evmreader/service_config_test.go @@ -53,18 +53,20 @@ func TestCreateAcceptsRequestTimeoutBelowPollingInterval(t *testing.T) { repo.On("RegisterObserver", mock.Anything, EvmReaderConfigKey, requested). Return(requested, nil).Once() - _, err = Create(t.Context(), &CreateInfo{ + created, err := Create(t.Context(), &CreateInfo{ Config: config.EvmreaderConfig{ LogLevel: logLevel, BlockchainDefaultBlock: DefaultBlock_Finalized, BlockchainHttpRequestTimeout: requestTimeout, BlockchainId: chainID, EvmReaderPollingInterval: pollInterval, + EvmReaderMaxInputsPerStep: 17, }, EthClient: client, Repository: repo, }) require.NoError(t, err) + require.Equal(t, uint64(17), created.(*Service).maxInputsPerStep) repo.AssertExpectations(t) } @@ -213,10 +215,36 @@ func TestReadinessBudget(t *testing.T) { func TestCreateRejectsNegativeReadinessBudget(t *testing.T) { _, err := Create(t.Context(), &CreateInfo{Config: config.EvmreaderConfig{ EvmReaderReadyMaxStaleness: -time.Second, + EvmReaderMaxInputsPerStep: 500, }}) require.ErrorContains(t, err, "CARTESI_EVM_READER_READY_MAX_STALENESS must be non-negative") } +func TestCreateValidatesInputBudgetBounds(t *testing.T) { + for _, tc := range []struct { + name string + limit uint64 + want string + }{ + {"zero", 0, "max input per step must be at least 1"}, + {"minimum", 1, "CARTESI_EVM_READER_READY_MAX_STALENESS must be non-negative"}, + {"maximum", 1000, "CARTESI_EVM_READER_READY_MAX_STALENESS must be non-negative"}, + {"above maximum", 1001, "max input per step must be at most 1000"}, + {"largest uint64", ^uint64(0), "max input per step must be at most 1000"}, + } { + t.Run(tc.name, func(t *testing.T) { + // A negative readiness budget stops initialization before external + // dependencies, proving valid input bounds reach the next validation. + created, err := Create(t.Context(), &CreateInfo{Config: config.EvmreaderConfig{ + EvmReaderMaxInputsPerStep: tc.limit, + EvmReaderReadyMaxStaleness: -time.Second, + }}) + require.Nil(t, created) + require.ErrorContains(t, err, tc.want) + }) + } +} + func (s *EvmReaderSuite) TestReadinessRefreshesAfterScan() { s.client.EnqueueNewHead(100).Once() s.repository.On("ListApplications", mock.Anything, mock.Anything, mock.Anything).Unset() diff --git a/pkg/ethutil/search.go b/pkg/ethutil/search.go index 7359c5075..4c27ea294 100644 --- a/pkg/ethutil/search.go +++ b/pkg/ethutil/search.go @@ -8,7 +8,6 @@ import ( "errors" "fmt" "math/big" - "slices" ) // ErrMonotonicViolation is returned by FindTransitions when the seed prevValue @@ -18,77 +17,87 @@ import ( // Callers can match it with errors.Is to distinguish this corruption signal // from transient query failures. var ErrMonotonicViolation = errors.New("monotonic assumption violated") +var ErrIncompleteProcessing = errors.New("incomplete processing of transitions") // TransitionQueryFn for binary search type TransitionQueryFn func(ctx context.Context, block uint64) (*big.Int, error) -type OnHitFn func(block uint64) error +type OnHitFn func(block uint64, prevValue, currValue *big.Int) error -// sortAndCallOnHit sorts the transition blocks and calls onHit for each in chronological order -func sortAndCallOnHit(transitionBlocks []uint64, onHit OnHitFn) (uint64, error) { - slices.Sort(transitionBlocks) - for _, block := range transitionBlocks { - if err := onHit(block); err != nil { - return 0, err - } - } - return uint64(len(transitionBlocks)), nil -} - -// FindTransitions performs divide-and-conquer search for transitions using oracle -// and calls onHit for each transition block in chronological order. -// The prevValue is optional and used to detect a transition exactly at startBlock. +// FindTransitions searches for transitions using transitionQuery and calls onHit +// during traversal in chronological order. The optional lowerBound seeds the +// previous counter value to detect a transition exactly at startBlock. +// The optional upperBound stops the search after the transition reaching or +// crossing that counter value; onHit still receives the full counter change. // It assumes the queried value is monotonic (e.g., increasing counter); if it can revert, // intermediate transitions may be missed when net change is zero. -func FindTransitions(ctx context.Context, startBlock, endBlock uint64, prevValue *big.Int, - transitionQuery TransitionQueryFn, onHit OnHitFn) (uint64, error) { +// +// The results are a transition count and a completed-block cursor. Callback +// errors, including ErrIncompleteProcessing, are propagated with the cursor set +// to the block before that transition, allowing callers to resume within it. +func FindTransitions(ctx context.Context, startBlock, endBlock uint64, lowerBound, upperBound *big.Int, + transitionQuery TransitionQueryFn, onHit OnHitFn) (uint64, uint64, error) { + if startBlock > endBlock { - return 0, nil + return 0, endBlock, nil } type interval struct { - StartBlock, EndBlock uint64 - StartValue, EndValue *big.Int + StartBlock, EndBlock, LastBlock uint64 + StartValue, EndValue *big.Int } startValue, err := transitionQuery(ctx, startBlock) if err != nil { - return 0, fmt.Errorf("transitionQuery(startBlock=%d): %w", startBlock, err) + return 0, startBlock - 1, fmt.Errorf("transitionQuery(startBlock=%d): %w", startBlock, err) } - var transitionBlocks []uint64 - if prevValue != nil { - comparisonResult := prevValue.Cmp(startValue) + var count uint64 + if lowerBound != nil { + comparisonResult := lowerBound.Cmp(startValue) if comparisonResult > 0 { - return 0, fmt.Errorf("%w: prevValue %s > startValue %s at block %d", - ErrMonotonicViolation, prevValue.String(), startValue.String(), startBlock) + return 0, startBlock - 1, fmt.Errorf("%w: lowerBound %s > startValue %s at block %d", + ErrMonotonicViolation, lowerBound.String(), startValue.String(), startBlock) } if comparisonResult < 0 { - transitionBlocks = append(transitionBlocks, startBlock) + if err := onHit(startBlock, lowerBound, startValue); err != nil { + // Resume at this block with the partially consumed counter, + // rather than querying an older block with a smaller counter. + return count, startBlock - 1, err + } + count++ } } - if startBlock == endBlock { - return sortAndCallOnHit(transitionBlocks, onHit) + if startBlock == endBlock || (upperBound != nil && startValue.Cmp(upperBound) >= 0) { + return count, startBlock, nil } endValue, err := transitionQuery(ctx, endBlock) if err != nil { - return 0, fmt.Errorf("transitionQuery(endBlock=%d): %w", endBlock, err) + return count, startBlock - 1, fmt.Errorf("transitionQuery(endBlock=%d): %w", endBlock, err) } if startValue.Cmp(endValue) == 0 { // No further transitions, but may have added startBlock - return sortAndCallOnHit(transitionBlocks, onHit) + return count, endBlock, nil } - // First phase: collect all transition blocks - stack := []interval{{StartBlock: startBlock, EndBlock: endBlock, StartValue: startValue, EndValue: endValue}} + lastProcessedBlock := startBlock + + // Visit intervals from left to right, invoking onHit as transitions are found. + stack := []interval{{ + StartBlock: startBlock, + EndBlock: endBlock, + LastBlock: endBlock, + StartValue: startValue, + EndValue: endValue, + }} for len(stack) > 0 { // Check for context cancellation select { case <-ctx.Done(): - return 0, ctx.Err() + return 0, lastProcessedBlock, ctx.Err() default: } @@ -98,25 +107,46 @@ func FindTransitions(ctx context.Context, startBlock, endBlock uint64, prevValue if iv.StartBlock == iv.EndBlock { // Found a transition block - transitionBlocks = append(transitionBlocks, iv.StartBlock) + if err := onHit(iv.StartBlock, iv.StartValue, iv.EndValue); err != nil { + // Resume at this block with the partially consumed counter, + // rather than querying an older block with a smaller counter. + return count, iv.StartBlock - 1, err + } + count++ + lastProcessedBlock = iv.LastBlock continue } midBlock := iv.StartBlock + (iv.EndBlock-iv.StartBlock)/2 //nolint:mnd midValue, err := transitionQuery(ctx, midBlock) if err != nil { - return 0, fmt.Errorf("transitionQuery(midBlock=%d): %w", midBlock, err) + return 0, lastProcessedBlock, fmt.Errorf("transitionQuery(midBlock=%d): %w", midBlock, err) } // Add new intervals to stack if there are transitions + lastBlock := iv.LastBlock if midValue.Cmp(iv.EndValue) != 0 { - stack = append(stack, interval{StartBlock: midBlock + 1, EndBlock: iv.EndBlock, StartValue: midValue, EndValue: iv.EndValue}) + if upperBound == nil || midValue.Cmp(upperBound) < 0 { + stack = append(stack, interval{ + StartBlock: midBlock + 1, + EndBlock: iv.EndBlock, + LastBlock: lastBlock, + StartValue: midValue, + EndValue: iv.EndValue, + }) + } + lastBlock = midBlock } if iv.StartValue.Cmp(midValue) != 0 { - stack = append(stack, interval{StartBlock: iv.StartBlock, EndBlock: midBlock, StartValue: iv.StartValue, EndValue: midValue}) + stack = append(stack, interval{ + StartBlock: iv.StartBlock, + EndBlock: midBlock, + LastBlock: lastBlock, + StartValue: iv.StartValue, + EndValue: midValue, + }) } } - // Second phase: sort transition blocks and call onHit in chronological order - return sortAndCallOnHit(transitionBlocks, onHit) + return count, lastProcessedBlock, nil } diff --git a/pkg/ethutil/search_test.go b/pkg/ethutil/search_test.go index 41f2f7b5f..e40956ad2 100644 --- a/pkg/ethutil/search_test.go +++ b/pkg/ethutil/search_test.go @@ -37,7 +37,7 @@ func mockTransitionQuery(values map[uint64]*big.Int) TransitionQueryFn { // Helper to create an onHit that collects blocks func collectOnHit(blocks *[]uint64) OnHitFn { - return func(block uint64) error { + return func(block uint64, _, _ *big.Int) error { *blocks = append(*blocks, block) return nil } @@ -45,7 +45,7 @@ func collectOnHit(blocks *[]uint64) OnHitFn { // Helper to create an onHit that returns error func errorOnHit(err error) OnHitFn { - return func(_ uint64) error { + return func(_ uint64, _, _ *big.Int) error { return err } } @@ -63,7 +63,7 @@ func (s *SearchSuite) TestNoTransitionsWhenValuesEqual() { 10: big.NewInt(1), } var blocks []uint64 - _, err := FindTransitions(s.ctx, 0, 10, nil, mockTransitionQuery(values), collectOnHit(&blocks)) + _, _, err := FindTransitions(s.ctx, 0, 10, nil, nil, mockTransitionQuery(values), collectOnHit(&blocks)) s.NoError(err) s.Empty(blocks) } @@ -74,7 +74,7 @@ func (s *SearchSuite) TestSingleTransition() { 1: big.NewInt(2), } var blocks []uint64 - _, err := FindTransitions(s.ctx, 0, 1, nil, mockTransitionQuery(values), collectOnHit(&blocks)) + _, _, err := FindTransitions(s.ctx, 0, 1, nil, nil, mockTransitionQuery(values), collectOnHit(&blocks)) s.NoError(err) s.Equal([]uint64{1}, blocks) } @@ -89,14 +89,14 @@ func (s *SearchSuite) TestMultipleTransitions() { 6: big.NewInt(3), } var blocks []uint64 - count, err := FindTransitions(s.ctx, 1, 6, nil, mockTransitionQuery(values), collectOnHit(&blocks)) + count, _, err := FindTransitions(s.ctx, 1, 6, nil, nil, mockTransitionQuery(values), collectOnHit(&blocks)) s.NoError(err) s.Equal(count, uint64(2)) s.Equal([]uint64{3, 5}, blocks) blocks = []uint64{} previousValue := big.NewInt(0) - count, err = FindTransitions(s.ctx, 1, 6, previousValue, mockTransitionQuery(values), collectOnHit(&blocks)) + count, _, err = FindTransitions(s.ctx, 1, 6, previousValue, nil, mockTransitionQuery(values), collectOnHit(&blocks)) s.NoError(err) s.Equal(count, uint64(3)) s.Equal([]uint64{1, 3, 5}, blocks) @@ -108,14 +108,14 @@ func (s *SearchSuite) TestTransitionAtBoundary() { 1: big.NewInt(2), } var blocks []uint64 - count, err := FindTransitions(s.ctx, 0, 1, nil, mockTransitionQuery(values), collectOnHit(&blocks)) + count, _, err := FindTransitions(s.ctx, 0, 1, nil, nil, mockTransitionQuery(values), collectOnHit(&blocks)) s.NoError(err) s.Equal(count, uint64(1)) s.Equal([]uint64{1}, blocks) blocks = []uint64{} previousValue := big.NewInt(0) - count, err = FindTransitions(s.ctx, 0, 1, previousValue, mockTransitionQuery(values), collectOnHit(&blocks)) + count, _, err = FindTransitions(s.ctx, 0, 1, previousValue, nil, mockTransitionQuery(values), collectOnHit(&blocks)) s.NoError(err) s.Equal(count, uint64(2)) s.Equal([]uint64{0, 1}, blocks) @@ -127,14 +127,14 @@ func (s *SearchSuite) TestStartEqualsEnd() { 5: big.NewInt(1), } var blocks []uint64 - count, err := FindTransitions(s.ctx, 5, 5, nil, mockTransitionQuery(values), collectOnHit(&blocks)) + count, _, err := FindTransitions(s.ctx, 5, 5, nil, nil, mockTransitionQuery(values), collectOnHit(&blocks)) s.NoError(err) s.Equal(count, uint64(0)) s.Empty(blocks) blocks = []uint64{} previousValue := big.NewInt(0) - count, err = FindTransitions(s.ctx, 5, 5, previousValue, mockTransitionQuery(values), collectOnHit(&blocks)) + count, _, err = FindTransitions(s.ctx, 5, 5, previousValue, nil, mockTransitionQuery(values), collectOnHit(&blocks)) s.NoError(err) s.Equal(count, uint64(1)) s.Equal([]uint64{5}, blocks) @@ -148,14 +148,14 @@ func (s *SearchSuite) TestContextCancellation() { ctx, cancel := context.WithCancel(s.ctx) cancel() var blocks []uint64 - _, err := FindTransitions(ctx, 0, 10, nil, mockTransitionQuery(values), collectOnHit(&blocks)) + _, _, err := FindTransitions(ctx, 0, 10, nil, nil, mockTransitionQuery(values), collectOnHit(&blocks)) s.Error(err) s.Equal(context.Canceled, err) } func (s *SearchSuite) TestTransitionQueryErrorAtStart() { var blocks []uint64 - _, err := FindTransitions(s.ctx, 0, 10, nil, errorTransitionQuery(fmt.Errorf("query error")), collectOnHit(&blocks)) + _, _, err := FindTransitions(s.ctx, 0, 10, nil, nil, errorTransitionQuery(fmt.Errorf("query error")), collectOnHit(&blocks)) s.Error(err) s.Contains(err.Error(), "transitionQuery(startBlock=0): query error") s.Empty(blocks) @@ -166,7 +166,7 @@ func (s *SearchSuite) TestTransitionQueryErrorAtEnd() { 0: big.NewInt(1), } var blocks []uint64 - _, err := FindTransitions(s.ctx, 0, 10, nil, func(_ context.Context, block uint64) (*big.Int, error) { + _, _, err := FindTransitions(s.ctx, 0, 10, nil, nil, func(_ context.Context, block uint64) (*big.Int, error) { if block == 10 { return nil, fmt.Errorf("query error at end") } @@ -183,7 +183,7 @@ func (s *SearchSuite) TestTransitionQueryErrorAtMid() { 10: big.NewInt(2), } var blocks []uint64 - _, err := FindTransitions(s.ctx, 0, 10, nil, func(_ context.Context, block uint64) (*big.Int, error) { + _, _, err := FindTransitions(s.ctx, 0, 10, nil, nil, func(_ context.Context, block uint64) (*big.Int, error) { if block == 5 { return nil, fmt.Errorf("query error at mid") } @@ -203,7 +203,7 @@ func (s *SearchSuite) TestOnHitError() { 1: big.NewInt(2), } var blocks []uint64 - _, err := FindTransitions(s.ctx, 0, 1, nil, mockTransitionQuery(values), errorOnHit(fmt.Errorf("onHit error"))) + _, _, err := FindTransitions(s.ctx, 0, 1, nil, nil, mockTransitionQuery(values), errorOnHit(fmt.Errorf("onHit error"))) s.Error(err) s.Contains(err.Error(), "onHit error") s.Empty(blocks) @@ -219,14 +219,14 @@ func (s *SearchSuite) TestLargeRange() { } } var blocks []uint64 - _, err := FindTransitions(s.ctx, 0, 100, nil, mockTransitionQuery(values), collectOnHit(&blocks)) + _, _, err := FindTransitions(s.ctx, 0, 100, nil, nil, mockTransitionQuery(values), collectOnHit(&blocks)) s.NoError(err) s.Equal([]uint64{50}, blocks) } func (s *SearchSuite) TestNoValuesDefined() { var blocks []uint64 - _, err := FindTransitions(s.ctx, 0, 10, nil, mockTransitionQuery(map[uint64]*big.Int{}), collectOnHit(&blocks)) + _, _, err := FindTransitions(s.ctx, 0, 10, nil, nil, mockTransitionQuery(map[uint64]*big.Int{}), collectOnHit(&blocks)) s.NoError(err) s.Empty(blocks) } @@ -238,9 +238,138 @@ func (s *SearchSuite) TestMonotonicViolation() { } var blocks []uint64 previousValue := big.NewInt(3) // 3 > 2, violation - _, err := FindTransitions(s.ctx, 0, 1, previousValue, mockTransitionQuery(values), collectOnHit(&blocks)) + _, _, err := FindTransitions(s.ctx, 0, 1, previousValue, nil, mockTransitionQuery(values), collectOnHit(&blocks)) s.Error(err) s.ErrorIs(err, ErrMonotonicViolation) s.Contains(err.Error(), "monotonic assumption violated") s.Empty(blocks) } + +func (s *SearchSuite) TestUpperBoundStopsAtReachingTransition() { + // Counters can jump by more than one in a block. The block reaching or + // crossing the bound must be processed, but no later transition may be. + values := map[uint64]*big.Int{ + 10: big.NewInt(2), 11: big.NewInt(2), + 12: big.NewInt(5), 13: big.NewInt(5), + 14: big.NewInt(9), 15: big.NewInt(9), + 16: big.NewInt(12), 17: big.NewInt(12), + } + for _, test := range []struct { + name string + lower *big.Int + upper *big.Int + want []uint64 + last uint64 + }{ + {name: "unbounded", want: []uint64{12, 14, 16}, last: 17}, + {name: "exact counter", upper: big.NewInt(5), want: []uint64{12}, last: 13}, + {name: "counter jumps past bound", upper: big.NewInt(6), want: []uint64{12, 14}, last: 15}, + {name: "bound above final counter", upper: big.NewInt(20), want: []uint64{12, 14, 16}, last: 17}, + {name: "bound already reached at start", upper: big.NewInt(2), last: 10}, + {name: "bound already exceeded at start", upper: big.NewInt(1), last: 10}, + {name: "initial transition reaches bound", lower: big.NewInt(0), upper: big.NewInt(2), want: []uint64{10}, last: 10}, + {name: "initial and later transitions", lower: big.NewInt(0), upper: big.NewInt(9), want: []uint64{10, 12, 14}, last: 15}, + } { + s.Run(test.name, func() { + var blocks []uint64 + count, last, err := FindTransitions(s.ctx, 10, 17, test.lower, test.upper, + mockTransitionQuery(values), collectOnHit(&blocks)) + s.Require().NoError(err) + s.Equal(test.want, blocks) + s.Equal(uint64(len(test.want)), count) + // Progress includes the unchanged blocks following a transition. + s.Equal(test.last, last) + }) + } +} + +func (s *SearchSuite) TestLastProcessedBlockOnEmptyAndSingleBlockRanges() { + for _, test := range []struct { + name string + start, end uint64 + lower *big.Int + want []uint64 + last uint64 + }{ + {name: "empty range", start: 11, end: 10, last: 10}, + {name: "single block without transition", start: 10, end: 10, last: 10}, + {name: "single block with transition", start: 10, end: 10, lower: big.NewInt(0), want: []uint64{10}, last: 10}, + } { + s.Run(test.name, func() { + var blocks []uint64 + count, last, err := FindTransitions(s.ctx, test.start, test.end, test.lower, nil, + mockTransitionQuery(map[uint64]*big.Int{10: big.NewInt(1)}), collectOnHit(&blocks)) + s.Require().NoError(err) + s.Equal(test.want, blocks) + s.Equal(uint64(len(test.want)), count) + s.Equal(test.last, last) + }) + } +} + +func (s *SearchSuite) TestLastProcessedBlockExcludesFailedCallback() { + // Include unchanged blocks to verify that progress can extend beyond the + // last successful callback, while excluding the failed transition. + for _, test := range []struct { + failBlock uint64 + last uint64 + completed []uint64 + }{ + {failBlock: 10, last: 9}, + {failBlock: 12, last: 11, completed: []uint64{10}}, + {failBlock: 14, last: 13, completed: []uint64{10, 12}}, + {failBlock: 16, last: 15, completed: []uint64{10, 12, 14}}, + } { + failBlock := test.failBlock + s.Run(fmt.Sprintf("failure_at_%d", failBlock), func() { + callbackErr := fmt.Errorf("callback failed at %d", failBlock) + var completed []uint64 + count, last, err := FindTransitions(s.ctx, 10, 17, big.NewInt(0), nil, + func(_ context.Context, block uint64) (*big.Int, error) { + return new(big.Int).SetUint64((block-10)/2 + 1), nil + }, func(block uint64, _, _ *big.Int) error { + if block == failBlock { + return callbackErr + } + completed = append(completed, block) + return nil + }) + s.ErrorIs(err, callbackErr) + s.Equal(test.last, last) + s.Equal(test.completed, completed) + s.Equal(uint64(len(completed)), count) + }) + } +} + +func (s *SearchSuite) TestIncompleteTransitionResumesAfterGap() { + for _, initial := range []bool{false, true} { + s.Run(fmt.Sprintf("initial_transition_%t", initial), func() { + var completed []uint64 + count, last, err := FindTransitions(s.ctx, 10, 30, big.NewInt(0), nil, + func(_ context.Context, block uint64) (*big.Int, error) { + if block >= 20 { + return big.NewInt(1000), nil + } + if initial { + return big.NewInt(1), nil + } + return big.NewInt(0), nil + }, func(block uint64, _, _ *big.Int) error { + if block == 20 { + return ErrIncompleteProcessing + } + completed = append(completed, block) + return nil + }) + s.Require().ErrorIs(err, ErrIncompleteProcessing) + s.Equal(uint64(19), last) + s.Equal(uint64(len(completed)), count) + if initial { + s.Equal([]uint64{10}, completed) + } else { + s.Empty(completed) + } + }) + } +} diff --git a/test/integration/memory_helpers_test.go b/test/integration/memory_helpers_test.go new file mode 100644 index 000000000..58580c59d --- /dev/null +++ b/test/integration/memory_helpers_test.go @@ -0,0 +1,58 @@ +// (c) Cartesi and individual authors (see AUTHORS) +// SPDX-License-Identifier: Apache-2.0 (see LICENSE) + +package integration + +import ( + "fmt" + "io" + "os" + "os/exec" + "strconv" + "strings" +) + +// processRSS reads resident memory, excluding virtual address-space reservations. +func processRSS(pid int) (uint64, error) { + data, err := os.ReadFile(fmt.Sprintf("/proc/%d/status", pid)) + if err != nil { + return 0, err + } + for _, line := range strings.Split(string(data), "\n") { + fields := strings.Fields(line) + if len(fields) == 3 && fields[0] == "VmRSS:" && fields[2] == "kB" { + kb, err := strconv.ParseUint(fields[1], 10, 64) + return kb * 1024, err + } + } + return 0, fmt.Errorf("process %d has no VmRSS", pid) +} + +// limitedNodeCommand applies a per-process Linux address-space limit before +// exec, preserving the node PID and signal handling. Allocation failures appear +// in the node log. This is virtual memory, not RSS: Go reservations and machine +// mappings count too. Children inherit the limit, each with its own allowance. +// CARTESI_TEST_NODE_MEMORY_LIMIT_BYTES overrides the default 2.5 GiB; 0 disables +// the limit. This setting belongs to the test runner, not extraEnv. Externally +// managed nodes are unaffected. Enforcing the limit requires util-linux prlimit. +func limitedNodeCommand(binary string, log io.Writer) (*exec.Cmd, error) { + value := os.Getenv("CARTESI_TEST_NODE_MEMORY_LIMIT_BYTES") + if value == "" { + value = "2900000000" + } + limit, err := strconv.ParseUint(value, 10, 64) + if err != nil { + return nil, fmt.Errorf("invalid CARTESI_TEST_NODE_MEMORY_LIMIT_BYTES %q: %w", value, err) + } + if limit == 0 { + fmt.Fprintf(log, "%s: memory limit disabled\n", binary) + return exec.Command(binary), nil //nolint:gosec + } + prlimit, err := exec.LookPath("prlimit") + if err != nil { + return nil, fmt.Errorf("memory limit requires util-linux prlimit: %w", err) + } + value = strconv.FormatUint(limit, 10) + fmt.Fprintf(log, "%s: virtual memory limit %s bytes per process\n", binary, value) + return exec.Command(prlimit, "--as="+value+":"+value, "--", binary), nil //nolint:gosec +} diff --git a/test/integration/memory_monitor_test.go b/test/integration/memory_monitor_test.go new file mode 100644 index 000000000..997de8912 --- /dev/null +++ b/test/integration/memory_monitor_test.go @@ -0,0 +1,76 @@ +// (c) Cartesi and individual authors (see AUTHORS) +// SPDX-License-Identifier: Apache-2.0 (see LICENSE) + +//go:build endtoendtests + +package integration + +import ( + "fmt" + "sync" + "time" +) + +type rssMeasurement struct { + baseline, peak uint64 + samples uint64 + err error +} + +// monitorNodeRSS sums the managed service processes in multiprocess mode. +// Shared resident pages can be counted more than once. Child machine processes +// are excluded. The peak is sampled every 10 ms; shorter spikes may be missed. +// Call stop before reading the result; repeated stops return the same result. +func monitorNodeRSS(node nodeHandle) (stop func() rssMeasurement, err error) { + var procs []*nodeSubprocess + switch n := node.(type) { + case *nodeProcess: + procs = []*nodeSubprocess{n.nodeSubprocess} + case *multiNode: + procs = n.procs + default: + return nil, fmt.Errorf("RSS measurement requires managed node processes") + } + read := func() (uint64, error) { + var total uint64 + for _, p := range procs { + rss, err := processRSS(p.cmd.Process.Pid) + if err != nil { + return 0, fmt.Errorf("read %s RSS: %w", p.name, err) + } + total += rss + } + return total, nil + } + baseline, err := read() + if err != nil { + return nil, err + } + result := rssMeasurement{baseline: baseline, peak: baseline, samples: 1} + quit, done := make(chan struct{}), make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(10 * time.Millisecond) + defer ticker.Stop() + for { + select { + case <-quit: + return + case <-ticker.C: + rss, err := read() + if err != nil { + result.err = err + return + } + result.peak = max(result.peak, rss) + result.samples++ + } + } + }() + var once sync.Once + return func() rssMeasurement { + once.Do(func() { close(quit) }) + <-done + return result + }, nil +} diff --git a/test/integration/memory_rss_test.go b/test/integration/memory_rss_test.go new file mode 100644 index 000000000..eb4c50d43 --- /dev/null +++ b/test/integration/memory_rss_test.go @@ -0,0 +1,22 @@ +// (c) Cartesi and individual authors (see AUTHORS) +// SPDX-License-Identifier: Apache-2.0 (see LICENSE) + +package integration + +import ( + "os" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestProcessRSS(t *testing.T) { + if _, err := os.Stat("/proc/self/status"); err != nil { + t.Skip("RSS measurement requires Linux procfs") + } + rss, err := processRSS(os.Getpid()) + require.NoError(t, err) + require.Positive(t, rss) + _, err = processRSS(-1) + require.Error(t, err, "missing processes must fail measurement") +} diff --git a/test/integration/multinode_helpers_test.go b/test/integration/multinode_helpers_test.go index a4f810361..1655ecd49 100644 --- a/test/integration/multinode_helpers_test.go +++ b/test/integration/multinode_helpers_test.go @@ -118,7 +118,11 @@ func startMultiNode(logPath string, extraEnv ...string) (*multiNode, error) { mn.stop(nil) return nil, fmt.Errorf("%s not found on PATH: %w", svc.name, err) } - cmd := exec.Command(svc.name) //nolint:gosec + cmd, err := limitedNodeCommand(svc.name, logFile) + if err != nil { + mn.stop(nil) + return nil, err + } cmd.Stdout = logFile cmd.Stderr = logFile cmd.Env = append(append(append([]string{}, base...), svc.extraEnv...), extraEnv...) diff --git a/test/integration/node_helpers_test.go b/test/integration/node_helpers_test.go index 58d20855f..65f54f819 100644 --- a/test/integration/node_helpers_test.go +++ b/test/integration/node_helpers_test.go @@ -142,7 +142,11 @@ func startNodeWithLog(logPath string, extraEnv ...string) (*nodeProcess, error) return nil, fmt.Errorf("open log file %s: %w", logPath, err) } - cmd := exec.Command(nodeBinary) + cmd, err := limitedNodeCommand(nodeBinary, logFile) + if err != nil { + logFile.Close() + return nil, err + } cmd.Stdout = logFile cmd.Stderr = logFile if workDir := os.Getenv("CARTESI_TEST_NODE_WORKDIR"); workDir != "" { @@ -166,13 +170,15 @@ func startNodeWithLog(logPath string, extraEnv ...string) (*nodeProcess, error) return nil, fmt.Errorf("start node: %w", err) } + n := &nodeProcess{nodeSubprocess: proc, logFile: logFile} + // Stream the log file to the terminal via a separate tail process. // We write to /dev/tty to bypass go test and gotestsum's output capture, // so the user sees node logs in real time just like the old Makefile did. // Falls back silently if /dev/tty is not available (e.g., CI, compose). tty, ttyErr := os.OpenFile("/dev/tty", os.O_WRONLY, 0) if ttyErr != nil { - return &nodeProcess{nodeSubprocess: proc, logFile: logFile}, nil + return n, nil } tail := exec.Command("tail", "-f", logPath) //nolint:gosec @@ -181,10 +187,11 @@ func startNodeWithLog(logPath string, extraEnv ...string) (*nodeProcess, error) tailProcess, err := startNodeSubprocess("node-log-tail", tail) if err != nil { tty.Close() - return &nodeProcess{nodeSubprocess: proc, logFile: logFile}, nil + return n, nil } - return &nodeProcess{nodeSubprocess: proc, logFile: logFile, tail: tailProcess, tty: tty}, nil + n.tail, n.tty = tailProcess, tty + return n, nil } // waitForHealth polls the node's readyz endpoint until it responds 200 OK diff --git a/test/integration/same_block_inputs_test.go b/test/integration/same_block_inputs_test.go index 16d1d65e5..2bf7db00a 100644 --- a/test/integration/same_block_inputs_test.go +++ b/test/integration/same_block_inputs_test.go @@ -16,6 +16,7 @@ import ( "math/big" "net/http" "os" + "path/filepath" "strconv" "strings" "testing" @@ -24,14 +25,17 @@ import ( "github.com/cartesi/rollups-node/internal/config" "github.com/cartesi/rollups-node/internal/jsonrpc/api" "github.com/cartesi/rollups-node/internal/model" + "github.com/cartesi/rollups-node/internal/repository" "github.com/cartesi/rollups-node/internal/repository/factory" "github.com/cartesi/rollups-node/pkg/contracts/idaveconsensus" "github.com/cartesi/rollups-node/pkg/contracts/iinputbox" + "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/accounts/abi" "github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethclient" + "github.com/ethereum/go-ethereum/rpc" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" ) @@ -66,7 +70,7 @@ var spamboxABIJSON string var spamboxBytecode string func (s *SameBlockInputsSuite) SetupSuite() { - s.ctx, s.cancel = context.WithTimeout(context.Background(), 15*time.Minute) + s.ctx, s.cancel = context.WithTimeout(context.Background(), 30*time.Minute) s.client = newIntegrationEthClient(s.ctx, s.T()) } @@ -210,6 +214,187 @@ func (s *SameBlockInputsSuite) TestMultipleInputsOneTransactionAuthority() { } } +// TestMaximumInputsOneBlockAuthority stresses the reader's in-memory input +// batch with the largest Spambox call that fits the block gas budget. It checks +// ingestion and process survival under the configured virtual-memory limit. +// It samples service RSS every 10 ms and requires peak growth <= 256 MiB, +// overridable with CARTESI_TEST_INPUT_STRESS_MAX_RSS_GROWTH_BYTES (positive). +// This measures the managed service processes, not child machines, and does +// not wait for machine execution. Shorter memory spikes may escape sampling. +// A normal block may not exhaust memory. CARTESI_TEST_INPUT_STRESS_GAS_LIMIT +// can raise Anvil's block limit for an OOM +// reproduction, together with CARTESI_TEST_NODE_MEMORY_LIMIT_BYTES. RPC gas caps +// must also allow the requested budget. The original block limit is restored. +func (s *SameBlockInputsSuite) TestMaximumInputsOneBlockAuthority() { + r := s.Require() + // Require an observable process before mutating the shared chain. + if sharedNode == nil { + s.T().Skip("memory stress test requires a test-managed node") + } + maxRSSGrowth, err := strconv.ParseUint(envOrDefault( + "CARTESI_TEST_INPUT_STRESS_MAX_RSS_GROWTH_BYTES", "268435456"), 10, 64) + r.NoError(err, "parse stress RSS growth limit") + r.Positive(maxRSSGrowth, "stress RSS growth limit must be positive") + s.appName = uniqueAppName("maximum-inputs-one-block-authority") + root, err := repositoryRoot() + r.NoError(err, "resolve repository root") + dappPath := envOrDefault("CARTESI_TEST_DAPP_PATH", filepath.Join(root, "applications", "echo-dapp")) + appAddrStr, err := deployApplication(s.ctx, s.appName, dappPath, "--salt", uniqueSalt()) + r.NoError(err, "deploy app") + appAddr := common.HexToAddress(appAddrStr) + inputBoxAddr := inputBoxAddress(s.T()) + spambox := deploySpambox(s.ctx, s.T(), s.client, inputBoxAddr) + r.Zero(inputBoxInputCount(s.ctx, s.T(), s.client, inputBoxAddr, appAddr), + "stress application must start without inputs") + + // Anvil applies gas-limit changes to the next block; latest may still + // contain the previous test's limit after its cleanup restored the setting. + header, err := s.client.HeaderByNumber(s.ctx, big.NewInt(int64(rpc.PendingBlockNumber))) + r.NoError(err, "read pending block gas limit") + gasLimit := uint64(500_000_000) + if value := os.Getenv("CARTESI_TEST_INPUT_STRESS_GAS_LIMIT"); value != "" { + gasLimit, err = strconv.ParseUint(value, 10, 64) + r.NoError(err, "parse CARTESI_TEST_INPUT_STRESS_GAS_LIMIT") + r.Greater(gasLimit, uint64(21_000), "stress gas budget must exceed intrinsic transaction gas") + } + + // Pause both mining modes so the measured state cannot change between the + // probes and the real transaction. Restore even after a fatal assertion, + // using a fresh context in case the suite deadline has expired. + defer func() { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + s.Assert().NoError(anvilRPC(ctx, "evm_setBlockGasLimit", fmt.Sprintf("0x%x", header.GasLimit))) + s.Assert().NoError(anvilRPC(ctx, "evm_setIntervalMining", 1)) + s.Assert().NoError(anvilRPC(ctx, "evm_setAutomine", true)) + }() + setAnvilAutomine(s.ctx, s.T(), false) + setAnvilIntervalMining(s.ctx, s.T(), 0) + r.NoError(anvilRPC(s.ctx, "evm_setBlockGasLimit", fmt.Sprintf("0x%x", gasLimit)), + "set stress block gas limit") + // Publish the limit so eth_estimateGas sees the same budget as mining. + r.NoError(anvilMine(s.ctx, 1), "mine block with stress gas limit") + + opts := transactorForMnemonicIndex(s.ctx, s.T(), s.client, 3) + // Fund only the stress sender explicitly: lack of ETH must not cap the + // workload. Restore its previous balance after the transaction is mined. + balance, err := s.client.BalanceAt(s.ctx, opts.From, nil) + r.NoError(err, "read stress sender balance") + defer func() { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + s.Assert().NoError(anvilSetBalance(ctx, opts.From.Hex(), "0x"+balance.Text(16))) + }() + // Use a fee above the frozen block's base fee and reserve the full gas cost. + opts.GasPrice = new(big.Int).Add(new(big.Int).Mul(header.BaseFee, big.NewInt(2)), big.NewInt(1_000_000_000)) + funds := new(big.Int).Mul(new(big.Int).SetUint64(gasLimit), opts.GasPrice) + funds.Add(funds, balance) + r.NoError(anvilSetBalance(s.ctx, opts.From.Hex(), "0x"+funds.Text(16)), "fund stress transaction") + + // Build calldata without sending or estimating. Probe with eth_estimateGas + // to include intrinsic gas and EIP-150 forwarding costs, not just call gas. + probe := *opts + probe.NoSend = true + probe.GasLimit = gasLimit + probe.Nonce = big.NewInt(0) + estimate := func(count uint64) (uint64, error) { + tx, err := spambox.Transact(&probe, "spam", appAddr, new(big.Int).SetUint64(count)) + if err != nil { + return 0, err + } + return s.client.EstimateGas(s.ctx, ethereum.CallMsg{ + From: opts.From, To: tx.To(), Gas: gasLimit, + GasPrice: opts.GasPrice, Data: tx.Data(), + }) + } + s.T().Logf("about to estimate count for gas budget %d", gasLimit) + _, err = estimate(1) + r.NoError(err, "one input must fit before searching the gas boundary") + // Each input emits a LOG costing at least 375 gas; this is a strict upper + // bound, even without storage, hashing, calldata, or memory expansion costs. + low, high := uint64(1), gasLimit/375+1 + for low+1 < high { + mid := low + (high-low)/2 + gas, err := estimate(mid) + if err != nil { + // An inner addInput running out of gas can surface as a revert. + // Transport, timeout, and other RPC failures must not shrink the batch. + message := strings.ToLower(err.Error()) + r.True(strings.Contains(message, "out of gas") || + strings.Contains(message, "gas required exceeds allowance") || + strings.Contains(message, "execution reverted"), + "estimate %d inputs failed unexpectedly: %v", mid, err) + high = mid + } else if gas > gasLimit { + high = mid + } else { + low = mid + } + } + count := low + s.T().Logf("stress batch: %d inputs, block gas budget %d, node memory limit setting %q", + count, gasLimit, os.Getenv("CARTESI_TEST_NODE_MEMORY_LIMIT_BYTES")) + opts.GasLimit = gasLimit + tx, err := spambox.Transact(opts, "spam", appAddr, new(big.Int).SetUint64(count)) + r.NoError(err, "submit maximum-input transaction") + stopMemory, err := monitorNodeRSS(sharedNode) + r.NoError(err, "start node RSS measurement before making stress inputs visible") + defer func() { + memory := stopMemory() + growth := memory.peak - memory.baseline + s.T().Logf("stress service RSS: baseline=%d peak=%d growth=%d limit=%d samples=%d interval=10ms inputs=%d block_gas_limit=%d", + memory.baseline, memory.peak, growth, maxRSSGrowth, memory.samples, count, gasLimit) + s.Assert().NoError(memory.err, "RSS measurement must remain available throughout ingestion") + s.Assert().Greater(memory.samples, uint64(1), "RSS measurement must cover ingestion") + s.Assert().LessOrEqual(growth, maxRSSGrowth, "sampled peak service RSS growth exceeds stress limit") + }() + // Mining makes the inputs visible to the reader. After confirming the + // receipt, poll ingestion progress and check whether the node has exited. + r.NoError(anvilMine(s.ctx, 1), "mine stress transaction in one block") + receiptCtx, receiptCancel := context.WithTimeout(s.ctx, 2*time.Minute) + defer receiptCancel() + receipt := waitReceipt(receiptCtx, s.T(), s.client, tx) + r.Equal(uint64(types.ReceiptStatusSuccessful), receipt.Status, "stress transaction must succeed") + r.Equal(count, inputBoxInputCount(s.ctx, s.T(), s.client, inputBoxAddr, appAddr), + "all stress inputs must exist on chain") + s.T().Logf("stress block %d: inputs=%d gas used=%d/%d tx=%s", + receipt.BlockNumber.Uint64(), count, receipt.GasUsed, gasLimit, tx.Hash()) + + dsn, err := config.GetDatabaseConnection() + r.NoError(err, "get database connection") + repo, err := factory.NewRepositoryFromConnectionString(s.ctx, dsn.Raw()) + r.NoError(err, "open repository") + defer repo.Close() + ctx, cancel := context.WithTimeout(s.ctx, inputProcessingTimeout) + defer cancel() + err = pollUntil(ctx, 2*time.Second, func() (bool, error) { + if checker, ok := sharedNode.(nodeExitChecker); ok { + if err := checker.exitedProcessError(); err != nil { + return false, err + } + } + stored, err := repo.GetNumberOfInputs(ctx, s.appName) + if err != nil || stored != count { + return false, err + } + app, err := repo.GetApplication(ctx, s.appName) + return app != nil && app.LastInputCheckBlock >= receipt.BlockNumber.Uint64(), err + }) + r.NoError(err, "reader must ingest %d inputs from block %d under its memory limit; check node log for allocation failures", + count, receipt.BlockNumber.Uint64()) + first, err := repo.GetInput(ctx, s.appName, 0) + r.NoError(err, "read first ingested input") + r.NotNil(first) + // Count in SQL with a one-row page, avoiding another large in-memory batch + // in the test runner. All inputs must belong to this transaction and epoch. + txHash := tx.Hash() + _, total, err := repo.ListInputs(ctx, s.appName, repository.InputFilter{ + EpochIndex: &first.EpochIndex, TransactionHash: &txHash, + }, repository.Pagination{Limit: 1}, false) + r.NoError(err, "count stress inputs in one epoch") + r.Equal(count, total, "entire stress batch must be stored in one epoch") +} + // runMultipleInputsOneBlock deploys an application, batches three inputs into a // single L1 block, and verifies the node ingests them with contiguous indices, // processes each through the machine, and assigns all of them to the same epoch.