Retry the first connect and bind on reconnect fromStart (#80)

* Regression tests for a client that retries its first connect and bind

* Retry the first connect and bind on reconnect: { fromStart: true }

* Pin that the signal bounding the first bind also closes the session

* Name a failed connect once, default the backoff where it lives, and fix the deadline advice

* Carry the failure that started the retries on the abort that ends them

* Pin one retry announcement per attempt when a link drops mid-bind

* Announce a retry when its wait is over, not when it is scheduled

* Say what the process-hold assertion expects, not what a failure would mean

src/client.ts

@@ -6,6 +6,7 @@ import type { SmsIdFormat } from './sms-id.ts';
 import type { Socket } from 'node:net';
 export type { BindType };
 
+import { ReconnectLoop } from './reconnect-loop.ts';
 import { Session } from './session.ts';
 import { checkSessionOptions, undeclaredInterfaceVersion } from './session-options.ts';
 import { connect as netConnect } from 'node:net';
@@ -13,6 +14,9 @@ import { connect as tlsConnect } from 'node:tls';
 import { defaultInterfaceVersion } from './defs/constants.ts';
 import { guardedLog } from './log.ts';
 
+/** `fromStart` puts the very first connect and bind through the same backoff loop as a drop. */
+type ReconnectTuning = { fromStart?: boolean; maxDelay?: number; minDelay?: number };
+
 export type ClientOptions = {
 	addressRange?: string;
 	addrNpi?: number;
@@ -26,7 +30,7 @@ export type ClientOptions = {
 	maxOutstanding?: number;
 	password?: string;
 	port?: number;
-	reconnect?: { maxDelay?: number; minDelay?: number } | false;
+	reconnect?: ReconnectTuning | false;
 	responseTimeout?: number;
 	shutdownTimeout?: number;
 	signal?: AbortSignal;
@@ -96,6 +100,21 @@ function openSocket(options: ClientOptions): Promise<Result<{ sock: Socket }>> {
 	});
 }
 
+/** Every connect this client makes goes through here, so a failed one is named the same way once. */
+async function connectSocket(options: ClientOptions, log: SmppLog): Promise<Result<{ sock: Socket }>> {
+	const opened = await openSocket(options);
+
+	if (opened.err) {
+		log.warn('client - could not connect', {
+			host: options.host ?? defaults.host,
+			message: opened.err.message,
+			port: options.port ?? defaults.port,
+		});
+	}
+
+	return opened;
+}
+
 function bindParams(options: ClientOptions, systemId: string) {
 	return {
 		address_range: options.addressRange ?? '',
@@ -139,13 +158,13 @@ async function bind(session: Session, options: ClientOptions): Promise<VoidResul
 	return {};
 }
 
-function reconnectFor(options: ClientOptions): ReconnectOptions | undefined {
+function reconnectFor(options: ClientOptions, log: SmppLog): ReconnectOptions | undefined {
 	if (options.reconnect === false) return undefined;
 
 	const tuning = options.reconnect ?? {};
 
 	return {
-		connect: () => openSocket(options),
+		connect: () => connectSocket(options, log),
 		maxDelay: tuning.maxDelay,
 		minDelay: tuning.minDelay,
 		onConnected: reconnected => bind(reconnected, options),
@@ -160,7 +179,7 @@ function createSession(options: ClientOptions, log: SmppLog, sock: Socket): Sess
 		idleTimeout: options.idleTimeout ?? enquireLinkInterval * defaults.idleTimeoutFactor,
 		log,
 		maxOutstanding: options.maxOutstanding,
-		reconnect: reconnectFor(options),
+		reconnect: reconnectFor(options, log),
 		responseTimeout: options.responseTimeout,
 		shutdownTimeout: options.shutdownTimeout,
 		smsIdFormat: options.smsIdFormat,
@@ -168,38 +187,22 @@ function createSession(options: ClientOptions, log: SmppLog, sock: Socket): Sess
 	});
 }
 
-async function connect(options: ClientOptions, log: SmppLog): Promise<Result<{ sock: Socket }>> {
-	const checked = checkSessionOptions(options);
-
-	if (checked.err) {
-		log.warn('client - unusable option', { message: checked.err.message });
-
-		return { err: checked.err };
-	}
-
-	const opened = await openSocket(options);
-
-	if (opened.err) {
-		log.warn('client - could not connect', {
-			host: options.host ?? defaults.host,
-			message: opened.err.message,
-			port: options.port ?? defaults.port,
-		});
-
-		return { err: opened.err };
-	}
-
-	return { sock: opened.sock };
-}
-
-/** Connects to an SMSC and binds. */
-export async function client(options: ClientOptions = {}): Promise<Result<{ session: Session }>> {
-	const log = guardedLog(options.log);
-	const opened = await connect(options, log);
+async function connectAndBind(
+	options: ClientOptions,
+	log: SmppLog,
+): Promise<Result<{ session: Session }>> {
+	const opened = await connectSocket(options, log);
 
 	if (opened.err) return { err: opened.err };
 
-	const session = createSession(options, log, opened.sock);
+	return bindOn(createSession(options, log, opened.sock), options);
+}
+
+/** Binds a session the caller has not seen yet, so a failure takes it down instead of surfacing. */
+async function bindOn(
+	session: Session,
+	options: ClientOptions,
+): Promise<Result<{ session: Session }>> {
 	const signal = options.signal;
 
 	if (signal?.aborted === true) {
@@ -208,12 +211,16 @@ export async function client(options: ClientOptions = {}): Promise<Result<{ sess
 		return { err: new Error('Aborted before binding') };
 	}
 
+	const onAbort = (): void => { void session.close({ signal }); };
+
 	// Registered before the bind: an abort landing while it is in flight has to close the session.
-	signal?.addEventListener('abort', () => { void session.close({ signal }); }, { once: true });
+	signal?.addEventListener('abort', onAbort, { once: true });
 
 	const bound = await bind(session, options);
 
 	if (bound.err) {
+		signal?.removeEventListener('abort', onAbort);
+		// close() must reach the loop's stop() before its first await, or this session retries too.
 		void session.close({ signal });
 
 		return { err: bound.err };
@@ -221,3 +228,96 @@ export async function client(options: ClientOptions = {}): Promise<Result<{ sess
 
 	return { session };
 }
+
+function retriesFromStart(reconnect: ClientOptions['reconnect']): reconnect is ReconnectTuning {
+	return reconnect !== undefined && reconnect !== false && reconnect.fromStart === true;
+}
+
+/** A fresh session per attempt, and the failure to answer an abort with when none of them binds. */
+function initialAttempts(options: ClientOptions, log: SmppLog, failed: Error) {
+	let lastErr = failed;
+
+	return {
+		bind: async (sock: Socket): Promise<Result<{ session: Session }>> => {
+			const bound = await bindOn(createSession(options, log, sock), options);
+
+			if (bound.err) lastErr = bound.err;
+
+			return bound;
+		},
+		connect: async (): Promise<Result<{ sock: Socket }>> => {
+			const opened = await connectSocket(options, log);
+
+			if (opened.err) lastErr = opened.err;
+
+			return opened;
+		},
+		lastErr: (): Error => lastErr,
+	};
+}
+
+/** Retries the first connect and bind, on the backoff a drop takes, until one of them binds. */
+function keepTrying(
+	options: ClientOptions,
+	log: SmppLog,
+	tuning: ReconnectTuning,
+	failed: Error,
+): Promise<Result<{ session: Session }>> {
+	return new Promise(resolve => {
+		const attempts = initialAttempts(options, log, failed);
+		const signal = options.signal;
+		let settled = false;
+		const loop = new ReconnectLoop({
+			connect: attempts.connect,
+			log,
+			maxDelay: tuning.maxDelay,
+			minDelay: tuning.minDelay,
+			onConnected: async sock => {
+				const bound = await attempts.bind(sock);
+
+				if (bound.err) return { err: bound.err };
+
+				settle({ session: bound.session });
+
+				return {};
+			},
+			// Awaited with no other handle, so an unref()'d wait would exit the process unbound.
+			unref: false,
+		});
+
+		function settle(result: Result<{ session: Session }>): void {
+			if (settled) return;
+
+			settled = true;
+			loop.stop();
+			signal?.removeEventListener('abort', onAbort);
+			resolve(result);
+		}
+
+		function onAbort(): void {
+			settle({ err: new Error('Aborted while connecting', { cause: attempts.lastErr() }) });
+		}
+
+		signal?.addEventListener('abort', onAbort, { once: true });
+		loop.schedule();
+	});
+}
+
+/** Connects to an SMSC and binds. */
+export async function client(options: ClientOptions = {}): Promise<Result<{ session: Session }>> {
+	const log = guardedLog(options.log);
+	const checked = checkSessionOptions(options);
+
+	if (checked.err) {
+		log.warn('client - unusable option', { message: checked.err.message });
+
+		return { err: checked.err };
+	}
+
+	const first = await connectAndBind(options, log);
+	const reconnect = options.reconnect;
+
+	if (!first.err || options.signal?.aborted === true || !retriesFromStart(reconnect)) return first;
+
+	return keepTrying(options, log, reconnect, first.err);
+}

src/reconnect-loop.ts

@@ -2,18 +2,27 @@ import type { Result, VoidResult } from './result.ts';
 import type { SmppLog } from './log.ts';
 import type { Socket } from 'node:net';
 
+export const backoffDefaults = {
+	maxDelay: 30_000,
+	minDelay: 1000,
+};
+
 export type ReconnectLoopOptions = {
 	connect: () => Promise<Result<{ sock: Socket }>>;
 	log: SmppLog;
-	maxDelay: number;
-	minDelay: number;
+	maxDelay?: number | undefined;
+	minDelay?: number | undefined;
 	now?: (() => number) | undefined;
 	/** Brings the owner back up on a freshly opened socket. An err means try again. */
 	onConnected: (sock: Socket) => Promise<VoidResult>;
+	/** Whether the wait between attempts lets the process exit. Default true. */
+	unref?: boolean | undefined;
 };
 
 /** Reopens a dropped connection, backing off between attempts until it is told to stop. */
 export class ReconnectLoop {
+	private readonly maxDelay: number;
+	private readonly minDelay: number;
 	private readonly now: () => number;
 	private readonly options: ReconnectLoopOptions;
 	private attempting = false;
@@ -23,9 +32,11 @@ export class ReconnectLoop {
 	private upAt: number | undefined;
 
 	constructor(options: ReconnectLoopOptions) {
+		this.maxDelay = options.maxDelay ?? backoffDefaults.maxDelay;
+		this.minDelay = options.minDelay ?? backoffDefaults.minDelay;
 		this.now = options.now ?? Date.now;
 		this.options = options;
-		this.delay = options.minDelay;
+		this.delay = this.minDelay;
 	}
 
 	/** Read through a method: stop() can land while an attempt is awaiting. */
@@ -37,23 +48,24 @@ export class ReconnectLoop {
 		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;
+		if (this.upAt !== undefined && this.now() - this.upAt >= this.maxDelay) {
+			this.delay = this.minDelay;
 		}
 
 		this.upAt = undefined;
 
 		const delay = this.delay;
 
-		this.options.log.info('reconnect - retrying after a drop', { delay });
-
+		// Announced when the wait is over rather than when it starts: a cancelled one never happened.
 		this.timer = setTimeout(() => {
 			this.timer = undefined;
+			this.options.log.info('reconnect - retrying', { delay });
 			void this.run();
 		}, delay);
-		this.timer.unref();
 
-		this.delay = Math.min(delay * 2, this.options.maxDelay);
+		if (this.options.unref ?? true) this.timer.unref();
+
+		this.delay = Math.min(delay * 2, this.maxDelay);
 	}
 
 	stop(): void {

src/session-options.ts

@@ -7,6 +7,7 @@ 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 { backoffDefaults } from './reconnect-loop.ts';
 import { isSmsIdNotation, smsIdNotations, smsIdPlaces } from './sms-id.ts';
 
 export type SessionEvents = {
@@ -103,12 +104,10 @@ export const defaults = {
 	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,
 	reassemblyTimeout: 300_000,
 	responseTimeout: 30_000,
 	shutdownTimeout: 5000,
@@ -120,6 +119,10 @@ export const defaults = {
  * queued behind a slot that is never freed, so the call never settles at all.
  */
 export function checkSessionOptions(options: CheckableOptions): VoidResult {
+	if (options.fromStart !== undefined) {
+		return { err: new Error('fromStart is part of the reconnect policy, spell it reconnect: { fromStart: true }') };
+	}
+
 	const checked = checkLimits([
 		['idleTimeout', options.idleTimeout ?? 0, 0],
 		['maxOutstanding', options.maxOutstanding ?? defaults.maxOutstanding, 1],
@@ -146,23 +149,27 @@ function checkLimits(limits: [string, number, number][]): VoidResult {
 	return {};
 }
 
-const backoffDelays: readonly string[] = ['maxDelay', 'minDelay'];
+const reconnectKeys: readonly string[] = ['fromStart', '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') };
+		return { err: new Error('reconnect takes { fromStart, 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 ')}`) };
+		if (!reconnectKeys.includes(key)) {
+			return { err: new Error(`reconnect has no ${key}, name ${reconnectKeys.join(', ')}`) };
 		}
 	}
 
-	const maxDelay = delayOr(reconnect.maxDelay, defaults.maxDelay);
-	const minDelay = delayOr(reconnect.minDelay, defaults.minDelay);
+	if (reconnect.fromStart !== undefined && typeof reconnect.fromStart !== 'boolean') {
+		return { err: new Error(`reconnect.fromStart must be true or false, got ${typeof reconnect.fromStart}`) };
+	}
+
+	const maxDelay = delayOr(reconnect.maxDelay, backoffDefaults.maxDelay);
+	const minDelay = delayOr(reconnect.minDelay, backoffDefaults.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]]);
 
@@ -211,6 +218,8 @@ function checkSmsIdFormat(smsIdFormat: unknown): VoidResult {
 
 /** What the checker reads, as it arrives: a caller without types can put anything in it. */
 export type CheckableOptions = {
+	/** Not an option: the one spelling is inside reconnect, and this is where the other is refused. */
+	fromStart?: unknown;
 	idleTimeout?: number | undefined;
 	maxOutstanding?: number | undefined;
 	maxReassembly?: number | undefined;

src/session.ts

@@ -265,8 +265,8 @@ export class Session extends EventEmitter<SessionEvents> {
 		return new ReconnectLoop({
 			connect: reconnect.connect,
 			log: this.log,
-			maxDelay: reconnect.maxDelay ?? defaults.maxDelay,
-			minDelay: reconnect.minDelay ?? defaults.minDelay,
+			maxDelay: reconnect.maxDelay,
+			minDelay: reconnect.minDelay,
 			onConnected: sock => this.comeBackUp(sock, reconnect.onConnected),
 		});
 	}