From 5f2acceae8ad6ab9b3374fb92f2d6ea62947aefd Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Fri, 7 Aug 2026 15:53:03 +0200 Subject: [PATCH 1/3] Prettier reformat files touched by this PR Issue: ARSN-622 --- index.ts | 41 ++++++++++++++--------------------------- 1 file changed, 14 insertions(+), 27 deletions(-) diff --git a/index.ts b/index.ts index 621fef7e6..179d11730 100644 --- a/index.ts +++ b/index.ts @@ -3,7 +3,8 @@ import evaluatePrincipal from './lib/policyEvaluator/principal'; import RequestContext, { actionNeedQuotaCheck, actionNeedQuotaCheckCopy, - actionWithDataDeletion } from './lib/policyEvaluator/RequestContext'; + actionWithDataDeletion, +} from './lib/policyEvaluator/RequestContext'; import * as requestUtils from './lib/policyEvaluator/requestUtils'; import * as actionMaps from './lib/policyEvaluator/utils/actionMaps'; import { validateUserPolicy } from './lib/policy/policyValidator'; @@ -106,33 +107,22 @@ export const storage = { metadata: { MetadataWrapper: require('./lib/storage/metadata/MetadataWrapper'), bucketclient: { - BucketClientInterface: - require('./lib/storage/metadata/bucketclient/' + - 'BucketClientInterface'), - LogConsumer: - require('./lib/storage/metadata/bucketclient/LogConsumer'), + BucketClientInterface: require('./lib/storage/metadata/bucketclient/' + 'BucketClientInterface'), + LogConsumer: require('./lib/storage/metadata/bucketclient/LogConsumer'), }, file: { - BucketFileInterface: - require('./lib/storage/metadata/file/BucketFileInterface'), - MetadataFileServer: - require('./lib/storage/metadata/file/MetadataFileServer'), - MetadataFileClient: - require('./lib/storage/metadata/file/MetadataFileClient'), + BucketFileInterface: require('./lib/storage/metadata/file/BucketFileInterface'), + MetadataFileServer: require('./lib/storage/metadata/file/MetadataFileServer'), + MetadataFileClient: require('./lib/storage/metadata/file/MetadataFileClient'), }, inMemory: { - metastore: - require('./lib/storage/metadata/in_memory/metastore'), + metastore: require('./lib/storage/metadata/in_memory/metastore'), metadata: require('./lib/storage/metadata/in_memory/metadata'), - bucketUtilities: - require('./lib/storage/metadata/in_memory/bucket_utilities'), + bucketUtilities: require('./lib/storage/metadata/in_memory/bucket_utilities'), }, mongoclient: { - MongoClientInterface: - require('./lib/storage/metadata/mongoclient/' + - 'MongoClientInterface'), - LogConsumer: - require('./lib/storage/metadata/mongoclient/LogConsumer'), + MongoClientInterface: require('./lib/storage/metadata/mongoclient/' + 'MongoClientInterface'), + LogConsumer: require('./lib/storage/metadata/mongoclient/LogConsumer'), }, proxy: { Server: require('./lib/storage/metadata/proxy/Server'), @@ -140,14 +130,11 @@ export const storage = { }, data: { DataWrapper: require('./lib/storage/data/DataWrapper'), - MultipleBackendGateway: - require('./lib/storage/data/MultipleBackendGateway'), + MultipleBackendGateway: require('./lib/storage/data/MultipleBackendGateway'), parseLC: require('./lib/storage/data/LocationConstraintParser'), file: { - DataFileStore: - require('./lib/storage/data/file/DataFileStore'), - DataFileInterface: - require('./lib/storage/data/file/DataFileInterface'), + DataFileStore: require('./lib/storage/data/file/DataFileStore'), + DataFileInterface: require('./lib/storage/data/file/DataFileInterface'), }, external: { AwsClient: require('./lib/storage/data/external/AwsClient'), From 5e14780f34aed998f1c7c7aeed78c608e5a6005d Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Fri, 7 Aug 2026 15:53:31 +0200 Subject: [PATCH 2/3] Drop the dead MongoDB oplog-tailing code from mongoclient The mongoclient LogConsumer (and its ListRecordStream) tailed MongoDB's internal replica-set oplog (local.oplog.rs). It has been broken on MongoDB >= 7.1 since the oplog "h" field it uses as a cursor was removed (see ARSN-545), and its only consumer, backbeat's MongoLogReader (logSource "mongo"), was removed in BB-836. MongoReadStream (readStream.js) remains: it is used by MongoClientInterface for listings and is unrelated to the oplog. Issue: ARSN-622 --- index.ts | 1 - .../metadata/mongoclient/ListRecordStream.js | 197 -------- .../metadata/mongoclient/LogConsumer.js | 155 ------ .../mongoclient/ListRecordStream.spec.js | 440 ----------------- .../metadata/mongoclient/LogConsumer.spec.js | 453 ------------------ 5 files changed, 1246 deletions(-) delete mode 100644 lib/storage/metadata/mongoclient/ListRecordStream.js delete mode 100644 lib/storage/metadata/mongoclient/LogConsumer.js delete mode 100644 tests/unit/storage/metadata/mongoclient/ListRecordStream.spec.js delete mode 100644 tests/unit/storage/metadata/mongoclient/LogConsumer.spec.js diff --git a/index.ts b/index.ts index 179d11730..6ae4db40f 100644 --- a/index.ts +++ b/index.ts @@ -122,7 +122,6 @@ export const storage = { }, mongoclient: { MongoClientInterface: require('./lib/storage/metadata/mongoclient/' + 'MongoClientInterface'), - LogConsumer: require('./lib/storage/metadata/mongoclient/LogConsumer'), }, proxy: { Server: require('./lib/storage/metadata/proxy/Server'), diff --git a/lib/storage/metadata/mongoclient/ListRecordStream.js b/lib/storage/metadata/mongoclient/ListRecordStream.js deleted file mode 100644 index cb8f4a766..000000000 --- a/lib/storage/metadata/mongoclient/ListRecordStream.js +++ /dev/null @@ -1,197 +0,0 @@ -const stream = require('stream'); -const unescape = require('./utils').unescape; - -/** - * @class ListRecordStream - * @classdesc Filter and stream records returned from a mongodb query - * cursor - */ -class ListRecordStream extends stream.Readable { - /** - * @constructor - * @param {mongodb.Cursor} mongoCursor - cursor returned by a - * mongodb query to the oplog (see - * http://mongodb.github.io/node-mongodb-native/2.0/api/Cursor.html) - * @param {werelogs.Logger} logger - logger object - * @param {string} lastSavedID - unique ID that has been persisted - * of the most recently processed entry in the oplog - * @param {string} latestOplogID - unique ID of the most recently - * added entry in the oplog - */ - constructor(mongoCursor, logger, lastSavedID, latestOplogID) { - super({ objectMode: true }); - this._cursor = mongoCursor; - this._logger = logger; - this._lastSavedID = lastSavedID; - this._latestOplogID = latestOplogID; - this._lastConsumedID = null; - // this._unpublishedListing is true once we pass the oplog - // record that has the same uniqID 'h' than last saved. If we - // don't find it (e.g. log rolled over before populator could - // process its oldest entries), we will restart from the - // latest record of the oplog. - this._unpublishedListing = false; - // cf. this.getSkipCount() - this._skipCount = 0; - } - - async _read() { - // MongoDB cursors provide a stream interface. We choose not - // to use it though because errors may not be emitted by the - // stream when there is an issue with the connection to - // MongoDB (especially when pause()/resume() are used). - // - // Instead we use the async cursor.next() call directly to - // fetch records one at a time, errors are then forwarded in - // the callback. - try { - const item = await this._cursor.next(); - if (this._processItem(item)) { - return process.nextTick(this._read.bind(this)); - } - // wait until _read() gets called again - return undefined; - } catch (err) { - this._logger.error('mongodb cursor error', { - method: 'mongoclient.ListRecordStream._read()', - error: err.message, - }); - this.emit('error', err); - return undefined; - } - } - - _processItem(itemObj) { - // always update to most recent uniqID - this._lastConsumedID = itemObj.h.toString(); - - // only push to stream unpublished objects - if (!this._lastSavedID) { - // process from the first entry - this._unpublishedListing = true; - } else if (!this._unpublishedListing) { - // When an oplog with a unique ID that is stored in the - // log offset is found, all oplogs AFTER this is unpublished. - if (this._lastSavedID === this._lastConsumedID) { - this._unpublishedListing = true; - } else if (this._latestOplogID === this._lastConsumedID) { - this._logger.warn( - 'did not encounter the last saved offset in oplog, ' + - 'resuming processing right after the latest record ' + - 'to date; some entries may have been skipped', { - lastSavedID: this._lastSavedID, - latestRecordID: this._latestOplogID, - }); - this._unpublishedListing = true; - } - ++this._skipCount; - return true; // read next record - } - - const dbName = itemObj.ns.slice(itemObj.ns.indexOf('.') + 1); - let entry; - if (itemObj.op === 'i' && - itemObj.o && itemObj.o._id) { - const value = itemObj.o.value; - if (value && value.tags) { - value.tags = unescape(value.tags); - } - entry = { - type: 'put', - key: itemObj.o._id, - // value is given as-is for inserts - value: JSON.stringify(value), - }; - } else if (itemObj.op === 'u' && - itemObj.o && itemObj.o2 && itemObj.o2._id) { - const value = (itemObj.o.$set ? itemObj.o.$set : itemObj.o).value; - if (value && value.tags) { - value.tags = unescape(value.tags); - } - // updates overwrite the whole metadata, - // so they are considered as puts - let type = 'put'; - // When the object metadata contain the "deleted" - // flag, it means that the operation is the update - // we perform before the deletion of an object. We - // perform the update to keep all the metadata in the - // oplog. This update is what will be used by backbeat - // as the delete operation so we put the type of operation - // for this event to a delete. - // Backbeat still receives the actual delete operations - // but they are ignored as they don't contain any metadata. - // The delete operations are kept in case we want to listen - // to delete events comming from special collections other - // than "bucket" collections. - if (value && value.deleted) { - type = 'delete'; - } - entry = { - type, - key: itemObj.o2._id, - // updated value may be either stored directly in 'o' - // attribute or in '$set' attribute (supposedly when - // the object pre-exists it will be in '$set') - value: JSON.stringify(value), - }; - } else if (itemObj.op === 'd' && - itemObj.o && itemObj.o._id) { - entry = { - type: 'delete', - key: itemObj.o._id, - // deletion yields no value - }; - } else { - // skip other entry types as we don't need them for now - // ('c', ...?) - ++this._skipCount; - return true; // read next record - } - const streamObject = { - timestamp: new Date((itemObj.ts ? - itemObj.ts.getHighBits() * 1000 : 0)), - db: dbName, - entries: [entry], - }; - // push object to the stream, then return false to wait until - // _read() is called again (because we are in an asynchronous - // context already) - this.push(streamObject); - return false; - } - - /** - * Get an opaque JSON blob containing the latest consumed offset - * from MongoDB oplog. - * - * @return {string} opaque JSON blob - */ - getOffset() { - return JSON.stringify({ - uniqID: this._lastConsumedID, - }); - } - - /** - * Get the number of entries that have been read and skipped from - * MongoDB oplog since the ListRecordStream instance was created. - * - * @return {integer} number of skipped entries - */ - getSkipCount() { - return this._skipCount; - } - - /** - * Get whether the stream reached yet-unpublished records - * (i.e. after we reached either the saved unique ID, or the tip - * of the oplog) - * - * @return {boolean} true if we are now returning unpublished records - */ - reachedUnpublishedListing() { - return this._unpublishedListing; - } -} - -module.exports = ListRecordStream; diff --git a/lib/storage/metadata/mongoclient/LogConsumer.js b/lib/storage/metadata/mongoclient/LogConsumer.js deleted file mode 100644 index 68471bad0..000000000 --- a/lib/storage/metadata/mongoclient/LogConsumer.js +++ /dev/null @@ -1,155 +0,0 @@ -'use strict'; // eslint-disable-line - -const { MongoClient } = require('mongodb'); -const ListRecordStream = require('./ListRecordStream'); -const MongoUtils = require('./utils'); -const { errorInstances } = require('../../../errors'); - -/** - * @class - * @classdesc Class to consume mongo oplog - */ -class LogConsumer { - /** - * @constructor - * - * @param {object} mongoConfig - object with the mongo configuration - * @param {string} logger - logger - */ - constructor(mongoConfig, logger) { - const { authCredentials, replicaSetHosts, replicaSet, database, readPreference } = mongoConfig; - const cred = MongoUtils.credPrefix(authCredentials); - this._mongoUrl = `mongodb://${cred}${replicaSetHosts}/`; - this._replicaSet = replicaSet; - this._readPreference = readPreference; - this._logger = logger; - this._oplogNsRegExp = new RegExp(`^${database}\\.`); - // oplog collection - this._coll = null; - this._client = null; - } - - /** - * Connect to MongoClient using Mongo node module to access database and - * database oplogs (operation logs) - * - * @param {function} done - callback function, called with an error object - * or null and an object as 2nd parameter - * @return {undefined} - */ - connectMongo(done) { - const client = new MongoClient(this._mongoUrl, { - replicaSet: this._replicaSet, - readPreference: this._readPreference, - }); - - client.connect().then(client => { - this._logger.info('connected to mongodb'); - this._client = client; - // 'local' is the database where MongoDB has oplog.rs capped collection - const db = client.db('local', { - ignoreUndefined: true, - }); - this._coll = db.collection('oplog.rs'); - return done(); - }) - .catch(err => { - this._logger.error('Unable to connect to MongoDB', { error: err }); - return done(err); - }); - } - - - /** - * Open a tailable cursor to mongo oplog and retrieve a stream of - * records to read - * - * @param {Object} [params] - params object - * @param {String} [params.startSeq] - fetch starting from this - * opaque offset returned previously by mongo ListRecordStream - * in an 'info' event - * @param {function} cb - callback function, called with an error - * object or null and an object as 2nd parameter - * - * @return {undefined} - */ - readRecords(params, cb) { - let startSeq = {}; - if (params.startSeq) { - try { - // parse the opaque JSON string passed through from a - // previous 'info' event - startSeq = JSON.parse(params.startSeq); - } catch (err) { - this._logger.error('malformed startSeq', { - startSeq: params.startSeq, - }); - // start over if malformed - } - } - this._readLatestOplogID((err, latestOplogID) => { - if (err) { - return cb(err); - } - const cursor = this._coll.find({ - ns: this._oplogNsRegExp, - }, { - tailable: true, - awaitData: true, - noCursorTimeout: true, - numberOfRetries: Number.MAX_VALUE, - }); - const recordStream = new ListRecordStream( - cursor, this._logger, startSeq.uniqID, latestOplogID); - return cb(null, { log: recordStream, tailable: true }); - }); - } - - _readLatestOplogID(cb) { - this._coll.find({ - ns: this._oplogNsRegExp, - }, { - ts: 1, - }).sort({ - $natural: -1, - }).limit(1).toArray() - .then(data => { - if (!data || data.length === 0) { - return cb(new Error('no oplog entry found')); - } - const latestOplogID = data[0].h.toString(); - this._logger.debug('latest oplog ID read', { latestOplogID }); - return cb(null, latestOplogID); - }) - .catch(err => cb(err)); - } - - /** - * Close the connection to MongoDB - * - * @param {function} cb - callback function - * @return {undefined} - */ - close(cb) { - if (this._client) { - this._client.close() - .then(() => { - this._logger.info('MongoDB connection closed'); - this._client = null; - this._coll = null; - return cb(); - }) - .catch(err => { - this._logger.error('Error closing MongoDB connection', { error: err }); - this._client = null; - this._coll = null; - return cb(err); - }); - } else { - this._coll = null; - return cb(); - } - } -} - -module.exports = LogConsumer; diff --git a/tests/unit/storage/metadata/mongoclient/ListRecordStream.spec.js b/tests/unit/storage/metadata/mongoclient/ListRecordStream.spec.js deleted file mode 100644 index 3957c5255..000000000 --- a/tests/unit/storage/metadata/mongoclient/ListRecordStream.spec.js +++ /dev/null @@ -1,440 +0,0 @@ -const assert = require('assert'); -const { Timestamp } = require('mongodb'); - -const ListRecordStream = require( - '../../../../../lib/storage/metadata/mongoclient/ListRecordStream'); -const DummyRequestLogger = require('./utils/DummyRequestLogger'); - -const logger = new DummyRequestLogger(); - -const mongoProcessedLogEntries = { - insert: { - h: -42, - ts: new Timestamp({ t: 1651144629 , i: 1 }), - op: 'i', - ns: 'metadata.replicated-bucket', - o: { - _id: 'replicated-key\u000098467518084696999999RG001 19.3', - value: { - someField: 'someValue', - }, - }, - }, - updateObject: { - h: -42, - ts: new Timestamp({ t: 1651144629 , i: 1 }), - op: 'u', - ns: 'metadata.replicated-bucket', - o2: { - _id: 'replicated-key\u000098467518084696999999RG001 19.3', - }, - o: { - $set: { - value: { - someField: 'someUpdatedValue', - }, - }, - }, - }, - deleteObject: { - h: -42, - ts: new Timestamp({ t: 1651144629 , i: 1 }), - op: 'd', - ns: 'metadata.replicated-bucket', - o: { - _id: 'replicated-key\u000098467518084696999999RG001 19.3', - }, - }, - updateBeforeDeleteObject: { - h: -42, - ts: new Timestamp({ t: 1651144629 , i: 1 }), - op: 'u', - ns: 'metadata.replicated-bucket', - o2: { - _id: 'replicated-key\u000098467518084696999999RG001 19.3', - }, - o: { - $set: { - value: { - deleted: true, - }, - }, - }, - }, - putBucketAttributes: { - h: -42, - ts: new Timestamp({ t: 1651144629 , i: 1 }), - op: 'u', - ns: 'metadata.__metastore', - o2: { - _id: 'new-bucket', - }, o: { - _id: 'new-bucket', - value: { - someField: 'someValue', - }, - }, - }, - deleteBucket: { - h: -42, - ts: new Timestamp({ t: 1651144629 , i: 1 }), - op: 'd', - ns: 'metadata.__metastore', - o: { - _id: 'new-bucket', - }, - }, -}; - -const mongoIgnoredLogEntries = { - createBucket: { - h: -42, - ts: new Timestamp({ t: 1651144629 , i: 1 }), - op: 'c', - ns: 'metadata.$cmd', - o: { - create: 'new-bucket', - idIndex: { - v: 2, - key: { _id: 1 }, - name: '_id_', - ns: 'metadata.new-bucket', - }, - }, - }, - dropBucketDb: { - h: -42, - ts: new Timestamp({ t: 1651144629 , i: 1 }), - op: 'c', - ns: 'metadata.$cmd', - o: { - drop: 'new-bucket', - }, - }, -}; - -const expectedStreamEntries = { - insert: { - db: 'replicated-bucket', - entries: [ - { - key: 'replicated-key\u000098467518084696999999RG001 19.3', - type: 'put', - value: '{"someField":"someValue"}', - }, - ], - timestamp: new Date(1651144629 * 1000), - }, - updateObject: { - db: 'replicated-bucket', - entries: [ - { - key: 'replicated-key\u000098467518084696999999RG001 19.3', - type: 'put', - value: '{"someField":"someUpdatedValue"}', - }, - ], - timestamp: new Date(1651144629 * 1000), - }, - deleteObject: { - db: 'replicated-bucket', - entries: [ - { - key: 'replicated-key\u000098467518084696999999RG001 19.3', - type: 'delete', - }, - ], - timestamp: new Date(1651144629 * 1000), - }, - updateBeforeDeleteObject: { - db: 'replicated-bucket', - entries: [ - { - key: 'replicated-key\u000098467518084696999999RG001 19.3', - type: 'delete', - value: '{"deleted":true}', - }, - ], - timestamp: new Date(1651144629 * 1000), - }, - putBucketAttributes: { - db: '__metastore', - entries: [ - { - key: 'new-bucket', - type: 'put', - value: '{"someField":"someValue"}', - }, - ], - timestamp: new Date(1651144629 * 1000), - }, - deleteBucket: { - db: '__metastore', - entries: [ - { - key: 'new-bucket', - type: 'delete', - }, - ], - timestamp: new Date(1651144629 * 1000), - }, - dropBucketDb: { - h: -42, - op: 'c', - ns: 'metadata.$cmd', - o: { - drop: 'new-bucket', - }, - }, -}; - -class MongoCursorMock { - constructor(itemsToYield, errorAtPos) { - this.itemsToYield = itemsToYield; - this.pos = 0; - this.errorAtPos = errorAtPos; - } - - async next() { - // if there's no more item, just hang out there waiting for - // items that will never come (this is how the real mongo - // tailable cursor would behave) - if (this.pos === this.errorAtPos) { - throw new Error('boo'); - } - if (!this.hasSentAllItems()) { - const pos = this.pos; - this.pos += 1; - return this.itemsToYield[pos]; - } - return new Promise(resolve => { - // Leave the Promise unresolved to simulate waiting for more data - }); - } - - hasSentAllItems() { - return this.pos === this.itemsToYield.length; - } -} - -describe('mongoclient.ListRecordStream', () => { - const lastEndIDEntry = { - h: -43, - ts: Timestamp.fromNumber(42), - }; - Object.keys(mongoProcessedLogEntries).forEach(entryType => { - it(`should transform ${entryType}`, done => { - // first write will be ignored by ListRecordStream because - // of the last end ID (-42), it's needed though to bootstrap it - const cursor = new MongoCursorMock([ - lastEndIDEntry, - mongoProcessedLogEntries[entryType], - ]); - const lrs = new ListRecordStream(cursor, logger, - lastEndIDEntry.h.toString()); - let hasReceivedData = false; - lrs.on('data', entry => { - assert.strictEqual(hasReceivedData, false); - hasReceivedData = true; - assert.deepStrictEqual(entry, expectedStreamEntries[entryType]); - if (cursor.hasSentAllItems()) { - assert.strictEqual(hasReceivedData, true); - assert.deepStrictEqual(JSON.parse(lrs.getOffset()), - { uniqID: '-42' }); - done(); - } - }); - }); - }); - - it('should ignore other entry types', done => { - // first write will be ignored by ListRecordStream because - // of the last end ID (-43), it's needed though to bootstrap it - const logEntries = [lastEndIDEntry]; - Object.keys(mongoIgnoredLogEntries).forEach(entryType => { - logEntries.push(mongoIgnoredLogEntries[entryType]); - }); - const cursor = new MongoCursorMock(logEntries); - const lrs = new ListRecordStream(cursor, logger, - lastEndIDEntry.h.toString()); - lrs.on('data', entry => { - assert(false, `ListRecordStream did not ignore entry ${entry}`); - }); - setTimeout(() => { - assert.strictEqual(cursor.hasSentAllItems(), true); - assert.deepStrictEqual(JSON.parse(lrs.getOffset()), - { uniqID: '-42' }); - done(); - }, 200); - }); - - it('should skip entries until uniqID is encountered', done => { - const logEntries = [ - Object.assign({}, mongoProcessedLogEntries.insert, - { h: 1234, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - Object.assign({}, mongoProcessedLogEntries.insert, - { h: 5678, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - Object.assign({}, mongoProcessedLogEntries.insert, - { h: -1234, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - Object.assign({}, mongoProcessedLogEntries.insert, - { h: 2345, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - ]; - const cursor = new MongoCursorMock(logEntries); - const lrs = new ListRecordStream(cursor, logger, '5678'); - assert.strictEqual(lrs.reachedUnpublishedListing(), false); - let nbReceivedEntries = 0; - lrs.on('data', entry => { - assert.deepStrictEqual(entry, expectedStreamEntries.insert); - assert.strictEqual(lrs.reachedUnpublishedListing(), true); - ++nbReceivedEntries; - if (cursor.hasSentAllItems()) { - assert.strictEqual(nbReceivedEntries, 2); - assert.deepStrictEqual(JSON.parse(lrs.getOffset()), - { uniqID: '2345' }); - assert.strictEqual(lrs.getSkipCount(), 2); - assert.strictEqual(lrs.reachedUnpublishedListing(), true); - done(); - } - }); - }); - - it('should start after latest entry if uniqID is not encountered', done => { - const logEntries = [ - Object.assign({}, mongoProcessedLogEntries.insert, - { h: 1234, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - Object.assign({}, mongoProcessedLogEntries.insert, - { h: 5678, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - Object.assign({}, mongoProcessedLogEntries.insert, - { h: -1234, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - Object.assign({}, mongoProcessedLogEntries.insert, - { h: 2345, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - ]; - const cursor = new MongoCursorMock(logEntries); - const lrs = new ListRecordStream(cursor, logger, '4242', '-1234'); - let nbReceivedEntries = 0; - lrs.on('data', entry => { - assert.deepStrictEqual(entry, expectedStreamEntries.insert); - ++nbReceivedEntries; - if (cursor.hasSentAllItems()) { - assert.strictEqual(nbReceivedEntries, 1); - assert.deepStrictEqual(JSON.parse(lrs.getOffset()), - { uniqID: '2345' }); - assert.strictEqual(lrs.getSkipCount(), 3); - assert.strictEqual(lrs.reachedUnpublishedListing(), true); - done(); - } - }); - }); - - it('should consume from the first entry if there is no saved ID', done => { - const logEntries = [ - Object.assign({}, mongoProcessedLogEntries.insert, - { h: 1234, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - Object.assign({}, mongoProcessedLogEntries.insert, - { h: 5678, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - Object.assign({}, mongoProcessedLogEntries.insert, - { h: -1234, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - Object.assign({}, mongoProcessedLogEntries.insert, - { h: 2345, ts: new Timestamp({ t: 1651144629 , i: 1 }) }), - ]; - const cursor = new MongoCursorMock(logEntries); - const lrs = new ListRecordStream(cursor, logger, undefined, '-1234'); - let nbReceivedEntries = 0; - lrs.on('data', entry => { - assert.deepStrictEqual(entry, expectedStreamEntries.insert); - ++nbReceivedEntries; - if (cursor.hasSentAllItems()) { - assert.strictEqual(nbReceivedEntries, 4); - assert.deepStrictEqual(JSON.parse(lrs.getOffset()), - { uniqID: '2345' }); - assert.strictEqual(lrs.getSkipCount(), 0); - assert.strictEqual(lrs.reachedUnpublishedListing(), true); - done(); - } - }); - }); - - it('should emit an error event when cursor returns an error', done => { - const cursor = new MongoCursorMock([], 0); - const lrs = new ListRecordStream(cursor, logger, '4242', '-1234'); - lrs.on('data', () => { - assert(false, 'did not expect data'); - }); - lrs.on('error', () => done()); - }); - - it('should support bucket names with dots', done => { - const logEntry = { - h: -42, - ts: new Timestamp({ t: 1651144629 , i: 1 }), - op: 'i', - ns: 'metadata.some.bucket.with.dots', - o: { - _id: 'replicated-key\u000098467518084696999999RG001 19.3', - value: { - someField: 'someValue', - }, - }, - }; - const expectedLogEntry = { - db: 'some.bucket.with.dots', - entries: [ - { - key: 'replicated-key\u000098467518084696999999RG001 19.3', - type: 'put', - value: '{"someField":"someValue"}', - }, - ], - timestamp: new Date(1651144629 * 1000), - }; - const cursor = new MongoCursorMock([ - lastEndIDEntry, - logEntry, - ]); - const lrs = new ListRecordStream(cursor, logger, - lastEndIDEntry.h.toString()); - lrs.on('data', entry => { - assert.deepStrictEqual(entry, expectedLogEntry); - done(); - }); - }); - - it('should support tags with dots and dollars', done => { - const logEntry = { - h: -42, - ts: new Timestamp({ t: 1651144629 , i: 1 }), - op: 'i', - ns: 'some-bucket', - o: { - _id: 'replicated-key\u000098467518084696999999RG001 19.3', - value: { - tags: { - 'some\uFF04weird\uFF0Ekey': - 'some$weird.value', - }, - }, - }, - }; - const expectedLogEntry = { - db: 'some-bucket', - entries: [ - { - key: 'replicated-key\u000098467518084696999999RG001 19.3', - type: 'put', - value: '{"tags":{"some$weird.key":"some$weird.value"}}', - }, - ], - timestamp: new Date(1651144629 * 1000), - }; - const cursor = new MongoCursorMock([ - lastEndIDEntry, - logEntry, - ]); - const lrs = new ListRecordStream(cursor, logger, - lastEndIDEntry.h.toString()); - lrs.on('data', entry => { - assert.deepStrictEqual(entry, expectedLogEntry); - done(); - }); - }); -}); diff --git a/tests/unit/storage/metadata/mongoclient/LogConsumer.spec.js b/tests/unit/storage/metadata/mongoclient/LogConsumer.spec.js deleted file mode 100644 index 52b28c39c..000000000 --- a/tests/unit/storage/metadata/mongoclient/LogConsumer.spec.js +++ /dev/null @@ -1,453 +0,0 @@ -'use strict'; // eslint-disable-line - -const assert = require('assert'); -const sinon = require('sinon'); -const werelogs = require('werelogs'); -const { MongoMemoryServer } = require('mongodb-memory-server'); -const { MongoClient } = require('mongodb'); - -const LogConsumer = require( - '../../../../../lib/storage/metadata/mongoclient/LogConsumer'); - -describe('MongoDB LogConsumer', () => { - let sandbox; - let logger; - let mongoConfig; - let mongoServer; - let activeConsumers = []; - - const cleanupConsumers = async () => { - const cleanups = activeConsumers.map(consumer => - new Promise(resolve => { - if (consumer && consumer._client) { - consumer.close(err => { - if (err) { - console.error('Error closing consumer:', err); - } - resolve(); - }); - } else { - resolve(); - } - }) - ); - await Promise.all(cleanups); - activeConsumers = []; - }; - - beforeAll(async () => { - mongoServer = await MongoMemoryServer.create(); - const uri = mongoServer.getUri(); - const parsedUri = new URL(uri); - const host = parsedUri.host; - - const dbName = parsedUri.pathname.substring(1) || 'test'; - - mongoConfig = { - replicaSetHosts: host, - database: dbName, - readPreference: 'primary', - connectOptions: { directConnection: true } - }; - }); - - afterAll(async () => { - await cleanupConsumers(); - if (mongoServer) { - await mongoServer.stop(); - } - }); - - beforeEach(() => { - sandbox = sinon.createSandbox(); - logger = new werelogs.Logger('LogConsumer'); - activeConsumers = []; - }); - - afterEach(async () => { - sandbox.restore(); - await cleanupConsumers(); - }); - - describe('constructor', () => { - it('should correctly initialize the class properties', () => { - const consumer = new LogConsumer(mongoConfig, logger); - - assert.strictEqual(consumer._mongoUrl, `mongodb://${mongoConfig.replicaSetHosts}/`); - assert.strictEqual(consumer._replicaSet, undefined); - assert.strictEqual(consumer._readPreference, 'primary'); - assert.strictEqual(consumer._logger, logger); - assert.deepStrictEqual(consumer._oplogNsRegExp, new RegExp(`^${mongoConfig.database}\\.`)); - assert.strictEqual(consumer._coll, null); - assert.strictEqual(consumer._client, null); - }); - }); - - describe('connectMongo', () => { - it('should successfully connect to MongoDB', done => { - const consumer = new LogConsumer(mongoConfig, logger); - activeConsumers.push(consumer); - - consumer.connectMongo(err => { - assert.ifError(err); - assert.notStrictEqual(consumer._coll, null); - assert.strictEqual(typeof consumer._client.close, 'function'); - done(); - }); - }); - - it('should handle connection errors with invalid config', done => { - const invalidConfig = { - ...mongoConfig, - replicaSetHosts: 'invalid-host:27017', - }; - - const consumer = new LogConsumer(invalidConfig, logger); - activeConsumers.push(consumer); - - const connectStub = sandbox.stub(MongoClient.prototype, 'connect'); - connectStub.rejects(new Error('Connection error')); - - consumer.connectMongo(err => { - assert.ok(err instanceof Error); - assert.strictEqual(consumer._coll, null); - done(); - }); - }); - }); - - describe('readRecords', () => { - it('should setup tailable cursor when no startSeq is provided', done => { - const consumer = new LogConsumer(mongoConfig, logger); - activeConsumers.push(consumer); - - const mockCursor = { - find: sandbox.stub().returns({}) - }; - consumer._coll = mockCursor; - - const latestOplogID = '123456789'; - sandbox.stub(consumer, '_readLatestOplogID').callsFake((cb) => { - cb(null, latestOplogID); - }); - - const mockStreamInstance = {}; - sandbox.stub(consumer, 'readRecords').callsFake((params, cb) => { - consumer._readLatestOplogID((err, latestOplogID) => { - if (err) { - return cb(err); - } - const cursor = consumer._coll.find({ - ns: consumer._oplogNsRegExp, - }, { - tailable: true, - awaitData: true, - noCursorTimeout: true, - numberOfRetries: Number.MAX_VALUE, - }); - return cb(null, { log: mockStreamInstance, tailable: true }); - }); - }).callThrough(); - - consumer.readRecords({}, (err, result) => { - assert.strictEqual(err, null); - assert.ok(result); - assert.strictEqual(result.tailable, true); - assert.strictEqual(result.log, mockStreamInstance); - - assert(consumer._readLatestOplogID.calledOnce); - assert(mockCursor.find.calledOnce); - assert.deepStrictEqual(mockCursor.find.firstCall.args[0].ns, consumer._oplogNsRegExp); - - const options = mockCursor.find.firstCall.args[1]; - assert.strictEqual(options.tailable, true); - assert.strictEqual(options.awaitData, true); - assert.strictEqual(options.noCursorTimeout, true); - assert.strictEqual(options.numberOfRetries, Number.MAX_VALUE); - - done(); - }); - }); - - it('should use startSeq if provided', done => { - const consumer = new LogConsumer(mongoConfig, logger); - activeConsumers.push(consumer); - - const mockCursor = { - find: sandbox.stub().returns({}) - }; - consumer._coll = mockCursor; - - const latestOplogID = '123456789'; - sandbox.stub(consumer, '_readLatestOplogID').callsFake((cb) => { - cb(null, latestOplogID); - }); - - const mockStreamInstance = {}; - let startSeqUniqID; - - sandbox.stub(consumer, 'readRecords').callsFake((params, cb) => { - let startSeq = {}; - if (params.startSeq) { - try { - startSeq = JSON.parse(params.startSeq); - } catch { - // Ignore parsing errors - } - } - - consumer._readLatestOplogID(err => { - if (err) { - return cb(err); - } - consumer._coll.find({ - ns: consumer._oplogNsRegExp, - }, { - tailable: true, - awaitData: true, - noCursorTimeout: true, - numberOfRetries: Number.MAX_VALUE, - }); - - startSeqUniqID = startSeq.uniqID; - - return cb(null, { log: mockStreamInstance, tailable: true }); - }); - }).callThrough(); - - const startSeq = JSON.stringify({ uniqID: '987654321' }); - - consumer.readRecords({ startSeq }, (err, result) => { - assert.strictEqual(err, null); - assert.ok(result); - assert.strictEqual(result.tailable, true); - assert.strictEqual(result.log, mockStreamInstance); - - assert(consumer._readLatestOplogID.calledOnce); - assert(mockCursor.find.calledOnce); - assert.strictEqual(startSeqUniqID, '987654321'); - - done(); - }); - }); - - it('should handle invalid startSeq JSON', done => { - const consumer = new LogConsumer(mongoConfig, logger); - activeConsumers.push(consumer); - - const mockCursor = { - find: sandbox.stub().returns({}) - }; - consumer._coll = mockCursor; - - const latestOplogID = '123456789'; - sandbox.stub(consumer, '_readLatestOplogID').callsFake((cb) => { - cb(null, latestOplogID); - }); - - const mockStreamInstance = {}; - let startSeqUniqID; - - sandbox.stub(consumer, 'readRecords').callsFake((params, cb) => { - let startSeq = {}; - if (params.startSeq) { - try { - startSeq = JSON.parse(params.startSeq); - } catch { - // Ignore parsing errors - } - } - - consumer._readLatestOplogID(err => { - if (err) { - return cb(err); - } - const cursor = consumer._coll.find({ - ns: consumer._oplogNsRegExp, - }, { - tailable: true, - awaitData: true, - noCursorTimeout: true, - numberOfRetries: Number.MAX_VALUE, - }); - - startSeqUniqID = startSeq.uniqID; - - return cb(null, { log: mockStreamInstance, tailable: true }); - }); - }).callThrough(); - - const startSeq = 'not a valid json'; - - consumer.readRecords({ startSeq }, (err, result) => { - assert.strictEqual(err, null); - assert.ok(result); - - assert(consumer._readLatestOplogID.calledOnce); - assert(mockCursor.find.calledOnce); - assert.strictEqual(startSeqUniqID, undefined); - - done(); - }); - }); - - it('should handle error from _readLatestOplogID', done => { - const consumer = new LogConsumer(mongoConfig, logger); - activeConsumers.push(consumer); - - const mockCursor = { - find: sandbox.stub().returns({}) - }; - consumer._coll = mockCursor; - - const testError = new Error('Failed to read latest oplog ID'); - sandbox.stub(consumer, '_readLatestOplogID').callsFake((cb) => { - cb(testError); - }); - - consumer.readRecords({}, (err, result) => { - assert.strictEqual(err, testError); - assert.strictEqual(result, undefined); - - assert(consumer._readLatestOplogID.calledOnce); - assert(!mockCursor.find.called); - - done(); - }); - }); - }); - - describe('_readLatestOplogID', () => { - it('should return latest oplog ID when documents exist', done => { - const consumer = new LogConsumer(mongoConfig, logger); - activeConsumers.push(consumer); - - const mockData = [{ h: 123456789 }]; - const mockCollectionStub = { - find: sandbox.stub().returns({ - sort: sandbox.stub().returns({ - limit: sandbox.stub().returns({ - toArray: sandbox.stub().resolves(mockData) - }) - }) - }) - }; - - consumer._coll = mockCollectionStub; - - consumer._readLatestOplogID((err, oplogID) => { - assert.strictEqual(err, null); - assert.strictEqual(oplogID, '123456789'); - done(); - }); - }); - - it('should return error when no documents exist', done => { - const consumer = new LogConsumer(mongoConfig, logger); - activeConsumers.push(consumer); - const mockCollectionStub = { - find: sandbox.stub().returns({ - sort: sandbox.stub().returns({ - limit: sandbox.stub().returns({ - toArray: sandbox.stub().resolves([]) - }) - }) - }) - }; - consumer._coll = mockCollectionStub; - consumer._readLatestOplogID((err, oplogID) => { - assert(err instanceof Error); - assert.strictEqual(err.message, 'no oplog entry found'); - assert.strictEqual(oplogID, undefined); - done(); - }); - }); - - it('should handle MongoDB query errors', done => { - const consumer = new LogConsumer(mongoConfig, logger); - activeConsumers.push(consumer); - const testError = new Error('MongoDB query failed'); - const mockCollectionStub = { - find: sandbox.stub().returns({ - sort: sandbox.stub().returns({ - limit: sandbox.stub().returns({ - toArray: sandbox.stub().rejects(testError) - }) - }) - }) - }; - - consumer._coll = mockCollectionStub; - - consumer._readLatestOplogID((err, oplogID) => { - assert.strictEqual(err, testError); - assert.strictEqual(oplogID, undefined); - done(); - }); - }); - }); - - describe('close', () => { - it('should close MongoDB client connection if connected', done => { - const consumer = new LogConsumer(mongoConfig, logger); - - const closeSpy = sandbox.stub().resolves(); - consumer._client = { - close: closeSpy - }; - - consumer.close(err => { - assert.ifError(err); - assert(closeSpy.calledOnce); - assert.strictEqual(consumer._client, null); - assert.strictEqual(consumer._coll, null); - done(); - }); - }); - - it('should handle errors when closing MongoDB client', done => { - const consumer = new LogConsumer(mongoConfig, logger); - - const testError = new Error('Error closing connection'); - const closeSpy = sandbox.stub().rejects(testError); - consumer._client = { - close: closeSpy - }; - - consumer.close(err => { - assert.strictEqual(err, testError); - assert(closeSpy.calledOnce); - assert.strictEqual(consumer._client, null); - assert.strictEqual(consumer._coll, null); - done(); - }); - }); - - it('should not fail if client is not connected', done => { - const consumer = new LogConsumer(mongoConfig, logger); - consumer._client = null; - - consumer.close(err => { - assert.ifError(err); - assert.strictEqual(consumer._client, null); - assert.strictEqual(consumer._coll, null); - done(); - }); - }); - - it('should properly close a real connection to MongoDB', done => { - const consumer = new LogConsumer(mongoConfig, logger); - consumer.connectMongo(connErr => { - assert.ifError(connErr); - assert.notStrictEqual(consumer._client, null); - assert.notStrictEqual(consumer._coll, null); - consumer.close(closeErr => { - assert.ifError(closeErr); - assert.strictEqual(consumer._client, null); - assert.strictEqual(consumer._coll, null); - done(); - }); - }); - }); - }); -}); \ No newline at end of file From a309450ccb07fd9d2a39dc900d50c1e62cee523f Mon Sep 17 00:00:00 2001 From: Thomas Flament Date: Fri, 7 Aug 2026 15:53:53 +0200 Subject: [PATCH 3/3] Bump package.json to 8.5.13 Issue: ARSN-622 --- package.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/package.json b/package.json index 8d4217cd5..e4f045262 100644 --- a/package.json +++ b/package.json @@ -3,7 +3,7 @@ "engines": { "node": ">=20" }, - "version": "8.5.12", + "version": "8.5.13", "config": { "mongodbMemoryServer": { "version": "8.0.23"