From 156024d65f06ad69864045dc76926a7efe211ec3 Mon Sep 17 00:00:00 2001 From: Manjusaka Date: Mon, 22 Jun 2026 22:53:14 +0800 Subject: [PATCH 1/3] feat: PostgreSQL-backed cross-replica Data Cache handler (P2) Make admin changes propagate across replicas instantly. PicImpact's public read path is memoised with unstable_cache + revalidateTag (server/lib/cache.ts), but Next's default cache handler keeps that data per-instance, so a revalidateTag on the replica handling an admin write doesn't reach the others. - Add a custom Next cacheHandler (server/lib/pg-cache-handler.cjs) backed by Postgres: cached values + per-tag invalidation timestamps live in shared tables (next_cache_entries / next_cache_tags, auto-created), with a bounded module-level in-memory L1 for hot reads. Wire it in next.config.mjs and set cacheMaxMemorySize: 0 so Next's per-instance memory cache doesn't mask it. - Cross-instance invalidation works by get() returning a miss when any of an entry's tags was invalidated after the entry was written. The tag manifest is Postgres-authoritative: refreshed on a short TTL (so a missed NOTIFY or a transaction-pooled connection that can't LISTEN still converges within ~1s) and busted instantly via LISTEN/NOTIFY when a direct connection is available (uses DIRECT_URL for the listener). Ordering uses the Postgres server clock for both writes and invalidations, so replica clock skew can't miss an invalidation. Consistency is instant (a revalidated tag forces recompute on the next read on every replica); per-entry safety-net TTLs are left to Next. - Add a long-unused-entry sweep (server/lib/cache-cleanup.ts) driven by the same single external cron as the preprocess tick, so it never runs as a per-replica timer. - Add pg (and @types/pg); pg was already declared in serverExternalPackages. - Document the shared cache + DIRECT_URL/CACHE_HANDLER_DEBUG in docs/multi-replica.md and .env.example. Verified with a 2-instance build against Postgres: shared reads, cross-replica invalidation, cold-start reload after a missed NOTIFY, and a stable steady state. Co-Authored-By: Claude Opus 4.8 --- .env.example | 5 +- docs/multi-replica.md | 44 ++++-- hono/preprocess-tasks.ts | 4 + next.config.mjs | 7 + package.json | 2 + pnpm-lock.yaml | 133 +++++++++++++++- server/lib/cache-cleanup.ts | 28 ++++ server/lib/pg-cache-handler.cjs | 260 ++++++++++++++++++++++++++++++++ 8 files changed, 464 insertions(+), 19 deletions(-) create mode 100644 server/lib/cache-cleanup.ts create mode 100644 server/lib/pg-cache-handler.cjs diff --git a/.env.example b/.env.example index 4da69bc1..db73221d 100644 --- a/.env.example +++ b/.env.example @@ -16,4 +16,7 @@ NODEJS_HELPERS=0 # PREPROCESS_TICKER_ENABLED=false # 给 /api/v1/preprocess-tasks/tick 加共享密钥（可选）。设置后,调用方需在 # x-preprocess-tick-secret 请求头携带该值,否则返回 401;不设置则该端点开放。 -# PREPROCESS_TICK_SECRET= \ No newline at end of file +# PREPROCESS_TICK_SECRET= +# 跨 replica 共享缓存 handler 的调试日志（可选）。设为 true 打印 get/set/失效活动。 +# 注：handler 用 DATABASE_URL 读写、用 DIRECT_URL（若设置）做 LISTEN/NOTIFY。 +# CACHE_HANDLER_DEBUG=false \ No newline at end of file diff --git a/docs/multi-replica.md b/docs/multi-replica.md index 2433aea2..446e9976 100644 --- a/docs/multi-replica.md +++ b/docs/multi-replica.md @@ -73,21 +73,39 @@ TTL is 60s, so a sign-out or session revocation propagates to every replica within ~60s. Lower it further in `server/auth/index.ts` (`session.cookieCache.maxAge`) if you need faster propagation. -## 4. Public data cache (current behavior) +## 4. Public data cache (shared across replicas) The public read paths (gallery listings, album nav, public config) are cached -with Next.js' Data Cache and invalidated on admin writes via `revalidateTag`. -That cache is **per-replica in-memory** by default, so a `revalidateTag` call only -busts the cache on the replica that handled the write. Other replicas pick up the -change when their cache entry's safety-net TTL expires: - -- gallery listings: ~60s -- album nav / public config: ~1h (admin writes still bust them instantly on the - writing replica; the TTL only bounds cross-replica propagation) - -If you need instant cross-replica consistency for admin changes, a shared -PostgreSQL-backed cache handler is the planned next step (it routes the Data Cache -through Postgres + `LISTEN/NOTIFY` so an invalidation on one replica is seen by all). +with Next.js' Data Cache (`unstable_cache` in `server/lib/cache.ts`) and +invalidated on admin writes via `revalidateTag`. To make that invalidation work +across replicas, the Data Cache is routed through a PostgreSQL-backed cache +handler (`server/lib/pg-cache-handler.cjs`, wired in `next.config.mjs`): + +- Cached values and per-tag invalidation timestamps live in Postgres + (`next_cache_entries` / `next_cache_tags`, created automatically), shared by all + replicas. Each replica also keeps a small bounded in-memory L1 for hot reads. +- An admin write calls `revalidateTag` on one replica, which records the + invalidation in Postgres and broadcasts it via `LISTEN/NOTIFY`; every replica's + next read of an affected entry recomputes — so admin changes are visible across + all replicas effectively immediately. +- The tag state is **Postgres-authoritative**: each replica refreshes it from + Postgres on a short interval and on (re)connect, so a missed `NOTIFY` (e.g. a + replica restart, or a connection behind a transaction-mode pooler that can't + `LISTEN`) self-heals within ~1s rather than requiring a restart. +- The per-entry safety-net TTLs still apply (gallery ~60s, album/config ~1h) as a + backstop and to bound the background preprocess ticker's `variants_ready` gap. + +Relevant environment: + +- The handler uses `DATABASE_URL` for reads/writes and `DIRECT_URL` (when set) for + the `LISTEN` connection, since `LISTEN` needs a persistent session that a + transaction-mode pooler would drop. If only `DATABASE_URL` is set and it points + at such a pooler, instant `NOTIFY` is skipped and propagation falls back to the + ~1s Postgres refresh — still correct, just not instant. +- Set `CACHE_HANDLER_DEBUG=true` to log handler get/set/invalidation activity. + +The long-unused-entry sweep is driven by the same single cron as the tick +endpoint (§1), so it does not run as a per-replica timer. ## 5. In-flight task cancellation diff --git a/hono/preprocess-tasks.ts b/hono/preprocess-tasks.ts index ce1c54ad..77bc1b02 100644 --- a/hono/preprocess-tasks.ts +++ b/hono/preprocess-tasks.ts @@ -14,6 +14,7 @@ import { tickPreprocessTaskRuns, } from '~/server/tasks/image-preprocess-service' import { ADMIN_TASK_KEY_PREPROCESS_IMAGES, normalizePreprocessTaskScope } from '~/types/admin-tasks' +import { cleanupStaleCacheEntries } from '~/server/lib/cache-cleanup' import { ok } from '~/hono/_lib/response' import { badRequest, conflict, notFound, serverError, unauthorized } from '~/hono/_lib/errors' @@ -161,6 +162,9 @@ app.post('/tick', async (c) => { assertTickAuthorized(c.req.header(TICK_SECRET_HEADER)) try { const data = await tickPreprocessTaskRuns() + // Piggyback the shared cache-entry sweep on the same single cron that drives + // the preprocess tick, so it never runs as a per-replica timer. Best-effort. + await cleanupStaleCacheEntries() return ok(c, data) } catch (error) { console.error('Preprocess task tick failed:', error) diff --git a/next.config.mjs b/next.config.mjs index 6e15993f..d4d2662b 100644 --- a/next.config.mjs +++ b/next.config.mjs @@ -14,6 +14,13 @@ let nextConfig = { }, }, serverExternalPackages: ['pg'], + // Route the Data Cache (unstable_cache in server/lib/cache.ts) through a + // PostgreSQL-backed handler so tag invalidation propagates across replicas + // (see docs/multi-replica.md). cacheMaxMemorySize: 0 disables Next's own + // per-instance in-memory cache in front of the handler — the handler keeps + // its own bounded module-level L1 and Postgres is the shared source of truth. + cacheHandler: new URL('./server/lib/pg-cache-handler.cjs', import.meta.url).pathname, + cacheMaxMemorySize: 0, typescript: { ignoreBuildErrors: true, }, diff --git a/package.json b/package.json index 1ff49864..84ea09f4 100644 --- a/package.json +++ b/package.json @@ -89,6 +89,7 @@ "next-pwa": "5.6.0", "next-qrcode": "2.5.1", "next-themes": "0.4.6", + "pg": "^8.22.0", "react": "19.2.5", "react-day-picker": "9.14.0", "react-dom": "19.2.5", @@ -116,6 +117,7 @@ "@next/eslint-plugin-next": "16.2.4", "@tailwindcss/postcss": "4.2.2", "@types/node": "24.12.4", + "@types/pg": "^8.20.0", "@types/react": "19.2.14", "@types/react-dom": "19.2.3", "@types/rss": "0.0.32", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index f4742ddc..1c9a6983 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -20,7 +20,7 @@ importers: version: 3.1030.0 '@better-auth/passkey': specifier: 1.6.5 - version: 1.6.5(@better-auth/core@1.6.5(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@opentelemetry/api@1.9.1)(better-call@1.3.5(zod@4.3.6))(jose@6.1.3)(kysely@0.28.14)(nanostores@1.1.1))(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(better-auth@1.6.5(@opentelemetry/api@1.9.1)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3))(mongodb@7.1.0)(next@16.2.4(@babel/core@7.27.1)(@opentelemetry/api@1.9.1)(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(prisma@6.19.3(typescript@5.9.3))(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(better-call@1.3.5(zod@4.3.6))(nanostores@1.1.1) + version: 1.6.5(@better-auth/core@1.6.5(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@opentelemetry/api@1.9.1)(better-call@1.3.5(zod@4.3.6))(jose@6.1.3)(kysely@0.28.14)(nanostores@1.1.1))(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(better-auth@1.6.5(@opentelemetry/api@1.9.1)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3))(mongodb@7.1.0)(next@16.2.4(@babel/core@7.27.1)(@opentelemetry/api@1.9.1)(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(pg@8.22.0)(prisma@6.19.3(typescript@5.9.3))(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(better-call@1.3.5(zod@4.3.6))(nanostores@1.1.1) '@heroui/react': specifier: 3.0.2 version: 3.0.2(@types/react-dom@19.2.3(@types/react@19.2.14))(@types/react@19.2.14)(react-dom@19.2.5(react@19.2.5))(react@19.2.5)(tailwindcss@4.2.2) @@ -125,7 +125,7 @@ importers: version: 0.9.9 better-auth: specifier: 1.6.5 - version: 1.6.5(@opentelemetry/api@1.9.1)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3))(mongodb@7.1.0)(next@16.2.4(@babel/core@7.27.1)(@opentelemetry/api@1.9.1)(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(prisma@6.19.3(typescript@5.9.3))(react-dom@19.2.5(react@19.2.5))(react@19.2.5) + version: 1.6.5(@opentelemetry/api@1.9.1)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3))(mongodb@7.1.0)(next@16.2.4(@babel/core@7.27.1)(@opentelemetry/api@1.9.1)(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(pg@8.22.0)(prisma@6.19.3(typescript@5.9.3))(react-dom@19.2.5(react@19.2.5))(react@19.2.5) class-variance-authority: specifier: 0.7.1 version: 0.7.1 @@ -195,6 +195,9 @@ importers: next-themes: specifier: 0.4.6 version: 0.4.6(react-dom@19.2.5(react@19.2.5))(react@19.2.5) + pg: + specifier: ^8.22.0 + version: 8.22.0 react: specifier: 19.2.5 version: 19.2.5 @@ -271,6 +274,9 @@ importers: '@types/node': specifier: 24.12.4 version: 24.12.4 + '@types/pg': + specifier: ^8.20.0 + version: 8.20.0 '@types/react': specifier: 19.2.14 version: 19.2.14 @@ -3861,6 +3867,9 @@ packages: '@types/node@24.12.4': resolution: {integrity: sha512-GUUEShf+PBCGW2KaXwcIt3Yk+e3pkKwWKb9GSyM9WQVE+ep2jzmHdGsHzu4wgcZy5fN9FBdVzjpBQsYlpfpgLA==} + '@types/pg@8.20.0': + resolution: {integrity: sha512-bEPFOaMAHTEP1EzpvHTbmwR8UsFyHSKsRisLIHVMXnpNefSbGA1bD6CVy+qKjGSqmZqNqBDV2azOBo8TgkcVow==} + '@types/react-dom@19.2.3': resolution: {integrity: sha512-jp2L/eY6fn+KgVVQAOqYItbF0VY/YApe5Mz2F0aykSO8gx31bYCZyvSeYxCHKvzHG5eZjc+zyaS5BrBWya2+kQ==} peerDependencies: @@ -5982,6 +5991,40 @@ packages: perfect-debounce@1.0.0: resolution: {integrity: sha512-xCy9V055GLEqoFaHoC1SoLIaLmWctgCUaBaWxDZ7/Zx4CTyX7cJQLJOok/orfjZAh9kEYpjJa4d0KcJmCbctZA==} + pg-cloudflare@1.4.0: + resolution: {integrity: sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==} + + pg-connection-string@2.14.0: + resolution: {integrity: sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==} + + pg-int8@1.0.1: + resolution: {integrity: sha512-WCtabS6t3c8SkpDBUlb1kjOs7l66xsGdKpIPZsg4wR+B3+u9UAum2odSsF9tnvxg80h4ZxLWMy4pRjOsFIqQpw==} + engines: {node: '>=4.0.0'} + + pg-pool@3.14.0: + resolution: {integrity: sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==} + peerDependencies: + pg: '>=8.0' + + pg-protocol@1.15.0: + resolution: {integrity: sha512-cq9sECI5s0+uPUXjbz8ioyPJni6RzsRib0US67i5IoTZKw8fNeYlVE7u8F4dG7vEJJtc5wdD1K189lCCUwqWTQ==} + + pg-types@2.2.0: + resolution: {integrity: sha512-qTAAlrEsl8s4OiEQY69wDvcMIdQN6wdz5ojQiOy6YRMuynxenON0O5oCpJI6lshc6scgAY8qvJ2On/p+CXY0GA==} + engines: {node: '>=4'} + + pg@8.22.0: + resolution: {integrity: sha512-8wih1vVIBMxoUM2oB4soJsD9tDnDpLv4OXBJ+EJzFsvycD+lfyIreC2gGHq78f8jbLLt+bvlPTFdFZfJkOuzAA==} + engines: {node: '>= 16.0.0'} + peerDependencies: + pg-native: '>=3.0.1' + peerDependenciesMeta: + pg-native: + optional: true + + pgpass@1.0.5: + resolution: {integrity: sha512-FdW9r/jQZhSeohs1Z3sI1yxFQNFvMcnmfuj4WBMUTxOrAyLMaTcE1aAMBiTlbMNaXvBCQuVi0R7hd8udDSP7ug==} + picocolors@1.1.1: resolution: {integrity: sha512-xceH2snhtb5M9liqDsmEw56le376mTZkEX/jEb/RxNFyegNul7eNslCXP9FDj/Lcu0X8KEyMceP2ntpaHrDEVA==} @@ -6035,6 +6078,22 @@ packages: resolution: {integrity: sha512-3Ybi1tAuwAP9s0r1UQ2J4n5Y0G05bJkpUIO0/bI9MhwmD70S5aTWbXGBwxHrelT+XM1k6dM0pk+SwNkpTRN7Pg==} engines: {node: ^10 || ^12 || >=14} + postgres-array@2.0.0: + resolution: {integrity: sha512-VpZrUqU5A69eQyW2c5CA1jtLecCsN2U/bD6VilrFDWq5+5UIEVO7nazS3TEcHf1zuPYO/sqGvUvW62g86RXZuA==} + engines: {node: '>=4'} + + postgres-bytea@1.0.1: + resolution: {integrity: sha512-5+5HqXnsZPE65IJZSMkZtURARZelel2oXUEO8rH83VS/hxH5vv1uHquPg5wZs8yMAfdv971IU+kcPUczi7NVBQ==} + engines: {node: '>=0.10.0'} + + postgres-date@1.0.7: + resolution: {integrity: sha512-suDmjLVQg78nMK2UZ454hAG+OAW+HQPZ6n++TNDUX+L0+uUlLywnoxJKDou51Zm+zTCjrCl0Nq6J9C5hP9vK/Q==} + engines: {node: '>=0.10.0'} + + postgres-interval@1.2.0: + resolution: {integrity: sha512-9ZhXKM/rw350N1ovuWHbGxnGh/SNJ4cnxHiM0rxE4VN41wsg8P8zWn9hv/buK00RP4WvlOyr/RBDiptyxVbkZQ==} + engines: {node: '>=0.10.0'} + potpack@2.1.0: resolution: {integrity: sha512-pcaShQc1Shq0y+E7GqJqvZj8DTthWV1KeHGdi0Z6IAin2Oi3JnLCOfwnCo84qc+HAp52wT9nK9H7FAJp5a44GQ==} @@ -6510,6 +6569,10 @@ packages: resolution: {integrity: sha512-NzNVhJDYpwceVVii8/Hu6DKfD2G+NrQHlS/V/qgv763EYudVwEcMQNxd2lh+0VrUByXN/oJkl5grOhYWvQUYiw==} engines: {node: '>=0.10.0'} + split2@4.2.0: + resolution: {integrity: sha512-UcjcJOWknrNkF6PLX83qcHM6KHgVKNkV62Y8a5uYDVv9ydGQVwAHMKqHdJje1VTWpljG0WYpCDhrCdAOYH4TWg==} + engines: {node: '>= 10.x'} + stable-hash@0.0.5: resolution: {integrity: sha512-+L3ccpzibovGXFK+Ap/f8LOS0ahMrHTf3xu7mMLSpEGU0EO9ucaysSylKo9eRDFNhWve/y275iPmIZ4z39a9iA==} @@ -7004,6 +7067,10 @@ packages: xml@1.0.1: resolution: {integrity: sha512-huCv9IH9Tcf95zuYCsQraZtWnJvBtLVE0QHMOs8bWyZAFZNDcYjsPq1nEx8jKA9y+Beo9v+7OBPRisQTjinQMw==} + xtend@4.0.2: + resolution: {integrity: sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==} + engines: {node: '>=0.4'} + y18n@4.0.3: resolution: {integrity: sha512-JKhqTOwSrqNA1NY5lSztJ1GrBiUodLMmIZuLiDaMRJ+itFd+ABVE8XBjOvIWL+rSqNDC74LCSFmlb/U4UZ4hJQ==} @@ -8203,14 +8270,14 @@ snapshots: optionalDependencies: mongodb: 7.1.0 - '@better-auth/passkey@1.6.5(@better-auth/core@1.6.5(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@opentelemetry/api@1.9.1)(better-call@1.3.5(zod@4.3.6))(jose@6.1.3)(kysely@0.28.14)(nanostores@1.1.1))(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(better-auth@1.6.5(@opentelemetry/api@1.9.1)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3))(mongodb@7.1.0)(next@16.2.4(@babel/core@7.27.1)(@opentelemetry/api@1.9.1)(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(prisma@6.19.3(typescript@5.9.3))(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(better-call@1.3.5(zod@4.3.6))(nanostores@1.1.1)': + '@better-auth/passkey@1.6.5(@better-auth/core@1.6.5(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@opentelemetry/api@1.9.1)(better-call@1.3.5(zod@4.3.6))(jose@6.1.3)(kysely@0.28.14)(nanostores@1.1.1))(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(better-auth@1.6.5(@opentelemetry/api@1.9.1)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3))(mongodb@7.1.0)(next@16.2.4(@babel/core@7.27.1)(@opentelemetry/api@1.9.1)(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(pg@8.22.0)(prisma@6.19.3(typescript@5.9.3))(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(better-call@1.3.5(zod@4.3.6))(nanostores@1.1.1)': dependencies: '@better-auth/core': 1.6.5(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@opentelemetry/api@1.9.1)(better-call@1.3.5(zod@4.3.6))(jose@6.1.3)(kysely@0.28.14)(nanostores@1.1.1) '@better-auth/utils': 0.4.0 '@better-fetch/fetch': 1.1.21 '@simplewebauthn/browser': 13.2.2 '@simplewebauthn/server': 13.2.3 - better-auth: 1.6.5(@opentelemetry/api@1.9.1)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3))(mongodb@7.1.0)(next@16.2.4(@babel/core@7.27.1)(@opentelemetry/api@1.9.1)(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(prisma@6.19.3(typescript@5.9.3))(react-dom@19.2.5(react@19.2.5))(react@19.2.5) + better-auth: 1.6.5(@opentelemetry/api@1.9.1)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3))(mongodb@7.1.0)(next@16.2.4(@babel/core@7.27.1)(@opentelemetry/api@1.9.1)(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(pg@8.22.0)(prisma@6.19.3(typescript@5.9.3))(react-dom@19.2.5(react@19.2.5))(react@19.2.5) better-call: 1.3.5(zod@4.3.6) nanostores: 1.1.1 zod: 4.3.6 @@ -11606,6 +11673,12 @@ snapshots: dependencies: undici-types: 7.16.0 + '@types/pg@8.20.0': + dependencies: + '@types/node': 24.12.4 + pg-protocol: 1.15.0 + pg-types: 2.2.0 + '@types/react-dom@19.2.3(@types/react@19.2.14)': dependencies: '@types/react': 19.2.14 @@ -12075,7 +12148,7 @@ snapshots: baseline-browser-mapping@2.10.10: {} - better-auth@1.6.5(@opentelemetry/api@1.9.1)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3))(mongodb@7.1.0)(next@16.2.4(@babel/core@7.27.1)(@opentelemetry/api@1.9.1)(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(prisma@6.19.3(typescript@5.9.3))(react-dom@19.2.5(react@19.2.5))(react@19.2.5): + better-auth@1.6.5(@opentelemetry/api@1.9.1)(@prisma/client@6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3))(mongodb@7.1.0)(next@16.2.4(@babel/core@7.27.1)(@opentelemetry/api@1.9.1)(react-dom@19.2.5(react@19.2.5))(react@19.2.5))(pg@8.22.0)(prisma@6.19.3(typescript@5.9.3))(react-dom@19.2.5(react@19.2.5))(react@19.2.5): dependencies: '@better-auth/core': 1.6.5(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@opentelemetry/api@1.9.1)(better-call@1.3.5(zod@4.3.6))(jose@6.1.3)(kysely@0.28.14)(nanostores@1.1.1) '@better-auth/drizzle-adapter': 1.6.5(@better-auth/core@1.6.5(@better-auth/utils@0.4.0)(@better-fetch/fetch@1.1.21)(@opentelemetry/api@1.9.1)(better-call@1.3.5(zod@4.3.6))(jose@6.1.3)(kysely@0.28.14)(nanostores@1.1.1))(@better-auth/utils@0.4.0) @@ -12098,6 +12171,7 @@ snapshots: '@prisma/client': 6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3) mongodb: 7.1.0 next: 16.2.4(@babel/core@7.27.1)(@opentelemetry/api@1.9.1)(react-dom@19.2.5(react@19.2.5))(react@19.2.5) + pg: 8.22.0 prisma: 6.19.3(typescript@5.9.3) react: 19.2.5 react-dom: 19.2.5(react@19.2.5) @@ -13901,6 +13975,41 @@ snapshots: perfect-debounce@1.0.0: {} + pg-cloudflare@1.4.0: + optional: true + + pg-connection-string@2.14.0: {} + + pg-int8@1.0.1: {} + + pg-pool@3.14.0(pg@8.22.0): + dependencies: + pg: 8.22.0 + + pg-protocol@1.15.0: {} + + pg-types@2.2.0: + dependencies: + pg-int8: 1.0.1 + postgres-array: 2.0.0 + postgres-bytea: 1.0.1 + postgres-date: 1.0.7 + postgres-interval: 1.2.0 + + pg@8.22.0: + dependencies: + pg-connection-string: 2.14.0 + pg-pool: 3.14.0(pg@8.22.0) + pg-protocol: 1.15.0 + pg-types: 2.2.0 + pgpass: 1.0.5 + optionalDependencies: + pg-cloudflare: 1.4.0 + + pgpass@1.0.5: + dependencies: + split2: 4.2.0 + picocolors@1.1.1: {} picomatch@2.3.1: {} @@ -13945,6 +14054,16 @@ snapshots: picocolors: 1.1.1 source-map-js: 1.2.1 + postgres-array@2.0.0: {} + + postgres-bytea@1.0.1: {} + + postgres-date@1.0.7: {} + + postgres-interval@1.2.0: + dependencies: + xtend: 4.0.2 + potpack@2.1.0: {} prelude-ls@1.2.1: {} @@ -14552,6 +14671,8 @@ snapshots: dependencies: extend-shallow: 3.0.2 + split2@4.2.0: {} + stable-hash@0.0.5: {} stop-iteration-iterator@1.1.0: @@ -15214,6 +15335,8 @@ snapshots: xml@1.0.1: {} + xtend@4.0.2: {} + y18n@4.0.3: {} yallist@3.1.1: {} diff --git a/server/lib/cache-cleanup.ts b/server/lib/cache-cleanup.ts new file mode 100644 index 00000000..0b3bf246 --- /dev/null +++ b/server/lib/cache-cleanup.ts @@ -0,0 +1,28 @@ +import 'server-only' + +import { db } from '~/server/lib/db' + +// Cached keys are deterministic (unstable_cache key + args), and active entries +// are rewritten well within their TTL (gallery ≤60s, albums/config ≤1h), so the +// entries table is naturally bounded by the number of distinct queries. This +// sweep just drops rows that haven't been written for a long time — i.e. keys +// no longer in use — so the table can't accumulate dead entries indefinitely. +const STALE_AFTER_MS = 7 * 24 * 60 * 60 * 1000 + +/** + * Best-effort deletion of long-unused cache entries. Driven by the same single + * external cron that drives the preprocess tick (see docs/multi-replica.md), so + * it does NOT run as a per-replica timer. Never throws — cache cleanup must not + * break the caller (e.g. the tick endpoint). + */ +export async function cleanupStaleCacheEntries(): Promise { + const cutoff = Date.now() - STALE_AFTER_MS + try { + // Table is owned by the PG cache handler (server/lib/pg-cache-handler.cjs). + return await db.$executeRaw`DELETE FROM next_cache_entries WHERE last_modified < ${cutoff}` + } catch (error) { + // Table may not exist yet (handler hasn't run) or DB hiccup — ignore. + console.warn('Cache cleanup skipped:', error instanceof Error ? error.message : error) + return 0 + } +} diff --git a/server/lib/pg-cache-handler.cjs b/server/lib/pg-cache-handler.cjs new file mode 100644 index 00000000..26689928 --- /dev/null +++ b/server/lib/pg-cache-handler.cjs @@ -0,0 +1,260 @@ +'use strict' + +/** + * Shared, cross-replica Next.js Data Cache handler backed by PostgreSQL. + * + * Why: PicImpact's public read path is memoised with `unstable_cache` + tag + * invalidation (see server/lib/cache.ts). Next's default cache handler keeps + * that data in per-instance memory, so when the app runs as multiple replicas a + * `revalidateTag` on the replica handling an admin write does NOT reach the + * others — they keep serving stale data until their entry's safety-net TTL + * expires. This handler routes the Data Cache through Postgres so an + * invalidation on one replica is seen by all of them. + * + * Design (validated by a 2-instance PoC + against the real build): + * - L2 = Postgres: `next_cache_entries` (cached values + when each was written) + * and `next_cache_tags` (per-tag last-invalidation time). + * - L1 = a bounded, MODULE-LEVEL in-memory cache. Next instantiates the + * handler class per request, so all shared state must live at module scope, + * not on the instance. + * - Cross-instance invalidation lever = `get()` returns null (cache miss) when + * any of an entry's tags was invalidated after the entry was written. Next's + * own in-process tag manifest can't be fed from another replica, but it only + * runs when `get()` returns data — so making `get()` the decision point, + * against the shared manifest, is what makes invalidation cross-replica. + * - The tag manifest is Postgres-authoritative: refreshed from PG on a short + * TTL (so a missed NOTIFY / a transaction-pooled connection that can't LISTEN + * still converges within ~1s) and busted instantly via LISTEN/NOTIFY when a + * direct connection is available. + * - Ordering uses the Postgres server clock for BOTH an entry's write time and + * a tag's invalidation time, so the "was this entry written before the tag + * was invalidated?" comparison can't be fooled by clock skew between replicas. + * - Consistency model = instant: any `revalidateTag` turns every entry written + * before it into a miss → immediate recompute on the next read, on every + * replica. We intentionally do not implement cross-instance + * stale-while-revalidate (which would let a replica serve one stale read + * after an admin change); instant visibility is the chosen behaviour. Per-entry + * safety-net TTLs (the `revalidate` seconds on each `unstable_cache`) are left + * to Next, which compares age against `revalidate` using the `lastModified` + * we return. + */ + +const { Pool, Client } = require('pg') + +const ENTRY_TABLE = 'next_cache_entries' +const TAG_TABLE = 'next_cache_tags' +const NOTIFY_CHANNEL = 'next_cache_tag_invalidation' +const L1_MAX_ENTRIES = 2000 +// How long the in-memory tag manifest is trusted before being re-read from PG. +// Bounds cross-replica staleness when LISTEN/NOTIFY is unavailable (e.g. behind +// a transaction-mode pooler); NOTIFY keeps it instant when a direct conn works. +const MANIFEST_TTL_MS = 1000 +// Postgres server clock in epoch milliseconds — one clock for all replicas. +const NOW_MS = `(EXTRACT(EPOCH FROM clock_timestamp()) * 1000)::bigint` + +const DEBUG = process.env.CACHE_HANDLER_DEBUG === 'true' +function debug(...args) { + if (DEBUG) console.log('[pg-cache-handler]', ...args) +} + +// ── Module-level shared state (one set per process) ───────────────────────── +let pool = null +let initPromise = null +const l1 = new Map() // key -> { value, tags, lastModified } +const tagManifest = new Map() // tag -> invalidatedAt (epoch ms) +let manifestLoadedAt = 0 +let manifestRefreshInFlight = null + +function getPool() { + if (!pool) { + pool = new Pool({ connectionString: process.env.DATABASE_URL, max: 4 }) + pool.on('error', (e) => console.warn('[pg-cache-handler] pool error:', e.message)) + } + return pool +} + +async function ensureInit() { + if (initPromise) return initPromise + initPromise = (async () => { + try { + await getPool().query(` + CREATE TABLE IF NOT EXISTS ${ENTRY_TABLE} ( + key TEXT PRIMARY KEY, + value JSONB NOT NULL, + tags TEXT[] NOT NULL DEFAULT '{}', + last_modified BIGINT NOT NULL + ); + CREATE TABLE IF NOT EXISTS ${TAG_TABLE} ( + tag TEXT PRIMARY KEY, + invalidated_at BIGINT NOT NULL DEFAULT 0 + ); + `) + } catch (e) { + // CREATE TABLE IF NOT EXISTS can race on pg_type when replicas boot + // together; the table still ends up created. + if (e.code !== '23505' && e.code !== '42P07') throw e + } + await refreshManifest() + startListener() // best-effort; not required for correctness + })() + return initPromise +} + +async function refreshManifest() { + if (manifestRefreshInFlight) return manifestRefreshInFlight + manifestRefreshInFlight = (async () => { + try { + const { rows } = await getPool().query(`SELECT tag, invalidated_at FROM ${TAG_TABLE}`) + tagManifest.clear() + for (const r of rows) tagManifest.set(r.tag, Number(r.invalidated_at)) + manifestLoadedAt = Date.now() + debug('manifest refreshed,', tagManifest.size, 'tags') + } catch (e) { + console.warn('[pg-cache-handler] manifest refresh failed:', e.message) + } finally { + manifestRefreshInFlight = null + } + })() + return manifestRefreshInFlight +} + +// Re-read the manifest from PG if our copy is older than the TTL. PG is the +// source of truth; the in-memory copy is only a short-lived cache. +async function ensureManifestFresh() { + if (Date.now() - manifestLoadedAt > MANIFEST_TTL_MS) { + await refreshManifest() + } +} + +// Optional instant-invalidation path. Uses DIRECT_URL when present because +// LISTEN needs a persistent session (transaction-mode poolers drop it). +function startListener() { + const conn = process.env.DIRECT_URL || process.env.DATABASE_URL + const connect = () => { + let client + try { + client = new Client({ connectionString: conn }) + } catch { + return + } + client.on('error', (e) => { + debug('listener error, reconnecting:', e.message) + try { client.end() } catch {} + setTimeout(connect, 2000) + }) + client.on('notification', (msg) => { + try { + const { tag, at } = JSON.parse(msg.payload) + const prev = tagManifest.get(tag) || 0 + if (at > prev) tagManifest.set(tag, Number(at)) + debug('NOTIFY', tag, at) + } catch {} + }) + client + .connect() + .then(() => client.query(`LISTEN ${NOTIFY_CHANNEL}`)) + .then(() => refreshManifest()) // reconcile after (re)connect — catch missed NOTIFYs + .then(() => debug('listening on', NOTIFY_CHANNEL)) + .catch((e) => { + debug('listener connect failed (falling back to TTL refresh):', e.message) + try { client.end() } catch {} + setTimeout(connect, 5000) + }) + } + connect() +} + +function l1Set(key, entry) { + l1.delete(key) + l1.set(key, entry) + if (l1.size > L1_MAX_ENTRIES) l1.delete(l1.keys().next().value) +} + +// True if an entry written at `lastModified` has had any of its tags invalidated +// since. PG-authoritative via the NOTIFY-warmed / TTL-refreshed manifest. +function isInvalidated(tags, lastModified) { + for (const tag of tags) { + const invalidatedAt = tagManifest.get(tag) + if (invalidatedAt && invalidatedAt > lastModified) return true + } + return false +} + +class PgCacheHandler { + constructor(ctx) { + // Tags revalidated within THIS request (Next-provided), for same-request + // read-your-writes consistency. + this.revalidatedTags = (ctx && ctx.revalidatedTags) || [] + } + + async get(key, ctx) { + await ensureInit() + await ensureManifestFresh() + + let entry = l1.get(key) + if (!entry) { + const { rows } = await getPool().query( + `SELECT value, tags, last_modified FROM ${ENTRY_TABLE} WHERE key = $1`, [key]) + if (!rows.length) return null + entry = { value: rows[0].value, tags: rows[0].tags || [], lastModified: Number(rows[0].last_modified) } + l1Set(key, entry) + } + + const ctxTags = ctx && ctx.kind === 'FETCH' ? [...(ctx.tags || []), ...(ctx.softTags || [])] : [] + const combinedTags = [...new Set([...(entry.tags || []), ...ctxTags])] + + if (combinedTags.some((t) => this.revalidatedTags.includes(t))) return null + if (isInvalidated(combinedTags, entry.lastModified)) return null + + return { value: entry.value, lastModified: entry.lastModified } + } + + async set(key, data, ctx) { + await ensureInit() + if (!data) { + l1.delete(key) + try { await getPool().query(`DELETE FROM ${ENTRY_TABLE} WHERE key = $1`, [key]) } catch (e) { debug('delete failed', e.message) } + return + } + const tags = (ctx && ctx.tags) || (data && data.tags) || [] + try { + // last_modified is stamped by the Postgres clock so it orders correctly + // against tag invalidation times regardless of which replica wrote it. + const { rows } = await getPool().query( + `INSERT INTO ${ENTRY_TABLE} (key, value, tags, last_modified) VALUES ($1, $2, $3, ${NOW_MS}) + ON CONFLICT (key) DO UPDATE SET value = $2, tags = $3, last_modified = ${NOW_MS} + RETURNING last_modified`, + [key, JSON.stringify(data), tags]) + l1Set(key, { value: data, tags, lastModified: Number(rows[0].last_modified) }) + } catch (e) { + // A non-serialisable value (e.g. a binary ISR payload, which PicImpact's + // all-dynamic routes don't produce) shouldn't crash the request path. + console.warn('[pg-cache-handler] set failed:', e.message) + } + } + + async revalidateTag(tags) { + await ensureInit() + tags = typeof tags === 'string' ? [tags] : tags + if (!tags || !tags.length) return + for (const tag of tags) { + try { + // invalidated_at is the Postgres clock; only ever moves forward. + const { rows } = await getPool().query( + `INSERT INTO ${TAG_TABLE} (tag, invalidated_at) VALUES ($1, ${NOW_MS}) + ON CONFLICT (tag) DO UPDATE SET invalidated_at = GREATEST(${TAG_TABLE}.invalidated_at, ${NOW_MS}) + RETURNING invalidated_at`, + [tag]) + const at = Number(rows[0].invalidated_at) + tagManifest.set(tag, at) + await getPool().query(`SELECT pg_notify($1, $2)`, [NOTIFY_CHANNEL, JSON.stringify({ tag, at })]) + } catch (e) { + console.warn('[pg-cache-handler] revalidateTag failed:', e.message) + } + } + } + + resetRequestCache() {} +} + +module.exports = PgCacheHandler From ce43d5e124394bbe8248fb087e4bfa7b1ad1aa70 Mon Sep 17 00:00:00 2001 From: Manjusaka Date: Mon, 22 Jun 2026 23:00:23 +0800 Subject: [PATCH 2/3] docs: address review notes on cache handler connection count + tag sweep MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - docs/multi-replica.md §2: account for the handler's own pool (max 4) + the persistent LISTEN client (~+5 connections per replica) in the max_connections budgeting, and cross-reference the DIRECT_URL requirement for LISTEN. - cache-cleanup.ts: note that only next_cache_entries is swept; next_cache_tags is a fixed small enum today, but flag that unbounded tags (e.g. per-image) would need their own sweep. Co-Authored-By: Claude Opus 4.8 --- docs/multi-replica.md | 11 +++++++---- server/lib/cache-cleanup.ts | 6 ++++++ 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/docs/multi-replica.md b/docs/multi-replica.md index 446e9976..1508a20b 100644 --- a/docs/multi-replica.md +++ b/docs/multi-replica.md @@ -60,10 +60,13 @@ pod) and set `PREPROCESS_TICKER_ENABLED=false` on the rest. ## 2. Database connections -Each replica opens its own Prisma connection pool. With N replicas the total -connection count is roughly `N × pool_size`. Make sure your PostgreSQL -`max_connections` covers that, or put a pooler such as PgBouncer (transaction -mode) in front of the database. +Each replica opens its own Prisma connection pool, plus the cache handler (§4) +adds its own small pool (max 4) and one persistent `LISTEN` client — roughly +`+5` connections per replica on top of Prisma. With N replicas the total is +about `N × (prisma_pool + 5)`. Make sure your PostgreSQL `max_connections` +covers that, or put a pooler such as PgBouncer (transaction mode) in front of +the database — note the handler's `LISTEN` connection must use a direct +(non-transaction-pooled) URL via `DIRECT_URL`, see §4. ## 3. Session revocation latency diff --git a/server/lib/cache-cleanup.ts b/server/lib/cache-cleanup.ts index 0b3bf246..0d957733 100644 --- a/server/lib/cache-cleanup.ts +++ b/server/lib/cache-cleanup.ts @@ -2,6 +2,12 @@ import 'server-only' import { db } from '~/server/lib/db' +// NOTE: this only sweeps `next_cache_entries`. The `next_cache_tags` table is +// intentionally not swept because the tag set is a small fixed enum +// (CACHE_TAG.gallery/albums/config in server/lib/cache.ts), so it can't grow. If +// unbounded tags are ever passed to `revalidateTag` (e.g. a per-image tag), +// next_cache_tags would need its own sweep — revisit this then. +// // Cached keys are deterministic (unstable_cache key + args), and active entries // are rewritten well within their TTL (gallery ≤60s, albums/config ≤1h), so the // entries table is naturally bounded by the number of distinct queries. This From 36db15935c516a19481e5691e19fa280a09457e6 Mon Sep 17 00:00:00 2001 From: Manjusaka Date: Mon, 22 Jun 2026 23:04:58 +0800 Subject: [PATCH 3/3] =?UTF-8?q?docs:=20correct=20multi-replica=20task-canc?= =?UTF-8?q?ellation=20section=20(=C2=A75)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cross-replica cancellation is already implemented, not a follow-up: the cancel endpoint sets status='cancelling' (atomic, cross-replica) and the running replica re-reads it at its per-image checkpoint and stops — within ~one item, not a lease window. Correct §5 which previously described it pessimistically. Co-Authored-By: Claude Opus 4.8 --- docs/multi-replica.md | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/docs/multi-replica.md b/docs/multi-replica.md index 1508a20b..ba6bdab4 100644 --- a/docs/multi-replica.md +++ b/docs/multi-replica.md @@ -112,8 +112,10 @@ endpoint (§1), so it does not run as a per-replica timer. ## 5. In-flight task cancellation -Cancelling a running preprocess/metadata task signals the replica that owns the -run. If the cancel request lands on a different replica, it is still recorded, and -the running replica stops at the next lease checkpoint (within the lease window). -For rare admin actions this lag is acceptable; a fully cross-replica cancel signal -is a planned follow-up. +Cancelling a running preprocess/metadata task is cross-replica safe. The cancel +endpoint flips the run's row to `status = 'cancelling'` in the database (an atomic +UPDATE visible to all replicas), and the replica actually running the task +re-reads that status at its per-image checkpoint and stops — so cancellation takes +effect within roughly one item's processing time, regardless of which replica +received the cancel request. (If the cancel lands on the same replica that's +running the task, an in-process abort also interrupts it immediately.)