Fix extra sink calls (#991)

* fix extranous sink calls

* extra logging in storage for debugging

Author: seefeldb

runner/src/cell.ts

@@ -20,6 +20,7 @@ import { type EntityId, getDocByEntityId, getEntityId } from "./doc-map.ts";
 import { type Cancel, isCancel, useCancelGroup } from "./cancel.ts";
 import { validateAndTransform } from "./schema.ts";
 import { type Schema } from "@commontools/builder";
+import { compactifyPaths } from "./scheduler.ts";
 
 /**
  * This is the regular Cell interface, generated by DocImpl.asCell().
@@ -555,14 +556,24 @@ function subscribeToReferencedDocs<T>(
 
   // Subscribe to the docs that are read (via logs), call callback on next change.
   const cancel = subscribe((log) => {
+    const newLog = {
+      reads: [],
+      writes: [],
+    } satisfies ReactivityLog;
+
     if (isCancel(cleanup)) cleanup();
     const newValue = validateAndTransform(
       doc,
       path,
       schema,
-      log,
+      newLog,
       rootSchema,
     ) as T;
+    // Copy reads to log _before_ calling the callback, as we're only interested
+    // in dependencies for the initial get, not further cells the callback might
+    // read. The callback is responsible for calling sink on those cells if it
+    // wants to stay updated.
+    log.reads.push(...newLog.reads);
     cleanup = callback(newValue);
   }, initialLog);
 

runner/src/storage.ts

@@ -441,20 +441,20 @@ class StorageImpl implements Storage {
       source: doc.sourceCell?.entityId,
     };
 
-    // 🤔 I'm guessing we should be storting schema here
+    // 🤔 I'm guessing we should be storing schema here
 
     if (JSON.stringify(value) !== JSON.stringify(this.writeValues.get(doc))) {
-      this.writeDependentDocs.set(doc, dependencies);
-      this.writeValues.set(doc, value);
-
-      this._addToBatch([{ doc, type: "storage" }]);
-
       log(() => [
         "prep for storage",
         JSON.stringify(doc.entityId),
         value,
+        this.writeValues.get(doc),
         [...dependencies].map((c) => JSON.stringify(c.entityId)),
       ]);
+      this.writeDependentDocs.set(doc, dependencies);
+      this.writeValues.set(doc, value);
+
+      this._addToBatch([{ doc, type: "storage" }]);
     }
   }
 
@@ -788,12 +788,13 @@ class StorageImpl implements Storage {
       return storage.send(jobs).then((result) => retryOnConflict(result))
         .then((result) => {
           if (result.ok) {
+            log(() => ["storage ok", JSON.stringify(result.ok, null, 2)]);
             // Apply updates from retry, if transaction ultimately succeeded
             updatesFromRetry.forEach(([doc, value]) =>
               this._batchForDoc(doc, value.value, value.source)
             );
           } else if (result.error) {
-            log(() => ["storage error", result.error]);
+            log(() => ["storage error", JSON.stringify(result.error, null, 2)]);
             console.error("storage error", result.error);
           }
           return result;
@@ -803,7 +804,7 @@ class StorageImpl implements Storage {
     // Write all storage jobs to storage, in parallel
     const promiseJobs = [];
     for (const [space, jobs] of storageJobsBySpace.entries()) {
-      promiseJobs.push(process(space, jobs));
+      if (jobs.length) promiseJobs.push(process(space, jobs));
     }
     await Promise.all(promiseJobs);
     log(() => ["storage jobs done"]);