From b656f479519a100b6e079fac8539e0c386bba24a Mon Sep 17 00:00:00 2001 From: AAA <35992542+TNTcraftHIM@users.noreply.github.com> Date: Sun, 23 Aug 2026 14:12:54 +0800 Subject: [PATCH] refactor(route): account reusable SFU subscriptions --- src/server/sfu-resource-admission.ts | 346 +++++++++++++++++++-------- tests/sfu-resource-admission.test.ts | 276 ++++++++++++++------- 2 files changed, 443 insertions(+), 179 deletions(-) diff --git a/src/server/sfu-resource-admission.ts b/src/server/sfu-resource-admission.ts index e365aaa..5349d68 100644 --- a/src/server/sfu-resource-admission.ts +++ b/src/server/sfu-resource-admission.ts @@ -4,6 +4,10 @@ export interface SfuResourceFence { publicationGeneration: string; } +export interface SfuSubscriptionFence extends SfuResourceFence { + viewerPeerId: string; +} + export interface SfuResourceAdmissionOptions { readonly ingressCapacity: number; readonly egressCapacity: number; @@ -14,19 +18,27 @@ export interface SfuResourceUsage { egress: number; } -interface SfuResourceEntry { - fence: SfuResourceFence; - egress: number; +type ResourceState = "reserved" | "committed" | "draining"; + +interface SubscriptionEntry { + fence: SfuSubscriptionFence; + state: ResourceState; + reservedFromDraining: boolean; } -interface RoomSfuResources { - committed?: SfuResourceEntry; - reserved?: SfuResourceEntry; - draining: Map; +interface PublicationEntry { + fence: SfuResourceFence; + state: ResourceState; + subscriptions: Map; + legacyEgress?: number; +} + +interface RoomResources { + publications: Map; } export class SfuResourceAdmission { - private readonly rooms = new Map(); + private readonly rooms = new Map(); private readonly ingressCapacity: number; private readonly egressCapacity: number; private ingressInUse = 0; @@ -39,110 +51,220 @@ export class SfuResourceAdmission { this.egressCapacity = options.egressCapacity; } - reserve(fence: SfuResourceFence, egress: number): boolean { + reservePublication(fence: SfuResourceFence): boolean { assertFence(fence); - assertPositiveSafeInteger(egress, "SFU egress reservation"); const room = this.rooms.get(fence.roomId); - if (room?.reserved) { - return sameEntry(room.reserved, fence, egress); - } - if (room?.committed && sameFence(room.committed.fence, fence)) { - return room.committed.egress === egress; - } - if (room?.draining.has(fenceKey(fence))) { - return false; + const existing = room?.publications.get(publicationKey(fence)); + if (existing) { + return existing.state !== "draining" && sameFence(existing.fence, fence); } if ( this.ingressInUse >= this.ingressCapacity || - egress > this.egressCapacity - this.egressInUse + [...(room?.publications.values() ?? [])].some( + (publication) => publication.state === "reserved", + ) ) { return false; } - const resources: RoomSfuResources = room ?? { draining: new Map() }; - resources.reserved = { fence: { ...fence }, egress }; + const resources = room ?? { publications: new Map() }; + resources.publications.set(publicationKey(fence), { + fence: cloneFence(fence), + state: "reserved", + subscriptions: new Map(), + }); + this.rooms.set(fence.roomId, resources); + this.ingressInUse += 1; + return true; + } + + reserveSubscription(fence: SfuSubscriptionFence): boolean { + assertSubscriptionFence(fence); + const publication = this.publication(fence); + if (!publication || publication.state === "draining") { + return false; + } + const existing = publication.subscriptions.get(fence.viewerPeerId); + if (existing) { + if (!sameSubscriptionFence(existing.fence, fence)) return false; + if (existing.state === "reserved" || existing.state === "committed") { + return true; + } + if (publication.state !== "committed") return false; + existing.state = "reserved"; + existing.reservedFromDraining = true; + return true; + } + if (this.egressInUse >= this.egressCapacity) return false; + + publication.subscriptions.set(fence.viewerPeerId, { + fence: cloneSubscriptionFence(fence), + state: "reserved", + reservedFromDraining: false, + }); + this.egressInUse += 1; + return true; + } + + commitPublication(fence: SfuResourceFence): readonly SfuResourceFence[] | null { + assertFence(fence); + const room = this.rooms.get(fence.roomId); + const publication = room?.publications.get(publicationKey(fence)); + if (!room || !publication || !sameFence(publication.fence, fence)) { + return null; + } + if (publication.state === "committed") return []; + if (publication.state !== "reserved") return null; + + const draining: SfuResourceFence[] = []; + for (const other of room.publications.values()) { + if (other === publication || other.state !== "committed") continue; + this.markDraining(other); + draining.push(cloneFence(other.fence)); + } + publication.state = "committed"; + for (const subscription of publication.subscriptions.values()) { + if (subscription.state === "reserved") { + subscription.state = "committed"; + subscription.reservedFromDraining = false; + } + } + return draining; + } + + commitSubscription(fence: SfuSubscriptionFence): boolean { + assertSubscriptionFence(fence); + const publication = this.publication(fence); + const subscription = publication?.subscriptions.get(fence.viewerPeerId); + if ( + publication?.state !== "committed" || + !subscription || + !sameSubscriptionFence(subscription.fence, fence) + ) { + return false; + } + if (subscription.state === "committed") return true; + if (subscription.state !== "reserved") return false; + subscription.state = "committed"; + subscription.reservedFromDraining = false; + return true; + } + + releaseSubscription(fence: SfuSubscriptionFence): boolean { + assertSubscriptionFence(fence); + const publication = this.publication(fence); + const subscription = publication?.subscriptions.get(fence.viewerPeerId); + if ( + !publication || + !subscription || + !sameSubscriptionFence(subscription.fence, fence) + ) { + return false; + } + if (subscription.state === "reserved") { + if (subscription.reservedFromDraining) { + subscription.state = "draining"; + subscription.reservedFromDraining = false; + } else { + publication.subscriptions.delete(fence.viewerPeerId); + this.subtractEgress(1); + } + return true; + } + if (subscription.state === "committed") { + subscription.state = "draining"; + } + return true; + } + + /** Compatibility wrapper for the pre-controller aggregate reservation. */ + reserve(fence: SfuResourceFence, egress: number): boolean { + assertFence(fence); + assertPositiveSafeInteger(egress, "SFU egress reservation"); + const existing = this.publication(fence); + if (existing) { + return ( + existing.state !== "draining" && existing.legacyEgress === egress + ); + } + const room = this.rooms.get(fence.roomId); + if ( + this.ingressInUse >= this.ingressCapacity || + egress > this.egressCapacity - this.egressInUse || + [...(room?.publications.values() ?? [])].some( + (publication) => publication.state === "reserved", + ) + ) { + return false; + } + + const resources = room ?? { publications: new Map() }; + const publication: PublicationEntry = { + fence: cloneFence(fence), + state: "reserved", + subscriptions: new Map(), + legacyEgress: egress, + }; + for (let index = 0; index < egress; index += 1) { + const viewerPeerId = `legacy_${index}`; + publication.subscriptions.set(viewerPeerId, { + fence: { ...cloneFence(fence), viewerPeerId }, + state: "reserved", + reservedFromDraining: false, + }); + } + resources.publications.set(publicationKey(fence), publication); this.rooms.set(fence.roomId, resources); this.ingressInUse += 1; this.egressInUse += egress; return true; } + /** Compatibility wrapper for the pre-controller aggregate commit. */ commit(fence: SfuResourceFence): readonly SfuResourceFence[] | null { - assertFence(fence); - const room = this.rooms.get(fence.roomId); - if (room?.committed && sameFence(room.committed.fence, fence)) { - return []; - } - if (!room?.reserved || !sameFence(room.reserved.fence, fence)) { - return null; - } - - const draining: SfuResourceFence[] = []; - if (room.committed) { - this.addDraining(room, room.committed); - draining.push({ ...room.committed.fence }); - } - room.committed = room.reserved; - room.reserved = undefined; - return draining; + return this.commitPublication(fence); } beginDrain(fence: SfuResourceFence): boolean { assertFence(fence); - const room = this.rooms.get(fence.roomId); - if (!room) { - return false; - } - - if (room.draining.has(fenceKey(fence))) { - return true; - } - let entry: SfuResourceEntry | undefined; - if (room.reserved && sameFence(room.reserved.fence, fence)) { - entry = room.reserved; - room.reserved = undefined; - } - if (room.committed && sameFence(room.committed.fence, fence)) { - entry = room.committed; - room.committed = undefined; - } - if (!entry) { - return false; - } - this.addDraining(room, entry); + const publication = this.publication(fence); + if (!publication) return false; + this.markDraining(publication); return true; } completeDrain(fence: SfuResourceFence): boolean { assertFence(fence); const room = this.rooms.get(fence.roomId); - const entry = room?.draining.get(fenceKey(fence)); - if (!room || !entry) { + const key = publicationKey(fence); + const publication = room?.publications.get(key); + if ( + !room || + !publication || + publication.state !== "draining" || + !sameFence(publication.fence, fence) + ) { return false; } - room.draining.delete(fenceKey(fence)); - this.subtract(entry); - this.deleteEmptyRoom(fence.roomId, room); + + room.publications.delete(key); + this.ingressInUse -= 1; + this.subtractEgress(publication.subscriptions.size); + this.assertNoUnderflow(); + if (room.publications.size === 0) this.rooms.delete(fence.roomId); return true; } beginDrainRoom(roomId: string): readonly SfuResourceFence[] { - if (!roomId) { - throw new Error("SFU resource room ID is invalid"); - } + if (!roomId) throw new Error("SFU resource room ID is invalid"); const room = this.rooms.get(roomId); - if (!room) { - return []; + if (!room) return []; + for (const publication of room.publications.values()) { + this.markDraining(publication); } - if (room.reserved) { - this.addDraining(room, room.reserved); - room.reserved = undefined; - } - if (room.committed) { - this.addDraining(room, room.committed); - room.committed = undefined; - } - return [...room.draining.values()].map((entry) => ({ ...entry.fence })); + return [...room.publications.values()].map((publication) => + cloneFence(publication.fence), + ); } beginDrainAll(): readonly SfuResourceFence[] { @@ -155,35 +277,49 @@ export class SfuResourceAdmission { return { ingress: this.ingressInUse, egress: this.egressInUse }; } - private subtract(entry: SfuResourceEntry): void { - this.ingressInUse -= 1; - this.egressInUse -= entry.egress; + private publication( + fence: SfuResourceFence, + ): PublicationEntry | undefined { + const publication = this.rooms + .get(fence.roomId) + ?.publications.get(publicationKey(fence)); + return publication && sameFence(publication.fence, fence) + ? publication + : undefined; + } + + private markDraining(publication: PublicationEntry): void { + publication.state = "draining"; + for (const subscription of publication.subscriptions.values()) { + subscription.state = "draining"; + subscription.reservedFromDraining = false; + } + } + + private subtractEgress(count: number): void { + this.egressInUse -= count; + this.assertNoUnderflow(); + } + + private assertNoUnderflow(): void { if (this.ingressInUse < 0 || this.egressInUse < 0) { throw new Error("SFU resource accounting underflow"); } } - - private addDraining(room: RoomSfuResources, entry: SfuResourceEntry): void { - room.draining.set(fenceKey(entry.fence), entry); - } - - private deleteEmptyRoom(roomId: string, room: RoomSfuResources): void { - if (!room.committed && !room.reserved && room.draining.size === 0) { - this.rooms.delete(roomId); - } - } } -function fenceKey(fence: SfuResourceFence): string { +function publicationKey(fence: SfuResourceFence): string { return `${fence.shareGeneration}\u0000${fence.publicationGeneration}`; } -function sameEntry( - entry: SfuResourceEntry, - fence: SfuResourceFence, - egress: number, -): boolean { - return entry.egress === egress && sameFence(entry.fence, fence); +function cloneFence(fence: SfuResourceFence): SfuResourceFence { + return { ...fence }; +} + +function cloneSubscriptionFence( + fence: SfuSubscriptionFence, +): SfuSubscriptionFence { + return { ...fence }; } function sameFence(left: SfuResourceFence, right: SfuResourceFence): boolean { @@ -194,12 +330,26 @@ function sameFence(left: SfuResourceFence, right: SfuResourceFence): boolean { ); } +function sameSubscriptionFence( + left: SfuSubscriptionFence, + right: SfuSubscriptionFence, +): boolean { + return sameFence(left, right) && left.viewerPeerId === right.viewerPeerId; +} + function assertFence(fence: SfuResourceFence): void { if (!fence.roomId || !fence.shareGeneration || !fence.publicationGeneration) { throw new Error("SFU resource fence is invalid"); } } +function assertSubscriptionFence(fence: SfuSubscriptionFence): void { + assertFence(fence); + if (!fence.viewerPeerId) { + throw new Error("SFU subscription fence is invalid"); + } +} + function assertPositiveSafeInteger(value: number, name: string): void { if (!Number.isSafeInteger(value) || value <= 0) { throw new Error(`${name} must be a positive safe integer`); diff --git a/tests/sfu-resource-admission.test.ts b/tests/sfu-resource-admission.test.ts index c122a19..cbc5c8e 100644 --- a/tests/sfu-resource-admission.test.ts +++ b/tests/sfu-resource-admission.test.ts @@ -3,9 +3,10 @@ import { describe, expect, it } from "vitest"; import { SfuResourceAdmission, type SfuResourceFence, + type SfuSubscriptionFence, } from "../src/server/sfu-resource-admission.ts"; -function fence( +function publication( roomId: string, publicationGeneration: string, shareGeneration = `share_${roomId}`, @@ -13,8 +14,15 @@ function fence( return { roomId, shareGeneration, publicationGeneration }; } +function subscription( + fence: SfuResourceFence, + viewerPeerId: string, +): SfuSubscriptionFence { + return { ...fence, viewerPeerId }; +} + describe("SfuResourceAdmission", () => { - it("requires explicit positive safe capacities", () => { + it("requires explicit capacities and enforces ingress and egress independently", () => { for (const ingressCapacity of [0, 1.5, Number.MAX_SAFE_INTEGER + 1]) { expect( () => new SfuResourceAdmission({ ingressCapacity, egressCapacity: 1 }), @@ -23,102 +31,208 @@ describe("SfuResourceAdmission", () => { expect( () => new SfuResourceAdmission({ ingressCapacity: 1, egressCapacity: 0 }), ).toThrow("positive safe integer"); - }); - it("enforces deployment-wide ingress and egress across rooms", () => { const admission = new SfuResourceAdmission({ ingressCapacity: 2, + egressCapacity: 2, + }); + const first = publication("1", "publication_first"); + const second = publication("2", "publication_second"); + expect(admission.reservePublication(first)).toBe(true); + expect(admission.reservePublication(second)).toBe(true); + expect( + admission.reservePublication(publication("3", "publication_third")), + ).toBe(false); + expect(admission.reserveSubscription(subscription(first, "viewer_a"))).toBe( + true, + ); + expect(admission.reserveSubscription(subscription(first, "viewer_b"))).toBe( + true, + ); + expect( + admission.reserveSubscription(subscription(second, "viewer_c")), + ).toBe(false); + expect(admission.usage()).toEqual({ ingress: 2, egress: 2 }); + }); + + it("keeps exact publication and subscription operations idempotent", () => { + const admission = new SfuResourceAdmission({ + ingressCapacity: 1, + egressCapacity: 1, + }); + const active = publication("1", "publication_active"); + const viewer = subscription(active, "viewer_a"); + expect(admission.reservePublication(active)).toBe(true); + expect(admission.reservePublication({ ...active })).toBe(true); + expect(admission.reserveSubscription(viewer)).toBe(true); + expect(admission.reserveSubscription({ ...viewer })).toBe(true); + expect(admission.commitSubscription(viewer)).toBe(false); + expect(admission.commitPublication(active)).toEqual([]); + expect(admission.commitPublication({ ...active })).toEqual([]); + expect(admission.commitSubscription(viewer)).toBe(true); + expect(admission.usage()).toEqual({ ingress: 1, egress: 1 }); + + const stale = publication("1", "publication_stale"); + expect(admission.reserveSubscription(subscription(stale, "viewer_a"))).toBe( + false, + ); + expect(admission.commitPublication(stale)).toBeNull(); + expect(admission.commitSubscription(subscription(active, "viewer_b"))).toBe( + false, + ); + expect(admission.releaseSubscription(subscription(active, "viewer_b"))).toBe( + false, + ); + expect(admission.beginDrain(stale)).toBe(false); + }); + + it("commits a new publication and its subscriptions before draining the old generation", () => { + const admission = new SfuResourceAdmission({ + ingressCapacity: 2, + egressCapacity: 3, + }); + const first = publication("1", "publication_first"); + const second = publication("1", "publication_second"); + const firstViewer = subscription(first, "viewer_a"); + expect(admission.reservePublication(first)).toBe(true); + expect(admission.reserveSubscription(firstViewer)).toBe(true); + expect(admission.commitPublication(first)).toEqual([]); + + expect(admission.reservePublication(second)).toBe(true); + expect(admission.reserveSubscription(subscription(second, "viewer_b"))).toBe( + true, + ); + expect(admission.reserveSubscription(subscription(second, "viewer_c"))).toBe( + true, + ); + expect(admission.usage()).toEqual({ ingress: 2, egress: 3 }); + expect(admission.commitPublication(second)).toEqual([first]); + expect(admission.commitSubscription(subscription(second, "viewer_b"))).toBe( + true, + ); + expect(admission.reservePublication(first)).toBe(false); + expect(admission.reserveSubscription(firstViewer)).toBe(false); + + expect(admission.completeDrain(first)).toBe(true); + expect(admission.completeDrain(first)).toBe(false); + expect(admission.usage()).toEqual({ ingress: 1, egress: 2 }); + }); + + it("refunds a new reserved subscription when its candidate aborts", () => { + const admission = new SfuResourceAdmission({ + ingressCapacity: 1, + egressCapacity: 2, + }); + const active = publication("1", "publication_active"); + const first = subscription(active, "viewer_a"); + const candidate = subscription(active, "viewer_b"); + expect(admission.reservePublication(active)).toBe(true); + expect(admission.reserveSubscription(first)).toBe(true); + expect(admission.commitPublication(active)).toEqual([]); + + expect(admission.reserveSubscription(candidate)).toBe(true); + expect(admission.usage()).toEqual({ ingress: 1, egress: 2 }); + expect(admission.releaseSubscription(candidate)).toBe(true); + expect(admission.releaseSubscription(candidate)).toBe(false); + expect(admission.commitSubscription(candidate)).toBe(false); + expect(admission.usage()).toEqual({ ingress: 1, egress: 1 }); + }); + + it("reactivates a draining subscription in the current generation without double charging", () => { + const admission = new SfuResourceAdmission({ + ingressCapacity: 1, + egressCapacity: 1, + }); + const active = publication("1", "publication_active"); + const viewer = subscription(active, "viewer_a"); + expect(admission.reservePublication(active)).toBe(true); + expect(admission.reserveSubscription(viewer)).toBe(true); + expect(admission.commitPublication(active)).toEqual([]); + expect(admission.releaseSubscription(viewer)).toBe(true); + expect(admission.usage()).toEqual({ ingress: 1, egress: 1 }); + + expect(admission.reserveSubscription(viewer)).toBe(true); + expect(admission.reserveSubscription(viewer)).toBe(true); + expect(admission.usage()).toEqual({ ingress: 1, egress: 1 }); + expect(admission.releaseSubscription(viewer)).toBe(true); + expect( + admission.reserveSubscription(subscription(active, "viewer_b")), + ).toBe(false); + expect(admission.reserveSubscription(viewer)).toBe(true); + expect(admission.commitSubscription(viewer)).toBe(true); + expect(admission.usage()).toEqual({ ingress: 1, egress: 1 }); + }); + + it("keeps departed subscription egress charged until publication drain proof", () => { + const admission = new SfuResourceAdmission({ + ingressCapacity: 1, + egressCapacity: 1, + }); + const active = publication("1", "publication_active"); + const viewer = subscription(active, "viewer_a"); + expect(admission.reservePublication(active)).toBe(true); + expect(admission.reserveSubscription(viewer)).toBe(true); + expect(admission.commitPublication(active)).toEqual([]); + expect(admission.releaseSubscription(viewer)).toBe(true); + expect( + admission.reserveSubscription(subscription(active, "viewer_b")), + ).toBe(false); + expect(admission.beginDrain(active)).toBe(true); + expect(admission.beginDrain(active)).toBe(true); + expect(admission.usage()).toEqual({ ingress: 1, egress: 1 }); + expect(admission.completeDrain(active)).toBe(true); + expect(admission.usage()).toEqual({ ingress: 0, egress: 0 }); + }); + + it("drains every reserved and committed generation exactly once", () => { + const admission = new SfuResourceAdmission({ + ingressCapacity: 3, egressCapacity: 4, }); - expect(admission.reserve(fence("1", "publication_a"), 2)).toBe(true); - expect(admission.reserve(fence("2", "publication_b"), 2)).toBe(true); - expect(admission.usage()).toEqual({ ingress: 2, egress: 4 }); + const active = publication("1", "publication_active"); + const replacement = publication("1", "publication_replacement"); + const other = publication("2", "publication_other"); + expect(admission.reservePublication(active)).toBe(true); + expect(admission.reserveSubscription(subscription(active, "viewer_a"))).toBe( + true, + ); + expect(admission.commitPublication(active)).toEqual([]); + expect(admission.reservePublication(replacement)).toBe(true); + expect( + admission.reserveSubscription(subscription(replacement, "viewer_b")), + ).toBe(true); + expect(admission.reservePublication(other)).toBe(true); + expect(admission.reserveSubscription(subscription(other, "viewer_c"))).toBe( + true, + ); - expect(admission.reserve(fence("3", "publication_c"), 1)).toBe(false); - expect(admission.usage()).toEqual({ ingress: 2, egress: 4 }); + const firstRoom = admission.beginDrainRoom("1"); + expect(firstRoom).toEqual([active, replacement]); + expect(admission.beginDrainRoom("1")).toEqual(firstRoom); + expect(admission.beginDrainAll()).toEqual([active, replacement, other]); + expect(admission.usage()).toEqual({ ingress: 3, egress: 3 }); + for (const fence of [active, replacement, other]) { + expect(admission.completeDrain(fence)).toBe(true); + expect(admission.completeDrain(fence)).toBe(false); + } + expect(admission.usage()).toEqual({ ingress: 0, egress: 0 }); }); - it("charges old and candidate generations until exact drain proof", () => { + it("keeps the aggregate API available while the router migrates", () => { const admission = new SfuResourceAdmission({ ingressCapacity: 2, - egressCapacity: 5, + egressCapacity: 3, }); - const first = fence("1", "publication_first"); - const second = fence("1", "publication_second"); + const first = publication("1", "publication_first"); + const second = publication("1", "publication_second"); expect(admission.reserve(first, 1)).toBe(true); + expect(admission.reserve(first, 1)).toBe(true); + expect(admission.reserve(first, 2)).toBe(false); expect(admission.commit(first)).toEqual([]); expect(admission.reserve(second, 2)).toBe(true); - expect(admission.usage()).toEqual({ ingress: 2, egress: 3 }); - - expect(admission.commit(fence("1", "publication_stale"))).toBeNull(); expect(admission.commit(second)).toEqual([first]); expect(admission.usage()).toEqual({ ingress: 2, egress: 3 }); expect(admission.completeDrain(first)).toBe(true); expect(admission.usage()).toEqual({ ingress: 1, egress: 2 }); - expect(admission.completeDrain(first)).toBe(false); - }); - - it("makes duplicate operations idempotent without weakening fences", () => { - const admission = new SfuResourceAdmission({ - ingressCapacity: 2, - egressCapacity: 3, - }); - const pending = fence("1", "publication_pending"); - expect(admission.reserve(pending, 2)).toBe(true); - expect(admission.reserve(pending, 2)).toBe(true); - expect(admission.reserve(pending, 1)).toBe(false); - expect(admission.reserve(fence("1", "publication_other"), 1)).toBe(false); - expect(admission.usage()).toEqual({ ingress: 1, egress: 2 }); - - expect(admission.beginDrain(pending)).toBe(true); - expect(admission.beginDrain(pending)).toBe(true); - expect(admission.usage()).toEqual({ ingress: 1, egress: 2 }); - expect(admission.completeDrain(pending)).toBe(true); - expect(admission.completeDrain(pending)).toBe(false); - expect(admission.usage()).toEqual({ ingress: 0, egress: 0 }); - }); - - it("moves a room's committed and reserved resources into draining together", () => { - const admission = new SfuResourceAdmission({ - ingressCapacity: 2, - egressCapacity: 4, - }); - const active = fence("1", "publication_active"); - expect(admission.reserve(active, 1)).toBe(true); - expect(admission.commit(active)).toEqual([]); - expect(admission.reserve(fence("1", "publication_candidate"), 2)).toBe( - true, - ); - - const draining = admission.beginDrainRoom("1"); - expect(draining).toHaveLength(2); - expect(admission.beginDrainRoom("1")).toEqual(draining); - expect(admission.usage()).toEqual({ ingress: 2, egress: 3 }); - for (const resourceFence of draining) { - expect(admission.completeDrain(resourceFence)).toBe(true); - } - expect(admission.usage()).toEqual({ ingress: 0, egress: 0 }); - }); - - it("moves every room to draining before its process owner closes", () => { - const admission = new SfuResourceAdmission({ - ingressCapacity: 2, - egressCapacity: 3, - }); - const active = fence("1", "publication_active"); - expect(admission.reserve(active, 1)).toBe(true); - expect(admission.commit(active)).toEqual([]); - expect(admission.reserve(fence("2", "publication_pending"), 2)).toBe( - true, - ); - - const draining = admission.beginDrainAll(); - expect(draining).toHaveLength(2); - expect(admission.beginDrainAll()).toEqual(draining); - expect(admission.usage()).toEqual({ ingress: 2, egress: 3 }); - for (const resourceFence of draining) { - expect(admission.completeDrain(resourceFence)).toBe(true); - } - expect(admission.usage()).toEqual({ ingress: 0, egress: 0 }); }); });