diff --git a/tools/preconf-rpc/handlers/handlers.go b/tools/preconf-rpc/handlers/handlers.go index d3d8fe9cb..d199658c7 100644 --- a/tools/preconf-rpc/handlers/handlers.go +++ b/tools/preconf-rpc/handlers/handlers.go @@ -4,6 +4,7 @@ import ( "context" "encoding/hex" "encoding/json" + "errors" "fmt" "log/slog" "math/big" @@ -14,6 +15,7 @@ import ( bidderapiv1 "github.com/primev/mev-commit/p2p/gen/go/bidderapi/v1" "github.com/primev/mev-commit/tools/preconf-rpc/rpcserver" "github.com/primev/mev-commit/tools/preconf-rpc/sender" + "github.com/primev/mev-commit/tools/preconf-rpc/store" ) const ( @@ -756,6 +758,10 @@ func (h *rpcMethodHandler) handleGetTxCommitments( txHash := common.HexToHash(txHashStr) commitments, err := h.store.GetTransactionCommitments(ctx, txHash) + if errors.Is(err, store.ErrNotFound) { + h.logger.Info("No commitments found for transaction", "txHash", txHash) + return json.RawMessage("[]"), false, nil + } if err != nil { return nil, false, rpcserver.NewJSONErr( rpcserver.CodeCustomError, @@ -763,11 +769,6 @@ func (h *rpcMethodHandler) handleGetTxCommitments( ) } - if len(commitments) == 0 { - h.logger.Info("No commitments found for transaction", "txHash", txHash) - return json.RawMessage("[]"), false, nil - } - commitmentsJSON, err := json.Marshal(commitments) if err != nil { h.logger.Error("Failed to marshal commitments to JSON", "error", err, "txHash", txHash) diff --git a/tools/preconf-rpc/sender/sender.go b/tools/preconf-rpc/sender/sender.go index e67127ce6..1291d2250 100644 --- a/tools/preconf-rpc/sender/sender.go +++ b/tools/preconf-rpc/sender/sender.go @@ -162,6 +162,8 @@ type Store interface { AddBalance(ctx context.Context, account common.Address, amount *big.Int) error DeductBalance(ctx context.Context, account common.Address, amount *big.Int) error StoreTransaction(ctx context.Context, txn *Transaction, commitments []*bidderapiv1.Commitment, logs []*types.Log) error + StoreCommitments(ctx context.Context, txnHash common.Hash, commitments []*bidderapiv1.Commitment) error + AppendCommitment(ctx context.Context, txnHash common.Hash, commitment *bidderapiv1.Commitment) error GetTransactionByHash(ctx context.Context, txnHash common.Hash) (*Transaction, error) StoreReceipt(ctx context.Context, receipt *types.Receipt) error } @@ -1008,13 +1010,24 @@ BID_LOOP: } switch bidStatus.Type { case bidder.BidStatusCommitment: - if len(txn.commitments) > 0 { - if txn.commitments[0].BlockNumber != int64(bidBlockNo) { - txn.commitments = nil // clear previous commitments for new block - } + blockChanged := len(txn.commitments) > 0 && + txn.commitments[0].BlockNumber != int64(bidBlockNo) + if blockChanged { + txn.commitments = nil // clear previous commitments for new block } cmt := bidStatus.Arg.(*bidderapiv1.Commitment) txn.commitments = append(txn.commitments, cmt) + // Persist commitments as they arrive so partial commitments are queryable via + // mevcommit_getTransactionCommitments while the block is still pending. On a + // block change replace the previous block's set; within a block append the new + // commitment. + if blockChanged { + if err := t.store.StoreCommitments(ctx, txn.Hash(), txn.commitments); err != nil { + logger.Error("Failed to store commitments", "error", err) + } + } else if err := t.store.AppendCommitment(ctx, txn.Hash(), cmt); err != nil { + logger.Error("Failed to store commitment", "error", err) + } if t.fastTrack(txn.commitments, optedInSlot) && txn.Status != TxStatusPreConfirmed { txn.Status = TxStatusPreConfirmed txn.BlockNumber = int64(bidBlockNo) diff --git a/tools/preconf-rpc/sender/sender_test.go b/tools/preconf-rpc/sender/sender_test.go index ee79128e1..417584ba2 100644 --- a/tools/preconf-rpc/sender/sender_test.go +++ b/tools/preconf-rpc/sender/sender_test.go @@ -32,6 +32,7 @@ type mockStore struct { nonce map[common.Address]uint64 balances map[common.Address]*big.Int byHash map[common.Hash]*sender.Transaction + commitments map[common.Hash][]*bidderapiv1.Commitment preconfirmedTxns chan result } @@ -40,6 +41,7 @@ func newMockStore() *mockStore { queued: make(map[common.Address][]*sender.Transaction), nonce: make(map[common.Address]uint64), balances: make(map[common.Address]*big.Int), + commitments: make(map[common.Hash][]*bidderapiv1.Commitment), preconfirmedTxns: make(chan result, 10), byHash: make(map[common.Hash]*sender.Transaction), } @@ -147,6 +149,38 @@ func (m *mockStore) StoreTransaction( return nil } +func (m *mockStore) StoreCommitments( + _ context.Context, + txnHash common.Hash, + commitments []*bidderapiv1.Commitment, +) error { + m.mu.Lock() + defer m.mu.Unlock() + stored := make([]*bidderapiv1.Commitment, len(commitments)) + copy(stored, commitments) + m.commitments[txnHash] = stored + return nil +} + +func (m *mockStore) AppendCommitment( + _ context.Context, + txnHash common.Hash, + commitment *bidderapiv1.Commitment, +) error { + m.mu.Lock() + defer m.mu.Unlock() + m.commitments[txnHash] = append(m.commitments[txnHash], commitment) + return nil +} + +func (m *mockStore) storedCommitments(txnHash common.Hash) []*bidderapiv1.Commitment { + m.mu.Lock() + defer m.mu.Unlock() + stored := make([]*bidderapiv1.Commitment, len(m.commitments[txnHash])) + copy(stored, m.commitments[txnHash]) + return stored +} + func (m *mockStore) GetTransactionByHash( ctx context.Context, hash common.Hash, @@ -457,6 +491,11 @@ func TestSender(t *testing.T) { if len(res.commitments) != 2 { t.Fatalf("expected 2 commitments, got %d", len(res.commitments)) } + // The commitments should also have been persisted incrementally as they arrived, + // so they are queryable via mevcommit_getTransactionCommitments while pending. + if stored := st.storedCommitments(tx1.Hash()); len(stored) != 2 { + t.Fatalf("expected 2 commitments persisted during bidding, got %d", len(stored)) + } tx2 := &sender.Transaction{ Transaction: types.NewTransaction( diff --git a/tools/preconf-rpc/store/store.go b/tools/preconf-rpc/store/store.go index 86b0b050b..41b8b0b72 100644 --- a/tools/preconf-rpc/store/store.go +++ b/tools/preconf-rpc/store/store.go @@ -46,6 +46,11 @@ CREATE TABLE IF NOT EXISTS commitments ( FOREIGN KEY (transaction_hash) REFERENCES mcTransactions (hash) ON DELETE CASCADE );` +// commitments are always looked up, replaced and cascade-deleted by transaction_hash, +// which Postgres does not index automatically for a foreign key. +var commitmentsIndex = ` +CREATE INDEX IF NOT EXISTS idx_commitments_transaction_hash ON commitments (transaction_hash);` + var balancesTable = ` CREATE TABLE IF NOT EXISTS balances ( account TEXT PRIMARY KEY, @@ -102,6 +107,7 @@ func New(db *sql.DB) (*rpcstore, error) { for _, table := range []string{ transactionsTable, commitmentsTable, + commitmentsIndex, balancesTable, subsidiesTable, simulationLogs, @@ -372,31 +378,18 @@ func (s *rpcstore) StoreTransaction( return fmt.Errorf("transaction %s not found for update", txn.Hash().Hex()) } + // Reconcile the stored commitments with the finalized set: drop any partial commitments + // persisted while bidding, then write the final set. A failed transaction writes nothing, + // so it no longer retains commitments from an abandoned block attempt. + deleteCommitments := `DELETE FROM commitments WHERE transaction_hash = $1;` + if _, err := dbTxn.ExecContext(ctx, deleteCommitments, txn.Hash().Hex()); err != nil { + _ = dbTxn.Rollback() + return fmt.Errorf("failed to delete existing commitments for transaction %s: %w", txn.Hash().Hex(), err) + } if txn.Status != sender.TxStatusFailed { - for _, commitment := range commitments { - 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; - ` - commitmentData, err := proto.Marshal(commitment) - if err != nil { - _ = dbTxn.Rollback() - return fmt.Errorf("failed to marshal commitment: %w", err) - } - - _, err = dbTxn.ExecContext( - ctx, - insertCommitment, - commitment.CommitmentDigest, - txn.Hash().Hex(), - commitment.ProviderAddress, - commitmentData, - ) - if err != nil { - _ = dbTxn.Rollback() - return fmt.Errorf("failed to insert commitment for transaction %s: %w", txn.Hash().Hex(), err) - } + if err := insertCommitments(ctx, dbTxn, txn.Hash().Hex(), commitments); err != nil { + _ = dbTxn.Rollback() + return err } } @@ -425,6 +418,98 @@ func (s *rpcstore) StoreTransaction( return nil } +// execer is satisfied by both *sql.DB and *sql.Tx so commitment inserts can run either +// standalone or inside a wider database transaction. +type execer interface { + ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) +} + +// insertCommitments upserts each commitment for the given transaction hash. A conflicting +// commitment digest updates the stored data so re-persisting the same commitment is +// idempotent. +func insertCommitments( + ctx context.Context, + db execer, + txnHash string, + commitments []*bidderapiv1.Commitment, +) error { + 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; + ` + for _, commitment := range commitments { + commitmentData, err := proto.Marshal(commitment) + if err != nil { + return fmt.Errorf("failed to marshal commitment: %w", err) + } + + if _, err := db.ExecContext( + ctx, + insertCommitment, + commitment.CommitmentDigest, + txnHash, + commitment.ProviderAddress, + commitmentData, + ); err != nil { + return fmt.Errorf("failed to insert commitment for transaction %s: %w", txnHash, err) + } + } + return nil +} + +// AppendCommitment records a single commitment for a transaction, adding to any already +// stored for the current block attempt. It is called as each commitment arrives while a +// block is still pending. The transaction row must already exist (added via +// AddQueuedTransaction) to satisfy the foreign key on the commitments table. +func (s *rpcstore) AppendCommitment( + ctx context.Context, + txnHash common.Hash, + commitment *bidderapiv1.Commitment, +) error { + return insertCommitments(ctx, s.db, txnHash.Hex(), []*bidderapiv1.Commitment{commitment}) +} + +// StoreCommitments atomically replaces the stored commitments for a transaction with the +// given set. It is used on a block change during bidding to drop the previous block's +// commitments and record the new block's set, so that mevcommit_getTransactionCommitments +// reflects the current block attempt; within a block, commitments are added incrementally +// via AppendCommitment. The transaction row must already exist (added via +// AddQueuedTransaction) to satisfy the foreign key on the commitments table. An empty set +// is a no-op; StoreTransaction reconciles the final set (including clearing commitments for +// a failed transaction) once the transaction settles. +func (s *rpcstore) StoreCommitments( + ctx context.Context, + txnHash common.Hash, + commitments []*bidderapiv1.Commitment, +) error { + if len(commitments) == 0 { + return nil + } + + dbTxn, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("failed to begin transaction: %w", err) + } + + deleteCommitments := `DELETE FROM commitments WHERE transaction_hash = $1;` + if _, err := dbTxn.ExecContext(ctx, deleteCommitments, txnHash.Hex()); err != nil { + _ = dbTxn.Rollback() + return fmt.Errorf("failed to delete existing commitments for transaction %s: %w", txnHash.Hex(), err) + } + + if err := insertCommitments(ctx, dbTxn, txnHash.Hex(), commitments); err != nil { + _ = dbTxn.Rollback() + return err + } + + if err := dbTxn.Commit(); err != nil { + return fmt.Errorf("failed to commit transaction: %w", err) + } + + return nil +} + func (s *rpcstore) GetTransactionLogs(ctx context.Context, txnHash common.Hash) ([]*types.Log, error) { query := ` SELECT logs diff --git a/tools/preconf-rpc/store/store_test.go b/tools/preconf-rpc/store/store_test.go index 8f4464527..18915efb7 100644 --- a/tools/preconf-rpc/store/store_test.go +++ b/tools/preconf-rpc/store/store_test.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "encoding/hex" + "errors" "fmt" "math/big" "testing" @@ -534,4 +535,119 @@ func TestStore(t *testing.T) { t.Fatalf("receipt mismatch (-want +got):\n%s", diff) } }) + + t.Run("StoreCommitments", func(t *testing.T) { + txn3 := types.NewTransaction( + 0, + common.HexToAddress("0x1111111111111111111111111111111111111111"), + big.NewInt(1000000000), + 21000, + big.NewInt(1000000000), + 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("0x2222222222222222222222222222222222222222"), + Type: sender.TxTypeRegular, + Status: sender.TxStatusPending, + } + // The transaction row must exist to satisfy the commitments foreign key. + if err := st.AddQueuedTransaction(context.Background(), wrappedTxn3); err != nil { + t.Fatalf("failed to queue transaction: %v", err) + } + + commitment := func(provider, digest string, block int64) *bidderapiv1.Commitment { + return &bidderapiv1.Commitment{ + TxHashes: []string{txn3.Hash().Hex()}, + BidAmount: big.NewInt(1000000000).String(), + BlockNumber: block, + ProviderAddress: provider, + CommitmentDigest: digest, + } + } + countCommitments := func(stage string) int { + got, err := st.GetTransactionCommitments(context.Background(), txn3.Hash()) + if err != nil { + t.Fatalf("failed to get commitments %s: %v", stage, err) + } + return len(got) + } + + // AppendCommitment adds a single commitment for the current block attempt. + if err := st.AppendCommitment(context.Background(), txn3.Hash(), commitment("0xaaaa", "0xcommit-a", 1)); err != nil { + t.Fatalf("failed to append first commitment: %v", err) + } + if err := st.AppendCommitment(context.Background(), txn3.Hash(), commitment("0xbbbb", "0xcommit-b", 1)); err != nil { + t.Fatalf("failed to append second commitment: %v", err) + } + if n := countCommitments("after appends"); n != 2 { + t.Fatalf("expected 2 commitments after appends, got %d", n) + } + // Re-appending the same digest is idempotent (upsert, not duplicate). + if err := st.AppendCommitment(context.Background(), txn3.Hash(), commitment("0xaaaa", "0xcommit-a", 1)); err != nil { + t.Fatalf("failed to re-append commitment: %v", err) + } + if n := countCommitments("after idempotent append"); n != 2 { + t.Fatalf("expected 2 commitments after idempotent append, got %d", n) + } + + // StoreCommitments replaces the previous block's set rather than appending. + replaced := []*bidderapiv1.Commitment{ + commitment("0xcccc", "0xcommit-c", 2), + commitment("0xdddd", "0xcommit-d", 2), + } + if err := st.StoreCommitments(context.Background(), txn3.Hash(), replaced); err != nil { + t.Fatalf("failed to store replaced commitments: %v", err) + } + got, err := st.GetTransactionCommitments(context.Background(), txn3.Hash()) + if err != nil { + t.Fatalf("failed to get commitments after replace: %v", err) + } + if len(got) != 2 { + t.Fatalf("expected 2 commitments after replace, got %d", len(got)) + } + for _, c := range got { + if c.CommitmentDigest == "0xcommit-a" || c.CommitmentDigest == "0xcommit-b" { + t.Fatalf("stale commitment %s should have been replaced", c.CommitmentDigest) + } + } + + // An empty set is a no-op and must not clear the last known commitments. + if err := st.StoreCommitments(context.Background(), txn3.Hash(), nil); err != nil { + t.Fatalf("failed to store empty commitments: %v", err) + } + if n := countCommitments("after empty store"); n != 2 { + t.Fatalf("expected 2 commitments after empty store, got %d", n) + } + + // StoreTransaction reconciles to the finalized set, dropping the partial + // commitments persisted while bidding (no union with an earlier block attempt). + wrappedTxn3.Status = sender.TxStatusPreConfirmed + wrappedTxn3.BlockNumber = 2 + if err := st.StoreTransaction(context.Background(), wrappedTxn3, []*bidderapiv1.Commitment{commitment("0xeeee", "0xcommit-e", 2)}, nil); err != nil { + t.Fatalf("failed to store preconfirmed transaction: %v", err) + } + got, err = st.GetTransactionCommitments(context.Background(), txn3.Hash()) + if err != nil { + t.Fatalf("failed to get commitments after finalize: %v", err) + } + if len(got) != 1 || got[0].CommitmentDigest != "0xcommit-e" { + t.Fatalf("expected only the finalized commitment, got %d: %+v", len(got), got) + } + + // A failed transaction clears its commitments so none linger from bidding. + wrappedTxn3.Status = sender.TxStatusFailed + wrappedTxn3.Details = "failed" + if err := st.StoreTransaction(context.Background(), wrappedTxn3, nil, nil); err != nil { + t.Fatalf("failed to store failed transaction: %v", err) + } + if _, err := st.GetTransactionCommitments(context.Background(), txn3.Hash()); !errors.Is(err, store.ErrNotFound) { + t.Fatalf("expected ErrNotFound after failed transaction cleared commitments, got %v", err) + } + }) }