Per-shard-pair transaction counters

nil/cmd/nil_block_generator/internal/commands/client_test.go

@@ -97,18 +97,18 @@ func (s *NilBlockGeneratorTestSuite) TearDownTest() {
 }
 
 func (s *NilBlockGeneratorTestSuite) TestGetBlock() {
-	smartAccountAdr, hexKey, err := CreateNewSmartAccount(s.url, s.logger)
+	smartAccountAddr, hexKey, err := CreateNewSmartAccount(s.url, s.logger)
 	s.Require().NoError(err)
-	s.Require().NotEmpty(smartAccountAdr)
+	s.Require().NotEmpty(smartAccountAddr)
 	s.Require().NotEmpty(hexKey)
 
-	contractAddress, err := DeployContract(s.url, smartAccountAdr, s.contractBasePath, hexKey, s.deployArgs, s.logger)
+	contractAddress, err := DeployContract(s.url, smartAccountAddr, s.contractBasePath, hexKey, s.deployArgs, s.logger)
 	s.Require().NoError(err)
 	s.Require().NotEmpty(contractAddress)
 
 	var calls []Call
 	calls = append(calls, *NewCall(s.contractName, s.method, s.contractBasePath+".abi", contractAddress, s.callArgs, 1))
-	blockHash, err := CallContract(s.url, smartAccountAdr, hexKey, calls, s.logger)
+	blockHash, err := CallContract(s.url, smartAccountAddr, hexKey, calls, s.logger)
 	s.Require().NoError(err)
 	s.Require().NotEmpty(blockHash)
 }

nil/internal/collate/proposer.go

@@ -347,10 +347,6 @@ func (p *proposer) handleTransactionsFromPool() error {
 	return nil
 }
 
-func (p *proposer) isTxProcessed(dbTx db.RoTx, txHash common.Hash) (bool, error) {
-	return dbTx.ExistsInShard(p.params.ShardId, db.BlockHashAndInTransactionIndexByTransactionHash, txHash.Bytes())
-}
-
 func (p *proposer) handleTransactionsFromNeighbors(tx db.RoTx) error {
 	state, err := db.ReadCollatorState(tx, p.params.ShardId)
 	if err != nil && !errors.Is(err, db.ErrKeyNotFound) {
@@ -376,6 +372,7 @@ func (p *proposer) handleTransactionsFromNeighbors(tx db.RoTx) error {
 			state.Neighbors = append(state.Neighbors, types.Neighbor{ShardId: neighborId})
 		}
 		neighbor := &state.Neighbors[position]
+		nextTx := p.executionState.InTxCounts[neighborId]
 
 		var lastBlockNumber types.BlockNumber
 		lastBlock, _, err := db.ReadLastBlock(tx, neighborId)
@@ -433,17 +430,19 @@ func (p *proposer) handleTransactionsFromNeighbors(tx db.RoTx) error {
 				}
 
 				if txn.To.ShardId() == p.params.ShardId {
-					txnHash := txn.Hash()
-					// TODO: Temporary workaround to prevent transaction duplication
-					isProcessed, err := p.isTxProcessed(tx, txnHash)
-					if err != nil {
-						return err
-					}
-					if isProcessed {
+					if txn.TxId < nextTx {
+						// When we become proposer, we start with an outdated CollatorState,
+						// so we need to skip transactions that were already processed.
+						p.logger.Debug().
+							Uint64("txId", uint64(txn.TxId)).Uint64("nextTx", uint64(nextTx)).
+							Msg("Already processed transaction")
 						continue
 					}
+					nextTx++
+
+					txnHash := txn.Hash()
 
-					if err := execution.ValidateInternalTransaction(txn); err != nil {
+					if err := p.executionState.ValidateInternalTransaction(txn); err != nil {
 						p.logger.Warn().Err(err).
 							Stringer(logging.FieldTransactionHash, txnHash).
 							Msg("Invalid internal transaction")

nil/internal/execution/block_generator.go

@@ -69,20 +69,20 @@ func NewBlockGenerator(
 	ctx context.Context,
 	params BlockGeneratorParams,
 	txFabric db.DB,
-	block *types.Block,
+	prevBlock *types.Block,
 ) (*BlockGenerator, error) {
 	rwTx, err := txFabric.CreateRwTx(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	configAccessor, err := config.NewConfigAccessorFromBlock(ctx, txFabric, block, params.ShardId)
+	configAccessor, err := config.NewConfigAccessorFromBlock(ctx, txFabric, prevBlock, params.ShardId)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create config accessor: %w", err)
 	}
 
 	executionState, err := NewExecutionState(rwTx, params.ShardId, StateParams{
-		Block:          block,
+		Block:          prevBlock,
 		ConfigAccessor: configAccessor,
 		FeeCalculator:  params.FeeCalculator,
 		Mode:           params.ExecutionMode,
@@ -344,17 +344,23 @@ func (g *BlockGenerator) GenerateBlock(
 	return g.finalize(proposal.PrevBlockId+1, params)
 }
 
-func ValidateInternalTransaction(transaction *types.Transaction) error {
+func (es *ExecutionState) ValidateInternalTransaction(transaction *types.Transaction) error {
 	check.PanicIfNot(transaction.IsInternal())
 
+	nextTx := es.InTxCounts[transaction.From.ShardId()]
+	if transaction.TxId != nextTx {
+		return types.NewError(types.ErrorTxIdGap)
+	}
+	es.InTxCounts[transaction.From.ShardId()] = nextTx + 1
+
 	if transaction.IsDeploy() {
 		return ValidateDeployTransaction(transaction)
 	}
 	return nil
 }
 
 func (g *BlockGenerator) handleInternalInTransaction(txn *types.Transaction) *ExecutionResult {
-	if err := ValidateInternalTransaction(txn); err != nil {
+	if err := g.executionState.ValidateInternalTransaction(txn); err != nil {
 		g.logger.Warn().Err(err).Msg("Invalid internal transaction")
 		return NewExecutionResult().SetError(types.KeepOrWrapError(types.ErrorValidation, err))
 	}

nil/internal/execution/contract_test.go

@@ -256,8 +256,8 @@ func TestAsyncCall(t *testing.T) {
 	callTransaction.MaxFeePerGas = defaultMaxFeePerGas
 	callTransaction.Data = calldata
 	callTransaction.To = addrCaller
-	txnHash := callTransaction.Hash()
 	res := state.AddAndHandleTransaction(ctx, callTransaction, dummyPayer{})
+	txnHash := callTransaction.Hash()
 	require.False(t, res.Failed())
 
 	require.Len(t, state.OutTransactions, 1)
@@ -278,8 +278,8 @@ func TestAsyncCall(t *testing.T) {
 	require.NoError(t, err)
 
 	callTransaction.Data = calldata
-	txnHash = callTransaction.Hash()
 	res = state.AddAndHandleTransaction(ctx, callTransaction, dummyPayer{})
+	txnHash = callTransaction.Hash()
 	require.False(t, res.Failed())
 
 	require.Len(t, state.OutTransactions, 2)
@@ -336,8 +336,8 @@ func TestSendTransaction(t *testing.T) {
 	callTransaction.Data = calldata
 	callTransaction.To = addrCaller
 	callTransaction.Seqno = 1
-	tx := callTransaction.Hash()
 	res := state.AddAndHandleTransaction(ctx, callTransaction, dummyPayer{})
+	tx := callTransaction.Hash()
 	require.False(t, res.Failed())
 	require.NotEmpty(t, state.Receipts)
 	require.True(t, state.Receipts[len(state.Receipts)-1].Success)

nil/internal/execution/proposal.go

@@ -130,7 +130,7 @@ func SplitOutTransactions(
 	})
 }
 
-func ConvertTxnRefs(refs []*InternalTxnReference, parentBlocks []*ParentBlock) ([]*types.Transaction, error) {
+func convertTxnRefs(refs []*InternalTxnReference, parentBlocks []*ParentBlock) ([]*types.Transaction, error) {
 	res := make([]*types.Transaction, len(refs))
 	for i, ref := range refs {
 		if ref.ParentBlockIndex >= uint32(len(parentBlocks)) {
@@ -158,11 +158,11 @@ func ConvertProposal(proposal *ProposalSSZ) (*Proposal, error) {
 		parentBlocks[i] = converted
 	}
 
-	internalTxns, err := ConvertTxnRefs(proposal.InternalTxnRefs, parentBlocks)
+	internalTxns, err := convertTxnRefs(proposal.InternalTxnRefs, parentBlocks)
 	if err != nil {
 		return nil, fmt.Errorf("invalid internal transactions: %w", err)
 	}
-	forwardTxns, err := ConvertTxnRefs(proposal.ForwardTxnRefs, parentBlocks)
+	forwardTxns, err := convertTxnRefs(proposal.ForwardTxnRefs, parentBlocks)
 	if err != nil {
 		return nil, fmt.Errorf("invalid forward transactions: %w", err)
 	}