diff --git a/packages/core/src/materializer.ts b/packages/core/src/materializer.ts index 15f1841..e775f12 100644 --- a/packages/core/src/materializer.ts +++ b/packages/core/src/materializer.ts @@ -47,8 +47,9 @@ import { validateRecord } from './validation.js' // CLAIMS are the one exception, and only because the rule that reads them is observer-local too: a // lease is a duration measured from when THIS observer first saw the version (`claim-lease.ts`), so // a claim in the index carries the stamp that decides its own liveness and the UI can show the -// deadline the fold actually applied. Nothing else does, and nothing convergent — the digest -// included — reads it. +// deadline the fold actually applied. `MaterializedIndex.staleAt` is the other surface of that same +// exception: it is derived from those stamps solely to say when this observer must refold. Nothing +// else does, and nothing convergent — the digest included — reads either value. export type IndexedRecord = Omit< StoredRecord, 'rev' | 'firstSeenAt' @@ -233,7 +234,7 @@ export interface MaterializedIndex { * already written stay blessed: a member vouched for those bytes, and a toggle is not a retraction. */ guestCommentsEnabled: boolean - /** Earliest instant the same records may fold differently because a live claim expires. */ + /** Observer-local earliest claim expiry; excluded from convergence comparisons and the digest. */ staleAt?: string edits: EditAnnotation[] ignored: IgnoredRecord[] @@ -1071,7 +1072,13 @@ export function materialize(store: RecordStore, options: MaterializeOptions): Ma }) const staleAt = claims .filter((claim) => claimLive(claim.value, asOf, claim.firstSeenAt)) - .map((claim) => claimDeadline(claim.value, claim.firstSeenAt) ?? claim.value.expiresAt) + .map((claim) => { + const deadline = claimDeadline(claim.value, claim.firstSeenAt) + if (deadline) return deadline + // The legacy fallback is live at expiresAt itself; refold one millisecond later so the cache + // boundary agrees with claimLive's inclusive comparison. + return new Date(Date.parse(claim.value.expiresAt) + 1).toISOString() + }) .filter((deadline) => Number.isFinite(Date.parse(deadline)) && Date.parse(asOf) < Date.parse(deadline)) .sort((left, right) => Date.parse(left) - Date.parse(right))[0] const targetUriForRequest = (request: IndexedRecord): string => diff --git a/packages/core/src/node.ts b/packages/core/src/node.ts index 11d9b78..a169259 100644 --- a/packages/core/src/node.ts +++ b/packages/core/src/node.ts @@ -46,6 +46,11 @@ import type { WantList } from './private/quarantine.js' export class SqliteRecordStore implements RecordStore { readonly #database: DatabaseSync + #revision = 0 + + get revision(): number { + return this.#revision + } /** * This store's connection, for the one other class that must write in the SAME transaction as a @@ -158,6 +163,9 @@ export class SqliteRecordStore implements RecordStore { `) .run(record.did, record.collection, record.rkey, record.cid, deviceKeyId) } + // SQLite may have inserted a version or reconciled earlier observer-local provenance. Bumping + // conservatively on a duplicate only costs a fold; failing to bump could serve a stale one. + this.#revision += 1 } #versions(): StoredRecord[] { diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index ec83738..9d8bdf7 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -55,6 +55,8 @@ export interface RepoSnapshot { } export interface RecordStore { + /** Monotonic observer-local generation, bumped whenever `put` teaches the store anything. */ + readonly revision: number put(record: StoredRecord): void get(did: string, collection: string, rkey: string): StoredRecord | undefined getSnapshot(did: string): RepoSnapshot | undefined @@ -297,9 +299,15 @@ export function collapse(versions: StoredRecord[]): { records: StoredRecord[]; e export class MemoryRecordStore implements RecordStore { readonly #versions: StoredRecord[] = [] readonly #snapshots = new Map() + #revision = 0 + + get revision(): number { + return this.#revision + } put(record: StoredRecord): void { this.#versions.push(structuredClone(record)) + this.#revision += 1 } records(): StoredRecord[] { diff --git a/packages/ingest/src/poller.ts b/packages/ingest/src/poller.ts index e420ab2..7e91abf 100644 --- a/packages/ingest/src/poller.ts +++ b/packages/ingest/src/poller.ts @@ -99,7 +99,8 @@ export class RepoPoller { const held = new Set(described.collections) collections = this.collections.filter((collection) => held.has(collection)) } catch { - // describeRepo is an optimization only. Older or incomplete PDSes get the full sweep. + // describeRepo is an optimization only. Errors get the full sweep. A successful response + // is trusted to describe that stable repo head completely, as required by the XRPC. } } const stagedByCollection: StoredRecord[][] = collections.map(() => []) diff --git a/packages/ingest/src/space.ts b/packages/ingest/src/space.ts index b409f12..1783f49 100644 --- a/packages/ingest/src/space.ts +++ b/packages/ingest/src/space.ts @@ -6,12 +6,14 @@ export class SpaceIngestor { readonly spaceDid: string readonly knownDids = new Set() #lastIndex: MaterializedIndex | undefined + #lastRevision = -1 constructor( readonly spaceUri: string, readonly poller: RepoPoller, readonly records: RecordStore, readonly now: () => string = () => new Date().toISOString(), + readonly memberConcurrency: number = 6, ) { this.spaceDid = parseAtUri(spaceUri).did this.knownDids.add(this.spaceDid) @@ -19,11 +21,12 @@ export class SpaceIngestor { async sync(signal?: AbortSignal): Promise { const initialDids = [...this.knownDids].sort(compareCodePoints) - const initial = await Promise.all(initialDids.map((did) => this.poller.pollDid(did, signal))) + const initial = await this.#pollDids(initialDids, signal) const polled = new Set(initialDids) const now = this.now() if ( this.#lastIndex && + this.records.revision === this.#lastRevision && initial.every((result) => !result.changed) && (!this.#lastIndex.staleAt || Date.parse(now) < Date.parse(this.#lastIndex.staleAt)) ) { @@ -43,15 +46,29 @@ export class SpaceIngestor { this.knownDids.add(this.spaceDid) for (const member of index.members) if (member.active) this.knownDids.add(member.did) this.#lastIndex = index + this.#lastRevision = this.records.revision return index } for (const member of pending) { polled.add(member.did) this.knownDids.add(member.did) } - await Promise.all(pending.map((member) => this.poller.pollDid(member.did, signal))) + await this.#pollDids(pending.map((member) => member.did), signal) } } + + async #pollDids(dids: string[], signal?: AbortSignal): Promise>[]> { + const results: Awaited>[] = new Array(dids.length) + let next = 0 + const poll = async (): Promise => { + const index = next++ + if (index >= dids.length) return + results[index] = await this.poller.pollDid(dids[index]!, signal) + await poll() + } + await Promise.all(Array.from({ length: Math.min(Math.max(1, this.memberConcurrency), dids.length) }, poll)) + return results + } } export function startPolling( diff --git a/packages/ingest/test/space.test.mjs b/packages/ingest/test/space.test.mjs index 669fc69..8cd2ae0 100644 --- a/packages/ingest/test/space.test.mjs +++ b/packages/ingest/test/space.test.mjs @@ -143,6 +143,49 @@ it('returns the cached index when every known repository is unchanged', async () assert.equal(second, first) }) +it('refolds after a direct store write even when every repository poll is unchanged', async () => { + const root = 'did:plc:root' + const store = new MemoryRecordStore() + const space = { + did: root, collection: COLLECTIONS.space, rkey: 'space', + uri: `at://${root}/${COLLECTIONS.space}/space`, cid: 'cid-space', rev: '0001', + value: { $type: COLLECTIONS.space, name: 'Before', description: 'test', createdAt: '2026-07-18T00:00:00Z' }, + } + store.put(space) + const poller = { + async pollDid(did) { + return { did, head: { rev: '0001', commitCid: 'head' }, changed: false, records: 0, rejected: [] } + }, + } + const ingestor = new SpaceIngestor(space.uri, poller, store) + const first = await ingestor.sync() + store.put({ ...space, rkey: 'other', uri: `at://${root}/${COLLECTIONS.space}/other`, cid: 'cid-other' }) + assert.notEqual(await ingestor.sync(), first) +}) + +it('bounds concurrent member polls', async () => { + const root = 'did:plc:root' + const store = new MemoryRecordStore() + store.put({ + did: root, collection: COLLECTIONS.space, rkey: 'space', + uri: `at://${root}/${COLLECTIONS.space}/space`, cid: 'cid-space', rev: '0001', + value: { $type: COLLECTIONS.space, name: 'Space', description: 'test', createdAt: '2026-07-18T00:00:00Z' }, + }) + let active = 0 + let peak = 0 + const poller = { async pollDid(did) { + active += 1 + peak = Math.max(peak, active) + await Promise.resolve() + active -= 1 + return { did, head: { rev: '0001', commitCid: 'head' }, changed: false, records: 0, rejected: [] } + } } + const ingestor = new SpaceIngestor(`at://${root}/${COLLECTIONS.space}/space`, poller, store, undefined, 2) + for (const did of ['did:plc:a', 'did:plc:b', 'did:plc:c', 'did:plc:d']) ingestor.knownDids.add(did) + await ingestor.sync() + assert.equal(peak, 2) +}) + it('refolds unchanged records after the cached claim deadline', async () => { const root = 'did:plc:root' const store = new MemoryRecordStore() diff --git a/packages/ui/src/lib/session.svelte.ts b/packages/ui/src/lib/session.svelte.ts index 1b0dc9e..9ff0ec2 100644 --- a/packages/ui/src/lib/session.svelte.ts +++ b/packages/ui/src/lib/session.svelte.ts @@ -186,6 +186,8 @@ interface Live { timer: ReturnType | undefined syncing: boolean stopped: boolean + /** Raw index most recently handed to Svelte; nested `$state` reads return a proxy instead. */ + published?: MaterializedIndex } let live: Live | undefined @@ -699,6 +701,7 @@ export function leave(): void { } function publish(index: MaterializedIndex, current = live): void { + if (current) current.published = index const uri = current?.uri ?? index.space.uri session.space = { uri, @@ -752,7 +755,7 @@ export async function sync(): Promise { if (current.waiting) return await catchUpWaiting(current, undialable) const index = await current.ingestor.sync() if (current.stopped) return - if (session.space?.index === index) { + if (current.published === index && session.space) { const now = new Date().toISOString() session.space.asOf = now session.space.syncedAt = now diff --git a/packages/ui/src/lib/session.test.ts b/packages/ui/src/lib/session.test.ts index 9e4f269..c2aa5bb 100644 --- a/packages/ui/src/lib/session.test.ts +++ b/packages/ui/src/lib/session.test.ts @@ -80,6 +80,19 @@ describe('opening a live space', () => { expect(live.network.counts('listRecords')).toBeLessThan(members * 26) }) + it('preserves the published Space on an unchanged tick while advancing its clocks', async () => { + const live = harness() + await open(live) + const published = session.space + if (!published) throw new Error('no space') + published.asOf = '2000-01-01T00:00:00.000Z' + published.syncedAt = published.asOf + await sync() + expect(session.space).toBe(published) + expect(session.space?.asOf).not.toBe('2000-01-01T00:00:00.000Z') + expect(session.space?.syncedAt).toBe(session.space?.asOf) + }) + it('publishes and remembers the latest distinct space', async () => { const uriB = `${fixture.spaceUri}-b` const spaceA = harness()