refactor(route): account reusable SFU subscriptions
This commit is contained in:
1 parent
cd534c7e68
commit
b656f47951
2 files changed
+443
-179
No files matched your search
@@ -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<string, SfuResourceEntry>;
|
||||
interface PublicationEntry {
|
||||
fence: SfuResourceFence;
|
||||
state: ResourceState;
|
||||
subscriptions: Map<string, SubscriptionEntry>;
|
||||
legacyEgress?: number;
|
||||
}
|
||||
|
||||
interface RoomResources {
|
||||
publications: Map<string, PublicationEntry>;
|
||||
}
|
||||
|
||||
export class SfuResourceAdmission {
|
||||
private readonly rooms = new Map<string, RoomSfuResources>();
|
||||
private readonly rooms = new Map<string, RoomResources>();
|
||||
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`);
|
||||
|
||||
@@ -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 });
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user