Refuse a single unreadable PDU instead of the whole link (#79)

* Regression tests: a PDU the codec cannot read costs that PDU, not the link

* Refuse a single unreadable PDU instead of the whole link

* smscsim interop asserts the first attempt gets every DLR

* Record the smscsim sequence number defect as fixed

* One framing rule and one response-command lookup, per the architecture review

* Apply the stability review's nits: honest sessionError docs and one link-survives assertion

src/outgoing-requests.ts

@@ -72,6 +72,11 @@ export class OutgoingRequests {
 		return this.pending.deliver(pduObj);
 	}
 
+	/** A response the codec refused settles its request instead of leaving it to time out. */
+	settleRefused(seqNr: number, err: Error): void {
+		this.pending.settle(seqNr, { err });
+	}
+
 	/** Sends a request and resolves with the peer's response. */
 	request(input: PduObjectInput, options: SendOptions): Promise<Result<{ pduObj: PduObject }>> {
 		// Ahead of the drain, so a misuse is named as one rather than blamed on the shutdown.

src/pdu-framer.ts

@@ -1,5 +1,5 @@
 import type { Result } from './result.ts';
-import { maxPduLength } from './pdu.ts';
+import { framingRefusal } from './pdu.ts';
 
 /**
  * Cuts a byte stream into whole PDUs.
@@ -31,10 +31,9 @@ export class PduFramer {
 
 		while (this.length >= 16) {
 			const cmdLength = this.join(16).readUInt32BE(0);
+			const unframable = framingRefusal(cmdLength);
 
-			if (cmdLength < 16 || cmdLength > maxPduLength) {
-				return { err: new Error(`Refusing a cmd_length of ${String(cmdLength)}`) };
-			}
+			if (unframable) return { err: unframable };
 
 			if (this.length < cmdLength) break;
 

src/pdu-transport.ts

@@ -3,7 +3,7 @@ import type { SmppLog } from './log.ts';
 import type { Socket } from 'node:net';
 import type { VoidResult } from './result.ts';
 import { PduFramer } from './pdu-framer.ts';
-import { pduToObj } from './pdu.ts';
+import { PduRefusedError, pduToObj } from './pdu.ts';
 
 export type PduTransportOptions = {
 	log: SmppLog;
@@ -14,6 +14,8 @@ export type PduTransportOptions = {
 	/** A complete PDU, before it is parsed. */
 	onFramed: (pdu: Buffer) => void;
 	onPdu: (pduObj: PduObject) => void;
+	/** A framed PDU the codec could not read. The stream is still in sync, so the link is not lost. */
+	onRefused: (refused: PduRefusedError) => void;
 	/** Nothing further can be read off this stream, whatever the socket does next. */
 	onUnreadable: (err: Error) => void;
 };
@@ -79,6 +81,17 @@ export class PduTransport {
 
 			const parsed = pduToObj(pdu);
 
+			if (parsed.err instanceof PduRefusedError) {
+				this.options.log.warn('transport - refusing a PDU it could not read', {
+					message: parsed.err.message,
+					reason: parsed.err.reason,
+				});
+				this.options.onRefused(parsed.err);
+
+				continue;
+			}
+
+			// The framer applies framingRefusal() first, so only a caller that skips it lands here.
 			if (parsed.err) {
 				this.options.log.warn('transport - could not parse an incoming PDU', {
 					message: parsed.err.message,

src/pdu.ts

@@ -11,12 +11,24 @@ import { errorNameById, errors, isErrorName } from './defs/errors.ts';
 import { paramNumber } from './defs/types.ts';
 import { tlvDefault, tlvs, tlvsById } from './defs/tlvs.ts';
 
-/** Sequence numbers are a 31-bit field; 0x7fffffff is reserved. */
+/** The highest sequence number this library hands out; SMPP 3.4 4.7.1 reserves 0x7fffffff. */
 export const maxSeqNr = 2147483646;
 
+/** What the field holds. Read and echoed in full, because peers do write above the spec's range. */
+const maxWireSeqNr = 0xFFFFFFFF;
+
 /** A hostile peer must not be able to make us allocate arbitrarily. */
 export const maxPduLength = 1024 * 1024;
 
+/** Why a command_length cannot frame a stream: past it nothing can say where the next PDU starts. */
+export function framingRefusal(cmdLength: number): Error | undefined {
+	if (cmdLength < 16 || cmdLength > maxPduLength) {
+		return new Error(`Refusing a cmd_length of ${String(cmdLength)}`);
+	}
+
+	return undefined;
+}
+
 export type TlvInput = {
 	/** Resolved from the record key; pass it for a tag the TLV table does not define. */
 	tagId?: number | undefined;
@@ -46,6 +58,57 @@ export type PduObject = {
 	tlvs: Record<string, Tlv>;
 };
 
+/** The 16 octets a framed PDU always has, whatever its body turns out to be. */
+export type PduHeader = {
+	cmdId: number;
+	cmdLength: number;
+	cmdName: CommandName | undefined;
+	cmdStatusId: number;
+	seqNr: number;
+};
+
+/** Which part of a PDU the codec could not read. */
+export type PduRefusalReason = 'body' | 'command' | 'tlvs';
+
+/** A PDU refused with the stream still in sync, so only this one PDU is lost. */
+export class PduRefusedError extends Error {
+	readonly header: PduHeader;
+	readonly reason: PduRefusalReason;
+
+	constructor(header: PduHeader, reason: PduRefusalReason, cause: Error) {
+		const named = header.cmdName ?? `command id ${String(header.cmdId)}`;
+
+		super(`Refused ${named} with seqNr ${String(header.seqNr)}: ${cause.message}`, { cause });
+		this.header = header;
+		this.name = 'PduRefusedError';
+		this.reason = reason;
+	}
+}
+
+// ESME_RINVTLVSTREAM is SMPP 5.0's name for 0xC0, which SMPP 3.4 spells ESME_RINVOPTPARSTREAM.
+const refusalStatus = {
+	body: 'ESME_RINVCMDLEN',
+	command: 'ESME_RINVCMDID',
+	tlvs: 'ESME_RINVTLVSTREAM',
+} as const satisfies Record<PduRefusalReason, ErrorName>;
+
+/** The response SMPP pairs with a request command, where it has one. */
+function respNameFor(cmdName: CommandName | undefined): CommandName | undefined {
+	if (cmdName === undefined) return undefined;
+
+	const respName = `${cmdName}_resp`;
+
+	return isCommandName(respName) ? respName : undefined;
+}
+
+/** SMPP 3.4 4.3: a PDU whose command has no response of its own is refused with generic_nack. */
+export function refusalAnswer(refused: PduRefusedError): { cmdName: CommandName; cmdStatus: ErrorName } {
+	return {
+		cmdName: respNameFor(refused.header.cmdName) ?? 'generic_nack',
+		cmdStatus: refusalStatus[refused.reason],
+	};
+}
+
 const respBit = 0x80000000;
 
 export function isResp(pduObj: Pick<PduObject, 'cmdId'>): boolean {
@@ -209,7 +272,7 @@ function buildPdu(
 		return { err: new Error(`Invalid cmdStatus: ${JSON.stringify(cmdStatus)}`) };
 	}
 
-	if (!Number.isInteger(seqNr) || seqNr < 0 || seqNr > maxSeqNr) {
+	if (!Number.isInteger(seqNr) || seqNr < 0 || seqNr > maxWireSeqNr) {
 		return { err: new Error(`Invalid seqNr: ${JSON.stringify(seqNr)}`) };
 	}
 
@@ -295,20 +358,24 @@ function readParams(
 	return { offset, params };
 }
 
-function parseOnce(pdu: Buffer, trailingNull: boolean): Result<{ aligned: boolean; pduObj: PduObject }> {
-	const cmdLength = pdu.readUInt32BE(0);
+function headerOf(pdu: Buffer): PduHeader {
 	const cmdId = pdu.readUInt32BE(4);
-	const cmdName = commandNameById(cmdId);
+
+	return {
+		cmdId,
+		cmdLength: pdu.readUInt32BE(0),
+		cmdName: commandNameById(cmdId),
+		cmdStatusId: pdu.readUInt32BE(8),
+		seqNr: pdu.readUInt32BE(12),
+	};
+}
+
+function parseOnce(pdu: Buffer, trailingNull: boolean): Result<{ aligned: boolean; pduObj: PduObject }> {
+	const header = headerOf(pdu);
+	const { cmdId, cmdLength, cmdName, cmdStatusId, seqNr } = header;
 
 	if (!cmdName) {
-		return { err: new Error(`Unknown PDU command id: ${String(cmdId)}`) };
-	}
-
-	const cmdStatusId = pdu.readUInt32BE(8);
-	const seqNr = pdu.readUInt32BE(12);
-
-	if (seqNr > maxSeqNr) {
-		return { err: new Error(`Invalid seqNr, exceeds ${String(maxSeqNr)}: ${String(seqNr)}`) };
+		return { err: new PduRefusedError(header, 'command', new Error('Unknown PDU command id')) };
 	}
 
 	// SMPP 3.4 4.4.2 and friends: a response with a non-zero status carries no body at all.
@@ -316,11 +383,11 @@ function parseOnce(pdu: Buffer, trailingNull: boolean): Result<{ aligned: boolea
 		? { offset: 16, params: {} }
 		: readParams(cmdName, pdu, trailingNull);
 
-	if (read.err) return { err: read.err };
+	if (read.err) return { err: new PduRefusedError(header, 'body', read.err) };
 
 	const parsed = parseTlvs(pdu, read.offset, cmdLength);
 
-	if (parsed.err) return { err: parsed.err };
+	if (parsed.err) return { err: new PduRefusedError(header, 'tlvs', parsed.err) };
 
 	const params = read.params;
 	const message = params.short_message;
@@ -352,10 +419,9 @@ function checkFraming(pdu: Buffer): VoidResult {
 	}
 
 	const cmdLength = pdu.readUInt32BE(0);
+	const unframable = framingRefusal(cmdLength);
 
-	if (cmdLength < 16 || cmdLength > maxPduLength) {
-		return { err: new Error(`Refusing a cmd_length of ${String(cmdLength)}`) };
-	}
+	if (unframable) return { err: unframable };
 
 	if (cmdLength > pdu.length) {
 		return { err: new Error(`cmd_length ${String(cmdLength)} exceeds the ${String(pdu.length)} octets given`) };
@@ -417,9 +483,9 @@ export function pduReturn(
 		return parsed.err ? { err: parsed.err } : pduReturn(parsed.pduObj, status, params, tlvs);
 	}
 
-	const respName = `${pdu.cmdName}_resp`;
+	const respName = respNameFor(pdu.cmdName);
 
-	if (!isCommandName(respName)) {
+	if (!respName) {
 		return { err: new Error(`"${pdu.cmdName}" has no response command`) };
 	}
 

src/session.ts

@@ -1,7 +1,7 @@
 import type { ErrorName } from './defs/errors.ts';
 import type { MessageDlr } from './dlr-merger.ts';
 import type { ParamValue } from './defs/types.ts';
-import type { PduObject, PduObjectInput, TlvInput } from './pdu.ts';
+import type { PduObject, PduObjectInput, PduRefusedError, TlvInput } from './pdu.ts';
 import type { BindType, CloseOptions, ReconnectOptions, SendOptions, SessionEvents, SessionOptions } from './session-options.ts';
 import type { Result, VoidResult } from './result.ts';
 import type { SendSmsOptions, SendSmsResult } from './send-sms.ts';
@@ -18,7 +18,7 @@ import { leftOf } from './idle-waiters.ts';
 import { errorFrom } from './error-from.ts';
 import { optionalParamsMinVersion } from './defs/constants.ts';
 import { bindCarries, bindCommands, defaultSystemId, defaults } from './session-options.ts';
-import { isResp, pduReturn } from './pdu.ts';
+import { isResp, objToPdu, pduReturn, refusalAnswer } from './pdu.ts';
 import { guardedLog } from './log.ts';
 import { submitSms, unsent } from './send-sms.ts';
 import { ConcatReference } from './udh.ts';
@@ -173,20 +173,25 @@ export class Session extends EventEmitter<SessionEvents> {
 		params: Record<string, ParamValue> = {},
 		tlvs?: Record<string, TlvInput>,
 	): Promise<VoidResult> {
-		const built = pduReturn(pdu, status, params, tlvs);
+		return Promise.resolve(
+			this.answer(pduReturn(pdu, status, params, tlvs), pdu.cmdName, pdu.seqNr),
+		);
+	}
+
+	private answer(built: Result<{ buffer: Buffer }>, cmdName: string, seqNr: number): VoidResult {
 		const sent = built.err ? { err: built.err } : this.transport.write(built.buffer);
 
 		// A peer that unbinds and drops the link takes our response with it; that is not a failure.
 		if (sent.err && !this.closed) {
 			this.log.warn('session - could not answer a request', {
-				cmdName: pdu.cmdName,
+				cmdName,
 				message: sent.err.message,
-				seqNr: pdu.seqNr,
+				seqNr,
 			});
 			this.emit('sessionError', sent.err);
 		}
 
-		return Promise.resolve(sent);
+		return sent;
 	}
 
 	async sendSms(sms: SendSmsOptions, options: SendOptions = {}): Promise<SendSmsResult> {
@@ -244,6 +249,7 @@ export class Session extends EventEmitter<SessionEvents> {
 			onError: err => { this.emit('sessionError', err); },
 			onFramed: pdu => { this.emit('incomingPdu', pdu); },
 			onPdu: pduObj => { this.dispatch(pduObj); },
+			onRefused: refused => { this.refuse(refused); },
 			onUnreadable: err => {
 				this.emit('sessionError', err);
 				this.teardown();
@@ -388,6 +394,22 @@ export class Session extends EventEmitter<SessionEvents> {
 		});
 	}
 
+	/** A PDU the codec refused. Its header parsed, so the peer gets an answer and the link stays. */
+	private refuse(refused: PduRefusedError): void {
+		const { cmdId, cmdName, seqNr } = refused.header;
+
+		this.emit('sessionError', refused);
+
+		// A response carries a sequence number of ours, so writing one back lands in the peer's space.
+		if (isResp(refused.header)) {
+			this.outgoing.settleRefused(seqNr, refused);
+
+			return;
+		}
+
+		this.answer(objToPdu({ ...refusalAnswer(refused), seqNr }), cmdName ?? String(cmdId), seqNr);
+	}
+
 	private resetTimers(): void {
 		if (this.closed) return;