Don't check subscriptions with read only transactions (#1823)

ubik2 · web-flow · commit f7d1d5df9ea9 · 2025-09-25T14:41:05.000-07:00
* Don't check subscriptions with read only transactions Some cleanup to avoid some of our complex nested classes. * Add a flag to Replica.push so that by default, we don't push transactions that are only Claims. * Alter transaction to prevent commit of nothing but claims. Add comment to Replica.push to note that we generally bypass that now (still used in tests and curl). * Remove Replica.push and change Provider.send to directly commit instead. (#1827)
diff --git a/packages/memory/provider.ts b/packages/memory/provider.ts
@@ -35,7 +35,7 @@ import {
 import * as SelectionBuilder from "./selection.ts";
 import * as Memory from "./memory.ts";
 import { refer } from "./reference.ts";
-import { redactCommit } from "./space.ts";
+import { redactCommitData } from "./space.ts";
 import * as Subscription from "./subscription.ts";
 import * as FactModule from "./fact.ts";
 import { setRevision } from "@commontools/memory/selection";
@@ -366,43 +366,50 @@ class MemoryProviderSession<
   }
 
   async commit(commit: Commit<Space>) {
-    // First, check to see if any of our schema queries need to be notified
-    // Any queries that lack access are skipped (with a console log)
-    const [lastId, maxSince, facts] = await this.getSchemaSubscriptionMatches(
-      commit,
-    );
-
-    // We need to remove any classified results from our commit.
-    // The schema subscription has a classification claim, but these don't.
-    const redactedCommit = redactCommit(commit);
-    const [[space, _ignored]] = Object.entries(commit);
+    // We should really only have one item, but it's technically legal to have
+    // multiple transactions in the same commit, so iterate
+    for (
+      const item of SelectionBuilder.iterate<{ is: Memory.CommitData }>(commit)
+    ) {
+      // We need to remove any classified results from our commit.
+      // The schema subscription has a classification claim, but these don't.
+      const redactedData = redactCommitData(item.value.is);
+      if (Subscription.isTransactionReadOnly(redactedData.transaction)) {
+        continue;
+      }
+      // First, check to see if any of our schema queries need to be notified
+      // Any queries that lack access are skipped (with a console log)
+      const [lastId, maxSince, facts] = await this.getSchemaSubscriptionMatches(
+        redactedData.transaction,
+      );
 
-    const jobIds: InvocationURL<Reference<Subscribe>>[] = [];
-    for (const [id, channels] of this.channels) {
-      if (Subscription.match(redactedCommit, channels)) {
-        jobIds.push(id);
+      const jobIds: InvocationURL<Reference<Subscribe>>[] = [];
+      for (const [id, channels] of this.channels) {
+        if (Subscription.match(redactedData.transaction, channels)) {
+          jobIds.push(id);
+        }
       }
-    }
 
-    if (jobIds.length > 0) {
-      // The client has a subscription to the space's commit log, but our
-      // subscriptions may trigger inclusion of other objects. Add these here.
-      const factsList = [...FactModule.iterate(facts[space as Space])];
-      const enhancedCommit: EnhancedCommit<Space> = {
-        commit: redactedCommit,
-        revisions: this.filterKnownFacts(factsList),
-      };
-
-      for (const id of jobIds) {
-        // this is sent to a standard subscription (application/commit+json)
-        this.perform({
-          the: "task/effect",
-          of: id,
-          is: enhancedCommit,
-        });
+      if (jobIds.length > 0) {
+        // The client has a subscription to the space's commit log, but our
+        // subscriptions may trigger inclusion of other objects. Add these here.
+        const enhancedCommit: EnhancedCommit<Space> = {
+          commit: {
+            [item.of]: { [item.the]: { [item.cause]: { is: redactedData } } },
+          } as Commit<Space>,
+          revisions: this.filterKnownFacts(facts),
+        };
+
+        for (const id of jobIds) {
+          // this is sent to a standard subscription (application/commit+json)
+          this.perform({
+            the: "task/effect",
+            of: id,
+            is: enhancedCommit,
+          });
+        }
       }
     }
-
     return { ok: {} };
   }
 
@@ -470,17 +477,18 @@ class MemoryProviderSession<
   }
 
   private async getSchemaSubscriptionMatches<Space extends MemorySpace>(
-    commit: Commit<Space>,
-  ): Promise<[JobId | undefined, number, Selection<Space>]> {
+    transaction: Transaction<Space>,
+  ): Promise<[JobId | undefined, number, Revision<Fact>[]]> {
     const schemaMatches = new Map<string, Revision<Fact>>();
+    const space = transaction.sub;
     let maxSince = -1;
     let lastId;
-    const [[spaceStr, _attributes]] = Object.entries(commit);
-    const space = spaceStr as Space;
     // Eventually, we should support multiple spaces, but currently the since handling is per-space
     // Our websockets are also per-space, so there's larger issues involved.
     for (const [id, subscription] of this.schemaChannels) {
-      if (Subscription.match(commit, subscription.watchedObjects)) {
+      if (
+        Subscription.match(transaction, subscription.watchedObjects)
+      ) {
         // Re-run our original query, but not as a subscription
         const newArgs = { ...subscription.invocation.args, subscribe: false };
         const newInvocation = { ...subscription.invocation, args: newArgs };
@@ -519,11 +527,7 @@ class MemoryProviderSession<
         maxSince = since > maxSince ? since : maxSince;
       }
     }
-    const selection = SelectionBuilder.from(
-      schemaMatches.values().map((item) => [item, item.since]),
-    );
-    const selectionSpace = { [space]: selection } as Selection<Space>;
-    return [lastId, maxSince, selectionSpace];
+    return [lastId, maxSince, [...schemaMatches.values()]];
   }
 }
 
diff --git a/packages/memory/space.ts b/packages/memory/space.ts
@@ -992,20 +992,9 @@ export function getClassifications(
   return classifications;
 }
 
-export function redactCommit(commit: Commit): Commit {
-  const newCommit = {};
-  for (const item of iterate(commit)) {
-    const redactedData = redactCommitData(item.value.is);
-    set(newCommit, item.of, item.the, item.cause, { is: redactedData });
-  }
-  return newCommit;
-}
-
 // Return the item with any classified results and the labels removed.
-export function redactCommitData(
-  commitData?: CommitData,
-): CommitData | undefined {
-  if (commitData === undefined || commitData.labels === undefined) {
+export function redactCommitData(commitData: CommitData): CommitData {
+  if (commitData.labels === undefined) {
     return commitData;
   }
   // Make a copy of the transaction with no changes
diff --git a/packages/memory/subscription.ts b/packages/memory/subscription.ts
@@ -1,39 +1,37 @@
 import {
   Cause,
-  Commit,
   Entity,
   MemorySpace,
   SchemaSelector,
   Selector,
   The,
+  Transaction,
 } from "./interface.ts";
+import { iterate } from "./selection.ts";
 import { COMMIT_LOG_TYPE } from "./commit.ts";
 
-export const match = (commit: Commit, watched: Set<string>) => {
-  for (const at of Object.keys(commit) as MemorySpace[]) {
-    const commitObj = commit[at][COMMIT_LOG_TYPE] ?? {};
-    for (const { is: { transaction } } of Object.values(commitObj)) {
-      // If commit on this space are watched we have a match
-      if (matchAddress(watched, { the: COMMIT_LOG_TYPE, of: at, at })) {
-        return true;
-      }
+export const match = (
+  transaction: Transaction<MemorySpace>,
+  watched: Set<string>,
+) => {
+  const space = transaction.sub;
+  // If commit on this space are watched we have a match
+  if (
+    matchAddress(watched, { at: space, of: space, the: COMMIT_LOG_TYPE })
+  ) {
+    return true;
+  }
 
-      // Otherwise we consider individual in the commit transaction to figure
-      // out if we have a match.
-      for (const [of, attributes] of Object.entries(transaction.args.changes)) {
-        for (const [the, changes] of Object.entries(attributes)) {
-          for (const change of Object.values(changes)) {
-            // If `change == true` we simply confirm that state has not changed
-            // so we don't need to notify those subscribers.
-            if (
-              change !== true &&
-              matchAddress(watched, { at: transaction.sub, the, of })
-            ) {
-              return true;
-            }
-          }
-        }
-      }
+  // Otherwise we consider individual in the commit transaction to figure
+  // out if we have a match.
+  for (const fact of iterate(transaction.args.changes)) {
+    // If `fact.value == true` we simply confirm that state has not changed
+    // so we don't need to notify those subscribers.
+    if (
+      fact.value !== true &&
+      matchAddress(watched, { at: space, of: fact.of, the: fact.the })
+    ) {
+      return true;
     }
   }
 
@@ -42,9 +40,9 @@ export const match = (commit: Commit, watched: Set<string>) => {
 
 const matchAddress = (
   watched: Set<string>,
-  { at, the, of }: { the: string; of: string; at: MemorySpace },
+  { at, of, the }: { at: MemorySpace; of: string; the: string },
 ) =>
-  watched.has(formatAddress({ at, the, of })) ||
+  watched.has(formatAddress({ at, of, the })) ||
   watched.has(formatAddress({ at, the })) ||
   watched.has(formatAddress({ at, of })) ||
   watched.has(formatAddress({ at }));
@@ -98,6 +96,17 @@ export const fromSelector = function* (selector: Selector | SchemaSelector) {
   }
 };
 
+export function isTransactionReadOnly(
+  transaction: Transaction<MemorySpace>,
+): boolean {
+  for (const fact of iterate(transaction.args.changes)) {
+    if (fact.value !== true) {
+      return false;
+    }
+  }
+  return true;
+}
+
 export const formatAddress = (
   { at = "_", of = "_", the = "_" }: {
     at?: string;
diff --git a/packages/runner/src/storage/cache.ts b/packages/runner/src/storage/cache.ts
@@ -971,73 +971,6 @@ export class Replica {
     this.syncTimer = setTimeout(this.pull, this.syncTimeout);
   }
 
-  /**
-   * Attempts to commit and push changes to the remote. It optimistically
-   * updates local state so that subsequent commits can be made without having
-   * awaiting for each commit to succeed. However if commit fails all the local
-   * changes made will be reverted back to the last merged state.
-   *
-   * ⚠️ Please note that if commits stack up e.g by incrementing value of the
-   * same entity, rejected commit will ripple through stack as the changes there
-   * would assume state that is rejected.
-   */
-  async push(
-    changes: (Assert | Retract | Claim)[],
-  ): Promise<
-    Result<
-      Commit,
-      PushError
-    >
-  > {
-    // Generate the selectors for the various change objects
-    const loadArgs: [BaseMemoryAddress, SchemaPathSelector?][] = changes.map((
-      change,
-    ) => {
-      const schema = generateSchemaFromLabels(change);
-      const selector = schema === undefined
-        ? undefined
-        : { path: [], schemaContext: { schema: schema, rootSchema: schema } };
-      return [{ id: change.of, type: change.the }, selector];
-    });
-    // First we pull all the affected entries into heap so we can build a
-    // transaction that is aware of latest state.
-    const { error } = await this.load(loadArgs);
-    if (error) {
-      return { error };
-    } else {
-      // Collect facts so that we can derive desired state and a corresponding
-      // transaction
-      const facts: Fact[] = [];
-      const claims: Invariant[] = [];
-      for (const { the, of, is, claim } of changes) {
-        const fact = this.get({ id: of, type: the });
-
-        if (claim) {
-          claims.push(claimState(fact!));
-        } else if (is === undefined) {
-          // If `is` is `undefined` we want to retract the fact.
-          // If local `is` in the local state is also `undefined` desired state
-          // matches current state in which case we omit this change from the
-          // transaction, otherwise we retract fact.
-          if (fact?.is !== undefined) {
-            facts.push(retract(fact));
-          }
-        } else {
-          facts.push(assert({
-            the,
-            of,
-            is,
-            // If fact has no `cause` it is unclaimed fact.
-            cause: fact?.cause ? fact : null,
-          }));
-        }
-      }
-
-      // These push transaction that will commit desired state to a remote.
-      return this.commit({ facts, claims });
-    }
-  }
-
   async commit(transaction: ITransaction, source?: IStorageTransaction) {
     const { facts, claims } = transaction;
     const changes = Differential.create().update(this, facts);
@@ -1616,6 +1549,7 @@ class ProviderConnection implements IStorageProvider {
   get<T = any>(uri: URI): StorageValue<T> | undefined {
     return this.provider.get(uri);
   }
+
   send<T = any>(
     batch: { uri: URI; value: StorageValue<T> }[],
   ) {
@@ -1760,6 +1694,8 @@ export class Provider implements IStorageProvider {
     return entity?.is as StorageValue<T> | undefined;
   }
 
+  // This is mostly just used by tests and tools, since the transactions will
+  // directly commit their results.
   async send<T = any>(
     batch: { uri: URI; value: StorageValue<T> }[],
   ): Promise<
@@ -1768,7 +1704,9 @@ export class Provider implements IStorageProvider {
     const { the, workspace } = this;
     const LABEL_TYPE = "application/label+json" as const;
 
-    const changes = [];
+    // Collect facts so that we can derive desired state and a corresponding
+    // transaction
+    const facts: Fact[] = [];
     for (const { uri, value } of batch) {
       const content = value.value !== undefined
         ? JSON.stringify({ value: value.value, source: value.source })
@@ -1779,29 +1717,37 @@ export class Provider implements IStorageProvider {
         if (content !== undefined) {
           // ⚠️ We do JSON roundtrips to strip off the undefined values that
           // cause problems with serialization.
-          changes.push({ of: uri, the, is: JSON.parse(content) as JSONValue });
+          facts.push(assert({
+            the,
+            of: uri,
+            is: JSON.parse(content) as JSONValue,
+            // If fact has no `cause` it is unclaimed fact.
+            cause: current?.cause ? current : null,
+          }));
         } else {
-          changes.push({ of: uri, the });
+          facts.push(retract(current as Consumer.Assertion));
         }
       }
       if (value.labels !== undefined) {
         const currentLabel = workspace.get({ id: uri, type: LABEL_TYPE });
         if (!deepEqual(currentLabel?.is, value.labels)) {
           if (value.labels !== undefined) {
-            changes.push({
-              of: uri,
+            facts.push(assert({
               the: LABEL_TYPE,
+              of: uri,
               is: value.labels as JSONValue,
-            });
+              // If fact has no `cause` it is unclaimed fact.
+              cause: currentLabel?.cause ? currentLabel : null,
+            }));
           } else {
-            changes.push({ of: uri, the: LABEL_TYPE });
+            facts.push(retract(currentLabel as Consumer.Assertion));
           }
         }
       }
     }
-
-    if (changes.length > 0) {
-      const result = await this.workspace.push(changes);
+    // If we don't have any writes, don't bother sending it.
+    if (facts.length > 0) {
+      const result = await this.workspace.commit({ facts, claims: [] });
       return result.error ? result : { ok: {} };
     } else {
       return { ok: {} };
diff --git a/packages/runner/src/storage/transaction.ts b/packages/runner/src/storage/transaction.ts
