TXM In-memory: step 3-02-CreateTransaction

common/txmgr/inmemory_store.go

@@ -110,7 +110,29 @@ func (ms *inMemoryStore[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE]) Creat
 	txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE],
 	error,
 ) {
-	return txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]{}, nil
+	tx := txmgrtypes.Tx[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]{}
+	if ms.chainID.String() != chainID.String() {
+		panic("invalid chain ID")
+	}
+
+	ms.addressStatesLock.Lock()
+	as, ok := ms.addressStates[txRequest.FromAddress]
+	if !ok {
+		as = newAddressState[CHAIN_ID, ADDR, TX_HASH, BLOCK_HASH, R, SEQ, FEE](ms.lggr, chainID, txRequest.FromAddress, ms.maxUnstarted, nil)
+		ms.addressStates[txRequest.FromAddress] = as
+	}
+	ms.addressStatesLock.Unlock()
+
+	// Persist Transaction to persistent storage
+	tx, err := ms.persistentTxStore.CreateTransaction(ctx, txRequest, chainID)
+	if err != nil {
+		return tx, fmt.Errorf("create_transaction: %w", err)
+	}
+
+	// Update in memory store
+	// Add the request to the Unstarted channel to be processed by the Broadcaster
+	as.addTxToUnstartedQueue(&tx)
+	return *ms.deepCopyTx(tx), nil
 }
 
 // FindTxWithIdempotencyKey returns a transaction with the given idempotency key

core/chains/evm/txmgr/evm_inmemory_store_test.go

@@ -14,6 +14,7 @@
 	commontxmgr "github.com/smartcontractkit/chainlink/v2/common/txmgr"
 	txmgrtypes "github.com/smartcontractkit/chainlink/v2/common/txmgr/types"
 
+	evmassets "github.com/smartcontractkit/chainlink/v2/core/chains/evm/assets"
 	evmgas "github.com/smartcontractkit/chainlink/v2/core/chains/evm/gas"
 	evmtxmgr "github.com/smartcontractkit/chainlink/v2/core/chains/evm/txmgr"
 	evmtypes "github.com/smartcontractkit/chainlink/v2/core/chains/evm/types"
@@ -23,6 +24,69 @@
 	"github.com/smartcontractkit/chainlink/v2/core/internal/testutils/pgtest"
 )
 
+func TestInMemoryStore_CreateTransaction(t *testing.T) {
+	t.Parallel()
+
+	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)
+
+	toAddress := testutils.NewAddress()
+	gasLimit := uint32(1000)
+	payload := []byte{1, 2, 3}
+
+	t.Run("with queue under capacity inserts eth_tx", func(t *testing.T) {
+		subject := uuid.New()
+		strategy := newMockTxStrategy(t)
+		strategy.On("Subject").Return(uuid.NullUUID{UUID: subject, Valid: true})
+		actTx, err := inMemoryStore.CreateTransaction(ctx, evmtxmgr.TxRequest{
+			FromAddress:    fromAddress,
+			ToAddress:      toAddress,
+			EncodedPayload: payload,
+			FeeLimit:       uint64(gasLimit),
+			Meta:           nil,
+			Strategy:       strategy,
+		}, chainID)
+		require.NoError(t, err)
+
+		// check that the transaction was inserted into the persistent store
+		cltest.AssertCount(t, db, "evm.txes", 1)
+
+		var dbEthTx evmtxmgr.DbEthTx
+		require.NoError(t, db.Get(&dbEthTx, `SELECT * FROM evm.txes ORDER BY id ASC LIMIT 1`))
+
+		assert.Equal(t, commontxmgr.TxUnstarted, dbEthTx.State)
+		assert.Equal(t, gasLimit, dbEthTx.GasLimit)
+		assert.Equal(t, fromAddress, dbEthTx.FromAddress)
+		assert.Equal(t, toAddress, dbEthTx.ToAddress)
+		assert.Equal(t, payload, dbEthTx.EncodedPayload)
+		assert.Equal(t, evmassets.NewEthValue(0), dbEthTx.Value)
+		assert.Equal(t, subject, dbEthTx.Subject.UUID)
+
+		var expTx evmtxmgr.Tx
+		dbEthTx.ToTx(&expTx)
+
+		// check that the in-memory store has the same transaction data as the persistent store
+		assertTxEqual(t, expTx, actTx)
+	})
+}
+
 func TestInMemoryStore_PruneUnstartedTxQueue(t *testing.T) {
 	t.Parallel()
 
@@ -50,7 +114,7 @@
 		maxQueueSize := uint32(5)
 		nTxs := 3
 		subject := uuid.NullUUID{UUID: uuid.New(), Valid: true}
 		strat := commontxmgr.NewDropOldestStrategy(subject.UUID, maxQueueSize, dbcfg.DefaultQueryTimeout())
 		for i := 0; i < nTxs; i++ {
 			inTx := cltest.NewEthTx(fromAddress)
 			inTx.Subject = subject
@@ -87,7 +151,7 @@
 		maxQueueSize := uint32(5)
 		nTxs := 5
 		subject := uuid.NullUUID{UUID: uuid.New(), Valid: true}
 		strat := commontxmgr.NewDropOldestStrategy(subject.UUID, maxQueueSize, dbcfg.DefaultQueryTimeout())
 		for i := 0; i < nTxs; i++ {
 			inTx := cltest.NewEthTx(fromAddress)
 			inTx.Subject = subject
@@ -99,7 +163,7 @@
 
 		ids, err := strat.PruneQueue(ctx, inMemoryStore)
 		require.NoError(t, err)
 		assert.Equal(t, int(nTxs)-int(maxQueueSize-1), len(ids))
 
 		AssertCountPerSubject(t, persistentStore, int64(maxQueueSize-1), subject.UUID)
 		fn := func(tx *evmtxmgr.Tx) bool { return true }