ChangeStream stops working after some time

[thaoula]

Hi All,

We are trying to utilise ChangeStreams using Mongoose and have encountered some challenges with reliability.

We have the following issues:

• For some unknown reason, our DB level ChangeStream which is connected to an Atlas Cluster stops receiving changes. We do not get any errors even though we are using ChangeStream.on('error') listener. Nothing is logged.

• When this happens, restarting the container but keeping the resume token i.e we are setting resumeAfter does not work. We only get changes after deleting the resume token and restarting the container.

Before logging some Github issues I wanted to see if anyone has experienced problems like this and whether you have some tips.

What Works

1. We are able to successfully open a ChangeStream at the database level using connection.watch().
2. We can receive the change event.
3. We can store the resume token after each change event is called.
4. We can restart the application and resumeAfter the previous resume token.

For some unknown reason (maybe low activity due to weekend or overnight, or primary changes), the change stream stops receiving change events. I am not sure how to debug this issue as no event listeners are called on the change stream. ie. error, resumeTokenChanged, closed events are never ever triggered.

By chance after reviewing the mongoose code and noticing the returned changestream was a wrapped object, I decided to just use connection.getClient().watch which returns a mongdb native ChangeStream and I am able to successfully listen to resumeTokenChanged, closed, change and error events as I would have expected when using connection.watch().

When using this approach, I suprisingly found that the resumeTokenChanged event is called almost once per second and the token is changing. Also, the token returned by this event is also much shorter than the resume token that is provided in the change event.

I also noticed that if i stop the database (test failure) the change stream is closed (it calls the close event listener) and it is able to resume when i restart the database server.

My questions are -

1. Is connection.watch Mongoose Changestream object meant to output all the events?
2. Is it safe to use connection.getClient().watch to get a native change stream?
3. Why is resumeTokenChanged called every second, should I persist the token provided in this event?
4. In general, how can I detect if the change stream stopped working. I am wondering if the resumeTokenChanged event can be used as a heartbeat and it it stops for a period of time .. i should reconnect the changestream.

Sorry for the all the questions.
Regards,
Tarek

[vkarpov15]

@thaoula can you please provide the code snippet you used to repro this behavior?

[thaoula]

Hi @vkarpov15,

Thanks for replying.

Please find the code below where I am using a MongoDB change stream obtained by calling .getClient().

I have been running the changestream for the last 3 to 4 days and seems to be still active and I am restart the app and resume where I left off.

The mongoose based changestream never raised any events and is missing RESUME_TOKEN_CHANGED which I think is very important when tracking changes on collections / databases where changes are infrequent. The token received in the last change event may have been removed from the oplog so you cannot resume.

 const changeStream = this.changeStream = this.db
            .connection

           // Remove these two lines to get a Mongoose changestream.
            .getClient()
            .db(this.configuration.database.mongo.dbName)
            /// 
            .watch(pipeline, options)
            .on(ChangeStream.INIT, () => {
               // Never called by getClient() based change stream or mongoose based changestream
            })
            .on(ChangeStream.ERROR, (error: any) => {
                /// This event is never by a mongoose based changestream. It called by a getClient() changestream and is very important
                /// for being able to resume correctly or handle a case where you cannot resume.

                const errorCodes = {
                    [9]: 'Resume of change stream was not possible, invalid resume token.',
                    [286]: 'Resume of change stream was not possible, as the resume point may no longer be in the oplog.',
                    [280]: 'Resume token was not found.'
                };

            })
            .on(ChangeStream.CLOSE, () => {
                this.logger.trace('Universal change stream closed');
            })
            .on(ChangeStream.END, () => {
                this.logger.trace('Universal change stream ended');
            })
            .on(ChangeStream.RESUME_TOKEN_CHANGED, async (token: any) => {
                /// This event does not exist when using Mongoose change stream. It is fired 1 time a second when using a changestream 
               ///  from getClient(). The token it provides is useful for collections or database that don't change a lot.
                this.changeStreamTokenChanged$.next(token);
            });

FYI we are using Mongo DB 6 and Mongoose 7.3.0. In production we are using Atlas cluster.

Regards,
Tarek

[vkarpov15]

Copying my response from #13607:

1. Yes. We put in a fix in fix(cursor): bubble up resumeTokenChanged event from change streams #13736 for resumeTokenChanged event
2. Yes. Mongoose's ChangeStream wrapper is just a convenience wrapper for buffering purposes.
3. You should persist the token provided, although the MongoDB Node driver does still emit this event even if the token hasn't changed. It looks like the MongoDB node driver emits 'resumeTokenChanged' every time it calls MongoDB to get more changes. It still emits 'resumeTokenChanged', even when the resume token is the same as the last event.
4. The change stream should be fine unless you receive an 'error' or 'close' event. For example, if the entire replica set shuts down in the middle of your change stream, you'll receive a 'close' event; although it can take up to a minute for that event to fire. For example, the following script:

'use strict';

const mongoose = require('mongoose');

void async function main() {
  const uri = 'mongodb://127.0.0.1:27017,127.0.0.1:27018/mongoose_test';
  await mongoose.connect(uri);
  const Person = mongoose.model('Person', mongoose.Schema({ name: String }));
  await Person.createCollection();

  // Create a change stream. The 'change' event gets emitted when there's a
  // change in the database
  const changeStream = await Person.watch();
  
  changeStream.on('change', data => console.log(new Date(), data));
  changeStream.on('resumeTokenChanged', data => console.log(new Date(), 'resumeTokenChanged', data));
  changeStream.on('error', data => console.log(new Date(), 'error', data));
  changeStream.on('close', data => console.log(new Date(), 'close', data));

  await new Promise(resolve => setTimeout(resolve, 1000));
  // Insert a doc, will trigger the change stream handler above
  console.log(new Date(), 'Inserting doc');
  await Person.create({ name: 'Axl Rose' });
  console.log(new Date(), 'Inserted doc');
}();

Prints the following output if I kill the entire replica set after a few seconds:

$ node gh-13607.js 
2023-08-14T19:55:13.537Z resumeTokenChanged { _data: '8264DA86A1000000022B0229296E04' }
2023-08-14T19:55:14.532Z Inserting doc
2023-08-14T19:55:14.545Z resumeTokenChanged { _data: '8264DA86A1000000022B0229296E04' }
2023-08-14T19:55:14.554Z Inserted doc
2023-08-14T19:55:14.554Z resumeTokenChanged {
  _data: '8264DA86A2000000012B022C0100296E5A1004B95EF18957CB4D9FA8B2D826162B6EA146645F6964006464DA86A257C9D18FDDBA9AE30004'
}
2023-08-14T19:55:14.555Z {
  _id: {
    _data: '8264DA86A2000000012B022C0100296E5A1004B95EF18957CB4D9FA8B2D826162B6EA146645F6964006464DA86A257C9D18FDDBA9AE30004'
  },
  operationType: 'insert',
  clusterTime: new Timestamp({ t: 1692042914, i: 1 }),
  fullDocument: {
    _id: new ObjectId("64da86a257c9d18fddba9ae3"),
    name: 'Axl Rose',
    __v: 0
  },
  ns: { db: 'mongoose_test', coll: 'people' },
  documentKey: { _id: new ObjectId("64da86a257c9d18fddba9ae3") }
}
2023-08-14T19:55:15.556Z resumeTokenChanged {
  _data: '8264DA86A2000000012B022C0100296E5A1004B95EF18957CB4D9FA8B2D826162B6EA146645F6964006464DA86A257C9D18FDDBA9AE30004'
}
2023-08-14T19:55:16.559Z resumeTokenChanged {
  _data: '8264DA86A2000000012B022C0100296E5A1004B95EF18957CB4D9FA8B2D826162B6EA146645F6964006464DA86A257C9D18FDDBA9AE30004'
}
2023-08-14T19:55:17.562Z resumeTokenChanged {
  _data: '8264DA86A2000000012B022C0100296E5A1004B95EF18957CB4D9FA8B2D826162B6EA146645F6964006464DA86A257C9D18FDDBA9AE30004'
}
2023-08-14T19:55:18.566Z resumeTokenChanged {
  _data: '8264DA86A2000000012B022C0100296E5A1004B95EF18957CB4D9FA8B2D826162B6EA146645F6964006464DA86A257C9D18FDDBA9AE30004'
}
2023-08-14T19:55:19.568Z resumeTokenChanged {
  _data: '8264DA86A2000000012B022C0100296E5A1004B95EF18957CB4D9FA8B2D826162B6EA146645F6964006464DA86A257C9D18FDDBA9AE30004'
}
2023-08-14T19:56:19.571Z close undefined
2023-08-14T19:56:19.571Z close undefined
2023-08-14T19:56:19.572Z close undefined
^C

[anantanandgupta]

I am facing the same problem. There is no absolute solution to this at least in mongoose. I have to try the native nodejs driver

[vkarpov15]

@anantanandgupta it is a bit strange that the close event is fired multiple times, we're investigating. But the general idea is that, as of this writing, you're responsible for recreating the change stream when a close event is emitted. In Node.js streams, close means the stream is complete and no more events will be emitted, so you should recreate the change stream. For example, the following code correctly picks up the change stream after a close event is emitted:

  let resumeAfter = null;
  let changeStream = await Person.watch();
  addChangeStreamListeners(changeStream);

  function addChangeStreamListeners(changeStream) {
    changeStream.on('change', data => console.log(new Date(), data));
    changeStream.on('resumeTokenChanged', data => {
      console.log(new Date(), 'resumeTokenChanged', data);
      resumeAfter = data;
    });
    changeStream.on('error', data => console.log(new Date(), 'error', data));
    changeStream.on('close', data => {
      console.log(new Date(), 'close', data);
      changeStream = Person.watch([], { resumeAfter });
      addChangeStreamListeners(changeStream);
    });
  }

[anantanandgupta]

Thank @vkarpov15. We were doing the same by exiting the process and restarting the process itself (being run in docker and relying on the capability of it to restart on error).

For multiple close event however i have no idea as the above code will create a new change handlers in this as.

I have tried earlier this approach only, but then the change event was firing multiple times and it kept growing on (for an instance in testing each change in database it sent almost 1024 emails).

recently i observed another issue that though the changestream is closed the event listers are somehow active and still processing the events even after we set the changestream variable to new watch in the close callback.

For your question on version number (i am sure we are using latest but will check again) and our current working code I will post another reply in some time once I go on my desk. (just wokeup 🥱🥱🥱)

[anantanandgupta]

here is my final piece of code

how to read this code:

1. there are fixed number of change streams and respective handlers (5 in count). since the ChangeStream is not an exported Type i had to use return type from the function.
2. the resume token is persisted in database on each change event. since the resumeTokenChanged event is fired too frequently, that is not the correct event to persist.
3. database connection wrapper takes a callback function, which gets attached to on connected event of database connection. this way whenever the application loses the connection (causing the change stream to close) and regains the connection, the change stream is recreated.
4. recreating the change stream was having some issue. somehow it was keeping the old listeners active in the memory even though the old stream was closed and a new stream is created, it was triggering old and new handlers simultaneously. to mitigate that i had to wrap the createChangeStream function call in a closeStream function, which removes the old listers and then calls the close() method.
5. there is no way for us to know if the stream is closed or not as the changeStream.closed property is for the cursor.

this code only fails when the resume token is not valid as there is no error thrown in that case. there should be a mechanism for lib to tell if the connection is closed, because the provided resume token is invalid. (you can test the use case by opening a changestream with invalid / expired resume token).

import { Constants, Types } from '@sap/common';
import * as database from '@sap/database';

import {
  onEmailVerificationQueue,
  onForgotPasswordQueue,
  onSchoolInvitationQueue,
  onSchoolRegistrationQueue,
  onStudentInvitationQueue,
} from './handlers';

export * from './handlers';

let forgotPasswordChangeStream: undefined | Awaited<ReturnType<typeof createChangeStream>> = undefined;
let schoolInvitationChangeStream: undefined | Awaited<ReturnType<typeof createChangeStream>> = undefined;
let studentInvitationChangeStream: undefined | Awaited<ReturnType<typeof createChangeStream>> = undefined;
let schoolRegistrationQueueChangeStream: undefined | Awaited<ReturnType<typeof createChangeStream>> = undefined;
let emailVerificationChangeStream: undefined | Awaited<ReturnType<typeof createChangeStream>> = undefined;

const createChangeStream = async (resumeTokenPurpose: Types.ResumeTokenPurpose, queueHandler?: (change: any) => void, operationType: string = 'insert') => {
  try {
    let resumeToken = (await database.Data.ResumeTokenData.readResumeToken(resumeTokenPurpose))?.toObject();
    const pipeline = [
      {
        $match: {
          $and: [
            { 'fullDocument.purpose': resumeTokenPurpose },
            { 'operationType': operationType },
          ],
        },
      },
    ];

    const changeStream = database.Data.VerificationCodeData.watch(pipeline, { resumeAfter: resumeToken ? { _data: resumeToken?._data } : undefined });

    changeStream.on('change', async (change: any) => {
      if (resumeToken?._data !== change?._id?._data) {
        resumeToken = await database.Data.ResumeTokenData.saveResumeToken(change._id, resumeTokenPurpose);
      }
      queueHandler?.(change.fullDocument);
    });

    changeStream.on('error', (...args) => {
      console.error(`[Change Stream] Error in ${resumeTokenPurpose} change stream:`, args);
      // we have to do this so that the docker container is destroyed and recreated, otherwise the change stream listener is never recovered.
      process.exit(1);
    });

    changeStream.on('close', (...args) => {
      console.error(`[Change Stream] Closed ${resumeTokenPurpose} change stream:`, args);
    });

    return changeStream;
  } catch (error) {
    console.error(`[Change Stream] Error connecting to the database for ${resumeTokenPurpose}:`, error);
    // we have to do this so that the docker container is destroyed and recreated, otherwise the change stream listener is never recovered.
    process.exit(1);
  }
};

const closeStream = async (stream: undefined | Awaited<ReturnType<typeof createChangeStream>>) => {
  if (stream && stream.listenerCount('change') > 0) {
    console.log('[Change Stream] Removing Listeners');
    stream?.removeAllListeners('change');

    return stream.close();
  } else {
    return;
  }
};

// initiate streams
database.Connection.getConnection(undefined, async () => {
  await closeStream(forgotPasswordChangeStream)
    .then(async () => {
      console.log('[Change Stream] Creating for Forgot Password');
      forgotPasswordChangeStream = await createChangeStream(Constants.ResumeTokenPurpose.forgot_password, onForgotPasswordQueue, 'insert');
    });
  await closeStream(schoolInvitationChangeStream)
    .then(async () => {
      console.log('[Change Stream] Creating for School Invitation');
      schoolInvitationChangeStream = await createChangeStream(Constants.ResumeTokenPurpose.school_invitation, onSchoolInvitationQueue, 'insert');
    });
  await closeStream(studentInvitationChangeStream)
    .then(async () => {
      console.log('[Change Stream] Creating for Student Invitation');
      studentInvitationChangeStream = await createChangeStream(Constants.ResumeTokenPurpose.student_invitation, onStudentInvitationQueue, 'insert');
    });
  await closeStream(schoolRegistrationQueueChangeStream)
    .then(async () => {
      console.log('[Change Stream] Creating for School Registration');
      schoolRegistrationQueueChangeStream = await createChangeStream(Constants.ResumeTokenPurpose.school_registration, onSchoolRegistrationQueue, 'insert');
    });
  await closeStream(emailVerificationChangeStream)
    .then(async () => {
      console.log('[Change Stream] Creating for Email Verification');
      emailVerificationChangeStream = await createChangeStream(Constants.ResumeTokenPurpose.email_verification, onEmailVerificationQueue, 'insert');
    });
});

[anantanandgupta]

also for the mongoose version, it is 7.6.3

[anantanandgupta]

updated to version 7.6.9