Skip to content

[Hold] Avoid subscriber re-fire when storage writes retry #792

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
elirangoshen wants to merge 1 commit into
base: main
Choose a base branch
from
+157 −9
Open

[Hold] Avoid subscriber re-fire when storage writes retry #792

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 35 additions & 9 deletions lib/OnyxUtils.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -769,9 +769,14 @@ function getCollectionDataAndSendAsObject<TKey extends OnyxKey>(matchingKeys: Co
/**
* Remove a key from Onyx and update the subscribers
*/
function remove<TKey extends OnyxKey>(key: TKey, isProcessingCollectionUpdate?: boolean): Promise<void> {
function remove<TKey extends OnyxKey>(key: TKey, isProcessingCollectionUpdate?: boolean, skipNotify?: boolean): Promise<void> {
cache.drop(key);
keyChanged(key, undefined as OnyxValue<TKey>, undefined, isProcessingCollectionUpdate);
// skipNotify is used by retryOperation's eviction branch — the imminent retry's cache.set
// will re-populate cache, so firing keyChanged(undefined) here would only strand subscribers
// in the "removed" state across the retry.
if (!skipNotify) {
keyChanged(key, undefined as OnyxValue<TKey>, undefined, isProcessingCollectionUpdate);
}

if (OnyxKeys.isRamOnlyKey(key)) {
return Promise.resolve();
Expand Down Expand Up @@ -842,8 +847,10 @@ function retryOperation<TMethod extends RetriableOnyxOperation>(error: Error, on
Logger.logInfo(`Out of storage. Evicting least recently accessed key (${keyForRemoval}) and retrying. Error: ${error}`);
reportStorageQuota(error);

// skipNotify=true: retry's orchestrator skips keysChanged on retryAttempt > 0, so we
// must not let remove() fire keyChanged(undefined) — cache.set on retry restores the value.
// @ts-expect-error No overload matches this call.
return remove(keyForRemoval).then(() => onyxMethod(defaultParams, nextRetryAttempt));
return remove(keyForRemoval, undefined, true).then(() => onyxMethod(defaultParams, nextRetryAttempt));
Copy link
Copy Markdown

@chatgpt-codex-connector chatgpt-codex-connector Bot May 28, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Notify subscribers when evicting unrelated keys

When a storage-capacity retry evicts the LRU key, that key is often unrelated to the write being retried (e.g. the existing eviction test writes TEST_KEY and evicts TEST_KEY1). Passing skipNotify=true here suppresses keyChanged(undefined) for the evicted key, but the retry only re-populates keys in defaultParams, so subscribers to the evicted LRU key keep showing stale data even though it was dropped from cache and removed from storage. The notification should only be skipped when the evicted key is actually restored by the retried operation.

Useful? React with 👍 / 👎.

}

/**
Expand Down Expand Up @@ -1395,14 +1402,21 @@ function multiSetWithRetry(data: OnyxMultiSetInput, retryAttempt?: number): Prom
// so re-entrant callbacks (e.g. Onyx.set inside a callback) see consistent cache
// and subscriber state, matching the original per-key notification semantics.
cache.set(key, value);
keyChanged(key, value);
// Skip subscriber notification on retry — already notified on attempt 0.
// waitForCollectionCallback subscribers re-fire on every keyChanged by contract.
if (!retryAttempt) {
keyChanged(key, value);
}
}
}

// One keysChanged() per collection — fires each collection-level subscriber once and lets
// keysChanged() internally decide which individual member subscribers need notification.
for (const [collectionKey, batch] of collectionBatches) {
keysChanged(collectionKey as CollectionKeyBase, batch.partial, batch.previous);
// Skip on retry — already notified on attempt 0 (see same-reason comment above).
if (!retryAttempt) {
for (const [collectionKey, batch] of collectionBatches) {
keysChanged(collectionKey as CollectionKeyBase, batch.partial, batch.previous);
}
}

const keyValuePairsToStore = keyValuePairsToSet.filter((keyValuePair) => {
Expand Down Expand Up @@ -1476,7 +1490,11 @@ function setCollectionWithRetry<TKey extends CollectionKeyBase>({collectionKey,

for (const [key, value] of keyValuePairs) cache.set(key, value);

keysChanged(collectionKey, mutableCollection, previousCollection);
// Skip subscriber notification on retry — already notified on attempt 0.
// waitForCollectionCallback subscribers re-fire on every keysChanged by contract.
if (!retryAttempt) {
keysChanged(collectionKey, mutableCollection, previousCollection);
}

// RAM-only keys are not supposed to be saved to storage
if (OnyxKeys.isRamOnlyKey(collectionKey)) {
Expand Down Expand Up @@ -1611,7 +1629,11 @@ function mergeCollectionWithPatches<TKey extends CollectionKeyBase>(
// write fails.
const previousCollection = getCachedCollection(collectionKey, existingKeys);
cache.merge(finalMergedCollection);
keysChanged(collectionKey, finalMergedCollection, previousCollection);
// Skip subscriber notification on retry — already notified on attempt 0.
// waitForCollectionCallback subscribers re-fire on every keysChanged by contract.
if (!retryAttempt) {
keysChanged(collectionKey, finalMergedCollection, previousCollection);
}

const promises = [];

Expand Down Expand Up @@ -1690,7 +1712,11 @@ function partialSetCollection<TKey extends CollectionKeyBase>({collectionKey, co

for (const [key, value] of keyValuePairs) cache.set(key, value);

keysChanged(collectionKey, mutableCollection, previousCollection);
// Skip subscriber notification on retry — already notified on attempt 0.
// waitForCollectionCallback subscribers re-fire on every keysChanged by contract.
if (!retryAttempt) {
keysChanged(collectionKey, mutableCollection, previousCollection);
}

if (OnyxKeys.isRamOnlyKey(collectionKey)) {
sendActionToDevTools(METHOD.SET_COLLECTION, undefined, mutableCollection);
Expand Down
122 changes: 122 additions & 0 deletions tests/unit/onyxUtilsTest.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -920,6 +920,128 @@ describe('OnyxUtils', () => {
});
});

describe('retry side-effect idempotency', () => {
// Save originals so each test can replace StorageMock.multiMerge / StorageMock.multiSet
// with a one-shot rejecting mock that triggers retryOperation's transient-error path.
// Restoring keeps mocks from leaking into the storage-eviction describe block below.
const originalMultiMerge = StorageMock.multiMerge;
const originalMultiSet = StorageMock.multiSet;

afterEach(() => {
StorageMock.multiMerge = originalMultiMerge;
StorageMock.multiSet = originalMultiSet;
});

// A retriable error: not in NON_RETRIABLE_ERRORS, not in STORAGE_ERRORS, so retryOperation
// re-enters the failing method on the next attempt.
const transientError = new Error('Transient storage error');

it('mergeCollection — waitForCollectionCallback subscriber fires once across retries', async () => {
const collectionKey = ONYXKEYS.COLLECTION.TEST_KEY;
const existingMemberKey = `${collectionKey}1`;
const newMemberKey = `${collectionKey}2`;

await Onyx.set(existingMemberKey, {value: 'initial'});

const collectionCallback = jest.fn();
Onyx.connect({
key: collectionKey,
waitForCollectionCallback: true,
callback: collectionCallback,
});
await waitForPromisesToResolve();
collectionCallback.mockClear();

StorageMock.multiMerge = jest.fn(originalMultiMerge).mockRejectedValueOnce(transientError);

await Onyx.mergeCollection(collectionKey, {
[existingMemberKey]: {value: 'merged'},
[newMemberKey]: {value: 'new'},
} as GenericCollection);

// Before this fix, every retry attempt re-fired keysChanged() — and
// waitForCollectionCallback subscribers fire on every keysChanged() call by contract.
// After the fix, retries skip the keysChanged re-fire, so subscribers are notified
// exactly once per logical operation.
expect(collectionCallback).toHaveBeenCalledTimes(1);
});

it('Onyx.multiSet — collection subscriber fires once across retries', async () => {
const collectionKey = ONYXKEYS.COLLECTION.TEST_KEY;
const memberKey1 = `${collectionKey}1`;
const memberKey2 = `${collectionKey}2`;

const collectionCallback = jest.fn();
Onyx.connect({
key: collectionKey,
waitForCollectionCallback: true,
callback: collectionCallback,
});
await waitForPromisesToResolve();
collectionCallback.mockClear();

StorageMock.multiSet = jest.fn(originalMultiSet).mockRejectedValueOnce(transientError);

await Onyx.multiSet({
[memberKey1]: {value: 'first'},
[memberKey2]: {value: 'second'},
});

expect(collectionCallback).toHaveBeenCalledTimes(1);
});

it('Onyx.setCollection — collection subscriber fires once across retries', async () => {
const collectionKey = ONYXKEYS.COLLECTION.TEST_KEY;
const memberKey1 = `${collectionKey}1`;
const memberKey2 = `${collectionKey}2`;

const collectionCallback = jest.fn();
Onyx.connect({
key: collectionKey,
waitForCollectionCallback: true,
callback: collectionCallback,
});
await waitForPromisesToResolve();
collectionCallback.mockClear();

StorageMock.multiSet = jest.fn(originalMultiSet).mockRejectedValueOnce(transientError);

await Onyx.setCollection(collectionKey, {
[memberKey1]: {value: 'first'},
[memberKey2]: {value: 'second'},
} as GenericCollection);

expect(collectionCallback).toHaveBeenCalledTimes(1);
});

it('OnyxUtils.partialSetCollection — collection subscriber fires once across retries', async () => {
const collectionKey = ONYXKEYS.COLLECTION.TEST_KEY;
const memberKey1 = `${collectionKey}1`;
const memberKey2 = `${collectionKey}2`;

const collectionCallback = jest.fn();
Onyx.connect({
key: collectionKey,
waitForCollectionCallback: true,
callback: collectionCallback,
});
await waitForPromisesToResolve();
collectionCallback.mockClear();

StorageMock.multiSet = jest.fn(originalMultiSet).mockRejectedValueOnce(transientError);

await OnyxUtils.partialSetCollection({
collectionKey,
collection: {
[memberKey1]: {value: 'first'},
[memberKey2]: {value: 'second'},
} as GenericCollection,
});

expect(collectionCallback).toHaveBeenCalledTimes(1);
});
});

describe('storage eviction', () => {
const diskFullError = new Error('database or disk is full');

Expand Down
Loading