8fb425344d
Replace the sequential for-loop with withTaskGroup running N concurrent uploads. Each slot gets its own SMB/SFTP connection so uploads never block each other. Default concurrency is 3 (configurable via ConnectionStore.maxConcurrentUploads, capped at 6). - ConnectionStore: add maxConcurrentUploads (default 3, persisted) - BackupEngine: build pool of N connections at start, seed TaskGroup with one task per slot, recycle slots as tasks complete - Manifest checkpoints, pause, cancel, retry logic all preserved - SlotResult extracted to BackupEngine extension for Sendable conformance - Fix: remove undefined .trackScrollForTabBar() from BackupView Co-Authored-By: Kutesir <kutesir@provoc.ug> Co-Authored-By: Sentry <sentry@provoc.ug>
594 lines
27 KiB
Swift
594 lines
27 KiB
Swift
import Foundation
|
||
import Photos
|
||
import os.log
|
||
|
||
private let logger = Logger(subsystem: "com.albert.nasbackup", category: "BackupEngine")
|
||
private let signposter = OSSignposter(subsystem: "com.albert.nasbackup", category: "BackupEngine")
|
||
|
||
@MainActor
|
||
final class BackupEngine: ObservableObject {
|
||
static let shared = BackupEngine()
|
||
|
||
@Published private(set) var job: BackupJob = BackupJob()
|
||
@Published private(set) var queueItems: [BackupQueueItem] = []
|
||
|
||
private let photoService: PhotoLibraryProtocol
|
||
private let transferFactory: (NASProtocol) -> any NASTransferProtocol
|
||
private var activeTransfer: (any NASTransferProtocol)?
|
||
private var isCancelled = false
|
||
|
||
private static let queueCacheURL: URL = {
|
||
let caches = FileManager.default.urls(for: .cachesDirectory, in: .userDomainMask)[0]
|
||
return caches.appendingPathComponent("kisani_last_queue.json")
|
||
}()
|
||
|
||
private init(
|
||
photos: PhotoLibraryProtocol = PhotoLibraryService(),
|
||
transferFactory: @escaping (NASProtocol) -> any NASTransferProtocol = { proto in
|
||
switch proto {
|
||
case .smb: return SMBService()
|
||
case .sftp: return SFTPService()
|
||
}
|
||
}
|
||
) {
|
||
self.photoService = photos
|
||
self.transferFactory = transferFactory
|
||
// Load last session's queue so SyncView has data to show on next launch.
|
||
if let data = try? Data(contentsOf: Self.queueCacheURL),
|
||
let items = try? JSONDecoder.kisani.decode([BackupQueueItem].self, from: data) {
|
||
self.queueItems = items
|
||
}
|
||
}
|
||
|
||
static func testInstance(
|
||
transfer: any NASTransferProtocol,
|
||
photos: PhotoLibraryProtocol
|
||
) -> BackupEngine {
|
||
BackupEngine(photos: photos, transferFactory: { _ in transfer })
|
||
}
|
||
|
||
func run(
|
||
connection: NASConnection,
|
||
filter: BackupFilter,
|
||
triggeredByLAN: Bool = false
|
||
) async throws -> BackupResult {
|
||
isCancelled = false
|
||
let startDate = Date()
|
||
let store = ConnectionStore.shared
|
||
let lan = LANMonitor.shared
|
||
|
||
if !triggeredByLAN && lan.isOnCellular && !store.allowCellularBackup {
|
||
throw BackupError.networkUnavailable
|
||
}
|
||
|
||
let host = resolveHost(connection: connection, store: store, lan: lan)
|
||
|
||
// ── 1. Fetch assets off the main actor ─────────────────────────────
|
||
// fetchAssets() enumerates every PHAsset and builds a [PhotoAsset] array.
|
||
// For a library of 10 k+ assets this is 100–500 ms of synchronous work.
|
||
// Running it in a detached task keeps the main thread free for animations.
|
||
job.prepare()
|
||
let photoSvc = self.photoService
|
||
let spFetch = signposter.beginInterval("FetchAssets")
|
||
let assets: [PhotoAsset] = await Task.detached(priority: .userInitiated) {
|
||
photoSvc.fetchAssets(filter: filter)
|
||
}.value
|
||
signposter.endInterval("FetchAssets", spFetch, "\(assets.count) assets")
|
||
logger.info("Photo fetch: \(assets.count) assets in library")
|
||
|
||
try Task.checkCancellation()
|
||
|
||
// ── 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")
|
||
for _ in 0..<concurrency {
|
||
let t = transferFactory(connection.nasProtocol)
|
||
try await t.connect(
|
||
to: host, port: connection.port,
|
||
username: connection.username, password: connection.password
|
||
)
|
||
transfers.append(t)
|
||
}
|
||
signposter.endInterval("NASConnect", spConnect, "\(concurrency) connections")
|
||
activeTransfer = transfers.first
|
||
defer { transfers.forEach { $0.disconnect() }; activeTransfer = nil }
|
||
|
||
try Task.checkCancellation()
|
||
|
||
// ── 3. Load manifest ───────────────────────────────────────────────
|
||
let manifestPath = "\(connection.remotePath)/\(BackupManifest.remoteFilename)"
|
||
let spManifest = signposter.beginInterval("ManifestLoad")
|
||
|
||
let manifestData = try? await transfers[0].downloadData(at: manifestPath)
|
||
|
||
// Decode NAS manifest off the main actor.
|
||
let nasManifest: BackupManifest? = await Task.detached(priority: .userInitiated) {
|
||
guard let data = manifestData else { return nil }
|
||
return try? JSONDecoder.kisani.decode(BackupManifest.self, from: data)
|
||
}.value
|
||
|
||
// Use the manifest with more entries as the authoritative base.
|
||
// CRITICAL: if the NAS manifest has fewer entries than the local cache (e.g. 1 vs 264),
|
||
// the cache preserves accumulated history and must win. Overwriting the cache with a
|
||
// smaller NAS manifest destroys history and causes every asset to re-upload.
|
||
let cachedManifest = await NASManifestCache.shared.manifest
|
||
let cacheCount = cachedManifest?.entries.count ?? 0
|
||
let nasCount = nasManifest?.entries.count ?? 0
|
||
|
||
let baseManifest: BackupManifest
|
||
if nasCount > 0, nasCount >= cacheCount {
|
||
baseManifest = nasManifest!
|
||
await NASManifestCache.shared.update(nasManifest!)
|
||
logger.info("Manifest from NAS — \(nasCount) entries")
|
||
} else if cacheCount > 0 {
|
||
baseManifest = cachedManifest!
|
||
logger.info("Manifest from cache — \(cacheCount) entries (NAS had \(nasCount))")
|
||
} else {
|
||
// Genuinely first backup — no manifest anywhere.
|
||
baseManifest = nasManifest ?? BackupManifest()
|
||
logger.info("Manifest: starting fresh — \(baseManifest.entries.count) entries")
|
||
}
|
||
let baseIndex = ManifestIndex(manifest: baseManifest)
|
||
|
||
signposter.endInterval("ManifestLoad", spManifest, "\(baseIndex.totalCount) entries")
|
||
logger.info("Manifest base: \(baseIndex.totalCount) entries, isFilenameOnly=\(baseIndex.isFilenameOnly)")
|
||
|
||
// ── 3b. Build filter index ─────────────────────────────────────────
|
||
// When the NAS directory has more files than Kisani's manifest (e.g. files
|
||
// uploaded by other tools), a manifest-only index will misidentify those files
|
||
// as "not backed up" and re-upload them on every "Back up again" tap.
|
||
// Use the broader NAS directory listing as the filter when the manifest is sparse.
|
||
// Build a hybrid filter index: manifest localIdentifiers (primary) + all NAS
|
||
// filenames (fallback). This ensures files uploaded by other tools are recognised
|
||
// as "already on NAS" by filename and are not re-uploaded on every backup run.
|
||
let filterIndex: ManifestIndex
|
||
let cachedDirCount = await NASManifestCache.shared.directoryCount ?? 0
|
||
if baseIndex.totalCount < cachedDirCount {
|
||
if let cachedFilenames = await NASManifestCache.shared.directoryFilenames,
|
||
cachedFilenames.count > baseIndex.totalCount {
|
||
filterIndex = await Task.detached(priority: .utility) {
|
||
ManifestIndex(manifest: baseManifest, nasFilenames: cachedFilenames, totalCount: cachedDirCount)
|
||
}.value
|
||
logger.info("Filter index: hybrid — \(baseIndex.totalCount) manifest + \(cachedFilenames.count) cached NAS filenames")
|
||
} else {
|
||
let nasItems = (try? await transfers[0].listDirectory(at: connection.remotePath)) ?? []
|
||
let names = nasItems.compactMap { item -> String? in
|
||
guard !item.isDirectory, !item.name.hasPrefix("."),
|
||
item.name != BackupManifest.remoteFilename else { return nil }
|
||
return item.name
|
||
}
|
||
await NASManifestCache.shared.setDirectoryFilenames(names)
|
||
filterIndex = await Task.detached(priority: .utility) {
|
||
ManifestIndex(manifest: baseManifest, nasFilenames: names, totalCount: names.count)
|
||
}.value
|
||
logger.info("Filter index: hybrid — \(baseIndex.totalCount) manifest + \(names.count) fresh NAS filenames")
|
||
}
|
||
} else {
|
||
filterIndex = baseIndex
|
||
}
|
||
|
||
try Task.checkCancellation()
|
||
|
||
// ── 4. Build pending queue off the main actor ──────────────────────
|
||
let spFilter = signposter.beginInterval("BuildPendingQueue")
|
||
let pendingAssets: [PhotoAsset] = await Task.detached(priority: .userInitiated) {
|
||
assets.filter { asset in
|
||
!filterIndex.matches(localIdentifier: asset.localIdentifier) &&
|
||
!filterIndex.matches(filename: asset.filename)
|
||
}
|
||
}.value
|
||
signposter.endInterval("BuildPendingQueue", spFilter, "\(pendingAssets.count) pending")
|
||
logger.info("Pending queue: \(pendingAssets.count) of \(assets.count) total")
|
||
|
||
guard !pendingAssets.isEmpty else {
|
||
job.allAlreadySafe()
|
||
BackupStatusService.shared.refresh(force: false)
|
||
return BackupResult.empty(date: startDate)
|
||
}
|
||
|
||
// ── 5. Upload loop ─────────────────────────────────────────────────
|
||
job.start(totalFiles: pendingAssets.count, totalBytes: 0)
|
||
if store.notifyOnStart {
|
||
NotificationService.notifyBackupStarted(count: pendingAssets.count)
|
||
}
|
||
|
||
// Build initial queue (all queued) for SyncView
|
||
queueItems = pendingAssets.map { asset in
|
||
BackupQueueItem(
|
||
id: asset.localIdentifier,
|
||
filename: asset.filename,
|
||
nasPath: "\(connection.remotePath)/\(asset.filename)",
|
||
status: .queued, bytesUploaded: 0, totalBytes: 0, speedBytesPerSec: 0,
|
||
error: nil, retryCount: 0, queuedAt: Date(), completedAt: nil
|
||
)
|
||
}
|
||
|
||
var uploaded = 0
|
||
var skipped = 0
|
||
var failed = 0
|
||
var totalBytes: Int64 = 0
|
||
var manifestEntries: [ManifestEntry] = []
|
||
manifestEntries.reserveCapacity(pendingAssets.count)
|
||
var lastCheckpointAt = 0
|
||
let checkpointInterval = 25
|
||
let maxRetries = 3
|
||
|
||
let spUpload = signposter.beginInterval("UploadLoop", "\(pendingAssets.count) files")
|
||
|
||
// Capture values needed inside non-isolated child tasks.
|
||
let remotePath = connection.remotePath
|
||
let sp = signposter
|
||
|
||
var nextIdx = 0
|
||
|
||
await withTaskGroup(of: SlotResult.self) { group in
|
||
|
||
// Seed one task per slot, up to the number of assets.
|
||
for slot in 0..<concurrency where nextIdx < pendingAssets.count {
|
||
let idx = nextIdx; nextIdx += 1
|
||
let asset = pendingAssets[idx]
|
||
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
|
||
}
|
||
}
|
||
}
|
||
)
|
||
}
|
||
}
|
||
|
||
// 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
|
||
job.fileCompleted(skipped: true)
|
||
if qi < queueItems.count {
|
||
queueItems[qi].status = .skipped
|
||
queueItems[qi].completedAt = Date()
|
||
}
|
||
case .failed(let domain, let message):
|
||
failed += 1
|
||
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
|
||
}
|
||
}
|
||
}
|
||
)
|
||
}
|
||
}
|
||
}
|
||
|
||
signposter.endInterval("UploadLoop", spUpload, "\(uploaded) uploaded, \(failed) failed")
|
||
|
||
// Update local manifest cache immediately with every completed entry
|
||
// (uploaded + collision-skipped + fileExists-skipped).
|
||
// writeManifest() will persist this to the NAS, but even if that network
|
||
// write fails the next fast-path reconcile reads the correct counts here.
|
||
// Without this, a failed NAS write leaves the cache at 1 entry and the
|
||
// next reconcile computes alreadySafe = 1 instead of the real count.
|
||
if !manifestEntries.isEmpty {
|
||
var localMerged = baseManifest
|
||
localMerged.merge(entries: manifestEntries)
|
||
await NASManifestCache.shared.update(localMerged)
|
||
logger.info("Local cache updated — \(localMerged.entries.count) total entries")
|
||
}
|
||
if uploaded > 0 {
|
||
await NASManifestCache.shared.addToDirectoryCount(uploaded)
|
||
}
|
||
|
||
let duration = Date().timeIntervalSince(startDate)
|
||
let result = BackupResult(
|
||
uploadedCount: uploaded,
|
||
skippedCount: skipped,
|
||
failedCount: failed,
|
||
duration: duration,
|
||
totalBytes: totalBytes,
|
||
date: startDate
|
||
)
|
||
|
||
job.finish(result: result)
|
||
BackupStatusService.shared.refreshAfterBackup(entries: manifestEntries, connection: connection)
|
||
|
||
let entry = BackupHistoryEntry(result: result, nasHost: host, triggeredByLAN: triggeredByLAN)
|
||
store.appendHistoryEntry(entry)
|
||
|
||
// Notifications
|
||
if result.failedCount > 0 {
|
||
if store.notifyOnErrors {
|
||
NotificationService.notifyBackupFailed(count: result.failedCount)
|
||
}
|
||
} else if store.notifyOnComplete {
|
||
NotificationService.notifyBackupCompleted(
|
||
uploaded: result.uploadedCount,
|
||
total: BackupStatusService.shared.snapshot.phoneTotal
|
||
)
|
||
}
|
||
|
||
// Persist queue so SyncView shows last session on next launch
|
||
let snapshot = queueItems
|
||
Task.detached(priority: .utility) { [snapshot, url = Self.queueCacheURL] in
|
||
guard let data = try? JSONEncoder.kisani.encode(snapshot) else { return }
|
||
try? data.write(to: url, options: .atomic)
|
||
}
|
||
|
||
return result
|
||
}
|
||
|
||
// Returns Tailscale host when off trusted LAN and tunnel is active.
|
||
private func resolveHost(connection: NASConnection, store: ConnectionStore, lan: LANMonitor) -> String {
|
||
let onTrustedLAN = store.trustedSSIDs.contains(lan.currentSSID ?? "")
|
||
guard !onTrustedLAN,
|
||
store.useTailscaleWhenRemote,
|
||
!store.tailscaleHost.isEmpty,
|
||
lan.isTailscaleActive else { return connection.host }
|
||
return store.tailscaleHost
|
||
}
|
||
|
||
func cancel() { isCancelled = true; job.cancel() }
|
||
func pause() { job.pause() }
|
||
func resume() { job.resume() }
|
||
|
||
func resolveSuccess() { job.resolveSuccess() }
|
||
|
||
}
|
||
|
||
// 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..<maxRetries {
|
||
if attempt > 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.
|
||
/// SMBClient surfaces this as ErrorResponse with NTStatus.objectNameCollision (0xC0000035),
|
||
/// whose localizedDescription is "Object Name Collision".
|
||
/// SFTP and POSIX use EEXIST (code 17) or the string "already exists".
|
||
private func isAlreadyExistsError(_ error: Error) -> Bool {
|
||
let base = unwrapUploadError(error)
|
||
let desc = base.localizedDescription.lowercased()
|
||
return desc.contains("object name collision") ||
|
||
desc.contains("already exist") ||
|
||
desc.contains("file exist") ||
|
||
(base as NSError).code == 17 // POSIX EEXIST
|
||
}
|
||
|
||
/// Returns the underlying transport error, unwrapping BackupError.uploadFailed if present.
|
||
private func unwrapUploadError(_ error: Error) -> Error {
|
||
if case let BackupError.uploadFailed(_, underlying) = error { return underlying }
|
||
return error
|
||
}
|
||
|
||
/// Human-readable description from the underlying transport error, not the wrapper.
|
||
private func underlyingDescription(_ error: Error) -> String {
|
||
unwrapUploadError(error).localizedDescription
|
||
}
|
||
|
||
/// NSError domain from the underlying transport error.
|
||
private func underlyingDomain(_ error: Error) -> String {
|
||
(unwrapUploadError(error) as NSError).domain
|
||
}
|
||
|
||
// MARK: — SpeedTracker
|
||
|
||
private struct SpeedTracker {
|
||
private var lastBytes: Int64 = 0
|
||
private var lastTime: Date = Date()
|
||
|
||
mutating func update(bytesSent: Int64) -> Double {
|
||
let now = Date()
|
||
let elapsed = now.timeIntervalSince(lastTime)
|
||
guard elapsed > 0.1 else { return 0 }
|
||
let speed = Double(bytesSent - lastBytes) / elapsed
|
||
lastBytes = bytesSent
|
||
lastTime = now
|
||
return max(0, speed)
|
||
}
|
||
}
|
||
|
||
// MARK: — ProgressThrottle
|
||
|
||
/// Limits the rate at which progress callbacks trigger @MainActor UI updates.
|
||
/// Without this, SMB/SFTP stacks fire hundreds of callbacks per second, causing
|
||
/// SwiftUI to redraw the full view tree on every tick.
|
||
private struct ProgressThrottle {
|
||
private var lastUpdate: Date = .distantPast
|
||
private let minInterval: TimeInterval
|
||
|
||
init(hz: Double = 8) { minInterval = 1.0 / max(1, hz) }
|
||
|
||
mutating func shouldUpdate() -> Bool {
|
||
let now = Date()
|
||
guard now.timeIntervalSince(lastUpdate) >= minInterval else { return false }
|
||
lastUpdate = now
|
||
return true
|
||
}
|
||
}
|