Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
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
8 changes: 1 addition & 7 deletions src/error.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1528,13 +1528,7 @@ export function isNodeShuttingDownError(err: MongoError): boolean {
*
* @see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.md#not-writable-primary-and-node-is-recovering
*/
export function isSDAMUnrecoverableError(error: MongoError): boolean {
// NOTE: null check is here for a strictly pre-CMAP world, a timeout or
// close event are considered unrecoverable
if (error instanceof MongoParseError || error == null) {
return true;
}

export function isStateChangeError(error: MongoError): boolean {
return isRecoveringError(error) || isNotWritablePrimaryError(error);
}

Expand Down
73 changes: 43 additions & 30 deletions src/sdam/server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,12 +22,13 @@ import {
import {
type AnyError,
isNodeShuttingDownError,
isSDAMUnrecoverableError,
isStateChangeError,
MONGODB_ERROR_CODES,
MongoError,
MongoErrorLabel,
MongoNetworkError,
MongoNetworkTimeoutError,
MongoParseError,
MongoRuntimeError,
MongoServerClosedError,
type MongoServerError,
Expand Down Expand Up @@ -391,9 +392,7 @@ export class Server extends TypedEventEmitter<ServerEvents> {
return;
}

const isStaleError =
error.connectionGeneration && error.connectionGeneration < this.pool.generation;
if (isStaleError) {
if (isStaleError(this, error)) {
return;
}

Expand All @@ -402,32 +401,40 @@ export class Server extends TypedEventEmitter<ServerEvents> {
const isNetworkTimeoutBeforeHandshakeError =
error instanceof MongoNetworkError && error.beforeHandshake;
const isAuthHandshakeError = error.hasErrorLabel(MongoErrorLabel.HandshakeError);
if (isNetworkNonTimeoutError || isNetworkTimeoutBeforeHandshakeError || isAuthHandshakeError) {
// In load balanced mode we never mark the server as unknown and always
// clear for the specific service id.

// Perhaps questionable and divergent from the spec, but considering MongoParseErrors like state change errors was legacy behavior.
if (isStateChangeError(error) || error instanceof MongoParseError) {
const shouldClearPool = isNodeShuttingDownError(error);

// from the SDAM spec: The driver MUST synchronize clearing the pool with updating the topology.
// In load balanced mode: there is no monitoring, so there is no topology to update. We simply clear the pool.
// For other topologies: the `ResetPool` label instructs the topology to clear the server's pool in `updateServer()`.
if (!this.loadBalanced) {
if (shouldClearPool) {
error.addErrorLabel(MongoErrorLabel.ResetPool);
}
markServerUnknown(this, error);
process.nextTick(() => this.requestCheck());
return;
}

if (connection && shouldClearPool) {
this.pool.clear({ serviceId: connection.serviceId });
}
} else if (
isNetworkNonTimeoutError ||
isNetworkTimeoutBeforeHandshakeError ||
isAuthHandshakeError
) {
// from the SDAM spec: The driver MUST synchronize clearing the pool with updating the topology.
// In load balanced mode: there is no monitoring, so there is no topology to update. We simply clear the pool.
// For other topologies: the `ResetPool` label instructs the topology to clear the server's pool in `updateServer()`.
if (!this.loadBalanced) {
error.addErrorLabel(MongoErrorLabel.ResetPool);
markServerUnknown(this, error);
} else if (connection) {
this.pool.clear({ serviceId: connection.serviceId });
}
} else {
if (isSDAMUnrecoverableError(error)) {
if (shouldHandleStateChangeError(this, error)) {
const shouldClearPool = isNodeShuttingDownError(error);
if (this.loadBalanced && connection && shouldClearPool) {
this.pool.clear({ serviceId: connection.serviceId });
}

if (!this.loadBalanced) {
if (shouldClearPool) {
error.addErrorLabel(MongoErrorLabel.ResetPool);
}
markServerUnknown(this, error);
process.nextTick(() => this.requestCheck());
}
}
}
}
}

Expand Down Expand Up @@ -560,12 +567,6 @@ function connectionIsStale(pool: ConnectionPool, connection: Connection) {
return connection.generation !== pool.generation;
}

function shouldHandleStateChangeError(server: Server, err: MongoError) {
const etv = err.topologyVersion;
const stv = server.description.topologyVersion;
return compareTopologyVersion(stv, etv) < 0;
}

function inActiveTransaction(session: ClientSession | undefined, cmd: Document) {
return session && session.inTransaction() && !isTransactionCommand(cmd);
}
Expand All @@ -575,3 +576,15 @@ function inActiveTransaction(session: ClientSession | undefined, cmd: Document)
function isRetryableWritesEnabled(topology: Topology) {
return topology.s.options.retryWrites !== false;
}

function isStaleError(server: Server, error: MongoError): boolean {
const currentGeneration = server.pool.generation;
const generation = error.connectionGeneration;

if (generation && generation < currentGeneration) {
return true;
}

const currentTopologyVersion = server.description.topologyVersion;
return compareTopologyVersion(currentTopologyVersion, error.topologyVersion) >= 0;
}
30 changes: 8 additions & 22 deletions test/unit/error.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ import * as importsFromErrorSrc from '../../src/error';
import {
isResumableError,
isRetryableReadError,
isSDAMUnrecoverableError,
isStateChangeError,
LEGACY_NOT_PRIMARY_OR_SECONDARY_ERROR_MESSAGE,
LEGACY_NOT_WRITABLE_PRIMARY_ERROR_MESSAGE,
MONGODB_ERROR_CODES,
Expand All @@ -26,7 +26,6 @@ import {
MongoNetworkError,
MongoNetworkTimeoutError,
MongoOperationTimeoutError,
MongoParseError,
MongoRuntimeError,
MongoServerError,
MongoSystemError,
Expand Down Expand Up @@ -211,26 +210,13 @@ describe('MongoErrors', () => {
});
});

describe('#isSDAMUnrecoverableError', function () {
context('when the error is a MongoParseError', function () {
it('returns true', function () {
const error = new MongoParseError('');
expect(isSDAMUnrecoverableError(error)).to.be.true;
});
});

context('when the error is null', function () {
it('returns true', function () {
expect(isSDAMUnrecoverableError(null)).to.be.true;
});
});

describe('#isStateChangeError', function () {
context('when the error has a "node is recovering" error code', function () {
it('returns true', function () {
const error = new MongoError('');
// Code for NotPrimaryOrSecondary
error.code = 13436;
expect(isSDAMUnrecoverableError(error)).to.be.true;
expect(isStateChangeError(error)).to.be.true;
});
});

Expand All @@ -239,7 +225,7 @@ describe('MongoErrors', () => {
const error = new MongoError('');
// Code for NotWritablePrimary
error.code = 10107;
expect(isSDAMUnrecoverableError(error)).to.be.true;
expect(isStateChangeError(error)).to.be.true;
});
});

Expand All @@ -250,7 +236,7 @@ describe('MongoErrors', () => {
// If the response includes an error code, it MUST be solely used to determine if error is a "node is recovering" or "not writable primary" error.
const error = new MongoError(NODE_IS_RECOVERING_ERROR_MESSAGE.source);
error.code = 555;
expect(isSDAMUnrecoverableError(error)).to.be.false;
expect(isStateChangeError(error)).to.be.false;
});
}
);
Expand All @@ -262,7 +248,7 @@ describe('MongoErrors', () => {
const error = new MongoError(
`this is ${LEGACY_NOT_WRITABLE_PRIMARY_ERROR_MESSAGE.source}.`
);
expect(isSDAMUnrecoverableError(error)).to.be.true;
expect(isStateChangeError(error)).to.be.true;
});
}
);
Expand All @@ -272,7 +258,7 @@ describe('MongoErrors', () => {
function () {
it('returns true', function () {
const error = new MongoError(`the ${NODE_IS_RECOVERING_ERROR_MESSAGE} from an error`);
expect(isSDAMUnrecoverableError(error)).to.be.true;
expect(isStateChangeError(error)).to.be.true;
});
}
);
Expand All @@ -284,7 +270,7 @@ describe('MongoErrors', () => {
const error = new MongoError(
`this is ${LEGACY_NOT_PRIMARY_OR_SECONDARY_ERROR_MESSAGE}, so we have a problem `
);
expect(isSDAMUnrecoverableError(error)).to.be.true;
expect(isStateChangeError(error)).to.be.true;
});
}
);
Expand Down