test: characterize the Redis connection layer before v5

test/async-context.test.ts

@@ -94,6 +94,41 @@ describe("Async context", () => {
     },
   );
 
+  test.runIf(process.env.DATASTORE != null)(
+    "runs pub/sub-driven drains outside the async context that constructed the limiter",
+    async ({ makeLimiter }) => {
+      const als = new AsyncLocalStorage<string>();
+      // Long heartbeats so the only thing that can drain `waiter` is the
+      // capacity message `holder` publishes when its job frees up.
+      const options = { id: "als-pubsub", maxConcurrent: 1, heartbeatInterval: 60_000 };
+      const holder = makeLimiter(options);
+      // Constructing inside a context must not leak it into the limiter's
+      // pub/sub socket: a drain triggered by a message on that socket would
+      // otherwise run (and report its Redis calls) inside the constructor's
+      // context for the limiter's whole lifetime.
+      const waiter = als.run("constructor-store", () => makeLimiter(options));
+      await Promise.all([holder.ready(), waiter.ready()]);
+
+      const hold = deferred();
+      const held = holder.schedule(() => hold.signal);
+      await enqueued(holder);
+      const waiting = waiter.schedule(() => Promise.resolve("ran"));
+      await enqueued(waiter);
+      expect(waiter.counts().QUEUED).toBe(1);
+
+      const registerStore = new Promise((resolve) => {
+        waiter.on("debug", (message: string) => {
+          if (message.includes("register.lua")) resolve(als.getStore());
+        });
+      });
+      hold.release();
+
+      await expect(registerStore).resolves.toBeUndefined();
+      await expect(waiting).resolves.toBe("ran");
+      await held;
+    },
+  );
+
   test("runDetached and setDetachedTimeout run outside the caller's context", async () => {
     const als = new AsyncLocalStorage<string>();
 

test/cluster-connection.test.ts

@@ -0,0 +1,126 @@
+import { describe, expect } from "vitest";
+import { defined } from "./helpers/defined";
+import { runCommand } from "./helpers/store";
+import { test, waitForState } from "./helpers/test-api";
+
+// Characterization tests for the Redis connection layer: how limiters and
+// Groups use, share, and tear down connections. Tests under "Ownership" pin
+// behavior that v5 intentionally changes (who closes what); the rest must keep
+// passing unchanged.
+
+// node-redis v4/v5 shape fails fast; ioredis stays flat.
+const failingClientOptions = () =>
+  process.env.DATASTORE === "redis"
+    ? { socket: { port: 1, reconnectStrategy: () => false } }
+    : { port: 1, retryStrategy: () => null };
+
+/** Whether a raw ioredis / node-redis client still has an open socket. */
+function isOpen(client: unknown): boolean {
+  if (typeof client !== "object" || client == null) throw new Error("Not a client");
+  if ("status" in client) return client.status !== "end"; // ioredis
+  if ("isOpen" in client) return client.isOpen === true; // node-redis
+  throw new Error("Unknown client");
+}
+
+describe("Cluster connection", () => {
+  test("Should keep running scripts after SCRIPT FLUSH", async ({ makeLimiter }) => {
+    const limiter = makeLimiter({ maxConcurrent: 1 });
+    await limiter.ready();
+    await expect(limiter.schedule(() => Promise.resolve(1))).resolves.toBe(1);
+
+    await runCommand(limiter, "script", ["flush"]);
+
+    await expect(limiter.schedule(() => Promise.resolve(2))).resolves.toBe(2);
+    expect(await limiter.running()).toBe(0);
+  });
+
+  test("Should stop delivering to a disconnected limiter while its connection's other limiters keep working", async ({
+    makeLimiter,
+    makeConnection,
+  }) => {
+    const connection = makeConnection();
+    const a = makeLimiter({ id: "shared-a", connection });
+    const b = makeLimiter({ id: "shared-b", connection });
+    // Both publishers share one connection and publish in order, and a and b
+    // share one subscriber, so Redis delivers in publish order: once b has its
+    // message, a would already have had its own.
+    const publishers = makeConnection();
+    const pubA = makeLimiter({ id: "shared-a", connection: publishers });
+    const pubB = makeLimiter({ id: "shared-b", connection: publishers });
+    await Promise.all([a.ready(), b.ready(), pubA.ready(), pubB.ready()]);
+
+    const received: string[] = [];
+    a.on("message", (message) => received.push(`a:${message}`));
+    b.on("message", (message) => received.push(`b:${message}`));
+
+    await a.disconnect();
+    await pubA.publish("1");
+    await pubB.publish("2");
+
+    await waitForState(() => expect(received).toContain("b:2"));
+    expect(received).toEqual(["b:2"]);
+    await expect(b.schedule(() => Promise.resolve("still works"))).resolves.toBe("still works");
+  });
+
+  describe("Ownership", () => {
+    test("Should close a connection the limiter built when it disconnects", async ({
+      makeLimiter,
+    }) => {
+      const limiter = makeLimiter();
+      await limiter.ready();
+      const { client, subscriber } = defined(limiter.connection);
+      expect([isOpen(client), isOpen(subscriber)]).toEqual([true, true]);
+
+      await limiter.disconnect();
+
+      await waitForState(() =>
+        expect([isOpen(client), isOpen(subscriber)]).toEqual([false, false]),
+      );
+    });
+
+    test("Should close a connection the Group built when it disconnects", async ({ makeGroup }) => {
+      const group = makeGroup({ datastore: process.env.DATASTORE });
+      const connection = defined(group.connection);
+      await connection.ready;
+      const { client, subscriber } = connection;
+
+      await group.disconnect();
+
+      await waitForState(() =>
+        expect([isOpen(client), isOpen(subscriber)]).toEqual([false, false]),
+      );
+    });
+
+    test("Should leave a shared connection open when a Group disconnects", async ({
+      makeGroup,
+      makeLimiter,
+      makeConnection,
+    }) => {
+      const connection = makeConnection();
+      const group = makeGroup({ connection });
+      const sibling = makeLimiter({ connection });
+      await Promise.all([group.key("a").ready(), sibling.ready()]);
+
+      await group.disconnect();
+
+      expect([isOpen(connection.client), isOpen(connection.subscriber)]).toEqual([true, true]);
+      await expect(sibling.schedule(() => Promise.resolve("ok"))).resolves.toBe("ok");
+    });
+
+    test("Should send errors from a connection the Group built to the Group", async ({
+      makeGroup,
+    }) => {
+      const group = makeGroup({
+        datastore: process.env.DATASTORE,
+        clientOptions: failingClientOptions(),
+      });
+      defined(group.connection).ready.catch(() => {});
+
+      await expect(
+        new Promise((resolve) => {
+          group.on("error", resolve);
+        }),
+      ).resolves.toBeTruthy();
+    });
+  });
+});

test/helpers/limiter.ts

@@ -66,3 +66,13 @@ function makeLimiter(
 
 export default makeLimiter;
 export { makeLimiter, buildClientOptions };
+
+/**
+ * Options for a connection to the test Redis for the current datastore, so
+ * tests can build one without caring how connections are constructed.
+ */
+export function defaultConnectionOptions(): Record<string, unknown> {
+  return process.env.DATASTORE === "ioredis"
+    ? { Redis: IORedis, clientOptions: buildClientOptions("ioredis") }
+    : { Redis: RedisClient, clientOptions: buildClientOptions("redis") };
+}

test/helpers/test-api.ts

@@ -4,7 +4,7 @@ import type { BottleneckOptions, JobOptions } from "../../src/types";
 import Bottleneck from "../bottleneck";
 import { isFakeClock } from "./clock";
 import { createTaskFns } from "./job-tasks";
-import makeLimiterHelper from "./limiter";
+import makeLimiterHelper, { defaultConnectionOptions } from "./limiter";
 import { wrongType } from "./wrong-type";
 
 export { waitForState } from "./wait-for-state";
@@ -240,7 +240,9 @@ export const test = baseTest.extend<{
         process.env.DATASTORE === "ioredis"
           ? Bottleneck.IORedisConnection
           : Bottleneck.RedisConnection;
-      return track(new Connection(wrongType(opts)) as unknown as Connection);
+      return track(
+        new Connection(wrongType(opts ?? defaultConnectionOptions())) as unknown as Connection,
+      );
     });
   },
   async limiter({ makeLimiter, limiterOptions, limiterMeta }, use) {

test/ioredis.test.ts

@@ -1,8 +1,8 @@
 import Redis from "ioredis";
 import ioredisPkg from "ioredis/package.json" with { type: "json" };
-import { describe, expect } from "vitest";
+import { describe, expect, vi } from "vitest";
 import { redisStore } from "./helpers/store";
-import { test } from "./helpers/test-api";
+import { test, waitForState } from "./helpers/test-api";
 import buildClientOptions from "./redis-client-options";
 
 describe("ioredis-only", () => {
@@ -42,21 +42,22 @@ describe("ioredis-only", () => {
     makeLimiter,
     makeConnection,
   }) => {
-    const client = new Redis.Cluster([]);
-    makeConnection({ client });
-    const limiter = makeLimiter({
-      maxConcurrent: 2,
-      clientOptions: {},
-      clusterNodes: [
-        {
-          host: process.env.REDIS_HOST,
-          port: process.env.REDIS_PORT,
-        },
-      ],
-    });
+    // The test server isn't a Redis Cluster, so this only checks wiring: the
+    // client never connects, its errors are expected, and with no offline
+    // queue its commands fail fast instead of hanging teardown.
+    const client = new Redis.Cluster(
+      [{ host: process.env.REDIS_HOST, port: Number(process.env.REDIS_PORT) }],
+      { enableOfflineQueue: false },
+    );
+    const connection = makeConnection({ client });
+    const limiter = makeLimiter({ maxConcurrent: 2, connection }, { expectErrors: true });
 
     expect(limiter.datastore).toStrictEqual("ioredis");
-    expect(redisStore(limiter).connection.client).toBeInstanceOf(Redis.Cluster);
+    expect(limiter.connection).toBe(connection);
+    expect(connection.client).toBe(client);
+    // The subscriber is a second Cluster client, not the premade one.
+    expect(connection.subscriber).toBeInstanceOf(Redis.Cluster);
+    expect(connection.subscriber).not.toBe(client);
   });
 
   test("Should accept existing connections", async ({
@@ -263,3 +264,37 @@ describeResp3("ioredis RESP3", () => {
     expect(await group.deleteKey("AAA")).toStrictEqual(false);
   });
 });
+
+// Pins what a connection does to a client passed in to it. v5 stops all of
+// this (the client belongs to the consumer); these tests change with it.
+describe("ioredis passed-in client side effects", () => {
+  test("Should install listeners and scripts on the client, then strip and close it", async ({
+    makeConnection,
+  }) => {
+    const client = new Redis(buildClientOptions("ioredis"));
+    const consumerListener = vi.fn<() => void>();
+    client.on("error", consumerListener);
+
+    const connection = makeConnection({ client });
+    await connection.ready;
+    expect(client.listenerCount("error")).toBe(2);
+    expect(client.getMaxListeners()).toBe(0);
+    // defineCommand installs each Lua script as a client method.
+    expect(client).toHaveProperty("submit", expect.any(Function));
+
+    await connection.disconnect(true);
+    expect(client.listeners("error")).not.toContain(consumerListener);
+    expect(client.listenerCount("error")).toBe(1);
+    await waitForState(() => expect(client.status).toBe("end"));
+  });
+
+  test("Should disconnect a passed-in client on disconnect(false)", async ({ makeConnection }) => {
+    const client = new Redis(buildClientOptions("ioredis"));
+    const connection = makeConnection({ client });
+    await connection.ready;
+
+    await connection.disconnect(false);
+
+    await waitForState(() => expect(client.status).toBe("end"));
+  });
+});

test/node_redis.test.ts

@@ -1,5 +1,5 @@
 import * as Redis from "redis";
-import { describe, expect } from "vitest";
+import { describe, expect, vi } from "vitest";
 import { test } from "./helpers/test-api";
 import buildClientOptions from "./redis-client-options";
 
@@ -103,3 +103,38 @@ describe("node_redis-only", () => {
     });
   });
 });
+
+// Pins what a connection does to a client passed in to it. v5 stops all of
+// this (the client belongs to the consumer); these tests change with it.
+describe("node_redis passed-in client side effects", () => {
+  test("Should connect the client and install listeners, then strip and close it", async ({
+    makeConnection,
+  }) => {
+    const client = Redis.createClient(buildClientOptions("redis"));
+    const consumerListener = vi.fn<() => void>();
+    client.on("error", consumerListener);
+    expect(client.isOpen).toBe(false);
+
+    const connection = makeConnection({ client });
+    await connection.ready;
+    expect(client.isOpen).toBe(true);
+    expect(client.listenerCount("error")).toBe(2);
+    expect(client.getMaxListeners()).toBe(0);
+
+    await connection.disconnect(true);
+    expect(client.listeners("error")).not.toContain(consumerListener);
+    expect(client.listenerCount("error")).toBe(1);
+    expect(client.isOpen).toBe(false);
+  });
+
+  test("Should destroy a passed-in client on disconnect(false)", async ({ makeConnection }) => {
+    const client = Redis.createClient(buildClientOptions("redis"));
+    await client.connect();
+    const connection = makeConnection({ client });
+    await connection.ready;
+
+    await connection.disconnect(false);
+
+    expect(client.isOpen).toBe(false);
+  });
+});