From 1353a31e941b18585177c5b20039f943ac4cbfd7 Mon Sep 17 00:00:00 2001 From: Rick Staa Date: Wed, 5 Aug 2026 14:15:44 +0200 Subject: [PATCH] pm: Defer ticket redemption when the sender cannot cover the face value livepeer/protocol#657 changed TicketBroker.redeemWinningTicket to require the sender's deposit and reserve to cover the full ticket face value. A redemption below it now reverts and leaves the ticket unused, where it previously consumed the ticket and paid out whatever remained (potentially dust). go-livepeer had no matching precondition, so it would keep building redemption transactions that cannot succeed. Mirror the broker's requirement before any RPC round-trip: availableFunds is served from the sender watcher cache, which the block watcher keeps current from DepositFunded and ReserveFunded, so a top-up is picked up on the next L1 block without a doomed transaction in between. The queue re-evaluates on every block, so a deferred ticket is redeemed as soon as the sender funds it, for as long as the ticket remains valid. Treat the underfunded case as an expected state rather than a failure: it is retryable and logged at V(5), since at one line per ticket per block it would otherwise flood the log for as long as a sender stays underfunded. Stop re-attempting the head of the queue within a single block event. Every iteration re-selects the earliest unredeemed ticket, so any outcome that leaves that ticket in place would re-attempt it once per queued ticket. The loop now stops and waits for the next block. Finally, do not infer that a ticket was consumed from an opaque error. CheckTx reports only that a transaction failed, without a revert reason, and isNonRetryableTicketErr matched that string to mark the ticket redeemed locally. Now that a reverted redemption leaves the ticket redeemable, ask the broker whether it was actually used before dropping it, so a still-valid ticket is not forfeited by a redemption that raced the sender's balance. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG_PENDING.md | 4 ++ pm/queue.go | 29 ++++++-- pm/queue_test.go | 50 +++++++++++++ pm/sendermonitor.go | 17 +++++ pm/sendermonitor_test.go | 152 +++++++++++++++++++++++++++++++++++++++ pm/stub.go | 8 ++- pm/validator.go | 17 +++++ 7 files changed, 272 insertions(+), 5 deletions(-) diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md index f9afe08498..85f23a8544 100644 --- a/CHANGELOG_PENDING.md +++ b/CHANGELOG_PENDING.md @@ -18,6 +18,10 @@ #### General +#### Orchestrator + +- [#4012](https://github.com/livepeer/go-livepeer/pull/4012) Defer ticket redemption while a sender's deposit and reserve cannot cover the full ticket face value, instead of submitting a redemption that reverts. Tickets stay queued and are redeemed on the first block after the sender tops up, for as long as they remain valid (@rickstaa) + #### Broadcaster #### CLI diff --git a/pm/queue.go b/pm/queue.go index 43fbdd3694..5f436c1618 100644 --- a/pm/queue.go +++ b/pm/queue.go @@ -1,6 +1,7 @@ package pm import ( + "errors" "math/big" "strings" "sync" @@ -121,6 +122,10 @@ func (q *ticketQueue) handleBlockEvent(latestL1Block *big.Int) { glog.Errorf("Error getting queue length err=%q", err) return } + // Each iteration re-selects the *earliest* unredeemed ticket, so the loop only makes + // progress when a ticket is marked redeemed. Any outcome that leaves the head of the + // queue in place would otherwise re-select the same ticket on every remaining + // iteration, so we stop and wait for the next block instead of spinning. for i := 0; i < int(numTickets); i++ { nextTicket, err := q.store.SelectEarliestWinningTicket(q.sender, new(big.Int).Sub(q.tm.LastInitializedRound(), big.NewInt(ticketValidityPeriod)).Int64()) if err != nil { @@ -132,7 +137,7 @@ func (q *ticketQueue) handleBlockEvent(latestL1Block *big.Int) { } if !q.isRecipientActive(nextTicket.Recipient) { glog.V(5).Infof("Ticket recipient is not active in this round, cannot redeem ticket recipient=%v", nextTicket.Recipient.Hex()) - continue + return } if nextTicket.ParamsExpirationBlock.Cmp(latestL1Block) <= 0 { resCh := make(chan struct { @@ -146,15 +151,22 @@ func (q *ticketQueue) handleBlockEvent(latestL1Block *big.Int) { // after receiving the response we can close the channel so it can be GC'd close(resCh) if res.err != nil { - glog.Errorf("Error redeeming err=%q", res.err) + if errors.Is(res.err, errInsufficientSenderFunds) { + // Expected while a sender is underfunded, not a failure: the + // ticket stays queued and is redeemed on the first block after + // the sender tops up, for as long as it remains valid. + glog.V(5).Infof("Deferring redemption until sender has sufficient funds sender=%v faceValue=%v", q.sender.Hex(), nextTicket.FaceValue) + } else { + glog.Errorf("Error redeeming err=%q", res.err) + } // If the error is non-retryable then we mark the ticket as redeemed if !isNonRetryableTicketErr(res.err) { - continue + return } } if err := q.store.MarkWinningTicketRedeemed(nextTicket, res.txHash); err != nil { glog.Error(err) - continue + return } case <-q.quit: return @@ -164,6 +176,15 @@ func (q *ticketQueue) handleBlockEvent(latestL1Block *big.Int) { } func isNonRetryableTicketErr(err error) bool { + // A redemption that reverted without consuming the ticket stays redeemable, even + // though CheckTx reports it as a plain transaction failure below. + var unconsumed unconsumedRedemptionErr + if errors.As(err, &unconsumed) { + return false + } + if errors.Is(err, errInsufficientSenderFunds) { + return false + } return err == errIsUsedTicket || // Depends on logic in eth.client.CheckTx() strings.Contains(err.Error(), "transaction failed") || diff --git a/pm/queue_test.go b/pm/queue_test.go index fdcdea392f..4d689ae627 100644 --- a/pm/queue_test.go +++ b/pm/queue_test.go @@ -4,6 +4,7 @@ import ( "fmt" "math/big" "sync" + "sync/atomic" "testing" "time" @@ -365,3 +366,52 @@ func TestIsRecipientActive(t *testing.T) { ts.err = errors.New("some error") assert.True(q.isRecipientActive(addr)) } + +// A retryable failure leaves the ticket at the head of the queue, and every iteration +// re-selects the earliest ticket. Without stopping, one block event would re-attempt the +// same ticket once per queued ticket. +func TestTicketQueue_RetryableErrorDoesNotSpinOnHeadOfQueue(t *testing.T) { + assert := assert.New(t) + + sender := RandAddress() + ts := newStubTicketStore() + tm := &stubTimeManager{round: big.NewInt(100)} + sm := &LocalSenderMonitor{ + ticketStore: ts, + tm: tm, + } + + q := newTicketQueue(sender, sm) + + numTickets := 10 + for i := 0; i < numTickets; i++ { + assert.Nil(q.Add(defaultSignedTicket(sender, uint32(i)))) + } + + var attempts int32 + stop := make(chan struct{}) + go func() { + for { + select { + case red := <-q.Redeemable(): + atomic.AddInt32(&attempts, 1) + red.resCh <- struct { + txHash ethcommon.Hash + err error + }{ethcommon.Hash{}, errInsufficientSenderFunds} + case <-stop: + return + } + } + }() + defer close(stop) + + q.handleBlockEvent(big.NewInt(1)) + + assert.Equal(int32(1), atomic.LoadInt32(&attempts), "expected a single attempt per block while the head of the queue is blocked") + + // Nothing may be dropped: the tickets stay queued for a later block. + qlen, err := q.Length() + assert.Nil(err) + assert.Equal(numTickets, qlen) +} diff --git a/pm/sendermonitor.go b/pm/sendermonitor.go index 5ef5dcfedd..c995ae74b9 100644 --- a/pm/sendermonitor.go +++ b/pm/sendermonitor.go @@ -359,6 +359,16 @@ func (sm *LocalSenderMonitor) redeemWinningTicket(ticket *SignedTicket) (*types. return nil, err } + // The broker requires the sender's deposit and reserve to cover the full face value, + // otherwise the redemption reverts without consuming the ticket. Check that here, + // before spending any RPC round-trips: for an underfunded sender this is the steady + // state, and the ticket queue re-evaluates it on every L1 block. availableFunds is + // served from the sender watcher's cache, which the block watcher keeps current, so + // a top-up is picked up on the next block without a doomed transaction in between. + if availableFunds.Cmp(ticket.FaceValue) < 0 { + return nil, errInsufficientSenderFunds + } + // Fail early if ticket is used used, err := sm.broker.IsUsedTicket(ticket.Ticket) if err != nil { @@ -427,6 +437,13 @@ func (sm *LocalSenderMonitor) redeemWinningTicket(ticket *SignedTicket) (*types. if monitor.Enabled { monitor.TicketRedemptionError(ticket.Sender.Hex()) } + // CheckTx only reports that the transaction failed, without a revert reason, so + // ask the broker whether the ticket was actually consumed rather than assuming + // it was. Since livepeer/protocol#657 a redemption that reverts leaves the + // ticket unused and still redeemable, and dropping it here would forfeit it. + if used, usedErr := sm.broker.IsUsedTicket(ticket.Ticket); usedErr == nil && !used { + return tx, unconsumedRedemptionErr{err} + } // Return tx so caller can utilize the tx if it fails return tx, err } diff --git a/pm/sendermonitor_test.go b/pm/sendermonitor_test.go index 88a32a957b..725398bca8 100644 --- a/pm/sendermonitor_test.go +++ b/pm/sendermonitor_test.go @@ -972,3 +972,155 @@ func stubLocalSenderMonitorCfg() *LocalSenderMonitorConfig { RPCTimeout: 5 * time.Minute, } } + +// The broker requires deposit + reserve to cover the full ticket face value +// (livepeer/protocol#657). The check must happen before any RPC round-trip, because an +// underfunded sender is a steady state that the queue re-evaluates on every L1 block. +func TestRedeemWinningTicket_InsufficientFundsForFaceValue(t *testing.T) { + assert := assert.New(t) + require := require.New(t) + + cfg, b, smgr, tm := localSenderMonitorFixture() + addr := ethcommon.BytesToAddress([]byte("foo")) + smgr.info[addr] = &SenderInfo{ + Deposit: big.NewInt(100), + WithdrawRound: big.NewInt(0), + Reserve: &ReserveInfo{ + FundsRemaining: big.NewInt(0), + ClaimedInCurrentRound: big.NewInt(0), + }, + } + smgr.claimedReserve[addr] = big.NewInt(0) + + // Any RPC reached past the gate turns into a distinguishable error, so seeing + // errInsufficientSenderFunds proves none of them were called. + b.isUsedErr = errors.New("IsUsedTicket should not be called") + cfg.SuggestGasPrice = func(_ context.Context) (*big.Int, error) { + return nil, errors.New("SuggestGasPrice should not be called") + } + + sm := NewSenderMonitor(cfg, b, smgr, tm, newStubTicketStore()) + + funds, err := sm.availableFunds(addr) + require.Nil(err) + require.Equal(int64(100), funds.Int64()) + + signedT := defaultSignedTicket(addr, uint32(0)) + signedT.FaceValue = new(big.Int).Add(funds, big.NewInt(1)) + + _, err = sm.redeemWinningTicket(signedT) + assert.ErrorIs(err, errInsufficientSenderFunds) + // Deferring must never drop the ticket. + assert.False(isNonRetryableTicketErr(err)) + + // Exactly covering the face value is enough, so the redemption proceeds. + b.isUsedErr = nil + cfg.SuggestGasPrice = func(_ context.Context) (*big.Int, error) { return big.NewInt(0), nil } + signedT.FaceValue = funds + + tx, err := sm.redeemWinningTicket(signedT) + assert.Nil(err) + assert.NotNil(tx) +} + +// CheckTx reports a failed transaction without a revert reason. Since a reverted +// redemption now leaves the ticket unused, the broker must be consulted rather than +// assuming the ticket was burned. +func TestRedeemWinningTicket_RevertedRedemptionKeepsTicket(t *testing.T) { + newFixture := func() (*LocalSenderMonitorConfig, *stubBroker, *stubSenderManager, *stubTimeManager, ethcommon.Address) { + cfg, b, smgr, tm := localSenderMonitorFixture() + addr := RandAddress() + smgr.info[addr] = &SenderInfo{ + Deposit: big.NewInt(500), + WithdrawRound: big.NewInt(0), + Reserve: &ReserveInfo{ + FundsRemaining: big.NewInt(1000), + ClaimedInCurrentRound: big.NewInt(0), + }, + } + smgr.claimedReserve[addr] = big.NewInt(0) + b.checkTxErr = errors.New("transaction failed txHash=0xdeadbeef") + return cfg, b, smgr, tm, addr + } + + t.Run("ticket not consumed on-chain is retryable", func(t *testing.T) { + assert := assert.New(t) + cfg, b, smgr, tm, addr := newFixture() + b.redeemDoesNotConsume = true + + sm := NewSenderMonitor(cfg, b, smgr, tm, newStubTicketStore()) + _, err := sm.redeemWinningTicket(defaultSignedTicket(addr, uint32(0))) + + assert.Error(err) + assert.False(isNonRetryableTicketErr(err), "a ticket left unused on-chain must be retried, not dropped") + }) + + t.Run("ticket consumed on-chain is not retryable", func(t *testing.T) { + assert := assert.New(t) + cfg, b, smgr, tm, addr := newFixture() + b.redeemDoesNotConsume = false + + sm := NewSenderMonitor(cfg, b, smgr, tm, newStubTicketStore()) + _, err := sm.redeemWinningTicket(defaultSignedTicket(addr, uint32(0))) + + assert.Error(err) + assert.True(isNonRetryableTicketErr(err), "a consumed ticket must not be retried") + }) +} + +func TestIsNonRetryableTicketErr(t *testing.T) { + txFailed := errors.New("transaction failed txHash=0xdeadbeef") + + tests := []struct { + name string + err error + want bool + }{ + {"used ticket", errIsUsedTicket, true}, + {"plain transaction failure", txFailed, true}, + {"missing creation round block hash", errors.New("ticket creationRound does not have a block hash"), true}, + {"insufficient sender funds", errInsufficientSenderFunds, false}, + {"wrapped insufficient sender funds", fmt.Errorf("redeem: %w", errInsufficientSenderFunds), false}, + {"reverted without consuming the ticket", unconsumedRedemptionErr{txFailed}, false}, + {"unrelated transient error", errors.New("connection reset by peer"), false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, isNonRetryableTicketErr(tt.err)) + }) + } +} + +// A ticket deferred for insufficient funds must redeem as soon as the sender tops up, +// with no trigger beyond the next block. availableFunds is served from the sender watcher +// cache, which DepositFunded/ReserveFunded keep current, so no extra wiring is needed. +func TestRedeemWinningTicket_RedeemsAfterSenderTopsUp(t *testing.T) { + assert := assert.New(t) + + cfg, b, smgr, tm := localSenderMonitorFixture() + addr := RandAddress() + smgr.info[addr] = &SenderInfo{ + Deposit: big.NewInt(10), + WithdrawRound: big.NewInt(0), + Reserve: &ReserveInfo{ + FundsRemaining: big.NewInt(0), + ClaimedInCurrentRound: big.NewInt(0), + }, + } + smgr.claimedReserve[addr] = big.NewInt(0) + + sm := NewSenderMonitor(cfg, b, smgr, tm, newStubTicketStore()) + signedT := defaultSignedTicket(addr, uint32(0)) // FaceValue 50 > deposit 10 + + _, err := sm.redeemWinningTicket(signedT) + assert.ErrorIs(err, errInsufficientSenderFunds) + assert.False(isNonRetryableTicketErr(err)) + + // The sender funds their deposit; the watcher cache picks it up. + smgr.info[addr].Deposit = big.NewInt(1000) + + tx, err := sm.redeemWinningTicket(signedT) + assert.Nil(err) + assert.NotNil(tx) +} diff --git a/pm/stub.go b/pm/stub.go index a1424bd743..688d1089f8 100644 --- a/pm/stub.go +++ b/pm/stub.go @@ -144,6 +144,10 @@ type stubBroker struct { redeemShouldFail bool getSenderInfoShouldFail bool claimableReserveShouldFail bool + // redeemDoesNotConsume models a redemption that reverts on-chain: a transaction is + // submitted but the ticket is left unused, as the broker does since + // livepeer/protocol#657 when the sender cannot cover the full face value. + redeemDoesNotConsume bool checkTxErr error isUsedErr error @@ -188,7 +192,9 @@ func (b *stubBroker) RedeemWinningTicket(ticket *Ticket, _ []byte, _ *big.Int) ( return nil, fmt.Errorf("stub broker redeem error") } - b.usedTickets[ticket.Hash()] = true + if !b.redeemDoesNotConsume { + b.usedTickets[ticket.Hash()] = true + } return types.NewTx(&types.DynamicFeeTx{}), nil } diff --git a/pm/validator.go b/pm/validator.go index eb459b9991..fc86430fa3 100644 --- a/pm/validator.go +++ b/pm/validator.go @@ -16,8 +16,25 @@ var ( errInvalidCreationRound = errors.New("invalid ticket creation round") errInvalidCreationRoundBlockHash = errors.New("invalid ticket creation round block hash") errIsUsedTicket = errors.New("ticket already used") + + // errInsufficientSenderFunds mirrors the TicketBroker precondition introduced in + // livepeer/protocol#657: a redemption only succeeds if the sender's deposit and + // reserve cover the full ticket face value. This is an expected, self-resolving + // state for an underfunded sender rather than a failure, so it is retryable. + errInsufficientSenderFunds = errors.New("sender deposit and reserve insufficient to cover ticket face value") ) +// unconsumedRedemptionErr wraps a redemption failure for which the broker reports that +// the ticket was not consumed on-chain. Since livepeer/protocol#657 a reverted redemption +// leaves the ticket unused and still redeemable, so it must not be dropped locally. +type unconsumedRedemptionErr struct { + err error +} + +func (e unconsumedRedemptionErr) Error() string { return e.err.Error() } + +func (e unconsumedRedemptionErr) Unwrap() error { return e.err } + // Validator is an interface which describes an object capable // of validating tickets type Validator interface {