common,pm: Redeem tickets only if recipient is active

CHANGELOG_PENDING.md

@@ -13,6 +13,7 @@
 - \#2171 Make transactions compatible with Arbitrum and fix setting max gas price (@leszko)
 - \#2252 Use different hardcoded redeemGas when connected to an Arbitrum network (@leszko)
 - \#2251 Add fail fast for Arbitrum One Mainnet when LIP-73 has not been activated yet (@leszko)
+- \#2253 Redeem tickets only when recipient is active (@leszko)
 
 #### Broadcaster
 

common/db.go

@@ -567,6 +567,19 @@ func (db *DB) OrchCount(filter *DBOrchFilter) (int, error) {
 	return int(count64), nil
 }
 
+func (db *DB) IsOrchActive(addr ethcommon.Address, round *big.Int) (bool, error) {
+	orchs, err := db.SelectOrchs(
+		&DBOrchFilter{
+			CurrentRound: round,
+			Addresses:    []ethcommon.Address{addr},
+		},
+	)
+	if err != nil {
+		return false, err
+	}
+	return len(orchs) > 0, nil
+}
+
 func (db *DB) InsertUnbondingLock(id *big.Int, delegator ethcommon.Address, amount, withdrawRound *big.Int) error {
 	glog.V(DEBUG).Infof("db: Inserting unbonding lock %v for delegator %v", id, delegator.Hex())
 	_, err := db.insertUnbondingLock.Exec(id.Int64(), delegator.Hex(), amount.String(), withdrawRound.Int64())

common/db_test.go

@@ -564,6 +564,34 @@ func TestDBFilterOrchs(t *testing.T) {
 	assert.Len(orchsFiltered, 0)
 }
 
+func TestIsOrchActive(t *testing.T) {
+	assert := assert.New(t)
+
+	dbh, dbraw, _ := TempDB(t)
+	defer dbh.Close()
+	defer dbraw.Close()
+
+	addr := pm.RandAddress().String()
+	activationRound := 5
+	orch := NewDBOrch(addr, "https://127.0.0.1:8936", 1, int64(activationRound), int64(activationRound+2), 0)
+	dbh.UpdateOrch(orch)
+
+	// inactive in round 4
+	isActive, err := dbh.IsOrchActive(ethcommon.HexToAddress(addr), big.NewInt(4))
+	assert.NoError(err)
+	assert.False(isActive)
+
+	// active in round 5
+	isActive, err = dbh.IsOrchActive(ethcommon.HexToAddress(addr), big.NewInt(5))
+	assert.NoError(err)
+	assert.True(isActive)
+
+	// active in round 6
+	isActive, err = dbh.IsOrchActive(ethcommon.HexToAddress(addr), big.NewInt(6))
+	assert.NoError(err)
+	assert.True(isActive)
+}
+
 func TestDBUnbondingLocks(t *testing.T) {
 	dbh, dbraw, err := TempDB(t)
 	defer dbh.Close()

pm/queue.go

@@ -124,13 +124,16 @@ func (q *ticketQueue) handleBlockEvent(latestL1Block *big.Int) {
 	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 {
-			glog.Errorf("Unable select earliest winning ticket err=%q", err)
+			glog.Errorf("Unable to select earliest winning ticket err=%q", err)
 			return
 		}
 		if nextTicket == nil {
 			return
 		}
-
+		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
+		}
 		if nextTicket.ParamsExpirationBlock.Cmp(latestL1Block) <= 0 {
 			resCh := make(chan struct {
 				txHash ethcommon.Hash
@@ -164,3 +167,13 @@ func isNonRetryableTicketErr(err error) bool {
 	// The latter check depends on logic in eth.client.CheckTx()
 	return err == errIsUsedTicket || strings.Contains(err.Error(), "transaction failed")
 }
+
+func (q *ticketQueue) isRecipientActive(addr ethcommon.Address) bool {
+	isActive, err := q.store.IsOrchActive(addr, q.tm.LastInitializedRound())
+	if err != nil {
+		glog.Errorf("Unable to select an active orchestrator")
+		// In the case of error, assume recipient is active in order to try to redeem the ticket
+		return true
+	}
+	return isActive
+}

pm/queue_test.go

@@ -329,3 +329,29 @@ func TestTicketQueue_Length(t *testing.T) {
 	assert.Nil(err)
 	assert.Equal(qlen, 3)
 }
+
+func TestIsRecipientActive(t *testing.T) {
+	assert := assert.New(t)
+
+	ts := newStubTicketStore()
+	tm := &stubTimeManager{round: big.NewInt(100)}
+	sm := &LocalSenderMonitor{
+		ticketStore: ts,
+		tm:          tm,
+	}
+	sender := RandAddress()
+	q := newTicketQueue(sender, sm)
+	addr := RandAddress()
+
+	// active
+	ts.isActive = true
+	assert.True(q.isRecipientActive(addr))
+
+	// inactive
+	ts.isActive = false
+	assert.False(q.isRecipientActive(addr))
+
+	// db error
+	ts.err = errors.New("some error")
+	assert.True(q.isRecipientActive(addr))
+}

pm/stub.go

@@ -16,6 +16,7 @@ import (
 type stubBlockStore struct {
 	lastBlock *big.Int
 	err       error
+	isActive  bool
 }
 
 type stubTicketStore struct {
@@ -32,6 +33,9 @@ func newStubTicketStore() *stubTicketStore {
 	return &stubTicketStore{
 		tickets:   make(map[ethcommon.Address][]*SignedTicket),
 		submitted: make(map[string]bool),
+		stubBlockStore: stubBlockStore{
+			isActive: true,
+		},
 	}
 }
 
@@ -110,6 +114,10 @@ func (ts *stubTicketStore) WinningTicketCount(sender ethcommon.Address, minCreat
 	return count, nil
 }
 
+func (ts *stubTicketStore) IsOrchActive(addr ethcommon.Address, round *big.Int) (bool, error) {
+	return ts.isActive, ts.err
+}
+
 func (ts *stubBlockStore) LastSeenBlock() (*big.Int, error) {
 	return ts.lastBlock, ts.err
 }

pm/ticketstore.go

@@ -2,6 +2,7 @@ package pm
 
 import (
 	ethcommon "github.com/ethereum/go-ethereum/common"
+	"math/big"
 )
 
 // TicketStore is an interface which describes an object capable
@@ -23,4 +24,7 @@ type TicketStore interface {
 
 	// WinningTicketCount returns the amount of non-redeemed winning tickets for a sender in the TicketStore
 	WinningTicketCount(sender ethcommon.Address, minCreationRound int64) (int, error)
+
+	// IsOrchActive returns true if the given orchestrator addr is active in the given round
+	IsOrchActive(addr ethcommon.Address, round *big.Int) (bool, error)
 }