feat(runner): Add onCommit callback support for Cell and Stream operations

packages/runner/src/cell.ts

@@ -159,8 +159,14 @@ import { ContextualFlowControl } from "./cfc.ts";
 declare module "@commontools/api" {
   interface Cell<T> {
     get(): Readonly<T>;
-    set(value: Cellify<T> | T): void;
-    send(value: Cellify<T> | T): void;
+    set(
+      value: Cellify<T> | T,
+      onCommit?: (tx: IExtendedStorageTransaction) => void,
+    ): void;
+    send(
+      value: Cellify<T> | T,
+      onCommit?: (tx: IExtendedStorageTransaction) => void,
+    ): void;
     update<V extends Cellify<Partial<T> | Partial<T>>>(
       values: V extends object ? V : never,
     ): void;
@@ -249,7 +255,7 @@ declare module "@commontools/api" {
   }
 
   interface Stream<T> {
-    send(event: T): void;
+    send(event: T, onCommit?: (tx: IExtendedStorageTransaction) => void): void;
     sink(callback: (event: Readonly<T>) => Cancel | undefined | void): Cancel;
     sync(): Promise<Stream<T>> | Stream<T>;
     getRaw(options?: IReadOptions): any;
@@ -350,11 +356,11 @@ class StreamCell<T> implements Stream<T> {
     return this.link.rootSchema;
   }
 
-  send(event: T): void {
+  send(event: T, onCommit?: (tx: IExtendedStorageTransaction) => void): void {
     event = convertCellsToLinks(event) as T;
 
     // Use runtime from doc if available
-    this.runtime.scheduler.queueEvent(this.link, event);
+    this.runtime.scheduler.queueEvent(this.link, event, undefined, onCommit);
 
     this.cleanup?.();
     const [cancel, addCancel] = useCancelGroup();
@@ -432,7 +438,10 @@ export class RegularCell<T> implements Cell<T> {
     return validateAndTransform(this.runtime, this.tx, this.link, this.synced);
   }
 
-  set(newValue: Cellify<T> | T): void {
+  set(
+    newValue: Cellify<T> | T,
+    onCommit?: (tx: IExtendedStorageTransaction) => void,
+  ): void {
     if (!this.tx) throw new Error("Transaction required for set");
 
     // No await for the sync, just kicking this off, so we have the data to
@@ -447,10 +456,18 @@ export class RegularCell<T> implements Cell<T> {
       newValue,
       getTopFrame()?.cause,
     );
+
+    // Register commit callback if provided
+    if (onCommit) {
+      this.tx.addCommitCallback(onCommit);
+    }
   }
 
-  send(newValue: Cellify<T> | T): void {
-    this.set(newValue);
+  send(
+    newValue: Cellify<T> | T,
+    onCommit?: (tx: IExtendedStorageTransaction) => void,
+  ): void {
+    this.set(newValue, onCommit);
   }
 
   update<V extends Cellify<Partial<T> | Partial<T>>>(

packages/runner/src/runtime.ts

@@ -209,7 +209,12 @@ export interface IScheduler {
   unsubscribe(action: Action): void;
   onConsole(fn: ConsoleHandler): void;
   onError(fn: ErrorHandler): void;
-  queueEvent(eventRef: NormalizedFullLink, event: any): void;
+  queueEvent(
+    eventRef: NormalizedFullLink,
+    event: any,
+    retries?: number,
+    onCommit?: (tx: IExtendedStorageTransaction) => void,
+  ): void;
   addEventHandler(handler: EventHandler, ref: NormalizedFullLink): Cancel;
   runningPromise: Promise<unknown> | undefined;
 }

packages/runner/src/scheduler.ts

@@ -82,7 +82,11 @@ const DEFAULT_RETRIES_FOR_EVENTS = 5;
 const MAX_RETRIES_FOR_REACTIVE = 10;
 
 export class Scheduler implements IScheduler {
-  private eventQueue: { action: Action; retriesLeft: number }[] = [];
+  private eventQueue: {
+    action: Action;
+    retriesLeft: number;
+    onCommit?: (tx: IExtendedStorageTransaction) => void;
+  }[] = [];
   private eventHandlers: [NormalizedFullLink, EventHandler][] = [];
 
   private pending = new Set<Action>();
@@ -313,13 +317,15 @@ export class Scheduler implements IScheduler {
     eventLink: NormalizedFullLink,
     event: any,
     retries: number = DEFAULT_RETRIES_FOR_EVENTS,
+    onCommit?: (tx: IExtendedStorageTransaction) => void,
   ): void {
     for (const [link, handler] of this.eventHandlers) {
       if (areNormalizedLinksSame(link, eventLink)) {
         this.queueExecution();
         this.eventQueue.push({
           action: (tx: IExtendedStorageTransaction) => handler(tx, event),
           retriesLeft: retries,
+          onCommit,
         });
       }
     }
@@ -480,7 +486,7 @@ export class Scheduler implements IScheduler {
     // the topological sort in the right way.
     const event = this.eventQueue.shift();
     if (event) {
-      const { action, retriesLeft } = event;
+      const { action, retriesLeft, onCommit } = event;
       this.runtime.telemetry.submit({
         type: "scheduler.invocation",
         handler: action,
@@ -496,10 +502,23 @@ export class Scheduler implements IScheduler {
             // enough, especially for a series of event that act on the same
             // conflicting data.
             if (error && retriesLeft > 0) {
-              this.eventQueue.unshift({ action, retriesLeft: retriesLeft - 1 });
+              this.eventQueue.unshift({
+                action,
+                retriesLeft: retriesLeft - 1,
+                onCommit,
+              });
               // Ensure the re-queued event gets processed even if the scheduler
               // finished this cycle before the commit completed.
               this.queueExecution();
+            } else if (onCommit) {
+              // Call commit callback when:
+              // - Commit succeeds (!error), OR
+              // - Commit fails but we're out of retries (retriesLeft === 0)
+              try {
+                onCommit(tx);
+              } catch (callbackError) {
+                logger.error("Error in event commit callback:", callbackError);
+              }
             }
           });
         }

packages/runner/src/storage/extended-storage-transaction.ts

@@ -42,6 +42,10 @@ const logResult = (
 };
 
 export class ExtendedStorageTransaction implements IExtendedStorageTransaction {
+  private commitCallbacks = new Set<
+    (tx: IExtendedStorageTransaction) => void
+  >();
+
   constructor(public tx: IStorageTransaction) {}
 
   get journal(): ITransactionJournal {
@@ -175,6 +179,38 @@ export class ExtendedStorageTransaction implements IExtendedStorageTransaction {
   }
 
   commit(): Promise<Result<Unit, CommitError>> {
-    return this.tx.commit();
+    const promise = this.tx.commit();
+
+    // Call commit callbacks after commit completes (success or failure) Note
+    // that promise always resolves, even if the commit fails, in which case it
+    // passes an error message as result. An exception here would be an internal
+    // error that should propagate.
+    promise.then((_result) => {
+      // Call all callbacks, wrapping each in try/catch to prevent one
+      // failing callback from breaking others
+      for (const callback of this.commitCallbacks) {
+        try {
+          callback(this);
+        } catch (error) {
+          logger.error("Error in commit callback:", error);
+        }
+      }
+    });
+
+    return promise;
+  }
+
+  /**
+   * Add a callback to be called when the transaction commit completes.
+   * The callback receives the transaction as a parameter and is called
+   * regardless of whether the commit succeeded or failed.
+   *
+   * Note: Callbacks are called synchronously after commit completes.
+   * If a callback throws, the error is logged but doesn't affect other callbacks.
+   *
+   * @param callback - Function to call after commit
+   */
+  addCommitCallback(callback: (tx: IExtendedStorageTransaction) => void): void {
+    this.commitCallbacks.add(callback);
   }
 }

packages/runner/src/storage/interface.ts

@@ -496,6 +496,18 @@ export interface IStorageTransaction {
 export interface IExtendedStorageTransaction extends IStorageTransaction {
   tx: IStorageTransaction;
 
+  /**
+   * Add a callback to be called when the transaction commit completes.
+   * The callback receives the transaction as a parameter and is called
+   * regardless of whether the commit succeeded or failed.
+   *
+   * Note: Callbacks are called synchronously after commit completes.
+   * If a callback throws, the error is logged but doesn't affect other callbacks.
+   *
+   * @param callback - Function to call after commit
+   */
+  addCommitCallback(callback: (tx: IExtendedStorageTransaction) => void): void;
+
   /**
    * Reads a value from a (local) memory address and throws on error, except for
    * `NotFoundError` which is returned as undefined.