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

@@ -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<keyof typeof real>).map(
+            (name) =>
+                vi.spyOn(redis, name).mockImplementation((async (
+                    ...args: unknown[]
+                ) => {
+                    const result = await (
+                        real[name] as (...a: unknown[]) => Promise<unknown>
+                    )(...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();

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<T>(suffix: string, fn: () => Promise<T>): Promise<T> {
-        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<T>(suffix: string, fn: () => Promise<T>): Promise<T> {
+        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' },
+                    );
+                },
+            },
+        );
     }
 
     /**

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<void> {
-        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<void> {
         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.
-            }
         }
     }
 

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<T>(suffix: string, run: () => Promise<T>): Promise<T> {
-        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<T>(suffix: string, run: () => Promise<T>): Promise<T> {
+        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. */

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 <https://www.gnu.org/licenses/>.
+ */
+
+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<unknown>
+                ).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<void>((resolve) => (finish = resolve));
+        let entered!: () => void;
+        const inside = new Promise<void>((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);
+        });
+    });
+});

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 <https://www.gnu.org/licenses/>.
+ */
+
+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<T> {
+    /** 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<T>;
+}
+
+/**
+ * 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 <T>(
+    redis: Pick<Cluster, 'set' | 'eval'>,
+    key: string,
+    fn: () => Promise<T>,
+    options: RedisLockOptions<T>,
+): Promise<T> => {
+    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.
+        }
+    }
+};