Answer every segment of an inbound concatenated message as it arrives (#83)

* Regression tests for answering every inbound segment as it arrives

* Answer every segment of an inbound concatenated message as it arrives

* Record the segment-by-segment answer in AGENTS.md and README

* Regression tests for an empty message_id on deliver_sm_resp

* Answer a deliver_sm with the empty message_id SMPP 3.4 makes it

* Record Jasmin's refusal of a deliver_sm_resp message_id

* Mark the Jasmin multipart deadlock fixed

* Regression tests for the architecture review's findings

* Give the segment id notation an owner, and every segment a status

* Correct what the ids reach and what a lost group tells the application

* Regression tests for a group lost to its own octet overrun

* Report a group lost to its own overrun, and refuse by the command it arrived on

* Keep the docs true about what a segment is answered with

* Count only the answered segments of a group lost to an overrun

* Report only what a lost group cost, and say which cap bit

src/dlr-merger.ts

@@ -2,6 +2,7 @@ import type { Dlr } from './dlr.ts';
 import type { MessageState } from './defs/constants.ts';
 import type { SmppLog } from './log.ts';
 import { ExpiringGroups } from './expiring-groups.ts';
+import { parseSegmentId } from './sms-id.ts';
 
 export type MessageDlr = Dlr & { segments: Dlr[]; smsId: string };
 
@@ -18,8 +19,6 @@ type Group = {
 	parts: Map<number, Dlr>;
 };
 
-const numbered = /^(.*)-(\d+)$/;
-
 /**
  * MESSAGE_STATE is a flat enum, not a ranking — ACCEPTED is 6 where UNDELIVERABLE is 5 — so reducing
  * on the wire value reports a part-failed message as delivered. Rank it deliberately instead.
@@ -43,14 +42,12 @@ function idNumbering(smsIds: string[]): { base: string; parts: Set<number> } | u
 	const parts = new Set<number>();
 
 	for (const smsId of smsIds) {
-		const match = numbered.exec(smsId);
-		const base = match?.[1];
-		const part = match?.[2];
+		const numbering = parseSegmentId(smsId);
 
-		if (base === undefined || part === undefined) return undefined;
+		if (!numbering) return undefined;
 
-		bases.add(base);
-		parts.add(Number(part));
+		bases.add(numbering.base);
+		parts.add(numbering.part);
 	}
 
 	const [base] = bases;
@@ -110,18 +107,15 @@ export class DlrMerger {
 
 		if (dlr.intermediate || dlr.smsId === undefined) return undefined;
 
-		const match = numbered.exec(dlr.smsId);
-		const base = match?.[1];
-		const part = match?.[2];
+		const numbering = parseSegmentId(dlr.smsId);
 
-		if (base === undefined || part === undefined) return undefined;
+		if (!numbering) return undefined;
 
+		const { base, part: number } = numbering;
 		const group = this.groups.get(base);
 
 		if (!group) return undefined;
 
-		const number = Number(part);
-
 		if (!group.expected.has(number)) return undefined;
 
 		group.parts.set(number, dlr);
@@ -137,8 +131,8 @@ export class DlrMerger {
 	}
 
 	clear(): void {
-		this.groups.clear();
-		this.spent.clear();
+		this.groups.takeAll();
+		this.spent.takeAll();
 	}
 
 	/** Drops every group past its deadline. Runs before each collect and on its own timer. */

src/expiring-groups.ts

@@ -58,9 +58,18 @@ export class ExpiringGroups<T> {
 		this.idle();
 	}
 
-	clear(): void {
+	/** Removes every group and hands them over, so an owner that must account for them can. */
+	takeAll(): [string, T][] {
+		const taken: [string, T][] = [];
+
+		for (const [key, entry] of this.entries) {
+			taken.push([key, entry.group]);
+		}
+
 		this.entries.clear();
 		this.idle();
+
+		return taken;
 	}
 
 	/** Removes every group past its deadline and hands them over. */

src/held-messages.ts

@@ -76,7 +76,7 @@ export class HeldMessages {
 
 	/** Drops every message: their segments went with the link, so no answer of ours correlates now. */
 	clear(): void {
-		this.held.clear();
+		this.held.takeAll();
 		this.idleWaiters.settle();
 	}
 

src/incoming-requests.ts

@@ -1,4 +1,7 @@
+import type { CommandName } from './defs/commands.ts';
 import type { DlrMerger } from './dlr-merger.ts';
+import type { ErrorName } from './defs/errors.ts';
+import type { LostGroup, Refusal } from './reassembly.ts';
 import type { OnRequest } from './session-options.ts';
 import type { PduObject, PduObjectInput } from './pdu.ts';
 import type { Result, VoidResult } from './result.ts';
@@ -13,6 +16,20 @@ import { hasUdh } from './defs/constants.ts';
 import { createSms } from './sms.ts';
 import { dlrFromPdu } from './dlr.ts';
 import { paramNumber, paramText } from './defs/types.ts';
+import { respIdParams, segmentId } from './sms-id.ts';
+
+/** SMPP 3.4 lists ESME_RMSGQFUL under submit_sm_resp only; 4.6.2's retryable code is another. */
+export function refusedSegmentStatus(cmdName: CommandName, refusal: Refusal): ErrorName {
+	if (refusal === 'unplaceable') return 'ESME_RINVESMCLASS';
+
+	return cmdName === 'deliver_sm' ? 'ESME_RX_T_APPN' : 'ESME_RMSGQFUL';
+}
+
+const lostReasons: Record<LostGroup['reason'], string> = {
+	evicted: 'the reassembly buffer filled',
+	expired: 'no further segment arrived in time',
+	linkGone: 'the link they arrived on went',
+};
 
 export type IncomingRequestsOptions = {
 	dlrMerger: DlrMerger;
@@ -56,6 +73,7 @@ export class IncomingRequests {
 			log: options.log,
 			max: options.maxReassembly ?? defaults.maxReassembly,
 			maxOctets: options.maxOctets,
+			onLost: lost => { this.reportLost(lost); },
 			timeout: options.reassemblyTimeout ?? defaults.reassemblyTimeout,
 		});
 		this.sendPastDrain = options.sendPastDrain;
@@ -94,7 +112,7 @@ export class IncomingRequests {
 				await this.session.sendReturn(pduObj);
 				break;
 			case 'submit_sm':
-				this.onMessage(pduObj);
+				await this.onMessage(pduObj);
 				break;
 			case 'unbind':
 				await this.session.sendReturn(pduObj);
@@ -148,7 +166,7 @@ export class IncomingRequests {
 		const dlr = dlrFromPdu(pduObj, this.smsIdFormat);
 
 		if (!dlr) {
-			this.onMessage(pduObj);
+			await this.onMessage(pduObj);
 
 			return;
 		}
@@ -162,7 +180,11 @@ export class IncomingRequests {
 		await this.session.sendReturn(pduObj);
 	}
 
-	private onMessage(pduObj: PduObject): void {
+	/**
+	 * A concatenated message is answered segment by segment as it arrives: a peer that dispatches
+	 * one request at a time never sends the second segment until the first has been answered.
+	 */
+	private async onMessage(pduObj: PduObject): Promise<void> {
 		const message = pduObj.params.short_message;
 		const carriesUdh = hasUdh(paramNumber(pduObj.params.esm_class, 0));
 		const concat = carriesUdh && Buffer.isBuffer(message) ? concatInfo(message) : undefined;
@@ -173,12 +195,30 @@ export class IncomingRequests {
 			return;
 		}
 
-		const whole = this.reassembler.collect(pduObj, concat);
+		const collected = this.reassembler.collect(pduObj, concat);
 
-		if (whole) this.emitSms(whole);
+		if (!collected.kept) {
+			await this.session.sendReturn(pduObj, refusedSegmentStatus(pduObj.cmdName, collected.refusal));
+
+			return;
+		}
+
+		await this.session.sendReturn(
+			pduObj,
+			'ESME_ROK',
+			respIdParams(pduObj.cmdName, segmentId(collected.smsId, concat.part - 1, concat.total)),
+		);
+
+		if (collected.whole) this.emitSms(collected.whole, collected.smsId);
 	}
 
-	private emitSms(pduObjs: PduObject[]): void {
+	private reportLost(lost: LostGroup): void {
+		this.session.emit('sessionError', new Error(
+			`Gave up ${String(lost.parts)} of ${String(lost.total)} segments of an incomplete concatenated message: ${lostReasons[lost.reason]}`,
+		));
+	}
+
+	private emitSms(pduObjs: PduObject[], answeredAs?: string): void {
 		const first = pduObjs[0];
 
 		if (!first) return;
@@ -188,6 +228,7 @@ export class IncomingRequests {
 		const release = (): void => { setImmediate(() => { this.held.release(pduObjs); }); };
 
 		const sms = createSms({
+			answeredAs,
 			from: paramText(first.params.source_addr),
 			message: decodeSegments(pduObjs),
 			pduObjs,

src/reassembly.ts

@@ -6,21 +6,48 @@ import type { Tlv } from './defs/tlvs.ts';
 import { ExpiringGroups } from './expiring-groups.ts';
 import { decodeMessage } from './message.ts';
 import { paramNumber, paramText } from './defs/types.ts';
+import { uuidv7 } from './uuid.ts';
+
+/** A concatenated message given up on, whose segments the peer has already been answered for. */
+export type LostGroup = {
+	parts: number;
+	reason: 'evicted' | 'expired' | 'linkGone';
+	smsId: string;
+	total: number;
+};
 
 export type ReassemblerOptions = {
 	log: SmppLog;
 	max: number;
 	maxOctets?: number | undefined;
+	/** Injected so the ids a group is answered with can be read back in a test. */
+	newId?: (() => string) | undefined;
 	/** Injected so expiry can be exercised without a wall clock. */
 	now?: (() => number) | undefined;
+	onLost: (lost: LostGroup) => void;
 	timeout: number;
 };
 
+/** Why a segment was not kept: its header joins no message here, or the store had no room for it. */
+export type Refusal = 'full' | 'unplaceable';
+
+/** What a segment did to its group. */
+export type Collected =
+	| { kept: false; refusal: Refusal }
+	| {
+		kept: true;
+		/** The id base the group's segments are answered with. */
+		smsId: string;
+		/** Every segment in order, on the one that completes the message. */
+		whole?: PduObject[] | undefined;
+	};
+
 const defaultMaxOctets = 64 * 1024 * 1024;
 
 type Group = {
 	octets: number;
 	parts: Map<number, PduObject>;
+	smsId: string;
 	total: number;
 };
 
@@ -101,6 +128,8 @@ export class Reassembler {
 	private readonly log: SmppLog;
 	private readonly max: number;
 	private readonly maxOctets: number;
+	private readonly newId: () => string;
+	private readonly onLost: (lost: LostGroup) => void;
 	private octets = 0;
 
 	constructor(options: ReassemblerOptions) {
@@ -113,38 +142,22 @@ export class Reassembler {
 		this.log = options.log;
 		this.max = options.max;
 		this.maxOctets = options.maxOctets ?? defaultMaxOctets;
+		this.newId = options.newId ?? uuidv7;
+		this.onLost = options.onLost;
 	}
 
 	get size(): number {
 		return this.groups.size;
 	}
 
-	/** Every segment in order, on the one that completes the message; nothing while it is short. */
-	collect(pduObj: PduObject, concat: ConcatInfo): PduObject[] | undefined {
+	/** The group the segment joined, and always an answer for it: an unanswered one stalls a peer. */
+	collect(pduObj: PduObject, concat: ConcatInfo): Collected {
 		this.sweep();
 
-		if (concat.part < 1 || concat.total < 1 || concat.part > concat.total) {
-			this.log.warn('reassembler - dropping a segment the UDH numbers impossibly', {
-				part: concat.part,
-				total: concat.total,
-			});
-
-			return undefined;
-		}
-
 		const key = groupKey(pduObj, concat.reference);
 		const existing = this.groups.get(key);
 
-		// Parts 1/2 and 2/3 would otherwise complete the stored two-part group as a truncated message.
-		if (existing && existing.total !== concat.total) {
-			this.log.warn('reassembler - dropping a segment with an inconsistent UDH total', {
-				existingTotal: existing.total,
-				part: concat.part,
-				total: concat.total,
-			});
-
-			return undefined;
-		}
+		if (!this.placeable(concat, existing)) return { kept: false, refusal: 'unplaceable' };
 
 		const group = existing ?? this.open(key, concat.total);
 		const replaced = group.parts.get(concat.part);
@@ -156,34 +169,69 @@ export class Reassembler {
 		this.octets += delta;
 
 		if (group.parts.size < group.total) {
-			this.trim();
+			this.trim(key);
 
-			return undefined;
+			// Its own arrival overran the octet cap, so the peer keeps it rather than being told we did.
+			if (this.groups.get(key) !== group) return { kept: false, refusal: 'full' };
+
+			return { kept: true, smsId: group.smsId };
 		}
 
 		this.groups.delete(key);
 		this.octets -= group.octets;
 
-		return [...group.parts.entries()].sort(([a], [b]) => a - b).map(([, part]) => part);
+		return {
+			kept: true,
+			smsId: group.smsId,
+			whole: [...group.parts.entries()].sort(([a], [b]) => a - b).map(([, part]) => part),
+		};
 	}
 
 	clear(): void {
-		this.groups.clear();
+		for (const [, group] of this.groups.takeAll()) {
+			this.lost(group, 'linkGone');
+		}
+
 		this.octets = 0;
 	}
 
 	/** Drops every group past its deadline. Runs before each collect and on its own timer. */
 	sweep(): void {
-		for (const [key, group] of this.groups.takeExpired()) {
-			this.log.info('reassembler - incomplete message expired', { key, total: group.total });
+		for (const [, group] of this.groups.takeExpired()) {
 			this.octets -= group.octets;
+			this.lost(group, 'expired');
 		}
 	}
 
+	/** Whether a segment's UDH can join a group at all: its own numbering, and the group's total. */
+	private placeable(concat: ConcatInfo, existing: Group | undefined): boolean {
+		if (concat.part < 1 || concat.total < 1 || concat.part > concat.total) {
+			this.log.warn('reassembler - dropping a segment the UDH numbers impossibly', {
+				part: concat.part,
+				total: concat.total,
+			});
+
+			return false;
+		}
+
+		// Parts 1/2 and 2/3 would otherwise complete the stored two-part group as a truncated message.
+		if (existing && existing.total !== concat.total) {
+			this.log.warn('reassembler - dropping a segment with an inconsistent UDH total', {
+				existingTotal: existing.total,
+				part: concat.part,
+				total: concat.total,
+			});
+
+			return false;
+		}
+
+		return true;
+	}
+
 	private open(key: string, total: number): Group {
 		if (this.groups.full) this.dropOldest();
 
-		const group: Group = { octets: 0, parts: new Map(), total };
+		const group: Group = { octets: 0, parts: new Map(), smsId: this.newId(), total };
 
 		this.groups.set(key, group);
 
@@ -191,24 +239,43 @@ export class Reassembler {
 	}
 
 	/** Drops the oldest groups until the retained payload is back under the octet cap. */
-	private trim(): void {
-		while (this.octets > this.maxOctets && this.groups.size > 0) {
-			this.dropOldest();
+	private trim(current: string): void {
+		while (this.octets > this.maxOctets) {
+			const oldest = this.takeOldest();
+
+			if (!oldest) return;
+
+			// The refused segment is in the group but stays with the peer, so it is none of the loss.
+			const answered = oldest[0] === current ? oldest[1].parts.size - 1 : oldest[1].parts.size;
+
+			if (answered > 0) this.lost(oldest[1], 'evicted', answered);
 		}
 	}
 
-	private dropOldest(): void {
+	private takeOldest(): [string, Group] | undefined {
 		const oldest = this.groups.takeOldest();
 
-		if (!oldest) return;
+		if (oldest) this.octets -= oldest[1].octets;
 
-		const [, group] = oldest;
+		return oldest;
+	}
 
-		this.log.warn('reassembler - buffer full, dropping the oldest message', {
+	private dropOldest(): void {
+		const oldest = this.takeOldest();
+
+		if (oldest) this.lost(oldest[1], 'evicted');
+	}
+
+	/** Its segments are answered, so the peer will not send them again: this is traffic gone. */
+	private lost(group: Group, reason: LostGroup['reason'], parts = group.parts.size): void {
+		const lost: LostGroup = { parts, reason, smsId: group.smsId, total: group.total };
+
+		this.log.warn('reassembler - gave up a concatenated message', {
+			...lost,
 			max: this.max,
 			maxOctets: this.maxOctets,
 			octets: this.octets,
 		});
-		this.octets -= group.octets;
+		this.onLost(lost);
 	}
 }

src/session.ts

@@ -359,12 +359,16 @@ export class Session extends EventEmitter<SessionEvents> {
 		if (this.closed) return;
 
 		this.closed = true;
-		this.outgoing.linkLost(this.retrying());
+
+		// Read once: clear() reports lost segments, and a listener could stop the loop between reads.
+		const retrying = this.retrying();
+
+		this.outgoing.linkLost(retrying);
 		this.timers.clear();
 		this.incoming.clear();
 		this.sock.destroy();
 
-		if (this.retrying()) this.emit('disconnected');
+		if (retrying) this.emit('disconnected');
 		else this.emitClose();
 	}
 

src/sms-id.ts

@@ -1,3 +1,6 @@
+import type { CommandName } from './defs/commands.ts';
+import type { ParamValue } from './defs/types.ts';
+
 const notations = {
 	decimal: { digits: /^[0-9]+$/, prefix: '' },
 	hex: { digits: /^[0-9a-f]+$/i, prefix: '0x' },
@@ -33,3 +36,26 @@ export function normaliseSmsId(id: string, notation: SmsIdNotation | undefined):
 
 	return digits.test(id) ? BigInt(`${prefix}${id}`).toString(10) : id;
 }
+
+const numbered = /^(.*)-(\d+)$/;
+
+/** Each segment of a multipart message gets its own message_id, as a separate submit_sm must. */
+export function segmentId(smsId: string, index: number, total: number): string {
+	return total === 1 ? smsId : `${smsId}-${String(index + 1)}`;
+}
+
+/** The message and the part an id names, or nothing where `segmentId()` did not write it. */
+export function parseSegmentId(smsId: string): { base: string; part: number } | undefined {
+	const match = numbered.exec(smsId);
+	const base = match?.[1];
+	const part = match?.[2];
+
+	if (base === undefined || part === undefined) return undefined;
+
+	return { base, part: Number(part) };
+}
+
+/** SMPP 3.4 4.6.2 makes `deliver_sm_resp`'s `message_id` unused, and Jasmin FINs the link over one. */
+export function respIdParams(cmdName: CommandName, smsId: string): Record<string, ParamValue> {
+	return cmdName === 'deliver_sm' ? {} : { message_id: smsId };
+}

src/sms.ts

@@ -7,6 +7,7 @@ import { UnansweredError } from './unanswered-error.ts';
 import { consts } from './defs/constants.ts';
 import { receiptCodes, transientStates } from './dlr.ts';
 import { smppDate } from './message.ts';
+import { respIdParams, segmentId } from './sms-id.ts';
 import { uuidv7 } from './uuid.ts';
 
 /** `pduObjs` holds what the peer took, so a partial failure names what is already receipted. */
@@ -28,6 +29,11 @@ export type SendRespOptions = {
  * every segment's PDU.
  */
 export type Sms = {
+	/**
+	 * Whether the peer was answered as the message's segments arrived, which is what a concatenated
+	 * message needs and a segment count cannot tell you. `sendResp()` then writes nothing.
+	 */
+	answeredOnArrival: boolean;
 	dlr: boolean;
 	flash: boolean;
 	from: string;
@@ -35,16 +41,22 @@ export type Sms = {
 	pduObjs: PduObject[];
 	/** Sends a delivery report back to the sender. Defaults to DELIVERED. */
 	sendDlr: (status?: MessageState) => Promise<SendDlrResult>;
-	/** Answers every segment. Part of the protocol, not optional. Defaults to ESME_ROK. */
+	/**
+	 * Answers the message, and says the application is done with it. A concatenated message was
+	 * answered segment by segment as it arrived, so there it only releases a shutdown's wait and
+	 * refuses an `smsId` or a refusing `status`. Part of the protocol, not optional.
+	 */
 	sendResp: (options?: SendRespOptions) => Promise<VoidResult>;
 	session: Session;
-	/** The id `sendResp()` was given, or a generated UUID v7. */
+	/** The id the segments were answered with, the id `sendResp()` was given, or a generated UUID v7. */
 	readonly smsId: string;
 	submitTime: Date;
 	to: string;
 };
 
 export type SmsInput = {
+	/** The id base the segments were already answered with; absent leaves the answer to `sendResp()`. */
+	answeredAs?: string | undefined;
 	from: string;
 	message: string;
 	pduObjs: PduObject[];
@@ -59,25 +71,23 @@ export type SmsHandlers = {
 	send: (input: PduObjectInput) => Promise<Result<{ pduObj: PduObject }>>;
 };
 
-/** Each segment of a multipart message gets its own message_id, as a separate submit_sm must. */
-function segmentId(smsId: string, index: number, total: number): string {
-	return total === 1 ? smsId : `${smsId}-${String(index + 1)}`;
-}
-
 export function createSms(input: SmsInput, handlers: SmsHandlers): Sms {
 	const first = input.pduObjs[0];
 	const registered = first?.params.registered_delivery;
 	const dataCoding = first?.params.data_coding;
-	const answered = { smsId: uuidv7() };
+	const answered = { smsId: input.answeredAs ?? uuidv7() };
 
 	const sms: Sms = {
+		answeredOnArrival: input.answeredAs !== undefined,
 		dlr: typeof registered === 'number' && registered !== 0,
 		flash: typeof dataCoding === 'number' && (dataCoding & 0xF0) === 0x10,
 		from: input.from,
 		message: input.message,
 		pduObjs: input.pduObjs,
 		sendDlr: status => sendDlr(sms, handlers.send, status),
-		sendResp: options => sendResp(sms, answered, options ?? {}, handlers),
+		sendResp: options => (input.answeredAs === undefined
+			? sendResp(sms, answered, options ?? {}, handlers)
+			: alreadyAnswered(options ?? {}, handlers)),
 		session: input.session,
 		get smsId(): string {
 			return answered.smsId;
@@ -89,6 +99,28 @@ export function createSms(input: SmsInput, handlers: SmsHandlers): Sms {
 	return sms;
 }
 
+/** Every segment went out answered, so the call is what the shutdown waits for and nothing else. */
+function alreadyAnswered(
+	options: SendRespOptions,
+	handlers: Pick<SmsHandlers, 'onAnswered'>,
+): Promise<VoidResult> {
+	if (options.smsId !== undefined) {
+		return Promise.resolve({
+			err: new Error('This message\'s id was fixed when its first segment arrived; read sms.smsId'),
+		});
+	}
+
+	if (options.status !== undefined && options.status !== 'ESME_ROK') {
+		return Promise.resolve({
+			err: new Error('Its segments were answered as they arrived, so there is nothing left to refuse; refuse a segment from onRequest instead'),
+		});
+	}
+
+	handlers.onAnswered();
+
+	return Promise.resolve({});
+}
+
 async function sendResp(
 	sms: Sms,
 	answered: { smsId: string },
@@ -115,7 +147,7 @@ async function sendResp(
 	const results = await Promise.all(sms.pduObjs.map((pduObj, index) => sms.session.sendReturn(
 		pduObj,
 		options.status ?? 'ESME_ROK',
-		{ message_id: segmentId(answered.smsId, index, total) },
+		respIdParams(pduObj.cmdName, segmentId(answered.smsId, index, total)),
 	)));
 
 	const failure = results.find(result => result.err);