pm: Defer ticket redemption when the sender cannot cover the face value

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

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") ||

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)
+}

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
 	}

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)
+}

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
 }

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 {