diff --git a/AGENTS.md b/AGENTS.md index b0e334a..d791e9d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,7 +1,7 @@ # AGENTS.md -Guidance for LLM agents working in this repository. Human-facing documentation lives in -[README.md](README.md); the remaining work is tracked in [todo.md](todo.md). +Guidance for LLM agents working in this repository. What each file in it is for is under +[Documentation](#documentation). ## What this is @@ -10,7 +10,35 @@ started from an orphan commit — no history from 0.4.0 is carried over. The 0.4 readable on the `master` branch of the same repository and is the reference for protocol behaviour, not for structure or style. -The library's value is its very small API. Do not grow the public surface without being asked. +## Goals + +In priority order, and the order is the point: where two of them pull against each other, the earlier +one wins. They do not override the hard rules below. + +1. **Correct on the wire.** SMPP 3.4 as SMSCs actually run it. Every other goal yields to this one; + the defect table below is what the alternative costs. +2. **Never give the application a wrong answer about what happened.** An outcome we cannot determine + is reported as undetermined rather than guessed; a request the peer may already have taken is + never re-sent on the library's own initiative; work the peer has no reason to send again is not + dropped. +3. **Strict in what we send, generous in what we read.** The library's own senders follow 3.4, and + the codec parses whatever arrives. Where the letter of the spec would discard traffic a real SMSC + sends, keep the traffic. +4. **A peer an operator never has to complain about.** No bind flooding, nothing a bind direction + forbids, no optional parameters to a peer that declared none, nothing held without a bound. +5. **The session layer is in here, and its defaults are what most applications should run.** + Keepalive, reconnect, the send window, reassembly and receipt correlation. An option retunes a + default or opts out of it; an option does not switch on the thing the caller obviously wanted. +6. **A small, stable public surface over reshapeable internals.** Only what `src/index.ts` exports is + published. A new option has to beat "the application can do this itself", and has to keep a + promise this library can verify. The low-level surface is a passthrough: policy binds what the + library composes, never what the caller wrote. +7. **Nothing that needs state wider than one session.** No throughput throttling, no persistence + across a restart, no coordination between processes. This is the scope floor, and it is why an + otherwise reasonable feature is declined without a fresh argument each time. +8. **It builds, tests and runs the same everywhere.** Container-only toolchain, no runtime + dependencies, the Node 18 floor verified in CI rather than asserted, every README example executed + by the suite. ## Hard rules @@ -38,18 +66,23 @@ src/ index.ts Public surface. Named exports only, no default export. client.ts client() -> { err, session } server.ts server() -> { err, server }, server owns the listener + close() - session.ts Session: dispatch, events, and the collaborators below + session.ts Session: the socket's life, dispatch, events, and the collaborators below 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 error-from.ts errorFrom(): whatever was thrown or rejected, as an Error expiring-groups.ts ExpiringGroups: the capped, expiring store both of those share + held-messages.ts HeldMessages: capped, expiring messages the application has not answered + idle-waiters.ts IdleWaiters: waiting for a count to fall to zero, and what is left of a budget incoming-requests.ts Every request the peer sends: messages, receipts, links, unknown commands + link-gate.ts LinkGate: where a request with no link to go out on waits for the next one link-timers.ts LinkTimers: the enquire_link heartbeat and the idle timeout log.ts SmppLog, the logger contract, and silentLog — the default message.ts Encoding detection, splitting, bit counting, SMPP date formatting + outgoing-requests.ts OutgoingRequests: the gate, the window, the pending map and the retry pdu.ts pduToObj / objToPdu / pduReturn — synchronous, result-returning pdu-framer.ts PduFramer: a byte stream cut into complete PDUs + pdu-transport.ts PduTransport: the socket a session reads complete PDUs off pending-requests.ts PendingRequests: sequence numbers, correlation, timeout, abort reassembly.ts Reassembler: capped, expiring multipart groups reconnect-loop.ts ReconnectLoop: backoff, retry timer, stopped-ness @@ -57,7 +90,9 @@ src/ send-sms.ts submitSms composition and the submitSmParams builder send-window.ts SendWindow: the maxOutstanding semaphore session-options.ts SessionOptions, ReconnectOptions, bind direction and the session defaults - udh.ts User data header: the concatenation fields of a long SMS + sms-id.ts The notation a peer writes message ids in, normalised for comparison + udh.ts User data header: the concatenation fields of a long SMS, and their reference + unanswered-error.ts UnansweredError: it went out and no answer came back uuid.ts uuidv7() — the ids the library generates for messages defs/ commands.ts The 33 commands, their ids and ordered parameter lists @@ -96,9 +131,10 @@ docker compose run --rm node npm run build ## Defects found in 0.4.0 -Confirmed by reading the 0.4.0 source. The rewrite fixes all of them; each needs a regression test -naming the behaviour, and the wire-affecting ones are cross-checked against a reference -implementation (see todo.md). +Every row names what 0.4.0's own code did, so it is not rebuilt here. +[README.md](README.md#behaviour-that-changed-on-the-wire) names what changed for a consumer, and is +the only place that does. Confirmed by reading the 0.4.0 source; each row has a regression test +naming the behaviour. | Defect | 0.4.0 behaviour | | --- | --- | @@ -115,13 +151,14 @@ implementation (see todo.md). | Unbounded reassembly | Incomplete long-SMS groups are capped by nothing and swept only when other traffic arrives, after 24 hours | | Dead DLR aggregation | `longSmsDlrs` is allocated to merge per-segment receipts and then never used | | Trailing NULL truncation | `types.buffer.size()` subtracts one whenever the value's last octet is `0x00`, so the PDU is allocated one octet short while `sm_length` still reports the full length. Any UCS2 message ending in a character like U+4E00 or U+3000 goes out corrupt | -| Dormant filters | `defs.filters` is declared on commands and TLVs but never invoked anywhere. Dropped in the rewrite; SMPP time formatting is exported as `smppTime` instead | -| Unchecked reads | Wire reads index straight into the buffer, so a short or malformed PDU throws out of the codec. Reads are bounds-checked and return results now | +| Dormant filters | `defs.filters` is declared on commands and TLVs but never invoked anywhere | +| Unchecked reads | Wire reads index straight into the buffer, so a short or malformed PDU throws out of the codec | | Unrangechecked writes | Integer params are handed to `writeUInt8`/`writeUInt16BE` unvalidated, so an out-of-range value throws from inside Node | | `submit_multi` missing `sm_length` | The field is commented out of the command table, so `short_message` never round-trips for that command | | Per-parameter defaults never applied | `calcCmdLength` reads `paramType.default` (the wire type's) rather than the parameter's, so `interface_version: 0x50` on the bind commands did nothing and every bind declared version 0x00 | | `source_telematics_id` width | Defined as a 2-octet integer; SMPP 3.4 5.3.2.8 makes it 1 octet, unlike `dest_telematics_id`, which really is 2 | -| Binary payloads decoded as text | `data_coding` 0x02, 0x04, 0x14 and 0xF4-0xF7 are 8-bit binary and land on the GSM 03.38 table, which rewrites every octet outside it. They resolve to LATIN1 now, so the payload survives as bytes | +| Binary payloads decoded as text | `data_coding` 0x02, 0x04, 0x14 and 0xF4-0xF7 are 8-bit binary and land on the GSM 03.38 table, which rewrites every octet outside it | +| Binary TLVs round-trip corrupt | `pduToObj` turns a `Buffer` TLV value into a hex string (`utils.js:307`), and `objToPdu` writes that string back as its own ASCII, so `message_payload`, `network_error_code`, `callback_num` and the rest are destroyed by any round trip | | `ESME_RINVBCASTCHANIND` typo | Defined as `0x011`, three hex digits; the spec value is `0x0112` | ## Multipart sends and the send window @@ -167,61 +204,214 @@ exactly 140. - `assert.equal` from `node:assert/strict` narrows its first argument, so a following `?.` on the same value is flagged as unnecessary. Assert once with `assert.ok(x)` and use plain access after. +## Documentation + +Each file answers one question, and a fact belongs to the file whose question it answers: + +- **README.md — what you can rely on.** Observable behaviour, for someone using the package. It + carries a reason only where the reason changes how you would call the thing. +- **AGENTS.md — what may not change, and why.** Goals, hard rules, architecture, conventions, and the + decisions the goals do not already settle. It does not restate behaviour README states. +- **todo.md** is a temporary working file that sets its own rules; nothing here governs it. + +A sentence living in two of them is a defect: delete the copy in the file whose question it does not +answer. The toolchain commands are the one deliberate exception — README's copy serves a contributor +who never opens this file, and this file's copy carries the constraint that nothing runs on the host. + +**Write a decision down only when it cannot be put better as a goal.** A goal decides every case that +follows from it; a decision record decides one. So reach for the goal list first — sharpen a goal, +add one, or move one up the order — and write a decision only for what is left over: a choice a +competent change would otherwise re-open, that no goal implies. Give the claim, the constraint that +settled it and the alternative rejected, and nothing the code or README already says. Where a +compiler or a test already forbids the other way, it is not a decision, it is a test name. Delete one +once it no longer constrains anything; this is not a changelog. + ## Decisions +Grouped by what each one constrains. + +### The public surface + +- **`Session` is publicly constructible, which is what makes `SessionOptions` and `ReconnectOptions` + public too.** Raised twice as a leak; it is not one. The collaborators `session.ts` delegates to + (`Reassembler`, `PendingRequests`, `SendWindow`, `ReconnectLoop`, `LinkTimers`, `LinkGate`, + `DlrMerger`, `PduTransport`, `submitSms`) stay unpublished so they can be reshaped. + +- **`acceptsOptionalParams()` and `bindAllows()` are predicates, not chokepoints.** The library's own + senders consult them; `session.send({ tlvs })` is passed through as written, because silently + stripping a caller's explicit TLVs off a deliberately public low-level surface would be worse than + sending them. Only `submit_sm` and `deliver_sm` are policed by bind direction — the only two the + library sends and dispatches by it. + +- **`session.sock` is a getter over `PduTransport`.** Reading it is unchanged; assigning it no longer + compiles, which never rewired the handlers and so never worked. + +- **Both emitters re-declare their listener methods to accept a promise.** Maintainer's call, + 2026-08-27: `EventEmitter` types every listener as void-returning, so the + `session.on('sms', async sms => …)` README documents reads as a misused promise in any strict + consumer. `declare on: …` and its six siblings re-type the inherited methods to return `unknown`, + which emits nothing and needs no cast; overriding them as real methods cannot work, because the + `super.on()` call needs one. The cost is that a subclass can no longer reach those seven through + `super` — re-declaring them the same way is its way out. `unknown` rather than + `void | Promise` because a listener may return anything: `session.on('close', () => + set.delete(session))` returns a boolean. This also settles what the drain can wait on: a listener's + own promise would be the better completion signal, and reaching it needs `listeners()`, which + cannot be re-declared the same way — Node types it invariantly enough that widening `void` to + `unknown` is `TS2416`. Re-probed 2026-09-01; `sendResp()` stays the signal. + +### The wire + +- **The declared interface version is an option on both `client()` and `server()`, and is not the + optional-parameter threshold.** That threshold is fixed at 0x34 by the spec, so an implementation + that must declare 5.0 throughout can, without moving it. + +- **A peer that declared no version is pre-3.4, and `undefined` means no bind yet.** `acceptBind()` + records what the ESME declared and the client's `bind()` records the `sc_interface_version` the + SMSC answered with; a peer that declared nothing is recorded as `undeclaredInterfaceVersion` (0x00) + and sent no optional parameters, which is how the spec reads an absent `sc_interface_version`. + +- **`esm_class` decides what a `deliver_sm` is, and the body is read only when it names nothing.** + `MC_DELIVERY_RECEIPT` (0x04) makes it a receipt whatever the body parses to, so a receipt in a + format `dlrFromPdu()` cannot read reaches `dlr` with `smsId` undefined instead of arriving as an + inbound SMS. Any other named type is not a receipt and its body is not scraped; a message type of 0 + or one of the ten reserved keeps the scrape, and a non-empty `receipted_message_id` TLV marks a + receipt on the same footing. A receipt this library recognises never reaches the reassembler, so an + SMSC that splits one across segments gets a `dlr` per segment rather than one merged report. The + `message_state` TLV is authoritative only where it names a state in the table — SMPP reserves + 0x80-0xFF for MC-vendor-specific values, so an unnameable one keeps its raw `statusId` and leaves + `statusMsg` to the body. + +- **`smsIdFormat` names a notation per place, and normalisation never reaches inside a `-` + id.** An SMSC may answer `submit_sm_resp` in hex and write the receipt's `id:` in decimal, so one + transform over both sides cannot make them equal. `submitResp` covers the `receipted_message_id` + TLV too, which SMPP 3.4 5.3.2.26 defines as the id the `submit_sm_resp` carried: naming one + notation for whichever id a receipt yields would break the peer that sends both. Omitting a place + is what leaving it alone means, so there is no `raw` notation, and a caller-supplied formatter is + refused because it would make the promise that the two ids are comparable unverifiable — `onRequest` + and the PDU on the `dlr` event are the escape hatches. A `-` id parses as no number and so + reaches `expect()` and `collect()` unchanged, which is what keeps `DlrMerger` working; normalising + the base instead would break that pair. The option is on `client()` only, since a `server()` session + writes both ids itself. + +### The session's life + - **A close arriving after our own `unbind` is a clean unbind, not an error.** Maintainer's call, 2026-08-26: most SMSCs drop the socket instead of answering, so the documented shutdown would otherwise always report a failure. It does mask a socket that died mid-unbind for an unrelated reason, which is accepted — the peer sees the same TCP close either way. -- **The published surface is frozen at what `src/index.ts` exports today.** `Session` is exported and - publicly constructible, which is why `SessionOptions` and `ReconnectOptions` are public too — that - is correct, not a leak, and it has been raised twice. The collaborators `session.ts` delegates to - (`Reassembler`, `PendingRequests`, `SendWindow`, `ReconnectLoop`, `LinkTimers`, `DlrMerger`, - `submitSms`) stay unpublished so they can be reshaped. -- **The sub-3.4 optional-parameter rule is a predicate, not a chokepoint.** `acceptsOptionalParams()` - is consulted by the library's own senders; `session.send({ tlvs })` is passed through as written, - because silently stripping a caller's explicit TLVs off a deliberately public low-level surface - would be worse than sending them. The guarantee is "what this library sends honours the rule", - never "the session cannot send optional parameters to an old peer". -- **Both ends feed `peerInterfaceVersion`, and a peer that declared nothing is pre-3.4.** - `acceptBind()` records what the ESME declared in its bind request; the client's `bind()` records - the `sc_interface_version` the SMSC answered with. A peer that declared no version is recorded as - `undeclaredInterfaceVersion` (0x00) and is sent no optional parameters — the spec reads an absent - `sc_interface_version` as an SMSC that supports none. `undefined` is left to mean one thing only: - no bind has been accepted on this session yet. -- **The library speaks SMPP 3.4 on the wire, and `defs/` keeps the 5.0 tables as a superset.** - Maintainer's call, 2026-08-26: 3.4 is what SMSCs actually run, while the wider tables let the codec - parse and build whatever a peer sends. The declared version is an option on both `client()` and - `server()`, so an implementation that needs 5.0 throughout can have it. The threshold at or above - which a peer may be sent optional parameters is fixed at 0x34 by the spec and is not the same - constant as the declared version. -- **Bind direction is enforced on the library's own senders and on everything incoming, not on - `send()`.** A receiver-bound ESME carries no `submit_sm` and a transmitter-bound one no - `deliver_sm`; `sendSms()` and `sendDlr()` refuse locally, and an arriving PDU is answered - `ESME_RINVBNDSTS`. `bindAllows()` is a predicate on the same footing as `acceptsOptionalParams()`, - so the deliberately public low-level `send()` stays a passthrough. Only those two commands are - policed, because they are the only ones the library sends and dispatches by direction. -- **The logger is a five-method contract this library declares, not a dependency.** `SmppLog` in - `log.ts` is what the code actually calls (`debug`, `error`, `info`, `verbose`, `warn`), so an - application can satisfy it with an object literal and `@larvit/smpp` ships with no runtime - dependencies. `@larvit/log` implements it structurally and stays a devDependency, where - `test/tls.test.ts` passing a real `Log` as the server's logger keeps that compatibility compiled. +- **`close` means the session is over, and a drop the loop will retry is `disconnected`.** + Maintainer's call, 2026-08-31: without the split, an application that opens a replacement client on + `close` ends up holding two binds on one account. `teardown()` picks the event by whether the + reconnect loop is still live, and `end()` stops that loop before tearing down, so every deliberate + shutdown emits `close`. A retry that opens a socket and then loses it clears `closed` through + `attach()`, which is why a second drop emits again. -- **`esm_class` decides what a `deliver_sm` is, and the body is only read when it names nothing.** - Message type `MC_DELIVERY_RECEIPT` (0x04) makes it a receipt whatever the body parses to, so a - receipt in a format `dlrFromPdu()` cannot read reaches `dlr` with `smsId` undefined instead of - arriving as an inbound SMS. Any other named type — delivery or user acknowledgement, conversation - abort, intermediate notification — is not a receipt and its body is not scraped. A message type of - 0, or one of the ten the spec reserves, keeps the scrape: SMSCs that send text-only receipts leave - `esm_class` at 0, and reading that as the spec's "default message type" would lose every one of - them. A non-empty `receipted_message_id` TLV marks a receipt on the same footing there, since - nothing but a receipt carries one. What gets scraped is the decoded `short_message` with any UDH - stripped; a receipt this library recognises never reaches the reassembler, so an SMSC that splits one across segments gets a - `dlr` per segment rather than one merged report. - The `message_state` TLV is authoritative only where it names a state in the table — SMPP reserves - 0x80-0xFF for MC-vendor-specific values, so an unnameable one keeps its raw `statusId` and leaves - `statusMsg` to the body. +- **`reconnect` takes `{ minDelay, maxDelay }` to retune and `false` to turn off**, so absent means + on and there is one spelling for each. Only `client()` reconnects — a `server()` session is a + connection the peer opened, and nothing at this end can reopen it. The retry timer is `unref()`'d, + so a process with nothing else left to do still exits between attempts. + +- **Coming up is not proof a link works, so only one that outlasted `maxDelay` resets the backoff.** + An unreadable stream is found after the bind returns, so resetting on connect gave a link that died + on arrival a fresh `minDelay` every cycle — one TCP connect and bind per second, forever. A drop + after a healthy link still retries at `minDelay`. + +- **A stream this library cannot read is a dead link, not a dead session.** Maintainer's call, + 2026-08-31: a framing or codec error tears the link down through `teardown()`, so the reconnect + loop retries it on a fresh socket with a fresh framer — which is what a desynced stream needs, and + the common cause. `sessionError` still carries every failure, so a peer that only ever sends + garbage is visible in the log rather than silent. + +- **A deliberate shutdown drains; an unusable link and an abort do not.** `close()` and `unbind()` + wait on the send window rather than the pending map — the map misses a segment still queued behind + a full window, and finishing a half-sent multipart message is the point. The window counts slots, + never outcomes, and empties on a drop too, where `teardown()` settles everything the link was + carrying, which is why `drain()` reads `closed` before it reads the count. A stream the framer or + the codec cannot read takes `teardown()` instead, and `close({ signal })` on an aborted signal and + a peer's own `unbind` take `end()`: nothing on a dead link can answer, an abort means stop now, and + a peer that has declared itself finished will not answer what it still owes, so draining any of the + three would only hold a socket open for the timeout. `unbind()` sends its own PDU through + `request()` past both the window and the drain gate, because it must go out either way. + `shutdownTimeout` stays a session option rather than a `close()` argument: `server()` builds + sessions on the caller's behalf, so the option is the only composition point. `SmppServer.close()` + reports each session's unfinished drain through `serverError`, because its own result says nothing + but that the listener stopped. + +- **The drain waits on the messages the application holds, and `sendResp()` is what says it is done + with one.** Maintainer's call, 2026-09-01: waiting on the send window alone tore a server session + down while the application was still answering a `submit_sm`, so the peer timed out and re-sent — + the duplicate goal 2 forbids, in the direction the window already covers. No completion signal was + added to the `sms` event: `sendResp()` is the answer the peer is waiting for, so it is the one the + drain waits for. Counting every inbound request until `sendReturn()` answered it was rejected — + an `onRequest` that deliberately answers nothing would then cost a full `shutdownTimeout` on every + close — and a message no listener took is released at once, since nothing is going to answer it. + `teardown()` drops what is still held for the same reason it drops inbound segments. The release + is one turn late, so a listener that sends its receipt straight after the response is still + holding when the drain looks; `sendDlr()` is the one send that goes out past the drain's refusal, + and only while the message is still held — past that it is an ordinary send, because the drain it + would slip past is no longer waiting for it. `shutdownTimeout: 0` does not carry over to this half: + waiting forever is safe for the peer, whose every request is bounded by `responseTimeout` unless the + caller set that to 0 as well, and unsafe for the application, which nothing bounds — `close()` is + what you reach for when the application is stuck, so it may not block on the application coming + unstuck. That half falls back to `responseTimeout`, the same answer the link gate's hold already + takes — and to that option's default where it is 0 as well, since neither option is an answer about + the application. What is held is capped and expiring like every other inbound store, on constants + rather than options, because a bound the application cannot raise is the point: an application that + answers nothing would otherwise grow it for the life of the link, which goal 4 forbids. A message + that falls out of the bound is one the drain stops waiting for, so `close()` can report fewer + unanswered than there were — accepted, because the alternative is holding what nothing will answer, + and both exits are logged. + +- **A reconnect keeps the delivery-receipt merges; everything else the link held is dropped.** + `onDeliverSm()` answers each receipt before the group it belongs to is complete, and `teardown()` + runs on every path — an idle timeout and a failed rebind, not only `close()` — so clearing the + merges there loses receipts no peer has a reason to send again. They are cleared where the session + is over instead. Inbound segments stay in `teardown()`, because they go unanswered until the + message is whole: the peer still holds them, and answering it on a later link with the old + segments' sequence numbers would correlate with nothing. + +- **A message id base is merged at most once.** A receipt carries nothing but `-`, so a + straggler for a message whose group is gone cannot be told from a receipt for a later message the + peer handed the same ids — an SMSC whose id counter restarts with its process is the realistic + case. `DlrMerger` remembers the bases it has finished with, capped and expiring exactly like the + groups, and refuses to open one a second time: the later message gets no `messageDlr`, and an + earlier one whose receipts are still arriving is dropped rather than left to collect the later + one's. Every segment still reaches the application as a `dlr`. `expect()` ignores a lone id, so a + single-part message never claims a base. + +- **A send that never reached the socket waits for the next link; one that did is counted, not + resent.** Maintainer's call, 2026-09-01: re-queueing everything unanswered would resend a + `submit_sm` the SMSC accepted and answered into a dead socket, which is delivered and billed twice, + while a request that never left this process can be lost for free. `attempt()` therefore wraps all + three ways a written request can fail in `UnansweredError`; counting only the dropped-link case, as + the first cut did, would have called the commonest one safe to resend. A count rather than a + boolean because `sendSms()` aggregates segments into one `err` slot, and required rather than + optional so every construction site answers. `UnansweredError` stays unexported: `unanswered` is + the one spelling on the public surface. The hold is bounded by `responseTimeout` rather than an + option of its own — that is already the answer to how long one request may wait — and its clock + starts when the send is issued rather than when it first finds the gate shut, so one budget covers + every hold a single call makes. That timer is the one here that is not `unref()`'d: a held request + is awaited with the socket already destroyed, so an unref'd one lets a process whose only remaining + work is that send exit without settling it. + +- **The gate decides whether a link can carry a request, and a bind is what makes it one.** + Maintainer's call, 2026-09-01: `attach()` clears `closed` the moment a socket is handed over, one + round trip before the bind is answered, so gating on `closed` let a send arriving in that window go + out unbound and come back `ESME_RINVBNDSTS`. `LinkGate` owns the answer instead — `shut(returning)` + on every teardown, `open()` only once `comeBackUp()` has a bound link — and + `OutgoingRequests.linkDown()` reads it rather than `closed`. The bind itself cannot wait for what it + creates, so `pastDrain()` lets the three bind commands past the gate and the window, the same door + `unbind()` takes through `now()`. The gate is told what happened and never reads back into the + session: a collaborator that has to ask does not own its decision, which is how the first cut ended + up answering the same question two different ways at admit and at release. For the same reason the + retry in `pastDrain()` asks `gate.isUp()` rather than `linkDown()`, which also reads the socket — a + condition that loops on something the gate does not gate on spins against a gate that admits it + straight back. `LinkGate.returning` is a copy of `retrying()` taken at teardown, and stays true + only because nothing stops the reconnect loop without `emitClose()` following it: `drain()` and + `end()` are the only callers of `stop()`. A third caller has to shut the gate itself. + +### Internals and tests - **A listener that rejects is routed by Node's `captureRejections`, not by hand-dispatching.** Both emitters construct with `captureRejections: true` and implement @@ -232,57 +422,13 @@ exactly 140. handlers normalise through `errorFrom()` rather than inline — a route out of the handler would land on a bare `process.nextTick` with nothing to catch it. -- **Both emitters re-declare their listener methods to accept a promise.** Maintainer's call, - 2026-08-27: `EventEmitter` types every listener as void-returning, so the - `session.on('sms', async sms => …)` the README documents reads as a misused promise in any strict - consumer. `declare on: …` and its six siblings re-type the inherited methods to return `unknown`, - which emits nothing, needs no cast and leaves the runtime method on the prototype. Overriding them - as real methods instead cannot work: the `super.on()` call needs a cast to satisfy the conditional - `Listener` type. The cost is that a subclass can no longer reach those seven through `super` or - override them as methods — re-declaring them the same way is its way out. `unknown` rather than `void | Promise` because a listener may return - anything — `session.on('close', () => set.delete(session))` returns a boolean. - -- **A reconnect keeps the delivery-receipt merges; everything else the link held is dropped.** - `onDeliverSm()` answers each receipt before the group it belongs to is complete, and `teardown()` - runs on every path — an idle timeout and a failed rebind, not only `close()` — so clearing the - merges there loses receipts no peer has a reason to send again. They are cleared where the session - is over instead: `close()`, or a drop with no reconnect loop left to bring the link back. Inbound - segments stay in `teardown()`, because they go unanswered until the message is whole: the peer - still holds them, and answering it on a later link with the old segments' sequence numbers would - correlate with nothing. Surviving a process restart is a separate, public-surface question, and is - in todo.md. - -- **A message id base is merged at most once.** A receipt carries nothing but `-`, so a - straggler for a message whose group is gone cannot be told from a receipt for a later message the - peer handed the same ids — an SMSC whose id counter restarts with its process is the realistic - case. `DlrMerger` remembers the bases it has finished with, capped and expiring exactly like the - groups, and refuses to open one a second time: the later message gets no `messageDlr`, and an - earlier one whose receipts are still arriving is dropped rather than left to collect the later - one's. Every segment still reaches the application as a `dlr`. The rule covers the bases the - merger opened — `expect()` ignores a lone id, so a single-part message never claims one. - -- **A deliberate shutdown drains; an unusable link and an abort do not.** `close()` and `unbind()` - refuse further sends and wait on the send window, not the pending map — the map misses a segment - still queued behind a full window, and finishing a half-sent multipart message is the point. The - window counts slots, never outcomes, so it says when to stop waiting and nothing about what - happened: it empties on a drop too, where `teardown()` settles everything the link was carrying, - which is why `drain()` reads `closed` before it reads the count. The wait covers what this end - sent — a request the peer sent us is answered through `sendReturn()`, which never enters the - window, so a server session waits for none of its inbound work; that half is in todo.md. - `shutdownTimeout` bounds the drain alone, and `0` waits forever like every other timeout here; - `unbind()` then waits `responseTimeout` for its own response, and sends that PDU through - `request()` past both the window and the drain gate because it must go out either way. A stream - the framer or the codec cannot read takes `end()` instead, and so does `close({ signal })` on an - aborted signal and a peer's own `unbind` — nothing on a dead link can answer, an abort means stop - now, and a peer that has declared itself finished will not answer what it still owes us, so - draining any of the three would only hold a socket open for the timeout. `SmppServer.close()` - reports each session's unfinished drain through `serverError`, because its own result says - nothing but that the listener stopped. `shutdownTimeout` stays a session option rather than a - `close()` argument: `server()` builds sessions on the caller's behalf, so the option - is the only composition point, and `close({ signal })` already covers a hard deadline. +- **`SmppLog` is a five-method contract this library declares, not a dependency.** `debug`, `error`, + `info`, `verbose` and `warn` are what the code actually calls, so an application can satisfy it + with an object literal. `@larvit/log` implements it structurally and stays a devDependency, where + `test/tls.test.ts` passing a real `Log` as the server's logger keeps that compatibility compiled. - **The TLS tests build their own self-signed certificate in DER** (`test/tls.test.ts`) instead of adding a devDependency or shelling out to openssl. Maintainer's call, 2026-08-26: the dev image - `node:24.18.0-bookworm-slim` ships no openssl binary, so a shelled-out fixture would pass in CI - and fail on every developer machine, and a committed key leaks in a public repository. Valid while - the dev image has no openssl. + `node:24.18.0-bookworm-slim` ships no openssl binary, so a shelled-out fixture would pass in CI and + fail on every developer machine, and a committed key leaks in a public repository. Valid while the + dev image has no openssl. diff --git a/README.md b/README.md index 38dc3cd..51acdc6 100644 --- a/README.md +++ b/README.md @@ -17,11 +17,11 @@ by hand on top of a library; it is built in here. | | | | --- | --- | | **Keepalive** | `enquire_link` every 20 s on a quiet link, and a peer that stops answering is dropped. | -| **Reconnect with backoff** | Opt-in `reconnect` reopens the socket and re-binds, 1 s doubling to 30 s. | +| **Reconnect with backoff** | A dropped client link reopens the socket and re-binds by default, 1 s doubling to 30 s. | | **Submit window** | `maxOutstanding` holds requests in flight at 10; further sends queue instead of overrunning the SMSC. | | **Delivery receipts** | Correlated by `receipted_message_id`/`message_state` where the SMSC sends them, falling back to parsing the receipt text — what Kannel and several others send. | | **Multipart** | Long messages split on send; concatenated `deliver_sm` reassembled into one `sms`. | -| **Graceful shutdown** | `close()` and `unbind()` wait out the requests this end already sent, so a submit the SMSC accepted is not reported as a failure. | +| **Graceful shutdown** | `close()` and `unbind()` wait out the requests this end already sent and the messages the application has not answered yet, so neither end has to guess whether a message got through. | | **Never throws** | Everything fallible resolves to `{ err?, … }`, the codec included. | Throughput throttling is deliberately absent: an operator's rate limit is scoped to the account, and @@ -99,11 +99,12 @@ 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; `0` waits forever. | -| `shutdownTimeout` | `5000` | How long `close()` and `unbind()` wait for the requests this end already sent; `0` waits forever. | +| `idleTimeout` | `2 × enquireLinkInterval` | Give up on a link the peer has stopped answering, and re-bind unless `reconnect` is `false`. | +| `responseTimeout` | `30000` | How long to wait for a response before giving up on it, and how long a send with no link waits for the next one; `0` waits forever. | +| `shutdownTimeout` | `5000` | How long `close()` and `unbind()` wait for the requests this end already sent and the messages the application has not answered. `0` waits forever for the requests, which end when the peer answers or `responseTimeout` expires — so setting both to `0` never ends. The messages fall back to `responseTimeout`, or to its default where that is `0` too, since nothing but the application ends that wait. | | `maxOutstanding` | `10` | Requests allowed on the wire at once; further sends queue. | -| `reconnect` | off | `{ minDelay, maxDelay }` to re-bind automatically after a drop or an idle timeout, with exponential backoff. | +| `smsIdFormat` | — | The notation the SMSC writes message ids in, per place it writes them: `{ receipt: 'decimal', submitResp: 'hex' }`. Only needed where the two disagree. | +| `reconnect` | on | Re-binds after a drop, an idle timeout, or a stream the library cannot read, backing off from `minDelay` 1 s to `maxDelay` 30 s and starting over at `minDelay` once a link has lasted `maxDelay`. `{ minDelay, maxDelay }` retunes it; `false` turns it off, so a drop ends the session. | | `log` | silent | Any object with `debug`, `error`, `info`, `verbose` and `warn` methods — see [Logging](#logging). | | `signal` | — | An `AbortSignal` that cancels connecting and tears the session down. | @@ -134,14 +135,17 @@ Messages too long for one SMS are split automatically and sent as a concatenated one id per segment: ```javascript -const { err, pduObjs, smsIds } = await session.sendSms({ from, message, to }); +const { err, pduObjs, smsIds, unanswered } = 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, `pduObjs` and `smsIds` then hold what the SMSC did accept — enough to reconcile against a later receipt, not enough to resend the rest, so treat a partial failure as a -failed message. A message needing more than 255 segments is refused before anything is sent, since -the concatenation header numbers segments in a single octet. `maxSegments` lowers that ceiling: +failed message. `unanswered` counts the segments that went out and were never answered: the SMSC may +have taken each of them and lost only the response, so a message with `unanswered` above zero cannot +be sent again without risking a duplicate, however empty `smsIds` is. A message needing more than 255 +segments is refused before anything is sent, since the concatenation header numbers segments in a +single octet. `maxSegments` lowers that ceiling: most handsets and SMSCs stop well short of 255, and refusing beats a message only half delivered. ### Receiving @@ -161,6 +165,21 @@ to tell the two apart. `esm_class` is what tells them apart; where it names no m `receipted_message_id` TLV does, and failing both the message body is read for the standard `id:` and `stat:` receipt fields. +Matching a receipt to a send means comparing `dlr.smsId` against the `smsIds` that `sendSms()` +returned. Some SMSCs write the two in different notations — a hex `message_id` on the +`submit_sm_resp` and a decimal `id:` in the receipt, or one of them zero-padded — and the comparison +then quietly matches nothing at all. Name each notation and both ids are read into plain decimal +before you see them: + +```javascript +const { err, session } = await client({ smsIdFormat: { receipt: 'decimal', submitResp: 'hex' } }); +``` + +`receipt` is the notation of the receipt body's `id:` field, `submitResp` that of the `message_id` +a `submit_sm_resp` carries — and of a receipt's `receipted_message_id` TLV, which is that same id. +An id that is not a number in the notation given is left exactly as it arrived, and the PDUs carry +what the peer wrote either way — `pduObjs` from the send, and the second argument of the `dlr` event. + ## Server The simplest possible server — no authentication, listening on port 2775: @@ -300,8 +319,9 @@ TypeScript users can import `SmppLog` to have the compiler check one. | `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. `smsId` is undefined when the peer marked a receipt whose body carries no readable id. `statusMsg` names `statusId` unless the peer sent a `message_state` this library cannot name — then `statusId` is that raw value and `statusMsg` is whatever the body said, or `UNKNOWN`. | | `messageDlr` | Every segment of a multipart message sent with `dlr: true` has been reported on, carrying the worst status of the segments. Merging needs the SMSC to number its segment ids `-`, which is this library's own server's convention — an SMSC that hands out unrelated ids per segment never fires it. A base is merged once: a later message the SMSC gives the same ids is reported on through `dlr` alone, and an earlier one still collecting loses its merged report as well. | -| `close` | The connection closed. | -| `reconnected` | The client re-bound after a drop (only with `reconnect` configured). | +| `close` | The session is over, because nothing will bring the link back. Fires once, whether you closed it or the link failed for good. | +| `disconnected` | The link dropped and the reconnect loop will retry it. Do not open a replacement client here — the session you hold comes back on its own, and `reconnected` says when. Fires again for each attempt that reconnects and then fails, so it is not one-to-one with `reconnected`. | +| `reconnected` | The client re-bound after a drop. | | `sessionError` | Something failed on a live session, including a hook or listener that threw or, if it was `async`, rejected. | | `data` | Raw bytes arrived on the socket. | | `incomingPdu` | A complete PDU arrived, as a buffer. | @@ -311,8 +331,12 @@ TypeScript users can import `SmppLog` to have the compiler check one. `sendSms()`, `send()`, `sendReturn()`, `unbind()` and `close()`. Both `close()` and `unbind()` refuse further sends, wait out the requests this end already sent for up to `shutdownTimeout`, and -then tear down whatever is left, resolving to an `err` that says what was lost. A request the peer -sent *us* is answered through `sendReturn()` and is not waited for. `close({ signal })` takes an +then tear down whatever is left, resolving to an `err` that says what was lost. They also wait for +every `sms` the application has not called `sendResp()` on, so a peer whose `submit_sm` is still +being handled is answered rather than left to re-send it — answering its PDUs through `sendReturn()` +instead leaves that wait running until it gives up. `sendDlr()` is the one send the refusal lets +past, and it catches the wait when issued straight after `sendResp()`; await anything in between and +it races the shutdown like any other send. `close({ signal })` takes an `AbortSignal` that cuts the wait short; `unbind()` takes none, and waits a further `responseTimeout` for its own response. `send()` reaches any of the 33 SMPP commands the codec knows, not just the four the session handles natively: @@ -324,6 +348,17 @@ const { err, pduObj } = await session.send({ }); ``` +A send issued while the link is down waits for the reconnect instead of failing, and goes out once +the new link is bound — up to `responseTimeout`, after which it gives up having sent nothing. A +request already on the wire is the other case: the SMSC may have taken it and lost only the response, +so it fails, and `sendSms()` and `sms.sendDlr()` count it in `unanswered`, whether the link dropped +under it, the peer never answered in time, or you aborted it after it went out. Neither applies with `reconnect: false`, +where a drop ends the session and every send after it is refused. + +`responseTimeout` bounds the wait for a link and the wait for an answer separately, and a send also +queues for a `maxOutstanding` slot, which nothing bounds — so it is not a deadline for the call. +Pass `{ signal: AbortSignal.timeout(ms) }` when you need one. + `acceptsOptionalParams()` answers whether the peer declared SMPP 3.4 or later, which is the version at and above which the spec allows optional parameters to be sent to it; `peerInterfaceVersion` is the version it declared, `0x00` if it declared none. The library's own senders consult the first before attaching a TLV — a diff --git a/src/client.ts b/src/client.ts index 58ac4ca..20c7fd9 100644 --- a/src/client.ts +++ b/src/client.ts @@ -1,7 +1,8 @@ import type { ConnectionOptions } from 'node:tls'; import type { Result, VoidResult } from './result.ts'; -import type { BindType } from './session-options.ts'; +import type { BindType, ReconnectOptions } from './session-options.ts'; import type { SmppLog } from './log.ts'; +import type { SmsIdFormat } from './sms-id.ts'; import type { Socket } from 'node:net'; export type { BindType }; @@ -25,10 +26,11 @@ export type ClientOptions = { maxOutstanding?: number; password?: string; port?: number; - reconnect?: { maxDelay?: number; minDelay?: number }; + reconnect?: { maxDelay?: number; minDelay?: number } | false; responseTimeout?: number; shutdownTimeout?: number; signal?: AbortSignal; + smsIdFormat?: SmsIdFormat; systemType?: string; tls?: ConnectionOptions | boolean; username?: string; @@ -137,6 +139,19 @@ async function bind(session: Session, options: ClientOptions): Promise openSocket(options), + maxDelay: tuning.maxDelay, + minDelay: tuning.minDelay, + onConnected: reconnected => bind(reconnected, options), + }; +} + function createSession(options: ClientOptions, log: SmppLog, sock: Socket): Session { const enquireLinkInterval = options.enquireLinkInterval ?? defaults.enquireLinkInterval; @@ -145,19 +160,11 @@ function createSession(options: ClientOptions, log: SmppLog, sock: Socket): Sess idleTimeout: options.idleTimeout ?? enquireLinkInterval * defaults.idleTimeoutFactor, log, maxOutstanding: options.maxOutstanding, + reconnect: reconnectFor(options), responseTimeout: options.responseTimeout, shutdownTimeout: options.shutdownTimeout, + smsIdFormat: options.smsIdFormat, sock, - ...(options.reconnect - ? { - reconnect: { - connect: () => openSocket(options), - maxDelay: options.reconnect.maxDelay, - minDelay: options.reconnect.minDelay, - onConnected: reconnected => bind(reconnected, options), - }, - } - : {}), }); } @@ -165,7 +172,7 @@ async function connect(options: ClientOptions, log: SmppLog): Promise number) | undefined; + timeout: number; +}; + +/** The peer's own sequence number, which is what our answer to this message will carry. */ +function keyOf(pduObjs: PduObject[]): string | undefined { + const first = pduObjs[0]; + + return first ? String(first.seqNr) : undefined; +} + +/** The messages handed to the application that it has not answered yet, held by their segments. */ +export class HeldMessages { + private readonly held: ExpiringGroups; + private readonly idleWaiters = new IdleWaiters(); + private readonly log: SmppLog; + private readonly max: number; + + constructor(options: HeldMessagesOptions) { + this.held = 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.held.size; + } + + /** An application that answers no message at all may not grow this without end. */ + hold(pduObjs: PduObject[]): void { + const key = keyOf(pduObjs); + + if (key === undefined) return; + + this.sweep(); + + if (this.held.get(key)) { + this.log.warn('heldMessages - replacing a message on a re-used sequence number', { seqNr: Number(key) }); + } else if (this.held.full) { + this.dropOldest(); + } + + this.held.set(key, pduObjs); + } + + /** Whether a drain is still waiting for this message to be answered. */ + has(pduObjs: PduObject[]): boolean { + const key = keyOf(pduObjs); + + return key !== undefined && this.held.get(key) === pduObjs; + } + + release(pduObjs: PduObject[]): void { + const key = keyOf(pduObjs); + + // Identity, not the key: a wrapped sequence number must not release someone else's message. + if (key === undefined || this.held.get(key) !== pduObjs) return; + + this.held.delete(key); + this.settle(); + } + + /** Drops every message: their segments went with the link, so no answer of ours correlates now. */ + clear(): void { + this.held.clear(); + this.idleWaiters.settle(); + } + + /** Resolves 0 once every message has been answered, or with how many have not. */ + idle(timeout: number, signal: AbortSignal | undefined): Promise { + return this.idleWaiters.wait(() => this.held.size, timeout, signal); + } + + private dropOldest(): void { + const oldest = this.held.takeOldest(); + + if (!oldest) return; + + const [seqNr] = oldest; + + this.log.warn('heldMessages - buffer full, dropping the oldest message', { + max: this.max, + seqNr: Number(seqNr), + }); + } + + /** Drops every message past its deadline. Runs before each hold and on its own timer. */ + sweep(): void { + const expired = this.held.takeExpired(); + + if (expired.length === 0) return; + + this.log.warn('heldMessages - messages the application never answered', { + messages: expired.length, + }); + this.settle(); + } + + private settle(): void { + if (this.held.size === 0) this.idleWaiters.settle(); + } +} diff --git a/src/idle-waiters.ts b/src/idle-waiters.ts new file mode 100644 index 0000000..dd29a9e --- /dev/null +++ b/src/idle-waiters.ts @@ -0,0 +1,47 @@ +/** What is left of a budget, in the shape a wait takes it: 0 waits forever. */ +export function leftOf(deadline: number): number { + return deadline === 0 ? 0 : Math.max(1, deadline - Date.now()); +} + +/** Everything waiting for a count to fall to zero, and how such a wait is cut short. */ +export class IdleWaiters { + private readonly waiting: (() => void)[] = []; + + /** Wakes everything waiting, whatever the count reads now. */ + settle(): void { + for (const resolve of this.waiting.splice(0)) { + resolve(); + } + } + + /** + * Resolves 0 once nothing is left, or with what still is when the timeout or the signal cuts the + * wait short. A timeout of 0 waits forever. + */ + wait(remaining: () => number, timeout: number, signal: AbortSignal | undefined): Promise { + if (remaining() === 0) return Promise.resolve(0); + + if (signal?.aborted === true) return Promise.resolve(remaining()); + + return new Promise(resolve => { + let timer: NodeJS.Timeout | undefined = undefined; + const done = (): void => { + const index = this.waiting.indexOf(done); + + if (timer) clearTimeout(timer); + if (index !== -1) this.waiting.splice(index, 1); + + signal?.removeEventListener('abort', done); + resolve(remaining()); + }; + + if (timeout > 0) { + timer = setTimeout(done, timeout); + timer.unref(); + } + + signal?.addEventListener('abort', done, { once: true }); + this.waiting.push(done); + }); + } +} diff --git a/src/incoming-requests.ts b/src/incoming-requests.ts index 4bca232..202f3af 100644 --- a/src/incoming-requests.ts +++ b/src/incoming-requests.ts @@ -1,8 +1,11 @@ import type { DlrMerger } from './dlr-merger.ts'; import type { OnRequest } from './session-options.ts'; -import type { PduObject } from './pdu.ts'; +import type { PduObject, PduObjectInput } from './pdu.ts'; +import type { Result, VoidResult } from './result.ts'; import type { Session } from './session.ts'; import type { SmppLog } from './log.ts'; +import type { SmsIdFormat } from './sms-id.ts'; +import { HeldMessages } from './held-messages.ts'; import { Reassembler, decodeSegments } from './reassembly.ts'; import { bindCommands, defaults } from './session-options.ts'; import { concatInfo } from './udh.ts'; @@ -18,21 +21,32 @@ export type IncomingRequestsOptions = { maxReassembly?: number | undefined; onRequest?: OnRequest | undefined; reassemblyTimeout?: number | undefined; + /** Past a drain's refusal, for a receipt the drain is itself waiting for. */ + sendPastDrain: (input: PduObjectInput) => Promise>; session: Session; + smsIdFormat?: SmsIdFormat | undefined; systemId?: string | undefined; }; /** Everything the peer asks of a session: messages, receipts, links and the answers to them. */ export class IncomingRequests { private readonly dlrMerger: DlrMerger; + private readonly held: HeldMessages; private readonly log: SmppLog; private readonly onRequest: OnRequest | undefined; private readonly reassembler: Reassembler; + private readonly sendPastDrain: IncomingRequestsOptions['sendPastDrain']; private readonly session: Session; + private readonly smsIdFormat: SmsIdFormat; private readonly systemId: string; constructor(options: IncomingRequestsOptions) { this.dlrMerger = options.dlrMerger; + this.held = new HeldMessages({ + log: options.log, + max: defaults.maxHeldMessages, + timeout: defaults.heldMessageTimeout, + }); this.log = options.log; this.onRequest = options.onRequest; this.reassembler = new Reassembler({ @@ -41,7 +55,9 @@ export class IncomingRequests { maxOctets: options.maxOctets, timeout: options.reassemblyTimeout ?? defaults.reassemblyTimeout, }); + this.sendPastDrain = options.sendPastDrain; this.session = options.session; + this.smsIdFormat = options.smsIdFormat ?? {}; this.systemId = options.systemId ?? defaults.systemId; } @@ -78,11 +94,23 @@ export class IncomingRequests { } } - /** Drops the segments of every message that never became whole. */ + /** Drops the segments of every message that never became whole, and of every one still held. */ clear(): void { + this.held.clear(); this.reassembler.clear(); } + /** Waits out the messages the application still holds, and says how many it never answered. */ + async drain(timeout: number, signal: AbortSignal | undefined): Promise { + const unanswered = await this.held.idle(timeout, signal); + + if (unanswered === 0) return {}; + + this.log.warn('session - shutting down with messages unanswered', { timeout, unanswered }); + + return { err: new Error(`Shut down with ${String(unanswered)} message(s) unanswered`) }; + } + private async unhandled(pduObj: PduObject): Promise { if (bindCommands.includes(pduObj.cmdName)) { this.log.info('session - bind on an already bound session', { cmdName: pduObj.cmdName }); @@ -97,7 +125,7 @@ export class IncomingRequests { /** SMPP carries a mobile-originated message and a delivery receipt on the same command. */ private async onDeliverSm(pduObj: PduObject): Promise { - const dlr = dlrFromPdu(pduObj); + const dlr = dlrFromPdu(pduObj, this.smsIdFormat); if (!dlr) { this.onMessage(pduObj); @@ -135,12 +163,22 @@ export class IncomingRequests { if (!first) return; - this.session.emit('sms', createSms({ + const sms = createSms({ from: paramText(first.params.source_addr), message: decodeSegments(pduObjs), pduObjs, session: this.session, to: paramText(first.params.destination_addr), - })); + }, { + // A turn later, so a listener sending its receipt straight after the response still holds. + onAnswered: () => { setImmediate(() => { this.held.release(pduObjs); }); }, + // Past the refusal only while a drain is still waiting for this message; an ordinary send after. + send: input => (this.held.has(pduObjs) ? this.sendPastDrain(input) : this.session.send(input)), + }); + + this.held.hold(pduObjs); + + // A message nobody took is not work a shutdown can wait for. + if (!this.session.emit('sms', sms)) this.held.release(pduObjs); } } diff --git a/src/index.ts b/src/index.ts index e0f724f..3cf22e5 100644 --- a/src/index.ts +++ b/src/index.ts @@ -35,10 +35,11 @@ export { uuidv7 } from './uuid.ts'; export type { BindType, ClientOptions } from './client.ts'; export type { Dlr, Receipt } from './dlr.ts'; -export type { SendRespOptions, Sms, SmsInput } from './sms.ts'; +export type { SendDlrResult, SendRespOptions, Sms, SmsInput } from './sms.ts'; export type { ConcatInfo } from './udh.ts'; export type { Result, VoidResult } from './result.ts'; export type { SmppLog } from './log.ts'; +export type { SmsIdFormat, SmsIdNotation } from './sms-id.ts'; export type { AuthenticateInput, AuthenticateResult, diff --git a/src/link-gate.ts b/src/link-gate.ts new file mode 100644 index 0000000..6711790 --- /dev/null +++ b/src/link-gate.ts @@ -0,0 +1,128 @@ +import type { SmppLog } from './log.ts'; +import type { VoidResult } from './result.ts'; + +export type LinkGateOptions = { + log: SmppLog; + now?: (() => number) | undefined; + /** How long a request may wait for a link. 0 waits for as long as one may still arrive. */ + timeout: number; +}; + +type Waiter = (result: VoidResult) => void; + +function aborted(): Error { + return new Error('Aborted while waiting for a link'); +} + +function expired(): Error { + return new Error('The link did not come back in time'); +} + +function over(): Error { + return new Error('Session is closed'); +} + +/** Where a request with no link to go out on waits for the next one. */ +export class LinkGate { + private readonly log: SmppLog; + private readonly now: () => number; + private readonly timeout: number; + private readonly waiting = new Set(); + private returning = false; + private up = true; + + constructor(options: LinkGateOptions) { + this.log = options.log; + this.now = options.now ?? Date.now; + this.timeout = options.timeout; + } + + /** Whether a request can go out right now. A link that is attached but not yet bound cannot. */ + isUp(): boolean { + return this.up; + } + + /** Why the gate will never admit a request, or undefined while one may still get through. */ + refusal(): Error | undefined { + return this.up || this.returning ? undefined : over(); + } + + /** One budget for a request, however many links it waits through. 0 never gives up. */ + hold(signal: AbortSignal | undefined): () => Promise { + const deadline = this.timeout > 0 ? this.now() + this.timeout : 0; + + return () => this.wait(deadline, signal); + } + + /** A link is up and bound: everything held goes out on it. */ + open(): void { + this.up = true; + this.returning = false; + + if (this.waiting.size > 0) { + this.log.verbose('linkGate - sending what was held for a link', { held: this.waiting.size }); + } + + this.release({}); + } + + /** The link is gone. `returning` says whether another one is on its way. */ + shut(returning: boolean): void { + this.up = false; + this.returning = returning; + + if (!returning) this.release({ err: over() }); + } + + /** Resolves once a link can carry the request, or with the reason none ever will. */ + private wait(deadline: number, signal: AbortSignal | undefined): Promise { + if (this.up) return Promise.resolve({}); + + const refused = this.refusal(); + + if (refused) return Promise.resolve({ err: refused }); + + if (signal?.aborted === true) return Promise.resolve({ err: aborted() }); + + const left = deadline === 0 ? 0 : deadline - this.now(); + + if (deadline !== 0 && left <= 0) return Promise.resolve({ err: expired() }); + + return this.waitForLink(left, signal); + } + + private waitForLink(left: number, signal: AbortSignal | undefined): Promise { + this.log.verbose('linkGate - holding a request until a link is back', { timeout: left }); + + return new Promise(resolve => { + let timer: NodeJS.Timeout | undefined = undefined; + const settle = (result: VoidResult): void => { + if (timer) clearTimeout(timer); + + signal?.removeEventListener('abort', onAbort); + this.waiting.delete(settle); + resolve(result); + }; + const giveUp = (): void => { + this.log.warn('linkGate - no link came back in time', { timeout: left }); + settle({ err: expired() }); + }; + + function onAbort(): void { + settle({ err: aborted() }); + } + + // Not unref()'d: a held request is awaited with no other handle, so the process would exit unsettled. + if (left > 0) timer = setTimeout(giveUp, left); + + signal?.addEventListener('abort', onAbort, { once: true }); + this.waiting.add(settle); + }); + } + + private release(result: VoidResult): void { + for (const settle of [...this.waiting]) { + settle(result); + } + } +} diff --git a/src/outgoing-requests.ts b/src/outgoing-requests.ts new file mode 100644 index 0000000..e3b4090 --- /dev/null +++ b/src/outgoing-requests.ts @@ -0,0 +1,177 @@ +import type { PduObject, PduObjectInput } from './pdu.ts'; +import type { PduTransport } from './pdu-transport.ts'; +import type { Result, VoidResult } from './result.ts'; +import type { SendOptions } from './session-options.ts'; +import type { SmppLog } from './log.ts'; +import { LinkGate } from './link-gate.ts'; +import { PendingRequests } from './pending-requests.ts'; +import { SendWindow } from './send-window.ts'; +import { UnansweredError } from './unanswered-error.ts'; +import { bindCommands } from './session-options.ts'; +import { objToPdu } from './pdu.ts'; + +export type OutgoingRequestsOptions = { + log: SmppLog; + maxOutstanding: number; + responseTimeout: number; + transport: PduTransport; +}; + +/** `retryOnNextLink`: the write failed, so nothing reached the socket and another link may carry it. */ +type Attempt = { result: Result<{ pduObj: PduObject }>; retryOnNextLink: boolean }; + +function abortedBeforeSend(): Error { + return new Error('Aborted before the request was sent'); +} + +/** A response carries the request's sequence number, which only sendReturn() has. */ +function misuse(input: PduObjectInput): Error | undefined { + return input.cmdName.endsWith('_resp') + ? new Error(`Use sendReturn() for responses, not send(): ${input.cmdName}`) + : undefined; +} + +/** Everything this end asks of the peer: which link carries it, how many at once, and the answer. */ +export class OutgoingRequests { + private readonly gate: LinkGate; + private readonly log: SmppLog; + private readonly pending: PendingRequests; + private readonly responseTimeout: number; + private readonly transport: PduTransport; + private readonly window: SendWindow; + + private draining = false; + + constructor(options: OutgoingRequestsOptions) { + this.gate = new LinkGate({ log: options.log, timeout: options.responseTimeout }); + this.log = options.log; + this.pending = new PendingRequests(options.log); + this.responseTimeout = options.responseTimeout; + this.transport = options.transport; + this.window = new SendWindow(options.maxOutstanding); + } + + /** Read through a method: a drop can land while a request is awaiting. */ + linkDown(): boolean { + return !this.gate.isUp() || this.transport.sock.destroyed; + } + + /** A link is up and bound, so everything held for one goes out on it. */ + linkUp(): void { + this.gate.open(); + } + + /** The link is gone; `returning` says whether another one is on its way. */ + linkLost(returning: boolean): void { + this.gate.shut(returning); + this.pending.settleAll(new Error('Session closed before a response arrived')); + } + + /** Hands a response to the request waiting for it. False means nothing was. */ + deliver(pduObj: PduObject): boolean { + return this.pending.deliver(pduObj); + } + + /** Sends a request and resolves with the peer's response. */ + request(input: PduObjectInput, options: SendOptions): Promise> { + // Ahead of the drain, so a misuse is named as one rather than blamed on the shutdown. + const wrong = misuse(input); + + if (wrong) return Promise.resolve({ err: wrong }); + + // A drain on a live link. A link that is down is the gate's answer, which says closed instead. + if (this.draining && !this.linkDown()) { + return Promise.resolve({ err: new Error('Session is shutting down') }); + } + + return this.pastDrain(input, options); + } + + /** The same path without that refusal, which a receipt for a held message has to take. */ + async pastDrain( + input: PduObjectInput, + options: SendOptions, + ): Promise> { + const refused = this.refuse(input, options); + + if (refused) return { err: refused }; + + // A bind is what makes a link usable, so it cannot wait for one: it takes the gate's answer now. + if (bindCommands.includes(input.cmdName)) { + const shut = this.gate.refusal(); + + return shut ? { err: shut } : this.now(input, options); + } + + const waitForLink = this.gate.hold(options.signal); + + for (;;) { + const held = await waitForLink(); + + if (held.err) return { err: held.err }; + + await this.window.acquire(); + + const attempt = await this.attempt(input, options).finally(() => { this.window.release(); }); + + // Nothing reached the socket, so the next link carries it instead of the caller resending. + if (!attempt.retryOnNextLink || this.gate.isUp() || this.gate.refusal()) return attempt.result; + } + } + + /** Past the gate, the window and a drain, for what has to go out either way. */ + async now(input: PduObjectInput, options: SendOptions = {}): Promise> { + return (await this.attempt(input, options)).result; + } + + /** Refuses every request from here on, on a link that is already down as much as a live one. */ + stopAccepting(): void { + this.draining = true; + } + + /** Waits out the requests already on the wire, and says how many never finished. */ + async drain(timeout: number, signal: AbortSignal | undefined): Promise { + const unfinished = await this.window.idle(timeout, signal); + + if (unfinished === 0) return {}; + + this.log.warn('outgoingRequests - shutting down with requests unfinished', { timeout, unfinished }); + + return { err: new Error(`Shut down with ${String(unfinished)} request(s) unfinished`) }; + } + + /** Why a request cannot go out at all, as opposed to not yet. */ + private refuse(input: PduObjectInput, options: SendOptions): Error | undefined { + // Before the gate and the window, or an aborted call waits for what it will never use. + return misuse(input) ?? (options.signal?.aborted === true ? abortedBeforeSend() : undefined); + } + + private async attempt(input: PduObjectInput, options: SendOptions): Promise { + // pending.wait() alone settles the caller while the request still goes out to the peer. + if (options.signal?.aborted === true) { + return { result: { err: abortedBeforeSend() }, retryOnNextLink: false }; + } + + const seqNr = this.pending.nextSeqNr(); + const built = objToPdu({ ...input, seqNr }); + + if (built.err) return { result: { err: built.err }, retryOnNextLink: false }; + + const response = this.pending.wait(seqNr, { + signal: options.signal, + timeout: this.responseTimeout, + }); + const written = this.transport.write(built.buffer); + + if (written.err) { + this.pending.settle(seqNr, { err: written.err }); + + return { result: { err: written.err }, retryOnNextLink: true }; + } + + const answered = await response; + + // It went out, so a failure now means the peer may have taken it and the answer was the loss. + return { result: answered.err ? { err: new UnansweredError(answered.err) } : answered, retryOnNextLink: false }; + } +} diff --git a/src/pdu-transport.ts b/src/pdu-transport.ts new file mode 100644 index 0000000..b1f74c0 --- /dev/null +++ b/src/pdu-transport.ts @@ -0,0 +1,94 @@ +import type { PduObject } from './pdu.ts'; +import type { SmppLog } from './log.ts'; +import type { Socket } from 'node:net'; +import type { VoidResult } from './result.ts'; +import { PduFramer } from './pdu-framer.ts'; +import { pduToObj } from './pdu.ts'; + +export type PduTransportOptions = { + log: SmppLog; + onClose: () => void; + /** Raw bytes, before framing. */ + onData: (chunk: Buffer) => void; + onError: (err: Error) => void; + /** A complete PDU, before it is parsed. */ + onFramed: (pdu: Buffer) => void; + onPdu: (pduObj: PduObject) => void; + /** Nothing further can be read off this stream, whatever the socket does next. */ + onUnreadable: (err: Error) => void; +}; + +/** A socket read as a stream of complete PDUs. A reconnect attaches a new socket in its place. */ +export class PduTransport { + private readonly options: PduTransportOptions; + private framer = new PduFramer(); + private socket: Socket; + + constructor(options: PduTransportOptions, sock: Socket) { + this.options = options; + this.socket = sock; + this.wire(sock); + } + + get sock(): Socket { + return this.socket; + } + + /** Takes over a freshly opened socket. Half a PDU left on the old one must not prefix this one. */ + attach(sock: Socket): void { + // The socket being replaced is already dead, and its three handlers still point here. + this.socket.removeAllListeners(); + this.socket = sock; + this.framer = new PduFramer(); + this.wire(sock); + } + + private wire(sock: Socket): void { + sock.on('data', chunk => { this.read(chunk); }); + sock.on('close', () => { this.options.onClose(); }); + sock.on('error', err => { + this.options.log.warn('transport - socket error', { message: err.message }); + this.options.onError(err); + this.options.onClose(); + }); + } + + write(pdu: Buffer): VoidResult { + if (this.socket.destroyed) return { err: new Error('Socket is closed') }; + + this.socket.write(pdu); + + return {}; + } + + private read(chunk: Buffer): void { + this.options.onData(chunk); + this.framer.push(chunk); + + const framed = this.framer.next(); + + if (framed.err) { + this.options.log.warn('transport - unusable stream', { message: framed.err.message }); + this.options.onUnreadable(framed.err); + + return; + } + + for (const pdu of framed.pdus) { + this.options.onFramed(pdu); + + const parsed = pduToObj(pdu); + + if (parsed.err) { + this.options.log.warn('transport - could not parse an incoming PDU', { + message: parsed.err.message, + }); + this.options.onUnreadable(parsed.err); + + return; + } + + this.options.onPdu(parsed.pduObj); + } + } +} diff --git a/src/reconnect-loop.ts b/src/reconnect-loop.ts index 0c6ec13..efa5306 100644 --- a/src/reconnect-loop.ts +++ b/src/reconnect-loop.ts @@ -7,19 +7,23 @@ export type ReconnectLoopOptions = { log: SmppLog; maxDelay: number; minDelay: number; + now?: (() => number) | undefined; /** Brings the owner back up on a freshly opened socket. An err means try again. */ onConnected: (sock: Socket) => Promise; }; /** Reopens a dropped connection, backing off between attempts until it is told to stop. */ export class ReconnectLoop { + private readonly now: () => number; private readonly options: ReconnectLoopOptions; private attempting = false; private delay: number; private halted = false; private timer: NodeJS.Timeout | undefined; + private upAt: number | undefined; constructor(options: ReconnectLoopOptions) { + this.now = options.now ?? Date.now; this.options = options; this.delay = options.minDelay; } @@ -32,6 +36,13 @@ export class ReconnectLoop { schedule(): void { if (this.timer || this.attempting || this.isStopped()) return; + // Coming up is not proof: a stream we cannot read is only found once the link is bound. + if (this.upAt !== undefined && this.now() - this.upAt >= this.options.maxDelay) { + this.delay = this.options.minDelay; + } + + this.upAt = undefined; + const delay = this.delay; this.options.log.info('reconnect - retrying after a drop', { delay }); @@ -98,7 +109,7 @@ export class ReconnectLoop { return true; } - this.delay = this.options.minDelay; + this.upAt = this.now(); return false; } diff --git a/src/send-sms.ts b/src/send-sms.ts index b424f20..8b018b4 100644 --- a/src/send-sms.ts +++ b/src/send-sms.ts @@ -3,8 +3,11 @@ import type { ParamValue } from './defs/types.ts'; import type { PduObject, PduObjectInput } from './pdu.ts'; import type { Result } from './result.ts'; import type { SmppLog } from './log.ts'; +import type { SmsIdNotation } from './sms-id.ts'; +import { UnansweredError } from './unanswered-error.ts'; import { consts } from './defs/constants.ts'; import { detect } from './defs/encodings.ts'; +import { normaliseSmsId } from './sms-id.ts'; import { paramText } from './defs/types.ts'; import { maxSegments, smppTime, splitMessage } from './message.ts'; @@ -26,12 +29,24 @@ export type SendSmsOptions = { }; /** 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[] }; +export type SendSmsResult = { + err?: Error; + pduObjs: PduObject[]; + smsIds: string[]; + /** Segments that went out unanswered. The peer may have taken them, so sending again may duplicate. */ + unanswered: number; +}; + +/** A message that never went out, in the shape a caller aggregating segments still reads. */ +export function unsent(err: Error): SendSmsResult { + return { err, pduObjs: [], smsIds: [], unanswered: 0 }; +} /** What sending needs from the session: a concat reference and a way onto the wire. */ export type SendSmsDeps = { log: SmppLog; reference: number; + respIdNotation?: SmsIdNotation | undefined; send: (input: PduObjectInput) => Promise>; }; @@ -80,7 +95,6 @@ export function submitSmParams( return params; } -/** Puts a message on the wire as one submit_sm per segment. */ /** Nothing goes on the wire until the whole message fits: a half-sent message bills twice. */ function checkSegments(allowed: number, segments: number): Error | undefined { if (!Number.isInteger(allowed) || allowed < 1 || allowed > maxSegments) { @@ -98,17 +112,23 @@ function checkSegments(allowed: number, segments: number): Error | undefined { return undefined; } -function collectSent(sent: Result<{ pduObj: PduObject }>[]): SendSmsResult { +function collectSent( + sent: Result<{ pduObj: PduObject }>[], + notation: SmsIdNotation | undefined, +): SendSmsResult { const pduObjs: PduObject[] = []; const smsIds: string[] = []; let failure: Error | undefined; + let unanswered = 0; for (const one of sent) { if (one.err) { + if (one.err instanceof UnansweredError) unanswered++; + failure ??= one.err; } else if (one.pduObj.cmdStatus === 'ESME_ROK') { pduObjs.push(one.pduObj); - smsIds.push(paramText(one.pduObj.params.message_id)); + smsIds.push(normaliseSmsId(paramText(one.pduObj.params.message_id), notation)); } else { const refusal = one.pduObj.cmdStatus ?? String(one.pduObj.cmdStatusId); @@ -116,16 +136,17 @@ function collectSent(sent: Result<{ pduObj: PduObject }>[]): SendSmsResult { } } - return failure ? { err: failure, pduObjs, smsIds } : { pduObjs, smsIds }; + return failure ? { err: failure, pduObjs, smsIds, unanswered } : { pduObjs, smsIds, unanswered }; } +/** Puts a message on the wire as one submit_sm per segment. */ export async function submitSms(deps: SendSmsDeps, sms: SendSmsOptions): Promise { const allowed = sms.maxSegments ?? maxSegments; const encoding = sms.encoding ?? detect(sms.message); const segments = splitMessage(sms.message, { encoding, reference: deps.reference }); const refused = checkSegments(allowed, segments.length); - if (refused) return { err: refused, pduObjs: [], smsIds: [] }; + if (refused) return unsent(refused); const multipart = segments.length > 1; @@ -138,5 +159,5 @@ export async function submitSms(deps: SendSmsDeps, sms: SendSmsOptions): Promise params: submitSmParams(sms, segment, { encoding, multipart }), }))); - return collectSent(sent); + return collectSent(sent, deps.respIdNotation); } diff --git a/src/send-window.ts b/src/send-window.ts index e1492ca..524df8d 100644 --- a/src/send-window.ts +++ b/src/send-window.ts @@ -1,8 +1,10 @@ +import { IdleWaiters } from './idle-waiters.ts'; + /** Caps how many requests are on the wire at once; anything past the limit waits its turn. */ export class SendWindow { + private readonly idleWaiters = new IdleWaiters(); private readonly limit: number; private readonly waiting: (() => void)[] = []; - private readonly waitingForIdle: (() => void)[] = []; private inFlight = 0; constructor(limit: number) { @@ -32,9 +34,7 @@ export class SendWindow { if (this.inFlight > 0) return; - for (const resolve of this.waitingForIdle.splice(0)) { - resolve(); - } + this.idleWaiters.settle(); } /** Everything the caller is still owed: on the wire, plus queued behind a full window. */ @@ -42,34 +42,8 @@ export class SendWindow { return this.inFlight + this.waiting.length; } - /** - * Resolves 0 once nothing is left, or with what still is when the timeout or the signal cuts the - * wait short. A timeout of 0 waits forever. - */ - idle(timeout: number, signal?: AbortSignal): Promise { - if (this.inFlight === 0) return Promise.resolve(0); - - if (signal?.aborted === true) return Promise.resolve(this.unfinished()); - - return new Promise(resolve => { - let timer: NodeJS.Timeout | undefined = undefined; - const done = (): void => { - const index = this.waitingForIdle.indexOf(done); - - if (timer) clearTimeout(timer); - if (index !== -1) this.waitingForIdle.splice(index, 1); - - signal?.removeEventListener('abort', done); - resolve(this.unfinished()); - }; - - if (timeout > 0) { - timer = setTimeout(done, timeout); - timer.unref(); - } - - signal?.addEventListener('abort', done, { once: true }); - this.waitingForIdle.push(done); - }); + /** Resolves 0 once nothing is left on the wire, or with what still is. */ + idle(timeout: number, signal: AbortSignal | undefined): Promise { + return this.idleWaiters.wait(() => this.unfinished(), timeout, signal); } } diff --git a/src/session-options.ts b/src/session-options.ts index 8347353..514541b 100644 --- a/src/session-options.ts +++ b/src/session-options.ts @@ -4,12 +4,15 @@ import type { PduObject } from './pdu.ts'; import type { Result, VoidResult } from './result.ts'; import type { Session } from './session.ts'; import type { SmppLog } from './log.ts'; +import type { SmsIdFormat } from './sms-id.ts'; import type { Sms } from './sms.ts'; import type { Socket } from 'node:net'; +import { isSmsIdNotation, smsIdNotations, smsIdPlaces } from './sms-id.ts'; export type SessionEvents = { close: []; data: [Buffer]; + disconnected: []; dlr: [Dlr, PduObject]; incomingPdu: [Buffer]; incomingPduObj: [PduObject]; @@ -83,6 +86,8 @@ export type SessionOptions = { responseTimeout?: number | undefined; /** How long a drain waits for the requests already on the wire. 0 waits forever. */ shutdownTimeout?: number | undefined; + /** The notation the peer writes message ids in, where it is not the one they are compared in. */ + smsIdFormat?: SmsIdFormat | undefined; sock: Socket; /** This end's own identity, answered to the peer in place of the one it sent. */ systemId?: string | undefined; @@ -96,8 +101,11 @@ export const undeclaredInterfaceVersion = 0x00; export const defaults = { /** Receipts of a multipart message can be a working day apart, so the cap does the bounding. */ dlrMergeTimeout: 86_400_000, + /** The peer gave up on an unanswered message long before this; the bound is against growth. */ + heldMessageTimeout: 300_000, maxDelay: 30_000, maxDlrMerges: 1000, + maxHeldMessages: 1000, maxOutstanding: 10, maxReassembly: 1000, minDelay: 1000, @@ -111,16 +119,24 @@ export const defaults = { * A count below 1 does not fail loudly anywhere downstream: `maxOutstanding: 0` leaves every send * queued behind a slot that is never freed, so the call never settles at all. */ -export function checkSessionOptions(options: SessionCounts): VoidResult { - const limits: [string, number, number][] = [ +export function checkSessionOptions(options: CheckableOptions): VoidResult { + const checked = checkLimits([ ['idleTimeout', options.idleTimeout ?? 0, 0], ['maxOutstanding', options.maxOutstanding ?? defaults.maxOutstanding, 1], ['maxReassembly', options.maxReassembly ?? defaults.maxReassembly, 1], ['reassemblyTimeout', options.reassemblyTimeout ?? defaults.reassemblyTimeout, 0], ['responseTimeout', options.responseTimeout ?? defaults.responseTimeout, 0], ['shutdownTimeout', options.shutdownTimeout ?? defaults.shutdownTimeout, 0], - ]; + ]); + if (checked.err) return checked; + + const backoff = checkReconnect(options.reconnect); + + return backoff.err ? backoff : checkSmsIdFormat(options.smsIdFormat); +} + +function checkLimits(limits: [string, number, number][]): VoidResult { for (const [name, value, min] of limits) { if (!Number.isInteger(value) || value < min) { return { err: new Error(`${name} must be ${String(min)} or more, got ${String(value)}`) }; @@ -130,11 +146,77 @@ export function checkSessionOptions(options: SessionCounts): VoidResult { return {}; } -export type SessionCounts = { +const backoffDelays: readonly string[] = ['maxDelay', 'minDelay']; + +function checkReconnect(reconnect: unknown): VoidResult { + if (reconnect === undefined || reconnect === false) return {}; + + if (!isRecord(reconnect)) { + return { err: new Error('reconnect takes { maxDelay, minDelay }, or false to turn it off') }; + } + + for (const key of Object.keys(reconnect)) { + if (!backoffDelays.includes(key)) { + return { err: new Error(`reconnect has no ${key}, name ${backoffDelays.join(' or ')}`) }; + } + } + + const maxDelay = delayOr(reconnect.maxDelay, defaults.maxDelay); + const minDelay = delayOr(reconnect.minDelay, defaults.minDelay); + // A delay of 0 never doubles, so the backoff never starts and every retry lands at once. + const checked = checkLimits([['maxDelay', maxDelay, 1], ['minDelay', minDelay, 1]]); + + if (checked.err) return checked; + + if (maxDelay < minDelay) { + return { err: new Error(`maxDelay must be minDelay (${String(minDelay)}) or more, got ${String(maxDelay)}`) }; + } + + return {}; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +/** A tuning value that is not a number lands on NaN, which the range check refuses by name. */ +function delayOr(value: unknown, fallback: number): number { + if (value === undefined) return fallback; + + return typeof value === 'number' ? value : NaN; +} + +function checkSmsIdFormat(smsIdFormat: unknown): VoidResult { + if (smsIdFormat === undefined) return {}; + + if (!isRecord(smsIdFormat)) { + return { err: new Error('smsIdFormat names a notation per place, as { receipt, submitResp }') }; + } + + for (const [place, notation] of Object.entries(smsIdFormat)) { + if (!smsIdPlaces.includes(place)) { + return { err: new Error(`smsIdFormat has no ${place}, name ${smsIdPlaces.join(' or ')}`) }; + } + + if (notation === undefined || isSmsIdNotation(notation)) continue; + + // String() throws on a null-prototype object, and this value is whatever the caller passed. + const got = typeof notation === 'string' ? notation : typeof notation; + + return { err: new Error(`smsIdFormat.${place} must be ${smsIdNotations.join(' or ')}, got ${got}`) }; + } + + return {}; +} + +/** What the checker reads, as it arrives: a caller without types can put anything in it. */ +export type CheckableOptions = { idleTimeout?: number | undefined; maxOutstanding?: number | undefined; maxReassembly?: number | undefined; reassemblyTimeout?: number | undefined; + reconnect?: unknown; responseTimeout?: number | undefined; shutdownTimeout?: number | undefined; + smsIdFormat?: unknown; }; diff --git a/src/session.ts b/src/session.ts index f5ddffa..34868e9 100644 --- a/src/session.ts +++ b/src/session.ts @@ -11,16 +11,17 @@ import { DlrMerger } from './dlr-merger.ts'; import { EventEmitter } from 'node:events'; import { IncomingRequests } from './incoming-requests.ts'; import { LinkTimers } from './link-timers.ts'; -import { PduFramer } from './pdu-framer.ts'; -import { PendingRequests } from './pending-requests.ts'; +import { OutgoingRequests } from './outgoing-requests.ts'; +import { PduTransport } from './pdu-transport.ts'; import { ReconnectLoop } from './reconnect-loop.ts'; -import { SendWindow } from './send-window.ts'; +import { leftOf } from './idle-waiters.ts'; import { errorFrom } from './error-from.ts'; import { optionalParamsMinVersion } from './defs/constants.ts'; import { bindCarries, bindCommands, defaultSystemId, defaults } from './session-options.ts'; -import { isResp, objToPdu, pduReturn, pduToObj } from './pdu.ts'; +import { isResp, pduReturn } from './pdu.ts'; import { silentLog } from './log.ts'; -import { submitSms } from './send-sms.ts'; +import { submitSms, unsent } from './send-sms.ts'; +import { ConcatReference } from './udh.ts'; export type { CloseOptions, @@ -47,8 +48,6 @@ export class Session extends EventEmitter { declare prependOnceListener: (event: K, listener: SessionListener) => this; declare removeListener: (event: K, listener: SessionListener) => this; - /** Replaced on reconnect, so hold the session rather than this. */ - sock: Socket; readonly log: SmppLog; /** The role the ESME bound with, whichever end of the link this is. Undefined before any bind. */ @@ -58,18 +57,17 @@ export class Session extends EventEmitter { peerInterfaceVersion: number | undefined = undefined; userData: unknown = undefined; + private readonly concatReference = new ConcatReference(); private readonly dlrMerger: DlrMerger; private readonly incoming: IncomingRequests; private readonly options: SessionOptions; - private readonly pending: PendingRequests; + private readonly outgoing: OutgoingRequests; private readonly reconnectLoop: ReconnectLoop | undefined; private readonly timers: LinkTimers; - private readonly window: SendWindow; + private readonly transport: PduTransport; private closed = false; - private concatReference = 0; - private draining = false; - private framer = new PduFramer(); + private ended = false; /** A listener that throws is the application's bug; it must not become ours. Hard rule 1. */ override emit( @@ -120,12 +118,12 @@ export class Session extends EventEmitter { maxReassembly: options.maxReassembly, onRequest: options.onRequest, reassemblyTimeout: options.reassemblyTimeout, + sendPastDrain: input => this.outgoing.pastDrain(input, {}), session: this, + smsIdFormat: options.smsIdFormat, systemId: options.systemId, }); - this.pending = new PendingRequests(this.log); this.reconnectLoop = this.loopFor(options.reconnect); - this.sock = options.sock; this.timers = new LinkTimers({ enquireLinkInterval: options.enquireLinkInterval, idleTimeout: options.idleTimeout, @@ -134,12 +132,22 @@ export class Session extends EventEmitter { // 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); + this.transport = this.transportFor(options.sock); + this.outgoing = new OutgoingRequests({ + log: this.log, + maxOutstanding: options.maxOutstanding ?? defaults.maxOutstanding, + responseTimeout: options.responseTimeout ?? defaults.responseTimeout, + transport: this.transport, + }); - this.attach(options.sock); this.resetTimers(); } + /** Replaced on reconnect, so hold the session rather than this. */ + get sock(): Socket { + return this.transport.sock; + } + /** Whether this session's bind direction carries a command. Consulted by the library's senders. */ bindAllows(cmdName: string): boolean { return bindCarries(this.boundAs, cmdName); @@ -152,30 +160,8 @@ export class Session extends EventEmitter { } /** Sends a request and resolves with the peer's response. */ - async send( - input: PduObjectInput, - options: SendOptions = {}, - ): Promise> { - if (input.cmdName.endsWith('_resp')) { - return { err: new Error(`Use sendReturn() for responses, not send(): ${input.cmdName}`) }; - } - - if (this.closed) return { err: new Error('Session is closed') }; - - if (this.draining) return { err: new Error('Session is shutting down') }; - - // Before the window, or a full window makes an aborted call wait for a slot it will not use. - if (options.signal?.aborted === true) { - return { err: new Error('Aborted before the request was sent') }; - } - - await this.window.acquire(); - - try { - return await this.request(input, options); - } finally { - this.window.release(); - } + send(input: PduObjectInput, options: SendOptions = {}): Promise> { + return this.outgoing.request(input, options); } /** Answers a request the peer sent us. Responses are never waited on. */ @@ -186,7 +172,7 @@ 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); + const sent = built.err ? { err: built.err } : this.transport.write(built.buffer); // A peer that unbinds and drops the link takes our response with it; that is not a failure. if (sent.err && !this.closed) { @@ -203,12 +189,13 @@ export class Session extends EventEmitter { async sendSms(sms: SendSmsOptions, options: SendOptions = {}): Promise { if (!this.bindAllows('submit_sm')) { - return { err: new Error('A receiver-bound session does not carry submit_sm'), pduObjs: [], smsIds: [] }; + return unsent(new Error('A receiver-bound session does not carry submit_sm')); } const sent = await submitSms({ log: this.log, - reference: this.nextConcatReference(), + reference: this.concatReference.next(), + respIdNotation: this.options.smsIdFormat?.submitResp, send: input => this.send(input, options), }, sms); @@ -224,9 +211,9 @@ export class Session extends EventEmitter { async unbind(): Promise { const drained = await this.drain(undefined); const wasOpen = !this.closed; - // request(), not send(): the drain gate refuses a send, and the unbind goes out either way. + // now(), not send(): a drain refuses a send, and the unbind goes out either way. const sent = wasOpen - ? await this.request({ cmdName: 'unbind' }, {}) + ? await this.outgoing.now({ cmdName: 'unbind' }) : { err: new Error('Session is closed') }; const closedOnUnbind = wasOpen && this.closed; @@ -247,6 +234,21 @@ export class Session extends EventEmitter { return drained; } + private transportFor(sock: Socket): PduTransport { + return new PduTransport({ + log: this.log, + onClose: () => { this.onClose(); }, + onData: chunk => { this.onData(chunk); }, + onError: err => { this.emit('sessionError', err); }, + onFramed: pdu => { this.emit('incomingPdu', pdu); }, + onPdu: pduObj => { this.dispatch(pduObj); }, + onUnreadable: err => { + this.emit('sessionError', err); + this.teardown(); + }, + }, sock); + } + private loopFor(reconnect: ReconnectOptions | undefined): ReconnectLoop | undefined { if (!reconnect) return undefined; @@ -281,82 +283,50 @@ export class Session extends EventEmitter { } this.resetTimers(); + this.outgoing.linkUp(); this.log.info('session - reconnected'); this.emit('reconnected'); return {}; } - /** 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.transport.attach(sock); this.closed = false; - - sock.on('data', chunk => { this.onData(chunk); }); - sock.on('close', () => { this.onClose(); }); - sock.on('error', err => { - this.log.warn('session - socket error', { message: err.message }); - this.emit('sessionError', err); - this.onClose(); - }); } - private async request( - input: PduObjectInput, - options: SendOptions, - ): Promise> { - // pending.wait() alone settles the caller while the request still goes out to the peer. - if (options.signal?.aborted === true) { - return { err: new Error('Aborted before the request was sent') }; - } - - const seqNr = this.pending.nextSeqNr(); - const built = objToPdu({ ...input, seqNr }); - - if (built.err) return { err: built.err }; - - const response = this.pending.wait(seqNr, { - signal: options.signal, - timeout: this.options.responseTimeout ?? defaults.responseTimeout, - }); - const written = this.write(built.buffer); - - if (written.err) { - this.pending.settle(seqNr, { err: written.err }); - - return { err: written.err }; - } - - return response; - } - - /** Stops new sends and waits out the ones already issued. */ + /** Stops new sends and waits out the messages we hold and the requests already issued. */ private async drain(signal: AbortSignal | undefined): Promise { this.reconnectLoop?.stop(); - this.draining = true; + this.outgoing.stopAccepting(); - if (this.closed || this.sock.destroyed) return {}; + if (this.outgoing.linkDown()) return {}; const timeout = this.options.shutdownTimeout ?? defaults.shutdownTimeout; - const unfinished = await this.window.idle(timeout, signal); + const deadline = timeout > 0 ? Date.now() + timeout : 0; + // Answering a message can put a receipt on the wire; nothing on the wire produces a message. + const messages = await this.incoming.drain(this.answering(timeout), signal); + const requests = await this.outgoing.drain(leftOf(deadline), signal); // The window empties on a teardown too, which settles everything the link was carrying. - if (this.isClosed()) return { err: new Error('The session closed before the drain finished') }; + if (this.outgoing.linkDown()) { + return { err: new Error('The session closed before the drain finished') }; + } - if (unfinished === 0) return {}; + if (!messages.err) return requests; - this.log.warn('session - shutting down with requests unfinished', { timeout, unfinished }); + if (!requests.err) return messages; - return { err: new Error(`Shut down with ${String(unfinished)} request(s) unfinished`) }; + return { err: new Error(`${messages.err.message}; ${requests.err.message}`) }; } - /** Read through a method: teardown() can land while the drain is awaiting. */ - private isClosed(): boolean { - return this.closed; + /** The application half's budget, which may never be "forever": nothing else ends that wait. */ + private answering(timeout: number): number { + if (timeout > 0) return timeout; + + const responseTimeout = this.options.responseTimeout ?? defaults.responseTimeout; + + return responseTimeout > 0 ? responseTimeout : defaults.responseTimeout; } /** The session is over now, drained or not. Nothing brings it back. */ @@ -364,79 +334,42 @@ export class Session extends EventEmitter { this.reconnectLoop?.stop(); this.teardown(); this.dlrMerger.clear(); + this.emitClose(); + } + + private emitClose(): void { + if (this.ended) return; + + this.ended = true; + this.outgoing.linkLost(false); + this.emit('close'); } private teardown(): void { if (this.closed) return; this.closed = true; + this.outgoing.linkLost(this.retrying()); this.timers.clear(); - this.pending.settleAll(new Error('Session closed before a response arrived')); this.incoming.clear(); this.sock.destroy(); - this.emit('close'); + + if (this.retrying()) this.emit('disconnected'); + else this.emitClose(); } - private nextConcatReference(): number { - this.concatReference = this.concatReference >= 255 ? 1 : this.concatReference + 1; - - return this.concatReference; - } - - private write(pdu: Buffer): VoidResult { - if (this.sock.destroyed) { - return { err: new Error('Socket is closed') }; - } - - this.sock.write(pdu); - - return {}; + private retrying(): boolean { + return this.reconnectLoop !== undefined && !this.reconnectLoop.isStopped(); } private onData(chunk: Buffer): void { this.emit('data', chunk); this.resetTimers(); - this.framer.push(chunk); - - const framed = this.framer.next(); - - if (framed.err) { - this.log.warn('session - unusable stream, closing', { message: framed.err.message }); - this.emit('sessionError', framed.err); - this.end(); - - return; - } - - for (const pdu of framed.pdus) { - if (!this.receive(pdu)) return; - } - } - - /** False means the PDU could not be read and the session has been closed. */ - private receive(pdu: Buffer): boolean { - this.emit('incomingPdu', pdu); - - const parsed = pduToObj(pdu); - - if (parsed.err) { - this.log.warn('session - could not parse an incoming PDU, closing', { - message: parsed.err.message, - }); - this.emit('sessionError', parsed.err); - this.end(); - - return false; - } - - this.dispatch(parsed.pduObj); - - return true; } private dispatch(pduObj: PduObject): void { if (isResp(pduObj)) { - if (!this.pending.deliver(pduObj)) { + if (!this.outgoing.deliver(pduObj)) { this.log.debug('session - response with no matching request', { seqNr: pduObj.seqNr }); } diff --git a/src/sms-id.ts b/src/sms-id.ts new file mode 100644 index 0000000..27146ff --- /dev/null +++ b/src/sms-id.ts @@ -0,0 +1,35 @@ +const notations = { + decimal: { digits: /^[0-9]+$/, prefix: '' }, + hex: { digits: /^[0-9a-f]+$/i, prefix: '0x' }, +}; + +const places = ['receipt', 'submitResp'] as const; + +/** The notation a peer writes message ids in. */ +export type SmsIdNotation = keyof typeof notations; + +/** The notation per place the peer writes an id. An omitted place is left as it arrived. */ +export type SmsIdFormat = Partial>; + +export const smsIdNotations: readonly string[] = Object.keys(notations); + +export const smsIdPlaces: readonly string[] = places; + +export function isSmsIdNotation(value: unknown): value is SmsIdNotation { + return typeof value === 'string' && Object.hasOwn(notations, value); +} + +// SMPP 3.4 caps message_id at 64 octets, and BigInt on a longer string is a peer-controlled cost. +const maxIdLength = 64; + +/** + * The id as a plain decimal value, so an SMSC that answers a submit in one notation and writes the + * receipt in another still correlates. + */ +export function normaliseSmsId(id: string, notation: SmsIdNotation | undefined): string { + if (!isSmsIdNotation(notation) || id.length > maxIdLength) return id; + + const { digits, prefix } = notations[notation]; + + return digits.test(id) ? BigInt(`${prefix}${id}`).toString(10) : id; +} diff --git a/src/sms.ts b/src/sms.ts index cefdcf9..a93cf6f 100644 --- a/src/sms.ts +++ b/src/sms.ts @@ -1,13 +1,22 @@ import type { ErrorName } from './defs/errors.ts'; import type { MessageState } from './defs/constants.ts'; -import type { PduObject, TlvInput } from './pdu.ts'; +import type { PduObject, PduObjectInput, TlvInput } from './pdu.ts'; import type { Result, VoidResult } from './result.ts'; import type { Session } from './session.ts'; +import { UnansweredError } from './unanswered-error.ts'; import { consts } from './defs/constants.ts'; import { receiptCodes } from './dlr.ts'; import { smppDate } from './message.ts'; import { uuidv7 } from './uuid.ts'; +/** `pduObjs` holds what the peer took, so a partial failure names what is already receipted. */ +export type SendDlrResult = { + err?: Error; + pduObjs: PduObject[]; + /** Segments that went out unanswered. The peer may have taken them, so sending again may duplicate. */ + unanswered: number; +}; + export type SendRespOptions = { /** The id the peer correlates a later delivery receipt by. Defaults to a generated UUID v7. */ smsId?: string; @@ -25,7 +34,7 @@ export type Sms = { message: string; pduObjs: PduObject[]; /** Sends a delivery report back to the sender. Defaults to DELIVERED. */ - sendDlr: (status?: MessageState) => Promise>; + sendDlr: (status?: MessageState) => Promise; /** Answers every segment. Part of the protocol, not optional. Defaults to ESME_ROK. */ sendResp: (options?: SendRespOptions) => Promise; session: Session; @@ -43,12 +52,18 @@ export type SmsInput = { to: string; }; +/** What the session's incoming side gives a message so it can be answered and accounted for. */ +export type SmsHandlers = { + onAnswered: () => void; + send: (input: PduObjectInput) => Promise>; +}; + /** 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): Sms { +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; @@ -60,8 +75,8 @@ export function createSms(input: SmsInput): Sms { from: input.from, message: input.message, pduObjs: input.pduObjs, - sendDlr: status => sendDlr(sms, status), - sendResp: options => sendResp(sms, answered, options ?? {}), + sendDlr: status => sendDlr(sms, handlers.send, status), + sendResp: options => sendResp(sms, answered, options ?? {}).finally(handlers.onAnswered), session: input.session, get smsId(): string { return answered.smsId; @@ -122,20 +137,47 @@ function receiptTlvs(smsId: string, status: MessageState): Record[]): SendDlrResult { + const pduObjs: PduObject[] = []; + let failure: Error | undefined; + let unanswered = 0; + + for (const one of sent) { + if (one.err) { + if (one.err instanceof UnansweredError) unanswered++; + + failure ??= one.err; + } else if (one.pduObj.cmdStatus === 'ESME_ROK') { + pduObjs.push(one.pduObj); + } else { + const refusal = one.pduObj.cmdStatus ?? String(one.pduObj.cmdStatusId); + + failure ??= new Error(`deliver_sm refused by the peer: ${refusal}`); + } + } + + return failure ? { err: failure, pduObjs, unanswered } : { pduObjs, unanswered }; +} + async function sendDlr( sms: Sms, + send: SmsHandlers['send'], status: MessageState = 'DELIVERED', -): Promise> { +): Promise { if (!sms.session.bindAllows('deliver_sm')) { - return { err: new Error('A transmitter-bound session does not carry deliver_sm') }; + return { + err: new Error('A transmitter-bound session does not carry deliver_sm'), + pduObjs: [], + unanswered: 0, + }; } const total = sms.pduObjs.length; - const pduObjs: PduObject[] = []; - - for (let index = 0; index < total; index++) { + // Together, not one after a response: a drain waiting for this message must see the whole receipt. + const sent = await Promise.all(sms.pduObjs.map((_segment, index) => { const smsId = segmentId(sms.smsId, index, total); - const sent = await sms.session.send({ + + return send({ cmdName: 'deliver_sm', params: { destination_addr: sms.from, @@ -145,11 +187,6 @@ async function sendDlr( }, ...(sms.session.acceptsOptionalParams() ? { tlvs: receiptTlvs(smsId, status) } : {}), }); - - if (sent.err) return { err: sent.err }; - - pduObjs.push(sent.pduObj); - } - - return { pduObjs }; + })); + return collectReceipt(sent); } diff --git a/src/udh.ts b/src/udh.ts index 37fba2d..1b6c399 100644 --- a/src/udh.ts +++ b/src/udh.ts @@ -1,3 +1,14 @@ +/** The 8-bit reference tying a long SMS's segments together, counted per session. */ +export class ConcatReference { + private current = 0; + + next(): number { + this.current = this.current >= 255 ? 1 : this.current + 1; + + return this.current; + } +} + export type ConcatInfo = { part: number; reference: number; diff --git a/src/unanswered-error.ts b/src/unanswered-error.ts new file mode 100644 index 0000000..b6f45b8 --- /dev/null +++ b/src/unanswered-error.ts @@ -0,0 +1,7 @@ +/** The request went out and no answer came back: the peer may have accepted it. */ +export class UnansweredError extends Error { + constructor(cause: Error) { + super(`No answer came back, so the peer may have accepted it: ${cause.message}`, { cause }); + this.name = 'UnansweredError'; + } +} diff --git a/test/dlr.test.ts b/test/dlr.test.ts index d1969b0..2dc499c 100644 --- a/test/dlr.test.ts +++ b/test/dlr.test.ts @@ -206,6 +206,40 @@ describe('dlrFromPdu()', () => { assert.equal(dlr.receipt.sub, 1); assert.equal(dlr.receipt.text, 'hello'); }); + + test('reads the id in the notation the peer writes receipts in', () => { + const hex = dlrFromPdu(deliverSm('id:1a2B stat:DELIVRD err:000 text:'), { receipt: 'hex' }); + + assert.ok(hex); + assert.equal(hex.smsId, '6699'); + assert.equal(hex.receipt?.id, '1a2B', 'the receipt itself keeps the id as it arrived'); + + assert.equal(dlrFromPdu(deliverSm('id:0000123 stat:DELIVRD'), { receipt: 'decimal' })?.smsId, '123'); + }); + + // SMPP 3.4 5.3.2.26 makes the TLV the id the submit_sm_resp carried, not the body's rendering. + test('reads the receipted_message_id TLV in the notation the peer answers a submit in', () => { + const marked = deliverSm('nothing scrapable here', { + receipted_message_id: { tagValue: 'FF' }, + }, 0); + + assert.equal(dlrFromPdu(marked, { submitResp: 'hex' })?.smsId, '255'); + assert.equal(dlrFromPdu(marked, { receipt: 'hex' })?.smsId, 'FF'); + }); + + test('leaves an id the notation cannot read as it arrived', () => { + assert.equal(dlrFromPdu(deliverSm('id:beef-1 stat:DELIVRD'), { receipt: 'hex' })?.smsId, 'beef-1'); + assert.equal(dlrFromPdu(deliverSm('id:1a2b stat:DELIVRD'), { receipt: 'decimal' })?.smsId, '1a2b'); + assert.equal(dlrFromPdu(deliverSm('id:0195f0c7 stat:DELIVRD'))?.smsId, '0195f0c7'); + }); + + // Number() reads 9007199254740993 as ...92, which correlates a receipt to the wrong send. + test('reads an id past the safe integer range without losing a digit', () => { + assert.equal( + dlrFromPdu(deliverSm('id:9007199254740993 stat:DELIVRD'), { receipt: 'decimal' })?.smsId, + '9007199254740993', + ); + }); }); describe('receiptCodes', () => { diff --git a/test/readme.test.ts b/test/readme.test.ts index 23b9b7b..d68ce28 100644 --- a/test/readme.test.ts +++ b/test/readme.test.ts @@ -83,7 +83,7 @@ describe('README: Client', () => { closeAfter(t, session); const reported = once(resolve => { session.on('dlr', resolve); }); - const { err: sendErr, smsIds } = await session.sendSms({ + const { err: sendErr, smsIds, unanswered } = await session.sendSms({ dlr: true, from: '46701113311', message: '«baff»', @@ -92,6 +92,29 @@ describe('README: Client', () => { assert.equal(sendErr, undefined); assert.equal(smsIds.length, 1); + assert.equal(unanswered, 0); + assert.equal((await reported).smsId, smsIds[0]); + }); + + test('naming the notation the SMSC writes message ids in', async t => { + await answeringServer(t); + + const { err, session } = await client({ smsIdFormat: { receipt: 'decimal', submitResp: 'hex' } }); + if (err) throw err; + + closeAfter(t, session); + + const reported = once(resolve => { session.on('dlr', resolve); }); + const { err: sendErr, smsIds } = await session.sendSms({ + dlr: true, + from: '46701113311', + message: 'Hello world', + to: '46709771337', + }); + + assert.equal(sendErr, undefined); + assert.equal(smsIds.length, 1); + // The generated ids the server answers with read as no notation, so they arrive untouched. assert.equal((await reported).smsId, smsIds[0]); }); diff --git a/test/session-extras.test.ts b/test/session-extras.test.ts index 62ec25a..91c41be 100644 --- a/test/session-extras.test.ts +++ b/test/session-extras.test.ts @@ -12,14 +12,20 @@ import type { SmppLog } from '../src/log.ts'; import type { Sms } from '../src/sms.ts'; import type { SmppServer } from '../src/server.ts'; import type { TestContext } from 'node:test'; +import { HeldMessages } from '../src/held-messages.ts'; +import { UnansweredError } from '../src/unanswered-error.ts'; +import { createSms } from '../src/sms.ts'; +import { LinkGate } from '../src/link-gate.ts'; import { Reassembler, decodeSegments } from '../src/reassembly.ts'; import { Session } from '../src/session.ts'; import { DlrMerger } from '../src/dlr-merger.ts'; +import { checkSessionOptions } from '../src/session-options.ts'; import { client } from '../src/client.ts'; import { closeAfter, closeListenerAfter } from './teardown.ts'; import { consts } from '../src/defs/constants.ts'; import { errors } from '../src/defs/errors.ts'; import { objToPdu } from '../src/pdu.ts'; +import { paramText } from '../src/defs/types.ts'; import { server } from '../src/server.ts'; import { silentLog } from '../src/log.ts'; import { submitSms } from '../src/send-sms.ts'; @@ -67,10 +73,54 @@ function delay(ms: number): Promise { return new Promise(resolve => { setTimeout(resolve, ms); }); } -type Gate = { open: () => void; passed: Promise }; +function submitPdu(seqNr: number, cmdStatus: ErrorName = 'ESME_ROK'): PduObject { + return { + cmdId: 0x00000004, + cmdLength: 0, + cmdName: 'submit_sm', + cmdStatus, + cmdStatusId: 0, + params: { destination_addr: '46709771337', short_message: 'held', source_addr: '46701113311' }, + seqNr, + tlvs: {}, + }; +} + +/** A cmd_length below the 16-octet header: a stream no framing can recover from. */ +const unreadablePdu = Buffer.from([0, 0, 0, 4, 0, 0, 0, 4, 0, 0, 0, 0, 0, 0, 0, 1]); + +/** The server's side of the one connection under test. */ +function peerOf(smpp: SmppServer): Session { + const [peer] = smpp.sessions; + + assert.equal(smpp.sessions.size, 1); + assert.ok(peer); + + return peer; +} + +async function sendReceipt(peer: Session, smsId: string, tlvSmsId = smsId): Promise { + const sent = await peer.send({ + cmdName: 'deliver_sm', + params: { + destination_addr: '46701113311', + esm_class: consts.ESM_CLASS.MC_DELIVERY_RECEIPT, + short_message: `id:${smsId} stat:DELIVRD err:000 text:`, + source_addr: '46709771337', + }, + tlvs: { + message_state: { tagValue: consts.MESSAGE_STATE.DELIVERED }, + receipted_message_id: { tagValue: tlvSmsId }, + }, + }); + + assert.equal(sent.err, undefined); +} + +type Latch = { open: () => void; passed: Promise }; /** A promise the test opens by hand, guarded by once() against waiting on one it never does. */ -function gate(): Gate { +function latch(): Latch { const opener: { open?: () => void } = {}; const passed = once(resolve => { opener.open = () => { resolve(true); }; }); @@ -296,33 +346,133 @@ describe('sendSms()', () => { }); describe('reconnect', () => { - /** The server's side of the one connection under test, replaced by every reconnect. */ - function peerOf(smpp: SmppServer): Session { - const [peer] = smpp.sessions; + test('re-binds after a drop with nothing asked for, since it is the default', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp); - assert.equal(smpp.sessions.size, 1); - assert.ok(peer); + assert.ok(session); - return peer; - } + const reconnected = once(resolve => { session.on('reconnected', () => { resolve(true); }); }); - async function sendReceipt(peer: Session, smsId: string): Promise { - const sent = await peer.send({ - cmdName: 'deliver_sm', - params: { - destination_addr: '46701113311', - esm_class: consts.ESM_CLASS.MC_DELIVERY_RECEIPT, - short_message: `id:${smsId} stat:DELIVRD err:000 text:`, - source_addr: '46709771337', - }, - tlvs: { - message_state: { tagValue: consts.MESSAGE_STATE.DELIVERED }, - receipted_message_id: { tagValue: smsId }, - }, - }); + const dropped = session.sock; - assert.equal(sent.err, undefined); - } + await peerOf(smpp).close(); + await reconnected; + + assert.notEqual(session.sock, dropped, 'the session should be live on a fresh socket'); + }); + + test('reports a drop it will retry as disconnected, keeping close for the end', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp, { reconnect: { maxDelay: 100, minDelay: 20 } }); + + assert.ok(session); + + const events: string[] = []; + + session.on('close', () => { events.push('close'); }); + session.on('disconnected', () => { events.push('disconnected'); }); + + const reconnected = once(resolve => { session.on('reconnected', () => { resolve(true); }); }); + + await peerOf(smpp).close(); + await reconnected; + + assert.deepEqual(events, ['disconnected'], 'a link the loop brings back is not the end'); + + await session.close(); + + assert.deepEqual(events, ['disconnected', 'close']); + }); + + test('emits close when the session ends while the link is still down', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp); + + assert.ok(session); + + const events: string[] = []; + + session.on('close', () => { events.push('close'); }); + session.on('disconnected', () => { events.push('disconnected'); }); + + const down = once(resolve => { session.on('disconnected', () => { resolve(true); }); }); + + await peerOf(smpp).close(); + await down; + await session.close(); + + assert.deepEqual(events, ['disconnected', 'close']); + + await session.close(); + + assert.deepEqual(events, ['disconnected', 'close'], 'closing twice is still one close'); + }); + + test('re-binds after a stream it cannot read, rather than ending the session', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp, { reconnect: { maxDelay: 100, minDelay: 20 } }); + + assert.ok(session); + + const events: string[] = []; + + session.on('close', () => { events.push('close'); }); + session.on('sessionError', () => { events.push('sessionError'); }); + + const reconnected = once(resolve => { session.on('reconnected', () => { resolve(true); }); }); + + peerOf(smpp).sock.write(unreadablePdu); + await reconnected; + + assert.deepEqual(events, ['sessionError']); + }); + + test('refuses a backoff that would retry without pausing', () => { + assert.match(checkSessionOptions({ reconnect: { minDelay: 0 } }).err?.message ?? '', /minDelay/); + assert.match(checkSessionOptions({ reconnect: { maxDelay: -1 } }).err?.message ?? '', /maxDelay/); + assert.match( + checkSessionOptions({ reconnect: null }).err?.message ?? '', + /false/, + 'off is spelled false, so nothing else may stand in for it', + ); + assert.match( + checkSessionOptions({ reconnect: { maxDelay: 1000, minDelay: 30_000 } }).err?.message ?? '', + /maxDelay/, + 'a transposed pair asks to never retry faster than 30 s and gets one every second', + ); + assert.match(checkSessionOptions({ reconnect: { minDelayMs: 20 } }).err?.message ?? '', /minDelayMs/); + assert.equal(checkSessionOptions({ reconnect: false }).err, undefined); + assert.equal(checkSessionOptions({ reconnect: { maxDelay: 60_000, minDelay: 500 } }).err, undefined); + }); + + test('reports a drop as close, and schedules nothing, when reconnect is false', async t => { + const smpp = await startServer(t); + const noop = (): void => undefined; + const infos: string[] = []; + const log: SmppLog = { + debug: noop, + error: noop, + info: msg => { infos.push(msg); }, + verbose: noop, + warn: noop, + }; + const { session } = await connect(t, smpp, { log, reconnect: false }); + + assert.ok(session); + + let disconnects = 0; + + session.on('disconnected', () => { disconnects++; }); + + const closed = once(resolve => { session.on('close', () => { resolve(true); }); }); + + peerOf(smpp).sock.write(unreadablePdu); + await closed; + + assert.equal(disconnects, 0); + assert.ok(!infos.includes('reconnect - retrying after a drop')); + }); test('re-binds after the connection drops, keeping the same session object', async t => { const smpp = await startServer(t); @@ -342,6 +492,11 @@ describe('reconnect', () => { assert.ok(session); const reconnected = once(resolve => { session.on('reconnected', () => { resolve(true); }); }); + const halfPdu = once(resolve => { session.on('data', () => { resolve(true); }); }); + + // A PDU header promising 32 octets and sending 8: the next link must not continue it. + peerOf(smpp).sock.write(Buffer.from([0, 0, 0, 32, 0, 0, 0, 4])); + await halfPdu; // Drop the connection from the server's side, as a peer restart would. for (const serverSession of smpp.sessions) { @@ -419,6 +574,398 @@ describe('reconnect', () => { }); }); +describe('sends across a reconnect', () => { + /** Answers every message after the first, which is left to hold the send window open. */ + function answerAfterTheFirst(smpp: SmppServer, arrived: string[]): Latch { + const first = latch(); + + smpp.on('session', peer => { + peer.on('sms', async sms => { + arrived.push(sms.message); + + if (arrived.length === 1) first.open(); + else await sms.sendResp(); + }); + }); + + return first; + } + + test('holds a send issued while the link is down and puts it on the new link', async t => { + const smpp = await startServer(t); + const arrived: string[] = []; + + smpp.on('session', peer => { peer.on('sms', async sms => { arrived.push(sms.message); await sms.sendResp(); }); }); + + const { session } = await connect(t, smpp, { reconnect: { maxDelay: 100, minDelay: 20 } }); + + assert.ok(session); + + const down = once(resolve => { session.on('disconnected', () => { resolve(true); }); }); + + await peerOf(smpp).close(); + await down; + + const sent = await session.sendSms({ from: '46701113311', message: 'held', to: '46709771337' }); + + assert.equal(sent.err, undefined); + assert.equal(sent.smsIds.length, 1); + assert.equal(sent.unanswered, 0); + assert.deepEqual(arrived, ['held']); + }); + + test('puts a segment still queued behind a full window on the new link', async t => { + const smpp = await startServer(t); + const arrived: string[] = []; + const first = answerAfterTheFirst(smpp, arrived); + const { session } = await connect(t, smpp, { + maxOutstanding: 1, + reconnect: { maxDelay: 100, minDelay: 20 }, + }); + + assert.ok(session); + + const holding = session.sendSms({ from: '46701113311', message: 'first', to: '46709771337' }); + + await first.passed; + + const queued = session.sendSms({ from: '46701113311', message: 'second', to: '46709771337' }); + + peerOf(smpp).sock.destroy(); + + const [dropped, resent] = await Promise.all([holding, queued]); + + assert.equal(dropped.unanswered, 1); + assert.equal(resent.err, undefined, 'a request that never reached the socket is not lost with it'); + assert.equal(resent.smsIds.length, 1); + assert.deepEqual(arrived, ['first', 'second']); + }); + + test('holds a send issued while the rebind is still binding', async t => { + const binding = latch(); + const release = latch(); + let binds = 0; + const smpp = await startServer(t, { + authenticate: async () => { + binds++; + + if (binds > 1) { + binding.open(); + await release.passed; + } + + return true; + }, + }); + + smpp.on('session', peer => { peer.on('sms', async sms => { await sms.sendResp(); }); }); + + const { session } = await connect(t, smpp, { reconnect: { maxDelay: 100, minDelay: 20 } }); + + assert.ok(session); + + peerOf(smpp).sock.destroy(); + + // The fresh socket is attached and the bind is in flight, so the link exists but carries nothing. + await binding.passed; + + const sending = session.sendSms({ from: '46701113311', message: 'mid-bind', to: '46709771337' }); + let settled = false; + + void sending.then(() => { settled = true; }); + await delay(30); + + assert.equal(settled, false, 'an unbound link must not take a submit_sm that would be refused'); + + release.open(); + + const sent = await sending; + + assert.equal(sent.err, undefined); + assert.equal(sent.smsIds.length, 1); + }); + + test('counts a segment the peer never answered in time as unanswered', async t => { + const smpp = await startServer(t); + + smpp.on('session', peer => { peer.on('sms', () => undefined); }); + + const { session } = await connect(t, smpp, { responseTimeout: 200 }); + + assert.ok(session); + + const sent = await session.sendSms({ from: '46701113311', message: 'no answer', to: '46709771337' }); + + assert.match(sent.err?.message ?? '', /may have accepted/); + assert.equal(sent.unanswered, 1, 'a slow SMSC may still have taken it'); + }); + + test('counts a segment aborted after it went out as unanswered', async t => { + const smpp = await startServer(t); + const arrived = once(resolve => { smpp.on('session', peer => peer.on('sms', resolve)); }); + const { session } = await connect(t, smpp); + + assert.ok(session); + + const controller = new AbortController(); + const sending = session.sendSms( + { from: '46701113311', message: 'aborted mid-flight', to: '46709771337' }, + { signal: controller.signal }, + ); + + await arrived; + controller.abort(); + + const sent = await sending; + + assert.match(sent.err?.message ?? '', /may have accepted/); + assert.equal(sent.unanswered, 1, 'the abort is ours; the peer still holds the request'); + }); + + test('reports a segment the link dropped under as unanswered, not as never sent', async t => { + const smpp = await startServer(t); + const arrived = once(resolve => { smpp.on('session', peer => peer.on('sms', resolve)); }); + const { session } = await connect(t, smpp, { reconnect: false }); + + assert.ok(session); + + const sending = session.sendSms({ from: '46701113311', message: 'in flight', to: '46709771337' }); + + await arrived; + peerOf(smpp).sock.destroy(); + + const sent = await sending; + + assert.match(sent.err?.message ?? '', /may have accepted/); + assert.equal(sent.unanswered, 1, 'the peer may have accepted it, so sending it again would duplicate'); + assert.deepEqual(sent.smsIds, []); + }); + + test('gives up a held send after responseTimeout, with nothing put on the wire', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp, { + reconnect: { maxDelay: 10_000, minDelay: 10_000 }, + responseTimeout: 200, + }); + + assert.ok(session); + + const down = once(resolve => { session.on('disconnected', () => { resolve(true); }); }); + + await peerOf(smpp).close(); + await down; + + const sent = await session.sendSms({ from: '46701113311', message: 'held', to: '46709771337' }); + + assert.match(sent.err?.message ?? '', /did not come back/); + assert.equal(sent.unanswered, 0, 'nothing reached the peer, so the message can be sent again'); + }); + + test('fails a held send when the session closes rather than leaving it waiting', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp, { + reconnect: { maxDelay: 10_000, minDelay: 10_000 }, + }); + + assert.ok(session); + + const down = once(resolve => { session.on('disconnected', () => { resolve(true); }); }); + + await peerOf(smpp).close(); + await down; + + const sending = session.sendSms({ from: '46701113311', message: 'held', to: '46709771337' }); + let settled = false; + + void sending.then(() => { settled = true; }); + await delay(30); + + assert.equal(settled, false, 'the send waits for a link rather than failing on the spot'); + + await session.close(); + + const sent = await sending; + + assert.match(sent.err?.message ?? '', /closed/); + assert.equal(sent.unanswered, 0); + }); + + test('aborts a held send instead of making it wait out the link', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp, { + reconnect: { maxDelay: 10_000, minDelay: 10_000 }, + }); + + assert.ok(session); + + const down = once(resolve => { session.on('disconnected', () => { resolve(true); }); }); + + await peerOf(smpp).close(); + await down; + + const controller = new AbortController(); + const sending = session.sendSms( + { from: '46701113311', message: 'held', to: '46709771337' }, + { signal: controller.signal }, + ); + + controller.abort(); + + const sent = await sending; + + assert.match(sent.err?.message ?? '', /Aborted while waiting for a link/); + assert.equal(sent.unanswered, 0); + }); + + test('refuses a send outright once the session is over', async t => { + const smpp = await startServer(t); + const { session } = await connect(t, smpp, { reconnect: false }); + + assert.ok(session); + await session.close(); + + const sent = await session.sendSms({ from: '46701113311', message: 'too late', to: '46709771337' }); + + assert.match(sent.err?.message ?? '', /closed/); + assert.equal(sent.unanswered, 0); + }); +}); + +describe('LinkGate', () => { + test('refuses a hold whose deadline has already passed', async () => { + let now = 0; + const gate = new LinkGate({ log: silentLog, now: () => now, timeout: 100 }); + const waitForLink = gate.hold(undefined); + + gate.shut(true); + now = 101; + + const held = await waitForLink(); + + assert.match(held.err?.message ?? '', /did not come back in time/); + }); + + test('holds on a timer that keeps the process alive', async () => { + const gate = new LinkGate({ log: silentLog, timeout: 10_000 }); + const timers = (): number => process.getActiveResourcesInfo().filter(name => name === 'Timeout').length; + + gate.shut(true); + + const before = timers(); + const held = gate.hold(undefined)(); + + assert.equal(timers(), before + 1, 'an unref\'d timer is not counted here, which is the point'); + + gate.open(); + + assert.deepEqual(await held, {}); + }); + + // addEventListener never fires for a signal that already aborted, so it would wait out the timeout. + test('gives up at once on a signal that was already aborted', async () => { + const gate = new LinkGate({ log: silentLog, timeout: 100 }); + + gate.shut(true); + + const held = await gate.hold(AbortSignal.abort())(); + + assert.match(held.err?.message ?? '', /Aborted while waiting for a link/); + }); +}); + +// Goal 4: an application that answers nothing must not grow this for the life of the link. +describe('held message bounds', () => { + function message(seqNr: number): PduObject[] { + return [submitPdu(seqNr)]; + } + + test('drops the message held longest rather than holding every one', () => { + const held = new HeldMessages({ log: silentLog, max: 2, timeout: 10_000 }); + const oldest = message(1); + + held.hold(oldest); + held.hold(message(2)); + held.hold(message(2)); + + assert.equal(held.size, 2, 'a re-used sequence number replaces rather than evicting'); + assert.equal(held.has(oldest), true); + + held.hold(message(3)); + + assert.equal(held.size, 2); + assert.equal(held.has(oldest), false); + + held.clear(); + }); + + test('gives up on a message the application never answers', () => { + let now = 0; + const held = new HeldMessages({ log: silentLog, max: 10, now: () => now, timeout: 60 }); + + held.hold(message(1)); + now = 61; + + // The next message sweeps the one that expired, so only the new one is still waited for. + held.hold(message(2)); + + assert.equal(held.size, 1); + + held.clear(); + }); + + // Without this the drain sits out its whole budget before returning what a sweep already settled. + test('wakes a waiting drain when the last message expires', async () => { + let now = 0; + const held = new HeldMessages({ log: silentLog, max: 10, now: () => now, timeout: 60 }); + + held.hold(message(1)); + + const waiting = held.idle(1000, undefined); + + now = 61; + held.sweep(); + + assert.equal(await waiting, 0); + }); +}); + +describe('sendDlr()', () => { + // A receipt cannot be resent wholesale without duplicating the segments that landed. + test('names the segments the peer took, refused, and may have taken', async t => { + const session = new Session({ sock: new net.Socket() }); + + closeAfter(t, session); + + let call = 0; + const sms = createSms({ + from: '46701113311', + message: 'three segments', + pduObjs: [submitPdu(1), submitPdu(2), submitPdu(3)], + session, + to: '46709771337', + }, { + onAnswered: () => undefined, + send: () => { + call++; + + if (call === 1) return Promise.resolve({ pduObj: submitPdu(1, 'ESME_RX_T_APPN') }); + + if (call === 2) { + return Promise.resolve({ err: new UnansweredError(new Error('nothing came back')) }); + } + + return Promise.resolve({ pduObj: submitPdu(3) }); + }, + }); + const report = await sms.sendDlr('DELIVERED'); + + assert.ok(report.err instanceof Error); + assert.match(report.err.message, /deliver_sm refused by the peer/); + assert.equal(report.pduObjs.length, 1); + assert.equal(report.unanswered, 1); + }); +}); + describe('reassembly bounds', () => { function segment(reference: number, part: number, total: number): PduObject { const udh = Buffer.from([0x05, 0x00, 0x03, reference, total, part]); @@ -599,8 +1146,13 @@ describe('AbortSignal on a send', () => { }); describe('graceful shutdown', () => { - async function submitInFlight(t: TestContext, options: Parameters[0] = {}) { - const smpp = await startServer(t); + async function submitInFlight( + t: TestContext, + options: Parameters[0] = {}, + serverOptions: Parameters[0] = {}, + message = 'answer me', + ) { + const smpp = await startServer(t, serverOptions); const incoming = once(resolve => { smpp.on('session', bound => bound.on('sms', resolve)); }); @@ -608,7 +1160,7 @@ describe('graceful shutdown', () => { assert.ok(session); - const sent = session.sendSms({ from: '46701113311', message: 'answer me', to: '46709771337' }); + const sent = session.sendSms({ from: '46701113311', message, to: '46709771337' }); return { sent, session, sms: await incoming, smpp }; } @@ -631,6 +1183,21 @@ describe('graceful shutdown', () => { assert.deepEqual(await closed, {}); }); + // The drain refuses sends; a response was never a send, and saying so is the more useful answer. + test('names a response put through send() as the misuse it is, even mid-shutdown', async t => { + const { sent, session, sms } = await submitInFlight(t); + const closing = session.close(); + const refused = await session.send({ cmdName: 'submit_sm_resp' }); + + assert.ok(refused.err instanceof Error); + assert.match(refused.err.message, /Use sendReturn\(\)/); + + await sms.sendResp({ smsId: 'answered-after-the-misuse' }); + + assert.deepEqual((await sent).smsIds, ['answered-after-the-misuse']); + assert.deepEqual(await closing, {}); + }); + test('unbind() waits out a submit already on the wire before it unbinds', async t => { const { sent, session, sms } = await submitInFlight(t); const unbound = session.unbind(); @@ -641,6 +1208,115 @@ describe('graceful shutdown', () => { assert.deepEqual(await unbound, {}); }); + test('close() waits for a message the application has not answered yet', async t => { + const { sent, smpp, sms } = await submitInFlight(t); + const closing = peerOf(smpp).close(); + + await delay(50); + await sms.sendResp({ smsId: 'answered-during-the-inbound-drain' }); + + assert.deepEqual(await closing, {}); + assert.deepEqual((await sent).smsIds, ['answered-during-the-inbound-drain']); + }); + + test('gives up on a message the application never answers', async t => { + const { smpp } = await submitInFlight(t, {}, { shutdownTimeout: 50 }); + const closed = await peerOf(smpp).close(); + + assert.ok(closed.err instanceof Error); + assert.match(closed.err.message, /1 message\(s\) unanswered/); + }); + + // Waiting forever is safe for the peer, which every request times out on. The application is not. + test('falls back to responseTimeout for a held message when the shutdown waits forever', async t => { + const { smpp } = await submitInFlight(t, {}, { responseTimeout: 200, shutdownTimeout: 0 }); + const started = Date.now(); + const closed = await peerOf(smpp).close(); + const waited = Date.now() - started; + + assert.ok(closed.err instanceof Error); + assert.match(closed.err.message, /1 message\(s\) unanswered/); + assert.ok(waited >= 190, `waited ${String(waited)} ms, so the fallback was not what bounded it`); + assert.ok(waited < 2000); + }); + + // leftOf() floors what is left at 1 ms: at 0 the request half would read "wait forever" instead. + test('still ends when the message half has spent the whole shutdown budget', async t => { + const { smpp } = await submitInFlight(t, {}, { shutdownTimeout: 100 }); + const bound = peerOf(smpp); + // The client listens for no 'sms', so this one is never answered and stays in the window. + const unanswered = bound.send({ + cmdName: 'submit_sm', + params: { + destination_addr: '46701113311', + short_message: 'nothing answers this', + source_addr: '46709771337', + }, + }); + const closed = await Promise.race([ + bound.close(), + new Promise<{ err?: Error }>(resolve => { + setTimeout(() => { resolve({ err: new Error('close() never returned') }); }, 2000).unref(); + }), + ]); + + assert.match(closed.err?.message ?? '', /1 message\(s\) unanswered; .*1 request\(s\) unfinished/); + assert.ok((await unanswered).err instanceof Error); + }); + + // The README's own listener answers and then sends its receipt, one turn later. Multipart, because + // a receipt sent one-after-a-response outruns that turn on every segment past the first. + test('a receipt sent right after the response still goes out mid-drain', async t => { + const { sent, session, smpp, sms } = await submitInFlight(t, {}, {}, 'x'.repeat(400)); + const received: Dlr[] = []; + const receipts = once(resolve => { + session.on('dlr', dlr => { + received.push(dlr); + + if (received.length === 3) resolve(received); + }); + }); + const closing = peerOf(smpp).close(); + + await sms.sendResp({ smsId: 'held-through-the-drain' }); + + const receiptSent = await sms.sendDlr('DELIVERED'); + const ids = ['held-through-the-drain-1', 'held-through-the-drain-2', 'held-through-the-drain-3']; + + assert.equal(receiptSent.err, undefined); + assert.deepEqual((await receipts).map(dlr => dlr.smsId), ids); + assert.deepEqual(await closing, {}); + assert.deepEqual((await sent).smsIds, ids); + }); + + test('a message no listener took does not hold the shutdown up', async t => { + const smpp = await startServer(t, { shutdownTimeout: 30_000 }); + const { session } = await connect(t, smpp); + + assert.ok(session); + + const bound = peerOf(smpp); + const arrived = once(resolve => { + bound.on('incomingPduObj', pduObj => { + if (pduObj.cmdName === 'submit_sm') resolve(pduObj); + }); + }); + const sent = session.sendSms({ + from: '46701113311', + message: 'nobody is listening', + to: '46709771337', + }); + + await arrived; + await delay(50); + + const started = Date.now(); + + assert.deepEqual(await bound.close(), {}); + assert.ok(Date.now() - started < 1000); + assert.ok((await sent).err instanceof Error); + }); + test('gives up on a request that outlasts shutdownTimeout', async t => { const { sent, session } = await submitInFlight(t, { shutdownTimeout: 50 }); const closed = await session.close(); @@ -651,7 +1327,8 @@ describe('graceful shutdown', () => { const result = await sent; assert.ok(result.err instanceof Error); - assert.equal(result.err.message, 'Session closed before a response arrived'); + assert.match(result.err.message, /may have accepted.*Session closed before a response arrived/); + assert.equal(result.unanswered, 1); }); // The window empties on a drop as well as on an answer, so it cannot be what the result reads. @@ -775,8 +1452,8 @@ describe('graceful shutdown', () => { assert.ok(first.sock); - const rebinding = gate(); - const release = gate(); + const rebinding = latch(); + const release = latch(); const session = new Session({ reconnect: { connect: open, @@ -810,3 +1487,78 @@ describe('graceful shutdown', () => { assert.deepEqual(reported, []); }); }); + +describe('message id notation', () => { + async function sendOne(session: Session, message: string): Promise { + return session.sendSms({ dlr: true, from: '46701113311', message, to: '46709771337' }); + } + + test('correlates a hex submit_sm_resp against a decimal receipt', async t => { + const smpp = await startServer(t); + + smpp.on('session', bound => { + bound.on('sms', sms => { void sms.sendResp({ smsId: '1a2b' }); }); + }); + + const { session } = await connect(t, smpp, { + smsIdFormat: { receipt: 'decimal', submitResp: 'hex' }, + }); + + assert.ok(session); + + const reported = once<[Dlr, PduObject]>(resolve => { + session.on('dlr', (dlr, pduObj) => { resolve([dlr, pduObj]); }); + }); + const sent = await sendOne(session, 'one segment'); + + assert.deepEqual(sent.smsIds, ['6699']); + assert.equal(paramText(sent.pduObjs[0]?.params.message_id), '1a2b', 'the PDU keeps the id it carried'); + + // The receipt renders the id in decimal and mirrors the answered one in its TLV, as the spec has it. + await sendReceipt(peerOf(smpp), '6699', '1a2b'); + + const [dlr, pduObj] = await reported; + + assert.equal(dlr.smsId, sent.smsIds[0]); + assert.equal(paramText(pduObj.tlvs.receipted_message_id?.tagValue), '1a2b'); + }); + + test('leaves the segment ids of a multipart send to merge as they are', async t => { + const smpp = await startServer(t); + + smpp.on('session', bound => { + bound.on('sms', sms => { void sms.sendResp({ smsId: 'beef' }); }); + }); + + const { session } = await connect(t, smpp, { + smsIdFormat: { receipt: 'decimal', submitResp: 'hex' }, + }); + + assert.ok(session); + + const merged = once(resolve => { session.on('messageDlr', resolve); }); + const sent = await sendOne(session, 'x'.repeat(200)); + + assert.deepEqual(sent.smsIds, ['beef-1', 'beef-2']); + + for (const smsId of sent.smsIds) { + await sendReceipt(peerOf(smpp), smsId); + } + + assert.equal((await merged).smsId, 'beef'); + }); + + test('refuses a notation it cannot apply', () => { + const checked = checkSessionOptions({ smsIdFormat: { receipt: 'octal' } }); + + assert.ok(checked.err instanceof Error); + assert.match(checked.err.message, /smsIdFormat\.receipt/); + assert.ok(checkSessionOptions({ smsIdFormat: 'hex' }).err instanceof Error); + assert.match( + checkSessionOptions({ smsIdFormat: { receipts: 'decimal' } }).err?.message ?? '', + /receipts/, + 'a misspelled place is the same silent no-op', + ); + assert.equal(checkSessionOptions({ smsIdFormat: { submitResp: 'hex' } }).err, undefined); + }); +}); diff --git a/test/session.test.ts b/test/session.test.ts index 5565ef5..d7a23d3 100644 --- a/test/session.test.ts +++ b/test/session.test.ts @@ -4,6 +4,7 @@ import test, { describe } from 'node:test'; import type { Dlr } from '../src/dlr.ts'; import type { PduObject, PduObjectInput } from '../src/pdu.ts'; import type { Sms } from '../src/sms.ts'; +import type { SmppLog } from '../src/log.ts'; import type { SmppServer } from '../src/server.ts'; import type { TestContext } from 'node:test'; import type { VoidResult } from '../src/result.ts'; @@ -235,7 +236,7 @@ describe('bind', () => { 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'); + assert.match(sent.err.message, /may have accepted.*Session closed before a response arrived/); }); test('reports an unbind the peer left unanswered on a link that stays up', async t => { @@ -1372,6 +1373,51 @@ describe('application hooks that throw or reject', () => { assert.ok(retried, 'a throwing connect should be retried, not left for the process to die on'); }); + test('keeps backing off when every link dies as soon as it comes up', async t => { + const clock = { now: 0 }; + const delays: number[] = []; + const noop = (): void => undefined; + const log: SmppLog = { + debug: noop, + error: noop, + info: (msg, metadata) => { + if (msg === 'reconnect - retrying after a drop') delays.push(Number(metadata?.delay)); + }, + verbose: noop, + warn: noop, + }; + let up = 0; + const loop = new ReconnectLoop({ + connect: () => Promise.resolve({ sock: new net.Socket() }), + log, + maxDelay: 80, + minDelay: 10, + now: () => clock.now, + onConnected: () => { + up++; + + return Promise.resolve({}); + }, + }); + + t.after(() => { loop.stop(); }); + + for (let died = 0; died < 4; died++) { + loop.schedule(); + await waitFor(() => up === died + 1); + await delay(5); + } + + assert.deepEqual(delays, [10, 20, 40, 80]); + + // A link that outlasted the longest wait earned a fresh start. + clock.now += 80; + loop.schedule(); + await waitFor(() => delays.length === 5); + + assert.deepEqual(delays, [10, 20, 40, 80, 10]); + }); + test('starts only one reconnect attempt at a time', async t => { let attempts = 0; let finish: (() => void) | undefined; @@ -1408,7 +1454,11 @@ describe('application hooks that throw or reject', () => { describe('link timers', () => { test('closes a client link the peer has stopped answering', async t => { const peer = await bindOnlyPeer(t); - const { err, session } = await client({ enquireLinkInterval: 50, port: peer.port }); + const { err, session } = await client({ + enquireLinkInterval: 50, + port: peer.port, + reconnect: false, + }); assert.equal(err, undefined); assert.ok(session); diff --git a/test/sms-id.test.ts b/test/sms-id.test.ts new file mode 100644 index 0000000..3259377 --- /dev/null +++ b/test/sms-id.test.ts @@ -0,0 +1,23 @@ +import assert from 'node:assert/strict'; +import test, { describe } from 'node:test'; +import { normaliseSmsId } from '../src/sms-id.ts'; + +describe('normaliseSmsId()', () => { + test('reads an id the length a message_id may be, and leaves a longer one alone', () => { + const padded = `0${'1'.repeat(63)}`; + const tooLong = `0${'1'.repeat(64)}`; + + assert.equal(normaliseSmsId(padded, 'decimal'), '1'.repeat(63)); + assert.equal(normaliseSmsId(tooLong, 'decimal'), tooLong); + }); + + test('leaves an id the notation cannot read as it arrived', () => { + assert.equal(normaliseSmsId('', 'hex'), ''); + assert.equal(normaliseSmsId('0x1f', 'hex'), '0x1f'); + assert.equal(normaliseSmsId('beef-1', 'hex'), 'beef-1', 'the segment convention stays whole'); + }); + + test('reads either case of a hexadecimal id', () => { + assert.equal(normaliseSmsId('1a2B', 'hex'), normaliseSmsId('1A2b', 'hex')); + }); +}); diff --git a/todo.md b/todo.md index 5bb0335..1f1cacb 100644 --- a/todo.md +++ b/todo.md @@ -1,17 +1,16 @@ # todo.md -Remaining work for the `@larvit/smpp` 1.0.0 rewrite. Read [AGENTS.md](AGENTS.md) first — the hard -rules there constrain every item below. +Remaining work for the `@larvit/smpp` 1.0.0 rewrite. Read [AGENTS.md](AGENTS.md) first — the goals +and hard rules there constrain every item below. + +This is a temporary working file: it is deleted when 1.0.0 ships, and until then it sets its own +rules. The documentation conventions in AGENTS.md do not govern it, and nothing here is a source +anything else may cite. ## Status -The rewrite is **feature complete and green**: 248 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 -docker compose run --rm node npm install -docker compose run --rm node npm test -``` +The rewrite is **feature complete and green**: the suite, lint and typecheck are clean 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. ## The agreed API @@ -55,7 +54,12 @@ Rules the API follows: | Delivery receipt parsing, TLV and text | `test/dlr.test.ts` | | Session, client, server: bind, auth, send, reassembly, DLRs, timeouts, abort, send window | `test/session.test.ts` | | Merged multipart DLRs including across a reconnect, reassembly bounds, per-send abort, the segment cap | `test/session-extras.test.ts` | +| `smsIdFormat`: a peer's `submit_sm_resp` and receipt ids read into one notation before they are compared | `test/dlr.test.ts`, `test/session-extras.test.ts` | | A draining `close()` and `unbind()`, bounded by `shutdownTimeout` or an abort | `test/session-extras.test.ts` | +| A drain that also waits out the messages the application has not answered, with `sendDlr()` the one send that passes it | `test/session-extras.test.ts` | +| `OutgoingRequests`: the gate, the window, the pending map and the retry under one owner, told when a link comes up or goes down | `test/session-extras.test.ts`, `test/session.test.ts` | +| Held messages capped and expiring, so an application that answers nothing cannot grow them | `test/session-extras.test.ts` | +| A send with no link held for the next one, and one the link dropped under counted as `unanswered` | `test/session-extras.test.ts` | | Every runnable README example | `test/readme.test.ts` | | Receipt-versus-message classification by `esm_class` | `test/dlr.test.ts`, `test/session.test.ts` | | A listener that throws, or rejects, reaching `sessionError`/`serverError` rather than the process | `test/session.test.ts`, `test/error-from.test.ts` | @@ -112,25 +116,36 @@ session message is a change to every call site. ## Worth doing, not blocking -- [ ] **In-flight sends across a reconnect.** They currently fail with "Session closed before a - response arrived" and the caller retries. Re-queueing them automatically would be friendlier - but risks duplicate delivery, so it needs a decision before it is built. -- [ ] **The drain covers only what this end sent.** `close()` and `unbind()` wait on the send window, - which `sendReturn()` never enters, so a server session tears down without waiting for the - application to answer the messages it is holding — the duplicate-on-retry outcome again, in - the SMSC direction. A full inbound drain needs a completion signal the `sms` event does not - carry, so it is a public-surface decision. Raised by review, 2026-08-30. - [ ] **Merge state does not survive a process restart.** A drop no longer discards it, but a restart loses every incomplete group, and a peer has no reason to resend a receipt it already had answered. Surviving one means exposing the merge state for the application to persist and hand back, which is a public-surface decision. -- [ ] **`session.ts` has one seam left in it**, a socket-to-PDU transport, which would move the - deliberately public `sock` field out of `Session` or turn it into a getter — a public-surface - change, so it waits for a decision. -- [ ] **Group the session's collaborators under `src/session/`.** Only `session.ts` imports - `reassembly`, `dlr-merger`, `send-window`, `link-timers`, `reconnect-loop`, `pending-requests` - and `send-sms`, so the directory would make that boundary visible. Do it on the next - extraction out of `session.ts`, not as a move of its own. +- [ ] **Group the session's collaborators under `src/session/`.** `session.ts` imports + `dlr-merger`, `incoming-requests`, `link-timers`, `outgoing-requests`, `pdu-transport`, + `reconnect-loop` and `send-sms`, and nothing else does, so the directory would make that + boundary visible. The `OutgoingRequests` extraction this was to be done with landed on + 2026-09-01, so it is the remaining half. Raised by review, 2026-09-01. + +- [ ] **`leftOf()` and the link gate's own budget are one concept counted twice.** + `idle-waiters.ts` reads what is left of a budget as `Math.max(1, deadline - now)`, because 0 + means "forever" there; `link-gate.ts` runs the same subtraction and calls `<= 0` expired. + Neither is reachable from the other, so nothing can disagree today, but a reader who learns one + and applies it to the other is wrong. A budget type both take would close it. Raised by review, + 2026-09-01. + +- [ ] **An `sms` listener that rejects before answering costs a whole `shutdownTimeout`.** + One that *throws* is fine: `emit()` catches it, returns false, and `emitSms()` releases the + hold. A rejecting `async` one reaches `sessionError` through `captureRejections`, which hands + the handler an `unknown[]` the `Sms` cannot be read out of without a cast, so nothing releases + until the message expires. Same cost as the `onRequest`-answers-nothing case that was declined, + but reached by a bug rather than a policy. Raised by review, 2026-09-01. + +- [ ] **A message the reconnect dropped is answered into the void, and reported as delivered.** + `teardown()` clears the held messages along with the inbound segments, which is right — but the + application still holds the `Sms`, so `sendResp()` writes the old link's sequence numbers to the + new socket, succeeds, and returns `{}` for a response that correlates with nothing at the peer. + `HeldMessages` now knows exactly which messages went that way, so saying so is a small + addition. Goal 2, low frequency. Raised by review, 2026-09-01. - [ ] **Does an intermediate delivery notification deserve to be a `dlr`?** `esm_class` message type `INTERMEDIATE_DELIVERY` (0x20) is classified as a message today, so a peer that reports non-final states with it hands the application a raw `id:… stat:ENROUTE` text as an inbound @@ -155,12 +170,10 @@ session message is a change to every call site. - [ ] **Coverage reporting.** `node --test --experimental-test-coverage` works today; nothing publishes the numbers. -- [ ] **Normalise the message id on both sides of a receipt.** An SMSC that answers `submit_sm_resp` - with a hex `message_id` and sends the receipt's `id:` in decimal — or pads it, or flips its - case — leaves `smsIds` and `dlr.smsId` unequal, so correlation silently yields nothing and the - application sees no receipts at all. A `dlrIdFormat` option (`'hex' | 'decimal' | 'raw'`, or a - function) applied to both ids before they are compared covers the whole class. The smallest - change on this list for the most real-world breakage removed. +- [ ] **A receipt whose fields are separated by anything but a space reads as one field.** + `parseReceipt()` takes `id:` as everything up to the next space, so a peer writing CRLF + between fields yields an id of `1a2b\r\nstat:DELIVRD` — one nothing correlates and no + notation can read. Every field pattern has the same shape. Raised by review, 2026-08-30. - [ ] **An `onReceipt` hook.** Receipt text is only loosely specified and operators disagree on it, but `dlrFromPdu()` is wired into `IncomingRequests` with no seam of its own: an application @@ -169,19 +182,11 @@ session message is a change to every call site. Mirror the `onRequest` seam — return a `Dlr` to own the receipt, `undefined` to fall through to the built-in parser. -- [ ] **Turn `reconnect` on by default in `client()`.** Surviving a dropped link is most of why the - session layer exists, and it is opt-in behind an empty object today, so an application that - does not read the options table gets none of it. A default change, so it needs a decision. - ## Declined -- **Throughput throttling — a TPS cap, and backing off on `ESME_RTHROTTLED`.** Two reasons, either - sufficient. An operator's rate limit is scoped to the account, while the widest thing this library - owns is a session: a bucket here cannot see a second process binding the same account, so it is - wrong in exactly the case it exists for. And a rate limiter's queue drains at a fixed ceiling - rather than at the peer's response rate, so a submit rate sustained above the limit grows it - without bound — and a queue holding messages the caller was told were accepted loses them on - restart, which is worse than refusing them up front. Pacing an account needs durable shared state - this library deliberately has none of. `sendSms()` surfaces `ESME_RTHROTTLED` to the caller - instead, and `maxOutstanding` stays: a window slot frees on the peer's next response, which is - self-limiting in a way a rate ceiling is not. +- **Throughput throttling — a TPS cap, and backing off on `ESME_RTHROTTLED`.** Declined by AGENTS.md + goal 7: an operator's rate limit is scoped to the account, while the widest thing this library owns + is a session, so a bucket here cannot see a second process binding the same account and is wrong in + exactly the case it exists for. `sendSms()` surfaces `ESME_RTHROTTLED` to the caller instead, and + `maxOutstanding` stays — a window slot frees on the peer's next response, which is self-limiting in + a way a rate ceiling is not.