From 0dfbceb04789a6ed68f258a54f1a5d4c7ad503f9 Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Wed, 2 Sep 2026 16:00:13 -0700 Subject: [PATCH] feat: durable event subscriptions store, cache, and routes (PUT-1673) (#3679) * feat: durable event subscriptions store, cache, and routes (PUT-1673) * fix: durable subscription hardening (PUT-1673) - Expired rows stop delivering at dispatch time and no longer count toward the per-account cap, instead of waiting for the sweep. - The expiry sweep runs hourly with a jittered first pass shortly after boot; a 24 h interval never fired on a fleet that redeploys more often than that. - Only a durable generation bump marks peer regions cold. A session subscribe/unsubscribe in one region used to force a primary read in every other region on its next dispatch. - `subject`/`anchor_path` widen to varchar(4096) to match `fsentries.path`, and subjects longer than that are refused with `invalid_subject` rather than failing the insert on MySQL/Postgres. - The dispatch and durable integration suites wait for the specific delivery they expect and assert only within their own folder; the old any-delivery `settle()` let a late event from a previous test satisfy or pollute the next one under CI load. --- .../database/SqliteDatabaseClient.test.ts | 2 +- .../clients/database/SqliteDatabaseClient.ts | 1 + .../migrations/mysql/mysql_mig_29.sql | 59 ++ .../migrations/postgres/postgres_mig_18.sql | 53 ++ .../sqlite/0075_event-subscriptions.sql | 90 +++ src/backend/clients/event/types.ts | 2 + .../controllers/events/EventsController.ts | 121 ++++ src/backend/controllers/events/limits.ts | 16 + src/backend/controllers/index.ts | 2 + .../services/events/EventsService.test.ts | 199 +++++- src/backend/services/events/EventsService.ts | 548 +++++++++++++-- .../events/dispatch.integration.test.ts | 52 +- .../events/durable.integration.test.ts | 636 ++++++++++++++++++ src/backend/services/events/subjects.test.ts | 21 + src/backend/services/events/subjects.ts | 4 + .../services/events/subscriptionCache.test.ts | 62 +- .../services/events/subscriptionCache.ts | 102 ++- ...rableSubscriptionStore.integration.test.ts | 421 ++++++++++++ .../stores/events/DurableSubscriptionStore.ts | 460 +++++++++++++ .../stores/events/EventSubscriptionStore.ts | 232 +++++-- src/backend/stores/events/types.ts | 94 +++ src/backend/stores/index.ts | 4 + src/docs/src/rate-limits-and-quotas.md | 8 +- 23 files changed, 3008 insertions(+), 181 deletions(-) create mode 100644 src/backend/clients/database/migrations/mysql/mysql_mig_29.sql create mode 100644 src/backend/clients/database/migrations/postgres/postgres_mig_18.sql create mode 100644 src/backend/clients/database/migrations/sqlite/0075_event-subscriptions.sql create mode 100644 src/backend/controllers/events/EventsController.ts create mode 100644 src/backend/services/events/durable.integration.test.ts create mode 100644 src/backend/stores/events/DurableSubscriptionStore.integration.test.ts create mode 100644 src/backend/stores/events/DurableSubscriptionStore.ts create mode 100644 src/backend/stores/events/types.ts diff --git a/src/backend/clients/database/SqliteDatabaseClient.test.ts b/src/backend/clients/database/SqliteDatabaseClient.test.ts index 0cb4d5461..f5e0523ff 100644 --- a/src/backend/clients/database/SqliteDatabaseClient.test.ts +++ b/src/backend/clients/database/SqliteDatabaseClient.test.ts @@ -27,7 +27,7 @@ import { DatabaseClientFactory } from './index.js'; import { SqliteDatabaseClient } from './SqliteDatabaseClient.js'; /** Highest schema version the migration table can reach. */ -const CURRENT_SCHEMA_VERSION = 70; +const CURRENT_SCHEMA_VERSION = 71; /** * These suites migrate real files on disk. Idle they finish in well under a diff --git a/src/backend/clients/database/SqliteDatabaseClient.ts b/src/backend/clients/database/SqliteDatabaseClient.ts index 31dc1f362..067638b02 100644 --- a/src/backend/clients/database/SqliteDatabaseClient.ts +++ b/src/backend/clients/database/SqliteDatabaseClient.ts @@ -104,6 +104,7 @@ const AVAILABLE_MIGRATIONS: [number, string[]][] = [ [67, ['0072_notification-scope.sql']], [68, ['0073_notification-created-at.sql']], [69, ['0074_add_user_home.sql']], + [70, ['0075_event-subscriptions.sql']], ]; export class SqliteDatabaseClient extends AbstractDatabaseClient { diff --git a/src/backend/clients/database/migrations/mysql/mysql_mig_29.sql b/src/backend/clients/database/migrations/mysql/mysql_mig_29.sql new file mode 100644 index 000000000..a2dac0936 --- /dev/null +++ b/src/backend/clients/database/migrations/mysql/mysql_mig_29.sql @@ -0,0 +1,59 @@ +-- 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 . + +-- Subscriptions that outlive the connection that made them. See +-- sqlite/0075_event-subscriptions.sql for the column rationale. +-- +-- Idempotent: `CREATE TABLE IF NOT EXISTS` with the indexes declared inline, +-- as mig_23. There is no per-file applied-state tracking, so a replay has to +-- be a no-op. + +CREATE TABLE IF NOT EXISTS `event_subscriptions` ( + `id` bigint unsigned NOT NULL AUTO_INCREMENT, + `sub_id` varchar(80) NOT NULL, + `token` varchar(255) NOT NULL, + `owner_user_id` int unsigned NOT NULL, + `holder_user_id` int unsigned NOT NULL, + -- Matches `apps`.`uid` exactly, charset included, so an equality against + -- one never falls back to a conversion. No foreign key: a row outlives the + -- app that made it, which is what keeps it revocable afterwards. + `app_uid` char(40) CHARACTER SET latin1 COLLATE latin1_swedish_ci + DEFAULT NULL, + `subject` varchar(4096) NOT NULL, + `anchor_uid` char(36) NOT NULL, + `anchor_path` varchar(4096) NOT NULL, + `match` varchar(1024) DEFAULT NULL, + `delivery` varchar(16) NOT NULL, + `ops` varchar(64) DEFAULT NULL, + `handler_name` varchar(128) DEFAULT NULL, + `targets` json NOT NULL, + `context` text, + `permission` varchar(32) NOT NULL, + `suspended_at` bigint DEFAULT NULL, + `suspended_reason` varchar(64) DEFAULT NULL, + `expires_at` bigint DEFAULT NULL, + `created_at` bigint NOT NULL, + PRIMARY KEY (`id`), + UNIQUE KEY `idx_event_subscriptions_sub_id` (`sub_id`), + KEY `idx_event_subscriptions_token` (`token`), + KEY `idx_event_subscriptions_holder` (`holder_user_id`, `app_uid`), + KEY `idx_event_subscriptions_owner` (`owner_user_id`), + CONSTRAINT `fk_event_subscriptions_owner` FOREIGN KEY (`owner_user_id`) + REFERENCES `user` (`id`) ON DELETE CASCADE ON UPDATE CASCADE, + CONSTRAINT `fk_event_subscriptions_holder` FOREIGN KEY (`holder_user_id`) + REFERENCES `user` (`id`) ON DELETE CASCADE ON UPDATE CASCADE +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; diff --git a/src/backend/clients/database/migrations/postgres/postgres_mig_18.sql b/src/backend/clients/database/migrations/postgres/postgres_mig_18.sql new file mode 100644 index 000000000..2f5e66ebb --- /dev/null +++ b/src/backend/clients/database/migrations/postgres/postgres_mig_18.sql @@ -0,0 +1,53 @@ +-- 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 . + +-- Subscriptions that outlive the connection that made them. See +-- sqlite/0075_event-subscriptions.sql for the column rationale. +-- +-- Idempotent via IF NOT EXISTS. + +CREATE TABLE IF NOT EXISTS event_subscriptions ( + id BIGSERIAL PRIMARY KEY, + sub_id VARCHAR(80) NOT NULL UNIQUE, + token VARCHAR(255) NOT NULL, + owner_user_id INTEGER NOT NULL + REFERENCES "user" (id) ON DELETE CASCADE ON UPDATE CASCADE, + holder_user_id INTEGER NOT NULL + REFERENCES "user" (id) ON DELETE CASCADE ON UPDATE CASCADE, + app_uid VARCHAR(40) DEFAULT NULL, + subject VARCHAR(4096) NOT NULL, + anchor_uid VARCHAR(36) NOT NULL, + anchor_path VARCHAR(4096) NOT NULL, + "match" VARCHAR(1024) DEFAULT NULL, + delivery VARCHAR(16) NOT NULL, + ops VARCHAR(64) DEFAULT NULL, + handler_name VARCHAR(128) DEFAULT NULL, + targets JSONB NOT NULL, + context TEXT DEFAULT NULL, + permission VARCHAR(32) NOT NULL, + suspended_at BIGINT DEFAULT NULL, + suspended_reason VARCHAR(64) DEFAULT NULL, + expires_at BIGINT DEFAULT NULL, + created_at BIGINT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_event_subscriptions_token + ON event_subscriptions (token); +CREATE INDEX IF NOT EXISTS idx_event_subscriptions_holder + ON event_subscriptions (holder_user_id, app_uid); +CREATE INDEX IF NOT EXISTS idx_event_subscriptions_owner + ON event_subscriptions (owner_user_id); diff --git a/src/backend/clients/database/migrations/sqlite/0075_event-subscriptions.sql b/src/backend/clients/database/migrations/sqlite/0075_event-subscriptions.sql new file mode 100644 index 000000000..aeb50b36d --- /dev/null +++ b/src/backend/clients/database/migrations/sqlite/0075_event-subscriptions.sql @@ -0,0 +1,90 @@ +-- 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 . + +-- Subscriptions that outlive the connection that made them. Session +-- subscriptions never reach this table — they are Redis only — so every row +-- here is one somebody has to be able to find and revoke later. +-- +-- - `sub_id` : `#`, or `user#` for a row a user +-- session created. Unique table-wide; it is what +-- `unsubscribe` names. +-- - `owner_user_id` : owner of the anchor node. Dispatch only knows whose +-- resource changed, so this is the side a shared-folder +-- subscription has to be findable from, and the key a +-- region's cache is rebuilt under. +-- - `holder_user_id` : who subscribed — the delivery target, whose access is +-- re-checked, and who the quota counts against. +-- - `app_uid` : NULL for a row a user session created. No foreign +-- key: a row outlives the app that made it, which is +-- what keeps it listable and revocable afterwards. +-- - `anchor_*` : the resolved anchor. `match` is a glob relative to +-- `anchor_path`, so the path is not decoration — the +-- dispatch filter reads it on every event. +-- - `ops` : op filter; a comma-separated set, NULL for every op. +-- - `targets` : JSON array over socket|worker|push. +-- - `context` : plaintext JSON, hard-capped at 4 KB, read only on the +-- delivery path and never returned by `list`. +-- - `permission` : the ACL mode the subscribe check passed under, +-- re-checked per delivery. +-- - `suspended_*` : set when a subscription stops delivering without +-- being removed; the state machine that drives them +-- lands with the delivery classes that need it. +-- +-- `created_at` and the two timestamps are unix seconds, matching +-- `app_feedback` and `user_block`. + +CREATE TABLE IF NOT EXISTS `event_subscriptions` ( + "id" INTEGER PRIMARY KEY AUTOINCREMENT, + "sub_id" TEXT NOT NULL, + "token" TEXT NOT NULL, + "owner_user_id" INTEGER NOT NULL + REFERENCES `user` ("id") ON DELETE CASCADE ON UPDATE CASCADE, + "holder_user_id" INTEGER NOT NULL + REFERENCES `user` ("id") ON DELETE CASCADE ON UPDATE CASCADE, + "app_uid" TEXT DEFAULT NULL, + "subject" TEXT NOT NULL, + "anchor_uid" TEXT NOT NULL, + "anchor_path" TEXT NOT NULL, + "match" TEXT DEFAULT NULL, + "delivery" TEXT NOT NULL, + "ops" TEXT DEFAULT NULL, + "handler_name" TEXT DEFAULT NULL, + "targets" TEXT NOT NULL, + "context" TEXT DEFAULT NULL, + "permission" TEXT NOT NULL, + "suspended_at" INTEGER DEFAULT NULL, + "suspended_reason" TEXT DEFAULT NULL, + "expires_at" INTEGER DEFAULT NULL, + "created_at" INTEGER NOT NULL -- unix seconds +); + +CREATE UNIQUE INDEX IF NOT EXISTS `idx_event_subscriptions_sub_id` + ON `event_subscriptions` (`sub_id`); + +-- Dispatch reads the table only when a region's cache is cold; this is the +-- lookup it falls back to. +CREATE INDEX IF NOT EXISTS `idx_event_subscriptions_token` + ON `event_subscriptions` (`token`); + +-- List, revoke and quota all key here — the index is the scope check rather +-- than a filter over a wider read. +CREATE INDEX IF NOT EXISTS `idx_event_subscriptions_holder` + ON `event_subscriptions` (`holder_user_id`, `app_uid`); + +-- Rebuilding one region's cache for one owner, and the sweep that follows it. +CREATE INDEX IF NOT EXISTS `idx_event_subscriptions_owner` + ON `event_subscriptions` (`owner_user_id`); diff --git a/src/backend/clients/event/types.ts b/src/backend/clients/event/types.ts index d206975a8..9f0540f26 100644 --- a/src/backend/clients/event/types.ts +++ b/src/backend/clients/event/types.ts @@ -447,6 +447,8 @@ export type EventMap = { 'outer.events.generationBumped': { userId: number; generation: number; + /** Whether the table changed; only then does a peer need to re-read it. */ + durable: boolean; }; 'outer.fs.write-hash': { hash: string; uuid: string }; /** diff --git a/src/backend/controllers/events/EventsController.ts b/src/backend/controllers/events/EventsController.ts new file mode 100644 index 000000000..7c80ce7ac --- /dev/null +++ b/src/backend/controllers/events/EventsController.ts @@ -0,0 +1,121 @@ +/* + * 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 type { Request, Response } from 'express'; +import type { Actor } from '../../core/actor.js'; +import { Controller, Get, Post } from '../../core/http/decorators.js'; +import { HttpError } from '../../core/http/HttpError.js'; +import { DURABLE_LIST_LIMIT_CAP } from '../../stores/events/DurableSubscriptionStore.js'; +import { normalizeLimit } from '../../util/pagination.js'; +import { PuterController } from '../types.js'; +import { EVENTS_LIST_LIMIT } from './limits.js'; + +/** + * The durable half of the events surface. Session subscriptions arrive over the + * socket that holds them and are not routable; these are rows that outlive + * every connection, so they need somewhere to be created, listed and revoked + * from without one. + * + * Gates are `subdomain: 'api'` (verb routes are root-origin by default) plus + * `allowAccessToken`, because API tokens are in the scoping matrix and access + * tokens are refused on authenticated routes otherwise. Deliberately **not** + * `allowedAppIds`: it does not restrict a route to app actors, and using it as + * though it did is the mis-gating this surface has to avoid. Scope is decided + * from `effectiveApp` inside the service, over the index the rows are stored + * under. + */ +@Controller('/events') +export class EventsController extends PuterController { + /** + * POST /events/subscribe — register a subscription that outlives the + * caller. + */ + @Post('/subscribe', { + subdomain: 'api', + requireAuth: true, + allowAccessToken: true, + }) + async subscribe(req: Request, res: Response): Promise { + const actor = this.#requireActor(req); + const { sub } = await this.services.events.subscribeDurable( + actor, + this.#body(req), + ); + res.json(sub); + } + + /** GET /events/subscriptions — what the caller holds, one page at a time. */ + @Get('/subscriptions', { + subdomain: 'api', + requireAuth: true, + allowAccessToken: true, + rateLimit: EVENTS_LIST_LIMIT, + }) + async list(req: Request, res: Response): Promise { + const actor = this.#requireActor(req); + const query = (req.query ?? {}) as Record; + + const page = await this.services.events.listDurable(actor, { + limit: normalizeLimit(query.limit, { cap: DURABLE_LIST_LIMIT_CAP }), + cursor: typeof query.cursor === 'string' ? query.cursor : undefined, + includeTotal: query.includeTotal === 'true', + }); + + res.json({ + items: page.items, + ...(page.cursor ? { cursor: page.cursor } : {}), + ...(page.total !== undefined ? { total: page.total } : {}), + }); + } + + /** + * POST /events/unsubscribe — an id the caller does not hold reads as + * absent. + */ + @Post('/unsubscribe', { + subdomain: 'api', + requireAuth: true, + allowAccessToken: true, + }) + async unsubscribe(req: Request, res: Response): Promise { + const actor = this.#requireActor(req); + await this.services.events.unsubscribeDurable(actor, this.#body(req)); + res.json({}); + } + + // -- Internals --------------------------------------------------- + + #requireActor(req: Request): Actor { + const actor = req.actor; + if (!actor?.user) + throw new HttpError(401, 'Unauthorized', { + legacyCode: 'unauthorized', + }); + return actor; + } + + #body(req: Request): Record { + const body = req.body; + if (!body || typeof body !== 'object' || Array.isArray(body)) + throw new HttpError(400, 'body must be an object', { + legacyCode: 'bad_request', + }); + return body as Record; + } +} diff --git a/src/backend/controllers/events/limits.ts b/src/backend/controllers/events/limits.ts index d600b632a..c4b170d86 100644 --- a/src/backend/controllers/events/limits.ts +++ b/src/backend/controllers/events/limits.ts @@ -50,6 +50,16 @@ const userWindow = ( */ export const EVENTS_SESSION_SUBSCRIPTIONS_PER_SOCKET = 50; +/** + * Durable subscriptions one account may hold, across every app. + * + * These are table rows that keep costing after the client that made them is + * gone — a delivery each time their anchor changes, and a cache entry in every + * region that sees a write. Counted over the holder index. Per-plan tiering + * arrives with the metering that prices them. + */ +export const EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER = 500; + /** * Subscribe + unsubscribe calls per minute, per user. * @@ -59,6 +69,12 @@ export const EVENTS_SESSION_SUBSCRIPTIONS_PER_SOCKET = 50; */ export const EVENTS_SUBSCRIBE_LIMIT = userWindow('events:subscribe', 60); +/** + * Subscription listings per minute, per user. Reads an index and returns one + * page, so it is budgeted well above the verbs that write. + */ +export const EVENTS_LIST_LIMIT = userWindow('events:list', 120); + // -- Dispatch fan-out ------------------------------------------------ /** diff --git a/src/backend/controllers/index.ts b/src/backend/controllers/index.ts index cda804451..2b6908807 100644 --- a/src/backend/controllers/index.ts +++ b/src/backend/controllers/index.ts @@ -23,6 +23,7 @@ import { AuthController } from './auth/AuthController.js'; import { BroadcastController } from './broadcast/BroadcastController.js'; import { DesktopController } from './desktop/DesktopController.js'; import { DriverController } from './drivers/DriverController.js'; +import { EventsController } from './events/EventsController.js'; import { FSController } from './fs/FSController.js'; import { HomepageController } from './homepage/HomepageController.js'; import { HostingController } from './hosting/HostingController.js'; @@ -54,6 +55,7 @@ export const puterControllers = { drivers: DriverController, broadcast: BroadcastController, notification: NotificationController, + events: EventsController, share: ShareController, webdav: WebDAVController, oidc: OIDCController, diff --git a/src/backend/services/events/EventsService.test.ts b/src/backend/services/events/EventsService.test.ts index 62ddcb2e8..f42c4e9c6 100644 --- a/src/backend/services/events/EventsService.test.ts +++ b/src/backend/services/events/EventsService.test.ts @@ -27,7 +27,10 @@ import { } from '../../controllers/events/limits.js'; import type { Actor } from '../../core/actor.js'; import { isHttpError } from '../../core/http/HttpError.js'; -import { EventSubscriptionStore } from '../../stores/events/EventSubscriptionStore.js'; +import { + EventSubscriptionStore, + type DurableSubscription, +} from '../../stores/events/EventSubscriptionStore.js'; import type { FSEntry } from '../../stores/fs/FSEntry.js'; import type { IConfig } from '../../types.js'; import { @@ -39,6 +42,7 @@ import { type EventSocket, } from './EventsService.js'; import { FILTER_EVALUATIONS_PER_EVENT } from './matcher.js'; +import { SUBSCRIPTION_CACHE_TTL_MS } from './subscriptionCache.js'; /** * The hot path is a cost claim before it is a behaviour claim, so the Redis @@ -172,6 +176,15 @@ const appStore = { getByUid: async (uid: string) => ({ uid, id: 1 }), }; +/** + * This suite is about session rows and what a dispatch spends on them, so the + * region is already warm and holds no durable rows — the table and its cache + * are covered against a real database in the durable suites. + */ +const durableSubscriptionStore = { + warmRegion: async () => false, +}; + /** * Each service gets its own outbox. A delivery still in flight when a test * ends must land in that test's record, not in the next one's. @@ -195,6 +208,7 @@ const buildService = ( } as never, { eventSubscription: store, + durableSubscription: durableSubscriptionStore, fsEntry: fsEntryStore, user: userStore, app: appStore, @@ -578,6 +592,147 @@ describe('cross-process invalidation', () => { expect(commands).toEqual([]); }); + + it('re-reads the table for a peer`s durable bump, but not for a session one', async () => { + const cold = vi.spyOn(store, 'markRegionCold'); + const handler = remoteGenerationBumpHandler(); + + handler?.( + 'outer.events.generationBumped', + { userId, generation: 1, durable: false }, + { from_outside: true }, + ); + expect(cold).not.toHaveBeenCalled(); + + handler?.( + 'outer.events.generationBumped', + { userId, generation: 2, durable: true }, + { from_outside: true }, + ); + expect(cold).toHaveBeenCalledWith(userId); + }); + + it('invalidates on a remote bump regardless of the number it carries', async () => { + vi.useFakeTimers(); + // `ev:g` is a region-local INCR: a peer's own counter can legitimately + // sit behind whatever this process has already recorded (it is + // counting something else entirely) — comparing the two would let a + // real remote invalidation be ignored as "already applied". So a + // `from_outside` bump forgets unconditionally instead; there is + // nothing to order it against. Push this process's own recorded + // generation ahead first, via genuine local activity, so a + // number-comparing implementation would wrongly ignore the bump below. + const { documents, file } = seedTree(); + for (let i = 0; i < 3; i++) { + const sub = await subscribe(`fs:${documents.uid}`, `local-${i}`); + await service.unsubscribe(actorFor(), `local-${i}`, { + subId: sub.subId, + }); + } + + await dispatch(file); // caches "nothing subscribed" at the higher generation + + await store.add({ + subId: 'behind-sub', + socketId: 'behind-socket', + holderUserId: userId, + ownerUserId: userId, + subject: `fs:${documents.uid}`, + token: `f#${documents.uid}`, + anchorUid: documents.uid, + anchorPath: documents.path, + match: null, + op: null, + appUid: null, + permission: 'list', + }); + + remoteGenerationBumpHandler()?.( + 'outer.events.generationBumped', + { userId, generation: 1 }, // behind this process's own recorded generation + { from_outside: true }, + ); + + await dispatch(file); + await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1); + + expect(sent).toHaveLength(1); + }); + + it('self-heals on the TTL alone when no bump ever arrives', async () => { + vi.useFakeTimers(); + // The backstop for a broadcast that never lands at all: nothing + // invalidates this process's cache, so only the read-side TTL can + // force it to look again. + const { documents, file } = seedTree(); + await dispatch(file); // warms this process's cache to "nothing subscribed" + + await store.add({ + subId: 'unseen-sub', + socketId: 'unseen-socket', + holderUserId: userId, + ownerUserId: userId, + subject: `fs:${documents.uid}`, + token: `f#${documents.uid}`, + anchorUid: documents.uid, + anchorPath: documents.path, + match: null, + op: null, + appUid: null, + permission: 'list', + }); + + await dispatch(file); + expect(sent).toEqual([]); // no bump landed, so still cached stale + + vi.advanceTimersByTime(SUBSCRIPTION_CACHE_TTL_MS + 1); + await dispatch(file); + await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1); + + expect(sent).toHaveLength(1); + }); +}); + +describe('cold-region rebuild concurrency', () => { + it('collapses concurrent misses on a cold region into one rebuild', async () => { + const { file } = seedTree(); + let calls = 0; + const warm = vi + .spyOn(durableSubscriptionStore, 'warmRegion') + .mockImplementation(async () => { + calls++; + await Promise.resolve(); + return false; + }); + try { + await Promise.all([dispatch(file), dispatch(file)]); + expect(calls).toBe(1); + } finally { + warm.mockRestore(); + } + }); + + it('does not wedge future dispatches when a rebuild throws', async () => { + vi.useFakeTimers(); + const { documents, file } = seedTree(); + const warm = vi + .spyOn(durableSubscriptionStore, 'warmRegion') + .mockRejectedValueOnce(new Error('boom')); + try { + // Not being able to tell must resolve as "nothing subscribed" + // rather than hang, and must not leave the in-flight lookup + // wedged for every dispatch after it. + await expect(dispatch(file)).resolves.toBeUndefined(); + + await subscribe(`fs:${documents.uid}`); + await dispatch(file); + await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1); + + expect(sent).toHaveLength(1); + } finally { + warm.mockRestore(); + } + }); }); // -- Matching -------------------------------------------------------- @@ -618,6 +773,48 @@ describe('matching', () => { expect(sent).toEqual([]); }); + it('stops delivering to a durable row the moment it expires', async () => { + const { documents, file } = seedTree(); + const now = Math.floor(Date.now() / 1000); + const row = (over: Partial): DurableSubscription => ({ + subId: `durable-${seq}-${over.expiresAt}`, + token: `f#${documents.uid}`, + ownerUserId: userId, + holderUserId: userId, + subject: `fs:${documents.uid}`, + anchorUid: documents.uid, + anchorPath: documents.path, + match: null, + op: null, + appUid: null, + permission: 'list', + durable: true, + delivery: 'broadcast', + targets: ['socket'], + handlerName: null, + context: null, + expiresAt: null, + suspendedAt: null, + suspendedReason: null, + createdAt: now, + ...over, + }); + // Straight into the region cache, as a cold rebuild would leave them: + // one still good for an hour, one that lapsed a minute ago and has not + // been swept yet. + await store.rebuildDurable(userId, [ + row({ expiresAt: now + 3600 }), + row({ expiresAt: now - 60 }), + ]); + + await dispatch(file); + await flush(); + + expect(sent.map((s) => s.envelope.subId)).toEqual([ + `durable-${seq}-${now + 3600}`, + ]); + }); + it('drops an event the match filter excludes', async () => { const { documents } = seedTree(); await subscribe(`fs:/u${userId}/Documents/reports/*.csv`); diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index ec1d474de..750ea94ef 100644 --- a/src/backend/services/events/EventsService.ts +++ b/src/backend/services/events/EventsService.ts @@ -30,12 +30,23 @@ import { HttpError } from '../../core/http/HttpError.js'; import { checkRateLimit } from '../../core/http/middleware/rateLimit.js'; import { SESSION_SUBSCRIPTION_TTL_SECONDS, + type DispatchSubscription, + type DurableSubscription, type GenerationBump, type SessionSubscription, } from '../../stores/events/EventSubscriptionStore.js'; +import { + isSubscriptionTarget, + type SubscriptionTarget, +} from '../../stores/events/types.js'; import type { FSEntry } from '../../stores/fs/FSEntry.js'; -import type { ResourceDescriptor } from '../acl/ACLService.js'; +import type { PageResult } from '../../util/pagination.js'; +import type { AclMode, ResourceDescriptor } from '../acl/ACLService.js'; import { resolveNode } from '../fs/resolveNode.js'; +import { + appSocketRoom, + type SocketSpecifier, +} from '../socket/SocketService.js'; import { PuterService } from '../types.js'; import { resolveFsAnchor, type FsAnchorDeps } from './anchors.js'; import { @@ -55,6 +66,7 @@ import { } from './matcher.js'; import { lookupPublicSubject, + type DeliveryClass, type EventContext, type ProjectedEvent, type PublicSubject, @@ -94,6 +106,21 @@ export interface UnsubscribeRequest { subId?: unknown; } +/** Body of `POST /events/subscribe`. */ +export interface DurableSubscribeRequest extends SubscribeRequest { + delivery?: unknown; + targets?: unknown; + handlerName?: unknown; + context?: unknown; + expiresAt?: unknown; +} + +export interface DurableListRequest { + limit?: number; + cursor?: string; + includeTotal?: boolean; +} + /** What a client gets back for a subscription it just made. */ export interface SubscriptionView { subId: string; @@ -103,6 +130,22 @@ export interface SubscriptionView { op: FsOp | null; } +/** + * A durable row as its holder sees it. `context` is deliberately absent: it is + * read on the delivery path and nowhere else, and a listing is the one surface + * an app can call repeatedly. + */ +export interface DurableSubscriptionView extends SubscriptionView { + delivery: DeliveryClass; + targets: SubscriptionTarget[]; + handlerName: string | null; + appUid: string | null; + createdAt: number; + expiresAt: number | null; + suspendedAt: number | null; + suspendedReason: string | null; +} + export type VerbAck = | ({ ok: true } & T) | { ok: false; error: { code: string; message: string } }; @@ -132,9 +175,9 @@ export interface DeliveryEnvelope { event: ProjectedEvent | GapMarker; } -/** The envelope plus where it goes. The socket id is not part of the wire. */ +/** The envelope plus where it goes. The address is not part of the wire. */ interface AddressedDelivery { - socketId: string; + target: SocketSpecifier; envelope: DeliveryEnvelope; } @@ -157,6 +200,18 @@ export const EVENTS_SUBSCRIBE_VERB = 'events.subscribe'; export const EVENTS_UNSUBSCRIBE_VERB = 'events.unsubscribe'; export const EVENTS_DELIVERY_CHANNEL = 'events.delivery'; +// -- Expiry sweep ----------------------------------------------------- + +/** An expiry is a date, not a deadline, so hourly is close enough. */ +const EXPIRY_SWEEP_INTERVAL_MS = 60 * 60 * 1000; +// The first pass lands within this long of boot, so a fleet that redeploys more +// often than the interval still sweeps. +const EXPIRY_SWEEP_INITIAL_DELAY_MS = 5 * 60 * 1000; +/** Rows one delete takes. Small enough not to hold a lock anyone waits on. */ +const EXPIRY_BATCH_SIZE = 500; +/** Batches one sweep takes, so a large backlog drains over several passes. */ +const EXPIRY_MAX_BATCHES = 50; + /** The part of a socket this service uses, so tests need not build one. */ export interface EventSocket { id: string; @@ -181,6 +236,15 @@ const tooManyCalls = (): HttpError => legacyCode: 'too_many_requests', }); +/** Stands until there is a pending-delivery store to take a `single` lease. */ +const deliveryClassUnavailable = (): HttpError => + new HttpError(501, 'Delivery class `single` is not available yet', { + legacyCode: 'delivery_class_unavailable', + }); + +const badRequest = (message: string, code: string): HttpError => + new HttpError(400, message, { legacyCode: code }); + const errorAck = (err: unknown): VerbAck => { if (err instanceof HttpError) return { @@ -196,7 +260,7 @@ const errorAck = (err: unknown): VerbAck => { }; }; -const toView = (sub: SessionSubscription): SubscriptionView => ({ +const toView = (sub: DispatchSubscription): SubscriptionView => ({ subId: sub.subId, subject: sub.subject, anchor: { uid: sub.anchorUid, path: sub.anchorPath }, @@ -204,15 +268,132 @@ const toView = (sub: SessionSubscription): SubscriptionView => ({ op: sub.op, }); +const toDurableView = (sub: DurableSubscription): DurableSubscriptionView => ({ + ...toView(sub), + delivery: sub.delivery, + targets: sub.targets, + handlerName: sub.handlerName, + appUid: sub.appUid, + createdAt: sub.createdAt, + expiresAt: sub.expiresAt, + suspendedAt: sub.suspendedAt, + suspendedReason: sub.suspendedReason, +}); + /** Coalescing is per (subscription, subject), which is what the key says. */ const coalesceKey = (subId: string, subject: string): string => `${subId}|${subject}`; +/** + * Where one row's deliveries go. A session row is addressed at the connection + * that made it. A durable row has no connection to name, so it is addressed at + * a room: the app's own room for a row an app created, and the holder's user + * room for one their session created — which is every desktop tab they have + * open, and the only handle that reaches an account rather than a connection. + */ +const deliveryTarget = (row: DispatchSubscription): SocketSpecifier => { + if (row.socketId !== undefined) return { socket: row.socketId }; + return { + room: row.appUid + ? appSocketRoom(row.holderUserId, row.appUid) + : String(row.holderUserId), + }; +}; + +/** + * Rows this pass can actually deliver. Durable `single` rows need the pending + * store to take a lease, and a row with no socket target has asked not to be + * delivered over one. + */ +const nowSeconds = (): number => Math.floor(Date.now() / 1000); + +/** Past `expiresAt` a durable row is finished, sweep or no sweep. */ +const unexpired = (row: DispatchSubscription): boolean => { + if (row.durable !== true) return true; + const { expiresAt } = row as DurableSubscription; + return expiresAt === null || expiresAt > nowSeconds(); +}; + +const deliverableOverSockets = (row: DispatchSubscription): boolean => + unexpired(row) && + (row.durable !== true || + (row.delivery === 'broadcast' && + (row.targets ?? []).includes('socket'))); + +// -- Durable request parsing ------------------------------------------ + +/** Transports a durable row takes unless the caller says otherwise. */ +const DEFAULT_DURABLE_TARGETS: SubscriptionTarget[] = ['socket', 'worker']; + +/** Longest a `handlerName` may be, matching the column that holds it. */ +const HANDLER_NAME_MAX_LENGTH = 128; + +const parseDelivery = (value: unknown): DeliveryClass => { + if (value === undefined || value === null || value === 'broadcast') + return 'broadcast'; + // Creatable but inert is worse than refused: a `single` row would take a + // lease nothing in this build can settle. + if (value === 'single') throw deliveryClassUnavailable(); + throw badRequest(`Unknown delivery class: ${String(value)}`, 'bad_request'); +}; + +const parseTargets = (value: unknown): SubscriptionTarget[] => { + if (value === undefined || value === null) return DEFAULT_DURABLE_TARGETS; + if (!Array.isArray(value) || value.length === 0) + throw badRequest( + 'targets must be a non-empty array', + 'invalid_targets', + ); + if (!value.every(isSubscriptionTarget)) + throw badRequest('Unknown delivery target', 'invalid_targets'); + return [...new Set(value)]; +}; + +const parseHandlerName = (value: unknown): string | null => { + if (value === undefined || value === null) return null; + if (typeof value !== 'string' || value.length === 0) + throw badRequest('handlerName must be a string', 'bad_request'); + if (value.length > HANDLER_NAME_MAX_LENGTH) + throw badRequest( + `handlerName may not exceed ${HANDLER_NAME_MAX_LENGTH} characters`, + 'bad_request', + ); + return value; +}; + +/** Stored as JSON text; the byte cap is the store's to enforce. */ +const parseContext = (value: unknown): string | null => { + if (value === undefined || value === null) return null; + try { + return JSON.stringify(value); + } catch { + throw badRequest('context must be JSON-serializable', 'bad_request'); + } +}; + +/** Unix seconds or an ISO-8601 string, and it has to be in the future. */ +const parseExpiresAt = (value: unknown): number | null => { + if (value === undefined || value === null) return null; + const seconds = + typeof value === 'number' + ? Math.floor(value) + : Math.floor(Date.parse(String(value)) / 1000); + if (!Number.isFinite(seconds) || seconds <= Math.floor(Date.now() / 1000)) + throw badRequest( + 'expiresAt must be a future time', + 'invalid_expires_at', + ); + return seconds; +}; + export class EventsService extends PuterService { readonly #cache = new SubscriptionCache(); readonly #compiled = new Map(); + readonly #lookups = new Map>(); readonly #refreshTimers = new Map>(); #coalescer: DeliveryCoalescer | null = null; + #expirySweep: ReturnType | null = null; + #expiryKick: ReturnType | null = null; // -- Lifecycle --------------------------------------------------- @@ -223,14 +404,22 @@ export class EventsService extends PuterService { // Our own emit reaches local listeners too, and that half has // already been applied. if (!(meta as { from_outside?: boolean })?.from_outside) return; - const { userId, generation } = (data ?? {}) as { + const { userId, durable } = (data ?? {}) as { userId?: number; - generation?: number; + durable?: boolean; }; if (typeof userId !== 'number') return; - this.#cache.bump(userId, generation); + this.invalidateUser(userId, { rebuild: durable === true }); }, ); + this.#armExpirySweep(); + } + + override onServerPrepareShutdown(): void { + if (this.#expiryKick) clearTimeout(this.#expiryKick); + this.#expiryKick = null; + if (this.#expirySweep) clearInterval(this.#expirySweep); + this.#expirySweep = null; } override onServerShutdown(): void { @@ -312,45 +501,13 @@ export class EventsService extends PuterService { await this.#spendCallBudget(holderUserId); const rawSubject = String(request?.subject ?? ''); - const parsed = parseSubject(rawSubject); - if (parsed.family !== 'fs') - throw new HttpError( - 400, - `Subject family not subscribable yet: ${parsed.family}`, - { legacyCode: 'invalid_subject' }, - ); - - const anchor = await resolveFsAnchor(parsed, this.#anchorDeps(), { - username: actor.user?.username, - }); - - // The resolver answers where a subscription keys, not whose it is, so - // the owner comes from the anchor node itself — and that is the - // keyspace the row is indexed in, because dispatch only ever knows - // whose resource changed. - const entry = await resolveNode(this.stores.fsEntry, { - uid: anchor.uid, - }); - if (!entry) - throw new HttpError(404, `No such entry: ${anchor.path}`, { - legacyCode: 'subject_does_not_exist', - }); - const permission = await assertSubscribeAuthorized( - actor, - { uid: anchor.uid, path: anchor.path }, - rawSubject, - this.#aclDeps(), - ); - - // Compile now so an unusable pattern fails this call rather than every - // event under the anchor. - if (anchor.match) compileMatch(anchor.match); + const anchor = await this.#resolveSubscribeAnchor(actor, rawSubject); const sub: SessionSubscription = { subId: randomUUID(), socketId, holderUserId, - ownerUserId: entry.userId, + ownerUserId: anchor.ownerUserId, subject: rawSubject, token: anchor.token, anchorUid: anchor.uid, @@ -358,11 +515,11 @@ export class EventsService extends PuterService { match: anchor.match, op: anchor.op, appUid: actor.effectiveApp?.uid ?? null, - permission, + permission: anchor.permission, }; const bump = await this.stores.eventSubscription.add(sub); - this.#publishGeneration(bump); + this.#publishGeneration(bump, false); this.#startRefresh(holderUserId, socketId); return { sub: toView(sub) }; @@ -393,7 +550,7 @@ export class EventsService extends PuterService { const bump = await this.stores.eventSubscription.remove(sub); this.#forget(subId); - this.#publishGeneration(bump); + this.#publishGeneration(bump, false); } /** What this actor holds on one connection, scoped to what it may see. */ @@ -412,6 +569,190 @@ export class EventsService extends PuterService { return held.filter((sub) => rowInActorScope(actor, sub)).map(toView); } + // -- Durable subscriptions --------------------------------------- + + /** + * Register a subscription that outlives the connection that made it. Same + * subject resolution and same ACL check as the session verb — what differs + * is where the row lands and who it is later addressed as. + */ + async subscribeDurable( + actor: Actor, + request: DurableSubscribeRequest, + ): Promise<{ sub: DurableSubscriptionView }> { + if (!this.enabled) throw disabled(); + const holderUserId = actor.user?.id; + if (holderUserId === undefined) throw disabled(); + + await this.#spendCallBudget(holderUserId); + + const delivery = parseDelivery(request?.delivery); + const targets = parseTargets(request?.targets); + const handlerName = parseHandlerName(request?.handlerName); + const context = parseContext(request?.context); + const expiresAt = parseExpiresAt(request?.expiresAt); + + const rawSubject = String(request?.subject ?? ''); + const anchor = await this.#resolveSubscribeAnchor(actor, rawSubject); + + const { row, bump } = await this.stores.durableSubscription.create({ + holderUserId, + ownerUserId: anchor.ownerUserId, + appUid: actor.effectiveApp?.uid ?? null, + subject: rawSubject, + token: anchor.token, + anchorUid: anchor.uid, + anchorPath: anchor.path, + match: anchor.match, + op: anchor.op, + delivery, + targets, + handlerName, + context, + permission: anchor.permission, + expiresAt, + }); + this.#publishGeneration(bump, true); + + return { sub: toDurableView(row) }; + } + + /** + * What this actor holds durably. An app-context actor is confined to its + * own rows by the index the query runs on; an account-context one sees + * across apps, which is what makes the account the revoke surface for a row + * whose app is long gone. + */ + async listDurable( + actor: Actor, + request: DurableListRequest = {}, + ): Promise> { + if (!this.enabled) throw disabled(); + const holderUserId = actor.user?.id; + if (holderUserId === undefined) throw disabled(); + + const app = actor.effectiveApp; + // Unresolved is not "no app" — reading it that way is what would hand + // an app the account-wide view. + if (app === undefined) return { items: [] }; + + const page = await this.stores.durableSubscription.listForHolder( + holderUserId, + { + appUid: app?.uid ?? null, + limit: request.limit, + cursor: request.cursor, + includeTotal: request.includeTotal, + }, + ); + return { + items: page.items.map(toDurableView), + ...(page.cursor ? { cursor: page.cursor } : {}), + ...(page.total !== undefined ? { total: page.total } : {}), + }; + } + + async unsubscribeDurable( + actor: Actor, + request: UnsubscribeRequest, + ): Promise { + if (!this.enabled) throw disabled(); + const holderUserId = actor.user?.id; + if (holderUserId === undefined) throw disabled(); + + await this.#spendCallBudget(holderUserId); + + const subId = String(request?.subId ?? ''); + if (!subId) throw unknownSubscription(); + + const row = await this.stores.durableSubscription.getBySubId(subId); + // Someone else's id — or one another app created — reads as absent + // rather than refused: a 403 here is an oracle for subIds. + if ( + !row || + row.holderUserId !== holderUserId || + !rowInActorScope(actor, row) + ) + throw unknownSubscription(); + + const bump = await this.stores.durableSubscription.remove(row); + this.#forget(subId); + this.#publishGeneration(bump, true); + } + + /** + * Drop rows past their expiry, in batches, and report how many went. Every + * node sweeps; the delete is idempotent, so two overlapping costs a few + * empty batches rather than correctness. + */ + async sweepExpired(): Promise { + if (!this.enabled) return 0; + let removed = 0; + for (let pass = 0; pass < EXPIRY_MAX_BATCHES; pass++) { + const batch = + await this.stores.durableSubscription.sweepExpired( + EXPIRY_BATCH_SIZE, + ); + removed += batch; + if (batch < EXPIRY_BATCH_SIZE) break; + } + return removed; + } + + /** + * Resolve, authorize and compile one subscribe request. Shared so a durable + * row cannot be created under a weaker check than a session one. + */ + async #resolveSubscribeAnchor( + actor: Actor, + rawSubject: string, + ): Promise<{ + token: string; + uid: string; + path: string; + match: string | null; + op: FsOp | null; + ownerUserId: number; + permission: AclMode; + }> { + const parsed = parseSubject(rawSubject); + if (parsed.family !== 'fs') + throw new HttpError( + 400, + `Subject family not subscribable yet: ${parsed.family}`, + { legacyCode: 'invalid_subject' }, + ); + + const anchor = await resolveFsAnchor(parsed, this.#anchorDeps(), { + username: actor.user?.username, + }); + + // The resolver answers where a subscription keys, not whose it is, so + // the owner comes from the anchor node itself — and that is the + // keyspace the row is indexed in, because dispatch only ever knows + // whose resource changed. + const entry = await resolveNode(this.stores.fsEntry, { + uid: anchor.uid, + }); + if (!entry) + throw new HttpError(404, `No such entry: ${anchor.path}`, { + legacyCode: 'subject_does_not_exist', + }); + + const permission = await assertSubscribeAuthorized( + actor, + { uid: anchor.uid, path: anchor.path }, + rawSubject, + this.#aclDeps(), + ); + + // Compile now so an unusable pattern fails this call rather than every + // event under the anchor. + if (anchor.match) compileMatch(anchor.match); + + return { ...anchor, ownerUserId: entry.userId, permission }; + } + /** Disconnect handler; also covers a socket the server dropped. */ async reapSocket(holderUserId: number, socketId: string): Promise { // Nothing could have been registered, so nothing has to be looked up. @@ -429,7 +770,7 @@ export class EventsService extends PuterService { holderUserId, socketId, ); - for (const bump of bumps) this.#publishGeneration(bump); + for (const bump of bumps) this.#publishGeneration(bump, false); } catch (err) { // The TTL backstop exists for exactly this. console.warn('[events] failed to reap socket subscriptions', err); @@ -484,9 +825,12 @@ export class EventsService extends PuterService { async #route( subject: PublicSubject, context: EventContext, - rows: SessionSubscription[], + candidates: DispatchSubscription[], actingUserId: number | undefined, ): Promise { + const rows = candidates.filter(deliverableOverSockets); + if (rows.length === 0) return; + // One throwaway projection reads the op off the registry entry rather // than re-deriving it from the subject string. const { op } = subject.project({ ...context, self: false, seq: 0 }); @@ -513,7 +857,7 @@ export class EventsService extends PuterService { seq: seq++, }); this.#coalesce().push(coalesceKey(row.subId, event.subject), { - socketId: row.socketId, + target: deliveryTarget(row), envelope: { subId: row.subId, event }, }); } @@ -540,7 +884,7 @@ export class EventsService extends PuterService { } /** Op filter first — a comparison, where the glob is not. */ - #passes(row: SessionSubscription, op: FsOp, matchOn: string): boolean { + #passes(row: DispatchSubscription, op: FsOp, matchOn: string): boolean { if (row.op !== null && row.op !== op) return false; if (!row.match) return true; @@ -560,9 +904,9 @@ export class EventsService extends PuterService { * and rows that share an identity and a grant share one decision. */ async #stillAuthorized( - rows: SessionSubscription[], + rows: DispatchSubscription[], context: EventContext, - ): Promise { + ): Promise { if (rows.length === 0) return rows; const node = this.#eventDescriptor(context); @@ -593,7 +937,7 @@ export class EventsService extends PuterService { ); } - #matcherFor(row: SessionSubscription): CompiledMatch { + #matcherFor(row: DispatchSubscription): CompiledMatch { const cached = this.#compiled.get(row.subId); if (cached && cached.pattern === row.match) return cached; const compiled = compileMatch(row.match as string); @@ -602,14 +946,14 @@ export class EventsService extends PuterService { } #gap( - rows: SessionSubscription[], + rows: DispatchSubscription[], subject: PublicSubject, context: EventContext, reason: GapReason, ): void { for (const row of rows) this.#send({ - socketId: row.socketId, + target: deliveryTarget(row), envelope: { subId: row.subId, event: { @@ -646,7 +990,7 @@ export class EventsService extends PuterService { } const event = delivery.envelope.event as ProjectedEvent; this.#send({ - socketId: delivery.socketId, + target: delivery.target, envelope: { subId: delivery.envelope.subId, event: { @@ -664,14 +1008,15 @@ export class EventsService extends PuterService { } /** - * Addressed at the socket's own id, which socket.io joins every socket to — - * so the adapter carries it to whichever node terminates the connection. + * Addressed at a socket id — which socket.io joins every socket to — or at + * a room, so either way the adapter carries it to whichever node terminates + * the connection. */ #send(delivery: AddressedDelivery): void { try { void this.services.socket .send( - { socket: delivery.socketId }, + delivery.target, EVENTS_DELIVERY_CHANNEL, delivery.envelope, ) @@ -698,18 +1043,42 @@ export class EventsService extends PuterService { /** * Whether this user has anything subscribed at all. Warm, a `Map` read; - * cold, one `EXISTS`. The generation is captured before the read, so a - * subscribe that lands mid-flight is not cached over. + * cold, one `EXISTS` — behind, at most once per warm window, the table read + * that teaches this region about durable rows it has never seen. Without + * that, an empty watched set is ambiguous: it means "nobody is listening" + * in a region that has looked, and nothing at all in one that has not. + * + * The epoch is captured before the read and is part of the in-flight key, + * so a bump landing mid-flight — a subscribe discovering the very folder a + * fire-and-forget dispatch is still warming up against, say — starts its + * own fresh lookup rather than being handed the answer an older, + * now-superseded one is about to compute. */ async #userHasAny(userId: number): Promise { const cached = this.#cache.read(userId); if (cached !== null) return cached; - const generation = this.#cache.generationOf(userId); + const epoch = this.#cache.generationOf(userId); + const key = `${userId}|${epoch}`; + + // Concurrent writes at the same epoch miss together, and each miss + // can cost a table read. One lookup answers all of them. + const inFlight = this.#lookups.get(key); + if (inFlight) return inFlight; + + const lookup = this.#lookUpHasAny(userId, epoch).finally(() => { + this.#lookups.delete(key); + }); + this.#lookups.set(key, lookup); + return lookup; + } + + async #lookUpHasAny(userId: number, epoch: number): Promise { try { + await this.stores.durableSubscription.warmRegion(userId); const hasAny = await this.stores.eventSubscription.userHasAny(userId); - this.#cache.write(userId, generation, hasAny); + this.#cache.write(userId, epoch, hasAny); return hasAny; } catch { // Not being able to tell is the same outcome as no subscribers, @@ -718,12 +1087,46 @@ export class EventsService extends PuterService { } } - #publishGeneration({ userId, generation }: GenerationBump): void { + /** + * Forget what this process, and this region, believe about one user's + * subscriptions. Where a generation bump from anywhere else lands: the + * process drops its answer, and the region rebuilds its durable rows from + * the table on the next dispatch. + * + * Bumps unconditionally rather than passing a remote generation through to + * the cache's own number check: `ev:g` is a region-local counter, so a + * peer's bump can carry a number this process's own counter has already + * passed for entirely unrelated reasons, and comparing them would let a + * real invalidation be silently ignored as "already applied". There is + * nothing to order a cross-region signal against, so every one just + * forgets, and `SubscriptionCache`'s read-side TTL is what bounds a process + * that never received one at all. + */ + invalidateUser( + userId: number, + { rebuild = true }: { rebuild?: boolean } = {}, + ): void { + this.#cache.bump(userId); + if (!rebuild) return; + void this.stores.eventSubscription + .markRegionCold(userId) + .catch(() => {}); + } + + /** + * `durable` says whether the table changed. Session rows live in this + * region's Redis alone, so a peer hearing about one has nothing to rebuild + * — only a durable bump is worth a primary read over there. + */ + #publishGeneration( + { userId, generation }: GenerationBump, + durable: boolean, + ): void { this.#cache.bump(userId, generation); try { this.clients.event.emit( 'outer.events.generationBumped', - { userId, generation }, + { userId, generation, durable }, {}, ); } catch { @@ -783,6 +1186,25 @@ export class EventsService extends PuterService { this.#refreshTimers.set(key, timer); } + #armExpirySweep(): void { + if (!this.enabled) return; + const run = () => { + void this.sweepExpired().catch((err) => { + console.warn('[events] expiry sweep failed', err); + }); + }; + // Jittered so a deploy does not have every node sweep at once. + const kick = setTimeout( + run, + EXPIRY_SWEEP_INITIAL_DELAY_MS * (0.5 + Math.random()), + ); + kick.unref?.(); + this.#expiryKick = kick; + const sweep = setInterval(run, EXPIRY_SWEEP_INTERVAL_MS); + sweep.unref?.(); + this.#expirySweep = sweep; + } + #stopRefresh(holderUserId: number, socketId: string): void { const key = `${holderUserId}|${socketId}`; const timer = this.#refreshTimers.get(key); diff --git a/src/backend/services/events/dispatch.integration.test.ts b/src/backend/services/events/dispatch.integration.test.ts index b2ea10300..b939f7504 100644 --- a/src/backend/services/events/dispatch.integration.test.ts +++ b/src/backend/services/events/dispatch.integration.test.ts @@ -43,8 +43,15 @@ let delivered: DeliveryEnvelope[]; const events = () => env.server.services.events; const fs = () => env.server.services.fs; -const settle = () => - vi.waitFor(() => expect(delivered.length).toBeGreaterThan(0), { +// The coalesce window runs on real time, so a delivery from one test can land +// after the next has begun. Waits name what they wait for, and assertions look +// only at their own folder. +const pathOf = (envelope: DeliveryEnvelope): string => + (envelope.event as { path?: string }).path ?? ''; +const deliveredUnder = (folder: string) => + delivered.filter((envelope) => pathOf(envelope).startsWith(`${folder}/`)); +const settle = (predicate: (envelope: DeliveryEnvelope) => boolean) => + vi.waitFor(() => expect(delivered.some(predicate)).toBe(true), { timeout: EVENTS_COALESCE_WINDOW_MS * 8, interval: 25, }); @@ -76,17 +83,15 @@ describe('the write path reaches subscribers', () => { const folder = `/${username}/watch-create`; await fs().mkdir(userId, { path: folder, createMissingParents: true }); const sub = await subscribeTo(`fs:${folder}`); - delivered.length = 0; - await fs().touch(userId, { path: `${folder}/made.txt` }); - await settle(); + const made = `${folder}/made.txt`; + await fs().touch(userId, { path: made }); + await settle((d) => pathOf(d) === made); - expect(delivered).toHaveLength(1); - expect(delivered[0].subId).toBe(sub.subId); - expect(delivered[0].event).toMatchObject({ - op: 'add', - path: `${folder}/made.txt`, - }); + const mine = deliveredUnder(folder); + expect(mine).toHaveLength(1); + expect(mine[0].subId).toBe(sub.subId); + expect(mine[0].event).toMatchObject({ op: 'add', path: made }); }); it('delivers a rename and a remove on the same subscription', async () => { @@ -94,17 +99,16 @@ describe('the write path reaches subscribers', () => { await fs().mkdir(userId, { path: folder, createMissingParents: true }); await subscribeTo(`fs:${folder}`); const file = await fs().touch(userId, { path: `${folder}/before.txt` }); - await settle(); - delivered.length = 0; + await settle((d) => pathOf(d) === `${folder}/before.txt`); + + const inFolder = (op: string) => (d: DeliveryEnvelope) => + d.event.op === op && pathOf(d).startsWith(`${folder}/`); const renamed = await fs().rename(userId, file, 'after.txt'); - await settle(); - expect(delivered.map((d) => d.event.op)).toContain('move'); - delivered.length = 0; + await settle(inFolder('move')); await fs().remove(userId, { entry: renamed }); - await settle(); - expect(delivered.map((d) => d.event.op)).toContain('remove'); + await settle(inFolder('remove')); }); it('addresses the delivery at the socket that subscribed', async () => { @@ -112,10 +116,10 @@ describe('the write path reaches subscribers', () => { await fs().mkdir(userId, { path: folder, createMissingParents: true }); await subscribeTo(`fs:${folder}`); const send = vi.spyOn(env.server.services.socket, 'send'); - delivered.length = 0; - await fs().touch(userId, { path: `${folder}/addressed.txt` }); - await settle(); + const addressed = `${folder}/addressed.txt`; + await fs().touch(userId, { path: addressed }); + await settle((d) => pathOf(d) === addressed); expect(send).toHaveBeenCalledWith( { socket: SOCKET_ID }, @@ -128,14 +132,13 @@ describe('the write path reaches subscribers', () => { it('leaves an unwatched folder alone', async () => { const folder = `/${username}/watch-nothing`; await fs().mkdir(userId, { path: folder, createMissingParents: true }); - delivered.length = 0; await fs().touch(userId, { path: `${folder}/ignored.txt` }); await new Promise((resolve) => setTimeout(resolve, EVENTS_COALESCE_WINDOW_MS * 2), ); - expect(delivered).toEqual([]); + expect(deliveredUnder(folder)).toEqual([]); }); }); @@ -188,13 +191,12 @@ describe('unsubscribing and disconnecting', () => { const sub = await subscribeTo(`fs:${folder}`); await events().unsubscribe(actor, SOCKET_ID, { subId: sub.subId }); - delivered.length = 0; await fs().touch(userId, { path: `${folder}/after.txt` }); await new Promise((resolve) => setTimeout(resolve, EVENTS_COALESCE_WINDOW_MS * 2), ); - expect(delivered).toEqual([]); + expect(deliveredUnder(folder)).toEqual([]); }); it('leaves no watched token behind when the socket goes', async () => { diff --git a/src/backend/services/events/durable.integration.test.ts b/src/backend/services/events/durable.integration.test.ts new file mode 100644 index 000000000..90f667680 --- /dev/null +++ b/src/backend/services/events/durable.integration.test.ts @@ -0,0 +1,636 @@ +/* + * 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 . + */ + +/** + * Durable subscriptions end to end: the routes that create and revoke them, who + * each credential shape may see, and what a write costs once one exists. + * + * The cost assertions are the point of the whole cache. A subscription made + * here is deliverable here without the table, and a region that has never seen + * the user pays for the table exactly once. + */ + +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; +import { v4 as uuidv4 } from 'uuid'; +import { EVENTS_COALESCE_WINDOW_MS } from '../../controllers/events/limits.js'; +import { makeActor } from '../../core/actor.js'; +import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; +import type { IConfig } from '../../types.js'; +import { appSocketRoom } from '../socket/SocketService.js'; +import type { DeliveryEnvelope } from './EventsService.js'; + +const BOOT_TIMEOUT_MS = 120_000; +const TABLE = 'event_subscriptions'; + +let env: PuterTestEnv; +let userId: number; +let username: string; +let anchor: string; +let appOneUid: string; +let appOneToken: string; +let appTwoUid: string; +let appTwoToken: string; +let appOneAccessToken: string; +let delivered: DeliveryEnvelope[]; + +const events = () => env.server.services.events; +const fs = () => env.server.services.fs; + +interface ApiResponse { + status: number; + body: Record; +} + +const call = async ( + method: 'GET' | 'POST', + path: string, + token: string, + body?: object, +): Promise => { + const response = await fetch(new URL(path, env.apiOrigin), { + method, + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${token}`, + }, + ...(body ? { body: JSON.stringify(body) } : {}), + }); + return { + status: response.status, + body: (await response.json()) as Record, + }; +}; + +const subscribe = (token: string, body: object = {}): Promise => + call('POST', '/events/subscribe', token, { subject: `fs:${anchor}`, ...body }); + +const listSubscriptions = ( + token: string, + query = '', +): Promise => + call('GET', `/events/subscriptions${query}`, token); + +const unsubscribe = (token: string, subId: string): Promise => + call('POST', '/events/unsubscribe', token, { subId }); + +const subIdsOf = (response: ApiResponse): string[] => + (response.body.items as Array<{ subId: string }>).map((row) => row.subId); + +/** An app the user has granted `list` on the shared anchor. */ +const makeApp = async (): Promise<{ uid: string; token: string }> => { + const uid = `app-${uuidv4()}`; + await env.server.clients.db.write( + 'INSERT INTO `apps` (`uid`, `name`, `title`, `index_url`, `owner_user_id`) VALUES (?, ?, ?, ?, ?)', + [uid, uid, uid, `https://${uid}.example/`, userId], + ); + const entry = await env.server.stores.fsEntry.getEntryByPath(anchor); + const actor = await env.server.services.auth.authenticate( + env.users.user.token, + ); + await env.server.services.permission.grantUserAppPermission( + actor.actor!, + uid, + `fs:${entry!.uid}:list`, + ); + return { + uid, + token: await env.server.services.auth.getUserAppToken(actor.actor!, uid), + }; +}; + +/** + * Table reads the block performed, counted where the driver sees them. On a + * single-node engine `pread` is `read`, so one query would otherwise be counted + * at both — the guard makes the count the number of statements, not of frames. + */ +const countTableReads = async (run: () => Promise): Promise => { + const db = env.server.clients.db; + const passThroughRead = db.read.bind(db); + const passThroughPread = db.pread.bind(db); + let reads = 0; + let insidePread = false; + + const pread = vi.spyOn(db, 'pread').mockImplementation(async (q, p) => { + if (q.includes(TABLE)) reads++; + insidePread = true; + try { + return await passThroughPread(q, p); + } finally { + insidePread = false; + } + }); + const read = vi.spyOn(db, 'read').mockImplementation(async (q, p) => { + if (q.includes(TABLE) && !insidePread) reads++; + return passThroughRead(q, p); + }); + try { + await run(); + return reads; + } finally { + read.mockRestore(); + pread.mockRestore(); + } +}; + +// The coalesce window runs on real time, so a delivery from one test can land +// after the next has begun. Every wait names the path it is waiting for. +const pathOf = (envelope: DeliveryEnvelope): string => + (envelope.event as { path?: string }).path ?? ''; +const deliveryOf = (path: string) => + delivered.find((envelope) => pathOf(envelope) === path); +const settle = (path: string) => + vi.waitFor(() => expect(deliveryOf(path)).toBeDefined(), { + timeout: EVENTS_COALESCE_WINDOW_MS * 12, + interval: 25, + }); + +const quiet = () => + new Promise((resolve) => + setTimeout(resolve, EVENTS_COALESCE_WINDOW_MS * 3), + ); + +beforeAll(async () => { + env = await setupPuterTestEnv({ events: { enabled: true } } as IConfig); + username = env.users.user.username; + const user = await env.server.stores.user.getByUsername(username); + userId = user!.id; + + anchor = `/${username}/durable`; + await fs().mkdir(userId, { path: anchor, createMissingParents: true }); + + ({ uid: appOneUid, token: appOneToken } = await makeApp()); + ({ uid: appTwoUid, token: appTwoToken } = await makeApp()); + + const appActor = await env.server.services.auth.authenticate(appOneToken); + const entry = await env.server.stores.fsEntry.getEntryByPath(anchor); + appOneAccessToken = await env.server.services.auth.createAccessToken( + appActor.actor!, + [[`fs:${entry!.uid}:list`]], + { label: 'durable' }, + ); + + delivered = []; + events().onDelivered = (envelope) => delivered.push(envelope); +}, BOOT_TIMEOUT_MS); + +afterAll(async () => { + await env?.shutdown(); +}); + +/** + * Start each case from a known state on both layers: no rows, this process + * holding no answer, and the region rebuilt from the now-empty table so it is + * warm and correct rather than merely empty. + */ +const clearRows = async () => { + await env.server.clients.db.write(`DELETE FROM \`${TABLE}\``, []); + events().invalidateUser(userId); + await env.server.stores.eventSubscription.markRegionCold(userId); + await env.server.stores.durableSubscription.warmRegion(userId); +}; + +describe('creating a durable subscription over HTTP', () => { + it('returns the row a client needs to revoke it later', async () => { + await clearRows(); + const created = await subscribe(env.users.user.token); + + expect(created.status).toBe(200); + expect(created.body).toMatchObject({ + subject: `fs:${anchor}`, + delivery: 'broadcast', + targets: ['socket', 'worker'], + appUid: null, + }); + expect(created.body.context).toBeUndefined(); + }); + + it('refuses the delivery class that has nowhere to queue yet', async () => { + const refused = await subscribe(env.users.user.token, { + delivery: 'single', + handlerName: 'onWrite', + }); + + expect(refused.status).toBe(501); + expect(refused.body.code).toBe('delivery_class_unavailable'); + }); + + it('refuses a target outside the known set', async () => { + const refused = await subscribe(env.users.user.token, { + targets: ['socket', 'carrier-pigeon'], + }); + + expect(refused.status).toBe(400); + expect(refused.body.code).toBe('invalid_targets'); + }); + + it('refuses an expiry in the past', async () => { + const refused = await subscribe(env.users.user.token, { + expiresAt: Math.floor(Date.now() / 1000) - 60, + }); + + expect(refused.status).toBe(400); + expect(refused.body.code).toBe('invalid_expires_at'); + }); + + it('answers a subject it cannot read as absent', async () => { + const refused = await call('POST', '/events/subscribe', appOneToken, { + subject: `fs:/${username}/not-granted`, + }); + + expect(refused.status).toBe(404); + expect(refused.body.code).toBe('subject_does_not_exist'); + }); +}); + +describe('what each credential sees and removes', () => { + it('confines an app to its own rows and lets the account see across them', async () => { + await clearRows(); + const mine = (await subscribe(appOneToken)).body.subId as string; + const theirs = (await subscribe(appTwoToken)).body.subId as string; + const account = (await subscribe(env.users.user.token)).body + .subId as string; + + expect(subIdsOf(await listSubscriptions(appOneToken))).toEqual([mine]); + expect(subIdsOf(await listSubscriptions(appTwoToken))).toEqual([ + theirs, + ]); + // An access token an app issued acts as that app, one hop through the + // issuer — which is what the whole scope keys on. + expect(subIdsOf(await listSubscriptions(appOneAccessToken))).toEqual([ + mine, + ]); + + for (const wide of [env.users.user.token, env.users.user.apiToken]) + expect(subIdsOf(await listSubscriptions(wide)).sort()).toEqual( + [mine, theirs, account].sort(), + ); + }); + + it('answers another app`s subscription id as absent', async () => { + await clearRows(); + const mine = (await subscribe(appOneToken)).body.subId as string; + + const refused = await unsubscribe(appTwoToken, mine); + expect(refused.status).toBe(404); + expect(refused.body.code).toBe('subscription_does_not_exist'); + expect(subIdsOf(await listSubscriptions(appOneToken))).toEqual([mine]); + }); + + it('answers an id that never existed the same way', async () => { + const refused = await unsubscribe( + env.users.user.token, + `${appOneUid}#${uuidv4()}`, + ); + expect(refused.status).toBe(404); + expect(refused.body.code).toBe('subscription_does_not_exist'); + }); + + it('lets the account remove what an app left behind', async () => { + await clearRows(); + const theirs = (await subscribe(appTwoToken)).body.subId as string; + expect(appTwoUid).not.toBe(appOneUid); + + const removed = await unsubscribe(env.users.user.token, theirs); + + expect(removed.status).toBe(200); + expect(subIdsOf(await listSubscriptions(env.users.user.token))).toEqual( + [], + ); + }); + + it('pages the listing and totals the scope, not the page', async () => { + await clearRows(); + for (let i = 0; i < 3; i++) await subscribe(env.users.user.token); + + const first = await listSubscriptions( + env.users.user.token, + '?limit=2&includeTotal=true', + ); + expect(first.body.items).toHaveLength(2); + expect(first.body.total).toBe(3); + expect(first.body.cursor).toBeDefined(); + + const second = await listSubscriptions( + env.users.user.token, + `?limit=2&cursor=${encodeURIComponent(String(first.body.cursor))}`, + ); + expect(second.body.items).toHaveLength(1); + expect(second.body.cursor).toBeUndefined(); + expect(second.body.total).toBeUndefined(); + }); +}); + +describe('what a write costs once a durable row exists', () => { + it('delivers straight after subscribe without reading the table', async () => { + await clearRows(); + const created = await subscribe(env.users.user.token); + delivered.length = 0; + + const path = `${anchor}/warm-${uuidv4().slice(0, 8)}.txt`; + const reads = await countTableReads(async () => { + await fs().touch(userId, { path }); + await settle(path); + }); + + expect(reads).toBe(0); + expect(deliveryOf(path)?.subId).toBe(created.body.subId); + }); + + it('rebuilds a cold region from the table once, then stops reading it', async () => { + await clearRows(); + const created = await subscribe(env.users.user.token); + + // A region that has never seen this user: nothing cached anywhere. + // Targeted rather than `flushall`: this shares a Redis instance (and + // an `ev:g` counter) with every other test in the process, and a + // global flush would reset it out from under their own generation + // bookkeeping too. + await env.server.clients.redis.del( + `ev:w:{${userId}}`, + `ev:dw:{${userId}}`, + `ev:dm:{${userId}}`, + ); + events().invalidateUser(userId); + delivered.length = 0; + + const coldPath = `${anchor}/cold-${uuidv4().slice(0, 8)}.txt`; + const cold = await countTableReads(async () => { + await fs().touch(userId, { path: coldPath }); + await settle(coldPath); + }); + expect(cold).toBe(1); + expect(deliveryOf(coldPath)?.subId).toBe(created.body.subId); + + const warmPath = `${anchor}/again-${uuidv4().slice(0, 8)}.txt`; + const warm = await countTableReads(async () => { + await fs().touch(userId, { path: warmPath }); + await settle(warmPath); + }); + expect(warm).toBe(0); + }); + + it('evicts a row a peer region removed, once its bump arrives', async () => { + await clearRows(); + const created = await subscribe(env.users.user.token); + delivered.length = 0; + const before = `${anchor}/pre-peer-${uuidv4().slice(0, 8)}.txt`; + await fs().touch(userId, { path: before }); + await settle(before); + expect(deliveryOf(before)?.subId).toBe(created.body.subId); + + // A peer region settling the row: the primary loses it, but this + // region is never told directly — only the bump such a removal would + // broadcast is simulated, over the real event bus rather than a + // direct call into the store. + await env.server.clients.db.write( + `DELETE FROM \`${TABLE}\` WHERE \`sub_id\` = ?`, + [created.body.subId], + ); + const generation = + await env.server.stores.eventSubscription.getGeneration(userId); + delivered.length = 0; + + const after = `${anchor}/post-peer-${uuidv4().slice(0, 8)}.txt`; + const reads = await countTableReads(async () => { + await env.server.clients.event.emitAndWait( + 'outer.events.generationBumped', + { userId, generation: generation + 1, durable: true }, + { from_outside: true }, + ); + await fs().touch(userId, { path: after }); + await quiet(); + }); + + expect(reads).toBe(1); // the bump alone forced exactly one rebuild + expect(deliveryOf(after)).toBeUndefined(); + }); + + it('does not re-read the table for a peer`s session bump', async () => { + await clearRows(); + const created = await subscribe(env.users.user.token); + const generation = + await env.server.stores.eventSubscription.getGeneration(userId); + delivered.length = 0; + + // A session subscribe in another region touches only that region's + // Redis; the table this region cached from is unchanged. + const path = `${anchor}/peer-session-${uuidv4().slice(0, 8)}.txt`; + const reads = await countTableReads(async () => { + await env.server.clients.event.emitAndWait( + 'outer.events.generationBumped', + { userId, generation: generation + 1, durable: false }, + { from_outside: true }, + ); + await fs().touch(userId, { path }); + await settle(path); + }); + + expect(reads).toBe(0); + expect(deliveryOf(path)?.subId).toBe(created.body.subId); + }); + + it('stops delivering once the subscription is revoked', async () => { + await clearRows(); + const created = await subscribe(env.users.user.token); + await unsubscribe(env.users.user.token, created.body.subId as string); + delivered.length = 0; + + const path = `${anchor}/revoked-${uuidv4().slice(0, 8)}.txt`; + await fs().touch(userId, { path }); + await quiet(); + + expect(deliveryOf(path)).toBeUndefined(); + }); + + it('stops delivering once the subscription is swept', async () => { + await clearRows(); + await subscribe(env.users.user.token, { + expiresAt: Math.floor(Date.now() / 1000) + 3600, + }); + await env.server.clients.db.write( + `UPDATE \`${TABLE}\` SET \`expires_at\` = ?`, + [Math.floor(Date.now() / 1000) - 1], + ); + + await expect(events().sweepExpired()).resolves.toBe(1); + delivered.length = 0; + + const path = `${anchor}/swept-${uuidv4().slice(0, 8)}.txt`; + await fs().touch(userId, { path }); + await quiet(); + + expect(deliveryOf(path)).toBeUndefined(); + }); +}); + +describe('where a durable delivery is addressed', () => { + it('sends an app`s row to the app`s own room', async () => { + await clearRows(); + await subscribe(appOneToken); + const send = vi.spyOn(env.server.services.socket, 'send'); + delivered.length = 0; + + const path = `${anchor}/app-${uuidv4().slice(0, 8)}.txt`; + await fs().touch(userId, { path }); + await settle(path); + + expect(send).toHaveBeenCalledWith( + { room: appSocketRoom(userId, appOneUid) }, + 'events.delivery', + expect.objectContaining({ subId: expect.any(String) }), + ); + send.mockRestore(); + }); + + it('sends a session`s row to the account`s own room', async () => { + await clearRows(); + await subscribe(env.users.user.token); + const send = vi.spyOn(env.server.services.socket, 'send'); + delivered.length = 0; + + const path = `${anchor}/session-${uuidv4().slice(0, 8)}.txt`; + await fs().touch(userId, { path }); + await settle(path); + + expect(send).toHaveBeenCalledWith( + { room: String(userId) }, + 'events.delivery', + expect.objectContaining({ subId: expect.any(String) }), + ); + send.mockRestore(); + }); +}); + +describe('a durable row`s match filter', () => { + it('filters deliveries the same way a session subscription`s would', async () => { + await clearRows(); + const created = await subscribe(env.users.user.token, { + subject: `fs:${anchor}/only-this.txt`, + }); + delivered.length = 0; + + await fs().touch(userId, { path: `${anchor}/not-this.txt` }); + await quiet(); + expect(deliveryOf(`${anchor}/not-this.txt`)).toBeUndefined(); + + await fs().touch(userId, { path: `${anchor}/only-this.txt` }); + await settle(`${anchor}/only-this.txt`); + expect(deliveryOf(`${anchor}/only-this.txt`)?.subId).toBe( + created.body.subId, + ); + }); +}); + +describe('a durable row across a share', () => { + it('stops delivering the moment the share is revoked', async () => { + await clearRows(); + const ownerRow = await env.server.stores.user.getByUsername(username); + const guestRow = await env.server.stores.user.getByUsername( + env.users.other.username, + ); + const ownerActor = makeActor({ user: ownerRow as never }); + const guestActor = makeActor({ user: guestRow as never }); + + const sharedPath = `${anchor}/shared-with-guest`; + await fs().mkdir(userId, { + path: sharedPath, + createMissingParents: true, + }); + await env.server.services.acl.setUserUser( + ownerActor, + guestActor, + { + path: sharedPath, + resolveAncestors: () => fs().getAncestorChain(sharedPath), + }, + 'list', + ); + + const created = await subscribe(env.users.other.token, { + subject: `fs:${sharedPath}`, + }); + delivered.length = 0; + + await fs().touch(userId, { path: `${sharedPath}/first.txt` }); + await settle(`${sharedPath}/first.txt`); + expect(deliveryOf(`${sharedPath}/first.txt`)?.subId).toBe( + created.body.subId, + ); + + const sharedEntry = + await env.server.stores.fsEntry.getEntryByPath(sharedPath); + await env.server.services.permission.revokeUserUserPermission( + ownerActor, + env.users.other.username, + `fs:${sharedEntry!.uid}:list`, + ); + delivered.length = 0; + + // The row is still registered; it just no longer authorizes anything. + await fs().touch(userId, { path: `${sharedPath}/second.txt` }); + await quiet(); + + expect(deliveryOf(`${sharedPath}/second.txt`)).toBeUndefined(); + }); +}); + +describe('with events switched off', () => { + let off: PuterTestEnv; + + beforeAll(async () => { + off = await setupPuterTestEnv(); + }, BOOT_TIMEOUT_MS); + + afterAll(async () => { + await off?.shutdown(); + }); + + const offCall = async ( + method: 'GET' | 'POST', + path: string, + body?: object, + ): Promise => { + const response = await fetch(new URL(path, off.apiOrigin), { + method, + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${off.users.user.token}`, + }, + ...(body ? { body: JSON.stringify(body) } : {}), + }); + return { + status: response.status, + body: (await response.json()) as Record, + }; + }; + + it('refuses every verb with one code', async () => { + for (const attempt of [ + offCall('POST', '/events/subscribe', { + subject: `fs:/${off.users.user.username}`, + }), + offCall('GET', '/events/subscriptions'), + offCall('POST', '/events/unsubscribe', { subId: 'user#nope' }), + ]) { + const response = await attempt; + expect(response.status).toBe(503); + expect(response.body.code).toBe('events_disabled'); + } + }); +}); diff --git a/src/backend/services/events/subjects.test.ts b/src/backend/services/events/subjects.test.ts index ccdae625f..b97e6c4fe 100644 --- a/src/backend/services/events/subjects.test.ts +++ b/src/backend/services/events/subjects.test.ts @@ -23,6 +23,7 @@ import { PermissionUtil } from '../permission/permissionUtil.js'; import { fsAnchorToken, parseSubject, + SUBJECT_MAX_LENGTH, type AnchorRef, type FsOp, } from './subjects.js'; @@ -205,6 +206,26 @@ describe('parseSubject', () => { }); }); +describe('subject length', () => { + it('rejects a subject longer than the widest path the filesystem stores', () => { + const subject = `fs:/alice/${'a'.repeat(SUBJECT_MAX_LENGTH)}`; + let thrown: unknown; + try { + parseSubject(subject); + } catch (err) { + thrown = err; + } + expect(thrown).toBeInstanceOf(HttpError); + expect((thrown as HttpError).legacyCode).toBe('invalid_subject'); + }); + + it('accepts one right at the cap', () => { + const prefix = 'fs:/alice/'; + const subject = `${prefix}${'a'.repeat(SUBJECT_MAX_LENGTH - prefix.length)}`; + expect(parseSubject(subject).family).toBe('fs'); + }); +}); + describe('fsAnchorToken', () => { it('namespaces node uids', () => { expect(fsAnchorToken('uid-abc')).toBe('f#uid-abc'); diff --git a/src/backend/services/events/subjects.ts b/src/backend/services/events/subjects.ts index a7e13abda..fd151a2b6 100644 --- a/src/backend/services/events/subjects.ts +++ b/src/backend/services/events/subjects.ts @@ -72,6 +72,9 @@ export const FS_OPS: readonly FsOp[] = Object.freeze([ */ export const KV_TOKEN_SEGMENT_CAP = 6; +/** Longest subject accepted: the widest path the filesystem itself stores. */ +export const SUBJECT_MAX_LENGTH = 4096; + const GLOB_CHARS = /[*?]/; // -- Anchor tokens ---------------------------------------------------- @@ -207,6 +210,7 @@ const parseNotifSubject = (subject: string, parts: string[]): ParsedSubject => { export function parseSubject(subject: string): ParsedSubject { if (typeof subject !== 'string' || subject.trim().length === 0) throw invalidSubject(String(subject)); + if (subject.length > SUBJECT_MAX_LENGTH) throw invalidSubject(subject); const trimmed = subject.trim(); const parts = PermissionUtil.split(trimmed); diff --git a/src/backend/services/events/subscriptionCache.test.ts b/src/backend/services/events/subscriptionCache.test.ts index 37df927c9..34e2ea413 100644 --- a/src/backend/services/events/subscriptionCache.test.ts +++ b/src/backend/services/events/subscriptionCache.test.ts @@ -17,7 +17,7 @@ * along with this program. If not, see . */ -import { describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { SubscriptionCache } from './subscriptionCache.js'; describe('answers', () => { @@ -65,9 +65,11 @@ describe('invalidation', () => { it('cannot be walked backwards by a bump that arrives late', () => { const cache = new SubscriptionCache(); cache.bump(1, 5); - cache.bump(1, 2); + // Captured as a lookup would, right after the generation that matters. + const epoch = cache.generationOf(1); + cache.bump(1, 2); // stale — must not undo generation 5's invalidation - cache.write(1, 5, true); + cache.write(1, epoch, true); expect(cache.read(1)).toBe(true); }); @@ -81,6 +83,60 @@ describe('invalidation', () => { expect(cache.read(1)).toBeNull(); expect(cache.generationOf(1)).toBe(1); }); + + it('does not let a number-less bump block a real one that follows it', () => { + // A number-less invalidation (`invalidateUser`'s bare call, or any + // other local reason to forget) must not plant a value a following + // *real* generation — a local subscribe's first-ever bump, often a + // small number — could compare behind and be ignored for. Mirrors a + // stale "nothing subscribed" answer surviving a subscribe that + // landed moments later. + const cache = new SubscriptionCache(); + cache.write(1, cache.generationOf(1), false); // cached stale "false" + cache.bump(1); // e.g. a number-less invalidation + cache.write(1, cache.generationOf(1), false); // re-checked, still "false" + + cache.bump(1, 1); // a real local publish landing shortly after + + expect(cache.read(1)).toBeNull(); // forced to look again, not stuck + }); +}); + +describe('cross-region generation mismatch', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('self-heals on a TTL even when a lower foreign generation is ignored', () => { + // A generation counter is region-local: a peer's bump can carry a + // number behind what this process already recorded for its own + // local traffic, purely because it was counted by a different + // counter. `bump()` treats that as already-applied and does not + // clear the cached answer — the TTL is what stops it surviving + // forever regardless. + const cache = new SubscriptionCache(10_000, 2_000); + cache.bump(1, 5); // a real local publish + cache.write(1, cache.generationOf(1), false); // cached under it + + cache.bump(1, 1); // a peer's bump, numbered behind this process's own + expect(cache.read(1)).toBe(false); // not yet invalidated by number + + vi.advanceTimersByTime(2_001); + expect(cache.read(1)).toBeNull(); // but stale past the TTL + }); + + it('keeps answering within the TTL without a bump at all', () => { + const cache = new SubscriptionCache(10_000, 2_000); + cache.write(1, 0, true); + + vi.advanceTimersByTime(1_000); + + expect(cache.read(1)).toBe(true); + }); }); describe('bounds', () => { diff --git a/src/backend/services/events/subscriptionCache.ts b/src/backend/services/events/subscriptionCache.ts index 83be4bcda..c5ac3caba 100644 --- a/src/backend/services/events/subscriptionCache.ts +++ b/src/backend/services/events/subscriptionCache.ts @@ -25,8 +25,35 @@ * nothing: after the first miss the answer is in memory and dispatch never * touches Redis again. That only holds if invalidation is pushed rather than * polled, so entries are keyed by a per-user generation the subscribe and - * unsubscribe paths bump and broadcast — never by a timer, which would put the - * round trip back on the hot path at whatever rate the timer expired. + * unsubscribe paths bump and broadcast. + * + * Two different numbers share the name "generation" in the surrounding code, + * and this cache keeps them as two separate fields rather than one: + * + * - `epoch` — purely local, bumped on every invalidation this process makes of + * its own answer, whatever the reason. `write()` captures it before a lookup + * starts and compares on the way back in: a mismatch means something + * invalidated while the lookup was in flight, and the computed answer is + * dropped rather than cached stale. It never needs to mean anything to + * another process. + * - `redisGeneration` — the store's `ev:g` value, real only when a _local_ + * subscribe/unsubscribe supplied one (`#publishGeneration`). Two such bumps + * can race and arrive out of order, and this is what lets the later one win + * regardless. + * + * Conflating them was the bug: `ev:g` is a region-local `INCR`, so a peer + * region's bump can carry a number behind one this process already recorded for + * entirely unrelated reasons (its own local traffic, or an earlier invalidation + * that had no number to report), and comparing them let a real invalidation be + * silently ignored as "already applied". A bump with no number — + * `invalidateUser`'s only mode; see its own comment — advances `epoch` + * unconditionally and leaves `redisGeneration` untouched, so it can never + * falsely outrank, or be outranked by, a store-issued one. `read()` is the + * remaining backstop: a definite answer only survives a short TTL before it is + * treated as unknown again, exactly the "~2 s local-gen TTL trade" the + * permission cache makes (`PermissionStore.ts`, + * `PERMISSION_CACHE_GENERATION_LOCAL_TTL_SECONDS`) — bounding staleness by time + * wherever a counter cannot order it. * * Bounded, because one process sees an unbounded number of users over its * lifetime and the useful entries are the ones being written to right now. @@ -35,34 +62,50 @@ */ interface CacheEntry { - generation: number; - /** `null` while unknown — a bump leaves the generation and clears this. */ + epoch: number; + /** `null` until a local subscribe/unsubscribe has supplied a real one. */ + redisGeneration: number | null; + /** `null` while unknown — a bump leaves the epoch and clears this. */ hasAny: boolean | null; + /** When `hasAny` was last written; what the read-side TTL measures from. */ + cachedAt: number; } export const SUBSCRIPTION_CACHE_MAX_USERS = 10_000; +/** + * How long a definite answer survives without a bump. Mirrors the permission + * cache's local TTL. + */ +export const SUBSCRIPTION_CACHE_TTL_MS = 2_000; + export class SubscriptionCache { readonly #entries = new Map(); readonly #maxUsers: number; + readonly #ttlMs: number; - constructor(maxUsers: number = SUBSCRIPTION_CACHE_MAX_USERS) { + constructor( + maxUsers: number = SUBSCRIPTION_CACHE_MAX_USERS, + ttlMs: number = SUBSCRIPTION_CACHE_TTL_MS, + ) { this.#maxUsers = Math.max(1, maxUsers); + this.#ttlMs = Math.max(0, ttlMs); } get size(): number { return this.#entries.size; } - /** The generation this process believes the user is on. */ + /** The epoch a lookup must capture before reading, to write safely after. */ generationOf(userId: number): number { - return this.#entries.get(userId)?.generation ?? 0; + return this.#entries.get(userId)?.epoch ?? 0; } /** The cached answer, or `null` when this process has to go and look. */ read(userId: number): boolean | null { const entry = this.#entries.get(userId); if (!entry) return null; + if (Date.now() - entry.cachedAt > this.#ttlMs) return null; // Touch on a hit so the hot users are the ones that survive eviction. this.#entries.delete(userId); this.#entries.set(userId, entry); @@ -70,30 +113,43 @@ export class SubscriptionCache { } /** - * Record an answer against the generation it was read under. A bump that - * landed while the read was in flight leaves the generations mismatched, - * and the answer is dropped rather than cached stale. + * Record an answer against the epoch it was read under. A bump that landed + * while the read was in flight leaves the epochs mismatched, and the answer + * is dropped rather than cached stale. */ - write(userId: number, generation: number, hasAny: boolean): void { + write(userId: number, epoch: number, hasAny: boolean): void { const entry = this.#entries.get(userId); - if (entry && entry.generation !== generation) return; - this.#set(userId, { generation, hasAny }); + if (entry && entry.epoch !== epoch) return; + this.#set(userId, { + epoch, + redisGeneration: entry?.redisGeneration ?? null, + hasAny, + cachedAt: Date.now(), + }); } /** - * Invalidate a user, moving them to `generation` when it is ahead of what - * this process has. Two bumps can arrive out of order — the counter is what - * orders them, so the later one cannot be undone by the earlier. + * Invalidate a user. With `generation`, this is a local subscribe or + * unsubscribe reporting the store's own new value: applied only when it is + * ahead of the last one this process recorded, so two racing local bumps + * can't land out of order. Without one — `invalidateUser`'s bare call — the + * epoch still advances unconditionally, because there is nothing to compare + * a number-less invalidation against; the previously recorded + * `redisGeneration`, if any, is left exactly as it was. */ bump(userId: number, generation?: number): void { - const current = this.#entries.get(userId)?.generation ?? 0; - if (generation === undefined) { - this.#set(userId, { generation: current + 1, hasAny: null }); - return; + const entry = this.#entries.get(userId); + if (generation !== undefined) { + const current = entry?.redisGeneration ?? 0; + // Already applied, or superseded by one that arrived first. + if (generation <= current) return; } - // Already applied, or superseded by one that arrived first. - if (generation <= current) return; - this.#set(userId, { generation, hasAny: null }); + this.#set(userId, { + epoch: (entry?.epoch ?? 0) + 1, + redisGeneration: generation ?? entry?.redisGeneration ?? null, + hasAny: null, + cachedAt: Date.now(), + }); } clear(): void { diff --git a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts new file mode 100644 index 000000000..5eb23d79c --- /dev/null +++ b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts @@ -0,0 +1,421 @@ +/* + * 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 . + */ + +/** + * The table and the region cache in front of it, against a real database. What + * is on the hook here is that the two never disagree — a row that exists is + * cached, a row that goes is uncached, and a region that has not looked knows + * it has not looked. + */ + +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; +import { v4 as uuidv4 } from 'uuid'; +import { EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER } from '../../controllers/events/limits.js'; +import { isHttpError } from '../../core/http/HttpError.js'; +import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; +import type { IConfig } from '../../types.js'; +import type { DurableSubscriptionInput } from './DurableSubscriptionStore.js'; + +const BOOT_TIMEOUT_MS = 120_000; + +let env: PuterTestEnv; +let userId: number; +let otherUserId: number; +let anchorUid: string; +let anchorPath: string; + +const durable = () => env.server.stores.durableSubscription; +const cache = () => env.server.stores.eventSubscription; + +const token = () => `f#${anchorUid}`; + +const input = ( + over: Partial = {}, +): DurableSubscriptionInput => ({ + holderUserId: userId, + ownerUserId: userId, + appUid: null, + subject: `fs:${anchorPath}`, + token: token(), + anchorUid, + anchorPath, + match: null, + op: null, + delivery: 'broadcast', + targets: ['socket', 'worker'], + handlerName: null, + context: null, + permission: 'list', + expiresAt: null, + ...over, +}); + +const codeOf = (code: string) => (err: unknown) => + isHttpError(err) && err.legacyCode === code; + +beforeAll(async () => { + env = await setupPuterTestEnv({ events: { enabled: true } } as IConfig); + const user = await env.server.stores.user.getByUsername( + env.users.user.username, + ); + userId = user!.id; + const other = await env.server.stores.user.getByUsername( + env.users.other.username, + ); + otherUserId = other!.id; + + anchorPath = `/${env.users.user.username}/durable-store`; + await env.server.services.fs.mkdir(userId, { + path: anchorPath, + createMissingParents: true, + }); + const entry = await env.server.stores.fsEntry.getEntryByPath(anchorPath); + anchorUid = entry!.uid; +}, BOOT_TIMEOUT_MS); + +afterAll(async () => { + await env?.shutdown(); +}); + +beforeEach(async () => { + await env.server.clients.db.write('DELETE FROM `event_subscriptions`', []); + await cache().markRegionCold(userId); + await cache().rebuildDurable(userId, []); +}); + +describe('creating a subscription', () => { + it('round-trips the row and names it after the app that made it', async () => { + const appUid = `app-${uuidv4()}`; + const { row } = await durable().create( + input({ appUid, handlerName: 'onWrite', match: '*.txt' }), + ); + + expect(row.subId.startsWith(`${appUid}#`)).toBe(true); + await expect(durable().getBySubId(row.subId)).resolves.toMatchObject({ + subId: row.subId, + appUid, + handlerName: 'onWrite', + match: '*.txt', + delivery: 'broadcast', + targets: ['socket', 'worker'], + durable: true, + }); + }); + + it('names a session`s row for the account, not an app', async () => { + const { row } = await durable().create(input()); + expect(row.subId.startsWith('user#')).toBe(true); + expect(row.appUid).toBeNull(); + }); + + it('caches the row and watches its token before returning', async () => { + const { row } = await durable().create(input()); + + await expect(cache().userHasAny(userId)).resolves.toBe(true); + await expect( + cache().watchedTokens(userId, [token()]), + ).resolves.toEqual([token()]); + await expect(cache().getForTokens(userId, [token()])).resolves.toEqual([ + row, + ]); + // The region has read the table, so nothing else has to. + await expect(cache().isRegionWarm(userId)).resolves.toBe(true); + }); + + it('indexes a shared anchor under its owner, not its subscriber', async () => { + const { row, bump } = await durable().create( + input({ holderUserId: otherUserId }), + ); + + expect(bump.userId).toBe(userId); + expect(row.holderUserId).toBe(otherUserId); + await expect(cache().userHasAny(otherUserId)).resolves.toBe(false); + await expect(cache().getForTokens(userId, [token()])).resolves.toEqual([ + row, + ]); + }); + + it('advances the owner`s generation', async () => { + const first = await durable().create(input()); + const second = await durable().create(input()); + expect(second.bump.generation).toBeGreaterThan(first.bump.generation); + }); +}); + +describe('validation at the row write', () => { + it('refuses a target outside the known set', async () => { + await expect( + durable().create( + input({ targets: ['socket', 'carrier-pigeon'] as never }), + ), + ).rejects.toSatisfy(codeOf('invalid_targets')); + }); + + it('refuses an empty target set', async () => { + await expect( + durable().create(input({ targets: [] })), + ).rejects.toSatisfy(codeOf('invalid_targets')); + }); + + it('refuses a context past the hard cap', async () => { + await expect( + durable().create(input({ context: 'x'.repeat(4097) })), + ).rejects.toSatisfy(codeOf('events_context_too_large')); + await expect(durable().countForHolder(userId)).resolves.toBe(0); + }); + + it('accepts a context right up to it', async () => { + const { row } = await durable().create( + input({ context: 'x'.repeat(4096) }), + ); + await expect(durable().getBySubId(row.subId)).resolves.toMatchObject({ + context: 'x'.repeat(4096), + }); + }); +}); + +describe('the per-account cap', () => { + it('refuses the one past the limit with a stable code', async () => { + const now = Math.floor(Date.now() / 1000); + for (let i = 0; i < EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER; i++) + await env.server.clients.db.insert('event_subscriptions', { + sub_id: `user#filler-${i}`, + token: token(), + owner_user_id: userId, + holder_user_id: userId, + app_uid: null, + subject: `fs:${anchorPath}`, + anchor_uid: anchorUid, + anchor_path: anchorPath, + match: null, + delivery: 'broadcast', + ops: null, + handler_name: null, + targets: '["socket"]', + context: null, + permission: 'list', + expires_at: null, + created_at: now, + }); + + await expect(durable().create(input())).rejects.toSatisfy( + codeOf('events_subscription_limit'), + ); + }); + + it('counts the holder, not the owner', async () => { + await durable().create(input({ holderUserId: otherUserId })); + await expect(durable().countForHolder(userId)).resolves.toBe(0); + await expect(durable().countForHolder(otherUserId)).resolves.toBe(1); + }); + + it('does not count a row that has expired but not yet been swept', async () => { + await durable().create( + input({ expiresAt: Math.floor(Date.now() / 1000) - 60 }), + ); + await durable().create(input()); + await expect(durable().countForHolder(userId)).resolves.toBe(1); + }); +}); + +describe('removal', () => { + it('takes the row out of the table and the cache together', async () => { + const { row } = await durable().create(input()); + + await durable().remove(row); + + await expect(durable().getBySubId(row.subId)).resolves.toBeNull(); + await expect(cache().getForTokens(userId, [token()])).resolves.toEqual( + [], + ); + await expect(cache().userHasAny(userId)).resolves.toBe(false); + }); + + it('leaves a sibling on the same anchor watched', async () => { + const { row } = await durable().create(input()); + const survivor = await durable().create(input()); + + await durable().remove(row); + + await expect( + cache().watchedTokens(userId, [token()]), + ).resolves.toEqual([token()]); + await expect(cache().getForTokens(userId, [token()])).resolves.toEqual([ + survivor.row, + ]); + }); +}); + +describe('the holder listing', () => { + const makeRows = async (count: number, appUid: string | null = null) => { + const rows = []; + for (let i = 0; i < count; i++) + rows.push((await durable().create(input({ appUid }))).row); + return rows; + }; + + it('pages with a cursor and stops offering one at the end', async () => { + const rows = await makeRows(5); + + const first = await durable().listForHolder(userId, { limit: 2 }); + expect(first.items.map((row) => row.subId)).toEqual( + rows.slice(0, 2).map((row) => row.subId), + ); + expect(first.cursor).toBeDefined(); + expect(first.total).toBeUndefined(); + + const second = await durable().listForHolder(userId, { + limit: 2, + cursor: first.cursor, + }); + expect(second.items.map((row) => row.subId)).toEqual( + rows.slice(2, 4).map((row) => row.subId), + ); + + const last = await durable().listForHolder(userId, { + limit: 2, + cursor: second.cursor, + }); + expect(last.items.map((row) => row.subId)).toEqual([rows[4].subId]); + expect(last.cursor).toBeUndefined(); + }); + + it('adds a total only when asked, over the scope and not the page', async () => { + await makeRows(3); + + const page = await durable().listForHolder(userId, { + limit: 1, + includeTotal: true, + }); + expect(page.items).toHaveLength(1); + expect(page.total).toBe(3); + }); + + it('confines an app to its own rows and shows a session everything', async () => { + const mine = `app-${uuidv4()}`; + const theirs = `app-${uuidv4()}`; + await makeRows(2, mine); + await makeRows(1, theirs); + await makeRows(1); + + const scoped = await durable().listForHolder(userId, { appUid: mine }); + expect(scoped.items).toHaveLength(2); + expect(scoped.items.every((row) => row.appUid === mine)).toBe(true); + + const account = await durable().listForHolder(userId, { + includeTotal: true, + }); + expect(account.total).toBe(4); + }); + + it('hides a row that has expired but not yet been swept', async () => { + const { row } = await durable().create( + input({ expiresAt: Math.floor(Date.now() / 1000) - 60 }), + ); + + const page = await durable().listForHolder(userId, { + includeTotal: true, + }); + expect(page.items).toEqual([]); + expect(page.total).toBe(0); + // Still on the table until the sweeper runs. + await expect(durable().getBySubId(row.subId)).resolves.not.toBeNull(); + }); + + it('returns an empty page for a holder with nothing', async () => { + await expect( + durable().listForHolder(otherUserId, { includeTotal: true }), + ).resolves.toEqual({ items: [], total: 0 }); + }); +}); + +describe('the expiry sweep', () => { + it('reaps an expired row and stops the region delivering against it', async () => { + const expired = await durable().create( + input({ expiresAt: Math.floor(Date.now() / 1000) - 1 }), + ); + const live = await durable().create(input()); + // The write-through caches only deliverable rows, so put the expired + // one back to prove the sweep is what removes it. + await cache().cacheDurable([expired.row]); + + await expect(durable().sweepExpired(500)).resolves.toBe(1); + + await expect(durable().getBySubId(expired.row.subId)).resolves.toBeNull(); + await expect(cache().getForTokens(userId, [token()])).resolves.toEqual([ + live.row, + ]); + }); + + it('leaves a subscription with no expiry alone', async () => { + await durable().create(input()); + await expect(durable().sweepExpired(500)).resolves.toBe(0); + await expect(durable().countForHolder(userId)).resolves.toBe(1); + }); +}); + +describe('warming a cold region', () => { + it('reads the table once and then answers from the cache', async () => { + const { row } = await durable().create(input()); + + // A region that has never seen this user: no keys, no marker. + await env.server.clients.redis.flushall(); + + await expect(durable().warmRegion(userId)).resolves.toBe(true); + await expect(cache().getForTokens(userId, [token()])).resolves.toEqual([ + row, + ]); + await expect(durable().warmRegion(userId)).resolves.toBe(false); + }); + + it('drops a row another region removed while this one was warm', async () => { + const { row } = await durable().create(input()); + await env.server.clients.db.write( + 'DELETE FROM `event_subscriptions` WHERE `sub_id` = ?', + [row.subId], + ); + + await cache().markRegionCold(userId); + await durable().warmRegion(userId); + + await expect(cache().getForTokens(userId, [token()])).resolves.toEqual( + [], + ); + await expect(cache().userHasAny(userId)).resolves.toBe(false); + }); + + it('does not cache a suspended row', async () => { + const { row } = await durable().create(input()); + await env.server.clients.db.write( + 'UPDATE `event_subscriptions` SET `suspended_at` = ?, ' + + '`suspended_reason` = ? WHERE `sub_id` = ?', + [Math.floor(Date.now() / 1000), 'permission_revoked', row.subId], + ); + + await cache().markRegionCold(userId); + await durable().warmRegion(userId); + + await expect(cache().getForTokens(userId, [token()])).resolves.toEqual( + [], + ); + // Suspended is still listed — it is a state, not a removal. + const page = await durable().listForHolder(userId); + expect(page.items[0]?.suspendedReason).toBe('permission_revoked'); + }); +}); diff --git a/src/backend/stores/events/DurableSubscriptionStore.ts b/src/backend/stores/events/DurableSubscriptionStore.ts new file mode 100644 index 000000000..c6ef533f7 --- /dev/null +++ b/src/backend/stores/events/DurableSubscriptionStore.ts @@ -0,0 +1,460 @@ +/* + * 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 { EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER } from '../../controllers/events/limits.js'; +import { HttpError } from '../../core/http/HttpError.js'; +import type { AclMode } from '../../services/acl/ACLService.js'; +import type { DeliveryClass } from '../../services/events/registry.js'; +import type { FsOp } from '../../services/events/subjects.js'; +import { + encodeCursor, + decodeCursor, + type PageResult, +} from '../../util/pagination.js'; +import { PuterStore } from '../types.js'; +import type { GenerationBump } from './EventSubscriptionStore.js'; +import { + isSubscriptionTarget, + SUBSCRIPTION_TARGETS, + type DurableSubscription, + type SubscriptionTarget, +} from './types.js'; + +/** + * Subscriptions that outlive the connection that made them. + * + * The table is the record; nothing on the dispatch path reads it directly. A + * subscribe is write-through — the row lands on the primary and in this + * region's cache before the call returns, so subscribe-then-write works here + * immediately — and every other region rebuilds lazily on its first dispatch + * for the owner. Rebuilds read the **primary** (`pread`): a replica that has + * not caught up would cache the absence of a subscription that exists, and no + * later event would correct it. + * + * Everything is keyed two ways, and the pair is the whole access story: + * `owner_user_id` is what dispatch rebuilds under, because a write only knows + * whose resource changed; `(holder_user_id, app_uid)` is what list, revoke and + * the quota use, and being the index it _is_ the scope check rather than a + * filter over a wider read. + */ + +// -- Limits ----------------------------------------------------------- + +/** Hard cap on the stored `context` blob. */ +export const DURABLE_CONTEXT_MAX_BYTES = 4096; + +/** Default page size for the holder listing, and its ceiling. */ +export const DURABLE_LIST_DEFAULT_LIMIT = 50; +export const DURABLE_LIST_LIMIT_CAP = 200; + +const TABLE = 'event_subscriptions'; + +// -- Wire shapes ------------------------------------------------------ + +/** Everything a caller must decide before a row can exist. */ +export interface DurableSubscriptionInput { + holderUserId: number; + ownerUserId: number; + appUid: string | null; + subject: string; + token: string; + anchorUid: string; + anchorPath: string; + match: string | null; + op: FsOp | null; + delivery: DeliveryClass; + targets: SubscriptionTarget[]; + handlerName: string | null; + context: string | null; + permission: AclMode; + expiresAt: number | null; +} + +export interface DurableListOptions { + /** Confines the listing to one app's rows; omitted is the account view. */ + appUid?: string | null; + limit?: number; + cursor?: string; + includeTotal?: boolean; +} + +// -- Errors ----------------------------------------------------------- + +const contextTooLarge = (): HttpError => + new HttpError( + 413, + `Subscription context may not exceed ${DURABLE_CONTEXT_MAX_BYTES} bytes`, + { legacyCode: 'events_context_too_large' }, + ); + +const invalidTargets = (): HttpError => + new HttpError( + 400, + `targets must be a subset of ${SUBSCRIPTION_TARGETS.join(', ')}`, + { legacyCode: 'invalid_targets' }, + ); + +const quotaReached = (): HttpError => + new HttpError( + 429, + `An account may hold ${EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER} durable subscriptions`, + { legacyCode: 'events_subscription_limit' }, + ); + +// -- Row mapping ------------------------------------------------------ + +const nowSeconds = (): number => Math.floor(Date.now() / 1000); + +const asNumber = (value: unknown): number | null => { + if (value === null || value === undefined) return null; + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : null; +}; + +/** + * `targets` is a JSON column on mysql and postgres and text on sqlite, so the + * driver hands back either an array or the string it was stored as. + */ +const parseTargets = (value: unknown): SubscriptionTarget[] => { + const raw = + typeof value === 'string' + ? (() => { + try { + return JSON.parse(value) as unknown; + } catch { + return []; + } + })() + : value; + if (!Array.isArray(raw)) return []; + return raw.filter(isSubscriptionTarget); +}; + +const toRow = (row: Record): DurableSubscription => ({ + durable: true, + subId: String(row.sub_id), + holderUserId: Number(row.holder_user_id), + ownerUserId: Number(row.owner_user_id), + subject: String(row.subject), + token: String(row.token), + anchorUid: String(row.anchor_uid), + anchorPath: String(row.anchor_path), + match: + row.match === null || row.match === undefined + ? null + : String(row.match), + op: row.ops ? (String(row.ops).split(',')[0] as FsOp) : null, + appUid: + row.app_uid === null || row.app_uid === undefined + ? null + : String(row.app_uid), + permission: String(row.permission) as AclMode, + delivery: String(row.delivery) as DeliveryClass, + targets: parseTargets(row.targets), + handlerName: + row.handler_name === null || row.handler_name === undefined + ? null + : String(row.handler_name), + context: + row.context === null || row.context === undefined + ? null + : String(row.context), + expiresAt: asNumber(row.expires_at), + suspendedAt: asNumber(row.suspended_at), + suspendedReason: + row.suspended_reason === null || row.suspended_reason === undefined + ? null + : String(row.suspended_reason), + createdAt: Number(row.created_at) || 0, +}); + +const SELECT_COLUMNS = + '`id`, `sub_id`, `token`, `owner_user_id`, `holder_user_id`, `app_uid`, ' + + '`subject`, `anchor_uid`, `anchor_path`, `match`, `delivery`, `ops`, ' + + '`handler_name`, `targets`, `context`, `permission`, `suspended_at`, ' + + '`suspended_reason`, `expires_at`, `created_at`'; + +export class DurableSubscriptionStore extends PuterStore { + // -- Writes ------------------------------------------------------ + + /** + * Insert one subscription and put it in this region's cache before + * returning. The generation bump is what tells every other region, and the + * caller broadcasts it. + */ + async create( + input: DurableSubscriptionInput, + ): Promise<{ row: DurableSubscription; bump: GenerationBump }> { + const targets = this.#assertTargets(input.targets); + this.#assertContext(input.context); + + const held = await this.countForHolder(input.holderUserId); + if (held >= EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER) throw quotaReached(); + + const row: DurableSubscription = { + durable: true, + subId: `${input.appUid ?? 'user'}#${randomUUID()}`, + holderUserId: input.holderUserId, + ownerUserId: input.ownerUserId, + subject: input.subject, + token: input.token, + anchorUid: input.anchorUid, + anchorPath: input.anchorPath, + match: input.match, + op: input.op, + appUid: input.appUid, + permission: input.permission, + delivery: input.delivery, + targets, + handlerName: input.handlerName, + context: input.context, + expiresAt: input.expiresAt, + suspendedAt: null, + suspendedReason: null, + createdAt: nowSeconds(), + }; + + await this.clients.db.insert(TABLE, { + sub_id: row.subId, + token: row.token, + owner_user_id: row.ownerUserId, + holder_user_id: row.holderUserId, + app_uid: row.appUid, + subject: row.subject, + anchor_uid: row.anchorUid, + anchor_path: row.anchorPath, + match: row.match, + delivery: row.delivery, + ops: row.op, + handler_name: row.handlerName, + targets: JSON.stringify(row.targets), + context: row.context, + permission: row.permission, + expires_at: row.expiresAt, + created_at: row.createdAt, + }); + + // Write-through, and over the whole owner rather than the one row: it + // costs the same indexed read as a warm would, on a path rate-limited + // to a few calls a minute, and it leaves this region needing nothing + // from the table before it can deliver. + await this.#rebuildRegion(row.ownerUserId); + return { row, bump: await this.#bump(row.ownerUserId) }; + } + + /** Remove one row and stop this region delivering against it. */ + async remove(row: DurableSubscription): Promise { + await this.clients.db.write( + `DELETE FROM \`${TABLE}\` WHERE \`sub_id\` = ?`, + [row.subId], + ); + await this.stores.eventSubscription.dropDurable(row); + return this.#bump(row.ownerUserId); + } + + /** + * Bring this region's cache for one owner up to date with the table, unless + * it already is. Returns whether the table was read, which is what the + * hot-path tests assert on. + */ + async warmRegion(ownerUserId: number): Promise { + if (await this.stores.eventSubscription.isRegionWarm(ownerUserId)) + return false; + await this.#rebuildRegion(ownerUserId); + return true; + } + + /** + * Reap rows past their expiry, dropping each from the cache so the region + * stops delivering against it without waiting for a rebuild. + */ + async sweepExpired(batchSize: number): Promise { + const rows = await this.#listExpired(nowSeconds(), batchSize); + if (rows.length === 0) return 0; + + await this.clients.db.write( + `DELETE FROM \`${TABLE}\` WHERE \`sub_id\` IN ` + + `(${rows.map(() => '?').join(', ')})`, + rows.map((row) => row.subId), + ); + + const owners = new Set(); + for (const row of rows) { + await this.stores.eventSubscription.dropDurable(row); + owners.add(row.ownerUserId); + } + for (const ownerUserId of owners) await this.#bump(ownerUserId); + return rows.length; + } + + // -- Reads ------------------------------------------------------- + + /** + * One row by id. Primary, because "not found" is how this surface answers a + * subscription that is not the caller's — and a replica behind by a moment + * would give that answer for a row the caller has only just created. + */ + async getBySubId(subId: string): Promise { + const rows = await this.clients.db.pread( + `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` WHERE \`sub_id\` = ?`, + [subId], + ); + return rows.length > 0 ? toRow(rows[0]) : null; + } + + /** + * What an actor may see, keyset-paginated on `id`. `appUid` is the scope: + * pass it for an app-context caller and the index answers the question, + * omit it for the account view that spans apps — including rows left behind + * by an app that has since been removed. + */ + async listForHolder( + holderUserId: number, + options: DurableListOptions = {}, + ): Promise> { + const limit = Math.min( + Math.max( + 1, + Math.floor(options.limit ?? DURABLE_LIST_DEFAULT_LIMIT), + ), + DURABLE_LIST_LIMIT_CAP, + ); + const after = asNumber(decodeCursor(options.cursor)?.id); + + // The scope half of the predicate is what the total counts over; the + // cursor half only positions one page inside it. + const scope = ['`holder_user_id` = ?', this.#unexpiredClause()]; + const scopeParams: unknown[] = [holderUserId, nowSeconds()]; + if (options.appUid !== undefined && options.appUid !== null) { + scope.push('`app_uid` = ?'); + scopeParams.push(options.appUid); + } + + const where = [...scope]; + const params = [...scopeParams]; + if (after !== null) { + where.push('`id` > ?'); + params.push(after); + } + + const rows = await this.clients.db.read( + `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` ` + + `WHERE ${where.join(' AND ')} ORDER BY \`id\` LIMIT ?`, + [...params, limit + 1], + ); + + const page = rows.slice(0, limit); + const result: PageResult = { + items: page.map(toRow), + }; + if (rows.length > limit) + result.cursor = encodeCursor({ + id: Number(page[page.length - 1].id), + }); + + if (options.includeTotal) { + const [count] = await this.clients.db.read( + `SELECT COUNT(*) AS \`total\` FROM \`${TABLE}\` ` + + `WHERE ${scope.join(' AND ')}`, + scopeParams, + ); + result.total = Number(count?.total ?? 0); + } + return result; + } + + /** + * Quota counting, over the same index the listing uses. Primary: a count + * read off a lagging replica is a cap a burst of subscribes walks straight + * through. + */ + async countForHolder(holderUserId: number): Promise { + const [row] = await this.clients.db.pread( + `SELECT COUNT(*) AS \`total\` FROM \`${TABLE}\` ` + + `WHERE \`holder_user_id\` = ? AND ${this.#unexpiredClause()}`, + [holderUserId, nowSeconds()], + ); + return Number(row?.total ?? 0); + } + + /** + * Every row a region has to be able to deliver for one owner. Read from the + * primary: this is what a cold region caches, and caching a replica's "no + * rows yet" would silence a subscription with nothing to correct it. + */ + async listDeliverableForOwner( + ownerUserId: number, + ): Promise { + const rows = await this.clients.db.pread( + `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` ` + + 'WHERE `owner_user_id` = ? AND `suspended_at` IS NULL AND ' + + `${this.#unexpiredClause()}`, + [ownerUserId, nowSeconds()], + ); + return rows.map(toRow); + } + + // -- Internals --------------------------------------------------- + + async #rebuildRegion(ownerUserId: number): Promise { + const rows = await this.listDeliverableForOwner(ownerUserId); + await this.stores.eventSubscription.rebuildDurable(ownerUserId, rows); + } + + /** `?` binds the cutoff, so no clock crosses the wire as SQL. */ + #unexpiredClause(): string { + return '(`expires_at` IS NULL OR `expires_at` > ?)'; + } + + async #listExpired( + cutoff: number, + batchSize: number, + ): Promise { + const limit = Math.max(1, Math.floor(batchSize)); + const rows = await this.clients.db.read( + `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` ` + + 'WHERE `expires_at` IS NOT NULL AND `expires_at` <= ? ' + + 'ORDER BY `id` LIMIT ?', + [cutoff, limit], + ); + return rows.map(toRow); + } + + #assertTargets(targets: readonly string[]): SubscriptionTarget[] { + if (!Array.isArray(targets) || targets.length === 0) + throw invalidTargets(); + if (!targets.every(isSubscriptionTarget)) throw invalidTargets(); + return [...new Set(targets as SubscriptionTarget[])]; + } + + #assertContext(context: string | null): void { + if (context === null) return; + if (Buffer.byteLength(context, 'utf8') > DURABLE_CONTEXT_MAX_BYTES) + throw contextTooLarge(); + } + + async #bump(ownerUserId: number): Promise { + return { + userId: ownerUserId, + generation: + await this.stores.eventSubscription.bumpGeneration(ownerUserId), + }; + } +} diff --git a/src/backend/stores/events/EventSubscriptionStore.ts b/src/backend/stores/events/EventSubscriptionStore.ts index 2247ed83f..3203145ce 100644 --- a/src/backend/stores/events/EventSubscriptionStore.ts +++ b/src/backend/stores/events/EventSubscriptionStore.ts @@ -19,14 +19,19 @@ import { EVENTS_SESSION_SUBSCRIPTIONS_PER_SOCKET } from '../../controllers/events/limits.js'; import { HttpError } from '../../core/http/HttpError.js'; -import type { FsOp } from '../../services/events/subjects.js'; -import type { AclMode } from '../../services/acl/ACLService.js'; import { PuterStore } from '../types.js'; +import type { + DispatchSubscription, + DurableSubscription, + GenerationBump, + SessionSubscription, +} from './types.js'; /** - * Session subscriptions: Redis only, keyed to the socket that holds them, gone - * when it disconnects. Nothing here outlives a connection, so none of it - * belongs in a table. + * The region's subscription keyspace. Session rows live here and nowhere else — + * keyed to the socket that holds them, gone when it disconnects — and durable + * rows are cached here over the table that owns them, so dispatch reads one + * place whichever kind answered. * * Rows are indexed by the **owner of the anchor node**, not by the subscriber: * dispatch knows only whose resource changed, and a subscription on a folder @@ -42,6 +47,8 @@ import { PuterStore } from '../types.js'; * ev:t:{}: HASH subId -> row, for one watched token * ev:s:{}: SET what this socket holds, for reaping * ev:g:{} STR subscription-set generation + * ev:dm:{} HASH subId -> token, durable rows cached here + * ev:dw:{} STR this region's durable cache is warm * * The socket set is the one keyed by the holder — it is read on disconnect, * when all that is known is whose connection went — so its members name the @@ -56,37 +63,21 @@ import { PuterStore } from '../types.js'; * disconnect handler leaves keys behind, and the TTL is what collects them — a * live socket refreshes its own, so the backstop only ever fires on rows whose * socket is gone. + * + * Durable rows share the row hash and the watched set rather than getting their + * own: `ev:w` is the one key on the hot path, and a token still wanted by a + * durable row has to stay in it when a session row on the same anchor goes — + * which the existing "drop the token once its hash is empty" rule gets right + * for free. What durable rows add is the warm marker, which is how a region + * tells "nobody is subscribed" apart from "this region has not looked yet". */ -// -- Types ------------------------------------------------------------ - -export interface SessionSubscription { - subId: string; - socketId: string; - /** Who subscribed: the delivery target, and whose access is re-checked. */ - holderUserId: number; - /** Owner of the anchor node: the keyspace this row is indexed in. */ - ownerUserId: number; - /** The subject as the client asked for it. */ - subject: string; - /** Anchor token the row is indexed under. */ - token: string; - anchorUid: string; - anchorPath: string; - /** Glob relative to the anchor, or `null` for a node-form subscription. */ - match: string | null; - op: FsOp | null; - /** The app that created the row, and the scope of the three verbs. */ - appUid: string | null; - /** ACL mode the subscribe check passed under; re-checked per delivery. */ - permission: AclMode; -} - -/** One owner's generation after a change to the set of rows keyed under them. */ -export interface GenerationBump { - userId: number; - generation: number; -} +export type { + DispatchSubscription, + DurableSubscription, + GenerationBump, + SessionSubscription, +} from './types.js'; // -- Keys ------------------------------------------------------------- @@ -96,6 +87,8 @@ const tokenKey = (userId: number | string, token: string): string => const socketKey = (userId: number | string, socketId: string): string => `ev:s:{${userId}}:${socketId}`; const generationKey = (userId: number | string): string => `ev:g:{${userId}}`; +const durableMapKey = (userId: number | string): string => `ev:dm:{${userId}}`; +const durableWarmKey = (userId: number | string): string => `ev:dw:{${userId}}`; /** `ev:s` members name the row they point at, and the keyspace it is in. */ interface SocketRef { @@ -144,6 +137,17 @@ export const SESSION_SUBSCRIPTION_TTL_SECONDS = 60 * 60; */ const GENERATION_TTL_SECONDS = 24 * 60 * 60; +/** How long a cached durable row stays readable without being rebuilt. */ +export const DURABLE_CACHE_TTL_SECONDS = 24 * 60 * 60; + +/** + * How long a region trusts its durable cache before reading the table again. + * Strictly under the session TTL, because a session refresh re-expires the + * shared keys down to that — so the marker always lapses first and a rebuild + * always precedes a row quietly expiring out from under it. + */ +export const DURABLE_WARM_TTL_SECONDS = 30 * 60; + const subscriptionLimitReached = (): HttpError => new HttpError( 429, @@ -240,12 +244,7 @@ export class EventSubscriptionStore extends PuterStore { return bumps; } - /** - * Remove rows and then any token whose rows are all gone. The token leaves - * the watched set only once its hash is empty, which is what keeps one - * socket's unsubscribe from silencing another's subscription on the same - * anchor. - */ + /** Forget a socket's refs, then the rows they point at. */ async #dropRefs( holderUserId: number, socketId: string, @@ -256,27 +255,37 @@ export class EventSubscriptionStore extends PuterStore { ...refs.map(socketRef), ); - for (const [ownerUserId, owned] of byOwner(refs)) { - const drop = this.clients.redis.pipeline(); - for (const { token, subId } of owned) - drop.hdel(tokenKey(ownerUserId, token), subId); - await drop.exec(); + for (const [ownerUserId, owned] of byOwner(refs)) + await this.#dropRows(ownerUserId, owned); + } - const tokens = [...new Set(owned.map((ref) => ref.token))]; - const counts = this.clients.redis.pipeline(); - for (const token of tokens) - counts.hlen(tokenKey(ownerUserId, token)); - const results = (await counts.exec()) ?? []; + /** + * Drop rows from one owner's keyspace and then any token whose rows are all + * gone. Emptiness is what un-watches a token, which is what keeps one + * socket's unsubscribe — or a durable row's removal — from silencing + * another subscription on the same anchor. + */ + async #dropRows( + ownerUserId: number, + rows: ReadonlyArray<{ token: string; subId: string }>, + ): Promise { + if (rows.length === 0) return; - const orphaned = tokens.filter( - (_token, i) => Number(results[i]?.[1] ?? 0) === 0, - ); - if (orphaned.length > 0) - await this.clients.redis.srem( - watchedKey(ownerUserId), - ...orphaned, - ); - } + const drop = this.clients.redis.pipeline(); + for (const { token, subId } of rows) + drop.hdel(tokenKey(ownerUserId, token), subId); + await drop.exec(); + + const tokens = [...new Set(rows.map((row) => row.token))]; + const counts = this.clients.redis.pipeline(); + for (const token of tokens) counts.hlen(tokenKey(ownerUserId, token)); + const results = (await counts.exec()) ?? []; + + const orphaned = tokens.filter( + (_token, i) => Number(results[i]?.[1] ?? 0) === 0, + ); + if (orphaned.length > 0) + await this.clients.redis.srem(watchedKey(ownerUserId), ...orphaned); } /** @@ -308,6 +317,96 @@ export class EventSubscriptionStore extends PuterStore { } } + // -- Durable rows in the region cache ---------------------------- + + /** + * Cache durable rows so dispatch finds them without the table. Ordering + * matches `add`: rows land before their tokens join the watched set. + */ + async cacheDurable(rows: readonly DurableSubscription[]): Promise { + if (rows.length === 0) return; + const ownerUserId = rows[0].ownerUserId; + + const write = this.clients.redis.pipeline(); + for (const row of rows) { + const key = tokenKey(ownerUserId, row.token); + write.hset(key, row.subId, JSON.stringify(row)); + write.expire(key, DURABLE_CACHE_TTL_SECONDS); + write.hset(durableMapKey(ownerUserId), row.subId, row.token); + } + write.sadd(watchedKey(ownerUserId), ...rows.map((row) => row.token)); + write.expire(watchedKey(ownerUserId), DURABLE_CACHE_TTL_SECONDS); + write.expire(durableMapKey(ownerUserId), DURABLE_CACHE_TTL_SECONDS); + await write.exec(); + } + + /** Forget one cached durable row, un-watching its token if it was the last. */ + async dropDurable(row: { + ownerUserId: number; + token: string; + subId: string; + }): Promise { + await this.clients.redis.hdel( + durableMapKey(row.ownerUserId), + row.subId, + ); + await this.#dropRows(row.ownerUserId, [ + { token: row.token, subId: row.subId }, + ]); + } + + /** + * Replace everything this region has cached for one owner. Rows that are no + * longer in the table go, which is how an unsubscribe taken in another + * region eventually stops delivering here. + */ + async rebuildDurable( + ownerUserId: number, + rows: readonly DurableSubscription[], + ): Promise { + const cached = await this.clients.redis.hgetall( + durableMapKey(ownerUserId), + ); + const fresh = new Set(rows.map((row) => row.subId)); + const stale = Object.entries(cached ?? {}) + .filter(([subId]) => !fresh.has(subId)) + .map(([subId, token]) => ({ subId, token: String(token) })); + + if (stale.length > 0) { + await this.clients.redis.hdel( + durableMapKey(ownerUserId), + ...stale.map((row) => row.subId), + ); + await this.#dropRows(ownerUserId, stale); + } + await this.cacheDurable(rows); + await this.markRegionWarm(ownerUserId); + } + + /** Whether this region has read the table for this owner recently. */ + async isRegionWarm(ownerUserId: number): Promise { + return ( + (await this.clients.redis.exists(durableWarmKey(ownerUserId))) === 1 + ); + } + + async markRegionWarm(ownerUserId: number): Promise { + await this.clients.redis.set( + durableWarmKey(ownerUserId), + '1', + 'EX', + DURABLE_WARM_TTL_SECONDS, + ); + } + + /** + * Force the next dispatch in this region to read the table again. What a + * generation bump from anywhere lands on. + */ + async markRegionCold(ownerUserId: number): Promise { + await this.clients.redis.del(durableWarmKey(ownerUserId)); + } + // -- Reads ------------------------------------------------------- /** @@ -334,22 +433,22 @@ export class EventSubscriptionStore extends PuterStore { return tokens.filter((_token, i) => Number(flags[i]) === 1); } - /** The rows behind a set of watched tokens. */ + /** The rows behind a set of watched tokens, session and durable alike. */ async getForTokens( ownerUserId: number, tokens: readonly string[], - ): Promise { + ): Promise { if (tokens.length === 0) return []; const pipeline = this.clients.redis.pipeline(); for (const token of tokens) pipeline.hvals(tokenKey(ownerUserId, token)); const results = (await pipeline.exec()) ?? []; - const subs: SessionSubscription[] = []; + const subs: DispatchSubscription[] = []; for (const [, raw] of results) { for (const row of (raw as string[] | null) ?? []) { try { - subs.push(JSON.parse(row) as SessionSubscription); + subs.push(JSON.parse(row) as DispatchSubscription); } catch { // A row we cannot read is a row we cannot deliver against. } @@ -374,7 +473,12 @@ export class EventSubscriptionStore extends PuterStore { const rows = await this.getForTokens(ownerUserId, [ ...new Set(owned.map((ref) => ref.token)), ]); - held.push(...rows.filter((row) => wanted.has(row.subId))); + held.push( + ...rows.filter( + (row): row is SessionSubscription => + wanted.has(row.subId) && row.socketId !== undefined, + ), + ); } return held; } diff --git a/src/backend/stores/events/types.ts b/src/backend/stores/events/types.ts new file mode 100644 index 000000000..019bf01bf --- /dev/null +++ b/src/backend/stores/events/types.ts @@ -0,0 +1,94 @@ +/* + * 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 type { AclMode } from '../../services/acl/ACLService.js'; +import type { DeliveryClass } from '../../services/events/registry.js'; +import type { FsOp } from '../../services/events/subjects.js'; + +/** + * Row shapes shared by the two subscription stores. Session rows live in Redis + * and die with their connection; durable rows live in a table and are cached + * into the same Redis keyspace. Dispatch reads both out of one hash, matches + * them with one matcher and re-checks them with one ACL call, so what it reads + * is declared once here rather than per store. + */ + +/** Transports a delivery may take. */ +export const SUBSCRIPTION_TARGETS = ['socket', 'worker', 'push'] as const; + +export type SubscriptionTarget = (typeof SUBSCRIPTION_TARGETS)[number]; + +export const isSubscriptionTarget = ( + value: unknown, +): value is SubscriptionTarget => + SUBSCRIPTION_TARGETS.includes(value as SubscriptionTarget); + +/** What dispatch needs from a subscription, whichever store it came from. */ +export interface DispatchSubscription { + subId: string; + /** Who subscribed: the delivery target, and whose access is re-checked. */ + holderUserId: number; + /** Owner of the anchor node: the keyspace this row is indexed in. */ + ownerUserId: number; + /** The subject as the client asked for it. */ + subject: string; + /** Anchor token the row is indexed under. */ + token: string; + anchorUid: string; + anchorPath: string; + /** Glob relative to the anchor, or `null` for a node-form subscription. */ + match: string | null; + op: FsOp | null; + /** The app that created the row, and the scope of the three verbs. */ + appUid: string | null; + /** ACL mode the subscribe check passed under; re-checked per delivery. */ + permission: AclMode; + /** Session rows only: the connection a delivery is addressed at. */ + socketId?: string; + /** Durable rows only: set on every row that outlives its connection. */ + durable?: true; + delivery?: DeliveryClass; + targets?: SubscriptionTarget[]; + handlerName?: string | null; +} + +export interface SessionSubscription extends DispatchSubscription { + socketId: string; +} + +/** A row in `event_subscriptions`, as the rest of the system sees it. */ +export interface DurableSubscription extends DispatchSubscription { + durable: true; + delivery: DeliveryClass; + targets: SubscriptionTarget[]; + handlerName: string | null; + /** Plaintext JSON, capped at 4 KB. Never returned by `list`. */ + context: string | null; + /** Unix seconds, or `null` for a subscription with no end. */ + expiresAt: number | null; + suspendedAt: number | null; + suspendedReason: string | null; + createdAt: number; +} + +/** One owner's generation after a change to the set of rows keyed under them. */ +export interface GenerationBump { + userId: number; + generation: number; +} diff --git a/src/backend/stores/index.ts b/src/backend/stores/index.ts index bb3ef3c49..4bcb1a738 100644 --- a/src/backend/stores/index.ts +++ b/src/backend/stores/index.ts @@ -21,6 +21,7 @@ import { AppFeedbackStore } from './appFeedback/AppFeedbackStore.js'; import { AppStore } from './app/AppStore.js'; import { FSEntryStore } from './fs/FSEntryStore.js'; import { GroupStore } from './group/GroupStore.js'; +import { DurableSubscriptionStore } from './events/DurableSubscriptionStore.js'; import { EventSubscriptionStore } from './events/EventSubscriptionStore.js'; import { CreditHoldStore } from './metering/CreditHoldStore.js'; import { MeteringBufferStore } from './metering/MeteringBufferStore.js'; @@ -62,6 +63,7 @@ declare module './types.js' { oidc: OIDCStore; userBlock: UserBlockStore; eventSubscription: EventSubscriptionStore; + durableSubscription: DurableSubscriptionStore; } } @@ -93,4 +95,6 @@ export const puterStores = { userBlock: UserBlockStore, // Redis only, no peer stores. eventSubscription: EventSubscriptionStore, + // Writes through the Redis keyspace above, so it comes after it. + durableSubscription: DurableSubscriptionStore, } satisfies IPuterStoreRegistry; diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md index 462e84eff..b8465dc68 100644 --- a/src/docs/src/rate-limits-and-quotas.md +++ b/src/docs/src/rate-limits-and-quotas.md @@ -162,12 +162,18 @@ One write can reach many subscriptions, so events are bounded on both halves: ho | Limit | All accounts | | -------------------------------------------- | ------------ | | Subscriptions per connection | 50 | +| Durable subscriptions per account | 500 | | `subscribe` / `unsubscribe` calls per minute | 60 | +| Subscription listings per minute | 120 | | Matched subscriptions per event | 50 | | Filter evaluations per event | 200 | | Deliveries per minute, per subscription | 600 | -Subscriptions live with the connection that made them: they are dropped when it closes, and a reconnecting client subscribes again. The 51st subscription on one connection fails with `events_subscription_limit`; over the call budget, `subscribe` and `unsubscribe` fail with `too_many_requests`. Subscribing to something you cannot read fails with `subject_does_not_exist` — the same answer as subscribing to something that is not there, so the call cannot be used to find out which. +Subscriptions come in two kinds. A **session** subscription lives with the connection that made it: it is dropped when the connection closes, and a reconnecting client subscribes again. A **durable** subscription outlives every connection — it is created over the API, listed and revoked from the account, and keeps delivering until you remove it or it expires. + +The 51st subscription on one connection, and the 501st durable subscription on one account, both fail with `events_subscription_limit`. Over the call budget, `subscribe` and `unsubscribe` fail with `too_many_requests`. Subscribing to something you cannot read fails with `subject_does_not_exist` — the same answer as subscribing to something that is not there, so the call cannot be used to find out which. + +A durable subscription may carry a `context`: JSON that is stored with it and handed to its handler on every delivery, capped at **4 KB** and rejected over that with `events_context_too_large`. Listings never return it. An app sees and revokes only the subscriptions it created; a session acting for the account sees them all, including ones left behind by an app that has since been removed. Match patterns are compiled once when you subscribe and are capped at **256 characters** and **16 segments**; anything larger is rejected with `invalid_subject_pattern`. `**` crosses directories and costs no more than `*`.