Sync → Prevent concurrent syncs from silently deleting files

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";
 }

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");

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<PutResult>;
+  putObject: (key: string, body: Uint8Array, condition?: PutCondition) => Promise<PutResult>;
   getObject: (key: string) => Promise<GetResult>;
   deleteObject: (key: string) => Promise<DeleteResult>;
   listObjects: (prefix?: string) => Promise<ListResult>;
@@ -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<string, string> {
+  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<PutResult> {
   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) };

src/sync/fake.ts

@@ -53,14 +53,34 @@ export function fakeReader(files: Record<string, string>): 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<string, string> = {}): {
   storage: StorageClient;
   objects: Map<string, string>;
 } {
   const store = new Map<string, string>(Object.entries(objects));
+  let revision = 0;
+  const etags = new Map<string, string>();
+  for (const key of store.keys()) {
+    revision++;
+    etags.set(key, `"v${revision}"`);
+  }
   const storage: StorageClient = {
-    putObject: async (key, body): Promise<PutResult> => {
+    putObject: async (key, body, condition): Promise<PutResult> => {
+      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<string, string> = {}): {
           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<DeleteResult> => {
       store.delete(key);
+      etags.delete(key);
       return { ok: true, status: "ok", message: "" };
     },
     listObjects: async (): Promise<ListResult> => {

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();