perf(redis): pipeline listSchedules to avoid N+1 queries

Author: RomainLanz

src/drivers/redis_adapter.ts

@@ -757,17 +757,36 @@ export class RedisAdapter implements Adapter {
 
   async listSchedules(options?: ScheduleListOptions): Promise<ScheduleData[]> {
     const ids = await this.#connection.smembers(schedulesIndexKey)
-    const schedules: ScheduleData[] = []
+    if (ids.length === 0) {
+      return []
+    }
+
+    const pipeline = this.#connection.pipeline()
 
     for (const id of ids) {
-      const schedule = await this.getSchedule(id)
-      if (schedule) {
-        // Filter by status if provided
-        if (options?.status && schedule.status !== options.status) {
-          continue
-        }
-        schedules.push(schedule)
+      pipeline.hgetall(`${schedulesKey}::${id}`)
+    }
+
+    const results = await pipeline.exec()
+    if (!results) {
+      return []
+    }
+
+    const schedules: ScheduleData[] = []
+
+    for (const [, data] of results) {
+      if (!data || Object.keys(data).length === 0) {
+        continue
+      }
+
+      const schedule = this.#hashToScheduleData(data as Record<string, string>)
+
+      // Filter by status if provided
+      if (options?.status && schedule.status !== options.status) {
+        continue
       }
+
+      schedules.push(schedule)
     }
 
     return schedules

tests/adapter.spec.ts

@@ -49,6 +49,43 @@ test.group('Adapter | Redis', (group) => {
     test,
     createAdapter: () => new RedisAdapter(connection),
   })
+
+  test('listSchedules should use bounded network round-trips as schedule count grows', async ({
+    assert,
+  }) => {
+    const adapter = new RedisAdapter(connection)
+
+    for (let i = 0; i < 50; i++) {
+      await adapter.upsertSchedule({
+        id: `perf-schedule-${i}`,
+        name: 'PerfJob',
+        payload: { i },
+        everyMs: 60_000,
+        timezone: 'UTC',
+      })
+    }
+
+    const stream = (connection as any).stream as { write: (...args: any[]) => any }
+    const originalWrite = stream.write.bind(stream)
+    let writes = 0
+
+    stream.write = ((...args: any[]) => {
+      writes++
+      return originalWrite(...args)
+    }) as typeof stream.write
+
+    try {
+      const schedules = await adapter.listSchedules()
+      assert.lengthOf(schedules, 50)
+      assert.isAtMost(
+        writes,
+        4,
+        `Expected bounded write count with pipelining, got ${writes} writes for 50 schedules`
+      )
+    } finally {
+      stream.write = originalWrite
+    }
+  })
 })
 
 test.group('Adapter | Knex (SQLite)', (group) => {