This commit is contained in:
velzie committed 2026-10-03 11:14:21 -04:00
1 parent 75e437013f
commit 8082541304
12 files changed
+301 -21

No files matched your search

+9 -2
View File
@@ -11,7 +11,7 @@ The `PuterPeerServer` object returned by [`puter.peer.serve()`](/Peer/serve/). I
#### `inviteCode` (String)
The code to share with other clients so they can connect with [`puter.peer.connect()`](/Peer/connect/).
The code to share with other clients so they can connect with [`puter.peer.connect()`](/Peer/connect/). It survives the signalling connection dropping and coming back; on the rare occasion it cannot, the `reconnect` event carries the new one.
#### `connections` (Map)
@@ -21,7 +21,7 @@ A `Map` of every connected client, keyed by connection id. The values are [`Pute
#### `close()`
Closes every client connection and the signalling connection. The invite code stops working.
Closes every client connection and the signalling connection, and gives the invite code up so it stops working at once.
## Events
@@ -32,6 +32,13 @@ Fired when a client connects. The event has the following attributes:
- `conn` ([`PuterPeerConnection`](/Objects/puterpeerconnection/)) - The connection to the client.
- `user` (Object) - Metadata about the connecting user, with `username` and `uuid` (if available).
#### `reconnect`
Fired when a dropped signalling connection has been re-established. A server dials the signaller again on its own, so a network blip, a laptop waking up or a signaller restart does not end the session — and existing client connections, being peer-to-peer, carry on throughout either way.
- `inviteCode` (String) - The code to share from now on.
- `resumed` (Boolean) - Whether the previous session was reclaimed. When `true` the invite code is unchanged and the clients already connected can still be renegotiated with, so there is nothing to do. When `false` the session was gone — clients connected under it can no longer be reached and will close — and `inviteCode` is a new code that has to be shared in its place.
## Example
```js
+1
View File
@@ -34,6 +34,7 @@ A string invite code created by `puter.peer.serve()`.
- `forceRelay` (`boolean`) Whether to force connections to route through a relay instead of attempting peer-to-peer (default). Metering charges may apply.
- `anonToken` (`String`) Join without a Puter session. Any uuid — it identifies this guest for the duration of the session, and no sign-in prompt is shown. The host sees the guest as `anonymous`, so anything you want to call them is yours to send over the connection.
- `turnGrant` (`String`) A grant from [`puter.peer.createGuestGrant()`](/Peer/createGuestGrant/). Lets a guest use the Puter-managed relays on the host's account. Without one, a guest connects only where a direct connection is possible; with `forceRelay`, a guest needs one.
- `recoveryTimeout` (`Number`) How long, in milliseconds, the connection keeps trying to repair a broken link before it closes, counted from when the link first showed trouble. Defaults to `60000`.
## Return value
+1
View File
@@ -29,6 +29,7 @@ const server = await puter.peer.serve(options);
- `iceServers` (`RTCIceServer[]`) Custom ICE servers (STUN/TURN) to use instead of the Puter-managed relays.
- `forceRelay` (`boolean`) Whether to force connections to route through a relay instead of attempting peer-to-peer (default). Metering charges will increase.
- `anonToken` (`String`) Host without a Puter session. Any uuid; no sign-in prompt is shown. An anonymous host has no account to attribute relay usage to, so it cannot issue guest grants and gets no relays of its own.
- `recoveryTimeout` (`Number`) How long, in milliseconds, each connection keeps trying to repair a broken link before it closes, counted from when the link first showed trouble. Defaults to `60000`.
To let people join your session without accounts of their own, keep hosting authenticated and give them a grant — see [`puter.peer.createGuestGrant()`](/Peer/createGuestGrant/).
@@ -256,7 +256,11 @@ export class PerfectNegotiator {
async #applyAnswer ( description, names ) {
const pc = this.#pc;
if ( pc.signalingState === 'closed' ) return;
// An answer with no offer of ours outstanding replies to one that has
// since been rolled back or replaced - several restart offers held up
// by a stalled socket arrive at once, and so do their answers. There
// is nothing for it to settle.
if ( pc.signalingState !== 'have-local-offer' ) return;
try {
await this.#adopt(description, names);
if ( pc.signalingState === 'stable' ) {
@@ -1,4 +1,4 @@
import { PerfectNegotiator } from './PerfectNegotiator.js';
import { DEFAULT_ANSWER_TIMEOUT, PerfectNegotiator } from './PerfectNegotiator.js';
import { TrackPublisher } from './tracks.js';
import { ClientSignallingChannel } from './signalling.js';
import {
@@ -14,7 +14,11 @@ import {
/** @typedef {import('./tracks.js').PuterPeerPublishOptions} PuterPeerPublishOptions */
/**
* How long to keep trying to rescue a link before giving it up.
* How long to keep trying to rescue a link before giving it up, counted from
* the first sign of trouble rather than from the transport failing: Chrome
* takes a further ten seconds or so to call a silent link `failed`, and an
* app waiting to act on a peer that vanished should not have to add that on.
* Apps can shorten it with the `recoveryTimeout` option.
*
* Counted in time rather than attempts, because what recovery is usually
* waiting on is the peer coming back - a lid reopened, a network rejoined -
@@ -23,6 +27,15 @@ import {
*/
const RECOVERY_BUDGET_MS = 60_000;
/**
* How long recovery keeps trying once the peer's own signalling session has
* ended. Until it comes back they cannot answer an offer at all, so the
* budget above would be spent on a peer that provably cannot reply - and an
* app waiting on this connection to close before it acts (to elect a new
* host, say) would wait that whole time for an answer it was never getting.
*/
const PEER_AWAY_GRACE_MS = 10_000;
/** Gap between restart attempts, doubling, so a long wait stays quiet. */
const RECOVERY_BACKOFF_MS = 2000;
const RECOVERY_BACKOFF_MAX_MS = 15_000;
@@ -66,16 +79,28 @@ export class PuterPeerConnection extends EventTarget {
#recovering = false;
/** The peer said goodbye, or closed the channel. Proof, not evidence. */
#peerHungUp = false;
/** The peer's signalling session ended and has not come back. */
#peerAway = false;
#transportWaiters = new Set();
#signallingWaiters = new Set();
#recoveryBudget;
#polite;
/** When the transport last left `connected`; recovery's budget runs from here. */
#troubleSince = null;
/**
* @param {object} peerConfig
* @param {{ polite?: boolean, channel?: import('./signalling.js').SignallingChannel }} [options]
* @param {{
* polite?: boolean,
* channel?: import('./signalling.js').SignallingChannel,
* recoveryTimeout?: number,
* }} [options]
*/
constructor ( peerConfig, { polite = false, channel } = {} ) {
constructor ( peerConfig, { polite = false, channel, recoveryTimeout = RECOVERY_BUDGET_MS } = {} ) {
super();
this.#peerConfig = peerConfig;
this.#polite = polite;
this.#recoveryBudget = recoveryTimeout;
this.#channel = channel ?? new ClientSignallingChannel(peerConfig);
this.peerconnection = new RTCPeerConnection({
@@ -122,6 +147,7 @@ export class PuterPeerConnection extends EventTarget {
this.#channel.onpeergone = (reason, resumable) => this.#onPeerGone(reason, resumable);
this.#channel.onunusable = () => this.#onSignallingLost();
this.#channel.onusable = () => this.#wakeSignallingWaiters();
this.#channel.onpeerback = () => this.#onPeerBack();
}
/**
@@ -170,11 +196,22 @@ export class PuterPeerConnection extends EventTarget {
*/
#onPeerGone ( reason, resumable ) {
if ( ! resumable ) this.#peerHungUp = true;
this.#peerAway = true;
// A handshake has nothing to wait on either way: whoever is dialling
// needs a definite answer rather than a socket that may come back.
if ( ! this.connected ) this.#doclose(reason, undefined);
}
/**
* The peer reclaimed the session it dropped, so it can answer again.
* Anything waiting on that gets to stop waiting now.
*/
#onPeerBack () {
this.#peerAway = false;
this.#wakeSignallingWaiters();
this.#wakeTransportWaiters();
}
/**
* Our own signalling socket dropped. Renegotiation is off the table until
* it returns, but an open data channel keeps working without it.
@@ -190,7 +227,10 @@ export class PuterPeerConnection extends EventTarget {
#onConnectionState () {
this.#wakeTransportWaiters();
switch ( this.peerconnection.connectionState ) {
const state = this.peerconnection.connectionState;
if ( state === 'connected' ) this.#troubleSince = null;
else if ( state === 'disconnected' || state === 'failed' ) this.#troubleSince ??= Date.now();
switch ( state ) {
case 'connected':
this.#setLinkState('connected');
break;
@@ -198,8 +238,10 @@ export class PuterPeerConnection extends EventTarget {
// escalates to 'failed', which is where recovery belongs. Nothing
// is done about it, but a watcher is told, because frozen video
// starts here rather than at 'failed'.
// Recovery is already saying more than that, and an unanswered
// restart lands here too.
case 'disconnected':
this.#setLinkState('unstable');
if ( ! this.#recovering ) this.#setLinkState('unstable');
break;
case 'failed':
this.#recover();
@@ -232,19 +274,45 @@ export class PuterPeerConnection extends EventTarget {
* or the budget is spent. The browser has no reason to raise another
* `connectionstatechange` while the state stays `failed`, so the retries
* belong in here rather than in the event.
*
* Recovery lasts until the transport is connected again, not merely until
* it stops reading `failed`: Chrome answers an unanswered restart offer by
* reporting `disconnected`, then `failed` again once that times out too.
* Letting that end recovery would start a fresh budget every cycle, and a
* peer that vanished would never be given up.
*
* Only the impolite side restarts; the polite side answers and waits.
* Restart offers from both sides made during an outage are held up
* together and collide the moment it ends, and the polite side rolling
* its own back then leaves Chrome's congestion control stuck near zero:
* the link reports connected, data flows, and the polite side's video
* never encodes another frame. Nor does a polite restart ever help - with
* the impolite side's signalling up it restarts by itself, and with it
* down the offer has nowhere to go.
*/
async #recover () {
if ( this.closed || this.#recovering ) return;
this.#recovering = true;
const deadline = Date.now() + RECOVERY_BUDGET_MS;
const deadline = (this.#troubleSince ?? Date.now()) + this.#recoveryBudget;
let attempt = 0;
/** Set while the peer is away; recovery outlives it only so long. */
let awayDeadline = null;
try {
let unanswered = false;
while ( ! this.closed && this.peerconnection.connectionState === 'failed' ) {
while ( ! this.closed && this.peerconnection.connectionState !== 'connected' ) {
if ( this.#peerHungUp ) {
this.#doclose('the peer hung up', undefined);
return;
}
if ( this.#peerAway ) {
awayDeadline ??= Date.now() + PEER_AWAY_GRACE_MS;
if ( Date.now() >= awayDeadline ) {
this.#doclose('the peer is no longer reachable', undefined);
return;
}
} else {
awayDeadline = null;
}
// No route left to offer over, and none coming.
if ( this.#channel.stranded ) {
this.#doclose('the peer is no longer reachable', undefined);
@@ -259,6 +327,15 @@ export class PuterPeerConnection extends EventTarget {
return;
}
if ( this.#polite ) {
if ( this.linkState !== 'recovering' ) this.#setLinkState('recovering');
// Woken early by the transport changing or the peer
// reclaiming its session; the checks above run again on
// the way round.
await this.#nextTransportChange(Math.min(RECOVERY_BACKOFF_MS, left));
continue;
}
// Nothing can be offered without signalling. It comes back
// on its own when a peer server reclaims its session, so
// wait on it rather than spend an attempt failing.
@@ -270,7 +347,8 @@ export class PuterPeerConnection extends EventTarget {
attempt++;
this.#setLinkState('recovering', { attempt });
try {
await this.#negotiator.restartIce();
// An answer after the deadline is too late to use.
await this.#negotiator.restartIce(Math.min(DEFAULT_ANSWER_TIMEOUT, deadline - Date.now()));
unanswered = false;
} catch {
unanswered = true;
@@ -278,7 +356,7 @@ export class PuterPeerConnection extends EventTarget {
// Either the new candidates take over or they do not, and
// only then is another attempt worth making.
if ( this.peerconnection.connectionState === 'failed' ) {
if ( this.peerconnection.connectionState !== 'connected' ) {
const backoff = Math.min(
RECOVERY_BACKOFF_MAX_MS,
RECOVERY_BACKOFF_MS * 2 ** (attempt - 1),
@@ -19,9 +19,13 @@ afterEach(() => {
vi.useRealTimers();
});
const makeConnection = async ({ connected = true } = {}) => {
/**
* A connection past its opening exchange. Impolite by default, since that is
* the side that restarts ICE; the polite side waits for its restarts.
*/
const makeConnection = async ({ connected = true, polite = false } = {}) => {
const channel = new LoopbackChannel('peer');
const conn = new PuterPeerConnection({ iceServers: [] }, { polite: true, channel });
const conn = new PuterPeerConnection({ iceServers: [] }, { polite, channel });
const pc = FakePeerConnection.instances.at(-1);
conn.acceptNegotiation();
@@ -121,6 +125,105 @@ describe('ICE recovery', () => {
expect(closes).toEqual(['the peer stopped responding']);
});
it('keeps one budget when an unanswered restart knocks the state back', async () => {
// Chrome reports 'disconnected' as soon as an unanswered restart offer
// is applied, and 'failed' again some 15s later. Each 'failed' must
// not start a fresh budget, or a vanished peer is never given up.
vi.useFakeTimers();
const { conn, pc, closes } = await makeConnection();
const restartIce = pc.restartIce.bind(pc);
pc.restartIce = () => {
restartIce();
queueMicrotask(() => pc.setConnectionState('disconnected'));
setTimeout(() => pc.setConnectionState('failed'), 15_000);
};
pc.setConnectionState('failed');
await vi.advanceTimersByTimeAsync(75_000);
expect(conn.closed).toBe(true);
expect(closes).toEqual(['the peer stopped responding']);
});
it('stays recovering while a restart knocks the state back', async () => {
vi.useFakeTimers();
const { conn, pc } = await makeConnection();
pc.setConnectionState('failed');
await vi.advanceTimersByTimeAsync(100);
pc.setConnectionState('disconnected');
await vi.advanceTimersByTimeAsync(100);
expect(conn.linkState).toBe('recovering');
});
it('counts the budget from the first sign of trouble, not from failure', async () => {
vi.useFakeTimers();
const { conn, pc, closes } = await makeConnection();
pc.setConnectionState('disconnected');
await vi.advanceTimersByTimeAsync(15_000);
pc.setConnectionState('failed');
await vi.advanceTimersByTimeAsync(46_000);
expect(conn.closed).toBe(true);
expect(closes).toEqual(['the peer stopped responding']);
});
it('gives up as soon as the app asked it to', async () => {
vi.useFakeTimers();
const channel = new LoopbackChannel('peer');
const conn = new PuterPeerConnection({ iceServers: [] }, { channel, recoveryTimeout: 20_000 });
const pc = FakePeerConnection.instances.at(-1);
conn.acceptNegotiation();
channel.onoffer({ type: 'offer', sdp: 'opening-offer' });
await flush();
pc.channels[0].open();
pc.setConnectionState('failed');
await vi.advanceTimersByTimeAsync(19_000);
expect(conn.closed).toBe(false);
await vi.advanceTimersByTimeAsync(2_000);
expect(conn.closed).toBe(true);
});
it('leaves restarting to the impolite side when it is the polite one', async () => {
// Restart offers from both sides, held up together by the outage,
// collide once it ends; the polite side rolling its own back leaves
// Chrome sending it no video. It answers the impolite side instead.
vi.useFakeTimers();
const { conn, pc, channel, closes } = await makeConnection({ polite: true });
const seen = [];
conn.addEventListener('linkstate', (e) => seen.push(e.state));
pc.setConnectionState('failed');
await vi.advanceTimersByTimeAsync(20_000);
expect(pc.restarts).toBe(0);
expect(seen).toEqual(['recovering']);
expect(conn.closed).toBe(false);
channel.onoffer({ type: 'offer', sdp: 'restart-offer' });
await flush();
pc.setConnectionState('connected');
await vi.advanceTimersByTimeAsync(100);
expect(conn.linkState).toBe('connected');
expect(closes).toEqual([]);
});
it('still gives the link up on the polite side once the budget is spent', async () => {
vi.useFakeTimers();
const { conn, pc, closes } = await makeConnection({ polite: true });
pc.setConnectionState('failed');
await vi.advanceTimersByTimeAsync(59_000);
expect(conn.closed).toBe(false);
await vi.advanceTimersByTimeAsync(3_000);
expect(pc.restarts).toBe(0);
expect(closes).toEqual(['could not restore the connection']);
});
it('waits for signalling rather than spending attempts it cannot send', async () => {
vi.useFakeTimers();
const { conn, pc, channel, closes } = await makeConnection();
@@ -155,18 +258,49 @@ describe('ICE recovery', () => {
expect(closes).toEqual(['the peer hung up']);
});
it('keeps recovering when the peer only lost its signalling session', async () => {
it('keeps recovering while a peer that lost its signalling might return', async () => {
vi.useFakeTimers();
const { conn, pc, channel, closes } = await makeConnection();
channel.onpeergone('the peer server’s signalling dropped', true);
pc.setConnectionState('failed');
await vi.advanceTimersByTimeAsync(10_000);
await vi.advanceTimersByTimeAsync(5_000);
expect(pc.restarts).toBeGreaterThan(0);
expect(conn.closed).toBe(false);
expect(closes).toEqual([]);
});
it('stops waiting on a peer whose signalling never comes back', async () => {
// Nothing it offers can be answered until that session returns, so
// spending the whole budget only holds up whoever is waiting on this
// connection to close before they act.
vi.useFakeTimers();
const { conn, pc, channel, closes } = await makeConnection();
channel.onpeergone('the peer server’s signalling dropped', true);
pc.setConnectionState('failed');
await vi.advanceTimersByTimeAsync(15_000);
expect(conn.closed).toBe(true);
expect(closes).toEqual(['the peer is no longer reachable']);
});
it('gets the full budget back when the peer reclaims its session', async () => {
vi.useFakeTimers();
const { conn, pc, channel, closes } = await makeConnection();
channel.onpeergone('the peer server’s signalling dropped', true);
pc.setConnectionState('failed');
await vi.advanceTimersByTimeAsync(5_000);
channel.onpeerback();
await vi.advanceTimersByTimeAsync(20_000);
expect(conn.closed).toBe(false);
expect(closes).toEqual([]);
expect(pc.restarts).toBeGreaterThan(1);
});
});
describe('signalling loss', () => {
@@ -142,7 +142,7 @@ export class PuterPeerServer extends EventTarget {
}
this.#resumeToken = reply.resumeToken ?? this.#resumeToken;
this.inviteCode = reply.invitecode ?? this.inviteCode;
return { inviteCode: this.inviteCode, resumed };
return { inviteCode: this.inviteCode, resumed, refused: reply.resumeRefused };
}
#scheduleReconnect () {
@@ -167,7 +167,11 @@ export class PuterPeerServer extends EventTarget {
}
this.#reconnectAttempts = 0;
this.dispatchEvent(
new PuterPeerServerReconnectEvent(registration.inviteCode, registration.resumed),
new PuterPeerServerReconnectEvent(
registration.inviteCode,
registration.resumed,
registration.refused,
),
);
}
@@ -238,6 +242,7 @@ export class PuterPeerServer extends EventTarget {
const connection = new PuterPeerConnection(this.#peerConfig, {
polite: true,
channel,
recoveryTimeout: this.#options.recoveryTimeout,
});
this.#channels.set(id, channel);
this.connections.set(id, connection);
@@ -325,3 +325,40 @@ describe('PuterPeerServer reclaiming a dropped session', () => {
expect(FakeWebSocket.latest).toBe(ws);
});
});
describe('a link already recovering when the session is lost', () => {
it('gives up as soon as the reclaim fails, not when its budget runs out', async () => {
// The order a network switch produces: ICE dies first, recovery is
// already under way, and only then does the socket come back under
// a session that cannot reach this client any more.
vi.useFakeTimers();
const server = await startServer();
const ws = FakeWebSocket.latest;
await ws.onmessage({
data: JSON.stringify({ server: { connect: { id: 'c1', user: {} } } }),
});
const conn = server.connections.get('c1');
const pc = FakePeerConnection.instances.at(-1);
pc.channels[0].open();
const closes = [];
conn.addEventListener('close', (e) => closes.push(e.reason));
// ICE goes first; recovery starts while the socket is still up.
pc.setConnectionState('failed');
await flushMicrotasks();
// Then the socket dies and comes back without the session.
ws.onclose({});
await vi.advanceTimersByTimeAsync(1_000);
await openSocket(FakeWebSocket.latest);
await answerCreate(FakeWebSocket.latest, {
success: true,
invitecode: 'invite-2',
resumeToken: 'tok-2',
resumed: false,
});
await vi.advanceTimersByTimeAsync(1_000);
expect(closes).toEqual(['the peer is no longer reachable']);
});
});
+4 -1
View File
@@ -16,10 +16,13 @@ export class PuterPeerServerConnectionEvent extends Event {
export class PuterPeerServerReconnectEvent extends Event {
inviteCode;
resumed;
constructor (inviteCode, resumed) {
/** Why a reclaim was refused, when one was asked for and turned down. */
refused;
constructor (inviteCode, resumed, refused) {
super('reconnect');
this.inviteCode = inviteCode;
this.resumed = resumed;
this.refused = refused;
}
}
+1 -1
View File
@@ -169,7 +169,7 @@ export class PeerModule extends PuterModule {
const peerConfig = await this.#resolvePeerConfig(options);
// Connecting is the impolite side: it makes the opening offer and
// keeps it when the two ends collide.
const conn = new PuterPeerConnection(peerConfig, { polite: false });
const conn = new PuterPeerConnection(peerConfig, { polite: false, recoveryTimeout: options?.recoveryTimeout });
await conn.connect(invitecode, options);
return conn;
}
@@ -33,6 +33,9 @@ export class SignallingChannel {
/** It came back, and anything waiting on it may go ahead. */
/** @type {() => void} */
onusable = () => {};
/** The peer reclaimed the session it dropped; it can answer again. */
/** @type {() => void} */
onpeerback = () => {};
/**
* Reads one relayed envelope and calls the handler for what it holds.
@@ -170,6 +173,10 @@ export class ClientSignallingChannel extends SignallingChannel {
}
return this.onrejected(new Error(msg.connect.error));
}
if ( msg.reconnect ) {
this.onpeerback();
return;
}
if ( msg.disconnect ) {
this.onpeergone(msg.disconnect.reason, !! msg.disconnect.resumable);
return;
+3
View File
@@ -10,6 +10,9 @@
* @property {string} [turnGrant] A grant from `puter.peer.createGuestGrant()`, letting a guest with
* no session use the Puter-managed relays on the granting account's allowance.
* @property {number} [port] Internal loopback port used by supported Puter environments.
* @property {number} [recoveryTimeout] How long, in milliseconds, a connection keeps trying to
* rescue a broken link before closing, counted from when the link first showed trouble. Defaults
* to 60000.
*/
/**