Fix isWhereSubset to handle AND subset with OR superset (#1275)

* fix(db): fix isNull predicate causing LiveQuery to never become ready when offline

The `isWhereSubsetInternal` function had a critical ordering issue in its
predicate checking logic. When a query with `and(eq(...), isNull(...))` was
checked against a union superset like `or(and(eq(...), isNull(...)), ...)`,
the AND decomposition ran before the OR superset check. This caused the
function to decompose the AND into individual conjuncts (eq and isNull),
neither of which could independently prove subset of the OR expression,
even though the full AND expression was structurally equal to one of the
OR disjuncts.

The fix reorders the checks so that:
1. superset AND is handled first (unchanged)
2. subset OR is handled next (moved up from after AND)
3. superset OR is handled next (moved up from after IN)
4. subset AND decomposition happens last

This ensures that when subset=AND and superset=OR, the full AND expression
is checked against each OR disjunct (where structural equality catches it),
before being decomposed into individual conjuncts that lose the ability to
match compound disjuncts.

Reported scenario: on-demand sync with `isNull(soft_deleted_at)` in a where
clause caused `isLoadingSubset` to stay true after going offline and
remounting, because the DeduplicatedLoadSubset couldn't recognize that
data for the predicate was already loaded.

https://claude.ai/code/session_01TmGem9Pj1NnXfaWY5FZJdb

* ci: apply automated fixes

* refactor: improve error handling and test coverage for review feedback

Separate forceDisconnectAndRefresh into its own try-catch with correct
error attribution, add negative guard test for isUpToDate check, add
rejection path test, trim verbose comments, and clarify wording.

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

* chore: add changeset for isNull predicate fix

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

* fix: handle forceDisconnectAndRefresh PauseLock error gracefully

When loadSubset is called during subscriber message processing (e.g.,
join pipelines via D2.step), the stream's PauseLock is held and
forceDisconnectAndRefresh deadlocks. Since the refresh is only an
optimization to break out of long-poll waits, log the error and fall
through to requestSnapshot which still works.

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

---------

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: 3 people

.changeset/fix-isnull-predicate-subset.md

@@ -0,0 +1,6 @@
+---
+'@tanstack/db': patch
+'@tanstack/electric-db-collection': patch
+---
+
+Fix isNull predicate causing LiveQuery to never become ready when offline. Reorder predicate checks in `isWhereSubsetInternal` so OR superset handling runs before AND subset decomposition, allowing `and(eq, isNull)` to match structurally equal disjuncts. Also separate `forceDisconnectAndRefresh` error handling into its own try-catch with correct error attribution.

packages/db/src/query/predicate-utils.ts

@@ -87,6 +87,24 @@ function isWhereSubsetInternal(
     )
   }
 
+  // Handle OR in subset: (A OR B) ⊆ C only if both A ⊆ C and B ⊆ C.
+  // Must be checked before OR superset so that or(A, B) ⊆ or(C, D)
+  // decomposes the subset first: A ⊆ or(C, D) AND B ⊆ or(C, D).
+  if (subset.type === `func` && subset.name === `or`) {
+    return subset.args.every((arg) =>
+      isWhereSubsetInternal(arg as BasicExpression<boolean>, superset),
+    )
+  }
+
+  // Handle OR in superset: subset ⊆ (A OR B) if subset ⊆ A or subset ⊆ B.
+  // Must be checked before decomposing AND subsets so that and(A, B) can
+  // match a structurally equal disjunct via areExpressionsEqual.
+  if (superset.type === `func` && superset.name === `or`) {
+    return superset.args.some((arg) =>
+      isWhereSubsetInternal(subset, arg as BasicExpression<boolean>),
+    )
+  }
+
   // Handle subset being an AND: (A AND B) implies both A and B
   if (subset.type === `func` && subset.name === `and`) {
     // For (A AND B) ⊆ C, since (A AND B) implies A, we check if any conjunct implies C
@@ -111,22 +129,6 @@ function isWhereSubsetInternal(
     }
   }
 
-  // Handle OR in subset: (A OR B) is subset of C only if both A and B are subsets of C
-  if (subset.type === `func` && subset.name === `or`) {
-    return subset.args.every((arg) =>
-      isWhereSubsetInternal(arg as BasicExpression<boolean>, superset),
-    )
-  }
-
-  // Handle OR in superset: subset ⊆ (A OR B) if subset ⊆ A or subset ⊆ B
-  // (A OR B) as superset means data can satisfy A or B
-  // If subset is contained in any disjunct, it's contained in the union
-  if (superset.type === `func` && superset.name === `or`) {
-    return superset.args.some((arg) =>
-      isWhereSubsetInternal(subset, arg as BasicExpression<boolean>),
-    )
-  }
-
   // Handle comparison operators on the same field
   if (subset.type === `func` && superset.type === `func`) {
     const subsetFunc = subset as Func

packages/db/tests/query/predicate-utils.test.ts

@@ -345,6 +345,71 @@ describe(`isWhereSubset`, () => {
     })
   })
 
+  describe(`AND subset with OR superset`, () => {
+    it(`should recognize and(eq, isNull) as subset of or(and(eq, isNull), and(eq, isNull))`, () => {
+      const projectX = `4e164373-31b4-4b42-95c9-9c395cfb4916`
+      const projectY = `2fd4c147-2547-4b02-9554-9cd067187409`
+
+      const queryX = and(
+        eq(ref(`project_id`), val(projectX)),
+        func(`isNull`, ref(`soft_deleted_at`)),
+      )
+      const queryY = and(
+        eq(ref(`project_id`), val(projectY)),
+        func(`isNull`, ref(`soft_deleted_at`)),
+      )
+
+      const unionPredicate = or(queryX, queryY)
+
+      expect(isWhereSubset(queryX, unionPredicate)).toBe(true)
+      expect(isWhereSubset(queryY, unionPredicate)).toBe(true)
+    })
+
+    it(`should recognize and(A, B) as subset of or(and(A, B), and(C, D))`, () => {
+      const subsetExpr = and(eq(ref(`id`), val(1)), gt(ref(`age`), val(20)))
+      const supersetExpr = or(
+        and(eq(ref(`id`), val(1)), gt(ref(`age`), val(20))),
+        and(eq(ref(`id`), val(2)), gt(ref(`age`), val(30))),
+      )
+      expect(isWhereSubset(subsetExpr, supersetExpr)).toBe(true)
+    })
+
+    it(`should return false when and(A, B) matches no disjunct`, () => {
+      const subsetExpr = and(eq(ref(`id`), val(3)), gt(ref(`age`), val(20)))
+      const supersetExpr = or(
+        and(eq(ref(`id`), val(1)), gt(ref(`age`), val(20))),
+        and(eq(ref(`id`), val(2)), gt(ref(`age`), val(30))),
+      )
+      expect(isWhereSubset(subsetExpr, supersetExpr)).toBe(false)
+    })
+  })
+
+  describe(`isNull predicates`, () => {
+    it(`should return true for identical isNull expressions`, () => {
+      const a = func(`isNull`, ref(`deleted_at`))
+      const b = func(`isNull`, ref(`deleted_at`))
+      expect(isWhereSubset(a, b)).toBe(true)
+    })
+
+    it(`should return false for isNull on different fields`, () => {
+      const a = func(`isNull`, ref(`deleted_at`))
+      const b = func(`isNull`, ref(`created_at`))
+      expect(isWhereSubset(a, b)).toBe(false)
+    })
+
+    it(`should return true for and(eq, isNull) subset of identical and(eq, isNull)`, () => {
+      const subset = and(
+        eq(ref(`project_id`), val(`abc`)),
+        func(`isNull`, ref(`soft_deleted_at`)),
+      )
+      const superset = and(
+        eq(ref(`project_id`), val(`abc`)),
+        func(`isNull`, ref(`soft_deleted_at`)),
+      )
+      expect(isWhereSubset(subset, superset)).toBe(true)
+    })
+  })
+
   describe(`different fields`, () => {
     it(`should return false for different fields with no relationship`, () => {
       expect(

packages/electric-db-collection/src/electric.ts

@@ -431,6 +431,25 @@ function createLoadSubsetDedupe<T extends Row<unknown>>({
 
     const { cursor, where, orderBy, limit } = opts
 
+    // When the stream is already up-to-date, it may be in a long-poll wait.
+    // Forcing a disconnect-and-refresh ensures requestSnapshot gets a response
+    // from a fresh server round-trip rather than waiting for the current poll to end.
+    // If the refresh fails (e.g., PauseLock held during subscriber processing in
+    // join pipelines), we fall through to requestSnapshot which still works.
+    if (stream.isUpToDate) {
+      try {
+        await stream.forceDisconnectAndRefresh()
+      } catch (error) {
+        if (handleSnapshotError(error, `forceDisconnectAndRefresh`)) {
+          return
+        }
+        debug(
+          `${logPrefix}forceDisconnectAndRefresh failed, proceeding to requestSnapshot: %o`,
+          error,
+        )
+      }
+    }
+
     try {
       if (cursor) {
         const whereCurrentOpts: LoadSubsetOptions = {

packages/electric-db-collection/tests/electric.test.ts

@@ -21,10 +21,13 @@ import type { StandardSchemaV1 } from '@standard-schema/spec'
 const mockSubscribe = vi.fn()
 const mockRequestSnapshot = vi.fn()
 const mockFetchSnapshot = vi.fn()
+const mockForceDisconnectAndRefresh = vi.fn()
 const mockStream = {
   subscribe: mockSubscribe,
   requestSnapshot: mockRequestSnapshot,
   fetchSnapshot: mockFetchSnapshot,
+  forceDisconnectAndRefresh: mockForceDisconnectAndRefresh,
+  isUpToDate: false,
 }
 
 vi.mock(`@electric-sql/client`, async () => {
@@ -56,6 +59,8 @@ describe(`Electric Integration`, () => {
 
     // Reset mock requestSnapshot
     mockRequestSnapshot.mockResolvedValue(undefined)
+    mockForceDisconnectAndRefresh.mockResolvedValue(undefined)
+    mockStream.isUpToDate = false
 
     // Create collection with Electric configuration
     const config = {
@@ -2376,7 +2381,7 @@ describe(`Electric Integration`, () => {
       // In on-demand mode, calling loadSubset should request a snapshot
       await testCollection._sync.loadSubset({ limit: 10 })
 
-      // Verify requestSnapshot was called
+      expect(mockForceDisconnectAndRefresh).not.toHaveBeenCalled()
       expect(mockRequestSnapshot).toHaveBeenCalledWith(
         expect.objectContaining({
           limit: 10,
@@ -2385,6 +2390,65 @@ describe(`Electric Integration`, () => {
       )
     })
 
+    it(`should refresh the stream before requesting on-demand snapshots when already up-to-date`, async () => {
+      vi.clearAllMocks()
+
+      const config = {
+        id: `on-demand-refresh-before-snapshot-test`,
+        shapeOptions: {
+          url: `http://test-url`,
+          params: {
+            table: `test_table`,
+          },
+        },
+        syncMode: `on-demand` as const,
+        getKey: (item: Row) => item.id as number,
+        startSync: true,
+      }
+
+      const testCollection = createCollection(electricCollectionOptions(config))
+
+      mockStream.isUpToDate = true
+
+      await testCollection._sync.loadSubset({ limit: 10 })
+
+      expect(mockForceDisconnectAndRefresh).toHaveBeenCalledTimes(1)
+      expect(mockRequestSnapshot).toHaveBeenCalledTimes(1)
+      const refreshCall =
+        mockForceDisconnectAndRefresh.mock.invocationCallOrder[0]!
+      const snapshotCall = mockRequestSnapshot.mock.invocationCallOrder[0]!
+      expect(refreshCall).toBeLessThan(snapshotCall)
+    })
+
+    it(`should fall through to requestSnapshot when forceDisconnectAndRefresh fails`, async () => {
+      vi.clearAllMocks()
+
+      const config = {
+        id: `on-demand-refresh-fallthrough-test`,
+        shapeOptions: {
+          url: `http://test-url`,
+          params: {
+            table: `test_table`,
+          },
+        },
+        syncMode: `on-demand` as const,
+        getKey: (item: Row) => item.id as number,
+        startSync: true,
+      }
+
+      const testCollection = createCollection(electricCollectionOptions(config))
+
+      mockStream.isUpToDate = true
+      mockForceDisconnectAndRefresh.mockImplementationOnce(async () => {
+        throw new Error(`PauseLock held`)
+      })
+
+      await testCollection._sync.loadSubset({ limit: 10 })
+
+      expect(mockForceDisconnectAndRefresh).toHaveBeenCalledTimes(1)
+      expect(mockRequestSnapshot).toHaveBeenCalledTimes(1)
+    })
+
     it(`should fetch snapshots in progressive mode when loadSubset is called before sync completes`, async () => {
       vi.clearAllMocks()