resumeToken understanding

[sibelius]

We are trying to make our change stream more resilient in production

Imagine that we have 1 pod that watch streams of a few collections

what if the pod crashes?

can we resume from where we stopped ?

We are trying to implement this using resumeToken

const resumeToken = getResumeToken();

const stream = model.watch([], {
    fullDocument: 'updateLookup',
    resumeAfter: resumeToken
  });

the problem is that we do not receive the events (insert, updates), that we missed until our pod was not working

what is the recommendation for this?

[sibelius]

I was using _id._data as resumeToken, but it shoud be the whole _id

[matiasgarcia]

@sibelius how did you fix this?

I do the following:

    const changeStream = this.chargeModel.watch([], {
      resumeAfter: this.resumeToken,
      fullDocument: 'updateLookup', // assess if there is a better way, this injects the fullDocument key
    });

and for example, this is a resumeToken I get:

    this.resumeToken = {
      _id: {
        _data:
          '8266968FCC000000022B042C0100296E5A1004F0298AB140B441B687E506FA66207E19463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B657900465A5F6964005A1004E447653E78C6415E9861BF6DB7BD2D6E000004',
      },
    };

However, this fails:

/Users/matias/projects/services-nuvem-pago-commons/node_modules/mongoose/node_modules/mongodb/src/cmap/connection.ts:526
          throw new MongoServerError((object ??= document.toObject(bsonOptions)));
                ^
MongoServerError: Bad resume token: _data of missing or of wrong type. Expected string, got {_id: {_data: "8266968FCC000000022B042C0100296E5A1004F0298AB140B441B687E506FA66207E19463C6F7065726174696F6E54797065003C7570646174650046646F63756D656E744B657900465A5F6964005A1004E447653E78C6415E9861BF6DB7BD2D6E000004"}}
    at Connection.sendCommand (/Users/matias/projects/services-nuvem-pago-commons/node_modules/mongoose/node_modules/mongodb/src/cmap/connection.ts:526:17)
    at processTicksAndRejections (node:internal/process/task_queues:95:5)
    at Connection.command (/Users/matias/projects/services-nuvem-pago-commons/node_modules/mongoose/node_modules/mongodb/src/cmap/connection.ts:609:22)
    at Server.command (/Users/matias/projects/services-nuvem-pago-commons/node_modules/mongoose/node_modules/mongodb/src/sdam/server.ts:326:16)
    at AggregateOperation.executeCommand (/Users/matias/projects/services-nuvem-pago-commons/node_modules/mongoose/node_modules/mongodb/src/operations/command.ts:155:12)
    at AggregateOperation.execute (/Users/matias/projects/services-nuvem-pago-commons/node_modules/mongoose/node_modules/mongodb/src/operations/aggregate.ts:137:33)
    at executeOperation (/Users/matias/projects/services-nuvem-pago-commons/node_modules/mongoose/node_modules/mongodb/src/operations/execute_operation.ts:196:12)
    at ChangeStreamCursor._initialize (/Users/matias/projects/services-nuvem-pago-commons/node_modules/mongoose/node_modules/mongodb/src/cursor/change_stream_cursor.ts:149:22)
    at ChangeStreamCursor.[kInit] (/Users/matias/projects/services-nuvem-pago-commons/node_modules/mongoose/node_modules/mongodb/src/cursor/abstract_cursor.ts:657:21)
    at next (/Users/matias/projects/services-nuvem-pago-commons/node_modules/mongoose/node_modules/mongodb/src/cursor/abstract_cursor.ts:763:7)

[sibelius]

export const changeStreamListen = async <T extends any>(
  model: Model<T>,
  fn: (data: ChangeStreamData<T>) => void,
) => {
  const name = model.collection.name;
  const key = `resumetoken:${name}`;
  const resumeToken = await RedisCache.get(key);
  const clusterTime = await RedisCache.get(`clustertime:${name}`);

  const getStartAtOperationTime = () => {
    if (clusterTime) {
      return Timestamp.fromString(clusterTime.$timestamp);
    }

    return null;
  };

  const getResumeOptions = () => {
    if (resumeToken) {
      return {
        resumeAfter: resumeToken,
        // startAfter: resumeToken,
        // startAtOperationTime,
      };
    }

    const startAtOperationTime = getStartAtOperationTime();

    if (startAtOperationTime) {
      return {
        startAtOperationTime,
      };
    }

    return {};
  };

  const stream = model.watch([], {
    fullDocument: 'updateLookup',
    // fullDocumentBeforeChange: 'whenAvailable',
    // fullDocumentBeforeChange: 'required',
    // batchSize: 100,
    ...getResumeOptions(),
  });

  stream.on('change', changeStreamMiddleware(name, fn)).on('error', (err) => {
    // eslint-disable-next-line
    console.log('change error:', err);
    Sentry.setExtra('error', err);
    Sentry.captureException(err);
  });
};