new updates for paddel
This commit is contained in:
Mark Gregory
@@ -0,0 +1,260 @@
|
||||
import Foundation
|
||||
|
||||
|
||||
@MainActor
|
||||
final class RsyncRunner: ObservableObject {
|
||||
@Published private(set) var isRunning = false
|
||||
@Published private(set) var currentJobID: UUID?
|
||||
|
||||
private var process: Process?
|
||||
|
||||
struct Result {
|
||||
let exitCode: Int32
|
||||
let stdout: String
|
||||
let stderr: String
|
||||
let wasCancelled: Bool
|
||||
}
|
||||
|
||||
func run(job: SyncJob, dryRun: Bool, trigger: SyncRunTrigger = .manual) async throws -> Result {
|
||||
guard !isRunning else { throw RunnerError.alreadyRunning }
|
||||
if let message = RsyncSecurityValidator.validationMessage(for: job, trigger: trigger) {
|
||||
throw RunnerError.securityValidation(message)
|
||||
}
|
||||
|
||||
let executable = RsyncLocator.locate()
|
||||
guard RsyncLocator.isTrustedExecutableCandidate(executable) else {
|
||||
throw RunnerError.invalidExecutable(executable)
|
||||
}
|
||||
let arguments = RsyncCommandBuilder.build(job: job, dryRun: dryRun, executable: executable)
|
||||
|
||||
isRunning = true
|
||||
currentJobID = job.id
|
||||
let activityToken = ProcessInfo.processInfo.beginActivity(
|
||||
options: [.userInitiated, .idleSystemSleepDisabled],
|
||||
reason: "Rsync Studio is synchronising \(job.name)"
|
||||
)
|
||||
|
||||
defer {
|
||||
ProcessInfo.processInfo.endActivity(activityToken)
|
||||
isRunning = false
|
||||
currentJobID = nil
|
||||
process = nil
|
||||
}
|
||||
|
||||
let process = Process()
|
||||
self.process = process
|
||||
process.executableURL = URL(fileURLWithPath: executable)
|
||||
process.arguments = arguments
|
||||
|
||||
let stdoutPipe = Pipe()
|
||||
let stderrPipe = Pipe()
|
||||
process.standardOutput = stdoutPipe
|
||||
process.standardError = stderrPipe
|
||||
|
||||
// Keep rsync's potentially very chatty output off the MainActor. Updating a
|
||||
// SwiftUI String for every rsync progress chunk makes the whole window jerk.
|
||||
// These locked buffers collect every byte for history/export without causing
|
||||
// any UI invalidation while the transfer is running.
|
||||
let stdoutBuffer = LockedDataBuffer()
|
||||
let stderrBuffer = LockedDataBuffer()
|
||||
|
||||
stdoutPipe.fileHandleForReading.readabilityHandler = { handle in
|
||||
let data = handle.availableData
|
||||
guard !data.isEmpty else { return }
|
||||
stdoutBuffer.append(data)
|
||||
}
|
||||
|
||||
stderrPipe.fileHandleForReading.readabilityHandler = { handle in
|
||||
let data = handle.availableData
|
||||
guard !data.isEmpty else { return }
|
||||
stderrBuffer.append(data)
|
||||
}
|
||||
|
||||
return try await withCheckedThrowingContinuation { continuation in
|
||||
process.terminationHandler = { proc in
|
||||
stdoutPipe.fileHandleForReading.readabilityHandler = nil
|
||||
stderrPipe.fileHandleForReading.readabilityHandler = nil
|
||||
|
||||
// Consume anything left between the final readability callback and EOF.
|
||||
let remainingOut = stdoutPipe.fileHandleForReading.readDataToEndOfFile()
|
||||
let remainingErr = stderrPipe.fileHandleForReading.readDataToEndOfFile()
|
||||
if !remainingOut.isEmpty { stdoutBuffer.append(remainingOut) }
|
||||
if !remainingErr.isEmpty { stderrBuffer.append(remainingErr) }
|
||||
|
||||
continuation.resume(returning: Result(
|
||||
exitCode: proc.terminationStatus,
|
||||
stdout: stdoutBuffer.stringValue,
|
||||
stderr: stderrBuffer.stringValue,
|
||||
wasCancelled: proc.terminationReason == .uncaughtSignal
|
||||
))
|
||||
}
|
||||
|
||||
do {
|
||||
try process.run()
|
||||
} catch {
|
||||
stdoutPipe.fileHandleForReading.readabilityHandler = nil
|
||||
stderrPipe.fileHandleForReading.readabilityHandler = nil
|
||||
continuation.resume(throwing: error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private func remoteDestinationLooksLikeHost(_ job: SyncJob) -> Bool {
|
||||
let destination = job.destinationPath.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
let host = job.remoteHost.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
return destination == host || destination == "\(job.remoteUser)@\(host)"
|
||||
}
|
||||
|
||||
func preflight(job: SyncJob) async -> JobPreflightReport {
|
||||
var checks: [JobPreflightCheck] = []
|
||||
|
||||
if let message = RsyncSecurityValidator.validationMessage(for: job, trigger: .manual) {
|
||||
checks.append(.init(title: "Configuration", detail: message, severity: .error))
|
||||
return JobPreflightReport(checks: checks)
|
||||
}
|
||||
checks.append(.init(title: "Configuration", detail: "Security and argument validation passed.", severity: .success))
|
||||
|
||||
let executable = RsyncLocator.locate()
|
||||
if RsyncLocator.isTrustedExecutableCandidate(executable) {
|
||||
checks.append(.init(title: "Rsync", detail: executable, severity: .success))
|
||||
} else {
|
||||
checks.append(.init(title: "Rsync", detail: "No trusted rsync executable is available.", severity: .error))
|
||||
}
|
||||
|
||||
if job.isRemote {
|
||||
let ssh = "/usr/bin/ssh"
|
||||
guard FileManager.default.isExecutableFile(atPath: ssh) else {
|
||||
checks.append(.init(title: "SSH", detail: "The system SSH client is unavailable.", severity: .error))
|
||||
return JobPreflightReport(checks: checks)
|
||||
}
|
||||
let userHost = job.remoteUser.isEmpty ? job.remoteHost : "\(job.remoteUser)@\(job.remoteHost)"
|
||||
var args = ["-o", "BatchMode=yes", "-o", "ConnectTimeout=8", "-o", "ServerAliveInterval=5", "-o", "ServerAliveCountMax=1", "-p", String(job.remotePort)]
|
||||
let key = RsyncCommandBuilder.normalizedPrivateKeyPath(job.sshKeyPath)
|
||||
if !key.isEmpty { args += ["-o", "IdentitiesOnly=yes", "-i", key] }
|
||||
let remotePath = RsyncCommandBuilder.shellEscape(job.destinationPath)
|
||||
args += [userHost, "command -v rsync >/dev/null 2>&1 && test -d \(remotePath)"]
|
||||
let result = await Self.runProbe(executable: ssh, arguments: args)
|
||||
if result == 0 {
|
||||
checks.append(.init(title: "Remote connection", detail: "SSH authentication, remote rsync and destination directory are available.", severity: .success))
|
||||
} else {
|
||||
checks.append(.init(title: "Remote connection", detail: "Could not authenticate, find remote rsync, or access the destination directory.", severity: .error))
|
||||
}
|
||||
} else {
|
||||
var isDir: ObjCBool = false
|
||||
if FileManager.default.fileExists(atPath: (job.sourcePath as NSString).expandingTildeInPath, isDirectory: &isDir) {
|
||||
checks.append(.init(title: "Source", detail: "Source exists and is reachable.", severity: .success))
|
||||
} else {
|
||||
checks.append(.init(title: "Source", detail: "Source does not exist or is not reachable.", severity: .error))
|
||||
}
|
||||
|
||||
let destination = (job.destinationPath as NSString).expandingTildeInPath
|
||||
let volume = SyncVolumeInspector.info(forPath: destination)
|
||||
if let volume {
|
||||
checks.append(.init(title: "Destination volume", detail: "\(volume.name) · \(volume.freeSpaceDescription) free", severity: volume.freeBytes >= Int64(job.minimumFreeSpaceMB) * 1_048_576 ? .success : .error))
|
||||
if job.requireExpectedVolume && !job.expectedVolumeIdentifier.isEmpty && volume.identifier != job.expectedVolumeIdentifier {
|
||||
checks.append(.init(title: "Volume identity", detail: "Expected \(job.expectedVolumeIdentifier), found \(volume.identifier).", severity: .error))
|
||||
} else if job.requireExpectedVolume {
|
||||
checks.append(.init(title: "Volume identity", detail: "Expected destination volume is mounted.", severity: .success))
|
||||
}
|
||||
} else {
|
||||
checks.append(.init(title: "Destination volume", detail: "Could not identify the destination volume.", severity: .error))
|
||||
}
|
||||
}
|
||||
return JobPreflightReport(checks: checks)
|
||||
}
|
||||
|
||||
private static func runProbe(executable: String, arguments: [String]) async -> Int32 {
|
||||
await withCheckedContinuation { continuation in
|
||||
let process = Process()
|
||||
process.executableURL = URL(fileURLWithPath: executable)
|
||||
process.arguments = arguments
|
||||
process.standardOutput = FileHandle.nullDevice
|
||||
process.standardError = FileHandle.nullDevice
|
||||
process.terminationHandler = { continuation.resume(returning: $0.terminationStatus) }
|
||||
do { try process.run() } catch { continuation.resume(returning: -1) }
|
||||
}
|
||||
}
|
||||
|
||||
func cancel() {
|
||||
guard let process, process.isRunning else { return }
|
||||
process.interrupt()
|
||||
}
|
||||
|
||||
enum RunnerError: LocalizedError {
|
||||
case alreadyRunning
|
||||
case securityValidation(String)
|
||||
case invalidExecutable(String)
|
||||
|
||||
var errorDescription: String? {
|
||||
switch self {
|
||||
case .alreadyRunning:
|
||||
return "Another rsync job is already running."
|
||||
case .securityValidation(let message):
|
||||
return message
|
||||
case .invalidExecutable(let path):
|
||||
return "The rsync executable at \(path) is unavailable or is not a trusted regular executable file."
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private final class LockedDataBuffer: @unchecked Sendable {
|
||||
private let lock = NSLock()
|
||||
private var data = Data()
|
||||
|
||||
func append(_ newData: Data) {
|
||||
lock.lock()
|
||||
data.append(newData)
|
||||
lock.unlock()
|
||||
}
|
||||
|
||||
var stringValue: String {
|
||||
lock.lock()
|
||||
let snapshot = data
|
||||
lock.unlock()
|
||||
return String(decoding: snapshot, as: UTF8.self)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
enum JobPreflightSeverity: Int, Comparable {
|
||||
case success = 0, warning = 1, error = 2
|
||||
static func < (lhs: Self, rhs: Self) -> Bool { lhs.rawValue < rhs.rawValue }
|
||||
}
|
||||
|
||||
struct JobPreflightCheck: Identifiable {
|
||||
let id = UUID()
|
||||
let title: String
|
||||
let detail: String
|
||||
let severity: JobPreflightSeverity
|
||||
}
|
||||
|
||||
struct JobPreflightReport {
|
||||
let checks: [JobPreflightCheck]
|
||||
var severity: JobPreflightSeverity { checks.map(\.severity).max() ?? .success }
|
||||
var passed: Bool { severity != .error }
|
||||
var title: String { passed ? (severity == .warning ? "Ready with warnings" : "Ready") : "Needs attention" }
|
||||
}
|
||||
|
||||
struct SyncVolumeInfo {
|
||||
let identifier: String
|
||||
let name: String
|
||||
let freeBytes: Int64
|
||||
var freeSpaceDescription: String { ByteCountFormatter.string(fromByteCount: freeBytes, countStyle: .file) }
|
||||
}
|
||||
|
||||
enum SyncVolumeInspector {
|
||||
static func info(forPath path: String) -> SyncVolumeInfo? {
|
||||
var url = URL(fileURLWithPath: path)
|
||||
while !FileManager.default.fileExists(atPath: url.path), url.path != "/" { url.deleteLastPathComponent() }
|
||||
do {
|
||||
let values = try url.resourceValues(forKeys: [.volumeIdentifierKey, .volumeNameKey, .volumeAvailableCapacityForImportantUsageKey])
|
||||
guard let rawID = values.volumeIdentifier else { return nil }
|
||||
return SyncVolumeInfo(
|
||||
identifier: String(describing: rawID),
|
||||
name: values.volumeName ?? url.path,
|
||||
freeBytes: values.volumeAvailableCapacityForImportantUsage ?? 0
|
||||
)
|
||||
} catch { return nil }
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user