diff --git a/.project/concepts/app-architecture/_debt.md b/.project/concepts/app-architecture/_debt.md new file mode 100644 index 0000000..b913ec1 --- /dev/null +++ b/.project/concepts/app-architecture/_debt.md @@ -0,0 +1,7 @@ +# Doc-debt — app-architecture + +> Presence of a block = doc to update. Processed → delete the block; no blocks left → delete this file. +> One block = one "big change": `why` + `files` + `verify` (leaves to review). + +## Raw markers (consolidate into blocks, then delete) +- TOUCHED src/shared/context/FestipodDataContext.tsx @2026-08-16 (session 0b064e8b-1717-421f-a20e-a4318ad217b1) diff --git a/.project/concepts/data-layer/_debt.md b/.project/concepts/data-layer/_debt.md new file mode 100644 index 0000000..c027904 --- /dev/null +++ b/.project/concepts/data-layer/_debt.md @@ -0,0 +1,15 @@ +# Doc-debt — data-layer + +> Presence of a block = doc to update. Processed → delete the block; no blocks left → delete this file. +> One block = one "big change": `why` + `files` + `verify` (leaves to review). + +## Raw markers (consolidate into blocks, then delete) +- TOUCHED src/shared/utils/documentNuri.ts @2026-08-16 (session 0b064e8b-1717-421f-a20e-a4318ad217b1) +- TOUCHED src/shared/utils/resolveOnce.ts @2026-08-16 (session 0b064e8b-1717-421f-a20e-a4318ad217b1) +- TOUCHED src/shared/utils/serialTask.ts @2026-08-16 (session 0b064e8b-1717-421f-a20e-a4318ad217b1) +- TOUCHED src/shared/utils/resolveOnce.test.ts @2026-08-16 (session 0b064e8b-1717-421f-a20e-a4318ad217b1) +- TOUCHED src/shared/utils/serialTask.test.ts @2026-08-16 (session 0b064e8b-1717-421f-a20e-a4318ad217b1) +- TOUCHED src/shared/utils/storeRegistry.ts @2026-08-16 (session 0b064e8b-1717-421f-a20e-a4318ad217b1) +- TOUCHED src/shared/data/registration.ts @2026-08-16 (session 0b064e8b-1717-421f-a20e-a4318ad217b1) +- TOUCHED src/shared/data/entityWrites.ts @2026-08-16 (session 0b064e8b-1717-421f-a20e-a4318ad217b1) +- TOUCHED src/shared/context/FestipodDataContext.tsx @2026-08-16 (session 0b064e8b-1717-421f-a20e-a4318ad217b1) diff --git a/src/shared/context/FestipodDataContext.tsx b/src/shared/context/FestipodDataContext.tsx index ac4a28e..3c5aecf 100644 --- a/src/shared/context/FestipodDataContext.tsx +++ b/src/shared/context/FestipodDataContext.tsx @@ -37,6 +37,7 @@ import { openDocumentInbox, } from '../utils/storeRegistry'; import { useCurrentPrincipal } from '../utils/currentPrincipal'; +import { createSerialTask } from '../utils/serialTask'; import { useShapeQuery } from '../data/useShapeQuery'; import { adaptEvents, adaptUsers, adaptParticipations } from '../data/shapeAdapters'; // The ORM generator emits BARE shape names (`EventShapeType`, `Event`, …), taken @@ -918,36 +919,77 @@ function useNgData(): FestipodDataContextValue { // failed, or this event arrived after the last one. Never 'not-mine' here. return 'unknown'; }, [ownedCanonicalIds, ruledOutEventIds]); - // Last count written per owned event, so we only persist a genuine change. - const materializedCountRef = useRef>(new Map()); + // Last count WRITTEN per owned event (keyed by canonical id), stamped with the + // cycle that wrote it. The stamp is what makes a STALE cycle harmless — see the + // guard below. + const materializedCountRef = useRef>(new Map()); + // Monotonic cycle number, shared by every materialize cycle of this session. It + // has to outlive the effect: when the owned set changes the effect re-runs, and + // a cycle started by the PREVIOUS run can still be in flight. + const cycleSeqRef = useRef(0); useEffect(() => { if (!ready) return; const owned = ownedEvents; if (owned.length === 0) return; let cancelled = false; - const materialize = async (trigger: string) => { + // ONE inbox address per owned event, resolved ONCE here and shared by the two + // things that must not disagree: the cycle that READS the inbox and the watch + // that SUBSCRIBES to it. Resolving them apart is how the owner ends up watching + // one address while a deposit lands in another — the sign-up is then never seen + // in the session that made it. `openDocumentInbox` is itself single-flight per + // document (utils/storeRegistry), so the app can no longer open a second inbox + // at all; this map is the narrower promise that the reader and the watcher hold + // the very same value. + // + // Resolved on first need and kept — but a FAILED resolution is released, not + // kept: it means the address is UNKNOWN, so the next trigger must really ask + // again instead of inheriting a permanent "no inbox". + let resolvingInboxes: Promise> | null = null; + const inboxesForOwnedEvents = (): Promise> => { + if (resolvingInboxes) return resolvingInboxes; + const attempt = (async () => { + const byEvent = new Map(); + for (const evId of owned) byEvent.set(evId, await openDocumentInbox(evId)); + return byEvent; + })().catch(err => { + if (resolvingInboxes === attempt) resolvingInboxes = null; + throw err; + }); + resolvingInboxes = attempt; + return attempt; + }; + + const runCycle = async (trigger: string) => { if (cancelled) return; + const seq = ++cycleSeqRef.current; try { + const inboxes = await inboxesForOwnedEvents(); console.log( - `${logPrefix} owner participation materialize START (trigger=${trigger}) — ${owned.length} owned ` + - `event(s)`, + `${logPrefix} owner participation materialize START (trigger=${trigger}, cycle=${seq}) — ` + + `${owned.length} owned event(s)`, ); const notifs: FpNotificationData[] = []; - for (const evId of owned) { - // Each event has its OWN inbox, and only its OWNER can open it. This - // call returns the address the owner reads and watches; a depositor - // never sees it (they name the document instead). - const targetInbox = await openDocumentInbox(evId); + // Each event has its OWN inbox, and only its OWNER can open it. The address + // comes from the shared map above — the same one the watch subscribed; a + // depositor never sees it (they name the document instead). + for (const [evId, targetInbox] of inboxes) { + const canonId = canonicalEventId(evId); // BEFORE — the event's readable detail (short id + title) and the // participantCount value as currently READ/exposed (the app-side `events` // state), captured before this cycle's derive+write. Comparing this to the // AFTER log below tells whether the counter is a DATA problem (never // incremented) or a DISPLAY/read problem (incremented but not re-read). - const knownEvent = events.find(e => e.id === evId); + // + // Read through the REF, never the closure: this effect only re-runs on + // [ready, ownedKey], so the `events` it captured is the snapshot from the + // render that wired the watch — which is why this line printed `(unknown)` + // for the whole session and told the last investigation nothing. Matched + // on the canonical id-form, like every other event-id comparison here. + const knownEvent = eventsRef.current.find(e => canonicalEventId(e.id) === canonId); const knownCount = knownEvent?.participantCount; console.log( - `${logPrefix} participation materialize — event=${canonicalEventId(evId)}` + + `${logPrefix} participation materialize — event=${canonId}` + (knownEvent?.title ? ` "${knownEvent.title}"` : '') + ` — participantCount before write (as currently read) = ${knownCount ?? '(unknown)'}`, ); @@ -957,18 +999,35 @@ function useNgData(): FestipodDataContextValue { // registrant's deposit is visible even on a cold session. const active = await materializeAttendance(targetInbox, evId); const nextCount = active.length; // no host baseline (creator not auto-in) - const prevCount = materializedCountRef.current.get(evId); - // Write ONLY when the derived value actually changes (anti-loop). This - // memo does NOT lock in a premature 0: the barrier-gated read above makes - // the first post-connection materialize see the real deposits, so once the - // set becomes non-empty `nextCount !== prevCount` and the correct count is - // written. A transient write failure reverts the memo so the next trigger - // retries. The guard's sole job is to avoid re-writing an UNCHANGED value. - if (prevCount !== nextCount) { - materializedCountRef.current.set(evId, nextCount); + const written = materializedCountRef.current.get(canonId); + // THE VALUE A CYCLE CARRIES IS ONLY AS FRESH AS THE READ IT CAME FROM. + // Cycles of this effect can no longer interleave (they are serialized + // below), but the effect re-runs whenever the owned set changes, and a + // cycle from the previous run can still be in flight — holding a count it + // derived BEFORE the fresher one's. Writing it would put the stale value + // back on the document, which is a count that goes backwards for no + // visible reason. So a cycle may only overwrite what an OLDER cycle wrote. + if (written && written.seq > seq) { console.log( - `${logPrefix} owner participation materialize — event=${canonicalEventId(evId)}: ` + - `participantCount ${prevCount ?? '(none)'} → ${nextCount} (writing own doc)`, + `${logPrefix} owner participation materialize — event=${canonId}: cycle ${seq} is STALE ` + + `(cycle ${written.seq} already wrote ${written.count}) — not writing ${nextCount}`, + ); + } else if (written?.count === nextCount) { + // Write ONLY when the derived value actually changes (anti-loop). This + // memo does NOT lock in a premature 0: the barrier-gated read above makes + // the first post-connection materialize see the real deposits, so once the + // set becomes non-empty the value differs and the correct count is + // written. A transient write failure reverts the memo so the next trigger + // retries. The guard's sole job is to avoid re-writing an UNCHANGED value. + console.log( + `${logPrefix} owner participation materialize — event=${canonId}: ` + + `participantCount unchanged (${nextCount}) — no write`, + ); + } else if (!cancelled) { + materializedCountRef.current.set(canonId, { count: nextCount, seq }); + console.log( + `${logPrefix} owner participation materialize — event=${canonId}: ` + + `participantCount ${written?.count ?? '(none)'} → ${nextCount} (writing own doc, cycle=${seq})`, ); // The write lands on the owned event doc, which `watchShape('public')` // already subscribes → the reactive read re-renders the new count on @@ -976,9 +1035,11 @@ function useNgData(): FestipodDataContextValue { let writeOk = true; await updateEntityField(evId, evId, 'participantCount', int(nextCount)) .catch(err => { - // Revert the memo so a transient write failure retries next trigger. + // Revert the memo so a transient write failure retries next trigger + // — but only if it is still OURS. A fresher cycle's value stands. writeOk = false; - materializedCountRef.current.delete(evId); + const current = materializedCountRef.current.get(canonId); + if (current && current.seq === seq) materializedCountRef.current.delete(canonId); console.error(`${logPrefix} owner participation materialize count WRITE FAILED:`, err); }); if (writeOk) { @@ -988,15 +1049,10 @@ function useNgData(): FestipodDataContextValue { // file) still showing the old N after this fires means the counter data // is fine and it is the read side that lags. console.log( - `${logPrefix} participation materialize — event=${canonicalEventId(evId)}: ` + + `${logPrefix} participation materialize — event=${canonId}: ` + `participantCount AFTER write = ${knownCount ?? '(unknown)'} → ${nextCount}`, ); } - } else { - console.log( - `${logPrefix} owner participation materialize — event=${canonicalEventId(evId)}: ` + - `participantCount unchanged (${nextCount}) — no write`, - ); } // (2) NOTIFICATIONS — surface "new participant" deposits (unchanged T02.c). const evNotifs = await readRegistrationNotifications(targetInbox, evId); @@ -1015,12 +1071,22 @@ function useNgData(): FestipodDataContextValue { } }; + // A CYCLE IS A READ-DERIVE-WRITE, AND TWO OF THEM MUST NOT OVERLAP. The two + // triggers below fire within the same instant on the connection that creates an + // event, and run concurrently they both read the inbox before either writes — + // so the one that finishes last puts its own, older reading back on the + // document. Serialized, a request arriving mid-cycle is served by ONE follow-up + // cycle once the current one has finished (a cycle re-derives everything from + // the inbox, so one follow-up covers however many requests it coalesces). + const materialize = createSerialTask(runCycle); + // (A) RELIABLE-AT-CONNECTION: run one materialization directly on this trigger // ([ready, ownedKey]). This is the spec's core — the owner, at its NEXT // CONNECTION, deterministically processes its owned events' inbox, reading // through the synced-view contract. It does NOT depend on a cross-session // inbox push arriving. - void materialize('connection'); + void materialize('connection').catch(err => + console.error(`${logPrefix} owner participation materialize cycle rejected:`, err)); // (B) SAME-SESSION LIVE: `inbox.watch` fires on the initial state push and on // every later deposit visible to THIS verifier (a local deposit, or a remote one @@ -1028,16 +1094,18 @@ function useNgData(): FestipodDataContextValue { // stays live when a deposit does push. Cross-session convergence does NOT rely on // this (it relies on (A) at the owner's next connection); this only sharpens the // same-session/live case. One watch PER owned event — each event has its OWN - // inbox document — resolved async, so wire them inside an IIFE and stash the - // unsubscribes for cleanup. + // inbox document — and the address watched is the one taken from the SHARED map + // above, so what is watched is exactly what the cycle reads. const unsubscribes: Array<() => void> = []; - (async () => { - for (const evId of owned) { - const targetInbox = await openDocumentInbox(evId); - if (cancelled) return; - unsubscribes.push(inbox.watch(targetInbox, () => void materialize('inbox-push'))); + void (async () => { + const inboxes = await inboxesForOwnedEvents(); + if (cancelled) return; + for (const targetInbox of inboxes.values()) { + unsubscribes.push(inbox.watch(targetInbox, () => + void materialize('inbox-push').catch(err => + console.error(`${logPrefix} owner participation materialize cycle rejected:`, err)))); } - })(); + })().catch(err => console.error(`${logPrefix} owner inbox watch wiring failed:`, err)); return () => { cancelled = true; for (const stop of unsubscribes) stop(); }; // eslint-disable-next-line react-hooks/exhaustive-deps }, [ready, ownedKey]); diff --git a/src/shared/data/entityWrites.ts b/src/shared/data/entityWrites.ts index 9de432a..d3fd126 100644 --- a/src/shared/data/entityWrites.ts +++ b/src/shared/data/entityWrites.ts @@ -102,12 +102,30 @@ export async function updateEntityField( // write shape (same as writeEntity / registration.ts); SDK graph details live in // `@ng-eventually/polyfill`, not here. `docs.sparqlUpdate` validates the anchor at // its own door — `subject` only needs escaping, as it lands in an IRI position. - const del = `DELETE WHERE { <${s}> <${pred}> ?o }`; - await docs.sparqlUpdate(sid, del, graphNuri); - if (obj !== null) { - const ins = `INSERT DATA { <${s}> <${pred}> ${obj} }`; - await docs.sparqlUpdate(sid, ins, graphNuri); + if (obj === null) { + // Clearing the field: there is nothing to put back, so the removal stands alone. + await docs.sparqlUpdate(sid, `DELETE WHERE { <${s}> <${pred}> ?o }`, graphNuri); + return; } + // ONE update, not a DELETE followed by an INSERT. Sent as two, the field is + // ABSENT between them, and a read landing in that window does not see "the old + // value" — it sees NO value, which the read side turns into the field's empty + // reading (0 for `participantCount`). The window is small and the reads are + // pushed, so it shows up as a count that flickers to 0 for no reason anyone can + // reproduce on demand. + // + // `DELETE … INSERT … WHERE` is ONE SPARQL modify operation — the surface takes + // an update string and this is a single one, so nothing here invents a + // transaction the SDK does not offer. The `OPTIONAL` is what makes it work on a + // field that is not there yet: the WHERE still yields one solution (with `?o` + // unbound, so the DELETE template drops out) and the INSERT applies. When the + // field is present, every one of its triples is removed and the new one written + // in the same operation. + const update = ` + DELETE { <${s}> <${pred}> ?o } + INSERT { <${s}> <${pred}> ${obj} } + WHERE { OPTIONAL { <${s}> <${pred}> ?o } }`; + await docs.sparqlUpdate(sid, update, graphNuri); } /** diff --git a/src/shared/data/registration.ts b/src/shared/data/registration.ts index 6872aa1..1d52021 100644 --- a/src/shared/data/registration.ts +++ b/src/shared/data/registration.ts @@ -20,6 +20,7 @@ import { inbox, docs } from '@ng-eventually/polyfill'; import type { Nuri, NuriLike } from '@ng-eventually/polyfill'; import { sessionPromise } from '../utils/ngSession'; import { listMyEntityDocs } from '../utils/storeRegistry'; +import { canonicalDocumentId } from '../utils/documentNuri'; import { escapeLiteral, escapeIri } from './sparqlEscape'; import type { FpNotificationData } from './types'; @@ -105,13 +106,12 @@ function mintDepositUid(): string { * target). * * A NURI with no `:v:` overlay (or a non-`did:ng:o:` id) passes through unchanged. + * + * The canonicalization itself is `canonicalDocumentId` (utils/documentNuri): the + * SAME invariant also keys the app's one-inbox-per-document resolution, and one + * document must not have two canonical forms depending on who is asking. */ -export function canonicalEventId(id: string): string { - // did:ng:o::v: → did:ng:o:. The overlay segment is the - // LAST `:v:`-introduced part; a base id (`did:ng:o:`) has no `:v:`. - const i = id.indexOf(':v:'); - return i === -1 ? id : id.slice(0, i); -} +export const canonicalEventId = canonicalDocumentId; /** * Build the host-facing notification from a registration deposit. The recipient diff --git a/src/shared/utils/documentNuri.ts b/src/shared/utils/documentNuri.ts new file mode 100644 index 0000000..82ee1a1 --- /dev/null +++ b/src/shared/utils/documentNuri.ts @@ -0,0 +1,23 @@ +/** + * The CANONICAL form of a document reference — the one form every comparison and + * every per-document memo in the app keys on. + * + * A document reference is a `did:ng:o:[:v:]`. The SAME document can + * legitimately be named with or without its `:v:` suffix depending on + * which boundary handed it over (a create, a listing, a read subject). Two forms + * of one document must never become two entries anywhere: two owned events, two + * counts — or, the defect this file was extracted for, two INBOXES for one + * document, one of which nobody reads. + * + * A reference with no `:v:` overlay (or one that is not a `did:ng:o:` id at all) + * passes through unchanged. + * + * `canonicalEventId` (data/registration) is this function under the name the event + * call sites use; it is re-exported there, not reimplemented. + */ +export function canonicalDocumentId(id: string): string { + // did:ng:o::v: → did:ng:o:. The overlay segment is the + // LAST `:v:`-introduced part; a base id (`did:ng:o:`) has no `:v:`. + const i = id.indexOf(':v:'); + return i === -1 ? id : id.slice(0, i); +} diff --git a/src/shared/utils/resolveOnce.test.ts b/src/shared/utils/resolveOnce.test.ts new file mode 100644 index 0000000..e605e6c --- /dev/null +++ b/src/shared/utils/resolveOnce.test.ts @@ -0,0 +1,106 @@ +import { expect, test } from 'bun:test'; +import { resolveOncePerKey } from './resolveOnce'; + +/** A resolution the test controls: it settles when the test says so. */ +function controllable() { + let calls = 0; + const gates: Array<{ resolve(v: string): void; reject(e: unknown): void }> = []; + const resolve = (arg: string) => { + calls++; + return new Promise((res, rej) => { + gates.push({ resolve: res, reject: rej }); + }).then(v => `${v}:${arg}`); + }; + return { get calls() { return calls; }, gates, resolve }; +} + +test('simultaneous callers for one key start ONE resolution and share its value', async () => { + const c = controllable(); + const resolveOnce = resolveOncePerKey(k => k, c.resolve); + + const a = resolveOnce('doc'); + const b = resolveOnce('doc'); + const d = resolveOnce('doc'); + expect(c.calls).toBe(1); // the second and third joined the one in flight + + c.gates[0]!.resolve('inbox-1'); + expect(await a).toBe('inbox-1:doc'); + expect(await b).toBe('inbox-1:doc'); + expect(await d).toBe('inbox-1:doc'); + expect(c.calls).toBe(1); +}); + +test('the answer is kept — a later caller never starts a second resolution', async () => { + const c = controllable(); + const resolveOnce = resolveOncePerKey(k => k, c.resolve); + + const first = resolveOnce('doc'); + c.gates[0]!.resolve('inbox-1'); + await first; + + expect(await resolveOnce('doc')).toBe('inbox-1:doc'); + expect(c.calls).toBe(1); +}); + +test('distinct keys resolve independently', async () => { + const c = controllable(); + const resolveOnce = resolveOncePerKey(k => k, c.resolve); + + const a = resolveOnce('doc-a'); + const b = resolveOnce('doc-b'); + expect(c.calls).toBe(2); + c.gates[0]!.resolve('inbox-a'); + c.gates[1]!.resolve('inbox-b'); + expect(await a).toBe('inbox-a:doc-a'); + expect(await b).toBe('inbox-b:doc-b'); +}); + +test('two spellings of one key are ONE resolution (the overlay case)', async () => { + const c = controllable(); + const resolveOnce = resolveOncePerKey( + arg => arg.split(':v:')[0]!, + c.resolve, + ); + + const bare = resolveOnce('did:ng:o:repo'); + const overlaid = resolveOnce('did:ng:o:repo:v:overlay'); + expect(c.calls).toBe(1); + c.gates[0]!.resolve('inbox-1'); + expect(await bare).toBe('inbox-1:did:ng:o:repo'); + expect(await overlaid).toBe('inbox-1:did:ng:o:repo'); +}); + +test('a rejection reaches every waiting caller and is NOT memoized', async () => { + const c = controllable(); + const resolveOnce = resolveOncePerKey(k => k, c.resolve); + + const a = resolveOnce('doc'); + const b = resolveOnce('doc'); + c.gates[0]!.reject(new Error('unknown')); + await expect(a).rejects.toThrow('unknown'); + await expect(b).rejects.toThrow('unknown'); + + // UNKNOWN is not "there is none": the next caller really retries. + const retry = resolveOnce('doc'); + expect(c.calls).toBe(2); + c.gates[1]!.resolve('inbox-1'); + expect(await retry).toBe('inbox-1:doc'); +}); + +test('the hooks report a join and the one resolution', async () => { + const c = controllable(); + const joined: string[] = []; + const resolved: Array<[string, string]> = []; + const resolveOnce = resolveOncePerKey(k => k, c.resolve, { + onJoined: key => joined.push(key), + onResolved: (key, value) => resolved.push([key, value]), + }); + + const a = resolveOnce('doc'); + resolveOnce('doc'); + c.gates[0]!.resolve('inbox-1'); + await a; + + expect(joined).toEqual(['doc']); + expect(resolved).toEqual([['doc', 'inbox-1:doc']]); +}); diff --git a/src/shared/utils/resolveOnce.ts b/src/shared/utils/resolveOnce.ts new file mode 100644 index 0000000..4a4f73b --- /dev/null +++ b/src/shared/utils/resolveOnce.ts @@ -0,0 +1,63 @@ +/** + * ONE resolution per key, whatever the concurrency — a single-flight memo. + * + * WHY THIS EXISTS. Some resolutions are not idempotent from the outside: asking + * twice does not hand back the same thing twice, it CREATES a second thing. A + * document's inbox is exactly that — "open the inbox of this document" answers + * with an address, and two callers racing each other end up with two addresses + * for one document, so the side that watches one never sees what was deposited + * in the other. Nothing about the calling code looks wrong: four independent + * call sites, each perfectly reasonable on its own, all firing within the same + * few hundred milliseconds. + * + * So the guarantee is not "we call it less often" (a cache) but "the application + * can never be the reason a second one exists": while a resolution is in flight, + * every other caller for the same key AWAITS THAT SAME PROMISE instead of + * starting its own, and once it has settled they all read the one value. + * + * A REJECTION IS NOT MEMOIZED. It means UNKNOWN, never "there is none": every + * caller waiting on it sees the failure, and the key is released so a later + * caller genuinely retries rather than inheriting a permanent "no". + * + * The memo lives as long as the returned function does. For a per-session + * resolution (a browser context is one identity for its whole life) that is the + * intended lifetime: keep the function at module scope and the answer is settled + * once for the session. + */ + +export interface ResolveOnceHooks { + /** A caller joined a resolution already in flight — nothing new was started. */ + onJoined?(key: string): void; + /** A resolution completed and became the key's one answer. */ + onResolved?(key: string, value: Value): void; +} + +export function resolveOncePerKey( + keyOf: (arg: Arg) => string, + resolve: (arg: Arg) => Promise, + hooks: ResolveOnceHooks = {}, +): (arg: Arg) => Promise { + const byKey = new Map>(); + + return (arg: Arg): Promise => { + const key = keyOf(arg); + const known = byKey.get(key); + if (known) { + hooks.onJoined?.(key); + return known; + } + const resolving = resolve(arg).then( + value => { + hooks.onResolved?.(key, value); + return value; + }, + err => { + // UNKNOWN, not "none" — release the key so a later caller can retry. + byKey.delete(key); + throw err; + }, + ); + byKey.set(key, resolving); + return resolving; + }; +} diff --git a/src/shared/utils/serialTask.test.ts b/src/shared/utils/serialTask.test.ts new file mode 100644 index 0000000..9960ba5 --- /dev/null +++ b/src/shared/utils/serialTask.test.ts @@ -0,0 +1,110 @@ +import { expect, test } from 'bun:test'; +import { createSerialTask } from './serialTask'; + +/** A run the test releases by hand, recording overlap as it goes. */ +function controllable() { + const started: string[] = []; + const releases: Array<() => void> = []; + const rejects: Array<(err: unknown) => void> = []; + let inFlight = 0; + let maxInFlight = 0; + const run = async (reason: string) => { + started.push(reason); + inFlight++; + maxInFlight = Math.max(maxInFlight, inFlight); + try { + await new Promise((res, rej) => { + releases.push(res); + rejects.push(rej); + }); + } finally { + inFlight--; + } + }; + return { started, releases, rejects, run, get maxInFlight() { return maxInFlight; } }; +} + +const tick = () => new Promise(r => setTimeout(r, 0)); + +test('a request made while a run is in flight does not start a second run', async () => { + const c = controllable(); + const task = createSerialTask(c.run); + + void task('connection').catch(() => {}); + void task('inbox-push').catch(() => {}); + await tick(); + + expect(c.started).toEqual(['connection']); + expect(c.maxInFlight).toBe(1); +}); + +test('the follow-up runs once the first has finished, and only once for N requests', async () => { + const c = controllable(); + const task = createSerialTask(c.run); + + void task('connection').catch(() => {}); + void task('inbox-push').catch(() => {}); + void task('inbox-push').catch(() => {}); + void task('inbox-push').catch(() => {}); + + c.releases[0]!(); // first run completes + await tick(); + expect(c.started).toEqual(['connection', 'inbox-push']); // ONE follow-up + expect(c.maxInFlight).toBe(1); + + c.releases[1]!(); + await tick(); + expect(c.started).toEqual(['connection', 'inbox-push']); +}); + +test('every request coalesced into one follow-up settles when that run does', async () => { + const c = controllable(); + const task = createSerialTask(c.run); + + const first = task('connection'); + const joinA = task('inbox-push'); + const joinB = task('inbox-push'); + + c.releases[0]!(); + await first; + c.releases[1]!(); + await joinA; + await joinB; // same run — both are served + expect(c.started.length).toBe(2); +}); + +test('requests made when idle each get their own run, in order', async () => { + const c = controllable(); + const task = createSerialTask(c.run); + + const a = task('connection'); + c.releases[0]!(); + await a; + const b = task('inbox-push'); + c.releases[1]!(); + await b; + + expect(c.started).toEqual(['connection', 'inbox-push']); + expect(c.maxInFlight).toBe(1); +}); + +test('a failed run rejects its requesters and does not wedge the task', async () => { + const c = controllable(); + const task = createSerialTask(c.run); + + const failing = task('connection'); + const queued = task('inbox-push'); + c.rejects[0]!(new Error('materialize failed')); + await expect(failing).rejects.toThrow('materialize failed'); + + // The follow-up still ran, and a later request is still served. + await tick(); + expect(c.started).toEqual(['connection', 'inbox-push']); + c.releases[1]!(); + await queued; + const later = task('connection'); + c.releases[2]!(); + await later; + expect(c.started).toEqual(['connection', 'inbox-push', 'connection']); + expect(c.maxInFlight).toBe(1); +}); diff --git a/src/shared/utils/serialTask.ts b/src/shared/utils/serialTask.ts new file mode 100644 index 0000000..8d69742 --- /dev/null +++ b/src/shared/utils/serialTask.ts @@ -0,0 +1,75 @@ +/** + * A task that NEVER runs concurrently with itself. + * + * WHY THIS EXISTS. A read-derive-write cycle (read an inbox, derive a value, + * write it) is only correct if nothing else is doing the same thing at the same + * time on the same target: two cycles started a few milliseconds apart both read + * before either writes, and the one that finishes last puts ITS (older) reading + * back on the document. The owner's participation materializer had two + * independent triggers — the connection and the inbox push — and nothing between + * them. + * + * COALESCING. A request made while a run is in flight does not queue behind an + * unbounded chain: at most ONE follow-up run is scheduled, and every request + * made during the current run shares it. That is sound precisely because a run + * re-derives everything from the current state — one follow-up observes whatever + * the N requests were about. What a run must NOT be is incremental (a `+1`); the + * caller keeps that property, this primitive assumes it. + * + * WHAT A CALLER GETS BACK. The promise of the run that will serve its request — + * the running one when it started idle, the coalesced follow-up otherwise. It + * settles with that run's outcome, so a failure is never swallowed here; a + * caller that does not await must attach its own rejection handler. + */ + +interface Settle { + promise: Promise; + resolve(): void; + reject(err: unknown): void; +} + +function settleLater(): Settle { + let resolve!: () => void; + let reject!: (err: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} + +export function createSerialTask( + run: (reason: Reason) => Promise, +): (reason: Reason) => Promise { + let busy = false; + let queued: { reason: Reason; settle: Settle } | null = null; + + const pump = async (reason: Reason, settle: Settle): Promise => { + busy = true; + try { + await run(reason); + settle.resolve(); + } catch (err) { + settle.reject(err); + } finally { + busy = false; + const next = queued; + queued = null; + // A failed run must not wedge the task: the follow-up starts either way. + if (next) void pump(next.reason, next.settle); + } + }; + + return (reason: Reason): Promise => { + if (!busy) { + const settle = settleLater(); + void pump(reason, settle); + return settle.promise; + } + // Already running: one follow-up is enough for every request made meanwhile. + // The FIRST such request names it — the ones that join it are, by definition, + // asking for the same thing. + if (!queued) queued = { reason, settle: settleLater() }; + return queued.settle.promise; + }; +} diff --git a/src/shared/utils/storeRegistry.ts b/src/shared/utils/storeRegistry.ts index 4c13b85..4104429 100644 --- a/src/shared/utils/storeRegistry.ts +++ b/src/shared/utils/storeRegistry.ts @@ -11,6 +11,9 @@ // a fact of the module graph rather than a convention. import './ngSession'; import { storeRegistry as sdkStoreRegistry } from '@ng-eventually/polyfill'; +import type { Nuri, NuriLike } from '@ng-eventually/polyfill'; +import { canonicalDocumentId } from './documentNuri'; +import { resolveOncePerKey } from './resolveOnce'; export type Scope = 'public' | 'protected' | 'private'; @@ -42,12 +45,42 @@ export const { // SDK-shaped scope resolvers — the app asks by scope, the SDK resolves // placement (no store id ever crosses the boundary). resolveScopeGraph, - // A document only HAS an inbox if its owner opened one. The app opens one on - // the documents meant to RECEIVE deposits (its events), and the address this - // returns is what the owner reads and watches. - openDocumentInbox, // Per-entity document creation. The SDK itself files the creator's key on // create, so the app declares NO access policy here: reading is possession, // and the creator holds what it created. createEntityDoc, } = sdkStoreRegistry; + +// --- A document's inbox: opened ONCE, by this session, whatever the concurrency - +// A document only HAS an inbox if its owner opened one. The app opens one on the +// documents meant to RECEIVE deposits (its events), and the address this returns +// is what the owner reads and watches. +// +// ASKING TWICE IS NOT FREE. "Open the inbox of this document" answers with an +// address; two callers racing each other get two, and then the owner watches one +// while a deposit lands in the other — the sign-up is never seen. Nothing looked +// wrong at any single call site: creating an event opens its inbox, the +// materializer opens it to read, the watch opens it to subscribe, and the watch +// callback re-enters the materializer — four calls within a fraction of a second, +// none of them aware of the others. +// +// So the app resolves it exactly ONCE PER DOCUMENT for the whole session, and +// simultaneous callers AWAIT THAT SAME RESOLUTION instead of starting their own +// (`resolveOncePerKey`). This wrapper is the ONLY place the SDK call is made — the +// raw entry is not re-exported, so no call site can bypass it. The key is the +// document's CANONICAL form, so the same document named with and without its +// overlay suffix is one document here too. +// +// A session is one identity for its whole life, so a session-long memo can never +// hand one person another's address. A REJECTION is not memoized: it means the +// answer is UNKNOWN, so the next caller genuinely retries. +export const openDocumentInbox: (doc: NuriLike) => Promise = resolveOncePerKey( + (doc: NuriLike) => canonicalDocumentId(doc), + (doc: NuriLike) => sdkStoreRegistry.openDocumentInbox(doc), + { + onResolved: (doc, address) => + console.log(`[app][inbox] doc=${doc} → inbox=${address} (this session's only one)`), + onJoined: doc => + console.log(`[app][inbox] doc=${doc} — joined the resolution already in flight (no second inbox)`), + }, +);