diff --git a/Services/BackupEngine.swift b/Services/BackupEngine.swift index 72fcda7..939c91d 100644 --- a/Services/BackupEngine.swift +++ b/Services/BackupEngine.swift @@ -78,17 +78,24 @@ final class BackupEngine: ObservableObject { try Task.checkCancellation() - // ── 2. Connect to NAS ─────────────────────────────────────────────── - // Network I/O suspends the main actor — UI stays live the whole time. - let transfer = transferFactory(connection.nasProtocol) + // ── 2. Connect to NAS — build connection pool ────────────────────── + // Each parallel slot gets its own independent SMB/SFTP session. + // SMB allows multiple sessions per host; SFTP channels multiplex over SSH. + let concurrency = min(max(1, store.maxConcurrentUploads), 6) + var transfers: [any NASTransferProtocol] = [] + transfers.reserveCapacity(concurrency) let spConnect = signposter.beginInterval("NASConnect") - try await transfer.connect( - to: host, port: connection.port, - username: connection.username, password: connection.password - ) - signposter.endInterval("NASConnect", spConnect) - activeTransfer = transfer - defer { transfer.disconnect(); activeTransfer = nil } + for _ in 0.. String? in guard !item.isDirectory, !item.name.hasPrefix("."), item.name != BackupManifest.remoteFilename else { return nil } @@ -204,198 +211,123 @@ final class BackupEngine: ObservableObject { var skipped = 0 var failed = 0 var totalBytes: Int64 = 0 - var speedTracker = SpeedTracker() - var throttle = ProgressThrottle(hz: 8) var manifestEntries: [ManifestEntry] = [] manifestEntries.reserveCapacity(pendingAssets.count) - var lastCheckpointAt = 0 // uploaded count at last manifest checkpoint write + var lastCheckpointAt = 0 let checkpointInterval = 25 + let maxRetries = 3 let spUpload = signposter.beginInterval("UploadLoop", "\(pendingAssets.count) files") - let maxRetries = 3 + // Capture values needed inside non-isolated child tasks. + let remotePath = connection.remotePath + let sp = signposter - for (assetIdx, asset) in pendingAssets.enumerated() { - if isCancelled { break } - while case .paused = job.status { - try await Task.sleep(nanoseconds: 500_000_000) - } + var nextIdx = 0 - let remotePath = "\(connection.remotePath)/\(asset.filename)" + await withTaskGroup(of: SlotResult.self) { group in - if assetIdx < queueItems.count { - queueItems[assetIdx].status = .uploading - } - - if (try? await transfer.fileExists(at: remotePath)) == true { - job.fileCompleted(skipped: true) - skipped += 1 - if assetIdx < queueItems.count { - queueItems[assetIdx].status = .skipped - queueItems[assetIdx].completedAt = Date() - } - // Record in manifest even though we didn't upload — the file is on NAS. - // Without this, skipped files stay "pending" forever and trigger backup loops. - manifestEntries.append(ManifestEntry( - localIdentifier: asset.localIdentifier, - filename: asset.filename, - creationDate: asset.creationDate, - fileSize: 0, - remotePath: remotePath, - uploadedAt: Date() - )) - logger.debug("Skipped (NAS exists): \(asset.filename, privacy: .public)") - continue - } - - var uploadSucceeded = false - var lastUploadError: Error? - - for attempt in 0.. 0 { - // Exponential backoff: 2s, 4s before retries 2 and 3 - let backoffNs = UInt64(pow(2.0, Double(attempt)) * 1_000_000_000) - try? await Task.sleep(nanoseconds: backoffNs) - if assetIdx < queueItems.count { - queueItems[assetIdx].retryCount = attempt - queueItems[assetIdx].status = .uploading - } - logger.info("Retry \(attempt)/\(maxRetries - 1) for \(asset.filename, privacy: .public)") - } - - do { - let spFile = signposter.beginInterval("UploadFile", - "\(asset.filename, privacy: .public)") - let localURL = try await photoService.exportAsset(asset) - defer { try? FileManager.default.removeItem(at: localURL) } - - let fileSize = (try? localURL.resourceValues(forKeys: [.fileSizeKey]).fileSize) - .flatMap { Int64($0) } ?? 0 - let bytesSoFar = totalBytes - let capturedIdx = assetIdx - - if capturedIdx < queueItems.count { - queueItems[capturedIdx].totalBytes = fileSize - queueItems[capturedIdx].bytesUploaded = 0 - } - - try await transfer.upload(localURL: localURL, remotePath: remotePath) { sent, _ in - let speed = speedTracker.update(bytesSent: sent) - guard throttle.shouldUpdate() else { return } - Task { @MainActor [weak self] in - guard let self else { return } - self.job.updateProgress( - fileName: asset.filename, - fileSize: fileSize, - bytesTransferred: bytesSoFar + sent, - speed: speed - ) - if capturedIdx < self.queueItems.count { - self.queueItems[capturedIdx].bytesUploaded = sent - self.queueItems[capturedIdx].speedBytesPerSec = speed + // Seed one task per slot, up to the number of assets. + for slot in 0..= checkpointInterval { - lastCheckpointAt = uploaded - var checkpoint = baseManifest - checkpoint.merge(entries: manifestEntries) - if let encoded = try? JSONEncoder.kisani.encode(checkpoint) { - try? await transfer.writeData(encoded, to: manifestPath) - await NASManifestCache.shared.update(checkpoint) - logger.info("Manifest checkpoint: \(checkpoint.entries.count) total entries") - } - } - } catch { - // "Object Name Collision" (NTStatus 0xC0000035) means the file is already - // on the NAS. Retrying won't help — treat it as safe immediately. - if isAlreadyExistsError(error) { - job.fileCompleted(skipped: true) - skipped += 1 - if assetIdx < queueItems.count { - queueItems[assetIdx].status = .skipped - queueItems[assetIdx].completedAt = Date() - } - manifestEntries.append(ManifestEntry( - localIdentifier: asset.localIdentifier, - filename: asset.filename, - creationDate: asset.creationDate, - fileSize: 0, - remotePath: remotePath, - uploadedAt: Date() - )) - logger.info("Collision — file already on NAS, marking safe: \(asset.filename, privacy: .public)") - uploadSucceeded = true - break - } - lastUploadError = error - logger.warning("Upload attempt \(attempt + 1)/\(maxRetries) failed for \(asset.filename, privacy: .public): \(underlyingDescription(error), privacy: .public)") + ) } - - if uploadSucceeded { break } } - if !uploadSucceeded, let error = lastUploadError { - // Last-resort check: maybe the file landed on NAS despite the errors - // (e.g. fileExists threw transiently on the pre-check and the upload - // actually succeeded before the connection reset). - if (try? await transfer.fileExists(at: remotePath)) == true { - job.fileCompleted(skipped: true) + // Collect results and immediately dispatch the next asset onto the freed slot. + for await result in group { + + // Drain any pause before updating state or queueing more work. + while case .paused = job.status { + try? await Task.sleep(nanoseconds: 300_000_000) + } + + // Update queue item display state. + let qi = result.assetIdx + switch result.outcome { + case .uploaded: + totalBytes += result.fileSize + uploaded += 1 + job.fileCompleted(skipped: false) + if qi < queueItems.count { + queueItems[qi].status = .uploaded + queueItems[qi].bytesUploaded = result.fileSize + queueItems[qi].completedAt = Date() + } + case .skipped: skipped += 1 - if assetIdx < queueItems.count { - queueItems[assetIdx].status = .skipped - queueItems[assetIdx].completedAt = Date() + job.fileCompleted(skipped: true) + if qi < queueItems.count { + queueItems[qi].status = .skipped + queueItems[qi].completedAt = Date() } - manifestEntries.append(ManifestEntry( - localIdentifier: asset.localIdentifier, - filename: asset.filename, - creationDate: asset.creationDate, - fileSize: 0, - remotePath: remotePath, - uploadedAt: Date() - )) - logger.info("Post-retry check — file on NAS, marking safe: \(asset.filename, privacy: .public)") - } else { - let reason = underlyingDescription(error) - logger.error("Upload permanently failed for \(asset.filename, privacy: .public): \(reason, privacy: .public)") - job.fileFailed() + case .failed(let domain, let message): failed += 1 - if assetIdx < queueItems.count { - queueItems[assetIdx].status = .failed - queueItems[assetIdx].completedAt = Date() - queueItems[assetIdx].error = BackupQueueItem.UploadError( - code: underlyingDomain(error), - message: reason, - timestamp: Date() - ) + job.fileFailed() + if qi < queueItems.count { + queueItems[qi].status = .failed + queueItems[qi].completedAt = Date() + queueItems[qi].error = BackupQueueItem.UploadError( + code: domain, message: message, timestamp: Date()) } } + + if let entry = result.entry { manifestEntries.append(entry) } + + // Manifest checkpoint every N completed uploads. + if uploaded - lastCheckpointAt >= checkpointInterval { + lastCheckpointAt = uploaded + var ckpt = baseManifest + ckpt.merge(entries: manifestEntries) + if let encoded = try? JSONEncoder.kisani.encode(ckpt) { + try? await transfers[0].writeData(encoded, to: manifestPath) + await NASManifestCache.shared.update(ckpt) + logger.info("Manifest checkpoint: \(ckpt.entries.count) entries") + } + } + + // Queue the next asset onto the now-free slot. + guard nextIdx < pendingAssets.count, !isCancelled else { continue } + let idx = nextIdx; nextIdx += 1 + let asset = pendingAssets[idx] + let slot = result.slotIdx + let t = transfers[slot] + group.addTask { [weak self] in + await BackupEngine.uploadSlot( + asset: asset, assetIdx: idx, slotIdx: slot, + transfer: t, remotePath: remotePath, + photoSvc: photoSvc, maxRetries: maxRetries, + signposter: sp, + progressHandler: { fileName, fileSize, bytesSoFar, sent in + Task { @MainActor [weak self] in + self?.job.updateProgress( + fileName: fileName, fileSize: fileSize, + bytesTransferred: bytesSoFar + sent, speed: 0) + if let qi = self?.queueItems, idx < qi.count { + self?.queueItems[idx].bytesUploaded = sent + } + } + } + ) + } } } @@ -473,6 +405,126 @@ final class BackupEngine: ObservableObject { } +// MARK: — Per-slot upload worker + +extension BackupEngine { + /// Runs entirely off the main actor. Returns a `SlotResult` that the + /// TaskGroup aggregation loop processes on the main actor. + nonisolated static func uploadSlot( + asset: PhotoAsset, + assetIdx: Int, + slotIdx: Int, + transfer: any NASTransferProtocol, + remotePath: String, + photoSvc: any PhotoLibraryProtocol, + maxRetries: Int, + signposter: OSSignposter, + progressHandler: @escaping @Sendable (String, Int64, Int64, Int64) -> Void + ) async -> SlotResult { + let filePath = "\(remotePath)/\(asset.filename)" + + // Fast-path: file already on NAS. + if (try? await transfer.fileExists(at: filePath)) == true { + logger.debug("Skipped (NAS exists): \(asset.filename, privacy: .public)") + return SlotResult( + assetIdx: assetIdx, slotIdx: slotIdx, fileSize: 0, + entry: ManifestEntry(localIdentifier: asset.localIdentifier, + filename: asset.filename, + creationDate: asset.creationDate, + fileSize: 0, remotePath: filePath, + uploadedAt: Date()), + outcome: .skipped) + } + + var lastError: Error? + var speedTracker = SpeedTracker() + var throttle = ProgressThrottle(hz: 8) + + for attempt in 0.. 0 { + let backoffNs = UInt64(pow(2.0, Double(attempt)) * 1_000_000_000) + try? await Task.sleep(nanoseconds: backoffNs) + logger.info("Retry \(attempt)/\(maxRetries - 1) for \(asset.filename, privacy: .public)") + } + + do { + let sp = signposter.beginInterval("UploadFile", "\(asset.filename, privacy: .public)") + let localURL = try await photoSvc.exportAsset(asset) + defer { try? FileManager.default.removeItem(at: localURL) } + + let fileSize = (try? localURL.resourceValues(forKeys: [.fileSizeKey]).fileSize) + .flatMap { Int64($0) } ?? 0 + + try await transfer.upload(localURL: localURL, remotePath: filePath) { sent, _ in + let speed = speedTracker.update(bytesSent: sent) + guard throttle.shouldUpdate() else { return } + _ = speed + progressHandler(asset.filename, fileSize, 0, sent) + } + + signposter.endInterval("UploadFile", sp, "\(fileSize) bytes") + + return SlotResult( + assetIdx: assetIdx, slotIdx: slotIdx, fileSize: fileSize, + entry: ManifestEntry(localIdentifier: asset.localIdentifier, + filename: asset.filename, + creationDate: asset.creationDate, + fileSize: fileSize, remotePath: filePath, + uploadedAt: Date()), + outcome: .uploaded) + + } catch { + if isAlreadyExistsError(error) { + logger.info("Collision — already on NAS: \(asset.filename, privacy: .public)") + return SlotResult( + assetIdx: assetIdx, slotIdx: slotIdx, fileSize: 0, + entry: ManifestEntry(localIdentifier: asset.localIdentifier, + filename: asset.filename, + creationDate: asset.creationDate, + fileSize: 0, remotePath: filePath, + uploadedAt: Date()), + outcome: .skipped) + } + lastError = error + logger.warning("Attempt \(attempt + 1)/\(maxRetries) failed for \(asset.filename, privacy: .public): \(underlyingDescription(error), privacy: .public)") + } + } + + // Last-resort existence check after all retries exhausted. + if (try? await transfer.fileExists(at: filePath)) == true { + return SlotResult( + assetIdx: assetIdx, slotIdx: slotIdx, fileSize: 0, + entry: ManifestEntry(localIdentifier: asset.localIdentifier, + filename: asset.filename, + creationDate: asset.creationDate, + fileSize: 0, remotePath: filePath, + uploadedAt: Date()), + outcome: .skipped) + } + + let reason = lastError.map { underlyingDescription($0) } ?? "unknown" + logger.error("Permanently failed: \(asset.filename, privacy: .public): \(reason, privacy: .public)") + return SlotResult( + assetIdx: assetIdx, slotIdx: slotIdx, fileSize: 0, + entry: nil, + outcome: .failed(lastError.map { underlyingDomain($0) } ?? "", reason)) + } + + // SlotResult must be visible to the TaskGroup closure inside run(). + struct SlotResult: Sendable { + let assetIdx: Int + let slotIdx: Int + let fileSize: Int64 + let entry: ManifestEntry? + enum Outcome: Sendable { + case uploaded + case skipped + case failed(String, String) + } + let outcome: Outcome + } +} + // MARK: — Error classification helpers /// True when the error indicates the remote file already exists. diff --git a/Shared/Persistence/ConnectionStore.swift b/Shared/Persistence/ConnectionStore.swift index 57925fd..d1c81a3 100644 --- a/Shared/Persistence/ConnectionStore.swift +++ b/Shared/Persistence/ConnectionStore.swift @@ -74,6 +74,11 @@ final class ConnectionStore: ObservableObject { didSet { UserDefaults.standard.set(notifyOnErrors, forKey: "notifyOnErrors") } } + // Parallel upload concurrency (1 = sequential, 3 = default, max 6) + @Published var maxConcurrentUploads: Int { + didSet { UserDefaults.standard.set(maxConcurrentUploads, forKey: "maxConcurrentUploads") } + } + // Auto-backup via LAN trigger (trusted Wi-Fi join) @Published var autoBackupEnabled: Bool { didSet { UserDefaults.standard.set(autoBackupEnabled, forKey: "autoBackupEnabled") } @@ -113,6 +118,7 @@ final class ConnectionStore: ObservableObject { notifyOnStart = UserDefaults.standard.object(forKey: "notifyOnStart") as? Bool ?? false notifyOnComplete = UserDefaults.standard.object(forKey: "notifyOnComplete") as? Bool ?? true notifyOnErrors = UserDefaults.standard.object(forKey: "notifyOnErrors") as? Bool ?? true + maxConcurrentUploads = UserDefaults.standard.object(forKey: "maxConcurrentUploads") as? Int ?? 3 autoBackupEnabled = UserDefaults.standard.object(forKey: "autoBackupEnabled") as? Bool ?? true autoBackupOnOpen = UserDefaults.standard.object(forKey: "autoBackupOnOpen") as? Bool ?? false allowCellularBackup = UserDefaults.standard.object(forKey: "allowCellularBackup") as? Bool ?? false