Merge juancastro/put-2172-notifications-settings into juancastro/put-2258-push-from-app-subdomains

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

@@ -942,7 +942,7 @@ describe('persistent delivery routing', () => {
         expect(sockets[0].sent.filter(s => s.verb === 'events.ack')).toEqual([]);
     });
 
-    it('runs nothing for a delivery the server marked as handled by the worker', async () => {
+    it('runs nothing for a delivery the server marked as past the handler-chain limit', async () => {
         const events = makeModule();
         const seen = [];
         events.channel.registerDurable('app-1#sub', ({ event }) => seen.push(event));
@@ -1019,7 +1019,7 @@ describe('persistent delivery routing', () => {
         ]);
     });
 
-    it('acknowledges nothing when the handler throws, so it is delivered again', async () => {
+    it('declines as failed and not retryable, and acknowledges nothing, when the handler throws', async () => {
         const events = makeModule();
         const errors = vi.spyOn(console, 'error').mockImplementation(() => {});
         events.channel.registerDurable('app-1#sub', async () => {
@@ -1031,15 +1031,98 @@ describe('persistent delivery routing', () => {
             durable('app-1#sub', '/user/a/one.txt', {
                 ackRequired: true,
                 ackId: 'entry-3',
+                origin: 'west',
             }),
         );
         await Promise.resolve();
         await Promise.resolve();
 
         expect(sockets[0].sent.filter(s => s.verb === 'events.ack')).toEqual([]);
+        expect(sockets[0].sent.filter(s => s.verb === 'events.nack')).toMatchObject([
+            {
+                payload: {
+                    subId: 'app-1#sub',
+                    id: 'entry-3',
+                    reason: 'failed',
+                    retry: false,
+                    origin: 'west',
+                },
+            },
+        ]);
         expect(errors).toHaveBeenCalled();
     });
 
+    it('declines a handler that throws synchronously, but not one that acked first', async () => {
+        const events = makeModule();
+        vi.spyOn(console, 'error').mockImplementation(() => {});
+        events.channel.registerDurable('app-1#sync', () => {
+            throw new Error('handler bug');
+        });
+        events.channel.registerDurable('app-1#late', async ({ ack }) => {
+            await ack();
+            throw new Error('after ack');
+        });
+
+        sockets[0].fire(
+            'events.delivery',
+            durable('app-1#sync', '/user/a/one.txt', { ackRequired: true, ackId: 'entry-s' }),
+        );
+        sockets[0].fire(
+            'events.delivery',
+            durable('app-1#late', '/user/a/two.txt', { ackRequired: true, ackId: 'entry-l' }),
+        );
+        for ( let i = 0; i < 5; i++ ) await Promise.resolve();
+
+        expect(sockets[0].sent.filter(s => s.verb === 'events.nack').map(s => s.payload.id))
+            .toEqual(['entry-s']);
+        expect(sockets[0].sent.filter(s => s.verb === 'events.ack').map(s => s.payload.id))
+            .toEqual(['entry-l']);
+    });
+
+    it('declines a retryable failure as one to try again', async () => {
+        const events = makeModule();
+        vi.spyOn(console, 'error').mockImplementation(() => {});
+        events.channel.registerDurable('app-1#net', async () => {
+            throw new TypeError('Failed to fetch');
+        });
+        events.channel.registerDurable('app-1#later', async () => {
+            throw Object.assign(new Error('try later'), { retry: true });
+        });
+
+        for ( const [subId, ackId] of [['app-1#net', 'entry-n'], ['app-1#later', 'entry-r']] ) {
+            sockets[0].fire(
+                'events.delivery',
+                durable(subId, '/user/a/one.txt', { ackRequired: true, ackId }),
+            );
+        }
+        for ( let i = 0; i < 5; i++ ) await Promise.resolve();
+
+        expect(sockets[0].sent.filter(s => s.verb === 'events.nack').map(s => s.payload))
+            .toMatchObject([
+                { id: 'entry-n', reason: 'failed', retry: true },
+                { id: 'entry-r', reason: 'failed', retry: true },
+            ]);
+    });
+
+    it('declines a queued delivery it has no handler for, at once', async () => {
+        const events = makeModule();
+        // Opens the connection without registering anything for `app-1#sub`.
+        events.channel.registerDurable('app-1#other', () => {});
+
+        sockets[0].fire(
+            'events.delivery',
+            durable('app-1#sub', '/user/a/one.txt', { ackRequired: true, ackId: 'entry-4' }),
+        );
+        // A copy everyone gets is not declined: nothing is waiting on it.
+        sockets[0].fire('events.delivery', durable('app-1#sub', '/user/a/two.txt'));
+        await Promise.resolve();
+
+        expect(sockets[0].sent.filter(s => s.verb === 'events.nack')).toMatchObject([
+            { payload: { subId: 'app-1#sub', id: 'entry-4', reason: 'no_handler', retry: false } },
+        ]);
+        expect(sockets[0].sent.filter(s => s.verb === 'events.ack')).toEqual([]);
+    });
+
     it('keeps routing across a reconnect without subscribing again', async () => {
         const events = makeModule();
         const seen = [];

src/puter-js/src/modules/events/lib/channel.js

@@ -2,6 +2,7 @@ import { io } from 'socket.io-client';
 import { PuterJSError } from '../../../lib/PuterJSError.js';
 import { socketAutoUnref } from '../../../lib/socketOptions.js';
 import { uuidv4 } from '../../../lib/utils.js';
+import { isRetryableError } from './retryable.js';
 import { EventSubscription } from './subscription.js';
 
 /** @typedef {import('../types.js').EventGapMarker} EventGapMarker */
@@ -35,11 +36,12 @@ const subscribePayload = (sub) => ({
     ...(sub.includeValue ? { includeValue: true } : {}),
 });
 
-// The wire, fixed by the server: three verbs answered with an ack, one channel
+// The wire, fixed by the server: four verbs answered with an ack, one channel
 // events arrive on.
 const SUBSCRIBE_VERB = 'events.subscribe';
 const UNSUBSCRIBE_VERB = 'events.unsubscribe';
 const ACK_VERB = 'events.ack';
+const NACK_VERB = 'events.nack';
 const DELIVERY_CHANNEL = 'events.delivery';
 const ENDED_CHANNEL = 'events.ended';
 
@@ -501,7 +503,7 @@ export class EventChannel {
 
     /**
      * @internal
-     * @param {{ subId?: string, event?: unknown, ackRequired?: boolean, ackId?: string, skipHandler?: boolean }} envelope
+     * @param {{ subId?: string, event?: unknown, ackRequired?: boolean, ackId?: string, origin?: string, skipHandler?: boolean }} envelope
      * @returns {void}
      */
     route (envelope) {
@@ -516,9 +518,15 @@ export class EventChannel {
             return;
         }
         const registered = this.durable.get(subId);
-        // An event for something this client has already unsubscribed from:
-        // in flight when `off()` was called, and no longer anybody's.
-        if ( registered ) this.runDurable(registered, envelope);
+        if ( registered ) {
+            this.runDurable(registered, envelope);
+            return;
+        }
+        // A queued delivery this client has no handler for goes back at once,
+        // so the events worker takes it instead of waiting out the lease.
+        if ( envelope.ackRequired && typeof envelope.ackId === 'string' ) {
+            this.nack(subId, envelope.ackId, envelope.origin, 'no_handler', false);
+        }
     }
 
     /**
@@ -539,14 +547,14 @@ export class EventChannel {
 
     /**
      * Run a persistent subscription's handler on one delivery, unless the
-     * server marked it `skipHandler`: the app's events worker takes it, or it
-     * is past the handler-chain limit.
+     * server marked it `skipHandler` because it is past the handler-chain
+     * limit.
      *
      * The handler is handed the same environment its published copy gets in the
      * app's events worker, so one body runs unchanged in either place. A
-     * delivery owed to exactly one consumer carries `ack`: calling it settles
-     * the delivery, resolving without calling it settles it anyway, and
-     * throwing settles nothing — the lease lapses and it is delivered again.
+     * queued delivery is acknowledged when the handler resolves (`ack` only
+     * releases it early), and a throw hands it back as failed: retried if
+     * `isRetryableError` says so, dropped otherwise.
      *
      * @internal
      * @param {DurableRegistration} registration
@@ -554,7 +562,7 @@ export class EventChannel {
      * @returns {void}
      */
     runDurable (registration, envelope) {
-        // `single` always carries `ackRequired` here and must still run —
+        // A queued delivery always carries `ackRequired` and must still run —
         // only an unacknowledged broadcast copy is ever marked this way.
         if ( envelope.skipHandler === true && ! envelope.ackRequired ) return;
 
@@ -585,12 +593,23 @@ export class EventChannel {
         settleHandler(
             () => registration.handler({ ...delivery, ack }),
             () => ack(),
+            error => {
+                if ( acked ) return;
+                acked = true;
+                this.nack(
+                    registration.subId,
+                    /** @type {string} */ (envelope.ackId),
+                    envelope.origin,
+                    'failed',
+                    isRetryableError(error),
+                );
+            },
         );
     }
 
     /**
      * Tell the server a delivery was taken. Best effort: a lost ack is a
-     * redelivery, which `single` callers are told to expect, and there is
+     * redelivery, which `queue` callers are told to expect, and there is
      * nothing a failure here leaves for the caller to fix.
      *
      * @internal
@@ -613,6 +632,31 @@ export class EventChannel {
         }
     }
 
+    /**
+     * Hand a queued delivery back so the server gives it to the events worker
+     * now. Best effort: a lost decline costs only the wait for the lease.
+     *
+     * @internal
+     * @param {string} subId
+     * @param {string} ackId
+     * @param {string | undefined} origin Echoed back untouched, as for `ack`.
+     * @param {'no_handler' | 'failed'} reason `failed` counts as a handler
+     *   failure; `no_handler` does not.
+     * @param {boolean} retry Whether a failure is worth trying again.
+     * @returns {Promise<void>}
+     */
+    async nack (subId, ackId, origin, reason, retry) {
+        try {
+            await this.request(
+                NACK_VERB,
+                { subId, id: ackId, reason, retry, ...(origin ? { origin } : {}) },
+                DEFAULT_TIMEOUT_MS,
+            );
+        } catch (error) {
+            console.warn('[puter.events] could not decline a delivery', error);
+        }
+    }
+
     /**
      * The connection is not coming back for now: fail what is waiting on it
      * and end every session subscription. A persistent registration is only
@@ -734,24 +778,26 @@ const timeoutFor = (sub) =>
 /**
  * Run a handler and, if it finishes without throwing, do whatever the delivery
  * still needs. A handler that throws is the app's bug: it is reported, and
- * nothing is settled on its behalf.
+ * `onFailed` hands the delivery back.
  *
  * @param {() => unknown} run
  * @param {() => unknown} [onResolved]
+ * @param {(error: unknown) => unknown} [onFailed]
  * @returns {void}
  */
-const settleHandler = (run, onResolved) => {
+const settleHandler = (run, onResolved, onFailed) => {
+    const failed = error => {
+        console.error('[puter.events] subscription handler failed', error);
+        onFailed?.(error);
+    };
     try {
         const result = run();
         if ( result instanceof Promise ) {
-            result.then(
-                () => onResolved?.(),
-                error => console.error('[puter.events] subscription handler failed', error),
-            );
+            result.then(() => onResolved?.(), failed);
             return;
         }
         onResolved?.();
     } catch (error) {
-        console.error('[puter.events] subscription handler failed', error);
+        failed(error);
     }
 };

src/puter-js/src/modules/events/lib/retryable.js

@@ -0,0 +1,37 @@
+/**
+ * Whether a handler failure is worth trying again. Anything else a handler
+ * throws drops the event. Shared by this client and the events worker runtime,
+ * which inlines this file, so it has no imports.
+ *
+ * Retryable: `retry: true` on the error; a 429 or 5xx `status`; a network,
+ * timeout or rate-limit error code; a `TimeoutError`; or a `TypeError` from a
+ * failed `fetch`. `terminal: true` always drops.
+ *
+ * @param {unknown} error What the handler threw.
+ * @returns {boolean}
+ */
+export const isRetryableError = (error) => {
+    if ( !error || typeof error !== 'object' ) return false;
+    const err = /** @type {Record<string, unknown>} */ (error);
+    if ( err.terminal === true || err.code === 'events_terminal' ) return false;
+    if ( err.retry === true ) return true;
+
+    const status = Number(err.status ?? err.statusCode);
+    if ( status === 429 || (status >= 500 && status < 600) ) return true;
+    if ( typeof err.code === 'string' && RETRYABLE_CODES.includes(err.code) ) return true;
+    if ( err.name === 'TimeoutError' ) return true;
+    return err.name === 'TypeError' &&
+        typeof err.message === 'string' &&
+        /fetch|network/i.test(err.message);
+};
+
+const RETRYABLE_CODES = [
+    'network_error',
+    'request_timeout',
+    'too_many_requests',
+    'rate_limited',
+    'events_connection_failed',
+    'ECONNRESET',
+    'ECONNREFUSED',
+    'ETIMEDOUT',
+];

src/puter-js/src/modules/events/lib/retryable.test.js

@@ -0,0 +1,39 @@
+import { describe, expect, it } from 'vitest';
+import { isRetryableError } from './retryable.js';
+
+const withFields = (fields, ErrorType = Error) =>
+    Object.assign(new ErrorType(fields.message ?? 'boom'), fields);
+
+describe('which handler failures are retried', () => {
+    it.each([
+        ['an error marked retry: true', withFields({ retry: true })],
+        ['a 429', withFields({ status: 429 })],
+        ['a 503', withFields({ status: 503 })],
+        ['a 500 as statusCode', withFields({ statusCode: 500 })],
+        ['a puter.js network error', withFields({ code: 'network_error' })],
+        ['a puter.js request timeout', withFields({ code: 'request_timeout' })],
+        ['a rate limit', withFields({ code: 'too_many_requests' })],
+        ['a dropped events connection', withFields({ code: 'events_connection_failed' })],
+        ['a reset socket', withFields({ code: 'ECONNRESET' })],
+        ['an abort signal timeout', withFields({ name: 'TimeoutError' })],
+        ['a browser fetch failure', new TypeError('Failed to fetch')],
+        ['a node fetch failure', new TypeError('fetch failed')],
+        ['a fetch network error', new TypeError('NetworkError when attempting to fetch resource.')],
+    ])('retries %s', (_label, error) => {
+        expect(isRetryableError(error)).toBe(true);
+    });
+
+    it.each([
+        ['a plain error', new Error('bad input')],
+        ['a 400', withFields({ status: 400 })],
+        ['a 404', withFields({ status: 404 })],
+        ['an unrelated TypeError', new TypeError('x is not a function')],
+        ['an unknown code', withFields({ code: 'subject_does_not_exist' })],
+        ['a thrown string', 'nope'],
+        ['undefined', undefined],
+        ['terminal: true, even on a 503', withFields({ terminal: true, status: 503 })],
+        ['events_terminal, even with retry: true', withFields({ code: 'events_terminal', retry: true })],
+    ])('drops %s', (_label, error) => {
+        expect(isRetryableError(error)).toBe(false);
+    });
+});

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

@@ -12,15 +12,14 @@ import { resolveSubject } from './lib/resolveSubject.js';
  *
  * Unlike `onLocal()`, the subscription is stored against the account, keeps
  * matching while the app is closed, and is ended by
- * `puter.events.unsubscribe()` rather than by navigating away. By default
- * (`broadcast`, targeting the worker) the app's published handler, named by
- * `handlerName`, is the one that runs — whether or not this client is open.
+ * `puter.events.unsubscribe()` rather than by navigating away. Each event runs
+ * the app's published handler, named by `handlerName`, in its events worker,
+ * and is also sent to every connected client. With `queue`, each event goes
+ * to exactly one consumer instead.
  *
- * Passing a function as `handler` also runs it in this client, on the
- * deliveries the server leaves to clients (see
- * [Where the handler runs](/Events/onPersistent/#where-the-handler-runs)):
- * the same body, the same `{ event, ctx }`, plus `user`, `fetch` and — for a
- * subscription owed to one consumer — `ack`.
+ * Passing a function as `handler` also runs it in this client for each
+ * delivery it receives: the same body, the same `{ event, ctx }`, plus `user`,
+ * `fetch` and — for a `queue` subscription — `ack`.
  *
  * `context` is evaluated **here, now** — serialized once and delivered to every
  * invocation as a frozen `ctx`. It never re-evaluates, so a value read from the
@@ -46,12 +45,11 @@ export async function onPersistent (options = {}) {
             'events_handler_name_required',
         );
     }
-    // A `single` delivery is owed to exactly one consumer, and the handler is
-    // the only one always there to take it — checked here rather than only
-    // discovered on the round trip.
-    if ( options.delivery === 'single' && ! handlerName ) {
+    // A queued delivery is owed to exactly one consumer, and the handler is
+    // the only one always there to take it.
+    if ( (options.queue === true || options.delivery === 'single') && ! handlerName ) {
         throw new PuterJSError(
-            'A `single` subscription needs a `handlerName`',
+            'A `queue` subscription needs a `handlerName`',
             'events_handler_required',
         );
     }
@@ -62,8 +60,8 @@ export async function onPersistent (options = {}) {
 
     const body = {
         subject,
+        ...(typeof options.queue === 'boolean' ? { queue: options.queue } : {}),
         ...(options.delivery ? { delivery: options.delivery } : {}),
-        ...(options.targets ? { targets: options.targets } : {}),
         ...(handlerName ? { handlerName } : {}),
         ...(inline ? { handlerHash: inline.hash } : {}),
         ...(options.expiresAt !== undefined && options.expiresAt !== null

src/puter-js/src/modules/events/persistent.test.js

@@ -76,11 +76,10 @@ describe('onPersistent', () => {
         expect(sub).toBe(view);
     });
 
-    it('carries delivery, targets, handlerName, expiry and includeValue when given', async () => {
+    it('carries queue, handlerName, expiry and includeValue when given', async () => {
         await makeModule().onPersistent({
             subject: SUBJECT,
-            delivery: 'single',
-            targets: ['worker'],
+            queue: true,
             handlerName: 'ingestUpload',
             expiresAt: 4102444800,
             includeValue: true,
@@ -88,14 +87,28 @@ describe('onPersistent', () => {
 
         expect(bodyOf()).toEqual({
             subject: SUBJECT,
-            delivery: 'single',
-            targets: ['worker'],
+            queue: true,
             handlerName: 'ingestUpload',
             expiresAt: 4102444800,
             includeValue: true,
         });
     });
 
+    it('still passes the older `delivery` option and drops `targets`', async () => {
+        await makeModule().onPersistent({
+            subject: SUBJECT,
+            delivery: 'single',
+            targets: ['worker'],
+            handlerName: 'ingestUpload',
+        });
+
+        expect(bodyOf()).toEqual({
+            subject: SUBJECT,
+            delivery: 'single',
+            handlerName: 'ingestUpload',
+        });
+    });
+
     it('sends an inline handler as a hash, never as source', async () => {
         await makeModule().onPersistent({
             subject: SUBJECT,
@@ -146,12 +159,13 @@ describe('onPersistent', () => {
         expect(bodyOf().handlerHash).toMatch(/^[0-9a-f]{64}$/);
     });
 
-    it('refuses a `single` delivery with no handlerName to match it against', async () => {
-        const error = await rejects(() =>
-            makeModule().onPersistent({ subject: SUBJECT, delivery: 'single' }),
-        );
-
-        expect(error.code).toBe('events_handler_required');
+    it('refuses a `queue` subscription with no handlerName to match it against', async () => {
+        for ( const options of [{ queue: true }, { delivery: 'single' }] ) {
+            const error = await rejects(() =>
+                makeModule().onPersistent({ subject: SUBJECT, ...options }),
+            );
+            expect(error.code).toBe('events_handler_required');
+        }
         expect(mockRequest).not.toHaveBeenCalled();
     });
 

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

@@ -157,10 +157,10 @@
  *   subscription — the ambient one when the handler runs in a client.
  * @property {(input: unknown, init?: unknown) => Promise<unknown>} [fetch]
  *   `puter.net.fetch` where it exists, the environment's `fetch` otherwise.
- * @property {() => Promise<void>} [ack] Settles the delivery. Present only on a
- *   `single` subscription: calling it hands the delivery back as taken,
- *   returning without calling it does the same, and throwing does neither — the
- *   delivery is offered again when its lease lapses.
+ * @property {() => Promise<void>} [ack] Optional. Marks a `queue` delivery done
+ *   before the handler finishes, so a later throw does not undo it. Resolving
+ *   marks it done anyway; a throw drops it unless it is retryable (a network,
+ *   timeout, rate-limit or 5xx error, or one thrown with `retry: true`).
  */
 
 /**
@@ -196,20 +196,20 @@
  * @typedef {Object} OnPersistentOptions
  * @property {string} subject What to watch — the same grammar `onLocal()`
  *   takes, e.g. `fs:~/Documents` or `fs:~/inbox/*.json:add`.
- * @property {'broadcast' | 'single'} [delivery] `broadcast` (the default)
- *   delivers to everything listening; `single` delivers to exactly one
- *   consumer, which must acknowledge, and requires a `handlerName`.
- * @property {Array<'socket' | 'worker' | 'push'>} [targets] Transports the
- *   deliveries may take. Defaults to `['socket', 'worker']` when an app
- *   subscribes and `['socket']` when an account session with no app does. A
- *   `single` subscription may not target `push`.
+ * @property {boolean} [queue] Work-queue delivery: each event goes to one
+ *   consumer at a time and is offered again until acknowledged. Requires a
+ *   `handlerName`. Without it, every event runs the app's published handler
+ *   in its events worker and also reaches every connected client, once each.
+ * @property {'broadcast' | 'single'} [delivery] Deprecated: use `queue`.
+ *   `'single'` is the same as `queue: true`.
+ * @property {unknown[]} [targets] Deprecated and ignored.
  * @property {string} [handlerName] The published handler this subscription
- *   binds to. Required for `single`.
+ *   binds to. Required for `queue`.
  * @property {Function | string | { file: string }} [handler] The handler
  *   source this subscription was written against. Sent as a hash, not as
  *   source: the subscription binds only if it matches what is published under
  *   `handlerName`, which is also required when this is given. A function also
- *   runs in this client, on the deliveries the server leaves to clients.
+ *   runs in this client for each delivery it receives.
  * @property {Record<string, unknown>} [context] Values the handler needs,
  *   evaluated **now** and delivered as a frozen `ctx` on every invocation.
  *   Capped at 4 KB serialized.
@@ -248,9 +248,8 @@
  *   covers everything under it.
  * @property {string | null} op The single operation it is limited to, or
  *   `null` for all of them.
- * @property {Array<'socket' | 'worker' | 'push'>} targets Transports its
- *   deliveries may take.
- * @property {'broadcast' | 'single'} delivery Its delivery class.
+ * @property {boolean} queue Whether each event goes to one consumer at a
+ *   time and is offered again until acknowledged.
  * @property {boolean} includeValue Whether its `kv:` deliveries carry the
  *   key's new value.
  * @property {string | null} handlerName The handler it is bound to.

src/puter-js/tests/api/suites/events.suite.ts

@@ -101,7 +101,7 @@ const recorded = (): Array<{ path?: string; label?: unknown }> =>
 
 /**
  * A broadcast handler, reporting whether it was handed the same environment a
- * `single` handler gets — `user` and `fetch` are not conditioned on an `ack`
+ * `queue` handler gets — `user` and `fetch` are not conditioned on an `ack`
  * existing, so every consumer of a broadcast delivery gets them too.
  */
 const BROADCAST_ENV_HANDLER = async ({
@@ -709,7 +709,7 @@ export default suite('events', {
         try {
             t.assert.ok(sub.subId, 'the subscription carries a server id');
             t.assert.equal(sub.subject, `fs:${dir}`);
-            t.assert.equal(sub.delivery, 'broadcast');
+            t.assert.equal(sub.queue, false);
             t.assert.equal(sub.suspendedAt, null);
 
             const held = await t.puter.events.list();
@@ -786,7 +786,7 @@ export default suite('events', {
 
             const sub = await t.puter.events.onPersistent({
                 subject: `fs:${dir}`,
-                delivery: 'single',
+                queue: true,
                 handlerName: 'ingestUpload',
                 handler: RECORDING_HANDLER,
                 context: { label: 'ingest' },
@@ -806,7 +806,7 @@ export default suite('events', {
                     'the handler is handed the context the subscription carries',
                 );
 
-                // A `single` hands out one delivery at a time, so a second one
+                // A queue hands out one delivery at a time, so a second one
                 // arriving is the acknowledgement of the first.
                 await t.puter.fs.write(`${dir}/second.txt`, 'two');
                 const settled = await waitFor(
@@ -843,7 +843,6 @@ export default suite('events', {
 
             const sub = await t.puter.events.onPersistent({
                 subject: `fs:${dir}`,
-                targets: ['socket'],
                 handlerName: 'ingestEvict',
                 handler: RECORDING_HANDLER,
                 context: { label: 'ingest' },
@@ -886,15 +885,14 @@ export default suite('events', {
             const live = await open(t, `fs:${dir}`, () => {});
             if (! live) return;
 
-            // Default delivery: `broadcast`, not `single` — nobody owes an
-            // `ack`, but the environment is the same either way. Pinned to
-            // `socket` so this client runs it whether or not a worker could.
+            // Default delivery, not `queue` — nobody owes an
+            // `ack`, but the environment is the same either way. This client
+            // runs it alongside the events worker.
             const sub = await t.puter.events.onPersistent({
                 subject: `fs:${dir}`,
                 handlerName: 'ingestBroadcastEnv',
                 handler: BROADCAST_ENV_HANDLER,
                 context: { label: 'broadcast-env' },
-                targets: ['socket'],
             });
             try {
                 await t.puter.fs.write(`${dir}/first.txt`, 'one');
@@ -925,7 +923,7 @@ export default suite('events', {
             t.assert.rejects(() =>
                 t.puter.events.onPersistent({
                     subject: `fs:${dir}`,
-                    delivery: 'single',
+                    queue: true,
                     handlerName: 'ingestUpload',
                     handler: RECORDING_HANDLER,
                 }),
@@ -937,7 +935,7 @@ export default suite('events', {
         const subId = await asApp(t, appUid, async () => {
             const sub = await t.puter.events.onPersistent({
                 subject: `fs:${dir}`,
-                delivery: 'single',
+                queue: true,
                 handlerName: 'ingestUpload',
                 handler: RECORDING_HANDLER,
             });
@@ -968,12 +966,11 @@ export default suite('events', {
             const sub = await open(t, `fs:${dir}`, () => {});
             if (sub) await sub.off();
 
-            // So is a durable one that asks for nothing else.
+            // So is a durable one with no handler to run.
             const durable = await t.puter.events.onPersistent({
                 subject: `fs:${dir}`,
-                targets: ['socket'],
             });
-            t.assert.deepEqual(durable.targets, ['socket']);
+            t.assert.equal(durable.handlerName, null);
             await durable.off!();
         });
     },