Fix/input processing too much memory

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)$$

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)

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)

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)

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)

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)
 			}
 		})
 	}

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,

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"