diff --git a/tools/preconf-rpc/bidder/bidder.go b/tools/preconf-rpc/bidder/bidder.go index 682c2b02f..8866b9809 100644 --- a/tools/preconf-rpc/bidder/bidder.go +++ b/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 + } } } }() diff --git a/tools/preconf-rpc/bidder/bidder_test.go b/tools/preconf-rpc/bidder/bidder_test.go index 16a909576..24cb24805 100644 --- a/tools/preconf-rpc/bidder/bidder_test.go +++ b/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 <- ¬ificationsapiv1.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 <- ¬ificationsapiv1.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) + } + } +} diff --git a/tools/preconf-rpc/main.go b/tools/preconf-rpc/main.go index 01ada4fa5..72a120cac 100644 --- a/tools/preconf-rpc/main.go +++ b/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), diff --git a/tools/preconf-rpc/sender/export_test.go b/tools/preconf-rpc/sender/export_test.go new file mode 100644 index 000000000..f053d73f0 --- /dev/null +++ b/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 + } + }) + } +} diff --git a/tools/preconf-rpc/sender/sender.go b/tools/preconf-rpc/sender/sender.go index e67127ce6..69b461850 100644 --- a/tools/preconf-rpc/sender/sender.go +++ b/tools/preconf-rpc/sender/sender.go @@ -25,6 +25,7 @@ import ( "github.com/primev/mev-commit/tools/preconf-rpc/sim" "github.com/prometheus/client_golang/prometheus" "golang.org/x/sync/errgroup" + "google.golang.org/protobuf/proto" ) type TxType int @@ -57,6 +58,7 @@ const ( transactionTimeout = 10 * time.Minute // timeout for transaction processing maxAttemptsPerBlock = 10 // maximum attempts per block defaultRetryDelay = 500 * time.Millisecond + maxBundleSize = 5 // maximum number of transactions in one bundle bid ) var ( @@ -172,7 +174,7 @@ type Bidder interface { ctx context.Context, bidAmount *big.Int, slashAmount *big.Int, - rawTx string, + rawTxs []string, opts *bidder.BidOpts, ) (chan bidder.BidStatus, error) ConnectedProviders(ctx context.Context) ([]string, error) @@ -251,6 +253,7 @@ type TxSender struct { metrics *metrics explorerSubmitter ExplorerSubmitter logEncryptionKey []byte + bundleBids bool } func noOpFastTrack(_ []*bidderapiv1.Commitment, _ bool) bool { @@ -269,6 +272,7 @@ func NewTxSender( settlementChainId *big.Int, explorerSubmitter ExplorerSubmitter, logEncryptionKey []byte, + bundleBids bool, logger *slog.Logger, ) (*TxSender, error) { txnAttemptHistory, err := lru.New[common.Hash, *txnAttempt](1000) @@ -306,6 +310,7 @@ func NewTxSender( metrics: newMetrics(), explorerSubmitter: explorerSubmitter, logEncryptionKey: logEncryptionKey, + bundleBids: bundleBids, }, nil } @@ -594,6 +599,124 @@ func (t *TxSender) markCompleted(txn *Transaction) { t.metrics.inflightTransactions.Dec() } +// bundleMember is the slot of one transaction in a bundle bid batch. The +// batch leader writes the cost, result and error fields before it closes +// the batch done channel. Followers read their slot only after the done +// channel is closed. The fastTracked flag records that the member got +// enough commitments to be stored and signaled as preconfirmed while +// the bid status loop was still open. +type bundleMember struct { + txn *Transaction + cancel <-chan struct{} + cost *big.Int + res bidResult + err error + fastTracked bool + // storedCommitments is the number of commitments that the + // fast-track write persisted. Later bundle commitments append in + // memory only; the success gate compares this count with the live + // count and stores the grown set again. + storedCommitments int +} + +// bundleBatch groups the transactions that share one bundle bid attempt. +// The first member is the leader. The leader performs the bid for the +// whole batch and closes the done channel when every member slot is set. +type bundleBatch struct { + members []*bundleMember + done chan struct{} +} + +func (b *bundleBatch) leader() *Transaction { + return b.members[0].txn +} + +func (b *bundleBatch) member(txn *Transaction) *bundleMember { + for _, m := range b.members { + if m.txn == txn { + return m + } + } + return nil +} + +// bundleEligible reports whether a transaction can join a bundle bid. +// Only regular transactions without a position constraint are eligible. +// All other transactions follow the solo bid path. The freshness check +// (no previous commitments) runs later, after the inflight mark, because +// the commitments field is only safe to read for transactions that no +// other goroutine owns. +func (t *TxSender) bundleEligible(txn *Transaction) bool { + return txn.Type == TxTypeRegular && txn.Constraint == nil +} + +// acquireWorkerSlots takes count worker pool slots before any goroutine +// starts. It returns false if the context ended first. In that case it +// puts the slots it already took back into the pool, so the caller can +// release every inflight mark safely: no goroutine has started. The +// context check before each select makes a context that is already done +// win deterministically; a select with a free pool slot and a done +// context picks a case at random. +func (t *TxSender) acquireWorkerSlots(ctx context.Context, count int) bool { + release := func(held int) { + for ; held > 0; held-- { + <-t.workerPool + } + } + for held := 0; held < count; held++ { + if ctx.Err() != nil { + release(held) + return false + } + select { + case <-ctx.Done(): + release(held) + return false + case t.workerPool <- struct{}{}: + } + } + return true +} + +// dispatchMarked takes one worker pool slot and starts the worker +// goroutine for a transaction that is already marked inflight. It +// returns false if the context ended before a slot was free. In that +// case the caller must release the inflight marks of the members that +// did not get a goroutine. +func (t *TxSender) dispatchMarked(ctx context.Context, txn *Transaction, cancel <-chan struct{}, batch *bundleBatch) bool { + if !t.acquireWorkerSlots(ctx, 1) { + return false + } + t.startMarked(ctx, txn, cancel, batch) + return true +} + +// startMarked starts the worker goroutine for a transaction that is +// already marked inflight and already holds one worker pool slot. The +// goroutine puts the slot back when it ends. +func (t *TxSender) startMarked(ctx context.Context, txn *Transaction, cancel <-chan struct{}, batch *bundleBatch) { + t.eg.Go(func() error { + defer func() { <-t.workerPool }() + defer t.triggerSender() // Trigger to reprocess after this transaction + defer t.markCompleted(txn) + + t.logger.Info("Processing transaction", "sender", txn.Sender.Hex(), "type", txn.Type) + if err := t.processTransaction(ctx, txn, cancel, batch); err != nil { + t.logger.Error("Failed to process transaction", "sender", txn.Sender.Hex(), "error", err) + txn.Status = TxStatusFailed + txn.Details = err.Error() + t.clearBlockAttemptHistory(txn, time.Now()) + // The terminal status write must survive the cancellation + // that caused the failure. A write on the cancelled + // context fails, the row keeps its old status, and the + // queue never selects it again: the failure would stay + // invisible. + return t.store.StoreTransaction(context.WithoutCancel(ctx), txn, nil, nil) + } + return nil + }) +} + func (t *TxSender) processQueuedTransactions(ctx context.Context) { txns, err := t.store.GetQueuedTransactions(ctx) if err != nil { @@ -606,7 +729,95 @@ func (t *TxSender) processQueuedTransactions(ctx context.Context) { return } t.logger.Debug("Processing queued transactions", "count", len(txns)) + + var solo, eligible []*Transaction for _, txn := range txns { + if t.bundleBids && t.bundleEligible(txn) { + eligible = append(eligible, txn) + } else { + solo = append(solo, txn) + } + } + if len(eligible) == 1 { + // A single eligible transaction cannot form a bundle. + solo = append(solo, eligible[0]) + eligible = nil + } + + release := func(members []*bundleMember) { + for _, m := range members { + t.markCompleted(m.txn) + } + } + + for start := 0; start < len(eligible); start += maxBundleSize { + end := min(start+maxBundleSize, len(eligible)) + // Build the roster under processMu. The historical check and the + // inflight mark run here, in the same order the solo goroutine + // uses. After this loop the roster is fixed. + roster := make([]*bundleMember, 0, end-start) + var preMarked []*bundleMember + for _, txn := range eligible[start:end] { + if t.historicalTxns.Contains(txn.Hash()) { + t.logger.Warn("Transaction already processed historically, skipping", "hash", txn.Hash().Hex()) + continue + } + canExecute, cancel := t.markInflight(txn) + if !canExecute { + // Transaction is already being processed or sender has an inflight transaction + continue + } + if len(txn.commitments) > 0 { + // Not a fresh transaction. It keeps its inflight mark + // and follows the solo path. + preMarked = append(preMarked, &bundleMember{txn: txn, cancel: cancel}) + continue + } + roster = append(roster, &bundleMember{txn: txn, cancel: cancel}) + } + if len(roster) == 1 { + // A single member cannot form a bundle. + preMarked = append(preMarked, roster[0]) + roster = nil + } + var batch *bundleBatch + if len(roster) > 1 { + batch = &bundleBatch{ + members: roster, + done: make(chan struct{}), + } + } + // All-or-nothing dispatch for the batch: every roster member + // gets its worker pool slot before any member goroutine starts. + // A partial dispatch would let the leader bid for a member that + // has no goroutine and whose inflight mark is released, and the + // next tick would re-dispatch that member: a duplicate bid. + // When the acquisition fails no goroutine has started, so every + // inflight mark can be released and the next tick retries the + // whole batch. + if len(roster) > 0 { + if !t.acquireWorkerSlots(ctx, len(roster)) { + release(roster) + release(preMarked) + t.logger.Info("Context cancelled, stopping transaction processing") + return + } + // The leader is started first so that the batch never runs + // without its leader goroutine. + for _, m := range roster { + t.startMarked(ctx, m.txn, m.cancel, batch) + } + } + for i, m := range preMarked { + if !t.dispatchMarked(ctx, m.txn, m.cancel, nil) { + release(preMarked[i:]) + t.logger.Info("Context cancelled, stopping transaction processing") + return + } + } + } + + for _, txn := range solo { txn := txn // capture range variable select { case <-ctx.Done(): @@ -630,12 +841,17 @@ func (t *TxSender) processQueuedTransactions(ctx context.Context) { defer t.markCompleted(txn) t.logger.Info("Processing transaction", "sender", txn.Sender.Hex(), "type", txn.Type) - if err := t.processTransaction(ctx, txn, cancel); err != nil { + if err := t.processTransaction(ctx, txn, cancel, nil); err != nil { t.logger.Error("Failed to process transaction", "sender", txn.Sender.Hex(), "error", err) txn.Status = TxStatusFailed txn.Details = err.Error() t.clearBlockAttemptHistory(txn, time.Now()) - return t.store.StoreTransaction(ctx, txn, nil, nil) + // The terminal status write must survive the + // cancellation that caused the failure. A write on + // the cancelled context fails, the row keeps its + // old status, and the queue never selects it + // again: the failure would stay invisible. + return t.store.StoreTransaction(context.WithoutCancel(ctx), txn, nil, nil) } return nil }) @@ -643,7 +859,7 @@ func (t *TxSender) processQueuedTransactions(ctx context.Context) { } } -func (t *TxSender) processTransaction(ctx context.Context, txn *Transaction, cancel <-chan struct{}) error { +func (t *TxSender) processTransaction(ctx context.Context, txn *Transaction, cancel <-chan struct{}, batch *bundleBatch) error { var ( result bidResult err error @@ -660,10 +876,34 @@ func (t *TxSender) processTransaction(ctx context.Context, txn *Transaction, can BID_LOOP: for { - result, err = t.sendBid(ctx, txn) + switch { + case batch == nil: + result, err = t.sendBid(ctx, txn) + case batch.leader() == txn: + // The leader sends one bid for the whole batch and fills + // every member slot. The batch is spent after the first + // iteration; later iterations use the solo path. + result, err = t.sendBundleBid(ctx, batch) + batch = nil + default: + // Followers wait for the leader to close the done channel. + // The wait is bounded by the leader's bid timeout. Cancel + // and context signals are handled by the select below. + <-batch.done + m := batch.member(txn) + result, err = m.res, m.err + batch = nil + } switch { case err != nil: - if retryErr, ok := err.(*errRetry); ok { + if errors.Is(err, errBundleAborted) { + // The bundle never placed a bid for this member. The + // attempt state is already rolled back, so the next + // iteration runs a solo bid with first-attempt + // semantics. + logger.Warn("Bundle aborted before the bid, retrying solo") + retryTicker.Reset(defaultRetryDelay) + } else if retryErr, ok := err.(*errRetry); ok { logger.Warn( "Retrying bid due to error", "error", retryErr.err, @@ -677,29 +917,63 @@ BID_LOOP: } case txn.noOfProviders == len(txn.commitments): if result.optedInSlot { - if txn.Status != TxStatusPreConfirmed { - t.metrics.timeToFirstPreconfirmation.Observe(float64(time.Since(result.startTime).Milliseconds())) - } - // This means that all builders have committed to the bid and it - // is a primev opted in slot. We can safely proceed to inform the - // user that the txn was successfully sent and will be processed - txn.Status = TxStatusPreConfirmed - txn.BlockNumber = int64(result.blockNumber) - logger.Info( - "Transaction pre-confirmed", - "blockNumber", result.blockNumber, - "bidAmount", result.bidAmount.String(), - ) - if err := t.store.StoreTransaction(ctx, txn, txn.commitments, txn.logs); err != nil { - return fmt.Errorf("failed to store preconfirmed transaction: %w", err) - } - t.signalReceiptAvailable(txn.Hash()) - if err := t.explorerSubmitter.Submit( - ctx, - txn.Transaction, - txn.Sender, - ); err != nil { - t.logger.Error("Failed to submit tx to explorer", "error", err) + if result.fastTracked { + // The batch leader already stored this member's + // preconfirmation, signaled its receipt and + // submitted it to the explorer when the member + // was fast-tracked. A repeat store here can run + // with a context that is already done, and its + // failure would flip the stored preconfirmation + // to a failed state. So the member skips that + // work and only resets the retry ticker below. + logger.Info( + "Fast-tracked bundle member keeps its stored preconfirmation", + "blockNumber", result.blockNumber, + "bidAmount", result.bidAmount.String(), + ) + if len(txn.commitments) > result.fastTrackCommitments { + // More commitments arrived after the + // fast-track write and exist in memory only. + // Store the grown set, so the receipt API + // shows every commitment, like the solo path. + // The write must survive a done context, and + // a failure only logs a warning: it must + // never flip the stored preconfirmation to a + // failed state. + if err := t.store.StoreTransaction( + context.WithoutCancel(ctx), txn, txn.commitments, txn.logs, + ); err != nil { + logger.Warn( + "Failed to store late bundle commitments", + "error", err, + ) + } + } + } else { + if txn.Status != TxStatusPreConfirmed { + t.metrics.timeToFirstPreconfirmation.Observe(float64(time.Since(result.startTime).Milliseconds())) + } + // This means that all builders have committed to the bid and it + // is a primev opted in slot. We can safely proceed to inform the + // user that the txn was successfully sent and will be processed + txn.Status = TxStatusPreConfirmed + txn.BlockNumber = int64(result.blockNumber) + logger.Info( + "Transaction pre-confirmed", + "blockNumber", result.blockNumber, + "bidAmount", result.bidAmount.String(), + ) + if err := t.store.StoreTransaction(ctx, txn, txn.commitments, txn.logs); err != nil { + return fmt.Errorf("failed to store preconfirmed transaction: %w", err) + } + t.signalReceiptAvailable(txn.Hash()) + if err := t.explorerSubmitter.Submit( + ctx, + txn.Transaction, + txn.Sender, + ); err != nil { + t.logger.Error("Failed to submit tx to explorer", "error", err) + } } } retryTicker.Reset(result.timeUntillNextBlock + 1*time.Second) @@ -718,6 +992,24 @@ BID_LOOP: } select { case <-ctx.Done(): + if txn.Status == TxStatusPreConfirmed && + txn.Type != TxTypeDeposit && txn.Type != TxTypeInstantBridge { + // The preconfirmation for this transaction is stored + // and signaled. A shutdown must not revert it to a + // failed state, so the loop ends without an error. + // This holds for the solo path and for bundle members + // in the same way. Only TxTypeDeposit and + // TxTypeInstantBridge are excluded: they run their + // post-loop obligation, the AddBalance credit or the + // bridge Transfer, after this loop, and the queue only + // selects pending transactions again. A silent success + // here would skip that step forever. Those two types + // return the context error, become visibly failed, and + // the user can retry them. Every other type, such as a + // preconfirmed fastswap, has no post-loop work and + // keeps its preconfirmation. + return nil + } return ctx.Err() case <-cancel: return ErrTransactionCancelled @@ -785,12 +1077,33 @@ func (e *errRetry) Error() string { return fmt.Sprintf("retry after %s: %v", e.retryAfter, e.err) } +// errBundleAborted tells a bundle member that the bundle bid was +// aborted before any bid went out. The member did not use up a real +// bid attempt, so its next solo bid must run with first-attempt +// semantics: it simulates again and sets noOfProviders from the live +// provider set. This keeps the zero-commitment outcome of the solo +// bid from passing the noOfProviders == len(commitments) success gate +// as 0 == 0. +var errBundleAborted = errors.New("bundle aborted before bid placement") + type bidResult struct { startTime time.Time timeUntillNextBlock time.Duration blockNumber uint64 optedInSlot bool bidAmount *big.Int + // fastTracked marks a bundle member result whose preconfirmation + // the batch leader already stored and signaled per member. The + // success gate must not repeat that work: a repeat store can run + // after the context ends, and its failure would flip the stored + // preconfirmation to a failed state. The solo path never sets it. + fastTracked bool + // fastTrackCommitments is the number of commitments that the + // fast-track write persisted for this member. When more + // commitments arrive before the bid loop ends, the success gate + // stores the grown set again, so the receipt API shows every + // commitment, like the solo path. + fastTrackCommitments int } func (t *TxSender) sendBid( @@ -953,15 +1266,8 @@ func (t *TxSender) sendBid( } txn.simFailed = false if !isRetry { - providers, err := t.bidder.ConnectedProviders(ctx) - if err != nil { - logger.Error("Failed to get connected providers", "error", err) - return bidResult{}, fmt.Errorf("failed to get connected providers: %w", err) - } txn.logs = logs txn.isSwap = isSwap - txn.noOfProviders = len(providers) - t.metrics.connectedProviders.Set(float64(len(providers))) // We could have already made a attempt on the previous block but the block // update hasn't happened yet. This means that the bid might fail, but // we should retain the previous commitments. Only clear if we get new @@ -969,12 +1275,27 @@ func (t *TxSender) sendBid( } } + // The success gate compares noOfProviders with the commitment + // count. A retry that never set noOfProviders must not let a + // zero-commitment outcome pass that gate as zero equals zero. So + // the provider count comes from the live provider set on every + // first attempt and on every retry that has no provider count yet. + if !isRetry || txn.noOfProviders == 0 { + providers, err := t.bidder.ConnectedProviders(ctx) + if err != nil { + logger.Error("Failed to get connected providers", "error", err) + return bidResult{}, fmt.Errorf("failed to get connected providers: %w", err) + } + txn.noOfProviders = len(providers) + t.metrics.connectedProviders.Set(float64(len(providers))) + } + bidStart := time.Now() bidC, err := t.bidder.Bid( cctx, cost, slashAmount, - strings.TrimPrefix(txn.Raw, "0x"), + []string{strings.TrimPrefix(txn.Raw, "0x")}, &bidder.BidOpts{ WaitForOptIn: false, BlockNumber: uint64(bidBlockNo), @@ -1066,6 +1387,344 @@ BID_LOOP: return result, nil } +// sendBundleBid sends one bid that carries the raw transactions of all +// batch members. It runs in the goroutine of the batch leader. It writes +// a result or an error into the slot of every member and closes the batch +// done channel when it returns. The return values are the slot values of +// the leader. A member that fails a pre-check drops out of the bundle and +// receives the same error the solo path produces for that failure. +func (t *TxSender) sendBundleBid(ctx context.Context, batch *bundleBatch) (bidResult, error) { + defer close(batch.done) + + start := time.Now() + leader := batch.members[0] + logger := t.logger.With( + "leaderTransactionHash", leader.txn.Hash().Hex(), + "bundleSize", len(batch.members), + ) + + distribute := func(members []*bundleMember, err error) { + for _, m := range members { + m.err = err + } + } + + timeToOptIn, err := t.bidder.Estimate() + if err != nil { + logger.Warn("Failed to estimate time to opt-in", "error", err) + // If we cannot estimate the time to opt-in, we assume a default value and + // proceed with the bid process. The default value should be higher than + // the typical block time to ensure we consider the next slot as a non-opt-in slot. + timeToOptIn = blockTime * 32 + } + + bidBlockNo, timeUntilNextBlock, err := t.blockTracker.NextBlockNumber() + if err != nil { + logger.Error("Failed to get next block number", "error", err) + distribute(batch.members, &errRetry{ + err: fmt.Errorf("failed to get next block number: %w", err), + retryAfter: time.Second, + }) + return leader.res, leader.err + } + logger.Debug("Next block info", "bidBlockNo", bidBlockNo, "timeUntilNextBlock", timeUntilNextBlock) + + if timeUntilNextBlock <= 500*time.Millisecond { + logger.Warn("Next block time is too short, skipping bid", "timeUntilNextBlock", timeUntilNextBlock) + distribute(batch.members, &errRetry{ + err: fmt.Errorf("next block time is too short: %s", timeUntilNextBlock), + retryAfter: defaultRetryDelay, + }) + return leader.res, leader.err + } + + prices := t.pricer.EstimatePrice(ctx) + + // Allow for certain level of tolerance w.r.t timestamps + optedInSlot := math.Abs(float64(timeToOptIn)-float64(timeUntilNextBlock.Seconds())) < float64(blockTime/3) + + cctx, cancel := context.WithTimeout(ctx, t.getBidTimeout()) + defer cancel() + + nextBaseFee := t.blockTracker.NextBaseFee() + latestBaseFee := t.blockTracker.LatestBaseFee() + if nextBaseFee.Sign() == 0 { + nextBaseFee = latestBaseFee + } + + // Per-member pre-checks. A member that fails a check keeps its own + // error and is left out of the bundle. This filters transactions + // that would make the providers reject the full bundle. + active := make([]*bundleMember, 0, len(batch.members)) + for _, m := range batch.members { + mlogger := t.logger.With( + "transactionHash", m.txn.Hash().Hex(), + "sender", m.txn.Sender.Hex(), + "type", m.txn.Type, + ) + + // Every drop below this price calculation rolls the recorded + // block attempt back: the member leaves the bundle before any + // bid goes out, so its solo retry must run with first-attempt + // semantics and set noOfProviders from the live provider set. + cost, _, err := t.calculatePriceForNextBlock(m.txn, bidBlockNo, prices, optedInSlot) + if err != nil { + mlogger.Error("Failed to calculate price for next block", "error", err) + if errors.Is(err, ErrTimeoutExceeded) || errors.Is(err, ErrMaxAttemptsPerBlockExceeded) { + // We propagate these errors as is. A timeout records + // no attempt, so there is nothing to roll back. + m.err = err + } else { + // The failed price lookup recorded an attempt for + // this block, but no bid goes out for this member. + t.rollbackBlockAttempt(m.txn, bidBlockNo) + m.err = &errRetry{ + err: fmt.Errorf("failed to calculate price: %w", err), + retryAfter: time.Second, + } + } + continue + } + + feePerGas := effectiveFeePerGas(m.txn.Transaction) + if nextBaseFee.Sign() > 0 && feePerGas.Cmp(nextBaseFee) < 0 { + mlogger.Warn( + "Fee per gas too low for next block", + "feePerGas", feePerGas.String(), + "nextBaseFee", nextBaseFee.String(), + ) + t.rollbackBlockAttempt(m.txn, bidBlockNo) + m.err = &errRetry{ + err: fmt.Errorf( + "fee per gas too low for next block: %s min %s", + feePerGas.String(), + nextBaseFee.String(), + ), + retryAfter: timeUntilNextBlock, + } + continue + } + + // Bundle members are always regular transactions, so the balance + // check mirrors the regular solo path. + if !t.store.HasBalance(ctx, m.txn.Sender, cost) { + mlogger.Error("Insufficient balance for sender") + t.rollbackBlockAttempt(m.txn, bidBlockNo) + m.err = fmt.Errorf("insufficient balance for sender: %s", m.txn.Sender.Hex()) + continue + } + + state := sim.Latest + if latestBaseFee.Sign() > 0 && feePerGas.Cmp(latestBaseFee) < 0 { + state = sim.Pending + } + + logs, isSwap, err := t.simulator.Simulate(ctx, m.txn.Raw, state) + if err != nil { + m.txn.simFailed = true + mlogger.Error("Failed to simulate transaction", "error", err, "blockNumber", bidBlockNo) + t.rollbackBlockAttempt(m.txn, bidBlockNo) + m.err = fmt.Errorf("failed to simulate transaction: %w", err) + continue + } + m.txn.simFailed = false + m.txn.logs = logs + m.txn.isSwap = isSwap + m.cost = cost + active = append(active, m) + } + + if len(active) < 2 { + logger.Info("Not enough members left for a bundle bid", "active", len(active)) + // The per-member price calculation consumed one block attempt + // for each survivor, but no bid went out. Roll the attempt + // back so the solo retry runs as a first attempt. + for _, m := range active { + t.rollbackBlockAttempt(m.txn, bidBlockNo) + } + distribute(active, errBundleAborted) + return leader.res, leader.err + } + + providers, err := t.bidder.ConnectedProviders(ctx) + if err != nil { + logger.Error("Failed to get connected providers", "error", err) + distribute(active, fmt.Errorf("failed to get connected providers: %w", err)) + return leader.res, leader.err + } + t.metrics.connectedProviders.Set(float64(len(providers))) + + // The bid amount is the sum of the per-member costs. Each member + // hash is listed as revertable, which matches the solo path where + // each bid lists its own transaction hash. + totalCost := big.NewInt(0) + rawTxs := make([]string, 0, len(active)) + revertingTxHashes := make([]string, 0, len(active)) + for _, m := range active { + m.txn.noOfProviders = len(providers) + totalCost = new(big.Int).Add(totalCost, m.cost) + rawTxs = append(rawTxs, strings.TrimPrefix(m.txn.Raw, "0x")) + revertingTxHashes = append(revertingTxHashes, m.txn.Hash().Hex()) + } + + bidStart := time.Now() + bidC, err := t.bidder.Bid( + cctx, + totalCost, + big.NewInt(0), + rawTxs, + &bidder.BidOpts{ + WaitForOptIn: false, + BlockNumber: bidBlockNo, + RevertingTxHashes: revertingTxHashes, + DecayDuration: t.getBidTimeout() * 2, + }, + ) + if err != nil { + logger.Error("Failed to place bundle bid", "error", err) + // The members retry on the solo path after a short delay. + distribute(active, &errRetry{ + err: fmt.Errorf("failed to place bundle bid: %w", err), + retryAfter: defaultRetryDelay, + }) + return leader.res, leader.err + } + + commitmentCount := 0 +BID_LOOP: + for { + select { + case <-ctx.Done(): + logger.Info("Context cancelled while waiting for bid status") + // A fast-tracked member already has its preconfirmation + // stored and signaled. It gets its success result, the + // same result the solo path returns for a fast-tracked + // transaction. Only members that are not resolved get + // the context error. + for _, m := range active { + if m.fastTracked { + m.res = bidResult{ + bidAmount: m.cost, + blockNumber: bidBlockNo, + startTime: start, + timeUntillNextBlock: timeUntilNextBlock, + optedInSlot: optedInSlot, + fastTracked: true, + fastTrackCommitments: m.storedCommitments, + } + continue + } + m.err = ctx.Err() + } + return leader.res, leader.err + case bidStatus, more := <-bidC: + if !more { + logger.Info("Bid channel closed, no more bid statuses") + break BID_LOOP + } + switch bidStatus.Type { + case bidder.BidStatusCommitment: + cmt := bidStatus.Arg.(*bidderapiv1.Commitment) + commitmentCount++ + for _, m := range active { + t.appendMemberCommitment(ctx, m, cmt, bidBlockNo, optedInSlot, start) + } + t.metrics.preconfDurationsProvider.WithLabelValues(cmt.ProviderAddress).Set(float64(time.Since(bidStart).Milliseconds())) + t.metrics.preconfCountsProvider.WithLabelValues(cmt.ProviderAddress).Inc() + case bidder.BidStatusCancelled: + logger.Warn("Bid context cancelled by the bidder") + break BID_LOOP + case bidder.BidStatusFailed: + logger.Error("Bid failed", "error", bidStatus.Arg) + break BID_LOOP + } + } + } + logger.Info( + "Bundle bid operation complete", + "noOfProviders", len(providers), + "noOfCommitments", commitmentCount, + "blockNumber", bidBlockNo, + "optedInSlot", optedInSlot, + ) + + for _, m := range active { + if len(m.txn.commitments) > 0 && m.txn.isSwap { + if err := t.backrunner.Backrun(ctx, m.txn.Raw, m.txn.commitments); err != nil { + logger.Error("Failed to backrun transaction", "error", err) + } + logger.Info("Backrun operation initiated for transaction", "hash", m.txn.Hash().Hex()) + } + m.res = bidResult{ + bidAmount: m.cost, + blockNumber: bidBlockNo, + startTime: start, + timeUntillNextBlock: timeUntilNextBlock, + optedInSlot: optedInSlot, + fastTracked: m.fastTracked, + fastTrackCommitments: m.storedCommitments, + } + } + + return leader.res, leader.err +} + +// appendMemberCommitment stores a copy of a bundle commitment on one +// member with the bid amount replaced by the member's own share. The sum +// of the member shares equals the aggregate bundle bid amount. +func (t *TxSender) appendMemberCommitment( + ctx context.Context, + m *bundleMember, + cmt *bidderapiv1.Commitment, + bidBlockNo uint64, + optedInSlot bool, + start time.Time, +) { + txn := m.txn + if len(txn.commitments) > 0 { + if txn.commitments[0].BlockNumber != int64(bidBlockNo) { + txn.commitments = nil // clear previous commitments for new block + } + } + memberCmt, ok := proto.Clone(cmt).(*bidderapiv1.Commitment) + if !ok { + t.logger.Error("Failed to clone commitment", "transactionHash", txn.Hash().Hex()) + return + } + memberCmt.BidAmount = m.cost.String() + txn.commitments = append(txn.commitments, memberCmt) + if !t.fastTrack(txn.commitments, optedInSlot) { + return + } + m.fastTracked = true + if txn.Status != TxStatusPreConfirmed { + txn.Status = TxStatusPreConfirmed + txn.BlockNumber = int64(bidBlockNo) + t.logger.Info( + "Transaction fast-tracked based on commitments", + "transactionHash", txn.Hash().Hex(), + "blockNumber", bidBlockNo, + "bidAmount", m.cost.String(), + ) + if err := t.store.StoreTransaction(ctx, txn, txn.commitments, txn.logs); err != nil { + t.logger.Error("Failed to store fast-tracked transaction", "error", err) + } + // Record how many commitments the fast-track write covered. + // The success gate stores the set again when it grew after + // this write. + m.storedCommitments = len(txn.commitments) + t.signalReceiptAvailable(txn.Hash()) + t.metrics.timeToFirstPreconfirmation.Observe(float64(time.Since(start).Milliseconds())) + if err := t.explorerSubmitter.Submit( + ctx, + txn.Transaction, + txn.Sender, + ); err != nil { + t.logger.Error("Failed to submit fast-tracked tx to explorer", "error", err) + } + } +} + func (t *TxSender) calculatePriceForNextBlock( txn *Transaction, bidBlockNo uint64, @@ -1146,9 +1805,48 @@ func (t *TxSender) calculatePriceForNextBlock( ) } +// rollbackBlockAttempt undoes one recorded attempt for a block. The +// bundle pre-check records an attempt through calculatePriceForNextBlock +// before the bundle bid goes out. When the bundle aborts before the bid, +// that attempt did not happen, so the next solo bid for the same block +// must see first-attempt state. The entry is removed when no attempts +// remain, which keeps clearBlockAttemptHistory safe: it reads the first +// element of a non-empty attempts list. +func (t *TxSender) rollbackBlockAttempt(txn *Transaction, bidBlockNo uint64) { + attempts, found := t.txnAttemptHistory.Get(txn.Hash()) + if !found { + return + } + + for i := len(attempts.attempts) - 1; i >= 0; i-- { + if attempts.attempts[i].blockNumber != bidBlockNo { + continue + } + attempts.attempts[i].attempts-- + if attempts.attempts[i].attempts <= 0 { + attempts.attempts = append(attempts.attempts[:i], attempts.attempts[i+1:]...) + } + break + } + + if len(attempts.attempts) == 0 { + _ = t.txnAttemptHistory.Remove(txn.Hash()) + return + } + _ = t.txnAttemptHistory.Add(txn.Hash(), attempts) +} + func (t *TxSender) clearBlockAttemptHistory(txn *Transaction, endTime time.Time) { attempts, found := t.txnAttemptHistory.Get(txn.Hash()) if !found { + // A rollback of the only recorded attempt removes the entry, + // so a terminal drop can land here without one. The user + // notification and the historical record must still happen: + // the historical record keeps the queue from processing the + // same hash again. Only the metrics that read the entry are + // skipped. + t.notifier.NotifyTransactionStatus(txn, 0, 0, 0) + _ = t.historicalTxns.Add(txn.Hash(), struct{}{}) return } diff --git a/tools/preconf-rpc/sender/sender_test.go b/tools/preconf-rpc/sender/sender_test.go index ee79128e1..5ea3e9505 100644 --- a/tools/preconf-rpc/sender/sender_test.go +++ b/tools/preconf-rpc/sender/sender_test.go @@ -33,6 +33,10 @@ type mockStore struct { balances map[common.Address]*big.Int byHash map[common.Hash]*sender.Transaction preconfirmedTxns chan result + addBalanceCalls map[common.Address]int + // honorCtx makes StoreTransaction fail after its context ends, + // like a real database write. Set it before the sender starts. + honorCtx bool } func newMockStore() *mockStore { @@ -42,6 +46,7 @@ func newMockStore() *mockStore { balances: make(map[common.Address]*big.Int), preconfirmedTxns: make(chan result, 10), byHash: make(map[common.Hash]*sender.Transaction), + addBalanceCalls: make(map[common.Address]int), } } @@ -97,6 +102,7 @@ func (m *mockStore) AddBalance(ctx context.Context, account common.Address, amou m.mu.Lock() defer m.mu.Unlock() + m.addBalanceCalls[account]++ if _, exists := m.balances[account]; !exists { m.balances[account] = amount } else { @@ -107,6 +113,13 @@ func (m *mockStore) AddBalance(ctx context.Context, account common.Address, amou return nil } +func (m *mockStore) addBalanceCount(account common.Address) int { + m.mu.Lock() + defer m.mu.Unlock() + + return m.addBalanceCalls[account] +} + func (m *mockStore) DeductBalance(ctx context.Context, account common.Address, amount *big.Int) error { m.mu.Lock() defer m.mu.Unlock() @@ -128,6 +141,9 @@ func (m *mockStore) StoreTransaction( commitments []*bidderapiv1.Commitment, logs []*types.Log, ) error { + if m.honorCtx && ctx.Err() != nil { + return ctx.Err() + } m.preconfirmedTxns <- result{ txn: txn, commitments: commitments, @@ -172,14 +188,19 @@ func (m *mockStore) StoreReceipt( type bidOp struct { bidAmount *big.Int slashAmount *big.Int - rawTx string + rawTxs []string opts *bidder.BidOpts } +type bidResponse struct { + statusCh chan bidder.BidStatus + err error +} + type mockBidder struct { optinEstimate chan int64 in chan bidOp - out chan chan bidder.BidStatus + out chan bidResponse } func (m *mockBidder) Estimate() (int64, error) { @@ -191,18 +212,18 @@ func (m *mockBidder) Bid( ctx context.Context, bidAmount *big.Int, slashAmount *big.Int, - rawTx string, + rawTxs []string, opts *bidder.BidOpts, ) (chan bidder.BidStatus, error) { m.in <- bidOp{ bidAmount: bidAmount, slashAmount: slashAmount, - rawTx: rawTx, + rawTxs: rawTxs, opts: opts, } res := <-m.out - return res, nil + return res.statusCh, res.err } func (m *mockBidder) ConnectedProviders(ctx context.Context) ([]string, error) { @@ -235,11 +256,39 @@ type mockBlockTracker struct { bnIn chan struct{} bnOut chan blockNoOp bnErr chan error + mu sync.Mutex lbf *big.Int nbf *big.Int + incl map[common.Hash]chan uint64 +} + +// inclusionFor registers a dedicated inclusion channel for one +// transaction hash. WaitForTxnInclusion returns that channel for the +// hash, so a test can signal inclusion per transaction instead of +// through the shared out channel. Register before the sender starts. +func (m *mockBlockTracker) inclusionFor(txnHash common.Hash) chan uint64 { + m.mu.Lock() + defer m.mu.Unlock() + + if m.incl == nil { + m.incl = make(map[common.Hash]chan uint64) + } + ch, found := m.incl[txnHash] + if !found { + ch = make(chan uint64, 1) + m.incl[txnHash] = ch + } + return ch } func (m *mockBlockTracker) WaitForTxnInclusion(txnHash common.Hash) chan uint64 { + m.mu.Lock() + ch, found := m.incl[txnHash] + m.mu.Unlock() + if found { + return ch + } + includedCh := make(chan uint64, 1) go func() { included := <-m.out @@ -268,6 +317,9 @@ func (m *mockBlockTracker) AccountNonce(ctx context.Context, account common.Addr } func (m *mockBlockTracker) LatestBaseFee() *big.Int { + m.mu.Lock() + defer m.mu.Unlock() + if m.lbf == nil { return big.NewInt(0) } @@ -275,12 +327,22 @@ func (m *mockBlockTracker) LatestBaseFee() *big.Int { } func (m *mockBlockTracker) NextBaseFee() *big.Int { + m.mu.Lock() + defer m.mu.Unlock() + if m.nbf == nil { return big.NewInt(0) } return new(big.Int).Set(m.nbf) } +func (m *mockBlockTracker) setNextBaseFee(fee *big.Int) { + m.mu.Lock() + defer m.mu.Unlock() + + m.nbf = fee +} + type mockTransferer struct{} func (m *mockTransferer) Transfer(ctx context.Context, to common.Address, chainID *big.Int, amount *big.Int) error { @@ -288,16 +350,35 @@ func (m *mockTransferer) Transfer(ctx context.Context, to common.Address, chainI } type mockNotifier struct { + mu sync.Mutex notifications []string } func (m *mockNotifier) NotifyTransactionStatus(txn *sender.Transaction, attempts, blocks int, start time.Duration) { + m.mu.Lock() + defer m.mu.Unlock() m.notifications = append(m.notifications, txn.Hash().Hex()) } -type mockSimulator struct{} +func (m *mockNotifier) notified(hash common.Hash) bool { + m.mu.Lock() + defer m.mu.Unlock() + for _, h := range m.notifications { + if h == hash.Hex() { + return true + } + } + return false +} + +type mockSimulator struct { + failRaw string +} func (m *mockSimulator) Simulate(ctx context.Context, rawTx string, _ sim.SimState) ([]*types.Log, bool, error) { + if m.failRaw != "" && rawTx == m.failRaw { + return nil, false, errors.New("simulation failed") + } return []*types.Log{}, false, nil } @@ -323,7 +404,7 @@ func TestSender(t *testing.T) { bidderImpl := &mockBidder{ optinEstimate: make(chan int64, 10), in: make(chan bidOp, 10), - out: make(chan chan bidder.BidStatus, 10), + out: make(chan bidResponse, 10), } blockTracker := &mockBlockTracker{ out: make(chan uint64, 10), @@ -344,7 +425,8 @@ func TestSender(t *testing.T) { &mockBackrunner{}, big.NewInt(1), // Settlement chain ID &MockExplorerSubmitter{}, - nil, // no log encryption key in tests + nil, // no log encryption key in tests + true, // bundle bids enabled util.NewTestLogger(os.Stdout), ) if err != nil { @@ -407,8 +489,8 @@ func TestSender(t *testing.T) { // Simulate a bid response bidOp := <-bidderImpl.in - if bidOp.rawTx != tx1.Raw[2:] { - t.Fatalf("expected raw transaction %s, got %s", tx1.Raw, bidOp.rawTx) + if len(bidOp.rawTxs) != 1 || bidOp.rawTxs[0] != tx1.Raw[2:] { + t.Fatalf("expected raw transaction %s, got %v", tx1.Raw, bidOp.rawTxs) } resC := make(chan bidder.BidStatus, 3) resC <- bidder.BidStatus{ @@ -430,7 +512,7 @@ func TestSender(t *testing.T) { }, } close(resC) - bidderImpl.out <- resC + bidderImpl.out <- bidResponse{statusCh: resC} <-waitCh @@ -495,13 +577,13 @@ func TestSender(t *testing.T) { // Simulate a bid response bidOp = <-bidderImpl.in - if bidOp.rawTx != tx2.Raw[2:] { - t.Fatalf("expected raw transaction %s, got %s", tx1.Raw, bidOp.rawTx) + if len(bidOp.rawTxs) != 1 || bidOp.rawTxs[0] != tx2.Raw[2:] { + t.Fatalf("expected raw transaction %s, got %v", tx2.Raw, bidOp.rawTxs) } resC = make(chan bidder.BidStatus, 3) // Simulate retry due to incomplete commitments close(resC) - bidderImpl.out <- resC + bidderImpl.out <- bidResponse{statusCh: resC} // Simulate non opted in block bidderImpl.optinEstimate <- 18 @@ -525,8 +607,8 @@ func TestSender(t *testing.T) { // Simulate a bid response bidOp = <-bidderImpl.in - if bidOp.rawTx != tx2.Raw[2:] { - t.Fatalf("expected raw transaction %s, got %s", tx1.Raw, bidOp.rawTx) + if len(bidOp.rawTxs) != 1 || bidOp.rawTxs[0] != tx2.Raw[2:] { + t.Fatalf("expected raw transaction %s, got %v", tx2.Raw, bidOp.rawTxs) } resC = make(chan bidder.BidStatus, 3) resC <- bidder.BidStatus{ @@ -539,7 +621,7 @@ func TestSender(t *testing.T) { }, } close(resC) - bidderImpl.out <- resC + bidderImpl.out <- bidResponse{statusCh: resC} res = <-st.preconfirmedTxns if res.txn == nil { @@ -583,7 +665,7 @@ func TestCancelTransaction(t *testing.T) { bidderImpl := &mockBidder{ optinEstimate: make(chan int64), in: make(chan bidOp, 10), - out: make(chan chan bidder.BidStatus, 10), + out: make(chan bidResponse, 10), } blockTracker := &mockBlockTracker{ out: make(chan uint64, 10), @@ -603,7 +685,8 @@ func TestCancelTransaction(t *testing.T) { &mockBackrunner{}, big.NewInt(1), // Settlement chain ID &MockExplorerSubmitter{}, - nil, // no log encryption key in tests + nil, // no log encryption key in tests + true, // bundle bids enabled util.NewTestLogger(os.Stdout), ) if err != nil { @@ -672,7 +755,7 @@ func TestIgnoreProvidersOnRetry(t *testing.T) { bidderImpl := &mockBidder{ optinEstimate: make(chan int64, 10), in: make(chan bidOp, 10), - out: make(chan chan bidder.BidStatus, 10), + out: make(chan bidResponse, 10), } blockTracker := &mockBlockTracker{ out: make(chan uint64, 10), @@ -693,7 +776,8 @@ func TestIgnoreProvidersOnRetry(t *testing.T) { &mockBackrunner{}, big.NewInt(1), // Settlement chain ID &MockExplorerSubmitter{}, - nil, // no log encryption key in tests + nil, // no log encryption key in tests + true, // bundle bids enabled util.NewTestLogger(io.Discard), ) if err != nil { @@ -746,8 +830,8 @@ func TestIgnoreProvidersOnRetry(t *testing.T) { // Simulate a bid response bidOp := <-bidderImpl.in - if bidOp.rawTx != tx1.Raw[2:] { - t.Fatalf("expected raw transaction %s, got %s", tx1.Raw, bidOp.rawTx) + if len(bidOp.rawTxs) != 1 || bidOp.rawTxs[0] != tx1.Raw[2:] { + t.Fatalf("expected raw transaction %s, got %v", tx1.Raw, bidOp.rawTxs) } resC := make(chan bidder.BidStatus, 3) resC <- bidder.BidStatus{ @@ -760,7 +844,7 @@ func TestIgnoreProvidersOnRetry(t *testing.T) { }, } close(resC) - bidderImpl.out <- resC + bidderImpl.out <- bidResponse{statusCh: resC} bidderImpl.optinEstimate <- 2 @@ -795,7 +879,7 @@ func TestIgnoreProvidersOnRetry(t *testing.T) { }, } close(resC) - bidderImpl.out <- resC + bidderImpl.out <- bidResponse{statusCh: resC} res := <-st.preconfirmedTxns if res.txn == nil { t.Fatal("expected a preconfirmed transaction, got nil") @@ -804,3 +888,1302 @@ func TestIgnoreProvidersOnRetry(t *testing.T) { cancel() <-done } + +type senderTestEnv struct { + st *mockStore + pricer *mockPricer + bidder *mockBidder + blockTracker *mockBlockTracker + notifier *mockNotifier + sndr *sender.TxSender +} + +func newSenderTestEnv(t *testing.T, simulator *mockSimulator) *senderTestEnv { + t.Helper() + + st := newMockStore() + testPricer := &mockPricer{ + out: make(chan map[int64]float64, 20), + } + bidderImpl := &mockBidder{ + optinEstimate: make(chan int64, 20), + in: make(chan bidOp, 20), + out: make(chan bidResponse, 20), + } + blockTracker := &mockBlockTracker{ + out: make(chan uint64, 20), + bnIn: make(chan struct{}, 20), + bnOut: make(chan blockNoOp, 20), + bnErr: make(chan error, 1), + } + notifier := &mockNotifier{} + + sndr, err := sender.NewTxSender( + st, + bidderImpl, + testPricer, + blockTracker, + &mockTransferer{}, + notifier, + simulator, + &mockBackrunner{}, + big.NewInt(1), // Settlement chain ID + &MockExplorerSubmitter{}, + nil, // no log encryption key in tests + true, // bundle bids enabled + util.NewTestLogger(io.Discard), + ) + if err != nil { + t.Fatalf("failed to create sender: %v", err) + } + + return &senderTestEnv{ + st: st, + pricer: testPricer, + bidder: bidderImpl, + blockTracker: blockTracker, + notifier: notifier, + sndr: sndr, + } +} + +func newTestTxn(senderAddr common.Address, value int64, raw string, txType sender.TxType) *sender.Transaction { + return newTestTxnWithGas(senderAddr, value, raw, 21000, txType) +} + +func newTestTxnWithGas(senderAddr common.Address, value int64, raw string, gas uint64, txType sender.TxType) *sender.Transaction { + return &sender.Transaction{ + Transaction: types.NewTransaction( + 0, + common.HexToAddress("0x1234567890123456789012345678901234567890"), + big.NewInt(value), + gas, + big.NewInt(1), + nil, + ), + Sender: senderAddr, + Type: txType, + Raw: raw, + } +} + +func newTestTxnWithGasPrice(senderAddr common.Address, value int64, raw string, gasPrice int64) *sender.Transaction { + return &sender.Transaction{ + Transaction: types.NewTransaction( + 0, + common.HexToAddress("0x1234567890123456789012345678901234567890"), + big.NewInt(value), + 21000, + big.NewInt(gasPrice), + nil, + ), + Sender: senderAddr, + Type: sender.TxTypeRegular, + Raw: raw, + } +} + +// bidCost is the cost of a transaction with the given gas limit at the +// 85 confidence price of 2.0 gwei. +func bidCost(gas int64) *big.Int { + return new(big.Int).Mul(big.NewInt(2_000_000_000), big.NewInt(gas)) +} + +func testPrices() map[int64]float64 { + return map[int64]float64{ + 70: 0.8, + 75: 1.0, + 80: 1.5, + 85: 2.0, + } +} + +// bundleShare is the cost of one 21000 gas transaction at the 85 +// confidence price of 2.0 gwei. +var bundleShare = big.NewInt(42_000_000_000_000) + +func TestBundleBid(t *testing.T) { + t.Parallel() + + env := newSenderTestEnv(t, &mockSimulator{}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Distinct gas limits give every member a distinct cost, so a + // share or attribution mix-up between members cannot hide behind + // equal amounts. + txns := []*sender.Transaction{ + newTestTxnWithGas(common.HexToAddress("0x1111111111111111111111111111111111111111"), 101, "0xaa01", 21000, sender.TxTypeRegular), + newTestTxnWithGas(common.HexToAddress("0x2222222222222222222222222222222222222222"), 102, "0xaa02", 30000, sender.TxTypeRegular), + newTestTxnWithGas(common.HexToAddress("0x3333333333333333333333333333333333333333"), 103, "0xaa03", 40000, sender.TxTypeRegular), + } + wantShare := map[common.Hash]*big.Int{ + txns[0].Hash(): bidCost(21000), + txns[1].Hash(): bidCost(30000), + txns[2].Hash(): bidCost(40000), + } + for _, txn := range txns { + if err := env.st.AddBalance(ctx, txn.Sender, big.NewInt(5e18)); err != nil { + t.Fatalf("failed to add balance: %v", err) + } + if err := env.sndr.Enqueue(ctx, txn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + } + + // Pre-feed the single bundle attempt. Only the leader calls these. + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + + done := env.sndr.Start(ctx) + + op := <-env.bidder.in + if len(op.rawTxs) != 3 { + t.Fatalf("expected 3 raw transactions in bundle bid, got %d", len(op.rawTxs)) + } + wantRaw := map[string]bool{"aa01": true, "aa02": true, "aa03": true} + for _, r := range op.rawTxs { + if !wantRaw[r] { + t.Fatalf("unexpected raw transaction in bundle bid: %s", r) + } + } + expectedTotal := big.NewInt(0) + for _, share := range wantShare { + expectedTotal = new(big.Int).Add(expectedTotal, share) + } + if op.bidAmount.Cmp(expectedTotal) != 0 { + t.Fatalf("expected bundle bid amount %s, got %s", expectedTotal, op.bidAmount) + } + if op.slashAmount.Sign() != 0 { + t.Fatalf("expected zero slash amount, got %s", op.slashAmount) + } + if op.opts.BlockNumber != 1 { + t.Fatalf("expected block number 1, got %d", op.opts.BlockNumber) + } + if len(op.opts.RevertingTxHashes) != 3 { + t.Fatalf("expected 3 reverting tx hashes, got %d", len(op.opts.RevertingTxHashes)) + } + wantHashes := map[string]bool{} + for _, txn := range txns { + wantHashes[txn.Hash().Hex()] = true + } + for _, h := range op.opts.RevertingTxHashes { + if !wantHashes[h] { + t.Fatalf("unexpected reverting tx hash: %s", h) + } + } + if op.opts.Constraint != nil { + t.Fatal("expected no constraint on bundle bid") + } + if len(op.opts.IgnoreProviders) != 0 { + t.Fatalf("expected no ignored providers, got %v", op.opts.IgnoreProviders) + } + + resC := make(chan bidder.BidStatus, 3) + for _, provider := range []string{"provider1", "provider2"} { + resC <- bidder.BidStatus{ + Type: bidder.BidStatusCommitment, + Arg: &bidderapiv1.Commitment{ + TxHashes: op.opts.RevertingTxHashes, + BidAmount: expectedTotal.String(), + BlockNumber: 1, + ProviderAddress: provider, + }, + } + } + close(resC) + env.bidder.out <- bidResponse{statusCh: resC} + + seen := map[common.Hash]result{} + for i := 0; i < 3; i++ { + res := <-env.st.preconfirmedTxns + seen[res.txn.Hash()] = res + } + for _, txn := range txns { + res, found := seen[txn.Hash()] + if !found { + t.Fatalf("missing preconfirmed result for %s", txn.Hash().Hex()) + } + if res.blockNumber != 1 { + t.Fatalf("expected block number 1, got %d", res.blockNumber) + } + if len(res.commitments) != 2 { + t.Fatalf("expected 2 commitments, got %d", len(res.commitments)) + } + for _, cmt := range res.commitments { + if cmt.BidAmount != wantShare[txn.Hash()].String() { + t.Fatalf( + "expected member share %s for %s, got %s", + wantShare[txn.Hash()], txn.Hash().Hex(), cmt.BidAmount, + ) + } + } + } + + // Let all members observe inclusion and finish. + for i := 0; i < 3; i++ { + env.blockTracker.out <- 1 + } + + cancel() + <-done +} + +func TestBundleFailureSoloFallback(t *testing.T) { + t.Parallel() + + env := newSenderTestEnv(t, &mockSimulator{}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + txns := []*sender.Transaction{ + newTestTxn(common.HexToAddress("0x1111111111111111111111111111111111111111"), 201, "0xbb01", sender.TxTypeRegular), + newTestTxn(common.HexToAddress("0x2222222222222222222222222222222222222222"), 202, "0xbb02", sender.TxTypeRegular), + newTestTxn(common.HexToAddress("0x3333333333333333333333333333333333333333"), 203, "0xbb03", sender.TxTypeRegular), + } + for _, txn := range txns { + if err := env.st.AddBalance(ctx, txn.Sender, big.NewInt(5e18)); err != nil { + t.Fatalf("failed to add balance: %v", err) + } + if err := env.sndr.Enqueue(ctx, txn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + } + + // Pre-feed one bundle attempt and three solo retries. + for i := 0; i < 4; i++ { + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + } + + done := env.sndr.Start(ctx) + + op := <-env.bidder.in + if len(op.rawTxs) != 3 { + t.Fatalf("expected 3 raw transactions in bundle bid, got %d", len(op.rawTxs)) + } + env.bidder.out <- bidResponse{err: errors.New("bundle rejected")} + + soloSeen := map[string]bool{} + for i := 0; i < 3; i++ { + op := <-env.bidder.in + if len(op.rawTxs) != 1 { + t.Fatalf("expected solo bid after bundle failure, got %d raw transactions", len(op.rawTxs)) + } + soloSeen[op.rawTxs[0]] = true + + resC := make(chan bidder.BidStatus, 3) + for _, provider := range []string{"provider1", "provider2"} { + resC <- bidder.BidStatus{ + Type: bidder.BidStatusCommitment, + Arg: &bidderapiv1.Commitment{ + BidAmount: bundleShare.String(), + BlockNumber: 1, + ProviderAddress: provider, + }, + } + } + close(resC) + env.bidder.out <- bidResponse{statusCh: resC} + } + if len(soloSeen) != 3 { + t.Fatalf("expected 3 distinct solo bids, got %d", len(soloSeen)) + } + + for i := 0; i < 3; i++ { + res := <-env.st.preconfirmedTxns + if res.txn.Status != sender.TxStatusPreConfirmed { + t.Fatalf("expected preconfirmed transaction, got %s", res.txn.Status) + } + } + + for i := 0; i < 3; i++ { + env.blockTracker.out <- 1 + } + + cancel() + <-done +} + +func TestBundleMemberSimFailureDropped(t *testing.T) { + t.Parallel() + + badTxn := newTestTxn(common.HexToAddress("0x3333333333333333333333333333333333333333"), 303, "0xcc03", sender.TxTypeRegular) + env := newSenderTestEnv(t, &mockSimulator{failRaw: badTxn.Raw}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + txns := []*sender.Transaction{ + newTestTxn(common.HexToAddress("0x1111111111111111111111111111111111111111"), 301, "0xcc01", sender.TxTypeRegular), + newTestTxn(common.HexToAddress("0x2222222222222222222222222222222222222222"), 302, "0xcc02", sender.TxTypeRegular), + badTxn, + } + for _, txn := range txns { + if err := env.st.AddBalance(ctx, txn.Sender, big.NewInt(5e18)); err != nil { + t.Fatalf("failed to add balance: %v", err) + } + if err := env.sndr.Enqueue(ctx, txn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + } + + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + + done := env.sndr.Start(ctx) + + op := <-env.bidder.in + if len(op.rawTxs) != 2 { + t.Fatalf("expected 2 raw transactions in bundle bid, got %d", len(op.rawTxs)) + } + for _, r := range op.rawTxs { + if r == "cc03" { + t.Fatal("failed member must not appear in the bundle bid") + } + } + expectedTotal := new(big.Int).Mul(bundleShare, big.NewInt(2)) + if op.bidAmount.Cmp(expectedTotal) != 0 { + t.Fatalf("expected bundle bid amount %s, got %s", expectedTotal, op.bidAmount) + } + + resC := make(chan bidder.BidStatus, 3) + for _, provider := range []string{"provider1", "provider2"} { + resC <- bidder.BidStatus{ + Type: bidder.BidStatusCommitment, + Arg: &bidderapiv1.Commitment{ + BidAmount: expectedTotal.String(), + BlockNumber: 1, + ProviderAddress: provider, + }, + } + } + close(resC) + env.bidder.out <- bidResponse{statusCh: resC} + + preconfirmed := 0 + failed := 0 + for i := 0; i < 3; i++ { + res := <-env.st.preconfirmedTxns + switch res.txn.Status { + case sender.TxStatusPreConfirmed: + preconfirmed++ + if len(res.commitments) != 2 { + t.Fatalf("expected 2 commitments, got %d", len(res.commitments)) + } + for _, cmt := range res.commitments { + if cmt.BidAmount != bundleShare.String() { + t.Fatalf("expected member share %s, got %s", bundleShare, cmt.BidAmount) + } + } + case sender.TxStatusFailed: + failed++ + if res.txn.Hash() != badTxn.Hash() { + t.Fatalf("expected failed transaction %s, got %s", badTxn.Hash().Hex(), res.txn.Hash().Hex()) + } + default: + t.Fatalf("unexpected transaction status %s", res.txn.Status) + } + } + if preconfirmed != 2 || failed != 1 { + t.Fatalf("expected 2 preconfirmed and 1 failed, got %d and %d", preconfirmed, failed) + } + + for i := 0; i < 3; i++ { + env.blockTracker.out <- 1 + } + + cancel() + <-done +} + +// TestBundleDispatchAllOrNothing checks that a batch dispatch is all or +// nothing: when the context ends while the batch waits for its second +// worker slot, no member goroutine starts, no bid goes out, and every +// inflight mark is released so the next pass forms the full bundle +// again. The worker pool keeps exactly one free slot, so the old +// per-member dispatch would start the leader goroutine with that slot +// and place a bundle bid from the pre-fed inputs below. +func TestBundleDispatchAllOrNothing(t *testing.T) { + t.Parallel() + + env := newSenderTestEnv(t, &mockSimulator{}) + + ctx := context.Background() + + txns := []*sender.Transaction{ + newTestTxn(common.HexToAddress("0x1111111111111111111111111111111111111111"), 601, "0xff01", sender.TxTypeRegular), + newTestTxn(common.HexToAddress("0x2222222222222222222222222222222222222222"), 602, "0xff02", sender.TxTypeRegular), + } + for _, txn := range txns { + if err := env.st.AddBalance(ctx, txn.Sender, big.NewInt(5e18)); err != nil { + t.Fatalf("failed to add balance: %v", err) + } + if err := env.sndr.Enqueue(ctx, txn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + } + + // Leave exactly one free worker slot. The two-member batch needs + // two slots, so the all-or-nothing acquisition blocks on the + // second slot until the dispatch context ends. + drain := env.sndr.FillWorkerPoolForTest(1) + defer drain() + + // One full round of inputs. The all-or-nothing dispatch starts no + // goroutine, so nothing consumes them here; the follow-up pass + // does. A leader goroutine started by a per-member dispatch would + // consume them and place a bundle bid. + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + + dispatchCtx, cancelDispatch := context.WithCancel(context.Background()) + defer cancelDispatch() + timer := time.AfterFunc(100*time.Millisecond, cancelDispatch) + defer timer.Stop() + + env.sndr.ProcessQueuedForTest(dispatchCtx) + + select { + case op := <-env.bidder.in: + t.Fatalf("expected no bid after blocked dispatch, got bid with %d raw transactions", len(op.rawTxs)) + default: + } + + // Every inflight mark is released, so the next pass with a free + // worker pool forms the full bundle from the pre-fed inputs. + drain() + + env.sndr.ProcessQueuedForTest(ctx) + + op := <-env.bidder.in + if len(op.rawTxs) != 2 { + t.Fatalf("expected the full bundle on the next pass, got %d raw transactions", len(op.rawTxs)) + } + + resC := make(chan bidder.BidStatus, 2) + for _, provider := range []string{"provider1", "provider2"} { + resC <- bidder.BidStatus{ + Type: bidder.BidStatusCommitment, + Arg: &bidderapiv1.Commitment{ + TxHashes: op.opts.RevertingTxHashes, + BidAmount: op.bidAmount.String(), + BlockNumber: 1, + ProviderAddress: provider, + }, + } + } + close(resC) + env.bidder.out <- bidResponse{statusCh: resC} + + for i := 0; i < 2; i++ { + res := <-env.st.preconfirmedTxns + if res.txn.Status != sender.TxStatusPreConfirmed { + t.Fatalf("expected preconfirmed transaction, got %s", res.txn.Status) + } + } + + for i := 0; i < 2; i++ { + env.blockTracker.out <- 1 + } + + if err := env.sndr.WaitWorkersForTest(); err != nil { + t.Fatalf("worker goroutines failed: %v", err) + } +} + +// TestBundleAbortSoloRetryFirstAttempt checks that a bundle abort with +// less than two survivors does not consume first-attempt semantics: the +// survivor's solo bid sets a real noOfProviders, so a zero-commitment +// outcome is not marked preconfirmed. +func TestBundleAbortSoloRetryFirstAttempt(t *testing.T) { + t.Parallel() + + badTxn := newTestTxn(common.HexToAddress("0x2222222222222222222222222222222222222222"), 702, "0xab02", sender.TxTypeRegular) + env := newSenderTestEnv(t, &mockSimulator{failRaw: badTxn.Raw}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + survivor := newTestTxn(common.HexToAddress("0x1111111111111111111111111111111111111111"), 701, "0xab01", sender.TxTypeRegular) + for _, txn := range []*sender.Transaction{survivor, badTxn} { + if err := env.st.AddBalance(ctx, txn.Sender, big.NewInt(5e18)); err != nil { + t.Fatalf("failed to add balance: %v", err) + } + if err := env.sndr.Enqueue(ctx, txn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + } + + // One bundle attempt that aborts before the bid, then the + // survivor's solo attempt. + for i := 0; i < 2; i++ { + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + } + + done := env.sndr.Start(ctx) + + // The bad member fails simulation, the bundle collapses to one + // member and aborts before the bid. The only bid on the wire is + // the survivor's solo bid. + op := <-env.bidder.in + if len(op.rawTxs) != 1 || op.rawTxs[0] != "ab01" { + t.Fatalf("expected solo bid for the survivor, got %v", op.rawTxs) + } + + // Both members registered an inclusion waiter; feed one signal + // for each before the bid response, so the survivor's loop can + // end through the inclusion path right after the empty bid. + for i := 0; i < 2; i++ { + env.blockTracker.out <- 1 + } + + // The solo bid resolves with zero commitments. + resC := make(chan bidder.BidStatus) + close(resC) + env.bidder.out <- bidResponse{statusCh: resC} + + for i := 0; i < 2; i++ { + res := <-env.st.preconfirmedTxns + switch res.txn.Hash() { + case survivor.Hash(): + if res.txn.Status == sender.TxStatusPreConfirmed { + t.Fatalf("survivor preconfirmed with %d commitments after aborted bundle", len(res.commitments)) + } + if res.txn.Status != sender.TxStatusConfirmed { + t.Fatalf("expected confirmed survivor, got %s", res.txn.Status) + } + case badTxn.Hash(): + if res.txn.Status != sender.TxStatusFailed { + t.Fatalf("expected failed member, got %s", res.txn.Status) + } + default: + t.Fatalf("unexpected stored transaction %s", res.txn.Hash().Hex()) + } + } + + cancel() + <-done +} + +// TestBundleCtxCancelKeepsFastTracked checks that a context end during +// the bundle bid status loop does not revert members that were already +// fast-tracked: they keep their stored preconfirmation, and only the +// members that are not resolved get the error path. The store honors +// the cancelled context, like a real database, so a repeat store on +// the dead context would fail: the fast-tracked member must not repeat +// the store the leader already did, and must not signal a second +// receipt. The unresolved member fails visibly with a terminal write +// that survives the cancellation. +func TestBundleCtxCancelKeepsFastTracked(t *testing.T) { + t.Parallel() + + env := newSenderTestEnv(t, &mockSimulator{}) + env.st.honorCtx = true + + // Distinct gas limits give distinct member costs, so the fast + // track function can single out member A by its own share. + txnA := newTestTxnWithGas(common.HexToAddress("0x1111111111111111111111111111111111111111"), 801, "0xac01", 21000, sender.TxTypeRegular) + txnB := newTestTxnWithGas(common.HexToAddress("0x2222222222222222222222222222222222222222"), 802, "0xac02", 42000, sender.TxTypeRegular) + // The fast track fires for member A on the second commitment. + // Reading A's fast-track store write below then proves that the + // status loop consumed both commitments before the context ends. + costA := bidCost(21000) + env.sndr.SetFastTrackFunc(func(cmts []*bidderapiv1.Commitment, _ bool) bool { + return len(cmts) >= 2 && cmts[0].BidAmount == costA.String() + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + for _, txn := range []*sender.Transaction{txnA, txnB} { + if err := env.st.AddBalance(ctx, txn.Sender, big.NewInt(5e18)); err != nil { + t.Fatalf("failed to add balance: %v", err) + } + if err := env.sndr.Enqueue(ctx, txn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + } + + waitA := env.sndr.WaitForReceiptAvailable(ctx, txnA.Hash()) + + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + + done := env.sndr.Start(ctx) + + op := <-env.bidder.in + if len(op.rawTxs) != 2 { + t.Fatalf("expected 2 raw transactions in bundle bid, got %d", len(op.rawTxs)) + } + + // Both commitments arrive; the first fast-tracks member A. The + // channel stays open, so the status loop still runs when the + // context ends. + resC := make(chan bidder.BidStatus, 2) + for _, provider := range []string{"provider1", "provider2"} { + resC <- bidder.BidStatus{ + Type: bidder.BidStatusCommitment, + Arg: &bidderapiv1.Commitment{ + TxHashes: op.opts.RevertingTxHashes, + BidAmount: op.bidAmount.String(), + BlockNumber: 1, + ProviderAddress: provider, + }, + } + } + env.bidder.out <- bidResponse{statusCh: resC} + + res := <-env.st.preconfirmedTxns + if res.txn.Hash() != txnA.Hash() { + t.Fatalf("expected fast-tracked store write for member A, got %s", res.txn.Hash().Hex()) + } + if res.txn.Status != sender.TxStatusPreConfirmed { + t.Fatalf("expected preconfirmed fast-tracked member, got %s", res.txn.Status) + } + if len(res.commitments) != 2 { + t.Fatalf("expected 2 commitments on fast track, got %d", len(res.commitments)) + } + for _, cmt := range res.commitments { + if cmt.BidAmount != costA.String() { + t.Fatalf("expected member A share %s, got %s", costA, cmt.BidAmount) + } + } + + // The fast track signaled member A's receipt once. A second + // waiter registered now must stay open: a duplicate receipt + // signal after the cancel would close it. + <-waitA + dupA := env.sndr.WaitForReceiptAvailable(ctx, txnA.Hash()) + + cancel() + <-done + + // After shutdown member A keeps its preconfirmation: the leader + // already stored and signaled it at fast-track time, and no + // commitment arrived after that write, so the success gate + // repeats no store write. Member B, which is not resolved, fails + // visibly; its terminal failed write runs on a context that + // survives the cancellation, so it lands even though the store + // honors context cancellation. + var post []result + drained := false + for !drained { + select { + case r := <-env.st.preconfirmedTxns: + post = append(post, r) + default: + drained = true + } + } + if len(post) != 1 { + t.Fatalf("expected one store write after cancel, got %d", len(post)) + } + if post[0].txn.Hash() != txnB.Hash() || post[0].txn.Status != sender.TxStatusFailed { + t.Fatalf( + "expected the failed store write for member B, got %s with status %s", + post[0].txn.Hash().Hex(), post[0].txn.Status, + ) + } + if txnA.Status != sender.TxStatusPreConfirmed { + t.Fatalf("fast-tracked member status reverted to %s", txnA.Status) + } + if txnB.Status != sender.TxStatusFailed { + t.Fatalf("expected failed unresolved member B, got %s", txnB.Status) + } + select { + case <-dupA: + t.Fatal("duplicate receipt signal for fast-tracked member") + default: + } +} + +// TestDepositCtxCancelFailsVisibly checks that a shutdown does not +// mask a preconfirmed deposit as a success. The balance credit runs +// only after the bid loop, and the queue only selects pending +// transactions again. So the deposit must end visibly failed, with no +// balance credit, and the user can retry it. +func TestDepositCtxCancelFailsVisibly(t *testing.T) { + t.Parallel() + + env := newSenderTestEnv(t, &mockSimulator{}) + // The store honors context cancellation, like a real database. + // The terminal failed write must land anyway: the sender writes + // it on a context that survives the cancellation. + env.st.honorCtx = true + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + depositTxn := newTestTxn(common.HexToAddress("0x1111111111111111111111111111111111111111"), 1e18, "0xdd11", sender.TxTypeDeposit) + if err := env.sndr.Enqueue(ctx, depositTxn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + + // Opted-in slot, so the deposit preconfirms after the bid. + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + + done := env.sndr.Start(ctx) + + op := <-env.bidder.in + if len(op.rawTxs) != 1 || op.rawTxs[0] != "dd11" { + t.Fatalf("expected solo deposit bid, got %v", op.rawTxs) + } + + resC := make(chan bidder.BidStatus, 2) + for _, provider := range []string{"provider1", "provider2"} { + resC <- bidder.BidStatus{ + Type: bidder.BidStatusCommitment, + Arg: &bidderapiv1.Commitment{ + TxHashes: []string{depositTxn.Hash().Hex()}, + BidAmount: bundleShare.String(), + BlockNumber: 1, + ProviderAddress: provider, + }, + } + } + close(resC) + env.bidder.out <- bidResponse{statusCh: resC} + + res := <-env.st.preconfirmedTxns + if res.txn.Status != sender.TxStatusPreConfirmed { + t.Fatalf("expected preconfirmed deposit, got %s", res.txn.Status) + } + + // The context ends before inclusion. The deposit has not credited + // its balance yet, so it must not end as a silent preconfirmed + // success that the queue never selects again. + cancel() + <-done + + var writes []result + drained := false + for !drained { + select { + case r := <-env.st.preconfirmedTxns: + writes = append(writes, r) + default: + drained = true + } + } + if len(writes) != 1 { + t.Fatalf("expected one store write after cancel, got %d", len(writes)) + } + if writes[0].txn.Status != sender.TxStatusFailed { + t.Fatalf("expected failed deposit after cancel, got %s", writes[0].txn.Status) + } + if depositTxn.Status != sender.TxStatusFailed { + t.Fatalf("expected failed deposit status, got %s", depositTxn.Status) + } + if count := env.st.addBalanceCount(depositTxn.Sender); count != 0 { + t.Fatalf("expected no balance credit for the cancelled deposit, got %d", count) + } +} + +// TestFastSwapCtxCancelKeepsPreconfirmed checks that a shutdown does +// not revert a preconfirmed fastswap to a failed state. A fastswap has +// no post-loop obligation, so the stored preconfirmation stands: no +// failed store write and no failure notification. +func TestFastSwapCtxCancelKeepsPreconfirmed(t *testing.T) { + t.Parallel() + + env := newSenderTestEnv(t, &mockSimulator{}) + env.st.honorCtx = true + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + swapTxn := newTestTxn(common.HexToAddress("0x1111111111111111111111111111111111111111"), 501, "0xfa01", sender.TxTypeFastSwap) + if err := env.sndr.Enqueue(ctx, swapTxn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + + // Opted-in slot, so the fastswap preconfirms after the bid. + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + + done := env.sndr.Start(ctx) + + op := <-env.bidder.in + if len(op.rawTxs) != 1 || op.rawTxs[0] != "fa01" { + t.Fatalf("expected solo fastswap bid, got %v", op.rawTxs) + } + + resC := make(chan bidder.BidStatus, 2) + for _, provider := range []string{"provider1", "provider2"} { + resC <- bidder.BidStatus{ + Type: bidder.BidStatusCommitment, + Arg: &bidderapiv1.Commitment{ + TxHashes: []string{swapTxn.Hash().Hex()}, + BidAmount: bundleShare.String(), + BlockNumber: 1, + ProviderAddress: provider, + }, + } + } + close(resC) + env.bidder.out <- bidResponse{statusCh: resC} + + res := <-env.st.preconfirmedTxns + if res.txn.Status != sender.TxStatusPreConfirmed { + t.Fatalf("expected preconfirmed fastswap, got %s", res.txn.Status) + } + + // The context ends before inclusion. The fastswap has no + // post-loop work, so the shutdown must not revert the stored + // preconfirmation to a failed state. + cancel() + <-done + + select { + case r := <-env.st.preconfirmedTxns: + t.Fatalf("unexpected store write after cancel with status %s", r.txn.Status) + default: + } + if swapTxn.Status != sender.TxStatusPreConfirmed { + t.Fatalf("expected preconfirmed fastswap after cancel, got %s", swapTxn.Status) + } + if env.notifier.notified(swapTxn.Hash()) { + t.Fatal("unexpected failure notification for the preconfirmed fastswap") + } +} + +// TestBundleLateCommitmentsRestored checks bundle and solo parity for +// commitments that arrive after the fast-track write: the fast-track +// write persists the early set, later commitments append in memory +// only, and the success gate stores the grown set again, on a context +// that survives the shutdown. The unresolved member still fails +// visibly with a terminal write that also survives the shutdown. +func TestBundleLateCommitmentsRestored(t *testing.T) { + t.Parallel() + + env := newSenderTestEnv(t, &mockSimulator{}) + env.st.honorCtx = true + + // Distinct gas limits give distinct member costs, so the fast + // track function can single out member A by its own share. + txnA := newTestTxnWithGas(common.HexToAddress("0x1111111111111111111111111111111111111111"), 811, "0xad01", 21000, sender.TxTypeRegular) + txnB := newTestTxnWithGas(common.HexToAddress("0x2222222222222222222222222222222222222222"), 812, "0xad02", 42000, sender.TxTypeRegular) + + // The fast track fires for member A on the first commitment, so + // the fast-track write persists one commitment and the second + // commitment exists in memory only. The channel signals when the + // second commitment reached member A. + costA := bidCost(21000) + secondSeen := make(chan struct{}) + var secondOnce sync.Once + env.sndr.SetFastTrackFunc(func(cmts []*bidderapiv1.Commitment, _ bool) bool { + if len(cmts) == 0 || cmts[0].BidAmount != costA.String() { + return false + } + if len(cmts) >= 2 { + secondOnce.Do(func() { close(secondSeen) }) + } + return true + }) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + for _, txn := range []*sender.Transaction{txnA, txnB} { + if err := env.st.AddBalance(ctx, txn.Sender, big.NewInt(5e18)); err != nil { + t.Fatalf("failed to add balance: %v", err) + } + if err := env.sndr.Enqueue(ctx, txn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + } + + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + + done := env.sndr.Start(ctx) + + op := <-env.bidder.in + if len(op.rawTxs) != 2 { + t.Fatalf("expected 2 raw transactions in bundle bid, got %d", len(op.rawTxs)) + } + + // Both commitments arrive; the first fast-tracks member A. The + // channel stays open, so the status loop still runs when the + // context ends. + resC := make(chan bidder.BidStatus, 2) + for _, provider := range []string{"provider1", "provider2"} { + resC <- bidder.BidStatus{ + Type: bidder.BidStatusCommitment, + Arg: &bidderapiv1.Commitment{ + TxHashes: op.opts.RevertingTxHashes, + BidAmount: op.bidAmount.String(), + BlockNumber: 1, + ProviderAddress: provider, + }, + } + } + env.bidder.out <- bidResponse{statusCh: resC} + + // The fast-track write persists exactly one commitment. + res := <-env.st.preconfirmedTxns + if res.txn.Hash() != txnA.Hash() { + t.Fatalf("expected fast-track store write for member A, got %s", res.txn.Hash().Hex()) + } + if res.txn.Status != sender.TxStatusPreConfirmed { + t.Fatalf("expected preconfirmed fast-tracked member, got %s", res.txn.Status) + } + if len(res.commitments) != 1 { + t.Fatalf("expected 1 commitment on the fast-track write, got %d", len(res.commitments)) + } + + // The second commitment reached member A in memory only. The + // shutdown ends the status loop. + <-secondSeen + cancel() + <-done + + var writes []result + drained := false + for !drained { + select { + case r := <-env.st.preconfirmedTxns: + writes = append(writes, r) + default: + drained = true + } + } + if len(writes) != 2 { + t.Fatalf("expected two store writes after the fast-track write, got %d", len(writes)) + } + for _, w := range writes { + switch w.txn.Hash() { + case txnA.Hash(): + if w.txn.Status != sender.TxStatusPreConfirmed { + t.Fatalf("expected preconfirmed re-store for member A, got %s", w.txn.Status) + } + if len(w.commitments) != 2 { + t.Fatalf("expected 2 commitments on the re-store, got %d", len(w.commitments)) + } + for _, cmt := range w.commitments { + if cmt.BidAmount != costA.String() { + t.Fatalf("expected member A share %s, got %s", costA, cmt.BidAmount) + } + } + case txnB.Hash(): + if w.txn.Status != sender.TxStatusFailed { + t.Fatalf("expected failed unresolved member B, got %s", w.txn.Status) + } + default: + t.Fatalf("unexpected store write for %s", w.txn.Hash().Hex()) + } + } + if txnA.Status != sender.TxStatusPreConfirmed { + t.Fatalf("fast-tracked member status reverted to %s", txnA.Status) + } +} + +// TestBundleBalanceDropRecordsHistory checks that a terminal pre-check +// drop whose only block attempt was rolled back still notifies the +// user and records the hash as historical: a later queue pass must not +// process the same hash again. +func TestBundleBalanceDropRecordsHistory(t *testing.T) { + t.Parallel() + + env := newSenderTestEnv(t, &mockSimulator{}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // Three members, so the bundle survives the drop: member B has no + // balance and drops terminally; members A and C bid. + txnA := newTestTxn(common.HexToAddress("0x1111111111111111111111111111111111111111"), 951, "0xba01", sender.TxTypeRegular) + txnB := newTestTxn(common.HexToAddress("0x2222222222222222222222222222222222222222"), 952, "0xba02", sender.TxTypeRegular) + txnC := newTestTxn(common.HexToAddress("0x3333333333333333333333333333333333333333"), 953, "0xba03", sender.TxTypeRegular) + + inclA := env.blockTracker.inclusionFor(txnA.Hash()) + env.blockTracker.inclusionFor(txnB.Hash()) + inclC := env.blockTracker.inclusionFor(txnC.Hash()) + + for _, txn := range []*sender.Transaction{txnA, txnC} { + if err := env.st.AddBalance(ctx, txn.Sender, big.NewInt(5e18)); err != nil { + t.Fatalf("failed to add balance: %v", err) + } + } + for _, txn := range []*sender.Transaction{txnA, txnB, txnC} { + if err := env.sndr.Enqueue(ctx, txn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + } + + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + + env.sndr.ProcessQueuedForTest(ctx) + + op := <-env.bidder.in + if len(op.rawTxs) != 2 { + t.Fatalf("expected 2 raw transactions in bundle bid, got %d", len(op.rawTxs)) + } + for _, r := range op.rawTxs { + if r == "ba02" { + t.Fatal("member without balance must not appear in the bundle bid") + } + } + + resC := make(chan bidder.BidStatus, 2) + for _, provider := range []string{"provider1", "provider2"} { + resC <- bidder.BidStatus{ + Type: bidder.BidStatusCommitment, + Arg: &bidderapiv1.Commitment{ + TxHashes: op.opts.RevertingTxHashes, + BidAmount: op.bidAmount.String(), + BlockNumber: 1, + ProviderAddress: provider, + }, + } + } + close(resC) + env.bidder.out <- bidResponse{statusCh: resC} + + inclA <- 1 + inclC <- 1 + + if err := env.sndr.WaitWorkersForTest(); err != nil { + t.Fatalf("worker goroutines failed: %v", err) + } + + failed := 0 + for i := 0; i < 3; i++ { + res := <-env.st.preconfirmedTxns + if res.txn.Hash() == txnB.Hash() { + if res.txn.Status != sender.TxStatusFailed { + t.Fatalf("expected failed dropped member, got %s", res.txn.Status) + } + failed++ + } + } + if failed != 1 { + t.Fatalf("expected one failed store write for the dropped member, got %d", failed) + } + + // The drop rolled its only block attempt back, so the attempt + // entry is gone. The user notification must still happen. + if !env.notifier.notified(txnB.Hash()) { + t.Fatal("expected a notification for the dropped member") + } + + // The historical record keeps a later queue pass from processing + // the same hash again: no bid goes out and no store write runs. + if err := env.st.AddQueuedTransaction(ctx, txnB); err != nil { + t.Fatalf("failed to re-queue transaction: %v", err) + } + env.sndr.ProcessQueuedForTest(ctx) + if err := env.sndr.WaitWorkersForTest(); err != nil { + t.Fatalf("worker goroutines failed: %v", err) + } + select { + case op := <-env.bidder.in: + t.Fatalf("historical transaction was bid again: %v", op.rawTxs) + default: + } + select { + case r := <-env.st.preconfirmedTxns: + t.Fatalf("unexpected store write for the historical transaction with status %s", r.txn.Status) + default: + } +} + +// TestBundlePreCheckDropSoloRetryFirstAttempt checks that a member +// dropped by a bundle pre-check before the bid keeps first-attempt +// semantics: the drop rolls its block attempt back, so the solo retry +// sets a real provider count and a zero-commitment outcome is not +// marked preconfirmed. +func TestBundlePreCheckDropSoloRetryFirstAttempt(t *testing.T) { + t.Parallel() + + env := newSenderTestEnv(t, &mockSimulator{}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + txnA := newTestTxnWithGasPrice(common.HexToAddress("0x1111111111111111111111111111111111111111"), 901, "0xae01", 10) + txnB := newTestTxnWithGasPrice(common.HexToAddress("0x2222222222222222222222222222222222222222"), 902, "0xae02", 1) + txnC := newTestTxnWithGasPrice(common.HexToAddress("0x3333333333333333333333333333333333333333"), 903, "0xae03", 10) + + // The next base fee is above member B's fee per gas, so the fee + // pre-check drops member B from the bundle before the bid. + env.blockTracker.setNextBaseFee(big.NewInt(5)) + + inclA := env.blockTracker.inclusionFor(txnA.Hash()) + inclB := env.blockTracker.inclusionFor(txnB.Hash()) + inclC := env.blockTracker.inclusionFor(txnC.Hash()) + + for _, txn := range []*sender.Transaction{txnA, txnB, txnC} { + if err := env.st.AddBalance(ctx, txn.Sender, big.NewInt(5e18)); err != nil { + t.Fatalf("failed to add balance: %v", err) + } + if err := env.sndr.Enqueue(ctx, txn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + } + + // Bundle round: not opted in, with a short next block time so + // member B's solo retry runs after about one second. + env.bidder.optinEstimate <- 20 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: time.Second} + env.pricer.out <- testPrices() + + done := env.sndr.Start(ctx) + + op := <-env.bidder.in + if len(op.rawTxs) != 2 { + t.Fatalf("expected 2 raw transactions in bundle bid, got %d", len(op.rawTxs)) + } + for _, r := range op.rawTxs { + if r == "ae02" { + t.Fatal("dropped member must not appear in the bundle bid") + } + } + + // Clear the base fee so member B's solo retry passes the fee + // check, and feed the solo round: opted in, so a zero-commitment + // retry that wrongly compares zero providers with zero + // commitments would be stored as preconfirmed. + env.blockTracker.setNextBaseFee(nil) + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + + // The bundle members A and C get both commitments. The slot is + // not opted in, so they end through the inclusion path. + resC := make(chan bidder.BidStatus, 2) + for _, provider := range []string{"provider1", "provider2"} { + resC <- bidder.BidStatus{ + Type: bidder.BidStatusCommitment, + Arg: &bidderapiv1.Commitment{ + TxHashes: op.opts.RevertingTxHashes, + BidAmount: op.bidAmount.String(), + BlockNumber: 1, + ProviderAddress: provider, + }, + } + } + close(resC) + env.bidder.out <- bidResponse{statusCh: resC} + + inclA <- 1 + inclC <- 1 + + // Member B retries solo after the fee drop. + opB := <-env.bidder.in + if len(opB.rawTxs) != 1 || opB.rawTxs[0] != "ae02" { + t.Fatalf("expected solo bid for the dropped member, got %v", opB.rawTxs) + } + + // The solo retry resolves with zero commitments. + resB := make(chan bidder.BidStatus) + close(resB) + env.bidder.out <- bidResponse{statusCh: resB} + + inclB <- 1 + + for i := 0; i < 3; i++ { + res := <-env.st.preconfirmedTxns + if res.txn.Hash() == txnB.Hash() && res.txn.Status == sender.TxStatusPreConfirmed { + t.Fatalf("dropped member preconfirmed with %d commitments", len(res.commitments)) + } + if res.txn.Status != sender.TxStatusConfirmed { + t.Fatalf("expected confirmed transaction %s, got %s", res.txn.Hash().Hex(), res.txn.Status) + } + } + + cancel() + <-done +} + +func TestIneligibleTxnsNeverBundle(t *testing.T) { + t.Parallel() + + env := newSenderTestEnv(t, &mockSimulator{}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + regularTxn := newTestTxn(common.HexToAddress("0x1111111111111111111111111111111111111111"), 401, "0xdd01", sender.TxTypeRegular) + depositTxn := newTestTxn(common.HexToAddress("0x2222222222222222222222222222222222222222"), 1e18, "0xdd02", sender.TxTypeDeposit) + constraintTxn := newTestTxn(common.HexToAddress("0x3333333333333333333333333333333333333333"), 403, "0xdd03", sender.TxTypeRegular) + constraintTxn.Constraint = &bidderapiv1.PositionConstraint{ + Anchor: bidderapiv1.PositionConstraint_ANCHOR_TOP, + Basis: bidderapiv1.PositionConstraint_BASIS_PERCENTILE, + Value: 10, + } + + txns := []*sender.Transaction{regularTxn, depositTxn, constraintTxn} + for _, txn := range txns { + if err := env.st.AddBalance(ctx, txn.Sender, big.NewInt(5e18)); err != nil { + t.Fatalf("failed to add balance: %v", err) + } + if err := env.sndr.Enqueue(ctx, txn); err != nil { + t.Fatalf("failed to enqueue transaction: %v", err) + } + } + + // Pre-feed three solo attempts. No bundle forms because only one + // transaction is eligible. + for i := 0; i < 3; i++ { + env.bidder.optinEstimate <- 7 + env.blockTracker.bnOut <- blockNoOp{block: 1, timeTillNextBlock: 5 * time.Second} + env.pricer.out <- testPrices() + } + + done := env.sndr.Start(ctx) + + for i := 0; i < 3; i++ { + op := <-env.bidder.in + if len(op.rawTxs) != 1 { + t.Fatalf("expected solo bid, got %d raw transactions", len(op.rawTxs)) + } + if op.rawTxs[0] == "dd03" && op.opts.Constraint == nil { + t.Fatal("expected constraint on constrained transaction bid") + } + + resC := make(chan bidder.BidStatus, 3) + for _, provider := range []string{"provider1", "provider2"} { + resC <- bidder.BidStatus{ + Type: bidder.BidStatusCommitment, + Arg: &bidderapiv1.Commitment{ + BidAmount: bundleShare.String(), + BlockNumber: 1, + ProviderAddress: provider, + }, + } + } + close(resC) + env.bidder.out <- bidResponse{statusCh: resC} + } + + for i := 0; i < 3; i++ { + res := <-env.st.preconfirmedTxns + if res.txn.Status != sender.TxStatusPreConfirmed { + t.Fatalf("expected preconfirmed transaction, got %s", res.txn.Status) + } + } + + for i := 0; i < 3; i++ { + env.blockTracker.out <- 1 + } + + cancel() + <-done +} diff --git a/tools/preconf-rpc/service/service.go b/tools/preconf-rpc/service/service.go index f2f6a2377..d87a9be62 100644 --- a/tools/preconf-rpc/service/service.go +++ b/tools/preconf-rpc/service/service.go @@ -81,6 +81,7 @@ type Config struct { Token string SimulatorURLs []string UseInlineSimulation bool + BundleBids bool BackrunnerRPC string BackrunnerAPIURL string BackrunnerAPIKey string @@ -355,6 +356,7 @@ func New(config *Config) (*Service, error) { settlementChainID, expSubmitter, config.LogEncryptionKey, + config.BundleBids, config.Logger.With("module", "txsender"), ) if err != nil { diff --git a/tools/preconf-rpc/settlement-tracker/tracker.go b/tools/preconf-rpc/settlement-tracker/tracker.go index edfadbaf0..369a9ecdc 100644 --- a/tools/preconf-rpc/settlement-tracker/tracker.go +++ b/tools/preconf-rpc/settlement-tracker/tracker.go @@ -4,12 +4,27 @@ import ( "context" "log/slog" "math/big" + "strings" + "sync" + "time" "github.com/ethereum/go-ethereum/common" "github.com/primev/mev-commit/tools/preconf-rpc/bidder" "golang.org/x/sync/errgroup" ) +// splitSourceCap bounds the split decision memo. Each entry serves one +// bundle payment notification burst and is evicted after one hit per +// bundle member. The cap protects the memo when member messages get +// lost and their entry never reaches its eviction count. +const splitSourceCap = 1024 + +// resubscribeBackoff is the wait before a new subscription after a +// subscription channel closed. A dead bidder connection returns a +// channel that closes at once; the wait keeps the subscribe loop from +// spinning against it. +const resubscribeBackoff = 5 * time.Second + type BidderClient interface { SubscribeSettlements(ctx context.Context) <-chan bidder.SettlementMsg SubscribePayments(ctx context.Context) <-chan bidder.PaymentMsg @@ -28,12 +43,34 @@ type Store interface { payment *big.Int, refund *big.Int, ) error + GetBidAmountsForTxns( + ctx context.Context, + txnHashes []common.Hash, + ) (map[common.Hash]*big.Int, error) +} + +// splitSource is the memoized split decision for one bundle payment +// notification burst. A nil amounts map marks the equal split path. +// The hits count tracks how many member messages of the burst used +// the decision; the entry is evicted after one hit per member. +type splitSource struct { + amounts map[common.Hash]*big.Int + total *big.Int + hits int } type tracker struct { client BidderClient store Store logger *slog.Logger + + // backoff is the wait before a resubscribe after a subscription + // channel closed. Tests shorten it. + backoff time.Duration + + splitMu sync.Mutex + splitSources map[string]*splitSource + splitOrder []string } func NewTracker( @@ -42,9 +79,11 @@ func NewTracker( logger *slog.Logger, ) *tracker { return &tracker{ - client: client, - store: store, - logger: logger, + client: client, + store: store, + logger: logger, + backoff: resubscribeBackoff, + splitSources: make(map[string]*splitSource), } } @@ -61,11 +100,21 @@ func (t *tracker) Start(ctx context.Context) <-chan struct{} { } sub := t.client.SubscribeSettlements(egCtx) + SETTLEMENTS: for { select { case <-egCtx.Done(): return egCtx.Err() - case msg := <-sub: + case msg, more := <-sub: + if !more { + // The subscription channel closed. Leave the + // receive loop, so the outer loop checks the + // context and subscribes again. A receive + // from the closed channel would yield endless + // zero-value messages for the zero hash. + t.logger.Warn("Settlement subscription closed, subscribing again") + break SETTLEMENTS + } // Process settlement message if err := t.store.UpdateSettlementStatus( egCtx, @@ -89,6 +138,14 @@ func (t *tracker) Start(ctx context.Context) <-chan struct{} { } } } + // The subscription channel closed. Wait before the next + // subscription, so a dead bidder connection does not + // cause a tight subscribe loop. + select { + case <-egCtx.Done(): + return egCtx.Err() + case <-time.After(t.backoff): + } } }) @@ -101,35 +158,54 @@ func (t *tracker) Start(ctx context.Context) <-chan struct{} { } sub := t.client.SubscribePayments(egCtx) + PAYMENTS: for { select { case <-egCtx.Done(): return egCtx.Err() - case msg := <-sub: + case msg, more := <-sub: + if !more { + // The subscription channel closed. Leave the + // receive loop, so the outer loop checks the + // context and subscribes again. A receive + // from the closed channel would yield endless + // zero-value messages for the zero hash. + t.logger.Warn("Payment subscription closed, subscribing again") + break PAYMENTS + } // Process payment message + payment, refund := t.splitPaymentShare(egCtx, msg) if err := t.store.UpdateSettlementPayment( egCtx, common.HexToHash(msg.TransactionHash), - msg.Payment, - msg.Refund, + payment, + refund, ); err != nil { t.logger.Error( "Failed to update settlement payment", "error", err, "txnHash", msg.TransactionHash, - "payment", msg.Payment, - "refund", msg.Refund, + "payment", payment, + "refund", refund, ) } else { t.logger.Info( "Updated settlement payment", "txnHash", msg.TransactionHash, - "payment", msg.Payment, - "refund", msg.Refund, + "payment", payment, + "refund", refund, ) } } } + // The subscription channel closed. Wait before the next + // subscription, so a dead bidder connection does not + // cause a tight subscribe loop. + select { + case <-egCtx.Done(): + return egCtx.Err() + case <-time.After(t.backoff): + } } }) @@ -142,3 +218,159 @@ func (t *tracker) Start(ctx context.Context) <-chan struct{} { }() return done } + +// splitPaymentShare computes the share of a payment message for its +// transaction. A bundle bid produces one aggregate payment for all bundle +// transactions. The share of each transaction is proportional to its +// stored bid amount. If the shares cannot be loaded, the aggregate is +// split equally between the bundle members. All members of one +// notification burst use one memoized split source: the first member +// message resolves the decision and the later member messages of the +// same burst reuse it, so pro-rata and equal shares never mix within +// one burst. In both cases the member with the first bundle hash adds +// the division remainder, so the sum of the member shares equals the +// aggregate exactly within each burst and the group is never charged +// more than the aggregate. Single transaction bids keep the full +// amount. +func (t *tracker) splitPaymentShare(ctx context.Context, msg bidder.PaymentMsg) (*big.Int, *big.Int) { + if len(msg.BundleTxnHashes) < 2 { + return msg.Payment, msg.Refund + } + + hashes := make([]common.Hash, 0, len(msg.BundleTxnHashes)) + for _, h := range msg.BundleTxnHashes { + hashes = append(hashes, common.HexToHash(h)) + } + own := common.HexToHash(msg.TransactionHash) + + src := t.splitSourceFor(ctx, msg, hashes) + if src.amounts == nil { + return equalShare(msg.Payment, own, hashes), equalShare(msg.Refund, own, hashes) + } + + payment := proRataShare(msg.Payment, src.amounts, hashes, own, src.total) + refund := proRataShare(msg.Refund, src.amounts, hashes, own, src.total) + return payment, refund +} + +// splitSourceFor returns the split decision for one bundle payment +// notification burst. The burst key combines the bundle hash list and +// the aggregate amounts. The first member message of a burst resolves +// the decision; every later member message of the same burst reuses +// it. An entry is evicted after one hit per bundle member. The memo +// is capped; when it is full, the oldest entry is evicted first. +func (t *tracker) splitSourceFor(ctx context.Context, msg bidder.PaymentMsg, hashes []common.Hash) *splitSource { + key := strings.Join(msg.BundleTxnHashes, ",") + "|" + msg.Payment.String() + "|" + msg.Refund.String() + + t.splitMu.Lock() + defer t.splitMu.Unlock() + + if src, found := t.splitSources[key]; found { + src.hits++ + if src.hits >= len(msg.BundleTxnHashes) { + t.evictSplitSource(key) + } + return src + } + + src := t.resolveSplitSource(ctx, msg, hashes) + src.hits = 1 + if len(t.splitOrder) >= splitSourceCap { + t.evictSplitSource(t.splitOrder[0]) + } + t.splitSources[key] = src + t.splitOrder = append(t.splitOrder, key) + return src +} + +// evictSplitSource removes one memo entry and its insertion order +// record. The caller holds splitMu. +func (t *tracker) evictSplitSource(key string) { + delete(t.splitSources, key) + order := make([]string, 0, len(t.splitOrder)) + for _, k := range t.splitOrder { + if k != key { + order = append(order, k) + } + } + t.splitOrder = order +} + +// resolveSplitSource loads the stored bid amounts for one bundle and +// builds the split decision: pro-rata over the stored amounts, or the +// equal split marker when the amounts cannot be loaded in full. +func (t *tracker) resolveSplitSource(ctx context.Context, msg bidder.PaymentMsg, hashes []common.Hash) *splitSource { + amounts, err := t.store.GetBidAmountsForTxns(ctx, hashes) + if err != nil { + t.logger.Error( + "Failed to get bid amounts for bundle, using equal split", + "error", err, + "txnHash", msg.TransactionHash, + ) + return &splitSource{} + } + + total := big.NewInt(0) + for _, h := range hashes { + amount, found := amounts[h] + if !found { + t.logger.Error( + "Missing bid amount for bundle member, using equal split", + "txnHash", h.Hex(), + ) + return &splitSource{} + } + total = new(big.Int).Add(total, amount) + } + + if total.Sign() <= 0 { + t.logger.Error( + "Non-positive bid amount total for bundle, using equal split", + "txnHash", msg.TransactionHash, + ) + return &splitSource{} + } + + return &splitSource{amounts: amounts, total: total} +} + +// equalShare splits an aggregate amount into equal parts, one for each +// bundle hash. The member with the first bundle hash adds the division +// remainder. Every member computes its share independently and the +// shares sum to the aggregate exactly. +func equalShare(aggregate *big.Int, own common.Hash, hashes []common.Hash) *big.Int { + count := big.NewInt(int64(len(hashes))) + share, remainder := new(big.Int).QuoRem(aggregate, count, new(big.Int)) + if own == hashes[0] { + return new(big.Int).Add(share, remainder) + } + return share +} + +// proRataShare computes one member's part of an aggregate amount, in +// proportion to the stored member bid amounts. The floor divisions can +// leave a remainder; the member with the first bundle hash adds it. +// Every member computes its share independently and the shares sum to +// the aggregate exactly. +func proRataShare( + aggregate *big.Int, + amounts map[common.Hash]*big.Int, + hashes []common.Hash, + own common.Hash, + total *big.Int, +) *big.Int { + share := big.NewInt(0) + floorSum := big.NewInt(0) + for _, h := range hashes { + memberShare := new(big.Int).Div(new(big.Int).Mul(aggregate, amounts[h]), total) + floorSum = new(big.Int).Add(floorSum, memberShare) + if h == own { + share = memberShare + } + } + if own == hashes[0] { + remainder := new(big.Int).Sub(aggregate, floorSum) + return new(big.Int).Add(share, remainder) + } + return share +} diff --git a/tools/preconf-rpc/settlement-tracker/tracker_test.go b/tools/preconf-rpc/settlement-tracker/tracker_test.go new file mode 100644 index 000000000..5d711e852 --- /dev/null +++ b/tools/preconf-rpc/settlement-tracker/tracker_test.go @@ -0,0 +1,419 @@ +package tracker + +import ( + "context" + "errors" + "io" + "log/slog" + "math/big" + "sync" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/primev/mev-commit/tools/preconf-rpc/bidder" +) + +type stubStore struct { + amounts map[common.Hash]*big.Int + err error +} + +func (s *stubStore) UpdateSettlementStatus( + _ context.Context, + _ common.Hash, + _ bool, + _ common.Address, +) error { + return nil +} + +func (s *stubStore) UpdateSettlementPayment( + _ context.Context, + _ common.Hash, + _ *big.Int, + _ *big.Int, +) error { + return nil +} + +func (s *stubStore) GetBidAmountsForTxns( + _ context.Context, + _ []common.Hash, +) (map[common.Hash]*big.Int, error) { + if s.err != nil { + return nil, s.err + } + return s.amounts, nil +} + +func testLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +// flakyStore returns the bid amounts on the first lookup and errors on +// every later lookup. +type flakyStore struct { + stubStore + mu sync.Mutex + calls int +} + +func (s *flakyStore) GetBidAmountsForTxns( + _ context.Context, + _ []common.Hash, +) (map[common.Hash]*big.Int, error) { + s.mu.Lock() + defer s.mu.Unlock() + + s.calls++ + if s.calls > 1 { + return nil, errors.New("transient lookup failure") + } + return s.amounts, nil +} + +func (s *flakyStore) callCount() int { + s.mu.Lock() + defer s.mu.Unlock() + + return s.calls +} + +// stubClient serves one closed subscription channel per topic first, +// and then open channels that never send. It signals when both loops +// have subscribed again after their closed channel, and records the +// time of every subscription per topic. +type stubClient struct { + mu sync.Mutex + settlementSubs int + paymentSubs int + settlementTimes []time.Time + paymentTimes []time.Time + resubscribed chan struct{} + signalled bool +} + +// maybeSignal runs with the mutex held. +func (c *stubClient) maybeSignal() { + if !c.signalled && c.settlementSubs >= 2 && c.paymentSubs >= 2 { + c.signalled = true + close(c.resubscribed) + } +} + +func (c *stubClient) SubscribeSettlements(_ context.Context) <-chan bidder.SettlementMsg { + c.mu.Lock() + defer c.mu.Unlock() + + c.settlementSubs++ + c.settlementTimes = append(c.settlementTimes, time.Now()) + if c.settlementSubs == 1 { + ch := make(chan bidder.SettlementMsg) + close(ch) + return ch + } + c.maybeSignal() + return make(chan bidder.SettlementMsg) +} + +func (c *stubClient) SubscribePayments(_ context.Context) <-chan bidder.PaymentMsg { + c.mu.Lock() + defer c.mu.Unlock() + + c.paymentSubs++ + c.paymentTimes = append(c.paymentTimes, time.Now()) + if c.paymentSubs == 1 { + ch := make(chan bidder.PaymentMsg) + close(ch) + return ch + } + c.maybeSignal() + return make(chan bidder.PaymentMsg) +} + +// recordingStore records every settlement and payment update. +type recordingStore struct { + stubStore + mu sync.Mutex + statusHashes []common.Hash + paymentHashes []common.Hash +} + +func (s *recordingStore) UpdateSettlementStatus( + _ context.Context, + txnHash common.Hash, + _ bool, + _ common.Address, +) error { + s.mu.Lock() + defer s.mu.Unlock() + + s.statusHashes = append(s.statusHashes, txnHash) + return nil +} + +func (s *recordingStore) UpdateSettlementPayment( + _ context.Context, + txnHash common.Hash, + _ *big.Int, + _ *big.Int, +) error { + s.mu.Lock() + defer s.mu.Unlock() + + s.paymentHashes = append(s.paymentHashes, txnHash) + return nil +} + +// TestSplitPaymentShareFallbackEqualSplit checks the fallback path: when +// the bid amount lookup fails, a bundle payment splits equally between +// the members, the first bundle hash carries the division remainder, +// and the member shares sum to the aggregate exactly. +func TestSplitPaymentShareFallbackEqualSplit(t *testing.T) { + t.Parallel() + + hashes := []string{ + common.HexToHash("0x01").Hex(), + common.HexToHash("0x02").Hex(), + common.HexToHash("0x03").Hex(), + } + tr := NewTracker(nil, &stubStore{err: errors.New("lookup failed")}, testLogger()) + + wantPayments := []*big.Int{big.NewInt(34), big.NewInt(33), big.NewInt(33)} + wantRefunds := []*big.Int{big.NewInt(4), big.NewInt(3), big.NewInt(3)} + + totalPayment := big.NewInt(0) + totalRefund := big.NewInt(0) + for i, h := range hashes { + payment, refund := tr.splitPaymentShare(context.Background(), bidder.PaymentMsg{ + TransactionHash: h, + Payment: big.NewInt(100), + Refund: big.NewInt(10), + BundleTxnHashes: hashes, + }) + if payment.Cmp(wantPayments[i]) != 0 { + t.Fatalf("expected payment share %s for member %d, got %s", wantPayments[i], i, payment) + } + if refund.Cmp(wantRefunds[i]) != 0 { + t.Fatalf("expected refund share %s for member %d, got %s", wantRefunds[i], i, refund) + } + totalPayment = new(big.Int).Add(totalPayment, payment) + totalRefund = new(big.Int).Add(totalRefund, refund) + } + if totalPayment.Cmp(big.NewInt(100)) != 0 { + t.Fatalf("expected payment shares to sum to 100, got %s", totalPayment) + } + if totalRefund.Cmp(big.NewInt(10)) != 0 { + t.Fatalf("expected refund shares to sum to 10, got %s", totalRefund) + } +} + +// TestSplitPaymentShareMemoizedAcrossBurst checks that every member of +// one bundle payment notification burst uses the same split source: a +// transient store error after the first member message must not mix +// pro-rata and equal shares, and the shares sum to the aggregate +// exactly. It also checks that the memo entry is evicted after one +// hit per member, so a later burst with the same key resolves fresh. +func TestSplitPaymentShareMemoizedAcrossBurst(t *testing.T) { + t.Parallel() + + hashes := []string{ + common.HexToHash("0x01").Hex(), + common.HexToHash("0x02").Hex(), + common.HexToHash("0x03").Hex(), + } + amounts := map[common.Hash]*big.Int{ + common.HexToHash(hashes[0]): big.NewInt(3), + common.HexToHash(hashes[1]): big.NewInt(5), + common.HexToHash(hashes[2]): big.NewInt(7), + } + st := &flakyStore{stubStore: stubStore{amounts: amounts}} + tr := NewTracker(nil, st, testLogger()) + + // The memoized pro-rata result of the single lookup covers the + // full burst: 100 over 3:5:7 floors to 20, 33 and 46, and the + // first hash adds the remainder 1. Refund 10 floors to 2, 3 and + // 4, and the first hash adds the remainder 1. + wantPayments := []*big.Int{big.NewInt(21), big.NewInt(33), big.NewInt(46)} + wantRefunds := []*big.Int{big.NewInt(3), big.NewInt(3), big.NewInt(4)} + + totalPayment := big.NewInt(0) + totalRefund := big.NewInt(0) + for i, h := range hashes { + payment, refund := tr.splitPaymentShare(context.Background(), bidder.PaymentMsg{ + TransactionHash: h, + Payment: big.NewInt(100), + Refund: big.NewInt(10), + BundleTxnHashes: hashes, + }) + if payment.Cmp(wantPayments[i]) != 0 { + t.Fatalf("expected payment share %s for member %d, got %s", wantPayments[i], i, payment) + } + if refund.Cmp(wantRefunds[i]) != 0 { + t.Fatalf("expected refund share %s for member %d, got %s", wantRefunds[i], i, refund) + } + totalPayment = new(big.Int).Add(totalPayment, payment) + totalRefund = new(big.Int).Add(totalRefund, refund) + } + if totalPayment.Cmp(big.NewInt(100)) != 0 { + t.Fatalf("expected payment shares to sum to 100, got %s", totalPayment) + } + if totalRefund.Cmp(big.NewInt(10)) != 0 { + t.Fatalf("expected refund shares to sum to 10, got %s", totalRefund) + } + if count := st.callCount(); count != 1 { + t.Fatalf("expected one store lookup for the burst, got %d", count) + } + + // The burst used up its memo entry, so the same key resolves + // fresh: the store errors now and the whole new burst takes the + // equal split path. + payment, refund := tr.splitPaymentShare(context.Background(), bidder.PaymentMsg{ + TransactionHash: hashes[0], + Payment: big.NewInt(100), + Refund: big.NewInt(10), + BundleTxnHashes: hashes, + }) + if payment.Cmp(big.NewInt(34)) != 0 { + t.Fatalf("expected equal payment share 34 after eviction, got %s", payment) + } + if refund.Cmp(big.NewInt(4)) != 0 { + t.Fatalf("expected equal refund share 4 after eviction, got %s", refund) + } + if count := st.callCount(); count != 2 { + t.Fatalf("expected a fresh store lookup after eviction, got %d", count) + } +} + +// TestClosedSubscriptionResubscribes checks that a closed subscription +// channel does not spin the tracker: the receive loops exit, no store +// update runs for the zero hash, and the outer loops subscribe again. +func TestClosedSubscriptionResubscribes(t *testing.T) { + t.Parallel() + + client := &stubClient{resubscribed: make(chan struct{})} + st := &recordingStore{} + tr := NewTracker(client, st, testLogger()) + // A short backoff keeps the test fast; the backoff itself is + // covered by TestResubscribeWaitsBackoff. + tr.backoff = 10 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + done := tr.Start(ctx) + + select { + case <-client.resubscribed: + case <-time.After(5 * time.Second): + t.Fatal("tracker did not subscribe again after the closed channel") + } + + cancel() + <-done + + st.mu.Lock() + defer st.mu.Unlock() + if len(st.statusHashes) != 0 { + t.Fatalf( + "expected no settlement updates from the closed channel, got %d (first %s)", + len(st.statusHashes), st.statusHashes[0].Hex(), + ) + } + if len(st.paymentHashes) != 0 { + t.Fatalf( + "expected no payment updates from the closed channel, got %d (first %s)", + len(st.paymentHashes), st.paymentHashes[0].Hex(), + ) + } +} + +// TestResubscribeWaitsBackoff checks that the tracker waits for the +// resubscribe backoff after a subscription channel closed: the second +// subscription of each topic starts only after the backoff, so a dead +// bidder connection cannot cause a tight subscribe loop. +func TestResubscribeWaitsBackoff(t *testing.T) { + t.Parallel() + + client := &stubClient{resubscribed: make(chan struct{})} + tr := NewTracker(client, &recordingStore{}, testLogger()) + tr.backoff = 100 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + done := tr.Start(ctx) + + select { + case <-client.resubscribed: + case <-time.After(5 * time.Second): + t.Fatal("tracker did not subscribe again after the closed channel") + } + + cancel() + <-done + + client.mu.Lock() + defer client.mu.Unlock() + for topic, times := range map[string][]time.Time{ + "settlement": client.settlementTimes, + "payment": client.paymentTimes, + } { + if len(times) < 2 { + t.Fatalf("expected at least two %s subscriptions, got %d", topic, len(times)) + } + if gap := times[1].Sub(times[0]); gap < tr.backoff { + t.Fatalf( + "expected the second %s subscription after at least %s, got %s", + topic, tr.backoff, gap, + ) + } + } +} + +// TestSplitPaymentShareProRataDust checks the pro-rata path with +// amounts that do not divide evenly: the shares sum to the aggregate +// exactly and the first bundle hash carries the dust. +func TestSplitPaymentShareProRataDust(t *testing.T) { + t.Parallel() + + hashes := []string{ + common.HexToHash("0x01").Hex(), + common.HexToHash("0x02").Hex(), + common.HexToHash("0x03").Hex(), + } + amounts := map[common.Hash]*big.Int{ + common.HexToHash(hashes[0]): big.NewInt(3), + common.HexToHash(hashes[1]): big.NewInt(5), + common.HexToHash(hashes[2]): big.NewInt(7), + } + tr := NewTracker(nil, &stubStore{amounts: amounts}, testLogger()) + + // Payment 100 over amounts 3:5:7 floors to 20, 33 and 46; the + // first hash adds the remainder 1. Refund 10 floors to 2, 3 and + // 4; the first hash adds the remainder 1. + wantPayments := []*big.Int{big.NewInt(21), big.NewInt(33), big.NewInt(46)} + wantRefunds := []*big.Int{big.NewInt(3), big.NewInt(3), big.NewInt(4)} + + totalPayment := big.NewInt(0) + totalRefund := big.NewInt(0) + for i, h := range hashes { + payment, refund := tr.splitPaymentShare(context.Background(), bidder.PaymentMsg{ + TransactionHash: h, + Payment: big.NewInt(100), + Refund: big.NewInt(10), + BundleTxnHashes: hashes, + }) + if payment.Cmp(wantPayments[i]) != 0 { + t.Fatalf("expected payment share %s for member %d, got %s", wantPayments[i], i, payment) + } + if refund.Cmp(wantRefunds[i]) != 0 { + t.Fatalf("expected refund share %s for member %d, got %s", wantRefunds[i], i, refund) + } + totalPayment = new(big.Int).Add(totalPayment, payment) + totalRefund = new(big.Int).Add(totalRefund, refund) + } + if totalPayment.Cmp(big.NewInt(100)) != 0 { + t.Fatalf("expected payment shares to sum to 100, got %s", totalPayment) + } + if totalRefund.Cmp(big.NewInt(10)) != 0 { + t.Fatalf("expected refund shares to sum to 10, got %s", totalRefund) + } +} diff --git a/tools/preconf-rpc/store/store.go b/tools/preconf-rpc/store/store.go index 86b0b050b..833caf4d3 100644 --- a/tools/preconf-rpc/store/store.go +++ b/tools/preconf-rpc/store/store.go @@ -39,13 +39,40 @@ CREATE TABLE IF NOT EXISTS mcTransactions ( var commitmentsTable = ` CREATE TABLE IF NOT EXISTS commitments ( - commitment_digest TEXT PRIMARY KEY, + commitment_digest TEXT, transaction_hash TEXT, provider_address TEXT, commitment_data BYTEA, + PRIMARY KEY (commitment_digest, transaction_hash), FOREIGN KEY (transaction_hash) REFERENCES mcTransactions (hash) ON DELETE CASCADE );` +// commitmentsPKMigration widens the primary key of the commitments table +// from (commitment_digest) to (commitment_digest, transaction_hash). One +// bundle bid produces one commitment digest that covers many transactions, +// so the digest alone is not unique. The block is idempotent: it only +// alters the table when the current primary key differs from the target. +var commitmentsPKMigration = ` +DO $$ +DECLARE + pk_name TEXT; + pk_cols TEXT[]; +BEGIN + SELECT c.conname, + (SELECT array_agg(a.attname::text ORDER BY k.ord) + FROM unnest(c.conkey) WITH ORDINALITY AS k(attnum, ord) + JOIN pg_attribute a ON a.attrelid = c.conrelid AND a.attnum = k.attnum) + INTO pk_name, pk_cols + FROM pg_constraint c + WHERE c.conrelid = 'commitments'::regclass AND c.contype = 'p'; + + IF pk_name IS NOT NULL AND pk_cols <> ARRAY['commitment_digest', 'transaction_hash'] THEN + EXECUTE format('ALTER TABLE commitments DROP CONSTRAINT %I', pk_name); + ALTER TABLE commitments ADD PRIMARY KEY (commitment_digest, transaction_hash); + END IF; +END +$$;` + var balancesTable = ` CREATE TABLE IF NOT EXISTS balances ( account TEXT PRIMARY KEY, @@ -115,6 +142,10 @@ func New(db *sql.DB) (*rpcstore, error) { } } + if _, err := db.Exec(commitmentsPKMigration); err != nil { + return nil, fmt.Errorf("failed to migrate commitments primary key: %w", err) + } + return &rpcstore{ db: db, }, nil @@ -377,7 +408,7 @@ func (s *rpcstore) StoreTransaction( insertCommitment := ` INSERT INTO commitments (commitment_digest, transaction_hash, provider_address, commitment_data) VALUES ($1, $2, $3, $4) - ON CONFLICT (commitment_digest) DO UPDATE SET commitment_data = EXCLUDED.commitment_data; + ON CONFLICT (commitment_digest, transaction_hash) DO UPDATE SET commitment_data = EXCLUDED.commitment_data; ` commitmentData, err := proto.Marshal(commitment) if err != nil { @@ -1034,6 +1065,104 @@ func (r *rpcstore) UpdateSettlementPayment( return nil } +// GetBidAmountsForTxns returns the stored bid amount for each of the given +// transaction hashes. The amount comes from the stored commitment data, +// which holds the per-transaction share for bundle bids. A transaction +// can have commitment rows under more than one digest, for example a +// bundle share and a later solo retry. The result comes from one digest +// only: among the digests whose rows cover every given hash, the one +// with the most recent commitment dispatch timestamp wins, and the +// digest text breaks a timestamp tie deterministically. The commitments +// table has no timestamp or serial column, so the dispatch timestamp +// inside the stored commitment data is the recency source. If no digest +// covers every given hash, the result is empty and the caller falls +// back to its own split rule. +func (r *rpcstore) GetBidAmountsForTxns( + ctx context.Context, + txnHashes []common.Hash, +) (map[common.Hash]*big.Int, error) { + wanted := make(map[common.Hash]struct{}, len(txnHashes)) + hashes := make([]string, 0, len(txnHashes)) + for _, h := range txnHashes { + if _, seen := wanted[h]; seen { + continue + } + wanted[h] = struct{}{} + hashes = append(hashes, h.Hex()) + } + + query := ` + SELECT commitment_digest, transaction_hash, commitment_data + FROM commitments + WHERE transaction_hash = ANY($1); + ` + + rows, err := r.db.QueryContext(ctx, query, pq.Array(hashes)) + if err != nil { + return nil, fmt.Errorf("failed to get bid amounts for transactions: %w", err) + } + defer func() { + _ = rows.Close() + }() + + type digestGroup struct { + amounts map[common.Hash]*big.Int + dispatch int64 + } + groups := make(map[string]*digestGroup) + for rows.Next() { + var ( + digest string + txnHash string + commitmentData []byte + ) + if err := rows.Scan(&digest, &txnHash, &commitmentData); err != nil { + return nil, fmt.Errorf("failed to scan commitment row: %w", err) + } + commitment := &bidderapiv1.Commitment{} + if err := proto.Unmarshal(commitmentData, commitment); err != nil { + return nil, fmt.Errorf("failed to unmarshal commitment data: %w", err) + } + amount, ok := new(big.Int).SetString(commitment.BidAmount, 10) + if !ok { + continue + } + group, found := groups[digest] + if !found { + group = &digestGroup{amounts: make(map[common.Hash]*big.Int)} + groups[digest] = group + } + group.amounts[common.HexToHash(txnHash)] = amount + if commitment.DispatchTimestamp > group.dispatch { + group.dispatch = commitment.DispatchTimestamp + } + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("error iterating commitment rows: %w", err) + } + + var ( + winner *digestGroup + winnerDigest string + ) + for digest, group := range groups { + if len(group.amounts) != len(wanted) { + continue + } + betterTime := winner == nil || group.dispatch > winner.dispatch + tieBreak := winner != nil && group.dispatch == winner.dispatch && digest > winnerDigest + if betterTime || tieBreak { + winner = group + winnerDigest = digest + } + } + if winner == nil { + return map[common.Hash]*big.Int{}, nil + } + + return winner.amounts, nil +} + func (r *rpcstore) StoreReceipt( ctx context.Context, receipt *types.Receipt, diff --git a/tools/preconf-rpc/store/store_test.go b/tools/preconf-rpc/store/store_test.go index 8f4464527..92ae9c7c8 100644 --- a/tools/preconf-rpc/store/store_test.go +++ b/tools/preconf-rpc/store/store_test.go @@ -534,4 +534,143 @@ func TestStore(t *testing.T) { t.Fatalf("receipt mismatch (-want +got):\n%s", diff) } }) + + t.Run("CommitmentsCompositePK", func(t *testing.T) { + // Recreate the legacy schema state with a single column primary key. + if _, err := db.Exec(`ALTER TABLE commitments DROP CONSTRAINT commitments_pkey`); err != nil { + t.Fatalf("failed to drop composite primary key: %v", err) + } + if _, err := db.Exec(`ALTER TABLE commitments ADD PRIMARY KEY (commitment_digest)`); err != nil { + t.Fatalf("failed to add legacy primary key: %v", err) + } + + // The init path must widen the primary key again. It must also be + // idempotent, so run it twice. + for i := 0; i < 2; i++ { + if _, err := store.New(db); err != nil { + t.Fatalf("failed to run store init on legacy schema: %v", err) + } + } + + txn3 := types.NewTransaction( + 0, + common.HexToAddress("0x1111111111111111111111111111111111111111"), + big.NewInt(3000000000), + 21000, + big.NewInt(3000000000), + nil, + ) + rawTxn3, err := txn3.MarshalBinary() + if err != nil { + t.Fatalf("failed to marshal transaction: %v", err) + } + wrappedTxn3 := &sender.Transaction{ + Transaction: txn3, + Raw: hex.EncodeToString(rawTxn3), + Sender: common.HexToAddress("0x9999999999999999999999999999999999999999"), + Type: sender.TxTypeRegular, + Status: sender.TxStatusPending, + } + if err := st.AddQueuedTransaction(context.Background(), wrappedTxn3); err != nil { + t.Fatalf("failed to add queued transaction: %v", err) + } + wrappedTxn3.Status = sender.TxStatusPreConfirmed + wrappedTxn3.BlockNumber = 1 + + // One bundle commitment covers many transactions with one digest. + // Each transaction stores its own share of the bid amount. + sharedDigest := commitments[0].CommitmentDigest + bundleCommitment := &bidderapiv1.Commitment{ + TxHashes: []string{txn1.Hash().Hex(), txn3.Hash().Hex()}, + BidAmount: big.NewInt(250000000).String(), + BlockNumber: 1, + CommitmentDigest: sharedDigest, + } + if err := st.StoreTransaction( + context.Background(), + wrappedTxn3, + []*bidderapiv1.Commitment{bundleCommitment}, + nil, + ); err != nil { + t.Fatalf("failed to store transaction with shared digest commitment: %v", err) + } + + var count int + if err := db.QueryRow( + `SELECT COUNT(*) FROM commitments WHERE commitment_digest = $1`, + sharedDigest, + ).Scan(&count); err != nil { + t.Fatalf("failed to count commitments: %v", err) + } + if count != 2 { + t.Fatalf("expected 2 commitment rows for shared digest, got %d", count) + } + + amounts, err := st.GetBidAmountsForTxns( + context.Background(), + []common.Hash{txn1.Hash(), txn3.Hash()}, + ) + if err != nil { + t.Fatalf("failed to get bid amounts: %v", err) + } + if len(amounts) != 2 { + t.Fatalf("expected 2 bid amounts, got %d", len(amounts)) + } + if amounts[txn3.Hash()].Cmp(big.NewInt(250000000)) != 0 { + t.Fatalf("expected bid amount 250000000 for txn3, got %s", amounts[txn3.Hash()]) + } + if amounts[txn1.Hash()].Cmp(big.NewInt(1000000000)) != 0 { + t.Fatalf("expected bid amount 1000000000 for txn1, got %s", amounts[txn1.Hash()]) + } + + // A solo retry stores a second commitment row for txn3 under + // its own digest with a different amount. The bundle lookup + // must keep returning the bundle digest's shares: the solo + // digest does not cover both hashes, so it cannot win. + soloRetryCommitment := &bidderapiv1.Commitment{ + TxHashes: []string{txn3.Hash().Hex()}, + BidAmount: big.NewInt(999999999).String(), + BlockNumber: 2, + CommitmentDigest: "0x5555555555555555555555555555555555555555555555555555555555555555", + DispatchTimestamp: time.Now().UnixMilli(), + } + if err := st.StoreTransaction( + context.Background(), + wrappedTxn3, + []*bidderapiv1.Commitment{soloRetryCommitment}, + nil, + ); err != nil { + t.Fatalf("failed to store solo retry commitment: %v", err) + } + + amounts, err = st.GetBidAmountsForTxns( + context.Background(), + []common.Hash{txn1.Hash(), txn3.Hash()}, + ) + if err != nil { + t.Fatalf("failed to get bid amounts: %v", err) + } + if len(amounts) != 2 { + t.Fatalf("expected 2 bid amounts, got %d", len(amounts)) + } + if amounts[txn3.Hash()].Cmp(big.NewInt(250000000)) != 0 { + t.Fatalf("expected bundle share 250000000 for txn3, got %s", amounts[txn3.Hash()]) + } + if amounts[txn1.Hash()].Cmp(big.NewInt(1000000000)) != 0 { + t.Fatalf("expected bid amount 1000000000 for txn1, got %s", amounts[txn1.Hash()]) + } + + // A lookup for txn3 alone must return the most recent digest + // that covers it: the solo retry. + soloAmounts, err := st.GetBidAmountsForTxns( + context.Background(), + []common.Hash{txn3.Hash()}, + ) + if err != nil { + t.Fatalf("failed to get solo bid amount: %v", err) + } + if soloAmounts[txn3.Hash()].Cmp(big.NewInt(999999999)) != 0 { + t.Fatalf("expected solo retry amount 999999999 for txn3, got %s", soloAmounts[txn3.Hash()]) + } + }) }