diff --git a/AGENTS.md b/AGENTS.md index dcc3e4a51..d7a9c6c80 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -94,6 +94,7 @@ Follow the same layered structure inside an extension — unless it only needs a - Vitest; test files sit next to the code they test (`*.test.ts` / `*.test.js`). Run with `npm run test:backend`. - **Mock data, not methods.** Stub inputs (fixtures, fake rows, payloads), not the function under test or the layer beneath it — over-mocking produces tests that pass while production breaks. If you must mock, mock at a real boundary (a client/external service). - **Prefer the test server over mocking deps.** `setupPuterTestEnv()` in [src/backend/testUtil.ts](src/backend/testUtil.ts) boots a fully in-memory backend; hit a real database/client shape where reasonable — integration shapes catch what mocked unit tests miss. +- **Test code, not docs.** Never write a test that reads a file under `src/docs/` and asserts on its wording or numbers. Docs are kept in step by the PR (see the limits rule above) and checked in review; a test that greps a markdown page fails on every rewording and verifies nothing about behavior. --- diff --git a/src/backend/controllers/events/limits.test.ts b/src/backend/controllers/events/limits.test.ts new file mode 100644 index 000000000..ffd367c7f --- /dev/null +++ b/src/backend/controllers/events/limits.test.ts @@ -0,0 +1,107 @@ +/* + * Copyright (C) 2024-present Puter Technologies Inc. + * + * This file is part of Puter. + * + * Puter is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published + * by the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see . + */ + +/** + * The event limits, and the page that publishes them. An undisclosed limit is + * one a developer meets as a service failure, so the numbers here and the + * numbers on the page are held against each other. + */ + +import { describe, expect, it } from 'vitest'; +import { + DEFAULT_FREE_SUBSCRIPTION, + DEFAULT_TEMP_SUBSCRIPTION, +} from '../../services/metering/consts.js'; +import { + EVENTS_BROADCAST_DELIVERY_LIMIT, + EVENTS_DURABLE_SUBSCRIPTIONS_MAX, + EVENTS_DURABLE_SUBSCRIPTIONS_PER_APP, + EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER, + EVENTS_SINGLE_DELIVERY_LIMIT, + EVENTS_WORKER_INVOCATION_LIMIT, + limitFor, + type TieredLimit, +} from './limits.js'; + +const tiers: Array<[string, TieredLimit]> = [ + ['durable subscriptions per account', EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER], + ['durable subscriptions per app', EVENTS_DURABLE_SUBSCRIPTIONS_PER_APP], +]; + +describe('the tiered subscription quotas', () => { + it.each(tiers)('%s never lets a free tier exceed paid', (_name, tier) => { + for (const n of Object.values(tier.bySubscription)) + expect(n).toBeLessThanOrEqual(tier.limit); + }); + + it.each(tiers)('%s gives a temporary account none', (_name, tier) => { + expect(tier.bySubscription[DEFAULT_TEMP_SUBSCRIPTION]).toBe(0); + }); + + it('keeps what one app may take below what the account may hold', () => { + for (const plan of [ + null, + DEFAULT_FREE_SUBSCRIPTION, + DEFAULT_TEMP_SUBSCRIPTION, + ]) { + expect( + limitFor(EVENTS_DURABLE_SUBSCRIPTIONS_PER_APP, plan), + ).toBeLessThanOrEqual( + limitFor(EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER, plan), + ); + } + }); + + it('holds an unrecognised plan to the paid base', () => { + expect(limitFor(EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER, 'some_plan')).toBe( + EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER.limit, + ); + expect(limitFor(EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER, null)).toBe( + EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER.limit, + ); + }); + + it('reads its structural maximum off the paid cap', () => { + expect(EVENTS_DURABLE_SUBSCRIPTIONS_MAX).toBe( + EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER.limit, + ); + }); +}); + +describe('the delivery budgets', () => { + it('keeps `single` well under broadcast — each one costs far more', () => { + expect(EVENTS_SINGLE_DELIVERY_LIMIT.limit).toBeLessThan( + EVENTS_BROADCAST_DELIVERY_LIMIT.limit, + ); + expect(EVENTS_WORKER_INVOCATION_LIMIT.limit).toBeLessThan( + EVENTS_SINGLE_DELIVERY_LIMIT.limit, + ); + }); + + it('pins an explicit scope on each, so two call sites share one counter', () => { + for (const spec of [ + EVENTS_BROADCAST_DELIVERY_LIMIT, + EVENTS_SINGLE_DELIVERY_LIMIT, + EVENTS_WORKER_INVOCATION_LIMIT, + ]) { + expect(spec.scope).toBeTruthy(); + expect(spec.window).toBe(60_000); + } + }); +}); diff --git a/src/backend/controllers/events/limits.ts b/src/backend/controllers/events/limits.ts index 294ba5b25..f66b29b63 100644 --- a/src/backend/controllers/events/limits.ts +++ b/src/backend/controllers/events/limits.ts @@ -18,6 +18,10 @@ */ import type { RouteRateLimit } from '../../core/http/types'; +import { + DEFAULT_FREE_SUBSCRIPTION, + DEFAULT_TEMP_SUBSCRIPTION, +} from '../../services/metering/consts.js'; // -- Shared event limits --------------------------------------------- // @@ -38,6 +42,40 @@ const userWindow = ( window = 60_000, ): RouteRateLimit => ({ scope, limit, window, key: 'user' }); +/** + * A cap that varies by plan, in the shape route gates already declare theirs + * in: the base is what a subscribed account sees, and `bySubscription` carves + * the free tiers out beneath it. A plan nobody enumerated falls through to the + * base, so a new one is generous rather than accidentally throttled. + */ +export interface TieredLimit { + limit: number; + bySubscription: Record; +} + +const tiered = (paid: number, free: number, temp: number): TieredLimit => ({ + limit: paid, + bySubscription: { + [DEFAULT_FREE_SUBSCRIPTION]: free, + [DEFAULT_TEMP_SUBSCRIPTION]: temp, + }, +}); + +/** The cap one plan sees. An unresolved plan is held to the base. */ +export const limitFor = ( + tier: TieredLimit, + subscriptionId: string | null, +): number => + (subscriptionId === null + ? undefined + : tier.bySubscription[subscriptionId]) ?? tier.limit; + +/** The two counts a durable subscribe is held to, already resolved by plan. */ +export interface SubscriptionQuota { + perUser: number; + perApp: number; +} + // -- Subscription surface -------------------------------------------- /** @@ -55,10 +93,23 @@ export const EVENTS_SESSION_SUBSCRIPTIONS_PER_SOCKET = 50; * * These are table rows that keep costing after the client that made them is * gone — a delivery each time their anchor changes, and a cache entry in every - * region that sees a write. Counted over the holder index. Per-plan tiering - * arrives with the metering that prices them. + * region that sees a write. Counted over the holder index. A temporary account + * holds none: nothing outlives its connection to deliver to. */ -export const EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER = 500; +export const EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER = tiered(500, 100, 0); + +/** + * Durable subscriptions one app may hold for one account. + * + * Below the per-account cap so that one app cannot spend an account's whole + * budget: the account-wide number is what the watched-token set costs, and this + * is what any single app may take of it. + */ +export const EVENTS_DURABLE_SUBSCRIPTIONS_PER_APP = tiered(100, 25, 0); + +/** Most rows any account can hold, whatever its plan — a read bound, not a gate. */ +export const EVENTS_DURABLE_SUBSCRIPTIONS_MAX = + EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER.limit; /** * How long a suspended durable subscription is kept before it is deleted. @@ -153,6 +204,33 @@ export const EVENTS_BROADCAST_DELIVERY_LIMIT = userWindow( 600, ); +/** + * `single` deliveries per minute, per subscription. + * + * A fifth of the broadcast budget: each one is leased, acknowledged and may run + * an app's handler, so it costs an order of magnitude more than a socket copy. + * Over it the event is not queued — a gap marker takes its place, so the + * consumer learns it fell behind instead of inheriting a backlog it can never + * work through. + */ +export const EVENTS_SINGLE_DELIVERY_LIMIT = userWindow( + 'events:delivery:single', + 120, +); + +/** + * Handler invocations per minute, per (account, app). + * + * The one term of the fan-out product that costs real compute, so it is capped + * per app rather than per subscription — an app cannot widen it by holding more + * subscriptions. A delivery that arrives over the budget is not failed: it + * stays owed, and its lease is the backoff. + */ +export const EVENTS_WORKER_INVOCATION_LIMIT = userWindow( + 'events:worker:invoke', + 60, +); + /** * Filter evaluations one event may spend. Lives with the matcher because the * primitive that enforces it does; re-exported here so every published number diff --git a/src/backend/services/events/EventsService.test.ts b/src/backend/services/events/EventsService.test.ts index 4b1787716..0e1539ec9 100644 --- a/src/backend/services/events/EventsService.test.ts +++ b/src/backend/services/events/EventsService.test.ts @@ -32,7 +32,9 @@ import { type DurableSubscription, } from '../../stores/events/EventSubscriptionStore.js'; import type { FSEntry } from '../../stores/fs/FSEntry.js'; +import type { UsageInput } from '../metering/types.js'; import type { IConfig } from '../../types.js'; +import { EVENTS_COSTS } from './costs.js'; import { EventsService, EVENTS_ACK_VERB, @@ -63,6 +65,7 @@ let store: EventSubscriptionStore; let service: EventsService; let sent: Array<{ socket?: string; envelope: DeliveryEnvelope }>; let delivered: DeliveryEnvelope[]; +let metered: MeteredLine[]; let entries: Map; let eventBus: { on: ReturnType; emit: ReturnType }; @@ -221,6 +224,18 @@ const durableSubscriptionStore = { getBySubId: async () => null, }; +/** One buffered usage line, with the identity it was written as. */ +interface MeteredLine { + userUuid: string | undefined; + appUid: string | null; + usageType: string; + usageAmount: number; + costOverride?: number; +} + +/** Whether the account being delivered to still has budget. */ +let hasCredits: boolean; + /** * Each service gets its own outbox. A delivery still in flight when a test * ends must land in that test's record, not in the next one's. @@ -231,10 +246,12 @@ const buildService = ( service: EventsService; sent: Array<{ socket?: string; envelope: DeliveryEnvelope }>; delivered: DeliveryEnvelope[]; + metered: MeteredLine[]; eventBus: { on: ReturnType; emit: ReturnType }; } => { const outbox: Array<{ socket?: string; envelope: DeliveryEnvelope }> = []; const counted: DeliveryEnvelope[] = []; + const lines: MeteredLine[] = []; const bus = { on: vi.fn(), emit: vi.fn() }; const built = new EventsService( config, @@ -266,11 +283,31 @@ const buildService = ( }, acl: aclService(), permission: permissionService(), + metering: { + bufferIncrementUsages: ( + actor: Actor, + usages: UsageInput[], + ) => { + for (const usage of usages) + lines.push({ + userUuid: actor.user?.uuid, + appUid: actor.app?.uid ?? null, + ...usage, + }); + }, + hasAnyUsageCached: async () => hasCredits, + }, } as never, ); built.onDelivered = (envelope) => counted.push(envelope); built.onServerStart(); - return { service: built, sent: outbox, delivered: counted, eventBus: bus }; + return { + service: built, + sent: outbox, + delivered: counted, + metered: lines, + eventBus: bus, + }; }; const fsEntryStore = { @@ -382,13 +419,14 @@ beforeEach(() => { grants = new Set(); permissionChecks = []; permissionGeneration = 1; + hasCredits = true; redis = countingRedis(new MockRedis.Cluster(['redis://localhost:7001'])); store = new EventSubscriptionStore( {} as IConfig, { redis } as never, {} as never, ); - ({ service, sent, delivered, eventBus } = buildService({ + ({ service, sent, delivered, metered, eventBus } = buildService({ events: { enabled: true }, } as IConfig)); }); @@ -940,6 +978,9 @@ describe('matching', () => { await flush(); expect(sent[0].envelope.event).toMatchObject({ self: false }); + // The subscription's holder is billed for their own delivery — never + // the actor whose write happened to trigger it. + expect(metered).toMatchObject([{ userUuid: `user-${userId}` }]); }); it('stops delivering the moment the holder`s access goes', async () => { @@ -1072,6 +1113,73 @@ describe('coalescing', () => { expect(delivered).toHaveLength(sent.length); expect(delivered).toHaveLength(1); + // Nine writes, one delivery, one line: what the coalescer collapses is + // never billed for. + expect(metered).toEqual([ + { + userUuid: `user-${userId}`, + appUid: null, + usageType: 'events:delivery:broadcast', + usageAmount: 1, + costOverride: EVENTS_COSTS['events:delivery:broadcast'], + }, + ]); + }); + + it('bills a filtered-out event to nobody', async () => { + const { file } = seedTree(); + await subscribe(`fs:/u${userId}/Documents/*.md`); + + await dispatch(file); + await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1); + + expect(delivered).toEqual([]); + expect(metered).toEqual([]); + }); + + it('bills a session subscription at the broadcast rate, to its holder', async () => { + const { documents, file } = seedTree(); + await subscribe(`fs:${documents.uid}`); + + await dispatch(file); + await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1); + + expect(metered).toMatchObject([ + { + userUuid: `user-${userId}`, + usageType: 'events:delivery:broadcast', + }, + ]); + }); + + it('does not bill a gap marker, which is a notice rather than a delivery', async () => { + const { documents, file } = seedTree(); + await seedSubscriptions(EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT + 5, { + token: `f#${documents.uid}`, + anchorUid: documents.uid, + anchorPath: documents.path, + match: null, + }); + + await dispatch(file); + await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1); + + const gaps = sent.filter((s) => s.envelope.event.op === 'gap'); + expect(gaps).toHaveLength(5); + expect(metered).toHaveLength(EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT); + }); + + it('stops delivering to a holder with nothing left to spend', async () => { + const { documents, file } = seedTree(); + await subscribe(`fs:${documents.uid}`); + hasCredits = false; + + await dispatch(file); + await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1); + + expect(sent).toEqual([]); + expect(delivered).toEqual([]); + expect(metered).toEqual([]); }); }); diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index 336b9a388..2b96d66dd 100644 --- a/src/backend/services/events/EventsService.ts +++ b/src/backend/services/events/EventsService.ts @@ -24,14 +24,20 @@ import { EVENTS_BROADCAST_DELIVERY_LIMIT, EVENTS_COALESCE_WINDOW_MS, EVENTS_CONSECUTIVE_FAILURES, + EVENTS_DURABLE_SUBSCRIPTIONS_PER_APP, + EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER, EVENTS_HANDLER_PUBLISH_BATCH, EVENTS_HANDLER_PUBLISH_LIMIT, EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT, + EVENTS_SINGLE_DELIVERY_LIMIT, EVENTS_SUBSCRIBE_LIMIT, + EVENTS_WORKER_INVOCATION_LIMIT, + limitFor, SUSPENDED_ROW_TTL_DAYS, + type SubscriptionQuota, } from '../../controllers/events/limits.js'; -import type { Actor } from '../../core/actor.js'; -import { HttpError } from '../../core/http/HttpError.js'; +import { makeActor, type Actor } from '../../core/actor.js'; +import { HttpError, isHttpError } from '../../core/http/HttpError.js'; import { checkRateLimit } from '../../core/http/middleware/rateLimit.js'; import type { ReanchorInput, @@ -70,6 +76,7 @@ import { parseKvNamespace } from '../../stores/systemKv/SystemKVStore.js'; import type { PageResult } from '../../util/pagination.js'; import type { AclMode, ResourceDescriptor } from '../acl/ACLService.js'; import { resolveNode } from '../fs/resolveNode.js'; +import { assertActorHasCredits } from '../metering/enforcement.js'; import { appSocketRoom, type SocketSpecifier, @@ -100,6 +107,12 @@ import { type SubscriptionGrant, } from './authorization.js'; import { DeliveryCoalescer } from './coalescer.js'; +import { + DELIVERY_USAGE_TYPES, + EVENTS_COSTS, + EVENTS_COST_UNITS, + type EventsUsageType, +} from './costs.js'; import { DeliveryAuthCache } from './deliveryAuthCache.js'; import { FILTER_EVALUATIONS_PER_EVENT, @@ -303,6 +316,28 @@ interface AddressedDelivery { /** False for a row that asked for its handler and no socket copy. */ socket: boolean; worker?: WorkerInvocation; + meter: DeliveryMeter; + /** + * Whether this specific call is the one that charges for the event. Always + * true for `broadcast` (one send, no retries); a `single` retry after a + * lease expiry carries `false` — the same event was already charged for on + * whichever attempt reached here first. + */ + bill: boolean; +} + +/** + * Who a delivery is billed to, and at which class's rate. Carried with the + * delivery rather than looked up when it lands: by then the row that answers + * both may already have been suspended or removed. + */ +interface DeliveryMeter { + holderUserId: number; + /** The app whose subscription this is, so usage is attributable to it. */ + appUid: string | null; + deliveryClass: DeliveryClass; + /** Only a durable row has a state to suspend when the balance runs out. */ + durable: boolean; } /** @@ -418,6 +453,30 @@ const PENDING_DRAIN_BATCH = 25; */ const DELIVERY_TOKEN_TTL = '5m'; +// -- Metering --------------------------------------------------------- + +/** + * How long a holder's metering identity is reused, and how many are held. + * + * Deliveries for one holder arrive in bursts, and the identity behind them — + * their user row, and the app the subscription belongs to — does not move + * between them. Without this, every delivered event pays a user lookup to + * record a line worth a fraction of a microcent. + */ +const METER_ACTOR_TTL_MS = 60_000; +const METER_ACTOR_LIMIT = 5_000; + +/** Rows one page of the credit sweep takes, and pages one sweep takes. */ +const NO_CREDIT_SWEEP_BATCH = 500; +const NO_CREDIT_SWEEP_MAX_BATCHES = 200; + +/** + * How often suspensions waiting on a restored balance are re-checked. Well + * inside the hour a `no_credit` backlog is held for, so a top-up gets the + * subscription back before what it was owed expires. + */ +const NO_CREDIT_SWEEP_INTERVAL_MS = 15 * 60 * 1000; + /** The part of a socket this service uses, so tests need not build one. */ export interface EventSocket { id: string; @@ -447,6 +506,19 @@ const handlerRequired = (): HttpError => legacyCode: 'events_handler_required', }); +/** + * A temporary account gets session subscriptions and nothing else. A durable + * row outlives the account itself and is revoked from a settings surface a + * temporary account never reaches — so this is refused outright rather than + * sold as a quota of zero. + */ +const durableNeedsAccount = (): HttpError => + new HttpError( + 403, + 'A temporary account may only subscribe for the life of its connection', + { legacyCode: 'events_durable_requires_account' }, + ); + const handlerNotFound = (name: string): HttpError => new HttpError(404, `No handler named \`${name}\` is published`, { legacyCode: 'events_handler_not_found', @@ -594,6 +666,23 @@ const deliverable = (row: DispatchSubscription): boolean => { const isSingle = (row: DispatchSubscription): boolean => row.durable === true && row.delivery === 'single'; +/** Who one row's deliveries are billed to, and at which rate. */ +const meterFor = (row: DispatchSubscription): DeliveryMeter => ({ + holderUserId: row.holderUserId, + appUid: row.appUid, + deliveryClass: isSingle(row) ? 'single' : 'broadcast', + durable: row.durable === true, +}); + +/** The marker that stands in for an event a delivery budget refused. */ +const rateLimitGap = (event: DeliverableEvent): GapMarker => ({ + id: event.id, + subject: event.subject, + op: 'gap', + reason: 'delivery_rate_limit', + ts: event.ts, +}); + /** * Whether a KV row reaches past its own namespace. Read off the app the row was * created by — which is the actor's `effectiveApp` at subscribe time, so an @@ -733,10 +822,16 @@ export class EventsService extends PuterService { readonly #compiled = new Map(); readonly #lookups = new Map>(); readonly #refreshTimers = new Map>(); + /** Holder identities the metering lines are written as. */ + readonly #meterActors = new Map< + string, + { actor: Actor; expiresAt: number } + >(); #coalescer: DeliveryCoalescer | null = null; #expirySweep: ReturnType | null = null; #expiryKick: ReturnType | null = null; #pendingSweep: ReturnType | null = null; + #creditSweep: ReturnType | null = null; /** * What runs an app's handler. Replaced at start-up by the invoker that @@ -790,6 +885,7 @@ export class EventsService extends PuterService { this.#armExpirySweep(); this.#armPendingSweep(); + this.#armCreditSweep(); } override onServerPrepareShutdown(): void { @@ -799,6 +895,8 @@ export class EventsService extends PuterService { this.#expirySweep = null; if (this.#pendingSweep) clearInterval(this.#pendingSweep); this.#pendingSweep = null; + if (this.#creditSweep) clearInterval(this.#creditSweep); + this.#creditSweep = null; } override onServerShutdown(): void { @@ -806,6 +904,22 @@ export class EventsService extends PuterService { this.#refreshTimers.clear(); } + /** + * What one delivery costs, in the shape the driver surfaces report theirs. + * Nothing consumes this at runtime — the rates are published on the + * rate-limits page, which is where a developer actually reads them. + */ + getReportedCosts(): Record[] { + return Object.entries(EVENTS_COSTS).map( + ([usageType, ucentsPerUnit]) => ({ + usageType, + ucentsPerUnit, + unit: EVENTS_COST_UNITS[usageType as EventsUsageType], + source: 'service:events', + }), + ); + } + /** The master switch. Read on every write, so it stays a field lookup. */ get enabled(): boolean { return this.config.events?.enabled === true; @@ -985,6 +1099,7 @@ export class EventsService extends PuterService { const delivery = parseDelivery(request?.delivery); const appUid = actor.effectiveApp?.uid ?? null; + const limits = await this.#subscriptionQuota(actor); const targets = parseTargets(request?.targets, delivery, appUid); const handlerName = parseHandlerName(request?.handlerName); const handlerHash = parseHandlerHash(request?.handlerHash); @@ -1025,12 +1140,46 @@ export class EventsService extends PuterService { context, permission: anchor.permission, expiresAt, + limits, }); this.#publishGeneration(bump, true); return { sub: toDurableView(row) }; } + /** + * How many durable rows this caller's plan allows, in total and for the app + * they are acting as. The store enforces them against one count, so the + * plan is read here and the counting stays where the index is. + * + * A deployment with no metering has no plans to read, and is held to the + * paid caps. + */ + async #subscriptionQuota(actor: Actor): Promise { + const plan = await this.#planId(actor); + if ( + plan !== null && + limitFor(EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER, plan) === 0 + ) + throw durableNeedsAccount(); + return { + perUser: limitFor(EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER, plan), + perApp: limitFor(EVENTS_DURABLE_SUBSCRIPTIONS_PER_APP, plan), + }; + } + + /** The plan this actor is on, or `null` where there are no plans. */ + async #planId(actor: Actor): Promise { + const metering = this.services.metering; + if (!metering) return null; + try { + return (await metering.getActorSubscription(actor)).id; + } catch (err) { + console.warn('[events] could not resolve a plan', err); + return null; + } + } + /** * What this actor holds durably. An app-context actor is confined to its * own rows by the index the query runs on; an account-context one sees @@ -1374,6 +1523,50 @@ export class EventsService extends PuterService { ); } + /** + * Put back subscriptions a restored balance releases. Lazy on purpose: a + * top-up is not something this service hears about, and coupling delivery + * to the payment path would make one more thing that has to be told. The + * cost of the delay is bounded by how long a `no_credit` backlog is held. + */ + async sweepNoCredit(): Promise { + if (!this.enabled || !this.services.metering) return 0; + + let resumed = 0; + let after = 0; + for (let pass = 0; pass < NO_CREDIT_SWEEP_MAX_BATCHES; pass++) { + const page = + await this.stores.durableSubscription.listSuspendedPage( + 'no_credit', + after, + NO_CREDIT_SWEEP_BATCH, + ); + const holders = new Set(page.rows.map((row) => row.holderUserId)); + for (const holderUserId of holders) { + const actor = await this.#meterActor({ + holderUserId, + appUid: null, + deliveryClass: 'broadcast', + durable: true, + }); + if (!actor) continue; + try { + await assertActorHasCredits( + this.services.metering, + actor, + this.config, + ); + } catch { + continue; + } + resumed += await this.resumeForCredit(holderUserId); + } + if (page.nextId === null) break; + after = page.nextId; + } + return resumed; + } + async #sweepInBatches( pass: (batchSize: number) => Promise, ): Promise { @@ -1963,7 +2156,7 @@ export class EventsService extends PuterService { // coalesced or broadcast — collapsing two of them would drop one // the subscription was promised. if (isSingle(row)) { - await this.#owe(row, event); + await this.#oweSingle(row, event); continue; } @@ -1973,6 +2166,9 @@ export class EventsService extends PuterService { envelope: { subId: row.subId, event }, socket: targets.includes('socket'), worker: this.#workerInvocation(row, event), + meter: meterFor(row), + // `broadcast` is one send per delivery — no retry to dedup. + bill: true, }); } @@ -2198,7 +2394,9 @@ export class EventsService extends PuterService { }; // A marker is a delivery, so it takes the same route its // subscription's events would: queued for a `single`, sent for the - // rest. + // rest. It is never rate limited and never metered — it exists to + // say something was lost, and charging for that would bill the + // holder for the loss. if (isSingle(row)) { void this.#owe(row, marker); continue; @@ -2206,10 +2404,14 @@ export class EventsService extends PuterService { // 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({ + void this.#send({ target: deliveryTarget(row), socket: true, envelope: { subId: row.subId, event: marker }, + meter: meterFor(row), + // A gap marker is never billed regardless — `#delivered`'s own + // op check is the real guard — but it never earns the claim. + bill: false, }); } } @@ -2571,6 +2773,24 @@ export class EventsService extends PuterService { // -- Owed deliveries --------------------------------------------- + /** + * One `single` event, or the marker that says its subscription is being + * delivered faster than the class allows. Spent before the queue: a budget + * checked at hand-out time would let a backlog build that nothing can ever + * work through. + */ + async #oweSingle( + row: DispatchSubscription, + event: DeliverableEvent, + ): Promise { + const allowed = await checkRateLimit( + `${EVENTS_SINGLE_DELIVERY_LIMIT.scope}:${row.subId}`, + EVENTS_SINGLE_DELIVERY_LIMIT.limit, + EVENTS_SINGLE_DELIVERY_LIMIT.window, + ); + await this.#owe(row, allowed ? event : rateLimitGap(event)); + } + /** * 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 @@ -2622,6 +2842,11 @@ export class EventsService extends PuterService { row: DispatchSubscription, claimed: ClaimedDelivery, ): Promise { + const meter = meterFor(row); + // Held, not dropped: a delivery its holder cannot pay for waits out the + // suspension's window and goes out if the balance comes back. + if (!(await this.#chargeable(meter, row.subId))) return false; + const targets = targetsOf(row); const target = deliveryTarget(row); @@ -2637,7 +2862,7 @@ export class EventsService extends PuterService { row.subId, claimed.entryId, ); - this.#send({ + await this.#send({ target, socket: true, envelope: { @@ -2646,6 +2871,8 @@ export class EventsService extends PuterService { ackRequired: true, ackId: claimed.entryId, }, + meter, + bill: await this.#firstAttempt(row.subId, claimed), }); return false; } @@ -2656,8 +2883,17 @@ export class EventsService extends PuterService { 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 }); + // Over the invocation budget nothing ran, so nothing was delivered and + // the lease is the backoff — and the budget refusal itself must not + // spend the one bill this entry gets, nor count against the handler. + const outcome = await this.#invokeHandler(invocation); + if (outcome === null) return false; + + this.#delivered( + { subId: row.subId, event: claimed.event }, + meter, + await this.#firstAttempt(row.subId, claimed), + ); if (outcome === 'settled') { await this.stores.pendingDelivery.clearFailures(row.subId); await this.stores.pendingDelivery.settle( @@ -2713,6 +2949,22 @@ export class EventsService extends PuterService { } } + /** + * Whether this is the first genuine attempt at this entry — across however + * many times it is claimed, retried and handed to a different transport. A + * `single` retries the same entry after a lease expiry, and that is one + * owed event, not a new one each time: only the attempt that gets here + * first is billed. A gap marker is never billed at all, so it never spends + * the claim either. + */ + async #firstAttempt( + subId: string, + claimed: ClaimedDelivery, + ): Promise { + if (claimed.event.op === 'gap') return false; + return this.stores.pendingDelivery.markBilled(subId, claimed.entryId); + } + /** What the handler seam is handed, or null for a row that wants none. */ #workerInvocation( row: DispatchSubscription, @@ -2833,29 +3085,35 @@ export class EventsService extends PuterService { async #flush(delivery: AddressedDelivery): Promise { try { + if ( + !(await this.#chargeable( + delivery.meter, + delivery.envelope.subId, + )) + ) + return; + const allowed = await checkRateLimit( `${EVENTS_BROADCAST_DELIVERY_LIMIT.scope}:${delivery.envelope.subId}`, EVENTS_BROADCAST_DELIVERY_LIMIT.limit, EVENTS_BROADCAST_DELIVERY_LIMIT.window, ); if (allowed) { - this.#send(delivery); + await this.#send(delivery); return; } - const event = delivery.envelope.event as ProjectedEvent; - this.#send({ + // The marker goes to the socket only: running the handler on a + // notice that its event was dropped is the invocation the budget + // was refusing. + await this.#send({ target: delivery.target, socket: delivery.socket, + meter: delivery.meter, envelope: { subId: delivery.envelope.subId, - event: { - id: event.id, - subject: event.subject, - op: 'gap', - reason: 'delivery_rate_limit', - ts: event.ts, - }, + event: rateLimitGap(delivery.envelope.event), }, + bill: false, }); } catch (err) { console.warn('[events] delivery failed', err); @@ -2868,7 +3126,7 @@ export class EventsService extends PuterService { * 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 { + async #send(delivery: AddressedDelivery): Promise { if (delivery.socket) { try { void this.services.socket @@ -2885,35 +3143,181 @@ export class EventsService extends PuterService { } } + let invoked = false; 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. It counts // toward nothing either — a row whose socket copies are arriving // must not be stopped by a handler nobody is waiting on. try { - void this.worker.invoke(invocation).catch((err: unknown) => { - console.warn('[events] handler invocation failed', err); - }); + invoked = (await this.#invokeHandler(delivery.worker)) !== null; } catch (err) { console.warn('[events] handler invocation failed', err); } } - this.onDelivered(delivery.envelope); + // Nothing carried it, so nothing was delivered — and a delivery that + // did not happen is not billed. + if (delivery.socket || invoked) + this.#delivered(delivery.envelope, delivery.meter, delivery.bill); + } + + /** + * One event reached a subscriber. Metering rides the same call the seam + * does, so a delivery cannot be reported without being charged for: nothing + * filtered out, coalesced away, rate limited or refused for credit gets + * here, and a gap marker — a notice of loss rather than a delivery — is + * reported without a line. `bill` is false for a `single` retry: the event + * it carries was already charged for on an earlier attempt. + */ + #delivered( + envelope: DeliveryEnvelope, + meter: DeliveryMeter, + bill: boolean, + ): void { + this.onDelivered(envelope); + if (envelope.event.op === 'gap' || !bill) return; + void this.#meterDelivery(meter); } /** * Called once per event that actually reached a subscriber, gap markers - * included. This is the seam metering hangs off — one delivered event is - * one line, which is why nothing filtered out, coalesced away or rate - * limited can arrive here. + * included. */ onDelivered(_envelope: DeliveryEnvelope): void { return; } + // -- Metering ---------------------------------------------------- + + /** + * Record one delivered event against its subscription's holder. Buffered + * rather than written: a line is worth a fraction of a microcent and they + * arrive per event, so writing each one would cost more than it records. + */ + async #meterDelivery(meter: DeliveryMeter): Promise { + const metering = this.services.metering; + if (!metering) return; + try { + const actor = await this.#meterActor(meter); + if (!actor) return; + const usageType = DELIVERY_USAGE_TYPES[meter.deliveryClass]; + metering.bufferIncrementUsages(actor, [ + { + usageType, + usageAmount: 1, + costOverride: EVENTS_COSTS[usageType], + }, + ]); + } catch (err) { + console.warn('[events] could not meter a delivery', err); + } + } + + /** + * Whether this holder's deliveries can still be charged for. Answered from + * the metering service's own cache, so it costs a map read per coalesced + * delivery rather than a lookup per event; the account that cannot pay has + * the subscription suspended and is told once, and comes back through the + * credit sweep. + */ + async #chargeable(meter: DeliveryMeter, subId: string): Promise { + const actor = await this.#meterActor(meter); + if (!actor) return true; + + try { + await assertActorHasCredits( + this.services.metering, + actor, + this.config, + ); + return true; + } catch (err) { + if (!isHttpError(err) || err.statusCode !== 402) { + // Not being able to read a balance is our problem, not a reason + // to stop delivering. + console.warn('[events] could not read a balance', err); + return true; + } + } + + // A session row has no state to suspend and nothing that outlives the + // connection: it simply stops being delivered to. + if (meter.durable && (await this.suspendForNoCredit(subId))) + await this.#notifyNoCredit(meter.holderUserId, meter.appUid); + return false; + } + + /** Tell a holder their subscriptions stopped, and what brings them back. */ + async #notifyNoCredit( + holderUserId: number, + appUid: string | null, + ): Promise { + try { + // The holder's news rather than the developer's — they are the one + // billed, and the one who can act on it. + await this.services.notification.notify( + [holderUserId], + { + title: 'Event delivery stopped', + reason: 'no_credit', + }, + { type: 'app.events.ended', appUid }, + ); + } catch (err) { + console.warn('[events] could not report an empty balance', err); + } + } + + /** + * Run a handler, unless this (account, app) has spent its invocations for + * the minute. `null` says nothing ran: a `single` stays owed and its lease + * paces the next attempt, and a `broadcast` copy is simply not made. + */ + async #invokeHandler( + invocation: WorkerInvocation, + ): Promise { + const allowed = await checkRateLimit( + `${EVENTS_WORKER_INVOCATION_LIMIT.scope}:${invocation.holderUserId}:${invocation.appUid ?? ''}`, + EVENTS_WORKER_INVOCATION_LIMIT.limit, + EVENTS_WORKER_INVOCATION_LIMIT.window, + ); + if (!allowed) return null; + return this.worker.invoke(invocation); + } + + /** + * The identity a holder's lines are written as: their account, acting as + * the app whose subscription this is, so usage lands where the account can + * see which app produced it. + */ + async #meterActor(meter: DeliveryMeter): Promise { + const key = `${meter.holderUserId}|${meter.appUid ?? ''}`; + const now = Date.now(); + const held = this.#meterActors.get(key); + if (held && held.expiresAt > now) return held.actor; + + const user = await this.stores.user.getById(meter.holderUserId); + if (!user) return null; + const actor = makeActor({ + user, + app: meter.appUid ? { uid: meter.appUid } : null, + }); + + // Insertion-ordered, so the oldest goes when a burst of one-off holders + // would otherwise grow this without bound. + if (this.#meterActors.size >= METER_ACTOR_LIMIT) { + const oldest = this.#meterActors.keys().next().value; + if (oldest !== undefined) this.#meterActors.delete(oldest); + } + this.#meterActors.set(key, { + actor, + expiresAt: now + METER_ACTOR_TTL_MS, + }); + return actor; + } + // -- Hot-path cache ---------------------------------------------- /** @@ -3094,6 +3498,17 @@ export class EventsService extends PuterService { this.#expirySweep = sweep; } + #armCreditSweep(): void { + if (!this.enabled) return; + const sweep = setInterval(() => { + void this.sweepNoCredit().catch((err) => { + console.warn('[events] credit sweep failed', err); + }); + }, NO_CREDIT_SWEEP_INTERVAL_MS); + sweep.unref?.(); + this.#creditSweep = sweep; + } + #armPendingSweep(): void { if (!this.enabled) return; const sweep = setInterval(() => { diff --git a/src/backend/services/events/anchorSettle.integration.test.ts b/src/backend/services/events/anchorSettle.integration.test.ts index 5c5bd5ac5..3b8807963 100644 --- a/src/backend/services/events/anchorSettle.integration.test.ts +++ b/src/backend/services/events/anchorSettle.integration.test.ts @@ -101,7 +101,13 @@ const gone = (subId: string) => ); beforeAll(async () => { - env = await setupPuterTestEnv({ events: { enabled: true } } as IConfig); + env = await setupPuterTestEnv({ + events: { enabled: true }, + // Seeded accounts carry no email, which the plan machinery reads as a + // temporary account — and a temporary account holds no durable rows. + // Plans are not what these cases are about. + unlimitedMetering: true, + } as IConfig); const row = await env.server.stores.user.getByUsername( env.users.user.username, ); diff --git a/src/backend/services/events/costs.ts b/src/backend/services/events/costs.ts new file mode 100644 index 000000000..831fc7e19 --- /dev/null +++ b/src/backend/services/events/costs.ts @@ -0,0 +1,44 @@ +/* + * 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 { DeliveryClass } from './registry.js'; + +// Microcents per delivered event. Cost is `EVENTS_COSTS[usageType] * units`. +// +// Rated per delivery class rather than blended: a `single` is leased, acked and +// queued where a broadcast copy is a socket write. The handler run itself is +// the worker's to meter, and a subscription that sits idle costs nothing — +// quotas, not a standing charge, bound how many an account may hold. +export const EVENTS_COSTS = { + 'events:delivery:broadcast': 10, + 'events:delivery:single': 100, +} as const; + +export type EventsUsageType = keyof typeof EVENTS_COSTS; + +export const DELIVERY_USAGE_TYPES: Record = { + broadcast: 'events:delivery:broadcast', + single: 'events:delivery:single', +}; + +/** What one line of each type is counted in, for the published rate table. */ +export const EVENTS_COST_UNITS: Record = { + 'events:delivery:broadcast': 'delivery', + 'events:delivery:single': 'delivery', +}; diff --git a/src/backend/services/events/durable.integration.test.ts b/src/backend/services/events/durable.integration.test.ts index e1a375d71..cb0285097 100644 --- a/src/backend/services/events/durable.integration.test.ts +++ b/src/backend/services/events/durable.integration.test.ts @@ -28,7 +28,12 @@ import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { v4 as uuidv4 } from 'uuid'; -import { EVENTS_COALESCE_WINDOW_MS } from '../../controllers/events/limits.js'; +import { + EVENTS_COALESCE_WINDOW_MS, + EVENTS_DURABLE_SUBSCRIPTIONS_PER_APP, + EVENTS_SUBSCRIBE_LIMIT, +} from '../../controllers/events/limits.js'; +import { DEFAULT_FREE_SUBSCRIPTION } from '../metering/consts.js'; import { makeActor } from '../../core/actor.js'; import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; import type { IConfig } from '../../types.js'; @@ -178,7 +183,13 @@ const quiet = () => ); beforeAll(async () => { - env = await setupPuterTestEnv({ events: { enabled: true } } as IConfig); + env = await setupPuterTestEnv({ + events: { enabled: true }, + // Seeded accounts carry no email, which the plan machinery reads as a + // temporary account — and a temporary account holds no durable rows. + // Plans are not what these cases are about. + unlimitedMetering: true, + } as IConfig); username = env.users.user.username; const user = await env.server.stores.user.getByUsername(username); userId = user!.id; @@ -744,3 +755,131 @@ describe('with events switched off', () => { } }); }); + +/** + * The tiering the rest of this file opts out of. Seeded accounts carry no + * email, which is exactly what the plan machinery reads as a temporary + * account — so this block boots with plans left on and gives the account an + * email when it wants to be a registered one. + */ +describe('what a plan lets an account hold', () => { + let tiered: PuterTestEnv; + let tieredUserId: number; + let tieredAnchor: string; + let tieredApp: { uid: string; token: string }; + + const tieredCall = async ( + path: string, + token: string, + body: object, + ): Promise => { + const response = await fetch(new URL(path, tiered.apiOrigin), { + method: 'POST', + headers: { + 'content-type': 'application/json', + authorization: `Bearer ${token}`, + }, + body: JSON.stringify(body), + }); + return { + status: response.status, + body: (await response.json()) as Record, + }; + }; + + const subscribeTiered = (token: string) => + tieredCall('/events/subscribe', token, { subject: `fs:${tieredAnchor}` }); + + /** + * Move the account between the plans the caps are written against: an + * address on file is what tells them apart. + */ + const setEmail = async (email: string | null) => { + await tiered.server.stores.user.update(tieredUserId, { email }); + const user = await tiered.server.stores.user.getById(tieredUserId); + tiered.server.services.metering.invalidateActorSubscription(user!.uuid); + }; + + beforeAll(async () => { + tiered = await setupPuterTestEnv({ + events: { enabled: true }, + } as IConfig); + const user = await tiered.server.stores.user.getByUsername( + tiered.users.user.username, + ); + tieredUserId = user!.id; + + tieredAnchor = `/${tiered.users.user.username}/tiered`; + await tiered.server.services.fs.mkdir(tieredUserId, { + path: tieredAnchor, + createMissingParents: true, + }); + + const uid = `app-${uuidv4()}`; + await tiered.server.clients.db.write( + 'INSERT INTO `apps` (`uid`, `name`, `title`, `index_url`, `owner_user_id`) VALUES (?, ?, ?, ?, ?)', + [uid, uid, uid, `https://${uid}.example/`, tieredUserId], + ); + const entry = + await tiered.server.stores.fsEntry.getEntryByPath(tieredAnchor); + const actor = await tiered.server.services.auth.authenticate( + tiered.users.user.token, + ); + await tiered.server.services.permission.grantUserAppPermission( + actor.actor!, + uid, + `fs:${entry!.uid}:list`, + ); + // A durable app row targets the worker by default, so the caps are only + // reachable once the background consent behind that target is given. + await tiered.server.services.permission.grantUserAppPermission( + actor.actor!, + uid, + EVENTS_BACKGROUND_PERMISSION, + ); + tieredApp = { + uid, + token: await tiered.server.services.auth.getUserAppToken( + actor.actor!, + uid, + ), + }; + }, BOOT_TIMEOUT_MS); + + afterAll(async () => { + await tiered?.shutdown(); + }); + + it('refuses a temporary account outright — it has session subscriptions', async () => { + await setEmail(null); + + const refused = await subscribeTiered(tiered.users.user.token); + + expect(refused.status).toBe(403); + expect(refused.body.code).toBe('events_durable_requires_account'); + }); + + it('holds a free account to the free per-app cap', async () => { + await setEmail(`${tiered.users.user.username}@example.invalid`); + const cap = + EVENTS_DURABLE_SUBSCRIPTIONS_PER_APP.bySubscription[ + DEFAULT_FREE_SUBSCRIPTION + ]; + + // Both servers in this file share the process-wide Redis mock, so this + // user id's call budget already carries the other server's subscribes. + await tiered.server.clients.redis.del( + `rate:${EVENTS_SUBSCRIBE_LIMIT.scope}:${tieredUserId}`, + ); + for (let i = 0; i < cap; i++) + expect((await subscribeTiered(tieredApp.token)).status).toBe(200); + + const refused = await subscribeTiered(tieredApp.token); + expect(refused.status).toBe(429); + expect(refused.body.code).toBe('events_subscription_limit'); + + // The account itself is nowhere near its own cap, so its own session + // may still subscribe. + expect((await subscribeTiered(tiered.users.user.token)).status).toBe(200); + }); +}); diff --git a/src/backend/services/events/handlers.integration.test.ts b/src/backend/services/events/handlers.integration.test.ts index 2ec2520a2..3a8293525 100644 --- a/src/backend/services/events/handlers.integration.test.ts +++ b/src/backend/services/events/handlers.integration.test.ts @@ -149,7 +149,13 @@ const touch = (holder: number, path: string) => env.server.services.fs.touch(holder, { path }); beforeAll(async () => { - env = await setupPuterTestEnv({ events: { enabled: true } } as IConfig); + env = await setupPuterTestEnv({ + events: { enabled: true }, + // Seeded accounts carry no email, which the plan machinery reads as a + // temporary account — and a temporary account holds no durable rows. + // Plans are not what these cases are about. + unlimitedMetering: true, + } as IConfig); const user = await env.server.stores.user.getByUsername( env.users.user.username, diff --git a/src/backend/services/events/kv.integration.test.ts b/src/backend/services/events/kv.integration.test.ts index 8f6f7e7a4..775a8e95f 100644 --- a/src/backend/services/events/kv.integration.test.ts +++ b/src/backend/services/events/kv.integration.test.ts @@ -124,6 +124,9 @@ const clearRows = async () => { beforeAll(async () => { env = await setupPuterTestEnv({ events: { enabled: true, crossAppKv: true }, + // Seeded accounts carry no email, which the plan machinery reads as a + // temporary account — and a temporary account holds no durable rows. + unlimitedMetering: true, } as IConfig); const user = await env.server.stores.user.getByUsername( env.users.user.username, diff --git a/src/backend/services/events/metering.test.ts b/src/backend/services/events/metering.test.ts new file mode 100644 index 000000000..fdbcdb2f6 --- /dev/null +++ b/src/backend/services/events/metering.test.ts @@ -0,0 +1,307 @@ +/* + * 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 . + */ + +/** + * What a plan lets an account hold, and how a suspension for an empty balance + * is lifted. + * + * Delivery metering is held against the delivery paths themselves; this is the + * half that happens on a timer rather than on an event. + */ + +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { + EVENTS_DURABLE_SUBSCRIPTIONS_PER_APP, + EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER, +} from '../../controllers/events/limits.js'; +import type { Actor } from '../../core/actor.js'; +import { isHttpError } from '../../core/http/HttpError.js'; +import type { DurableSubscriptionInput } from '../../stores/events/DurableSubscriptionStore.js'; +import type { DurableSubscription } from '../../stores/events/types.js'; +import type { FSEntry } from '../../stores/fs/FSEntry.js'; +import { + DEFAULT_FREE_SUBSCRIPTION, + DEFAULT_TEMP_SUBSCRIPTION, +} from '../metering/consts.js'; +import type { UsageInput } from '../metering/types.js'; +import type { IConfig } from '../../types.js'; +import { EVENTS_COSTS } from './costs.js'; +import { EventsService } from './EventsService.js'; +import { fsAnchorToken } from './subjects.js'; + +let seq = 0; +let userId = 0; +let service: EventsService; +let rows: DurableSubscription[]; +let metered: Array<{ userUuid: string | undefined } & UsageInput>; +let created: DurableSubscriptionInput[]; +let resumed: string[]; +/** The plan the metering service reports for the acting account. */ +let plan: string; +/** Accounts that still have budget, by user id. */ +let solvent: Set; + +const anchorUid = (): string => `docs-${seq}`; +const anchorPath = (): string => `/u${userId}/Documents`; + +const durableRow = ( + over: Partial = {}, +): DurableSubscription => ({ + durable: true, + subId: `sub-${rows.length}`, + holderUserId: userId, + ownerUserId: userId, + subject: `fs:${anchorPath()}`, + token: fsAnchorToken(anchorUid()), + anchorUid: anchorUid(), + anchorPath: anchorPath(), + match: null, + op: null, + appUid: null, + permission: 'list', + delivery: 'broadcast', + targets: ['socket'], + handlerName: null, + context: null, + expiresAt: null, + suspendedAt: null, + suspendedReason: null, + createdAt: Math.floor(Date.now() / 1000), + ...over, +}); + +const actorFor = (over: Partial = {}): Actor => + ({ + user: { id: userId, uuid: `user-${userId}`, username: `u${userId}` }, + effectiveApp: null, + ...over, + }) as unknown as Actor; + +const appActor = (appUid: string): Actor => + actorFor({ app: { uid: appUid, id: 1 }, effectiveApp: { uid: appUid, id: 1 } }); + +const anchorEntry = (): FSEntry => + ({ + uid: anchorUid(), + uuid: anchorUid(), + path: anchorPath(), + userId, + isDir: true, + }) as FSEntry; + +const codeOf = (code: string) => (err: unknown) => + isHttpError(err) && err.legacyCode === code; + +beforeEach(() => { + seq++; + userId = 5000 + seq; + rows = []; + metered = []; + created = []; + resumed = []; + plan = 'paid_plan'; + solvent = new Set([userId]); + + service = new EventsService( + { events: { enabled: true } } as IConfig, + { + redis: {}, + event: { on: vi.fn(), emit: vi.fn() }, + alarm: { create: vi.fn() }, + } as never, + { + durableSubscription: { + create: async (input: DurableSubscriptionInput) => { + created.push(input); + const row = durableRow({ + appUid: input.appUid, + delivery: input.delivery, + targets: input.targets, + }); + rows.push(row); + return { + row, + bump: { userId: input.ownerUserId, generation: 1 }, + }; + }, + listSuspendedPage: async (reason: string, afterId: number) => + afterId === 0 + ? { + rows: rows.filter( + (row) => row.suspendedReason === reason, + ), + nextId: null, + } + : { rows: [], nextId: null }, + listSuspendedForHolder: async ( + holderUserId: number, + reason: string, + ) => + rows.filter( + (row) => + row.holderUserId === holderUserId && + row.suspendedReason === reason, + ), + resume: async (resuming: readonly DurableSubscription[]) => { + for (const row of resuming) { + resumed.push(row.subId); + row.suspendedAt = null; + row.suspendedReason = null; + } + return [{ userId, generation: 1 }]; + }, + }, + pendingDelivery: { + releaseHold: async () => undefined, + claim: async () => null, + }, + fsEntry: { + getEntryByUuid: async (uid: string) => + uid === anchorUid() ? anchorEntry() : null, + getEntryByPath: async (path: string) => + path === anchorPath() ? anchorEntry() : null, + getEntryById: async () => null, + }, + user: { + getById: async (id: number) => ({ id, uuid: `user-${id}` }), + }, + app: { getByUid: async (uid: string) => ({ uid, id: 1 }) }, + permission: { getCacheGeneration: async () => 1 }, + } as never, + { + socket: { send: vi.fn(), has: () => false }, + fs: { getAncestorChain: async () => [] }, + acl: { + check: async () => true, + getSafeAclError: async () => ({ + status: 404, + message: 'Subject does not exist', + fields: { code: 'subject_does_not_exist' }, + }), + }, + notification: { notify: vi.fn() }, + // Background consent is a given here: this suite is about what a + // plan allows, not about who agreed to it. + permission: { check: async () => true }, + metering: { + bufferIncrementUsages: ( + actor: Actor, + usages: UsageInput[], + ) => { + for (const usage of usages) + metered.push({ + userUuid: actor.user?.uuid, + ...usage, + }); + }, + hasAnyUsageCached: async (actor: Actor) => + solvent.has(Number(actor.user?.id ?? -1)), + getActorSubscription: async () => ({ id: plan }), + }, + } as never, + ); +}); + +describe('a suspension waiting on a balance', () => { + const suspended = () => + durableRow({ + suspendedAt: Math.floor(Date.now() / 1000), + suspendedReason: 'no_credit', + }); + + it('comes back once the account has budget again', async () => { + const row = suspended(); + rows.push(row); + + await expect(service.sweepNoCredit()).resolves.toBe(1); + expect(resumed).toEqual([row.subId]); + expect(row.suspendedReason).toBeNull(); + }); + + it('stays out of service while the account still has none', async () => { + rows.push(suspended()); + solvent.clear(); + + await expect(service.sweepNoCredit()).resolves.toBe(0); + expect(resumed).toEqual([]); + }); + + it('leaves a suspension it did not cause alone', async () => { + rows.push( + durableRow({ + suspendedAt: Math.floor(Date.now() / 1000), + suspendedReason: 'permission_revoked', + }), + ); + + await expect(service.sweepNoCredit()).resolves.toBe(0); + expect(resumed).toEqual([]); + }); +}); + +describe('what a plan lets an account subscribe to', () => { + const subscribe = (actor = actorFor()) => + service.subscribeDurable(actor, { subject: `fs:${anchorUid()}` }); + + it('hands the store the caps the plan resolved to', async () => { + plan = DEFAULT_FREE_SUBSCRIPTION; + + await subscribe(appActor(`app-${seq}`)); + + expect(created[0].limits).toEqual({ + perUser: EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER.bySubscription[ + DEFAULT_FREE_SUBSCRIPTION + ], + perApp: EVENTS_DURABLE_SUBSCRIPTIONS_PER_APP.bySubscription[ + DEFAULT_FREE_SUBSCRIPTION + ], + }); + }); + + it('gives a paid plan the base caps', async () => { + await subscribe(); + + expect(created[0].limits).toEqual({ + perUser: EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER.limit, + perApp: EVENTS_DURABLE_SUBSCRIPTIONS_PER_APP.limit, + }); + }); + + it('refuses a temporary account outright, with a stable code', async () => { + plan = DEFAULT_TEMP_SUBSCRIPTION; + + await expect(subscribe()).rejects.toSatisfy( + codeOf('events_durable_requires_account'), + ); + expect(created).toEqual([]); + }); +}); + +describe('the rates this service reports', () => { + it('names every line it can write, in microcents', () => { + expect(service.getReportedCosts()).toEqual( + Object.entries(EVENTS_COSTS).map(([usageType, ucentsPerUnit]) => ({ + usageType, + ucentsPerUnit, + unit: expect.any(String), + source: 'service:events', + })), + ); + }); +}); diff --git a/src/backend/services/events/revocationSettle.integration.test.ts b/src/backend/services/events/revocationSettle.integration.test.ts index df6db96c7..f30adc199 100644 --- a/src/backend/services/events/revocationSettle.integration.test.ts +++ b/src/backend/services/events/revocationSettle.integration.test.ts @@ -223,7 +223,13 @@ const clearRows = async () => { }; beforeAll(async () => { - env = await setupPuterTestEnv({ events: { enabled: true } } as IConfig); + env = await setupPuterTestEnv({ + events: { enabled: true }, + // Seeded accounts carry no email, which the plan machinery reads as a + // temporary account — and a temporary account holds no durable rows. + // Plans are not what these cases are about. + unlimitedMetering: true, + } as IConfig); const ownerRow = await env.server.stores.user.getByUsername( env.users.user.username, diff --git a/src/backend/services/events/singleDelivery.test.ts b/src/backend/services/events/singleDelivery.test.ts index b49ac2bd8..80f8ad8cc 100644 --- a/src/backend/services/events/singleDelivery.test.ts +++ b/src/backend/services/events/singleDelivery.test.ts @@ -22,6 +22,8 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { EVENTS_COALESCE_WINDOW_MS, EVENTS_REGION_PENDING_CEILING, + EVENTS_SINGLE_DELIVERY_LIMIT, + EVENTS_WORKER_INVOCATION_LIMIT, deliveryBackoffMs, } from '../../controllers/events/limits.js'; import type { Actor } from '../../core/actor.js'; @@ -33,6 +35,8 @@ import type { } from '../../stores/events/types.js'; import type { FSEntry } from '../../stores/fs/FSEntry.js'; import type { IConfig } from '../../types.js'; +import type { UsageInput } from '../metering/types.js'; +import { EVENTS_COSTS } from './costs.js'; import { EventsService, type DeliveryEnvelope } from './EventsService.js'; import { fsAnchorToken } from './subjects.js'; import type { @@ -62,6 +66,19 @@ let invoked: WorkerInvocation[]; let rows: Map; let entries: Map; let alarms: ReturnType; +let notified: ReturnType; +let metered: MeteredLine[]; +/** Whether the holder being delivered to still has budget. */ +let hasCredits: boolean; + +/** One buffered usage line, with the identity it was written as. */ +interface MeteredLine { + userUuid: string | undefined; + appUid: string | null; + usageType: string; + usageAmount: number; + costOverride?: number; +} /** Whether this region holds a connection for the row being delivered to. */ let socketConnected = true; @@ -164,6 +181,9 @@ beforeEach(async () => { rows = new Map(); entries = new Map(); alarms = vi.fn(); + notified = vi.fn(); + metered = []; + hasCredits = true; redis = new MockRedis.Cluster(['redis://localhost:7001']); await redis.del('ev:qx', 'ev:qc'); @@ -198,6 +218,22 @@ beforeEach(async () => { rows.delete(row.subId); return { userId: row.holderUserId, generation: 1 }; }, + suspend: async ( + suspending: readonly DurableSubscription[], + reason: string, + ) => { + const suspended: DurableSubscription[] = []; + for (const row of suspending) { + const next = { + ...row, + suspendedAt: Math.floor(Date.now() / 1000), + suspendedReason: reason, + }; + rows.set(row.subId, next); + suspended.push(next); + } + return { suspended, bumps: [{ userId, generation: 1 }] }; + }, }, fsEntry: { getEntryByUuid: async (uid: string) => @@ -227,6 +263,18 @@ beforeEach(async () => { fields: { code: 'subject_does_not_exist' }, }), }, + notification: { notify: notified }, + metering: { + bufferIncrementUsages: (actor: Actor, usages: UsageInput[]) => { + for (const usage of usages) + metered.push({ + userUuid: actor.user?.uuid, + appUid: actor.app?.uid ?? null, + ...usage, + }); + }, + hasAnyUsageCached: async () => hasCredits, + }, } as never, ); service.onDelivered = (envelope) => delivered.push(envelope); @@ -343,6 +391,36 @@ describe('a delivery owed to exactly one consumer', () => { handlerName: 'onWrite', holderUserId: userId, }); + // Three attempts at the one event it was never acked for — not three + // bills. A retry after a lease expiry is not a new delivery. + expect(metered).toHaveLength(1); + }); + + it('bills a retried delivery once, however many attempts an outage costs', async () => { + socketConnected = false; + const row = await register({ targets: ['worker'] }); + await dispatch(); + + expect(invoked).toHaveLength(1); + expect(metered).toHaveLength(1); + + // The handler keeps failing to settle it: every retry is the same + // owed event, so none of these further attempts bills again. + for (let i = 0; i < 4; i++) { + jump(31_000); + await service.sweepPending(); + } + expect(invoked).toHaveLength(5); + expect(metered).toHaveLength(1); + + // It finally lands. Still one bill for the one event delivered. + workerOutcome = 'settled'; + jump(31_000); + await service.sweepPending(); + + expect(invoked).toHaveLength(6); + expect(metered).toHaveLength(1); + await expect(pending.depth(row.subId)).resolves.toBe(0); }); it('goes straight to the handler when nothing is connected', async () => { @@ -552,3 +630,100 @@ describe('keeping the region counter honest', () => { await expect(pending.regionDepth()).resolves.toBe(1); }); }); + +describe('what a delivery costs its holder', () => { + it('bills a `single` at its own rate, to the account whose row it is', async () => { + const appUid = `app-${seq}`; + await register({ appUid }); + + await dispatch(); + + expect(delivered).toHaveLength(1); + expect(metered).toEqual([ + { + userUuid: `user-${userId}`, + appUid, + usageType: 'events:delivery:single', + usageAmount: 1, + costOverride: EVENTS_COSTS['events:delivery:single'], + }, + ]); + }); + + it('bills a handler run once, and only when one actually ran', async () => { + socketConnected = false; + await register(); + + await dispatch(); + + expect(invoked).toHaveLength(1); + expect(metered).toHaveLength(1); + }); + + it('stops handing out, suspends and tells the holder when the balance is gone', async () => { + const row = await register(); + hasCredits = false; + + await dispatch(); + + expect(sent).toEqual([]); + expect(invoked).toEqual([]); + expect(metered).toEqual([]); + expect(rows.get(row.subId)).toMatchObject({ + suspendedReason: 'no_credit', + }); + expect(notified).toHaveBeenCalledWith( + [userId], + expect.objectContaining({ reason: 'no_credit' }), + expect.objectContaining({ type: 'app.events.ended' }), + ); + // The event itself is held rather than dropped: the suspension's own + // window is what decides how long it survives. + await expect(pending.depth(row.subId)).resolves.toBe(1); + }); +}); + +describe('the budgets a `single` is delivered under', () => { + it('stands a gap marker in for an event past the per-minute budget', async () => { + const row = await register({ targets: ['socket'] }); + + // One consumer holds one lease at a time, so this is the client loop: + // take the delivery, ack it, and let the next one out. + for (let i = 0; i <= EVENTS_SINGLE_DELIVERY_LIMIT.limit; i++) { + await dispatch(entry({ uid: `file-${seq}-${i}` })); + const last = sent[sent.length - 1]; + if (last?.ackId) + await service.ackDelivery(actorFor(), { + subId: row.subId, + id: last.ackId, + }); + } + + const ops = sent.map((envelope) => envelope.event.op); + expect(ops.filter((op) => op !== 'gap')).toHaveLength( + EVENTS_SINGLE_DELIVERY_LIMIT.limit, + ); + expect(sent[sent.length - 1].event).toMatchObject({ + op: 'gap', + reason: 'delivery_rate_limit', + }); + // A marker is not a delivery, so the last one is not billed. + expect(metered).toHaveLength(EVENTS_SINGLE_DELIVERY_LIMIT.limit); + }); + + it('holds a delivery whose app has spent its invocations, without failing it', async () => { + socketConnected = false; + workerOutcome = 'settled'; + const appUid = `app-${seq}`; + const row = await register({ appUid, targets: ['worker'] }); + + for (let i = 0; i <= EVENTS_WORKER_INVOCATION_LIMIT.limit; i++) + await dispatch(entry({ uid: `file-${seq}-${i}` })); + + expect(invoked).toHaveLength(EVENTS_WORKER_INVOCATION_LIMIT.limit); + // Nothing ran for the last one, so nothing was delivered or billed — + // and it is still owed rather than failed. + expect(metered).toHaveLength(EVENTS_WORKER_INVOCATION_LIMIT.limit); + await expect(pending.depth(row.subId)).resolves.toBe(1); + }); +}); diff --git a/src/backend/services/events/suspension.test.ts b/src/backend/services/events/suspension.test.ts index 32e9dec3e..a45c86589 100644 --- a/src/backend/services/events/suspension.test.ts +++ b/src/backend/services/events/suspension.test.ts @@ -26,7 +26,6 @@ import { import { SUSPENDED_REASONS } from '../../stores/events/DurableSubscriptionStore.js'; import { backlogPolicyFor, - isMetered, isResumable, suspendedFor, } from './suspension.js'; @@ -80,11 +79,6 @@ describe('what a suspension state answers', () => { expect(isResumable('permission_revoked')).toBe(false); }); - it('stops metering while a row is out of service', () => { - expect(isMetered({ suspendedAt: null })).toBe(true); - expect(isMetered({ suspendedAt: 1_700_000_000 })).toBe(false); - }); - it('matches a row against the reason a resume would lift', () => { const row = { suspendedAt: 1, suspendedReason: 'handler_not_found' }; expect(suspendedFor(row, 'handler_not_found')).toBe(true); diff --git a/src/backend/services/events/suspension.ts b/src/backend/services/events/suspension.ts index 953da3ea7..d0b83654c 100644 --- a/src/backend/services/events/suspension.ts +++ b/src/backend/services/events/suspension.ts @@ -76,15 +76,6 @@ export const backlogPolicyFor = (reason: SuspendedReason): BacklogPolicy => export const isResumable = (reason: SuspendedReason): boolean => BACKLOG_POLICY[reason].resumable; -/** - * Whether a subscription accrues metering lines. The seam the metering work - * reads: a suspended subscription is not delivering, so it is not billed for - * standing there. - */ -export const isMetered = ( - row: Pick, -): boolean => row.suspendedAt === null; - /** Whether a suspended row is in the state a given resume would lift. */ export const suspendedFor = ( row: Pick, diff --git a/src/backend/services/events/workerInvoker.integration.test.ts b/src/backend/services/events/workerInvoker.integration.test.ts index 62ac2b6c9..bfffcbfa5 100644 --- a/src/backend/services/events/workerInvoker.integration.test.ts +++ b/src/backend/services/events/workerInvoker.integration.test.ts @@ -168,6 +168,9 @@ const heldForMs = async (subId: string): Promise => { beforeAll(async () => { env = await setupPuterTestEnv({ events: { enabled: true, invokeTimeoutMs: INVOKE_TIMEOUT_MS }, + // Seeded accounts carry no email, which the plan machinery reads as a + // temporary account — and a temporary account holds no durable rows. + unlimitedMetering: true, } as IConfig); const user = await env.server.stores.user.getByUsername( diff --git a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts index 33e1ad2a3..9dc67ff7c 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts +++ b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts @@ -26,7 +26,7 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; import { v4 as uuidv4 } from 'uuid'; -import { EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER } from '../../controllers/events/limits.js'; +import { EVENTS_DURABLE_SUBSCRIPTIONS_MAX } from '../../controllers/events/limits.js'; import { isHttpError } from '../../core/http/HttpError.js'; import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; import type { IConfig } from '../../types.js'; @@ -192,7 +192,9 @@ describe('validation at the row write', () => { }), ), ).rejects.toSatisfy(codeOf('invalid_targets')); - await expect(durable().countForHolder(userId)).resolves.toBe(0); + await expect(durable().countForHolder(userId)).resolves.toMatchObject({ + total: 0, + }); }); it('refuses an app`s `single` row with no worker to fall back to', async () => { @@ -232,7 +234,9 @@ describe('validation at the row write', () => { await expect( durable().create(input({ appUid: null, targets: ['socket', 'worker'] })), ).rejects.toSatisfy(codeOf('invalid_targets')); - await expect(durable().countForHolder(userId)).resolves.toBe(0); + await expect(durable().countForHolder(userId)).resolves.toMatchObject({ + total: 0, + }); }); it('allows a `worker` target once the row has an app', async () => { @@ -249,7 +253,9 @@ describe('validation at the row write', () => { await expect( durable().create(input({ context: 'x'.repeat(4097) })), ).rejects.toSatisfy(codeOf('events_context_too_large')); - await expect(durable().countForHolder(userId)).resolves.toBe(0); + await expect(durable().countForHolder(userId)).resolves.toMatchObject({ + total: 0, + }); }); it('accepts a context right up to it', async () => { @@ -263,15 +269,15 @@ describe('validation at the row write', () => { }); describe('the per-account cap', () => { - it('refuses the one past the limit with a stable code', async () => { + const fill = async (count: number, appUid: string | null = null) => { const now = Math.floor(Date.now() / 1000); - for (let i = 0; i < EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER; i++) + for (let i = 0; i < count; i++) await env.server.clients.db.insert('event_subscriptions', { - sub_id: `user#filler-${i}`, + sub_id: `${appUid ?? 'user'}#filler-${i}`, token: token(), owner_user_id: userId, holder_user_id: userId, - app_uid: null, + app_uid: appUid, subject: `fs:${anchorPath}`, anchor_uid: anchorUid, anchor_path: anchorPath, @@ -285,16 +291,57 @@ describe('the per-account cap', () => { expires_at: null, created_at: now, }); + }; + + it('refuses the one past the limit with a stable code', async () => { + await fill(EVENTS_DURABLE_SUBSCRIPTIONS_MAX); await expect(durable().create(input())).rejects.toSatisfy( codeOf('events_subscription_limit'), ); }); - it('counts the holder, not the owner', async () => { + it('holds a row to the caps its caller resolved from a plan', async () => { + const appUid = `app-${uuidv4()}`; + await fill(2, appUid); + + await expect( + durable().create( + input({ appUid, limits: { perUser: 10, perApp: 2 } }), + ), + ).rejects.toSatisfy(codeOf('events_subscription_limit')); + + // The app is full; the account it belongs to is not. + await expect( + durable().create(input({ limits: { perUser: 10, perApp: 2 } })), + ).resolves.toMatchObject({ row: { appUid: null } }); + }); + + it('never lets a resolved cap exceed the structural one', async () => { + await fill(EVENTS_DURABLE_SUBSCRIPTIONS_MAX); + + await expect( + durable().create( + input({ limits: { perUser: 10_000, perApp: 10_000 } }), + ), + ).rejects.toSatisfy(codeOf('events_subscription_limit')); + }); + + it('counts the holder, not the owner, and one app`s share of it', async () => { + const appUid = `app-${uuidv4()}`; await durable().create(input({ holderUserId: otherUserId })); - await expect(durable().countForHolder(userId)).resolves.toBe(0); - await expect(durable().countForHolder(otherUserId)).resolves.toBe(1); + await durable().create(input({ appUid })); + + await expect(durable().countForHolder(userId)).resolves.toMatchObject({ + total: 1, + forApp: 0, + }); + await expect( + durable().countForHolder(userId, appUid), + ).resolves.toMatchObject({ total: 1, forApp: 1 }); + await expect( + durable().countForHolder(otherUserId), + ).resolves.toMatchObject({ total: 1 }); }); it('does not count a row that has expired but not yet been swept', async () => { @@ -302,7 +349,7 @@ describe('the per-account cap', () => { input({ expiresAt: Math.floor(Date.now() / 1000) - 60 }), ); await durable().create(input()); - await expect(durable().countForHolder(userId)).resolves.toBe(1); + await expect(durable().countForHolder(userId)).resolves.toMatchObject({ total: 1 }); }); }); @@ -438,7 +485,9 @@ describe('the expiry sweep', () => { it('leaves a subscription with no expiry alone', async () => { await durable().create(input()); await expect(durable().sweepExpired(500)).resolves.toBe(0); - await expect(durable().countForHolder(userId)).resolves.toBe(1); + await expect(durable().countForHolder(userId)).resolves.toMatchObject({ + total: 1, + }); }); }); diff --git a/src/backend/stores/events/DurableSubscriptionStore.ts b/src/backend/stores/events/DurableSubscriptionStore.ts index a3635f0f8..1da6c0868 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.ts +++ b/src/backend/stores/events/DurableSubscriptionStore.ts @@ -18,7 +18,10 @@ */ import { randomUUID } from 'node:crypto'; -import { EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER } from '../../controllers/events/limits.js'; +import { + EVENTS_DURABLE_SUBSCRIPTIONS_MAX, + type SubscriptionQuota, +} from '../../controllers/events/limits.js'; import { HttpError } from '../../core/http/HttpError.js'; import type { AclMode } from '../../services/acl/ACLService.js'; import type { DeliveryClass } from '../../services/events/registry.js'; @@ -92,6 +95,19 @@ export interface DurableSubscriptionInput { context: string | null; permission: AclMode; expiresAt: number | null; + /** + * Plan-resolved caps this subscribe is held to. Omitted falls back to the + * structural maximum, so a writer that never resolved a plan still cannot + * leave an account holding more rows than the design allows. + */ + limits?: SubscriptionQuota; +} + +/** One keyset page of rows, and where the next page starts. */ +export interface DurablePage { + rows: DurableSubscription[]; + /** `null` once the scan is exhausted. */ + nextId: number | null; } export interface DurableListOptions { @@ -161,10 +177,12 @@ const workerNeedsApp = (): HttpError => { legacyCode: 'invalid_targets' }, ); -const quotaReached = (): HttpError => +const quotaReached = (limit: number, scope: 'account' | 'app'): HttpError => new HttpError( 429, - `An account may hold ${EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER} durable subscriptions`, + scope === 'app' + ? `An app may hold ${limit} durable subscriptions for one account` + : `An account may hold ${limit} durable subscriptions`, { legacyCode: 'events_subscription_limit' }, ); @@ -259,8 +277,18 @@ export class DurableSubscriptionStore extends PuterStore { ); this.#assertContext(input.context); - const held = await this.countForHolder(input.holderUserId); - if (held >= EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER) throw quotaReached(); + const perUser = Math.min( + input.limits?.perUser ?? EVENTS_DURABLE_SUBSCRIPTIONS_MAX, + EVENTS_DURABLE_SUBSCRIPTIONS_MAX, + ); + const perApp = input.limits?.perApp ?? perUser; + const held = await this.countForHolder( + input.holderUserId, + input.appUid, + ); + if (held.total >= perUser) throw quotaReached(perUser, 'account'); + if (input.appUid !== null && held.forApp >= perApp) + throw quotaReached(perApp, 'app'); const row: DurableSubscription = { durable: true, @@ -547,17 +575,25 @@ export class DurableSubscriptionStore extends PuterStore { } /** - * Quota counting, over the same index the listing uses. Primary: a count - * read off a lagging replica is a cap a burst of subscribes walks straight - * through. + * Quota counting, over the same index the listing uses: what the account + * holds, and how much of that is one app's. Both caps come off one read. + * Primary — a count read off a lagging replica is a cap a burst of + * subscribes walks straight through. */ - async countForHolder(holderUserId: number): Promise { + async countForHolder( + holderUserId: number, + appUid: string | null = null, + ): Promise<{ total: number; forApp: number }> { const [row] = await this.clients.db.pread( - `SELECT COUNT(*) AS \`total\` FROM \`${TABLE}\` ` + - `WHERE \`holder_user_id\` = ? AND ${this.#unexpiredClause()}`, - [holderUserId, nowSeconds()], + `SELECT COUNT(*) AS \`total\`, ` + + 'SUM(CASE WHEN `app_uid` = ? THEN 1 ELSE 0 END) AS `for_app` ' + + `FROM \`${TABLE}\` WHERE \`holder_user_id\` = ? AND ${this.#unexpiredClause()}`, + [appUid, holderUserId, nowSeconds()], ); - return Number(row?.total ?? 0); + return { + total: Number(row?.total ?? 0), + forApp: Number(row?.for_app ?? 0), + }; } /** @@ -584,7 +620,7 @@ export class DurableSubscriptionStore extends PuterStore { const rows = await this.clients.db.pread( `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` ` + `WHERE ${where.join(' AND ')} ORDER BY \`id\` LIMIT ?`, - [...params, EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER], + [...params, EVENTS_DURABLE_SUBSCRIPTIONS_MAX], ); return rows.map(toRow); } @@ -639,12 +675,31 @@ export class DurableSubscriptionStore extends PuterStore { holderUserId, reason, nowSeconds(), - EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER, + EVENTS_DURABLE_SUBSCRIPTIONS_MAX, ], ); return rows.map(toRow); } + /** + * Rows suspended for one reason across every holder, one keyset page at a + * time. The credit sweep is the one pass that has to see all of them rather + * than one account's, and it walks the primary key so a long scan never + * holds a position anything else waits on. + */ + async listSuspendedPage( + reason: SuspendedReason, + afterId: number, + batchSize: number, + ): Promise { + return this.#page( + ['`suspended_at` IS NOT NULL', '`suspended_reason` = ?'], + [reason], + afterId, + batchSize, + ); + } + /** * Every row a region has to be able to deliver for one owner. Read from the * primary: this is what a cold region caches, and caching a replica's "no @@ -664,6 +719,29 @@ export class DurableSubscriptionStore extends PuterStore { // -- Internals --------------------------------------------------- + /** One page of a whole-table scan, ordered and positioned by primary key. */ + async #page( + where: string[], + params: unknown[], + afterId: number, + batchSize: number, + ): Promise { + const limit = Math.max(1, Math.floor(batchSize)); + const rows = await this.clients.db.read( + `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` ` + + `WHERE ${[...where, '`id` > ?'].join(' AND ')} ` + + 'ORDER BY `id` LIMIT ?', + [...params, afterId, limit], + ); + return { + rows: rows.map(toRow), + nextId: + rows.length < limit + ? null + : Number(rows[rows.length - 1].id) || null, + }; + } + async #rebuildRegion(ownerUserId: number): Promise { const rows = await this.listDeliverableForOwner(ownerUserId); await this.stores.eventSubscription.rebuildDurable(ownerUserId, rows); diff --git a/src/backend/stores/events/PendingDeliveryStore.ts b/src/backend/stores/events/PendingDeliveryStore.ts index 16559fa3a..89fe5bf0b 100644 --- a/src/backend/stores/events/PendingDeliveryStore.ts +++ b/src/backend/stores/events/PendingDeliveryStore.ts @@ -44,7 +44,7 @@ import { PuterStore } from '../types.js'; * all deleted the moment the last one settles — a subscription that is keeping * up owns nothing: * - * ev:q:{} HASH entryId -> the delivery and its attempt counts + * ev:q:{} HASH entryId -> the delivery, its attempt counts and billed flag * ev:qp:{} ZSET entryId -> enqueued at; membership means unsettled * ev:ql:{} ZSET entryId -> lease expiry; membership means in flight * ev:qf:{} STR handler failures in a row, expiring on its own @@ -201,6 +201,8 @@ interface StoredEntry { socketAttempts: number; /** Handler attempts spent, which is what the retry wait is derived from. */ handlerAttempts?: number; + /** Set once this entry has been charged for, however many attempts follow. */ + billed?: boolean; } const parseEntry = (raw: string | null): StoredEntry | null => { @@ -408,6 +410,26 @@ export class PendingDeliveryStore extends PuterStore { return true; } + /** + * Claim this entry's one bill, if nothing already has. A `single` retries + * the same entry across a lease expiry — a second socket attempt, then the + * handler — and every one of those is the same undelivered event, not a new + * one: only the attempt that gets here first is charged for it. + */ + async markBilled(subId: string, entryId: string): Promise { + const entry = parseEntry( + await this.clients.redis.hget(entriesKey(subId), entryId), + ); + if (!entry || entry.billed) return false; + + await this.clients.redis.hset( + entriesKey(subId), + entryId, + JSON.stringify({ ...entry, billed: true }), + ); + return true; + } + /** * Count one failure against the subscription and answer how many are in a * row. Region-local like the rest of the delivery state: the lease that diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md index a9acc98c8..b978fd80a 100644 --- a/src/docs/src/rate-limits-and-quotas.md +++ b/src/docs/src/rate-limits-and-quotas.md @@ -159,15 +159,25 @@ Over these, **the share still succeeds** — only the announcement is dropped. T One write can reach many subscriptions, so events are bounded on both halves: how much you may register, and how much any one event may turn into. +Durable subscriptions are the ones that outlive a connection, so they are the half that varies by plan: + +| Limit | Paid | Free | Anonymous | +| ----------------------------------------- | ---- | ---- | --------- | +| Durable subscriptions per account | 500 | 100 | — | +| Durable subscriptions per app, per account | 100 | 25 | — | + +A temporary (anonymous) account cannot create durable subscriptions at all — `subscribe` fails with `events_durable_requires_account`, and session subscriptions, which live and die with the connection, are the surface it has. Past either cap the call fails with `events_subscription_limit`; unsubscribing frees a slot immediately. + | Limit | All accounts | | -------------------------------------------- | ------------ | | Subscriptions per connection | 50 | -| Durable subscriptions per account | 500 | | `subscribe` / `unsubscribe` calls per minute | 60 | | Subscription listings per minute | 120 | | Matched subscriptions per event | 50 | | Filter evaluations per event | 200 | -| Deliveries per minute, per subscription | 600 | +| Broadcast deliveries per minute, per subscription | 600 | +| `single` deliveries per minute, per subscription | 120 | +| Handler invocations per minute, per app | 60 | | Acknowledgements per minute | 600 | | Undelivered deliveries per subscription | 10,000 | | Undelivered deliveries per *suspended* subscription | 100 | @@ -184,7 +194,7 @@ One write can reach many subscriptions, so events are bounded on both halves: ho Subscriptions come in two kinds. A **session** subscription lives with the connection that made it: it is dropped when the connection closes, and a reconnecting client subscribes again. A **durable** subscription outlives every connection — it is created over the API, listed and revoked from the account, and keeps delivering until you remove it or it expires. -The 51st subscription on one connection, and the 501st durable subscription on one account, both fail with `events_subscription_limit`. Over the call budget, `subscribe` and `unsubscribe` fail with `too_many_requests`. Subscribing to something you cannot read fails with `subject_does_not_exist` — the same answer as subscribing to something that is not there, so the call cannot be used to find out which. +The 51st subscription on one connection, and the durable subscription past your plan's cap, both fail with `events_subscription_limit`. Over the call budget, `subscribe` and `unsubscribe` fail with `too_many_requests`. Subscribing to something you cannot read fails with `subject_does_not_exist` — the same answer as subscribing to something that is not there, so the call cannot be used to find out which. A durable subscription may carry a `context`: JSON that is stored with it and handed to its handler on every delivery, capped at a hard **4 KB** and rejected over that with `events_context_too_large` — client-side, before the request. It is stored in plaintext and read only on the delivery path; listings return its **key names and a content hash**, never its values. For anything larger, store it in a file and put the path in `context`. An app sees and revokes only the subscriptions it created; a session acting for the account sees them all, including ones left behind by an app that has since been removed. @@ -215,6 +225,23 @@ A **background delivery** — one that runs your app's handler with nobody there 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. +Both per-minute delivery budgets are spent per subscription and answered with a `gap` marker carrying `reason: 'delivery_rate_limit'` rather than an error. The handler budget is different: a delivery that arrives when its app has spent the minute's invocations is **not** failed and does not count as a handler failure — it stays owed and goes out on a later attempt. + +#### What events cost + +Deliveries are metered to the **subscription's holder** — your data, your subscriptions, your bill. A subscription that sits idle costs nothing; the plan quotas above are what bound how many you can hold. + +| Line | Rate | Counted per | +| --------------------------- | -------------- | ------------------------------- | +| `events:delivery:broadcast` | 10 µ¢ | delivered event | +| `events:delivery:single` | 100 µ¢ | delivered event | + +A `single` costs more because it is leased and acknowledged; a broadcast copy is a socket write. Handler runs bill separately through the usual worker path. + +Only deliveries that actually happen are billed. An event a filter excluded, several writes the 250 ms window collapsed into one, a delivery a permission re-check stopped, and every `gap` marker are all free — a marker says something was lost, and charging for the loss would be charging you twice. Session subscriptions are billed at the broadcast rate like any other. + +Deliveries stop when the holder's balance runs out: the subscription is suspended with `suspendedReason: 'no_credit'`, the holder is notified, and nothing further is metered against it. What it was owed is held for **1 hour**. Restoring the balance resumes it — checked periodically rather than the instant a payment lands, so allow a few minutes after topping up. + ### Peer connections | Limit | Paid | Free | Anonymous |