diff --git a/src/held-messages.ts b/src/held-messages.ts index cfacbd9..dc43272 100644 --- a/src/held-messages.ts +++ b/src/held-messages.ts @@ -25,7 +25,10 @@ type HoldEntry = { release: () => void; }; -/** One message offered to the application, held until it is answered or every listener gives up. */ +/** + * One message offered to the application. A drain waits on it until the first of: `answered()`, + * every listener that took it rejecting, no listener taking it, its deadline, or the link going. + */ export class MessageHold { private readonly entry: HoldEntry; private working: number; @@ -64,6 +67,8 @@ export class HeldMessages { private readonly idleWaiters = new IdleWaiters(); private readonly log: SmppLog; private readonly maxOctets: number; + /** A rejecting listener hands the message back as an `unknown`, so its hold is found by identity. */ + private readonly offered = new WeakMap(); constructor(options: HeldMessagesOptions) { this.held = new ExpiringGroups({ @@ -112,6 +117,28 @@ export class HeldMessages { return hold; } + /** Holds a message, builds what the application is handed, and emits it. */ + offer( + pduObjs: PduObject[], + listeners: number, + build: (hold: MessageHold) => T, + emit: (message: T) => boolean, + ): void { + const hold = this.hold(pduObjs, listeners); + const message = build(hold); + + this.offered.set(message, hold); + + if (!emit(message)) hold.release(); + } + + /** One listener gave up on a message; the last one to do so is what releases it. */ + listenerRejected(message: unknown): void { + if (typeof message !== 'object' || message === null) return; + + this.offered.get(message)?.listenerGaveUp(); + } + private has(pduObjs: PduObject[]): boolean { const key = keyOf(pduObjs); diff --git a/src/incoming-requests.ts b/src/incoming-requests.ts index 4f926de..e3a3d18 100644 --- a/src/incoming-requests.ts +++ b/src/incoming-requests.ts @@ -10,7 +10,6 @@ import type { Result, VoidResult } from './result.ts'; import type { SmppLog } from './log.ts'; import type { Sms, SmsHandlers, SmsInput } from './sms.ts'; import type { SmsIdFormat } from './sms-id.ts'; -import type { MessageHold } from './held-messages.ts'; import { HeldMessages } from './held-messages.ts'; import { Reassembler, decodeSegments } from './reassembly.ts'; import { bindCommands, defaults, standsInFor } from './session-options.ts'; @@ -85,8 +84,6 @@ export class IncomingRequests { private readonly deps: IncomingDeps; private readonly dlrMerger: DlrMerger; private readonly held: HeldMessages; - /** The rejection handler is handed the Sms back as an `unknown`, so its hold is found by identity. */ - private readonly holds = new WeakMap(); private readonly log: SmppLog; private readonly reassembler: Reassembler; private readonly smsIdFormat: SmsIdFormat; @@ -172,11 +169,8 @@ export class IncomingRequests { this.reassembler.clear(); } - /** One `sms` listener gave up on a message; the last one to do so is what releases the hold. */ listenerRejected(sms: unknown): void { - if (typeof sms !== 'object' || sms === null) return; - - this.holds.get(sms)?.listenerGaveUp(); + this.held.listenerRejected(sms); } /** Waits out the messages the application still holds, and says how many it never answered. */ @@ -310,9 +304,8 @@ export class IncomingRequests { if (!first) return; const generation = this.linkGeneration; - const hold = this.held.hold(pduObjs, this.deps.smsListeners()); - const sms = this.deps.createSms({ + this.held.offer(pduObjs, this.deps.smsListeners(), hold => this.deps.createSms({ answeredAs, from: paramText(first.params.source_addr), message: decodeSegments(pduObjs), @@ -325,10 +318,6 @@ export class IncomingRequests { lostLink: () => this.linkGeneration !== generation, onAnswered: () => { hold.answered(); }, send: input => (hold.isHeld() ? this.deps.sendPastDrain(input) : this.deps.send(input)), - }); - - this.holds.set(sms, hold); - - if (!this.deps.offerSms(sms)) hold.release(); + }), sms => this.deps.offerSms(sms)); } } diff --git a/todo.md b/todo.md index 4602348..123c150 100644 --- a/todo.md +++ b/todo.md @@ -204,9 +204,6 @@ and 5. Every seat ranked the session's lifecycle hardest and least wanted to mod - [ ] **Lift Locality to 7, and confirm it with a scoring run.** A run reading 7.0 or above also retires the #30 decision. The sub-items are what the 2026-09-28 run named, most seats first. -- [ ] **Give the held-message flow one place a reader can follow it.** Whether a drain still waits - on a message is spread over `emitSms()`, `MessageHold`, the session's rejection route and - `Sms.isHeld()`. Four seats. - [ ] **Shrink the `IncomingDeps` closure bag.** 16 lambdas, six of them repeated in `SmsHandlers`, which makes every inbound call path indirect. Two seats.