TxOrderingMiddleware Implementation

Overview

The TxOrderingMiddleware ensures that transactions for the same document are broadcasted to clients in the correct order, preventing race conditions that cause unnecessary getCurrentDoc calls.

Problem Description

The Race Condition

When multiple transactions for the same document are processed concurrently:

1. Transaction 1 arrives with modifiedOn=101
2. Transaction 2 arrives with modifiedOn=102
3. Due to network/processing delays, tx2's handleBroadcast() might complete before tx1's
4. Client receives tx2 first, then tx1
5. Client's LiveQuery sees out-of-order transactions and calls getCurrentDoc unnecessarily

This was confirmed in the unit test liveQuery.race.test.ts.

Solution Design

Key Insights from client.ts

Looking at how transactions flow in client.ts:

async txRaw(ctx: ClientSessionCtx, tx: Tx): Promise<{...}> {
  // 1. Process transaction
  result = await ctx.pipeline.tx(ctx.ctx, [tx])

  // 2. Send result to client
  await ctx.sendResponse(ctx.requestId, result)

  // 3. Broadcast to other clients (returns promise)
  const broadcastPromise = ctx.pipeline.handleBroadcast(ctx.ctx)

  return { result, broadcastPromise, asyncsPromise }
}

The critical observation: Multiple txRaw() calls can be in-flight simultaneously, and their handleBroadcast() promises may resolve out-of-order.

Two-Phase Approach

Phase 1: tx() - Record Transaction Order

When a transaction is processed:

1. Extract the target document ID from the transaction
2. Create a promise that will be resolved when this tx's broadcast completes
3. Store the transaction entry in a queue: Map<docId, TxOrderEntry[]>

interface TxOrderEntry {
  txId: string
  modifiedOn: number
  broadcastPromise?: Promise<void>
  broadcastResolve?: () => void
}

Phase 2: handleBroadcast() - Wait Before Broadcasting

When broadcasting transactions:

1. For each transaction in the broadcast, find all previous transactions for the same document
2. Wait for all previous broadcasts to complete: await Promise.all(waitPromises)
3. Call the next middleware's handleBroadcast() to perform actual broadcast
4. Resolve this transaction's promise to signal completion
5. Remove the transaction from the queue

Example Scenario

Time  | Thread 1 (tx1, modifiedOn=101)      | Thread 2 (tx2, modifiedOn=102)
------|-------------------------------------|-----------------------------------
T1    | tx() called                         |
      | - Create promise1                   |
      | - Queue: [tx1]                      |
------|-------------------------------------|-----------------------------------
T2    |                                     | tx() called
      |                                     | - Create promise2
      |                                     | - Queue: [tx1, tx2]
------|-------------------------------------|-----------------------------------
T3    |                                     | handleBroadcast() called
      |                                     | - Find tx1 before tx2
      |                                     | - await promise1 (BLOCKS)
------|-------------------------------------|-----------------------------------
T4    | handleBroadcast() called            |
      | - No previous txes                  | (still waiting on promise1)
      | - Broadcast tx1                     |
      | - Resolve promise1                  |
      | - Remove tx1 from queue             |
------|-------------------------------------|-----------------------------------
T5    |                                     | promise1 resolved, continues
      |                                     | - Broadcast tx2
      |                                     | - Resolve promise2
      |                                     | - Remove tx2 from queue

Implementation Details

Data Structure

private readonly docTxQueue = new Map<Ref<Doc>, TxOrderEntry[]>()

• Key: Document ID
• Value: Array of transaction entries in order of arrival
• Automatically cleaned up when queues are empty

Memory Management

To prevent memory leaks:

1. Remove transactions from queue after broadcast completes
2. Delete empty queues
3. If a queue grows beyond 1000 entries, remove oldest entries (with safety resolution)

Key Methods

tx(ctx, txes)

Records transaction order:

for (const tx of txes) {
  const docId = this.getTargetDocId(tx)
  if (docId !== undefined) {
    let queue = this.docTxQueue.get(docId)
    if (queue === undefined) {
      queue = []
      this.docTxQueue.set(docId, queue)
    }

    // Create promise for this transaction's broadcast
    let broadcastResolve
    const broadcastPromise = new Promise<void>((resolve) => {
      broadcastResolve = resolve
    })

    queue.push({
      txId: tx._id,
      modifiedOn: tx.modifiedOn,
      broadcastPromise,
      broadcastResolve
    })
  }
}

handleBroadcast(ctx)

Enforces order and waits:

const waitPromises: Promise<void>[] = []

for (const tx of txes) {
  const queue = this.docTxQueue.get(docId)
  const txIndex = queue.findIndex((entry) => entry.txId === tx._id)

  // Wait for all previous transactions
  for (let i = 0; i < txIndex; i++) {
    const prevEntry = queue[i]
    if (prevEntry.broadcastPromise !== undefined) {
      waitPromises.push(prevEntry.broadcastPromise)
    }
  }
}

// Block until all previous broadcasts complete
await Promise.all(waitPromises)

// Now broadcast this transaction
await this.next?.handleBroadcast(ctx)

// Mark as complete
for (const tx of txes) {
  // Resolve promise and remove from queue
  entry.broadcastResolve()
  queue.splice(txIndex, 1)
}

Testing

Unit Tests (txOrdering.test.ts)

1. Basic ordering: Verify out-of-order transactions are broadcasted in correct sequence
2. Multiple documents: Ensure independent ordering per document
3. Concurrent broadcasts: Test that later transactions wait for earlier ones
4. Statistics tracking: Verify queue size monitoring
5. Memory cleanup: Confirm transactions removed after broadcast

Integration Test

Simulates the full pipeline flow with mocked next middleware.

Performance Considerations

Overhead

• Memory: O(N) where N is number of in-flight transactions per document
• CPU: O(M) where M is position in queue (typically small)

Benefits

• Prevents unnecessary getCurrentDoc calls on client
• Reduces database load
• Improves client-side responsiveness

Scalability

• Independent ordering per document (no global locks)
• Automatic cleanup prevents memory leaks
• Works with any number of concurrent transactions

Deployment

Adding to Pipeline

In your middleware configuration:

import { TxOrderingMiddleware } from '@hcengineering/middleware'

const pipeline = [
  // ... other middleware
  TxOrderingMiddleware.create()
  // ... more middleware
]

Future Enhancements

1. Configurable queue size limit: Currently hardcoded to 1000
2. Metrics integration: Export stats to monitoring system
3. Timeout handling: Add timeout for stuck broadcasts
4. Priority-based ordering: Allow high-priority transactions to bypass queue

Related Documents

• /docs/livequery-race-condition-analysis.md - Detailed problem analysis
• /packages/middleware/src/tests/liveQuery.race.test.ts - Original race condition test
• /packages/middleware/src/tests/txOrdering.test.ts - Middleware unit tests