diff --git a/.github/workflows/docker.yaml b/.github/workflows/docker.yaml index 5ecfc8379c..56813d0d07 100644 --- a/.github/workflows/docker.yaml +++ b/.github/workflows/docker.yaml @@ -82,6 +82,7 @@ jobs: with: build-args: | BUILD_TAGS=${{ steps.build-tag.outputs.build-tags }} + GIT_REVISION=${{ github.event.pull_request.head.sha || github.sha }} context: . provenance: mode=max sbom: true @@ -92,8 +93,10 @@ jobs: tags: ${{ steps.meta.outputs.tags }} annotations: ${{ steps.meta.outputs.annotations }} labels: ${{ steps.meta.outputs.labels }} - cache-from: type=registry,ref=livepeerci/build:cache - cache-to: type=registry,ref=livepeerci/build:cache,mode=max + # PR builds must not reuse the shared registry cache — it can serve a + # stale livepeer binary while metadata tags the PR head SHA (see #3947). + cache-from: ${{ github.event_name != 'pull_request' && 'type=registry,ref=livepeerci/build:cache' || '' }} + cache-to: ${{ github.event_name != 'pull_request' && 'type=registry,ref=livepeerci/build:cache,mode=max' || '' }} builder: name: go-livepeer builder docker image generation diff --git a/core/capabilities.go b/core/capabilities.go index e1ecd62576..838ede18f5 100644 --- a/core/capabilities.go +++ b/core/capabilities.go @@ -5,6 +5,8 @@ import ( "encoding/json" "errors" "fmt" + "sort" + "strings" "sync" "github.com/Masterminds/semver/v3" @@ -791,6 +793,80 @@ func (bcast *Capabilities) PerCapability() PerCapabilityConstraints { return nil } +// ModelIDForCapability returns the constrained model ID for the given capability. +// When multiple models are constrained it returns the lexicographically first one +// so the result is deterministic. Returns "" when no model is constrained. +func (bcast *Capabilities) ModelIDForCapability(cap Capability) string { + if bcast == nil { + return "" + } + capConstraints, ok := bcast.constraints.perCapability[cap] + if !ok || capConstraints == nil { + return "" + } + modelIDs := make([]string, 0, len(capConstraints.Models)) + for modelID := range capConstraints.Models { + if modelID != "" { + modelIDs = append(modelIDs, modelID) + } + } + if len(modelIDs) == 0 { + return "" + } + sort.Strings(modelIDs) + return modelIDs[0] +} + +// CapabilityToPipeline converts a capability to a pipeline slug. +// Returns "" when the capability is unknown. +func CapabilityToPipeline(cap Capability) string { + capName, ok := CapabilityNameLookup[cap] + if !ok { + return "" + } + return strings.ReplaceAll(strings.ToLower(capName), " ", "-") +} + +// ConstrainedPipelineModelID returns a deterministic pipeline/model_id pair +// from per-capability constraints. When multiple capabilities have constrained +// models, it returns the lexicographically first pipeline. +func (bcast *Capabilities) ConstrainedPipelineModelID() (string, string) { + if bcast == nil { + return "", "" + } + + type capPipeline struct { + capability Capability + pipeline string + } + + candidates := []capPipeline{} + for capability := range bcast.constraints.perCapability { + modelID := bcast.ModelIDForCapability(capability) + if modelID == "" { + continue + } + pipeline := CapabilityToPipeline(capability) + if pipeline == "" { + continue + } + candidates = append(candidates, capPipeline{ + capability: capability, + pipeline: pipeline, + }) + } + if len(candidates) == 0 { + return "", "" + } + + sort.Slice(candidates, func(i, j int) bool { + return candidates[i].pipeline < candidates[j].pipeline + }) + + selected := candidates[0] + return selected.pipeline, bcast.ModelIDForCapability(selected.capability) +} + func (bcast *Capabilities) SetMinVersionConstraint(minVersionConstraint string) { if bcast != nil { bcast.constraints.minVersion = minVersionConstraint diff --git a/core/capabilities_test.go b/core/capabilities_test.go index 59f45b0582..db3f0a6e61 100644 --- a/core/capabilities_test.go +++ b/core/capabilities_test.go @@ -751,6 +751,110 @@ func TestMinRunnerVersion(t *testing.T) { assert.Equal("", c.MinRunnerVersionConstraint(Capability_LiveVideoToVideo, "other")) } +func TestModelIDForCapability(t *testing.T) { + assert := assert.New(t) + + // nil receiver is safe + var nilCaps *Capabilities + assert.Equal("", nilCaps.ModelIDForCapability(Capability_LiveVideoToVideo)) + + // no constraints for the capability + c := NewCapabilities([]Capability{Capability_LiveVideoToVideo}, nil) + assert.Equal("", c.ModelIDForCapability(Capability_LiveVideoToVideo)) + + // single constrained model + c.SetPerCapabilityConstraints(PerCapabilityConstraints{ + Capability_LiveVideoToVideo: &CapabilityConstraints{ + Models: ModelConstraints{"streamdiffusion": &ModelConstraint{Warm: true}}, + }, + }) + assert.Equal("streamdiffusion", c.ModelIDForCapability(Capability_LiveVideoToVideo)) + + // different capability has no constraint + assert.Equal("", c.ModelIDForCapability(Capability_TextToImage)) + + // empty model IDs are ignored + c.SetPerCapabilityConstraints(PerCapabilityConstraints{ + Capability_LiveVideoToVideo: &CapabilityConstraints{ + Models: ModelConstraints{"": &ModelConstraint{}}, + }, + }) + assert.Equal("", c.ModelIDForCapability(Capability_LiveVideoToVideo)) + + // one real model plus empty key resolves to the real model + c.SetPerCapabilityConstraints(PerCapabilityConstraints{ + Capability_LiveVideoToVideo: &CapabilityConstraints{ + Models: ModelConstraints{ + "": &ModelConstraint{}, + "streamdiffusion": &ModelConstraint{Warm: true}, + }, + }, + }) + assert.Equal("streamdiffusion", c.ModelIDForCapability(Capability_LiveVideoToVideo)) + + // multiple models resolve to the lexicographically first ID + c.SetPerCapabilityConstraints(PerCapabilityConstraints{ + Capability_LiveVideoToVideo: &CapabilityConstraints{ + Models: ModelConstraints{ + "comfyui": &ModelConstraint{}, + "streamdiffusion": &ModelConstraint{}, + "noop": &ModelConstraint{}, + }, + }, + }) + assert.Equal("comfyui", c.ModelIDForCapability(Capability_LiveVideoToVideo)) +} + +func TestCapabilityToPipeline(t *testing.T) { + assert := assert.New(t) + + assert.Equal("live-video-to-video", CapabilityToPipeline(Capability_LiveVideoToVideo)) + assert.Equal("text-to-image", CapabilityToPipeline(Capability_TextToImage)) + assert.Equal("", CapabilityToPipeline(Capability(-999))) +} + +func TestConstrainedPipelineModelID(t *testing.T) { + assert := assert.New(t) + + var nilCaps *Capabilities + pipeline, modelID := nilCaps.ConstrainedPipelineModelID() + assert.Equal("", pipeline) + assert.Equal("", modelID) + + c := NewCapabilities([]Capability{Capability_LiveVideoToVideo, Capability_TextToImage}, nil) + pipeline, modelID = c.ConstrainedPipelineModelID() + assert.Equal("", pipeline) + assert.Equal("", modelID) + + c.SetPerCapabilityConstraints(PerCapabilityConstraints{ + Capability_TextToImage: &CapabilityConstraints{ + Models: ModelConstraints{ + "stabilityai/sd-turbo": &ModelConstraint{Warm: true}, + }, + }, + }) + pipeline, modelID = c.ConstrainedPipelineModelID() + assert.Equal("text-to-image", pipeline) + assert.Equal("stabilityai/sd-turbo", modelID) + + // Multiple constrained capabilities resolve deterministically. + c.SetPerCapabilityConstraints(PerCapabilityConstraints{ + Capability_LiveVideoToVideo: &CapabilityConstraints{ + Models: ModelConstraints{ + "streamdiffusion": &ModelConstraint{}, + }, + }, + Capability_AudioToText: &CapabilityConstraints{ + Models: ModelConstraints{ + "whisper-large": &ModelConstraint{}, + }, + }, + }) + pipeline, modelID = c.ConstrainedPipelineModelID() + assert.Equal("audio-to-text", pipeline) + assert.Equal("whisper-large", modelID) +} + func (c *Constraints) addCapabilityConstraints(cap Capability, constraint CapabilityConstraints) { // the capability should be added by AddCapacity for modelID, modelConstraint := range constraint.Models { diff --git a/docker/Dockerfile b/docker/Dockerfile index efa33a4d47..1ae4ed8032 100644 --- a/docker/Dockerfile +++ b/docker/Dockerfile @@ -61,6 +61,9 @@ COPY ./install_ffmpeg.sh ./install_ffmpeg.sh ARG BUILD_TAGS ENV BUILD_TAGS=${BUILD_TAGS} +ARG GIT_REVISION=unknown +RUN echo "Building livepeer at ${GIT_REVISION}" + COPY go.mod go.sum ./ RUN go mod download diff --git a/server/remote_signer.go b/server/remote_signer.go index fd926cf574..f34cdead52 100644 --- a/server/remote_signer.go +++ b/server/remote_signer.go @@ -269,7 +269,7 @@ type authResponse struct { // Unix timestamp (seconds) until which auth is considered valid. // Allows for skipping webhook callbacks until this time is exceeded. Expiry int64 `json:"expiry,omitempty"` - // Optional opaque identifier. + // Optional opaque identifier for metering attribution. AuthID string `json:"auth_id,omitempty"` } @@ -681,7 +681,11 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req sessionStatus = "new" } pipeline := "" - if req.Type == RemoteType_LiveVideoToVideo { + modelID := "" + if streamParams.Capabilities != nil { + pipeline, modelID = streamParams.Capabilities.ConstrainedPipelineModelID() + } + if pipeline == "" && req.Type == RemoteType_LiveVideoToVideo { pipeline = PipelineLiveVideoToVideo } // NB: This could could drop events if tha Kafka queue is full! @@ -689,6 +693,7 @@ func (ls *LivepeerServer) GenerateLivePayment(w http.ResponseWriter, r *http.Req "session_id": state.StateID, "session_status": sessionStatus, "pipeline": pipeline, + "model_id": modelID, "request_id": requestID, "orch_address": orchAddr.Hex(), "orch_url": oInfo.Transcoder,