fix: harden events (#3813)

* fix(events): rename carries from, one retry on deploy-timeout, forward-path counters

- FSService.rename passes the pre-rename path so a folder subscription
  sees op move with from, like a real move does
- deploy-timeout from the dispatcher gets one retry with the deployed
  header and no second upload (ALREADY_DEPLOYED_MISS_REASONS)
- OTel counters events.forward.sent/received and events.single.attempt
- cross-app KV subscribe error hints at the three-segment parse
- docs: lease is 60 s, kv prefix example is fully qualified, move covers
  rename

* feat(events): forward session subscriptions across regions, fast bumps, worker session cleanup

- session (onLocal) subscriptions now receive writes committed in other
  regions: a transition-maintained remote-watch index (ev:sc / ev:rw),
  watch/event forward items, replay through dispatchForwarded against
  session rows only; events.forwardSession=false is the kill switch
- subscription and presence generation bumps also ride the addressed
  forward channel (kind bump) so a peer sees a new durable row within
  a queue window; the webhook fan stays as backstop
- workers.destroy revokes every holder's events:handlers session; app
  deletion reaps the app's rows, backlog and handlers and revokes the
  sessions; an hourly sweep revokes sessions whose app is gone
- docs: cross-region latency, per-region caps footnote, session
  lifecycle

src/backend/clients/events/EventsWorkerInvokerClient.test.ts

@@ -216,6 +216,27 @@ const alwaysMissingFetch = (calls: unknown[]): FetchImpl =>
         });
     }) as unknown as FetchImpl;
 
+/**
+ * Answers `deploy-timeout` once, then 200 once it sees the deployed header —
+ * the shape of a script that finished propagating between the two calls.
+ */
+const deployTimeoutThenOkFetch = (
+    headersSeen: Array<Record<string, string>>,
+): FetchImpl =>
+    (async (_url: string, init?: RequestInit) => {
+        const headers = (init?.headers ?? {}) as Record<string, string>;
+        headersSeen.push(headers);
+        if (headers[EVENTS_DEPLOYED_HEADER] === '1')
+            return new Response(null, {
+                status: 200,
+                headers: { [EVENTS_HANDLED_HEADER]: '1' },
+            });
+        return new Response(null, {
+            status: 503,
+            headers: { [EVENTS_DISPATCH_ERROR_HEADER]: 'deploy-timeout' },
+        });
+    }) as unknown as FetchImpl;
+
 describe('EventsWorkerInvokerClient deploy-on-miss', () => {
     it('deploys the script and retries once with the deployed header', async () => {
         const headersSeen: Array<Record<string, string>> = [];
@@ -326,4 +347,64 @@ describe('EventsWorkerInvokerClient deploy-on-miss', () => {
             error: 'dispatcher: forbidden (403)',
         });
     });
+
+    it('retries a deploy-timeout with the deployed header instead of redeploying', async () => {
+        const headersSeen: Array<Record<string, string>> = [];
+        const client = makeClient();
+        client.setTransport(
+            new DispatcherInvokeTransport('http://dispatcher', 's', {
+                fetchImpl: deployTimeoutThenOkFetch(headersSeen),
+            }),
+        );
+        const deployCalls: unknown[] = [];
+        client.setMissHandler(async () => {
+            deployCalls.push(null);
+            return 'deployed';
+        });
+
+        const result = await client.invoke(REQUEST);
+
+        // The script was already deployed once — the one that led to this
+        // deploy-timeout answer — so this client must not add a second.
+        expect(deployCalls).toHaveLength(0);
+        expect(headersSeen).toHaveLength(2);
+        expect(headersSeen[0][EVENTS_DEPLOYED_HEADER]).toBeUndefined();
+        expect(headersSeen[1][EVENTS_DEPLOYED_HEADER]).toBe('1');
+        expect(result).toEqual({ outcome: 'settled', status: 200 });
+    });
+
+    it('gives up after one retry when propagation is still not done', async () => {
+        const calls: unknown[] = [];
+        const client = makeClient();
+        client.setTransport(
+            new DispatcherInvokeTransport('http://dispatcher', 's', {
+                fetchImpl: (async () => {
+                    calls.push(null);
+                    return new Response(null, {
+                        status: 503,
+                        headers: {
+                            [EVENTS_DISPATCH_ERROR_HEADER]: 'deploy-timeout',
+                        },
+                    });
+                }) as unknown as FetchImpl,
+            }),
+        );
+        const deployCalls: unknown[] = [];
+        client.setMissHandler(async () => {
+            deployCalls.push(null);
+            return 'deployed';
+        });
+
+        const result = await client.invoke(REQUEST);
+
+        // The first send plus the one retry — never a third, and never a
+        // deploy. The delivery comes back on its own backoff instead.
+        expect(calls).toHaveLength(2);
+        expect(deployCalls).toHaveLength(0);
+        expect(result).toEqual({
+            outcome: 'retriable',
+            status: null,
+            error: 'dispatcher: deploy-timeout (503)',
+        });
+    });
 });

src/backend/clients/events/EventsWorkerInvokerClient.ts

@@ -78,15 +78,23 @@ export const EVENTS_DEPLOYED_HEADER = 'x-puter-events-deployed';
 /**
  * Dispatch-error reasons this backend can resolve itself by deploying, rather
  * than wait on the rehydrate callback finding a backend behind the dispatcher's
- * own hostname.
+ * own hostname. `deploy-timeout` is not one of them: it is only ever answered
+ * after a deploy that succeeded, so the script exists and another upload of it
+ * cannot make the namespace see it any sooner.
  */
 const SELF_DEPLOYABLE_MISS_REASONS = new Set([
     'missing',
     'deploy-failed',
-    'deploy-timeout',
     'deploy-error',
 ]);
 
+/**
+ * Dispatch-error reasons where the deploy already happened and only propagation
+ * is still catching up — one prompt retry with the deployed header is worth it,
+ * without spending another upload on it.
+ */
+const ALREADY_DEPLOYED_MISS_REASONS = new Set(['deploy-timeout']);
+
 /** What a self-deploy attempt did, and to which app/script it applies. */
 export type WorkerMissHandler = (
     appUid: string,
@@ -219,23 +227,29 @@ export class EventsWorkerInvokerClient extends PuterClient {
         };
 
         let result = await transport.send(call);
-        if (
-            result.status === null &&
-            this.#missHandler &&
-            result.dispatchReason &&
-            SELF_DEPLOYABLE_MISS_REASONS.has(result.dispatchReason)
-        ) {
-            const outcome = await this.#missHandler(
-                request.appUid,
-                request.script,
-            );
-            if (outcome !== 'deployed')
-                return {
-                    outcome: 'retriable',
-                    status: null,
-                    error: `deploy: ${outcome}`,
-                };
-            result = await transport.send({ ...call, deployed: true });
+        if (result.status === null && result.dispatchReason) {
+            if (
+                this.#missHandler &&
+                SELF_DEPLOYABLE_MISS_REASONS.has(result.dispatchReason)
+            ) {
+                const outcome = await this.#missHandler(
+                    request.appUid,
+                    request.script,
+                );
+                if (outcome !== 'deployed')
+                    return {
+                        outcome: 'retriable',
+                        status: null,
+                        error: `deploy: ${outcome}`,
+                    };
+                result = await transport.send({ ...call, deployed: true });
+            } else if (
+                ALREADY_DEPLOYED_MISS_REASONS.has(result.dispatchReason)
+            ) {
+                // No deploy to trigger — just give propagation the one retry
+                // it usually needs, without spending an upload on it.
+                result = await transport.send({ ...call, deployed: true });
+            }
         }
 
         const { status, handled, error } = result;

src/backend/services/events/EventForwardService.ts

@@ -34,10 +34,14 @@ import {
     PeerForwardQueue,
     type ForwardAck,
     type ForwardBatch,
+    type ForwardBump,
     type ForwardDelivery,
+    type ForwardEvent,
     type ForwardItem,
     type ForwardReply,
+    type ForwardWatch,
 } from './forwardQueue.js';
+import { forwardReceived, forwardSent, sessionForward } from './metrics.js';
 import { PresenceCache, remoteRegions } from './presenceCache.js';
 import type { DeliverableEvent, GapMarker } from './registry.js';
 
@@ -164,6 +168,18 @@ export class EventForwardService extends PuterService {
         );
     }
 
+    /**
+     * Whether session (`onLocal`) subscriptions are forwarded across regions.
+     * On wherever peers exist, since it is what makes the documented `onLocal`
+     * promise true; a separate switch from {@link active} only so a deployment
+     * can hold back the standing per-write index it adds. Explicitly `false` ⇒
+     * no announcements, no `ev:rw` writes on either side, and
+     * `dispatchFs`/`dispatchKv`'s remote arm is never forwarded against.
+     */
+    get forwardSessionActive(): boolean {
+        return this.active && this.config.events?.forwardSession !== false;
+    }
+
     /** What this deployment calls itself in a presence row. */
     get region(): string {
         return this.services.broadcast.regionId;
@@ -369,6 +385,90 @@ export class EventForwardService extends PuterService {
         return this.services.broadcast.addressablePeers.includes(region);
     }
 
+    /**
+     * Tell every peer this region now has (or no longer has) a session watcher
+     * on one anchor token. Fires only on the transition — the first session row
+     * for a token, or the last one going — never per subscribe.
+     */
+    announceWatch(
+        ownerUserId: number,
+        token: string,
+        op: 'add' | 'drop',
+    ): void {
+        if (!this.forwardSessionActive) return;
+        for (const region of this.services.broadcast.addressablePeers) {
+            forwardSent.add(1, {
+                from: this.region,
+                to: region,
+                class: 'watch',
+            });
+            const item: ForwardWatch = {
+                kind: 'watch',
+                op,
+                userId: ownerUserId,
+                token,
+            };
+            this.#queueFor().push(region, item);
+        }
+    }
+
+    /**
+     * Replay one committed change against named regions' session rows. Only
+     * regions that announced a session watcher on one of the event's tokens are
+     * named — see `EventsService#dispatchFs`/`#dispatchKv`, which read the
+     * remote-watch index this answers to.
+     */
+    forwardEvent(
+        regions: readonly string[],
+        item: Omit<ForwardEvent, 'kind' | 'sessionOnly' | 'hop'>,
+    ): void {
+        if (!this.forwardSessionActive) return;
+        for (const region of regions) {
+            if (!this.isPeer(region)) continue;
+            forwardSent.add(1, {
+                from: this.region,
+                to: region,
+                class: 'session',
+            });
+            this.#queueFor().push(region, {
+                ...item,
+                kind: 'event',
+                sessionOnly: true,
+                hop: 1,
+            });
+        }
+    }
+
+    /**
+     * Fan a subscription-set or presence generation bump to every peer over the
+     * addressed channel, so a cold region there marks itself so within one
+     * queue window instead of waiting for the slower all-peers webhook. That
+     * webhook stays as the backstop for a peer that drops this, or is still
+     * running old code that does not know the `bump` kind.
+     */
+    announceGeneration(
+        bump: { userId: number; generation: number },
+        durable: boolean,
+        scope: 'subscription' | 'presence' = 'subscription',
+    ): void {
+        if (!this.active) return;
+        for (const region of this.services.broadcast.addressablePeers) {
+            forwardSent.add(1, {
+                from: this.region,
+                to: region,
+                class: 'bump',
+            });
+            const item: ForwardBump = {
+                kind: 'bump',
+                userId: bump.userId,
+                generation: bump.generation,
+                scope,
+                durable,
+            };
+            this.#queueFor().push(region, item);
+        }
+    }
+
     /**
      * Regions other than this one holding a socket for the pair, read through
      * the generation-keyed cache and narrowed to regions this deployment can
@@ -425,6 +525,10 @@ export class EventForwardService extends PuterService {
      */
     async receive(batch: ForwardBatch): Promise<ForwardReply> {
         const items = batch.items ?? [];
+        forwardReceived.add(items.length, {
+            from: batch.from ?? 'unknown',
+            to: this.region,
+        });
 
         for (const item of items) {
             if (item.kind !== 'delivery') continue;
@@ -435,6 +539,56 @@ export class EventForwardService extends PuterService {
             }
         }
 
+        for (const item of items) {
+            if (item.kind !== 'watch') continue;
+            // Turned off here means the index is not kept here either: a peer
+            // still running the announce has nothing to forward against.
+            if (!this.forwardSessionActive) continue;
+            try {
+                await this.stores.eventSubscription.noteRemoteWatch(
+                    item.userId,
+                    item.token,
+                    batch.from,
+                    item.op,
+                );
+            } catch (err) {
+                console.warn('[events] remote-watch note failed', err);
+            }
+        }
+
+        for (const item of items) {
+            if (item.kind !== 'bump') continue;
+            if (item.scope === 'presence') {
+                this.#cache.bump(item.userId);
+                continue;
+            }
+            this.services.events.invalidateUser(item.userId, {
+                rebuild: item.durable,
+            });
+        }
+
+        // Deliveries keep the batch's order (see the delivery loop above); a
+        // raw event replayed against session rows runs through the same
+        // region-local dispatch path a local write does, so it is walked the
+        // same way rather than under the ack's concurrency bound.
+        const noWatch: Array<{ userId: number; token: string }> = [];
+        for (const item of items) {
+            if (item.kind !== 'event') continue;
+            try {
+                const { matched, tokens } =
+                    await this.services.events.dispatchForwarded(item);
+                sessionForward.add(1, {
+                    from: batch.from,
+                    result: matched ? 'matched' : 'no-rows',
+                });
+                if (!matched)
+                    for (const token of tokens)
+                        noWatch.push({ userId: item.ownerUserId, token });
+            } catch (err) {
+                console.warn('[events] forwarded session event failed', err);
+            }
+        }
+
         const acks = items.filter(
             (item): item is ForwardAck => item.kind === 'ack',
         );
@@ -487,7 +641,10 @@ export class EventForwardService extends PuterService {
                 );
         });
 
-        return noSocket.length > 0 ? { noSocket } : {};
+        const reply: ForwardReply = {};
+        if (noSocket.length > 0) reply.noSocket = noSocket;
+        if (noWatch.length > 0) reply.noWatch = noWatch;
+        return reply;
     }
 
     /**
@@ -526,6 +683,11 @@ export class EventForwardService extends PuterService {
 
     #send(region: string, delivery: ForwardableDelivery): void {
         if (!this.isPeer(region)) return;
+        forwardSent.add(1, {
+            from: this.region,
+            to: region,
+            class: delivery.ackRequired ? 'single' : 'broadcast',
+        });
         const item: ForwardDelivery = {
             kind: 'delivery',
             userId: delivery.holderUserId,
@@ -566,6 +728,16 @@ export class EventForwardService extends PuterService {
 
         for (const missing of reply?.noSocket ?? [])
             await this.#repair(peerId, missing.userId, missing.appUid);
+
+        // A peer answering "I hold no session for this token" is
+        // authoritative the same way `noSocket` is — this region's own
+        // remote-watch entry for it is stale, so it stops sending there.
+        for (const stale of reply?.noWatch ?? [])
+            await this.stores.eventSubscription
+                .noteRemoteWatch(stale.userId, stale.token, peerId, 'drop')
+                .catch((err: unknown) => {
+                    console.warn('[events] remote-watch repair failed', err);
+                });
     }
 
     /**
@@ -675,6 +847,9 @@ export class EventForwardService extends PuterService {
                 { userId, generation },
                 {},
             );
+            // Addressed alongside the webhook emit above, which stays as the
+            // backstop for a peer that drops this or runs old code.
+            this.announceGeneration({ userId, generation }, false, 'presence');
         } catch (err) {
             this.#cache.bump(userId);
             console.warn('[events] presence generation bump failed', err);

src/backend/services/events/EventsService.test.ts

@@ -294,6 +294,9 @@ const buildService = (
                 fanOut: async () => undefined,
                 handOff: () => undefined,
                 relayAck: () => undefined,
+                announceWatch: () => undefined,
+                forwardEvent: () => undefined,
+                announceGeneration: () => undefined,
             },
             socket: {
                 send: vi.fn(async (spec: { socket?: string }, _key, data) => {
@@ -641,7 +644,9 @@ describe('what a dispatch costs', () => {
 
         await dispatch(elsewhere);
 
-        expect(commands).toEqual(['smismember']);
+        // One pipeline: the local membership test alongside the remote-watch
+        // read, still one round trip.
+        expect(commands).toEqual(['pipeline']);
         expect(sent).toEqual([]);
     });
 
@@ -653,8 +658,9 @@ describe('what a dispatch costs', () => {
 
         await dispatch(file);
 
-        // The membership test, then one pipelined read of the one hit.
-        expect(commands).toEqual(['smismember', 'pipeline']);
+        // The membership + remote-watch pipeline, then one pipelined read of
+        // the one hit.
+        expect(commands).toEqual(['pipeline', 'pipeline']);
     });
 
     it('walks the tree only for a user who has subscriptions', async () => {
@@ -869,7 +875,7 @@ describe('cold-region rebuild concurrency', () => {
             // Not being able to tell must resolve as "nothing subscribed"
             // rather than hang, and must not leave the in-flight lookup
             // wedged for every dispatch after it.
-            await expect(dispatch(file)).resolves.toBeUndefined();
+            await expect(dispatch(file)).resolves.toBe(false);
 
             await subscribe(`fs:${documents.uid}`);
             await dispatch(file);
@@ -1541,7 +1547,8 @@ describe('failure containment', () => {
         const { file } = seedTree();
         vi.spyOn(store, 'userHasAny').mockRejectedValue(new Error('down'));
 
-        await expect(dispatch(file)).resolves.toBeUndefined();
+        // Not being able to tell resolves as "nothing subscribed".
+        await expect(dispatch(file)).resolves.toBe(false);
         expect(sent).toEqual([]);
     });
 
@@ -1555,7 +1562,9 @@ describe('failure containment', () => {
             }
         ).services.socket.send.mockRejectedValue(new Error('no socket'));
 
-        await expect(dispatch(file)).resolves.toBeUndefined();
+        // The row still matched and was routed; the socket failure is caught
+        // asynchronously downstream of the return.
+        await expect(dispatch(file)).resolves.toBe(true);
         await vi.advanceTimersByTimeAsync(EVENTS_COALESCE_WINDOW_MS + 1);
         expect(delivered).toHaveLength(1);
     });
@@ -1803,7 +1812,9 @@ describe('what a kv dispatch costs', () => {
 
         await dispatchKv(['elsewhere']);
 
-        expect(commands).toEqual(['smismember']);
+        // One pipeline: the local membership test alongside the remote-watch
+        // read, still one round trip.
+        expect(commands).toEqual(['pipeline']);
         expect(sent).toEqual([]);
     });
 
@@ -1815,7 +1826,7 @@ describe('what a kv dispatch costs', () => {
 
         await dispatchKv(keys);
 
-        expect(commands).toEqual(['smismember', 'pipeline']);
+        expect(commands).toEqual(['pipeline', 'pipeline']);
     });
 });
 

src/backend/services/events/EventsService.ts

@@ -143,7 +143,7 @@ import {
 } from './authorization.js';
 import { DeliveryCoalescer } from './coalescer.js';
 import { forwardTarget } from './EventForwardService.js';
-import type { ForwardDelivery } from './forwardQueue.js';
+import type { ForwardDelivery, ForwardEvent } from './forwardQueue.js';
 import {
     DELIVERY_USAGE_TYPES,
     EVENTS_COSTS,
@@ -203,6 +203,7 @@ import {
     type SubjectOp,
 } from './subjects.js';
 import { backlogPolicyFor, isResumable } from './suspension.js';
+import { singleAttempt } from './metrics.js';
 import type {
     EventsInvokeTransport,
     WorkerMissHandler,
@@ -543,6 +544,26 @@ export interface FsDispatchOptions {
         path: string;
         ancestors: () => Promise<ReadonlyArray<{ uid: string; path: string }>>;
     };
+    /**
+     * Set by {@link EventsService#dispatchForwarded}: a replay of another
+     * region's committed change, evaluated against session rows only — durable
+     * rows already crossed by row, and re-forwarding would loop.
+     */
+    forwarded?: true;
+    /** Carried over from the emitting region so both copies share one id/ts. */
+    id?: string;
+    ts?: number;
+}
+
+/** What a `dispatchKv` call site can supply beyond the bus payload. */
+export interface KvDispatchOptions {
+    /** Who performed the write, for the `self` flag. */
+    actingUserId?: number;
+    /** See {@link FsDispatchOptions.forwarded}. */
+    forwarded?: true;
+    /** Carried over from the emitting region so both copies share one id/ts. */
+    id?: string;
+    ts?: number;
 }
 
 /** One persisted notification, as the bus reports it. */
@@ -593,6 +614,14 @@ const EXPIRY_MAX_BATCHES = 50;
  */
 const ENDED_SUBJECTS_LISTED = 20;
 
+/**
+ * Worker sessions one page of the stray-session sweep reads, and pages one pass
+ * takes. Bounded low: the sweep is a safety net for sessions no API path could
+ * reach any more, not a bulk cleanup, and it runs every hour.
+ */
+const WORKER_SESSION_SWEEP_BATCH = 500;
+const WORKER_SESSION_SWEEP_MAX_PAGES = 2;
+
 // -- Owed deliveries --------------------------------------------------
 
 /**
@@ -907,6 +936,26 @@ const deliverable = (row: DispatchSubscription): boolean => {
 const isSingle = (row: DispatchSubscription): boolean =>
     row.durable === true && row.delivery === 'single';
 
+/**
+ * The three fields a forwarded copy needs and nothing else: `fsProject` and
+ * `fsTokens` read `uid`/`path`, `#eventDescriptor` reads `{uid, path}`, and
+ * `dispatchFs` reads `entry.userId` as the owner. Nothing else about the row
+ * crosses a region.
+ */
+const slimFsEntry = (
+    entry: FSEntry,
+): { uid: string; path: string; userId: number; isDir?: boolean } => ({
+    uid: entry.uid,
+    path: entry.path,
+    userId: entry.userId,
+    ...(entry.isDir !== undefined ? { isDir: entry.isDir } : {}),
+});
+
+/** Every region a set of matched remote tokens named, deduplicated. */
+const regionsIn = (remote: ReadonlyMap<string, string[]>): string[] => [
+    ...new Set([...remote.values()].flat()),
+];
+
 /**
  * `row.anchorPath` is the anchor's path at subscribe time, not now — a rename
  * or move leaves it stale, and matching against it would silently drop every
@@ -1199,6 +1248,35 @@ export class EventsService extends PuterService {
             });
         });
 
+        // A deleted app's rows can never deliver again — the grant identity
+        // they are re-checked under no longer resolves — so they are torn
+        // down rather than left to cost their holder an anchor slot forever.
+        // Best-effort: `#emitAppChanged` is fire-and-forget, so this must not
+        // be able to fail the delete that triggered it.
+        this.clients.event.on('app.changed', async (_key, data) => {
+            const {
+                app_uid: appUid,
+                action,
+                old_app: oldApp,
+            } = (data ?? {}) as {
+                app_uid?: string;
+                action?: string;
+                old_app?: { owner_user_id?: unknown };
+            };
+            if (action !== 'deleted' || !appUid) return;
+            const ownerUserId = Number(oldApp?.owner_user_id);
+            try {
+                await this.settleDeletedApp(
+                    appUid,
+                    Number.isFinite(ownerUserId) && ownerUserId > 0
+                        ? ownerUserId
+                        : undefined,
+                );
+            } catch (err) {
+                console.warn('[events] deleted-app settle failed', err);
+            }
+        });
+
         // The KV store is a store, so the bus is the only seam it has to reach
         // a service. Same posture as the FS hook: post-commit, and nothing here
         // can fail the write that produced it.
@@ -1980,7 +2058,14 @@ export class EventsService extends PuterService {
         } finally {
             // A pass that stops partway may still have taken the last handler
             // with it, and the announce is what a rent listener stops on.
-            if (removed > 0) await this.#maybeAnnounceWorkerDestroy(appUid);
+            if (removed > 0) {
+                await this.#maybeAnnounceWorkerDestroy(appUid);
+                // The worker is gone; nothing should still be able to invoke
+                // under the session it was running deliveries with. Rows stay
+                // suspended `handler_not_found` — publishing again mints a
+                // fresh session through the unchanged `#mintSubscriberToken`.
+                await this.#revokeWorkerSessions(appUid);
+            }
         }
         return { appUid, removed, suspended };
     }
@@ -2440,6 +2525,51 @@ export class EventsService extends PuterService {
         );
     }
 
+    /**
+     * Revoke `events:handlers` worker sessions whose app is already gone — the
+     * safety net for whatever `workers.destroy` and app deletion did not catch
+     * (a caller that skipped both, or a row stranded before either hook
+     * existed). Scoped to this one worker name; the same strand for
+     * `WorkerDriver`-minted sessions is a separate finding, not covered here.
+     */
+    async sweepStrandedWorkerSessions(): Promise<number> {
+        if (!this.enabled) return 0;
+
+        let revoked = 0;
+        let afterId = 0;
+        for (let page = 0; page < WORKER_SESSION_SWEEP_MAX_PAGES; page++) {
+            const rows = await this.stores.session.listWorkerSessions({
+                workerName: EVENTS_WORKER_SESSION_NAME,
+                afterId,
+                limit: WORKER_SESSION_SWEEP_BATCH,
+            });
+            if (rows.length === 0) break;
+            afterId = rows[rows.length - 1].id;
+
+            const appUids = [
+                ...new Set(rows.map((row) => row.appUid).filter(Boolean)),
+            ] as string[];
+            const apps = await this.stores.app.getByUids(appUids);
+
+            for (const row of rows) {
+                if (apps.has(row.appUid)) continue;
+                try {
+                    await this.services.auth.revokeSession(row.uuid);
+                    revoked += 1;
+                } catch (err) {
+                    console.warn(
+                        '[events] could not revoke a stranded worker session',
+                        row.uuid,
+                        err,
+                    );
+                }
+            }
+
+            if (rows.length < WORKER_SESSION_SWEEP_BATCH) break;
+        }
+        return revoked;
+    }
+
     /**
      * Put back subscriptions a restored balance releases. Lazy on purpose: a
      * top-up is not something this service hears about, and coupling delivery
@@ -2679,10 +2809,19 @@ export class EventsService extends PuterService {
 
     /**
      * The other half of {@link #maybeAnnounceWorkerCreate}, for a 1→0
-     * transition.
+     * transition. `ownerUserId`, when given, is used in place of an app-row
+     * lookup — needed when the app row is already gone, as it is by the time
+     * `app.changed`'s `deleted` action reaches {@link settleDeletedApp}.
      */
-    async #maybeAnnounceWorkerDestroy(appUid: string): Promise<void> {
-        await this.#emitWorkerLifecycle('events.worker.destroy', appUid);
+    async #maybeAnnounceWorkerDestroy(
+        appUid: string,
+        ownerUserId?: number,
+    ): Promise<void> {
+        await this.#emitWorkerLifecycle(
+            'events.worker.destroy',
+            appUid,
+            ownerUserId,
+        );
     }
 
     /**
@@ -2699,6 +2838,7 @@ export class EventsService extends PuterService {
     async #emitWorkerLifecycle(
         name: 'events.worker.create' | 'events.worker.destroy',
         appUid: string,
+        knownOwnerUserId?: number,
     ): Promise<void> {
         try {
             // Only a count now on the far side of the transition is one: a
@@ -2707,11 +2847,15 @@ export class EventsService extends PuterService {
             if (name === 'events.worker.create' ? count === 0 : count > 0)
                 return;
 
-            const app = await this.stores.app.getByUid(appUid);
-            const ownerUserId = Number(
-                (app as { owner_user_id?: unknown } | null)?.owner_user_id,
-            );
-            if (!Number.isFinite(ownerUserId) || ownerUserId <= 0) return;
+            let ownerUserId = knownOwnerUserId;
+            if (ownerUserId === undefined) {
+                const app = await this.stores.app.getByUid(appUid);
+                ownerUserId = Number(
+                    (app as { owner_user_id?: unknown } | null)?.owner_user_id,
+                );
+            }
+            if (!Number.isFinite(ownerUserId) || (ownerUserId as number) <= 0)
+                return;
             const owner = await this.stores.user.getById(ownerUserId);
             if (!owner) return;
 
@@ -3138,16 +3282,16 @@ export class EventsService extends PuterService {
         key: EventKey,
         entry: FSEntry,
         options: FsDispatchOptions = {},
-    ): Promise<void> {
-        if (!this.enabled) return;
+    ): Promise<boolean> {
+        if (!this.enabled) return false;
 
         const subject = lookupFsSubject(key);
-        if (!subject) return;
+        if (!subject) return false;
 
         const ownerUserId = entry?.userId;
-        if (typeof ownerUserId !== 'number') return;
+        if (typeof ownerUserId !== 'number') return false;
 
-        if (!(await this.#userHasAny(ownerUserId))) return;
+        if (!(await this.#userHasAny(ownerUserId))) return false;
 
         const ancestors = options.ancestors ? await options.ancestors() : [];
         const movedFrom = options.movedFrom
@@ -3161,21 +3305,49 @@ export class EventsService extends PuterService {
             entry,
             ancestors,
             movedFrom,
-            id: randomUUID(),
-            ts: Date.now(),
+            id: options.id ?? randomUUID(),
+            ts: options.ts ?? Date.now(),
         };
 
-        const watched = await this.stores.eventSubscription.watchedTokens(
-            ownerUserId,
-            subject.tokens(context),
-        );
-        if (watched.length === 0) return;
+        const { local, remote } =
+            await this.stores.eventSubscription.watchedFor(
+                ownerUserId,
+                subject.tokens(context),
+            );
+        if (local.length === 0 && remote.size === 0) return false;
 
-        const rows = await this.stores.eventSubscription.getForTokens(
+        // Ship first: the far side should not wait on this region's ACL
+        // re-checks. A forwarded copy is never re-forwarded — its rows are
+        // already session-local by construction, and `hop` is not read.
+        if (!options.forwarded && remote.size > 0)
+            this.services.eventForward.forwardEvent(regionsIn(remote), {
+                family: 'fs',
+                ownerUserId,
+                actingUserId: options.actingUserId,
+                id: context.id,
+                ts: context.ts,
+                fs: {
+                    key,
+                    entry: slimFsEntry(entry),
+                    ancestors: [...ancestors],
+                    movedFrom: movedFrom && {
+                        path: movedFrom.path,
+                        ancestors: [...movedFrom.ancestors],
+                    },
+                },
+            });
+
+        if (local.length === 0) return false;
+
+        let rows = await this.stores.eventSubscription.getForTokens(
             ownerUserId,
-            watched,
+            local,
         );
-        if (rows.length === 0) return;
+        // A forwarded copy evaluates session rows only: durable rows already
+        // crossed by row (`warmRegion` rebuilds them in every region).
+        if (options.forwarded)
+            rows = rows.filter((row) => row.socketId !== undefined);
+        if (rows.length === 0) return false;
 
         await this.#route(
             subject,
@@ -3190,6 +3362,7 @@ export class EventsService extends PuterService {
         // now gone.
         if (key === 'fs.remove.node')
             await this.#settleDeletedAnchor(context, rows);
+        return true;
     }
 
     /**
@@ -3201,45 +3374,75 @@ export class EventsService extends PuterService {
      * A batch is one bus event over many keys, so the watched-set check is one
      * command for the whole batch rather than one per key.
      */
-    async dispatchKv(input: KvDispatchInput): Promise<void> {
-        if (!this.enabled) return;
+    async dispatchKv(
+        input: KvDispatchInput,
+        options: KvDispatchOptions = {},
+    ): Promise<boolean> {
+        if (!this.enabled) return false;
 
         const subject = lookupKvSubject('kv.mutated');
-        if (!subject) return;
+        if (!subject) return false;
 
         const ownerUserId = input?.userId;
-        if (typeof ownerUserId !== 'number') return;
-        if (!input.keys?.length) return;
+        if (typeof ownerUserId !== 'number') return false;
+        if (!input.keys?.length) return false;
 
-        if (!(await this.#userHasAny(ownerUserId))) return;
+        if (!(await this.#userHasAny(ownerUserId))) return false;
 
         const namespace = parseKvNamespace(input.namespace);
-        if (!namespace) return;
+        if (!namespace) return false;
 
-        const ts = Date.now();
+        const ts = options.ts ?? Date.now();
         const contexts: KvEventContext[] = input.keys.map((kvKey) => ({
             key: 'kv.mutated',
             userUuid: namespace.userUuid,
             appUid: namespace.appUid,
             kvKey,
             op: input.op,
-            id: randomUUID(),
+            // A forwarded replay is always a single key, carrying the
+            // emitter's own id so both copies match.
+            id: options.forwarded && options.id ? options.id : randomUUID(),
             ts,
         }));
 
         const tokensPerKey = contexts.map((context) => subject.tokens(context));
-        const watched = new Set(
-            await this.stores.eventSubscription.watchedTokens(ownerUserId, [
+        const { local, remote } =
+            await this.stores.eventSubscription.watchedFor(ownerUserId, [
                 ...new Set(tokensPerKey.flat()),
-            ]),
-        );
-        if (watched.size === 0) return;
+            ]);
+        if (local.length === 0 && remote.size === 0) return false;
 
-        const rows = await this.stores.eventSubscription.getForTokens(
+        if (!options.forwarded && remote.size > 0)
+            contexts.forEach((context, i) => {
+                const regions = new Set<string>();
+                for (const token of tokensPerKey[i])
+                    for (const region of remote.get(token) ?? [])
+                        regions.add(region);
+                if (regions.size === 0) return;
+                this.services.eventForward.forwardEvent([...regions], {
+                    family: 'kv',
+                    ownerUserId,
+                    actingUserId: options.actingUserId,
+                    id: context.id,
+                    ts: context.ts,
+                    kv: {
+                        userUuid: namespace.userUuid,
+                        appUid: namespace.appUid,
+                        kvKey: context.kvKey,
+                        op: context.op,
+                    },
+                });
+            });
+
+        if (local.length === 0) return false;
+
+        let rows = await this.stores.eventSubscription.getForTokens(
             ownerUserId,
-            [...watched],
+            local,
         );
-        if (rows.length === 0) return;
+        if (options.forwarded)
+            rows = rows.filter((row) => row.socketId !== undefined);
+        if (rows.length === 0) return false;
 
         // Indexed once: a row holds one token, so a key's candidates are the
         // rows under the tokens it enumerated.
@@ -3247,11 +3450,13 @@ export class EventsService extends PuterService {
         for (const row of rows)
             byToken.set(row.token, [...(byToken.get(row.token) ?? []), row]);
 
+        let matchedAny = false;
         for (const [i, context] of contexts.entries()) {
             const forKey = tokensPerKey[i].flatMap(
                 (token) => byToken.get(token) ?? [],
             );
             if (forKey.length === 0) continue;
+            matchedAny = true;
 
             await this.#route(
                 subject,
@@ -3261,6 +3466,87 @@ export class EventsService extends PuterService {
                 (matched) => this.#kvStillAuthorized(matched, namespace.appUid),
             );
         }
+        return matchedAny;
+    }
+
+    /**
+     * Replay a peer's committed change against this region's session rows.
+     * Reusing `dispatchFs`/`dispatchKv` is the point: gate → watched set → rows
+     * → filter → authorize → coalesce → meter → emit, all region-local and all
+     * code already tested. `matched` reports whether this event's tokens were
+     * watched here at all — the index-staleness signal a `false` turns into a
+     * `noWatch` reply — and is independent of whether anything downstream (a
+     * filter, an ACL re-check) actually delivered.
+     */
+    async dispatchForwarded(
+        item: ForwardEvent,
+    ): Promise<{ matched: boolean; tokens: readonly string[] }> {
+        if (!this.enabled) return { matched: false, tokens: [] };
+
+        if (item.family === 'fs' && item.fs) {
+            const subject = lookupFsSubject(item.fs.key);
+            // Only `entry.uid`/`ancestors`/`movedFrom.ancestors` are read to
+            // compute tokens — the slim entry a forward carries has exactly
+            // those, never a full `FSEntry`.
+            const tokens = subject
+                ? subject.tokens({
+                      key: item.fs.key,
+                      entry: item.fs.entry,
+                      ancestors: item.fs.ancestors,
+                      movedFrom: item.fs.movedFrom,
+                      id: item.id,
+                      ts: item.ts,
+                  } as unknown as FsEventContext)
+                : [];
+            const matched = await this.dispatchFs(
+                item.fs.key,
+                item.fs.entry as FSEntry,
+                {
+                    ancestors: async () => item.fs!.ancestors,
+                    movedFrom: item.fs!.movedFrom && {
+                        path: item.fs!.movedFrom.path,
+                        ancestors: async () => item.fs!.movedFrom!.ancestors,
+                    },
+                    actingUserId: item.actingUserId,
+                    forwarded: true,
+                    id: item.id,
+                    ts: item.ts,
+                },
+            );
+            return { matched, tokens };
+        }
+
+        if (item.family === 'kv' && item.kv) {
+            const subject = lookupKvSubject('kv.mutated');
+            const tokens = subject
+                ? subject.tokens({
+                      key: 'kv.mutated',
+                      userUuid: item.kv.userUuid,
+                      appUid: item.kv.appUid,
+                      kvKey: item.kv.kvKey,
+                      op: item.kv.op,
+                      id: item.id,
+                      ts: item.ts,
+                  } as KvEventContext)
+                : [];
+            const matched = await this.dispatchKv(
+                {
+                    userId: item.ownerUserId,
+                    namespace: `v1:${item.kv.userUuid}:${item.kv.appUid}`,
+                    keys: [item.kv.kvKey],
+                    op: item.kv.op,
+                },
+                {
+                    actingUserId: item.actingUserId,
+                    forwarded: true,
+                    id: item.id,
+                    ts: item.ts,
+                },
+            );
+            return { matched, tokens };
+        }
+
+        return { matched: false, tokens: [] };
     }
 
     /**
@@ -3812,6 +4098,43 @@ export class EventsService extends PuterService {
         }
     }
 
+    /**
+     * Retire every holder's `events:handlers` session for one app. A session is
+     * only ever minted for a holder who owns a durable row bound to one of the
+     * app's handlers, so that table is the candidate list — indexed on
+     * `app_uid`, so no new lookup is needed to find them.
+     */
+    async #revokeWorkerSessions(appUid: string): Promise<void> {
+        const holders =
+            await this.stores.durableSubscription.listHolderIdsForApp(appUid);
+        for (const userId of holders)
+            await this.#revokeWorkerSession(userId, appUid);
+    }
+
+    /**
+     * Retire everything a deleted app left behind. Its rows can never deliver
+     * again — the grant identity they are re-checked under no longer resolves,
+     * `#recheck` refuses a row whose identity cannot be resolved — so they are
+     * deleted rather than suspended, unlike `workers.destroy` (a reversible
+     * developer action) which only ever suspends.
+     */
+    async settleDeletedApp(
+        appUid: string,
+        ownerUserId?: number,
+    ): Promise<void> {
+        if (!this.enabled) return;
+
+        const holders =
+            await this.stores.durableSubscription.listHolderIdsForApp(appUid);
+        await this.#sweepInBatches((batchSize) =>
+            this.stores.durableSubscription.reapForApp(appUid, batchSize),
+        );
+        await this.stores.eventHandler.deleteForApp(appUid);
+        await this.#maybeAnnounceWorkerDestroy(appUid, ownerUserId);
+        for (const userId of holders)
+            await this.#revokeWorkerSession(userId, appUid);
+    }
+
     /**
      * Which of a holder's rows one withdrawn grant actually stops.
      *
@@ -4328,11 +4651,18 @@ export class EventsService extends PuterService {
                     bill,
                 });
             }
+            singleAttempt.add(1, {
+                target: region ? 'remote-socket' : 'local-socket',
+                result: 'sent',
+            });
             return false;
         }
 
         // Nowhere to put it yet: the lease is what paces the next attempt.
-        if (!hasWorkerFallback) return false;
+        if (!hasWorkerFallback) {
+            singleAttempt.add(1, { target: 'none', result: 'no-target' });
+            return false;
+        }
 
         const invocation = this.#workerInvocation(row, claimed.event);
         if (!invocation) return false;
@@ -4341,6 +4671,10 @@ export class EventsService extends PuterService {
         // the lease is the backoff — and the budget refusal itself must not
         // spend the one bill this entry gets, nor count against the handler.
         const outcome = await this.#invokeHandler(invocation);
+        singleAttempt.add(1, {
+            target: 'worker',
+            result: outcome ?? 'over-budget',
+        });
         if (outcome === null) return false;
 
         // Only a settled outcome is a delivery: billing and reporting it
@@ -4939,11 +5273,18 @@ export class EventsService extends PuterService {
      * `durable` says whether the table changed. Session rows live in this
      * region's Redis alone, so a peer hearing about one has nothing to rebuild
      * — only a durable bump is worth a primary read over there.
+     *
+     * Also fans the bump over the addressed peer channel — a 25 ms queue window
+     * and a pooled round trip, rather than the ~2 s all-peers webhook batch
+     * above, which stays as the backstop for a dropped addressed item or a peer
+     * running old code. And where the store reported a session-watcher
+     * transition (`bump.announce`), tells every peer to start or stop
+     * forwarding raw events for that token — the whole of how subscribe,
+     * unsubscribe, socket reap and reanchor keep the remote-watch index honest,
+     * since all of them land here.
      */
-    #publishGeneration(
-        { userId, generation }: GenerationBump,
-        durable: boolean,
-    ): void {
+    #publishGeneration(bump: GenerationBump, durable: boolean): void {
+        const { userId, generation } = bump;
         this.#cache.bump(userId, generation);
         try {
             this.clients.event.emit(
@@ -4954,6 +5295,17 @@ export class EventsService extends PuterService {
         } catch {
             // A peer that misses the bump rebuilds on its own next miss.
         }
+        this.services.eventForward.announceGeneration(
+            { userId, generation },
+            durable,
+            'subscription',
+        );
+        for (const announce of bump.announce ?? [])
+            this.services.eventForward.announceWatch(
+                userId,
+                announce.token,
+                announce.op,
+            );
     }
 
     // -- Plumbing ----------------------------------------------------
@@ -5020,6 +5372,19 @@ export class EventsService extends PuterService {
             () => {
                 void this.stores.eventSubscription
                     .refresh(holderUserId, socketId)
+                    .then((reasserted) => {
+                        // The whole of the refresh story for a peer's
+                        // remote-watch entry: re-assert whatever this socket
+                        // still holds, so a lost `ev:rw` field — or its TTL
+                        // simply lapsing — heals within one window rather
+                        // than waiting on the next transition.
+                        for (const { ownerUserId, token } of reasserted)
+                            this.services.eventForward.announceWatch(
+                                ownerUserId,
+                                token,
+                                'add',
+                            );
+                    })
                     .catch(() => {});
             },
             Math.floor((SESSION_SUBSCRIPTION_TTL_SECONDS * 1000) / 3),
@@ -5033,6 +5398,7 @@ export class EventsService extends PuterService {
         const run = () => {
             void this.sweepExpired()
                 .then(() => this.sweepSuspended())
+                .then(() => this.sweepStrandedWorkerSessions())
                 .catch((err) => {
                     console.warn('[events] expiry sweep failed', err);
                 });

src/backend/services/events/appDeletion.integration.test.ts

@@ -0,0 +1,343 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+/**
+ * What deleting an app leaves behind on the events side: its durable rows,
+ * their backlog and its published handlers can never deliver again — the
+ * grant identity they are re-checked under no longer resolves — so
+ * `settleDeletedApp` tears them down on the `app.changed` bus key rather than
+ * leaving them to cost their holder an anchor slot forever. Also the hourly
+ * sweep that catches whatever neither this nor `workers.destroy` reached.
+ */
+
+import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
+import { v4 as uuidv4 } from 'uuid';
+import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js';
+import type { IConfig } from '../../types.js';
+import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js';
+import { EVENTS_WORKER_SESSION_NAME } from './workerRuntime.js';
+
+const BOOT_TIMEOUT_MS = 120_000;
+
+let env: PuterTestEnv;
+let userId: number;
+let otherUserId: number;
+
+beforeAll(async () => {
+    env = await setupPuterTestEnv({
+        events: { enabled: true },
+        unlimitedMetering: true,
+    } as IConfig);
+
+    const user = await env.server.stores.user.getByUsername(
+        env.users.user.username,
+    );
+    userId = user!.id;
+    const other = await env.server.stores.user.getByUsername(
+        env.users.other.username,
+    );
+    otherUserId = other!.id;
+}, BOOT_TIMEOUT_MS);
+
+afterAll(async () => {
+    await env?.shutdown();
+});
+
+beforeEach(async () => {
+    await env.server.clients.db.write('DELETE FROM `event_handlers`', []);
+    await env.server.clients.db.write('DELETE FROM `event_subscriptions`', []);
+    await env.server.clients.db.write('DELETE FROM `apps`', []);
+});
+
+interface ApiResponse {
+    status: number;
+    body: Record<string, unknown>;
+}
+
+const call = async (
+    method: 'GET' | 'POST',
+    path: string,
+    token: string,
+    body?: object,
+): Promise<ApiResponse> => {
+    const response = await fetch(new URL(path, env.apiOrigin), {
+        method,
+        headers: {
+            'content-type': 'application/json',
+            authorization: `Bearer ${token}`,
+        },
+        ...(body ? { body: JSON.stringify(body) } : {}),
+    });
+    return {
+        status: response.status,
+        body: (await response.json()) as Record<string, unknown>,
+    };
+};
+
+/** An app owned by `ownerUserId`, with a token that acts as it for that owner. */
+const makeApp = async (
+    ownerUserId: number,
+): Promise<{ uid: string; token: string }> => {
+    const uid = `app-${uuidv4()}`;
+    await env.server.clients.db.write(
+        'INSERT INTO `apps` (`uid`, `name`, `title`, `index_url`, `owner_user_id`) VALUES (?, ?, ?, ?, ?)',
+        [uid, uid, uid, `https://${uid}.example/`, ownerUserId],
+    );
+    const ownerToken =
+        ownerUserId === userId ? env.users.user.token : env.users.other.token;
+    const { actor } = await env.server.services.auth.authenticate(ownerToken);
+    const token = await env.server.services.auth.getUserAppToken(actor!, uid);
+    return { uid, token };
+};
+
+const SOURCE = 'async ({ event }) => { console.log(event.path); }';
+
+/**
+ * Emit the same event `AppDriver` emits on delete — `old_app` carries the row
+ * as it was, which is where `owner_user_id` comes from once the row itself is
+ * gone.
+ */
+const emitDeleted = async (appUid: string): Promise<void> => {
+    const app = await env.server.stores.app.getByUid(appUid);
+    await env.server.clients.event.emitAndWait(
+        'app.changed',
+        { app_uid: appUid, app: null, old_app: app, action: 'deleted' },
+        {},
+    );
+};
+
+const publish = (
+    token: string,
+    body: { appUid: string; name: string; source: string },
+): Promise<ApiResponse> => call('POST', '/events/handlers/publish', token, body);
+
+/** A durable `single` subscription bound to a published handler, with the
+ * grants a background delivery needs. Returns the subscribed subId. */
+const subscribeBackgroundHandler = async (
+    app: { uid: string; token: string },
+    handlerName: string,
+): Promise<string> => {
+    const anchor = `/${env.users.user.username}/${uuidv4()}`;
+    await env.server.services.fs.mkdir(userId, {
+        path: anchor,
+        createMissingParents: true,
+    });
+    const { actor } = await env.server.services.auth.authenticate(
+        env.users.user.token,
+    );
+    const entry = await env.server.stores.fsEntry.getEntryByPath(anchor);
+    await env.server.services.permission.grantUserAppPermission(
+        actor!,
+        app.uid,
+        `fs:${entry!.uid}:list`,
+    );
+    await env.server.services.permission.grantUserAppPermission(
+        actor!,
+        app.uid,
+        EVENTS_BACKGROUND_PERMISSION,
+    );
+
+    const subscribed = await call('POST', '/events/subscribe', app.token, {
+        subject: `fs:${anchor}`,
+        delivery: 'single',
+        handlerName,
+        targets: ['worker'],
+    });
+    expect(subscribed.status).toBe(200);
+    return subscribed.body.subId as string;
+};
+
+describe('an app deleted while it has events state', () => {
+    it('clears the rows, the backlog and the handlers when the app is deleted', async () => {
+        const app = await makeApp(userId);
+        await publish(app.token, {
+            appUid: app.uid,
+            name: 'ingestUpload',
+            source: SOURCE,
+        });
+        const subId = await subscribeBackgroundHandler(app, 'ingestUpload');
+
+        await env.server.stores.pendingDelivery.enqueue(subId, {
+            id: 'ev-1',
+            subject: `fs:${app.uid}`,
+            op: 'write',
+            uid: 'node-1',
+            path: '/somewhere',
+            self: true,
+            seq: 0,
+            ts: Date.now(),
+        });
+        expect(await env.server.stores.pendingDelivery.depth(subId)).toBe(1);
+
+        await emitDeleted(app.uid);
+
+        const listed = await env.server.stores.durableSubscription.listForHolder(
+            userId,
+            { appUid: app.uid },
+        );
+        expect(listed.items).toEqual([]);
+        expect(await env.server.stores.pendingDelivery.depth(subId)).toBe(0);
+        expect(await env.server.stores.eventHandler.listForApp(app.uid)).toEqual(
+            [],
+        );
+    });
+
+    it('retires every holder session for a deleted app', async () => {
+        const app = await makeApp(userId);
+        await publish(app.token, {
+            appUid: app.uid,
+            name: 'ingestUpload',
+            source: SOURCE,
+        });
+
+        // The account's own row, from the real subscribe path.
+        await subscribeBackgroundHandler(app, 'ingestUpload');
+        // A second holder bound to the same app — the shared-anchor case, not
+        // worth a second app-token dance to reach through the API.
+        await env.server.stores.durableSubscription.create({
+            holderUserId: otherUserId,
+            ownerUserId: userId,
+            appUid: app.uid,
+            subject: `fs:${app.uid}`,
+            token: `fs:${app.uid}`,
+            anchorUid: app.uid,
+            anchorPath: '/shared',
+            match: null,
+            op: null,
+            delivery: 'single',
+            targets: ['worker'],
+            handlerName: 'ingestUpload',
+            context: null,
+            permission: 'list',
+            expiresAt: null,
+        });
+
+        const { actor: ownerActor } = await env.server.services.auth.authenticate(
+            env.users.user.token,
+        );
+        const { actor: otherActor } = await env.server.services.auth.authenticate(
+            env.users.other.token,
+        );
+        const ownerToken = await env.server.services.auth.createWorkerAppToken(
+            ownerActor!,
+            app.uid,
+            EVENTS_WORKER_SESSION_NAME,
+        );
+        const otherToken = await env.server.services.auth.createWorkerAppToken(
+            otherActor!,
+            app.uid,
+            EVENTS_WORKER_SESSION_NAME,
+        );
+
+        await emitDeleted(app.uid);
+
+        for (const token of [ownerToken, otherToken]) {
+            const reauth = await env.server.services.auth.authenticate(token);
+            expect(reauth).toMatchObject({
+                reauth: { reason: 'session_revoked' },
+            });
+        }
+    });
+
+    it('still deletes the app when the settle throws', async () => {
+        const app = await makeApp(userId);
+        await publish(app.token, {
+            appUid: app.uid,
+            name: 'ingestUpload',
+            source: SOURCE,
+        });
+        const spy = vi
+            .spyOn(env.server.stores.eventHandler, 'deleteForApp')
+            .mockRejectedValue(new Error('db down'));
+        try {
+            await expect(emitDeleted(app.uid)).resolves.not.toThrow();
+        } finally {
+            spy.mockRestore();
+        }
+    });
+
+    it('revokes a session whose app is already gone', async () => {
+        const app = await makeApp(userId);
+        await publish(app.token, {
+            appUid: app.uid,
+            name: 'ingestUpload',
+            source: SOURCE,
+        });
+        const { actor } = await env.server.services.auth.authenticate(
+            env.users.user.token,
+        );
+        await env.server.services.auth.createWorkerAppToken(
+            actor!,
+            app.uid,
+            EVENTS_WORKER_SESSION_NAME,
+        );
+
+        // Reproduces the already-stranded rows: the app row goes without
+        // either hook running — no `app.changed`, no `workers.destroy`. Goes
+        // through the store (not a raw DELETE) so its cache is invalidated
+        // the same way a real deletion leaves it.
+        const row = await env.server.stores.app.getByUid(app.uid);
+        await env.server.stores.app.delete((row as { id: number }).id);
+
+        const revoked =
+            await env.server.services.events.sweepStrandedWorkerSessions();
+        expect(revoked).toBeGreaterThanOrEqual(1);
+
+        // `authenticate()` can no longer resolve the token at all — the app
+        // it names is truly gone, not just the session — so the row itself
+        // is what confirms the revoke.
+        const sessions = await env.server.stores.session.getByUserId(userId, {
+            includeRevoked: true,
+        });
+        const session = sessions.find(
+            (row: {
+                kind: string;
+                app_uid: string | null;
+                meta?: { worker_name?: string };
+                revoked_at: unknown;
+            }) =>
+                row.kind === 'worker' &&
+                row.app_uid === app.uid &&
+                row.meta?.worker_name === EVENTS_WORKER_SESSION_NAME,
+        );
+        expect(session?.revoked_at).not.toBeNull();
+    });
+
+    it('leaves a stranded session alone when its app still exists', async () => {
+        const app = await makeApp(userId);
+        await publish(app.token, {
+            appUid: app.uid,
+            name: 'ingestUpload',
+            source: SOURCE,
+        });
+        const { actor } = await env.server.services.auth.authenticate(
+            env.users.user.token,
+        );
+        const token = await env.server.services.auth.createWorkerAppToken(
+            actor!,
+            app.uid,
+            EVENTS_WORKER_SESSION_NAME,
+        );
+
+        await env.server.services.events.sweepStrandedWorkerSessions();
+
+        const reauth = await env.server.services.auth.authenticate(token);
+        expect(reauth).not.toHaveProperty('reauth');
+    });
+});

src/backend/services/events/authorization.test.ts

@@ -0,0 +1,93 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+/**
+ * `assertCrossAppKvAuthorized` against fake deps — no server needed, since
+ * `CrossAppKvDeps` is already the injection seam. Pins that the
+ * three-segment-subject hint shows up only when the second segment does not
+ * look like an app uid, and that error codes never move.
+ */
+
+import { describe, expect, it } from 'vitest';
+import { makeActor } from '../../core/actor.js';
+import { isHttpError } from '../../core/http/HttpError.js';
+import {
+    assertCrossAppKvAuthorized,
+    type CrossAppKvDeps,
+} from './authorization.js';
+
+const actor = makeActor({ user: { id: 1, uuid: 'u-1', username: 'alice' } });
+
+const disabledDeps: CrossAppKvDeps = {
+    enabled: false,
+    getApp: async () => ({}),
+    checkPermission: async () => true,
+};
+
+const unknownAppDeps: CrossAppKvDeps = {
+    enabled: true,
+    getApp: async () => null,
+    checkPermission: async () => true,
+};
+
+const expectRejection = async (
+    promise: Promise<void>,
+): Promise<{ message: string; legacyCode?: string }> => {
+    try {
+        await promise;
+    } catch (err) {
+        if (!isHttpError(err)) throw err;
+        return { message: err.message, legacyCode: err.legacyCode };
+    }
+    throw new Error('expected assertCrossAppKvAuthorized to reject');
+};
+
+describe('assertCrossAppKvAuthorized hint', () => {
+    it('adds the hint when the segment does not look like an app uid', async () => {
+        const { message, legacyCode } = await expectRejection(
+            assertCrossAppKvAuthorized(actor, 'cart', disabledDeps),
+        );
+        expect(legacyCode).toBe('events_cross_app_disabled');
+        expect(message).toContain('three-segment');
+    });
+
+    it('omits the hint when the segment is an app uid', async () => {
+        const { message, legacyCode } = await expectRejection(
+            assertCrossAppKvAuthorized(actor, 'app-abc123', disabledDeps),
+        );
+        expect(legacyCode).toBe('events_cross_app_disabled');
+        expect(message).not.toContain('three-segment');
+    });
+
+    it('also adds the hint on the unknown_app path for a non-app-uid segment', async () => {
+        const { message, legacyCode } = await expectRejection(
+            assertCrossAppKvAuthorized(actor, 'cart', unknownAppDeps),
+        );
+        expect(legacyCode).toBe('subject_does_not_exist');
+        expect(message).toContain('three-segment');
+    });
+
+    it('omits the hint on the unknown_app path for an app uid', async () => {
+        const { message, legacyCode } = await expectRejection(
+            assertCrossAppKvAuthorized(actor, 'app-abc123', unknownAppDeps),
+        );
+        expect(legacyCode).toBe('subject_does_not_exist');
+        expect(message).not.toContain('three-segment');
+    });
+});

src/backend/services/events/authorization.ts

@@ -328,6 +328,13 @@ export const crossAppKvPermissions = (targetAppUid: string): string[] => [
     appDataPermission(targetAppUid, 'kv', CROSS_APP_KV_CLASS),
 ];
 
+// A key containing `:` written without an app id parses as `kv:<app>:<key>`,
+// which is the commonest way to arrive here by accident.
+const readAsAppSlot = (uid: string): string =>
+    uid.startsWith('app-')
+        ? ''
+        : ' — a three-segment `kv:` subject names an app in its second segment';
+
 /** Subscribe-time form. Codes match the ones a cross-app KV read answers with. */
 export const assertCrossAppKvAuthorized = async (
     actor: Actor,
@@ -339,13 +346,15 @@ export const assertCrossAppKvAuthorized = async (
     if (denial === 'disabled')
         throw new HttpError(
             403,
-            'kv: subscribing to another app’s data is not available',
+            `kv: subscribing to another app’s data is not available${readAsAppSlot(targetAppUid)}`,
             { legacyCode: 'events_cross_app_disabled' },
         );
     if (denial === 'unknown_app')
-        throw new HttpError(404, `entity_not_found: app:${targetAppUid}`, {
-            legacyCode: 'subject_does_not_exist',
-        });
+        throw new HttpError(
+            404,
+            `entity_not_found: app:${targetAppUid}${readAsAppSlot(targetAppUid)}`,
+            { legacyCode: 'subject_does_not_exist' },
+        );
     if (denial === 'sharing_off')
         throw new HttpError(
             403,

src/backend/services/events/dispatch.integration.test.ts

@@ -288,6 +288,29 @@ describe('a move out of a watched folder', () => {
         });
     });
 
+    it('tells the folder where a renamed node used to be', async () => {
+        const folder = `/${username}/watch-rename-from`;
+        await fs().mkdir(userId, { path: folder, createMissingParents: true });
+        await subscribeTo(`fs:${folder}`);
+
+        const file = await fs().touch(userId, {
+            path: `${folder}/before.txt`,
+        });
+        await settle((d) => pathOf(d) === `${folder}/before.txt`);
+
+        const renamed = await fs().rename(userId, file, 'after.txt');
+
+        const leftFrom = (d: DeliveryEnvelope) =>
+            d.event.op === 'move' &&
+            (d.event as { from?: string }).from === file.path;
+        await settle(leftFrom);
+
+        expect(delivered.find(leftFrom)?.event).toMatchObject({
+            path: renamed.path,
+            from: file.path,
+        });
+    });
+
     it('tells a filtered subscription on the folder too', async () => {
         const from = `/${username}/watch-move-out-glob-from`;
         const to = `/${username}/watch-move-out-glob-to`;

src/backend/services/events/forwardQueue.ts

@@ -17,6 +17,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
+import type { EventKey, KvOp } from '../../clients/event/types.js';
 import type { DeliverableEvent } from './registry.js';
 
 /**
@@ -62,7 +63,58 @@ export interface ForwardAck {
     entryId: string;
 }
 
-export type ForwardItem = ForwardDelivery | ForwardAck;
+/** Region R now has (or no longer has) session watchers on one anchor token. */
+export interface ForwardWatch {
+    kind: 'watch';
+    op: 'add' | 'drop';
+    /** Owner of the anchor keyspace the token lives in. */
+    userId: number;
+    token: string;
+}
+
+/** One committed change, replayed against another region's session rows. */
+export interface ForwardEvent {
+    kind: 'event';
+    family: 'fs' | 'kv';
+    ownerUserId: number;
+    /** Who caused it, so `self` is right on the far side. */
+    actingUserId?: number;
+    /** The emitter's event id and timestamp, kept so the two copies match. */
+    id: string;
+    ts: number;
+    /** Only session rows are evaluated: durable rows already crossed by row. */
+    sessionOnly: true;
+    /** Never re-forwarded. Present so a future multi-hop cannot loop. */
+    hop: 1;
+    fs?: {
+        key: EventKey;
+        entry: { uid: string; path: string; userId: number; isDir?: boolean };
+        ancestors: Array<{ uid: string; path: string }>;
+        movedFrom?: {
+            path: string;
+            ancestors: Array<{ uid: string; path: string }>;
+        };
+    };
+    kv?: { userUuid: string; appUid: string; kvKey: string; op: KvOp };
+}
+
+/** A subscription-set or presence generation moved in another region. */
+export interface ForwardBump {
+    kind: 'bump';
+    userId: number;
+    generation: number;
+    /** Which cache the receiver invalidates. */
+    scope: 'subscription' | 'presence';
+    /** `subscription` scope only: whether the table changed. */
+    durable: boolean;
+}
+
+export type ForwardItem =
+    | ForwardDelivery
+    | ForwardAck
+    | ForwardWatch
+    | ForwardEvent
+    | ForwardBump;
 
 /** One batch, as a peer receives it. */
 export interface ForwardBatch {
@@ -74,9 +126,12 @@ export interface ForwardBatch {
 /**
  * What a peer answers. `noSocket` names the pairs it holds no connection for,
  * read off its own socket registry — the only signal that authorises a repair.
+ * `noWatch` names anchor tokens the receiver holds no session row for at all —
+ * the sender prunes those out of its remote-watch index the same way.
  */
 export interface ForwardReply {
     noSocket?: Array<{ userId: number; appUid: string | null }>;
+    noWatch?: Array<{ userId: number; token: string }>;
 }
 
 // -- Bounds ----------------------------------------------------------

src/backend/services/events/forwarding.test.ts

@@ -54,6 +54,32 @@ import type {
     WorkerInvocationOutcome,
 } from './workerSeam.js';
 
+// A fake meter, so the forward-path counters can be asserted on without a
+// real OTel pipeline: every counter this module creates records its adds
+// here instead of exporting them anywhere.
+const { metricCalls } = vi.hoisted(() => ({
+    metricCalls: [] as Array<{
+        name: string;
+        value: number;
+        attributes: Record<string, unknown>;
+    }>,
+}));
+vi.mock(import('@opentelemetry/api'), async (importOriginal) => ({
+    ...(await importOriginal()),
+    metrics: {
+        getMeter: () => ({
+            createCounter: (name: string) => ({
+                add: (
+                    value: number,
+                    attributes: Record<string, unknown> = {},
+                ) => {
+                    metricCalls.push({ name, value, attributes });
+                },
+            }),
+        }),
+    },
+}));
+
 // -- The replicated table --------------------------------------------
 //
 // One item per (pair, region) now, not one item with a `regions` map field:
@@ -136,6 +162,9 @@ interface Region {
     /** Peers whose POSTs never come back — a timeout, not a refusal. */
     unreachable: Set<string>;
     handlers: Map<string, (key: string, data: unknown, meta: unknown) => void>;
+    /** `outer.*` emits held here instead of reaching peers, when `deferBus`. */
+    busQueue: Array<{ key: string; data: unknown; meta: object }>;
+    deferBus: boolean;
 }
 
 let regions: Map<string, Region>;
@@ -158,9 +187,16 @@ const entry = (over: Partial<FSEntry> = {}): FSEntry =>
         ...over,
     }) as FSEntry;
 
-const actorFor = (appUid: string | null = null): Actor =>
+const actorFor = (
+    appUid: string | null = null,
+    holderUserId: number = userId,
+): Actor =>
     ({
-        user: { id: userId, uuid: `user-${userId}`, username: `u${userId}` },
+        user: {
+            id: holderUserId,
+            uuid: `user-${holderUserId}`,
+            username: `u${holderUserId}`,
+        },
         effectiveApp: appUid ? { uid: appUid } : null,
         app: appUid ? { uid: appUid } : null,
     }) as unknown as Actor;
@@ -195,6 +231,7 @@ const makeRegion = (
     name: string,
     peers: string[],
     config: Partial<IConfig> = {},
+    options: { deferBus?: boolean } = {},
 ): Region => {
     // A keyspace per region, because that is what a region is here: leases,
     // pending queues and connection counts are local by construction, and a
@@ -212,6 +249,8 @@ const makeRegion = (
         alarms: vi.fn(),
         unreachable: new Set(),
         handlers: new Map(),
+        busQueue: [],
+        deferBus: options.deferBus === true,
     } as unknown as Region;
 
     const fullConfig = {
@@ -234,8 +273,14 @@ const makeRegion = (
                 );
             },
             // The broadcast channel, simulated: an `outer.*` emit reaches every
-            // peer tagged `from_outside`, and never its own emitter.
+            // peer tagged `from_outside`, and never its own emitter — unless
+            // `deferBus` holds it for `flushBus` to deliver later, the way the
+            // real ~2 s batch delays it behind the addressed channel.
             emit: (key: string, data: unknown, meta: object) => {
+                if (region.deferBus) {
+                    region.busQueue.push({ key, data, meta });
+                    return;
+                }
                 for (const other of regions.values()) {
                     if (other === region) continue;
                     other.handlers.get(key)?.(key, data, {
@@ -308,13 +353,62 @@ const makeRegion = (
         pendingDelivery: pending,
         eventSubscription: subscriptions,
         durableSubscription: {
-            warmRegion: async () => false,
+            // A real rebuild off the shared `rows` table, not a stub that
+            // always answers "already warm" — item 13's tests need a region
+            // that actually goes cold and rebuilds from what another region
+            // wrote.
+            warmRegion: async (ownerUserId: number) => {
+                const owned = [...rows.values()].filter(
+                    (row) => row.ownerUserId === ownerUserId,
+                );
+                await subscriptions.rebuildDurable(ownerUserId, owned);
+                return true;
+            },
             getBySubId: async (subId: string) => rows.get(subId) ?? null,
             remove: async (row: DurableSubscription) => {
                 rows.delete(row.subId);
                 return { userId: row.holderUserId, generation: 1 };
             },
             suspend: async () => [{ userId, generation: 1 }],
+            // Enough of the real store's write-through contract for
+            // `subscribeDurable` to run against this harness: land the row in
+            // the shared table and this region's cache, then bump.
+            create: async (
+                input: Record<string, unknown>,
+            ): Promise<{
+                row: DurableSubscription;
+                bump: { userId: number; generation: number };
+            }> => {
+                const row = durableRow({
+                    holderUserId: input.holderUserId as number,
+                    ownerUserId: input.ownerUserId as number,
+                    subject: input.subject as string,
+                    token: input.token as string,
+                    anchorUid: input.anchorUid as string,
+                    anchorPath: input.anchorPath as string,
+                    match: (input.match as string | null) ?? null,
+                    op: (input.op as DurableSubscription['op']) ?? null,
+                    appUid: (input.appUid as string | null) ?? null,
+                    delivery: input.delivery as DurableSubscription['delivery'],
+                    targets: input.targets as DurableSubscription['targets'],
+                    handlerName: (input.handlerName as string | null) ?? null,
+                    context: (input.context as string | null) ?? null,
+                    permission:
+                        input.permission as DurableSubscription['permission'],
+                    expiresAt: (input.expiresAt as number | null) ?? null,
+                });
+                rows.set(row.subId, row);
+                await subscriptions.cacheDurable([row]);
+                return {
+                    row,
+                    bump: {
+                        userId: row.ownerUserId,
+                        generation: await subscriptions.bumpGeneration(
+                            row.ownerUserId,
+                        ),
+                    },
+                };
+            },
         },
         fsEntry: {
             getEntryByUuid: async (uid: string) =>
@@ -354,6 +448,24 @@ const makeRegion = (
         },
     };
     region.forward.onServerStart();
+    // The one listener `EventsService.onServerStart()` registers that these
+    // tests need — the webhook backstop for a durable generation bump. The
+    // rest of it (kv.mutated, permission wiring, sweep timers) is out of
+    // scope here, and `services.permission` is not stubbed to support it.
+    clients.event.on(
+        'outer.pubsub.events.generationBumped',
+        ((_key: string, data: unknown, meta: unknown) => {
+            if (!(meta as { from_outside?: boolean })?.from_outside) return;
+            const { userId: bumpedUserId, durable } = (data ?? {}) as {
+                userId?: number;
+                durable?: boolean;
+            };
+            if (typeof bumpedUserId === 'number')
+                region.events.invalidateUser(bumpedUserId, {
+                    rebuild: durable === true,
+                });
+        }) as (...args: never[]) => void,
+    );
     regions.set(name, region);
     return region;
 };
@@ -390,6 +502,48 @@ const dispatch = (region: Region, node = entry()): Promise<void> =>
         ancestors: async () => ancestors(),
     });
 
+/** A session (`onLocal`) row on the shared anchor, through the real subscribe path. */
+const subscribeSession = async (
+    region: Region,
+    opts: {
+        holderUserId?: number;
+        socketId?: string;
+        appUid?: string | null;
+    } = {},
+): Promise<{ subId: string; socketId: string }> => {
+    const socketId = opts.socketId ?? `socket-${seq}`;
+    const { sub } = await region.events.subscribe(
+        actorFor(opts.appUid ?? null, opts.holderUserId ?? userId),
+        socketId,
+        { subject: `fs:${anchorUid()}` },
+    );
+    return { subId: sub.subId, socketId };
+};
+
+const unsubscribeSession = (
+    region: Region,
+    subId: string,
+    socketId: string,
+    holderUserId = userId,
+): Promise<void> =>
+    region.events.unsubscribe(actorFor(null, holderUserId), socketId, {
+        subId,
+    });
+
+/** Deliver whatever `outer.*` emits a `deferBus` region is holding. */
+const flushBus = (region: Region): void => {
+    const queued = region.busQueue;
+    region.busQueue = [];
+    for (const { key, data, meta } of queued)
+        for (const other of regions.values()) {
+            if (other === region) continue;
+            other.handlers.get(key)?.(key, data, {
+                ...meta,
+                from_outside: true,
+            });
+        }
+};
+
 const posted = (region: Region, count = 1): Promise<void> =>
     vi.waitFor(
         () => expect(region.posts.length).toBeGreaterThanOrEqual(count),
@@ -430,6 +584,7 @@ beforeEach(() => {
     regions = new Map();
     rows = new Map();
     workerOutcome = 'deferred';
+    metricCalls.length = 0;
     EventForwardService.LEAVE_DELAY_MIN_MS = 60;
     EventForwardService.LEAVE_DELAY_MAX_MS = 60;
 });
@@ -1161,3 +1316,380 @@ describe('receiving a batch', () => {
         expect(settled.at(-1)).toBe('slow');
     });
 });
+
+// -- Observability ------------------------------------------------------
+
+describe('forward-path metrics', () => {
+    it('counts a broadcast fan-out as sent, and the receiving region as received', async () => {
+        const west = makeRegion('west', ['east', 'south']);
+        const east = makeRegion('east', ['west', 'south']);
+        makeRegion('south', ['west', 'east']);
+        east.rooms.add(String(userId));
+        await east.forward.noteConnect(actorFor());
+        await register(west);
+
+        await dispatch(west);
+        await posted(west);
+        await arrived(east);
+
+        const sent = metricCalls.filter(
+            (call) => call.name === 'events.forward.sent',
+        );
+        expect(sent).toContainEqual({
+            name: 'events.forward.sent',
+            value: 1,
+            attributes: { from: 'west', to: 'east', class: 'broadcast' },
+        });
+
+        const received = metricCalls.filter(
+            (call) => call.name === 'events.forward.received',
+        );
+        expect(received).toContainEqual({
+            name: 'events.forward.received',
+            value: 1,
+            attributes: { from: 'west', to: 'east' },
+        });
+    });
+
+    it('counts a single hand-off as sent, class single', async () => {
+        const west = makeRegion('west', ['east', 'south']);
+        const east = makeRegion('east', ['west', 'south']);
+        const south = makeRegion('south', ['west', 'east']);
+        east.rooms.add(String(userId));
+        south.rooms.add(String(userId));
+        await east.forward.noteConnect(actorFor());
+        await quiet(5);
+        await south.forward.noteConnect(actorFor());
+        await register(west, {
+            delivery: 'single',
+            targets: ['socket', 'worker'] as SubscriptionTarget[],
+            handlerName: 'onWrite',
+        });
+
+        await dispatch(west);
+        await posted(west);
+        await arrived(south);
+
+        const sent = metricCalls.filter(
+            (call) => call.name === 'events.forward.sent',
+        );
+        expect(sent).toContainEqual({
+            name: 'events.forward.sent',
+            value: 1,
+            attributes: { from: 'west', to: 'south', class: 'single' },
+        });
+        expect(metricCalls).toContainEqual({
+            name: 'events.single.attempt',
+            value: 1,
+            attributes: { target: 'remote-socket', result: 'sent' },
+        });
+    });
+
+    it("counts a single attempt landing on this region's own socket", async () => {
+        const west = makeRegion('west', ['east']);
+        const east = makeRegion('east', ['west']);
+        west.rooms.add(String(userId));
+        east.rooms.add(String(userId));
+        await west.forward.noteConnect(actorFor());
+        await east.forward.noteConnect(actorFor());
+        await register(west, {
+            delivery: 'single',
+            targets: ['socket', 'worker'] as SubscriptionTarget[],
+            handlerName: 'onWrite',
+        });
+
+        await dispatch(west);
+        await arrived(west);
+
+        expect(metricCalls).toContainEqual({
+            name: 'events.single.attempt',
+            value: 1,
+            attributes: { target: 'local-socket', result: 'sent' },
+        });
+    });
+
+    it('counts a single attempt that reaches the handler', async () => {
+        const west = makeRegion('west', ['east']);
+        const east = makeRegion('east', ['west']);
+        east.rooms.add(String(userId));
+        await east.forward.noteConnect(actorFor());
+        workerOutcome = 'settled';
+        await register(west, {
+            delivery: 'single',
+            targets: ['socket', 'worker'] as SubscriptionTarget[],
+            handlerName: 'onWrite',
+        });
+
+        await dispatch(west);
+        await posted(west);
+
+        for (let attempt = 0; attempt < 3; attempt++) {
+            jump(61_000);
+            await west.events.sweepPending();
+        }
+
+        expect(metricCalls).toContainEqual({
+            name: 'events.single.attempt',
+            value: 1,
+            attributes: { target: 'worker', result: 'settled' },
+        });
+    });
+});
+
+// -- Cross-region session subscriptions (item 2) -----------------------
+
+/** Session forwarding is on by default, so this is the ordinary config. */
+const forwardCfg = { events: { enabled: true } } as Partial<IConfig>;
+
+describe('a session subscription in another region', () => {
+    it('delivers a write committed in the region that has no rows', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        const east = makeRegion('east', ['west'], forwardCfg);
+
+        await subscribeSession(east);
+        // The `watch` announce reaching west over the 25 ms queue.
+        await posted(east);
+
+        await dispatch(west);
+
+        await arrived(east);
+        expect(east.sent[0].event).toMatchObject({ op: 'write', self: true });
+        expect(west.sent).toEqual([]);
+    });
+
+    it('delivers to a subscriber who is not the writer', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        const east = makeRegion('east', ['west'], forwardCfg);
+        const holderUserId = userId + 1;
+
+        await subscribeSession(east, { holderUserId });
+        await posted(east);
+
+        // `actingUserId` defaults to the anchor's owner — the shared-file
+        // case, where the holder is someone else entirely.
+        await dispatch(west);
+
+        await arrived(east);
+        expect(east.sent[0].event).toMatchObject({ self: false });
+    });
+
+    it('sends nothing to a region that never announced the token', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        makeRegion('east', ['west'], forwardCfg);
+
+        await dispatch(west);
+        await quiet(150);
+
+        expect(west.posts).toEqual([]);
+    });
+
+    it('stops forwarding once the last session row goes', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        const east = makeRegion('east', ['west'], forwardCfg);
+
+        const { subId, socketId } = await subscribeSession(east);
+        await posted(east);
+        const afterAnnounce = east.posts.length;
+
+        await unsubscribeSession(east, subId, socketId);
+        await vi.waitFor(() =>
+            expect(east.posts.length).toBeGreaterThan(afterAnnounce),
+        );
+
+        await dispatch(west);
+        await quiet(150);
+
+        expect(west.posts).toEqual([]);
+        const remoteAfter = await west.subscriptions.watchedFor(userId, [
+            fsAnchorToken(anchorUid()),
+        ]);
+        expect(remoteAfter.remote.size).toBe(0);
+    });
+
+    it('does not deliver a durable row twice when one shares the anchor with a session row', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        const east = makeRegion('east', ['west'], forwardCfg);
+        east.rooms.add(String(userId));
+        await east.forward.noteConnect(actorFor());
+        const durable = await register(west);
+        const { subId: sessionSubId } = await subscribeSession(east);
+        await posted(east);
+
+        await dispatch(west);
+        await arrived(east, 2);
+
+        const subIds = east.sent.map((envelope) => envelope.subId);
+        expect(subIds.filter((id) => id === durable.subId)).toHaveLength(1);
+        expect(subIds.filter((id) => id === sessionSubId)).toHaveLength(1);
+    });
+
+    it('prunes a token the peer no longer holds', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        makeRegion('east', ['west'], forwardCfg);
+        const token = fsAnchorToken(anchorUid());
+        // A stale entry, written by hand rather than through a real
+        // subscribe: east never actually holds a session row for it.
+        await west.subscriptions.noteRemoteWatch(userId, token, 'east', 'add');
+
+        await dispatch(west);
+        await posted(west);
+
+        await vi.waitFor(async () => {
+            const { remote } = await west.subscriptions.watchedFor(userId, [
+                token,
+            ]);
+            expect(remote.size).toBe(0);
+        });
+    });
+
+    it('ignores an item kind it does not recognize, and still applies the ones it does', async () => {
+        const east = makeRegion('east', ['west'], forwardCfg);
+
+        await expect(
+            east.forward.receive({
+                from: 'west',
+                items: [
+                    { kind: 'from-the-future' } as never,
+                    {
+                        kind: 'bump',
+                        userId,
+                        generation: 1,
+                        scope: 'subscription',
+                        durable: true,
+                    },
+                ],
+            }),
+        ).resolves.toEqual({});
+    });
+
+    it('records watch and session-forward metrics', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        const east = makeRegion('east', ['west'], forwardCfg);
+        metricCalls.length = 0;
+
+        await subscribeSession(east);
+        await posted(east);
+
+        await dispatch(west);
+        await arrived(east);
+
+        expect(metricCalls).toContainEqual(
+            expect.objectContaining({
+                name: 'events.forward.sent',
+                attributes: expect.objectContaining({ class: 'watch' }),
+            }),
+        );
+        expect(metricCalls).toContainEqual(
+            expect.objectContaining({
+                name: 'events.forward.sent',
+                attributes: expect.objectContaining({ class: 'session' }),
+            }),
+        );
+        expect(metricCalls).toContainEqual(
+            expect.objectContaining({
+                name: 'events.session.forward',
+                attributes: expect.objectContaining({ result: 'matched' }),
+            }),
+        );
+    });
+
+    it('does nothing when session forwarding is turned off, even with a matching announce', async () => {
+        const west = makeRegion('west', ['east'], forwardCfg);
+        const east = makeRegion('east', ['west'], {
+            events: { enabled: true, forwardSession: false },
+        } as Partial<IConfig>);
+
+        await subscribeSession(east);
+        await quiet(80);
+
+        await dispatch(west);
+        await quiet(150);
+
+        // East never announced (its own forwarding is off), so west has
+        // nothing in its remote-watch index to forward against.
+        expect(west.posts).toEqual([]);
+    });
+
+    it('writes no remote-watch entry on a region that has it turned off', async () => {
+        const west = makeRegion('west', ['east'], {
+            events: { enabled: true, forwardSession: false },
+        } as Partial<IConfig>);
+        const east = makeRegion('east', ['west'], forwardCfg);
+        const token = fsAnchorToken(anchorUid());
+
+        await subscribeSession(east);
+        await posted(east);
+
+        const { remote } = await west.subscriptions.watchedFor(userId, [token]);
+        expect(remote.size).toBe(0);
+        expect(await west.subscriptions.userHasAny(userId)).toBe(false);
+    });
+});
+
+// -- The first seconds after a cross-region durable subscribe (item 13) --
+
+describe('a durable row created in another region', () => {
+    it('reaches a peer before the batched bus does', async () => {
+        const west = makeRegion('west', ['east']);
+        const east = makeRegion('east', ['west'], {}, { deferBus: true });
+        east.rooms.add(String(userId));
+        await east.forward.noteConnect(actorFor());
+
+        // West marks itself warm-and-empty before the row exists to find.
+        await dispatch(west);
+
+        await east.events.subscribeDurable(actorFor(), {
+            subject: `fs:${anchorUid()}`,
+            targets: ['socket'],
+        });
+        // A tick for the 25 ms addressed queue — the bus stays held.
+        await quiet(80);
+
+        await dispatch(west);
+
+        await arrived(east);
+        expect(east.sent[0].event).toMatchObject({ op: 'write' });
+    });
+
+    it('still learns about the row when the addressed send fails', async () => {
+        const west = makeRegion('west', ['east']);
+        const east = makeRegion('east', ['west'], {}, { deferBus: true });
+        east.unreachable.add('west');
+        east.rooms.add(String(userId));
+        await east.forward.noteConnect(actorFor());
+
+        await dispatch(west);
+
+        await east.events.subscribeDurable(actorFor(), {
+            subject: `fs:${anchorUid()}`,
+            targets: ['socket'],
+        });
+        await quiet(80);
+
+        await dispatch(west);
+        await quiet(150);
+        expect(west.sent).toEqual([]);
+
+        // The backstop: the ~2 s all-peers webhook finally lands.
+        flushBus(east);
+        await dispatch(west);
+
+        await arrived(east);
+        expect(east.sent[0].event).toMatchObject({ op: 'write' });
+    });
+
+    it('also fans a presence generation bump onto the addressed channel', async () => {
+        const west = makeRegion('west', ['east']);
+        makeRegion('east', ['west']);
+        metricCalls.length = 0;
+
+        await west.forward.noteConnect(actorFor());
+
+        expect(metricCalls).toContainEqual(
+            expect.objectContaining({
+                name: 'events.forward.sent',
+                attributes: expect.objectContaining({ class: 'bump' }),
+            }),
+        );
+    });
+});

src/backend/services/events/metering.test.ts

@@ -196,6 +196,9 @@ beforeEach(() => {
                 fanOut: async () => undefined,
                 handOff: () => undefined,
                 relayAck: () => undefined,
+                announceWatch: () => undefined,
+                forwardEvent: () => undefined,
+                announceGeneration: () => undefined,
             },
             socket: { send: vi.fn(), has: () => false },
             fs: { getAncestorChain: async () => [] },

src/backend/services/events/metrics.ts

@@ -0,0 +1,53 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+import { metrics } from '@opentelemetry/api';
+
+/**
+ * Counters for the cross-region forward path — which region a delivery went to,
+ * whether the peer accepted it, and where a single-delivery attempt actually
+ * landed. Attributes carry only region names and enum labels, never user data.
+ */
+
+const meter = metrics.getMeter('puter-backend');
+
+/** Deliveries handed to a peer region, by class. */
+export const forwardSent = meter.createCounter('events.forward.sent', {
+    description: 'Deliveries queued for a peer region',
+});
+
+/** Batches a peer handed us, and what was in them. */
+export const forwardReceived = meter.createCounter('events.forward.received', {
+    description: 'Forwarded items accepted from a peer region',
+});
+
+/** One attempt at one owed `single` delivery, and where it went. */
+export const singleAttempt = meter.createCounter('events.single.attempt', {
+    description: 'Attempts at a single-delivery subscription, by target',
+});
+
+/**
+ * What a forwarded session event found on the far side: `no-rows` is the
+ * remote-watch index going stale (the token a peer announced is no longer
+ * watched here), and is what drives a `noWatch` reply.
+ */
+export const sessionForward = meter.createCounter('events.session.forward', {
+    description:
+        'Forwarded session events, by whether the receiver matched one',
+});

src/backend/services/events/singleDelivery.test.ts

@@ -276,6 +276,9 @@ beforeEach(async () => {
                 fanOut: async () => undefined,
                 handOff: () => undefined,
                 relayAck: () => undefined,
+                announceWatch: () => undefined,
+                forwardEvent: () => undefined,
+                announceGeneration: () => undefined,
             },
             socket: {
                 send: vi.fn(async (_spec, _key, data) => {
@@ -673,7 +676,9 @@ describe('when a delivery cannot be held', () => {
             new Error('the cache is unreachable'),
         );
 
-        await expect(dispatch()).resolves.toBeUndefined();
+        // The row still matched; the enqueue failure is caught asynchronously
+        // downstream of the return.
+        await expect(dispatch()).resolves.toBe(true);
 
         expect(sent).toEqual([]);
         expect(alarms).toHaveBeenCalledWith(

src/backend/services/events/workers.integration.test.ts

@@ -28,6 +28,7 @@ import { v4 as uuidv4 } from 'uuid';
 import { setupPuterTestEnv, type PuterTestEnv } from '../../testUtil.js';
 import type { IConfig } from '../../types.js';
 import { EVENTS_BACKGROUND_PERMISSION } from './authorization.js';
+import { EVENTS_WORKER_SESSION_NAME } from './workerRuntime.js';
 
 const BOOT_TIMEOUT_MS = 120_000;
 
@@ -87,6 +88,58 @@ const makeApp = async (ownerUserId: number): Promise<{ uid: string; token: strin
 const SOURCE = 'async ({ event }) => { console.log(event.path); }';
 const OTHER_SOURCE = 'async ({ event, ctx }) => { console.log(ctx.url); }';
 
+/** The reused `events:handlers` worker session for (userId, appUid), if any. */
+const workerSessionFor = async (forAppUid: string) => {
+    const rows = await env.server.stores.session.getByUserId(userId, {
+        includeRevoked: true,
+    });
+    return rows.find(
+        (row: {
+            kind: string;
+            app_uid: string;
+            meta?: { worker_name?: string };
+        }) =>
+            row.kind === 'worker' &&
+            row.app_uid === forAppUid &&
+            row.meta?.worker_name === EVENTS_WORKER_SESSION_NAME,
+    );
+};
+
+/** A durable `single` subscription bound to a published handler, with the
+ * grants a background delivery needs — the setup a session gets minted for. */
+const subscribeBackgroundHandler = async (
+    app: { uid: string; token: string },
+    handlerName: string,
+): Promise<void> => {
+    const anchor = `/${env.users.user.username}/${uuidv4()}`;
+    await env.server.services.fs.mkdir(userId, {
+        path: anchor,
+        createMissingParents: true,
+    });
+    const { actor } = await env.server.services.auth.authenticate(
+        env.users.user.token,
+    );
+    const entry = await env.server.stores.fsEntry.getEntryByPath(anchor);
+    await env.server.services.permission.grantUserAppPermission(
+        actor!,
+        app.uid,
+        `fs:${entry!.uid}:list`,
+    );
+    await env.server.services.permission.grantUserAppPermission(
+        actor!,
+        app.uid,
+        EVENTS_BACKGROUND_PERMISSION,
+    );
+
+    const subscribed = await call('POST', '/events/subscribe', app.token, {
+        subject: `fs:${anchor}`,
+        delivery: 'single',
+        handlerName,
+        targets: ['worker'],
+    });
+    expect(subscribed.status).toBe(200);
+};
+
 let creates: Array<{ appUid: string; ownerId: number | undefined }>;
 let destroys: Array<{ appUid: string; ownerId: number | undefined }>;
 
@@ -345,6 +398,96 @@ describe('POST /events/workers/destroy', () => {
         expect(destroyed.body).toMatchObject({ removed: 1, suspended: 1 });
     });
 
+    it('retires the app worker session it minted', async () => {
+        const app = await makeApp(userId);
+        await publish(app.token, {
+            appUid: app.uid,
+            name: 'ingestUpload',
+            source: SOURCE,
+        });
+        await subscribeBackgroundHandler(app, 'ingestUpload');
+
+        // Stands in for the session a real delivery would have minted.
+        const { actor } = await env.server.services.auth.authenticate(
+            env.users.user.token,
+        );
+        const workerToken = await env.server.services.auth.createWorkerAppToken(
+            actor!,
+            app.uid,
+            EVENTS_WORKER_SESSION_NAME,
+        );
+
+        await call('POST', '/events/workers/destroy', app.token, {
+            appUid: app.uid,
+        });
+
+        const session = await workerSessionFor(app.uid);
+        expect(session?.revoked_at).not.toBeNull();
+
+        const reauth = await env.server.services.auth.authenticate(workerToken);
+        expect(reauth).toMatchObject({
+            reauth: { reason: 'session_revoked' },
+        });
+    });
+
+    it('leaves the rows suspended so a republish resumes them, and mints a fresh session', async () => {
+        const app = await makeApp(userId);
+        await publish(app.token, {
+            appUid: app.uid,
+            name: 'ingestUpload',
+            source: SOURCE,
+        });
+        await subscribeBackgroundHandler(app, 'ingestUpload');
+        const { actor } = await env.server.services.auth.authenticate(
+            env.users.user.token,
+        );
+        await env.server.services.auth.createWorkerAppToken(
+            actor!,
+            app.uid,
+            EVENTS_WORKER_SESSION_NAME,
+        );
+
+        await call('POST', '/events/workers/destroy', app.token, {
+            appUid: app.uid,
+        });
+
+        const suspended = await env.server.stores.durableSubscription.listByHandler(
+            app.uid,
+            'ingestUpload',
+            { suspendedReason: 'handler_not_found' },
+        );
+        expect(suspended).toHaveLength(1);
+
+        const republished = await publish(app.token, {
+            appUid: app.uid,
+            name: 'ingestUpload',
+            source: SOURCE,
+        });
+        expect(republished.status).toBe(200);
+
+        const stillSuspended =
+            await env.server.stores.durableSubscription.listByHandler(
+                app.uid,
+                'ingestUpload',
+                { suspendedReason: 'handler_not_found' },
+            );
+        expect(stillSuspended).toHaveLength(0);
+
+        // The next delivery mints a fresh session through the unchanged
+        // `#mintSubscriberToken` — nothing here refuses a new one.
+        const freshToken = await env.server.services.auth.createWorkerAppToken(
+            actor!,
+            app.uid,
+            EVENTS_WORKER_SESSION_NAME,
+        );
+        expect(typeof freshToken).toBe('string');
+        const freshSession = await env.server.stores.session.getWorker(userId, {
+            appUid: app.uid,
+            workerName: EVENTS_WORKER_SESSION_NAME,
+        });
+        expect(freshSession?.revoked_at).toBeNull();
+    });
+
     it('refuses an app token destroying an app it is not', async () => {
         const app = await makeApp(userId);
         await publish(app.token, { appUid: app.uid, name: 'a', source: SOURCE });

src/backend/services/fs/FSService.ts

@@ -3450,12 +3450,17 @@ export class FSService extends PuterService {
                 newPath,
             );
         }
-        this.#emitFsEvent('fs.rename', updated, {
-            old_name: entry.name,
-            new_name: newName,
-            old_path: entry.path,
-            new_path: newPath,
-        });
+        this.#emitFsEvent(
+            'fs.rename',
+            updated,
+            {
+                old_name: entry.name,
+                new_name: newName,
+                old_path: entry.path,
+                new_path: newPath,
+            },
+            { path: entry.path },
+        );
         return updated;
     }
 
@@ -3854,7 +3859,8 @@ export class FSService extends PuterService {
      *
      * Currently emitted: fs.create.{file,directory,shortcut,symlink}
      * fs.write.file — overwrite of an existing file fs.rename — in-place name
-     * change (move emits fs.move.node separately)
+     * change (move emits fs.move.node separately); both carry the path the node
+     * left via `movedFrom`.
      *
      * Skipped intentionally: `fs.pending.*` (no real entry yet at signed-URL
      * issue time) and per-flavor `fs.move.file` (move already emits
@@ -3864,6 +3870,7 @@ export class FSService extends PuterService {
         name: T,
         entry: FSEntry,
         extras: Record<string, unknown> = {},
+        movedFrom?: { path: string },
     ): void {
         try {
             this.clients.event.emit(
@@ -3881,7 +3888,7 @@ export class FSService extends PuterService {
         } catch {
             console.warn('missing event emissions');
         }
-        this.#dispatchEvents(name, entry);
+        this.#dispatchEvents(name, entry, movedFrom);
     }
 
     /**

src/backend/stores/events/DurableSubscriptionStore.integration.test.ts

@@ -491,6 +491,87 @@ describe('the expiry sweep', () => {
     });
 });
 
+describe('listHolderIdsForApp', () => {
+    it('names every distinct holder bound to the app`s handlers', async () => {
+        const appUid = `app-${uuidv4()}`;
+        await durable().create(
+            input({ appUid, holderUserId: userId, targets: ['socket', 'worker'] }),
+        );
+        await durable().create(
+            input({
+                appUid,
+                holderUserId: otherUserId,
+                targets: ['socket', 'worker'],
+            }),
+        );
+        // A second row for the same holder must not double the name.
+        await durable().create(
+            input({ appUid, holderUserId: userId, targets: ['socket', 'worker'] }),
+        );
+
+        const holders = await durable().listHolderIdsForApp(appUid);
+        expect(holders.sort()).toEqual([userId, otherUserId].sort());
+    });
+
+    it('finds a suspended row`s holder too', async () => {
+        const appUid = `app-${uuidv4()}`;
+        const { row } = await durable().create(
+            input({ appUid, targets: ['socket', 'worker'] }),
+        );
+        await durable().suspend([row], 'handler_not_found');
+
+        expect(await durable().listHolderIdsForApp(appUid)).toEqual([userId]);
+    });
+
+    it('answers an empty list for an app with nothing bound', async () => {
+        expect(
+            await durable().listHolderIdsForApp(`app-${uuidv4()}`),
+        ).toEqual([]);
+    });
+});
+
+describe('reapForApp', () => {
+    it('deletes every row bound to the app and bumps its owner`s generation', async () => {
+        const appUid = `app-${uuidv4()}`;
+        const mine = await durable().create(
+            input({ appUid, targets: ['socket', 'worker'] }),
+        );
+        const untouched = await durable().create(input());
+        const before = await cache().getGeneration(userId);
+
+        await expect(durable().reapForApp(appUid, 500)).resolves.toBe(1);
+
+        await expect(durable().getBySubId(mine.row.subId)).resolves.toBeNull();
+        await expect(
+            durable().getBySubId(untouched.row.subId),
+        ).resolves.not.toBeNull();
+        await expect(cache().getGeneration(userId)).resolves.toBeGreaterThan(
+            before,
+        );
+    });
+
+    it('never touches another app`s rows', async () => {
+        const appUid = `app-${uuidv4()}`;
+        const otherAppUid = `app-${uuidv4()}`;
+        await durable().create(input({ appUid, targets: ['socket', 'worker'] }));
+        const theirs = await durable().create(
+            input({ appUid: otherAppUid, targets: ['socket', 'worker'] }),
+        );
+
+        await durable().reapForApp(appUid, 500);
+
+        await expect(
+            durable().getBySubId(theirs.row.subId),
+        ).resolves.not.toBeNull();
+    });
+
+    it('answers 0 for an app with nothing to reap', async () => {
+        await expect(
+            durable().reapForApp(`app-${uuidv4()}`, 500),
+        ).resolves.toBe(0);
+    });
+});
+
 describe('warming a cold region', () => {
     it('reads the table once and then answers from the cache', async () => {
         const { row } = await durable().create(input());

src/backend/stores/events/DurableSubscriptionStore.ts

@@ -717,8 +717,51 @@ export class DurableSubscriptionStore extends PuterStore {
         return rows.map(toRow);
     }
 
+    /**
+     * Every holder with a row bound to this app — who may hold a reused
+     * `events:handlers` worker session for it.
+     * `idx_event_subscriptions_app_handler` is `app_uid`-leading, so this is an
+     * indexed scan rather than a table scan, with no new index needed.
+     * Suspended rows included: a row `handler_not_found` still names a holder
+     * who may hold the session.
+     */
+    async listHolderIdsForApp(appUid: string): Promise<number[]> {
+        const rows = await this.clients.db.pread(
+            `SELECT DISTINCT \`holder_user_id\` FROM \`${TABLE}\` WHERE \`app_uid\` = ?`,
+            [appUid],
+        );
+        return rows.map((row) => Number(row.holder_user_id));
+    }
+
+    /**
+     * Page and delete every row bound to one app, wherever its holder lives.
+     * What app deletion reaps: `#reap` already purges the backlog, drops the
+     * cache and bumps a generation per owner, so this is the whole of it.
+     */
+    async reapForApp(appUid: string, batchSize: number): Promise<number> {
+        return this.#reap(await this.#listForApp(appUid, batchSize));
+    }
+
     // -- Internals ---------------------------------------------------
 
+    /**
+     * Primary, unlike the timed sweeps: this runs the moment an app is deleted,
+     * and a replica still a beat behind would answer "nothing to reap" and
+     * leave the rows with nothing to come back for them.
+     */
+    async #listForApp(
+        appUid: string,
+        batchSize: number,
+    ): Promise<DurableSubscription[]> {
+        const limit = Math.max(1, Math.floor(batchSize));
+        const rows = await this.clients.db.pread(
+            `SELECT ${SELECT_COLUMNS} FROM \`${TABLE}\` ` +
+                'WHERE `app_uid` = ? ORDER BY `id` LIMIT ?',
+            [appUid, limit],
+        );
+        return rows.map(toRow);
+    }
+
     /** One page of a whole-table scan, ordered and positioned by primary key. */
     async #page(
         where: string[],

src/backend/stores/events/EventHandlerStore.integration.test.ts

@@ -372,6 +372,32 @@ describe('removing a handler', () => {
     });
 });
 
+describe('deleteForApp', () => {
+    it('drops every handler an app published in one call', async () => {
+        await handlers().publish({ appUid, name: 'a', source: SOURCE });
+        await handlers().publish({ appUid, name: 'b', source: OTHER_SOURCE });
+
+        await expect(handlers().deleteForApp(appUid)).resolves.toBe(2);
+        expect(await handlers().listForApp(appUid)).toEqual([]);
+    });
+
+    it('never touches another app`s handlers', async () => {
+        await handlers().publish({ appUid, name: 'a', source: SOURCE });
+        await handlers().publish({
+            appUid: otherAppUid,
+            name: 'a',
+            source: SOURCE,
+        });
+
+        await handlers().deleteForApp(appUid);
+        expect(await handlers().listForApp(otherAppUid)).toHaveLength(1);
+    });
+
+    it('answers 0 for an app with nothing published', async () => {
+        await expect(handlers().deleteForApp(appUid)).resolves.toBe(0);
+    });
+});
+
 describe('totalSourceBytesForApp', () => {
     it('is zero for an app with nothing published', async () => {
         expect(await handlers().totalSourceBytesForApp(appUid)).toBe(0);

src/backend/stores/events/EventHandlerStore.ts

@@ -257,6 +257,18 @@ export class EventHandlerStore extends PuterStore {
         return existing;
     }
 
+    /**
+     * Drop every handler an app has published — what a deleted app leaves
+     * behind.
+     */
+    async deleteForApp(appUid: string): Promise<number> {
+        const result = await this.clients.db.write(
+            `DELETE FROM \`${TABLE}\` WHERE \`app_uid\` = ?`,
+            [appUid],
+        );
+        return result.affectedRows;
+    }
+
     // -- Reads -------------------------------------------------------
 
     /**

src/backend/stores/events/EventSubscriptionStore.ts

@@ -24,6 +24,7 @@ import type {
     DispatchSubscription,
     DurableSubscription,
     GenerationBump,
+    RemoteWatchAnnounce,
     SessionSubscription,
 } from './types.js';
 
@@ -49,6 +50,8 @@ import type {
  *     ev:g:{<ownerId>}             STR   subscription-set generation
  *     ev:dm:{<ownerId>}            HASH  subId -> token, durable rows cached here
  *     ev:dw:{<ownerId>}            STR   this region's durable cache is warm
+ *     ev:sc:{<ownerId>}            HASH  token -> session-row count, this region
+ *     ev:rw:{<ownerId>}            HASH  token -> {region: announcedAtMs}, peer-written
  *
  * The socket set is the one keyed by the holder — it is read on disconnect,
  * when all that is known is whose connection went — so its members name the
@@ -70,6 +73,15 @@ import type {
  * which the existing "drop the token once its hash is empty" rule gets right
  * for free. What durable rows add is the warm marker, which is how a region
  * tells "nobody is subscribed" apart from "this region has not looked yet".
+ *
+ * `ev:sc` and `ev:rw` extend this for cross-region session subscriptions.
+ * `ev:sc` counts this region's own live session rows per token — separate from
+ * `ev:w` because that set is shared with durable rows, so a `sadd` returning 1
+ * does not mean a _session_ watcher just arrived. Crossing 0<->1+ here is what
+ * announces to (or withdraws from) every peer, which writes the announcement
+ * into _its own_ `ev:rw`: which regions currently have a session watcher on one
+ * of this owner's tokens. A write by this owner reads `ev:rw` alongside `ev:w`
+ * to decide which peers, if any, get the raw event forwarded to them.
  */
 
 export type {
@@ -89,6 +101,17 @@ const socketKey = (userId: number | string, socketId: string): string =>
 const generationKey = (userId: number | string): string => `ev:g:{${userId}}`;
 const durableMapKey = (userId: number | string): string => `ev:dm:{${userId}}`;
 const durableWarmKey = (userId: number | string): string => `ev:dw:{${userId}}`;
+const sessionCountKey = (userId: number | string): string =>
+    `ev:sc:{${userId}}`;
+const remoteWatchKey = (userId: number | string): string => `ev:rw:{${userId}}`;
+
+const safeParseRegions = (raw: string): Record<string, number> | null => {
+    try {
+        return JSON.parse(raw) as Record<string, number>;
+    } catch {
+        return null;
+    }
+};
 
 /** `ev:s` members name the row they point at, and the keyspace it is in. */
 interface SocketRef {
@@ -110,6 +133,20 @@ const parseSocketRef = (ref: string): SocketRef => {
     };
 };
 
+/** A session token whose region-local watcher count just crossed to zero. */
+interface DroppedSessionToken {
+    ownerUserId: number;
+    token: string;
+}
+
+const toAnnounces = (
+    dropped: readonly DroppedSessionToken[],
+    op: 'add' | 'drop',
+): RemoteWatchAnnounce[] | undefined =>
+    dropped.length > 0
+        ? dropped.map((entry) => ({ token: entry.token, op }))
+        : undefined;
+
 /** Group refs by the keyspace they live in, so no pipeline crosses slots. */
 const byOwner = (refs: readonly SocketRef[]): Map<number, SocketRef[]> => {
     const grouped = new Map<number, SocketRef[]>();
@@ -149,6 +186,14 @@ export const DURABLE_CACHE_TTL_SECONDS = 24 * 60 * 60;
  */
 export const DURABLE_WARM_TTL_SECONDS = 6 * 60 * 60;
 
+/**
+ * How long a peer's remote-watch announcement is trusted without a re-announce.
+ * Well past the ~20 min socket refresh that re-asserts live tokens, so only a
+ * lost `drop` — or a peer that vanished outright — is ever caught by this
+ * rather than by the refresh or the `noWatch` repair.
+ */
+export const REMOTE_WATCH_TTL_SECONDS = 2 * 60 * 60;
+
 const subscriptionLimitReached = (): HttpError =>
     new HttpError(
         429,
@@ -204,13 +249,20 @@ export class EventSubscriptionStore extends PuterStore {
         );
         rows.sadd(watchedKey(ownerUserId), token);
         rows.expire(watchedKey(ownerUserId), SESSION_SUBSCRIPTION_TTL_SECONDS);
-        await rows.exec();
+        rows.hincrby(sessionCountKey(ownerUserId), token, 1);
+        rows.expire(
+            sessionCountKey(ownerUserId),
+            SESSION_SUBSCRIPTION_TTL_SECONDS,
+        );
+        const results = (await rows.exec()) ?? [];
+        const sessionCount = Number(results[4]?.[1]);
 
         await this.#keepDurableWindow(ownerUserId, [token]);
 
         return {
             userId: ownerUserId,
             generation: await this.bumpGeneration(ownerUserId),
+            announce: sessionCount === 1 ? [{ token, op: 'add' }] : undefined,
         };
     }
 
@@ -250,7 +302,7 @@ export class EventSubscriptionStore extends PuterStore {
      * app's — with the actor, where it belongs.
      */
     async remove(sub: SessionSubscription): Promise<GenerationBump> {
-        await this.#dropRefs(sub.holderUserId, sub.socketId, [
+        const dropped = await this.#dropRefs(sub.holderUserId, sub.socketId, [
             {
                 ownerUserId: sub.ownerUserId,
                 token: sub.token,
@@ -260,6 +312,7 @@ export class EventSubscriptionStore extends PuterStore {
         return {
             userId: sub.ownerUserId,
             generation: await this.bumpGeneration(sub.ownerUserId),
+            announce: toAnnounces(dropped, 'drop'),
         };
     }
 
@@ -273,13 +326,17 @@ export class EventSubscriptionStore extends PuterStore {
         previous: SessionSubscription,
         next: SessionSubscription,
     ): Promise<GenerationBump[]> {
-        await this.#dropRefs(previous.holderUserId, previous.socketId, [
-            {
-                ownerUserId: previous.ownerUserId,
-                token: previous.token,
-                subId: previous.subId,
-            },
-        ]);
+        const dropped = 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);
@@ -290,7 +347,13 @@ export class EventSubscriptionStore extends PuterStore {
             watchedKey(next.ownerUserId),
             SESSION_SUBSCRIPTION_TTL_SECONDS,
         );
-        await rows.exec();
+        rows.hincrby(sessionCountKey(next.ownerUserId), next.token, 1);
+        rows.expire(
+            sessionCountKey(next.ownerUserId),
+            SESSION_SUBSCRIPTION_TTL_SECONDS,
+        );
+        const results = (await rows.exec()) ?? [];
+        const nextCount = Number(results[4]?.[1]);
 
         const holder = this.clients.redis.pipeline();
         holder.sadd(
@@ -309,11 +372,24 @@ export class EventSubscriptionStore extends PuterStore {
 
         await this.#keepDurableWindow(next.ownerUserId, [next.token]);
 
+        const announceByOwner = new Map<number, RemoteWatchAnnounce[]>();
+        for (const { ownerUserId, token } of dropped)
+            announceByOwner.set(ownerUserId, [
+                ...(announceByOwner.get(ownerUserId) ?? []),
+                { token, op: 'drop' },
+            ]);
+        if (nextCount === 1)
+            announceByOwner.set(next.ownerUserId, [
+                ...(announceByOwner.get(next.ownerUserId) ?? []),
+                { token: next.token, op: 'add' },
+            ]);
+
         const owners = new Set([previous.ownerUserId, next.ownerUserId]);
         return Promise.all(
             [...owners].map(async (userId) => ({
                 userId,
                 generation: await this.bumpGeneration(userId),
+                announce: announceByOwner.get(userId),
             })),
         );
     }
@@ -332,12 +408,20 @@ export class EventSubscriptionStore extends PuterStore {
         ).map(parseSocketRef);
         if (refs.length === 0) return [];
 
-        await this.#dropRefs(holderUserId, socketId, refs);
+        const dropped = await this.#dropRefs(holderUserId, socketId, refs);
+        const announceByOwner = new Map<number, RemoteWatchAnnounce[]>();
+        for (const { ownerUserId, token } of dropped)
+            announceByOwner.set(ownerUserId, [
+                ...(announceByOwner.get(ownerUserId) ?? []),
+                { token, op: 'drop' },
+            ]);
+
         const bumps: GenerationBump[] = [];
         for (const ownerUserId of byOwner(refs).keys())
             bumps.push({
                 userId: ownerUserId,
                 generation: await this.bumpGeneration(ownerUserId),
+                announce: announceByOwner.get(ownerUserId),
             });
         return bumps;
     }
@@ -347,14 +431,59 @@ export class EventSubscriptionStore extends PuterStore {
         holderUserId: number,
         socketId: string,
         refs: readonly SocketRef[],
-    ): Promise<void> {
+    ): Promise<DroppedSessionToken[]> {
         await this.clients.redis.srem(
             socketKey(holderUserId, socketId),
             ...refs.map(socketRef),
         );
 
-        for (const [ownerUserId, owned] of byOwner(refs))
+        const dropped: DroppedSessionToken[] = [];
+        for (const [ownerUserId, owned] of byOwner(refs)) {
             await this.#dropRows(ownerUserId, owned);
+            dropped.push(
+                ...(await this.#dropSessionCounts(
+                    ownerUserId,
+                    owned.map((ref) => ref.token),
+                )),
+            );
+        }
+        return dropped;
+    }
+
+    /**
+     * Decrement this region's live-session count for each dropped token, and
+     * report which crossed to zero — those are what a peer needs to be told to
+     * stop for. A token can appear more than once (several rows on the same
+     * anchor), so each is decremented by its own occurrence count in one
+     * pipeline rather than one command per row.
+     */
+    async #dropSessionCounts(
+        ownerUserId: number,
+        tokens: readonly string[],
+    ): Promise<DroppedSessionToken[]> {
+        if (tokens.length === 0) return [];
+        const counts = new Map<string, number>();
+        for (const token of tokens)
+            counts.set(token, (counts.get(token) ?? 0) + 1);
+
+        const key = sessionCountKey(ownerUserId);
+        const tokenList = [...counts.keys()];
+        const pipeline = this.clients.redis.pipeline();
+        for (const token of tokenList)
+            pipeline.hincrby(key, token, -(counts.get(token) as number));
+        const results = (await pipeline.exec()) ?? [];
+
+        const zeroed: string[] = [];
+        const dropped: DroppedSessionToken[] = [];
+        tokenList.forEach((token, i) => {
+            const remaining = Number(results[i]?.[1]);
+            if (Number.isFinite(remaining) && remaining <= 0) {
+                zeroed.push(token);
+                dropped.push({ ownerUserId, token });
+            }
+        });
+        if (zeroed.length > 0) await this.clients.redis.hdel(key, ...zeroed);
+        return dropped;
     }
 
     /**
@@ -409,8 +538,16 @@ export class EventSubscriptionStore extends PuterStore {
      * TTL: a live row is proof its token belongs there, so a race that silently
      * dropped it (see `#dropRows`) heals itself on the next refresh even if
      * nothing catches it sooner.
+     *
+     * Returns every (owner, token) pair the socket still holds, so the caller
+     * can re-announce them to peers — the whole of how a remote-watch
+     * announcement survives longer than one refresh window without a second
+     * timer.
      */
-    async refresh(holderUserId: number, socketId: string): Promise<void> {
+    async refresh(
+        holderUserId: number,
+        socketId: string,
+    ): Promise<Array<{ ownerUserId: number; token: string }>> {
         const refs = (
             await this.clients.redis.smembers(socketKey(holderUserId, socketId))
         ).map(parseSocketRef);
@@ -420,6 +557,7 @@ export class EventSubscriptionStore extends PuterStore {
             SESSION_SUBSCRIPTION_TTL_SECONDS,
         );
 
+        const reasserted: Array<{ ownerUserId: number; token: string }> = [];
         for (const [ownerUserId, owned] of byOwner(refs)) {
             const tokens = [...new Set(owned.map((ref) => ref.token))];
             const pipeline = this.clients.redis.pipeline();
@@ -428,6 +566,10 @@ export class EventSubscriptionStore extends PuterStore {
                 watchedKey(ownerUserId),
                 SESSION_SUBSCRIPTION_TTL_SECONDS,
             );
+            pipeline.expire(
+                sessionCountKey(ownerUserId),
+                SESSION_SUBSCRIPTION_TTL_SECONDS,
+            );
             for (const token of tokens)
                 pipeline.expire(
                     tokenKey(ownerUserId, token),
@@ -436,7 +578,9 @@ export class EventSubscriptionStore extends PuterStore {
             await pipeline.exec();
 
             await this.#keepDurableWindow(ownerUserId, tokens);
+            for (const token of tokens) reasserted.push({ ownerUserId, token });
         }
+        return reasserted;
     }
 
     // -- Durable rows in the region cache ----------------------------
@@ -532,27 +676,102 @@ export class EventSubscriptionStore extends PuterStore {
     // -- Reads -------------------------------------------------------
 
     /**
-     * Whether anyone watches anything of this owner's at all. One command, and
-     * the only thing a cold process needs before it can answer from memory.
+     * Whether anyone watches anything of this owner's at all — locally, or a
+     * peer holding a session watcher on one of their tokens. One pipeline, two
+     * commands, still the only thing a cold process needs before it can answer
+     * from memory: a region with no local rows but a peer watching must not
+     * read as "nobody is subscribed".
      */
     async userHasAny(ownerUserId: number): Promise<boolean> {
-        return (await this.clients.redis.exists(watchedKey(ownerUserId))) === 1;
+        const pipeline = this.clients.redis.pipeline();
+        pipeline.exists(watchedKey(ownerUserId));
+        pipeline.exists(remoteWatchKey(ownerUserId));
+        const results = (await pipeline.exec()) ?? [];
+        return Number(results[0]?.[1]) === 1 || Number(results[1]?.[1]) === 1;
     }
 
     /**
-     * Which of an event's tokens anyone is watching — the dispatch hot path,
-     * and one command whatever the depth of the tree.
+     * Which of an event's tokens anyone is watching locally — the dispatch hot
+     * path, and one command whatever the depth of the tree. Delegates to
+     * {@link watchedFor} so no existing caller has to change.
      */
     async watchedTokens(
         ownerUserId: number,
         tokens: readonly string[],
     ): Promise<string[]> {
-        if (tokens.length === 0) return [];
-        const flags = await this.clients.redis.smismember(
-            watchedKey(ownerUserId),
-            ...tokens,
-        );
-        return tokens.filter((_token, i) => Number(flags[i]) === 1);
+        return (await this.watchedFor(ownerUserId, tokens)).local;
+    }
+
+    /**
+     * Which of an event's tokens anyone watches — here, and in which peers. One
+     * round trip, one cluster slot: `ev:w` and `ev:rw` share the owner's hash
+     * tag. Regions whose announcement has aged past
+     * {@link REMOTE_WATCH_TTL_SECONDS} are pruned from the answer in memory, not
+     * written back — a peer that is actually still watching re-announces on its
+     * own refresh well inside that window.
+     */
+    async watchedFor(
+        ownerUserId: number,
+        tokens: readonly string[],
+    ): Promise<{ local: string[]; remote: Map<string, string[]> }> {
+        if (tokens.length === 0) return { local: [], remote: new Map() };
+
+        const pipeline = this.clients.redis.pipeline();
+        pipeline.smismember(watchedKey(ownerUserId), ...tokens);
+        pipeline.hmget(remoteWatchKey(ownerUserId), ...tokens);
+        const results = (await pipeline.exec()) ?? [];
+
+        const flags = (results[0]?.[1] as number[] | undefined) ?? [];
+        const local = tokens.filter((_token, i) => Number(flags[i]) === 1);
+
+        const rawRemote =
+            (results[1]?.[1] as Array<string | null> | undefined) ?? [];
+        const cutoffMs = Date.now() - REMOTE_WATCH_TTL_SECONDS * 1000;
+        const remote = new Map<string, string[]>();
+        tokens.forEach((token, i) => {
+            const raw = rawRemote[i];
+            if (!raw) return;
+            const regions = safeParseRegions(raw);
+            if (!regions) return;
+            const live = Object.entries(regions)
+                .filter(
+                    ([, announcedAt]) =>
+                        typeof announcedAt === 'number' &&
+                        announcedAt >= cutoffMs,
+                )
+                .map(([region]) => region);
+            if (live.length > 0) remote.set(token, live);
+        });
+
+        return { local, remote };
+    }
+
+    /**
+     * Record (or clear) one peer's session watch on one of our tokens — the
+     * write side of {@link watchedFor}'s remote arm. Read-modify-write, since
+     * several peers can hold the same token; a race between two peers'
+     * announcements is bounded by the TTL, the periodic re-announce and the
+     * `noWatch` repair, the same three things that bound a lost `drop`.
+     */
+    async noteRemoteWatch(
+        ownerUserId: number,
+        token: string,
+        region: string,
+        op: 'add' | 'drop',
+    ): Promise<void> {
+        const key = remoteWatchKey(ownerUserId);
+        const raw = await this.clients.redis.hget(key, token);
+        const regions = raw ? (safeParseRegions(raw) ?? {}) : {};
+
+        if (op === 'drop') delete regions[region];
+        else regions[region] = Date.now();
+
+        if (Object.keys(regions).length === 0) {
+            await this.clients.redis.hdel(key, token);
+            return;
+        }
+        await this.clients.redis.hset(key, token, JSON.stringify(regions));
+        await this.clients.redis.expire(key, REMOTE_WATCH_TTL_SECONDS);
     }
 
     /** The rows behind a set of watched tokens, session and durable alike. */

src/backend/stores/events/types.ts

@@ -135,8 +135,16 @@ export interface DurableSubscription extends DispatchSubscription {
     createdAt: number;
 }
 
+/** A region-local session-watcher count crossed 0<->1+ for one token. */
+export interface RemoteWatchAnnounce {
+    token: string;
+    op: 'add' | 'drop';
+}
+
 /** One owner's generation after a change to the set of rows keyed under them. */
 export interface GenerationBump {
     userId: number;
     generation: number;
+    /** Session tokens whose local watcher count just crossed 0<->1+, if any. */
+    announce?: RemoteWatchAnnounce[];
 }

src/backend/stores/session/SessionStore.js

@@ -578,6 +578,42 @@ export class SessionStore extends PuterStore {
         return existing;
     }
 
+    /**
+     * A page of live worker sessions for one worker name, oldest id first —
+     * what the stray-session sweep pages through to find rows whose app no
+     * longer exists. `idx_sessions_kind_user` gives `kind = 'worker'` as a
+     * leading equality, so this range-scans worker sessions only, not the whole
+     * table; the `worker_name` predicate is evaluated on that slice. No
+     * migration.
+     *
+     * @param {{
+     *     workerName: string;
+     *     afterId?: number;
+     *     limit?: number;
+     * }} args
+     *   - `workerName` selects the worker; `afterId` is the keyset cursor (0 for
+     *       the first page); `limit` bounds the page size.
+     */
+    async listWorkerSessions({ workerName, afterId = 0, limit = 500 } = {}) {
+        if (!workerName) return [];
+        const workerNameExpr = this.clients.db.jsonTextExtract('`meta`', [
+            'worker_name',
+        ]);
+        const rows = await this.clients.db.read(
+            `SELECT \`id\`, \`uuid\`, \`user_id\`, \`app_uid\` FROM \`sessions\` ` +
+                `WHERE \`kind\` = 'worker' AND \`revoked_at\` IS NULL AND ` +
+                `\`app_uid\` IS NOT NULL AND ${workerNameExpr} = ? AND \`id\` > ? ` +
+                'ORDER BY `id` LIMIT ?',
+            [workerName, afterId, Math.max(1, Math.floor(limit))],
+        );
+        return rows.map((row) => ({
+            id: Number(row.id),
+            uuid: String(row.uuid),
+            userId: Number(row.user_id),
+            appUid: String(row.app_uid),
+        }));
+    }
+
     /**
      * Bump `last_activity` and slide `expires_at` per the row's kind in a
      * single UPDATE. Sliding kinds (web/app/asset) get their `expires_at`

src/backend/stores/session/SessionStore.test.ts

@@ -507,6 +507,107 @@ describe('SessionStore', () => {
         });
     });
 
+    describe('listWorkerSessions', () => {
+        it('pages live sessions for one worker name, oldest id first', async () => {
+            const user = await makeUser();
+            const workerName = `wk-${Math.random().toString(36).slice(2, 8)}`;
+            const appA = `app-${uuidv4()}`;
+            const appB = `app-${uuidv4()}`;
+            const a = await target.getOrCreateWorker(user.id, {
+                appUid: appA,
+                workerName,
+            });
+            const b = await target.getOrCreateWorker(user.id, {
+                appUid: appB,
+                workerName,
+            });
+
+            const rows = await target.listWorkerSessions({ workerName });
+            const uuids = rows.map((row: { uuid: string }) => row.uuid);
+            expect(uuids).toContain(a.uuid);
+            expect(uuids).toContain(b.uuid);
+            expect(rows.every((row: { id: number }) => Number.isFinite(row.id)))
+                .toBe(true);
+        });
+
+        it('does not return a revoked session', async () => {
+            const user = await makeUser();
+            const workerName = `wk-${Math.random().toString(36).slice(2, 8)}`;
+            const row = await target.getOrCreateWorker(user.id, {
+                appUid: `app-${uuidv4()}`,
+                workerName,
+            });
+            await target.removeByUuid(row.uuid);
+
+            const rows = await target.listWorkerSessions({ workerName });
+            expect(rows.map((r: { uuid: string }) => r.uuid)).not.toContain(
+                row.uuid,
+            );
+        });
+
+        it('never returns a different worker name', async () => {
+            const user = await makeUser();
+            const appUid = `app-${uuidv4()}`;
+            const wanted = `wk-${Math.random().toString(36).slice(2, 8)}`;
+            const other = `wk-${Math.random().toString(36).slice(2, 8)}`;
+            await target.getOrCreateWorker(user.id, {
+                appUid,
+                workerName: other,
+            });
+
+            const rows = await target.listWorkerSessions({
+                workerName: wanted,
+            });
+            expect(rows).toEqual([]);
+        });
+
+        it('a user-scoped worker (no app) is never returned', async () => {
+            // The stray-session sweep only ever finds rows with an app to
+            // check for existence — a worker session with no app has
+            // nothing for it to resolve.
+            const user = await makeUser();
+            const workerName = `wk-${Math.random().toString(36).slice(2, 8)}`;
+            await target.getOrCreateWorker(user.id, {
+                appUid: null,
+                workerName,
+            });
+
+            const rows = await target.listWorkerSessions({ workerName });
+            expect(rows).toEqual([]);
+        });
+
+        it('respects the keyset cursor and limit', async () => {
+            const user = await makeUser();
+            const workerName = `wk-${Math.random().toString(36).slice(2, 8)}`;
+            const first = await target.getOrCreateWorker(user.id, {
+                appUid: `app-${uuidv4()}`,
+                workerName,
+            });
+            const second = await target.getOrCreateWorker(user.id, {
+                appUid: `app-${uuidv4()}`,
+                workerName,
+            });
+
+            const page = await target.listWorkerSessions({
+                workerName,
+                limit: 1,
+            });
+            expect(page).toHaveLength(1);
+            expect(page[0].uuid).toBe(first.uuid);
+
+            const next = await target.listWorkerSessions({
+                workerName,
+                afterId: page[0].id,
+            });
+            expect(next.map((r: { uuid: string }) => r.uuid)).toContain(
+                second.uuid,
+            );
+            expect(next.map((r: { uuid: string }) => r.uuid)).not.toContain(
+                first.uuid,
+            );
+        });
+    });
+
     describe('error propagation (no silent swallow)', () => {
         // INSERT-IGNORE used to mask every constraint violation, not just
         // the partial-unique-index conflict the `getOrCreate*` paths rely

src/backend/types.ts

@@ -239,7 +239,12 @@ export interface IPreludeConfig {
      * an RCS agent provisioned in the Prelude account to actually use RCS.
      */
     preferredChannel?:
-        'sms' | 'rcs' | 'whatsapp' | 'viber' | 'zalo' | 'telegram';
+        | 'sms'
+        | 'rcs'
+        | 'whatsapp'
+        | 'viber'
+        | 'zalo'
+        | 'telegram';
 }
 
 /**
@@ -1151,6 +1156,12 @@ interface IConfigOptional {
      *   worker and delivery invokes it. Absent means off: publish stores rows
      *   and nothing is deployed, and the invoker keeps its null resolver, so
      *   worker-target deliveries stay retriable until something answers.
+     * - `forwardSession` — whether `onLocal` (session) subscriptions are
+     *   forwarded across regions. On unless set to `false`, which holds back
+     *   the per-token remote-watch index: a write in another region then never
+     *   reaches a session subscription here. Durable (`onPersistent`)
+     *   subscriptions and the rest of the addressed channel are unaffected
+     *   either way.
      */
     events?: {
         enabled?: boolean;
@@ -1158,6 +1169,7 @@ interface IConfigOptional {
         notificationsFoldIn?: boolean;
         kvHandles?: boolean;
         workerRuntime?: boolean;
+        forwardSession?: boolean;
         /** How long a handler has to answer an invocation. Default 30 s. */
         invokeTimeoutMs?: number;
         /**
@@ -1227,8 +1239,7 @@ export interface WithLifecycle extends Object {
 }
 
 export interface WithCostsReporting extends WithLifecycle {
-    getReportedCosts?: () =>
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+    getReportedCosts?: () => // eslint-disable-next-line @typescript-eslint/no-explicit-any
         | Promise<Record<string, any>[]>
         // eslint-disable-next-line @typescript-eslint/no-explicit-any
         | Record<string, any>[];

src/docs/src/Events.md

@@ -164,7 +164,7 @@ The handler is called with `{ event }`. A filesystem change carries:
 | --- | --- | --- |
 | `id` | String | Unique id for the event. |
 | `subject` | String | The subject the change was projected onto, naming the node it happened to (`fs:<uid>:<op>`) — not the subject string you subscribed with. |
-| `op` | String | `add`, `write`, `move`, or `remove`. |
+| `op` | String | `add`, `write`, `move`, or `remove`. `move` covers a move and an in-place rename. |
 | `uid` | String | The uid of the node that changed. |
 | `path` | String | The path of the node that changed. |
 | `from` | String | On a `move`, the path the node left. Only present when the subscription was watching that side — a subscription on the destination folder alone is not told where the node came from. |
@@ -261,6 +261,8 @@ Puter runs in several places, and a client connects to whichever one is nearest.
 
 The one consequence worth knowing is the one already stated: a `single` delivery is **at-least-once**. Undelivered events are held where the change happened, so a deployment going down loses only what it was still holding — the subscription itself, and everything already delivered, is unaffected. Handlers are asked to be idempotent for this reason, and `event.id` is the key to deduplicate on.
 
+Ordering follows the same shape: a subscription's own deliveries stay in order within the region that emits them, but the ordering is best effort across regions, and the 250 ms coalescing window is applied per region rather than globally. Two writes made moments apart can therefore arrive coalesced into one event in a region near the writer and as two separate ones somewhere farther away.
+
 ## Limits
 
 Subscriptions per connection, persistent subscriptions per account, published handlers per app, subscribe calls per minute, and how much one event may fan out are all capped — see [Rate Limits and Quotas](/rate-limits-and-quotas/). Deliveries are coalesced over 250 ms per subject, so a multipart upload or a save loop arrives as one event rather than one per write.

src/docs/src/Events/onLocal.md

@@ -6,7 +6,7 @@ platforms: [websites, apps, nodejs]
 
 <div class="info">The Events API is in beta. Event shapes, limits, and behavior may change between releases.</div>
 
-Subscribes to a subject and calls `handler` every time something matching it changes. The subscription belongs to this client's connection: nothing is stored, nothing runs while the page is closed, and it ends when the connection does. See [Events](/Events/) for the subject grammar and the event shape.
+Subscribes to a subject and calls `handler` every time something matching it changes. The subscription belongs to this client's connection: nothing is stored, nothing runs while the page is closed, and it ends when the connection does. A change made from another device, another browser, or another part of the world reaches it too — one made in another region typically arrives a few hundred milliseconds later than one made locally. See [Events](/Events/) for the subject grammar and the event shape.
 
 Not for a Puter worker: a worker invocation is short-lived, so a subscription here only lasts as long as that one invocation. To react to changes from a worker, use [`onPersistent()`](/Events/onPersistent/) with a `worker` target and a published handler.
 
@@ -142,8 +142,9 @@ The promise rejects with `{ message, code }`:
             // (1) Exactly one key, and separately every key under a prefix.
             const one = await puter.events.onLocal('kv:cart', ({ event }) =>
                 puter.print(`${event.op}: ${event.key}<br>`));
-            const many = await puter.events.onLocal('kv:cart:*', ({ event }) =>
-                puter.print(`under cart: ${event.key}<br>`));
+            const many = await puter.events.onLocal(
+                `kv:${puter.appID}:cart:*`,
+                ({ event }) => puter.print(`under cart: ${event.key}<br>`));
 
             // (2) `cart` reaches the first, `cart:items` only the second.
             await puter.kv.set('cart', { total: 0 });

src/docs/src/Events/onPersistent.md

@@ -8,6 +8,8 @@ platforms: [websites, apps, nodejs, workers]
 
 Creates a subscription that outlives this connection. It is stored against the account, keeps matching while your app is closed, and runs a handler your app published with [`puter.events.handlers.publish()`](/Events/handlers/). Contrast [`puter.events.onLocal()`](/Events/onLocal/), which lives and dies with the page.
 
+The subscription is live immediately in the region it was created in. A change made in another region in the first moment after this call resolves may take a little longer to reach it — usually well under a second — while that region catches up.
+
 See [Events](/Events/) for the subject grammar and the event shape.
 
 ## Syntax
@@ -37,7 +39,7 @@ await puter.perms.request(['events:background']);
 
 The user can revoke it wherever they manage an app's access. Doing so suspends every worker-target subscription that app holds for them with `permission_revoked`; re-granting the permission does not resume them, so subscribe again. A subscription that only wants deliveries while your app is open needs no consent at all: pass `targets: ['socket']`.
 
-A background delivery runs as a session, the same as any other your app is granted — it shows up in the user's own sessions list as a worker session, and revoking it there stops background handlers for your app the same way withdrawing `events:background` does. Withdrawing `events:background` or uninstalling the app revokes that session in turn, so a copied-out token stops working too.
+A background delivery runs as a session, the same as any other your app is granted — it shows up in the user's own sessions list as a worker session, and revoking it there stops background handlers for your app the same way withdrawing `events:background` does. Withdrawing `events:background` or uninstalling the app revokes that session in turn, so a copied-out token stops working too — and so does destroying the app's events worker or deleting the app outright.
 
 ## Where the handler runs, and what it is handed
 
@@ -61,7 +63,7 @@ A `single` delivery is owed to exactly one consumer, so it stays owed until it i
 
 - Calling `ack()` takes the delivery.
 - Returning **without** calling it acknowledges it anyway — a handler that finished did the work.
-- **Throwing acknowledges nothing.** The lease lapses after 30 seconds and the delivery is offered again, so a handler that throws sees the same event twice. `event.id` is stable across redeliveries; use it to make the second one a no-op.
+- **Throwing acknowledges nothing.** The lease lapses after 60 seconds — twice the handler invocation timeout — and the delivery is offered again, so a handler that throws sees the same event twice. `event.id` is stable across redeliveries; use it to make the second one a no-op.
 
 In the events worker the same three outcomes are the response status: `2xx` takes the delivery, `4xx` refuses it (it is dropped with a `gap` marker carrying `reason: 'handler_rejected'`), and `5xx`, `429` or no answer within 30 seconds means "not now" — the delivery is retried after 2 seconds, doubling to at most 5 minutes. **Five failures in a row, refusals included, suspend the subscription** with `failures`; the developer is notified and republishing the handler puts it back in service.
 

src/docs/src/Events/workers.md

@@ -37,6 +37,8 @@ puter.events.workers.destroy(appUid)
 
 Removes **every** handler the named app has published, in one call — the same consequences as calling [`puter.events.handlers.remove()`](/Events/handlers/) on each of them: a name nothing is bound to is deleted outright, and a name with subscriptions on it is deleted with those subscriptions *suspended* (`suspendedReason: 'handler_not_found'`), never dropped. Publishing new handlers for the app afterwards resumes them, exactly as republishing a single removed handler would.
 
+It also retires the worker session it was running background deliveries under, for every holder — see [`onPersistent()`](/Events/onPersistent/) — the same session revoking it from the user's sessions list would end. The session is not gone for good: the first delivery after a republish mints a fresh one.
+
 Resolves to `{ appUid, removed, suspended }` — `removed` is how many handlers were deleted, `suspended` how many subscriptions that left suspended across all of them. An app with nothing published rejects with `events_handler_not_found`.
 
 ## Errors

src/docs/src/rate-limits-and-quotas.md

@@ -247,9 +247,9 @@ A `kv:` subject is indexed on the first **6** `:`-segments, or **160 bytes**, of
 
 **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.
 
-The two per-event ceilings — matched subscriptions and filter evaluations — do not fail your call: they truncate the delivery and send a `gap` marker in its place, with `reason: 'matched_subscription_limit'` or `reason: 'filter_evaluation_limit'` respectively — an event with `op: 'gap'` and no `uid` or `path`. A gap means something happened that you were not told the details of, so a client that must not miss changes should re-read the anchor when it sees one rather than treat the silence as "nothing changed".
+The two per-event ceilings — matched subscriptions and filter evaluations — do not fail your call: they truncate the delivery and send a `gap` marker in its place, with `reason: 'matched_subscription_limit'` or `reason: 'filter_evaluation_limit'` respectively — an event with `op: 'gap'` and no `uid` or `path`. A gap means something happened that you were not told the details of, so a client that must not miss changes should re-read the anchor when it sees one rather than treat the silence as "nothing changed". Both ceilings are counted **per region**: a change is evaluated against every matching region's own copy of your subscriptions, so an account with subscribers spread across several regions can see more than 50 matched, or 200 evaluated, in total for one event, even though no single region ever exceeds its own cap.
 
-A **background delivery** — one that runs your app's handler with nobody there — takes the user's consent, the per-app permission `events:background`, and a subscription targeting `worker` without it is refused with `events_background_consent_required`. The handler runs as your app's own session for that user — the same reach it has from a tab, not a credential cut down to this one subscription's grant — and that session is what the consent authorizes running unattended; it shows up in the user's sessions list as a worker session, and revoking it there stops every background delivery for your app the same way withdrawing the permission does. A handler has **30 seconds** to answer each invocation. Answering `2xx` takes the delivery; `4xx` refuses it, and it is dropped with a `gap` marker carrying `reason: 'handler_rejected'` rather than sent again to the same answer; `5xx`, `429` and a timeout are all "not now", and the delivery is held **2 seconds** before the next attempt, doubling each time up to **5 minutes**. **Five failures in a row** — refusals included — suspend the subscription with `failures`, hold what it is owed under the suspended-backlog rules above, and notify the app's developer. Publishing a handler is all the deployment there is: the app's events worker is brought up the first time a delivery needs it, and again if it has been idle long enough to be evicted, so the first background delivery after a publish pays a short cold start. Nothing else can invoke it — it answers one platform route, and only the platform can reach it.
+A **background delivery** — one that runs your app's handler with nobody there — takes the user's consent, the per-app permission `events:background`, and a subscription targeting `worker` without it is refused with `events_background_consent_required`. The handler runs as your app's own session for that user — the same reach it has from a tab, not a credential cut down to this one subscription's grant — and that session is what the consent authorizes running unattended; it shows up in the user's sessions list as a worker session, and revoking it there stops every background delivery for your app the same way withdrawing the permission does. Destroying the app's events worker ([`puter.events.workers.destroy()`](/Events/workers/)) retires that session too, and deleting the app ends it along with every subscription and anything they were owed. A handler has **30 seconds** to answer each invocation. Answering `2xx` takes the delivery; `4xx` refuses it, and it is dropped with a `gap` marker carrying `reason: 'handler_rejected'` rather than sent again to the same answer; `5xx`, `429` and a timeout are all "not now", and the delivery is held **2 seconds** before the next attempt, doubling each time up to **5 minutes**. **Five failures in a row** — refusals included — suspend the subscription with `failures`, hold what it is owed under the suspended-backlog rules above, and notify the app's developer. Publishing a handler is all the deployment there is: the app's events worker is brought up the first time a delivery needs it, and again if it has been idle long enough to be evicted, so the first background delivery after a publish pays a short cold start. Nothing else can invoke it — it answers one platform route, and only the platform can reach it.
 
 A `single` subscription is delivered to exactly one consumer, which has **60 seconds** — twice the handler invocation timeout, so a slow but successful handler is never re-invoked mid-run — to acknowledge each delivery before it is offered again, twice to a connected client and then to the subscription's handler. Until it is acknowledged it is held for you, so a consumer that is away is a backlog that grows: **10,000** undelivered deliveries per subscription, after which the oldest are dropped and one `gap` marker with `reason: 'backlog_overflow'` takes their place. Each region also holds at most **1,000,000** undelivered deliveries across every subscription it serves, and sheds the oldest first — with the same marker — before it reaches that. A redelivery after a missed acknowledgement is normal and expected: deliveries are at-least-once, `event.id` is stable across them, and a handler that runs twice on the same id should do nothing the second time.
 

src/puter-js/src/modules/events/types.js

@@ -24,6 +24,7 @@
  *   naming the node it happened to — `fs:<uid>:<op>`. Not the subject string
  *   you subscribed with.
  * @property {'add' | 'write' | 'move' | 'remove' | 'meta'} op What happened.
+ *   `move` covers a move and an in-place rename.
  * @property {string} uid The uid of the node the event is about.
  * @property {string} path The path of the node the event is about.
  * @property {string} [from] On a `move`, the path the node left — present only