Reduce payment receipt contention

orchestrator/payment_engine.go

@@ -121,6 +121,8 @@ func (e *PaymentEngine) makeChallenge(ctx context.Context, appID, manifest strin
 	info := wire.OrchestratorInfo{Transcoder: service, Address: e.recipient.Bytes(), Price: priceInfo,
 		TicketParams: wire.TicketParams{Recipient: e.recipient.Bytes(), FaceValue: e.faceValue.Bytes(), WinProb: e.winProb.Bytes(), RecipientRandHash: randHash.Bytes(), Seed: seed, ExpirationBlock: expiresAtBlock.Bytes(), Expiration: wire.ExpirationParams{CreationRound: expiration.CreationRound, CreationRoundBlockHash: expiration.CreationRoundBlockHash.Bytes()}},
 		Auth:         e.authToken(manifest, time.Now().Add(time.Hour).Unix())}
+	encoded := wire.EncodeOrchestratorInfo(info)
+	challenge := paymentChallenge(appID, manifest, service, encoded)
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	if err := snapshot.Validate(); err != nil {
@@ -137,8 +139,8 @@ func (e *PaymentEngine) makeChallenge(ctx context.Context, appID, manifest strin
 	} else if current.appID != appID || current.registration != registration || current.payer != payer || current.price != priceInfo || current.unit != unit {
 		return Challenge{}, errors.New("payment session scope conflict")
 	}
-	current.info, current.updated = wire.EncodeOrchestratorInfo(info), time.Now()
-	return paymentChallenge(appID, manifest, service, current.info), nil
+	current.info, current.updated = encoded, time.Now()
+	return challenge, nil
 }
 
 func paymentChallenge(appID, manifest, service string, info []byte) Challenge {
@@ -276,6 +278,34 @@ func (e *PaymentEngine) Receive(ctx context.Context, appID, manifest, paymentHea
 		}
 		return ethcommon.Address{}, nil, pm.ErrPayerUnavailable
 	}
+	batchNonces := make(map[uint32]bool, len(payment.PayerParams))
+	batchCredit := new(big.Rat)
+	var winners []*pm.SignedTicket
+	validator := pm.NewValidator(pm.DefaultSigVerifier{})
+	for _, sp := range payment.PayerParams {
+		if sp.TicketNonce == 0 || sp.TicketNonce >= 600 || len(sp.Sig) != 65 || batchNonces[sp.TicketNonce] {
+			return ethcommon.Address{}, nil, ErrInvalidPayment
+		}
+		ticket := pm.NewTicket(&params, params.ExpirationParams, payer, sp.TicketNonce)
+		if err := validator.ValidateTicket(e.recipient, ticket, sp.Sig, recipientRand); err != nil {
+			return ethcommon.Address{}, nil, ErrInvalidPayment
+		}
+		batchNonces[sp.TicketNonce] = true
+		batchCredit.Add(batchCredit, ticket.EV())
+		if validator.IsWinningTicket(ticket, sp.Sig, recipientRand) {
+			winners = append(winners, &pm.SignedTicket{Ticket: ticket, Sig: sp.Sig, RecipientRand: recipientRand})
+		}
+	}
+
+	// Record accepted credit after unlocking; credit is returned only on success.
+	defer func() {
+		if credit != nil {
+			e.metrics.Add("payment_credit_eth_total", observability.ETH(batchCredit), labels)
+			e.metrics.Add("payment_tickets_total", float64(len(batchNonces)), labels.With("kind", "accepted"))
+			e.metrics.Add("payment_tickets_total", float64(len(winners)), labels.With("kind", "winning"))
+		}
+	}()
+
 	e.mu.Lock()
 	defer e.mu.Unlock()
 	// Store winning tickets before publishing in-memory replay guards and credit.
@@ -295,27 +325,12 @@ func (e *PaymentEngine) Receive(ctx context.Context, appID, manifest, paymentHea
 	}
 	randKey := recipientRand.String()
 	previous := e.ticketNonces[randKey]
-	batchNonces := make(map[uint32]bool, len(payment.PayerParams))
-	var winners []*pm.SignedTicket
-	balance := new(big.Rat).Set(session.balance)
-	validator := pm.NewValidator(pm.DefaultSigVerifier{})
-	for _, sp := range payment.PayerParams {
-		if sp.TicketNonce == 0 || sp.TicketNonce >= 600 || len(sp.Sig) != 65 {
-			return ethcommon.Address{}, nil, ErrInvalidPayment
-		}
-		ticket := pm.NewTicket(&params, params.ExpirationParams, payer, sp.TicketNonce)
-		if err := validator.ValidateTicket(e.recipient, ticket, sp.Sig, recipientRand); err != nil {
-			return ethcommon.Address{}, nil, ErrInvalidPayment
-		}
-		if batchNonces[sp.TicketNonce] || (previous != nil && previous.nonceSeen[sp.TicketNonce]) {
+	for nonce := range batchNonces {
+		if previous != nil && previous.nonceSeen[nonce] {
 			return ethcommon.Address{}, nil, ErrInvalidPayment
 		}
-		batchNonces[sp.TicketNonce] = true
-		balance.Add(balance, ticket.EV())
-		if validator.IsWinningTicket(ticket, sp.Sig, recipientRand) {
-			winners = append(winners, &pm.SignedTicket{Ticket: ticket, Sig: sp.Sig, RecipientRand: recipientRand})
-		}
 	}
+	balance := new(big.Rat).Add(session.balance, batchCredit)
 	if err := e.store.storeWinningTickets(ctx, payer, params.RecipientRandHash, snapshot.Round.Int64()-2, new(big.Int).Add(funds.Deposit, funds.Reserve), winners, snapshot); err != nil {
 		return ethcommon.Address{}, nil, err
 	}
@@ -327,9 +342,6 @@ func (e *PaymentEngine) Receive(ctx context.Context, appID, manifest, paymentHea
 		previous.nonceSeen[nonce] = true
 	}
 	e.nonceCount += len(batchNonces)
-	e.metrics.Add("payment_credit_eth_total", observability.ETH(new(big.Rat).Sub(balance, session.balance)), labels)
-	e.metrics.Add("payment_tickets_total", float64(len(batchNonces)), labels.With("kind", "accepted"))
-	e.metrics.Add("payment_tickets_total", float64(len(winners)), labels.With("kind", "winning"))
 	session.balance, session.updated = balance, time.Now()
 	return payer, new(big.Rat).Set(balance), nil
 }

orchestrator/payment_recipient_test.go

@@ -303,6 +303,46 @@ func TestRecipientRestartRotatesSecretsAndPreservesWinningTickets(t *testing.T)
 	require.NoError(t, receiveRecipientPayment(t, restarted, key, fresh, 1))
 }
 
+func TestRecipientConcurrentReplay(t *testing.T) {
+	e, _, key, _ := newRecipientTestEngine(t)
+	e.metrics = observability.New("orchestrator")
+	info := issueRecipientParams(t, e, key)
+	payment, segment := recipientPayment(t, key, info, 1, 2)
+	paymentHeader := base64.StdEncoding.EncodeToString(wire.EncodePayment(payment))
+	segmentHeader := base64.StdEncoding.EncodeToString(wire.EncodeSegData(segment))
+	start := make(chan struct{})
+	results := make(chan error, 32)
+	for range cap(results) {
+		go func() {
+			<-start
+			_, _, err := e.Receive(t.Context(), "runner", "session", paymentHeader, segmentHeader)
+			results <- err
+		}()
+	}
+	close(start)
+	accepted := 0
+	for range cap(results) {
+		if err := <-results; err == nil {
+			accepted++
+		} else {
+			require.ErrorIs(t, err, ErrInvalidPayment)
+		}
+	}
+	require.Equal(t, 1, accepted)
+	require.EqualValues(t, 2, test.MetricSum(t, e.metrics.Registry, "payment_tickets_total", map[string]string{"kind": "accepted"}))
+	require.EqualValues(t, 2, test.MetricSum(t, e.metrics.Registry, "payment_tickets_total", map[string]string{"kind": "winning"}))
+	require.Positive(t, test.MetricSum(t, e.metrics.Registry, "payment_credit_eth_total", nil))
+	require.Equal(t, 2, e.nonceCount)
+	count, err := e.store.WinningTicketCount(crypto.PubkeyToAddress(key.PublicKey), 0)
+	require.NoError(t, err)
+	require.Equal(t, 2, count)
+	balance, err := e.Balance("session")
+	require.NoError(t, err)
+	expected := new(big.Rat).Mul(new(big.Rat).SetFrac(e.faceValue, maxWinProb), new(big.Rat).SetInt(e.winProb))
+	expected.Mul(expected, big.NewRat(2, 1))
+	require.Zero(t, balance.Cmp(expected), "concurrent replays must credit the batch once")
+}
+
 func TestInactiveOrchestratorDoesNotIssuePaidChallenge(t *testing.T) {
 	e, chain, key, _ := newRecipientTestEngine(t)
 	chain.active = false
@@ -345,6 +385,8 @@ func TestRecipientStorageFailureDoesNotPublishCreditOrConsumeNonces(t *testing.T
 	require.NoError(t, err)
 	require.Zero(t, balance.Sign())
 	require.Zero(t, e.nonceCount)
+	require.Zero(t, test.MetricSum(t, e.metrics.Registry, "payment_tickets_total", nil))
+	require.Zero(t, test.MetricSum(t, e.metrics.Registry, "payment_credit_eth_total", nil))
 	_, err = e.store.db.Exec(`DROP TRIGGER reject_second_ticket`)
 	require.NoError(t, err)
 	require.NoError(t, receiveRecipientPayment(t, e, key, info, 1, 2), "storage failure must leave both nonces retryable")

orchestrator/redeemer_db.go

@@ -36,6 +36,8 @@ type RedeemerDB struct {
 	metrics    *observability.Metrics
 	readiness  *paidReadiness
 	db         *sql.DB
+	admission  *sql.Stmt
+	liability  *sql.Stmt
 	recoveryMu sync.Mutex
 	recovery   *redemptionRecovery
 }
@@ -54,7 +56,21 @@ func OpenRedeemerDB(path string) (*RedeemerDB, error) {
 		_ = db.Close()
 		return nil, fmt.Errorf("initialize redeemer SQLite: %w", err)
 	}
-	return &RedeemerDB{db: db}, nil
+	store := &RedeemerDB{db: db}
+	for _, check := range []struct {
+		stmt  **sql.Stmt
+		query string
+	}{
+		{&store.admission, receiptAdmissionQuery},
+		{&store.liability, receiptLiabilityQuery},
+	} {
+		*check.stmt, err = db.PrepareContext(ctx, check.query)
+		if err != nil {
+			_ = db.Close()
+			return nil, fmt.Errorf("prepare payment admission checks: %w", err)
+		}
+	}
+	return store, nil
 }
 
 func openRedeemerDB(ctx context.Context, path string, create bool) (*sql.DB, error) {
@@ -126,6 +142,14 @@ func (s *RedeemerDB) StoreWinningTicket(t *pm.SignedTicket) error {
 	return err
 }
 
+const (
+	receiptAdmissionQuery = `SELECT m.number,m.hash,EXISTS(SELECT 1 FROM winning_tickets w JOIN redemption_attempts a ON a.sig=w.sig WHERE w.recipient_rand_hash=? AND (a.phase!='expired' OR EXISTS(SELECT 1 FROM redemption_transactions t WHERE t.sig=w.sig)))
+ FROM (SELECT 1) LEFT JOIN collateral_watermark m ON m.id=1`
+	receiptLiabilityQuery = `SELECT face_value FROM winning_tickets WHERE payer_address=? AND redeemed_at IS NULL AND (
+ (creation_round>=? AND sig NOT IN (SELECT sig FROM redemption_attempts WHERE phase IN ('used','invalid','expired','reverted')))
+ OR EXISTS(SELECT 1 FROM redemption_transactions t WHERE t.sig=winning_tickets.sig AND t.phase NOT IN ('confirmed','reverted','superseded')))`
+)
+
 // storeWinningTickets commits a validated receipt batch before the payment
 // engine publishes replay guards or credit. Check exposure and liability in the
 // same transaction so redemption and concurrent receipts cannot race these gates.
@@ -137,19 +161,17 @@ func (s *RedeemerDB) storeWinningTickets(ctx context.Context, payer ethcommon.Ad
 	defer tx.Rollback()
 	// A singleton durable floor prevents old collateral being reused after any
 	// settlement releases liability. It is deliberately conservative across payers.
-	var number int64
-	var hash string
-	if err := tx.QueryRowContext(ctx, `SELECT number,hash FROM collateral_watermark WHERE id=1`).Scan(&number, &hash); err != nil && !errors.Is(err, sql.ErrNoRows) {
+	var number sql.NullInt64
+	var hash sql.NullString
+	var exposed int
+	if err := tx.StmtContext(ctx, s.admission).QueryRowContext(ctx, randHash.Hex()).Scan(&number, &hash, &exposed); err != nil {
 		return err
-	} else if err == nil && (snapshot.Number == nil || snapshot.Number.Cmp(big.NewInt(number)) < 0 || (snapshot.Number.Cmp(big.NewInt(number)) == 0 && snapshot.BlockHash.Hex() != hash)) {
+	}
+	if number.Valid && (snapshot.Number == nil || snapshot.Number.Cmp(big.NewInt(number.Int64)) < 0 || (snapshot.Number.Cmp(big.NewInt(number.Int64)) == 0 && snapshot.BlockHash.Hex() != hash.String)) {
 		return pm.ErrPayerUnavailable
 	}
 	// Once redemption can reveal randomness, this epoch must never accept
 	// more tickets, even if a reorg moves the observed L1 clock backwards.
-	var exposed int
-	if err := tx.QueryRow(`SELECT EXISTS(SELECT 1 FROM winning_tickets w JOIN redemption_attempts a ON a.sig=w.sig WHERE w.recipient_rand_hash=? AND (a.phase!='expired' OR EXISTS(SELECT 1 FROM redemption_transactions t WHERE t.sig=w.sig)))`, randHash.Hex()).Scan(&exposed); err != nil {
-		return err
-	}
 	if exposed != 0 {
 		return ErrInvalidPayment
 	}
@@ -162,9 +184,7 @@ func (s *RedeemerDB) storeWinningTickets(ctx context.Context, payer ethcommon.Ad
 	}
 	// Include every unconfirmed winner, across all sessions for this payer.
 	// Keep uncertain transactions reserved until finalized receipt reconciliation.
-	rows, err := tx.Query(`SELECT face_value FROM winning_tickets WHERE payer_address=? AND redeemed_at IS NULL AND (
- (creation_round>=? AND sig NOT IN (SELECT sig FROM redemption_attempts WHERE phase IN ('used','invalid','expired','reverted')))
- OR EXISTS(SELECT 1 FROM redemption_transactions t WHERE t.sig=winning_tickets.sig AND t.phase NOT IN ('confirmed','reverted','superseded')))`, payer.Hex(), minRound)
+	rows, err := tx.StmtContext(ctx, s.liability).QueryContext(ctx, payer.Hex(), minRound)
 	if err != nil {
 		return err
 	}