From 490dc274426f033319641594af5fb5f46dc7cdb0 Mon Sep 17 00:00:00 2001 From: Sean Derrow Date: Sun, 4 Oct 2026 22:07:15 -0400 Subject: [PATCH] test: characterize the Redis connection layer before v5 Pin today's behavior in everything the v5 client-injection change touches. Test-only; no src changes. Passes on ioredis 5/6 and redis 4/5/6. Coverage audit (existing coverage kept as-is): - Lua scripts: all 15 exercised end to end on both datastores. - Error mapping (OVERWEIGHT, SETTINGS_KEY_NOT_FOUND, UNKNOWN_CLIENT): covered. - Pub/sub capacity / capacity-priority drains, publish/message: covered. - Connection sharing, deleteKey / clusterKeys / auto-cleanup: covered. - RESP2/RESP3 normalization, heartbeat / orphaned-jobs, clearDatastore: covered. Gaps filled. Preserve through v5: - SCRIPT FLUSH mid-run: both datastores recover today. - A disconnected limiter stops receiving messages while siblings on its connection keep receiving and scheduling. - Pub/sub-driven drains run outside the constructing caller's AsyncLocalStorage context. Fails if RedisDatastore stops detaching connection setup. - Premade ioredis Cluster client: the old test never passed its connection to the limiter. It now does and asserts a separate Cluster subscriber. Gaps filled. Intentional v5 changes, edited when the behavior changes: - limiter.disconnect() / group.disconnect() close connections they built; a Group leaves a shared connection open. - Errors from a Group-built connection reach group.on("error"). - A connection connects (node-redis), adds an error listener to, and sets maxListeners on a passed-in client. ioredis also defines script methods on it. disconnect() strips the consumer's error listeners and closes the client. makeConnection() with no options now builds a default connection for the current datastore, so test bodies needn't change when construction does. --- test/async-context.test.ts | 35 +++++++++ test/cluster-connection.test.ts | 126 ++++++++++++++++++++++++++++++++ test/helpers/limiter.ts | 10 +++ test/helpers/test-api.ts | 6 +- test/ioredis.test.ts | 65 ++++++++++++---- test/node_redis.test.ts | 37 +++++++++- 6 files changed, 261 insertions(+), 18 deletions(-) create mode 100644 test/cluster-connection.test.ts diff --git a/test/async-context.test.ts b/test/async-context.test.ts index 184d675..6315e96 100644 --- a/test/async-context.test.ts +++ b/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(); + // 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(); diff --git a/test/cluster-connection.test.ts b/test/cluster-connection.test.ts new file mode 100644 index 0000000..c6ba0d3 --- /dev/null +++ b/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(); + }); + }); +}); diff --git a/test/helpers/limiter.ts b/test/helpers/limiter.ts index 0fe0757..18bce0c 100644 --- a/test/helpers/limiter.ts +++ b/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 { + return process.env.DATASTORE === "ioredis" + ? { Redis: IORedis, clientOptions: buildClientOptions("ioredis") } + : { Redis: RedisClient, clientOptions: buildClientOptions("redis") }; +} diff --git a/test/helpers/test-api.ts b/test/helpers/test-api.ts index 7358caf..787b85f 100644 --- a/test/helpers/test-api.ts +++ b/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) { diff --git a/test/ioredis.test.ts b/test/ioredis.test.ts index 3110d61..6c41872 100644 --- a/test/ioredis.test.ts +++ b/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")); + }); +}); diff --git a/test/node_redis.test.ts b/test/node_redis.test.ts index cf7dc77..2d9e862 100644 --- a/test/node_redis.test.ts +++ b/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); + }); +});