Guard file uploads with remote object preconditions

src/sync/execute.ts

@@ -1,17 +1,22 @@
-import type { StorageClient } from "../storage/storage.ts";
+import type { PutCondition, StorageClient } from "../storage/storage.ts";
 import { byPath, type FileState, hashBytes, type Reader, type Snapshot } from "../vault/vault.ts";
 import { conflictCopyPath, type SyncAction } from "./plan.ts";
 
 // DRIFT_MESSAGE is the failure reported when a local file changed after the snapshot an action
 // was planned from; the next sync re-snapshots and replans the path as a conflict.
 const DRIFT_MESSAGE = "changed locally mid sync; sync again to reconcile";
 
+const REMOTE_DRIFT_MESSAGE = "changed remotely mid sync; sync again to reconcile";
+const REMOTE_ETAG_MESSAGE = "remote object has no etag";
+
 // ExecuteResult reports what executeSyncPlan carried out: completed holds every action fully
-// applied, failed the actions that weren't, and failures the per file detail of why. failed is
-// carried separately from failures because a conflict's failure can name its copy path rather
-// than the action's own path.
+// applied, failed the actions that weren't, failures the per file detail of why, and concurrent
+// whether a file precondition proved the remote snapshot stale. failed is carried separately
+// from failures because a conflict's failure can name its copy path rather than the action's own
+// path.
 export type ExecuteResult = {
   completed: SyncAction[];
+  concurrent: boolean;
   failed: SyncAction[];
   failures: SyncFailure[];
 };
@@ -30,45 +35,66 @@ export type SyncFailure = {
   message: string;
 };
 
+type ActionResult = {
+  concurrent: boolean;
+  failures: SyncFailure[];
+};
+
+type PutConditionResult =
+  | { ok: true; kind: "done" }
+  | { ok: true; kind: "put"; condition: PutCondition }
+  | { ok: false; concurrent: boolean; failure: SyncFailure };
+
 // executeSyncPlan carries out every action against reader/localWriter (the local vault) and
 // storage (the remote bucket), and reports what completed and what couldn't be, so one failed
 // file never discards the progress of the rest of the pass (#87). local is the snapshot the plan
 // was made from, so each destructive local write can first check the file hasn't changed since
 // (#86). now is passed in rather than read internally so a conflict's copy name is deterministic
-// under test.
+// under test. remote is the manifest the plan was made from, used to make file PUTs conditional;
+// its empty default makes callers that lack a remote view create-only rather than overwrite.
 export async function executeSyncPlan(
   actions: SyncAction[],
   local: Snapshot,
   reader: Reader,
   localWriter: LocalWriter,
   storage: StorageClient,
   now: number,
+  remote: Snapshot = { files: [] },
 ): Promise<ExecuteResult> {
   const completed: SyncAction[] = [];
+  let concurrent = false;
   const failed: SyncAction[] = [];
   const failures: SyncFailure[] = [];
   const localByPath = byPath(local.files);
+  const remoteByPath = byPath(remote.files);
 
   for (const action of actions) {
-    const actionFailures = await executeAction(
+    const actionResult = await executeAction(
       action,
       localByPath,
+      remoteByPath,
       reader,
       localWriter,
       storage,
       now,
     );
-    if (actionFailures.length === 0) {
+    if (actionResult.failures.length === 0) {
       completed.push(action);
       continue;
     }
     failed.push(action);
-    for (const failure of actionFailures) {
+    if (actionResult.concurrent) {
+      concurrent = true;
+    }
+    for (const failure of actionResult.failures) {
       failures.push(failure);
     }
+    if (actionResult.concurrent) {
+      break;
+    }
   }
 
-  return { completed, failed, failures };
+  return { completed, concurrent, failed, failures };
 }
 
 // applyLocalWrite runs one localWriter mutation, converting a thrown I/O error into a SyncFailure
@@ -119,73 +145,88 @@ async function checkLocalDrift(
   return null;
 }
 
-// executeAction carries out a single action and returns every failure it hit, an empty array
-// meaning the action fully completed. An action can report more than one failure: a conflict
-// whose copy push fails still pulls the remote version, so the diverged local edit lands on disk
-// even when the bucket refuses the copy.
+// executeAction carries out a single action and reports its failures and whether remote
+// concurrency invalidated the plan. An action can report more than one failure: a conflict whose
+// copy push fails still pulls the remote version, so the diverged local edit lands on disk even
+// when the bucket refuses the copy.
 async function executeAction(
   action: SyncAction,
   localByPath: Map<string, FileState>,
+  remoteByPath: Map<string, FileState>,
   reader: Reader,
   localWriter: LocalWriter,
   storage: StorageClient,
   now: number,
-): Promise<SyncFailure[]> {
+): Promise<ActionResult> {
   if (action.kind === "push") {
     let bytes: Uint8Array;
     try {
       bytes = await reader.readFile(action.path);
     } catch (err) {
-      return [{ path: action.path, message: localFailureMessage(err) }];
+      return failedAction(action.path, localFailureMessage(err), false);
+    }
+    const checked = await putCondition(action.path, bytes, remoteByPath.get(action.path), storage);
+    if (!checked.ok) {
+      return { concurrent: checked.concurrent, failures: [checked.failure] };
     }
-    const result = await storage.putObject(action.path, bytes);
+    if (checked.kind === "done") {
+      return successfulAction();
+    }
+    const result = await storage.putObject(action.path, bytes, checked.condition);
     if (!result.ok) {
-      return [{ path: action.path, message: result.message }];
+      if (result.status !== "conflict") {
+        return failedAction(action.path, result.message, false);
+      }
+      const matches = await remoteMatches(action.path, bytes, storage);
+      if (matches) {
+        return successfulAction();
+      }
+      return failedAction(action.path, result.message, true);
     }
 
-    return [];
+    return successfulAction();
   }
 
   if (action.kind === "pushDelete") {
     const result = await storage.deleteObject(action.path);
     if (!result.ok) {
-      return [{ path: action.path, message: result.message }];
+      return failedAction(action.path, result.message, false);
     }
 
-    return [];
+    return successfulAction();
   }
 
   if (action.kind === "pull") {
     const drift = await checkLocalDrift(reader, action.path, localByPath.get(action.path));
     if (drift !== null) {
-      return [drift];
+      return { concurrent: false, failures: [drift] };
     }
     const result = await storage.getObject(action.path);
     if (!result.ok || result.body === null) {
-      return [{ path: action.path, message: result.message }];
+      return failedAction(action.path, result.message, false);
     }
     const body = result.body;
     const failure = await applyLocalWrite(action.path, () =>
       localWriter.writeFile(action.path, body),
     );
     if (failure !== null) {
-      return [failure];
+      return { concurrent: false, failures: [failure] };
     }
 
-    return [];
+    return successfulAction();
   }
 
   if (action.kind === "pullDelete") {
     const drift = await checkLocalDrift(reader, action.path, localByPath.get(action.path));
     if (drift !== null) {
-      return [drift];
+      return { concurrent: false, failures: [drift] };
     }
     const failure = await applyLocalWrite(action.path, () => localWriter.deleteFile(action.path));
     if (failure !== null) {
-      return [failure];
+      return { concurrent: false, failures: [failure] };
     }
 
-    return [];
+    return successfulAction();
   }
 
   // conflict, deletedSide "local": the user deleted their copy, so there is no local edit to
@@ -194,21 +235,21 @@ async function executeAction(
   if (action.deletedSide === "local") {
     const drift = await checkLocalDrift(reader, action.path, localByPath.get(action.path));
     if (drift !== null) {
-      return [drift];
+      return { concurrent: false, failures: [drift] };
     }
     const result = await storage.getObject(action.path);
     if (!result.ok || result.body === null) {
-      return [{ path: action.path, message: result.message }];
+      return failedAction(action.path, result.message, false);
     }
     const body = result.body;
     const failure = await applyLocalWrite(action.path, () =>
       localWriter.writeFile(action.path, body),
     );
     if (failure !== null) {
-      return [failure];
+      return { concurrent: false, failures: [failure] };
     }
 
-    return [];
+    return successfulAction();
   }
 
   // conflict, deletedSide "remote" or "none": preserve the local edit under a new name and
@@ -220,7 +261,7 @@ async function executeAction(
   try {
     localBytes = await reader.readFile(action.path);
   } catch (err) {
-    return [{ path: action.path, message: localFailureMessage(err) }];
+    return failedAction(action.path, localFailureMessage(err), false);
   }
   // A failed rename means the local edit is still sitting at action.path untouched. Bail before
   // the pull below would overwrite it, so a diverged edit is never silently discarded by an I/O
@@ -229,24 +270,26 @@ async function executeAction(
     localWriter.renameFile(action.path, copyPath),
   );
   if (renameFailure !== null) {
-    return [renameFailure];
+    return { concurrent: false, failures: [renameFailure] };
   }
   const failures: SyncFailure[] = [];
-  const pushed = await storage.putObject(copyPath, localBytes);
+  let concurrent = false;
+  const pushed = await storage.putObject(copyPath, localBytes, { kind: "ifAbsent" });
   if (!pushed.ok) {
     failures.push({ path: copyPath, message: pushed.message });
+    concurrent = pushed.status === "conflict";
   }
 
   // deletedSide "remote": there is nothing at this path remotely to pull, the rename above
   // already vacated it locally, and that is the correct final state, not a failure to report.
   if (action.deletedSide === "remote") {
-    return failures;
+    return { concurrent, failures };
   }
 
   const result = await storage.getObject(action.path);
   if (!result.ok || result.body === null) {
     failures.push({ path: action.path, message: result.message });
-    return failures;
+    return { concurrent, failures };
   }
   const body = result.body;
   const writeFailure = await applyLocalWrite(action.path, () =>
@@ -256,7 +299,78 @@ async function executeAction(
     failures.push(writeFailure);
   }
 
-  return failures;
+  return { concurrent, failures };
+}
+
+// failedAction returns one failed action result with its concurrency classification.
+function failedAction(path: string, message: string, concurrent: boolean): ActionResult {
+  return { concurrent, failures: [{ path, message }] };
+}
+
+// putCondition returns the precondition that keeps a file PUT tied to the remote snapshot this
+// pass planned from. Existing objects are also hashed before their ETag is trusted, because an
+// ETag fetched after another pass's PUT describes that newer object rather than the snapshot.
+async function putCondition(
+  path: string,
+  bytes: Uint8Array,
+  expected: FileState | undefined,
+  storage: StorageClient,
+): Promise<PutConditionResult> {
+  if (expected === undefined) {
+    return { ok: true, kind: "put", condition: { kind: "ifAbsent" } };
+  }
+  const fetched = await storage.getObject(path);
+  if (!fetched.ok || fetched.body === null) {
+    if (fetched.status === "not_found") {
+      return { ok: true, kind: "put", condition: { kind: "ifAbsent" } };
+    }
+    return {
+      ok: false,
+      concurrent: false,
+      failure: { path, message: fetched.message },
+    };
+  }
+  const remoteHash = await hashBytes(fetched.body);
+  if (remoteHash !== expected.hash) {
+    if (remoteHash === (await hashBytes(bytes))) {
+      return { ok: true, kind: "done" };
+    }
+    return {
+      ok: false,
+      concurrent: true,
+      failure: { path, message: REMOTE_DRIFT_MESSAGE },
+    };
+  }
+  if (fetched.etag === null) {
+    return {
+      ok: false,
+      concurrent: false,
+      failure: { path, message: REMOTE_ETAG_MESSAGE },
+    };
+  }
+
+  return { ok: true, kind: "put", condition: { kind: "ifMatch", etag: fetched.etag } };
+}
+
+// remoteMatches reports whether path already holds bytes, making a failed create idempotent. A
+// previous pass can leave an unmanifested object after losing the manifest CAS; accepting those
+// same bytes lets the retry fold it into the manifest without an unsafe overwrite.
+async function remoteMatches(
+  path: string,
+  bytes: Uint8Array,
+  storage: StorageClient,
+): Promise<boolean> {
+  const fetched = await storage.getObject(path);
+  if (!fetched.ok || fetched.body === null) {
+    return false;
+  }
+
+  return (await hashBytes(fetched.body)) === (await hashBytes(bytes));
+}
+
+// successfulAction returns the zero failure result for a completed action.
+function successfulAction(): ActionResult {
+  return { concurrent: false, failures: [] };
 }
 
 // localFailureMessage turns whatever a local vault operation threw into a SyncFailure message.