Fix RxState write queue permanently breaking after a modifier throws (#8398)

When a user-supplied modifier passed to RxState.set() throws, the
shared write queue promise became permanently rejected, causing all
subsequent set() calls to reject with an unrelated SNH error instead
of performing the write.

The fix splits the per-call write promise from the shared queue
promise so a failing write still propagates its error to the caller
but does not block future writes.

Co-authored-by: Claude <noreply@anthropic.com>

Author: pubkey

orga/changelog/fix-rxstate-write-queue-recover-after-modifier-throws.md

@@ -0,0 +1 @@
+- FIX `RxState.set()` permanently breaking the write queue when a user-supplied modifier throws, causing all subsequent `set()` calls to reject with an unrelated `SNH` error instead of performing the write

src/plugins/state/rx-state.ts

@@ -42,7 +42,6 @@ import {
     RxStateOperation,
     RxStateModifier
 } from './types.ts';
-import { newRxError } from '../../rx-error.ts';
 import { runPluginHooks } from '../../hooks.ts';
 
 
@@ -144,7 +143,7 @@ export class RxStateBase<T, Reactivity = unknown> {
      * that would throw conflict errors and trigger a retry.
      */
     _triggerWrite() {
-        this._writeQueue = this._writeQueue.then(async () => {
+        const next = this._writeQueue.then(async () => {
             if (this._nonPersisted.length === 0) {
                 return;
             }
@@ -207,13 +206,10 @@ export class RxStateBase<T, Reactivity = unknown> {
                     await promiseWait(0);
                 }
             }
-        }).catch(error => {
-            throw newRxError('SNH', {
-                name: 'RxState WRITE QUEUE ERROR',
-                error
-            });
         });
-        return this._writeQueue;
+        // Keep the shared queue alive so a failing write does not block subsequent ones.
+        this._writeQueue = next.catch(() => { });
+        return next;
     }
 
     mergeOperationsIntoState(

test/unit/rx-state.test.ts

@@ -743,6 +743,38 @@ addRxPlugin(RxDBJsonDumpPlugin);
                 // State must still be correct
                 assert.strictEqual(state.a, 6);
 
+                state.collection.database.remove();
+            });
+            /**
+             * When a modifier passed to set() throws, the write queue
+             * must not be permanently broken. Subsequent writes with
+             * valid modifiers should still succeed.
+             */
+            it('write queue should recover after a modifier throws', async () => {
+                const state = await getState();
+
+                await state.set('a', () => 1);
+                assert.strictEqual(state.get('a'), 1);
+
+                const thrownError = new Error('bad modifier');
+                let caughtError: any;
+                try {
+                    await state.set('a', () => {
+                        throw thrownError;
+                    });
+                } catch (err) {
+                    caughtError = err;
+                }
+                assert.ok(caughtError, 'the first set() should reject');
+
+                // a subsequent valid write must still succeed
+                await state.set('a', () => 2);
+                assert.strictEqual(state.get('a'), 2);
+
+                // and a later write should still work
+                await state.set('a', (prev: any) => prev + 1);
+                assert.strictEqual(state.get('a'), 3);
+
                 state.collection.database.remove();
             });
         });