feat: optimize retry logic

Author: janthurau

packages/common/src/SkipFurtherHooksError.ts

@@ -0,0 +1,6 @@
+export class SkipFurtherHooksError extends Error {
+	constructor(message?: string) {
+		super(message ?? "Further hooks skipped");
+		this.name = "SkipFurtherHooksError";
+	}
+}

packages/common/src/index.ts

@@ -2,4 +2,5 @@ export * from "./auth.ts";
 export * from "./CloseEvents.ts";
 export * from "./awarenessStatesToArray.ts";
 export * from "./routingKey.ts";
+export * from "./SkipFurtherHooksError.ts";
 export * from "./types.ts";

packages/extension-redis/package.json

@@ -31,6 +31,7 @@
 		"@types/lodash.debounce": "^4.0.6"
 	},
 	"dependencies": {
+		"@hocuspocus/common": "workspace:^",
 		"@hocuspocus/server": "workspace:^",
 		"@sesamecare-oss/redlock": "^1.4.0",
 		"ioredis": "~5.6.1",

packages/extension-redis/src/Redis.ts

@@ -1,24 +1,25 @@
 import crypto from "node:crypto";
 import type {
-	Document,
-	Extension,
-	Hocuspocus,
-	RedisTransactionOrigin,
 	afterLoadDocumentPayload,
 	afterStoreDocumentPayload,
 	afterUnloadDocumentPayload,
 	beforeBroadcastStatelessPayload,
 	beforeUnloadDocumentPayload,
+	Document,
+	Extension,
+	Hocuspocus,
 	onAwarenessUpdatePayload,
 	onChangePayload,
 	onConfigurePayload,
 	onStoreDocumentPayload,
+	RedisTransactionOrigin,
 } from "@hocuspocus/server";
+import { SkipFurtherHooksError } from "@hocuspocus/common";
 import {
 	IncomingMessage,
+	isTransactionOrigin,
 	MessageReceiver,
 	OutgoingMessage,
-	isTransactionOrigin,
 } from "@hocuspocus/server";
 import {
 	ExecutionError,
@@ -266,10 +267,8 @@ export class Redis implements Extension {
 					"The operation was unable to achieve a quorum during its retry window."
 			) {
 				// Expected behavior: Could not acquire lock, another instance locked it already.
-				// No further `onStoreDocument` hooks will be executed; should throw a silent error with no message.
-				throw new Error("", {
-					cause: "Could not acquire lock, another instance locked it already.",
-				});
+				// Skip further hooks and retry — the data is safe on the other instance.
+				throw new SkipFurtherHooksError("Another instance is already storing this document");
 			}
 			//unexpected error
 			console.error("unexpected error:", error);

packages/server/src/Hocuspocus.ts

@@ -1,5 +1,5 @@
 import crypto from "node:crypto";
-import { awarenessStatesToArray, ResetConnection } from "@hocuspocus/common";
+import { awarenessStatesToArray, ResetConnection, SkipFurtherHooksError } from "@hocuspocus/common";
 import { applyUpdate, Doc, encodeStateAsUpdate } from "yjs";
 import meta from "../package.json" with { type: "json" };
 
@@ -321,7 +321,7 @@ export class Hocuspocus<Context = any> {
 		context?: Context,
 	): Promise<Document> {
 		if (!documentName.trim()) {
-			throw new Error('Document name must not be empty')
+			throw new Error("Document name must not be empty");
 		}
 
 		const existingLoadingDoc = this.loadingDocuments.get(documentName);
@@ -473,9 +473,20 @@ export class Hocuspocus<Context = any> {
 						await this.hooks("afterStoreDocument", hookPayload);
 					});
 				} catch (error: any) {
-					console.error("Caught error during storeDocumentHooks, retrying", error);
-					// Retry to avoid data loss — the document stays in memory until the store succeeds
-					this.storeDocumentHooks(document, hookPayload);
+					if (error instanceof SkipFurtherHooksError) {
+						// Another extension handled this — proceed to unload
+						setTimeout(() => {
+							if (this.shouldUnloadDocument(document)) {
+								this.unloadDocument(document);
+							}
+						}, 0);
+						return;
+					}
+
+					console.error(
+						"Caught error during storeDocumentHooks. Document stays in memory to avoid data loss",
+						error,
+					);
 					return;
 				}
 

pnpm-lock.yaml

@@ -146,7 +146,7 @@ test('document gets unloaded on both servers after disconnection', async t => {
             t.is(server.documents.size, 0)
 
             resolve('')
-          }, 5000) // must be higher than RedisExtension.disconnectDelay
+          }, 10000) // must be higher than CustomStorageExtension delay (3s) + RedisExtension.disconnectDelay (1s)
         }, 1500)
 
       },

tests/extension-redis/onStoreDocument.ts

@@ -233,29 +233,29 @@ test("stops when one of the onStoreDocument hooks throws a non-Error value", asy
 	});
 });
 
-test("retries onStoreDocument on transient failure and succeeds", async (t) => {
+test("keeps document in memory when onStoreDocument fails", async (t) => {
 	await new Promise(async (resolve) => {
-		let callCount = 0;
-
 		const server = await newHocuspocus(t, {
 			debounce: 100,
-			async onStoreDocument({ document }) {
-				callCount++;
-				if (callCount < 3) {
-					throw new Error("transient failure");
-				}
-				// Third attempt succeeds
-				const value = document.getArray("foo").get(0);
-				t.is(value, "bar");
-				t.is(callCount, 3, "should succeed on third attempt");
-				resolve("done");
+			async onStoreDocument() {
+				throw new Error("storage unavailable");
 			},
 		});
 
-		const provider = newHocuspocusProvider(t, server);
+		const socket = newHocuspocusProviderWebsocket(t, server);
 
-		provider.on("synced", () => {
-			provider.document.getArray("foo").insert(0, ["bar"]);
+		const provider = newHocuspocusProvider(t, server, {
+			websocketProvider: socket,
+			onSynced() {
+				provider.document.getArray("foo").push(["bar"]);
+				socket.destroy();
+
+				setTimeout(() => {
+					// Document must remain in memory since the store failed
+					t.is(server.getDocumentsCount(), 1);
+					resolve("done");
+				}, 500);
+			},
 		});
 	});
 });