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.

Author: janthurau

packages/server/src/Connection.ts

@@ -9,7 +9,11 @@ import type Document from "./Document.ts";
 import { IncomingMessage } from "./IncomingMessage.ts";
 import { MessageReceiver } from "./MessageReceiver.ts";
 import { OutgoingMessage } from "./OutgoingMessage.ts";
-import type { beforeSyncPayload, onTokenSyncPayload, onStatelessPayload } from "./types.ts";
+import type {
+	beforeSyncPayload,
+	onStatelessPayload,
+	onTokenSyncPayload,
+} from "./types.ts";
 
 export class Connection {
 	webSocket: WebSocket;
@@ -29,7 +33,8 @@ 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;
@@ -107,7 +112,7 @@ export class Connection {
 		return this;
 	}
 
-  /**
+	/**
 	 * Set a callback that will be triggered when on token sync message is received
 	 */
 	onTokenSyncCallback(
@@ -154,7 +159,9 @@ 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());
 	}
@@ -209,18 +216,18 @@ export class Connection {
 		this.callbacks
 			.beforeHandleMessage(this, data)
 			.then(() => {
-				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,
-						});
+				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,
 					});
+				}
 			})
 			.catch((e: any) => {
 				console.error(

packages/server/src/MessageReceiver.ts

@@ -1,3 +1,4 @@
+import { AuthMessageType } from "@hocuspocus/common";
 import * as decoding from "lib0/decoding";
 import { readVarString } from "lib0/decoding";
 import { applyAwarenessUpdate } from "y-protocols/awareness";
@@ -15,7 +16,6 @@ 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 async apply(
+	public 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);
-				await this.readSyncMessage(
+				this.readSyncMessage(
 					message,
 					document,
 					connection,
@@ -65,7 +65,7 @@ export class MessageReceiver {
 				break;
 			}
 			case MessageType.Awareness: {
-				await applyAwarenessUpdate(
+				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 {
 		}
 	}
 
-	async readSyncMessage(
+	readSyncMessage(
 		message: IncomingMessage,
 		document: Document,
 		connection?: Connection,
@@ -136,7 +136,7 @@ export class MessageReceiver {
 		const type = message.readVarUint();
 
 		if (connection) {
-			await connection.callbacks.beforeSync(connection, {
+			connection.callbacks.beforeSync(connection, {
 				type,
 				payload: message.peekVarUint8Array(),
 			});