mirror of
https://github.com/HeyPuter/puter.git
synced 2026-09-11 15:55:44 +00:00
feat: background delivery seam, retries, and consent (PUT-1682) (#3684)
This commit is contained in:
Daniel Salazar
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
// 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 worker>/__events/invoke
|
||||
* puter-auth: <subscriber-scoped access token>
|
||||
* { 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<string | null>;
|
||||
}
|
||||
|
||||
/** The stand-in until something can address an app's events worker. */
|
||||
export class UnresolvedEventsWorkerResolver implements EventsWorkerResolver {
|
||||
resolveInvokeUrl(): Promise<string | null> {
|
||||
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<WorkerInvokeResult> {
|
||||
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';
|
||||
};
|
||||
@@ -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,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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<typeof setInterval> | 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<number> {
|
||||
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<boolean> {
|
||||
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<DurableSubscription[]> {
|
||||
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<DurableSubscription[]> {
|
||||
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<WorkerInvocationOutcome, 'settled' | 'deferred'>,
|
||||
): Promise<void> {
|
||||
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<string | null> {
|
||||
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);
|
||||
|
||||
@@ -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 ------------------------------------------------------
|
||||
|
||||
/**
|
||||
|
||||
@@ -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<ApiResponse> =>
|
||||
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'],
|
||||
|
||||
@@ -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', () => {
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<Actor> => {
|
||||
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`);
|
||||
|
||||
@@ -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', () => {
|
||||
|
||||
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
/**
|
||||
* 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<string, unknown>;
|
||||
ctx?: Record<string, unknown>;
|
||||
};
|
||||
}
|
||||
|
||||
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<string> => {
|
||||
const chunks: Buffer[] = [];
|
||||
for await (const chunk of req) chunks.push(chunk as Buffer);
|
||||
return Buffer.concat(chunks).toString('utf8');
|
||||
};
|
||||
|
||||
const startStub = async (): Promise<string> => {
|
||||
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<void>((resolve) =>
|
||||
stub.listen(0, '127.0.0.1', () => resolve()),
|
||||
);
|
||||
return `http://127.0.0.1:${(stub.address() as AddressInfo).port}`;
|
||||
};
|
||||
|
||||
const subscribe = async (): Promise<string> => {
|
||||
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<string> => {
|
||||
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<void> => {
|
||||
await env.server.services.fs.touch(userId, { path: `${anchor}/${name}` });
|
||||
};
|
||||
|
||||
const invoked = (count: number): Promise<void> =>
|
||||
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<number> => {
|
||||
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<void>((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),
|
||||
});
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -17,16 +17,15 @@
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
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<WorkerInvocationOutcome>;
|
||||
}
|
||||
|
||||
/** Mints the token one invocation carries, scoped to the subscriber. */
|
||||
export type SubscriberTokenMinter = (
|
||||
invocation: WorkerInvocation,
|
||||
) => Promise<string | null>;
|
||||
|
||||
/**
|
||||
* 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<EventsWorkerInvokerClient, 'invoke'>;
|
||||
readonly #mintToken: SubscriberTokenMinter;
|
||||
|
||||
constructor(
|
||||
client: Pick<EventsWorkerInvokerClient, 'invoke'>,
|
||||
mintToken: SubscriberTokenMinter,
|
||||
) {
|
||||
this.#client = client;
|
||||
this.#mintToken = mintToken;
|
||||
}
|
||||
|
||||
async invoke(
|
||||
invocation: WorkerInvocation,
|
||||
): Promise<WorkerInvocationOutcome> {
|
||||
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;
|
||||
|
||||
|
||||
@@ -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'] }),
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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:{<subId>} HASH entryId -> the delivery and its attempt count
|
||||
* ev:q:{<subId>} HASH entryId -> the delivery and its attempt counts
|
||||
* ev:qp:{<subId>} ZSET entryId -> enqueued at; membership means unsettled
|
||||
* ev:ql:{<subId>} ZSET entryId -> lease expiry; membership means in flight
|
||||
* ev:qf:{<subId>} 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<DeferredDelivery> {
|
||||
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<boolean> {
|
||||
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<number> {
|
||||
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<void> {
|
||||
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);
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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;
|
||||
};
|
||||
|
||||
/**
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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. |
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:<app_uid>:<read|write>
|
||||
if ( parts[2] === 'read' ) {
|
||||
|
||||
@@ -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.',
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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<string, EventSubscription>} */
|
||||
this.byId = new Map();
|
||||
/**
|
||||
* @internal Persistent subscriptions this client is running the
|
||||
* handler for, by their server-side id.
|
||||
* @type {Map<string, DurableRegistration>}
|
||||
*/
|
||||
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<string, unknown>} [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<string, unknown> | 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<string, unknown> | 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<void>}
|
||||
*/
|
||||
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<Record<string, unknown>>} 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);
|
||||
}
|
||||
};
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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', () => {
|
||||
|
||||
@@ -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<Record<string, unknown>>} [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<unknown>} [fetch]
|
||||
* `puter.net.fetch` where it exists, the environment's `fetch` otherwise.
|
||||
* @property {() => Promise<void>} [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<void>} [off] Ends the subscription: stops running
|
||||
* its handler here and unsubscribes it. Present on the subscription
|
||||
* `onPersistent()` returns, not on one a listing reports.
|
||||
*/
|
||||
|
||||
/**
|
||||
|
||||
@@ -30,15 +30,15 @@ const sleep = (ms: number): Promise<void> =>
|
||||
new Promise((resolve) => setTimeout(resolve, ms));
|
||||
|
||||
const waitFor = async (
|
||||
condition: () => boolean,
|
||||
condition: () => boolean | Promise<boolean>,
|
||||
timeoutMs: number,
|
||||
): Promise<boolean> => {
|
||||
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<string> => {
|
||||
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<string, unknown>;
|
||||
}) => {
|
||||
const seen = ((globalThis as Record<string, unknown>).puterEventsSeen ??
|
||||
[]) as unknown[];
|
||||
seen.push({ path: event.path, label: ctx.label });
|
||||
(globalThis as Record<string, unknown>).puterEventsSeen = seen;
|
||||
};
|
||||
|
||||
const recorded = (): Array<{ path?: string; label?: unknown }> =>
|
||||
((globalThis as Record<string, unknown>).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>(
|
||||
t: TestContext,
|
||||
appUid: string,
|
||||
run: () => Promise<T>,
|
||||
): Promise<T> => {
|
||||
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/<app> 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(() =>
|
||||
|
||||
Reference in New Issue
Block a user