// Client WebSocket multiplexé — UNE seule connexion pour toute l'app. // Handshake hello/hello_ok, reconnexion avec backoff (ré-attache des terminaux // ouverts + re-sub), corrélation FIFO des 'attached', flow control par ACK // (cumul d'octets traités PAR CANAL, remis à zéro à chaque RESYNC). import { ref, type Ref } from 'vue'; import { BINARY_FRAME, FLOW, PROTOCOL_VERSION, decodeBinaryFrame, type ClientMessage, type ServerMessage, } from '@arboretum/shared'; export type WsStatus = 'idle' | 'connecting' | 'open' | 'reconnecting'; export type DetachReason = 'client' | 'session_exit' | 'replaced'; /** Interface minimale attendue côté terminal (implémentée par xterm). */ export interface TerminalSink { /** le callback est invoqué quand xterm a réellement traité le chunk (base du flow control) */ write(data: Uint8Array, callback?: () => void): void; /** reset complet avant le replay d'une frame RESYNC */ reset(): void; onDetached(reason: DetachReason): void; onControlChanged(controlling: boolean): void; } export type SessionEvent = Extract; export interface AttachOptions { sessionId: string; mode: 'interactive' | 'observer'; cols: number; rows: number; sink: TerminalSink; } const BACKOFF_MIN_MS = 500; const BACKOFF_MAX_MS = 10_000; export class Attachment { /** canal courant — réassigné à chaque reconnexion, -1 tant que non attaché */ channel = -1; controlling = false; closed = false; cols: number; rows: number; /** époque de resync : invalide les callbacks d'écritures encore en vol après un reset */ epoch = 0; /** octets de payload réellement traités par le terminal depuis le dernier resync */ processedBytes = 0; /** dernier cumul envoyé en ack (repart à 0 à chaque resync, comme côté serveur) */ lastAckBytes = 0; /** ACK traînant : garantit l'envoi du reliquat même quand le flux s'arrête (anti-deadlock) */ trailingAckTimer: ReturnType | null = null; /** résolution du premier 'attached' de ce canal (corrélation FIFO) */ pending: { resolve: (a: Attachment) => void; reject: (err: Error) => void } | null = null; constructor( private readonly client: WsClient, readonly sessionId: string, readonly mode: 'interactive' | 'observer', readonly sink: TerminalSink, cols: number, rows: number, ) { this.cols = cols; this.rows = rows; } sendStdin(data: string): void { if (this.closed || this.channel < 0) return; this.client.sendControl({ type: 'stdin', channel: this.channel, data }); } /** mémorise toujours la taille (réutilisée à la ré-attache) ; n'envoie que si controlling */ resize(cols: number, rows: number): void { this.cols = cols; this.rows = rows; if (this.closed || this.channel < 0 || !this.controlling) return; this.client.sendControl({ type: 'resize', channel: this.channel, cols, rows }); } detach(): void { if (this.closed) return; this.closed = true; this.pending = null; this.client.releaseAttachment(this); } } export class WsClient { readonly status: Ref = ref('idle'); private socket: WebSocket | null = null; private ready = false; private stopped = true; private backoffMs = BACKOFF_MIN_MS; private reconnectTimer: ReturnType | null = null; /** terminaux ouverts, ré-attachés à chaque reconnexion (ordre d'insertion = ordre FIFO) */ private readonly attachments = new Set(); private readonly byChannel = new Map(); /** FIFO des attaches en attente : le serveur répond aux 'attach' dans l'ordre */ private awaitingAttached: Attachment[] = []; private readonly sessionListeners = new Set<(e: SessionEvent) => void>(); connect(): void { this.stopped = false; if (this.socket || this.reconnectTimer) return; if (this.status.value === 'idle') this.status.value = 'connecting'; this.openSocket(); } disconnect(): void { this.stopped = true; this.clearReconnectTimer(); const socket = this.socket; this.socket = null; this.ready = false; this.status.value = 'idle'; for (const att of this.attachments) { att.closed = true; att.pending?.reject(new Error('disconnected')); att.pending = null; att.sink.onDetached('client'); } this.attachments.clear(); this.byChannel.clear(); this.awaitingAttached = []; socket?.close(); } /** résout sur le prochain 'attached' corrélé ; rejette si la session est introuvable/terminée */ attach(opts: AttachOptions): Promise { const att = new Attachment(this, opts.sessionId, opts.mode, opts.sink, opts.cols, opts.rows); const promise = new Promise((resolve, reject) => { att.pending = { resolve, reject }; }); this.attachments.add(att); this.connect(); if (this.ready) this.sendAttach(att); return promise; } subscribeSessions(listener: (e: SessionEvent) => void): () => void { this.sessionListeners.add(listener); this.connect(); if (this.sessionListeners.size === 1) this.sendControl({ type: 'sub', topics: ['sessions'] }); return () => { this.sessionListeners.delete(listener); if (this.sessionListeners.size === 0) this.sendControl({ type: 'sub', topics: [] }); }; } sendControl(msg: ClientMessage): void { if (!this.ready || this.socket?.readyState !== WebSocket.OPEN) return; this.socket.send(JSON.stringify(msg)); } /** détache côté serveur et oublie l'attache (appelé par Attachment.detach) */ releaseAttachment(att: Attachment): void { this.attachments.delete(att); if (att.channel >= 0) { this.byChannel.delete(att.channel); this.sendControl({ type: 'detach', channel: att.channel }); att.channel = -1; } // si l'attache attend encore son 'attached', elle reste dans la FIFO pour // préserver la corrélation : closed=true la fera détacher à la réception } // ---- interne ---- private openSocket(): void { const socket = new WebSocket(`${location.origin.replace(/^http/, 'ws')}/ws`); socket.binaryType = 'arraybuffer'; this.socket = socket; socket.onopen = () => { const hello: ClientMessage = { type: 'hello', protocol: PROTOCOL_VERSION }; socket.send(JSON.stringify(hello)); }; socket.onmessage = (ev: MessageEvent) => this.handleMessage(ev); socket.onclose = () => this.handleClose(socket); socket.onerror = () => { // onclose suit systématiquement : la reconnexion se joue là-bas }; } private handleClose(socket: WebSocket): void { if (this.socket !== socket) return; this.socket = null; this.ready = false; this.byChannel.clear(); this.awaitingAttached = []; for (const att of this.attachments) { att.channel = -1; att.epoch += 1; } if (this.stopped) { this.status.value = 'idle'; return; } this.status.value = 'reconnecting'; this.scheduleReconnect(); } private scheduleReconnect(): void { this.clearReconnectTimer(); this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; if (!this.stopped) this.openSocket(); }, this.backoffMs); this.backoffMs = Math.min(this.backoffMs * 2, BACKOFF_MAX_MS); } private clearReconnectTimer(): void { if (this.reconnectTimer !== null) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; } } private sendAttach(att: Attachment): void { this.awaitingAttached.push(att); this.sendControl({ type: 'attach', sessionId: att.sessionId, mode: att.mode, cols: att.cols, rows: att.rows }); } private handleMessage(ev: MessageEvent): void { if (ev.data instanceof ArrayBuffer) { this.handleBinary(new Uint8Array(ev.data)); return; } let msg: ServerMessage; try { msg = JSON.parse(String(ev.data)) as ServerMessage; } catch { return; } this.handleServerMessage(msg); } private handleBinary(data: Uint8Array): void { if (data.byteLength < BINARY_FRAME.HEADER_BYTES) return; const frame = decodeBinaryFrame(data); const att = this.byChannel.get(frame.channel); if (!att || att.closed) return; if (frame.type === BINARY_FRAME.RESYNC) { // le serveur repart de sentBytes=0 après un resync : reset terminal, replay // du ring, et compteurs d'octets remis à zéro (le payload du replay ne compte pas) att.epoch += 1; att.processedBytes = 0; att.lastAckBytes = 0; att.sink.reset(); if (frame.payload.byteLength > 0) att.sink.write(frame.payload); return; } if (frame.type !== BINARY_FRAME.OUTPUT) return; const epoch = att.epoch; const length = frame.payload.byteLength; att.sink.write(frame.payload, () => { // compté seulement si aucun resync/reconnexion n'est passé entre temps if (att.closed || att.epoch !== epoch || att.channel < 0) return; att.processedBytes += length; if (att.processedBytes - att.lastAckBytes >= FLOW.ACK_EVERY_BYTES) { att.lastAckBytes = att.processedBytes; this.sendControl({ type: 'ack', channel: att.channel, bytes: att.processedBytes }); } // ACK traînant débouncé : si le flux s'arrête avec un reliquat non-ACKé, // le serveur ne doit jamais rester en pause faute d'ACK suivant. if (att.trailingAckTimer) clearTimeout(att.trailingAckTimer); att.trailingAckTimer = setTimeout(() => { att.trailingAckTimer = null; if (att.closed || att.epoch !== epoch || att.channel < 0) return; if (att.processedBytes > att.lastAckBytes) { att.lastAckBytes = att.processedBytes; this.sendControl({ type: 'ack', channel: att.channel, bytes: att.processedBytes }); } }, 200); }); } private handleServerMessage(msg: ServerMessage): void { switch (msg.type) { case 'hello_ok': { this.ready = true; this.backoffMs = BACKOFF_MIN_MS; this.status.value = 'open'; if (this.sessionListeners.size > 0) this.sendControl({ type: 'sub', topics: ['sessions'] }); // ré-attache des terminaux encore ouverts (nouveaux canaux côté serveur) for (const att of this.attachments) { if (!att.closed) this.sendAttach(att); } return; } case 'attached': { const att = this.awaitingAttached.shift(); if (!att) return; if (att.closed) { // détaché pendant l'attente : on libère le canal côté serveur this.sendControl({ type: 'detach', channel: msg.channel }); return; } const wasControlling = att.controlling; att.channel = msg.channel; att.controlling = msg.controlling; this.byChannel.set(msg.channel, att); if (att.pending) { att.pending.resolve(att); att.pending = null; } else if (wasControlling !== msg.controlling) { // ré-attache après reconnexion : le statut controlling peut avoir changé att.sink.onControlChanged(msg.controlling); } return; } case 'detached': { const att = this.byChannel.get(msg.channel); if (!att) return; this.byChannel.delete(msg.channel); att.channel = -1; if (msg.reason === 'client') return; att.closed = true; this.attachments.delete(att); att.sink.onDetached(msg.reason); return; } case 'control_changed': { const att = this.byChannel.get(msg.channel); if (!att || att.closed) return; att.controlling = msg.controlling; att.sink.onControlChanged(msg.controlling); return; } case 'session_update': case 'session_exit': { for (const cb of this.sessionListeners) cb(msg); return; } case 'error': { if (msg.channel !== undefined) { console.warn(`[ws] channel ${msg.channel}: ${msg.code} — ${msg.message}`); return; } // un échec d'attach est la seule erreur sans canal corrélable à une requête if ((msg.code === 'NOT_FOUND' || msg.code === 'SESSION_EXITED') && this.awaitingAttached.length > 0) { const att = this.awaitingAttached.shift(); if (!att) return; this.attachments.delete(att); if (att.pending) { att.pending.reject(new Error(msg.code)); att.pending = null; } else if (!att.closed) { att.closed = true; att.sink.onDetached('session_exit'); } return; } console.warn(`[ws] ${msg.code} — ${msg.message}`); return; } case 'pong': return; } } } export const wsClient = new WsClient();