Keep a join or a leave on screen until a read taken after it confirms it
After leaving a meeting point, the view could stay at "1 inscrit" while the collection was already empty (caught by the rights matrix, 1 run in 5): a read issued before the act was taken as confirmation, the overlay was dropped, and nothing re-read afterwards. Now only a read issued after the act confirms it; while unconfirmed, only the collection the act changed is re-read (immediately, then at 1.5, 3, 6, 12, 20 and 30 s, the last read wins), and a pre-act read never undoes it. The join direction had the mirror race. Foreign records are read one URI at a time.
This commit is contained in:
@@ -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<string, unknown>[],
|
||||
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<Response> {
|
||||
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<typeof createRoot> | undefined;
|
||||
|
||||
beforeAll(async () => {
|
||||
const window = new Window({ url: 'http://localhost:4000/' });
|
||||
const g = globalThis as Record<string, unknown>;
|
||||
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<string, unknown>)[name];
|
||||
});
|
||||
|
||||
// --- helpers -----------------------------------------------------------------------
|
||||
async function until(condition: () => boolean, what: string, timeoutMs = 10000): Promise<void> {
|
||||
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<typeof view>) => boolean, timeoutMs: number): Promise<ReturnType<typeof view>> {
|
||||
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<string, unknown>[] = [], 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(<RefineDataProvider><Probe /></RefineDataProvider>);
|
||||
}
|
||||
|
||||
async function mount(members: string[]): Promise<void> {
|
||||
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<number> {
|
||||
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);
|
||||
@@ -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<string>();
|
||||
// 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<string, string>(); // 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<MembershipOverlay>(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<PendingActs>(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<ReturnType<typeof setTimeout> | 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<string, ReturnType<typeof setTimeout>[]>());
|
||||
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<FpEventData, 'id'>): Promise<FpEventData> => {
|
||||
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) => {
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
|
||||
@@ -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<string, PendingAct>;
|
||||
|
||||
/**
|
||||
* 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<string, PendingAct> | 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;
|
||||
}
|
||||
|
||||
@@ -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<Response> {
|
||||
const uri = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url;
|
||||
return new Promise<Response>((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<typeof createRoot> | 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<string, unknown>;
|
||||
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<string, unknown>)[name];
|
||||
});
|
||||
|
||||
const tick = () => new Promise(resolve => setTimeout(resolve, 30));
|
||||
|
||||
async function show(uris: readonly string[], options: PublicRecordsOptions = {}): Promise<void> {
|
||||
if (!root) root = createRoot(document.createElement('div') as unknown as Element);
|
||||
root.render(<Probe uris={uris} options={options} />);
|
||||
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();
|
||||
});
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -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<string, JsonRecord>;
|
||||
/** Per URI, the generation of the read that produced the record in `records`. */
|
||||
readAt: ReadonlyMap<string, number>;
|
||||
/** Per URI, the generation of the last read that landed, successful or failed. */
|
||||
attemptedAt: ReadonlyMap<string, number>;
|
||||
pending: ReadonlySet<string>;
|
||||
}
|
||||
|
||||
export interface PublicRecordsOptions {
|
||||
refresh?: number;
|
||||
/** Per-URI read generation (absent = 0): moving it re-reads that URI. */
|
||||
refresh?: ReadonlyMap<string, number>;
|
||||
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<string, number> = 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<string, number>, (uri: string) => number] {
|
||||
const [generations, setGenerations] = useState<ReadonlyMap<string, number>>(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<ReadonlyMap<string, Landed>>(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<string, { generation: number; controller: AbortController }>());
|
||||
// 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<string, JsonRecord>();
|
||||
for (const [uri, entry] of landed) if (entry.record) out.set(uri, entry.record);
|
||||
return out;
|
||||
}, [landed]);
|
||||
const readAt = useMemo(() => {
|
||||
const out = new Map<string, number>();
|
||||
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<string, number>();
|
||||
for (const [uri, entry] of landed) out.set(uri, entry.refresh);
|
||||
return out;
|
||||
}, [landed]);
|
||||
const pending = useMemo(() => {
|
||||
const out = new Set<string>();
|
||||
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 };
|
||||
}
|
||||
|
||||
@@ -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<string, unknown>();
|
||||
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<Response>((_, 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<Response>((_, 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);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -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<void> {
|
||||
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`,
|
||||
|
||||
Reference in New Issue
Block a user