Feature: effect cursor / startAfter for durable replay

[KyleAMathews]

Use Case

We're building a tool where entities wake, run effects, and go idle. On re-wake, the runtime replays effects from persisted manifest entries. The problem: replayed effects fire onEnter for all existing rows that match the query, including rows that were already processed on a previous wake.

The scenario

1. Entity wakes, creates an effect watching a child entity's runs collection
2. Effect's onEnter fires when a child's run completes → sends a "child-done" message
3. Entity goes idle
4. Entity re-wakes (new message arrives)
5. Runtime replays the effect from the manifest
6. Effect fires onEnter again for the already-completed run → duplicate "child-done" message

What we need

A way to tell an effect: "only fire callbacks for mutations that happen after this point." Something like:

createEffect({
  query: (q) => q.from({ runs: childDb.collections.runs }),
  startAfter: cursor, // <-- "skip everything before this point"
  onEnter: (event) => {
    // Only fires for new completions, not ones from previous wakes
  },
})

Current workarounds

• skipInitial: true — too aggressive. Skips ALL initial data, including events that arrived while the entity was idle (which are exactly the ones we need to process).
• Making every effect handler idempotent (check state before acting) — works but pushes complexity to every effect author. We'd prefer the runtime handles this.

The Challenge

TanStack DB doesn't have a global offset/sequence number on collection mutations. Effects work on collection state (rows matching a query), not on an event log. So "start after offset X" doesn't have a natural mapping.

Possible approaches

1. Row-level versioning: Each row gets a monotonic version number on mutation. Effects with startAfter skip rows whose version is ≤ the cursor. New mutations (inserts, updates, deletes) after the cursor fire normally.

2. Subscription-level cursor: The effect subscription could be configured to treat the current snapshot as "already seen" and only fire callbacks for future changes. This is similar to skipInitial but more precise — it skips the specific rows that exist at creation time, not all initial load data.

3. Collection-level epoch: Collections track a mutation counter. startAfter: epoch means "fire onEnter/onExit only for mutations with epoch > startAfter." The runtime would store the epoch when the entity goes idle and pass it back when replaying effects.

4. External filtering in onEnter: The caller filters in the callback (our current workaround). Not ideal but works.

What we're doing with this

Effects are TanStack DB reactive queries that watch entity collections (which are projections of the stream). The entity lifecycle is:

• Wake → preload stream into collections → create/replay effects → process messages → go idle
• On re-wake, effects must distinguish "data that was there last time" from "data that arrived while I was asleep"

We're happy to contribute an implementation if there's alignment on the approach. The existing cursor machinery in live/utils.ts (computeOrderedLoadCursor, requestLimitedSnapshot with minValues) looks like it could be extended, but we'd want guidance on the right direction.

Environment

• @tanstack/db version: 0.5.33
• Runtime: Node.js / Deno
• Context: Server-side durable execution, not browser

[samwillis]

I've been chatting with our agent friends on this one to try and work out what we can do. I'm very cautious of trying to ensure we have exact deployability of all events since an offset. This isn't really something DB is designed to do, the collections sync mechanism could collapse multiple changes into one logical change, there could be external compaction, and some collections may not support it.

What we can do is provide a route to add a $version virtual prop on each row, and each event on the createEffect. The sync layer can then add this version (from the stream offset) on each write() to the collection.

If the stream was modelled as an append only event source, this will work well, you can filter on the $version prop to get the events from the last saved offset. But is they are mutable rows there is a high risk of missing changes.

This is what the LLMs and me came up with:

---

$version Virtual Prop Design

Status

Draft proposal for a lightweight replay/idempotency primitive.

Summary

Add a virtual row prop, $version, so sync adapters can expose a mutation-order
token (offset/cursor/sequence). TanStack DB stays state-delta oriented; $version
is a building block for user-defined replay and idempotency patterns.

Goals

• Expose an adapter-defined ordering token on row transitions.
• Keep collection/query/effect semantics unchanged.
• Support durable checkpoint patterns without introducing a core event-log model.

Non-Goals

• Exact historical replay of every upstream mutation through effects.
• Changing subscribeChanges into a raw event stream.
• Global startAfter semantics across all query/subscription paths.

API Surface

Virtual prop

Add $version to virtual row props.

• Type is collection-defined, default unknown.
• Value is optional: $version: TVersion | undefined.
• Available in selects/where/effects like other virtual props.

Type sketch:

interface VirtualRowProps<
  TKey extends string | number = string | number,
  TVersion = unknown,
> {
  $key: TKey
  $collectionId: string
  $synced: boolean
  $origin: 'local' | 'remote'
  $version: TVersion | undefined
}

Core defaults TVersion to unknown; adapters can narrow (for example to
string offsets).

Metadata mapping convention

Core maps $version from:

• ChangeMessage.metadata.version

Adapters may include more metadata, but metadata.version is the canonical key.

Semantics

Insert

value.$version is the version token of the mutation that produced the visible row.

Update

value.$version is the update version.
previousValue.$version is the prior visible version when available.

Delete / Exit

exit.value.$version must be the delete-time version, not only the previous row
version. This is required for safe checkpoint advancement.

Derived rows (joins, aggregates, includes)

$version remains on namespaced source rows. Core does not auto-combine versions
across sources. Users decide projection/composition in query shape or at the
effect boundary.

Usage

Effect-side checkpointing

createEffect({
  query: (q) => q.from({ runs: db.collections.runs }),
  onEnter: async ({ value }) => {
    const version = value.$version
    if (version == null) return
    if (!isAfter(version, state.lastProcessedVersion)) return

    await sendChildDone(value.id)
    state.lastProcessedVersion = version
  },
})

Query-side filtering (when comparable)

createEffect({
  query: (q) =>
    q
      .from({ runs: db.collections.runs })
      .where(({ runs }) => gt(runs.$version, storedVersion)),
  onEnter: ({ value }) => {
    // handle only rows newer than storedVersion
  },
})

gt/lt is only appropriate when adapter version types are comparable.
For opaque versions, compare in handler code.

Limitations

• $version is an ordering/idempotency hint on observed state transitions.
• It does not imply complete event-log replay.
• Multiple mutations can collapse into one visible transition; intermediate
  versions may not be observable.

Important nuance:

• If user data is modeled as append-only events, $version-based replay can be
  robust.
• If user data collapses actions into mutable state (for example "tool call"
  rows toggled to done), intermediate actions are not reconstructable from
  $version alone.

Implementation Plan

1) Virtual prop plumbing

• Add $version to virtual prop types.
• Thread version generic (TVersion) through row/ref/query typings.
• Include $version in enrichment/pass-through logic.

2) Sync write path

Adapters write version metadata per change:

sync: ({ begin, write, commit }) => {
  begin()
  write({
    type: `update`,
    value: row,
    metadata: { version: sourceVersion },
  })
  commit()
}

Electric today likely uses stream.lastOffset at processing time, which is
batch-level granularity unless upstream provides per-message offsets.

3) Delete/exit propagation (required)

Because deleted-row metadata is removed from persisted synced metadata, delete
version must be carried transiently through event construction:

1. Keep persisted metadata deletion behavior for deleted rows.
2. Preserve delete-time version in pending sync transaction event metadata.
3. Emit delete change messages with that metadata.
4. Enrichment sets delete value $version from delete-event metadata.
5. Effects see exit.value.$version as delete-time version.

Adapter Notes

Electric

• Set metadata.version on every write.
• Include delete version on delete writes.
• Document current granularity (batch-level with lastOffset) until/if
  per-message offsets are available.

Durable streams / StreamDB-like sources

• Map stream cursor/sequence to metadata.version per mutation.
• Apply same insert/update/delete rules as Electric.

Testing

• Type tests for $version on refs/results.
• Collection/effect tests for insert/update/delete/exit version behavior.
• Delete-path tests proving delete-time version reaches exit.
• Collapsed-update tests proving only latest visible version is emitted.
• Optimistic-to-synced tests for $version and $synced interaction.
• Adapter tests for version mapping and resume checkpoint patterns.

Rollout

• Backward-compatible: $version optional, default undefined.
• No semantic changes to skipInitial, subscribeChanges, or query execution.
• $version remains virtual (not required in user schema fields).

Open Questions

• Should core expose a VersionComparator<TVersion> helper contract?
• Is Electric batch-level granularity sufficient, or should adapter pursue
  per-message positions?