ai/live: Remote signer implementation for tickets (#3822)

This PR completes the remote signing feature, allowing gateways to retrieve PM tickets for Live AI (live-video-to-video) without requiring any on-chain connectivity or possession of an Ethereum signing key. See the design background for additional motivation and design detail around remote signers. Refer to doc/remote-signer.md for instructions on how to enable this feature.

Retrieving tickets is mostly done via implementing the LivePaymentSender interface with a new implementation: `remotePaymentSender` in `live_payment.go`. The LivePaymentSender implementations (signer or non-signer) is also initialized earlier in the process, before an orchestrator is requested, and stored in the LiveParams struct. This is so the gateway can send an upfront payment to the orchestrator using remote signers. Processing remote payment signing requests happens in the `remote_signer.go` file.

When a job first starts, the gateway sends an upfront payment to the orchestrator encoded in the initial request header. To support this, the API for the `remotePaymentSender` also offers a standalone `RequestPayment` method to retrieve signed tickets without sending them. The non-remote signer does not have a clean, singular method to retrieve tickets; at some point we may codify this behind a proper interface and clean up this bit, but that can come later to avoid introducing additional concepts to an already involved PR.

### Remote Signing Protocol

Refer to the design document for context behind the design of the protocol. Here is some more detail on that:

* There are 2 bits of state: the remote signer's state, and the orchestrator's state (OrchestratorInfo ticket parameters). The remote signing protocol is stateless, and each call to sign tickets returns an updated state. The gateway is responsible for retaining both bits of state in between calls, and re-sending the state to the remote signer.
* The remote signer's state is itself signed to prevent tampering. The OrchestratorInfo data is already signed.
* There is a loose requirement for the gateway to store the payment response since it contains updated OrchestratorInfo data. However, this is not strictly necessary; the existing OrchestratorInfo can be reused until its parameters expire.
* If expired OrchestratorInfo parameters are sent to the signer, the signer will respond with an internal status code of 480 ("HTTPStatusRefreshSession") indicating the client should retrieve a fresh set of parameters using an GetOrchestratorInfo RPC request. This comes at the cost of an additional set of requests to the O and the signer, but the impact should be negligible given that Live AI payments are asynchronous and there is typically a bit of a buffer before the gateway depletes its balance with the O.

```mermaid
sequenceDiagram
    participant O as Orchestrator
    participant G as Gateway
    participant S as Signer

    %% Initial session setup
    G->>S: getOrchInfoSig()
    S-->>G: gatewaySig
    G->>O: getOrchInfo(gatewaySig)
    O-->>G: ticketParams₀

    %% First signing call (no prior signer state)
    Note over S: state is null → create fresh signer state
    G->>S: signTicket(state=null, ticketParams₀)
    S-->>G: signedTicket₀, signerState₀

    G->>O: pay(signedTicket₀)
    O-->>G: ticketParams₁

    G->>S: signTicket(signerState₀, ticketParams₁)
    S-->>G: signedTicket₁, signerState₁

    %% Subsequent calls (k = 1..N)
    loop For each k = 1..N
        Note over S: NB: ticketParamsₖ reusable between<br>calls as long as it is valid but not<br>signedTicketₖ or signerStateₖ
        G->>S: signTicket(signerStateₖ₋₁, ticketParamsₖ)
        S-->>G: signedTicketₖ, signerStateₖ

        G->>O: pay(signedTicketₖ)
        O-->>G: ticketParamsₖ₊₁
    end
```


### PM Changes

All the changes here are used only by the remote signer, so the impact on the existing code is minimal.

The `Sender` interface adds two new methods: a StartSessionWithNonce constructor, and a `Nonce` accessor. The nonce is a (mostly internal) PM construct that allows for multiple tickets to be generated using the same set of PM parameters. For ordinary signers, the `Sender` persists for the duration of the session, so the nonce would stay internal and be incremented as necessary. However, since remote signers are stateless, the nonce needs to be extracted and set with each signing call, and that is what we do here.

The `Balance` struct has a new `Reserve()` method added to zero out the current balance. This addition makes the `Balance` more closely mirror the API of the nested `AddressBalances()` list. (Otherwise I would have chosen a better name than "Reserve" to zero out a balance.)

Note that the `Balances` object itself hides quite a bit of nested global accouting that we don't strictly need here, and it would be much neater to not have to use these in favor of strictly request-local accounting. However, this would make the rest of the implementation more complex, since the `BroadcastSession` works on the global `Balances` and most of the payment helper functions themselves take a `BroadcastSession` ... so here we go.

There is also a small change in starter.go to initialize more PM and Ethereum scaffolding (watchers etc) when the node starts up in remote signer mode.

Author: j0sh

cmd/livepeer/starter/starter.go

@@ -1092,7 +1092,7 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 			}
 
 		}
-		if n.NodeType == core.BroadcasterNode {
+		if n.NodeType == core.BroadcasterNode || n.NodeType == core.RemoteSignerNode {
 			maxEV, _ := new(big.Rat).SetString(*cfg.MaxTicketEV)
 			if maxEV == nil {
 				panic(fmt.Errorf("-maxTicketEV must be a valid rational number, but %v provided. Restart the node with a valid value for -maxTicketEV", *cfg.MaxTicketEV))

core/accounting.go

@@ -31,6 +31,11 @@ func (b *Balance) Credit(amount *big.Rat) {
 	b.balances.Credit(b.addr, b.manifestID, amount)
 }
 
+// Reserve zeroes the balance and returns the current balance
+func (b *Balance) Reserve() *big.Rat {
+	return b.balances.Reserve(b.addr, b.manifestID)
+}
+
 // StageUpdate prepares a balance update by reserving the current balance and returning the number of tickets
 // to send with a payment, the new credit represented by the payment and the existing credit (i.e reserved balance)
 func (b *Balance) StageUpdate(minCredit, ev *big.Rat) (int, *big.Rat, *big.Rat) {

doc/remote-signer.md

@@ -89,6 +89,23 @@ Example:
   -v 6
 ```
 
+### Pricing checks (gateway vs remote signer)
+
+When running a gateway in offchain mode (ie, with `-remoteSignerUrl` and no Ethereum flags), the gateway does not check orchestrator pricing. Instead, price checks happen in the remote signer during payment generation.
+
+- **Remote signer configuration**: configure the signer with the same pricing and PM knobs you would normally configure on a gateway, e.g.:
+  - `-maxPricePerUnit`, `-pixelsPerUnit`
+  - `-maxPricePerCapability` (optional, capability/model pricing config)
+  - `-maxTicketEV`, `-maxTotalEV`, etc.
+- **Selection behavior**: if an orchestrator’s price is above the signer’s configured limits, the signer rejects the request (HTTP 481) and the gateway will retry with a different orchestrator session.
+- **LV2V session price is fixed**: like a traditional gateway setup, Live Video-to-Video (LV2V) jobs treat price as fixed for the lifetime of the session, captured at session initialization time.
+
+### Tuning ticket EV to avoid “too many tickets” errors
+
+If there are errors about too many tickets (eg `numTickets ... exceeds maximum of 100`), increase the ticket EV on the remote signer so each signing call produces fewer tickets. A good target is ~1–3 tickets per remote signer call.
+
+For PM configuration details and how these knobs interact, see `doc/payments.md`.
+
 ## Operational + security guidance
 
 For the moment, remote signers are intended to sit behind infrastructure controls rather than being exposed directly to end-users. For example, run the remote signer on a private network or behind an authenticated proxy. Do not expose the remote signer to unauthenticated end-users. Run the remote signer close to gateways on a private network; protect it like you would an internal wallet service.

pm/sender.go

@@ -22,6 +22,9 @@ type Sender interface {
 	// for creating new tickets
 	StartSession(ticketParams TicketParams) string
 
+	// StartSessionWithNonce is like StartSession with a non-default nonce
+	StartSessionWithNonce(ticketParams TicketParams, nonce uint32) string
+
 	// CleanupSession deletes session from the internal map
 	CleanupSession(sessionID string)
 
@@ -33,6 +36,9 @@ type Sender interface {
 
 	// EV returns the ticket EV for a session
 	EV(sessionID string) (*big.Rat, error)
+
+	// Nonce returns the current nonce for a session
+	Nonce(sessionID string) (uint32, error)
 }
 
 type session struct {
@@ -75,6 +81,17 @@ func (s *sender) StartSession(ticketParams TicketParams) string {
 	return sessionID
 }
 
+func (s *sender) StartSessionWithNonce(ticketParams TicketParams, nonce uint32) string {
+	sessionID := ticketParams.RecipientRandHash.Hex()
+
+	s.sessions.Store(sessionID, &session{
+		ticketParams: ticketParams,
+		senderNonce:  nonce,
+	})
+
+	return sessionID
+}
+
 // EV returns the ticket EV for a session
 func (s *sender) EV(sessionID string) (*big.Rat, error) {
 	session, err := s.loadSession(sessionID)
@@ -85,6 +102,14 @@ func (s *sender) EV(sessionID string) (*big.Rat, error) {
 	return ticketEV(session.ticketParams.FaceValue, session.ticketParams.WinProb), nil
 }
 
+func (s *sender) Nonce(sessionID string) (uint32, error) {
+	session, err := s.loadSession(sessionID)
+	if err != nil {
+		return 0, err
+	}
+	return session.senderNonce, nil
+}
+
 func (s *sender) CleanupSession(sessionID string) {
 	s.sessions.Delete(sessionID)
 }

pm/stub.go

@@ -511,6 +511,11 @@ func (m *MockSender) StartSession(ticketParams TicketParams) string {
 	return args.String(0)
 }
 
+func (m *MockSender) StartSessionWithNonce(ticketParams TicketParams, nonce uint32) string {
+	args := m.Called(ticketParams, nonce)
+	return args.String(0)
+}
+
 // CleanupSession deletes session from the internal ma
 func (m *MockSender) CleanupSession(sessionID string) {
 	m.Called(sessionID)
@@ -545,3 +550,8 @@ func (m *MockSender) ValidateTicketParams(ticketParams *TicketParams) error {
 	args := m.Called(ticketParams)
 	return args.Error(0)
 }
+
+func (m *MockSender) Nonce(sessionID string) (uint32, error) {
+	args := m.Called(sessionID)
+	return uint32(args.Int(0)), args.Error(1)
+}

server/ai_live_video.go

@@ -83,7 +83,7 @@ func startTricklePublish(ctx context.Context, url *url.URL, params aiRequestPara
 	priceInfo := sess.OrchestratorInfo.PriceInfo
 	var paymentProcessor *LivePaymentProcessor
 	if priceInfo != nil && priceInfo.PricePerUnit != 0 {
-		paymentSender := livePaymentSender{}
+		paymentSender := params.liveParams.paymentSender
 		sendPaymentFunc := func(inPixels int64) error {
 			return paymentSender.SendPayment(context.Background(), &SegmentInfoSender{
 				sess:      sess.BroadcastSession,

server/ai_mediaserver.go

@@ -695,7 +695,7 @@ func processStream(ctx context.Context, params aiRequestParams, req worker.GenLi
 		var err error
 		for {
 			perOrchCtx, perOrchCancel := context.WithCancelCause(ctx)
-			params.liveParams = newParams(params.liveParams, perOrchCancel)
+			params.liveParams = newLiveParams(params, perOrchCancel)
 			var resp interface{}
 			resp, err = processAIRequest(perOrchCtx, params, req)
 			if err != nil {
@@ -761,7 +761,8 @@ func processStream(ctx context.Context, params aiRequestParams, req worker.GenLi
 	<-firstProcessed
 }
 
-func newParams(params *liveRequestParams, cancelOrch context.CancelCauseFunc) *liveRequestParams {
+func newLiveParams(aiParams aiRequestParams, cancelOrch context.CancelCauseFunc) *liveRequestParams {
+	params := aiParams.liveParams
 	return &liveRequestParams{
 		segmentReader:          params.segmentReader,
 		rtmpOutputs:            params.rtmpOutputs,
@@ -778,8 +779,15 @@ func newParams(params *liveRequestParams, cancelOrch context.CancelCauseFunc) *l
 		orchestrator:           params.orchestrator,
 		startTime:              time.Now(),
 		kickOrch:               cancelOrch,
+		paymentSender:          choosePaymentSender(aiParams),
 	}
+}
 
+func choosePaymentSender(params aiRequestParams) LivePaymentSender {
+	if hasRemoteSigner(params) {
+		return NewRemotePaymentSender(params.node)
+	}
+	return &livePaymentSender{}
 }
 
 func startProcessing(ctx context.Context, params aiRequestParams, res interface{}) error {

server/ai_process.go

@@ -104,6 +104,7 @@ type liveRequestParams struct {
 	pipeline      string
 	orchestrator  string
 
+	paymentSender          LivePaymentSender
 	paymentProcessInterval time.Duration
 	outSegmentTimeout      time.Duration
 
@@ -1051,28 +1052,54 @@ func submitLiveVideoToVideo(ctx context.Context, params aiRequestParams, sess *A
 	params.liveParams.sess = sess
 	params.liveParams.startTime = time.Now()
 
-	// Live Video should not reuse the existing session balance, because it could lead to not sending the init
-	// payment, which in turns may cause "Insufficient Balance" on the Orchestrator's side.
-	// It works differently than other AI Jobs, because Live Video is accounted by mid on the Orchestrator's side.
-	clearSessionBalance(sess.BroadcastSession, core.RandomManifestID())
+	var paymentHeaders worker.RequestEditorFn
+	if hasRemoteSigner(params) {
+		rpp, ok := params.liveParams.paymentSender.(*remotePaymentSender)
+		if !ok {
+			return nil, errors.New("remote sender was not the correct type")
+		}
+		res, err := rpp.RequestPayment(ctx, &SegmentInfoSender{
+			sess: sess.BroadcastSession,
+		})
+		if err != nil {
+			return nil, err
+		}
+		paymentHeaders = func(_ context.Context, req *http.Request) error {
+			req.Header.Set(segmentHeader, res.SegCreds)
+			req.Header.Set(paymentHeader, res.Payment)
+			req.Header.Set("Authorization", protoVerAIWorker)
+			return nil
+		}
+	} else {
 
-	client, err := worker.NewClientWithResponses(sess.Transcoder(), worker.WithHTTPClient(httpClient))
-	if err != nil {
-		if monitor.Enabled {
-			monitor.AIRequestError(err.Error(), "LiveVideoToVideo", *req.ModelId, sess.OrchestratorInfo)
+		// Live Video should not reuse the existing session balance, because it could lead to not sending the init
+		// payment, which in turns may cause "Insufficient Balance" on the Orchestrator's side.
+		// It works differently than other AI Jobs, because Live Video is accounted by mid on the Orchestrator's side.
+		clearSessionBalance(sess.BroadcastSession, core.RandomManifestID())
+
+		var (
+			balUpdate *BalanceUpdate
+			err       error
+		)
+		paymentHeaders, balUpdate, err = prepareAIPayment(ctx, sess, initPixelsToPay)
+		if err != nil {
+			if monitor.Enabled {
+				monitor.AIRequestError(err.Error(), "LiveVideoToVideo", *req.ModelId, sess.OrchestratorInfo)
+			}
+			return nil, err
 		}
-		return nil, err
+		defer completeBalanceUpdate(sess.BroadcastSession, balUpdate)
 	}
-	paymentHeaders, balUpdate, err := prepareAIPayment(ctx, sess, initPixelsToPay)
+
+	// Send request to orchestrator
+	client, err := worker.NewClientWithResponses(sess.Transcoder(), worker.WithHTTPClient(httpClient))
 	if err != nil {
 		if monitor.Enabled {
 			monitor.AIRequestError(err.Error(), "LiveVideoToVideo", *req.ModelId, sess.OrchestratorInfo)
 		}
 		return nil, err
 	}
-	defer completeBalanceUpdate(sess.BroadcastSession, balUpdate)
 
-	// Send request to orchestrator
 	reqTimeout := 5 * time.Second
 	reqCtx, cancel := context.WithTimeout(ctx, reqTimeout)
 	defer cancel()
@@ -1669,3 +1696,7 @@ func encodeReqMetadata(metadata map[string]string) string {
 	metadataBytes, _ := json.Marshal(metadata)
 	return string(metadataBytes)
 }
+
+func hasRemoteSigner(params aiRequestParams) bool {
+	return params.node != nil && params.node.RemoteSignerUrl != nil
+}

server/handlers_test.go

@@ -243,6 +243,10 @@ func TestSetBroadcastConfigHandler_TranscodingOptionsError(t *testing.T) {
 func TestSetBroadcastConfigHandler_Success(t *testing.T) {
 	assert := assert.New(t)
 
+	oldProfs := BroadcastJobVideoProfiles
+	defer func() { BroadcastJobVideoProfiles = oldProfs }()
+	defer BroadcastCfg.SetMaxPrice(nil)
+
 	handler := setBroadcastConfigHandler()
 	status, _ := postForm(handler, url.Values{
 		"maxPricePerUnit":    {"1"},
@@ -269,6 +273,7 @@ func TestSetMaxPriceForCapabilityHandler(t *testing.T) {
 	//set default max price
 	basePrice, _ := core.NewAutoConvertedPrice("WEI", big.NewRat(10, 1), nil)
 	BroadcastCfg.SetMaxPrice(basePrice)
+	defer BroadcastCfg.SetMaxPrice(nil)
 
 	//set price per unit for specific pipeline
 	p1, _ := core.NewAutoConvertedPrice("WEI", big.NewRat(1, 1), nil)
@@ -281,6 +286,10 @@ func TestSetMaxPriceForCapabilityHandler(t *testing.T) {
 	p2_pipeline_cap, _ := core.PipelineToCapability(p2_pipeline)
 	p2_modelID := "default"
 
+	defer BroadcastCfg.SetCapabilityMaxPrice(p1_pipeline_cap, "default", nil)
+	defer BroadcastCfg.SetCapabilityMaxPrice(p2_pipeline_cap, "default", nil)
+	defer BroadcastCfg.SetCapabilityMaxPrice(p1_pipeline_cap, "stabilityai/sd-turbo", nil)
+
 	status1, _ := postForm(handler, url.Values{
 		"maxPricePerUnit": {"1"},
 		"pixelsPerUnit":   {"1"},
@@ -450,6 +459,10 @@ func TestGetNetworkCapabilitiesHandler(t *testing.T) {
 func TestGetBroadcastConfigHandler(t *testing.T) {
 	assert := assert.New(t)
 
+	oldProfs := BroadcastJobVideoProfiles
+	defer func() { BroadcastJobVideoProfiles = oldProfs }()
+	defer BroadcastCfg.SetMaxPrice(nil)
+
 	BroadcastCfg.SetMaxPrice(core.NewFixedPrice(big.NewRat(1, 2)))
 	BroadcastJobVideoProfiles = []ffmpeg.VideoProfile{
 		ffmpeg.VideoProfileLookup["P240p25fps16x9"],