Add min acceptable livepeer version as the broadcaster param

cmd/livepeer/livepeer.go

@@ -127,6 +127,7 @@ func parseLivepeerConfig() starter.LivepeerConfig {
 	cfg.OrchAddr = flag.String("orchAddr", *cfg.OrchAddr, "Comma-separated list of orchestrators to connect to")
 	cfg.OrchWebhookURL = flag.String("orchWebhookUrl", *cfg.OrchWebhookURL, "Orchestrator discovery callback URL")
 	cfg.OrchBlacklist = flag.String("orchBlocklist", "", "Comma-separated list of blocklisted orchestrators")
+	cfg.OrchMinLivepeerVersion = flag.String("orchMinLivepeerVersion", "", "Minimal go-livepeer version orchestrator should have to be selected")
 	cfg.SelectRandWeight = flag.Float64("selectRandFreq", *cfg.SelectRandWeight, "Weight of the random factor in the orchestrator selection algorithm")
 	cfg.SelectStakeWeight = flag.Float64("selectStakeWeight", *cfg.SelectStakeWeight, "Weight of the stake factor in the orchestrator selection algorithm")
 	cfg.SelectPriceWeight = flag.Float64("selectPriceWeight", *cfg.SelectPriceWeight, "Weight of the price factor in the orchestrator selection algorithm")

cmd/livepeer/starter/starter.go

@@ -144,6 +144,7 @@ type LivepeerConfig struct {
 	AuthWebhookURL          *string
 	OrchWebhookURL          *string
 	OrchBlacklist           *string
+	OrchMinLivepeerVersion  *string
 	TestOrchAvail           *bool
 }
 
@@ -1169,6 +1170,9 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 	}
 
 	n.Capabilities = core.NewCapabilities(transcoderCaps, core.MandatoryOCapabilities())
+	if cfg.OrchMinLivepeerVersion != nil {
+		n.Capabilities.AddMinVersion(core.NewMinVersionContraint(*cfg.OrchMinLivepeerVersion))
+	}
 
 	if drivers.NodeStorage == nil {
 		// base URI will be empty for broadcasters; that's OK

core/capabilities.go

@@ -329,6 +329,7 @@ func (bcast *Capabilities) CompatibleWith(orch *net.Capabilities) bool {
 		return false
 	}
 
+	// TODO Compare bcast.constraints.minVersion <= orch.Version instead of this mock equality check
 	if bcast.constraints.minVersion != orch.Version {
 		return false
 	}
@@ -353,7 +354,7 @@ func (c *Capabilities) ToNetCapabilities() *net.Capabilities {
 	}
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
-	netCaps := &net.Capabilities{Bitstring: c.bitstring, Mandatories: c.mandatories, Version: c.version, Capacities: make(map[uint32]uint32)}
+	netCaps := &net.Capabilities{Bitstring: c.bitstring, Mandatories: c.mandatories, Version: c.version, Capacities: make(map[uint32]uint32), Constraints: &net.Capabilities_Constraints{MinVersion: c.constraints.minVersion}}
 	for capability, capacity := range c.capacities {
 		netCaps.Capacities[uint32(capability)] = uint32(capacity)
 	}
@@ -389,8 +390,12 @@ func CapabilitiesFromNetCapabilities(caps *net.Capabilities) *Capabilities {
 	return coreCaps
 }
 
+func NewMinVersionContraint(minVersion string) *Capabilities {
+	return &Capabilities{constraints: Constraints{minVersion: minVersion}}
+}
+
 func NewCapabilities(caps []Capability, m []Capability) *Capabilities {
-	c := &Capabilities{capacities: make(map[Capability]int)}
+	c := &Capabilities{capacities: make(map[Capability]int), version: LivepeerVersion}
 	if len(caps) > 0 {
 		c.bitstring = NewCapabilityString(caps)
 		// initialize capacities to 1 by default, mandatory capabilities doesn't have capacities
@@ -576,3 +581,9 @@ func (bcast *Capabilities) LegacyOnly() bool {
 	}
 	return bcast.bitstring.CompatibleWith(legacyCapabilityString)
 }
+
+func (bcast *Capabilities) AddMinVersion(capabilities *Capabilities) {
+	if capabilities.constraints.minVersion != "" {
+		bcast.constraints.minVersion = capabilities.constraints.minVersion
+	}
+}

server/broadcast.go

@@ -448,6 +448,7 @@ func (bsm *BroadcastSessionsManager) shouldSkipVerification(sessions []*Broadcas
 }
 
 func NewSessionManager(ctx context.Context, node *core.LivepeerNode, params *core.StreamParameters, sel BroadcastSessionsSelectorFactory) *BroadcastSessionsManager {
+	params.Capabilities.AddMinVersion(node.Capabilities)
 	var trustedPoolSize, untrustedPoolSize float64
 	if node.OrchestratorPool != nil {
 		trustedPoolSize = float64(node.OrchestratorPool.SizeWith(common.ScoreAtLeast(common.Score_Trusted)))

server/mediaserver.go

@@ -433,6 +433,7 @@ func gotRTMPStreamHandler(s *LivepeerServer) func(url *url.URL, rtmpStrm stream.
 func endRTMPStreamHandler(s *LivepeerServer) func(url *url.URL, rtmpStrm stream.RTMPVideoStream) error {
 	return func(url *url.URL, rtmpStrm stream.RTMPVideoStream) error {
 		params := streamParams(rtmpStrm.AppData())
+		params.Capabilities.AddMinVersion(s.LivepeerNode.Capabilities)
 		if params == nil {
 			return errMismatchedParams
 		}