agw/overleaf-cep
1
0
Fork 0
mirror of https://github.com/yu-i-i/overleaf-cep.git synced 2026-11-12 13:13:45 +01:00
Code Issues Packages Projects Releases Wiki Activity

[monorepo] prepare for per-project pub/sub for applied ops (#35663)

Browse Source 
GitOrigin-RevId: 11d45addf366c3449cad432272ed03d4c8dae49d
This commit is contained in:
Jakob Ackermann
2026-09-17 11:33:01 +00:00
committed by Copybot
parent 1b2da240d5
commit 3565f51a84
13 changed files with 723 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/real-time/app/js/DocumentUpdaterController.js
+53 -28
View File
@@ -66,42 +66,61 @@ export default DocumentUpdaterController = {
logger.error({ err: error, channel }, 'error parsing JSON')
return
}
if (message.op) {
if (message._id && settings.checkEventOrder) {
const status = EventLogger.checkEventOrder(
'applied-ops',
message._id,
message
)
if (status === 'duplicate') {
return // skip duplicate events
}
}
DocumentUpdaterController._applyUpdateFromDocumentUpdater(
io,
message.doc_id,
message.op
)
} else if (message.error) {
DocumentUpdaterController._processErrorFromDocumentUpdater(
io,
message.doc_id,
message.error,
message
)
} else if (message.health_check) {
if (message.health_check) {
logger.debug(
{ message },
'got health check message in applied ops channel'
)
HealthCheckManager.check(channel, message.key)
return
}
// Messages on the legacy applied-ops channel are broadcast to the doc
// room, matching the per-doc channel granularity.
DocumentUpdaterController.handleAppliedOpMessage(
io,
message,
message.doc_id
)
})
},
_applyUpdateFromDocumentUpdater(io, docId, update) {
// Handle an already-parsed message from document-updater (an applied op
// or an error) and broadcast it to the given room: the doc room for
// messages from the legacy per-doc applied-ops channel, the project room
// for messages forwarded from the editor-events channel by the
// WebsocketLoadBalancer.
handleAppliedOpMessage(io, message, roomId) {
if (message.op) {
if (message._id && settings.checkEventOrder) {
const status = EventLogger.checkEventOrder(
'applied-ops',
message._id,
message
)
if (status === 'duplicate') {
return // skip duplicate events
}
}
DocumentUpdaterController._applyUpdateFromDocumentUpdater(
io,
roomId,
message.doc_id,
message.op
)
} else if (message.error) {
DocumentUpdaterController._processErrorFromDocumentUpdater(
io,
roomId,
message.doc_id,
message.error,
message
)
}
},
_applyUpdateFromDocumentUpdater(io, roomId, docId, update) {
let client
const clientList = io.sockets.clients(docId)
const clientList = io.sockets.clients(roomId)
// avoid unnecessary work if no clients are connected
if (clientList.length === 0) {
return
@@ -172,8 +191,14 @@ export default DocumentUpdaterController = {
}
},
_processErrorFromDocumentUpdater(io, docId, error, message) {
for (const client of io.sockets.clients(docId)) {
_processErrorFromDocumentUpdater(io, roomId, docId, error, message) {
for (const client of io.sockets.clients(roomId)) {
// When broadcasting via the project room, only clients that have
// joined the doc are affected by the error. (Always true for clients
// in the doc room.)
if (client.ol_context?.[`doc:${docId}`] !== 'allowed') {
continue
}
logger.warn(
{ err: error, docId, clientId: client.id },
'error from document updater, disconnecting client'
services/real-time/app/js/WebsocketLoadBalancer.js
+23
View File
@@ -8,6 +8,7 @@ import HealthCheckManager from './HealthCheckManager.js'
import RoomManager from './RoomManager.js'
import ChannelManager from './ChannelManager.js'
import ConnectedUsersManager from './ConnectedUsersManager.js'
import DocumentUpdaterController from './DocumentUpdaterController.js'
const RESTRICTED_USER_MESSAGE_TYPE_PASS_LIST = [
'otUpdateApplied',
@@ -171,6 +172,28 @@ export default WebsocketLoadBalancer = {
estimateBandwidth(projectId, 'per-project')
estimateBandwidth(docId, 'per-doc')
} else if (
message.message === 'otUpdateApplied' ||
message.message === 'otUpdateError'
) {
// An applied op (or error) from document-updater, published on the
// per-project editor-events channel instead of the legacy per-doc
// applied-ops channel. Forward it for broadcasting to the project
// room.
const status = message.message === 'otUpdateApplied' ? 'op' : 'error'
Metrics.inc('applied-ops-via-editor-events', 1, { status })
if (!message.project_id) {
logger.error(
{ channel, message },
'applied-ops message without project_id'
)
} else {
DocumentUpdaterController.handleAppliedOpMessage(
io,
message,
message.project_id
)
}
} else if (message.room_id) {
if (message._id && Settings.checkEventOrder) {
const status = EventLogger.checkEventOrder(
services/real-time/test/acceptance/js/MatrixTests.js
+28
View File
@@ -389,6 +389,34 @@ describe('MatrixTests', function () {
})
})
describe('receive updates via editor-events', function () {
const receivedMessages = []
beforeEach(function publishAnUpdateInRedis(done) {
const update = {
project_id: privateProjectId,
doc_id: privateDocId,
message: 'otUpdateApplied',
op: {
meta: { source: privateClient.publicId },
v: 42,
doc: privateDocId,
op: [{ i: 'foo', p: 50 }],
},
}
client.on('otUpdateApplied', update => {
receivedMessages.push(update)
})
privateClient.once('otUpdateApplied', () => {
setTimeout(done, 10)
})
rclient.publish('editor-events', JSON.stringify(update))
})
it('should send nothing to client', function () {
expect(receivedMessages).to.have.length(0)
})
})
describe('receive messages from web', function () {
const receivedMessages = []
beforeEach(function publishAMessageInRedis(done) {
services/real-time/test/acceptance/js/ReceiveUpdateTests.js
+128 -1
View File
@@ -69,6 +69,11 @@ describe('receiveUpdate', function () {
return this.clientB.emit('joinDoc', this.doc_id, cb)
},
cb => {
// clientD is in the first project, but has not joined the doc
this.clientD = RealTimeClient.connect(this.project_id, cb)
},
cb => {
return FixturesManager.setUpProject(
{
@@ -116,6 +121,10 @@ describe('receiveUpdate', function () {
this.clientC.on('otUpdateApplied', update =>
this.clientCUpdates.push(update)
)
this.clientDUpdates = []
this.clientD.on('otUpdateApplied', update =>
this.clientDUpdates.push(update)
)
this.clientAErrors = []
this.clientA.on('otUpdateError', error =>
@@ -129,6 +138,10 @@ describe('receiveUpdate', function () {
this.clientC.on('otUpdateError', error =>
this.clientCErrors.push(error)
)
this.clientDErrors = []
this.clientD.on('otUpdateError', error =>
this.clientDErrors.push(error)
)
return cb()
},
],
@@ -143,7 +156,10 @@ describe('receiveUpdate', function () {
if (this.clientB != null) {
this.clientB.disconnect()
}
return this.clientC != null ? this.clientC.disconnect() : undefined
if (this.clientC != null) {
this.clientC.disconnect()
}
return this.clientD != null ? this.clientD.disconnect() : undefined
})
describe('with an update from clientA', function () {
@@ -246,6 +262,117 @@ describe('receiveUpdate', function () {
})
})
describe('with an update from clientA published on the editor-events channel', function () {
beforeEach(function (done) {
this.update = {
project_id: this.project_id,
doc_id: this.doc_id,
message: 'otUpdateApplied',
op: {
meta: {
source: this.clientA.publicId,
},
v: this.version,
doc: this.doc_id,
op: [{ i: 'foo', p: 50 }],
},
}
rclient.publish(
`editor-events:${this.project_id}`,
JSON.stringify(this.update)
)
setTimeout(done, 200)
}) // Give clients time to get message
it('should send the full op to clientB', function () {
this.clientBUpdates.should.deep.equal([this.update.op])
})
it('should send an ack to clientA', function () {
this.clientAUpdates.should.deep.equal([
{
v: this.version,
doc: this.doc_id,
},
])
})
it('should send the full op to clientD, who has not joined the doc', function () {
this.clientDUpdates.should.deep.equal([this.update.op])
})
it('should send nothing to clientC', function () {
this.clientCUpdates.should.deep.equal([])
})
})
describe('with an update from a remote client published on the base editor-events channel', function () {
// producers with PUBLISH_ON_INDIVIDUAL_CHANNELS disabled publish on the
// base channel instead of the per-project channel
beforeEach(function (done) {
this.update = {
project_id: this.project_id,
doc_id: this.doc_id,
message: 'otUpdateApplied',
op: {
meta: {
source: 'this-is-a-remote-client-id',
},
v: this.version,
doc: this.doc_id,
op: [{ i: 'foo', p: 50 }],
},
}
rclient.publish('editor-events', JSON.stringify(this.update))
setTimeout(done, 200)
}) // Give clients time to get message
it('should send the full op to the clients in the first project', function () {
this.clientAUpdates.should.deep.equal([this.update.op])
this.clientBUpdates.should.deep.equal([this.update.op])
this.clientDUpdates.should.deep.equal([this.update.op])
})
it('should send nothing to clientC', function () {
this.clientCUpdates.should.deep.equal([])
})
})
describe('with an error for the first project published on the editor-events channel', function () {
beforeEach(function (done) {
rclient.publish(
`editor-events:${this.project_id}`,
JSON.stringify({
project_id: this.project_id,
doc_id: this.doc_id,
message: 'otUpdateError',
error: (this.error = 'something went wrong'),
})
)
setTimeout(done, 200)
}) // Give clients time to get message
it('should send the error to the clients that joined the doc', function () {
this.clientAErrors.should.deep.equal([this.error])
this.clientBErrors.should.deep.equal([this.error])
})
it('should disconnect the clients that joined the doc', function () {
this.clientA.socket.connected.should.equal(false)
this.clientB.socket.connected.should.equal(false)
})
it('should not send the error to clientD, who has not joined the doc', function () {
this.clientDErrors.should.deep.equal([])
this.clientD.socket.connected.should.equal(true)
})
it('should not send any errors to the client in the second project', function () {
this.clientCErrors.should.deep.equal([])
this.clientC.socket.connected.should.equal(true)
})
})
describe('with an error for the first project', function () {
beforeEach(function (done) {
rclient.publish(
services/real-time/test/unit/js/DocumentUpdaterController.test.js
+166 -17
View File
@@ -138,9 +138,9 @@ describe('DocumentUpdaterController', function () {
)
})
it('should apply the update', function (ctx) {
it('should apply the update to the doc room', function (ctx) {
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater
.calledWith(ctx.io, ctx.doc_id, ctx.message.op)
.calledWith(ctx.io, ctx.doc_id, ctx.doc_id, ctx.message.op)
.should.equal(true)
})
})
@@ -160,9 +160,89 @@ describe('DocumentUpdaterController', function () {
)
})
it('should process the error', function (ctx) {
it('should process the error in the doc room', function (ctx) {
ctx.EditorUpdatesController._processErrorFromDocumentUpdater
.calledWith(ctx.io, ctx.doc_id, ctx.message.error)
.calledWith(ctx.io, ctx.doc_id, ctx.doc_id, ctx.message.error)
.should.equal(true)
})
})
})
describe('handleAppliedOpMessage', function () {
describe('with an update forwarded from the editor-events channel', function () {
beforeEach(function (ctx) {
ctx.message = {
project_id: ctx.project_id,
doc_id: ctx.doc_id,
op: { t: 'foo', p: 12 },
}
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater =
sinon.stub()
ctx.EditorUpdatesController.handleAppliedOpMessage(
ctx.io,
ctx.message,
ctx.project_id
)
})
it('should apply the update to the project room', function (ctx) {
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater
.calledWith(ctx.io, ctx.project_id, ctx.doc_id, ctx.message.op)
.should.equal(true)
})
})
describe('with a duplicate update', function () {
beforeEach(function (ctx) {
ctx.settings.checkEventOrder = true
ctx.message = {
project_id: ctx.project_id,
doc_id: ctx.doc_id,
op: { t: 'foo', p: 12 },
_id: 'doc:host:rnd-1',
}
ctx.EventLogger.checkEventOrder.returns('duplicate')
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater =
sinon.stub()
ctx.EditorUpdatesController.handleAppliedOpMessage(
ctx.io,
ctx.message,
ctx.project_id
)
})
it('should check the event order on the applied-ops channel', function (ctx) {
ctx.EventLogger.checkEventOrder
.calledWith('applied-ops', ctx.message._id, ctx.message)
.should.equal(true)
})
it('should skip the update', function (ctx) {
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater.called.should.equal(
false
)
})
})
describe('with an error forwarded from the editor-events channel', function () {
beforeEach(function (ctx) {
ctx.message = {
project_id: ctx.project_id,
doc_id: ctx.doc_id,
error: 'Something went wrong',
}
ctx.EditorUpdatesController._processErrorFromDocumentUpdater =
sinon.stub()
ctx.EditorUpdatesController.handleAppliedOpMessage(
ctx.io,
ctx.message,
ctx.project_id
)
})
it('should process the error in the project room', function (ctx) {
ctx.EditorUpdatesController._processErrorFromDocumentUpdater
.calledWith(ctx.io, ctx.project_id, ctx.doc_id, ctx.message.error)
.should.equal(true)
})
})
@@ -194,6 +274,7 @@ describe('DocumentUpdaterController', function () {
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater(
ctx.io,
ctx.doc_id,
ctx.doc_id,
ctx.update
)
})
@@ -224,6 +305,7 @@ describe('DocumentUpdaterController', function () {
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater(
ctx.io,
ctx.doc_id,
ctx.doc_id,
ctx.update
)
})
@@ -240,28 +322,95 @@ describe('DocumentUpdaterController', function () {
)
})
})
describe('when broadcasting to the project room', function () {
beforeEach(function (ctx) {
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater(
ctx.io,
ctx.project_id,
ctx.doc_id,
ctx.update
)
})
it('should get the clients connected to the project', function (ctx) {
ctx.io.sockets.clients.calledWith(ctx.project_id).should.equal(true)
})
it('should send a version bump to the source client', function (ctx) {
ctx.sourceClient.emit
.calledWith('otUpdateApplied', { v: ctx.version, doc: ctx.doc_id })
.should.equal(true)
ctx.sourceClient.emit.calledOnce.should.equal(true)
})
it('should send the full update to the other clients', function (ctx) {
Array.from(ctx.otherClients).map(client =>
client.emit
.calledWith('otUpdateApplied', ctx.update)
.should.equal(true)
)
})
})
})
describe('_processErrorFromDocumentUpdater', function () {
beforeEach(function (ctx) {
ctx.clients = [new MockClient(), new MockClient()]
ctx.io.sockets = { clients: sinon.stub().returns(ctx.clients) }
ctx.EditorUpdatesController._processErrorFromDocumentUpdater(
ctx.io,
ctx.doc_id,
'Something went wrong'
)
for (const client of ctx.clients) {
client.ol_context[`doc:${ctx.doc_id}`] = 'allowed'
}
})
it('should log a warning', function (ctx) {
ctx.logger.warn.called.should.equal(true)
describe('in the doc room', function () {
beforeEach(function (ctx) {
ctx.io.sockets = { clients: sinon.stub().returns(ctx.clients) }
ctx.EditorUpdatesController._processErrorFromDocumentUpdater(
ctx.io,
ctx.doc_id,
ctx.doc_id,
'Something went wrong'
)
})
it('should log a warning', function (ctx) {
ctx.logger.warn.called.should.equal(true)
})
it('should disconnect all clients in that document', function (ctx) {
ctx.io.sockets.clients.calledWith(ctx.doc_id).should.equal(true)
Array.from(ctx.clients).map(client =>
client.disconnect.called.should.equal(true)
)
})
})
it('should disconnect all clients in that document', function (ctx) {
ctx.io.sockets.clients.calledWith(ctx.doc_id).should.equal(true)
Array.from(ctx.clients).map(client =>
client.disconnect.called.should.equal(true)
)
describe('in the project room', function () {
beforeEach(function (ctx) {
ctx.otherDocClient = new MockClient()
ctx.otherDocClient.ol_context['doc:other-doc-id'] = 'allowed'
ctx.io.sockets = {
clients: sinon.stub().returns([...ctx.clients, ctx.otherDocClient]),
}
ctx.EditorUpdatesController._processErrorFromDocumentUpdater(
ctx.io,
ctx.project_id,
ctx.doc_id,
'Something went wrong'
)
})
it('should disconnect the clients that joined the doc', function (ctx) {
ctx.io.sockets.clients.calledWith(ctx.project_id).should.equal(true)
Array.from(ctx.clients).map(client =>
client.disconnect.called.should.equal(true)
)
})
it('should not disconnect clients that did not join the doc', function (ctx) {
ctx.otherDocClient.emit.called.should.equal(false)
ctx.otherDocClient.disconnect.called.should.equal(false)
})
})
})
})
services/real-time/test/unit/js/WebsocketLoadBalancer.test.js
+92 -1
View File
@@ -53,6 +53,12 @@ describe('WebsocketLoadBalancer', function () {
}),
}))
vi.doMock('../../../app/js/DocumentUpdaterController', () => ({
default: (ctx.DocumentUpdaterController = {
handleAppliedOpMessage: sinon.stub(),
}),
}))
ctx.WebsocketLoadBalancer = (await import(modulePath)).default
ctx.io = {}
ctx.WebsocketLoadBalancer.rclientPubList = [{ publish: sinon.stub() }]
@@ -64,7 +70,9 @@ describe('WebsocketLoadBalancer', function () {
]
ctx.room_id = 'room-id'
ctx.message = 'otUpdateApplied'
// any non-restricted message name (must be on the restricted-user message
// pass list so the restricted-client cases receive it)
ctx.message = 'projectNameUpdated'
ctx.payload = ['argument one', 42]
})
@@ -337,8 +345,91 @@ describe('WebsocketLoadBalancer', function () {
.should.equal(true)
ctx.emit3.called.should.equal(false)
})
it('should not forward the message to the DocumentUpdaterController', function (ctx) {
ctx.DocumentUpdaterController.handleAppliedOpMessage.called.should.equal(
false
)
})
}) // duplicate client should be ignored
describe('with an applied op from document-updater', function () {
beforeEach(function (ctx) {
ctx.appliedOpMessage = {
project_id: 'project-id-123',
doc_id: 'doc-id-123',
op: { v: 42, doc: 'doc-id-123' },
_id: 'doc:host:rnd-1',
message: 'otUpdateApplied',
}
ctx.io.sockets = { clients: sinon.stub().returns([]) }
ctx.WebsocketLoadBalancer._processEditorEvent(
ctx.io,
'editor-events:project-id-123',
JSON.stringify(ctx.appliedOpMessage)
)
})
it('should forward the message for broadcasting to the project room', function (ctx) {
ctx.DocumentUpdaterController.handleAppliedOpMessage
.calledWith(ctx.io, ctx.appliedOpMessage, 'project-id-123')
.should.equal(true)
})
it('should not emit the message as an editor event', function (ctx) {
ctx.io.sockets.clients.called.should.equal(false)
})
})
describe('with an applied-op error from document-updater', function () {
beforeEach(function (ctx) {
ctx.appliedOpMessage = {
project_id: 'project-id-123',
doc_id: 'doc-id-123',
error: 'Something went wrong',
message: 'otUpdateError',
}
ctx.io.sockets = { clients: sinon.stub().returns([]) }
ctx.WebsocketLoadBalancer._processEditorEvent(
ctx.io,
'editor-events:project-id-123',
JSON.stringify(ctx.appliedOpMessage)
)
})
it('should forward the message for broadcasting to the project room', function (ctx) {
ctx.DocumentUpdaterController.handleAppliedOpMessage
.calledWith(ctx.io, ctx.appliedOpMessage, 'project-id-123')
.should.equal(true)
})
})
describe('with an applied op without a project_id', function () {
beforeEach(function (ctx) {
ctx.appliedOpMessage = {
doc_id: 'doc-id-123',
op: { v: 42, doc: 'doc-id-123' },
message: 'otUpdateApplied',
}
ctx.io.sockets = { clients: sinon.stub().returns([]) }
ctx.WebsocketLoadBalancer._processEditorEvent(
ctx.io,
'editor-events',
JSON.stringify(ctx.appliedOpMessage)
)
})
it('should log an error', function (ctx) {
ctx.logger.error.called.should.equal(true)
})
it('should not forward the message', function (ctx) {
ctx.DocumentUpdaterController.handleAppliedOpMessage.called.should.equal(
false
)
})
})
describe('with a designated room, and restricted clients, not restricted message', function () {
beforeEach(function (ctx) {
ctx.io.sockets = {