fix(route): enforce authenticated endpoint capacity

native/sender/internal/app/server_test.go

@@ -632,7 +632,7 @@ func newAppRemoteServer(t *testing.T, terminal chan<- []byte) *httptest.Server {
 		if _, _, err = connection.Read(request.Context()); err != nil {
 			return
 		}
-		authenticated := []byte(`{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}`)
+		authenticated := []byte(`{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"endpointMediaCopyCapacity":2,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}`)
 		if err = connection.Write(request.Context(), websocket.MessageText, authenticated); err != nil {
 			return
 		}
@@ -688,7 +688,7 @@ func newDelayedAppRemoteServer(
 		<-releaseAuthentication
 		writeContext, cancelWrite := context.WithTimeout(context.Background(), time.Second)
 		defer cancelWrite()
-		_ = connection.Write(writeContext, websocket.MessageText, []byte(`{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}`))
+		_ = connection.Write(writeContext, websocket.MessageText, []byte(`{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"endpointMediaCopyCapacity":2,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}`))
 	})
 	return server
 }

native/sender/internal/remote/admission.go

@@ -17,8 +17,8 @@ type viewerAdmission struct {
 	waitSet   map[string]struct{}
 }
 
-func newViewerAdmission(maxViewers int) *viewerAdmission {
-	edgeLimit := maxHostEdges
+func newViewerAdmission(maxViewers, endpointMediaCopyCapacity int) *viewerAdmission {
+	edgeLimit := endpointMediaCopyCapacity
 	if maxViewers < edgeLimit {
 		edgeLimit = maxViewers
 	}
@@ -30,33 +30,53 @@ func newViewerAdmission(maxViewers int) *viewerAdmission {
 	}
 }
 
-func (admission *viewerAdmission) Join(peerID string) admissionAction {
+func (admission *viewerAdmission) Activate(peerID string) admissionAction {
 	if _, exists := admission.active[peerID]; exists {
 		return admissionDuplicate
 	}
+	_, wasWaiting := admission.waitSet[peerID]
+	if !wasWaiting && len(admission.active)+len(admission.waiting) >= admission.roomLimit {
+		return admissionFull
+	}
+	if len(admission.active) >= admission.edgeLimit {
+		return admissionFull
+	}
+	if wasWaiting {
+		admission.removeWaiting(peerID)
+	}
+	admission.active[peerID] = struct{}{}
+	return admissionActivate
+}
+
+func (admission *viewerAdmission) Wait(peerID string) admissionAction {
 	if _, exists := admission.waitSet[peerID]; exists {
 		return admissionDuplicate
 	}
-	if len(admission.active) < admission.edgeLimit {
-		admission.active[peerID] = struct{}{}
-		return admissionActivate
-	}
-	if len(admission.active)+len(admission.waiting) >= admission.roomLimit {
+	_, wasActive := admission.active[peerID]
+	if !wasActive && len(admission.active)+len(admission.waiting) >= admission.roomLimit {
 		return admissionFull
 	}
+	if wasActive {
+		delete(admission.active, peerID)
+	}
 	admission.waiting = append(admission.waiting, peerID)
 	admission.waitSet[peerID] = struct{}{}
 	return admissionWait
 }
 
-func (admission *viewerAdmission) Leave(peerID string) (wasActive bool, promoted string) {
+func (admission *viewerAdmission) Leave(peerID string) (wasActive bool) {
 	if _, exists := admission.active[peerID]; exists {
 		delete(admission.active, peerID)
-		return true, admission.promoteNext()
+		return true
 	}
 	if _, exists := admission.waitSet[peerID]; !exists {
-		return false, ""
+		return false
 	}
+	admission.removeWaiting(peerID)
+	return false
+}
+
+func (admission *viewerAdmission) removeWaiting(peerID string) {
 	delete(admission.waitSet, peerID)
 	for index, waitingID := range admission.waiting {
 		if waitingID != peerID {
@@ -67,7 +87,6 @@ func (admission *viewerAdmission) Leave(peerID string) (wasActive bool, promoted
 		admission.waiting = admission.waiting[:len(admission.waiting)-1]
 		break
 	}
-	return false, ""
 }
 
 func (admission *viewerAdmission) AbortActivation(peerID string) {
@@ -82,15 +101,3 @@ func (admission *viewerAdmission) IsActive(peerID string) bool {
 	_, exists := admission.active[peerID]
 	return exists
 }
-
-func (admission *viewerAdmission) promoteNext() string {
-	if len(admission.waiting) == 0 {
-		return ""
-	}
-	promoted := admission.waiting[0]
-	admission.waiting[0] = ""
-	admission.waiting = admission.waiting[1:]
-	delete(admission.waitSet, promoted)
-	admission.active[promoted] = struct{}{}
-	return promoted
-}

native/sender/internal/remote/admission_test.go

@@ -2,41 +2,44 @@ package remote
 
 import "testing"
 
-func TestViewerAdmissionNeverExceedsTwoEdgesAndPromotesInJoinOrder(t *testing.T) {
-	admission := newViewerAdmission(4)
-	assertAdmission(t, admission, "viewer-a", admissionActivate)
-	assertAdmission(t, admission, "viewer-b", admissionActivate)
-	assertAdmission(t, admission, "viewer-c", admissionWait)
-	assertAdmission(t, admission, "viewer-d", admissionWait)
-	assertAdmission(t, admission, "viewer-c", admissionDuplicate)
-	assertAdmission(t, admission, "viewer-e", admissionFull)
+func TestViewerAdmissionMirrorsAuthoritativeActiveAndWaitingSets(t *testing.T) {
+	admission := newViewerAdmission(4, 2)
+	assertActiveAdmission(t, admission, "viewer-a", admissionActivate)
+	assertActiveAdmission(t, admission, "viewer-b", admissionActivate)
+	assertWaitingAdmission(t, admission, "viewer-c", admissionWait)
+	assertWaitingAdmission(t, admission, "viewer-d", admissionWait)
+	assertWaitingAdmission(t, admission, "viewer-c", admissionDuplicate)
+	assertWaitingAdmission(t, admission, "viewer-e", admissionFull)
 
 	active, waiting := admission.Counts()
 	if active != 2 || waiting != 2 {
 		t.Fatalf("counts = %d active, %d waiting", active, waiting)
 	}
-	wasActive, promoted := admission.Leave("viewer-b")
-	if !wasActive || promoted != "viewer-c" {
-		t.Fatalf("leave promoted %q from active=%t", promoted, wasActive)
+	if !admission.Leave("viewer-b") {
+		t.Fatal("active leave was not recorded")
 	}
-	if admission.IsActive("viewer-b") || !admission.IsActive("viewer-c") {
-		t.Fatal("promotion did not atomically replace the freed edge")
+	if admission.IsActive("viewer-b") || admission.IsActive("viewer-c") {
+		t.Fatal("leave promoted a Viewer without server authority")
 	}
 	active, waiting = admission.Counts()
+	if active != 1 || waiting != 2 {
+		t.Fatalf("post-leave counts = %d active, %d waiting", active, waiting)
+	}
+	assertActiveAdmission(t, admission, "viewer-c", admissionActivate)
+	active, waiting = admission.Counts()
 	if active != 2 || waiting != 1 {
-		t.Fatalf("post-promotion counts = %d active, %d waiting", active, waiting)
+		t.Fatalf("server promotion counts = %d active, %d waiting", active, waiting)
 	}
 }
 
 func TestViewerAdmissionRemovesAWaitingViewerWithoutPromotion(t *testing.T) {
-	admission := newViewerAdmission(3)
-	assertAdmission(t, admission, "viewer-a", admissionActivate)
-	assertAdmission(t, admission, "viewer-b", admissionActivate)
-	assertAdmission(t, admission, "viewer-c", admissionWait)
+	admission := newViewerAdmission(3, 2)
+	assertActiveAdmission(t, admission, "viewer-a", admissionActivate)
+	assertActiveAdmission(t, admission, "viewer-b", admissionActivate)
+	assertWaitingAdmission(t, admission, "viewer-c", admissionWait)
 
-	wasActive, promoted := admission.Leave("viewer-c")
-	if wasActive || promoted != "" {
-		t.Fatalf("waiting leave = active %t, promotion %q", wasActive, promoted)
+	if admission.Leave("viewer-c") {
+		t.Fatal("waiting leave was reported as active")
 	}
 	active, waiting := admission.Counts()
 	if active != 2 || waiting != 0 {
@@ -45,9 +48,9 @@ func TestViewerAdmissionRemovesAWaitingViewerWithoutPromotion(t *testing.T) {
 }
 
 func TestViewerAdmissionUsesTheSmallerRoomLimit(t *testing.T) {
-	admission := newViewerAdmission(1)
-	assertAdmission(t, admission, "viewer-a", admissionActivate)
-	assertAdmission(t, admission, "viewer-b", admissionFull)
+	admission := newViewerAdmission(1, 3)
+	assertActiveAdmission(t, admission, "viewer-a", admissionActivate)
+	assertWaitingAdmission(t, admission, "viewer-b", admissionFull)
 	active, waiting := admission.Counts()
 	if active != 1 || waiting != 0 {
 		t.Fatalf("counts = %d active, %d waiting", active, waiting)
@@ -55,17 +58,15 @@ func TestViewerAdmissionUsesTheSmallerRoomLimit(t *testing.T) {
 }
 
 func TestViewerAdmissionClearsFailedActivation(t *testing.T) {
-	admission := newViewerAdmission(4)
-	assertAdmission(t, admission, "viewer-a", admissionActivate)
-	assertAdmission(t, admission, "viewer-b", admissionActivate)
-	assertAdmission(t, admission, "viewer-c", admissionWait)
-	assertAdmission(t, admission, "viewer-d", admissionWait)
+	admission := newViewerAdmission(4, 2)
+	assertActiveAdmission(t, admission, "viewer-a", admissionActivate)
+	assertActiveAdmission(t, admission, "viewer-b", admissionActivate)
+	assertWaitingAdmission(t, admission, "viewer-c", admissionWait)
+	assertWaitingAdmission(t, admission, "viewer-d", admissionWait)
 
-	_, promoted := admission.Leave("viewer-a")
-	if promoted != "viewer-c" {
-		t.Fatalf("first promotion = %q", promoted)
-	}
-	admission.AbortActivation(promoted)
+	admission.Leave("viewer-a")
+	assertActiveAdmission(t, admission, "viewer-c", admissionActivate)
+	admission.AbortActivation("viewer-c")
 	if admission.IsActive("viewer-c") || admission.IsActive("viewer-d") {
 		t.Fatal("failed activation left a viewer in the active set")
 	}
@@ -75,9 +76,25 @@ func TestViewerAdmissionClearsFailedActivation(t *testing.T) {
 	}
 }
 
-func assertAdmission(t *testing.T, admission *viewerAdmission, peerID string, want admissionAction) {
-	t.Helper()
-	if got := admission.Join(peerID); got != want {
-		t.Fatalf("join %q = %d, want %d", peerID, got, want)
+func TestViewerAdmissionNeverActivatesAWaitingViewerIntoAFreeSlot(t *testing.T) {
+	admission := newViewerAdmission(4, 2)
+	assertWaitingAdmission(t, admission, "viewer-a", admissionWait)
+	active, waiting := admission.Counts()
+	if active != 0 || waiting != 1 || admission.IsActive("viewer-a") {
+		t.Fatalf("waiting authority = %d active, %d waiting", active, waiting)
+	}
+}
+
+func assertActiveAdmission(t *testing.T, admission *viewerAdmission, peerID string, want admissionAction) {
+	t.Helper()
+	if got := admission.Activate(peerID); got != want {
+		t.Fatalf("activate %q = %d, want %d", peerID, got, want)
+	}
+}
+
+func assertWaitingAdmission(t *testing.T, admission *viewerAdmission, peerID string, want admissionAction) {
+	t.Helper()
+	if got := admission.Wait(peerID); got != want {
+		t.Fatalf("wait %q = %d, want %d", peerID, got, want)
 	}
 }

native/sender/internal/remote/session.go

@@ -24,7 +24,6 @@ import (
 )
 
 const (
-	maxHostEdges          = 3
 	maxRemoteResponse     = 64 << 10
 	remoteRequestTimeout  = 10 * time.Second
 	signalWriteTimeout    = 5 * time.Second
@@ -77,28 +76,29 @@ type Session struct {
 	shareGeneration string
 	onEvent         func(Event)
 
-	mu               sync.Mutex
-	writeMu          sync.Mutex
-	conn             *websocket.Conn
-	ice              iceConfig
-	peers            map[string]*peer
-	admission        *viewerAdmission
-	peerAssisted     bool
-	routeRevision    int64
-	routePhase       string
-	routeFailures    [2]string
-	fanout           *media.Fanout
-	audioFanout      *media.AudioFanout
-	codec            media.Codec
-	peerAPI          *webrtc.API
-	codecPreferences []webrtc.RTPCodecParameters
-	closeOnce        sync.Once
-	fatalOnce        sync.Once
-	terminalOnce     sync.Once
-	readDone         chan struct{}
-	terminalResult   chan error
-	readerStarted    atomic.Bool
-	closing          atomic.Bool
+	mu                        sync.Mutex
+	writeMu                   sync.Mutex
+	conn                      *websocket.Conn
+	ice                       iceConfig
+	peers                     map[string]*peer
+	admission                 *viewerAdmission
+	endpointMediaCopyCapacity int
+	peerAssisted              bool
+	routeRevision             int64
+	routePhase                string
+	routeFailures             [2]string
+	fanout                    *media.Fanout
+	audioFanout               *media.AudioFanout
+	codec                     media.Codec
+	peerAPI                   *webrtc.API
+	codecPreferences          []webrtc.RTPCodecParameters
+	closeOnce                 sync.Once
+	fatalOnce                 sync.Once
+	terminalOnce              sync.Once
+	readDone                  chan struct{}
+	terminalResult            chan error
+	readerStarted             atomic.Bool
+	closing                   atomic.Bool
 }
 
 type createRoomResponse struct {
@@ -338,12 +338,13 @@ func (session *Session) connect(ctx context.Context) error {
 	}
 	session.mu.Lock()
 	session.ice = message.IceConfig
+	session.endpointMediaCopyCapacity = message.EndpointMediaCopyCapacity
 	session.peerAssisted = message.PeerAssisted
 	if message.PeerAssisted {
 		session.routeRevision = message.RouteRevision
 		session.routePhase = "active"
 	} else {
-		session.admission = newViewerAdmission(message.MaxViewers)
+		session.admission = newViewerAdmission(message.MaxViewers, message.EndpointMediaCopyCapacity)
 	}
 	session.mu.Unlock()
 	if message.PeerAssisted {
@@ -415,6 +416,11 @@ func (session *Session) handle(message serverMessage) error {
 			return nil
 		}
 		return session.addViewer(message.PeerID)
+	case "peer-waiting":
+		if session.peerAssisted {
+			return nil
+		}
+		return session.waitViewer(message.PeerID)
 	case "peer-left":
 		if session.peerAssisted {
 			return nil
@@ -474,16 +480,15 @@ func (session *Session) addViewer(peerID string) error {
 		session.mu.Unlock()
 		return errors.New("viewer joined before host authentication completed")
 	}
-	action := session.admission.Join(peerID)
+	action := session.admission.Activate(peerID)
 	session.mu.Unlock()
 	switch action {
 	case admissionDuplicate:
 		return nil
 	case admissionWait:
-		session.emitViewerCounts()
-		return nil
+		return errors.New("active Viewer was assigned to the waiting set")
 	case admissionFull:
-		return errors.New("viewer presence exceeded the authenticated room capacity")
+		return errors.New("active Viewer exceeds authenticated endpoint or room capacity")
 	case admissionActivate:
 		if err := session.activatePeer(peerID); err != nil {
 			session.mu.Lock()
@@ -498,6 +503,29 @@ func (session *Session) addViewer(peerID string) error {
 	}
 }
 
+func (session *Session) waitViewer(peerID string) error {
+	session.mu.Lock()
+	if session.admission == nil {
+		session.mu.Unlock()
+		return errors.New("waiting Viewer arrived before host authentication completed")
+	}
+	wasActive := session.admission.IsActive(peerID)
+	action := session.admission.Wait(peerID)
+	peer := session.peers[peerID]
+	if wasActive {
+		delete(session.peers, peerID)
+	}
+	session.mu.Unlock()
+	if peer != nil && wasActive {
+		peer.close()
+	}
+	if action == admissionFull {
+		return errors.New("waiting Viewer exceeds authenticated room capacity")
+	}
+	session.emitViewerCounts()
+	return nil
+}
+
 func (session *Session) activatePeer(peerID string) error {
 	session.mu.Lock()
 	configuration, err := pionConfiguration(session.ice)
@@ -537,19 +565,11 @@ func (session *Session) removeViewer(peerID string) error {
 		}
 		return nil
 	}
-	_, promoted := session.admission.Leave(peerID)
+	session.admission.Leave(peerID)
 	session.mu.Unlock()
 	if peer != nil {
 		peer.close()
 	}
-	if promoted != "" {
-		if err := session.activatePeer(promoted); err != nil {
-			session.mu.Lock()
-			session.admission.AbortActivation(promoted)
-			session.mu.Unlock()
-			return fmt.Errorf("promote waiting viewer: %w", err)
-		}
-	}
 	session.emitViewerCounts()
 	return nil
 }
@@ -583,6 +603,9 @@ func (session *Session) rebuildPeer(peerID string) error {
 }
 
 func (session *Session) reconcileAuthoritativeChildren(peerIDs []string) error {
+	if len(peerIDs) > session.endpointMediaCopyCapacity {
+		return errors.New("authoritative children exceed the authenticated endpoint capacity")
+	}
 	desired := make(map[string]struct{}, len(peerIDs))
 	for _, peerID := range peerIDs {
 		desired[peerID] = struct{}{}
@@ -622,12 +645,22 @@ func (session *Session) handleRouteUpdate(message serverMessage) error {
 	if message.RouteRevision < session.routeRevision {
 		return nil
 	}
+	if !hostRouteAssignmentFitsCapacity(
+		message.RouteAssignment,
+		session.endpointMediaCopyCapacity,
+		message.RoutePhase == "prepare",
+	) {
+		return errors.New("route update exceeds the authenticated endpoint capacity")
+	}
 	session.routeRevision = message.RouteRevision
 	session.routePhase = message.RoutePhase
 	publicationGeneration, _ := decodeNullableOpaqueID(message.RouteAssignment.SFUPublicationGeneration)
 	if publicationGeneration != "" {
 		return session.failUnsupportedRoute(message.RouteRevision, message.RoutePhase, nil)
 	}
+	if message.RoutePhase == "prepare" && !session.hasExactChildren(message.RouteAssignment.ChildPeerIDs) {
+		return session.failUnsupportedRoute(message.RouteRevision, message.RoutePhase, nil)
+	}
 	if message.RoutePhase == "active" {
 		if err := session.reconcileAuthoritativeChildren(message.RouteAssignment.ChildPeerIDs); err != nil {
 			return err
@@ -636,6 +669,20 @@ func (session *Session) handleRouteUpdate(message serverMessage) error {
 	return session.send(map[string]any{"type": "route-ready", "revision": message.RouteRevision, "phase": message.RoutePhase})
 }
 
+func (session *Session) hasExactChildren(peerIDs []string) bool {
+	session.mu.Lock()
+	defer session.mu.Unlock()
+	if len(session.peers) != len(peerIDs) {
+		return false
+	}
+	for _, peerID := range peerIDs {
+		if session.peers[peerID] == nil {
+			return false
+		}
+	}
+	return true
+}
+
 func (session *Session) handleUnsupportedRouteMessage(message serverMessage, connectionID *string) error {
 	if !session.peerAssisted || message.RouteRevision != session.routeRevision {
 		return nil
@@ -754,13 +801,13 @@ func (session *Session) diagnosticsLoop() {
 }
 
 func (session *Session) availableSlotLocked() int {
-	used := [maxHostEdges + 1]bool{}
+	used := [maxEndpointMediaCopyCapacity + 1]bool{}
 	for _, activePeer := range session.peers {
-		if activePeer.slot > 0 && activePeer.slot <= maxHostEdges {
+		if activePeer.slot > 0 && activePeer.slot <= session.endpointMediaCopyCapacity {
 			used[activePeer.slot] = true
 		}
 	}
-	for slot := 1; slot <= maxHostEdges; slot++ {
+	for slot := 1; slot <= session.endpointMediaCopyCapacity; slot++ {
 		if !used[slot] {
 			return slot
 		}

native/sender/internal/remote/session_test.go

@@ -92,7 +92,7 @@ func TestStartRetriesTheSameProvisionalRoomOnce(t *testing.T) {
 		expiresAt := time.Now().UTC().Add(24 * time.Hour).Truncate(time.Second)
 		authenticated, _ := json.Marshal(map[string]any{
 			"type": "authenticated", "protocol": signalingProtocol, "role": "host", "peerId": "host-peer",
-			"roomExpiresAt": expiresAt.Format(time.RFC3339), "maxViewers": 3, "hostOnline": true,
+			"roomExpiresAt": expiresAt.Format(time.RFC3339), "maxViewers": 3, "endpointMediaCopyCapacity": 2, "hostOnline": true,
 			"connectionId": nil, "viewerPeerIds": []string{},
 			"iceConfig": map[string]any{"iceServers": []any{}}, "viewerPolicy": privateViewerPolicy,
 			"viewerAuthorizationGeneration": "viewer_generation_12345678",
@@ -387,7 +387,8 @@ func TestPeerAssistedHostUsesOnlyAuthoritativeRoutes(t *testing.T) {
 	defer cancel()
 	session := &Session{
 		signalCtx: ctx, conn: connection, peerAssisted: true,
-		peers: make(map[string]*peer), routeRevision: 7, routePhase: "active",
+		peers: make(map[string]*peer), endpointMediaCopyCapacity: 2,
+		routeRevision: 7, routePhase: "active",
 	}
 
 	if err = session.handle(serverMessage{Type: "peer-joined", PeerID: "viewer-new"}); err != nil || len(session.peers) != 0 {
@@ -407,6 +408,13 @@ func TestPeerAssistedHostUsesOnlyAuthoritativeRoutes(t *testing.T) {
 	case <-time.After(50 * time.Millisecond):
 	}
 
+	unsupportedPrepare := direct
+	unsupportedPrepare.ChildPeerIDs = []string{"viewer-new"}
+	if err = session.handle(serverMessage{Type: "route-update", RouteRevision: 8, RoutePhase: "prepare", RouteAssignment: unsupportedPrepare}); err != nil {
+		t.Fatal(err)
+	}
+	assertRouteStatus(t, receiveRouteStatus(t, received), "route-failed", 8, "prepare", nil)
+
 	if err = session.handle(serverMessage{Type: "route-update", RouteRevision: 8, RoutePhase: "active", RouteAssignment: direct}); err != nil {
 		t.Fatal(err)
 	}
@@ -745,7 +753,7 @@ func newRemoteHarnessWithPassword(t *testing.T, roster []string, password string
 		harness.authentication <- append([]byte(nil), payload...)
 		authenticated, _ := json.Marshal(map[string]any{
 			"type": "authenticated", "protocol": signalingProtocol, "role": "host", "peerId": "host-peer",
-			"roomExpiresAt": nil, "maxViewers": 3, "hostOnline": true,
+			"roomExpiresAt": nil, "maxViewers": 3, "endpointMediaCopyCapacity": 2, "hostOnline": true,
 			"connectionId": nil, "viewerPeerIds": roster,
 			"iceConfig":    map[string]any{"iceServers": []any{}},
 			"viewerPolicy": privateViewerPolicy, "viewerAuthorizationGeneration": "viewer_generation_12345678",

native/sender/internal/remote/wire.go

@@ -16,11 +16,12 @@ import (
 )
 
 const (
-	signalingProtocol  = "screener-v6"
-	maxProtocolViewers = 16
-	maxRouteRevision   = int64(1<<53 - 1)
-	maxSignalSDPBytes  = 48 << 10
-	maxQualityEvidence = 2 << 10
+	signalingProtocol            = "screener-v6"
+	maxProtocolViewers           = 16
+	maxEndpointMediaCopyCapacity = 3
+	maxRouteRevision             = int64(1<<53 - 1)
+	maxSignalSDPBytes            = 48 << 10
+	maxQualityEvidence           = 2 << 10
 )
 
 var opaqueIDPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{8,128}$`)
@@ -91,27 +92,28 @@ type participantRouteAssignment struct {
 }
 
 type serverMessage struct {
-	Type            string
-	Role            string
-	PeerID          string
-	FromPeerID      string
-	ViewerPeerIDs   []string
-	ConnectionID    string
-	Rebuild         bool
-	MaxViewers      int
-	IceConfig       iceConfig
-	ViewerPolicy    string
-	Payload         inboundSignalPayload
-	Code            string
-	Message         string
-	Reason          string
-	PeerAssisted    bool
-	MediaAssignment mediaAssignment
-	RouteAssignment participantRouteAssignment
-	RouteRevision   int64
-	RoutePhase      string
-	EdgeKind        string
-	NewConnectionID string
+	Type                      string
+	Role                      string
+	PeerID                    string
+	FromPeerID                string
+	ViewerPeerIDs             []string
+	ConnectionID              string
+	Rebuild                   bool
+	MaxViewers                int
+	EndpointMediaCopyCapacity int
+	IceConfig                 iceConfig
+	ViewerPolicy              string
+	Payload                   inboundSignalPayload
+	Code                      string
+	Message                   string
+	Reason                    string
+	PeerAssisted              bool
+	MediaAssignment           mediaAssignment
+	RouteAssignment           participantRouteAssignment
+	RouteRevision             int64
+	RoutePhase                string
+	EdgeKind                  string
+	NewConnectionID           string
 }
 
 func readServerMessage(ctx context.Context, conn *websocket.Conn) (serverMessage, error) {
@@ -135,7 +137,7 @@ func decodeServerMessage(payload []byte) (serverMessage, error) {
 	switch discriminator.Type {
 	case "authenticated":
 		return decodeAuthenticatedMessage(payload)
-	case "peer-joined", "peer-left":
+	case "peer-joined", "peer-waiting", "peer-left":
 		var wire struct {
 			Type   string `json:"type"`
 			PeerID string `json:"peerId"`
@@ -327,6 +329,7 @@ func decodeAuthenticatedMessage(payload []byte) (serverMessage, error) {
 		PeerID                        string          `json:"peerId"`
 		RoomExpiresAt                 json.RawMessage `json:"roomExpiresAt"`
 		MaxViewers                    *int            `json:"maxViewers"`
+		EndpointMediaCopyCapacity     *int            `json:"endpointMediaCopyCapacity"`
 		HostOnline                    *bool           `json:"hostOnline"`
 		ConnectionID                  json.RawMessage `json:"connectionId"`
 		ViewerPeerIDs                 []string        `json:"viewerPeerIds"`
@@ -343,7 +346,9 @@ func decodeAuthenticatedMessage(payload []byte) (serverMessage, error) {
 	if err := decodeStrict(payload, &wire); err != nil || wire.Type != "authenticated" ||
 		wire.Protocol != signalingProtocol ||
 		wire.Role != "host" || !validOpaqueID(wire.PeerID) || wire.MaxViewers == nil ||
-		*wire.MaxViewers < 1 || *wire.MaxViewers > maxProtocolViewers || wire.HostOnline == nil ||
+		*wire.MaxViewers < 1 || *wire.MaxViewers > maxProtocolViewers ||
+		wire.EndpointMediaCopyCapacity == nil || *wire.EndpointMediaCopyCapacity < 1 ||
+		*wire.EndpointMediaCopyCapacity > maxEndpointMediaCopyCapacity || wire.HostOnline == nil ||
 		!*wire.HostOnline || !bytes.Equal(bytes.TrimSpace(wire.ConnectionID), []byte("null")) ||
 		wire.ViewerPeerIDs == nil || len(wire.ViewerPeerIDs) > *wire.MaxViewers ||
 		(wire.ViewerPolicy != "private-link" && wire.ViewerPolicy != "public-watch") ||
@@ -382,23 +387,27 @@ func decodeAuthenticatedMessage(payload []byte) (serverMessage, error) {
 		if err != nil || strings.Join(assignment.ChildPeerIDs, "\x00") != strings.Join(routeAssignment.ChildPeerIDs, "\x00") {
 			return serverMessage{}, errors.New("host authentication response is invalid")
 		}
+		if !hostRouteAssignmentFitsCapacity(routeAssignment, *wire.EndpointMediaCopyCapacity, false) {
+			return serverMessage{}, errors.New("host authentication response is invalid")
+		}
 		if wire.RouteRevision == nil || *wire.RouteRevision < 0 || *wire.RouteRevision > maxRouteRevision || len(wire.QualitySettings) == 0 {
 			return serverMessage{}, errors.New("host authentication response is invalid")
 		}
 		routeRevision = *wire.RouteRevision
 	}
 	return serverMessage{
-		Type:            wire.Type,
-		Role:            wire.Role,
-		PeerID:          wire.PeerID,
-		MaxViewers:      *wire.MaxViewers,
-		ViewerPeerIDs:   append([]string(nil), wire.ViewerPeerIDs...),
-		IceConfig:       config,
-		ViewerPolicy:    wire.ViewerPolicy,
-		PeerAssisted:    peerAssisted,
-		MediaAssignment: assignment,
-		RouteAssignment: routeAssignment,
-		RouteRevision:   routeRevision,
+		Type:                      wire.Type,
+		Role:                      wire.Role,
+		PeerID:                    wire.PeerID,
+		MaxViewers:                *wire.MaxViewers,
+		EndpointMediaCopyCapacity: *wire.EndpointMediaCopyCapacity,
+		ViewerPeerIDs:             append([]string(nil), wire.ViewerPeerIDs...),
+		IceConfig:                 config,
+		ViewerPolicy:              wire.ViewerPolicy,
+		PeerAssisted:              peerAssisted,
+		MediaAssignment:           assignment,
+		RouteAssignment:           routeAssignment,
+		RouteRevision:             routeRevision,
 	}, nil
 }
 
@@ -433,7 +442,7 @@ func decodeNullableOpaqueID(payload []byte) (string, error) {
 }
 
 func validChildPeerIDs(peerIDs []string) bool {
-	if peerIDs == nil || len(peerIDs) > maxHostEdges {
+	if peerIDs == nil || len(peerIDs) > maxEndpointMediaCopyCapacity {
 		return false
 	}
 	seen := make(map[string]struct{}, len(peerIDs))
@@ -449,6 +458,22 @@ func validChildPeerIDs(peerIDs []string) bool {
 	return true
 }
 
+func hostRouteAssignmentFitsCapacity(assignment participantRouteAssignment, capacity int, transition bool) bool {
+	limit := capacity
+	if transition && limit < maxEndpointMediaCopyCapacity {
+		limit++
+	}
+	publicationGeneration, err := decodeNullableOpaqueID(assignment.SFUPublicationGeneration)
+	if err != nil {
+		return false
+	}
+	count := len(assignment.ChildPeerIDs)
+	if publicationGeneration != "" {
+		count++
+	}
+	return count <= limit
+}
+
 func decodeInboundSignalPayload(payload []byte) (inboundSignalPayload, error) {
 	var discriminator struct {
 		Kind string `json:"kind"`

native/sender/internal/remote/wire_test.go

@@ -6,9 +6,9 @@ import (
 	"testing"
 )
 
-const validHostAuthenticated = `{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[{"urls":["stun:example.test"]}]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}`
+const validHostAuthenticated = `{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"endpointMediaCopyCapacity":2,"hostOnline":true,"connectionId":null,"viewerPeerIds":[],"iceConfig":{"iceServers":[{"urls":["stun:example.test"]}]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678"}`
 
-const validPeerAssistedHostAuthenticated = `{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":["viewer-1","viewer-2","viewer-3"],"iceConfig":{"iceServers":[{"urls":["stun:example.test"]}]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678","mediaMode":"peer-assisted","mediaAssignment":{"parentPeerId":null,"childPeerIds":["viewer-1","viewer-2","viewer-3"]},"routeRevision":7,"routeAssignment":{"upstream":{"kind":"none"},"childPeerIds":["viewer-1","viewer-2","viewer-3"],"sfuPublicationGeneration":null},"qualitySettings":{"resolution":"1080p","maxFramerate":60,"maxBitrate":8000000,"degradationPreference":"maintain-resolution","videoCodec":"automatic"}}`
+const validPeerAssistedHostAuthenticated = `{"type":"authenticated","protocol":"screener-v6","role":"host","peerId":"host-peer","roomExpiresAt":null,"maxViewers":3,"endpointMediaCopyCapacity":3,"hostOnline":true,"connectionId":null,"viewerPeerIds":["viewer-1","viewer-2","viewer-3"],"iceConfig":{"iceServers":[{"urls":["stun:example.test"]}]},"viewerPolicy":"private-link","viewerAuthorizationGeneration":"viewer_generation_12345678","mediaMode":"peer-assisted","mediaAssignment":{"parentPeerId":null,"childPeerIds":["viewer-1","viewer-2","viewer-3"]},"routeRevision":7,"routeAssignment":{"upstream":{"kind":"none"},"childPeerIds":["viewer-1","viewer-2","viewer-3"],"sfuPublicationGeneration":null},"qualitySettings":{"resolution":"1080p","maxFramerate":60,"maxBitrate":8000000,"degradationPreference":"maintain-resolution","videoCodec":"automatic"}}`
 
 func TestMarshalHostSignalShapes(t *testing.T) {
 	mid := "0"
@@ -58,7 +58,7 @@ func TestDecodeOrdinaryHostAuthentication(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if message.MaxViewers != 3 || message.Type != "authenticated" || message.Role != "host" ||
+	if message.MaxViewers != 3 || message.EndpointMediaCopyCapacity != 2 || message.Type != "authenticated" || message.Role != "host" ||
 		message.ViewerPolicy != "private-link" {
 		t.Fatalf("authentication = %+v", message)
 	}
@@ -138,13 +138,16 @@ func TestDecodePeerAssistedRouteMessages(t *testing.T) {
 
 func TestDecodeOrdinaryHostAuthenticationRejectsInvalidVariants(t *testing.T) {
 	tests := map[string]func(map[string]any){
-		"missing-protocol": func(message map[string]any) { delete(message, "protocol") },
-		"wrong-protocol":   func(message map[string]any) { message["protocol"] = "screener-v5" },
-		"missing-max":      func(message map[string]any) { delete(message, "maxViewers") },
-		"zero-max":         func(message map[string]any) { message["maxViewers"] = 0 },
-		"too-large-max":    func(message map[string]any) { message["maxViewers"] = 17 },
-		"peer-mode-extra":  func(message map[string]any) { message["mediaMode"] = "peer-assisted" },
-		"wrong-role":       func(message map[string]any) { message["role"] = "viewer" },
+		"missing-protocol":       func(message map[string]any) { delete(message, "protocol") },
+		"wrong-protocol":         func(message map[string]any) { message["protocol"] = "screener-v5" },
+		"missing-max":            func(message map[string]any) { delete(message, "maxViewers") },
+		"zero-max":               func(message map[string]any) { message["maxViewers"] = 0 },
+		"too-large-max":          func(message map[string]any) { message["maxViewers"] = 17 },
+		"missing-endpoint-cap":   func(message map[string]any) { delete(message, "endpointMediaCopyCapacity") },
+		"zero-endpoint-cap":      func(message map[string]any) { message["endpointMediaCopyCapacity"] = 0 },
+		"too-large-endpoint-cap": func(message map[string]any) { message["endpointMediaCopyCapacity"] = 4 },
+		"peer-mode-extra":        func(message map[string]any) { message["mediaMode"] = "peer-assisted" },
+		"wrong-role":             func(message map[string]any) { message["role"] = "viewer" },
 		"legacy-ice-shape": func(message map[string]any) {
 			message["iceConfig"].(map[string]any)["expiresAt"] = nil
 		},

src/client/pages/HostPage.tsx

@@ -1414,6 +1414,12 @@ export function HostPage({ onAuthorizationRequired }: HostPageProps = {}) {
       });
       return;
     }
+    if (message.type === "peer-waiting") {
+      if (!peerAssistedRef.current) {
+        removePeer(message.peerId);
+      }
+      return;
+    }
     if (message.type === "peer-left") {
       if (peerAssistedRef.current) {
         return;

src/server/room-store.ts

@@ -563,6 +563,13 @@ export class RoomStore {
     return viewer?.sessionId ? { peerId, sessionId: viewer.sessionId } : undefined;
   }
 
+  getViewerPeerIds(roomId: string): string[] {
+    const room = this.rooms.get(roomId);
+    return room
+      ? [...room.viewers.values()].map((viewer) => viewer.peerId)
+      : [];
+  }
+
   getConnectedViewers(roomId: string): ConnectedPeer[] {
     const room = this.rooms.get(roomId);
     if (!room) {

src/server/signaling.ts

@@ -21,6 +21,7 @@ import {
   type ServerMessage,
   type ParticipantPresenceEntry,
 } from "../shared/protocol.js";
+import { assertEndpointMediaCopyCapacity } from "../shared/media-copy-accounting.js";
 import {
   RoomStore,
   RoomStoreError,
@@ -123,6 +124,10 @@ export class SignalingServer {
   private readonly shareGenerationsByRoom = new Map<string, string>();
   private readonly pausedShareGenerationsByRoom = new Map<string, string>();
   private readonly deferredViewerPresenceRooms = new Set<string>();
+  private readonly ordinaryActiveHostChildrenByRoom = new Map<
+    string,
+    Set<string>
+  >();
   private readonly hybridMediaRouter?: HybridMediaRouter;
   private readonly now: () => number;
   private readonly authenticationTimeoutMs: number;
@@ -136,6 +141,7 @@ export class SignalingServer {
   private closing = false;
 
   constructor(private readonly options: SignalingOptions) {
+    assertEndpointMediaCopyCapacity(options.endpointMediaCopyCapacity);
     this.now = options.now ?? Date.now;
     this.authenticationTimeoutMs = options.authenticationTimeoutMs ?? 5_000;
     this.viewerDisconnectGraceMs = options.viewerDisconnectGraceMs ?? 5_000;
@@ -255,6 +261,7 @@ export class SignalingServer {
     this.viewerMediaReadyByRoom.clear();
     this.shareGenerationsByRoom.clear();
     this.pausedShareGenerationsByRoom.clear();
+    this.ordinaryActiveHostChildrenByRoom.clear();
     this.options.server.off("upgrade", this.upgradeHandler);
 
     for (const socket of this.webSocketServer.clients) {
@@ -531,6 +538,9 @@ export class SignalingServer {
       participant.role === "host"
         ? this.options.roomStore.getConnectedViewers(participant.roomId)
         : [];
+    if (!this.isHybridMediaEnabled()) {
+      this.fillOrdinaryHostChildren(participant.roomId);
+    }
 
     const authenticatedMessage = {
       type: "authenticated" as const,
@@ -539,6 +549,7 @@ export class SignalingServer {
       peerId: participant.peerId,
       roomExpiresAt: participant.expiresAt,
       maxViewers: this.options.roomStore.maxViewersPerRoom,
+      endpointMediaCopyCapacity: this.options.endpointMediaCopyCapacity,
       hostOnline: participant.hostOnline,
       ...(participant.role === "viewer"
         ? {
@@ -549,7 +560,10 @@ export class SignalingServer {
           }
         : {}),
       connectionId,
-      viewerPeerIds: [...participant.viewerPeerIds],
+      viewerPeerIds:
+        participant.role === "host" && !this.isHybridMediaEnabled()
+          ? [...this.ordinaryHostChildPeerIds(participant.roomId)]
+          : [...participant.viewerPeerIds],
       iceConfig: this.iceConfig(),
       viewerPolicy: participant.viewerPolicy,
       viewerAuthorizationGeneration:
@@ -606,8 +620,18 @@ export class SignalingServer {
             this.pausedShareGenerationsByRoom.get(participant.roomId) ===
             shareGeneration,
         });
-        if (!this.isHybridMediaEnabled()) {
-          this.send(socket, { type: "peer-joined", peerId: viewer.peerId });
+        if (
+          !this.isHybridMediaEnabled() &&
+          !this.isOrdinaryHostChild(participant.roomId, viewer.peerId)
+        ) {
+          this.send(socket, { type: "peer-waiting", peerId: viewer.peerId });
+        }
+      }
+      if (!this.isHybridMediaEnabled()) {
+        for (const viewer of connectedViewers) {
+          if (this.isOrdinaryHostChild(participant.roomId, viewer.peerId)) {
+            this.send(socket, { type: "peer-joined", peerId: viewer.peerId });
+          }
         }
       }
       return;
@@ -625,7 +649,9 @@ export class SignalingServer {
     // event idempotently, which closes races where its local peer was rebuilt
     // while the viewer retained the same server-side connection ID.
     this.sendToSession(host.sessionId, {
-      type: "peer-joined",
+      type: this.isOrdinaryHostChild(participant.roomId, participant.peerId)
+        ? "peer-joined"
+        : "peer-waiting",
       peerId: participant.peerId,
     });
   }
@@ -675,6 +701,12 @@ export class SignalingServer {
           this.sendError(socket, "FORBIDDEN", "P2P restart requests target the host implicitly");
           return;
         }
+        if (
+          !this.isOrdinaryHostChild(authenticated.roomId, authenticated.peerId)
+        ) {
+          this.sendError(socket, "FORBIDDEN", "Viewer has no active Host edge");
+          return;
+        }
         this.routeToHost(socket, authenticated, {
           type: "restart-request",
           fromPeerId: authenticated.peerId,
@@ -954,6 +986,7 @@ export class SignalingServer {
     if (update.revokedViewers.length > 0) {
       this.clearRoomGraceTimers(authenticated.roomId);
       this.clearRoomConnectionIds(authenticated.roomId);
+      this.ordinaryActiveHostChildrenByRoom.delete(authenticated.roomId);
 
       this.deferredViewerPresenceRooms.add(authenticated.roomId);
       try {
@@ -1257,6 +1290,10 @@ export class SignalingServer {
         this.sendError(sourceSocket, "PEER_NOT_FOUND", "Viewer is not connected");
         return;
       }
+      if (!this.isOrdinaryHostChild(source.roomId, viewer.peerId)) {
+        this.sendError(sourceSocket, "FORBIDDEN", "Viewer has no active Host edge");
+        return;
+      }
       if (description?.type === "offer") {
         this.setViewerConnectionId(
           source.roomId,
@@ -1276,6 +1313,10 @@ export class SignalingServer {
       this.sendError(sourceSocket, "FORBIDDEN", "Viewer signals may only reply to the host");
       return;
     }
+    if (!this.isOrdinaryHostChild(source.roomId, source.peerId)) {
+      this.sendError(sourceSocket, "FORBIDDEN", "Viewer has no active Host edge");
+      return;
+    }
     this.routeToHost(sourceSocket, source, {
       type: "signal",
       fromPeerId: source.peerId,
@@ -1488,12 +1529,25 @@ export class SignalingServer {
           );
           return;
         }
+        this.releaseOrdinaryHostChild(disconnected.roomId, disconnected.peerId);
+        const promotedPeerIds = this.fillOrdinaryHostChildren(
+          disconnected.roomId,
+        );
         const host = this.options.roomStore.getConnectedHost(disconnected.roomId);
         if (host) {
           this.sendToSession(host.sessionId, {
             type: "peer-left",
             peerId: disconnected.peerId,
           });
+          for (const peerId of promotedPeerIds) {
+            this.sendToSession(host.sessionId, {
+              type: "peer-joined",
+              peerId,
+            });
+          }
+        }
+        if (promotedPeerIds.length > 0) {
+          this.sendViewerPresence(disconnected.roomId);
         }
       }
     }, this.viewerDisconnectGraceMs);
@@ -1524,6 +1578,7 @@ export class SignalingServer {
     }
     this.clearRoomGraceTimers(roomId);
     this.clearRoomConnectionIds(roomId);
+    this.ordinaryActiveHostChildrenByRoom.delete(roomId);
     if (this.isHybridMediaEnabled()) {
       this.hybridMediaRouter!.deleteRoom(roomId);
     }
@@ -1544,6 +1599,7 @@ export class SignalingServer {
     for (const expired of this.options.roomStore.expireRooms(this.now())) {
       this.clearRoomGraceTimers(expired.roomId);
       this.clearRoomConnectionIds(expired.roomId);
+      this.ordinaryActiveHostChildrenByRoom.delete(expired.roomId);
       if (this.isHybridMediaEnabled()) {
         this.hybridMediaRouter!.deleteRoom(expired.roomId);
       }
@@ -1603,6 +1659,60 @@ export class SignalingServer {
     return this.hybridMediaRouter !== undefined;
   }
 
+  private ordinaryHostChildPeerIds(roomId: string): readonly string[] {
+    const active = this.ordinaryActiveHostChildren(roomId);
+    return this.options.roomStore
+      .getViewerPeerIds(roomId)
+      .filter((peerId) => active.has(peerId));
+  }
+
+  private isOrdinaryHostChild(roomId: string, peerId: string): boolean {
+    return this.ordinaryActiveHostChildren(roomId).has(peerId);
+  }
+
+  private ordinaryActiveHostChildren(roomId: string): Set<string> {
+    let active = this.ordinaryActiveHostChildrenByRoom.get(roomId);
+    if (!active) {
+      active = new Set<string>();
+      this.ordinaryActiveHostChildrenByRoom.set(roomId, active);
+      return active;
+    }
+    const members = new Set(this.options.roomStore.getViewerPeerIds(roomId));
+    for (const peerId of active) {
+      if (!members.has(peerId)) {
+        active.delete(peerId);
+      }
+    }
+    return active;
+  }
+
+  private fillOrdinaryHostChildren(roomId: string): string[] {
+    const active = this.ordinaryActiveHostChildren(roomId);
+    const promotedPeerIds: string[] = [];
+    for (const viewer of this.options.roomStore.getConnectedViewers(roomId)) {
+      if (active.size >= this.options.endpointMediaCopyCapacity) {
+        break;
+      }
+      if (active.has(viewer.peerId)) {
+        continue;
+      }
+      active.add(viewer.peerId);
+      promotedPeerIds.push(viewer.peerId);
+    }
+    return promotedPeerIds;
+  }
+
+  private releaseOrdinaryHostChild(roomId: string, peerId: string): void {
+    const active = this.ordinaryActiveHostChildrenByRoom.get(roomId);
+    if (!active) {
+      return;
+    }
+    active.delete(peerId);
+    if (active.size === 0) {
+      this.ordinaryActiveHostChildrenByRoom.delete(roomId);
+    }
+  }
+
   private hasConnectionCapacity(): boolean {
     return (
       !this.closing &&
@@ -1690,7 +1800,7 @@ export class SignalingServer {
       }
       const upstream = this.isHybridMediaEnabled()
         ? this.hybridMediaRouter!.getViewerRouteUpstream(roomId, viewer.peerId)
-        : host
+        : host && this.isOrdinaryHostChild(roomId, viewer.peerId)
           ? { kind: "peer" as const, peerId: host.peerId }
           : { kind: "none" as const };
       const ready = readyByViewer?.get(viewer.peerId);

src/shared/protocol.ts

@@ -661,6 +661,11 @@ const authenticatedMessageShape = {
   peerId: opaqueIdSchema,
   roomExpiresAt: z.string().datetime().nullable(),
   maxViewers: z.number().int().min(1).max(MAX_VIEWERS_PER_ROOM_LIMIT),
+  endpointMediaCopyCapacity: z
+    .number()
+    .int()
+    .min(1)
+    .max(MAX_ENDPOINT_MEDIA_COPY_CAPACITY),
   hostOnline: z.boolean(),
   hostPaused: z.boolean().optional(),
   connectionId: opaqueIdSchema.nullable(),
@@ -699,6 +704,12 @@ export const serverMessageSchema = z.union([
       peerId: opaqueIdSchema,
     })
     .strict(),
+  z
+    .object({
+      type: z.literal("peer-waiting"),
+      peerId: opaqueIdSchema,
+    })
+    .strict(),
   z
     .object({
       type: z.literal("peer-left"),

tests/client-state.test.ts

@@ -645,6 +645,7 @@ describe("client signaling recovery policy", () => {
         peerId: "viewer_12345678",
         roomExpiresAt: null,
         maxViewers: 8,
+        endpointMediaCopyCapacity: 2,
         hostOnline: true,
         connectionId: null,
         viewerPeerIds: [],
@@ -718,6 +719,7 @@ describe("client signaling recovery policy", () => {
         peerId: "host_12345678",
         roomExpiresAt: null,
         maxViewers: 8,
+        endpointMediaCopyCapacity: 2,
         hostOnline: true,
         connectionId: null,
         viewerPeerIds: [],
@@ -842,6 +844,7 @@ describe("client signaling recovery policy", () => {
       peerId: "viewer_12345678",
       roomExpiresAt: null,
       maxViewers: 8,
+      endpointMediaCopyCapacity: 2,
       hostOnline: true,
       connectionId: "connection_12345678",
       viewerPeerIds: [],

tests/protocol.test.ts

@@ -85,8 +85,8 @@ describe("client signaling protocol", () => {
       "utf8",
     );
 
-    expect(nativeWire).toContain(
-      `signalingProtocol  = "${SIGNALING_PROTOCOL}"`,
+    expect(nativeWire).toMatch(
+      new RegExp(`signalingProtocol\\s*=\\s*"${SIGNALING_PROTOCOL}"`),
     );
   });
 
@@ -848,6 +848,7 @@ describe("server signaling protocol", () => {
       peerId: "host_12345678",
       roomExpiresAt: "2026-08-18T18:00:00.000Z",
       maxViewers,
+      endpointMediaCopyCapacity: 2,
       hostOnline: true,
       connectionId: null,
       viewerPeerIds,
@@ -873,6 +874,46 @@ describe("server signaling protocol", () => {
     ).toBe(true);
   });
 
+  it("requires the authenticated endpoint capacity to be 1, 2, or 3", () => {
+    for (const endpointMediaCopyCapacity of [1, 2, 3]) {
+      expect(
+        serverMessageSchema.safeParse({
+          ...authenticatedMessage(8),
+          endpointMediaCopyCapacity,
+        }).success,
+      ).toBe(true);
+    }
+    for (const endpointMediaCopyCapacity of [undefined, 0, 4, 1.5]) {
+      const message = authenticatedMessage(8) as Record<string, unknown>;
+      if (endpointMediaCopyCapacity === undefined) {
+        delete message.endpointMediaCopyCapacity;
+      } else {
+        message.endpointMediaCopyCapacity = endpointMediaCopyCapacity;
+      }
+      expect(serverMessageSchema.safeParse(message).success).toBe(false);
+    }
+  });
+
+  it("accepts only a strict peer-waiting notification", () => {
+    expect(
+      serverMessageSchema.safeParse({
+        type: "peer-waiting",
+        peerId: "viewer_12345678",
+      }).success,
+    ).toBe(true);
+    expect(
+      serverMessageSchema.safeParse({
+        type: "peer-waiting",
+        peerId: "viewer_12345678",
+        active: false,
+      }).success,
+    ).toBe(false);
+    expect(
+      serverMessageSchema.safeParse({ type: "peer-waiting", peerId: "short" })
+        .success,
+    ).toBe(false);
+  });
+
   it("accepts a strict, unique and bounded Viewer presence snapshot", () => {
     const viewer = {
       role: "viewer",

tests/server-signal.test.ts

@@ -9231,6 +9231,258 @@ describe("WebSocket signaling", () => {
     expect((await host.inbox.next("peer-left", 500)).peerId).toBe(firstAuth.peerId);
   });
 
+  it.each([1, 2, 3] as const)(
+    "authorizes exactly %i ordinary Host children independently of room admission",
+    async (endpointMediaCopyCapacity) => {
+      const maxViewersPerRoom = endpointMediaCopyCapacity + 1;
+      const harness = await startHarness({
+        endpointMediaCopyCapacity,
+        maxViewersPerRoom,
+      });
+      const viewers: TestClient[] = [];
+      const viewerPeerIds: string[] = [];
+      for (let index = 0; index < maxViewersPerRoom; index += 1) {
+        const viewer = await openClient(harness.webSocketUrl);
+        const authenticated = await authenticate(
+          viewer,
+          harness.room,
+          "viewer",
+          `ordinary-cap-viewer-${index}`,
+          1,
+          undefined,
+          { viewerPresence: true },
+        );
+        expect(authenticated.endpointMediaCopyCapacity).toBe(
+          endpointMediaCopyCapacity,
+        );
+        viewers.push(viewer);
+        viewerPeerIds.push(authenticated.peerId);
+      }
+
+      const host = await openClient(harness.webSocketUrl);
+      const hostAuth = await authenticate(
+        host,
+        harness.room,
+        "host",
+        "ordinary-cap-host",
+        1,
+        undefined,
+        { viewerPresence: true },
+      );
+      expect(hostAuth).toMatchObject({
+        maxViewers: maxViewersPerRoom,
+        endpointMediaCopyCapacity,
+        viewerPeerIds: viewerPeerIds.slice(0, endpointMediaCopyCapacity),
+      });
+
+      const presence = viewerPresenceEntries(
+        await host.inbox.next("viewer-presence"),
+      );
+      for (let index = 0; index < presence.length; index += 1) {
+        expect(presence[index]).toMatchObject({
+          peerId: viewerPeerIds[index],
+          upstream:
+            index < endpointMediaCopyCapacity
+              ? { kind: "peer", peerId: hostAuth.peerId }
+              : { kind: "none" },
+        });
+      }
+
+      const waitingPeerId = viewerPeerIds.at(-1)!;
+      expect((await host.inbox.next("peer-waiting")).peerId).toBe(
+        waitingPeerId,
+      );
+      for (const activePeerId of viewerPeerIds.slice(
+        0,
+        endpointMediaCopyCapacity,
+      )) {
+        expect((await host.inbox.next("peer-joined")).peerId).toBe(
+          activePeerId,
+        );
+      }
+
+      host.socket.send(
+        JSON.stringify({
+          type: "signal",
+          targetPeerId: waitingPeerId,
+          payload: {
+            kind: "description",
+            connectionId: "waiting-host-offer",
+            description: { type: "offer", sdp: "v=0\r\n" },
+          },
+        }),
+      );
+      expect((await host.inbox.next("error")).code).toBe("FORBIDDEN");
+
+      host.socket.send(
+        JSON.stringify({
+          type: "signal",
+          targetPeerId: waitingPeerId,
+          payload: {
+            kind: "candidate",
+            connectionId: "waiting-host-candidate",
+            candidate: null,
+          },
+        }),
+      );
+      expect((await host.inbox.next("error")).code).toBe("FORBIDDEN");
+
+      const waitingViewer = viewers.at(-1)!;
+      waitingViewer.socket.send(
+        JSON.stringify({
+          type: "signal",
+          payload: {
+            kind: "description",
+            connectionId: "waiting-viewer-answer",
+            description: { type: "answer", sdp: "v=0\r\n" },
+          },
+        }),
+      );
+      expect((await waitingViewer.inbox.next("error")).code).toBe("FORBIDDEN");
+
+      waitingViewer.socket.send(
+        JSON.stringify({
+          type: "signal",
+          payload: {
+            kind: "candidate",
+            connectionId: "waiting-viewer-candidate",
+            candidate: null,
+          },
+        }),
+      );
+      expect((await waitingViewer.inbox.next("error")).code).toBe("FORBIDDEN");
+
+      waitingViewer.socket.send(
+        JSON.stringify({
+          type: "restart-request",
+          connectionId: "waiting-viewer-restart",
+          rebuild: true,
+        }),
+      );
+      expect((await waitingViewer.inbox.next("error")).code).toBe("FORBIDDEN");
+
+      host.socket.send(
+        JSON.stringify({
+          type: "signal",
+          targetPeerId: viewerPeerIds[0],
+          payload: {
+            kind: "description",
+            connectionId: "active-host-offer",
+            description: { type: "offer", sdp: "v=0\r\n" },
+          },
+        }),
+      );
+      expect(await viewers[0]!.inbox.next("signal")).toMatchObject({
+        fromPeerId: hostAuth.peerId,
+        payload: { connectionId: "active-host-offer" },
+      });
+    },
+  );
+
+  it("holds an active ordinary slot through grace and skips offline waiters", async () => {
+    const harness = await startHarness({
+      endpointMediaCopyCapacity: 1,
+      maxViewersPerRoom: 3,
+      viewerDisconnectGraceMs: 100,
+    });
+    const host = await openClient(harness.webSocketUrl);
+    const hostAuth = await authenticate(
+      host,
+      harness.room,
+      "host",
+      "ordinary-grace-host",
+      1,
+      undefined,
+      { viewerPresence: true },
+    );
+    expect(
+      viewerPresenceEntries(await host.inbox.next("viewer-presence")),
+    ).toEqual([]);
+
+    const activeViewer = await openClient(harness.webSocketUrl);
+    const activeAuth = await authenticate(
+      activeViewer,
+      harness.room,
+      "viewer",
+      "ordinary-grace-active",
+    );
+    expect((await host.inbox.next("peer-joined")).peerId).toBe(
+      activeAuth.peerId,
+    );
+    expect(
+      viewerPresenceEntries(await host.inbox.next("viewer-presence")),
+    ).toEqual([
+      expect.objectContaining({
+        peerId: activeAuth.peerId,
+        upstream: { kind: "peer", peerId: hostAuth.peerId },
+      }),
+    ]);
+
+    const offlineWaitingViewer = await openClient(harness.webSocketUrl);
+    const offlineWaitingAuth = await authenticate(
+      offlineWaitingViewer,
+      harness.room,
+      "viewer",
+      "ordinary-grace-offline-waiting",
+    );
+    expect((await host.inbox.next("peer-waiting")).peerId).toBe(
+      offlineWaitingAuth.peerId,
+    );
+    await host.inbox.next("viewer-presence");
+
+    const connectedWaitingViewer = await openClient(harness.webSocketUrl);
+    const connectedWaitingAuth = await authenticate(
+      connectedWaitingViewer,
+      harness.room,
+      "viewer",
+      "ordinary-grace-connected-waiting",
+    );
+    expect((await host.inbox.next("peer-waiting")).peerId).toBe(
+      connectedWaitingAuth.peerId,
+    );
+    await host.inbox.next("viewer-presence");
+
+    await closeClient(activeViewer);
+    await host.inbox.next("viewer-presence");
+    await new Promise((resolve) => setTimeout(resolve, 20));
+    await closeClient(offlineWaitingViewer);
+    await host.inbox.next("viewer-presence");
+    await host.inbox.expectNone(40);
+
+    expect((await host.inbox.next("peer-left", 300)).peerId).toBe(
+      activeAuth.peerId,
+    );
+    expect((await host.inbox.next("peer-joined", 300)).peerId).toBe(
+      connectedWaitingAuth.peerId,
+    );
+    expect(
+      viewerPresenceEntries(await host.inbox.next("viewer-presence")),
+    ).toEqual([
+      expect.objectContaining({
+        peerId: connectedWaitingAuth.peerId,
+        upstream: { kind: "peer", peerId: hostAuth.peerId },
+      }),
+    ]);
+
+    host.socket.send(
+      JSON.stringify({
+        type: "signal",
+        targetPeerId: connectedWaitingAuth.peerId,
+        payload: {
+          kind: "description",
+          connectionId: "promoted-host-offer",
+          description: { type: "offer", sdp: "v=0\r\n" },
+        },
+      }),
+    );
+    expect(await connectedWaitingViewer.inbox.next("signal")).toMatchObject({
+      payload: { connectionId: "promoted-host-offer" },
+    });
+    expect((await host.inbox.next("peer-left", 300)).peerId).toBe(
+      offlineWaitingAuth.peerId,
+    );
+  });
+
   it("clears the previous media generation when sharing stops", async () => {
     const harness = await startHarness({ viewerDisconnectGraceMs: 300 });
     const host = await openClient(harness.webSocketUrl);
@@ -9636,7 +9888,7 @@ describe("WebSocket signaling", () => {
         `viewer-client-${index}`,
       );
       expect(viewerAuth.maxViewers).toBe(maxViewersPerRoom);
-      await host.inbox.next("peer-joined");
+      await host.inbox.next(index <= 2 ? "peer-joined" : "peer-waiting");
       viewers.push(viewer);
     }