Revert "fix: temporarily removes beforeSync callback (#919, partially reverts a6a7bcd). Making applySync async causes timing issues that leads to issues when using PermanentUserData. This can be reproduced by using the playground server with sqlite and a sleep(1000). Additionally, use Y.PermanentUserData, or try to read data from the ydoc in the onSynced event."

This reverts commit 7192664.

Author: janthurau

packages/server/src/Connection.ts

@@ -9,11 +9,7 @@ import type Document from "./Document.ts";
 import { IncomingMessage } from "./IncomingMessage.ts";
 import { MessageReceiver } from "./MessageReceiver.ts";
 import { OutgoingMessage } from "./OutgoingMessage.ts";
-import type {
-	beforeSyncPayload,
-	onStatelessPayload,
-	onTokenSyncPayload,
-} from "./types.ts";
+import type { beforeSyncPayload, onTokenSyncPayload, onStatelessPayload } from "./types.ts";
 
 export class Connection {
 	webSocket: WebSocket;
@@ -33,8 +29,7 @@ export class Connection {
 			payload: Pick<beforeSyncPayload, "type" | "payload">,
 		) => Promise.resolve(),
 		statelessCallback: (payload: onStatelessPayload) => Promise.resolve(),
-		onTokenSyncCallback: (payload: Partial<onTokenSyncPayload>) =>
-			Promise.resolve(),
+		onTokenSyncCallback: (payload: Partial<onTokenSyncPayload>) => Promise.resolve(),
 	};
 
 	socketId: string;
@@ -112,7 +107,7 @@ export class Connection {
 		return this;
 	}
 
-	/**
+  /**
 	 * Set a callback that will be triggered when on token sync message is received
 	 */
 	onTokenSyncCallback(
@@ -159,9 +154,7 @@ export class Connection {
 	 * Request current token from the client
 	 */
 	public requestToken(): void {
-		const message = new OutgoingMessage(
-			this.document.name,
-		).writeTokenSyncRequest();
+		const message = new OutgoingMessage(this.document.name).writeTokenSyncRequest();
 
 		this.send(message.toUint8Array());
 	}
@@ -216,18 +209,18 @@ export class Connection {
 		this.callbacks
 			.beforeHandleMessage(this, data)
 			.then(() => {
-				try {
-					new MessageReceiver(message).apply(this.document, this);
-				} catch (e) {
-					console.error(
-						`closing connection ${this.socketId} (while handling ${documentName}) because of exception`,
-						e,
-					);
-					this.close({
-						code: "code" in e ? e.code : ResetConnection.code,
-						reason: "reason" in e ? e.reason : ResetConnection.reason,
+				new MessageReceiver(message)
+					.apply(this.document, this)
+					.catch((e: any) => {
+						console.error(
+							`closing connection ${this.socketId} (while handling ${documentName}) because of exception`,
+							e,
+						);
+						this.close({
+							code: "code" in e ? e.code : ResetConnection.code,
+							reason: "reason" in e ? e.reason : ResetConnection.reason,
+						});
 					});
-				}
 			})
 			.catch((e: any) => {
 				console.error(

packages/server/src/MessageReceiver.ts

@@ -1,4 +1,3 @@
-import { AuthMessageType } from "@hocuspocus/common";
 import * as decoding from "lib0/decoding";
 import { readVarString } from "lib0/decoding";
 import { applyAwarenessUpdate } from "y-protocols/awareness";
@@ -16,6 +15,7 @@ import type Document from "./Document.ts";
 import type { IncomingMessage } from "./IncomingMessage.ts";
 import { OutgoingMessage } from "./OutgoingMessage.ts";
 import { MessageType } from "./types.ts";
+import { AuthMessageType } from "@hocuspocus/common";
 
 export class MessageReceiver {
 	message: IncomingMessage;
@@ -27,7 +27,7 @@ export class MessageReceiver {
 		this.defaultTransactionOrigin = defaultTransactionOrigin;
 	}
 
-	public apply(
+	public async apply(
 		document: Document,
 		connection?: Connection,
 		reply?: (message: Uint8Array) => void,
@@ -40,7 +40,7 @@ export class MessageReceiver {
 			case MessageType.Sync:
 			case MessageType.SyncReply: {
 				message.writeVarUint(MessageType.Sync);
-				this.readSyncMessage(
+				await this.readSyncMessage(
 					message,
 					document,
 					connection,
@@ -65,7 +65,7 @@ export class MessageReceiver {
 				break;
 			}
 			case MessageType.Awareness: {
-				applyAwarenessUpdate(
+				await applyAwarenessUpdate(
 					document.awareness,
 					message.readVarUint8Array(),
 					connection?.webSocket,
@@ -111,7 +111,7 @@ export class MessageReceiver {
 						token: message.readVarString(),
 					});
 					break;
-				}
+  				}
 				console.error(
 					"Received an authentication message on a connection that is already fully authenticated. Probably your provider has been destroyed + recreated really fast.",
 				);
@@ -126,7 +126,7 @@ export class MessageReceiver {
 		}
 	}
 
-	readSyncMessage(
+	async readSyncMessage(
 		message: IncomingMessage,
 		document: Document,
 		connection?: Connection,
@@ -136,7 +136,7 @@ export class MessageReceiver {
 		const type = message.readVarUint();
 
 		if (connection) {
-			connection.callbacks.beforeSync(connection, {
+			await connection.callbacks.beforeSync(connection, {
 				type,
 				payload: message.peekVarUint8Array(),
 			});