fix: close query consistency tails

Author: rockyshi1993

changelogs/unreleased.md

@@ -14,6 +14,7 @@
 
 ## Fixed / changed
 
+- Fixed aggregate direct `.toArray()` to honor cache/meta execution paths, extended `find()` ObjectId auto-conversion to comparison operators, forwarded CountQueue abort signals into MongoDB count options, and added a warning when sync idempotency falls back to in-memory storage.
 - Added a short-lived read-cache dirty barrier around writes and transaction commits. Cached reads now bypass and avoid refilling query cache while a namespace is being invalidated, reducing stale-cache windows when a process exits between a database write and post-write invalidation.
 - Added optional Change Stream sync idempotency gates (`sync.idempotency`) with per-target keys and duplicate stats, so supervised restarts can skip targets already marked as applied before saving the shared resume token.
 - Added strict optimistic-locking support to Model `updateBatch(..., { versionMode: 'strict' })`; default `counter` behavior remains unchanged.

src/adapters/mongodb/common/collection-accessor.ts

@@ -294,7 +294,11 @@ export class MongoCollectionAccessor<TSchema extends Document = Document> {
         const cacheTTL = typeof merged.cache === 'number' ? merged.cache : 0;
         const { cache: _cache, ...keyOptions } = merged;
         void _cache;
-        const executeCount = () => countDocuments(this.collectionRef, normalizedQuery ?? {}, keyOptions as Parameters<Collection<TSchema>['countDocuments']>[1]);
+        const executeCount = (signal?: AbortSignal) => countDocuments(
+            this.collectionRef,
+            normalizedQuery ?? {},
+            (signal ? { ...keyOptions, signal } : keyOptions) as Parameters<Collection<TSchema>['countDocuments']>[1],
+        );
         const countQueue = this.management.defaults?.countQueue;
         if (
             cacheTTL > 0

src/adapters/mongodb/queries/index.ts

@@ -413,7 +413,7 @@ class AggregateChain<TResult = unknown, TSchema extends Document = Document> imp
         return wrapQueryResultWithMeta(this.collection, this.defaults, op, this.options, startTs, result);
     }
 
-    toArray(): Promise<TResult[]> {
+    private runToArray(): Promise<TResult[]> {
         if (this.executed) {
             throw createError(ErrorCodes.INVALID_OPERATION, 'Query already executed.');
         }
@@ -426,6 +426,13 @@ class AggregateChain<TResult = unknown, TSchema extends Document = Document> imp
             });
     }
 
+    toArray(): Promise<TResult[]> {
+        if (this.executed) {
+            throw createError(ErrorCodes.INVALID_OPERATION, 'Query already executed.');
+        }
+        return this.executeResult() as Promise<TResult[]>;
+    }
+
     private async executeResult(): Promise<TResult[] | ResultWithMeta<TResult[]> | NodeJS.ReadableStream> {
         const startTs = Date.now();
 
@@ -453,11 +460,11 @@ class AggregateChain<TResult = unknown, TSchema extends Document = Document> imp
                 return this.wrapResult('aggregate', startTs, cached as TResult[]) as TResult[] | ResultWithMeta<TResult[]>;
             }
             const qc = this.queryCache;
-            const result = await this.toArray();
+            const result = await this.runToArray();
             await Promise.resolve(qc.set(cacheKey, result, cacheTTL));
             return this.wrapResult('aggregate', startTs, result) as TResult[] | ResultWithMeta<TResult[]>;
         }
-        return this.toArray().then((result) => this.wrapResult('aggregate', startTs, result) as TResult[] | ResultWithMeta<TResult[]>);
+        return this.runToArray().then((result) => this.wrapResult('aggregate', startTs, result) as TResult[] | ResultWithMeta<TResult[]>);
     }
 
     then<TResult1 = TResult[], TResult2 = never>(
@@ -582,6 +589,7 @@ export async function countDocuments<TSchema extends Document = Document>(
     const countOptions = buildCountDriverOptions<TSchema>(rawOptions);
     const maxTimeMS = rawOptions.maxTimeMS as number | undefined;
     const comment = rawOptions.comment as string | undefined;
+    const signal = rawOptions.signal as AbortSignal | undefined;
     const canUseEstimatedCount = isEmptyQuery
         && !hasSessionOption(rawOptions)
         && rawOptions.collation === undefined
@@ -610,6 +618,7 @@ export async function countDocuments<TSchema extends Document = Document>(
         const estimatedOptions: Record<string, unknown> = {};
         if (maxTimeMS !== undefined) estimatedOptions.maxTimeMS = maxTimeMS;
         if (comment) estimatedOptions.comment = comment;
+        if (signal) estimatedOptions.signal = signal;
         return collection.estimatedDocumentCount(estimatedOptions as Parameters<Collection<TSchema>['estimatedDocumentCount']>[0]);
     }
 

src/adapters/mongodb/queries/query-helpers.ts

@@ -144,13 +144,16 @@ function normalizeQueryArray(
     });
 }
 
+const QUERY_OBJECTID_ARRAY_OPERATORS = new Set(['$in', '$nin', '$all']);
+const QUERY_OBJECTID_SCALAR_OPERATORS = new Set(['$eq', '$ne', '$gt', '$gte', '$lt', '$lte']);
+
 /**
  * Recursively normalizes a query filter, converting eligible 24-char hex strings to ObjectId.
  *
  * Notes:
  * - This is the query-path-specific filter normalizer.
  * - Unlike `utils/objectid-converter.ts` (general object traversal), this one preserves
- *   query operator semantics: $and/$or/$nor/$elemMatch/$in/$nin/$all/$eq/$ne.
+ *   query operator semantics: $and/$or/$nor/$elemMatch/$in/$nin/$all/$eq/$ne/$gt/$gte/$lt/$lte.
  */
 export function normalizeQueryFilter(
     filter: Record<string, unknown>,
@@ -188,13 +191,13 @@ export function normalizeQueryFilter(
             if (hasOperators) {
                 const nestedResult: Record<string, unknown> = {};
                 for (const [op, opVal] of Object.entries(nested)) {
-                    if (shouldConvert && (op === '$in' || op === '$nin' || op === '$all') && Array.isArray(opVal)) {
+                    if (shouldConvert && QUERY_OBJECTID_ARRAY_OPERATORS.has(op) && Array.isArray(opVal)) {
                         nestedResult[op] = opVal.map((item) =>
                             typeof item === 'string' && item.length === 24 && ObjectId.isValid(item)
                                 ? new ObjectId(item)
                                 : item,
                         );
-                    } else if (shouldConvert && (op === '$eq' || op === '$ne') && typeof opVal === 'string' && opVal.length === 24 && ObjectId.isValid(opVal)) {
+                    } else if (shouldConvert && QUERY_OBJECTID_SCALAR_OPERATORS.has(op) && typeof opVal === 'string' && opVal.length === 24 && ObjectId.isValid(opVal)) {
                         nestedResult[op] = new ObjectId(opVal);
                     } else if (op === '$elemMatch' && opVal && typeof opVal === 'object' && !Array.isArray(opVal)) {
                         nestedResult[op] = normalizeQueryFilter(opVal as Record<string, unknown>, autoConvert, currentPath, depth + 1);
@@ -488,6 +491,7 @@ export function buildCountDriverOptions<TSchema extends Document = Document>(
         'session',
         'readConcern',
         'readPreference',
+        'signal',
     ]);
     for (const key of Object.keys(driverOptions)) {
         if (!allowed.has(key)) {

src/capabilities/sync/index.ts

@@ -409,6 +409,9 @@ export class ChangeStreamSyncManager {
         });
         this.clientFactory = options.clientFactory ?? defaultClientFactory;
         this.idempotency = resolveSyncIdempotencyRuntime(options.config.idempotency);
+        if (options.config.idempotency?.enabled === true && options.config.idempotency.store === undefined) {
+            this.logger?.warn?.('[Sync] sync.idempotency is enabled without a durable store; the memory fallback only protects replay within the current process.');
+        }
     }
 
     /**

test/integration/mongodb/aggregate.test.ts

@@ -220,6 +220,36 @@ describe('aggregate() / distinct() / count() / explain()', () => {
             assert.equal(third[0].total, 21);
         });
 
+        it('direct .toArray() honors aggregate cache option', async () => {
+            const db = runtime._adapter.db;
+            const pipeline = [
+                { $match: { category: 'clothing' } },
+                { $group: { _id: null, total: { $sum: 1 } } },
+            ];
+
+            const first = await col.aggregate(pipeline, { cache: 60_000 }).toArray();
+            assert.equal(first[0].total, 20);
+
+            await db.collection('products').insertOne({
+                sku: 'SKU-CACHE-AGG-TOARRAY',
+                category: 'clothing',
+                price: 1,
+                stock: 1,
+                rating: 5,
+                active: true,
+            });
+
+            const second = await col.aggregate(pipeline, { cache: 60_000 }).toArray();
+            assert.equal(second[0].total, 20);
+        });
+
+        it('direct .toArray() honors aggregate meta option', async () => {
+            const result = await col.aggregate([{ $match: { category: 'food' } }], { meta: true }).toArray() as any;
+            assert.ok(Array.isArray(result.data));
+            assert.equal(result.meta.op, 'aggregate');
+            assert.equal(result.meta.ns.coll, 'products');
+        });
+
         it('invalidates target collection cache after $merge', async () => {
             const db = runtime._adapter.db;
             const summary = runtime.collection('product_summary');

test/integration/mongodb/queries-chain-extended.test.ts

@@ -289,6 +289,7 @@ describe('ObjectId auto-conversion — integration', () => {
     let runtime2: any;
     let col2: any;
     const VALID_HEX = '507f1f77bcf86cd799439011';
+    const NEXT_HEX = '507f1f77bcf86cd799439012';
 
     before(async () => {
         const ctx = await bootstrap2.setup();
@@ -300,7 +301,7 @@ describe('ObjectId auto-conversion — integration', () => {
         const { ObjectId } = require('mongodb');
         await col2.insertMany([
             { name: 'X', userId: new ObjectId(VALID_HEX), score: 5 },
-            { name: 'Y', userId: new ObjectId('507f1f77bcf86cd799439012'), score: 3 },
+            { name: 'Y', userId: new ObjectId(NEXT_HEX), score: 3 },
             { name: 'Z', score: 1, tags: [new ObjectId(VALID_HEX)] },
         ]);
     });
@@ -323,6 +324,20 @@ describe('ObjectId auto-conversion — integration', () => {
         assert.ok(docs.length >= 0); // might or might not match depending on conversion
     });
 
+    it('find converts ObjectId comparison operator values', async () => {
+        const gtDocs = await col2.find({ userId: { $gt: VALID_HEX } }).toArray();
+        assert.deepEqual(gtDocs.map((doc: { name: string }) => doc.name), ['Y']);
+
+        const gteDocs = await col2.find({ userId: { $gte: VALID_HEX } }).sort({ name: 1 }).toArray();
+        assert.deepEqual(gteDocs.map((doc: { name: string }) => doc.name), ['X', 'Y']);
+
+        const ltDocs = await col2.find({ userId: { $lt: NEXT_HEX } }).toArray();
+        assert.deepEqual(ltDocs.map((doc: { name: string }) => doc.name), ['X']);
+
+        const lteDocs = await col2.find({ userId: { $lte: VALID_HEX } }).toArray();
+        assert.deepEqual(lteDocs.map((doc: { name: string }) => doc.name), ['X']);
+    });
+
     it('find with $expr operator — SPECIAL_OPERATORS skips conversion', async () => {
         const docs = await col2.find({ $expr: { $gt: ['$score', 2] } }).toArray();
         assert.ok(Array.isArray(docs));

test/unit/runtime/runtime-compat.test.ts

@@ -525,6 +525,33 @@ describe('P6 runtime compat mock path', () => {
         assert.equal(calls.distinct, 1);
     });
 
+    it('passes CountQueue AbortSignal into count driver options', async () => {
+        const controller = new AbortController();
+        let capturedOptions: Record<string, unknown> | undefined;
+        const nativeCollection = {
+            namespace: 'compat_db.users',
+            countDocuments(_query: unknown, options?: Record<string, unknown>) {
+                capturedOptions = options;
+                return resolved(7);
+            },
+        };
+        const accessor = new MongoCollectionAccessor(
+            'compat_db',
+            'users',
+            nativeCollection as never,
+            {
+                defaults: {
+                    countQueue: {
+                        execute: <T>(fn: (signal?: AbortSignal) => Promise<T>) => fn(controller.signal),
+                    },
+                },
+            },
+        );
+
+        assert.equal(await accessor.count({ active: true } as never), 7);
+        assert.equal(capturedOptions?.signal, controller.signal);
+    });
+
     it('records write cache invalidation on the active transaction session', async () => {
         const recorded: string[] = [];
         let deletedPatterns = 0;

test/unit/sync/sync.test.ts

@@ -490,6 +490,62 @@ describe('P4-C sync', () => {
         assert.equal(stats.duplicateEventCount, 1);
     });
 
+    it('warns when sync idempotency uses the in-memory fallback store', () => {
+        const warnings: string[] = [];
+        const db = {
+            databaseName: 'source_db',
+            watch: () => ({ close: () => Promise.resolve(true) }),
+        };
+
+        new MonSQLize.ChangeStreamSyncManager({
+            db,
+            config: {
+                enabled: true,
+                idempotency: { enabled: true },
+                targets: [{ name: 'target', apply: async () => undefined }],
+            },
+            logger: {
+                warn: (message: string) => warnings.push(message),
+                error: () => undefined,
+                info: () => undefined,
+                debug: () => undefined,
+            },
+        });
+
+        assert.ok(warnings.some((message) => message.includes('memory fallback only protects replay within the current process')));
+    });
+
+    it('does not warn when sync idempotency receives an explicit store', () => {
+        const warnings: string[] = [];
+        const db = {
+            databaseName: 'source_db',
+            watch: () => ({ close: () => Promise.resolve(true) }),
+        };
+
+        new MonSQLize.ChangeStreamSyncManager({
+            db,
+            config: {
+                enabled: true,
+                idempotency: {
+                    enabled: true,
+                    store: {
+                        get: () => undefined,
+                        set: () => undefined,
+                    },
+                },
+                targets: [{ name: 'target', apply: async () => undefined }],
+            },
+            logger: {
+                warn: (message: string) => warnings.push(message),
+                error: () => undefined,
+                info: () => undefined,
+                debug: () => undefined,
+            },
+        });
+
+        assert.equal(warnings.length, 0);
+    });
+
     it('stops processing changes when resume token persistence fails', async () => {
         const liveStream = new EventEmitter() as EventEmitter & { close(): Promise<boolean> };
         let closed = false;