Add SendAsync to TxMgr (#11843)

* Add SendAsync to TxMgr

Adds a SendAsync method to TxMgr. I'd like to use this for `op-deployer`, which needs to send multiple transactions in parallel but with predictable nonces. `SendAsync` returns a channel that resolves with the result of each send, but synchronously increases the nonce and prepares the first send prior to returning.

* review updates + tests

op-challenger/sender/sender_test.go

@@ -128,6 +128,10 @@ func (s *stubTxMgr) Send(ctx context.Context, candidate txmgr.TxCandidate) (*typ
 	return <-ch, nil
 }
 
+func (s *stubTxMgr) SendAsync(ctx context.Context, candidate txmgr.TxCandidate, ch chan txmgr.SendResponse) {
+	panic("unimplemented")
+}
+
 func (s *stubTxMgr) recordTx(candidate txmgr.TxCandidate) chan *types.Receipt {
 	s.m.Lock()
 	defer s.m.Unlock()

op-e2e/actions/l2_proposer.go

@@ -69,6 +69,10 @@ func (f fakeTxMgr) Send(_ context.Context, _ txmgr.TxCandidate) (*types.Receipt,
 	panic("unimplemented")
 }
 
+func (f fakeTxMgr) SendAsync(ctx context.Context, candidate txmgr.TxCandidate, ch chan txmgr.SendResponse) {
+	panic("unimplemented")
+}
+
 func (f fakeTxMgr) Close() {
 }
 

op-service/txmgr/mocks/TxManager.go

@@ -120,6 +120,11 @@ func (_m *TxManager) Send(ctx context.Context, candidate txmgr.TxCandidate) (*ty
 	return r0, r1
 }
 
+// SendAsync provides a mock function with given fields: ctx, candidate, ch
+func (_m *TxManager) SendAsync(ctx context.Context, candidate txmgr.TxCandidate, ch chan txmgr.SendResponse) {
+	_m.Called(ctx, candidate, ch)
+}
+
 type mockConstructorTestingTNewTxManager interface {
 	mock.TestingT
 	Cleanup(func())

op-service/txmgr/txmgr.go

@@ -46,6 +46,12 @@ var (
 	ErrClosed       = errors.New("transaction manager is closed")
 )
 
+type SendResponse struct {
+	Receipt *types.Receipt
+	Nonce   uint64
+	Err     error
+}
+
 // TxManager is an interface that allows callers to reliably publish txs,
 // bumping the gas price if needed, and obtain the receipt of the resulting tx.
 //
@@ -63,6 +69,14 @@ type TxManager interface {
 	// mempool and is in need of replacement or cancellation.
 	Send(ctx context.Context, candidate TxCandidate) (*types.Receipt, error)
 
+	// SendAsync is used to create & send a transaction asynchronously. It has similar internal
+	// semantics to Send, however it returns a channel that will receive the result of the
+	// send operation once it completes. Transactions crafted synchronously - that is, nonce
+	// management and gas estimation happen prior to the method returning. This allows callers
+	// that rely on predictable nonces to send multiple transactions in parallel while preserving
+	// the order of nonce increments.
+	SendAsync(ctx context.Context, candidate TxCandidate, ch chan SendResponse)
+
 	// From returns the sending address associated with the instance of the transaction manager.
 	// It is static for a single instance of a TxManager.
 	From() common.Address
@@ -222,24 +236,83 @@ func (m *SimpleTxManager) Send(ctx context.Context, candidate TxCandidate) (*typ
 	if m.closed.Load() {
 		return nil, ErrClosed
 	}
+
 	m.metr.RecordPendingTx(m.pending.Add(1))
-	defer func() {
-		m.metr.RecordPendingTx(m.pending.Add(-1))
-	}()
-	receipt, err := m.send(ctx, candidate)
+	defer m.metr.RecordPendingTx(m.pending.Add(-1))
+
+	var cancel context.CancelFunc
+	if m.cfg.TxSendTimeout == 0 {
+		ctx, cancel = context.WithCancel(ctx)
+	} else {
+		ctx, cancel = context.WithTimeout(ctx, m.cfg.TxSendTimeout)
+	}
+	defer cancel()
+
+	tx, err := m.prepare(ctx, candidate)
 	if err != nil {
 		m.resetNonce()
+		return nil, err
+	}
+	receipt, err := m.sendTx(ctx, tx)
+	if err != nil {
+		m.resetNonce()
+		return nil, err
 	}
 	return receipt, err
 }
 
-// send performs the actual transaction creation and sending.
-func (m *SimpleTxManager) send(ctx context.Context, candidate TxCandidate) (*types.Receipt, error) {
-	if m.cfg.TxSendTimeout != 0 {
-		var cancel context.CancelFunc
+func (m *SimpleTxManager) SendAsync(ctx context.Context, candidate TxCandidate, ch chan SendResponse) {
+	if cap(ch) == 0 {
+		panic("SendAsync: channel must be buffered")
+	}
+
+	// refuse new requests if the tx manager is closed
+	if m.closed.Load() {
+		ch <- SendResponse{
+			Receipt: nil,
+			Err:     ErrClosed,
+		}
+		return
+	}
+
+	m.metr.RecordPendingTx(m.pending.Add(1))
+
+	var cancel context.CancelFunc
+	if m.cfg.TxSendTimeout == 0 {
+		ctx, cancel = context.WithCancel(ctx)
+	} else {
 		ctx, cancel = context.WithTimeout(ctx, m.cfg.TxSendTimeout)
-		defer cancel()
 	}
+
+	tx, err := m.prepare(ctx, candidate)
+	if err != nil {
+		m.resetNonce()
+		cancel()
+		m.metr.RecordPendingTx(m.pending.Add(-1))
+		ch <- SendResponse{
+			Receipt: nil,
+			Err:     err,
+		}
+		return
+	}
+
+	go func() {
+		defer m.metr.RecordPendingTx(m.pending.Add(-1))
+		defer cancel()
+		receipt, err := m.sendTx(ctx, tx)
+		if err != nil {
+			m.resetNonce()
+		}
+		ch <- SendResponse{
+			Receipt: receipt,
+			Nonce:   tx.Nonce(),
+			Err:     err,
+		}
+	}()
+}
+
+// prepare prepares the transaction for sending.
+func (m *SimpleTxManager) prepare(ctx context.Context, candidate TxCandidate) (*types.Transaction, error) {
 	tx, err := retry.Do(ctx, 30, retry.Fixed(2*time.Second), func() (*types.Transaction, error) {
 		if m.closed.Load() {
 			return nil, ErrClosed
@@ -253,7 +326,7 @@ func (m *SimpleTxManager) send(ctx context.Context, candidate TxCandidate) (*typ
 	if err != nil {
 		return nil, fmt.Errorf("failed to create the tx: %w", err)
 	}
-	return m.sendTx(ctx, tx)
+	return tx, nil
 }
 
 // craftTx creates the signed transaction

op-service/txmgr/txmgr_test.go

@@ -338,6 +338,27 @@ func (b *mockBackend) TransactionReceipt(ctx context.Context, txHash common.Hash
 func (b *mockBackend) Close() {
 }
 
+type testSendVariantsFn func(ctx context.Context, h *testHarness, tx TxCandidate) (*types.Receipt, error)
+
+func testSendVariants(t *testing.T, testFn func(t *testing.T, send testSendVariantsFn)) {
+	t.Parallel()
+
+	t.Run("Send", func(t *testing.T) {
+		testFn(t, func(ctx context.Context, h *testHarness, tx TxCandidate) (*types.Receipt, error) {
+			return h.mgr.Send(ctx, tx)
+		})
+	})
+
+	t.Run("SendAsync", func(t *testing.T) {
+		testFn(t, func(ctx context.Context, h *testHarness, tx TxCandidate) (*types.Receipt, error) {
+			ch := make(chan SendResponse, 1)
+			h.mgr.SendAsync(ctx, tx, ch)
+			res := <-ch
+			return res.Receipt, res.Err
+		})
+	})
+}
+
 // TestTxMgrConfirmAtMinGasPrice asserts that Send returns the min gas price tx
 // if the tx is mined instantly.
 func TestTxMgrConfirmAtMinGasPrice(t *testing.T) {
@@ -400,32 +421,32 @@ func TestTxMgrNeverConfirmCancel(t *testing.T) {
 // TestTxMgrTxSendTimeout tests that the TxSendTimeout is respected when trying to send a
 // transaction, even if NetworkTimeout expires first.
 func TestTxMgrTxSendTimeout(t *testing.T) {
-	t.Parallel()
-
-	conf := configWithNumConfs(1)
-	conf.TxSendTimeout = 3 * time.Second
-	conf.NetworkTimeout = 1 * time.Second
+	testSendVariants(t, func(t *testing.T, send testSendVariantsFn) {
+		conf := configWithNumConfs(1)
+		conf.TxSendTimeout = 3 * time.Second
+		conf.NetworkTimeout = 1 * time.Second
+
+		h := newTestHarnessWithConfig(t, conf)
+
+		txCandidate := h.createTxCandidate()
+		sendCount := 0
+		sendTx := func(ctx context.Context, tx *types.Transaction) error {
+			sendCount++
+			<-ctx.Done()
+			return context.DeadlineExceeded
+		}
+		h.backend.setTxSender(sendTx)
 
-	h := newTestHarnessWithConfig(t, conf)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
+		defer cancel()
 
-	txCandidate := h.createTxCandidate()
-	sendCount := 0
-	sendTx := func(ctx context.Context, tx *types.Transaction) error {
-		sendCount++
-		<-ctx.Done()
-		return context.DeadlineExceeded
-	}
-	h.backend.setTxSender(sendTx)
-
-	ctx, cancel := context.WithTimeout(context.Background(), time.Hour)
-	defer cancel()
-
-	receipt, err := h.mgr.send(ctx, txCandidate)
-	require.ErrorIs(t, err, context.DeadlineExceeded)
-	// Because network timeout is much shorter than send timeout, we should see multiple send attempts
-	// before the overall send fails.
-	require.Greater(t, sendCount, 1)
-	require.Nil(t, receipt)
+		receipt, err := send(ctx, h, txCandidate)
+		require.ErrorIs(t, err, context.DeadlineExceeded)
+		// Because network timeout is much shorter than send timeout, we should see multiple send attempts
+		// before the overall send fails.
+		require.Greater(t, sendCount, 1)
+		require.Nil(t, receipt)
+	})
 }
 
 // TestAlreadyReserved tests that AlreadyReserved error results in immediate abort of transaction
@@ -1326,40 +1347,42 @@ func TestErrStringMatch(t *testing.T) {
 }
 
 func TestNonceReset(t *testing.T) {
-	conf := configWithNumConfs(1)
-	conf.SafeAbortNonceTooLowCount = 1
-	h := newTestHarnessWithConfig(t, conf)
-
-	index := -1
-	var nonces []uint64
-	sendTx := func(ctx context.Context, tx *types.Transaction) error {
-		index++
-		nonces = append(nonces, tx.Nonce())
-		// fail every 3rd tx
-		if index%3 == 0 {
-			return core.ErrNonceTooLow
+	testSendVariants(t, func(t *testing.T, send testSendVariantsFn) {
+		conf := configWithNumConfs(1)
+		conf.SafeAbortNonceTooLowCount = 1
+		h := newTestHarnessWithConfig(t, conf)
+
+		index := -1
+		var nonces []uint64
+		sendTx := func(ctx context.Context, tx *types.Transaction) error {
+			index++
+			nonces = append(nonces, tx.Nonce())
+			// fail every 3rd tx
+			if index%3 == 0 {
+				return core.ErrNonceTooLow
+			}
+			txHash := tx.Hash()
+			h.backend.mine(&txHash, tx.GasFeeCap(), nil)
+			return nil
 		}
-		txHash := tx.Hash()
-		h.backend.mine(&txHash, tx.GasFeeCap(), nil)
-		return nil
-	}
-	h.backend.setTxSender(sendTx)
+		h.backend.setTxSender(sendTx)
 
-	ctx := context.Background()
-	for i := 0; i < 8; i++ {
-		_, err := h.mgr.Send(ctx, TxCandidate{
-			To: &common.Address{},
-		})
-		// expect every 3rd tx to fail
-		if i%3 == 0 {
-			require.Error(t, err)
-		} else {
-			require.NoError(t, err)
+		ctx := context.Background()
+		for i := 0; i < 8; i++ {
+			_, err := send(ctx, h, TxCandidate{
+				To: &common.Address{},
+			})
+			// expect every 3rd tx to fail
+			if i%3 == 0 {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+			}
 		}
-	}
 
-	// internal nonce tracking should be reset to startingNonce value every 3rd tx
-	require.Equal(t, []uint64{1, 1, 2, 3, 1, 2, 3, 1}, nonces)
+		// internal nonce tracking should be reset to startingNonce value every 3rd tx
+		require.Equal(t, []uint64{1, 1, 2, 3, 1, 2, 3, 1}, nonces)
+	})
 }
 
 func TestMinFees(t *testing.T) {
@@ -1431,114 +1454,118 @@ func TestMinFees(t *testing.T) {
 
 // TestClose ensures that the tx manager will refuse new work and cancel any in progress
 func TestClose(t *testing.T) {
-	conf := configWithNumConfs(1)
-	h := newTestHarnessWithConfig(t, conf)
-
-	sendingSignal := make(chan struct{})
-
-	// Ensure the manager is not closed
-	require.False(t, h.mgr.closed.Load())
-
-	// sendTx will fail until it is called a retry-number of times
-	called := 0
-	const retries = 4
-	sendTx := func(ctx context.Context, tx *types.Transaction) (err error) {
-		called += 1
-		// sendingSignal is used when the tx begins to be sent
-		if called == 1 {
-			sendingSignal <- struct{}{}
-		}
-		if called%retries == 0 {
-			txHash := tx.Hash()
-			h.backend.mine(&txHash, tx.GasFeeCap(), big.NewInt(1))
-		} else {
-			time.Sleep(10 * time.Millisecond)
-			err = errRpcFailure
+	testSendVariants(t, func(t *testing.T, send testSendVariantsFn) {
+		conf := configWithNumConfs(1)
+		h := newTestHarnessWithConfig(t, conf)
+
+		sendingSignal := make(chan struct{})
+
+		// Ensure the manager is not closed
+		require.False(t, h.mgr.closed.Load())
+
+		// sendTx will fail until it is called a retry-number of times
+		called := 0
+		const retries = 4
+		sendTx := func(ctx context.Context, tx *types.Transaction) (err error) {
+			called += 1
+			// sendingSignal is used when the tx begins to be sent
+			if called == 1 {
+				sendingSignal <- struct{}{}
+			}
+			if called%retries == 0 {
+				txHash := tx.Hash()
+				h.backend.mine(&txHash, tx.GasFeeCap(), big.NewInt(1))
+			} else {
+				time.Sleep(10 * time.Millisecond)
+				err = errRpcFailure
+			}
+			return
 		}
-		return
-	}
-	h.backend.setTxSender(sendTx)
-
-	// on the first call, we don't use the sending signal but we still need to drain it
-	go func() {
-		<-sendingSignal
-	}()
-	// demonstrate that a tx is sent, even when it must retry repeatedly
-	ctx := context.Background()
-	_, err := h.mgr.Send(ctx, TxCandidate{
-		To: &common.Address{},
-	})
-	require.NoError(t, err)
-	require.Equal(t, retries, called)
-	called = 0
-	// Ensure the manager is *still* not closed
-	require.False(t, h.mgr.closed.Load())
-
-	// on the second call, we close the manager while the tx is in progress by consuming the sending signal
-	go func() {
-		<-sendingSignal
-		h.mgr.Close()
-	}()
-	// demonstrate that a tx will cancel if it is in progress when the manager is closed
-	_, err = h.mgr.Send(ctx, TxCandidate{
-		To: &common.Address{},
-	})
-	require.ErrorIs(t, err, ErrClosed)
-	// confirm that the tx was canceled before it retried to completion
-	require.Less(t, called, retries)
-	require.True(t, h.mgr.closed.Load())
-	called = 0
-
-	// demonstrate that new calls to Send will also fail when the manager is closed
-	// there should be no need to capture the sending signal here because the manager is already closed and will return immediately
-	_, err = h.mgr.Send(ctx, TxCandidate{
-		To: &common.Address{},
+		h.backend.setTxSender(sendTx)
+
+		// on the first call, we don't use the sending signal but we still need to drain it
+		go func() {
+			<-sendingSignal
+		}()
+		// demonstrate that a tx is sent, even when it must retry repeatedly
+		ctx := context.Background()
+		_, err := send(ctx, h, TxCandidate{
+			To: &common.Address{},
+		})
+		require.NoError(t, err)
+		require.Equal(t, retries, called)
+		called = 0
+		// Ensure the manager is *still* not closed
+		require.False(t, h.mgr.closed.Load())
+
+		// on the second call, we close the manager while the tx is in progress by consuming the sending signal
+		go func() {
+			<-sendingSignal
+			h.mgr.Close()
+		}()
+		// demonstrate that a tx will cancel if it is in progress when the manager is closed
+		_, err = send(ctx, h, TxCandidate{
+			To: &common.Address{},
+		})
+		require.ErrorIs(t, err, ErrClosed)
+		// confirm that the tx was canceled before it retried to completion
+		require.Less(t, called, retries)
+		require.True(t, h.mgr.closed.Load())
+		called = 0
+
+		// demonstrate that new calls to Send will also fail when the manager is closed
+		// there should be no need to capture the sending signal here because the manager is already closed and will return immediately
+		_, err = send(ctx, h, TxCandidate{
+			To: &common.Address{},
+		})
+		require.ErrorIs(t, err, ErrClosed)
+		// confirm that the tx was canceled before it ever made it to the backend
+		require.Equal(t, 0, called)
 	})
-	require.ErrorIs(t, err, ErrClosed)
-	// confirm that the tx was canceled before it ever made it to the backend
-	require.Equal(t, 0, called)
 }
 
 // TestCloseWaitingForConfirmation ensures that the tx manager will wait for confirmation of a tx in flight, even when closed
 func TestCloseWaitingForConfirmation(t *testing.T) {
-	// two confirmations required so that we can mine and not yet be fully confirmed
-	conf := configWithNumConfs(2)
-	h := newTestHarnessWithConfig(t, conf)
+	testSendVariants(t, func(t *testing.T, send testSendVariantsFn) {
+		// two confirmations required so that we can mine and not yet be fully confirmed
+		conf := configWithNumConfs(2)
+		h := newTestHarnessWithConfig(t, conf)
 
-	// sendDone is a signal that the tx has been sent from the sendTx function
-	sendDone := make(chan struct{})
-	// closeDone is a signal that the txmanager has closed
-	closeDone := make(chan struct{})
+		// sendDone is a signal that the tx has been sent from the sendTx function
+		sendDone := make(chan struct{})
+		// closeDone is a signal that the txmanager has closed
+		closeDone := make(chan struct{})
 
-	sendTx := func(ctx context.Context, tx *types.Transaction) error {
-		txHash := tx.Hash()
-		h.backend.mine(&txHash, tx.GasFeeCap(), big.NewInt(1))
-		close(sendDone)
-		return nil
-	}
-	h.backend.setTxSender(sendTx)
-
-	// this goroutine will close the manager when the tx sending is complete
-	// the transaction is not yet confirmed, so the manager will wait for confirmation
-	go func() {
-		<-sendDone
-		h.mgr.Close()
-		close(closeDone)
-	}()
-
-	// this goroutine will complete confirmation of the tx when the manager is closed
-	// by forcing this to happen after close, we are able to observe a closing manager waiting for confirmation
-	go func() {
-		<-closeDone
-		h.backend.mine(nil, nil, big.NewInt(1))
-	}()
-
-	ctx := context.Background()
-	_, err := h.mgr.Send(ctx, TxCandidate{
-		To: &common.Address{},
+		sendTx := func(ctx context.Context, tx *types.Transaction) error {
+			txHash := tx.Hash()
+			h.backend.mine(&txHash, tx.GasFeeCap(), big.NewInt(1))
+			close(sendDone)
+			return nil
+		}
+		h.backend.setTxSender(sendTx)
+
+		// this goroutine will close the manager when the tx sending is complete
+		// the transaction is not yet confirmed, so the manager will wait for confirmation
+		go func() {
+			<-sendDone
+			h.mgr.Close()
+			close(closeDone)
+		}()
+
+		// this goroutine will complete confirmation of the tx when the manager is closed
+		// by forcing this to happen after close, we are able to observe a closing manager waiting for confirmation
+		go func() {
+			<-closeDone
+			h.backend.mine(nil, nil, big.NewInt(1))
+		}()
+
+		ctx := context.Background()
+		_, err := send(ctx, h, TxCandidate{
+			To: &common.Address{},
+		})
+		require.True(t, h.mgr.closed.Load())
+		require.NoError(t, err)
 	})
-	require.True(t, h.mgr.closed.Load())
-	require.NoError(t, err)
 }
 
 func TestMakeSidecar(t *testing.T) {
@@ -1561,3 +1588,12 @@ func TestMakeSidecar(t *testing.T) {
 		require.Equal(t, hashes[i], eth.KZGToVersionedHash(commit))
 	}
 }
+
+func TestSendAsyncUnbufferedChan(t *testing.T) {
+	conf := configWithNumConfs(2)
+	h := newTestHarnessWithConfig(t, conf)
+
+	require.Panics(t, func() {
+		h.mgr.SendAsync(context.Background(), TxCandidate{}, make(chan SendResponse))
+	})
+}