mirror of
https://github.com/yu-i-i/overleaf-cep.git
synced 2026-11-14 06:03:45 +01:00
Merge pull request #35434 from overleaf/em-change-library-mongo-cluster-switch-reads
Switch library collection reads and authoritative writes to the auxiliary Mongo cluster GitOrigin-RevId: 5bbaa30b50c4d98d70d5d310dcd1209837e3f921
This commit is contained in:
@@ -87,18 +87,19 @@ if (auxMongoClient) {
|
||||
*/
|
||||
|
||||
/**
|
||||
* Wraps a collection so that writes are mirrored to the auxiliary Mongo
|
||||
* cluster (best-effort) while the primary cluster remains the source of
|
||||
* truth. Falls back to the plain primary collection when no auxiliary
|
||||
* client is configured.
|
||||
* Wraps a collection so that reads and the authoritative write happen
|
||||
* against the auxiliary Mongo cluster, while writes are additionally
|
||||
* mirrored to the main cluster (best-effort), so that reads could be
|
||||
* switched back to the main cluster if needed. Falls back to the plain
|
||||
* main-cluster collection when no auxiliary client is configured.
|
||||
*
|
||||
* @param {string} name
|
||||
* @returns {DualWriteCollection}
|
||||
*/
|
||||
function dualWriteCollection(name) {
|
||||
const primary = internalDb.collection(name)
|
||||
const main = internalDb.collection(name)
|
||||
if (!auxInternalDb) {
|
||||
return primary
|
||||
return main
|
||||
}
|
||||
const auxiliary = auxInternalDb.collection(name)
|
||||
|
||||
@@ -106,21 +107,24 @@ function dualWriteCollection(name) {
|
||||
Promise.resolve()
|
||||
.then(op)
|
||||
.catch(err =>
|
||||
logger.warn({ err, collection: name }, 'auxiliary mongo write failed')
|
||||
logger.warn(
|
||||
{ err, collection: name },
|
||||
'main cluster mongo write failed'
|
||||
)
|
||||
)
|
||||
|
||||
return {
|
||||
find: (...args) => primary.find(...args),
|
||||
findOne: (...args) => primary.findOne(...args),
|
||||
aggregate: (...args) => primary.aggregate(...args),
|
||||
countDocuments: (...args) => primary.countDocuments(...args),
|
||||
find: (...args) => auxiliary.find(...args),
|
||||
findOne: (...args) => auxiliary.findOne(...args),
|
||||
aggregate: (...args) => auxiliary.aggregate(...args),
|
||||
countDocuments: (...args) => auxiliary.countDocuments(...args),
|
||||
/**
|
||||
* @param {OptionalUnlessRequiredId<Document>[]} docs
|
||||
* @param {BulkWriteOptions} [opts]
|
||||
*/
|
||||
async insertMany(docs, opts) {
|
||||
const result = await primary.insertMany(docs, opts)
|
||||
void mirror(() => auxiliary.insertMany(docs, opts))
|
||||
const result = await auxiliary.insertMany(docs, opts)
|
||||
void mirror(() => main.insertMany(docs, opts))
|
||||
return result
|
||||
},
|
||||
/**
|
||||
@@ -128,8 +132,8 @@ function dualWriteCollection(name) {
|
||||
* @param {BulkWriteOptions} [opts]
|
||||
*/
|
||||
async bulkWrite(ops, opts) {
|
||||
const result = await primary.bulkWrite(ops, opts)
|
||||
void mirror(() => auxiliary.bulkWrite(ops, opts))
|
||||
const result = await auxiliary.bulkWrite(ops, opts)
|
||||
void mirror(() => main.bulkWrite(ops, opts))
|
||||
return result
|
||||
},
|
||||
/**
|
||||
@@ -138,8 +142,8 @@ function dualWriteCollection(name) {
|
||||
* @param {UpdateOptions} [opts]
|
||||
*/
|
||||
async updateOne(filter, update, opts) {
|
||||
const result = await primary.updateOne(filter, update, opts)
|
||||
void mirror(() => auxiliary.updateOne(filter, update, opts))
|
||||
const result = await auxiliary.updateOne(filter, update, opts)
|
||||
void mirror(() => main.updateOne(filter, update, opts))
|
||||
return result
|
||||
},
|
||||
/**
|
||||
@@ -148,8 +152,8 @@ function dualWriteCollection(name) {
|
||||
* @param {FindOneAndUpdateOptions} [opts]
|
||||
*/
|
||||
async findOneAndUpdate(filter, update, opts) {
|
||||
const result = await primary.findOneAndUpdate(filter, update, opts)
|
||||
void mirror(() => auxiliary.findOneAndUpdate(filter, update, opts))
|
||||
const result = await auxiliary.findOneAndUpdate(filter, update, opts)
|
||||
void mirror(() => main.findOneAndUpdate(filter, update, opts))
|
||||
return result
|
||||
},
|
||||
/**
|
||||
@@ -157,16 +161,16 @@ function dualWriteCollection(name) {
|
||||
* @param {FindOneAndDeleteOptions} [opts]
|
||||
*/
|
||||
async findOneAndDelete(filter, opts) {
|
||||
const result = await primary.findOneAndDelete(filter, opts)
|
||||
void mirror(() => auxiliary.findOneAndDelete(filter, opts))
|
||||
const result = await auxiliary.findOneAndDelete(filter, opts)
|
||||
void mirror(() => main.findOneAndDelete(filter, opts))
|
||||
return result
|
||||
},
|
||||
/**
|
||||
* @param {Filter<Document>} filter
|
||||
*/
|
||||
async deleteMany(filter) {
|
||||
const result = await primary.deleteMany(filter)
|
||||
void mirror(() => auxiliary.deleteMany(filter))
|
||||
const result = await auxiliary.deleteMany(filter)
|
||||
void mirror(() => main.deleteMany(filter))
|
||||
return result
|
||||
},
|
||||
}
|
||||
|
||||
@@ -6,7 +6,7 @@ import logger from '@overleaf/logger'
|
||||
const { ObjectId } = mongodb
|
||||
|
||||
const MODULE_PATH = '../../../../app/src/infrastructure/mongodb.mjs'
|
||||
const PRIMARY_URL = 'mongodb://primary'
|
||||
const MAIN_URL = 'mongodb://main'
|
||||
const AUX_URL = 'mongodb://aux'
|
||||
|
||||
function makeFakeCollection() {
|
||||
@@ -14,11 +14,11 @@ function makeFakeCollection() {
|
||||
find: sinon.stub(),
|
||||
findOne: sinon.stub(),
|
||||
aggregate: sinon.stub(),
|
||||
insertMany: sinon.stub().resolves({ from: 'primary' }),
|
||||
bulkWrite: sinon.stub().resolves({ from: 'primary' }),
|
||||
updateOne: sinon.stub().resolves({ from: 'primary' }),
|
||||
findOneAndUpdate: sinon.stub().resolves({ from: 'primary' }),
|
||||
findOneAndDelete: sinon.stub().resolves({ from: 'primary' }),
|
||||
insertMany: sinon.stub().resolves({ ok: true }),
|
||||
bulkWrite: sinon.stub().resolves({ ok: true }),
|
||||
updateOne: sinon.stub().resolves({ ok: true }),
|
||||
findOneAndUpdate: sinon.stub().resolves({ ok: true }),
|
||||
findOneAndDelete: sinon.stub().resolves({ ok: true }),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -176,7 +176,7 @@ describe('auxiliary cluster support', function () {
|
||||
|
||||
beforeEach(function () {
|
||||
collectionsByUrl = {
|
||||
[PRIMARY_URL]: new Map(),
|
||||
[MAIN_URL]: new Map(),
|
||||
[AUX_URL]: new Map(),
|
||||
}
|
||||
clients = []
|
||||
@@ -218,7 +218,7 @@ describe('auxiliary cluster support', function () {
|
||||
vi.doMock('@overleaf/settings', () => ({
|
||||
default: {
|
||||
mongo: {
|
||||
url: PRIMARY_URL,
|
||||
url: MAIN_URL,
|
||||
auxUrl,
|
||||
options: {},
|
||||
hasSecondaries: false,
|
||||
@@ -229,20 +229,19 @@ describe('auxiliary cluster support', function () {
|
||||
}
|
||||
|
||||
describe('when no auxiliary cluster is configured', function () {
|
||||
it('only creates a client for the primary cluster', async function () {
|
||||
it('only creates a client for the main cluster', async function () {
|
||||
await loadModule(undefined)
|
||||
expect(clients).to.have.lengthOf(1)
|
||||
expect(clients[0].url).to.equal(PRIMARY_URL)
|
||||
expect(clients[0].url).to.equal(MAIN_URL)
|
||||
})
|
||||
|
||||
it('exposes the plain primary collection, unwrapped', async function () {
|
||||
it('exposes the plain main-cluster collection, unwrapped', async function () {
|
||||
const { db } = await loadModule(undefined)
|
||||
const primaryCollection =
|
||||
collectionsByUrl[PRIMARY_URL].get('libraryReferences')
|
||||
expect(db.libraryReferences).to.equal(primaryCollection)
|
||||
const mainCollection = collectionsByUrl[MAIN_URL].get('libraryReferences')
|
||||
expect(db.libraryReferences).to.equal(mainCollection)
|
||||
})
|
||||
|
||||
it('waits for the primary connection only', async function () {
|
||||
it('waits for the main cluster connection only', async function () {
|
||||
const { waitForDb } = await loadModule(undefined)
|
||||
await expect(waitForDb()).to.eventually.be.fulfilled
|
||||
expect(clients[0].connect).to.have.been.called
|
||||
@@ -253,58 +252,68 @@ describe('auxiliary cluster support', function () {
|
||||
it('creates a client for both clusters', async function () {
|
||||
await loadModule(AUX_URL)
|
||||
expect(clients).to.have.lengthOf(2)
|
||||
expect(clients.map(c => c.url)).to.include.members([PRIMARY_URL, AUX_URL])
|
||||
expect(clients.map(c => c.url)).to.include.members([MAIN_URL, AUX_URL])
|
||||
})
|
||||
|
||||
it('mirrors writes to the auxiliary collection and returns the primary result', async function () {
|
||||
it('writes authoritatively to the auxiliary collection, mirrors to the main collection, and returns the auxiliary result', async function () {
|
||||
const { db } = await loadModule(AUX_URL)
|
||||
const mainCollection = collectionsByUrl[MAIN_URL].get('libraryReferences')
|
||||
const auxCollection = collectionsByUrl[AUX_URL].get('libraryReferences')
|
||||
mainCollection.insertMany.resolves({ from: 'main' })
|
||||
auxCollection.insertMany.resolves({ from: 'aux' })
|
||||
|
||||
const result = await db.libraryReferences.insertMany([{ _id: 1 }], {
|
||||
ordered: false,
|
||||
})
|
||||
|
||||
expect(result).to.deep.equal({ from: 'primary' })
|
||||
const primaryCollection =
|
||||
collectionsByUrl[PRIMARY_URL].get('libraryReferences')
|
||||
const auxCollection = collectionsByUrl[AUX_URL].get('libraryReferences')
|
||||
expect(primaryCollection.insertMany).to.have.been.calledWith(
|
||||
[{ _id: 1 }],
|
||||
{ ordered: false }
|
||||
)
|
||||
expect(result).to.deep.equal({ from: 'aux' })
|
||||
expect(auxCollection.insertMany).to.have.been.calledWith([{ _id: 1 }], {
|
||||
ordered: false,
|
||||
})
|
||||
expect(mainCollection.insertMany).to.have.been.calledWith([{ _id: 1 }], {
|
||||
ordered: false,
|
||||
})
|
||||
})
|
||||
|
||||
it('only reads from the primary collection', async function () {
|
||||
it('only reads from the auxiliary collection', async function () {
|
||||
const { db } = await loadModule(AUX_URL)
|
||||
await db.libraryReferences.find({ userId: 'abc' })
|
||||
|
||||
const primaryCollection =
|
||||
collectionsByUrl[PRIMARY_URL].get('libraryReferences')
|
||||
const mainCollection = collectionsByUrl[MAIN_URL].get('libraryReferences')
|
||||
const auxCollection = collectionsByUrl[AUX_URL].get('libraryReferences')
|
||||
expect(primaryCollection.find).to.have.been.calledWith({
|
||||
expect(auxCollection.find).to.have.been.calledWith({
|
||||
userId: 'abc',
|
||||
})
|
||||
expect(auxCollection.find).to.not.have.been.called
|
||||
expect(mainCollection.find).to.not.have.been.called
|
||||
})
|
||||
|
||||
it('swallows auxiliary write failures without affecting the primary result', async function () {
|
||||
it('swallows main cluster write failures without affecting the auxiliary result', async function () {
|
||||
const { db } = await loadModule(AUX_URL)
|
||||
const auxCollection = collectionsByUrl[AUX_URL].get('librarySizes')
|
||||
auxCollection.updateOne.rejects(new Error('aux write failed'))
|
||||
const mainCollection = collectionsByUrl[MAIN_URL].get('librarySizes')
|
||||
mainCollection.updateOne.rejects(new Error('main cluster write failed'))
|
||||
|
||||
const result = await db.librarySizes.updateOne(
|
||||
{ _id: 'user1' },
|
||||
{ $set: { size: 1 } }
|
||||
)
|
||||
|
||||
expect(result).to.deep.equal({ from: 'primary' })
|
||||
expect(result).to.deep.equal({ ok: true })
|
||||
await vi.waitFor(() => {
|
||||
expect(logger.warn).toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
|
||||
it('waits for both the primary and auxiliary connections', async function () {
|
||||
it('does not swallow auxiliary write failures, since it is the authoritative write', async function () {
|
||||
const { db } = await loadModule(AUX_URL)
|
||||
const auxCollection = collectionsByUrl[AUX_URL].get('librarySizes')
|
||||
auxCollection.updateOne.rejects(new Error('auxiliary write failed'))
|
||||
|
||||
await expect(
|
||||
db.librarySizes.updateOne({ _id: 'user1' }, { $set: { size: 1 } })
|
||||
).to.eventually.be.rejectedWith('auxiliary write failed')
|
||||
})
|
||||
|
||||
it('waits for both the main and auxiliary connections', async function () {
|
||||
const { waitForDb } = await loadModule(AUX_URL)
|
||||
await expect(waitForDb()).to.eventually.be.fulfilled
|
||||
expect(clients[0].connect).to.have.been.called
|
||||
|
||||
Reference in New Issue
Block a user