sync-runner.ts

1// Production sync runner — composes the live vault adapter, WebSocket2// transport, and coordinator into the function the dispatcher injects via3// `setSyncRunner`. The runner is responsible for opening one Noise KK4// session against a single paired device, driving the sync protocol, and5// surfacing a `SyncRunResult` (or throwing on failure). The dispatcher in6// sync-trigger.ts iterates over hosts and aggregates results.7//8// Connection strategy:9//   1. Try `getLastKnownIp(deviceLongTermPubkey)` if cached — fast path.10//   2. Fall back to `probeSubnetForPeer` over the local /24. The subnet11//      probe is itself rate-limited (60s per-pubkey cooldown) inside the12//      transport, so we don't add another layer here.13//   3. On a successful session, persist the IP we connected on so the next14//      sync round skips the probe.15//16// We do NOT verify vault_id here — the coordinator's SYNC_HELLO exchange17// does that (returns `status: "vault_mismatch"` if the peer holds a18// different vault). The runner just translates the coordinator's outcome19// into the throw-or-return contract the dispatcher expects.2021import { runSyncInitiator, type SyncVaultAdapter } from "./sync-coordinator";22import { createSyncVaultAdapter } from "./sync-vault-adapter";23import {24  getLastKnownIp,25  setLastKnownIp,26} from "./paired-devices";27import {28  openRelaySyncSession,29  openSyncSession,30  probeSubnetForPeer,31  type ProbedSyncEndpoint,32  type SyncSession,33  type SyncEndpoint,34  type SyncPeerIdentity,35} from "../shared/sync-transport";36import { hexToPubkey } from "../shared/sync-identity";37import { getLocalSubnet } from "../shared/local-network";38import { buildLocalIdentity, deriveLocalSyncKeypair } from "./device-identity";39import { PAIRING_PORT } from "../shared/pairing-dial";40import { constantTimeEqualHex } from "../shared/constant-time";41import { getRootKeyHex, isUnlocked } from "./vault";42import type { PairedDevice } from "../shared/sync-types";43import type { SyncRunResult } from "./sync-trigger";44import {45  dialingRendezvousIds,46  rendezvousSecretFromHex,47} from "../shared/rendezvous";48import {49  getRelaySettings,50  type RelaySettings,51} from "./relay-settings";5253/** Same fixed port pairing dials — SyncReconnectListener.SYNC_PORT on Android. */54const DEFAULT_PORT = PAIRING_PORT;5556/**57 * Test/injection seam. Production wires in the real `openSyncSession`;58 * unit tests inject a fake to avoid real WebSocket + Noise KK.59 */60export interface SyncRunnerDeps {61  openSession: (62    endpoint: SyncEndpoint,63    ourIdentity: SyncPeerIdentity,64    expectedRemoteHex: string,65    rendezvousId: string,66  ) => Promise<SyncSession>;67  probeSubnet: (68    subnet: string,69    port: number,70    ourIdentity: SyncPeerIdentity,71    expectedRemoteHex: string,72    rendezvousIds: readonly string[],73  ) => Promise<ProbedSyncEndpoint | null>;74  openRelaySession?: (75    relayUrl: string,76    ourIdentity: SyncPeerIdentity,77    expectedRemoteHex: string,78    rendezvousId: string,79  ) => Promise<SyncSession>;80  getRelaySettings?: () => Promise<RelaySettings>;81  getSubnet: () => Promise<string | null>;82  createAdapter: (context?: { deviceName?: string }) => SyncVaultAdapter;83}8485const productionDeps: SyncRunnerDeps = {86  openSession: (endpoint, identity, expected, rendezvousId) =>87    openSyncSession(endpoint, identity, expected, rendezvousId),88  probeSubnet: (subnet, port, identity, expected, rendezvousIds) =>89    probeSubnetForPeer(subnet, port, identity, expected, rendezvousIds),90  openRelaySession: (relayUrl, identity, expected, rendezvousId) =>91    openRelaySyncSession(relayUrl, identity, expected, rendezvousId),92  getRelaySettings,93  getSubnet: getLocalSubnet,94  createAdapter: (context) => createSyncVaultAdapter(context),95};9697export async function productionSyncRunner(device: PairedDevice): Promise<SyncRunResult> {98  return runWithDeps(device, productionDeps);99}100101/**102 * Inner runner with explicit deps — exported so tests can inject fakes.103 */104export async function runWithDeps(105  device: PairedDevice,106  deps: SyncRunnerDeps,107): Promise<SyncRunResult> {108  if (!isUnlocked()) return { ok: false, error: "VAULT_LOCKED" };109  if (!device.longTermPublicKeyHex) {110    throw new Error("paired_device_missing_pubkey");111  }112  if (!device.rendezvousSecretHex) {113    throw new Error("paired_device_missing_rendezvous_secret");114  }115  const expectedRemoteHex = device.longTermPublicKeyHex;116  const reconnectIds = dialingRendezvousIds(117    rendezvousSecretFromHex(device.rendezvousSecretHex),118    Date.now() / 1000,119  );120  const port = device.wsPort ?? DEFAULT_PORT;121122  // Sync identity derives from the stable root key, not the password-derived123  // key. Reconnect uses both halves directly in standard Noise KK.124  const masterKeyHex = getRootKeyHex();125  if (!masterKeyHex) return { ok: false, error: "VAULT_LOCKED" };126  const masterKey = hexToBytesLocal(masterKeyHex);127  const publicIdentity = await buildLocalIdentity(masterKey);128  const localKeys = deriveLocalSyncKeypair(masterKey, publicIdentity.deviceId);129  if (!constantTimeEqualHex(130    publicIdentity.longTermPublicKeyHex,131    bytesToHex(localKeys.publicKey),132  )) {133    masterKey.fill(0);134    localKeys.secretKey.fill(0);135    throw new Error("derived reconnect identity mismatch");136  }137  const ourIdentity: SyncPeerIdentity = {138    ...publicIdentity,139    longTermPrivateKey: localKeys.secretKey,140  };141142  // Sanity — the remote pubkey must be 32-byte hex so noise can compare it.143  void hexToPubkey(expectedRemoteHex);144145  // HV-1b: pass the peer being synced so remote-mutation alerts can name it.146  const adapter = deps.createAdapter({ deviceName: device.deviceName });147  let session: SyncSession;148  try {149    try {150      session = await connectLan(151        expectedRemoteHex,152        port,153        ourIdentity,154        reconnectIds,155        deps,156      );157    } catch (lanError) {158      const relaySettings = await deps.getRelaySettings?.();159      if (160        !relaySettings ||161        relaySettings.lanOnly ||162        !deps.openRelaySession163      ) {164        throw lanError;165      }166      session = await connectRelay(167        device.relayUrl ?? relaySettings.relayUrl,168        expectedRemoteHex,169        ourIdentity,170        reconnectIds,171        deps.openRelaySession,172      );173    }174  } finally {175    masterKey.fill(0);176    localKeys.secretKey.fill(0);177  }178179  try {180    const outcome = await runSyncInitiator(session, adapter);181    if (outcome.status === "completed") {182      return {183        ok: true,184        entriesReceived: outcome.entriesReceived,185        entriesSent: outcome.entriesSent,186      };187    }188    if (outcome.status === "vault_mismatch") {189      throw new Error("vault_id_mismatch");190    }191    if (outcome.status === "timeout") {192      throw new Error("sync_timeout");193    }194    throw new Error(outcome.error ?? "sync_errored");195  } finally {196    // The coordinator does NOT close the session (it returns and leaves the197    // session open). This runner is the sole owner of session lifetime.198    await session.close().catch(() => undefined);199  }200}201202async function connectLan(203  expectedRemoteHex: string,204  port: number,205  ourIdentity: SyncPeerIdentity,206  rendezvousIds: string[],207  deps: SyncRunnerDeps,208): Promise<SyncSession> {209  const cached = await getLastKnownIp(expectedRemoteHex);210  if (cached) {211    for (const rendezvousId of rendezvousIds) {212      try {213        const session = await deps.openSession(214          { ip: cached, port },215          ourIdentity,216          expectedRemoteHex,217          rendezvousId,218        );219        return session;220      } catch {221        // Try the next clock-skew window before treating the IP as stale.222      }223    }224  }225226  const subnet = await deps.getSubnet();227  if (!subnet) {228    throw new Error(cached ? "host_unreachable" : "subnet_unknown");229  }230231  const found = await deps.probeSubnet(232    subnet,233    port,234    ourIdentity,235    expectedRemoteHex,236    rendezvousIds,237  );238  if (!found) throw new Error("host_unreachable");239  const session = await deps.openSession(240    { ip: found.ip, port: found.port },241    ourIdentity,242    expectedRemoteHex,243    found.rendezvousId,244  );245  await setLastKnownIp(expectedRemoteHex, found.ip).catch(() => undefined);246  return session;247}248249async function connectRelay(250  relayUrl: string,251  expectedRemoteHex: string,252  ourIdentity: SyncPeerIdentity,253  rendezvousIds: string[],254  openRelaySession: NonNullable<SyncRunnerDeps["openRelaySession"]>,255): Promise<SyncSession> {256  let lastError: unknown;257  for (const rendezvousId of rendezvousIds) {258    try {259      return await openRelaySession(260        relayUrl,261        ourIdentity,262        expectedRemoteHex,263        rendezvousId,264      );265    } catch (error) {266      lastError = error;267    }268  }269  const detail = lastError instanceof Error ? lastError.message : "unreachable";270  throw new Error(`relay_sync_failed:${detail}`);271}272273function hexToBytesLocal(hex: string): Uint8Array {274  const out = new Uint8Array(hex.length / 2);275  for (let i = 0; i < out.length; i++) {276    out[i] = parseInt(hex.slice(i * 2, i * 2 + 2), 16);277  }278  return out;279}280281function bytesToHex(bytes: Uint8Array): string {282  return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join("");283}