diff --git a/src/__tests__/returns-guest-auth.test.ts b/src/__tests__/returns-guest-auth.test.ts index 3de5db4f..04294cfc 100644 --- a/src/__tests__/returns-guest-auth.test.ts +++ b/src/__tests__/returns-guest-auth.test.ts @@ -16,7 +16,7 @@ * mocked — no live services required. */ -import { describe, it, expect, vi, beforeAll, beforeEach } from 'vitest'; +import { describe, it, expect, vi, beforeAll, beforeEach, afterEach } from 'vitest'; import type { FastifyInstance } from 'fastify'; import type { GuestInviteDoc, GuestSessionDoc, ProductionDoc } from '../db/types.js'; @@ -352,6 +352,84 @@ describe('Return-picture WHEP guest-token auth (issue #380)', () => { }); }); +describe('Return-picture session binding under concurrent writes to the guest session', () => { + const bearer = (token: string) => ({ authorization: `Bearer ${token}` }); + const revNumber = (rev: string | undefined) => Number(rev?.split('-')[0] ?? 0); + const defaultInsert = sessionsDb.insert.getMockImplementation()!; + + // Enforce `_rev` like CouchDB: an update must carry the stored doc's current + // `_rev` or it gets a 409; each write bumps the revision. + beforeEach(() => { + sessionsDb.insert.mockImplementation(async (doc: GuestSessionDoc) => { + const stored = sessionsStore.get(doc._id); + if (stored && doc._rev !== stored._rev) throw Object.assign(new Error('conflict'), { statusCode: 409 }); + sessionsStore.set(doc._id, { ...doc, _rev: `${revNumber(stored?._rev) + 1}-x` }); + return { ok: true }; + }); + }); + afterEach(() => { + sessionsDb.insert.mockImplementation(defaultInsert); + }); + + /** Answers the Strom WHEP POST after `during` has changed the guest session, as a concurrent request would. */ + function stromAnswersAfter(during: () => void) { + vi.stubGlobal('fetch', vi.fn().mockImplementation(async () => { + during(); + return { + ok: true, + status: 201, + text: async () => 'v=0 mock-answer-sdp', + headers: { get: (name: string) => (name === 'Location' ? '/whep/session-picture' : null) }, + }; + })); + } + + it('keeps the binding, and the other write, when the session changed during the Strom round trip', async () => { + seedActiveProduction(); + const invite = await createInvite('prod-1', { mixerInput: 'video_in_0' }); + const { guestId } = await joinGuest(invite.id, invite.token); + // The guest page's talkback WHIP binding (guests.ts) lands while the + // return-picture offer is with Strom: the session doc this request read + // is no longer current when it writes its own binding. + stromAnswersAfter(() => { + const stored = sessionsStore.get(guestId)!; + sessionsStore.set(guestId, { ...stored, talkbackWhipSessionId: 'tb-1', _rev: `${revNumber(stored._rev) + 1}-x` }); + }); + + const { res, sessionId } = await postPicture('video_in_0', bearer(invite.token)); + expect(res.statusCode).toBe(201); + const session = sessionsStore.get(guestId)!; + expect(session.returnWhepSessionId).toBe(sessionId); + expect(session.talkbackWhipSessionId).toBe('tb-1'); + + // So the guest can still tear their own return session down. + const del = await app.inject({ + method: 'DELETE', + url: `/api/v1/productions/prod-1/returns/video_in_0/picture/whep/${sessionId}`, + headers: bearer(invite.token), + }); + expect(del.statusCode).toBe(204); + }); + + it('does not bind for a guest who left while the offer was with Strom', async () => { + seedActiveProduction(); + const invite = await createInvite('prod-1', { mixerInput: 'video_in_0' }); + const { guestId } = await joinGuest(invite.id, invite.token); + stromAnswersAfter(() => { + const stored = sessionsStore.get(guestId)!; + sessionsStore.set(guestId, { ...stored, state: 'left', _rev: `${revNumber(stored._rev) + 1}-x` }); + }); + + const { res } = await postPicture('video_in_0', bearer(invite.token)); + // The upstream session already exists, so the offer still succeeds… + expect(res.statusCode).toBe(201); + // …but the left session is not brought back to life with a binding. + const session = sessionsStore.get(guestId)!; + expect(session.state).toBe('left'); + expect(session.returnWhepSessionId).toBeUndefined(); + }); +}); + describe('Guest return mode: GET / PUT /api/v1/guests/:inviteId/session/return', () => { const returnUrl = (inviteId: string) => `/api/v1/guests/${inviteId}/session/return`; const bearer = (token: string) => ({ authorization: `Bearer ${token}` }); diff --git a/src/lib/db-errors.ts b/src/lib/db-errors.ts index 69f1986a..065fdf02 100644 --- a/src/lib/db-errors.ts +++ b/src/lib/db-errors.ts @@ -13,3 +13,8 @@ export function isDocNotFoundError(err: unknown): boolean { if ((err as { statusCode?: number } | null)?.statusCode === 404) return true; return err instanceof Error && err.message.startsWith('Document type mismatch'); } + +/** A CouchDB `.insert()` refused because the document's `_rev` is no longer current (409). */ +export function isDocConflictError(err: unknown): boolean { + return (err as { statusCode?: number } | null)?.statusCode === 409; +} diff --git a/src/routes/returns.ts b/src/routes/returns.ts index 9977bb8d..811763e1 100644 --- a/src/routes/returns.ts +++ b/src/routes/returns.ts @@ -3,6 +3,7 @@ import { z } from 'zod'; import { getDb, getGuestSessionsDb } from '../db/index.js'; import type { GuestSessionDoc, ProductionDoc, ProductionSourceAssignment } from '../db/types.js'; import { getStromToken } from '../lib/strom-token.js'; +import { isDocConflictError } from '../lib/db-errors.js'; import { assertSameStromOrigin } from '../lib/url-validation.js'; import { config, isGuestCallingEnabled } from '../config.js'; import { resolveGuestSession } from '../lib/guest-scope.js'; @@ -116,23 +117,71 @@ async function proxyReturnPictureWhep(args: { } // Bind this WHEP session id to the guest's live session (issue #380) so the // matching DELETE can verify a guest caller only tears down their OWN return - // session — never another guest's :sessionId. Best-effort: a persist failure - // here does not fail the (already-established) upstream session; the guest - // just fails closed on their next DELETE instead of silently trusting an - // unbound id (see the DELETE handlers below). + // session — never another guest's :sessionId. if (guestSession && sessionId) { + await bindWhepSession(reply, 'return picture WHEP', { + session: guestSession, + apply: (doc, whepSessionId) => ({ ...doc, returnWhepSessionId: whepSessionId, updatedAt: new Date().toISOString() }), + }, sessionId); + } + reply.header('Content-Type', 'application/sdp'); + return reply.status(201).send(answerSdp); +} + +/** + * How a guest-scoped WHEP POST records its session id: the guest's session doc + * as read at the start of the request, and the one change to make to it. + * `apply` must only set its own key, because it may be re-applied to a newer + * copy of the doc (see `bindWhepSession`). + */ +interface GuestSessionBinding { + session: GuestSessionDoc; + apply: (doc: GuestSessionDoc, whepSessionId: string) => GuestSessionDoc; +} + +/** Attempts at writing a WHEP session id into a guest session doc (one write plus retries on conflict). */ +const MAX_BIND_ATTEMPTS = 4; + +/** + * Writes the WHEP session id into the guest's session doc. The request read + * that doc at its start; while the Strom round trip ran, other requests of the + * same guest page may have written it (the talkback WHIP/WHEP bindings at join, + * a mute or a status change), so the insert can get a CouchDB 409. On a + * conflict the doc is read again, still has to be live (`state !== 'left'`), + * and only this request's key is applied to it, so neither this binding nor the + * other request's change is lost. Best-effort after that: a failure does not + * fail the (already-established) upstream session; the guest just fails closed + * on their next DELETE instead of trusting an unbound id. + */ +async function bindWhepSession( + reply: FastifyReply, + logName: string, + binding: GuestSessionBinding, + whepSessionId: string, +): Promise { + const db = getGuestSessionsDb(); + let doc = binding.session; + for (let attempt = 1; ; attempt++) { try { - await getGuestSessionsDb().insert({ - ...guestSession, - returnWhepSessionId: sessionId, - updatedAt: new Date().toISOString(), - }); + await db.insert(binding.apply(doc, whepSessionId)); + return; + } catch (err) { + if (!isDocConflictError(err) || attempt >= MAX_BIND_ATTEMPTS) { + reply.log.warn({ err, attempt }, `${logName}: guest session-id bind failed`); + return; + } + } + try { + doc = await db.get(doc._id); } catch (err) { - reply.log.warn({ err }, 'return picture WHEP: guest session-id bind failed'); + reply.log.warn({ err }, `${logName}: guest session re-read failed, session-id not bound`); + return; + } + if (doc.state === 'left') { + reply.log.info(`${logName}: guest left during the request, session-id not bound`); + return; } } - reply.header('Content-Type', 'application/sdp'); - return reply.status(201).send(answerSdp); } /**