Success

All signal.set() calls inside the transaction are committed, and when the outermost level exits, we call flushJobs() once, so our effects rerun only once.

Failure (throw / reject)

All affected signals are restored to the values they had when entering the current transaction level. We do not flush, so invalid intermediate state is never pushed into effects. Downstream computed nodes are marked as stale and will lazily recompute on the next read.

Nested transactions

Each level maintains its own write log.

Equality semantics

Consistent with the core runtime, Object.is(prev, next) means “no change”: no log entry, no scheduling.

if (inAtomic()) recordAtomicWrite(node, prev)

import { markStale } from "./computed.js";
import type { Node } from "./graph.js";

export interface Schedulable { run(): void; disposed?: boolean }

// Internal node shape used by signal/computed
export type InternalNode<T = unknown> = { value: T };

// Write log for atomic transactions
type WriteLog = Map<(Node & InternalNode<unknown>), unknown>;

const queue = new Set<Schedulable>();
let scheduled = false;

// > 0 means we are inside batch/transaction mode (delay microtask flushing)
let batchDepth = 0;

// Atomic transaction depth and log stack
let atomicDepth = 0;
const atomicLogs: WriteLog[] = [];

// Mute scheduling during rollback to prevent scheduleJob from creating new work
let muted = 0;

export function scheduleJob(job: Schedulable) {
  if (job.disposed) return;
  queue.add(job);
  if (!scheduled && batchDepth === 0) {
    scheduled = true;
    queueMicrotask(flushJobs);
  }
}

export function batch<T>(fn: () => T): T {
  batchDepth++;
  try {
    return fn();
  } finally {
    batchDepth--;
    if (batchDepth === 0) flushJobs();
  }
}

// Promise detection
function isPromiseLike<T = unknown>(v: any): v is PromiseLike<T> {
  return v != null && typeof v.then === "function";
}

export function transaction<T>(fn: () => T): T;
export function transaction<T>(fn: () => Promise<T>): Promise<T>;
export function transaction<T>(fn: () => T | Promise<T>): T | Promise<T> {
  batchDepth++;
  try {
    const out = fn();
    if (isPromiseLike<T>(out)) {
      return Promise.resolve(out).finally(() => {
        batchDepth--;
        if (batchDepth === 0) flushJobs();
      });
    }
    batchDepth--;
    if (batchDepth === 0) flushJobs();
    return out as T;
  } catch (e) {
    batchDepth--;
    if (batchDepth === 0) flushJobs();
    throw e;
  }
}

// Atomic transaction (with rollback)
export function inAtomic() {
  return atomicDepth > 0;
}

// Record the "first write in this level"; called by signal.set() when a write is confirmed
export function recordAtomicWrite<T>(node: Node & InternalNode<T>, prevValue: T) {
  const log = atomicLogs[atomicLogs.length - 1];
  if (!log) return; // safety guard: no active atomic layer
  if (!log.has(node)) log.set(node, prevValue);
}

function writeNodeValue<T>(node: Node & InternalNode<T>, v: T) {
  if ("value" in node) (node as { value: T }).value = v;
}

function mergeChildIntoParent(child: WriteLog, parent: WriteLog) {
  for (const [node, prev] of child) {
    if (!parent.has(node)) parent.set(node, prev);
  }
}

export function atomic<T>(fn: () => T): T;
export function atomic<T>(fn: () => Promise<T>): Promise<T>;
export function atomic<T>(fn: () => T | Promise<T>): T | Promise<T> {
  // Enter atomic layer: suppress flushing (shared batchDepth), start write logging
  batchDepth++;
  atomicDepth++;
  atomicLogs.push(new Map<(Node & InternalNode<unknown>), unknown>());

  const exitCommit = () => {
    const log = atomicLogs.pop()!;
    atomicDepth--;
    // Inner success -> merge first-seen old values into parent
    if (atomicDepth > 0) {
      mergeChildIntoParent(log, atomicLogs[atomicLogs.length - 1]!);
    }
    // Only flush when the outermost layer exits
    batchDepth--;
    if (batchDepth === 0) flushJobs();
  };

  const exitRollback = () => {
    const log = atomicLogs.pop()!;
    atomicDepth--;
    // Silent rollback: avoid scheduling while restoring values
    muted++;
    try {
      for (const [node, prev] of log) {
        writeNodeValue(node, prev);
        if ((node as Node).kind === "signal") {
          for (const sub of (node as Node).subs) {
            if (sub.kind === "computed") markStale(sub);
            // sub.kind === "effect" does not need scheduling
            // muted blocks it, and we also do not flush afterward
          }
        }
      }
      queue.clear(); // clear jobs created during this level
      scheduled = false;
    } finally {
      muted--;
    }
    // No flush on failure; just exit batch/atomic depth
    batchDepth--;
  };

  try {
    const out = fn();
    if (isPromiseLike<T>(out)) {
      return Promise.resolve(out).then(
        (v) => { exitCommit(); return v; },
        (err) => { exitRollback(); throw err; }
      );
    }
    // Synchronous success
    exitCommit();
    return out as T;
  } catch (e) {
    // Synchronous failure -> rollback
    exitRollback();
    throw e;
  }
}

export function flushSync() {
  if (!scheduled && queue.size === 0) return;
  flushJobs();
}

function flushJobs() {
  scheduled = false;
  let guard = 0;
  while (queue.size) {
    const list = Array.from(queue);
    queue.clear();
    for (const job of list) job.run();
    if (++guard > 10000) throw new Error("Infinite update loop");
  }
}

import { markStale } from "./computed.js";
import { link, track, unlink, type Node } from "./graph.js";
import { SymbolRegistry as Effects } from "./registry.js";
import { inAtomic, recordAtomicWrite, type InternalNode } from "./scheduler.js";

type Comparator<T> = (a: T, b: T) => boolean;
const defaultEquals = Object.is;

export function signal<T>(initial: T, equals: Comparator<T> = defaultEquals) {
  const node: Node & InternalNode<T> & { kind: "signal"; equals: Comparator<T> } = {
    kind: "signal",
    deps: new Set(),
    subs: new Set(),
    value: initial,
    equals,
  };

  const get = () => {
    track(node);
    return node.value;
  };

  const set = (next: T | ((prev: T) => T)) => {
    const prev = node.value;
    const nxtVal = typeof next === "function" ? (next as (p: T) => T)(node.value) : next;
    if (node.equals(node.value, nxtVal)) return;

    // Atomic hook: record the previous value only when the write is confirmed,
    // and only the first time this level touches the node
    if (inAtomic()) recordAtomicWrite(node, prev);

    // Perform the actual write
    node.value = nxtVal;

    // No downstream subscribers -> exit early to avoid unnecessary work
    if (node.subs.size === 0) return;

    // Has downstream subscribers -> follow the original propagation logic
    for (const sub of node.subs) {
      if (sub.kind === "effect") {
        Effects.get(sub)?.schedule();
      } else if (sub.kind === "computed") {
        markStale(sub);
      }
    }
  };

  const subscribe = (observer: Node) => {
    if (observer.kind === "signal") {
      throw new Error("A signal cannot subscribe to another node");
    }
    link(observer, node);
    return () => unlink(observer, node);
  };

  return { get, set, subscribe, peek: () => node.value };
}

const a = signal(0);
const b = signal(0);

await atomic(async () => { // outer
  a.set(1); // OK
  try {
    await atomic(async () => { // inner
      b.set(1);
      throw new Error("boom"); // inner fails -> rollback b to 0
    });
  } catch {}
  // At this point: a = 1, b = 0
}); // outer succeeds -> one flush

const a = signal(0);
const b = signal(0);

try {
  await atomic(async () => {
    a.set(1);
    await Promise.resolve();
    b.set(2);
    throw new Error("oops"); // entire transaction fails -> rollback both a and b
  });
} catch {}

// a = 0, b = 0
// and there was no flush in this transaction

import { useState, useEffect } from "react";
import { signal } from "../core/signal.js";
import { atomic } from "../core/scheduler.js";
import {
  useSignalValue,
  useSignalState,
  useComputed,
} from "../hook/react_adapter.js";

// ---- mock API ----
async function postTitle(v: string, shouldFail = false) {
  await new Promise((r) => setTimeout(r, 300)); // simulate latency
  if (shouldFail) throw new Error("server says no");
  return true;
}

// ---- state ----
const titleSig = signal("Hello");

// for unit test
export type EditorTestProps = { __sig?: ReturnType<typeof signal<string>> };

export function Editor({ __sig }: EditorTestProps = {}) {
  const sig = __sig ?? titleSig; // default: module-scope signal; tests can override it

  const committed = useSignalValue(sig); // read snapshot from external signal
  const [draft, setDraft] = useSignalState(committed); // local draft state
  useEffect(() => setDraft(committed), [committed]);

  const len = useComputed(() => titleSig.get().length); // ✅ hook returns a value

  const [saving, setSaving] = useState(false);
  const [error, setError] = useState<string | null>(null);
  const [shouldFail, setShouldFail] = useState(false);

  const save = async () => {
    setSaving(true);
    setError(null);
    try {
      await atomic(async () => {
        sig.set(draft); // optimistic write (does not flush immediately)
        await postTitle(draft, shouldFail); // may throw -> rollback + no flush
      });
      // Success: only flush when atomic exits, so committed/len update together
    } catch (e: any) {
      setError(e?.message ?? "save failed");
    } finally {
      setSaving(false);
    }
  };

  return (
    <section>
      <input
        value={draft}
        onChange={(e) => setDraft(e.target.value)}
        disabled={saving}
      />
      <button onClick={save} disabled={saving}>
        {saving ? "Saving..." : "Save"}
      </button>
      <label style={{ marginLeft: 8 }}>
        <input
          type="checkbox"
          checked={shouldFail}
          onChange={(e) => setShouldFail(e.target.checked)}
        />
        simulate failure
      </label>

      <hr />
      <p>
        Committed title: <b>{committed}</b>
      </p>
      <p>
        Derived length (computed): <b>{len}</b>
      </p>
      {error && <p style={{ color: "crimson" }}>Error: {error}</p>}
    </section>
  );
}

<script setup lang="ts">
import { ref, watch } from "vue";
import { signal } from "../core/signal.js";
import { atomic } from "../core/scheduler.js";
import { useSignalRef, useComputedRef } from "../hook/vue_adapter.js";

// ---- mock API ----
async function postTitle(v: string, shouldFail = false) {
  await new Promise((r) => setTimeout(r, 300));
  if (shouldFail) throw new Error("server says no");
  return true;
}

// ---- state ----
const titleSig = signal("Hello");
const committed = useSignalRef(titleSig);
const titleLen = useComputedRef(() => titleSig.get().length);

const draft = ref(committed.value);
watch(committed, (v) => (draft.value = v)); // sync draft when external value changes

const saving = ref(false);
const error = ref<string | null>(null);
const shouldFail = ref(false);

async function save() {
  saving.value = true;
  error.value = null;
  try {
    await atomic(async () => {
      // optimistic write, but no immediate flush
      titleSig.set(draft.value);
      await postTitle(draft.value, shouldFail.value); // may throw
    });
    // Success: only flush after atomic exits -> one template update
  } catch (e: any) {
    // Failure: rollback, no flush -> template keeps showing the old value
    error.value = e?.message ?? "save failed";
  } finally {
    saving.value = false;
  }
}
</script>

<template>
  <section>
    <div>
      <label>
        Draft:
        <input v-model="draft" :disabled="saving" />
      </label>
      <button @click="save" :disabled="saving">
        {{ saving ? "Saving..." : "Save" }}
      </button>
      <label style="margin-left: 8px">
        <input type="checkbox" v-model="shouldFail" :disabled="saving" />
        simulate failure
      </label>
    </div>

    <hr />

    <p>
      Committed title: <b>{{ committed }}</b>
    </p>
    <p>
      Derived length (computed): <b>{{ titleLen }}</b>
    </p>
    <p v-if="error" style="color: crimson">Error: {{ error }}</p>
  </section>
</template>