From 48807e9931a47afc45b3802b32132c5530cb7c8e Mon Sep 17 00:00:00 2001 From: Brendan Baldwin Date: Sun, 30 Aug 2026 19:52:03 -0600 Subject: [PATCH 1/4] Add better-sync plan and archive completed sync design docs. Co-authored-by: Cursor --- docs/{ => archive}/local-embedding-plan.md | 0 docs/{ => archive}/multi-doc-get-plan.md | 0 docs/{ => archive}/sync-and-history-plan.md | 0 docs/better-sync.plan | 334 ++++++++++++++++++++ 4 files changed, 334 insertions(+) rename docs/{ => archive}/local-embedding-plan.md (100%) rename docs/{ => archive}/multi-doc-get-plan.md (100%) rename docs/{ => archive}/sync-and-history-plan.md (100%) create mode 100644 docs/better-sync.plan diff --git a/docs/local-embedding-plan.md b/docs/archive/local-embedding-plan.md similarity index 100% rename from docs/local-embedding-plan.md rename to docs/archive/local-embedding-plan.md diff --git a/docs/multi-doc-get-plan.md b/docs/archive/multi-doc-get-plan.md similarity index 100% rename from docs/multi-doc-get-plan.md rename to docs/archive/multi-doc-get-plan.md diff --git a/docs/sync-and-history-plan.md b/docs/archive/sync-and-history-plan.md similarity index 100% rename from docs/sync-and-history-plan.md rename to docs/archive/sync-and-history-plan.md diff --git a/docs/better-sync.plan b/docs/better-sync.plan new file mode 100644 index 0000000..7931072 --- /dev/null +++ b/docs/better-sync.plan @@ -0,0 +1,334 @@ +# Better Sync Plan — defer hot files, rebase before park, in-memory conflict hold + +Status: **design**. Extends [sync-and-history-plan.md](archive/sync-and-history-plan.md) +(§4.3–4.9). Targets the UX failure mode where a conflict sibling +(`-.md`) lands beside a note the user is actively editing, and the +related case where local bytes are ahead of stale embedded provenance but inbound sync +parks a sibling instead of rebasing in place. + +This plan does **not** introduce three-way markdown merge or a persistent conflict +database. It tightens the existing hash-gate + optimistic-concurrency model so +siblings are a last resort, not the default reaction to a hot or rebaseable file. + +## 0. Problem + +Today, inbound sync (feed + startup reconcile) applies this rule without nuance: + +> local dirty **and** incoming remote ≠ embedded `$version` → park remote as ignored sibling + +See `applyRef` in `src/sync/feed.ts` and row 5 of the §4.9 table in +`src/sync/reconcile.ts`. Outbound push is already smarter: `recoverStalePut` in +`src/sync/push.ts` tries to put local bytes on top of remote current before parking. + +Gaps: + +1. **Feed has no edit-awareness.** Push is debounced; feed polls every `--interval` + and may park a sibling while the user is still typing. +2. **`--settle` is misdocumented.** CLI help claims mtime-age filtering; the daemon + wires it only to chokidar `awaitWriteFinish` on the watcher (push direction). +3. **Rebase logic is push-only.** The highest-leverage fix for “disk is ahead, `$version` + is stale, remote also advanced” never runs on inbound paths. +4. **Conflicts are immediate and filesystem-visible.** There is no in-memory hold to + retry once the file settles or a rebase becomes possible. + +## 1. Principles (unchanged) + +These north stars from the sync plan still govern every change here: + +- **Never clobber local bytes** on the canonical path without proof it is safe. +- **Hash equality licenses adopt** — inject provenance in place, no server write. +- **Optimistic put with `prev_version_id`** is the preferred convergence path when + bytes diverge but the editor’s work should win as the new head. +- **Conflict siblings remain the last resort** — remote parked beside local with + `$sync: ignore`, deterministic name, idempotent replay. +- **No new persistent client state** beyond the existing `.mrplex/sync.json` cursor. + Deferred conflicts live in daemon memory only; restart replays from the feed. + +## 2. Scope + +**In:** + +1. **Feed deferral** — skip inbound materialize/conflict/park on paths whose mtime is + too recent (or otherwise “hot”). +2. **Rebase-before-park** — shared inbound logic mirroring `recoverStalePut` / + `occupiedPath`: hash-equal adopt, else try `docs.put`, park sibling only on failure. +3. **`--settle` semantics** — restore documented mtime-age meaning; apply symmetrically + to inbound touch decisions and outbound push scheduling. +4. **In-memory deferred conflicts** — hold would-be sibling writes in daemon memory; + retry on later polls when the file is settled; no log line, no disk artifact. + +**Out (deliberately):** + +- Three-way or block-level markdown merge (`docs.merge_preview`, etc.). +- Persistent conflict queue / sidecar manifest. +- Obsidian open-buffer integration (no API available). +- Changes to kernel concurrency (`stale_prev`, `create_conflict` codes stay as-is). + +## 3. Architecture + +### 3.1 New module: `src/sync/hot-path.ts` + +Centralize “may we touch this path right now?” and “how do we converge inbound?” + +```ts +/** True when path mtime is within settleMs of now (0 = no gate). */ +export function isPathHot(store: FileStore, path: string, settleMs: number): Promise; + +/** Inbound convergence: adopt | rebase-put | defer | park-sibling. */ +export type InboundVerdict = + | { action: "noop" } + | { action: "adopt"; version: Version } + | { action: "rebase"; prevVersionId: string; user: UserContent } + | { action: "defer"; ref: VersionRef } // hold in memory + | { action: "park"; version: Version }; // write ignored sibling + +export function resolveInbound( + client, store, repo, path, localText, ref, opts: { settleMs: number; hot: boolean }, +): Promise; +``` + +`resolveInbound` encapsulates the decision order in §4.2. Callers (`feed.ts`, +`reconcile.ts`) replace ad-hoc dirty/version checks with this function. + +### 3.2 Daemon state: in-memory deferred conflicts + +Add to `startDaemon` in `src/sync/daemon.ts`: + +```ts +/** Paths we would have conflict-parked but are holding until settled or rebase succeeds. */ +const deferred = new Map(); +``` + +Properties: + +- **Key:** canonical doc path (repo-relative POSIX). +- **Value:** the `VersionRef` we deferred (plus `since` timestamp for optional TTL). +- **Not logged** — no `deferred-conflict` stderr line; optional `--verbose` may mention + retries, but default output stays unchanged. +- **Not persisted** — daemon restart drops the map; the next `history.since` drain + re-encounters the ref (idempotent — no duplicate siblings if one was already written). +- **Processed before each feed page** — `retryDeferred(deferred)` runs at the start of + `pollFeedOnce` (inside the existing serialize chain). + +When a deferred entry succeeds (adopt/rebase/noop), delete the map entry. When rebase +fails and the path is no longer hot, fall through to park and delete the entry. + +Optional safety cap: drop deferred entries older than e.g. 30 minutes and park immediately +(prevents infinite deferral during an all-day editing session). Tunable constant, not CLI +for v1. + +### 3.3 `--settle` restored semantics + +| Direction | Today | After | +|-----------|-------|-------| +| Push (watcher) | `awaitWriteFinish.stabilityThreshold = settleMs` | Mtime age ≥ `settleMs` before `pushPath`; keep `awaitWriteFinish` with `min(settleMs, 2000)` when `settleMs > 0` so rapid saves still coalesce | +| Feed / reconcile | ignored | Skip materialize + conflict park when `isPathHot`; enqueue defer instead | +| Provenance ack | `preserveMtime: true` | unchanged — ack writes never participate in the hot gate | + +Default remains `0` (no gate) for backward compatibility. + +Update CLI help in `src/cli/main.ts` to match behavior: + +``` +--settle minimum mtime age before sync reads or writes a path (default 0) +``` + +## 4. Workstreams + +### WS1 — Feed deferral (mtime hot gate) + +**Goal:** inbound sync does not materialize or park siblings on files the editor likely +still has open. + +**Behavior:** + +Before any inbound filesystem effect at `path` (materialize, fast-forward, conflict +park, delete of clean file): + +1. If `settleMs === 0`, proceed (current behavior modulo WS2). +2. If `isPathHot(path)`, do **not** write. For conflict-shaped situations, insert + `{ path → ref }` into `deferred` (WS4). For benign noops (hash-equal adopt), still + defer the provenance stamp — stamping during an active edit can confuse Obsidian Sync + even with `preserveMtime`; retry when cold. +3. Feed cursor **still advances** for deferred refs — the ref is reflected in memory, not + on disk. Rationale: stalling the global cursor blocks the entire vault for one hot + file. The deferred map is the per-path hold. + +**Exception:** witnessed **clean delete** on a cold file stays immediate. Hot + dirty +delete already no-ops (existing §4.3). + +**Files:** `src/sync/feed.ts`, `src/sync/hot-path.ts`, `src/sync/daemon.ts` (pass +`settleMs`, `deferred` into `applyFeed` options). + +**Acceptance:** + +- Feed poll while mtime is fresh → no sibling, entry in `deferred`. +- After mtime ages past `settleMs` → retry applies adopt/rebase/park per WS2. +- Cursor advances even when all refs in a page defer. +- `--settle 0` → no deferrals from mtime. + +### WS2 — Rebase-before-park (feed + reconcile) + +**Goal:** extend outbound `recoverStalePut` / `occupiedPath` logic to inbound paths so +“local ahead + stale `$version` + remote advanced” converges in place when the kernel +accepts the put. + +**Decision order** (for `resolveInbound` when local file exists and incoming remote +current ≠ embedded `$version`, or reconcile equivalent): + +| Step | Condition | Action | +|------|-----------|--------| +| 1 | `computed_hash === remote.content_hash` | **Adopt** — `renderMaterialized` in place, `preserveMtime: true` | +| 2 | `versionAtOrAhead(local, remote)` | **Noop** — local already names this or a later version | +| 3 | `!isDirty(local)` | **Materialize** — fast-forward (existing row 3 / feed materialize) | +| 4 | hot (WS1) | **Defer** — memory hold | +| 5 | dirty + divergent | **Rebase** — strip user content, `docs.put(repo, remote.version_id, path, user)` | +| 6 | rebase → success | **Ack** — `ackLocalWrite` / `renderMaterialized` with `preserveMtime` | +| 7 | rebase → `stale_prev` / `path_taken` / hash mismatch after race | **Park** — ignored sibling (§4.8) | + +Rebase (step 5) is the new branch. It matches `recoverStalePut`’s “same-path continuation” +comment: the editor is on stale provenance but has newer bytes; put those bytes on top of +remote current instead of parking a sibling that Obsidian then syncs into the vault. + +**Reconcile changes** (`src/sync/reconcile.ts`): + +- Row 5 (stale + dirty): call `resolveInbound` instead of unconditional `parkConflict`. +- Row 6 (occupied, no provenance): already partially handled; route through shared helper. +- `parkConflict` / `parkConflictForCurrent` become the terminal action of `resolveInbound`. + +**Feed changes** (`src/sync/feed.ts`): + +- Replace the block at lines 132–143 (dirty → immediate park) with `resolveInbound`. +- Hash-equal branch (lines 122–130) folds into step 1. + +**Shared push helpers:** + +- Extract `occupiedPath`, `commitPut`, `ackLocalWrite`, `stripToUserContent` usage into + imports from `push.ts` or a thin `src/sync/converge.ts` to avoid circular deps + (feed/reconcile must not import the whole push burst machinery). + +**Acceptance:** + +- Stale `$version`, dirty local, local bytes = remote current → adopt, no sibling. +- Stale `$version`, dirty local, local bytes differ, put succeeds → in-place head, no sibling. +- True divergence (put fails) → sibling, same as today. +- Existing `test/sync-feed.test.ts`, `test/sync-reconcile.test.ts`, `test/sync-push.test.ts` + scenarios still pass; new cases cover rebase path. + +### WS3 — Fix `--settle` semantics + +**Goal:** CLI flag matches implementation; one knob for “don’t touch files still being +written.” + +**Implementation:** + +1. Implement `isPathHot` in `src/sync/hot-path.ts` using `fs.stat` mtime vs + `Date.now() - settleMs`. +2. Push path: in `schedulePush` / before `pushBurst`, re-queue paths that fail the hot + check (reset debounce timer, do not call `pushPath` yet). +3. Feed path: WS1. +4. chokidar: when `settleMs > 0`, set + `awaitWriteFinish: { stabilityThreshold: Math.min(settleMs, 2000), pollInterval: 100 }` + as a complementary write-stability guard (not a substitute for mtime age). +5. Update CLI help and `docs/archive/sync-and-history-plan.md` cross-reference (one line + in this plan’s §7 changelog; full doc edit optional). + +**Acceptance:** + +- `--settle 3000` + file modified 1s ago → push and inbound touch both wait. +- `--settle 0` → no mtime gate (stability threshold off unless we keep a separate default; + prefer fully off for 0). +- `test/sync-fs-store.test.ts` preserveMtime cases unaffected. + +### WS4 — In-memory deferred conflicts + +**Goal:** when WS1 or WS2 would park but the path is hot, hold the ref in memory and +retry silently until the file is cold or rebase succeeds. + +**Behavior:** + +1. **Enqueue:** `deferred.set(path, { ref, since: Date.now() })` when `resolveInbound` + returns `{ action: "defer" }`. No stderr output. +2. **Retry:** at the start of each serialized `pollFeedOnce`: + - For each deferred entry (snapshot copy of map to avoid mutation during iteration): + - Re-read local file; re-fetch remote version for `ref.version_id` if needed. + - Call `resolveInbound` with `hot: isPathHot(...)`. + - On adopt/rebase/noop → remove entry. + - On park (cold + rebase failed) → write sibling, remove entry. + - On defer → keep entry. +3. **Dedup:** if `deferred` already has `path`, replace `ref` only when incoming + `ref.version_id` is newer (decode integer ids). +4. **Interaction with feed cursor:** deferred refs correspond to feed positions already + passed; retry uses `client.docs.get` / `get_version` at canonical path, not replay of + old refs. The held `VersionRef` is a hint for which remote version we need to converge + with. +5. **Restart:** map empty; next feed poll may re-park if still divergent and cold — acceptable. + +**Not in scope:** logging deferred state, persisting queue, UI indicators. + +**Acceptance:** + +- Active edit + inbound remote change → no sibling during hot window; sibling only if still + divergent after cold + rebase failure. +- `--verbose` may print retry outcomes; default silent. +- `test/sync-daemon.test.ts`: simulate hot mtime → deferred; advance clock or touch mtime → + convergence. + +## 5. Scenario notes + +### A — Obsidian local → Obsidian Sync → server vault → mrplex sync + +- Sibling originates on server; Obsidian Sync propagates it locally. +- WS2 reduces false siblings when server file is dirty with stale `$version` but content + is rebaseable onto remote current. +- WS1/WS4 reduce siblings when Obsidian Sync delivers the edit while server mrplex poll runs. +- Mtime on server may lag Obsidian Sync delivery — defer+retry still helps on the next poll. + +### B — Obsidian local + mrplex sync on same machine + +- WS1/WS4 directly address “sibling appeared while I was editing.” +- WS2 addresses stale provenance after interrupted ack (crash between put and restamp). +- Recommend `--settle 2000` or `--settle 3000` in docs as a sensible Obsidian default once + implemented (not changed as CLI default in code). + +## 6. Test plan + +| Area | Tests | +|------|-------| +| `hot-path.ts` | unit: `isPathHot`, `resolveInbound` decision table | +| Feed | hot file → defer; cold + rebase → in place; cold + fail → sibling | +| Reconcile | row 5 rebase; row 5 park unchanged for true conflict | +| Daemon | deferred map retry loop; cursor advances with deferrals | +| Push + settle | push delayed until mtime age; `--settle 0` unchanged | +| Regression | echo suppression, hash-equal adopt, `$sync: ignore`, move/delete | + +Use in-memory `FileStore` test doubles with controllable mtimes (extend test store or mock +`stat` via injectable clock + mtime map). + +## 7. Implementation order + +1. **WS3 + WS1 foundation** — `hot-path.ts`, `isPathHot`, wire `settleMs` through feed opts. +2. **WS4** — deferred map in daemon, feed defer enqueue, retry loop. +3. **WS2** — `resolveInbound` rebase branch; refactor feed + reconcile to use it. +4. **CLI/docs** — help text, optional README note on Obsidian `--settle`. + +Estimated touch surface: `src/sync/hot-path.ts` (new), `src/sync/converge.ts` (new, optional), +`src/sync/feed.ts`, `src/sync/reconcile.ts`, `src/sync/push.ts` (extract shared ack/put), +`src/sync/daemon.ts`, `src/cli/main.ts`, tests under `test/sync-*.test.ts`. + +## 8. Open questions + +1. **Cursor advance vs defer** — this plan advances the global cursor while holding per-path + deferred state. Alternative: stall cursor at first deferred ref (simpler replay, blocks + whole vault). Recommendation: advance + per-path hold unless integration tests show + missed convergence. +2. **Defer TTL** — 30-minute cap before forced park: keep, drop, or make configurable? +3. **`--once` reconcile** — apply hot gate on startup reconcile? Recommendation: yes for + conflict park; no for remote-only materialize of files absent locally (less surprising). +4. **Rebase on feed delete** — if remote deleted and local is dirty+hot, existing keep-local + behavior stands; no defer needed. + +--- + +*Changelog: initial draft — four workstreams (deferral, rebase-before-park, settle fix, +in-memory deferred conflicts without logging).* From 52daee820bd4d57cd5c342f255c5e37b9f1db7e5 Mon Sep 17 00:00:00 2001 From: Brendan Baldwin Date: Sun, 30 Aug 2026 23:49:41 -0600 Subject: [PATCH 2/4] Implement better-sync: defer hot paths, rebase before park. Inbound feed and reconcile now try optimistic rebase onto remote current before parking conflict siblings, and the daemon holds hot-file work in memory with settle-based mtime gating on both push and feed directions. Co-authored-by: Cursor --- src/cli/main.ts | 2 +- src/sync/converge.ts | 76 +++++++++ src/sync/daemon.ts | 67 +++++++- src/sync/feed.ts | 173 ++++++++++++-------- src/sync/fs-store.ts | 8 + src/sync/hot-path.ts | 304 ++++++++++++++++++++++++++++++++++++ src/sync/push.ts | 60 ++----- src/sync/reconcile.ts | 112 +++++++------ test/sync-feed.test.ts | 90 +++++++++-- test/sync-hot-path.test.ts | 186 ++++++++++++++++++++++ test/sync-push.test.ts | 12 +- test/sync-reconcile.test.ts | 42 +++-- test/sync-scenarios.test.ts | 9 +- 13 files changed, 942 insertions(+), 199 deletions(-) create mode 100644 src/sync/converge.ts create mode 100644 src/sync/hot-path.ts create mode 100644 test/sync-hot-path.test.ts diff --git a/src/cli/main.ts b/src/cli/main.ts index 62627f2..311c328 100644 --- a/src/cli/main.ts +++ b/src/cli/main.ts @@ -1684,7 +1684,7 @@ function buildProgram(): Command { .option("--debounce ", "burst debounce in ms (daemon; default 5000)", parsePositiveInt) .option( "--settle ", - "skip files younger than this many ms (partial saves)", + "minimum mtime age in ms before sync touches a path (default 0)", parsePositiveInt, ) .option( diff --git a/src/sync/converge.ts b/src/sync/converge.ts new file mode 100644 index 0000000..addd834 --- /dev/null +++ b/src/sync/converge.ts @@ -0,0 +1,76 @@ +/** + * Shared disk↔kernel write helpers used by push, feed, and reconcile so + * inbound rebase and outbound ack stay byte-identical (better-sync.plan). + */ + +import type { KernelClient } from "../client/kernel-client.js"; +import { withVersionSuffix } from "../kernel/deletion.js"; +import type { Version } from "../kernel/wire.js"; +import { extractSystemProperties, split } from "../markdown/frontmatter.js"; +import { renderIgnoredSibling, renderMaterialized, stampProvenance } from "./intrinsics.js"; +import type { FileStore } from "./reconcile.js"; + +export type UserContent = { frontmatter_raw: string; body: string }; + +/** Split a file into stored-shape fields, dropping all `$*` intrinsic lines. */ +export function stripToUserContent(text: string): UserContent { + const lf = text.replace(/\r\n/g, "\n"); + const { frontmatter_raw, body } = split(lf); + return { frontmatter_raw: extractSystemProperties(frontmatter_raw).raw, body }; +} + +/** + * After a successful kernel write, restamp the local file. If the editor saved + * again during the round-trip, keep those bytes and only update provenance — + * never write the snapshot we just pushed over newer typing. + */ +export async function ackLocalWrite( + store: FileStore, + path: string, + v: Version, + pushed: UserContent, +): Promise { + const now = await store.read(path); + if (now === null) return; + const current = stripToUserContent(now); + if (current.body === pushed.body && current.frontmatter_raw === pushed.frontmatter_raw) { + await store.write(path, renderMaterialized(v), { preserveMtime: true }); + return; + } + await store.write(path, stampProvenance(now, v.version_id, v.content_hash), { + preserveMtime: true, + }); +} + +/** Optimistic put of local user bytes, then ack provenance on disk. */ +export async function putAndAck( + client: KernelClient, + store: FileStore, + repo: string, + path: string, + prevVersionId: string, + user: UserContent, +): Promise { + const v = await client.docs.put(repo, prevVersionId, path, user); + await ackLocalWrite(store, path, v, user); + return v; +} + +/** Park remote as `-.md` with `$sync: ignore` (§4.8). */ +export async function parkIgnoredSibling( + store: FileStore, + path: string, + version: Version, +): Promise { + await store.write(withVersionSuffix(path, version.version_id), renderIgnoredSibling(version)); +} + +/** Write remote version bytes at the canonical path (fast-forward / materialize). */ +export async function materializeAt( + store: FileStore, + path: string, + version: Version, + opts?: { preserveMtime?: boolean }, +): Promise { + await store.write(path, renderMaterialized(version), opts); +} diff --git a/src/sync/daemon.ts b/src/sync/daemon.ts index 0bd26f3..79ef46c 100644 --- a/src/sync/daemon.ts +++ b/src/sync/daemon.ts @@ -9,6 +9,10 @@ * first so unlink+add of a rename is a move, not a delete (§4.7); each * path is still stated, not trusted by event type (§4.6). * + * better-sync.plan: `--settle` is an mtime-age gate on both directions; inbound + * conflicts on hot files are held in an in-memory deferred map and retried on + * later polls (cursor still advances). + * * chokidar is confined to this module (§4.4). Echo suppression falls out of * self-description (§4.5): our own pushes come back on the feed but the local * file already embeds that version+hash, and our own writes trip the watcher @@ -21,6 +25,11 @@ import type { KernelClient } from "../client/kernel-client.js"; import { type SyncCursor, readCursor, sourceFields, writeCursor } from "./cursor.js"; import { applyFeed } from "./feed.js"; import { createFsStore } from "./fs-store.js"; +import { + type DeferredMap, + pathIsHot, + retryDeferredEntry, +} from "./hot-path.js"; import { isIgnored, readFileIntrinsics } from "./intrinsics.js"; import { SYNC_DIR, type ScopeFilter, makeScopeFilter, toDocPath } from "./paths.js"; import { type RemoteMap, pushBurst, pushPath } from "./push.js"; @@ -55,8 +64,10 @@ export function startDaemon(client: KernelClient, opts: DaemonOptions): Daemon { const store = createFsStore(opts.root); const scope = makeScopeFilter({ include: opts.include, exclude: opts.exclude }); const map: RemoteMap = new Map(); + const deferred: DeferredMap = new Map(); const intervalMs = opts.intervalMs ?? 5000; const debounceMs = opts.debounceMs ?? 5000; + const settleMs = opts.settleMs ?? 0; let stopped = false; let watcher: FSWatcher | undefined; @@ -99,12 +110,36 @@ export function startDaemon(client: KernelClient, opts: DaemonOptions): Daemon { } } + /** Retry in-memory deferred inbound holds before draining new feed pages. */ + async function retryDeferred(): Promise { + if (deferred.size === 0) return; + for (const [path, entry] of [...deferred.entries()]) { + if (!scope.matches(path) && entry.ref.op !== "delete") { + deferred.delete(path); + continue; + } + try { + const { done } = await retryDeferredEntry(client, store, opts.repo, path, entry, { + settleMs, + map, + deferred, + }); + if (done) deferred.delete(path); + } catch (err) { + log(`defer retry error\t${path}\t${(err as Error).message}`); + } + } + } + async function pollFeedOnce(): Promise { + await retryDeferred(); const { cursor: next } = await applyFeed(client, store, scope, { repo: opts.repo, since: cursor, log, map, + settleMs, + deferred, }); if (next !== cursor) { cursor = next; @@ -112,9 +147,7 @@ export function startDaemon(client: KernelClient, opts: DaemonOptions): Daemon { } } - function schedulePush(docPath: string): void { - if (!scope.matches(docPath)) return; - pending.add(docPath); + function armBurstTimer(): void { if (burstTimer) clearTimeout(burstTimer); burstTimer = setTimeout(() => { burstTimer = undefined; @@ -122,11 +155,28 @@ export function startDaemon(client: KernelClient, opts: DaemonOptions): Daemon { pending.clear(); void serialize(async () => { if (stopped) return; - await pushBurst(batch, { client, store, repo: opts.repo, map, log }); + const ready: string[] = []; + for (const path of batch) { + if (settleMs > 0 && (await pathIsHot(store, path, settleMs))) { + pending.add(path); + continue; + } + ready.push(path); + } + if (pending.size > 0) armBurstTimer(); + if (ready.length > 0) { + await pushBurst(ready, { client, store, repo: opts.repo, map, log }); + } }); }, debounceMs); } + function schedulePush(docPath: string): void { + if (!scope.matches(docPath)) return; + pending.add(docPath); + armBurstTimer(); + } + const ready = (async () => { // 1. Startup is deterministic on the cursor marker (§4.9, §7). const existing = await readCursor(opts.root); @@ -160,8 +210,13 @@ export function startDaemon(client: KernelClient, opts: DaemonOptions): Daemon { ignoreInitial: true, // Prune the sync state dir; the scope filter still guards everything else. ignored: (p: string) => p.includes(`/${SYNC_DIR}/`) || p.endsWith(`/${SYNC_DIR}`), - ...(opts.settleMs - ? { awaitWriteFinish: { stabilityThreshold: opts.settleMs, pollInterval: 100 } } + ...(settleMs > 0 + ? { + awaitWriteFinish: { + stabilityThreshold: Math.min(settleMs, 2000), + pollInterval: 100, + }, + } : {}), }); const onEvent = (abs: string): void => { diff --git a/src/sync/feed.ts b/src/sync/feed.ts index 70ae8d9..f351b48 100644 --- a/src/sync/feed.ts +++ b/src/sync/feed.ts @@ -8,19 +8,24 @@ * absent file is a no-op; the dirty check prevents a replay from clobbering an * edit made in a crash window. So a crash between apply and cursor-advance * simply replays the batch harmlessly. + * + * better-sync.plan: hot files defer inbound effects into an in-memory map + * (cursor still advances); dirty+stale local files rebase onto remote current + * before parking a sibling. */ import type { KernelClient } from "../client/kernel-client.js"; -import { withVersionSuffix } from "../kernel/deletion.js"; -import { decodeVersionId } from "../kernel/version-id.js"; import type { VersionRef } from "../kernel/wire.js"; +import { materializeAt } from "./converge.js"; import { - isDirty, - isIgnored, - readFileIntrinsics, - renderIgnoredSibling, - renderMaterialized, -} from "./intrinsics.js"; + type DeferredMap, + decideInbound, + effectInbound, + enqueueDeferred, + pathIsHot, + versionAtOrAhead, +} from "./hot-path.js"; +import { isDirty, isIgnored, readFileIntrinsics } from "./intrinsics.js"; import type { ScopeFilter } from "./paths.js"; import type { RemoteMap } from "./push.js"; import type { FileStore } from "./reconcile.js"; @@ -35,6 +40,15 @@ export type ApplyFeedOptions = { * witnessed unlink knows the file's prev_version_id. */ map?: RemoteMap; + /** + * Minimum mtime age before inbound touch (better-sync `--settle`). When a + * deferred map is also supplied, hot paths enqueue instead of writing. + */ + settleMs?: number; + /** In-memory deferred holds (daemon only). Absent → never defer (e.g. `--once`). */ + deferred?: DeferredMap; + /** Injectable clock for tests. */ + nowMs?: () => number; }; export type ApplyFeedResult = { @@ -42,6 +56,8 @@ export type ApplyFeedResult = { cursor: string; /** Number of refs whose filesystem effect was applied. */ applied: number; + /** Number of refs held in the deferred map (hot path). */ + deferred: number; }; /** @@ -58,10 +74,13 @@ export async function applyFeed( const log = opts.log ?? (() => {}); let cursor = opts.since; let applied = 0; + let deferredCount = 0; for (;;) { const page = await client.history.since({ after_version: cursor, repo: opts.repo }); for (const ref of page.refs) { - if (await applyRef(client, store, scope, opts.repo, ref, log, opts.map)) applied++; + const outcome = await applyRef(client, store, scope, opts, ref, log); + if (outcome === "applied") applied++; + else if (outcome === "deferred") deferredCount++; } // No forward progress → caught up (or a hot gap). Stop draining. if (page.next_since === cursor || page.refs.length === 0) { @@ -70,91 +89,121 @@ export async function applyFeed( } cursor = page.next_since; } - return { cursor, applied }; + return { cursor, applied, deferred: deferredCount }; } -/** Apply one feed ref to the local store. Returns true if it changed a file. */ +type ApplyOutcome = "applied" | "deferred" | "noop"; + +/** Apply one feed ref to the local store. */ async function applyRef( client: KernelClient, store: FileStore, scope: ScopeFilter, - repo: string, + opts: ApplyFeedOptions, ref: VersionRef, log: (msg: string) => void, - map?: RemoteMap, -): Promise { +): Promise { + const settleMs = opts.settleMs ?? 0; + const nowMs = opts.nowMs?.() ?? Date.now(); + const canDefer = opts.deferred !== undefined; + if (ref.op === "delete") { - // Remove the local file at prev_path IF clean; dirty ⇒ keep (resurrection - // happens on its next local-change cycle). §4.3 delete. const target = ref.prev_path; - if (target === null || !scope.matches(target)) return false; - map?.delete(target); + if (target === null || !scope.matches(target)) return "noop"; + opts.map?.delete(target); const text = await store.read(target); - if (text === null) return false; + if (text === null) return "noop"; const intr = readFileIntrinsics(text); - if (isIgnored(intr) || isDirty(intr)) return false; // preserve local work + if (isIgnored(intr) || isDirty(intr)) return "noop"; // preserve local work + if (canDefer && (await pathIsHot(store, target, settleMs, nowMs))) { + enqueueDeferred(opts.deferred!, target, ref, nowMs); + return "deferred"; + } await store.remove(target); log(`feed delete\t${target}`); - return true; + return "applied"; } if (ref.op === "move") { const from = ref.prev_path; if (from !== null && scope.matches(from)) { - map?.delete(from); + opts.map?.delete(from); const text = await store.read(from); if (text !== null) { const intr = readFileIntrinsics(text); - if (!isIgnored(intr) && !isDirty(intr)) await store.remove(from); + if (!isIgnored(intr) && !isDirty(intr)) { + if (canDefer && (await pathIsHot(store, from, settleMs, nowMs))) { + // Hold the whole move (including dest) until source is cold. + enqueueDeferred(opts.deferred!, ref.path, ref, nowMs); + return "deferred"; + } + await store.remove(from); + } } } // Fall through to materialize at the destination path. } - // create / update / move-destination: materialize at ref.path unless the - // local file already holds these bytes, is at-or-ahead of this ref, is - // dirty against a *newer* remote, or is $sync: ignore. - if (!scope.matches(ref.path)) return false; + if (!scope.matches(ref.path)) return "noop"; const existing = await store.read(ref.path); if (existing !== null) { const intr = readFileIntrinsics(existing); - if (isIgnored(intr)) return false; + if (isIgnored(intr)) return "noop"; + + // Fast path: bytes already match — may still need provenance repair. if (intr.computed_hash === ref.content_hash) { - // Bytes already present (our own push echoing back, or a vault copy). - map?.set(ref.path, { version_id: ref.version_id, content_hash: ref.content_hash }); - // Repair provenance only if the embedded version lags. - if (intr.version === ref.version_id) return false; - if (versionAtOrAhead(intr.version, ref.version_id)) return false; - const v = await client.docs.get_version(repo, ref.version_id); - await store.write(ref.path, renderMaterialized(v), { preserveMtime: true }); - return true; - } - // Local is at or ahead of this ref (echo of our push, or a replay of an - // older version). Never clobber and never park a sibling of something we - // already have — including when the user has typed more since (dirty). - if (versionAtOrAhead(intr.version, ref.version_id)) return false; - if (isDirty(intr)) { - // A local edit collides with a *newer* incoming version → conflict, not - // overwrite. Park the remote as an ignored sibling; keep local bytes. - const v = await client.docs.get_version(repo, ref.version_id); - await store.write(withVersionSuffix(ref.path, ref.version_id), renderIgnoredSibling(v)); - log(`feed conflict\t${ref.path}`); - return true; + opts.map?.set(ref.path, { version_id: ref.version_id, content_hash: ref.content_hash }); + if (intr.version === ref.version_id) return "noop"; + if (versionAtOrAhead(intr.version, ref.version_id)) return "noop"; + const hot = await pathIsHot(store, ref.path, settleMs, nowMs); + const v = await client.docs.get_version(opts.repo, ref.version_id); + const decision = decideInbound({ + localText: existing, + remote: v, + hot, + canDefer, + }); + const result = await effectInbound(client, store, opts.repo, ref.path, decision, { + deferred: opts.deferred, + ref, + map: opts.map, + nowMs, + }); + if (result === "deferred") return "deferred"; + if (result === "noop") return "noop"; + log(`feed adopt\t${ref.path}`); + return "applied"; } + + if (versionAtOrAhead(intr.version, ref.version_id)) return "noop"; + + // Divergent local vs newer remote → decideInbound (rebase / defer / park). + const hot = await pathIsHot(store, ref.path, settleMs, nowMs); + const v = await client.docs.get_version(opts.repo, ref.version_id); + const decision = decideInbound({ + localText: existing, + remote: v, + hot, + canDefer, + }); + const result = await effectInbound(client, store, opts.repo, ref.path, decision, { + deferred: opts.deferred, + ref, + map: opts.map, + nowMs, + }); + if (result === "deferred") return "deferred"; + if (result === "noop") return "noop"; + if (result === "parked") log(`feed conflict\t${ref.path}`); + else if (decision.action === "rebase") log(`feed rebase\t${ref.path}`); + else log(`feed ${ref.op}\t${ref.path}`); + return "applied"; } - const v = await client.docs.get_version(repo, ref.version_id); - await store.write(ref.path, renderMaterialized(v)); - map?.set(ref.path, { version_id: ref.version_id, content_hash: ref.content_hash }); - log(`feed ${ref.op}\t${ref.path}`); - return true; -} -/** True when the local embedded version is this ref or a later one. */ -function versionAtOrAhead(localVersion: string | undefined, refVersion: string): boolean { - if (!localVersion) return false; - if (localVersion === refVersion) return true; - const local = decodeVersionId(localVersion); - const ref = decodeVersionId(refVersion); - if (local === null || ref === null) return false; - return local > ref; + // Absent locally → materialize (unless somehow hot — can't be; no mtime). + const v = await client.docs.get_version(opts.repo, ref.version_id); + await materializeAt(store, ref.path, v); + opts.map?.set(ref.path, { version_id: ref.version_id, content_hash: ref.content_hash }); + log(`feed ${ref.op}\t${ref.path}`); + return "applied"; } diff --git a/src/sync/fs-store.ts b/src/sync/fs-store.ts index f6f3443..0e0556c 100644 --- a/src/sync/fs-store.ts +++ b/src/sync/fs-store.ts @@ -42,6 +42,14 @@ export function createFsStore(root: string): FileStore { async remove(docPath: string): Promise { await rm(toLocalPath(root, docPath), { force: true }); }, + async mtime(docPath: string): Promise { + try { + return (await stat(toLocalPath(root, docPath))).mtimeMs; + } catch (err) { + if ((err as NodeJS.ErrnoException).code === "ENOENT") return null; + throw err; + } + }, }; } diff --git a/src/sync/hot-path.ts b/src/sync/hot-path.ts new file mode 100644 index 0000000..6f98f25 --- /dev/null +++ b/src/sync/hot-path.ts @@ -0,0 +1,304 @@ +/** + * Hot-path detection and inbound convergence decisions (better-sync.plan). + * Feed/reconcile call `decideInbound` then `effectInbound`; the daemon holds + * deferred refs in memory and retries them on later polls. + */ + +import type { KernelClient } from "../client/kernel-client.js"; +import { KernelError } from "../kernel/errors.js"; +import { decodeVersionId } from "../kernel/version-id.js"; +import type { Version, VersionRef } from "../kernel/wire.js"; +import { + materializeAt, + parkIgnoredSibling, + putAndAck, + stripToUserContent, + type UserContent, +} from "./converge.js"; +import { isDirty, isIgnored, readFileIntrinsics } from "./intrinsics.js"; +import type { FileStore } from "./reconcile.js"; +import type { RemoteMap } from "./push.js"; + +/** Drop deferred entries older than this and force cold treatment (park/rebase). */ +export const DEFER_TTL_MS = 30 * 60 * 1000; + +export type DeferredEntry = { ref: VersionRef; since: number }; +export type DeferredMap = Map; + +export type InboundDecision = + | { action: "noop" } + | { action: "adopt"; version: Version } + | { action: "materialize"; version: Version } + | { action: "rebase"; prevVersionId: string; user: UserContent } + | { action: "defer" }; + +/** + * Pure mtime gate: true when the file was modified within settleMs of now. + * settleMs ≤ 0 disables the gate. Absent files (null mtime) are never hot. + */ +export function isPathHot( + mtimeMs: number | null, + settleMs: number, + nowMs: number = Date.now(), +): boolean { + if (settleMs <= 0) return false; + if (mtimeMs === null) return false; + return nowMs - mtimeMs < settleMs; +} + +/** Store-backed hot check. */ +export async function pathIsHot( + store: FileStore, + path: string, + settleMs: number, + nowMs: number = Date.now(), +): Promise { + return isPathHot(await store.mtime(path), settleMs, nowMs); +} + +/** True when the local embedded version is this ref or a later one. */ +export function versionAtOrAhead(localVersion: string | undefined, refVersion: string): boolean { + if (!localVersion) return false; + if (localVersion === refVersion) return true; + const local = decodeVersionId(localVersion); + const ref = decodeVersionId(refVersion); + if (local === null || ref === null) return false; + return local > ref; +} + +export function isDeferExpired( + entry: DeferredEntry, + nowMs: number, + ttlMs: number = DEFER_TTL_MS, +): boolean { + return nowMs - entry.since >= ttlMs; +} + +/** + * Insert or replace a deferred hold. Newer version_id wins when both decode. + */ +export function enqueueDeferred( + deferred: DeferredMap, + path: string, + ref: VersionRef, + nowMs: number = Date.now(), +): void { + const existing = deferred.get(path); + if (existing) { + const old = decodeVersionId(existing.ref.version_id); + const neu = decodeVersionId(ref.version_id); + if (old !== null && neu !== null && neu <= old) return; + // Keep original `since` so TTL is from first deferral of this path. + deferred.set(path, { ref, since: existing.since }); + return; + } + deferred.set(path, { ref, since: nowMs }); +} + +/** + * Decide how to converge an existing local file with an incoming remote version. + * Does not touch the filesystem or kernel — callers run `effectInbound`. + * + * When `canDefer` is false (e.g. `--once` with no deferred map), hot files are + * treated as cold so the pass still converges. + */ +export function decideInbound(opts: { + localText: string; + remote: Version; + hot: boolean; + canDefer: boolean; +}): InboundDecision { + const { localText, remote } = opts; + const treatHot = opts.hot && opts.canDefer; + const intr = readFileIntrinsics(localText); + if (isIgnored(intr)) return { action: "noop" }; + + if (intr.computed_hash === remote.content_hash) { + if (intr.version === remote.version_id && intr.content_hash === remote.content_hash) { + return { action: "noop" }; + } + if (treatHot) return { action: "defer" }; + return { action: "adopt", version: remote }; + } + + if (versionAtOrAhead(intr.version, remote.version_id)) { + return { action: "noop" }; + } + + if (!isDirty(intr)) { + if (treatHot) return { action: "defer" }; + return { action: "materialize", version: remote }; + } + + // Dirty + divergent: rebase local bytes onto remote current (better-sync WS2). + if (treatHot) return { action: "defer" }; + return { + action: "rebase", + prevVersionId: remote.version_id, + user: stripToUserContent(localText), + }; +} + +export type EffectResult = "noop" | "applied" | "deferred" | "parked"; + +/** + * Apply a decision from `decideInbound`. Rebase failures (`stale_prev` / + * `path_taken`) fall through to parking the remote as an ignored sibling. + */ +export async function effectInbound( + client: KernelClient, + store: FileStore, + repo: string, + path: string, + decision: InboundDecision, + opts: { + deferred?: DeferredMap; + ref?: VersionRef; + map?: RemoteMap; + nowMs?: number; + } = {}, +): Promise { + const nowMs = opts.nowMs ?? Date.now(); + + switch (decision.action) { + case "noop": + return "noop"; + case "defer": { + if (opts.deferred && opts.ref) { + enqueueDeferred(opts.deferred, path, opts.ref, nowMs); + return "deferred"; + } + // No place to hold it — treat as cold materialize of the ref's version. + return "noop"; + } + case "adopt": { + await materializeAt(store, path, decision.version, { preserveMtime: true }); + opts.map?.set(path, { + version_id: decision.version.version_id, + content_hash: decision.version.content_hash, + }); + return "applied"; + } + case "materialize": { + await materializeAt(store, path, decision.version); + opts.map?.set(path, { + version_id: decision.version.version_id, + content_hash: decision.version.content_hash, + }); + return "applied"; + } + case "rebase": { + try { + const v = await putAndAck( + client, + store, + repo, + path, + decision.prevVersionId, + decision.user, + ); + opts.map?.set(path, { version_id: v.version_id, content_hash: v.content_hash }); + return "applied"; + } catch (err) { + if ( + err instanceof KernelError && + (err.code === "stale_prev" || err.code === "path_taken") + ) { + const current = await client.docs.get_version(repo, decision.prevVersionId); + // Prefer live current at path if available (may have advanced). + let park: Version = current; + try { + park = await client.docs.get(repo, path); + } catch { + // Keep get_version result. + } + await parkIgnoredSibling(store, path, park); + return "parked"; + } + throw err; + } + } + } +} + +/** + * Retry one deferred hold: re-read local, fetch remote, decide, effect. + * Returns whether the entry should be removed from the deferred map. + */ +export async function retryDeferredEntry( + client: KernelClient, + store: FileStore, + repo: string, + path: string, + entry: DeferredEntry, + opts: { + settleMs: number; + map?: RemoteMap; + nowMs?: number; + deferred?: DeferredMap; + }, +): Promise<{ done: boolean; result: EffectResult }> { + const nowMs = opts.nowMs ?? Date.now(); + const forceCold = isDeferExpired(entry, nowMs); + const hot = forceCold ? false : await pathIsHot(store, path, opts.settleMs, nowMs); + const canDefer = opts.deferred !== undefined && !forceCold; + + if (entry.ref.op === "delete") { + const text = await store.read(path); + if (text === null) return { done: true, result: "noop" }; + const intr = readFileIntrinsics(text); + if (isIgnored(intr) || isDirty(intr)) return { done: true, result: "noop" }; + if (hot && canDefer) return { done: false, result: "deferred" }; + await store.remove(path); + opts.map?.delete(path); + return { done: true, result: "applied" }; + } + + const text = await store.read(path); + if (text === null) { + // Local gone — materialize remote current if still live. + try { + const current = await client.docs.get(repo, path); + if (hot && canDefer) return { done: false, result: "deferred" }; + await materializeAt(store, path, current); + opts.map?.set(path, { + version_id: current.version_id, + content_hash: current.content_hash, + }); + return { done: true, result: "applied" }; + } catch (err) { + if (err instanceof KernelError && err.code === "doc_not_found") { + return { done: true, result: "noop" }; + } + throw err; + } + } + + // Prefer live current at path; fall back to the held version. + let remote: Version; + try { + remote = await client.docs.get(repo, path); + } catch (err) { + if (!(err instanceof KernelError && err.code === "doc_not_found")) throw err; + try { + remote = await client.docs.get_version(repo, entry.ref.version_id); + } catch { + return { done: true, result: "noop" }; + } + } + + const decision = decideInbound({ + localText: text, + remote, + hot, + canDefer, + }); + if (decision.action === "defer") { + return { done: false, result: "deferred" }; + } + const result = await effectInbound(client, store, repo, path, decision, { + map: opts.map, + nowMs, + }); + return { done: true, result }; +} diff --git a/src/sync/push.ts b/src/sync/push.ts index 192ac37..e355b4c 100644 --- a/src/sync/push.ts +++ b/src/sync/push.ts @@ -19,19 +19,18 @@ */ import type { KernelClient } from "../client/kernel-client.js"; -import { pathIsInSystemNamespace, withVersionSuffix } from "../kernel/deletion.js"; +import { pathIsInSystemNamespace } from "../kernel/deletion.js"; import { KernelError } from "../kernel/errors.js"; import { HARDCODED_DEFAULTS } from "../kernel/path-config.js"; import type { Version } from "../kernel/wire.js"; -import { extractSystemProperties, split } from "../markdown/frontmatter.js"; import { - isDirty, - isIgnored, - readFileIntrinsics, - renderIgnoredSibling, - renderMaterialized, - stampProvenance, -} from "./intrinsics.js"; + ackLocalWrite, + parkIgnoredSibling, + putAndAck, + stripToUserContent, + type UserContent, +} from "./converge.js"; +import { isDirty, isIgnored, readFileIntrinsics, renderMaterialized } from "./intrinsics.js"; import type { FileStore } from "./reconcile.js"; /** Last-known remote state per path (§4.2 tier 3). */ @@ -55,8 +54,6 @@ export type PushDeps = { log?: (msg: string) => void; }; -type UserContent = { frontmatter_raw: string; body: string }; - type StalePrevData = { current_version_id?: string | null; current_path?: string | null; @@ -247,10 +244,7 @@ async function recoverStalePut( ) { throw rebaseErr; } - await store.write( - withVersionSuffix(path, destCurrent.version_id), - renderIgnoredSibling(destCurrent), - ); + await parkIgnoredSibling(store, path, destCurrent); log(`conflict\t${path}`); return "conflict"; } @@ -281,7 +275,7 @@ async function recoverStalePut( // Still live at another path with a newer version — local rename vs // remote edit. Park the remote current; keep local bytes. const current = await client.docs.get_version(repo, currentId); - await store.write(withVersionSuffix(path, current.version_id), renderIgnoredSibling(current)); + await parkIgnoredSibling(store, path, current); log(`conflict\t${path}`); return "conflict"; } @@ -323,37 +317,13 @@ async function commitPut( ): Promise { const { client, store, repo, map } = deps; const log = deps.log ?? (() => {}); - const v = await client.docs.put(repo, prevVersionId, path, user); - await ackLocalWrite(store, path, v, user); + const v = await putAndAck(client, store, repo, path, prevVersionId, user); dropMapEntriesForVersion(map, prevVersionId, path); map.set(path, { version_id: v.version_id, content_hash: v.content_hash }); log(`${logLabel}\t${path}`); return "updated"; } -/** - * After a successful kernel write, restamp the local file. If the editor saved - * again during the round-trip, keep those bytes and only update provenance — - * never write the snapshot we just pushed over newer typing. - */ -async function ackLocalWrite( - store: FileStore, - path: string, - v: Version, - pushed: UserContent, -): Promise { - const now = await store.read(path); - if (now === null) return; - const current = stripToUserContent(now); - if (current.body === pushed.body && current.frontmatter_raw === pushed.frontmatter_raw) { - await store.write(path, renderMaterialized(v), { preserveMtime: true }); - return; - } - await store.write(path, stampProvenance(now, v.version_id, v.content_hash), { - preserveMtime: true, - }); -} - /** The occupied-path rule (§4.4): hash match → adopt; differ → conflict park. */ async function occupiedPath( deps: PushDeps, @@ -370,7 +340,7 @@ async function occupiedPath( map.set(path, { version_id: current.version_id, content_hash: current.content_hash }); return "clean"; } - await store.write(withVersionSuffix(path, current.version_id), renderIgnoredSibling(current)); + await parkIgnoredSibling(store, path, current); log(`conflict\t${path}`); return "conflict"; } @@ -405,9 +375,3 @@ function dropMapEntriesForVersion(map: RemoteMap, versionId: string, keepPath: s function isDeletedPath(path: string): boolean { return pathIsInSystemNamespace(path, HARDCODED_DEFAULTS.system_sigils); } - -function stripToUserContent(text: string): UserContent { - const lf = text.replace(/\r\n/g, "\n"); - const { frontmatter_raw, body } = split(lf); - return { frontmatter_raw: extractSystemProperties(frontmatter_raw).raw, body }; -} diff --git a/src/sync/reconcile.ts b/src/sync/reconcile.ts index 1252379..2bdffea 100644 --- a/src/sync/reconcile.ts +++ b/src/sync/reconcile.ts @@ -14,16 +14,20 @@ */ import type { KernelClient } from "../client/kernel-client.js"; -import { withVersionSuffix } from "../kernel/deletion.js"; import { KernelError } from "../kernel/errors.js"; import type { IndexItem } from "../kernel/wire.js"; -import { extractSystemProperties, split } from "../markdown/frontmatter.js"; +import { + materializeAt, + parkIgnoredSibling, + putAndAck, + stripToUserContent, +} from "./converge.js"; +import { decideInbound, effectInbound } from "./hot-path.js"; import { type FileIntrinsics, isDirty, isIgnored, readFileIntrinsics, - renderIgnoredSibling, renderMaterialized, } from "./intrinsics.js"; import type { ScopeFilter } from "./paths.js"; @@ -43,6 +47,12 @@ export type FileStore = { write(docPath: string, text: string, opts?: { preserveMtime?: boolean }): Promise; /** Remove the file at a doc path (no-op if already absent). */ remove(docPath: string): Promise; + /** + * File mtime in epoch milliseconds, or null if absent. Used by the settle / + * hot-path gate (better-sync.plan); provenance-preserving writes must not + * advance this value. + */ + mtime(docPath: string): Promise; }; /** One reconciliation action, for reporting + dry-run (§4.1 `--dry-run`). */ @@ -53,6 +63,7 @@ export type SyncAction = { | "adopt" // metadata repair: inject remote provenance into a clean local copy | "materialize" // write remote current locally (new or fast-forward) | "push" // local edit → docs.put / docs.create + | "rebase" // dirty local put onto advanced remote current (better-sync) | "conflict" // park remote as ignored sibling; local bytes preserved | "delete-local" // remote-deleted, local clean → remove | "resurrect" // remote-deleted, local dirty → push as create @@ -181,10 +192,12 @@ async function resolveLocalPath( return { path, verdict: "adopt", detail: "content matches remote; provenance repaired" }; } if (!intr.version) { - // No embedded version and bytes differ from remote → occupied-path - // conflict (row 6). Never stamp divergent local bytes with remote id. - if (!dryRun) await parkConflict(client, store, opts.repo, path, remote.version_id); - return { path, verdict: "conflict", detail: "occupied path, no local provenance" }; + // No embedded version and bytes differ → rebase onto remote, else park + // (better-sync WS2; formerly unconditional §4.9 row 6 park). + if (dryRun) { + return { path, verdict: "rebase", detail: "occupied path, no local provenance" }; + } + return convergeOccupied(client, store, opts.repo, path); } if (intr.version === remote.version_id) { // Embedded version is the remote current. @@ -199,9 +212,11 @@ async function resolveLocalPath( if (!dryRun) await materializeVersion(client, store, opts.repo, path, remote.version_id); return { path, verdict: "materialize", detail: "fast-forward to remote current" }; } - // Local edited AND remote advanced → conflict (row 5). - if (!dryRun) await parkConflict(client, store, opts.repo, path, remote.version_id); - return { path, verdict: "conflict", detail: "local edit vs advanced remote" }; + // Local edited AND remote advanced → rebase, else park (better-sync WS2). + if (dryRun) { + return { path, verdict: "rebase", detail: "local edit vs advanced remote" }; + } + return convergeOccupied(client, store, opts.repo, path); } // No remote doc at this path. @@ -253,14 +268,6 @@ async function readUserContent( return stripToUserContent(text); } -/** Split a file into stored-shape fields, dropping all `$*` intrinsic lines. */ -function stripToUserContent(text: string): { frontmatter_raw: string; body: string } { - const lf = text.replace(/\r\n/g, "\n"); - // Reuse the same split/strip path the hash uses so bytes are canonical. - const { frontmatter_raw, body } = split(lf); - return { frontmatter_raw: extractSystemProperties(frontmatter_raw).raw, body }; -} - async function materializeVersion( client: KernelClient, store: FileStore, @@ -269,7 +276,7 @@ async function materializeVersion( versionId: string, ): Promise { const v = await client.docs.get_version(repo, versionId); - await store.write(path, renderMaterialized(v)); + await materializeAt(store, path, v); } /** Repair a clean local file's embedded provenance to the given version. */ @@ -280,11 +287,8 @@ async function materializeInPlace( path: string, versionId: string, ): Promise { - // The content already equals the version; re-render from the authoritative - // remote version so intrinsics are exact. Keep mtime so a metadata-only - // stamp cannot win an Obsidian/iCloud race against unsynced typing. const v = await client.docs.get_version(repo, versionId); - await store.write(path, renderMaterialized(v), { preserveMtime: true }); + await materializeAt(store, path, v, { preserveMtime: true }); } async function pushEdit( @@ -294,9 +298,8 @@ async function pushEdit( path: string, prevVersionId: string, ): Promise { - const { frontmatter_raw, body } = await readUserContent(store, path); - const v = await client.docs.put(repo, prevVersionId, path, { frontmatter_raw, body }); - await store.write(path, renderMaterialized(v), { preserveMtime: true }); + const user = await readUserContent(store, path); + await putAndAck(client, store, repo, path, prevVersionId, user); } async function pushMove( @@ -306,10 +309,9 @@ async function pushMove( path: string, prevVersionId: string, ): Promise { - const { frontmatter_raw, body } = await readUserContent(store, path); + const user = await readUserContent(store, path); // A put whose path differs from prev's path is a move preserving identity. - const v = await client.docs.put(repo, prevVersionId, path, { frontmatter_raw, body }); - await store.write(path, renderMaterialized(v), { preserveMtime: true }); + await putAndAck(client, store, repo, path, prevVersionId, user); } async function pushCreate( @@ -318,16 +320,14 @@ async function pushCreate( repo: string, path: string, ): Promise { - const { frontmatter_raw, body } = await readUserContent(store, path); + const user = await readUserContent(store, path); try { - const v = await client.docs.create(repo, path, { frontmatter_raw, body }); + const v = await client.docs.create(repo, path, user); await store.write(path, renderMaterialized(v), { preserveMtime: true }); } catch (err) { if (err instanceof KernelError && err.code === "create_conflict") { - // A doc appeared at this path since the index scan; treat the occupied - // path as a conflict rather than clobbering it. - const current = (err.data as { current_version_id?: string }).current_version_id; - if (current) await parkConflict(client, store, repo, path, current); + // A doc appeared at this path since the index scan; rebase or park. + await convergeOccupied(client, store, repo, path); return; } throw err; @@ -335,21 +335,37 @@ async function pushCreate( } /** - * Park a conflict (§4.8): keep the local file's bytes and path untouched; - * materialize the remote current beside it as `-.md` with - * `$sync: ignore`. Collision-safe (version ids are unique), so replay never - * sprays duplicates. + * Rebase local bytes onto the live remote current at `path`, else park a + * sibling (better-sync WS2). Used for §4.9 rows 5–6 and create races. */ -async function parkConflict( +async function convergeOccupied( client: KernelClient, store: FileStore, repo: string, path: string, - remoteVersionId: string, -): Promise { - const v = await client.docs.get_version(repo, remoteVersionId); - const siblingPath = withVersionSuffix(path, remoteVersionId); - await store.write(siblingPath, renderIgnoredSibling(v)); +): Promise { + const text = (await store.read(path)) ?? ""; + const remote = await client.docs.get(repo, path); + const decision = decideInbound({ + localText: text, + remote, + hot: false, + canDefer: false, + }); + const result = await effectInbound(client, store, repo, path, decision); + if (result === "parked" || decision.action === "noop") { + // noop here would mean ignored; occupied diverge shouldn't noop. + if (result === "parked") { + return { path, verdict: "conflict", detail: "rebase failed; parked remote sibling" }; + } + } + if (decision.action === "rebase" || result === "applied") { + return { path, verdict: "rebase", detail: "local bytes put onto remote current" }; + } + if (decision.action === "adopt") { + return { path, verdict: "adopt", detail: "content matches remote; provenance repaired" }; + } + return { path, verdict: "conflict", detail: "could not converge occupied path" }; } /** Conflict park keyed off the doc's *current* remote version (move race). */ @@ -360,11 +376,7 @@ async function parkConflictForCurrent( path: string, embeddedVersionId: string, ): Promise { - // Resolve the document's current version from the embedded (now-superseded) - // one, then park it. get_version gives us the doc; docs.get by its path finds - // the live current. const superseded = await client.docs.get_version(repo, embeddedVersionId); const current = await client.docs.get(repo, superseded.path); - const siblingPath = withVersionSuffix(path, current.version_id); - await store.write(siblingPath, renderIgnoredSibling(current)); + await parkIgnoredSibling(store, path, current); } diff --git a/test/sync-feed.test.ts b/test/sync-feed.test.ts index 217b62d..0c2f0f6 100644 --- a/test/sync-feed.test.ts +++ b/test/sync-feed.test.ts @@ -15,10 +15,15 @@ import type { FileStore } from "../src/sync/reconcile.js"; let client: KernelClient; -function memStore(initial?: Record): FileStore & { files: Map } { +function memStore(initial?: Record): FileStore & { + files: Map; + mtimes: Map; +} { const files = new Map(Object.entries(initial ?? {})); + const mtimes = new Map(); return { files, + mtimes, async list() { return [...files.keys()]; }, @@ -30,6 +35,11 @@ function memStore(initial?: Record): FileStore & { files: Map { expect(store.files.get("new.md")).toContain("x"); }); - it("parks a conflict when the move destination holds divergent dirty bytes", async () => { + it("rebases a dirty destination onto the moved remote instead of parking", async () => { const v1 = await client.docs.create("notes", "old.md", { body: "x\n", frontmatter_raw: "" }); const store = memStore({ "old.md": materialized(await client.docs.get_version("notes", v1.version_id)), // A dirty file already occupies the destination (no provenance). "new.md": "my local dest work\n", }); - const v2 = await client.docs.put("notes", v1.version_id, "new.md", { + await client.docs.put("notes", v1.version_id, "new.md", { body: "moved body\n", frontmatter_raw: "", }); await applyFeed(client, store, scope, { repo: "notes", since: v1.version_id }); - // Local destination bytes preserved; remote parked as an ignored sibling. - expect(store.files.get("new.md")).toBe("my local dest work\n"); - expect(store.files.get(`new-${v2.version_id}.md`)).toContain("moved body"); - expect(store.files.get(`new-${v2.version_id}.md`)).toContain("$sync: ignore"); + // Local destination bytes become the new head; no ignored sibling. + expect(store.files.get("new.md")).toContain("my local dest work"); + expect([...store.files.keys()].some((k) => /-v\d+\.md$/.test(k))).toBe(false); + const remote = await client.docs.get("notes", "new.md"); + expect(remote.body).toBe("my local dest work\n"); }); }); @@ -142,21 +153,22 @@ describe("applyFeed — delete", () => { }); describe("applyFeed — conflict + ignore", () => { - it("parks an incoming version beside a dirty local file", async () => { + it("rebases a dirty local file onto a newer incoming version (no sibling)", async () => { const v1 = await client.docs.create("notes", "a.md", { body: "base\n", frontmatter_raw: "" }); const dirty = materialized(await client.docs.get_version("notes", v1.version_id)).replace( "base", "my local edit", ); const store = memStore({ "a.md": dirty }); - const v2 = await client.docs.put("notes", v1.version_id, "a.md", { + await client.docs.put("notes", v1.version_id, "a.md", { body: "remote edit\n", frontmatter_raw: "", }); await applyFeed(client, store, scope, { repo: "notes", since: v1.version_id }); - expect(store.files.get("a.md")).toBe(dirty); - expect(store.files.get(`a-${v2.version_id}.md`)).toContain("remote edit"); - expect(store.files.get(`a-${v2.version_id}.md`)).toContain("$sync: ignore"); + expect(store.files.get("a.md")).toContain("my local edit"); + expect([...store.files.keys()].some((k) => /-v\d+\.md$/.test(k))).toBe(false); + const remote = await client.docs.get("notes", "a.md"); + expect(remote.body).toBe("my local edit\n"); }); it("does not park or clobber when local is dirty on the same $version (push echo)", async () => { @@ -231,3 +243,57 @@ describe("applyFeed — idempotent replay (§4.3)", () => { expect(replay.cursor).toBe(first.cursor); }); }); + +describe("applyFeed — settle / defer (better-sync)", () => { + it("defers inbound rebase on a hot file and still advances the cursor", async () => { + const v1 = await client.docs.create("notes", "a.md", { body: "base\n", frontmatter_raw: "" }); + const dirty = materialized(await client.docs.get_version("notes", v1.version_id)).replace( + "base", + "my local edit", + ); + const store = memStore({ "a.md": dirty }); + store.mtimes.set("a.md", 10_000); + const v2 = await client.docs.put("notes", v1.version_id, "a.md", { + body: "remote edit\n", + frontmatter_raw: "", + }); + const deferred = new Map(); + const result = await applyFeed(client, store, scope, { + repo: "notes", + since: v1.version_id, + settleMs: 5_000, + deferred, + nowMs: () => 12_000, + }); + expect(store.files.get("a.md")).toBe(dirty); + expect([...store.files.keys()].some((k) => /-v\d+\.md$/.test(k))).toBe(false); + expect(deferred.has("a.md")).toBe(true); + expect(deferred.get("a.md")?.ref.version_id).toBe(v2.version_id); + expect(result.deferred).toBeGreaterThan(0); + expect(result.cursor).not.toBe(v1.version_id); + }); + + it("does not defer when settleMs is 0 even if mtime is fresh", async () => { + const v1 = await client.docs.create("notes", "a.md", { body: "base\n", frontmatter_raw: "" }); + const dirty = materialized(await client.docs.get_version("notes", v1.version_id)).replace( + "base", + "my local edit", + ); + const store = memStore({ "a.md": dirty }); + store.mtimes.set("a.md", Date.now()); + await client.docs.put("notes", v1.version_id, "a.md", { + body: "remote edit\n", + frontmatter_raw: "", + }); + const deferred = new Map(); + await applyFeed(client, store, scope, { + repo: "notes", + since: v1.version_id, + settleMs: 0, + deferred, + }); + expect(deferred.size).toBe(0); + expect(store.files.get("a.md")).toContain("my local edit"); + expect(await client.docs.get("notes", "a.md")).toMatchObject({ body: "my local edit\n" }); + }); +}); diff --git a/test/sync-hot-path.test.ts b/test/sync-hot-path.test.ts new file mode 100644 index 0000000..3902acd --- /dev/null +++ b/test/sync-hot-path.test.ts @@ -0,0 +1,186 @@ +/** + * Hot-path gate and inbound decision table (better-sync.plan). + */ + +import { describe, expect, it } from "vitest"; +import { contentHashOfFile } from "../src/markdown/content-hash.js"; +import { + decideInbound, + enqueueDeferred, + isDeferExpired, + isPathHot, + versionAtOrAhead, +} from "../src/sync/hot-path.js"; +import type { Version, VersionRef } from "../src/kernel/wire.js"; + +function version(partial: Partial & { version_id: string; content_hash: string }): Version { + return { + prev_version_id: null, + next_version_id: null, + repo: "notes", + path: "a.md", + frontmatter: {}, + frontmatter_raw: "", + body: "x\n", + author: "t", + created_at: "2020-01-01T00:00:00.000Z", + ...partial, + }; +} + +describe("isPathHot", () => { + it("is never hot when settleMs is 0", () => { + expect(isPathHot(Date.now(), 0, Date.now())).toBe(false); + }); + + it("is never hot when the file is absent", () => { + expect(isPathHot(null, 5000, Date.now())).toBe(false); + }); + + it("is hot when mtime is within the settle window", () => { + expect(isPathHot(1000, 500, 1400)).toBe(true); + }); + + it("is cold when mtime is at or older than settleMs", () => { + expect(isPathHot(1000, 500, 1500)).toBe(false); + expect(isPathHot(1000, 500, 2000)).toBe(false); + }); +}); + +describe("versionAtOrAhead", () => { + it("treats missing local version as behind", () => { + expect(versionAtOrAhead(undefined, "v2")).toBe(false); + }); + + it("treats equal and later ids as ahead", () => { + expect(versionAtOrAhead("v3", "v3")).toBe(true); + expect(versionAtOrAhead("v4", "v3")).toBe(true); + expect(versionAtOrAhead("v2", "v3")).toBe(false); + }); +}); + +describe("decideInbound", () => { + const remote = version({ version_id: "v2", content_hash: "abc", body: "remote\n" }); + + it("noops an ignored file", () => { + const d = decideInbound({ + localText: "---\n$sync: ignore\n---\nwhatever\n", + remote, + hot: true, + canDefer: true, + }); + expect(d.action).toBe("noop"); + }); + + it("noops when local already names the remote version and hashes match", () => { + const d = decideInbound({ + localText: "---\n$version: v2\n$content_hash: abc\n---\nremote\n", + remote, + hot: false, + canDefer: false, + }); + expect(d.action).toBe("noop"); + }); + + it("adopts when hashes match but provenance lags (cold)", () => { + const body = "same bytes\n"; + const local = "---\n$version: v1\n$content_hash: not-the-real-hash\n---\n" + body; + const hash = contentHashOfFile(local); + const d = decideInbound({ + localText: local, + remote: version({ version_id: "v2", content_hash: hash, body }), + hot: false, + canDefer: true, + }); + expect(d.action).toBe("adopt"); + }); + + it("defers dirty+divergent when hot and canDefer", () => { + const d = decideInbound({ + localText: "---\n$version: v1\n$content_hash: deadbeef\n---\nlocal edit\n", + remote, + hot: true, + canDefer: true, + }); + expect(d.action).toBe("defer"); + }); + + it("rebases dirty+divergent when cold", () => { + const d = decideInbound({ + localText: "---\n$version: v1\n$content_hash: deadbeef\n---\nlocal edit\n", + remote, + hot: false, + canDefer: true, + }); + expect(d.action).toBe("rebase"); + if (d.action === "rebase") expect(d.prevVersionId).toBe("v2"); + }); + + it("does not defer when canDefer is false even if hot", () => { + const d = decideInbound({ + localText: "---\n$version: v1\n$content_hash: deadbeef\n---\nlocal edit\n", + remote, + hot: true, + canDefer: false, + }); + expect(d.action).toBe("rebase"); + }); +}); + +describe("enqueueDeferred", () => { + it("keeps the newer version_id and original since", () => { + const map = new Map(); + const older: VersionRef = { + version_id: "v2", + prev_version_id: "v1", + repo: "notes", + path: "a.md", + prev_path: "a.md", + content_hash: "x", + op: "update", + created_at: "", + }; + const newer: VersionRef = { ...older, version_id: "v5" }; + enqueueDeferred(map, "a.md", older, 1000); + enqueueDeferred(map, "a.md", newer, 2000); + expect(map.get("a.md")?.ref.version_id).toBe("v5"); + expect(map.get("a.md")?.since).toBe(1000); + }); + + it("does not replace with an older version_id", () => { + const map = new Map(); + const newer: VersionRef = { + version_id: "v5", + prev_version_id: "v4", + repo: "notes", + path: "a.md", + prev_path: "a.md", + content_hash: "x", + op: "update", + created_at: "", + }; + enqueueDeferred(map, "a.md", newer, 1000); + enqueueDeferred(map, "a.md", { ...newer, version_id: "v3" }, 2000); + expect(map.get("a.md")?.ref.version_id).toBe("v5"); + }); +}); + +describe("isDeferExpired", () => { + it("expires after the TTL", () => { + const entry = { + ref: { + version_id: "v1", + prev_version_id: null, + repo: "notes", + path: "a.md", + prev_path: null, + content_hash: "x", + op: "update" as const, + created_at: "", + }, + since: 0, + }; + expect(isDeferExpired(entry, 1000, 5000)).toBe(false); + expect(isDeferExpired(entry, 5000, 5000)).toBe(true); + }); +}); diff --git a/test/sync-push.test.ts b/test/sync-push.test.ts index 4715554..8f32470 100644 --- a/test/sync-push.test.ts +++ b/test/sync-push.test.ts @@ -14,10 +14,15 @@ import type { FileStore } from "../src/sync/reconcile.js"; let client: KernelClient; -function memStore(initial?: Record): FileStore & { files: Map } { +function memStore(initial?: Record): FileStore & { + files: Map; + mtimes: Map; +} { const files = new Map(Object.entries(initial ?? {})); + const mtimes = new Map(); return { files, + mtimes, async list() { return [...files.keys()]; }, @@ -29,6 +34,11 @@ function memStore(initial?: Record): FileStore & { files: Map): FileStore & { files: Map } { +function memStore(initial?: Record): FileStore & { + files: Map; + mtimes: Map; +} { const files = new Map(Object.entries(initial ?? {})); + const mtimes = new Map(); return { files, + mtimes, async list() { return [...files.keys()]; }, @@ -30,6 +35,11 @@ function memStore(initial?: Record): FileStore & { files: Map { expect(store.files.get("a.md")).toContain("v2"); }); - it("row 5: local edit AND remote advanced → conflict (park sibling, keep local)", async () => { + it("row 5: local edit AND remote advanced → rebase (local becomes new head)", async () => { const v1 = await client.docs.create("notes", "a.md", { body: "base\n", frontmatter_raw: "" }); const injected = renderInjected(await client.docs.get("notes", "a.md")); const localEdited = injected.replace("base", "my local edit"); const store = memStore({ "a.md": localEdited }); - // Remote advances to v2. - const v2 = await client.docs.put("notes", v1.version_id, "a.md", { + await client.docs.put("notes", v1.version_id, "a.md", { body: "their remote edit\n", frontmatter_raw: "", }); const report = await reconcile(store); - expect(verdictFor(report, "a.md")).toBe("conflict"); - // Local bytes preserved. - expect(store.files.get("a.md")).toBe(localEdited); - // Remote current parked as an ignored sibling. - const sibling = store.files.get(`a-${v2.version_id}.md`) ?? ""; - expect(sibling).toContain("their remote edit"); - expect(sibling).toContain("$sync: ignore"); + expect(verdictFor(report, "a.md")).toBe("rebase"); + expect(store.files.get("a.md")).toContain("my local edit"); + expect([...store.files.keys()].some((k) => /-v\d+\.md$/.test(k))).toBe(false); + const remote = await client.docs.get("notes", "a.md"); + expect(remote.body).toBe("my local edit\n"); }); - it("row 6: occupied path, no local provenance, bytes differ → conflict", async () => { - const v = await client.docs.create("notes", "a.md", { + it("row 6: occupied path, no local provenance, bytes differ → rebase", async () => { + await client.docs.create("notes", "a.md", { body: "remote body\n", frontmatter_raw: "", }); - // Local file at same path, different bytes, NO provenance. const store = memStore({ "a.md": "totally different local\n" }); const report = await reconcile(store); - expect(verdictFor(report, "a.md")).toBe("conflict"); - expect(store.files.get("a.md")).toBe("totally different local\n"); - expect(store.files.get(`a-${v.version_id}.md`)).toContain("remote body"); + expect(verdictFor(report, "a.md")).toBe("rebase"); + expect(store.files.get("a.md")).toContain("totally different local"); + expect([...store.files.keys()].some((k) => /-v\d+\.md$/.test(k))).toBe(false); + const remote = await client.docs.get("notes", "a.md"); + expect(remote.body).toBe("totally different local\n"); }); it("remote-deleted, local clean → delete-local", async () => { diff --git a/test/sync-scenarios.test.ts b/test/sync-scenarios.test.ts index 30b4a26..ab49931 100644 --- a/test/sync-scenarios.test.ts +++ b/test/sync-scenarios.test.ts @@ -72,6 +72,10 @@ async function run(steps: Step[]) { async remove(p) { files.delete(p); }, + async mtime(p) { + if (!files.has(p)) return null; + return 0; + }, }; const wrapped: KernelClient = { @@ -344,7 +348,7 @@ describe("sync timelines", () => { expect(await v.remote("note.md")).toBe("idea\n"); }); - it("two-writer: feed of newer remote vs dirty local parks sibling, keeps local", async () => { + it("two-writer: feed of newer remote vs dirty local rebases; local wins, no sibling", async () => { const v = await run([ ["write", "a.md", "base\n"], ["push"], @@ -353,6 +357,7 @@ describe("sync timelines", () => { ["feed"], ]); expect(v.bodies["a.md"]).toBe("my local edit\n"); - expect(v.ignored.length).toBe(1); + expect(v.ignored.length).toBe(0); + expect(await v.remote("a.md")).toBe("my local edit\n"); }); }); From 93ec11e83d5a3b3efbe93b3108c049fe3dfdb5ae Mon Sep 17 00:00:00 2001 From: Brendan Baldwin Date: Mon, 31 Aug 2026 00:19:37 -0600 Subject: [PATCH 3/4] Fix review feedback on reconcile verdicts and error handling. Return accurate SyncActions from pushCreate, map noop convergence to clean, narrow deferred/rebase catch blocks to doc_not_found, and streamline the hash-equal feed adopt path. Co-authored-by: Cursor --- src/sync/feed.ts | 21 ++++++--------------- src/sync/hot-path.ts | 11 +++++++---- src/sync/reconcile.ts | 32 +++++++++++++++++++++++--------- 3 files changed, 36 insertions(+), 28 deletions(-) diff --git a/src/sync/feed.ts b/src/sync/feed.ts index f351b48..5db9669 100644 --- a/src/sync/feed.ts +++ b/src/sync/feed.ts @@ -150,27 +150,18 @@ async function applyRef( const intr = readFileIntrinsics(existing); if (isIgnored(intr)) return "noop"; - // Fast path: bytes already match — may still need provenance repair. + // Fast path: bytes already match — provenance repair or defer when hot. if (intr.computed_hash === ref.content_hash) { opts.map?.set(ref.path, { version_id: ref.version_id, content_hash: ref.content_hash }); if (intr.version === ref.version_id) return "noop"; if (versionAtOrAhead(intr.version, ref.version_id)) return "noop"; const hot = await pathIsHot(store, ref.path, settleMs, nowMs); + if (hot && canDefer) { + enqueueDeferred(opts.deferred!, ref.path, ref, nowMs); + return "deferred"; + } const v = await client.docs.get_version(opts.repo, ref.version_id); - const decision = decideInbound({ - localText: existing, - remote: v, - hot, - canDefer, - }); - const result = await effectInbound(client, store, opts.repo, ref.path, decision, { - deferred: opts.deferred, - ref, - map: opts.map, - nowMs, - }); - if (result === "deferred") return "deferred"; - if (result === "noop") return "noop"; + await materializeAt(store, ref.path, v, { preserveMtime: true }); log(`feed adopt\t${ref.path}`); return "applied"; } diff --git a/src/sync/hot-path.ts b/src/sync/hot-path.ts index 6f98f25..a47760b 100644 --- a/src/sync/hot-path.ts +++ b/src/sync/hot-path.ts @@ -209,8 +209,8 @@ export async function effectInbound( let park: Version = current; try { park = await client.docs.get(repo, path); - } catch { - // Keep get_version result. + } catch (err) { + if (!(err instanceof KernelError && err.code === "doc_not_found")) throw err; } await parkIgnoredSibling(store, path, park); return "parked"; @@ -282,8 +282,11 @@ export async function retryDeferredEntry( if (!(err instanceof KernelError && err.code === "doc_not_found")) throw err; try { remote = await client.docs.get_version(repo, entry.ref.version_id); - } catch { - return { done: true, result: "noop" }; + } catch (err) { + if (err instanceof KernelError && err.code === "doc_not_found") { + return { done: true, result: "noop" }; + } + throw err; } } diff --git a/src/sync/reconcile.ts b/src/sync/reconcile.ts index 2bdffea..8fa1bf6 100644 --- a/src/sync/reconcile.ts +++ b/src/sync/reconcile.ts @@ -243,7 +243,7 @@ async function resolveLocalPath( if (!intr.version) { // No provenance, path absent remotely → a genuine local creation (row 7). - if (!dryRun) await pushCreate(client, store, opts.repo, path); + if (!dryRun) return await pushCreate(client, store, opts.repo, path); return { path, verdict: "push", detail: "local creation" }; } @@ -254,7 +254,16 @@ async function resolveLocalPath( if (!dryRun) await store.remove(path); return { path, verdict: "delete-local", detail: "remote-deleted, local clean" }; } - if (!dryRun) await pushCreate(client, store, opts.repo, path); + if (!dryRun) { + return await pushCreate( + client, + store, + opts.repo, + path, + "resurrect", + "remote-deleted, local dirty", + ); + } return { path, verdict: "resurrect", detail: "remote-deleted, local dirty" }; } @@ -319,16 +328,18 @@ async function pushCreate( store: FileStore, repo: string, path: string, -): Promise { + successVerdict: SyncAction["verdict"] = "push", + successDetail = "local creation", +): Promise { const user = await readUserContent(store, path); try { const v = await client.docs.create(repo, path, user); await store.write(path, renderMaterialized(v), { preserveMtime: true }); + return { path, verdict: successVerdict, detail: successDetail }; } catch (err) { if (err instanceof KernelError && err.code === "create_conflict") { // A doc appeared at this path since the index scan; rebase or park. - await convergeOccupied(client, store, repo, path); - return; + return await convergeOccupied(client, store, repo, path); } throw err; } @@ -353,11 +364,14 @@ async function convergeOccupied( canDefer: false, }); const result = await effectInbound(client, store, repo, path, decision); - if (result === "parked" || decision.action === "noop") { - // noop here would mean ignored; occupied diverge shouldn't noop. - if (result === "parked") { - return { path, verdict: "conflict", detail: "rebase failed; parked remote sibling" }; + if (result === "noop" || decision.action === "noop") { + if (isIgnored(readFileIntrinsics(text))) { + return { path, verdict: "ignored" }; } + return { path, verdict: "clean", detail: "already converged" }; + } + if (result === "parked") { + return { path, verdict: "conflict", detail: "rebase failed; parked remote sibling" }; } if (decision.action === "rebase" || result === "applied") { return { path, verdict: "rebase", detail: "local bytes put onto remote current" }; From 8a26e4d3f18cd3a7c511c2864156a2290aa61036 Mon Sep 17 00:00:00 2001 From: Brendan Baldwin Date: Mon, 31 Aug 2026 00:31:44 -0600 Subject: [PATCH 4/4] Give watcher push daemon test 20s waitFor headroom under load. Matches the rename case so chokidar/debounce latency in a busy vitest worker does not flake at the default 8s deadline. Co-authored-by: Cursor --- test/sync-daemon.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/test/sync-daemon.test.ts b/test/sync-daemon.test.ts index e8a5fbe..1a6011d 100644 --- a/test/sync-daemon.test.ts +++ b/test/sync-daemon.test.ts @@ -82,19 +82,21 @@ describe("sync daemon", () => { it("pushes a new local file to the remote via the watcher", async () => { daemon = await start(); writeFileSync(join(vault, "local.md"), "brand new\n"); + // chokidar + debounce can lag under a loaded vitest worker; same headroom + // as the rename case below. const v = await waitFor(async () => { try { return await client.docs.get("notes", "local.md"); } catch { return undefined; } - }); + }, 20_000); expect(v.body).toBe("brand new\n"); // The local file gets its provenance rewritten by the ack. const text = await waitFor(() => { const t = readFileSync(join(vault, "local.md"), "utf8"); return t.includes("$version") ? t : undefined; - }); + }, 20_000); expect(text).toMatch(/\$version: v\d+/); });