Core Concepts

Sync Anchors & Delta Catch-Up

A client that goes offline and reconnects needs to recover lost state without re-downloading the entire database, and it needs a way to know when it does not need to download anything at all. Flux records the moment each channel was last confirmed current, asks your backend only what changed after it, and treats a confirmed empty answer as proof the channel is up to date.

Why delta catch-up replaces full re-bootstrap

Re-fetching entire datasets on every reconnect scales poorly and introduces state race conditions. Delta catch-up reduces recovery to one per-channel question: "What changed after timestamp X?" The answer is either rows to apply, a confirmation that nothing changed, or an honest "I cannot tell". Only that last answer ever costs a refetch.

Two records: the anchor and the coverage record

Each tab keeps an in-memory sync anchor per subscribed channel — the highest server timestamp it has applied — and hands the set to the adapter on every reconnect. This is the SyncAnchorMap. Anchors live in memory, so they are rebuilt after a reload.

sync_anchor_map.ts

1// What Flux hands your adapter on every reconnect attempt.
2type SyncAnchorMap = Record<string, number>
3 
4// Example — one entry per subscribed channel:
5{
6  'items': 1718000000000,
7  'notifications': 1718000001234,
8}

What survives a reload is the coverage record, persisted per channel in IndexedDB. Its confirmed-through time, not the age of the cache entry, is where the next catch-up starts. When the app loads from disk, anchors are re-seeded from it, so a returning visitor asks "what changed since I was last confirmed", not "what changed since I last wrote anything". Wherever an anchor and a coverage record disagree, the coverage record is the one that is trusted.

The coverage record

One record per channel, stored in the global scope or under the user's own namespace. User records are removed with the user's data on logout. Two clocks are used on purpose. confirmedThrough is server time, because it is sent to the backend as the "since" of a catch-up and must be comparable with server timestamps. Everything else is device time, only ever compared against other values written by tabs on the same device, so a wrong client clock cannot corrupt liveness checks.

coverage.ts

1// Persisted per channel in IndexedDB (global scope, or one record per user).
2interface ChannelCoverage {
3  // SERVER clock (ms). The disk copy is known complete through this instant.
4  // This is the "since" for the next catch-up.
5  confirmedThrough: number
6 
7  // DEVICE clock (ms) of the last confirmation, and how it was made.
8  confirmedAt: number
9  source: 'bootstrap' | 'refetch' | 'delta'
10 
11  // True while a leader tab holds an intact live subscription.
12  live: boolean
13  owner: string | null          // session id of that leader
14  heartbeatAt: number | null    // device clock of the leader's last beat
15 
16  // Why coverage is not live (null while live).
17  gapReason:
18    | 'not-yet-live'     // confirmed, but live coverage has not begun this session
19    | 'offline'          // the network dropped
20    | 'realtime-down'    // network up, socket not live for this channel
21    | 'unclean-exit'     // the leader went away without closing its claim
22    | 'catchup-failed'   // a catch-up failed or could not be verified (sticky)
23    | 'too-old'          // the gap is older than the backend can answer (sticky)
24    | null
25  gapOpenedAt: number | null
26}

Confirmed: a bootstrap, a selective refetch, or a catch-up that the backend answered sets the confirmed time. This happens only after the data reached disk.
Live: once a channel is confirmed and its socket is live, the leader tab claims it. While the claim holds, the leader writes a heartbeat (15 seconds by default), and each beat advances the confirmed time. A channel with a live claim and a fresh heartbeat is covered.
Gap: when the network drops or the socket drops, the claim ends at that instant and the confirmed time is advanced to it, since everything up to then was delivered. The gap and its reason are persisted, so closing the tab while offline loses nothing.
Crashed or closed leader: a leader that disappears cannot close its claim. Its heartbeat simply stops, and the next leader to be elected converts the orphaned claim into a gap that starts at the last heartbeat, so the catch-up begins at exactly the right point.

Two reasons are sticky

A gap of catchup-failed or too-old is a data problem, not a connectivity state. Going offline or losing the socket again does not overwrite it. Only a successful confirmation, normally a refetch of that channel, clears it. A channel can therefore never drift back to "looks fine" while a real question about it is still unanswered.

How adapters transmit sync anchors

Before opening or re-authenticating a stream connection, Flux supplies current anchors to the adapter. Each adapter conveys these anchors according to its underlying protocol:

WebSocket: Encodes anchors in the initial URL query (?syncAnchors=...) for eager catch-up, then executes an explicit SYNC_REQUEST handshake frame after subscription confirmation.
Socket.IO: Emits FLUX_SYNC_ANCHOR on connect, followed by a channel-scoped flux_sync_request handshake.
SSE: Appends ?syncAnchor=<unixMs> directly to the stream endpoint URL and processes backfill events upon connection.
Supabase: Invokes the flux_request_delta RPC function over Postgres Realtime.

The adapter contract

An adapter declares what it can do, reports a typed outcome for every catch-up, and optionally explains why a connection failed. All of it is optional, and anything left out is inferred from the methods you implement.

realtime_adapter.ts

1interface RealtimeAdapter {
2  // Declare what this adapter can do. Anything left out is inferred from the
3  // methods you implement, so older adapters keep working.
4  capabilities?: {
5    realtime?:       boolean  // false for the noop adapter
6    delta?:          boolean  // can answer "what changed since T"
7    reconcile?:      boolean
8    failureReasons?: boolean  // reports WHY a connection failed
9    requiresTable?:  boolean  // a registration with no 'table' has no binding
10  }
11 
12  // Called before every reconnect — encode these into your handshake
13  // however your transport supports it (query string, initial frame, headers).
14  notifyReconnect?(anchors: SyncAnchorMap): void
15 
16  // Catch up one channel, or many in one request. Return a typed outcome.
17  sync?(channel: string, lastKnownServerTime: number): Promise<SyncOutcome | void>
18  syncMany?(requests: { channel: string; since: number }[]):
19    Promise<Record<string, SyncOutcome> | void>
20 
21  // Fire this the moment a channel's subscription is confirmed by your
22  // server, so Flux can trigger catch-up.
23  onChannelSubscribed?(cb: (channel: string) => void): void
24 
25  // Report why a connection or channel failed, so an unstable network can be
26  // told apart from a quota limit (which must back off slowly).
27  onConnectFailure?(cb: (failure: RealtimeFailure) => void): void
28 
29  // How far back (ms) this backend can enumerate ALL changes, deletions
30  // included. A gap older than this is answered 'too-old' and refetched.
31  getDeltaRetentionWindow?(): number
32 
33  // ...connect, disconnect, subscribe, unsubscribe, onConnect, onDisconnect, destroy
34}

The outcome is what lets Flux tell "nothing changed" from "I could not find out". Those two must never look alike, because one means the channel is current and the other means it might not be.

sync_outcome.ts

1type SyncOutcome =
2  // Rows newer than 'since' were delivered through the subscribe callback.
3  | { status: 'applied'; rows: number; asOf?: number }
4 
5  // The backend POSITIVELY stated it checked 'since' to 'asOf' and nothing
6  // changed. Never return this if you could not see the change log.
7  | { status: 'confirmed-current'; asOf?: number }
8 
9  // 'since' is older than the backend can answer. Flux will refetch.
10  | { status: 'too-old' }
11 
12  // Unknown result. Coverage is not advanced and the gap stays open.
13  | { status: 'failed'; reason: 'timeout' | 'rpc-error' | 'not-subscribed'
14                              | 'disconnected' | 'unknown'; detail?: string }
15 
16type RealtimeFailure = {
17  reason: 'network' | 'limit' | 'auth' | 'unknown'
18  channel?: string        // when the failure is specific to one channel
19  retryAfterMs?: number   // backend hint, when it gives one
20}

The one rule for adapter authors

Return confirmed-current only when your backend positively answered. If your table has no changelog trigger, the function is missing, you lack permission to read the log, or the request timed out, return failed or declare delta: false. An empty reply that merely arrived is not a confirmation, and an adapter that reports one anyway can leave a channel permanently "up to date" while it silently misses changes.

What Flux does with each outcome

Applied or confirmed current. The channel is confirmed through the answer's asOf time and marked live. If the adapter gives no asOf, Flux uses the request start on the server clock, backed off by a few seconds so a later catch-up can never start after a change it missed. No refetch.

Too old. Either the adapter says so, or the gap is longer than the retention window it declared. Rows are not requested, since an empty answer would be meaningless. The channel opens a sticky gap and is refetched.

Failed, or nothing returned. Coverage is not advanced, the cache entry is not restamped fresh, and the channel opens a sticky gap that a refetch resolves. A sync() that returns nothing is treated as unverified: Flux does not assume it worked.

Before recording any of these, the engine waits for the rows that arrived with the answer to finish writing to disk. A failed write, or writes that do not finish in time, downgrade the result to failed. Coverage never gets ahead of the disk.

Catch-ups requested together are coalesced into one request (a 75 millisecond window), and channels on the same table share a single question that uses the oldest "since" among them. Each channel then keeps only the rows newer than its own.

The Single-State Changelog Architecture

A plain query on a live table cannot detect deletions that occurred while a client was offline. To guarantee deletion recovery without infinite log growth, Flux backends utilize a single-state changelog (FluxChangelog).

At most one record exists per row per table. Subsequent mutations perform an upsert, merging sparse field updates and preserving deleted tombstones.

Language & Database Agnostic

Every code sample on this page past this point is illustrative, shown in TypeScript and Node because that's what the Flux client and reference servers use — but nothing about the server side requires Node or JavaScript at all. Flux's adapters only care about the wire contract: specific event names for Socket.IO, specific JSON frame shapes for raw WebSockets, and specific query-string parameters for SSE. Whatever language your backend runs in — Python, Go, Ruby, Java, PHP — can implement the same single-state changelog pattern and query contract.

Supabase / Postgres Delta catch-up

In Supabase, delta backfill is handled directly in Postgres via a single-state changelog trigger and an RPC function that streams rows back over Realtime broadcast. Realtime subscribes to a table, so every registration you want covered needs a table. A registration without one is TTL-only.

lib/flux.ts (excerpt)

1const flux = createFlux({
2  adapter: createSupabaseAdapter(supabase, {
3    // Must match your flux_changelog cleanup retention. A gap older than this
4    // is answered 'too-old' locally, without a request, and refetched.
5    deltaRetentionMs: 14 * 24 * 60 * 60 * 1000,
6  }),
7})
8 
9flux.register({
10  channel: 'pricing_table',
11  table:   'pricing_table',   // required for realtime coverage on Supabase
12  idbKey:  'pricing_table:latest',
13  ttl:     'long',
14  event:   'UPDATE',
15  store:   pricingStoreAdapter,
16})

The adapter shares one channel per table, asks once per table (batching all tables into a single flux_request_delta_batch call when that function is installed, and falling back to one flux_request_delta call per table otherwise), and fans the returned rows out to each channel that asked. Rows that arrive are applied. A reply that arrives with no rows is reported as confirmed current. A channel whose table is not subscribed reports failed with not-subscribed, an RPC error reports failed with rpc-error, and a reply that never arrives reports failed with timeout after 30 seconds.

Every table you register needs the trigger

A table with no flux_changelog_trigger attached returns an empty reply every time, and an empty reply is read as "nothing changed". Attach the trigger to every table you register for realtime, and keep deltaRetentionMs equal to your changelog cleanup retention so a gap past it is refetched instead of trusted.

supabase_changelog.sql

1-- Flux Delta Sync for Supabase — paste this once in your project's
2-- SQL Editor. Nothing here runs on our servers.
3 
4-- 1. Single-state Changelog table — holds at most ONE row per entity.
5create table if not exists public.flux_changelog (
6  id         bigint generated always as identity primary key,
7  table_name text        not null,
8  row_id     text        not null,
9  op         text        not null check (op in ('INSERT', 'UPDATE', 'DELETE')),
10  row_data   jsonb,
11  changed_at timestamptz not null default now()
12);
13 
14create unique index if not exists flux_changelog_table_row_idx
15  on public.flux_changelog (table_name, row_id);
16 
17create index if not exists flux_changelog_table_changed_idx
18  on public.flux_changelog (table_name, changed_at);
19 
20-- 2. Sparse field-level diff for UPDATE operations.
21create or replace function public.flux_jsonb_diff(p_old jsonb, p_new jsonb)
22returns jsonb
23language sql
24immutable
25as $$
26  select jsonb_object_agg(key, value)
27  from jsonb_each(p_new)
28  where not (p_old ? key)
29     or (p_old -> key) <> value
30     or key = 'id';
31$$;
32 
33-- 3. Single-state Trigger function.
34create or replace function public.flux_log_change()
35returns trigger
36language plpgsql
37security definer
38set search_path = public
39as $$
40declare
41  pk_value      text;
42  v_op          text;
43  v_data        jsonb;
44  v_existing_op text;
45  v_existing    jsonb;
46begin
47  if TG_OP = 'DELETE' then
48    execute 'select ($1).id::text' into pk_value using OLD;
49    v_op   := 'DELETE';
50    v_data := jsonb_build_object('id', pk_value);
51 
52  elsif TG_OP = 'UPDATE' then
53    execute 'select ($1).id::text' into pk_value using NEW;
54    v_op   := 'UPDATE';
55    v_data := public.flux_jsonb_diff(to_jsonb(OLD), to_jsonb(NEW));
56 
57    select op, row_data into v_existing_op, v_existing
58    from public.flux_changelog
59    where table_name = TG_TABLE_NAME and row_id = pk_value;
60 
61    if v_existing_op = 'INSERT' then
62      v_op   := 'INSERT';
63      v_data := to_jsonb(NEW);
64    elsif v_existing_op = 'UPDATE' then
65      v_data := v_existing || v_data;
66    end if;
67 
68  else -- INSERT
69    execute 'select ($1).id::text' into pk_value using NEW;
70    v_op   := 'INSERT';
71    v_data := to_jsonb(NEW);
72  end if;
73 
74  delete from public.flux_changelog
75  where table_name = TG_TABLE_NAME
76    and row_id = pk_value;
77 
78  insert into public.flux_changelog (table_name, row_id, op, row_data)
79  values (TG_TABLE_NAME, pk_value, v_op, v_data);
80 
81  if TG_OP = 'DELETE' then
82    return OLD;
83  else
84    return NEW;
85  end if;
86end;
87$$;
88 
89-- 4. Attach to target table.
90drop trigger if exists flux_changelog_trigger on public.your_table;
91create trigger flux_changelog_trigger
92  after insert or update or delete on public.your_table
93  for each row execute function public.flux_log_change();
94 
95-- 5. RPC Function for delta retrieval over Supabase Realtime broadcast.
96create or replace function public.flux_request_delta(
97  p_channel  text,
98  p_table    text,
99  p_since_ms bigint,
100  p_limit    int default 2000
101)
102returns void
103language plpgsql
104security definer
105set search_path = public
106as $$
107declare
108  v_rows jsonb;
109begin
110  select coalesce(
111           jsonb_agg(
112             jsonb_build_object(
113               'op',        latest.op,
114               'data',      latest.row_data,
115               'timestamp', (extract(epoch from latest.changed_at) * 1000)::bigint
116             )
117             order by latest.changed_at asc
118           ),
119           '[]'::jsonb
120         )
121    into v_rows
122  from (
123    select distinct on (c.row_id)
124      c.row_id, c.op, c.row_data, c.changed_at
125    from public.flux_changelog c
126    where c.table_name = p_table
127      and c.changed_at > to_timestamp(p_since_ms / 1000.0)
128    order by c.row_id, c.changed_at desc
129    limit p_limit
130  ) latest;
131 
132  perform realtime.send(
133    jsonb_build_object('rows', v_rows),
134    'flux_sync_complete',
135    'flux:' || p_channel,
136    false
137  );
138end;
139$$;
140 
141grant execute on function public.flux_request_delta(text, text, bigint, int) to anon, authenticated;
142alter table public.flux_changelog enable row level security;

Socket.IO Delta catch-up

The Socket.IO adapter issues flux_sync_request for each channel and expects a flux_sync_complete response payload containing the array of delta rows from queryChangesSince.

server/socket_handler.ts

1// Node/Socket.IO server — queryChangesSince reads single-state changelog
2//
3// Client adapter emits 'flux_sync_request' { channel, last_sync_anchor }
4// Server emits back '${channel}:flux_sync_complete' { rows }
5socket.on('flux_sync_request', async ({ channel, last_sync_anchor }) => {
6  try {
7    const rows = await queryChangesSince(channel, last_sync_anchor)
8    socket.emit(`${channel}:flux_sync_complete`, {
9      rows: rows.map((r) => ({
10        op:        r.op,        // 'CREATE' | 'UPDATE' | 'DELETE'
11        data:      r.data,      // normalized record or sparse diff
12        timestamp: r.timestamp, // unix ms
13      })),
14    })
15  } catch (err) {
16    socket.emit(`${channel}:flux_sync_complete`, { rows: [] })
17  }
18})

server/socket_handler.py

1# Python/Socket.IO equivalent (python-socketio)
2@sio.on("flux_sync_request")
3async def on_sync_request(sid, data):
4    channel = data.get("channel")
5    anchor = data.get("last_sync_anchor", 0)
6    try:
7        rows = await query_changes_since(channel, anchor)
8        payload = [{"op": r["op"], "data": r["data"], "timestamp": r["timestamp"]} for r in rows]
9        await sio.emit(f"{channel}:flux_sync_complete", {"rows": payload}, to=sid)
10    except Exception:
11        await sio.emit(f"{channel}:flux_sync_complete", {"rows": []}, to=sid)

WebSocket Delta catch-up

The raw WebSocket adapter emits a JSON frame with action: 'SYNC_REQUEST' and expects action: 'SYNC_COMPLETE' carrying the backfill rows.

server/ws_handler.ts

1// Node/WebSocket server — queryChangesSince reads single-state changelog
2//
3// Client adapter sends JSON: { action: 'SYNC_REQUEST', channel, last_sync_anchor }
4// Server responds JSON: { action: 'SYNC_COMPLETE', channel, rows }
5ws.on('message', async (raw) => {
6  const frame = JSON.parse(raw.toString())
7  if (frame.action === 'SYNC_REQUEST' && frame.channel) {
8    const rows = await queryChangesSince(frame.channel, frame.last_sync_anchor)
9    ws.send(JSON.stringify({
10      action:  'SYNC_COMPLETE',
11      channel: frame.channel,
12      rows: rows.map((r) => ({
13        op:        r.op,
14        data:      r.data,
15        timestamp: r.timestamp,
16      })),
17    }))
18  }
19})

SSE Delta catch-up

The SSE adapter encodes the sync anchor into the stream HTTP query. The server reads req.query.syncAnchor, executes queryChangesSince, and streams backfill events down the open response body immediately after client registration.

server/routes/sse.js

1// Node/Express SSE route — GET /api/stream/:channel?syncAnchor=<unixMs>
2//
3// SSE backfill is streamed directly down the open response on connect/reconnect
4router.get('/:channel', async (req, res) => {
5  const { channel } = req.params
6  const syncAnchor = req.query.syncAnchor ? Number(req.query.syncAnchor) : 0
7 
8  res.setHeader('Content-Type', 'text/event-stream')
9  res.setHeader('Cache-Control', 'no-cache')
10  res.setHeader('Connection', 'keep-alive')
11  res.flushHeaders()
12 
13  registerSseClient(channel, res)
14 
15  // Unconditional backfill query (runs for syncAnchor >= 0)
16  const rows = await queryChangesSince(channel, syncAnchor)
17  rows.forEach((r) => {
18    const event = {
19      entity: channel,
20      id: r.data?.id ?? '',
21      op: r.op,
22      data: r.data,
23      timestamp: r.timestamp,
24      source: 'flux_sse_backfill',
25    }
26    res.write(`event: UPDATE\ndata: ${JSON.stringify(event)}\n\n`)
27  })
28})

MongoDB Change Stream Catch-Up

MongoDB backends consume driver Change Streams to execute upserts into a FluxChangelog collection. Catch-up queries scan this collection to return deduplicated delta rows.

server/watchers/items.js

1// Mongo Change Stream watcher creating single-state FluxChangelog
2const changeStream = Item.watch([], { fullDocument: 'updateLookup' })
3 
4changeStream.on('change', async (change) => {
5  const { operationType, documentKey, fullDocument, updateDescription } = change
6  const opMap = { insert: 'CREATE', update: 'UPDATE', replace: 'UPDATE', delete: 'DELETE' }
7  const op = opMap[operationType]
8  if (!op) return
9 
10  const rowId = documentKey._id.toString()
11 
12  let updatedFields = null
13  if (operationType === 'update' && updateDescription?.updatedFields) {
14    updatedFields = { ...updateDescription.updatedFields }
15    delete updatedFields.updatedAt
16    delete updatedFields.__v
17  }
18 
19  const rowData = fullDocument
20    ? { ...fullDocument.toObject?.() ?? fullDocument, id: rowId }
21    : { id: rowId }
22 
23  // logChange executes upsert on { tableName, rowId }, keeping changelog single-state
24  await logChange('items', rowId, op, rowData, updatedFields)
25})

The query handler queries the changelog using standard Mongoose/MongoDB logic:

server/lib/changelog.js

1// queryChangesSince implementation for MongoDB / Mongoose
2async function queryChangesSince(tableName, sinceMs) {
3  const sinceDate = new Date(Number(sinceMs) || 0)
4 
5  const rawRows = await FluxChangelog.find({
6    tableName,
7    changedAt: { $gt: sinceDate },
8  }).sort({ changedAt: -1 }).lean()
9 
10  const latestByRowId = new Map()
11  for (const row of rawRows) {
12    if (!latestByRowId.has(row.rowId)) latestByRowId.set(row.rowId, row)
13  }
14 
15  return Array.from(latestByRowId.values())
16    .sort((a, b) => new Date(a.changedAt).getTime() - new Date(b.changedAt).getTime())
17    .slice(0, 2000)
18    .map((r) => ({
19      op: r.op,
20      data: r.rowData,
21      timestamp: new Date(r.changedAt).getTime(),
22    }))
23}

Optional Adapter Features

Everything beyond subscribe and unsubscribe is optional on the adapter contract, and each missing piece degrades gracefully. With no delta support, a channel is realtime-only: live events keep it current, but any gap is closed by refetching that channel. With no realtime at all (the noop adapter, or a registration with no table on a table-based backend), a channel is TTL-only. An adapter that returns nothing from sync() is treated as unverified: Flux cannot prove the channel current, so it refetches it rather than guess. Delta catch-up is an enhancement a transport opts into.

Two engine settings control the coverage heartbeat. The defaults suit most apps; change them only if you have an unusual tab-throttling environment.

lib/flux.ts (excerpt)

1const flux = createFlux({
2  adapter: createSupabaseAdapter(supabase, {
3    // Must match your flux_changelog cleanup retention.
4    deltaRetentionMs: 14 * 24 * 60 * 60 * 1000,
5  }),
6 
7  // How often the leader proves it is still watching (ms). Default 15000.
8  coverageHeartbeatMs: 15_000,
9 
10  // How long a live claim survives without a beat before other tabs treat the
11  // channel as having a gap (ms). Keep it well above the beat interval:
12  // browsers throttle timers in background tabs. Default 120000.
13  coverageHeartbeatStaleMs: 120_000,
14})

For complete backend setups, refer to getting_started/websocket, getting_started/socketio, and getting_started/sse. For how catch-up results are tracked, see core_concepts/activity_bus.