From f2d5fd721ced521c17df0377e6c520a20c46d997 Mon Sep 17 00:00:00 2001 From: Joe Heck Date: Tue, 30 Apr 2024 12:22:53 -0700 Subject: [PATCH] run through and make the log output more consistent for each component --- .../Networking/NetworkSubsystem.swift | 10 +- .../Providers/PeerToPeerProvider.swift | 110 +++++++++--------- .../Providers/WebSocketProvider.swift | 34 +++--- Sources/AutomergeRepo/Repo.swift | 52 ++++----- .../Storage/DocumentStorage.swift | 1 - .../Sync/SyncV1Msg+encode+decode.swift | 22 ++-- Sources/AutomergeRepo/Sync/SyncV1Msg.swift | 1 - .../extensions/OSLog+extensions.swift | 3 + 8 files changed, 114 insertions(+), 119 deletions(-) diff --git a/Sources/AutomergeRepo/Networking/NetworkSubsystem.swift b/Sources/AutomergeRepo/Networking/NetworkSubsystem.swift index 4e44fd2..7797702 100644 --- a/Sources/AutomergeRepo/Networking/NetworkSubsystem.swift +++ b/Sources/AutomergeRepo/Networking/NetworkSubsystem.swift @@ -56,13 +56,13 @@ final class NetworkSubsystem { // invariant that there should be a valid doc handle available from the repo throw Errors.Unavailable(id: id) } - Logger.network.trace("REPONET - Initiating remote fetch for \(id)") + Logger.network.trace("REPONET: Initiating remote fetch for \(id)") let newDocument = Document() for adapter in adapters { for peerConnection in adapter.peeredConnections { Logger.network .trace( - "REPONET - requesting \(id) from peer \(peerConnection.peerId) at \(peerConnection.endpoint)" + "REPONET: requesting \(id) from peer \(peerConnection.peerId) at \(peerConnection.endpoint)" ) // upsert the requested document into the list by peer if var existingList = requestedDocuments[id] { @@ -84,7 +84,7 @@ final class NetworkSubsystem { } } } - Logger.network.trace("REPONET - remote fetch for \(id) finished") + Logger.network.trace("REPONET: remote fetch for \(id) finished") } func send(message: SyncV1Msg, to: PEER_ID?) async { @@ -126,7 +126,7 @@ extension NetworkSubsystem: NetworkEventReceiver { // ERROR FOR THESE MSG TYPES - expected to be handled at adapter Logger.network .error( - "Unexpected message type received by network subsystem: \(payload.debugDescription, privacy: .public)" + "REPONET: Unexpected message type received by network subsystem: \(payload.debugDescription, privacy: .public)" ) #if DEBUG fatalError("UNEXPECTED MSG") @@ -148,7 +148,7 @@ extension NetworkSubsystem: NetworkEventReceiver { ) return } - Logger.network.trace("Received \(event.debugDescription) event") + Logger.network.trace("REPONET: Received \(event.debugDescription) event") if let peersRequested = requestedDocuments[docId] { Logger.network.trace("REPONET: We've requested \(docId) from \(peersRequested.count) peers:") for p in peersRequested { diff --git a/Sources/AutomergeRepo/Networking/Providers/PeerToPeerProvider.swift b/Sources/AutomergeRepo/Networking/Providers/PeerToPeerProvider.swift index 251c473..73db692 100644 --- a/Sources/AutomergeRepo/Networking/Providers/PeerToPeerProvider.swift +++ b/Sources/AutomergeRepo/Networking/Providers/PeerToPeerProvider.swift @@ -181,7 +181,7 @@ public final class PeerToPeerProvider: NetworkProvider { for holder in connections.values { holder.connection.cancel() } - Logger.peerProtocol.debug("Terminating \(self.connections.count) connections") + Logger.peerProtocol.debug("P2PNET: Terminating \(self.connections.count) connections") connections.removeAll() // could be connectionPublisher.send(allConnections()), but we just removed them all... connectionPublisher.send([]) @@ -193,12 +193,12 @@ public final class PeerToPeerProvider: NetworkProvider { /// connected peers. public func send(message: SyncV1Msg, to peer: PEER_ID?) async { if let peerId = peer { - Logger.peerProtocol.trace("Sending \(message.debugDescription) to peer \(peerId)") + Logger.peerProtocol.trace("P2PNET: Sending \(message.debugDescription) to peer \(peerId)") let holdersWithPeer: [PeerToPeerConnection] = connections.values.filter { h in h.peerId == peerId } if holdersWithPeer.isEmpty { - Logger.peerProtocol.warning("Unable to find a connection to peer \(peerId)") + Logger.peerProtocol.warning("P2PNET: Unable to find a connection to peer \(peerId)") for c in connections.values { Logger.peerProtocol .warning( @@ -213,14 +213,14 @@ public final class PeerToPeerProvider: NetworkProvider { } catch { Logger.peerProtocol .warning( - "error encoding message \(targetedMessage.debugDescription, privacy: .public). Unable to send to peer \(peerId)" + "P2PNET: error encoding message \(targetedMessage.debugDescription, privacy: .public). Unable to send to peer \(peerId)" ) } } } } else { // nil peerId means send to everyone... - Logger.peerProtocol.trace("Sending \(message.debugDescription) to all peers") + Logger.peerProtocol.trace("P2PNET: Sending \(message.debugDescription) to all peers") for holder in connections.values { // only send to connections with a set PeerId if let peerId = holder.peerId { @@ -230,7 +230,7 @@ public final class PeerToPeerProvider: NetworkProvider { } catch { Logger.peerProtocol .warning( - "error encoding message \(targetedMessage.debugDescription, privacy: .public). Unable to send to endpoint \(peerId)" + "P2PNET: error encoding message \(targetedMessage.debugDescription, privacy: .public). Unable to send to endpoint \(peerId)" ) } } @@ -293,16 +293,16 @@ public final class PeerToPeerProvider: NetworkProvider { throw Errors.NetworkProviderError(msg: "No peer name is set on the provider") } - Logger.peerProtocol.debug("Starting Bonjour browser") + Logger.peerProtocol.debug("P2PNET: Starting Bonjour browser") if browser == nil { self.startBrowsing() } - Logger.peerProtocol.debug("Starting Bonjour listener as \(self.peerName)") - Logger.peerProtocol.debug(" - PeerId: \(self.peerId ?? "unset")") - Logger.peerProtocol.debug(" - PeerMetadata: \(self.peerMetadata?.debugDescription ?? "nil")") - Logger.peerProtocol.debug(" - Autoconnect on appearing host: \(self.config.autoconnect)") - Logger.peerProtocol.debug(" - Delegate: \(String(describing: self.delegate))") + Logger.peerProtocol.debug("P2PNET: Starting Bonjour listener as \(self.peerName)") + Logger.peerProtocol.debug("P2PNET: - PeerId: \(self.peerId ?? "unset")") + Logger.peerProtocol.debug("P2PNET: - PeerMetadata: \(self.peerMetadata?.debugDescription ?? "nil")") + Logger.peerProtocol.debug("P2PNET: - Autoconnect on appearing host: \(self.config.autoconnect)") + Logger.peerProtocol.debug("P2PNET: - Delegate: \(String(describing: self.delegate))") if listener == nil { self.setupBonjourListener() } @@ -312,11 +312,11 @@ public final class PeerToPeerProvider: NetworkProvider { /// /// This terminates all connections, incoming and outgoing, and disables future connections. public func stopListening() { - Logger.peerProtocol.debug("Stopping Bonjour browser") + Logger.peerProtocol.debug("P2PNET: Stopping Bonjour browser") self.stopBrowsing() browser = nil - Logger.peerProtocol.debug("Stopping Bonjour listener") + Logger.peerProtocol.debug("P2PNET: Stopping Bonjour listener") disconnect() listener?.cancel() listener = nil @@ -349,7 +349,7 @@ public final class PeerToPeerProvider: NetworkProvider { // start process to "peer" with endpoint Logger.peerProtocol .trace( - "Connection established, requesting peering with \(destination.debugDescription, privacy: .public)" + "P2PNET: Connection established, requesting peering with \(destination.debugDescription, privacy: .public)" ) // since we initiated the connection, it's on us to send an initial 'join' // protocol message to start the handshake phase of the protocol @@ -381,7 +381,7 @@ public final class PeerToPeerProvider: NetworkProvider { peered: peerConnection.peered ) await delegate.receiveEvent(event: .ready(payload: peerConnectionDetails)) - Logger.peerProtocol.trace("Peered to: \(peerMsg.senderId) \(peerMsg.debugDescription)") + Logger.peerProtocol.trace("P2PNET: Peered to: \(peerMsg.senderId) \(peerMsg.debugDescription)") connectionPublisher.send(allConnections()) return true @@ -395,7 +395,7 @@ public final class PeerToPeerProvider: NetworkProvider { connectionPublisher.send(allConnections()) Logger.peerProtocol .error( - "Failed to peer with \(destination.debugDescription, privacy: .public): \(error.localizedDescription, privacy: .public)" + "P2PNET: Failed to peer with \(destination.debugDescription, privacy: .public): \(error.localizedDescription, privacy: .public)" ) throw error } @@ -450,7 +450,7 @@ public final class PeerToPeerProvider: NetworkProvider { await handleMessage(msg: msg) } catch { // error scenario with the connection - Logger.peerProtocol.warning("Error reading from connection: \(error.localizedDescription)") + Logger.peerProtocol.warning("P2PNET: Error reading from connection: \(error.localizedDescription)") // update the stored copy of the holder with peered as false to indicate a // broken connection that can be re-attempted holder.peered = false @@ -458,7 +458,7 @@ public final class PeerToPeerProvider: NetworkProvider { connectionPublisher.send(allConnections()) } } - Logger.peerProtocol.log("receive and reconnect loop terminated") + Logger.peerProtocol.log("P2PNET: receive and reconnect loop terminated") } private func handleMessage(msg: SyncV1Msg) async { @@ -468,12 +468,12 @@ public final class PeerToPeerProvider: NetworkProvider { // - otherwise forward the message to the delegate to work with switch msg { case let .leave(msg): - Logger.peerProtocol.trace("\(msg.senderId) requests to kill the connection") + Logger.peerProtocol.trace("P2PNET: \(msg.senderId) requests to kill the connection") disconnect(peerId: msg.senderId) case let .join(msg): - Logger.peerProtocol.error("Unexpected message received: \(msg.debugDescription)") + Logger.peerProtocol.error("P2PNET: Unexpected message received: \(msg.debugDescription)") case let .peer(msg): - Logger.peerProtocol.error("Unexpected message received: \(msg.debugDescription)") + Logger.peerProtocol.error("P2PNET: Unexpected message received: \(msg.debugDescription)") default: await delegate?.receiveEvent(event: .message(payload: msg)) } @@ -515,24 +515,24 @@ public final class PeerToPeerProvider: NetworkProvider { self.browserResultUpdateContinuation.yield(BrowserResultUpdate(newResults: results, changes: changes)) } - Logger.peerProtocol.info("Activating NWBrowser \(newNetworkBrowser.debugDescription, privacy: .public)") + Logger.peerProtocol.info("P2PNET: Activating NWBrowser \(newNetworkBrowser.debugDescription, privacy: .public)") browser = newNetworkBrowser // Start browsing and ask for updates on the main queue. newNetworkBrowser.start(queue: .main) } private func reactToNWBrowserStateUpdate(_ newState: NWBrowser.State) async { - Logger.peerProtocol.trace("\(self.peerName) NWBrowser state -> \(String(describing: newState))") + Logger.peerProtocol.trace("P2PNET: \(self.peerName) NWBrowser state -> \(String(describing: newState))") browserStatePublisher.send(newState) switch newState { case let .failed(error): // Restart the browser if it loses its connection. if error == NWError.dns(DNSServiceErrorType(kDNSServiceErr_DefunctConnection)) { - Logger.peerProtocol.info("Browser failed with \(error, privacy: .public), restarting") + Logger.peerProtocol.info("P2PNET: Browser failed with \(error, privacy: .public), restarting") self.browser?.cancel() self.startBrowsing() } else { - Logger.peerProtocol.warning("Browser failed with \(error, privacy: .public), stopping") + Logger.peerProtocol.warning("P2PNET: Browser failed with \(error, privacy: .public), stopping") self.browser?.cancel() } case .ready: @@ -556,7 +556,7 @@ public final class PeerToPeerProvider: NetworkProvider { private func handleNWBrowserUpdates(_ update: BrowserResultUpdate) async { Logger.peerProtocol - .debug("\(self.peerName) NWBrowser update with \(update.newResults.count, privacy: .public) result(s):") + .debug("P2PNET: NWBrowser update with \(update.newResults.count, privacy: .public) result(s):") let availablePeers = update.newResults.compactMap { browserResult in Logger.peerProtocol @@ -578,13 +578,13 @@ public final class PeerToPeerProvider: NetworkProvider { do { Logger.peerProtocol .debug( - "\(self.peerName) AutoConnect attempting to connect to \(availablePeer.debugDescription)" + "P2PNET: AutoConnect attempting to connect to \(availablePeer.debugDescription)" ) try await connect(to: result.endpoint) } catch { Logger.peerProtocol .warning( - "Failed to connect to \(result.endpoint.debugDescription): \(error.localizedDescription)" + "P2PNET: Failed to connect to \(result.endpoint.debugDescription): \(error.localizedDescription)" ) } } @@ -595,7 +595,7 @@ public final class PeerToPeerProvider: NetworkProvider { fileprivate func stopBrowsing() { guard let browser else { return } - Logger.peerProtocol.info("Terminating NWBrowser") + Logger.peerProtocol.info("P2PNET: Terminating NWBrowser") browser.cancel() self.browser = nil } @@ -614,21 +614,21 @@ public final class PeerToPeerProvider: NetworkProvider { case .ready: if let port = listener.port { Logger.peerProtocol - .info("Bonjour listener ready on \(port.rawValue, privacy: .public)") + .info("P2PNET: Bonjour listener ready on \(port.rawValue, privacy: .public)") } else { Logger.peerProtocol - .info("Bonjour listener ready (no port listed)") + .info("P2PNET: Bonjour listener ready (no port listed)") } case let .failed(error): if error == NWError.dns(DNSServiceErrorType(kDNSServiceErr_DefunctConnection)) { Logger.peerProtocol - .warning("Bonjour listener failed with \(error, privacy: .public), restarting.") + .warning("P2PNET: Bonjour listener failed with \(error, privacy: .public), restarting.") listener.cancel() self.listener = nil self.setupBonjourListener() } else { Logger.peerProtocol - .error("Bonjour listener failed with \(error, privacy: .public), stopping.") + .error("P2PNET: Bonjour listener failed with \(error, privacy: .public), stopping.") listener.cancel() } case .setup: @@ -679,11 +679,11 @@ public final class PeerToPeerProvider: NetworkProvider { listener.start(queue: .main) self.listener = listener Logger.peerProtocol - .debug("Starting bonjour network listener") + .debug("P2PNET: Starting bonjour network listener") } catch { Logger.peerProtocol - .critical("Failed to create bonjour listener") + .critical("P2PNET: Failed to create bonjour listener") } } @@ -695,25 +695,25 @@ public final class PeerToPeerProvider: NetworkProvider { } Logger.peerProtocol .debug( - "Receiving connection request from \(newConnection.endpoint.debugDescription, privacy: .public)" + "P2PNET: Receiving connection request from \(newConnection.endpoint.debugDescription, privacy: .public)" ) Logger.peerProtocol .debug( - " Connection details: \(newConnection.debugDescription, privacy: .public)" + "P2PNET: Connection details: \(newConnection.debugDescription, privacy: .public)" ) - Logger.peerProtocol.debug("Existing connections:") - Logger.peerProtocol.debug("----------------------------------------------------------") + Logger.peerProtocol.debug("P2PNET: Existing connections:") + Logger.peerProtocol.debug("P2PNET: ----------------------------------------------------------") for (k, v) in connections { let peeredString = v.peered ? "true" : "false" let initiatedString = v.initiated ? "true" : "false" let peerString = v.peerId ?? "nil" - Logger.peerProtocol.debug("\(k.debugDescription)") - Logger.peerProtocol.debug(" :: peerId: \(peerString)") - Logger.peerProtocol.debug(" :: initiated: \(initiatedString)") - Logger.peerProtocol.debug(" :: peered: \(peeredString)") - Logger.peerProtocol.debug("----------------------------------------------------------") + Logger.peerProtocol.debug("P2PNET: \(k.debugDescription)") + Logger.peerProtocol.debug("P2PNET: :: peerId: \(peerString)") + Logger.peerProtocol.debug("P2PNET: :: initiated: \(initiatedString)") + Logger.peerProtocol.debug("P2PNET: :: peered: \(peeredString)") + Logger.peerProtocol.debug("P2PNET: ----------------------------------------------------------") } // check to see if there's already a connection with this endpoint, if there is @@ -721,15 +721,9 @@ public final class PeerToPeerProvider: NetworkProvider { if connections[newConnection.endpoint] == nil { Logger.peerProtocol .info( - "Endpoint not yet recorded, accepting connection from \(newConnection.endpoint.debugDescription, privacy: .public)" + "P2PNET: Endpoint not yet recorded, accepting connection from \(newConnection.endpoint.debugDescription, privacy: .public)" ) let peerConnection = PeerToPeerConnection(connection: newConnection) -// let holder = ConnectionHolder( -// connection: peerConnection, -// initiated: false, -// peered: false, -// endpoint: newConnection.endpoint -// ) connections[newConnection.endpoint] = peerConnection connectionPublisher.send(allConnections()) @@ -751,7 +745,7 @@ public final class PeerToPeerProvider: NetworkProvider { } else { Logger.peerProtocol .info( - "Inbound connection already exists for \(newConnection.endpoint.debugDescription, privacy: .public), cancelling the connection request." + "P2PNET: Inbound connection already exists for \(newConnection.endpoint.debugDescription, privacy: .public), cancelling the connection request." ) // If we already have a connection to that endpoint, don't add another newConnection.cancel() @@ -806,7 +800,7 @@ public final class PeerToPeerProvider: NetworkProvider { connectionPublisher.send(allConnections()) Logger.peerProtocol - .trace("Accepting peer connection from \(holder.endpoint.debugDescription, privacy: .public)") + .trace("P2PNET: Accepting peer connection from \(holder.endpoint.debugDescription, privacy: .public)") // reply with the corresponding "peer" message let peerMessage = SyncV1Msg.PeerMsg( @@ -817,7 +811,7 @@ public final class PeerToPeerProvider: NetworkProvider { ) try await holder.send(.peer(peerMessage)) - Logger.peerProtocol.trace("SEND: \(peerMessage.debugDescription)") + Logger.peerProtocol.trace("P2PNET: SEND: \(peerMessage.debugDescription)") return peerConnectionDetails } @@ -839,12 +833,12 @@ public final class PeerToPeerProvider: NetworkProvider { await handleMessage(msg: msg) } catch { // error scenario with the PeerToPeer connection - Logger.peerProtocol.warning("Error reading connection: \(error.localizedDescription)") + Logger.peerProtocol.warning("P2PNET: Error reading connection: \(error.localizedDescription)") disconnect(peerId: peerId) break } } - Logger.peerProtocol.warning("receive and reconnect loop for \(endpoint.debugDescription) terminated") + Logger.peerProtocol.warning("P2PNET: receive and reconnect loop for \(endpoint.debugDescription) terminated") } // Update the advertised name on the network. @@ -857,6 +851,6 @@ public final class PeerToPeerProvider: NetworkProvider { type: P2PAutomergeSyncProtocol.bonjourType, txtRecord: txtRecord ) - Logger.peerProtocol.info("Updated bonjour network listener to advertise name \(name, privacy: .public)") + Logger.peerProtocol.info("P2PNET: Updated bonjour network listener to advertise name \(name, privacy: .public)") } } diff --git a/Sources/AutomergeRepo/Networking/Providers/WebSocketProvider.swift b/Sources/AutomergeRepo/Networking/Providers/WebSocketProvider.swift index 3085464..172771a 100644 --- a/Sources/AutomergeRepo/Networking/Providers/WebSocketProvider.swift +++ b/Sources/AutomergeRepo/Networking/Providers/WebSocketProvider.swift @@ -105,7 +105,7 @@ public final class WebSocketProvider: NetworkProvider { /// connected peers. public func send(message: SyncV1Msg, to: PEER_ID?) async { guard let webSocketTask, let peer = peerId else { - Logger.webSocket.warning("Attempt to send a message without a connection or defined remote peer") + Logger.webSocket.warning("WEBSOCKET: Attempt to send a message without a connection or defined remote peer") return } let msgToSend = message.setTarget(to ?? peer) @@ -113,7 +113,7 @@ public final class WebSocketProvider: NetworkProvider { let data = try SyncV1Msg.encode(msgToSend) try await webSocketTask.send(.data(data)) } catch { - Logger.webSocket.error("Unable to encode and send message: \(error.localizedDescription, privacy: .public)") + Logger.webSocket.error("WEBSOCKET: Unable to encode and send message: \(error.localizedDescription, privacy: .public)") } } @@ -153,7 +153,7 @@ public final class WebSocketProvider: NetworkProvider { let decodeAttempted = SyncV1Msg.decode(raw_data) Logger.webSocket .warning( - "Decoding websocket message, expecting peer only - and it wasn't a peer message. RECEIVED MSG: \(String(describing: decodeAttempted))" + "WEBSOCKET: Decoding message, expecting peer only - and it wasn't a peer message. RECEIVED MSG: \(String(describing: decodeAttempted))" ) throw Errors.UnexpectedMsg(msg: String(describing: decodeAttempted)) } @@ -168,12 +168,12 @@ public final class WebSocketProvider: NetworkProvider { case let .string(string): // In the handshake phase and received anything other than a valid peer message Logger.webSocket - .warning("Unknown websocket message received: .string(\(string))") + .warning("WEBSOCKET: Unknown message received: .string(\(string))") throw Errors.UnexpectedMsg(msg: string) @unknown default: // In the handshake phase and received anything other than a valid peer message Logger.webSocket - .error("Unknown websocket message received: \(String(describing: msg))") + .error("WEBSOCKET: Unknown message received: \(String(describing: msg))") throw Errors.UnexpectedMsg(msg: String(describing: msg)) } } @@ -194,7 +194,7 @@ public final class WebSocketProvider: NetworkProvider { let request = URLRequest(url: url) let webSocketTask = URLSession.shared.webSocketTask(with: request) - Logger.webSocket.trace("Activating websocket to \(url, privacy: .public)") + Logger.webSocket.trace("WEBSOCKET: Activating websocket to \(url, privacy: .public)") // start the websocket processing things webSocketTask.resume() @@ -203,7 +203,7 @@ public final class WebSocketProvider: NetworkProvider { let joinMessage = SyncV1Msg.JoinMsg(senderId: peerId, metadata: peerMetadata) let data = try SyncV1Msg.encode(joinMessage) try await webSocketTask.send(.data(data)) - Logger.webSocket.trace("SEND: \(joinMessage.debugDescription)") + Logger.webSocket.trace("WEBSOCKET: SEND: \(joinMessage.debugDescription)") do { // Race a timeout against receiving a Peer message from the other side @@ -228,7 +228,7 @@ public final class WebSocketProvider: NetworkProvider { ) peeredConnections = [peerConnectionDetails] await delegate.receiveEvent(event: .ready(payload: peerConnectionDetails)) - Logger.webSocket.trace("Peered to targetId: \(peerMsg.senderId) \(peerMsg.debugDescription)") + Logger.webSocket.trace("WEBSOCKET: Peered to targetId: \(peerMsg.senderId) \(peerMsg.debugDescription)") } catch { // if there's an error, disconnect anything that's lingering and cancel it down. // an error here means we contacted the server successfully, but were unable to @@ -237,7 +237,7 @@ public final class WebSocketProvider: NetworkProvider { // which will force us to fail reconnects. Logger.webSocket .error( - "Failed to peer with \(url.absoluteString, privacy: .public): \(error.localizedDescription, privacy: .public)" + "WEBSOCKET: Failed to peer with \(url.absoluteString, privacy: .public): \(error.localizedDescription, privacy: .public)" ) await disconnect() throw error @@ -331,7 +331,7 @@ public final class WebSocketProvider: NetworkProvider { } guard let webSocketTask else { - Logger.webSocket.warning("Receive Handler: webSocketTask is nil, terminating handler loop") + Logger.webSocket.warning("WEBSOCKET: Receive Handler: webSocketTask is nil, terminating handler loop") break // terminates the while loop - no more reconnect attempts } @@ -341,7 +341,7 @@ public final class WebSocketProvider: NetworkProvider { msgFromWebSocket = try await webSocketTask.receive() } catch { // error scenario with the WebSocket connection - Logger.webSocket.warning("Error reading websocket: \(error.localizedDescription)") + Logger.webSocket.warning("WEBSOCKET: Error reading websocket: \(error.localizedDescription)") } if let encodedMessage = msgFromWebSocket { @@ -352,11 +352,11 @@ public final class WebSocketProvider: NetworkProvider { // catch decode failures, but don't terminate the whole shebang // on a failure Logger.webSocket - .warning("Unable to decode websocket message: \(error.localizedDescription, privacy: .public)") + .warning("WEBSOCKET: Unable to decode websocket message: \(error.localizedDescription, privacy: .public)") } } } - Logger.webSocket.log("receive and reconnect loop terminated") + Logger.webSocket.log("WEBSOCKET: receive and reconnect loop terminated") } func handleMessage(msg: SyncV1Msg) async { @@ -364,15 +364,15 @@ public final class WebSocketProvider: NetworkProvider { // in this method (all handling of them should happen before getting here) // - .leave invokes the disconnect, and associated messages to the delegate // - otherwise forward the message to the delegate to work with - Logger.webSocket.trace("WebSocket received: \(msg.debugDescription)") + Logger.webSocket.trace("WEBSOCKET: handle message: \(msg.debugDescription)") switch msg { case let .leave(msg): - Logger.webSocket.trace("\(msg.senderId) requests to kill the connection") + Logger.webSocket.trace("WEBSOCKET: \(msg.senderId) requests to kill the connection") await disconnect() case let .join(msg): - Logger.webSocket.error("Unexpected message received: \(msg.debugDescription)") + Logger.webSocket.error("WEBSOCKET: Unexpected message received: \(msg.debugDescription)") case let .peer(msg): - Logger.webSocket.error("Unexpected message received: \(msg.debugDescription)") + Logger.webSocket.error("WEBSOCKET: Unexpected message received: \(msg.debugDescription)") default: await delegate?.receiveEvent(event: .message(payload: msg)) } diff --git a/Sources/AutomergeRepo/Repo.swift b/Sources/AutomergeRepo/Repo.swift index 7d1360c..3e59d0d 100644 --- a/Sources/AutomergeRepo/Repo.swift +++ b/Sources/AutomergeRepo/Repo.swift @@ -270,16 +270,16 @@ public final class Repo { // MARK: Synchronization Pieces - For Network Subsystem Access func handleSync(msg: SyncV1Msg.SyncMsg) async { - Logger.repo.trace("PEER: \(self.peerId) - handling a sync msg from \(msg.senderId) to \(msg.targetId)") + Logger.repo.trace("REPO: \(self.peerId) - handling a sync msg from \(msg.senderId) to \(msg.targetId)") guard let docId = DocumentId(msg.documentId) else { Logger.repo - .warning("Invalid documentId \(msg.documentId) received in a sync message \(msg.debugDescription)") + .warning("REPO: Invalid documentId \(msg.documentId) received in a sync message \(msg.debugDescription)") return } - Logger.repo.trace(" - Sync request received for document \(docId)") + Logger.repo.trace("REPO: - Sync request received for document \(docId)") do { if handles[docId] == nil { - Logger.repo.trace(" - No recorded handle for \(docId), creating one") + Logger.repo.trace("REPO: - No recorded handle for \(docId), creating one") // There is no in-memory handle for the document being synced, so this is a request // to create a local copy of the document encapsulated in the sync message. let newDocument = Document() @@ -291,7 +291,7 @@ public final class Repo { _ = try await resolveDocHandle(id: docId) } guard let handle = handles[docId] else { fatalError("HANDLE DOESN'T EXIST") } - Logger.repo.trace(" - working on handle for \(docId), state: \(String(describing: handle.state))") + Logger.repo.trace("REPO: - working on handle for \(docId), state: \(String(describing: handle.state))") let docFromHandle = handle.doc ?? Document() let syncState = syncState(id: docId, peer: msg.senderId) // Apply the request message as a sync update @@ -312,7 +312,7 @@ public final class Repo { targetId: msg.senderId, sync_message: syncData )) - Logger.repo.trace("Sync received and applied, replying with a sync msg back to \(msg.senderId)") + Logger.repo.trace("REPO: Sync received and applied, replying with a sync msg back to \(msg.senderId)") await network.send(message: syncMsg, to: msg.senderId) } // else no sync is needed, as the last sync state reports that it knows about @@ -320,7 +320,7 @@ public final class Repo { } catch { let err: SyncV1Msg = .error(.init(message: "Error receiving sync: \(error.localizedDescription)")) - Logger.repo.warning("Error receiving initial sync for \(docId, privacy: .public)") + Logger.repo.warning("REPO: Error receiving initial sync for \(docId, privacy: .public)") await network.send(message: err, to: msg.senderId) } } @@ -328,7 +328,7 @@ public final class Repo { func handleRequest(msg: SyncV1Msg.RequestMsg) async { guard let docId = DocumentId(msg.documentId) else { Logger.repo - .warning("Invalid documentId \(msg.documentId) received in a sync message \(msg.debugDescription)") + .warning("REPO: Invalid documentId \(msg.documentId) received in a sync message \(msg.debugDescription)") return } if handles[docId] != nil { @@ -543,12 +543,12 @@ public final class Repo { fatalError("No stored dochandle for id: \(id)") } if let handleSyncState = handle.syncStates[peer] { - Logger.repo.trace("Providing stored sync state for doc \(id)") + Logger.repo.trace("REPO: Providing stored sync state for doc \(id)") return handleSyncState } else { // TODO: add attempt to load from storage and return it before creating a new one - Logger.repo.trace("No stored sync state for doc \(id) and peer \(peer).") - Logger.repo.trace("Creating a new sync state for doc \(id)") + Logger.repo.trace("REPO: No stored sync state for doc \(id) and peer \(peer).") + Logger.repo.trace("REPO: Creating a new sync state for doc \(id)") return SyncState() } } @@ -557,14 +557,14 @@ public final class Repo { guard let handle = handles[id] else { fatalError("No stored dochandle for id: \(id)") } - Logger.repo.trace("Storing updated sync state for doc \(id) and peer \(peer).") + Logger.repo.trace("REPO: Storing updated sync state for doc \(id) and peer \(peer).") handle.syncStates[peer] = syncState } func markDocUnavailable(id: DocumentId) async { // handling a requested document being marked as unavailable after all peers have been checked guard let handle = handles[id] else { - Logger.repo.error("missing handle for documentId \(id.description) while attempt to mark unavailable") + Logger.repo.error("REPO: missing handle for documentId \(id.description) while attempt to mark unavailable") return } assert(handle.state == .requesting) @@ -577,7 +577,7 @@ public final class Repo { guard let handle = handles[id] else { fatalError("No stored document handle for document id: \(id)") } - Logger.repo.trace("Updated contents of document \(id), state: \(String(describing: handle.state))") + Logger.repo.trace("REPO: Updated contents of document \(id), state: \(String(describing: handle.state))") // Automerge-repo https://github.com/automerge/automerge-repo/issues/343 is sending two responses, // the first being UNAVAILABLE, which we use to change the state, but that triggers this unexpected // assertion, we we later receive the SYNC update to set the document as expected @@ -591,7 +591,7 @@ public final class Repo { } catch { Logger.repo .warning( - "Error received while attempting to store document ID \(id): \(error.localizedDescription)" + "REPO: Error received while attempting to store document ID \(id): \(error.localizedDescription)" ) } } @@ -641,7 +641,7 @@ public final class Repo { private func resolveDocHandle(id: DocumentId) async throws -> DocHandle { if let handle: InternalDocHandle = handles[id] { - Logger.resolver.trace("RESOLVE document id \(id) [\(String(describing: handle.state))]") + Logger.resolver.trace("RESOLVE: document id \(id) [\(String(describing: handle.state))]") switch handle.state { case .idle: if handle.doc != nil { @@ -706,7 +706,7 @@ public final class Repo { } case .requesting: guard let updatedHandle = handles[id] else { - Logger.resolver.trace("RESOLVED - X :: Missing \(id) -> [UNAVAILABLE]") + Logger.resolver.error("RESOLVE: :: Missing \(id) -> [UNAVAILABLE]") throw Errors.Unavailable(id: handle.id) } if updatedHandle.doc != nil, updatedHandle.state == .ready { @@ -716,16 +716,16 @@ public final class Repo { } else { guard let previousRequests = pendingRequestReadAttempts[id] else { Logger.resolver - .trace("RESOLVED - X :: Missing \(id) from pending request read attempts -> [UNAVAILABLE]") + .error("RESOLVE: :: Missing \(id) from pending request read attempts -> [UNAVAILABLE]") throw Errors.Unavailable(id: id) } if previousRequests < maxRetriesForFetch { // we are racing against the receipt of a network result // to see what we get at the end - Logger.resolver.trace(" :: \(id) -> [\(String(describing: handle.state))]") + Logger.resolver.trace("RESOLVE: :: \(id) -> [\(String(describing: handle.state))]") Logger.resolver .trace( - " :: check # \(previousRequests) (of \(self.maxRetriesForFetch), waiting \(self.pendingRequestWaitDuration) seconds for remote fetch" + "RESOLVE: :: check # \(previousRequests) (of \(self.maxRetriesForFetch), waiting \(self.pendingRequestWaitDuration) seconds for remote fetch" ) try await Task.sleep(for: pendingRequestWaitDuration) pendingRequestReadAttempts[id] = previousRequests + 1 @@ -735,26 +735,26 @@ public final class Repo { updatedHandle.state = .unavailable docHandlePublisher.send(handle.snapshot()) Logger.resolver - .trace( - "RESOLVED - X :: failed waiting \(previousRequests) of \(self.maxRetriesForFetch) requests for \(id) -> [UNAVAILABLE]" + .error( + "RESOLVE: :: failed waiting \(previousRequests) of \(self.maxRetriesForFetch) requests for \(id) -> [UNAVAILABLE]" ) throw Errors.Unavailable(id: id) } } case .ready: guard let doc = handle.doc else { fatalError("DocHandle state is ready, but ._doc is null") } - Logger.resolver.trace("RESOLVED! :: \(id) [\(String(describing: handle.state))]") + Logger.resolver.trace("RESOLVE :: \(id) [\(String(describing: handle.state))]") watchDocForChanges(id: id) return DocHandle(id: id, doc: doc) case .unavailable: - Logger.resolver.trace("RESOLVED - X :: \(id) -> [MARKED UNAVAILABLE]") + Logger.resolver.error("RESOLVE: :: \(id) -> [MARKED UNAVAILABLE]") throw Errors.Unavailable(id: handle.id) case .deleted: - Logger.resolver.trace("RESOLVED - X :: \(id) -> [MARKED DELETED]") + Logger.resolver.error("RESOLVE: :: \(id) -> [MARKED DELETED]") throw Errors.DocDeleted(id: handle.id) } } else { - Logger.resolver.error("RESOLVED - X :: Error Resolving document: Repo doesn't have a handle for \(id).") + Logger.resolver.error("RESOLVE: :: Error Resolving document: Repo doesn't have a handle for \(id).") throw Errors.Unavailable(id: id) } } diff --git a/Sources/AutomergeRepo/Storage/DocumentStorage.swift b/Sources/AutomergeRepo/Storage/DocumentStorage.swift index 7f979e7..5aaac68 100644 --- a/Sources/AutomergeRepo/Storage/DocumentStorage.swift +++ b/Sources/AutomergeRepo/Storage/DocumentStorage.swift @@ -1,6 +1,5 @@ import Automerge import Foundation -import OSLog // inspired from automerge-repo: // https://github.com/automerge/automerge-repo/blob/main/packages/automerge-repo/src/storage/StorageSubsystem.ts diff --git a/Sources/AutomergeRepo/Sync/SyncV1Msg+encode+decode.swift b/Sources/AutomergeRepo/Sync/SyncV1Msg+encode+decode.swift index 109c2ee..5753356 100644 --- a/Sources/AutomergeRepo/Sync/SyncV1Msg+encode+decode.swift +++ b/Sources/AutomergeRepo/Sync/SyncV1Msg+encode+decode.swift @@ -48,7 +48,7 @@ public extension SyncV1Msg { do { cborMsg = try CBORSerialization.cbor(from: data) } catch { - Logger.webSocket.warning("Unable to CBOR decode incoming data: \(data)") + Logger.coder.warning("CBOR: Unable to CBOR decode incoming data: \(data)") return .unknown(data) } // read the "type" of the message in order to choose the appropriate decoding path @@ -106,7 +106,7 @@ public extension SyncV1Msg { do { return try CBORCoder.decoder.decode(SyncMsg.self, from: data) } catch { - Logger.webSocket.warning("Failed to decode data as SyncMsg") + Logger.coder.warning("CBOR: Failed to decode data as SyncMsg") } return nil } @@ -115,7 +115,7 @@ public extension SyncV1Msg { do { return try CBORCoder.decoder.decode(RequestMsg.self, from: data) } catch { - Logger.webSocket.warning("Failed to decode data as RequestMsg") + Logger.coder.warning("CBOR: Failed to decode data as RequestMsg") } return nil } @@ -124,7 +124,7 @@ public extension SyncV1Msg { do { return try CBORCoder.decoder.decode(UnavailableMsg.self, from: data) } catch { - Logger.webSocket.warning("Failed to decode data as UnavailableMsg") + Logger.coder.warning("CBOR: Failed to decode data as UnavailableMsg") } return nil } @@ -135,7 +135,7 @@ public extension SyncV1Msg { do { return try CBORCoder.decoder.decode(PeerMsg.self, from: data) } catch { - Logger.webSocket.warning("Failed to decode data as PeerMsg") + Logger.coder.warning("CBOR: Failed to decode data as PeerMsg") } return nil } @@ -144,7 +144,7 @@ public extension SyncV1Msg { do { return try CBORCoder.decoder.decode(JoinMsg.self, from: data) } catch { - Logger.webSocket.warning("Failed to decode data as JoinMsg") + Logger.coder.warning("CBOR: Failed to decode data as JoinMsg") } return nil } @@ -153,7 +153,7 @@ public extension SyncV1Msg { do { return try CBORCoder.decoder.decode(LeaveMsg.self, from: data) } catch { - Logger.webSocket.warning("Failed to decode data as LeaveMsg") + Logger.coder.warning("CBOR: Failed to decode data as LeaveMsg") } return nil } @@ -164,7 +164,7 @@ public extension SyncV1Msg { do { return try CBORCoder.decoder.decode(ErrorMsg.self, from: data) } catch { - Logger.webSocket.warning("Failed to decode data as ErrorMsg") + Logger.coder.warning("CBOR: Failed to decode data as ErrorMsg") } return nil } @@ -175,7 +175,7 @@ public extension SyncV1Msg { do { return try CBORCoder.decoder.decode(EphemeralMsg.self, from: data) } catch { - Logger.webSocket.warning("Failed to decode data as EphemeralMsg") + Logger.coder.warning("CBOR: Failed to decode data as EphemeralMsg") } return nil } @@ -186,7 +186,7 @@ public extension SyncV1Msg { do { return try CBORCoder.decoder.decode(RemoteHeadsChangedMsg.self, from: data) } catch { - Logger.webSocket.warning("Failed to decode data as RemoteHeadsChangedMsg") + Logger.coder.warning("CBOR: Failed to decode data as RemoteHeadsChangedMsg") } return nil } @@ -195,7 +195,7 @@ public extension SyncV1Msg { do { return try CBORCoder.decoder.decode(RemoteSubscriptionChangeMsg.self, from: data) } catch { - Logger.webSocket.warning("Failed to decode data as RemoteSubscriptionChangeMsg") + Logger.coder.warning("CBOR: Failed to decode data as RemoteSubscriptionChangeMsg") } return nil } diff --git a/Sources/AutomergeRepo/Sync/SyncV1Msg.swift b/Sources/AutomergeRepo/Sync/SyncV1Msg.swift index ea0722f..f389cd1 100644 --- a/Sources/AutomergeRepo/Sync/SyncV1Msg.swift +++ b/Sources/AutomergeRepo/Sync/SyncV1Msg.swift @@ -6,7 +6,6 @@ // import Foundation -import OSLog import PotentCBOR // Automerge Repo WebSocket sync details: diff --git a/Sources/AutomergeRepo/extensions/OSLog+extensions.swift b/Sources/AutomergeRepo/extensions/OSLog+extensions.swift index fa60073..fb3f06a 100644 --- a/Sources/AutomergeRepo/extensions/OSLog+extensions.swift +++ b/Sources/AutomergeRepo/extensions/OSLog+extensions.swift @@ -20,6 +20,9 @@ extension Logger { /// Logs updates and interaction related to the process of synchronization over the network. static let peerConnection = Logger(subsystem: subsystem, category: "SyncConnection") + /// Logs updates and interations performed by the sync protocol encoder and decoder. + static let coder = Logger(subsystem: subsystem, category: "SyncCoderDecoder") + /// Logs updates and interaction related to the process of synchronization over the network. static let webSocket = Logger(subsystem: subsystem, category: "WebSocket")