Ryan/docmost
1
0
Fork 0
mirror of https://github.com/docmost/docmost.git synced 2026-08-24 19:52:14 +10:00
Code Issues Packages Projects Releases Wiki Activity

feat: collab hocuspocus v4 upgrade (#2351)

Browse Source 
* WIP 1

* complete v4 migration

* add flushDelay

* feat: multiplexing

* fix: survive hocuspocus v4 message timeout

* fix: searchTerm type error
This commit is contained in:
Philip Okugbe
2026-07-20 20:34:09 +01:00
committed by GitHub
parent ce8fbb86ff
commit a55057db37
13 changed files with 876 additions and 787 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/src/collaboration/extensions/redis-sync/redis-sync.extension.ts
+116 -66
View File
@@ -3,27 +3,30 @@ import {
Extension,
Hocuspocus,
IncomingMessage,
afterUnloadDocumentPayload,
onConfigurePayload,
onLoadDocumentPayload,
afterUnloadDocumentPayload,
WebSocketLike,
} from '@hocuspocus/server';
import { ConnectionTimeout, Unauthorized } from '@hocuspocus/common';
import RedisClient from 'ioredis';
import { readVarString } from 'lib0/decoding.js';
import { CollabProxySocket } from './collab-proxy-socket';
import {
BaseWebSocket,
Configuration,
CustomEvents,
Pack,
RSAMessage,
RSAMessageClose,
RSAMessageCloseProxy,
RSAMessageCustomEventComplete,
RSAMessageCustomEventStart,
RSAMessagePong,
RSAMessageProxy,
RSAMessageUnload,
SerializedHTTPRequest,
Unpack,
OriginConnection,
ProxyConnection,
toWebRequest,
} from './redis-sync.types';
export type { Pack, SerializedHTTPRequest } from './redis-sync.types';
@@ -38,10 +41,10 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
private sub: RedisClient;
private readonly pack: Pack;
private readonly unpack: Unpack;
private originSockets: Record<SocketId, BaseWebSocket> = {};
private originConnections: Record<SocketId, OriginConnection> = {};
private locks: Record<DocumentName, NodeJS.Timeout> = {};
private lockPromises: Record<DocumentName, Promise<ServerId | null>> = {};
private proxySockets: Record<SocketId, CollabProxySocket> = {};
private proxyConnections: Record<SocketId, ProxyConnection> = {};
private readonly prefix: string;
private readonly lockPrefix: string;
private readonly msgChannel: string;
@@ -54,6 +57,9 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
// @ts-ignore
private pendingReplies: Record<number, PromiseWithResolvers<any>['resolve']> =
{};
private deriveContext: (
serializedHTTPRequest: SerializedHTTPRequest,
) => Record<string, any>;
constructor(configuration: Configuration<TCE>) {
const {
@@ -65,6 +71,7 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
prefix,
customEvents,
customEventTTL,
deriveContext,
} = configuration;
this.pub = redis.duplicate();
this.sub = redis.duplicate();
@@ -77,6 +84,7 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
this.lockPrefix = `${this.prefix}Lock`;
this.msgChannel = `${this.prefix}Msg`;
this.customEvents = (customEvents as any) ?? ({} as any as CustomEvents);
this.deriveContext = deriveContext ?? (() => ({}));
this.sub.subscribe(this.msgChannel, `${this.msgChannel}:${this.serverId}`);
this.sub.on('messageBuffer', this.handleRedisMessage);
this.pub.on('error', () => {});
@@ -87,44 +95,63 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
}
private closeProxy(socketId: string) {
const proxySocket = this.proxySockets[socketId];
if (proxySocket) {
proxySocket.emit(
'close',
1000,
Buffer.from('provider_initiated', 'utf-8'),
);
delete this.proxySockets[socketId];
const entry = this.proxyConnections[socketId];
if (entry) {
delete this.proxyConnections[socketId];
const { socket, clientConnection } = entry;
// The origin socket is already gone; don't echo a close message back
socket.markClosed();
clientConnection.handleClose({
code: 1000,
reason: 'provider_initiated',
});
}
}
private pongProxy(socketId: string) {
this.proxySockets[socketId]?.emit('pong');
}
private handleProxyMessage(
msg: Pick<RSAMessageProxy, 'replyTo' | 'message' | 'serializedHTTPRequest'>,
) {
const { replyTo, message, serializedHTTPRequest } = msg;
const { headers } = serializedHTTPRequest;
const socketId = headers['sec-websocket-key']!;
let socket = this.proxySockets[socketId];
if (!socket) {
socket = new CollabProxySocket(
const socketId = headers['sec-websocket-key'];
let entry = this.proxyConnections[socketId];
if (!entry) {
const socket = new CollabProxySocket(
this.pub,
this.pack,
replyTo,
`${this.msgChannel}:${this.serverId}`,
socketId,
);
this.proxySockets[socketId] = socket;
this.instance.handleConnection(
socket as any,
serializedHTTPRequest as any,
{},
// A proxy connection with no live documents (client left the page, auth
// failed, or the origin server crashed) is reaped by hocuspocus' message
// timeout. Dispose it silently in that case: relaying the timeout close
// to the origin would kill the client's real socket, which may be busy
// serving other documents. Genuine protocol closes are still relayed.
socket.onClose = (code, reason) => {
delete this.proxyConnections[socketId];
if (code !== ConnectionTimeout.code) {
const msg: RSAMessageClose = {
type: 'close',
code,
reason,
socketId,
};
this.pub.publish(replyTo, this.pack(msg));
}
};
const clientConnection = this.instance.handleConnection(
socket,
toWebRequest(serializedHTTPRequest),
this.deriveContext(serializedHTTPRequest),
);
entry = { clientConnection, socket };
this.proxyConnections[socketId] = entry;
}
socket.emit('message', message);
entry.clientConnection.handleMessage(message);
}
private getLock(documentName: string) {
return this.pub.get(this.getKey(documentName));
}
private getOrClaimLock(documentName: string) {
@@ -166,10 +193,6 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
this.closeProxy(msg.socketId);
return;
}
if (type === 'pong') {
this.pongProxy(msg.socketId);
return;
}
if (type === 'unload') {
delete this.lockPromises[msg.documentName];
return;
@@ -198,22 +221,14 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
return;
}
const { socketId } = msg;
const socket = this.originSockets[socketId];
if (!socket) {
const entry = this.originConnections[socketId];
if (!entry) {
// origin socket already cleaned up
return;
}
const { socket } = entry;
if (type === 'close') {
socket.close(msg.code, msg.reason);
} else if (type === 'ping') {
// Reply instantly to the proxy socket, without forwarding to client
// The origin socket handles heartbeat for itself
const { replyTo, socketId } = msg;
const reply: RSAMessagePong = {
type: 'pong',
socketId,
};
this.pub.publish(`${replyTo}`, this.pack(reply));
} else if (type === 'send') {
socket.send(msg.message);
}
@@ -251,6 +266,8 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
eventName: TName,
documentName: string,
payload: any,
// if true, don't claim the lock. Useful for targeting pages that are currently open
onlyIfOpen = false,
) {
const isDocLoadedOnInstance = this.instance.documents.has(documentName);
@@ -258,7 +275,14 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
return this.handleEventLocally(eventName, documentName, payload);
}
const proxyTo = await this.getOrClaimLockThrottled(documentName);
const proxyTo = await (onlyIfOpen
? this.getLock(documentName)
: this.getOrClaimLockThrottled(documentName));
if (!proxyTo && onlyIfOpen) {
return;
}
if (proxyTo && proxyTo !== this.serverId) {
++this.replyIdCounter; // bug in biome thinks this.replyIdCounter is not used if written on the line below
const replyId = this.replyIdCounter;
@@ -277,7 +301,8 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
const { promise, resolve, reject } = Promise.withResolvers();
this.pendingReplies[replyId] = resolve;
setTimeout(() => {
reject('TIMEOUT');
delete this.pendingReplies[replyId];
reject(new Error('TIMEOUT'));
}, this.customEventTTL);
return promise as Promise<ReturnType<TCE[TName]>>;
}
@@ -296,36 +321,59 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
/* WebSocket Server Hooks */
onSocketOpen(
ws: BaseWebSocket,
ws: WebSocketLike,
serializedHTTPRequest: SerializedHTTPRequest,
context = {},
) {
const socketId = serializedHTTPRequest.headers['sec-websocket-key']!;
this.originSockets[socketId] = ws;
this.instance.handleConnection(
ws as any,
serializedHTTPRequest as any,
context,
const socketId = serializedHTTPRequest.headers['sec-websocket-key'];
const clientConnection = this.instance.handleConnection(
ws,
toWebRequest(serializedHTTPRequest),
this.deriveContext(serializedHTTPRequest),
);
this.originConnections[socketId] = { clientConnection, socket: ws };
}
async onSocketMessage(
ws: BaseWebSocket,
serializedHTTPRequest: SerializedHTTPRequest,
detachableMsg: ArrayBuffer,
) {
const message = new Uint8Array(detachableMsg.slice());
const tmpMsg = new IncomingMessage(detachableMsg);
const documentName = readVarString(tmpMsg.decoder);
const socketId = serializedHTTPRequest.headers['sec-websocket-key'];
const entry = this.originConnections[socketId];
if (!entry) return;
const { clientConnection } = entry;
let message: Uint8Array;
let documentName: string;
try {
message = new Uint8Array(detachableMsg.slice());
const tmpMsg = new IncomingMessage(detachableMsg);
const documentNameAndSessionId = tmpMsg.readVarString();
// session-aware providers suffix the documentName with \0sessionId
const sepIdx = documentNameAndSessionId.indexOf('\0');
documentName =
sepIdx === -1
? documentNameAndSessionId
: documentNameAndSessionId.slice(0, sepIdx);
} catch (error) {
entry.socket.close(Unauthorized.code, Unauthorized.reason);
return;
}
const isDocLoadedOnInstance = this.instance.documents.has(documentName);
if (isDocLoadedOnInstance) {
ws.emit('message', message);
clientConnection.handleMessage(message);
return;
}
const proxyTo = await this.getOrClaimLockThrottled(documentName);
if (proxyTo && proxyTo !== this.serverId) {
// Proxied messages bypass handleMessage, so refresh the connection's
// liveness fields manually or hocuspocus' message timeout would reap the
// real socket every `timeout` ms. connectionEstablishedAt is the
// reference while unauthenticated (auth for remote docs is proxied too)
// and is private upstream.
clientConnection.lastMessageReceivedAt = Date.now();
(clientConnection as any).connectionEstablishedAt = Date.now();
// another server owns the doc
const proxyMessage: RSAMessageProxy = {
serializedHTTPRequest: serializedHTTPRequest,
@@ -338,16 +386,17 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
return;
}
// This server owns the document, but hocuspocus hasn't loaded it yet
ws.emit('message', message);
clientConnection.handleMessage(message);
}
onSocketClose(socketId: string, code?: number, reason?: ArrayBuffer) {
const socket = this.originSockets[socketId];
if (!socket) return;
// at this point the socket is considered GC'd and we cannot call close
// The origin socket did not set up any connections for the proxy, so none of the hooks will work if we just emit
socket?.emit('close', code, reason);
delete this.originSockets[socketId];
const entry = this.originConnections[socketId];
if (!entry) return;
delete this.originConnections[socketId];
entry.clientConnection.handleClose({
code: code ?? 1000,
reason: reason ? Buffer.from(reason).toString() : '',
});
const msg: RSAMessageCloseProxy = { type: 'closeProxy', socketId };
this.pub.publish(this.msgChannel, this.pack(msg)).catch(() => {});
}
@@ -372,6 +421,7 @@ export class RedisSyncExtension<TCE extends CustomEvents> implements Extension {
}
async onDestroy() {
this.pendingReplies = {};
this.pub.disconnect(false);
this.sub.disconnect(false);
}