Skip to content

chore: remove left over oplog code #38313

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sampaiodiego wants to merge 3 commits into
base: develop
Choose a base branch
from
Open

chore: remove left over oplog code #38313

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
a7b4143
chore: remove oplog stuff
sampaiodiego Jan 23, 2026
1a91d73
refactor: move function to a new file
sampaiodiego Jan 23, 2026
5b2d342
Update apps/meteor/server/services/meteor/userReactivity.ts
sampaiodiego Jan 29, 2026
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
1 change: 0 additions & 1 deletion apps/meteor/app/lib/server/lib/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@
library files.
*/
import './notifyUsersOnMessage';
import './meteorFixes';

export { sendNotification } from './sendNotificationsOnMessage';
export { passwordPolicy } from './passwordPolicy';
Expand Down
44 changes: 0 additions & 44 deletions apps/meteor/app/lib/server/lib/meteorFixes.js
View file
Open in desktop

This file was deleted.

7 changes: 0 additions & 7 deletions apps/meteor/app/metrics/server/lib/collectMetrics.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ import { tracerSpan } from '@rocket.chat/tracing';
import connect from 'connect';
import { Facts } from 'meteor/facts-base';
import { Meteor } from 'meteor/meteor';
import { MongoInternals } from 'meteor/mongo';
import client from 'prom-client';
import gcStats from 'prometheus-gc-stats';
import _ from 'underscore';
Expand All @@ -17,8 +16,6 @@ import { settings } from '../../../settings/server';
import { getAppsStatistics } from '../../../statistics/server/lib/getAppsStatistics';
import { Info } from '../../../utils/rocketchat.info';

const { mongo } = MongoInternals.defaultRemoteCollectionDriver();

Facts.incrementServerFact = function (pkg: 'pkg' | 'fact', fact: string | number, increment: number): void {
metrics.meteorFacts.inc({ pkg, fact }, increment);
};
Expand Down Expand Up @@ -46,9 +43,6 @@ const setPrometheusData = async (): Promise<void> => {
metrics.totalAppsEnabled.set(totalActive || 0);
metrics.totalAppsFailed.set(totalFailed || 0);

const oplogQueue = (mongo as any)._oplogHandle?._entryQueue?.length || 0;
metrics.oplogQueue.set(oplogQueue);

const statistics = await Statistics.findLast();
if (!statistics) {
return;
Expand All @@ -57,7 +51,6 @@ const setPrometheusData = async (): Promise<void> => {
metrics.version.set({ version: statistics.version }, 1);
metrics.migration.set((await getControl()).version);
metrics.instanceCount.set(statistics.instanceCount);
metrics.oplogEnabled.set({ enabled: `${statistics.oplogEnabled}` }, 1);

// User statistics
metrics.totalUsers.set(statistics.totalUsers);
Expand Down
16 changes: 0 additions & 16 deletions apps/meteor/app/metrics/server/lib/metrics.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -100,22 +100,6 @@ export const metrics = {
name: 'rocketchat_instance_count',
help: 'instances running',
}),
oplogEnabled: new client.Gauge({
name: 'rocketchat_oplog_enabled',
labelNames: ['enabled'],
help: 'oplog enabled',
}),
oplogQueue: new client.Gauge({
name: 'rocketchat_oplog_queue',
labelNames: ['queue'],
help: 'oplog queue',
}),
oplog: new client.Counter({
name: 'rocketchat_oplog',
help: 'summary of oplog operations',
labelNames: ['collection', 'op'],
}),

pushQueue: new client.Gauge({
name: 'rocketchat_push_queue',
labelNames: ['queue'],
Expand Down
6 changes: 2 additions & 4 deletions apps/meteor/packages/rocketchat-mongo-config/server/index.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,8 @@ import { PassThrough } from 'stream';
import { Email } from 'meteor/email';
import { Mongo } from 'meteor/mongo';

const shouldUseNativeOplog = ['yes', 'true'].includes(String(process.env.USE_NATIVE_OPLOG).toLowerCase());
if (!shouldUseNativeOplog) {
Package['disable-oplog'] = {};
}
// we always want Meteor to disable oplog tailing
Package['disable-oplog'] = {};

// FIX For TLS error see more here https://github.com/RocketChat/Rocket.Chat/issues/9316
// TODO: Remove after NodeJS fix it, more information
Expand Down
132 changes: 16 additions & 116 deletions apps/meteor/server/services/meteor/service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,8 @@ import type { ILivechatAgent, LoginServiceConfiguration, UserStatus } from '@roc
import { LoginServiceConfiguration as LoginServiceConfigurationModel, Users } from '@rocket.chat/models';
import { wrapExceptions } from '@rocket.chat/tools';
import { Meteor } from 'meteor/meteor';
import { MongoInternals } from 'meteor/mongo';

import { processOnChange, serviceConfigCallbacks } from './userReactivity';
import { isOutgoingIntegration } from '../../../app/integrations/server/lib/definition';
import { triggerHandler } from '../../../app/integrations/server/lib/triggerHandler';
import { notifyGuestStatusChanged } from '../../../app/livechat/server/lib/guests';
Expand All @@ -21,104 +21,8 @@ import { configureEmailInboxes } from '../../features/EmailInbox/EmailInbox';
import { roomCoordinator } from '../../lib/rooms/roomCoordinator';
import { ListenersModule } from '../../modules/listeners/listeners.module';

type Callbacks = {
added(id: string, record: object): void;
changed(id: string, record: object): void;
removed(id: string): void;
};

let processOnChange: (diff: Record<string, any>, id: string) => void;
// eslint-disable-next-line no-undef
const disableOplog = !!(Package as any)['disable-oplog'];
const serviceConfigCallbacks = new Set<Callbacks>();

const disableMsgRoundtripTracking = ['yes', 'true'].includes(String(process.env.DISABLE_MESSAGE_ROUNDTRIP_TRACKING).toLowerCase());

if (disableOplog) {
// Stores the callbacks for the disconnection reactivity bellow
const userCallbacks = new Map();

// Overrides the native observe changes to prevent database polling and stores the callbacks
// for the users' tokens to re-implement the reactivity based on our database listeners
const { mongo } = MongoInternals.defaultRemoteCollectionDriver();
MongoInternals.Connection.prototype._observeChanges = async function (
{
collectionName,
selector,
options = {},
}: {
collectionName: string;
selector: Record<string, any>;
options?: {
projection?: Record<string, number>;
fields?: Record<string, number>;
};
},
_ordered: boolean,
callbacks: Callbacks,
): Promise<any> {
// console.error('Connection.Collection.prototype._observeChanges', collectionName, selector, options);
let cbs: Set<{ hashedToken: string; callbacks: Callbacks }>;
let data: { hashedToken: string; callbacks: Callbacks };
if (callbacks?.added) {
const records = await mongo
.rawCollection(collectionName)
.find(selector, {
...(options.projection || options.fields ? { projection: options.projection || options.fields } : {}),
})
.toArray();

for (const { _id, ...fields } of records) {
callbacks.added(String(_id), fields);
}

if (collectionName === 'users' && selector['services.resume.loginTokens.hashedToken']) {
cbs = userCallbacks.get(selector._id) || new Set();
data = {
hashedToken: selector['services.resume.loginTokens.hashedToken'],
callbacks,
};

cbs.add(data);
userCallbacks.set(selector._id, cbs);
}
}

if (collectionName === 'meteor_accounts_loginServiceConfiguration') {
serviceConfigCallbacks.add(callbacks);
}

return {
stop(): void {
if (cbs) {
cbs.delete(data);
}
serviceConfigCallbacks.delete(callbacks);
},
};
};

// Re-implement meteor's reactivity that uses observe to disconnect sessions when the token
// associated was removed
processOnChange = (diff: Record<string, any>, id: string): void => {
if (!diff || !('services.resume.loginTokens' in diff)) {
return;
}
const loginTokens: undefined | { hashedToken: string }[] = diff['services.resume.loginTokens'];
const tokens = loginTokens?.map(({ hashedToken }) => hashedToken);

const cbs = userCallbacks.get(id);
if (cbs) {
[...cbs]
.filter(({ hashedToken }) => tokens === undefined || !tokens.includes(hashedToken))
.forEach((item) => {
item.callbacks.removed(id);
cbs.delete(item);
});
}
};
}

settings.set = use(settings.set, (context, next) => {
next(...context);
const [record] = context;
Expand All @@ -145,28 +49,24 @@ export class MeteorService extends ServiceClassInternal implements IMeteor {
setValue(setting._id, undefined);
});

if (disableOplog) {
this.onEvent('watch.loginServiceConfiguration', ({ clientAction, id, data }) => {
if (clientAction === 'removed') {
serviceConfigCallbacks.forEach((callbacks) => {
wrapExceptions(() => callbacks.removed?.(id)).suppress();
});
return;
}
this.onEvent('watch.loginServiceConfiguration', ({ clientAction, id, data }) => {
if (clientAction === 'removed') {
serviceConfigCallbacks.forEach((callbacks) => {
wrapExceptions(() => callbacks.removed?.(id)).suppress();
});
return;
}

if (data) {
serviceConfigCallbacks.forEach((callbacks) => {
wrapExceptions(() => callbacks[clientAction === 'inserted' ? 'added' : 'changed']?.(id, data)).suppress();
});
}
});
}
if (data) {
serviceConfigCallbacks.forEach((callbacks) => {
wrapExceptions(() => callbacks[clientAction === 'inserted' ? 'added' : 'changed']?.(id, data)).suppress();
});
}
});

this.onEvent('watch.users', async (data) => {
if (disableOplog) {
if (data.clientAction === 'updated' && data.diff) {
processOnChange(data.diff, data.id);
}
if (data.clientAction === 'updated' && data.diff) {
processOnChange(data.diff, data.id);
}

if (!monitorAgents) {
Expand Down
91 changes: 91 additions & 0 deletions apps/meteor/server/services/meteor/userReactivity.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
import { MongoInternals } from 'meteor/mongo';

type Callbacks = {
added(id: string, record: object): void;
changed(id: string, record: object): void;
removed(id: string): void;
};

export const serviceConfigCallbacks = new Set<Callbacks>();

// Stores the callbacks for the disconnection reactivity bellow
const userCallbacks = new Map();

// Overrides the native observe changes to prevent database polling and stores the callbacks
// for the users' tokens to re-implement the reactivity based on our database listeners
const { mongo } = MongoInternals.defaultRemoteCollectionDriver();
MongoInternals.Connection.prototype._observeChanges = async function (
{
collectionName,
selector,
options = {},
}: {
collectionName: string;
selector: Record<string, any>;
options?: {
projection?: Record<string, number>;
fields?: Record<string, number>;
};
},
_ordered: boolean,
callbacks: Callbacks,
): Promise<any> {
let cbs: Set<{ hashedToken: string; callbacks: Callbacks }>;
let data: { hashedToken: string; callbacks: Callbacks };
if (callbacks?.added) {
const records = await mongo
.rawCollection(collectionName)
.find(selector, {
...(options.projection || options.fields ? { projection: options.projection || options.fields } : {}),
})
.toArray();

for (const { _id, ...fields } of records) {
callbacks.added(String(_id), fields);
}

if (collectionName === 'users' && selector['services.resume.loginTokens.hashedToken']) {
cbs = userCallbacks.get(selector._id) || new Set();
data = {
hashedToken: selector['services.resume.loginTokens.hashedToken'],
callbacks,
};

cbs.add(data);
userCallbacks.set(selector._id, cbs);
}
}

if (collectionName === 'meteor_accounts_loginServiceConfiguration') {
serviceConfigCallbacks.add(callbacks);
}

return {
stop(): void {
if (cbs) {
cbs.delete(data);
}
serviceConfigCallbacks.delete(callbacks);
},
};
};
Comment on lines 17 to 71
Copy link
Contributor

@coderabbitai coderabbitai bot Jan 29, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Search for observeChanges and _observeChanges usage
rg -n -g '*.{ts,tsx,js}' '\.observeChanges\s*\(' -C 3

Repository: RocketChat/Rocket.Chat

Length of output: 1553

🏁 Script executed:

#!/bin/bash
# Search for _observeChanges references
rg -n -g '*.{ts,tsx,js}' '_observeChanges' -C 3

Repository: RocketChat/Rocket.Chat

Length of output: 2804

🏁 Script executed:

#!/bin/bash
# Look for callbacks.ready usage in the codebase to understand the pattern
rg -n -g '*.{ts,tsx,js}' 'callbacks\.ready' -C 2

Repository: RocketChat/Rocket.Chat

Length of output: 48

🏁 Script executed:

#!/bin/bash
# Check the entire file to understand context
wc -l apps/meteor/server/services/meteor/userReactivity.ts

Repository: RocketChat/Rocket.Chat

Length of output: 121

🏁 Script executed:

#!/bin/bash
cat -n apps/meteor/server/services/meteor/userReactivity.ts

Repository: RocketChat/Rocket.Chat

Length of output: 3434

🏁 Script executed:

#!/bin/bash
# Search for how _observeChanges return value is used - check call sites
rg -n -g '*.{ts,tsx,js}' 'Collection.*\.find\(' -C 2
rg -n -g '*.{ts,tsx,js}' 'mongo\.collection\(' -C 2

Repository: RocketChat/Rocket.Chat

Length of output: 48

🏁 Script executed:

#!/bin/bash
# Look for Meteor/MongoDB connection usage to understand the contract better
rg -n -g '*.{ts,tsx,js}' 'MongoInternals\.Connection\.prototype' -A 5

Repository: RocketChat/Rocket.Chat

Length of output: 527

🏁 Script executed:

#!/bin/bash
# Check if there are other observeChanges overrides or implementations
rg -n -g '*.{ts,tsx,js}' 'observeChanges.*=' -B 2 -A 5

Repository: RocketChat/Rocket.Chat

Length of output: 793

🏁 Script executed:

#!/bin/bash
# Look for usage of observe/watch patterns and handle properties
rg -n -g '*.{ts,tsx,js}' '\.stop\(\)' -B 3 -A 1

Repository: RocketChat/Rocket.Chat

Length of output: 30556

🏁 Script executed:

#!/bin/bash
# Search for sort/limit in find operations related to observeChanges
rg -n -g '*.{ts,tsx,js}' 'sort|limit' -B 2 -A 2 | head -100

Repository: RocketChat/Rocket.Chat

Length of output: 9011

🏁 Script executed:

#!/bin/bash
# Look for Collection.find calls with options
rg -n -g '*.{ts,tsx,js}' 'collection.*observe' -i -B 2 -A 2

Repository: RocketChat/Rocket.Chat

Length of output: 3795

🏁 Script executed:

#!/bin/bash
# Check if there's any Meteor documentation or type definitions in the repo
fd -e 'd.ts' -e 'types.ts' | xargs grep -l 'observeChanges\|_observeChanges' 2>/dev/null || echo "No type defs found"

Repository: RocketChat/Rocket.Chat

Length of output: 84

🏁 Script executed:

#!/bin/bash
# Look for uses of observe or observeChanges on server that might use the handle
rg -n -g '*.{ts,tsx,js}' '\.observe\(' -B 2 -A 3 | head -80

Repository: RocketChat/Rocket.Chat

Length of output: 5969

🏁 Script executed:

#!/bin/bash
# Check if Collection.observe is called anywhere (the method that would call _observeChanges)
rg -n -g '*.{ts,tsx,js}' 'collection\.observe|Collection\.observe' -B 2 -A 3

Repository: RocketChat/Rocket.Chat

Length of output: 960

🏁 Script executed:

#!/bin/bash
# Look for await _observeChanges or any async calls to observe
rg -n -g '*.{ts,tsx,js}' 'await.*observe|observeChanges.*await' -B 1 -A 1

Repository: RocketChat/Rocket.Chat

Length of output: 524

The _observeChanges override needs to return a synchronous handle with isReady and isReadyPromise properties.

This async override may break the Meteor contract. The original _observeChanges should return a synchronous handle immediately (not a Promise), with the initial query run asynchronously. The returned handle must include isReady (boolean) and isReadyPromise (Promise) properties—currently only stop() is provided. Additionally, the options parameter only forwards projection/fields but omits sort, limit, and other common query options, which could cause unexpected query behavior.

🤖 Prompt for AI Agents 
In `@apps/meteor/server/services/meteor/userReactivity.ts` around lines 17 - 72,
The override MongoInternals.Connection.prototype._observeChanges must be
synchronous and return a handle immediately (not an async function); change the
async function to a normal function that starts the initial DB query
asynchronously, immediately returns an object with stop(), isReady (initially
false) and isReadyPromise (a Promise resolved when the initial query completes),
and when the async query finishes set isReady=true and resolve isReadyPromise
and invoke callbacks.added as currently implemented; also forward additional
query options (e.g., sort, limit, skip, hint, etc.) from the options param when
calling mongo.rawCollection(...).find(...) so you don't only pass
projection/fields; keep the userCallbacks and serviceConfigCallbacks logic but
ensure stop() can cancel/cleanup even if called before the initial query
resolves.

Copy link
Member Author

@sampaiodiego sampaiodiego Jan 29, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this is completely wrong, where did you get this info from?

Copy link
Contributor

@coderabbitai coderabbitai bot Jan 29, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧩 Analysis chain

@sampaiodiego I apologize for the incorrect analysis. Let me search for the actual Meteor documentation to understand the correct contract for _observeChanges.

🌐 Web query:

Meteor MongoInternals Connection _observeChanges contract

💡 Result:

MongoInternals.Connection#_observeChanges is Meteor’s server-side internal hook that powers Mongo.Cursor#observeChanges(...) / observeChangesAsync(...). While it’s not a public API, the “contract” it must satisfy is effectively the same as the public observeChanges* semantics:

Inputs (conceptually)

  • A cursor/query description (collection + selector + options like sort/limit/skip/fields, etc.).
  • An “ordered” vs “unordered” mode:
    • Unordered observers use added/changed/removed.
    • Ordered observers use addedBefore/movedBefore (and may still use changed/removed).
  • A callbacks object.

Callback contract

Meteor will call (some subset of) these callbacks, with these meanings:

  • added(id, fields) or addedBefore(id, fields, before)
    • fields is the full document minus _id.
    • before is the _id of the doc that follows it in sort order, or null for “at end”.
  • changed(id, fields)
    • fields contains only changed keys; removed keys appear with value undefined.
  • removed(id)
  • movedBefore(id, before) (ordered observers only)
    • before is the following doc’s id, or null for “at end”.
      These definitions are documented for observeChangesAsync, and apply equivalently to the underlying implementation. [1]

Timing / lifecycle contract

  • Initial results are delivered before the call returns: Meteor calls added/addedBefore zero-or-more times to publish the initial result set, then returns the handle. [1]
  • _observeChanges returns a live query handle with at least:
    • stop() — stops observation and releases resources. [1]

Ordering rule (practical requirement)

If the query uses skip/limit (and generally any ordering-sensitive query), Meteor expects an ordered observe (callbacks like addedBefore instead of added), otherwise you’ll hit errors like: “must use ordered observe … with skip or limit”. [2]

If you share what you’re implementing (e.g., a custom observe driver / alternative Mongo driver), I can map the exact callback set you must support for your specific query shapes (ordered vs unordered) and how to translate DB events into these callbacks.

Sources:
[1] https://docs.meteor.com/api/collections (section: Cursor.observeChangesAsync callbacks + initial-add-before-return + handle.stop)
[2] https://stackoverflow.com/questions/36099080/server-function-call-on-document-observe-changes (ordered observe requirement/error)

@sampaiodiego I apologize for the incorrect review comment. Based on the Meteor documentation I found, I can see that I was wrong about the isReady and isReadyPromise properties being required—the standard contract only requires a stop() method on the returned handle.

Could you help me understand what was incorrect about my analysis? I want to learn from this mistake so I can provide better reviews in the future. Specifically:

  1. Is the async implementation appropriate for your use case?
  2. Are the query options (sort, limit, skip) not needed for the specific queries being observed here?
  3. Is there internal Rocket.Chat context about how this observer pattern works that I'm missing?

Thank you for your patience in correcting me.

🧠 Learnings used 
Learnt from: sampaiodiego
Repo: RocketChat/Rocket.Chat PR: 37357
File: apps/meteor/ee/server/startup/federation.ts:39-74
Timestamp: 2025-11-05T20:53:57.761Z
Learning: In Rocket.Chat (apps/meteor/app/settings/server/CachedSettings.ts), the settings.watchMultiple() method immediately invokes its callback with current values if all requested settings exist in the store, then continues watching for subsequent changes. It does not wait for a setting to change before the first invocation.

Learnt from: Dnouv
Repo: RocketChat/Rocket.Chat PR: 37057
File: packages/apps-engine/src/definition/accessors/IUserRead.ts:23-27
Timestamp: 2025-09-25T09:59:26.461Z
Learning: AppUserBridge.getUserRoomIds in apps/meteor/app/apps/server/bridges/users.ts always returns an array of strings by mapping subscription documents to room IDs, never undefined, even when user has no room subscriptions.

Learnt from: tassoevan
Repo: RocketChat/Rocket.Chat PR: 38219
File: packages/core-typings/src/cloud/Announcement.ts:5-6
Timestamp: 2026-01-17T01:51:47.764Z
Learning: In packages/core-typings/src/cloud/Announcement.ts, the AnnouncementSchema.createdBy field intentionally overrides IBannerSchema.createdBy (object with _id and optional username) with a string enum ['cloud', 'system'] to match existing runtime behavior. This is documented as technical debt with a FIXME comment at apps/meteor/app/cloud/server/functions/syncWorkspace/handleCommsSync.ts:53 and should not be flagged as an error until the runtime behavior is corrected.

Learnt from: MartinSchoeler
Repo: RocketChat/Rocket.Chat PR: 37408
File: apps/meteor/client/views/admin/ABAC/useRoomAttributeOptions.tsx:53-69
Timestamp: 2025-11-10T19:06:20.146Z
Learning: In the Rocket.Chat repository, do not provide suggestions or recommendations about code sections marked with TODO comments. The maintainers have already identified these as future work and external reviewers lack the full context about implementation plans and timing.

Copy link
Member Author

@sampaiodiego sampaiodiego Jan 29, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't know where you get the definitions for _observeChanges

Copy link
Contributor

@coderabbitai coderabbitai bot Jan 29, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Search for any Meteor internal documentation or type definitions in the repo
rg -n -g '*.{ts,d.ts,js}' 'MongoInternals\.Connection' -A 10 -B 2

Length of output: 1156

🏁 Script executed:

#!/bin/bash
# Look for any comments or documentation about _observeChanges in the codebase
rg -n -g '*.{ts,js}' '_observeChanges' -B 5 -A 5

Length of output: 2947

🌐 Web query:

Meteor MongoInternals Connection _observeChanges source code implementation

💡 Result:

In Meteor core, MongoInternals.Connection.prototype._observeChanges lives in the server-side mongo package and is implemented (as of the Meteor 3.3.x line) in:

  • packages/mongo/mongo_connection.js (the connection class) (fossies.org)
  • It delegates most of the livequery wiring to packages/mongo/observe_multiplex.ts (the “observe multiplexer” that deduplicates identical live queries) (fossies.org)

How _observeChanges is structured (high level)

  1. Build a cursor description (collection name, selector, options/projection/sort/limit, ordered vs unordered callbacks, etc.).
  2. Look up or create an ObserveMultiplexer keyed by that cursor description, so multiple callers observing the same query share one underlying driver and fan out callbacks.
  3. The ObserveMultiplexer chooses an observe driver:
    • Prefer OplogObserveDriver (oplog tailing) when the query is supported.
    • Fall back to PollingObserveDriver when it’s not. (fossies.org)
  4. The chosen driver runs the initial query (emitting initial added callbacks) and then streams changes via oplog tailing or polling/diff.

The underlying drivers you’ll end up reading

  • Oplog driver: packages/mongo/oplog_observe_driver.js (exports OplogObserveDriver) (fossies.org)
  • Polling fallback: packages/mongo/polling_observe_driver.ts (fossies.org)
  • Background/architecture explanation of the decision and unsupported query shapes: docs/long-form/oplog-observe-driver.md (fossies.org)

If you tell me which Meteor version/branch you’re targeting (e.g., Meteor 3.3.2, devel, etc.), I can point you to the exact function signature and the specific call chain inside that version’s mongo_connection.js / observe_multiplex.ts.

Citations:

@sampaiodiego I apologize for the confusion. I was making inferences based on:

  1. Public Meteor documentation about observeChanges() (the public API)
  2. Web searches that pointed to Meteor's internal source code structure

I don't have access to the actual Meteor internal source code or its type definitions in this repository, so I was incorrectly assuming that the internal _observeChanges method would follow the same patterns as the public API. This was a mistake on my part.

Could you help me understand:

  1. What is the actual contract for MongoInternals.Connection.prototype._observeChanges?
  2. Is the async implementation in this PR the correct approach for replacing the oplog-based reactivity?
  3. Are there any specific behaviors or return values that must be maintained?

I want to learn from this so I can provide more accurate reviews in the future, especially when it comes to internal Meteor APIs that your team is working with.

🧠 Learnings used 
Learnt from: sampaiodiego
Repo: RocketChat/Rocket.Chat PR: 37357
File: apps/meteor/ee/server/startup/federation.ts:39-74
Timestamp: 2025-11-05T20:53:57.761Z
Learning: In Rocket.Chat (apps/meteor/app/settings/server/CachedSettings.ts), the settings.watchMultiple() method immediately invokes its callback with current values if all requested settings exist in the store, then continues watching for subsequent changes. It does not wait for a setting to change before the first invocation.

Learnt from: Dnouv
Repo: RocketChat/Rocket.Chat PR: 37057
File: packages/apps-engine/src/definition/accessors/IUserRead.ts:23-27
Timestamp: 2025-09-25T09:59:26.461Z
Learning: AppUserBridge.getUserRoomIds in apps/meteor/app/apps/server/bridges/users.ts always returns an array of strings by mapping subscription documents to room IDs, never undefined, even when user has no room subscriptions.

Learnt from: tassoevan
Repo: RocketChat/Rocket.Chat PR: 38219
File: packages/core-typings/src/cloud/Announcement.ts:5-6
Timestamp: 2026-01-17T01:51:47.764Z
Learning: In packages/core-typings/src/cloud/Announcement.ts, the AnnouncementSchema.createdBy field intentionally overrides IBannerSchema.createdBy (object with _id and optional username) with a string enum ['cloud', 'system'] to match existing runtime behavior. This is documented as technical debt with a FIXME comment at apps/meteor/app/cloud/server/functions/syncWorkspace/handleCommsSync.ts:53 and should not be flagged as an error until the runtime behavior is corrected.

Learnt from: MartinSchoeler
Repo: RocketChat/Rocket.Chat PR: 37408
File: apps/meteor/client/views/admin/ABAC/useRoomAttributeOptions.tsx:53-69
Timestamp: 2025-11-10T19:06:20.146Z
Learning: In the Rocket.Chat repository, do not provide suggestions or recommendations about code sections marked with TODO comments. The maintainers have already identified these as future work and external reviewers lack the full context about implementation plans and timing.


// Re-implement meteor's reactivity that uses observe to disconnect sessions when the token
// associated was removed
export const processOnChange = (diff: Record<string, any>, id: string): void => {
if (!diff || !('services.resume.loginTokens' in diff)) {
return;
}
const loginTokens: undefined | { hashedToken: string }[] = diff['services.resume.loginTokens'];
const tokens = loginTokens?.map(({ hashedToken }) => hashedToken);

const cbs = userCallbacks.get(id);
if (cbs) {
[...cbs]
.filter(({ hashedToken }) => tokens === undefined || !tokens.includes(hashedToken))
.forEach((item) => {
item.callbacks.removed(id);
cbs.delete(item);
});
}
};
Loading