Skip to content

fix: Only hand polling over to the chat relay once the room is joined on the HPB #2689

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Ivansss wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: Only hand polling over to the chat relay once the room is joined on the HPB #2689

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 109 additions & 10 deletions NextcloudTalk/Chat/NCChatController.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,17 @@ public class NCChatController: NSObject {
private var getHistoryTask: URLSessionDataTask?
private var pullMessagesTask: URLSessionDataTask?

private enum ChatRelayState {
case inactive, active, catchingUp
enum ChatRelayState {
// We are polling the chat API, the relay does not process messages yet
case inactive
// The chat is up to date, but our signaling session did not join the room yet, so nothing
// would be relayed to us: keep polling until the join is acked
case waitingForJoin
// The relay took over, polling stopped
case active
// The relay is active, but a message could not be rendered from its payload, so we are
// fetching the missing messages over the chat API
case catchingUp
}

private var chatRelayState: ChatRelayState = .inactive
Expand All @@ -58,7 +67,9 @@ public class NCChatController: NSObject {

super.init()

setupChatRelay()
let signalingController = NCSettingsController.sharedInstance().externalSignalingController(forAccountId: account.accountId)
setupChatRelay(with: signalingController)

AllocationTracker.shared.addAllocation("NCChatController")
}

Expand All @@ -71,7 +82,9 @@ public class NCChatController: NSObject {

super.init()

setupChatRelay()
let signalingController = NCSettingsController.sharedInstance().externalSignalingController(forAccountId: account.accountId)
setupChatRelay(with: signalingController)

AllocationTracker.shared.addAllocation("NCChatController")
}

Expand Down Expand Up @@ -425,14 +438,23 @@ public class NCChatController: NSObject {

// MARK: - External Signaling / Chat Relay

private func setupChatRelay() {
guard let signalingController = NCSettingsController.sharedInstance().externalSignalingController(forAccountId: account.accountId),
signalingController.hasChatRelay else { return }
// The signaling server only sends us the events of the room our session joined, so this is the
// condition for letting the relay take over from polling the chat API.
private var canChatRelayTakeOverPolling: Bool {
guard let externalSignalingController, externalSignalingController.hasChatRelay else { return false }

return externalSignalingController.joinedRoomToken == room.token
}

private func setupChatRelay(with signalingController: NCExternalSignalingController?) {
guard let signalingController, signalingController.hasChatRelay else { return }

externalSignalingController = signalingController
chatRelayMessagesQueue = DispatchQueue(label: "chat.relay.message.queue")
NotificationCenter.default.addObserver(self, selector: #selector(didReceiveChatMessageFromExternalSignaling(_:)), name: .extSignalingDidReceiveChatMessage, object: signalingController)
NotificationCenter.default.addObserver(self, selector: #selector(didRequestChatRefreshFromExternalSignaling(_:)), name: .extSignalingDidRequestChatRefresh, object: signalingController)
NotificationCenter.default.addObserver(self, selector: #selector(didReconnectExternalSignaling(_:)), name: .extSignalingDidReconnect, object: signalingController)
NotificationCenter.default.addObserver(self, selector: #selector(didJoinRoomOnExternalSignaling(_:)), name: .extSignalingDidJoinRoom, object: signalingController)
}

@objc private func didReceiveChatMessageFromExternalSignaling(_ notification: Notification) {
Expand Down Expand Up @@ -463,6 +485,63 @@ public class NCChatController: NSObject {
}
}

// The signaling server only sends us the events of the room our session joined, so the relay can
// take over from polling as soon as the join is acked. Poll once more without a timeout instead of
// waiting for the running long poll to time out, that poll then arms the relay on its 304.
@objc private func didJoinRoomOnExternalSignaling(_ notification: Notification) {
guard let roomToken = notification.userInfo?["roomToken"] as? String, roomToken == room.token else { return }
guard !stopChatMessagesPoll else { return }

chatRelayMessagesQueue?.async {
guard self.chatRelayState == .waitingForJoin else { return }

// We are not waiting for the join anymore, we are waiting for the poll below to confirm
// that the chat is up to date. Also keeps a second join from polling twice.
self.chatRelayState = .inactive

DispatchQueue.main.async {
let lastChatBlock = self.chatBlocksForRoomOrThread().last
self.startReceivingChatMessages(fromMessagesId: lastChatBlock?.newestMessageId ?? 0, withTimeout: false)
}
}
}

// Called when the chat is up to date and the server supports the chat relay. The relay only
// delivers messages of the room our signaling session joined, so handing polling over to it before
// the join is acked would stop polling while nothing is relayed to us yet, and every message posted
// in that window would be lost for good: the next relayed message advances the chat block past the
// gap, so it is never requested again. Keep polling until we are in the room instead.
private func handOverPollingToChatRelay(fromMessagesId messageId: Int) {
if canChatRelayTakeOverPolling {
print("Chat is up to date, now processing new messages from the chat relay")
startProcessingChatRelayMessages()
return
}

print("Chat is up to date, but we did not join the room on the signaling server yet, keep polling")

chatRelayMessagesQueue?.async {
self.chatRelayState = .waitingForJoin

// The join might have been acked while we were switching queues, in that case poll once
// without a timeout so the relay is armed right away instead of after the long poll.
let hasJoinedMeanwhile = self.canChatRelayTakeOverPolling

DispatchQueue.main.async {
self.startReceivingChatMessages(fromMessagesId: messageId, withTimeout: !hasJoinedMeanwhile)
}
}
}

// Puts the relay back into its initial state after a poll ended without arming it. Without this, a
// catch-up that doesn't repoll (brute-force protection, blocked chat, thread not found) would leave
// the relay in `.catchingUp` forever, buffering messages that are never flushed.
private func resetChatRelayState() {
chatRelayMessagesQueue?.async {
self.chatRelayState = .inactive
}
}

private func startProcessingChatRelayMessages() {
chatRelayMessagesQueue?.async {
self.chatRelayState = .active
Expand Down Expand Up @@ -1147,18 +1226,21 @@ public class NCChatController: NSObject {

if let error {
if self.isChatBeingBlocked(statusCode) {
self.resetChatRelayState()
self.notifyChatIsBlocked()
return
}

if statusCode == 404 {
NCLog.log("Thread not found error: \(error.description)")
self.resetChatRelayState()
NotificationCenter.default.post(name: .NCChatControllerDidReceiveThreadNotFound, object: self, userInfo: nil)
return
}

if statusCode == 429 {
NCLog.log("Brute-force protected, received 429 while receiving messages. No further polling.")
self.resetChatRelayState()
return
}

Expand All @@ -1180,6 +1262,7 @@ public class NCChatController: NSObject {
// When we receive a "history_cleared" message, we don't continue here, as otherwise
// we would request new messages, but instead, we need to request the initial history again
if message?.systemMessage == "history_cleared" {
self.resetChatRelayState()
return
}
}
Expand All @@ -1194,9 +1277,8 @@ public class NCChatController: NSObject {
let chatIsUpToDate = statusCode == 304
let lastChatBlock = self.chatBlocksForRoomOrThread().last

if chatIsUpToDate, let extSignaling = self.externalSignalingController, extSignaling.hasChatRelay {
print("Chat is up to date, now processing new messages from the chat relay")
self.startProcessingChatRelayMessages()
if chatIsUpToDate, self.externalSignalingController?.hasChatRelay == true {
self.handOverPollingToChatRelay(fromMessagesId: lastChatBlock?.newestMessageId ?? 0)
return
}

Expand Down Expand Up @@ -1436,4 +1518,21 @@ extension NCChatController {
// triggerChatRelayCatchUpForTesting() actually schedules the restart on the main queue, mirroring
// a catch-up that fires while the user is still in the room (just before they leave).
func markChatRelayActiveForTesting() { chatRelayState = .active }

// Puts the relay state machine into .waitingForJoin — the state the controller is in while it
// keeps polling because our signaling session did not join the room yet.
func markChatRelayWaitingForJoinForTesting() { chatRelayState = .waitingForJoin }

var chatRelayStateForTesting: ChatRelayState { chatRelayState }

// The condition the chat relay is gated on (see canChatRelayTakeOverPolling).
var canChatRelayTakeOverPollingForTesting: Bool { canChatRelayTakeOverPolling }

// Waits until everything queued on the relay queue ran, so tests don't have to guess timings.
func waitForChatRelayQueueForTesting() { chatRelayMessagesQueue?.sync {} }

// There is no signaling controller configured for the fake account, so tests pass in their own.
func setupChatRelayForTesting(with signalingController: NCExternalSignalingController) {
setupChatRelay(with: signalingController)
}
}
34 changes: 33 additions & 1 deletion NextcloudTalk/WebRTC/NCExternalSignalingController.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ extension Notification.Name {
static let extSignalingDidReceiveChatMessage = Notification.Name(rawValue: "NCExternalSignalingControllerDidReceiveChatMessageNotification")
static let extSignalingDidRequestChatRefresh = Notification.Name(rawValue: "NCExternalSignalingControllerDidRequestChatRefreshNotification")
static let extSignalingDidReconnect = Notification.Name(rawValue: "NCExternalSignalingControllerDidReconnectNotification")
static let extSignalingDidJoinRoom = Notification.Name(rawValue: "NCExternalSignalingControllerDidJoinRoomNotification")
}

public typealias SendMessageCompletionBlock = (_ task: URLSessionWebSocketTask?, _ status: NCExternalSignalingSendMessageStatus) -> Void
Expand All @@ -43,6 +44,10 @@ public enum NCExternalSignalingSendMessageStatus {

public var currentRoom: String?

// The room the signaling server acked us into. Unlike `currentRoom` this is cleared on every
// reconnect, so it answers "will the server relay this room's chat to our session right now?"
public private(set) var joinedRoomToken: String?

public private(set) var account: TalkAccount
public private(set) var disconnected: Bool = true
public private(set) var hasMCU: Bool = false
Expand Down Expand Up @@ -210,6 +215,7 @@ public enum NCExternalSignalingSendMessageStatus {
self.webSocket?.cancel()
self.webSocket = nil
self.helloResponseReceived = false
self.joinedRoomToken = nil
self.helloMessage?.ignoreCompletionBlock()
self.helloMessage = nil
self.disconnected = true
Expand Down Expand Up @@ -340,6 +346,15 @@ public enum NCExternalSignalingSendMessageStatus {
let sessionChanged = self.sessionId != newSessionId
self.sessionId = newSessionId

if sessionChanged {
// The new session did not join any room yet, the re-join below takes care of that
self.joinedRoomToken = nil
} else {
// The session was resumed, so the server kept us in the room we joined before and replays
// the messages we missed while being disconnected
self.setJoinedRoomToken(self.currentRoom)
}

guard let serverDict = helloDict["server"] as? [AnyHashable: Any],
let serverFeatures = serverDict["features"] as? [String],
let serverVersion = serverDict["version"] as? String
Expand Down Expand Up @@ -402,8 +417,10 @@ public enum NCExternalSignalingSendMessageStatus {
let roomId = roomDict["roomid"] as? String
else { return }

// If we are aware that we were in this room before, we should treat this as a success
// If we are aware that we were in this room before, we should treat this as a success.
// No room message follows in this case, so we have to set the joined room ourselves.
if currentRoom == roomId {
self.setJoinedRoomToken(roomId)
self.executeCompletionBlock(forMessageId: messageId, withStatus: .success)
return
}
Expand Down Expand Up @@ -462,6 +479,7 @@ public enum NCExternalSignalingSendMessageStatus {
func leaveRoom(withRoomId roomId: String) {
if self.currentRoom == roomId {
self.currentRoom = nil
self.joinedRoomToken = nil
self.joinRoom(withRoomId: "", withSessionId: "", withFederation: nil, withCompletionBlock: nil)
} else {
print("External signaling: Not leaving because it's not the room we joined")
Expand Down Expand Up @@ -556,11 +574,25 @@ public enum NCExternalSignalingSendMessageStatus {
self.currentRoom = newRoomId.isEmpty ? nil : newRoomId
}

// Set outside the check above: after a reconnect we re-join the same room, so `currentRoom` is
// unchanged, but this is exactly the moment we are part of the room on the signaling server again.
self.setJoinedRoomToken(newRoomId.isEmpty ? nil : newRoomId)

if let messageId = messageDict["id"] as? String {
self.executeCompletionBlock(forMessageId: messageId, withStatus: .success)
}
}

// Notifies about the room we are joined to on the signaling server, so components relying on the
// server pushing room events to us (e.g. the chat relay) know when they can start doing so.
private func setJoinedRoomToken(_ roomToken: String?) {
self.joinedRoomToken = roomToken

guard let roomToken else { return }

NotificationCenter.default.post(name: .extSignalingDidJoinRoom, object: self, userInfo: ["roomToken": roomToken])
}

func eventMessageReceived(eventDict: [AnyHashable: Any]) {
let eventTarget = eventDict["target"] as? String

Expand Down
105 changes: 105 additions & 0 deletions NextcloudTalkTests/Unit/Chat/UnitChatViewControllerTest.swift
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,111 @@ final class UnitChatViewControllerTest: TestBaseRealm {
"A chat-relay catch-up scheduled before stop() must not resume polling once stop() has run")
}

// MARK: - Chat relay

private func inertSignalingController(withChatRelay hasChatRelay: Bool = true) -> NCExternalSignalingController {
let account = NCDatabaseManager.sharedInstance().activeAccount()
let signalingController = NCExternalSignalingController(account: account, serverUrl: TestConstants.server, ticket: "fakeTicket")

// Make the controller inert: without a websocket the delegate callbacks of the failing
// connection attempt are ignored, so nothing reconnects underneath the assertions.
signalingController.disconnect()

if hasChatRelay {
let helloMessage: [AnyHashable: Any] = [
"type": "hello",
"id": "1",
"hello": [
"sessionid": "session-1",
"server": ["version": "2.0.0", "features": ["chat-relay"]]
]
]

signalingController.helloResponseReceived(messageDict: helloMessage)
}

return signalingController
}

private func drainMainQueue() {
let exp = expectation(description: "\(#function)\(#line)")
DispatchQueue.main.async { exp.fulfill() }
waitForExpectations(timeout: TestConstants.timeoutShort, handler: nil)
}

func testChatRelayOnlyTakesOverPollingOnceTheRoomIsJoinedOnTheSignalingServer() throws {
let room = addRoom(withToken: "relayGateRoom")
let chatController = NCChatController(for: room)!
let signalingController = inertSignalingController()

chatController.setupChatRelayForTesting(with: signalingController)

// The server supports the chat relay, but our signaling session is not in the room yet, so
// the relay would not deliver anything to us. Handing over to it here loses every message
// posted until the join is acked, since polling stops for good.
XCTAssertFalse(chatController.canChatRelayTakeOverPollingForTesting)

// Being in *another* room is not enough either
signalingController.roomMessageReceived(messageDict: ["type": "room", "room": ["roomid": "someOtherRoom"]])
XCTAssertFalse(chatController.canChatRelayTakeOverPollingForTesting)

signalingController.roomMessageReceived(messageDict: ["type": "room", "room": ["roomid": room.token]])
XCTAssertTrue(chatController.canChatRelayTakeOverPollingForTesting)

// We left the room again (or the connection dropped)
signalingController.resetWebSocket()
XCTAssertFalse(chatController.canChatRelayTakeOverPollingForTesting)

chatController.stop()
drainMainQueue()
}

func testJoiningTheRoomRepollsWhileWaitingForTheJoin() throws {
let room = addRoom(withToken: "relayJoinRoom")
let chatController = NCChatController(for: room)!
let signalingController = inertSignalingController()

chatController.setupChatRelayForTesting(with: signalingController)

// The chat is up to date, but we kept polling because we were not in the room yet
chatController.markChatRelayWaitingForJoinForTesting()

// The join of another room must not make us repoll
signalingController.roomMessageReceived(messageDict: ["type": "room", "room": ["roomid": "someOtherRoom"]])
chatController.waitForChatRelayQueueForTesting()
drainMainQueue()
XCTAssertEqual(chatController.chatRelayStateForTesting, .waitingForJoin)

// Our room was acked: repoll right away instead of waiting for the running long poll to time
// out, so the relay is armed by that poll's 304
signalingController.roomMessageReceived(messageDict: ["type": "room", "room": ["roomid": room.token]])
chatController.waitForChatRelayQueueForTesting()
drainMainQueue()
XCTAssertEqual(chatController.chatRelayStateForTesting, .inactive)

chatController.stop()
drainMainQueue()
}

func testJoiningTheRoomDoesNotRepollWhenTheRelayIsAlreadyActive() throws {
let room = addRoom(withToken: "relayActiveRoom")
let chatController = NCChatController(for: room)!
let signalingController = inertSignalingController()

chatController.setupChatRelayForTesting(with: signalingController)
chatController.markChatRelayActiveForTesting()

// A re-join while the relay is already processing messages must not restart polling
signalingController.roomMessageReceived(messageDict: ["type": "room", "room": ["roomid": room.token]])
chatController.waitForChatRelayQueueForTesting()
drainMainQueue()

XCTAssertEqual(chatController.chatRelayStateForTesting, .active)

chatController.stop()
drainMainQueue()
}

func testContentInsetAdjustsForOverlayViews() throws {
let activeAccount = NCDatabaseManager.sharedInstance().activeAccount()
let room = NCRoom()
Expand Down
Loading
Loading