Skip to content

Commit 6d25e09

Browse files
JulusianJulusian
committed
feat: reimplement limit and skip for observers
1 parent 13a2890 commit 6d25e09

10 files changed

Lines changed: 412 additions & 73 deletions

File tree

‎meteor/server/collections/changeStream/__tests__/observeMultiplexer.test.ts‎

Lines changed: 31 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import {
55
observeChangesViaChangeStream,
66
getActiveMultiplexerCount,
77
} from '../observeMultiplexer'
8+
import type { ObserveViewShape } from '@sofie-automation/corelib/dist/memoryCollection/observeView'
89

910
interface TestDoc {
1011
_id: any
@@ -83,9 +84,10 @@ async function startChanges(
8384
projection: any,
8485
deps: ObserveMultiplexerDeps<TestDoc>,
8586
cb: ReturnType<typeof changesCallbacks>,
86-
nonMutating = false
87+
nonMutating = false,
88+
shape?: ObserveViewShape<TestDoc>
8789
): Promise<ObserveMultiplexer<TestDoc>> {
88-
const m = new ObserveMultiplexer<TestDoc>(selector, projection, deps, () => undefined)
90+
const m = new ObserveMultiplexer<TestDoc>(selector, projection, shape, deps, () => undefined)
8991
await subscribeChanges(m, cb, nonMutating)
9092
return m
9193
}
@@ -100,10 +102,15 @@ async function joinChangesFeed(
100102
selector: any,
101103
projection: any,
102104
cb: any,
103-
makeDeps: () => ObserveMultiplexerDeps<TestDoc>
105+
makeDeps: () => ObserveMultiplexerDeps<TestDoc>,
106+
shape?: ObserveViewShape<TestDoc>
104107
): Promise<AbortController> {
105108
const controller = new AbortController()
106-
await observeChangesViaChangeStream(collectionName, selector, projection, cb, controller.signal, false, makeDeps)
109+
// Tear this subscriber down when EITHER the test's own controller aborts OR the shared per-test signal
110+
// does (afterEach). Otherwise a test that throws before its manual `.abort()` would leak this subscriber
111+
// into the global multiplexer registry, corrupting `getActiveMultiplexerCount()` for later tests.
112+
const signal = AbortSignal.any([controller.signal, observers.signal])
113+
await observeChangesViaChangeStream(collectionName, selector, projection, shape, cb, signal, false, makeDeps)
107114
return controller
108115
}
109116

@@ -252,7 +259,7 @@ describe('ObserveMultiplexer transition logic', () => {
252259
describe('ObserveMultiplexer snapshot-vs-stream race', () => {
253260
test('events arriving before a subscriber goes live are folded into its replay (no dup, latest state)', async () => {
254261
const h = makeHarness([{ _id: id('A'), val: 1 }])
255-
const m = new ObserveMultiplexer<TestDoc>({}, undefined, h.deps, () => undefined)
262+
const m = new ObserveMultiplexer<TestDoc>({}, undefined, undefined, h.deps, () => undefined)
256263

257264
// This event is enqueued after the initial #sync but before the subscriber's replay
258265
h.pushChange(updateEv({ _id: id('A'), val: 2 }))
@@ -294,7 +301,7 @@ describe('ObserveMultiplexer reconnect resync', () => {
294301
describe('ObserveMultiplexer observe (full-document) subscribers', () => {
295302
test('added(doc), changed(newDoc, oldDoc), removed(doc)', async () => {
296303
const h = makeHarness([{ _id: id('A'), val: 1 }])
297-
const m = new ObserveMultiplexer<TestDoc>({}, undefined, h.deps, () => undefined)
304+
const m = new ObserveMultiplexer<TestDoc>({}, undefined, undefined, h.deps, () => undefined)
298305
const cb = { added: jest.fn(), changed: jest.fn(), removed: jest.fn() }
299306
await subscribeObserve(m, cb)
300307

@@ -313,7 +320,7 @@ describe('ObserveMultiplexer observe (full-document) subscribers', () => {
313320
describe('ObserveMultiplexer nonMutatingCallbacks', () => {
314321
test('clones per subscriber by default; shares the object when nonMutating', async () => {
315322
const h = makeHarness([])
316-
const m = new ObserveMultiplexer<TestDoc>({}, undefined, h.deps, () => undefined)
323+
const m = new ObserveMultiplexer<TestDoc>({}, undefined, undefined, h.deps, () => undefined)
317324

318325
const captured: Record<string, any> = {}
319326
const cbDefault = { added: jest.fn((_id, f) => (captured.def = f)), changed: jest.fn(), removed: jest.fn() }
@@ -381,6 +388,22 @@ describe('ObserveMultiplexer dedup registry', () => {
381388
subB.abort()
382389
})
383390

391+
test('same selector+projection but different window (limit) → different multiplexers', async () => {
392+
const h = makeHarness([])
393+
const base = getActiveMultiplexerCount()
394+
const sub1 = await joinChangesFeed('coll', {}, undefined, changesCallbacks() as any, () => h.deps, {
395+
limit: 10,
396+
})
397+
const sub2 = await joinChangesFeed('coll', {}, undefined, changesCallbacks() as any, () => h.deps, {
398+
limit: 20,
399+
})
400+
// Different published windows must NOT be merged onto one multiplexer.
401+
expect(getActiveMultiplexerCount()).toBe(base + 2)
402+
sub1.abort()
403+
sub2.abort()
404+
expect(getActiveMultiplexerCount()).toBe(base)
405+
})
406+
384407
test('selectors equal up to key order share one multiplexer (canonical key)', async () => {
385408
const h = makeHarness([])
386409
const base = getActiveMultiplexerCount()
@@ -422,6 +445,7 @@ describe('ObserveMultiplexer dedup registry', () => {
422445
'coll',
423446
{},
424447
undefined,
448+
undefined,
425449
cb2 as any,
426450
ctrl2.signal,
427451
false,

‎meteor/server/collections/changeStream/changeStreamCursor.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,12 +9,15 @@ import {
99
} from '@sofie-automation/corelib/dist/mongo'
1010
import { PromisifyCallbacks } from '@sofie-automation/shared-lib/dist/lib/types'
1111
import { observeChangesViaChangeStream, observeViaChangeStream, ObserveMultiplexerDeps } from './observeMultiplexer'
12+
import type { ObserveViewShape } from '@sofie-automation/corelib/dist/memoryCollection/observeView'
1213
import type { MinimalMongoCursor } from '../collection'
1314

1415
export interface ChangeStreamCursorConfig<TDoc extends { _id: ProtectedString<any> }> {
1516
collectionName: string
1617
selector: MongoQuery<TDoc>
1718
projection: MongoFieldSpecifier<TDoc> | undefined
19+
/** Cursor shaping (sort/skip/limit) applied to the published window; `undefined` = no window. */
20+
shape: ObserveViewShape<TDoc> | undefined
1821
/** Build the deps for an observe multiplexer over this cursor's query */
1922
makeDeps: () => ObserveMultiplexerDeps<TDoc>
2023
}
@@ -45,6 +48,7 @@ export class ChangeStreamCursor<TDoc extends { _id: ProtectedString<any> }> impl
4548
this.#config.collectionName,
4649
this.#config.selector,
4750
this.#config.projection,
51+
this.#config.shape,
4852
callbacks,
4953
abort.signal,
5054
!!options?.nonMutatingCallbacks,
@@ -70,6 +74,7 @@ export class ChangeStreamCursor<TDoc extends { _id: ProtectedString<any> }> impl
7074
this.#config.collectionName,
7175
this.#config.selector,
7276
this.#config.projection,
77+
this.#config.shape,
7378
callbacks,
7479
abort.signal,
7580
false,

‎meteor/server/collections/changeStream/observeMultiplexer.ts‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,7 @@ import { PromisifyCallbacks } from '@sofie-automation/shared-lib/dist/lib/types'
1111
import { stringifyError } from '@sofie-automation/shared-lib/dist/lib/stringifyError'
1212
import { EJSON } from 'meteor/ejson'
1313
import { logger } from '../../logging'
14-
import { ObserveView, fieldsFor } from '@sofie-automation/corelib/dist/memoryCollection/observeView'
14+
import { ObserveView, ObserveViewShape, fieldsFor } from '@sofie-automation/corelib/dist/memoryCollection/observeView'
1515
import { CollectionFeedHandle } from './collectionChangeFeed'
1616

1717
type Doc = { _id: ProtectedString<any> }
@@ -80,12 +80,13 @@ export class ObserveMultiplexer<TDoc extends Doc> {
8080
constructor(
8181
selector: MongoQuery<TDoc>,
8282
projection: MongoFieldSpecifier<TDoc> | undefined,
83+
shape: ObserveViewShape<TDoc> | undefined,
8384
deps: ObserveMultiplexerDeps<TDoc>,
8485
onEmpty: () => void
8586
) {
8687
this.#deps = deps
8788
this.#onEmpty = onEmpty
88-
this.#view = new ObserveView<TDoc>(selector, projection, {
89+
this.#view = new ObserveView<TDoc>(selector, projection, shape, {
8990
added: (id, doc) => this.#pending.push({ kind: 'added', id, doc }),
9091
changed: (id, newDoc, oldDoc, fields) =>
9192
this.#pending.push({ kind: 'changed', id, newDoc, oldDoc, fields }),
@@ -308,21 +309,24 @@ const multiplexers = new Map<string, ObserveMultiplexer<any>>()
308309
* collide) which is fine for hashing but would incorrectly merge two different queries onto one
309310
* multiplexer here. EJSON canonical is injective and also serialises Date/RegExp/etc. faithfully.
310311
*/
311-
function multiplexerKey(collectionName: string, selector: unknown, projection: unknown): string {
312+
function multiplexerKey(collectionName: string, selector: unknown, projection: unknown, shape: unknown): string {
312313
const canonical = (v: unknown) => EJSON.stringify(v as any, { canonical: true })
313-
return `${collectionName}::${canonical(selector)}::${canonical(projection ?? null)}`
314+
// Include the shape (sort/skip/limit): observers with the same selector+projection but a different
315+
// window publish different sets, so they must NOT share a multiplexer.
316+
return `${collectionName}::${canonical(selector)}::${canonical(projection ?? null)}::${canonical(shape ?? null)}`
314317
}
315318

316319
function getOrCreateMultiplexer<TDoc extends Doc>(
317320
collectionName: string,
318321
selector: MongoQuery<TDoc>,
319322
projection: MongoFieldSpecifier<TDoc> | undefined,
323+
shape: ObserveViewShape<TDoc> | undefined,
320324
makeDeps: () => ObserveMultiplexerDeps<TDoc>
321325
): ObserveMultiplexer<TDoc> {
322-
const key = multiplexerKey(collectionName, selector, projection)
326+
const key = multiplexerKey(collectionName, selector, projection, shape)
323327
let m = multiplexers.get(key)
324328
if (!m) {
325-
m = new ObserveMultiplexer<TDoc>(selector, projection, makeDeps(), () => multiplexers.delete(key))
329+
m = new ObserveMultiplexer<TDoc>(selector, projection, shape, makeDeps(), () => multiplexers.delete(key))
326330
multiplexers.set(key, m)
327331
}
328332
return m
@@ -340,6 +344,7 @@ export async function observeChangesViaChangeStream<TDoc extends Doc>(
340344
collectionName: string,
341345
selector: MongoQuery<TDoc>,
342346
projection: MongoFieldSpecifier<TDoc> | undefined,
347+
shape: ObserveViewShape<TDoc> | undefined,
343348
callbacks: PromisifyCallbacks<ObserveChangesCallbacks<TDoc>>,
344349
signal: AbortSignal,
345350
nonMutating: boolean,
@@ -348,7 +353,7 @@ export async function observeChangesViaChangeStream<TDoc extends Doc>(
348353
// If the caller has already aborted, do nothing - crucially, don't create a multiplexer (which would
349354
// open a change feed) that would never gain a subscriber and so never be torn down.
350355
if (signal.aborted) return
351-
const m = getOrCreateMultiplexer(collectionName, selector, projection, makeDeps)
356+
const m = getOrCreateMultiplexer(collectionName, selector, projection, shape, makeDeps)
352357
return m.addObserveChangesSubscriber(callbacks, signal, nonMutating)
353358
}
354359

@@ -357,6 +362,7 @@ export async function observeViaChangeStream<TDoc extends Doc>(
357362
collectionName: string,
358363
selector: MongoQuery<TDoc>,
359364
projection: MongoFieldSpecifier<TDoc> | undefined,
365+
shape: ObserveViewShape<TDoc> | undefined,
360366
callbacks: PromisifyCallbacks<ObserveCallbacks<TDoc>>,
361367
signal: AbortSignal,
362368
nonMutating: boolean,
@@ -365,6 +371,6 @@ export async function observeViaChangeStream<TDoc extends Doc>(
365371
// If the caller has already aborted, do nothing - crucially, don't create a multiplexer (which would
366372
// open a change feed) that would never gain a subscriber and so never be torn down.
367373
if (signal.aborted) return
368-
const m = getOrCreateMultiplexer(collectionName, selector, projection, makeDeps)
374+
const m = getOrCreateMultiplexer(collectionName, selector, projection, shape, makeDeps)
369375
return m.addObserveSubscriber(callbacks, signal, nonMutating)
370376
}

‎meteor/server/collections/implementations/asyncCollection.ts‎

Lines changed: 20 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,9 @@ import {
22
MongoBulkWriteOperation,
33
MongoModifier,
44
MongoQuery,
5-
ObserveChangesOptions,
65
ObserveChangesCallbacks,
76
ObserveCallbacks,
7+
FindObserveChangesOptions,
88
} from '@sofie-automation/corelib/dist/mongo'
99
import { ProtectedString, unprotectString } from '@sofie-automation/corelib/dist/protectedString'
1010
import { Meteor } from 'meteor/meteor'
@@ -29,6 +29,7 @@ import {
2929
} from '../changeStream/observeMultiplexer'
3030
import { ChangeStreamCursor } from '../changeStream/changeStreamCursor'
3131
import { subscribeToCollectionChangeFeed } from '../changeStream/collectionChangeFeed'
32+
import type { ObserveViewShape } from '@sofie-automation/corelib/dist/memoryCollection/observeView'
3233

3334
/**
3435
* Translate a meteor-lib {@link FindOptions} into the options the native `mongodb` driver accepts.
@@ -152,6 +153,16 @@ export class WrappedAsyncMongoCollection<
152153
return (options?.projection ?? options?.fields) as MongoFieldSpecifier<DBInterface> | undefined
153154
}
154155

156+
/**
157+
* Extract the observe window shaping (sort/skip/limit) from find-options. Returns `undefined` when none
158+
* is set, so the observe kernel keeps its fast (non-windowed) path and multiplexers de-dup correctly.
159+
*/
160+
private shapeOf(options: FindOptions<DBInterface> | undefined): ObserveViewShape<DBInterface> | undefined {
161+
if (!options) return undefined
162+
if (options.sort === undefined && options.skip === undefined && options.limit === undefined) return undefined
163+
return { sort: options.sort, skip: options.skip, limit: options.limit }
164+
}
165+
155166
async findWithCursor(
156167
selector?: MongoQuery<DBInterface> | DBInterface['_id'],
157168
options?: Omit<FindOptions<DBInterface>, 'fields'>
@@ -161,15 +172,15 @@ export class WrappedAsyncMongoCollection<
161172
collectionName: this.name,
162173
selector: sel,
163174
projection: this.projectionOf(options),
175+
shape: this.shapeOf(options),
164176
makeDeps: () => this.observeDeps(sel),
165177
})
166178
}
167179

168180
async observeChanges(
169181
selector: MongoQuery<DBInterface> | DBInterface['_id'],
170182
callbacks: PromisifyCallbacks<ObserveChangesCallbacks<DBInterface>>,
171-
findOptions?: FindOptions<DBInterface>,
172-
callbackOptions?: ObserveChangesOptions
183+
options?: FindObserveChangesOptions<DBInterface>
173184
): Promise<Meteor.LiveQueryHandle> {
174185
// Note: this span only covers the observer setup (initial snapshot + diff), not the lifetime of the observer
175186
const span = profiler.startSpan(`MongoCollection.${this.name}.observeChanges`)
@@ -185,10 +196,11 @@ export class WrappedAsyncMongoCollection<
185196
await observeChangesViaChangeStream(
186197
this.name,
187198
sel,
188-
this.projectionOf(findOptions),
199+
this.projectionOf(options),
200+
this.shapeOf(options),
189201
callbacks,
190202
abort.signal,
191-
!!callbackOptions?.nonMutatingCallbacks,
203+
!!options?.nonMutatingCallbacks,
192204
() => this.observeDeps(sel)
193205
)
194206
if (span) span.end()
@@ -207,7 +219,7 @@ export class WrappedAsyncMongoCollection<
207219
async observe(
208220
selector: MongoQuery<DBInterface> | DBInterface['_id'],
209221
callbacks: PromisifyCallbacks<ObserveCallbacks<DBInterface>>,
210-
options?: FindOptions<DBInterface>
222+
options?: FindObserveChangesOptions<DBInterface>
211223
): Promise<Meteor.LiveQueryHandle> {
212224
// Note: this span only covers the observer setup (initial snapshot + diff), not the lifetime of the observer
213225
const span = profiler.startSpan(`MongoCollection.${this.name}.observe`)
@@ -224,9 +236,10 @@ export class WrappedAsyncMongoCollection<
224236
this.name,
225237
sel,
226238
this.projectionOf(options),
239+
this.shapeOf(options),
227240
callbacks,
228241
abort.signal,
229-
false,
242+
!!options?.nonMutatingCallbacks,
230243
() => this.observeDeps(sel)
231244
)
232245
if (span) span.end()

‎packages/corelib/src/memoryCollection/InMemoryMongoCollection.ts‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import {
1717
} from '../mongo.js'
1818
import {
1919
ObserveView,
20+
ObserveViewShape,
2021
MongoLiveQueryHandle,
2122
ObserverDeliveryScheduler,
2223
makeObserveSink,
@@ -262,6 +263,7 @@ export class InMemoryMongoCollection<TDoc extends Doc> {
262263
'observe',
263264
this.#normalizeSelector(selector),
264265
projectionOf(options),
266+
shapeOf(options),
265267
callbacks,
266268
!!options?.nonMutatingCallbacks
267269
)
@@ -276,6 +278,7 @@ export class InMemoryMongoCollection<TDoc extends Doc> {
276278
'changes',
277279
this.#normalizeSelector(selector),
278280
projectionOf(options),
281+
shapeOf(options),
279282
callbacks,
280283
!!options?.nonMutatingCallbacks
281284
)
@@ -286,6 +289,7 @@ export class InMemoryMongoCollection<TDoc extends Doc> {
286289
kind: 'observe' | 'changes',
287290
selector: MongoQuery<TDoc>,
288291
projection: MongoFieldSpecifier<TDoc> | undefined,
292+
shape: ObserveViewShape<TDoc> | undefined,
289293
callbacks: ObserveCallbacks<TDoc> | ObserveChangesCallbacks<TDoc>,
290294
nonMutating: boolean
291295
): MongoLiveQueryHandle {
@@ -298,7 +302,7 @@ export class InMemoryMongoCollection<TDoc extends Doc> {
298302
this.#deliveryScheduler
299303
)
300304

301-
const view = new ObserveView<TDoc>(selector, projection, sink)
305+
const view = new ObserveView<TDoc>(selector, projection, shape, sink)
302306

303307
const entry: InMemoryObserverEntry<TDoc> =
304308
kind === 'observe'
@@ -357,3 +361,15 @@ export class InMemoryMongoCollection<TDoc extends Doc> {
357361
function projectionOf<TDoc>(options: FindOptions<TDoc> | undefined): MongoFieldSpecifier<TDoc> | undefined {
358362
return (options?.projection ?? options?.fields) as MongoFieldSpecifier<TDoc> | undefined
359363
}
364+
365+
/**
366+
* Extract the observe window shaping (sort/skip/limit). Returns `undefined` when none is set, so the
367+
* {@link ObserveView} keeps its fast (non-windowed) path.
368+
*/
369+
function shapeOf<TDoc extends Doc>(
370+
options: FindObserveChangesOptions<TDoc> | undefined
371+
): ObserveViewShape<TDoc> | undefined {
372+
if (!options) return undefined
373+
if (options.sort === undefined && options.skip === undefined && options.limit === undefined) return undefined
374+
return { sort: options.sort, skip: options.skip, limit: options.limit }
375+
}

0 commit comments

Comments
 (0)