Revert accepting pending Os to accept traffic

CHANGELOG_PENDING.md

@@ -7,6 +7,7 @@
 ### Features ⚒
 
 #### General
+- \#2758 Accept only active Os to receive traffic and redeem tickets (@leszko)
 
 #### Broadcaster
 

core/orch_test.go

@@ -789,7 +789,7 @@ func TestProcessPayment_ActiveOrchestrator(t *testing.T) {
 
 	// orchestrator inactive -> error
 	err := orch.ProcessPayment(context.Background(), defaultPayment(t), ManifestID("some manifest"))
-	expErr := fmt.Sprintf("orchestrator %v is not eligible for payments in round %v, cannot process payments", addr.Hex(), 10)
+	expErr := fmt.Sprintf("orchestrator %v is inactive in round %v, cannot process payments", addr.Hex(), 10)
 	assert.EqualError(err, expErr)
 
 	// orchestrator is active -> no error
@@ -1180,43 +1180,30 @@ func TestProcessPayment_PaymentError_DoesNotIncreaseCreditBalance(t *testing.T)
 	assert.Nil(orch.node.Balances.Balance(ethcommon.BytesToAddress(payment.Sender), manifestID))
 }
 
-func TestIsPaymentEligible(t *testing.T) {
+func TestIsActive(t *testing.T) {
 	assert := assert.New(t)
 	addr := defaultRecipient
 	dbh, dbraw := tempDBWithOrch(t, &common.DBOrch{
 		EthereumAddr:      addr.Hex(),
-		ActivationRound:   2,
+		ActivationRound:   1,
 		DeactivationRound: 999,
 	})
 	defer dbh.Close()
 	defer dbraw.Close()
 
-	// not active yet
 	n, _ := NewLivepeerNode(nil, "", dbh)
 	rm := &stubRoundsManager{
-		round: big.NewInt(0),
+		round: big.NewInt(10),
 	}
 	orch := NewOrchestrator(n, rm)
 
-	ok, err := orch.isPaymentEligible(addr)
-	assert.False(ok)
-	assert.NoError(err)
-
-	// pending activation
-	rm.round = big.NewInt(1)
-	ok, err = orch.isPaymentEligible(addr)
-	assert.True(ok)
-	assert.NoError(err)
-
-	// active
-	rm.round = big.NewInt(10)
-	ok, err = orch.isPaymentEligible(addr)
+	ok, err := orch.isActive(addr)
 	assert.True(ok)
 	assert.NoError(err)
 
 	// inactive
 	rm.round = big.NewInt(1000)
-	ok, err = orch.isPaymentEligible(addr)
+	ok, err = orch.isActive(addr)
 	assert.False(ok)
 	assert.NoError(err)
 }

core/orchestrator.go

@@ -119,13 +119,13 @@ func (orch *orchestrator) ProcessPayment(ctx context.Context, payment net.Paymen
 	sender := ethcommon.BytesToAddress(payment.Sender)
 
 	recipientAddr := ethcommon.BytesToAddress(payment.TicketParams.Recipient)
-	ok, err := orch.isPaymentEligible(recipientAddr)
+	ok, err := orch.isActive(recipientAddr)
 	if err != nil {
 		return err
 	}
 
 	if !ok {
-		return fmt.Errorf("orchestrator %v is not eligible for payments in round %v, cannot process payments", recipientAddr.Hex(), orch.rm.LastInitializedRound())
+		return fmt.Errorf("orchestrator %v is inactive in round %v, cannot process payments", recipientAddr.Hex(), orch.rm.LastInitializedRound())
 	}
 
 	priceInfo := payment.GetExpectedPrice()
@@ -349,11 +349,9 @@ func (orch *orchestrator) AuthToken(sessionID string, expiration int64) *net.Aut
 	}
 }
 
-func (orch *orchestrator) isPaymentEligible(addr ethcommon.Address) (bool, error) {
-	// Accept payments when already activated or will be activated in the next round
-	nextRound := new(big.Int).Add(orch.rm.LastInitializedRound(), big.NewInt(1))
+func (orch *orchestrator) isActive(addr ethcommon.Address) (bool, error) {
 	filter := &common.DBOrchFilter{
-		CurrentRound: nextRound,
+		CurrentRound: orch.rm.LastInitializedRound(),
 		Addresses:    []ethcommon.Address{addr},
 	}
 	orchs, err := orch.node.Database.SelectOrchs(filter)

discovery/db_discovery.go

@@ -70,7 +70,7 @@ func (dbo *DBOrchestratorPoolCache) getURLs() ([]*url.URL, error) {
 	orchs, err := dbo.store.SelectOrchs(
 		&common.DBOrchFilter{
 			MaxPrice:       server.BroadcastCfg.MaxPrice(),
-			CurrentRound:   dbo.nextRound(),
+			CurrentRound:   dbo.rm.LastInitializedRound(),
 			UpdatedLastDay: true,
 		},
 	)
@@ -153,7 +153,7 @@ func (dbo *DBOrchestratorPoolCache) Size() int {
 	count, _ := dbo.store.OrchCount(
 		&common.DBOrchFilter{
 			MaxPrice:       server.BroadcastCfg.MaxPrice(),
-			CurrentRound:   dbo.nextRound(),
+			CurrentRound:   dbo.rm.LastInitializedRound(),
 			UpdatedLastDay: true,
 		},
 	)
@@ -185,7 +185,7 @@ func (dbo *DBOrchestratorPoolCache) cacheTranscoderPool() error {
 func (dbo *DBOrchestratorPoolCache) cacheOrchestratorStake() error {
 	orchs, err := dbo.store.SelectOrchs(
 		&common.DBOrchFilter{
-			CurrentRound: dbo.nextRound(),
+			CurrentRound: dbo.rm.LastInitializedRound(),
 		},
 	)
 	if err != nil {
@@ -261,7 +261,7 @@ func (dbo *DBOrchestratorPoolCache) pollOrchestratorInfo(ctx context.Context) er
 func (dbo *DBOrchestratorPoolCache) cacheDBOrchs() error {
 	orchs, err := dbo.store.SelectOrchs(
 		&common.DBOrchFilter{
-			CurrentRound: dbo.nextRound(),
+			CurrentRound: dbo.rm.LastInitializedRound(),
 		},
 	)
 	if err != nil {
@@ -338,13 +338,6 @@ func (dbo *DBOrchestratorPoolCache) cacheDBOrchs() error {
 	return nil
 }
 
-func (dbo *DBOrchestratorPoolCache) nextRound() *big.Int {
-	if dbo.rm.LastInitializedRound() == nil {
-		return nil
-	}
-	return new(big.Int).Add(dbo.rm.LastInitializedRound(), big.NewInt(1))
-}
-
 func parseURI(addr string) (*url.URL, error) {
 	if !strings.HasPrefix(addr, "http") {
 		addr = "https://" + addr

discovery/discovery_test.go

@@ -1071,13 +1071,12 @@ func TestCachedPool_GetOrchestrators_OnlyActiveOrchestrators(t *testing.T) {
 	}
 
 	// check size
-	// 25 active Os + 1 pending O
-	assert.Equal(26, pool.Size())
+	assert.Equal(25, pool.Size())
 
 	infos := pool.GetInfos()
-	assert.Len(infos, 26)
+	assert.Len(infos, 25)
 	for _, info := range infos {
-		assert.Contains(addresses[:26], info.URL.String())
+		assert.Contains(addresses[:25], info.URL.String())
 	}
 	oinfos, err := pool.GetOrchestrators(context.TODO(), 50, newStubSuspender(), newStubCapabilities(), common.ScoreAtLeast(0))
 	for _, info := range oinfos {
@@ -1086,7 +1085,7 @@ func TestCachedPool_GetOrchestrators_OnlyActiveOrchestrators(t *testing.T) {
 	}
 
 	assert.Nil(err, "Should not be error")
-	assert.Len(infos, 26)
+	assert.Len(infos, 25)
 }
 
 func TestNewWHOrchestratorPoolCache(t *testing.T) {