diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 909d491f..0bc1128b 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -116,6 +116,12 @@ jobs: - name: Pin client version run: pnpm add -D ${{ matrix.client }}@${{ matrix.version }} + # Connections are generic over the consumer's client and constrain it + # with structural types (src/cluster/redis-types.ts): every supported + # client major must still type-check against them. + - name: Type check against the pinned client + run: pnpm exec tsc + - name: Run cluster tests run: pnpm exec vitest run --project ${{ matrix.project }} diff --git a/README.md b/README.md index e8aa7e64..493470a5 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,17 @@ The original [bottleneck library](https://github.com/SGrondin/bottleneck) appear More importantly, this library has been rewritten with modern-day JS (courtesy of [decaffeinate](https://www.npmjs.com/package/decaffeinate)) since CoffeeScript isn't necessary anymore. Much of the packages and the overall toolchain has been upgraded as well. There is still more work to do of course, such as writing the whole thing natively in TypeScript. +### Breaking changes in v5 + +- Redis-backed limiters and Groups take a `connection` built from your own client: `new Bottleneck({ connection: new IORedisConnection({ client }) })`. The `datastore`, `Redis`, `client`, `clientOptions` and `clusterNodes` options are gone, and passing them throws a `LEGACY_REDIS_OPTIONS` error. `datastore: "local"` is still accepted. +- Bottleneck never modifies or closes your client. It only sends commands on it: no listeners, no `setMaxListeners`, no script methods defined on it, no `connect()`. node-redis clients must be `connect()`ed before you build a connection (`CLIENT_NOT_OPEN` otherwise). +- You close what you create. `limiter.disconnect()` and `group.disconnect()` no longer close the connection, and `connection.disconnect()` only closes the pub/sub subscriber it duplicated from your client. Close your client yourself. +- Errors from your client are yours to handle on the client. Errors from the connection's own subscriber still reach `connection.on("error")` and every limiter using the connection. +- `limiter.clients()` and the `ClientsList` type are removed: use `connection.client` / `connection.subscriber`, typed from the client you passed in (`IORedisConnection`, `RedisConnection`). `limiter.ready()` resolves to `undefined`. +- Lua scripts run by SHA with an `EVAL` fallback, so there's no `SCRIPT LOAD` at startup and limiters recover by themselves after `SCRIPT FLUSH` or a Redis restart. + +See [Upgrading to v5](#upgrading-to-v5) for migration steps. + ### Breaking changes in v4 - The callback-style `submit()` method has been removed — Bottleneck is now Promise-only. Use `schedule()`, wrapping callback-style functions with [`util.promisify`](https://nodejs.org/api/util.html#utilpromisifyoriginal). The `Bottleneck.Callback` type is gone from the typings. @@ -56,6 +67,7 @@ See [Upgrading to v4](#upgrading-to-v4) for migration steps. - [Upgrading To v2](#upgrading-to-v2) - [Upgrading To v3](#upgrading-to-v3) - [Upgrading To v4](#upgrading-to-v4) +- [Upgrading To v5](#upgrading-to-v5) - [Contributing](#contributing) @@ -814,7 +826,7 @@ Batching doesn't throttle requests, it only groups them up optimally according t ## Clustering -Clustering lets many limiters access the same shared state, stored in Redis. Changes to the state are Atomic, Consistent and Isolated (and fully [ACID](https://en.wikipedia.org/wiki/ACID) with the right [Durability](https://redis.io/topics/persistence) configuration), to eliminate any chances of race conditions or state corruption. Your settings, such as `maxConcurrent`, `minTime`, etc., are shared across the whole cluster, which means —for example— that `{ maxConcurrent: 5 }` guarantees no more than 5 jobs can ever run at a time in the entire cluster of limiters. 100% of Bottleneck's features are supported in Clustering mode. Enabling Clustering is as simple as changing a few settings. It's also a convenient way to store or export state for later use. +Clustering lets many limiters access the same shared state, stored in Redis. Changes to the state are Atomic, Consistent and Isolated (and fully [ACID](https://en.wikipedia.org/wiki/ACID) with the right [Durability](https://redis.io/topics/persistence) configuration), to eliminate any chances of race conditions or state corruption. Your settings, such as `maxConcurrent`, `minTime`, etc., are shared across the whole cluster, which means —for example— that `{ maxConcurrent: 5 }` guarantees no more than 5 jobs can ever run at a time in the entire cluster of limiters. 100% of Bottleneck's features are supported in Clustering mode. Enabling Clustering is as simple as passing a `connection`. It's also a convenient way to store or export state for later use. Bottleneck will attempt to spread load evenly across limiters. @@ -823,17 +835,21 @@ Bottleneck will attempt to spread load evenly across limiters. First, add `redis` or `ioredis` to your application's dependencies. Both are optional **peer dependencies** of Bottleneck — install whichever you intend to use: ```bash -# node-redis (https://github.com/redis/node-redis) — v4 or v5 -pnpm add redis@^4 # or: npm install redis@^4 / yarn add redis@^4 +# node-redis (https://github.com/redis/node-redis) — v4, v5 or v6 +pnpm add redis # or: npm install redis / yarn add redis -# or ioredis (https://github.com/redis/ioredis) — v5 -pnpm add ioredis@^5 +# or ioredis (https://github.com/redis/ioredis) — v5 or v6 +pnpm add ioredis ``` -Then import the client and pass it to Bottleneck via the `Redis` constructor option (Bottleneck does not implicitly require either client at runtime): +Then create your client, wrap it in a connection, and pass the connection to your limiters and Groups: ```js -import Redis from "ioredis"; // or: import { createClient } from "redis"; +import Redis from "ioredis"; +import Bottleneck, { IORedisConnection } from "@sderrow/bottleneck"; + +const client = new Redis({ host: "127.0.0.1", port: 6379 }); // your client, your options +const connection = new IORedisConnection({ client }); const limiter = new Bottleneck({ /* Some basic options */ @@ -842,28 +858,31 @@ const limiter = new Bottleneck({ id: "my-super-app", // All limiters with the same id will be clustered together /* Clustering options */ - datastore: "ioredis", // or "redis" - Redis, // required (unless you pass a pre-built `client` or `connection`) + connection, clearDatastore: false, - clientOptions: { - host: "127.0.0.1", - port: 6379, - - // Redis client options - // Using node-redis? See https://github.com/redis/node-redis#createclient-configuration - // Using ioredis? See https://github.com/redis/ioredis#connect-to-redis - }, }); ``` -| Option | Default | Description | -| ---------------- | --------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `datastore` | `"local"` | Where the limiter stores its internal state. The default (`"local"`) keeps the state in the limiter itself. Set it to `"redis"` or `"ioredis"` to enable Clustering. | -| `Redis` | `null` | The imported Redis library, e.g. `require("ioredis")` or `require("redis")`. **Required** when `datastore` is `"redis"` or `"ioredis"`, unless a pre-built `client` or `connection` is provided. Bottleneck does not implicitly require either client at runtime. | -| `clearDatastore` | `false` | When set to `true`, on initial startup, the limiter will wipe any existing Bottleneck state data on the Redis db. | -| `clientOptions` | `{}` | This object is passed directly to the redis client library you've selected. | -| `clusterNodes` | `null` | **ioredis only.** When `clusterNodes` is not null, the client will be instantiated by calling `new Redis.Cluster(clusterNodes, clientOptions)` instead of `new Redis(clientOptions)`. | -| `timeout` | `null` (no TTL) | The Redis TTL in milliseconds ([TTL](https://redis.io/commands/ttl)) for the keys created by the limiter. When `timeout` is set, the limiter's state will be automatically removed from Redis after `timeout` milliseconds of inactivity. | +With node-redis, connect the client first (no need to await it) and use `RedisConnection`: + +```js +import { createClient } from "redis"; +import Bottleneck, { RedisConnection } from "@sderrow/bottleneck"; + +const client = createClient({ url: "redis://127.0.0.1:6379" }); +client.on("error", (err) => { + /* your client's errors are yours to handle */ +}); +client.connect(); + +const limiter = new Bottleneck({ id: "my-super-app", connection: new RedisConnection({ client }) }); +``` + +| Option | Default | Description | +| ---------------- | --------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `connection` | `null` (local) | A `RedisConnection` or `IORedisConnection` built from your client. Without one, the limiter keeps its state in memory. See [Managing Redis Connections](#managing-redis-connections). | +| `clearDatastore` | `false` | When set to `true`, on initial startup, the limiter will wipe any existing Bottleneck state data on the Redis db. | +| `timeout` | `null` (no TTL) | The Redis TTL in milliseconds ([TTL](https://redis.io/commands/ttl)) for the keys created by the limiter. When `timeout` is set, the limiter's state will be automatically removed from Redis after `timeout` milliseconds of inactivity. | **Note: When using Groups**, the `timeout` option has a default of `300000` milliseconds and the generated limiters automatically receive an `id` with the pattern `${group.id}-${KEY}`. @@ -891,15 +910,15 @@ It is **strongly recommended** that you set an `expiration` (See [Job Options](# Network latency between Node.js and Redis is not taken into account when calculating timings (such as `minTime`). To minimize the impact of latency, Bottleneck only performs a single Redis call per [lifecycle transition](#jobs-lifecycle). Keeping the Redis server close to your limiters will help you get a more consistent experience. Keeping the system time consistent across all clients will also help. -It is **strongly recommended** to [set up an `"error"` listener](#events) on all your limiters and on your Groups. +It is **strongly recommended** to [set up an `"error"` listener](#events) on all your limiters and on your Groups, and an `"error"` listener on your Redis client itself: Bottleneck doesn't listen to your client. ### Clustering Methods -The `ready()`, `publish()` and `clients()` methods also exist when using the `local` datastore, for code compatibility reasons: code written for `redis`/`ioredis` won't break with `local`. +The `ready()` and `publish()` methods also exist on local limiters, for code compatibility reasons: code written for a Redis-backed limiter won't break without a `connection`. #### ready() -This method returns a promise that resolves once the limiter is connected to Redis. +This method returns a promise that resolves (to `undefined`) once the limiter is connected to Redis. As of v2.9.0, it's no longer necessary to wait for `.ready()` to resolve before issuing commands to a limiter. The commands will be queued until the limiter successfully connects. Make sure to listen to the `"error"` event to handle connection errors. @@ -939,84 +958,59 @@ limiter.on("message", (msg) => { limiter.publish(JSON.stringify({ hello: "world" })); ``` -#### clients() - -If you need direct access to the redis clients, use `.clients()`: - -```js -console.log(limiter.clients()); -// { client: , subscriber: } -``` - ### Additional Clustering information -- Bottleneck is compatible with [Redis Clusters](https://redis.io/topics/cluster-tutorial), but you must use the `ioredis` datastore and the `clusterNodes` option. -- Bottleneck is compatible with Redis Sentinel, but you must use the `ioredis` datastore. +- Bottleneck is compatible with [Redis Clusters](https://redis.io/topics/cluster-tutorial) through ioredis: pass a `new Redis.Cluster(nodes, options)` client to `IORedisConnection`. +- Bottleneck is compatible with Redis Sentinel through ioredis: configure Sentinel on your ioredis client. - Bottleneck's data is stored in Redis keys starting with `b_`. It also uses pubsub channels starting with `b_` It will not interfere with any other data stored on the server. -- Bottleneck loads a few Lua scripts on the Redis server using the `SCRIPT LOAD` command. These scripts only take up a few Kb of memory. Running the `SCRIPT FLUSH` command will cause any connected limiters to experience critical errors until a new limiter connects to Redis and loads the scripts again. +- Bottleneck runs a few Lua scripts by their SHA (`EVALSHA`), sending the full script with `EVAL` the first time a server (or a Cluster node) hasn't seen it. These scripts only take up a few Kb of memory. `SCRIPT FLUSH` and server restarts are safe: limiters reload the scripts on their next call. - The Lua scripts are highly optimized and designed to use as few resources as possible. ### Managing Redis Connections -Bottleneck needs to create 2 Redis Clients to function, one for normal operations and one for pubsub subscriptions. These 2 clients are kept in a `Bottleneck.RedisConnection` (NodeRedis) or a `Bottleneck.IORedisConnection` (ioredis) object, referred to as the Connection object. - -By default, every Group and every standalone limiter (a limiter not created by a Group) will create their own Connection object, but it is possible to manually control this behavior. In this example, every Group and limiter is sharing the same Connection object and therefore the same 2 clients: +Bottleneck needs 2 Redis clients: one for normal operations, and one for pub/sub subscriptions. A `RedisConnection` (node-redis) or `IORedisConnection` (ioredis) holds them: the client you pass in, and a subscriber it creates with `client.duplicate()`. Share one connection across as many limiters and Groups as you like: ```js -import Redis from "redis"; // or ioredis: import Redis from "ioredis"; - -// Use Bottleneck.IORedisConnection when using ioredis -const connection = new Bottleneck.RedisConnection({ - Redis, - clientOptions: {/* node-redis/ioredis options */}, - // Bottleneck.IORedisConnection also accepts `clusterNodes` here -}); +const connection = new IORedisConnection({ client }); -const limiter = new Bottleneck({ connection: connection }); -const group = new Bottleneck.Group({ connection: connection }); +const limiter = new Bottleneck({ connection }); +const group = new Bottleneck.Group({ connection }); +const other = new Bottleneck({ connection: limiter.connection }); ``` -You can access and reuse the Connection object of any Group or limiter: +`connection.client` and `connection.subscriber` are typed from the client you passed in (`IORedisConnection`, `IORedisConnection`, `RedisConnection>`). -```js -const group = new Bottleneck.Group({ connection: limiter.connection }); -``` +#### What Bottleneck does with your client + +It only sends commands on it. It doesn't add listeners to it, change its settings, define methods on it, connect it, or close it. With node-redis, call `client.connect()` before building the connection (awaiting it is optional: commands queue until the client is ready). -When a Connection object is created manually, the connectivity `"error"` events are emitted on the Connection itself. +The subscriber the connection duplicates is Bottleneck's own: it connects it, listens to it, and closes it. To use a subscriber you manage yourself instead, pass it in. It must be dedicated to Bottleneck, since a client in subscriber mode can't run other commands, and Bottleneck leaves it open: ```js -connection.on("error", (err) => { - /* handle connectivity errors here */ -}); +const connection = new IORedisConnection({ client, subscriber: client.duplicate() }); ``` -If you already have a node-redis/ioredis client, you can ask Bottleneck to reuse it, although currently the Connection object will still create a second client for pubsub operations: +#### Errors -```js -import { createClient } from "redis"; -const client = createClient({/* options */}); -await client.connect(); +Errors on your client are yours to handle, with your own `client.on("error", ...)` listener. Commands that fail because of them reject like any other failure, so they still reach your jobs and your limiters' `"error"` events. Errors on the connection's own subscriber are emitted on the connection and on every limiter using it: -const connection = new Bottleneck.RedisConnection({ - // `clientOptions` and `clusterNodes` will be ignored since we're passing a raw client - client: client, +```js +connection.on("error", (err) => { + /* handle pub/sub connectivity errors here */ }); - -const limiter = new Bottleneck({ connection: connection }); -const group = new Bottleneck.Group({ connection: connection }); ``` -Depending on your application, using more clients can improve performance. +#### Disconnecting -Use the `disconnect(flush)` method to close the Redis clients. +You close what you create. `limiter.disconnect()` only removes the limiter from its connection, and `group.disconnect()` disconnects the Group's limiters. Neither closes the connection, which other limiters may be sharing. `connection.disconnect()` stops the connection's limiters and closes the subscriber it duplicated. Your client stays open: ```js -limiter.disconnect(); -group.disconnect(); +await limiter.disconnect(); +await group.disconnect(); +await connection.disconnect(); // closes the subscriber Bottleneck created +await client.quit(); // yours ``` -If you created the Connection object manually, you need to call `connection.disconnect()` instead, for safety reasons. - ## Debugging your application Debugging complex scheduling logic can be difficult, especially when priorities, weights, and network latency all interact with one another. @@ -1057,16 +1051,18 @@ try { } ``` -| `code` | Meaning | -| ------------------- | ------------------------------------------------------------------- | -| `DROPPED` | The job was shed by the queue strategy, or dropped by `stop()`. | -| `EXPIRED` | The job ran longer than its `expiration`. | -| `STOPPED` | The job was submitted after `stop()`, or `stop()` was called twice. | -| `DUPLICATE_JOB_ID` | A job with the same `id` already exists. | -| `OVERWEIGHT` | The job's `weight` exceeds the limiter's `maxConcurrent`. | -| `INVALID_DATASTORE` | Unknown `datastore` value. | -| `INVALID_ARGUMENTS` | The constructor received a non-object argument (v1-style usage). | -| `MISSING_CLIENT` | Cluster mode without a `Redis` library or pre-built `client`. | +| `code` | Meaning | +| ---------------------- | ----------------------------------------------------------------------------------------------- | +| `DROPPED` | The job was shed by the queue strategy, or dropped by `stop()`. | +| `EXPIRED` | The job ran longer than its `expiration`. | +| `STOPPED` | The job was submitted after `stop()`, or `stop()` was called twice. | +| `DUPLICATE_JOB_ID` | A job with the same `id` already exists. | +| `OVERWEIGHT` | The job's `weight` exceeds the limiter's `maxConcurrent`. | +| `INVALID_DATASTORE` | Unknown `datastore` value (only `"local"` is accepted). | +| `INVALID_ARGUMENTS` | The constructor received a non-object argument (v1-style usage). | +| `MISSING_CLIENT` | A connection was built without a `client`. | +| `CLIENT_NOT_OPEN` | A node-redis client was passed in before `client.connect()`. | +| `LEGACY_REDIS_OPTIONS` | A removed pre-v5 Redis option (`datastore: "redis"`, `Redis`, `clientOptions`, ...) was passed. | ## Upgrading to v2 @@ -1138,7 +1134,7 @@ Pass the imported library to Bottleneck explicitly. Bottleneck no longer require }); ``` -The same applies to `Bottleneck.Group` and to the standalone `Bottleneck.RedisConnection` / `Bottleneck.IORedisConnection` constructors. Pass `Redis` once at the top of your module and reuse the resulting Connection or Group across as many limiters as you want — see [Managing Redis Connections](#managing-redis-connections). +(v5 replaces these options with a `connection` built from your own client: see [Upgrading to v5](#upgrading-to-v5).) The same applies to `Bottleneck.Group` and to the standalone `Bottleneck.RedisConnection` / `Bottleneck.IORedisConnection` constructors. Pass `Redis` once at the top of your module and reuse the resulting Connection or Group across as many limiters as you want — see [Managing Redis Connections](#managing-redis-connections). If you previously hit "Bottleneck failed to require ioredis at runtime", that workaround paragraph is no longer needed. The implicit-require hack has been removed entirely. @@ -1188,6 +1184,66 @@ The minimum supported `redis` package version is now v4. The v2/v3 callback-styl The `bottleneck/es5` import path has been removed. If you still need a build that runs in older browsers, use the ESM `@sderrow/bottleneck/light` build (which excludes cluster mode) and transpile in your own toolchain. +## Upgrading to v5 + +The v5 release makes Redis connections explicit: you bring the client, and Bottleneck never modifies or closes it. If your limiters don't use Redis, no changes are required. + +### Build a connection from your client + +Instead of handing Bottleneck a Redis library and options, create the client yourself and pass a connection: + +```diff + import Redis from "ioredis"; ++ import Bottleneck, { IORedisConnection } from "@sderrow/bottleneck"; ++ ++ const client = new Redis({ host: "127.0.0.1", port: 6379 }); ++ const connection = new IORedisConnection({ client }); + + const limiter = new Bottleneck({ + id: "my-app", +- datastore: "ioredis", +- Redis, +- clientOptions: { host: "127.0.0.1", port: 6379 }, ++ connection, + }); +``` + +- **node-redis:** `createClient(clientOptions)`, call `client.connect()`, and use `RedisConnection`. +- **`clusterNodes`:** build `new Redis.Cluster(clusterNodes, clientOptions)` and pass it as the client. +- **Groups:** pass the same `connection` option. A Group no longer creates a connection of its own. +- **A pre-built `client` or `connection`:** if you already passed either one, only the `connection.disconnect()` behavior below changes for you. + +### Close what you create + +`limiter.disconnect()` and `group.disconnect()` used to close a connection the limiter or Group had created. Now they only detach from the connection, so close it, and then your client, at shutdown: + +```js +await limiter.disconnect(); +await connection.disconnect(); +await client.quit(); +``` + +`connection.disconnect()` used to close a client you passed in. Now it only closes the subscriber it duplicated. + +### Handle your client's errors + +Bottleneck no longer listens for errors on your client, so `limiter.on("error")` doesn't see them. Listen on the client yourself (node-redis throws on an unhandled `"error"` event). Failed commands still reject, and errors on the connection's own subscriber still reach `connection.on("error")` and its limiters. + +### `clients()` and `ready()` + +`limiter.clients()` is removed. Use the connection, whose `client` and `subscriber` are typed from what you passed in: + +```diff +- const { client } = limiter.clients(); ++ const { client } = connection; +``` + +`limiter.ready()` resolves to `undefined` instead of the clients. + +### TypeScript users + +`BottleneckOptions` no longer has `datastore`, `Redis`, `client`, `clientOptions` or `clusterNodes`, and `connection` is typed as `RedisConnection | IORedisConnection`. `RedisConnectionOptions` and `IORedisConnectionOptions` take `{ client, subscriber? }`. `ClientsList` is removed. `NodeRedisClient` and `IORedisClient` describe the clients Bottleneck accepts. + ## Contributing This README is always in need of improvements. If wording can be clearer and simpler, please consider forking this repo and submitting a Pull Request, or simply opening an issue. diff --git a/scripts/test-ci.mts b/scripts/test-ci.mts index c53e38ef..3c62293a 100644 --- a/scripts/test-ci.mts +++ b/scripts/test-ci.mts @@ -172,6 +172,7 @@ if (selectedLegs.has("clients")) { cmd: ["pnpm", "add", "-D", `${pin.client}@${pin.version}`], pin, }, + { title: "tsc", cmd: ["pnpm", "exec", "tsc"] }, { title: `vitest run --project ${pin.project}`, cmd: ["pnpm", "exec", "vitest", "run", "--project", pin.project], diff --git a/src/Bottleneck.ts b/src/Bottleneck.ts index dbc7143f..53008aac 100644 --- a/src/Bottleneck.ts +++ b/src/Bottleneck.ts @@ -1,7 +1,6 @@ import type { BottleneckEvents, BottleneckOptions, - ClientsList, Counts, EventInfo, JobDefaults, @@ -18,6 +17,7 @@ import BottleneckError from "./BottleneckError"; import IORedisConnection from "./cluster/IORedisConnection"; import RedisConnection from "./cluster/RedisConnection"; import RedisDatastore from "./cluster/RedisDatastore"; +import { validateDatastoreOptions } from "./datastore-options"; import Events from "./Events"; import Group from "./Group"; import Job from "./Job"; @@ -102,15 +102,11 @@ class Bottleneck { timeout: null as number | null, heartbeatInterval: 5000, clientTimeout: 10000, - Redis: null, - clientOptions: {} as object, - clusterNodes: null as unknown, clearDatastore: false, connection: null, }; /** @internal */ instanceDefaults = { - datastore: "local", connection: null, id: "", rejectOnDrop: true, @@ -126,8 +122,9 @@ class Bottleneck { dropErrorMessage: "This limiter has been stopped.", }; - // Populated from instanceDefaults via parser.load in the constructor. + /** `"local"`, or the connection's `"redis"` / `"ioredis"`. */ datastore: string = "local"; + // Populated from instanceDefaults via parser.load in the constructor. connection: RedisConnection | IORedisConnection | null = null; id: string = ""; /** @internal */ @@ -186,14 +183,12 @@ class Bottleneck { this._registerLock = new Sync("register"); const storeOptions = load(options, this.storeDefaults, {}); - if (this.datastore === "redis" || this.datastore === "ioredis" || this.connection != null) { + if (this.connection != null) { const opts = load(options, this.redisStoreDefaults, {}); this._store = new RedisDatastore(this, storeOptions, opts); - } else if (this.datastore === "local") { + } else { const opts = load(options, this.localStoreDefaults, {}); this._store = new LocalDatastore(this, storeOptions, opts); - } else { - throw new BottleneckError(`Invalid datastore type: ${this.datastore}`, "INVALID_DATASTORE"); } this._queues.on("leftzero", () => this._store.heartbeat?.ref?.()); @@ -208,14 +203,12 @@ class Bottleneck { "INVALID_ARGUMENTS", ); } + validateDatastoreOptions(options); } - ready(): Promise { - return this._store.ready; - } - - clients(): ClientsList { - return this._store.clients as ClientsList; + /** Resolves once the limiter is connected to Redis (immediately for a local limiter). */ + async ready(): Promise { + await this._store.ready; } channel(): string { diff --git a/src/BottleneckError.ts b/src/BottleneckError.ts index 1dbda8f1..e4f7d282 100644 --- a/src/BottleneckError.ts +++ b/src/BottleneckError.ts @@ -11,7 +11,9 @@ export type BottleneckErrorCode = | "OVERWEIGHT" | "INVALID_DATASTORE" | "INVALID_ARGUMENTS" - | "MISSING_CLIENT"; + | "MISSING_CLIENT" + | "CLIENT_NOT_OPEN" + | "LEGACY_REDIS_OPTIONS"; class BottleneckError extends Error { /** Stable machine-readable reason; absent on internal invariant failures. */ diff --git a/src/Group.ts b/src/Group.ts index ebdc897f..27c5f34a 100644 --- a/src/Group.ts +++ b/src/Group.ts @@ -1,15 +1,10 @@ -import type { - BottleneckOptions, - GroupEvents, - GroupLimiterPair, - IORedisConnectionOptions, - RedisConnectionOptions, -} from "./types"; -import { runDetached, setDetachedInterval } from "./async-context"; +import type IORedisConnection from "./cluster/IORedisConnection"; +import type RedisConnection from "./cluster/RedisConnection"; +import type { BottleneckOptions, GroupEvents, GroupLimiterPair } from "./types"; +import { setDetachedInterval } from "./async-context"; import Bottleneck from "./Bottleneck"; -import IORedisConnection from "./cluster/IORedisConnection"; -import RedisConnection from "./cluster/RedisConnection"; import * as Scripts from "./cluster/Scripts"; +import { validateDatastoreOptions } from "./datastore-options"; import Events from "./Events"; import { load, overwrite } from "./parser"; @@ -47,43 +42,17 @@ class Group { /** @internal */ interval: ReturnType | undefined; /** @internal */ - sharedConnection: boolean; - /** @internal */ Bottleneck: typeof Bottleneck; constructor(limiterOptions: BottleneckOptions = {}) { + validateDatastoreOptions(limiterOptions); this.deleteKey = this.deleteKey.bind(this); this.limiterOptions = { ...limiterOptions } as Record; load(this.limiterOptions, this.defaults, this); this.Events = new Events(this); this.instances = {}; this._startAutoCleanup(); - this.sharedConnection = this.connection != null; this.Bottleneck = Bottleneck; - - if (this.connection == null) { - if (this.limiterOptions.datastore === "redis") { - // Same detachment rationale as RedisDatastore: socket setup must not - // capture the constructing caller's async context. - this.connection = runDetached( - () => - new RedisConnection( - Object.assign({}, this.limiterOptions, { - Events: this.Events, - }) as unknown as RedisConnectionOptions, - ), - ); - } else if (this.limiterOptions.datastore === "ioredis") { - this.connection = runDetached( - () => - new IORedisConnection( - Object.assign({}, this.limiterOptions, { - Events: this.Events, - }) as unknown as IORedisConnectionOptions, - ), - ); - } - } } key(key = ""): Bottleneck { @@ -182,10 +151,14 @@ class Group { } } - disconnect(flush = true): Promise | undefined { + /** + * Stop the Group's auto-cleanup and disconnect its Redis-backed limiters. + * The connection stays open: close it (and its client) yourself. + */ + async disconnect(flush = true): Promise { clearInterval(this.interval); - if (!this.sharedConnection) { - return this.connection?.disconnect(flush); + if (this.connection != null) { + await Promise.all(Object.values(this.instances).map((limiter) => limiter.disconnect(flush))); } } } diff --git a/src/LocalDatastore.ts b/src/LocalDatastore.ts index 21e44643..57700376 100644 --- a/src/LocalDatastore.ts +++ b/src/LocalDatastore.ts @@ -34,7 +34,6 @@ class LocalDatastore { /** @internal */ _unblockTime = 0; ready: Promise = Promise.resolve(); - clients: Record = {}; constructor( instance: Bottleneck, @@ -50,7 +49,6 @@ class LocalDatastore { this._done = 0; this._unblockTime = 0; this.ready = Promise.resolve(); - this.clients = {}; this._startHeartbeat(); } diff --git a/src/cluster/IORedisConnection.ts b/src/cluster/IORedisConnection.ts index 8b412076..7e0f69f1 100644 --- a/src/cluster/IORedisConnection.ts +++ b/src/cluster/IORedisConnection.ts @@ -1,21 +1,28 @@ import type Bottleneck from "../Bottleneck"; import type { IORedisConnectionOptions } from "../types"; -import type { IORedisClient, IORedisLib, RedisClients } from "./redis-types"; +import type { IORedisClient, RedisClients } from "./redis-types"; +import { runDetached } from "../async-context"; import BottleneckError from "../BottleneckError"; import Events from "../Events"; -import { load } from "../parser"; import { normalizeReply } from "./normalizeReply"; import * as Scripts from "./Scripts"; -class IORedisConnection { - /** @internal */ - Redis: IORedisLib | null; - /** @internal */ - clientOptions: object; - /** @internal */ - clusterNodes: unknown; - /** @internal */ - client: IORedisClient; +const noop = () => {}; + +const whenReady = (c: IORedisClient) => + new Promise((resolve) => { + if (c.status === "ready") { + resolve(); + } else { + c.once("ready", resolve); + } + }); + +class IORedisConnection { + /** The client passed in. The connection only sends commands on it. */ + readonly client: C; + /** The pub/sub client: `options.subscriber`, or a `client.duplicate()` the connection owns. */ + readonly subscriber: C; /** @internal */ Events: Events; /** @internal */ @@ -23,10 +30,12 @@ class IORedisConnection { /** @internal */ terminated = false; /** @internal */ - subscriber: IORedisClient; - /** @internal */ limiters: Record = {}; - ready: Promise>; + /** @internal Limiters using this connection, which receive its subscriber's errors. */ + instances = new Set(); + /** @internal Whether the connection duplicated (and so closes) its subscriber. */ + ownsSubscriber: boolean; + ready: Promise>; // Installed on the instance by Events (see Events constructor). declare on: { @@ -39,96 +48,53 @@ class IORedisConnection { }; declare removeAllListeners: (name?: string | null) => void; - /** @internal */ - defaults: { - Redis: IORedisLib | null; - clientOptions: object; - clusterNodes: unknown; - client: IORedisClient | null; - Events: Events | null; - } = { - Redis: null, - clientOptions: {}, - clusterNodes: null, - client: null, - Events: null, - }; - - constructor(options?: IORedisConnectionOptions) { - const opts = load(options ?? {}, this.defaults); - this.Redis = opts.Redis; - this.clientOptions = opts.clientOptions; - this.clusterNodes = opts.clusterNodes; - - const requireRedis = (): IORedisLib => { - if (opts.Redis == null) { - throw new BottleneckError( - "Bottleneck cluster mode requires a `Redis` library reference or a pre-built `client`. " + - "Pass it explicitly: `new Bottleneck({ datastore: 'ioredis', Redis: require('ioredis'), clientOptions })`.", - "MISSING_CLIENT", - ); - } - return opts.Redis; - }; - if (opts.client == null) requireRedis(); - - this.Events = opts.Events ?? new Events(this); - this.terminated = false; - - if (opts.clusterNodes != null) { - const { Cluster } = requireRedis(); - this.client = new Cluster(opts.clusterNodes, opts.clientOptions); - this.subscriber = new Cluster(opts.clusterNodes, opts.clientOptions); - } else { - const client = opts.client ?? new (requireRedis())(opts.clientOptions); - this.client = client; - // ioredis Cluster clients have no duplicate(); build a sibling Cluster instead. - this.subscriber = - client.duplicate?.() ?? new (requireRedis().Cluster)(client.startupNodes, client.options); + constructor(options: IORedisConnectionOptions) { + const { client, subscriber }: Partial> = options ?? {}; + if (client == null) { + throw new BottleneckError( + "IORedisConnection requires an ioredis `client`: `new IORedisConnection({ client })`.", + "MISSING_CLIENT", + ); } - this.limiters = {}; - this.ready = this._initReady(); + this.Events = new Events(this); + this.client = client; + this.ownsSubscriber = subscriber == null; + // Sockets the connection creates must not capture the constructing + // caller's async context: pub/sub-driven work would otherwise run (and + // trace) inside it for the connection's whole lifetime. + this.subscriber = subscriber ?? runDetached(() => client.duplicate() as C); + + this.ready = runDetached(() => this._initReady()); } /** @internal */ - async _initReady() { - await Promise.all([this._setup(this.client, false), this._setup(this.subscriber, true)]); - this._loadScripts(); - return { client: this.client, subscriber: this.subscriber }; - } + _onSubscriberError = (e: unknown): void => { + if (this.terminated) return; + this.Events.trigger("error", e); + for (const instance of this.instances) { + instance.Events.trigger("error", e); + } + }; /** @internal */ - _setup(client: IORedisClient, sub: boolean): Promise { - client.setMaxListeners(0); - return new Promise((resolve) => { - client.on("error", (e: unknown) => { - if (!this.terminated) { - this.Events.trigger("error", e); - } - }); - if (sub) { - client.on("message", (channel: string, message: string) => { - ( - this.limiters[channel]?._store as unknown as { - onMessage?: (channel: string, message: string) => Promise; - } - )?.onMessage?.(channel, message); - }); + _onMessage = (channel: string, message: string): void => { + ( + this.limiters[channel]?._store as unknown as { + onMessage?: (channel: string, message: string) => Promise; } - if (client.status === "ready") { - resolve(); - } else { - client.once("ready", resolve); - } - }); - } + )?.onMessage?.(channel, message); + }; /** @internal */ - _loadScripts(): void { - Scripts.names.forEach((name) => - this.client.defineCommand(name, { lua: Scripts.payload(name) }), - ); + async _initReady(): Promise> { + this.subscriber.on("message", this._onMessage); + if (this.ownsSubscriber) { + this.subscriber.setMaxListeners(0); + this.subscriber.on("error", this._onSubscriberError); + await whenReady(this.subscriber); + } + return { client: this.client, subscriber: this.subscriber }; } /** @internal */ @@ -143,7 +109,12 @@ class IORedisConnection { /** @internal */ __runScript__(name: Scripts.ScriptName, id: string, args: unknown[]): Promise { const keys = Scripts.keys(name, id); - return this.client[name](keys.length, ...keys, ...args); + const all = [...keys, ...Scripts.stringifyArgs(args)]; + return Scripts.run( + name, + (sha) => this.client.evalsha(sha, keys.length, ...all), + (script) => this.client.eval(script, keys.length, ...all), + ); } /** @internal */ @@ -161,31 +132,45 @@ class IORedisConnection { async __removeLimiter__(instance: Bottleneck): Promise { await Promise.all( [instance.channel(), instance.channel_client()].map(async (channel) => { - if (!this.terminated) { + // Only release channels this limiter holds: one whose setup failed + // never subscribed (unsubscribing on its dead subscriber would never + // settle), and a later limiter with the same id may own `channel()`. + if (this.limiters[channel] !== instance) return; + delete this.limiters[channel]; + if (this.terminated) return; + try { await this.subscriber.unsubscribe(channel); + } catch (e) { + // Closing the connection mid-UNSUBSCRIBE releases the channel anyway. + if (!this.terminated) throw e; } - delete this.limiters[channel]; }), ); } + /** + * Stop the connection's limiters and close the subscriber it duplicated. + * The client (and a subscriber passed in) stay open: close them yourself. + */ async disconnect(flush = true): Promise { for (const v of Object.values(this.limiters)) { clearInterval(v._store?.heartbeat); } this.limiters = {}; + this.instances.clear(); + if (this.terminated) return; this.terminated = true; - this.client.removeAllListeners?.("error"); - this.client.on?.("error", () => {}); - this.subscriber.removeAllListeners?.("error"); - this.subscriber.on?.("error", () => {}); - - if (flush) { - await Promise.all([this.client.quit(), this.subscriber.quit()]); - } else { - this.client.disconnect(); - this.subscriber.disconnect(); + this.subscriber.removeListener("message", this._onMessage); + if (this.ownsSubscriber) { + // Absorb late socket errors from the closing subscriber. + this.subscriber.removeListener("error", this._onSubscriberError); + this.subscriber.on("error", noop); + if (flush) { + await this.subscriber.quit(); + } else { + this.subscriber.disconnect(); + } } } } diff --git a/src/cluster/RedisConnection.ts b/src/cluster/RedisConnection.ts index 64b22bbb..ce7b5f52 100644 --- a/src/cluster/RedisConnection.ts +++ b/src/cluster/RedisConnection.ts @@ -1,9 +1,9 @@ import type Bottleneck from "../Bottleneck"; import type { RedisConnectionOptions } from "../types"; -import type { NodeRedisClient, NodeRedisLib, RedisClients } from "./redis-types"; +import type { NodeRedisClient, RedisClients } from "./redis-types"; +import { runDetached } from "../async-context"; import BottleneckError from "../BottleneckError"; import Events from "../Events"; -import { load } from "../parser"; import { normalizeReply } from "./normalizeReply"; import * as Scripts from "./Scripts"; @@ -18,23 +18,13 @@ const safe = async (run: () => unknown): Promise => { return undefined; } }; +const noop = () => {}; -const connectIfNeeded = async (c: NodeRedisClient) => { - if (c.isOpen === false) { - await c.connect(); - } -}; - -const stringifyArgs = (args: unknown[]): string[] => - args.map((a) => (a == null ? "" : typeof a === "string" ? a : String(a))); - -class RedisConnection { - /** @internal */ - Redis: NodeRedisLib | null; - /** @internal */ - clientOptions: object; - /** @internal */ - client: NodeRedisClient; +class RedisConnection { + /** The client passed in. The connection only sends commands on it. */ + readonly client: C; + /** The pub/sub client: `options.subscriber`, or a `client.duplicate()` the connection owns. */ + readonly subscriber: C; /** @internal */ Events: Events; /** @internal */ @@ -42,12 +32,12 @@ class RedisConnection { /** @internal */ terminated = false; /** @internal */ - shas: Partial> = {}; - /** @internal */ - subscriber: NodeRedisClient; - /** @internal */ limiters: Record = {}; - ready: Promise>; + /** @internal Limiters using this connection, which receive its subscriber's errors. */ + instances = new Set(); + /** @internal Whether the connection duplicated (and so connects and closes) its subscriber. */ + ownsSubscriber: boolean; + ready: Promise>; // Installed on the instance by Events (see Events constructor). declare on: { @@ -60,108 +50,73 @@ class RedisConnection { }; declare removeAllListeners: (name?: string | null) => void; - constructor(options?: RedisConnectionOptions) { - const opts = load(options ?? {}, this.defaults); - this.Redis = opts.Redis; - this.clientOptions = opts.clientOptions; - - const client = opts.client ?? opts.Redis?.createClient(opts.clientOptions); + constructor(options: RedisConnectionOptions) { + const { client, subscriber }: Partial> = options ?? {}; if (client == null) { throw new BottleneckError( - "Bottleneck cluster mode requires a `Redis` library reference or a pre-built `client`. " + - "Pass it explicitly: `new Bottleneck({ datastore: 'redis', Redis: require('redis'), clientOptions })`.", + "RedisConnection requires a node-redis `client`: `new RedisConnection({ client })`.", "MISSING_CLIENT", ); } + for (const c of [client, subscriber]) { + if (c?.isOpen === false) { + throw new BottleneckError( + "RedisConnection requires a connected node-redis client: call `client.connect()` first.", + "CLIENT_NOT_OPEN", + ); + } + } - this.Events = opts.Events ?? new Events(this); - this.terminated = false; - + this.Events = new Events(this); this.client = client; - this.subscriber = this.client.duplicate(); - this.limiters = {}; + this.ownsSubscriber = subscriber == null; + // Sockets the connection creates must not capture the constructing + // caller's async context: pub/sub-driven work would otherwise run (and + // trace) inside it for the connection's whole lifetime. + this.subscriber = subscriber ?? runDetached(() => client.duplicate() as C); - this.ready = this._initReady(); + this.ready = runDetached(() => this._initReady()); // Stored init promise: consumers await `ready` lazily, so suppress the // unhandled-rejection that would fire before anyone attaches a handler. this.ready.catch(() => {}); } /** @internal */ - defaults: { - Redis: NodeRedisLib | null; - clientOptions: object; - client: NodeRedisClient | null; - Events: Events | null; - } = { - Redis: null, - clientOptions: {}, - client: null, - Events: null, + _onSubscriberError = (e: unknown): void => { + if (this.terminated) return; + this.Events.trigger("error", e); + for (const instance of this.instances) { + instance.Events.trigger("error", e); + } }; /** @internal */ - async _initReady() { - await Promise.all([this._setup(this.client, false), this._setup(this.subscriber, true)]); - await this._loadScripts(); - return { client: this.client, subscriber: this.subscriber }; - } - - /** @internal */ - async _setup(client: NodeRedisClient, _sub: boolean): Promise { - client.setMaxListeners?.(0); - client.on("error", (e: unknown) => { - if (!this.terminated) { - this.Events.trigger("error", e); + async _initReady(): Promise> { + if (this.ownsSubscriber) { + this.subscriber.setMaxListeners?.(0); + this.subscriber.on("error", this._onSubscriberError); + if (this.subscriber.isOpen === false) { + await this.subscriber.connect(); } - }); - await connectIfNeeded(client); - } - - /** @internal */ - async _loadScript(name: Scripts.ScriptName): Promise { - const sha = await this.client.scriptLoad(Scripts.payload(name)); - this.shas[name] = sha; - return sha; - } - - /** @internal */ - _loadScripts(): Promise { - return Promise.all( - Scripts.names.map(async (k) => { - try { - return await this._loadScript(k); - } catch (e) { - if (!this.terminated) throw e; - } - }), - ); + } + return { client: this.client, subscriber: this.subscriber }; } /** @internal */ async __runCommand__(cmd: unknown[]): Promise { await this.ready; - const reply = await this.client.sendCommand(stringifyArgs(cmd)); + const reply = await this.client.sendCommand(Scripts.stringifyArgs(cmd)); return normalizeReply(cmd, reply); } /** @internal */ - async __runScript__(name: Scripts.ScriptName, id: string, args: unknown[]): Promise { - const keys = Scripts.keys(name, id); - const stringArgs = stringifyArgs(args); - try { - const sha = this.shas[name] ?? (await this._loadScript(name)); - return await this.client.evalSha(sha, { keys, arguments: stringArgs }); - } catch (e) { - if ( - typeof (e as Error)?.message === "string" && - (e as Error).message.startsWith("NOSCRIPT") - ) { - const sha = await this._loadScript(name); - return await this.client.evalSha(sha, { keys, arguments: stringArgs }); - } - throw e; - } + __runScript__(name: Scripts.ScriptName, id: string, args: unknown[]): Promise { + const options = { keys: Scripts.keys(name, id), arguments: Scripts.stringifyArgs(args) }; + return Scripts.run( + name, + (sha) => this.client.evalSha(sha, options), + (script) => this.client.eval(script, options), + ); } /** @internal */ @@ -184,37 +139,40 @@ class RedisConnection { async __removeLimiter__(instance: Bottleneck): Promise { await Promise.all( [instance.channel(), instance.channel_client()].map(async (channel) => { - if (!this.terminated) { + // Only release channels this limiter holds: one whose setup failed + // never subscribed (unsubscribing on its dead subscriber would never + // settle), and a later limiter with the same id may own `channel()`. + if (this.limiters[channel] !== instance) return; + delete this.limiters[channel]; + if (this.terminated) return; + try { await this.subscriber.unsubscribe(channel); + } catch (e) { + // Closing the connection mid-UNSUBSCRIBE releases the channel anyway. + if (!this.terminated) throw e; } - delete this.limiters[channel]; }), ); } + /** + * Stop the connection's limiters and close the subscriber it duplicated. + * The client (and a subscriber passed in) stay open: close them yourself. + */ async disconnect(flush = true): Promise { for (const v of Object.values(this.limiters)) { clearInterval(v._store?.heartbeat); } this.limiters = {}; + this.instances.clear(); if (this.terminated) return; this.terminated = true; - this.client.removeAllListeners?.("error"); - this.client.on?.("error", () => {}); - this.subscriber.removeAllListeners?.("error"); - this.subscriber.on?.("error", () => {}); - - if (flush) { - await Promise.all([ - safe(() => closeClient(this.client)), - safe(() => closeClient(this.subscriber)), - ]); - } else { - await Promise.all([ - safe(() => destroyClient(this.client)), - safe(() => destroyClient(this.subscriber)), - ]); + if (this.ownsSubscriber) { + // Absorb late socket errors from the closing subscriber. + this.subscriber.removeListener("error", this._onSubscriberError); + this.subscriber.on("error", noop); + await safe(() => (flush ? closeClient(this.subscriber) : destroyClient(this.subscriber))); } } } diff --git a/src/cluster/RedisDatastore.ts b/src/cluster/RedisDatastore.ts index 687f8e04..d1a3c9cc 100644 --- a/src/cluster/RedisDatastore.ts +++ b/src/cluster/RedisDatastore.ts @@ -1,12 +1,11 @@ import type Bottleneck from "../Bottleneck"; import type { StoreOptions } from "../types"; -import type { IORedisLib, NodeRedisLib, RedisClients } from "./redis-types"; +import type IORedisConnection from "./IORedisConnection"; +import type RedisConnection from "./RedisConnection"; import type { ScriptName } from "./Scripts"; -import { runDetached, setDetachedInterval, setDetachedTimeout } from "../async-context"; +import { setDetachedInterval, setDetachedTimeout } from "../async-context"; import BottleneckError from "../BottleneckError"; import { load, overwrite } from "../parser"; -import IORedisConnection from "./IORedisConnection"; -import RedisConnection from "./RedisConnection"; class RedisDatastore { /** @internal */ @@ -14,9 +13,6 @@ class RedisDatastore { _orphaned = 0; instance: Bottleneck; storeOptions: StoreOptions; - Redis: NodeRedisLib | IORedisLib | null = null; - clientOptions: object = {}; - clusterNodes: unknown = null; Promise: PromiseConstructor = Promise; timeout: number | null = null; heartbeatInterval = 5000; @@ -25,11 +21,9 @@ class RedisDatastore { connection!: RedisConnection | IORedisConnection; originalId: string; clientId: string; - sharedConnection: boolean; - clients: Partial = {}; capacityPriorityCounters: Record> = {}; heartbeat: ReturnType | undefined; - ready: Promise; + ready: Promise; constructor(instance: Bottleneck, storeOptions: StoreOptions, storeInstanceOptions: object) { this.instance = instance; @@ -37,41 +31,9 @@ class RedisDatastore { this.originalId = this.instance.id; this.clientId = this.instance._randomIndex(); load(storeInstanceOptions, storeInstanceOptions, this); - this.clients = {}; this.capacityPriorityCounters = {}; - this.sharedConnection = this.connection != null; - - if (!this.connection) { - if (this.instance.datastore === "redis") { - // Socket setup outlives the constructing caller: the subscriber - // socket keeps its creation context, so creating it inside a traced - // request would drain pub/sub jobs (and their EVALSHA calls) into - // that request's trace for the limiter's whole lifetime. - this.connection = runDetached( - () => - new RedisConnection({ - Redis: this.Redis, - clientOptions: this.clientOptions, - Events: this.instance.Events, - }), - ); - } else if (this.instance.datastore === "ioredis") { - this.connection = runDetached( - () => - new IORedisConnection({ - Redis: this.Redis, - clientOptions: this.clientOptions, - clusterNodes: this.clusterNodes, - Events: this.instance.Events, - }), - ); - } else { - throw new BottleneckError( - `Invalid datastore type: ${this.instance.datastore}`, - "INVALID_DATASTORE", - ); - } - } + // Errors from the connection's own sockets reach every limiter on it. + this.connection.instances.add(this.instance); this.instance.connection = this.connection; this.instance.datastore = this.connection.datastore; @@ -83,9 +45,8 @@ class RedisDatastore { } /** @internal */ - async _initReady(): Promise { - const clients: RedisClients = await this.connection.ready; - this.clients = clients; + async _initReady(): Promise { + await this.connection.ready; await this.runScript("init", this.prepareInitSettings(this.clearDatastore)); await this.connection.__addLimiter__(this.instance); await this.runScript("register_client", [this.instance.queued()]); @@ -104,13 +65,12 @@ class RedisDatastore { } }, this.heartbeatInterval).unref?.(); } - return clients; } /** @internal */ async __publish__(message: string): Promise { - const { client } = await this.ready; - return client.publish(this.instance.channel(), `message:${message.toString()}`); + await this.ready; + return this.connection.client.publish(this.instance.channel(), `message:${message.toString()}`); } async onMessage(_channel: string, message: string): Promise { @@ -126,11 +86,7 @@ class RedisDatastore { this.instance.Events.trigger("capacity-priority", capacity); const drained = await this.instance._drainAll(capacity); const newCapacity = capacity != null ? capacity - (drained || 0) : ""; - const { client } = this.clients; - if (client == null) { - throw new BottleneckError("Received a Redis message before the client was ready"); - } - return await client.publish( + return await this.connection.client.publish( this.instance.channel(), `capacity-priority:${newCapacity}::${counter}`, ); @@ -165,14 +121,11 @@ class RedisDatastore { } /** @internal */ - async __disconnect__(flush?: boolean): Promise { + async __disconnect__(_flush?: boolean): Promise { this._disconnecting = true; clearInterval(this.heartbeat); - if (this.sharedConnection) { - await this.connection.__removeLimiter__(this.instance); - } else { - return this.connection.disconnect(flush); - } + this.connection.instances.delete(this.instance); + await this.connection.__removeLimiter__(this.instance); } async runScript(name: ScriptName, args: unknown[]): Promise { diff --git a/src/cluster/Scripts.ts b/src/cluster/Scripts.ts index 63c6975b..4969f52d 100644 --- a/src/cluster/Scripts.ts +++ b/src/cluster/Scripts.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import lua from "./lua/index"; const headers: Record = { @@ -160,3 +161,39 @@ export const payload = (name: ScriptName): string => { ) .join("\n"); }; + +const shas = new Map(); + +/** The SHA1 Redis assigns `name`'s script, computed locally (no SCRIPT LOAD). */ +export const sha = (name: ScriptName): string => { + let digest = shas.get(name); + if (digest == null) { + digest = createHash("sha1").update(payload(name)).digest("hex"); + shas.set(name, digest); + } + return digest; +}; + +/** Redis arguments are strings; null/undefined become "" (as ioredis does). */ +export const stringifyArgs = (args: unknown[]): string[] => + args.map((a) => (a == null ? "" : typeof a === "string" ? a : String(a))); + +const isNoScript = (e: unknown) => e instanceof Error && e.message.startsWith("NOSCRIPT"); + +/** + * Run `name` by its SHA, falling back to sending the full script when the + * server doesn't have it cached yet (first use, SCRIPT FLUSH, a restart, or + * another node of a Cluster). EVAL also caches it, so later calls hit EVALSHA. + */ +export const run = async ( + name: ScriptName, + evalsha: (sha: string) => Promise, + evaluate: (script: string) => Promise, +): Promise => { + try { + return await evalsha(sha(name)); + } catch (e) { + if (isNoScript(e)) return evaluate(payload(name)); + throw e; + } +}; diff --git a/src/cluster/redis-types.ts b/src/cluster/redis-types.ts index 8c122bd6..94b656ac 100644 --- a/src/cluster/redis-types.ts +++ b/src/cluster/redis-types.ts @@ -4,68 +4,54 @@ * and must never be imported at runtime (the `light` build contains no cluster * code at all), so the client surface is described structurally. Methods every * supported version has are required; methods that vary by version are not. + * + * These are the constraints on the clients passed to `RedisConnection` and + * `IORedisConnection`: real clients from the supported versions satisfy them. */ -import type { ScriptName } from "./Scripts"; +// Method-shaped so listeners are checked bivariantly: real clients type them as +// EventEmitter's `(...args: any[]) => void`, and Bottleneck passes narrower +// callbacks like `(channel: string, message: string) => void`. +type Listener = { bivarianceHack(...args: unknown[]): void }["bivarianceHack"]; type BaseClient = { - on(event: string, cb: (...args: never[]) => void): unknown; - removeAllListeners?(event?: string): unknown; + on(event: string, cb: Listener): unknown; + removeListener(event: string, cb: Listener): unknown; publish(channel: string, message: string): Promise; unsubscribe(channel: string): unknown; }; -/** node-redis v4 closes with `quit`/`disconnect`; v5+ adds `close`/`destroy`. */ +type ScriptOptions = { keys: string[]; arguments: string[] }; + +/** A node-redis client. v4 closes with `quit`/`disconnect`; v5+ adds `close`/`destroy`. */ export type NodeRedisClient = BaseClient & ({ close(): unknown } | { close?: undefined; quit(): unknown }) & ({ destroy(): unknown } | { destroy?: undefined; disconnect(): unknown }) & { connect(): unknown; - isOpen: boolean; + readonly isOpen: boolean; duplicate(): NodeRedisClient; setMaxListeners?(n: number): unknown; sendCommand(args: string[]): Promise; - scriptLoad(script: string): Promise; - evalSha(sha: string, options: { keys: string[]; arguments: string[] }): Promise; + evalSha(sha: string, options: ScriptOptions): Promise; + eval(script: string, options: ScriptOptions): Promise; subscribe(channel: string, cb: (message: string) => void): unknown; }; -/** - * Installed on ioredis clients by `defineCommand()` in - * IORedisConnection._loadScripts, which runs before the connection is ready. - */ -type IORedisScriptCommands = Record< - ScriptName, - (numKeys: number, ...args: unknown[]) => Promise ->; - +/** An ioredis client: a standalone `Redis` or a `Redis.Cluster`. */ export type IORedisClient = BaseClient & { - status: string; - once(event: string, cb: (...args: never[]) => void): unknown; + readonly status: string; + once(event: string, cb: Listener): unknown; setMaxListeners(n: number): unknown; - /** Absent on ioredis Cluster clients. */ - duplicate?(): IORedisClient; + duplicate(): IORedisClient; quit(): Promise; disconnect(): unknown; subscribe(channel: string): Promise; - pipeline(commands: unknown[]): { exec(): Promise }; - defineCommand(name: string, options: { lua: string }): unknown; - // ioredis Cluster clients only - startupNodes?: unknown; - options?: unknown; -} & IORedisScriptCommands; + pipeline(commands: unknown[][]): { exec(): Promise }; + evalsha(sha: string, numKeys: number, ...args: string[]): Promise; + eval(script: string, numKeys: number, ...args: string[]): Promise; +}; export type RedisClients = { client: C; subscriber: C; }; - -/** The `redis` module statics Bottleneck touches. */ -export type NodeRedisLib = { - createClient(options: unknown): NodeRedisClient; -}; - -/** The `ioredis` constructor (`new Redis(...)`) plus its `Cluster` static. */ -export type IORedisLib = { - new (options: unknown): IORedisClient; - Cluster: new (nodes: unknown, options: unknown) => IORedisClient; -}; diff --git a/src/datastore-options.ts b/src/datastore-options.ts new file mode 100644 index 00000000..3494550e --- /dev/null +++ b/src/datastore-options.ts @@ -0,0 +1,27 @@ +import BottleneckError from "./BottleneckError"; + +/** Options Bottleneck v5 removed in favor of `connection`. */ +const REMOVED_OPTIONS = ["Redis", "client", "clientOptions", "clusterNodes"]; + +/** + * Reject the pre-v5 ways of asking for Redis (rather than silently running a + * local limiter): a Redis-backed limiter or Group now takes a `connection` + * built from the consumer's own client. `datastore: "local"` stays harmless. + */ +export function validateDatastoreOptions(options: object): void { + const values = options as Record; + const removed = REMOVED_OPTIONS.filter((name) => values[name] != null); + const { datastore } = values; + if (datastore === "redis" || datastore === "ioredis") removed.unshift("datastore"); + if (removed.length > 0) { + throw new BottleneckError( + `Bottleneck v5 removed the ${removed.map((name) => `\`${name}\``).join(", ")} option(s). ` + + "Build a connection from your own Redis client and pass it instead: " + + "`new Bottleneck({ connection: new IORedisConnection({ client }) })`.", + "LEGACY_REDIS_OPTIONS", + ); + } + if (datastore != null && datastore !== "local") { + throw new BottleneckError(`Invalid datastore type: ${String(datastore)}`, "INVALID_DATASTORE"); + } +} diff --git a/src/index.ts b/src/index.ts index d6560fd4..37c5306e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -8,7 +8,6 @@ export type { BatcherOptions, BottleneckEvents, BottleneckOptions, - ClientsList, Counts, EventInfo, EventInfoDropped, @@ -25,3 +24,4 @@ export type { Strategy, } from "./types"; export type { BottleneckErrorCode } from "./BottleneckError"; +export type { IORedisClient, NodeRedisClient } from "./cluster/redis-types"; diff --git a/src/types.ts b/src/types.ts index 457cefcf..78e23631 100644 --- a/src/types.ts +++ b/src/types.ts @@ -6,6 +6,9 @@ * opaquely through parser.load/overwrite. */ import type Bottleneck from "./Bottleneck"; +import type IORedisConnection from "./cluster/IORedisConnection"; +import type { IORedisClient, NodeRedisClient } from "./cluster/redis-types"; +import type RedisConnection from "./cluster/RedisConnection"; /** Queue-shedding strategy when the queue length reaches `highWater`. */ export type Strategy = 1 | 2 | 3 | 4; @@ -55,20 +58,14 @@ export type BottleneckOptions = { readonly scheduleStackLabel?: string | null; /** Set to true to track done jobs with counts() and jobStatus(). Uses more memory. */ readonly trackDoneStatus?: boolean | null; - /** Where the limiter stores its internal state: `local` (default) or `redis`/`ioredis` for Clustering. */ - readonly datastore?: string | null; /** Override the Promise library used by Bottleneck. */ readonly Promise?: unknown; - /** Passed directly to the redis client library you've selected. */ - readonly clientOptions?: unknown; - /** **ioredis only.** When set, the client is created via `new Redis.Cluster(clusterNodes, clientOptions)`. */ - readonly clusterNodes?: unknown; - /** The imported client library (`redis` or `ioredis`), required for clustering unless `client`/`connection` is provided. */ - readonly Redis?: unknown; - /** A pre-built client to use instead of creating one from `clientOptions`. */ - readonly client?: unknown; - /** A connection object from `new Bottleneck.RedisConnection` / `new Bottleneck.IORedisConnection`. */ - readonly connection?: unknown; + /** + * Store the limiter's state in Redis (Clustering) through this connection, + * built from your own client: `new IORedisConnection({ client })` or + * `new RedisConnection({ client })`. Without one, the limiter is local. + */ + readonly connection?: RedisConnection | IORedisConnection | null; /** When true, the limiter wipes existing Bottleneck state on the Redis db at startup. */ readonly clearDatastore?: boolean | null; /** Redis TTL in ms for the limiter's keys (state removed after this much inactivity). Defaults to 300000 under a Group. */ @@ -110,58 +107,32 @@ export type BatcherOptions = { }; /** - * Options for `new Bottleneck.RedisConnection(...)`. Either `Redis` (the - * imported `redis` library) or `client` (a pre-built NodeRedis client) is - * required; passing both is rejected by the union. + * Options for `new Bottleneck.RedisConnection(...)`. The connection only sends + * commands on `client`: it never connects, closes, or adds listeners to it. */ -export type RedisConnectionOptions = - | { - /** The `redis` library, e.g. `Redis from "redis"`. Required unless `client` is provided. */ - readonly Redis: unknown; - /** Passed directly to NodeRedis' `createClient()` method. */ - readonly clientOptions?: unknown; - readonly client?: never; - /** Internal: pubsub event sink wired up by RedisDatastore. */ - readonly Events?: unknown; - } - | { - /** An existing NodeRedis client to use. When provided, `clientOptions` is ignored. */ - readonly client: unknown; - readonly Redis?: never; - readonly clientOptions?: unknown; - /** Internal: pubsub event sink wired up by RedisDatastore. */ - readonly Events?: unknown; - }; +export type RedisConnectionOptions = { + /** A node-redis client. Call `client.connect()` first (no need to await it). */ + readonly client: C; + /** + * A client dedicated to Bottleneck's pub/sub subscriptions. Defaults to + * `client.duplicate()`, which the connection connects and closes itself. + */ + readonly subscriber?: C; +}; /** - * Options for `new Bottleneck.IORedisConnection(...)`. Either `Redis` (the - * imported `ioredis` library) or `client` (a pre-built ioredis client) is - * required; passing both is rejected by the union. + * Options for `new Bottleneck.IORedisConnection(...)`. The connection only sends + * commands on `client`: it never connects, closes, or adds listeners to it. */ -export type IORedisConnectionOptions = - | { - /** The `ioredis` library, e.g. `Redis from "ioredis"`. Required unless `client` is provided. */ - readonly Redis: unknown; - /** Passed directly to ioredis' constructor. */ - readonly clientOptions?: unknown; - /** When set, clients are created via `new Redis.Cluster(clusterNodes, clientOptions)`. */ - readonly clusterNodes?: unknown; - readonly client?: never; - /** Internal: pubsub event sink wired up by RedisDatastore. */ - readonly Events?: unknown; - } - | { - /** An existing ioredis client to use. When provided, `clientOptions` and `clusterNodes` are ignored. */ - readonly client: unknown; - readonly Redis?: never; - readonly clientOptions?: unknown; - readonly clusterNodes?: unknown; - /** Internal: pubsub event sink wired up by RedisDatastore. */ - readonly Events?: unknown; - }; - -/** Datastore-specific map of raw redis clients. */ -export type ClientsList = { client?: unknown; subscriber?: unknown }; +export type IORedisConnectionOptions = { + /** An ioredis client: `new Redis(...)` or `new Redis.Cluster(...)`. */ + readonly client: C; + /** + * A client dedicated to Bottleneck's pub/sub subscriptions. Defaults to + * `client.duplicate()`, which the connection closes itself. + */ + readonly subscriber?: C; +}; export type GroupLimiterPair = { key: string; limiter: Bottleneck }; diff --git a/test/bottleneck.ts b/test/bottleneck.ts index 395eb40b..411b2da6 100644 --- a/test/bottleneck.ts +++ b/test/bottleneck.ts @@ -14,13 +14,14 @@ const distPath: string | null = const Bottleneck: BottleneckClass = distPath == null ? SourceBottleneck : (await import(/* @vite-ignore */ distPath)).default; -// A limiter (or group) is Redis-backed if its options either name a Redis -// datastore explicitly OR provide a pre-built `connection` (in which case -// the Bottleneck constructor infers the datastore from the connection). -// Both paths must get the test heartbeat override applied, otherwise -// child limiters created from a "connection-only" Group inherit the 5000ms -// production default and produce 5-second flakes (see cluster.test.ts:75 -// and similar). +// Test-harness convention: `datastore: "redis" | "ioredis"` (and optionally +// `clientOptions`) asks for a Redis-backed limiter or Group with its own test +// client and connection, which the wrapper builds, owns, and closes on +// disconnect(). The product itself only takes `connection`. Options with a +// `connection` are Redis-backed too. Both paths must get the test heartbeat +// override applied, otherwise child limiters created from a +// "connection-only" Group inherit the 5000ms production default and produce +// 5-second flakes (see cluster.test.ts:75 and similar). const isRedisBacked = ( options: Record | undefined, ): options is Record => @@ -31,10 +32,9 @@ const isRedisBacked = ( const usingRedis = process.env.DATASTORE === "redis" || process.env.DATASTORE === "ioredis"; async function makeTestBottleneck(Base: BottleneckClass): Promise { - const Redis = - process.env.DATASTORE === "redis" - ? (await import("redis")).default - : (await import("ioredis")).default; + const { closeTestClient, makeTestClient } = await import("./helpers/clients.ts"); + type TestClient = ReturnType; + type Owned = { client: TestClient; connection: { disconnect(flush?: boolean): Promise } }; // One prefix per fork. Every test limiter's `id` is namespaced with this so // workers running in parallel against the single shared Redis (started by @@ -50,29 +50,38 @@ async function makeTestBottleneck(Base: BottleneckClass): Promise { + const options = + (clientOptions as Record | undefined) ?? + buildClientOptions(String(datastore)); + // The client's errors surface through the limiter (failed commands, and + // the connection's subscriber errors), so the harness client stays quiet. + const client = makeTestClient(options, { expectErrors: true, datastore: String(datastore) }); + // Only ioredis clients have a `status`. + const connection = + "status" in client + ? new Base.IORedisConnection({ client }) + : new Base.RedisConnection({ client }); + return { client, connection }; + }; + const withRedis = (options: Record | undefined) => { - if (!isRedisBacked(options)) return options; - const next = { ...options }; - // Only inject the Redis library / clientOptions when the test isn't - // bringing its own pre-built client or connection. Both of those carry - // their own clientOptions and the Bottleneck constructor would reject - // duplicates. - if (options.connection == null && options.client == null) { - if (next.Redis == null) next.Redis = Redis; - if (next.clientOptions == null) { - next.clientOptions = buildClientOptions(String(next.datastore)); - } + if (!isRedisBacked(options)) return { options, owned: null }; + const { datastore, clientOptions, ...next } = options; + let owned: Owned | null = null; + if (next.connection == null) { + owned = ownConnection(datastore, clientOptions); + next.connection = owned.connection; } // The default `heartbeatInterval` for Redis-backed limiters is 5000ms, // which bounds cross-limiter capacity-message recovery: when a @@ -101,7 +110,13 @@ async function makeTestBottleneck(Base: BottleneckClass): Promise { + if (owned == null) return; + await owned.connection.disconnect(flush); + await closeTestClient(owned.client); }; // `Group.limiters()` returns instances of the real `Bottleneck`, so we override @@ -110,8 +125,16 @@ async function makeTestBottleneck(Base: BottleneckClass): Promise) { - super(withRedis(options)); + const { options: next, owned } = withRedis(options); + super(next); + this.#owned = owned; + } + override async disconnect(flush?: boolean): Promise { + await super.disconnect(flush); + await closeOwned(this.#owned, flush); + this.#owned = null; } } @@ -119,8 +142,11 @@ async function makeTestBottleneck(Base: BottleneckClass): Promise) { - super(withRedis(options)); + const { options: next, owned } = withRedis(options); + super(next); + this.#owned = owned; // Group.key() instantiates child limiters via `this.Bottleneck`, which // the parent Group constructor sets to the library's Bottleneck class. // That bypasses our test-wrapper's heartbeatInterval injection, so @@ -129,6 +155,11 @@ async function makeTestBottleneck(Base: BottleneckClass): Promise { + await super.disconnect(flush); + await closeOwned(this.#owned, flush); + this.#owned = null; + } }; return TestBottleneck; diff --git a/test/cluster-connection.test.ts b/test/cluster-connection.test.ts index c6ba0d37..a7e37c20 100644 --- a/test/cluster-connection.test.ts +++ b/test/cluster-connection.test.ts @@ -1,5 +1,4 @@ import { describe, expect } from "vitest"; -import { defined } from "./helpers/defined"; import { runCommand } from "./helpers/store"; import { test, waitForState } from "./helpers/test-api"; @@ -14,6 +13,12 @@ const failingClientOptions = () => ? { socket: { port: 1, reconnectStrategy: () => false } } : { port: 1, retryStrategy: () => null }; +/** Resolves with the first "error" event the emitter triggers. */ +const firstError = (emitter: { on(event: "error", cb: (e: unknown) => void): unknown }) => + new Promise((resolve) => { + emitter.on("error", resolve); + }); + /** 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"); @@ -63,32 +68,34 @@ describe("Cluster connection", () => { }); describe("Ownership", () => { - test("Should close a connection the limiter built when it disconnects", async ({ + test("Should leave its connection open when a limiter disconnects", async ({ makeLimiter, + makeConnection, }) => { - const limiter = makeLimiter(); + const connection = makeConnection(); + const limiter = makeLimiter({ connection }); 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]), - ); + expect([isOpen(connection.client), isOpen(connection.subscriber)]).toEqual([true, true]); + const next = makeLimiter({ connection }); + await expect(next.schedule(() => Promise.resolve("ok"))).resolves.toBe("ok"); }); - 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; + test("Should disconnect a Group's limiters but leave its connection open", async ({ + makeGroup, + makeConnection, + }) => { + const connection = makeConnection(); + const group = makeGroup({ connection }); + await Promise.all([group.key("a").ready(), group.key("b").ready()]); + expect(Object.keys(connection.limiters)).toHaveLength(4); await group.disconnect(); - await waitForState(() => - expect([isOpen(client), isOpen(subscriber)]).toEqual([false, false]), - ); + expect(connection.limiters).toEqual({}); + expect([isOpen(connection.client), isOpen(connection.subscriber)]).toEqual([true, true]); }); test("Should leave a shared connection open when a Group disconnects", async ({ @@ -107,20 +114,19 @@ describe("Cluster connection", () => { 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, + test("Should send its subscriber's errors to the connection and every limiter on it", async ({ + makeClient, + makeConnection, + makeLimiter, }) => { - const group = makeGroup({ - datastore: process.env.DATASTORE, - clientOptions: failingClientOptions(), + const connection = makeConnection({ + client: makeClient(failingClientOptions(), { expectErrors: true }), }); - defined(group.connection).ready.catch(() => {}); + const limiters = [1, 2].map(() => makeLimiter({ connection }, { expectErrors: true })); await expect( - new Promise((resolve) => { - group.on("error", resolve); - }), - ).resolves.toBeTruthy(); + Promise.all([connection, ...limiters].map((emitter) => firstError(emitter))), + ).resolves.toHaveLength(3); }); }); }); diff --git a/test/cluster.test.ts b/test/cluster.test.ts index 70e03b46..9f65cc45 100644 --- a/test/cluster.test.ts +++ b/test/cluster.test.ts @@ -1,5 +1,5 @@ import assert from "node:assert"; -import { describe, expect } from "vitest"; +import { describe, expect, vi } from "vitest"; import type BottleneckBase from "../src/Bottleneck"; import type { ScriptName } from "../src/cluster/Scripts"; import type { JobOptions } from "../src/types"; @@ -7,7 +7,6 @@ import sleep from "../src/sleep"; import { defined } from "./helpers/defined"; import { limiterKeys, redisStore, runCommand } from "./helpers/store"; import { test, waitForState, deferred, enqueued } from "./helpers/test-api"; -import { wrongType } from "./helpers/wrong-type"; // Causality policy (Workstream B): observe product-timer effects via waitForState // and state counts — never assert wall-clock bounds around real network time. @@ -55,12 +54,14 @@ describe("Cluster-only", () => { return ready; }); - test("Should return clients", async ({ makeLimiter }) => { + test("Should expose its connection's clients", async ({ makeLimiter }) => { const rootLimiter = makeLimiter({ maxConcurrent: 2 }); - const clients = await rootLimiter.ready(); - expect(Object.keys(clients as Record)).toEqual(["client", "subscriber"]); - expect(Object.keys(rootLimiter.clients())).toEqual(["client", "subscriber"]); + await expect(rootLimiter.ready()).resolves.toBeUndefined(); + const { client, subscriber } = defined(rootLimiter.connection); + expect(client).toBeDefined(); + expect(subscriber).toBeDefined(); + expect(subscriber).not.toBe(client); }); test("Should return a promise when disconnecting", async ({ makeLimiter }) => { @@ -1072,8 +1073,9 @@ describe("Cluster-only", () => { await limiter.ready(); const store = redisStore(limiter); const published: [string, string][] = []; - store.clients.client = wrongType({ - publish: async (ch: string, msg: string) => published.push([ch, msg]), + vi.spyOn(store.connection.client, "publish").mockImplementation(async (ch, msg) => { + published.push([ch, msg]); + return 0; }); const capacities: (number | undefined)[] = []; diff --git a/test/connection.test.ts b/test/connection.test.ts index 31818827..d9096382 100644 --- a/test/connection.test.ts +++ b/test/connection.test.ts @@ -14,30 +14,23 @@ import { wrongType } from "./helpers/wrong-type"; type Handler = (...args: unknown[]) => void; type Overrides = Record; -const fakes = new WeakSet(); - -type FakeNodeClient = ReturnType; -type FakeIOClient = ReturnType; - -/** Retype a client the connection created from one of the fake factories. */ -function fake(client: NodeRedisClient): FakeNodeClient; -function fake(client: IORedisClient): FakeIOClient; -function fake(client: object): FakeNodeClient | FakeIOClient { - if (!fakes.has(client)) throw new Error("Expected a fake client"); - return wrongType(client); -} - const makeEmitter = () => { const handlers: Record = {}; return { on(name: string, cb: Handler) { (handlers[name] ??= []).push(cb); }, - emit(name: string, ...args: unknown[]) { - for (const cb of handlers[name] ?? []) cb(...args); + removeListener(name: string, cb: Handler) { + handlers[name] = (handlers[name] ?? []).filter((h) => h !== cb); + }, + listenerCount(name: string) { + return handlers[name]?.length ?? 0; }, - removeAllListeners(name?: string) { - if (name != null) delete handlers[name]; + emit(name: string, ...args: unknown[]) { + const listeners = handlers[name] ?? []; + // Like EventEmitter: an "error" nobody listens to throws. + if (name === "error" && listeners.length === 0) throw args[0]; + for (const cb of listeners) cb(...args); }, }; }; @@ -48,13 +41,18 @@ const makeNodeClient = (overrides: Overrides = {}, dupOverrides: Overrides = {}) ...makeEmitter(), _subscriptions: subscriptions, isOpen: true, - scriptLoad: vi.fn<(payload: string) => Promise>( - async (payload) => `sha-${payload.length}`, - ), + connect: vi.fn<() => Promise>(async () => { + client.isOpen = true; + }), + setMaxListeners: vi.fn<(n: number) => void>(), evalSha: vi.fn< - (sha: string, options: { keys: string[]; arguments: string[] }) => Promise + (sha: string, options: { keys: string[]; arguments: string[] }) => Promise + >(async () => 1), + eval: vi.fn< + (script: string, options: { keys: string[]; arguments: string[] }) => Promise >(async () => 1), sendCommand: vi.fn<() => Promise>(async () => "OK"), + publish: vi.fn<() => Promise>(async () => 0), subscribe: vi.fn<(channel: string, cb: Handler) => Promise>(async (channel, cb) => { subscriptions[channel] = cb; }), @@ -63,46 +61,40 @@ const makeNodeClient = (overrides: Overrides = {}, dupOverrides: Overrides = {}) quit: vi.fn<() => Promise>(async () => {}), destroy: vi.fn<() => Promise>(async () => {}), disconnect: vi.fn<() => void>(), - duplicate: (): NodeRedisClient => makeNodeClient(dupOverrides), + // node-redis duplicates start disconnected. + duplicate: (): NodeRedisClient => makeNodeClient({ isOpen: false, ...dupOverrides }), }; Object.assign(client, overrides); - fakes.add(client); // The fake implements only what RedisConnection calls. return wrongType(client); }; -const makeIOClient = (overrides: Overrides = {}) => { +const makeIOClient = (overrides: Overrides = {}, dupOverrides: Overrides = {}) => { const client = { ...makeEmitter(), - status: "connect", + status: "ready", once(name: string, cb: Handler) { client.on(name, cb); }, - setMaxListeners() {}, - defineCommand: vi.fn<(...args: unknown[]) => void>(), - subscribe: vi.fn<(...args: unknown[]) => Promise>(async () => 1), - unsubscribe: vi.fn<(...args: unknown[]) => Promise>(async () => 1), + setMaxListeners: vi.fn<(n: number) => void>(), + evalsha: vi.fn<(sha: string, numKeys: number, ...args: string[]) => Promise>( + async () => 1, + ), + eval: vi.fn<(script: string, numKeys: number, ...args: string[]) => Promise>( + async () => 1, + ), + publish: vi.fn<() => Promise>(async () => 0), + subscribe: vi.fn<(channel: string) => Promise>(async () => 1), + unsubscribe: vi.fn<(channel: string) => Promise>(async () => 1), + quit: vi.fn<() => Promise>(async () => "OK"), + disconnect: vi.fn<() => void>(), + duplicate: (): IORedisClient => makeIOClient(dupOverrides), }; Object.assign(client, overrides); - fakes.add(client); // The fake implements only what IORedisConnection calls. return wrongType(client); }; -const makeFakeRedis = (mainOverrides: Overrides = {}, subOverrides: Overrides = {}) => { - // Plain functions, not vi.fn(): `new FakeRedis(...)` must return the fake - // client object the factory builds. - const calls: { nodes: unknown; options: unknown }[] = []; - function Cluster(nodes: unknown, options: unknown) { - calls.push({ nodes, options }); - return makeIOClient({ status: "ready" }); - } - function FakeRedis() { - return makeIOClient({ ...mainOverrides, duplicate: () => makeIOClient(subOverrides) }); - } - return Object.assign(FakeRedis, { Cluster: Object.assign(Cluster, { calls }) }); -}; - /** Just the limiter surface the connections touch when (un)registering. */ const makeLimiterInstance = () => { const instance = { @@ -113,88 +105,140 @@ const makeLimiterInstance = () => { return { instance, limiter: wrongType(instance) }; }; +const noscript = () => new Error("NOSCRIPT No matching script. Please use EVAL."); + describe("RedisConnection (node-redis)", () => { - test("Should refuse to build without a Redis reference or a pre-built client", () => { + test("Should not fail a limiter's removal when the connection closes mid-unsubscribe", async () => { + const conn = new RedisConnection({ client: makeNodeClient() }); + await conn.ready; + const { limiter } = makeLimiterInstance(); + await conn.__addLimiter__(limiter); + conn.subscriber.unsubscribe.mockImplementation(async () => { + await conn.disconnect(false); + throw new Error("Disconnects client"); + }); + + await expect(conn.__removeLimiter__(limiter)).resolves.toBeUndefined(); + }); + + test("Should only unsubscribe channels a limiter holds", async () => { + const conn = new RedisConnection({ client: makeNodeClient() }); + await conn.ready; + + // A limiter whose setup failed never subscribed: there's nothing to release + // (and its dead subscriber would never answer an UNSUBSCRIBE). + const { limiter: neverAdded } = makeLimiterInstance(); + await conn.__removeLimiter__(neverAdded); + expect(conn.subscriber.unsubscribe).not.toHaveBeenCalled(); + + // A later limiter with the same channels takes them over. + const { limiter: first } = makeLimiterInstance(); + const { limiter: second } = makeLimiterInstance(); + await conn.__addLimiter__(first); + await conn.__addLimiter__(second); + await conn.__removeLimiter__(first); + expect(conn.subscriber.unsubscribe).not.toHaveBeenCalled(); + expect(conn.limiters["ch-one"]).toBe(second); + }); + + test("Should refuse to build without a client", () => { // {} is not a valid RedisConnectionOptions; the runtime rejection is the contract expect(() => new RedisConnection({} as unknown as RedisConnectionOptions)).toThrow( - /requires a `Redis` library reference or a pre-built `client`/, + /requires a node-redis `client`/, ); }); - test("Should connect a client that is not open", async () => { - const connect = vi.fn<() => Promise>(async () => {}); - const client = makeNodeClient({ isOpen: false, connect }); - const conn = new RedisConnection({ client }); + test("Should refuse a client that hasn't been connected", () => { + const client = makeNodeClient({ isOpen: false }); + expect(() => new RedisConnection({ client })).toThrow(/call `client.connect\(\)` first/); + expect(client.connect).not.toHaveBeenCalled(); + }); + test("Should connect, listen to, and close the subscriber it duplicates", async () => { + const conn = new RedisConnection({ client: makeNodeClient() }); await conn.ready; - expect(connect).toHaveBeenCalledTimes(1); + expect(conn.subscriber.connect).toHaveBeenCalledTimes(1); + expect(conn.subscriber.setMaxListeners).toHaveBeenCalledWith(0); + expect(conn.subscriber.listenerCount("error")).toBe(1); + + await conn.disconnect(true); + expect(conn.subscriber.close).toHaveBeenCalledTimes(1); + // A no-op handler absorbs late socket errors from the closing subscriber. + expect(() => conn.subscriber.emit("error", new Error("late"))).not.toThrow(); }); - test("Should reload and retry a script after NOSCRIPT", async () => { + test("Should only send commands on the consumer's client", async () => { const client = makeNodeClient(); - client.evalSha - .mockRejectedValueOnce(new Error("NOSCRIPT Invalid script hash.")) - .mockResolvedValueOnce(7); - + const keys = Object.keys(client); const conn = new RedisConnection({ client }); await conn.ready; - await expect(conn.__runScript__("register", "id", [null, 5])).resolves.toBe(7); - expect(client.scriptLoad).toHaveBeenCalledTimes(Scripts.names.length + 1); - expect(client.evalSha.mock.calls[1]?.[1].arguments[0]).toBe(""); + const { limiter } = makeLimiterInstance(); + await conn.__addLimiter__(limiter); + await conn.__runScript__("register", "id", []); + await conn.__runCommand__(["get", "key"]); + await conn.__removeLimiter__(limiter); + await conn.disconnect(true); + + expect(Object.keys(client)).toEqual(keys); + expect(client.listenerCount("error")).toBe(0); + for (const method of [ + client.connect, + client.setMaxListeners, + client.close, + client.quit, + client.destroy, + client.disconnect, + ]) { + expect(method).not.toHaveBeenCalled(); + } }); - test("Should reject ready when script loading fails", async () => { - const client = makeNodeClient({ - scriptLoad: vi.fn<() => Promise>(async () => { - throw new Error("script load failed"); - }), - }); + test("Should run scripts by SHA and fall back to EVAL on NOSCRIPT", async () => { + const client = makeNodeClient(); + client.evalSha.mockRejectedValueOnce(noscript()); + client.eval.mockResolvedValueOnce(7); const conn = new RedisConnection({ client }); - await expect(conn.ready).rejects.toThrow("script load failed"); + await expect(conn.__runScript__("register", "id", [null, 5])).resolves.toBe(7); + const options = { keys: Scripts.keys("register", "id"), arguments: ["", "5"] }; + expect(client.evalSha).toHaveBeenCalledWith(Scripts.sha("register"), options); + expect(client.eval).toHaveBeenCalledWith(Scripts.payload("register"), options); }); - test("Should swallow script loading failures after termination", async () => { - const client = makeNodeClient({ - scriptLoad: vi.fn<() => Promise>(async () => { - throw new Error("script load failed"); - }), - }); + test("Should rethrow script errors other than NOSCRIPT", async () => { + const client = makeNodeClient(); + client.evalSha.mockRejectedValueOnce(new Error("ERR boom")); const conn = new RedisConnection({ client }); - conn.terminated = true; - await expect(conn.ready).resolves.toBeDefined(); + await expect(conn.__runScript__("register", "id", [])).rejects.toThrow("ERR boom"); + expect(client.eval).not.toHaveBeenCalled(); }); - test("Should trigger error events only while not terminated", async () => { - const client = makeNodeClient(); - const conn = new RedisConnection({ client }); + test("Should trigger error events from its own subscriber only while not terminated", async () => { + const conn = new RedisConnection({ client: makeNodeClient() }); await conn.ready; const errors: unknown[] = []; conn.on("error", (e: unknown) => errors.push(e)); - client.emit("error", new Error("boom")); - fake(conn.subscriber).emit("error", new Error("boom")); - expect(errors.length).toBe(2); + conn.subscriber.emit("error", new Error("boom")); + expect(errors.length).toBe(1); conn.terminated = true; - client.emit("error", new Error("ignored")); - fake(conn.subscriber).emit("error", new Error("ignored")); - expect(errors.length).toBe(2); + conn.subscriber.emit("error", new Error("ignored")); + expect(errors.length).toBe(1); }); test("Should route subscribed messages to the limiter's store", async () => { - const client = makeNodeClient(); - const conn = new RedisConnection({ client }); + const conn = new RedisConnection({ client: makeNodeClient() }); await conn.ready; const { instance, limiter } = makeLimiterInstance(); await conn.__addLimiter__(limiter); expect(conn.subscriber.subscribe).toHaveBeenCalledTimes(2); - const onMessage = fake(conn.subscriber)._subscriptions["ch-one"]; + const onMessage = conn.subscriber._subscriptions["ch-one"]; onMessage?.("hello"); expect(instance._store.onMessage).toHaveBeenCalledWith("ch-one", "hello"); @@ -203,91 +247,151 @@ describe("RedisConnection (node-redis)", () => { expect(conn.limiters["ch-one"]).toBeUndefined(); }); - test("Should close clients on flush and destroy them otherwise", async () => { - const client1 = makeNodeClient({}, { close: undefined }); - const conn1 = new RedisConnection({ client: client1 }); + test("Should fall back to quit and disconnect on node-redis v4 subscribers", async () => { + const conn1 = new RedisConnection({ client: makeNodeClient({}, { close: undefined }) }); await conn1.ready; await conn1.disconnect(true); - expect(client1.close).toHaveBeenCalledTimes(1); - expect(client1).not.toHaveProperty("subscriber"); - expect(fake(conn1.subscriber).quit).toHaveBeenCalledTimes(1); - expect(conn1.subscriber.close).toBeUndefined(); - - // The no-op error handlers installed by disconnect() must absorb late - // client error events. - fake(conn1.client).emit("error", new Error("late")); - fake(conn1.subscriber).emit("error", new Error("late")); - - const client2 = makeNodeClient({}, { close: undefined, destroy: undefined }); - const conn2 = new RedisConnection({ client: client2 }); + expect(conn1.subscriber.quit).toHaveBeenCalledTimes(1); + + const conn2 = new RedisConnection({ client: makeNodeClient({}, { destroy: undefined }) }); await conn2.ready; await conn2.disconnect(false); - expect(client2.destroy).toHaveBeenCalledTimes(1); - expect(fake(conn2.subscriber).disconnect).toHaveBeenCalledTimes(1); + expect(conn2.subscriber.disconnect).toHaveBeenCalledTimes(1); + }); + + test("Should leave a consumer-supplied subscriber open and unlistened", async () => { + const subscriber = makeNodeClient(); + const conn = new RedisConnection({ client: makeNodeClient(), subscriber }); + await conn.ready; + expect(conn.subscriber).toBe(subscriber); + + await conn.disconnect(true); + expect(subscriber.listenerCount("error")).toBe(0); + expect(subscriber.setMaxListeners).not.toHaveBeenCalled(); + expect(subscriber.close).not.toHaveBeenCalled(); }); }); describe("IORedisConnection", () => { - test("Should refuse to build without a Redis reference or a pre-built client", () => { - expect(() => new IORedisConnection({} as unknown as IORedisConnectionOptions)).toThrow( - /requires a `Redis` library reference or a pre-built `client`/, - ); + test("Should not fail a limiter's removal when the connection closes mid-unsubscribe", async () => { + const conn = new IORedisConnection({ client: makeIOClient() }); + await conn.ready; + const { limiter } = makeLimiterInstance(); + await conn.__addLimiter__(limiter); + conn.subscriber.unsubscribe.mockImplementation(async () => { + await conn.disconnect(false); + throw new Error("Disconnects client"); + }); + + await expect(conn.__removeLimiter__(limiter)).resolves.toBeUndefined(); }); - test("Should resolve ready immediately for already-ready clients", async () => { - const Redis = makeFakeRedis({ status: "ready" }, { status: "ready" }); - const conn = new IORedisConnection({ Redis, clientOptions: {} }); + test("Should only unsubscribe channels a limiter holds", async () => { + const conn = new IORedisConnection({ client: makeIOClient() }); + await conn.ready; - const { client, subscriber } = await conn.ready; - expect(client).toBe(conn.client); - expect(subscriber).toBe(conn.subscriber); - expect(client.defineCommand).toHaveBeenCalledTimes(Scripts.names.length); + // A limiter whose setup failed never subscribed: there's nothing to release + // (and its dead subscriber would never answer an UNSUBSCRIBE). + const { limiter: neverAdded } = makeLimiterInstance(); + await conn.__removeLimiter__(neverAdded); + expect(conn.subscriber.unsubscribe).not.toHaveBeenCalled(); + + // A later limiter with the same channels takes them over. + const { limiter: first } = makeLimiterInstance(); + const { limiter: second } = makeLimiterInstance(); + await conn.__addLimiter__(first); + await conn.__addLimiter__(second); + await conn.__removeLimiter__(first); + expect(conn.subscriber.unsubscribe).not.toHaveBeenCalled(); + expect(conn.limiters["ch-one"]).toBe(second); + }); + + test("Should refuse to build without a client", () => { + expect(() => new IORedisConnection({} as unknown as IORedisConnectionOptions)).toThrow( + /requires an ioredis `client`/, + ); }); - test("Should wait for the ready event on a connecting client", async () => { - const Redis = makeFakeRedis({ status: "ready" }, { status: "connect" }); - const conn = new IORedisConnection({ Redis, clientOptions: {} }); + test("Should resolve ready once the subscriber it duplicates is ready", async () => { + // The consumer's client still connecting doesn't hold up `ready`: its + // commands queue until it connects. + const conn = new IORedisConnection({ + client: makeIOClient({ status: "connect" }, { status: "connect" }), + }); let resolved = false; - conn.ready.then(() => { + void conn.ready.then(() => { resolved = true; return resolved; }); await Promise.resolve(); expect(resolved).toBe(false); - fake(conn.subscriber).emit("ready"); + conn.subscriber.emit("ready"); await conn.ready; expect(resolved).toBe(true); }); - test("Should build a Cluster subscriber when the client cannot duplicate", async () => { - const Redis = makeFakeRedis(); - const client = makeIOClient({ - status: "ready", - startupNodes: ["node-1"], - options: { keyPrefix: "x" }, - duplicate: undefined, - }); + test("Should listen to and close the subscriber it duplicates", async () => { + const conn = new IORedisConnection({ client: makeIOClient() }); + await conn.ready; + expect(conn.subscriber.setMaxListeners).toHaveBeenCalledWith(0); + expect(conn.subscriber.listenerCount("error")).toBe(1); + + await conn.disconnect(true); + expect(conn.subscriber.quit).toHaveBeenCalledTimes(1); + expect(conn.subscriber.listenerCount("message")).toBe(0); + expect(() => conn.subscriber.emit("error", new Error("late"))).not.toThrow(); - // The options type forbids `Redis` alongside `client`, but the Cluster - // fallback for a non-duplicating client reads it. - const conn = new IORedisConnection(wrongType({ Redis, client })); + const conn2 = new IORedisConnection({ client: makeIOClient() }); + await conn2.ready; + await conn2.disconnect(false); + expect(conn2.subscriber.disconnect).toHaveBeenCalledTimes(1); + }); + test("Should only send commands on the consumer's client", async () => { + const client = makeIOClient(); + const keys = Object.keys(client); + const conn = new IORedisConnection({ client }); await conn.ready; - expect(Redis.Cluster.calls).toEqual([{ nodes: client.startupNodes, options: client.options }]); + + const { limiter } = makeLimiterInstance(); + await conn.__addLimiter__(limiter); + await conn.__runScript__("register", "id", []); + await conn.__removeLimiter__(limiter); + await conn.disconnect(true); + + // No script methods (or anything else) defined on it. + expect(Object.keys(client)).toEqual(keys); + expect(client.listenerCount("error")).toBe(0); + expect(client.listenerCount("ready")).toBe(0); + expect(client.setMaxListeners).not.toHaveBeenCalled(); + expect(client.quit).not.toHaveBeenCalled(); + expect(client.disconnect).not.toHaveBeenCalled(); + }); + + test("Should run scripts by SHA and fall back to EVAL on NOSCRIPT", async () => { + const client = makeIOClient(); + client.evalsha.mockRejectedValueOnce(noscript()); + client.eval.mockResolvedValueOnce(7); + const conn = new IORedisConnection({ client }); + + await expect(conn.__runScript__("register", "id", [null, 5])).resolves.toBe(7); + const keys = Scripts.keys("register", "id"); + const args = [keys.length, ...keys, "", "5"]; + expect(client.evalsha).toHaveBeenCalledWith(Scripts.sha("register"), ...args); + expect(client.eval).toHaveBeenCalledWith(Scripts.payload("register"), ...args); }); test("Should route subscribed messages and clean up on removal", async () => { - const Redis = makeFakeRedis({ status: "ready" }, { status: "ready" }); - const conn = new IORedisConnection({ Redis, clientOptions: {} }); + const conn = new IORedisConnection({ client: makeIOClient() }); await conn.ready; const { instance, limiter } = makeLimiterInstance(); await conn.__addLimiter__(limiter); expect(conn.subscriber.subscribe).toHaveBeenCalledTimes(2); - fake(conn.subscriber).emit("message", "ch-one", "hello"); + conn.subscriber.emit("message", "ch-one", "hello"); expect(instance._store.onMessage).toHaveBeenCalledWith("ch-one", "hello"); await conn.__removeLimiter__(limiter); @@ -295,44 +399,32 @@ describe("IORedisConnection", () => { expect(conn.limiters["ch-one"]).toBeUndefined(); }); - test("Should trigger error events only while not terminated", async () => { - const Redis = makeFakeRedis({ status: "ready" }, { status: "ready" }); - const conn = new IORedisConnection({ Redis, clientOptions: {} }); + test("Should trigger error events from its own subscriber only while not terminated", async () => { + const conn = new IORedisConnection({ client: makeIOClient() }); await conn.ready; - const errors = []; + const errors: unknown[] = []; conn.on("error", (e: unknown) => errors.push(e)); - fake(conn.client).emit("error", new Error("boom")); - fake(conn.subscriber).emit("error", new Error("boom")); - expect(errors.length).toBe(2); + conn.subscriber.emit("error", new Error("boom")); + expect(errors.length).toBe(1); conn.terminated = true; - fake(conn.client).emit("error", new Error("ignored")); - expect(errors.length).toBe(2); + conn.subscriber.emit("error", new Error("ignored")); + expect(errors.length).toBe(1); }); - test("Should quit or disconnect clients on disconnect", async () => { - const Redis1 = makeFakeRedis({ status: "ready" }, { status: "ready" }); - const conn1 = new IORedisConnection({ Redis: Redis1, clientOptions: {} }); - await conn1.ready; - conn1.client.quit = vi.fn<() => Promise>(async () => "OK"); - conn1.subscriber.quit = vi.fn<() => Promise>(async () => "OK"); - await conn1.disconnect(true); - expect(conn1.client.quit).toHaveBeenCalledTimes(1); - expect(conn1.subscriber.quit).toHaveBeenCalledTimes(1); - - // Late errors are absorbed by the no-op handlers disconnect() installs. - fake(conn1.client).emit("error", new Error("late")); - fake(conn1.subscriber).emit("error", new Error("late")); - - const Redis2 = makeFakeRedis({ status: "ready" }, { status: "ready" }); - const conn2 = new IORedisConnection({ Redis: Redis2, clientOptions: {} }); - await conn2.ready; - conn2.client.disconnect = vi.fn<() => void>(); - conn2.subscriber.disconnect = vi.fn<() => void>(); - await conn2.disconnect(false); - expect(conn2.client.disconnect).toHaveBeenCalledTimes(1); - expect(conn2.subscriber.disconnect).toHaveBeenCalledTimes(1); + test("Should leave a consumer-supplied subscriber open, removing only its message listener", async () => { + const subscriber = makeIOClient(); + const conn = new IORedisConnection({ client: makeIOClient(), subscriber }); + await conn.ready; + expect(conn.subscriber).toBe(subscriber); + expect(subscriber.listenerCount("message")).toBe(1); + + await conn.disconnect(true); + expect(subscriber.listenerCount("message")).toBe(0); + expect(subscriber.listenerCount("error")).toBe(0); + expect(subscriber.setMaxListeners).not.toHaveBeenCalled(); + expect(subscriber.quit).not.toHaveBeenCalled(); }); }); diff --git a/test/error-codes.test.ts b/test/error-codes.test.ts index 24a9b262..44163e92 100644 --- a/test/error-codes.test.ts +++ b/test/error-codes.test.ts @@ -1,7 +1,9 @@ import { describe, expect } from "vitest"; -import { BottleneckError } from "../src/index"; +import type { NodeRedisClient } from "../src/cluster/redis-types"; +import SourceBottleneck, { BottleneckError, Group as SourceGroup } from "../src/index"; import Bottleneck from "./bottleneck"; import { deferred, enqueued, test } from "./helpers/test-api"; +import { wrongType } from "./helpers/wrong-type"; /* * The `code` on BottleneckError is the stable programmatic signal (messages @@ -101,15 +103,46 @@ describe("BottleneckError codes", () => { expect(codeOf(invalidArgs)).toEqual("INVALID_ARGUMENTS"); const badStore = await capture( - Promise.resolve().then(() => new Bottleneck({ datastore: "bogus" })), + Promise.resolve().then(() => new Bottleneck(wrongType({ datastore: "bogus" }))), "bad datastore should have thrown", ); expect(codeOf(badStore)).toEqual("INVALID_DATASTORE"); + // The product class, not the test wrapper: under the redis projects the + // wrapper treats `datastore: "redis"` as its own flag for a test connection. + for (const legacy of [ + { datastore: "redis" }, + { datastore: "ioredis" }, + { Redis: {} }, + { client: {} }, + { clientOptions: {} }, + { clusterNodes: [] }, + ]) { + const removed = await capture( + Promise.resolve().then(() => new SourceBottleneck(wrongType(legacy))), + `${JSON.stringify(legacy)} should have thrown`, + ); + expect(codeOf(removed)).toEqual("LEGACY_REDIS_OPTIONS"); + const removedFromGroup = await capture( + Promise.resolve().then(() => new SourceGroup(wrongType(legacy))), + `Group ${JSON.stringify(legacy)} should have thrown`, + ); + expect(codeOf(removedFromGroup)).toEqual("LEGACY_REDIS_OPTIONS"); + } + const noClient = await capture( Promise.resolve().then(() => new Bottleneck.RedisConnection({} as never)), "clientless connection should have thrown", ); expect(codeOf(noClient)).toEqual("MISSING_CLIENT"); + + const closedClient = await capture( + Promise.resolve().then( + () => + new Bottleneck.RedisConnection({ client: wrongType({ isOpen: false }) }), + ), + "unconnected client should have thrown", + ); + expect(codeOf(closedClient)).toEqual("CLIENT_NOT_OPEN"); }); }); diff --git a/test/general.test.ts b/test/general.test.ts index 576741b4..577d39d7 100644 --- a/test/general.test.ts +++ b/test/general.test.ts @@ -830,7 +830,7 @@ describe("General", () => { describe("Datastore errors", () => { test("Should refuse an unknown datastore type", () => { - expect(() => new Bottleneck({ datastore: "carrier-pigeon" })).toThrow( + expect(() => new Bottleneck(wrongType({ datastore: "carrier-pigeon" }))).toThrow( "Invalid datastore type: carrier-pigeon", ); }); diff --git a/test/helpers/clients.ts b/test/helpers/clients.ts new file mode 100644 index 00000000..e929251d --- /dev/null +++ b/test/helpers/clients.ts @@ -0,0 +1,55 @@ +import IORedis from "ioredis"; +import { ConnectionTimeoutError, SocketClosedUnexpectedlyError } from "redis"; +import RedisClient from "redis"; +import type { IORedisClient, NodeRedisClient } from "../../src/cluster/redis-types"; +import buildClientOptions from "../redis-client-options"; + +/** Log an error event unless it's a known transient connection error. */ +export function logUnexpectedError(source: string, err: unknown): void { + const e = err as { code?: string; syscall?: string }; + const isIoredisConnectTimeout = e?.code === "ETIMEDOUT" && e?.syscall === "connect"; + const isTransientNodeRedisError = + err instanceof ConnectionTimeoutError || err instanceof SocketClosedUnexpectedlyError; + if (isIoredisConnectTimeout || isTransientNodeRedisError) return; + + console.log(`(${source}) ERROR EVENT`, err); +} + +/** + * A client for the test Redis of the current datastore (connection started, + * error listener attached), which the caller owns and must close with + * closeTestClient. Unexpected errors are logged unless `expectErrors`. + */ +export function makeTestClient( + clientOptions?: Record, + meta: { expectErrors?: boolean; datastore?: string } = {}, +): NodeRedisClient | IORedisClient { + const datastore = meta.datastore ?? process.env.DATASTORE; + const onError = (err: unknown) => { + if (!meta.expectErrors) logUnexpectedError("makeTestClient", err); + }; + if (datastore === "ioredis") { + const client = new IORedis(clientOptions ?? buildClientOptions("ioredis")); + client.on("error", onError); + return client; + } + const client = RedisClient.createClient(clientOptions ?? buildClientOptions("redis")); + client.on("error", onError); + // isOpen turns true synchronously, which is all RedisConnection needs. + client.connect().catch(onError); + return client; +} + +/** + * Close a client from makeTestClient. node-redis clients close gracefully: + * destroy() rejects whatever node-redis itself still has queued, which + * surfaced as unhandled "Disconnects client" rejections in Group teardown + * even with none of Bottleneck's commands in flight. + */ +export async function closeTestClient(client: NodeRedisClient | IORedisClient): Promise { + if ("status" in client) { + client.disconnect(); + } else if (client.isOpen) { + await (typeof client.close === "function" ? client.close() : client.quit()); + } +} diff --git a/test/helpers/limiter.ts b/test/helpers/limiter.ts index 18bce0c9..e40d8f4b 100644 --- a/test/helpers/limiter.ts +++ b/test/helpers/limiter.ts @@ -1,21 +1,15 @@ -import IORedis from "ioredis"; -import { ConnectionTimeoutError, SocketClosedUnexpectedlyError } from "redis"; -import RedisClient from "redis"; import type BottleneckBase from "../../src/Bottleneck"; import type { BottleneckOptions } from "../../src/types"; import Bottleneck from "../bottleneck"; import buildClientOptions from "../redis-client-options"; - -function setRedisClientOptions(options: Record) { - if (options.clientOptions == null) { - options.clientOptions = buildClientOptions(String(options.datastore)); - } -} +import { logUnexpectedError } from "./clients"; /** * Create a Bottleneck limiter pre-wired for the current test environment. * - * @param {object} [options] - Bottleneck constructor options (datastore/clientOptions auto-filled from env) + * @param {object} [options] - Bottleneck constructor options. Without a `connection`, the + * `datastore` harness flag defaults to the env's DATASTORE: test/bottleneck.ts builds a + * Redis-backed limiter (from `clientOptions`, if given) or a local one from it. * @param {{ expectErrors?: boolean }} [meta] - Test-level flags kept separate from Bottleneck options * @returns {import("../../src/Bottleneck").default} */ @@ -23,37 +17,17 @@ function makeLimiter( options: Record = {}, meta: { expectErrors?: boolean } = {}, ): BottleneckBase { - const assigned = Object.assign({}, options) as Record; - options = assigned; - - if (options.datastore == null) { - if (process.env.DATASTORE === "redis") { - assigned.datastore = "redis"; - assigned.Redis ??= RedisClient; - } else if (process.env.DATASTORE === "ioredis") { - assigned.datastore = "ioredis"; - assigned.Redis ??= IORedis; - } else { - assigned.datastore = "local"; - } - } - - if (options.datastore === "redis" || options.datastore === "ioredis") { - setRedisClientOptions(options); + const assigned = { ...options }; + if (assigned.datastore == null && assigned.connection == null) { + assigned.datastore = process.env.DATASTORE ?? "local"; } - const limiter = new Bottleneck(options as BottleneckOptions); + // Harness options (the `datastore` flag, `clientOptions`) are translated + // into product options by test/bottleneck.ts. + const limiter = new Bottleneck(assigned as BottleneckOptions); if (!meta.expectErrors) { - limiter.on("error", (err) => { - const e = err as { code?: string; syscall?: string }; - const isIoredisConnectTimeout = e?.code === "ETIMEDOUT" && e?.syscall === "connect"; - const isTransientNodeRedisError = - err instanceof ConnectionTimeoutError || err instanceof SocketClosedUnexpectedlyError; - if (isIoredisConnectTimeout || isTransientNodeRedisError) return; - - console.log("(makeLimiter) ERROR EVENT", err); - }); + limiter.on("error", (err) => logUnexpectedError("makeLimiter", err)); } // makeLimiter is synchronous; suppress the unhandled-rejection from ready() @@ -66,13 +40,3 @@ 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 787b85f0..9cb61a19 100644 --- a/test/helpers/test-api.ts +++ b/test/helpers/test-api.ts @@ -1,10 +1,17 @@ import { test as baseTest, expect as vitestExpect, vi } from "vitest"; import type BottleneckBase from "../../src/Bottleneck"; -import type { BottleneckOptions, JobOptions } from "../../src/types"; +import type { IORedisClient, NodeRedisClient } from "../../src/cluster/redis-types"; +import type { + BottleneckOptions, + IORedisConnectionOptions, + JobOptions, + RedisConnectionOptions, +} from "../../src/types"; import Bottleneck from "../bottleneck"; +import { closeTestClient, makeTestClient } from "./clients"; import { isFakeClock } from "./clock"; import { createTaskFns } from "./job-tasks"; -import makeLimiterHelper, { defaultConnectionOptions } from "./limiter"; +import makeLimiterHelper from "./limiter"; import { wrongType } from "./wrong-type"; export { waitForState } from "./wait-for-state"; @@ -24,13 +31,19 @@ export type JobHarness = ReturnType; type MakeLimiter = (opts?: LimiterOptions, meta?: { expectErrors?: boolean }) => Limiter; type MakeGroup = (opts?: LimiterOptions) => Group; -type MakeConnection = (opts?: LimiterOptions) => Connection; +type Client = NodeRedisClient | IORedisClient; +type MakeClient = ( + clientOptions?: Record, + meta?: { expectErrors?: boolean }, +) => Client; +type MakeConnection = (opts?: { client: Client; subscriber?: Client }) => Connection; export interface TestFixtures { harness: JobHarness; track: Track; makeLimiter: MakeLimiter; makeGroup: MakeGroup; + makeClient: MakeClient; makeConnection: MakeConnection; limiter: Limiter; limiterOptions: LimiterOptions; @@ -198,6 +211,7 @@ export const test = baseTest.extend<{ track: (resource: T) => T; makeLimiter: MakeLimiter; makeGroup: MakeGroup; + makeClient: MakeClient; makeConnection: MakeConnection; limiter: Limiter; limiterOptions: Record; @@ -234,15 +248,23 @@ export const test = baseTest.extend<{ track(new Bottleneck.Group(opts ?? {}) as unknown as Group), ); }, - async makeConnection({ track }, use) { - await use((opts?: LimiterOptions) => { - const Connection = + async makeClient({ track }, use) { + await use((clientOptions, meta) => { + const client = makeTestClient(clientOptions, meta); + track({ disconnect: () => closeTestClient(client) }); + return client; + }); + }, + async makeConnection({ track, makeClient }, use) { + await use((opts) => { + // The fixture picks the connection class for the current datastore, so + // the client's type isn't tied to it here. + const options = opts ?? { client: makeClient() }; + const connection = process.env.DATASTORE === "ioredis" - ? Bottleneck.IORedisConnection - : Bottleneck.RedisConnection; - return track( - new Connection(wrongType(opts ?? defaultConnectionOptions())) as unknown as Connection, - ); + ? new Bottleneck.IORedisConnection(wrongType(options)) + : new Bottleneck.RedisConnection(wrongType(options)); + return track(connection); }); }, async limiter({ makeLimiter, limiterOptions, limiterMeta }, use) { diff --git a/test/ioredis.test.ts b/test/ioredis.test.ts index 6c418721..2b4b2d63 100644 --- a/test/ioredis.test.ts +++ b/test/ioredis.test.ts @@ -1,43 +1,11 @@ import Redis from "ioredis"; import ioredisPkg from "ioredis/package.json" with { type: "json" }; -import { describe, expect, vi } from "vitest"; -import { redisStore } from "./helpers/store"; -import { test, waitForState } from "./helpers/test-api"; +import { describe, expect, onTestFinished, vi } from "vitest"; +import { limiterKeys } from "./helpers/store"; +import { test } from "./helpers/test-api"; import buildClientOptions from "./redis-client-options"; describe("ioredis-only", () => { - test("Should accept ioredis lib override", ({ makeLimiter }) => { - const limiter = makeLimiter({ - maxConcurrent: 2, - Redis, - clientOptions: {}, - clusterNodes: [ - { - host: process.env.REDIS_HOST, - port: process.env.REDIS_PORT, - }, - ], - }); - - expect(limiter.datastore).toStrictEqual("ioredis"); - }); - - test("Should connect in Redis Cluster mode", ({ makeLimiter }) => { - const limiter = makeLimiter({ - maxConcurrent: 2, - clientOptions: {}, - clusterNodes: [ - { - host: process.env.REDIS_HOST, - port: process.env.REDIS_PORT, - }, - ], - }); - - expect(limiter.datastore).toStrictEqual("ioredis"); - expect(redisStore(limiter).connection.client).toBeInstanceOf(Redis.Cluster); - }); - test("Should connect in Redis Cluster mode with premade client", ({ makeLimiter, makeConnection, @@ -65,10 +33,7 @@ describe("ioredis-only", () => { makeLimiter, makeConnection, }) => { - const connection = makeConnection({ - Redis, - clientOptions: buildClientOptions("ioredis"), - }); + const connection = makeConnection(); const limiter = makeLimiter({ minTime: 50, connection, @@ -84,7 +49,7 @@ describe("ioredis-only", () => { expect(limiter.datastore).toStrictEqual("ioredis"); await limiter.disconnect(); - expect(limiter.clients().client).toHaveProperty("status", "ready"); + expect(connection.client).toHaveProperty("status", "ready"); await Promise.all([expect(p1).resolves.toEqual([1]), expect(p2).resolves.toEqual([2])]); }); @@ -94,6 +59,7 @@ describe("ioredis-only", () => { makeConnection, }) => { const client = new Redis(buildClientOptions("ioredis")); + onTestFinished(() => client.disconnect()); const connection = makeConnection({ client }); const limiter = makeLimiter({ @@ -107,26 +73,24 @@ describe("ioredis-only", () => { await h.flushLimiter(limiter); expect(h.log).toHaveCallOrder([[1], [2]]); expect(h).toHaveFinalCallAt(50); - expect(limiter.clients().client).toBe(client); + expect(connection.client).toBe(client); expect(limiter.connection).toBe(connection); expect(limiter.datastore).toStrictEqual("ioredis"); await limiter.disconnect(); - expect(limiter.clients().client).toHaveProperty("status", "ready"); + expect(connection.client).toHaveProperty("status", "ready"); await Promise.all([expect(p1).resolves.toEqual([1]), expect(p2).resolves.toEqual([2])]); }); test("Should trigger error events on the shared connection", ({ makeLimiter, + makeClient, makeConnection, }) => { expect.hasAssertions(); return new Promise((resolve, reject) => { const connection = makeConnection({ - Redis, - clientOptions: { - port: 1, - }, + client: makeClient({ port: 1 }, { expectErrors: true }), }); let fired = false; const limiter = makeLimiter({ connection }); @@ -185,9 +149,10 @@ describeResp3("ioredis RESP3", () => { }); test("Should normalize RESP3 hgetall and withscores replies in __runCommand__", async ({ + makeClient, makeConnection, }) => { - const connection = makeConnection({ Redis, clientOptions: resp3ClientOptions() }); + const connection = makeConnection({ client: makeClient(resp3ClientOptions()) }); await connection.ready; // Guard against silently testing RESP2: if this ever fails on an ioredis // upgrade, the option name or default protocol changed and these tests @@ -211,14 +176,14 @@ describeResp3("ioredis RESP3", () => { }); test("Should normalize RESP2 replies in __runCommand__ when protocol 2 is forced", async ({ + makeClient, makeConnection, }) => { // ioredis 5 (still a supported peer dep) is RESP2-only; force the same // wire protocol under ioredis 6 so the RESP2 branches of normalizeReply // (HGETALL flat array, WITHSCORES passthrough) run against a real server. const connection = makeConnection({ - Redis, - clientOptions: { ...buildClientOptions("ioredis"), protocol: 2 }, + client: makeClient({ ...buildClientOptions("ioredis"), protocol: 2 }), }); await connection.ready; // ioredis 5 has no protocol option (RESP2-only); only assert under v6+. @@ -265,36 +230,60 @@ describeResp3("ioredis RESP3", () => { }); }); -// 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 ({ +// The client belongs to the consumer: the connection only sends commands on +// it, never adding listeners or script methods, and never closing it. +describe("ioredis passed-in client", () => { + test("Should only send commands on a passed-in client", async ({ + makeLimiter, makeConnection, }) => { const client = new Redis(buildClientOptions("ioredis")); + onTestFinished(() => client.disconnect()); const consumerListener = vi.fn<() => void>(); client.on("error", consumerListener); + const maxListeners = client.getMaxListeners(); 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)); + const limiter = makeLimiter({ connection }); + await expect(limiter.schedule(() => Promise.resolve("ran"))).resolves.toBe("ran"); + expect(client.listeners("error")).toEqual([consumerListener]); + expect(client.getMaxListeners()).toBe(maxListeners); + // Scripts run through EVALSHA/EVAL, not methods defined on the client. + expect(client).not.toHaveProperty("submit"); + await limiter.disconnect(); await connection.disconnect(true); - expect(client.listeners("error")).not.toContain(consumerListener); - expect(client.listenerCount("error")).toBe(1); - await waitForState(() => expect(client.status).toBe("end")); + expect(client.listeners("error")).toEqual([consumerListener]); + expect(client.status).toBe("ready"); + }); + + test("Should apply the client's keyPrefix to script keys", async ({ + makeClient, + makeConnection, + makeLimiter, + }) => { + // Inside the per-fork prefix, so test cleanup still finds these keys. + const keyPrefix = `b_${process.env.BOTTLENECK_TEST_PREFIX}kp:`; + const connection = makeConnection({ + client: makeClient({ ...buildClientOptions("ioredis"), keyPrefix }), + }); + const limiter = makeLimiter({ id: "prefixed", connection }); + await expect(limiter.schedule(() => Promise.resolve("ran"))).resolves.toBe("ran"); + + const [settingsKey] = limiterKeys(limiter); + const unprefixed = makeConnection(); + expect(await unprefixed.__runCommand__(["exists", `${keyPrefix}${settingsKey}`])).toBe(1); + expect(await unprefixed.__runCommand__(["exists", settingsKey])).toBe(0); }); - test("Should disconnect a passed-in client on disconnect(false)", async ({ makeConnection }) => { + test("Should leave a passed-in client open on disconnect(false)", async ({ makeConnection }) => { const client = new Redis(buildClientOptions("ioredis")); + onTestFinished(() => client.disconnect()); const connection = makeConnection({ client }); await connection.ready; await connection.disconnect(false); - await waitForState(() => expect(client.status).toBe("end")); + await expect(client.ping()).resolves.toBe("PONG"); }); }); diff --git a/test/memory/limiter-memory.test.ts b/test/memory/limiter-memory.test.ts index 0ac36cab..92d5833d 100644 --- a/test/memory/limiter-memory.test.ts +++ b/test/memory/limiter-memory.test.ts @@ -9,7 +9,7 @@ describe("Limiter memory", () => { await iterateAsync( async () => { calls++; - const limiter = new Bottleneck({ datastore: "local" }); + const limiter = new Bottleneck(); await limiter.ready(); return limiter.disconnect(false); }, @@ -19,7 +19,7 @@ describe("Limiter memory", () => { }); it("Should not leak memory running jobs", { timeout: 12000 }, async () => { - const limiter = new Bottleneck({ datastore: "local", maxConcurrent: 1, minTime: 10 }); + const limiter = new Bottleneck({ maxConcurrent: 1, minTime: 10 }); await limiter.ready(); let i = 0; diff --git a/test/node_redis.test.ts b/test/node_redis.test.ts index 2d9e8623..c83ff609 100644 --- a/test/node_redis.test.ts +++ b/test/node_redis.test.ts @@ -1,27 +1,15 @@ import * as Redis from "redis"; -import { describe, expect, vi } from "vitest"; +import { describe, expect, onTestFinished, vi } from "vitest"; import { test } from "./helpers/test-api"; import buildClientOptions from "./redis-client-options"; describe("node_redis-only", () => { - test("Should accept node_redis lib override", ({ makeLimiter }) => { - const limiter = makeLimiter({ - maxConcurrent: 2, - Redis, - }); - - expect(limiter.datastore).toStrictEqual("redis"); - }); - test("Should accept existing connections", async ({ harness: h, makeLimiter, makeConnection, }) => { - const connection = makeConnection({ - Redis, - clientOptions: buildClientOptions("redis"), - }); + const connection = makeConnection(); const limiter = makeLimiter({ minTime: 50, connection, @@ -37,7 +25,7 @@ describe("node_redis-only", () => { expect(limiter.datastore).toStrictEqual("redis"); await limiter.disconnect(); - expect(limiter.clients().client).toHaveProperty("isReady", true); + expect(connection.client).toHaveProperty("isReady", true); await Promise.all([expect(p1).resolves.toEqual([1]), expect(p2).resolves.toEqual([2])]); }); @@ -47,6 +35,7 @@ describe("node_redis-only", () => { makeConnection, }) => { const client = Redis.createClient(buildClientOptions("redis")); + onTestFinished(() => client.disconnect()); await client.connect(); const connection = makeConnection({ client }); @@ -61,29 +50,27 @@ describe("node_redis-only", () => { await h.flushLimiter(limiter); expect(h.log).toHaveCallOrder([[1], [2]]); expect(h).toHaveFinalCallAt(50); - expect(limiter.clients().client).toBe(client); + expect(connection.client).toBe(client); expect(limiter.connection).toBe(connection); expect(limiter.datastore).toStrictEqual("redis"); await limiter.disconnect(); - expect(limiter.clients().client).toHaveProperty("isReady", true); + expect(connection.client).toHaveProperty("isReady", true); await Promise.all([expect(p1).resolves.toEqual([1]), expect(p2).resolves.toEqual([2])]); }); test("Should trigger error events on the shared connection", ({ makeLimiter, + makeClient, makeConnection, }) => { expect.hasAssertions(); return new Promise((resolve, reject) => { const connection = makeConnection({ - Redis, - clientOptions: { - socket: { - port: 1, - reconnectStrategy: () => false, - }, - }, + client: makeClient( + { socket: { port: 1, reconnectStrategy: () => false } }, + { expectErrors: true }, + ), }); connection.ready.catch(() => {}); let fired = false; @@ -104,37 +91,42 @@ 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 ({ +// The client belongs to the consumer: the connection only sends commands on +// it, never connecting it, adding listeners, or closing it. +describe("node_redis passed-in client", () => { + test("Should require a connected client and only send commands on it", async ({ + makeLimiter, makeConnection, }) => { const client = Redis.createClient(buildClientOptions("redis")); + onTestFinished(() => (client.isOpen ? client.disconnect() : undefined)); const consumerListener = vi.fn<() => void>(); client.on("error", consumerListener); - expect(client.isOpen).toBe(false); + expect(() => makeConnection({ client })).toThrow(/call `client.connect\(\)` first/); + await client.connect(); + const maxListeners = client.getMaxListeners(); const connection = makeConnection({ client }); - await connection.ready; - expect(client.isOpen).toBe(true); - expect(client.listenerCount("error")).toBe(2); - expect(client.getMaxListeners()).toBe(0); + const limiter = makeLimiter({ connection }); + await expect(limiter.schedule(() => Promise.resolve("ran"))).resolves.toBe("ran"); + expect(client.listeners("error")).toEqual([consumerListener]); + expect(client.getMaxListeners()).toBe(maxListeners); + await limiter.disconnect(); await connection.disconnect(true); - expect(client.listeners("error")).not.toContain(consumerListener); - expect(client.listenerCount("error")).toBe(1); - expect(client.isOpen).toBe(false); + expect(client.listeners("error")).toEqual([consumerListener]); + expect(client.isOpen).toBe(true); }); - test("Should destroy a passed-in client on disconnect(false)", async ({ makeConnection }) => { + test("Should leave a passed-in client open on disconnect(false)", async ({ makeConnection }) => { const client = Redis.createClient(buildClientOptions("redis")); + onTestFinished(() => (client.isOpen ? client.disconnect() : undefined)); await client.connect(); const connection = makeConnection({ client }); await connection.ready; await connection.disconnect(false); - expect(client.isOpen).toBe(false); + await expect(client.ping()).resolves.toBe("PONG"); }); }); diff --git a/test/schedule-stack.test.ts b/test/schedule-stack.test.ts index 7816d4b5..094563c7 100644 --- a/test/schedule-stack.test.ts +++ b/test/schedule-stack.test.ts @@ -220,7 +220,7 @@ describe("Schedule stacks", () => { return (super.schedule as (...a: unknown[]) => Promise)(...args); } } - const arrowPatched = track(new Bottleneck({ id: "override-arrow", datastore: "local" })); + const arrowPatched = track(new Bottleneck({ id: "override-arrow" })); const original = Bottleneck.prototype.schedule as (...a: unknown[]) => Promise; function scheduleWithOpts(args: unknown[]): Promise { return original.apply(arrowPatched, args); @@ -229,8 +229,8 @@ describe("Schedule stacks", () => { scheduleWithOpts(args); const limiters = [ - track(new SubclassLimiter({ id: "override-subclass", datastore: "local" })), - track(new AsyncSubclassLimiter({ id: "override-async", datastore: "local" })), + track(new SubclassLimiter({ id: "override-subclass" })), + track(new AsyncSubclassLimiter({ id: "override-async" })), arrowPatched, ]; for (const limiter of limiters) { @@ -259,7 +259,7 @@ describe("Schedule stacks", () => { test("a bound schedule override keeps its wrapper frames", async ({ track }) => { // V8 can't cut at a bound function, so the base cutoff is used instead. - const bound = track(new Bottleneck({ id: "override-bound", datastore: "local" })); + const bound = track(new Bottleneck({ id: "override-bound" })); const original = Bottleneck.prototype.schedule as (...a: unknown[]) => Promise; (bound as { schedule: unknown }).schedule = function scheduleWithOpts( this: Bottleneck, diff --git a/test/smoke/lib.test.ts b/test/smoke/lib.test.ts index 8bfd9f83..610c80ab 100644 --- a/test/smoke/lib.test.ts +++ b/test/smoke/lib.test.ts @@ -12,15 +12,18 @@ // Redis-backed limiter actually evaluates a script. // // Loaded via `test/bottleneck.mjs` with `BOTTLENECK_ENTRY=lib` and -// `DATASTORE=redis` so the wrapper injects the test heartbeat override, -// the per-fork id prefix, and node-redis client options pointing at the +// `DATASTORE=redis` so the wrapper injects the test heartbeat override and +// the per-fork id prefix. The connections are built the way consumers do: +// the bundle's RedisConnection around a node-redis client pointed at the // shared testcontainer Redis (started by the root globalSetup). The // `lib-smoke` project's globalSetup builds the lib bundle before any // test file is collected. import { createRequire } from "node:module"; +import { createClient } from "redis"; import { describe, it, expect } from "vitest"; import Bottleneck from "../bottleneck"; +import buildClientOptions from "../redis-client-options"; describe("dist/index full smoke", () => { it("exposes .default/.Bottleneck on the CJS build for non-interop consumers", () => { @@ -56,20 +59,23 @@ describe("dist/index full smoke", () => { // lua files were missing from the inlined bundle, ioredis/node-redis // would surface a "ERR Error compiling script" or the body would silently // contain `undefined` and Redis would reply with a runtime lua error. + const client = createClient(buildClientOptions("redis")); + await client.connect(); + const connection = new Bottleneck.RedisConnection({ client }); const limiter = new Bottleneck({ id: "lib-smoke", - datastore: "redis", + connection, maxConcurrent: 1, clearDatastore: true, }); try { - const clients = await limiter.ready(); - expect(Object.keys(clients as Record)).toEqual(["client", "subscriber"]); - + await limiter.ready(); const result = await limiter.schedule(() => "ok"); expect(result).toBe("ok"); } finally { await limiter.disconnect(false); + await connection.disconnect(false); + await client.disconnect(); } }); @@ -79,9 +85,12 @@ describe("dist/index full smoke", () => { // wiring path (Scripts loaded through the Group's RedisDatastore) and // catches regressions where the inlined lua map is reachable from one // entry point but not another. + const client = createClient(buildClientOptions("redis")); + await client.connect(); + const connection = new Bottleneck.RedisConnection({ client }); const group = new Bottleneck.Group({ id: "lib-smoke-group", - datastore: "redis", + connection, maxConcurrent: 1, clearDatastore: true, }); @@ -93,6 +102,8 @@ describe("dist/index full smoke", () => { expect(rb).toBe("b"); } finally { await group.disconnect(false); + await connection.disconnect(false); + await client.disconnect(); } }); }); diff --git a/test/smoke/light.test.ts b/test/smoke/light.test.ts index ba72d8af..ff06af00 100644 --- a/test/smoke/light.test.ts +++ b/test/smoke/light.test.ts @@ -1,5 +1,7 @@ import { describe, it, expect } from "vitest"; +import type { IORedisConnectionOptions, RedisConnectionOptions } from "../../src/types"; import Bottleneck from "../bottleneck"; +import { wrongType } from "../helpers/wrong-type"; describe("dist/light smoke", () => { it("loads", () => { @@ -14,9 +16,12 @@ describe("dist/light smoke", () => { await limiter.disconnect(false); }); - it("throws when clustering datastore is requested", () => { - expect(() => { - void new Bottleneck({ datastore: "redis" }); - }).toThrow(/full version of Bottleneck/i); + it("throws when a clustering connection is requested", () => { + expect(() => new Bottleneck.RedisConnection(wrongType({}))).toThrow( + /full version of Bottleneck/i, + ); + expect(() => new Bottleneck.IORedisConnection(wrongType({}))).toThrow( + /full version of Bottleneck/i, + ); }); }); diff --git a/test/types.test.ts b/test/types.test.ts index 31ede541..af16442f 100644 --- a/test/types.test.ts +++ b/test/types.test.ts @@ -1,7 +1,9 @@ +import type { Cluster, Redis as IORedis } from "ioredis"; +import type { createClient } from "redis"; import { describe, expect, expectTypeOf, it } from "vitest"; import type { BatcherOptions, - ClientsList, + BottleneckOptions, Counts, EventInfoDropped, EventInfoQueued, @@ -18,19 +20,19 @@ import Bottleneck, { RedisConnection, } from "../src/index"; +type NodeRedis = ReturnType; + +// Never called: the connection type assertions below are compile-time. +const connectIORedis = (client: IORedis) => new IORedisConnection({ client }); +const connectCluster = (client: Cluster) => new IORedisConnection({ client }); +const connectNodeRedis = (client: NodeRedis) => new RedisConnection({ client }); + /* * Type-level contract test for the published surface. The generated dts * (dist/index.d.mts) is produced from these same source types, so asserting * on them here guards the published type contract. */ -const makeFakeRedisClient = () => ({ - setMaxListeners() {}, - on() {}, - once() {}, - duplicate: () => makeFakeRedisClient(), -}); - describe("Bottleneck type contract", () => { it("exposes the strategy constants as literal types", () => { expectTypeOf(Bottleneck.strategy.LEAK).toEqualTypeOf<1>(); @@ -145,41 +147,37 @@ describe("Bottleneck type contract", () => { batcher.on("batch", (batch) => expectTypeOf(batch).toEqualTypeOf()); }); - it("exposes connection clients and channels", () => { + it("exposes the connection, readiness, and channels", () => { const limiter = new Bottleneck(); - expectTypeOf(limiter.clients()).toEqualTypeOf(); + expectTypeOf(limiter.ready()).toEqualTypeOf>(); + expectTypeOf(limiter.connection).toEqualTypeOf(); expectTypeOf(limiter.channel()).toEqualTypeOf(); }); - it("rejects connection options with both Redis and client", () => { - // Minimal stand-ins: the constructor runs for real (ready() stays pending - // because these fakes never emit "ready"), the assertions here are - // compile-time. They must satisfy _setup's synchronous calls. - const nodeRedisFake = { - createClient: makeFakeRedisClient, - }; - const clientFake = makeFakeRedisClient(); - class IORedisFake { - setMaxListeners() {} - on() {} - once() {} - duplicate() { - return makeFakeRedisClient(); - } - } - - // Valid: either branch alone constructs. - void new Bottleneck.RedisConnection({ Redis: nodeRedisFake }); - void new Bottleneck.RedisConnection({ client: clientFake }); - void new Bottleneck.IORedisConnection({ Redis: IORedisFake }); - - // @ts-expect-error Redis and client are mutually exclusive - void new Bottleneck.RedisConnection({ Redis: nodeRedisFake, client: clientFake }); - // @ts-expect-error ditto for ioredis - void new Bottleneck.IORedisConnection({ Redis: IORedisFake, client: clientFake }); - - expect(typeof Bottleneck.RedisConnection).toBe("function"); - expect(typeof Bottleneck.IORedisConnection).toBe("function"); + it("takes Redis only as a connection", () => { + expectTypeOf().toEqualTypeOf< + RedisConnection | IORedisConnection | null | undefined + >(); + // Never called: the assertions are compile-time. + void ((client: IORedis) => new Bottleneck({ connection: connectIORedis(client) })); + void ((client: NodeRedis) => new Bottleneck.Group({ connection: connectNodeRedis(client) })); + + // @ts-expect-error the pre-v5 library options are gone + void ((Redis: unknown) => new Bottleneck({ Redis, clientOptions: {} })); + // @ts-expect-error so is the datastore switch + void (() => new Bottleneck({ datastore: "ioredis" })); + }); + + it("infers a connection's client type from the client passed in", () => { + expectTypeOf["client"]>().toEqualTypeOf(); + expectTypeOf["subscriber"]>().toEqualTypeOf(); + expectTypeOf["client"]>().toEqualTypeOf(); + expectTypeOf["client"]>().toEqualTypeOf(); + + // @ts-expect-error a node-redis client isn't an ioredis client + void ((client: NodeRedis) => new IORedisConnection({ client })); + // @ts-expect-error connections take a client, not the library + void ((Redis: unknown) => new RedisConnection({ Redis })); }); it("keeps option types structural", () => {