Add adapter-backed transactions

CHANGELOG.md

@@ -22,6 +22,12 @@ scheduling, retries, flush, and discard. `defineMutationQueue()` gives reconnect
 queues a stable policy and key namespace. Queue writes and ordinary writes share the same
 per-record ordering.
 
+Adapters can opt into atomic multi-service transactions. `figbird.transaction()` collects
+schema-typed CRUD calls, projects and rolls them back as one cache update, and coordinates the
+affected record lanes before making one adapter request. The Feathers adapter includes an
+opt-in `feathersBatchTransactions()` transport for `api/batch`-style services; adapters without
+an atomic capability never fall back to sequential requests.
+
 Writes can use `optimisticPatch` when the local projection differs from the server payload.
 Relational filters apply projected changes locally, then reconcile once after the record's
 writes settle.

docs/content/_index.md

@@ -577,6 +577,47 @@ queries update immediately. Relations assemble from cached data and fetch missin
 When the server must decide query membership or ordering, Figbird waits for the record's
 writes to settle before reconciling the query.
 
+### Adapter-backed transactions
+
+Use `figbird.transaction()` when several CRUD writes must commit or roll back as one server
+operation. Transactions are a capability, not an emulation: Figbird throws if the adapter
+does not provide an atomic transport and never falls back to sequential requests.
+
+```ts
+await figbird.transaction(tx => {
+  tx.m.tasks.patch(taskId, { columnId: nextColumnId })
+  tx.m.columns.patch(previousColumnId, { count: previousCount - 1 })
+  tx.m.columns.patch(nextColumnId, { count: nextCount + 1 })
+})
+```
+
+The callback collects synchronously; its methods return `void`, and the outer promise settles
+when the transaction commits. Payloads, patches, service names, and `confirmed` options retain
+the same schema inference as `m`. Optimistic operations project as one observer-visible cache
+update and roll back together. `tx.m.columns.confirmed.patch(...)` keeps that operation hidden
+until commit.
+
+Transaction operations wait behind earlier writes to every affected record, then reserve those
+record lanes until the adapter transaction settles. Each entity can appear only once in a
+transaction. Creates must be collected one at a time and carry a stable id, including confirmed
+creates, because multi-record ordering needs an identity before dispatch. The transaction DSL is
+CRUD-only; custom methods have adapter-specific argument and cache semantics.
+
+For Feathers, opt into the `api/batch` tuple contract explicitly:
+
+```ts
+import { FeathersAdapter, feathersBatchTransactions } from 'figbird'
+
+const adapter = new FeathersAdapter(feathers, {
+  transactions: feathersBatchTransactions(), // defaults to api/batch
+})
+```
+
+The batch service must return one ordered `{ status: 'fulfilled', value }` or
+`{ status: 'rejected', reason }` entry per call and guarantee that any rejection rolls back the
+whole batch. `feathersBatchTransactions()` rejects the Figbird transaction if any entry rejects.
+Use `serviceName` and `params` options when the batch endpoint differs.
+
 ### Ordered autosave with mutation queues
 
 `m` sends independent records in parallel. A feature such as a workflow editor may instead
@@ -1663,6 +1704,7 @@ const figbird = new Figbird({
 | `prefetch(request, opts?)`                        | Idempotent speculative warming. Argumentless definitions can be passed directly. See [figbird.prefetch](#figbirdprefetch).                                      |
 | `refetch(service?)`                               | Manual refetch escape hatch for changes Figbird can’t observe, such as custom methods without events or out-of-band writes. Call `figbird.refetch(...)`.                  |
 | `m`                                               | The instance’s write proxy: `figbird.m.issues.patch(...)`, or `figbird.m(service)` for dynamic names. In React, access the provider instance through `useMutations()`. See [m](#m). |
+| `transaction(fn)`                                 | Adapter-backed atomic CRUD collector. Optimistic projection, commit, and rollback are grouped across services. See [Adapter-backed transactions](#adapter-backed-transactions). |
 | `createMutationQueue(config?)`                    | Explicitly owned serial writes across records or services. See [figbird.createMutationQueue](#figbirdcreatemutationqueue).                                |
 | `mutating`                                        | Synchronous active-mutation tracker (`subscribe`/`getSnapshot`) — `useMutating` is its React binding.                                                       |
 | `explain(...)`                                    | Static classification report — see [figbird.explain](#figbirdexplain).                                                                                      |
@@ -1691,6 +1733,7 @@ const adapter = new FeathersAdapter(feathers, options)
   - `defaultPageSizeWhenFetchingAll` — default `query.$limit` when fetching with `allPages`
   - `pagination` — cursor strategies keyed by service path; see [Cursor pagination](#cursor-pagination)
   - `operators` — custom query operators the client can evaluate (`{ $asOf: asOf => item => boolean }`); queries using them stay realtime-mergeable. See [Teaching the client custom operators](#teaching-the-client-custom-operators)
+  - `transactions` — optional atomic transaction transport; use `feathersBatchTransactions()` for an `api/batch`-style service
 
 Meta behavior: `find` returns `{ data, meta }` (`FindMeta`: `{ total, limit, skip }`); `get` returns only the item.
 

lib/adapters/adapter.ts

@@ -66,6 +66,13 @@ export interface MatcherContext {
   serviceName: string
 }
 
+/** One mutation inside an adapter-backed atomic transaction. */
+export interface AdapterTransactionOperation {
+  serviceName: string
+  method: 'create' | 'update' | 'patch' | 'remove'
+  args: readonly unknown[]
+}
+
 /**
  * Unified adapter interface
  * The adapter is service-agnostic and works with unknown items
@@ -92,6 +99,16 @@ export interface Adapter<
 
   mutate(serviceName: string, method: string, args: unknown[]): Promise<unknown>
 
+  /**
+   * Atomically commit several mutations, returning one result per operation in
+   * the same order. Omit this capability when the backend cannot guarantee
+   * all-or-nothing commit semantics; Figbird never falls back to sequential
+   * requests.
+   */
+  transaction?:
+    | ((operations: readonly AdapterTransactionOperation[]) => Promise<readonly unknown[]>)
+    | undefined
+
   /** Return false when retrying a failed query cannot help. Errors retry by default. */
   isRetryableError?(error: Error): boolean
 

lib/adapters/feathers.ts

@@ -1,5 +1,6 @@
 import type {
   Adapter,
+  AdapterTransactionOperation,
   EventHandlers,
   MatcherContext,
   PageCursor,
@@ -298,6 +299,77 @@ export type CustomOperatorRegistration =
       byService: Record<string, CustomOperator>
     }
 
+/** Feathers-specific transport for an adapter-backed atomic transaction. */
+export type FeathersTransaction = (
+  feathers: FeathersClient,
+  operations: readonly AdapterTransactionOperation[],
+) => Promise<readonly unknown[]>
+
+export interface FeathersBatchTransactionsOptions {
+  /** Feathers service implementing the batch contract. Defaults to `api/batch`. */
+  serviceName?: string
+  /** Params passed to the batch service's `create` call. */
+  params?: FeathersParams
+}
+
+interface FeathersBatchResult {
+  data: Array<{ status: 'fulfilled'; value?: unknown } | { status: 'rejected'; reason: unknown }>
+}
+
+/** Error returned when an atomic Feathers batch contains rejected operations. */
+export class FeathersTransactionError extends Error {
+  readonly result: FeathersBatchResult
+
+  constructor(result: FeathersBatchResult) {
+    const rejected = result.data.filter(entry => entry.status === 'rejected').length
+    super(`Feathers transaction rejected ${rejected} operation${rejected === 1 ? '' : 's'}`)
+    this.name = 'FeathersTransactionError'
+    this.result = result
+  }
+}
+
+/**
+ * Adapt a Feathers `api/batch`-style service to Figbird's atomic transaction
+ * capability. The service must return ordered `{ status, value/reason }`
+ * entries and roll the entire batch back when any entry rejects.
+ */
+export function feathersBatchTransactions({
+  serviceName = 'api/batch',
+  params,
+}: FeathersBatchTransactionsOptions = {}): FeathersTransaction {
+  return async (feathers, operations) => {
+    const result = await feathers.service(serviceName).create(
+      {
+        calls: operations.map(operation => [
+          operation.method,
+          operation.serviceName,
+          ...operation.args,
+        ]),
+      },
+      params,
+    )
+    if (!isFeathersBatchResult(result) || result.data.length !== operations.length) {
+      throw new Error(`Feathers transaction service "${serviceName}" returned an invalid result`)
+    }
+    if (result.data.some(entry => entry.status === 'rejected')) {
+      throw new FeathersTransactionError(result)
+    }
+    return result.data.map(entry => (entry.status === 'fulfilled' ? entry.value : undefined))
+  }
+}
+
+function isFeathersBatchResult(value: unknown): value is FeathersBatchResult {
+  if (!value || typeof value !== 'object' || !Array.isArray((value as FeathersBatchResult).data)) {
+    return false
+  }
+  return (value as FeathersBatchResult).data.every(
+    entry =>
+      entry &&
+      typeof entry === 'object' &&
+      (entry.status === 'fulfilled' || entry.status === 'rejected'),
+  )
+}
+
 export interface FeathersAdapterOptions {
   idField?: IdFieldType
   updatedAtField?: UpdatedAtFieldType
@@ -326,6 +398,8 @@ export interface FeathersAdapterOptions {
   defaultPagination?: FeathersPagination
   /** Pagination overrides selected by Feathers service path. */
   pagination?: Record<string, FeathersPagination>
+  /** Opt-in atomic transaction transport. Omit when the backend has no such capability. */
+  transactions?: FeathersTransaction
 }
 
 /**
@@ -353,6 +427,7 @@ export class FeathersAdapter<TQuery = Record<string, unknown>> implements Adapte
   #operators: Record<string, CustomOperatorRegistration>
   #defaultPagination: FeathersPagination | undefined
   #pagination: Record<string, FeathersPagination>
+  transaction?: Adapter['transaction']
 
   /** Names of custom operators registered for every service. */
   get customOperators(): readonly string[] {
@@ -400,6 +475,7 @@ export class FeathersAdapter<TQuery = Record<string, unknown>> implements Adapte
       operators = {},
       defaultPagination,
       pagination = {},
+      transactions,
     }: FeathersAdapterOptions = {},
   ) {
     this.feathers = feathers
@@ -410,6 +486,9 @@ export class FeathersAdapter<TQuery = Record<string, unknown>> implements Adapte
     this.#operators = operators
     this.#defaultPagination = defaultPagination
     this.#pagination = pagination
+    if (transactions) {
+      this.transaction = operations => transactions(this.feathers, operations)
+    }
   }
 
   #paginationFor(serviceName: string): FeathersPagination | undefined {

lib/core/figbird.ts

@@ -59,6 +59,7 @@ import type {
   ServiceUpdate,
 } from './schema.js'
 import { resolveServicePath } from './schema.js'
+import { createTransactionContext, type TransactionContext } from './transactions.js'
 
 type DescriptorWriteProjection<TItem> =
   | {
@@ -91,6 +92,11 @@ export type {
   MutationsProxy,
   WriteMutationOptions,
 } from './mutations.js'
+export type {
+  TransactionContext,
+  TransactionMutationsHandle,
+  TransactionMutationsProxy,
+} from './transactions.js'
 export {
   defineMutationQueue,
   MutationQueueDiscardedError,
@@ -664,6 +670,44 @@ export class Figbird<
     return this.#mutationsProxy
   }
 
+  /**
+   * Atomically commit several typed CRUD mutations through the configured
+   * adapter. The callback is a synchronous collector: its calls project as one
+   * cache update, wait for affected record lanes, and are dispatched together.
+   * No sequential-request fallback is provided.
+   */
+  transaction(collect: (transaction: TransactionContext<S>) => undefined): Promise<void> {
+    if (!this.queryStore.supportsTransactions) {
+      throw new Error('figbird: the configured adapter does not support transactions')
+    }
+    const transaction = createTransactionContext<S>()
+    let returned: unknown
+    try {
+      returned = collect(transaction.context)
+    } catch (error) {
+      // Close on failure too, so a leaked handle cannot append work later.
+      transaction.close()
+      throw error
+    }
+    if (
+      returned !== null &&
+      (typeof returned === 'object' || typeof returned === 'function') &&
+      'then' in returned &&
+      typeof (returned as { then?: unknown }).then === 'function'
+    ) {
+      transaction.close()
+      // The callback may continue after its first await and hit the closed
+      // collector. Observe that misuse promise so it cannot become unhandled.
+      void Promise.resolve(returned).catch(() => {})
+      throw new Error('figbird: transaction callbacks must be synchronous')
+    }
+    const descs = transaction.close().map(desc => ({
+      ...desc,
+      serviceName: resolveServicePath(this.schema, desc.serviceName),
+    }))
+    return this.queryStore.transaction(descs)
+  }
+
   /**
    * Create an explicitly owned serial mutation queue. Calls made through the
    * queue's `m` proxy project immediately, preserve queue order across records,

lib/core/mutationLanes.ts

@@ -139,6 +139,22 @@ export class MutationLanes<TEntry extends MutationLaneEntry> {
     lane: MutationLane,
     entry: TEntry,
     outcome: MutationOutcome,
+  ): LaneSettlement<TEntry> | null {
+    return this.#complete(lane, entry, outcome)
+  }
+
+  /**
+   * Remove an entry that will never reach the adapter while preserving the
+   * create/remove lifetime rules that apply to an ordinary failed mutation.
+   */
+  abort(lane: MutationLane, entry: TEntry, error: Error): LaneSettlement<TEntry> | null {
+    return this.#complete(lane, entry, { ok: false, error })
+  }
+
+  #complete(
+    lane: MutationLane,
+    entry: TEntry,
+    outcome: MutationOutcome,
   ): LaneSettlement<TEntry> | null {
     const state = this.#lanes.get(lane.key)
     if (state !== lane) return null
@@ -147,18 +163,22 @@ export class MutationLanes<TEntry extends MutationLaneEntry> {
     if (index === -1) return null
 
     state.entries.splice(index, 1)
-    state.running = false
+    if (index === 0) state.running = false
 
     let cancelled: TEntry[] = []
     if (!outcome.ok && entry.desc.method === 'create') {
-      cancelled = state.entries.splice(0)
+      cancelled = state.entries.splice(index)
     } else if (!outcome.ok && entry.desc.method === 'remove') {
-      const nextCreate = state.entries.findIndex(queued => queued.desc.method === 'create')
+      const nextCreate = state.entries.findIndex(
+        (queued, queuedIndex) => queuedIndex >= index && queued.desc.method === 'create',
+      )
       if (nextCreate !== -1) cancelled = state.entries.splice(nextCreate)
     } else if (outcome.ok && entry.desc.method === 'remove') {
-      const nextCreate = state.entries.findIndex(queued => queued.desc.method === 'create')
+      const nextCreate = state.entries.findIndex(
+        (queued, queuedIndex) => queuedIndex >= index && queued.desc.method === 'create',
+      )
       const end = nextCreate === -1 ? state.entries.length : nextCreate
-      cancelled = state.entries.splice(0, end)
+      cancelled = state.entries.splice(index, end - index)
     }
     let authoritativeEvent: ProcessedRealtimeEvent | null = null