Merge pull request #35377 from overleaf/em-change-library-mongo-cluster

Add auxiliary Mongo cluster support for library collections

GitOrigin-RevId: 1cfdbe4bbeb6c95f1158834ae60a4e34bef7bb87

services/web/app/src/infrastructure/mongodb.mjs

@@ -5,6 +5,22 @@ 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'
+ */
 
 // Ensure Mongoose is using the same mongodb instance as the mongodb module,
 // otherwise we will get multiple versions of the ObjectId class. Mongoose
@@ -35,6 +51,127 @@ addConnectionDrainer('mongodb', async () => {
 })
 
 const internalDb = mongoClient.db()
+
+const auxMongoClient = Settings.mongo.auxUrl
+  ? new mongodb.MongoClient(Settings.mongo.auxUrl, Settings.mongo.options)
+  : null
+
+let auxInternalDb = null
+let auxConnectionPromise = Promise.resolve()
+
+if (auxMongoClient) {
+  Metrics.mongodb.monitor(auxMongoClient, 'native-aux')
+
+  addConnectionDrainer('mongodb-aux', async () => {
+    await auxMongoClient.close()
+  })
+
+  auxInternalDb = auxMongoClient.db()
+  auxConnectionPromise = auxMongoClient.connect()
+}
+
+/**
+ * @typedef {Pick<
+ *   Collection<Document>,
+ *   | 'find'
+ *   | 'findOne'
+ *   | 'aggregate'
+ *   | 'countDocuments'
+ *   | 'insertMany'
+ *   | 'bulkWrite'
+ *   | 'updateOne'
+ *   | 'findOneAndUpdate'
+ *   | 'findOneAndDelete'
+ *   | 'deleteMany'
+ * >} DualWriteCollection
+ */
+
+/**
+ * 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.
+ *
+ * @param {string} name
+ * @returns {DualWriteCollection}
+ */
+function dualWriteCollection(name) {
+  const primary = internalDb.collection(name)
+  if (!auxInternalDb) {
+    return primary
+  }
+  const auxiliary = auxInternalDb.collection(name)
+
+  const mirror = op =>
+    Promise.resolve()
+      .then(op)
+      .catch(err =>
+        logger.warn({ err, collection: name }, 'auxiliary mongo write failed')
+      )
+
+  return {
+    find: (...args) => primary.find(...args),
+    findOne: (...args) => primary.findOne(...args),
+    aggregate: (...args) => primary.aggregate(...args),
+    countDocuments: (...args) => primary.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))
+      return result
+    },
+    /**
+     * @param {AnyBulkWriteOperation[]} ops
+     * @param {BulkWriteOptions} [opts]
+     */
+    async bulkWrite(ops, opts) {
+      const result = await primary.bulkWrite(ops, opts)
+      void mirror(() => auxiliary.bulkWrite(ops, opts))
+      return result
+    },
+    /**
+     * @param {Filter<Document>} filter
+     * @param {UpdateFilter<Document>} update
+     * @param {UpdateOptions} [opts]
+     */
+    async updateOne(filter, update, opts) {
+      const result = await primary.updateOne(filter, update, opts)
+      void mirror(() => auxiliary.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 primary.findOneAndUpdate(filter, update, opts)
+      void mirror(() => auxiliary.findOneAndUpdate(filter, update, opts))
+      return result
+    },
+    /**
+     * @param {Filter<Document>} filter
+     * @param {FindOneAndDeleteOptions} [opts]
+     */
+    async findOneAndDelete(filter, opts) {
+      const result = await primary.findOneAndDelete(filter, opts)
+      void mirror(() => auxiliary.findOneAndDelete(filter, opts))
+      return result
+    },
+    /**
+     * @param {Filter<Document>} filter
+     */
+    async deleteMany(filter) {
+      const result = await primary.deleteMany(filter)
+      void mirror(() => auxiliary.deleteMany(filter))
+      return result
+    },
+  }
+}
+
 export const db = {
   contacts: internalDb.collection('contacts'),
   deletedProjects: internalDb.collection('deletedProjects'),
@@ -53,9 +190,9 @@ export const db = {
   grouppolicies: internalDb.collection('grouppolicies'),
   groupAuditLogEntries: internalDb.collection('groupAuditLogEntries'),
   institutions: internalDb.collection('institutions'),
-  libraryReferences: internalDb.collection('libraryReferences'),
-  librarySizes: internalDb.collection('librarySizes'),
-  librarySyncStates: internalDb.collection('librarySyncStates'),
+  libraryReferences: dualWriteCollection('libraryReferences'),
+  librarySizes: dualWriteCollection('librarySizes'),
+  librarySyncStates: dualWriteCollection('librarySyncStates'),
   messages: internalDb.collection('messages'),
   migrations: internalDb.collection('migrations'),
   notifications: internalDb.collection('notifications'),
@@ -119,6 +256,7 @@ export async function getCollectionInternal(name) {
 
 export async function waitForDb() {
   await connectionPromise
+  await auxConnectionPromise
 }
 
 export default {

services/web/config/settings.defaults.js

@@ -148,6 +148,7 @@ module.exports = {
       process.env.MONGO_URL ||
       `mongodb://${process.env.MONGO_HOST || '127.0.0.1'}/sharelatex`,
     hasSecondaries: process.env.MONGO_HAS_SECONDARIES === 'true',
+    auxUrl: process.env.MONGO_AUX_CONNECTION_STRING,
   },
 
   redis: {

services/web/test/unit/src/infrastructure/Mongodb.test.mjs

@@ -1,9 +1,27 @@
 import mongodb from 'mongodb-legacy'
-import { describe, expect, it } from 'vitest'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
 import sinon from 'sinon'
+import logger from '@overleaf/logger'
 
 const { ObjectId } = mongodb
 
+const MODULE_PATH = '../../../../app/src/infrastructure/mongodb.mjs'
+const PRIMARY_URL = 'mongodb://primary'
+const AUX_URL = 'mongodb://aux'
+
+function makeFakeCollection() {
+  return {
+    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' }),
+  }
+}
+
 describe('Mongo ObjectId comparison', function () {
   const ObjectId1 = new ObjectId('111111111111111111111111')
   const ObjectId2 = new ObjectId('65d8607441ef95170e89bd2a')
@@ -152,3 +170,145 @@ describe('Mongo ObjectId comparison', function () {
     })
   })
 })
+
+describe('auxiliary cluster support', function () {
+  let collectionsByUrl, clients
+
+  beforeEach(function () {
+    collectionsByUrl = {
+      [PRIMARY_URL]: new Map(),
+      [AUX_URL]: new Map(),
+    }
+    clients = []
+
+    class FakeMongoClient {
+      constructor(url, options) {
+        this.url = url
+        this.options = options
+        this.connect = sinon.stub().resolves(this)
+        this.close = sinon.stub().resolves()
+        clients.push(this)
+      }
+
+      on() {}
+
+      db() {
+        const store = collectionsByUrl[this.url]
+        return {
+          collection(name) {
+            if (!store.has(name)) {
+              store.set(name, makeFakeCollection())
+            }
+            return store.get(name)
+          },
+        }
+      }
+    }
+
+    vi.doMock('mongodb-legacy', () => ({
+      default: { ...mongodb, MongoClient: FakeMongoClient },
+    }))
+
+    vi.doMock('../../../../app/src/infrastructure/Mongoose.mjs', () => ({
+      default: { mongo: { ObjectId } },
+    }))
+  })
+
+  async function loadModule(auxUrl) {
+    vi.doMock('@overleaf/settings', () => ({
+      default: {
+        mongo: {
+          url: PRIMARY_URL,
+          auxUrl,
+          options: {},
+          hasSecondaries: false,
+        },
+      },
+    }))
+    return import(MODULE_PATH)
+  }
+
+  describe('when no auxiliary cluster is configured', function () {
+    it('only creates a client for the primary cluster', async function () {
+      await loadModule(undefined)
+      expect(clients).to.have.lengthOf(1)
+      expect(clients[0].url).to.equal(PRIMARY_URL)
+    })
+
+    it('exposes the plain primary collection, unwrapped', async function () {
+      const { db } = await loadModule(undefined)
+      const primaryCollection =
+        collectionsByUrl[PRIMARY_URL].get('libraryReferences')
+      expect(db.libraryReferences).to.equal(primaryCollection)
+    })
+
+    it('waits for the primary connection only', async function () {
+      const { waitForDb } = await loadModule(undefined)
+      await expect(waitForDb()).to.eventually.be.fulfilled
+      expect(clients[0].connect).to.have.been.called
+    })
+  })
+
+  describe('when an auxiliary cluster is configured', 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])
+    })
+
+    it('mirrors writes to the auxiliary collection and returns the primary result', async function () {
+      const { db } = await loadModule(AUX_URL)
+      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(auxCollection.insertMany).to.have.been.calledWith([{ _id: 1 }], {
+        ordered: false,
+      })
+    })
+
+    it('only reads from the primary collection', async function () {
+      const { db } = await loadModule(AUX_URL)
+      await db.libraryReferences.find({ userId: 'abc' })
+
+      const primaryCollection =
+        collectionsByUrl[PRIMARY_URL].get('libraryReferences')
+      const auxCollection = collectionsByUrl[AUX_URL].get('libraryReferences')
+      expect(primaryCollection.find).to.have.been.calledWith({
+        userId: 'abc',
+      })
+      expect(auxCollection.find).to.not.have.been.called
+    })
+
+    it('swallows auxiliary write failures without affecting the primary 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 result = await db.librarySizes.updateOne(
+        { _id: 'user1' },
+        { $set: { size: 1 } }
+      )
+
+      expect(result).to.deep.equal({ from: 'primary' })
+      await vi.waitFor(() => {
+        expect(logger.warn).toHaveBeenCalled()
+      })
+    })
+
+    it('waits for both the primary 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
+      expect(clients[1].connect).to.have.been.called
+    })
+  })
+})

tools/migrations/20260420120000_create_libraryReferences_userId_updatedAt_index.mjs

@@ -1,6 +1,6 @@
 import Helpers from './lib/helpers.mjs'
 
-const tags = ['saas']
+const tags = ['saas', 'auxiliary']
 
 const indexes = [
   {

tools/migrations/20260430120000_update_libraryReferences_userId_updatedAt_index.mjs

@@ -1,6 +1,6 @@
 import Helpers from './lib/helpers.mjs'
 
-const tags = ['saas']
+const tags = ['saas', 'auxiliary']
 
 const oldIndex = {
   key: { userId: 1, updatedAt: 1 },

tools/migrations/20260507120000_create_libraryReferences_search_fields.mjs

@@ -1,6 +1,6 @@
 import Helpers from './lib/helpers.mjs'
 
-const tags = ['saas']
+const tags = ['saas', 'auxiliary']
 
 const indexes = [
   {

tools/migrations/20260603120000_recreate_libraryReferences_search_indexes_without_collation.mjs

@@ -1,6 +1,6 @@
 import Helpers from './lib/helpers.mjs'
 
-const tags = ['saas']
+const tags = ['saas', 'auxiliary']
 
 const oldIndexes = [
   {

tools/migrations/20260604174200_zotero_libraryReferences_sources.mjs

@@ -1,7 +1,7 @@
 import { batchedUpdate } from '@overleaf/mongo-utils/batchedUpdate.js'
 import Helpers from './lib/helpers.mjs'
 
-const tags = ['saas']
+const tags = ['saas', 'auxiliary']
 
 const indexes = [
   {

tools/migrations/20260609120000_create_librarySyncStates_collection.mjs

@@ -1,7 +1,6 @@
 import Helpers from './lib/helpers.mjs'
-import { getCollectionInternal } from './lib/mongodb.mjs'
 
-const tags = ['saas']
+const tags = ['saas', 'auxiliary']
 
 const indexes = [
   {
@@ -11,14 +10,14 @@ const indexes = [
   },
 ]
 
-const migrate = async () => {
-  const librarySyncStates = await getCollectionInternal('librarySyncStates')
-  await Helpers.addIndexesToCollection(librarySyncStates, indexes)
+const migrate = async client => {
+  const { db } = client
+  await Helpers.addIndexesToCollection(db.librarySyncStates, indexes)
 }
 
-const rollback = async () => {
-  const librarySyncStates = await getCollectionInternal('librarySyncStates')
-  await Helpers.dropIndexesFromCollection(librarySyncStates, indexes)
+const rollback = async client => {
+  const { db } = client
+  await Helpers.dropIndexesFromCollection(db.librarySyncStates, indexes)
 }
 
 export default {

tools/migrations/20260612120000_replace_libraryReferences_searchTokens_index.mjs

@@ -1,6 +1,6 @@
 import Helpers from './lib/helpers.mjs'
 
-const tags = ['saas']
+const tags = ['saas', 'auxiliary']
 
 const newIndexes = [
   {

tools/migrations/20260619120000_create_libraryReferences_match_indexes.mjs

@@ -1,6 +1,6 @@
 import Helpers from './lib/helpers.mjs'
 
-const tags = ['saas']
+const tags = ['saas', 'auxiliary']
 
 // Partial indexes so docs without a given match token are not indexed. They
 // back the batch duplicate-detection query in findMatchingReferences.

tools/migrations/20260708120000_drop_libraryReferences_searchValue_index.mjs

@@ -1,6 +1,6 @@
 import Helpers from './lib/helpers.mjs'
 
-const tags = ['saas']
+const tags = ['saas', 'auxiliary']
 
 const oldIndexes = [
   {

tools/migrations/config/settings.defaults.js

@@ -16,6 +16,7 @@ module.exports = {
       process.env.MONGO_CONNECTION_STRING ||
       process.env.MONGO_URL ||
       `mongodb://${process.env.MONGO_HOST || '127.0.0.1'}/sharelatex`,
+    auxUrl: process.env.MONGO_AUX_CONNECTION_STRING,
     hasSecondaries: process.env.MONGO_HAS_SECONDARIES === 'true',
   },
 }

tools/migrations/lib/adapter.mjs

@@ -1,10 +1,19 @@
 import Path from 'node:path'
-import { db } from './mongodb.mjs'
+import { db, auxInternalDb } from './mongodb.mjs'
 import { fileURLToPath } from 'node:url'
 
 const __filename = fileURLToPath(import.meta.url)
 const __dirname = Path.dirname(__filename)
 
+// Migrations tagged 'auxiliary' track their execution history on the
+// auxiliary Mongo cluster (when one is configured) instead of the primary
+// one, so a migration already marked executed against the primary cluster
+// still runs once the auxiliary cluster is provisioned. When no auxiliary
+// cluster is configured, everything falls back to the primary cluster.
+const auxMigrations = auxInternalDb
+  ? auxInternalDb.collection('migrations')
+  : null
+
 class Adapter {
   constructor(params) {
     if (
@@ -16,6 +25,7 @@ class Adapter {
       process.exit(1)
     }
     this.params = params || {}
+    this.migrationTagsByName = new Map()
   }
 
   getTemplatePath() {
@@ -30,22 +40,75 @@ class Adapter {
     return Promise.resolve()
   }
 
+  async _getMigrationTags(name) {
+    if (!this.migrationTagsByName.has(name)) {
+      const { dir, migrationExtension } = this.params
+      const migrationPath = Path.resolve(dir, `${name}.${migrationExtension}`)
+      let tags = []
+      try {
+        const { default: migration } = await import(migrationPath)
+        tags = migration.tags || []
+      } catch (err) {
+        if (err.code !== 'ERR_MODULE_NOT_FOUND') {
+          throw err
+        }
+        // The migration file no longer exists on disk (e.g. it was removed
+        // long after it ran everywhere). We can't know if it was tagged
+        // 'auxiliary', so it stays tracked as a primary-only migration.
+      }
+      this.migrationTagsByName.set(name, tags)
+    }
+    return this.migrationTagsByName.get(name)
+  }
+
+  async _migrationsCollectionFor(name) {
+    if (!auxMigrations) {
+      return db.migrations
+    }
+    const tags = await this._getMigrationTags(name)
+    return tags.includes('auxiliary') ? auxMigrations : db.migrations
+  }
+
   async getExecutedMigrationNames() {
-    const migrations = await db.migrations
-      .find({}, { sort: { migratedAt: 1 }, projection: { name: 1 } })
-      .toArray()
-    return migrations.map(migration => migration.name)
+    const primaryNames = (
+      await db.migrations.find({}, { projection: { name: 1 } }).toArray()
+    ).map(migration => migration.name)
+
+    if (!auxMigrations) {
+      return primaryNames.sort()
+    }
+
+    // Once an auxiliary cluster is configured, an 'auxiliary'-tagged
+    // migration counts as executed only if it ran against the auxiliary
+    // cluster — a stale primary-side record from before the auxiliary
+    // cluster existed is ignored, so the migration runs again on the
+    // auxiliary cluster.
+    const nonAuxPrimaryNames = []
+    for (const name of primaryNames) {
+      const tags = await this._getMigrationTags(name)
+      if (!tags.includes('auxiliary')) {
+        nonAuxPrimaryNames.push(name)
+      }
+    }
+
+    const auxNames = (
+      await auxMigrations.find({}, { projection: { name: 1 } }).toArray()
+    ).map(migration => migration.name)
+
+    return [...nonAuxPrimaryNames, ...auxNames].sort()
   }
 
   async markExecuted(name) {
-    return await db.migrations.insertOne({
+    const collection = await this._migrationsCollectionFor(name)
+    return await collection.insertOne({
       name,
       migratedAt: new Date(),
     })
   }
 
   async unmarkExecuted(name) {
-    return await db.migrations.deleteOne({
+    const collection = await this._migrationsCollectionFor(name)
+    return await collection.deleteOne({
       name,
     })
   }

tools/migrations/lib/mongodb.mjs

@@ -11,6 +11,18 @@ export const READ_PREFERENCE_SECONDARY = Settings.mongo.hasSecondaries
 const mongoClient = new MongoClient(Settings.mongo.url, Settings.mongo.options)
 
 const internalDb = mongoClient.db()
+
+const auxMongoClient = Settings.mongo.auxUrl
+  ? new MongoClient(Settings.mongo.auxUrl, Settings.mongo.options)
+  : null
+
+export const auxInternalDb = auxMongoClient ? auxMongoClient.db() : null
+const auxConnectionPromise = auxMongoClient
+  ? auxMongoClient.connect()
+  : Promise.resolve()
+
+const libraryDb = auxInternalDb || internalDb
+
 export const db = {
   contacts: internalDb.collection('contacts'),
   deletedProjects: internalDb.collection('deletedProjects'),
@@ -29,7 +41,9 @@ export const db = {
   grouppolicies: internalDb.collection('grouppolicies'),
   groupAuditLogEntries: internalDb.collection('groupAuditLogEntries'),
   institutions: internalDb.collection('institutions'),
-  libraryReferences: internalDb.collection('libraryReferences'),
+  libraryReferences: libraryDb.collection('libraryReferences'),
+  librarySizes: libraryDb.collection('librarySizes'),
+  librarySyncStates: libraryDb.collection('librarySyncStates'),
   messages: internalDb.collection('messages'),
   migrations: internalDb.collection('migrations'),
   notifications: internalDb.collection('notifications'),
@@ -83,6 +97,7 @@ export async function getCollectionInternal(name) {
 
 export async function waitForDb() {
   await connectionPromise
+  await auxConnectionPromise
 }
 
 const mongodb = {