diff --git a/CHANGELOG.md b/CHANGELOG.md index 4811236e..189b005c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -116,6 +116,22 @@ pending until the returned function is called, so a test can model a native that is slow to answer. Later enqueues wait behind it, as they do on a device. +A key that `define()` registers after `configure()` now receives its held +outcomes in the same session. Before, an outcome that replayed before its +key had a definition emitted the `unhandled-key` row and then waited for the +next launch. This happened when a bundler loaded a module lazily, such as +Metro with `inlineRequires`. The outcomes of one id and key still deliver +in order; across keys, the key defined first delivers first. The +`unhandled-key` row is not followed by a new `state` row when the outcome is +handled. `cancel(id)` drops the held outcomes of that id. An outcome whose +key is never defined stays unacknowledged, as before. + +`configure()` takes an optional `definitions` list, the handles that +`define()` returned. Listing a handle imports its module, so its `define()` +runs before the replay. In development, a later `define()` of a key that is +not listed logs a warning. Nothing else changes: an unlisted key still +receives its outcomes. The new `AnyDefined` type names one entry of the list. + ## 11.1.0 Added `takeDiagnostics()`. On Android it returns the module's recent diff --git a/README.md b/README.md index 6e84bf0d..347e3ef1 100644 --- a/README.md +++ b/README.md @@ -293,13 +293,25 @@ while the app was dead from being sent twice. outcomes wait too. They stay journaled and unacknowledged, so none is lost. 5. **Replay starts after `configure()`.** Outcomes journaled by a dead session - deliver then. Call every `define()` first. -6. **Unknown key is loud.** An outcome whose key has no definition stays - unacknowledged and reaches `state` listeners with `reason: 'unhandled-key'`. + deliver then. Call every `define()` first. A bundler that loads modules + lazily, such as Metro with `inlineRequires`, can run a `define()` later. + That key then receives its held outcomes when its `define()` runs, in the + same session and in their original order per id. Order holds among the + outcomes of one key. Across keys, the key defined first delivers first. + `cancel(id)` drops the held outcomes of that id. +6. **Unknown key is loud.** An outcome whose key has no definition reaches + `state` listeners with `reason: 'unhandled-key'`, and the library holds it. + When no `define()` of the key comes in this session, it stays + unacknowledged and redelivers at the next launch. When a later `define()` + handles it, no new `state` row replaces the `unhandled-key` row. After the + ack, a `completed` or `cancelled` entry leaves `getRequests()` without a + `state` event, and an `error` entry stays with its `error` row. 7. **Completed entries are forgotten after ack.** Row and bytes go. An `error` or `expired` entry keeps both until `cancel()` or a same-id `mutate()`. 8. **Ordering holds per id only.** The outcomes of one id are delivered in - order. There is no ordering guarantee between different ids. + order. One exception: when one id has held outcomes under two keys, the + key defined first delivers first (rule 5). There is no ordering guarantee + between different ids. 9. **`attempt` events are live-only.** They are never journaled or replayed. Retry classes: @@ -401,10 +413,13 @@ whose `request` takes no vars, call `mutate()` with no arguments; native stores ### `configure(options): void` Call one time at boot, after every `define()`. Starts replay of journaled -outcomes. A second call updates the settings and does not replay again. +outcomes. A second call updates the settings and does not replay again. A +`define()` that runs after `configure()` still receives the outcomes that +reached its key before it; see rule 5 of "Reliable delivery". | Option | Notes | | --- | --- | +| `definitions` | Optional. Every definition, as `define()` returned it. Listing a handle imports its module, so its `define()` runs before replay starts. In development, a later `define()` of a key that is not listed logs a warning. That key still receives its outcomes, and `mutate()` works for it. Throws when it is not an array of handles. | | `lifetimeMs` | Default `expiresAt` distance. Default 14 days. | | `maxVarsBytes` | Cap on the JSON length of `vars`. Default 1 MB. `mutate()` rejects above it. JS-side only. | | `retry` | `{ backoff?: { baseMs, maxMs, jitter }, terminalHttp?: { exempt }, transient?: Array<{ status, bodyIncludes?, maxRetries? }> }`. Each object is optional, but one you give must be complete. Defaults 1 s, 2 h, 0.2, `[404]`, `[]`. A descriptor's `retry` replaces each field it names. | diff --git a/src/__tests__/client.test.ts b/src/__tests__/client.test.ts index 6db7909c..e97dce6e 100644 --- a/src/__tests__/client.test.ts +++ b/src/__tests__/client.test.ts @@ -830,7 +830,7 @@ describe('onWarning', () => { expect.objectContaining({ message: 'handler failed' }), ], [ - `delivery: no definition for key "nobody" (id ${id}). The outcome stays unacknowledged.`, + `delivery: no definition for key "nobody" (id ${id}). The outcome stays unacknowledged until define() registers the key, or until the next launch.`, undefined, ], [ @@ -993,3 +993,325 @@ describe('with a fake native', () => { ]); }); }); + +describe('a definition that arrives after configure()', () => { + // A lazily loaded module can run its define() after configure(). These + // outcomes come from an earlier session, so this one replays them. + const NOT_LISTED = /is not in configure\(\{ definitions \}\)/; + const request = (_v: null) => ({ url: 'https://x' }); + let warn: jest.SpyInstance; + + beforeEach(() => { + warn = jest.spyOn(console, 'warn').mockImplementation(() => {}); + }); + + afterEach(() => { + warn.mockRestore(); + }); + + /** + * Queues an entry of `key` through a client that never configures, as an + * earlier session did. No listener exists, so settle() only journals. + */ + const queuedEarlier = async ( + fake: ReturnType, + key: string, + ): Promise => { + const earlier = createUploadClient({ native: fake }); + const { id } = await earlier.define({ key, request }).mutate(); + return id; + }; + + const notListedWarnings = () => + warn.mock.calls.filter(([message]) => NOT_LISTED.test(String(message))); + + it('delivers a held outcome to the late definition and acks it in the same session', async () => { + const fake = createFakeNative(); + const id = await queuedEarlier(fake, 'late'); + const acked = fake.settle(id, { + kind: 'completed', + response: { body: '{}' }, + }); + const client = createUploadClient({ native: fake }); + const state = jest.fn(); + client.addListener('state', state); + + client.configure({}); + await flush(); + expect(state).toHaveBeenCalledWith( + expect.objectContaining({ id, key: 'late', reason: 'unhandled-key' }), + ); + expect(fake.ackedEventIds).toEqual([]); + + const onSuccess = jest.fn(); + client.define({ key: 'late', request, onSuccess }); + const event = await acked; + + expect(onSuccess).toHaveBeenCalledWith( + { status: 200, body: '{}', bodyTruncated: false }, + null, + expect.objectContaining({ id, key: 'late', deliveries: 1 }), + ); + expect(fake.ackedEventIds).toEqual([event.eventId]); + }); + + it('delivers two held outcomes of one id in journal order', async () => { + const fake = createFakeNative(); + const id = await queuedEarlier(fake, 'late'); + const first = fake.settle(id, { + kind: 'error', + error: { errorKind: 'http', response: { status: 500 } }, + }); + const second = fake.settle(id, { kind: 'completed' }); + const client = createUploadClient({ native: fake }); + client.configure({}); + await flush(); + + const order: string[] = []; + client.define({ + key: 'late', + request, + // The first handler is slow. The second must still wait for it. + onError: async () => { + await new Promise((resolve) => setTimeout(resolve, 20)); + order.push('error'); + }, + onSuccess: () => { + order.push('completed'); + }, + }); + const [a, b] = await Promise.all([first, second]); + + expect(order).toEqual(['error', 'completed']); + expect(fake.ackedEventIds).toEqual([a.eventId, b.eventId]); + }); + + it('leaves an outcome whose key is never defined unacknowledged', async () => { + const fake = createFakeNative({ ackTimeoutMs: 100 }); + const id = await queuedEarlier(fake, 'gone'); + const acked = fake.settle(id, { kind: 'completed' }); + const client = createUploadClient({ native: fake }); + const state = jest.fn(); + client.addListener('state', state); + + client.configure({}); + client.define({ key: 'other', request }); + + await expect(acked).rejects.toThrow(/was not acknowledged/); + expect(state).toHaveBeenCalledWith( + expect.objectContaining({ id, key: 'gone', reason: 'unhandled-key' }), + ); + expect(fake.ackedEventIds).toEqual([]); + expect(await fake.getUnacknowledgedEvents()).toEqual([ + expect.objectContaining({ id, key: 'gone' }), + ]); + }); + + it('does not deliver twice when define() replaces a key with held outcomes', async () => { + const fake = createFakeNative(); + const id = await queuedEarlier(fake, 'late'); + const acked = fake.settle(id, { kind: 'completed' }); + const client = createUploadClient({ native: fake }); + client.configure({}); + await flush(); + + // A hot reload runs the module's define() again. + const replaced = jest.fn(); + const current = jest.fn(); + client.define({ key: 'late', request, onSuccess: replaced }); + client.define({ key: 'late', request, onSuccess: current }); + const event = await acked; + client.define({ key: 'late', request, onSuccess: current }); + await flush(); + + expect(replaced).not.toHaveBeenCalled(); + expect(current).toHaveBeenCalledTimes(1); + expect(fake.ackedEventIds).toEqual([event.eventId]); + }); + + it('with configure({ definitions }) listing every handle, behaves as before', async () => { + const fake = createFakeNative(); + const id = await queuedEarlier(fake, 'item.send'); + const acked = fake.settle(id, { kind: 'completed' }); + const client = createUploadClient({ native: fake }); + const onSuccess = jest.fn(); + const send = client.define({ key: 'item.send', request, onSuccess }); + const ping = client.define({ key: 'ping', request }); + + client.configure({ definitions: [send, ping], lifetimeMs: 1000 }); + await acked; + expect(onSuccess).toHaveBeenCalledTimes(1); + // The list stays on the JS side. + expect(fake.calls.configure).toEqual([{ lifetimeMs: 1000 }]); + + // A hot reload of a listed key replaces it without the list warning. + client.define({ key: 'ping', request }); + expect(notListedWarnings()).toEqual([]); + const { id: pingId } = await ping.mutate(); + await fake.settle(pingId, { kind: 'completed' }); + }); + + it('rejects a definitions option that is not a list of handles', () => { + const client = createUploadClient({ native: createFakeNative() }); + expect(() => + client.configure({ definitions: 'all' as unknown as [] }), + ).toThrow( + 'configure: definitions must be an array of the handles that define() returned', + ); + expect(() => + client.configure({ definitions: [null as unknown as never] }), + ).toThrow(/definitions must be an array/); + }); + + it('in development, warns once about a key defined later that the list left out, and still delivers to it', async () => { + const fake = createFakeNative(); + const id = await queuedEarlier(fake, 'late'); + const acked = fake.settle(id, { kind: 'completed' }); + const client = createUploadClient({ native: fake }); + const listed = client.define({ key: 'listed', request }); + client.configure({ definitions: [listed] }); + await flush(); + + const onSuccess = jest.fn(); + client.define({ key: 'late', request, onSuccess }); + client.define({ key: 'late', request, onSuccess }); + await acked; + + expect(notListedWarnings()).toEqual([ + [ + 'define: "late" is not in configure({ definitions }). It still receives its outcomes, but list it so that its module loads before configure().', + ], + ]); + expect(onSuccess).toHaveBeenCalledTimes(1); + }); + + it('drops the held outcomes of a cancelled id, so a later define() runs no handler', async () => { + const fake = createFakeNative({ ackTimeoutMs: 100 }); + const id = await queuedEarlier(fake, 'late'); + const acked = fake.settle(id, { kind: 'completed' }); + const client = createUploadClient({ native: fake }); + client.configure({}); + await flush(); + + // On a settled entry, native forgets the row and its outcomes. + await client.cancel(id); + expect(await fake.getUnacknowledgedEvents()).toEqual([]); + const onSuccess = jest.fn(); + client.define({ key: 'late', request, onSuccess }); + + await expect(acked).rejects.toThrow(/was not acknowledged/); + expect(onSuccess).not.toHaveBeenCalled(); + expect(fake.ackedEventIds).toEqual([]); + }); + + it('drops the held outcomes of an id at the start of cancel(), so a define() while native cancels runs no handler', async () => { + const fake = createFakeNative({ ackTimeoutMs: 100 }); + const id = await queuedEarlier(fake, 'late'); + const acked = fake.settle(id, { kind: 'completed' }); + // Native's answer to cancel() waits until the test lets it go. + let answer!: () => void; + const answered = new Promise((resolve) => { + answer = resolve; + }); + const nativeCancel = fake.cancel; + jest.spyOn(fake, 'cancel').mockImplementation(async (cancelId) => { + await answered; + return nativeCancel(cancelId); + }); + const client = createUploadClient({ native: fake }); + client.configure({}); + await flush(); + + const cancelling = client.cancel(id); + const onSuccess = jest.fn(); + client.define({ key: 'late', request, onSuccess }); + await flush(); + answer(); + await cancelling; + + await expect(acked).rejects.toThrow(/was not acknowledged/); + expect(onSuccess).not.toHaveBeenCalled(); + expect(fake.ackedEventIds).toEqual([]); + }); + + it('sends no second state row when a late definition handles a held outcome', async () => { + const fake = createFakeNative(); + const id = await queuedEarlier(fake, 'late'); + const acked = fake.settle(id, { kind: 'completed' }); + const client = createUploadClient({ native: fake }); + const state = jest.fn(); + client.addListener('state', state); + client.configure({}); + await flush(); + + const onSuccess = jest.fn(); + client.define({ key: 'late', request, onSuccess }); + await acked; + await flush(); + + expect(onSuccess).toHaveBeenCalledTimes(1); + expect(state.mock.calls).toEqual([ + [expect.objectContaining({ id, key: 'late', reason: 'unhandled-key' })], + ]); + }); + + it('a later configure() replaces the list the dev warning checks, and no list turns it off', () => { + const client = createUploadClient({ native: createFakeNative() }); + const listed = client.define({ key: 'listed', request }); + client.configure({ definitions: [listed] }); + client.configure({ + definitions: [listed, { key: 'added', mutate: listed.mutate }], + }); + client.define({ key: 'added', request }); + client.define({ key: 'unlisted', request }); + expect(notListedWarnings()).toEqual([ + [expect.stringContaining('define: "unlisted" is not in')], + ]); + + client.configure({}); + client.define({ key: 'after.no.list', request }); + expect(notListedWarnings()).toHaveLength(1); + }); + + it('does not warn about an unlisted key without the list, or outside development', async () => { + const withoutList = createUploadClient({ native: createFakeNative() }); + withoutList.configure({}); + withoutList.define({ key: 'late', request }); + + const global = globalThis as { __DEV__?: boolean }; + global.__DEV__ = false; + try { + const production = createUploadClient({ native: createFakeNative() }); + production.configure({ definitions: [] }); + production.define({ key: 'late', request }); + } finally { + delete global.__DEV__; + } + + expect(notListedWarnings()).toEqual([]); + }); +}); + +describe('late definitions with onWarning', () => { + const request = (_v: null) => ({ url: 'https://x' }); + let consoleWarn: jest.SpyInstance; + beforeEach(() => { + consoleWarn = jest.spyOn(console, 'warn').mockImplementation(() => {}); + }); + afterEach(() => consoleWarn.mockRestore()); + + it('sends the warning for an unlisted late define() to onWarning', async () => { + const fake = createFakeNative(); + const client = createUploadClient({ native: fake }); + const onWarning = jest.fn(); + const listed = client.define({ key: 'listed', request }); + client.configure({ definitions: [listed], onWarning }); + await flush(); + client.define({ key: 'late', request }); + expect(onWarning).toHaveBeenCalledWith( + 'define: "late" is not in configure({ definitions }). It still receives its outcomes, but list it so that its module loads before configure().', + undefined, + ); + expect(consoleWarn).not.toHaveBeenCalled(); + }); +}); diff --git a/src/__tests__/delivery.test.ts b/src/__tests__/delivery.test.ts index ef82730f..07eb21d6 100644 --- a/src/__tests__/delivery.test.ts +++ b/src/__tests__/delivery.test.ts @@ -417,6 +417,209 @@ describe('unknown key', () => { }); }); +describe('release', () => { + it('routes a released outcome before a later outcome of its id that waited on mutate()', async () => { + const definitions: Record = {}; + const { start, emit, trackMutate, release, native } = setup(definitions); + start(); + await flush(); + emit(completed({ eventId: 'first', key: 'late', vars: { n: 1 } })); + await flush(); + // The second outcome of the id waits on a same-id mutate() in its lane. + const enqueue = deferred(); + trackMutate('u1', enqueue.promise); + emit(completed({ eventId: 'second', key: 'late', vars: { n: 2 } })); + await flush(); + + const order: number[] = []; + definitions.late = { + key: 'late', + request: jest.fn(), + onSuccess: (_data: unknown, vars: { n: number }) => { + order.push(vars.n); + }, + }; + release('late'); + enqueue.resolve('u1'); + await tick(); + await flush(); + + expect(order).toEqual([1, 2]); + expect(native.ackEvents.mock.calls).toEqual([[['first']], [['second']]]); + }); + + it('routes a held outcome before a newer outcome of its id and key when define() runs a few microtasks in', async () => { + // The newer outcome waits in the lane while the older one is held. A + // define() can run in any microtask of that wait. Try each depth. + for (let depth = 0; depth < 12; depth++) { + const definitions: Record = {}; + const { start, emit, release, native } = setup(definitions); + start(); + await flush(); + emit(completed({ eventId: 'old', key: 'late', vars: { n: 1 } })); + await flush(); + emit(completed({ eventId: 'new', key: 'late', vars: { n: 2 } })); + + const order: number[] = []; + let continuation = Promise.resolve(); + for (let i = 0; i < depth; i++) { + continuation = continuation.then(() => undefined); + } + void continuation.then(() => { + definitions.late = { + key: 'late', + request: jest.fn(), + onSuccess: (_data: unknown, vars: { n: number }) => { + order.push(vars.n); + }, + }; + release('late'); + }); + await flush(); + + expect({ depth, order }).toEqual({ depth, order: [1, 2] }); + expect(native.ackEvents.mock.calls).toEqual([[['old']], [['new']]]); + } + }); + + it('waits for a pending same-id mutate() before it routes a released outcome on an idle lane', async () => { + const definitions: Record = {}; + const { start, emit, trackMutate, release, native } = setup(definitions); + start(); + await flush(); + emit(completed({ key: 'late' })); + await flush(); + + const onSuccess = jest.fn(); + definitions.late = { key: 'late', request: jest.fn(), onSuccess }; + const enqueue = deferred(); + trackMutate('u1', enqueue.promise); + release('late'); + await flush(); + expect(onSuccess).not.toHaveBeenCalled(); + + enqueue.resolve('u1'); + await tick(); + await flush(); + expect(onSuccess).toHaveBeenCalledTimes(1); + expect(native.ackEvents.mock.calls).toEqual([[['e1']]]); + }); + + it('acks a cancelled outcome of an undefined key once, and does not route it again on release', async () => { + const definitions: Record = {}; + const { start, emit, release, native } = setup(definitions); + start(); + await flush(); + emit( + completed({ + eventId: 'cancelled', + key: 'late', + kind: 'cancelled', + state: 'cancelled', + response: undefined, + cancelReason: 'user', + }), + ); + await flush(); + expect(native.ackEvents.mock.calls).toEqual([[['cancelled']]]); + + definitions.late = { key: 'late', request: jest.fn() }; + release('late'); + await flush(); + expect(native.ackEvents.mock.calls).toEqual([[['cancelled']]]); + }); + + it('drops the held outcomes of a forgotten id and keeps the others', async () => { + const definitions: Record = {}; + const { start, emit, release, forget, native } = setup(definitions); + start(); + await flush(); + emit(completed({ eventId: 'gone', id: 'u1', key: 'late' })); + emit(completed({ eventId: 'kept', id: 'u2', key: 'late' })); + await flush(); + + const onSuccess = jest.fn(); + definitions.late = { key: 'late', request: jest.fn(), onSuccess }; + forget('u1'); + release('late'); + await flush(); + + expect(onSuccess).toHaveBeenCalledTimes(1); + expect(onSuccess.mock.calls[0]?.[2]).toMatchObject({ id: 'u2' }); + expect(native.ackEvents.mock.calls).toEqual([[['kept']]]); + }); + + it('drops a released outcome of a forgotten id that still waits for its turn', async () => { + const definitions: Record = {}; + const { start, emit, trackMutate, release, forget, native } = + setup(definitions); + start(); + await flush(); + emit(completed({ key: 'late' })); + await flush(); + + const onSuccess = jest.fn(); + definitions.late = { key: 'late', request: jest.fn(), onSuccess }; + const enqueue = deferred(); + trackMutate('u1', enqueue.promise); + release('late'); + forget('u1'); + enqueue.resolve('u1'); + await tick(); + await flush(); + + expect(onSuccess).not.toHaveBeenCalled(); + expect(native.ackEvents).not.toHaveBeenCalled(); + }); + + it('delivers the held outcomes of one id under two keys in define() order', async () => { + // A same-id mutate() from another definition gives one id outcomes + // under two keys. Order holds per key; the key defined first goes first. + const definitions: Record = {}; + const { start, emit, release, native } = setup(definitions); + start(); + await flush(); + emit(completed({ eventId: 'older', key: 'a' })); + emit(completed({ eventId: 'newer', key: 'b' })); + await flush(); + + const order: string[] = []; + const define = (key: string): void => { + definitions[key] = { + key, + request: jest.fn(), + onSuccess: () => { + order.push(key); + }, + }; + release(key); + }; + define('b'); + await flush(); + define('a'); + await flush(); + + expect(order).toEqual(['b', 'a']); + expect(native.ackEvents.mock.calls).toEqual([[['newer']], [['older']]]); + }); + + it('does nothing for a key with no held outcomes', async () => { + const onSuccess = jest.fn(); + const { start, emit, release, native } = setup({ + k: { key: 'k', request: jest.fn(), onSuccess }, + }); + start(); + await flush(); + emit(completed()); + await flush(); + release('k'); + release('never-seen'); + await flush(); + expect(onSuccess).toHaveBeenCalledTimes(1); + expect(native.ackEvents).toHaveBeenCalledTimes(1); + }); +}); + describe('completed', () => { it('passes the RawResponse to onSuccess when there is no parser, with vars and meta', async () => { const onSuccess = jest.fn(); diff --git a/src/__typetests__/define.ts b/src/__typetests__/define.ts index b85a3831..0f2351e4 100644 --- a/src/__typetests__/define.ts +++ b/src/__typetests__/define.ts @@ -306,6 +306,15 @@ client.define({ request: (_vars: null) => ({ url: 'https://x', wifiOnly: 'yes' }), }); +// configure() takes handles of any vars and response types, and the option +// is optional. +client.configure({}); +client.configure({ + definitions: [addComment, noVars, interfaceVars, readonlyVars, upsert], +}); +// @ts-expect-error a key alone is not a handle +client.configure({ definitions: [{ key: 'bare' }] }); + // onWarning takes a message and an optional detail. client.configure({ onWarning: (_message, _detail) => { diff --git a/src/delivery.ts b/src/delivery.ts index 685145d1..5a09fe03 100644 --- a/src/delivery.ts +++ b/src/delivery.ts @@ -58,6 +58,16 @@ export type Delivery = { start: () => void; /** Delivery for `id` waits until this promise has settled. */ trackMutate: (id: string, pending: Promise) => void; + /** + * Call after `define()` registers `key`. The outcomes that reached `key` + * before it had a definition are routed again, in their original order. + */ + release: (key: string) => void; + /** + * Call before and after native `cancel(id)`. Drops the held outcomes of + * `id`, so a later `define()` does not run a handler for them. + */ + forget: (id: string) => void; }; const errorMessage = (e: unknown): string => @@ -70,7 +80,8 @@ const errorMessage = (e: unknown): string => * the id's in-flight mutate(), ack a cancelled outcome, look up the key, run * the handler, ack after its promise resolves. An unknown key or a rejected * handler leaves the outcome unacknowledged, so native redelivers it at the - * next launch. + * next launch. An outcome with an unknown key is also held, and `release()` + * routes it again when `define()` registers the key in the same session. */ export const createDelivery = ({ native, @@ -89,6 +100,12 @@ export const createDelivery = ({ // outcomes deliver first. const buffer: SettledEvent[] = []; let live = false; + // Outcomes whose key had no definition when they were routed, by key, then + // by eventId. A Map keeps insertion order, which is each id's lane order. + const held = new Map>(); + // Held outcomes whose key is now defined, by id, in their original order. + // Each waits for its turn in its id's lane. + const released = new Map(); const trackMutate = (id: string, pending: Promise): void => { const settled = pending.then( @@ -241,9 +258,14 @@ export const createDelivery = ({ await new Promise((resolve) => setTimeout(resolve, 0)); }; - /** The outcome's own steps, run inside its id's lane. */ - const route = async (event: SettledEvent): Promise => { - await waitForMutate(event.id); + const hold = (event: SettledEvent): void => { + const forKey = held.get(event.key) ?? new Map(); + forKey.set(event.eventId, event); + held.set(event.key, forKey); + }; + + /** Settles one outcome after its id's mutate() wait. Runs in its lane. */ + const settle = async (event: SettledEvent): Promise => { // A cancelled outcome has no handler, so it acks whether or not the key // is still defined. if (event.kind === 'cancelled') { @@ -253,8 +275,9 @@ export const createDelivery = ({ const definition = lookup(event.key); if (!definition) { warn( - `delivery: no definition for key "${event.key}" (id ${event.id}). The outcome stays unacknowledged.`, + `delivery: no definition for key "${event.key}" (id ${event.id}). The outcome stays unacknowledged until define() registers the key, or until the next launch.`, ); + hold(event); emitState(unhandledRow(event)); return; } @@ -263,6 +286,46 @@ export const createDelivery = ({ } }; + const takeReleased = (id: string): SettledEvent | undefined => { + const queue = released.get(id); + const next = queue?.shift(); + if (queue && queue.length === 0) { + released.delete(id); + } + return next; + }; + + /** The task that release() adds to an idle lane. */ + const drainReleased = async (id: string): Promise => { + for ( + let event = takeReleased(id); + event !== undefined; + event = takeReleased(id) + ) { + await waitForMutate(id); + await settle(event); + } + }; + + /** The outcome's own steps, run inside its id's lane. */ + const route = async (event: SettledEvent): Promise => { + await waitForMutate(event.id); + // define() can release an earlier outcome of this id while this one + // waited. That outcome goes first, so the id's order holds. The loop is + // inline, not drainReleased(): the last empty check and settle()'s + // lookup must run in one synchronous step, with no await between them + // where a define() could slip in. + for ( + let next = takeReleased(event.id); + next !== undefined; + next = takeReleased(event.id) + ) { + await waitForMutate(event.id); + await settle(next); + } + await settle(event); + }; + const isWellFormed = (event: SettledEvent): boolean => typeof event.key === 'string' && typeof event.kind === 'string' && @@ -318,5 +381,38 @@ export const createDelivery = ({ }); }; - return { start, trackMutate }; + const release = (key: string): void => { + const forKey = held.get(key); + if (!forKey) { + return; + } + held.delete(key); + const ids = new Set(); + forKey.forEach((event) => { + const queue = released.get(event.id) ?? []; + queue.push(event); + released.set(event.id, queue); + ids.add(event.id); + }); + // When the lane is idle this task routes them. When an outcome of the id + // already waits in the lane, that one routes them first and this task + // finds nothing left. + ids.forEach((id) => enqueueForId(id, () => drainReleased(id))); + }; + + const forget = (id: string): void => { + released.delete(id); + held.forEach((forKey, key) => { + forKey.forEach((event, eventId) => { + if (event.id === id) { + forKey.delete(eventId); + } + }); + if (forKey.size === 0) { + held.delete(key); + } + }); + }; + + return { start, trackMutate, release, forget }; }; diff --git a/src/index.ts b/src/index.ts index 9abc1e55..89c45b58 100644 --- a/src/index.ts +++ b/src/index.ts @@ -12,6 +12,7 @@ import { createRegistry, DEFAULT_ENQUEUE_TIMEOUT_MS, DEFAULT_LIFETIME_MS, + isDev, MAX_VARS_BYTES, type AnyDefinition, type Settings, @@ -133,11 +134,32 @@ export const createUploadClient = ({ emitState, warn, }); + // The keys of the last configure({ definitions }), and the unlisted keys + // already warned about. + let listedKeys: Set | undefined; + const warnedKeys = new Set(); + + const onDefine = (key: string): void => { + if ( + listedKeys !== undefined && + !listedKeys.has(key) && + !warnedKeys.has(key) && + isDev() + ) { + warnedKeys.add(key); + warn( + `define: "${key}" is not in configure({ definitions }). It still receives its outcomes, but list it so that its module loads before configure().`, + ); + } + delivery.release(key); + }; + const { define } = createRegistry({ native, definitions, getSettings: () => settings, trackMutate: delivery.trackMutate, + onDefine, warn, }); @@ -148,6 +170,10 @@ export const createUploadClient = ({ * lifetime, retry and Android notification settings to native, then starts * replaying journaled outcomes. A second call updates the settings and does * not replay again. Each call replaces the full configuration. + * + * `definitions` is optional. Each listed handle came from define(), so its + * key is registered already. The list only pins the keys, so that a later + * define() of another key warns in development. */ const configure = (options: ConfigureOptions): void => { // Every check runs before any setting changes, so a rejected call leaves @@ -155,6 +181,16 @@ export const createUploadClient = ({ if (options.retry !== undefined) { validateRetryPolicy(options.retry, 'configure'); } + const listed = options.definitions; + if ( + listed !== undefined && + (!Array.isArray(listed) || + !listed.every((handle) => typeof handle?.key === 'string')) + ) { + throw new Error( + 'configure: definitions must be an array of the handles that define() returned', + ); + } const lifetimeMs = options.lifetimeMs ?? DEFAULT_LIFETIME_MS; if (!Number.isFinite(lifetimeMs) || lifetimeMs <= 0) { throw new Error( @@ -195,6 +231,10 @@ export const createUploadClient = ({ if (options.retry !== undefined) { forwarded.retry = options.retry; } + listedKeys = + listed === undefined + ? undefined + : new Set(listed.map((handle) => handle.key)); native.configure(forwarded); delivery.start(); }; @@ -222,7 +262,20 @@ export const createUploadClient = ({ * when it has no entry file: a v9 upload's bytes, or a chunked file kept * after an enqueue that rejected E_STORAGE. */ - const cancel = (id: string): Promise => native.cancel(id); + const cancel = async (id: string): Promise => { + // On a settled entry native also drops the unacknowledged outcomes, so + // the held ones must not reach a handler later. On a live entry native + // keeps an earlier generation's outcomes; dropping them here only defers + // them to the next launch, as before late definitions existed. + // Forget before the await, so a define() of the key that runs while + // native cancels does not run a handler for this id. When native cancel + // rejects, the dropped outcomes were never acknowledged, so native + // redelivers them at the next launch. + delivery.forget(id); + await native.cancel(id); + // Forget again: an outcome of the id can be held during the await. + delivery.forget(id); + }; /** * Persisted natively. Applies to queued and future entries whose diff --git a/src/registry.ts b/src/registry.ts index a2bc0c4e..9145c869 100644 --- a/src/registry.ts +++ b/src/registry.ts @@ -116,6 +116,8 @@ type RegistryDeps = { getSettings: () => Settings; /** Called with every enqueue promise, so delivery can wait for it. */ trackMutate: (id: string, pending: Promise) => void; + /** Called after `define()` has registered `key`. */ + onDefine?: (key: string) => void; warn?: (message: string) => void; now?: () => number; }; @@ -575,6 +577,7 @@ export const createRegistry = ({ definitions, getSettings, trackMutate, + onDefine, warn = console.warn, now = Date.now, }: RegistryDeps): Registry => { @@ -598,6 +601,7 @@ export const createRegistry = ({ ); } definitions.set(key, definition); + onDefine?.(key); const mutate = async ( input: V | undefined, diff --git a/src/types.ts b/src/types.ts index 9fcc2035..c4c1f71b 100644 --- a/src/types.ts +++ b/src/types.ts @@ -318,6 +318,12 @@ export type Defined = { : (vars: V, options?: { id?: string }) => Promise<{ id: string }>; }; +/** A `Defined` of any vars and response types, as `configure()` lists them. */ +export type AnyDefined = { + key: string; + mutate: (...args: never[]) => Promise<{ id: string }>; +}; + /** * `define()`. `V` infers from the `request` parameter and defaults to `null` * when `request` declares none. `T` infers from the `response` return type. @@ -349,6 +355,13 @@ export type AndroidNotificationConfig = { }; export type ConfigureOptions = { + /** + * Optional. Every definition, as `define()` returned it. Listing a handle + * imports its module, so its `define()` runs before the replay starts. In + * development, a later `define()` of a key that is not listed warns. An + * unlisted key still receives its outcomes. + */ + definitions?: readonly AnyDefined[]; /** Default 14 days. Sets the default `expiresAt` of every entry. */ lifetimeMs?: number; /**