Merge pull request #29 from smartprocure/daveyarwood/GS-9973-regularly-update-resume-token

0.51.0: Bump mongochangestream to pick up refresh token improvements

Author: dubiousdavid

CHANGELOG.md

@@ -1,3 +1,9 @@
+# 0.51.0
+
+- Latest `mongochangestream` - Performance improvements related to updating the
+  refresh token more often. See [mongochangestream
+  changelog](https://github.com/smartprocure/mongochangestream/blob/master/CHANGELOG.md#0590)
+
 # 0.50.0
 
 - Remove try/catch and rely on retry logic in `mongochangestream`.

README.md

@@ -40,3 +40,16 @@ const options = { batchSize: 1000 }
 const initialScan = await sync.runInitialScan(options)
 initialScan.start()
 ```
+
+## Run the tests locally
+
+Create a .env file with the following variables set to the appropriate values:
+
+```
+MONGO_CONN="mongodb+srv://..."
+ELASTIC_NODE='https://elastic-node-url-here.com'
+ELASTIC_USERNAME="username-here"
+ELASTIC_PASSWORD="password-here"
+```
+
+Then run `npm test` to run the tests.

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "mongo2elastic",
-  "version": "0.50.0",
+  "version": "0.51.0",
   "description": "Sync MongoDB collections to Elasticsearch",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",
@@ -47,7 +47,7 @@
     "@types/node": "^22.10.2",
     "@typescript-eslint/eslint-plugin": "^8.19.0",
     "globals": "^15.14.0",
-    "mongochangestream-testing": "^0.1.0",
+    "mongochangestream-testing": "^0.5.0",
     "ms": "^2.1.3",
     "prettier": "^3.4.2",
     "typescript": "^5.7.2",
@@ -64,16 +64,20 @@
     "lodash": "^4.17.21",
     "make-error": "^1.3.6",
     "minimatch": "^10.0.1",
-    "mongochangestream": "^0.58.0",
+    "mongochangestream": "^0.59.0",
     "obj-walker": "^2.4.0",
     "prom-utils": "^0.14.0"
   },
   "prettier": {
     "semi": false,
     "singleQuote": true,
     "trailingComma": "es5",
-    "plugins": ["@trivago/prettier-plugin-sort-imports"],
-    "importOrder": ["^[./]"],
+    "plugins": [
+      "@trivago/prettier-plugin-sort-imports"
+    ],
+    "importOrder": [
+      "^[./]"
+    ],
     "importOrderSortSpecifiers": true,
     "importOrderCaseInsensitive": true,
     "importOrderSeparation": true

package.json

@@ -1,22 +1,23 @@
 import elasticsearch from '@elastic/elasticsearch'
+import debug from 'debug'
 import Redis from 'ioredis'
 import _ from 'lodash/fp.js'
 import {
+  assertEventually,
   initState as initRedisAndMongoState,
   numDocs,
 } from 'mongochangestream-testing'
 import { type Db, MongoClient } from 'mongodb'
 import ms from 'ms'
 import { setTimeout } from 'node:timers/promises'
-import { describe, expect, test } from 'vitest'
-import debug from 'debug'
+import { describe, test } from 'vitest'
+
+import { initSync, SyncOptions } from './index.js'
 
 // Output via console.info (stdout) instead of stderr.
 // Without this debug statements are swallowed by vitest.
 debug.log = console.info.bind(console)
 
-import { initSync, SyncOptions } from './index.js'
-
 const index = 'testing'
 
 const getConns = _.memoize(async () => {
@@ -79,13 +80,14 @@ describe.sequential('syncCollection', () => {
     await initElasticState(sync, db)
 
     const initialScan = await sync.runInitialScan()
-    // Wait for initial scan to complete
     await initialScan.start()
-    await setTimeout(ms('1s'))
+    // Test that all of the records are eventually synced.
+    await assertEventually(async () => {
+      const countResponse = await elasticClient.count({ index })
+      return countResponse.count == numDocs
+    }, `Less than ${numDocs} records were processed`)
     // Stop
     await initialScan.stop()
-    const countResponse = await elasticClient.count({ index })
-    expect(countResponse.count).toBe(numDocs)
   })
   test('should process records via change stream', async () => {
     const { coll, db, elasticClient } = await getConns()
@@ -95,18 +97,20 @@ describe.sequential('syncCollection', () => {
 
     const changeStream = await sync.processChangeStream()
     changeStream.start()
+    // Give change stream time to connect.
     await setTimeout(ms('1s'))
     const date = new Date()
     // Update records
     coll.updateMany({}, { $set: { createdAt: date } })
-    // Wait for the change stream events to be processed
-    await setTimeout(ms('2s'))
-    const countResponse = await elasticClient.count({
-      index,
-      query: { range: { createdAt: { gte: date } } },
-    })
+    // Test that all of the records are eventually synced.
+    await assertEventually(async () => {
+      const countResponse = await elasticClient.count({
+        index,
+        query: { range: { createdAt: { gte: date } } },
+      })
+      return countResponse.count == numDocs
+    }, `Less than ${numDocs} records were processed`)
     // Stop
     await changeStream.stop()
-    expect(countResponse.count).toBe(numDocs)
   })
 })

src/syncData.test.ts

@@ -4,6 +4,6 @@ import { defineConfig } from 'vitest/config'
 export default defineConfig(({ mode }) => ({
   test: {
     env: loadEnv(mode, process.cwd(), ''),
-    testTimeout: 30000,
+    testTimeout: 60000,
   },
 }))