From ce93e647a0235779209ffb2aea708e28aebe8066 Mon Sep 17 00:00:00 2001 From: Josh Allmann Date: Thu, 5 Nov 2026 14:00:58 -0800 Subject: [PATCH] Reduce payment receipt contention Verify request-local tickets and encode challenges before acquiring the receipt mutex. Reuse prepared admission queries, combine watermark and exposure reads, and record accepted receipt counters after unlocking. Cover concurrent replay acceptance, durable winners, exact credit, and rollback counters with regression tests. --- orchestrator/payment_engine.go | 58 ++++++++++++++++---------- orchestrator/payment_recipient_test.go | 42 +++++++++++++++++++ orchestrator/redeemer_db.go | 44 +++++++++++++------ 3 files changed, 109 insertions(+), 35 deletions(-) diff --git a/orchestrator/payment_engine.go b/orchestrator/payment_engine.go index f9ccce2..a2ed9da 100644 --- a/orchestrator/payment_engine.go +++ b/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(¶ms, 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(¶ms, 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 } diff --git a/orchestrator/payment_recipient_test.go b/orchestrator/payment_recipient_test.go index 0f13154..f0b443f 100644 --- a/orchestrator/payment_recipient_test.go +++ b/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") diff --git a/orchestrator/redeemer_db.go b/orchestrator/redeemer_db.go index 96c231d..42b3d4d 100644 --- a/orchestrator/redeemer_db.go +++ b/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 }