diff --git a/.changeset/parallel-dependency-queue.md b/.changeset/parallel-dependency-queue.md new file mode 100644 index 000000000..22df2bd77 --- /dev/null +++ b/.changeset/parallel-dependency-queue.md @@ -0,0 +1,30 @@ +--- +'@tanstack/db': minor +--- + +Add `dependencyQueueStrategy` for parallel execution of independent mutations. + +This new paced mutations strategy enables mutations on different records to run in parallel while still serializing mutations that share dependencies (based on `globalKey`). This provides better performance than `queueStrategy` when mutations are independent, without sacrificing correctness. + +Key features: + +- Mutations on different records run concurrently +- Mutations on the same record are serialized in order +- Supports custom dependency declarations via `getDependencies` callback for semantic relationships + +```ts +import { dependencyQueueStrategy } from '@tanstack/db' + +const mutate = usePacedMutations({ + onMutate: (vars) => collection.update(vars.id, ...), + mutationFn: async ({ transaction }) => api.save(transaction.mutations), + strategy: dependencyQueueStrategy() +}) + +// These run in parallel (different records): +mutate({ id: 1, title: 'A' }) +mutate({ id: 2, title: 'B' }) + +// This waits for the first one (same record): +mutate({ id: 1, title: 'C' }) +``` diff --git a/docs/guides/mutations.md b/docs/guides/mutations.md index 4c1662560..500bfdf29 100644 --- a/docs/guides/mutations.md +++ b/docs/guides/mutations.md @@ -1002,6 +1002,7 @@ The fundamental difference between strategies is how they handle transactions: | **`debounceStrategy`** | Wait for inactivity before persisting. Only final state is saved. | Auto-save forms, search-as-you-type | | **`throttleStrategy`** | Ensure minimum spacing between executions. Mutations between executions are merged. | Sliders, progress updates, analytics | | **`queueStrategy`** | Each mutation becomes a separate transaction, processed sequentially in order (FIFO by default, configurable to LIFO). All mutations guaranteed to persist. | Sequential workflows, file uploads, rate-limited APIs | +| **`dependencyQueueStrategy`** | Parallel execution for independent records, serialized for same-record mutations. Based on `globalKey` tracking. | Multi-item editors, reordering operations, complex forms with independent fields | ### Debounce Strategy @@ -1139,6 +1140,83 @@ function FileUploader() { - All mutations guaranteed to persist - Waits for each transaction to complete before starting the next +### Dependency Queue Strategy + +The dependency queue strategy provides the best of both worlds: parallel execution for mutations on different records, while maintaining serialization for mutations on the same record. This is ideal for complex UIs where users might be editing multiple independent items simultaneously. + +```tsx +import { usePacedMutations, dependencyQueueStrategy } from "@tanstack/react-db" + +function MultiItemEditor() { + const mutate = usePacedMutations<{ id: string; title: string }>({ + onMutate: ({ id, title }) => { + // Apply optimistic update immediately + itemCollection.update(id, (draft) => { + draft.title = title + }) + }, + mutationFn: async ({ transaction }) => { + await api.items.update(transaction.mutations) + }, + // Parallel for different items, serial for same item + strategy: dependencyQueueStrategy(), + }) + + const handleTitleChange = (id: string, title: string) => { + mutate({ id, title }) + } + + return ( +
+ {items.map((item) => ( + handleTitleChange(item.id, e.target.value)} + /> + ))} +
+ ) +} +``` + +In this example: +- Editing `item-1` and `item-2` simultaneously fires both mutations in parallel +- Editing `item-1` twice in quick succession serializes those mutations (second waits for first) + +**Key characteristics**: +- Mutations on different records run in parallel +- Mutations on the same record are serialized in order +- Uses `globalKey` (`KEY::{collectionId}/{itemKey}`) for dependency tracking +- Supports custom dependency declarations for semantic relationships + +#### Custom Dependencies + +For complex scenarios where you need to express dependencies beyond the automatic `globalKey` tracking, use the `getDependencies` option: + +```tsx +const mutate = usePacedMutations<{ listId: string; changes: Partial }>({ + onMutate: ({ listId, changes }) => { + listCollection.update(listId, (draft) => { + Object.assign(draft, changes) + }) + }, + mutationFn: async ({ transaction }) => { + await api.lists.update(transaction.mutations) + }, + strategy: dependencyQueueStrategy({ + getDependencies: (tx) => { + // List mutations should wait for any item mutations in that list + return tx.mutations + .filter((m) => m.collection.id === 'lists') + .map((m) => `list-items:${m.key}`) + }, + }), +}) +``` + +This allows you to express semantic dependencies like "updating a list should wait for all its items to finish updating." + ### Choosing a Strategy Use this guide to pick the right strategy for your use case: @@ -1162,6 +1240,13 @@ Use this guide to pick the right strategy for your use case: - You need sequential processing with delays - Examples: file uploads, batch operations, audit trails, multi-step wizards +**Use `dependencyQueueStrategy` when:** +- You want parallel execution for independent records +- Same-record mutations need to be serialized +- You're building multi-item editors or list UIs +- Users might edit multiple items simultaneously +- Examples: kanban boards, multi-item forms, reorder operations, batch editing + ### Using in React The `usePacedMutations` hook makes it easy to use paced mutations in React components: diff --git a/packages/db/src/paced-mutations.ts b/packages/db/src/paced-mutations.ts index d1d6d9a3e..ebf4f7d2d 100644 --- a/packages/db/src/paced-mutations.ts +++ b/packages/db/src/paced-mutations.ts @@ -157,6 +157,12 @@ export function createPacedMutations< }) return capturedTx }) + } else if (strategy._type === `dependencyQueue`) { + // For dependency queue strategy, pass the transaction directly so it can + // extract globalKeys and manage dependencies before committing + const capturedTx = activeTransaction + activeTransaction = null // Clear so next mutation creates a new transaction + strategy.executeWithTx(capturedTx) } else { // For debounce/throttle, use commitCallback which manages activeTransaction strategy.execute(commitCallback) diff --git a/packages/db/src/strategies/dependencyQueueStrategy.ts b/packages/db/src/strategies/dependencyQueueStrategy.ts new file mode 100644 index 000000000..d98358f70 --- /dev/null +++ b/packages/db/src/strategies/dependencyQueueStrategy.ts @@ -0,0 +1,259 @@ +import type { Transaction } from '../transactions' +import type { + DependencyQueueStrategy, + DependencyQueueStrategyOptions, +} from './types' + +interface QueuedItem> { + tx: Transaction + keys: Set + resolve: () => void +} + +/** + * Creates a dependency-aware queue strategy that: + * - Runs independent mutations in parallel + * - Serializes mutations that share dependencies (same globalKey) + * - Supports custom dependency declarations for semantic relationships + * + * This solves the problem where a global queue forces everything to run sequentially, + * even when mutations affect completely unrelated records. + * + * @example + * ```ts + * // Basic usage - automatically parallelizes based on globalKey + * const mutate = usePacedMutations({ + * onMutate: (variables) => { + * itemsCollection.update(variables.id, draft => { + * draft.title = variables.title + * }) + * }, + * mutationFn: async ({ transaction }) => { + * await api.save(transaction.mutations) + * }, + * strategy: dependencyQueueStrategy() + * }) + * + * // These run in parallel (different items): + * mutate({ id: 'item-1', title: 'New title 1' }) + * mutate({ id: 'item-2', title: 'New title 2' }) + * + * // This waits for the above (touches both items): + * mutate({ id: 'item-1', reorderWith: 'item-2' }) + * ``` + * + * @example + * ```ts + * // With custom dependencies for semantic relationships + * const mutate = usePacedMutations({ + * onMutate: (variables) => { + * listsCollection.update(variables.listId, draft => { + * draft.title = variables.title + * }) + * }, + * mutationFn: async ({ transaction }) => { + * await api.save(transaction.mutations) + * }, + * strategy: dependencyQueueStrategy({ + * getDependencies: (tx) => { + * // List mutations should wait for any item mutations in that list + * const listIds = tx.mutations + * .filter(m => m.collection.id === 'lists') + * .map(m => `list-items:${m.key}`) + * return listIds + * } + * }) + * }) + * ``` + */ +export function dependencyQueueStrategy( + options?: DependencyQueueStrategyOptions, +): DependencyQueueStrategy { + // Map from key -> Set of transactions currently using that key + const inFlightByKey = new Map>() + + // Pending items waiting for dependencies to clear + const pendingQueue: Array = [] + + // Timer for optional wait + let waitTimer: ReturnType | null = null + + /** + * Get all dependency keys for a transaction + */ + function getDependencyKeys( + tx: Transaction, + ): Set { + const keys = new Set() + + // Add all globalKeys from mutations + for (const mutation of tx.mutations) { + keys.add(mutation.globalKey) + } + + // Add custom dependencies if provided + if (options?.getDependencies) { + // Cast to base Transaction type for the callback + const customDeps = options.getDependencies(tx as unknown as Transaction) + for (const dep of customDeps) { + keys.add(dep) + } + } + + return keys + } + + /** + * Check if any of the given keys are currently in-flight + */ + function hasInFlightDependencies(keys: Set): boolean { + for (const key of keys) { + if (inFlightByKey.has(key) && inFlightByKey.get(key)!.size > 0) { + return true + } + } + return false + } + + /** + * Mark keys as in-flight for a transaction + */ + function markInFlight( + tx: Transaction, + keys: Set, + ): void { + for (const key of keys) { + if (!inFlightByKey.has(key)) { + inFlightByKey.set(key, new Set()) + } + inFlightByKey.get(key)!.add(tx as Transaction) + } + } + + /** + * Remove transaction from in-flight tracking and process pending items + */ + function markCompleted( + tx: Transaction, + keys: Set, + ): void { + for (const key of keys) { + const inFlight = inFlightByKey.get(key) + if (inFlight) { + inFlight.delete(tx as Transaction) + if (inFlight.size === 0) { + inFlightByKey.delete(key) + } + } + } + + // Try to process pending items that may now be unblocked + processQueue() + } + + /** + * Process the pending queue, starting any items whose dependencies are clear + */ + function processQueue(): void { + // Process in order, but allow multiple items to start if they're independent + let i = 0 + while (i < pendingQueue.length) { + const item = pendingQueue[i]! + + if (!hasInFlightDependencies(item.keys)) { + // Remove from queue + pendingQueue.splice(i, 1) + + // Mark as in-flight + markInFlight(item.tx, item.keys) + + // Start the commit + startCommit(item.tx, item.keys) + + // Don't increment i since we removed an item + } else { + i++ + } + } + } + + /** + * Start committing a transaction and handle completion + */ + function startCommit( + tx: Transaction, + keys: Set, + ): void { + tx.commit() + .catch(() => { + // Errors are handled via transaction.isPersisted.promise + }) + .finally(() => { + markCompleted(tx, keys) + }) + } + + /** + * Add a transaction to the queue + */ + function enqueue(tx: Transaction): void { + const keys = getDependencyKeys(tx) + + if (!hasInFlightDependencies(keys)) { + // No dependencies blocking - start immediately + markInFlight(tx, keys) + startCommit(tx, keys) + } else { + // Add to pending queue + pendingQueue.push({ + tx: tx as Transaction, + keys, + resolve: () => {}, + }) + } + } + + /** + * Handle optional wait time + */ + function scheduleEnqueue(tx: Transaction): void { + if (options?.wait && options.wait > 0) { + // Clear any existing timer + if (waitTimer) { + clearTimeout(waitTimer) + } + waitTimer = setTimeout(() => { + enqueue(tx) + waitTimer = null + }, options.wait) + } else { + enqueue(tx) + } + } + + return { + _type: `dependencyQueue`, + options, + + executeWithTx: (tx: Transaction) => { + scheduleEnqueue(tx) + }, + + // Fallback for standard execute API - immediately calls fn and extracts tx + execute: (fn: () => Transaction) => { + const tx = fn() + // Note: The fn() may have already started committing in some cases + // This is a compatibility mode - prefer executeWithTx for proper behavior + scheduleEnqueue(tx) + }, + + cleanup: () => { + if (waitTimer) { + clearTimeout(waitTimer) + waitTimer = null + } + pendingQueue.length = 0 + inFlightByKey.clear() + }, + } +} diff --git a/packages/db/src/strategies/index.ts b/packages/db/src/strategies/index.ts index 64d41acea..d23cce574 100644 --- a/packages/db/src/strategies/index.ts +++ b/packages/db/src/strategies/index.ts @@ -2,6 +2,7 @@ export { debounceStrategy } from './debounceStrategy' export { queueStrategy } from './queueStrategy' export { throttleStrategy } from './throttleStrategy' +export { dependencyQueueStrategy } from './dependencyQueueStrategy' // Export strategy types export type { @@ -13,5 +14,7 @@ export type { QueueStrategyOptions, ThrottleStrategy, ThrottleStrategyOptions, + DependencyQueueStrategy, + DependencyQueueStrategyOptions, StrategyOptions, } from './types' diff --git a/packages/db/src/strategies/types.ts b/packages/db/src/strategies/types.ts index 70102c87a..5e8acab6e 100644 --- a/packages/db/src/strategies/types.ts +++ b/packages/db/src/strategies/types.ts @@ -107,6 +107,33 @@ export interface BatchStrategy extends BaseStrategy<`batch`> { options?: BatchStrategyOptions } +/** + * Options for dependency queue strategy + * Processes mutations in parallel when they affect different records, + * but serializes mutations that share dependencies + */ +export interface DependencyQueueStrategyOptions { + /** Optional wait time before processing (milliseconds) */ + wait?: number + /** + * Optional function to extract additional dependency keys beyond globalKey. + * Use this to declare semantic dependencies (e.g., "list depends on its items"). + */ + getDependencies?: (tx: Transaction) => Array +} + +/** + * Dependency queue strategy that enables parallel execution of independent mutations + * while serializing mutations that share record dependencies (via globalKey) + */ +export interface DependencyQueueStrategy extends BaseStrategy<`dependencyQueue`> { + options?: DependencyQueueStrategyOptions + /** + * Execute with direct transaction access for dependency analysis + */ + executeWithTx: (tx: Transaction) => void +} + /** * Union type of all available strategies */ @@ -115,6 +142,7 @@ export type Strategy = | QueueStrategy | ThrottleStrategy | BatchStrategy + | DependencyQueueStrategy /** * Extract the options type from a strategy @@ -127,4 +155,6 @@ export type StrategyOptions = T extends DebounceStrategy ? ThrottleStrategyOptions : T extends BatchStrategy ? BatchStrategyOptions - : never + : T extends DependencyQueueStrategy + ? DependencyQueueStrategyOptions + : never diff --git a/packages/db/tests/paced-mutations.test.ts b/packages/db/tests/paced-mutations.test.ts index 026ed314a..7fd16ec8d 100644 --- a/packages/db/tests/paced-mutations.test.ts +++ b/packages/db/tests/paced-mutations.test.ts @@ -3,6 +3,7 @@ import { createCollection } from '../src/collection' import { createPacedMutations } from '../src/paced-mutations' import { debounceStrategy, + dependencyQueueStrategy, queueStrategy, throttleStrategy, } from '../src/strategies' @@ -455,6 +456,281 @@ describe(`createPacedMutations`, () => { }) }) + describe(`with dependency queue strategy`, () => { + it(`should run independent mutations in parallel`, async () => { + const executionLog: Array<{ id: number; action: string }> = [] + const mutationFn = vi.fn(async ({ transaction }) => { + const id = transaction.mutations[0].changes.id + executionLog.push({ id, action: `start` }) + // Simulate async work + await new Promise((resolve) => setTimeout(resolve, 30)) + executionLog.push({ id, action: `end` }) + }) + + const collection = await createReadyCollection({ + id: `test`, + getKey: (item) => item.id, + }) + + const mutate = createPacedMutations({ + onMutate: (item) => { + collection.insert(item) + }, + mutationFn, + strategy: dependencyQueueStrategy(), + }) + + // Trigger mutations on DIFFERENT items - should run in parallel + const tx1 = mutate({ id: 1, value: 1 }) + const tx2 = mutate({ id: 2, value: 2 }) + const tx3 = mutate({ id: 3, value: 3 }) + + // Each should be a different transaction + expect(tx1).not.toBe(tx2) + expect(tx2).not.toBe(tx3) + + // Wait for all to complete + await Promise.all([ + tx1.isPersisted.promise, + tx2.isPersisted.promise, + tx3.isPersisted.promise, + ]) + + // All should be completed + expect(tx1.state).toBe(`completed`) + expect(tx2.state).toBe(`completed`) + expect(tx3.state).toBe(`completed`) + + // Key test: they should have started before others finished (parallel execution) + // Check that at least 2 started before the first one ended + const firstEndIndex = executionLog.findIndex( + (log) => log.action === `end`, + ) + const startsBeforeFirstEnd = executionLog + .slice(0, firstEndIndex) + .filter((log) => log.action === `start`).length + + expect(startsBeforeFirstEnd).toBeGreaterThanOrEqual(2) + }) + + it(`should serialize mutations on the same key`, async () => { + const executionOrder: Array = [] + let currentlyExecuting = false + + const mutationFn = vi.fn(async ({ transaction }) => { + // Verify no concurrent execution on same key + expect(currentlyExecuting).toBe(false) + currentlyExecuting = true + + const mutation = transaction.mutations[0] + executionOrder.push(`${mutation.type}:${mutation.changes.value}`) + + // Simulate async work + await new Promise((resolve) => setTimeout(resolve, 20)) + + currentlyExecuting = false + }) + + const collection = await createReadyCollection({ + id: `test`, + getKey: (item) => item.id, + }) + + let insertCount = 0 + const mutate = createPacedMutations({ + onMutate: (item) => { + if (insertCount === 0) { + collection.insert(item) + insertCount++ + } else { + collection.update(item.id, (draft) => { + draft.value = item.value + }) + } + }, + mutationFn, + strategy: dependencyQueueStrategy(), + }) + + // Trigger mutations on SAME item - should serialize + const tx1 = mutate({ id: 1, value: 1 }) + const tx2 = mutate({ id: 1, value: 2 }) + const tx3 = mutate({ id: 1, value: 3 }) + + // Each should be a different transaction + expect(tx1).not.toBe(tx2) + expect(tx2).not.toBe(tx3) + + // Wait for all to complete + await Promise.all([ + tx1.isPersisted.promise, + tx2.isPersisted.promise, + tx3.isPersisted.promise, + ]) + + // Should have executed in order (serialized) + expect(executionOrder).toEqual([`insert:1`, `update:2`, `update:3`]) + }) + + it(`should handle mixed dependencies - parallel where possible, serial where needed`, async () => { + const executionLog: Array<{ + id: number + action: string + time: number + }> = [] + const startTime = Date.now() + + const mutationFn = vi.fn(async ({ transaction }) => { + // Use 'key' field instead of changes.id since updates may not include id in changes + const id = transaction.mutations[0].key as number + executionLog.push({ id, action: `start`, time: Date.now() - startTime }) + await new Promise((resolve) => setTimeout(resolve, 30)) + executionLog.push({ id, action: `end`, time: Date.now() - startTime }) + }) + + const collection = await createReadyCollection({ + id: `test`, + getKey: (item) => item.id, + }) + + let firstInsert = true + let secondInsert = true + const mutate = createPacedMutations({ + onMutate: (item) => { + if (item.id === 1) { + if (firstInsert) { + collection.insert(item) + firstInsert = false + } else { + collection.update(item.id, (draft) => { + draft.value = item.value + }) + } + } else { + if (secondInsert) { + collection.insert(item) + secondInsert = false + } else { + collection.update(item.id, (draft) => { + draft.value = item.value + }) + } + } + }, + mutationFn, + strategy: dependencyQueueStrategy(), + }) + + // Pattern: item 1 twice, item 2 once + // tx1 (item 1) and tx3 (item 2) can run in parallel + // tx2 (item 1) must wait for tx1 + const tx1 = mutate({ id: 1, value: 1 }) + const tx2 = mutate({ id: 1, value: 2 }) + const tx3 = mutate({ id: 2, value: 1 }) + + await Promise.all([ + tx1.isPersisted.promise, + tx2.isPersisted.promise, + tx3.isPersisted.promise, + ]) + + // tx1 and tx3 should start at roughly the same time (parallel) + const tx1Start = executionLog.find( + (l) => l.id === 1 && l.action === `start`, + )! + const tx3Start = executionLog.find( + (l) => l.id === 2 && l.action === `start`, + )! + + // They should start within 20ms of each other (accounting for test timing variance) + expect(Math.abs(tx1Start.time - tx3Start.time)).toBeLessThan(20) + + // tx2 (second item 1 mutation) should start after tx1 ends + const tx1End = executionLog.find((l) => l.id === 1 && l.action === `end`)! + const tx2Start = executionLog.filter( + (l) => l.id === 1 && l.action === `start`, + )[1]! + + expect(tx2Start.time).toBeGreaterThanOrEqual(tx1End.time - 5) // small tolerance + }) + + it(`should continue processing after an error`, async () => { + const mutationFn = vi + .fn() + .mockRejectedValueOnce(new Error(`First failed`)) + .mockResolvedValueOnce(undefined) + + const collection = await createReadyCollection({ + id: `test`, + getKey: (item) => item.id, + }) + + let insertCount = 0 + const mutate = createPacedMutations({ + onMutate: (item) => { + if (insertCount === 0) { + collection.insert(item) + insertCount++ + } else { + collection.update(item.id, (draft) => { + draft.value = item.value + }) + } + }, + mutationFn, + strategy: dependencyQueueStrategy(), + }) + + // First mutation will fail, second should still process + const tx1 = mutate({ id: 1, value: 1 }) + const tx2 = mutate({ id: 1, value: 2 }) + + await Promise.allSettled([ + tx1.isPersisted.promise, + tx2.isPersisted.promise, + ]) + + expect(tx1.state).toBe(`failed`) + expect(tx2.state).toBe(`failed`) // Cascaded rollback due to same key + expect(mutationFn).toHaveBeenCalledTimes(1) // Second didn't run due to rollback + }) + + it(`should process independent items after one fails`, async () => { + const mutationFn = vi.fn().mockImplementation(async ({ transaction }) => { + const id = transaction.mutations[0].changes.id + if (id === 1) { + throw new Error(`Item 1 failed`) + } + await new Promise((resolve) => setTimeout(resolve, 10)) + }) + + const collection = await createReadyCollection({ + id: `test`, + getKey: (item) => item.id, + }) + + const mutate = createPacedMutations({ + onMutate: (item) => { + collection.insert(item) + }, + mutationFn, + strategy: dependencyQueueStrategy(), + }) + + // item 1 will fail, item 2 should still succeed (independent) + const tx1 = mutate({ id: 1, value: 1 }) + const tx2 = mutate({ id: 2, value: 2 }) + + await Promise.allSettled([ + tx1.isPersisted.promise, + tx2.isPersisted.promise, + ]) + + expect(tx1.state).toBe(`failed`) + expect(tx2.state).toBe(`completed`) // Independent item should succeed + }) + }) + describe(`error handling`, () => { it(`should handle mutationFn errors and set transaction to failed state`, async () => { const error = new Error(`Mutation failed`)