zebra-spaces: SubscribeFSM — formalises renegotiation queue + reconnect

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.
This commit is contained in:
Russell Ballestrini 2026-06-02 11:03:48 -04:00
parent ebda460574
commit 57013ec9ad
No known key found for this signature in database
2 changed files with 224 additions and 5 deletions

View file

@ -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);

View file

@ -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');
<footer style="margin:2.2rem auto 0;font-size:0.65rem;color:#999;line-height:1.7;word-break:break-all;font-family:monospace">
<span id="pi-seal" style="color:#777;cursor:default;user-select:none" title="">page integrity</span> &nbsp;·&nbsp; built <span class="stamp-date">2026-06-02</span><br>
md5 <span class="stamp-md5">41b1bef910d4b758a3c3170540b2c4ec</span><br>
sha256 <span class="stamp-sha">c89827a1cf4ddbc122642de2365f0fcf2dc60cae7725a10c64548b4e40c055ff</span><br>
md5 <span class="stamp-md5">fabb116d48d04e00c0d689f14e475ead</span><br>
sha256 <span class="stamp-sha">f449dab274cac4a6ecf36a7329eb759d1ea494789b054e1431408452168004aa</span><br>
<span style="color:#bbb">hashes are of this page with these two fields zeroed — to verify, blank them and re-hash</span><br>
<span style="color:#bbb">one self-contained file — <strong>save a copy</strong> and verify against these hashes; point at your own servers with ?signal= and ?turncred=, or <a href="host-your-own.html" style="color:#999">host your own community</a></span>
</footer>