mirror of
https://github.com/yu-i-i/overleaf-cep.git
synced 2026-11-12 13:13:45 +01:00
Merge pull request #35435 from overleaf/em-library-stop-main-cluster-writes
Stop writes to the main Mongo cluster for library collections GitOrigin-RevId: 39c9ccc2a02e291befcf197fa806ee5aab900a8d
This commit is contained in:
@@ -5,21 +5,9 @@ import MongoUtils from '@overleaf/mongo-utils'
|
||||
import Mongoose from './Mongoose.mjs'
|
||||
import { addConnectionDrainer } from './GracefulShutdown.mjs'
|
||||
import Metrics from '@overleaf/metrics'
|
||||
import logger from '@overleaf/logger'
|
||||
|
||||
/**
|
||||
* @import {
|
||||
* AnyBulkWriteOperation,
|
||||
* BulkWriteOptions,
|
||||
* Collection,
|
||||
* Document,
|
||||
* Filter,
|
||||
* FindOneAndDeleteOptions,
|
||||
* FindOneAndUpdateOptions,
|
||||
* OptionalUnlessRequiredId,
|
||||
* UpdateFilter,
|
||||
* UpdateOptions,
|
||||
* } from 'mongodb-legacy'
|
||||
* @import { Collection, Document } from 'mongodb-legacy'
|
||||
*/
|
||||
|
||||
// Ensure Mongoose is using the same mongodb instance as the mongodb module,
|
||||
@@ -71,109 +59,15 @@ if (auxMongoClient) {
|
||||
}
|
||||
|
||||
/**
|
||||
* @typedef {Pick<
|
||||
* Collection<Document>,
|
||||
* | 'find'
|
||||
* | 'findOne'
|
||||
* | 'aggregate'
|
||||
* | 'countDocuments'
|
||||
* | 'insertMany'
|
||||
* | 'bulkWrite'
|
||||
* | 'updateOne'
|
||||
* | 'findOneAndUpdate'
|
||||
* | 'findOneAndDelete'
|
||||
* | 'deleteMany'
|
||||
* >} DualWriteCollection
|
||||
*/
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* Returns the collection to use for the library collections, which live on
|
||||
* the auxiliary Mongo cluster. Falls back to the main cluster when no
|
||||
* auxiliary client is configured.
|
||||
*
|
||||
* @param {string} name
|
||||
* @returns {DualWriteCollection}
|
||||
* @returns {Collection<Document>}
|
||||
*/
|
||||
function dualWriteCollection(name) {
|
||||
const main = internalDb.collection(name)
|
||||
if (!auxInternalDb) {
|
||||
return main
|
||||
}
|
||||
const auxiliary = auxInternalDb.collection(name)
|
||||
|
||||
const mirror = op =>
|
||||
Promise.resolve()
|
||||
.then(op)
|
||||
.catch(err =>
|
||||
logger.warn(
|
||||
{ err, collection: name },
|
||||
'main cluster mongo write failed'
|
||||
)
|
||||
)
|
||||
|
||||
return {
|
||||
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 auxiliary.insertMany(docs, opts)
|
||||
void mirror(() => main.insertMany(docs, opts))
|
||||
return result
|
||||
},
|
||||
/**
|
||||
* @param {AnyBulkWriteOperation[]} ops
|
||||
* @param {BulkWriteOptions} [opts]
|
||||
*/
|
||||
async bulkWrite(ops, opts) {
|
||||
const result = await auxiliary.bulkWrite(ops, opts)
|
||||
void mirror(() => main.bulkWrite(ops, opts))
|
||||
return result
|
||||
},
|
||||
/**
|
||||
* @param {Filter<Document>} filter
|
||||
* @param {UpdateFilter<Document>} update
|
||||
* @param {UpdateOptions} [opts]
|
||||
*/
|
||||
async updateOne(filter, update, opts) {
|
||||
const result = await auxiliary.updateOne(filter, update, opts)
|
||||
void mirror(() => main.updateOne(filter, update, opts))
|
||||
return result
|
||||
},
|
||||
/**
|
||||
* @param {Filter<Document>} filter
|
||||
* @param {UpdateFilter<Document>} update
|
||||
* @param {FindOneAndUpdateOptions} [opts]
|
||||
*/
|
||||
async findOneAndUpdate(filter, update, opts) {
|
||||
const result = await auxiliary.findOneAndUpdate(filter, update, opts)
|
||||
void mirror(() => main.findOneAndUpdate(filter, update, opts))
|
||||
return result
|
||||
},
|
||||
/**
|
||||
* @param {Filter<Document>} filter
|
||||
* @param {FindOneAndDeleteOptions} [opts]
|
||||
*/
|
||||
async 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 auxiliary.deleteMany(filter)
|
||||
void mirror(() => main.deleteMany(filter))
|
||||
return result
|
||||
},
|
||||
}
|
||||
function libraryCollection(name) {
|
||||
return (auxInternalDb || internalDb).collection(name)
|
||||
}
|
||||
|
||||
export const db = {
|
||||
@@ -194,9 +88,9 @@ export const db = {
|
||||
grouppolicies: internalDb.collection('grouppolicies'),
|
||||
groupAuditLogEntries: internalDb.collection('groupAuditLogEntries'),
|
||||
institutions: internalDb.collection('institutions'),
|
||||
libraryReferences: dualWriteCollection('libraryReferences'),
|
||||
librarySizes: dualWriteCollection('librarySizes'),
|
||||
librarySyncStates: dualWriteCollection('librarySyncStates'),
|
||||
libraryReferences: libraryCollection('libraryReferences'),
|
||||
librarySizes: libraryCollection('librarySizes'),
|
||||
librarySyncStates: libraryCollection('librarySyncStates'),
|
||||
messages: internalDb.collection('messages'),
|
||||
migrations: internalDb.collection('migrations'),
|
||||
notifications: internalDb.collection('notifications'),
|
||||
@@ -258,18 +152,6 @@ export async function getCollectionInternal(name) {
|
||||
return internalDb.collection(name)
|
||||
}
|
||||
|
||||
/**
|
||||
* Direct access to a collection on the auxiliary Mongo cluster, bypassing
|
||||
* the dual-write mirror in `db`. For use by data-migration scripts that need
|
||||
* to write to the auxiliary cluster without also writing to the primary one.
|
||||
*/
|
||||
export async function getAuxCollectionInternal(name) {
|
||||
if (!auxInternalDb) {
|
||||
throw new OError('no auxiliary Mongo cluster configured')
|
||||
}
|
||||
return auxInternalDb.collection(name)
|
||||
}
|
||||
|
||||
export async function waitForDb() {
|
||||
await connectionPromise
|
||||
await auxConnectionPromise
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
import mongodb from 'mongodb-legacy'
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import sinon from 'sinon'
|
||||
import logger from '@overleaf/logger'
|
||||
|
||||
const { ObjectId } = mongodb
|
||||
|
||||
@@ -255,62 +254,25 @@ describe('auxiliary cluster support', function () {
|
||||
expect(clients.map(c => c.url)).to.include.members([MAIN_URL, AUX_URL])
|
||||
})
|
||||
|
||||
it('writes authoritatively to the auxiliary collection, mirrors to the main collection, and returns the auxiliary result', async function () {
|
||||
it('exposes the plain auxiliary-cluster collection, unwrapped', 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: 'aux' })
|
||||
expect(auxCollection.insertMany).to.have.been.calledWith([{ _id: 1 }], {
|
||||
ordered: false,
|
||||
})
|
||||
expect(mainCollection.insertMany).to.have.been.calledWith([{ _id: 1 }], {
|
||||
ordered: false,
|
||||
})
|
||||
expect(db.libraryReferences).to.equal(auxCollection)
|
||||
})
|
||||
|
||||
it('only reads from the auxiliary collection', async function () {
|
||||
it('only reads and writes to the auxiliary collection', async function () {
|
||||
const { db } = await loadModule(AUX_URL)
|
||||
await db.libraryReferences.find({ userId: 'abc' })
|
||||
await db.libraryReferences.insertMany([{ _id: 1 }], { ordered: false })
|
||||
|
||||
const mainCollection = collectionsByUrl[MAIN_URL].get('libraryReferences')
|
||||
const auxCollection = collectionsByUrl[AUX_URL].get('libraryReferences')
|
||||
expect(auxCollection.find).to.have.been.calledWith({
|
||||
userId: 'abc',
|
||||
})
|
||||
expect(mainCollection.find).to.not.have.been.called
|
||||
})
|
||||
|
||||
it('swallows main cluster write failures without affecting the auxiliary result', async function () {
|
||||
const { db } = await loadModule(AUX_URL)
|
||||
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({ ok: true })
|
||||
await vi.waitFor(() => {
|
||||
expect(logger.warn).toHaveBeenCalled()
|
||||
expect(auxCollection.insertMany).to.have.been.calledWith([{ _id: 1 }], {
|
||||
ordered: false,
|
||||
})
|
||||
})
|
||||
|
||||
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')
|
||||
expect(collectionsByUrl[MAIN_URL].has('libraryReferences')).to.be.false
|
||||
})
|
||||
|
||||
it('waits for both the main and auxiliary connections', async function () {
|
||||
|
||||
Reference in New Issue
Block a user