INC-2328 Add supportsDualBroadcast to txmgr (#87)

Author: cll-gg

chains/txmgr/txmgr.go

@@ -61,6 +61,11 @@ type TxManager[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THA
 	GetTransactionFee(ctx context.Context, transactionID string) (fee *evmtypes.TransactionFee, err error)
 	GetTransactionReceipt(ctx context.Context, transactionID string) (receipt *txmgrtypes.ChainReceipt[THASH, BHASH], err error)
 	CalculateFee(feeParts FeeParts) *big.Int
+	// SupportsDualBroadcast reports whether this TXM will route transactions marked with
+	// DualBroadcast=true to an OFA rather than the public mempool.
+	// Jobs that configure a secondary EOA check this at startup and refuse to run if it
+	// returns false, preventing accidental public-mempool exposure of secondary transactions.
+	SupportsDualBroadcast() bool
 }
 
 type TxmV2Wrapper[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THASH chains.Hashable, BHASH chains.Hashable, SEQ chains.Sequence, FEE fees.Fee] interface {
@@ -98,16 +103,17 @@ type Txm[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THASH cha
 	chSubbed chan struct{}
 	wg       sync.WaitGroup
 
-	reaper             *Reaper[CID]
-	resender           *Resender[CID, ADDR, THASH, BHASH, R, SEQ, FEE]
-	broadcaster        *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]
-	confirmer          *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]
-	tracker            *Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]
-	finalizer          txmgrtypes.Finalizer[BHASH, HEAD]
-	fwdMgr             txmgrtypes.ForwarderManager[ADDR]
-	txAttemptBuilder   txmgrtypes.TxAttemptBuilder[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]
-	newErrorClassifier NewErrorClassifier
-	txmv2wrapper       TxmV2Wrapper[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]
+	reaper               *Reaper[CID]
+	resender             *Resender[CID, ADDR, THASH, BHASH, R, SEQ, FEE]
+	broadcaster          *Broadcaster[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]
+	confirmer            *Confirmer[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]
+	tracker              *Tracker[CID, ADDR, THASH, BHASH, R, SEQ, FEE]
+	finalizer            txmgrtypes.Finalizer[BHASH, HEAD]
+	fwdMgr               txmgrtypes.ForwarderManager[ADDR]
+	txAttemptBuilder     txmgrtypes.TxAttemptBuilder[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]
+	newErrorClassifier   NewErrorClassifier
+	txmv2wrapper         TxmV2Wrapper[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]
+	dualBroadcastEnabled bool
 
 	enabledAddrs []ADDR // sorted as strings
 }
@@ -137,29 +143,31 @@ func NewTxm[CID chains.ID, HEAD chains.Head[BHASH], ADDR chains.Hashable, THASH
 	finalizer txmgrtypes.Finalizer[BHASH, HEAD],
 	newErrorClassifierFunc NewErrorClassifier,
 	txmv2wrapper TxmV2Wrapper[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE],
+	dualBroadcastEnabled bool,
 ) *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE] {
 	b := Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]{
-		logger:             logger.Sugared(lggr),
-		txStore:            txStore,
-		config:             cfg,
-		txConfig:           txCfg,
-		keyStore:           keyStore,
-		chainID:            chainID,
-		checkerFactory:     checkerFactory,
-		chHeads:            make(chan HEAD),
-		trigger:            make(chan ADDR),
-		chStop:             make(chan struct{}),
-		chSubbed:           make(chan struct{}),
-		reset:              make(chan reset),
-		fwdMgr:             fwdMgr,
-		txAttemptBuilder:   txAttemptBuilder,
-		broadcaster:        broadcaster,
-		confirmer:          confirmer,
-		resender:           resender,
-		tracker:            tracker,
-		newErrorClassifier: newErrorClassifierFunc,
-		finalizer:          finalizer,
-		txmv2wrapper:       txmv2wrapper,
+		logger:               logger.Sugared(lggr),
+		txStore:              txStore,
+		config:               cfg,
+		txConfig:             txCfg,
+		keyStore:             keyStore,
+		chainID:              chainID,
+		checkerFactory:       checkerFactory,
+		chHeads:              make(chan HEAD),
+		trigger:              make(chan ADDR),
+		chStop:               make(chan struct{}),
+		chSubbed:             make(chan struct{}),
+		reset:                make(chan reset),
+		fwdMgr:               fwdMgr,
+		txAttemptBuilder:     txAttemptBuilder,
+		broadcaster:          broadcaster,
+		confirmer:            confirmer,
+		resender:             resender,
+		tracker:              tracker,
+		newErrorClassifier:   newErrorClassifierFunc,
+		finalizer:            finalizer,
+		txmv2wrapper:         txmv2wrapper,
+		dualBroadcastEnabled: dualBroadcastEnabled,
 	}
 
 	if txCfg.ResendAfterThreshold() <= 0 {
@@ -603,6 +611,13 @@ func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetForwarderForEOA(ctx
 	return
 }
 
+// SupportsDualBroadcast reports whether this TXM will route DualBroadcast transactions to an
+// OFA rather than the public mempool. The value is set at construction time by the
+// txmgr builder based on node config.
+func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) SupportsDualBroadcast() bool {
+	return b.dualBroadcastEnabled
+}
+
 // GetForwarderForEOAOCR2Feeds calls forwarderMgr to get a proper forwarder for a given EOA and checks if its set as a transmitter on the OCR2Aggregator contract.
 func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) GetForwarderForEOAOCR2Feeds(ctx context.Context, eoa, ocr2Aggregator ADDR) (forwarder ADDR, err error) {
 	if !b.txConfig.ForwardersEnabled() {
@@ -881,6 +896,10 @@ func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) GetTransactionR
 	return
 }
 
+func (n *NullTxManager[CID, HEAD, ADDR, THASH, BHASH, SEQ, FEE]) SupportsDualBroadcast() bool {
+	return false
+}
+
 func (b *Txm[CID, HEAD, ADDR, THASH, BHASH, R, SEQ, FEE]) pruneQueueAndCreateTxn(
 	ctx context.Context,
 	txRequest txmgrtypes.TxRequest[ADDR, THASH],

chains/txmgr/txmgrtest/err.go

@@ -99,3 +99,7 @@ func (n *ErrTxManager[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) Coun
 func (n *ErrTxManager[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) GetTransactionStatus(ctx context.Context, transactionID string) (status commontypes.TransactionStatus, err error) {
 	return
 }
+
+func (n *ErrTxManager[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) SupportsDualBroadcast() bool {
+	return false
+}