From 5af75a3c57c98cf8dc97a3fb508394d4b634988a Mon Sep 17 00:00:00 2001 From: Lilleman auf Larv Date: Thu, 27 Aug 2026 00:17:48 +0200 Subject: [PATCH] Close the stability findings and settle the sendResp, sendSms and unbind contracts --- AGENTS.md | 2 + README.md | 21 +- src/client.ts | 72 ++++-- src/defs/types.ts | 55 ++++- src/dlr-merger.ts | 56 ++++- src/dlr.ts | 2 +- src/expiring-groups.ts | 102 ++++++++ src/message.ts | 8 +- src/pdu.ts | 13 +- src/reassembly.ts | 119 ++++++--- src/reconnect-loop.ts | 33 ++- src/send-sms.ts | 34 ++- src/server.ts | 35 ++- src/session-options.ts | 53 ++++ src/session.ts | 93 +++---- src/sms.ts | 35 ++- test/message.test.ts | 5 + test/pdu.test.ts | 51 ++++ test/session-extras.test.ts | 147 ++++++++++- test/session.test.ts | 468 +++++++++++++++++++++++++++++++++++- test/tls.test.ts | 38 +-- test/types.test.ts | 20 ++ todo.md | 2 +- 23 files changed, 1262 insertions(+), 202 deletions(-) create mode 100644 src/expiring-groups.ts create mode 100644 src/session-options.ts diff --git a/AGENTS.md b/AGENTS.md index 4c81610..fafafe9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -42,6 +42,7 @@ src/ sms.ts The live handle emitted as the 'sms' event (sendResp/sendDlr) dlr.ts Delivery receipts: text and TLV parsing, receipt status codes dlr-merger.ts DlrMerger: per-segment receipts counted into one MessageDlr + expiring-groups.ts ExpiringGroups: the capped, expiring store both of those share link-timers.ts LinkTimers: the enquire_link heartbeat and the idle timeout log.ts silentLog — the default when the application passes none message.ts Encoding detection, splitting, bit counting, SMPP date formatting @@ -53,6 +54,7 @@ src/ result.ts Result — the shape every fallible call returns send-sms.ts submitSms composition and the submitSmParams builder send-window.ts SendWindow: the maxOutstanding semaphore + session-options.ts SessionOptions, ReconnectOptions and the session defaults udh.ts User data header: the concatenation fields of a long SMS uuid.ts uuidv7() — the ids the library generates for messages defs/ diff --git a/README.md b/README.md index 511f822..499a232 100644 --- a/README.md +++ b/README.md @@ -81,9 +81,10 @@ Every one is optional. | `systemType`, `addressRange`, `addrTon`, `addrNpi` | `''`, `''`, `0`, `0` | The remaining bind fields, for operators that require them. | | `tls` | `false` | `true` for defaults, or a `tls.ConnectionOptions` object for a private CA or a client certificate. | | `enquireLinkInterval` | `20000` | How often to send `enquire_link` on a quiet link. | +| `idleTimeout` | `2 × enquireLinkInterval` | Give up on a link the peer has stopped answering; with `reconnect` set, it re-binds. | | `responseTimeout` | `30000` | How long to wait for a response before giving up on it. | | `maxOutstanding` | `10` | Requests allowed on the wire at once; further sends queue. | -| `reconnect` | off | `{ minDelay, maxDelay }` to re-bind automatically after a drop, with exponential backoff. | +| `reconnect` | off | `{ minDelay, maxDelay }` to re-bind automatically after a drop or an idle timeout, with exponential backoff. | | `log` | silent | A `@larvit/log` instance. | | `signal` | — | An `AbortSignal` that cancels connecting and tears the session down. | @@ -109,6 +110,11 @@ one id per segment: const { err, pduObjs, smsIds } = await session.sendSms({ from, message, to }); ``` +`err` is set when the SMSC refuses a segment, and it names the status it refused with. Because every +segment goes on the wire together, `smsIds` then holds the ids of the segments the SMSC did accept — +retry only what is missing from it. A message needing more than 255 segments is refused before +anything is sent, since the concatenation header numbers segments in a single octet. + ## Server The simplest possible server — no authentication, listening on port 2775: @@ -143,9 +149,10 @@ if (err) throw err; smpp.on('session', session => { session.on('sms', async sms => { - // Responding is part of the protocol, not optional. - // Defaults to ESME_ROK; see the SMPP spec for the other status codes. + // Responding is part of the protocol, not optional. Without arguments it answers + // ESME_ROK with a generated id; pass your own, and a status to refuse the message. await sms.sendResp(); + // await sms.sendResp({ smsId: yourOwnId, status: 'ESME_RMSGQFUL' }); if (sms.dlr) { await sms.sendDlr(); // same as sms.sendDlr('DELIVERED') @@ -196,12 +203,12 @@ is exactly what this library promises not to do. | Event | Fires when | | --- | --- | -| `sms` | An SMS arrives, reassembled if it was multipart. Carries `sendResp()` and `sendDlr()`. | +| `sms` | An SMS arrives, reassembled if it was multipart. Carries `sendResp()`, `sendDlr()` and the `smsId` it was answered with. | | `dlr` | A delivery report arrives, one per segment. | -| `messageDlr` | Every segment of a multipart message has been reported on. | +| `messageDlr` | Every segment of a multipart message sent with `dlr: true` has been reported on. | | `close` | The connection closed. | | `reconnected` | The client re-bound after a drop (only with `reconnect` configured). | -| `sessionError` | Something failed on a live session. | +| `sessionError` | Something failed on a live session, including a hook or listener that threw. | | `data` | Raw bytes arrived on the socket. | | `incomingPdu` | A complete PDU arrived, as a buffer. | | `incomingPduObj` | The same PDU, parsed into an object. | @@ -248,6 +255,8 @@ The spec tables are exported both individually (`cmds`, `consts`, `encodings`, ` resolving to a result object with an optional `err`. Nothing rejects. - **`server()` resolves once, when it is listening**, and gives you a handle with `close()`, `port` and a `session` event. It no longer calls your callback once per incoming connection. +- **The id a message is answered with goes to `sendResp({ smsId })`**, and `sms.smsId` is read-only. + Assigning it no longer works, so set the id where the response is sent rather than before it. - **`checkuserpass` is now `authenticate`**, takes `{ password, session, systemId, systemType }` and returns `false` or `{ userData }`. - **Renamed options:** `enqLinkTiming` → `enquireLinkInterval`, server `timeout` → `idleTimeout`. diff --git a/src/client.ts b/src/client.ts index 9f951c1..f418726 100644 --- a/src/client.ts +++ b/src/client.ts @@ -17,6 +17,7 @@ export type ClientOptions = { bindType?: BindType; enquireLinkInterval?: number; host?: string; + idleTimeout?: number; interfaceVersion?: number; log?: LogInt; maxOutstanding?: number; @@ -34,16 +35,14 @@ const defaults = { bindType: 'transceiver', enquireLinkInterval: 20_000, host: 'localhost', + /** The idle timeout is what notices a dead link, so it has to outlast one silent probe. */ + idleTimeoutFactor: 2, interfaceVersion: defaultInterfaceVersion, password: 'pass', port: 2775, username: 'user', } as const; -/** - * Opens the socket. 0.4.0 built a bare `new tls.Socket()` for `tls: true`, which never performs a - * handshake, so those connections were not encrypted at all. - */ function openSocket(options: ClientOptions): Promise> { const host = options.host ?? defaults.host; const port = options.port ?? defaults.port; @@ -59,7 +58,15 @@ function openSocket(options: ClientOptions): Promise> { return; } - const sock = secure ? tlsConnect({ host, port, ...tlsOptions }) : netConnect({ host, port }); + let sock: Socket; + + try { + sock = secure ? tlsConnect({ host, port, ...tlsOptions }) : netConnect({ host, port }); + } catch (thrown: unknown) { + resolve({ err: thrown instanceof Error ? thrown : new Error(String(thrown)) }); + + return; + } const settle = (result: Result<{ sock: Socket }>): void => { sock.removeListener('error', onError); @@ -122,6 +129,29 @@ async function bind(session: Session, options: ClientOptions): Promise openSocket(options), + maxDelay: options.reconnect.maxDelay, + minDelay: options.reconnect.minDelay, + onConnected: reconnected => bind(reconnected, options), + }, + } + : {}), + }); +} + /** Connects to an SMSC and binds. */ export async function client(options: ClientOptions = {}): Promise> { const log = options.log ?? silentLog; @@ -137,23 +167,17 @@ export async function client(options: ClientOptions = {}): Promise openSocket(options), - maxDelay: options.reconnect.maxDelay, - minDelay: options.reconnect.minDelay, - onConnected: reconnected => bind(reconnected, options), - }, - } - : {}), - }); + const session = createSession(options, log, opened.sock); + const signal = options.signal; + + if (signal?.aborted === true) { + session.close(); + + return { err: new Error('Aborted before binding') }; + } + + // Registered before the bind: an abort landing while it is in flight has to close the session. + signal?.addEventListener('abort', () => { session.close(); }, { once: true }); const bound = await bind(session, options); @@ -163,9 +187,5 @@ export async function client(options: ClientOptions = {}): Promise { session.close(); }, { once: true }); - } - return { session }; } diff --git a/src/defs/types.ts b/src/defs/types.ts index b029e07..4617da4 100644 --- a/src/defs/types.ts +++ b/src/defs/types.ts @@ -55,6 +55,26 @@ function wantInt(value: ParamValue, max: number): Result<{ int: number }> { return { int: value }; } +function writeInt8(value: ParamValue, buf: Buffer, offset: number): VoidResult { + const { err, int } = wantInt(value, 0xFF); + + if (err) return { err }; + + buf.writeUInt8(int, offset); + + return {}; +} + +function writeInt32(value: ParamValue, buf: Buffer, offset: number): VoidResult { + const { err, int } = wantInt(value, 0xFFFFFFFF); + + if (err) return { err }; + + buf.writeUInt32BE(int, offset); + + return {}; +} + function wantText(value: ParamValue): Result<{ text: string }> { if (typeof value === 'string') return { text: value }; if (typeof value === 'number') return { text: value.toString() }; @@ -323,7 +343,9 @@ export const dest_address_array: WireType = { if (rangeErr) return { err: rangeErr }; - buf.writeUInt8(addresses.length, offset++); + const count = writeInt8(addresses.length, buf, offset++); + + if (count.err) return { err: count.err }; for (const dest of addresses) { if ('dl_name' in dest) { @@ -332,8 +354,15 @@ export const dest_address_array: WireType = { offset += dest.dl_name.length + 1; } else { buf.writeUInt8(1, offset++); - buf.writeUInt8(dest.dest_addr_ton, offset++); - buf.writeUInt8(dest.dest_addr_npi, offset++); + + const ton = writeInt8(dest.dest_addr_ton, buf, offset++); + + if (ton.err) return { err: ton.err }; + + const npi = writeInt8(dest.dest_addr_npi, buf, offset++); + + if (npi.err) return { err: npi.err }; + writeCstring(dest.destination_addr, buf, offset); offset += dest.destination_addr.length + 1; } @@ -406,14 +435,26 @@ export const unsuccess_sme_array: WireType = { if (rangeErr) return { err: rangeErr }; - buf.writeUInt8(smes.length, offset++); + const count = writeInt8(smes.length, buf, offset++); + + if (count.err) return { err: count.err }; for (const sme of smes) { - buf.writeUInt8(sme.dest_addr_ton, offset++); - buf.writeUInt8(sme.dest_addr_npi, offset++); + const ton = writeInt8(sme.dest_addr_ton, buf, offset++); + + if (ton.err) return { err: ton.err }; + + const npi = writeInt8(sme.dest_addr_npi, buf, offset++); + + if (npi.err) return { err: npi.err }; + writeCstring(sme.destination_addr, buf, offset); offset += sme.destination_addr.length + 1; - buf.writeUInt32BE(sme.error_status_code, offset); + + const status = writeInt32(sme.error_status_code, buf, offset); + + if (status.err) return { err: status.err }; + offset += 4; } diff --git a/src/dlr-merger.ts b/src/dlr-merger.ts index 2ed09ab..4a848e8 100644 --- a/src/dlr-merger.ts +++ b/src/dlr-merger.ts @@ -1,7 +1,17 @@ import type { Dlr } from './dlr.ts'; +import type { LogInt } from '@larvit/log'; +import { ExpiringGroups } from './expiring-groups.ts'; export type MessageDlr = Dlr & { segments: Dlr[] }; +export type DlrMergerOptions = { + log: LogInt; + max: number; + /** Injected so expiry can be exercised without a wall clock. */ + now?: (() => number) | undefined; + timeout: number; +}; + type Group = { expected: number; parts: Map; @@ -15,7 +25,24 @@ const numbered = /^(.*)-(\d+)$/; * SMSC that hands out unrelated ids per segment cannot be merged, so nothing is reported for it. */ export class DlrMerger { - private readonly groups = new Map(); + private readonly groups: ExpiringGroups; + private readonly log: LogInt; + private readonly max: number; + + constructor(options: DlrMergerOptions) { + this.groups = new ExpiringGroups({ + max: options.max, + now: options.now, + onSweep: () => { this.sweep(); }, + timeout: options.timeout, + }); + this.log = options.log; + this.max = options.max; + } + + get size(): number { + return this.groups.size; + } /** Registers the ids one multipart send got back, so their receipts can be merged. */ expect(smsIds: string[]): void { @@ -34,12 +61,14 @@ export class DlrMerger { if (bases.size !== 1) return; for (const base of bases) { - this.groups.set(base, { expected: smsIds.length, parts: new Map() }); + this.open(base, smsIds.length); } } /** The whole message's report, on the receipt that completes it. */ collect(dlr: Dlr): MessageDlr | undefined { + this.sweep(); + const match = numbered.exec(dlr.smsId); const base = match?.[1]; const part = match?.[2]; @@ -61,4 +90,27 @@ export class DlrMerger { return { ...worst, segments, smsId: base }; } + + clear(): void { + this.groups.clear(); + } + + /** Drops every group past its deadline. Runs before each collect and on its own timer. */ + sweep(): void { + for (const [base, group] of this.groups.takeExpired()) { + this.log.info('dlrMerger - incomplete receipts expired', { base, expected: group.expected }); + } + } + + private open(base: string, expected: number): void { + if (this.groups.full) this.dropOldest(); + + this.groups.set(base, { expected, parts: new Map() }); + } + + private dropOldest(): void { + if (!this.groups.takeOldest()) return; + + this.log.warn('dlrMerger - buffer full, dropping the oldest message', { max: this.max }); + } } diff --git a/src/dlr.ts b/src/dlr.ts index 19b63bb..61b807f 100644 --- a/src/dlr.ts +++ b/src/dlr.ts @@ -115,7 +115,7 @@ export function parseReceipt(message: string): Receipt { /** * Builds a delivery report from a deliver_sm. The message_state and receipted_message_id TLVs are * authoritative when present; otherwise the receipt text is parsed, which is the only thing Kannel - * and several other SMSCs send. 0.4.0 required the TLVs and rejected everything else. + * and several other SMSCs send. */ export function dlrFromPdu(pduObj: PduObject): Dlr | undefined { const message = pduObj.params.short_message; diff --git a/src/expiring-groups.ts b/src/expiring-groups.ts new file mode 100644 index 0000000..584a1a2 --- /dev/null +++ b/src/expiring-groups.ts @@ -0,0 +1,102 @@ +export type ExpiringGroupsOptions = { + max: number; + /** Injected so expiry can be exercised without a wall clock. */ + now?: (() => number) | undefined; + /** Runs on the sweeper's own timer; the owner reports whatever it takes out. */ + onSweep: () => void; + timeout: number; +}; + +type Entry = { + deadline: number; + group: T; +}; + +/** + * A capped store of groups that expire. Nothing is dropped silently: the owner takes the expired + * and the evicted out itself, so the accounting and the log line stay where the group is understood. + */ +export class ExpiringGroups { + private readonly entries = new Map>(); + private readonly max: number; + private readonly now: () => number; + private readonly onSweep: () => void; + private readonly timeout: number; + private sweeper: NodeJS.Timeout | undefined; + + constructor(options: ExpiringGroupsOptions) { + this.max = options.max; + this.now = options.now ?? Date.now; + this.onSweep = options.onSweep; + this.timeout = options.timeout; + } + + get full(): boolean { + return this.entries.size >= this.max; + } + + get size(): number { + return this.entries.size; + } + + get(key: string): T | undefined { + return this.entries.get(key)?.group; + } + + /** Starts the group's deadline, and the sweeper if this is the only group held. */ + set(key: string, group: T): void { + this.entries.set(key, { deadline: this.now() + this.timeout, group }); + + if (this.sweeper) return; + + this.sweeper = setInterval(() => { this.onSweep(); }, this.timeout); + this.sweeper.unref(); + } + + delete(key: string): void { + this.entries.delete(key); + this.idle(); + } + + clear(): void { + this.entries.clear(); + this.idle(); + } + + /** Removes every group past its deadline and hands them over. */ + takeExpired(): [string, T][] { + const now = this.now(); + const taken: [string, T][] = []; + + for (const [key, entry] of this.entries) { + if (entry.deadline > now) continue; + + taken.push([key, entry.group]); + this.entries.delete(key); + } + + this.idle(); + + return taken; + } + + /** Removes the group held longest and hands it over. Undefined means there was none. */ + takeOldest(): [string, T] | undefined { + const oldest = this.entries.entries().next(); + + if (oldest.done) return undefined; + + const [key, entry] = oldest.value; + + this.delete(key); + + return [key, entry.group]; + } + + private idle(): void { + if (!this.sweeper || this.entries.size > 0) return; + + clearInterval(this.sweeper); + this.sweeper = undefined; + } +} diff --git a/src/message.ts b/src/message.ts index 4796831..621d1b0 100644 --- a/src/message.ts +++ b/src/message.ts @@ -6,6 +6,9 @@ import { detect, encodingByDataCoding, encodings } from './defs/encodings.ts'; /** A single SMS carries 1120 bits, whatever the alphabet. */ const singleMessageBits = 1120; +/** The concatenation UDH numbers the segments of a message in a single octet. */ +export const maxSegments = 255; + /** Budget per concatenated segment: septets for GSM, octets for UCS2. */ const segmentUnits = { ASCII: 153, UCS2: 67 * 2 } as const; @@ -52,7 +55,8 @@ export function bitCount(message: string, encoding?: EncodingName): number { /** * Splits a message into concatenation segments, each prefixed with a UDH. A message that fits in a - * single SMS is returned as one segment with no UDH. + * single SMS is returned as one segment with no UDH, and one needing more than `maxSegments` as no + * segments at all — a UDH cannot number them. * * Splitting walks code points, so an escaped GSM character never straddles a segment boundary and a * surrogate pair is never cut in half. @@ -84,6 +88,8 @@ export function splitMessage(message: string, options: SplitOptions): Buffer[] { if (current !== '') parts.push(current); + if (parts.length > maxSegments) return []; + return parts.map((part, index) => Buffer.concat([ Buffer.from([0x05, 0x00, 0x03, options.reference & 0xFF, parts.length, index + 1]), encodings[encoding].encode(part), diff --git a/src/pdu.ts b/src/pdu.ts index aa5af95..f9c6927 100644 --- a/src/pdu.ts +++ b/src/pdu.ts @@ -79,12 +79,18 @@ function tagIdOf(name: string, input: TlvInput): Result<{ tagId: number }> { return { tagId }; } -/** Encoding a string short_message also settles data_coding and sm_length. */ +/** Encoding a string short_message settles data_coding and sm_length; a buffer settles sm_length. */ function resolveShortMessage( params: Record, ): Record { const message = params.short_message; + if (Buffer.isBuffer(message)) { + return params.sm_length === undefined + ? { ...params, sm_length: message.length } + : { ...params }; + } + if (typeof message !== 'string') return { ...params }; const dataCoding = params.data_coding; @@ -355,7 +361,10 @@ export function pduToObj(pdu: Buffer): Result<{ pduObj: PduObject }> { return { err: plain.err }; } -/** Fields the response shares with the request are echoed back unless the caller overrode them. */ +/** + * Fields the response shares with the request are echoed back unless the caller overrode them, so a + * response that has to state its own value for one — an SMSC's system_id — must pass it. + */ function echoParams( respName: CommandName, pdu: PduObject, diff --git a/src/reassembly.ts b/src/reassembly.ts index c472591..7906f16 100644 --- a/src/reassembly.ts +++ b/src/reassembly.ts @@ -2,23 +2,60 @@ import type { ConcatInfo } from './udh.ts'; import type { LogInt } from '@larvit/log'; import type { ParamValue } from './defs/types.ts'; import type { PduObject } from './pdu.ts'; +import type { Tlv } from './defs/tlvs.ts'; +import { ExpiringGroups } from './expiring-groups.ts'; import { decodeMessage } from './message.ts'; import { paramText } from './defs/types.ts'; export type ReassemblerOptions = { log: LogInt; max: number; + maxOctets?: number | undefined; /** Injected so expiry can be exercised without a wall clock. */ now?: (() => number) | undefined; timeout: number; }; +const defaultMaxOctets = 64 * 1024 * 1024; + type Group = { - deadline: number; + octets: number; parts: Map; total: number; }; +/** Wire reads hand back views, so retaining one segment would pin the whole PDU it arrived in. */ +function detach(pduObj: PduObject): PduObject { + const params: Record = {}; + const tlvs: Record = {}; + + for (const [name, value] of Object.entries(pduObj.params)) { + params[name] = Buffer.isBuffer(value) ? Buffer.from(value) : value; + } + + for (const [name, tlv] of Object.entries(pduObj.tlvs)) { + tlvs[name] = Buffer.isBuffer(tlv.tagValue) + ? { ...tlv, tagValue: Buffer.from(tlv.tagValue) } + : tlv; + } + + return { ...pduObj, params, tlvs }; +} + +function octetsOf(pduObj: PduObject): number { + let octets = 0; + + for (const value of Object.values(pduObj.params)) { + if (Buffer.isBuffer(value)) octets += value.length; + } + + for (const tlv of Object.values(pduObj.tlvs)) { + if (Buffer.isBuffer(tlv.tagValue)) octets += tlv.tagValue.length; + } + + return octets; +} + function groupKey(pduObj: PduObject, reference: number): string { return [ paramText(pduObj.params.source_addr), @@ -52,18 +89,22 @@ export function decodeSegments(pduObjs: PduObject[]): string { /** Holds the segments of incomplete multipart messages until they are whole, capped and expiring. */ export class Reassembler { - private readonly groups = new Map(); + private readonly groups: ExpiringGroups; private readonly log: LogInt; private readonly max: number; - private readonly now: () => number; - private readonly timeout: number; - private sweeper: NodeJS.Timeout | undefined; + private readonly maxOctets: number; + private octets = 0; constructor(options: ReassemblerOptions) { + this.groups = new ExpiringGroups({ + max: options.max, + now: options.now, + onSweep: () => { this.sweep(); }, + timeout: options.timeout, + }); this.log = options.log; this.max = options.max; - this.now = options.now ?? Date.now; - this.timeout = options.timeout; + this.maxOctets = options.maxOctets ?? defaultMaxOctets; } get size(): number { @@ -76,64 +117,68 @@ export class Reassembler { const key = groupKey(pduObj, concat.reference); const group = this.groups.get(key) ?? this.open(key, concat.total); + const replaced = group.parts.get(concat.part); + const segment = detach(pduObj); + const delta = octetsOf(segment) - (replaced === undefined ? 0 : octetsOf(replaced)); - group.parts.set(concat.part, pduObj); + group.parts.set(concat.part, segment); + group.octets += delta; + this.octets += delta; - if (group.parts.size < group.total) return undefined; + if (group.parts.size < group.total) { + this.trim(); + + return undefined; + } this.groups.delete(key); - this.idle(); + this.octets -= group.octets; return [...group.parts.entries()].sort(([a], [b]) => a - b).map(([, part]) => part); } clear(): void { this.groups.clear(); - this.idle(); + this.octets = 0; } /** Drops every group past its deadline. Runs before each collect and on its own timer. */ sweep(): void { - const now = this.now(); - - for (const [key, group] of this.groups) { - if (group.deadline > now) continue; - + for (const [key, group] of this.groups.takeExpired()) { this.log.info('reassembler - incomplete message expired', { key, total: group.total }); - this.groups.delete(key); + this.octets -= group.octets; } - - this.idle(); } private open(key: string, total: number): Group { - if (this.groups.size >= this.max) this.dropOldest(); + if (this.groups.full) this.dropOldest(); - const group: Group = { deadline: this.now() + this.timeout, parts: new Map(), total }; + const group: Group = { octets: 0, parts: new Map(), total }; this.groups.set(key, group); - if (!this.sweeper) { - this.sweeper = setInterval(() => { this.sweep(); }, this.timeout); - this.sweeper.unref(); - } - return group; } - private dropOldest(): void { - const oldest = this.groups.keys().next(); - - if (oldest.done) return; - - this.log.warn('reassembler - buffer full, dropping the oldest message', { max: this.max }); - this.groups.delete(oldest.value); + /** 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 idle(): void { - if (!this.sweeper || this.groups.size > 0) return; + private dropOldest(): void { + const oldest = this.groups.takeOldest(); - clearInterval(this.sweeper); - this.sweeper = undefined; + if (!oldest) return; + + const [, group] = oldest; + + this.log.warn('reassembler - buffer full, dropping the oldest message', { + max: this.max, + maxOctets: this.maxOctets, + octets: this.octets, + }); + this.octets -= group.octets; } } diff --git a/src/reconnect-loop.ts b/src/reconnect-loop.ts index 8d63c1c..3fcc554 100644 --- a/src/reconnect-loop.ts +++ b/src/reconnect-loop.ts @@ -14,6 +14,7 @@ export type ReconnectLoopOptions = { /** Reopens a dropped connection, backing off between attempts until it is told to stop. */ export class ReconnectLoop { private readonly options: ReconnectLoopOptions; + private attempting = false; private delay: number; private halted = false; private timer: NodeJS.Timeout | undefined; @@ -29,7 +30,7 @@ export class ReconnectLoop { } schedule(): void { - if (this.timer || this.isStopped()) return; + if (this.timer || this.attempting || this.isStopped()) return; const delay = this.delay; @@ -53,7 +54,25 @@ export class ReconnectLoop { } private async run(): Promise { - if (this.isStopped()) return; + this.attempting = true; + + // connect() and onConnected() are the application's, so a throw from either lands here. + const retry = await this.attempt().catch((thrown: unknown) => { + const err = thrown instanceof Error ? thrown : new Error(String(thrown)); + + this.options.log.error('reconnect - an attempt threw', { message: err.message }); + + return true; + }); + + this.attempting = false; + + if (retry) this.schedule(); + } + + /** True means the attempt failed and the loop should try again. */ + private async attempt(): Promise { + if (this.isStopped()) return false; const opened = await this.options.connect(); @@ -61,26 +80,26 @@ export class ReconnectLoop { this.options.log.warn('reconnect - could not open a socket', { message: opened.err.message, }); - this.schedule(); - return; + return true; } if (this.isStopped()) { opened.sock.destroy(); - return; + return false; } const up = await this.options.onConnected(opened.sock); if (up.err) { this.options.log.warn('reconnect - could not come back up', { message: up.err.message }); - this.schedule(); - return; + return true; } this.delay = this.options.minDelay; + + return false; } } diff --git a/src/send-sms.ts b/src/send-sms.ts index bb1a6d9..57f7fc0 100644 --- a/src/send-sms.ts +++ b/src/send-sms.ts @@ -6,7 +6,7 @@ import type { Result } from './result.ts'; import { consts } from './defs/constants.ts'; import { detect } from './defs/encodings.ts'; import { paramText } from './defs/types.ts'; -import { smppTime, splitMessage } from './message.ts'; +import { maxSegments, smppTime, splitMessage } from './message.ts'; export type SendSmsOptions = { dlr?: boolean; @@ -23,7 +23,8 @@ export type SendSmsOptions = { validityPeriod?: Date | number | string; }; -export type SendSmsResult = Result<{ pduObjs: PduObject[]; smsIds: string[] }>; +/** Both arrays hold what the peer accepted, so a partial failure names what is already delivered. */ +export type SendSmsResult = { err?: Error; pduObjs: PduObject[]; smsIds: string[] }; /** What sending needs from the session: a concat reference and a way onto the wire. */ export type SendSmsDeps = { @@ -37,7 +38,7 @@ type SegmentOptions = { multipart: boolean; }; -/** Alphanumeric senders must be TON 5; 0.4.0 sent everything as TON 1 (international). */ +/** Alphanumeric senders must be TON 5. */ function addressTon(address: string): number { return /^\+?\d+$/.test(address) ? consts.TON.INTERNATIONAL : consts.TON.ALPHANUMERIC; } @@ -82,6 +83,15 @@ export function submitSmParams( export async function submitSms(deps: SendSmsDeps, sms: SendSmsOptions): Promise { const encoding = sms.encoding ?? detect(sms.message); const segments = splitMessage(sms.message, { encoding, reference: deps.reference }); + + if (segments.length === 0) { + return { + err: new Error(`Message needs more than ${String(maxSegments)} segments, the concatenation limit`), + pduObjs: [], + smsIds: [], + }; + } + const multipart = segments.length > 1; const pduObjs: PduObject[] = []; const smsIds: string[] = []; @@ -95,12 +105,20 @@ export async function submitSms(deps: SendSmsDeps, sms: SendSmsOptions): Promise params: submitSmParams(sms, segment, { encoding, multipart }), }))); - for (const one of sent) { - if (one.err) return { err: one.err }; + let failure: Error | undefined; - pduObjs.push(one.pduObj); - smsIds.push(paramText(one.pduObj.params.message_id)); + for (const one of sent) { + if (one.err) { + failure ??= one.err; + } else if (one.pduObj.cmdStatus === 'ESME_ROK') { + pduObjs.push(one.pduObj); + smsIds.push(paramText(one.pduObj.params.message_id)); + } else { + const refusal = one.pduObj.cmdStatus ?? String(one.pduObj.cmdStatusId); + + failure ??= new Error(`submit_sm refused by the peer: ${refusal}`); + } } - return { pduObjs, smsIds }; + return failure ? { err: failure, pduObjs, smsIds } : { pduObjs, smsIds }; } diff --git a/src/server.ts b/src/server.ts index fbbf70e..bd451db 100644 --- a/src/server.ts +++ b/src/server.ts @@ -128,7 +128,6 @@ async function acceptBind( async function onBind(session: Session, pduObj: PduObject, options: ServerOptions): Promise { const log = options.log ?? silentLog; - // Explicit, or pduReturn's echo answers the ESME with its own system_id instead of ours. const identity = { system_id: options.systemId ?? defaults.systemId }; const systemId = paramText(pduObj.params.system_id); @@ -211,6 +210,7 @@ function createListener( ): Result<{ listener: NetServer | TlsServer; useTls: boolean }> { const useTls = options.tls !== undefined && options.tls !== false; const tlsOptions = typeof options.tls === 'object' ? options.tls : undefined; + const version = options.interfaceVersion ?? defaults.interfaceVersion; if (useTls && !tlsOptions) { log.warn('server - tls without a certificate', { port }); @@ -218,6 +218,13 @@ function createListener( return { err: new Error('Listening over TLS needs tls: { cert, key }') }; } + // An int8 TLV on every bind response: out of range here means no ESME can ever bind. + if (!Number.isInteger(version) || version < 0 || version > 0xFF) { + log.warn('server - interface version out of range', { interfaceVersion: version }); + + return { err: new Error(`interfaceVersion must be 0-255, got ${String(version)}`) }; + } + return { listener: tlsOptions ? createSecureListener(tlsOptions, log) : createNetServer(), useTls, @@ -235,9 +242,15 @@ function onListening(listener: NetServer, smpp: SmppServer, options: ServerOptio log.info('server - listening', { host: options.host ?? '*', port: smpp.port }); - if (options.signal) { - options.signal.addEventListener('abort', () => { void smpp.close(); }, { once: true }); - } + // close() runs the application's own 'close' listeners, so a throw from one lands here. + options.signal?.addEventListener('abort', () => { + void smpp.close().catch((thrown: unknown) => { + const err = thrown instanceof Error ? thrown : new Error(String(thrown)); + + log.warn('server - could not close on abort', { message: err.message }); + smpp.emit('serverError', err); + }); + }, { once: true }); } /** Starts listening for SMPP connections. Resolves once the socket is bound. */ @@ -264,10 +277,14 @@ export function server(options: ServerOptions = {}): Promise { - listener.removeListener('error', onStartupError); - onListening(listener, smpp, options); - resolve({ server: smpp }); - }); + try { + listener.listen(port, options.host, () => { + listener.removeListener('error', onStartupError); + onListening(listener, smpp, options); + resolve({ server: smpp }); + }); + } catch (thrown: unknown) { + onStartupError(thrown instanceof Error ? thrown : new Error(String(thrown))); + } }); } diff --git a/src/session-options.ts b/src/session-options.ts new file mode 100644 index 0000000..a8682e2 --- /dev/null +++ b/src/session-options.ts @@ -0,0 +1,53 @@ +import type { LogInt } from '@larvit/log'; +import type { PduObject } from './pdu.ts'; +import type { Result, VoidResult } from './result.ts'; +import type { Session } from './session.ts'; +import type { Socket } from 'node:net'; + +export type SendOptions = { signal?: AbortSignal | undefined }; + +/** + * How to come back after an unexpected disconnect. The session owns the retry loop; the caller + * supplies how to open a socket and what to do once it is open (bind, for a client). + */ +export type ReconnectOptions = { + connect: () => Promise>; + maxDelay?: number | undefined; + minDelay?: number | undefined; + onConnected: (session: Session) => Promise; +}; + +export type SessionOptions = { + enquireLinkInterval?: number | undefined; + idleTimeout?: number | undefined; + log?: LogInt | undefined; + maxOutstanding?: number | undefined; + maxReassembly?: number | undefined; + /** + * First refusal on every incoming request. Returning true means the hook answered it and the + * built-in handling is skipped — this is how the server owns bind without the session also + * replying "invalid command". + */ + onRequest?: ((session: Session, pduObj: PduObject) => Promise) | undefined; + reassemblyTimeout?: number | undefined; + reconnect?: ReconnectOptions | undefined; + responseTimeout?: number | undefined; + sock: Socket; + /** This end's own identity, answered to the peer in place of the one it sent. */ + systemId?: string | undefined; +}; + +export const defaultSystemId = ''; + +export const defaults = { + /** Receipts of a multipart message can be a working day apart, so the cap does the bounding. */ + dlrMergeTimeout: 86_400_000, + maxDelay: 30_000, + maxDlrMerges: 1000, + maxOutstanding: 10, + maxReassembly: 1000, + minDelay: 1000, + reassemblyTimeout: 300_000, + responseTimeout: 30_000, + systemId: defaultSystemId, +}; diff --git a/src/session.ts b/src/session.ts index 2cd377f..699de6d 100644 --- a/src/session.ts +++ b/src/session.ts @@ -4,6 +4,7 @@ import type { LogInt } from '@larvit/log'; import type { MessageDlr } from './dlr-merger.ts'; import type { ParamValue } from './defs/types.ts'; import type { PduObject, PduObjectInput, TlvInput } from './pdu.ts'; +import type { ReconnectOptions, SendOptions, SessionOptions } from './session-options.ts'; import type { Result, VoidResult } from './result.ts'; import type { SendSmsOptions, SendSmsResult } from './send-sms.ts'; import type { Sms } from './sms.ts'; @@ -19,13 +20,15 @@ import { SendWindow } from './send-window.ts'; import { concatInfo } from './udh.ts'; import { consts, optionalParamsMinVersion } from './defs/constants.ts'; import { createSms } from './sms.ts'; +import { defaultSystemId, defaults } from './session-options.ts'; import { dlrFromPdu } from './dlr.ts'; import { isResp, objToPdu, pduReturn, pduToObj } from './pdu.ts'; import { paramText } from './defs/types.ts'; import { silentLog } from './log.ts'; import { submitSms } from './send-sms.ts'; -export type { MessageDlr, SendSmsOptions }; +export type { MessageDlr, ReconnectOptions, SendOptions, SendSmsOptions, SessionOptions }; +export { defaultSystemId }; export type SessionEvents = { close: []; @@ -39,51 +42,6 @@ export type SessionEvents = { sms: [Sms]; }; -export type SendOptions = { signal?: AbortSignal | undefined }; - -/** - * How to come back after an unexpected disconnect. The session owns the retry loop; the caller - * supplies how to open a socket and what to do once it is open (bind, for a client). - */ -export type ReconnectOptions = { - connect: () => Promise>; - maxDelay?: number | undefined; - minDelay?: number | undefined; - onConnected: (session: Session) => Promise; -}; - -export type SessionOptions = { - enquireLinkInterval?: number | undefined; - idleTimeout?: number | undefined; - log?: LogInt | undefined; - maxOutstanding?: number | undefined; - maxReassembly?: number | undefined; - /** - * First refusal on every incoming request. Returning true means the hook answered it and the - * built-in handling is skipped — this is how the server owns bind without the session also - * replying "invalid command". - */ - onRequest?: ((session: Session, pduObj: PduObject) => Promise) | undefined; - reassemblyTimeout?: number | undefined; - reconnect?: ReconnectOptions | undefined; - responseTimeout?: number | undefined; - sock: Socket; - /** This end's own identity, answered to the peer in place of the one it sent. */ - systemId?: string | undefined; -}; - -export const defaultSystemId = ''; - -const defaults = { - maxDelay: 30_000, - maxOutstanding: 10, - maxReassembly: 1000, - minDelay: 1000, - reassemblyTimeout: 300_000, - responseTimeout: 30_000, - systemId: defaultSystemId, -}; - export const bindCommands: readonly string[] = [ 'bind_receiver', 'bind_transceiver', @@ -100,7 +58,7 @@ export class Session extends EventEmitter { peerInterfaceVersion: number | undefined = undefined; userData: unknown = undefined; - private readonly dlrMerger = new DlrMerger(); + private readonly dlrMerger: DlrMerger; private readonly options: SessionOptions; private readonly pending: PendingRequests; private readonly reassembler: Reassembler; @@ -117,6 +75,11 @@ export class Session extends EventEmitter { this.log = options.log ?? silentLog; this.options = options; + this.dlrMerger = new DlrMerger({ + log: this.log, + max: defaults.maxDlrMerges, + timeout: defaults.dlrMergeTimeout, + }); this.pending = new PendingRequests(this.log); this.reassembler = new Reassembler({ log: this.log, @@ -130,7 +93,8 @@ export class Session extends EventEmitter { idleTimeout: options.idleTimeout, log: this.log, onEnquireLink: () => { void this.send({ cmdName: 'enquire_link' }); }, - onIdle: () => { this.close(); }, + // Not close(): a link that went quiet is a drop, and a drop is what reconnect is for. + onIdle: () => { this.teardown(); }, }); this.window = new SendWindow(options.maxOutstanding ?? defaults.maxOutstanding); @@ -172,10 +136,18 @@ export class Session extends EventEmitter { tlvs?: Record, ): Promise { const built = pduReturn(pdu, status, params, tlvs); + const sent = built.err ? { err: built.err } : this.write(built.buffer); - if (built.err) return { err: built.err }; + if (sent.err) { + this.log.warn('session - could not answer a request', { + cmdName: pdu.cmdName, + message: sent.err.message, + seqNr: pdu.seqNr, + }); + this.emit('sessionError', sent.err); + } - return Promise.resolve(this.write(built.buffer)); + return Promise.resolve(sent); } async sendSms(sms: SendSmsOptions, options: SendOptions = {}): Promise { @@ -185,18 +157,20 @@ export class Session extends EventEmitter { send: input => this.send(input, options), }, sms); - if (!sent.err) this.dlrMerger.expect(sent.smsIds); + if (!sent.err && sms.dlr === true) this.dlrMerger.expect(sent.smsIds); return sent; } - /** Unbinds politely, then closes. */ + /** Unbinds politely, then closes. Many SMSCs drop the link instead of answering, which is fine. */ async unbind(): Promise { + const wasOpen = !this.closed; const sent = await this.send({ cmdName: 'unbind' }); + const closedOnUnbind = wasOpen && this.closed; this.close(); - return sent.err ? { err: sent.err } : {}; + return sent.err && !closedOnUnbind ? { err: sent.err } : {}; } /** Closes for good. A session closed this way never reconnects. */ @@ -240,6 +214,9 @@ export class Session extends EventEmitter { /** Wires a freshly opened socket into this session, replacing any previous one. */ private attach(sock: Socket): void { + // The socket being replaced is already dead, and its three handlers still point here. + if (this.sock !== sock) this.sock.removeAllListeners(); + this.sock = sock; this.framer = new PduFramer(); this.closed = false; @@ -283,6 +260,7 @@ export class Session extends EventEmitter { this.closed = true; this.timers.clear(); this.pending.settleAll(new Error('Session closed before a response arrived')); + this.dlrMerger.clear(); this.reassembler.clear(); this.sock.destroy(); this.emit('close'); @@ -355,7 +333,13 @@ export class Session extends EventEmitter { } this.emit('incomingPduObj', pduObj); - void this.handle(pduObj); + // Every application hook and listener reached from an incoming PDU funnels through here. + void this.handle(pduObj).catch((thrown: unknown) => { + const err = thrown instanceof Error ? thrown : new Error(String(thrown)); + + this.log.error('session - a handler threw', { message: err.message }); + this.emit('sessionError', err); + }); } private async handle(pduObj: PduObject): Promise { @@ -385,7 +369,6 @@ export class Session extends EventEmitter { private async unhandled(pduObj: PduObject): Promise { if (bindCommands.includes(pduObj.cmdName)) { this.log.info('session - bind on an already bound session', { cmdName: pduObj.cmdName }); - // Explicit, or pduReturn's echo answers the peer with its own system_id instead of ours. await this.sendReturn(pduObj, 'ESME_RALYBND', { system_id: this.options.systemId ?? defaults.systemId, }); diff --git a/src/sms.ts b/src/sms.ts index e8b3ba3..ee4c09e 100644 --- a/src/sms.ts +++ b/src/sms.ts @@ -8,6 +8,12 @@ import { receiptCodes } from './dlr.ts'; import { smppDate } from './message.ts'; import { uuidv7 } from './uuid.ts'; +export type SendRespOptions = { + /** The id the peer correlates a later delivery receipt by. Defaults to a generated UUID v7. */ + smsId?: string; + status?: ErrorName; +}; + /** * A received SMS, and the handle for answering it. Multipart messages arrive as one Sms carrying * every segment's PDU. @@ -21,10 +27,10 @@ export type Sms = { /** Sends a delivery report back to the sender. Defaults to DELIVERED. */ sendDlr: (status?: MessageState) => Promise>; /** Answers every segment. Part of the protocol, not optional. Defaults to ESME_ROK. */ - sendResp: (status?: ErrorName) => Promise; + sendResp: (options?: SendRespOptions) => Promise; session: Session; - /** Generated as a UUID v7 unless the application sets its own before answering. */ - smsId: string; + /** The id answered to the peer: what sendResp() was given, or a generated UUID v7. */ + readonly smsId: string; submitTime: Date; to: string; }; @@ -46,6 +52,7 @@ export function createSms(input: SmsInput): Sms { const first = input.pduObjs[0]; const registered = first?.params.registered_delivery; const dataCoding = first?.params.data_coding; + const answered = { smsId: uuidv7() }; const sms: Sms = { dlr: typeof registered === 'number' && registered !== 0, @@ -54,9 +61,11 @@ export function createSms(input: SmsInput): Sms { message: input.message, pduObjs: input.pduObjs, sendDlr: status => sendDlr(sms, status), - sendResp: status => sendResp(sms, status), + sendResp: options => sendResp(sms, answered, options ?? {}), session: input.session, - smsId: uuidv7(), + get smsId(): string { + return answered.smsId; + }, submitTime: new Date(), to: input.to, }; @@ -64,17 +73,27 @@ export function createSms(input: SmsInput): Sms { return sms; } -async function sendResp(sms: Sms, status: ErrorName = 'ESME_ROK'): Promise { +async function sendResp( + sms: Sms, + answered: { smsId: string }, + options: SendRespOptions, +): Promise { const total = sms.pduObjs.length; if (total === 0) { return { err: new Error('No PDUs to answer') }; } + if (options.smsId === '') { + return { err: new Error('smsId must not be empty') }; + } + + if (options.smsId !== undefined) answered.smsId = options.smsId; + const results = await Promise.all(sms.pduObjs.map((pduObj, index) => sms.session.sendReturn( pduObj, - status, - { message_id: segmentId(sms.smsId, index, total) }, + options.status ?? 'ESME_ROK', + { message_id: segmentId(answered.smsId, index, total) }, ))); return results.find(result => result.err) ?? {}; diff --git a/test/message.test.ts b/test/message.test.ts index 3a9ba1b..db95e5c 100644 --- a/test/message.test.ts +++ b/test/message.test.ts @@ -73,6 +73,11 @@ describe('splitMessage()', () => { assert.deepEqual(segments[1]?.subarray(0, 6), Buffer.from([0x05, 0x00, 0x03, 0x2A, 2, 2])); }); + test('produces no segments at all for a message no UDH can number', () => { + assert.equal(splitMessage('a'.repeat(153 * 255), { reference: 1 }).length, 255); + assert.equal(splitMessage('a'.repeat(153 * 255 + 1), { reference: 1 }).length, 0); + }); + test('splits on characters, never inside an escape sequence', () => { const segments = splitMessage('€'.repeat(100), { reference: 1 }); const rejoined = segments diff --git a/test/pdu.test.ts b/test/pdu.test.ts index 7355d9e..f0222bb 100644 --- a/test/pdu.test.ts +++ b/test/pdu.test.ts @@ -157,6 +157,23 @@ describe('encoding submit_sm', () => { assert.equal(pduObj.params.short_message.toString('hex'), '05000301010168656a2076c3a4726c64656e'); }); + test('derives sm_length from a Buffer short_message the caller gave no length for', () => { + const message = Buffer.concat([Buffer.from('050003010201', 'hex'), Buffer.from('hej')]); + const pduObj = decode(encode({ + cmdName: 'submit_sm', + params: { + destination_addr: '46709771337', + esm_class: 0x40, + short_message: message, + source_addr: '46701113311', + }, + seqNr: 12, + })); + + assert.equal(pduObj.params.sm_length, message.length); + assert.deepEqual(pduObj.params.short_message, message); + }); + test('accepts a number for a C-string parameter', () => { const pduObj = decode(encode({ cmdName: 'submit_sm_resp', @@ -185,6 +202,40 @@ describe('encoding submit_sm', () => { }); }); +describe('encoding submit_multi', () => { + const dest = { dest_addr_npi: 1, dest_addr_ton: 1, destination_addr: '46709771337' }; + + test('reports a value the destination structures cannot hold instead of throwing', () => { + const badTon = objToPdu({ + cmdName: 'submit_multi', + params: { + dest_address: [{ ...dest, dest_addr_ton: 999 }], + short_message: 'hi', + source_addr: '46701113311', + }, + }); + const tooMany = objToPdu({ + cmdName: 'submit_multi', + params: { + dest_address: Array.from({ length: 300 }, () => dest), + short_message: 'hi', + source_addr: '46701113311', + }, + }); + const badStatus = objToPdu({ + cmdName: 'submit_multi_resp', + params: { + message_id: '01a03ff4-737f-7c01-91db-cb14aa779bcf', + unsuccess_sme: [{ ...dest, error_status_code: 0x1FFFFFFFF }], + }, + }); + + assert.ok(badTon.err instanceof Error); + assert.ok(tooMany.err instanceof Error); + assert.ok(badStatus.err instanceof Error); + }); +}); + describe('TLVs', () => { test('extracts TLVs from a delivery receipt captured from an SMSC', () => { const pduObj = decode(Buffer.from( diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 7501161..ad6f99d 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -1,14 +1,19 @@ import assert from 'node:assert/strict'; import net from 'node:net'; import test, { describe } from 'node:test'; +import type { ErrorName } from '../src/defs/errors.ts'; import type { MessageDlr } from '../src/session.ts'; -import type { PduObject } from '../src/pdu.ts'; +import type { PduObject, PduObjectInput } from '../src/pdu.ts'; +import type { Result } from '../src/result.ts'; +import type { SendSmsResult } from '../src/send-sms.ts'; import type { Sms } from '../src/sms.ts'; import type { SmppServer } from '../src/server.ts'; -import { Reassembler } from '../src/reassembly.ts'; +import { Reassembler, decodeSegments } from '../src/reassembly.ts'; import { client } from '../src/client.ts'; +import { errors } from '../src/defs/errors.ts'; import { server } from '../src/server.ts'; import { silentLog } from '../src/log.ts'; +import { submitSms } from '../src/send-sms.ts'; async function startServer(options: Parameters[0] = {}): Promise { const { err, server: smpp } = await server({ ...options, port: 0 }); @@ -41,8 +46,7 @@ describe('merged delivery reports', () => { const [sms] = await Promise.all([ incoming.then(async received => { - received.smsId = 'merge-me'; - await received.sendResp(); + await received.sendResp({ smsId: 'merge-me' }); return received; }), @@ -80,8 +84,7 @@ describe('merged delivery reports', () => { const [sms] = await Promise.all([ incoming.then(async received => { - received.smsId = 'partly-failed'; - await received.sendResp(); + await received.sendResp({ smsId: 'partly-failed' }); return received; }), @@ -105,6 +108,100 @@ describe('merged delivery reports', () => { }); }); +describe('sendSms()', () => { + function submitResp(seqNr: number, messageId: string, status: ErrorName = 'ESME_ROK'): PduObject { + return { + cmdId: 0x80000004, + cmdLength: 0, + cmdName: 'submit_sm_resp', + cmdStatus: status, + cmdStatusId: errors[status], + params: { message_id: messageId }, + seqNr, + tlvs: {}, + }; + } + + /** Three segments, each answered by whatever the caller decides for that part. */ + function sendSegments( + answer: (part: number) => Result<{ pduObj: PduObject }>, + ): Promise { + let part = 0; + + return submitSms( + { + log: silentLog, + reference: 7, + send: () => { + part++; + + return Promise.resolve(answer(part)); + }, + }, + { from: '46701113311', message: 'x'.repeat(400), to: '46709771337' }, + ); + } + + test('reports a submit_sm the peer refused instead of an empty message id', async () => { + const smpp = await startServer(); + const incoming = once(resolve => { + smpp.on('session', session => session.on('sms', resolve)); + }); + const { session } = await client({ port: smpp.port }); + + assert.ok(session); + + const [, sent] = await Promise.all([ + incoming.then(received => received.sendResp({ status: 'ESME_RMSGQFUL' })), + session.sendSms({ from: '46701113311', message: 'the queue is full', to: '46709771337' }), + ]); + + assert.ok(sent.err instanceof Error); + assert.match(sent.err.message, /ESME_RMSGQFUL/); + + session.close(); + await smpp.close(); + }); + + // A retry that repeats the segments the SMSC already took bills the recipient twice. + test('hands back the ids that landed when a segment fails', async () => { + const refused = await sendSegments(part => part === 2 + ? { pduObj: submitResp(part, '', 'ESME_RMSGQFUL') } + : { pduObj: submitResp(part, `landed-${String(part)}`) }); + + assert.ok(refused.err instanceof Error); + assert.match(refused.err.message, /ESME_RMSGQFUL/); + assert.equal(refused.pduObjs.length, 2); + assert.deepEqual(refused.smsIds, ['landed-1', 'landed-3']); + + const unanswered = await sendSegments(part => part === 2 + ? { err: new Error('No response to seqNr 2') } + : { pduObj: submitResp(part, `landed-${String(part)}`) }); + + assert.ok(unanswered.err instanceof Error); + assert.deepEqual(unanswered.smsIds, ['landed-1', 'landed-3']); + }); + + test('refuses a message needing more segments than a UDH can number', async () => { + const attempts: PduObjectInput[] = []; + const sent = await submitSms( + { + log: silentLog, + reference: 1, + send: input => { + attempts.push(input); + + return Promise.resolve({ err: new Error('nothing should reach the wire') }); + }, + }, + { from: '46701113311', message: 'a'.repeat(153 * 256), to: '46709771337' }, + ); + + assert.ok(sent.err instanceof Error); + assert.equal(attempts.length, 0); + }); +}); + describe('reconnect', () => { test('re-binds after the connection drops, keeping the same session object', async () => { const smpp = await startServer(); @@ -231,6 +328,44 @@ describe('reassembly bounds', () => { reassembler.clear(); }); + test('drops the oldest incomplete message once the retained octets exceed the cap', () => { + const reassembler = new Reassembler({ + log: silentLog, + max: 10, + maxOctets: 30, + now: () => 0, + timeout: 60_000, + }); + + for (const reference of [1, 2, 3]) { + assert.equal(collect(reassembler, reference, 1, 2), undefined); + } + + assert.equal(reassembler.size, 2); + assert.equal(collect(reassembler, 1, 2, 2), undefined); + + reassembler.clear(); + }); + + // A retained subarray keeps its whole framed PDU alive, up to maxPduLength per segment. + test('copies a segment out of the buffer it arrived in', () => { + const reassembler = new Reassembler({ log: silentLog, max: 10, now: () => 0, timeout: 60_000 }); + const framed = Buffer.alloc(1024); + const first = segment(6, 1, 2); + + Buffer.concat([Buffer.from([0x05, 0x00, 0x03, 6, 2, 1]), Buffer.from('fragment')]).copy(framed); + first.params.short_message = framed.subarray(0, 14); + + assert.equal(reassembler.collect(first, { part: 1, reference: 6, total: 2 }), undefined); + + framed.fill(0x00); + + const whole = collect(reassembler, 6, 2, 2); + + assert.ok(whole); + assert.equal(decodeSegments(whole), 'fragmentfragment'); + }); + test('expires an incomplete message once its timeout has passed', () => { let now = 0; const reassembler = new Reassembler({ log: silentLog, max: 10, now: () => now, timeout: 60 }); diff --git a/test/session.test.ts b/test/session.test.ts index 87e0cca..9bcf0be 100644 --- a/test/session.test.ts +++ b/test/session.test.ts @@ -1,15 +1,19 @@ import assert from 'node:assert/strict'; import net from 'node:net'; import test, { describe } from 'node:test'; +import type { Dlr } from '../src/dlr.ts'; import type { PduObject, PduObjectInput } from '../src/pdu.ts'; -import type { Session } from '../src/session.ts'; import type { Sms } from '../src/sms.ts'; import type { SmppServer } from '../src/server.ts'; +import { DlrMerger } from '../src/dlr-merger.ts'; import { PduFramer } from '../src/pdu-framer.ts'; +import { ReconnectLoop } from '../src/reconnect-loop.ts'; +import { Session, bindCommands } from '../src/session.ts'; import { client } from '../src/client.ts'; -import { isCommand, objToPdu, pduToObj } from '../src/pdu.ts'; +import { isCommand, objToPdu, pduReturn, pduToObj } from '../src/pdu.ts'; import { paramText } from '../src/defs/types.ts'; import { server } from '../src/server.ts'; +import { silentLog } from '../src/log.ts'; async function startServer(options: Parameters[0] = {}): Promise { const { err, server: smpp } = await server({ ...options, port: 0 }); @@ -28,6 +32,86 @@ function once(register: (resolve: (value: T) => void) => void): Promise { return new Promise(resolve => { register(resolve); }); } +function delay(ms: number): Promise { + return new Promise(resolve => { setTimeout(resolve, ms); }); +} + +/** Polls until the condition holds; false means it never did within the budget. */ +async function waitFor(condition: () => boolean, budget = 2000): Promise { + const deadline = Date.now() + budget; + + while (!condition()) { + if (Date.now() > deadline) return false; + + await delay(5); + } + + return true; +} + +function raceWithin(ms: number, promise: Promise): Promise { + return Promise.race([promise, delay(ms).then((): false => false)]); +} + +type Peer = { close: () => Promise; port: number }; + +/** Answers binds and nothing else, which is what a link the peer has stopped serving looks like. */ +async function bindOnlyPeer(options: { dropOn?: string } = {}): Promise { + const sockets: net.Socket[] = []; + const listener = net.createServer(sock => { + const framer = new PduFramer(); + + sockets.push(sock); + sock.on('data', chunk => { + framer.push(chunk); + + for (const pdu of framer.next().pdus ?? []) { + const { pduObj } = pduToObj(pdu); + + if (pduObj && pduObj.cmdName === options.dropOn) { + sock.destroy(); + + return; + } + + if (!pduObj || !bindCommands.includes(pduObj.cmdName)) continue; + + const { buffer } = pduReturn(pduObj, 'ESME_ROK', { system_id: 'silent' }); + + if (buffer) sock.write(buffer); + } + }); + }); + + await new Promise(resolve => { listener.listen(0, resolve); }); + + const address = listener.address(); + + return { + close: async () => { + for (const sock of sockets) { + sock.destroy(); + } + + await new Promise(resolve => { listener.close(() => { resolve(); }); }); + }, + port: typeof address === 'object' && address !== null ? address.port : 0, + }; +} + +function enquireLink(seqNr: number): PduObject { + return { + cmdId: 0x00000015, + cmdLength: 16, + cmdName: 'enquire_link', + cmdStatus: 'ESME_ROK', + cmdStatusId: 0, + params: {}, + seqNr, + tlvs: {}, + }; +} + type RawPeer = { close: () => void; /** The next PDU the server sends, queued so none is missed between reads. */ @@ -110,6 +194,42 @@ describe('bind', () => { await smpp.close(); }); + // Plenty of SMSCs drop the connection on unbind instead of answering it. + test('takes a close that follows our unbind as a clean unbind', async () => { + const peer = await bindOnlyPeer({ dropOn: 'unbind' }); + const { session } = await client({ port: peer.port, responseTimeout: 2000 }); + + assert.ok(session); + assert.deepEqual(await session.unbind(), {}); + + await peer.close(); + }); + + test('still reports a close that lands on another in-flight request', async () => { + const peer = await bindOnlyPeer({ dropOn: 'enquire_link' }); + const { session } = await client({ port: peer.port, responseTimeout: 2000 }); + + assert.ok(session); + + const sent = await session.send({ cmdName: 'enquire_link' }); + + assert.ok(sent.err instanceof Error); + assert.equal(sent.err.message, 'Session closed before a response arrived'); + + session.close(); + await peer.close(); + }); + + test('reports an unbind the peer left unanswered on a link that stays up', async () => { + const peer = await bindOnlyPeer(); + const { session } = await client({ port: peer.port, responseTimeout: 150 }); + + assert.ok(session); + assert.ok((await session.unbind()).err instanceof Error); + + await peer.close(); + }); + test('reports the resolved port when 0 was requested', async () => { const smpp = await startServer(); @@ -283,8 +403,10 @@ describe('sending', () => { const [sms, sent] = await Promise.all([ incoming.then(async received => { - received.smsId = 'fixed-id'; - await received.sendResp(); + const refused = await received.sendResp({ smsId: '' }); + + assert.ok(refused.err instanceof Error); + await received.sendResp({ smsId: 'fixed-id' }); return received; }), @@ -295,6 +417,7 @@ describe('sending', () => { assert.equal(sms.to, '46709771337'); assert.equal(sms.message, 'hello world'); assert.equal(sms.dlr, false); + assert.equal(sms.smsId, 'fixed-id'); assert.equal(sent.err, undefined); assert.deepEqual(sent.smsIds, ['fixed-id']); @@ -321,8 +444,7 @@ describe('sending', () => { const [sms, sent] = await Promise.all([ incoming.then(async received => { - received.smsId = 'long-id'; - await received.sendResp(); + await received.sendResp({ smsId: 'long-id' }); return received; }), @@ -332,6 +454,7 @@ describe('sending', () => { assert.equal(sms.message, message); assert.ok(sms.pduObjs.length > 1); assert.equal(sent.err, undefined); + assert.equal(sent.pduObjs.length, 4); assert.deepEqual(sent.smsIds, ['long-id-1', 'long-id-2', 'long-id-3', 'long-id-4']); session.close(); @@ -358,6 +481,7 @@ describe('sending', () => { ]); assert.equal(sms.message, message); + assert.match(sms.smsId, /^[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/); session.close(); await smpp.close(); @@ -407,8 +531,7 @@ describe('delivery reports', () => { const [sms] = await Promise.all([ incoming.then(async received => { - received.smsId = 'dlr-id'; - await received.sendResp(); + await received.sendResp({ smsId: 'dlr-id' }); return received; }), @@ -449,8 +572,7 @@ describe('delivery reports', () => { const [sms] = await Promise.all([ incoming.then(async received => { - received.smsId = 'fail-id'; - await received.sendResp(); + await received.sendResp({ smsId: 'fail-id' }); return received; }), @@ -506,6 +628,39 @@ describe('delivery reports', () => { peer.close(); await smpp.close(); }); + + test('merges nothing for a message that asked for no receipt', async () => { + const smpp = await startServer(); + const incoming = once(resolve => { + smpp.on('session', session => session.on('sms', resolve)); + }); + const { session } = await connect(smpp); + + assert.ok(session); + + const perSegment: string[] = []; + let merged = 0; + + session.on('dlr', dlr => perSegment.push(dlr.smsId)); + session.on('messageDlr', () => { merged++; }); + + const [sms] = await Promise.all([ + incoming.then(async received => { + await received.sendResp({ smsId: 'unrequested' }); + + return received; + }), + session.sendSms({ from: '46701113311', message: 'x'.repeat(400), to: '46709771337' }), + ]); + + await sms.sendDlr(); + + assert.deepEqual(perSegment, ['unrequested-1', 'unrequested-2', 'unrequested-3']); + assert.equal(merged, 0); + + session.close(); + await smpp.close(); + }); }); describe('a session captured from Kannel', () => { @@ -632,4 +787,297 @@ describe('robustness', () => { session.close(); await smpp.close(); }); + + test('reports a response it could not send', async () => { + const sock = new net.Socket(); + + sock.destroy(); + + const session = new Session({ sock }); + const failed = once(resolve => { session.on('sessionError', resolve); }); + const sent = await session.sendReturn(enquireLink(7)); + const reported = await raceWithin(500, failed); + + assert.ok(sent.err instanceof Error); + assert.ok(reported instanceof Error, 'a response that never reached the wire should be reported'); + assert.equal(reported.message, sent.err.message); + + session.close(); + }); + + test('ignores events from the socket it left behind on a reconnect', async () => { + const smpp = await startServer(); + + smpp.on('session', bound => { + bound.on('sms', sms => { void sms.sendResp(); }); + }); + + const { session } = await connect(smpp, { reconnect: { maxDelay: 50, minDelay: 10 } }); + + assert.ok(session); + + const reconnected = once(resolve => { session.on('reconnected', () => { resolve(true); }); }); + const dead = session.sock; + + for (const serverSession of smpp.sessions) { + serverSession.close(); + } + + await reconnected; + + let closes = 0; + + session.on('close', () => { closes++; }); + dead.emit('close'); + + const sent = await session.sendSms({ + from: '46701113311', + message: 'still up', + to: '46709771337', + }); + + assert.equal(closes, 0); + assert.equal(sent.err, undefined); + + session.close(); + await smpp.close(); + }); + + test('closes the session when the signal aborts after the bind', async () => { + const smpp = await startServer(); + const controller = new AbortController(); + const { err, session } = await connect(smpp, { signal: controller.signal }); + + assert.equal(err, undefined); + assert.ok(session); + + const closed = once(resolve => { session.on('close', () => { resolve(true); }); }); + + controller.abort(); + + assert.ok(await closed); + await smpp.close(); + }); +}); + +describe('application hooks that throw', () => { + test('turns a throwing authenticate into a session error', async () => { + const smpp = await startServer({ + authenticate: () => { throw new Error('authenticate exploded'); }, + }); + const failed = once(resolve => { + smpp.on('session', session => { session.on('sessionError', resolve); }); + }); + const { err } = await connect(smpp, { responseTimeout: 200 }); + const reported = await raceWithin(500, failed); + + assert.ok(err instanceof Error); + assert.ok(reported instanceof Error, 'a throwing authenticate should reach the session'); + assert.equal(reported.message, 'authenticate exploded'); + + await smpp.close(); + }); + + test('turns a throwing sms listener into a session error', async () => { + const smpp = await startServer(); + const failed = once(resolve => { + smpp.on('session', session => { + session.on('sessionError', resolve); + session.on('sms', () => { throw new Error('listener exploded'); }); + }); + }); + const { session } = await connect(smpp, { responseTimeout: 200 }); + + assert.ok(session); + + const sent = await session.sendSms({ + from: '46701113311', + message: 'blows up the listener', + to: '46709771337', + }); + const reported = await raceWithin(500, failed); + + assert.ok(sent.err instanceof Error); + assert.ok(reported instanceof Error, 'a throwing sms listener should reach the session'); + assert.equal(reported.message, 'listener exploded'); + + session.close(); + await smpp.close(); + }); + + test('keeps the reconnect loop alive when connect throws', async () => { + let attempts = 0; + const loop = new ReconnectLoop({ + connect: () => { + attempts++; + + throw new Error('connect exploded'); + }, + log: silentLog, + maxDelay: 10, + minDelay: 1, + onConnected: () => Promise.resolve({}), + }); + + loop.schedule(); + + const retried = await waitFor(() => attempts >= 2); + + loop.stop(); + + assert.ok(retried, 'a throwing connect should be retried, not left for the process to die on'); + }); + + test('starts only one reconnect attempt at a time', async () => { + let attempts = 0; + let finish: (() => void) | undefined; + const loop = new ReconnectLoop({ + connect: () => { + attempts++; + + return new Promise(resolve => { + finish = () => { resolve({ err: new Error('no socket') }); }; + }); + }, + log: silentLog, + maxDelay: 5, + minDelay: 1, + onConnected: () => Promise.resolve({}), + }); + + loop.schedule(); + + assert.ok(await waitFor(() => attempts === 1)); + + // A second drop landing while the first attempt is still inside connect(). + loop.schedule(); + await delay(30); + + assert.equal(attempts, 1); + + loop.stop(); + finish?.(); + }); +}); + +describe('link timers', () => { + test('closes a client link the peer has stopped answering', async () => { + const peer = await bindOnlyPeer(); + const { err, session } = await client({ enquireLinkInterval: 50, port: peer.port }); + + assert.equal(err, undefined); + assert.ok(session); + + const closed = once(resolve => { session.on('close', () => { resolve(true); }); }); + + assert.ok( + await raceWithin(1000, closed), + 'a peer that answers nothing should time the link out', + ); + + session.close(); + await peer.close(); + }); + + test('reconnects a link that timed out', async () => { + const peer = await bindOnlyPeer(); + const { session } = await client({ + enquireLinkInterval: 40, + port: peer.port, + reconnect: { maxDelay: 20, minDelay: 10 }, + }); + + assert.ok(session); + + const back = once(resolve => { session.on('reconnected', () => { resolve(true); }); }); + + assert.ok(await raceWithin(2000, back), 'a link that timed out should be reconnected'); + + session.close(); + await peer.close(); + }); +}); + +describe('merged delivery report bounds', () => { + function receipt(smsId: string): Dlr { + return { + doneDate: undefined, + errorCode: undefined, + receipt: undefined, + smsId, + statusId: 2, + statusMsg: 'DELIVERED', + }; + } + + function merger(options: { max?: number; now?: () => number } = {}): DlrMerger { + return new DlrMerger({ + log: silentLog, + max: options.max ?? 10, + now: options.now ?? (() => 0), + timeout: 60, + }); + } + + test('merges the receipts of one message and forgets the group', () => { + const dlrMerger = merger(); + + dlrMerger.expect(['whole-1', 'whole-2']); + + assert.equal(dlrMerger.collect(receipt('whole-1')), undefined); + + const merged = dlrMerger.collect(receipt('whole-2')); + + assert.ok(merged); + assert.equal(merged.smsId, 'whole'); + assert.equal(merged.segments.length, 2); + assert.equal(dlrMerger.size, 0); + }); + + // Every multipart send registered a group, and only a complete set of receipts ever removed it. + test('drops the oldest group once the cap is reached', () => { + const dlrMerger = merger({ max: 2 }); + + for (const base of ['first', 'second', 'third']) { + dlrMerger.expect([`${base}-1`, `${base}-2`]); + } + + assert.equal(dlrMerger.size, 2); + assert.equal(dlrMerger.collect(receipt('first-1')), undefined); + assert.equal(dlrMerger.collect(receipt('first-2')), undefined); + + dlrMerger.clear(); + assert.equal(dlrMerger.size, 0); + }); + + test('expires a group whose receipts never all arrived', () => { + let now = 0; + const dlrMerger = merger({ now: () => now }); + + dlrMerger.expect(['late-1', 'late-2']); + now = 61; + + assert.equal(dlrMerger.collect(receipt('late-1')), undefined); + assert.equal(dlrMerger.size, 0); + }); +}); + +describe('option validation', () => { + test('refuses an interface version that cannot go on the wire', async () => { + const { err, server: smpp } = await server({ interfaceVersion: 0x100, port: 0 }); + + if (smpp) await smpp.close(); + + assert.ok(err instanceof Error); + }); + + test('returns an error rather than rejecting on an impossible port', async () => { + const listening = await server({ port: 70_000 }); + + assert.ok(listening.err instanceof Error); + + const connected = await client({ port: 70_000 }); + + assert.ok(connected.err instanceof Error); + }); }); diff --git a/test/tls.test.ts b/test/tls.test.ts index dd16b2c..4109a0a 100644 --- a/test/tls.test.ts +++ b/test/tls.test.ts @@ -142,8 +142,7 @@ describe('tls', () => { const [sms, sent] = await Promise.all([ incoming.then(async received => { - received.smsId = 'tls-id'; - await received.sendResp(); + await received.sendResp({ smsId: 'tls-id' }); return received; }), @@ -193,22 +192,29 @@ describe('tls', () => { }); test('logs a handshake the server turned away', async () => { - const warned = once(resolve => { - const log = new Log({ logLevel: 'warn', stderr: resolve, stdout: resolve }); - - void server({ log, port: 0, tls: { cert: certificate.cert, key: certificate.key } }) - .then(({ server: smpp }) => { - assert.ok(smpp); - - const sock = net.connect({ port: smpp.port }, () => { - sock.end('not a client hello'); - }); - - sock.on('close', () => { void smpp.close(); }); - sock.resume(); - }); + let onWarning: ((message: string) => void) | undefined; + const warned = once(resolve => { onWarning = resolve; }); + const log = new Log({ + logLevel: 'warn', + stderr: message => onWarning?.(message), + stdout: message => onWarning?.(message), + }); + const { err, server: smpp } = await server({ + log, + port: 0, + tls: { cert: certificate.cert, key: certificate.key }, }); + assert.equal(err, undefined); + assert.ok(smpp); + + const sock = net.connect({ port: smpp.port }, () => { sock.end('not a client hello'); }); + + sock.resume(); + assert.match(await warned, /client handshake failed/); + + sock.destroy(); + await smpp.close(); }); }); diff --git a/test/types.test.ts b/test/types.test.ts index 4dc158d..b53f2f3 100644 --- a/test/types.test.ts +++ b/test/types.test.ts @@ -155,6 +155,14 @@ describe('dest_address_array', () => { assert.deepEqual(target, encoded); }); + + test('refuses a field value the wire cannot hold instead of throwing', () => { + const badTon: DestAddress[] = [{ dest_addr_npi: 0, dest_addr_ton: 999, destination_addr: '123' }]; + const tooMany: DestAddress[] = Array.from({ length: 300 }, () => ({ dl_name: 'a' })); + + assert.ok(types.dest_address_array.write(badTon, Buffer.alloc(8), 0).err instanceof Error); + assert.ok(types.dest_address_array.write(tooMany, Buffer.alloc(901), 0).err instanceof Error); + }); }); describe('unsuccess_sme_array', () => { @@ -186,6 +194,18 @@ describe('unsuccess_sme_array', () => { assert.deepEqual(target, encoded); }); + + test('refuses a field value the wire cannot hold instead of throwing', () => { + const badStatus: UnsuccessSme[] = [ + { dest_addr_npi: 0, dest_addr_ton: 0, destination_addr: 'abc', error_status_code: 0x1FFFFFFFF }, + ]; + const badTon: UnsuccessSme[] = [ + { dest_addr_npi: 0, dest_addr_ton: 999, destination_addr: 'abc', error_status_code: 0 }, + ]; + + assert.ok(types.unsuccess_sme_array.write(badStatus, Buffer.alloc(11), 0).err instanceof Error); + assert.ok(types.unsuccess_sme_array.write(badTon, Buffer.alloc(11), 0).err instanceof Error); + }); }); describe('bounds checking', () => { diff --git a/todo.md b/todo.md index 3b3d428..3d7b15c 100644 --- a/todo.md +++ b/todo.md @@ -5,7 +5,7 @@ rules there constrain every item below. ## Status -The rewrite is **feature complete and green**: 155 tests, lint and typecheck clean, verified on Node +The rewrite is **feature complete and green**: 184 tests, lint and typecheck clean, verified on Node 18, 20, 22 and 24. What is left is release work and a few things worth adding before or after 1.0.0. ```bash