Skip to content

Commit 4c6478e

Browse files
nbbeekennbbeeken
committed
wip
1 parent e338669 commit 4c6478e

22 files changed

+132
-92
lines changed

‎src/change_stream.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -664,6 +664,8 @@ export class ChangeStream<
664664
this.isClosed = false;
665665
this.mode = false;
666666

667+
this.on('error', () => null);
668+
667669
// Listen for any `change` listeners being added to ChangeStream
668670
this.on('newListener', eventName => {
669671
if (eventName === 'change' && this.cursor && this.listenerCount('change') === 0) {

‎src/cmap/auth/mongodb_oidc.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -100,7 +100,8 @@ export interface Workflow {
100100
execute(
101101
connection: Connection,
102102
credentials: MongoCredentials,
103-
response?: Document
103+
response?: Document,
104+
closeSignal: AbortSignal
104105
): Promise<void>;
105106

106107
/**

‎src/cmap/connect.ts‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ import {
1616
MongoRuntimeError,
1717
needsRetryableWriteLabel
1818
} from '../error';
19-
import { addAbortSignalToStream, HostAddress, ns, promiseWithResolvers } from '../utils';
19+
import { addAbortSignalToStream, HostAddress, noop, ns, promiseWithResolvers } from '../utils';
2020
import { AuthContext } from './auth/auth_provider';
2121
import { AuthMechanism } from './auth/providers';
2222
import {
@@ -389,6 +389,7 @@ export async function makeSocket(
389389

390390
addAbortSignalToStream(closeSignal, socket);
391391

392+
socket.unref();
392393
socket.setKeepAlive(true, 300000);
393394
socket.setTimeout(connectTimeoutMS);
394395
socket.setNoDelay(noDelay);

‎src/cmap/connection.ts‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -298,6 +298,14 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
298298
);
299299
}
300300

301+
unref() {
302+
this.socket.unref();
303+
}
304+
305+
ref() {
306+
this.socket.ref();
307+
}
308+
301309
public markAvailable(): void {
302310
this.lastUseTime = now();
303311
}

‎src/cmap/connection_pool.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -413,6 +413,8 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
413413
if (!this.checkedOut.has(connection)) {
414414
return;
415415
}
416+
417+
connection.unref();
416418
const poolClosed = this.closed;
417419
const stale = this.connectionIsStale(connection);
418420
const willDestroy = !!(poolClosed || stale || connection.closed);
@@ -788,6 +790,7 @@ export class ConnectionPool extends TypedEventEmitter<ConnectionPoolEvents> {
788790
);
789791

790792
this.waitQueue.shift();
793+
connection.ref();
791794
waitQueueMember.resolve(connection);
792795
}
793796
}

‎src/mongo_client.ts‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -691,14 +691,15 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
691691

692692
/* @internal */
693693
private async _close(force = false): Promise<void> {
694-
this.closeController.abort();
695-
// There's no way to set hasBeenClosed back to false
696-
Object.defineProperty(this.s, 'hasBeenClosed', {
697-
value: true,
698-
enumerable: true,
699-
configurable: false,
700-
writable: false
701-
});
694+
try {
695+
this.closeController.abort();
696+
// There's no way to set hasBeenClosed back to false
697+
Object.defineProperty(this.s, 'hasBeenClosed', {
698+
value: true,
699+
enumerable: true,
700+
configurable: false,
701+
writable: false
702+
});
702703

703704
const activeCursorCloses = Array.from(this.s.activeCursors, cursor => cursor.close());
704705
this.s.activeCursors.clear();

‎src/sdam/monitor.ts‎

Lines changed: 8 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -384,20 +384,13 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
384384
}
385385

386386
// connecting does an implicit `hello`
387-
(async () => {
387+
const makeMonitoringConnection = async () => {
388388
const socket = await makeSocket(monitor.connectOptions, monitor.closeSignal);
389389
const connection = makeConnection(monitor.connectOptions, socket);
390390
// The start time is after socket creation but before the handshake
391391
start = now();
392392
try {
393393
await performInitialHandshake(connection, monitor.connectOptions, monitor.closeSignal);
394-
return connection;
395-
} catch (error) {
396-
connection.destroy();
397-
throw error;
398-
}
399-
})().then(
400-
connection => {
401394
if (isInCloseState(monitor)) {
402395
connection.destroy();
403396
return;
@@ -417,15 +410,16 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
417410
useStreamingProtocol(monitor, connection.hello?.topologyVersion)
418411
)
419412
);
420-
421-
callback(undefined, connection.hello);
422-
},
423-
error => {
413+
return connection.hello;
414+
} catch (error) {
415+
connection.destroy();
424416
monitor.connection = null;
425417
awaited = false;
426-
onHeartbeatFailed(error);
418+
throw error;
427419
}
428-
);
420+
};
421+
422+
makeMonitoringConnection().then(callback.bind(undefined, undefined), onHeartbeatFailed);
429423
}
430424

431425
function monitorServer(monitor: Monitor) {

‎src/utils.ts‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1571,7 +1571,16 @@ export function addAbortSignalToStream(
15711571

15721572
const abortListener = addAbortListener(signal, function () {
15731573
stream.off('close', abortListener[kDispose]).off('error', abortListener[kDispose]);
1574-
stream.destroy(this.reason);
1574+
stream.destroy(
1575+
new (class extends Error {
1576+
s = stream;
1577+
})(
1578+
`sad: ${stream.___socketId}: error listeners: ${stream.listenerCount('error')} + ${stream.___stack}`,
1579+
{
1580+
cause: this.reason
1581+
}
1582+
)
1583+
);
15751584
});
15761585
// not nearly as complex as node's eos() but... do we need all that?? sobbing emoji.
15771586
stream.once('close', abortListener[kDispose]).once('error', abortListener[kDispose]);

‎test/integration/change-streams/change_stream.test.ts‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import {
1818
MongoChangeStreamError,
1919
type MongoClient,
2020
MongoServerError,
21+
promiseWithResolvers,
2122
ReadPreference,
2223
type ResumeToken
2324
} from '../../mongodb';
@@ -62,6 +63,7 @@ describe('Change Streams', function () {
6263
await csDb.createCollection('test').catch(() => null);
6364
collection = csDb.collection('test');
6465
changeStream = collection.watch();
66+
changeStream.once('error', error => this.error(error));
6567
});
6668

6769
afterEach(async () => {
@@ -695,10 +697,18 @@ describe('Change Streams', function () {
695697
async test() {
696698
await initIteratorMode(changeStream);
697699

700+
const { promise, resolve, reject } = promiseWithResolvers<void>();
701+
698702
const outStream = new PassThrough({ objectMode: true });
699703

700-
// @ts-expect-error: transform requires a Document return type
701-
changeStream.stream({ transform: JSON.stringify }).pipe(outStream);
704+
const csStream = changeStream
705+
// @ts-expect-error: transform requires a Document return type
706+
.stream({ transform: JSON.stringify });
707+
708+
csStream.once('error', reject).pipe(outStream).once('error', reject);
709+
710+
outStream.on('close', resolve);
711+
csStream.on('close', resolve);
702712

703713
const willBeData = once(outStream, 'data');
704714

@@ -709,6 +719,8 @@ describe('Change Streams', function () {
709719
expect(parsedEvent).to.have.nested.property('fullDocument.a', 1);
710720

711721
outStream.destroy();
722+
csStream.destroy();
723+
await promise;
712724
}
713725
});
714726

‎test/integration/client-side-encryption/client_side_encryption.prose.18.azure_kms_mock_server.test.ts‎

Lines changed: 19 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,8 @@ const metadata: MongoDBMetadataUI = {
3030
}
3131
};
3232

33+
const closeSignal = new AbortController().signal;
34+
3335
context('Azure KMS Mock Server Tests', function () {
3436
context('Case 1: Success', metadata, function () {
3537
// Do not set an ``X-MongoDB-HTTP-TestParams`` header.
@@ -44,7 +46,7 @@ context('Azure KMS Mock Server Tests', function () {
4446
// 5. The token will have a resource of ``"https://vault.azure.net"``
4547

4648
it('returns a properly formatted access token', async () => {
47-
const credentials = await fetchAzureKMSToken(new KMSRequestOptions());
49+
const credentials = await fetchAzureKMSToken(new KMSRequestOptions(), closeSignal);
4850
expect(credentials).to.have.property('accessToken', 'magic-cookie');
4951
});
5052
});
@@ -59,7 +61,10 @@ context('Azure KMS Mock Server Tests', function () {
5961
// The test case should ensure that this error condition is handled gracefully.
6062

6163
it('returns an error', async () => {
62-
const error = await fetchAzureKMSToken(new KMSRequestOptions('empty-json')).catch(e => e);
64+
const error = await fetchAzureKMSToken(
65+
new KMSRequestOptions('empty-json'),
66+
closeSignal
67+
).catch(e => e);
6368

6469
expect(error).to.be.instanceof(MongoCryptAzureKMSRequestError);
6570
});
@@ -74,7 +79,9 @@ context('Azure KMS Mock Server Tests', function () {
7479
// The test case should ensure that this error condition is handled gracefully.
7580

7681
it('returns an error', async () => {
77-
const error = await fetchAzureKMSToken(new KMSRequestOptions('bad-json')).catch(e => e);
82+
const error = await fetchAzureKMSToken(new KMSRequestOptions('bad-json'), closeSignal).catch(
83+
e => e
84+
);
7885

7986
expect(error).to.be.instanceof(MongoCryptAzureKMSRequestError);
8087
});
@@ -89,7 +96,9 @@ context('Azure KMS Mock Server Tests', function () {
8996
// 2. The response body is unspecified.
9097
// The test case should ensure that this error condition is handled gracefully.
9198
it('returns an error', async () => {
92-
const error = await fetchAzureKMSToken(new KMSRequestOptions('404')).catch(e => e);
99+
const error = await fetchAzureKMSToken(new KMSRequestOptions('404'), closeSignal).catch(
100+
e => e
101+
);
93102

94103
expect(error).to.be.instanceof(MongoCryptAzureKMSRequestError);
95104
});
@@ -104,7 +113,9 @@ context('Azure KMS Mock Server Tests', function () {
104113
// 2. The response body is unspecified.
105114
// The test case should ensure that this error condition is handled gracefully.
106115
it('returns an error', async () => {
107-
const error = await fetchAzureKMSToken(new KMSRequestOptions('500')).catch(e => e);
116+
const error = await fetchAzureKMSToken(new KMSRequestOptions('500'), closeSignal).catch(
117+
e => e
118+
);
108119

109120
expect(error).to.be.instanceof(MongoCryptAzureKMSRequestError);
110121
});
@@ -117,7 +128,9 @@ context('Azure KMS Mock Server Tests', function () {
117128
// The HTTP response from the ``fake_azure`` server will take at least 1000 seconds
118129
// to complete. The request should fail with a timeout.
119130
it('returns an error after the request times out', async () => {
120-
const error = await fetchAzureKMSToken(new KMSRequestOptions('slow')).catch(e => e);
131+
const error = await fetchAzureKMSToken(new KMSRequestOptions('slow'), closeSignal).catch(
132+
e => e
133+
);
121134

122135
expect(error).to.be.instanceof(MongoCryptAzureKMSRequestError);
123136
});

0 commit comments

Comments
 (0)