Skip to content

refactor: use native Promise.withResolvers and extract a shared abort-race helper #1754

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
arthurschreiber merged 1 commit into from
Jul 17, 2026
+37 −79
Merged

refactor: use native Promise.withResolvers and extract a shared abort-race helper #1754

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 36 additions & 78 deletions src/connection.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -877,20 +877,32 @@ interface RoutingData {
}

/**
* Helper function, equivalent to `Promise.withResolvers()`.
* Runs the given function, providing it with a promise that will reject
* with the given signal's abort reason once the signal is aborted.
*
* @returns An object with the properties `promise`, `resolve`, and `reject`.
* The function can race this promise against any asynchronous work it
* performs (potentially multiple times) to make itself abortable.
*
* The rejection is always considered handled, and the signal's abort
* listener is removed once the function settles.
*/
function withResolvers<T>() {
let resolve: (value: T | PromiseLike<T>) => void;
let reject: (reason?: any) => void;
async function withAbortRace<T>(signal: AbortSignal, func: (signalAborted: Promise<never>) => Promise<T>): Promise<T> {
signal.throwIfAborted();

const { promise: signalAborted, reject } = Promise.withResolvers<never>();

const promise = new Promise<T>((res, rej) => {
resolve = res;
reject = rej;
});
// Prevent unhandled rejections if the signal is aborted while
// `func` is not currently racing against `signalAborted`.
signalAborted.catch(() => {});

return { promise, resolve: resolve!, reject: reject! };
const onAbort = () => { reject(signal.reason); };
signal.addEventListener('abort', onAbort, { once: true });

try {
return await func(signalAborted);
} finally {
signal.removeEventListener('abort', onAbort);
}
}

/**
Expand Down Expand Up @@ -2216,7 +2228,7 @@ class Connection extends EventEmitter {
servername: this.config.options.serverName ? this.config.options.serverName : serverName,
};

const { promise, resolve, reject } = withResolvers<tls.TLSSocket>();
const { promise, resolve, reject } = Promise.withResolvers<tls.TLSSocket>();
const encryptsocket = tls.connect(encryptOptions);

try {
Expand Down Expand Up @@ -3333,14 +3345,7 @@ class Connection extends EventEmitter {
* @private
*/
async performTlsNegotiation(preloginPayload: PreloginPayload, signal: AbortSignal) {
signal.throwIfAborted();

const { promise: signalAborted, reject } = withResolvers<never>();

const onAbort = () => { reject(signal.reason); };
signal.addEventListener('abort', onAbort, { once: true });

try {
await withAbortRace(signal, async (signalAborted) => {
if (preloginPayload.fedAuthRequired === 1) {
this.fedAuthRequired = true;
}
Expand All @@ -3357,22 +3362,13 @@ class Connection extends EventEmitter {
signalAborted
]);
}
} finally {
signal.removeEventListener('abort', onAbort);
}
});
}

async readPreloginResponse(signal: AbortSignal): Promise<PreloginPayload> {
signal.throwIfAborted();

let messageBuffer = Buffer.alloc(0);

const { promise: signalAborted, reject } = withResolvers<never>();

const onAbort = () => { reject(signal.reason); };
signal.addEventListener('abort', onAbort, { once: true });

try {
await withAbortRace(signal, async (signalAborted) => {
const message = await Promise.race([
this.messageIo.readMessage().catch((err) => {
throw this.wrapSocketError(err);
Expand All @@ -3399,9 +3395,7 @@ class Connection extends EventEmitter {
await iterator.return();
}
}
} finally {
signal.removeEventListener('abort', onAbort);
}
});

const preloginPayload = new PreloginPayload(messageBuffer);
this.debug.payload(function() {
Expand Down Expand Up @@ -3449,7 +3443,7 @@ class Connection extends EventEmitter {
const closeSignal = this.closeController!.signal;
closeSignal.throwIfAborted();

const { promise, resolve, reject } = withResolvers<void>();
const { promise, resolve, reject } = Promise.withResolvers<void>();

const onAbort = () => { reject(closeSignal.reason); };
closeSignal.addEventListener('abort', onAbort, { once: true });
Expand All @@ -3471,14 +3465,7 @@ class Connection extends EventEmitter {
* @private
*/
async performSentLogin7WithStandardLogin(signal: AbortSignal): Promise<RoutingData | undefined> {
signal.throwIfAborted();

const { promise: signalAborted, reject } = withResolvers<never>();

const onAbort = () => { reject(signal.reason); };
signal.addEventListener('abort', onAbort, { once: true });

try {
return await withAbortRace(signal, async (signalAborted) => {
const message = await Promise.race([
this.messageIo.readMessage().catch((err) => {
throw this.wrapSocketError(err);
Expand All @@ -3500,23 +3487,14 @@ class Connection extends EventEmitter {
} else {
throw new ConnectionError('Login failed.', 'ELOGIN');
}
} finally {
signal.removeEventListener('abort', onAbort);
}
});
}

/**
* @private
*/
async performSentLogin7WithNTLMLogin(signal: AbortSignal): Promise<RoutingData | undefined> {
signal.throwIfAborted();

const { promise: signalAborted, reject } = withResolvers<never>();

const onAbort = () => { reject(signal.reason); };
signal.addEventListener('abort', onAbort, { once: true });

try {
return await withAbortRace(signal, async (signalAborted) => {
while (true) {
const message = await Promise.race([
this.messageIo.readMessage().catch((err) => {
Expand Down Expand Up @@ -3556,23 +3534,14 @@ class Connection extends EventEmitter {
throw new ConnectionError('Login failed.', 'ELOGIN');
}
}
} finally {
signal.removeEventListener('abort', onAbort);
}
});
}

/**
* @private
*/
async performSentLogin7WithFedAuth(signal: AbortSignal): Promise<RoutingData | undefined> {
signal.throwIfAborted();

const { promise: signalAborted, reject } = withResolvers<never>();

const onAbort = () => { reject(signal.reason); };
signal.addEventListener('abort', onAbort, { once: true });

try {
return await withAbortRace(signal, async (signalAborted) => {
const message = await Promise.race([
this.messageIo.readMessage().catch((err) => {
throw this.wrapSocketError(err);
Expand Down Expand Up @@ -3662,23 +3631,14 @@ class Connection extends EventEmitter {
} else {
throw new ConnectionError('Login failed.', 'ELOGIN');
}
} finally {
signal.removeEventListener('abort', onAbort);
}
});
}

/**
* @private
*/
async performLoggedInSendingInitialSql(signal: AbortSignal) {
signal.throwIfAborted();

const { promise: signalAborted, reject } = withResolvers<never>();

const onAbort = () => { reject(signal.reason); };
signal.addEventListener('abort', onAbort, { once: true });

try {
await withAbortRace(signal, async (signalAborted) => {
this.sendInitialSql();

const message = await Promise.race([
Expand All @@ -3693,9 +3653,7 @@ class Connection extends EventEmitter {
once(tokenStreamParser, 'end'),
signalAborted
]);
} finally {
signal.removeEventListener('abort', onAbort);
}
});
}
}

Expand Down
2 changes: 1 addition & 1 deletion tsconfig.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
"strict": true,
"esModuleInterop": true,
"allowSyntheticDefaultImports": true,
"lib": [ "ES2022" ],
"lib": [ "ES2024" ],
"skipLibCheck": true,
"resolveJsonModule": true,
"exactOptionalPropertyTypes": true,
Expand Down
Loading