mirror of
https://github.com/vrtmrz/obsidian-livesync.git
synced 2026-10-24 18:13:20 +00:00
Use stable P2P service views in the CLI
This commit is contained in:
1 parent
be03c25904
commit
4dcc783a71
8 files changed
+230
-118
No files matched your search
@@ -1,10 +1,9 @@
|
|||||||
import type { LiveSyncBaseCore } from "@/LiveSyncBaseCore";
|
import type { LiveSyncBaseCore } from "@/LiveSyncBaseCore";
|
||||||
import { P2P_DEFAULT_SETTINGS } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
import { P2P_DEFAULT_SETTINGS } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||||
import type { ServiceContext } from "@vrtmrz/livesync-commonlib/context";
|
import type { ServiceContext } from "@vrtmrz/livesync-commonlib/context";
|
||||||
import { LiveSyncTrysteroReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/LiveSyncTrysteroReplicator";
|
|
||||||
import { compatGlobal } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions";
|
import { compatGlobal } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions";
|
||||||
import { LiveSyncError } from "@vrtmrz/livesync-commonlib/compat/common/LSError";
|
import { LiveSyncError } from "@vrtmrz/livesync-commonlib/compat/common/LSError";
|
||||||
import { getPeerConnectionStats } from "@vrtmrz/livesync-commonlib/compat/rpc/transports/DiagRTCPeerConnections.utils";
|
import type { P2PPeerConnectionMetrics, P2PServiceViews } from "@vrtmrz/livesync-commonlib/p2p";
|
||||||
import { fsPromises } from "@vrtmrz/livesync-commonlib/node";
|
import { fsPromises } from "@vrtmrz/livesync-commonlib/node";
|
||||||
|
|
||||||
type CLIP2PPeer = {
|
type CLIP2PPeer = {
|
||||||
@@ -12,12 +11,7 @@ type CLIP2PPeer = {
|
|||||||
name: string;
|
name: string;
|
||||||
};
|
};
|
||||||
|
|
||||||
type CandidateSummary = {
|
type CLIP2PService = Pick<P2PServiceViews, "transportLifecycle" | "peerDirectory" | "targetedTransfer" | "diagnostics">;
|
||||||
id: string;
|
|
||||||
candidateType: string;
|
|
||||||
protocol: string;
|
|
||||||
relayProtocol: string;
|
|
||||||
};
|
|
||||||
|
|
||||||
function delay(ms: number): Promise<void> {
|
function delay(ms: number): Promise<void> {
|
||||||
return new Promise((resolve) => compatGlobal.setTimeout(resolve, ms));
|
return new Promise((resolve) => compatGlobal.setTimeout(resolve, ms));
|
||||||
@@ -43,35 +37,35 @@ function validateP2PSettings(core: LiveSyncBaseCore<ServiceContext, never>) {
|
|||||||
settings.P2P_IsHeadless = true;
|
settings.P2P_IsHeadless = true;
|
||||||
}
|
}
|
||||||
|
|
||||||
async function createReplicator(core: LiveSyncBaseCore<ServiceContext, never>): Promise<LiveSyncTrysteroReplicator> {
|
function requireP2PService(
|
||||||
|
core: LiveSyncBaseCore<ServiceContext, never>,
|
||||||
|
service: CLIP2PService | undefined
|
||||||
|
): CLIP2PService {
|
||||||
validateP2PSettings(core);
|
validateP2PSettings(core);
|
||||||
const replicator = await core.services.replicator.getNewReplicator();
|
if (!service) {
|
||||||
if (!replicator) {
|
throw new Error("P2P service is not available. Ensure the P2P feature was composed for this CLI process.");
|
||||||
throw new Error("Failed to create replicator instance. Ensure P2P is enabled in settings.");
|
|
||||||
}
|
}
|
||||||
if (!(replicator instanceof LiveSyncTrysteroReplicator)) {
|
return service;
|
||||||
throw new Error("Unexpected replicator type. Expected LiveSyncTrysteroReplicator.");
|
|
||||||
}
|
|
||||||
return replicator;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
function getSortedPeers(replicator: LiveSyncTrysteroReplicator): CLIP2PPeer[] {
|
function getSortedPeers(service: Pick<P2PServiceViews, "peerDirectory">): CLIP2PPeer[] {
|
||||||
return [...replicator.knownAdvertisements]
|
return [...service.peerDirectory.getPeers()]
|
||||||
.map((peer) => ({ peerId: peer.peerId, name: peer.name }))
|
.map((peer) => ({ peerId: peer.peerId, name: peer.name }))
|
||||||
.sort((a, b) => a.peerId.localeCompare(b.peerId));
|
.sort((a, b) => a.peerId.localeCompare(b.peerId));
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function collectPeers(
|
export async function collectPeers(
|
||||||
core: LiveSyncBaseCore<ServiceContext, never>,
|
core: LiveSyncBaseCore<ServiceContext, never>,
|
||||||
|
p2pService: CLIP2PService | undefined,
|
||||||
timeoutSec: number
|
timeoutSec: number
|
||||||
): Promise<CLIP2PPeer[]> {
|
): Promise<CLIP2PPeer[]> {
|
||||||
const replicator = await createReplicator(core);
|
const service = requireP2PService(core, p2pService);
|
||||||
await replicator.open();
|
await service.transportLifecycle.connect();
|
||||||
try {
|
try {
|
||||||
await delay(timeoutSec * 1000);
|
await delay(timeoutSec * 1000);
|
||||||
return getSortedPeers(replicator);
|
return getSortedPeers(service);
|
||||||
} finally {
|
} finally {
|
||||||
await replicator.close();
|
await service.transportLifecycle.disconnect();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -90,32 +84,8 @@ function resolvePeer(peers: CLIP2PPeer[], peerToken: string): CLIP2PPeer | undef
|
|||||||
return undefined;
|
return undefined;
|
||||||
}
|
}
|
||||||
|
|
||||||
function getReportValue<T extends string | number>(
|
|
||||||
report: Record<string, unknown> | undefined,
|
|
||||||
key: string
|
|
||||||
): T | "unknown" {
|
|
||||||
const value = report?.[key];
|
|
||||||
return typeof value === "string" || typeof value === "number" ? (value as T) : "unknown";
|
|
||||||
}
|
|
||||||
|
|
||||||
function summariseCandidate(reports: unknown[], candidateId: string): CandidateSummary | undefined {
|
|
||||||
if (candidateId === "unknown") {
|
|
||||||
return undefined;
|
|
||||||
}
|
|
||||||
const report = reports.map((r) => r as Record<string, unknown>).find((r) => r.id === candidateId);
|
|
||||||
if (!report) {
|
|
||||||
return undefined;
|
|
||||||
}
|
|
||||||
return {
|
|
||||||
id: candidateId,
|
|
||||||
candidateType: getReportValue<string>(report, "candidateType"),
|
|
||||||
protocol: getReportValue<string>(report, "protocol"),
|
|
||||||
relayProtocol: getReportValue<string>(report, "relayProtocol"),
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
async function writePeerConnectionStatsIfRequested(
|
async function writePeerConnectionStatsIfRequested(
|
||||||
replicator: LiveSyncTrysteroReplicator,
|
service: Pick<P2PServiceViews, "diagnostics">,
|
||||||
peer: CLIP2PPeer
|
peer: CLIP2PPeer
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
const outputPath = process.env.LIVESYNC_P2P_STATS_JSONL?.trim();
|
const outputPath = process.env.LIVESYNC_P2P_STATS_JSONL?.trim();
|
||||||
@@ -123,21 +93,30 @@ async function writePeerConnectionStatsIfRequested(
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
const peerConnection = replicator.rawHost?.room?.getPeers()[peer.peerId];
|
const stats = await service.diagnostics.getPeerConnectionMetrics(peer.peerId);
|
||||||
const stats = peerConnection ? await getPeerConnectionStats(`cli-p2p-${peer.peerId}`, peerConnection) : undefined;
|
const payload = createPeerConnectionStatsPayload(peer, stats, new Date().toISOString());
|
||||||
const localCandidate = summariseCandidate(stats?.reports ?? [], stats?.localCandidateId ?? "unknown");
|
await fsPromises.appendFile(outputPath, `${JSON.stringify(payload)}\n`, "utf8");
|
||||||
const remoteCandidate = summariseCandidate(stats?.reports ?? [], stats?.remoteCandidateId ?? "unknown");
|
}
|
||||||
|
|
||||||
|
/** Build the stable JSONL record consumed by the P2P benchmark harnesses. */
|
||||||
|
export function createPeerConnectionStatsPayload(
|
||||||
|
peer: CLIP2PPeer,
|
||||||
|
stats: P2PPeerConnectionMetrics | undefined,
|
||||||
|
generatedAt: string
|
||||||
|
) {
|
||||||
|
const localCandidate = stats?.localCandidate;
|
||||||
|
const remoteCandidate = stats?.remoteCandidate;
|
||||||
const selectedPath =
|
const selectedPath =
|
||||||
localCandidate && remoteCandidate
|
localCandidate && remoteCandidate
|
||||||
? `${localCandidate.candidateType}<->${remoteCandidate.candidateType}`
|
? `${localCandidate.candidateType}<->${remoteCandidate.candidateType}`
|
||||||
: "unknown";
|
: "unknown";
|
||||||
|
|
||||||
const payload = {
|
const payload = {
|
||||||
generatedAt: new Date().toISOString(),
|
generatedAt,
|
||||||
command: "p2p-sync",
|
command: "p2p-sync",
|
||||||
peerId: peer.peerId,
|
peerId: peer.peerId,
|
||||||
peerName: peer.name,
|
peerName: peer.name,
|
||||||
candidatePathCollected: !!stats?.selectedPair,
|
candidatePathCollected: stats?.selectedPairPresent ?? false,
|
||||||
selectedPath,
|
selectedPath,
|
||||||
selectedPair: stats
|
selectedPair: stats
|
||||||
? {
|
? {
|
||||||
@@ -155,23 +134,24 @@ async function writePeerConnectionStatsIfRequested(
|
|||||||
localCandidate,
|
localCandidate,
|
||||||
remoteCandidate,
|
remoteCandidate,
|
||||||
};
|
};
|
||||||
await fsPromises.appendFile(outputPath, `${JSON.stringify(payload)}\n`, "utf8");
|
return payload;
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function syncWithPeer(
|
export async function syncWithPeer(
|
||||||
core: LiveSyncBaseCore<ServiceContext, never>,
|
core: LiveSyncBaseCore<ServiceContext, never>,
|
||||||
|
p2pService: CLIP2PService | undefined,
|
||||||
peerToken: string,
|
peerToken: string,
|
||||||
timeoutSec: number
|
timeoutSec: number
|
||||||
): Promise<CLIP2PPeer> {
|
): Promise<CLIP2PPeer> {
|
||||||
const replicator = await createReplicator(core);
|
const service = requireP2PService(core, p2pService);
|
||||||
await replicator.open();
|
await service.transportLifecycle.connect();
|
||||||
try {
|
try {
|
||||||
const timeoutMs = timeoutSec * 1000;
|
const timeoutMs = timeoutSec * 1000;
|
||||||
const start = Date.now();
|
const start = Date.now();
|
||||||
let targetPeer: CLIP2PPeer | undefined;
|
let targetPeer: CLIP2PPeer | undefined;
|
||||||
|
|
||||||
while (Date.now() - start <= timeoutMs) {
|
while (Date.now() - start <= timeoutMs) {
|
||||||
const peers = getSortedPeers(replicator);
|
const peers = getSortedPeers(service);
|
||||||
targetPeer = resolvePeer(peers, peerToken);
|
targetPeer = resolvePeer(peers, peerToken);
|
||||||
if (targetPeer) {
|
if (targetPeer) {
|
||||||
break;
|
break;
|
||||||
@@ -183,11 +163,11 @@ export async function syncWithPeer(
|
|||||||
throw new Error(`Peer '${peerToken}' was not found within ${timeoutSec} seconds`);
|
throw new Error(`Peer '${peerToken}' was not found within ${timeoutSec} seconds`);
|
||||||
}
|
}
|
||||||
|
|
||||||
const pullResult = await replicator.replicateFrom(targetPeer.peerId, false);
|
const pullResult = await service.targetedTransfer.pullFromPeer(targetPeer.peerId, { showNotice: false });
|
||||||
if (pullResult && "error" in pullResult && pullResult.error) {
|
if (pullResult && "error" in pullResult && pullResult.error) {
|
||||||
throw pullResult.error instanceof Error ? pullResult.error : LiveSyncError.fromError(pullResult.error);
|
throw pullResult.error instanceof Error ? pullResult.error : LiveSyncError.fromError(pullResult.error);
|
||||||
}
|
}
|
||||||
const pushResult = await replicator.requestSynchroniseToPeer(targetPeer.peerId);
|
const pushResult = await service.targetedTransfer.requestPushToPeer(targetPeer.peerId);
|
||||||
if (!pushResult || pushResult.ok !== true) {
|
if (!pushResult || pushResult.ok !== true) {
|
||||||
const err: unknown = pushResult && "error" in pushResult ? pushResult.error : undefined;
|
const err: unknown = pushResult && "error" in pushResult ? pushResult.error : undefined;
|
||||||
throw err instanceof Error
|
throw err instanceof Error
|
||||||
@@ -195,15 +175,18 @@ export async function syncWithPeer(
|
|||||||
: LiveSyncError.fromError(err ?? "P2P sync failed while requesting remote sync");
|
: LiveSyncError.fromError(err ?? "P2P sync failed while requesting remote sync");
|
||||||
}
|
}
|
||||||
|
|
||||||
await writePeerConnectionStatsIfRequested(replicator, targetPeer);
|
await writePeerConnectionStatsIfRequested(service, targetPeer);
|
||||||
return targetPeer;
|
return targetPeer;
|
||||||
} finally {
|
} finally {
|
||||||
await replicator.close();
|
await service.transportLifecycle.disconnect();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function openP2PHost(core: LiveSyncBaseCore<ServiceContext, never>): Promise<LiveSyncTrysteroReplicator> {
|
export async function openP2PHost(
|
||||||
const replicator = await createReplicator(core);
|
core: LiveSyncBaseCore<ServiceContext, never>,
|
||||||
await replicator.open();
|
p2pService: CLIP2PService | undefined
|
||||||
return replicator;
|
): Promise<CLIP2PService> {
|
||||||
|
const service = requireP2PService(core, p2pService);
|
||||||
|
await service.transportLifecycle.connect();
|
||||||
|
return service;
|
||||||
}
|
}
|
||||||
@@ -1,5 +1,40 @@
|
|||||||
import { describe, expect, it } from "vitest";
|
import { describe, expect, it, vi } from "vitest";
|
||||||
import { parseTimeoutSeconds } from "./p2p";
|
import { collectPeers, createPeerConnectionStatsPayload, parseTimeoutSeconds, syncWithPeer } from "./p2p";
|
||||||
|
|
||||||
|
function createCore() {
|
||||||
|
const settings = { P2P_Enabled: true, P2P_AppID: "app-id", P2P_IsHeadless: false };
|
||||||
|
return {
|
||||||
|
services: {
|
||||||
|
setting: { currentSettings: () => settings },
|
||||||
|
replicator: { getNewReplicator: vi.fn(() => Promise.reject(new Error("must not be called"))) },
|
||||||
|
},
|
||||||
|
} as never;
|
||||||
|
}
|
||||||
|
|
||||||
|
function createP2PService() {
|
||||||
|
const connect = vi.fn(async () => undefined);
|
||||||
|
const disconnect = vi.fn(async () => undefined);
|
||||||
|
const pullFromPeer = vi.fn(async () => ({ ok: true }));
|
||||||
|
const requestPushToPeer = vi.fn(async () => ({ ok: true }));
|
||||||
|
return {
|
||||||
|
service: {
|
||||||
|
transportLifecycle: { isConnected: false, connect, disconnect },
|
||||||
|
peerDirectory: {
|
||||||
|
getPeers: () => [{ peerId: "peer-a", name: "Peer A", platform: "test" }],
|
||||||
|
},
|
||||||
|
targetedTransfer: {
|
||||||
|
pullFromPeer,
|
||||||
|
requestPushToPeer,
|
||||||
|
synchroniseWithPeer: vi.fn(),
|
||||||
|
},
|
||||||
|
diagnostics: { requestStatus: vi.fn(), getPeerConnectionMetrics: vi.fn() },
|
||||||
|
},
|
||||||
|
connect,
|
||||||
|
disconnect,
|
||||||
|
pullFromPeer,
|
||||||
|
requestPushToPeer,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
describe("p2p command helpers", () => {
|
describe("p2p command helpers", () => {
|
||||||
it("accepts non-negative timeout", () => {
|
it("accepts non-negative timeout", () => {
|
||||||
@@ -15,4 +50,88 @@ describe("p2p command helpers", () => {
|
|||||||
"p2p-sync requires a non-negative timeout in seconds"
|
"p2p-sync requires a non-negative timeout in seconds"
|
||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("collects peers through service views without acquiring a concrete replicator", async () => {
|
||||||
|
const { service, connect, disconnect } = createP2PService();
|
||||||
|
|
||||||
|
await expect(collectPeers(createCore(), service as never, 0)).resolves.toEqual([
|
||||||
|
{ peerId: "peer-a", name: "Peer A" },
|
||||||
|
]);
|
||||||
|
expect(connect).toHaveBeenCalledOnce();
|
||||||
|
expect(disconnect).toHaveBeenCalledOnce();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("synchronises through the targeted-transfer view", async () => {
|
||||||
|
const { service, pullFromPeer, requestPushToPeer } = createP2PService();
|
||||||
|
|
||||||
|
await expect(syncWithPeer(createCore(), service as never, "peer-a", 0)).resolves.toEqual({
|
||||||
|
peerId: "peer-a",
|
||||||
|
name: "Peer A",
|
||||||
|
});
|
||||||
|
expect(pullFromPeer).toHaveBeenCalledWith("peer-a", { showNotice: false });
|
||||||
|
expect(requestPushToPeer).toHaveBeenCalledWith("peer-a");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("preserves the benchmark diagnostics JSONL contract", () => {
|
||||||
|
expect(
|
||||||
|
createPeerConnectionStatsPayload(
|
||||||
|
{ peerId: "peer-a", name: "Peer A" },
|
||||||
|
{
|
||||||
|
selectedPairPresent: true,
|
||||||
|
selectedPairId: "pair-1",
|
||||||
|
state: "succeeded",
|
||||||
|
currentRoundTripTime: 0.01,
|
||||||
|
totalRoundTripTime: 0.1,
|
||||||
|
requestsSent: 3,
|
||||||
|
responsesReceived: 3,
|
||||||
|
packetsDiscardedOnSend: 0,
|
||||||
|
bytesSent: 100,
|
||||||
|
bytesReceived: 200,
|
||||||
|
localCandidate: {
|
||||||
|
id: "local-1",
|
||||||
|
candidateType: "host",
|
||||||
|
protocol: "udp",
|
||||||
|
relayProtocol: "unknown",
|
||||||
|
},
|
||||||
|
remoteCandidate: {
|
||||||
|
id: "remote-1",
|
||||||
|
candidateType: "relay",
|
||||||
|
protocol: "udp",
|
||||||
|
relayProtocol: "udp",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
"2026-08-27T00:00:00.000Z"
|
||||||
|
)
|
||||||
|
).toEqual({
|
||||||
|
generatedAt: "2026-08-27T00:00:00.000Z",
|
||||||
|
command: "p2p-sync",
|
||||||
|
peerId: "peer-a",
|
||||||
|
peerName: "Peer A",
|
||||||
|
candidatePathCollected: true,
|
||||||
|
selectedPath: "host<->relay",
|
||||||
|
selectedPair: {
|
||||||
|
id: "pair-1",
|
||||||
|
state: "succeeded",
|
||||||
|
currentRoundTripTime: 0.01,
|
||||||
|
totalRoundTripTime: 0.1,
|
||||||
|
requestsSent: 3,
|
||||||
|
responsesReceived: 3,
|
||||||
|
packetsDiscardedOnSend: 0,
|
||||||
|
bytesSent: 100,
|
||||||
|
bytesReceived: 200,
|
||||||
|
},
|
||||||
|
localCandidate: {
|
||||||
|
id: "local-1",
|
||||||
|
candidateType: "host",
|
||||||
|
protocol: "udp",
|
||||||
|
relayProtocol: "unknown",
|
||||||
|
},
|
||||||
|
remoteCandidate: {
|
||||||
|
id: "remote-1",
|
||||||
|
candidateType: "relay",
|
||||||
|
protocol: "udp",
|
||||||
|
relayProtocol: "udp",
|
||||||
|
},
|
||||||
|
});
|
||||||
|
});
|
||||||
});
|
});
|
||||||
@@ -252,7 +252,7 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext
|
|||||||
}
|
}
|
||||||
const timeoutSec = parseTimeoutSeconds(options.commandArgs[0], "p2p-peers");
|
const timeoutSec = parseTimeoutSeconds(options.commandArgs[0], "p2p-peers");
|
||||||
writeStderrLine(standardIo, `[Command] p2p-peers timeout=${timeoutSec}s`);
|
writeStderrLine(standardIo, `[Command] p2p-peers timeout=${timeoutSec}s`);
|
||||||
const peers = await collectPeers(core, timeoutSec);
|
const peers = await collectPeers(core, context.p2pReplicator, timeoutSec);
|
||||||
if (peers.length > 0) {
|
if (peers.length > 0) {
|
||||||
standardIo.writeStdout(peers.map((peer) => `[peer]\t${peer.peerId}\t${peer.name}`).join("\n") + "\n");
|
standardIo.writeStdout(peers.map((peer) => `[peer]\t${peer.peerId}\t${peer.name}`).join("\n") + "\n");
|
||||||
}
|
}
|
||||||
@@ -269,14 +269,14 @@ export async function runCommand(options: CLIOptions, context: CLICommandContext
|
|||||||
}
|
}
|
||||||
const timeoutSec = parseTimeoutSeconds(options.commandArgs[1], "p2p-sync");
|
const timeoutSec = parseTimeoutSeconds(options.commandArgs[1], "p2p-sync");
|
||||||
writeStderrLine(standardIo, `[Command] p2p-sync peer=${peerToken} timeout=${timeoutSec}s`);
|
writeStderrLine(standardIo, `[Command] p2p-sync peer=${peerToken} timeout=${timeoutSec}s`);
|
||||||
const peer = await syncWithPeer(core, peerToken, timeoutSec);
|
const peer = await syncWithPeer(core, context.p2pReplicator, peerToken, timeoutSec);
|
||||||
writeStderrLine(standardIo, `[Done] P2P sync completed with ${peer.name} (${peer.peerId})`);
|
writeStderrLine(standardIo, `[Done] P2P sync completed with ${peer.name} (${peer.peerId})`);
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (options.command === "p2p-host") {
|
if (options.command === "p2p-host") {
|
||||||
writeStderrLine(standardIo, "[Command] p2p-host");
|
writeStderrLine(standardIo, "[Command] p2p-host");
|
||||||
await openP2PHost(core);
|
await openP2PHost(core, context.p2pReplicator);
|
||||||
writeStderrLine(standardIo, "[Ready] P2P host is running. Press Ctrl+C to stop.");
|
writeStderrLine(standardIo, "[Ready] P2P host is running. Press Ctrl+C to stop.");
|
||||||
await new Promise(() => {});
|
await new Promise(() => {});
|
||||||
return true;
|
return true;
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
import { LiveSyncBaseCore } from "@/LiveSyncBaseCore";
|
import { LiveSyncBaseCore } from "@/LiveSyncBaseCore";
|
||||||
import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
import type { ObsidianLiveSyncSettings } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||||
import type { NodeServiceContext } from "@/apps/cli/services/NodeServiceContext";
|
import type { NodeServiceContext } from "@/apps/cli/services/NodeServiceContext";
|
||||||
import type { UseP2PReplicatorResult } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/UseP2PReplicatorResult";
|
import type { UseP2PReplicatorResult } from "@vrtmrz/livesync-commonlib/p2p";
|
||||||
|
|
||||||
export type CLICommand =
|
export type CLICommand =
|
||||||
| "daemon"
|
| "daemon"
|
||||||
|
|||||||
@@ -23,8 +23,7 @@ import type { CLICommand, CLICommandContext, CLIOptions } from "./commands/types
|
|||||||
import { getPathFromUXFileInfo } from "@vrtmrz/livesync-commonlib/compat/common/typeUtils";
|
import { getPathFromUXFileInfo } from "@vrtmrz/livesync-commonlib/compat/common/typeUtils";
|
||||||
import { stripAllPrefixes } from "@vrtmrz/livesync-commonlib/compat/string_and_binary/path";
|
import { stripAllPrefixes } from "@vrtmrz/livesync-commonlib/compat/string_and_binary/path";
|
||||||
import { IgnoreRules } from "./serviceModules/IgnoreRules";
|
import { IgnoreRules } from "./serviceModules/IgnoreRules";
|
||||||
import { useP2PReplicatorFeature } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/useP2PReplicatorFeature";
|
import { useP2PReplicatorFeature, type UseP2PReplicatorResult } from "@vrtmrz/livesync-commonlib/p2p";
|
||||||
import type { UseP2PReplicatorResult } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/UseP2PReplicatorResult";
|
|
||||||
import { createNodeStandardIo, fsPromises as fs, path } from "@vrtmrz/livesync-commonlib/node";
|
import { createNodeStandardIo, fsPromises as fs, path } from "@vrtmrz/livesync-commonlib/node";
|
||||||
import type { StandardIo } from "@vrtmrz/livesync-commonlib/context";
|
import type { StandardIo } from "@vrtmrz/livesync-commonlib/context";
|
||||||
import { writeStderrLine, writeStdoutLine } from "./cliOutput";
|
import { writeStderrLine, writeStdoutLine } from "./cliOutput";
|
||||||
@@ -290,7 +289,10 @@ export async function main(
|
|||||||
) {
|
) {
|
||||||
const options = parseArgs(standardIo);
|
const options = parseArgs(standardIo);
|
||||||
if (options.interval && options.command !== "daemon") {
|
if (options.interval && options.command !== "daemon") {
|
||||||
writeStderrLine(standardIo, `Warning: --interval is only used in daemon mode, ignored for '${options.command}'`);
|
writeStderrLine(
|
||||||
|
standardIo,
|
||||||
|
`Warning: --interval is only used in daemon mode, ignored for '${options.command}'`
|
||||||
|
);
|
||||||
}
|
}
|
||||||
const avoidStdoutNoise =
|
const avoidStdoutNoise =
|
||||||
options.command === "cat" ||
|
options.command === "cat" ||
|
||||||
@@ -420,7 +422,10 @@ export async function main(
|
|||||||
// In daemon mode the default handler must run so changes are applied to the filesystem.
|
// In daemon mode the default handler must run so changes are applied to the filesystem.
|
||||||
if (options.command !== "daemon") {
|
if (options.command !== "daemon") {
|
||||||
serviceHubInstance.replication.processSynchroniseResult.addHandler(async () => {
|
serviceHubInstance.replication.processSynchroniseResult.addHandler(async () => {
|
||||||
writeStderrLine(standardIo, `[Info] Replication result received, but not processed automatically in CLI mode.`);
|
writeStderrLine(
|
||||||
|
standardIo,
|
||||||
|
`[Info] Replication result received, but not processed automatically in CLI mode.`
|
||||||
|
);
|
||||||
return await Promise.resolve(true);
|
return await Promise.resolve(true);
|
||||||
}, -100);
|
}, -100);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
import type { FilePathWithPrefix } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
import type { FilePathWithPrefix } from "@vrtmrz/livesync-commonlib/compat/common/types";
|
||||||
import { LiveSyncTrysteroReplicator } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/LiveSyncTrysteroReplicator";
|
|
||||||
import { compatGlobal } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions";
|
import { compatGlobal } from "@vrtmrz/livesync-commonlib/compat/common/coreEnvFunctions";
|
||||||
|
import type { P2PServiceViews } from "@vrtmrz/livesync-commonlib/p2p";
|
||||||
import type { CLICommandContext } from "@/apps/cli/commands/types";
|
import type { CLICommandContext } from "@/apps/cli/commands/types";
|
||||||
import { openP2PHost } from "@/apps/cli/commands/p2p";
|
import { openP2PHost } from "@/apps/cli/commands/p2p";
|
||||||
|
|
||||||
@@ -15,32 +15,35 @@ function describeError(value: unknown): string {
|
|||||||
return value instanceof Error ? (value.stack ?? value.message) : String(value);
|
return value instanceof Error ? (value.stack ?? value.message) : String(value);
|
||||||
}
|
}
|
||||||
|
|
||||||
async function waitForServing(replicator: LiveSyncTrysteroReplicator, timeoutMs: number): Promise<void> {
|
type ProbeP2PService = Pick<P2PServiceViews, "transportLifecycle" | "peerDirectory" | "targetedTransfer">;
|
||||||
|
|
||||||
|
async function waitForServing(service: ProbeP2PService, timeoutMs: number): Promise<void> {
|
||||||
const started = Date.now();
|
const started = Date.now();
|
||||||
while (Date.now() - started <= timeoutMs) {
|
while (Date.now() - started <= timeoutMs) {
|
||||||
if (replicator.server?.isServing) return;
|
if (service.transportLifecycle.isConnected) return;
|
||||||
await delay(200);
|
await delay(200);
|
||||||
}
|
}
|
||||||
throw new Error("The replacement P2P replicator did not start serving within the timeout");
|
throw new Error("The stable P2P service did not start serving within the timeout");
|
||||||
}
|
}
|
||||||
|
|
||||||
async function waitForPeer(
|
async function waitForPeer(
|
||||||
replicator: LiveSyncTrysteroReplicator,
|
service: ProbeP2PService,
|
||||||
targetPeer: string,
|
targetPeer: string,
|
||||||
timeoutMs: number
|
timeoutMs: number
|
||||||
): Promise<{ peerId: string; name: string }> {
|
): Promise<{ peerId: string; name: string }> {
|
||||||
const started = Date.now();
|
const started = Date.now();
|
||||||
while (Date.now() - started <= timeoutMs) {
|
while (Date.now() - started <= timeoutMs) {
|
||||||
const peer = replicator.knownAdvertisements.find(
|
const peer = service.peerDirectory
|
||||||
(candidate) => candidate.name === targetPeer || candidate.peerId === targetPeer
|
.getPeers()
|
||||||
);
|
.find((candidate) => candidate.name === targetPeer || candidate.peerId === targetPeer);
|
||||||
if (peer) return peer;
|
if (peer) return peer;
|
||||||
await delay(200);
|
await delay(200);
|
||||||
}
|
}
|
||||||
const knownPeers = replicator.knownAdvertisements.map((peer) => `${peer.name} (${peer.peerId})`).join(", ");
|
const knownPeers = service.peerDirectory
|
||||||
throw new Error(
|
.getPeers()
|
||||||
`Peer '${targetPeer}' was not discovered within the timeout. Known peers: ${knownPeers || "none"}`
|
.map((peer) => `${peer.name} (${peer.peerId})`)
|
||||||
);
|
.join(", ");
|
||||||
|
throw new Error(`Peer '${targetPeer}' was not discovered within the timeout. Known peers: ${knownPeers || "none"}`);
|
||||||
}
|
}
|
||||||
|
|
||||||
function assertPullSucceeded(result: unknown): void {
|
function assertPullSucceeded(result: unknown): void {
|
||||||
@@ -50,15 +53,17 @@ function assertPullSucceeded(result: unknown): void {
|
|||||||
}
|
}
|
||||||
|
|
||||||
async function communicateWithPeer(
|
async function communicateWithPeer(
|
||||||
replicator: LiveSyncTrysteroReplicator,
|
service: ProbeP2PService,
|
||||||
targetPeer: string,
|
targetPeer: string,
|
||||||
timeoutMs: number
|
timeoutMs: number
|
||||||
): Promise<{ peerId: string; name: string }> {
|
): Promise<{ peerId: string; name: string }> {
|
||||||
await replicator.open();
|
if (!service.transportLifecycle.isConnected) {
|
||||||
await waitForServing(replicator, timeoutMs);
|
await service.transportLifecycle.connect();
|
||||||
const peer = await waitForPeer(replicator, targetPeer, timeoutMs);
|
}
|
||||||
assertPullSucceeded(await replicator.replicateFrom(peer.peerId, false));
|
await waitForServing(service, timeoutMs);
|
||||||
const pushResult = await replicator.requestSynchroniseToPeer(peer.peerId);
|
const peer = await waitForPeer(service, targetPeer, timeoutMs);
|
||||||
|
assertPullSucceeded(await service.targetedTransfer.pullFromPeer(peer.peerId, { showNotice: false }));
|
||||||
|
const pushResult = await service.targetedTransfer.requestPushToPeer(peer.peerId);
|
||||||
if (!pushResult || pushResult.ok !== true) {
|
if (!pushResult || pushResult.ok !== true) {
|
||||||
throw new Error(`P2P push failed: ${describeError(pushResult?.error)}`);
|
throw new Error(`P2P push failed: ${describeError(pushResult?.error)}`);
|
||||||
}
|
}
|
||||||
@@ -78,35 +83,39 @@ export async function runP2PReplicatorReplacementProbe(
|
|||||||
throw new Error("The CLI did not expose its P2P service-feature result to the integration probe");
|
throw new Error("The CLI did not expose its P2P service-feature result to the integration probe");
|
||||||
}
|
}
|
||||||
|
|
||||||
const firstReplicator = await openP2PHost(core);
|
const initialActiveReplicator = core.services.replicator.getActiveReplicator();
|
||||||
if (p2pReplicator.replicator !== firstReplicator) {
|
if (!initialActiveReplicator) {
|
||||||
throw new Error("The P2P service feature did not expose the newly created replicator");
|
throw new Error("The CLI did not activate the initial P2P Replicator adapter");
|
||||||
}
|
}
|
||||||
|
const compatibilityFacade = p2pReplicator.replicator;
|
||||||
|
const p2pService = await openP2PHost(core, p2pReplicator);
|
||||||
|
|
||||||
const firstPeer = await communicateWithPeer(firstReplicator, targetPeer, timeoutMs);
|
const firstPeer = await communicateWithPeer(p2pService, targetPeer, timeoutMs);
|
||||||
const initialised = await core.services.databaseEvents.initialiseDatabase(false, true, false);
|
const initialised = await core.services.databaseEvents.initialiseDatabase(false, true, false);
|
||||||
if (!initialised) {
|
if (!initialised) {
|
||||||
throw new Error("Database reinitialisation failed during the P2P replacement probe");
|
throw new Error("Database reinitialisation failed during the P2P replacement probe");
|
||||||
}
|
}
|
||||||
|
|
||||||
const replacementReplicator = p2pReplicator.replicator;
|
const replacementActiveReplicator = core.services.replicator.getActiveReplicator();
|
||||||
if (core.services.replicator.getActiveReplicator() !== replacementReplicator) {
|
if (!replacementActiveReplicator) {
|
||||||
throw new Error("ReplicatorService did not activate the P2P service feature's replacement replicator");
|
throw new Error("ReplicatorService did not activate a replacement P2P Replicator adapter");
|
||||||
}
|
}
|
||||||
if (replacementReplicator === firstReplicator) {
|
if (replacementActiveReplicator === initialActiveReplicator) {
|
||||||
throw new Error("Database reinitialisation retained the previous P2P replicator instance");
|
throw new Error("Database reinitialisation retained the previous active P2P Replicator adapter");
|
||||||
}
|
}
|
||||||
if (firstReplicator.server !== undefined) {
|
if (p2pReplicator.replicator !== compatibilityFacade) {
|
||||||
throw new Error("The previous P2P replicator remained open after replacement");
|
throw new Error("Database reinitialisation replaced the stable P2P service compatibility facade");
|
||||||
|
}
|
||||||
|
if (p2pService.transportLifecycle.isConnected) {
|
||||||
|
throw new Error("Database reinitialisation left the database-bound P2P room open");
|
||||||
}
|
}
|
||||||
|
|
||||||
const settings = core.services.setting.currentSettings();
|
const settings = core.services.setting.currentSettings();
|
||||||
settings.P2P_AutoStart = true;
|
settings.P2P_AutoStart = true;
|
||||||
await core.services.control.applySettings();
|
await core.services.control.applySettings();
|
||||||
const resumedReplicator = p2pReplicator.replicator;
|
await waitForServing(p2pService, timeoutMs);
|
||||||
await waitForServing(resumedReplicator, timeoutMs);
|
if (p2pReplicator.replicator !== compatibilityFacade) {
|
||||||
if (firstReplicator.server !== undefined) {
|
throw new Error("A setting event replaced the stable P2P service compatibility facade");
|
||||||
throw new Error("A setting event reopened the previous P2P replicator");
|
|
||||||
}
|
}
|
||||||
|
|
||||||
const encoded = new TextEncoder().encode(noteContent);
|
const encoded = new TextEncoder().encode(noteContent);
|
||||||
@@ -118,7 +127,7 @@ export async function runP2PReplicatorReplacementProbe(
|
|||||||
});
|
});
|
||||||
await core.serviceModules.fileHandler.storeFileToDB(notePath as FilePathWithPrefix, true);
|
await core.serviceModules.fileHandler.storeFileToDB(notePath as FilePathWithPrefix, true);
|
||||||
|
|
||||||
const replacementPeer = await communicateWithPeer(resumedReplicator, targetPeer, timeoutMs);
|
const replacementPeer = await communicateWithPeer(p2pService, targetPeer, timeoutMs);
|
||||||
if (replacementPeer.name !== firstPeer.name) {
|
if (replacementPeer.name !== firstPeer.name) {
|
||||||
throw new Error(
|
throw new Error(
|
||||||
`The replacement replicator reached '${replacementPeer.name}' instead of the original peer '${firstPeer.name}'`
|
`The replacement replicator reached '${replacementPeer.name}' instead of the original peer '${firstPeer.name}'`
|
||||||
@@ -126,7 +135,7 @@ export async function runP2PReplicatorReplacementProbe(
|
|||||||
}
|
}
|
||||||
|
|
||||||
core.services.context.standardIo.writeStdout(
|
core.services.context.standardIo.writeStdout(
|
||||||
`[Probe] P2P replicator replaced, old transport stayed closed, and ${notePath} was sent through the replacement.\n`
|
`[Probe] The active P2P adapter was replaced, the stable service reopened, and ${notePath} was sent through it.\n`
|
||||||
);
|
);
|
||||||
return true;
|
return true;
|
||||||
}
|
}
|
||||||
@@ -39,7 +39,7 @@ async function runReplacementProbe(
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
Deno.test("p2p lifecycle: replacement keeps real CLI communication on the current replicator", async () => {
|
Deno.test("p2p lifecycle: active-adapter replacement keeps real CLI communication on the stable service", async () => {
|
||||||
const relay = Deno.env.get("RELAY") ?? "ws://localhost:4000/";
|
const relay = Deno.env.get("RELAY") ?? "ws://localhost:4000/";
|
||||||
const peersTimeout = Number(Deno.env.get("PEERS_TIMEOUT") ?? "20");
|
const peersTimeout = Number(Deno.env.get("PEERS_TIMEOUT") ?? "20");
|
||||||
const syncTimeout = Number(Deno.env.get("SYNC_TIMEOUT") ?? "60");
|
const syncTimeout = Number(Deno.env.get("SYNC_TIMEOUT") ?? "60");
|
||||||
@@ -82,11 +82,8 @@ Deno.test("p2p lifecycle: replacement keeps real CLI communication on the curren
|
|||||||
try {
|
try {
|
||||||
await host.waitUntilContains("P2P host is running", 20000);
|
await host.waitUntilContains("P2P host is running", 20000);
|
||||||
const probe = await runReplacementProbe(probeVault, probeSettings, hostPeerName, probeTimeoutMs);
|
const probe = await runReplacementProbe(probeVault, probeSettings, hostPeerName, probeTimeoutMs);
|
||||||
assert(
|
assert(probe.code === 0, `P2P replacement probe failed\nstdout: ${probe.stdout}\nstderr: ${probe.stderr}`);
|
||||||
probe.code === 0,
|
assertStringIncludes(probe.stdout, "[Probe] The active P2P adapter was replaced");
|
||||||
`P2P replacement probe failed\nstdout: ${probe.stdout}\nstderr: ${probe.stderr}`
|
|
||||||
);
|
|
||||||
assertStringIncludes(probe.stdout, "[Probe] P2P replicator replaced");
|
|
||||||
|
|
||||||
const syncResult = await runCli(
|
const syncResult = await runCli(
|
||||||
verifierVault,
|
verifierVault,
|
||||||
|
|||||||
+1
-2
@@ -39,8 +39,7 @@ import { useSetupProtocolFeature } from "./serviceFeatures/setupObsidian/setupPr
|
|||||||
import { useSetupQRCodeFeature } from "@/serviceFeatures/setupObsidian/qrCode";
|
import { useSetupQRCodeFeature } from "@/serviceFeatures/setupObsidian/qrCode";
|
||||||
import { useSetupURIFeature } from "@/serviceFeatures/setupObsidian/setupUri";
|
import { useSetupURIFeature } from "@/serviceFeatures/setupObsidian/setupUri";
|
||||||
import { useSetupManagerHandlersFeature } from "./serviceFeatures/setupObsidian/setupManagerHandlers.ts";
|
import { useSetupManagerHandlersFeature } from "./serviceFeatures/setupObsidian/setupManagerHandlers.ts";
|
||||||
import { useP2PReplicatorFeature } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/useP2PReplicatorFeature";
|
import { useP2PReplicatorCommands, useP2PReplicatorFeature } from "@vrtmrz/livesync-commonlib/p2p";
|
||||||
import { useP2PReplicatorCommands } from "@vrtmrz/livesync-commonlib/compat/replication/trystero/useP2PReplicatorCommands";
|
|
||||||
import { useP2PReplicatorUI } from "./serviceFeatures/useP2PReplicatorUI.ts";
|
import { useP2PReplicatorUI } from "./serviceFeatures/useP2PReplicatorUI.ts";
|
||||||
import { useReviewHarness } from "./serviceFeatures/useReviewHarness.ts";
|
import { useReviewHarness } from "./serviceFeatures/useReviewHarness.ts";
|
||||||
import { createOpenReplicationUI, createOpenRebuildUI } from "./features/P2PSync/P2PReplicator/P2PReplicationUI.ts";
|
import { createOpenReplicationUI, createOpenRebuildUI } from "./features/P2PSync/P2PReplicator/P2PReplicationUI.ts";
|
||||||
|
|||||||
Reference in new issue
Block a user