Merge pull request #3513 from OpenNeuroOrg/immediate-indexing

Add Redis queue for search reindexing

Author: nellh

.pnp.cjs

@@ -20,7 +20,7 @@
     "@apollo/utils.keyvadapter": "3.0.0",
     "@elastic/elasticsearch": "8.13.1",
     "@graphql-tools/schema": "^10.0.0",
-    "@keyv/redis": "^2.7.0",
+    "@keyv/redis": "^4.5.0",
     "@openneuro/search": "^4.36.2",
     "@sentry/node": "^8.25.0",
     "@sentry/profiling-node": "^8.25.0",
@@ -37,10 +37,10 @@
     "graphql-iso-date": "^3.6.1",
     "graphql-tools": "9.0.0",
     "immutable": "^3.8.2",
-    "ioredis": "4.17.3",
+    "ioredis": "^5.6.1",
     "jsdom": "24.0.0",
     "jsonwebtoken": "^9.0.0",
-    "keyv": "^4.5.3",
+    "keyv": "^5.3.4",
     "mime-types": "^2.1.19",
     "mongodb-memory-server": "^9.2.0",
     "mongoose": "^8.9.5",
@@ -55,6 +55,8 @@
     "passport-orcid": "0.0.4",
     "react": "^18.2.0",
     "react-dom": "^18.2.0",
+    "redis-smq": "^8.3.1",
+    "redis-smq-common": "^8.3.1",
     "redlock": "^4.0.0",
     "request": "^2.83.0",
     "semver": "^5.5.0",

packages/openneuro-server/package.json

@@ -4,6 +4,10 @@ const config = {
       secret: "123456",
     },
   },
+  redis: {
+    port: 6379,
+    host: "localhost",
+  },
   datalad: {
     uri: "datalad",
     workers: 4,

packages/openneuro-server/src/__mocks__/config.ts

@@ -65,6 +65,7 @@ export async function expressApolloSetup() {
     schema,
     // Always allow introspection - our schema is public
     introspection: true,
+    // @ts-expect-error Type mismatch for keyv and ioredis recent releases
     cache: new KeyvAdapter(new Keyv({ store: new KeyvRedis(redis) })),
     plugins: [
       ApolloServerPluginLandingPageLocalDefault(),

packages/openneuro-server/src/app.ts

@@ -3,7 +3,6 @@
  */
 import * as Sentry from "@sentry/node"
 import request from "superagent"
-import { reindexDataset } from "../elasticsearch/reindex-dataset"
 import { redis, redlock } from "../libs/redis"
 import CacheItem, { CacheType } from "../cache/item"
 import config from "../config"
@@ -23,6 +22,7 @@ import { updateDatasetRevision } from "./draft"
 import { getDatasetWorker } from "../libs/datalad-service"
 import { join } from "path"
 import { createEvent, updateEvent } from "../libs/events"
+import { queueIndexDataset } from "../queues/producer-methods"
 
 const lockSnapshot = (datasetId, tag) => {
   return redlock.lock(
@@ -177,7 +177,7 @@ export const createSnapshot = async (
     await updateEvent(event)
 
     // Immediate indexing for new snapshots
-    await reindexDataset(datasetId)
+    queueIndexDataset(datasetId)
 
     announceNewSnapshot(snapshot, datasetId, user)
     return snapshot

packages/openneuro-server/src/datalad/snapshots.ts

@@ -5,6 +5,7 @@ import Dataset from "../../models/dataset"
 import Permission from "../../models/permission"
 import Comment from "../../models/comment"
 import Deletion from "../../models/deletion"
+import { queueIndexDataset } from "../../queues/producer-methods"
 import * as Sentry from "@sentry/node"
 
 /**
@@ -20,6 +21,7 @@ import * as Sentry from "@sentry/node"
 export async function userMigration(orcid: string, userId: string) {
   const session = await mongoose.startSession()
   try {
+    const updateDatasets: Record<string, boolean> = {}
     await session.withTransaction(async () => {
       try {
         // Load both original records
@@ -55,6 +57,7 @@ export async function userMigration(orcid: string, userId: string) {
           // Record this dataset uploader as migrated
           migration.datasets.push(dataset.id)
           await dataset.save({ session })
+          updateDatasets[dataset.id] = true
         }
 
         // Migrate dataset permissions
@@ -70,6 +73,7 @@ export async function userMigration(orcid: string, userId: string) {
           // Record this permission as migrated
           migration.permissions.push(permission.toObject())
           await permission.save({ session })
+          updateDatasets[permission.datasetId] = true
         }
 
         // Migrate dataset deletions
@@ -110,6 +114,10 @@ export async function userMigration(orcid: string, userId: string) {
         // Save success
         migration.success = true
         await migration.save({ session })
+        // Request reindexing
+        for (const updateDataset of Object.keys(updateDatasets)) {
+          queueIndexDataset(updateDataset)
+        }
       } catch (err) {
         Sentry.captureException(err)
         throw err

packages/openneuro-server/src/libs/authentication/user-migration.ts

@@ -3,28 +3,7 @@
 // dependencies --------------------------------------------------
 import Redis from "ioredis"
 import Redlock from "redlock"
+import config from "../config"
 
-let redis = null
-let redlock = null
-
-const connect = async (config) => {
-  return new Promise((resolve) => {
-    if (!redis) {
-      console.log(
-        'Connecting to Redis "redis://%s:%d/0"',
-        config.host,
-        config.port,
-      )
-      redis = new Redis(config)
-      redlock = new Redlock([redis])
-      redis.on("connect", () => {
-        resolve(redis)
-      })
-    } else {
-      resolve(redis)
-    }
-  })
-}
-
-export default { connect }
-export { connect, redis, redlock }
+export const redis = new Redis(config.redis)
+export const redlock = new Redlock([redis])

packages/openneuro-server/src/libs/redis.ts

@@ -0,0 +1,18 @@
+import type { Consumer } from "redis-smq"
+import { reindexDataset } from "../elasticsearch/reindex-dataset"
+import { OpenNeuroQueues } from "./queues"
+import * as Sentry from "@sentry/node"
+
+export function startConsumer(consumer: Consumer) {
+  const reindexMessageHandler = async (msg, cb) => {
+    // Index one dataset
+    reindexDataset(msg.body.datasetId).then(cb)
+  }
+
+  consumer.consume(OpenNeuroQueues.INDEXING, reindexMessageHandler, (err) => {
+    if (err) {
+      Sentry.captureException(err)
+    }
+  })
+  return consumer
+}

packages/openneuro-server/src/queues/consumer.ts

@@ -0,0 +1,18 @@
+import { ProducibleMessage } from "redis-smq"
+import { producer } from "./setup"
+import { OpenNeuroQueues } from "./queues"
+import * as Sentry from "@sentry/node"
+
+/**
+ * Queue search indexing for a dataset
+ * @param datasetId Dataset to index
+ */
+export function queueIndexDataset(datasetId: string) {
+  const msg = new ProducibleMessage()
+  msg.setQueue(OpenNeuroQueues.INDEXING).setBody({ datasetId })
+  producer.produce(msg, (err) => {
+    if (err) {
+      Sentry.captureException(err)
+    }
+  })
+}

packages/openneuro-server/src/queues/producer-methods.ts

@@ -0,0 +1,34 @@
+import { Queue } from "redis-smq"
+import { EQueueDeliveryModel, EQueueType, QueueRateLimit } from "redis-smq"
+import * as Sentry from "@sentry/node"
+
+export enum OpenNeuroQueues {
+  INDEXING = "elasticsearch_indexing",
+}
+
+export function setupQueues() {
+  const indexingQueue = new Queue()
+  indexingQueue.save(
+    OpenNeuroQueues.INDEXING,
+    EQueueType.FIFO_QUEUE,
+    EQueueDeliveryModel.POINT_TO_POINT,
+    (err) => {
+      // The queue may already exist, don't log that error
+      if (err !== "QueueQueueExistsError") {
+        Sentry.captureException(err)
+      }
+    },
+  )
+
+  // Limit indexing queue to 8 runs per minute to avoid stacking indexing excessively
+  const queueRateLimit = new QueueRateLimit()
+  queueRateLimit.set(
+    OpenNeuroQueues.INDEXING,
+    { limit: 8, interval: 60000 },
+    (err) => {
+      if (err) {
+        Sentry.captureException(err)
+      }
+    },
+  )
+}