Tell a refused PDU apart from a session that failed (#82)

* Regression tests for telling a refused PDU from a failed session

* Export PduRefusedError so a refusal can be told from a session failure

* Document telling a refused PDU from a session failure

* Share the malformed-PDU fixtures and pin that a dead stream is no refusal

* Say once what a refusal is, and why the header type is published

* Fold the refused-response test into the one that already staged it

* Drop the refusal-rate claim the interop finding does not support

* Build a raw PDU one way, sequence number included

* Follow PduRefusedError into pdu-refusal.ts

test/raw-pdus.ts

@@ -0,0 +1,40 @@
+import assert from 'node:assert/strict';
+import type { PduObjectInput } from '../src/pdu.ts';
+import { objToPdu } from '../src/pdu.ts';
+
+/** The octets a test writes straight to a socket, which objToPdu builds for every valid PDU. */
+export function pduBytes(input: PduObjectInput): Buffer {
+	const { buffer } = objToPdu(input);
+
+	assert.ok(buffer);
+
+	return buffer;
+}
+
+/** The same, wearing a command id the command table defines nothing for. */
+export function withUnknownCmdId(input: PduObjectInput): Buffer {
+	const buffer = pduBytes(input);
+
+	buffer.writeUInt32BE(0x00010001, 4);
+
+	return buffer;
+}
+
+/** command_length honoured, so the stream stays in sync, with the declared body cut short. */
+export function shortened(input: PduObjectInput, octets: number): Buffer {
+	const buffer = pduBytes(input);
+	const cut = buffer.subarray(0, buffer.length - octets);
+
+	cut.writeUInt32BE(cut.length, 0);
+
+	return cut;
+}
+
+/** The same, with a message_state TLV declaring four octets of value and carrying one. */
+export function truncatedTlv(input: PduObjectInput): Buffer {
+	const appended = Buffer.concat([pduBytes(input), Buffer.from('0427000401', 'hex')]);
+
+	appended.writeUInt32BE(appended.length, 0);
+
+	return appended;
+}

test/readme.test.ts

@@ -6,6 +6,8 @@ import type { Sms } from '../src/sms.ts';
 import type { SmppLog } from '../src/log.ts';
 import type { SmppServer } from '../src/server.ts';
 import type { TestContext } from 'node:test';
+import { PduRefusedError } from '../src/pdu-refusal.ts';
+import { objToPdu } from '../src/pdu.ts';
 import { client } from '../src/client.ts';
 import { closeAfter } from './teardown.ts';
 import { server } from '../src/server.ts';
@@ -258,4 +260,52 @@ describe('README: Errors', () => {
 		assert.ok(err instanceof Error);
 		assert.equal(session, undefined);
 	});
+
+	test('telling a refused PDU from a session that failed', async t => {
+		const smpp = await answeringServer(t);
+		const bound = once<Session>(resolve => { smpp.on('session', resolve); });
+		const { err, session } = await client();
+		if (err) throw err;
+
+		closeAfter(t, session);
+
+		const warned: Record<string, boolean | number | string>[] = [];
+		const log: SmppLog = {
+			debug: () => undefined,
+			error: () => undefined,
+			info: () => undefined,
+			verbose: () => undefined,
+			warn: (msg, metadata) => { warned.push({ msg, ...metadata }); },
+		};
+
+		session.on('sessionError', err => {
+			if (err instanceof PduRefusedError) {
+				log.warn('the peer sent a PDU that could not be read', {
+					cmdName: err.header.cmdName ?? err.header.cmdId,
+					reason: err.reason,
+				});
+
+				return;
+			}
+
+			log.error('the session failed', { message: err.message });
+		});
+
+		const reported = once<Error>(resolve => { session.on('sessionError', resolve); });
+		const peer = await bound;
+		const { buffer } = objToPdu({ cmdName: 'enquire_link', seqNr: 5 });
+
+		assert.ok(buffer);
+		// A command id no command is defined for: refused on its own, with the link untouched.
+		buffer.writeUInt32BE(0x00010001, 4);
+		peer.sock.write(buffer);
+
+		await reported;
+
+		assert.deepEqual(warned, [{
+			cmdName: 0x00010001,
+			msg: 'the peer sent a PDU that could not be read',
+			reason: 'command',
+		}]);
+	});
 });

test/session-error.test.ts

@@ -0,0 +1,122 @@
+import assert from 'node:assert/strict';
+import test, { describe } from 'node:test';
+import type { PduHeader, Session } from '../src/index.ts';
+import type { TestContext } from 'node:test';
+import { PduRefusedError, client, server } from '../src/index.ts';
+import { closeAfter } from './teardown.ts';
+import { shortened, truncatedTlv, withUnknownCmdId } from './raw-pdus.ts';
+
+const receipt = {
+	destination_addr: '46709771337',
+	short_message: 'id:01a07342-a1d0-7479-88d6-c80082e19aec stat:DELIVRD err:000 text:',
+	source_addr: '46701113311',
+};
+
+function once<T>(register: (resolve: (value: T) => void) => void): Promise<T> {
+	return new Promise<T>(resolve => { register(resolve); });
+}
+
+function delay(ms: number): Promise<void> {
+	return new Promise(resolve => { setTimeout(resolve, ms); });
+}
+
+function raceWithin<T>(ms: number, promise: Promise<T>): Promise<T | false> {
+	return Promise.race([promise, delay(ms).then((): false => false)]);
+}
+
+/** Polls until the condition holds; false means it never did within the budget. */
+async function waitFor(condition: () => boolean, budget = 2000): Promise<boolean> {
+	const deadline = Date.now() + budget;
+
+	while (!condition()) {
+		if (Date.now() > deadline) return false;
+
+		await delay(5);
+	}
+
+	return true;
+}
+
+/** A bound client, and the server session at the far end whose socket the tests write from. */
+async function linked(t: TestContext, options: Parameters<typeof client>[0] = {}) {
+	const { err, server: smpp } = await server({ port: 0 });
+
+	assert.equal(err, undefined);
+	assert.ok(smpp);
+	closeAfter(t, smpp);
+
+	const accepted = once<Session>(resolve => { smpp.on('session', resolve); });
+	const { session } = await client({ port: smpp.port, ...options });
+
+	assert.ok(session);
+	closeAfter(t, session);
+
+	const peer = await raceWithin(2000, accepted);
+
+	assert.ok(peer, 'the server never accepted a session');
+
+	return { peer, session };
+}
+
+describe('telling a refused PDU from a failed session', () => {
+	test('narrows a refusal to its class, command and reason, from the entry point alone', async t => {
+		const { peer, session } = await linked(t);
+		const failed = once<Error>(resolve => { session.on('sessionError', resolve); });
+
+		peer.sock.write(truncatedTlv({ cmdName: 'deliver_sm', params: receipt, seqNr: 5 }));
+
+		const reported = await raceWithin(2000, failed);
+
+		assert.ok(reported instanceof PduRefusedError, 'a refusal narrows without a cast');
+
+		const header: PduHeader = reported.header;
+
+		assert.equal(header.cmdName, 'deliver_sm');
+		assert.equal(header.seqNr, 5);
+		assert.equal(reported.reason, 'tlvs');
+	});
+
+	test('names which part of the PDU it could not read, one refusal at a time', async t => {
+		const { peer, session } = await linked(t);
+		const seen: Error[] = [];
+
+		session.on('sessionError', err => { seen.push(err); });
+
+		peer.sock.write(withUnknownCmdId({ cmdName: 'enquire_link', seqNr: 9 }));
+		peer.sock.write(shortened({ cmdName: 'deliver_sm', params: receipt, seqNr: 6 }, 3));
+		peer.sock.write(truncatedTlv({ cmdName: 'deliver_sm', params: receipt, seqNr: 7 }));
+
+		assert.ok(await waitFor(() => seen.length === 3), 'one sessionError per refused PDU');
+		assert.deepEqual(
+			seen.map(err => (err instanceof PduRefusedError ? err.reason : err.message)),
+			['command', 'body', 'tlvs'],
+		);
+	});
+
+	test('reports a listener that threw as an error that is no refusal', async t => {
+		const { peer, session } = await linked(t, { responseTimeout: 200 });
+		const failed = once<Error>(resolve => { peer.on('sessionError', resolve); });
+
+		peer.on('sms', () => { throw new Error('listener exploded'); });
+
+		await session.sendSms({ from: '46701113311', message: 'blows the listener up', to: '46709771337' });
+
+		const reported = await raceWithin(2000, failed);
+
+		assert.ok(reported, 'the listener that threw never reached the session');
+		assert.ok(!(reported instanceof PduRefusedError), 'a session failure is not a refused PDU');
+		assert.equal(reported.message, 'listener exploded');
+	});
+
+	test('reports a socket the peer reset as an error that is no refusal', async t => {
+		const { peer, session } = await linked(t, { reconnect: false });
+		const failed = once<Error>(resolve => { session.on('sessionError', resolve); });
+
+		peer.sock.resetAndDestroy();
+
+		const reported = await raceWithin(2000, failed);
+
+		assert.ok(reported, 'a reset socket never reached the session');
+		assert.ok(!(reported instanceof PduRefusedError), 'a dead socket is not a refused PDU');
+	});
+});

test/session-extras.test.ts

@@ -20,12 +20,13 @@ import { LinkGate } from '../src/link-gate.ts';
 import { Reassembler, decodeSegments } from '../src/reassembly.ts';
 import { Session } from '../src/session.ts';
 import { DlrMerger } from '../src/dlr-merger.ts';
+import { PduRefusedError } from '../src/pdu-refusal.ts';
+import { objToPdu } from '../src/pdu.ts';
 import { checkSessionOptions } from '../src/session-options.ts';
 import { client } from '../src/client.ts';
 import { closeAfter, closeListenerAfter } from './teardown.ts';
 import { consts } from '../src/defs/constants.ts';
 import { errors } from '../src/defs/errors.ts';
-import { objToPdu } from '../src/pdu.ts';
 import { paramNumber, paramText } from '../src/defs/types.ts';
 import { server } from '../src/server.ts';
 import { silentLog } from '../src/log.ts';
@@ -499,14 +500,16 @@ describe('reconnect', () => {
 		const events: string[] = [];
 
 		session.on('close', () => { events.push('close'); });
-		session.on('sessionError', () => { events.push('sessionError'); });
+		session.on('sessionError', err => {
+			events.push(err instanceof PduRefusedError ? 'refused' : 'sessionError');
+		});
 
 		const reconnected = once<true>(resolve => { session.on('reconnected', () => { resolve(true); }); });
 
 		peerOf(smpp).sock.write(unreadablePdu);
 		await reconnected;
 
-		assert.deepEqual(events, ['sessionError']);
+		assert.deepEqual(events, ['sessionError'], 'an unframeable stream is the link dying, not one PDU refused');
 	});
 
 	test('refuses a backoff that would retry without pausing', () => {

test/session.test.ts

@@ -15,9 +15,11 @@ import { Session, bindCommands } from '../src/session.ts';
 import { client } from '../src/client.ts';
 import { closeAfter, closeListenerAfter } from './teardown.ts';
 import { consts } from '../src/defs/constants.ts';
+import { PduRefusedError } from '../src/pdu-refusal.ts';
 import { isCommand, objToPdu, pduReturn, pduToObj } from '../src/pdu.ts';
 import { paramText } from '../src/defs/types.ts';
 import { server } from '../src/server.ts';
+import { pduBytes, shortened, truncatedTlv, withUnknownCmdId } from './raw-pdus.ts';
 import { silentLog } from '../src/log.ts';
 import { splitMessage } from '../src/message.ts';
 
@@ -1241,42 +1243,6 @@ describe('a PDU the codec cannot read', () => {
 		source_addr: '46701113311',
 	};
 
-	/** The octets objToPdu built, wearing a sequence number it would refuse to write itself. */
-	function withSeqNr(input: PduObjectInput, seqNr: number): Buffer {
-		const { buffer } = objToPdu(input);
-
-		assert.ok(buffer);
-		buffer.writeUInt32BE(seqNr, 12);
-
-		return buffer;
-	}
-
-	/** command_length honoured, so the stream stays in sync, with the declared body cut short. */
-	function shortened(input: PduObjectInput, octets: number): Buffer {
-		const { buffer } = objToPdu(input);
-
-		assert.ok(buffer);
-
-		const cut = buffer.subarray(0, buffer.length - octets);
-
-		cut.writeUInt32BE(cut.length, 0);
-
-		return cut;
-	}
-
-	/** The same, with a message_state TLV declaring four octets of value and carrying one. */
-	function truncatedTlv(input: PduObjectInput): Buffer {
-		const { buffer } = objToPdu(input);
-
-		assert.ok(buffer);
-
-		const appended = Buffer.concat([buffer, Buffer.from('0427000401', 'hex')]);
-
-		appended.writeUInt32BE(appended.length, 0);
-
-		return appended;
-	}
-
 	/** The peer's next PDU within a budget: a dropped link must fail the test, not hang it. */
 	async function answerTo(peer: Peer): Promise<PduObject> {
 		const pduObj = await raceWithin(2000, peer.next());
@@ -1302,7 +1268,7 @@ describe('a PDU the codec cannot read', () => {
 		const { peer, session } = await bound(t);
 		const reported = once<Dlr>(resolve => { session.on('dlr', resolve); });
 
-		peer.writeRaw(withSeqNr({ cmdName: 'deliver_sm', params: receipt, seqNr: 1 }, 0x80000001));
+		peer.writeRaw(pduBytes({ cmdName: 'deliver_sm', params: receipt, seqNr: 0x80000001 }));
 
 		const answered = await answerTo(peer);
 
@@ -1315,7 +1281,7 @@ describe('a PDU the codec cannot read', () => {
 		assert.ok(dlr, 'the receipt is a report, not a reason to drop the link');
 		assert.equal(dlr.statusMsg, 'DELIVERED');
 
-		peer.writeRaw(withSeqNr({ cmdName: 'enquire_link', seqNr: 1 }, 0xFFFFFFFF));
+		peer.writeRaw(pduBytes({ cmdName: 'enquire_link', seqNr: 0xFFFFFFFF }));
 
 		const pinged = await answerTo(peer);
 
@@ -1326,10 +1292,8 @@ describe('a PDU the codec cannot read', () => {
 	test('answers an unknown command id with generic_nack ESME_RINVCMDID', async t => {
 		const { peer, session } = await bound(t);
 		const failed = once<Error>(resolve => { session.on('sessionError', resolve); });
-		const vendorSpecific = withSeqNr({ cmdName: 'enquire_link', seqNr: 1 }, 9);
 
-		vendorSpecific.writeUInt32BE(0x00010001, 4);
-		peer.writeRaw(vendorSpecific);
+		peer.writeRaw(withUnknownCmdId({ cmdName: 'enquire_link', seqNr: 9 }));
 
 		const answered = await answerTo(peer);
 
@@ -1354,7 +1318,7 @@ describe('a PDU the codec cannot read', () => {
 		assert.equal(reports, 0, 'a refused PDU is not a report');
 
 		// The regression this fixes: the link, and the stream's sync, outlive the refused PDU.
-		peer.writeRaw(withSeqNr({ cmdName: 'enquire_link', seqNr: 1 }, 78));
+		peer.writeRaw(pduBytes({ cmdName: 'enquire_link', seqNr: 78 }));
 		assert.equal((await answerTo(peer)).cmdName, 'enquire_link_resp');
 	});
 
@@ -1370,8 +1334,9 @@ describe('a PDU the codec cannot read', () => {
 		assert.equal(answered.seqNr, 6);
 	});
 
-	test('settles the request a response it could not read was answering', async t => {
+	test('settles the request a response it could not read was answering, and reports it', async t => {
 		const { peer, session } = await bound(t, { responseTimeout: 60000 });
+		const failed = once<Error>(resolve => { session.on('sessionError', resolve); });
 		const sending = session.sendSms({ from: '46701113311', message: 'hi', to: '46709771337' });
 		const submitted = await answerTo(peer);
 
@@ -1388,8 +1353,13 @@ describe('a PDU the codec cannot read', () => {
 		assert.ok(sent.err instanceof Error);
 		assert.equal(sent.unanswered, 1);
 
+		const reported = await raceWithin(2000, failed);
+
+		assert.ok(reported instanceof PduRefusedError);
+		assert.equal(reported.header.cmdName, 'submit_sm_resp');
+
 		// Nothing goes back: a response carries a sequence number of ours, not one of the peer's.
-		peer.writeRaw(withSeqNr({ cmdName: 'enquire_link', seqNr: 1 }, 77));
+		peer.writeRaw(pduBytes({ cmdName: 'enquire_link', seqNr: 77 }));
 		assert.equal((await answerTo(peer)).cmdName, 'enquire_link_resp');
 	});