Investigate deadlock (#178)

.github/workflows/e2e.yml

@@ -49,3 +49,7 @@ jobs:
           cd ..
 
           scripts/e2e.sh 16
+
+      - name: Cleanup
+        if: always()
+        run: scripts/clean-up.sh

frontend/sync-client/src/index.ts

@@ -1,3 +1,4 @@
+import { awaitAll } from "./utils/await-all";
 import { logToConsole } from "./utils/debugging/log-to-console";
 import { slowFetchFactory } from "./utils/debugging/slow-fetch-factory";
 import { slowWebSocketFactory } from "./utils/debugging/slow-web-socket-factory";
@@ -41,5 +42,6 @@ export const debugging = {
 export const utils = {
 	getRandomColor,
 	positionToLineAndColumn,
-	lineAndColumnToPosition
+	lineAndColumnToPosition,
+	awaitAll
 };

frontend/sync-client/src/services/sync-service.ts

@@ -70,6 +70,10 @@ export class SyncService {
 				new Blob([new Uint8Array(contentBytes)])
 			);
 
+			this.logger.debug(
+				`Creating document with id ${documentId} and relative path ${relativePath}`
+			);
+
 			const response = await this.client(this.getUrl("/documents"), {
 				method: "POST",
 				body: formData,
@@ -110,7 +114,7 @@ export class SyncService {
 	}): Promise<DocumentUpdateResponse> {
 		return this.retryForever(async () => {
 			this.logger.debug(
-				`Updating text document ${documentId} with parent version ${parentVersionId} and relative path ${relativePath}`
+				`Updating text document ${documentId} with parent version ${parentVersionId} and relative path ${relativePath}, content [${content.join(", ")}]`
 			);
 
 			const request: UpdateTextDocumentVersion = {
@@ -213,6 +217,11 @@ export class SyncService {
 			const request: DeleteDocumentVersion = {
 				relativePath
 			};
+
+			this.logger.debug(
+				`Delete document with id ${documentId} and relative path ${relativePath}`
+			);
+
 			const response = await this.client(
 				this.getUrl(`/documents/${documentId}`),
 				{
@@ -247,6 +256,8 @@ export class SyncService {
 		documentId: DocumentId;
 	}): Promise<DocumentVersion> {
 		return this.retryForever(async () => {
+			this.logger.debug(`Getting document with id ${documentId}`);
+
 			const response = await this.client(
 				this.getUrl(`/documents/${documentId}`),
 				{
@@ -275,6 +286,11 @@ export class SyncService {
 		since?: VaultUpdateId
 	): Promise<FetchLatestDocumentsResponse> {
 		return this.retryForever(async () => {
+			this.logger.debug(
+				"Getting all documents" +
+					(since != null ? ` since ${since}` : "")
+			);
+
 			const url = new URL(this.getUrl("/documents"));
 			if (since !== undefined) {
 				url.searchParams.append("since", since.toString());
@@ -303,6 +319,7 @@ export class SyncService {
 	}
 
 	public async ping(): Promise<PingResponse> {
+		this.logger.debug("Pinging server");
 		const response = await this.pingClient(this.getUrl("/ping"), {
 			headers: this.getDefaultHeaders()
 		});

frontend/test-client/src/agent/mock-agent.ts

@@ -2,10 +2,13 @@ import { choose } from "../utils/choose";
 import { v4 as uuidv4 } from "uuid";
 import { assert } from "../utils/assert";
 import type { RelativePath, SyncSettings } from "sync-client";
-import { debugging, Logger, LogLevel } from "sync-client";
+import { debugging, Logger, LogLevel, utils } from "sync-client";
 import { MockClient } from "./mock-client";
 import { sleep } from "../utils/sleep";
-import type { LogLine } from "sync-client/dist/types/tracing/logger";
+import type { LogLine } from "sync-client";
+import { withTimeout } from "../utils/with-timeout";
+
+const TIMEOUT_MS = 10 * 60 * 1000;
 
 export class MockAgent extends MockClient {
 	private readonly writtenContents: string[] = [];
@@ -134,15 +137,26 @@ export class MockAgent extends MockClient {
 	}
 
 	public async finish(): Promise<void> {
+		await withTimeout(
+			(async (): Promise<void> => {
 				await this.client.setSetting("isSyncEnabled", true);
-		// eslint-disable-next-line no-restricted-properties
+				await utils.awaitAll(this.pendingActions);
-		await Promise.all(this.pendingActions);
 				await this.client.waitUntilFinished();
+			})(),
+			TIMEOUT_MS,
+			"finish()"
+		);
 	}
 
 	public async destroy(): Promise<void> {
+		await withTimeout(
+			(async (): Promise<void> => {
 				await this.client.waitUntilFinished();
 				await this.client.destroy();
+			})(),
+			TIMEOUT_MS,
+			"destroy()"
+		);
 	}
 
 	public assertFileSystemsAreConsistent(otherAgent: MockAgent): void {

frontend/test-client/src/cli.ts

@@ -1,4 +1,5 @@
 import type { SyncSettings } from "sync-client";
+import { utils } from "sync-client";
 import { MockAgent } from "./agent/mock-agent";
 import { sleep } from "./utils/sleep";
 import { v4 as uuidv4 } from "uuid";
@@ -56,14 +57,12 @@ async function runTest({
 	}
 
 	try {
-		// eslint-disable-next-line no-restricted-properties
+		await utils.awaitAll(clients.map(async (client) => client.init()));
-		await Promise.all(clients.map(async (client) => client.init()));
 
 		for (let i = 0; i < iterations; i++) {
 			console.info(`Iteration ${i + 1}/${iterations}`);
-			// eslint-disable-next-line no-restricted-properties
+			await utils.awaitAll(clients.map(async (client) => client.act()));
-			await Promise.all(clients.map(async (client) => client.act()));
+			await sleep(Math.random() * 200);
-			await sleep(100);
 		}
 
 		console.info("Stopping agents");
@@ -71,6 +70,7 @@ async function runTest({
 		// Each agent can have unpushed changes which might conflict with eachother so each has to resolve the conflicts & push, and
 		for (const client of clients) {
 			try {
+				console.info(`Finishing up ${client.name}`);
 				await client.finish();
 			} catch (err) {
 				if (!slowFileEvents) {
@@ -82,6 +82,7 @@ async function runTest({
 		// then we need a second pass to ensure that all agents pull the same state.
 		for (const client of clients) {
 			try {
+				console.info(`Destroying ${client.name}`);
 				await client.destroy();
 			} catch (err) {
 				if (!slowFileEvents) {

frontend/test-client/src/utils/with-timeout.ts

@@ -0,0 +1,16 @@
+export async function withTimeout<T>(
+	promise: Promise<T>,
+	timeoutMs: number,
+	operationName: string
+): Promise<T> {
+	return Promise.race([
+		promise,
+		new Promise<T>((_, reject) =>
+			setTimeout(() => {
+				reject(
+					new Error(`${operationName} timed out after ${timeoutMs}ms`)
+				);
+			}, timeoutMs)
+		)
+	]);
+}

scripts/e2e.sh

@@ -3,6 +3,9 @@
 set -e
 set -o pipefail
 
+NO_COLOR=1
+FORCE_COLOR=0
+
 node_version=$(node -v | sed 's/^v\([0-9]*\).*/\1/')
 if [ "$node_version" != "22" ]; then
     echo "Error: This script requires Node.js version 22, found: $node_version"
@@ -37,8 +40,18 @@ cd frontend
 
 pids=()
 for i in $(seq 1 $process_count); do
-    node test-client/dist/cli.js > "../logs/log_${i}.log" 2>&1 &
+    # Create a named pipe for this process
-    pids+=($!)
+    pipe="/tmp/vaultlink_pipe_$$_$i"
+    mkfifo "$pipe"
+
+    # Start the node process writing to the pipe
+    node test-client/dist/cli.js > "$pipe" 2>&1 &
+    pid=$!
+    pids+=($pid)
+    echo "Started process $i with PID: $pid"
+
+    # Read from pipe, prefix with PID, and write to log file
+    (sed "s/^/[PID $pid] /" < "$pipe" > "../logs/log_${i}.log"; rm "$pipe") &
 done
 
 cd ..
@@ -52,6 +65,7 @@ print_failed_log() {
 
             # Only consider non-zero exit codes as failures
             if [ $exit_code -ne 0 ]; then
+                echo "----- Log for process ${pids[$i-1]} (log_${i}.log) -----"
                 cat "$(pwd)/logs/log_${i}.log"
                 echo "Process ${pids[$i-1]} failed with exit code $exit_code. Log file: $(pwd)/logs/log_${i}.log"
                 return 0