Real Time Sync

Real-time sync

@0xx0lostcause0xx0/polypack/sync is a transport-agnostic sync layer for a PolyGraph: acknowledgements, retry, deduplication, reconnect recovery, and echo suppression. It provides optional durable server and client logs, authentication and conflict hooks, bounded batches, checksums, cursor recovery, and filtered subscriptions — it does not provide identity management, application permissions, or a domain-specific conflict resolver.

Client

import { SyncClient, MemoryTransport } from '@0xx0lostcause0xx0/polypack/sync'

const [clientTransport, serverTransport] = MemoryTransport.pair()
const client = new SyncClient({ graph, transport: clientTransport })

• SyncClient({ graph, transport, clientId?, autoFlush?, retryMs?, activationSyncThreshold?, stateStore? }) captures graph events, retains operations until acknowledged, retries unacknowledged deltas, detects server-cursor gaps, and applies remote operations with echo suppression. retryMs defaults to 1,000ms (0 disables automatic retry). activationSyncThreshold (default 0.05) drops coalesced activation deltas below that magnitude instead of syncing them.
• flush() — manual send. pendingOps — inspect outstanding operations. requestSync(fromStart?) — request catch-up. syncCursor — inspect progress. reconnect(transport) — replace a transport, resend pending work, and request missing server operations. disconnect() — flush pending operations and close.
• await SyncClient.restore({ graph, transport, stateStore, clientId? }) — resume a client after restart from durable state.
• await client.persist() — wait for the latest local state to be durable.

Client state stores

MemorySyncClientStateStore and FileSyncClientStateStore persist pending operations plus the acknowledged client and server cursors. The file store includes a checksum and rejects corrupted state.

Server

import { SyncServer } from '@0xx0lostcause0xx0/polypack/sync'

const server = new SyncServer()
const handleFromClient = server.addClient(handle)

SyncServer is an in-memory relay by default. Pass a SyncOperationLog as operationLog and await ready() before accepting clients, to restore and durably append the server history.

• maxBatchOps bounds an incoming envelope; maxPendingOps applies back-pressure to the durable queue. A rejected envelope receives an ack with pending_too_large — retry after the queue drains. A storage failure returns a persistence_error acknowledgement and causes await server.flush() to reject with the underlying error, so callers can retry after repairing the log.
• await server.flush() waits for all durable submissions accepted so far to reach the operation log.
• addClient(handle) returns that client's incoming-message handler; removeClient(handle) unregisters it; ops exposes the received log.
• The server deduplicates operations by client and sequence, operation, and transaction identity, acknowledges both first-time and repeated delivery, and serves full operation snapshots or cursor-based deltas for late and reconnecting clients.
• Authorization and conflict hooks validate transaction groups atomically — if one operation in a transaction is rejected, none of it is committed or broadcast.
• await server.logStats() reports cursor retention, retained operations, and operation/transaction identity counts for monitoring and compaction tooling.

Building blocks

• OpLog(clientId, existing?) — appends sequenced SyncOp values; exposes since(seq), all, latestSeq, size.
• SyncAdapter(inner, clientId) — wraps a persistence adapter and records successful node/edge writes in its oplog; set onOp to observe them.
• SyncTransport — the transport contract: send, onMessage, close. MemoryTransport.pair() creates linked asynchronous in-process transports, useful for tests and same-process client/server wiring.

Applications must detect transport failure themselves and supply a replacement transport to client.reconnect().

---

Back to Home.