diff --git a/doc/offline-query-membership.md b/doc/offline-query-membership.md new file mode 100644 index 00000000000..32abbf11823 --- /dev/null +++ b/doc/offline-query-membership.md @@ -0,0 +1,140 @@ +# Offline query membership and stale snapshot reconciliation + +This document explains a subtle class of bugs in the frontend realtime layer (`xforge-common`), +and how `RealtimeQuery`/`RealtimeDoc` guard against it. It exists because the complexity is easy +to lose in scattered comments; see SF-3893 for the bug that motivated it. + +## Background: three copies of every doc + +For each realtime doc the client can hold up to three copies: + +1. **Server** (ShareDB/Mongo) — authoritative. +2. **In-memory doc** (`RealtimeDoc`/ShareDB client doc) — kept up to date by ops _only while the + client is subscribed to the doc, or the doc is in the results of a subscribed query_. +3. **Offline store** (IndexedDB) — written by `RealtimeDoc.updateOfflineData()`, used to render + before the server responds and to work offline. + +A subscribed `RealtimeQuery` gets its membership (which docs are in the results) from two +sources: + +- **Remote**: the ShareDB query subscription pushes diffs when the server's result set changes. +- **Local**: `RealtimeQuery.localQuery()` re-runs the query's Mongo-style filter against the + offline store (via mingo). This happens on initial subscribe (offline-first render) and — via + `RealtimeService.onLocalDocUpdate()` — on **every local submit to any doc in the collection**, + so that local edits are reflected in query results immediately and while offline. + +## The problem: snapshots nobody will ever correct + +The server only pushes ops for docs the client is subscribed to (directly, or via the doc being +in a subscribed query's current results). Consequently, **a doc that changes on the server while +this client is not listening ends up with a stale offline snapshot, and no invalidation for it +will ever arrive.** Typical causes: the page was closed or reloaded between the change and the +next look, or a websocket reconnect window. + +Nothing in the doc lifecycle repairs this on its own: + +- `updateOfflineData()` early-returns when the adapter's version equals the version already + stored offline — which is exactly the case when the doc was loaded _from_ the stale snapshot. +- `checkExists()` only answers "was it deleted?", which handles server-side deletions (the + offline entry is purged) but not modifications. +- Docs are never disposed during normal navigation, so the stale entry survives indefinitely. + +The failure mode (SF-3893): a checker's client had an archived question cached offline with +`isArchived: false`. The live query correctly showed the server's results — until the checker +answered a _different_ question. That local submit triggered `localQuery()`, the stale snapshot +matched the `isArchived: false` filter, and the archived question was spliced back into the live +results. The server never corrects this, because _its_ result set did not change. The same +mechanism works in the opposite direction (a doc whose stale snapshot wrongly fails the filter +vanishes from results on any unrelated local write). + +Note that per-consumer defensive filtering (e.g. `.filter(q => !q.data.isArchived)` in a +component) does **not** fix this: the resurrected doc's in-memory data comes from the same stale +snapshot, so the filter passes it. + +## The solution + +Two cooperating mechanisms. The principle: **while the remote query subscription is live, the +server's membership is authoritative; local results may only diverge from it for docs with +pending (unacknowledged) local ops. Any other disagreement proves an offline snapshot is stale +and triggers its repair.** + +### 1. Membership gate — `RealtimeQuery.reconcileWithRemote()` + +When `localQuery()` runs while the remote query is live (`adapter.ready && adapter.subscribed`), +its results are merged with the server's current results. Per doc: + +| server includes | offline matches filter | pending local ops | in results? | notes | +| --------------- | ---------------------- | ----------------- | ----------- | --------------------------------------------------- | +| yes | yes | — | yes | agreement | +| no | no | — | no | agreement | +| no | yes | yes | yes | optimistic add: client just created/changed it | +| no | yes | no | no | stale snapshot → reconcile (SF-3893) | +| yes | no | yes | no | optimistic removal: client just archived/changed it | +| yes | no | no | yes | stale snapshot (inverse direction) → reconcile | + +- "Pending local ops" (`RealtimeDoc.hasPendingOps`) includes the in-flight op, so it is `true` + at the moment `submit()` triggers the local re-query, before the server acknowledges. +- Paged queries (`$skip`/`$limit`): two differently-paged result sets cannot be meaningfully + merged, so the server's page is used as-is while live. +- Row 6 appends docs out of sort order; the order self-corrects once reconciliation refreshes + the offline snapshot. +- Before the remote query is ready (initial load, fully offline), local results are used + unchanged — offline-first behavior is unaffected, and all offline mutations carry pending ops, + so they survive the gate after reconnecting. + +### 1b. Serialized change application — `RealtimeQuery.onChange()` + +Implementing the gate surfaced a latent race: `onChange()` diffs the new result ids against the +current results and applies the diff with index-based splices, but it is async (inserting docs +awaits their offline data loading). Two overlapping invocations — e.g. a server-driven change +interleaving with a local re-query at an await point — each capture a `before` snapshot and can +splice against state the other has already changed, duplicating or misplacing docs. Changes are +now applied strictly one at a time via an internal lock. When no change is in flight, a change +still starts synchronously, preserving the previous timing in the common case. + +(Implementation note: the lock is deliberately written with async/await only. ts-mockito +discovers mockable method names by scanning the class _source text_, so a call to a promise's +"then" method anywhere in the `RealtimeQuery` source would make every mocked `RealtimeQuery` +instance a thenable that never settles when awaited or passed to `Promise.resolve()` in tests.) + +### 2. Snapshot repair — `RealtimeDoc.reconcileOfflineData()` + +Fired for the two "stale snapshot" rows above, and also when the server removes a doc from a +subscribed query's results while the doc's local data still matches the query filter (which +proves staleness without waiting for the next local write — this repairs SF-3893-style staleness +at load time, before the user does anything). One server fetch resolves all cases: + +- doc still exists → rewrite the offline snapshot (forced, since the doc may no longer be in any + subscribed query); +- doc deleted → purge the offline entry and emit `delete$` (a generalization of `checkExists()`); +- fetch fails (offline, or the user may no longer read the doc) → leave the offline copy; the + membership gate already excludes the doc, so nothing incorrect is shown. + +Repair is self-limiting (once the snapshot agrees with the server the trigger disappears) and +concurrent triggers share one round trip. + +## Known limitations + +- **Initial flash**: with a stale offline store, a since-archived doc can appear briefly on load + until the remote query becomes ready. Fixing this would mean not rendering offline results — + a product tradeoff. Repair at least limits it to one stale session per doc. +- **Ack window**: after an op is acknowledged but before the server's query diff arrives, an + unrelated local write can briefly drop a just-added doc; the diff restores it moments later. +- **Permission revocation**: `sharedb-access` rejects whole read requests with a 403 when any + snapshot is unreadable, so a no-longer-readable doc looks like an error, not like "gone". Its + offline data is _not_ purged (only logout's `deleteDB()` clears it), and project-level + removal is handled at the application layer (components navigate away). A deliberate purge of + unreadable docs' offline data is possible future work. +- **No offline sweep**: docs that never pass through a subscribed query again (e.g. a whole + project the user lost access to) keep their offline entries until logout. + +## Tests + +- `xforge-common/models/realtime-query.spec.ts` — the decision table, including the SF-3893 + regression (resurrection on unrelated local write) and the proactive repair on remote removal. +- `xforge-common/models/realtime-doc.spec.ts` — `reconcileOfflineData()` semantics + (update/purge/leave-on-error/deduplication). +- The memory test doubles (`memory-realtime-remote-store.ts`) model an _instantly consistent_ + server: local submits are written back to the remote store and query adapters re-query it on + access. Without this, tests would exercise a state (acknowledged op, stale server result set) + that the real ShareDB adapters only pass through transiently. diff --git a/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/memory-realtime-remote-store.ts b/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/memory-realtime-remote-store.ts index c8e20d6b22f..d3b47136afd 100644 --- a/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/memory-realtime-remote-store.ts +++ b/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/memory-realtime-remote-store.ts @@ -34,6 +34,10 @@ export class MemoryRealtimeRemoteStore extends RealtimeRemoteStore { return collectionSnapshots.values(); } + removeSnapshot(collection: string, id: string): void { + this.snapshots.get(collection)?.delete(id); + } + clear(): void { this.snapshots.clear(); } @@ -45,9 +49,9 @@ export class MemoryRealtimeRemoteStore extends RealtimeRemoteStore { snapshot = collectionSnapshots.get(id); } if (snapshot == null) { - return new MemoryRealtimeDocAdapter(collection, id); + return new MemoryRealtimeDocAdapter(collection, id, undefined, undefined, undefined, this); } - return new MemoryRealtimeDocAdapter(collection, id, snapshot.data, types.map[snapshot.type], snapshot.v); + return new MemoryRealtimeDocAdapter(collection, id, snapshot.data, types.map[snapshot.type], snapshot.v, this); } createQueryAdapter(collection: string, parameters: QueryParameters): RealtimeQueryAdapter { @@ -101,7 +105,8 @@ export class MemoryRealtimeDocAdapter implements RealtimeDocAdapter { public readonly id: string, public data?: any, public type: OTType | undefined = OTJson0.type, - version?: number + version?: number, + private readonly remoteStore?: MemoryRealtimeRemoteStore ) { if (version != null) { this.version = version; @@ -115,6 +120,7 @@ export class MemoryRealtimeDocAdapter implements RealtimeDocAdapter { this.data = data; this.type = types.map[type]; this.version = 0; + this.syncSnapshotToStore(); this.emitCreate(); return Promise.resolve(); } @@ -143,6 +149,7 @@ export class MemoryRealtimeDocAdapter implements RealtimeDocAdapter { } this.data = this.type.apply(this.data, op); this.version++; + this.syncSnapshotToStore(); this.emitChange(op); if (!source) { this.emitRemoteChange(op); @@ -162,6 +169,7 @@ export class MemoryRealtimeDocAdapter implements RealtimeDocAdapter { this.data = undefined; this.version = -1; this.type = undefined; + this.remoteStore?.removeSnapshot(this.collection, this.id); this.emitDelete(); return Promise.resolve(); } @@ -174,6 +182,23 @@ export class MemoryRealtimeDocAdapter implements RealtimeDocAdapter { return Promise.resolve(); } + /** + * Keeps the remote store's snapshot in sync with this adapter, so that the memory + * implementation behaves like an instantly-consistent server: local submits are immediately + * reflected in query results (MemoryRealtimeQueryAdapter re-queries the store on access). + */ + private syncSnapshotToStore(): void { + if (this.remoteStore == null || this.type == null) { + return; + } + this.remoteStore.addSnapshot(this.collection, { + id: this.id, + data: this.data, + v: this.version, + type: this.type.name + }); + } + emitChange(op?: any): void { this.changes$.next(op); } @@ -194,62 +219,77 @@ export class MemoryRealtimeDocAdapter implements RealtimeDocAdapter { export class MemoryRealtimeQueryAdapter implements RealtimeQueryAdapter { subscribed: boolean = false; ready: boolean = true; - unpagedCount: number = 0; - docIds: string[] = []; - count: number = 0; readonly ready$ = new Subject(); readonly remoteChanges$ = new Subject(); + private lastDocIds: string[] = []; + private lastCount: number = 0; + constructor( private readonly remoteStore: MemoryRealtimeRemoteStore, public readonly collection: string, public readonly parameters: QueryParameters ) {} + // The results are re-queried from the remote store on every access, so that the memory + // implementation behaves like an instantly-consistent server (there is no notion of an + // in-flight op or a not-yet-polled query subscription, as there is with the ShareDB adapters). + get docIds(): string[] { + return this.performQuery().docIds; + } + + get count(): number { + return this.performQuery().count; + } + + get unpagedCount(): number { + return this.performQuery().unpagedCount; + } + fetch(): Promise { - this.performQuery(); + this.rememberResults(); this.ready = true; this.ready$.next(); return Promise.resolve(); } subscribe(_initialDocIds?: string[]): void { - this.performQuery(); + this.rememberResults(); this.subscribed = true; this.ready = true; this.ready$.next(); } updateResults(): void { - if (this.performQuery()) { + if (this.rememberResults()) { this.remoteChanges$.next(); } } destroy(): void {} - private performQuery(): boolean { - let changed = false; + /** Re-queries the results and reports whether they changed since the last remembered results. */ + private rememberResults(): boolean { + const { docIds, count } = this.performQuery(); + const changed: boolean = !isEqual(this.lastDocIds, docIds) || this.lastCount !== count; + this.lastDocIds = docIds; + this.lastCount = count; + return changed; + } + + private performQuery(): { docIds: string[]; count: number; unpagedCount: number } { const snapshots = Array.from(this.remoteStore.getSnapshots(this.collection)); const { results, unpagedCount } = performQuery(this.parameters, snapshots); + let docIds: string[]; let count: number; if (results instanceof Array) { - const before = this.docIds; - const after = results.map(s => s.id); - this.docIds = after; - if (!isEqual(before, after)) { - changed = true; - } + docIds = results.map(s => s.id); count = results.length; } else { + docIds = []; count = results; } - if (this.count !== count) { - this.count = count; - changed = true; - } - this.unpagedCount = unpagedCount; - return changed; + return { docIds: docIds, count: count, unpagedCount: unpagedCount }; } } diff --git a/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-doc.spec.ts b/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-doc.spec.ts new file mode 100644 index 00000000000..31842574c54 --- /dev/null +++ b/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-doc.spec.ts @@ -0,0 +1,136 @@ +import { fakeAsync, TestBed, tick } from '@angular/core/testing'; +import { getQuestionDocId, Question } from 'realtime-server/lib/esm/scriptureforge/models/question'; +import { QuestionDoc } from '../../app/core/models/question-doc'; +import { MemoryOfflineStore } from '../memory-offline-store'; +import { MemoryRealtimeDocAdapter } from '../memory-realtime-remote-store'; +import { provideTestRealtime } from '../test-realtime-providers'; +import { TestRealtimeService } from '../test-realtime.service'; +import { configureTestingModule } from '../test-utils'; +import { TypeRegistry } from '../type-registry'; +import { FileType } from './file-offline-data'; +import { RealtimeOfflineData } from './realtime-offline-data'; + +describe('RealtimeDoc', () => { + configureTestingModule(() => ({ + providers: [provideTestRealtime(new TypeRegistry([QuestionDoc], [FileType.Audio], []))] + })); + + describe('reconcileOfflineData', () => { + it('should update the offline copy from the server', fakeAsync(() => { + const env = new TestEnvironment(); + env.addRemoteQuestion(true, 2); + env.addOfflineQuestion(false, 1); // stale: archived on the server while this client was away + const doc: QuestionDoc = env.getQuestionDoc(); + + void doc.reconcileOfflineData(); + tick(); + + const offlineData: RealtimeOfflineData | undefined = env.getOfflineData(); + expect(offlineData?.data.isArchived).toBe(true); + expect(offlineData?.v).toBe(2); + })); + + it('should remove the offline copy when the doc was deleted on the server', fakeAsync(() => { + const env = new TestEnvironment(); + env.addRemoteQuestion(false, 1); + env.addOfflineQuestion(false, 1); + const doc: QuestionDoc = env.getQuestionDoc(); + let deleted: boolean = false; + doc.delete$.subscribe(() => (deleted = true)); + env.simulateDeletedOnServer(doc); + + void doc.reconcileOfflineData(); + tick(); + + expect(env.getOfflineData()).toBeUndefined(); + expect(deleted).toBe(true); + })); + + it('should leave the offline copy as-is when the fetch fails', fakeAsync(() => { + const env = new TestEnvironment(); + env.addRemoteQuestion(true, 2); + env.addOfflineQuestion(false, 1); + const doc: QuestionDoc = env.getQuestionDoc(); + spyOn(doc.adapter, 'fetch').and.returnValue(Promise.reject(new Error('offline'))); + + void doc.reconcileOfflineData(); + tick(); + + const offlineData: RealtimeOfflineData | undefined = env.getOfflineData(); + expect(offlineData?.data.isArchived).toBe(false); + expect(offlineData?.v).toBe(1); + })); + + it('should share a single fetch between concurrent calls', fakeAsync(() => { + const env = new TestEnvironment(); + env.addRemoteQuestion(true, 2); + env.addOfflineQuestion(false, 1); + const doc: QuestionDoc = env.getQuestionDoc(); + const fetchSpy: jasmine.Spy = spyOn(doc.adapter, 'fetch').and.callThrough(); + + void doc.reconcileOfflineData(); + void doc.reconcileOfflineData(); + tick(); + + expect(fetchSpy).toHaveBeenCalledTimes(1); + })); + }); +}); + +class TestEnvironment { + static readonly projectId: string = 'project1'; + static readonly questionId: string = getQuestionDocId(TestEnvironment.projectId, 'q1'); + + readonly realtimeService: TestRealtimeService = TestBed.inject(TestRealtimeService); + + addRemoteQuestion(isArchived: boolean, v: number): void { + this.realtimeService.addSnapshot(QuestionDoc.COLLECTION, { + id: TestEnvironment.questionId, + data: this.questionData(isArchived), + v: v + }); + } + + addOfflineQuestion(isArchived: boolean, v: number): void { + const offlineStore = this.realtimeService.offlineStore as MemoryOfflineStore; + offlineStore.addData(QuestionDoc.COLLECTION, { + id: TestEnvironment.questionId, + v: v, + data: this.questionData(isArchived), + pendingOps: [] + } as any); + } + + getQuestionDoc(): QuestionDoc { + return this.realtimeService.get(QuestionDoc.COLLECTION, TestEnvironment.questionId); + } + + getOfflineData(): RealtimeOfflineData | undefined { + const offlineStore = this.realtimeService.offlineStore as MemoryOfflineStore; + return offlineStore.getData(QuestionDoc.COLLECTION, TestEnvironment.questionId); + } + + simulateDeletedOnServer(doc: QuestionDoc): void { + const adapter = doc.adapter as MemoryRealtimeDocAdapter; + spyOn(adapter, 'fetch').and.callFake(() => { + adapter.data = undefined; + adapter.type = undefined; + return Promise.resolve(); + }); + } + + private questionData(isArchived: boolean): Question { + const date = new Date(2026, 0, 1).toISOString(); + return { + dataId: 'q1', + projectRef: TestEnvironment.projectId, + ownerRef: 'user1', + verseRef: { bookNum: 1, chapterNum: 1, verseNum: 1 }, + text: 'Question 1', + isArchived: isArchived, + dateCreated: date, + dateModified: date, + answers: [] + }; + } +} diff --git a/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-doc.ts b/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-doc.ts index 2e92104f3fc..0d11e176669 100644 --- a/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-doc.ts +++ b/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-doc.ts @@ -33,6 +33,7 @@ export abstract class RealtimeDoc { private subscribedState: boolean = false; private subscribeQueryCount: number = 0; private loadOfflineDataPromise?: Promise; + private reconcileOfflineDataPromise?: Promise; constructor( protected readonly realtimeService: RealtimeService, @@ -70,6 +71,14 @@ export abstract class RealtimeDoc { return this.subscribeQueryCount; } + /** + * Whether this doc has local changes that the server has not yet acknowledged. This includes an + * op that is currently in flight. + */ + get hasPendingOps(): boolean { + return this.adapter.pendingOps.length > 0; + } + get collection(): string { return this.adapter.collection; } @@ -170,6 +179,19 @@ export abstract class RealtimeDoc { return this.adapter.previousSnapshot(); } + /** + * Re-reads this doc from the server and updates (or removes, if the doc was deleted) its copy in + * the offline store. This is used when the doc is found to have changed on the server while this + * client was not subscribed to it, which the client is not otherwise notified about — see + * doc/offline-query-membership.md. Concurrent calls share a single server round trip. + */ + reconcileOfflineData(): Promise { + this.reconcileOfflineDataPromise ??= this.reconcileWithServer().finally( + () => (this.reconcileOfflineDataPromise = undefined) + ); + return this.reconcileOfflineDataPromise; + } + /** * Unsubscribes and destroys this realtime data model. * @@ -271,6 +293,24 @@ export abstract class RealtimeDoc { await this.onSubscribe(); } + private async reconcileWithServer(): Promise { + try { + await this.adapter.fetch(); + } catch { + // The fetch failed (e.g. offline, or this user is no longer permitted to read the doc). + // Leave the offline copy as-is; query membership already excludes the doc. + return; + } + if (this.adapter.type == null) { + // The doc no longer exists on the server. + await this.onDelete(); + this.localDelete$.next(); + } else { + // Force the update, since the doc may no longer be in any subscribed query. + await this.updateOfflineData(true); + } + } + private async checkExists(): Promise { if (!(await this.adapter.exists())) { void this.onDelete(); diff --git a/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-query.spec.ts b/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-query.spec.ts new file mode 100644 index 00000000000..924efb06fec --- /dev/null +++ b/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-query.spec.ts @@ -0,0 +1,191 @@ +import { fakeAsync, TestBed, tick } from '@angular/core/testing'; +import { getQuestionDocId, Question } from 'realtime-server/lib/esm/scriptureforge/models/question'; +import { QuestionDoc } from '../../app/core/models/question-doc'; +import { MemoryOfflineStore } from '../memory-offline-store'; +import { MemoryRealtimeDocAdapter } from '../memory-realtime-remote-store'; +import { noopDestroyRef } from '../realtime.service'; +import { provideTestRealtime } from '../test-realtime-providers'; +import { TestRealtimeService } from '../test-realtime.service'; +import { configureTestingModule } from '../test-utils'; +import { TypeRegistry } from '../type-registry'; +import { FileType } from './file-offline-data'; +import { RealtimeQuery } from './realtime-query'; + +// These tests cover the reconciliation of offline (IndexedDB) query results with the server's +// query results, which guards against stale offline snapshots corrupting live query membership +// (SF-3893). See doc/offline-query-membership.md. +describe('RealtimeQuery', () => { + configureTestingModule(() => ({ + providers: [provideTestRealtime(new TypeRegistry([QuestionDoc], [FileType.Audio], []))] + })); + + // SF-3893: a question archived while the client was away must not reappear in a subscribed + // activeOnly query when an unrelated local write (e.g. answering another question) causes the + // query to re-run against the offline store, which still holds the stale pre-archive snapshot. + it('should not resurrect remotely-excluded docs on unrelated local writes', fakeAsync(() => { + const env = new TestEnvironment(); + env.addRemoteQuestion(1, false); + env.addRemoteQuestion(2, true, 2); // archived on the server while this client was away + env.addRemoteQuestion(3, false); + env.addOfflineQuestion(1, false); + env.addOfflineQuestion(2, false); // stale offline snapshot from before the archive + env.addOfflineQuestion(3, false); + env.subscribeQuery(); + tick(); + + // The server's query results are authoritative: the archived question is excluded + expect(env.queryDocIds()).toEqual([env.docId(1), env.docId(3)]); + + // SUT + env.submitUnrelatedOp(1); + tick(); + + expect(env.queryDocIds()).toEqual([env.docId(1), env.docId(3)]); + })); + + it('should keep docs the server includes when the offline snapshot wrongly excludes them', fakeAsync(() => { + const env = new TestEnvironment(); + env.addRemoteQuestion(1, false); + env.addRemoteQuestion(2, false, 2); // unarchived on the server while this client was away + env.addRemoteQuestion(3, false); + env.addOfflineQuestion(1, false); + env.addOfflineQuestion(2, true); // stale offline snapshot from before the unarchive + env.addOfflineQuestion(3, false); + env.subscribeQuery(); + tick(); + + expect(env.queryDocIds()).toEqual([env.docId(1), env.docId(2), env.docId(3)]); + + // SUT + env.submitUnrelatedOp(1); + tick(); + + // Order is not guaranteed until the offline snapshot is reconciled, but membership is + expect(env.queryDocIds().sort()).toEqual([env.docId(1), env.docId(2), env.docId(3)]); + })); + + it('should keep locally-changed docs the server has not acknowledged yet', fakeAsync(() => { + const env = new TestEnvironment(); + env.addRemoteQuestion(1, false); + env.addRemoteQuestion(2, true); // the server still sees the question as archived + env.addOfflineQuestion(1, false); + env.addOfflineQuestion(2, false, 2); // this client has unarchived it... + env.subscribeQuery(); + tick(); + env.simulatePendingOps(2); // ...and the op has not been acknowledged yet + + // SUT + env.submitUnrelatedOp(1); + tick(); + + expect(env.queryDocIds()).toContain(env.docId(2)); + })); + + it('should not re-add locally-removed docs the server has not acknowledged yet', fakeAsync(() => { + const env = new TestEnvironment(); + env.addRemoteQuestion(1, false); + env.addRemoteQuestion(2, false); // the server still sees the question as active + env.addOfflineQuestion(1, false); + env.addOfflineQuestion(2, true, 2); // this client has archived it... + env.subscribeQuery(); + tick(); + env.simulatePendingOps(2); // ...and the op has not been acknowledged yet + + // SUT + env.submitUnrelatedOp(1); + tick(); + + expect(env.queryDocIds()).toEqual([env.docId(1)]); + })); + + it('should reconcile the offline copy of a doc the server removed while its local data still matches', fakeAsync(() => { + const env = new TestEnvironment(); + env.addRemoteQuestion(1, false); + env.addRemoteQuestion(2, false); + env.subscribeQuery(); + tick(); + expect(env.queryDocIds()).toEqual([env.docId(1), env.docId(2)]); + const questionDoc: QuestionDoc = env.getQuestionDoc(2); + const reconcileSpy: jasmine.Spy = spyOn(questionDoc, 'reconcileOfflineData').and.callThrough(); + + // SUT + // Archive the question on the server without the client seeing the op, then notify the query + env.addRemoteQuestion(2, true, 2); + env.realtimeService.updateQueryAdaptersRemote(); + tick(); + + expect(env.queryDocIds()).toEqual([env.docId(1)]); + expect(reconcileSpy).toHaveBeenCalled(); + })); +}); + +class TestEnvironment { + static readonly projectId: string = 'project1'; + + readonly realtimeService: TestRealtimeService = TestBed.inject(TestRealtimeService); + query?: RealtimeQuery; + + docId(num: number): string { + return getQuestionDocId(TestEnvironment.projectId, `q${num}`); + } + + addRemoteQuestion(num: number, isArchived: boolean, v: number = 1): void { + this.realtimeService.addSnapshot(QuestionDoc.COLLECTION, { + id: this.docId(num), + data: this.questionData(num, isArchived), + v: v + }); + } + + addOfflineQuestion(num: number, isArchived: boolean, v: number = 1): void { + const offlineStore = this.realtimeService.offlineStore as MemoryOfflineStore; + offlineStore.addData(QuestionDoc.COLLECTION, { + id: this.docId(num), + v: v, + data: this.questionData(num, isArchived), + pendingOps: [] + } as any); + } + + subscribeQuery(): void { + void this.realtimeService + .subscribeQuery( + QuestionDoc.COLLECTION, + { projectRef: TestEnvironment.projectId, isArchived: false }, + noopDestroyRef + ) + .then(query => (this.query = query)); + } + + queryDocIds(): string[] { + return this.query!.docs.map(d => d.id); + } + + getQuestionDoc(num: number): QuestionDoc { + return this.realtimeService.get(QuestionDoc.COLLECTION, this.docId(num)); + } + + simulatePendingOps(num: number): void { + const adapter = this.getQuestionDoc(num).adapter as MemoryRealtimeDocAdapter; + adapter.pendingOps.push({ op: [] }); + } + + submitUnrelatedOp(num: number): void { + void this.getQuestionDoc(num).submitJson0Op(op => op.set(q => q.dateModified, new Date(2026, 0, 2).toISOString())); + } + + private questionData(num: number, isArchived: boolean): Question { + const date = new Date(2026, 0, 1, num).toISOString(); + return { + dataId: `q${num}`, + projectRef: TestEnvironment.projectId, + ownerRef: 'user1', + verseRef: { bookNum: 1, chapterNum: 1, verseNum: num }, + text: `Question ${num}`, + isArchived: isArchived, + dateCreated: date, + dateModified: date, + answers: [] + }; + } +} diff --git a/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-query.ts b/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-query.ts index a8c90c0c8e3..1bff01e6eba 100644 --- a/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-query.ts +++ b/src/SIL.XForge.Scripture/ClientApp/src/xforge-common/models/realtime-query.ts @@ -1,6 +1,7 @@ import arrayDiff, { InsertDiff, MoveDiff, RemoveDiff } from 'arraydiff'; import { BehaviorSubject, Observable, Subject, Subscription } from 'rxjs'; import { takeUntil } from 'rxjs/operators'; +import { performQuery, QueryParameters } from '../query-parameters'; import { RealtimeQueryAdapter } from '../realtime-remote-store'; import { RealtimeService } from '../realtime.service'; import { RealtimeDoc } from './realtime-doc'; @@ -12,6 +13,8 @@ import { RealtimeDoc } from './realtime-doc'; export class RealtimeQuery { private _docs: T[] = []; private unsubscribe$ = new Subject(); + private changeLock?: Promise; + private latestChangeId: number = 0; private _count: number = 0; private _unpagedCount: number = 0; private isDisposed = false; @@ -134,6 +137,9 @@ export class RealtimeQuery { let count: number; if (results instanceof Array) { docIds = results.map(s => s.id); + if (this.adapter.ready && this.adapter.subscribed) { + docIds = this.reconcileWithRemote(docIds); + } count = docIds.length; } else { count = results; @@ -142,6 +148,51 @@ export class RealtimeQuery { return docIds; } + /** + * Merges the results of a query of the offline store with the server's current results. The + * offline store can hold stale snapshots of docs that changed on the server while this client + * was not subscribed to them, so while the remote query is live, its membership is authoritative + * and the local results may only diverge from it for docs with pending local ops (i.e. changes + * the server has not acknowledged yet). Docs whose offline snapshots are thereby proven stale + * are refreshed in the background. See doc/offline-query-membership.md. + */ + private reconcileWithRemote(localDocIds: string[]): string[] { + // Two differently-paged result sets cannot be meaningfully merged, so for paged queries use + // the server's results as-is. + if (this.adapter.parameters.$skip != null || this.adapter.parameters.$limit != null) { + return Array.from(this.adapter.docIds); + } + + const serverDocIds: string[] = this.adapter.docIds; + const serverDocIdSet = new Set(serverDocIds); + const localDocIdSet = new Set(localDocIds); + const docIds: string[] = []; + for (const docId of localDocIds) { + const doc: T = this.realtimeService.get(this.collection, docId); + if (serverDocIdSet.has(docId) || doc.hasPendingOps) { + docIds.push(docId); + } else { + // The offline snapshot matches the query, but the server excludes the doc and this client + // has no unacknowledged changes to it, so the offline snapshot must be stale (e.g. the doc + // was archived while this client was not subscribed to it). + void doc.reconcileOfflineData(); + } + } + for (const docId of serverDocIds) { + if (!localDocIdSet.has(docId)) { + const doc: T = this.realtimeService.get(this.collection, docId); + if (!doc.hasPendingOps) { + // The server includes the doc, but the offline snapshot is missing or does not match + // the query, so the offline snapshot must be stale. The doc is appended out of sort + // order until reconciliation refreshes the offline snapshot. + docIds.push(docId); + void doc.reconcileOfflineData(); + } + } + } + return docIds; + } + private async onReady(): Promise { if (this.subscribed) { await this.onChange(true, this.adapter.docIds, this.adapter.count, this.adapter.unpagedCount); @@ -153,12 +204,65 @@ export class RealtimeQuery { } } - private async onChange( + /** + * Applies a change to the query results. Changes are applied strictly one at a time: a change + * computes its diff against the current results and applies it with index-based splices, so two + * changes being applied concurrently (e.g. a server-driven change interleaving with a local + * re-query at an await point) would corrupt the results. When no change is in flight, the + * change starts synchronously so that timing is unaffected in the common non-overlapping case. + * + * NOTE: this must be written with async/await rather than promise method calls: ts-mockito + * finds method names by scanning the class source, so a call to a promise's "then" method + * anywhere in this class (even in a comment) would give mocked RealtimeQuery instances a + * stubbed "then" method, making them thenables that never settle when awaited or passed to + * Promise.resolve() in tests. + */ + private onChange( + emitRemoteChanges: boolean, + docIds: string[] | undefined, + count: number, + unpagedCount: number + ): Promise { + const change: Promise = + this.changeLock == null + ? this.applyChange(emitRemoteChanges, docIds, count, unpagedCount) + : this.applyChangeAfter(this.changeLock, emitRemoteChanges, docIds, count, unpagedCount); + const changeId: number = ++this.latestChangeId; + this.changeLock = this.releaseLockWhenDone(change, changeId); + return change; + } + + private async applyChangeAfter( + lock: Promise, + emitRemoteChanges: boolean, + docIds: string[] | undefined, + count: number, + unpagedCount: number + ): Promise { + await lock; + await this.applyChange(emitRemoteChanges, docIds, count, unpagedCount); + } + + private async releaseLockWhenDone(change: Promise, changeId: number): Promise { + try { + await change; + } catch { + // A failed change is reported to the onChange() caller; the lock just needs to be released. + } + if (this.latestChangeId === changeId) { + this.changeLock = undefined; + } + } + + private async applyChange( emitRemoteChanges: boolean, docIds: string[] | undefined, count: number, unpagedCount: number ): Promise { + if (this.isDisposed) { + return; + } let changed = false; if (this.count !== count) { this._count = count; @@ -177,7 +281,11 @@ export class RealtimeQuery { case 'remove': const removeDiff = diff as RemoveDiff; - this.onRemove(removeDiff.index, before.slice(removeDiff.index, removeDiff.index + removeDiff.howMany)); + this.onRemove( + removeDiff.index, + before.slice(removeDiff.index, removeDiff.index + removeDiff.howMany), + emitRemoteChanges + ); break; case 'move': @@ -217,9 +325,16 @@ export class RealtimeQuery { this._docs.splice(index, 0, ...newDocs); } - private onRemove(index: number, docIds: string[]): void { + private onRemove(index: number, docIds: string[], removedByServer: boolean = false): void { const removedDocs = this._docs.splice(index, docIds.length); for (const doc of removedDocs) { + if (removedByServer && !doc.hasPendingOps && this.matchesQueryFilter(doc)) { + // The server removed the doc from the results, yet the local copy of the doc still + // matches the query, so the local copy (and the offline snapshot it was loaded from) must + // be stale. Refresh it so that a later query of the offline store cannot re-add the doc. + // See doc/offline-query-membership.md. + void doc.reconcileOfflineData(); + } doc.onRemovedFromSubscribeQuery(); const subscription = this.docSubscriptions.get(doc.id); if (subscription != null) { @@ -229,6 +344,20 @@ export class RealtimeQuery { } } + /** Checks whether the doc's current data matches this query's filter (ignoring paging/sorting). */ + private matchesQueryFilter(doc: T): boolean { + if (doc.data == null) { + return false; + } + const filter: QueryParameters = { ...this.adapter.parameters }; + delete filter.$sort; + delete filter.$skip; + delete filter.$limit; + delete filter.$count; + const { results } = performQuery(filter, [{ id: doc.id, data: doc.data }]); + return results instanceof Array && results.length > 0; + } + private onMove(from: number, to: number, length: number): void { const removedDocs = this._docs.splice(from, length); this._docs.splice(to, 0, ...removedDocs);