TXM In-memory: step 3-04-MarkAllConfirmedMissingReceipt (#12232)

* implement MarkAllConfirmedMissingReceipt

* implement tests for MarkAllConfirmedMissingReceipt

* add panic if incorrect chainID

* address comments

* address comments

common/txmgr/inmemory_store.go

@@ -356,7 +356,53 @@ func (ms *inMemoryStore[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) FindT
 	return nil, nil
 }
 func (ms *inMemoryStore[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) MarkAllConfirmedMissingReceipt(ctx context.Context, chainID CHAIN_ID) error {
-	return nil
+	if ms.chainID.String() != chainID.String() {
+		panic("invalid chain ID")
+	}
+
+	// Persist to persistent storage
+	if err := ms.persistentTxStore.MarkAllConfirmedMissingReceipt(ctx, chainID); err != nil {
+		return err
+	}
+
+	// Update in memory store
+	var errs error
+	ms.addressStatesLock.RLock()
+	defer ms.addressStatesLock.RUnlock()
+	for _, as := range ms.addressStates {
+		// Get the max confirmed sequence
+		filter := func(tx *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) bool { return true }
+		states := []txmgrtypes.TxState{TxConfirmed}
+		txs := as.findTxs(states, filter)
+		var maxConfirmedSequence SEQ
+		for _, tx := range txs {
+			if tx.Sequence == nil {
+				continue
+			}
+			if (*tx.Sequence).Int64() > maxConfirmedSequence.Int64() {
+				maxConfirmedSequence = *tx.Sequence
+			}
+		}
+
+		// Mark all unconfirmed txs with a sequence less than the max confirmed sequence as confirmed_missing_receipt
+		filter = func(tx *txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) bool {
+			if tx.Sequence == nil {
+				return false
+			}
+
+			return (*tx.Sequence).Int64() < maxConfirmedSequence.Int64()
+		}
+		states = []txmgrtypes.TxState{TxUnconfirmed}
+		txs = as.findTxs(states, filter)
+		for _, tx := range txs {
+			if err := as.moveUnconfirmedToConfirmedMissingReceipt(tx.ID); err != nil {
+				err = fmt.Errorf("mark_all_confirmed_missing_receipt: address: %s: %w", as.fromAddress, err)
+				errs = errors.Join(errs, err)
+			}
+		}
+	}
+
+	return errs
 }
 func (ms *inMemoryStore[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) MarkOldTxesMissingReceiptAsErrored(ctx context.Context, blockNum int64, finalityDepth uint32, chainID CHAIN_ID) error {
 	return nil

core/chains/evm/txmgr/evm_inmemory_store_test.go

@@ -120,7 +120,65 @@ func TestInMemoryStore_UpdateTxForRebroadcast(t *testing.T) {
 			assert.Nil(t, actErr)
 			inTx.FromAddress = fromAddress
 		})
+	})
+}
+
+func TestInMemoryStore_MarkAllConfirmedMissingReceipt(t *testing.T) {
+	t.Parallel()
+
+	t.Run("successfully mark all confirmed missing receipt", func(t *testing.T) {
+		db := pgtest.NewSqlxDB(t)
+		_, dbcfg, evmcfg := evmtxmgr.MakeTestConfigs(t)
+		persistentStore := cltest.NewTestTxStore(t, db)
+		kst := cltest.NewKeyStore(t, db, dbcfg)
+		_, fromAddress := cltest.MustInsertRandomKey(t, kst.Eth())
 
+		ethClient := evmtest.NewEthClientMockWithDefaultChain(t)
+		lggr := logger.TestSugared(t)
+		chainID := ethClient.ConfiguredChainID()
+		ctx := testutils.Context(t)
+
+		inMemoryStore, err := commontxmgr.NewInMemoryStore[
+			*big.Int,
+			common.Address, common.Hash, common.Hash,
+			*evmtypes.Receipt,
+			evmtypes.Nonce,
+			evmgas.EvmFee,
+		](ctx, lggr, chainID, kst.Eth(), persistentStore, evmcfg.Transactions())
+		require.NoError(t, err)
+
+		// create transaction 0 that is unconfirmed (block 7)
+		// Insert a transaction into persistent store
+		blocknum := int64(7)
+		inTx_0 := cltest.MustInsertUnconfirmedEthTx(t, persistentStore, 0, fromAddress)
+		inTxAttempt_0 := newBroadcastLegacyEthTxAttempt(t, inTx_0.ID, int64(1))
+		inTxAttempt_0.BroadcastBeforeBlockNum = &blocknum
+		require.NoError(t, persistentStore.InsertTxAttempt(ctx, &inTxAttempt_0))
+		assert.Equal(t, commontxmgr.TxUnconfirmed, inTx_0.State)
+		// Insert the transaction into the in-memory store
+		inTx_0.TxAttempts = []evmtxmgr.TxAttempt{inTxAttempt_0}
+		require.NoError(t, inMemoryStore.XXXTestInsertTx(fromAddress, &inTx_0))
+
+		// create transaction 1 that is confirmed (block 77)
+		inTx_1 := mustInsertConfirmedEthTxBySaveFetchedReceipts(t, persistentStore, fromAddress, 1, 77, *chainID)
+		assert.Equal(t, commontxmgr.TxConfirmed, inTx_1.State)
+		// Insert the transaction into the in-memory store
+		require.NoError(t, inMemoryStore.XXXTestInsertTx(fromAddress, &inTx_1))
+
+		// mark transaction 0 as confirmed missing receipt
+		err = inMemoryStore.MarkAllConfirmedMissingReceipt(ctx, chainID)
+		require.NoError(t, err)
+
+		expTx, err := persistentStore.FindTxWithAttempts(ctx, inTx_0.ID)
+		require.NoError(t, err)
+
+		fn := func(tx *evmtxmgr.Tx) bool { return true }
+		actTxs := inMemoryStore.XXXTestFindTxs(nil, fn, inTx_0.ID)
+		require.Equal(t, 1, len(actTxs))
+		actTx := actTxs[0]
+		assert.Equal(t, commontxmgr.TxConfirmedMissingReceipt, actTx.State)
+		assert.Equal(t, txmgrtypes.TxAttemptBroadcast, actTx.TxAttempts[0].State)
+		assertTxEqual(t, expTx, actTx)
 	})
 }