diff --git a/cmd/livepeer/starter/flags.go b/cmd/livepeer/starter/flags.go index 8a8c724f2e..b8c144e27c 100644 --- a/cmd/livepeer/starter/flags.go +++ b/cmd/livepeer/starter/flags.go @@ -101,6 +101,8 @@ func NewLivepeerConfig(fs *flag.FlagSet) LivepeerConfig { cfg.MaxTotalEV = fs.String("maxTotalEV", *cfg.MaxTotalEV, "The maximum acceptable expected value for one PM payment") // Broadcaster deposit multiplier to determine max acceptable ticket faceValue cfg.DepositMultiplier = fs.Int("depositMultiplier", *cfg.DepositMultiplier, "The deposit multiplier used to determine max acceptable faceValue for PM tickets") + // Orchestrator automatic pruning of dead winning tickets + cfg.TicketPrune = fs.Bool("ticketPrune", *cfg.TicketPrune, "Enable automatic pruning of expired and permanently failed winning tickets from the ticket queue database") // Orchestrator base pricing info cfg.PricePerUnit = fs.String("pricePerUnit", "0", "The price per 'pixelsPerUnit' amount pixels. Can be specified in wei or a custom currency in the format (e.g. 0.50USD). When using a custom currency, a corresponding price feed must be configured with -priceFeedAddr") // Unit of pixels for both O's pricePerUnit and B's maxPricePerUnit diff --git a/cmd/livepeer/starter/starter.go b/cmd/livepeer/starter/starter.go index cf3aca3dbd..8f7327c528 100755 --- a/cmd/livepeer/starter/starter.go +++ b/cmd/livepeer/starter/starter.go @@ -138,6 +138,7 @@ type LivepeerConfig struct { MaxTicketEV *string MaxTotalEV *string DepositMultiplier *int + TicketPrune *bool PricePerUnit *string PixelsPerUnit *string PriceFeedAddr *string @@ -268,6 +269,7 @@ func DefaultLivepeerConfig() LivepeerConfig { defaultMaxTicketEV := "3000000000000" defaultMaxTotalEV := "20000000000000" defaultDepositMultiplier := 1 + defaultTicketPrune := true defaultMaxPricePerUnit := "0" defaultMaxPricePerCapability := "" defaultIgnoreMaxPriceIfNeeded := false @@ -393,6 +395,7 @@ func DefaultLivepeerConfig() LivepeerConfig { MaxTicketEV: &defaultMaxTicketEV, MaxTotalEV: &defaultMaxTotalEV, DepositMultiplier: &defaultDepositMultiplier, + TicketPrune: &defaultTicketPrune, MaxPricePerUnit: &defaultMaxPricePerUnit, MaxPricePerCapability: &defaultMaxPricePerCapability, IgnoreMaxPriceIfNeeded: &defaultIgnoreMaxPriceIfNeeded, @@ -987,6 +990,7 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) { RedeemGas: redeemGas, SuggestGasPrice: client.Backend().SuggestGasPrice, RPCTimeout: ethRPCTimeout, + TicketPrune: *cfg.TicketPrune, } if *cfg.Orchestrator { diff --git a/common/db.go b/common/db.go index d43232e853..96df132008 100644 --- a/common/db.go +++ b/common/db.go @@ -37,6 +37,7 @@ type DB struct { winningTicketCount *sql.Stmt markWinningTicketRedeemed *sql.Stmt removeWinningTicket *sql.Stmt + clearDeadWinningTickets *sql.Stmt insertMiniHeader *sql.Stmt findLatestMiniHeader *sql.Stmt findAllMiniHeadersSortedByNumber *sql.Stmt @@ -324,6 +325,15 @@ func InitDB(dbPath string) (*DB, error) { } d.markWinningTicketRedeemed = stmt + // Clear dead tickets + stmt, err = db.Prepare("DELETE FROM ticketQueue WHERE (redeemedAt IS NULL AND txHash IS NULL AND creationRound < ?) OR (redeemedAt IS NOT NULL AND txHash=?)") + if err != nil { + glog.Error("Unable to prepare clearDeadWinningTickets ", err) + d.Close() + return nil, err + } + d.clearDeadWinningTickets = stmt + // Insert block header stmt, err = db.Prepare("INSERT INTO blockheaders(number, parent, hash, logs) VALUES(?, ?, ?, ?)") if err != nil { @@ -405,6 +415,9 @@ func (db *DB) Close() { if db.removeWinningTicket != nil { db.removeWinningTicket.Close() } + if db.clearDeadWinningTickets != nil { + db.clearDeadWinningTickets.Close() + } if db.insertMiniHeader != nil { db.insertMiniHeader.Close() } @@ -749,6 +762,19 @@ func (db *DB) RemoveWinningTicket(ticket *pm.SignedTicket) error { return nil } +// ClearDeadWinningTickets removes winning tickets that can never be redeemed: unredeemed +// tickets with a creationRound older than minCreationRound (outside the redemption validity +// window) and tickets marked redeemed with a zero txHash (permanently failed redemptions). +// Live tickets within the validity window and successfully redeemed tickets are kept. +// It returns the number of removed tickets. +func (db *DB) ClearDeadWinningTickets(minCreationRound int64) (int64, error) { + res, err := db.clearDeadWinningTickets.Exec(minCreationRound, ethcommon.Hash{}.Hex()) + if err != nil { + return 0, errors.Wrapf(err, "failed clearing dead winning tickets minCreationRound=%v", minCreationRound) + } + return res.RowsAffected() +} + // SelectEarliestWinningTicket selects the earliest stored winning ticket for a 'sender' that is not expired and not yet redeemed func (db *DB) SelectEarliestWinningTicket(sender ethcommon.Address, minCreationRound int64) (*pm.SignedTicket, error) { diff --git a/common/db_test.go b/common/db_test.go index de5e06397d..e8fdb7e8b7 100644 --- a/common/db_test.go +++ b/common/db_test.go @@ -1188,6 +1188,83 @@ func TestRemoveWinningTicket(t *testing.T) { require.Equal(count, 0) } +func TestClearDeadWinningTickets(t *testing.T) { + assert := assert.New(t) + dbh, dbraw, err := TempDB(t) + defer dbh.Close() + defer dbraw.Close() + require := require.New(t) + require.Nil(err) + + minCreationRound := int64(100) + + // live unredeemed ticket within the redemption validity window + _, ticket, sig, recipientRand := defaultWinningTicket(t) + ticket.CreationRound = minCreationRound + liveTicket := &pm.SignedTicket{ + Ticket: ticket, + Sig: sig, + RecipientRand: recipientRand, + } + err = dbh.StoreWinningTicket(liveTicket) + require.Nil(err) + + // expired unredeemed ticket outside the redemption validity window + _, ticket, sig, recipientRand = defaultWinningTicket(t) + ticket.CreationRound = minCreationRound - 1 + expiredTicket := &pm.SignedTicket{ + Ticket: ticket, + Sig: sig, + RecipientRand: recipientRand, + } + err = dbh.StoreWinningTicket(expiredTicket) + require.Nil(err) + + // ticket whose redemption permanently failed (marked redeemed with a zero txHash) + _, ticket, sig, recipientRand = defaultWinningTicket(t) + ticket.CreationRound = minCreationRound + failedTicket := &pm.SignedTicket{ + Ticket: ticket, + Sig: sig, + RecipientRand: recipientRand, + } + err = dbh.StoreWinningTicket(failedTicket) + require.Nil(err) + err = dbh.MarkWinningTicketRedeemed(failedTicket, ethcommon.Hash{}) + require.Nil(err) + + // successfully redeemed ticket (real txHash) + _, ticket, sig, recipientRand = defaultWinningTicket(t) + ticket.CreationRound = minCreationRound - 1 + redeemedTicket := &pm.SignedTicket{ + Ticket: ticket, + Sig: sig, + RecipientRand: recipientRand, + } + err = dbh.StoreWinningTicket(redeemedTicket) + require.Nil(err) + err = dbh.MarkWinningTicketRedeemed(redeemedTicket, pm.RandHash()) + require.Nil(err) + + require.Equal(4, getRowCountOrFatal("SELECT count(sig) FROM ticketQueue", dbraw, t)) + + // removes only the expired unredeemed and permanently failed tickets + count, err := dbh.ClearDeadWinningTickets(minCreationRound) + assert.Nil(err) + assert.Equal(int64(2), count) + + assert.Equal(2, getRowCountOrFatal("SELECT count(sig) FROM ticketQueue", dbraw, t)) + // the live unredeemed ticket is kept + assert.Equal(1, getRowCountOrFatal("SELECT count(sig) FROM ticketQueue WHERE redeemedAt IS NULL AND txHash IS NULL", dbraw, t)) + // the successfully redeemed ticket is kept + assert.Equal(1, getRowCountOrFatal("SELECT count(sig) FROM ticketQueue WHERE redeemedAt IS NOT NULL", dbraw, t)) + + // clearing again removes nothing + count, err = dbh.ClearDeadWinningTickets(minCreationRound) + assert.Nil(err) + assert.Equal(int64(0), count) +} + func TestInsertMiniHeader_ReturnsFindLatestMiniHeader(t *testing.T) { dbh, dbraw, err := TempDB(t) defer dbh.Close() diff --git a/pm/queue.go b/pm/queue.go index 43fbdd3694..0d1b44f9a1 100644 --- a/pm/queue.go +++ b/pm/queue.go @@ -168,7 +168,9 @@ func isNonRetryableTicketErr(err error) bool { // Depends on logic in eth.client.CheckTx() strings.Contains(err.Error(), "transaction failed") || // Arbitrum L2 happens to return zero as the L1 block hash which results in this non-retryable error - strings.Contains(err.Error(), "ticket creationRound does not have a block hash") + strings.Contains(err.Error(), "ticket creationRound does not have a block hash") || + // Ticket is expired on-chain (past its redemption window); retrying can never succeed + strings.Contains(err.Error(), "ticket is expired") } func (q *ticketQueue) isRecipientActive(addr ethcommon.Address) bool { diff --git a/pm/queue_test.go b/pm/queue_test.go index fdcdea392f..631c629143 100644 --- a/pm/queue_test.go +++ b/pm/queue_test.go @@ -189,10 +189,20 @@ func TestTicketQueueLoop_IsNonRetryableTicketErr_MarkAsRedeemed(t *testing.T) { consumeQueue(qc) assert.True(ts.submitted[fmt.Sprintf("%x", ticket.Sig)]) - // Test that ticket is not marked as redeemed if there is an error checking the tx, but the tx did not fail + // Test that ticket is marked as redeemed if it is expired on-chain ticket = defaultSignedTicket(sender, 2) addTicket(ticket) + qc = &queueConsumer{ + redemptionErr: errors.New("execution reverted: ticket is expired"), + } + consumeQueue(qc) + assert.True(ts.submitted[fmt.Sprintf("%x", ticket.Sig)]) + + // Test that ticket is not marked as redeemed if there is an error checking the tx, but the tx did not fail + ticket = defaultSignedTicket(sender, 3) + addTicket(ticket) + qc = &queueConsumer{ redemptionErr: errors.New("some other error"), } diff --git a/pm/sendermonitor.go b/pm/sendermonitor.go index 5ef5dcfedd..944dd1658e 100644 --- a/pm/sendermonitor.go +++ b/pm/sendermonitor.go @@ -73,6 +73,16 @@ type LocalSenderMonitorConfig struct { RedeemGas int SuggestGasPrice func(context.Context) (*big.Int, error) RPCTimeout time.Duration + + // TicketPrune enables automatic pruning of dead winning tickets + // (expired unredeemed or permanently failed) from the ticket store + TicketPrune bool +} + +// ticketStoreCleaner describes a TicketStore implementation that is also +// capable of clearing dead winning tickets from persistent storage +type ticketStoreCleaner interface { + ClearDeadWinningTickets(minCreationRound int64) (int64, error) } type LocalSenderMonitor struct { @@ -328,12 +338,41 @@ func (sm *LocalSenderMonitor) startCleanupLoop() { select { case <-ticker.C: sm.cleanup() + if sm.cfg.TicketPrune { + sm.pruneDeadTickets() + } case <-sm.quit: return } } } +// pruneDeadTickets removes dead winning tickets from the ticket store. +// A ticket is dead if it is unredeemed and outside of the redemption validity +// window (the redemption loop can never select it again) or if its redemption +// permanently failed. The cutoff matches the redemption window exactly +// (ticketValidityPeriod rounds behind the last initialized round): a ticket is +// pruned as soon as it falls out of the window, on the next cleanup tick, +// rather than a round later. The prune condition is the exact complement of the +// selection condition (creationRound >= LastInit-ticketValidityPeriod), so a +// still-redeemable ticket is never pruned. +func (sm *LocalSenderMonitor) pruneDeadTickets() { + cleaner, ok := sm.ticketStore.(ticketStoreCleaner) + if !ok { + return + } + + minCreationRound := new(big.Int).Sub(sm.tm.LastInitializedRound(), big.NewInt(ticketValidityPeriod)).Int64() + count, err := cleaner.ClearDeadWinningTickets(minCreationRound) + if err != nil { + glog.Errorf("Unable to prune dead winning tickets err=%q", err) + return + } + if count > 0 { + glog.Infof("Pruned dead winning tickets count=%d", count) + } +} + // cleanup removes tracked remote senders that have exceeded // their ttl func (sm *LocalSenderMonitor) cleanup() {