feat(relayer): Allow relayer to run in "target single transaction has…

…h" mode (#15546)

Co-authored-by: RogerLamTd <[email protected]>

packages/relayer/bridge.go

@@ -26,4 +26,5 @@ type Bridge interface {
 		sink chan<- *bridge.BridgeMessageStatusChanged,
 		msgHash [][32]byte,
 	) (event.Subscription, error)
+	ParseMessageSent(log types.Log) (*bridge.BridgeMessageSent, error)
 }

packages/relayer/cmd/flags/processor.go

@@ -126,6 +126,13 @@ var (
 		Category: processorCategory,
 		EnvVars:  []string{"HOP_RPC_URLS"},
 	}
+	TargetTxHash = &cli.StringFlag{
+		Name:     "targetTxHash",
+		Usage:    "Target transaction hash, set to ignore processing from queue and only process this individual transaction",
+		Category: processorCategory,
+		Required: false,
+		EnvVars:  []string{"TARGET_TX_HASH"},
+	}
 )
 
 var ProcessorFlags = MergeFlags(CommonFlags, QueueFlags, []cli.Flag{
@@ -148,4 +155,5 @@ var ProcessorFlags = MergeFlags(CommonFlags, QueueFlags, []cli.Flag{
 	HopSignalServiceAddresses,
 	HopTaikoAddresses,
 	DestBridgeAddress,
+	TargetTxHash,
 })

packages/relayer/pkg/mock/bridge.go

@@ -149,3 +149,7 @@ func (b *Bridge) ProveMessageReceived(opts *bind.CallOpts, message bridge.IBridg
 
 	return false, nil
 }
+
+func (b *Bridge) ParseMessageSent(log types.Log) (*bridge.BridgeMessageSent, error) {
+	return &bridge.BridgeMessageSent{}, nil
+}

packages/relayer/processor/config.go

@@ -33,6 +33,8 @@ type Config struct {
 	// private key
 	ProcessorPrivateKey *ecdsa.PrivateKey
 
+	TargetTxHash *common.Hash
+
 	// processing configs
 	HeaderSyncInterval   uint64
 	Confirmations        uint64
@@ -96,6 +98,13 @@ func NewConfigFromCliContext(c *cli.Context) (*Config, error) {
 		})
 	}
 
+	var targetTxHash *common.Hash
+
+	if c.IsSet(flags.TargetTxHash.Name) {
+		hash := common.HexToHash(c.String(flags.TargetTxHash.Name))
+		targetTxHash = &hash
+	}
+
 	return &Config{
 		hopConfigs:              hopConfigs,
 		ProcessorPrivateKey:     processorPrivateKey,
@@ -127,6 +136,7 @@ func NewConfigFromCliContext(c *cli.Context) (*Config, error) {
 		BackoffRetryInterval:    c.Uint64(flags.BackOffRetryInterval.Name),
 		BackOffMaxRetrys:        c.Uint64(flags.BackOffMaxRetrys.Name),
 		ETHClientTimeout:        c.Uint64(flags.ETHClientTimeout.Name),
+		TargetTxHash:            targetTxHash,
 		OpenDBFunc: func() (DB, error) {
 			return db.OpenDBConnection(db.DBConnectionOpts{
 				Name:            c.String(flags.DatabaseUsername.Name),

packages/relayer/processor/process_message.go

@@ -299,9 +299,13 @@ func (p *Processor) processMessage(
 		relayer.DoneEvents.Inc()
 	}
 
-	// update message status
-	if err := p.eventRepo.UpdateStatus(ctx, msgBody.ID, relayer.EventStatus(messageStatus)); err != nil {
-		return errors.Wrap(err, fmt.Sprintf("p.eventRepo.UpdateStatus, id: %v", msgBody.ID))
+	// internal will only be set if it's an actual queue message, not a targeted
+	// transaction hash.
+	if msg.Internal != nil {
+		// update message status
+		if err := p.eventRepo.UpdateStatus(ctx, msgBody.ID, relayer.EventStatus(messageStatus)); err != nil {
+			return errors.Wrap(err, fmt.Sprintf("p.eventRepo.UpdateStatus, id: %v", msgBody.ID))
+		}
 	}
 
 	return nil

packages/relayer/processor/process_single.go

@@ -0,0 +1,52 @@
+package processor
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/taikoxyz/taiko-mono/packages/relayer/bindings/bridge"
+	"github.com/taikoxyz/taiko-mono/packages/relayer/pkg/queue"
+)
+
+func (p *Processor) processSingle(ctx context.Context) error {
+	bridgeAbi, err := abi.JSON(strings.NewReader(bridge.BridgeABI))
+	if err != nil {
+		return err
+	}
+
+	receipt, err := p.srcEthClient.TransactionReceipt(ctx, *p.targetTxHash)
+	if err != nil {
+		return err
+	}
+
+	for _, log := range receipt.Logs {
+		topic := log.Topics[0]
+
+		if topic == bridgeAbi.Events["MessageSent"].ID {
+			event, err := p.destBridge.ParseMessageSent(*log)
+			if err != nil {
+				return err
+			}
+
+			msg := queue.QueueMessageBody{
+				ID:    0,
+				Event: event,
+			}
+
+			marshalledMsg, err := json.Marshal(msg)
+			if err != nil {
+				return err
+			}
+
+			if err := p.processMessage(ctx, queue.Message{
+				Body: marshalledMsg,
+			}); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}

packages/relayer/processor/processor.go

@@ -112,6 +112,8 @@ type Processor struct {
 	destChainId *big.Int
 
 	taikoL2 *taikol2.TaikoL2
+
+	targetTxHash *common.Hash // optional, set to target processing a specific txHash only
 }
 
 func (p *Processor) InitFromCli(ctx context.Context, c *cli.Context) error {
@@ -207,11 +209,6 @@ func InitFromConfig(ctx context.Context, p *Processor, cfg *Config) error {
 		})
 	}
 
-	q, err := cfg.OpenQueueFunc()
-	if err != nil {
-		return err
-	}
-
 	srcSignalService, err := signalservice.NewSignalService(
 		cfg.SrcSignalServiceAddress,
 		srcEthClient,
@@ -294,6 +291,14 @@ func InitFromConfig(ctx context.Context, p *Processor, cfg *Config) error {
 		p.taikoL2 = taikoL2
 	}
 
+	var q queue.Queue
+	if p.targetTxHash != nil {
+		q, err = cfg.OpenQueueFunc()
+		if err != nil {
+			return err
+		}
+	}
+
 	p.hops = hops
 	p.prover = prover
 	p.eventRepo = eventRepository
@@ -335,6 +340,8 @@ func InitFromConfig(ctx context.Context, p *Processor, cfg *Config) error {
 	p.backOffMaxRetries = cfg.BackOffMaxRetrys
 	p.ethClientTimeout = time.Duration(cfg.ETHClientTimeout) * time.Second
 
+	p.targetTxHash = cfg.TargetTxHash
+
 	return nil
 }
 
@@ -353,6 +360,14 @@ func (p *Processor) Start() error {
 
 	p.cancel = cancel
 
+	// if a targetTxHash is set, we only want to process that specific one.
+	if p.targetTxHash != nil {
+		return p.processSingle(ctx)
+	}
+
+	// otherwise, we can start the queue, and process messages from it
+	// via eventloop.
+
 	if err := p.queue.Start(ctx, p.queueName()); err != nil {
 		return err
 	}