diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml index eaa318e4c..e3cda5d33 100644 --- a/.github/workflows/main.yaml +++ b/.github/workflows/main.yaml @@ -80,7 +80,7 @@ jobs: ports: - 5432:5432 redis: - image: redis + image: redis:7.2.17 options: >- --health-cmd "redis-cli ping" --health-interval 10s @@ -171,7 +171,7 @@ jobs: ports: - 5432:5432 redis: - image: redis + image: redis:7.2.17 options: >- --health-cmd "redis-cli ping" --health-interval 10s @@ -235,7 +235,7 @@ jobs: ports: - 5432:5432 redis: - image: redis + image: redis:7.2.17 options: >- --health-cmd "redis-cli ping" --health-interval 10s @@ -270,6 +270,73 @@ jobs: env: AMQP_ORDERING_TEST: "true" + test-redis-atomic: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + redis: ["7.2.17", "7.4.12", "8.10.3"] + services: + redis: + image: redis:${{ matrix.redis }} + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 5s + --health-timeout 5s + --health-retries 5 + redis-admin: + image: redis:${{ matrix.redis }} + ports: + - 6380:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 5s + --health-timeout 5s + --health-retries 5 + env: + FEDIFY_REDIS_REQUIRE_ATOMIC_TESTS: "true" + REDIS_URL: redis://127.0.0.1:6379 + REDIS_ATOMIC_ADMIN_URL: redis://127.0.0.1:6380 + REDIS_CLUSTER_NODES: 127.0.0.1:7000,127.0.0.1:7001,127.0.0.1:7002 + REDIS_ATOMIC_CLUSTER_ADMIN_NODES: 127.0.0.1:7000,127.0.0.1:7001,127.0.0.1:7002 + REDIS_ATOMIC_MIGRATION_NODES: 127.0.0.1:7000,127.0.0.1:7001,127.0.0.1:7002 + steps: + - uses: actions/checkout@v6 + - uses: ./.github/actions/setup-mise + - name: Start a dedicated three-primary Redis Cluster + env: + REDIS_IMAGE: redis:${{ matrix.redis }} + run: | + for port in 7000 7001 7002; do + docker run --detach --name "redis-cluster-$port" --network host \ + "$REDIS_IMAGE" redis-server --bind 127.0.0.1 --port "$port" \ + --cluster-enabled yes --cluster-config-file /data/nodes.conf \ + --cluster-announce-ip 127.0.0.1 --save "" --appendonly no + for _ in $(seq 1 30); do + if docker exec "redis-cluster-$port" redis-cli -p "$port" ping | grep -q PONG; then + break + fi + sleep 1 + done + docker exec "redis-cluster-$port" redis-cli -p "$port" ping | grep -q PONG + done + docker exec redis-cluster-7000 redis-cli --cluster create \ + 127.0.0.1:7000 127.0.0.1:7001 127.0.0.1:7002 \ + --cluster-replicas 0 --cluster-yes + for port in 7000 7001 7002; do + for _ in $(seq 1 30); do + if docker exec "redis-cluster-$port" redis-cli -p "$port" cluster info | grep -q 'cluster_state:ok'; then + break + fi + sleep 1 + done + docker exec "redis-cluster-$port" redis-cli -p "$port" cluster info | grep -q 'cluster_state:ok' + done + # The task runs Deno, Node.js and Bun sequentially on exclusive fixtures. + - run: mise run test-each redis + test-cfworkers: runs-on: ubuntu-latest env: diff --git a/CHANGES.md b/CHANGES.md index 012ee8b83..44feb87ba 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -429,10 +429,37 @@ To be released. the index exists, writes whose expiration overflows PostgreSQL's timestamp range fail atomically, including writes from older versions. [[#1309], [#1334]] + - Fixed duplicate expiry-index creation when reinitializing `PostgresKvStore` + with an existing `AT TIME ZONE` index on PostgreSQL 18.7. Index detection + now checks the stored expression rather than index-definition syntax, + preserving both documented `timezone()` and `AT TIME ZONE` forms. + [[#1434]] [index migration guide]: https://fedify.dev/manual/kv#postgresql-expiry-index [#1309]: https://github.com/fedify-dev/fedify/issues/1309 [#1334]: https://github.com/fedify-dev/fedify/pull/1334 +[#1434]: https://github.com/fedify-dev/fedify/pull/1434 + +### @fedify/redis + + - Added atomic snapshots and multi-key writes, deletes, and compare-and-swap + through `RedisKvStore.getMany()`, `setMany()`, `deleteMany()`, and + `casMany()`. Batches support standalone Redis and one Redis Cluster hash + slot on official Redis 7.2.17, 7.4.12, and 8.10.3; other releases reject + nonempty batches until audited and added in a Fedify update. + These restrictions preserve the audited atomicity guarantees; select a + supported Redis release for nonempty batches. Existing keys are not + relocated. [[#1361], [#1365], [#1434]] + + Storing `undefined` or a top-level value that JSON cannot encode now + rejects instead of writing undecodable bytes. CAS deletions with a + non-null TTL reject even when the comparison would mismatch. CAS + success results now also handle ioredis `stringNumbers` clients. Remove + TTLs from CAS deletions when upgrading affected calls. Ordinary + single-key methods retain their existing Redis version and TTL range + support. + +[#1365]: https://github.com/fedify-dev/fedify/issues/1365 ### @fedify/sqlite diff --git a/changes.d/postgres/expiry-index-expression.md b/changes.d/postgres/expiry-index-expression.md new file mode 100644 index 000000000..b981ac029 --- /dev/null +++ b/changes.d/postgres/expiry-index-expression.md @@ -0,0 +1,9 @@ +--- +links: + '#1434': https://github.com/fedify-dev/fedify/pull/1434 +--- + - Fixed duplicate expiry-index creation when reinitializing `PostgresKvStore` + with an existing `AT TIME ZONE` index on PostgreSQL 18.7. Index detection + now checks the stored expression rather than index-definition syntax, + preserving both documented `timezone()` and `AT TIME ZONE` forms. + [[#1434]] diff --git a/changes.d/redis/atomic-kv-store.md b/changes.d/redis/atomic-kv-store.md new file mode 100644 index 000000000..706e2f519 --- /dev/null +++ b/changes.d/redis/atomic-kv-store.md @@ -0,0 +1,22 @@ +--- +links: + '#1361': https://github.com/fedify-dev/fedify/issues/1361 + '#1365': https://github.com/fedify-dev/fedify/issues/1365 + '#1434': https://github.com/fedify-dev/fedify/pull/1434 +--- + - Added atomic snapshots and multi-key writes, deletes, and compare-and-swap + through `RedisKvStore.getMany()`, `setMany()`, `deleteMany()`, and + `casMany()`. Batches support standalone Redis and one Redis Cluster hash + slot on official Redis 7.2.17, 7.4.12, and 8.10.3; other releases reject + nonempty batches until audited and added in a Fedify update. + These restrictions preserve the audited atomicity guarantees; select a + supported Redis release for nonempty batches. Existing keys are not + relocated. [[#1361], [#1365], [#1434]] + + Storing `undefined` or a top-level value that JSON cannot encode now + rejects instead of writing undecodable bytes. CAS deletions with a + non-null TTL reject even when the comparison would mismatch. CAS + success results now also handle ioredis `stringNumbers` clients. Remove + TTLs from CAS deletions when upgrading affected calls. Ordinary + single-key methods retain their existing Redis version and TTL range + support. diff --git a/claude-plugin/skills/fedify/SKILL.md b/claude-plugin/skills/fedify/SKILL.md index b7f3a6665..ea633aa5f 100644 --- a/claude-plugin/skills/fedify/SKILL.md +++ b/claude-plugin/skills/fedify/SKILL.md @@ -482,8 +482,8 @@ for a shared backend. Separate calls do not form a transaction; uncertain commit results need receipts or version checks, and ABA detection needs a non-reused version. -`MemoryKvStore`, `SqliteKvStore`, and `PgliteKvStore` implement the full atomic -interface. +`MemoryKvStore`, `RedisKvStore`, `SqliteKvStore`, and `PgliteKvStore` +implement the atomic interface within their documented scopes. SQLite covers one configured table/database on a writable, normally locking connection. Nonempty batches and ordinary mutations reject caller-owned transactions and use one transaction; snapshots and mismatching CAS can reject @@ -504,10 +504,24 @@ engine insertion failures roll back, while mismatch does not prove proposed writes would fit. Do not infer crash durability, browser persistence, or worker-proxy support. Read the PGlite package README's synchronization policy. -The PostgreSQL, Redis, and MySQL atomic extensions are tracked separately in -issues #1364, #1365, and #1368; Deno KV and Netlify Blobs have no atomic adapter -child issue. `WorkersKvStore` lacks CAS. `WorkersD1KvStore` supports -single-key CAS; its atomic extension is tracked in #1379. Check the +Redis batches currently require official Redis 7.2.17, +7.4.12, or 8.10.3, with no debugger or module command hooks; other releases +reject nonempty batches until audited and added in a Fedify update. The version +guard does not authenticate forks or modified engines. On Cluster, all final +physical keys must share one slot in database zero, including ioredis's client +prefix. Choose an explicit hash tag for a new namespace; existing keys are not +automatically moved. Each batch request is limited to 128 entries and 1 MiB of +key, encoded value/expectation, and argument data. TTLs round up to seconds +with a minimum of one second and batch maximum of 2,147,483,647 seconds. With +`JsonCodec`, CAS compares encoded bytes, so object property order matters. +Grant read/write key access for `EVAL`, including snapshots. Snapshot reads +route to primaries, but Cluster `list()` retains its existing `SCAN` coverage +limitation. Redis persistence/failover and ioredis replays remain separate from +per-call atomicity; a later CAS mismatch does not prove a prior attempt never +committed. The PostgreSQL and MySQL atomic extensions are tracked separately in +issues #1364 and #1368; Deno KV and Netlify Blobs have no atomic adapter +child issue. `WorkersKvStore` lacks CAS. `WorkersD1KvStore` supports +single-key CAS; its atomic extension is tracked in #1379. Check the [atomic support matrix] and installed declarations before selecting a production adapter. diff --git a/docs/manual/kv.md b/docs/manual/kv.md index 672166774..270bb3d0c 100644 --- a/docs/manual/kv.md +++ b/docs/manual/kv.md @@ -260,7 +260,68 @@ const federation = createFederation({ }); ~~~~ +#### Atomic operations + +Since Fedify 2.5.0, `RedisKvStore` implements `AtomicKvStore` on official Redis +7.2.17, 7.4.12, and 8.10.3. Other releases reject nonempty batch calls before +reading or writing. Supporting another Redis release requires a new audit and +Fedify update; ordinary single-key methods keep their existing compatibility. +Modified engines, forks, script debugging, and module command hooks are not +supported atomic configurations, and the version guard does not authenticate +the server. + +On standalone Redis, batches cover one selected database on the primary. On +Cluster, all effective physical keys must share a slot in database zero. For a +new namespace, the existing `keyPrefix` option can select that scope: + +~~~~ typescript twoslash +import { RedisKvStore } from "@fedify/redis"; +import { Cluster } from "ioredis"; + +const redis = new Cluster([{ host: "127.0.0.1", port: 7000 }]); +const store = new RedisKvStore(redis, { keyPrefix: "{inventory}::" }); +const values = await store.getMany([["available"], ["reserved"]]); +~~~~ + +Keys are not automatically tagged or moved. Changing a prefix selects a new +namespace and needs an application migration for existing data. The first `{` +and following `}` select a hash tag only if the pair is nonempty; an empty pair +hashes the whole key. A tag in the ioredis client's own `keyPrefix` can take +precedence over a tag in the adapter's `keyPrefix`. Cross-slot calls reject, +including different slots on the same master; they never split into separate +operations. Missing or expired keys can cause an entire batch to reject during +resharding. + +Each request is limited to 128 entries and 1 MiB of effective key, encoded +values and expectations, and argument data, excluding the fixed script and +protocol framing. Batch TTLs round up to whole seconds, with a minimum of one +second and maximum of 2,147,483,647 seconds. Zero and negative TTLs therefore +keep a value for one second. Omitting a TTL clears the previous expiry. +Single-key TTLs keep their existing range handling. CAS compares encoded bytes; +with the default JSON codec, object property order matters. Values must +round-trip through their codec; `undefined` is never stored (as a CAS +`newValue` it deletes the key), while `null` remains a value. + +Before changing keys, the scripts read every target and check each write +command's permissions. +They require `EVAL`, `GET` or `EXISTS`, and the applicable `SET`, `SETEX`, or +`DEL` access. Redis checks `EVAL`'s read/write key access before entering the +script, so key patterns must grant both read and write access even for +`getMany()` snapshots. Snapshots route to the primary despite Cluster +`scaleReads`. The existing Cluster prefix-listing `SCAN` limitation remains: +listing does not reliably visit every node, and `getMany()` does not snapshot +`list()`. + +Configured memory pressure can reject batch deletes as well as writes; +ordinary `delete()` remains available. A successful batch is not a persistence +or failover guarantee. ioredis can replay requests after disconnection, so even +a CAS mismatch reply does not prove an earlier attempt never committed. +Re-read and re-evaluate application rules after a mismatch or uncertain +response. See the [Redis package guide] +for codecs, supported configurations, limits, and retry behavior. + [`RedisKvStore`]: https://jsr.io/@fedify/redis/doc/kv/~/RedisKvStore +[Redis package guide]: https://github.com/fedify-dev/fedify/tree/main/packages/redis#atomic-keyvalue-operations ### [`PostgresKvStore`] @@ -1008,16 +1069,16 @@ check or an operation receipt before repeating effects. ### Current support -`MemoryKvStore`, `SqliteKvStore`, and `PgliteKvStore` implement the full -`AtomicKvStore` contract. -Existing single-key CAS support does not imply atomic multi-key support. -The linked issues track separate adapter implementations: +`MemoryKvStore`, `RedisKvStore`, `SqliteKvStore`, and `PgliteKvStore` +implement `AtomicKvStore` within their documented scopes. Existing single-key +CAS support in other adapters does not imply atomic multi-key support. +The linked issues track their implementations: | Implementation | `AtomicKvStore` | Status | | --------------------- | --------------- | ------------------------------------------------------------------------------------------------- | | `MemoryKvStore` | Yes | Development and testing only; one instance. | | `PostgresKvStore` | No | Planned in [#1364]. | -| `RedisKvStore` | No | Planned in [#1365]. | +| `RedisKvStore` | Yes | Official Redis 7.2.17, 7.4.12, or 8.10.3; standalone or one Cluster slot; 128 entries/1 MiB. | | `PgliteKvStore` | Yes | Since 2.5.0; one normal PGlite owner and table, JSONB comparison. | | `SqliteKvStore` | Yes | One configured table/database; writable handles, JSON values; see [SQLite scope](#sqlitekvstore). | | `MysqlKvStore` | No | Planned in [#1368]. | @@ -1028,7 +1089,6 @@ The linked issues track separate adapter implementations: [PR #1360]: https://github.com/fedify-dev/fedify/pull/1360 [#1364]: https://github.com/fedify-dev/fedify/issues/1364 -[#1365]: https://github.com/fedify-dev/fedify/issues/1365 [#1368]: https://github.com/fedify-dev/fedify/issues/1368 [#1379]: https://github.com/fedify-dev/fedify/issues/1379 @@ -1622,6 +1682,13 @@ when no new TTL is supplied. Reject unsupported batches before any mutation; do not split calls to fit backend limits. If a later statement fails, roll back all earlier writes or deletes. +`RedisKvStore` uses bounded scripts and a restricted engine list because Redis +scripts do not undo earlier writes when a later command fails. Its complete +input encoding, execution-time key reads, command-permission checks, and script +bounds are part of the all-or-nothing argument. An isolated `EVAL` or a +`MULTI`/`EXEC` sequence alone does not establish this contract. Keep those Redis +constraints when extending its implementation. + #### Implement `~AtomicKvStore.casMany()` method Protect the complete read set against competing clients, including absent and diff --git a/packages/postgres/src/kv.index.test.ts b/packages/postgres/src/kv.index.test.ts index fc985ad77..842341ddf 100644 --- a/packages/postgres/src/kv.index.test.ts +++ b/packages/postgres/src/kv.index.test.ts @@ -118,19 +118,21 @@ test( assert.equal(initial.length, 1); assert.equal(initial[0].indisvalid, true); assert.match(initial[0].definition, /timezone\('UTC'/); - const [{ name }] = await sql` - SELECT c.relname AS name FROM pg_index i - JOIN pg_class c ON c.oid = i.indexrelid - WHERE i.indrelid = ${tableName}::regclass AND NOT i.indisprimary - `; - await sql`DROP INDEX ${sql(name)}`; - await manualIndex(sql, tableName, true); - statements.length = 0; - await new PostgresKvStore(sql, { - tableName: `public.${tableName}`, - }).initialize(); - assert.equal((await indexes(sql, tableName)).length, 1); - assert(!statements.some((q) => /LOCK TABLE|CREATE INDEX/.test(q))); + for (const atTimeZone of [false, true]) { + const [{ name }] = await sql` + SELECT c.relname AS name FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + WHERE i.indrelid = ${tableName}::regclass AND NOT i.indisprimary + `; + await sql`DROP INDEX ${sql(name)}`; + await manualIndex(sql, tableName, atTimeZone); + statements.length = 0; + await new PostgresKvStore(sql, { + tableName: `public.${tableName}`, + }).initialize(); + assert.equal((await indexes(sql, tableName)).length, 1); + assert(!statements.some((q) => /LOCK TABLE|CREATE INDEX/.test(q))); + } } finally { await store.drop(); await sql.end(); diff --git a/packages/postgres/src/kv.ts b/packages/postgres/src/kv.ts index 0e1e1d553..8649f544c 100644 --- a/packages/postgres/src/kv.ts +++ b/packages/postgres/src/kv.ts @@ -397,6 +397,8 @@ export class PostgresKvStore implements KvStore { WHERE c.oid = (SELECT typrelid FROM pg_type WHERE oid = pg_typeof(NULL::${sql(this.#tableName)})::oid) `; + // Compare the stored expression, not index-definition syntax whose + // enclosing parentheses can change between PostgreSQL minor releases. const indexed = async () => { const [result] = await sql` SELECT EXISTS ( @@ -409,7 +411,7 @@ export class PostgresKvStore implements KvStore { AND NOT i.indisunique AND i.indnatts = 1 AND a.amname = 'btree' AND i.indkey[0] = 0 AND o.opcdefault AND o.opcintype = 'timestamptz'::regtype - AND pg_get_indexdef(i.indexrelid, 1, false) IN ( + AND pg_get_expr(i.indexprs, i.indrelid, false) IN ( ${"timezone('UTC'::text, (timezone('UTC'::text, created) + ttl))"}, ${"(((created AT TIME ZONE 'UTC'::text) + ttl) AT TIME ZONE 'UTC'::text)"} ) diff --git a/packages/redis/README.md b/packages/redis/README.md index 3372a840c..5e82590e9 100644 --- a/packages/redis/README.md +++ b/packages/redis/README.md @@ -57,3 +57,139 @@ yarn add @fedify/redis # Yarn bun add @fedify/redis # Bun deno add jsr:@fedify/redis # Deno ~~~~ + + +Atomic key–value operations +--------------------------- + +Since Fedify 2.5.0, `RedisKvStore` implements `AtomicKvStore` on official Redis +7.2.17, 7.4.12, and 8.10.3. Nonempty `getMany()`, `setMany()`, `deleteMany()`, +and `casMany()` calls reject on other releases before reading or writing. +The supported release list is deliberately exact: adding a Redis release +requires auditing its scripting and Cluster behavior and updating Fedify. +Ordinary single-key methods, including `cas()`, retain their existing Redis +version support. The version check does not authenticate the server; modified +engines, forks, script debugging, and module command hooks are outside the +atomic support scope. + +On standalone Redis, an operation covers the selected database on the primary. +On Redis Cluster, all effective physical keys must share a hash slot in +database zero. Redis checks the slot after ioredis applies its client prefix. +A batch never splits into separate operations. Cross-slot calls reject even +when both slots belong to the same master. + +For a new Cluster namespace, choose an explicit hash tag in the existing +`keyPrefix` option: + +~~~~ typescript +import type { AtomicKvStore } from "@fedify/fedify"; +import { RedisKvStore } from "@fedify/redis"; +import { Cluster } from "ioredis"; + +const redis = new Cluster([{ host: "127.0.0.1", port: 7000 }]); +const store: AtomicKvStore = new RedisKvStore(redis, { + keyPrefix: "{inventory}::", +}); +const keys = [["available"], ["reserved"]] as const; +const [available, reserved] = await store.getMany(keys); +if (available !== undefined && available > 0) { + const committed = await store.casMany([ + { key: keys[0], expectedValue: available, newValue: available - 1 }, + { key: keys[1], expectedValue: reserved, newValue: (reserved ?? 0) + 1 }, + ]); + // On false, an earlier attempt may have committed. Check a receipt or + // version before re-evaluating the reservation rule (see below). +} +~~~~ + +The adapter preserves its key encoding and never adds a tag or relocates data. +Changing a prefix selects different keys; existing data needs an explicit +application migration. Hash tags affect routing, not key identity. Choose +prefixes whose encoded keys cannot overlap, such as `"{application}:one::"` and +`"{application}:two::"`. Redis uses the first `{` and the first following `}`: +an empty pair hashes the whole key rather than searching for a later tag. A tag +in the ioredis client's own `keyPrefix` can therefore take precedence over a +tag in the adapter's `keyPrefix`. + +Each batch request is limited to 128 entries and 1 MiB of data, counting +effective physical key bytes, encoded expectations and replacements, and +argument strings. The fixed script and protocol framing are excluded. Oversized +calls reject without chunking. Keys in a mutating batch must be distinct; two +keys that encode to the same bytes count as duplicates. Repeated keys in +`getMany()` repeat the same snapshot value. Empty arrays are local no-ops; +`casMany([])` returns `true`. The bounds and the fixed scripts are part of the +atomicity guarantee, not throughput settings. + +All values must round-trip through the selected codec. `JsonCodec` supports +JSON-representable values; it rejects top-level values that JSON cannot encode. +`undefined` is never stored, while `null` is a value. Single and batch CAS +compare encoded bytes, so JSON object property order matters. Custom codecs +must encode equal values deterministically; binary values and empty encoded +buffers are supported. A CAS `newValue` of `undefined` deletes the key, and a +non-null TTL on deletion rejects even when the comparison would mismatch. + +TTLs round up to whole seconds with a minimum of one second, so zero and +negative durations keep a value for one second. A batch TTL must not exceed +2,147,483,647 seconds; this additional cap does not apply to single-key calls. +Replacing a value without a TTL clears its previous expiry. Redis checks expiry +against one server time throughout each batch. A mismatch leaves values and +their TTLs unchanged. + +Scripts require `EVAL`, read access through `GET` or `EXISTS`, and the relevant +`SET`, `SETEX`, or `DEL` permissions for every key. Redis checks `EVAL`'s +read/write key access before entering the script, even for snapshots: key +patterns must grant both read and write access to every batch key. Inputs and +all command permissions are checked before application writes; Redis script +isolation is not rollback for a later command error. During Cluster resharding, +missing or expired keys can cause the whole call to reject with a routing +error. Under configured memory pressure, batch writes and deletes can reject +before any mutation; ordinary `delete()` remains available to free space. + +Snapshots use primary routing even when the Cluster client's `scaleReads` +option selects replicas for ordinary reads. Prefix listing retains the +existing Cluster `SCAN` limitation: it can visit only part of the Cluster, and +its cursors have no slot affinity. `getMany()` snapshots only the requested +keys; it does not establish a snapshot of `list()`. + +Separate calls do not form a transaction. The adapter adds no retry loop and +does not alter the supplied client's options. Batch scripts dispatch directly +and do not join ioredis's automatic pipelines. ioredis redirects and connection +replays can retry a request: a later CAS `false` reply does not establish that +an earlier transport attempt never committed. Re-read and re-evaluate after a +mismatch or uncertain response; use non-reused versions or receipts when the +application needs them. Atomicity does not imply crash or failover durability. +Configure Redis persistence, replication, and eviction for authoritative state; +a successful call alone does not make an external effect exactly once. + +### Adapter verification + +Set `REDIS_URL` to a standalone server URL and, for a disposable three-master +Cluster, `REDIS_CLUSTER_NODES` to comma-separated `host:port` addresses, then +run the package suites: + +~~~~ sh +mise run test-each redis +~~~~ + +Conformance runs only on the supported releases. Node.js marks other releases +as skipped; Bun and Deno log the reason and return without running conformance. +A separate case still checks that every batch method rejects without changing +values or TTLs. Administrative tests require a dedicated Redis server at +`REDIS_ATOMIC_ADMIN_URL`, distinct from the server at `REDIS_URL`: they change +server-wide configuration. Cluster +administrative and migration tests additionally opt in through +`REDIS_ATOMIC_CLUSTER_ADMIN_NODES` and `REDIS_ATOMIC_MIGRATION_NODES`. +Run *src/kv.atomic-admin.test.ts* and the migration case in +*src/kv.atomic-cluster.test.ts* separately against an otherwise idle disposable +server or Cluster: they change ACLs, memory settings, or slot ownership, and +restore their state in cleanup. These tests do not establish replica-lag or +failover durability. + +The `test-redis-atomic` CI matrix pins Redis 7.2.17, 7.4.12, and 8.10.3 and +uses dedicated standalone, administrative, and three-primary Cluster fixtures. +Each job runs Deno, Node.js, and Bun in sequence. It sets +`FEDIFY_REDIS_REQUIRE_ATOMIC_TESTS=true` so missing fixtures and unsupported +Redis versions fail instead of skipping the conformance tests. The isolated +configuration probes clear this flag while checking absent and blank fixture +settings. Under Node.js, the package's test files run sequentially so the +administrative and migration tests cannot interfere with each other. diff --git a/packages/redis/deno.json b/packages/redis/deno.json index 82b23dce7..7bffa4621 100644 --- a/packages/redis/deno.json +++ b/packages/redis/deno.json @@ -25,6 +25,6 @@ }, "tasks": { "check": "deno check src/*.ts", - "test": "deno test --allow-net --allow-env --doc --no-check=leaks" + "test": "deno test --allow-net --allow-env --allow-read --allow-write --allow-run --allow-sys --doc --no-check=leaks" } } diff --git a/packages/redis/package.json b/packages/redis/package.json index a1d7d8a3c..e6429c9a2 100644 --- a/packages/redis/package.json +++ b/packages/redis/package.json @@ -108,7 +108,7 @@ "prepack": "pnpm build", "prepublish": "pnpm build", "pretest": "pnpm build", - "test": "node --experimental-transform-types --test", + "test": "node --experimental-transform-types --test --test-concurrency=1", "pretest:bun": "pnpm build", "test:bun": "bun test --timeout=60000" } diff --git a/packages/redis/src/atomic.ts b/packages/redis/src/atomic.ts new file mode 100644 index 000000000..2246ddfbf --- /dev/null +++ b/packages/redis/src/atomic.ts @@ -0,0 +1,83 @@ +// Private implementation constants, also used by the adapter tests. +export const atomicRedisVersions = ["7.2.17", "7.4.12", "8.10.3"] as const; +export const atomicMaxEntries = 128; +export const atomicMaxBytes = 1024 * 1024; +export const atomicMaxTtlSeconds = 2 ** 31 - 1; + +const versionNumbers = atomicRedisVersions.map((version) => + version.split(".").reduce((number, part) => number * 256 + Number(part), 0), +); +const versionCheck = `local version = redis.REDIS_VERSION_NUM +if ${versionNumbers.map((version) => `version ~= ${version}`).join(" and ")} then + return redis.error_reply('ERR atomic KV batches require official Redis ${atomicRedisVersions.slice(0, -1).join(", ")}, or ${atomicRedisVersions.at(-1)}') +end +`; + +// The complete scripts must stay below Redis's per-invocation 100,000 Lua +// instruction hook. Otherwise slow-script event handling can change Cluster +// topology between writes. Audited Lua 5.1 listings have 190/37 instructions, +// no Lua functions, and only separate loops over at most 128 keys: each static +// instruction runs at most 129 times (24,510/4,773 total). Native command paths +// are synchronous on the exact releases above. Re-audit these bounds and the +// engine internals when changing a script, limit, or supported release. +// See Redis src/script_lua.c luaCallFunction(), deps/lua/src/ldebug.c +// lua_sethook(), src/script.c scriptVerifyClusterState(), and src/cluster.c +// getNodeByQuery(). Debugger execution, module hooks, and modified engines are +// outside this guarantee. All keys must be declared in KEYS; never split calls. + +export const atomicMutationScript = `#!lua +${versionCheck}-- Fixed shape ARGV: CAS flag, then five arguments/key: +-- absence flag, expected bytes, deletion flag, replacement bytes, TTL seconds. +if #KEYS > ${atomicMaxEntries} then return redis.error_reply('ERR batch too large') end +if #ARGV ~= 1 + 5 * #KEYS then + return redis.error_reply('ERR invalid atomic KV arguments') +end +local seen = {} +local current = {} +local matches = true +for i, key in ipairs(KEYS) do + if seen[key] then return redis.error_reply('ERR duplicate atomic KV key') end + seen[key] = true + local base = 2 + (i - 1) * 5 + if ARGV[1] == '1' then + current[i] = redis.call('GET', key) + if ARGV[base] == '1' then + if current[i] then matches = false end + elseif not current[i] or current[i] ~= ARGV[base+1] then + matches = false + end + else + redis.call('EXISTS', key) + end + local permitted + if ARGV[base+2] == '1' then + permitted = redis.acl_check_cmd('DEL', key) + elseif ARGV[base+4] == '0' then + permitted = redis.acl_check_cmd('SET', key, ARGV[base+3]) + else + permitted = redis.acl_check_cmd('SETEX', key, ARGV[base+4], ARGV[base+3]) + end + if not permitted then return redis.error_reply('NOPERM atomic KV mutation denied') end +end +if not matches then return 0 end +for i, key in ipairs(KEYS) do + local base = 2 + (i - 1) * 5 + if ARGV[base+2] == '1' then + redis.call('DEL', key) + elseif ARGV[base+4] == '0' then + redis.call('SET', key, ARGV[base+3]) + else + redis.call('SETEX', key, ARGV[base+4], ARGV[base+3]) + end +end +return 1 +`; + +export const atomicSnapshotScript = `#!lua flags=no-writes +${versionCheck}if #KEYS > ${atomicMaxEntries} then return redis.error_reply('ERR batch too large') end +local values = {} +for i, key in ipairs(KEYS) do + values[i] = redis.call('GET', key) +end +return values +`; diff --git a/packages/redis/src/codec.ts b/packages/redis/src/codec.ts index 1674089d6..409a8b0cd 100644 --- a/packages/redis/src/codec.ts +++ b/packages/redis/src/codec.ts @@ -66,6 +66,9 @@ export class JsonCodec implements Codec { if (e instanceof TypeError) throw new EncodingError(e.message); throw e; } + if (typeof json !== "string") { + throw new EncodingError("The value cannot be represented as JSON."); + } return Buffer.from(this.#textEncoder.encode(json)); } diff --git a/packages/redis/src/kv.atomic-admin.test.ts b/packages/redis/src/kv.atomic-admin.test.ts new file mode 100644 index 000000000..e1050a86a --- /dev/null +++ b/packages/redis/src/kv.atomic-admin.test.ts @@ -0,0 +1,225 @@ +// These opt-in tests change ACL/configuration on dedicated disposable servers. +import { RedisKvStore } from "@fedify/redis/kv"; +import * as temporal from "@js-temporal/polyfill"; +import { Redis } from "ioredis"; +import assert from "node:assert/strict"; +import process from "node:process"; +import { test, type TestContext } from "node:test"; +import { atomicRedisVersions } from "./atomic.ts"; + +const Temporal = globalThis.Temporal ?? temporal.Temporal; +const adminUrl = process.env.REDIS_ATOMIC_ADMIN_URL?.trim() || undefined; +const clusterAdminNodes = + process.env.REDIS_ATOMIC_CLUSTER_ADMIN_NODES?.trim() || undefined; +const versions = new Set(atomicRedisVersions); + +function skipUnsupported(t: TestContext, reason: string): void { + // Bun lacks dynamic skip; Deno may report a dynamic skip as a pass. + assert.notEqual( + process.env.FEDIFY_REDIS_REQUIRE_ATOMIC_TESTS, + "true", + `Atomic CI requires an audited Redis engine: ${reason}`, + ); + console.warn(`Skipping Redis atomic test: ${reason}`); + if (process.versions.bun == null) t.skip(reason); +} + +async function failures(admin: Redis, prefix: string): Promise { + const store = new RedisKvStore(admin, { keyPrefix: prefix }); + const user = `fedify_atomic_${crypto.randomUUID()}`; + const password = crypto.randomUUID(); + const ttl = Temporal.Duration.from({ seconds: 30 }); + const short = Temporal.Duration.from({ seconds: 1 }); + await store.setMany([ + { key: ["a"], value: 1, options: { ttl } }, + { key: ["b"], value: 2, options: { ttl } }, + ]); + const started = Date.now(); + const before = await Promise.all([ + admin.pttl(`${prefix}a`), + admin.pttl(`${prefix}b`), + ]); + async function unchanged(): Promise { + assert.deepEqual([await store.get(["a"]), await store.get(["b"])], [1, 2]); + const after = await Promise.all([ + admin.pttl(`${prefix}a`), + admin.pttl(`${prefix}b`), + ]); + for (let i = 0; i < after.length; i++) { + assert( + after[i] <= before[i] && + after[i] >= before[i] - (Date.now() - started) - 100, + "rejection must not change expiry", + ); + } + } + await admin.acl( + "SETUSER", + user, + "reset", + "on", + `>${password}`, + `~${prefix}*`, + "+eval", + "+get", + "+exists", + "+set", + "+del", + "+ping", + "-setex", + ); + const limited = new Redis({ + ...admin.options, + username: user, + password, + enableReadyCheck: false, + }); + const restricted = new RedisKvStore(limited, { keyPrefix: prefix }); + try { + await limited.ping(); + await assert.rejects( + restricted.setMany([ + { key: ["a"], value: 10 }, + { key: ["b"], value: 20, options: { ttl: short } }, + ]), + ); + await unchanged(); + await assert.rejects( + restricted.casMany([ + { key: ["a"], expectedValue: "wrong", newValue: 10 }, + { key: ["b"], expectedValue: 2, newValue: 20, options: { ttl: short } }, + ]), + ); + await unchanged(); + await admin.acl("SETUSER", user, "+setex", "-del"); + await assert.rejects( + restricted.casMany([ + { key: ["a"], expectedValue: 1, newValue: 10 }, + { key: ["b"], expectedValue: 2, newValue: undefined }, + ]), + ); + await unchanged(); + await admin.acl("SETUSER", user, "+del", "resetkeys", `~${prefix}a`); + await assert.rejects( + restricted.setMany([ + { key: ["a"], value: 10 }, + { key: ["b"], value: 20 }, + ]), + ); + await unchanged(); + // EVAL checks both read/write key access before entering the script. + await admin.acl("SETUSER", user, "resetkeys", `%W~${prefix}*`); + await assert.rejects( + restricted.setMany([ + { key: ["a"], value: 10 }, + { key: ["b"], value: 20 }, + ]), + ); + await unchanged(); + await admin.acl("SETUSER", user, "resetkeys", `%R~${prefix}*`); + await assert.rejects(restricted.getMany([["a"], ["b"]])); + await unchanged(); + } finally { + limited.disconnect(); + await admin.acl("DELUSER", user); + } + + const originalMemory = await admin.config("GET", "maxmemory"); + const originalPolicy = await admin.config("GET", "maxmemory-policy"); + assert( + Array.isArray(originalMemory) && typeof originalMemory[1] === "string", + ); + assert( + Array.isArray(originalPolicy) && typeof originalPolicy[1] === "string", + ); + try { + await admin.config("SET", "maxmemory-policy", "noeviction"); + await admin.config("SET", "maxmemory", "1"); + await assert.rejects( + store.casMany([ + { key: ["a"], expectedValue: 1, newValue: undefined }, + { key: ["b"], expectedValue: 2, newValue: 20 }, + ]), + /OOM/, + ); + await assert.rejects( + store.setMany([ + { key: ["a"], value: 10 }, + { key: ["b"], value: 20 }, + ]), + /OOM/, + ); + await assert.rejects(store.deleteMany([["a"], ["b"]]), /OOM/); + assert.deepEqual(await store.getMany([["a"], ["b"]]), [1, 2]); + await unchanged(); + } finally { + await admin.config("SET", "maxmemory", originalMemory[1]); + await admin.config("SET", "maxmemory-policy", originalPolicy[1]); + await admin.del(`${prefix}a`, `${prefix}b`); + } +} + +test( + "Redis atomic standalone administrative failures", + { skip: adminUrl == null }, + async (t) => { + if (adminUrl == null) return; // Bun ignores node:test skip. + assert.notEqual( + adminUrl, + process.env.REDIS_URL, + "use a dedicated administrative fixture, separate from ordinary package tests", + ); + const admin = new Redis(adminUrl!); + try { + const actual = String(await admin.eval("return redis.REDIS_VERSION", 0)); + if (!versions.has(actual)) { + skipUnsupported(t, `Redis ${actual} is not an audited batch engine`); + return; + } + await failures(admin, `{fedify_admin_${crypto.randomUUID()}}::`); + } finally { + await admin.quit(); + } + }, +); + +test( + "Redis atomic Cluster administrative failures", + { skip: clusterAdminNodes == null }, + async (t) => { + if (clusterAdminNodes == null) return; // Bun ignores node:test skip. + // Opt in only when no other tests/clients are using this disposable Cluster. + const nodes = clusterAdminNodes!.split(",").map((node) => { + const [host, port] = node.split(":"); + return new Redis(Number(port), host); + }); + try { + const actual = String( + await nodes[0].eval("return redis.REDIS_VERSION", 0), + ); + if (!versions.has(actual)) { + skipUnsupported(t, `Redis ${actual} is not an audited batch engine`); + return; + } + const prefix = `{fedify_cluster_admin_${crypto.randomUUID()}}::`; + const slot = Number(await nodes[0].cluster("KEYSLOT", `${prefix}a`)); + const slots = await nodes[0].cluster("SLOTS"); + assert(Array.isArray(slots)); + const range = slots.find( + (range: unknown) => + Array.isArray(range) && slot >= range[0] && slot <= range[1], + ); + assert(Array.isArray(range) && Array.isArray(range[2])); + const nodeIds = await Promise.all( + nodes.map(async (node) => String(await node.cluster("MYID"))), + ); + const owner = nodes.find( + (_, index) => nodeIds[index] === String(range[2][2]), + ); + assert(owner); + await failures(owner, prefix); + } finally { + await Promise.all(nodes.map((node) => node.quit())); + } + }, +); diff --git a/packages/redis/src/kv.atomic-cluster.test.ts b/packages/redis/src/kv.atomic-cluster.test.ts new file mode 100644 index 000000000..a56d53ebc --- /dev/null +++ b/packages/redis/src/kv.atomic-cluster.test.ts @@ -0,0 +1,517 @@ +// Slot migration is opt-in and exclusively owns a disposable Cluster fixture. +/* oxlint-disable eslint/no-await-in-loop */ +import type { KvKey } from "@fedify/fedify"; +import { RedisKvStore } from "@fedify/redis/kv"; +import { Cluster, Redis, type Command, type Pipeline } from "ioredis"; +import assert from "node:assert/strict"; +import process from "node:process"; +import { test, type TestContext } from "node:test"; +import { atomicRedisVersions, atomicSnapshotScript } from "./atomic.ts"; + +const addresses = process.env.REDIS_CLUSTER_NODES?.trim() || undefined; +const migrationAddresses = + process.env.REDIS_ATOMIC_MIGRATION_NODES?.trim() || undefined; +const versions = new Set(atomicRedisVersions); + +function skipUnsupported(t: TestContext, reason: string): void { + // Bun lacks dynamic skip; Deno may report a dynamic skip as a pass. + assert.notEqual( + process.env.FEDIFY_REDIS_REQUIRE_ATOMIC_TESTS, + "true", + `Atomic CI requires an audited Redis engine: ${reason}`, + ); + console.warn(`Skipping Redis atomic test: ${reason}`); + if (process.versions.bun == null) t.skip(reason); +} + +function startup(addresses: string): { host: string; port: number }[] { + return addresses.split(",").map((address) => { + const [host, port] = address.split(":"); + return { host, port: Number(port) }; + }); +} + +async function supported(redis: Redis | Cluster): Promise { + return versions.has( + String(await redis.eval("return redis.REDIS_VERSION", 0)), + ); +} + +test( + "Redis atomic Cluster primary routing and namespace isolation", + { skip: addresses == null }, + async (t) => { + if (addresses == null) return; // Bun ignores node:test skip. + let readRouting = 0; + const cluster = new Cluster(startup(addresses!), { + scaleReads(nodes: Redis[]) { + readRouting++; + return nodes[0]; + }, + }); + const tag = `{fedify_routing_${crypto.randomUUID()}}`; + const firstPrefix = `${tag}:one::`, + secondPrefix = `${tag}:two::`; + const first = new RedisKvStore(cluster, { keyPrefix: firstPrefix }); + const second = new RedisKvStore(cluster, { keyPrefix: secondPrefix }); + try { + if (!(await supported(cluster))) { + skipUnsupported(t, "Redis is not an audited batch engine"); + return; + } + await first.setMany([ + { key: ["a"], value: 1 }, + { key: ["b"], value: 2 }, + ]); + await second.setMany([ + { key: ["a"], value: 10 }, + { key: ["b"], value: 20 }, + ]); + readRouting = 0; + assert.deepEqual(await first.getMany([["a"], ["b"]]), [1, 2]); + assert.deepEqual(await second.getMany([["a"], ["b"]]), [10, 20]); + assert.equal( + readRouting, + 0, + "ordinary EVAL must bypass replica/custom read routing", + ); + await first.get(["a"]); + assert( + readRouting > 0, + "the custom read route must be active for ordinary GET", + ); + } finally { + await cluster.del( + `${firstPrefix}a`, + `${firstPrefix}b`, + `${secondPrefix}a`, + `${secondPrefix}b`, + ); + await cluster.quit(); + } + }, +); + +for (const autoPipeline of [false, true]) { + test( + `Redis atomic Cluster rejects different slots on one master (auto pipeline ${autoPipeline})`, + { skip: addresses == null }, + async (t) => { + if (addresses == null) return; // Bun ignores node:test skip. + const cluster = new Cluster(startup(addresses!), { + enableAutoPipelining: autoPipeline, + }); + const prefix = `fedify_cross_${crypto.randomUUID()}::`; + const store = new RedisKvStore(cluster, { keyPrefix: prefix }); + let keys: KvKey[] = []; + try { + if (!(await supported(cluster))) { + skipUnsupported(t, "Redis is not an audited batch engine"); + return; + } + const slots = await cluster.cluster("SLOTS"); + assert(Array.isArray(slots) && Array.isArray(slots[0])); + const [start, end] = slots[0]; + const found: { key: KvKey; slot: number }[] = []; + for (let i = 0; i < 100 && found.length < 2; i++) { + const key: KvKey = [String(i)]; + const slot = Number( + await cluster.cluster("KEYSLOT", `${prefix}${i}`), + ); + if ( + slot >= start && + slot <= end && + !found.some((entry) => entry.slot === slot) + ) + found.push({ key, slot }); + } + assert.equal(found.length, 2); + keys = found.map((entry) => entry.key); + await store.set(keys[0], 1); + await store.set(keys[1], 2); + await assert.rejects(store.getMany(keys)); + await assert.rejects( + store.setMany([ + { key: keys[0], value: 10 }, + { key: keys[1], value: 20 }, + ]), + ); + await assert.rejects(store.deleteMany(keys)); + await assert.rejects( + store.casMany([ + { key: keys[0], expectedValue: 1, newValue: 10 }, + { key: keys[1], expectedValue: 2, newValue: 20 }, + ]), + ); + assert.deepEqual( + [await store.get(keys[0]), await store.get(keys[1])], + [1, 2], + ); + } finally { + // DEL itself cannot combine the deliberately different slots. + for (const key of keys) await store.delete(key); + await cluster.quit(); + } + }, + ); +} + +test( + "Redis atomic Cluster an empty first brace pair does not select a later tag", + { skip: addresses == null }, + async (t) => { + if (addresses == null) return; // Bun ignores node:test skip. + const cluster = new Cluster(startup(addresses!)); + const prefix = `{}{fedify_empty_${crypto.randomUUID()}}::`; + const store = new RedisKvStore(cluster, { keyPrefix: prefix }); + try { + if (!(await supported(cluster))) { + skipUnsupported(t, "Redis is not an audited batch engine"); + return; + } + assert.notEqual( + await cluster.cluster("KEYSLOT", `${prefix}a`), + await cluster.cluster("KEYSLOT", `${prefix}b`), + ); + await store.set(["a"], 1); + await store.set(["b"], 2); + await assert.rejects( + store.setMany([ + { key: ["a"], value: 10 }, + { key: ["b"], value: 20 }, + ]), + ); + assert.deepEqual( + [await store.get(["a"]), await store.get(["b"])], + [1, 2], + ); + } finally { + await store.delete(["a"]); + await store.delete(["b"]); + await cluster.quit(); + } + }, +); + +test( + "Redis atomic Cluster concurrent batches across masters and ordinary commands", + { skip: addresses == null }, + async (t) => { + if (addresses == null) return; // Bun ignores node:test skip. + const cluster = new Cluster(startup(addresses), { + enableAutoPipelining: true, + }); + const prefixes: string[] = []; + let ordinaryKey = ""; + try { + if (!(await supported(cluster))) { + skipUnsupported(t, "Redis is not an audited batch engine"); + return; + } + const slots = await cluster.cluster("SLOTS"); + assert(Array.isArray(slots)); + const owner = (slot: number): string => { + const range = slots.find( + (range: unknown) => + Array.isArray(range) && slot >= range[0] && slot <= range[1], + ); + assert(Array.isArray(range) && Array.isArray(range[2])); + return String(range[2][2]); + }; + const scriptOwner = owner( + Number(await cluster.cluster("KEYSLOT", atomicSnapshotScript)), + ); + const found = new Map(); + for ( + let i = 0; + i < 100 && (found.size < 2 || !found.has(scriptOwner)); + i++ + ) { + const prefix = `{fedify_multimaster_${crypto.randomUUID()}}::`; + const node = owner( + Number(await cluster.cluster("KEYSLOT", `${prefix}a`)), + ); + found.set(node, prefix); + } + assert( + found.size >= 2 && found.has(scriptOwner), + "a multi-master Cluster fixture is required", + ); + const firstPrefix = found.get(scriptOwner)!; + const secondPrefix = [...found].find( + ([node]) => node !== scriptOwner, + )![1]; + prefixes.push(firstPrefix, secondPrefix); + const first = new RedisKvStore(cluster, { keyPrefix: firstPrefix }); + const second = new RedisKvStore(cluster, { keyPrefix: secondPrefix }); + await Promise.all([ + first.setMany([{ key: ["a"], value: 1 }]), + second.setMany([{ key: ["a"], value: 10 }]), + ]); + assert.deepEqual( + await Promise.all([first.getMany([["a"]]), second.getMany([["a"]])]), + [[1], [10]], + ); + ordinaryKey = `${firstPrefix}ordinary`; + await cluster.set(ordinaryKey, "ordinary"); + const [snapshot, ordinary] = await Promise.all([ + second.getMany([["a"]]), + cluster.get(ordinaryKey), + ]); + assert.deepEqual(snapshot, [10]); + assert.equal(ordinary, "ordinary"); + assert.deepEqual( + await Promise.all([ + first.casMany([{ key: ["a"], expectedValue: 1, newValue: 2 }]), + second.casMany([{ key: ["a"], expectedValue: 10, newValue: 20 }]), + ]), + [true, true], + ); + assert.deepEqual( + await Promise.all([first.getMany([["a"]]), second.getMany([["a"]])]), + [[2], [20]], + ); + await Promise.all([ + first.deleteMany([["a"]]), + second.deleteMany([["a"]]), + ]); + assert.deepEqual( + await Promise.all([first.getMany([["a"]]), second.getMany([["a"]])]), + [[undefined], [undefined]], + ); + } finally { + for (const prefix of prefixes) await cluster.del(`${prefix}a`); + if (ordinaryKey !== "") await cluster.del(ordinaryKey); + await cluster.quit(); + } + }, +); + +test( + "Redis atomic Cluster resharding and routing-to-execution expiry regression", + { skip: migrationAddresses == null }, + async (t) => { + if (migrationAddresses == null) return; // Bun ignores node:test skip. + const nodes = startup(migrationAddresses!).map( + ({ host, port }) => new Redis(port, host), + ); + const routed = new Cluster(startup(migrationAddresses!)); + const bounded = new Cluster(startup(migrationAddresses!), { + maxRedirections: 1, + retryDelayOnTryAgain: 0, + }); + let source: Redis | undefined, target: Redis | undefined; + let sourceId = "", + targetId = "", + slot = -1, + prefix = ""; + let committed = false; + let slowClient: Redis | undefined; + // Redis TIME reads a cached clock inside a script. Use finite work instead; + // the expiry assertions below require this delay to outlast the 50 ms TTL. + const slow = `local total = 0 +for i = 1, 50000000 do total = total + i end +return total`; + try { + if (!(await supported(nodes[0]))) { + skipUnsupported(t, "Redis is not an audited batch engine"); + return; + } + assert.equal(nodes.length, 3, "use a dedicated three-master fixture"); + const slots = await nodes[0].cluster("SLOTS"); + assert(Array.isArray(slots)); + const nodeIds = await Promise.all( + nodes.map(async (node) => String(await node.cluster("MYID"))), + ); + for (let attempt = 0; attempt < 100; attempt++) { + prefix = `{fedify_migration_${crypto.randomUUID()}}::`; + slot = Number(await nodes[0].cluster("KEYSLOT", `${prefix}a`)); + const range = slots.find( + (range: unknown) => + Array.isArray(range) && slot >= range[0] && slot <= range[1], + ); + assert(Array.isArray(range) && Array.isArray(range[2])); + source = nodes.find( + (_, index) => nodeIds[index] === String(range[2][2]), + ); + assert(source); + if (Number(await source.cluster("COUNTKEYSINSLOT", slot)) === 0) break; + slot = -1; + } + assert(slot >= 0 && source, "an empty test slot is required"); + target = nodes.find((node) => node !== source); + assert(target); + sourceId = String(await source.cluster("MYID")); + targetId = String(await target.cluster("MYID")); + slowClient = new Redis({ + host: source.options.host, + port: source.options.port, + enableAutoPipelining: true, + }); + const store = new RedisKvStore(routed, { keyPrefix: prefix }); + const limited = new RedisKvStore(bounded, { keyPrefix: prefix }); + const keys = [`${prefix}a`, `${prefix}b`]; + + // Warm routing before changing ownership, then force the older routing + // snapshot/newer script snapshot window on one auto-pipelined connection. + await store.setMany([ + { key: ["a"], value: 1 }, + { key: ["b"], value: 2 }, + ]); + await target.cluster("SETSLOT", slot, "IMPORTING", sourceId); + await source.cluster("SETSLOT", slot, "MIGRATING", targetId); + for (const operation of ["set", "delete", "cas"] as const) { + await source.cluster("SETSLOT", slot, "STABLE"); + await source.set(keys[0], "1", "PX", 50); + await source.set(keys[1], "2"); + await source.cluster("SETSLOT", slot, "MIGRATING", targetId); + // Send both requests in one explicit pipeline so the routing clock + // predates the slow script but the batch execution clock follows it. + // Capture the actual adapter Command, including its declared keys. + const pipeline = slowClient.pipeline() as Pipeline; + pipeline.eval(slow, 0); + const queued = { + options: slowClient.options, + sendCommand(command: Command): Promise { + pipeline.sendCommand(command); + return command.promise; + }, + } as unknown as Redis; + const direct = new RedisKvStore(queued, { keyPrefix: prefix }); + const change = + operation === "set" + ? direct.setMany([ + { key: ["b"], value: 100 }, + { key: ["a"], value: 200 }, + ]) + : operation === "delete" + ? direct.deleteMany([["b"], ["a"]]) + : direct.casMany([ + { key: ["b"], expectedValue: 2, newValue: undefined }, + { key: ["a"], expectedValue: 1, newValue: 200 }, + ]); + const results: PromiseSettledResult[] = + await Promise.allSettled([pipeline.exec(), change]); + assert.equal(results[0].status, "fulfilled"); + assert.equal( + results[1].status, + "rejected", + "a later expired migrating key must reject before all writes", + ); + assert.equal( + await source.get(keys[1]), + "2", + "the earlier persistent key must be untouched", + ); + } + await source.cluster("SETSLOT", slot, "STABLE"); + await store.setMany([ + { key: ["a"], value: 1 }, + { key: ["b"], value: 2 }, + ]); + await source.cluster("SETSLOT", slot, "MIGRATING", targetId); + assert.equal( + await store.casMany([ + { key: ["a"], expectedValue: 1, newValue: 3 }, + { key: ["b"], expectedValue: 2, newValue: 4 }, + ]), + true, + ); + await source.migrate( + target.options.host ?? "127.0.0.1", + target.options.port ?? 6379, + keys[0], + 0, + 5000, + ); + await assert.rejects(limited.getMany([["a"], ["b"]])); + await assert.rejects( + limited.setMany([ + { key: ["a"], value: 10 }, + { key: ["b"], value: 20 }, + ]), + ); + await assert.rejects(limited.deleteMany([["a"], ["b"]])); + await assert.rejects( + limited.casMany([ + { key: ["a"], expectedValue: 3, newValue: 10 }, + { key: ["b"], expectedValue: 4, newValue: 20 }, + ]), + ); + await target.call("ASKING"); + assert.equal(await target.get(keys[0]), "3"); + assert.equal(await source.get(keys[1]), "4"); + await assert.rejects( + limited.casMany([ + { key: ["missing-a"], expectedValue: undefined, newValue: 1 }, + { key: ["missing-b"], expectedValue: undefined, newValue: 2 }, + ]), + ); + await source.migrate( + target.options.host ?? "127.0.0.1", + target.options.port ?? 6379, + keys[1], + 0, + 5000, + ); + await assert.rejects( + source.callBuffer("eval", atomicSnapshotScript, 2, ...keys), + /ASK/, + ); + assert.deepEqual( + await store.getMany([["a"], ["b"]]), + [3, 4], + "the Cluster client must follow ASK/ASKING", + ); + assert.equal( + await store.casMany([ + { key: ["a"], expectedValue: 3, newValue: 5 }, + { key: ["b"], expectedValue: 4, newValue: 6 }, + ]), + true, + ); + await target.cluster("SETSLOT", slot, "NODE", targetId); + await source.cluster("SETSLOT", slot, "NODE", targetId); + for (const node of nodes) + if (node !== source && node !== target) + await node.cluster("SETSLOT", slot, "NODE", targetId); + committed = true; + await assert.rejects(source.get(keys[0]), /MOVED/); + assert.deepEqual( + await store.getMany([["a"], ["b"]]), + [5, 6], + "routing must recover after ownership moves", + ); + await store.deleteMany([["a"], ["b"]]); + assert.equal(await target.exists(...keys), 0); + } finally { + try { + if (source && target && slot >= 0 && sourceId) { + const owner = committed ? target : source; + const other = committed ? source : target; + const ownerId = committed ? targetId : sourceId; + await owner.cluster("SETSLOT", slot, "STABLE"); + await other.cluster("SETSLOT", slot, "IMPORTING", ownerId); + for (const suffix of ["a", "b", "missing-a", "missing-b"]) { + await other.call("ASKING"); + await other.del(`${prefix}${suffix}`); + } + await owner.del( + `${prefix}a`, + `${prefix}b`, + `${prefix}missing-a`, + `${prefix}missing-b`, + ); + await other.cluster("SETSLOT", slot, "STABLE"); + for (const node of nodes) + await node.cluster("SETSLOT", slot, "NODE", sourceId); + } + } finally { + slowClient?.disconnect(); + routed.disconnect(); + bounded.disconnect(); + await Promise.all(nodes.map((node) => node.quit())); + } + } + }, +); diff --git a/packages/redis/src/kv.atomic-gates.test.ts b/packages/redis/src/kv.atomic-gates.test.ts new file mode 100644 index 000000000..b7e80d703 --- /dev/null +++ b/packages/redis/src/kv.atomic-gates.test.ts @@ -0,0 +1,128 @@ +// Run isolated profiles sequentially to keep child runtimes bounded. +import assert from "node:assert/strict"; +import { Buffer } from "node:buffer"; +import { spawnSync } from "node:child_process"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import process from "node:process"; +import { test } from "node:test"; +import { fileURLToPath, pathToFileURL } from "node:url"; + +const variables = [ + "REDIS_URL", + "REDIS_CLUSTER_NODES", + "REDIS_ATOMIC_ADMIN_URL", + "REDIS_ATOMIC_CLUSTER_ADMIN_NODES", + "REDIS_ATOMIC_MIGRATION_NODES", +]; + +test("Redis atomic fixture opt-ins reject absent and blank configuration before client startup", () => { + const packageRoot = fileURLToPath(new URL("../", import.meta.url)); + const directory = mkdtempSync(join(tmpdir(), "fedify-redis-fixture-gates-")); + const probe = join(directory, "guard.test.js"); + const files = [ + "kv.atomic.test.ts", + "kv.atomic-admin.test.ts", + "kv.atomic-cluster.test.ts", + ].map((file) => pathToFileURL(join(packageRoot, "src", file)).href); + // The child intercepts eager constructor startup and socket connections. + // A regressed opt-in fails before it can reach a real default server. + const source = ` +import assert from "node:assert/strict"; +import { Buffer } from "node:buffer"; +import { createRequire } from "node:module"; +import net from "node:net"; +import process from "node:process"; +import { test } from "node:test"; +// These probes intentionally test missing fixtures, not CI conformance. +delete process.env.FEDIFY_REDIS_REQUIRE_ATOMIC_TESTS; +assert.equal(process.env.FEDIFY_REDIS_REQUIRE_ATOMIC_TESTS, undefined); +const rawProfile = process.env.FEDIFY_REDIS_FIXTURE_PROFILE; +assert.notEqual(rawProfile, undefined, "fixture profile must reach the child"); +const profile = JSON.parse(rawProfile); +// Deno's spawnSync inherits omitted env keys: clear them inside the child too. +for (const key of Object.keys(process.env)) { + if (key.startsWith("REDIS_")) delete process.env[key]; +} +for (const key of ${JSON.stringify(variables)}) { + if (profile.present) process.env[key] = profile.value; + assert.equal(process.env[key], profile.present ? profile.value : undefined); +} +console.log("FIXTURE_PROFILE_HEX " + Buffer.from(JSON.stringify(profile)).toString("hex")); +const require = createRequire(${JSON.stringify(join(packageRoot, "package.json"))}); +const { Redis, Cluster } = require("ioredis"); +const calls = { redis: 0, cluster: 0, sockets: 0 }; +Redis.prototype.connect = function () { + calls.redis++; + throw new Error("GUARD: unexpected Redis construction/startup"); +}; +Cluster.prototype.connect = function () { + calls.cluster++; + throw new Error("GUARD: unexpected Cluster construction/startup"); +}; +net.Socket.prototype.connect = function () { + calls.sockets++; + throw new Error("GUARD: unexpected socket connection"); +}; +for (const file of ${JSON.stringify(files)}) await import(file); +test("fixture guard records no client startup or socket connection", () => { + assert.deepEqual(calls, { redis: 0, cluster: 0, sockets: 0 }); + console.log("FIXTURE_GUARD_OK"); +}); +`; + writeFileSync(join(directory, "package.json"), '{"type":"module"}\n'); + writeFileSync(probe, source); + let passed = false; + try { + for (const value of [undefined, "", " \t "]) { + const env = { ...process.env }; + for (const key of Object.keys(env)) { + if (key.startsWith("REDIS_")) delete env[key]; + } + delete env.NODE_TEST_CONTEXT; + delete env.NODE_OPTIONS; + delete env.NODE_V8_COVERAGE; + const profile = JSON.stringify({ + present: value !== undefined, + value: value ?? "", + }); + env.FEDIFY_REDIS_FIXTURE_PROFILE = profile; + if (value !== undefined) { + for (const key of variables) env[key] = value; + } + const args = + process.versions.bun != null + ? ["test", "--timeout=10000", probe] + : process.versions.deno != null + ? [ + "test", + "--no-check", + "--allow-read", + "--allow-env", + "--allow-sys", + probe, + ] + : ["--experimental-transform-types", "--test", probe]; + const result = spawnSync(process.execPath, args, { + cwd: resolve(packageRoot, "../.."), + env, + encoding: "utf8", + timeout: 15_000, + }); + const detail = `Fixture ${JSON.stringify(value)}; probe retained at ${directory}\n${result.stdout}${result.stderr}`; + assert.ifError(result.error); + assert.equal(result.status, 0, detail); + assert.match(result.stdout + result.stderr, /FIXTURE_GUARD_OK/, detail); + assert( + (result.stdout + result.stderr).includes( + `FIXTURE_PROFILE_HEX ${Buffer.from(profile).toString("hex")}`, + ), + detail, + ); + } + passed = true; + } finally { + if (passed) rmSync(directory, { recursive: true, force: true }); + } +}); diff --git a/packages/redis/src/kv.atomic.test.ts b/packages/redis/src/kv.atomic.test.ts new file mode 100644 index 000000000..0f0422085 --- /dev/null +++ b/packages/redis/src/kv.atomic.test.ts @@ -0,0 +1,817 @@ +// Independent cases and namespace cleanup must run sequentially. +/* oxlint-disable eslint/no-await-in-loop */ +import type { KvKey, KvStoreCasEntry, KvStoreSetEntry } from "@fedify/fedify"; +import { RedisKvStore } from "@fedify/redis/kv"; +import { EncodingError, JsonCodec, type Codec } from "@fedify/redis/codec"; +import { testAtomicKvStore } from "@fedify/testing"; +import * as temporal from "@js-temporal/polyfill"; +import { Cluster, Redis } from "ioredis"; +import assert from "node:assert/strict"; +import { Buffer } from "node:buffer"; +import process from "node:process"; +import { test, type TestContext } from "node:test"; +import { + atomicMaxBytes, + atomicMaxEntries, + atomicRedisVersions, +} from "./atomic.ts"; + +const Temporal = globalThis.Temporal ?? temporal.Temporal; +const redisUrl = process.env.REDIS_URL?.trim() || undefined; +const clusterNodes = process.env.REDIS_CLUSTER_NODES?.trim() || undefined; +const versions = new Set(atomicRedisVersions); + +function skipUnsupported(t: TestContext, reason: string): void { + // Bun lacks dynamic skip; Deno may report a dynamic skip as a pass. + assert.notEqual( + process.env.FEDIFY_REDIS_REQUIRE_ATOMIC_TESTS, + "true", + `Atomic CI requires an audited Redis engine: ${reason}`, + ); + console.warn(`Skipping Redis atomic test: ${reason}`); + if (process.versions.bun == null) t.skip(reason); +} +if (process.env.FEDIFY_REDIS_REQUIRE_ATOMIC_TESTS === "true") { + test("Redis atomic CI requires every service fixture", () => { + for (const name of [ + "REDIS_URL", + "REDIS_CLUSTER_NODES", + "REDIS_ATOMIC_ADMIN_URL", + "REDIS_ATOMIC_CLUSTER_ADMIN_NODES", + "REDIS_ATOMIC_MIGRATION_NODES", + ]) { + assert(process.env[name]?.trim(), `Atomic CI requires ${name}`); + } + }); +} + +const modes = ["standalone", "cluster"] as const; + +function client(mode: (typeof modes)[number]): Redis | Cluster { + if (mode === "standalone") return new Redis(redisUrl!); + return new Cluster( + clusterNodes!.split(",").map((node) => { + const [host, port] = node.split(":"); + return { host, port: Number(port) }; + }), + ); +} + +async function cleanup(redis: Redis | Cluster, prefix: string): Promise { + // SCAN on Cluster itself has no slot affinity. Clean every primary directly. + const nodes = redis instanceof Cluster ? redis.nodes("master") : [redis]; + for (const node of nodes) { + let cursor = "0"; + do { + const [next, keys] = await node.scan( + cursor, + "MATCH", + `${prefix}*`, + "COUNT", + 100, + ); + cursor = next; + if (keys.length > 0) await node.del(...keys); + } while (cursor !== "0"); + } +} + +async function version(redis: Redis | Cluster): Promise { + const info = await redis.info("server"); + const match = /^redis_version:([^\r\n]+)/m.exec(info); + assert(match, "Redis must identify its version for these integration tests"); + return match[1]; +} + +for (const mode of modes) { + const skip = mode === "standalone" ? redisUrl == null : clusterNodes == null; + test(`Redis atomic ${mode} conformance`, { skip }, async (t) => { + if (skip) return; // Bun ignores the node:test skip option. + const probe = client(mode); + try { + const actual = await version(probe); + if (!versions.has(actual)) { + skipUnsupported( + t, + `Redis ${actual} is outside the audited batch engine list; rejection is tested separately`, + ); + return; + } + } finally { + await probe.quit(); + } + await testAtomicKvStore( + async () => { + const first = client(mode); + const second = client(mode); + const prefix = `{fedify_atomic_${crypto.randomUUID()}}::`; + try { + await Promise.all([first.ping(), second.ping()]); + return { + first: new RedisKvStore(first, { keyPrefix: prefix }), + second: new RedisKvStore(second, { keyPrefix: prefix }), + async dispose() { + try { + await cleanup(first, prefix); + } finally { + await Promise.all([first.quit(), second.quit()]); + } + }, + }; + } catch (error) { + first.disconnect(); + second.disconnect(); + throw error; + } + }, + { + objectKeyOrder: "significant", + ttl: { + short: Temporal.Duration.from({ seconds: 1 }), + wait: Temporal.Duration.from({ milliseconds: 2500 }), + }, + rejections(makeKey) { + const keys = Array.from({ length: atomicMaxEntries + 1 }, (_, i) => + makeKey("limit", String(i)), + ); + return { + getMany: keys, + setMany: keys.map((key, value) => ({ key, value })), + deleteMany: keys, + casMany: keys.map((key, value) => ({ + key, + expectedValue: value, + newValue: value + 1, + })), + }; + }, + }, + ); + }); + + test( + `Redis atomic ${mode} engine rejection or guarded acceptance`, + { skip }, + async () => { + if (skip) return; // Bun ignores the node:test skip option. + const redis = client(mode); + const prefix = `{fedify_engine_${crypto.randomUUID()}}::`; + const store = new RedisKvStore(redis, { keyPrefix: prefix }); + try { + await store.set(["a"], 1, { + ttl: Temporal.Duration.from({ seconds: 30 }), + }); + await store.set(["b"], 2); + const actual = await version(redis); + if (versions.has(actual)) { + const identity = await redis.eval( + "return redis.REDIS_VERSION_NUM", + 0, + ); + assert.equal( + identity, + actual.split(".").reduce((n, part) => n * 256 + Number(part), 0), + ); + assert.deepEqual(await store.getMany([["a"], ["b"]]), [1, 2]); + return; + } + const before = await redis.pttl(`${prefix}a`); + const calls = [ + () => store.getMany([["a"], ["b"]]), + () => + store.setMany([ + { key: ["a"], value: 10 }, + { key: ["b"], value: 20 }, + ]), + () => store.deleteMany([["a"], ["b"]]), + () => + store.casMany([ + { key: ["a"], expectedValue: 1, newValue: 10 }, + { key: ["b"], expectedValue: 2, newValue: 20 }, + ]), + ]; + for (const call of calls) { + // Redis before 7.0 rejects the script shebang during compilation, + // before the in-script version guard can provide its own message. + if (Number(actual.split(".")[0]) < 7) await assert.rejects(call); + else + await assert.rejects( + call, + /atomic KV batches require official Redis/, + ); + } + assert.deepEqual( + [await store.get(["a"]), await store.get(["b"])], + [1, 2], + ); + const after = await redis.pttl(`${prefix}a`); + assert( + after > 0 && after <= before, + "rejection must preserve the original expiry", + ); + assert.deepEqual(await store.getMany([]), []); + await store.setMany([]); + await store.deleteMany([]); + assert.equal(await store.casMany([]), true); + } finally { + await cleanup(redis, prefix); + await redis.quit(); + } + }, + ); + + test( + `Redis atomic ${mode} validation, codec, and physical deletion`, + { skip }, + async (t) => { + if (skip) return; // Bun ignores the node:test skip option. + const redis = client(mode); + const prefix = `{fedify_edges_${crypto.randomUUID()}}::`; + const store = new RedisKvStore(redis, { keyPrefix: prefix }); + try { + const actual = await version(redis); + if (!versions.has(actual)) { + skipUnsupported(t, `Redis ${actual} is not an audited batch engine`); + return; + } + await store.setMany([ + { key: ["a"], value: 1 }, + { key: ["b"], value: 2 }, + ]); + const ttl = Temporal.Duration.from({ seconds: 30 }); + await assert.rejects(store.set(["a"], undefined)); + await assert.rejects(store.cas(["a"], "wrong", undefined, { ttl })); + await assert.rejects(store.cas(["a"], 1, undefined, { ttl })); + await assert.rejects( + store.setMany([ + { key: ["a"], value: 10 }, + { key: ["b"], value: undefined }, + ]), + ); + await assert.rejects( + store.setMany([ + { key: ["a"], value: 10 }, + { + key: ["b"], + value: 20, + options: { ttl: Temporal.Duration.from({ months: 1 }) }, + }, + ]), + ); + await assert.rejects( + store.casMany([ + { key: ["a"], expectedValue: "wrong", newValue: 10 }, + { + key: ["b"], + expectedValue: 2, + newValue: undefined, + options: { ttl }, + }, + ]), + ); + await assert.rejects( + store.setMany([ + { key: ["a"], value: 10 }, + { + key: ["b"], + value: 20, + options: { ttl: Temporal.Duration.from({ seconds: 2 ** 31 }) }, + }, + ]), + ); + await assert.rejects( + store.setMany([ + { key: ["a"], value: 10 }, + { key: ["b"], value: "x".repeat(atomicMaxBytes) }, + ]), + ); + await assert.rejects( + store.casMany([ + { + key: ["a"], + expectedValue: "x".repeat(atomicMaxBytes), + newValue: 10, + }, + ]), + ); + // Single-key methods retain the old large-duration range; batch + // scripts deliberately accept only their audited, bounded TTL range. + const hugeTtl = Temporal.Duration.from({ + seconds: Number.MAX_SAFE_INTEGER, + milliseconds: 999, + }); + await store.set(["huge"], 1, { ttl: hugeTtl }); + assert.equal(await store.cas(["huge"], 1, 2, { ttl: hugeTtl }), true); + assert.equal(await store.get(["huge"]), 2); + await assert.rejects( + store.setMany([ + { key: ["huge"], value: 3, options: { ttl: hugeTtl } }, + ]), + RangeError, + ); + assert.equal(await store.get(["huge"]), 2); + await assert.rejects(store.deleteMany([["\ud800"], ["\ud801"]])); + await assert.rejects(store.getMany(new Array(2))); + await assert.rejects(store.setMany(new Array(2))); + await assert.rejects(store.casMany(new Array(2))); + assert.deepEqual(await store.getMany([["a"], ["b"]]), [1, 2]); + const throwing: Codec = { + encode(value) { + if (value === "fail") throw new Error("late codec failure"); + return new JsonCodec().encode(value); + }, + decode(value) { + return new JsonCodec().decode(value); + }, + }; + const failed = new RedisKvStore(redis, { + keyPrefix: prefix, + codec: throwing, + }); + await assert.rejects( + failed.setMany([ + { key: ["a"], value: 10 }, + { key: ["b"], value: "fail" }, + ]), + /late codec failure/, + ); + assert.deepEqual(await store.getMany([["a"], ["b"]]), [1, 2]); + await redis.del(`${prefix}b`); + await redis.rpush(`${prefix}b`, "nonstring"); + await assert.rejects( + store.casMany([ + { key: ["a"], expectedValue: 1, newValue: 10 }, + { key: ["b"], expectedValue: 2, newValue: 20 }, + ]), + /WRONGTYPE/, + ); + assert.equal(await store.get(["a"]), 1); + await store.setMany([ + { key: ["a"], value: 3 }, + { key: ["b"], value: 4 }, + ]); + await store.deleteMany([["a"]]); + assert.equal(await redis.exists(`${prefix}a`), 0); + assert.equal( + await store.casMany([ + { key: ["b"], expectedValue: 4, newValue: undefined }, + ]), + true, + ); + assert.equal(await redis.exists(`${prefix}b`), 0); + await store.setMany([ + { + key: ["expires"], + value: 1, + options: { ttl: Temporal.Duration.from({ milliseconds: 1 }) }, + }, + ]); + const remaining = await redis.pttl(`${prefix}expires`); + assert(remaining > 0 && remaining <= 1000); + await new Promise((resolve) => setTimeout(resolve, 1100)); + assert.equal(await store.get(["expires"]), undefined); + assert.equal(await redis.exists(`${prefix}expires`), 0); + await store.setMany([{ key: ["reopen"], value: { persistent: true } }]); + const reopened = client(mode); + try { + assert.deepEqual( + await new RedisKvStore(reopened, { keyPrefix: prefix }).getMany([ + ["reopen"], + ]), + [{ persistent: true }], + ); + } finally { + await reopened.quit(); + } + } finally { + await cleanup(redis, prefix); + await redis.quit(); + } + }, + ); +} + +test("JsonCodec rejects unencodable top-level values", () => { + const codec = new JsonCodec(); + for (const value of [undefined, Symbol("value"), () => 1]) + assert.throws(() => codec.encode(value), EncodingError); +}); + +for (const mode of modes) { + const skip = mode === "standalone" ? redisUrl == null : clusterNodes == null; + test( + `Redis atomic ${mode} binary codecs, capture, and prefixes`, + { skip }, + async (t) => { + if (skip) return; // Bun ignores the node:test skip option. + const redis = client(mode); + const prefix = `{fedify_binary_${crypto.randomUUID()}}::`; + try { + const actual = await version(redis); + if (!versions.has(actual)) { + skipUnsupported(t, `Redis ${actual} is not an audited batch engine`); + return; + } + const scratch = Buffer.alloc(32); + const codec: Codec = { + encode(value) { + assert.equal(typeof value, "string"); + if (value === "") return Buffer.alloc(0); + scratch.fill(0); + scratch.write(String(value)); + return scratch; + }, + decode(value) { + return value.toString().replace(/\0+$/, ""); + }, + }; + const store = new RedisKvStore(redis, { keyPrefix: prefix, codec }); + await store.setMany([ + { key: ["a"], value: "one" }, + { key: ["b"], value: "two" }, + ]); + assert.deepEqual(await store.getMany([["a"], ["b"]]), ["one", "two"]); + assert.equal(await store.cas(["a"], "one", "three"), true); + assert.equal( + await store.casMany([ + { key: ["a"], expectedValue: "three", newValue: "four" }, + { key: ["b"], expectedValue: "two", newValue: "" }, + ]), + true, + ); + assert.deepEqual(await store.getMany([["b"], ["missing"], ["a"]]), [ + "", + undefined, + "four", + ]); + assert.equal( + await store.casMany([ + { key: ["b"], expectedValue: "", newValue: "five" }, + ]), + true, + ); + + const identity: Codec = { + encode(value) { + assert(Buffer.isBuffer(value)); + return value; + }, + decode(value) { + return Buffer.from(value); + }, + }; + const binary = new RedisKvStore(redis, { + keyPrefix: prefix, + codec: identity, + }); + const value = Buffer.from([0, 255, 128, 1]); + const write = binary.setMany([{ key: ["bytes"], value }]); + value.fill(7); + await write; + assert.deepEqual(await binary.getMany([["bytes"]]), [ + Buffer.from([0, 255, 128, 1]), + ]); + assert.equal( + await binary.casMany([ + { + key: ["bytes"], + expectedValue: Buffer.from([0, 255, 128, 1]), + newValue: Buffer.from([254, 0]), + }, + ]), + true, + ); + assert.deepEqual(await binary.get(["bytes"]), Buffer.from([254, 0])); + + const json = new RedisKvStore(redis, { keyPrefix: prefix }); + const later = { key: ["captured-b"] as KvKey, value: 2 }; + let reads = 0; + const first = { + key: ["captured-a"] as KvKey, + get value() { + reads++; + return { + toJSON() { + later.value = 99; + return 1; + }, + }; + }, + }; + await json.setMany([first, later]); + assert.equal(reads, 1); + assert.deepEqual( + await json.getMany([["captured-a"], ["captured-b"]]), + [1, 2], + ); + + const binaryPrefix = Buffer.concat([ + Buffer.from(prefix), + Buffer.from([255]), + ]); + const frozen = Buffer.from(binaryPrefix); + const prefixed = new RedisKvStore(redis, { keyPrefix: binaryPrefix }); + binaryPrefix.fill(0); + try { + await prefixed.setMany([ + { key: ["a"], value: 1 }, + { key: ["b"], value: 2 }, + ]); + assert.deepEqual(await prefixed.getMany([["a"], ["b"]]), [1, 2]); + assert.equal( + await redis.get(Buffer.concat([frozen, Buffer.from("a")])), + "1", + ); + } finally { + await redis.del( + Buffer.concat([frozen, Buffer.from("a")]), + Buffer.concat([frozen, Buffer.from("b")]), + ); + } + } finally { + await cleanup(redis, prefix); + await redis.quit(); + } + + // A string prefix ending in a high surrogate joins a low surrogate key + // before ioredis encodes UTF-8. These are distinct physical keys. + const clientPrefix = `{fedify_client_${crypto.randomUUID()}}\ud800`; + const configured = + mode === "standalone" + ? new Redis(redisUrl!, { keyPrefix: clientPrefix }) + : new Cluster( + clusterNodes!.split(",").map((node) => { + const [host, port] = node.split(":"); + return { host, port: Number(port) }; + }), + { redisOptions: { keyPrefix: clientPrefix } }, + ); + const raw = client(mode); + const store = new RedisKvStore(configured, { keyPrefix: "" }); + try { + await store.setMany([ + { key: ["\udc00"], value: 1 }, + { key: ["\udc01"], value: 2 }, + ]); + assert.deepEqual(await store.getMany([["\udc00"], ["\udc01"]]), [1, 2]); + assert.equal(await raw.get(Buffer.from(clientPrefix + "\udc00")), "1"); + } finally { + await raw.del( + Buffer.from(clientPrefix + "\udc00"), + Buffer.from(clientPrefix + "\udc01"), + ); + await Promise.all([configured.quit(), raw.quit()]); + } + + const oversizedPrefix = `{'budget'}${"x".repeat(atomicMaxBytes)}`; + const oversized = + mode === "standalone" + ? new Redis(redisUrl!, { keyPrefix: oversizedPrefix }) + : new Cluster( + clusterNodes!.split(",").map((node) => { + const [host, port] = node.split(":"); + return { host, port: Number(port) }; + }), + { redisOptions: { keyPrefix: oversizedPrefix } }, + ); + try { + const store = new RedisKvStore(oversized); + await assert.rejects(store.getMany([["a"]]), RangeError); + await assert.rejects( + store.setMany([{ key: ["a"], value: 1 }]), + RangeError, + ); + } finally { + await oversized.quit(); + } + }, + ); +} + +for (const mode of modes) { + const skip = mode === "standalone" ? redisUrl == null : clusterNodes == null; + test(`Redis atomic ${mode} stringNumbers replies`, { skip }, async (t) => { + if (skip) return; // Bun ignores the node:test skip option. + const redis = + mode === "standalone" + ? new Redis(redisUrl!, { stringNumbers: true }) + : new Cluster( + clusterNodes!.split(",").map((node) => { + const [host, port] = node.split(":"); + return { host, port: Number(port) }; + }), + { redisOptions: { stringNumbers: true } }, + ); + const prefix = `{fedify_numbers_${crypto.randomUUID()}}::`; + const store = new RedisKvStore(redis, { keyPrefix: prefix }); + try { + const actual = await version(redis); + if (!versions.has(actual)) { + skipUnsupported(t, `Redis ${actual} is not an audited batch engine`); + return; + } + assert.equal( + await store.casMany([ + { key: ["a"], expectedValue: undefined, newValue: 1 }, + { key: ["b"], expectedValue: undefined, newValue: 2 }, + ]), + true, + ); + assert.equal(await store.cas(["a"], 1, 3), true); + assert.equal(await store.cas(["a"], 1, 4), false); + assert.equal( + await store.casMany([ + { key: ["a"], expectedValue: 1, newValue: 4 }, + { key: ["b"], expectedValue: 2, newValue: 5 }, + ]), + false, + ); + assert.deepEqual(await store.getMany([["a"], ["b"]]), [3, 2]); + } finally { + await redis.del(`${prefix}a`, `${prefix}b`); + await redis.quit(); + } + }); +} + +test("Redis CAS validates numeric, string, and binary replies without retrying", async () => { + const cases: [unknown, boolean][] = [ + [1, true], + ["1", true], + [Buffer.from("1"), true], + [0, false], + ["0", false], + [Buffer.from("0"), false], + ]; + const invalid = [ + undefined, + null, + 2, + -1, + true, + "01", + " 1", + Buffer.from([0xb1]), + Buffer.alloc(0), + ]; + for (const [reply, expected] of cases) { + let calls = 0; + const redis = { + options: {}, + eval() { + calls++; + return Promise.resolve(reply); + }, + sendCommand() { + calls++; + return Promise.resolve(reply); + }, + } as unknown as Redis; + const store = new RedisKvStore(redis); + assert.equal(await store.cas(["key"], 1, 2), expected); + assert.equal( + await store.casMany([{ key: ["key"], expectedValue: 1, newValue: 2 }]), + expected, + ); + assert.equal(calls, 2, "each CAS must dispatch once without a retry"); + } + for (const reply of invalid) { + let calls = 0; + const redis = { + options: {}, + eval() { + calls++; + return Promise.resolve(reply); + }, + sendCommand() { + calls++; + return Promise.resolve(reply); + }, + } as unknown as Redis; + const store = new RedisKvStore(redis); + await assert.rejects(store.cas(["key"], 1, 2), /Invalid Redis CAS reply/); + await assert.rejects( + store.casMany([{ key: ["key"], expectedValue: 1, newValue: 2 }]), + /Invalid Redis CAS reply/, + ); + assert.equal(calls, 2, "an invalid reply must not replay the request"); + } +}); + +for (const mode of modes) { + const skip = mode === "standalone" ? redisUrl == null : clusterNodes == null; + test( + `Redis atomic ${mode} successful auto-pipelined snapshots`, + { skip }, + async (t) => { + if (skip) return; // Bun ignores the node:test skip option. + const redis = + mode === "standalone" + ? new Redis(redisUrl!, { enableAutoPipelining: true }) + : new Cluster( + clusterNodes!.split(",").map((node) => { + const [host, port] = node.split(":"); + return { host, port: Number(port) }; + }), + { enableAutoPipelining: true }, + ); + const prefix = `{fedify_pipeline_${crypto.randomUUID()}}::`; + const store = new RedisKvStore(redis, { keyPrefix: prefix }); + try { + const actual = await version(redis); + if (!versions.has(actual)) { + skipUnsupported(t, `Redis ${actual} is not an audited batch engine`); + return; + } + await store.setMany([ + { key: ["a"], value: 1 }, + { key: ["b"], value: null }, + ]); + assert.deepEqual( + await store.getMany([["b"], ["missing"], ["a"], ["a"]]), + [null, undefined, 1, 1], + ); + assert.equal( + await store.casMany([{ key: ["a"], expectedValue: 1, newValue: 2 }]), + true, + ); + const codec: Codec = { + encode(value) { + assert(Buffer.isBuffer(value)); + return value; + }, + decode(value) { + return Buffer.from(value); + }, + }; + const binary = new RedisKvStore(redis, { keyPrefix: prefix, codec }); + await binary.setMany([ + { key: ["bytes"], value: Buffer.from([0, 255, 128]) }, + { key: ["empty"], value: Buffer.alloc(0) }, + ]); + const [first, second] = await Promise.all([ + binary.getMany([["bytes"], ["missing"], ["empty"], ["bytes"]]), + store.getMany([["a"], ["b"]]), + ]); + assert.deepEqual(first, [ + Buffer.from([0, 255, 128]), + undefined, + Buffer.alloc(0), + Buffer.from([0, 255, 128]), + ]); + assert.deepEqual(second, [2, null]); + await store.deleteMany([["a"], ["b"]]); + assert.deepEqual(await store.getMany([["a"], ["b"]]), [ + undefined, + undefined, + ]); + } finally { + await cleanup(redis, prefix); + await redis.quit(); + } + }, + ); +} + +test("Redis KV rejects malformed keys and non-array batches before dispatch", async () => { + let dispatched = 0; + function dispatch(): Promise { + dispatched++; + return Promise.resolve(null); + } + const redis = { + options: {}, + getBuffer: dispatch, + set: dispatch, + setex: dispatch, + del: dispatch, + eval: dispatch, + sendCommand: dispatch, + } as unknown as Redis; + const store = new RedisKvStore(redis); + const empty = [] as unknown as KvKey; + const sparse = new Array(1) as unknown as KvKey; + const nonString = ["a", 1] as unknown as KvKey; + const arrayLike = { 0: ["valid"], length: 1 } as unknown as readonly KvKey[]; + const operations = [ + () => store.get(empty), + () => store.set(empty, 1), + () => store.get(sparse), + () => store.getMany([empty]), + () => store.deleteMany([empty]), + () => store.setMany([{ key: nonString, value: 1 }]), + () => + store.casMany([{ key: sparse, expectedValue: undefined, newValue: 1 }]), + () => store.getMany(arrayLike), + () => store.deleteMany("x" as unknown as readonly KvKey[]), + () => + store.setMany({ + 0: { key: ["valid"], value: 1 }, + length: 1, + } as unknown as readonly KvStoreSetEntry[]), + ]; + for (const operation of operations) { + await assert.rejects(operation, TypeError); + assert.equal(dispatched, 0, "invalid input must not dispatch a command"); + } +}); diff --git a/packages/redis/src/kv.test.ts b/packages/redis/src/kv.test.ts index 99e1dc3f5..cae268dd8 100644 --- a/packages/redis/src/kv.test.ts +++ b/packages/redis/src/kv.test.ts @@ -322,10 +322,21 @@ nodeTest("RedisKvStore.cas() routes one key and rounds TTL", async () => { Buffer.from('"value"'), ]); assert.equal(calls[0].args[4], 1); - await store.cas(["key"], "value", undefined, { + await store.cas(["key"], "value", "updated", { ttl: Temporal.Duration.from({ seconds: -1 }), }); assert.equal(calls[1].args[4], 1); + await assert.rejects( + store.cas(["key"], "wrong", undefined, { + ttl: Temporal.Duration.from({ seconds: -1 }), + }), + TypeError, + ); + assert.equal( + calls.length, + 2, + "invalid deletions must reject before dispatch", + ); }); nodeTest("RedisKvStore.set() rounds a TTL up to whole seconds", async () => { diff --git a/packages/redis/src/kv.ts b/packages/redis/src/kv.ts index d88e8a68c..9ffd1fbdb 100644 --- a/packages/redis/src/kv.ts +++ b/packages/redis/src/kv.ts @@ -1,11 +1,20 @@ import type { + AtomicKvStore, KvKey, - KvStore, + KvStoreCasEntry, KvStoreListEntry, + KvStoreSetEntry, KvStoreSetOptions, } from "@fedify/fedify"; -import type { Cluster, Redis, RedisKey } from "ioredis"; +import { Command, type Cluster, type Redis, type RedisKey } from "ioredis"; import { Buffer } from "node:buffer"; +import { + atomicMaxBytes, + atomicMaxEntries, + atomicMaxTtlSeconds, + atomicMutationScript, + atomicSnapshotScript, +} from "./atomic.ts"; import { type Codec, JsonCodec } from "./codec.ts"; /** @@ -36,6 +45,15 @@ function expirySeconds(ttl: Temporal.Duration): number { return Math.max(1, Math.ceil(ttl.total("second"))); } +// EVAL integer replies can be strings with ioredis stringNumbers. Accept +// exact binary forms too, and never turn an unexpected reply into a mismatch. +function casResult(reply: unknown): boolean { + const result = Buffer.isBuffer(reply) ? reply.toString("utf8") : reply; + if (result === 1 || result === "1") return true; + if (result === 0 || result === "0") return false; + throw new TypeError("Invalid Redis CAS reply."); +} + const casScript = ` local current = redis.call('GET', KEYS[1]) if ARGV[1] == '1' then @@ -66,7 +84,9 @@ export interface RedisKvStoreOptions { /** * The codec to use for encoding and decoding values in the key–value store. * Defaults to {@link JsonCodec}. - * For {@link RedisKvStore.cas}, equal values must encode to identical bytes. + * For {@link RedisKvStore.cas} and {@link RedisKvStore.casMany}, equal values + * must encode to identical bytes. Values must round-trip through the codec; + * `undefined` is not storable. JSON object property order affects comparison. */ readonly codec?: Codec; } @@ -74,6 +94,36 @@ export interface RedisKvStoreOptions { /** * A key–value store that uses Redis as the underlying storage. * + * The atomic batch methods support official Redis 7.2.17, 7.4.12, and 8.10.3. + * Other versions reject nonempty batches before reading or writing. Supporting + * another release requires auditing its scripting/Cluster behavior and updating + * Fedify. Ordinary single-key methods retain their existing version support. + * Modified engines, forks, script debugging, and module command hooks are not + * supported atomic configurations; the version guard does not authenticate + * the server. + * + * A batch request is limited to 128 entries and 1 MiB of data: effective physical key + * bytes (including the client's prefix), encoded values/expectations, and + * argument strings, excluding the fixed script and protocol framing. Batch + * TTLs are rounded up to whole seconds, with a minimum of one second and a + * maximum of 2,147,483,647 seconds. Omitting a TTL clears the old expiry. + * + * On Cluster, every key must share one hash slot in database zero. An explicit + * prefix such as `"{application}::"` can select that scope; existing keys are + * never relocated. Redis rejects cross-slot calls without splitting them. + * Batches may reject during resharding when keys are missing or expired. + * Scripts need `EVAL`, `GET` or `EXISTS` as appropriate, and `SET`, `SETEX`, or + * `DEL` permissions on all keys. EVAL also requires both read and write key + * access, including for snapshots. Snapshot reads route to the primary. + * + * Atomicity does not imply persistence or failover durability. A lost reply or + * client replay can leave commit status unknown, even when a later CAS reply + * is `false`; re-read and re-evaluate before retrying. This adapter adds no + * retries or changes to client options. Use ordinary `delete()` to relieve + * memory pressure when flagged batch scripts are rejected under `maxmemory`. + * Cluster prefix listing retains its existing node-local SCAN limitation; + * `getMany()` is a snapshot of the requested keys, not a snapshot of `list()`. + * * @example * ```ts ignore * import { createFederation } from "@fedify/fedify"; @@ -98,7 +148,7 @@ export interface RedisKvStoreOptions { * }); * ``` */ -export class RedisKvStore implements KvStore { +export class RedisKvStore implements AtomicKvStore { #redis: Redis | Cluster; #keyPrefix: RedisKey; #keyPrefixStr: string; @@ -112,7 +162,9 @@ export class RedisKvStore implements KvStore { */ constructor(redis: Redis | Cluster, options: RedisKvStoreOptions = {}) { this.#redis = redis; - this.#keyPrefix = options.keyPrefix ?? "fedify::"; + const keyPrefix = options.keyPrefix ?? "fedify::"; + this.#keyPrefix = + typeof keyPrefix === "string" ? keyPrefix : Buffer.from(keyPrefix); this.#keyPrefixStr = typeof this.#keyPrefix === "string" ? this.#keyPrefix @@ -121,9 +173,18 @@ export class RedisKvStore implements KvStore { } #serializeKey(key: KvKey): RedisKey { - const suffix = key - .map((part: string) => part.replaceAll(":", "_:")) - .join("::"); + if (!Array.isArray(key) || key.length === 0) { + throw new TypeError("A KV key must be a nonempty array of strings."); + } + const parts: string[] = []; + for (let index = 0; index < key.length; index++) { + const part = key[index]; + if (!Object.hasOwn(key, index) || typeof part !== "string") { + throw new TypeError("A KV key must be a dense array of strings."); + } + parts.push(part.replaceAll(":", "_:")); + } + const suffix = parts.join("::"); if (typeof this.#keyPrefix === "string") { return `${this.#keyPrefix}${suffix}`; } @@ -131,6 +192,134 @@ export class RedisKvStore implements KvStore { return Buffer.concat([new Uint8Array(this.#keyPrefix), suffixBytes]); } + #encode(value: unknown): Buffer { + if (value === undefined) { + throw new TypeError("undefined is not a storable KV value."); + } + const encoded = this.#codec.encode(value); + if (!Buffer.isBuffer(encoded)) { + throw new TypeError("Codec.encode() must return a Buffer."); + } + // A codec can reuse its scratch buffer on the next encode(). + return Buffer.from(encoded); + } + + #batch(items: readonly T[]): T[] { + if (!Array.isArray(items)) { + throw new TypeError("A KV batch must be an array."); + } + const length = items.length; + if (length > atomicMaxEntries) { + throw new RangeError( + `A KV batch cannot exceed ${atomicMaxEntries} entries.`, + ); + } + const captured: T[] = []; + for (let index = 0; index < length; index++) { + if (!Object.hasOwn(items, index)) { + throw new TypeError("A KV batch must not be sparse."); + } + captured.push(items[index]); + } + return captured; + } + + #physicalKey(key: RedisKey): Buffer { + const prefix = this.#redis.options.keyPrefix; + // Match ioredis's string concatenation before UTF-8 encoding. In + // particular, a surrogate pair can span the prefix/key boundary. + if (typeof prefix === "string" && typeof key === "string") { + return Buffer.from(prefix + key); + } + return Buffer.concat([Buffer.from(prefix ?? ""), Buffer.from(key)]); + } + + #unique(keys: readonly RedisKey[]): void { + const encoded = keys.map((key) => this.#physicalKey(key).toString("hex")); + // This also prevents delete-then-write aliases on a migrating slot. + if (new Set(encoded).size !== keys.length) { + throw new TypeError("Duplicate keys in an atomic mutation."); + } + } + + #budget(keys: readonly RedisKey[], args: readonly (string | Buffer)[]): void { + let bytes = 0; + for (const key of keys) bytes += this.#physicalKey(key).length; + for (const arg of args) bytes += Buffer.byteLength(arg); + if (bytes > atomicMaxBytes) { + throw new RangeError( + `A KV batch cannot exceed ${atomicMaxBytes} data bytes.`, + ); + } + } + + #eval( + script: string, + keys: readonly RedisKey[], + args: readonly (string | Buffer)[], + replyEncoding: "utf8" | null, + ): Promise { + // ioredis groups automatic pipelines by EVAL's script argument rather + // than its first key. Direct dispatch keeps independent Cluster scopes + // apart and still uses the client's normal prefixing and primary routing. + return Promise.resolve( + this.#redis.sendCommand( + new Command("eval", [script, keys.length, ...keys, ...args], { + keyPrefix: this.#redis.options.keyPrefix, + replyEncoding, + }), + ), + ); + } + + async #mutate( + cas: boolean, + entries: readonly { + key: RedisKey; + expectedValue: unknown; + newValue: unknown; + ttl: Temporal.Duration | undefined; + }[], + ): Promise { + if (entries.length === 0) return true; + const keys = entries.map((entry) => entry.key); + this.#unique(keys); + // Normalize every TTL before invoking any caller-supplied codec. + const ttls = entries.map((entry) => { + if (entry.newValue === undefined && entry.ttl != null) { + throw new TypeError("A CAS deletion must not specify a TTL."); + } + if (entry.ttl == null) return "0"; + const seconds = expirySeconds(entry.ttl); + if (!Number.isSafeInteger(seconds) || seconds > atomicMaxTtlSeconds) { + throw new RangeError( + `A batch TTL cannot exceed ${atomicMaxTtlSeconds} seconds.`, + ); + } + return String(seconds); + }); + const args: (string | Buffer)[] = [cas ? "1" : "0"]; + for (let index = 0; index < entries.length; index++) { + const entry = entries[index]; + args.push( + entry.expectedValue === undefined ? "1" : "0", + entry.expectedValue === undefined + ? Buffer.alloc(0) + : this.#encode(entry.expectedValue), + entry.newValue === undefined ? "1" : "0", + entry.newValue === undefined + ? Buffer.alloc(0) + : this.#encode(entry.newValue), + ttls[index], + ); + } + // Recheck the effective prefix after codec callbacks, before dispatch. + this.#unique(keys); + this.#budget(keys, args); + const result = await this.#eval(atomicMutationScript, keys, args, "utf8"); + return casResult(result); + } + async get(key: KvKey): Promise { const serializedKey = this.#serializeKey(key); const encodedValue = await this.#redis.getBuffer(serializedKey); @@ -153,13 +342,10 @@ export class RedisKvStore implements KvStore { options?: KvStoreSetOptions | undefined, ): Promise { const serializedKey = this.#serializeKey(key); - const encodedValue = this.#codec.encode(value); - if (options?.ttl != null) { - await this.#redis.setex( - serializedKey, - expirySeconds(options.ttl), - encodedValue, - ); + const ttl = options?.ttl; + const encodedValue = this.#encode(value); + if (ttl != null) { + await this.#redis.setex(serializedKey, expirySeconds(ttl), encodedValue); } else { await this.#redis.set(serializedKey, encodedValue); } @@ -186,18 +372,93 @@ export class RedisKvStore implements KvStore { newValue: unknown, options?: KvStoreSetOptions, ): Promise { + const keyBytes = this.#serializeKey(key); + const ttl = options?.ttl; + if (newValue === undefined && ttl != null) { + throw new TypeError("A CAS deletion must not specify a TTL."); + } const absent = Buffer.alloc(0); const result = await this.#redis.eval( casScript, 1, - this.#serializeKey(key), + keyBytes, expectedValue === undefined ? "1" : "0", - expectedValue === undefined ? absent : this.#codec.encode(expectedValue), + expectedValue === undefined ? absent : this.#encode(expectedValue), newValue === undefined ? "1" : "0", - newValue === undefined ? absent : this.#codec.encode(newValue), - options?.ttl == null ? 0 : expirySeconds(options.ttl), + newValue === undefined ? absent : this.#encode(newValue), + ttl == null ? 0 : expirySeconds(ttl), ); - return result === 1; + return casResult(result); + } + + /** + * {@inheritDoc AtomicKvStore.getMany} + * @since 2.5.0 + */ + async getMany( + keys: readonly KvKey[], + ): Promise<(T | undefined)[]> { + const serialized = this.#batch(keys).map((key) => this.#serializeKey(key)); + if (serialized.length === 0) return []; + this.#budget(serialized, []); + const result = await this.#eval(atomicSnapshotScript, serialized, [], null); + if (!Array.isArray(result) || result.length !== serialized.length) { + throw new TypeError("Invalid Redis atomic snapshot reply."); + } + return result.map((encoded: unknown) => { + if (encoded === null) return undefined; + if (!Buffer.isBuffer(encoded)) { + throw new TypeError("Invalid Redis atomic snapshot value."); + } + return this.#codec.decode(encoded) as T; + }); + } + + /** + * {@inheritDoc AtomicKvStore.setMany} + * @since 2.5.0 + */ + async setMany(entries: readonly KvStoreSetEntry[]): Promise { + const captured = this.#batch(entries).map((entry) => { + const key = this.#serializeKey(entry.key); + const value = entry.value; + const ttl = entry.options?.ttl; + if (value === undefined) { + throw new TypeError("undefined is not a storable KV value."); + } + return { key, expectedValue: undefined, newValue: value, ttl }; + }); + await this.#mutate(false, captured); + } + + /** + * {@inheritDoc AtomicKvStore.deleteMany} + * @since 2.5.0 + */ + async deleteMany(keys: readonly KvKey[]): Promise { + await this.#mutate( + false, + this.#batch(keys).map((key) => ({ + key: this.#serializeKey(key), + expectedValue: undefined, + newValue: undefined, + ttl: undefined, + })), + ); + } + + /** + * {@inheritDoc AtomicKvStore.casMany} + * @since 2.5.0 + */ + async casMany(entries: readonly KvStoreCasEntry[]): Promise { + const captured = this.#batch(entries).map((entry) => ({ + key: this.#serializeKey(entry.key), + expectedValue: entry.expectedValue, + newValue: entry.newValue, + ttl: entry.options?.ttl, + })); + return await this.#mutate(true, captured); } #deserializeKey(redisKey: string): KvKey {