From 8d9b77d8557221091a0f27c4ca63622cc2623856 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Johannes=20K=C3=B6hler?= Date: Sat, 31 Oct 2026 08:10:54 +0100 Subject: [PATCH 1/5] Add Collection.observeRecord for records that may be missing findAndObserve(id) errors when the record does not exist and completes when it is deleted. In withObservables this rethrows in render, so a row whose related record is missing crashes the tree, and a record that arrives later (e.g. via sync) is never shown. experimentalFailsafe only catches the error and emits its fallback once, so it never recovers. observeRecord(id) emits null while the record is missing, the record once it exists or a record with that id is created, and null again instead of completing when it is deleted. It keeps observing throughout. Relations get a failsafe option that makes observe() use observeRecord, and @experimentalFailsafe sets it, mapping a missing record to the fallback value, so failsafe relations recover when the record appears. Adds tests for findAndObserve, observeRecord, failsafe relations and withObservables rendering a null prop, plus docs and a changelog entry. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG-Unreleased.md | 2 + docs-website/docs/docs/CRUD.md | 3 +- docs-website/docs/docs/Components.md | 7 + docs-website/docs/docs/Relation.md | 11 + src/Collection/index.d.ts | 8 + src/Collection/index.js | 97 ++++++++- src/Collection/test.js | 191 ++++++++++++++++++ src/Relation/helpers.js | 22 +- src/Relation/index.d.ts | 5 + src/Relation/index.js | 6 + src/Relation/test.js | 100 +++++++++ src/decorators/experimentalFailsafe/index.js | 20 +- src/decorators/experimentalFailsafe/test.js | 144 +++++++++++++ .../withObservables/withObservables.test.js | 51 +++++ 14 files changed, 650 insertions(+), 17 deletions(-) create mode 100644 src/decorators/experimentalFailsafe/test.js diff --git a/CHANGELOG-Unreleased.md b/CHANGELOG-Unreleased.md index 9d43e2bdd..5f71752ab 100644 --- a/CHANGELOG-Unreleased.md +++ b/CHANGELOG-Unreleased.md @@ -6,6 +6,8 @@ ### New features +- New `Collection.observeRecord(id)`: a failsafe version of `findAndObserve(id)` that emits `null` while the record does not exist (instead of erroring), emits the record once it is created, and emits `null` (instead of completing) when it is deleted. `@experimentalFailsafe` relations now use it, so their `observe()` recovers when a missing related record is created later + ### Fixes - [LokiJS] Multitab sync issue fix diff --git a/docs-website/docs/docs/CRUD.md b/docs-website/docs/docs/CRUD.md index fc443f211..af5279f97 100644 --- a/docs-website/docs/docs/CRUD.md +++ b/docs-website/docs/docs/CRUD.md @@ -111,7 +111,8 @@ await somePost.destroyPermanently() // permanent - `Model.observe()` - usually you only use this [when connecting records to components](./Components.md), but you can manually observe a record outside of React components. The returned [RxJS](https://github.com/reactivex/rxjs) `Observable` will emit the record immediately upon subscription, and then every time the record is updated. If the record is deleted, the Observable will complete. - `Query.observe()`, `Relation.observe()` — analagous to the above, but for [Queries](./Query.md) and [Relations](./Relation.md) - `Query.observeWithColumns()` - used for [sorted lists](./Components.md) -- `Collection.findAndObserve(id)` — same as using `.find(id)` and then calling `record.observe()` +- `Collection.findAndObserve(id)` — same as using `.find(id)` and then calling `record.observe()`. The Observable errors if the record does not exist, and completes if it's deleted. +- `Collection.observeRecord(id)` — failsafe version of `findAndObserve(id)` for records that might not exist (yet). Emits `null` while the record is missing, emits the record once it exists or is created later (e.g. arrives via sync), and emits `null` again (instead of completing) when it's deleted. Use it when rendering a missing record as a placeholder is better than an error. - `Model.prepareUpdate()`, `Collection.prepareCreate`, `Database.batch` — used for [batch updates](./Writers.md) - `Database.unsafeResetDatabase()` destroys the whole database - [be sure to see this comment before using it](https://github.com/Nozbe/WatermelonDB/blob/22188ee5b6e3af08e48e8af52d14e0d90db72925/src/Database/index.js#L131) - To override the `record.id` during the creation, e.g. to sync with a remote database, you can do it by `record._raw` property. Be aware that the `id` must be of type `string`. diff --git a/docs-website/docs/docs/Components.md b/docs-website/docs/docs/Components.md index ceb9cd98f..bb59e361d 100644 --- a/docs-website/docs/docs/Components.md +++ b/docs-website/docs/docs/Components.md @@ -161,6 +161,13 @@ withObservables(['post'], ({ post }) => ({ post: database.get('posts').findAndObserve(postId) })) ``` + If the post does not exist, `findAndObserve` errors, and the error is re-thrown when rendering (use an Error Boundary to catch it). If the record might legitimately be missing — for example, it hasn't been synced yet, or it was deleted — use **observeRecord** instead: + ```js + withObservables(['postId'], ({ postId, database }) => ({ + post: database.get('posts').observeRecord(postId) + })) + ``` + Your component then receives `post={null}` while the post is missing (render a placeholder), and is re-rendered with the post as soon as it's created. If the post is deleted, `post` becomes `null` again. Similarly, mark a relation with `@experimentalFailsafe` to make `Relation.observe()` emit a fallback value (instead of erroring) while the related record is missing, and recover once it's created. 1. **RxJS transformations**. The values returned by `Model.observe()`, `Query.observe()`, `Relation.observe()` are [RxJS Observables](https://github.com/ReactiveX/rxjs). You can use standard transforms like mapping, filtering, throttling, startWith to change when and how the component is re-rendered. 1. **Custom Observables**. `withObservables` is a general-purpose HOC for Observables, not just Watermelon. You can create new props from any `Observable`. diff --git a/docs-website/docs/docs/Relation.md b/docs-website/docs/docs/Relation.md index acddddda4..17af67875 100644 --- a/docs-website/docs/docs/Relation.md +++ b/docs-website/docs/docs/Relation.md @@ -63,6 +63,17 @@ withObservables(['comment'], ({ comment }) => ({ The component will now have an `author` prop containing a `User`, and will re-render both when the user changes (e.g. comment's author changes its name), but also when a new author is assigned to the comment (if that was possible). +If the related record might not exist (e.g. it hasn't been synced yet), observing the relation errors. To render a placeholder instead, mark the relation with `@experimentalFailsafe`. `author` will then be `undefined` (or the fallback value you pass to the decorator) while the user is missing, and the component re-renders with the `User` once it's created: + +```js +import { experimentalFailsafe } from '@nozbe/watermelondb/decorators' + +class Comment extends Model { + // ... + @experimentalFailsafe() @relation('users', 'author_id') author +} +``` + ### Fetching To simply get the related record, use `fetch`. You might need it [in a Writer](./Writers.md) diff --git a/src/Collection/index.d.ts b/src/Collection/index.d.ts index af47bb402..5595a0162 100644 --- a/src/Collection/index.d.ts +++ b/src/Collection/index.d.ts @@ -41,6 +41,11 @@ export default class Collection { // (with the same semantics as when calling `model.observe()`) findAndObserve(id: RecordId): Observable + // Observes the record with the given ID, treating a missing record as `null`. + // Failsafe version of `findAndObserve`: emits `null` while the record does not exist, + // the record once it exists (or is created later), and `null` again when it is deleted + observeRecord(id: RecordId): Observable + // Query records of this type query: ArrayOrSpreadFn> @@ -87,6 +92,9 @@ export default class Collection { // Fetches exactly one record (See: Collection.find) _fetchRecord(id: RecordId, callback: ResultCallback): void + // Fetches exactly one record, or `null` if it does not exist (See: Collection.observeRecord) + _fetchRecordIfExists(id: RecordId, callback: ResultCallback): void + _applyChangesToCache(operations: CollectionChangeSet): void _notify(operations: CollectionChangeSet): void diff --git a/src/Collection/index.js b/src/Collection/index.js index 5b8405e51..8858cc081 100644 --- a/src/Collection/index.js +++ b/src/Collection/index.js @@ -116,6 +116,83 @@ export default class Collection { }) } + /** + * Observes the record with the given ID, treating a missing record as `null`. + * + * This is a failsafe version of `findAndObserve()`. The returned Observable never errors or + * completes because of the record's existence. Instead, it emits: + * - the record, if it exists, and then again every time it changes + * - `null`, if the record does not exist (yet) + * - the record, once a record with this ID is created (e.g. arrives via sync) + * - `null` again, if the record is deleted — and the record again, if it is re-created + * + * Use this instead of `findAndObserve()` when the record might not exist, e.g. when observing + * a related record that may arrive later, so that your UI can render a placeholder instead of + * throwing an error. + */ + observeRecord(id: RecordId): Observable { + return Observable.create((observer) => { + let unsubscribed = false + let observedRecord: ?Record = null + let unsubscribeFromRecord: ?Unsubscribe = null + + const stopObservingRecord = () => { + unsubscribeFromRecord && unsubscribeFromRecord() + unsubscribeFromRecord = null + observedRecord = null + } + + const observeExistingRecord = (record: Record) => { + stopObservingRecord() + observedRecord = record + observer.next(record) + unsubscribeFromRecord = record.experimentalSubscribe((isDeleted) => { + if (unsubscribed) { + return + } + if (isDeleted) { + stopObservingRecord() + observer.next(null) + } else { + observer.next(record) + } + }) + } + + // Watch for a record with this ID being created (or re-created after deletion) + const unsubscribeFromCollection = this.experimentalSubscribe((changes) => { + if (unsubscribed) { + return + } + changes.forEach(({ record, type }) => { + if (type === 'created' && record.id === id && record !== observedRecord) { + observeExistingRecord(record) + } + }) + }) + + this._fetchRecordIfExists(id, (result) => { + if (unsubscribed || observedRecord) { + // Unsubscribed, or a `created` notification arrived while we were fetching + return + } + if (result.error) { + observer.error(result.error) + } else if (result.value) { + observeExistingRecord(result.value) + } else { + observer.next(null) + } + }) + + return () => { + unsubscribed = true + unsubscribeFromCollection() + stopObservingRecord() + } + }) + } + /*:: query: ArrayOrSpreadFn> */ /** * Returns a `Query` with conditions given. @@ -217,6 +294,18 @@ export default class Collection { // Fetches exactly one record (See: Collection.find) _fetchRecord(id: RecordId, callback: ResultCallback): void { + this._fetchRecordIfExists(id, (result) => + callback( + mapValue((record) => { + invariant(record, `Record ${this.table}#${id} not found`) + return record + }, result), + ), + ) + } + + // Fetches exactly one record, or `null` if it does not exist (See: Collection.observeRecord) + _fetchRecordIfExists(id: RecordId, callback: ResultCallback): void { if (typeof id !== 'string') { callback({ error: new Error(`Invalid record ID ${this.table}#${id}`) }) return @@ -231,10 +320,10 @@ export default class Collection { this.database.adapter.underlyingAdapter.find(this.table, id, (result) => callback( - mapValue((rawRecord) => { - invariant(rawRecord, `Record ${this.table}#${id} not found`) - return this._cache.recordFromQueryResult(rawRecord) - }, result), + mapValue( + (rawRecord) => (rawRecord ? this._cache.recordFromQueryResult(rawRecord) : null), + result, + ), ), ) } diff --git a/src/Collection/test.js b/src/Collection/test.js index 3607aab9b..00d670e37 100644 --- a/src/Collection/test.js +++ b/src/Collection/test.js @@ -355,3 +355,194 @@ describe('Collection observation', () => { expect(subscriber).toHaveBeenCalledTimes(4) }) }) + +describe('findAndObserve', () => { + it('observes an existing record and completes when it is deleted', async () => { + const { database, tasks } = mockDatabase() + const t1 = await database.write(() => tasks.create()) + + const observer = jest.fn() + const complete = jest.fn() + tasks.findAndObserve(t1.id).subscribe(observer, undefined, complete) + + await new Promise(process.nextTick) + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(t1) + + await database.write(() => t1.update()) + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(t1) + + await database.write(() => t1.markAsDeleted()) + expect(observer).toHaveBeenCalledTimes(2) + expect(complete).toHaveBeenCalledTimes(1) + }) + it('errors if the record does not exist', async () => { + const { tasks } = mockDatabase() + + const observer = jest.fn() + const error = jest.fn() + tasks.findAndObserve('missing').subscribe(observer, error) + + await new Promise(process.nextTick) + expect(observer).toHaveBeenCalledTimes(0) + expect(error).toHaveBeenCalledTimes(1) + expect(error.mock.calls[0][0].message).toMatch('Record mock_tasks#missing not found') + }) +}) + +describe('observeRecord', () => { + const createWithId = (database, collection, id) => + database.write(() => + collection.create((record) => { + record._raw.id = id + }), + ) + + it('emits null if the record does not exist', async () => { + const { tasks } = mockDatabase() + + const observer = jest.fn() + const error = jest.fn() + const complete = jest.fn() + tasks.observeRecord('missing').subscribe(observer, error, complete) + + await new Promise(process.nextTick) + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + expect(error).toHaveBeenCalledTimes(0) + expect(complete).toHaveBeenCalledTimes(0) + }) + it('emits the record once a record with this id is created', async () => { + const { database, tasks } = mockDatabase() + + const observer = jest.fn() + const subscription = tasks.observeRecord('later').subscribe(observer) + + await new Promise(process.nextTick) + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + + // unrelated records don't matter + await database.write(() => tasks.create()) + expect(observer).toHaveBeenCalledTimes(1) + + const t1 = await createWithId(database, tasks, 'later') + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(t1) + + // ...and then keeps observing the record + await database.write(() => t1.update()) + expect(observer).toHaveBeenCalledTimes(3) + expect(observer).toHaveBeenLastCalledWith(t1) + + subscription.unsubscribe() + await database.write(() => t1.update()) + expect(observer).toHaveBeenCalledTimes(3) + }) + it('emits null instead of completing when the record is deleted, and recovers', async () => { + const { database, tasks } = mockDatabase() + const t1 = await createWithId(database, tasks, 't1') + + const observer = jest.fn() + const error = jest.fn() + const complete = jest.fn() + const subscription = tasks.observeRecord('t1').subscribe(observer, error, complete) + + // record is cached, so it's emitted synchronously + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(t1) + + await database.write(() => t1.markAsDeleted()) + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(null) + expect(complete).toHaveBeenCalledTimes(0) + + // purge the deleted record (like sync does), then re-create a record with the same id + await database.adapter.destroyDeletedRecords('mock_tasks', ['t1']) + const t1Again = await createWithId(database, tasks, 't1') + expect(t1Again).not.toBe(t1) + expect(observer).toHaveBeenCalledTimes(3) + expect(observer).toHaveBeenLastCalledWith(t1Again) + + await database.write(() => t1Again.destroyPermanently()) + expect(observer).toHaveBeenCalledTimes(4) + expect(observer).toHaveBeenLastCalledWith(null) + + expect(error).toHaveBeenCalledTimes(0) + expect(complete).toHaveBeenCalledTimes(0) + subscription.unsubscribe() + }) + it('fetches records that are not cached, including ids containing colons', async () => { + const { database, tasks: originalTasks, cloneDatabase } = mockDatabase() + const id = '12:Quiz:hand' + await createWithId(database, originalTasks, id) + + // simulate reload - the record is in the database, but not in cache + const reloadedDb = await cloneDatabase() + const tasks = reloadedDb.get('mock_tasks') + const findSpy = jest.spyOn(reloadedDb.adapter.underlyingAdapter, 'find') + + const observer = jest.fn() + tasks.observeRecord(id).subscribe(observer) + expect(observer).toHaveBeenCalledTimes(0) + + await new Promise(process.nextTick) + expect(findSpy).toHaveBeenCalledTimes(1) + expect(findSpy.mock.calls[0].slice(0, 2)).toEqual(['mock_tasks', id]) + expect(observer).toHaveBeenCalledTimes(1) + expect(observer.mock.calls[0][0].id).toBe(id) + + // a missing record with such an id is also fine + const observer2 = jest.fn() + tasks.observeRecord('12:Quiz:missing').subscribe(observer2) + await new Promise(process.nextTick) + expect(observer2).toHaveBeenLastCalledWith(null) + }) + it('does not emit twice if the record is created while fetching', async () => { + const { database, tasks, adapter } = mockDatabase() + + let finishFind = null + adapter.find = jest.fn().mockImplementation((table, id, callback) => { + finishFind = () => callback({ value: null }) + }) + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + expect(adapter.find).toHaveBeenCalledTimes(1) + + const t1 = await createWithId(database, tasks, 't1') + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(t1) + + finishFind() + expect(observer).toHaveBeenCalledTimes(1) + }) + it('errors for invalid ids', async () => { + const { tasks } = mockDatabase() + + const observer = jest.fn() + const error = jest.fn() + tasks.observeRecord(null).subscribe(observer, error) + + expect(observer).toHaveBeenCalledTimes(0) + expect(error).toHaveBeenCalledTimes(1) + expect(error.mock.calls[0][0].message).toMatch('Invalid record ID') + }) + it('stops observing after unsubscribing', async () => { + const { database, tasks } = mockDatabase() + + const observer = jest.fn() + const subscription = tasks.observeRecord('t1').subscribe(observer) + await new Promise(process.nextTick) + expect(observer).toHaveBeenCalledTimes(1) + + subscription.unsubscribe() + expect(tasks._subscribers.length).toBe(0) + + const t1 = await createWithId(database, tasks, 't1') + await database.write(() => t1.markAsDeleted()) + expect(observer).toHaveBeenCalledTimes(1) + expect(t1._subscribers.length).toBe(0) + }) +}) diff --git a/src/Relation/helpers.js b/src/Relation/helpers.js index 49fe9838a..18f17f79c 100644 --- a/src/Relation/helpers.js +++ b/src/Relation/helpers.js @@ -9,13 +9,19 @@ import { } from '../utils/rx' import type Relation from './index' -import type Model from '../Model' +import type Model, { RecordId } from '../Model' + +// Failsafe relations emit `null` (and keep observing) instead of erroring when the related +// record does not exist, so that they recover when the record is created later +const observeRecord = (relation: Relation, id: RecordId): Observable => { + const collection = relation._model.collections.get(relation._relationTableName) + // $FlowFixMe + return relation._isFailsafe ? collection.observeRecord(id) : collection.findAndObserve(id) +} const getImmutableObservable = (relation: Relation): Observable => - relation._model.collections - .get(relation._relationTableName) - // $FlowFixMe - .findAndObserve(relation.id) + // $FlowFixMe + observeRecord(relation, relation.id) const getObservable = (relation: Relation): Observable => relation._model @@ -24,11 +30,7 @@ const getObservable = (relation: Relation): Observable => .pipe( map$((model) => model._getRaw(relation._columnName)), distinctUntilChanged(), - switchMap((id) => - id - ? relation._model.collections.get(relation._relationTableName).findAndObserve(id) - : of$(null), - ), + switchMap((id) => (id ? observeRecord(relation, id) : of$(null))), ) // eslint-disable-next-line diff --git a/src/Relation/index.d.ts b/src/Relation/index.d.ts index e863d2afd..4f24dda6c 100644 --- a/src/Relation/index.d.ts +++ b/src/Relation/index.d.ts @@ -11,6 +11,9 @@ type ExtractRecordId = ExtractRecordIdNonOptional & ExtractRecordIdOptional export type Options = $Exact<{ isImmutable: boolean + // If true, observing a related record that does not exist emits `null` (and recovers once the + // record is created) instead of erroring. See `Collection.observeRecord` and `@experimentalFailsafe` + failsafe?: boolean }> // Defines a one-to-one relation between two Models (two tables in db) @@ -27,6 +30,8 @@ export default class Relation { _isImmutable: boolean + _isFailsafe: boolean + // TODO: FIX TS // @lazy _cachedObservable: Observable diff --git a/src/Relation/index.js b/src/Relation/index.js index 0ad850679..84740f36e 100644 --- a/src/Relation/index.js +++ b/src/Relation/index.js @@ -16,6 +16,9 @@ type ExtractRecordId = ExtractRecordIdNonOptional & ExtractRecordIdOptional export type Options = $Exact<{ isImmutable: boolean, + // If true, observing a related record that does not exist emits `null` (and recovers once the + // record is created) instead of erroring. See `Collection.observeRecord` and `@experimentalFailsafe` + failsafe?: boolean, }> // Defines a one-to-one relation between two Models (two tables in db) @@ -32,6 +35,8 @@ export default class Relation { _isImmutable: boolean + _isFailsafe: boolean + @lazy _cachedObservable: Observable = createObservable(this) .pipe(publishReplayLatestWhileConnected) @@ -47,6 +52,7 @@ export default class Relation { this._relationTableName = relationTableName this._columnName = columnName this._isImmutable = options.isImmutable + this._isFailsafe = !!options.failsafe } get id(): $Call { diff --git a/src/Relation/test.js b/src/Relation/test.js index 79e7e8501..56de2d751 100644 --- a/src/Relation/test.js +++ b/src/Relation/test.js @@ -148,6 +148,106 @@ describe('Relation', () => { expect(await relation).toBe(currentRecord) expect(await relation.then((model) => [model])).toEqual([currentRecord]) }) + it('errors when observing a related record that does not exist', async () => { + const { tasks, db } = mockDatabase() + const primary = await db.write(() => + tasks.create((mock) => { + mock.projectId = 'missing' + }), + ) + const relation = new Relation(primary, 'mock_projects', 'project_id', { isImmutable: false }) + + const observer = jest.fn() + const error = jest.fn() + relation.observe().subscribe(observer, error) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(0) + expect(error).toHaveBeenCalledTimes(1) + }) + it('observes a missing related record as null when failsafe, and recovers', async () => { + const { tasks, projects, db } = mockDatabase() + const primary = await db.write(() => + tasks.create((mock) => { + mock.projectId = 'missing' + }), + ) + const relation = new Relation(primary, 'mock_projects', 'project_id', { + isImmutable: false, + failsafe: true, + }) + expect(relation._isFailsafe).toBe(true) + + const observer = jest.fn() + const error = jest.fn() + const complete = jest.fn() + const subscription = relation.observe().subscribe(observer, error, complete) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + + // related record arrives later (e.g. via sync) + const secondary = await db.write(() => + projects.create((mock) => { + mock._raw.id = 'missing' + mock.name = 'foo' + }), + ) + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(secondary) + + await db.write(() => + secondary.update((mock) => { + mock.name = 'bar' + }), + ) + expect(observer).toHaveBeenCalledTimes(3) + expect(observer).toHaveBeenLastCalledWith(secondary) + + // related record is deleted + await db.write(() => secondary.markAsDeleted()) + expect(observer).toHaveBeenCalledTimes(4) + expect(observer).toHaveBeenLastCalledWith(null) + + // relation is re-pointed at another record + const secondary2 = await db.write(() => projects.create()) + await db.write(() => + primary.update((mock) => { + mock.projectId = secondary2.id + }), + ) + expect(observer).toHaveBeenLastCalledWith(secondary2) + + expect(error).toHaveBeenCalledTimes(0) + expect(complete).toHaveBeenCalledTimes(0) + subscription.unsubscribe() + }) + it('observes a missing immutable related record as null when failsafe, and recovers', async () => { + const { tasks, comments, db } = mockDatabase() + const primary = await db.write(() => + comments.create((mock) => { + mock.task.id = 'missing' + }), + ) + const relation = new Relation(primary, 'mock_tasks', 'task_id', { + isImmutable: true, + failsafe: true, + }) + + const observer = jest.fn() + const subscription = relation.observe().subscribe(observer) + await new Promise(process.nextTick) + expect(observer).toHaveBeenLastCalledWith(null) + + const secondary = await db.write(() => + tasks.create((mock) => { + mock._raw.id = 'missing' + }), + ) + expect(observer).toHaveBeenLastCalledWith(secondary) + subscription.unsubscribe() + }) it('caches observable', () => { const { tasks } = mockDatabase() const model = new MockTask(tasks, {}) diff --git a/src/decorators/experimentalFailsafe/index.js b/src/decorators/experimentalFailsafe/index.js index 7fc40df69..6ae6bf65d 100644 --- a/src/decorators/experimentalFailsafe/index.js +++ b/src/decorators/experimentalFailsafe/index.js @@ -1,7 +1,7 @@ // @flow import { isObj } from '../../utils/fp' -import { catchError, of } from '../../utils/rx' +import { catchError, map, of } from '../../utils/rx' type FailsafeDecorator = ( fallback?: any, @@ -29,6 +29,14 @@ const failsafe: FailsafeDecorator = const originalFetch = value.fetch const originalObserve = value.observe + // A Relation can recover from a missing related record: when marked as failsafe, its + // `observe()` emits `null` while the record does not exist (and the record once it + // is created) instead of erroring — which, if caught below, would end the observation + const isRelation = !!value.constructor && value.constructor._wmelonTag === 'relation' + if (isRelation) { + value._isFailsafe = true + } + if (typeof originalFetch === 'function') { value.fetch = function fetch(...args): any { const result = originalFetch.apply(value, args) @@ -43,7 +51,15 @@ const failsafe: FailsafeDecorator = value.observe = function observe(...args): any { const result = originalObserve.apply(value, args) if (isObj(result) && typeof result.pipe === 'function') { - return result.pipe(catchError(() => of(fallback))) + const operators = [catchError(() => of(fallback))] + if (isRelation) { + // Missing related record is emitted as `null` — map it to the fallback value + // (a relation whose ID is null legitimately emits `null`, so leave that alone) + operators.unshift( + map((record) => (record === null && value.id ? fallback : record)), + ) + } + return result.pipe(...operators) } return result } diff --git a/src/decorators/experimentalFailsafe/test.js b/src/decorators/experimentalFailsafe/test.js new file mode 100644 index 000000000..1d3396c53 --- /dev/null +++ b/src/decorators/experimentalFailsafe/test.js @@ -0,0 +1,144 @@ +import { of, throwError } from 'rxjs' +import { MockTask, mockDatabase } from '../../__tests__/testModels' + +import experimentalFailsafe from './index' +import relation from '../relation' +import immutableRelation from '../immutableRelation' + +const makeProxy = (task, fallback) => { + class TaskProxy { + asModel = task + + @experimentalFailsafe(fallback) + @relation('mock_projects', 'project_id') + project + + @experimentalFailsafe(fallback) + @immutableRelation('mock_project_sections', 'project_section_id') + projectSection + } + return new TaskProxy() +} + +describe('decorators/experimentalFailsafe', () => { + it('marks relations as failsafe', () => { + const { tasks } = mockDatabase() + const proxy = makeProxy(new MockTask(tasks, { project_id: 'p1' })) + expect(proxy.project._isFailsafe).toBe(true) + expect(proxy.projectSection._isFailsafe).toBe(true) + expect(new MockTask(tasks, {}).project._isFailsafe).toBe(false) + }) + it('fetches missing related records as the fallback value', async () => { + const { tasks, projects, db } = mockDatabase() + const task = await db.write(() => + tasks.create((mock) => { + mock.projectId = 'missing' + }), + ) + expect(await makeProxy(task).project.fetch()).toBe(undefined) + expect(await makeProxy(task, 'fallback').project.fetch()).toBe('fallback') + expect(await makeProxy(task).projectSection.fetch()).toBe(null) // null id + + const project = await db.write(() => + projects.create((mock) => { + mock._raw.id = 'missing' + }), + ) + expect(await makeProxy(task).project.fetch()).toBe(project) + }) + it('observes missing related records as the fallback value, and recovers', async () => { + const { tasks, projects, db } = mockDatabase() + const task = await db.write(() => + tasks.create((mock) => { + mock.projectId = 'missing' + }), + ) + const proxy = makeProxy(task, 'fallback') + + const observer = jest.fn() + const error = jest.fn() + const complete = jest.fn() + const subscription = proxy.project.observe().subscribe(observer, error, complete) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith('fallback') + + // related record arrives later + const project = await db.write(() => + projects.create((mock) => { + mock._raw.id = 'missing' + }), + ) + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(project) + + // ...and is deleted again + await db.write(() => project.markAsDeleted()) + expect(observer).toHaveBeenCalledTimes(3) + expect(observer).toHaveBeenLastCalledWith('fallback') + + // a null relation id still yields null, not the fallback + await db.write(() => + task.update((mock) => { + mock.projectId = null + }), + ) + expect(observer).toHaveBeenLastCalledWith(null) + + expect(error).toHaveBeenCalledTimes(0) + expect(complete).toHaveBeenCalledTimes(0) + subscription.unsubscribe() + }) + it('observes missing immutable related records as the fallback value, and recovers', async () => { + const { tasks, projectSections, db } = mockDatabase() + const task = await db.write(() => + tasks.create((mock) => { + mock._setRaw('project_section_id', 'missing') + }), + ) + const proxy = makeProxy(task) + + const observer = jest.fn() + const subscription = proxy.projectSection.observe().subscribe(observer) + await new Promise(process.nextTick) + expect(observer).toHaveBeenLastCalledWith(undefined) + + const section = await db.write(() => + projectSections.create((mock) => { + mock._raw.id = 'missing' + }), + ) + expect(observer).toHaveBeenLastCalledWith(section) + subscription.unsubscribe() + }) + it('still catches errors of arbitrary observable-like values', async () => { + class Proxy { + @experimentalFailsafe('fallback') + get broken() { + return { + fetch: () => Promise.reject(new Error('nope')), + observe: () => throwError(() => new Error('nope')), + } + } + + @experimentalFailsafe('fallback') + get working() { + return { + fetch: () => Promise.resolve('value'), + observe: () => of('value'), + } + } + } + const proxy = new Proxy() + + expect(await proxy.broken.fetch()).toBe('fallback') + expect(await proxy.working.fetch()).toBe('value') + + const observer = jest.fn() + proxy.broken.observe().subscribe(observer) + expect(observer).toHaveBeenLastCalledWith('fallback') + proxy.working.observe().subscribe(observer) + expect(observer).toHaveBeenLastCalledWith('value') + }) +}) diff --git a/src/react/withObservables/withObservables.test.js b/src/react/withObservables/withObservables.test.js index 6c6fe5469..9a6533cc1 100644 --- a/src/react/withObservables/withObservables.test.js +++ b/src/react/withObservables/withObservables.test.js @@ -1,4 +1,7 @@ import * as React from 'react' +import * as TestRenderer from 'react-test-renderer' +import { Subject } from 'rxjs' +import { mockDatabase } from '../../__tests__/testModels' import withObservables from './index' describe('withObservables', () => { @@ -13,4 +16,52 @@ describe('withObservables', () => { const WrappedComponent = withObservables([], getObservables)(A) expect(WrappedComponent.nonReactProp).toBe(A.nonReactProp) }) + it('renders a null prop without throwing, and re-renders when the record arrives', async () => { + const { database, tasks } = mockDatabase() + + // eslint-disable-next-line react/prop-types + const Task = ({ task }) => {task ? `task:${task.id}` : 'missing'} + const EnhancedTask = withObservables(['taskId'], ({ taskId }) => ({ + task: tasks.observeRecord(taskId), + }))(Task) + + let renderer + await TestRenderer.act(async () => { + renderer = TestRenderer.create() + }) + expect(renderer.root.findByType('span').children).toEqual(['missing']) + + await TestRenderer.act(async () => { + await database.write(() => + tasks.create((task) => { + task._raw.id = 't1' + }), + ) + }) + expect(renderer.root.findByType('span').children).toEqual(['task:t1']) + + await TestRenderer.act(async () => { + await database.write(async () => { + const task = await tasks.find('t1') + await task.markAsDeleted() + }) + }) + expect(renderer.root.findByType('span').children).toEqual(['missing']) + + renderer.unmount() + }) + it('rethrows observable errors in render', () => { + const subject = new Subject() + const Base = () => ok + const Enhanced = withObservables([], () => ({ value: subject }))(Base) + const consoleError = jest.spyOn(console, 'error').mockImplementation(() => {}) + + expect(() => { + TestRenderer.act(() => { + TestRenderer.create() + subject.error(new Error('boom')) + }) + }).toThrow('boom') + consoleError.mockRestore() + }) }) From 84680a34acc3eff40f77a2df53ebf9b7d45f4b02 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Johannes=20K=C3=B6hler?= Date: Sat, 31 Oct 2026 08:19:49 +0100 Subject: [PATCH 2/5] Fix observeRecord edge cases found in review - Model and Collection notify a copy of their subscriber list, so a subscriber that unsubscribes while being notified no longer makes the next subscriber miss the notification (e.g. a second observeRecord subscription missing the null on deletion). - observeRecord subscribes to the record before emitting it, so a consumer that unsubscribes synchronously does not leak the listener. - observeRecord treats a fetched record marked as deleted as missing. - Failsafe relations with a null id emit null instead of erroring. Co-Authored-By: Claude Opus 5.5 --- src/Collection/index.js | 15 ++++-- src/Collection/test.js | 55 +++++++++++++++++++++ src/Model/index.js | 6 ++- src/Relation/helpers.js | 8 ++- src/Relation/test.js | 18 +++++++ src/decorators/experimentalFailsafe/test.js | 13 +++++ 6 files changed, 106 insertions(+), 9 deletions(-) diff --git a/src/Collection/index.js b/src/Collection/index.js index 8858cc081..c6f7a7139 100644 --- a/src/Collection/index.js +++ b/src/Collection/index.js @@ -122,7 +122,7 @@ export default class Collection { * This is a failsafe version of `findAndObserve()`. The returned Observable never errors or * completes because of the record's existence. Instead, it emits: * - the record, if it exists, and then again every time it changes - * - `null`, if the record does not exist (yet) + * - `null`, if the record does not exist (yet) or is marked as deleted * - the record, once a record with this ID is created (e.g. arrives via sync) * - `null` again, if the record is deleted — and the record again, if it is re-created * @@ -145,7 +145,8 @@ export default class Collection { const observeExistingRecord = (record: Record) => { stopObservingRecord() observedRecord = record - observer.next(record) + // Subscribe before emitting, so that if the consumer unsubscribes synchronously, + // the teardown can remove this listener unsubscribeFromRecord = record.experimentalSubscribe((isDeleted) => { if (unsubscribed) { return @@ -157,6 +158,7 @@ export default class Collection { observer.next(record) } }) + observer.next(record) } // Watch for a record with this ID being created (or re-created after deletion) @@ -176,11 +178,13 @@ export default class Collection { // Unsubscribed, or a `created` notification arrived while we were fetching return } + const record = result.value if (result.error) { observer.error(result.error) - } else if (result.value) { - observeExistingRecord(result.value) + } else if (record && record.syncStatus !== 'deleted') { + observeExistingRecord(record) } else { + // Missing, or marked as deleted (adapters still find records marked as deleted) observer.next(null) } }) @@ -346,7 +350,8 @@ export default class Collection { ]): void => { subscriber(operations) } - this._subscribers.forEach(collectionChangeNotifySubscribers) + // Iterate over a copy, so that subscribers can safely unsubscribe while being notified + this._subscribers.slice().forEach(collectionChangeNotifySubscribers) this.changes.next(operations) const collectionChangeNotifyModels = ({ record, type }: CollectionChange): void => { diff --git a/src/Collection/test.js b/src/Collection/test.js index 00d670e37..85b8e8dc0 100644 --- a/src/Collection/test.js +++ b/src/Collection/test.js @@ -1,3 +1,4 @@ +import { filter, take } from 'rxjs/operators' import { expectToRejectWithMessage } from '../__tests__/utils' import { noop } from '../utils/fp' @@ -518,6 +519,60 @@ describe('observeRecord', () => { finishFind() expect(observer).toHaveBeenCalledTimes(1) }) + it('notifies every subscription when the record is deleted', async () => { + const { database, tasks } = mockDatabase() + const t1 = await createWithId(database, tasks, 't1') + + const observers = [jest.fn(), jest.fn(), jest.fn()] + observers.forEach((observer) => tasks.observeRecord('t1').subscribe(observer)) + observers.forEach((observer) => expect(observer).toHaveBeenLastCalledWith(t1)) + + await database.write(() => t1.markAsDeleted()) + observers.forEach((observer) => { + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(null) + }) + expect(t1._subscribers.length).toBe(0) + }) + it('emits null for records marked as deleted', async () => { + const { database, tasks } = mockDatabase() + const t1 = await createWithId(database, tasks, 't1') + await database.write(() => t1.markAsDeleted()) + expect(tasks._cache.get('t1')).toBe(undefined) + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + + // find() semantics are unchanged + expect((await tasks.find('t1')).syncStatus).toBe('deleted') + }) + it('does not leak listeners when the consumer unsubscribes synchronously', async () => { + const { database, tasks } = mockDatabase() + + // record created later + const observer = jest.fn() + tasks + .observeRecord('t1') + .pipe(filter(Boolean), take(1)) + .subscribe(observer) + await new Promise(process.nextTick) + const t1 = await createWithId(database, tasks, 't1') + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(t1) + expect(t1._subscribers.length).toBe(0) + expect(tasks._subscribers.length).toBe(0) + + // cached record emitted synchronously on subscription + const observer2 = jest.fn() + tasks.observeRecord('t1').pipe(take(1)).subscribe(observer2) + expect(observer2).toHaveBeenLastCalledWith(t1) + expect(t1._subscribers.length).toBe(0) + expect(tasks._subscribers.length).toBe(0) + }) it('errors for invalid ids', async () => { const { tasks } = mockDatabase() diff --git a/src/Model/index.js b/src/Model/index.js index 29596fa00..e21f77ce5 100644 --- a/src/Model/index.js +++ b/src/Model/index.js @@ -416,14 +416,16 @@ export default class Model { _notifyChanged(): void { this._getChanges().next(this) - this._subscribers.forEach(([subscriber]) => { + // Iterate over a copy, so that subscribers can safely unsubscribe while being notified + this._subscribers.slice().forEach(([subscriber]) => { subscriber(false) }) } _notifyDestroyed(): void { this._getChanges().complete() - this._subscribers.forEach(([subscriber]) => { + // Iterate over a copy, so that subscribers can safely unsubscribe while being notified + this._subscribers.slice().forEach(([subscriber]) => { subscriber(true) }) } diff --git a/src/Relation/helpers.js b/src/Relation/helpers.js index 18f17f79c..a62beda4f 100644 --- a/src/Relation/helpers.js +++ b/src/Relation/helpers.js @@ -13,10 +13,14 @@ import type Model, { RecordId } from '../Model' // Failsafe relations emit `null` (and keep observing) instead of erroring when the related // record does not exist, so that they recover when the record is created later -const observeRecord = (relation: Relation, id: RecordId): Observable => { +const observeRecord = (relation: Relation, id: ?RecordId): Observable => { const collection = relation._model.collections.get(relation._relationTableName) + if (relation._isFailsafe) { + // $FlowFixMe + return id ? collection.observeRecord(id) : of$(null) + } // $FlowFixMe - return relation._isFailsafe ? collection.observeRecord(id) : collection.findAndObserve(id) + return collection.findAndObserve(id) } const getImmutableObservable = (relation: Relation): Observable => diff --git a/src/Relation/test.js b/src/Relation/test.js index 56de2d751..48e3ef9e1 100644 --- a/src/Relation/test.js +++ b/src/Relation/test.js @@ -248,6 +248,24 @@ describe('Relation', () => { expect(observer).toHaveBeenLastCalledWith(secondary) subscription.unsubscribe() }) + it('observes a failsafe immutable relation with a null id as null', async () => { + const { tasks, db } = mockDatabase() + const primary = await db.write(() => tasks.create()) + const relation = new Relation(primary, 'mock_project_sections', 'project_section_id', { + isImmutable: true, + failsafe: true, + }) + expect(relation.id).toBe(null) + + const observer = jest.fn() + const error = jest.fn() + relation.observe().subscribe(observer, error) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + expect(error).toHaveBeenCalledTimes(0) + }) it('caches observable', () => { const { tasks } = mockDatabase() const model = new MockTask(tasks, {}) diff --git a/src/decorators/experimentalFailsafe/test.js b/src/decorators/experimentalFailsafe/test.js index 1d3396c53..2d4acb1c5 100644 --- a/src/decorators/experimentalFailsafe/test.js +++ b/src/decorators/experimentalFailsafe/test.js @@ -112,6 +112,19 @@ describe('decorators/experimentalFailsafe', () => { expect(observer).toHaveBeenLastCalledWith(section) subscription.unsubscribe() }) + it('observes an immutable relation with a null id as null, not the fallback', async () => { + const { tasks, db } = mockDatabase() + const task = await db.write(() => tasks.create()) + const proxy = makeProxy(task, 'fallback') + + const observer = jest.fn() + const complete = jest.fn() + proxy.projectSection.observe().subscribe(observer, undefined, complete) + await new Promise(process.nextTick) + + expect(observer).toHaveBeenCalledTimes(1) + expect(observer).toHaveBeenLastCalledWith(null) + }) it('still catches errors of arbitrary observable-like values', async () => { class Proxy { @experimentalFailsafe('fallback') From 1f57968699951dae470aefe8c0156fafc3e8cf87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Johannes=20K=C3=B6hler?= Date: Sat, 31 Oct 2026 08:28:09 +0100 Subject: [PATCH 3/5] Make observeRecord follow the record cache and listen only while missing - On a change notification for the observed id, look the record up in the record cache (which is updated before notifications are sent) instead of trusting the notification's record. A record created and deleted within batched notifications is no longer emitted, and a record deleted and re-created within one batch is switched to directly. - Only subscribe to collection changes while the record is missing, so observed live records no longer scan every change batch. - Ignore the initial fetch result once a change notification for the id has arrived, so a late result cannot emit an outdated record. Co-Authored-By: Claude Opus 5.5 --- src/Collection/index.js | 51 +++++++++++++++++++++-------- src/Collection/test.js | 72 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 110 insertions(+), 13 deletions(-) diff --git a/src/Collection/index.js b/src/Collection/index.js index c6f7a7139..746bc557e 100644 --- a/src/Collection/index.js +++ b/src/Collection/index.js @@ -135,6 +135,10 @@ export default class Collection { let unsubscribed = false let observedRecord: ?Record = null let unsubscribeFromRecord: ?Unsubscribe = null + let unsubscribeFromCollection: ?Unsubscribe = null + // Set once a change notification for this ID arrives - from then on, the record cache is + // the source of truth, and the (possibly outdated) result of the initial fetch is ignored + let initialFetchSuperseded = false const stopObservingRecord = () => { unsubscribeFromRecord && unsubscribeFromRecord() @@ -142,7 +146,20 @@ export default class Collection { observedRecord = null } + const stopObservingCollection = () => { + unsubscribeFromCollection && unsubscribeFromCollection() + unsubscribeFromCollection = null + } + + // Caches are updated before change notifications are sent, so at notification time the + // cache tells us if a (non-deleted) record with this ID currently exists + const getLiveCachedRecord = (): ?Record => { + const record = this._cache.get(id) + return record && record.syncStatus !== 'deleted' ? record : null + } + const observeExistingRecord = (record: Record) => { + stopObservingCollection() stopObservingRecord() observedRecord = record // Subscribe before emitting, so that if the consumer unsubscribes synchronously, @@ -153,7 +170,9 @@ export default class Collection { } if (isDeleted) { stopObservingRecord() - observer.next(null) + // The record might have been re-created in the same batch + const recreatedRecord = getLiveCachedRecord() + recreatedRecord ? observeExistingRecord(recreatedRecord) : observeMissingRecord() } else { observer.next(record) } @@ -161,21 +180,27 @@ export default class Collection { observer.next(record) } - // Watch for a record with this ID being created (or re-created after deletion) - const unsubscribeFromCollection = this.experimentalSubscribe((changes) => { - if (unsubscribed) { - return - } - changes.forEach(({ record, type }) => { - if (type === 'created' && record.id === id && record !== observedRecord) { - observeExistingRecord(record) + // While the record is missing, watch for a record with this ID being created. (Once it + // exists, we only listen to that record, so that we don't have to scan every change) + const startObservingCollection = () => { + unsubscribeFromCollection = this.experimentalSubscribe((changes) => { + if (unsubscribed || !changes.some(({ record }) => record.id === id)) { + return } + initialFetchSuperseded = true + const record = getLiveCachedRecord() + record && observeExistingRecord(record) }) - }) + } + + const observeMissingRecord = () => { + startObservingCollection() + observer.next(null) + } + startObservingCollection() this._fetchRecordIfExists(id, (result) => { - if (unsubscribed || observedRecord) { - // Unsubscribed, or a `created` notification arrived while we were fetching + if (unsubscribed || initialFetchSuperseded || observedRecord) { return } const record = result.value @@ -191,7 +216,7 @@ export default class Collection { return () => { unsubscribed = true - unsubscribeFromCollection() + stopObservingCollection() stopObservingRecord() } }) diff --git a/src/Collection/test.js b/src/Collection/test.js index 85b8e8dc0..dc7ecd4be 100644 --- a/src/Collection/test.js +++ b/src/Collection/test.js @@ -573,6 +573,78 @@ describe('observeRecord', () => { expect(t1._subscribers.length).toBe(0) expect(tasks._subscribers.length).toBe(0) }) + it('only listens to collection changes while the record is missing', async () => { + const { database, tasks } = mockDatabase() + + const observer = jest.fn() + const subscription = tasks.observeRecord('t1').subscribe(observer) + await new Promise(process.nextTick) + expect(tasks._subscribers.length).toBe(1) + + const t1 = await createWithId(database, tasks, 't1') + expect(observer).toHaveBeenLastCalledWith(t1) + expect(tasks._subscribers.length).toBe(0) + expect(t1._subscribers.length).toBe(1) + + await database.write(() => t1.markAsDeleted()) + expect(observer).toHaveBeenLastCalledWith(null) + expect(tasks._subscribers.length).toBe(1) + expect(t1._subscribers.length).toBe(0) + + subscription.unsubscribe() + expect(tasks._subscribers.length).toBe(0) + }) + it('does not emit a record created and deleted within batched notifications', async () => { + const { database, tasks } = mockDatabase() + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + await new Promise(process.nextTick) + expect(observer).toHaveBeenLastCalledWith(null) + + await database.experimentalBatchNotifications(async () => { + const t1 = await createWithId(database, tasks, 't1') + await database.write(() => t1.destroyPermanently()) + }) + expect(observer).toHaveBeenCalledTimes(1) + }) + it('switches to a record deleted and re-created within batched notifications', async () => { + const { database, tasks } = mockDatabase() + const t1 = await createWithId(database, tasks, 't1') + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + expect(observer).toHaveBeenLastCalledWith(t1) + + let t1Again + await database.experimentalBatchNotifications(async () => { + await database.write(() => t1.destroyPermanently()) + t1Again = await createWithId(database, tasks, 't1') + }) + expect(observer).toHaveBeenCalledTimes(2) + expect(observer).toHaveBeenLastCalledWith(t1Again) + expect(t1._subscribers.length).toBe(0) + expect(t1Again._subscribers.length).toBe(1) + }) + it('ignores the initial fetch result if the record changed in the meantime', async () => { + const { database, tasks, adapter } = mockDatabase() + + let finishFind = null + adapter.find = jest.fn().mockImplementation((table, id, callback) => { + finishFind = () => callback({ value: { id: 't1', _status: 'synced', _changed: '' } }) + }) + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + + const t1 = await createWithId(database, tasks, 't1') + await database.write(() => t1.markAsDeleted()) + expect(observer.mock.calls).toEqual([[t1], [null]]) + + // outdated result of the initial fetch arrives late + finishFind() + expect(observer.mock.calls).toEqual([[t1], [null]]) + }) it('errors for invalid ids', async () => { const { tasks } = mockDatabase() From 0fbf35a197277e4aea458df67a4d1e1c6bad3512 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Johannes=20K=C3=B6hler?= Date: Sat, 31 Oct 2026 08:34:53 +0100 Subject: [PATCH 4/5] Emit initial null when observeRecord's fetch is superseded by changes If a record was created and deleted while the initial fetch was still pending, the change notification superseded the fetch without emitting anything, so the subscription never emitted its initial null. Emit null in that case, without emitting null twice in a row. Also document why a superseded fetch result is still turned into a cached model: the adapter marks the record as sent to JS, so dropping it would leave the adapter-side and JS caches out of sync. Co-Authored-By: Claude Opus 5.5 --- src/Collection/index.js | 22 +++++++++++++++++++--- src/Collection/test.js | 25 +++++++++++++++++++++++++ 2 files changed, 44 insertions(+), 3 deletions(-) diff --git a/src/Collection/index.js b/src/Collection/index.js index 746bc557e..f7bb74ea3 100644 --- a/src/Collection/index.js +++ b/src/Collection/index.js @@ -139,6 +139,15 @@ export default class Collection { // Set once a change notification for this ID arrives - from then on, the record cache is // the source of truth, and the (possibly outdated) result of the initial fetch is ignored let initialFetchSuperseded = false + // Whether the last value emitted was `null` (so that we don't emit it twice in a row) + let isMissingEmitted = false + + const emitMissing = () => { + if (!isMissingEmitted) { + isMissingEmitted = true + observer.next(null) + } + } const stopObservingRecord = () => { unsubscribeFromRecord && unsubscribeFromRecord() @@ -177,6 +186,7 @@ export default class Collection { observer.next(record) } }) + isMissingEmitted = false observer.next(record) } @@ -189,16 +199,22 @@ export default class Collection { } initialFetchSuperseded = true const record = getLiveCachedRecord() - record && observeExistingRecord(record) + // (If the record was created and deleted while the initial fetch is pending, we still + // need to emit the initial `null`) + record ? observeExistingRecord(record) : emitMissing() }) } const observeMissingRecord = () => { startObservingCollection() - observer.next(null) + emitMissing() } startObservingCollection() + // NOTE: A fetched raw record is always turned into a cached model (even if this fetch is + // superseded), because the adapter marks it as sent to JS - dropping it would leave the + // adapter-side and JS record caches out of sync. Adapters process calls in order, so a find + // result can't be older than a change notification that arrives before it. this._fetchRecordIfExists(id, (result) => { if (unsubscribed || initialFetchSuperseded || observedRecord) { return @@ -210,7 +226,7 @@ export default class Collection { observeExistingRecord(record) } else { // Missing, or marked as deleted (adapters still find records marked as deleted) - observer.next(null) + emitMissing() } }) diff --git a/src/Collection/test.js b/src/Collection/test.js index dc7ecd4be..a2d4fd53e 100644 --- a/src/Collection/test.js +++ b/src/Collection/test.js @@ -626,6 +626,31 @@ describe('observeRecord', () => { expect(t1._subscribers.length).toBe(0) expect(t1Again._subscribers.length).toBe(1) }) + it('emits null if the record is created and deleted while the initial fetch is pending', async () => { + const { database, tasks, adapter } = mockDatabase() + + let finishFind = null + adapter.find = jest.fn().mockImplementation((table, id, callback) => { + finishFind = () => callback({ value: null }) + }) + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + expect(observer).toHaveBeenCalledTimes(0) + + await database.experimentalBatchNotifications(async () => { + const t1 = await createWithId(database, tasks, 't1') + await database.write(() => t1.destroyPermanently()) + }) + expect(observer.mock.calls).toEqual([[null]]) + + finishFind() + expect(observer.mock.calls).toEqual([[null]]) + + // ...and still recovers + const t1Again = await createWithId(database, tasks, 't1') + expect(observer.mock.calls).toEqual([[null], [t1Again]]) + }) it('ignores the initial fetch result if the record changed in the meantime', async () => { const { database, tasks, adapter } = mockDatabase() From 06abc7f7cd8b084309b326d50effaa517ec486ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Johannes=20K=C3=B6hler?= Date: Sat, 31 Oct 2026 08:42:10 +0100 Subject: [PATCH 5/5] Skip stale updates in observeRecord and allow nullable Relation types - With batched notifications, an update notification can arrive after the record was deleted. observeRecord now only emits an update while the record is still the cached one; the deletion notification that follows emits null. - In TypeScript, Relation now accepts a nullable record type (as Flow already does), so failsafe relations and relations with an optional ID can be declared as e.g. Relation, and observe() and fetch() are typed accordingly. Documented in Relation.md. Co-Authored-By: Claude Opus 5.5 --- docs-website/docs/docs/Relation.md | 2 ++ src/Collection/index.js | 4 +++- src/Collection/test.js | 14 ++++++++++++++ src/Relation/helpers.d.ts | 4 +++- src/Relation/index.d.ts | 11 +++++++---- 5 files changed, 29 insertions(+), 6 deletions(-) diff --git a/docs-website/docs/docs/Relation.md b/docs-website/docs/docs/Relation.md index 17af67875..be0b0a041 100644 --- a/docs-website/docs/docs/Relation.md +++ b/docs-website/docs/docs/Relation.md @@ -74,6 +74,8 @@ class Comment extends Model { } ``` +If you use TypeScript or Flow, declare such a relation with a nullable type (e.g. `author: Relation`), since it can emit `undefined` (or your fallback). The same goes for relations whose ID is optional, which emit `null` when the ID is not set. + ### Fetching To simply get the related record, use `fetch`. You might need it [in a Writer](./Writers.md) diff --git a/src/Collection/index.js b/src/Collection/index.js index f7bb74ea3..cf6b9c075 100644 --- a/src/Collection/index.js +++ b/src/Collection/index.js @@ -182,7 +182,9 @@ export default class Collection { // The record might have been re-created in the same batch const recreatedRecord = getLiveCachedRecord() recreatedRecord ? observeExistingRecord(recreatedRecord) : observeMissingRecord() - } else { + } else if (this._cache.get(id) === record) { + // (With batched notifications, the record might have been deleted in the meantime - + // in that case, a deletion notification follows, so don't emit the deleted record) observer.next(record) } }) diff --git a/src/Collection/test.js b/src/Collection/test.js index a2d4fd53e..54ff4c1b3 100644 --- a/src/Collection/test.js +++ b/src/Collection/test.js @@ -670,6 +670,20 @@ describe('observeRecord', () => { finishFind() expect(observer.mock.calls).toEqual([[t1], [null]]) }) + it('does not emit an update of a record deleted within batched notifications', async () => { + const { database, tasks } = mockDatabase() + const t1 = await createWithId(database, tasks, 't1') + + const observer = jest.fn() + tasks.observeRecord('t1').subscribe(observer) + expect(observer.mock.calls).toEqual([[t1]]) + + await database.experimentalBatchNotifications(async () => { + await database.write(() => t1.update()) + await database.write(() => t1.markAsDeleted()) + }) + expect(observer.mock.calls).toEqual([[t1], [null]]) + }) it('errors for invalid ids', async () => { const { tasks } = mockDatabase() diff --git a/src/Relation/helpers.d.ts b/src/Relation/helpers.d.ts index aa6bf44a4..37c00b4f6 100644 --- a/src/Relation/helpers.d.ts +++ b/src/Relation/helpers.d.ts @@ -5,4 +5,6 @@ import type { Observable } from '../utils/rx' import type Relation from './index' import type Model from '../Model' -export declare function createObservable(relation: Relation): Observable +export declare function createObservable( + relation: Relation, +): Observable diff --git a/src/Relation/index.d.ts b/src/Relation/index.d.ts index 4f24dda6c..f371a1fc6 100644 --- a/src/Relation/index.d.ts +++ b/src/Relation/index.d.ts @@ -1,4 +1,4 @@ -import { $NonMaybeType, $Exact, $Call } from '../types' +import { $Exact, $Call } from '../types' import type { Observable } from '../utils/rx' import type Model from '../Model' @@ -18,7 +18,10 @@ export type Options = $Exact<{ // Defines a one-to-one relation between two Models (two tables in db) // Do not create this object directly! Use `relation` or `immutableRelation` decorators instead -export default class Relation { +// +// Use a nullable type (e.g. `Relation`) if the relation ID is optional, or if the +// relation is marked with `@experimentalFailsafe` (`Relation` by default) +export default class Relation { // Used by withObservables to differentiate between object types static _wmelonTag: string @@ -26,7 +29,7 @@ export default class Relation { _columnName: ColumnName - _relationTableName: TableName<$NonMaybeType> + _relationTableName: TableName> _isImmutable: boolean @@ -38,7 +41,7 @@ export default class Relation { constructor( model: Model, - relationTableName: TableName<$NonMaybeType>, + relationTableName: TableName>, columnName: ColumnName, options: Options, )