Sync

@pleach/core/sync ships the cross-client sync primitives: version vectors for conflict detection, a hybrid logical clock for total ordering, durable outboxes that survive reloads and disconnects, and a SyncCoordinator that ties it all together.

The version vector is one of three concepts in the state-and-persistence cluster — alongside the storage adapter and the checkpointer — that together carry session state across restarts, rewinds, and concurrent writers. This page covers the concurrent-writer axis: Record<clientId, number> increments per write, compareVectors flags concurrent outcomes, and SyncCoordinator.resolveConflicts settles them by last-writer-wins rather than clobbering by arrival order.

The substrate's sync property is: concurrent writes are detected, not blindly applied in arrival order. Two clients writing the same session bump their respective version-vector entries; when one pushes to the server, the coordinator compares vectors and, on a concurrent outcome, resolves with last-writer-wins on updatedAt. Interactive conflict resolution — surfacing each conflict for the application to settle by hand — is an enterprise-tier surface, not part of open @pleach/core.

import {
  SyncCoordinator,
  type SyncCoordinatorConfig,
  type SyncCoordinatorLayeredConfig,
  type SyncCoordinatorFlatConfig,
  type SyncTransportConfig,
  type SyncEngineConfig,
  type OutboxFlushResult,
  HybridLogicalClock,
  InMemoryOutbox,
  IndexedDBOutbox,
  SupabaseOutbox,
  type OutboxEntry,
  type OutboxStorage,
  type SupabaseOutboxClient,
  emptyVector,
  createVector,
  incrementVersion,
  mergeVectors,
  compareVectors,
  hasSeen,
  getMissing,
} from "@pleach/core/sync";

Picking an outbox

The outbox buffers SessionChanges until the server confirms acceptance. Match it to your storage adapter.

All three implement OutboxStorage; build your own (Redis, SQLite) against the same interface if you need a different backing.

Wiring a SyncCoordinator

The config carries two distinct concerns: transport (where the sync service lives) and engine (how the client buffers, retries, and paces). The layered shape nests them separately so call sites communicate the layering at a glance.

import { SyncCoordinator, IndexedDBOutbox } from "@pleach/core/sync";

const outbox = new IndexedDBOutbox("pleach-outbox");

const coordinator = new SyncCoordinator({
  clientId,
  transport: {
    endpoint:  "/api/pleach/sync",
    authToken: await getAuthToken(),
  },
  engine: {
    intervalMs:  0,    // manual flushes only
    outbox,
    maxAttempts: 5,
  },
});

The SyncCoordinator is itself a SyncAdapter — it implements sync(request) against transport.endpoint via fetch. Hand it to the runtime as your transport, or call flushOutbox() directly to drain buffered changes on app start and on connectivity-restored.

SyncCoordinatorConfig

SyncCoordinatorConfig accepts either the layered shape (preferred, shown above) or the flat shape (back-compat, original v1.0). The constructor discriminates by the presence of a transport field and normalizes both into the same internal form, so existing code keeps working unchanged.

Layered shape — SyncCoordinatorLayeredConfig

SyncTransportConfig

SyncEngineConfig

Flat shape — SyncCoordinatorFlatConfig (deprecated)

The original v1.0 shape keeps every field at one nesting level:

const coordinator = new SyncCoordinator({
  syncEndpoint:      "/api/pleach/sync",
  clientId,
  authToken:         await getAuthToken(),
  outbox,
  maxOutboxAttempts: 5,
});

The flat shape is marked @deprecated but stays valid indefinitely for back-compat. Migrate at your own pace — the constructor accepts both and produces byte-identical runtime behavior.

OutboxFlushResult

Each flushOutbox() call returns per-cycle counts — surface them in the UI for a "syncing… / synced N changes" affordance.

const result: OutboxFlushResult = await coordinator.flushOutbox();
console.log(result.inspected, result.synced, result.failed, result.parked);

parked is the count of entries that hit engine.maxAttempts and are kept in the outbox with lastError populated, but skipped by future flushes until you reset them manually. The lastError field carries a free-text message from the most recent failed attempt — the server's error string, or the caught exception's message for a transport failure. It is a human-readable reason, not a structured code. Reach for DevTools' forceSync() to re-attempt the flush once you've confirmed the underlying issue (downed endpoint, expired auth token, full disk on the outbox store) is fixed.

Outbox constructors

IndexedDBOutbox and SupabaseOutbox both take their store name as a positional argument, defaulting to "harness-outbox". Pass a SupabaseOutboxClient (the minimal Supabase shape) to SupabaseOutbox as the first arg.

const inMemory = new InMemoryOutbox();
const indexed  = new IndexedDBOutbox("pleach-outbox");
const remote   = new SupabaseOutbox(supabase, "harness_outbox");

Version vectors

A version vector is Record<clientId, number>. Each client increments its own entry on every write. When two clients sync, comparing vectors gives one of four outcomes:

The free functions handle the math:

import {
  emptyVector,
  createVector,
  incrementVersion,
  mergeVectors,
  compareVectors,
  hasSeen,
  getMissing,
} from "@pleach/core/sync";

const local  = incrementVersion({ [clientId]: 2 }, clientId);
const remote = createVector(otherClient, 1);
// → createVector takes (clientId, version=1) — one entry at a time.

compareVectors(local, remote);
// → "concurrent" — neither dominates; conflict.

mergeVectors(local, remote);
// → { [clientId]: 3, [otherClient]: 1 } — element-wise max.

hasSeen(local, { [otherClient]: 1 });
// → false — local hasn't observed otherClient's change.

getMissing(local, remote);
// → { [otherClient]: 1 } — entries in remote that local is missing.

All functions are pure. Use them inside a custom SyncAdapter or in tests that simulate divergent client states.

HybridLogicalClock

Causal ordering across clients without wall-clock dependence.

import { HybridLogicalClock } from "@pleach/core/sync";

const clock = new HybridLogicalClock(clientId);
const ts1   = clock.now();                      // local tick
const ts2   = clock.receive(remoteTimestamp);   // advance past a remote ts

HybridLogicalClock.compare(ts1, ts2);
// → -1 | 0 | 1 (wallTime → logical → clientId tie-break)

The constructor takes clientId as a positional string. The returned HybridTimestamp is { wallTime, logical, clientId }; the static compare orders by wall time, falls back to logical counter, then breaks ties by clientId.localeCompare.

Use the HLC timestamp for ordering changes that need a single total order across clients (e.g. message insert order). The version vector is for conflict detection; the HLC is for ordering.

Conflict resolution

When compareVectors returns concurrent, the coordinator's resolveConflicts(local, remote) decides between local and remote using last-writer-wins on updatedAt:

const { resolved, strategy } = coordinator.resolveConflicts(local, remote);
// strategy: "local" | "remote"

This is the real, working self-host surface. resolveConflicts takes two SessionState values, compares their version vectors, and returns the winning state plus a coarse strategy label and the list of diverged paths. On a concurrent outcome it keeps whichever side has the later updatedAt (ties favor local). The published SyncCoordinator ships only this last-writer-wins path. Call it on your push handler where two vectors compare concurrent.

What's enterprise-tier / planned, not in open @pleach/core

Interactive conflict resolution — the surface a UI would use to let a person settle a conflict — is not shipped in the open package:

• The sync.conflict stream event is declared in the event-map type, but no code path emits it. A self-host build never receives one.
• runtime.resolveConflict(sessionId, conflictId, "local" | "remote") exists on the runtime, but in @pleach/core it is a stub: it logs and returns a receipt without merging or persisting anything, and there is no conflict store to look up by conflictId. CRDT merge and an interactive conflict surface are the enterprise upgrade.
• "merged" is likewise reserved for the CRDT path; the open build never returns it.

The 3xxx sync-error range is defined, not yet thrown

The 3xxx code range is a defined-but-not-yet-thrown enum. The constants live in errors/codes.ts, but no code path in open @pleach/core throws or attaches one — they describe a planned structured-error surface. Today a failed sync records a free-text reason instead: the outbox entry's lastError and each OutboxFlushResult.failedEntries[].reason carry the server's error string or a caught exception's message, not a 3xxx code.

The planned range:

A worked example of the working last-writer-wins path: two tabs write to session-018f-7a within the same second. Both pushes reach the coordinator; compareVectors reports concurrent, and resolveConflicts keeps whichever SessionState has the later updatedAt. The losing tab's concurrent edit is dropped — there is no interactive prompt and no sync.conflict event in the open package.

Connectivity awareness

ConnectivityMonitor watches navigator.onLine and actively probes a configurable URL on an interval. Use it to gate flushOutbox() calls rather than retrying through a downed network.

import { ConnectivityMonitor } from "@pleach/core/sync";

const monitor = new ConnectivityMonitor({
  probeUrl:        "/api/health",
  probeIntervalMs: 30_000,
  probeTimeoutMs:  5_000,
});

const unsubscribe = monitor.subscribe((state) => {
  if (state.online) void coordinator.flushOutbox();
});

monitor.start();

ConnectivityState carries { online, lastOnlineAt, lastOfflineAt, rttMs }. The active probe pauses while document.hidden is true, to avoid the tab-return storm where every accumulated probe resolves at once.

The IndexedDBOutbox makes this fully offline-safe — writes land in IndexedDB, the coordinator pushes when online, and the per-entry retry ladder (capped by engine.maxAttempts) handles transient failures before parking.

Force-syncing from DevTools

__HARNESS_DEVTOOLS__.syncStatus();   // current state
__HARNESS_DEVTOOLS__.forceSync();    // drain the outbox now

Useful during development when you want to verify a write made it through without waiting for the next flushIntervalMs tick.

Where to go next