Exit the oplog populator when the metastore stream is invalidated

extensions/oplogPopulator/OplogPopulator.js

@@ -199,6 +199,18 @@ class OplogPopulator {
                 }
                 break;
             }
+            case 'drop':
+            case 'rename':
+            case 'dropDatabase':
+            case 'invalidate':
+                // The change stream is closed for good, and buckets may have been removed or
+                // restored meanwhile: restart so that setup() re-syncs them on a new stream.
+                this._logger.error('Metastore change stream invalidated, exiting', {
+                    method: 'OplogPopulator._handleChangeStreamChange',
+                    type: change.operationType,
+                });
+                process.exit(1);
+                return;
             default:
                 this._logger.info('Skipping unsupported change stream event', {
                     method: 'OplogPopulator._handleChangeStreamChange',

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@scality/backbeat",
-  "version": "9.6.0-preview.2",
+  "version": "9.6.0-preview.3",
   "description": "Asynchronous queue and job manager",
   "main": "index.js",
   "scripts": {

tests/unit/oplogPopulator/oplogPopulator.js

@@ -857,7 +857,7 @@ describe('OplogPopulator', () => {
                 has: () => true,
             };
             const changeDocument = {
-                operationType: 'drop',
+                operationType: 'modify',
                 documentKey: {
                     _id: 'example-bucket',
                 },
@@ -878,16 +878,35 @@ describe('OplogPopulator', () => {
                 has: () => true,
             };
             const eventProcessed = sinon.stub(oplogPopulator._metricsHandler, 'onOplogEventProcessed');
-            // "drop" and "invalidate" events carry no documentKey
             const changeDocument = {
-                operationType: 'drop',
+                operationType: 'createIndexes',
             };
             await oplogPopulator._handleChangeStreamChangeEvent(changeDocument);
             assert(stopListening.notCalled);
             assert(startListening.notCalled);
             assert(eventProcessed.calledOnce);
         });
 
+        ['drop', 'rename', 'dropDatabase', 'invalidate'].forEach(opType => {
+            it(`should exit when the change stream gets invalidated (${opType})`, async () => {
+                const exit = sinon.stub(process, 'exit');
+                const stopListening = sinon.stub().resolves();
+                const startListening = sinon.stub().resolves();
+                oplogPopulator._allocator = {
+                    stopListeningToBucket: stopListening,
+                    listenToBucket: startListening,
+                    has: () => true,
+                };
+                const eventProcessed = sinon.stub(oplogPopulator._metricsHandler, 'onOplogEventProcessed');
+                // these events carry no documentKey
+                await oplogPopulator._handleChangeStreamChangeEvent({ operationType: opType });
+                assert(exit.calledOnceWith(1));
+                assert(stopListening.notCalled);
+                assert(startListening.notCalled);
+                assert(eventProcessed.notCalled);
+            });
+        });
+
         ['replace', 'insert', 'update'].forEach(opType => {
             it(`should stop listening if bucket becomes invalid (${opType})`, () => {
                 const stopListening = sinon.stub().resolves();