import ByreAPI import Foundation private let maxOutput = 64 << 10 private let runsDir = FileManager.default.urls(for: .cachesDirectory, in: .userDomainMask)[0].appending(path: "run.fc.byre.mac/runs", directoryHint: .isDirectory) @MainActor private var watching: Set = [] @MainActor func runOnThisMac(_ api: API, _ item: InboxItem) async throws { let dir = runsDir.appending(path: UUID().uuidString, directoryHint: .isDirectory) watching.insert(dir.path) do { try await start(api, item, dir) } catch { try? FileManager.default.removeItem(at: dir) watching.remove(dir.path) throw error } poll(api, dir) } @MainActor private func start(_ api: API, _ item: InboxItem, _ dir: URL) async throws { let files = dir.appending(path: "files", directoryHint: .isDirectory) try FileManager.default.createDirectory(at: files, withIntermediateDirectories: true) try item.id.write(to: dir.appending(path: "item"), atomically: true, encoding: .utf8) try api.origin.absoluteString.write(to: dir.appending(path: "server"), atomically: true, encoding: .utf8) try str(item.payload.value, "command")!.write(to: dir.appending(path: "command.sh"), atomically: true, encoding: .utf8) let (bundle, response) = try await api.session.download(from: api.url("/api/inbox/\(item.id)/bundle")) let code = (response as! HTTPURLResponse).statusCode if code != 200 { let text = (try? String(contentsOf: bundle, encoding: .utf8)) ?? "" throw ApiError(status: code, message: "download bundle: \(code) \(text)") } try await untar(bundle, into: files, errors: dir.appending(path: "untar.log")) try? FileManager.default.removeItem(at: bundle) let p = Process() p.executableURL = URL(filePath: "/usr/bin/lockf") p.arguments = ["-k", "lock", "/bin/zsh", "-c", "cd files && /bin/zsh -l ../command.sh >../output.log 2>&1; echo $? >../exit.tmp; mv ../exit.tmp ../exit"] p.currentDirectoryURL = dir p.standardInput = FileHandle.nullDevice p.standardOutput = FileHandle.nullDevice p.standardError = FileHandle.nullDevice try p.run() let lock = dir.appending(path: "lock") while !FileManager.default.fileExists(atPath: lock.path) { if !p.isRunning { throw ApiError(status: 0, message: "start command: lockf exited \(p.terminationStatus) before taking its lock") } try await Task.sleep(for: .milliseconds(20)) } log.info("client run started: item \(item.id, privacy: .public) in \(dir.path, privacy: .public)") } private func untar(_ bundle: URL, into dir: URL, errors: URL) async throws { FileManager.default.createFile(atPath: errors.path, contents: nil) let errFile = try FileHandle(forWritingTo: errors) defer { try? errFile.close() } let p = Process() p.executableURL = URL(filePath: "/usr/bin/tar") p.arguments = ["-xzf", bundle.path, "-C", dir.path] p.standardError = errFile p.standardOutput = FileHandle.nullDevice try await withCheckedThrowingContinuation { (c: CheckedContinuation) in p.terminationHandler = { _ in c.resume() } do { try p.run() } catch { c.resume(throwing: error) } } if p.terminationStatus != 0 { let text = tail(errors) throw ApiError(status: 0, message: "unpack bundle: tar exited \(p.terminationStatus): \(text)") } } @MainActor func resumeClientRuns(_ api: API) { let dirs = (try? FileManager.default.contentsOfDirectory(at: runsDir, includingPropertiesForKeys: nil)) ?? [] for dir in dirs where !watching.contains(dir.path) && (try? String(contentsOf: dir.appending(path: "server"), encoding: .utf8)) == api.origin.absoluteString { if !FileManager.default.fileExists(atPath: dir.appending(path: "lock").path) { log.info("client run never started; removing \(dir.path, privacy: .public)") try? FileManager.default.removeItem(at: dir) continue } watching.insert(dir.path) poll(api, dir) } } @MainActor private func poll(_ api: API, _ dir: URL) { Task { let exitFile = dir.appending(path: "exit").path var interrupted = false while !FileManager.default.fileExists(atPath: exitFile) { if lockFree(dir.appending(path: "lock").path) { interrupted = !FileManager.default.fileExists(atPath: exitFile) break } try? await Task.sleep(for: .seconds(1)) } await report(api, dir, interrupted: interrupted) watching.remove(dir.path) } } private func lockFree(_ path: String) -> Bool { guard let h = FileHandle(forReadingAtPath: path) else { return false } defer { try? h.close() } let fd = h.fileDescriptor if flock(fd, LOCK_EX | LOCK_NB) != 0 { return false } flock(fd, LOCK_UN) return true } @MainActor private func report(_ api: API, _ dir: URL, interrupted: Bool) async { do { let itemId = trimmed(try String(contentsOf: dir.appending(path: "item"), encoding: .utf8)) var exit = -1 var output = tail(dir.appending(path: "output.log")) if interrupted { output += "\n[byre: the command stopped without finishing, for example because the Mac restarted]" } else { exit = Int(trimmed(try String(contentsOf: dir.appending(path: "exit"), encoding: .utf8))) ?? -1 } let client = Host.current().localizedName ?? ProcessInfo.processInfo.hostName _ = try await api.client.reportRun(path: .init(itemId: itemId), body: .json(.init(runId: dir.lastPathComponent, client: client, exitCode: exit, output: output))).ok log.info("client run reported: item \(itemId, privacy: .public) exit \(exit)") } catch { log.error("report client run \(dir.path, privacy: .public): \(errorText(error), privacy: .public)") guard let e = rootError(error) as? ApiError, (400..<500).contains(e.status) else { return } } try? FileManager.default.removeItem(at: dir) } private func tail(_ file: URL) -> String { guard let h = try? FileHandle(forReadingFrom: file) else { return "" } defer { try? h.close() } let size = (try? h.seekToEnd()) ?? 0 try? h.seek(toOffset: size > UInt64(maxOutput) ? size - UInt64(maxOutput) : 0) return String(decoding: (try? h.readToEnd()) ?? Data(), as: UTF8.self) }