diff --git a/src/backend/controllers/events/limits.ts b/src/backend/controllers/events/limits.ts index c4b170d86..7ba803d3e 100644 --- a/src/backend/controllers/events/limits.ts +++ b/src/backend/controllers/events/limits.ts @@ -75,6 +75,15 @@ export const EVENTS_SUBSCRIBE_LIMIT = userWindow('events:subscribe', 60); */ export const EVENTS_LIST_LIMIT = userWindow('events:list', 120); +/** + * Delivery acknowledgements per minute, per user. + * + * One ack per `single` delivery, so this sits level with what one subscription + * may be delivered — a client acking faster than that is acking things it was + * never sent. + */ +export const EVENTS_ACK_LIMIT = userWindow('events:ack', 600); + // -- Dispatch fan-out ------------------------------------------------ /** @@ -103,6 +112,29 @@ export const EVENTS_BROADCAST_DELIVERY_LIMIT = userWindow( */ export { FILTER_EVALUATIONS_PER_EVENT } from '../../services/events/matcher.js'; +// -- Undelivered backlog --------------------------------------------- + +/** + * Deliveries one subscription may hold undelivered. + * + * A `single` delivery waits until something takes it, so a subscription whose + * consumer is gone accumulates. Over the cap the oldest go and one gap marker + * takes their place, which is what keeps "at-least-once" honest: what was lost + * is visible rather than silent. + */ +export const EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION = 10_000; + +/** + * Undelivered deliveries one region may hold across every subscription. + * + * The per-subscription cap bounds one backlog and nothing in aggregate — + * multiply it by the subscriptions that can exist and the region's memory is + * the only remaining limit. Over this, the oldest deliveries in the region are + * shed first, each shedding subscription gets a gap marker, and an alarm says + * it happened. + */ +export const EVENTS_REGION_PENDING_CEILING = 1_000_000; + // -- Coalescing ------------------------------------------------------ /** diff --git a/src/backend/services/events/EventsService.test.ts b/src/backend/services/events/EventsService.test.ts index f42c4e9c6..f1214b32a 100644 --- a/src/backend/services/events/EventsService.test.ts +++ b/src/backend/services/events/EventsService.test.ts @@ -35,6 +35,7 @@ import type { FSEntry } from '../../stores/fs/FSEntry.js'; import type { IConfig } from '../../types.js'; import { EventsService, + EVENTS_ACK_VERB, EVENTS_DELIVERY_CHANNEL, EVENTS_SUBSCRIBE_VERB, EVENTS_UNSUBSCRIBE_VERB, @@ -183,6 +184,7 @@ const appStore = { */ const durableSubscriptionStore = { warmRegion: async () => false, + getBySubId: async () => null, }; /** @@ -383,9 +385,27 @@ describe('subscribing', () => { anchor: { uid: documents.uid, path: documents.path }, match: null, op: null, + // The connection is the only thing a session row can be delivered + // to, so it is the only transport it can ask for. + targets: ['socket'], }); }); + it('refuses a session subscription a connection could not carry', async () => { + seedTree(); + + for (const targets of [['worker'], ['socket', 'push']]) + await expect( + service.subscribe(actorFor(), socketId, { + subject: `fs:/u${userId}/Documents`, + targets, + }), + ).rejects.toSatisfy( + (err: unknown) => + isHttpError(err) && err.legacyCode === 'invalid_targets', + ); + }); + it('rejects a subject that resolves to nothing', async () => { await expect(subscribe('fs:/nowhere/at/all')).rejects.toSatisfy( (err: unknown) => @@ -1154,6 +1174,26 @@ describe('the socket surface', () => { expect(ack.mock.calls[0][0]).toEqual({ ok: true }); }); + it('answers an ack for a subscription the caller does not hold', async () => { + const socket = fakeSocket(); + service.attachSocket(socket, actorFor()); + + const ack = vi.fn(); + socket.fire( + EVENTS_ACK_VERB, + { subId: 'app#someone-elses', id: '1-0' }, + ack, + ); + await vi.waitFor(() => expect(ack).toHaveBeenCalled()); + + expect(ack.mock.calls[0][0]).toEqual({ + ok: false, + error: expect.objectContaining({ + code: 'subscription_does_not_exist', + }), + }); + }); + it('addresses deliveries at the socket that asked for them', async () => { vi.useFakeTimers(); const { documents, file } = seedTree(); diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index 750ea94ef..6abba55aa 100644 --- a/src/backend/services/events/EventsService.ts +++ b/src/backend/services/events/EventsService.ts @@ -20,6 +20,7 @@ import { randomUUID } from 'node:crypto'; import type { EventKey } from '../../clients/event/types.js'; import { + EVENTS_ACK_LIMIT, EVENTS_BROADCAST_DELIVERY_LIMIT, EVENTS_COALESCE_WINDOW_MS, EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT, @@ -35,8 +36,15 @@ import { type GenerationBump, type SessionSubscription, } from '../../stores/events/EventSubscriptionStore.js'; +import type { + ClaimedDelivery, + PendingShed, +} from '../../stores/events/PendingDeliveryStore.js'; import { + DEFAULT_DURABLE_TARGETS, + SESSION_TARGETS, isSubscriptionTarget, + targetsAllowedForDelivery, type SubscriptionTarget, } from '../../stores/events/types.js'; import type { FSEntry } from '../../stores/fs/FSEntry.js'; @@ -66,13 +74,21 @@ import { } from './matcher.js'; import { lookupPublicSubject, + type DeliverableEvent, type DeliveryClass, type EventContext, + type GapMarker, + type GapReason, type ProjectedEvent, type PublicSubject, } from './registry.js'; import { SubscriptionCache } from './subscriptionCache.js'; import { parseSubject, type FsOp } from './subjects.js'; +import { + RecordingWorkerInvoker, + type WorkerInvocation, + type WorkerInvokerSeam, +} from './workerSeam.js'; /** * Subscribe, unsubscribe, and the dispatch hot path. @@ -100,12 +116,19 @@ import { parseSubject, type FsOp } from './subjects.js'; export interface SubscribeRequest { subject?: unknown; + targets?: unknown; } export interface UnsubscribeRequest { subId?: unknown; } +/** Body of the `events.ack` verb: which subscription, and which delivery. */ +export interface AckRequest { + subId?: unknown; + id?: unknown; +} + /** Body of `POST /events/subscribe`. */ export interface DurableSubscribeRequest extends SubscribeRequest { delivery?: unknown; @@ -128,6 +151,7 @@ export interface SubscriptionView { anchor: { uid: string; path: string }; match: string | null; op: FsOp | null; + targets: SubscriptionTarget[]; } /** @@ -137,7 +161,6 @@ export interface SubscriptionView { */ export interface DurableSubscriptionView extends SubscriptionView { delivery: DeliveryClass; - targets: SubscriptionTarget[]; handlerName: string | null; appUid: string | null; createdAt: number; @@ -150,35 +173,28 @@ export type VerbAck = | ({ ok: true } & T) | { ok: false; error: { code: string; message: string } }; -export type GapReason = - | 'matched_subscription_limit' - | 'filter_evaluation_limit' - | 'delivery_rate_limit'; - -/** - * `gap` says an event existed and was not delivered. It rides the delivery - * channel because a subscriber that never saw one would read the silence as - * "nothing happened", and carries no `uid`/`path` — what was dropped is exactly - * what it cannot name. - */ -export interface GapMarker { - id: string; - subject: string; - op: 'gap'; - reason: GapReason; - ts: number; -} +export type { GapMarker, GapReason } from './registry.js'; /** One delivery, as the client receives it. */ export interface DeliveryEnvelope { subId: string; - event: ProjectedEvent | GapMarker; + event: DeliverableEvent; + /** Set on a `single`: nothing else takes this until `events.ack` settles it. */ + ackRequired?: true; + /** The handle `events.ack` names — the delivery, not the event. */ + ackId?: string; } -/** The envelope plus where it goes. The address is not part of the wire. */ +/** + * The envelope plus where it goes. The address is not part of the wire, and + * neither is the handler a durable row may want run alongside the socket copy. + */ interface AddressedDelivery { target: SocketSpecifier; envelope: DeliveryEnvelope; + /** False for a row that asked for its handler and no socket copy. */ + socket: boolean; + worker?: WorkerInvocation; } /** What a dispatch call site can supply that the event itself does not carry. */ @@ -198,6 +214,7 @@ export interface FsDispatchOptions { export const EVENTS_SUBSCRIBE_VERB = 'events.subscribe'; export const EVENTS_UNSUBSCRIBE_VERB = 'events.unsubscribe'; +export const EVENTS_ACK_VERB = 'events.ack'; export const EVENTS_DELIVERY_CHANNEL = 'events.delivery'; // -- Expiry sweep ----------------------------------------------------- @@ -212,6 +229,27 @@ const EXPIRY_BATCH_SIZE = 500; /** Batches one sweep takes, so a large backlog drains over several passes. */ const EXPIRY_MAX_BATCHES = 50; +// -- Owed deliveries -------------------------------------------------- + +/** + * Socket attempts one `single` delivery gets before its handler takes it. Two, + * because the second is what a client that reconnected — or a second connection + * of the same account — is worth trying; a third is just a slower hand-off. + */ +const SINGLE_SOCKET_ATTEMPTS = 2; + +/** How often expired leases are reclaimed and owed deliveries retried. */ +const PENDING_SWEEP_INTERVAL_MS = 10_000; + +/** Subscriptions one sweep pass looks at, taken from the oldest end. */ +const PENDING_SWEEP_SUBSCRIPTIONS = 100; + +/** + * Deliveries one subscription may be handed in a row. A consumer that settles + * inline would otherwise drain a whole backlog inside one ack. + */ +const PENDING_DRAIN_BATCH = 25; + /** The part of a socket this service uses, so tests need not build one. */ export interface EventSocket { id: string; @@ -236,10 +274,9 @@ const tooManyCalls = (): HttpError => legacyCode: 'too_many_requests', }); -/** Stands until there is a pending-delivery store to take a `single` lease. */ -const deliveryClassUnavailable = (): HttpError => - new HttpError(501, 'Delivery class `single` is not available yet', { - legacyCode: 'delivery_class_unavailable', +const handlerRequired = (): HttpError => + new HttpError(400, 'A `single` subscription needs a handlerName', { + legacyCode: 'events_handler_required', }); const badRequest = (message: string, code: string): HttpError => @@ -266,12 +303,12 @@ const toView = (sub: DispatchSubscription): SubscriptionView => ({ anchor: { uid: sub.anchorUid, path: sub.anchorPath }, match: sub.match, op: sub.op, + targets: sub.targets ?? SESSION_TARGETS, }); const toDurableView = (sub: DurableSubscription): DurableSubscriptionView => ({ ...toView(sub), delivery: sub.delivery, - targets: sub.targets, handlerName: sub.handlerName, appUid: sub.appUid, createdAt: sub.createdAt, @@ -300,10 +337,17 @@ const deliveryTarget = (row: DispatchSubscription): SocketSpecifier => { }; }; +/** Transports a row is asking for, whichever store it came from. */ +const targetsOf = (row: DispatchSubscription): SubscriptionTarget[] => + row.durable === true + ? (row.targets ?? DEFAULT_DURABLE_TARGETS) + : SESSION_TARGETS; + /** - * Rows this pass can actually deliver. Durable `single` rows need the pending - * store to take a lease, and a row with no socket target has asked not to be - * delivered over one. + * Whether this pass has anywhere to put the row. A `single` is queued whether + * or not anything is listening right now — that is what it is for — while a + * `broadcast` whose only transport is one this build cannot carry has no + * target, and an event with no target is dropped rather than held. */ const nowSeconds = (): number => Math.floor(Date.now() / 1000); @@ -314,30 +358,33 @@ const unexpired = (row: DispatchSubscription): boolean => { return expiresAt === null || expiresAt > nowSeconds(); }; -const deliverableOverSockets = (row: DispatchSubscription): boolean => - unexpired(row) && - (row.durable !== true || - (row.delivery === 'broadcast' && - (row.targets ?? []).includes('socket'))); +const deliverable = (row: DispatchSubscription): boolean => { + if (row.durable !== true) return true; + if (!unexpired(row)) return false; + if (row.delivery === 'single') return true; + const targets = targetsOf(row); + return targets.includes('socket') || targets.includes('worker'); +}; + +const isSingle = (row: DispatchSubscription): boolean => + row.durable === true && row.delivery === 'single'; // -- Durable request parsing ------------------------------------------ -/** Transports a durable row takes unless the caller says otherwise. */ -const DEFAULT_DURABLE_TARGETS: SubscriptionTarget[] = ['socket', 'worker']; - /** Longest a `handlerName` may be, matching the column that holds it. */ const HANDLER_NAME_MAX_LENGTH = 128; const parseDelivery = (value: unknown): DeliveryClass => { if (value === undefined || value === null || value === 'broadcast') return 'broadcast'; - // Creatable but inert is worse than refused: a `single` row would take a - // lease nothing in this build can settle. - if (value === 'single') throw deliveryClassUnavailable(); + if (value === 'single') return 'single'; throw badRequest(`Unknown delivery class: ${String(value)}`, 'bad_request'); }; -const parseTargets = (value: unknown): SubscriptionTarget[] => { +const parseTargets = ( + value: unknown, + delivery: DeliveryClass, +): SubscriptionTarget[] => { if (value === undefined || value === null) return DEFAULT_DURABLE_TARGETS; if (!Array.isArray(value) || value.length === 0) throw badRequest( @@ -346,7 +393,34 @@ const parseTargets = (value: unknown): SubscriptionTarget[] => { ); if (!value.every(isSubscriptionTarget)) throw badRequest('Unknown delivery target', 'invalid_targets'); - return [...new Set(value)]; + + const targets = [...new Set(value)]; + if (!targetsAllowedForDelivery(delivery, targets)) + throw badRequest( + 'A `single` subscription needs a `worker` target and may not target `push`', + 'invalid_targets', + ); + return targets; +}; + +/** + * A session row is one connection, so the socket is the only transport it can + * have: a handler runs long after the connection is gone, and a device + * notification is not addressed to a connection at all. + */ +const parseSessionTargets = (value: unknown): SubscriptionTarget[] => { + if (value === undefined || value === null) return SESSION_TARGETS; + if (!Array.isArray(value) || value.length === 0) + throw badRequest( + 'targets must be a non-empty array', + 'invalid_targets', + ); + if (!value.every((target) => target === 'socket')) + throw badRequest( + 'A session subscription may only target `socket`', + 'invalid_targets', + ); + return SESSION_TARGETS; }; const parseHandlerName = (value: unknown): string | null => { @@ -394,6 +468,14 @@ export class EventsService extends PuterService { #coalescer: DeliveryCoalescer | null = null; #expirySweep: ReturnType | null = null; #expiryKick: ReturnType | null = null; + #pendingSweep: ReturnType | null = null; + + /** + * What runs an app's handler. The default records the intent and settles + * nothing, so a delivery handed to it stays owed until there is a real + * invoker to take it. + */ + worker: WorkerInvokerSeam = new RecordingWorkerInvoker(); // -- Lifecycle --------------------------------------------------- @@ -413,6 +495,7 @@ export class EventsService extends PuterService { }, ); this.#armExpirySweep(); + this.#armPendingSweep(); } override onServerPrepareShutdown(): void { @@ -420,6 +503,8 @@ export class EventsService extends PuterService { this.#expiryKick = null; if (this.#expirySweep) clearInterval(this.#expirySweep); this.#expirySweep = null; + if (this.#pendingSweep) clearInterval(this.#pendingSweep); + this.#pendingSweep = null; } override onServerShutdown(): void { @@ -462,6 +547,13 @@ export class EventsService extends PuterService { }); }) as (...args: never[]) => void); + socket.on(EVENTS_ACK_VERB, ((payload: AckRequest, ack: unknown) => { + void this.#answer(ack, async () => { + await this.ackDelivery(actor, payload); + return {}; + }); + }) as (...args: never[]) => void); + socket.once('disconnect', (() => { void this.reapSocket(userId, socket.id); }) as (...args: never[]) => void); @@ -500,6 +592,7 @@ export class EventsService extends PuterService { await this.#spendCallBudget(holderUserId); + const targets = parseSessionTargets(request?.targets); const rawSubject = String(request?.subject ?? ''); const anchor = await this.#resolveSubscribeAnchor(actor, rawSubject); @@ -516,6 +609,7 @@ export class EventsService extends PuterService { op: anchor.op, appUid: actor.effectiveApp?.uid ?? null, permission: anchor.permission, + targets, }; const bump = await this.stores.eventSubscription.add(sub); @@ -587,11 +681,16 @@ export class EventsService extends PuterService { await this.#spendCallBudget(holderUserId); const delivery = parseDelivery(request?.delivery); - const targets = parseTargets(request?.targets); + const targets = parseTargets(request?.targets, delivery); const handlerName = parseHandlerName(request?.handlerName); const context = parseContext(request?.context); const expiresAt = parseExpiresAt(request?.expiresAt); + // A `single` is owed to exactly one consumer, and the handler is the + // only one that is always there to take it. Whether the handler exists + // is the publish surface's question, not this one's. + if (delivery === 'single' && !handlerName) throw handlerRequired(); + const rawSubject = String(request?.subject ?? ''); const anchor = await this.#resolveSubscribeAnchor(actor, rawSubject); @@ -676,10 +775,47 @@ export class EventsService extends PuterService { throw unknownSubscription(); const bump = await this.stores.durableSubscription.remove(row); + // Whatever it was still owed goes with it: a backlog held for a + // subscription nobody can consume is memory, and the paths it names are + // ones its holder just gave up asking about. + await this.stores.pendingDelivery.purge(subId); this.#forget(subId); this.#publishGeneration(bump, true); } + /** + * Settle a `single` delivery the client took. An id this subscription is + * not holding — one already settled, or one reclaimed and handed on — is + * not an error: at-least-once makes a duplicate ack routine. + */ + async ackDelivery(actor: Actor, request: AckRequest): Promise { + if (!this.enabled) throw disabled(); + const holderUserId = actor.user?.id; + if (holderUserId === undefined) throw disabled(); + + const subId = String(request?.subId ?? ''); + const entryId = String(request?.id ?? ''); + if (!subId || !entryId) throw unknownSubscription(); + + const ok = await checkRateLimit( + `${EVENTS_ACK_LIMIT.scope}:${holderUserId}`, + EVENTS_ACK_LIMIT.limit, + EVENTS_ACK_LIMIT.window, + ); + if (!ok) throw tooManyCalls(); + + const row = await this.stores.durableSubscription.getBySubId(subId); + if ( + !row || + row.holderUserId !== holderUserId || + !rowInActorScope(actor, row) + ) + throw unknownSubscription(); + + await this.stores.pendingDelivery.settle(subId, entryId); + await this.#drain(row); + } + /** * Drop rows past their expiry, in batches, and report how many went. Every * node sweeps; the delete is idempotent, so two overlapping costs a few @@ -699,6 +835,59 @@ export class EventsService extends PuterService { return removed; } + /** + * Retry what nobody took. Reads the pending index from its oldest end — the + * subscriptions that have waited longest — so finding the work costs one + * ordered read rather than a walk of the keyspace, which is exactly the + * thing that gets expensive when there is a backlog to find. + * + * A lease that lapsed is what makes a delivery claimable again, so this + * needs no notion of failure: it retries whatever is not currently held. + */ + async sweepPending(): Promise { + if (!this.enabled) return 0; + + // A partial write can move a pending set without its share of the + // region counter going with it, in either direction. This is what + // keeps that drift from being permanent. + await this.stores.pendingDelivery + .reconcileRegionDepth() + .catch((err) => { + console.warn('[events] pending counter reconcile failed', err); + }); + + let attempted = 0; + for (const { subId } of await this.stores.pendingDelivery.head( + PENDING_SWEEP_SUBSCRIPTIONS, + )) { + try { + const row = + await this.stores.durableSubscription.getBySubId(subId); + // Nothing left to deliver to, so nothing left to hold. + if (!row || this.#isOver(row)) { + await this.stores.pendingDelivery.purge(subId); + continue; + } + // A suspended row keeps what it is owed — what happens to that + // backlog is the suspension's decision, not the sweeper's. It + // goes to the back of the line so it cannot hold the head. + if (row.suspendedAt !== null) { + await this.stores.pendingDelivery.defer(subId); + continue; + } + attempted += await this.#drain(row); + } catch (err) { + console.warn('[events] pending sweep failed', subId, err); + } + } + return attempted; + } + + /** Past its expiry, so the daily reaper is only a matter of time. */ + #isOver(row: DurableSubscription): boolean { + return row.expiresAt !== null && row.expiresAt <= Date.now() / 1000; + } + /** * Resolve, authorize and compile one subscribe request. Shared so a durable * row cannot be created under a weaker check than a session one. @@ -828,7 +1017,7 @@ export class EventsService extends PuterService { candidates: DispatchSubscription[], actingUserId: number | undefined, ): Promise { - const rows = candidates.filter(deliverableOverSockets); + const rows = candidates.filter(deliverable); if (rows.length === 0) return; // One throwaway projection reads the op off the registry entry rather @@ -856,9 +1045,21 @@ export class EventsService extends PuterService { actingUserId === row.holderUserId, seq: seq++, }); + + // A `single` is owed rather than sent: it is queued, and never + // coalesced or broadcast — collapsing two of them would drop one + // the subscription was promised. + if (isSingle(row)) { + await this.#owe(row, event); + continue; + } + + const targets = targetsOf(row); this.#coalesce().push(coalesceKey(row.subId, event.subject), { target: deliveryTarget(row), envelope: { subId: row.subId, event }, + socket: targets.includes('socket'), + worker: this.#workerInvocation(row, event), }); } @@ -951,20 +1152,191 @@ export class EventsService extends PuterService { context: EventContext, reason: GapReason, ): void { - for (const row of rows) + for (const row of rows) { + const marker: GapMarker = { + id: context.id, + subject: subject.subject, + op: 'gap', + reason, + ts: context.ts, + }; + // A marker is a delivery, so it takes the same route its + // subscription's events would: queued for a `single`, sent for the + // rest. + if (isSingle(row)) { + void this.#owe(row, marker); + continue; + } + // A marker rides the socket; a row with none has nowhere to hear + // it, and sending nothing must not count as a delivery. + if (!targetsOf(row).includes('socket')) continue; this.#send({ target: deliveryTarget(row), + socket: true, + envelope: { subId: row.subId, event: marker }, + }); + } + } + + // -- Owed deliveries --------------------------------------------- + + /** + * Queue a `single` and try to hand it straight over. The queue comes first: + * an attempt that fails after it is recorded is a retry, where one that + * fails before it is a lost event. + */ + async #owe( + row: DispatchSubscription, + event: DeliverableEvent, + ): Promise { + try { + const { shed } = await this.stores.pendingDelivery.enqueue( + row.subId, + event, + ); + this.#reportShed(shed); + await this.#drain(row); + } catch (err) { + this.#enqueueFailed(row, err); + } + } + + /** + * Hand this subscription what it is owed, oldest first, until something + * takes a lease and holds it. Stops after a batch so one consumer that + * settles inline cannot drain a whole backlog inside one call. + */ + async #drain(row: DispatchSubscription): Promise { + let handed = 0; + for (let pass = 0; pass < PENDING_DRAIN_BATCH; pass++) { + const claimed = await this.stores.pendingDelivery.claim(row.subId); + if (!claimed) return handed; + handed++; + // Anything still holding the lease is the next consumer's answer to + // give, so this pass is over. + if (!(await this.#handOut(row, claimed))) return handed; + } + return handed; + } + + /** + * One attempt at one owed delivery. Sockets first and one at a time, the + * handler once they are spent — a client that is there answers faster than + * anything else, and one that is not must not stall the delivery forever. + * + * Returns whether the delivery settled, which is what says the next one may + * go out now rather than when this lease lapses. + */ + async #handOut( + row: DispatchSubscription, + claimed: ClaimedDelivery, + ): Promise { + const targets = targetsOf(row); + const target = deliveryTarget(row); + + // Only this region's own connections are visible here; the ones other + // regions hold arrive with presence. + const overSocket = + targets.includes('socket') && + claimed.socketAttempts < SINGLE_SOCKET_ATTEMPTS && + this.services.socket.has(target); + + if (overSocket) { + await this.stores.pendingDelivery.recordSocketAttempt( + row.subId, + claimed.entryId, + ); + this.#send({ + target, + socket: true, envelope: { subId: row.subId, - event: { - id: context.id, - subject: subject.subject, - op: 'gap', - reason, - ts: context.ts, - }, + event: claimed.event, + ackRequired: true, + ackId: claimed.entryId, }, }); + return false; + } + + // Nowhere to put it yet: the lease is what paces the next attempt. + if (!targets.includes('worker')) return false; + + const invocation = this.#workerInvocation(row, claimed.event); + if (!invocation) return false; + + const outcome = await this.worker.invoke(invocation); + this.onDelivered({ subId: row.subId, event: claimed.event }); + if (outcome !== 'settled') return false; + + await this.stores.pendingDelivery.settle(row.subId, claimed.entryId); + return true; + } + + /** What the handler seam is handed, or null for a row that wants none. */ + #workerInvocation( + row: DispatchSubscription, + event: DeliverableEvent, + ): WorkerInvocation | null { + if (row.durable !== true) return null; + if (!targetsOf(row).includes('worker')) return null; + return { + subId: row.subId, + holderUserId: row.holderUserId, + appUid: row.appUid, + handlerName: row.handlerName ?? null, + event, + context: row.context ?? 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 + * place; this is the half nobody else would see. + */ + #reportShed(shed: readonly PendingShed[]): void { + for (const dropped of shed) { + if (dropped.scope === 'subscription') { + console.warn( + `[events] dropped ${dropped.dropped} undelivered event(s) of ${dropped.subId}: backlog full`, + ); + continue; + } + console.error( + `[events] dropped ${dropped.dropped} undelivered event(s) of ${dropped.subId}: too many held in this region`, + ); + this.clients.alarm.create( + 'events_pending_ceiling', + 'Too many undelivered events are being held here — the oldest were dropped', + { subId: dropped.subId, dropped: dropped.dropped }, + 'warning', + { dedup: true }, + ); + } + } + + /** + * A `single` that could not be queued is an event its subscription was + * promised and will never see, and nothing downstream will notice on its + * own. The write that produced it still succeeded. + */ + #enqueueFailed(row: DispatchSubscription, err: unknown): void { + console.error( + `[events] could not queue an event for ${row.subId}`, + err, + ); + this.clients.alarm.create( + 'events_pending_enqueue_failed', + 'An event owed to a subscription could not be queued and is lost', + { + subId: row.subId, + holderUserId: row.holderUserId, + error: err instanceof Error ? err : new Error(String(err)), + }, + 'warning', + { dedup: true }, + ); } // -- Delivery ---------------------------------------------------- @@ -991,6 +1363,7 @@ export class EventsService extends PuterService { const event = delivery.envelope.event as ProjectedEvent; this.#send({ target: delivery.target, + socket: delivery.socket, envelope: { subId: delivery.envelope.subId, event: { @@ -1010,22 +1383,40 @@ export class EventsService extends PuterService { /** * Addressed at a socket id — which socket.io joins every socket to — or at * a room, so either way the adapter carries it to whichever node terminates - * the connection. + * the connection. A durable row may also want its handler run, which + * happens alongside the socket copy and at most once per delivery. */ #send(delivery: AddressedDelivery): void { - try { - void this.services.socket - .send( - delivery.target, - EVENTS_DELIVERY_CHANNEL, - delivery.envelope, - ) - .catch((err: unknown) => { - console.warn('[events] socket send failed', err); - }); - } catch (err) { - console.warn('[events] socket send failed', err); + if (delivery.socket) { + try { + void this.services.socket + .send( + delivery.target, + EVENTS_DELIVERY_CHANNEL, + delivery.envelope, + ) + .catch((err: unknown) => { + console.warn('[events] socket send failed', err); + }); + } catch (err) { + console.warn('[events] socket send failed', err); + } } + + if (delivery.worker) { + 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. + try { + void this.worker.invoke(invocation).catch((err: unknown) => { + console.warn('[events] handler invocation failed', err); + }); + } catch (err) { + console.warn('[events] handler invocation failed', err); + } + } + this.onDelivered(delivery.envelope); } @@ -1205,6 +1596,17 @@ export class EventsService extends PuterService { this.#expirySweep = sweep; } + #armPendingSweep(): void { + if (!this.enabled) return; + const sweep = setInterval(() => { + void this.sweepPending().catch((err) => { + console.warn('[events] pending sweep failed', err); + }); + }, PENDING_SWEEP_INTERVAL_MS); + sweep.unref?.(); + this.#pendingSweep = sweep; + } + #stopRefresh(holderUserId: number, socketId: string): void { const key = `${holderUserId}|${socketId}`; const timer = this.#refreshTimers.get(key); diff --git a/src/backend/services/events/durable.integration.test.ts b/src/backend/services/events/durable.integration.test.ts index 90f667680..28f08cc5a 100644 --- a/src/backend/services/events/durable.integration.test.ts +++ b/src/backend/services/events/durable.integration.test.ts @@ -220,14 +220,39 @@ describe('creating a durable subscription over HTTP', () => { expect(created.body.context).toBeUndefined(); }); - it('refuses the delivery class that has nowhere to queue yet', async () => { - const refused = await subscribe(env.users.user.token, { + it('registers a subscription owed to one consumer', async () => { + await clearRows(); + const created = await subscribe(env.users.user.token, { delivery: 'single', handlerName: 'onWrite', }); - expect(refused.status).toBe(501); - expect(refused.body.code).toBe('delivery_class_unavailable'); + expect(created.status).toBe(200); + expect(created.body).toMatchObject({ + delivery: 'single', + handlerName: 'onWrite', + targets: ['socket', 'worker'], + }); + }); + + it('refuses one owed to a consumer it cannot name', async () => { + const refused = await subscribe(env.users.user.token, { + delivery: 'single', + }); + + expect(refused.status).toBe(400); + expect(refused.body.code).toBe('events_handler_required'); + }); + + it('refuses one owed to a device notification', async () => { + const refused = await subscribe(env.users.user.token, { + delivery: 'single', + handlerName: 'onWrite', + targets: ['push'], + }); + + expect(refused.status).toBe(400); + expect(refused.body.code).toBe('invalid_targets'); }); it('refuses a target outside the known set', async () => { @@ -239,6 +264,17 @@ 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, { + delivery: 'single', + handlerName: 'onWrite', + targets: ['socket'], + }); + + expect(refused.status).toBe(400); + expect(refused.body.code).toBe('invalid_targets'); + }); + it('refuses an expiry in the past', async () => { const refused = await subscribe(env.users.user.token, { expiresAt: Math.floor(Date.now() / 1000) - 60, diff --git a/src/backend/services/events/registry.ts b/src/backend/services/events/registry.ts index 9b7cbd9b3..ac6968e7a 100644 --- a/src/backend/services/events/registry.ts +++ b/src/backend/services/events/registry.ts @@ -46,6 +46,29 @@ export interface ProjectedEvent { seq: number; } +export type GapReason = + | 'matched_subscription_limit' + | 'filter_evaluation_limit' + | 'delivery_rate_limit' + | 'backlog_overflow'; + +/** + * `gap` says an event existed and was not delivered. It rides the delivery + * channel because a subscriber that never saw one would read the silence as + * "nothing happened", and carries no `uid`/`path` — what was dropped is exactly + * what it cannot name. + */ +export interface GapMarker { + id: string; + subject: string; + op: 'gap'; + reason: GapReason; + ts: number; +} + +/** Either shape a subscriber can be handed. */ +export type DeliverableEvent = ProjectedEvent | GapMarker; + /** What dispatch knows about one internal emit, before it picks subscribers. */ export interface EventContext { key: EventKey; diff --git a/src/backend/services/events/singleDelivery.test.ts b/src/backend/services/events/singleDelivery.test.ts new file mode 100644 index 000000000..ff5148b71 --- /dev/null +++ b/src/backend/services/events/singleDelivery.test.ts @@ -0,0 +1,499 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import MockRedis from 'ioredis-mock'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + EVENTS_COALESCE_WINDOW_MS, + EVENTS_REGION_PENDING_CEILING, +} from '../../controllers/events/limits.js'; +import type { Actor } from '../../core/actor.js'; +import { EventSubscriptionStore } from '../../stores/events/EventSubscriptionStore.js'; +import { PendingDeliveryStore } from '../../stores/events/PendingDeliveryStore.js'; +import type { + DurableSubscription, + SubscriptionTarget, +} from '../../stores/events/types.js'; +import type { FSEntry } from '../../stores/fs/FSEntry.js'; +import type { IConfig } from '../../types.js'; +import { EventsService, type DeliveryEnvelope } from './EventsService.js'; +import { fsAnchorToken } from './subjects.js'; +import type { + WorkerInvocation, + WorkerInvocationOutcome, +} from './workerSeam.js'; + +/** + * The promises each delivery class makes, held against each other. + * + * A `single` is owed to exactly one consumer: the socket if someone is there, + * the handler once the socket has had its turns, and never both. A `broadcast` + * is at-most-once to everyone who is: its handler runs alongside the socket + * copies rather than instead of them, and a row nothing can carry delivers — + * and meters — nothing at all. + */ + +let seq = 0; +let userId = 0; +let redis: InstanceType; +let subscriptions: EventSubscriptionStore; +let pending: PendingDeliveryStore; +let service: EventsService; +let sent: DeliveryEnvelope[]; +let delivered: DeliveryEnvelope[]; +let invoked: WorkerInvocation[]; +let rows: Map; +let entries: Map; +let alarms: ReturnType; + +/** Whether this region holds a connection for the row being delivered to. */ +let socketConnected = true; +/** What the handler seam reports back, which is what settles a lease or not. */ +let workerOutcome: WorkerInvocationOutcome = 'deferred'; + +const entry = (over: Partial = {}): FSEntry => + ({ + uid: `file-${seq}`, + uuid: `file-${seq}`, + path: `/u${userId}/Documents/notes.txt`, + userId, + isDir: false, + ...over, + }) as FSEntry; + +const actorFor = (asUserId = userId): Actor => + ({ + user: { + id: asUserId, + uuid: `user-${asUserId}`, + username: `u${asUserId}`, + }, + effectiveApp: null, + }) as unknown as Actor; + +const anchorUid = (): string => `docs-${seq}`; +const anchorPath = (): string => `/u${userId}/Documents`; + +const ancestors = (): Array<{ uid: string; path: string }> => [ + { uid: anchorUid(), path: anchorPath() }, +]; + +const durableRow = ( + over: Partial = {}, +): DurableSubscription => ({ + durable: true, + subId: `app-${seq}#${over.subId ?? 'sub'}`, + holderUserId: userId, + ownerUserId: userId, + subject: `fs:${anchorPath()}`, + token: fsAnchorToken(anchorUid()), + anchorUid: anchorUid(), + anchorPath: anchorPath(), + match: null, + op: null, + appUid: null, + permission: 'list', + delivery: 'single', + targets: ['socket', 'worker'] as SubscriptionTarget[], + handlerName: 'onWrite', + context: null, + expiresAt: null, + suspendedAt: null, + suspendedReason: null, + createdAt: Math.floor(Date.now() / 1000), + ...over, +}); + +/** Put a row where dispatch reads it, and where a later ack looks it up. */ +const register = async ( + over: Partial = {}, +): Promise => { + const row = durableRow(over); + rows.set(row.subId, row); + await subscriptions.cacheDurable([row]); + service.invalidateUser(userId); + return row; +}; + +const dispatch = (node = entry()): Promise => + service.dispatchFs('fs.write.file', node, { + actingUserId: userId, + ancestors: async () => ancestors(), + }); + +/** Wait out the coalescing window a `broadcast` delivery sits in. */ +const flushed = (count = 1): Promise => + vi.waitFor(() => expect(sent.length).toBeGreaterThanOrEqual(count), { + timeout: EVENTS_COALESCE_WINDOW_MS * 12, + interval: 25, + }); + +const jump = (ms: number): void => { + vi.useFakeTimers({ toFake: ['Date'] }); + vi.setSystemTime(Date.now() + ms); +}; + +const keysOf = async (subId: string): Promise => + (await redis.keys('*')).filter((key: string) => key.includes(subId)); + +beforeEach(async () => { + seq++; + userId = 7000 + seq; + socketConnected = true; + workerOutcome = 'deferred'; + sent = []; + delivered = []; + invoked = []; + rows = new Map(); + entries = new Map(); + alarms = vi.fn(); + + redis = new MockRedis.Cluster(['redis://localhost:7001']); + await redis.del('ev:qx', 'ev:qc'); + + subscriptions = new EventSubscriptionStore( + {} as IConfig, + { redis } as never, + {} as never, + ); + pending = new PendingDeliveryStore( + {} as IConfig, + { redis } as never, + {} as never, + ); + + entries.set(`uid:${anchorUid()}`, entry({ uid: anchorUid(), isDir: true })); + + service = new EventsService( + { events: { enabled: true } } as IConfig, + { + redis, + event: { on: vi.fn(), emit: vi.fn() }, + alarm: { create: alarms }, + } as never, + { + eventSubscription: subscriptions, + pendingDelivery: pending, + durableSubscription: { + warmRegion: async () => false, + getBySubId: async (subId: string) => rows.get(subId) ?? null, + remove: async (row: DurableSubscription) => { + rows.delete(row.subId); + return { userId: row.holderUserId, generation: 1 }; + }, + }, + fsEntry: { + getEntryByUuid: async (uid: string) => + entries.get(`uid:${uid}`) ?? null, + getEntryByPath: async () => null, + getEntryById: async () => null, + }, + user: { + getById: async (id: number) => ({ id, uuid: `user-${id}` }), + }, + app: { getByUid: async (uid: string) => ({ uid, id: 1 }) }, + } as never, + { + socket: { + send: vi.fn(async (_spec, _key, data) => { + sent.push(data as DeliveryEnvelope); + }), + has: () => socketConnected, + }, + fs: { getAncestorChain: async () => ancestors() }, + acl: { + check: async () => true, + getSafeAclError: async () => ({ + status: 404, + message: 'Subject does not exist', + fields: { code: 'subject_does_not_exist' }, + }), + }, + } as never, + ); + service.onDelivered = (envelope) => delivered.push(envelope); + service.worker = { + invoke: async (invocation: WorkerInvocation) => { + invoked.push(invocation); + return workerOutcome; + }, + }; +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('a delivery owed to exactly one consumer', () => { + it('goes to the connected socket, and not to the handler as well', async () => { + const row = await register(); + + await dispatch(); + + expect(invoked).toEqual([]); + expect(sent).toHaveLength(1); + expect(sent[0]).toMatchObject({ + subId: row.subId, + ackRequired: true, + event: { op: 'write' }, + }); + expect(sent[0].ackId).toEqual(expect.any(String)); + expect(delivered).toHaveLength(1); + }); + + it('stays owed until the client acks it', async () => { + const row = await register(); + await dispatch(); + + await expect(pending.depth(row.subId)).resolves.toBe(1); + + await service.ackDelivery(actorFor(), { + subId: row.subId, + id: sent[0].ackId, + }); + + await expect(pending.depth(row.subId)).resolves.toBe(0); + // Nothing owed, nothing held. + await expect(keysOf(row.subId)).resolves.toEqual([]); + }); + + it('takes a second ack for the same delivery as nothing to do', async () => { + const row = await register(); + await dispatch(); + const ack = { subId: row.subId, id: sent[0].ackId }; + + await service.ackDelivery(actorFor(), ack); + await expect( + service.ackDelivery(actorFor(), ack), + ).resolves.toBeUndefined(); + }); + + it('refuses to settle a delivery for a subscription someone else holds', async () => { + const row = await register(); + await dispatch(); + const ack = { subId: row.subId, id: sent[0].ackId }; + + await expect( + service.ackDelivery(actorFor(userId + 1), ack), + ).rejects.toMatchObject({ + statusCode: 404, + legacyCode: 'subscription_does_not_exist', + }); + + // Untouched: the real holder can still take it. + await expect(pending.depth(row.subId)).resolves.toBe(1); + await service.ackDelivery(actorFor(), ack); + await expect(pending.depth(row.subId)).resolves.toBe(0); + }); + + it('hands the next one over as soon as the last is settled', async () => { + const row = await register(); + await dispatch(entry({ uid: `file-a-${seq}` })); + await dispatch(entry({ uid: `file-b-${seq}` })); + + // One at a time: the second waits on the first being taken. + expect(sent).toHaveLength(1); + + await service.ackDelivery(actorFor(), { + subId: row.subId, + id: sent[0].ackId, + }); + + expect(sent).toHaveLength(2); + expect(sent[1].event.id).not.toBe(sent[0].event.id); + }); + + it('tries the socket twice, then hands it to the handler', async () => { + const row = await register(); + await dispatch(); + expect(sent).toHaveLength(1); + + // First lease lapses with no ack: a second socket attempt, which is + // what a reconnected client is worth. + jump(31_000); + await service.sweepPending(); + expect(sent).toHaveLength(2); + expect(invoked).toEqual([]); + + // Second lapses too. Sockets are spent, so the handler takes it. + jump(31_000); + await service.sweepPending(); + expect(sent).toHaveLength(2); + expect(invoked).toHaveLength(1); + expect(invoked[0]).toMatchObject({ + subId: row.subId, + handlerName: 'onWrite', + holderUserId: userId, + }); + }); + + it('goes straight to the handler when nothing is connected', async () => { + socketConnected = false; + const row = await register(); + + await dispatch(); + + expect(sent).toEqual([]); + expect(invoked).toHaveLength(1); + // Not taken yet, so still owed. + await expect(pending.depth(row.subId)).resolves.toBe(1); + }); + + it('settles the delivery once the handler reports it took it', async () => { + socketConnected = false; + workerOutcome = 'settled'; + const row = await register(); + + await dispatch(); + + expect(invoked).toHaveLength(1); + await expect(pending.depth(row.subId)).resolves.toBe(0); + await expect(keysOf(row.subId)).resolves.toEqual([]); + }); + + it('is never delivered as a broadcast copy as well', async () => { + await register(); + + await dispatch(); + await new Promise((resolve) => + setTimeout(resolve, EVENTS_COALESCE_WINDOW_MS * 3), + ); + + expect(sent).toHaveLength(1); + expect(sent[0].ackRequired).toBe(true); + }); +}); + +describe('a delivery everyone connected gets', () => { + it('runs the handler once, alongside the socket copy', async () => { + const row = await register({ delivery: 'broadcast' }); + + await dispatch(); + await flushed(); + + expect(sent).toHaveLength(1); + expect(sent[0]).toMatchObject({ subId: row.subId }); + expect(sent[0].ackRequired).toBeUndefined(); + expect(invoked).toHaveLength(1); + expect(invoked[0]).toMatchObject({ subId: row.subId }); + // One delivery is one line, however many transports carried it. + expect(delivered).toHaveLength(1); + }); + + it('runs the handler once per delivery, not once per transport', async () => { + await register({ delivery: 'broadcast', targets: ['worker'] }); + + await dispatch(); + await vi.waitFor(() => expect(invoked.length).toBe(1), { + timeout: EVENTS_COALESCE_WINDOW_MS * 12, + interval: 25, + }); + + expect(sent).toEqual([]); + expect(delivered).toHaveLength(1); + }); + + it('delivers and meters nothing when nothing can carry it', async () => { + await register({ delivery: 'broadcast', targets: ['push'] }); + + await dispatch(); + await new Promise((resolve) => + setTimeout(resolve, EVENTS_COALESCE_WINDOW_MS * 3), + ); + + expect(sent).toEqual([]); + expect(invoked).toEqual([]); + expect(delivered).toEqual([]); + }); +}); + +/** No alarm text may name the storage this runs on — see AGENTS.md. */ +const noInfraWording = (): RegExp => + /^(?!.*\b(redis|elasticache|dynamo|dynamodb|aws|memcached)\b).*$/is; + +describe('when a delivery cannot be held', () => { + it('does not fail the write, and does not let the loss pass quietly', async () => { + await register(); + vi.spyOn(pending, 'enqueue').mockRejectedValue( + new Error('the cache is unreachable'), + ); + + await expect(dispatch()).resolves.toBeUndefined(); + + expect(sent).toEqual([]); + expect(alarms).toHaveBeenCalledWith( + 'events_pending_enqueue_failed', + expect.stringMatching(noInfraWording()), + expect.objectContaining({ subId: expect.any(String) }), + 'warning', + { dedup: true }, + ); + }); + + it('sheds the oldest and alarms when the region is holding too much', async () => { + const row = await register(); + await redis.incrby('ev:qc', EVENTS_REGION_PENDING_CEILING); + + await dispatch(); + + expect(alarms).toHaveBeenCalledWith( + 'events_pending_ceiling', + expect.stringMatching(noInfraWording()), + expect.objectContaining({ subId: row.subId }), + 'warning', + { dedup: true }, + ); + // What it shed, it was told about: the marker took the delivery's place. + expect(sent).toHaveLength(1); + expect(sent[0].event).toMatchObject({ + op: 'gap', + reason: 'backlog_overflow', + }); + }); +}); + +describe('giving up a subscription', () => { + it('drops what it was owed, and corrects the region counter', async () => { + socketConnected = false; + const row = await register(); + await dispatch(); + await expect(pending.regionDepth()).resolves.toBe(1); + + await service.unsubscribeDurable(actorFor(), { subId: row.subId }); + + await expect(pending.depth(row.subId)).resolves.toBe(0); + await expect(pending.regionDepth()).resolves.toBe(0); + await expect(keysOf(row.subId)).resolves.toEqual([]); + }); +}); + +describe('keeping the region counter honest', () => { + it('self-heals a drifted counter on the periodic sweep', async () => { + socketConnected = false; + await register(); + await dispatch(); + // Drift: as if a settle's decrement had been lost somewhere else. + await redis.incrby('ev:qc', 41); + await expect(pending.regionDepth()).resolves.toBe(42); + + await service.sweepPending(); + + await expect(pending.regionDepth()).resolves.toBe(1); + }); +}); diff --git a/src/backend/services/events/workerSeam.ts b/src/backend/services/events/workerSeam.ts new file mode 100644 index 000000000..f0759f4f8 --- /dev/null +++ b/src/backend/services/events/workerSeam.ts @@ -0,0 +1,76 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import type { 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. + * + * 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 + * ran from looking like one that succeeded. + */ + +/** One handler call, as the seam receives it. */ +export interface WorkerInvocation { + subId: string; + /** Whose subscription this is, and who is billed for the work. */ + holderUserId: number; + /** The app whose handler runs, or null for an account-owned row. */ + appUid: string | null; + handlerName: string | null; + event: DeliverableEvent; + /** The subscription's stored context, delivered to the handler as `ctx`. */ + context: string | null; +} + +/** + * 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. + */ +export type WorkerInvocationOutcome = 'settled' | 'deferred'; + +export interface WorkerInvokerSeam { + invoke(invocation: WorkerInvocation): Promise; +} + +/** Invocations one recorder holds, so it cannot grow with event volume. */ +const RECORDED_INVOCATIONS = 100; + +/** + * The seam until there is something to run handlers. Records what would have + * been invoked and settles nothing, so a delivery handed here stays owed and is + * retried rather than quietly disappearing. + */ +export class RecordingWorkerInvoker implements WorkerInvokerSeam { + readonly recorded: WorkerInvocation[] = []; + + invoke(invocation: WorkerInvocation): Promise { + this.recorded.push(invocation); + if (this.recorded.length > RECORDED_INVOCATIONS) this.recorded.shift(); + return Promise.resolve('deferred'); + } +} diff --git a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts index 5eb23d79c..766fd2b4d 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts +++ b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts @@ -173,6 +173,38 @@ describe('validation at the row write', () => { ).rejects.toSatisfy(codeOf('invalid_targets')); }); + it('refuses a `single` row that wants a device notification', async () => { + await expect( + durable().create( + input({ + delivery: 'single', + handlerName: 'onWrite', + targets: ['socket', 'push'], + }), + ), + ).rejects.toSatisfy(codeOf('invalid_targets')); + await expect(durable().countForHolder(userId)).resolves.toBe(0); + }); + + it('refuses a `single` row with no handler to fall back to', async () => { + await expect( + durable().create( + input({ + delivery: 'single', + handlerName: 'onWrite', + targets: ['socket'], + }), + ), + ).rejects.toSatisfy(codeOf('invalid_targets')); + }); + + it('keeps the same targets on a `broadcast` row, where push is fine', async () => { + const { row } = await durable().create( + input({ targets: ['socket', 'push'] }), + ); + expect(row.targets).toEqual(['socket', 'push']); + }); + it('refuses a context past the hard cap', async () => { await expect( durable().create(input({ context: 'x'.repeat(4097) })), diff --git a/src/backend/stores/events/DurableSubscriptionStore.ts b/src/backend/stores/events/DurableSubscriptionStore.ts index c6ef533f7..f3e82f984 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.ts +++ b/src/backend/stores/events/DurableSubscriptionStore.ts @@ -33,6 +33,7 @@ import type { GenerationBump } from './EventSubscriptionStore.js'; import { isSubscriptionTarget, SUBSCRIPTION_TARGETS, + targetsAllowedForDelivery, type DurableSubscription, type SubscriptionTarget, } from './types.js'; @@ -111,6 +112,13 @@ const invalidTargets = (): HttpError => { legacyCode: 'invalid_targets' }, ); +const pushOnSingle = (): HttpError => + new HttpError( + 400, + 'A `single` subscription needs a `worker` target and may not target `push`', + { legacyCode: 'invalid_targets' }, + ); + const quotaReached = (): HttpError => new HttpError( 429, @@ -202,7 +210,7 @@ export class DurableSubscriptionStore extends PuterStore { async create( input: DurableSubscriptionInput, ): Promise<{ row: DurableSubscription; bump: GenerationBump }> { - const targets = this.#assertTargets(input.targets); + const targets = this.#assertTargets(input.delivery, input.targets); this.#assertContext(input.context); const held = await this.countForHolder(input.holderUserId); @@ -437,11 +445,22 @@ export class DurableSubscriptionStore extends PuterStore { return rows.map(toRow); } - #assertTargets(targets: readonly string[]): SubscriptionTarget[] { + /** + * The row cannot exist with transports its delivery class cannot use. Held + * here rather than only at the API, so a writer that never passes through + * one cannot leave an unsatisfiable row behind. + */ + #assertTargets( + delivery: DeliveryClass, + targets: readonly string[], + ): SubscriptionTarget[] { if (!Array.isArray(targets) || targets.length === 0) throw invalidTargets(); if (!targets.every(isSubscriptionTarget)) throw invalidTargets(); - return [...new Set(targets as SubscriptionTarget[])]; + + const unique = [...new Set(targets as SubscriptionTarget[])]; + if (!targetsAllowedForDelivery(delivery, unique)) throw pushOnSingle(); + return unique; } #assertContext(context: string | null): void { diff --git a/src/backend/stores/events/PendingDeliveryStore.test.ts b/src/backend/stores/events/PendingDeliveryStore.test.ts new file mode 100644 index 000000000..26e526df6 --- /dev/null +++ b/src/backend/stores/events/PendingDeliveryStore.test.ts @@ -0,0 +1,427 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import MockRedis from 'ioredis-mock'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { + EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION, + EVENTS_REGION_PENDING_CEILING, +} from '../../controllers/events/limits.js'; +import type { ProjectedEvent } from '../../services/events/registry.js'; +import type { IConfig } from '../../types.js'; +import { PendingDeliveryStore } from './PendingDeliveryStore.js'; + +/** + * What a subscription is owed, and what holding it costs. Two claims run + * through everything here: a delivery is never lost quietly — a drop is a gap + * marker the subscriber receives — and a subscription that is keeping up owns + * nothing at all. + */ + +// The keyspace is shared per process, so each test gets its own subscription +// and the two region-wide keys are cleared between them. +let seq = 0; +let subId = ''; +let redis: InstanceType; +let store: PendingDeliveryStore; +let commands: string[]; + +const INDEX_KEY = 'ev:qx'; +const COUNTER_KEY = 'ev:qc'; + +const entriesKey = (id = subId): string => `ev:q:{${id}}`; +const pendingKey = (id = subId): string => `ev:qp:{${id}}`; + +/** Every command that crossed the client, so a test can say what was not. */ +const recordingRedis = ( + inner: InstanceType, +): InstanceType => + new Proxy(inner, { + get(target, property, receiver) { + const value = Reflect.get(target, property, receiver); + if (typeof property !== 'string' || typeof value !== 'function') + return value; + return (...args: unknown[]) => { + commands.push(property); + return (value as (...a: unknown[]) => unknown).apply( + target, + args, + ); + }; + }, + }) as InstanceType; + +const event = (id: string): ProjectedEvent => ({ + id, + subject: 'fs:/u/Documents', + op: 'write', + uid: `uid-${id}`, + path: `/u/Documents/${id}.txt`, + self: true, + ts: Date.now(), + seq: 0, +}); + +/** Keys this subscription owns right now, whatever their type. */ +const keysOf = async (id = subId): Promise => + (await redis.keys('*')).filter((key: string) => key.includes(id)); + +/** + * Put a backlog in place without paying for it one delivery at a time. The caps + * are the point of these tests, not the path that fills them. + */ +const seedBacklog = async (id: string, count: number): Promise => { + const scored: Array = []; + const fields: string[] = []; + const base = Date.now() - count; + for (let i = 0; i < count; i++) { + const entryId = `${base + i}-seed${i}`; + scored.push(base + i, entryId); + fields.push( + entryId, + JSON.stringify({ event: event(`seed-${i}`), socketAttempts: 0 }), + ); + } + await redis.zadd(pendingKey(id), ...scored); + await redis.hset(entriesKey(id), ...fields); + await redis.zadd(INDEX_KEY, base, id); + await redis.incrby(COUNTER_KEY, count); +}; + +const pendingEvents = async (id = subId): Promise => { + const held = await redis.hvals(entriesKey(id)); + return held.map( + (raw: string) => (JSON.parse(raw) as { event: ProjectedEvent }).event, + ); +}; + +beforeEach(async () => { + seq++; + subId = `app-x#sub-${seq}`; + commands = []; + redis = recordingRedis(new MockRedis.Cluster(['redis://localhost:7001'])); + await redis.del(INDEX_KEY, COUNTER_KEY); + store = new PendingDeliveryStore( + {} as IConfig, + { redis } as never, + {} as never, + ); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('what a subscription owns', () => { + it('owns nothing until it is owed something', async () => { + await expect(keysOf()).resolves.toEqual([]); + await expect(store.depth(subId)).resolves.toBe(0); + + await store.enqueue(subId, event('a')); + + expect(await keysOf()).not.toEqual([]); + await expect(store.depth(subId)).resolves.toBe(1); + }); + + it('owns nothing again once the last delivery settles', async () => { + const { entryId } = await store.enqueue(subId, event('a')); + + await expect(store.settle(subId, entryId)).resolves.toBe(true); + + await expect(keysOf()).resolves.toEqual([]); + await expect(store.head(10)).resolves.toEqual([]); + await expect(store.regionDepth()).resolves.toBe(0); + }); +}); + +describe('handing a delivery out', () => { + it('hands out the oldest first, and only one at a time', async () => { + const first = await store.enqueue(subId, event('a')); + await store.enqueue(subId, event('b')); + + const claimed = await store.claim(subId); + expect(claimed?.entryId).toBe(first.entryId); + expect(claimed?.event.id).toBe('a'); + + // The second is owed, but nothing else may take it while the first is + // out — `single` promises one consumer per event. + await expect(store.claim(subId)).resolves.toBeNull(); + + await store.settle(subId, first.entryId); + await expect(store.claim(subId)).resolves.toMatchObject({ + event: { id: 'b' }, + }); + }); + + it('counts the socket attempts a delivery has spent', async () => { + const { entryId } = await store.enqueue(subId, event('a')); + await store.claim(subId, { leaseMs: 1 }); + + await expect(store.recordSocketAttempt(subId, entryId)).resolves.toBe( + 1, + ); + + vi.useFakeTimers({ toFake: ['Date'] }); + vi.setSystemTime(Date.now() + 1_000); + await expect(store.claim(subId)).resolves.toMatchObject({ + entryId, + socketAttempts: 1, + }); + }); + + it('offers a delivery again once its lease lapses', async () => { + const { entryId } = await store.enqueue(subId, event('a')); + expect(await store.claim(subId, { leaseMs: 30_000 })).not.toBeNull(); + await expect(store.claim(subId)).resolves.toBeNull(); + + vi.useFakeTimers({ toFake: ['Date'] }); + vi.setSystemTime(Date.now() + 31_000); + + await expect(store.claim(subId)).resolves.toMatchObject({ entryId }); + }); + + it('settles once, and treats a second ack as nothing to do', async () => { + const { entryId } = await store.enqueue(subId, event('a')); + await store.claim(subId); + + await expect(store.settle(subId, entryId)).resolves.toBe(true); + await expect(store.settle(subId, entryId)).resolves.toBe(false); + }); +}); + +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); + + const { shed } = await store.enqueue(subId, event('newest')); + + expect(shed).toEqual([{ subId, dropped: 2, scope: 'subscription' }]); + await expect(store.depth(subId)).resolves.toBe( + EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION, + ); + + const held = await pendingEvents(); + const markers = held.filter((held) => held.op === 'gap'); + expect(markers).toHaveLength(1); + expect(markers[0]).toMatchObject({ + op: 'gap', + reason: 'backlog_overflow', + subject: 'fs:/u/Documents', + }); + // The oldest went, the newest stayed. + expect(held.some((event) => event.id === 'seed-0')).toBe(false); + expect(held.some((event) => event.id === 'newest')).toBe(true); + }); + + it('sheds the region`s oldest backlog first, and says which', async () => { + const older = `app-x#older-${seq}`; + await seedBacklog(older, 4); + // Two over the ceiling once this enqueue lands. Getting back under it + // takes three, because the marker left behind holds a place too — and + // the shed stops inside the oldest backlog, never reaching the newest. + await redis.incrby(COUNTER_KEY, EVENTS_REGION_PENDING_CEILING - 3); + + const { shed } = await store.enqueue(subId, event('a')); + + expect(shed).toEqual([{ subId: older, dropped: 3, scope: 'region' }]); + // What it lost, it was told about. + const marked = await pendingEvents(older); + expect(marked.filter((event) => event.op === 'gap')).toHaveLength(1); + // What it did not lose is still owed; the newest end survives a shed. + expect(marked.some((event) => event.id === 'seed-3')).toBe(true); + expect(marked.some((event) => event.id === 'seed-2')).toBe(false); + await expect(store.depth(subId)).resolves.toBe(1); + // The region actually got back under, marker included. + await expect(store.regionDepth()).resolves.toBe( + EVENTS_REGION_PENDING_CEILING, + ); + }); +}); + +describe('sharing the sweeper between backlogs', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + it('hands the head to exactly one of two concurrent claimers', async () => { + await store.enqueue(subId, event('a')); + + const claims = await Promise.all([ + store.claim(subId), + store.claim(subId), + ]); + + expect(claims.filter(Boolean)).toHaveLength(1); + }); + + it('moves a claimed subscription behind the ones still waiting', async () => { + const other = `app-x#other-${seq}`; + vi.setSystemTime(1_000_000); + await store.enqueue(subId, event('first')); + vi.setSystemTime(1_001_000); + await store.enqueue(other, event('second')); + expect((await store.head(2)).map((h) => h.subId)).toEqual([ + subId, + other, + ]); + + vi.setSystemTime(1_002_000); + await store.claim(subId); + + expect((await store.head(2)).map((h) => h.subId)).toEqual([ + other, + subId, + ]); + }); + + it('defers a subscription behind the others, holdings untouched', async () => { + const other = `app-x#other-${seq}`; + vi.setSystemTime(1_000_000); + await store.enqueue(subId, event('first')); + vi.setSystemTime(1_001_000); + await store.enqueue(other, event('second')); + + vi.setSystemTime(1_002_000); + await store.defer(subId); + + expect((await store.head(2)).map((h) => h.subId)).toEqual([ + other, + subId, + ]); + await expect(store.depth(subId)).resolves.toBe(1); + }); + + it('puts a lifetime on a backlog`s keys', async () => { + await store.enqueue(subId, event('a')); + + await expect(redis.ttl(pendingKey())).resolves.toBeGreaterThan(0); + await expect(redis.ttl(entriesKey())).resolves.toBeGreaterThan(0); + }); +}); + +describe('finding the work', () => { + it('reads the oldest end of the index rather than the keyspace', async () => { + const older = `app-x#older-${seq}`; + await seedBacklog(older, 2); + await store.enqueue(subId, event('a')); + + commands = []; + const head = await store.head(10); + + expect(head.map((entry) => entry.subId)).toEqual([older, subId]); + expect(head[0].oldestAt).toBeLessThan(head[1].oldestAt); + expect( + commands.filter((command) => + ['scan', 'keys', 'hscan', 'sscan', 'zscan'].includes(command), + ), + ).toEqual([]); + }); + + it('never scans the keyspace, whatever it is asked to do', async () => { + const { entryId } = await store.enqueue(subId, event('a')); + await store.claim(subId); + await store.recordSocketAttempt(subId, entryId); + await store.settle(subId, entryId); + await store.enqueue(subId, event('b')); + await store.purge(subId); + await store.regionDepth(); + + expect( + commands.filter((command) => + ['scan', 'keys', 'hscan', 'sscan', 'zscan'].includes(command), + ), + ).toEqual([]); + }); +}); + +describe('surviving a crash between the write and the reindex', () => { + it('keeps a first entry discoverable even if the follow-up reindex never runs', async () => { + // The counter write is the first command after the entry's own + // transaction lands — exactly where a crash would fall between the + // entry existing and the follow-up reindex. + vi.spyOn(redis, 'incrby').mockRejectedValueOnce( + new Error('connection lost'), + ); + + await expect(store.enqueue(subId, event('a'))).rejects.toThrow(); + vi.restoreAllMocks(); + + // The crash was after the index write, not before it: the sweeper + // still finds this subscription, and the entry is still claimable. + await expect(store.head(10)).resolves.toEqual([ + { subId, oldestAt: expect.any(Number) }, + ]); + await expect(store.claim(subId)).resolves.toMatchObject({ + event: { id: 'a' }, + }); + }); +}); + +describe('keeping the region counter honest', () => { + it('corrects an inflated counter back to what is actually held', async () => { + await store.enqueue(subId, event('a')); + // Drift: some decrement that should have landed did not. + await redis.incrby(COUNTER_KEY, 500); + + await expect(store.reconcileRegionDepth()).resolves.toBe(1); + await expect(store.regionDepth()).resolves.toBe(1); + }); + + it('corrects an undercounted or negative counter the same way', async () => { + await store.enqueue(subId, event('a')); + await redis.set(COUNTER_KEY, -50); + + await expect(store.reconcileRegionDepth()).resolves.toBe(1); + await expect(store.regionDepth()).resolves.toBe(1); + }); + + it('zeroes the counter when nothing is actually pending', async () => { + await redis.incrby(COUNTER_KEY, 12); + + await expect(store.reconcileRegionDepth()).resolves.toBe(0); + await expect(store.regionDepth()).resolves.toBe(0); + }); + + it('never scans the keyspace to do it', async () => { + await store.enqueue(subId, event('a')); + commands = []; + + await store.reconcileRegionDepth(); + + expect( + commands.filter((command) => + ['scan', 'keys', 'hscan', 'sscan', 'zscan'].includes(command), + ), + ).toEqual([]); + }); +}); + +describe('giving up a backlog', () => { + it('drops everything a subscription held and gives the room back', async () => { + await store.enqueue(subId, event('a')); + await store.enqueue(subId, event('b')); + await expect(store.regionDepth()).resolves.toBe(2); + + await expect(store.purge(subId)).resolves.toBe(2); + + await expect(keysOf()).resolves.toEqual([]); + await expect(store.regionDepth()).resolves.toBe(0); + await expect(store.head(10)).resolves.toEqual([]); + }); +}); diff --git a/src/backend/stores/events/PendingDeliveryStore.ts b/src/backend/stores/events/PendingDeliveryStore.ts new file mode 100644 index 000000000..beae7eae5 --- /dev/null +++ b/src/backend/stores/events/PendingDeliveryStore.ts @@ -0,0 +1,590 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +import { randomUUID } from 'node:crypto'; +import { + EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION, + EVENTS_REGION_PENDING_CEILING, +} from '../../controllers/events/limits.js'; +import type { + DeliverableEvent, + GapMarker, +} from '../../services/events/registry.js'; +import { PuterStore } from '../types.js'; + +/** + * Deliveries that are waiting for a consumer, in the region that emitted them. + * + * A `broadcast` delivery is gone the moment it is sent. A `single` one is owed + * to exactly one consumer, so it has to survive until that consumer says it + * took it — which is what everything here holds. Nothing is replicated: the + * lease, the retry count and the queue belong to the region that emitted the + * event, and the published promise is at-least-once **while that region is + * available**. + * + * Three keys per subscription, all created on the first pending delivery and + * all deleted the moment the last one settles — a subscription that is keeping + * up owns nothing: + * + * ev:q:{} HASH entryId -> the delivery and its attempt count + * ev:qp:{} ZSET entryId -> enqueued at; membership means unsettled + * ev:ql:{} ZSET entryId -> lease expiry; membership means in flight + * + * And two the region shares: + * + * ev:qx ZSET subId -> oldest pending delivery, for the sweeper + * ev:qc STR how many deliveries the region is holding + * + * `ev:qx` is why nothing here ever scans the keyspace: the sweeper reads its + * head to find the subscriptions that are behind, and a scan is exactly what + * goes pathological when the system already is. The counter is what the region + * ceiling is read from, and it moves only where the pending set does. + * + * The pending set, the lease and the attempt count are explicit rather than a + * queue primitive's implicit ones: reclaiming an expired lease is then a score + * comparison, and dropping a whole subscription is one delete. + * + * One delivery is in flight per subscription at a time. `single` promises one + * consumer per event, and handing out the next while the last is unsettled + * would make ordering — and the retry count that decides socket-versus-worker — + * meaningless. + */ + +// -- Keys ------------------------------------------------------------- + +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 INDEX_KEY = 'ev:qx'; +const COUNTER_KEY = 'ev:qc'; + +// -- Lease ------------------------------------------------------------ + +/** + * How long a consumer has to settle a delivery before someone else may take it. + * Long enough for a handler doing real work, short enough that a tab that + * closed mid-delivery does not hold the queue. + */ +export const PENDING_LEASE_TTL_MS = 30_000; + +/** Subscriptions one region-ceiling shed may take deliveries from. */ +const REGION_SHED_SUBSCRIPTIONS = 32; + +/** Deliveries one region-ceiling shed may drop, so a burst cannot stall here. */ +const REGION_SHED_MAX_ENTRIES = 1_000; + +/** Arguments one variadic command carries, so a large shed stays in batches. */ +const COMMAND_BATCH = 500; + +/** Queues the reconciler measures at once, one command each. */ +const RECONCILE_CONCURRENCY = 50; + +/** + * Backstop for keys nothing indexes any more: a claim refreshes it, so a + * backlog still being retried never lapses, while one left behind by a purge + * that raced an enqueue does not sit in Redis forever. + */ +export const PENDING_BACKLOG_TTL_SECONDS = 7 * 24 * 60 * 60; + +// -- Scripts ---------------------------------------------------------- +// The three keys a subscription owns share its `{subId}` hash tag, so a script +// over them runs on one node under cluster mode. The index and the counter hash +// elsewhere and are moved by plain commands around the scripts. + +/** + * Lease the oldest delivery, if none is in flight. KEYS: entries, pending, + * lease. ARGV: now, leaseUntil, ttlSeconds. Returns a status-first tuple. + */ +const CLAIM_SCRIPT = ` +local inflight = redis.call('ZRANGEBYSCORE', KEYS[3], ARGV[1], '+inf', 'LIMIT', 0, 1) +if #inflight > 0 then return { 'inflight' } end +local head = redis.call('ZRANGE', KEYS[2], 0, 0) +if #head == 0 then return { 'empty' } end +local raw = redis.call('HGET', KEYS[1], head[1]) +if not raw then + redis.call('ZREM', KEYS[2], head[1]) + redis.call('ZREM', KEYS[3], head[1]) + return { 'missing', head[1] } +end +redis.call('ZADD', KEYS[3], ARGV[2], head[1]) +redis.call('EXPIRE', KEYS[1], ARGV[3]) +redis.call('EXPIRE', KEYS[2], ARGV[3]) +redis.call('EXPIRE', KEYS[3], ARGV[3]) +return { 'claimed', head[1], raw } +`; + +/** + * The oldest pending score, or '-1' after deleting a drained subscription's + * keys. Checking and deleting in one step is what keeps a concurrent append + * from being wiped between the two. KEYS: entries, pending, lease. + */ +const REINDEX_SCRIPT = ` +local head = redis.call('ZRANGE', KEYS[2], 0, 0, 'WITHSCORES') +if #head == 0 then + redis.call('DEL', KEYS[1], KEYS[2], KEYS[3]) + return '-1' +end +return head[2] +`; + +interface PendingScripts { + pendingClaim( + entries: string, + pending: string, + lease: string, + now: string, + leaseUntil: string, + ttlSeconds: string, + ): Promise<[string, string?, string?]>; + pendingReindex( + entries: string, + pending: string, + lease: string, + ): Promise; +} + +// -- Shapes ----------------------------------------------------------- + +/** One delivery, handed out under a lease. */ +export interface ClaimedDelivery { + entryId: string; + event: DeliverableEvent; + /** Socket attempts already spent, which is what decides the next one. */ + socketAttempts: number; +} + +/** What one shed took, for the marker and the alarm that follow it. */ +export interface PendingShed { + subId: string; + dropped: number; + scope: 'subscription' | 'region'; +} + +/** A subscription with undelivered deliveries, oldest first. */ +export interface PendingHead { + subId: string; + oldestAt: number; +} + +interface StoredEntry { + event: DeliverableEvent; + socketAttempts: number; +} + +const parseEntry = (raw: string | null): StoredEntry | null => { + if (!raw) return null; + try { + const parsed = JSON.parse(raw) as StoredEntry; + return parsed?.event ? parsed : null; + } catch { + return null; + } +}; + +const gapMarker = (subject: string): GapMarker => ({ + id: randomUUID(), + subject, + op: 'gap', + reason: 'backlog_overflow', + ts: Date.now(), +}); + +/** `ZPOPMIN`/`ZRANGE … WITHSCORES` answer flat, and scores come back typed. */ +const membersOf = (flat: readonly unknown[]): string[] => { + const members: string[] = []; + for (let i = 0; i < flat.length; i += 2) members.push(String(flat[i])); + return members; +}; + +const batched = (items: readonly T[], size = COMMAND_BATCH): T[][] => { + const batches: T[][] = []; + for (let i = 0; i < items.length; i += size) + batches.push(items.slice(i, i + size)); + return batches; +}; + +export class PendingDeliveryStore extends PuterStore { + /** Tells this process's ids from a peer's, so two cannot mint the same. */ + readonly #minter = randomUUID().slice(0, 8); + #minted = 0; + #definedScripts = false; + + #scripts(): PendingScripts { + if (!this.#definedScripts) { + this.#definedScripts = true; + this.clients.redis.defineCommand('pendingClaim', { + numberOfKeys: 3, + lua: CLAIM_SCRIPT, + }); + this.clients.redis.defineCommand('pendingReindex', { + numberOfKeys: 3, + lua: REINDEX_SCRIPT, + }); + } + return this.clients.redis as unknown as PendingScripts; + } + + // -- Writes ------------------------------------------------------ + + /** + * Hold one delivery until a consumer takes it. Returns what had to be shed + * to make room — the subscription's own cap first, then the region's — so + * the caller can say so out loud. + */ + async enqueue( + subId: string, + event: DeliverableEvent, + ): Promise<{ entryId: string; shed: PendingShed[] }> { + const entryId = await this.#append(subId, event); + + const shed: PendingShed[] = []; + const overflowed = await this.#capSubscription(subId); + if (overflowed) shed.push(overflowed); + shed.push(...(await this.#capRegion())); + + return { entryId, shed }; + } + + /** + * Take the oldest delivery this subscription is owed, under a lease. Null + * when one is already in flight or nothing is waiting; an expired lease + * makes its delivery claimable again, which is the whole retry mechanism. + */ + async claim( + subId: string, + options: { leaseMs?: number } = {}, + ): Promise { + const now = Date.now(); + // One script, so two claimers racing for the same head cannot both + // walk away holding it. + const [status, entryId, raw] = await this.#scripts().pendingClaim( + entriesKey(subId), + pendingKey(subId), + leaseKey(subId), + String(now), + String(now + (options.leaseMs ?? PENDING_LEASE_TTL_MS)), + String(PENDING_BACKLOG_TTL_SECONDS), + ); + if (status === 'inflight') return null; + if (status === 'empty') { + // Nothing left, which is also how a drained subscription's keys go. + await this.#reindex(subId); + return null; + } + if (status === 'missing') { + // A queue position with no entry behind it: half a write. The + // script already dropped the position; this drops its share. + await this.clients.redis.decrby(COUNTER_KEY, 1); + await this.#reindex(subId); + return null; + } + + const entry = parseEntry(raw ?? null); + if (!entryId || !entry) { + await this.clients.redis.zrem(pendingKey(subId), String(entryId)); + await this.#forget(subId, [String(entryId)]); + await this.#reindex(subId); + return null; + } + + // To the back of the sweeper's line: a delivery nobody ever settles + // must not hold the head against every other backlog in the region. + await this.clients.redis.zadd(INDEX_KEY, 'XX', now, subId); + return { + entryId, + event: entry.event, + socketAttempts: entry.socketAttempts, + }; + } + + /** + * Send a subscription to the back of the sweeper's line, holdings + * untouched. + */ + async defer(subId: string): Promise { + await this.clients.redis.zadd(INDEX_KEY, 'XX', Date.now(), subId); + } + + /** Count one socket attempt against a claimed delivery. */ + async recordSocketAttempt(subId: string, entryId: string): Promise { + const entry = parseEntry( + await this.clients.redis.hget(entriesKey(subId), entryId), + ); + if (!entry) return 0; + + const socketAttempts = entry.socketAttempts + 1; + await this.clients.redis.hset( + entriesKey(subId), + entryId, + JSON.stringify({ ...entry, socketAttempts }), + ); + return socketAttempts; + } + + /** + * 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 + * makes routine and which nothing should treat as an error. + */ + async settle(subId: string, entryId: string): Promise { + const removed = await this.clients.redis.zrem( + pendingKey(subId), + entryId, + ); + if (Number(removed) !== 1) return false; + + await this.#forget(subId, [entryId]); + await this.#reindex(subId); + return true; + } + + /** Drop everything a subscription is holding, for one that is going away. */ + async purge(subId: string): Promise { + const held = await this.depth(subId); + await this.clients.redis.del( + entriesKey(subId), + pendingKey(subId), + leaseKey(subId), + ); + await this.clients.redis.zrem(INDEX_KEY, subId); + if (held > 0) await this.clients.redis.decrby(COUNTER_KEY, held); + return held; + } + + // -- Reads ------------------------------------------------------- + + /** + * The subscriptions that have waited longest, oldest first. The sweeper's + * whole input, and the backlog-age metric's. + */ + async head(limit: number): Promise { + if (limit <= 0) return []; + const flat = await this.clients.redis.zrange( + INDEX_KEY, + 0, + limit - 1, + 'WITHSCORES', + ); + const heads: PendingHead[] = []; + for (let i = 0; i < flat.length; i += 2) + heads.push({ + subId: String(flat[i]), + oldestAt: Number(flat[i + 1]) || 0, + }); + return heads; + } + + async depth(subId: string): Promise { + return Number(await this.clients.redis.zcard(pendingKey(subId))) || 0; + } + + /** What the region is holding in total, and what its ceiling is read from. */ + async regionDepth(): Promise { + const raw = await this.clients.redis.get(COUNTER_KEY); + const held = raw === null ? 0 : Number.parseInt(raw, 10); + return Number.isFinite(held) && held > 0 ? held : 0; + } + + /** + * Recompute the region counter from the pending sets it is supposed to + * total. The counter moves in a separate write from the removal it follows + * (a settle's decrement, a shed's decrement-then-append), so a crash + * between the two drifts it in either direction — silently undercounting + * lets the ceiling never trip, silently overcounting trips it forever. + * Cheap in the steady state the index is sized for: one read per + * subscription that actually has a backlog, never the keyspace. + */ + async reconcileRegionDepth(): Promise { + const subIds = await this.clients.redis.zrange(INDEX_KEY, 0, -1); + + let total = 0; + // One command per subscription: the queues hash to different slots, + // so under cluster mode they cannot share a pipeline. + for (const batch of batched(subIds, RECONCILE_CONCURRENCY)) { + const counts = await Promise.all( + batch.map((subId) => this.depth(subId)), + ); + for (const count of counts) total += count; + } + + await this.clients.redis.set(COUNTER_KEY, total); + return total; + } + + // -- Internals --------------------------------------------------- + + /** + * The id and the order of one delivery. Deliveries landing in the same + * millisecond share a score, and a sorted set breaks that tie on the member + * — so the counter is padded and comes first, and a queue keeps its order + * under a burst instead of shuffling it. + */ + #mintEntryId(at: number): string { + const minted = String(++this.#minted).padStart(12, '0'); + return `${at}-${minted}-${this.#minter}`; + } + + /** + * Add one delivery without asking whether there is room for it. + * + * The index write goes in the same pipeline as the entry it describes — + * `NX` so it never disturbs an existing (older) score — rather than waiting + * for the follow-up `#reindex` below. A subscription's first pending entry + * would otherwise be fully written and claimable, yet invisible to the + * sweeper forever, if this process died in the gap between the two: `ev:qx` + * is the only thing the sweeper ever reads, so an entry it does not know + * about is never retried. + */ + async #append(subId: string, event: DeliverableEvent): Promise { + const now = Date.now(); + const entryId = this.#mintEntryId(now); + + // Index first, so a crash after the entry lands still leaves the + // sweeper a way to find it; `NX` keeps an older score in place. + await this.clients.redis.zadd(INDEX_KEY, 'NX', now, subId); + // The entry and its queue position land together, so a concurrent + // drain can never see one without the other. Same slot, so this is a + // real transaction under cluster mode. + const write = this.clients.redis.multi(); + write.hset( + entriesKey(subId), + entryId, + JSON.stringify({ event, socketAttempts: 0 } satisfies StoredEntry), + ); + write.zadd(pendingKey(subId), now, entryId); + write.expire(entriesKey(subId), PENDING_BACKLOG_TTL_SECONDS); + write.expire(pendingKey(subId), PENDING_BACKLOG_TTL_SECONDS); + await write.exec(); + await this.clients.redis.incrby(COUNTER_KEY, 1); + + // A drain that emptied this subscription in between took it out of + // the index again; this puts it back, with the right score. + await this.#reindex(subId); + return entryId; + } + + async #capSubscription(subId: string): Promise { + const held = await this.depth(subId); + const over = held - EVENTS_PENDING_DELIVERIES_PER_SUBSCRIPTION; + if (over <= 0) return null; + // One more than the overflow, because the marker that replaces them + // takes a place of its own. + return this.#shedOldest(subId, over + 1, 'subscription'); + } + + /** + * Shed the region's oldest deliveries until it is back under the ceiling. + * Oldest-first across subscriptions, because the index is already ordered + * that way and a backlog nobody is draining is the one to lose. + */ + async #capRegion(): Promise { + const held = await this.regionDepth(); + let over = Math.min( + held - EVENTS_REGION_PENDING_CEILING, + REGION_SHED_MAX_ENTRIES, + ); + if (over <= 0) return []; + + const shed: PendingShed[] = []; + for (const { subId } of await this.head(REGION_SHED_SUBSCRIPTIONS)) { + if (over <= 0) break; + // One more than the overflow, as the marker left behind takes a + // place of its own; counting it is what lets the region actually + // get back under the ceiling rather than hover one over it. + const dropped = await this.#shedOldest(subId, over + 1, 'region'); + if (!dropped) continue; + over -= dropped.dropped - 1; + shed.push(dropped); + } + return shed; + } + + /** + * Drop a subscription's oldest deliveries and leave one gap marker in their + * place — the marker is a delivery like any other, so it queues where they + * were and reaches the subscriber the same way. + */ + async #shedOldest( + subId: string, + count: number, + scope: PendingShed['scope'], + ): Promise { + const dropped = membersOf( + await this.clients.redis.zpopmin(pendingKey(subId), count), + ); + if (dropped.length === 0) return null; + + const subject = await this.#subjectOf( + subId, + dropped[dropped.length - 1], + ); + await this.#forget(subId, dropped); + await this.#append(subId, gapMarker(subject)); + + return { subId, dropped: dropped.length, scope }; + } + + /** The subject a shed delivery carried, so its marker can name it. */ + async #subjectOf(subId: string, entryId: string): Promise { + const entry = parseEntry( + await this.clients.redis.hget(entriesKey(subId), entryId), + ); + return entry?.event.subject ?? ''; + } + + /** Forget entries already out of the pending set, and the space they held. */ + async #forget(subId: string, entryIds: readonly string[]): Promise { + if (entryIds.length === 0) return; + for (const batch of batched(entryIds)) { + const drop = this.clients.redis.pipeline(); + drop.hdel(entriesKey(subId), ...batch); + drop.zrem(leaseKey(subId), ...batch); + await drop.exec(); + } + await this.clients.redis.decrby(COUNTER_KEY, entryIds.length); + } + + /** + * Point the sweeper's index at this subscription's oldest delivery, or take + * the subscription out of the region entirely once it has none — which is + * where every key it owned goes. + */ + async #reindex(subId: string): Promise { + const oldest = await this.#scripts().pendingReindex( + entriesKey(subId), + pendingKey(subId), + leaseKey(subId), + ); + if (oldest !== '-1') { + await this.clients.redis.zadd( + INDEX_KEY, + Number(oldest) || 0, + subId, + ); + return; + } + await this.clients.redis.zrem(INDEX_KEY, subId); + // The index hashes to another slot, so no script can cover both it and + // the queue: an append that landed since the script ran has entries + // the ZREM just hid from the sweeper. Put it back if so. + if ((await this.depth(subId)) > 0) + await this.clients.redis.zadd(INDEX_KEY, 'NX', Date.now(), subId); + } +} diff --git a/src/backend/stores/events/types.ts b/src/backend/stores/events/types.ts index 019bf01bf..de718d393 100644 --- a/src/backend/stores/events/types.ts +++ b/src/backend/stores/events/types.ts @@ -39,6 +39,30 @@ export const isSubscriptionTarget = ( ): value is SubscriptionTarget => SUBSCRIPTION_TARGETS.includes(value as SubscriptionTarget); +/** Transports a session row may take: it has one connection and no handler. */ +export const SESSION_TARGETS: SubscriptionTarget[] = ['socket']; + +/** Transports a durable row takes unless the caller says otherwise. */ +export const DEFAULT_DURABLE_TARGETS: SubscriptionTarget[] = [ + 'socket', + 'worker', +]; + +/** + * 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. + */ +export const targetsAllowedForDelivery = ( + delivery: DeliveryClass, + targets: readonly SubscriptionTarget[], +): boolean => + delivery !== 'single' || + (!targets.includes('push') && targets.includes('worker')); + /** What dispatch needs from a subscription, whichever store it came from. */ export interface DispatchSubscription { subId: string; @@ -59,13 +83,16 @@ export interface DispatchSubscription { appUid: string | null; /** ACL mode the subscribe check passed under; re-checked per delivery. */ permission: AclMode; + /** Transports this row's deliveries may take. */ + targets?: SubscriptionTarget[]; /** Session rows only: the connection a delivery is addressed at. */ socketId?: string; /** Durable rows only: set on every row that outlives its connection. */ durable?: true; delivery?: DeliveryClass; - targets?: SubscriptionTarget[]; handlerName?: string | null; + /** Durable rows only: handed to the handler, and read nowhere else. */ + context?: string | null; } export interface SessionSubscription extends DispatchSubscription { diff --git a/src/backend/stores/index.ts b/src/backend/stores/index.ts index 4bcb1a738..8e1cef9c1 100644 --- a/src/backend/stores/index.ts +++ b/src/backend/stores/index.ts @@ -23,6 +23,7 @@ import { FSEntryStore } from './fs/FSEntryStore.js'; import { GroupStore } from './group/GroupStore.js'; import { DurableSubscriptionStore } from './events/DurableSubscriptionStore.js'; import { EventSubscriptionStore } from './events/EventSubscriptionStore.js'; +import { PendingDeliveryStore } from './events/PendingDeliveryStore.js'; import { CreditHoldStore } from './metering/CreditHoldStore.js'; import { MeteringBufferStore } from './metering/MeteringBufferStore.js'; import { NotificationStore } from './notification/NotificationStore.js'; @@ -64,6 +65,7 @@ declare module './types.js' { userBlock: UserBlockStore; eventSubscription: EventSubscriptionStore; durableSubscription: DurableSubscriptionStore; + pendingDelivery: PendingDeliveryStore; } } @@ -95,6 +97,7 @@ export const puterStores = { userBlock: UserBlockStore, // Redis only, no peer stores. eventSubscription: EventSubscriptionStore, + pendingDelivery: PendingDeliveryStore, // Writes through the Redis keyspace above, so it comes after it. durableSubscription: DurableSubscriptionStore, } satisfies IPuterStoreRegistry; diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md index b8465dc68..1efbc54eb 100644 --- a/src/docs/src/rate-limits-and-quotas.md +++ b/src/docs/src/rate-limits-and-quotas.md @@ -168,6 +168,8 @@ One write can reach many subscriptions, so events are bounded on both halves: ho | Matched subscriptions per event | 50 | | Filter evaluations per event | 200 | | Deliveries per minute, per subscription | 600 | +| Acknowledgements per minute | 600 | +| Undelivered deliveries per subscription | 10,000 | Subscriptions come in two kinds. A **session** subscription lives with the connection that made it: it is dropped when the connection closes, and a reconnecting client subscribes again. A **durable** subscription outlives every connection — it is created over the API, listed and revoked from the account, and keeps delivering until you remove it or it expires. @@ -181,6 +183,8 @@ Match patterns are compiled once when you subscribe and are capped at **256 char 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 `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 | Limit | Paid | Free | Anonymous | diff --git a/src/puter-js/src/modules/events/types.js b/src/puter-js/src/modules/events/types.js index c9eb916db..a8baef596 100644 --- a/src/puter-js/src/modules/events/types.js +++ b/src/puter-js/src/modules/events/types.js @@ -43,8 +43,9 @@ * @property {string} subject The subject that was being delivered. * @property {'gap'} op Always `'gap'`. * @property {string} reason Why the delivery was dropped — - * `matched_subscription_limit`, `filter_evaluation_limit`, or - * `delivery_rate_limit`. + * `matched_subscription_limit`, `filter_evaluation_limit`, + * `delivery_rate_limit`, or `backlog_overflow` when undelivered events were + * shed to stay inside a backlog cap. * @property {number} ts Milliseconds since the epoch. */