diff --git a/native/sender/internal/app/server_test.go b/native/sender/internal/app/server_test.go index 2504389..e016275 100644 --- a/native/sender/internal/app/server_test.go +++ b/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 } diff --git a/native/sender/internal/remote/admission.go b/native/sender/internal/remote/admission.go index beb6644..90684ee 100644 --- a/native/sender/internal/remote/admission.go +++ b/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 -} diff --git a/native/sender/internal/remote/admission_test.go b/native/sender/internal/remote/admission_test.go index 30c8763..e21d604 100644 --- a/native/sender/internal/remote/admission_test.go +++ b/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) } } diff --git a/native/sender/internal/remote/session.go b/native/sender/internal/remote/session.go index 03946e3..6ca6d35 100644 --- a/native/sender/internal/remote/session.go +++ b/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 } diff --git a/native/sender/internal/remote/session_test.go b/native/sender/internal/remote/session_test.go index 34cb3d2..4f5cc32 100644 --- a/native/sender/internal/remote/session_test.go +++ b/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", diff --git a/native/sender/internal/remote/wire.go b/native/sender/internal/remote/wire.go index 5492911..ef6242b 100644 --- a/native/sender/internal/remote/wire.go +++ b/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"` diff --git a/native/sender/internal/remote/wire_test.go b/native/sender/internal/remote/wire_test.go index 87efc06..a0c3d2f 100644 --- a/native/sender/internal/remote/wire_test.go +++ b/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 }, diff --git a/src/client/pages/HostPage.tsx b/src/client/pages/HostPage.tsx index 00b6a73..97c10ad 100644 --- a/src/client/pages/HostPage.tsx +++ b/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; diff --git a/src/server/room-store.ts b/src/server/room-store.ts index a71bed4..26ca58d 100644 --- a/src/server/room-store.ts +++ b/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) { diff --git a/src/server/signaling.ts b/src/server/signaling.ts index a3295e2..77c85c3 100644 --- a/src/server/signaling.ts +++ b/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(); private readonly pausedShareGenerationsByRoom = new Map(); private readonly deferredViewerPresenceRooms = new Set(); + private readonly ordinaryActiveHostChildrenByRoom = new Map< + string, + Set + >(); 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 { + let active = this.ordinaryActiveHostChildrenByRoom.get(roomId); + if (!active) { + active = new Set(); + 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); diff --git a/src/shared/protocol.ts b/src/shared/protocol.ts index 332a78b..a67e46c 100644 --- a/src/shared/protocol.ts +++ b/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"), diff --git a/tests/client-state.test.ts b/tests/client-state.test.ts index 8a7c891..553ecc8 100644 --- a/tests/client-state.test.ts +++ b/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: [], diff --git a/tests/protocol.test.ts b/tests/protocol.test.ts index c3a36fc..16fa78d 100644 --- a/tests/protocol.test.ts +++ b/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; + 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", diff --git a/tests/server-signal.test.ts b/tests/server-signal.test.ts index 4bd4f16..8159a57 100644 --- a/tests/server-signal.test.ts +++ b/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); }