server: Allow Os price to increase up to 2x mid-session

CHANGELOG_PENDING.md

@@ -10,6 +10,8 @@
 
 #### Broadcaster
 
+- [#2995](https://github.com/livepeer/go-livepeer/pull/2995) server: Allow Os price to increase up to 2x mid-session (@victorges)
+
 #### Orchestrator
 
 #### Transcoder

server/rpc_test.go

@@ -550,10 +550,10 @@ func TestGenPayment(t *testing.T) {
 	s.Sender = sender
 
 	// Test changing O price
-	s.InitialPrice = &net.PriceInfo{PricePerUnit: 1, PixelsPerUnit: 5}
+	s.InitialPrice = &net.PriceInfo{PricePerUnit: 1, PixelsPerUnit: 7}
 	payment, err = genPayment(context.TODO(), s, 1)
 	assert.Equal("", payment)
-	assert.Errorf(err, "Orchestrator price has changed, Orchestrator price: %v, Orchestrator initial price: %v", "1/3", "1/5")
+	assert.Errorf(err, "Orchestrator price has more than doubled, Orchestrator price: %v, Orchestrator initial price: %v", "1/3", "1/7")
 
 	s.InitialPrice = nil
 
@@ -694,10 +694,15 @@ func TestValidatePrice(t *testing.T) {
 	err = validatePrice(s)
 	assert.Nil(err)
 
-	// O Initial Price lower than O Price
-	s.InitialPrice = &net.PriceInfo{PricePerUnit: 1, PixelsPerUnit: 10}
+	// O Price higher but up to 2x Initial Price
+	s.InitialPrice = &net.PriceInfo{PricePerUnit: 1, PixelsPerUnit: 6}
 	err = validatePrice(s)
-	assert.ErrorContains(err, "price has changed")
+	assert.Nil(err)
+
+	// O Price higher than 2x Initial Price
+	s.InitialPrice = &net.PriceInfo{PricePerUnit: 1000, PixelsPerUnit: 6001}
+	err = validatePrice(s)
+	assert.ErrorContains(err, "price has more than doubled")
 
 	// O.PriceInfo is nil
 	s.OrchestratorInfo.PriceInfo = nil

server/segment_rpc.go

@@ -36,6 +36,9 @@ const segmentHeader = "Livepeer-Segment"
 
 const pixelEstimateMultiplier = 1.02
 
+// Maximum price change allowed in orchestrator pricing before the session is swapped.
+var priceIncreaseThreshold = big.NewRat(2, 1)
+
 var errSegEncoding = errors.New("ErrorSegEncoding")
 var errSegSig = errors.New("ErrSegSig")
 var errFormat = errors.New("unrecognized profile output format")
@@ -826,9 +829,18 @@ func validatePrice(sess *BroadcastSession) error {
 		return errors.New("missing orchestrator price")
 	}
 
-	iPrice, err := common.RatPriceInfo(sess.InitialPrice)
-	if err == nil && iPrice != nil && oPrice.Cmp(iPrice) == 1 {
-		return fmt.Errorf("Orchestrator price has changed, Orchestrator price: %v, Orchestrator initial price: %v", oPrice, iPrice)
+	initPrice, err := common.RatPriceInfo(sess.InitialPrice)
+	if err != nil {
+		glog.Warningf("Error parsing session initial price (%d / %d): %v",
+			sess.InitialPrice.PricePerUnit, sess.InitialPrice.PixelsPerUnit, err)
+	}
+	if initPrice != nil {
+		// Prices are dynamic if configured with a custom currency, so we need to allow some change during the session.
+		// TODO: Make sure prices stay the same during a session so we can make this logic more strict, disallowing any price changes.
+		maxIncreasedPrice := new(big.Rat).Mul(initPrice, priceIncreaseThreshold)
+		if oPrice.Cmp(maxIncreasedPrice) > 0 {
+			return fmt.Errorf("Orchestrator price has more than doubled, Orchestrator price: %v, Orchestrator initial price: %v", oPrice.RatString(), initPrice.RatString())
+		}
 	}
 
 	return nil

server/segment_rpc_test.go

@@ -1679,13 +1679,13 @@ func TestSubmitSegment_GenPaymentError_ValidatePriceError(t *testing.T) {
 		OrchestratorInfo: oinfo,
 		InitialPrice: &net.PriceInfo{
 			PricePerUnit:  1,
-			PixelsPerUnit: 5,
+			PixelsPerUnit: 7,
 		},
 	}
 
 	_, err := SubmitSegment(context.TODO(), s, &stream.HLSSegment{}, nil, 0, false, true)
 
-	assert.EqualError(t, err, fmt.Sprintf("Orchestrator price has changed, Orchestrator price: %v, Orchestrator initial price: %v", "1/3", "1/5"))
+	assert.EqualError(t, err, fmt.Sprintf("Orchestrator price has more than doubled, Orchestrator price: %v, Orchestrator initial price: %v", "1/3", "1/7"))
 	balance.AssertCalled(t, "Credit", existingCredit)
 }