diff --git a/src/backend/clients/event/types.ts b/src/backend/clients/event/types.ts index 9f0540f26..7cdf0e5d0 100644 --- a/src/backend/clients/event/types.ts +++ b/src/backend/clients/event/types.ts @@ -516,6 +516,23 @@ export type EventMap = { */ 'auth.sessions.revoked': { user_id: number; session_uids: string[] }; + /** + * A grant was withdrawn, so whatever was standing on it has to be settled + * rather than left to fail its next check. + * + * Local rather than `outer.*` on purpose: a listener settles shared state, + * and one region doing it covers every other. Emitted once per revoke call, + * not once per row it affects. + */ + 'permission.revoked': { + /** Whose access went. */ + holderUserId: number; + /** The app the grant was to, or `null` for a user-to-user grant. */ + appUid: string | null; + /** The grant string, or `null` when every grant to the app went. */ + permission: string | null; + }; + // ---- Extension hooks / misc ---- 'puter.gui.addons': { prependHeadContent?: string[]; diff --git a/src/backend/controllers/events/limits.ts b/src/backend/controllers/events/limits.ts index 7ba803d3e..cba4f0d1a 100644 --- a/src/backend/controllers/events/limits.ts +++ b/src/backend/controllers/events/limits.ts @@ -60,6 +60,15 @@ export const EVENTS_SESSION_SUBSCRIPTIONS_PER_SOCKET = 50; */ export const EVENTS_DURABLE_SUBSCRIPTIONS_PER_USER = 500; +/** + * How long a suspended durable subscription is kept before it is deleted. + * + * A suspension caused by a revoked grant never resumes — consent to watch is + * re-established by subscribing again — so the row survives only long enough + * for its holder to see, in `list`, that it stopped and why. + */ +export const SUSPENDED_ROW_TTL_DAYS = 30; + /** * Subscribe + unsubscribe calls per minute, per user. * diff --git a/src/backend/services/acl/ACLService.ts b/src/backend/services/acl/ACLService.ts index d78401d49..87031b23b 100644 --- a/src/backend/services/acl/ACLService.ts +++ b/src/backend/services/acl/ACLService.ts @@ -46,7 +46,11 @@ export interface ResourceDescriptor { } export type AclMode = - 'see' | 'list' | 'read' | 'write' | typeof MANAGE_PERM_PREFIX; + | 'see' + | 'list' + | 'read' + | 'write' + | typeof MANAGE_PERM_PREFIX; /** Duck-typed error shape compatible with APIError consumers (fsv2). */ export interface AclError { @@ -180,7 +184,7 @@ export class ACLService extends PuterService { if (actor.accessToken.fullAccess) return true; for (const ancestor of ancestors) { - for (const permission of this.#permissionsFor( + for (const permission of this.permissionsFor( ancestor.uid, mode, )) { @@ -220,7 +224,7 @@ export class ACLService extends PuterService { for (const ancestor of ancestors) { const reading = await this.services.permission.scan( actor, - this.#permissionsFor(ancestor.uid, mode), + this.permissionsFor(ancestor.uid, mode), ); const options = PermissionUtil.readingToOptions(reading); if (options.length > 0) return true; @@ -232,8 +236,11 @@ export class ACLService extends PuterService { /** * Permissions on `uid` that satisfy `mode`. `manage` sits above the whole * family — it answers any mode, but nothing answers it. + * + * Public because it also answers the reverse question: given a grant that + * was just withdrawn, which stored checks did it hold up? */ - #permissionsFor(uid: string, mode: AclMode): string[] { + permissionsFor(uid: string, mode: AclMode): string[] { const manage = PermissionUtil.join(MANAGE_PERM_PREFIX, 'fs', uid); if (mode === MANAGE_PERM_PREFIX) return [manage]; return [ diff --git a/src/backend/services/events/EventsService.test.ts b/src/backend/services/events/EventsService.test.ts index f1214b32a..1e5c518c9 100644 --- a/src/backend/services/events/EventsService.test.ts +++ b/src/backend/services/events/EventsService.test.ts @@ -177,6 +177,17 @@ const appStore = { getByUid: async (uid: string) => ({ uid, id: 1 }), }; +/** + * The counter delivery decisions are cached under. Held here so a test can move + * it and watch the re-check happen again; what bumps it in production is any + * grant or revoke. + */ +let permissionGeneration: number; + +const permissionStore = { + getCacheGeneration: async () => permissionGeneration, +}; + /** * This suite is about session rows and what a dispatch spends on them, so the * region is already warm and holds no durable rows — the table and its cache @@ -214,6 +225,7 @@ const buildService = ( fsEntry: fsEntryStore, user: userStore, app: appStore, + permission: permissionStore, } as never, { socket: { @@ -322,6 +334,7 @@ beforeEach(() => { commands = []; entries = new Map(); denied = new Map(); + permissionGeneration = 1; redis = countingRedis(new MockRedis.Cluster(['redis://localhost:7001'])); store = new EventSubscriptionStore( {} as IConfig, @@ -895,6 +908,33 @@ describe('matching', () => { expect(sent).toEqual([]); }); + it('holds its answer across events, and re-asks when the generation moves', async () => { + const { documents, file } = seedTree(); + await subscribe(`fs:${documents.uid}`); + + await dispatch(file); + await flush(); + expect(sent).toHaveLength(1); + + // Access changing with nothing to announce it leaves the cached answer + // standing, which is the trade a generation-keyed cache makes. + denied.set(file.path, 'hidden'); + sent.length = 0; + await dispatch(file); + await flush(); + expect(sent).toHaveLength(1); + + // Any grant or revoke moves the counter, and the question is asked + // again — with nothing delivered and nothing metered when it now fails. + permissionGeneration++; + sent.length = 0; + delivered.length = 0; + await dispatch(file); + await flush(); + expect(sent).toEqual([]); + expect(delivered).toEqual([]); + }); + it('re-checks the node the event is about, not the anchor', async () => { const { documents } = seedTree(); await subscribe(`fs:/u${userId}/Documents/**`); diff --git a/src/backend/services/events/EventsService.ts b/src/backend/services/events/EventsService.ts index 6abba55aa..23ed51bbf 100644 --- a/src/backend/services/events/EventsService.ts +++ b/src/backend/services/events/EventsService.ts @@ -25,10 +25,12 @@ import { EVENTS_COALESCE_WINDOW_MS, EVENTS_MATCHED_SUBSCRIPTIONS_PER_EVENT, EVENTS_SUBSCRIBE_LIMIT, + SUSPENDED_ROW_TTL_DAYS, } from '../../controllers/events/limits.js'; import type { Actor } from '../../core/actor.js'; import { HttpError } from '../../core/http/HttpError.js'; import { checkRateLimit } from '../../core/http/middleware/rateLimit.js'; +import type { ReanchorInput } from '../../stores/events/DurableSubscriptionStore.js'; import { SESSION_SUBSCRIPTION_TTL_SECONDS, type DispatchSubscription, @@ -60,11 +62,14 @@ import { resolveFsAnchor, type FsAnchorDeps } from './anchors.js'; import { assertSubscribeAuthorized, checkDeliveryAuthorized, + deliveryGenerationTag, nodeDescriptor, + resolveGrantActor, rowInActorScope, type EventAclDeps, } from './authorization.js'; import { DeliveryCoalescer } from './coalescer.js'; +import { DeliveryAuthCache } from './deliveryAuthCache.js'; import { FILTER_EVALUATIONS_PER_EVENT, compileMatch, @@ -83,7 +88,7 @@ import { type PublicSubject, } from './registry.js'; import { SubscriptionCache } from './subscriptionCache.js'; -import { parseSubject, type FsOp } from './subjects.js'; +import { fsAnchorToken, parseSubject, type FsOp } from './subjects.js'; import { RecordingWorkerInvoker, type WorkerInvocation, @@ -197,6 +202,27 @@ interface AddressedDelivery { worker?: WorkerInvocation; } +/** + * Why a subscription ended without its holder unsubscribing. Both are terminal: + * the grant is gone, or the node is, and neither comes back by itself. + */ +export type SubscriptionEndReason = 'permission_revoked' | 'anchor_deleted'; + +/** Who a stored row acts as, and what its cached decisions hang on. */ +interface GrantIdentity { + actor: Actor; + generation: string; +} + +/** A withdrawn grant, as the settle pass reads it off the bus. */ +export interface RevokedGrant { + holderUserId: number; + /** The app the grant was to, or `null` for a user-to-user grant. */ + appUid: string | null; + /** The grant string, or `null` when every grant to the app went. */ + permission: string | null; +} + /** What a dispatch call site can supply that the event itself does not carry. */ export interface FsDispatchOptions { /** @@ -228,6 +254,11 @@ const EXPIRY_SWEEP_INITIAL_DELAY_MS = 5 * 60 * 1000; const EXPIRY_BATCH_SIZE = 500; /** Batches one sweep takes, so a large backlog drains over several passes. */ const EXPIRY_MAX_BATCHES = 50; +/** + * Subjects one "subscriptions ended" notification names before it stops + * listing. + */ +const ENDED_SUBJECTS_LISTED = 20; // -- Owed deliveries -------------------------------------------------- @@ -462,6 +493,7 @@ const parseExpiresAt = (value: unknown): number | null => { export class EventsService extends PuterService { readonly #cache = new SubscriptionCache(); + readonly #deliveryAuth = new DeliveryAuthCache(); readonly #compiled = new Map(); readonly #lookups = new Map>(); readonly #refreshTimers = new Map>(); @@ -494,6 +526,16 @@ export class EventsService extends PuterService { this.invalidateUser(userId, { rebuild: durable === true }); }, ); + + // The delivery re-check would already deny these, but a row nobody can + // deliver to still holds an anchor slot, still bills, and still costs a + // filter evaluation on every event under it. + this.clients.event.on('permission.revoked', (_key, data) => { + void this.settleRevokedGrant(data as RevokedGrant).catch((err) => { + console.warn('[events] revocation settle failed', err); + }); + }); + this.#armExpirySweep(); this.#armPendingSweep(); } @@ -823,12 +865,32 @@ export class EventsService extends PuterService { */ async sweepExpired(): Promise { if (!this.enabled) return 0; + return this.#sweepInBatches((batchSize) => + this.stores.durableSubscription.sweepExpired(batchSize), + ); + } + + /** + * Drop rows that have been suspended longer than they are worth keeping. A + * revoked subscription never resumes, so the row survives only as the + * answer its holder gets from `list` when they ask why it stopped. + */ + async sweepSuspended(): Promise { + if (!this.enabled) return 0; + const cutoff = + Math.floor(Date.now() / 1000) - + SUSPENDED_ROW_TTL_DAYS * 24 * 60 * 60; + return this.#sweepInBatches((batchSize) => + this.stores.durableSubscription.sweepSuspended(cutoff, batchSize), + ); + } + + async #sweepInBatches( + pass: (batchSize: number) => Promise, + ): Promise { let removed = 0; - for (let pass = 0; pass < EXPIRY_MAX_BATCHES; pass++) { - const batch = - await this.stores.durableSubscription.sweepExpired( - EXPIRY_BATCH_SIZE, - ); + for (let i = 0; i < EXPIRY_MAX_BATCHES; i++) { + const batch = await pass(EXPIRY_BATCH_SIZE); removed += batch; if (batch < EXPIRY_BATCH_SIZE) break; } @@ -872,7 +934,12 @@ export class EventsService extends PuterService { // 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); + // Except a revoked row's, which names paths its holder may + // no longer see: anything a dispatch in flight queued after + // the settle's own purge goes now, not at the reap. + if (row.suspendedReason === 'permission_revoked') + await this.stores.pendingDelivery.purge(subId); + else await this.stores.pendingDelivery.defer(subId); continue; } attempted += await this.#drain(row); @@ -1009,6 +1076,12 @@ export class EventsService extends PuterService { if (rows.length === 0) return; await this.#route(subject, context, rows, options.actingUserId); + + // Only a removal can invalidate an anchor, so nothing else pays for + // this — and this pass is already holding the rows that key on the uid + // now gone. + if (key === 'fs.remove.node') + await this.#settleDeletedAnchor(context, rows); } async #route( @@ -1101,8 +1174,12 @@ export class EventsService extends PuterService { * subscription was made must stop delivering at once rather than when the * row is next touched. * - * Last of the filters, because it is the only one that can cost a lookup — - * and rows that share an identity and a grant share one decision. + * Last of the filters, because it is the only one that can cost a lookup. + * Two layers keep that lookup rare: the cross-event cache, keyed by the + * permission cache's own generation, answers a subscription being written + * to repeatedly without asking anything; and within one event, rows sharing + * an identity and a grant share one decision — which is what a fan-out over + * one folder is. */ async #stillAuthorized( rows: DispatchSubscription[], @@ -1111,25 +1188,81 @@ export class EventsService extends PuterService { if (rows.length === 0) return rows; const node = this.#eventDescriptor(context); + const identities = new Map>(); const decisions = new Map>(); const allowed = await Promise.all( - rows.map((row) => { - const key = `${row.holderUserId}|${row.appUid ?? ''}|${row.permission}`; - let decision = decisions.get(key); - if (!decision) { - decision = checkDeliveryAuthorized( - row, - node, - this.#aclDeps(), - ); - decisions.set(key, decision); - } - return decision; - }), + rows.map((row) => + this.#recheck(row, node, context.entry.uid, { + identities, + decisions, + }), + ), ); return rows.filter((_row, i) => allowed[i]); } + async #recheck( + row: DispatchSubscription, + node: ResourceDescriptor, + nodeUid: string, + memo: { + identities: Map>; + decisions: Map>; + }, + ): Promise { + const identityKey = `${row.holderUserId}|${row.appUid ?? ''}`; + let identity = memo.identities.get(identityKey); + if (!identity) { + identity = this.#grantIdentity(row); + memo.identities.set(identityKey, identity); + } + // An identity that cannot be resolved — a deleted app, a deleted user — + // is one nothing may be delivered to. + const resolved = await identity; + if (!resolved) return false; + + const key = { + subId: row.subId, + generation: resolved.generation, + nodeUid, + }; + const cached = this.#deliveryAuth.read(key); + if (cached !== null) return cached; + + const decisionKey = `${identityKey}|${row.permission}`; + let decision = memo.decisions.get(decisionKey); + if (!decision) { + decision = checkDeliveryAuthorized( + resolved.actor, + row.permission, + node, + this.#aclDeps(), + ); + memo.decisions.set(decisionKey, decision); + } + + const allowed = await decision; + this.#deliveryAuth.write(key, allowed); + return allowed; + } + + /** Who a row acts as, and the counter its answers are keyed by. */ + async #grantIdentity( + row: DispatchSubscription, + ): Promise { + try { + const deps = this.#aclDeps(); + const actor = await resolveGrantActor(row, deps); + if (!actor) return null; + return { + actor, + generation: await deliveryGenerationTag(actor, deps), + }; + } catch { + return null; + } + } + /** The event's node as ACL wants it, reusing the walk dispatch already did. */ #eventDescriptor(context: EventContext): ResourceDescriptor { return nodeDescriptor( @@ -1178,6 +1311,330 @@ export class EventsService extends PuterService { } } + // -- Settling ---------------------------------------------------- + + /** + * Take out of service every durable subscription a withdrawn grant was + * holding up. One holder-index read per revocation, not one per row. + * + * The delivery re-check already refuses these, so this is not what makes a + * revocation safe — it is what keeps a revoked subscription from going on + * costing its holder an anchor slot, a daily line and a filter evaluation + * per event forever. Re-granting does not bring one back: the consent to + * watch is re-established by subscribing again. + * + * Session rows are left to the re-check: finding them means knowing which + * connections a holder has, which is a keyspace scan, and one that ends + * with the connection anyway. + */ + async settleRevokedGrant(revocation: RevokedGrant): Promise { + if (!this.enabled) return 0; + + const held = await this.stores.durableSubscription.listActiveForHolder( + revocation.holderUserId, + revocation.appUid, + ); + // Withdrawing an app's access wholesale is the user saying the app is + // done, so it takes everything the app holds — no per-row question, + // and none of the standing exemptions an app enjoys over its own data + // keep a background subscription alive past the consent that made it. + const settling = + revocation.permission === null + ? held + : await this.#leftUnauthorized(held, revocation.permission); + if (settling.length === 0) return 0; + + // Only the rows this pass was the one to suspend are its to purge and + // announce: an unshare withdraws several grant strings in a row, and + // every one of them runs this settle. + const { suspended, bumps } = + await this.stores.durableSubscription.suspend( + settling, + 'permission_revoked', + ); + for (const row of suspended) { + // Unlike every other suspension, this backlog goes at once: it + // holds the paths of a resource its holder has just lost the right + // to see, and keeping it for a resume that by design never comes + // turns a revocation into a delayed disclosure. + await this.stores.pendingDelivery.purge(row.subId).catch(() => {}); + // Takes the coalesced deliveries with it: one still in flight names + // exactly what its holder has stopped being allowed to see. + this.#forget(row.subId); + } + for (const bump of bumps) this.#publishGeneration(bump, true); + await this.#notifyEnded(suspended, 'permission_revoked'); + return suspended.length; + } + + /** + * Of a holder's rows, the ones a named withdrawn grant has actually left + * without access. + * + * Two steps, and both are needed. The first is implication-aware rather + * than string equality: a row's anchor and stored mode compose the + * permission its subscribe check ran against, and `list` is answered by + * `read`, `write` and `manage`, so withdrawing any of them puts the row in + * question. That only narrows, though — the second step asks the access + * question for real, because a share whose mode merely _changed_ is + * recorded as a grant followed by a revoke, and settling on the revoke + * alone would end a subscription whose reach just got wider. + * + * A grant withdrawn on an _ancestor_ of an anchor is not narrowed to here — + * the row names its own node, not the chain above it. Those are left to the + * delivery re-check, which stops them immediately and permanently. + */ + async #leftUnauthorized( + rows: readonly DurableSubscription[], + permission: string, + ): Promise { + const settling: DurableSubscription[] = []; + for (const row of rows) { + const covered = this.services.acl + .permissionsFor(row.anchorUid, row.permission) + .includes(permission); + if (covered && !(await this.#anchorStillReachable(row))) + settling.push(row); + } + return settling; + } + + /** Whether a row's holder can still reach its anchor, asked fresh. */ + #anchorStillReachable(row: DurableSubscription): Promise { + return this.#reachable(row, row); + } + + /** + * Whether a row's holder may watch from `at`, under the mode it subscribed + * with. + */ + async #reachable( + row: DispatchSubscription, + at: { anchorUid: string; anchorPath: string }, + ): Promise { + const deps = this.#aclDeps(); + const actor = await resolveGrantActor(row, deps); + if (!actor) return false; + return checkDeliveryAuthorized( + actor, + row.permission, + nodeDescriptor({ uid: at.anchorUid, path: at.anchorPath }, deps), + deps, + ); + } + + /** + * Move a path-form row up to the nearest surviving ancestor its holder may + * still watch, or end it. Asked again after each move: a recursive delete + * works from the leaves up, so the level just moved to may be gone by the + * time the row is written there — and its own removal pass ran before the + * row was visible on it. + */ + async #carryForward( + row: DispatchSubscription, + ancestors: ReadonlyArray<{ uid: string; path: string }>, + ): Promise { + let current = row; + for (let hop = 0; hop <= ancestors.length; hop++) { + const next = await this.#nextAnchor(current, ancestors); + // Climbing must not land a row where its holder could never have + // subscribed: the re-check would deny every delivery, but the row + // would still hold an anchor slot and a filter evaluation there. + if (!next || !(await this.#reachable(current, next))) { + await this.#endSubscription(current, 'anchor_deleted'); + return; + } + await this.#reanchor(current, next); + if (await resolveNode(this.stores.fsEntry, { uid: next.anchorUid })) + return; + current = { + ...current, + token: next.token, + anchorUid: next.anchorUid, + anchorPath: next.anchorPath, + match: next.match, + ownerUserId: next.ownerUserId, + }; + } + } + + /** + * What the removal of an anchor node does to the rows keyed on it. Runs in + * the pass that delivered the `remove`, because nothing will ever come + * looking for them again: the uid they key on is gone. + * + * A row carrying a match asked about a _path_, so it follows that path up + * to whatever still exists, its match rewritten to lead with the segments + * that went. A row with no match asked about that node, whose uid is never + * coming back, so it ends. + */ + async #settleDeletedAnchor( + context: EventContext, + candidates: readonly DispatchSubscription[], + ): Promise { + const onAnchor = candidates.filter( + (row) => row.anchorUid === context.entry.uid, + ); + if (onAnchor.length === 0) return; + + for (const row of onAnchor) { + try { + if (row.match) await this.#carryForward(row, context.ancestors); + else await this.#endSubscription(row, 'anchor_deleted'); + } catch (err) { + console.warn( + '[events] could not settle a subscription whose anchor was removed', + row.subId, + err, + ); + } + } + } + + /** + * Where a path-form row moves to, or `null` when there is nowhere to move + * it: nothing left above it, or a rewritten pattern past what may be + * compiled. + * + * The chain holds existing ancestors only, deepest first, so its head is + * already the nearest survivor however many levels a recursive delete took + * at once. It is still walked rather than indexed, because a delete works + * from the leaves up and the level above may have gone since the walk. Root + * is where climbing stops on its own — its uid never changes. + */ + async #nextAnchor( + row: DispatchSubscription, + ancestors: ReadonlyArray<{ uid: string; path: string }>, + ): Promise { + for (const survivor of ancestors) { + const climbed = relativeTo(survivor.path, row.anchorPath); + if (climbed === null) continue; + + const match = climbed + ? `${climbed}/${row.match}` + : String(row.match); + try { + compileMatch(match); + } catch { + return null; + } + + // The keyspace a row is indexed in is the anchor owner's, so a + // climb that crosses an ownership boundary moves with it. + const entry = await resolveNode(this.stores.fsEntry, { + uid: survivor.uid, + }); + if (!entry) continue; + + return { + token: fsAnchorToken(survivor.uid), + anchorUid: survivor.uid, + anchorPath: survivor.path, + match, + ownerUserId: entry.userId, + }; + } + return null; + } + + async #reanchor( + row: DispatchSubscription, + next: ReanchorInput, + ): Promise { + const bumps = + row.durable === true + ? ( + await this.stores.durableSubscription.reanchor( + row as DurableSubscription, + next, + ) + ).bumps + : await this.stores.eventSubscription.reanchorSession( + row as SessionSubscription, + { ...(row as SessionSubscription), ...next }, + ); + // The matcher cache keys on the pattern it compiled, so it corrects + // itself; the access decisions were about a node this row no longer + // watches. + this.#deliveryAuth.forget(row.subId); + for (const bump of bumps) + this.#publishGeneration(bump, row.durable === true); + } + + /** + * End one subscription that cannot be carried forward. Anything already + * coalesced for it is deliberately left alone — the final `remove` is the + * last thing it is owed, and cancelling it here would be the delivery this + * whole pass exists to make. + */ + async #endSubscription( + row: DispatchSubscription, + reason: SubscriptionEndReason, + ): Promise { + this.#compiled.delete(row.subId); + this.#deliveryAuth.forget(row.subId); + + if (row.durable !== true) { + const bump = await this.stores.eventSubscription.remove( + row as SessionSubscription, + ); + this.#publishGeneration(bump, false); + return; + } + + const durable = row as DurableSubscription; + const bump = await this.stores.durableSubscription.remove(durable); + // Nothing can drain a stream whose row is gone, so what is still owed + // goes with it — the same trade an explicit unsubscribe makes. + await this.stores.pendingDelivery.purge(durable.subId).catch(() => {}); + this.#publishGeneration(bump, true); + await this.#notifyEnded([durable], reason); + } + + /** + * Tell durable holders their subscriptions are over. They did not ask for + * this, and silence would read as "still watching" — so it goes to the + * holder, not the app's developer. One notification per holder and app: + * withdrawing an app's access ends everything it held at once, and that is + * one piece of news, not one per row. + */ + async #notifyEnded( + rows: readonly DurableSubscription[], + reason: SubscriptionEndReason, + ): Promise { + const groups = new Map(); + for (const row of rows) { + const key = `${row.holderUserId}|${row.appUid ?? ''}`; + groups.set(key, [...(groups.get(key) ?? []), row]); + } + for (const group of groups.values()) { + const [first] = group; + try { + await this.services.notification.notify( + [first.holderUserId], + { + title: + group.length === 1 + ? 'A subscription ended' + : `${group.length} subscriptions ended`, + subject: first.subject, + subjects: group + .slice(0, ENDED_SUBJECTS_LISTED) + .map((row) => row.subject), + count: group.length, + reason, + }, + { type: 'app.events.ended', appUid: first.appUid }, + ); + } catch (err) { + console.warn( + '[events] could not report a subscription ending', + err, + ); + } + } + } + // -- Owed deliveries --------------------------------------------- /** @@ -1529,6 +1986,7 @@ export class EventsService extends PuterService { #forget(subId: string): void { this.#compiled.delete(subId); + this.#deliveryAuth.forget(subId); this.#coalesce().cancel((key) => key.startsWith(`${subId}|`)); } @@ -1545,6 +2003,8 @@ export class EventsService extends PuterService { getAncestorChain: (path) => this.services.fs.getAncestorChain(path), getUser: (userId) => this.stores.user.getById(userId), getApp: (uid) => this.stores.app.getByUid(uid), + getCacheGeneration: (uid) => + this.stores.permission.getCacheGeneration(uid), }; } @@ -1580,9 +2040,11 @@ export class EventsService extends PuterService { #armExpirySweep(): void { if (!this.enabled) return; const run = () => { - void this.sweepExpired().catch((err) => { - console.warn('[events] expiry sweep failed', err); - }); + void this.sweepExpired() + .then(() => this.sweepSuspended()) + .catch((err) => { + console.warn('[events] expiry sweep failed', err); + }); }; // Jittered so a deploy does not have every node sweep at once. const kick = setTimeout( diff --git a/src/backend/services/events/anchorSettle.integration.test.ts b/src/backend/services/events/anchorSettle.integration.test.ts new file mode 100644 index 000000000..5c5bd5ac5 --- /dev/null +++ b/src/backend/services/events/anchorSettle.integration.test.ts @@ -0,0 +1,322 @@ +/* + * 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 . + */ + +/** + * A subscription keys on a uid, and deleting the node it keys on is the one + * thing that takes that uid away for good. What happens next depends on what + * the subscriber asked for: a path-form subscription follows the path up to + * whatever still exists and keeps watching, while a node-form one is over, + * because the node it named is never coming back. + */ + +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; +import { EVENTS_COALESCE_WINDOW_MS } from '../../controllers/events/limits.js'; +import { makeActor, type Actor } from '../../core/actor.js'; +import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js'; +import type { IConfig } from '../../types.js'; +import type { DeliveryEnvelope } from './EventsService.js'; + +const BOOT_TIMEOUT_MS = 120_000; + +let env: PuterTestEnv; +let user: { actor: Actor; username: string; id: number }; +let home: string; +let delivered: DeliveryEnvelope[]; + +const events = () => env.server.services.events; +const fs = () => env.server.services.fs; + +const folder = async (path: string): Promise => { + await fs().mkdir(user.id, { path, createMissingParents: true }); + return path; +}; + +const entryAt = (path: string) => + env.server.stores.fsEntry.getEntryByPath(path); + +const removeAt = async (path: string): Promise => { + const entry = await entryAt(path); + await fs().remove(user.id, { entry: entry!, recursive: true }); +}; + +const subscribe = async (socketId: string, subject: string) => + (await events().subscribe(user.actor, socketId, { subject })).sub; + +const held = (socketId: string) => + events().listSubscriptions(user.actor, socketId); + +const settled = (count = 1) => + vi.waitFor(() => expect(delivered.length).toBeGreaterThanOrEqual(count), { + timeout: EVENTS_COALESCE_WINDOW_MS * 12, + interval: 25, + }); + +const quiet = () => + new Promise((resolve) => + setTimeout(resolve, EVENTS_COALESCE_WINDOW_MS * 3), + ); + +/** Let every window in flight close, then start counting from nothing. */ +const drain = async (): Promise => { + await quiet(); + delivered.length = 0; +}; + +/** Wait for the settle the removal kicked off in its dispatch pass. */ +const anchoredAt = (socketId: string, subId: string, path: string) => + vi.waitFor( + async () => { + const row = (await held(socketId)).find( + (sub) => sub.subId === subId, + ); + expect(row?.anchor.path).toBe(path); + return row!; + }, + { timeout: 5_000, interval: 25 }, + ); + +const gone = (subId: string) => + vi.waitFor( + async () => + expect( + await env.server.stores.durableSubscription.getBySubId(subId), + ).toBeNull(), + { timeout: 5_000, interval: 25 }, + ); + +beforeAll(async () => { + env = await setupPuterTestEnv({ events: { enabled: true } } as IConfig); + const row = await env.server.stores.user.getByUsername( + env.users.user.username, + ); + user = { + actor: makeActor({ user: row as never }), + username: env.users.user.username, + id: row!.id, + }; + home = `/${user.username}`; + + delivered = []; + events().onDelivered = (envelope) => delivered.push(envelope); +}, BOOT_TIMEOUT_MS); + +afterAll(async () => { + await env?.shutdown(); +}); + +describe('a path-form subscription whose anchor is deleted', () => { + it('survives the folder it was waiting inside being deleted and recreated', async () => { + const docs = await folder(`${home}/reanchor-docs`); + const sub = await subscribe('sock-path', `fs:${docs}/trigger:add`); + expect(sub.anchor.path).toBe(docs); + expect(sub.match).toBe('trigger'); + + await removeAt(docs); + const moved = await anchoredAt('sock-path', sub.subId, home); + // The segments that went lead the pattern now, so it means the same + // thing measured from further up. + expect(moved.match).toBe('reanchor-docs/trigger'); + + await folder(docs); + await drain(); + await fs().touch(user.id, { path: `${docs}/trigger` }); + await settled(); + + expect(delivered.map((d) => d.subId)).toEqual([sub.subId]); + }); + + it('climbs another level when the level it moved to is deleted too', async () => { + const outer = `${home}/reanchor-outer`; + const inner = await folder(`${outer}/inner`); + const sub = await subscribe('sock-climb', `fs:${inner}/trigger:add`); + expect(sub.anchor.path).toBe(inner); + + await removeAt(inner); + expect( + (await anchoredAt('sock-climb', sub.subId, outer)).match, + ).toBe('inner/trigger'); + + await removeAt(outer); + expect((await anchoredAt('sock-climb', sub.subId, home)).match).toBe( + 'reanchor-outer/inner/trigger', + ); + + await folder(inner); + await drain(); + await fs().touch(user.id, { path: `${inner}/trigger` }); + await settled(); + + expect(delivered.map((d) => d.subId)).toEqual([sub.subId]); + }); + + it('moves a durable row, its cache entry and its stored anchor together', async () => { + const docs = await folder(`${home}/reanchor-durable`); + const sub = ( + await events().subscribeDurable(user.actor, { + subject: `fs:${docs}/**`, + }) + ).sub; + + await removeAt(docs); + await vi.waitFor( + async () => { + const row = + await env.server.stores.durableSubscription.getBySubId( + sub.subId, + ); + expect(row?.anchorPath).toBe(home); + expect(row?.match).toBe('reanchor-durable/**'); + }, + { timeout: 5_000, interval: 25 }, + ); + + await folder(docs); + await drain(); + await fs().touch(user.id, { path: `${docs}/after.txt` }); + await settled(); + + expect(delivered.map((d) => d.subId)).toEqual([sub.subId]); + }); +}); + +describe('a node-form subscription whose anchor is deleted', () => { + it('ends, while a path-form sibling on the same node re-anchors', async () => { + const dir = await folder(`${home}/node-form`); + const nodeForm = await subscribe('sock-node', `fs:${dir}`); + const pathForm = await subscribe('sock-node', `fs:${dir}/**`); + expect(nodeForm.match).toBeNull(); + expect(pathForm.match).toBe('**'); + + await removeAt(dir); + await vi.waitFor( + async () => + expect( + (await held('sock-node')).map((row) => row.subId), + ).toEqual([pathForm.subId]), + { timeout: 5_000, interval: 25 }, + ); + + // The path is back, with a uid the ended subscription never named. + await folder(dir); + await drain(); + await fs().touch(user.id, { path: `${dir}/after.txt` }); + await settled(); + + expect(delivered.map((d) => d.subId)).toEqual([pathForm.subId]); + }); + + it('is delivered its final removal before it ends', async () => { + const dir = await folder(`${home}/node-form-final`); + const sub = await subscribe('sock-final', `fs:${dir}`); + await drain(); + + await removeAt(dir); + await settled(); + + expect(delivered.map((d) => d.subId)).toEqual([sub.subId]); + expect(delivered[0].event).toMatchObject({ op: 'remove' }); + }); + + it('deletes a durable row, and tells its holder the anchor went', async () => { + const dir = await folder(`${home}/node-form-durable`); + const sub = ( + await events().subscribeDurable(user.actor, { subject: `fs:${dir}` }) + ).sub; + + await removeAt(dir); + await gone(sub.subId); + + const listed = await events().listDurable(user.actor); + expect(listed.items.map((row) => row.subId)).not.toContain(sub.subId); + + const ended = await vi.waitFor( + async () => { + const rows = await env.server.stores.notification.listByUserId( + user.id, + {}, + ); + const match = rows.find( + (row: { type?: string }) => row.type === 'app.events.ended', + ); + expect(match).toBeDefined(); + return match as { value: unknown }; + }, + { timeout: 5_000, interval: 25 }, + ); + expect(ended.value).toMatchObject({ + subject: `fs:${dir}`, + reason: 'anchor_deleted', + }); + + // Recreating the path mints a new uid, which nothing is watching. + await folder(dir); + await drain(); + await fs().touch(user.id, { path: `${dir}/after.txt` }); + await quiet(); + expect(delivered).toEqual([]); + }); +}); + +describe('a path-form subscription held over a share', () => { + it('ends rather than climbing onto a folder its holder cannot see', async () => { + const guestRow = await env.server.stores.user.getByUsername( + env.users.other.username, + ); + const guest = makeActor({ user: guestRow as never }); + const shared = await folder(`${home}/reanchor-shared`); + await env.server.services.acl.setUserUser( + user.actor, + guest, + { + path: shared, + resolveAncestors: () => fs().getAncestorChain(shared), + }, + 'list', + ); + const sub = ( + await events().subscribeDurable(guest, { + subject: `fs:${shared}/**`, + }) + ).sub; + + // The nearest survivor is the owner's home, which the guest was never + // allowed to watch; the row ends instead of moving there. + await removeAt(shared); + await gone(sub.subId); + + const ended = await vi.waitFor( + async () => { + const rows = await env.server.stores.notification.listByUserId( + guestRow!.id, + {}, + ); + const match = rows.find( + (row: { type?: string }) => row.type === 'app.events.ended', + ); + expect(match).toBeDefined(); + return match as { value: unknown }; + }, + { timeout: 5_000, interval: 25 }, + ); + expect(ended.value).toMatchObject({ + subject: `fs:${shared}/**`, + reason: 'anchor_deleted', + }); + }); +}); diff --git a/src/backend/services/events/authorization.ts b/src/backend/services/events/authorization.ts index 492bff9c2..705351050 100644 --- a/src/backend/services/events/authorization.ts +++ b/src/backend/services/events/authorization.ts @@ -17,7 +17,7 @@ * along with this program. If not, see . */ -import { makeActor, type Actor } from '../../core/actor.js'; +import { actorUid, makeActor, type Actor } from '../../core/actor.js'; import { HttpError } from '../../core/http/HttpError.js'; import type { UserRow } from '../../stores/user/UserStore.js'; import type { @@ -36,7 +36,9 @@ import type { * re-run per event, against the node the event is about rather than the anchor: * that is what keeps a match filter from reaching anything the holder could not * have subscribed to directly, and what makes a revoked share stop delivering - * without anyone having to find and delete the row. + * without anyone having to find and delete the row. The answer is cached by the + * permission cache's own generation (`deliveryAuthCache`), so a grant or a + * revoke is what re-opens the question. * * The failure is a `getSafeAclError` failure, not a bare 403: a node the caller * cannot even `see` is answered as absent, because a distinguishable @@ -78,6 +80,8 @@ export interface EventAclDeps { ) => Promise>; getUser: (userId: number) => Promise; getApp: (uid: string) => Promise<{ id?: number } | null>; + /** `PermissionStore.getCacheGeneration`, for the re-check cache's key. */ + getCacheGeneration: (actorUid: string) => Promise; } /** @@ -114,17 +118,37 @@ export const nodeDescriptor = ( * the app is resolved rather than taken from the row, because a grant to an app * is stored against its numeric id and is invisible to an actor without one. */ -const grantActor = async ( +export const resolveGrantActor = async ( grant: SubscriptionGrant, - user: UserRow, deps: EventAclDeps, ): Promise => { + const user = await deps.getUser(grant.holderUserId); + if (!user) return null; if (!grant.appUid) return makeActor({ user, app: null }); const app = await deps.getApp(grant.appUid); if (!app) return null; return makeActor({ user, app: { uid: grant.appUid, id: app.id } }); }; +/** + * The permission-cache counters this identity's readings hang on, read as one + * value. An app acts through its user, so either counter moving has to change + * the answer — the same pair the permission cache folds into its own keys. + * + * Joined rather than summed: two counter states must never collide on one tag. + */ +export const deliveryGenerationTag = async ( + actor: Actor, + deps: Pick, +): Promise => { + const keys = [actorUid(actor)]; + if (actor.app && actor.user?.uuid) keys.push(`user:${actor.user.uuid}`); + const generations = await Promise.all( + keys.map((key) => deps.getCacheGeneration(key)), + ); + return generations.join('.'); +}; + /** * Authorize a subscribe. Returns the mode the check succeeded under, which the * row stores; throws the safe error otherwise. @@ -157,18 +181,18 @@ export const assertSubscribeAuthorized = async ( * Whether a stored subscription may still be delivered an event about `node`. * Anything that cannot be decided is a no: a delivery is not worth failing a * write over, and silence is the safe direction. + * + * Takes the resolved identity rather than the row, because the caller has to + * resolve it first anyway to know which generation the answer is keyed by. */ export const checkDeliveryAuthorized = async ( - grant: SubscriptionGrant, + actor: Actor, + permission: AclMode, node: ResourceDescriptor, - deps: EventAclDeps, + deps: Pick, ): Promise => { try { - const user = await deps.getUser(grant.holderUserId); - if (!user) return false; - const actor = await grantActor(grant, user, deps); - if (!actor) return false; - return await deps.acl.check(actor, node, grant.permission); + return await deps.acl.check(actor, node, permission); } catch { return false; } diff --git a/src/backend/services/events/deliveryAuthCache.test.ts b/src/backend/services/events/deliveryAuthCache.test.ts new file mode 100644 index 000000000..265c3a809 --- /dev/null +++ b/src/backend/services/events/deliveryAuthCache.test.ts @@ -0,0 +1,108 @@ +/* + * 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 { afterEach, describe, expect, it, vi } from 'vitest'; +import { DeliveryAuthCache } from './deliveryAuthCache.js'; + +const key = ( + subId: string, + generation: string, + nodeUid = 'node-1', +): { subId: string; generation: string; nodeUid: string } => ({ + subId, + generation, + nodeUid, +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('DeliveryAuthCache', () => { + it('answers what it was told, and nothing it was not', () => { + const cache = new DeliveryAuthCache(); + cache.write(key('sub-1', 'g1'), true); + + expect(cache.read(key('sub-1', 'g1'))).toBe(true); + expect(cache.read(key('sub-2', 'g1'))).toBeNull(); + }); + + it('keeps a denial as firmly as an approval', () => { + const cache = new DeliveryAuthCache(); + cache.write(key('sub-1', 'g1'), false); + + expect(cache.read(key('sub-1', 'g1'))).toBe(false); + }); + + it('has no answer once the permission generation moves', () => { + const cache = new DeliveryAuthCache(); + cache.write(key('sub-1', 'g1'), true); + + expect(cache.read(key('sub-1', 'g2'))).toBeNull(); + }); + + it('decides per node, because the check is about the node', () => { + const cache = new DeliveryAuthCache(); + cache.write(key('sub-1', 'g1', 'allowed'), true); + cache.write(key('sub-1', 'g1', 'closed'), false); + + expect(cache.read(key('sub-1', 'g1', 'allowed'))).toBe(true); + expect(cache.read(key('sub-1', 'g1', 'closed'))).toBe(false); + expect(cache.read(key('sub-1', 'g1', 'unseen'))).toBeNull(); + }); + + it('stops trusting an answer nothing has refreshed', () => { + vi.useFakeTimers(); + const cache = new DeliveryAuthCache(100, 1_000); + cache.write(key('sub-1', 'g1'), true); + + vi.advanceTimersByTime(999); + expect(cache.read(key('sub-1', 'g1'))).toBe(true); + + vi.advanceTimersByTime(2); + expect(cache.read(key('sub-1', 'g1'))).toBeNull(); + }); + + it('drops the least recently used rather than growing', () => { + const cache = new DeliveryAuthCache(2); + cache.write(key('sub-1', 'g1'), true); + cache.write(key('sub-2', 'g1'), true); + // Touching the oldest is what makes it the youngest. + cache.read(key('sub-1', 'g1')); + cache.write(key('sub-3', 'g1'), true); + + expect(cache.size).toBe(2); + expect(cache.read(key('sub-1', 'g1'))).toBe(true); + expect(cache.read(key('sub-2', 'g1'))).toBeNull(); + expect(cache.read(key('sub-3', 'g1'))).toBe(true); + }); + + it('forgets every answer about one subscription', () => { + const cache = new DeliveryAuthCache(); + cache.write(key('sub-1', 'g1', 'a'), true); + cache.write(key('sub-1', 'g2', 'b'), true); + cache.write(key('sub-2', 'g1', 'a'), true); + + cache.forget('sub-1'); + + expect(cache.read(key('sub-1', 'g1', 'a'))).toBeNull(); + expect(cache.read(key('sub-1', 'g2', 'b'))).toBeNull(); + expect(cache.read(key('sub-2', 'g1', 'a'))).toBe(true); + }); +}); diff --git a/src/backend/services/events/deliveryAuthCache.ts b/src/backend/services/events/deliveryAuthCache.ts new file mode 100644 index 000000000..a3f0863ae --- /dev/null +++ b/src/backend/services/events/deliveryAuthCache.ts @@ -0,0 +1,138 @@ +/* + * 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 { PERMISSION_SCAN_CACHE_TTL_SECONDS } from '../permission/consts.js'; + +/** + * Delivery-time access decisions, held across events. + * + * The re-check is the most expensive thing on the dispatch path — a permission + * scan per surviving row — and a busy anchor asks the same question of the same + * subscription over and over. Memoizing inside one event is not enough for a + * save loop, which is many events. + * + * Entries are keyed by the permission cache's **own** generation for the row's + * identity, so nothing here needs its own invalidation: a grant or revoke + * already bumps that counter, cluster- and region-wide, and a bumped generation + * simply produces a key nothing has answered yet. The node is part of the key + * because the decision is about the node the event is about, not the anchor — a + * filter reaching into a subfolder the holder cannot list is denied there and + * allowed elsewhere under the same subscription. + * + * The TTL is a backstop for the paths that change access without bumping + * anything, and is the scan cache's own, so this can never serve an answer + * staler than the layer it caches over. Bounded, and least-recently-used: a + * `Map` iterates in insertion order, so re-inserting on read moves an entry to + * the young end. + */ + +export interface DeliveryAuthKey { + subId: string; + /** Permission-cache generation(s) the row's identity depends on. */ + generation: string; + /** Uid of the node the decision is about. */ + nodeUid: string; +} + +export const DELIVERY_AUTH_CACHE_MAX_ENTRIES = 20_000; + +export const DELIVERY_AUTH_CACHE_TTL_MS = + PERMISSION_SCAN_CACHE_TTL_SECONDS * 1000; + +const cacheKey = (key: DeliveryAuthKey): string => + `${key.subId}|${key.generation}|${key.nodeUid}`; + +interface CacheEntry { + allowed: boolean; + cachedAt: number; +} + +export class DeliveryAuthCache { + readonly #entries = new Map(); + /** Entry ids by subscription, so forgetting one is not a scan of all. */ + readonly #bySub = new Map>(); + readonly #maxEntries: number; + readonly #ttlMs: number; + + constructor( + maxEntries: number = DELIVERY_AUTH_CACHE_MAX_ENTRIES, + ttlMs: number = DELIVERY_AUTH_CACHE_TTL_MS, + ) { + this.#maxEntries = Math.max(1, maxEntries); + this.#ttlMs = Math.max(0, ttlMs); + } + + get size(): number { + return this.#entries.size; + } + + /** The decision, or `null` when it has to be made again. */ + read(key: DeliveryAuthKey): boolean | null { + const id = cacheKey(key); + const entry = this.#entries.get(id); + if (!entry) return null; + if (Date.now() - entry.cachedAt > this.#ttlMs) { + this.#drop(key.subId, id); + return null; + } + this.#entries.delete(id); + this.#entries.set(id, entry); + return entry.allowed; + } + + write(key: DeliveryAuthKey, allowed: boolean): void { + const id = cacheKey(key); + this.#entries.delete(id); + this.#entries.set(id, { allowed, cachedAt: Date.now() }); + const ids = this.#bySub.get(key.subId) ?? new Set(); + ids.add(id); + this.#bySub.set(key.subId, ids); + while (this.#entries.size > this.#maxEntries) { + const oldest = this.#entries.keys().next(); + if (oldest.done) break; + this.#drop( + oldest.value.slice(0, oldest.value.indexOf('|')), + oldest.value, + ); + } + } + + #drop(subId: string, id: string): void { + this.#entries.delete(id); + const ids = this.#bySub.get(subId); + if (!ids) return; + ids.delete(id); + if (ids.size === 0) this.#bySub.delete(subId); + } + + /** + * Drop every decision about one subscription. For a row that changed shape + * rather than access — a re-anchor — where the generation it was keyed by + * has not moved and the old answers are about a node it no longer watches. + */ + forget(subId: string): void { + for (const id of this.#bySub.get(subId) ?? []) this.#entries.delete(id); + this.#bySub.delete(subId); + } + + clear(): void { + this.#entries.clear(); + this.#bySub.clear(); + } +} diff --git a/src/backend/services/events/durable.integration.test.ts b/src/backend/services/events/durable.integration.test.ts index 28f08cc5a..8ab39c0af 100644 --- a/src/backend/services/events/durable.integration.test.ts +++ b/src/backend/services/events/durable.integration.test.ts @@ -618,7 +618,8 @@ describe('a durable row across a share', () => { ); delivered.length = 0; - // The row is still registered; it just no longer authorizes anything. + // The re-check refuses it at once; the settle then takes the row out + // of service behind it. await fs().touch(userId, { path: `${sharedPath}/second.txt` }); await quiet(); diff --git a/src/backend/services/events/revocationSettle.integration.test.ts b/src/backend/services/events/revocationSettle.integration.test.ts new file mode 100644 index 000000000..b5b9c5d7a --- /dev/null +++ b/src/backend/services/events/revocationSettle.integration.test.ts @@ -0,0 +1,684 @@ +/* + * 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 happens to a subscription when the grant under it is taken away. + * + * Two mechanisms, and the tests keep them apart on purpose. The delivery + * re-check is the backstop: it denies from the next event, on its own, with + * nothing having to find the row — which is what the session cases prove, since + * nothing settles those. The settle is the cleanup: it takes the durable rows + * out of service so a revoked subscription stops costing anything at all. + */ + +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 { SUSPENDED_ROW_TTL_DAYS } from '../../controllers/events/limits.js'; +import { makeActor, type Actor } from '../../core/actor.js'; +import { runWithContext } from '../../core/context.js'; +import { + createTestUser, + setupPuterTestEnv, + type PuterTestEnv, +} from '../../testUtil.js'; +import type { IConfig } from '../../types.js'; +import type { AclMode } from '../acl/ACLService.js'; +import type { DeliveryEnvelope } from './EventsService.js'; +import { fsAnchorToken } from './subjects.js'; + +const BOOT_TIMEOUT_MS = 120_000; +const TABLE = 'event_subscriptions'; + +let env: PuterTestEnv; +let owner: { actor: Actor; username: string; id: number }; +let guest: { actor: Actor; username: string; id: number }; +let delivered: DeliveryEnvelope[]; + +const events = () => env.server.services.events; +const fs = () => env.server.services.fs; + +const descriptor = (path: string) => ({ + path, + resolveAncestors: () => fs().getAncestorChain(path), +}); + +const folder = async (path: string): Promise => { + await fs().mkdir(owner.id, { path, createMissingParents: true }); + return path; +}; + +const share = (path: string, mode: AclMode): Promise => + env.server.services.acl.setUserUser( + owner.actor, + guest.actor, + descriptor(path), + mode, + ); + +const unshare = async (path: string, mode: AclMode): Promise => { + const entry = await env.server.stores.fsEntry.getEntryByPath(path); + await env.server.services.permission.revokeUserUserPermission( + owner.actor, + guest.username, + `fs:${entry!.uid}:${mode}`, + ); +}; + +/** + * The real user-facing surface: `ShareService`, not the ACL/permission layer + * it settles on top of. What the settle mechanism actually has to survive is + * everything this service does around the grant — index-row bookkeeping, + * authorization, delegate resolution — not just the grant itself. + */ +const shareViaService = ( + path: string, + recipientUsername: string, + mode: AclMode, +) => + runWithContext({ actor: owner.actor }, () => + env.server.services.share.share(owner.actor, { + path, + recipient: { username: recipientUsername }, + mode, + }), + ); + +const unshareViaService = (path: string, recipientUsername: string) => + runWithContext({ actor: owner.actor }, () => + env.server.services.share.unshare(owner.actor, { + path, + recipient: { username: recipientUsername }, + }), + ); + +/** A second guest, for proving a revoke settles only the holder it named. */ +const makeGuest = async (): Promise<{ + actor: Actor; + username: string; + id: number; +}> => { + const username = `settle-guest-${uuidv4().slice(0, 8)}`; + await createTestUser(env.server, { username, password: 'pw-test-1234' }); + const row = await env.server.stores.user.getByUsername(username); + return { + actor: makeActor({ user: row as never }), + username, + id: row!.id, + }; +}; + +const uidOf = async (path: string): Promise => { + const entry = await env.server.stores.fsEntry.getEntryByPath(path); + return entry!.uid; +}; + +const write = (path: string): Promise => + fs().touch(owner.id, { path }); + +const uniquePath = (base: string) => + `${base}/n-${Math.random().toString(36).slice(2, 8)}`; + +const settled = (count = 1) => + vi.waitFor(() => expect(delivered.length).toBeGreaterThanOrEqual(count), { + timeout: EVENTS_COALESCE_WINDOW_MS * 12, + interval: 25, + }); + +const quiet = () => + new Promise((resolve) => + setTimeout(resolve, EVENTS_COALESCE_WINDOW_MS * 3), + ); + +const rowOf = async ( + subId: string, +): Promise<{ suspended_at: unknown; suspended_reason: unknown }> => { + const [row] = await env.server.clients.db.pread( + `SELECT \`suspended_at\`, \`suspended_reason\` FROM \`${TABLE}\` WHERE \`sub_id\` = ?`, + [subId], + ); + return row as { suspended_at: unknown; suspended_reason: unknown }; +}; + +/** Wait for the settle the revoke kicked off on the bus to land. */ +const endedNotifications = async ( + userId: number, +): Promise> => + (await env.server.stores.notification.listByUserId(userId, {})).filter( + (row: { type?: string }) => row.type === 'app.events.ended', + ) as Array<{ value: unknown }>; + +const suspendedRow = (subId: string) => + vi.waitFor( + async () => { + const row = await rowOf(subId); + expect(row?.suspended_reason).toBe('permission_revoked'); + }, + { timeout: 5_000, interval: 25 }, + ); + +/** Whether `path`'s anchor token is watched in its own owner's keyspace. */ +const watches = async ( + path: string, + ownerUserId: number = owner.id, +): Promise => { + const watched = await env.server.stores.eventSubscription.watchedTokens( + ownerUserId, + [fsAnchorToken(await uidOf(path))], + ); + return watched.length > 0; +}; + +/** An app of the owner's, granted `list` on one folder. */ +const makeApp = async (path: string): Promise => { + const uid = `app-${uuidv4()}`; + await env.server.clients.db.write( + 'INSERT INTO `apps` (`uid`, `name`, `title`, `index_url`, `owner_user_id`) VALUES (?, ?, ?, ?, ?)', + [uid, uid, uid, `https://${uid}.example/`, owner.id], + ); + await env.server.services.permission.grantUserAppPermission( + owner.actor, + uid, + `fs:${await uidOf(path)}:list`, + ); + const app = await env.server.stores.app.getByUid(uid); + return makeActor({ + user: owner.actor.user as never, + app: { uid, id: app!.id }, + }); +}; + +const clearRows = async () => { + await env.server.clients.db.write(`DELETE FROM \`${TABLE}\``, []); + for (const id of [owner.id, guest.id]) { + events().invalidateUser(id); + await env.server.stores.eventSubscription.markRegionCold(id); + await env.server.stores.durableSubscription.warmRegion(id); + } + delivered.length = 0; +}; + +beforeAll(async () => { + env = await setupPuterTestEnv({ events: { enabled: true } } as IConfig); + + const ownerRow = await env.server.stores.user.getByUsername( + env.users.user.username, + ); + const guestRow = await env.server.stores.user.getByUsername( + env.users.other.username, + ); + owner = { + actor: makeActor({ user: ownerRow as never }), + username: env.users.user.username, + id: ownerRow!.id, + }; + guest = { + actor: makeActor({ user: guestRow as never }), + username: env.users.other.username, + id: guestRow!.id, + }; + + delivered = []; + events().onDelivered = (envelope) => delivered.push(envelope); +}, BOOT_TIMEOUT_MS); + +afterAll(async () => { + await env?.shutdown(); +}); + +describe('the delivery re-check on its own', () => { + it('stops a session subscription, and meters nothing, without any settle', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/backstop-session`); + await share(path, 'list'); + const sub = ( + await events().subscribe(guest.actor, 'guest-backstop', { + subject: `fs:${path}`, + }) + ).sub; + + await write(uniquePath(path)); + await settled(); + expect(delivered.map((d) => d.subId)).toEqual([sub.subId]); + + await unshare(path, 'list'); + delivered.length = 0; + + await write(uniquePath(path)); + await quiet(); + + // Nothing reached a subscriber, so the seam metering hangs off saw + // nothing either — and the row is still exactly where it was, which is + // what makes this the backstop working alone. + expect(delivered).toEqual([]); + const held = await env.server.stores.eventSubscription.listForSocket( + guest.id, + 'guest-backstop', + ); + expect(held.map((row) => row.subId)).toEqual([sub.subId]); + }); + + it('answers from cache until the permission generation moves', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/backstop-cache`); + await share(path, 'list'); + await events().subscribe(guest.actor, 'guest-cache', { + subject: `fs:${path}`, + }); + + // Only the guest's own re-check counts: the owner's write is checked + // as the owner, on its own path into ACL. + const acl = env.server.services.acl; + const passThrough = acl.check.bind(acl); + let checks = 0; + const spy = vi + .spyOn(acl, 'check') + .mockImplementation(async (actor, resource, mode) => { + if (actor?.user?.id === guest.id) checks++; + return passThrough(actor, resource, mode); + }); + + // Renames keep the uid, so these are several events about one node — + // which is what the cache is keyed by, alongside the generation. + const probe = await fs().touch(owner.id, { + path: `${path}/probe.txt`, + }); + try { + await settled(); + expect(checks).toBe(1); + + delivered.length = 0; + checks = 0; + await fs().rename(owner.id, probe, 'probe-again.txt'); + await settled(); + expect(checks).toBe(0); + + // The revoke bumps the permission generation, which is the key the + // cached answer was filed under. + await unshare(path, 'list'); + delivered.length = 0; + checks = 0; + const renamed = await env.server.stores.fsEntry.getEntryByUuid( + probe.uuid, + ); + await fs().rename(owner.id, renamed!, 'probe-once-more.txt'); + await quiet(); + + expect(checks).toBe(1); + expect(delivered).toEqual([]); + } finally { + spy.mockRestore(); + } + }); +}); + +describe('what a revoked grant settles', () => { + it('suspends the durable rows it was holding up and purges their backlog', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/settle-durable`); + await share(path, 'list'); + const sub = ( + await events().subscribeDurable(guest.actor, { + subject: `fs:${path}`, + delivery: 'single', + handlerName: 'onChange', + }) + ).sub; + + await write(uniquePath(path)); + await vi.waitFor( + async () => + expect( + await env.server.stores.pendingDelivery.depth(sub.subId), + ).toBeGreaterThan(0), + { timeout: 5_000, interval: 25 }, + ); + expect(await watches(path)).toBe(true); + + await unshare(path, 'list'); + await suspendedRow(sub.subId); + + expect(await watches(path)).toBe(false); + // A revoked backlog names paths its holder just lost the right to see. + expect(await env.server.stores.pendingDelivery.depth(sub.subId)).toBe( + 0, + ); + + delivered.length = 0; + await write(uniquePath(path)); + await quiet(); + expect(delivered).toEqual([]); + }); + + it('tells the holder their subscription ended, and why', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/settle-notified`); + await share(path, 'list'); + const sub = ( + await events().subscribeDurable(guest.actor, { + subject: `fs:${path}`, + }) + ).sub; + + await unshare(path, 'list'); + await suspendedRow(sub.subId); + + const ended = await vi.waitFor( + async () => { + const rows = await env.server.stores.notification.listByUserId( + guest.id, + {}, + ); + const match = rows.find( + (row: { type?: string }) => + row.type === 'app.events.ended', + ); + expect(match).toBeDefined(); + return match as { audience: string; value: unknown }; + }, + { timeout: 5_000, interval: 25 }, + ); + + expect(ended.audience).toBe('app-user'); + expect(ended.value).toMatchObject({ + subject: `fs:${path}`, + reason: 'permission_revoked', + }); + }); + + it('leaves a share whose mode only changed exactly where it was', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/settle-upgraded`); + await share(path, 'list'); + const sub = ( + await events().subscribeDurable(guest.actor, { + subject: `fs:${path}`, + }) + ).sub; + + // Widening a share is stored as a grant plus a revoke of the mode it + // replaces; settling on the revoke alone would end a subscription + // whose reach just grew. + await share(path, 'write'); + await quiet(); + + expect((await rowOf(sub.subId)).suspended_at).toBeFalsy(); + delivered.length = 0; + await write(uniquePath(path)); + await settled(); + expect(delivered.map((d) => d.subId)).toEqual([sub.subId]); + }); + + it('does not resume on a re-grant, and a fresh subscribe still works', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/settle-regrant`); + await share(path, 'list'); + const stale = ( + await events().subscribeDurable(guest.actor, { + subject: `fs:${path}`, + }) + ).sub; + + await unshare(path, 'list'); + await suspendedRow(stale.subId); + + await share(path, 'list'); + delivered.length = 0; + await write(uniquePath(path)); + await quiet(); + expect(delivered).toEqual([]); + + const fresh = ( + await events().subscribeDurable(guest.actor, { + subject: `fs:${path}`, + }) + ).sub; + delivered.length = 0; + await write(uniquePath(path)); + await settled(); + expect(delivered.map((d) => d.subId)).toEqual([fresh.subId]); + }); + + it('settles everything an app holds when its access is withdrawn wholesale', async () => { + await clearRows(); + const one = await folder(`/${owner.username}/settle-app-one`); + const two = await folder(`/${owner.username}/settle-app-two`); + const appActor = await makeApp(one); + await env.server.services.permission.grantUserAppPermission( + owner.actor, + appActor.app!.uid, + `fs:${await uidOf(two)}:list`, + ); + + const held = [ + (await events().subscribeDurable(appActor, { subject: `fs:${one}` })) + .sub, + (await events().subscribeDurable(appActor, { subject: `fs:${two}` })) + .sub, + ]; + + await env.server.services.permission.revokeUserAppAll( + owner.actor, + appActor.app!.uid, + ); + + for (const sub of held) await suspendedRow(sub.subId); + expect(await watches(one)).toBe(false); + expect(await watches(two)).toBe(false); + }); + + it('settles a subscription when the owner unshares through ShareService', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/settle-share-service`); + await shareViaService(path, guest.username, 'list'); + const sub = ( + await events().subscribeDurable(guest.actor, { + subject: `fs:${path}`, + }) + ).sub; + + await unshareViaService(path, guest.username); + await suspendedRow(sub.subId); + + expect(await watches(path)).toBe(false); + delivered.length = 0; + await write(uniquePath(path)); + await quiet(); + expect(delivered).toEqual([]); + }); + + it('settles a subscription when one app permission is revoked, not just all of them', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/settle-app-single`); + const appActor = await makeApp(path); + const sub = ( + await events().subscribeDurable(appActor, { subject: `fs:${path}` }) + ).sub; + + await env.server.services.permission.revokeUserAppPermission( + owner.actor, + appActor.app!.uid, + `fs:${await uidOf(path)}:list`, + ); + await suspendedRow(sub.subId); + + expect(await watches(path)).toBe(false); + }); + + it('leaves another holder, and an unrelated anchor, out of a revoke', async () => { + await clearRows(); + const shared = await folder(`/${owner.username}/settle-scope-shared`); + const otherGuest = await makeGuest(); + + await share(shared, 'list'); + const guestSub = ( + await events().subscribeDurable(guest.actor, { + subject: `fs:${shared}`, + }) + ).sub; + + await env.server.services.acl.setUserUser( + owner.actor, + otherGuest.actor, + descriptor(shared), + 'list', + ); + const otherSub = ( + await events().subscribeDurable(otherGuest.actor, { + subject: `fs:${shared}`, + }) + ).sub; + + // The guest's own folder has nothing to do with the share about to be + // revoked — a different anchor, held by the same user. Made as the + // guest, not `owner`: `folder()` is bound to owner.id, and this path + // is outside owner's tree. + const own = `/${guest.username}/settle-scope-own`; + await fs().mkdir(guest.id, { path: own, createMissingParents: true }); + const ownSub = ( + await events().subscribeDurable(guest.actor, { subject: `fs:${own}` }) + ).sub; + + await unshare(shared, 'list'); + await suspendedRow(guestSub.subId); + + expect((await rowOf(otherSub.subId)).suspended_at).toBeFalsy(); + expect((await rowOf(ownSub.subId)).suspended_at).toBeFalsy(); + // `own` is anchored in the guest's own keyspace, not the owner's. + expect(await watches(own, guest.id)).toBe(true); + }); + + it('reaps a row that has been suspended past the retention window', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/settle-reaped`); + await share(path, 'list'); + const sub = ( + await events().subscribeDurable(guest.actor, { + subject: `fs:${path}`, + }) + ).sub; + + await unshare(path, 'list'); + await suspendedRow(sub.subId); + + // Still there the day it is suspended — that is how its holder finds + // out what happened. + expect(await events().sweepSuspended()).toBe(0); + expect(await rowOf(sub.subId)).toBeDefined(); + + const aged = + Math.floor(Date.now() / 1000) - + (SUSPENDED_ROW_TTL_DAYS + 1) * 24 * 60 * 60; + await env.server.clients.db.write( + `UPDATE \`${TABLE}\` SET \`suspended_at\` = ? WHERE \`sub_id\` = ?`, + [aged, sub.subId], + ); + + expect(await events().sweepSuspended()).toBe(1); + expect(await rowOf(sub.subId)).toBeUndefined(); + }); + + it('tells an app holder once when its access is withdrawn wholesale', async () => { + await clearRows(); + const one = await folder(`/${owner.username}/settle-once-one`); + const two = await folder(`/${owner.username}/settle-once-two`); + const appActor = await makeApp(one); + await env.server.services.permission.grantUserAppPermission( + owner.actor, + appActor.app!.uid, + `fs:${await uidOf(two)}:list`, + ); + const held = [ + (await events().subscribeDurable(appActor, { subject: `fs:${one}` })) + .sub, + (await events().subscribeDurable(appActor, { subject: `fs:${two}` })) + .sub, + ]; + const before = (await endedNotifications(owner.id)).length; + + await env.server.services.permission.revokeUserAppAll( + owner.actor, + appActor.app!.uid, + ); + for (const sub of held) await suspendedRow(sub.subId); + await quiet(); + + const ended = await endedNotifications(owner.id); + expect(ended).toHaveLength(before + 1); + expect(ended[0].value).toMatchObject({ + count: 2, + reason: 'permission_revoked', + }); + expect( + (ended[0].value as { subjects: string[] }).subjects.sort(), + ).toEqual([`fs:${one}`, `fs:${two}`].sort()); + }); + + it('settles a row once however many times the same withdrawal is heard', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/settle-twice`); + await share(path, 'list'); + const sub = ( + await events().subscribeDurable(guest.actor, { + subject: `fs:${path}`, + }) + ).sub; + const before = (await endedNotifications(guest.id)).length; + const revocation = { + holderUserId: guest.id, + appUid: null, + permission: `fs:${await uidOf(path)}:list`, + }; + + // The unshare announces once on its own; two more passes race it. + await unshare(path, 'list'); + const settled = await Promise.all([ + events().settleRevokedGrant(revocation), + events().settleRevokedGrant(revocation), + ]); + await suspendedRow(sub.subId); + await quiet(); + + expect(settled.reduce((sum, n) => sum + n, 0)).toBeLessThanOrEqual(1); + expect(await endedNotifications(guest.id)).toHaveLength(before + 1); + }); + + it('never fails the revoke when the settle listener throws', async () => { + await clearRows(); + const path = await folder(`/${owner.username}/settle-listener-throws`); + await share(path, 'list'); + + const boom = vi + .spyOn(events(), 'settleRevokedGrant') + .mockRejectedValueOnce(new Error('settle boom')); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + try { + // The revoke itself — not just the announce — has to come back + // clean even though the thing listening for it just failed. + await expect(unshare(path, 'list')).resolves.toBeUndefined(); + await vi.waitFor(() => expect(boom).toHaveBeenCalled(), { + timeout: 5_000, + interval: 25, + }); + } finally { + boom.mockRestore(); + warn.mockRestore(); + } + }); +}); diff --git a/src/backend/services/events/singleDelivery.test.ts b/src/backend/services/events/singleDelivery.test.ts index ff5148b71..ae3e9df05 100644 --- a/src/backend/services/events/singleDelivery.test.ts +++ b/src/backend/services/events/singleDelivery.test.ts @@ -208,6 +208,7 @@ beforeEach(async () => { getById: async (id: number) => ({ id, uuid: `user-${id}` }), }, app: { getByUid: async (uid: string) => ({ uid, id: 1 }) }, + permission: { getCacheGeneration: async () => 1 }, } as never, { socket: { diff --git a/src/backend/services/notification/NotificationService.test.ts b/src/backend/services/notification/NotificationService.test.ts index 749b86539..c5eaa5326 100644 --- a/src/backend/services/notification/NotificationService.test.ts +++ b/src/backend/services/notification/NotificationService.test.ts @@ -196,7 +196,7 @@ describe('NotificationService.notify', () => { {}, ); expect(row.type).toBe('app.events.ended'); - expect(row.audience).toBe('developer'); + expect(row.audience).toBe('app-user'); expect(row.app_uid).toBe(appUid); persisted.stop(); }); @@ -235,7 +235,11 @@ describe('NotificationService.notify', () => { ).rejects.toThrow('cannot name an app'); await expect( - notifications.notify([user.id], {}, { type: 'app.events.ended' }), + notifications.notify( + [user.id], + {}, + { type: 'app.events.suspended' }, + ), ).rejects.toThrow('requires an app uid'); expect(pushed.seen).toEqual([]); diff --git a/src/backend/services/notification/notificationAudience.test.ts b/src/backend/services/notification/notificationAudience.test.ts index 2ab3f45f7..895550a08 100644 --- a/src/backend/services/notification/notificationAudience.test.ts +++ b/src/backend/services/notification/notificationAudience.test.ts @@ -136,6 +136,17 @@ describe('canViewNotification — app-user rows', () => { it('denies an actor whose app was never resolved', () => { expect(canViewNotification(own, unresolved)).toBe(false); }); + + it('treats a row naming no app as the recipient`s own, and no app`s', () => { + // A subscription a plain session made and then lost holds no app — + // the row is the holder's own, exactly like an unattributed + // `developer` row, and no app may read it in its place. + const unattributed = { audience: 'app-user', appUid: null }; + expect(canViewNotification(unattributed, session)).toBe(true); + expect(canViewNotification(unattributed, userIssuedToken)).toBe(true); + expect(canViewNotification(unattributed, appUnderUser)).toBe(false); + expect(canViewNotification(unattributed, appIssuedToken)).toBe(false); + }); }); describe('canViewNotification — unknown audiences', () => { diff --git a/src/backend/services/notification/notificationTypes.test.ts b/src/backend/services/notification/notificationTypes.test.ts index 0208cb652..8f86a52da 100644 --- a/src/backend/services/notification/notificationTypes.test.ts +++ b/src/backend/services/notification/notificationTypes.test.ts @@ -45,13 +45,16 @@ describe('NOTIFICATION_TYPES', () => { }); it('carries the types the events work will emit', () => { - for (const type of ['app.events.ended', 'app.events.suspended']) { - const entry = findNotificationType(type); - expect(entry).toMatchObject({ - audience: 'developer', - appScoped: true, - }); - } + // A handler that keeps failing is the developer's problem; a + // subscription that ended is the holder's. + expect(findNotificationType('app.events.suspended')).toMatchObject({ + audience: 'developer', + appScoped: true, + }); + expect(findNotificationType('app.events.ended')).toMatchObject({ + audience: 'app-user', + appScoped: false, + }); }); it('projects a subject naming the app, or the recipient when there is none', () => { @@ -60,11 +63,16 @@ describe('NOTIFICATION_TYPES', () => { account.subject({ userUuid: 'user-uuid', appUid: null }), ).toBe('notif:user-uuid:account'); - const developer = findNotificationType('app.events.ended')!; + const developer = findNotificationType('app.events.suspended')!; expect( developer.subject({ userUuid: 'user-uuid', appUid: 'app-1' }), ).toBe('notif:app-1:developer'); + const holder = findNotificationType('app.events.ended')!; + expect(holder.subject({ userUuid: 'user-uuid', appUid: 'app-1' })).toBe( + 'notif:app-1:app-user', + ); + // A worker bound to no app has no app to name. const worker = findNotificationType('app.worker.deployed')!; expect(worker.subject({ userUuid: 'user-uuid', appUid: null })).toBe( @@ -106,11 +114,17 @@ describe('resolveNotificationWrite', () => { }); it('rejects an app-scoped type with no app uid', () => { - expect(() => resolveNotificationWrite('app.events.ended', null)).toThrow( - 'requires an app uid', - ); - expect(() => resolveNotificationWrite('app.events.ended', '')).toThrow( - 'requires an app uid', + expect(() => + resolveNotificationWrite('app.events.suspended', null), + ).toThrow('requires an app uid'); + expect(() => + resolveNotificationWrite('app.events.suspended', ''), + ).toThrow('requires an app uid'); + }); + + it('lets a subscription an account made end without naming an app', () => { + expect(resolveNotificationWrite('app.events.ended', null).audience).toBe( + 'app-user', ); }); }); diff --git a/src/backend/services/notification/notificationTypes.ts b/src/backend/services/notification/notificationTypes.ts index 70c3478a0..d562eca05 100644 --- a/src/backend/services/notification/notificationTypes.ts +++ b/src/backend/services/notification/notificationTypes.ts @@ -92,12 +92,15 @@ export const NOTIFICATION_TYPES = [ groupable: false, subject: subjectFor('developer'), }, + // The recipient is the subscription's holder, not the app's owner: a + // subscription that ended without an unsubscribe is news for whoever made + // it. An account made its own carries no app, exactly as a worker row does. { type: 'app.events.ended', - audience: 'developer', - appScoped: true, + audience: 'app-user', + appScoped: false, groupable: false, - subject: subjectFor('developer'), + subject: subjectFor('app-user'), }, { type: 'app.events.suspended', diff --git a/src/backend/services/permission/PermissionService.ts b/src/backend/services/permission/PermissionService.ts index e72ab3c59..910c8e620 100644 --- a/src/backend/services/permission/PermissionService.ts +++ b/src/backend/services/permission/PermissionService.ts @@ -977,6 +977,9 @@ export class PermissionService extends PuterService { // Unconditional: the flat delete above can't report what it removed, so // skipping the bump on a no-op risks leaving a cached allow standing. if (user.uuid) await this.#bumpUserCacheGeneration(user.uuid); + // A revoke of a grant that was not there settled nothing; announcing + // it would only cost every listener a read. + if (revoked) this.#announceRevoked(user.id, null, permission); return revoked; } @@ -1158,6 +1161,7 @@ export class PermissionService extends PuterService { app.uid, ); } + this.#announceRevoked(actor.user.id, app.uid, permission); } async revokeUserAppAll( @@ -1196,6 +1200,7 @@ export class PermissionService extends PuterService { app.uid, ); } + this.#announceRevoked(actor.user.id, app.uid, null); } async grantDevAppPermission( @@ -1426,6 +1431,27 @@ export class PermissionService extends PuterService { return gens.join('.'); } + /** + * Tell whatever was standing on a grant that it is gone. Fire-and-forget: a + * listener that fails must not fail the revoke, and every check the grant + * used to answer already denies on its own. + */ + #announceRevoked( + holderUserId: number, + appUid: string | null, + permission: string | null, + ): void { + try { + this.clients.event.emit( + 'permission.revoked', + { holderUserId, appUid, permission }, + {}, + ); + } catch (err) { + console.warn('[PermissionService] revoke announce failed:', err); + } + } + /** Bump a plain user holder (`user:`). */ async #bumpUserCacheGeneration(userUuid: string): Promise { await this.stores.permission.bumpCacheGeneration(`user:${userUuid}`); diff --git a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts index 766fd2b4d..a71e650c4 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts +++ b/src/backend/stores/events/DurableSubscriptionStore.integration.test.ts @@ -451,3 +451,17 @@ describe('warming a cold region', () => { expect(page.items[0]?.suspendedReason).toBe('permission_revoked'); }); }); + +describe('suspending', () => { + it('suspends a row once, and tells a later pass it was not the one', async () => { + const { row } = await durable().create(input()); + + const first = await durable().suspend([row], 'permission_revoked'); + expect(first.suspended.map((r) => r.subId)).toEqual([row.subId]); + expect(first.bumps).toHaveLength(1); + + const second = await durable().suspend([row], 'permission_revoked'); + expect(second.suspended).toEqual([]); + expect(second.bumps).toEqual([]); + }); +}); diff --git a/src/backend/stores/events/DurableSubscriptionStore.ts b/src/backend/stores/events/DurableSubscriptionStore.ts index f3e82f984..aa8034b36 100644 --- a/src/backend/stores/events/DurableSubscriptionStore.ts +++ b/src/backend/stores/events/DurableSubscriptionStore.ts @@ -96,6 +96,19 @@ export interface DurableListOptions { includeTotal?: boolean; } +/** Why a row is out of service. Never auto-resumes for `permission_revoked`. */ +export type SuspendedReason = 'permission_revoked'; + +/** Where a row is moving to when its anchor is deleted under it. */ +export interface ReanchorInput { + token: string; + anchorUid: string; + anchorPath: string; + match: string; + /** The new anchor's owner, which is the keyspace the row is indexed in. */ + ownerUserId: number; +} + // -- Errors ----------------------------------------------------------- const contextTooLarge = (): HttpError => @@ -277,6 +290,86 @@ export class DurableSubscriptionStore extends PuterStore { return this.#bump(row.ownerUserId); } + /** + * Take a set of rows out of service without deleting them: one statement + * for the table, then each row out of the cache, then one generation per + * owner. Suspended rows stay listable — that is how their holder finds out + * what happened — but nothing rebuilds them into a watched set again. + */ + async suspend( + rows: readonly DurableSubscription[], + reason: SuspendedReason, + ): Promise<{ suspended: DurableSubscription[]; bumps: GenerationBump[] }> { + if (rows.length === 0) return { suspended: [], bumps: [] }; + + // One conditional write per row, so two settles racing over the same + // rows — an unshare withdraws several grant strings in a row — each + // learn exactly which rows they were the one to suspend. + const at = nowSeconds(); + const suspended: DurableSubscription[] = []; + for (const row of rows) { + const written = await this.clients.db.write( + `UPDATE \`${TABLE}\` SET \`suspended_at\` = ?, ` + + '`suspended_reason` = ? ' + + 'WHERE `sub_id` = ? AND `suspended_at` IS NULL', + [at, reason, row.subId], + ); + if (written.anyRowsAffected) + suspended.push({ + ...row, + suspendedAt: at, + suspendedReason: reason, + }); + } + + const owners = new Set(); + for (const row of suspended) { + await this.stores.eventSubscription.dropDurable(row); + owners.add(row.ownerUserId); + } + const bumps = await Promise.all( + [...owners].map((owner) => this.#bump(owner)), + ); + return { suspended, bumps }; + } + + /** + * Move one row onto a different anchor, keeping its identity. The cache + * entry moves with it — including across owners, which is a different + * keyspace — so both sides advance and neither is left holding a row that + * is no longer theirs. + */ + async reanchor( + row: DurableSubscription, + next: ReanchorInput, + ): Promise<{ row: DurableSubscription; bumps: GenerationBump[] }> { + await this.clients.db.write( + `UPDATE \`${TABLE}\` SET \`token\` = ?, \`anchor_uid\` = ?, ` + + '`anchor_path` = ?, `match` = ?, `owner_user_id` = ? ' + + 'WHERE `sub_id` = ?', + [ + next.token, + next.anchorUid, + next.anchorPath, + next.match, + next.ownerUserId, + row.subId, + ], + ); + + const moved: DurableSubscription = { ...row, ...next }; + await this.stores.eventSubscription.dropDurable(row); + await this.stores.eventSubscription.cacheDurable([moved]); + + const owners = new Set([row.ownerUserId, next.ownerUserId]); + return { + row: moved, + bumps: await Promise.all( + [...owners].map((owner) => this.#bump(owner)), + ), + }; + } + /** * Bring this region's cache for one owner up to date with the table, unless * it already is. Returns whether the table was read, which is what the @@ -294,22 +387,16 @@ export class DurableSubscriptionStore extends PuterStore { * stops delivering against it without waiting for a rebuild. */ async sweepExpired(batchSize: number): Promise { - const rows = await this.#listExpired(nowSeconds(), batchSize); - if (rows.length === 0) return 0; + return this.#reap(await this.#listExpired(nowSeconds(), batchSize)); + } - await this.clients.db.write( - `DELETE FROM \`${TABLE}\` WHERE \`sub_id\` IN ` + - `(${rows.map(() => '?').join(', ')})`, - rows.map((row) => row.subId), - ); - - const owners = new Set(); - for (const row of rows) { - await this.stores.eventSubscription.dropDurable(row); - owners.add(row.ownerUserId); - } - for (const ownerUserId of owners) await this.#bump(ownerUserId); - return rows.length; + /** + * Reap rows suspended longer than the retention window. A suspension that + * never resumes is a row kept only so its holder can see why it stopped, + * and that answer has a shelf life. + */ + async sweepSuspended(cutoff: number, batchSize: number): Promise { + return this.#reap(await this.#listSuspendedBefore(cutoff, batchSize)); } // -- Reads ------------------------------------------------------- @@ -402,6 +489,35 @@ export class DurableSubscriptionStore extends PuterStore { return Number(row?.total ?? 0); } + /** + * The holder's live rows, which is what a revocation has to consider. Same + * index the listing and the quota use — passing `appUid` narrows to one + * app's rows, which is what a grant made to that app can have authorized. + * + * Bounded by the per-account quota, so the whole set fits one read. + */ + async listActiveForHolder( + holderUserId: number, + appUid: string | null, + ): Promise { + const where = [ + '`holder_user_id` = ?', + '`suspended_at` IS NULL', + this.#unexpiredClause(), + ]; + const params: unknown[] = [holderUserId, nowSeconds()]; + if (appUid !== null) { + where.push('`app_uid` = ?'); + params.push(appUid); + } + 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], + ); + return rows.map(toRow); + } + /** * 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 @@ -445,6 +561,41 @@ export class DurableSubscriptionStore extends PuterStore { return rows.map(toRow); } + async #listSuspendedBefore( + cutoff: number, + batchSize: number, + ): Promise { + const limit = Math.max(1, Math.floor(batchSize)); + const rows = await this.clients.db.read( + `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` ` + + 'WHERE `suspended_at` IS NOT NULL AND `suspended_at` <= ? ' + + 'ORDER BY `id` LIMIT ?', + [cutoff, limit], + ); + return rows.map(toRow); + } + + /** Delete a batch and leave no region delivering against any of it. */ + async #reap(rows: readonly DurableSubscription[]): Promise { + if (rows.length === 0) return 0; + + await this.clients.db.write( + `DELETE FROM \`${TABLE}\` WHERE \`sub_id\` IN ` + + `(${rows.map(() => '?').join(', ')})`, + rows.map((row) => row.subId), + ); + + const owners = new Set(); + for (const row of rows) { + await this.stores.eventSubscription.dropDurable(row); + // Whatever was still owed to a row that no longer exists. + await this.stores.pendingDelivery.purge(row.subId).catch(() => {}); + owners.add(row.ownerUserId); + } + for (const ownerUserId of owners) await this.#bump(ownerUserId); + return rows.length; + } + /** * 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 diff --git a/src/backend/stores/events/EventSubscriptionStore.ts b/src/backend/stores/events/EventSubscriptionStore.ts index 3203145ce..af5ecbf2b 100644 --- a/src/backend/stores/events/EventSubscriptionStore.ts +++ b/src/backend/stores/events/EventSubscriptionStore.ts @@ -220,6 +220,59 @@ export class EventSubscriptionStore extends PuterStore { }; } + /** + * Move one session row onto a different anchor, keeping its id and its + * socket. Not `remove` then `add`: this is the same subscription, so it + * must not be turned away by the per-connection cap it already occupies a + * slot in, and the new anchor may sit in a different owner's keyspace. + */ + async reanchorSession( + previous: SessionSubscription, + next: SessionSubscription, + ): Promise { + await this.#dropRefs(previous.holderUserId, previous.socketId, [ + { + ownerUserId: previous.ownerUserId, + token: previous.token, + subId: previous.subId, + }, + ]); + + const rows = this.clients.redis.pipeline(); + const key = tokenKey(next.ownerUserId, next.token); + rows.hset(key, next.subId, JSON.stringify(next)); + rows.expire(key, SESSION_SUBSCRIPTION_TTL_SECONDS); + rows.sadd(watchedKey(next.ownerUserId), next.token); + rows.expire( + watchedKey(next.ownerUserId), + SESSION_SUBSCRIPTION_TTL_SECONDS, + ); + await rows.exec(); + + const holder = this.clients.redis.pipeline(); + holder.sadd( + socketKey(next.holderUserId, next.socketId), + socketRef({ + ownerUserId: next.ownerUserId, + token: next.token, + subId: next.subId, + }), + ); + holder.expire( + socketKey(next.holderUserId, next.socketId), + SESSION_SUBSCRIPTION_TTL_SECONDS, + ); + await holder.exec(); + + const owners = new Set([previous.ownerUserId, next.ownerUserId]); + return Promise.all( + [...owners].map(async (userId) => ({ + userId, + generation: await this.bumpGeneration(userId), + })), + ); + } + /** * Drop everything a socket held. Runs on disconnect; the TTL is what covers * the disconnect that never runs. One socket can hold rows in several diff --git a/src/backend/stores/notification/NotificationStore.test.js b/src/backend/stores/notification/NotificationStore.test.js index 1b16cb0d2..923014ea3 100644 --- a/src/backend/stores/notification/NotificationStore.test.js +++ b/src/backend/stores/notification/NotificationStore.test.js @@ -133,7 +133,7 @@ describe('NotificationStore', () => { userId: u.id, value: {}, type: 'app.events.ended', - audience: 'developer', + audience: 'app-user', appUid, }); @@ -148,7 +148,7 @@ describe('NotificationStore', () => { appUid: null, }); expect(byType['app.events.ended']).toEqual({ - audience: 'developer', + audience: 'app-user', appUid, }); }); diff --git a/src/docs/src/rate-limits-and-quotas.md b/src/docs/src/rate-limits-and-quotas.md index 1efbc54eb..e1943a54b 100644 --- a/src/docs/src/rate-limits-and-quotas.md +++ b/src/docs/src/rate-limits-and-quotas.md @@ -170,6 +170,7 @@ One write can reach many subscriptions, so events are bounded on both halves: ho | Deliveries per minute, per subscription | 600 | | Acknowledgements per minute | 600 | | Undelivered deliveries per subscription | 10,000 | +| Suspended subscriptions kept for | 30 days | 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. @@ -177,6 +178,8 @@ The 51st subscription on one connection, and the 501st durable subscription on o A durable subscription may carry a `context`: JSON that is stored with it and handed to its handler on every delivery, capped at **4 KB** and rejected over that with `events_context_too_large`. Listings never return it. An app sees and revokes only the subscriptions it created; a session acting for the account sees them all, including ones left behind by an app that has since been removed. +**A subscription can end without you unsubscribing.** Access is re-checked against the stored permission on every delivery, so a share that is taken back stops delivering immediately; the subscription is then *suspended*, with `suspendedAt` and `suspendedReason: 'permission_revoked'` in `list` and a notification to whoever holds it. The same happens to every subscription an app holds for you when you withdraw that app's access. Re-granting does not bring a suspended subscription back — subscribe again, which is how consent to watch is re-established — and a suspended row is deleted **30 days** after it stops. Deleting the node a subscription is anchored on ends it too, unless the subject named a path or a pattern, in which case it follows that path up to the nearest folder that still exists and keeps watching, so recreating the path resumes delivery. + Match patterns are compiled once when you subscribe and are capped at **256 characters** and **16 segments**; anything larger is rejected with `invalid_subject_pattern`. `**` crosses directories and costs no more than `*`. **Deliveries are coalesced over 250 ms per subject.** A multipart upload, a save loop, or a recursive delete is one thing the user did, and it arrives as one event carrying the newest state rather than as one event per write. Two different files in the same window are two deliveries.