github/meteor
1
1
Fork 0
mirror of https://github.com/meteor/meteor.git synced 2026-05-02 03:01:46 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fix oplog_observe_driver

Browse Source
This commit is contained in:
harryadel
2024-08-24 13:40:28 +03:00
parent fa10502b30
commit 277cae2e1a
1 changed files with 28 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
packages/mongo/oplog_observe_driver.js
View File

@@ -192,25 +192,20 @@ _.extend(OplogObserveDriver.prototype, {
}
}
});
});
}
));
// When Mongo fails over, we need to repoll the query, in case we processed an
// oplog entry that got rolled back.
self._stopHandles.push(self._mongoHandle._onFailover(finishIfNeedToPollQuery(
function () {
self._needToPollQuery();
})));
// Give _observeChanges a chance to add the new ObserveHandle to our
// multiplexer, so that the added calls get streamed.
Meteor.defer(finishIfNeedToPollQuery(function () {
self._runInitialQuery();
}));
};
_.extend(OplogObserveDriver.prototype, {
}
));
// When Mongo fails over, we need to repoll the query, in case we processed an
// oplog entry that got rolled back.
self._addStopHandles(self._mongoHandle._onFailover(finishIfNeedToPollQuery(
function () {
return self._needToPollQuery();
})));
// Give _observeChanges a chance to add the new ObserveHandle to our
// multiplexer, so that the added calls get streamed.
return self._runInitialQuery();
},
_addPublished: function (id, doc) {
var self = this;
Meteor._noYieldsAllowed(function () {
@@ -584,15 +579,17 @@ _.extend(OplogObserveDriver.prototype, {
},
_beSteady: async function () {
var self = this;
Meteor._noYieldsAllowed(function () {
self._registerPhaseChange(PHASE.STEADY);
var writes = self._writesToCommitWhenWeReachSteady;
self._writesToCommitWhenWeReachSteady = [];
self._multiplexer.onFlush(function () {
_.each(writes, function (w) {
w.committed();
});
});
self._registerPhaseChange(PHASE.STEADY);
var writes = self._writesToCommitWhenWeReachSteady || [];
self._writesToCommitWhenWeReachSteady = [];
await self._multiplexer.onFlush(async function () {
try {
for (const w of writes) {
await w.committed();
}
} catch (e) {
console.error("_beSteady error", {writes}, e);
}
});
},
_handleOplogEntryQuerying: function (op) {
@@ -956,18 +953,15 @@ _.extend(OplogObserveDriver.prototype, {
if (self._stopped)
return;
self._stopped = true;
_.each(self._stopHandles, function (handle) {
handle.stop();
});
// Note: we *don't* use multiplexer.onFlush here because this stop
// callback is actually invoked by the multiplexer itself when it has
// determined that there are no handles left. So nothing is actually going
// to get flushed (and it's probably not valid to call methods on the
// dying multiplexer).
_.each(self._writesToCommitWhenWeReachSteady, function (w) {
w.committed(); // maybe yields?
});
for (const w of self._writesToCommitWhenWeReachSteady) {
await w.committed();
}
self._writesToCommitWhenWeReachSteady = null;
// Proactively drop references to potentially big things.