From bb34ca2d7ed8106a1146942b370b115aaf682fa1 Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Sat, 10 Oct 2026 17:24:11 -0400 Subject: [PATCH] fix(redis): release locks with an atomic compare-and-delete ACLService, TeamService and the share digest flush each hand-rolled a SET NX lock whose release was a GET-then-DEL (or a blind DEL), so a holder whose TTL lapsed could delete the next holder's lock. They now share withRedisLock, which claims with a random token and releases via a Lua compare-and-delete. --- src/backend/services/acl/ACLService.test.ts | 48 ++++ src/backend/services/acl/ACLService.ts | 66 ++--- .../share/ShareNotificationService.ts | 45 ++-- src/backend/services/team/TeamService.ts | 67 ++--- src/backend/util/redisLock.test.ts | 242 ++++++++++++++++++ src/backend/util/redisLock.ts | 79 ++++++ 6 files changed, 429 insertions(+), 118 deletions(-) create mode 100644 src/backend/util/redisLock.test.ts create mode 100644 src/backend/util/redisLock.ts diff --git a/src/backend/services/acl/ACLService.test.ts b/src/backend/services/acl/ACLService.test.ts index b95362a04..d5b121052 100644 --- a/src/backend/services/acl/ACLService.test.ts +++ b/src/backend/services/acl/ACLService.test.ts @@ -863,6 +863,54 @@ describe('ACLService.statUserUser / setUserUser (integration)', () => { }); }); + it("does not release the next holder's lock when its own TTL lapses mid-release", async () => { + const issuer = await makeUser(); + const holder = await makeUser(); + const res = await ownedResource(issuer); + const key = `acl:set-user-user:${issuer.user!.id}:${holder.user!.id}:${res.uid}`; + const redis = server.clients.redis; + const real = { + set: redis.set.bind(redis), + get: redis.get.bind(redis), + del: redis.del.bind(redis), + eval: redis.eval.bind(redis), + }; + // After the first command on the held lock (the release), the lock + // lapses and another node takes it, as if the TTL ran out in between. + let held = false; + let lapsed = false; + const spies = (Object.keys(real) as Array).map( + (name) => + vi.spyOn(redis, name).mockImplementation((async ( + ...args: unknown[] + ) => { + const result = await ( + real[name] as (...a: unknown[]) => Promise + )(...args); + const touched = name === 'eval' ? args[2] : args[0]; + if (touched !== key || lapsed) return result; + if (!held) { + held = name === 'set' && result === 'OK'; + return result; + } + lapsed = true; + await real.del(key); + await real.set(key, 'next-holder', 'PX', 60_000, 'NX'); + return result; + }) as never), + ); + try { + await runWithContext({ actor: issuer }, () => + acl.setUserUser(issuer, holder, res, 'read'), + ); + } finally { + for (const spy of spies) spy.mockRestore(); + } + expect(lapsed).toBe(true); + expect(await redis.get(key)).toBe('next-holder'); + await redis.del(key); + }); + it('onlyIfHigher declines to downgrade an existing stronger grant', async () => { const issuer = await makeUser(); const holder = await makeUser(); diff --git a/src/backend/services/acl/ACLService.ts b/src/backend/services/acl/ACLService.ts index 8c5ce3b96..d3e6628e8 100644 --- a/src/backend/services/acl/ACLService.ts +++ b/src/backend/services/acl/ACLService.ts @@ -31,6 +31,7 @@ import { PermissionUtil } from '../permission/permissionUtil'; import { MANAGE_PERM_PREFIX } from '../permission/consts'; import { HttpError } from '../../core/http/HttpError.js'; import { actorHasSubscription } from '../metering/enforcement'; +import { withRedisLock } from '../../util/redisLock.js'; // -- Types ------------------------------------------------------------ @@ -618,52 +619,25 @@ export class ACLService extends PuterService { * Serialize writes to one (issuer, holder, node) triple. Fails open on a * Redis error — a narrow race beats taking sharing down with the cache. */ - async #withNodeLock(suffix: string, fn: () => Promise): Promise { - const key = `acl:set-user-user:${suffix}`; - const token = `${process.pid}:${Date.now()}:${Math.random()}`; - let held = false; - - try { - for (let attempt = 0; attempt < SET_USER_LOCK_ATTEMPTS; attempt++) { - const claimed = await this.clients.redis.set( - key, - token, - 'EX', - SET_USER_LOCK_TTL_SECONDS, - 'NX', - ); - if (claimed === 'OK') { - held = true; - break; - } - await new Promise((resolve) => - setTimeout(resolve, SET_USER_LOCK_RETRY_MS), - ); - } - if (!held) { - throw new HttpError( - 409, - 'another change to this share is in progress', - { legacyCode: 'conflict' }, - ); - } - } catch (err) { - if (err instanceof HttpError) throw err; - // Redis unavailable — proceed unserialized rather than fail. - return fn(); - } - - try { - return await fn(); - } finally { - try { - // Only clear our own claim — a lapsed TTL may have reassigned it. - const current = await this.clients.redis.get(key); - if (current === token) await this.clients.redis.del(key); - } catch { - // The TTL cleans up regardless. - } - } + #withNodeLock(suffix: string, fn: () => Promise): Promise { + return withRedisLock( + this.clients.redis, + `acl:set-user-user:${suffix}`, + fn, + { + ttlMs: SET_USER_LOCK_TTL_SECONDS * 1000, + attempts: SET_USER_LOCK_ATTEMPTS, + retryMs: SET_USER_LOCK_RETRY_MS, + onUnavailable: 'run', + onBusy: () => { + throw new HttpError( + 409, + 'another change to this share is in progress', + { legacyCode: 'conflict' }, + ); + }, + }, + ); } /** diff --git a/src/backend/services/share/ShareNotificationService.ts b/src/backend/services/share/ShareNotificationService.ts index 9d483be0a..c6a98d4f5 100644 --- a/src/backend/services/share/ShareNotificationService.ts +++ b/src/backend/services/share/ShareNotificationService.ts @@ -42,6 +42,7 @@ import { } from './shareDeepLink'; import { NOTIFY_FANOUT_CAP } from '../../stores/team/TeamStore'; import { blocksAllShares, type ResolvedShare } from './ShareService'; +import { withRedisLock } from '../../util/redisLock.js'; /** * How long one sharer stays quiet after reaching a recipient, and how long a @@ -874,25 +875,27 @@ export class ShareNotificationService extends PuterService { * handing each entry to exactly one flusher. */ async #flushDigest(key: string): Promise { - const lockKey = `share:digest:lock:${key}`; - try { - const locked = await this.clients.redis.set( - lockKey, - '1', - 'EX', - DIGEST_LOCK_SECONDS, - 'NX', - ); - if (locked !== 'OK') { - console.log('[share-notify] another flush holds the lock:', { - key, - }); - return; - } - } catch { - // No lock beats no mail; `take` still prevents double sends. - } + await withRedisLock( + this.clients.redis, + `share:digest:lock:${key}`, + () => this.#flushDigestLocked(key), + { + ttlMs: DIGEST_LOCK_SECONDS * 1000, + // No lock beats no mail; `take` still prevents double sends. + onUnavailable: 'run', + onBusy: () => { + console.log( + '[share-notify] another flush holds the lock:', + { + key, + }, + ); + }, + }, + ); + } + async #flushDigestLocked(key: string): Promise { try { const prefix = `share:digest:${key}:`; const { res } = await this.stores.kv.list({ @@ -1027,12 +1030,6 @@ export class ShareNotificationService extends PuterService { } } catch (err) { console.warn('[share-notify] digest flush failed:', err); - } finally { - try { - await this.clients.redis.del(lockKey); - } catch { - // The lock self-expires. - } } } diff --git a/src/backend/services/team/TeamService.ts b/src/backend/services/team/TeamService.ts index 1e79a9921..14a167d5e 100644 --- a/src/backend/services/team/TeamService.ts +++ b/src/backend/services/team/TeamService.ts @@ -31,6 +31,7 @@ import { ORG_SEAT_RESOLVER_PRIORITY, } from '../metering/consts.js'; import { runWithConcurrencyLimitSettled } from '../../util/concurrency.js'; +import { withRedisLock } from '../../util/redisLock.js'; // A free team is small on purpose; paying widens it. Both overridable in config. const FREE_SEAT_CAP = 3; @@ -397,55 +398,25 @@ export class TeamService extends PuterService { } /** - * Serializes a count-then-insert; same shape as `ACLService.#withNodeLock`. - * Refuses rather than running unserialized, or the cap is unenforced under - * concurrency. A lone create never contends. + * Serializes a count-then-insert. Refuses rather than running unserialized, + * or the cap is unenforced under concurrency. A lone create never + * contends. */ - async #withCapLock(suffix: string, run: () => Promise): Promise { - const key = `team:cap:${suffix}`; - const token = `${process.pid}:${Date.now()}:${Math.random()}`; - let held = false; - - try { - for (let attempt = 0; attempt < CAP_LOCK_ATTEMPTS; attempt++) { - const claimed = await this.clients.redis.set( - key, - token, - 'EX', - CAP_LOCK_TTL_SECONDS, - 'NX', - ); - if (claimed === 'OK') { - held = true; - break; - } - await new Promise((resolve) => - setTimeout(resolve, CAP_LOCK_RETRY_MS), - ); - } - } catch (e) { - // Same answer as contention below — the caller retries either way. - // Logged, because unlike contention the cause is ours. - console.warn('[team] cap lock unavailable:', e); - held = false; - } - if (!held) { - throw new HttpError(409, 'Busy — try that again in a moment', { - legacyCode: 'conflict', - }); - } - - try { - return await run(); - } finally { - try { - // Only clear our own claim — a lapsed TTL may have reassigned it. - const current = await this.clients.redis.get(key); - if (current === token) await this.clients.redis.del(key); - } catch { - /* the TTL clears it */ - } - } + #withCapLock(suffix: string, run: () => Promise): Promise { + return withRedisLock(this.clients.redis, `team:cap:${suffix}`, run, { + ttlMs: CAP_LOCK_TTL_SECONDS * 1000, + attempts: CAP_LOCK_ATTEMPTS, + retryMs: CAP_LOCK_RETRY_MS, + onUnavailable: 'busy', + onBusy: (err) => { + // Same answer as contention — the caller retries either way. + // Logged, because unlike contention the cause is ours. + if (err) console.warn('[team] cap lock unavailable:', err); + throw new HttpError(409, 'Busy — try that again in a moment', { + legacyCode: 'conflict', + }); + }, + }); } /** Renames or re-handles a team, refusing an unusable handle. */ diff --git a/src/backend/util/redisLock.test.ts b/src/backend/util/redisLock.test.ts new file mode 100644 index 000000000..dc89978d8 --- /dev/null +++ b/src/backend/util/redisLock.test.ts @@ -0,0 +1,242 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { randomUUID } from 'node:crypto'; +import { afterAll, describe, expect, it, vi } from 'vitest'; +import { RedisClient } from '../clients/redis/RedisClient.ts'; +import type { IConfig } from '../types'; +import { withRedisLock } from './redisLock.ts'; + +const redis = new RedisClient({ + port: 0, + extensions: [], + redis: {}, +} as unknown as IConfig); + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +const lockKey = () => `test:lock:${randomUUID()}`; + +const busy = () => 'busy'; + +const unexpectedBusy = (): never => { + throw new Error('lock unexpectedly busy'); +}; + +/** + * `redis`, except that right after the first command on `key` once it is held, + * the claim lapses and `nextToken` takes the key: the TTL running out between a + * holder's release round trips. + */ +const lapsingMidRelease = (key: string, nextToken: string) => { + const state = { held: false, lapsed: false }; + const proxy = new Proxy(redis, { + get(target, property, receiver) { + const value = Reflect.get(target, property, receiver); + if (!['set', 'get', 'del', 'eval'].includes(String(property))) + return value; + return async (...args: unknown[]) => { + const result = await ( + value as (...a: unknown[]) => Promise + ).apply(target, args); + const touched = property === 'eval' ? args[2] : args[0]; + if (touched !== key || state.lapsed) return result; + if (!state.held) { + state.held = property === 'set' && result === 'OK'; + return result; + } + state.lapsed = true; + await target.del(key); + await target.set(key, nextToken, 'PX', 60_000, 'NX'); + return result; + }; + }, + }); + return { proxy, state }; +}; + +describe('withRedisLock', () => { + afterAll(async () => { + await redis.onServerShutdown?.(); + }); + + it('runs fn under the lock and releases it after', async () => { + const key = lockKey(); + const result = await withRedisLock( + redis, + key, + async () => { + expect(await redis.get(key)).not.toBeNull(); + return 42; + }, + { ttlMs: 5_000, onUnavailable: 'busy', onBusy: unexpectedBusy }, + ); + expect(result).toBe(42); + expect(await redis.get(key)).toBeNull(); + }); + + it('releases the lock when fn throws', async () => { + const key = lockKey(); + await expect( + withRedisLock( + redis, + key, + async () => { + throw new Error('boom'); + }, + { ttlMs: 5_000, onUnavailable: 'busy', onBusy: unexpectedBusy }, + ), + ).rejects.toThrow('boom'); + expect(await redis.get(key)).toBeNull(); + }); + + it("keeps the next holder's lock when the TTL lapsed before release", async () => { + const key = lockKey(); + let finish!: () => void; + const finished = new Promise((resolve) => (finish = resolve)); + let entered!: () => void; + const inside = new Promise((resolve) => (entered = resolve)); + + const first = withRedisLock( + redis, + key, + async () => { + entered(); + await finished; + }, + { ttlMs: 20, onUnavailable: 'busy', onBusy: unexpectedBusy }, + ); + await inside; + await sleep(60); + expect(await redis.set(key, 'next-holder', 'PX', 60_000, 'NX')).toBe( + 'OK', + ); + finish(); + await first; + + expect(await redis.get(key)).toBe('next-holder'); + }); + + it("keeps the next holder's lock when the TTL lapses mid-release", async () => { + const key = lockKey(); + const { proxy, state } = lapsingMidRelease(key, 'next-holder'); + + await withRedisLock(proxy, key, async () => undefined, { + ttlMs: 5_000, + onUnavailable: 'busy', + onBusy: unexpectedBusy, + }); + + expect(state.lapsed).toBe(true); + expect(await redis.get(key)).toBe('next-holder'); + }); + + it('serializes concurrent callers', async () => { + const key = lockKey(); + let active = 0; + let peak = 0; + const results = await Promise.all( + Array.from({ length: 5 }, (_, i) => + withRedisLock( + redis, + key, + async () => { + active++; + peak = Math.max(peak, active); + await sleep(10); + active--; + return i; + }, + { + ttlMs: 5_000, + attempts: 200, + retryMs: 5, + onUnavailable: 'busy', + onBusy: unexpectedBusy, + }, + ), + ), + ); + expect(results).toEqual([0, 1, 2, 3, 4]); + expect(peak).toBe(1); + expect(await redis.get(key)).toBeNull(); + }); + + it('answers onBusy once every attempt finds the lock held', async () => { + const key = lockKey(); + await redis.set(key, 'other-holder', 'PX', 60_000); + const set = vi.spyOn(redis, 'set'); + const fn = vi.fn(async () => 'ran'); + const onBusy = vi.fn(busy); + try { + const result = await withRedisLock(redis, key, fn, { + ttlMs: 5_000, + attempts: 3, + retryMs: 1, + onUnavailable: 'run', + onBusy, + }); + expect(result).toBe('busy'); + expect(set).toHaveBeenCalledTimes(3); + } finally { + set.mockRestore(); + } + expect(fn).not.toHaveBeenCalled(); + expect(onBusy).toHaveBeenCalledWith(); + expect(await redis.get(key)).toBe('other-holder'); + }); + + describe('when Redis is unavailable', () => { + const down = new Error('redis down'); + const unavailable = { + set: async () => { + throw down; + }, + eval: async () => { + throw down; + }, + } as unknown as RedisClient; + + it("runs fn unlocked under 'run'", async () => { + const onBusy = vi.fn(busy); + const result = await withRedisLock( + unavailable, + lockKey(), + async () => 'ran', + { ttlMs: 5_000, attempts: 3, onUnavailable: 'run', onBusy }, + ); + expect(result).toBe('ran'); + expect(onBusy).not.toHaveBeenCalled(); + }); + + it("answers onBusy with the error under 'busy'", async () => { + const fn = vi.fn(async () => 'ran'); + const onBusy = vi.fn(busy); + const result = await withRedisLock(unavailable, lockKey(), fn, { + ttlMs: 5_000, + attempts: 3, + onUnavailable: 'busy', + onBusy, + }); + expect(result).toBe('busy'); + expect(fn).not.toHaveBeenCalled(); + expect(onBusy).toHaveBeenCalledWith(down); + }); + }); +}); diff --git a/src/backend/util/redisLock.ts b/src/backend/util/redisLock.ts new file mode 100644 index 000000000..379d37285 --- /dev/null +++ b/src/backend/util/redisLock.ts @@ -0,0 +1,79 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { randomUUID } from 'node:crypto'; +import type { Cluster } from 'ioredis'; + +// Compare-and-delete in one step: a GET then DEL could delete a lock another +// holder took after ours lapsed in between. +const RELEASE_SCRIPT = + "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) end return 0"; + +export interface RedisLockOptions { + /** How long a holder that never releases (crashed, stalled) blocks others. */ + ttlMs: number; + /** Claims tried before giving up, `retryMs` apart. Defaults to one. */ + attempts?: number; + retryMs?: number; + /** On a Redis error while claiming: run `fn` unlocked, or call `onBusy`. */ + onUnavailable: 'run' | 'busy'; + /** Every attempt found the lock held; `err` is set when Redis failed. */ + onBusy: (err?: unknown) => T | Promise; +} + +/** + * Run `fn` holding `key`, claimed with a per-call token and released only while + * that token is still the one stored. + */ +export const withRedisLock = async ( + redis: Pick, + key: string, + fn: () => Promise, + options: RedisLockOptions, +): Promise => { + const { ttlMs, attempts = 1, retryMs = 0, onUnavailable, onBusy } = options; + const token = randomUUID(); + let held = false; + + try { + for (let attempt = 0; attempt < attempts; attempt++) { + if ((await redis.set(key, token, 'PX', ttlMs, 'NX')) === 'OK') { + held = true; + break; + } + if (attempt + 1 < attempts) { + await new Promise((resolve) => setTimeout(resolve, retryMs)); + } + } + } catch (err) { + if (onUnavailable === 'run') return fn(); + return onBusy(err); + } + if (!held) return onBusy(); + + try { + return await fn(); + } finally { + try { + await redis.eval(RELEASE_SCRIPT, 1, key, token); + } catch { + // The TTL clears it. + } + } +};