txmgr: Add custom publish error message handling (#13856)

* txmgr: Add custom publish error message handling

Fixes #13739

* add prefix TXMGR to env var
Co-authored-by: protolambda <proto@protolambda.com>

---------
Co-authored-by: protolambda <proto@protolambda.com>

op-service/txmgr/cli.go

@@ -38,6 +38,7 @@ const (
 	TxSendTimeoutFlagName              = "txmgr.send-timeout"
 	TxNotInMempoolTimeoutFlagName      = "txmgr.not-in-mempool-timeout"
 	ReceiptQueryIntervalFlagName       = "txmgr.receipt-query-interval"
+	AlreadyPublishedCustomErrsFlagName = "txmgr.already-published-custom-errs"
 )
 
 var (
@@ -191,6 +192,11 @@ func CLIFlagsWithDefaults(envPrefix string, defaults DefaultFlagValues) []cli.Fl
 			Value:   defaults.ReceiptQueryInterval,
 			EnvVars: prefixEnvVars("TXMGR_RECEIPT_QUERY_INTERVAL"),
 		},
+		&cli.StringSliceFlag{
+			Name:    AlreadyPublishedCustomErrsFlagName,
+			Usage:   "List of custom RPC error messages that indicate that a transaction has already been published.",
+			EnvVars: prefixEnvVars("TXMGR_ALREADY_PUBLISHED_CUSTOM_ERRS"),
+		},
 	}, opsigner.CLIFlags(envPrefix, "")...)
 }
 
@@ -213,6 +219,7 @@ type CLIConfig struct {
 	NetworkTimeout             time.Duration
 	TxSendTimeout              time.Duration
 	TxNotInMempoolTimeout      time.Duration
+	AlreadyPublishedCustomErrs []string
 }
 
 func NewCLIConfig(l1RPCURL string, defaults DefaultFlagValues) CLIConfig {
@@ -288,6 +295,7 @@ func ReadCLIConfig(ctx *cli.Context) CLIConfig {
 		NetworkTimeout:             ctx.Duration(NetworkTimeoutFlagName),
 		TxSendTimeout:              ctx.Duration(TxSendTimeoutFlagName),
 		TxNotInMempoolTimeout:      ctx.Duration(TxNotInMempoolTimeoutFlagName),
+		AlreadyPublishedCustomErrs: ctx.StringSlice(AlreadyPublishedCustomErrsFlagName),
 	}
 }
 
@@ -341,14 +349,16 @@ func NewConfig(cfg CLIConfig, l log.Logger) (*Config, error) {
 	res := Config{
 		Backend: l1,
 		ChainID: chainID,
+		Signer:  signerFactory(chainID),
+		From:    from,
+
 		TxSendTimeout:              cfg.TxSendTimeout,
 		TxNotInMempoolTimeout:      cfg.TxNotInMempoolTimeout,
 		NetworkTimeout:             cfg.NetworkTimeout,
 		ReceiptQueryInterval:       cfg.ReceiptQueryInterval,
 		NumConfirmations:           cfg.NumConfirmations,
 		SafeAbortNonceTooLowCount:  cfg.SafeAbortNonceTooLowCount,
-		Signer:                    signerFactory(chainID),
-		From:                      from,
+		AlreadyPublishedCustomErrs: cfg.AlreadyPublishedCustomErrs,
 	}
 
 	res.ResubmissionTimeout.Store(int64(cfg.ResubmissionTimeout))
@@ -422,6 +432,10 @@ type Config struct {
 	// GasPriceEstimatorFn is used to estimate the gas price for a transaction.
 	// If nil, DefaultGasPriceEstimatorFn is used.
 	GasPriceEstimatorFn GasPriceEstimatorFn
+
+	// List of custom RPC error messages that indicate that a transaction has
+	// already been published.
+	AlreadyPublishedCustomErrs []string
 }
 
 func (m *Config) Check() error {

op-service/txmgr/txmgr.go

@@ -638,9 +638,14 @@ func (m *SimpleTxManager) publishTx(ctx context.Context, tx *types.Transaction,
 		cancel()
 		sendState.ProcessSendError(err)
 
-		if err == nil {
+		if err == nil || errStringContainsAny(err, m.cfg.AlreadyPublishedCustomErrs) {
+			// only empty error strings are recorded as successful publishes
 			m.metr.TxPublished("")
+			if err == nil {
 				l.Info("Transaction successfully published", "tx", tx.Hash())
+			} else {
+				l.Info("Transaction successfully published (custom RPC error)", "tx", tx.Hash(), "err", err)
+			}
 			// Tx made it into the mempool, so we'll need a fee bump if we end up trying to replace
 			// it with another publish attempt.
 			sendState.bumpFees = true
@@ -1053,6 +1058,19 @@ func errStringMatch(err, target error) bool {
 	return strings.Contains(err.Error(), target.Error())
 }
 
+func errStringContainsAny(err error, targets []string) bool {
+	if err == nil || len(targets) == 0 {
+		return false
+	}
+
+	for _, target := range targets {
+		if strings.Contains(err.Error(), target) {
+			return true
+		}
+	}
+	return false
+}
+
 // finishBlobTx finishes creating a blob tx message by safely converting bigints to uint256
 func finishBlobTx(message *types.BlobTx, chainID, tip, fee, blobFee, value *big.Int) error {
 	var o bool

op-service/txmgr/txmgr_test.go

@@ -1581,6 +1581,29 @@ func TestCloseWaitingForConfirmation(t *testing.T) {
 	})
 }
 
+func TestTxMgrCustomPublishError(t *testing.T) {
+	customErr := errors.New("custom test error")
+
+	testSendVariants(t, func(t *testing.T, send testSendVariantsFn) {
+		cfg := configWithNumConfs(1)
+		cfg.AlreadyPublishedCustomErrs = []string{customErr.Error()}
+		h := newTestHarnessWithConfig(t, cfg)
+
+		sendTx := func(ctx context.Context, tx *types.Transaction) error {
+			txHash := tx.Hash()
+			h.backend.mine(&txHash, tx.GasFeeCap(), nil)
+			return customErr
+		}
+		h.backend.setTxSender(sendTx)
+
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		receipt, err := send(ctx, h, h.createTxCandidate())
+		require.Nil(t, err)
+		require.NotNil(t, receipt)
+	})
+}
+
 func TestMakeSidecar(t *testing.T) {
 	var blob eth.Blob
 	_, err := rand.Read(blob[:])