feat: bundle pending bids into a single multi-tx bid

tools/preconf-rpc/bidder/bidder.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"log/slog"
 	"math/big"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -203,7 +204,7 @@ func (b *BidderClient) Bid(
 	ctx context.Context,
 	bidAmount *big.Int,
 	slashAmount *big.Int,
-	rawTx string,
+	rawTxs []string,
 	opts *BidOpts,
 ) (chan BidStatus, error) {
 	if opts == nil {
@@ -264,7 +265,7 @@ func (b *BidderClient) Bid(
 		bidReq := &bidderapiv1.Bid{
 			Amount:              bidAmount.String(),
 			BlockNumber:         int64(blkNumber),
-			RawTransactions:     []string{rawTx},
+			RawTransactions:     rawTxs,
 			DecayStartTimestamp: nowFunc().Add(200 * time.Millisecond).UnixMilli(),
 			SlashAmount:         slashAmount.String(),
 			RevertingTxHashes:   opts.RevertingTxHashes,
@@ -366,6 +367,22 @@ type SettlementMsg struct {
 	IsSlash         bool
 }
 
+// splitTxnHashes splits the comma-joined transaction_hashes value of a
+// notification. A bid with multiple raw transactions produces one
+// commitment that carries all transaction hashes in one joined string.
+// Empty entries are removed.
+func splitTxnHashes(joined string) []string {
+	parts := strings.Split(joined, ",")
+	hashes := make([]string, 0, len(parts))
+	for _, p := range parts {
+		if p == "" {
+			continue
+		}
+		hashes = append(hashes, p)
+	}
+	return hashes
+}
+
 func (b *BidderClient) SubscribeSettlements(ctx context.Context) <-chan SettlementMsg {
 	outCh := make(chan SettlementMsg)
 
@@ -394,14 +411,25 @@ func (b *BidderClient) SubscribeSettlements(ctx context.Context) <-chan Settleme
 				return
 			}
 
-			txHash := msg.Value.Fields["transaction_hashes"].GetStringValue()
+			txHashes := splitTxnHashes(msg.Value.Fields["transaction_hashes"].GetStringValue())
+			if len(txHashes) == 0 {
+				b.logger.Error("settlement notification has no transaction hashes, skipping")
+				continue
+			}
 			provider := msg.Value.Fields["provider"].GetStringValue()
 			isSlash := msg.Value.Fields["is_slashed"].GetBoolValue()
 
-			outCh <- SettlementMsg{
-				TransactionHash: txHash,
-				Provider:        provider,
-				IsSlash:         isSlash,
+			for _, txHash := range txHashes {
+				select {
+				case outCh <- SettlementMsg{
+					TransactionHash: txHash,
+					Provider:        provider,
+					IsSlash:         isSlash,
+				}:
+				case <-ctx.Done():
+					b.logger.Info("settlement subscription context done")
+					return
+				}
 			}
 		}
 	}()
@@ -413,6 +441,11 @@ type PaymentMsg struct {
 	TransactionHash string
 	Payment         *big.Int
 	Refund          *big.Int
+	// BundleTxnHashes lists all transaction hashes covered by the same
+	// payment notification. For a single transaction bid it has one
+	// entry. The payment and refund amounts are the aggregate values
+	// for the full list.
+	BundleTxnHashes []string
 }
 
 func (b *BidderClient) SubscribePayments(ctx context.Context) <-chan PaymentMsg {
@@ -443,7 +476,11 @@ func (b *BidderClient) SubscribePayments(ctx context.Context) <-chan PaymentMsg
 				return
 			}
 
-			txHash := msg.Value.Fields["transaction_hashes"].GetStringValue()
+			txHashes := splitTxnHashes(msg.Value.Fields["transaction_hashes"].GetStringValue())
+			if len(txHashes) == 0 {
+				b.logger.Error("payment notification has no transaction hashes, skipping")
+				continue
+			}
 			paymentStr := msg.Value.Fields["payment"].GetStringValue()
 			refundStr := msg.Value.Fields["refund"].GetStringValue()
 
@@ -456,10 +493,18 @@ func (b *BidderClient) SubscribePayments(ctx context.Context) <-chan PaymentMsg
 				refund = big.NewInt(0)
 			}
 
-			outCh <- PaymentMsg{
-				TransactionHash: txHash,
-				Payment:         payment,
-				Refund:          refund,
+			for _, txHash := range txHashes {
+				select {
+				case outCh <- PaymentMsg{
+					TransactionHash: txHash,
+					Payment:         new(big.Int).Set(payment),
+					Refund:          new(big.Int).Set(refund),
+					BundleTxnHashes: txHashes,
+				}:
+				case <-ctx.Done():
+					b.logger.Info("payment subscription context done")
+					return
+				}
 			}
 		}
 	}()

tools/preconf-rpc/bidder/bidder_test.go

@@ -195,6 +195,10 @@ func TestBidderClient(t *testing.T) {
 	_, _ = rand.Read(buf)
 	txString := hex.EncodeToString(buf)
 
+	buf2 := make([]byte, 32)
+	_, _ = rand.Read(buf2)
+	txString2 := hex.EncodeToString(buf2)
+
 	rpcServices.topo = &debugapiv1.TopologyResponse{
 		Topology: topoVal,
 	}
@@ -207,7 +211,7 @@ func TestBidderClient(t *testing.T) {
 		t.Fatalf("expected 2 providers, got %d", len(providers))
 	}
 
-	statusC, err := bidderClient.Bid(ctx, big.NewInt(1), big.NewInt(1), txString, nil)
+	statusC, err := bidderClient.Bid(ctx, big.NewInt(1), big.NewInt(1), []string{txString, txString2}, nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -242,9 +246,15 @@ waitLoop:
 			if bid.BlockNumber != 11 {
 				t.Fatalf("expected block number 11, got %d", bid.BlockNumber)
 			}
+			if len(bid.RawTransactions) != 2 {
+				t.Fatalf("expected 2 raw transactions, got %d", len(bid.RawTransactions))
+			}
 			if bid.RawTransactions[0] != txString {
 				t.Fatalf("expected raw transaction %x, got %s", buf, bid.RawTransactions[0])
 			}
+			if bid.RawTransactions[1] != txString2 {
+				t.Fatalf("expected raw transaction %x, got %s", buf2, bid.RawTransactions[1])
+			}
 			rpcServices.commitmentChan <- &bidderapiv1.Commitment{
 				BlockNumber: 11,
 			}
@@ -262,3 +272,100 @@ waitLoop:
 	cancel()
 	<-done
 }
+
+func TestSubscribeSettlementsSplit(t *testing.T) {
+	t.Parallel()
+
+	rpcServices := &testRPCServices{
+		notificationChan: make(chan *notificationsapiv1.Notification),
+	}
+	bidderClient := bidder.NewBidderClient(
+		util.NewTestLogger(os.Stdout),
+		rpcServices,
+		rpcServices,
+		rpcServices,
+		&testBlockNumberGetter{blockNumber: 10},
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	settlements := bidderClient.SubscribeSettlements(ctx)
+
+	val, err := structpb.NewStruct(map[string]interface{}{
+		"transaction_hashes": "aaa,bbb",
+		"provider":           "provider1",
+		"is_slashed":         true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rpcServices.notificationChan <- &notificationsapiv1.Notification{
+		Topic: "transaction_settled",
+		Value: val,
+	}
+
+	for _, want := range []string{"aaa", "bbb"} {
+		msg := <-settlements
+		if msg.TransactionHash != want {
+			t.Fatalf("expected transaction hash %s, got %s", want, msg.TransactionHash)
+		}
+		if msg.Provider != "provider1" {
+			t.Fatalf("expected provider provider1, got %s", msg.Provider)
+		}
+		if !msg.IsSlash {
+			t.Fatal("expected is slashed to be true")
+		}
+	}
+}
+
+func TestSubscribePaymentsSplit(t *testing.T) {
+	t.Parallel()
+
+	rpcServices := &testRPCServices{
+		notificationChan: make(chan *notificationsapiv1.Notification),
+	}
+	bidderClient := bidder.NewBidderClient(
+		util.NewTestLogger(os.Stdout),
+		rpcServices,
+		rpcServices,
+		rpcServices,
+		&testBlockNumberGetter{blockNumber: 10},
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	payments := bidderClient.SubscribePayments(ctx)
+
+	val, err := structpb.NewStruct(map[string]interface{}{
+		"transaction_hashes": "aaa,bbb",
+		"payment":            "100",
+		"refund":             "40",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rpcServices.notificationChan <- &notificationsapiv1.Notification{
+		Topic: "transaction_payment",
+		Value: val,
+	}
+
+	for _, want := range []string{"aaa", "bbb"} {
+		msg := <-payments
+		if msg.TransactionHash != want {
+			t.Fatalf("expected transaction hash %s, got %s", want, msg.TransactionHash)
+		}
+		if msg.Payment.Cmp(big.NewInt(100)) != 0 {
+			t.Fatalf("expected payment 100, got %s", msg.Payment)
+		}
+		if msg.Refund.Cmp(big.NewInt(40)) != 0 {
+			t.Fatalf("expected refund 40, got %s", msg.Refund)
+		}
+		if len(msg.BundleTxnHashes) != 2 || msg.BundleTxnHashes[0] != "aaa" || msg.BundleTxnHashes[1] != "bbb" {
+			t.Fatalf("expected bundle hashes [aaa bbb], got %v", msg.BundleTxnHashes)
+		}
+	}
+}

tools/preconf-rpc/main.go

@@ -277,6 +277,13 @@ var (
 		EnvVars: []string{"PRECONF_RPC_LOG_ENCRYPTION_KEY"},
 	}
 
+	optionBundleBids = &cli.BoolFlag{
+		Name:    "bundle-bids",
+		Usage:   "Send one bundle bid for multiple pending transactions. Set to false to send one bid per transaction.",
+		EnvVars: []string{"PRECONF_RPC_BUNDLE_BIDS"},
+		Value:   true,
+	}
+
 	optionLogFmt = &cli.StringFlag{
 		Name:    "log-fmt",
 		Usage:   "log format to use, options are 'text' or 'json'",
@@ -411,6 +418,7 @@ func main() {
 			optionAuthToken,
 			optionSimulationURLs,
 			optionUseInlineSimulation,
+			optionBundleBids,
 			optionBackrunnerAPIURL,
 			optionBackrunnerRPCURL,
 			optionBackrunnerAPIKey,
@@ -536,6 +544,7 @@ func main() {
 				Token:                  c.String(optionAuthToken.Name),
 				SimulatorURLs:          c.StringSlice(optionSimulationURLs.Name),
 				UseInlineSimulation:    c.Bool(optionUseInlineSimulation.Name),
+				BundleBids:             c.Bool(optionBundleBids.Name),
 				BackrunnerAPIURL:       c.String(optionBackrunnerAPIURL.Name),
 				BackrunnerRPC:          c.String(optionBackrunnerRPCURL.Name),
 				BackrunnerAPIKey:       c.String(optionBackrunnerAPIKey.Name),

tools/preconf-rpc/sender/export_test.go

@@ -0,0 +1,47 @@
+package sender
+
+import (
+	"context"
+	"sync"
+
+	"golang.org/x/sync/errgroup"
+)
+
+// ProcessQueuedForTest runs one queue processing pass with the given
+// context. Tests use it to drive the dispatch path directly, without
+// the ticker loop that Start runs.
+func (t *TxSender) ProcessQueuedForTest(ctx context.Context) {
+	if t.eg == nil {
+		t.eg, t.egCtx = errgroup.WithContext(context.Background())
+	}
+	t.processMu.Lock()
+	defer t.processMu.Unlock()
+	t.processQueuedTransactions(ctx)
+}
+
+// WaitWorkersForTest waits for every worker goroutine that a
+// ProcessQueuedForTest call started.
+func (t *TxSender) WaitWorkersForTest() error {
+	return t.eg.Wait()
+}
+
+// FillWorkerPoolForTest fills the worker pool until only free slots
+// remain. Tests use it to make a batch slot acquisition block partway
+// through. It returns a drain function that releases the filled slots;
+// the drain function is safe to call more than once. Call it only
+// while no worker goroutine runs.
+func (t *TxSender) FillWorkerPoolForTest(free int) func() {
+	filled := 0
+	for cap(t.workerPool)-len(t.workerPool) > free {
+		t.workerPool <- struct{}{}
+		filled++
+	}
+	var once sync.Once
+	return func() {
+		once.Do(func() {
+			for ; filled > 0; filled-- {
+				<-t.workerPool
+			}
+		})
+	}
+}