Repro: TanStack DB on-demand subset load forces an extra same-offset shape refresh, tripping Electric's fast-retry-loop guard

README.md

Repro: each on-demand subset load forces a stream refresh at the same offset

@tanstack/db@0.9.2 · @tanstack/electric-db-collection@0.4.10 · @electric-sql/client@1.5.28 (latest at time of writing)

docker compose up -d     # Postgres + Electric, the pair TanStack DB's own e2e suite expects
npm install
node repro.mjs

Nothing here fakes the wire — it runs against a real Electric.

What it does

Ten live queries, each with a distinct where predicate, each fully awaited before the next is created. No concurrency, nothing superseded, nothing racing. The underlying table never changes.

What happens

distinct subsets requested   : 10
non-live shape requests      : 20
  snapshots (POST)           : 0
  chunk refetches (GET)      : 20
distinct offsets requested   : 4
requests per subset          : 2.00
subset requests              : 10

first six non-live requests:
  GET ?log=changes_only&offset=now&subset__params=...&subset__where="n" >= $1&table=items
  GET ?handle=...&log=changes_only&offset=0_inf&table=items
  GET ?handle=...&log=changes_only&offset=0_inf&subset__params=...&subset__where=...
  GET ?handle=...&log=changes_only&offset=0_inf&table=items
  GET ?handle=...&log=changes_only&offset=0_inf&subset__params=...&subset__where=...
  GET ?handle=...&log=changes_only&offset=0_inf&table=items

The requests alternate. Ten of them are the subset requests themselves, which are expected. The other ten carry no subset__ parameters at all and repeat the same offset=0_inf — they are the forceDisconnectAndRefresh() that precedes each requestSnapshot.

Ten identical same-offset requests is exactly the pattern @electric-sql/client treats as a broken cache, so its guard fires:

[Electric] Detected fast retry loop (5 requests in 500ms at the same offset).
Clearing client-side caches and resetting stream to recover. If this persists,
check that your proxy includes all query parameters (especially 'handle' and
'offset') in its cache key, and that required Electric headers are forwarded to
the client.

The proxy is fine. The duplicate requests come from the adapter.

Scope of this repro

It demonstrates the request amplification and the spurious guard trip. It does not demonstrate a permanently dead stream — the client recovers here. Higher subset churn is where that would be worth investigating, and this repro does not establish it.

docker-compose.yml

# Postgres + Electric, the same pair TanStack DB's own e2e suite expects
# (packages/db-collection-e2e/support/global-setup.ts). Ports are shifted off
# the defaults so this can run beside an existing stack.
services:
  postgres:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: password
      POSTGRES_DB: repro
    command:
      - postgres
      - -c
      - wal_level=logical
    ports:
      - "54399:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres"]
      interval: 1s
      timeout: 3s
      retries: 30

  electric:
    image: electricsql/electric:1.8.1
    environment:
      DATABASE_URL: postgresql://postgres:password@postgres:5432/repro?sslmode=disable
      ELECTRIC_INSECURE: "true"
    ports:
      - "3100:3000"
    depends_on:
      postgres:
        condition: service_healthy

package.json

{
  "name": "tanstack-db-ondemand-subset-refresh-repro",
  "private": true,
  "type": "module",
  "scripts": {
    "start": "node repro.mjs"
  },
  "dependencies": {
    "@tanstack/db": "0.9.2",
    "@tanstack/electric-db-collection": "0.4.10",
    "@electric-sql/client": "1.5.28",
    "pg": "^8"
  }
}

repro.mjs

/**
 * Repro: every `loadSubset` on an `on-demand` Electric collection costs more
 * than one non-live shape request, because each one forces a stream refresh.
 *
 *   docker compose up -d
 *   npm install
 *   node repro.mjs
 *
 * Runs against a real Postgres + Electric — the same pair TanStack DB's own
 * e2e suite expects (`packages/db-collection-e2e/support/global-setup.ts`), so
 * nothing here fakes the wire.
 *
 * Measured: N live queries with distinct predicates, each awaited in turn,
 * against a log that never changes. Reports how many non-live (chunk/snapshot)
 * requests those N subsets cost.
 */
import pg from "pg";

import { createCollection, createLiveQueryCollection, gte } from "@tanstack/db";
import { electricCollectionOptions } from "@tanstack/electric-db-collection";

const ELECTRIC = process.env.ELECTRIC_URL ?? `http://localhost:3100`;
const PG = process.env.DATABASE_URL ?? `postgresql://postgres:password@localhost:54399/repro`;
const SUBSETS = Number(process.env.SUBSETS ?? 10);

const seed = async () => {
  const client = new pg.Client({ connectionString: PG });
  await client.connect();
  await client.query(`drop table if exists items`);
  await client.query(`create table items (id text primary key, n int not null)`);
  for (let i = 0; i < 50; i++) {
    await client.query(`insert into items (id, n) values ($1, $2)`, [`row-${i}`, i]);
  }
  await client.end();
};

const run = async () => {
  await seed();

  const nonLive = [];

  const fetchClient = async (input, init) => {
    const url = new URL(typeof input === `string` ? input : input.url);

    if (url.searchParams.get(`live`) !== `true`) {
      nonLive.push({
        offset: url.searchParams.get(`offset`),
        method: init?.method ?? `GET`,
        subset: [...url.searchParams.keys()].some((k) => k.startsWith(`subset__`)),
        search: url.search.slice(0, 110),
      });
    }

    return fetch(input, init);
  };

  const collection = createCollection(
    electricCollectionOptions({
      id: `items`,
      syncMode: `on-demand`,
      getKey: (row) => row.id,
      shapeOptions: {
        url: `${ELECTRIC}/v1/shape`,
        params: { table: `items` },
        fetchClient,
      },
    }),
  );

  // One live query per distinct predicate, each fully awaited before the next
  // — no concurrency, nothing racing, nothing superseded.
  for (let cutoff = 0; cutoff < SUBSETS; cutoff++) {
    const query = createLiveQueryCollection({
      startSync: true,
      query: (q) => q.from({ row: collection }).where(({ row }) => gte(row.n, cutoff)),
    });

    await query.preload();
  }

  const snapshots = nonLive.filter((request) => request.method === `POST`).length;
  const chunks = nonLive.length - snapshots;

  console.log(`distinct subsets requested   : ${SUBSETS}`);
  console.log(`non-live shape requests      : ${nonLive.length}`);
  console.log(`  snapshots (POST)           : ${snapshots}`);
  console.log(`  chunk refetches (GET)      : ${chunks}`);
  console.log(`distinct offsets requested   : ${new Set(nonLive.map((r) => r.offset)).size}`);
  console.log(`requests per subset          : ${(nonLive.length / SUBSETS).toFixed(2)}`);
  console.log(`subset requests              : ${nonLive.filter((r) => r.subset).length}`);
  console.log(`collection status            : ${collection.status}`);
  console.log();
  console.log(`first six non-live requests:`);
  for (const r of nonLive.slice(0, 6)) console.log(`  ${r.method} ${r.search}`);

  process.exit(0);
};

await run();