observeCount() Can Remain Stale After Rapid Writes

Code Review: Two Notification Reliability Issues

I found two issues that can cause stale values or missed notifications under normal edge-case usage.

Both are relatively small fixes, but they can lead to difficult-to-debug behavior because the application may appear to work correctly under normal conditions while silently missing updates during rapid changes or subscriber mutations.

Finding 1 — observeCount() Can Remain Stale After Rapid Writes

Severity: High
Location: subscribeToCount/index.js

Problem

observeCount() currently uses:

throttleTime(250)

RxJS throttleTime() uses leading-only behavior by default.

This means the first event in a 250 ms window is emitted, while subsequent events during that window are discarded.

For a database count observer, this is problematic because the last database change can be the most important one.

Example

Suppose the current count is 0 and two writes happen very quickly:

const subscription = query.observeCount().subscribe(console.log);

await database.write(() => tasks.create());
await database.write(() => tasks.create());

If both writes occur within the same 250 ms throttle window, the first change may be emitted while the second change is discarded.

The observer can therefore end up with:

Actual database count: 2
Observer's last emitted count: 1

and remain at 1 until another database change occurs.

In some timing scenarios, the observer can similarly remain at 0 even though the database has already changed.

Why This Happens

Current behavior:

Write #1
   │
   ▼
Emit count ────────────────┐
                           │ 250 ms throttle window
Write #2                   │
   │                       │
   ▼                       │
Discarded ❌               │
                           ▼
                      Window closes

The second write is lost because throttleTime() does not emit the trailing value by default.

For a state-observation API such as observeCount(), losing the final state is undesirable.

️ Suggested Fix

Enable trailing emissions:

throttleTime(250, undefined, {
  trailing: true
})

This changes the behavior to:

Write #1
   │
   ▼
Emit immediately
   │
   │ 250 ms window
   │
Write #2
   │
   ▼
Remember latest value
   │
   ▼
Emit latest value when window closes

This preserves throttling while ensuring the latest database state is eventually delivered.

Recommended Regression Test

A regression test should perform multiple writes within the throttle window and verify that the final count is eventually emitted.

For example:

it('should emit the final count after rapid writes', async () => {
  const counts = [];

  const subscription = query
    .observeCount()
    .subscribe(count => counts.push(count));

  await database.write(() => tasks.create());
  await database.write(() => tasks.create());

  await new Promise(resolve => setTimeout(resolve, 300));

  expect(counts.at(-1)).toBe(2);

  subscription.unsubscribe();
});

The exact test setup should follow the repository's existing testing utilities.

Important

Existing tests that wait more than 250 ms between writes may accidentally hide this issue.

A regression test should intentionally perform writes within the 250 ms throttle window.

✅ Acceptance Criteria

• observeCount() does not permanently miss the latest count after rapid writes.

• Multiple writes occurring within 250 ms are handled correctly.

• The latest count is eventually emitted after the throttle window.

• Existing throttling behavior is preserved.

• A regression test covers multiple writes within the throttle window.

• Existing tests continue to pass.

Finding 2 — A Subscriber Can Prevent the Next Subscriber From Receiving Notifications

Severity: Medium
Location: SharedSubscribable/index.js

Problem

SharedSubscribable._notify() currently iterates directly over the live _subscribers array.

If a subscriber unsubscribes itself while notification is in progress, the underlying array is modified during iteration.

This can cause the next subscriber to be skipped.

For example:

const unsubscribeA = shared.subscribe(() => {
  unsubscribeA();
});

shared.subscribe(secondSubscriber);

sourceEmit(value);

Depending on the implementation of _notify(), removing the first subscriber while iterating can shift the second subscriber into the current index.

The result is that:

Subscriber A → receives notification
Subscriber A → unsubscribes itself
Subscriber B → skipped ❌

This can cause missed updates and stale UI state.

Why This Happens

Consider the subscriber list:

Before notification:

_subscribers
┌──────────────┬──────────────┐
│ Subscriber A │ Subscriber B │
└──────────────┴──────────────┘
       ▲
       │
    index = 0

A receives the notification and removes itself:

During notification:

_subscribers
┌──────────────┐
│ Subscriber B │
└──────────────┘
       ▲
       │
    index = 1

Because the array has changed while it is being iterated, Subscriber B can be skipped.

️ Suggested Fix

Iterate over a snapshot of the subscriber list instead of the live array:

this._subscribers.slice().forEach(([subscriber]) => {
  subscriber(value);
});

Or, equivalently:

[...this._subscribers].forEach(([subscriber]) => {
  subscriber(value);
});

This creates a stable snapshot before notification begins.

The notification then operates on:

Original subscribers
       │
       ▼
Create snapshot
       │
       ▼
┌──────────────┬──────────────┐
│ Subscriber A │ Subscriber B │
└──────────────┴──────────────┘
       │              │
       ▼              ▼
    Notify A       Notify B
       │
       ▼
 A unsubscribes
       │
       ▼
Original list changes
       │
       ▼
Snapshot remains unchanged

Therefore, removing a subscriber during notification does not affect the current notification cycle.

Recommended Regression Test

Add a test where the first subscriber unsubscribes itself during notification:

it('should notify all subscribers even when one unsubscribes during notification', () => {
  const firstSubscriber = jest.fn();
  const secondSubscriber = jest.fn();

  let unsubscribeFirst;

  unsubscribeFirst = shared.subscribe(() => {
    firstSubscriber();
    unsubscribeFirst();
  });

  shared.subscribe(secondSubscriber);

  sourceEmit('value');

  expect(firstSubscriber).toHaveBeenCalledTimes(1);
  expect(secondSubscriber).toHaveBeenCalledTimes(1);
});

A second notification can verify that the unsubscribed subscriber is actually removed:

sourceEmit('next-value');

expect(firstSubscriber).toHaveBeenCalledTimes(1);
expect(secondSubscriber).toHaveBeenCalledTimes(2);

This ensures both behaviors are correct:

1. The current notification reaches every subscriber.

2. A subscriber that unsubscribes during notification does not receive future notifications.

Quick Summary

Overall Recommendation

Both issues are related to notification reliability rather than the core functionality itself.

They may therefore be difficult to detect through ordinary tests because the happy path continues to work.

I recommend adding regression coverage for both cases:

Rapid state changes
       │
       ▼
observeCount()
       │
       ▼
Latest state must eventually be emitted

and:

Subscriber notification
       │
       ▼
Subscriber mutates subscription
       │
       ▼
Remaining subscribers must still be notified

These changes would make the observable behavior significantly more deterministic and prevent subtle stale-state or missed-update bugs.