diff --git a/src/storage/errors.ts b/src/storage/errors.ts index 74858e0..f87b5a6 100644 --- a/src/storage/errors.ts +++ b/src/storage/errors.ts @@ -17,5 +17,8 @@ export function statusForHttp(code: number): ResultStatus { if (code === 403 || code === 401) { return "auth"; } + if (code === 412) { + return "conflict"; + } return "server"; } diff --git a/src/storage/storage.itest.ts b/src/storage/storage.itest.ts index a860461..ade9a54 100644 --- a/src/storage/storage.itest.ts +++ b/src/storage/storage.itest.ts @@ -38,6 +38,60 @@ test("putObject then getObject round trips the same bytes", async () => { assert.deepEqual(getResult.body, body); }); +test("getObject returns the object's etag for a later conditional put", async () => { + const client = createS3Client(liveSettings, SECRET_ACCESS_KEY); + await client.putObject("etag-test/note.md", new TextEncoder().encode("v1")); + + const getResult = await client.getObject("etag-test/note.md"); + assert.equal(getResult.ok, true); + assert.notEqual(getResult.etag, null); +}); + +test("putObject ifAbsent creates a missing key but is rejected once the key exists", async () => { + const client = createS3Client(liveSettings, SECRET_ACCESS_KEY); + const key = `conditional-test/absent-${Date.now()}.md`; + + const first = await client.putObject(key, new TextEncoder().encode("v1"), { kind: "ifAbsent" }); + assert.equal(first.ok, true); + + const second = await client.putObject(key, new TextEncoder().encode("v2"), { kind: "ifAbsent" }); + assert.equal(second.ok, false); + assert.equal(second.status, "conflict"); + + await client.deleteObject(key); +}); + +test("putObject ifMatch succeeds with the current etag and is rejected once it goes stale", async () => { + const client = createS3Client(liveSettings, SECRET_ACCESS_KEY); + const key = "conditional-test/match.md"; + try { + await client.putObject(key, new TextEncoder().encode("v1")); + + const getResult = await client.getObject(key); + assert.equal(getResult.ok, true); + assert.ok(getResult.etag !== null); + const etag = getResult.etag; + + const fresh = await client.putObject(key, new TextEncoder().encode("v2 longer"), { + kind: "ifMatch", + etag, + }); + assert.equal(fresh.ok, true); + + // The etag read before the v2 write is now stale, exactly a concurrent writer's position. + const stale = await client.putObject(key, new TextEncoder().encode("v3"), { + kind: "ifMatch", + etag, + }); + assert.equal(stale.ok, false); + assert.equal(stale.status, "conflict"); + } finally { + // The key is fixed, so a mid test assertion failure must not leave a leftover object that + // changes the next run's behaviour. + await client.deleteObject(key); + } +}); + test("getObject on a missing key fails without a body", async () => { const client = createS3Client(liveSettings, SECRET_ACCESS_KEY); const result = await client.getObject("does/not/exist.md"); diff --git a/src/storage/storage.ts b/src/storage/storage.ts index 96fa4aa..9df8bc6 100644 --- a/src/storage/storage.ts +++ b/src/storage/storage.ts @@ -20,12 +20,15 @@ export type DeleteResult = { message: string; }; -// GetResult reports whether an object was read. Body is null when ok is false. +// GetResult reports whether an object was read. Body is null when ok is false. Etag is the +// object's ETag exactly as the server sent it (quotes included, opaque to us), for handing back +// in a later conditional put; null when ok is false or the server sent none. export type GetResult = { ok: boolean; status: ResultStatus; message: string; body: Uint8Array | null; + etag: string | null; }; // ListResult reports whether a bucket listing succeeded. Objects is empty when ok is false. @@ -43,6 +46,12 @@ export type ObjectMeta = { lastModified: string; }; +// PutCondition makes a put conditional: "ifMatch" succeeds only while the object's ETag still +// equals etag, "ifAbsent" only while no object exists at the key. A failed precondition comes +// back as a "conflict" status, how a caller detects a concurrent writer instead of silently +// overwriting what that writer just stored. +export type PutCondition = { kind: "ifMatch"; etag: string } | { kind: "ifAbsent" }; + // PutResult reports whether an object was written. Message is the empty string when ok is true. export type PutResult = { ok: boolean; @@ -51,14 +60,15 @@ export type PutResult = { }; // ResultStatus classifies the outcome of a storage operation so callers can distinguish absent -// objects from transient failures without parsing the message string. -export type ResultStatus = "ok" | "not_found" | "auth" | "server" | "network"; +// objects and failed put preconditions from transient failures without parsing the message +// string. +export type ResultStatus = "ok" | "not_found" | "conflict" | "auth" | "server" | "network"; // StorageClient reads, writes, deletes, and lists objects in a bucket. Every method takes and // returns plain data, never provider credentials or settings, so a future WebDAV or Dropbox // client can satisfy this same shape without changing anything that depends on it. export type StorageClient = { - putObject: (key: string, body: Uint8Array) => Promise; + putObject: (key: string, body: Uint8Array, condition?: PutCondition) => Promise; getObject: (key: string) => Promise; deleteObject: (key: string) => Promise; listObjects: (prefix?: string) => Promise; @@ -75,7 +85,7 @@ export function createS3Client(settings: GeodeSettings, secretAccessKey: string) const baseUrl = `${endpointFor(settings)}/${settings.bucket}`; return { - putObject: (key, body) => s3PutObject(client, baseUrl, key, body), + putObject: (key, body, condition) => s3PutObject(client, baseUrl, key, body, condition), getObject: (key) => s3GetObject(client, baseUrl, key), deleteObject: (key) => s3DeleteObject(client, baseUrl, key), listObjects: (prefix) => s3ListObjects(client, baseUrl, prefix), @@ -118,6 +128,19 @@ export async function testConnection( return { ok: true, status: "ok", message: "" }; } +// conditionHeaders converts a PutCondition into the HTTP precondition headers an S3 compatible +// server evaluates before accepting a write. +function conditionHeaders(condition: PutCondition | undefined): Record { + if (condition === undefined) { + return {}; + } + if (condition.kind === "ifAbsent") { + return { "If-None-Match": "*" }; + } + + return { "If-Match": condition.etag }; +} + // missingFieldFor returns the name of the first field testConnection needs but doesn't have, or // "" if everything required is present. function missingFieldFor(settings: GeodeSettings, secretAccessKey: string): string { @@ -162,7 +185,7 @@ async function s3GetObject(client: AwsClient, baseUrl: string, key: string): Pro try { response = await client.fetch(`${baseUrl}/${encodeKey(key)}`, { method: "GET" }); } catch (err) { - return { ok: false, status: "network", message: messageFor(err), body: null }; + return { ok: false, status: "network", message: messageFor(err), body: null, etag: null }; } if (!response.ok) { @@ -171,10 +194,17 @@ async function s3GetObject(client: AwsClient, baseUrl: string, key: string): Pro status: statusForHttp(response.status), message: `Storage rejected the read (${response.status})`, body: null, + etag: null, }; } const buffer = await response.arrayBuffer(); - return { ok: true, status: "ok", message: "", body: new Uint8Array(buffer) }; + return { + ok: true, + status: "ok", + message: "", + body: new Uint8Array(buffer), + etag: response.headers.get("etag"), + }; } // s3ListObjects lists objects in the bucket, optionally restricted to a key prefix. S3 caps a @@ -221,12 +251,14 @@ async function s3ListObjects( return { ok: true, status: "ok", message: "", objects }; } -// s3PutObject writes body to key, creating or overwriting it. +// s3PutObject writes body to key, creating or overwriting it. When condition is set, the write +// only lands if its precondition still holds; a 412 from the server surfaces as "conflict". async function s3PutObject( client: AwsClient, baseUrl: string, key: string, body: Uint8Array, + condition: PutCondition | undefined, ): Promise { let response: Response; try { @@ -235,6 +267,7 @@ async function s3PutObject( response = await client.fetch(`${baseUrl}/${encodeKey(key)}`, { method: "PUT", body: body as BodyInit, + headers: conditionHeaders(condition), }); } catch (err) { return { ok: false, status: "network", message: messageFor(err) }; diff --git a/src/sync/fake.ts b/src/sync/fake.ts index b484fc5..66b3801 100644 --- a/src/sync/fake.ts +++ b/src/sync/fake.ts @@ -53,14 +53,34 @@ export function fakeReader(files: Record): Reader { }; } -// fakeStorage returns a StorageClient backed by an in-memory map of key to content. +// fakeStorage returns a StorageClient backed by an in-memory map of key to content. Etags are +// fake but real shaped: a quoted revision counter bumped on every write, so a conditional put +// detects a concurrent writer exactly the way a real ETag would. export function fakeStorage(objects: Record = {}): { storage: StorageClient; objects: Map; } { const store = new Map(Object.entries(objects)); + let revision = 0; + const etags = new Map(); + for (const key of store.keys()) { + revision++; + etags.set(key, `"v${revision}"`); + } const storage: StorageClient = { - putObject: async (key, body): Promise => { + putObject: async (key, body, condition): Promise => { + if (condition !== undefined && condition.kind === "ifAbsent" && store.has(key)) { + return { ok: false, status: "conflict", message: "Storage rejected the write (412)" }; + } + if ( + condition !== undefined && + condition.kind === "ifMatch" && + etags.get(key) !== condition.etag + ) { + return { ok: false, status: "conflict", message: "Storage rejected the write (412)" }; + } + revision++; + etags.set(key, `"v${revision}"`); store.set(key, new TextDecoder().decode(body)); return { ok: true, status: "ok", message: "" }; }, @@ -72,12 +92,25 @@ export function fakeStorage(objects: Record = {}): { status: "not_found", message: "Storage rejected the read (404)", body: null, + etag: null, }; } - return { ok: true, status: "ok", message: "", body: new TextEncoder().encode(content) }; + let etag: string | null = null; + const stored = etags.get(key); + if (stored !== undefined) { + etag = stored; + } + return { + ok: true, + status: "ok", + message: "", + body: new TextEncoder().encode(content), + etag, + }; }, deleteObject: async (key): Promise => { store.delete(key); + etags.delete(key); return { ok: true, status: "ok", message: "" }; }, listObjects: async (): Promise => { diff --git a/src/sync/sync.itest.ts b/src/sync/sync.itest.ts index 22b65da..1736d37 100644 --- a/src/sync/sync.itest.ts +++ b/src/sync/sync.itest.ts @@ -9,7 +9,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { test } from "node:test"; import { DEFAULT_SETTINGS, type GeodeSettings } from "../settings/settings.ts"; -import { createS3Client } from "../storage/storage.ts"; +import { createS3Client, type StorageClient } from "../storage/storage.ts"; import { nodeVault } from "../vault/fs.ts"; import { createObsidianLocalWriter, @@ -309,6 +309,51 @@ test("sync: a stale state.json from an older build never deletes the vault on th } }); +test("sync: two devices syncing at overlapping times never silently delete a file", async () => { + // Reproduces #83 against a real bucket. B's entire sync pass lands while A's pass sits between + // reading the manifest and uploading its own, the exact interleaving overlapping automatic + // syncs produce. Before the fix A's unconditional manifest upload clobbered B's, so B's next + // sync read from-b.md as a remote deletion and silently deleted it. + await resetRemote("eight/"); + const a = newDevice(); + const b = newDevice(); + try { + await writeLocal(a, "eight/base.md", "shared base"); + assert.equal((await sync(a)).ok, true); + assert.equal((await sync(b)).ok, true); + + await writeLocal(a, "eight/from-a.md", "a's new note"); + await writeLocal(b, "eight/from-b.md", "b's new note here"); + + let interleaved = false; + const racingStorage: StorageClient = { + ...storage, + putObject: async (key, body, condition) => { + if (key === MANIFEST_KEY && !interleaved) { + interleaved = true; + assert.equal((await sync(b)).ok, true); + } + return storage.putObject(key, body, condition); + }, + }; + const previous = await a.stateStore.read(); + const outcome = await syncOnce(previous, a.reader, a.writer, racingStorage, Date.now()); + + // A lost the race: the pass fails loudly and state.json does not advance. + assert.equal(outcome.ok, false); + + // A's next ordinary sync reconciles both devices' work; nothing was lost anywhere. + assert.equal((await sync(a)).ok, true); + assert.equal((await sync(b)).ok, true); + assert.equal(await readLocal(a, "eight/from-b.md"), "b's new note here"); + assert.equal(await readLocal(b, "eight/from-a.md"), "a's new note"); + assert.equal(await readLocal(a, "eight/base.md"), "shared base"); + assert.equal(await readLocal(b, "eight/base.md"), "shared base"); + } finally { + cleanup(a, b); + } +}); + test("sync: an edit on one device and a delete on another preserves the edit as a copy, no phantom pull failure", async () => { await resetRemote("six/"); const a = newDevice(); diff --git a/src/sync/sync.test.ts b/src/sync/sync.test.ts index e8f8814..81ea8b8 100644 --- a/src/sync/sync.test.ts +++ b/src/sync/sync.test.ts @@ -13,13 +13,28 @@ test("readRemoteManifest: a 404 is treated as an empty snapshot", async () => { assert.deepEqual(result, { ok: true, snapshot: { files: [] }, firstSync: true }); }); -test("readRemoteManifest: valid JSON is parsed into a snapshot", async () => { +test("readRemoteManifest: valid JSON is parsed into a snapshot, with the manifest's etag", async () => { const want: Snapshot = snapshot(file("a.md", "h1")); const { storage } = fakeStorage({ [MANIFEST_KEY]: JSON.stringify(want) }); const result = await readRemoteManifest(storage); - assert.deepEqual(result, { ok: true, snapshot: want, firstSync: false }); + assert.deepEqual(result, { ok: true, snapshot: want, firstSync: false, etag: '"v1"' }); +}); + +test("readRemoteManifest: a manifest without an etag is refused, not synced unsafely", async () => { + // Without an etag the manifest upload can't be conditional, and an unconditional upload is the + // concurrent clobber #83 fixed; the pass must refuse rather than proceed. + const { storage } = fakeStorage({ [MANIFEST_KEY]: JSON.stringify(empty) }); + const inner = storage.getObject; + storage.getObject = async (key) => { + const result = await inner(key); + return { ...result, etag: null }; + }; + + const result = await readRemoteManifest(storage); + + assert.deepEqual(result, { ok: false, message: "remote manifest has no etag" }); }); test("readRemoteManifest: corrupt JSON is reported as a failure, not an empty snapshot", async () => { @@ -92,9 +107,84 @@ test("readRemoteManifest: a non 404 failure is reported, never guessed at as emp status: "server", message: "Storage rejected the read (500)", body: null, + etag: null, }); const result = await readRemoteManifest(storage); assert.deepEqual(result, { ok: false, message: "Storage rejected the read (500)" }); }); + +test("syncOnce: a manifest overwritten by another device mid sync fails the pass instead of clobbering it", async () => { + // Reproduces #83. Device A (under test) and device B share a synced vault containing a.md, then + // sync at overlapping times: B's whole pass (pushing b.md and its manifest) lands while A is + // between reading the manifest and uploading its own. Before the fix A's unconditional upload + // clobbered B's manifest, so b.md read as a remote deletion on B's next sync and was silently + // deleted. A's conditional upload must instead lose the race and fail the pass. + const ancestor = snapshot(file("a.md", "h1")); + const { storage, objects } = fakeStorage({ [MANIFEST_KEY]: JSON.stringify(ancestor) }); + const bManifest = JSON.stringify(snapshot(file("a.md", "h1"), file("b.md", "h2"))); + const inner = storage.putObject; + let raced = false; + storage.putObject = async (key, body, condition) => { + if (key === MANIFEST_KEY && !raced) { + raced = true; + await inner("b.md", new TextEncoder().encode("bee")); + await inner(MANIFEST_KEY, new TextEncoder().encode(bManifest)); + } + return inner(key, body, condition); + }; + // a.md matches the ancestor's size and mtime so takeSnapshot reuses its hash and sees no local + // change there; c.md is A's new local file, so A has something to push and a manifest to upload. + const reader = fakeReader({ "a.md": "xy", "c.md": "ccc" }); + const { writer, files } = fakeLocalWriter(); + files.set("a.md", "xy"); + files.set("c.md", "ccc"); + + const outcome = await syncOnce(ancestor, reader, writer, storage, 1); + + assert.deepEqual(outcome, { + ok: false, + message: "another device synced at the same time; sync again", + failures: [], + }); + // B's manifest survived; A's never landed. + assert.equal(objects.get(MANIFEST_KEY), bManifest); + // A's push still reached the bucket (harmless: the next pass folds it into the manifest). + assert.equal(objects.get("c.md"), "ccc"); + // Nothing was touched locally. + assert.equal(files.get("a.md"), "xy"); + assert.equal(files.get("c.md"), "ccc"); + + // The failed pass never advanced state.json, so A retries with the same ancestor, now against + // B's manifest: b.md is pulled, nothing is deleted, and the pass completes. + const retry = await syncOnce(ancestor, reader, writer, storage, 1); + assert.equal(retry.ok, true); + assert.equal(files.get("b.md"), "bee"); + assert.equal(files.get("a.md"), "xy"); +}); + +test("syncOnce: two first syncs racing for an empty bucket, the loser fails instead of clobbering", async () => { + // Both devices see no manifest and plan a first sync. The other device's manifest lands while + // this one is mid pass; the "ifAbsent" conditional upload must lose rather than overwrite it. + const { storage, objects } = fakeStorage(); + const otherManifest = JSON.stringify(snapshot(file("b.md", "h2"))); + const inner = storage.putObject; + let raced = false; + storage.putObject = async (key, body, condition) => { + if (key === MANIFEST_KEY && !raced) { + raced = true; + await inner(MANIFEST_KEY, new TextEncoder().encode(otherManifest)); + } + return inner(key, body, condition); + }; + const reader = fakeReader({ "a.md": "alpha" }); + const { writer, files } = fakeLocalWriter(); + files.set("a.md", "alpha"); + + const outcome = await syncOnce(empty, reader, writer, storage, 1); + + assert.equal(outcome.ok, false); + assert.equal(objects.get(MANIFEST_KEY), otherManifest); + assert.equal(files.get("a.md"), "alpha"); +}); diff --git a/src/sync/sync.ts b/src/sync/sync.ts index 9171212..92db9e5 100644 --- a/src/sync/sync.ts +++ b/src/sync/sync.ts @@ -1,4 +1,4 @@ -import type { StorageClient } from "../storage/storage.ts"; +import type { PutCondition, StorageClient } from "../storage/storage.ts"; import { isSnapshot, type Reader, type Snapshot, takeSnapshot } from "../vault/vault.ts"; import { executeSyncPlan, type LocalWriter, type SyncFailure } from "./execute.ts"; import { MANIFEST_KEY, planSync } from "./plan.ts"; @@ -20,9 +20,17 @@ export type SyncOutcome = // genuinely empty": syncOnce must ignore the local ancestor in the former (nothing has ever been // synced, so state.json cannot be a valid common ancestor) but trust it in the latter (an empty // remote that a prior sync really produced, where a local file absent from it was deleted). +// +// etag rides along with an existing manifest so syncOnce can make its manifest upload conditional +// on the remote still being exactly this version, the guard against two devices syncing at +// overlapping times (#83). export async function readRemoteManifest( storage: StorageClient, -): Promise<{ ok: true; snapshot: Snapshot; firstSync: boolean } | { ok: false; message: string }> { +): Promise< + | { ok: true; snapshot: Snapshot; firstSync: true } + | { ok: true; snapshot: Snapshot; firstSync: false; etag: string } + | { ok: false; message: string } +> { const fetched = await storage.getObject(MANIFEST_KEY); if (fetched.ok && fetched.body !== null) { @@ -35,7 +43,14 @@ export async function readRemoteManifest( if (!isSnapshot(parsed)) { return { ok: false, message: "remote manifest is corrupt" }; } - return { ok: true, snapshot: parsed, firstSync: false }; + // Every S3 compatible server returns an ETag on a successful read; without one (a stripping + // proxy, a broken provider) the manifest upload can't be made conditional, and uploading it + // unconditionally is exactly the concurrent clobber #83 fixed, so refuse rather than sync + // unsafely. + if (fetched.etag === null) { + return { ok: false, message: "remote manifest has no etag" }; + } + return { ok: true, snapshot: parsed, firstSync: false, etag: fetched.etag }; } // TODO(#41): GetResult conflates 404 with every other failure; swap this for a real status @@ -88,8 +103,26 @@ export async function syncOnce( // conflict rename just applied. const final = await takeSnapshot(reader, local); const manifestBody = new TextEncoder().encode(JSON.stringify(final)); - const uploaded = await storage.putObject(MANIFEST_KEY, manifestBody); + + // The upload is conditional on the remote manifest still being exactly what this pass read at + // the start (or still absent, on a first sync). An unconditional put would last-writer-win + // against a device syncing at overlapping times, and the loser's pushes would then read as + // remote deletions on the winner's next sync: files silently deleted (#83). Losing the race + // fails this pass loudly instead; state.json doesn't advance, and the next sync re-reads the + // fresh manifest and reconciles both devices' work with nothing lost. + let condition: PutCondition = { kind: "ifAbsent" }; + if (!remote.firstSync) { + condition = { kind: "ifMatch", etag: remote.etag }; + } + const uploaded = await storage.putObject(MANIFEST_KEY, manifestBody, condition); if (!uploaded.ok) { + if (uploaded.status === "conflict") { + return { + ok: false, + message: "another device synced at the same time; sync again", + failures: [], + }; + } return { ok: false, message: uploaded.message, failures: [] }; }