diff --git a/src/shared/context/refineData.test.tsx b/src/shared/context/refineData.test.tsx new file mode 100644 index 0000000..9fb2fbd --- /dev/null +++ b/src/shared/context/refineData.test.tsx @@ -0,0 +1,356 @@ +/** + * Read-your-writes of a join/leave, driven through the REAL data provider + * (`RefineDataProvider` over `useFestipodData`) against a scripted server. + * + * THE RACE (rights matrix, line `leave.ui-counts`, 1 run in 5). A join or leave + * is posted to the user's outbox and the backend updates the meeting point's + * attendees collection LATER. When the reads that follow one act all land + * before the backend has processed it, the act is never confirmed; if the user + * then makes the opposite act, the read still on screen (from BEFORE the first + * act) already shows what the second act asserts. Then the next read — the + * first act processed, the second not yet — shows the opposite, and the view + * must not follow it. + * + * THE SCRIPT IS ORDERED BY READS, NOT BY CLOCK. The server answers the + * collection from `server.members`, and the test changes it only after counting + * the reads served since an act — so the ordering "act → stale read → backend + * processes → read" is the same on every run. Only the data provider's own + * re-read timers run on the real clock. + * + * What is stubbed: the Refine data provider (Alice's own list: empty, or her + * own meeting point) and `fetch` (catalog, records, collection, actor + * document, outbox). Everything between them — the foreign read set, the + * collection reads, the overlay, the counts — is the production code. + * + * THE OVERLAY MUST ALSO GO. A view that only ever shows the overlay would pass + * every count check above; what proves the act was RELEASED (confirmed by a + * read) is that the re-reads stop: they run only while an act is pending. + */ + +import { afterAll, afterEach, beforeAll, expect, test } from 'bun:test'; +import { Window } from 'happy-dom'; +import { useLayoutEffect, type ReactNode } from 'react'; +import type { DataProvider } from '@refinedev/core'; +import { activitypods } from '../refine/providers/activitypods'; +import { setCurrentSession } from '../utils/currentSession'; +import { setRuntimeConfig } from '../utils/runtimeConfig'; +import { useFestipodData, type FestipodDataContextValue } from './festipodDataBase'; + +const ALICE = 'http://alice.pod.test/alice'; +const ALICE_OUTBOX = 'http://alice.pod.test/alice/outbox'; +const BOB = 'http://bob.pod.test/bob'; +const E = 'http://bob.pod.test/bob/data/events/concert'; +const M = 'http://bob.pod.test/bob/data/events/concert-bar'; +const C = 'http://bob.pod.test/bob/data/events/concert-bar/attendees'; +const BACKEND = 'http://backend.test'; + +const EVENT_RECORD = { id: E, type: 'Event', name: 'Concert', startTime: '2026-11-01T20:00:00', 'dc:creator': BOB, to: 'as:Public' }; +const MP_RECORD = { id: M, type: 'Event', location: 'Au bar', 'as:context': E, 'dc:creator': BOB, 'apods:attendees': C, to: 'as:Public' }; +/** M as Bob's pod serves it before the backend attached its collection. */ +const MP_RECORD_UNLINKED = { id: M, type: 'Event', location: 'Au bar', 'as:context': E, 'dc:creator': BOB, to: 'as:Public' }; +/** M as ALICE's own meeting point on Bob's event, as her pod lists it before the backend attached its collection. */ +const OWN_MP_UNLINKED = { id: M, type: 'Event', location: 'Au bar', 'as:context': E, 'dc:creator': ALICE, to: 'as:Public' }; + +/** + * The scripted server: the collection's members, how many reads of it were + * served, and Alice's own list (`own`, with M's collection link once `ownLinked`). + */ +const server = { + members: [] as string[], + reads: 0, + posted: [] as string[], + own: [] as Record[], + ownLinked: false, + /** Whether Bob's pod serves M with its collection link yet. */ + mpLinked: true, +}; + +function json(body: unknown): Response { + return new Response(JSON.stringify(body), { status: 200, headers: { 'Content-Type': 'application/ld+json' } }); +} + +async function scriptedFetch(input: RequestInfo | URL, init?: RequestInit): Promise { + const url = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url; + if (init?.signal?.aborted) throw new DOMException('aborted', 'AbortError'); + if ((init?.method ?? 'GET') === 'POST') { + if (url !== ALICE_OUTBOX) return new Response(null, { status: 405 }); + server.posted.push(String((JSON.parse(String(init?.body)) as { type?: unknown }).type)); + return new Response(null, { status: 201 }); + } + switch (url) { + case ALICE: return json({ id: ALICE, type: 'Person', outbox: ALICE_OUTBOX }); + case `${BACKEND}/catalog`: return json({ type: 'OrderedCollection', items: [EVENT_RECORD, MP_RECORD] }); + case E: return json(EVENT_RECORD); + case M: return json(server.mpLinked ? MP_RECORD : MP_RECORD_UNLINKED); + case C: + server.reads++; + return json({ id: C, type: 'Collection', items: [...server.members] }); + default: return new Response('not found', { status: 404 }); + } +} + +/** Alice's pod: her own list is `server.own` (the attendees link appears once the backend attached it). */ +const alicePod = { + getList: async () => { + const data = server.own.map(record => (server.ownLinked ? { ...record, 'apods:attendees': C } : record)); + return { data, total: data.length }; + }, + getOne: async () => { throw new Error('no record'); }, + getMany: async () => ({ data: [] }), + create: async () => { throw new Error('read-only'); }, + update: async () => { throw new Error('read-only'); }, + deleteOne: async () => { throw new Error('read-only'); }, + getApiUrl: () => 'http://alice.pod.test', +} as unknown as DataProvider; + +let latest: FestipodDataContextValue | undefined; +function Probe() { + const data = useFestipodData(); + latest = data; + // What the event detail screen does: ask for the event it shows. + const { openEvent } = data; + useLayoutEffect(() => { openEvent(E); }, [openEvent]); + return null; +} + +// --- environment --------------------------------------------------------------- +const saved = { fetch: globalThis.fetch, build: activitypods.build, log: console.log, warn: console.warn }; +const installed: string[] = []; +let createRoot: (typeof import('react-dom/client'))['createRoot']; +let RefineDataProvider: (props: { children: ReactNode }) => ReactNode; +let root: ReturnType | undefined; + +beforeAll(async () => { + const window = new Window({ url: 'http://localhost:4000/' }); + const g = globalThis as Record; + for (const [name, value] of Object.entries({ window, document: window.document, navigator: window.navigator })) { + if (g[name] !== undefined) continue; + Object.defineProperty(globalThis, name, { value, writable: true, configurable: true }); + installed.push(name); + } + globalThis.fetch = scriptedFetch as typeof fetch; + activitypods.build = () => ({ dataProvider: alicePod }); + console.log = () => {}; + console.warn = () => {}; + createRoot = (await import('react-dom/client')).createRoot; + RefineDataProvider = (await import('./refineData')).RefineDataProvider; + setRuntimeConfig({ backendUrl: BACKEND }); + setCurrentSession({ webId: ALICE, token: 'alice-token' }); +}); + +afterEach(async () => { + root?.unmount(); + root = undefined; + latest = undefined; + // Let React run the work the unmount scheduled while the DOM globals still exist. + await new Promise(resolve => setTimeout(resolve, 50)); +}); + +afterAll(() => { + setCurrentSession(undefined); + setRuntimeConfig({}); + globalThis.fetch = saved.fetch; + activitypods.build = saved.build; + console.log = saved.log; + console.warn = saved.warn; + for (const name of installed) delete (globalThis as Record)[name]; +}); + +// --- helpers ----------------------------------------------------------------------- +async function until(condition: () => boolean, what: string, timeoutMs = 10000): Promise { + const start = Date.now(); + while (!condition()) { + if (Date.now() - start > timeoutMs) throw new Error(`timed out waiting for ${what}`); + await new Promise(resolve => setTimeout(resolve, 10)); + } +} + +/** Long enough for a served read to land in React state (in-process, no network). */ +const land = () => new Promise(resolve => setTimeout(resolve, 150)); + +/** What Alice's event detail shows: the event header count, M's count, and M's button. */ +function view(): { event?: number; mp?: number; joined: boolean } { + const data = latest; + if (!data) return { joined: false }; + const mp = data.getEventMeetingPoints(E).find(point => point.id === M); + return { event: data.getEvent(E)?.participantCount, mp: mp?.participantCount, joined: data.isParticipating(M) }; +} + +/** Poll `view()` until it matches, or give back the last one seen. */ +async function settledView(match: (v: ReturnType) => boolean, timeoutMs: number): Promise> { + const start = Date.now(); + let current = view(); + while (!match(current) && Date.now() - start < timeoutMs) { + await new Promise(resolve => setTimeout(resolve, 20)); + current = view(); + } + return current; +} + +const sleep = (ms: number) => new Promise(resolve => setTimeout(resolve, ms)); + +function render(members: string[], own: Record[] = [], mpLinked = true): void { + server.members = members; + server.reads = 0; + server.posted = []; + server.own = own; + server.ownLinked = false; + server.mpLinked = mpLinked; + const host = document.createElement('div'); + root = createRoot(host as unknown as Element); + root.render(); +} + +async function mount(members: string[]): Promise { + render(members); + await until(() => server.reads >= 1, 'the first read of the attendees collection'); + await until(() => view().mp === members.length, `M listed with ${members.length} inscrit`); +} + +/** + * How many reads followed the act by `untilMs` after it — a point later than + * the step of the re-read schedule after the confirming read. The schedule runs + * only while an act is pending, so an act released by its confirming read + * leaves exactly that many; one never released (or never re-read) does not. + * Never throws: the count goes into the compared timeline. + */ +async function readsBy(servedBefore: number, actAt: number, untilMs: number): Promise { + await sleep(Math.max(0, actAt + untilMs - Date.now())); + return server.reads - servedBefore; +} + +const NOT_JOINED = { event: 0, mp: 0, joined: false }; +const JOINED = { event: 1, mp: 1, joined: true }; + +// --- the race, both directions ------------------------------------------------------- + +test('a leave following a still-unconfirmed join is never undone by a read that predates the leave', async () => { + await mount([]); + + // JOIN — the backend has not processed it when the reads that follow it land. + let served = server.reads; + await latest!.joinMeetingPoint(M); + await until(() => server.reads >= served + 2, 'two reads after the join'); + await land(); + const afterJoin = view(); + + // The backend processes the join now — no read has seen it. + server.members = [ALICE]; + + // LEAVE — the backend processes it only after two more reads. + served = server.reads; + await latest!.leaveMeetingPoint(M); + const leftAt = Date.now(); + await until(() => server.reads >= served + 1, 'the first read after the leave'); + await land(); + const afterLeaveFirstRead = view(); + await until(() => server.reads >= served + 2, 'the second read after the leave'); + await land(); + const afterLeaveSecondRead = view(); + + // The backend processes the leave. + server.members = []; + const afterBackendProcessedLeave = await settledView(v => v.mp === 0 && v.event === 0 && !v.joined, 8000); + // The third read (the 3 s step) confirms the leave; the 6 s step must find nothing pending. + const readsAfterLeave = await readsBy(served, leftAt, 7000); + + expect(server.posted).toEqual(['Join', 'Leave']); + expect({ afterJoin, afterLeaveFirstRead, afterLeaveSecondRead, afterBackendProcessedLeave, readsAfterLeave }).toEqual({ + afterJoin: JOINED, + afterLeaveFirstRead: NOT_JOINED, + afterLeaveSecondRead: NOT_JOINED, + afterBackendProcessedLeave: NOT_JOINED, + readsAfterLeave: 3, + }); +}, 30000); + +test('a join following a still-unconfirmed leave is never undone by a read that predates the join', async () => { + await mount([ALICE]); + expect(view()).toEqual(JOINED); + + // LEAVE — the backend has not processed it when the reads that follow it land. + let served = server.reads; + await latest!.leaveMeetingPoint(M); + await until(() => server.reads >= served + 2, 'two reads after the leave'); + await land(); + const afterLeave = view(); + + // The backend processes the leave now — no read has seen it. + server.members = []; + + // JOIN — the backend processes it only after two more reads. + served = server.reads; + await latest!.joinMeetingPoint(M); + const joinedAt = Date.now(); + await until(() => server.reads >= served + 1, 'the first read after the join'); + await land(); + const afterJoinFirstRead = view(); + await until(() => server.reads >= served + 2, 'the second read after the join'); + await land(); + const afterJoinSecondRead = view(); + + // The backend processes the join. + server.members = [ALICE]; + const afterBackendProcessedJoin = await settledView(v => v.mp === 1 && v.event === 1 && v.joined, 8000); + const readsAfterJoin = await readsBy(served, joinedAt, 7000); + + expect(server.posted).toEqual(['Leave', 'Join']); + expect({ afterLeave, afterJoinFirstRead, afterJoinSecondRead, afterBackendProcessedJoin, readsAfterJoin }).toEqual({ + afterLeave: NOT_JOINED, + afterJoinFirstRead: JOINED, + afterJoinSecondRead: JOINED, + afterBackendProcessedJoin: JOINED, + readsAfterJoin: 3, + }); +}, 30000); + +test('an own meeting point whose collection link arrives after the act is still re-read until a read confirms the act', async () => { + // Alice's own list shows M before the backend attached its attendees collection. + render([], [OWN_MP_UNLINKED]); + await until(() => view().mp === 0, 'M listed (own, no collection linked yet)'); + // The backend attaches the collection now — no list read has seen it. + server.ownLinked = true; + + const served = server.reads; + await latest!.joinMeetingPoint(M); + const joinedAt = Date.now(); + // The list read the act triggers brings the link; the collection's first read predates the backend's processing. + await until(() => server.reads >= served + 1, 'the first read of the newly linked collection'); + await land(); + const afterFirstRead = view(); + server.members = [ALICE]; // the backend processes the join + + // The next scheduled step (1.5 s) re-reads the collection it did not know at + // act time, that read confirms the join, and the 3 s step finds nothing pending. + const readsAfterJoin = await readsBy(served, joinedAt, 4000); + expect({ afterFirstRead, afterConfirmed: view(), readsAfterJoin }).toEqual({ + afterFirstRead: JOINED, + afterConfirmed: JOINED, + readsAfterJoin: 2, + }); +}, 30000); + +test("another user's meeting point read before its collection was attached is re-read until a read confirms the act", async () => { + // Bob's pod served M before the backend attached its attendees collection. + render([], [], false); + await until(() => view().mp === 0, 'M listed (no collection linked yet)'); + // The backend attaches the collection now — no read of M has seen it. + server.mpLinked = true; + + const served = server.reads; + await latest!.joinMeetingPoint(M); + const joinedAt = Date.now(); + // M is read again by URI, which brings the link; the collection's first read predates the backend's processing. + await until(() => server.reads >= served + 1, 'the first read of the newly linked collection'); + await land(); + const afterFirstRead = view(); + server.members = [ALICE]; // the backend processes the join + + // The 1.5 s step re-reads the collection, that read confirms the join, and the 3 s step finds nothing pending. + const readsAfterJoin = await readsBy(served, joinedAt, 4000); + expect({ afterFirstRead, afterConfirmed: view(), readsAfterJoin }).toEqual({ + afterFirstRead: JOINED, + afterConfirmed: JOINED, + readsAfterJoin: 2, + }); +}, 30000); diff --git a/src/shared/context/refineData.tsx b/src/shared/context/refineData.tsx index d9b724b..f263a28 100644 --- a/src/shared/context/refineData.tsx +++ b/src/shared/context/refineData.tsx @@ -38,12 +38,13 @@ * refine-providers package implements no `custom` method: joining or leaving * a meeting point POSTs an `as:Join`/`as:Leave` activity to the user's own * outbox (`data/outbox.ts`), and the on-behalf backend maintains the - * `apods:attendees` collection. The collection travels inside the event - * record and is read by the participation binding (`membersOf`). A write made - * beside Refine does not refresh its cache, and the backend updates the - * collection asynchronously — so the join/leave reflects in an optimistic - * overlay here and the lists are invalidated for the write to land when it - * lands. + * `apods:attendees` collection. The collection is read by URI and embedded + * for the participation binding (`membersOf`). A write made beside Refine does + * not refresh its cache, and the backend updates the collection + * asynchronously — so the join/leave reflects in an optimistic overlay here, + * held until a read issued AFTER the act confirms it (read-your-writes: + * `settleMembershipActs` in `refineDerive.ts`), with a bounded schedule of + * re-reads of that one collection meanwhile. * * THE COUNT AND THE ALICE/BOB RULE. The participant count of a meeting point * is the collection's total — never the number of RESOLVED profiles, so it @@ -81,9 +82,15 @@ import { applyMembershipOverlay, countsWithOverlay, deriveContainer, - type MembershipOverlay, + expireMembershipAct, + MEMBERSHIP_REREAD_DELAYS_MS, + overlayOfPendingActs, + recordMembershipAct, + settleMembershipActs, + type MembershipOp, + type PendingActs, } from './refineDerive'; -import { usePublicRecords } from './usePublicRecords'; +import { usePublicRecords, useReadGenerations } from './usePublicRecords'; export function RefineDataProvider({ children }: { children: ReactNode }) { return ( @@ -176,6 +183,8 @@ function useRefineData(backend: RefineBackend): FestipodDataContextValue { () => new Set(ownSplit.meetingPoints.map(mp => mp.id)), [ownSplit.meetingPoints], ); + const ownMeetingPointIdsRef = useRef(ownMeetingPointIds); + ownMeetingPointIdsRef.current = ownMeetingPointIds; // --- OPENED EVENTS: the events screens asked to see --------------------------- // Grows only: an opened event stays in the read set for the page's life @@ -242,7 +251,14 @@ function useRefineData(backend: RefineBackend): FestipodDataContextValue { () => (foreignMeetingPointKey ? foreignMeetingPointKey.split('\n') : []), [foreignMeetingPointKey], ); - const foreignMeetingPointsRead = usePublicRecords(foreignMeetingPointIds, { retry: openCount, tokenOf }); + // Re-read on purpose only while an act waits on a meeting point whose record + // carried no collection link yet (`afterMembershipAct`). + const [meetingPointGenerations, readMeetingPointAgain] = useReadGenerations(); + const foreignMeetingPointsRead = usePublicRecords(foreignMeetingPointIds, { + refresh: meetingPointGenerations, + retry: openCount, + tokenOf, + }); // What was read is trusted for what it IS, not for what the catalog or the // route said it was: an "event" URI that reads as a meeting point (and the @@ -267,21 +283,35 @@ function useRefineData(backend: RefineBackend): FestipodDataContextValue { // --- ATTENDEES COLLECTIONS, read by URI and embedded -------------------------- // Every meeting point's linked collection, own and foreign alike — public by // design, read by URI with the same token policy, keyed by the URI the record - // links (never by what the fetched document claims). Re-read after a - // join/leave (`membershipRefresh`); a failed one is retried on the next open. - const [membershipRefresh, setMembershipRefresh] = useState(0); - const collectionKey = useMemo(() => { - const links = new Set(); + // links (never by what the fetched document claims). A join/leave re-reads + // the ONE collection it changed (`readCollectionAgain` moves its generation); + // a failed one is retried on the next open. + const collectionOf = useMemo(() => { + const links = new Map(); // meeting point id → collection URI for (const record of readRecords) { if (!(bindings.meetingPoint.isInstance?.(record) ?? false)) continue; const link = attendeesCollectionLink(record); - if (link) links.add(link); + if (!link) continue; + try { + links.set(bindings.meetingPoint.domainId(record), link); + } catch { + // No identifier: the derive skips this record too. + } } - return [...links].join('\n'); + return links; }, [readRecords, bindings]); + // Read when a re-read FIRES, not when the act was made: an own meeting point + // can be listed before the backend attached its collection, and the link + // arrives with a later list read. + const collectionOfRef = useRef(collectionOf); + collectionOfRef.current = collectionOf; + const collectionKey = useMemo(() => [...new Set(collectionOf.values())].join('\n'), [collectionOf]); const collectionIds = useMemo(() => (collectionKey ? collectionKey.split('\n') : []), [collectionKey]); + // An act learns the generation of the read it triggers at once — that + // number is what may confirm the act. + const [collectionGenerations, readCollectionAgain] = useReadGenerations(); const collectionsRead = usePublicRecords(session ? collectionIds : [], { - refresh: membershipRefresh, + refresh: collectionGenerations, retry: openCount, tokenOf, }); @@ -295,12 +325,17 @@ function useRefineData(backend: RefineBackend): FestipodDataContextValue { // --- SPLIT + PARTICIPATION/COUNT DERIVATION (vocabulary lives in bindings) -- const derived = useMemo(() => deriveContainer(embeddedRecords, bindings), [embeddedRecords, bindings]); - // --- OPTIMISTIC MEMBERSHIP OVERLAY ------------------------------------------ - // `watchShape`-style latency, Refine edition: the outbox post returns before - // the backend has maintained the collection, and the container re-fetch can - // still precede it. The overlay reflects the join/leave immediately and - // reconciles away once the read carries the truth — no polling. - const [overlay, setOverlay] = useState(new Map()); + // --- OPTIMISTIC MEMBERSHIP OVERLAY (read-your-writes) ---------------------- + // The outbox post returns before the backend has maintained the collection, + // and a read can land on either side of that. Each act is held as a pending + // act and reflected at once; it goes only when a read ISSUED AFTER it + // confirms it, or when the bounded wait runs out and a fresh re-read is + // taken as the truth (`settleMembershipActs`). The read on screen merely + // showing the asserted state is not a confirmation: it may predate the act. + const [pendingActs, setPendingActs] = useState(new Map()); + const pendingActsRef = useRef(pendingActs); + pendingActsRef.current = pendingActs; + const overlay = useMemo(() => overlayOfPendingActs(pendingActs), [pendingActs]); const participations = useMemo( () => applyMembershipOverlay(derived.participations, overlay, webId), @@ -316,23 +351,20 @@ function useRefineData(backend: RefineBackend): FestipodDataContextValue { [allMeetingPoints, ownMeetingPointIds], ); - // RECONCILIATION — drop each overlay entry once the container read carries - // what it asserts (join: membership present; leave: membership absent). + // RECONCILIATION — confront each pending act with the read: what it shows for + // the session on that meeting point, and which read generation it comes from. useEffect(() => { - if (overlay.size === 0 || !webId) return; - setOverlay(prev => { - let changed = false; - const next = new Map(prev); - for (const [mpId, op] of prev) { - const present = derived.participations.some(p => p.eventId === mpId && p.userId === webId); - if ((op === 'join' && present) || (op === 'leave' && !present)) { - next.delete(mpId); - changed = true; - } - } - return changed ? next : prev; - }); - }, [derived.participations, overlay, webId]); + if (pendingActs.size === 0 || !webId) return; + setPendingActs(prev => settleMembershipActs(prev, mpId => { + if (!derived.meetingPoints.some(mp => mp.id === mpId)) return undefined; + const collection = collectionOf.get(mpId); + return { + member: derived.participations.some(p => p.eventId === mpId && p.userId === webId), + generation: collection === undefined ? undefined : collectionsRead.readAt.get(collection), + attempted: collection === undefined ? undefined : collectionsRead.attemptedAt.get(collection), + }; + })); + }, [pendingActs, derived, collectionOf, collectionsRead.readAt, collectionsRead.attemptedAt, webId]); // --- MY PROFILE (the app's own entity id space) ------------------------------ // The actor (WebID) document links the `as:Profile` record via `url`; the @@ -521,29 +553,47 @@ function useRefineData(backend: RefineBackend): FestipodDataContextValue { const { mutateAsync: updateRecord } = useUpdate(); const invalidate = useInvalidate(); - // A join/leave is maintained asynchronously by the backend, so the immediate - // invalidate can re-fetch BEFORE the collection is updated — and nothing - // re-reads afterwards (the list subscription only reports membership changes). - // ONE delayed re-invalidation per action closes that gap: no polling, at most - // one pending timer, cancelled on unmount. - const revalidateTimer = useRef | null>(null); + // A join/leave is maintained asynchronously by the backend, so the read that + // follows the act can still precede the backend's update. The collection the + // act changed is re-read at once, then on a FINITE schedule + // (`MEMBERSHIP_REREAD_DELAYS_MS`) while the act is unconfirmed; the last + // re-read ends the wait and its result wins. While the meeting point links no + // collection yet (read before the backend attached it), a step re-reads the + // meeting point's RECORD instead, where the link will arrive: the own list + // for an own meeting point, the record by URI for another user's. A newer act + // on the same meeting point replaces the schedule; every timer is cleared on + // unmount. + const actTimers = useRef(new Map[]>()); useEffect(() => () => { - if (revalidateTimer.current) clearTimeout(revalidateTimer.current); - revalidateTimer.current = null; + for (const timers of actTimers.current.values()) timers.forEach(clearTimeout); + actTimers.current.clear(); }, []); - // What a join/leave changes: the attendees collections (read by URI) and the - // own list. - const refreshMemberships = useCallback(() => { - void invalidate({ resource: bindings.event.resource, invalidates: ['list'] }); - setMembershipRefresh(n => n + 1); - }, [invalidate, bindings]); - const scheduleListRevalidate = useCallback(() => { - if (revalidateTimer.current) clearTimeout(revalidateTimer.current); - revalidateTimer.current = setTimeout(() => { - revalidateTimer.current = null; - refreshMemberships(); - }, 1500); - }, [refreshMemberships]); + const afterMembershipAct = useCallback((mpId: string, op: MembershipOp) => { + const invalidateOwnList = () => void invalidate({ resource: bindings.event.resource, invalidates: ['list'] }); + // Re-read what can confirm the act: its collection — the generation of that + // read is the first one issued after the outbox accepted the act. With no + // collection linked yet, re-read the record that will carry the link (no + // generation: any read of a collection that appears later is after the act). + const readAgain = (): number | undefined => { + const collection = collectionOfRef.current.get(mpId); + if (collection !== undefined) return readCollectionAgain(collection); + if (ownMeetingPointIdsRef.current.has(mpId)) invalidateOwnList(); + else readMeetingPointAgain(mpId); + return undefined; + }; + const confirmFrom = readAgain(); + setPendingActs(prev => recordMembershipAct(prev, mpId, op, confirmFrom ?? 0)); + // The own list, as after every write (unless the line above just did). + if (confirmFrom !== undefined || !ownMeetingPointIdsRef.current.has(mpId)) invalidateOwnList(); + for (const timer of actTimers.current.get(mpId) ?? []) clearTimeout(timer); + const last = MEMBERSHIP_REREAD_DELAYS_MS.length - 1; + actTimers.current.set(mpId, MEMBERSHIP_REREAD_DELAYS_MS.map((delay, step) => setTimeout(() => { + // Confirmed meanwhile: nothing left to read for it. + if (!pendingActsRef.current.has(mpId)) return; + const generation = readAgain(); + if (step === last) setPendingActs(prev => expireMembershipAct(prev, mpId, generation ?? 0)); + }, delay))); + }, [readCollectionAgain, readMeetingPointAgain, invalidate, bindings]); const createEvent = useCallback(async (event: Omit): Promise => { console.log(`${logPrefix} createEvent (refine):`, event.title); @@ -610,10 +660,8 @@ function useRefineData(backend: RefineBackend): FestipodDataContextValue { } console.log(`${logPrefix} joinMeetingPoint (refine):`, meetingPointId); await postJoinActivity({ meetingPointUri: mp.id, hostUri: mp.hostId, session }); - setOverlay(prev => new Map(prev).set(mp.id, 'join')); - refreshMemberships(); - scheduleListRevalidate(); - }, [logPrefix, allMeetingPoints, session, webId, refreshMemberships, scheduleListRevalidate]); + afterMembershipAct(mp.id, 'join'); + }, [logPrefix, allMeetingPoints, session, webId, afterMembershipAct]); const leaveMeetingPoint = useCallback(async (meetingPointId: string, userId?: string) => { if (userId && userId !== webId) { @@ -633,10 +681,8 @@ function useRefineData(backend: RefineBackend): FestipodDataContextValue { } console.log(`${logPrefix} leaveMeetingPoint (refine):`, meetingPointId); await postLeaveActivity({ meetingPointUri: mp.id, hostUri: mp.hostId, session }); - setOverlay(prev => new Map(prev).set(mp.id, 'leave')); - refreshMemberships(); - scheduleListRevalidate(); - }, [logPrefix, allMeetingPoints, participations, session, webId, refreshMemberships, scheduleListRevalidate]); + afterMembershipAct(mp.id, 'leave'); + }, [logPrefix, allMeetingPoints, participations, session, webId, afterMembershipAct]); // --- The rest of the contract ------------------------------------------------------- const addFriend = useCallback((friendId: string) => { diff --git a/src/shared/context/refineDerive.test.ts b/src/shared/context/refineDerive.test.ts index 17d0bc8..024c580 100644 --- a/src/shared/context/refineDerive.test.ts +++ b/src/shared/context/refineDerive.test.ts @@ -1,5 +1,16 @@ import { expect, spyOn, test } from 'bun:test'; -import { deriveContainer, applyMembershipOverlay, countsWithOverlay } from './refineDerive'; +import { + deriveContainer, + applyMembershipOverlay, + countsWithOverlay, + expireMembershipAct, + MEMBERSHIP_REREAD_DELAYS_MS, + overlayOfPendingActs, + recordMembershipAct, + settleMembershipActs, + type MembershipObservation, + type PendingActs, +} from './refineDerive'; import { getBackend } from '../refine/providers'; import type { FpParticipationData } from '../data/types'; @@ -212,3 +223,114 @@ test('a record the bindings reject is skipped; the others are derived', () => { expect(derived.meetingPoints).toEqual([]); warn.mockRestore(); }); + +// ── Read-your-writes: when a pending join/leave may go ────────────────────── +// State: the act (join | leave), how many unconfirmed acts it replaced (0, 1, +// 2+), the read's generation (none, before the act, after it, the fresh one at +// the end of the wait) and what the read shows (asserted state or not). + +const NONE: PendingActs = new Map(); +/** The read says: the session is (not) a member, from a read of `generation`. */ +const reads = (member: boolean, generation: number | undefined) => + (): MembershipObservation => ({ member, generation }); + +test('a clean act is confirmed by a read issued after it that shows the asserted state', () => { + const join = recordMembershipAct(NONE, MP, 'join', 3); + expect(settleMembershipActs(join, reads(true, 3)).has(MP)).toBe(false); + const leave = recordMembershipAct(NONE, MP, 'leave', 3); + expect(settleMembershipActs(leave, reads(false, 4)).has(MP)).toBe(false); +}); + +test('a read issued after a clean act that does not show it yet keeps the act (the backend has not processed it)', () => { + const join = recordMembershipAct(NONE, MP, 'join', 3); + expect(settleMembershipActs(join, reads(false, 3))).toBe(join); + const leave = recordMembershipAct(NONE, MP, 'leave', 3); + expect(settleMembershipActs(leave, reads(true, 5))).toBe(leave); +}); + +test('a read from BEFORE the act confirms nothing, even when it shows the asserted state', () => { + const leave = recordMembershipAct(NONE, MP, 'leave', 3); + expect(settleMembershipActs(leave, reads(false, 2))).toBe(leave); + expect(settleMembershipActs(leave, reads(false, undefined))).toBe(leave); + const join = recordMembershipAct(NONE, MP, 'join', 3); + expect(settleMembershipActs(join, reads(true, 2))).toBe(join); +}); + +test('an act replacing ONE unconfirmed opposite act: the asserted state alone confirms nothing', () => { + // Join (unconfirmed), then leave: "neither processed" and "both processed" both read as absent. + const chain = recordMembershipAct(recordMembershipAct(NONE, MP, 'join', 1), MP, 'leave', 3); + expect(chain.get(MP)?.unconfirmedBefore).toBe(1); + expect(settleMembershipActs(chain, reads(false, 3))).toBe(chain); +}); + +test('…until a read after it shows the opposite state: the earlier act is processed, this one now confirms normally', () => { + const chain = recordMembershipAct(recordMembershipAct(NONE, MP, 'join', 1), MP, 'leave', 3); + const decided = settleMembershipActs(chain, reads(true, 3)); // the join landed, the leave not yet + expect(decided.get(MP)).toEqual({ op: 'leave', confirmFrom: 3, unconfirmedBefore: 0 }); + expect(settleMembershipActs(decided, reads(false, 4)).has(MP)).toBe(false); + // Symmetric: leave (unconfirmed), then join. + const joinChain = recordMembershipAct(recordMembershipAct(NONE, MP, 'leave', 1), MP, 'join', 3); + expect(settleMembershipActs(joinChain, reads(true, 3))).toBe(joinChain); + const joinDecided = settleMembershipActs(joinChain, reads(false, 3)); + expect(settleMembershipActs(joinDecided, reads(true, 4)).has(MP)).toBe(false); +}); + +test('with two or more unconfirmed acts before it, no read decides: only the end of the wait does', () => { + const chain = recordMembershipAct( + recordMembershipAct(recordMembershipAct(NONE, MP, 'join', 1), MP, 'leave', 2), MP, 'join', 3); + expect(chain.get(MP)?.unconfirmedBefore).toBe(2); + expect(settleMembershipActs(chain, reads(true, 9))).toBe(chain); + expect(settleMembershipActs(chain, reads(false, 9))).toBe(chain); + const expired = expireMembershipAct(chain, MP, 10); + expect(settleMembershipActs(expired, reads(false, 9))).toBe(expired); + expect(settleMembershipActs(expired, reads(false, 10)).has(MP)).toBe(false); +}); + +test('a repeated act asserts nothing new: it inherits the earlier act\'s standing', () => { + const twice = recordMembershipAct(recordMembershipAct(NONE, MP, 'join', 1), MP, 'join', 3); + expect(twice.get(MP)).toEqual({ op: 'join', confirmFrom: 3, unconfirmedBefore: 0 }); +}); + +test('when the bounded wait runs out, the fresh re-read wins whatever it shows', () => { + const join = recordMembershipAct(NONE, MP, 'join', 3); + const expired = expireMembershipAct(join, MP, 9); + expect(expired.get(MP)?.settleFrom).toBe(9); + // A read older than the fresh one still cannot drop it… + expect(settleMembershipActs(expired, reads(false, 8))).toBe(expired); + // …the fresh one does, even showing the act never landed. + expect(settleMembershipActs(expired, reads(false, 9)).has(MP)).toBe(false); + // Expiring twice keeps the first fresh generation. + expect(expireMembershipAct(expired, MP, 12)).toBe(expired); + // With no read issued after the act at all — none readable, or the meeting point gone from the read set — + // nothing settles it, even once the wait is over: a read from before the act never undoes it. + expect(settleMembershipActs(expired, () => undefined)).toBe(expired); + expect(settleMembershipActs(expired, reads(false, undefined))).toBe(expired); + expect(settleMembershipActs(expired, reads(false, 2))).toBe(expired); + expect(settleMembershipActs(join, () => undefined)).toBe(join); +}); + +test('when the fresh re-read at the end of the wait FAILS, the last read issued after the act wins — never one from before it', () => { + const expired = expireMembershipAct(recordMembershipAct(NONE, MP, 'join', 3), MP, 9); + // Generation 9 came back failed: the record shown is still generation 8's, issued after the act. + expect(settleMembershipActs(expired, () => ({ member: false, generation: 8, attempted: 9 })).has(MP)).toBe(false); + // Every read since the act failed (offline): the record shown predates the act, which stays. + expect(settleMembershipActs(expired, () => ({ member: false, generation: 2, attempted: 9 }))).toBe(expired); + // Before generation 9 came back at all, the act stays. + expect(settleMembershipActs(expired, () => ({ member: false, generation: 8, attempted: 8 }))).toBe(expired); + // An attempt alone never confirms an act whose wait is not over. + const join = recordMembershipAct(NONE, MP, 'join', 3); + expect(settleMembershipActs(join, () => ({ member: true, generation: 2, attempted: 5 }))).toBe(join); +}); + +test('acts on different meeting points settle independently; the overlay is their ops', () => { + const both = recordMembershipAct(recordMembershipAct(NONE, MP, 'join', 1), MP2, 'leave', 1); + const settled = settleMembershipActs(both, id => (id === MP ? { member: true, generation: 1 } : { member: true, generation: 1 })); + expect([...settled.keys()]).toEqual([MP2]); + expect([...overlayOfPendingActs(both)]).toEqual([[MP, 'join'], [MP2, 'leave']]); +}); + +test('the re-read schedule is finite and increasing', () => { + expect(MEMBERSHIP_REREAD_DELAYS_MS.length).toBeGreaterThan(0); + const previous = [0, ...MEMBERSHIP_REREAD_DELAYS_MS]; + expect(MEMBERSHIP_REREAD_DELAYS_MS.every((delay, i) => delay > (previous[i] ?? delay))).toBe(true); +}); diff --git a/src/shared/context/refineDerive.ts b/src/shared/context/refineDerive.ts index c522fcc..0619353 100644 --- a/src/shared/context/refineDerive.ts +++ b/src/shared/context/refineDerive.ts @@ -16,10 +16,11 @@ * (the Alice/Bob rule). * * 2. `applyMembershipOverlay` reflects a join/leave made beside Refine (through - * the outbox) before the collection read carries it. It reconciles away on - * its own once the read shows the truth — no polling. `countsWithOverlay` + * the outbox) before the collection read carries it. `countsWithOverlay` * applies the same overlay to the COUNTS, so a join shows in the number the - * moment it shows in the membership. + * moment it shows in the membership. WHEN the overlay may go is decided by + * `settleMembershipActs` (read-your-writes, below): only a read issued after + * the act can confirm it, or the bounded wait runs out. * * THE COLLECTION MUST BE EMBEDDED TO BE COUNTED. The pod links a meeting * point's collection by URI and never embeds it (verified 2026-10-12); the @@ -181,3 +182,130 @@ export function countsWithOverlay( }); return { events, meetingPoints }; } + +/** + * READ-YOUR-WRITES FOR A JOIN/LEAVE — when the overlay may go. + * + * The outbox accepts an act before the backend has updated the attendees + * collection, so a read can land on either side of the backend's processing. + * The overlay entry for a meeting point is therefore dropped only when a read + * CONFIRMS it — never because the read on screen happens to show the asserted + * state: + * + * - Only a read ISSUED AFTER the act can confirm it (`confirmFrom`: the first + * read generation of the collection made once the outbox accepted the act). + * A read from before says nothing about it. + * - An act made while an earlier act of this session on the same meeting point + * is still unconfirmed is NOT CLEAN: the collection shows the same thing + * before both acts and after both ("neither processed" and "both processed" + * look alike), so the asserted state alone confirms nothing. One case is + * decidable: with ONE unconfirmed act before it, a read showing the opposite + * of the asserted state proves the earlier act processed and this one not + * yet — from then on the act is clean. With two or more, no read decides. + * - The wait is BOUNDED: when it runs out, the caller issues one fresh re-read + * (`expireMembershipAct` records its generation) and that read wins, whatever + * it shows — and if it fails, the last good read issued after the act wins. + * With NO read issued after the act at all (offline, a collection that never + * became readable), nothing can settle it and the act stays shown: a read + * from before the act never undoes it, at the end of the wait included. + * Until then the caller re-reads on a fixed, finite schedule + * (`MEMBERSHIP_REREAD_DELAYS_MS`) — never an open-ended poll. + */ + +/** A join or a leave this session made on a meeting point. */ +export type MembershipOp = 'join' | 'leave'; + +/** One act of this session on a meeting point, not yet confirmed by a read. */ +export interface PendingAct { + op: MembershipOp; + /** Reads of the collection made for this generation or later were issued after the act. */ + confirmFrom: number; + /** How many acts of this session on the same meeting point were still unconfirmed when this one was made. */ + unconfirmedBefore: number; + /** Once the bounded wait ran out: the generation of the fresh re-read then issued — its result wins. */ + settleFrom?: number; +} + +/** The pending acts, keyed by meeting point id. */ +export type PendingActs = ReadonlyMap; + +/** + * What the CURRENT read says about one meeting point: whether it carries the + * session's membership, the generation of the collection read it comes from + * (`undefined` when no collection read has succeeded for it), and the + * generation of the last collection read that came back, failed or not. + */ +export interface MembershipObservation { + member: boolean; + generation: number | undefined; + attempted?: number; +} + +/** + * When the collection is re-read after an act, in ms from the act — the first + * re-read is immediate and not listed. The LAST one is the fresh re-read whose + * result wins (`expireMembershipAct`). Finite by construction. + */ +export const MEMBERSHIP_REREAD_DELAYS_MS: readonly number[] = [1500, 3000, 6000, 12000, 20000, 30000]; + +/** Record an act the outbox accepted; it replaces any earlier act on the same meeting point. */ +export function recordMembershipAct( + pending: PendingActs, + meetingPointId: string, + op: MembershipOp, + confirmFrom: number, +): PendingActs { + const previous = pending.get(meetingPointId); + // A repeated act asserts nothing new: it inherits what was still unconfirmed + // under the earlier one. An opposite act stacks on it. + const unconfirmedBefore = previous === undefined + ? 0 + : previous.op === op ? previous.unconfirmedBefore : previous.unconfirmedBefore + 1; + return new Map(pending).set(meetingPointId, { op, confirmFrom, unconfirmedBefore }); +} + +/** The bounded wait ran out: the read of generation `settleFrom` (or later) will win. */ +export function expireMembershipAct(pending: PendingActs, meetingPointId: string, settleFrom: number): PendingActs { + const act = pending.get(meetingPointId); + if (!act || act.settleFrom !== undefined) return pending; + return new Map(pending).set(meetingPointId, { ...act, settleFrom }); +} + +/** + * Confront the pending acts with what the current read says. Returns the SAME + * map when nothing changes (a state setter can bail out on it). + */ +export function settleMembershipActs( + pending: PendingActs, + observe: (meetingPointId: string) => MembershipObservation | undefined, +): PendingActs { + let next: Map | undefined; + const set = (id: string, act: PendingAct | undefined) => { + next ??= new Map(pending); + if (act) next.set(id, act); + else next.delete(id); + }; + for (const [id, act] of pending) { + const seen = observe(id); + const generation = seen?.generation; + // Only a read issued after the act says anything about it. + if (seen === undefined || generation === undefined || generation < act.confirmFrom) continue; + if (act.settleFrom !== undefined && (generation >= act.settleFrom || (seen.attempted ?? -1) >= act.settleFrom)) { + // The wait is over: the fresh read wins — or, if it failed, the last + // read issued after the act. + set(id, undefined); + continue; + } + const asserted = seen.member === (act.op === 'join'); + if (asserted && act.unconfirmedBefore === 0) set(id, undefined); + else if (!asserted && act.unconfirmedBefore === 1) set(id, { ...act, unconfirmedBefore: 0 }); + } + return next ?? pending; +} + +/** The overlay the views apply: each pending act's op. */ +export function overlayOfPendingActs(pending: PendingActs): MembershipOverlay { + const overlay: MembershipOverlay = new Map(); + for (const [id, act] of pending) overlay.set(id, act.op); + return overlay; +} diff --git a/src/shared/context/usePublicRecords.hook.test.tsx b/src/shared/context/usePublicRecords.hook.test.tsx new file mode 100644 index 0000000..89eb160 --- /dev/null +++ b/src/shared/context/usePublicRecords.hook.test.tsx @@ -0,0 +1,187 @@ +/** + * `usePublicRecords` mounted for real (happy-dom), against a fetch whose + * answers the test releases by hand — so "in flight" is a state the test holds, + * not a race it hopes for. + * + * What is pinned here: each URI is read ON ITS OWN — a slow or failing one + * costs that URI only; moving one URI's generation re-reads that URI and never + * aborts or restarts another's read in flight; a URI whose own generation moves + * on while in flight is read again and the late answer of the old read is + * never taken; a re-open (retry) while a read is in flight does not restart it, + * yet still gets a read if that one fails — nothing is ever left pending with + * no read in flight; the token goes only where the caller grants it. + */ + +import { afterAll, afterEach, beforeAll, expect, test } from 'bun:test'; +import { Window } from 'happy-dom'; +import type { PublicRecordsOptions, PublicRecordsRead } from './usePublicRecords'; + +const A = 'http://bob.pod.test/bob/data/a'; +const B = 'http://bob.pod.test/bob/data/b'; +const OWN = 'http://alice.pod.test/alice/data/c'; + +interface Call { uri: string; auth: string | null; aborted: () => boolean; answer: (body: unknown, status?: number) => void } +let calls: Call[] = []; + +/** Every GET waits until the test answers it (or the read is aborted). */ +function heldFetch(input: RequestInfo | URL, init?: RequestInit): Promise { + const uri = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url; + return new Promise((resolve, reject) => { + const signal = init?.signal; + signal?.addEventListener('abort', () => reject(new DOMException('aborted', 'AbortError'))); + calls.push({ + uri, + auth: new Headers(init?.headers).get('Authorization'), + aborted: () => signal?.aborted ?? false, + answer: (body, status = 200) => + resolve(new Response(JSON.stringify(body), { status, headers: { 'Content-Type': 'application/ld+json' } })), + }); + }); +} + +const callsTo = (uri: string) => calls.filter(c => c.uri === uri); +const lastCall = (uri: string): Call => { + const found = callsTo(uri).at(-1); + if (!found) throw new Error(`no read of ${uri}`); + return found; +}; + +// --- environment ------------------------------------------------------------------- +const saved = { fetch: globalThis.fetch }; +const installed: string[] = []; +let createRoot: (typeof import('react-dom/client'))['createRoot']; +let usePublicRecords: (uris: readonly string[], options?: PublicRecordsOptions) => PublicRecordsRead; +let root: ReturnType | undefined; +let latest: PublicRecordsRead | undefined; + +function Probe({ uris, options }: { uris: readonly string[]; options: PublicRecordsOptions }) { + latest = usePublicRecords(uris, options); + return null; +} + +beforeAll(async () => { + const window = new Window({ url: 'http://localhost:4000/' }); + const g = globalThis as Record; + for (const [name, value] of Object.entries({ window, document: window.document, navigator: window.navigator })) { + if (g[name] !== undefined) continue; + Object.defineProperty(globalThis, name, { value, writable: true, configurable: true }); + installed.push(name); + } + globalThis.fetch = heldFetch as typeof fetch; + createRoot = (await import('react-dom/client')).createRoot; + usePublicRecords = (await import('./usePublicRecords')).usePublicRecords; +}); + +afterEach(async () => { + root?.unmount(); + root = undefined; + latest = undefined; + calls = []; + await new Promise(resolve => setTimeout(resolve, 50)); +}); + +afterAll(() => { + globalThis.fetch = saved.fetch; + for (const name of installed) delete (globalThis as Record)[name]; +}); + +const tick = () => new Promise(resolve => setTimeout(resolve, 30)); + +async function show(uris: readonly string[], options: PublicRecordsOptions = {}): Promise { + if (!root) root = createRoot(document.createElement('div') as unknown as Element); + root.render(); + await tick(); +} + +// --- the cases ------------------------------------------------------------------------ + +test('each URI lands on its own: a slow one withholds nothing, a failing one costs only itself', async () => { + await show([A, B]); + lastCall(B).answer({ id: B, type: 'Event' }); + await tick(); + expect([...latest!.records.keys()]).toEqual([B]); + expect([...latest!.pending]).toEqual([A]); + lastCall(A).answer({}, 404); + await tick(); + expect([...latest!.records.keys()]).toEqual([B]); + expect(latest!.pending.size).toBe(0); + expect(latest!.attemptedAt.get(A)).toBe(0); +}); + +test("moving one URI's generation re-reads that URI only — another's read in flight is not restarted", async () => { + await show([A, B]); + lastCall(B).answer({ id: B }); + await tick(); + // A is still in flight when B is asked again. + await show([A, B], { refresh: new Map([[B, 1]]) }); + expect(callsTo(A)).toHaveLength(1); + expect(lastCall(A).aborted()).toBe(false); + expect(callsTo(B)).toHaveLength(2); + lastCall(A).answer({ id: A }); + lastCall(B).answer({ id: B, name: 'again' }); + await tick(); + expect(latest!.records.get(A)).toEqual({ id: A }); + expect(latest!.readAt.get(A)).toBe(0); + expect(latest!.readAt.get(B)).toBe(1); +}); + +test("a URI whose own generation moves on while in flight is read again, and the old read's answer is never taken", async () => { + await show([A]); + const first = lastCall(A); + await show([A], { refresh: new Map([[A, 1]]) }); + expect(first.aborted()).toBe(true); + expect(callsTo(A)).toHaveLength(2); + first.answer({ id: A, name: 'before' }); + lastCall(A).answer({ id: A, name: 'after' }); + await tick(); + expect(latest!.records.get(A)).toEqual({ id: A, name: 'after' }); + expect(latest!.readAt.get(A)).toBe(1); +}); + +test('a re-open while a read is in flight does not restart it; a success then answers it', async () => { + await show([A]); + await show([A], { retry: 1 }); + expect(callsTo(A)).toHaveLength(1); + lastCall(A).answer({ id: A }); + await tick(); + expect(callsTo(A)).toHaveLength(1); + expect(latest!.pending.size).toBe(0); + expect(latest!.records.get(A)).toEqual({ id: A }); +}); + +test('…but if that read FAILS, the re-open still gets its read — nothing stays pending with nothing in flight', async () => { + await show([A]); + await show([A], { retry: 1 }); + lastCall(A).answer({}, 504); // the read issued before the re-open fails + await tick(); + expect(callsTo(A)).toHaveLength(2); + expect([...latest!.pending]).toEqual([A]); + lastCall(A).answer({ id: A }); + await tick(); + expect(latest!.pending.size).toBe(0); + expect(latest!.records.get(A)).toEqual({ id: A }); + // A failure with no re-open since it was issued waits for the next re-open. + await show([A, B], { retry: 1 }); + lastCall(B).answer({}, 503); + await tick(); + expect(callsTo(B)).toHaveLength(1); + expect(latest!.pending.size).toBe(0); + await show([A, B], { retry: 2 }); + expect(callsTo(B)).toHaveLength(2); +}); + +test('a URI that leaves the set has its read aborted; unmounting aborts the rest', async () => { + await show([A, B]); + await show([B]); + expect(lastCall(A).aborted()).toBe(true); + expect(lastCall(B).aborted()).toBe(false); + root?.unmount(); + root = undefined; + expect(lastCall(B).aborted()).toBe(true); +}); + +test('the token goes only where the caller grants it', async () => { + await show([A, OWN], { tokenOf: uri => (uri.startsWith('http://alice.pod.test/') ? 'alice-token' : undefined) }); + expect(lastCall(OWN).auth).toBe('Bearer alice-token'); + expect(lastCall(A).auth).toBeNull(); +}); diff --git a/src/shared/context/usePublicRecords.test.ts b/src/shared/context/usePublicRecords.test.ts index d09ddf4..686293f 100644 --- a/src/shared/context/usePublicRecords.test.ts +++ b/src/shared/context/usePublicRecords.test.ts @@ -4,15 +4,22 @@ import { landedAfter, needsRead } from './usePublicRecords'; const RECORD = { id: 'http://localhost:3000/bob/data/e1', type: 'Event' }; describe('landedAfter — last good wins', () => { - test('a successful read stores the record', () => { - expect(landedAfter(undefined, 0, 0, RECORD)).toEqual({ refresh: 0, retry: 0, record: RECORD, failed: false }); + test('a successful read stores the record and the generation that produced it', () => { + expect(landedAfter(undefined, 0, 0, RECORD)).toEqual({ refresh: 0, retry: 0, record: RECORD, recordRefresh: 0, failed: false }); }); - test('a failed re-read keeps the record already shown, and is marked failed', () => { + test('a failed re-read keeps the record already shown — and its OLDER generation — and is marked failed', () => { const shown = landedAfter(undefined, 0, 0, RECORD); - expect(landedAfter(shown, 1, 0, undefined)).toEqual({ refresh: 1, retry: 0, record: RECORD, failed: true }); + // The record kept is still generation 0's: it must not pass for a read issued after a write. + expect(landedAfter(shown, 1, 0, undefined)).toEqual({ refresh: 1, retry: 0, record: RECORD, recordRefresh: 0, failed: true }); + }); + test('a retry that succeeds is a read of the current generation', () => { + const failed = landedAfter(landedAfter(undefined, 0, 0, RECORD), 2, 0, undefined); + expect(landedAfter(failed, 2, 1, RECORD).recordRefresh).toBe(2); }); test('a first read that fails stores nothing', () => { - expect(landedAfter(undefined, 0, 0, undefined).record).toBeUndefined(); + const landed = landedAfter(undefined, 0, 0, undefined); + expect(landed.record).toBeUndefined(); + expect(landed.recordRefresh).toBeUndefined(); }); }); diff --git a/src/shared/context/usePublicRecords.ts b/src/shared/context/usePublicRecords.ts index ff1cc91..f16fe5a 100644 --- a/src/shared/context/usePublicRecords.ts +++ b/src/shared/context/usePublicRecords.ts @@ -5,47 +5,67 @@ * - `records` is keyed by the REQUESTED URI. A URI never read successfully is * absent. LAST GOOD WINS: a re-read that fails keeps the record already * shown (a blip never blanks the screen or turns a membership off). - * - `refresh` (a counter) re-reads every URI — what a join/leave needs for the - * attendees collections. `retry` (a counter) re-reads only the URIs whose + * - `refresh` (a generation PER URI, absent = 0) re-reads a URI each time its + * generation moves — what a join/leave needs for the one attendees + * collection it changed. `retry` (a counter) re-reads only the URIs whose * last read FAILED — what re-opening an event needs, without re-reading * everything that already landed. - * - `pending` holds the current URIs whose read for the current counters has - * not landed yet. It is computed from state, so a URI is pending from the - * very render it enters the set — "not read yet" is never mistaken for - * "absent". - * - Each URI lands on its own: one slow or dead host delays that URI only - * (and gives up after the read timeout), never the others. + * - `readAt` gives, per URI, the generation of the read that produced the + * record SHOWN — so a caller can tell a read issued after a write from one + * issued before it. A failed re-read leaves it where the record came from: + * the record kept is still the older read's. `attemptedAt` gives the + * generation of the last read that LANDED, failed or not — "the read asked + * for has come back", whatever it brought. + * - `pending` holds the current URIs whose read for the current generation and + * retry has not landed yet. It is computed from state, so a URI is pending + * from the very render it enters the set — "not read yet" is never mistaken + * for "absent". + * - Each URI is read ON ITS OWN: one slow or dead host delays that URI only + * (and gives up after the read timeout), never the others — and re-reading + * one URI (its generation moved) never aborts or restarts another's read + * still in flight. A read in flight is aborted only when its URI leaves the + * set, when its own generation moves on, or on unmount. */ -import { useEffect, useMemo, useRef, useState } from 'react'; +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; import type { JsonRecord } from '../data/jsonRecord'; -import { readEachRecord } from '../data/publicRead'; +import { fetchPublicRecord } from '../data/publicRead'; export interface Landed { - /** The counters this read was made for. */ + /** The generation and retry this read was made for. */ refresh: number; retry: number; /** The last record read successfully (kept across a failed re-read). */ record: JsonRecord | undefined; + /** The generation of the read that produced `record`. */ + recordRefresh: number | undefined; /** Whether THIS read failed. */ failed: boolean; } export interface PublicRecordsRead { records: ReadonlyMap; + /** Per URI, the generation of the read that produced the record in `records`. */ + readAt: ReadonlyMap; + /** Per URI, the generation of the last read that landed, successful or failed. */ + attemptedAt: ReadonlyMap; pending: ReadonlySet; } export interface PublicRecordsOptions { - refresh?: number; + /** Per-URI read generation (absent = 0): moving it re-reads that URI. */ + refresh?: ReadonlyMap; retry?: number; /** The token to attach for a URI — the caller grants one only for its own origin. */ tokenOf?: (uri: string) => string | undefined; } +const NO_GENERATIONS: ReadonlyMap = new Map(); + /** - * What a read landing leaves for its URI: the counters it was made for, and - * the LAST GOOD record — a failed read keeps the previous one. + * What a read landing leaves for its URI: the generation and retry it was made + * for, and the LAST GOOD record with the generation that produced it — a failed + * read keeps the previous one. */ export function landedAfter( previous: Landed | undefined, @@ -53,18 +73,40 @@ export function landedAfter( retry: number, record: JsonRecord | undefined, ): Landed { - return { refresh, retry, record: record ?? previous?.record, failed: record === undefined }; + return record === undefined + ? { refresh, retry, record: previous?.record, recordRefresh: previous?.recordRefresh, failed: true } + : { refresh, retry, record, recordRefresh: refresh, failed: false }; } -/** Does this URI need a read for the current counters? */ +/** Does this URI need a read for its current generation and retry? */ export function needsRead(entry: Landed | undefined, refresh: number, retry: number): boolean { if (!entry || entry.refresh !== refresh) return true; return entry.failed && entry.retry !== retry; } +/** + * Per-URI read generations for `usePublicRecords`' `refresh`: `readAgain(uri)` + * moves that URI's generation and returns the new value AT ONCE — the number a + * caller compares `readAt` against to know a read was issued after this call. + */ +export function useReadGenerations(): [ReadonlyMap, (uri: string) => number] { + const [generations, setGenerations] = useState>(new Map()); + const generationsRef = useRef(generations); + const readAgain = useCallback((uri: string): number => { + const generation = (generationsRef.current.get(uri) ?? 0) + 1; + generationsRef.current = new Map(generationsRef.current).set(uri, generation); + setGenerations(generationsRef.current); + return generation; + }, []); + return [generations, readAgain]; +} + export function usePublicRecords(uris: readonly string[], options: PublicRecordsOptions = {}): PublicRecordsRead { - const { refresh = 0, retry = 0, tokenOf } = options; + const { refresh = NO_GENERATIONS, retry = 0, tokenOf } = options; const key = uris.join('\n'); + // The generations of the CURRENT URIs, as a string: the effect re-runs when + // one of them moves, not when the caller rebuilds an equal map. + const generationKey = key ? key.split('\n').map(uri => refresh.get(uri) ?? 0).join(',') : ''; const [landed, setLanded] = useState>(new Map()); // Read by the effect without re-running it: what already landed is not read // again when the set grows, and the token policy is the caller's current one. @@ -72,29 +114,71 @@ export function usePublicRecords(uris: readonly string[], options: PublicRecords landedRef.current = landed; const tokenOfRef = useRef(tokenOf); tokenOfRef.current = tokenOf; + const retryRef = useRef(retry); + retryRef.current = retry; + // The reads in flight, per URI, with the generation each was issued for. + const inFlight = useRef(new Map()); + // Moved when a read FAILS after a re-open that came while it was in flight: + // that failure is no answer to the re-open, so the effect reads again. + const [relaunch, setRelaunch] = useState(0); useEffect(() => { - if (!key) return; - const wanted = key.split('\n').filter(uri => needsRead(landedRef.current.get(uri), refresh, retry)); - if (wanted.length === 0) return; - const controller = new AbortController(); - void readEachRecord( - wanted, - (uri, record) => setLanded(prev => new Map(prev).set(uri, landedAfter(prev.get(uri), refresh, retry, record))), - { signal: controller.signal, tokenOf: uri => tokenOfRef.current?.(uri) }, - ); - return () => controller.abort(); - }, [key, refresh, retry]); + const all = key ? key.split('\n') : []; + const generations = generationKey ? generationKey.split(',').map(Number) : []; + const generationOf = new Map(all.map((uri, i) => [uri, generations[i] ?? 0])); + // A read in flight for a URI that left the set, or whose generation moved + // on, answers a question nobody asks any more. Every other read in flight + // is left alone: one URI's re-read never restarts another's. + for (const [uri, read] of inFlight.current) { + if (generationOf.get(uri) === read.generation) continue; + read.controller.abort(); + inFlight.current.delete(uri); + } + for (const uri of all) { + const generation = generationOf.get(uri) ?? 0; + if (inFlight.current.has(uri) || !needsRead(landedRef.current.get(uri), generation, retry)) continue; + const controller = new AbortController(); + inFlight.current.set(uri, { generation, controller }); + void fetchPublicRecord(uri, { signal: controller.signal, token: tokenOfRef.current?.(uri) }).then(record => { + if (controller.signal.aborted) return; + inFlight.current.delete(uri); + setLanded(prev => new Map(prev).set(uri, landedAfter(prev.get(uri), generation, retry, record))); + // A re-open while this read was in flight did not restart it (a slow + // read is not a failed one) — but if it fails, the re-open still wants + // its read: issue it now. + if (record === undefined && retryRef.current !== retry) setRelaunch(n => n + 1); + }); + } + }, [key, generationKey, retry, relaunch]); + useEffect(() => () => { + for (const read of inFlight.current.values()) read.controller.abort(); + inFlight.current.clear(); + }, []); const records = useMemo(() => { const out = new Map(); for (const [uri, entry] of landed) if (entry.record) out.set(uri, entry.record); return out; }, [landed]); + const readAt = useMemo(() => { + const out = new Map(); + for (const [uri, entry] of landed) if (entry.record && entry.recordRefresh !== undefined) out.set(uri, entry.recordRefresh); + return out; + }, [landed]); + const attemptedAt = useMemo(() => { + const out = new Map(); + for (const [uri, entry] of landed) out.set(uri, entry.refresh); + return out; + }, [landed]); const pending = useMemo(() => { const out = new Set(); - if (key) for (const uri of key.split('\n')) if (needsRead(landed.get(uri), refresh, retry)) out.add(uri); + if (key) { + const generations = generationKey.split(',').map(Number); + key.split('\n').forEach((uri, i) => { + if (needsRead(landed.get(uri), generations[i] ?? 0, retry)) out.add(uri); + }); + } return out; - }, [key, landed, refresh, retry]); - return { records, pending }; + }, [key, generationKey, landed, retry]); + return { records, readAt, attemptedAt, pending }; } diff --git a/src/shared/data/publicRead.test.ts b/src/shared/data/publicRead.test.ts index ab2fbd2..ff7a582 100644 --- a/src/shared/data/publicRead.test.ts +++ b/src/shared/data/publicRead.test.ts @@ -1,5 +1,5 @@ import { afterEach, describe, expect, spyOn, test } from 'bun:test'; -import { fetchPublicRecord, readEachRecord, tokenFor } from './publicRead'; +import { fetchPublicRecord, tokenFor } from './publicRead'; import { fetchActorProfileRecord } from './actor'; const original = globalThis.fetch; @@ -53,44 +53,23 @@ describe('fetchPublicRecord — anonymous, trusted for what was asked', () => { }); }); -describe('readEachRecord — per URI, as each lands', () => { - test('one unreadable URI costs that URI only; duplicates are read once; keyed by the requested URI', async () => { - const calls = install(url => (url === M ? json({}, 404) : json({ id: url, type: 'Event' }))); - const landed = new Map(); - await readEachRecord([E, M, E], (uri, record) => landed.set(uri, record)); - expect(landed.get(E)).toEqual({ id: E, type: 'Event' }); - expect(landed.has(M)).toBe(true); - expect(landed.get(M)).toBeUndefined(); - expect(calls.map(c => c.url).sort()).toEqual([E, M]); - }); - - test('a host that never answers withholds nothing else, and gives up at the timeout', async () => { - globalThis.fetch = ((url: string | URL | Request, init?: RequestInit) => { - const u = typeof url === 'string' ? url : url instanceof URL ? url.toString() : url.url; - if (u === EVIL) { - return new Promise((_, reject) => { - init?.signal?.addEventListener('abort', () => reject(new Error('aborted'))); - }); - } - return Promise.resolve(json({ id: u })); - }) as typeof fetch; - const order: string[] = []; +describe('fetchPublicRecord — bounded', () => { + test('a host that never answers gives up at the timeout', async () => { + globalThis.fetch = ((_url: string | URL | Request, init?: RequestInit) => + new Promise((_, reject) => { + init?.signal?.addEventListener('abort', () => reject(new Error('aborted'))); + })) as typeof fetch; const started = Date.now(); - await readEachRecord([EVIL, E], (uri, record) => order.push(`${uri}:${record ? 'read' : 'absent'}`), { timeoutMs: 150 }); - expect(order).toEqual([`${E}:read`, `${EVIL}:absent`]); + expect(await fetchPublicRecord(EVIL, { timeoutMs: 150 })).toBeUndefined(); expect(Date.now() - started).toBeLessThan(2000); }); - test('the token goes only where the caller grants it; nothing is reported after an abort', async () => { + test('an already-aborted read sends nothing', async () => { const calls = install(url => json({ id: url })); - await readEachRecord([E, EVIL], () => {}, { tokenOf: uri => tokenFor(uri, 'tok', 'http://localhost:3000') }); - expect(calls.map(c => [c.url, authOf(c)]).sort()).toEqual([[E, 'Bearer tok'], [EVIL, null]]); - const controller = new AbortController(); controller.abort(); - const reported: string[] = []; - await readEachRecord([E], uri => reported.push(uri), { signal: controller.signal }); - expect(reported).toEqual([]); + expect(await fetchPublicRecord(E, { signal: controller.signal })).toBeUndefined(); + expect(calls).toHaveLength(0); }); }); diff --git a/src/shared/data/publicRead.ts b/src/shared/data/publicRead.ts index 1395c1d..bcd8194 100644 --- a/src/shared/data/publicRead.ts +++ b/src/shared/data/publicRead.ts @@ -14,8 +14,8 @@ * already read, and covers a resource whose public grant did not land. * * BOUNDED. Each read gives up after `timeoutMs` (a host that never answers is - * an unreadable URI, not a stalled page), and `readEachRecord` reports every - * URI as it lands — one slow host never withholds the others. + * an unreadable URI, not a stalled page); the caller reads each URI on its own + * (`context/usePublicRecords.ts`), so one slow host never withholds the others. * * TRUSTED FOR WHAT WAS ASKED, NOT FOR WHAT IT CLAIMS. A record is kept only * when its own identifier (`id`/`@id`) is the URI that was requested, and the @@ -86,24 +86,6 @@ export async function fetchPublicRecord(uri: string, options: PublicReadOptions } } -/** - * Read each URI independently and report it AS IT LANDS (`onRead(uri, record)`, - * `record` undefined when unreadable) — keyed by the REQUESTED URI, duplicates - * read once. Nothing is reported after `signal` aborts. Resolves once every - * URI has been reported (or the signal aborted). - */ -export async function readEachRecord( - uris: readonly string[], - onRead: (uri: string, record: JsonRecord | undefined) => void, - options: { signal?: AbortSignal; tokenOf?: (uri: string) => string | undefined; timeoutMs?: number } = {}, -): Promise { - const { signal, tokenOf, timeoutMs } = options; - await Promise.all([...new Set(uris)].map(async uri => { - const record = await fetchPublicRecord(uri, { signal, token: tokenOf?.(uri), timeoutMs }); - if (!signal?.aborted) onRead(uri, record); - })); -} - /** * The `Authorization` the session may attach to a request for `uri`: its * bearer token, but ONLY on the origin that issued the session (`trustedOrigin`,