P1 complete: web front, test suite, CI — acceptance ALL GREEN
Fan-out integration + fixes found by the test/acceptance pass: - FIX ring-buffer: chunks >= capacity skipped bytes now count into the monotonic offset (invariant: stream byte k lives at k % capacity) — window order was corrupted on unaligned big chunks - FIX auth: non-numeric cookie expiry no longer bypasses expiration - FIX protocol: safe-integer validation on ack.bytes / hello.protocol - FIX @fastify/websocket v11: websocket route must be registered in an encapsulated context after plugin load (handler got REST signature) - FIX flow-control deadlock found by e2e acceptance: client only ACKs on data receipt, so pausing with an unACKed residue in (LOW, ACK_EVERY] stalled both sides at 0.9 MB. ACK_EVERY now 64 KiB (<= LOW invariant, tested) + trailing debounced ACK in the web client - Web: Vue 3 + Vite + Pinia + Tailwind 4 + vue-i18n (EN/FR) + xterm 6 (fit + webgl fallback), multiplexed ws-client with reconnect/backoff and resync epochs - Tests: 100 vitest (protocol fuzz, ring edges, auth, pty-manager flow control with mocked pty, REST e2e) ; CI Node 22/24 + pack-smoke - scripts/acceptance-p1.mjs: real daemon + real WS client — boot, login, attach, stdin, 10 MB flood w/ ACK (13.7 MB/1.9s, RSS bounded), brutal disconnect + replay resync, kill broadcast, SIGTERM drain Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
38
packages/web/src/lib/api.ts
Normal file
38
packages/web/src/lib/api.ts
Normal file
@@ -0,0 +1,38 @@
|
||||
// Mini client REST : JSON, cookies de session, erreurs API normalisées.
|
||||
import type { ApiError as ApiErrorBody } from '@arboretum/shared';
|
||||
|
||||
export class ApiError extends Error {
|
||||
constructor(
|
||||
readonly status: number,
|
||||
readonly code: string,
|
||||
message: string,
|
||||
) {
|
||||
super(message);
|
||||
this.name = 'ApiError';
|
||||
}
|
||||
}
|
||||
|
||||
async function request<T>(path: string, method: string, body?: unknown): Promise<T> {
|
||||
const init: RequestInit = { method, credentials: 'same-origin' };
|
||||
if (body !== undefined) {
|
||||
init.headers = { 'content-type': 'application/json' };
|
||||
init.body = JSON.stringify(body);
|
||||
}
|
||||
const res = await fetch(path, init);
|
||||
if (!res.ok) {
|
||||
let parsed: ApiErrorBody | null = null;
|
||||
try {
|
||||
parsed = (await res.json()) as ApiErrorBody;
|
||||
} catch {
|
||||
// réponse non-JSON : on retombe sur le statusText
|
||||
}
|
||||
throw new ApiError(res.status, parsed?.error.code ?? 'UNKNOWN', parsed?.error.message ?? res.statusText);
|
||||
}
|
||||
return (await res.json()) as T;
|
||||
}
|
||||
|
||||
export const api = {
|
||||
get: <T>(path: string): Promise<T> => request<T>(path, 'GET'),
|
||||
post: <T>(path: string, body?: unknown): Promise<T> => request<T>(path, 'POST', body),
|
||||
delete: <T>(path: string): Promise<T> => request<T>(path, 'DELETE'),
|
||||
};
|
||||
370
packages/web/src/lib/ws-client.ts
Normal file
370
packages/web/src/lib/ws-client.ts
Normal file
@@ -0,0 +1,370 @@
|
||||
// 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<ServerMessage, { type: 'session_update' | 'session_exit' }>;
|
||||
|
||||
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<typeof setTimeout> | 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<WsStatus> = ref('idle');
|
||||
|
||||
private socket: WebSocket | null = null;
|
||||
private ready = false;
|
||||
private stopped = true;
|
||||
private backoffMs = BACKOFF_MIN_MS;
|
||||
private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
/** terminaux ouverts, ré-attachés à chaque reconnexion (ordre d'insertion = ordre FIFO) */
|
||||
private readonly attachments = new Set<Attachment>();
|
||||
private readonly byChannel = new Map<number, Attachment>();
|
||||
/** 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<Attachment> {
|
||||
const att = new Attachment(this, opts.sessionId, opts.mode, opts.sink, opts.cols, opts.rows);
|
||||
const promise = new Promise<Attachment>((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();
|
||||
Reference in New Issue
Block a user