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