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

[real-time] remove per-doc rooms (#35691)

Browse Source 
* [real-time] remove applied-ops pub/sub

* [real-time] remove per-doc rooms

GitOrigin-RevId: ec3b76b37a029df15bf20d6f2832fb7c089c0042
This commit is contained in:
Jakob Ackermann
2026-09-17 11:34:11 +00:00
committed by Copybot
parent 2dee2b31d9
commit e714810632
20 changed files with 215 additions and 1408 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/real-time/app.js
+1 -4
View File
@@ -20,7 +20,6 @@ import socketIOClient from 'socket.io-client'
import http from 'node:http'
import Router from './app/js/Router.js'
import WebsocketLoadBalancer from './app/js/WebsocketLoadBalancer.js'
import DocumentUpdaterController from './app/js/DocumentUpdaterController.js'
logger.initialize('real-time')
Metrics.event_loop.monitor(logger)
@@ -196,8 +195,6 @@ Router.configure(app, io, sessionSockets)
WebsocketLoadBalancer.listenForEditorEvents(io)
DocumentUpdaterController.listenForUpdatesFromDocumentUpdater(io)
const { port } = Settings.internal.realTime
const { host } = Settings.internal.realTime
@@ -346,7 +343,7 @@ if (Settings.continualPubsubTraffic) {
}
const runPubSubTraffic = () =>
async.map(['applied-ops', 'editor-events'], publishJob, () =>
async.map(['editor-events'], publishJob, () =>
setTimeout(runPubSubTraffic, 1000 * 20)
)
services/real-time/app/js/DocumentUpdaterController.js
+5 -81
View File
@@ -1,94 +1,18 @@
import logger from '@overleaf/logger'
import settings from '@overleaf/settings'
import RedisClientManager from './RedisClientManager.js'
import SafeJsonParse from './SafeJsonParse.js'
import EventLogger from './EventLogger.js'
import HealthCheckManager from './HealthCheckManager.js'
import RoomManager from './RoomManager.js'
import ChannelManager from './ChannelManager.js'
import metrics from '@overleaf/metrics'
let DocumentUpdaterController
export default DocumentUpdaterController = {
// DocumentUpdaterController is responsible for updates that come via Redis
// Pub/Sub from the document updater.
rclientList: RedisClientManager.createClientList(settings.redis.pubsub),
listenForUpdatesFromDocumentUpdater(io) {
logger.debug(
{ rclients: this.rclientList.length },
'listening for applied-ops events'
)
for (const rclient of this.rclientList) {
rclient.subscribe('applied-ops')
rclient.on('message', function (channel, message) {
metrics.inc('rclient', 0.001) // global event rate metric
if (settings.debugEvents > 0) {
EventLogger.debugEvent(channel, message)
}
DocumentUpdaterController._processMessageFromDocumentUpdater(
io,
channel,
message
)
})
}
// create metrics for each redis instance only when we have multiple redis clients
if (this.rclientList.length > 1) {
this.rclientList.forEach((rclient, i) => {
// per client event rate metric
const metricName = `rclient-${i}`
rclient.on('message', () => metrics.inc(metricName, 0.001))
})
}
this.handleRoomUpdates(this.rclientList)
},
handleRoomUpdates(rclientSubList) {
const roomEvents = RoomManager.eventSource()
roomEvents.on('doc-active', function (docId) {
const subscribePromises = rclientSubList.map(rclient =>
ChannelManager.subscribe(rclient, 'applied-ops', docId)
)
RoomManager.emitOnCompletion(subscribePromises, `doc-subscribed-${docId}`)
})
roomEvents.on('doc-empty', docId =>
rclientSubList.map(rclient =>
ChannelManager.unsubscribe(rclient, 'applied-ops', docId)
)
)
},
_processMessageFromDocumentUpdater(io, channel, message) {
SafeJsonParse.parse(message, function (error, message) {
if (error) {
logger.error({ err: error, channel }, 'error parsing JSON')
return
}
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
)
})
},
// DocumentUpdaterController broadcasts applied ops (and errors) from
// document-updater to socket.io clients. The messages arrive on the
// per-project editor-events channel and are forwarded here by the
// WebsocketLoadBalancer.
// 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.
// or an error) and broadcast it to the project room (roomId = project_id).
handleAppliedOpMessage(io, message, roomId) {
if (message.op) {
if (message._id && settings.checkEventOrder) {
services/real-time/app/js/DocumentUpdaterManager.js
-13
View File
@@ -76,17 +76,6 @@ async function getDocument(projectId, docId, fromVersion) {
}
}
/**
* Check that a doc is loadable from the document-updater.
*
* @param {string} projectId
* @param {string} docId
*/
async function checkDocument(projectId, docId) {
// in this call fromVersion = -1 means get document without docOps
return await getDocument(projectId, docId, -1)
}
/**
* Ask the document-updater to flush a project to mongo and remove it from
* redis.
@@ -185,13 +174,11 @@ async function queueChange(projectId, docId, change) {
export default {
getDocument: callbackify(getDocument),
checkDocument: callbackify(checkDocument),
flushProjectToMongoAndDelete: callbackify(flushProjectToMongoAndDelete),
_getPendingUpdateListKey,
queueChange: callbackify(queueChange),
promises: {
getDocument,
checkDocument,
flushProjectToMongoAndDelete,
queueChange,
},
services/real-time/app/js/Errors.js
-7
View File
@@ -39,12 +39,6 @@ class DocumentUpdaterRequestFailedError extends OError {
}
}
class JoinLeaveEpochMismatchError extends OError {
constructor() {
super('joinLeaveEpoch mismatch')
}
}
class MissingSessionError extends OError {
constructor() {
super('could not look up session by key')
@@ -93,7 +87,6 @@ export default {
ClientRequestedMissingOpsError,
DataTooLargeToParseError,
DocumentUpdaterRequestFailedError,
JoinLeaveEpochMismatchError,
MissingSessionError,
NotAuthorizedError,
NotJoinedError,
services/real-time/app/js/RoomManager.js
+3 -13
View File
@@ -6,13 +6,11 @@ import OError from '@overleaf/o-error'
const IdMap = new Map() // keep track of whether ids are from projects or docs
const RoomEvents = new EventEmitter() // emits {project,doc}-active and {project,doc}-empty events
// Manage socket.io rooms for individual projects and docs
// Manage socket.io rooms for individual projects.
//
// The first time someone joins a project or doc we emit a 'project-active' or
// 'doc-active' event.
// The first time someone joins a project we emit a 'project-active' event.
//
// When the last person leaves a project or doc, we emit 'project-empty' or
// 'doc-empty' event.
// When the last person leaves a project, we emit a 'project-empty' event.
//
// The pubsub side is handled by ChannelManager
@@ -21,14 +19,6 @@ export default {
this.joinEntity(client, 'project', projectId, callback)
},
joinDoc(client, docId, callback) {
this.joinEntity(client, 'doc', docId, callback)
},
leaveDoc(client, docId) {
this.leaveEntity(client, 'doc', docId)
},
leaveProjectAndDocs(client) {
// what rooms is this client in? we need to leave them all. socket.io
// will cause us to leave the rooms, so we only need to manage our
services/real-time/app/js/Router.js
-3
View File
@@ -109,7 +109,6 @@ export default Router = {
} else if (
[
'not authorized',
'joinLeaveEpoch mismatch',
'doc updater could not load requested ops',
'no project_id found on client',
'cannot join multiple projects',
@@ -177,8 +176,6 @@ export default Router = {
// init client context, we may access it in Router._handleError before
// setting any values
client.ol_context = {}
// bail out from joinDoc when a parallel joinDoc or leaveDoc is running
client.joinLeaveEpoch = 0
if (client) {
client.on('error', function (err) {
services/real-time/app/js/WebsocketController.js
+97 -158
View File
@@ -12,7 +12,6 @@ import { EditOperationBuilder } from 'overleaf-editor-core'
const {
CodedError,
JoinLeaveEpochMismatchError,
NotAuthorizedError,
NotJoinedError,
ClientRequestedMissingOpsError,
@@ -205,7 +204,6 @@ export default WebsocketController = {
return callback()
}
const joinLeaveEpoch = ++client.joinLeaveEpoch
metrics.inc('editor.join-doc', 1, { status: client.transport })
const {
project_id: projectId,
@@ -252,185 +250,126 @@ export default WebsocketController = {
)
}
WebsocketController._assertClientAuthorization(
client,
docId,
function (error) {
if (error) {
return callback(error)
}
if (client.disconnected) {
metrics.inc('editor.join-doc.disconnected', 1, {
status: 'after-client-auth-check',
})
// the client will not read the response anyways
return callback()
}
if (joinLeaveEpoch !== client.joinLeaveEpoch) {
// another joinDoc or leaveDoc rpc overtook us
return callback(new JoinLeaveEpochMismatchError())
}
// ensure the per-doc applied-ops channel is subscribed before sending the
// doc to the client, so that no events are missed.
RoomManager.joinDoc(client, docId, function (error) {
AuthorizationManager.assertClientCanViewProject(client, function (error) {
if (error) {
return callback(error)
}
if (client.disconnected) {
metrics.inc('editor.join-doc.disconnected', 1, {
status: 'after-client-auth-check',
})
// the client will not read the response anyways
return callback()
}
DocumentUpdaterManager.getDocument(
projectId,
docId,
fromVersion,
function (error, result) {
if (error) {
if (error instanceof ClientRequestedMissingOpsError) {
emitJoinDocCatchUpMetrics('missing', error.info)
}
return callback(error)
}
const { lines, version, ranges, ops, ttlInS, type } = result
emitJoinDocCatchUpMetrics('success', { version, ttlInS })
if (client.disconnected) {
metrics.inc('editor.join-doc.disconnected', 1, {
status: 'after-joining-room',
status: 'after-doc-updater-call',
})
// the client will not read the response anyways
return callback()
}
DocumentUpdaterManager.getDocument(
projectId,
docId,
fromVersion,
function (error, result) {
if (error) {
if (error instanceof ClientRequestedMissingOpsError) {
emitJoinDocCatchUpMetrics('missing', error.info)
}
return callback(error)
}
const { lines, version, ranges, ops, ttlInS, type } = result
emitJoinDocCatchUpMetrics('success', { version, ttlInS })
if (client.disconnected) {
metrics.inc('editor.join-doc.disconnected', 1, {
status: 'after-doc-updater-call',
})
// the client will not read the response anyways
return callback()
}
if (isRestrictedUser && ranges && ranges.comments) {
ranges.comments = []
}
if (isRestrictedUser && ranges && ranges.comments) {
ranges.comments = []
}
// Encode any binary bits of data so it can go via WebSockets
// See http://ecmanaut.blogspot.co.uk/2006/07/encoding-decoding-utf8-in-javascript.html
const encodeForWebsockets = text =>
unescape(encodeURIComponent(text))
metrics.inc('client_supports_history_v1_ot', 1, {
status: options.supportsHistoryOT ? 'success' : 'failure',
})
let escapedLines
if (type === 'history-ot') {
if (!options.supportsHistoryOT) {
RoomManager.leaveDoc(client, docId)
// TODO(24596): ask the user to reload the editor page (via out-of-sync modal when there are pending ops).
return callback(
new CodedError('client does not support history-ot')
)
}
escapedLines = lines
} else {
escapedLines = []
for (let line of lines) {
try {
line = encodeForWebsockets(line)
} catch (err) {
OError.tag(err, 'error encoding line uri component', {
line,
})
return callback(err)
}
escapedLines.push(line)
}
if (options.encodeRanges) {
try {
for (const comment of (ranges && ranges.comments) || []) {
if (comment.op.c) {
comment.op.c = encodeForWebsockets(comment.op.c)
}
}
for (const change of (ranges && ranges.changes) || []) {
if (change.op.i) {
change.op.i = encodeForWebsockets(change.op.i)
}
if (change.op.d) {
change.op.d = encodeForWebsockets(change.op.d)
}
}
} catch (err) {
OError.tag(err, 'error encoding range uri component', {
ranges,
})
return callback(err)
}
}
}
AuthorizationManager.addAccessToDoc(client, docId, () => {})
logger.debug(
{
userId,
projectId,
docId,
fromVersion,
clientId: client.id,
},
'client joined doc'
// Encode any binary bits of data so it can go via WebSockets
// See http://ecmanaut.blogspot.co.uk/2006/07/encoding-decoding-utf8-in-javascript.html
const encodeForWebsockets = text => unescape(encodeURIComponent(text))
metrics.inc('client_supports_history_v1_ot', 1, {
status: options.supportsHistoryOT ? 'success' : 'failure',
})
let escapedLines
if (type === 'history-ot') {
if (!options.supportsHistoryOT) {
// TODO(24596): ask the user to reload the editor page (via out-of-sync modal when there are pending ops).
return callback(
new CodedError('client does not support history-ot')
)
callback(null, escapedLines, version, ops, ranges, type)
}
)
})
}
)
},
escapedLines = lines
} else {
escapedLines = []
for (let line of lines) {
try {
line = encodeForWebsockets(line)
} catch (err) {
OError.tag(err, 'error encoding line uri component', {
line,
})
return callback(err)
}
escapedLines.push(line)
}
if (options.encodeRanges) {
try {
for (const comment of (ranges && ranges.comments) || []) {
if (comment.op.c) {
comment.op.c = encodeForWebsockets(comment.op.c)
}
}
for (const change of (ranges && ranges.changes) || []) {
if (change.op.i) {
change.op.i = encodeForWebsockets(change.op.i)
}
if (change.op.d) {
change.op.d = encodeForWebsockets(change.op.d)
}
}
} catch (err) {
OError.tag(err, 'error encoding range uri component', {
ranges,
})
return callback(err)
}
}
}
_assertClientAuthorization(client, docId, callback) {
// Check for project-level access first
AuthorizationManager.assertClientCanViewProject(client, function (error) {
if (error) {
return callback(error)
}
// Check for doc-level access next
AuthorizationManager.assertClientCanViewProjectAndDoc(
client,
docId,
function (error) {
if (error) {
// No cached access, check docupdater
const { project_id: projectId } = client.ol_context
DocumentUpdaterManager.checkDocument(
AuthorizationManager.addAccessToDoc(client, docId, () => {})
logger.debug(
{
userId,
projectId,
docId,
function (error) {
if (error) {
return callback(error)
} else {
// Success
AuthorizationManager.addAccessToDoc(client, docId, callback)
}
}
)
} else {
// Access already cached
callback()
}
fromVersion,
clientId: client.id,
},
'client joined doc'
)
// The trailing options bag carries capability flags for the client.
// canSkipLeaveDoc tells the client it may skip the leaveDoc RPC,
// because this real-time's leaveDoc is a no-op (doc rooms have been
// retired; see leaveDoc below). Older clients ignore the extra
// argument and keep calling leaveDoc, which is harmless.
callback(null, escapedLines, version, ops, ranges, type, {
canSkipLeaveDoc: true,
})
}
)
})
},
leaveDoc(client, docId, callback) {
// client may have disconnected, but we have to cleanup internal state.
client.joinLeaveEpoch++
// No-op: real-time no longer tracks per-doc room membership (applied ops
// are broadcast to the project room and filtered by doc id client-side).
// The RPC is kept so older/in-flight clients still receive an ack; the
// frontend no longer emits leaveDoc. Doc access is intentionally not
// removed here (the connection is per-project and already authorised).
metrics.inc('editor.leave-doc', 1, { status: client.transport })
const { project_id: projectId, user_id: userId } = client.ol_context
logger.debug(
{ userId, projectId, docId, clientId: client.id },
'client leaving doc'
)
RoomManager.leaveDoc(client, docId)
// we could remove permission when user leaves a doc, but because
// the connection is per-project, we continue to allow access
// after the initial joinDoc since we know they are already authorised.
// # AuthorizationManager.removeAccessToDoc client, doc_id
callback()
},
updateClientPosition(client, cursorData, callback) {
services/real-time/test/acceptance/js/EarlyDisconnect.js
+1 -8
View File
@@ -158,14 +158,7 @@ describe('EarlyDisconnect', function () {
return done(err)
}
expect(resp).to.not.include(`editor-events:${this.project_id}`)
return rclient.pubsub('CHANNELS', (err, resp) => {
if (err) {
return done(err)
}
expect(resp).to.not.include(`applied-ops:${this.doc_id}`)
return done()
})
return done()
})
return null
})
services/real-time/test/acceptance/js/HttpControllerTests.js
+1 -1
View File
@@ -83,7 +83,7 @@ describe('HttpControllerTests', function () {
last_name: 'Bloggs',
project_id: this.project_id,
user_id: this.user_id,
rooms: [this.project_id, this.doc_id],
rooms: [this.project_id],
})
})
})
services/real-time/test/acceptance/js/JoinDocTests.js
+7 -101
View File
@@ -89,19 +89,9 @@ describe('joinDoc', function () {
this.ops,
this.ranges,
'sharejs-text-ot',
{ canSkipLeaveDoc: true },
])
})
return it('should have joined the doc room', function (done) {
return RealTimeClient.getConnectedClient(
this.client.socket.sessionid,
(error, client) => {
if (error) return done(error)
expect(Array.from(client.rooms).includes(this.doc_id)).to.equal(true)
return done()
}
)
})
})
describe('when authorised readOnly', function () {
@@ -169,19 +159,9 @@ describe('joinDoc', function () {
this.ops,
this.ranges,
'sharejs-text-ot',
{ canSkipLeaveDoc: true },
])
})
return it('should have joined the doc room', function (done) {
return RealTimeClient.getConnectedClient(
this.client.socket.sessionid,
(error, client) => {
if (error) return done(error)
expect(Array.from(client.rooms).includes(this.doc_id)).to.equal(true)
return done()
}
)
})
})
describe('when authorised as owner', function () {
@@ -249,19 +229,9 @@ describe('joinDoc', function () {
this.ops,
this.ranges,
'sharejs-text-ot',
{ canSkipLeaveDoc: true },
])
})
return it('should have joined the doc room', function (done) {
return RealTimeClient.getConnectedClient(
this.client.socket.sessionid,
(error, client) => {
if (error) return done(error)
expect(Array.from(client.rooms).includes(this.doc_id)).to.equal(true)
return done()
}
)
})
})
// It is impossible to write an acceptance test to test joining an unauthorized
@@ -329,19 +299,6 @@ describe('joinDoc', function () {
it('should return an invalid id error', function () {
this.error.message.should.equal('invalid Mongo ObjectId')
})
return it('should not have joined the doc room', function (done) {
return RealTimeClient.getConnectedClient(
this.client.socket.sessionid,
(error, client) => {
if (error) return done(error)
expect(Array.from(client.rooms).includes('invalid-doc-id')).to.equal(
false
)
return done()
}
)
})
})
describe('with a fromVersion', function () {
@@ -411,19 +368,9 @@ describe('joinDoc', function () {
this.ops,
this.ranges,
'sharejs-text-ot',
{ canSkipLeaveDoc: true },
])
})
return it('should have joined the doc room', function (done) {
return RealTimeClient.getConnectedClient(
this.client.socket.sessionid,
(error, client) => {
if (error) return done(error)
expect(Array.from(client.rooms).includes(this.doc_id)).to.equal(true)
return done()
}
)
})
})
describe('with options', function () {
@@ -493,19 +440,9 @@ describe('joinDoc', function () {
this.ops,
this.ranges,
'sharejs-text-ot',
{ canSkipLeaveDoc: true },
])
})
return it('should have joined the doc room', function (done) {
return RealTimeClient.getConnectedClient(
this.client.socket.sessionid,
(error, client) => {
if (error) return done(error)
expect(Array.from(client.rooms).includes(this.doc_id)).to.equal(true)
return done()
}
)
})
})
describe('with fromVersion and options', function () {
@@ -577,19 +514,9 @@ describe('joinDoc', function () {
this.ops,
this.ranges,
'sharejs-text-ot',
{ canSkipLeaveDoc: true },
])
})
return it('should have joined the doc room', function (done) {
return RealTimeClient.getConnectedClient(
this.client.socket.sessionid,
(error, client) => {
if (error) return done(error)
expect(Array.from(client.rooms).includes(this.doc_id)).to.equal(true)
return done()
}
)
})
})
describe('with type=history-ot', function () {
@@ -664,19 +591,9 @@ describe('joinDoc', function () {
this.ops,
this.ranges,
'history-ot',
{ canSkipLeaveDoc: true },
])
})
it('should have joined the doc room', function (done) {
RealTimeClient.getConnectedClient(
this.client.socket.sessionid,
(error, client) => {
if (error) return done(error)
expect(client.rooms).to.deep.equal([this.project_id, this.doc_id])
done()
}
)
})
})
describe('when support is not indicated', function () {
@@ -713,17 +630,6 @@ describe('joinDoc', function () {
it('should not return the doc lines, version, ranges and ops', function () {
this.returnedArgs.should.deep.equal([])
})
it('should leave the doc room again', function (done) {
RealTimeClient.getConnectedClient(
this.client.socket.sessionid,
(error, client) => {
if (error) return done(error)
expect(client.rooms).to.deep.equal([this.project_id])
done()
}
)
})
})
})
})
services/real-time/test/acceptance/js/LeaveDocTests.js
+8 -94
View File
@@ -1,40 +1,18 @@
/* eslint-disable
no-return-assign,
no-unused-vars,
*/
// TODO: This file was created by bulk-decaffeinate.
// Fix any style issues and re-enable lint.
/*
* decaffeinate suggestions:
* DS101: Remove unnecessary use of Array.from
* DS102: Remove unnecessary code created because of implicit returns
* DS207: Consider shorter variations of null checks
* Full docs: https://github.com/decaffeinate/decaffeinate/blob/master/docs/suggestions.md
*/
import { expect } from 'chai'
import sinon from 'sinon'
import RealTimeClient from './helpers/RealTimeClient.js'
import MockDocUpdaterServer from './helpers/MockDocUpdaterServer.js'
import FixturesManager from './helpers/FixturesManager.js'
import logger from '@overleaf/logger'
import async from 'async'
// leaveDoc is a no-op on the server (real-time no longer tracks per-doc room
// membership); the RPC is kept so clients still receive an ack. These tests
// cover that contract: a valid leaveDoc acks without error, an invalid doc id
// is rejected by the router validation.
describe('leaveDoc', function () {
before(function () {
this.lines = ['test', 'doc', 'lines']
this.version = 42
this.ops = ['mock', 'doc', 'ops']
sinon.spy(logger, 'error')
sinon.spy(logger, 'warn')
sinon.spy(logger, 'debug')
return (this.other_doc_id = FixturesManager.getRandomId())
})
after(function () {
logger.error.restore() // remove the spy
logger.warn.restore()
return logger.debug.restore()
})
return describe('when joined to a doc', function () {
@@ -87,28 +65,17 @@ describe('leaveDoc', function () {
describe('then leaving the doc', function () {
beforeEach(function (done) {
return this.client.emit('leaveDoc', this.doc_id, error => {
if (error != null) {
throw error
}
this.error = error
return done()
})
})
return it('should have left the doc room', function (done) {
return RealTimeClient.getConnectedClient(
this.client.socket.sessionid,
(error, client) => {
if (error) return done(error)
expect(Array.from(client.rooms).includes(this.doc_id)).to.equal(
false
)
return done()
}
)
return it('should not return an error', function () {
return expect(this.error).to.not.exist
})
})
describe('then leaving an invalid doc', function () {
return describe('then leaving an invalid doc', function () {
beforeEach(function (done) {
return this.client.emit('leaveDoc', 'bad-id', error => {
this.error = error
@@ -120,58 +87,5 @@ describe('leaveDoc', function () {
return expect(this.error).to.exist
})
})
describe('when sending a leaveDoc request before the previous joinDoc request has completed', function () {
beforeEach(function (done) {
this.client.emit('leaveDoc', this.doc_id, () => {})
this.client.emit('joinDoc', this.doc_id, () => {})
return this.client.emit('leaveDoc', this.doc_id, error => {
if (error != null) {
throw error
}
return done()
})
})
it('should not trigger an error', function () {
return sinon.assert.neverCalledWith(
logger.error,
sinon.match.any,
"not subscribed - shouldn't happen"
)
})
return it('should have left the doc room', function (done) {
return RealTimeClient.getConnectedClient(
this.client.socket.sessionid,
(error, client) => {
if (error) return done(error)
expect(Array.from(client.rooms).includes(this.doc_id)).to.equal(
false
)
return done()
}
)
})
})
return describe('when sending a leaveDoc for a room the client has not joined ', function () {
beforeEach(function (done) {
return this.client.emit('leaveDoc', this.other_doc_id, error => {
if (error != null) {
throw error
}
return done()
})
})
return it('should trigger a low level message only', function () {
return sinon.assert.calledWith(
logger.debug,
sinon.match.any,
'ignoring request from client to leave room it is not in'
)
})
})
})
})
services/real-time/test/acceptance/js/LeaveProjectTests.js
-22
View File
@@ -129,17 +129,6 @@ describe('leaveProject', function () {
})
return null
})
return it('should remain subscribed to the applied-ops channels', function (done) {
rclient.pubsub('CHANNELS', (err, resp) => {
if (err) {
return done(err)
}
resp.should.include(`applied-ops:${this.doc_id}`)
return done()
})
return null
})
})
return describe('with no other clients in the project', function () {
@@ -219,16 +208,5 @@ describe('leaveProject', function () {
})
return null
})
return it('should not subscribe to the applied-ops channels anymore', function (done) {
rclient.pubsub('CHANNELS', (err, resp) => {
if (err) {
return done(err)
}
resp.should.not.include(`applied-ops:${this.doc_id}`)
return done()
})
return null
})
})
})
services/real-time/test/acceptance/js/MatrixTests.js
-38
View File
@@ -349,44 +349,6 @@ describe('MatrixTests', function () {
}
)
})
it('should not add the user into the privateDoc room', function (done) {
RealTimeClient.getConnectedClient(
client.socket.sessionid,
(error, client) => {
if (error?.message === 'not found') return done() // disconnected
if (error) return done(error)
expect(client.rooms).to.not.include(privateDocId)
done()
}
)
})
})
describe('receive updates', function () {
const receivedMessages = []
beforeEach(function publishAnUpdateInRedis(done) {
const update = {
doc_id: privateDocId,
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('applied-ops', JSON.stringify(update))
})
it('should send nothing to client', function () {
expect(receivedMessages).to.have.length(0)
})
})
describe('receive updates via editor-events', function () {
services/real-time/test/acceptance/js/PubSubRace.js
-244
View File
@@ -1,13 +1,3 @@
/* eslint-disable
no-return-assign,
*/
// TODO: This file was created by bulk-decaffeinate.
// Fix any style issues and re-enable lint.
/*
* decaffeinate suggestions:
* DS102: Remove unnecessary code created because of implicit returns
* Full docs: https://github.com/decaffeinate/decaffeinate/blob/master/docs/suggestions.md
*/
import RealTimeClient from './helpers/RealTimeClient.js'
import MockDocUpdaterServer from './helpers/MockDocUpdaterServer.js'
@@ -22,229 +12,6 @@ describe('PubSubRace', function () {
return MockDocUpdaterServer.run(done)
})
describe('when the client leaves a doc before joinDoc completes', function () {
before(function (done) {
return async.series(
[
cb => {
return FixturesManager.setUpProject(
{
privilegeLevel: 'owner',
project: {
name: 'Test Project',
},
},
(e, { project_id: projectId, user_id: userId }) => {
this.project_id = projectId
this.user_id = userId
return cb()
}
)
},
cb => {
this.clientA = RealTimeClient.connect(
this.project_id,
(error, project, privilegeLevel, protocolVersion) => {
this.project = project
this.privilegeLevel = privilegeLevel
this.protocolVersion = protocolVersion
return cb(error)
}
)
},
cb => {
return FixturesManager.setUpDoc(
this.project_id,
{ lines: this.lines, version: this.version, ops: this.ops },
(e, { doc_id: docId }) => {
this.doc_id = docId
return cb(e)
}
)
},
cb => {
this.clientA.emit('joinDoc', this.doc_id, () => {})
// leave before joinDoc completes
return this.clientA.emit('leaveDoc', this.doc_id, cb)
},
cb => {
// wait for subscribe and unsubscribe
return setTimeout(cb, 100)
},
],
done
)
})
return it('should not subscribe to the applied-ops channels anymore', function (done) {
rclient.pubsub('CHANNELS', (err, resp) => {
if (err) {
return done(err)
}
resp.should.not.include(`applied-ops:${this.doc_id}`)
return done()
})
return null
})
})
describe('when the client emits joinDoc and leaveDoc requests frequently and leaves eventually', function () {
before(function (done) {
return async.series(
[
cb => {
return FixturesManager.setUpProject(
{
privilegeLevel: 'owner',
project: {
name: 'Test Project',
},
},
(e, { project_id: projectId, user_id: userId }) => {
this.project_id = projectId
this.user_id = userId
return cb()
}
)
},
cb => {
this.clientA = RealTimeClient.connect(
this.project_id,
(error, project, privilegeLevel, protocolVersion) => {
this.project = project
this.privilegeLevel = privilegeLevel
this.protocolVersion = protocolVersion
return cb(error)
}
)
},
cb => {
return FixturesManager.setUpDoc(
this.project_id,
{ lines: this.lines, version: this.version, ops: this.ops },
(e, { doc_id: docId }) => {
this.doc_id = docId
return cb(e)
}
)
},
cb => {
this.clientA.emit('joinDoc', this.doc_id, () => {})
this.clientA.emit('leaveDoc', this.doc_id, () => {})
this.clientA.emit('joinDoc', this.doc_id, () => {})
this.clientA.emit('leaveDoc', this.doc_id, () => {})
this.clientA.emit('joinDoc', this.doc_id, () => {})
this.clientA.emit('leaveDoc', this.doc_id, () => {})
this.clientA.emit('joinDoc', this.doc_id, () => {})
this.clientA.emit('leaveDoc', this.doc_id, () => {})
this.clientA.emit('joinDoc', this.doc_id, () => {})
return this.clientA.emit('leaveDoc', this.doc_id, cb)
},
cb => {
// wait for subscribe and unsubscribe
return setTimeout(cb, 100)
},
],
done
)
})
return it('should not subscribe to the applied-ops channels anymore', function (done) {
rclient.pubsub('CHANNELS', (err, resp) => {
if (err) {
return done(err)
}
resp.should.not.include(`applied-ops:${this.doc_id}`)
return done()
})
return null
})
})
describe('when the client emits joinDoc and leaveDoc requests frequently and remains in the doc', function () {
before(function (done) {
return async.series(
[
cb => {
return FixturesManager.setUpProject(
{
privilegeLevel: 'owner',
project: {
name: 'Test Project',
},
},
(e, { project_id: projectId, user_id: userId }) => {
this.project_id = projectId
this.user_id = userId
return cb()
}
)
},
cb => {
this.clientA = RealTimeClient.connect(
this.project_id,
(error, project, privilegeLevel, protocolVersion) => {
this.project = project
this.privilegeLevel = privilegeLevel
this.protocolVersion = protocolVersion
return cb(error)
}
)
},
cb => {
return FixturesManager.setUpDoc(
this.project_id,
{ lines: this.lines, version: this.version, ops: this.ops },
(e, { doc_id: docId }) => {
this.doc_id = docId
return cb(e)
}
)
},
cb => {
this.clientA.emit('joinDoc', this.doc_id, () => {})
this.clientA.emit('leaveDoc', this.doc_id, () => {})
this.clientA.emit('joinDoc', this.doc_id, () => {})
this.clientA.emit('leaveDoc', this.doc_id, () => {})
this.clientA.emit('joinDoc', this.doc_id, () => {})
this.clientA.emit('leaveDoc', this.doc_id, () => {})
this.clientA.emit('joinDoc', this.doc_id, () => {})
this.clientA.emit('leaveDoc', this.doc_id, () => {})
return this.clientA.emit('joinDoc', this.doc_id, cb)
},
cb => {
// wait for subscribe and unsubscribe
return setTimeout(cb, 100)
},
],
done
)
})
return it('should subscribe to the applied-ops channels', function (done) {
rclient.pubsub('CHANNELS', (err, resp) => {
if (err) {
return done(err)
}
resp.should.include(`applied-ops:${this.doc_id}`)
return done()
})
return null
})
})
return describe('when the client disconnects before joinDoc completes', function () {
before(function (done) {
return async.series(
@@ -332,16 +99,5 @@ describe('PubSubRace', function () {
})
return null
})
return it('should not subscribe to the applied-ops channels anymore', function (done) {
rclient.pubsub('CHANNELS', (err, resp) => {
if (err) {
return done(err)
}
resp.should.not.include(`applied-ops:${this.doc_id}`)
return done()
})
return null
})
})
})
services/real-time/test/acceptance/js/ReceiveUpdateTests.js
+41 -137
View File
@@ -162,106 +162,6 @@ describe('receiveUpdate', function () {
return this.clientD != null ? this.clientD.disconnect() : undefined
})
describe('with an update from clientA', function () {
beforeEach(function (done) {
this.update = {
doc_id: this.doc_id,
op: {
meta: {
source: this.clientA.publicId,
},
v: this.version,
doc: this.doc_id,
op: [{ i: 'foo', p: 50 }],
},
}
rclient.publish('applied-ops', JSON.stringify(this.update))
return setTimeout(done, 200)
}) // Give clients time to get message
it('should send the full op to clientB', function () {
return this.clientBUpdates.should.deep.equal([this.update.op])
})
it('should send an ack to clientA', function () {
return this.clientAUpdates.should.deep.equal([
{
v: this.version,
doc: this.doc_id,
},
])
})
return it('should send nothing to clientC', function () {
return this.clientCUpdates.should.deep.equal([])
})
})
describe('with an update from clientC', function () {
beforeEach(function (done) {
this.update = {
doc_id: this.doc_id_second,
op: {
meta: {
source: this.clientC.publicId,
},
v: this.version,
doc: this.doc_id_second,
op: [{ i: 'update from clientC', p: 50 }],
},
}
rclient.publish('applied-ops', JSON.stringify(this.update))
return setTimeout(done, 200)
}) // Give clients time to get message
it('should send nothing to clientA', function () {
return this.clientAUpdates.should.deep.equal([])
})
it('should send nothing to clientB', function () {
return this.clientBUpdates.should.deep.equal([])
})
return it('should send an ack to clientC', function () {
return this.clientCUpdates.should.deep.equal([
{
v: this.version,
doc: this.doc_id_second,
},
])
})
})
describe('with an update from a remote client for project 1', function () {
beforeEach(function (done) {
this.update = {
doc_id: this.doc_id,
op: {
meta: {
source: 'this-is-a-remote-client-id',
},
v: this.version,
doc: this.doc_id,
op: [{ i: 'foo', p: 50 }],
},
}
rclient.publish('applied-ops', JSON.stringify(this.update))
return setTimeout(done, 200)
}) // Give clients time to get message
it('should send the full op to clientA', function () {
return this.clientAUpdates.should.deep.equal([this.update.op])
})
it('should send the full op to clientB', function () {
return this.clientBUpdates.should.deep.equal([this.update.op])
})
return it('should send nothing to clientC', function () {
return this.clientCUpdates.should.deep.equal([])
})
})
describe('with an update from clientA published on the editor-events channel', function () {
beforeEach(function (done) {
this.update = {
@@ -373,65 +273,69 @@ describe('receiveUpdate', function () {
})
})
describe('with an error for the first project', function () {
describe('with an update for the second project on the editor-events channel', function () {
beforeEach(function (done) {
this.update = {
project_id: this.project_id_second,
doc_id: this.doc_id_second,
message: 'otUpdateApplied',
op: {
meta: {
source: this.clientC.publicId,
},
v: this.version,
doc: this.doc_id_second,
op: [{ i: 'bar', p: 50 }],
},
}
rclient.publish(
'applied-ops',
JSON.stringify({
doc_id: this.doc_id,
error: (this.error = 'something went wrong'),
})
`editor-events:${this.project_id_second}`,
JSON.stringify(this.update)
)
return setTimeout(done, 200)
setTimeout(done, 200)
}) // Give clients time to get message
it('should send the error to the clients in the first project', function () {
this.clientAErrors.should.deep.equal([this.error])
return this.clientBErrors.should.deep.equal([this.error])
it('should send an ack to clientC', function () {
this.clientCUpdates.should.deep.equal([
{
v: this.version,
doc: this.doc_id_second,
},
])
})
it('should not send any errors to the client in the second project', function () {
return this.clientCErrors.should.deep.equal([])
})
it('should disconnect the clients of the first project', function () {
this.clientA.socket.connected.should.equal(false)
return this.clientB.socket.connected.should.equal(false)
})
return it('should not disconnect the client in the second project', function () {
return this.clientC.socket.connected.should.equal(true)
it('should send nothing to the clients in the first project', function () {
this.clientAUpdates.should.deep.equal([])
this.clientBUpdates.should.deep.equal([])
this.clientDUpdates.should.deep.equal([])
})
})
return describe('with an error for the second project', function () {
describe('with an error for the second project on the editor-events channel', function () {
beforeEach(function (done) {
rclient.publish(
'applied-ops',
`editor-events:${this.project_id_second}`,
JSON.stringify({
project_id: this.project_id_second,
doc_id: this.doc_id_second,
message: 'otUpdateError',
error: (this.error = 'something went wrong'),
})
)
return setTimeout(done, 200)
setTimeout(done, 200)
}) // Give clients time to get message
it('should send the error to the client in the second project', function () {
this.clientCErrors.should.deep.equal([this.error])
this.clientC.socket.connected.should.equal(false)
})
it('should not send any errors to the clients in the first project', function () {
this.clientAErrors.should.deep.equal([])
return this.clientBErrors.should.deep.equal([])
})
it('should send the error to the client in the second project', function () {
return this.clientCErrors.should.deep.equal([this.error])
})
it('should not disconnect the clients of the first project', function () {
this.clientBErrors.should.deep.equal([])
this.clientDErrors.should.deep.equal([])
this.clientA.socket.connected.should.equal(true)
return this.clientB.socket.connected.should.equal(true)
})
return it('should disconnect the client in the second project', function () {
return this.clientC.socket.connected.should.equal(false)
this.clientB.socket.connected.should.equal(true)
})
})
})
services/real-time/test/unit/js/DocumentUpdaterController.test.js
+4 -184
View File
@@ -15,8 +15,6 @@ describe('DocumentUpdaterController', function () {
ctx.doc_id = 'doc-id-123'
ctx.callback = sinon.stub()
ctx.io = { mock: 'socket.io' }
ctx.rclient = []
ctx.RoomEvents = { on: sinon.stub() }
vi.doMock('@overleaf/settings', () => ({
default: (ctx.settings = {
@@ -29,34 +27,10 @@ describe('DocumentUpdaterController', function () {
}),
}))
vi.doMock('../../../app/js/RedisClientManager', () => ({
default: {
createClientList: () => {
ctx.redis = {
createClient: name => {
let rclientStub
ctx.rclient.push((rclientStub = { name }))
return rclientStub
},
}
},
},
}))
vi.doMock('../../../app/js/SafeJsonParse', () => ({
default: (ctx.SafeJsonParse = {
parse: (data, cb) => cb(null, JSON.parse(data)),
}),
}))
vi.doMock('../../../app/js/EventLogger', () => ({
default: (ctx.EventLogger = { checkEventOrder: sinon.stub() }),
}))
vi.doMock('../../../app/js/HealthCheckManager', () => ({
default: { check: sinon.stub() },
}))
vi.doMock('@overleaf/metrics', () => ({
default: (ctx.metrics = {
inc: sinon.stub(),
@@ -64,110 +38,9 @@ describe('DocumentUpdaterController', function () {
}),
}))
vi.doMock('../../../app/js/RoomManager', () => ({
default: (ctx.RoomManager = {
eventSource: sinon.stub().returns(ctx.RoomEvents),
}),
}))
vi.doMock('../../../app/js/ChannelManager', () => ({
default: (ctx.ChannelManager = {}),
}))
ctx.EditorUpdatesController = (await import(modulePath)).default
})
describe('listenForUpdatesFromDocumentUpdater', function () {
beforeEach(function (ctx) {
ctx.rclient.length = 0 // clear any existing clients
ctx.EditorUpdatesController.rclientList = [
ctx.redis.createClient('first'),
ctx.redis.createClient('second'),
]
ctx.rclient[0].subscribe = sinon.stub()
ctx.rclient[0].on = sinon.stub()
ctx.rclient[1].subscribe = sinon.stub()
ctx.rclient[1].on = sinon.stub()
ctx.EditorUpdatesController.listenForUpdatesFromDocumentUpdater()
})
it('should subscribe to the doc-updater stream', function (ctx) {
ctx.rclient[0].subscribe.calledWith('applied-ops').should.equal(true)
})
it('should register a callback to handle updates', function (ctx) {
ctx.rclient[0].on.calledWith('message').should.equal(true)
})
it('should subscribe to any additional doc-updater stream', function (ctx) {
ctx.rclient[1].subscribe.calledWith('applied-ops').should.equal(true)
ctx.rclient[1].on.calledWith('message').should.equal(true)
})
})
describe('_processMessageFromDocumentUpdater', function () {
describe('with bad JSON', function () {
beforeEach(function (ctx) {
ctx.SafeJsonParse.parse = sinon
.stub()
.callsArgWith(1, new Error('oops'))
ctx.EditorUpdatesController._processMessageFromDocumentUpdater(
ctx.io,
'applied-ops',
'blah'
)
})
it('should log an error', function (ctx) {
ctx.logger.error.called.should.equal(true)
})
})
describe('with update', function () {
beforeEach(function (ctx) {
ctx.message = {
doc_id: ctx.doc_id,
op: { t: 'foo', p: 12 },
}
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater =
sinon.stub()
ctx.EditorUpdatesController._processMessageFromDocumentUpdater(
ctx.io,
'applied-ops',
JSON.stringify(ctx.message)
)
})
it('should apply the update to the doc room', function (ctx) {
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater
.calledWith(ctx.io, ctx.doc_id, ctx.doc_id, ctx.message.op)
.should.equal(true)
})
})
describe('with error', function () {
beforeEach(function (ctx) {
ctx.message = {
doc_id: ctx.doc_id,
error: 'Something went wrong',
}
ctx.EditorUpdatesController._processErrorFromDocumentUpdater =
sinon.stub()
ctx.EditorUpdatesController._processMessageFromDocumentUpdater(
ctx.io,
'applied-ops',
JSON.stringify(ctx.message)
)
})
it('should process the error in the doc room', function (ctx) {
ctx.EditorUpdatesController._processErrorFromDocumentUpdater
.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) {
@@ -273,7 +146,7 @@ describe('DocumentUpdaterController', function () {
beforeEach(function (ctx) {
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater(
ctx.io,
ctx.doc_id,
ctx.project_id,
ctx.doc_id,
ctx.update
)
@@ -286,8 +159,8 @@ describe('DocumentUpdaterController', function () {
ctx.sourceClient.emit.calledOnce.should.equal(true)
})
it('should get the clients connected to the document', function (ctx) {
ctx.io.sockets.clients.calledWith(ctx.doc_id).should.equal(true)
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 the full update to the other clients', function (ctx) {
@@ -304,7 +177,7 @@ describe('DocumentUpdaterController', function () {
ctx.update.dup = true
ctx.EditorUpdatesController._applyUpdateFromDocumentUpdater(
ctx.io,
ctx.doc_id,
ctx.project_id,
ctx.doc_id,
ctx.update
)
@@ -322,36 +195,6 @@ 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 () {
@@ -362,29 +205,6 @@ describe('DocumentUpdaterController', function () {
}
})
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)
)
})
})
describe('in the project room', function () {
beforeEach(function (ctx) {
ctx.otherDocClient = new MockClient()
services/real-time/test/unit/js/RoomManager.test.js
+11 -142
View File
@@ -127,142 +127,6 @@ describe('RoomManager', function () {
})
})
describe('joinDoc', function () {
describe('when the doc room is empty', function () {
beforeEach(async function (ctx) {
await new Promise((resolve, reject) => {
ctx.RoomManager._clientsInRoom
.withArgs(ctx.client, ctx.doc_id)
.onFirstCall()
.returns(0)
ctx.client.join = sinon.stub()
ctx.callback = sinon.stub()
ctx.RoomEvents.on('doc-active', id => {
setTimeout(() => {
ctx.RoomEvents.emit(`doc-subscribed-${id}`)
}, 100)
})
ctx.RoomManager.joinDoc(ctx.client, ctx.doc_id, err => {
ctx.callback(err)
resolve()
})
})
})
it("should emit a 'doc-active' event with the id", function (ctx) {
ctx.RoomEvents.emit
.calledWithExactly('doc-active', ctx.doc_id)
.should.equal(true)
})
it("should listen for the 'doc-subscribed-id' event", function (ctx) {
ctx.RoomEvents.once
.calledWith(`doc-subscribed-${ctx.doc_id}`)
.should.equal(true)
})
it('should join the room using the id', function (ctx) {
ctx.client.join.calledWithExactly(ctx.doc_id).should.equal(true)
})
})
describe('when there are other clients in the doc room', function () {
beforeEach(async function (ctx) {
await new Promise((resolve, reject) => {
ctx.RoomManager._clientsInRoom
.withArgs(ctx.client, ctx.doc_id)
.onFirstCall()
.returns(123)
.onSecondCall()
.returns(124)
ctx.client.join = sinon.stub()
ctx.RoomManager.joinDoc(ctx.client, ctx.doc_id, err => {
if (err) return reject(err)
resolve()
})
})
})
it('should join the room using the id', function (ctx) {
ctx.client.join.called.should.equal(true)
})
it('should not emit any events', function (ctx) {
ctx.RoomEvents.emit.called.should.equal(false)
})
})
})
describe('leaveDoc', function () {
describe('when doc room will be empty after this client has left', function () {
beforeEach(function (ctx) {
ctx.RoomManager._clientAlreadyInRoom
.withArgs(ctx.client, ctx.doc_id)
.returns(true)
ctx.RoomManager._clientsInRoom
.withArgs(ctx.client, ctx.doc_id)
.onCall(0)
.returns(0)
ctx.client.leave = sinon.stub()
ctx.RoomManager.leaveDoc(ctx.client, ctx.doc_id)
})
it('should leave the room using the id', function (ctx) {
ctx.client.leave.calledWithExactly(ctx.doc_id).should.equal(true)
})
it("should emit a 'doc-empty' event with the id", function (ctx) {
ctx.RoomEvents.emit
.calledWithExactly('doc-empty', ctx.doc_id)
.should.equal(true)
})
})
describe('when there are other clients in the doc room', function () {
beforeEach(function (ctx) {
ctx.RoomManager._clientAlreadyInRoom
.withArgs(ctx.client, ctx.doc_id)
.returns(true)
ctx.RoomManager._clientsInRoom
.withArgs(ctx.client, ctx.doc_id)
.onCall(0)
.returns(123)
ctx.client.leave = sinon.stub()
ctx.RoomManager.leaveDoc(ctx.client, ctx.doc_id)
})
it('should leave the room using the id', function (ctx) {
ctx.client.leave.calledWithExactly(ctx.doc_id).should.equal(true)
})
it('should not emit any events', function (ctx) {
ctx.RoomEvents.emit.called.should.equal(false)
})
})
describe('when the client is not in the doc room', function () {
beforeEach(function (ctx) {
ctx.RoomManager._clientAlreadyInRoom
.withArgs(ctx.client, ctx.doc_id)
.returns(false)
ctx.RoomManager._clientsInRoom
.withArgs(ctx.client, ctx.doc_id)
.onCall(0)
.returns(0)
ctx.client.leave = sinon.stub()
ctx.RoomManager.leaveDoc(ctx.client, ctx.doc_id)
})
it('should not leave the room', function (ctx) {
ctx.client.leave.called.should.equal(false)
})
it('should not emit any events', function (ctx) {
ctx.RoomEvents.emit.called.should.equal(false)
})
})
})
describe('leaveProjectAndDocs', function () {
describe('when the client is connected to the project and multiple docs', function () {
beforeEach(function (ctx) {
@@ -315,12 +179,17 @@ describe('RoomManager', function () {
})
// put the client in the rooms
ctx.RoomManager.joinProject(ctx.client, ctx.project_id, () => {
ctx.RoomManager.joinDoc(ctx.client, ctx.doc_id, () => {
ctx.RoomManager.joinDoc(ctx.client, ctx.other_doc_id, () => {
// now leave the project
ctx.RoomManager.leaveProjectAndDocs(ctx.client)
resolve()
})
ctx.RoomManager.joinEntity(ctx.client, 'doc', ctx.doc_id, () => {
ctx.RoomManager.joinEntity(
ctx.client,
'doc',
ctx.other_doc_id,
() => {
// now leave the project
ctx.RoomManager.leaveProjectAndDocs(ctx.client)
resolve()
}
)
})
})
})
services/real-time/test/unit/js/WebsocketController.test.js
+4 -149
View File
@@ -37,7 +37,6 @@ describe('WebsocketController', function () {
id: (ctx.client_id = 'mock-client-id-123'),
publicId: `other-id-${Math.random()}`,
ol_context: {},
joinLeaveEpoch: 0,
join: sinon.stub(),
leave: sinon.stub(),
}
@@ -545,9 +544,6 @@ describe('WebsocketController', function () {
ctx.AuthorizationManager.assertClientCanViewProject = sinon
.stub()
.callsArgWith(1, null)
ctx.AuthorizationManager.assertClientCanViewProjectAndDoc = sinon
.stub()
.callsArgWith(2, null)
ctx.DocumentUpdaterManager.getDocument = sinon
.stub()
.callsArgWith(3, null, {
@@ -556,7 +552,6 @@ describe('WebsocketController', function () {
ranges: ctx.ranges,
ops: ctx.ops,
})
ctx.RoomManager.joinDoc = sinon.stub().callsArg(2)
})
describe('works', function () {
@@ -570,10 +565,6 @@ describe('WebsocketController', function () {
)
})
it('should inc the joinLeaveEpoch', function (ctx) {
expect(ctx.client.joinLeaveEpoch).to.equal(1)
})
it('should check that the client is authorized to view the project', function (ctx) {
ctx.AuthorizationManager.assertClientCanViewProject
.calledWith(ctx.client)
@@ -592,12 +583,6 @@ describe('WebsocketController', function () {
.should.equal(true)
})
it('should join the client to room for the doc_id', function (ctx) {
ctx.RoomManager.joinDoc
.calledWith(ctx.client, ctx.doc_id)
.should.equal(true)
})
it('should call the callback with the lines, version, ranges and ops', function (ctx) {
ctx.callback
.calledWith(null, ctx.doc_lines, ctx.version, ctx.ops, ctx.ranges)
@@ -801,10 +786,10 @@ describe('WebsocketController', function () {
describe('when the client disconnects while auth checks are running', function () {
beforeEach(async function (ctx) {
await new Promise((resolve, reject) => {
ctx.AuthorizationManager.assertClientCanViewProjectAndDoc.yields(
new Error()
)
ctx.DocumentUpdaterManager.checkDocument = (projectId, docId, cb) => {
ctx.AuthorizationManager.assertClientCanViewProject = (
client,
cb
) => {
ctx.client.disconnected = true
cb()
}
@@ -840,121 +825,6 @@ describe('WebsocketController', function () {
})
})
describe('when the client starts a parallel joinDoc request', function () {
beforeEach(async function (ctx) {
await new Promise((resolve, reject) => {
ctx.AuthorizationManager.assertClientCanViewProjectAndDoc.yields(
new Error()
)
ctx.DocumentUpdaterManager.checkDocument = (projectId, docId, cb) => {
ctx.DocumentUpdaterManager.checkDocument = sinon.stub().yields()
ctx.WebsocketController.joinDoc(
ctx.client,
ctx.doc_id,
-1,
{},
() => {}
)
cb()
}
ctx.WebsocketController.joinDoc(
ctx.client,
ctx.doc_id,
-1,
ctx.options,
(...args) => {
ctx.callback(...args)
// make sure the other joinDoc request completed
setTimeout(resolve, 5)
}
)
})
})
it('should call the callback with an error', function (ctx) {
expect(ctx.callback.called).to.equal(true)
expect(ctx.callback.args[0][0].message).to.equal(
'joinLeaveEpoch mismatch'
)
})
it('should get the document once (the parallel request wins)', function (ctx) {
expect(ctx.DocumentUpdaterManager.getDocument.callCount).to.equal(1)
})
})
describe('when the client starts a parallel leaveDoc request', function () {
beforeEach(async function (ctx) {
await new Promise((resolve, reject) => {
ctx.RoomManager.leaveDoc = sinon.stub()
ctx.AuthorizationManager.assertClientCanViewProjectAndDoc.yields(
new Error()
)
ctx.DocumentUpdaterManager.checkDocument = (projectId, docId, cb) => {
ctx.WebsocketController.leaveDoc(ctx.client, ctx.doc_id, () => {})
cb()
}
ctx.WebsocketController.joinDoc(
ctx.client,
ctx.doc_id,
-1,
ctx.options,
(...args) => {
ctx.callback(...args)
resolve()
}
)
})
})
it('should call the callback with an error', function (ctx) {
expect(ctx.callback.called).to.equal(true)
expect(ctx.callback.args[0][0].message).to.equal(
'joinLeaveEpoch mismatch'
)
})
it('should not get the document', function (ctx) {
expect(ctx.DocumentUpdaterManager.getDocument.called).to.equal(false)
})
})
describe('when the client disconnects while RoomManager.joinDoc is running', function () {
beforeEach(function (ctx) {
ctx.RoomManager.joinDoc = (client, docId, cb) => {
ctx.client.disconnected = true
cb()
}
ctx.WebsocketController.joinDoc(
ctx.client,
ctx.doc_id,
-1,
ctx.options,
ctx.callback
)
})
it('should call the callback with no details', function (ctx) {
expect(ctx.callback.args[0]).to.deep.equal([])
})
it('should increment the editor.join-doc.disconnected metric with a status', function (ctx) {
expect(
ctx.metrics.inc.calledWith('editor.join-doc.disconnected', 1, {
status: 'after-joining-room',
})
).to.equal(true)
})
it('should not get the document', function (ctx) {
expect(ctx.DocumentUpdaterManager.getDocument.called).to.equal(false)
})
})
describe('when the client disconnects while DocumentUpdaterManager.getDocument is running', function () {
beforeEach(function (ctx) {
ctx.DocumentUpdaterManager.getDocument = (
@@ -999,27 +869,12 @@ describe('WebsocketController', function () {
beforeEach(function (ctx) {
ctx.doc_id = 'doc-id-123'
ctx.client.ol_context.project_id = ctx.project_id
ctx.RoomManager.leaveDoc = sinon.stub()
ctx.WebsocketController.leaveDoc(ctx.client, ctx.doc_id, ctx.callback)
})
it('should inc the joinLeaveEpoch', function (ctx) {
expect(ctx.client.joinLeaveEpoch).to.equal(1)
})
it('should remove the client from the doc_id room', function (ctx) {
ctx.RoomManager.leaveDoc
.calledWith(ctx.client, ctx.doc_id)
.should.equal(true)
})
it('should call the callback', function (ctx) {
ctx.callback.called.should.equal(true)
})
it('should increment the leave-doc metric', function (ctx) {
ctx.metrics.inc.calledWith('editor.leave-doc').should.equal(true)
})
})
describe('getConnectedUsers', function () {
services/real-time/test/unit/js/helpers/MockClient.js
-1
View File
@@ -16,7 +16,6 @@ export default MockClient = class MockClient {
this.disconnect = sinon.stub()
this.id = idCounter++
this.publicId = idCounter++
this.joinLeaveEpoch = 0
}
disconnect() {}
services/web/frontend/js/features/ide-react/editor/document-container.ts
+32 -8
View File
@@ -40,6 +40,11 @@ const MAX_PENDING_OP_SIZE = 64
type JoinCallback = (error?: Error) => void
type LeaveCallback = JoinCallback
// Trailing options bag on the joinDoc response
type JoinDocResponseOptions = {
canSkipLeaveDoc?: boolean
}
type Update =
| {
v: number
@@ -102,6 +107,11 @@ export class DocumentContainer extends EventEmitter {
private joinCallbacks: JoinCallback[] = []
private leaveCallbacks: LeaveCallback[] = []
// Set from the joinDoc response. When true, the server signalled that its
// leaveDoc handler is a no-op (doc rooms retired), so we can skip the
// leaveDoc RPC and just clean up locally.
private canSkipLeaveDoc = false
doc?: ShareJsDoc
cm6?: EditorFacade
oldInflightOp?: ShareJsOperation
@@ -491,13 +501,15 @@ export class DocumentContainer extends EventEmitter {
version,
updates,
ranges,
type = 'sharejs-text-ot'
type = 'sharejs-text-ot',
options: JoinDocResponseOptions = {}
) => {
if (error) {
callback?.(error)
return
}
this.joined = true
this.canSkipLeaveDoc = options.canSkipLeaveDoc ?? false
this.doc?.catchUp(updates)
if (this.doc?.getType() !== type) {
// TODO(24596): page reload after checking for pending ops?
@@ -527,13 +539,15 @@ export class DocumentContainer extends EventEmitter {
version,
updates,
ranges,
type: OTType = 'sharejs-text-ot'
type: OTType = 'sharejs-text-ot',
options: JoinDocResponseOptions = {}
) => {
if (error) {
callback?.(error)
return
}
this.joined = true
this.canSkipLeaveDoc = options.canSkipLeaveDoc ?? false
this.doc = new ShareJsDoc(
this.doc_id,
docLines,
@@ -582,22 +596,32 @@ export class DocumentContainer extends EventEmitter {
}
private leaveDoc(callback?: LeaveCallback) {
if (this.canSkipLeaveDoc) {
debugConsole.log('[leaveDoc] Leaving doc (server round-trip skipped)')
this.finishLeaveDoc()
callback?.()
return
}
debugConsole.log('[leaveDoc] Sending leaveDoc request')
this.socket.emit('leaveDoc', this.doc_id, error => {
if (error) {
callback?.(error)
return
}
this.joined = false
for (const leaveCallback of this.leaveCallbacks) {
debugConsole.log('[_leaveDoc] Calling buffered callback', leaveCallback)
leaveCallback(error)
}
this.leaveCallbacks = []
this.finishLeaveDoc()
callback?.()
})
}
private finishLeaveDoc() {
this.joined = false
for (const leaveCallback of this.leaveCallbacks) {
debugConsole.log('[_leaveDoc] Calling buffered callback', leaveCallback)
leaveCallback()
}
this.leaveCallbacks = []
}
cleanUp() {
// if we arrive here from _onError the pending and inflight ops will have been cleared
if (this.hasBufferedOps()) {