fix: validate monotonic cursor ordering and document same-key limitation

Reject non-monotonic cursor sequences within a sync batch via
onSourceError instead of silently leaking replay changes. Document
the known limitation that same-key replay+live changes within one
transaction produce an enter event instead of update due to
transaction-scoped graph coalescing.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: KyleAMathews

packages/db/src/query/effect.ts

@@ -738,11 +738,19 @@ class EffectPipelineRunner<TRow extends object, TKey extends string | number> {
         continue
       }
       try {
-        if (
-          this.pendingBatchCursor === undefined ||
-          compareCollectionCursors(cursor, this.pendingBatchCursor) > 0
-        ) {
+        if (this.pendingBatchCursor === undefined) {
           this.pendingBatchCursor = cursor
+        } else {
+          const cmp = compareCollectionCursors(cursor, this.pendingBatchCursor)
+          if (cmp < 0) {
+            throw new Error(
+              `Cursors within a sync batch must be monotonically ordered. ` +
+                `Saw ${String(cursor)} after ${String(this.pendingBatchCursor)}.`,
+            )
+          }
+          if (cmp > 0) {
+            this.pendingBatchCursor = cursor
+          }
         }
       } catch (error) {
         this.onSourceError(
@@ -1227,11 +1235,19 @@ function findFirstChangeAfterCursor(
   startAfter: CollectionCursor,
 ): number {
   let anyCursor = false
+  let lastCursor: CollectionCursor | undefined
   for (let index = 0; index < changes.length; index++) {
     const cursor = changes[index]!.cursor
     if (cursor === undefined) {
       continue
     }
+    if (lastCursor !== undefined && compareCollectionCursors(cursor, lastCursor) < 0) {
+      throw new Error(
+        `Cursors within a sync batch must be monotonically ordered. ` +
+          `Saw ${String(cursor)} after ${String(lastCursor)}.`,
+      )
+    }
+    lastCursor = cursor
     anyCursor = true
     if (compareCollectionCursors(cursor, startAfter) > 0) {
       // Walk backwards to include any preceding uncursored changes —

packages/db/tests/effect.test.ts

@@ -569,6 +569,99 @@ describe(`createEffect`, () => {
       await effect.dispose()
     })
 
+    it(`should emit enter when same key crosses startAfter within one transaction (known limitation)`, async () => {
+      // Known limitation: when the same key has a replay insert and a live
+      // update within one transaction, the transaction-scoped scheduler
+      // coalesces both into a single graph run. D2 sees the net result as
+      // a new row, so the event type is `enter` instead of `update`.
+      // Correct behavior would be `update` with previousValue from the
+      // replay insert, but fixing this requires per-change D2 fencing
+      // within transactions, which is too invasive for the initial cut.
+      const users = createCollection(
+        mockSyncCollectionOptionsNoInitialState<User>({
+          id: `same-key-straddle-users`,
+          getKey: (user) => user.id,
+        }),
+      )
+      const events: Array<DeltaEvent<User, number>> = []
+
+      const effect = createEffect<User, number>({
+        query: (q) => q.from({ user: users }),
+        onBatch: collectBatchEvents(events),
+        startAfter: 2,
+      })
+
+      users.utils.markReady()
+      await flushPromises()
+
+      users.utils.begin()
+      users.utils.write({
+        type: `insert`,
+        value: { id: 1, name: `Alice`, active: true },
+        cursor: 2,
+      })
+      users.utils.write({
+        type: `update`,
+        value: { id: 1, name: `Alice v2`, active: true },
+        previousValue: { id: 1, name: `Alice`, active: true },
+        cursor: 3,
+      })
+      users.utils.commit()
+      await flushPromises()
+
+      expect(events).toHaveLength(1)
+      // Ideally this would be `update`, but coalescing produces `enter`
+      expect(events[0]!.type).toBe(`enter`)
+      expect(events[0]!.value.name).toBe(`Alice v2`)
+
+      await effect.dispose()
+    })
+
+    it(`should reject non-monotonic cursor sequences within a batch`, async () => {
+      const users = createCollection(
+        mockSyncCollectionOptionsNoInitialState<User>({
+          id: `non-monotonic-cursor-users`,
+          getKey: (user) => user.id,
+        }),
+      )
+      const errors: Array<Error> = []
+
+      const effect = createEffect<User, number>({
+        query: (q) => q.from({ user: users }),
+        onBatch: () => {},
+        startAfter: 2,
+        onSourceError: (err) => errors.push(err),
+      })
+
+      users.utils.markReady()
+      await flushPromises()
+
+      // Non-monotonic batch: [cursor 1, cursor 3, cursor 2]
+      users.utils.begin()
+      users.utils.write({
+        type: `insert`,
+        value: { id: 1, name: `Alice`, active: true },
+        cursor: 1,
+      })
+      users.utils.write({
+        type: `insert`,
+        value: { id: 2, name: `Bob`, active: true },
+        cursor: 3,
+      })
+      users.utils.write({
+        type: `insert`,
+        value: { id: 3, name: `Charlie`, active: true },
+        cursor: 2,
+      })
+      users.utils.commit()
+      await flushPromises()
+
+      expect(errors).toHaveLength(1)
+      expect(errors[0]!.message).toMatch(/monotonically/)
+
+      await effect.dispose()
+    })
+
     it(`should throw when startAfter is used with multi-source effects`, () => {
       const users = createUsersCollection()
       const issues = createIssuesCollection()