Skip to content

Commit

Permalink
fix(NODE-6362): cache cursor deserialization options across deseriali…
Browse files Browse the repository at this point in the history
…ze calls (#4221)
  • Loading branch information
baileympearson committed Sep 9, 2024
1 parent d6c147d commit 833eaa4
Show file tree
Hide file tree
Showing 10 changed files with 129 additions and 241 deletions.
11 changes: 11 additions & 0 deletions src/bson.ts
Original file line number Diff line number Diff line change
Expand Up @@ -132,3 +132,14 @@ export function resolveBSONOptions(
options?.enableUtf8Validation ?? parentOptions?.enableUtf8Validation ?? true
};
}

/** @internal */
export function parseUtf8ValidationOption(options?: { enableUtf8Validation?: boolean }): {
utf8: { writeErrors: false } | false;
} {
const enableUtf8Validation = options?.enableUtf8Validation;
if (enableUtf8Validation === false) {
return { utf8: false };
}
return { utf8: { writeErrors: false } };
}
3 changes: 2 additions & 1 deletion src/cmap/connection.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { type DeserializeOptions } from 'bson';
import { type Readable, Transform, type TransformCallback } from 'stream';
import { clearTimeout, setTimeout } from 'timers';

Expand Down Expand Up @@ -487,7 +488,7 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {

// If `documentsReturnedIn` not set or raw is not enabled, use input bson options
// Otherwise, support raw flag. Raw only works for cursors that hardcode firstBatch/nextBatch fields
const bsonOptions =
const bsonOptions: DeserializeOptions =
options.documentsReturnedIn == null || !options.raw
? options
: {
Expand Down
32 changes: 14 additions & 18 deletions src/cmap/wire_protocol/on_demand/document.ts
Original file line number Diff line number Diff line change
@@ -1,16 +1,16 @@
import { type DeserializeOptions } from 'bson';

import {
Binary,
type BSONElement,
BSONError,
type BSONSerializeOptions,
BSONType,
deserialize,
getBigInt64LE,
getFloat64LE,
getInt32LE,
ObjectId,
parseToElementsToArray,
pluckBSONSerializeOptions,
Timestamp,
toUTF8
} from '../../../bson';
Expand Down Expand Up @@ -44,6 +44,14 @@ export type JSTypeOf = {
/** @internal */
type CachedBSONElement = { element: BSONElement; value: any | undefined };

/**
* @internal
*
* Options for `OnDemandDocument.toObject()`. Validation is required to ensure
* that callers provide utf8 validation options. */
export type OnDemandDocumentDeserializeOptions = Omit<DeserializeOptions, 'validation'> &
Required<Pick<DeserializeOptions, 'validation'>>;

/** @internal */
export class OnDemandDocument {
/**
Expand Down Expand Up @@ -330,24 +338,12 @@ export class OnDemandDocument {
* Deserialize this object, DOES NOT cache result so avoid multiple invocations
* @param options - BSON deserialization options
*/
public toObject(options?: BSONSerializeOptions): Record<string, any> {
const exactBSONOptions = {
...pluckBSONSerializeOptions(options ?? {}),
validation: this.parseBsonSerializationOptions(options),
public toObject(options?: OnDemandDocumentDeserializeOptions): Record<string, any> {
return deserialize(this.bson, {
...options,
index: this.offset,
allowObjectSmallerThanBufferSize: true
};
return deserialize(this.bson, exactBSONOptions);
}

private parseBsonSerializationOptions(options?: { enableUtf8Validation?: boolean }): {
utf8: { writeErrors: false } | false;
} {
const enableUtf8Validation = options?.enableUtf8Validation;
if (enableUtf8Validation === false) {
return { utf8: false };
}
return { utf8: { writeErrors: false } };
});
}

/** Returns this document's bytes only */
Expand Down
28 changes: 23 additions & 5 deletions src/cmap/wire_protocol/responses.ts
Original file line number Diff line number Diff line change
@@ -1,16 +1,24 @@
import { type DeserializeOptions } from 'bson';

import {
type BSONElement,
type BSONSerializeOptions,
BSONType,
type Document,
Long,
parseToElementsToArray,
parseUtf8ValidationOption,
pluckBSONSerializeOptions,
type Timestamp
} from '../../bson';
import { MongoUnexpectedServerResponseError } from '../../error';
import { type ClusterTime } from '../../sdam/common';
import { decorateDecryptionResult, ns } from '../../utils';
import { type JSTypeOf, OnDemandDocument } from './on_demand/document';
import {
type JSTypeOf,
OnDemandDocument,
type OnDemandDocumentDeserializeOptions
} from './on_demand/document';

// eslint-disable-next-line no-restricted-syntax
const enum BSONElementOffset {
Expand Down Expand Up @@ -112,7 +120,8 @@ export class MongoDBResponse extends OnDemandDocument {
this.get('recoveryToken', BSONType.object)?.toObject({
promoteValues: false,
promoteLongs: false,
promoteBuffers: false
promoteBuffers: false,
validation: { utf8: true }
}) ?? null
);
}
Expand Down Expand Up @@ -165,6 +174,14 @@ export class MongoDBResponse extends OnDemandDocument {
}
return this.clusterTime ?? null;
}

public override toObject(options?: BSONSerializeOptions): Record<string, any> {
const exactBSONOptions = {
...pluckBSONSerializeOptions(options ?? {}),
validation: parseUtf8ValidationOption(options)
};
return super.toObject(exactBSONOptions);
}
}

/** @internal */
Expand Down Expand Up @@ -248,12 +265,13 @@ export class CursorResponse extends MongoDBResponse {
this.cursor.get('postBatchResumeToken', BSONType.object)?.toObject({
promoteValues: false,
promoteLongs: false,
promoteBuffers: false
promoteBuffers: false,
validation: { utf8: true }
}) ?? null
);
}

public shift(options?: BSONSerializeOptions): any {
public shift(options: OnDemandDocumentDeserializeOptions): any {
if (this.iterated >= this.batchSize) {
return null;
}
Expand Down Expand Up @@ -305,7 +323,7 @@ export class ExplainedCursorResponse extends CursorResponse {
return this._length;
}

override shift(options?: BSONSerializeOptions | undefined) {
override shift(options?: DeserializeOptions) {
if (this._length === 0) return null;
this._length -= 1;
return this.toObject(options);
Expand Down
19 changes: 15 additions & 4 deletions src/cursor/abstract_cursor.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import { Readable, Transform } from 'stream';

import { type BSONSerializeOptions, type Document, Long, pluckBSONSerializeOptions } from '../bson';
import { type OnDemandDocumentDeserializeOptions } from '../cmap/wire_protocol/on_demand/document';
import { type CursorResponse } from '../cmap/wire_protocol/responses';
import {
MongoAPIError,
Expand Down Expand Up @@ -157,6 +158,9 @@ export abstract class AbstractCursor<
/** @event */
static readonly CLOSE = 'close' as const;

/** @internal */
protected deserializationOptions: OnDemandDocumentDeserializeOptions;

/** @internal */
protected constructor(
client: MongoClient,
Expand Down Expand Up @@ -211,6 +215,13 @@ export abstract class AbstractCursor<
} else {
this.cursorSession = this.cursorClient.startSession({ owner: this, explicit: false });
}

this.deserializationOptions = {
...this.cursorOptions,
validation: {
utf8: options?.enableUtf8Validation === false ? false : true
}
};
}

/**
Expand Down Expand Up @@ -304,7 +315,7 @@ export abstract class AbstractCursor<
);

for (let count = 0; count < documentsToRead; count++) {
const document = this.documents?.shift(this.cursorOptions);
const document = this.documents?.shift(this.deserializationOptions);
if (document != null) {
bufferedDocs.push(document);
}
Expand Down Expand Up @@ -406,7 +417,7 @@ export abstract class AbstractCursor<
}

do {
const doc = this.documents?.shift(this.cursorOptions);
const doc = this.documents?.shift(this.deserializationOptions);
if (doc != null) {
if (this.transform != null) return await this.transformDocument(doc);
return doc;
Expand All @@ -425,15 +436,15 @@ export abstract class AbstractCursor<
throw new MongoCursorExhaustedError();
}

let doc = this.documents?.shift(this.cursorOptions);
let doc = this.documents?.shift(this.deserializationOptions);
if (doc != null) {
if (this.transform != null) return await this.transformDocument(doc);
return doc;
}

await this.fetchBatch();

doc = this.documents?.shift(this.cursorOptions);
doc = this.documents?.shift(this.deserializationOptions);
if (doc != null) {
if (this.transform != null) return await this.transformDocument(doc);
return doc;
Expand Down
2 changes: 1 addition & 1 deletion src/cursor/aggregation_cursor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ export class AggregationCursor<TSchema = any> extends AbstractCursor<TSchema> {
explain: verbosity ?? true
})
)
).shift(this.aggregateOptions);
).shift(this.deserializationOptions);
}

/** Add a stage to the aggregation pipeline
Expand Down
2 changes: 1 addition & 1 deletion src/cursor/find_cursor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -143,7 +143,7 @@ export class FindCursor<TSchema = any> extends AbstractCursor<TSchema> {
explain: verbosity ?? true
})
)
).shift(this.findOptions);
).shift(this.deserializationOptions);
}

/** Set the cursor query */
Expand Down
6 changes: 5 additions & 1 deletion src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -302,7 +302,11 @@ export type { ClientMetadata, ClientMetadataOptions } from './cmap/handshake/cli
export type { ConnectionPoolMetrics } from './cmap/metrics';
export type { StreamDescription, StreamDescriptionOptions } from './cmap/stream_description';
export type { CompressorName } from './cmap/wire_protocol/compression';
export type { JSTypeOf, OnDemandDocument } from './cmap/wire_protocol/on_demand/document';
export type {
JSTypeOf,
OnDemandDocument,
OnDemandDocumentDeserializeOptions
} from './cmap/wire_protocol/on_demand/document';
export type {
CursorResponse,
MongoDBResponse,
Expand Down
Loading

0 comments on commit 833eaa4

Please sign in to comment.