/******************************************************************************** * Copyright (c) 2026 CrossBreeze, EclipseSource and others. * * This program and the accompanying materials are made available under the * terms of the MIT License which is available in the project root. * * SPDX-License-Identifier: MIT ********************************************************************************/ import { Emitter, type Event } from '@theia/core/lib/common/event'; import { SocketWriteBuffer } from '@theia/core/lib/common/messaging/socket-write-buffer'; import { type ConnectionResilienceOptions } from './connection-resilience-options'; /** * Prefix on every connection log line, deliberately shared by the browser and the * server so one pattern greps a console and a pod log at once. */ export const CONNECTION_LOG_PREFIX = '[connection]'; /** * The part of a socket this buffer uses. Narrower than the socket type Theia * declares, which keeps the dependency honest and lets the class be tested * without a connection. */ export interface MessageSink { /** `sequence` travels as a second socket.io argument, which a reader that does not expect it ignores. */ send(data: Uint8Array, sequence: number): void; } /** * Reported when the buffer runs out of room. The message that did not fit is * rejected, and so is every further one for as long as the buffer stays full. * Reconnecting drains it and sending resumes, but whatever was rejected * meanwhile is gone, so the two sides no longer agree about the session. */ export interface ConnectionBufferOverflow { /** Messages held when the limit was reached. */ readonly messages: number; /** Bytes held when the limit was reached. */ readonly bytes: number; /** The limit in force, so a reader can tell what to raise it to. */ readonly maxBytes: number; } /** * Holds messages while the peer is away and re-sends them when it returns. * * Theia copies them into one growing byte array and flushes that with a single * `send`. A websocket delivers whole messages rather than a byte stream, so * Theia's encoding adds no length prefix and every reader assumes one delivery * is one message. The concatenated flush breaks that assumption: the reader * decodes the first message and throws away the bytes behind it. * * Changed here: one entry per message, sent one at a time, so each arrives as * its own delivery. The buffer also decides whether a message may go out * directly — see {@link sendOrQueue} — because a caller that decides for itself * is how messages end up overtaking a backlog. * * Unchanged from Theia: exceeding the limit rejects the message rather than * dropping something silently. The limit is settable, since the right value * depends on how long an outage has to survive. * * Added here: every message sent is numbered and kept until the peer * acknowledges it, and {@link flush} resends what is still unacknowledged ahead * of the backlog. A socket whose far end died unnoticed still accepts writes, so * a message is not delivered just because it was sent. * * Logging goes to `console` rather than through a `Logger`. On the browser side * Theia's preloader builds this before any Output channel exists, so there is * nothing else to write to; the {@link CONNECTION_LOG_PREFIX} keeps the lines * findable on both sides. */ export class FramedSocketWriteBuffer extends SocketWriteBuffer { /** Theia's fixed limit, kept as the default. */ static readonly DEFAULT_MAX_BYTES = 100 * 1024; /** * Settable so a deployment can trade memory for longer outages. * * Consulted directly rather than through the base class's `maxBufferSize`, * which does not exist before Theia 1.71 — an `override` of it would not * compile against the oldest release this package supports. Nothing is lost: * `buffer`, `flush` and `drain` are all replaced here, so the base's own * limit is never reached. */ maxBytes = FramedSocketWriteBuffer.DEFAULT_MAX_BYTES; /** Kept copies past this are discarded, so a peer that never acknowledges cannot grow them without bound. */ static readonly DEFAULT_MAX_UNACKNOWLEDGED_BYTES = 1024 * 1024; /** * Separate from {@link maxBytes}, which counts only what is produced while * the peer is away: copies kept on a healthy connection must not use up the * room an outage needs. */ maxUnacknowledgedBytes = FramedSocketWriteBuffer.DEFAULT_MAX_UNACKNOWLEDGED_BYTES; protected pending: Uint8Array[] = []; protected pendingBytes = 0; protected overflowReported = false; protected readonly onOverflowEmitter = new Emitter(); protected nextSequence = 1; /** Oldest first: {@link acknowledge} stops at the first entry the peer has not accepted. */ protected unacknowledged: { readonly sequence: number; readonly data: Uint8Array }[] = []; protected unacknowledgedBytes = 0; protected unacknowledgedOverflowReported = false; /** * Whether the peer drops a message it already has. One without the hardening * applies a resent message a second time, so nothing is resent until the * peer has shown otherwise. */ protected peerDeduplicates = false; /** Fires once per outage when the limit is reached. */ get onOverflow(): Event { return this.onOverflowEmitter.event; } /** * Sends `data` now only if nothing is waiting, and queues it otherwise. * * Callers say whether the peer is ready for a message; the buffer decides * whether it may go ahead of one already waiting. Splitting it this way is * the point: a caller that answers both questions from one flag is how a * fresh message ends up overtaking a backlog. */ sendOrQueue(socket: MessageSink | undefined, data: Uint8Array): void { if (socket && !this.hasBacklog) { this.send(socket, data); } else { this.buffer(data); } } override buffer(data: Uint8Array): void { if (this.pendingBytes + data.byteLength > this.maxBytes) { this.reportOverflow(); throw new Error(`Max disconnected buffer size exceeded by adding ${data.byteLength} bytes`); } // Copied for the same reason the base class copies: the caller owns `data`. this.pending.push(data.slice()); this.pendingBytes += data.byteLength; if (this.pending.length === 1) { console.info(`${CONNECTION_LOG_PREFIX} buffering messages, the peer is disconnected`); } } /** * Kept only once `send` returns: a message whose send threw stays in the * backlog, and keeping it as well would send it again under a second number, * which the receiver cannot recognise as a duplicate. */ protected send(socket: MessageSink, data: Uint8Array): void { const sequence = this.nextSequence++; socket.send(data, sequence); this.keep(sequence, data); } protected keep(sequence: number, data: Uint8Array): void { if (this.unacknowledgedBytes + data.byteLength > this.maxUnacknowledgedBytes) { // Said only to a peer known to acknowledge: one without the hardening never does, and // nothing would be resent to it anyway. if (this.peerDeduplicates && !this.unacknowledgedOverflowReported) { this.unacknowledgedOverflowReported = true; console.warn( `${CONNECTION_LOG_PREFIX} the peer has not acknowledged ${this.unacknowledged.length} message(s), ` + `${this.unacknowledgedBytes} bytes; discarding their copies, so a connection dropped now loses them` ); } this.unacknowledged = []; this.unacknowledgedBytes = 0; return; } // Not copied: a committed write buffer hands over a fresh array and is disposed. this.unacknowledged.push({ sequence, data }); this.unacknowledgedBytes += data.byteLength; } /** * Records that the peer numbers its messages, which only a peer that also * drops duplicates does, so resending to it is safe. */ markPeerDeduplicates(): void { this.peerDeduplicates = true; } /** Releases every kept message numbered up to `sequence`, which the peer has accepted. */ acknowledge(sequence: number): void { this.peerDeduplicates = true; let index = 0; while (index < this.unacknowledged.length && this.unacknowledged[index].sequence <= sequence) { this.unacknowledgedBytes -= this.unacknowledged[index].data.byteLength; index++; } if (index > 0) { this.unacknowledged = this.unacknowledged.slice(index); } } /** * Resends what the peer has not acknowledged, under the same numbers, then * sends the backlog. Some of the resent messages may have arrived already; * the receiver recognises them by number and drops them. * * Resends only to a peer known to drop duplicates, and otherwise discards the * copies, leaving any message lost with the old socket lost. */ override flush(socket: MessageSink): void { if (this.peerDeduplicates) { for (const { sequence, data } of this.unacknowledged) { socket.send(data, sequence); } if (this.unacknowledged.length > 0) { console.info( `${CONNECTION_LOG_PREFIX} resent ${this.unacknowledged.length} unacknowledged message(s), ${this.unacknowledgedBytes} bytes` ); } } else { this.unacknowledged = []; this.unacknowledgedBytes = 0; } if (this.pending.length === 0) { return; } const count = this.pending.length; const bytes = this.pendingBytes; // Walked with an index and dropped in one slice, not shifted per message: a shift re-indexes // the whole array, so the drain would cost the square of a backlog `bufferBytes` invites an // adopter to enlarge. Leaving the queue in place is what keeps `hasBacklog` true until the // last message is out, and re-reading the length is what drains one appended by a nested send. let index = 0; try { while (index < this.pending.length) { const message = this.pending[index]; this.send(socket, message); index++; this.pendingBytes -= message.byteLength; } } catch (error: unknown) { // Said here because nothing retries: Theia flushes once, from a reconnect handler that has // deregistered by now, so the stall would otherwise surface much later as an overflow — // which names the wrong cause, the socket being up rather than absent. console.error( `${CONNECTION_LOG_PREFIX} send failed ${index} message(s) into a ${count}-message backlog; ` + `${this.pending.length - index} still queued and nothing will retry them`, error ); throw error; } finally { this.pending = this.pending.slice(index); } this.overflowReported = false; console.info(`${CONNECTION_LOG_PREFIX} sent ${count} buffered message(s), ${bytes} bytes`); } override drain(): void { if (this.pending.length > 0 || this.unacknowledged.length > 0) { console.warn( `${CONNECTION_LOG_PREFIX} discarded ${this.pending.length} buffered message(s), ${this.pendingBytes} bytes, ` + `and ${this.unacknowledged.length} sent but unacknowledged, ${this.unacknowledgedBytes} bytes` ); } this.reset(); } /** Whether messages are still waiting to go out. */ get hasBacklog(): boolean { return this.pending.length > 0; } protected reportOverflow(): void { if (this.overflowReported) { return; } this.overflowReported = true; const overflow: ConnectionBufferOverflow = { messages: this.pending.length, bytes: this.pendingBytes, maxBytes: this.maxBytes }; console.error( `${CONNECTION_LOG_PREFIX} buffer full at ${overflow.maxBytes} bytes after ${overflow.messages} message(s); ` + 'this message is rejected, and so is every further one until the peer returns' ); this.onOverflowEmitter.fire(overflow); } /** Also discards the kept copies: a drain ends the session, and resending them would replay it into the next one. */ protected reset(): void { this.pending = []; this.pendingBytes = 0; this.overflowReported = false; this.nextSequence = 1; this.unacknowledged = []; this.unacknowledgedBytes = 0; this.unacknowledgedOverflowReported = false; this.peerDeduplicates = false; } } /** Named in the warning a skipped install emits, so the reader is told what to upgrade to. */ export const REQUIRED_THEIA_HINT = '@theia/core 1.71 or newer'; /** * Whether this Theia exposes the seam the reconnect hardening replaces. * * Theia only began binding {@link SocketWriteBuffer} in 1.71; before that each * connection built one privately, so there is no binding to rebind and no way * in. Every other API involved has been there throughout, which makes this one * probe the whole compatibility question — and it is a binding check rather * than a version parse, so it answers about the container actually in front of * us instead of a number that may have been overridden or vendored. * * Callers that get `false` must leave Theia's own wiring alone: rebinding half * of it would be worse than not rebinding at all. */ export function supportsConnectionResilience(isBound: (identifier: typeof SocketWriteBuffer) => boolean): boolean { return isBound(SocketWriteBuffer); } /** * The warning a skipped install emits. Says what is lost rather than only what * is missing: the failures this guards against are silent, so an adopter who * never sees this line has no other way to learn the guard is absent. */ export function warnConnectionResilienceUnavailable(tier: 'frontend' | 'backend'): void { console.warn( `${CONNECTION_LOG_PREFIX} ${tier} reconnect hardening NOT installed: this Theia does not expose an injectable ` + `socket write buffer (needs ${REQUIRED_THEIA_HINT}). Theia's own behaviour is left in place, which can lose or ` + 'reorder messages when a socket drops and reconnects.' ); } /** Builds a buffer with the configured limits, or the defaults for any not configured. */ export function createFramedSocketWriteBuffer(options: ConnectionResilienceOptions = {}): FramedSocketWriteBuffer { const buffer = new FramedSocketWriteBuffer(); if (isPositive(options.bufferBytes)) { buffer.maxBytes = options.bufferBytes; } if (isPositive(options.unacknowledgedBytes)) { buffer.maxUnacknowledgedBytes = options.unacknowledgedBytes; } return buffer; } function isPositive(value: number | undefined): value is number { return typeof value === 'number' && value > 0; }