diff --git a/examples/playground/package.json b/examples/playground/package.json index 11dccab4..acb5b7ed 100644 --- a/examples/playground/package.json +++ b/examples/playground/package.json @@ -8,7 +8,7 @@ "typecheck": "tsc -p tsconfig.json --noEmit", "build": "vite build", "preview": "vite preview", - "test:e2e": "pnpm -C ../../packages/discovery run build && pnpm -C ../../packages/sync-protocol/protocol run build && pnpm -C ../../packages/sync-protocol/server/core run build && pnpm -C ../../packages/treecrdt-auth run build && pnpm -C ../../packages/sync-protocol/material/sqlite run build && pnpm -C ../../packages/treecrdt-wa-sqlite-vendor run build && pnpm -C ../../packages/treecrdt-wa-sqlite run build:ts && playwright test", + "test:e2e": "pnpm -C ../../packages/discovery run build && pnpm -C ../../packages/sync-protocol/protocol run build && pnpm -C ../../packages/sync-protocol/server/core run build && pnpm -C ../../packages/treecrdt-auth run build && pnpm -C ../../packages/sync-protocol/material/sqlite run build && pnpm -C ../../packages/treecrdt-sync run build && pnpm -C ../../packages/treecrdt-wa-sqlite-vendor run build && pnpm -C ../../packages/treecrdt-wa-sqlite run build:ts && playwright test", "deploy": "pnpm run build && gh-pages -d dist" }, "dependencies": { @@ -17,6 +17,7 @@ "@treecrdt/crypto": "workspace:*", "@treecrdt/discovery": "workspace:*", "@treecrdt/interface": "workspace:*", + "@treecrdt/sync": "workspace:*", "@treecrdt/sync-protocol": "workspace:*", "@treecrdt/sync-sqlite": "workspace:*", "@treecrdt/wa-sqlite": "workspace:*", diff --git a/examples/playground/src/App.tsx b/examples/playground/src/App.tsx index 8db5a0a4..95f45a3d 100644 --- a/examples/playground/src/App.tsx +++ b/examples/playground/src/App.tsx @@ -468,7 +468,7 @@ export default function App() { liveAllEnabled, setLiveAllEnabled, toggleLiveChildren, - queueLocalOpsForSync, + queueOps, handleSync, handleScopedSync, postBroadcastMessage, @@ -498,11 +498,11 @@ export default function App() { const handleCommittedLocalOps = React.useCallback( (ops: Operation[]) => { - // The local write already committed. This only feeds playground sync and debug UI state. - queueLocalOpsForSync(ops); + // The local write already committed; publish it to sync and debug UI state. + queueOps(ops); recordOps(ops, { assumeSorted: true }); }, - [queueLocalOpsForSync, recordOps] + [queueOps, recordOps] ); const grantSubtreeToReplicaPubkey = React.useCallback( diff --git a/examples/playground/src/playground/hooks/usePlaygroundLiveSubscriptions.ts b/examples/playground/src/playground/hooks/usePlaygroundLiveSubscriptions.ts index 729d6617..7eb38e82 100644 --- a/examples/playground/src/playground/hooks/usePlaygroundLiveSubscriptions.ts +++ b/examples/playground/src/playground/hooks/usePlaygroundLiveSubscriptions.ts @@ -7,7 +7,8 @@ import { type SetStateAction, } from 'react'; import type { Operation } from '@treecrdt/interface'; -import type { SyncPeer, SyncSubscription } from '@treecrdt/sync-protocol'; +import { createInboundSync, type InboundSync, type InboundSyncOnceOptions } from '@treecrdt/sync'; +import type { Filter, SyncMessage, SyncPeer } from '@treecrdt/sync-protocol'; import type { DuplexTransport } from '@treecrdt/sync-protocol/transport'; import { hexToBytes16 } from '../../sync-v0'; @@ -18,156 +19,83 @@ export type PlaygroundSyncConnection = { detach: () => void; }; +function subscriptionFilterLabel(filter: Filter): string { + return 'all' in filter ? 'all' : 'children'; +} + +function childrenFilter(parentId: string): Filter { + return { children: { parent: hexToBytes16(parentId) } }; +} + export function usePlaygroundLiveSubscriptions(opts: { syncPeerRef: MutableRefObject | null>; - syncConnRef: MutableRefObject>; setSyncError: Dispatch>; authCanSyncAll: boolean; }) { - const { syncPeerRef, syncConnRef, setSyncError, authCanSyncAll } = opts; + const { syncPeerRef, setSyncError, authCanSyncAll } = opts; const [liveBusy, setLiveBusy] = useState(false); const [liveChildrenParents, setLiveChildrenParents] = useState>(() => new Set()); const [liveAllEnabled, setLiveAllEnabled] = useState(false); const liveChildrenParentsRef = useRef>(new Set()); - const liveChildSubsRef = useRef>>(new Map()); const liveAllEnabledRef = useRef(false); - const liveAllSubsRef = useRef>(new Map()); - const liveAllStartingRef = useRef>(new Set()); - const liveChildrenStartingRef = useRef>(new Set()); - const liveBusyCountRef = useRef(0); - - const beginLiveWork = () => { - liveBusyCountRef.current += 1; - setLiveBusy(true); - }; - - const endLiveWork = () => { - liveBusyCountRef.current = Math.max(0, liveBusyCountRef.current - 1); - setLiveBusy(liveBusyCountRef.current > 0); - }; - - const stopLiveAllForPeer = (peerId: string) => { - const existing = liveAllSubsRef.current.get(peerId); - if (!existing) return; - existing.stop(); - liveAllSubsRef.current.delete(peerId); - }; + const inboundSyncRef = useRef | null>(null); + const inboundSyncPeerRef = useRef | null>(null); - const stopAllLiveAll = () => { - for (const sub of liveAllSubsRef.current.values()) sub.stop(); - liveAllSubsRef.current.clear(); + const currentSubscriptionFilters = () => { + if (liveAllEnabledRef.current) return [{ all: {} } satisfies Filter]; + return Array.from(liveChildrenParentsRef.current).map(childrenFilter); }; - const startLiveAll = (peerId: string) => { - const conn = syncConnRef.current.get(peerId); + const ensureInboundSync = () => { const peer = syncPeerRef.current; - if (!conn || !peer) return; - - if (liveAllSubsRef.current.has(peerId)) return; - if (liveAllStartingRef.current.has(peerId)) return; - liveAllStartingRef.current.add(peerId); - beginLiveWork(); - - void (async () => { - let started = false; - const sub = peer.subscribe( - conn.transport, - { all: {} }, - { - immediate: true, - intervalMs: 0, - ...syncOnceOptionsForPeer(peerId, 1024), - }, - ); - liveAllSubsRef.current.set(peerId, sub); - void sub.done.catch((err) => { - if (!started) return; - console.error('Live sync(all) failed', err); - stopLiveAllForPeer(peerId); - setSyncError(formatSyncError(err)); - }); - - try { - await sub.ready; - started = true; - } catch (err) { - console.error('Live sync(all) initial catch-up failed', err); - stopLiveAllForPeer(peerId); - setSyncError(formatSyncError(err)); - } - })().finally(() => { - liveAllStartingRef.current.delete(peerId); - endLiveWork(); + if (!peer) return null; + if (inboundSyncRef.current && inboundSyncPeerRef.current === peer) { + return inboundSyncRef.current; + } + + inboundSyncRef.current?.close(); + + const inbound = createInboundSync({ + localPeer: peer, + syncOptions: (peerId) => syncOnceOptionsForPeer(peerId, 2048), + subscribeOptions: (peerId) => syncOnceOptionsForPeer(peerId, 1024), + onStatus: (status) => setLiveBusy(status.busy), + onError: ({ peerId, filter, error, phase }) => { + console.error( + `Inbound sync(${subscriptionFilterLabel(filter)}) ${phase} failed`, + peerId, + error, + ); + setSyncError(formatSyncError(error)); + }, }); + inboundSyncRef.current = inbound; + inboundSyncPeerRef.current = peer; + return inbound; }; - const stopLiveChildrenForPeer = (peerId: string) => { - const byParent = liveChildSubsRef.current.get(peerId); - if (!byParent) return; - for (const sub of byParent.values()) sub.stop(); - liveChildSubsRef.current.delete(peerId); + const applySubscriptions = () => { + ensureInboundSync()?.subscribe(currentSubscriptionFilters()); }; - const stopLiveChildren = (peerId: string, parentId: string) => { - const byParent = liveChildSubsRef.current.get(peerId); - if (!byParent) return; - const sub = byParent.get(parentId); - if (!sub) return; - sub.stop(); - byParent.delete(parentId); - if (byParent.size === 0) liveChildSubsRef.current.delete(peerId); + const addInboundPeer = (peerId: string, conn: PlaygroundSyncConnection) => { + const inbound = ensureInboundSync(); + if (!inbound) return; + inbound.addPeer(peerId, conn.transport as DuplexTransport>); + inbound.subscribe(currentSubscriptionFilters()); }; - const stopAllLiveChildren = () => { - for (const peerId of Array.from(liveChildSubsRef.current.keys())) - stopLiveChildrenForPeer(peerId); + const removeLivePeer = (peerId: string) => { + inboundSyncRef.current?.removePeer(peerId); }; - const startLiveChildren = (peerId: string, parentId: string) => { - const conn = syncConnRef.current.get(peerId); - const peer = syncPeerRef.current; - if (!conn || !peer) return; - - const existing = liveChildSubsRef.current.get(peerId); - if (existing?.has(parentId)) return; - const startKey = `${peerId}\u0000${parentId}`; - if (liveChildrenStartingRef.current.has(startKey)) return; - liveChildrenStartingRef.current.add(startKey); - beginLiveWork(); - - const byParent = existing ?? new Map(); - void (async () => { - let started = false; - const sub = peer.subscribe( - conn.transport, - { children: { parent: hexToBytes16(parentId) } }, - { - immediate: true, - intervalMs: 0, - ...syncOnceOptionsForPeer(peerId, 1024), - }, - ); - byParent.set(parentId, sub); - liveChildSubsRef.current.set(peerId, byParent); - void sub.done.catch((err) => { - if (!started) return; - console.error('Live sync failed', err); - stopLiveChildren(peerId, parentId); - setSyncError(formatSyncError(err)); - }); - - try { - await sub.ready; - started = true; - } catch (err) { - console.error('Live sync(children) initial catch-up failed', err); - stopLiveChildren(peerId, parentId); - setSyncError(formatSyncError(err)); - } - })().finally(() => { - liveChildrenStartingRef.current.delete(startKey); - endLiveWork(); - }); + const syncInboundOnce = async ( + filters: Filter | readonly Filter[], + opts?: InboundSyncOnceOptions, + ) => { + const inbound = ensureInboundSync(); + if (!inbound) throw new Error('Inbound sync is not ready yet.'); + await inbound.syncOnce(filters, opts); }; const toggleLiveChildren = (parentId: string) => { @@ -180,42 +108,21 @@ export function usePlaygroundLiveSubscriptions(opts: { }; const resetLiveWork = () => { - liveAllStartingRef.current.clear(); - liveChildrenStartingRef.current.clear(); - liveBusyCountRef.current = 0; + inboundSyncRef.current?.close(); + inboundSyncRef.current = null; + inboundSyncPeerRef.current = null; setLiveBusy(false); }; useEffect(() => { liveChildrenParentsRef.current = liveChildrenParents; - - const connections = syncConnRef.current; - for (const peerId of connections.keys()) { - for (const parentId of liveChildrenParents) startLiveChildren(peerId, parentId); - } - - for (const peerId of Array.from(liveChildSubsRef.current.keys())) { - if (!connections.has(peerId)) { - stopLiveChildrenForPeer(peerId); - continue; - } - const byParent = liveChildSubsRef.current.get(peerId); - if (!byParent) continue; - for (const parentId of Array.from(byParent.keys())) { - if (!liveChildrenParents.has(parentId)) stopLiveChildren(peerId, parentId); - } - } + applySubscriptions(); // eslint-disable-next-line react-hooks/exhaustive-deps }, [liveChildrenParents]); useEffect(() => { liveAllEnabledRef.current = liveAllEnabled; - const connections = syncConnRef.current; - if (liveAllEnabled) { - for (const peerId of connections.keys()) startLiveAll(peerId); - } else { - stopAllLiveAll(); - } + applySubscriptions(); // eslint-disable-next-line react-hooks/exhaustive-deps }, [liveAllEnabled]); @@ -230,16 +137,9 @@ export function usePlaygroundLiveSubscriptions(opts: { liveAllEnabled, setLiveAllEnabled, toggleLiveChildren, - liveChildrenParentsRef, - liveAllEnabledRef, - beginLiveWork, - endLiveWork, - startLiveAll, - stopLiveAllForPeer, - stopAllLiveAll, - startLiveChildren, - stopLiveChildrenForPeer, - stopAllLiveChildren, + addInboundPeer, + removeLivePeer, + syncInboundOnce, resetLiveWork, }; } diff --git a/examples/playground/src/playground/hooks/usePlaygroundSync.ts b/examples/playground/src/playground/hooks/usePlaygroundSync.ts index cf29d297..605ec5dd 100644 --- a/examples/playground/src/playground/hooks/usePlaygroundSync.ts +++ b/examples/playground/src/playground/hooks/usePlaygroundSync.ts @@ -5,12 +5,8 @@ import { resolveWebSocketAttachment, type ResolveWebSocketAttachmentResult, } from '@treecrdt/discovery'; -import { - SyncPeer, - deriveOpRefV0, - type Filter, - type SyncAuth, -} from '@treecrdt/sync-protocol'; +import { SyncPeer, type Filter, type SyncAuth } from '@treecrdt/sync-protocol'; +import { createOutboundSync, type OutboundSync } from '@treecrdt/sync'; import { createTreecrdtSyncBackendFromClient } from '@treecrdt/sync-sqlite'; import type { BroadcastPresenceAckMessageV1, @@ -21,7 +17,10 @@ import { createBrowserWebSocketTransport, } from '@treecrdt/sync-protocol/browser'; import { treecrdtSyncV0ProtobufCodec } from '@treecrdt/sync-protocol/protobuf'; -import { wrapDuplexTransportWithCodec, type DuplexTransport } from '@treecrdt/sync-protocol/transport'; +import { + wrapDuplexTransportWithCodec, + type DuplexTransport, +} from '@treecrdt/sync-protocol/transport'; import type { TreecrdtClient } from '@treecrdt/wa-sqlite'; import { hexToBytes16, type AuthGrantMessageV1 } from '../../sync-v0'; @@ -44,17 +43,27 @@ import { formatRemoteRouteDetail, formatSyncError, getBrowserDiscoveryRouteCache, + inboundSyncPeerIdsToDrop, isCapabilityRevokedError, isDiscoveryBootstrapUrl, isRemotePeerId, isTransientRemoteConnectError, - localOpUploadKey, normalizeSyncServerUrl, previewDiscoveryHost, - syncOnceOptionsForPeer, syncTimeoutMsForPeer, - withTimeout, } from '../syncHelpers'; + +const RECENT_SYNC_TARGET_MS = 5_000; +const NODE_ID_HEX_RE = /^[0-9a-f]{32}$/i; + +function isNodeIdHex(id: string): boolean { + return NODE_ID_HEX_RE.test(id); +} + +function childrenFilter(parentId: string): Filter { + return { children: { parent: hexToBytes16(parentId) } }; +} + type PlaygroundSyncApi = { peers: PeerInfo[]; remoteSyncStatus: RemoteSyncStatus; @@ -67,7 +76,7 @@ type PlaygroundSyncApi = { liveAllEnabled: boolean; setLiveAllEnabled: React.Dispatch>; toggleLiveChildren: (parentId: string) => void; - queueLocalOpsForSync: (ops?: Operation[]) => void; + queueOps: (ops: Operation[]) => void; handleSync: (filter: Filter) => Promise; handleScopedSync: () => Promise; postBroadcastMessage: ( @@ -123,6 +132,7 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn } = opts; const [syncBusy, setSyncBusy] = useState(false); + const [outboundBusy, setOutboundBusy] = useState(false); const [syncError, setSyncError] = useState(null); const [remoteSyncStatus, setRemoteSyncStatus] = useState({ state: 'disabled', @@ -133,6 +143,7 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn const onlineRef = useRef(true); useEffect(() => { onlineRef.current = online; + if (online) void outboundSyncRef.current?.flush(); }, [online]); const autoSyncJoinInitial = useRef(autoSyncJoin).current; @@ -144,6 +155,7 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn const syncPeerRef = useRef | null>(null); const syncConnRef = useRef>(new Map()); + const outboundSyncRef = useRef | null>(null); const { liveBusy, liveChildrenParents, @@ -151,128 +163,52 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn liveAllEnabled, setLiveAllEnabled, toggleLiveChildren, - liveChildrenParentsRef, - liveAllEnabledRef, - beginLiveWork, - endLiveWork, - startLiveAll, - stopLiveAllForPeer, - stopAllLiveAll, - startLiveChildren, - stopLiveChildrenForPeer, - stopAllLiveChildren, + addInboundPeer, + removeLivePeer, + syncInboundOnce, resetLiveWork, } = usePlaygroundLiveSubscriptions({ syncPeerRef, - syncConnRef, setSyncError, authCanSyncAll, }); - const remoteLivePushScheduledRef = useRef(false); - const remoteLivePushRunningRef = useRef(false); - const remoteLivePushNeedsFullSyncRef = useRef(false); - const remoteLivePushPendingOpsRef = useRef>(new Map()); const autoSyncDoneRef = useRef(false); const autoSyncInFlightRef = useRef(false); const autoSyncAttemptRef = useRef(0); const autoSyncPeerIdRef = useRef(null); - const queueRemoteUploadHints = (ops?: Operation[]) => { - if (!ops || ops.length === 0) { - remoteLivePushNeedsFullSyncRef.current = true; - return; - } - const pendingOps = remoteLivePushPendingOpsRef.current; - for (const op of ops) { - pendingOps.set(localOpUploadKey(op), op); - } + const queueOps = (ops: Operation[]) => { + outboundSyncRef.current?.queueOps(ops); }; - const queueLocalOpsForSync = (ops?: Operation[]) => { - void syncPeerRef.current?.notifyLocalUpdate(ops); - queueRemoteUploadHints(ops); - if (remoteLivePushRunningRef.current) { - remoteLivePushScheduledRef.current = true; - return; - } - remoteLivePushScheduledRef.current = true; - remoteLivePushRunningRef.current = true; - beginLiveWork(); - void (async () => { - try { - while (remoteLivePushScheduledRef.current) { - remoteLivePushScheduledRef.current = false; - if (!onlineRef.current) continue; - - const peer = syncPeerRef.current; - if (!peer) continue; - - const connections = syncConnRef.current; - const remotePeerIds = Array.from(connections.keys()).filter(isRemotePeerId); - if (remotePeerIds.length === 0) continue; - - const liveChildren = Array.from(liveChildrenParentsRef.current).filter((id) => - /^[0-9a-f]{32}$/i.test(id), - ); - const pendingOps = Array.from(remoteLivePushPendingOpsRef.current.values()); - const needsFullSync = remoteLivePushNeedsFullSyncRef.current; - remoteLivePushPendingOpsRef.current.clear(); - remoteLivePushNeedsFullSyncRef.current = false; - if ( - !needsFullSync && - pendingOps.length === 0 && - !liveAllEnabledRef.current && - liveChildren.length === 0 - ) - continue; - - for (const peerId of remotePeerIds) { - const conn = connections.get(peerId); - if (!conn) continue; - try { - if (pendingOps.length > 0) { - await withTimeout( - peer.pushOps(conn.transport, pendingOps, { - maxOpsPerBatch: PLAYGROUND_SYNC_MAX_OPS_PER_BATCH, - }), - syncTimeoutMsForPeer(peerId, { autoSync: true }), - `live push with ${peerId.slice(0, 8)}… timed out`, - ); - continue; - } - - if (needsFullSync || (liveAllEnabledRef.current && liveChildren.length === 0)) { - await withTimeout( - peer.syncOnce(conn.transport, { all: {} }, syncOnceOptionsForPeer(peerId, 1024)), - syncTimeoutMsForPeer(peerId, { autoSync: true }), - `live sync with ${peerId.slice(0, 8)}… timed out`, - ); - continue; - } - - for (const parentId of liveChildren) { - await withTimeout( - peer.syncOnce( - conn.transport, - { children: { parent: hexToBytes16(parentId) } }, - syncOnceOptionsForPeer(peerId, 1024), - ), - syncTimeoutMsForPeer(peerId, { autoSync: true }), - `live sync(children ${parentId.slice(0, 8)}…) with ${peerId.slice(0, 8)}… timed out`, - ); - } - } catch (err) { - console.error('Remote live sync push failed', err); - setSyncError(formatSyncError(err)); - if (!isCapabilityRevokedError(err)) dropPeerConnection(peerId); - } - } + const removePeerConnection = ( + peerId: string, + opts: { detach?: boolean; closeTransport?: boolean } = {}, + ) => { + const connections = syncConnRef.current; + const conn = connections.get(peerId); + if (conn) { + if (opts.detach) { + try { + conn.detach(); + } catch { + // ignore } - } finally { - remoteLivePushRunningRef.current = false; - endLiveWork(); } - })(); + if (opts.closeTransport) { + try { + (conn.transport as any).close?.(); + } catch { + // ignore + } + } + connections.delete(peerId); + } + + removeLivePeer(peerId); + + if (isRemotePeerId(peerId)) setRemotePeer(null); + else removeMeshPeer(peerId); }; const dropPeerConnection = (peerId: string) => { @@ -282,28 +218,19 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn return; } - const connections = syncConnRef.current; - const conn = connections.get(peerId); - if (!conn) return; - try { - conn.detach(); - } catch { - // ignore - } - try { - (conn.transport as any).close?.(); - } catch { - // ignore - } - connections.delete(peerId); - stopLiveAllForPeer(peerId); - stopLiveChildrenForPeer(peerId); + removePeerConnection(peerId, { detach: true, closeTransport: true }); + }; - if (isRemotePeerId(peerId)) setRemotePeer(null); - else removeMeshPeer(peerId); + const selectSyncTargetIds = (connections: ReadonlyMap) => { + const now = Date.now(); + const recentPeerIds = peers + .filter((p) => now - p.lastSeen < RECENT_SYNC_TARGET_MS) + .sort((a, b) => b.lastSeen - a.lastSeen) + .map((p) => p.id); + return recentPeerIds.length > 0 ? recentPeerIds : Array.from(connections.keys()); }; - const handleSync = async (filter: Filter) => { + const syncFiltersWithTargets = async (filters: readonly Filter[], label: string) => { if (!onlineRef.current) { setSyncError('Offline: toggle Online to sync.'); return; @@ -322,115 +249,38 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn setSyncBusy(true); setSyncError(null); try { - const now = Date.now(); - const recentPeerIds = peers - .filter((p) => now - p.lastSeen < 5_000) - .sort((a, b) => b.lastSeen - a.lastSeen) - .map((p) => p.id); - const targets = recentPeerIds.length > 0 ? recentPeerIds : Array.from(connections.keys()); - let successes = 0; - let lastErr: unknown = null; + const targets = selectSyncTargetIds(connections).filter((peerId) => connections.has(peerId)); for (const peerId of targets) { const conn = connections.get(peerId); - if (!conn) continue; - const perPeerTimeoutMs = syncTimeoutMsForPeer(peerId, { - multipleTargets: targets.length > 1, - }); - try { - await withTimeout( - peer.syncOnce(conn.transport, filter, syncOnceOptionsForPeer(peerId, 2048)), - perPeerTimeoutMs, - `sync with ${peerId.slice(0, 8)}… timed out`, - ); - successes += 1; - } catch (err) { - lastErr = err; - console.error('Sync failed for peer', peerId, err); - if (!isCapabilityRevokedError(err)) dropPeerConnection(peerId); - } - } - if (successes === 0) { - if (lastErr) throw lastErr; - throw new Error('No peers responded to sync.'); + if (conn) addInboundPeer(peerId, conn); } + await syncInboundOnce(filters, { + peerIds: targets, + syncTimeoutMs: (peerId) => + syncTimeoutMsForPeer(peerId, { multipleTargets: targets.length > 1 }), + }); await refreshMeta(); } catch (err) { - console.error('Sync failed', err); + console.error(`${label} failed`, err); + for (const peerId of inboundSyncPeerIdsToDrop(err)) dropPeerConnection(peerId); setSyncError(formatSyncError(err)); } finally { setSyncBusy(false); } }; + const handleSync = async (filter: Filter) => { + await syncFiltersWithTargets([filter], 'Sync'); + }; + const handleScopedSync = async () => { const parents = new Set(getLoadedParentIds()); parents.add(viewRootId); if (viewRootId !== ROOT_ID) parents.delete(ROOT_ID); - const parentIds = Array.from(parents).filter((id) => /^[0-9a-f]{32}$/i.test(id)); + const parentIds = Array.from(parents).filter(isNodeIdHex); parentIds.sort(); - if (!onlineRef.current) { - setSyncError('Offline: toggle Online to sync.'); - return; - } - const peer = syncPeerRef.current; - if (!peer) { - setSyncError('Sync peer is not ready yet.'); - return; - } - const connections = syncConnRef.current; - if (connections.size === 0) { - setSyncError('No peers discovered yet.'); - return; - } - - setSyncBusy(true); - setSyncError(null); - try { - const now = Date.now(); - const recentPeerIds = peers - .filter((p) => now - p.lastSeen < 5_000) - .sort((a, b) => b.lastSeen - a.lastSeen) - .map((p) => p.id); - const targets = recentPeerIds.length > 0 ? recentPeerIds : Array.from(connections.keys()); - let successes = 0; - let lastErr: unknown = null; - for (const peerId of targets) { - const conn = connections.get(peerId); - if (!conn) continue; - const perPeerTimeoutMs = syncTimeoutMsForPeer(peerId, { - multipleTargets: targets.length > 1, - }); - try { - for (const parentId of parentIds) { - await withTimeout( - peer.syncOnce( - conn.transport, - { children: { parent: hexToBytes16(parentId) } }, - syncOnceOptionsForPeer(peerId, 2048), - ), - perPeerTimeoutMs, - `sync(children ${parentId.slice(0, 8)}…) with ${peerId.slice(0, 8)}… timed out`, - ); - } - successes += 1; - } catch (err) { - lastErr = err; - console.error('Scoped sync failed for peer', peerId, err); - if (!isCapabilityRevokedError(err)) dropPeerConnection(peerId); - } - } - if (successes === 0) { - if (lastErr) throw lastErr; - throw new Error('No peers responded to sync.'); - } - await refreshMeta(); - } catch (err) { - console.error('Scoped sync failed', err); - setSyncError(formatSyncError(err)); - } finally { - setSyncBusy(false); - } + await syncFiltersWithTargets(parentIds.map(childrenFilter), 'Scoped sync'); }; const postBroadcastMessage = ( @@ -472,13 +322,12 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn } if (!peerId) return; - const peer = syncPeerRef.current; const conn = connections.get(peerId); - if (!peer || !conn) return; + if (!conn) return; if (!authCanSyncAll) { const clean = viewRootId.toLowerCase(); - if (clean === ROOT_ID || !/^[0-9a-f]{32}$/.test(clean)) return; + if (clean === ROOT_ID || !isNodeIdHex(clean)) return; } if (autoSyncAttemptRef.current >= 3) return; @@ -489,23 +338,12 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn setSyncBusy(true); setSyncError(null); try { - if (authCanSyncAll) { - await withTimeout( - peer.syncOnce(conn.transport, { all: {} }, syncOnceOptionsForPeer(peerId, 2048)), - syncTimeoutMsForPeer(peerId, { autoSync: true }), - `auto sync with ${peerId.slice(0, 8)}… timed out`, - ); - } else { - await withTimeout( - peer.syncOnce( - conn.transport, - { children: { parent: hexToBytes16(viewRootId) } }, - syncOnceOptionsForPeer(peerId, 2048), - ), - syncTimeoutMsForPeer(peerId, { autoSync: true }), - `auto sync(children ${viewRootId.slice(0, 8)}…) with ${peerId.slice(0, 8)}… timed out`, - ); - } + const filter: Filter = authCanSyncAll ? { all: {} } : childrenFilter(viewRootId); + addInboundPeer(peerId, conn); + await syncInboundOnce(filter, { + peerIds: [peerId], + syncTimeoutMs: (targetPeerId) => syncTimeoutMsForPeer(targetPeerId, { autoSync: true }), + }); await refreshMeta(); @@ -519,7 +357,9 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn console.error('Auto sync failed', err); setSyncError(formatSyncError(err)); autoSyncPeerIdRef.current = null; - if (!isCapabilityRevokedError(err)) dropPeerConnection(peerId); + for (const failedPeerId of inboundSyncPeerIdsToDrop(err, [peerId])) { + dropPeerConnection(failedPeerId); + } } finally { autoSyncInFlightRef.current = false; setSyncBusy(false); @@ -661,29 +501,66 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn }, }; - const sharedPeer = new SyncPeer(backend, { + const localPeer = new SyncPeer(backend, { maxCodewords: PLAYGROUND_SYNC_MAX_CODEWORDS, maxOpsPerBatch: PLAYGROUND_SYNC_MAX_OPS_PER_BATCH, - deriveOpRef: (op, ctx) => - deriveOpRefV0(ctx.docId, { replica: op.meta.id.replica, counter: op.meta.id.counter }), ...(syncAuth ? { auth: syncAuth, } : {}), }); - syncPeerRef.current = sharedPeer; + syncPeerRef.current = localPeer; const connections = new Map; detach: () => void }>(); syncConnRef.current = connections; + const outboundSync = createOutboundSync({ + localPeer, + isOnline: () => onlineRef.current, + pushOptions: () => ({ + maxOpsPerBatch: PLAYGROUND_SYNC_MAX_OPS_PER_BATCH, + }), + pushTimeoutMs: (peerId) => syncTimeoutMsForPeer(peerId, { autoSync: true }), + onStatus: (status) => setOutboundBusy(status.running || status.scheduled), + onError: ({ targetId, error }) => { + console.error('Remote op upload failed', error); + setSyncError(formatSyncError(error)); + if (!isCapabilityRevokedError(error)) dropPeerConnection(targetId); + }, + }); + outboundSyncRef.current = outboundSync; + const maybeStartLiveForPeer = (peerId: string) => { if (!isRemotePeerId(peerId)) { const mesh = presenceMeshRef.current; if (!mesh || !mesh.isPeerReady(peerId)) return; } - if (liveAllEnabledRef.current) startLiveAll(peerId); - for (const parentId of liveChildrenParentsRef.current) startLiveChildren(peerId, parentId); + const conn = connections.get(peerId); + if (!conn) return; + addInboundPeer(peerId, conn); + }; + + const queueAutoSyncForPeer = (peerId: string) => { + if (!autoSyncJoinInitial || !joinMode || autoSyncDoneRef.current) return; + autoSyncPeerIdRef.current = peerId; + // Ensure the auto-sync effect runs even if peer readiness toggles without changing `peers.length`. + bumpAutoSyncJoinTick((t) => t + 1); + }; + + const registerPeerConnection = ( + peerId: string, + transport: DuplexTransport, + opts: { outbound?: boolean; markRemoteSeen?: boolean } = {}, + ) => { + const detach = opts.outbound + ? outboundSync.attachTarget(peerId, transport) + : localPeer.attach(transport); + connections.set(peerId, { transport, detach }); + if (opts.markRemoteSeen) setRemotePeer({ id: peerId, lastSeen: Date.now() }); + maybeStartLiveForPeer(peerId); + queueAutoSyncForPeer(peerId); + return detach; }; const mesh = channel @@ -698,27 +575,13 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn }, onPeerReady: (peerId) => { maybeStartLiveForPeer(peerId); - if (autoSyncJoinInitial && joinMode && !autoSyncDoneRef.current) { - autoSyncPeerIdRef.current = peerId; - // Ensure the auto-sync effect runs even if peer readiness toggles without changing `peers.length`. - bumpAutoSyncJoinTick((t) => t + 1); - } + queueAutoSyncForPeer(peerId); }, onPeerTransport: (peerId, transport) => { - const detach = sharedPeer.attach(transport); - connections.set(peerId, { transport, detach }); - maybeStartLiveForPeer(peerId); - if (autoSyncJoinInitial && joinMode && !autoSyncDoneRef.current) { - autoSyncPeerIdRef.current = peerId; - bumpAutoSyncJoinTick((t) => t + 1); - } - return detach; + return registerPeerConnection(peerId, transport); }, onPeerDisconnected: (peerId) => { - connections.delete(peerId); - stopLiveAllForPeer(peerId); - stopLiveChildrenForPeer(peerId); - removeMeshPeer(peerId); + removePeerConnection(peerId); }, onBroadcastMessage: (data) => { if (!data || typeof data !== 'object') return; @@ -745,12 +608,15 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn let remoteSocket: WebSocket | null = null; let remotePeerId: string | null = null; - let disposed = false; + let remoteSocketDisposed = false; let remoteOpened = false; let resolvedRemote: ResolveWebSocketAttachmentResult | null = null; - const discoveryRouteCache = getBrowserDiscoveryRouteCache(); if (remoteSyncUrl.length > 0) { + const discoveryRouteCache = getBrowserDiscoveryRouteCache(); + const isRemoteSocketCurrent = () => + !remoteSocketDisposed && syncConnRef.current === connections; + void (async () => { try { setSyncError((prev) => (isTransientRemoteConnectError(prev) ? null : prev)); @@ -766,7 +632,8 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn ? window.fetch.bind(window) : undefined, }); - if (disposed || syncConnRef.current !== connections) return; + if (!isRemoteSocketCurrent()) return; + const remoteUrl = resolvedRemote.url; const connectVerb = resolvedRemote.source === 'network' @@ -783,7 +650,7 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn remoteSocket.binaryType = 'arraybuffer'; remoteSocket.addEventListener('open', () => { - if (disposed || syncConnRef.current !== connections) return; + if (!isRemoteSocketCurrent()) return; if (!remoteSocket || remoteSocket.readyState !== WebSocket.OPEN || !remotePeerId) return; remoteOpened = true; @@ -797,20 +664,14 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn wire, treecrdtSyncV0ProtobufCodec as any, ); - const detach = sharedPeer.attach(transport); - syncConnRef.current.set(remotePeerId, { transport, detach }); - setRemotePeer({ id: remotePeerId, lastSeen: Date.now() }); - maybeStartLiveForPeer(remotePeerId); - - if (autoSyncJoinInitial && joinMode && !autoSyncDoneRef.current) { - autoSyncPeerIdRef.current = remotePeerId; - bumpAutoSyncJoinTick((t) => t + 1); - } + registerPeerConnection(remotePeerId, transport, { + outbound: true, + markRemoteSeen: true, + }); }); remoteSocket.addEventListener('message', () => { - if (disposed || syncConnRef.current !== connections) return; - if (!remotePeerId) return; + if (!isRemoteSocketCurrent() || !remotePeerId) return; setRemotePeer({ id: remotePeerId, lastSeen: Date.now() }); setRemoteSyncStatus((prev) => prev.state === 'connected' @@ -824,7 +685,7 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn remoteSocket.addEventListener('close', () => { if (syncConnRef.current !== connections) return; - if (!disposed) { + if (!remoteSocketDisposed) { setRemoteSyncStatus({ detail: formatRemoteErrorDetail( remoteOpened ? 'disconnected' : 'could_not_connect', @@ -837,12 +698,11 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn if (!remoteOpened && resolvedRemote?.source === 'cache' && resolvedRemote.cacheKey) { void discoveryRouteCache?.delete(resolvedRemote.cacheKey); } - if (!remotePeerId) return; - dropPeerConnection(remotePeerId); + if (remotePeerId) dropPeerConnection(remotePeerId); }); remoteSocket.addEventListener('error', () => { - if (syncConnRef.current !== connections) return; + if (!isRemoteSocketCurrent()) return; setRemoteSyncStatus({ detail: formatRemoteErrorDetail( remoteOpened ? 'connection_error' : 'could_not_reach', @@ -857,7 +717,7 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn setSyncError((prev) => prev ?? `Remote sync socket error (${remoteUrl.host})`); }); } catch (err) { - if (disposed || syncConnRef.current !== connections) return; + if (!isRemoteSocketCurrent()) return; setRemoteSyncStatus({ state: isDiscoveryBootstrapUrl(remoteSyncUrl) ? 'error' : 'invalid', detail: formatSyncError(err), @@ -867,24 +727,27 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn })(); } + const stopRemoteSocket = () => { + remoteSocketDisposed = true; + if (!remoteSocket) return; + try { + remoteSocket.close(); + } catch { + // ignore + } + }; + return () => { - disposed = true; - stopAllLiveAll(); - stopAllLiveChildren(); if (presenceMeshRef.current === mesh) presenceMeshRef.current = null; mesh?.stop(); - if (remoteSocket) { - try { - remoteSocket.close(); - } catch { - // ignore - } - } + stopRemoteSocket(); if (broadcastChannelRef.current === channel) broadcastChannelRef.current = null; - if (syncPeerRef.current === sharedPeer) syncPeerRef.current = null; + if (syncPeerRef.current === localPeer) syncPeerRef.current = null; + outboundSync.close(); + if (outboundSyncRef.current === outboundSync) { + outboundSyncRef.current = null; + } channel?.close(); - remoteLivePushScheduledRef.current = false; - remoteLivePushRunningRef.current = false; resetLiveWork(); connections.clear(); resetPeers(); @@ -910,7 +773,7 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn peers, remoteSyncStatus, syncBusy, - liveBusy, + liveBusy: liveBusy || outboundBusy, syncError, setSyncError, liveChildrenParents, @@ -918,7 +781,7 @@ export function usePlaygroundSync(opts: UsePlaygroundSyncOptions): PlaygroundSyn liveAllEnabled, setLiveAllEnabled, toggleLiveChildren, - queueLocalOpsForSync, + queueOps, handleSync, handleScopedSync, postBroadcastMessage, diff --git a/examples/playground/src/playground/syncErrorHelpers.ts b/examples/playground/src/playground/syncErrorHelpers.ts new file mode 100644 index 00000000..f77a110b --- /dev/null +++ b/examples/playground/src/playground/syncErrorHelpers.ts @@ -0,0 +1,64 @@ +type InboundSyncFailureLike = { + peerId: string; + error: unknown; +}; + +function inboundSyncFailures(error: unknown): readonly InboundSyncFailureLike[] | undefined { + if (!(error instanceof Error) || error.name !== 'InboundSyncAggregateError') return undefined; + const failures = (error as Error & { failures?: unknown }).failures; + if (!Array.isArray(failures)) return undefined; + + return failures.filter( + (failure): failure is InboundSyncFailureLike => + typeof failure === 'object' && + failure !== null && + typeof (failure as { peerId?: unknown }).peerId === 'string' && + 'error' in failure, + ); +} + +function syncErrorCauses(error: unknown): readonly unknown[] { + if (!(error instanceof Error)) return [error]; + const errors = (error as Error & { errors?: unknown }).errors; + return Array.isArray(errors) ? errors : [error]; +} + +export function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +export function isCapabilityRevokedError(error: unknown): boolean { + return syncErrorCauses(error).some((cause) => + /capability token revoked/i.test(errorMessage(cause)), + ); +} + +export function formatSyncError(error: unknown): string { + const causes = syncErrorCauses(error); + if (causes.some(isCapabilityRevokedError)) { + return 'Access revoked for this capability. Import/update access, then sync again.'; + } + if (causes.some((cause) => /unknown author:/i.test(errorMessage(cause)))) { + return 'This document contains ops from an author whose capability token is not available here yet. Sync from a peer that has the full author history, or try a fresh doc.'; + } + + const messages = Array.from(new Set(causes.map(errorMessage))); + if (messages.length === 1) return messages[0]!; + return `${errorMessage(error)}: ${messages.join('; ')}`; +} + +export function inboundSyncPeerIdsToDrop( + error: unknown, + fallbackPeerIds: readonly string[] = [], +): readonly string[] { + const failures = inboundSyncFailures(error); + if (!failures) { + return isCapabilityRevokedError(error) ? [] : Array.from(new Set(fallbackPeerIds)); + } + + const peerIds = new Set(); + for (const failure of failures) { + if (!isCapabilityRevokedError(failure.error)) peerIds.add(failure.peerId); + } + return Array.from(peerIds); +} diff --git a/examples/playground/src/playground/syncHelpers.ts b/examples/playground/src/playground/syncHelpers.ts index 6b2249e1..e51dcc83 100644 --- a/examples/playground/src/playground/syncHelpers.ts +++ b/examples/playground/src/playground/syncHelpers.ts @@ -1,5 +1,3 @@ -import type { Operation } from '@treecrdt/interface'; -import { bytesToHex } from '@treecrdt/interface/ids'; import { createStringStoreRouteCache, isDiscoveryBootstrapUrl, @@ -16,22 +14,6 @@ import { const REMOTE_SYNC_CODEWORDS_PER_MESSAGE = 512; -export function withTimeout(promise: Promise, ms: number, message: string): Promise { - return new Promise((resolve, reject) => { - const timer = setTimeout(() => reject(new Error(message)), ms); - promise.then( - (value) => { - clearTimeout(timer); - resolve(value); - }, - (err) => { - clearTimeout(timer); - reject(err); - }, - ); - }); -} - export function normalizeSyncServerUrl(raw: string, docId: string): URL { return normalizeDirectSyncWebSocketUrl(raw, docId); } @@ -60,24 +42,6 @@ export function previewDiscoveryHost(raw: string): string { return url.host; } -export function errorMessage(err: unknown): string { - return err instanceof Error ? err.message : String(err); -} - -export function isCapabilityRevokedError(err: unknown): boolean { - return /capability token revoked/i.test(errorMessage(err)); -} - -export function formatSyncError(err: unknown): string { - if (isCapabilityRevokedError(err)) { - return 'Access revoked for this capability. Import/update access, then sync again.'; - } - if (/unknown author:/i.test(errorMessage(err))) { - return 'This document contains ops from an author whose capability token is not available here yet. Sync from a peer that has the full author history, or try a fresh doc.'; - } - return errorMessage(err); -} - export function isTransientRemoteConnectError(message: string | null): boolean { if (!message) return false; return ( @@ -150,8 +114,10 @@ export function syncTimeoutMsForPeer( return opts.multipleTargets ? 8_000 : 15_000; } -export function localOpUploadKey(op: Operation): string { - return `${bytesToHex(op.meta.id.replica)}:${op.meta.id.counter}`; -} - export { isDiscoveryBootstrapUrl }; +export { + errorMessage, + formatSyncError, + inboundSyncPeerIdsToDrop, + isCapabilityRevokedError, +} from './syncErrorHelpers'; diff --git a/examples/playground/tests/playground.spec.ts b/examples/playground/tests/playground.spec.ts index 6c563fd2..9c367035 100644 --- a/examples/playground/tests/playground.spec.ts +++ b/examples/playground/tests/playground.spec.ts @@ -687,13 +687,12 @@ test("remote live sync all pushes new ops without a manual sync click", async ({ })(), ]); await Promise.race([ - server.waitForServedOps(doc, 1), + expect(treeRowByLabel(pageB, nodeLabel)).toBeVisible({ timeout: 60_000 }), (async () => { - await pageB.getByTestId("sync-error").waitFor({ state: "visible", timeout: 30_000 }); + await pageB.getByTestId("sync-error").waitFor({ state: "visible", timeout: 60_000 }); throw new Error(`sync error (B): ${await pageB.getByTestId("sync-error").textContent()}`); })(), ]); - await expect(treeRowByLabel(pageB, nodeLabel)).toBeVisible({ timeout: 60_000 }); await expect(pageA.getByTestId("sync-error")).toBeHidden(); await expect(pageB.getByTestId("sync-error")).toBeHidden(); diff --git a/examples/playground/tests/sync-errors.spec.ts b/examples/playground/tests/sync-errors.spec.ts new file mode 100644 index 00000000..f7064512 --- /dev/null +++ b/examples/playground/tests/sync-errors.spec.ts @@ -0,0 +1,42 @@ +import { expect, test } from "@playwright/test"; +import { InboundSyncAggregateError } from "@treecrdt/sync"; + +import { + formatSyncError, + inboundSyncPeerIdsToDrop, + isCapabilityRevokedError, +} from "../src/playground/syncErrorHelpers.js"; + +function inboundAggregate( + failures: readonly { peerId: string; error: unknown }[], +): InboundSyncAggregateError { + return new InboundSyncAggregateError( + failures.map((failure) => ({ ...failure, filter: { all: {} } })), + ); +} + +test("drops only non-revoked peers from an inbound aggregate", () => { + const error = inboundAggregate([ + { peerId: "peer-offline", error: new Error("transport closed") }, + { peerId: "peer-revoked", error: new Error("capability token revoked") }, + { peerId: "peer-offline", error: new Error("retry also failed") }, + ]); + + expect(inboundSyncPeerIdsToDrop(error)).toEqual(["peer-offline"]); + expect(isCapabilityRevokedError(error)).toBe(true); + expect(formatSyncError(error)).toBe( + "Access revoked for this capability. Import/update access, then sync again.", + ); +}); + +test("preserves the underlying single-target aggregate error", () => { + const error = inboundAggregate([{ peerId: "peer-a", error: new Error("transport closed") }]); + + expect(formatSyncError(error)).toBe("transport closed"); + expect(inboundSyncPeerIdsToDrop(error)).toEqual(["peer-a"]); +}); + +test("applies fallback peer handling only to non-revoked direct errors", () => { + expect(inboundSyncPeerIdsToDrop(new Error("offline"), ["peer-a"])).toEqual(["peer-a"]); + expect(inboundSyncPeerIdsToDrop(new Error("capability token revoked"), ["peer-a"])).toEqual([]); +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 947bd490..8b4cb10b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -41,6 +41,9 @@ importers: '@treecrdt/interface': specifier: workspace:* version: link:../../packages/treecrdt-ts + '@treecrdt/sync': + specifier: workspace:* + version: link:../../packages/treecrdt-sync '@treecrdt/sync-protocol': specifier: workspace:* version: link:../../packages/sync-protocol/protocol