Fix sync propagation for mutating listener deletions

src/mergeable-store/index.ts

@@ -43,6 +43,7 @@ import {
   objForEach,
   objFreeze,
   objHas,
+  objIsEmpty,
   objMap,
   objNew,
   objValidate,
@@ -153,6 +154,7 @@ export const createMergeableStore = ((
   getNow?: GetNow,
 ): MergeableStore => {
   let listeningToRawStoreChanges = 1;
+  let incomingChanges: Changes | undefined;
   let contentStampMap = newContentStampMap();
   let defaultingContent: 0 | 1 = 0;
   const touchedCells: IdSet3 = mapNew();
@@ -327,15 +329,28 @@ export const createMergeableStore = ((
     cellId: Id,
     newCell: CellOrUndefined,
   ) => {
-    setAdd(
-      mapEnsure(
-        mapEnsure(touchedCells, tableId, mapNew<Id, IdSet>),
-        rowId,
-        setNew<Id>,
-      ),
-      cellId,
-    );
-    if (listeningToRawStoreChanges) {
+    const tableChanges = incomingChanges?.[0]?.[tableId];
+    const rowChanges = tableChanges?.[rowId];
+    const hasExpected = !!rowChanges && objHas(rowChanges, cellId);
+    const expectedCell = rowChanges?.[cellId];
+    const isExpected = hasExpected && Object.is(expectedCell, newCell);
+    if (isExpected) {
+      delete incomingChanges?.[0]?.[tableId]?.[rowId]?.[cellId];
+      if (objIsEmpty(incomingChanges?.[0]?.[tableId]?.[rowId])) {
+        delete incomingChanges?.[0]?.[tableId]?.[rowId];
+      }
+      if (objIsEmpty(incomingChanges?.[0]?.[tableId])) {
+        delete incomingChanges?.[0]?.[tableId];
+      }
+    } else if (listeningToRawStoreChanges || incomingChanges) {
+      setAdd(
+        mapEnsure(
+          mapEnsure(touchedCells, tableId, mapNew<Id, IdSet>),
+          rowId,
+          setNew<Id>,
+        ),
+        cellId,
+      );
       mergeContentOrChanges([
         [
           {
@@ -360,8 +375,14 @@ export const createMergeableStore = ((
   };
 
   const valueChanged = (valueId: Id, newValue: ValueOrUndefined) => {
-    setAdd(touchedValues, valueId);
-    if (listeningToRawStoreChanges) {
+    const valueChanges = incomingChanges?.[1];
+    const hasExpected = !!valueChanges && objHas(valueChanges, valueId);
+    const expectedValue = valueChanges?.[valueId];
+    const isExpected = hasExpected && Object.is(expectedValue, newValue);
+    if (isExpected) {
+      delete incomingChanges?.[1]?.[valueId];
+    } else if (listeningToRawStoreChanges || incomingChanges) {
+      setAdd(touchedValues, valueId);
       mergeContentOrChanges([
         [{}],
         [
@@ -607,10 +628,17 @@ export const createMergeableStore = ((
 
   const applyMergeableChanges = (
     mergeableChanges: MergeableChanges | MergeableContent,
-  ): MergeableStore =>
-    disableListeningToRawStoreChanges(() =>
-      store.applyChanges(mergeContentOrChanges(mergeableChanges)),
-    );
+  ): MergeableStore => {
+    const changes = mergeContentOrChanges(mergeableChanges);
+    return disableListeningToRawStoreChanges(() => {
+      incomingChanges = changes;
+      try {
+        store.applyChanges(changes);
+      } finally {
+        incomingChanges = undefined;
+      }
+    });
+  };
 
   const merge = (mergeableStore2: MergeableStore) => {
     const mergeableChanges = getMergeableContent();

src/persisters/common/create.ts

@@ -270,7 +270,10 @@ export const createCustomPersister = <
     >,
   ): Promise<Persister<Persist>> => {
     /*! istanbul ignore else */
-    if (status != StatusValues.Loading) {
+    if (
+      status != StatusValues.Loading ||
+      ((isSynchronizer as 0 | 1) && !isUndefined(changes))
+    ) {
       setStatus(StatusValues.Saving);
       saves++;
       await schedule(async () => {

test/unit/synchronizers/synchronizers.test.ts

@@ -548,6 +548,29 @@ describe.each([
         ]);
       });
 
+      test('mutating listener deletions during sync are propagated back', async () => {
+        store2.addRowListener(
+          'tasks',
+          null,
+          (_store, _tableId, rowId) => {
+            if (store2.getCell('tasks', rowId, 'pending') === false) {
+              store2.delRow('tasks', rowId);
+            }
+          },
+          true,
+        );
+
+        store1.setRow('tasks', 'task1', {
+          pending: false,
+          title: 'demo',
+        });
+
+        await pause(synchronizable.pauseMilliseconds);
+
+        expect(store1.getTable('tasks')).toEqual({});
+        expect(store2.getTable('tasks')).toEqual({});
+      });
+
       test('deleted cell', async () => {
         store1.setContent([
           {t1: {r1: {c1: 1, c2: 2}, r2: {c2: 2}}, t2: {r2: {c2: 2}}},