Retry a stream we cannot read, and cover the framer reset on attach
A framing or codec error tears the link down rather than the session, so the loop retries it on a fresh socket with a fresh framer — which is what a desynced stream needs. Removing that reset failed nothing before; the reconnect test now leaves half a PDU on the dying link, and does. `disconnected` counts failed links rather than outages, which the README now says, and the transport wires its socket as it is built.
This commit is contained in:
@@ -14,7 +14,7 @@ export type PduTransportOptions = {
|
||||
/** A complete PDU, before it is parsed. */
|
||||
onFramed: (pdu: Buffer) => void;
|
||||
onPdu: (pduObj: PduObject) => void;
|
||||
/** Nothing further can be read off this stream, whatever the socket does next. */
|
||||
/** Nothing further can be read off this stream; the socket has to go. */
|
||||
onUnreadable: (err: Error) => void;
|
||||
};
|
||||
|
||||
@@ -27,20 +27,23 @@ export class PduTransport {
|
||||
constructor(options: PduTransportOptions, sock: Socket) {
|
||||
this.options = options;
|
||||
this.socket = sock;
|
||||
this.wire(sock);
|
||||
}
|
||||
|
||||
get sock(): Socket {
|
||||
return this.socket;
|
||||
}
|
||||
|
||||
/** Wires a freshly opened socket in, replacing any previous one. */
|
||||
/** Takes over a freshly opened socket. Half a PDU left on the old one must not prefix this one. */
|
||||
attach(sock: Socket): void {
|
||||
// The socket being replaced is already dead, and its three handlers still point here.
|
||||
if (this.socket !== sock) this.socket.removeAllListeners();
|
||||
|
||||
this.socket.removeAllListeners();
|
||||
this.socket = sock;
|
||||
this.framer = new PduFramer();
|
||||
this.wire(sock);
|
||||
}
|
||||
|
||||
private wire(sock: Socket): void {
|
||||
sock.on('data', chunk => { this.read(chunk); });
|
||||
sock.on('close', () => { this.options.onClose(); });
|
||||
sock.on('error', err => {
|
||||
@@ -65,7 +68,7 @@ export class PduTransport {
|
||||
const framed = this.framer.next();
|
||||
|
||||
if (framed.err) {
|
||||
this.options.log.warn('transport - unusable stream, closing', { message: framed.err.message });
|
||||
this.options.log.warn('transport - unusable stream', { message: framed.err.message });
|
||||
this.options.onUnreadable(framed.err);
|
||||
|
||||
return;
|
||||
@@ -77,7 +80,7 @@ export class PduTransport {
|
||||
const parsed = pduToObj(pdu);
|
||||
|
||||
if (parsed.err) {
|
||||
this.options.log.warn('transport - could not parse an incoming PDU, closing', {
|
||||
this.options.log.warn('transport - could not parse an incoming PDU', {
|
||||
message: parsed.err.message,
|
||||
});
|
||||
this.options.onUnreadable(parsed.err);
|
||||
|
||||
@@ -163,9 +163,13 @@ function checkReconnect(reconnect: unknown): VoidResult {
|
||||
// A delay of 0 never doubles, so the backoff never starts and every retry lands at once.
|
||||
const checked = checkLimits([['maxDelay', maxDelay, 1], ['minDelay', minDelay, 1]]);
|
||||
|
||||
if (checked.err || maxDelay >= minDelay) return checked;
|
||||
if (checked.err) return checked;
|
||||
|
||||
return { err: new Error(`maxDelay must be minDelay or more, got ${String(maxDelay)}`) };
|
||||
if (maxDelay < minDelay) {
|
||||
return { err: new Error(`maxDelay must be minDelay (${String(minDelay)}) or more, got ${String(maxDelay)}`) };
|
||||
}
|
||||
|
||||
return {};
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
|
||||
+1
-3
@@ -136,7 +136,6 @@ export class Session extends EventEmitter<SessionEvents> {
|
||||
this.transport = this.transportFor(options.sock);
|
||||
this.window = new SendWindow(options.maxOutstanding ?? defaults.maxOutstanding);
|
||||
|
||||
this.attach(options.sock);
|
||||
this.resetTimers();
|
||||
}
|
||||
|
||||
@@ -263,7 +262,7 @@ export class Session extends EventEmitter<SessionEvents> {
|
||||
onPdu: pduObj => { this.dispatch(pduObj); },
|
||||
onUnreadable: err => {
|
||||
this.emit('sessionError', err);
|
||||
this.end();
|
||||
this.teardown();
|
||||
},
|
||||
}, sock);
|
||||
}
|
||||
@@ -375,7 +374,6 @@ export class Session extends EventEmitter<SessionEvents> {
|
||||
this.emitClose();
|
||||
}
|
||||
|
||||
/** A session torn down by a drop the loop was retrying reaches here with nothing left to tear down. */
|
||||
private emitClose(): void {
|
||||
if (this.ended) return;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user