feat: return partial commitments for pending blocks

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,18 +758,17 @@ 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,
 			"failed to get transaction commitments",
 		)
 	}
 
-	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)

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)

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(

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