|
| 1 | +import Foundation |
| 2 | +import GRPC |
| 3 | +import NIO |
| 4 | +import os |
| 5 | +import Subprocess |
| 6 | + |
| 7 | +@MainActor |
| 8 | +public protocol FileSyncDaemon: ObservableObject { |
| 9 | + var state: DaemonState { get } |
| 10 | + func start() async |
| 11 | + func stop() async |
| 12 | +} |
| 13 | + |
| 14 | +@MainActor |
| 15 | +public class MutagenDaemon: FileSyncDaemon { |
| 16 | + private let logger = Logger(subsystem: Bundle.main.bundleIdentifier!, category: "mutagen") |
| 17 | + |
| 18 | + @Published public var state: DaemonState = .stopped { |
| 19 | + didSet { |
| 20 | + logger.info("daemon state changed: \(self.state.description, privacy: .public)") |
| 21 | + } |
| 22 | + } |
| 23 | + |
| 24 | + private var mutagenProcess: Subprocess? |
| 25 | + private let mutagenPath: URL! |
| 26 | + private let mutagenDataDirectory: URL |
| 27 | + private let mutagenDaemonSocket: URL |
| 28 | + |
| 29 | + private var group: MultiThreadedEventLoopGroup? |
| 30 | + private var channel: GRPCChannel? |
| 31 | + private var client: Daemon_DaemonAsyncClient? |
| 32 | + |
| 33 | + public init() { |
| 34 | + #if arch(arm64) |
| 35 | + mutagenPath = Bundle.main.url(forResource: "mutagen-darwin-arm64", withExtension: nil) |
| 36 | + #elseif arch(x86_64) |
| 37 | + mutagenPath = Bundle.main.url(forResource: "mutagen-darwin-amd64", withExtension: nil) |
| 38 | + #else |
| 39 | + fatalError("unknown architecture") |
| 40 | + #endif |
| 41 | + mutagenDataDirectory = FileManager.default.urls( |
| 42 | + for: .applicationSupportDirectory, |
| 43 | + in: .userDomainMask |
| 44 | + ).first!.appending(path: "Coder Desktop").appending(path: "Mutagen") |
| 45 | + mutagenDaemonSocket = mutagenDataDirectory.appending(path: "daemon").appending(path: "daemon.sock") |
| 46 | + // It shouldn't be fatal if the app was built without Mutagen embedded, |
| 47 | + // but file sync will be unavailable. |
| 48 | + if mutagenPath == nil { |
| 49 | + logger.warning("Mutagen not embedded in app, file sync will be unavailable") |
| 50 | + state = .unavailable |
| 51 | + } |
| 52 | + } |
| 53 | + |
| 54 | + public func start() async { |
| 55 | + if case .unavailable = state { return } |
| 56 | + |
| 57 | + // Stop an orphaned daemon, if there is one |
| 58 | + try? await connect() |
| 59 | + await stop() |
| 60 | + |
| 61 | + mutagenProcess = createMutagenProcess() |
| 62 | + // swiftlint:disable:next large_tuple |
| 63 | + let (standardOutput, standardError, waitForExit): (Pipe.AsyncBytes, Pipe.AsyncBytes, @Sendable () async -> Void) |
| 64 | + do { |
| 65 | + (standardOutput, standardError, waitForExit) = try mutagenProcess!.run() |
| 66 | + } catch { |
| 67 | + state = .failed(DaemonError.daemonStartFailure(error)) |
| 68 | + return |
| 69 | + } |
| 70 | + |
| 71 | + Task { |
| 72 | + await streamHandler(io: standardOutput) |
| 73 | + logger.info("standard output stream closed") |
| 74 | + } |
| 75 | + |
| 76 | + Task { |
| 77 | + await streamHandler(io: standardError) |
| 78 | + logger.info("standard error stream closed") |
| 79 | + } |
| 80 | + |
| 81 | + Task { |
| 82 | + await terminationHandler(waitForExit: waitForExit) |
| 83 | + } |
| 84 | + |
| 85 | + do { |
| 86 | + try await connect() |
| 87 | + } catch { |
| 88 | + state = .failed(DaemonError.daemonStartFailure(error)) |
| 89 | + return |
| 90 | + } |
| 91 | + |
| 92 | + state = .running |
| 93 | + logger.info( |
| 94 | + """ |
| 95 | + mutagen daemon started, pid: |
| 96 | + \(self.mutagenProcess?.pid.description ?? "unknown", privacy: .public) |
| 97 | + """ |
| 98 | + ) |
| 99 | + } |
| 100 | + |
| 101 | + private func connect() async throws(DaemonError) { |
| 102 | + guard client == nil else { |
| 103 | + // Already connected |
| 104 | + return |
| 105 | + } |
| 106 | + group = MultiThreadedEventLoopGroup(numberOfThreads: 1) |
| 107 | + do { |
| 108 | + channel = try GRPCChannelPool.with( |
| 109 | + target: .unixDomainSocket(mutagenDaemonSocket.path), |
| 110 | + transportSecurity: .plaintext, |
| 111 | + eventLoopGroup: group! |
| 112 | + ) |
| 113 | + client = Daemon_DaemonAsyncClient(channel: channel!) |
| 114 | + logger.info( |
| 115 | + "Successfully connected to mutagen daemon, socket: \(self.mutagenDaemonSocket.path, privacy: .public)" |
| 116 | + ) |
| 117 | + } catch { |
| 118 | + logger.error("Failed to connect to gRPC: \(error)") |
| 119 | + try? await cleanupGRPC() |
| 120 | + throw DaemonError.connectionFailure(error) |
| 121 | + } |
| 122 | + } |
| 123 | + |
| 124 | + private func cleanupGRPC() async throws { |
| 125 | + try? await channel?.close().get() |
| 126 | + try? await group?.shutdownGracefully() |
| 127 | + |
| 128 | + client = nil |
| 129 | + channel = nil |
| 130 | + group = nil |
| 131 | + } |
| 132 | + |
| 133 | + public func stop() async { |
| 134 | + if case .unavailable = state { return } |
| 135 | + state = .stopped |
| 136 | + guard FileManager.default.fileExists(atPath: mutagenDaemonSocket.path) else { |
| 137 | + // Already stopped |
| 138 | + return |
| 139 | + } |
| 140 | + |
| 141 | + // "We don't check the response or error, because the daemon |
| 142 | + // may terminate before it has a chance to send the response." |
| 143 | + _ = try? await client?.terminate( |
| 144 | + Daemon_TerminateRequest(), |
| 145 | + callOptions: .init(timeLimit: .timeout(.milliseconds(500))) |
| 146 | + ) |
| 147 | + |
| 148 | + try? await cleanupGRPC() |
| 149 | + |
| 150 | + mutagenProcess?.kill() |
| 151 | + mutagenProcess = nil |
| 152 | + logger.info("Daemon stopped and gRPC connection closed") |
| 153 | + } |
| 154 | + |
| 155 | + private func createMutagenProcess() -> Subprocess { |
| 156 | + let process = Subprocess([mutagenPath.path, "daemon", "run"]) |
| 157 | + process.environment = [ |
| 158 | + "MUTAGEN_DATA_DIRECTORY": mutagenDataDirectory.path, |
| 159 | + ] |
| 160 | + logger.info("setting mutagen data directory: \(self.mutagenDataDirectory.path, privacy: .public)") |
| 161 | + return process |
| 162 | + } |
| 163 | + |
| 164 | + private func terminationHandler(waitForExit: @Sendable () async -> Void) async { |
| 165 | + await waitForExit() |
| 166 | + |
| 167 | + switch state { |
| 168 | + case .stopped: |
| 169 | + logger.info("mutagen daemon stopped") |
| 170 | + default: |
| 171 | + logger.error( |
| 172 | + """ |
| 173 | + mutagen daemon exited unexpectedly with code: |
| 174 | + \(self.mutagenProcess?.exitCode.description ?? "unknown") |
| 175 | + """ |
| 176 | + ) |
| 177 | + state = .failed(.terminatedUnexpectedly) |
| 178 | + } |
| 179 | + } |
| 180 | + |
| 181 | + private func streamHandler(io: Pipe.AsyncBytes) async { |
| 182 | + for await line in io.lines { |
| 183 | + logger.info("\(line, privacy: .public)") |
| 184 | + } |
| 185 | + } |
| 186 | +} |
| 187 | + |
| 188 | +public enum DaemonState { |
| 189 | + case running |
| 190 | + case stopped |
| 191 | + case failed(DaemonError) |
| 192 | + case unavailable |
| 193 | + |
| 194 | + var description: String { |
| 195 | + switch self { |
| 196 | + case .running: |
| 197 | + "Running" |
| 198 | + case .stopped: |
| 199 | + "Stopped" |
| 200 | + case let .failed(error): |
| 201 | + "Failed: \(error)" |
| 202 | + case .unavailable: |
| 203 | + "Unavailable" |
| 204 | + } |
| 205 | + } |
| 206 | +} |
| 207 | + |
| 208 | +public enum DaemonError: Error { |
| 209 | + case daemonStartFailure(Error) |
| 210 | + case connectionFailure(Error) |
| 211 | + case terminatedUnexpectedly |
| 212 | + |
| 213 | + var description: String { |
| 214 | + switch self { |
| 215 | + case let .daemonStartFailure(error): |
| 216 | + "Daemon start failure: \(error)" |
| 217 | + case let .connectionFailure(error): |
| 218 | + "Connection failure: \(error)" |
| 219 | + case .terminatedUnexpectedly: |
| 220 | + "Daemon terminated unexpectedly" |
| 221 | + } |
| 222 | + } |
| 223 | + |
| 224 | + var localizedDescription: String { description } |
| 225 | +} |
0 commit comments