Activity

1. robacourt

   added a commit that references this issue on Jul 15, 2026

   fix(sync-service): stop flaky ShapeCache await-consumer test under CI…

   ...

   4051e0f
2. simpautus commented on Oct 13, 2026

   

   simpautus

   on Oct 13, 2026

   More actions

   We hit something in our app's integration tests that might be related to the await_snapshot_start/4 should wait for consumer to come up flake here, though I'm not sure. Sharing in case it helps. Our reading of the code may well be off.

   What we see (1.8.1, Docker image)

   TanStack DB's electric collection in progressive mode sends the first GET offset=-1 and a subset POST offset=-1 (no handle) at about the same time for a new shape. Now and then the POST gets no response. 45 s later Electric logs:

   [error] Failed to await snapshot start for shape: timeout
   

   The GET for the same shape returns normally and goes live. In our suite this happened in about 3% of runs, with one log line per hang.

   Repro without our stack

   New shape per iteration: an insert, then one GET and 8 concurrent subset POSTs straight at Electric (script below). Against a local 1.8.1:

   • POSTs sent together with the GET: 10 of 400 shapes had a POST that never returned
   • POSTs held until the GET returned: 0 of 600

   What we think is happening

   With ELECTRIC_LOG_LEVEL=debug, the POST that hung was the only one whose Starting a wait on the snapshot line came before Writer for <handle> initialized. All the others came after it and got a reply:

   20:53:48.083 [info]  Creating new shape ... with handle 73340091-1791924828083643
   20:53:48.083 [debug] Starting Electric.Shapes.Consumer for 73340091-...
   20:53:48.084 [debug] Starting a wait on the snapshot 73340091-... for {#PID<0.3732.0>, ...}   <- hung request
   20:53:48.084 [debug] Writer for 73340091-... initialized
   20:53:48.084 [debug] Starting a wait on the snapshot 73340091-... for {#PID<0.3729.0>, ...}   (x7, all answered)
   20:53:48.085 [debug] Snapshot started shape_handle: 73340091-...
   20:54:33.087 request_id=... [error] Failed to await snapshot start for shape: timeout       <- from PID<0.3732.0>
   

   From reading the code, our guess is:

   1. ShapeCache adds the handle to ShapeStatus, starts the consumer and then sends it {:initialize_shape, ...}.
   2. A request that sees the handle can call :await_snapshot_start after the consumer has registered but before that message arrives. State.add_waiter/2 then puts it on the default InitialSnapshot from State.new/2.
   3. State.initialize/3 replaces initial_snapshot_state with InitialSnapshot.new(pg_snapshot), which would drop the waiter. Nothing replies to it, and the call runs into the 45 s timeout.

   If that's right, the test in this issue might hit the same window. Its await_snapshot_start polls while consumer startup is delayed, and a retry that lands just after the consumer registers would wait the full 45 s, so the 15 s bound wouldn't help. We haven't run your test suite, so this part is only a guess from reading it.

   Happy to be told we've misread this. Repro script, in case it's useful:

   race.mjs

   // node race.mjs <iterations>
   // Env: ELECTRIC_URL, ELECTRIC_SECRET, DATABASE_URL (the db Electric replicates)
   // Table: CREATE TABLE race_items (id text PRIMARY KEY, group_id text NOT NULL, status text NOT NULL);
   import { randomUUID } from 'node:crypto';
   import pg from 'pg';
   
   const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
   const N = Number(process.argv[2] ?? 300);
   const POSTS = 8;
   
   function shapeUrl(group) {
     const url = new URL('/v1/shape', process.env.ELECTRIC_URL);
     url.searchParams.set('offset', '-1');
     url.searchParams.set('table', 'race_items');
     url.searchParams.set('where', 'group_id = $1');
     url.searchParams.set('params[1]', group);
     url.searchParams.set('secret', process.env.ELECTRIC_SECRET);
     return url;
   }
   
   const timeout = (ms) => new Promise((resolve) => setTimeout(() => resolve('TIMEOUT'), ms));
   let hangs = 0;
   
   for (let i = 0; i < N; i++) {
     const group = `race-${randomUUID()}`;
     await pool.query('INSERT INTO race_items VALUES ($1, $2, $3)', [`${group}-1`, group, 'new']);
   
     const get = fetch(shapeUrl(group)).then((r) => r.text().then(() => r.status));
     const posts = Array.from({ length: POSTS }, () =>
       fetch(shapeUrl(group), {
         method: 'POST',
         headers: { 'content-type': 'application/json' },
         body: JSON.stringify({ where: 'id = $1', params: { 1: `${group}-1` } }),
       }).then((r) => r.text().then(() => r.status)),
     );
     const [, ...results] = await Promise.all([get, ...posts.map((p) => Promise.race([p, timeout(3000)]))]);
   
     if (results.includes('TIMEOUT')) {
       hangs++;
       console.log(`hang at ${i}: ${results.join(',')}`);
     }
   }
   
   console.log(`${hangs}/${N} shapes had a POST that never returned`);
   await pool.end();