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 }