From abce4949e58d0c75dfa57a08c6d3966aeb05c5a0 Mon Sep 17 00:00:00 2001 From: Juan Castro Date: Wed, 12 Aug 2026 16:08:29 -0400 Subject: [PATCH] fix(cache): apply cache updates broadcast from peer regions --- .../cache/CacheReplicationService.test.ts | 84 +++++++++++++++++++ .../services/cache/CacheReplicationService.ts | 60 +++++++++++++ 2 files changed, 144 insertions(+) create mode 100644 src/backend/services/cache/CacheReplicationService.test.ts create mode 100644 src/backend/services/cache/CacheReplicationService.ts diff --git a/src/backend/services/cache/CacheReplicationService.test.ts b/src/backend/services/cache/CacheReplicationService.test.ts new file mode 100644 index 000000000..814e001cf --- /dev/null +++ b/src/backend/services/cache/CacheReplicationService.test.ts @@ -0,0 +1,84 @@ +/* + * 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 { v4 as uuidv4 } from 'uuid'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { PuterServer } from '../../server.ts'; +import { setupTestServer } from '../../testUtil.ts'; + +describe('CacheReplicationService', () => { + let server: PuterServer; + + beforeAll(async () => { + server = await setupTestServer(); + }); + + afterAll(async () => { + await server?.shutdown(); + }); + + const emitRemote = (cacheKey: unknown) => + server.clients.event.emitAndWait( + 'outer.cacheUpdate', + { cacheKey } as { cacheKey: string[] }, + { from_outside: true }, + ); + + it('drops keys a peer region invalidated', async () => { + const key = `cacherepl-${uuidv4()}`; + await server.clients.redis.set(key, 'stale'); + + await emitRemote([key]); + + expect(await server.clients.redis.get(key)).toBeNull(); + }); + + it('ignores a locally-emitted update, which already applied itself', async () => { + const key = `cacherepl-${uuidv4()}`; + await server.clients.redis.set(key, 'fresh'); + + await server.clients.event.emitAndWait( + 'outer.cacheUpdate', + { cacheKey: [key] }, + {}, + ); + + expect(await server.clients.redis.get(key)).toBe('fresh'); + }); + + it('deletes rather than adopting the sender payload', async () => { + const key = `cacherepl-${uuidv4()}`; + await server.clients.redis.set(key, 'ours'); + + await server.clients.event.emitAndWait( + 'outer.cacheUpdate', + { cacheKey: [key], data: 'theirs', ttlSeconds: 60 } as never, + { from_outside: true }, + ); + + // Their value came from their own replica; force a local re-read. + expect(await server.clients.redis.get(key)).toBeNull(); + }); + + it('survives a malformed payload', async () => { + await expect(emitRemote('not-an-array')).resolves.not.toThrow(); + await expect(emitRemote([123, '', null])).resolves.not.toThrow(); + await expect(emitRemote(undefined)).resolves.not.toThrow(); + }); +}); diff --git a/src/backend/services/cache/CacheReplicationService.ts b/src/backend/services/cache/CacheReplicationService.ts new file mode 100644 index 000000000..b821ea7e0 --- /dev/null +++ b/src/backend/services/cache/CacheReplicationService.ts @@ -0,0 +1,60 @@ +/* + * 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 { PuterService } from '../types'; + +/** + * Applies `outer.cacheUpdate` from peer regions. + * + * `PuterStore.publishCacheKeys({ broadcast: true })` writes its own cluster's + * Redis and emits the same mutation for peers; `BroadcastService` ships it over + * a webhook. Nothing consumed it on the far side, so cross-region cache + * replication silently no-oped. + * + * Always deletes, never re-writes the sender's payload: their value was derived + * from their own replica, so forcing a re-read here is the conservative move. + */ +export class CacheReplicationService extends PuterService { + override onServerStart(): void { + this.clients.event.on('outer.cacheUpdate', (_key, data, meta) => { + if (!(meta as { from_outside?: boolean })?.from_outside) return; + const raw = (data as { cacheKey?: unknown })?.cacheKey; + if (!Array.isArray(raw)) return; + const keys = raw.filter( + (key): key is string => typeof key === 'string' && key !== '', + ); + if (keys.length === 0) return; + void this.#invalidate(keys); + }); + } + + // Pipelined rather than a multi-key DEL, which would CROSSSLOT on Valkey. + async #invalidate(keys: string[]): Promise { + try { + const pipeline = this.clients.redis.pipeline(); + for (const key of keys) pipeline.del(key); + await pipeline.exec(); + } catch { + console.warn( + '[CacheReplicationService] failed to apply remote cache update:', + keys, + ); + } + } +}