ai/live: Remote signer updates (#3870)

* ai/live: Special-case zero NumTickets

Per the code: if the balance is already large enough to cover the
required minimum credit (fee with ticket EV as the floor) then
the code currently does not generate a ticket.

Checking for this case prevents us from creating an empty-ticket
payment. While those are not a huge problem for the orchestrators,
it still leads to undesirable noise across the network.

* ai/live: Check for a required ManifestID

Help out SDK implementers in case they miss this, since the errors
are non-obvious: the orchestrator runs out of balance within a few
minutes, even if payments are being sent regularly.

* ai/live: Simplify state in remote signer unit tests.

Compute the signature for test cases automatically if one was not
provided as part of the test. Saves a bunch of boilerplate.

Author: j0sh

server/remote_signer.go

@@ -26,6 +26,7 @@ import (
 
 const HTTPStatusRefreshSession = 480
 const HTTPStatusPriceExceeded = 481
+const HTTPStatusNoTickets = 482
 const RemoteType_LiveVideoToVideo = "lv2v"
 
 // SignOrchestratorInfo handles signing GetOrchestratorInfo requests for multiple orchestrators
@@ -248,7 +249,8 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 		err   error
 	)
 	reqState, reqSig := req.State.State, req.State.Sig
-	if len(reqState) != 0 || len(reqSig) != 0 {
+	hasState := len(reqState) != 0 || len(reqSig) != 0
+	if hasState {
 		if err := verifyStateSignature(ls, reqState, reqSig); err != nil {
 			err = errors.New("invalid sig")
 			respondJsonError(ctx, w, err, http.StatusBadRequest)
@@ -278,6 +280,12 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 
 	manifestID := req.ManifestID
 	if manifestID == "" {
+		if hasState {
+			// Required for lv2v so stateful requests stay tied to the same id.
+			err := errors.New("missing manifestID")
+			respondJsonError(ctx, w, err, http.StatusBadRequest)
+			return
+		}
 		manifestID = string(core.RandomManifestID())
 	}
 	ctx = clog.AddVal(ctx, "manifest_id", manifestID)
@@ -393,6 +401,15 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req
 		respondJsonError(ctx, w, err, http.StatusInternalServerError)
 		return
 	}
+	if balUpdate.NumTickets <= 0 {
+		// No new tickets are needed when reserved balance already covers the
+		// required minimum credit (fee with ticket EV as the floor). Caller
+		// should retry once balance has been run down further.
+		err = errors.New("no tickets")
+		clog.Errorf(ctx, "No tickets")
+		respondJsonError(ctx, w, err, HTTPStatusNoTickets)
+		return
+	}
 	if balUpdate.NumTickets > 100 {
 		// Prevent both draining funds and perf issues
 		ev, err := sender.EV(sess.PMSessionID)

server/remote_signer_test.go

@@ -347,25 +347,42 @@ func TestGenerateLivePayment_StateValidationErrors(t *testing.T) {
 		return sig
 	}
 
-	priceIncreaseStateBytes, priceIncreaseStateSig := func() ([]byte, []byte) {
+	priceIncreaseStateBytes := func() []byte {
 		stateBytes, err := json.Marshal(RemotePaymentState{
 			StateID:              "state",
 			OrchestratorAddress:  ethcommon.BytesToAddress(orchInfo.Address),
 			InitialPricePerUnit:  100,
 			InitialPixelsPerUnit: 1,
 		})
 		require.NoError(err)
-		return stateBytes, sign(stateBytes)
+		return stateBytes
 	}()
 
 	tests := []struct {
-		name       string
-		stateBytes []byte
-		stateSig   []byte
-		orchInfo   *net.OrchestratorInfo
-		wantStatus int
-		wantMsg    string
+		name           string
+		stateBytes     []byte
+		stateSig       []byte
+		orchInfo       *net.OrchestratorInfo
+		omitManifestID bool
+		wantStatus     int
+		wantMsg        string
 	}{
+		{
+			name: "missing manifest id with state",
+			stateBytes: func() []byte {
+				state, err := json.Marshal(RemotePaymentState{
+					StateID:              "state",
+					OrchestratorAddress:  ethcommon.BytesToAddress(orchInfo.Address),
+					InitialPricePerUnit:  1,
+					InitialPixelsPerUnit: 1,
+				})
+				require.NoError(err)
+				return state
+			}(),
+			omitManifestID: true,
+			wantStatus:     http.StatusBadRequest,
+			wantMsg:        "missing manifestID",
+		},
 		{
 			name:       "invalid state signature",
 			stateBytes: []byte(`{"stateID":"state","orchestratorAddress":"0x1"}`),
@@ -376,7 +393,6 @@ func TestGenerateLivePayment_StateValidationErrors(t *testing.T) {
 		{
 			name:       "invalid state json",
 			stateBytes: []byte("not-json"),
-			stateSig:   sign([]byte("not-json")),
 			wantStatus: http.StatusBadRequest,
 			wantMsg:    "invalid state",
 		},
@@ -391,15 +407,6 @@ func TestGenerateLivePayment_StateValidationErrors(t *testing.T) {
 				require.NoError(err)
 				return state
 			}(),
-			stateSig: sign(func() []byte {
-				state, err := json.Marshal(RemotePaymentState{
-					StateID:              "state",
-					OrchestratorAddress:  ethcommon.HexToAddress("0x1"),
-					InitialPixelsPerUnit: 1,
-				})
-				require.NoError(err)
-				return state
-			}()),
 			orchInfo: func() *net.OrchestratorInfo {
 				oInfo := proto.Clone(orchInfo).(*net.OrchestratorInfo)
 				oInfo.Address = ethcommon.HexToAddress("0x2").Bytes()
@@ -411,7 +418,6 @@ func TestGenerateLivePayment_StateValidationErrors(t *testing.T) {
 		{
 			name:       "orchestrator price increased more than 2x",
 			stateBytes: priceIncreaseStateBytes,
-			stateSig:   priceIncreaseStateSig,
 			orchInfo: func() *net.OrchestratorInfo {
 				oInfo := proto.Clone(orchInfo).(*net.OrchestratorInfo)
 				oInfo.PriceInfo = &net.PriceInfo{PricePerUnit: 250, PixelsPerUnit: 1}
@@ -420,6 +426,23 @@ func TestGenerateLivePayment_StateValidationErrors(t *testing.T) {
 			wantStatus: HTTPStatusPriceExceeded,
 			wantMsg:    "Orchestrator price has more than doubled",
 		},
+		{
+			name: "zero tickets returns 482",
+			stateBytes: func() []byte {
+				stateBytes, err := json.Marshal(RemotePaymentState{
+					StateID:             "state",
+					OrchestratorAddress: ethcommon.BytesToAddress(orchInfo.Address),
+					// Existing balance large enough so StageUpdate yields NumTickets == 0.
+					Balance:              "1000",
+					InitialPricePerUnit:  1,
+					InitialPixelsPerUnit: 1,
+				})
+				require.NoError(err)
+				return stateBytes
+			}(),
+			wantStatus: HTTPStatusNoTickets,
+			wantMsg:    "no tickets",
+		},
 	}
 
 	for _, tt := range tests {
@@ -431,10 +454,20 @@ func TestGenerateLivePayment_StateValidationErrors(t *testing.T) {
 			orchBlob, err := proto.Marshal(oInfo)
 			require.NoError(err)
 
+			var manifestID string
+			if !tt.omitManifestID {
+				manifestID = "manifest"
+			}
+			stateSig := tt.stateSig
+			if stateSig == nil {
+				stateSig = sign(tt.stateBytes)
+			}
+
 			reqBody, err := json.Marshal(RemotePaymentRequest{
 				Orchestrator: orchBlob,
+				ManifestID:   manifestID,
 				InPixels:     1,
-				State:        RemotePaymentStateSig{State: tt.stateBytes, Sig: tt.stateSig},
+				State:        RemotePaymentStateSig{State: tt.stateBytes, Sig: stateSig},
 			})
 			require.NoError(err)
 
@@ -533,9 +566,11 @@ func TestGenerateLivePayment_LV2V_Succeeds(t *testing.T) {
 	}
 	orchBlob, err := proto.Marshal(oInfo)
 	require.NoError(err)
+	const manifestID = "lv2v-manifest"
 
 	resp, payment := doPayment(RemotePaymentRequest{
 		Orchestrator: orchBlob,
+		ManifestID:   manifestID,
 		InPixels:     inPixels,
 	})
 	require.NotEmpty(resp.Payment)
@@ -579,6 +614,7 @@ func TestGenerateLivePayment_LV2V_Succeeds(t *testing.T) {
 	const inPixelsUpdated int64 = 2500
 	resp2, payment2 := doPayment(RemotePaymentRequest{
 		Orchestrator: orchBlob,
+		ManifestID:   manifestID,
 		InPixels:     inPixelsUpdated,
 		State:        resp.State,
 	})