From 57013ec9ad16416b8fb13e8d928bdfa534aa93e7 Mon Sep 17 00:00:00 2001 From: Russell Ballestrini Date: Tue, 2 Jun 2026 11:03:48 -0400 Subject: [PATCH] =?UTF-8?q?zebra-spaces:=20SubscribeFSM=20=E2=80=94=20form?= =?UTF-8?q?alises=20renegotiation=20queue=20+=20reconnect?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Second state machine. Models the SFU subscribe leg explicitly: off ──START──▶ connecting ──CONNECTED──▶ subscribed │ FAILED │ RENEG ▼ ▼ off renegotiating │ RENEG_DONE / RENEG_FAILED ▼ subscribed │ LOST ▼ reconnecting ──CONNECTED──▶ subscribed │ STOP │ FAILED ▼ ▼ stopping off │ DONE ▼ off Renegotiation is its own state so concurrent SSE offers can't race setRemoteDescription (the bug ae9721e patched imperatively with a promise queue). A RENEG event during renegotiating parks the SDP on ctx.pendingOffers; the runtime will drain that queue from an observer when RENEG_DONE fires. RENEG_FAILED returns to subscribed without killing the PC — the negotiation attempt is what failed, the channel itself is still up. LOST during renegotiating jumps straight to reconnecting (drops the in-flight reneg cleanly; when the connection comes back the runtime will re-deliver any still-relevant SDP via fresh RENEGs). + 15 new unit tests covering connect, queue, drops, teardown, illegal transitions. test-fsm now reports 38 passed. --- test/zebra-fsm.test.js | 141 ++++++++++++++++++++++++++++++++++++++++- web/zebra-spaces.html | 88 ++++++++++++++++++++++++- 2 files changed, 224 insertions(+), 5 deletions(-) diff --git a/test/zebra-fsm.test.js b/test/zebra-fsm.test.js index 28092f8..fe13780 100644 --- a/test/zebra-fsm.test.js +++ b/test/zebra-fsm.test.js @@ -29,15 +29,17 @@ function extract(re){ return src.slice(m.index, j); } -const createFSMSrc = extract(/function createFSM\(/); +const createFSMSrc = extract(/function createFSM\(/); const publishSpecSrc = extract(/const publishSpec = /); +const subscribeSpecSrc = extract(/const subscribeSpec = /); /* Function-constructor scope so `const` declarations are visible at the * harness's `return` — they would NOT leak through a bare `eval()`. */ const harness = new Function( - createFSMSrc + '\n' + publishSpecSrc + '\nreturn { createFSM, publishSpec };' + createFSMSrc + '\n' + publishSpecSrc + '\n' + subscribeSpecSrc + + '\nreturn { createFSM, publishSpec, subscribeSpec };' ); -const { createFSM, publishSpec } = harness(); +const { createFSM, publishSpec, subscribeSpec } = harness(); let pass = 0, fail = 0; function test(name, fn){ @@ -254,5 +256,138 @@ test('stopping + STOP is a no-op (already on the way out)', () => { eq(m.state, 'stopping'); }); +console.log('subscribeSpec — connect:'); + +test('starts in off', () => { + const m = createFSM(subscribeSpec); + eq(m.state, 'off'); +}); + +test('off → connecting on START', () => { + const m = createFSM(subscribeSpec); + m.send('START'); + eq(m.state, 'connecting'); +}); + +test('connecting → subscribed on CONNECTED, payload lands in ctx', () => { + const m = createFSM(subscribeSpec); + m.send('START'); + const pc = { id: 'pc' }, events = { id: 'es' }; + m.send('CONNECTED', { pc, peerID: 'peer-x', events }); + eq(m.state, 'subscribed'); + eq(m.context.pc, pc); + eq(m.context.peerID, 'peer-x'); + eq(m.context.events, events); +}); + +test('connecting + FAILED → off with error', () => { + const m = createFSM(subscribeSpec); + m.send('START'); + m.send('FAILED', { error: 'sfu 502' }); + eq(m.state, 'off'); + eq(m.context.lastError, 'sfu 502'); +}); + +console.log('subscribeSpec — renegotiation queue:'); + +test('subscribed + RENEG → renegotiating, sdp parked on pendingOffers', () => { + const m = createFSM(subscribeSpec); + m.send('START'); m.send('CONNECTED', { pc: {} }); + m.send('RENEG', { sdp: 'sdp-1' }); + eq(m.state, 'renegotiating'); + eq(m.context.pendingOffers.length, 1); + eq(m.context.pendingOffers[0], 'sdp-1'); +}); + +test('a second RENEG while renegotiating queues the next sdp', () => { + const m = createFSM(subscribeSpec); + m.send('START'); m.send('CONNECTED', { pc: {} }); + m.send('RENEG', { sdp: 'sdp-1' }); + m.send('RENEG', { sdp: 'sdp-2' }); + m.send('RENEG', { sdp: 'sdp-3' }); + eq(m.state, 'renegotiating'); + eq(m.context.pendingOffers.length, 3); +}); + +test('renegotiating + RENEG_DONE → subscribed', () => { + const m = createFSM(subscribeSpec); + m.send('START'); m.send('CONNECTED', { pc: {} }); + m.send('RENEG', { sdp: 'sdp-1' }); + m.send('RENEG_DONE'); + eq(m.state, 'subscribed'); +}); + +test('renegotiating + RENEG_FAILED → subscribed, error captured (recoverable)', () => { + const m = createFSM(subscribeSpec); + m.send('START'); m.send('CONNECTED', { pc: {} }); + m.send('RENEG', { sdp: 'sdp-1' }); + m.send('RENEG_FAILED', { error: 'sdp parse' }); + eq(m.state, 'subscribed'); + eq(m.context.lastError, 'sdp parse'); +}); + +console.log('subscribeSpec — drops + teardown:'); + +test('subscribed + LOST → reconnecting', () => { + const m = createFSM(subscribeSpec); + m.send('START'); m.send('CONNECTED', { pc: {} }); + m.send('LOST'); + eq(m.state, 'reconnecting'); +}); + +test('reconnecting + CONNECTED → subscribed', () => { + const m = createFSM(subscribeSpec); + m.send('START'); m.send('CONNECTED', { pc: {} }); + m.send('LOST'); + m.send('CONNECTED'); + eq(m.state, 'subscribed'); +}); + +test('renegotiating + LOST → reconnecting (drops in-flight reneg)', () => { + const m = createFSM(subscribeSpec); + m.send('START'); m.send('CONNECTED', { pc: {} }); + m.send('RENEG', { sdp: 'sdp-1' }); + m.send('LOST'); + eq(m.state, 'reconnecting'); +}); + +test('subscribed + STOP → stopping → off (DONE clears ctx)', () => { + const m = createFSM(subscribeSpec); + m.send('START'); m.send('CONNECTED', { pc: { id: 'pc' }, peerID: 'p', events: { id: 'es' } }); + m.send('STOP'); + eq(m.state, 'stopping'); + m.send('DONE'); + eq(m.state, 'off'); + eq(m.context.pc, null); + eq(m.context.peerID, null); + eq(m.context.events, null); + eq(m.context.pendingOffers.length, 0); +}); + +console.log('subscribeSpec — illegal transitions are no-ops:'); + +test('off + RENEG is a no-op', () => { + const m = createFSM(subscribeSpec); + eq(m.send('RENEG', { sdp: 's' }), false); + eq(m.state, 'off'); +}); + +test('connecting + RENEG is a no-op (handshake not complete)', () => { + const m = createFSM(subscribeSpec); + m.send('START'); + eq(m.send('RENEG', { sdp: 's' }), false); + eq(m.state, 'connecting'); +}); + +test('stopping + RENEG / LOST / CONNECTED are all no-ops', () => { + const m = createFSM(subscribeSpec); + m.send('START'); m.send('CONNECTED', { pc: {} }); + m.send('STOP'); + eq(m.send('RENEG', { sdp: 's' }), false); + eq(m.send('LOST'), false); + eq(m.send('CONNECTED'), false); + eq(m.state, 'stopping'); +}); + console.log('\n' + pass + ' passed, ' + fail + ' failed'); process.exit(fail === 0 ? 0 : 1); diff --git a/web/zebra-spaces.html b/web/zebra-spaces.html index d662815..f9a568b 100644 --- a/web/zebra-spaces.html +++ b/web/zebra-spaces.html @@ -821,6 +821,90 @@ const publishSpec = { }, }; +/* ================================================================== + * SubscribeFSM — one instance for the SFU subscribe leg. + * + * off ──START──▶ connecting ──CONNECTED──▶ subscribed + * ▲ │ FAILED │ RENEG (SSE offer arrives) + * │ ▼ ▼ + * │ off renegotiating + * │ │ RENEG_DONE / RENEG_FAILED + * │ ▼ + * │ subscribed + * │ │ LOST (SSE drop) + * │ ▼ + * │ reconnecting ──CONNECTED──▶ subscribed + * │ │ + * │ STOP + * │ ▼ + * └────────────── DONE ──────────────── stopping + * + * Renegotiation lives in its own state so RENEG offers queued during an + * in-flight one can't race setRemoteDescription (the bug we hit in + * commit ae9721e). When RENEG fires while already renegotiating, the + * spec parks the SDP on ctx.pendingOffers; the runtime drains the queue + * on RENEG_DONE via an observer, so the FSM itself stays pure. */ +const subscribeSpec = { + initial: 'off', + context: { pc: null, peerID: null, events: null, lastError: null, pendingOffers: [] }, + states: { + off: { + entry: (ctx) => { ctx.pc = null; ctx.peerID = null; ctx.events = null; ctx.pendingOffers = []; }, + on: { START: 'connecting' }, + }, + connecting: { + on: { + CONNECTED: { + target: 'subscribed', + action: (ctx, ev) => { + if (ev.payload){ + ctx.pc = ev.payload.pc || ctx.pc; + ctx.peerID = ev.payload.peerID || ctx.peerID; + ctx.events = ev.payload.events || ctx.events; + } + }, + }, + FAILED: { target: 'off', action: (ctx, ev) => { ctx.lastError = ev.payload && ev.payload.error; } }, + STOP: 'off', + }, + }, + subscribed: { + on: { + RENEG: { + target: 'renegotiating', + action: (ctx, ev) => { const sdp = ev.payload && ev.payload.sdp; if (sdp) ctx.pendingOffers.push(sdp); }, + }, + LOST: 'reconnecting', + STOP: 'stopping', + }, + }, + renegotiating: { + on: { + /* extra offers queue up; we re-enter renegotiating so observers + * can see the transition but the existing run continues */ + RENEG: { + target: 'renegotiating', + action: (ctx, ev) => { const sdp = ev.payload && ev.payload.sdp; if (sdp) ctx.pendingOffers.push(sdp); }, + }, + RENEG_DONE: 'subscribed', + RENEG_FAILED: { target: 'subscribed', action: (ctx, ev) => { ctx.lastError = ev.payload && ev.payload.error; } }, + LOST: 'reconnecting', + STOP: 'stopping', + }, + }, + reconnecting: { + on: { + CONNECTED: 'subscribed', + FAILED: { target: 'off', action: (ctx, ev) => { ctx.lastError = ev.payload && ev.payload.error; } }, + STOP: 'stopping', + }, + }, + stopping: { + on: { DONE: 'off' }, + }, + }, +}; + /* ================================================================== * identity — ed25519 keypair, persisted in localStorage as JWK. * @@ -2889,8 +2973,8 @@ logLine('', 'ready — pick a handle, type a rendezvous code, enter the space');