Subscriptions

Subscriptions are how application code reacts to causl's commits. A subscription registers an observer against a node; the engine guarantees that, for each commit in which that node's value changes, the observer fires exactly once, with the new value and the new GraphTime. That sentence is the whole contract. Everything else on this page is the fine print every production adopter eventually needs: how observers are ordered, what happens when you subscribe from inside a compute, what unsubscribe really does, when transient observers drop, and where thrown observers go.

This page covers graph.subscribe(node, observer). For commit-level firehose, see the graph.subscribeCommits section below; for selector-style projections without a derived node, see the subscribeReads section below; for React, see the React tutorial.

1. The basic shape

import { createCausl, input } from '@causl/core';

const graph = createCausl();
const count = input(graph, 0);

const unsubscribe = graph.subscribe(count, (value, time) => {
  console.log(`count = ${value} at ${time}`);
});
// fires synchronously: "count = 0 at 0"

graph.commit(() => {
  graph.set(count, 5);
});
// fires from Phase G:   "count = 5 at 1"

unsubscribe();

Three things happen at subscribe():

The engine validates that node exists. Disposed or unknown nodes throw UnknownNodeError / NodeDisposedError before bookkeeping is allocated.
The observer fires once synchronously with the node's current value and the current GraphTime. This initial fire is wrapped in a try/catch so a faulty observer cannot abort registration.
The engine returns an Unsubscribe closure that drops the entry from the per-node index. Calling it twice is a documented no-op.

After registration, the observer is the engine's responsibility until you call the returned unsubscribe or the node is disposed.

2. When observers fire: Phase G

Every successful graph.commit() walks the eight-phase pipeline laid out in SPEC §5.1. Subscribers do not run in the middle of that walk; they run in a designated late phase, Phase G, after the engine has assembled a frozen Commit record (Phase E), appended it to history (Phase F), and refreshed the commitLog and any commit-metadata derived nodes (Phases F.4 and F.5).

By the time Phase G runs, three invariants hold:

The commit is durable. If your observer reads graph.commitLog or any derived value, it sees the post-commit state. There is no fractional time at which Phase G could observe a half-applied write.
The changed set is fixed. Equality-cutoff has already removed any node whose new value is Object.is-equal to its previous value. Subscribers fire only when the value actually moved.
Every observed node fires once. A commit that writes to ten inputs and propagates through a node observed by one subscriber will fire that subscriber exactly once, not ten times.

2.1 Fire ordering per node

When multiple observers are registered on the same node, the engine fires them in insertion order: the order in which subscribe() was called for that node. This is the order the per-node index preserves (a JavaScript Set's iteration order, which is insertion order by the language specification), and it is the order Phase G's dispatch loop walks.

const log: string[] = [];
graph.subscribe(count, v => log.push(`A:${v}`));
graph.subscribe(count, v => log.push(`B:${v}`));
graph.subscribe(count, v => log.push(`C:${v}`));

graph.commit(() => { graph.set(count, 1) });
// log[3..6] === ['A:1', 'B:1', 'C:1']

Across different nodes within a single commit, the firing order is implementation- defined; do not write code that depends on observer A on node X firing before observer B on node Y. The only contract is per-node insertion order. If your application needs a cross-node ordering guarantee, use subscribeMany([X, Y], ...) — a single observer registered against both nodes fires once per commit and receives the value tuple atomically.

Detailed explanation. The reason per-node insertion order is the contract and not, say, lexicographic order by callback identity is that the per-node index (subscriptionsByNode: Map<NodeId, Set<SubscriptionEntry>>) is derived from a single insertion-ordered Set. Any ordering scheme that wasn't "the order things were inserted" would require a parallel sort, which would cost O(n log n) on every subscribe() call and produce nothing the application couldn't achieve by registering observers in the desired order itself.

3. The subscribe-inside-compute hazard (H3)

A derived(graph, g => ...) compute callback runs inside Phase D, the Kahn-topology recompute. Inside that callback, the engine is the only thing reading and writing engine state. Calling graph.subscribe() from inside a derived compute mutates the subscription registry mid-pipeline. The engine cannot promise sensible behaviour for that subscriber: the synchronous initial fire happens at a fractional time the rest of the pipeline has not reached, and the entry may or may not be observed by Phase G depending on when in the dispatch loop the registration landed.

This is hazard H3 in causl's hazard taxonomy. The mitigation is categorical: do not call subscribe, unsubscribe, or commit from a derived compute. Computes are pure functions of their reads. If you need a subscription that depends on derived state, register it in application code at mount time (the React adapter handles this for you), or register it from inside a subscriber callback — Phase G is a safe place to call subscribe(), because the in-flight commit has already published.

// WRONG — subscribes inside a compute. H3 hazard.
const bad = derived(graph, g => {
  graph.subscribe(otherNode, v => console.log(v));  // do not do this
  return g.read(input1) * 2;
});

// RIGHT — subscribe at the application boundary, read inside the compute.
const good = derived(graph, g => g.read(input1) * 2);
graph.subscribe(good, v => console.log(v));

Calling graph.commit() from inside a derived compute is a related but stricter violation: the engine throws CommitInProgressError at runtime, and the @causl/checker static pass causl/commit-in-subscribe flags it at compile time. subscribe from a compute is not currently a runtime error — the engine tolerates the call and lets the entry land — but the resulting fire ordering is unspecified, so the static linter will eventually grow a warning here too.

4. Unsubscribe semantics

subscribe() returns an Unsubscribe, a zero-argument function whose job is to drop the entry from the engine's bookkeeping. The semantics are deliberately small:

Idempotent. Calling the returned closure twice is a no-op. The second call finds the entry already gone and exits without touching any counters.
Synchronous. After unsubscribe() returns, the observer will not fire again, including for any commit that started before the unsubscribe call landed — Phase G's dispatch loop checks live entries each pass.
Per-registration. Each subscribe() call produces its own independent Unsubscribe. Re-registering the same observer function against the same node creates a second entry, and you get a second Unsubscribe to clean it up.
Lifecycle-bound. A subscription is also auto-cleaned when its target node is disposed via graph.dispose(node). Any subsequent read / subscribe / tx.set against the disposed node throws NodeDisposedError.

const handle = graph.subscribe(count, observer);
handle();   // observer detached
handle();   // no-op, no double-decrement, no throw

graph.dispose(count);
// any retained handle is now a stale closure; calling it
// is still a no-op, but you cannot resubscribe to `count`.

Long-running processes and subscriber leaks. Every long-running causl application should pair subscriptions with a clear disposal scope. The @causl/checker SubscribeWithoutDispose pass enforces this statically against the IR exported by graph.exportModel(): every IRSubscribe must have a matching IRDispose reachable in the same scope, or the scope must be marked infinite: true. In React, the useCauslNode hook handles this for you (the effect's cleanup arm fires the unsubscribe); in raw-engine scripts you are responsible for the pairing.

5. Transient subscribers (one-shot observers)

Sometimes you want an observer to fire once and then go away — "wake me when the loading flag flips, then stop watching." The { transient: true } option produces this shape:

graph.subscribe(loading, (v, t) => {
  if (v === false) console.log(`finished loading at ${t}`);
}, { transient: true });

The transient flag changes the lifecycle, not the semantics:

The synchronous initial fire happens normally. It does not consume the one-shot slot — only the next Phase G fire does.
On the first commit in which the observed node changes, Phase G fires the observer once as usual, then adds the entry to the engine's pendingTransientDrops set.
At the end of the commit, Phase H's drain (in a finally arm of the commit body, so the drop still happens even if a later observer in the same commit threw) removes every entry in pendingTransientDrops from the subscription index.

After the drop, the entry is gone: subsequent commits do not fire the observer, and the Unsubscribe closure originally returned at registration is a no-op. This is the "transient subscribers drop in Phase H" behaviour the SPEC names — the auto-dispose is scheduled by Phase G and applied by Phase H so it cannot tear down state the in-flight dispatch loop is still walking.

Transient observers compose with subscribeMany: registering a multi-node group with { transient: true } auto-disposes the entire group after the first commit in which any member of the group changed. The whole group is the unit of one-shot lifetime; you cannot half-drop it.

6. The onObserverError contract

Observers are user code. User code throws. The engine cannot let an observer's exception tear down the pipeline — Phase G is past the point where rollback is meaningful (the commit has already published) and observer faults must not bring down other subscribers.

The contract per SPEC §15.3.2 is:

Every observer dispatch is wrapped in a try/catch.
A thrown error is routed to the onObserverError hook configured on createCausl({ onObserverError }).
The hook receives the thrown value and an ObserverErrorContext tagging the call site.
Dispatch continues with the next observer.

import { createCausl } from '@causl/core';

const graph = createCausl({
  onObserverError(error, ctx) {
    // ctx.source identifies where the throw originated
    console.error(`[causl] observer threw from ${ctx.source}`, error);
    metrics.increment('causl.observer_error', { source: ctx.source });
  },
});

The ObserverErrorContext.source tag is a closed discriminated union. There are six tags:

source	fires from
`'subscribe-initial'`	the synchronous initial fire inside `graph.subscribe()`
`'node-subscriber'`	Phase G per-node observer dispatch
`'commit-subscriber'`	Phase H `subscribeCommits` observer dispatch
`'subscribe-reads-initial'`	the synchronous initial fire inside `graph.subscribeReads()`
`'subscribe-reads'`	Phase G projection-observer dispatch from `subscribeReads`
`'subscribe-reads-projection'`	the projection closure threw during a Phase G re-run; the registration is preserved and the next commit will retry

Per SPEC §5.3, these are call-site labels on the error edges leaving the engine's Publishing state, not states the engine occupies. The set is closed by contract: any new dispatch path must add a row to the union before it ships, so the switch (ctx.source) in your hook stays exhaustive (TypeScript will tell you).

If you do not pass onObserverError, the engine uses console.error as the default. Production deployments typically replace this with a structured logger so observer crashes go into the same telemetry pipeline as everything else. Passing a no-op function silences the channel entirely — the engine is fine with that; it just means observer faults will not surface anywhere.

What the hook is and isn't. The hook is a sink, not a recovery point. Calling graph.commit() or graph.set() from inside onObserverError is the same kind of re-entrancy hazard as calling them from a subscriber, with the same outcome (CommitInProgressError). The hook's only legitimate jobs are logging, metrics, and queuing a deferred action (e.g. queueMicrotask(() => graph.commit(...))). The engine has already paid for the failure by the time you receive it; the hook reports the bill.

7. Putting it together

A subscription is a four-line contract: register, fire once now, fire once per commit that moves the value, drop on unsubscribe-or-dispose. The engine puts a lot of machinery behind that contract — per-node insertion-ordered dispatch, equality-cutoff, transient auto-disposal, observer-error routing — but the contract itself is small enough to teach in a tutorial. The rest of the rules on this page are the ones an adopter eventually hits when their application graph grows past the toy size: do not subscribe from a compute, configure onObserverError, pair every long-lived subscription with an unsubscribe, and use { transient: true } when the observer only needs to wake once.

Cross-references: Error handling, Structuring graphs, React adapter tutorial, @causl/core API reference.