From 28f370d611f3b6ad191e147bbef40f4670cdd1a3 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 04:30:43 +0900 Subject: [PATCH 1/8] =?UTF-8?q?Add=20atomic=20Redis=20key=E2=80=93value=20?= =?UTF-8?q?operations?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Support snapshot reads and all-or-nothing multi-key mutations on standalone Redis and one Redis Cluster hash slot without relocating keys. Encode and validate complete batches before dispatch, then preflight reads and command permissions before the bounded Lua mutation phase. Restrict nonempty batches to audited Redis releases and fixed size limits to preserve the Cluster execution argument. Keep existing single-key TTL ranges while rejecting undefined storage and deletion TTLs. Preserve binary snapshots, auto-pipelining, and correct CAS replies with ioredis stringNumbers clients. Closes https://github.com/fedify-dev/fedify/issues/1365 Related to https://github.com/fedify-dev/fedify/issues/1361 Assisted-by: Codex:gpt-6.1-sol Assisted-by: Claude Code:claude-fable-5-2 Assisted-by: OpenCode:kimi-k3.1-pro Assisted-by: Codex:gpt-6-astra Assisted-by: Claude Code:claude-sonnet-5-5 --- changes.d/redis/atomic-kv-store.md | 19 + claude-plugin/skills/fedify/SKILL.md | 26 +- docs/manual/kv.md | 79 +- packages/redis/README.md | 126 +++ packages/redis/src/atomic.ts | 83 ++ packages/redis/src/codec.ts | 3 + packages/redis/src/kv.atomic-admin.test.ts | 215 +++++ packages/redis/src/kv.atomic-cluster.test.ts | 509 ++++++++++++ packages/redis/src/kv.atomic-gates.test.ts | 125 +++ packages/redis/src/kv.atomic.test.ts | 798 +++++++++++++++++++ packages/redis/src/kv.test.ts | 13 +- packages/redis/src/kv.ts | 301 ++++++- 12 files changed, 2264 insertions(+), 33 deletions(-) create mode 100644 changes.d/redis/atomic-kv-store.md create mode 100644 packages/redis/src/atomic.ts create mode 100644 packages/redis/src/kv.atomic-admin.test.ts create mode 100644 packages/redis/src/kv.atomic-cluster.test.ts create mode 100644 packages/redis/src/kv.atomic-gates.test.ts create mode 100644 packages/redis/src/kv.atomic.test.ts diff --git a/changes.d/redis/atomic-kv-store.md b/changes.d/redis/atomic-kv-store.md new file mode 100644 index 000000000..535bc7a0e --- /dev/null +++ b/changes.d/redis/atomic-kv-store.md @@ -0,0 +1,19 @@ +--- +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. 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. 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/redis/README.md b/packages/redis/README.md index 3372a840c..1af4d9f55 100644 --- a/packages/redis/README.md +++ b/packages/redis/README.md @@ -57,3 +57,129 @@ 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, re-read both values and re-evaluate the reservation rule. +} +~~~~ + +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. 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..750762457 --- /dev/null +++ b/packages/redis/src/kv.atomic-admin.test.ts @@ -0,0 +1,215 @@ +// 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. + 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 owner = nodes.find((node) => node.options.port === range[2][1]); + 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..d02b136a2 --- /dev/null +++ b/packages/redis/src/kv.atomic-cluster.test.ts @@ -0,0 +1,509 @@ +// 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. + 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; + const slow = `local start = redis.call('TIME') +local deadline = tonumber(start[1]) * 1000000 + tonumber(start[2]) + 100000 +repeat + local now = redis.call('TIME') + if tonumber(now[1]) * 1000000 + tonumber(now[2]) >= deadline then break end +until false +return 1`; + 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)); + 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((node) => node.options.port === range[2][1]); + 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..d243375d6 --- /dev/null +++ b/packages/redis/src/kv.atomic-gates.test.ts @@ -0,0 +1,125 @@ +// 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"; +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..856061a14 --- /dev/null +++ b/packages/redis/src/kv.atomic.test.ts @@ -0,0 +1,798 @@ +// 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. + console.warn(`Skipping Redis atomic test: ${reason}`); + if (process.versions.bun == null) t.skip(reason); +} +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 { From 6c687250af34e83e8a108051988af6ceea623cb9 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 08:47:51 +0900 Subject: [PATCH 2/8] Clarify Redis upgrade and retry guidance Explain why batch release restrictions preserve atomicity and what to change when upgrading CAS deletions. Warn in the reservation example that a replay can return false after a commit, requiring a receipt or version check before re-evaluating the rule. https://github.com/fedify-dev/fedify/pull/1434#discussion_r4555812313 https://github.com/fedify-dev/fedify/pull/1434#discussion_r4555812323 Assisted-by: Codex:gpt-6.1-sol --- CHANGES.md | 19 +++++++++++++++++++ changes.d/redis/atomic-kv-store.md | 7 +++++-- packages/redis/README.md | 3 ++- 3 files changed, 26 insertions(+), 3 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 012ee8b83..f927a8d53 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -434,6 +434,25 @@ To be released. [#1309]: https://github.com/fedify-dev/fedify/issues/1309 [#1334]: https://github.com/fedify-dev/fedify/pull/1334 +### @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. 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. Ordinary + single-key methods retain their existing Redis version and TTL range + support. + +[#1365]: https://github.com/fedify-dev/fedify/issues/1365 +[#1434]: https://github.com/fedify-dev/fedify/pull/1434 + ### @fedify/sqlite - Added `AtomicKvStore` support to `SqliteKvStore` for consistent snapshots diff --git a/changes.d/redis/atomic-kv-store.md b/changes.d/redis/atomic-kv-store.md index 535bc7a0e..7f93bb8d6 100644 --- a/changes.d/redis/atomic-kv-store.md +++ b/changes.d/redis/atomic-kv-store.md @@ -9,11 +9,14 @@ links: `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. Existing - keys are not relocated. [[#1361], [#1365], [#1434]] + 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. Ordinary + 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/packages/redis/README.md b/packages/redis/README.md index 1af4d9f55..bf07b235e 100644 --- a/packages/redis/README.md +++ b/packages/redis/README.md @@ -97,7 +97,8 @@ if (available !== undefined && available > 0) { { key: keys[0], expectedValue: available, newValue: available - 1 }, { key: keys[1], expectedValue: reserved, newValue: (reserved ?? 0) + 1 }, ]); - // On false, re-read both values and re-evaluate the reservation rule. + // On false, an earlier attempt may have committed. Check a receipt or + // version before re-evaluating the reservation rule (see below). } ~~~~ From 28d17a7a842e27725250faa8836e0453369f22fe Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 08:49:02 +0900 Subject: [PATCH 3/8] Match Cluster test owners by node ID Resolve each direct fixture connection with CLUSTER MYID and match the ID reported by CLUSTER SLOTS. Distinct hosts can share a port, so port matching can select the wrong owner in administrative and migration tests. https://github.com/fedify-dev/fedify/pull/1434#discussion_r4555812352 Assisted-by: Codex:gpt-6.1-sol --- packages/redis/src/kv.atomic-admin.test.ts | 7 ++++++- packages/redis/src/kv.atomic-cluster.test.ts | 7 ++++++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/packages/redis/src/kv.atomic-admin.test.ts b/packages/redis/src/kv.atomic-admin.test.ts index 750762457..4b97744c4 100644 --- a/packages/redis/src/kv.atomic-admin.test.ts +++ b/packages/redis/src/kv.atomic-admin.test.ts @@ -205,7 +205,12 @@ test( Array.isArray(range) && slot >= range[0] && slot <= range[1], ); assert(Array.isArray(range) && Array.isArray(range[2])); - const owner = nodes.find((node) => node.options.port === range[2][1]); + 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 { diff --git a/packages/redis/src/kv.atomic-cluster.test.ts b/packages/redis/src/kv.atomic-cluster.test.ts index d02b136a2..9e0fb2d17 100644 --- a/packages/redis/src/kv.atomic-cluster.test.ts +++ b/packages/redis/src/kv.atomic-cluster.test.ts @@ -317,6 +317,9 @@ return 1`; 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`)); @@ -325,7 +328,9 @@ return 1`; Array.isArray(range) && slot >= range[0] && slot <= range[1], ); assert(Array.isArray(range) && Array.isArray(range[2])); - source = nodes.find((node) => node.options.port === range[2][1]); + source = nodes.find( + (_, index) => nodeIds[index] === String(range[2][2]), + ); assert(source); if (Number(await source.cluster("COUNTKEYSINSLOT", slot)) === 0) break; slot = -1; From 4317b31be3356e2dfc20433a45b4b2ef85670015 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 08:50:29 +0900 Subject: [PATCH 4/8] Bound the Cluster migration test delay Use a finite Lua summation instead of polling TIME, which reads the cached Redis clock inside the script. Retain the expiry and unchanged value assertions that exercise the routing-to-execution clock gap. https://github.com/fedify-dev/fedify/pull/1434#discussion_r4555812370 Assisted-by: Codex:gpt-6.1-sol --- packages/redis/src/kv.atomic-cluster.test.ts | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/packages/redis/src/kv.atomic-cluster.test.ts b/packages/redis/src/kv.atomic-cluster.test.ts index 9e0fb2d17..683bc7c83 100644 --- a/packages/redis/src/kv.atomic-cluster.test.ts +++ b/packages/redis/src/kv.atomic-cluster.test.ts @@ -302,13 +302,11 @@ test( prefix = ""; let committed = false; let slowClient: Redis | undefined; - const slow = `local start = redis.call('TIME') -local deadline = tonumber(start[1]) * 1000000 + tonumber(start[2]) + 100000 -repeat - local now = redis.call('TIME') - if tonumber(now[1]) * 1000000 + tonumber(now[2]) >= deadline then break end -until false -return 1`; + // 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"); From b3115a76b35e8f8523f89601dcb6580598b0071a Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 08:53:51 +0900 Subject: [PATCH 5/8] Permit Deno fixture guard subprocesses Grant filesystem, subprocess, and system access only to the Redis package test task so the isolated fixture guard can create its probe and launch the matching runtime. Child Deno probes still omit network permission and intercept all client startup and socket attempts. https://github.com/fedify-dev/fedify/pull/1434#discussion_r4555812381 Assisted-by: Codex:gpt-6.1-sol --- packages/redis/deno.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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" } } From 9b9c972a294bd59ca9fdb2a650df7fe89beea5dc Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 09:04:15 +0900 Subject: [PATCH 6/8] Remove a stray word from Redis release notes Keep the release restriction explanation grammatical in the fragment and regenerated changelog. https://github.com/fedify-dev/fedify/pull/1434#discussion_r4557984786 Assisted-by: Codex:gpt-6.1-sol --- CHANGES.md | 7 +++++-- changes.d/redis/atomic-kv-store.md | 2 +- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index f927a8d53..7d9a4e03f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -441,12 +441,15 @@ To be released. `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. Existing - keys are not relocated. [[#1361], [#1365], [#1434]] + 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. Ordinary + 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/changes.d/redis/atomic-kv-store.md b/changes.d/redis/atomic-kv-store.md index 7f93bb8d6..706e2f519 100644 --- a/changes.d/redis/atomic-kv-store.md +++ b/changes.d/redis/atomic-kv-store.md @@ -8,7 +8,7 @@ links: 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. Existing + 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]] From 3fb6cbb32885674487959acad8892646c160c4f0 Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 09:16:05 +0900 Subject: [PATCH 7/8] Recognize PostgreSQL expiry expressions PostgreSQL 18.7 adds parentheses when pg_get_indexdef() prints SQL-syntax expressions. Compare pg_get_expr() over the stored index expression so an existing AT TIME ZONE expiry index is reused. Retain the index shape, opclass and predicate guards, and exercise both documented spellings during reinitialization. https://github.com/fedify-dev/fedify/actions/runs/43332852964/job/131581558684 https://github.com/fedify-dev/fedify/actions/runs/43332852964/job/131581558123 Assisted-by: Codex:gpt-6.1-sol Assisted-by: Claude Code:claude-sonnet-5-5 --- CHANGES.md | 2 +- changes.d/postgres/expiry-index-expression.md | 9 ++++++ packages/postgres/src/kv.index.test.ts | 28 ++++++++++--------- packages/postgres/src/kv.ts | 4 ++- 4 files changed, 28 insertions(+), 15 deletions(-) create mode 100644 changes.d/postgres/expiry-index-expression.md diff --git a/CHANGES.md b/CHANGES.md index 7d9a4e03f..b8c5c05e1 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -440,7 +440,7 @@ To be released. 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. Existing + 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]] 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/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)"} ) From c05a17563195c6b01e55baa7054aa68b8e4755eb Mon Sep 17 00:00:00 2001 From: Hong Minhee Date: Fri, 13 Nov 2026 13:29:18 +0900 Subject: [PATCH 8/8] Require Redis atomic conformance in CI Pin ordinary Redis services and test every audited release with standalone, administrative and three-primary Cluster fixtures. Run the package suites across Deno, Node.js and Bun and fail when required fixtures or supported Redis versions are missing. Serialize Node.js test files so administrative and slot-migration tests cannot interfere. Keep absent/blank configuration probes independent of the CI fixture requirement. https://github.com/fedify-dev/fedify/pull/1434#discussion_r4558929290 Assisted-by: Codex:gpt-6.1-sol Assisted-by: Claude Code:claude-sonnet-5-5 --- .github/workflows/main.yaml | 73 +++++++++++++++++++- CHANGES.md | 7 +- packages/redis/README.md | 9 +++ packages/redis/package.json | 2 +- packages/redis/src/kv.atomic-admin.test.ts | 5 ++ packages/redis/src/kv.atomic-cluster.test.ts | 5 ++ packages/redis/src/kv.atomic-gates.test.ts | 3 + packages/redis/src/kv.atomic.test.ts | 19 +++++ 8 files changed, 118 insertions(+), 5 deletions(-) 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 b8c5c05e1..44feb87ba 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -429,10 +429,16 @@ 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 @@ -454,7 +460,6 @@ To be released. support. [#1365]: https://github.com/fedify-dev/fedify/issues/1365 -[#1434]: https://github.com/fedify-dev/fedify/pull/1434 ### @fedify/sqlite diff --git a/packages/redis/README.md b/packages/redis/README.md index bf07b235e..5e82590e9 100644 --- a/packages/redis/README.md +++ b/packages/redis/README.md @@ -184,3 +184,12 @@ Run *src/kv.atomic-admin.test.ts* and the migration case in 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/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/kv.atomic-admin.test.ts b/packages/redis/src/kv.atomic-admin.test.ts index 4b97744c4..e1050a86a 100644 --- a/packages/redis/src/kv.atomic-admin.test.ts +++ b/packages/redis/src/kv.atomic-admin.test.ts @@ -15,6 +15,11 @@ 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); } diff --git a/packages/redis/src/kv.atomic-cluster.test.ts b/packages/redis/src/kv.atomic-cluster.test.ts index 683bc7c83..a56d53ebc 100644 --- a/packages/redis/src/kv.atomic-cluster.test.ts +++ b/packages/redis/src/kv.atomic-cluster.test.ts @@ -15,6 +15,11 @@ 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); } diff --git a/packages/redis/src/kv.atomic-gates.test.ts b/packages/redis/src/kv.atomic-gates.test.ts index d243375d6..b7e80d703 100644 --- a/packages/redis/src/kv.atomic-gates.test.ts +++ b/packages/redis/src/kv.atomic-gates.test.ts @@ -35,6 +35,9 @@ 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); diff --git a/packages/redis/src/kv.atomic.test.ts b/packages/redis/src/kv.atomic.test.ts index 856061a14..0f0422085 100644 --- a/packages/redis/src/kv.atomic.test.ts +++ b/packages/redis/src/kv.atomic.test.ts @@ -23,9 +23,28 @@ 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 {