From 41b837b398ce0014b7409bc343cb04428aed626a Mon Sep 17 00:00:00 2001 From: Daniel Salazar Date: Thu, 3 Sep 2026 01:39:06 -0700 Subject: [PATCH] feat: background delivery seam, retries, and consent (PUT-1682) (#3684) --- .../events/EventsWorkerInvokerClient.ts | 197 +++++++ src/backend/clients/index.ts | 2 + src/backend/controllers/events/limits.ts | 35 ++ src/backend/data/hardcoded-permissions.js | 4 + src/backend/services/events/EventsService.ts | 245 ++++++++- src/backend/services/events/authorization.ts | 48 ++ .../events/durable.integration.test.ts | 64 ++- .../events/handlers.integration.test.ts | 27 + .../services/events/kv.integration.test.ts | 8 + src/backend/services/events/registry.ts | 4 +- .../revocationSettle.integration.test.ts | 35 ++ .../services/events/singleDelivery.test.ts | 54 ++ .../events/workerInvoker.integration.test.ts | 486 ++++++++++++++++++ src/backend/services/events/workerSeam.ts | 90 +++- ...rableSubscriptionStore.integration.test.ts | 14 +- .../stores/events/DurableSubscriptionStore.ts | 5 +- .../events/PendingDeliveryStore.test.ts | 69 +++ .../stores/events/PendingDeliveryStore.ts | 89 +++- src/backend/stores/events/types.ts | 14 +- src/backend/types.ts | 2 + src/docs/src/Events.md | 8 +- src/docs/src/Events/onPersistent.md | 36 ++ src/docs/src/rate-limits-and-quotas.md | 6 + src/gui/src/UI/UIPermissionDialog.js | 4 + src/gui/src/i18n/translations/en.js | 1 + .../src/modules/events/events.test.js | 117 +++++ .../src/modules/events/lib/channel.js | 162 +++++- .../src/modules/events/onPersistent.js | 27 +- .../src/modules/events/persistent.test.js | 57 ++ src/puter-js/src/modules/events/types.js | 19 +- src/puter-js/tests/api/suites/events.suite.ts | 199 ++++++- 31 files changed, 2057 insertions(+), 71 deletions(-) create mode 100644 src/backend/clients/events/EventsWorkerInvokerClient.ts create mode 100644 src/backend/services/events/workerInvoker.integration.test.ts diff --git a/src/backend/clients/events/EventsWorkerInvokerClient.ts b/src/backend/clients/events/EventsWorkerInvokerClient.ts new file mode 100644 index 000000000..4b4aaba0d --- /dev/null +++ b/src/backend/clients/events/EventsWorkerInvokerClient.ts @@ -0,0 +1,197 @@ +/* + * 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 . + */ + +// Claims the global dispatcher slot for Node's own fetch before npm `undici` +// can — see the guard's own comment. +import '../../util/nodeFetchDispatcherGuard.js'; + +import { Agent, fetch as undiciFetch } from 'undici'; +import { PuterClient } from '../types.js'; + +/** + * The call that leaves the platform and runs an app's own code. + * + * The protocol is fixed and lives here rather than in the events service + * because it is a wire format, not a delivery decision: one POST, one header, + * and a status code that says whether the handler took the delivery, refused + * it, or could not answer. + * + * POST /__events/invoke + * puter-auth: + * { handler, event, ctx } + * + * The token is minted above this layer and passed in: what a delivery is + * allowed to do is an authorization question, and this only carries the + * answer. + */ + +/** The one route an events worker reserves for the platform. */ +export const EVENTS_INVOKE_PATH = '/__events/invoke'; + +/** How long a handler has to answer before the attempt is abandoned. */ +export const EVENTS_INVOKE_TIMEOUT_MS = 30_000; + +/** Connections held open per worker host, so a busy app is not re-handshaking. */ +const INVOKE_CONNECTIONS = 64; +const INVOKE_KEEP_ALIVE_MS = 30_000; + +/** + * Where an app's events worker can be reached. + * + * Building and deploying that worker is a separate concern, so this is a seam: + * the default resolves nothing, and the deployment that owns worker addressing + * registers an implementation. A `null` answer means the app has no events + * worker to invoke. + */ +export interface EventsWorkerResolver { + resolveInvokeUrl(appUid: string): Promise; +} + +/** The stand-in until something can address an app's events worker. */ +export class UnresolvedEventsWorkerResolver implements EventsWorkerResolver { + resolveInvokeUrl(): Promise { + return Promise.resolve(null); + } +} + +/** + * What the call did. + * + * - `settled` — the handler took the delivery (2xx). + * - `terminal` — it refused it (4xx). Retrying sends the same body to the same + * code, so it is not retried. + * - `retriable` — it could not answer: 5xx, 429, a timeout, a transport failure, + * or no worker to address at all. + */ +export type WorkerInvokeOutcome = 'settled' | 'terminal' | 'retriable'; + +export interface WorkerInvokeRequest { + appUid: string; + handler: string; + /** Subscriber-scoped access token, sent in `puter-auth`. */ + token: string; + event: unknown; + ctx: unknown; +} + +export interface WorkerInvokeResult { + outcome: WorkerInvokeOutcome; + /** The status the handler answered with, or `null` when it never did. */ + status: number | null; + /** Why it could not answer, when nothing did. */ + error?: string; +} + +const RESOLVER_UNAVAILABLE = 'no events worker is deployed for this app'; + +export class EventsWorkerInvokerClient extends PuterClient { + /** Replaced by whatever owns worker addressing on this deployment. */ + #resolver: EventsWorkerResolver = new UnresolvedEventsWorkerResolver(); + #agent: Agent | null = null; + + setResolver(resolver: EventsWorkerResolver): void { + this.#resolver = resolver; + } + + override onServerShutdown(): void { + void this.#agent?.close().catch(() => {}); + this.#agent = null; + } + + async invoke(request: WorkerInvokeRequest): Promise { + let url: string | null = null; + try { + url = await this.#resolver.resolveInvokeUrl(request.appUid); + } catch (err) { + return { + outcome: 'retriable', + status: null, + error: err instanceof Error ? err.message : String(err), + }; + } + // Nothing to call yet. Retriable rather than terminal: the address is + // the platform's to provide, and an app whose worker is not there has + // done nothing wrong — the consecutive-failure rule is what stops this + // from retrying forever. + if (!url) + return { + outcome: 'retriable', + status: null, + error: RESOLVER_UNAVAILABLE, + }; + + try { + const response = await undiciFetch( + new URL(EVENTS_INVOKE_PATH, url).toString(), + { + method: 'POST', + headers: { + 'content-type': 'application/json', + 'puter-auth': request.token, + }, + body: JSON.stringify({ + handler: request.handler, + event: request.event, + ctx: request.ctx, + }), + signal: AbortSignal.timeout(this.#timeoutMs()), + dispatcher: this.#dispatcher(), + }, + ); + // The body is not read: a handler answers with a status, and + // whatever else it writes is its own business. + void response.body?.cancel().catch(() => {}); + return { + outcome: outcomeForStatus(response.status), + status: response.status, + }; + } catch (err) { + return { + outcome: 'retriable', + status: null, + error: err instanceof Error ? err.message : String(err), + }; + } + } + + #timeoutMs(): number { + const configured = this.config.events?.invokeTimeoutMs; + return typeof configured === 'number' && configured > 0 + ? configured + : EVENTS_INVOKE_TIMEOUT_MS; + } + + /** One pool per process, built on first use and closed with the server. */ + #dispatcher(): Agent { + this.#agent ??= new Agent({ + connections: INVOKE_CONNECTIONS, + keepAliveTimeout: INVOKE_KEEP_ALIVE_MS, + keepAliveMaxTimeout: INVOKE_KEEP_ALIVE_MS, + }); + return this.#agent; + } +} + +/** 2xx took it, 429 is "not now", any other 4xx is a refusal. */ +export const outcomeForStatus = (status: number): WorkerInvokeOutcome => { + if (status >= 200 && status < 300) return 'settled'; + if (status === 429) return 'retriable'; + if (status >= 400 && status < 500) return 'terminal'; + return 'retriable'; +}; diff --git a/src/backend/clients/index.ts b/src/backend/clients/index.ts index aa6de932a..2fc257a37 100644 --- a/src/backend/clients/index.ts +++ b/src/backend/clients/index.ts @@ -21,6 +21,7 @@ import { AlarmClient } from './alarm/AlarmClient'; import { DatabaseClientFactory } from './database'; import { EmailClient } from './email/EmailClient'; import { EventClient } from './event/EventClient'; +import { EventsWorkerInvokerClient } from './events/EventsWorkerInvokerClient'; import { DDBClient } from './dynamodb/DDBClient'; import { RedisClient } from './redis/RedisClient'; import { S3Client } from './s3/S3Client'; @@ -32,6 +33,7 @@ export const puterClients = { db: DatabaseClientFactory, email: EmailClient, event: EventClient, + eventsWorkerInvoker: EventsWorkerInvokerClient, dynamo: DDBClient, redis: RedisClient, s3: S3Client, diff --git a/src/backend/controllers/events/limits.ts b/src/backend/controllers/events/limits.ts index 6ec9ba37e..294ba5b25 100644 --- a/src/backend/controllers/events/limits.ts +++ b/src/backend/controllers/events/limits.ts @@ -183,6 +183,41 @@ export const EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION = 10_000; */ export const EVENTS_REGION_PENDING_CEILING = 1_000_000; +// -- Handler retries ------------------------------------------------- +// +// A handler that answers "not now" — a 5xx, a timeout, a 429 — is retried, and +// a handler that answers "no" is not. Retrying on a fixed cadence turns one +// broken deploy into a permanent load on whatever is failing, so the wait +// doubles per attempt up to a ceiling, and a run of failures stops the +// subscription rather than retrying it forever. + +/** Wait before the first retry of a delivery a handler could not take. */ +export const EVENTS_RETRY_BASE_MS = 2_000; + +/** Longest wait between retries, however many have failed. */ +export const EVENTS_RETRY_MAX_MS = 5 * 60 * 1000; + +/** + * Failures in a row before a subscription is suspended. Counted per + * subscription and reset by the first delivery a handler takes, so an + * occasional failure never accumulates into one. + */ +export const EVENTS_CONSECUTIVE_FAILURES = 5; + +/** + * How long a run of failures is remembered. Five failures at the capped wait + * span well under this, so a counter nothing has touched for an hour describes + * a run that ended. + */ +export const EVENTS_FAILURE_COUNTER_TTL_MS = 60 * 60 * 1000; + +/** How long to hold a delivery whose handler has failed `attempts` times. */ +export const deliveryBackoffMs = (attempts: number): number => + Math.min( + EVENTS_RETRY_MAX_MS, + EVENTS_RETRY_BASE_MS * 2 ** Math.max(0, attempts - 1), + ); + // -- Suspended backlog ----------------------------------------------- // // A suspended subscription stops metering but keeps what it is owed, and that diff --git a/src/backend/data/hardcoded-permissions.js b/src/backend/data/hardcoded-permissions.js index 1f0b81b4a..a0c5d4a8c 100644 --- a/src/backend/data/hardcoded-permissions.js +++ b/src/backend/data/hardcoded-permissions.js @@ -100,6 +100,10 @@ const implicit_user_app_permissions = [ const default_user_permissions = { driver: {}, service: {}, + // Every account may have events delivered to an app's handler while nobody + // is there; which app may is the per-app grant, which is what the user is + // actually asked about. + 'events:background': {}, }; module.exports = { diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index 308941b99..336b9a388 100644 --- a/src/backend/services/events/EventsService.ts +++ b/src/backend/services/events/EventsService.ts @@ -23,6 +23,7 @@ import { EVENTS_ACK_LIMIT, EVENTS_BROADCAST_DELIVERY_LIMIT, EVENTS_COALESCE_WINDOW_MS, + EVENTS_CONSECUTIVE_FAILURES, EVENTS_HANDLER_PUBLISH_BATCH, EVENTS_HANDLER_PUBLISH_LIMIT, EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT, @@ -82,13 +83,17 @@ import { import { assertCrossAppKvAuthorized, assertSubscribeAuthorized, + backgroundConsentRequired, checkDeliveryAuthorized, crossAppKvDenial, crossAppKvPermissions, deliveryGenerationTag, + EVENTS_BACKGROUND_PERMISSION, + needsBackgroundConsent, nodeDescriptor, resolveGrantActor, rowInActorScope, + subscriptionTokenPermissions, SUBSCRIBE_MODE, type CrossAppKvDeps, type EventAclDeps, @@ -130,8 +135,10 @@ import { } from './subjects.js'; import { backlogPolicyFor, isResumable } from './suspension.js'; import { + EventsWorkerInvoker, RecordingWorkerInvoker, type WorkerInvocation, + type WorkerInvocationOutcome, type WorkerInvokerSeam, } from './workerSeam.js'; @@ -404,6 +411,13 @@ const PENDING_SWEEP_SUBSCRIPTIONS = 100; */ const PENDING_DRAIN_BATCH = 25; +/** + * How long the token an invocation carries is good for. Long enough for a + * handler doing real work with the account's own data, short enough that a copy + * of it is worth nothing by the time anyone finds it. + */ +const DELIVERY_TOKEN_TTL = '5m'; + /** The part of a socket this service uses, so tests need not build one. */ export interface EventSocket { id: string; @@ -624,9 +638,9 @@ const parseTargets = ( throw badRequest('Unknown delivery target', 'invalid_targets'); const targets = [...new Set(value)]; - if (!targetsAllowedForDelivery(delivery, targets)) + if (!targetsAllowedForDelivery(delivery, targets, appUid)) throw badRequest( - 'A `single` subscription needs a `worker` target and may not target `push`', + 'A `single` subscription may not target `push`, and an app`s needs a `worker` target', 'invalid_targets', ); if (appUid === null && targets.includes('worker')) @@ -725,15 +739,22 @@ export class EventsService extends PuterService { #pendingSweep: ReturnType | null = null; /** - * What runs an app's handler. The default records the intent and settles - * nothing, so a delivery handed to it stays owed until there is a real - * invoker to take it. + * What runs an app's handler. Replaced at start-up by the invoker that + * calls the app's events worker; the recorder stands in until then and + * settles nothing, so a delivery handed to it stays owed. */ worker: WorkerInvokerSeam = new RecordingWorkerInvoker(); // -- Lifecycle --------------------------------------------------- override onServerStart(): void { + // Left alone when something has already put its own invoker here. + if (this.worker instanceof RecordingWorkerInvoker) + this.worker = new EventsWorkerInvoker( + this.clients.eventsWorkerInvoker, + (invocation) => this.#mintSubscriberToken(invocation), + ); + this.clients.event.on( 'outer.events.generationBumped', (_key, data, meta) => { @@ -974,6 +995,14 @@ export class EventsService extends PuterService { // only one that is always there to take it. if (delivery === 'single' && !handlerName) throw handlerRequired(); + // Consent first: a row that would run the app's code with nobody + // present is refused before anything about it is resolved or stored. + if ( + needsBackgroundConsent(targets) && + !(await this.#hasBackgroundConsent(actor)) + ) + throw backgroundConsentRequired(); + if (handlerName) await this.#assertHandlerBinding(appUid, handlerName, handlerHash); @@ -1099,7 +1128,9 @@ export class EventsService extends PuterService { throw unknownSubscription(); await this.stores.pendingDelivery.settle(subId, entryId); - await this.#drain(row); + // A suspended row is not delivered to: settling what it already handed + // out must not be the trigger that hands out the next one. + if (row.suspendedAt === null) await this.#drain(row); } // -- Handlers ---------------------------------------------------- @@ -1511,21 +1542,28 @@ export class EventsService extends PuterService { return suspended; } - /** Bring back what was waiting on this name. The other half of a removal. */ + /** + * Bring back what was waiting on this name. The other half of a removal — + * and of a handler that kept failing, since new source under the name is + * the fix for one that could not take its deliveries. + */ async #resumeHandlerDependents( appUid: string, name: string, ): Promise { let resumed = 0; - for (;;) { - const batch = await this.stores.durableSubscription.listByHandler( - appUid, - name, - { suspendedReason: 'handler_not_found' }, - ); - if (batch.length === 0) break; - resumed += await this.resumeSubscriptions(batch); - if (batch.length < HANDLER_SETTLE_BATCH) break; + for (const reason of ['handler_not_found', 'failures'] as const) { + for (;;) { + const batch = + await this.stores.durableSubscription.listByHandler( + appUid, + name, + { suspendedReason: reason }, + ); + if (batch.length === 0) break; + resumed += await this.resumeSubscriptions(batch); + if (batch.length < HANDLER_SETTLE_BATCH) break; + } } return resumed; } @@ -1581,6 +1619,25 @@ export class EventsService extends PuterService { } } + /** + * Whether this actor may have its handler run in the background. + * + * Asked of the actor rather than of the app, so an access token an app + * issued has to carry the consent itself — a delivery that runs code with + * nobody present is not something a narrower credential inherits. + */ + async #hasBackgroundConsent(actor: Actor): Promise { + try { + return await this.services.permission.check( + actor, + EVENTS_BACKGROUND_PERMISSION, + ); + } catch (err) { + console.warn('[events] background consent check failed', err); + return false; + } + } + /** * Whether a subscription may bind the handler name it asked for. * @@ -2180,14 +2237,7 @@ export class EventsService extends PuterService { revocation.holderUserId, revocation.appUid, ); - // Withdrawing an app's access wholesale is the user saying the app is - // done, so it takes everything the app holds — no per-row question, - // and none of the standing exemptions an app enjoys over its own data - // keep a background subscription alive past the consent that made it. - const settling = - revocation.permission === null - ? held - : await this.#leftUnauthorized(held, revocation.permission); + const settling = await this.#leftSettling(held, revocation.permission); if (settling.length === 0) return 0; // The `permission_revoked` arm of the shared policy purges the backlog @@ -2199,6 +2249,52 @@ export class EventsService extends PuterService { return suspended.length; } + /** + * Which of a holder's rows one withdrawn grant actually stops. + * + * Withdrawing an app's access wholesale is the user saying the app is done, + * so it takes everything the app holds — no per-row question, and none of + * the standing exemptions an app enjoys over its own data keep a + * subscription alive past the consent that made it. + */ + async #leftSettling( + held: readonly DurableSubscription[], + permission: string | null, + ): Promise { + if (permission === null) return [...held]; + if (permission === EVENTS_BACKGROUND_PERMISSION) + return this.#leftWithoutConsent(held); + return this.#leftUnauthorized(held, permission); + } + + /** + * Of a holder's rows, the ones that were running the app's code in the + * background on a consent that has just been withdrawn. Rows delivered only + * to a connection are untouched: what was withdrawn is the right to run + * with nobody there, not the app itself. + * + * The consent is asked for again rather than assumed gone — a mode change + * is recorded as a revoke followed by a grant, and settling on the revoke + * alone would end subscriptions whose consent still stands. + */ + async #leftWithoutConsent( + rows: readonly DurableSubscription[], + ): Promise { + const background = rows.filter((row) => + needsBackgroundConsent(targetsOf(row)), + ); + if (background.length === 0) return []; + + const deps = this.#aclDeps(); + const settling: DurableSubscription[] = []; + for (const row of background) { + const actor = await resolveGrantActor(row, deps); + if (actor && (await this.#hasBackgroundConsent(actor))) continue; + settling.push(row); + } + return settling; + } + /** * Of a holder's rows, the ones a named withdrawn grant has actually left * without access. @@ -2562,10 +2658,59 @@ export class EventsService extends PuterService { const outcome = await this.worker.invoke(invocation); this.onDelivered({ subId: row.subId, event: claimed.event }); - if (outcome !== 'settled') return false; + if (outcome === 'settled') { + await this.stores.pendingDelivery.clearFailures(row.subId); + await this.stores.pendingDelivery.settle( + row.subId, + claimed.entryId, + ); + return true; + } + // Nothing was attempted, so nothing failed: the lease paces the retry. + if (outcome === 'deferred') return false; - await this.stores.pendingDelivery.settle(row.subId, claimed.entryId); - return true; + await this.#handlerFailed(row, claimed, outcome); + return false; + } + + /** + * What a handler that would not take a delivery costs it, and what a run of + * them costs the subscription. + * + * A refusal is the handler's answer, so the delivery is dropped with a gap + * marker rather than sent again to the same answer; anything else is "not + * now", and the delivery waits longer each time. Both count: five failures + * in a row is a handler that is not working, whichever way it is failing, + * and retrying it forever is how one bad deploy becomes a standing load on + * whatever it is calling. + */ + async #handlerFailed( + row: DispatchSubscription, + claimed: ClaimedDelivery, + outcome: Exclude, + ): Promise { + const pending = this.stores.pendingDelivery; + try { + if (outcome === 'terminal') + await pending.discard( + row.subId, + claimed.entryId, + 'handler_rejected', + ); + else await pending.deferAfterFailure(row.subId, claimed.entryId); + + const failures = await pending.recordFailure(row.subId); + if (failures < EVENTS_CONSECUTIVE_FAILURES) return; + + await pending.clearFailures(row.subId); + await this.suspendForFailures(row.subId); + } catch (err) { + console.warn( + '[events] could not record a failed handler attempt', + row.subId, + err, + ); + } } /** What the handler seam is handed, or null for a row that wants none. */ @@ -2582,9 +2727,51 @@ export class EventsService extends PuterService { handlerName: row.handlerName ?? null, event, context: row.context ?? null, + permissions: subscriptionTokenPermissions(row), }; } + /** + * The token one invocation carries: the subscriber's own identity, the app + * whose handler runs, and exactly the grant the subscription was made + * under. Short-lived, because it leaves the platform — a handler that needs + * longer than the delivery it was given is asking for standing access the + * subscription never granted. + * + * Null when the identity cannot be rebuilt (the holder or the app is gone), + * which is not a handler failure: there is nothing to invoke. + */ + async #mintSubscriberToken( + invocation: WorkerInvocation, + ): Promise { + const actor = await resolveGrantActor( + { + holderUserId: invocation.holderUserId, + appUid: invocation.appUid, + permission: SUBSCRIBE_MODE, + }, + this.#aclDeps(), + ); + if (!actor) return null; + + try { + return await this.services.auth.createAccessToken( + actor, + invocation.permissions.map( + (permission) => [permission] as [string], + ), + { expiresIn: DELIVERY_TOKEN_TTL }, + ); + } catch (err) { + console.warn( + '[events] could not mint a delivery token', + invocation.subId, + err, + ); + return null; + } + } + /** * Say that deliveries were dropped to stay inside a cap. The subscriptions * that lost them are told by the gap marker the store queued in their @@ -2702,7 +2889,9 @@ export class EventsService extends PuterService { const invocation = delivery.worker; // At-most-once by construction: a `broadcast` invocation is never // retried, which is why the docs ask handlers to be idempotent - // rather than promising them each event exactly once. + // rather than promising them each event exactly once. It counts + // toward nothing either — a row whose socket copies are arriving + // must not be stopped by a handler nobody is waiting on. try { void this.worker.invoke(invocation).catch((err: unknown) => { console.warn('[events] handler invocation failed', err); diff --git a/src/backend/services/events/authorization.ts b/src/backend/services/events/authorization.ts index 4f182d10f..b58484b6b 100644 --- a/src/backend/services/events/authorization.ts +++ b/src/backend/services/events/authorization.ts @@ -20,6 +20,7 @@ import { actorUid, makeActor, type Actor } from '../../core/actor.js'; import { HttpError } from '../../core/http/HttpError.js'; import type { UserRow } from '../../stores/user/UserStore.js'; +import type { SubscriptionTarget } from '../../stores/events/types.js'; import type { AclError, AclMode, @@ -29,6 +30,8 @@ import { appDataPermission, appDataSharingAllowed, } from '../permission/appDataScopes.js'; +import { PermissionUtil } from '../permission/permissionUtil.js'; +import { isKvToken } from './subjects.js'; /** * Who may subscribe to an anchor, who may still be delivered from it, and which @@ -202,6 +205,51 @@ export const checkDeliveryAuthorized = async ( } }; +// -- Background delivery ---------------------------------------------- + +/** + * Consent to run an app's handler with nobody present. + * + * Delivering to a connected client is the app doing what the user opened it to + * do; invoking its handler hours later, on their account and their bill, is a + * different thing to agree to — so it is a per-app grant of its own, asked for + * through the same flow as every other one and revocable in the same place. + */ +export const EVENTS_BACKGROUND_PERMISSION = 'events:background'; + +/** Whether a row's transports include one that runs code with nobody there. */ +export const needsBackgroundConsent = ( + targets: readonly SubscriptionTarget[], +): boolean => targets.includes('worker'); + +export const backgroundConsentRequired = (): HttpError => + new HttpError( + 403, + 'Background delivery needs this app’s `events:background` permission', + { legacyCode: 'events_background_consent_required' }, + ); + +/** + * The grant a delivery's token carries: exactly the one its subscribe check + * passed under, so the token can reach what the subscription watches and + * nothing else. The issuer-subset check refuses anything wider. + * + * A row on the app's own key-value namespace has no grant behind it — the app + * has always been able to read its own data — so its token carries none and + * stands only for who the delivery is for. + */ +export const subscriptionTokenPermissions = (row: { + token: string; + anchorUid: string; + appUid: string | null; + permission: AclMode; +}): string[] => { + if (!isKvToken(row.token)) + return [PermissionUtil.join('fs', row.anchorUid, row.permission)]; + if (row.appUid === null || row.appUid === row.anchorUid) return []; + return [appDataPermission(row.anchorUid, 'kv', CROSS_APP_KV_CLASS)]; +}; + // -- Cross-app KV ------------------------------------------------------ /** diff --git a/src/backend/services/events/durable.integration.test.ts b/src/backend/services/events/durable.integration.test.ts index c47d3f5b4..e1a375d71 100644 --- a/src/backend/services/events/durable.integration.test.ts +++ b/src/backend/services/events/durable.integration.test.ts @@ -33,6 +33,7 @@ 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 { EVENTS_BACKGROUND_PERMISSION } from './authorization.js'; import type { DeliveryEnvelope } from './EventsService.js'; const BOOT_TIMEOUT_MS = 120_000; @@ -92,8 +93,13 @@ const unsubscribe = (token: string, subId: string): Promise => 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 }> => { +/** + * An app the user has granted `list` on the shared anchor, and consent to run + * its handler in the background — which durable rows target by default. + */ +const makeApp = async ( + options: { background?: boolean } = {}, +): 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 (?, ?, ?, ?, ?)', @@ -108,6 +114,12 @@ const makeApp = async (): Promise<{ uid: string; token: string }> => { uid, `fs:${entry!.uid}:list`, ); + if (options.background !== false) + await env.server.services.permission.grantUserAppPermission( + actor.actor!, + uid, + EVENTS_BACKGROUND_PERMISSION, + ); return { uid, token: await env.server.services.auth.getUserAppToken(actor.actor!, uid), @@ -272,6 +284,50 @@ describe('creating a durable subscription over HTTP', () => { expect(refused.body.code).toBe('invalid_targets'); }); + it('refuses background delivery an app has no consent for', async () => { + // An app of its own, so nothing it holds includes the consent. + const { token } = await makeApp({ background: false }); + + const refused = await call('POST', '/events/subscribe', token, { + subject: `fs:${anchor}`, + delivery: 'single', + handlerName: 'onWrite', + targets: ['worker'], + }); + + expect(refused.status).toBe(403); + expect(refused.body.code).toBe('events_background_consent_required'); + // Refused before the handler is even looked up: consent comes first. + expect(refused.body.message).toContain('events:background'); + }); + + it('requires consent for the default targets an app row gets, even with none named', async () => { + // An app of its own, so nothing it holds includes the consent. + const { token } = await makeApp({ background: false }); + + // No `targets` at all: an app row defaults to `['socket', 'worker']`, + // and that default still needs the consent — the gate runs on the + // resolved targets, not only on an explicit ask for `worker`. + const refused = await call('POST', '/events/subscribe', token, { + subject: `fs:${anchor}`, + }); + + expect(refused.status).toBe(403); + expect(refused.body.code).toBe('events_background_consent_required'); + }); + + it('needs no consent for a subscription only a connection hears', async () => { + const { token } = await makeApp({ background: false }); + + const created = await call('POST', '/events/subscribe', token, { + subject: `fs:${anchor}`, + targets: ['socket'], + }); + + expect(created.status).toBe(200); + expect(created.body.targets).toEqual(['socket']); + }); + it('refuses a target outside the known set', async () => { const refused = await subscribe(env.users.user.token, { targets: ['socket', 'carrier-pigeon'], @@ -281,8 +337,8 @@ describe('creating a durable subscription over HTTP', () => { expect(refused.body.code).toBe('invalid_targets'); }); - it('refuses a `single` subscription with no handler to fall back to', async () => { - const refused = await subscribe(env.users.user.token, { + it('refuses an app`s `single` subscription with no worker to fall back to', async () => { + const refused = await subscribe(appOneToken, { delivery: 'single', handlerName: 'onWrite', targets: ['socket'], diff --git a/src/backend/services/events/handlers.integration.test.ts b/src/backend/services/events/handlers.integration.test.ts index 383c561c6..2ec2520a2 100644 --- a/src/backend/services/events/handlers.integration.test.ts +++ b/src/backend/services/events/handlers.integration.test.ts @@ -35,6 +35,7 @@ import { import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; import { hashContent } from '../../stores/events/EventHandlerStore.js'; import type { IConfig } from '../../types.js'; +import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js'; import type { DurableSubscriptionView } from './EventsService.js'; import { RecordingWorkerInvoker } from './workerSeam.js'; @@ -119,6 +120,13 @@ const makeApp = async ( uid, `fs:${entry!.uid}:list`, ); + // Durable rows target the app's worker by default, which takes its own + // consent. + await env.server.services.permission.grantUserAppPermission( + actor!, + uid, + EVENTS_BACKGROUND_PERMISSION, + ); tokens.push( await env.server.services.auth.getUserAppToken(actor!, uid), ); @@ -676,6 +684,25 @@ describe('the handler lifecycle', () => { // A credit restore does not lift a suspension it did not cause. expect(await events().resumeForCredit(userId)).toBe(0); }); + + it('puts a row a failing handler stopped back in service on a republish', async () => { + const subId = await bind(); + await events().suspendForFailures(subId); + + // New source under the name is the fix for a handler that could not + // take its deliveries, so it is what brings its subscriptions back. + const republished = await call('POST', '/events/handlers/publish', appToken, { + name: 'ingestUpload', + source: NEXT_SOURCE, + replace: true, + }); + + expect(republished.body.resumed).toBe(1); + expect(await rowOf(subId)).toMatchObject({ + suspendedAt: null, + suspendedReason: null, + }); + }); }); describe('the context a subscription carries', () => { diff --git a/src/backend/services/events/kv.integration.test.ts b/src/backend/services/events/kv.integration.test.ts index 6f8675971..8f6f7e7a4 100644 --- a/src/backend/services/events/kv.integration.test.ts +++ b/src/backend/services/events/kv.integration.test.ts @@ -33,6 +33,7 @@ import { runWithContext } from '../../core/context.js'; import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; import type { IConfig } from '../../types.js'; import { appDataPermission } from '../permission/appDataScopes.js'; +import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js'; import type { DeliveryEnvelope } from './EventsService.js'; const BOOT_TIMEOUT_MS = 120_000; @@ -136,6 +137,13 @@ beforeAll(async () => { userActor, ownAppUid, ); + // Durable rows target the app's worker by default, which takes its own + // consent. + await env.server.services.permission.grantUserAppPermission( + userActor, + ownAppUid, + EVENTS_BACKGROUND_PERMISSION, + ); delivered = []; events().onDelivered = (envelope) => delivered.push(envelope); diff --git a/src/backend/services/events/registry.ts b/src/backend/services/events/registry.ts index aba9c6ec0..a4b928847 100644 --- a/src/backend/services/events/registry.ts +++ b/src/backend/services/events/registry.ts @@ -79,7 +79,9 @@ export type GapReason = | 'backlog_overflow' // A suspension holds what it is owed only for as long as the suspension is // plausibly recoverable; past that the backlog goes and this stands in. - | 'suspended_backlog_expired'; + | 'suspended_backlog_expired' + // The handler refused the delivery outright, so it is not offered again. + | 'handler_rejected'; /** * `gap` says an event existed and was not delivered. It rides the delivery diff --git a/src/backend/services/events/revocationSettle.integration.test.ts b/src/backend/services/events/revocationSettle.integration.test.ts index b5b9c5d7a..df6db96c7 100644 --- a/src/backend/services/events/revocationSettle.integration.test.ts +++ b/src/backend/services/events/revocationSettle.integration.test.ts @@ -40,6 +40,7 @@ import { } from '../../testUtil.js'; import type { IConfig } from '../../types.js'; import type { AclMode } from '../acl/ACLService.js'; +import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js'; import type { DeliveryEnvelope } from './EventsService.js'; import { fsAnchorToken } from './subjects.js'; @@ -197,6 +198,13 @@ const makeApp = async (path: string): Promise => { uid, `fs:${await uidOf(path)}:list`, ); + // Durable rows target the app's worker by default, which takes its own + // consent. + await env.server.services.permission.grantUserAppPermission( + owner.actor, + uid, + EVENTS_BACKGROUND_PERMISSION, + ); const app = await env.server.stores.app.getByUid(uid); return makeActor({ user: owner.actor.user as never, @@ -503,6 +511,33 @@ describe('what a revoked grant settles', () => { expect(delivered).toEqual([]); }); + it('settles what background consent allowed, and leaves the rest running', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/settle-background`); + const appActor = await makeApp(path); + + const background = ( + await events().subscribeDurable(appActor, { subject: `fs:${path}` }) + ).sub; + // Delivered to a connection and nowhere else, so the consent that just + // went was never what allowed it. + const foreground = ( + await events().subscribeDurable(appActor, { + subject: `fs:${path}`, + targets: ['socket'], + }) + ).sub; + + await env.server.services.permission.revokeUserAppPermission( + owner.actor, + appActor.app!.uid, + EVENTS_BACKGROUND_PERMISSION, + ); + + await suspendedRow(background.subId); + expect((await rowOf(foreground.subId)).suspended_at).toBeFalsy(); + }); + it('settles a subscription when one app permission is revoked, not just all of them', async () => { await clearRows(); const path = await folder(`/${owner.username}/settle-app-single`); diff --git a/src/backend/services/events/singleDelivery.test.ts b/src/backend/services/events/singleDelivery.test.ts index ae3e9df05..b49ac2bd8 100644 --- a/src/backend/services/events/singleDelivery.test.ts +++ b/src/backend/services/events/singleDelivery.test.ts @@ -22,6 +22,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { EVENTS_COALESCE_WINDOW_MS, EVENTS_REGION_PENDING_CEILING, + deliveryBackoffMs, } from '../../controllers/events/limits.js'; import type { Actor } from '../../core/actor.js'; import { EventSubscriptionStore } from '../../stores/events/EventSubscriptionStore.js'; @@ -379,6 +380,59 @@ describe('a delivery owed to exactly one consumer', () => { expect(sent).toHaveLength(1); expect(sent[0].ackRequired).toBe(true); }); + + it('does not clear a run of handler failures when a socket ack settles a later delivery', async () => { + socketConnected = false; + workerOutcome = 'retriable'; + const row = await register(); + + // Nothing is connected, so this one goes to the handler and fails. + await dispatch(); + expect(invoked).toHaveLength(1); + await expect(redis.get(`ev:qf:{${row.subId}}`)).resolves.toBe('1'); + + // The client reconnects before the backoff clears; the retry finds a + // socket this time and is acked there rather than run again. + socketConnected = true; + jump(deliveryBackoffMs(1) + 1); + await service.sweepPending(); + expect(sent).toHaveLength(1); + + await service.ackDelivery(actorFor(), { + subId: row.subId, + id: sent[0].ackId, + }); + + await expect(pending.depth(row.subId)).resolves.toBe(0); + // A socket taking a delivery is not the handler answering: the strike + // the handler earned earlier stands. + await expect(redis.get(`ev:qf:{${row.subId}}`)).resolves.toBe('1'); + }); + + it('does not hand out the next delivery when an ack lands on a suspended row', async () => { + const row = await register(); + await dispatch(entry({ uid: `file-a-${seq}` })); + await dispatch(entry({ uid: `file-b-${seq}` })); + expect(sent).toHaveLength(1); + + // Suspended between the first delivery and its ack — a settle hook + // mid-flight, not a fresh dispatch decision. + rows.set(row.subId, { + ...row, + suspendedAt: Math.floor(Date.now() / 1000), + suspendedReason: 'failures', + }); + + await service.ackDelivery(actorFor(), { + subId: row.subId, + id: sent[0].ackId, + }); + + // The ack itself still settles; what it must not do is hand the next + // owed delivery out to a subscription that is no longer in service. + expect(sent).toHaveLength(1); + await expect(pending.depth(row.subId)).resolves.toBe(1); + }); }); describe('a delivery everyone connected gets', () => { diff --git a/src/backend/services/events/workerInvoker.integration.test.ts b/src/backend/services/events/workerInvoker.integration.test.ts new file mode 100644 index 000000000..62ac2b6c9 --- /dev/null +++ b/src/backend/services/events/workerInvoker.integration.test.ts @@ -0,0 +1,486 @@ +/* + * 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 call that leaves the platform, against a stub standing in for an app's + * events worker. + * + * The worker runtime is not built yet, so what is pinned here is everything on + * this side of the wire: the body, the token it carries and what that token is + * allowed to do, and what each answer does to the delivery — settled, dropped, + * or held for longer each time until the subscription stops. + */ + +import http from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { v4 as uuidv4 } from 'uuid'; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + EVENTS_CONSECUTIVE_FAILURES, + deliveryBackoffMs, +} from '../../controllers/events/limits.js'; +import { runWithContext } from '../../core/context.js'; +import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; +import type { IConfig } from '../../types.js'; +import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js'; + +const BOOT_TIMEOUT_MS = 120_000; +/** Short enough that a hung handler is a test case rather than a stall. */ +const INVOKE_TIMEOUT_MS = 300; +const HANDLER = 'ingestUpload'; +const SOURCE = 'async ({ event, ctx }) => { console.log(event.path, ctx.label); }'; + +interface StubCall { + method: string; + path: string; + auth: string | undefined; + body: { + handler?: string; + event?: Record; + ctx?: Record; + }; +} + +let env: PuterTestEnv; +let userId: number; +let anchor: string; +let anchorUid: string; +let appUid: string; +let appToken: string; +let stub: http.Server; +let calls: StubCall[]; +/** What the stub answers with, or `hang` for a handler that never does. */ +let answer: number | 'hang' = 200; + +const events = () => env.server.services.events; +const pending = () => env.server.stores.pendingDelivery; +const durable = () => env.server.stores.durableSubscription; + +const readBody = async (req: http.IncomingMessage): Promise => { + const chunks: Buffer[] = []; + for await (const chunk of req) chunks.push(chunk as Buffer); + return Buffer.concat(chunks).toString('utf8'); +}; + +const startStub = async (): Promise => { + stub = http.createServer((req, res) => { + void readBody(req).then((raw) => { + calls.push({ + method: req.method ?? '', + path: req.url ?? '', + auth: req.headers['puter-auth'] as string | undefined, + body: JSON.parse(raw || '{}') as StubCall['body'], + }); + if (answer === 'hang') return; + res.writeHead(answer).end(); + }); + }); + await new Promise((resolve) => + stub.listen(0, '127.0.0.1', () => resolve()), + ); + return `http://127.0.0.1:${(stub.address() as AddressInfo).port}`; +}; + +const subscribe = async (): Promise => { + const response = await fetch(new URL('/events/subscribe', env.apiOrigin), { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${appToken}`, + }, + body: JSON.stringify({ + subject: `fs:${anchor}`, + delivery: 'single', + handlerName: HANDLER, + targets: ['worker'], + context: { label: 'ingest' }, + }), + }); + const body = (await response.json()) as { subId: string }; + expect(response.status).toBe(200); + return body.subId; +}; + +/** A durable KV subscription on the app's own namespace, targeting the worker. */ +const subscribeKv = async (key: string): Promise => { + const response = await fetch(new URL('/events/subscribe', env.apiOrigin), { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${appToken}`, + }, + body: JSON.stringify({ + subject: `kv:${key}`, + delivery: 'single', + handlerName: HANDLER, + targets: ['worker'], + }), + }); + const body = (await response.json()) as { subId: string }; + expect(response.status).toBe(200); + return body.subId; +}; + +/** One write under the anchor, which is one delivery owed to the handler. */ +const touch = async (name: string): Promise => { + await env.server.services.fs.touch(userId, { path: `${anchor}/${name}` }); +}; + +const invoked = (count: number): Promise => + vi.waitFor(() => expect(calls.length).toBeGreaterThanOrEqual(count), { + timeout: 5_000, + interval: 25, + }); + +/** Move time past whatever the failed delivery is being held for. */ +const jump = (ms: number): void => { + vi.useFakeTimers({ toFake: ['Date'] }); + vi.setSystemTime(Date.now() + ms); +}; + +/** How much longer the one in-flight delivery is being held. */ +const heldForMs = async (subId: string): Promise => { + const [, score] = await env.server.clients.redis.zrange( + `ev:ql:{${subId}}`, + 0, + 0, + 'WITHSCORES', + ); + return Number(score) - Date.now(); +}; + +beforeAll(async () => { + env = await setupPuterTestEnv({ + events: { enabled: true, invokeTimeoutMs: INVOKE_TIMEOUT_MS }, + } as IConfig); + + const user = await env.server.stores.user.getByUsername( + env.users.user.username, + ); + userId = user!.id; + anchor = `/${env.users.user.username}/invoker`; + await env.server.services.fs.mkdir(userId, { + path: anchor, + createMissingParents: true, + }); + anchorUid = (await env.server.stores.fsEntry.getEntryByPath(anchor))!.uid; + + appUid = `app-${uuidv4()}`; + await env.server.clients.db.write( + 'INSERT INTO `apps` (`uid`, `name`, `title`, `index_url`, `owner_user_id`) VALUES (?, ?, ?, ?, ?)', + [appUid, appUid, appUid, `https://${appUid}.example/`, userId], + ); + const { actor } = await env.server.services.auth.authenticate( + env.users.user.token, + ); + await env.server.services.permission.grantUserAppPermission( + actor!, + appUid, + `fs:${anchorUid}:list`, + ); + await env.server.services.permission.grantUserAppPermission( + actor!, + appUid, + EVENTS_BACKGROUND_PERMISSION, + ); + appToken = await env.server.services.auth.getUserAppToken(actor!, appUid); + await env.server.stores.eventHandler.publish({ + appUid, + name: HANDLER, + source: SOURCE, + }); + + calls = []; + const origin = await startStub(); + env.server.clients.eventsWorkerInvoker.setResolver({ + resolveInvokeUrl: () => Promise.resolve(origin), + }); +}, BOOT_TIMEOUT_MS); + +afterAll(async () => { + await new Promise((resolve) => stub?.close(() => resolve())); + await env?.shutdown(); +}); + +beforeEach(async () => { + vi.useRealTimers(); + calls = []; + answer = 200; + await env.server.clients.db.write( + 'DELETE FROM `event_subscriptions`', + [], + ); + events().invalidateUser(userId); + await env.server.stores.eventSubscription.markRegionCold(userId); + await durable().warmRegion(userId); +}); + +describe('the call an owed delivery makes', () => { + it('posts the handler, the event and the context to the app`s worker', async () => { + await subscribe(); + + await touch('note.txt'); + await invoked(1); + + const call = calls[0]; + expect(call.method).toBe('POST'); + expect(call.path).toBe('/__events/invoke'); + expect(call.body.handler).toBe(HANDLER); + expect(call.body.ctx).toEqual({ label: 'ingest' }); + expect(Object.keys(call.body.event ?? {}).sort()).toEqual([ + 'id', + 'op', + 'path', + 'self', + 'seq', + 'subject', + 'ts', + 'uid', + ]); + expect(call.body.event).toMatchObject({ + path: `${anchor}/note.txt`, + self: true, + }); + }); + + it('carries a token that is the subscriber, the app, and nothing wider', async () => { + await subscribe(); + + await touch('token.txt'); + await invoked(1); + + const token = calls[0].auth; + expect(typeof token).toBe('string'); + + const { actor } = await env.server.services.auth.authenticate(token!); + expect(actor!.user?.id).toBe(userId); + expect(actor!.effectiveApp?.uid).toBe(appUid); + + const permission = env.server.services.permission; + await expect( + permission.check(actor!, `fs:${anchorUid}:list`), + ).resolves.toBe(true); + // The subscription was made under `list`, so the token stops there — + // even though the account it acts for owns the folder outright. + await expect( + permission.check(actor!, `fs:${anchorUid}:write`), + ).resolves.toBe(false); + + const decoded = env.server.services.token.verify('auth', token!) as { + exp: number; + iat: number; + }; + expect(decoded.exp - decoded.iat).toBe(5 * 60); + }); + + it('mints no extra grant for a row on the app`s own kv namespace', async () => { + // The app already holds `fs:${anchorUid}:list` at the account level + // (see `beforeAll`) — proof, below, that a kv delivery's token cannot + // spend a grant the app holds for a different reason entirely. + await subscribeKv('widget'); + + const actor = (await env.server.services.auth.authenticate(appToken)) + .actor!; + await runWithContext({ actor }, () => + env.server.drivers.kvStore.set({ key: 'widget', value: 'x' }), + ); + await invoked(1); + + const token = calls[0].auth!; + const decoded = env.server.services.token.verify('auth', token) as { + token_uid: string; + }; + const rows = await env.server.clients.db.read( + 'SELECT `permission` FROM `access_token_permissions` WHERE `token_uid` = ?', + [decoded.token_uid], + ); + // Own-namespace kv has no grant behind it: nothing was minted at all. + expect(rows).toEqual([]); + + const tokenActor = ( + await env.server.services.auth.authenticate(token) + ).actor!; + expect(tokenActor.effectiveApp?.uid).toBe(appUid); + await expect( + env.server.services.permission.check( + tokenActor, + `fs:${anchorUid}:list`, + ), + ).resolves.toBe(false); + }); + + it('settles the delivery when the handler takes it', async () => { + const subId = await subscribe(); + + await touch('settled.txt'); + await invoked(1); + + await vi.waitFor(async () => + expect(await pending().depth(subId)).toBe(0), + ); + }); +}); + +describe('what each answer does to the delivery', () => { + it('drops one the handler refused, and says so with a gap marker', async () => { + answer = 400; + const subId = await subscribe(); + + await touch('refused.txt'); + await invoked(1); + + // The event is gone and a marker stands in its place, so the + // subscription learns there was one rather than reading silence. + await vi.waitFor(async () => + expect(await pending().depth(subId)).toBe(1), + ); + answer = 200; + const claimed = await pending().claim(subId, { leaseMs: 0 }); + expect(claimed?.event).toMatchObject({ + op: 'gap', + reason: 'handler_rejected', + }); + // It counted: a refusal is still a handler that did not work. + await expect( + env.server.clients.redis.get(`ev:qf:{${subId}}`), + ).resolves.toBe('1'); + }); + + it('holds one it could not answer, for longer each time', async () => { + answer = 500; + const subId = await subscribe(); + + await touch('failing.txt'); + await invoked(1); + + const waits: number[] = []; + for (let attempt = 1; attempt < EVENTS_CONSECUTIVE_FAILURES; attempt++) { + await vi.waitFor(async () => + expect(await heldForMs(subId)).toBeGreaterThan(0), + ); + waits.push(await heldForMs(subId)); + + // Nothing may take it while it is held. + await events().sweepPending(); + expect(calls).toHaveLength(attempt); + + jump(deliveryBackoffMs(attempt) + 50); + await events().sweepPending(); + await invoked(attempt + 1); + } + + expect(waits.map((wait) => Math.round(wait / 1000))).toEqual([2, 4, 8, 16]); + + // The fifth failure in a row is the one that stops it. + await vi.waitFor(async () => + expect((await durable().getBySubId(subId))?.suspendedReason).toBe( + 'failures', + ), + ); + expect(calls).toHaveLength(EVENTS_CONSECUTIVE_FAILURES); + }); + + it('tells the developer their handler stopped working', async () => { + answer = 503; + const notify = vi.spyOn(env.server.services.notification, 'notify'); + const subId = await subscribe(); + + await touch('notified.txt'); + for (let attempt = 1; attempt < EVENTS_CONSECUTIVE_FAILURES; attempt++) { + await invoked(attempt); + jump(deliveryBackoffMs(attempt) + 50); + await events().sweepPending(); + } + + await vi.waitFor(() => + expect(notify).toHaveBeenCalledWith( + [userId], + expect.objectContaining({ handler: HANDLER }), + expect.objectContaining({ type: 'app.events.suspended' }), + ), + ); + expect((await durable().getBySubId(subId))?.suspendedAt).not.toBeNull(); + notify.mockRestore(); + }); + + it('takes a handler that never answers as one that could not', async () => { + answer = 'hang'; + const subId = await subscribe(); + + await touch('hanging.txt'); + await invoked(1); + + // Still owed, and held rather than dropped: nobody said no. + await vi.waitFor(async () => + expect(await heldForMs(subId)).toBeGreaterThan(0), + ); + expect(await pending().depth(subId)).toBe(1); + answer = 200; + }); + + it('takes a handler that is too busy as one that could not', async () => { + answer = 429; + const subId = await subscribe(); + + await touch('busy.txt'); + await invoked(1); + + await vi.waitFor(async () => + expect(await heldForMs(subId)).toBeGreaterThan(0), + ); + expect(await pending().depth(subId)).toBe(1); + }); +}); + +describe('with no events worker to address', () => { + it('counts the same as a handler that could not answer', async () => { + env.server.clients.eventsWorkerInvoker.setResolver({ + resolveInvokeUrl: () => Promise.resolve(null), + }); + const subId = await subscribe(); + + try { + await touch('unresolved.txt'); + for ( + let attempt = 1; + attempt < EVENTS_CONSECUTIVE_FAILURES; + attempt++ + ) { + await vi.waitFor(async () => + expect(await heldForMs(subId)).toBeGreaterThan(0), + ); + jump(deliveryBackoffMs(attempt) + 50); + await events().sweepPending(); + } + + await vi.waitFor(async () => + expect( + (await durable().getBySubId(subId))?.suspendedReason, + ).toBe('failures'), + ); + // Nothing was ever called: there was nowhere to call. + expect(calls).toEqual([]); + } finally { + const origin = `http://127.0.0.1:${(stub.address() as AddressInfo).port}`; + env.server.clients.eventsWorkerInvoker.setResolver({ + resolveInvokeUrl: () => Promise.resolve(origin), + }); + } + }); +}); diff --git a/src/backend/services/events/workerSeam.ts b/src/backend/services/events/workerSeam.ts index f0759f4f8..28b4effa7 100644 --- a/src/backend/services/events/workerSeam.ts +++ b/src/backend/services/events/workerSeam.ts @@ -17,16 +17,15 @@ * along with this program. If not, see . */ +import type { EventsWorkerInvokerClient } from '../../clients/events/EventsWorkerInvokerClient.js'; import type { DeliverableEvent } from './registry.js'; /** * Where a delivery leaves the event system for the app's own code. * - * The invoker itself — minting the subscriber-scoped token, the call, its - * retries — is not built yet, and delivery semantics must not wait for it: what - * runs a handler is one decision, and how many times a `single` may be handed - * out is another. So the seam is the boundary, and the default records the - * intent without acting on it. + * The seam is the boundary between "this delivery is owed to a handler" and + * "something ran it": what runs a handler is one decision, and how many times a + * `single` may be handed out is another. * * An invocation is not an ack. A `single` delivery stays leased until the * invoker reports the handler took it, which is what keeps a handler that never @@ -44,19 +43,92 @@ export interface WorkerInvocation { event: DeliverableEvent; /** The subscription's stored context, delivered to the handler as `ctx`. */ context: string | null; + /** + * The grant the subscription was made under, and the whole of what its + * token may carry. Resolved with the row so the invoker needs no lookup of + * its own. + */ + permissions: string[]; } /** - * What the invoker did with it. `settled` means the handler took the delivery - * and its lease may be released; `deferred` means it has not, and the delivery - * stays owed. + * What the invoker did with it. + * + * - `settled` — the handler took the delivery and its lease may be released. + * - `terminal` — the handler refused it. Nothing is gained by sending it again. + * - `retriable` — nobody could answer; the delivery is owed and comes back. + * - `deferred` — nothing was attempted, so nothing failed either. + * + * `terminal` and `retriable` are both failures and both count toward the run + * that suspends a subscription; they differ only in whether the delivery itself + * gets another turn. */ -export type WorkerInvocationOutcome = 'settled' | 'deferred'; +export type WorkerInvocationOutcome = + | 'settled' + | 'terminal' + | 'retriable' + | 'deferred'; export interface WorkerInvokerSeam { invoke(invocation: WorkerInvocation): Promise; } +/** Mints the token one invocation carries, scoped to the subscriber. */ +export type SubscriberTokenMinter = ( + invocation: WorkerInvocation, +) => Promise; + +/** + * The invoker that actually calls an app's events worker: mint the + * subscriber-scoped token, hand the call to the protocol client, and report + * what the handler said. + * + * A row with no app, no handler name, or no token that can be minted for it has + * nothing to invoke, and says so rather than reporting a failure — there is no + * handler to blame for a row that never named one. + */ +export class EventsWorkerInvoker implements WorkerInvokerSeam { + readonly #client: Pick; + readonly #mintToken: SubscriberTokenMinter; + + constructor( + client: Pick, + mintToken: SubscriberTokenMinter, + ) { + this.#client = client; + this.#mintToken = mintToken; + } + + async invoke( + invocation: WorkerInvocation, + ): Promise { + const { appUid, handlerName } = invocation; + if (!appUid || !handlerName) return 'deferred'; + + const token = await this.#mintToken(invocation); + if (token === null) return 'deferred'; + + const result = await this.#client.invoke({ + appUid, + handler: handlerName, + token, + event: invocation.event, + ctx: parseContext(invocation.context), + }); + return result.outcome; + } +} + +/** Context is stored as JSON text and delivered as the object it was. */ +const parseContext = (context: string | null): unknown => { + if (context === null) return undefined; + try { + return JSON.parse(context) as unknown; + } catch { + return undefined; + } +}; + /** Invocations one recorder holds, so it cannot grow with event volume. */ const RECORDED_INVOCATIONS = 100; diff --git a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts index 10641a667..33e1ad2a3 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts +++ b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts @@ -195,10 +195,11 @@ describe('validation at the row write', () => { await expect(durable().countForHolder(userId)).resolves.toBe(0); }); - it('refuses a `single` row with no handler to fall back to', async () => { + it('refuses an app`s `single` row with no worker to fall back to', async () => { await expect( durable().create( input({ + appUid: 'app-single-rule', delivery: 'single', handlerName: 'onWrite', targets: ['socket'], @@ -207,6 +208,17 @@ describe('validation at the row write', () => { ).rejects.toSatisfy(codeOf('invalid_targets')); }); + it('lets an account`s own `single` row stand on the socket alone', async () => { + const { row } = await durable().create( + input({ + delivery: 'single', + handlerName: 'onWrite', + targets: ['socket'], + }), + ); + expect(row.targets).toEqual(['socket']); + }); + it('keeps the same targets on a `broadcast` row, where push is fine', async () => { const { row } = await durable().create( input({ targets: ['socket', 'push'] }), diff --git a/src/backend/stores/events/DurableSubscriptionStore.ts b/src/backend/stores/events/DurableSubscriptionStore.ts index a243135b0..a3635f0f8 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.ts +++ b/src/backend/stores/events/DurableSubscriptionStore.ts @@ -150,7 +150,7 @@ const invalidTargets = (): HttpError => const pushOnSingle = (): HttpError => new HttpError( 400, - 'A `single` subscription needs a `worker` target and may not target `push`', + 'A `single` subscription may not target `push`, and an app`s needs a `worker` target', { legacyCode: 'invalid_targets' }, ); @@ -740,7 +740,8 @@ export class DurableSubscriptionStore extends PuterStore { if (!targets.every(isSubscriptionTarget)) throw invalidTargets(); const unique = [...new Set(targets as SubscriptionTarget[])]; - if (!targetsAllowedForDelivery(delivery, unique)) throw pushOnSingle(); + if (!targetsAllowedForDelivery(delivery, unique, appUid)) + throw pushOnSingle(); if (appUid === null && unique.includes('worker')) throw workerNeedsApp(); return unique; diff --git a/src/backend/stores/events/PendingDeliveryStore.test.ts b/src/backend/stores/events/PendingDeliveryStore.test.ts index 26e526df6..5264421c9 100644 --- a/src/backend/stores/events/PendingDeliveryStore.test.ts +++ b/src/backend/stores/events/PendingDeliveryStore.test.ts @@ -22,6 +22,9 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION, EVENTS_REGION_PENDING_CEILING, + EVENTS_RETRY_BASE_MS, + EVENTS_RETRY_MAX_MS, + deliveryBackoffMs, } from '../../controllers/events/limits.js'; import type { ProjectedEvent } from '../../services/events/registry.js'; import type { IConfig } from '../../types.js'; @@ -205,6 +208,72 @@ describe('handing a delivery out', () => { }); }); +describe('when a handler could not take it', () => { + it('holds the delivery longer after each failed attempt', async () => { + const { entryId } = await store.enqueue(subId, event('a')); + await store.claim(subId); + + await expect(store.deferAfterFailure(subId, entryId)).resolves.toEqual({ + attempts: 1, + retryInMs: EVENTS_RETRY_BASE_MS, + }); + // Held: nothing may take it while the wait stands. + await expect(store.claim(subId)).resolves.toBeNull(); + + vi.useFakeTimers({ toFake: ['Date'] }); + vi.setSystemTime(Date.now() + EVENTS_RETRY_BASE_MS + 1); + await expect(store.claim(subId)).resolves.toMatchObject({ entryId }); + + await expect(store.deferAfterFailure(subId, entryId)).resolves.toEqual({ + attempts: 2, + retryInMs: EVENTS_RETRY_BASE_MS * 2, + }); + }); + + it('never waits longer than the cap, however many have failed', async () => { + expect(deliveryBackoffMs(1)).toBe(EVENTS_RETRY_BASE_MS); + expect(deliveryBackoffMs(50)).toBe(EVENTS_RETRY_MAX_MS); + }); + + it('drops a refused delivery and leaves a marker naming why', async () => { + const { entryId } = await store.enqueue(subId, event('a')); + + await expect( + store.discard(subId, entryId, 'handler_rejected'), + ).resolves.toBe(true); + + const held = await pendingEvents(); + expect(held).toHaveLength(1); + expect(held[0]).toMatchObject({ + op: 'gap', + reason: 'handler_rejected', + subject: 'fs:/u/Documents', + }); + // A delivery that is already gone is not dropped twice. + await expect( + store.discard(subId, entryId, 'handler_rejected'), + ).resolves.toBe(false); + }); + + it('counts failures in a row, and forgets them when one lands', async () => { + await expect(store.recordFailure(subId)).resolves.toBe(1); + await expect(store.recordFailure(subId)).resolves.toBe(2); + + await store.clearFailures(subId); + + await expect(store.recordFailure(subId)).resolves.toBe(1); + }); + + it('takes the failure count with the subscription it belongs to', async () => { + await store.enqueue(subId, event('a')); + await store.recordFailure(subId); + + await store.purge(subId); + + await expect(keysOf()).resolves.toEqual([]); + }); +}); + describe('when there is more than can be held', () => { it('drops the oldest of one backlog and leaves one marker', async () => { await seedBacklog(subId, EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION); diff --git a/src/backend/stores/events/PendingDeliveryStore.ts b/src/backend/stores/events/PendingDeliveryStore.ts index e49e5201e..16559fa3a 100644 --- a/src/backend/stores/events/PendingDeliveryStore.ts +++ b/src/backend/stores/events/PendingDeliveryStore.ts @@ -19,8 +19,10 @@ import { randomUUID } from 'node:crypto'; import { + EVENTS_FAILURE_COUNTER_TTL_MS, EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION, EVENTS_REGION_PENDING_CEILING, + deliveryBackoffMs, } from '../../controllers/events/limits.js'; import type { DeliverableEvent, @@ -42,9 +44,10 @@ import { PuterStore } from '../types.js'; * all deleted the moment the last one settles — a subscription that is keeping * up owns nothing: * - * ev:q:{} HASH entryId -> the delivery and its attempt count + * ev:q:{} HASH entryId -> the delivery and its attempt counts * ev:qp:{} ZSET entryId -> enqueued at; membership means unsettled * ev:ql:{} ZSET entryId -> lease expiry; membership means in flight + * ev:qf:{} STR handler failures in a row, expiring on its own * * And two the region shares: * @@ -72,6 +75,7 @@ const entriesKey = (subId: string): string => `ev:q:{${subId}}`; const pendingKey = (subId: string): string => `ev:qp:{${subId}}`; const leaseKey = (subId: string): string => `ev:ql:{${subId}}`; const holdKey = (subId: string): string => `ev:qt:{${subId}}`; +const failureKey = (subId: string): string => `ev:qf:{${subId}}`; const INDEX_KEY = 'ev:qx'; const COUNTER_KEY = 'ev:qc'; @@ -171,6 +175,14 @@ export interface ClaimedDelivery { socketAttempts: number; } +/** What one failed handler attempt left behind. */ +export interface DeferredDelivery { + /** Handler attempts this delivery has now had. */ + attempts: number; + /** How long it is held before anything may claim it again. */ + retryInMs: number; +} + /** What one shed took, for the marker and the alarm that follow it. */ export interface PendingShed { subId: string; @@ -187,6 +199,8 @@ export interface PendingHead { interface StoredEntry { event: DeliverableEvent; socketAttempts: number; + /** Handler attempts spent, which is what the retry wait is derived from. */ + handlerAttempts?: number; } const parseEntry = (raw: string | null): StoredEntry | null => { @@ -342,6 +356,78 @@ export class PendingDeliveryStore extends PuterStore { return socketAttempts; } + /** + * Count one failed handler attempt and hold the delivery until its wait is + * over. The lease is the hold: a score in the future is what `claim` reads + * as "in flight", so pushing it out paces the retry with no second + * mechanism to keep in step. + */ + async deferAfterFailure( + subId: string, + entryId: string, + ): Promise { + const entry = parseEntry( + await this.clients.redis.hget(entriesKey(subId), entryId), + ); + const attempts = (entry?.handlerAttempts ?? 0) + 1; + const retryInMs = deliveryBackoffMs(attempts); + + const write = this.clients.redis.pipeline(); + if (entry) + write.hset( + entriesKey(subId), + entryId, + JSON.stringify({ ...entry, handlerAttempts: attempts }), + ); + write.zadd(leaseKey(subId), Date.now() + retryInMs, entryId); + await write.exec(); + + return { attempts, retryInMs }; + } + + /** + * Drop one delivery nothing will ever take, leaving a gap marker in its + * place. A refused delivery is still an event its subscription was + * promised, so the marker is what keeps the silence from reading as + * "nothing happened". + */ + async discard( + subId: string, + entryId: string, + reason: GapMarker['reason'], + ): Promise { + const subject = await this.#subjectOf(subId, entryId); + const removed = await this.clients.redis.zrem( + pendingKey(subId), + entryId, + ); + if (Number(removed) !== 1) return false; + + await this.#forget(subId, [entryId]); + await this.#append(subId, gapMarker(subject, reason)); + return true; + } + + /** + * Count one failure against the subscription and answer how many are in a + * row. Region-local like the rest of the delivery state: the lease that + * produced the failure is this region's, and a row update per 5xx would put + * a write on the table for every failed attempt. + */ + async recordFailure(subId: string): Promise { + const count = Number(await this.clients.redis.incr(failureKey(subId))); + await this.clients.redis.pexpire( + failureKey(subId), + EVENTS_FAILURE_COUNTER_TTL_MS, + ); + return Number.isFinite(count) ? count : 0; + } + + /** Forget a run of failures, for a delivery that landed. */ + async clearFailures(subId: string): Promise { + await this.clients.redis.del(failureKey(subId)); + } + /** * Settle a delivery a consumer took. False for an id this subscription is * not holding — a second ack for one already settled, which at-least-once @@ -367,6 +453,7 @@ export class PendingDeliveryStore extends PuterStore { pendingKey(subId), leaseKey(subId), holdKey(subId), + failureKey(subId), ); await this.clients.redis.zrem(INDEX_KEY, subId); if (held > 0) await this.clients.redis.decrby(COUNTER_KEY, held); diff --git a/src/backend/stores/events/types.ts b/src/backend/stores/events/types.ts index c89fea22b..1d2cc90e1 100644 --- a/src/backend/stores/events/types.ts +++ b/src/backend/stores/events/types.ts @@ -51,17 +51,21 @@ export const DEFAULT_DURABLE_TARGETS: SubscriptionTarget[] = [ /** * Whether a delivery class may carry these transports. `single` and `push` are * incompatible by construction — a lease can only be settled by a consumer that - * reports back, and a device notification never does — and `single` needs a - * `worker` to fall back to once the connected clients have had their turns, or - * an unacknowledged delivery sits at the head of its queue for good. Both are - * invariants every writer is held to, not defaults. + * reports back, and a device notification never does. An app's `single` row + * also needs a `worker` to fall back to once the connected clients have had + * their turns, or an unacknowledged delivery sits at the head of its queue for + * good. An account's own row has no app to run a worker for, so it is held on + * the socket alone and the pending hold is its backstop. Both are invariants + * every writer is held to, not defaults. */ export const targetsAllowedForDelivery = ( delivery: DeliveryClass, targets: readonly SubscriptionTarget[], + appUid: string | null, ): boolean => delivery !== 'single' || - (!targets.includes('push') && targets.includes('worker')); + (!targets.includes('push') && + (appUid === null || targets.includes('worker'))); /** What dispatch needs from a subscription, whichever store it came from. */ export interface DispatchSubscription { diff --git a/src/backend/types.ts b/src/backend/types.ts index f1c2e7e3a..fb2fbe542 100644 --- a/src/backend/types.ts +++ b/src/backend/types.ts @@ -1139,6 +1139,8 @@ interface IConfigOptional { events?: { enabled?: boolean; crossAppKv?: boolean; + /** How long a handler has to answer an invocation. Default 30 s. */ + invokeTimeoutMs?: number; }; /** diff --git a/src/docs/src/Events.md b/src/docs/src/Events.md index c13040d54..fb80a7bcc 100644 --- a/src/docs/src/Events.md +++ b/src/docs/src/Events.md @@ -166,7 +166,13 @@ Values reach a handler through **`context`**, which is evaluated **once, at subs See [`puter.events.handlers`](/Events/handlers/) for the deploy side — publishing, replacing, and what removing a name does to the subscriptions bound to it. -A persistent subscription can also stop without you unsubscribing: its handler was removed, its holder ran out of credit, or the share it was made under was withdrawn. It is then *suspended* rather than deleted, and [`list()`](/Events/list/) reports `suspendedAt` and `suspendedReason`. Everything but a withdrawn grant can resume. +### Running when nobody is there takes consent + +A persistent subscription delivers to a connected client when there is one and runs the app's handler in the background when there is not. The background half is a separate thing to agree to — your code running on the user's account with nobody watching — so it takes the per-app permission **`events:background`**, requested with [`puter.perms.request()`](/Perms/request/) and revocable wherever the user manages the app's access. Without it, subscribing with `worker` among its `targets` (the default for an app) fails with `events_background_consent_required`; taking it back suspends every worker-target subscription that app holds for that user. A subscription that only wants deliveries while your app is open asks for `targets: ['socket']` and needs no consent. + +Pass `handler` as a **function** and it runs here too, whenever this client is the one the delivery goes to — the same body that runs in the worker, with the same `{ event, ctx, user, fetch, ack }`. See [`onPersistent()`](/Events/onPersistent/) for the acknowledgement rules; the short version is that a `single` delivery is settled by returning from the handler, and a handler that throws sees the event again. + +A persistent subscription can also stop without you unsubscribing: its handler was removed, its holder ran out of credit, the handler kept failing, or the share it was made under was withdrawn. It is then *suspended* rather than deleted, and [`list()`](/Events/list/) reports `suspendedAt` and `suspendedReason`. Everything but a withdrawn grant can resume. ## Limits diff --git a/src/docs/src/Events/onPersistent.md b/src/docs/src/Events/onPersistent.md index 75d40376f..40c4791bd 100644 --- a/src/docs/src/Events/onPersistent.md +++ b/src/docs/src/Events/onPersistent.md @@ -27,6 +27,40 @@ puter.events.onPersistent(options) - `context` (Object): Values the handler needs, delivered to it as a frozen `ctx`. **Capped at 4 KB serialized** — see below. - `expiresAt` (Number | String): When the subscription ends by itself — unix seconds or an ISO-8601 string, and it has to be in the future. +## Background delivery takes the user's consent + +A persistent subscription can run your handler when nobody is there — a different thing from delivering to a page the user has open — so it takes its own per-app permission, **`events:background`**. Subscribing with `worker` among its `targets` without it fails with `events_background_consent_required`, and `['socket', 'worker']` is the default for a subscription an app creates. Ask for it the way you ask for anything else: + +```js +await puter.perms.request(['events:background']); +``` + +The user can take it back wherever they manage an app's access; every worker-target subscription that app holds for them is then suspended with `permission_revoked`, and re-granting does not bring one back — subscribe again. A subscription that only wants deliveries while your app is open needs no consent at all: pass `targets: ['socket']`. + +## Where the handler runs, and what it is handed + +The handler runs **in this client while it is connected**, and in the app's events worker when it is not. It is the same body either way, called with: + +| Binding | What it is | +| --- | --- | +| `event` | The projected event, or a gap marker. | +| `ctx` | The frozen `context` this subscription was created with. | +| `user` | A `puter` bound to the account holding the subscription — the ambient one in a client. | +| `fetch` | [`puter.net.fetch`](/Networking/fetch/) where it exists, the environment's `fetch` otherwise. | +| `ack` | On a `single` subscription only — see below. | + +Passing `handler` as a **function** is what registers it to run here; a source string or `{ file }` is sent as a hash only, and nothing runs client-side. Either way the hash must match what is published under `handlerName`. + +### Acknowledging a `single` delivery + +A `single` delivery is owed to exactly one consumer, so it stays owed until it is acknowledged: + +- Calling `ack()` takes the delivery. +- Returning **without** calling it acknowledges it anyway — a handler that finished did the work. +- **Throwing acknowledges nothing.** The lease lapses after 30 seconds and the delivery is offered again, so a handler that throws sees the same event twice. `event.id` is stable across redeliveries; use it to make the second one a no-op. + +In the events worker the same three outcomes are the response status: `2xx` takes the delivery, `4xx` refuses it (it is dropped with a `gap` marker carrying `reason: 'handler_rejected'`), and `5xx`, `429` or no answer within 30 seconds means "not now" — the delivery is retried after 2 seconds, doubling to at most 5 minutes. **Five failures in a row, refusals included, suspend the subscription** with `failures`; the developer is notified and republishing the handler puts it back in service. + ## `context` is evaluated once, and capped at 4 KB A handler is deployed, not called: it is serialized and run later, somewhere else, so it cannot close over anything. `context` is how values reach it — and it is evaluated **at this call**, serialized, and never re-evaluated. `ctx.endpoint` is whatever `process.env.INGEST_URL` was when you subscribed, forever, until you subscribe again. @@ -52,6 +86,7 @@ A `Promise` that resolves to the subscription: - `contextKeys` (Array | null), `contextHash` (String | null): the shape of the stored context, never its values. - `createdAt`, `expiresAt` (Number | null): unix seconds. - `suspendedAt` (Number | null), `suspendedReason` (String | null): why it stopped delivering without being removed — see [`puter.events.handlers.remove()`](/Events/handlers/). +- `off()` (Function): ends the subscription — stops running its handler here and unsubscribes it. The same thing as [`puter.events.unsubscribe(subId)`](/Events/unsubscribe/), with nothing to pass. The promise rejects with `{ message, code }`: @@ -65,6 +100,7 @@ The promise rejects with `{ message, code }`: | `events_handler_not_found` | No handler is published under `handlerName`. The subscription is **not** created. | | `events_handler_hash_mismatch` | The published handler is not the source this subscription was written against. | | `events_handler_required` | `delivery: 'single'` without a `handlerName`. | +| `events_background_consent_required` | The subscription targets `worker` and the user has not granted this app `events:background`. | | `events_context_too_large` | The serialized `context` is over 4 KB. | | `events_context_invalid` | `context` is not JSON-serializable. | | `invalid_targets` | A target outside `socket`/`worker`/`push`, `push` on a `single` subscription, or `worker` on a subscription with no app. | diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md index 08d1fcfb9..a9acc98c8 100644 --- a/src/docs/src/rate-limits-and-quotas.md +++ b/src/docs/src/rate-limits-and-quotas.md @@ -172,6 +172,10 @@ One write can reach many subscriptions, so events are bounded on both halves: ho | Undelivered deliveries per subscription | 10,000 | | Undelivered deliveries per *suspended* subscription | 100 | | Suspended subscriptions kept for | 30 days | +| Handler invocation timeout | 30 seconds | +| Wait before retrying a failed handler | 2 seconds, doubling | +| Longest wait between retries | 5 minutes | +| Handler failures in a row before suspension | 5 | | Published handlers per app | 100 | | Handler source size | 64 KB | | Handlers per `publishAll` call | 50 | @@ -207,6 +211,8 @@ A `kv:` subject is indexed on the first **6** `:`-segments, or **160 bytes**, of The three per-event ceilings do not fail your call — they truncate the delivery and send a `gap` marker in its place, an event with `op: 'gap'` and no `uid` or `path`. A gap means something happened that you were not told the details of, so a client that must not miss changes should re-read the anchor when it sees one rather than treat the silence as "nothing changed". +A **background delivery** — one that runs your app's handler with nobody there — takes the user's consent, the per-app permission `events:background`, and a subscription targeting `worker` without it is refused with `events_background_consent_required`. A handler has **30 seconds** to answer each invocation. Answering `2xx` takes the delivery; `4xx` refuses it, and it is dropped with a `gap` marker carrying `reason: 'handler_rejected'` rather than sent again to the same answer; `5xx`, `429` and a timeout are all "not now", and the delivery is held **2 seconds** before the next attempt, doubling each time up to **5 minutes**. **Five failures in a row** — refusals included — suspend the subscription with `failures`, hold what it is owed under the suspended-backlog rules above, and notify the app's developer. Until an events worker is deployed for an app there is nothing to invoke, so a worker-target subscription self-limits along exactly this path. + A `single` subscription is delivered to exactly one consumer, which has **30 seconds** to acknowledge each delivery before it is offered again — twice to a connected client, then to the subscription's handler. Until it is acknowledged it is held for you, so a consumer that is away is a backlog that grows: **10,000** undelivered deliveries per subscription, after which the oldest are dropped and one `gap` marker with `reason: 'backlog_overflow'` takes their place. Each region also holds at most **1,000,000** undelivered deliveries across every subscription it serves, and sheds the oldest first — with the same marker — before it reaches that. A redelivery after a missed acknowledgement is normal and expected: deliveries are at-least-once, `event.id` is stable across them, and a handler that runs twice on the same id should do nothing the second time. ### Peer connections diff --git a/src/gui/src/UI/UIPermissionDialog.js b/src/gui/src/UI/UIPermissionDialog.js index 7aac5e58f..bc4ac639f 100644 --- a/src/gui/src/UI/UIPermissionDialog.js +++ b/src/gui/src/UI/UIPermissionDialog.js @@ -646,6 +646,10 @@ async function get_permission_description (permission, options = {}) { return await get_app_data_description(parts, options); } + if ( parts[0] === 'events' && parts[1] === 'background' ) { + return { html: i18n('perm_events_background'), icon: 'zap' }; + } + if ( parts[0] === 'app-root-dir' ) { // Format: app-root-dir:: if ( parts[2] === 'read' ) { diff --git a/src/gui/src/i18n/translations/en.js b/src/gui/src/i18n/translations/en.js index ede010a43..97ed37c88 100644 --- a/src/gui/src/i18n/translations/en.js +++ b/src/gui/src/i18n/translations/en.js @@ -772,6 +772,7 @@ const en = { 'perm_app_data_delete': 'delete entries from {{subject}}.', 'perm_app_data_store_all': 'read, change and delete {{subject}}.', 'perm_app_data_all': "read, change and delete everything {{app}} has saved for you, including any saved logins.", + 'perm_events_background': 'run in the background when your files or data change, even while it is closed', 'perm_dialog_wants_to': 'wants permission to', 'perm_dialog_footnote': 'You can change this anytime in Settings.', 'perm_dialog_error': 'Something went wrong. Please try again.', diff --git a/src/puter-js/src/modules/events/events.test.js b/src/puter-js/src/modules/events/events.test.js index 10433af6e..67d3297a2 100644 --- a/src/puter-js/src/modules/events/events.test.js +++ b/src/puter-js/src/modules/events/events.test.js @@ -411,6 +411,123 @@ describe('reconnect', () => { }); }); +describe('persistent delivery routing', () => { + /** One durable envelope, as the server addresses it at a room. */ + const durable = (subId, path, over = {}) => ({ + ...projected(subId, path), + ...over, + }); + + it('runs the handler on a delivery the server sends here', async () => { + const events = makeModule(); + const seen = []; + events.channel.registerDurable( + 'app-1#sub', + delivery => seen.push(delivery), + { label: 'ingest' }, + ); + + sockets[0].fire('events.delivery', durable('app-1#sub', '/user/a/one.txt')); + + expect(seen).toHaveLength(1); + expect(seen[0].event.path).toBe('/user/a/one.txt'); + expect(seen[0].ctx).toEqual({ label: 'ingest' }); + expect(Object.isFrozen(seen[0].ctx)).toBe(true); + // Nothing to acknowledge: everyone connected gets a copy of this one. + expect(sockets[0].sent.filter(s => s.verb === 'events.ack')).toEqual([]); + }); + + it('hands a delivery owed to one consumer the environment its worker gets', async () => { + const events = makeModule(); + let handed; + events.channel.registerDurable('app-1#sub', delivery => { + handed = delivery; + return delivery.ack(); + }); + + sockets[0].fire( + 'events.delivery', + durable('app-1#sub', '/user/a/one.txt', { + ackRequired: true, + ackId: 'entry-1', + }), + ); + await Promise.resolve(); + + expect(handed.user).toBe(events.puter); + expect(typeof handed.fetch).toBe('function'); + expect(sockets[0].sent.filter(s => s.verb === 'events.ack')).toMatchObject([ + { payload: { subId: 'app-1#sub', id: 'entry-1' } }, + ]); + }); + + it('acknowledges a handler that returns without doing so itself', async () => { + const events = makeModule(); + events.channel.registerDurable('app-1#sub', async () => 'done'); + + sockets[0].fire( + 'events.delivery', + durable('app-1#sub', '/user/a/one.txt', { + ackRequired: true, + ackId: 'entry-2', + }), + ); + await Promise.resolve(); + await Promise.resolve(); + + expect(sockets[0].sent.filter(s => s.verb === 'events.ack')).toMatchObject([ + { payload: { subId: 'app-1#sub', id: 'entry-2' } }, + ]); + }); + + it('acknowledges nothing when the handler throws, so it is delivered again', async () => { + const events = makeModule(); + const errors = vi.spyOn(console, 'error').mockImplementation(() => {}); + events.channel.registerDurable('app-1#sub', async () => { + throw new Error('handler bug'); + }); + + sockets[0].fire( + 'events.delivery', + durable('app-1#sub', '/user/a/one.txt', { + ackRequired: true, + ackId: 'entry-3', + }), + ); + await Promise.resolve(); + await Promise.resolve(); + + expect(sockets[0].sent.filter(s => s.verb === 'events.ack')).toEqual([]); + expect(errors).toHaveBeenCalled(); + }); + + it('keeps routing across a reconnect without subscribing again', async () => { + const events = makeModule(); + const seen = []; + events.channel.registerDurable('app-1#sub', ({ event }) => seen.push(event)); + + sockets[0].fire('disconnect'); + sockets[0].fire('connect'); + + expect(sockets[0].sent.filter(s => s.verb === 'events.subscribe')).toEqual([]); + + sockets[0].fire('events.delivery', durable('app-1#sub', '/user/a/back.txt')); + expect(seen).toHaveLength(1); + }); + + it('stops routing once the subscription is let go, and closes with the last', async () => { + const events = makeModule(); + const seen = []; + events.channel.registerDurable('app-1#sub', ({ event }) => seen.push(event)); + + events.channel.deregisterDurable('app-1#sub'); + sockets[0].fire('events.delivery', durable('app-1#sub', '/user/a/late.txt')); + + expect(seen).toEqual([]); + expect(sockets[0].disconnected).toBe(true); + }); +}); + describe('ack timeout', () => { afterEach(() => { vi.useRealTimers(); diff --git a/src/puter-js/src/modules/events/lib/channel.js b/src/puter-js/src/modules/events/lib/channel.js index a2f155d25..e1d8be844 100644 --- a/src/puter-js/src/modules/events/lib/channel.js +++ b/src/puter-js/src/modules/events/lib/channel.js @@ -22,10 +22,11 @@ import { EventSubscription } from './subscription.js'; /** @typedef {{ ok: true, sub?: SubscriptionView }} VerbAck */ -// The wire, fixed by the server: two verbs answered with an ack, one channel +// The wire, fixed by the server: three verbs answered with an ack, one channel // events arrive on. const SUBSCRIBE_VERB = 'events.subscribe'; const UNSUBSCRIBE_VERB = 'events.unsubscribe'; +const ACK_VERB = 'events.ack'; const DELIVERY_CHANNEL = 'events.delivery'; /** How long a verb waits for its ack before the call is called lost. */ @@ -82,6 +83,11 @@ const handshakeError = (error) => { * so a reconnect is not transparent server-side — this re-issues each live * subscription and re-points its handle at the new id, which is what keeps * `onLocal` a thing you call once. + * + * A persistent subscription is the other way round: the server holds it, its + * id outlives every connection, and what a reconnect has to rebuild is only + * this side's routing — so those registrations are kept apart from the session + * ones and are never re-subscribed. */ export class EventChannel { /** @param {import('../index.js').EventsModule} module */ @@ -94,6 +100,12 @@ export class EventChannel { this.subscriptions = new Set(); /** @internal @type {Map} */ this.byId = new Map(); + /** + * @internal Persistent subscriptions this client is running the + * handler for, by their server-side id. + * @type {Map} + */ + this.durable = new Map(); /** @internal Rejectors for verbs still waiting on an ack. */ this.waiters = new Set(); /** @internal Subscribes that have not resolved yet. */ @@ -154,6 +166,38 @@ export class EventChannel { } } + /** + * Run a persistent subscription's handler here whenever this client is the + * one the server delivers to. The subscription itself already exists and is + * not re-registered by any of this — what is registered is only where its + * deliveries go while this page is open. + * + * @internal + * @param {string} subId + * @param {import('../types.js').EventHandler} handler + * @param {Record} [ctx] The context the subscription was + * created with, delivered frozen alongside every event. + * @returns {void} + */ + registerDurable (subId, handler, ctx) { + this.durable.set(subId, { subId, handler, ctx: Object.freeze({ ...(ctx ?? {}) }) }); + this.connect(); + } + + /** + * Stop routing a persistent subscription here. The subscription is + * untouched — ending it is `unsubscribe()`, which is a different thing from + * this page no longer running its handler. + * + * @internal + * @param {string} subId + * @returns {void} + */ + deregisterDurable (subId) { + if ( ! this.durable.delete(subId) ) return; + this.closeIfIdle(); + } + /** * Rebuild the connection against the current token and origin. Live * subscriptions are re-issued once it is up. @@ -163,7 +207,7 @@ export class EventChannel { */ reset () { this.close(); - if ( this.subscriptions.size > 0 ) this.connect(); + if ( this.subscriptions.size > 0 || this.durable.size > 0 ) this.connect(); } /** @@ -319,21 +363,86 @@ export class EventChannel { /** * @internal - * @param {{ subId?: string, event?: unknown }} envelope + * @param {{ subId?: string, event?: unknown, ackRequired?: boolean, ackId?: string }} envelope * @returns {void} */ route (envelope) { - if ( ! envelope || typeof envelope !== 'object' ) return; - const sub = this.byId.get(/** @type {string} */ (envelope.subId)); + if ( ! envelope || typeof envelope !== 'object' || ! envelope.event ) return; + const subId = /** @type {string} */ (envelope.subId); + const sub = this.byId.get(subId); + if ( sub ) { + sub.deliver( + /** @type {PuterEvent | PuterKvEvent | EventGapMarker} */ (envelope.event), + /** @type {Record | undefined} */ (envelope.ctx), + ); + return; + } + const registered = this.durable.get(subId); // An event for something this client has already unsubscribed from: // in flight when `off()` was called, and no longer anybody's. - if ( ! sub || ! envelope.event ) return; - sub.deliver( - /** @type {PuterEvent | PuterKvEvent | EventGapMarker} */ (envelope.event), - /** @type {Record | undefined} */ (envelope.ctx), + if ( registered ) this.runDurable(registered, envelope); + } + + /** + * Run a persistent subscription's handler on one delivery. + * + * The handler is handed the same environment its published copy gets in the + * app's events worker, so one body runs unchanged in either place. A + * delivery owed to exactly one consumer carries `ack`: calling it settles + * the delivery, resolving without calling it settles it anyway, and + * throwing settles nothing — the lease lapses and it is delivered again. + * + * @internal + * @param {DurableRegistration} registration + * @param {{ event?: unknown, ackRequired?: boolean, ackId?: string }} envelope + * @returns {void} + */ + runDurable (registration, envelope) { + const event = /** @type {PuterEvent | PuterKvEvent | EventGapMarker} */ (envelope.event); + const delivery = { event, ctx: registration.ctx }; + + if ( ! envelope.ackRequired || typeof envelope.ackId !== 'string' ) { + settleHandler(() => registration.handler(delivery)); + return; + } + + let acked = false; + const ack = () => { + if ( acked ) return Promise.resolve(); + acked = true; + return this.ack(registration.subId, /** @type {string} */ (envelope.ackId)); + }; + const { puter } = this.module; + settleHandler( + () => + registration.handler({ + ...delivery, + user: puter, + fetch: puter?.net?.fetch ?? globalThis.fetch, + ack, + }), + () => ack(), ); } + /** + * Tell the server a delivery was taken. Best effort: a lost ack is a + * redelivery, which `single` callers are told to expect, and there is + * nothing a failure here leaves for the caller to fix. + * + * @internal + * @param {string} subId + * @param {string} ackId + * @returns {Promise} + */ + async ack (subId, ackId) { + try { + await this.request(ACK_VERB, { subId, id: ackId }, DEFAULT_TIMEOUT_MS); + } catch (error) { + console.warn('[puter.events] could not acknowledge a delivery', error); + } + } + /** * The connection is not coming back: fail what is waiting on it and end * every subscription it was carrying. @@ -408,6 +517,7 @@ export class EventChannel { */ closeIfIdle () { if ( this.subscriptions.size > 0 || this.inflight > 0 ) return; + if ( this.durable.size > 0 ) return; this.close(); } @@ -433,3 +543,37 @@ export class EventChannel { */ const timeoutFor = (sub) => typeof sub.timeout === 'number' && sub.timeout > 0 ? sub.timeout : DEFAULT_TIMEOUT_MS; + +/** + * One persistent subscription this client runs the handler for. + * + * @typedef {Object} DurableRegistration + * @property {string} subId + * @property {import('../types.js').EventHandler} handler + * @property {Readonly>} ctx + */ + +/** + * Run a handler and, if it finishes without throwing, do whatever the delivery + * still needs. A handler that throws is the app's bug: it is reported, and + * nothing is settled on its behalf. + * + * @param {() => unknown} run + * @param {() => unknown} [onResolved] + * @returns {void} + */ +const settleHandler = (run, onResolved) => { + try { + const result = run(); + if ( result instanceof Promise ) { + result.then( + () => onResolved?.(), + error => console.error('[puter.events] subscription handler failed', error), + ); + return; + } + onResolved?.(); + } catch (error) { + console.error('[puter.events] subscription handler failed', error); + } +}; diff --git a/src/puter-js/src/modules/events/onPersistent.js b/src/puter-js/src/modules/events/onPersistent.js index 7b91bb95e..c12a8a2b2 100644 --- a/src/puter-js/src/modules/events/onPersistent.js +++ b/src/puter-js/src/modules/events/onPersistent.js @@ -9,10 +9,14 @@ import { assertSubject } from './lib/validate.js'; /** * Subscribes to a subject with a subscription that outlives this connection. * - * Unlike `onLocal()`, nothing about this lives in the page: the subscription is - * stored against the account, keeps matching while the app is closed, and is - * ended by `puter.events.unsubscribe()` rather than by navigating away. What - * runs is the app's published handler, named by `handlerName`. + * Unlike `onLocal()`, the subscription is stored against the account, keeps + * matching while the app is closed, and is ended by + * `puter.events.unsubscribe()` rather than by navigating away. What runs it + * while nothing is open is the app's published handler, named by `handlerName`. + * + * While this client *is* open it runs the handler itself, if one was passed as + * a function: the same body, the same `{ event, ctx }`, plus `user`, `fetch` + * and — for a subscription owed to one consumer — `ack`. * * `context` is evaluated **here, now** — serialized once and delivered to every * invocation as a frozen `ctx`. It never re-evaluates, so a value read from the @@ -56,7 +60,20 @@ export async function onPersistent (options = {}) { if ( serializeContext(options.context) !== undefined ) body.context = options.context; - return /** @type {PersistentSubscription} */ ( + const sub = /** @type {PersistentSubscription} */ ( await request(puter, '/events/subscribe', body) ); + + // Durable ids are the server's and survive every reconnect, so routing is + // registered once and never re-subscribed. + if ( typeof handler === 'function' && typeof sub?.subId === 'string' ) { + this.channel.registerDurable(sub.subId, handler, options.context); + } + + const { channel } = this; + sub.off = async () => { + if ( typeof sub.subId === 'string' ) channel.deregisterDurable(sub.subId); + await this.unsubscribe(sub.subId); + }; + return sub; } diff --git a/src/puter-js/src/modules/events/persistent.test.js b/src/puter-js/src/modules/events/persistent.test.js index 42f1b1754..adae16b19 100644 --- a/src/puter-js/src/modules/events/persistent.test.js +++ b/src/puter-js/src/modules/events/persistent.test.js @@ -23,6 +23,16 @@ const makeModule = (fsRead) => { APIOrigin: 'https://api.test', fs: { read: fsRead ?? vi.fn() }, }, + channel: { + registered: [], + deregistered: [], + registerDurable (subId, handler, ctx) { + this.registered.push({ subId, handler, ctx }); + }, + deregisterDurable (subId) { + this.deregistered.push(subId); + }, + }, onPersistent, unsubscribe, list, @@ -188,6 +198,53 @@ describe('onPersistent', () => { expect(error.code).toBe('events_context_invalid'); }); }); + + describe('running the handler here as well', () => { + it('routes the subscription`s deliveries to the function it was given', async () => { + mockRequest.mockResolvedValue({ subId: 'app-1#a', subject: SUBJECT }); + const module = makeModule(); + + await module.onPersistent({ + subject: SUBJECT, + handlerName: 'ingestUpload', + handler: HANDLER, + context: { url: 'https://ingest.example' }, + }); + + expect(module.channel.registered).toMatchObject([ + { subId: 'app-1#a', handler: HANDLER, ctx: { url: 'https://ingest.example' } }, + ]); + }); + + it('routes nothing when the handler is source this client cannot run', async () => { + mockRequest.mockResolvedValue({ subId: 'app-1#a', subject: SUBJECT }); + const module = makeModule(); + + await module.onPersistent({ + subject: SUBJECT, + handlerName: 'ingestUpload', + handler: '({ event }) => console.log(event.path)', + }); + + expect(module.channel.registered).toEqual([]); + }); + + it('off() stops routing and ends the subscription', async () => { + mockRequest.mockResolvedValue({ subId: 'app-1#a', subject: SUBJECT }); + const module = makeModule(); + const sub = await module.onPersistent({ + subject: SUBJECT, + handlerName: 'ingestUpload', + handler: HANDLER, + }); + + await sub.off(); + + expect(module.channel.deregistered).toEqual(['app-1#a']); + expect(routeOf(1)).toBe('/events/unsubscribe'); + expect(bodyOf(1)).toEqual({ subId: 'app-1#a' }); + }); + }); }); describe('unsubscribe', () => { diff --git a/src/puter-js/src/modules/events/types.js b/src/puter-js/src/modules/events/types.js index 5dae4f9aa..dcafb9744 100644 --- a/src/puter-js/src/modules/events/types.js +++ b/src/puter-js/src/modules/events/types.js @@ -64,8 +64,9 @@ * @property {string} reason Why the delivery was dropped — * `matched_subscription_limit`, `filter_evaluation_limit`, * `delivery_rate_limit`, `backlog_overflow` when undelivered events were - * shed to stay inside a backlog cap, or `suspended_backlog_expired` when a - * suspended subscription held them past its deadline. + * shed to stay inside a backlog cap, `suspended_backlog_expired` when a + * suspended subscription held them past its deadline, or `handler_rejected` + * when the subscription's handler refused the delivery outright. * @property {number} ts Milliseconds since the epoch. */ @@ -73,12 +74,23 @@ * What a handler is called with. An object rather than the event itself, so * more can be added to the call without breaking existing handlers. * + * The last three arrive only on a persistent subscription, and are what let one + * handler body run unchanged here and in the app's events worker. + * * @typedef {Object} EventDelivery * @property {PuterEvent | PuterKvEvent | EventGapMarker} event The delivered * event, or a gap marker in place of events that were dropped. * @property {Readonly>} [ctx] The `context` the * subscription was created with, frozen. Present only for a persistent * subscription; a session subscription carries none. + * @property {unknown} [user] A `puter` bound to the account holding the + * subscription — the ambient one when the handler runs in a client. + * @property {(input: unknown, init?: unknown) => Promise} [fetch] + * `puter.net.fetch` where it exists, the environment's `fetch` otherwise. + * @property {() => Promise} [ack] Settles the delivery. Present only on a + * `single` subscription: calling it hands the delivery back as taken, + * returning without calling it does the same, and throwing does neither — the + * delivery is offered again when its lease lapses. */ /** @@ -161,6 +173,9 @@ * being removed, or `null` while it is live. * @property {string | null} suspendedReason Why it stopped — * `handler_not_found`, `failures`, `no_credit`, or `permission_revoked`. + * @property {() => Promise} [off] Ends the subscription: stops running + * its handler here and unsubscribes it. Present on the subscription + * `onPersistent()` returns, not on one a listing reports. */ /** diff --git a/src/puter-js/tests/api/suites/events.suite.ts b/src/puter-js/tests/api/suites/events.suite.ts index 2db58390c..961edb5ae 100644 --- a/src/puter-js/tests/api/suites/events.suite.ts +++ b/src/puter-js/tests/api/suites/events.suite.ts @@ -30,15 +30,15 @@ const sleep = (ms: number): Promise => new Promise((resolve) => setTimeout(resolve, ms)); const waitFor = async ( - condition: () => boolean, + condition: () => boolean | Promise, timeoutMs: number, ): Promise => { const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { - if (condition()) return true; + if (await condition()) return true; await sleep(50); } - return condition(); + return await condition(); }; const unique = (prefix: string): string => @@ -71,6 +71,61 @@ const makeApp = async (t: TestContext): Promise => { return (app as unknown as { uid: string }).uid; }; +/** + * A handler that closes over nothing and reports what it ran on through the + * one thing it is allowed to reach. Published and subscribed as the same + * function, so the hash the subscribe sends is the hash that was published. + */ +const RECORDING_HANDLER = async ({ + event, + ctx, +}: { + event: { path?: string }; + ctx: Record; +}) => { + const seen = ((globalThis as Record).puterEventsSeen ?? + []) as unknown[]; + seen.push({ path: event.path, label: ctx.label }); + (globalThis as Record).puterEventsSeen = seen; +}; + +const recorded = (): Array<{ path?: string; label?: unknown }> => + ((globalThis as Record).puterEventsSeen ?? []) as Array<{ + path?: string; + label?: unknown; + }>; + +/** Run as the app rather than as the account, then put the session back. */ +const asApp = async ( + t: TestContext, + appUid: string, + run: () => Promise, +): Promise => { + const response = await fetch(`${t.env.apiOrigin}/auth/get-user-app-token`, { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${t.env.users.user.token}`, + }, + body: JSON.stringify({ app_uid: appUid }), + }); + const { token } = (await response.json()) as { token: string }; + // setAuthToken adopts the app identity from an app token's claims and never + // drops it, so the shared instance would keep resolving relative paths + // under ~/AppData/ for every later test unless it is put back here. + const sdk = t.puter as unknown as { appID?: string; appDataPath?: string }; + const { appID, appDataPath } = sdk; + t.puter.setAuthToken(token); + try { + return await run(); + } finally { + t.puter.setAuthToken(t.env.users.user.token); + sdk.appID = appID; + sdk.appDataPath = appDataPath; + } +}; + + export default suite('events', { 'exposes onLocal': async (t) => { t.assert.ok(t.puter.events, 'puter.events is registered'); @@ -438,6 +493,144 @@ export default suite('events', { t.assert.equal(codeOf(error), 'subscription_does_not_exist'); }, + 'ends a persistent subscription through the handle it returns': async (t) => { + const dir = await makeDir(t, 'events-off'); + + const sub = await t.puter.events.onPersistent({ subject: `fs:${dir}` }); + t.assert.equal(typeof sub.off, 'function'); + await sub.off!(); + + const after = await t.puter.events.list(); + t.assert.ok( + ! after.some((row) => row.subId === sub.subId), + 'off() ends the subscription it was called on', + ); + }, + + 'runs the handler here while this client is the one connected': async (t) => { + const dir = await makeDir(t, 'events-durable'); + const appUid = await makeApp(t); + await t.puter.events.handlers.publish('ingestUpload', RECORDING_HANDLER, { + appUid, + }); + // Enough to watch the folder, write into it, and run in the background. + await t.puter.perms.grantApp(appUid, `fs:${dir}:write`); + await t.puter.perms.grantApp(appUid, 'events:background'); + + const before = recorded().length; + await asApp(t, appUid, async () => { + // A session subscription first: its answer is proof the connection + // is up, and the persistent deliveries ride the same one. Without + // it the write can land before the socket registers, and the + // delivery goes looking for the app's events worker instead. + const live = await open(t, `fs:${dir}`, () => {}); + if (! live) return; + + const sub = await t.puter.events.onPersistent({ + subject: `fs:${dir}`, + delivery: 'single', + handlerName: 'ingestUpload', + handler: RECORDING_HANDLER, + context: { label: 'ingest' }, + }); + try { + await t.puter.fs.write(`${dir}/first.txt`, 'one'); + await waitFor( + () => recorded().length > before, + DELIVERY_TIMEOUT_MS, + ); + const first = recorded()[before]; + t.assert.ok(first, 'the persistent handler ran in this client'); + t.assert.equal(first?.path, `${dir}/first.txt`); + t.assert.equal( + first?.label, + 'ingest', + 'the handler is handed the context the subscription carries', + ); + + // A `single` hands out one delivery at a time, so a second one + // arriving is the acknowledgement of the first. + await t.puter.fs.write(`${dir}/second.txt`, 'two'); + const settled = await waitFor( + () => recorded().length > before + 1, + DELIVERY_TIMEOUT_MS, + ); + t.assert.ok( + settled, + 'the first delivery was acknowledged, so the next was handed over', + ); + } finally { + await sub.off!(); + await live.off(); + } + }); + }, + + 'takes background delivery only once the app is allowed it': async (t) => { + const dir = await makeDir(t, 'events-consent'); + const appUid = await makeApp(t); + await t.puter.events.handlers.publish('ingestUpload', RECORDING_HANDLER, { + appUid, + }); + await t.puter.perms.grantApp(appUid, `fs:${dir}:list`); + + const refused = await asApp(t, appUid, () => + t.assert.rejects(() => + t.puter.events.onPersistent({ + subject: `fs:${dir}`, + delivery: 'single', + handlerName: 'ingestUpload', + handler: RECORDING_HANDLER, + }), + ), + ); + t.assert.equal(codeOf(refused), 'events_background_consent_required'); + + await t.puter.perms.grantApp(appUid, 'events:background'); + const subId = await asApp(t, appUid, async () => { + const sub = await t.puter.events.onPersistent({ + subject: `fs:${dir}`, + delivery: 'single', + handlerName: 'ingestUpload', + handler: RECORDING_HANDLER, + }); + return sub.subId; + }); + + // Taking the consent back stops the subscription it allowed. + await t.puter.perms.revokeApp(appUid, 'events:background'); + const settled = await waitFor(async () => { + const held = await t.puter.events.list(); + return ( + held.find((row) => row.subId === subId)?.suspendedReason === + 'permission_revoked' + ); + }, DELIVERY_TIMEOUT_MS); + t.assert.ok(settled, 'the subscription settled when consent was taken back'); + + await t.puter.events.unsubscribe(subId); + }, + + 'needs no consent for a subscription only this connection hears': async (t) => { + const dir = await makeDir(t, 'events-no-consent'); + const appUid = await makeApp(t); + await t.puter.perms.grantApp(appUid, `fs:${dir}:list`); + + await asApp(t, appUid, async () => { + // Session subscriptions are socket-only by construction. + const sub = await open(t, `fs:${dir}`, () => {}); + if (sub) await sub.off(); + + // So is a durable one that asks for nothing else. + const durable = await t.puter.events.onPersistent({ + subject: `fs:${dir}`, + targets: ['socket'], + }); + t.assert.deepEqual(durable.targets, ['socket']); + await durable.off!(); + }); + }, + 'refuses to bind an inline handler nothing is published for': async (t) => { const dir = await makeDir(t, 'events-unbound'); const error = await t.assert.rejects(() =>