Skip to content

refactor(playground): adopt the sync controllers #205

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
marcus-pousette wants to merge 1 commit into
base: stack/inbound-sync-controller
Choose a base branch
from
Open

refactor(playground): adopt the sync controllers #205

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion examples/playground/package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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": {
Expand All @@ -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:*",
Expand Down
8 changes: 4 additions & 4 deletions examples/playground/src/App.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -468,7 +468,7 @@ export default function App() {
liveAllEnabled,
setLiveAllEnabled,
toggleLiveChildren,
queueLocalOpsForSync,
queueOps,
handleSync,
handleScopedSync,
postBroadcastMessage,
Expand Down Expand Up @@ -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(
Expand Down
228 changes: 64 additions & 164 deletions examples/playground/src/playground/hooks/usePlaygroundLiveSubscriptions.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand All @@ -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<SyncPeer<Operation> | null>;
syncConnRef: MutableRefObject<Map<string, PlaygroundSyncConnection>>;
setSyncError: Dispatch<SetStateAction<string | null>>;
authCanSyncAll: boolean;
}) {
const { syncPeerRef, syncConnRef, setSyncError, authCanSyncAll } = opts;
const { syncPeerRef, setSyncError, authCanSyncAll } = opts;
const [liveBusy, setLiveBusy] = useState(false);
const [liveChildrenParents, setLiveChildrenParents] = useState<Set<string>>(() => new Set());
const [liveAllEnabled, setLiveAllEnabled] = useState(false);
const liveChildrenParentsRef = useRef<Set<string>>(new Set());
const liveChildSubsRef = useRef<Map<string, Map<string, SyncSubscription>>>(new Map());
const liveAllEnabledRef = useRef(false);
const liveAllSubsRef = useRef<Map<string, SyncSubscription>>(new Map());
const liveAllStartingRef = useRef<Set<string>>(new Set());
const liveChildrenStartingRef = useRef<Set<string>>(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<InboundSync<Operation> | null>(null);
const inboundSyncPeerRef = useRef<SyncPeer<Operation> | 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<Operation>({
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<SyncMessage<Operation>>);
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<string, SyncSubscription>();
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) => {
Expand All @@ -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]);

Expand All @@ -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,
};
}
Loading
Loading