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 { getRootKeyHex, isUnlocked } from "./vault";41import type { PairedDevice } from "../shared/sync-types";42import type { SyncRunResult } from "./sync-trigger";43import {44  dialingRendezvousIds,45  rendezvousSecretFromHex,46} from "../shared/rendezvous";47import {48  getRelaySettings,49  type RelaySettings,50} from "./relay-settings";5152/** Same fixed port pairing dials — SyncReconnectListener.SYNC_PORT on Android. */53const DEFAULT_PORT = PAIRING_PORT;5455/**56 * Test/injection seam. Production wires in the real `openSyncSession`;57 * unit tests inject a fake to avoid real WebSocket + Noise KK.58 */59export interface SyncRunnerDeps {60  openSession: (61    endpoint: SyncEndpoint,62    ourIdentity: SyncPeerIdentity,63    expectedRemoteHex: string,64    rendezvousId: string,65  ) => Promise<SyncSession>;66  probeSubnet: (67    subnet: string,68    port: number,69    ourIdentity: SyncPeerIdentity,70    expectedRemoteHex: string,71    rendezvousIds: readonly string[],72  ) => Promise<ProbedSyncEndpoint | null>;73  openRelaySession?: (74    relayUrl: string,75    ourIdentity: SyncPeerIdentity,76    expectedRemoteHex: string,77    rendezvousId: string,78  ) => Promise<SyncSession>;79  getRelaySettings?: () => Promise<RelaySettings>;80  getSubnet: () => Promise<string | null>;81  createAdapter: (context?: { deviceName?: string }) => SyncVaultAdapter;82}8384const productionDeps: SyncRunnerDeps = {85  openSession: (endpoint, identity, expected, rendezvousId) =>86    openSyncSession(endpoint, identity, expected, rendezvousId),87  probeSubnet: (subnet, port, identity, expected, rendezvousIds) =>88    probeSubnetForPeer(subnet, port, identity, expected, rendezvousIds),89  openRelaySession: (relayUrl, identity, expected, rendezvousId) =>90    openRelaySyncSession(relayUrl, identity, expected, rendezvousId),91  getRelaySettings,92  getSubnet: getLocalSubnet,93  createAdapter: (context) => createSyncVaultAdapter(context),94};9596export async function productionSyncRunner(device: PairedDevice): Promise<SyncRunResult> {97  return runWithDeps(device, productionDeps);98}99100/**101 * Inner runner with explicit deps — exported so tests can inject fakes.102 */103export async function runWithDeps(104  device: PairedDevice,105  deps: SyncRunnerDeps,106): Promise<SyncRunResult> {107  if (!isUnlocked()) return { ok: false, error: "VAULT_LOCKED" };108  if (!device.longTermPublicKeyHex) {109    throw new Error("paired_device_missing_pubkey");110  }111  if (!device.rendezvousSecretHex) {112    throw new Error("paired_device_missing_rendezvous_secret");113  }114  const expectedRemoteHex = device.longTermPublicKeyHex;115  const reconnectIds = dialingRendezvousIds(116    rendezvousSecretFromHex(device.rendezvousSecretHex),117    Date.now() / 1000,118  );119  const port = device.wsPort ?? DEFAULT_PORT;120121  // Sync identity derives from the stable root key, not the password-derived122  // key. Reconnect uses both halves directly in standard Noise KK.123  const masterKeyHex = getRootKeyHex();124  if (!masterKeyHex) return { ok: false, error: "VAULT_LOCKED" };125  const masterKey = hexToBytesLocal(masterKeyHex);126  const publicIdentity = await buildLocalIdentity(masterKey);127  const localKeys = deriveLocalSyncKeypair(masterKey, publicIdentity.deviceId);128  if (publicIdentity.longTermPublicKeyHex !== bytesToHex(localKeys.publicKey)) {129    masterKey.fill(0);130    localKeys.secretKey.fill(0);131    throw new Error("derived reconnect identity mismatch");132  }133  const ourIdentity: SyncPeerIdentity = {134    ...publicIdentity,135    longTermPrivateKey: localKeys.secretKey,136  };137138  // Sanity — the remote pubkey must be 32-byte hex so noise can compare it.139  void hexToPubkey(expectedRemoteHex);140141  // HV-1b: pass the peer being synced so remote-mutation alerts can name it.142  const adapter = deps.createAdapter({ deviceName: device.deviceName });143  let session: SyncSession;144  try {145    try {146      session = await connectLan(147        expectedRemoteHex,148        port,149        ourIdentity,150        reconnectIds,151        deps,152      );153    } catch (lanError) {154      const relaySettings = await deps.getRelaySettings?.();155      if (156        !relaySettings ||157        relaySettings.lanOnly ||158        !deps.openRelaySession159      ) {160        throw lanError;161      }162      session = await connectRelay(163        device.relayUrl ?? relaySettings.relayUrl,164        expectedRemoteHex,165        ourIdentity,166        reconnectIds,167        deps.openRelaySession,168      );169    }170  } finally {171    masterKey.fill(0);172    localKeys.secretKey.fill(0);173  }174175  try {176    const outcome = await runSyncInitiator(session, adapter);177    if (outcome.status === "completed") {178      return {179        ok: true,180        entriesReceived: outcome.entriesReceived,181        entriesSent: outcome.entriesSent,182      };183    }184    if (outcome.status === "vault_mismatch") {185      throw new Error("vault_id_mismatch");186    }187    if (outcome.status === "timeout") {188      throw new Error("sync_timeout");189    }190    throw new Error(outcome.error ?? "sync_errored");191  } finally {192    // The coordinator does NOT close the session (it returns and leaves the193    // session open). This runner is the sole owner of session lifetime.194    await session.close().catch(() => undefined);195  }196}197198async function connectLan(199  expectedRemoteHex: string,200  port: number,201  ourIdentity: SyncPeerIdentity,202  rendezvousIds: string[],203  deps: SyncRunnerDeps,204): Promise<SyncSession> {205  const cached = await getLastKnownIp(expectedRemoteHex);206  if (cached) {207    for (const rendezvousId of rendezvousIds) {208      try {209        const session = await deps.openSession(210          { ip: cached, port },211          ourIdentity,212          expectedRemoteHex,213          rendezvousId,214        );215        return session;216      } catch {217        // Try the next clock-skew window before treating the IP as stale.218      }219    }220  }221222  const subnet = await deps.getSubnet();223  if (!subnet) {224    throw new Error(cached ? "host_unreachable" : "subnet_unknown");225  }226227  const found = await deps.probeSubnet(228    subnet,229    port,230    ourIdentity,231    expectedRemoteHex,232    rendezvousIds,233  );234  if (!found) throw new Error("host_unreachable");235  const session = await deps.openSession(236    { ip: found.ip, port: found.port },237    ourIdentity,238    expectedRemoteHex,239    found.rendezvousId,240  );241  await setLastKnownIp(expectedRemoteHex, found.ip).catch(() => undefined);242  return session;243}244245async function connectRelay(246  relayUrl: string,247  expectedRemoteHex: string,248  ourIdentity: SyncPeerIdentity,249  rendezvousIds: string[],250  openRelaySession: NonNullable<SyncRunnerDeps["openRelaySession"]>,251): Promise<SyncSession> {252  let lastError: unknown;253  for (const rendezvousId of rendezvousIds) {254    try {255      return await openRelaySession(256        relayUrl,257        ourIdentity,258        expectedRemoteHex,259        rendezvousId,260      );261    } catch (error) {262      lastError = error;263    }264  }265  const detail = lastError instanceof Error ? lastError.message : "unreachable";266  throw new Error(`relay_sync_failed:${detail}`);267}268269function hexToBytesLocal(hex: string): Uint8Array {270  const out = new Uint8Array(hex.length / 2);271  for (let i = 0; i < out.length; i++) {272    out[i] = parseInt(hex.slice(i * 2, i * 2 + 2), 16);273  }274  return out;275}276277function bytesToHex(bytes: Uint8Array): string {278  return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join("");279}