perf(auth): coalesce session activity writes across processes (#4360)

src/backend/stores/session/SessionStore.js

@@ -38,10 +38,11 @@ import { PuterStore } from '../types';
 const CACHE_KEY_PREFIX = 'sessions:v2';
 const CACHE_TTL_SECONDS = 15 * 60;
 // Min interval between successive activity flushes per session/user.
-// In-memory throttle keeps DB writes bounded; multi-node duplicates
-// are harmless because the SQL guard `last_activity < ?` makes the
-// UPDATE idempotent.
+// The in-memory throttle is per process; a Redis `SET NX` gate behind it
+// keeps the other nodes from writing the same rows in the same window.
 const TOUCH_THROTTLE_MS = 60 * 1000;
+const TOUCH_GATE_TTL_SECONDS = TOUCH_THROTTLE_MS / 1000;
+const TOUCH_GATE_PREFIX = 'sessions:touch';
 // Hard cap to keep the throttle map from growing unbounded for
 // long-lived processes. Clearing only loses throttling — at worst a
 // brief burst of redundant UPDATEs.
@@ -1085,25 +1086,61 @@ export class SessionStore extends PuterStore {
             this.#lastUserTouchMs.set(userId, nowMs);
         }
 
+        const [sessionClaimed, userClaimed] = await Promise.all([
+            sessionDue &&
+                this.#claimTouchGate(`${TOUCH_GATE_PREFIX}:s:${uuid}`),
+            userDue && this.#claimTouchGate(`${TOUCH_GATE_PREFIX}:u:${userId}`),
+        ]);
+
         const tasks = [];
-        if (sessionDue) {
+        if (sessionClaimed) {
             tasks.push(
-                this.updateActivity(uuid, Math.floor(nowMs / 1000), {
-                    ip,
-                    userAgent,
-                }).catch(() => {}),
+                (async () => {
+                    try {
+                        await this.updateActivity(
+                            uuid,
+                            Math.floor(nowMs / 1000),
+                            { ip, userAgent },
+                        );
+                    } catch {
+                        // best-effort
+                    }
+                })(),
             );
         }
-        if (userDue) {
+        if (userClaimed) {
             tasks.push(
-                this.updateUserActivity(userId, sqlTimestamp(nowMs)).catch(
-                    () => {},
-                ),
+                (async () => {
+                    try {
+                        await this.updateUserActivity(
+                            userId,
+                            sqlTimestamp(nowMs),
+                        );
+                    } catch {
+                        // best-effort
+                    }
+                })(),
             );
         }
         await Promise.all(tasks);
     }
 
+    /** True when this node won the touch window; fails open on Redis errors. */
+    async #claimTouchGate(key) {
+        try {
+            const result = await this.clients.redis.set(
+                key,
+                '1',
+                'EX',
+                TOUCH_GATE_TTL_SECONDS,
+                'NX',
+            );
+            return result === 'OK';
+        } catch {
+            return true;
+        }
+    }
+
     // -- Internals ---------------------------------------------------
 
     #cacheKey(uuid) {

src/backend/stores/session/SessionStore.test.ts

@@ -1372,6 +1372,68 @@ describe('SessionStore', () => {
         });
     });
 
+    describe('touch shares its throttle across nodes', () => {
+        // A second store instance over the same DB and Redis stands in for
+        // another node, whose in-process throttle hasn't seen the session.
+        const peerStore = () =>
+            new target.constructor(
+                target.config,
+                target.clients,
+                target.stores,
+            );
+
+        const countTouchWrites = (spy: ReturnType<typeof vi.spyOn>) => {
+            const sqls = spy.mock.calls.map((c) => String(c[0]));
+            return {
+                session: sqls.filter((q) =>
+                    q.startsWith('UPDATE `sessions` SET `last_activity`'),
+                ).length,
+                user: sqls.filter((q) =>
+                    q.startsWith('UPDATE `user` SET `last_activity_ts`'),
+                ).length,
+            };
+        };
+
+        it('writes once per window when two nodes touch the same session', async () => {
+            const user = await makeUser();
+            const session = await target.create(user.id, { kind: 'web' });
+            const spy = vi.spyOn(server.clients.db, 'write');
+            try {
+                await target.touch({ uuid: session.uuid, userId: user.id });
+                await peerStore().touch({
+                    uuid: session.uuid,
+                    userId: user.id,
+                });
+                expect(countTouchWrites(spy)).toEqual({ session: 1, user: 1 });
+            } finally {
+                spy.mockRestore();
+            }
+        });
+
+        it('writes anyway when the Redis gate errors', async () => {
+            const user = await makeUser();
+            const session = await target.create(user.id, { kind: 'web' });
+            const writeSpy = vi.spyOn(server.clients.db, 'write');
+            const redisSpy = vi
+                .spyOn(server.clients.redis, 'set')
+                .mockRejectedValue(new Error('redis down'));
+            try {
+                await target.touch({ uuid: session.uuid, userId: user.id });
+                await peerStore().touch({
+                    uuid: session.uuid,
+                    userId: user.id,
+                });
+                expect(countTouchWrites(writeSpy)).toEqual({
+                    session: 2,
+                    user: 2,
+                });
+            } finally {
+                redisSpy.mockRestore();
+                writeSpy.mockRestore();
+            }
+        });
+    });
+
     describe('touch slides expires_at per kind', () => {
         it('extends expires_at on web sessions', async () => {
             const user = await makeUser();