Handle potentially conflicting keys

Author: s-h-a-d-o-w

.github/workflows/ci.yml

@@ -0,0 +1,23 @@
+name: CI
+
+on:
+  push:
+    branches:
+      - master
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install pnpm
+        uses: pnpm/action-setup@v4
+      - name: Use Node.js 22
+        uses: actions/setup-node@v4
+        with:
+          node-version: 22
+      - name: Install, test
+        run: |
+          pnpm install
+          pnpm test

client/src/biDirectionalSync.ts

@@ -2,14 +2,17 @@ import { statSync } from "node:fs";
 import { readdir } from "node:fs/promises";
 import { join } from "node:path";
 import { logger } from "@s3-smart-sync/shared/logger.js";
-import { LOCAL_DIR } from "./consts.js";
+import { LOCAL_DIR, S3_BUCKET } from "./consts.js";
 import {
   convertAbsolutePathToKey,
+  deleteObject,
   download,
   listS3Files,
+  s3Client,
   upload,
 } from "./s3Operations.js";
 import { destroyTrayIcon } from "./trayWrapper.js";
+import { _Object, ListObjectsV2Command } from "@aws-sdk/client-s3";
 
 async function listLocalFiles(dir: string) {
   const files: Array<{
@@ -45,10 +48,7 @@ export async function biDirectionalSync() {
   logger.info("Starting full sync...");
   isSyncInProgress = true;
 
-  const [localFiles, [s3Files, noLastModifiedInfo]] = await Promise.all([
-    listLocalFiles(LOCAL_DIR),
-    listS3Files(),
-  ]);
+  const [preliminaryS3Files, noLastModifiedInfo] = await listS3Files();
   if (noLastModifiedInfo.length > 0) {
     logger.error(
       `No "last modified" date from S3 for file(s): ${noLastModifiedInfo.join(", ")}.\nPlease address this before starting the client again.`,
@@ -57,6 +57,15 @@ export async function biDirectionalSync() {
     process.exit(1);
   }
 
+  const fixConflictsPromises = await fixConflicts(preliminaryS3Files);
+  // It's possible that some things are attempted to be deleted twice for various reasons. Obviously, we don't care about errors thrown because something that caused the config was already deleted.
+  await Promise.allSettled(fixConflictsPromises);
+
+  const [localFiles, [s3Files]] = await Promise.all([
+    listLocalFiles(LOCAL_DIR),
+    listS3Files(),
+  ]);
+
   const localPromises = localFiles
     .map(({ key, lastModified }) => {
       const s3File = s3Files.find((s3File) => s3File.key === key);
@@ -95,3 +104,73 @@ export async function biDirectionalSync() {
   isSyncInProgress = false;
   logger.info("Done.\n");
 }
+
+async function fixConflicts(
+  s3Files: Array<{ key: string; lastModified: Date }>,
+) {
+  const deletePromises: Promise<void>[] = [];
+
+  const keyMap = new Map<string, Array<{ key: string; lastModified: Date }>>();
+  for (const file of s3Files) {
+    const baseKey = file.key.endsWith("/") ? file.key.slice(0, -1) : file.key;
+    if (!keyMap.has(baseKey)) {
+      keyMap.set(baseKey, []);
+    }
+    keyMap.get(baseKey)!.push(file);
+  }
+
+  // Find conflicts (where both a file and directory with the same baseKey exist)
+  for (const entries of keyMap.values()) {
+    if (entries.length === 2) {
+      const directory = entries[0]!.key.endsWith("/")
+        ? entries[0]!
+        : entries[1]!;
+      const file = entries[0]!.key.endsWith("/") ? entries[1]! : entries[0]!;
+
+      if (file.lastModified < directory.lastModified) {
+        logger.info(
+          `Resolving conflict: deleting file ${file.key} (keeping ${directory.key})`,
+        );
+        deletePromises.push(deleteObject(file.key));
+      } else {
+        logger.info(
+          `Resolving conflict: deleting directory ${directory.key} (keeping ${file.key})`,
+        );
+        deletePromises.push(deleteObject(directory.key));
+
+        // Also delete all objects found under this directory
+        let continuationToken: string | undefined = undefined;
+        do {
+          const { Contents, NextContinuationToken } = (await s3Client.send(
+            new ListObjectsV2Command({
+              Bucket: S3_BUCKET,
+              Prefix: directory.key,
+              ...(continuationToken
+                ? { ContinuationToken: continuationToken }
+                : {}),
+            }),
+          )) as {
+            Contents?: _Object[];
+            NextContinuationToken: string | undefined;
+          };
+
+          if (Contents && Contents.length > 0) {
+            for (const { Key } of Contents) {
+              if (Key) {
+                deletePromises.push(deleteObject(Key));
+              }
+            }
+          }
+
+          continuationToken = NextContinuationToken;
+        } while (continuationToken);
+      }
+    } else if (entries.length > 2) {
+      logger.error(
+        `Found ${entries.length} conflicting entries for "${entries[0]!.key}": ${entries.map((e) => e.key).join(", ")}`,
+      );
+    }
+  }
+
+  return deletePromises;
+}

client/src/s3Operations.ts

@@ -9,7 +9,7 @@ import {
 import { Upload } from "@aws-sdk/lib-storage";
 import { logger } from "@s3-smart-sync/shared/logger.js";
 import { createReadStream, createWriteStream } from "node:fs";
-import { mkdir, stat, utimes } from "node:fs/promises";
+import { mkdir, rm, stat, utimes } from "node:fs/promises";
 import { dirname, join, relative } from "node:path";
 import { pipeline } from "stream/promises";
 import {
@@ -78,14 +78,30 @@ export async function deleteObject(key: string) {
 export async function download(key: string, localPath: string) {
   logger.info(`Downloading: ${key}`);
 
-  if (key.endsWith("/")) {
-    const { LastModified } = await s3Client.send(
-      new GetObjectCommand({
-        Bucket: S3_BUCKET,
-        Key: key,
-      }),
-    );
+  const { Body, LastModified } = await s3Client.send(
+    new GetObjectCommand({
+      Bucket: S3_BUCKET,
+      Key: key,
+    }),
+  );
 
+  // delete a possibly outdated local file that will be replaced by a directory
+  try {
+    const stats = await stat(localPath);
+    if (
+      stats.mtime < (LastModified ?? 0) &&
+      key.endsWith("/") &&
+      stats.isFile()
+    ) {
+      ignore(FileOperationType.Remove, localPath);
+      await rm(localPath);
+      unignore(FileOperationType.Remove, localPath);
+    }
+  } catch (_) {
+    // local file doesn't exist yet
+  }
+
+  if (key.endsWith("/")) {
     ignore(FileOperationType.Sync, localPath);
     try {
       await mkdir(localPath, { recursive: true });
@@ -98,13 +114,6 @@ export async function download(key: string, localPath: string) {
     return;
   }
 
-  const { Body, LastModified } = await s3Client.send(
-    new GetObjectCommand({
-      Bucket: S3_BUCKET,
-      Key: key,
-    }),
-  );
-
   if (Body) {
     // We don't manage ignoring potentially new created directories here because that would be a lot of overhead. Instead, if syncing is triggered, we let the upload of the directory handle breaking that chain. (via updating modification time and that timestamp then being the same)
     await mkdir(dirname(localPath), { recursive: true });

client/tests/e2e.spec.mts

@@ -10,12 +10,14 @@ import {
   cleanupS3,
   createClientDirectories,
   createFile,
+  list,
   pause,
   sendSnsMessage,
   startClients,
   startServer,
   stopClients,
   stopServer,
+  upload,
   waitUntil,
 } from "./utilities.js";
 
@@ -25,8 +27,7 @@ const clientDirectories = await createClientDirectories(clientIds);
 describe("E2E Tests", () => {
   beforeAll(async () => {
     await startServer();
-    // startClients([0, 1]);
-    startClients([0, 1], [0, 1]);
+    startClients([0, 1]);
     await pause(1000);
   });
 
@@ -121,10 +122,8 @@ describe("E2E Tests", () => {
 
   it("should handle replacing a file with an empty directory", async () => {
     await createFile(0, "file-then-directory", "starts as a file");
-    await waitUntil(async () =>
-      expect(
-        await fileExists(join(clientDirectories[1], "file-then-directory")),
-      ).toBe(true),
+    await waitUntil(() =>
+      fileExists(join(clientDirectories[1], "file-then-directory")),
     );
 
     await rm(join(clientDirectories[0], "file-then-directory"));
@@ -137,25 +136,23 @@ describe("E2E Tests", () => {
     await pause(WATCHER_DEBOUNCE_DURATION + 300);
     await sendSnsMessage("file-then-directory/", "put");
 
-    await waitUntil(async () => {
-      const stats = await stat(
-        join(clientDirectories[1], "file-then-directory"),
-      );
-      return stats.isDirectory();
-    });
+    await waitUntil(async () =>
+      (
+        await stat(join(clientDirectories[1], "file-then-directory"))
+      ).isDirectory(),
+    );
   });
 
   it("should handle replacing an empty directory with a file", async () => {
     await mkdir(join(clientDirectories[0], "directory-then-file"));
     // First, the debounced upload. Then we have to wait for the upload to actually have finished
     await pause(WATCHER_DEBOUNCE_DURATION + 300);
     await sendSnsMessage("directory-then-file/", "put");
-    await waitUntil(async () => {
-      const stats = await stat(
-        join(clientDirectories[1], "directory-then-file"),
-      );
-      return stats.isDirectory();
-    });
+    await waitUntil(async () =>
+      (
+        await stat(join(clientDirectories[1], "directory-then-file"))
+      ).isDirectory(),
+    );
 
     await rm(join(clientDirectories[0], "directory-then-file"), {
       recursive: true,
@@ -174,35 +171,20 @@ describe("E2E Tests", () => {
     });
   });
 
-  // it("handles duplicate file/directory on S3", async () => {
-  //   await s3Client.send(
-  //     new PutObjectCommand({
-  //       Bucket: S3_BUCKET,
-  //       Key: "duplicate-file",
-  //       Body: Buffer.from(""),
-  //     }),
-  //   );
-  //   await s3Client.send(
-  //     new PutObjectCommand({
-  //       Bucket: S3_BUCKET,
-  //       Key: "duplicate-file/",
-  //       Body: Buffer.from(""),
-  //     }),
-  //   );
-
-  //   await sendSnsMessage("duplicate-file/", "put");
-  //   await waitUntil(async () => {
-  //     const stats = await stat(join(CLIENT_1_DIR, "duplicate-file/"));
-  //     return stats.isDirectory();
-  //   });
-
-  //   const { Contents } = await s3Client.send(
-  //     new ListObjectsV2Command({
-  //       Bucket: S3_BUCKET,
-  //       Prefix: "duplicate-file/",
-  //     }),
-  //   );
-  //   expect(Contents?.length).toBe(1);
-  //   expect(Contents?.[0]?.Key).toBe("duplicate-file/");
-  // });
+  it("handles duplicate file/directory on S3", async () => {
+    await stopClients([0, 1]);
+    await upload("duplicate-file/", "");
+    await upload("duplicate-file/nested/", "");
+    await upload("duplicate-file/nested/file.txt", "...");
+    await upload("duplicate-file", "");
+
+    startClients([0, 1]);
+    await waitUntil(() =>
+      readFile(join(clientDirectories[0], "duplicate-file"), "utf-8"),
+    );
+
+    const { Contents } = await list("duplicate-file");
+    expect(Contents?.length).toBe(1);
+    expect(Contents?.[0]?.Key).toBe("duplicate-file");
+  });
 });