diff --git a/BitMatch-iPadTests/QueueAccessoryPresentationTests.swift b/BitMatch-iPadTests/QueueAccessoryPresentationTests.swift index 69d01714..51a81b85 100644 --- a/BitMatch-iPadTests/QueueAccessoryPresentationTests.swift +++ b/BitMatch-iPadTests/QueueAccessoryPresentationTests.swift @@ -174,7 +174,8 @@ struct QueueAccessoryPresentationTests { id: id, cardName: name, evidence: nil, destinations: "Backup", safetyState: state, progressFraction: progress, action: nil, cause: cause, copySummary: name, outcome: nil, destinationNames: ["Backup"], - verificationModeName: VerificationMode.standard.rawValue + verificationModeName: VerificationMode.standard.rawValue, + estimatedWorkBytes: 100 ) } } diff --git a/BitMatchTests/ByteWeightedProgressPresentationTests.swift b/BitMatchTests/ByteWeightedProgressPresentationTests.swift new file mode 100644 index 00000000..ccbb6524 --- /dev/null +++ b/BitMatchTests/ByteWeightedProgressPresentationTests.swift @@ -0,0 +1,76 @@ +import Foundation +import Testing +import BitMatchEngine +@testable import BitMatch + +@MainActor +struct ByteWeightedProgressPresentationTests { + @Test func handoffUsesSelectedByteWorkAndNeverResetsOrFinishesEarly() { + let pipeline = OperationProgress(overallProgress: 1, currentFile: nil, + filesProcessed: 2, totalFiles: 2, currentStage: .completed, + speed: nil, elapsedTime: nil, averageSpeed: nil, peakSpeed: nil, + bytesProcessed: 200, totalBytes: 100, workBytesProcessed: 500, totalWorkBytes: 500) + let withMHL = CopyVerifyExecutor.pipelineProgress(pipeline, destinationCount: 2, includeMHL: true) + #expect(abs(withMHL.overallProgress - 5.0 / 7.0) < 0.00001) + let beginning = CopyVerifyExecutor.mhlProgress(.init(bytesProcessed: 0, totalBytes: 200, + filesProcessed: 0, totalFiles: 2, published: false), precedingWorkBytes: 500) + let midway = CopyVerifyExecutor.mhlProgress(.init(bytesProcessed: 100, totalBytes: 200, + filesProcessed: 1, totalFiles: 2, published: false), precedingWorkBytes: 500) + let published = CopyVerifyExecutor.mhlProgress(.init(bytesProcessed: 200, totalBytes: 200, + filesProcessed: 2, totalFiles: 2, published: true), precedingWorkBytes: 500) + #expect(beginning.overallProgress == withMHL.overallProgress) + #expect(midway.overallProgress > beginning.overallProgress) + #expect(published.overallProgress == 0.99) + #expect(published.currentStage == .generating) + #expect(CopyVerifyExecutor.pipelineProgress(pipeline, destinationCount: 2, includeMHL: false).overallProgress == 0.99) + // Mapping the bar cannot change actual copied bytes. + #expect(withMHL.bytesProcessed == pipeline.bytesProcessed) + } + + @Test func failedBackupKeepsItsUnfinishedBudgetDuringPartialHandoff() { + // Two backups reserved 500 pipeline units + 200 handoff units. + // One failed early: only 300 pipeline units and one 100-byte job remain. + let partial = CopyVerifyExecutor.mhlProgress(.init(bytesProcessed: 0, totalBytes: 100, + filesProcessed: 0, totalFiles: 1, published: false), precedingWorkBytes: 500, + precedingCompletedWorkBytes: 300, reservedHandoffBytes: 200) + let onePublished = CopyVerifyExecutor.mhlProgress(.init(bytesProcessed: 100, totalBytes: 100, + filesProcessed: 1, totalFiles: 1, published: true), precedingWorkBytes: 500, + precedingCompletedWorkBytes: 300, reservedHandoffBytes: 200) + let noJobs = CopyVerifyExecutor.mhlProgress(.init(bytesProcessed: 0, totalBytes: 0, + filesProcessed: 0, totalFiles: 0, published: false), precedingWorkBytes: 500, + precedingCompletedWorkBytes: 300, reservedHandoffBytes: 200) + #expect(partial.overallProgress == 3.0 / 7.0) + #expect(onePublished.overallProgress == 4.0 / 7.0) + #expect(noJobs.overallProgress == partial.overallProgress) + } + + @Test func tinyFinishedCardCannotSpendAHugeWaitingCardsProgress() { + let tiny = row(state: .safeToErase, work: 1) + let huge = row(state: .copying(progress: 0), fraction: 0, work: 999) + let started = QueueAccessoryPresentation.weightedProgress(rows: [tiny, huge]) + #expect(started == 0.001) + var half = huge + half = row(state: .verifying(progress: 50), fraction: 0.5, work: 999) + #expect(QueueAccessoryPresentation.weightedProgress(rows: [tiny, half]) == 0.5005) + #expect(QueueAccessoryPresentation.weightedProgress(rows: [tiny, row(state: .waiting, work: nil)]) == nil) + #expect(QueueAccessoryPresentation.weightedProgress(rows: [row(state: .finishing(progress: 100), fraction: 1, work: 100)]) == 0.99) + } + + @Test func destinationBarsPreferByteWorkOverEqualFileCounts() throws { + let update = OperationProgress(overallProgress: 0.4, currentFile: nil, + filesProcessed: 2, totalFiles: 4, currentStage: .copying, speed: nil, + elapsedTime: nil, averageSpeed: nil, peakSpeed: nil, bytesProcessed: 2, totalBytes: 1000, + perDestinationTotals: [2, 2], perDestinationCompleted: [1, 1], perDestinationProgress: [0.001, 0.7]) + let rows = TransferProgressPresentation.rows(destinations: [URL(fileURLWithPath: "/one"), URL(fileURLWithPath: "/two")], + progress: update, phase: .copying) + #expect(rows[0].fraction == 0.001) + #expect(rows[1].fraction == 0.7) + #expect(rows[0].countText == "1 of 2") + } + + private func row(state: CardSafetyState, fraction: Double? = nil, work: Double?) -> QueueSessionRow { + QueueSessionRow(id: UUID(), cardName: "Card", evidence: nil, destinations: "Backup", + safetyState: state, progressFraction: fraction, action: nil, cause: nil, copySummary: "", + outcome: nil, destinationNames: ["Backup"], verificationModeName: "Standard", estimatedWorkBytes: work) + } +} diff --git a/BitMatchTests/CopyVerifyExecutorIntegrityTests.swift b/BitMatchTests/CopyVerifyExecutorIntegrityTests.swift index 815b7b22..659e4b2a 100644 --- a/BitMatchTests/CopyVerifyExecutorIntegrityTests.swift +++ b/BitMatchTests/CopyVerifyExecutorIntegrityTests.swift @@ -326,6 +326,20 @@ final class CopyVerifyExecutorIntegrityTests: XCTestCase { XCTAssertEqual(harness.lastState, .cancelled) } + func testImmediateFinalCallbackReachesAuthoritativeResultsWithoutWaitingForUITask() async throws { + let final = OperationProgress(overallProgress: 1, currentFile: nil, + filesProcessed: 1, totalFiles: 1, currentStage: .completed, speed: nil, + elapsedTime: nil, averageSpeed: nil, peakSpeed: nil, bytesProcessed: 10, + totalBytes: 10, workBytesProcessed: 30, totalWorkBytes: 30) + let harness = ExecutorHarness(returnedResults: [verifiedResult()], emittedResults: [], terminalProgress: final) + harness.onAuthoritativeResults = { + XCTAssertEqual(harness.observedProgress.last?.totalWorkBytes, 30) + XCTAssertEqual(harness.observedProgress.last?.overallProgress, 0.99) + } + _ = try await harness.execute() + XCTAssertEqual(harness.terminalInfo?.success, true) + } + func testMHLReadProgressIsNotFinishedBeforePublication() { let update = CopyVerifyExecutor.mhlProgress(.init(bytesProcessed: 50, totalBytes: 100, filesProcessed: 0, totalFiles: 1, published: false)) @@ -1014,6 +1028,7 @@ private final class ExecutorHarness { reportPrefs: ReportPrefs? = nil, cameraSettings: CameraLabelSettings = CameraLabelSettings(), thrownError: Error? = nil, + terminalProgress: OperationProgress? = nil, realFileOperations: FileOperationsService? = nil, estimatedFiles: Int? = nil, estimatedBytes: Int64? = nil, @@ -1022,7 +1037,8 @@ private final class ExecutorHarness { let fileOperations = ExecutorFileOperationsService( returnedResults: returnedResults, emittedResults: emittedResults, - thrownError: thrownError + thrownError: thrownError, + terminalProgress: terminalProgress ) let platform = ExecutorPlatformManager(fileOperations: realFileOperations ?? fileOperations) self.platform = platform @@ -1089,11 +1105,13 @@ private final class ExecutorFileOperationsService: FileOperationsService { private let returnedResults: [FileOperationResult] private let emittedResults: [FileOperationResult] private let thrownError: Error? + private let terminalProgress: OperationProgress? - init(returnedResults: [FileOperationResult], emittedResults: [FileOperationResult], thrownError: Error? = nil) { + init(returnedResults: [FileOperationResult], emittedResults: [FileOperationResult], thrownError: Error? = nil, terminalProgress: OperationProgress? = nil) { self.returnedResults = returnedResults self.emittedResults = emittedResults self.thrownError = thrownError + self.terminalProgress = terminalProgress } func performFileOperation( @@ -1109,6 +1127,7 @@ private final class ExecutorFileOperationsService: FileOperationsService { await onFileResult?(result) } if let thrownError { throw thrownError } + if let terminalProgress { progressCallback(terminalProgress) } let sourceManifest = returnedResults.reduce(into: [URL]()) { files, result in if !files.contains(result.sourceURL) { files.append(result.sourceURL) } } diff --git a/BitMatchTests/QueueSessionPresentationTests.swift b/BitMatchTests/QueueSessionPresentationTests.swift index a3786fc6..c2bb8814 100644 --- a/BitMatchTests/QueueSessionPresentationTests.swift +++ b/BitMatchTests/QueueSessionPresentationTests.swift @@ -50,7 +50,7 @@ struct QueueSessionPresentationTests { records: [running], sessionIDs: [running.id], progress: verifyingProgress, mountedSourceIDs: [] ) - #expect(verifying.rows.first?.oneLineStatus().contains("Verifying 40%") == true) + #expect(verifying.rows.first?.oneLineStatus().contains("Verifying 70%") == true) } @Test func fileWorkAtOneHundredPercentIsStillFinishingWhileRecordRuns() throws { diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift index f83b729b..17636372 100644 --- a/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift @@ -562,7 +562,7 @@ public final class PinnedDestinationFile: @unchecked Sendable { #endif public final class DestinationWriter { - typealias IOByteCallback = @Sendable (Int, TransferIOPhase, Int, TimeInterval, TimeInterval) -> Void + typealias IOByteCallback = @Sendable (Int, String, TransferIOPhase, Int, TimeInterval, TimeInterval) -> Void struct SourceReadEvidence: Sendable { let digests: VerifiedDigests let identity: VerifiedFileIdentity @@ -956,7 +956,10 @@ public final class DestinationWriter { destinationIndex: destination.index, relativePath: relativePath, hooks: hooks, - onIOBytes: onIOBytes + onIOBytes: { index, path, phase, bytes, duration, ended in + let reusePhase: TransferIOPhase = phase == .readingBack ? .validatingExisting : .sourceValidatingExisting + onIOBytes?(index, path, reusePhase, bytes, duration, ended) + } ) pendingReuse.append(destination) } else { @@ -1026,6 +1029,7 @@ public final class DestinationWriter { throw sourceChangedError() } bytesRead += Int64(data.count) + onIOBytes?(-1, relativePath, .sourceReading, data.count, 0, ProcessInfo.processInfo.systemUptime) if let sourceDidRead = hooks?.sourceDidRead { let readCount = data.count data = try await sourceDidRead(source, data) @@ -1058,7 +1062,7 @@ public final class DestinationWriter { try file.destination.root.requireIncompletePublicationStatus() } try writeAll(chunkData, to: file.temporaryFD) { count, seconds, ended in - onIOBytes?(file.destination.index, .writing, count, seconds, ended) + onIOBytes?(file.destination.index, relativePath, .writing, count, seconds, ended) } return FanOutWriteResult(file: file, error: nil) } catch { @@ -1109,7 +1113,7 @@ public final class DestinationWriter { source, algorithms: verificationMode.checksumTypes + [.md5], relativePath: relativePath, - hooks: hooks + hooks: hooks, onIOBytes: onIOBytes ) ) } else if verificationMode != .quick { @@ -1573,6 +1577,23 @@ public final class DestinationWriter { ) async throws -> VerificationResult { let startTime = Date() var sourceValues: [ChecksumAlgorithm: String] = [:] + let sourceSize = try sourceFileSize(source) + func sourceChecksum(_ type: ChecksumAlgorithm) async throws -> String { + let lastBytes = Mutex(0) + return try await checksumService.generateChecksum(for: source, type: type) { fraction, _ in + guard fraction.isFinite else { return } + let current: Int64 = fraction >= 1 ? sourceSize + : max(0, Int64(Double(sourceSize) * max(0, fraction))) + let delta = lastBytes.withLock { last -> Int64 in + let delta = max(0, current - last) + last = max(last, current) + return delta + } + if delta > 0 { + onIOBytes?(destinationIndex, relativePath, .sourceVerifying, Int(delta), 0, ProcessInfo.processInfo.systemUptime) + } + } + } if let sourceReadEvidence { guard try sourceIdentity(at: source) == sourceReadEvidence.identity else { throw sourceChangedError() @@ -1581,11 +1602,7 @@ public final class DestinationWriter { if let value = sourceReadEvidence.digests[type] { sourceValues[type] = value } else { - sourceValues[type] = try await checksumService.generateChecksum( - for: source, - type: type, - progressCallback: nil - ) + sourceValues[type] = try await sourceChecksum(type) } } guard try sourceIdentity(at: source) == sourceReadEvidence.identity else { @@ -1593,11 +1610,7 @@ public final class DestinationWriter { } } else { for type in requiredTypes { - sourceValues[type] = try await checksumService.generateChecksum( - for: source, - type: type, - progressCallback: nil - ) + sourceValues[type] = try await sourceChecksum(type) } } // MD5 is collected in the same pass even when SHA-256 is the only @@ -1700,7 +1713,8 @@ public final class DestinationWriter { _ source: URL, algorithms: [ChecksumAlgorithm], relativePath: String, - hooks: FanOutHooks? + hooks: FanOutHooks?, + onIOBytes: IOByteCallback? = nil ) async throws -> SourceReadEvidence { let handle = try uncachedSourceHandle(for: source) hooks?.verificationSourceDidOpen?(relativePath) @@ -1726,6 +1740,7 @@ public final class DestinationWriter { } guard count > 0 else { throw sourceChangedError() } hooks?.verificationSourceDidRead?(relativePath, count) + onIOBytes?(-1, relativePath, .sourceVerifying, count, 0, ProcessInfo.processInfo.systemUptime) bytesRead += Int64(count) } let trailingData = try handle.read(upToCount: 1) ?? Data() @@ -1779,7 +1794,7 @@ public final class DestinationWriter { } if count == 0 { break } hooks?.destinationDidRead?(destinationIndex, relativePath, count) - onIOBytes?(destinationIndex, .readingBack, count, readSeconds, readEnded) + onIOBytes?(destinationIndex, relativePath, .readingBack, count, readSeconds, readEnded) bytesRead += Int64(count) if ordinal != nil, bytesRead >= nextBreadcrumb { nextBreadcrumb += verifyReadBreadcrumbInterval @@ -1844,7 +1859,8 @@ public final class DestinationWriter { } hooks?.verificationSourceDidRead?(relativePath, sourceCount) hooks?.destinationDidRead?(destinationIndex, relativePath, destinationCount) - onIOBytes?(destinationIndex, .readingBack, destinationCount, readSeconds, readEnded) + onIOBytes?(destinationIndex, relativePath, .readingBack, destinationCount, readSeconds, readEnded) + onIOBytes?(destinationIndex, relativePath, .sourceVerifying, sourceCount, 0, readEnded) if !same { return false } bytesRead += Int64(sourceCount) } diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/OperationProgress.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/OperationProgress.swift index 74a706fa..db34c91d 100644 --- a/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/OperationProgress.swift +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/OperationProgress.swift @@ -43,6 +43,10 @@ public struct OperationProgress: Codable, Sendable { // Per-destination progress (optional) public let perDestinationTotals: [Int]? public let perDestinationCompleted: [Int]? + /// Presentation work only, distinct from actual copied bytes/evidence. + public let workBytesProcessed: Double? + public let totalWorkBytes: Double? + public let perDestinationProgress: [Double]? // Convenience initializer for backward compatibility public init(overallProgress: Double, currentFile: String?, filesProcessed: Int, totalFiles: Int, currentStage: ProgressStage, speed: Double?, reusedCopies: Int? = nil) { @@ -62,10 +66,13 @@ public struct OperationProgress: Codable, Sendable { self.isASCMHL = nil self.perDestinationTotals = nil self.perDestinationCompleted = nil + self.workBytesProcessed = nil + self.totalWorkBytes = nil + self.perDestinationProgress = nil } // Full initializer with timing information - public init(overallProgress: Double, currentFile: String?, filesProcessed: Int, totalFiles: Int, currentStage: ProgressStage, speed: Double?, elapsedTime: TimeInterval?, averageSpeed: Double?, peakSpeed: Double?, bytesProcessed: Int64?, totalBytes: Int64?, stageProgress: Double? = nil, reusedCopies: Int? = nil, perDestinationTotals: [Int]? = nil, perDestinationCompleted: [Int]? = nil, isASCMHL: Bool? = nil) { + public init(overallProgress: Double, currentFile: String?, filesProcessed: Int, totalFiles: Int, currentStage: ProgressStage, speed: Double?, elapsedTime: TimeInterval?, averageSpeed: Double?, peakSpeed: Double?, bytesProcessed: Int64?, totalBytes: Int64?, stageProgress: Double? = nil, reusedCopies: Int? = nil, perDestinationTotals: [Int]? = nil, perDestinationCompleted: [Int]? = nil, isASCMHL: Bool? = nil, workBytesProcessed: Double? = nil, totalWorkBytes: Double? = nil, perDestinationProgress: [Double]? = nil) { self.overallProgress = overallProgress self.currentFile = currentFile self.filesProcessed = filesProcessed @@ -82,5 +89,20 @@ public struct OperationProgress: Codable, Sendable { self.isASCMHL = isASCMHL self.perDestinationTotals = perDestinationTotals self.perDestinationCompleted = perDestinationCompleted + self.workBytesProcessed = workBytesProcessed + self.totalWorkBytes = totalWorkBytes + self.perDestinationProgress = perDestinationProgress + } + /// Preserve the engine's measurements while mapping the bar across handoff + /// and report work. This changes presentation only. + public func replacingOverallProgress(_ fraction: Double) -> OperationProgress { + OperationProgress(overallProgress: fraction, currentFile: currentFile, + filesProcessed: filesProcessed, totalFiles: totalFiles, currentStage: currentStage, + speed: speed, elapsedTime: elapsedTime, averageSpeed: averageSpeed, peakSpeed: peakSpeed, + bytesProcessed: bytesProcessed, totalBytes: totalBytes, stageProgress: stageProgress, + reusedCopies: reusedCopies, perDestinationTotals: perDestinationTotals, + perDestinationCompleted: perDestinationCompleted, isASCMHL: isASCMHL, + workBytesProcessed: workBytesProcessed, totalWorkBytes: totalWorkBytes, + perDestinationProgress: perDestinationProgress) } } diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift index 40699811..5e71c1ed 100644 --- a/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift @@ -44,6 +44,10 @@ public struct TransferIOProgress: Equatable, Sendable { enum TransferIOPhase: Sendable { case writing case readingBack + case validatingExisting + case sourceReading + case sourceVerifying + case sourceValidatingExisting } /// Counts bytes only after an actual destination I/O call succeeds. A short @@ -136,8 +140,10 @@ final class TransferIOMeter: Sendable { switch phase { case .writing: state.destinations[destination].writing.add(bytes: bytes, duration: duration, now: now) - case .readingBack: + case .readingBack, .validatingExisting: state.destinations[destination].readingBack.add(bytes: bytes, duration: duration, now: now) + case .sourceReading, .sourceVerifying, .sourceValidatingExisting: + return nil } guard state.lastPublishUptime.map({ now - $0 >= 0.25 }) ?? true else { return nil } state.lastPublishUptime = now diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferWorkProgress.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferWorkProgress.swift new file mode 100644 index 00000000..d5bb2ba7 --- /dev/null +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferWorkProgress.swift @@ -0,0 +1,131 @@ +import Foundation +import Synchronization + +/// A presentation budget measured in selected-file bytes, never transfer +/// evidence. Validated reuse settles the corresponding work without claiming +/// that bytes were written. Each file has its own caps, so repeated checks or +/// a fast backup cannot spend another file or backup's budget. +final class TransferWorkMeter: Sendable { + struct Snapshot: Sendable { + let completed: Double + let total: Double + let perDestination: [Double] + var fraction: Double { total > 0 ? min(0.99, completed / total) : 0 } + } + + private struct FileWork { + let size: Double + var sourceRead: Double = 0 + var sourceReread: Double = 0 + var copied: [Double] + var verified: [Double] + } + private struct State { + var files: [String: FileWork] + var lastPublish: TimeInterval? + var completed = 0.0 + var total = 0.0 + var perDestinationCompleted: [Double] + var perDestinationTotal = 0.0 + + mutating func replace(_ file: FileWork, at path: String, previous: FileWork) { + let sharedDelta = file.sourceRead + file.sourceReread - previous.sourceRead - previous.sourceReread + var delta = sharedDelta + let count = perDestinationCompleted.count + for index in perDestinationCompleted.indices { + let destinationDelta = file.copied[index] + file.verified[index] + - previous.copied[index] - previous.verified[index] + delta += destinationDelta + perDestinationCompleted[index] += destinationDelta + sharedDelta / Double(max(1, count)) + } + completed += delta + files[path] = file + } + } + private let destinations: Int + private let destinationPasses: Double + private let sourceRereadPasses: Double + private let state: Mutex + + init(files: [FileEntry], destinationCount: Int, mode: VerificationMode) { + destinations = destinationCount + destinationPasses = mode == .quick ? 0 : (mode == .paranoid ? 2 : 1) + sourceRereadPasses = mode == .thorough ? 1 : (mode == .paranoid ? Double(destinationCount) : 0) + // Planning is presentation-only. A malformed duplicate must not trap + // the process or authorize anything; the source/coverage gates still + // own whether that manifest is admissible for copying. + let plannedFiles = Dictionary(files.map { + // Empty files need a scheduling unit, never a fabricated byte of + // copy/read evidence. + ($0.relativePath, FileWork(size: Double(max(1, $0.size)), + copied: Array(repeating: 0, count: destinationCount), + verified: Array(repeating: 0, count: destinationCount))) + }, uniquingKeysWith: { first, _ in first }) + let selectedWork = plannedFiles.values.reduce(0.0) { $0 + $1.size } + let sharedWork = selectedWork * (1 + sourceRereadPasses) + state = Mutex(State(files: plannedFiles, + total: sharedWork + selectedWork * Double(destinationCount) * (1 + destinationPasses), + perDestinationCompleted: Array(repeating: 0, count: destinationCount), + perDestinationTotal: selectedWork * (1 + destinationPasses) + sharedWork / Double(max(1, destinationCount)))) + } + + func record(path: String, destination: Int, phase: TransferIOPhase, bytes: Int, + now: TimeInterval) -> Bool { + state.withLock { state in + guard bytes > 0, var file = state.files[path] else { return false } + let previous = file + let count = Double(bytes) + switch phase { + case .sourceReading, .sourceValidatingExisting: + file.sourceRead = min(file.size, file.sourceRead + count) + case .sourceVerifying: + file.sourceReread = min(file.size * sourceRereadPasses, file.sourceReread + count) + case .writing, .validatingExisting: + guard file.copied.indices.contains(destination) else { return false } + file.copied[destination] = min(file.size, file.copied[destination] + count) + case .readingBack: + guard file.verified.indices.contains(destination) else { return false } + file.verified[destination] = min(file.size * destinationPasses, file.verified[destination] + count) + } + state.replace(file, at: path, previous: previous) + guard state.lastPublish.map({ now - $0 >= 0.25 }) ?? true else { return false } + state.lastPublish = now + return true + } + } + + func copied(path: String, destination: Int) { + state.withLock { state in + guard var file = state.files[path], file.copied.indices.contains(destination) else { return } + let previous = file + file.sourceRead = file.size + file.copied[destination] = file.size + state.replace(file, at: path, previous: previous) + } + } + + func verified(path: String, destination: Int) { + state.withLock { state in + guard var file = state.files[path], file.verified.indices.contains(destination) else { return } + let previous = file + file.verified[destination] = file.size * destinationPasses + // Shared source evidence can avoid a separate read per backup. + // Settle unused work only after every destination actually checks. + if file.verified.allSatisfy({ $0 == file.size * destinationPasses }) { + file.sourceReread = file.size * sourceRereadPasses + } + state.replace(file, at: path, previous: previous) + } + } + + func snapshot() -> Snapshot { + state.withLock { state in + // No file traversal here: a 10,000-file card has the same + // publication cost as a one-file card (per selected destination). + Snapshot(completed: state.completed, total: state.total, + perDestination: state.perDestinationCompleted.map { + state.perDestinationTotal > 0 ? min(0.99, $0 / state.perDestinationTotal) : 0 + }) + } + } +} diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferJournal.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferJournal.swift index ff1282a6..ebf1c268 100644 --- a/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferJournal.swift +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferJournal.swift @@ -127,6 +127,8 @@ public struct LocalTransferRecord: Identifiable, Codable, Sendable { public let retryOf: UUID? /// Selected files per destination, from the engine manifest (exclusions omitted). public var plannedFileCounts: [Int]? + /// Selection-size estimate for queue progress only. Never copy evidence. + public var estimatedSourceBytes: Int64? /// True only when Quick's size-only check was the sole reason the attempt /// could not be verified. False when reporting, coverage, or another part /// of the attempt also failed; nil for journals written before this field. @@ -160,7 +162,7 @@ public struct LocalTransferRecord: Identifiable, Codable, Sendable { case id, createdAt, source, destinations, verificationMode, cameraSettings, reportSettings case generateASCMHL, projectID, projectCardID, state, startedAt, endedAt case copyDurationSeconds, verifyDurationSeconds, performanceTelemetry, summary, results, sourceFingerprint - case independentDestinationCount, retryOf, plannedFileCounts, quickOnlyUnverified + case independentDestinationCount, retryOf, plannedFileCounts, quickOnlyUnverified, estimatedSourceBytes } public init(from decoder: Decoder) throws { @@ -193,6 +195,7 @@ public struct LocalTransferRecord: Identifiable, Codable, Sendable { independentDestinationCount = try c.decodeIfPresent(Int.self, forKey: .independentDestinationCount) retryOf = try c.decodeIfPresent(UUID.self, forKey: .retryOf) plannedFileCounts = try c.decodeIfPresent([Int].self, forKey: .plannedFileCounts) + estimatedSourceBytes = try c.decodeIfPresent(Int64.self, forKey: .estimatedSourceBytes) quickOnlyUnverified = try c.decodeIfPresent(Bool.self, forKey: .quickOnlyUnverified) } public var issueCount: Int { results.filter { !$0.isSuccessStatus }.count } @@ -302,12 +305,13 @@ public final class TransferJournal: Sendable { @discardableResult public func enqueue(sourceURL: URL, destinationURLs: [URL], verificationMode: VerificationMode, cameraSettings: CameraLabelSettings, reportSettings: ReportPrefs, generateASCMHL: Bool = true, - projectID: UUID? = nil, projectCardID: UUID? = nil) throws -> UUID { + projectID: UUID? = nil, projectCardID: UUID? = nil, estimatedSourceBytes: Int64? = nil) throws -> UUID { guard !destinationURLs.isEmpty else { throw LocalTransferJournalError.missingDestinations } - let record = LocalTransferRecord(id: UUID(), createdAt: Date(), source: try LocalTransferResource(url: sourceURL), + var record = LocalTransferRecord(id: UUID(), createdAt: Date(), source: try LocalTransferResource(url: sourceURL), destinations: try destinationURLs.map(LocalTransferResource.init(url:)), verificationMode: verificationMode, cameraSettings: cameraSettings, reportSettings: reportSettings, generateASCMHL: generateASCMHL, projectID: projectID, projectCardID: projectCardID) + record.estimatedSourceBytes = estimatedSourceBytes.flatMap { $0 >= 0 ? $0 : nil } try commit { [record] + $0 } return record.id } @@ -319,11 +323,12 @@ public final class TransferJournal: Sendable { original.state != .queued else { throw LocalTransferJournalError.invalidState } let access = try prepareToRun(id: id) defer { access.release() } - let retry = LocalTransferRecord(id: UUID(), createdAt: Date(), source: original.source, + var retry = LocalTransferRecord(id: UUID(), createdAt: Date(), source: original.source, destinations: original.destinations, verificationMode: original.verificationMode, cameraSettings: original.cameraSettings, reportSettings: original.reportSettings, generateASCMHL: generateASCMHL ?? original.generateASCMHL, projectID: original.projectID, projectCardID: original.projectCardID, retryOf: original.id) + retry.estimatedSourceBytes = original.estimatedSourceBytes try commit { [retry] + $0 } return retry.id } @@ -475,7 +480,7 @@ public final class TransferJournal: Sendable { } /// Save once, before results settle. Unknown counts in older records stay unknown. - public func recordPlan(id: UUID, fileCounts: [Int]) throws { + public func recordPlan(id: UUID, fileCounts: [Int], selectedSourceBytes: Int64? = nil) throws { try update(id: id) { record in guard record.state == .running, fileCounts.count == record.destinations.count, fileCounts.allSatisfy({ $0 >= 0 }), @@ -483,6 +488,11 @@ public final class TransferJournal: Sendable { throw LocalTransferJournalError.invalidState } record.plannedFileCounts = fileCounts + // Keep an existing queue estimate stable throughout this attempt. + // A late scan correction must not rewind the whole queue bar. + if record.estimatedSourceBytes == nil, let selectedSourceBytes, selectedSourceBytes >= 0 { + record.estimatedSourceBytes = selectedSourceBytes + } } } @@ -552,7 +562,8 @@ public final class TransferJournal: Sendable { verificationMode: VerificationMode, cameraSettings: CameraLabelSettings, reportSettings: ReportPrefs, - generateASCMHL: Bool + generateASCMHL: Bool, + estimatedSourceBytes: Int64? = nil ) throws { guard !destinationURLs.isEmpty else { throw LocalTransferJournalError.missingDestinations } let source = try LocalTransferResource(url: sourceURL) @@ -575,6 +586,7 @@ public final class TransferJournal: Sendable { && source.resourceID == record.source.resourceID && source.creationDate == record.source.creationDate ? record.retryOf : nil ) + record.estimatedSourceBytes = estimatedSourceBytes.flatMap { $0 >= 0 ? $0 : nil } } } } diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift index a34bc61c..d9426677 100644 --- a/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift @@ -540,17 +540,9 @@ public final class TransferPipeline: FileOperationsService, Sendable { let totalFiles = perSourceFileCount * operation.destinationURLs.count SharedLogger.debug("Prep: source files=\(perSourceFileCount), destinations=\(operation.destinationURLs.count), planned total rows=\(totalFiles)", category: .transfer) let destinationCount = operation.destinationURLs.count - let totalStageUnits = operation.verificationMode == .quick ? 1 : 2 // Perf 2: time-based throttle on progress callbacks (500ms) let ledger = RunLedger(destinationCount: destinationCount, filesPerDestination: perSourceFileCount, throttle: 0.5) let ioMeter = TransferIOMeter(destinationCount: destinationCount) - let onIOBytes: DestinationWriter.IOByteCallback = { index, phase, bytes, duration, ended in - if let snapshot = ioMeter.record(destination: index, phase: phase, - bytes: bytes, duration: duration, now: ended) { - ioProgressCallback(snapshot) - } - } - defer { ioProgressCallback(ioMeter.finalSnapshot()) } // Retain an exclusion row for every source/destination pair before any // copying, so cancellation and saved history keep the selected scope. for (index, destination) in operation.destinationURLs.enumerated() { @@ -577,44 +569,62 @@ public final class TransferPipeline: FileOperationsService, Sendable { SharedLogger.verifyConfiguration(run: diagnosticRun, concurrency: shouldPipelineVerify ? verifyConcurrency : 1) let fileOrdinals = Dictionary(uniqueKeysWithValues: sourceManifest.enumerated().map { ($0.element.relativePath, $0.offset + 1) }) - // The one way progress is reported. Total bytes: the caller's - // estimate, else the average copied file size times the plan. - let makeProgress: @Sendable (ProgressStage, String?, RunLedger.Snapshot, Date, Double?) -> OperationProgress = { - stage, file, snapshot, now, stageProgress in - let elapsed = now.timeIntervalSince(startTime) - let speed = elapsed > 0 ? Double(snapshot.bytesCopied) / elapsed : nil - let totalBytes: Int64 = { - // The caller's scan includes omitted metadata. For a filtered - // run use the measured selected bytes per backup instead. - if !excludedPaths.isEmpty { return manifestBytes } - if let estimate = operation.estimatedTotalBytes, estimate > 0 { return estimate } - if snapshot.filesCopied > 0 { - return safeMultiply(Int64(totalFiles), snapshot.bytesCopied / Int64(snapshot.filesCopied)) + // Progress uses one stable per-file byte-work plan. Copied-byte + // telemetry below remains actual selected copy/reuse coverage only. + let workMeter = TransferWorkMeter(files: sourceManifest, destinationCount: destinationCount, + mode: operation.verificationMode) + let progressEmission = Mutex(nil) + let emitProgress: @Sendable (ProgressStage, String?, RunLedger.Snapshot?, Date, Double?) -> Void = { + stage, file, incoming, now, _ in + // Serialize callbacks, not merely counter updates: independent + // copy and verify tasks must not publish an older percentage last. + progressEmission.withLock { cached in + if let incoming { + if let previous = cached { + cached = RunLedger.Snapshot(filesCopied: max(previous.filesCopied, incoming.filesCopied), + bytesCopied: max(previous.bytesCopied, incoming.bytesCopied), + filesVerified: max(previous.filesVerified, incoming.filesVerified), + perDestinationTotals: incoming.perDestinationTotals, + perDestinationCompleted: zip(previous.perDestinationCompleted, incoming.perDestinationCompleted).map { max($0, $1) }) + } else { cached = incoming } } - return safeMultiply(50 * 1024 * 1024, Int64(totalFiles)) - }() - let unitsDone = stage == .copying ? snapshot.filesCopied : snapshot.filesCopied + snapshot.filesVerified - return OperationProgress( - // Reserve 100% for the terminal event, after coverage checks - // and every selected row's outcome are known. A failed final - // verify can otherwise briefly display 100% as it finishes. - overallProgress: min(0.99, Double(unitsDone) / Double(max(1, totalFiles * totalStageUnits))), - currentFile: file, - filesProcessed: snapshot.filesCopied, - totalFiles: totalFiles, - currentStage: stage, - speed: speed, - elapsedTime: elapsed, - averageSpeed: speed, - peakSpeed: nil, - bytesProcessed: snapshot.bytesCopied, - totalBytes: totalBytes, - stageProgress: stageProgress.map { min(0.99, $0) }, - reusedCopies: nil, - perDestinationTotals: snapshot.perDestinationTotals, - perDestinationCompleted: snapshot.perDestinationCompleted - ) + guard let snapshot = cached else { return } + let work = workMeter.snapshot() + let elapsed = now.timeIntervalSince(startTime) + let speed = elapsed > 0 ? Double(snapshot.bytesCopied) / elapsed : nil + let activeStage: ProgressStage = stage == .completed ? .completed + : (snapshot.filesCopied >= totalFiles && operation.verificationMode != .quick ? .verifying : .copying) + progressCallback(OperationProgress( + overallProgress: work.fraction, + currentFile: file, + filesProcessed: snapshot.filesCopied, + totalFiles: totalFiles, + currentStage: activeStage, + speed: speed, + elapsedTime: elapsed, + averageSpeed: speed, + peakSpeed: nil, + bytesProcessed: snapshot.bytesCopied, + totalBytes: manifestBytes, + stageProgress: nil, + reusedCopies: nil, + perDestinationTotals: snapshot.perDestinationTotals, + perDestinationCompleted: snapshot.perDestinationCompleted, + workBytesProcessed: work.completed, totalWorkBytes: work.total, + perDestinationProgress: work.perDestination + )) + } } + let onIOBytes: DestinationWriter.IOByteCallback = { index, path, phase, bytes, duration, ended in + if let snapshot = ioMeter.record(destination: index, phase: phase, + bytes: bytes, duration: duration, now: ended) { + ioProgressCallback(snapshot) + } + if workMeter.record(path: path, destination: index, phase: phase, bytes: bytes, now: ended) { + emitProgress(.copying, nil, nil, Date(), nil) + } + } + defer { ioProgressCallback(ioMeter.finalSnapshot()) } let sourceFileURLs = sourceManifest.map(\.url) let sourceReadEvidence = SourceReadEvidenceStore() @@ -655,6 +665,7 @@ public final class TransferPipeline: FileOperationsService, Sendable { } } diagnosticOutcome = checked.verification.matches ? .matched : .mismatched + if checked.verification.matches { workMeter.verified(path: job.relativePath, destination: job.destinationIndex) } let verificationResult = checked.verification let verified = FileOperationResult( sourceURL: job.source, @@ -673,10 +684,10 @@ public final class TransferPipeline: FileOperationsService, Sendable { now: Date() ) if event.emit { - progressCallback(makeProgress( + emitProgress( .verifying, job.source.lastPathComponent, event.snapshot, Date(), Double(event.snapshot.filesVerified) / Double(max(1, totalFiles)) - )) + ) } await onFileResult?(verified) } catch is CancellationError { @@ -819,7 +830,7 @@ public final class TransferPipeline: FileOperationsService, Sendable { // first completed-file callback, without advancing any result or // percentage. Live destination I/O can then appear immediately. if !fanOutDestinations.isEmpty { - progressCallback(makeProgress(.copying, nil, await ledger.snapshot(), Date(), nil)) + emitProgress(.copying, nil, await ledger.snapshot(), Date(), nil) } try await DestinationWriter.copyAllSafelyFanOut( @@ -851,6 +862,7 @@ public final class TransferPipeline: FileOperationsService, Sendable { processingTime: 0, wasReused: wasReused ) + workMeter.copied(path: relativePath, destination: destIndex) let copied = await ledger.recordCopy( copyResult, relativePath: relativePath, @@ -874,7 +886,7 @@ public final class TransferPipeline: FileOperationsService, Sendable { } if copied.emit { - progressCallback(makeProgress(.copying, relativePath, copied.snapshot, Date(), nil)) + emitProgress(.copying, relativePath, copied.snapshot, Date(), nil) } }, onError: { destIndex, relativePath, error in @@ -930,10 +942,10 @@ public final class TransferPipeline: FileOperationsService, Sendable { // directory descriptor; this URL is report metadata only. let event = await ledger.beginSequentialVerify(now: Date()) if event.emit { - progressCallback(makeProgress( + emitProgress( .verifying, fileURL.lastPathComponent, event.snapshot, Date(), Double(event.snapshot.filesVerified) / Double(max(1, totalFiles)) - )) + ) } let checked = try await TransferDiagnostics.$runID.withValue(diagnosticRun) { try await TransferDiagnostics.$verifyOrdinal.withValue(ordinal) { @@ -952,6 +964,7 @@ public final class TransferPipeline: FileOperationsService, Sendable { } } diagnosticOutcome = checked.verification.matches ? .matched : .mismatched + if checked.verification.matches { workMeter.verified(path: relativePath, destination: destIndex) } let verificationResult = checked.verification let fileSize = sizeForVerify @@ -971,6 +984,7 @@ public final class TransferPipeline: FileOperationsService, Sendable { destination: destIndex ) await onFileResult?(result) + emitProgress(.verifying, fileURL.lastPathComponent, await ledger.snapshot(), Date(), nil) } catch is CancellationError { diagnosticOutcome = .cancelled @@ -1204,7 +1218,7 @@ public final class TransferPipeline: FileOperationsService, Sendable { // one selected copy or verification row failed or is missing. if selectedFilesSucceeded { let final = await ledger.snapshot() - let finalProgress = makeProgress(.completed, nil, final, Date(), nil) + let finalWork = workMeter.snapshot() progressCallback(OperationProgress( overallProgress: 1.0, currentFile: nil, @@ -1212,14 +1226,16 @@ public final class TransferPipeline: FileOperationsService, Sendable { totalFiles: totalFiles, currentStage: .completed, speed: nil, - elapsedTime: finalProgress.elapsedTime, + elapsedTime: Date().timeIntervalSince(startTime), averageSpeed: nil, peakSpeed: nil, bytesProcessed: final.bytesCopied, - totalBytes: finalProgress.totalBytes, + totalBytes: manifestBytes, stageProgress: nil, perDestinationTotals: final.perDestinationTotals, - perDestinationCompleted: final.perDestinationCompleted + perDestinationCompleted: final.perDestinationCompleted, + workBytesProcessed: finalWork.completed, totalWorkBytes: finalWork.total, + perDestinationProgress: finalWork.perDestination )) } diff --git a/Packages/BitMatchEngine/Tests/BitMatchEngineTests/ByteWeightedProgressTests.swift b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/ByteWeightedProgressTests.swift new file mode 100644 index 00000000..f85c28e4 --- /dev/null +++ b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/ByteWeightedProgressTests.swift @@ -0,0 +1,179 @@ +import Foundation +import Synchronization +import Testing +@testable import BitMatchEngine + +struct ByteWeightedProgressTests { + private func entry(_ path: String, _ bytes: Int64) -> FileEntry { + FileEntry(url: URL(fileURLWithPath: "/unused/\(path)"), relativePath: path, size: bytes) + } + + @Test func perFileAndDestinationCapsDoNotSpendAnotherClipsWork() { + let meter = TransferWorkMeter(files: [entry("sidecar", 1), entry("clip", 999)], destinationCount: 2, mode: .standard) + meter.copied(path: "sidecar", destination: 0) + meter.verified(path: "sidecar", destination: 0) + _ = meter.record(path: "sidecar", destination: 0, phase: .readingBack, bytes: 10_000, now: 1) + let tiny = meter.snapshot() + #expect(tiny.fraction < 0.001) + _ = meter.record(path: "clip", destination: -1, phase: .sourceReading, bytes: 999, now: 2) + _ = meter.record(path: "clip", destination: 0, phase: .writing, bytes: 999, now: 3) + let uneven = meter.snapshot() + #expect(uneven.perDestination[0] > uneven.perDestination[1]) + #expect(uneven.perDestination[1] < 0.21) + #expect(uneven.fraction < 0.5) + } + + @Test func reuseValidationLeavesIndependentReadbackOutstanding() { + let meter = TransferWorkMeter(files: [entry("clip", 100)], destinationCount: 1, mode: .standard) + _ = meter.record(path: "clip", destination: 0, phase: .sourceValidatingExisting, bytes: 100, now: 1) + _ = meter.record(path: "clip", destination: 0, phase: .validatingExisting, bytes: 100, now: 2) + meter.copied(path: "clip", destination: 0) + #expect(abs(meter.snapshot().fraction - 2.0 / 3.0) < 0.0001) + _ = meter.record(path: "clip", destination: 0, phase: .readingBack, bytes: 50, now: 3) + #expect(meter.snapshot().fraction > 2.0 / 3.0) + #expect(meter.snapshot().fraction < 0.99) + } + + @Test func thoroughAndParanoidReserveRequiredSourceReads() { + let thorough = TransferWorkMeter(files: [entry("clip", 100)], destinationCount: 2, mode: .thorough) + #expect(thorough.snapshot().total == 600) // source + fresh source + two writes + two reads + let paranoid = TransferWorkMeter(files: [entry("clip", 100)], destinationCount: 2, mode: .paranoid) + #expect(paranoid.snapshot().total == 900) // source + two source compares + two writes + four reads + let quick = TransferWorkMeter(files: [entry("clip", 100)], destinationCount: 2, mode: .quick) + #expect(quick.snapshot().total == 300) + } + + @Test func tenThousandSmallFilesKeepExactIncrementalProgress() { + let count = 10_000 + let files = (0..(([], nil)) + let pipeline = TransferPipeline(fileSystem: LocalFileAccess(), checksum: ChecksumEngine.shared, + fanOutHooks: .init(sourceDidRead: { url, data in + if url.lastPathComponent == "clip.bin" { + let deadline = Date().addingTimeInterval(10) + while !released.withLock({ $0 }) { + guard Date() < deadline else { throw CocoaError(.fileReadUnknown) } + try await Task.sleep(for: .milliseconds(1)) + } + } + return data + })) + let operation = try await pipeline.performFileOperation(sourceURL: source, destinationURLs: [destination], + verificationMode: .standard, settings: CameraLabelSettings(), estimatedTotalBytes: nil, + progressCallback: { update in observations.withLock { $0.updates.append(update) } }, + onFileResult: { result in + if result.sourceURL.lastPathComponent == "sidecar.txt", result.verificationResult != nil || !result.success { + observations.withLock { $0.tiny = $0.updates.last?.overallProgress } + released.withLock { $0 = true } + } + }) + let observed = observations.withLock { $0 } + let tinyFraction = try #require(observed.tiny) + #expect(tinyFraction < 0.1) + #expect(operation.results.count == 2) + #expect(operation.results.allSatisfy { $0.success && $0.verificationResult?.matches == true }) + #expect(observed.updates.count >= 4) + #expect(observed.updates.last?.overallProgress == 1) + #expect(zip(observed.updates, observed.updates.dropFirst()).allSatisfy { $0.overallProgress <= $1.overallProgress }) + } + } + + @Test(arguments: [VerificationMode.standard, .thorough, .paranoid]) + func repeatedTransfersRemainMonotonicAndVerified(mode: VerificationMode) async throws { + try await FileOperationsTestLock.shared.run { + let root = FileManager.default.temporaryDirectory.appendingPathComponent("byte-reuse-\(UUID())") + let source = root.appendingPathComponent("source") + let destinations = [root.appendingPathComponent("one"), root.appendingPathComponent("two")] + for folder in [source] + destinations { try FileManager.default.createDirectory(at: folder, withIntermediateDirectories: true) } + defer { try? FileManager.default.removeItem(at: root) } + try Data([1]).write(to: source.appendingPathComponent("tiny")) + try Data(repeating: 9, count: 2 * 1024 * 1024).write(to: source.appendingPathComponent("large")) + let pipeline = TransferPipeline(fileSystem: LocalFileAccess(), checksum: ChecksumEngine.shared) + for attempt in 0..<2 { + let updates = Mutex<[OperationProgress]>([]) + let operation = try await pipeline.performFileOperation(sourceURL: source, destinationURLs: destinations, + verificationMode: mode, settings: CameraLabelSettings(), estimatedTotalBytes: 999_999_999, + progressCallback: { update in updates.withLock { $0.append(update) } }, onFileResult: nil) + let observed = updates.withLock { $0 } + #expect(observed.count >= 3) + let expectedBytes: Int64 = 2 * 1024 * 1024 + 1 + let selectedBytes = try #require(observed.last?.totalBytes) + #expect(selectedBytes == expectedBytes) + #expect(observed.last?.overallProgress == 1) + #expect(zip(observed, observed.dropFirst()).allSatisfy { $0.overallProgress <= $1.overallProgress }) + #expect(observed.dropLast().allSatisfy { $0.overallProgress < 1 }) + #expect(operation.results.count == 4) + #expect(operation.results.allSatisfy { $0.success && $0.verificationResult?.matches == true && $0.wasReused == (attempt == 1) }) + } + } + } + + @Test(arguments: [false, true]) + func interruptedWriteNeverPublishesCompletedProgress(cancel: Bool) async throws { + try await FileOperationsTestLock.shared.run { + let root = FileManager.default.temporaryDirectory.appendingPathComponent("byte-stop-\(UUID())") + let source = root.appendingPathComponent("source") + let destination = root.appendingPathComponent("backup") + for folder in [source, destination] { try FileManager.default.createDirectory(at: folder, withIntermediateDirectories: true) } + defer { try? FileManager.default.removeItem(at: root) } + try Data(repeating: 4, count: 8 * 1024 * 1024).write(to: source.appendingPathComponent("clip")) + let updates = Mutex<[OperationProgress]>([]) + let pipeline = TransferPipeline(fileSystem: LocalFileAccess(), checksum: ChecksumEngine.shared, + fanOutHooks: .init(beforeWrite: { _, _, chunk in + if chunk == 1 { + if cancel { throw CancellationError() } + throw CocoaError(.fileWriteOutOfSpace) + } + })) + do { + let operation = try await pipeline.performFileOperation(sourceURL: source, destinationURLs: [destination], + verificationMode: .standard, settings: CameraLabelSettings(), estimatedTotalBytes: nil, + progressCallback: { update in updates.withLock { $0.append(update) } }, onFileResult: nil) + #expect(!cancel) + #expect(operation.results.contains { !$0.success }) + } catch is CancellationError { + #expect(cancel) + } + let observed = updates.withLock { $0 } + #expect(observed.count >= 2) + #expect(observed.contains { $0.overallProgress > 0 }) + #expect(observed.allSatisfy { $0.overallProgress < 1 && $0.currentStage != .completed }) + #expect(zip(observed, observed.dropFirst()).allSatisfy { $0.overallProgress <= $1.overallProgress }) + } + } +} diff --git a/Packages/BitMatchEngine/Tests/BitMatchEngineTests/StreamingMemoryProbeTests.swift b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/StreamingMemoryProbeTests.swift index 58cf9908..b16b9b36 100644 --- a/Packages/BitMatchEngine/Tests/BitMatchEngineTests/StreamingMemoryProbeTests.swift +++ b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/StreamingMemoryProbeTests.swift @@ -96,10 +96,10 @@ final class StreamingMemoryProbeTests: XCTestCase { checksumService: ChecksumEngine.shared, preEnumeratedFiles: [clip], hooks: hooks, - onIOBytes: { index, phase, count, _, _ in + onIOBytes: { index, _, phase, count, _, _ in switch phase { case .writing: samples.wrote(index: index, count: count) - case .readingBack: break + case .readingBack, .validatingExisting, .sourceReading, .sourceVerifying, .sourceValidatingExisting: break } }, onProgress: { index, _, byteCount, wasReused in diff --git a/Shared/Core/Models/QueueAccessoryPresentation.swift b/Shared/Core/Models/QueueAccessoryPresentation.swift index 0f995b84..bfeb888b 100644 --- a/Shared/Core/Models/QueueAccessoryPresentation.swift +++ b/Shared/Core/Models/QueueAccessoryPresentation.swift @@ -75,13 +75,9 @@ struct QueueAccessoryPresentation: Equatable, Sendable { } if let running, let index = queue.rows.firstIndex(where: { $0.id == running.id }) { - let completed = queue.rows.filter(\.isFinished).count - let cardProgress = min(max(running.progressFraction ?? 0, 0), 0.99) - let overall = min( - max((Double(completed) + cardProgress) / Double(queue.rows.count), 0), - 0.99 - ) - var details = ["\(index + 1) of \(queue.rows.count)", percent(overall)] + let overall = weightedProgress(rows: queue.rows) + var details = ["\(index + 1) of \(queue.rows.count)"] + if let overall { details.append(percent(overall)) } if let timeRemaining { details.append("\(timeRemaining) left") } return Self( title: running.cardName, @@ -118,6 +114,16 @@ struct QueueAccessoryPresentation: Equatable, Sendable { ) } + static func weightedProgress(rows: [QueueSessionRow]) -> Double? { + guard !rows.isEmpty, rows.allSatisfy({ $0.estimatedWorkBytes.map { $0.isFinite && $0 > 0 } == true }) else { return nil } + let total = rows.reduce(0.0) { $0 + ($1.estimatedWorkBytes ?? 0) } + let done = rows.reduce(0.0) { sum, row in + let fraction = row.isFinished ? 1.0 : (row.isRunning ? min(0.99, max(0, row.progressFraction ?? 0)) : 0) + return sum + (row.estimatedWorkBytes ?? 0) * fraction + } + return min(0.99, max(0, done / total)) + } + private static func problemPresentation(_ row: QueueSessionRow) -> Self { let failure = row.safetyState == .failed return Self( diff --git a/Shared/Core/Models/QueueSessionPresentation.swift b/Shared/Core/Models/QueueSessionPresentation.swift index ad661c10..a40a743c 100644 --- a/Shared/Core/Models/QueueSessionPresentation.swift +++ b/Shared/Core/Models/QueueSessionPresentation.swift @@ -20,6 +20,8 @@ struct QueueSessionRow: Identifiable, Equatable, Sendable { let outcome: TransferOutcomePresentation? let destinationNames: [String] let verificationModeName: String + /// Progress estimate only; nil for old or unmeasured queue records. + var estimatedWorkBytes: Double? = nil var isEditable: Bool { safetyState == .waiting } var isRunning: Bool { safetyState.isActiveQueueState } @@ -217,7 +219,7 @@ struct QueueSessionPresentation: Equatable, Sendable { // A completed file-work stage is not a finished transfer. MHL, // report writing and journal settlement can still fail or stop. let activeFraction = progress.map { - min(max($0.stageProgress ?? $0.overallProgress, 0), 0.99) + min(max($0.overallProgress, 0), 0.99) } let percent = activeFraction.map { Int(($0 * 100).rounded(.down)) } switch progress?.currentStage { @@ -261,7 +263,7 @@ struct QueueSessionPresentation: Equatable, Sendable { destinations: destinationSummary(destinationNames), safetyState: state, progressFraction: record.state == .running ? progress.map { - min(max($0.stageProgress ?? $0.overallProgress, 0), 0.99) + min(max($0.overallProgress, 0), 0.99) } : nil, action: action, cause: cause, @@ -276,10 +278,20 @@ struct QueueSessionPresentation: Equatable, Sendable { ), outcome: outcome, destinationNames: destinationNames, - verificationModeName: record.verificationMode.rawValue + verificationModeName: record.verificationMode.rawValue, + estimatedWorkBytes: estimatedWorkBytes(for: record) ) } + static func estimatedWorkBytes(for record: LocalTransferRecord) -> Double? { + guard let bytes = record.estimatedSourceBytes, bytes >= 0 else { return nil } + let destinations = Double(record.destinations.count) + let destinationPasses = record.verificationMode == .quick ? 0.0 : (record.verificationMode == .paranoid ? 2.0 : 1.0) + let sourcePasses = record.verificationMode == .thorough ? 2.0 : (record.verificationMode == .paranoid ? 1 + destinations : 1.0) + let handoffPasses = record.generateASCMHL && record.verificationMode != .quick ? destinations : 0 + return max(1, Double(bytes)) * (sourcePasses + destinations * (1 + destinationPasses) + handoffPasses) + } + private static func finishedOutcome( for record: LocalTransferRecord, state: CardSafetyState diff --git a/Shared/Core/Models/TransferProgressPresentation.swift b/Shared/Core/Models/TransferProgressPresentation.swift index 5af33263..15c81ac6 100644 --- a/Shared/Core/Models/TransferProgressPresentation.swift +++ b/Shared/Core/Models/TransferProgressPresentation.swift @@ -156,7 +156,7 @@ struct TransferProgressPresentation: Equatable, Sendable { /// "A001 to 2 backups", or why the run is paused. let detail: String? let tone: ProgressTone - /// The engine's (copied + verified) / (files × stages), capped below 100% + /// The engine's selected-byte work budget, capped below 100% /// while the journal still owns an active attempt. Nil before the engine /// reports, so the bar is indeterminate rather than a false 0%. let fraction: Double? @@ -389,6 +389,10 @@ struct TransferProgressPresentation: Equatable, Sendable { state = .copying } } + if let measuredFraction = progress?.perDestinationProgress, + measuredFraction.count == destinations.count { + fraction = min(0.99, max(0, measuredFraction[index])) + } let name = destinationNames.flatMap { names in names.indices.contains(index) ? names[index] : nil } ?? DestinationVolumeLabel.name(for: url) diff --git a/Shared/Core/Services/CopyVerifyExecutor.swift b/Shared/Core/Services/CopyVerifyExecutor.swift index 66f07ae8..62420142 100644 --- a/Shared/Core/Services/CopyVerifyExecutor.swift +++ b/Shared/Core/Services/CopyVerifyExecutor.swift @@ -1,5 +1,6 @@ // CopyVerifyExecutor.swift - Handles copy and verify operation execution import Foundation +import Synchronization import BitMatchEngine struct PhotographerFinalizationResult { @@ -116,6 +117,10 @@ final class CopyVerifyExecutor { private var acceptsPipelineProgress = false private var lastDiagnosticStage: ProgressStage? private var destinationRoots: [URL] = [] + private var highestOverallProgress = 0.0 + private var pipelineWorkBytes = 0.0 + private var pipelineCompletedWorkBytes = 0.0 + private var reservedHandoffBytes = 0.0 private(set) var completedPhaseDurations = OperationPhaseDurations(copySeconds: nil, verifySeconds: nil) private(set) var completedPerformanceTelemetry = TransferPerformanceTelemetry() @@ -156,6 +161,10 @@ final class CopyVerifyExecutor { private func executeRun(config: CopyVerifyConfig, callbacks: CopyVerifyCallbacks) async throws -> FileOperation? { cancellationRequested = false + highestOverallProgress = 0 + pipelineWorkBytes = 0 + pipelineCompletedWorkBytes = 0 + reservedHandoffBytes = 0 diagnosticRun = config.operationId acceptsPipelineProgress = true lastDiagnosticStage = nil @@ -204,10 +213,18 @@ final class CopyVerifyExecutor { try checkCancellation() timingService.updateStage(.copying) + // Capture synchronously before dispatching UI work. The executor + // can resume before the MainActor Task for the final callback. + let lastPipelineProgress = Mutex(nil) let progressCallback: FileOperationsService.ProgressCallback = { [weak self] progressUpdate in + lastPipelineProgress.withLock { previous in + if previous.map({ progressUpdate.overallProgress >= $0.overallProgress }) ?? true { previous = progressUpdate } + } Task { @MainActor in guard let self, self.acceptsPipelineProgress, self.diagnosticRun == config.operationId else { return } - self.handleProgress(progressUpdate, callbacks: callbacks) + self.handleProgress(Self.pipelineProgress(progressUpdate, + destinationCount: config.destinationURLs.count, + includeMHL: config.generateASCMHL && config.verificationMode != .quick), callbacks: callbacks) } } let ioProgressCallback: FileOperationsService.IOProgressCallback = { [weak self] update in @@ -268,6 +285,15 @@ final class CopyVerifyExecutor { ) #endif + if let finalUpdate = lastPipelineProgress.withLock({ $0 }) { + pipelineWorkBytes = finalUpdate.totalWorkBytes ?? 0 + pipelineCompletedWorkBytes = min(pipelineWorkBytes, max(0, finalUpdate.workBytesProcessed ?? 0)) + reservedHandoffBytes = config.generateASCMHL && config.verificationMode != .quick + ? Double(max(0, finalUpdate.totalBytes ?? 0)) * Double(config.destinationURLs.count) : 0 + handleProgress(Self.pipelineProgress(finalUpdate, + destinationCount: config.destinationURLs.count, + includeMHL: config.generateASCMHL && config.verificationMode != .quick), callbacks: callbacks) + } acceptsPipelineProgress = false SharedLogger.transferEvent(.pipelineDrained, run: config.operationId) SharedLogger.correlatePipeline(run: config.operationId, pipeline: operation.id) @@ -294,6 +320,11 @@ final class CopyVerifyExecutor { // MARK: - Progress Handling private func handleProgress(_ progressUpdate: OperationProgress, callbacks: CopyVerifyCallbacks) { + // A phase boundary or delayed child callback may not rewind the bar. + // 100% belongs to the settled outcome, never this active executor. + let fraction = progressUpdate.overallProgress.isFinite ? progressUpdate.overallProgress : highestOverallProgress + highestOverallProgress = min(0.99, max(highestOverallProgress, fraction)) + let progressUpdate = progressUpdate.replacingOverallProgress(highestOverallProgress) callbacks.onProgress(progressUpdate) if lastDiagnosticStage != progressUpdate.currentStage { lastDiagnosticStage = progressUpdate.currentStage @@ -393,9 +424,9 @@ final class CopyVerifyExecutor { let reportIssue: String? if config.reportSettings.makeReport && !allResults.isEmpty { timingService.updateStage(.generating) - callbacks.onProgress(OperationProgress( - overallProgress: 0, currentFile: "Writing transfer reports…", - filesProcessed: 0, totalFiles: allResults.count, currentStage: .generating, speed: nil)) + handleProgress(OperationProgress( + overallProgress: highestOverallProgress, currentFile: "Writing transfer reports…", + filesProcessed: 0, totalFiles: allResults.count, currentStage: .generating, speed: nil), callbacks: callbacks) reportIssue = try await generateReport( operation: operation, results: allResults, @@ -468,12 +499,15 @@ final class CopyVerifyExecutor { let channel = AsyncStream.makeStream(bufferingPolicy: .bufferingNewest(1)) let consumer = Task { @MainActor in for await update in channel.stream { - self.handleProgress(Self.mhlProgress(update), callbacks: callbacks) + self.handleProgress(Self.mhlProgress(update, precedingWorkBytes: self.pipelineWorkBytes, + precedingCompletedWorkBytes: self.pipelineCompletedWorkBytes, reservedHandoffBytes: self.reservedHandoffBytes), callbacks: callbacks) } } - callbacks.onProgress(Self.mhlProgress(.init(bytesProcessed: 0, totalBytes: mhlBytes, + handleProgress(Self.mhlProgress(.init(bytesProcessed: 0, totalBytes: mhlBytes, filesProcessed: 0, totalFiles: plan.jobs.flatMap(\.files).count, - published: false))) + published: false), precedingWorkBytes: pipelineWorkBytes, + precedingCompletedWorkBytes: pipelineCompletedWorkBytes, + reservedHandoffBytes: reservedHandoffBytes), callbacks: callbacks) let sourceURL = config.sourceURL let startTime = operation.startTime let run = config.operationId @@ -504,11 +538,25 @@ final class CopyVerifyExecutor { } } - static func mhlProgress(_ update: ASCMHLGenerator.Progress) -> OperationProgress { + static func pipelineProgress(_ update: OperationProgress, destinationCount: Int, includeMHL: Bool) -> OperationProgress { + guard includeMHL, let total = update.totalWorkBytes, total > 0, + let done = update.workBytesProcessed, let sourceBytes = update.totalBytes else { + return update.replacingOverallProgress(min(0.99, update.overallProgress)) + } + // Every selected byte must either be independently read for ASC MHL or + // have its prior readback identity/digests validated. The generator + // settles reused evidence immediately; exFAT advances as it rereads. + let handoff = Double(max(0, sourceBytes)) * Double(max(0, destinationCount)) + return update.replacingOverallProgress(min(0.99, done / (total + handoff))) + } + + static func mhlProgress(_ update: ASCMHLGenerator.Progress, precedingWorkBytes: Double = 0, + precedingCompletedWorkBytes: Double? = nil, reservedHandoffBytes: Double? = nil) -> OperationProgress { let fraction = update.totalBytes > 0 ? min(1, Double(update.bytesProcessed) / Double(update.totalBytes)) : 0 return OperationProgress( - overallProgress: update.published ? 1 : min(0.99, fraction), + overallProgress: min(0.99, ((precedingCompletedWorkBytes ?? precedingWorkBytes) + Double(max(0, update.bytesProcessed))) + / max(1, precedingWorkBytes + (reservedHandoffBytes ?? Double(max(0, update.totalBytes))))), currentFile: update.published ? "ASC MHL records saved" : "Creating ASC MHL — checking destination media…", filesProcessed: update.filesProcessed, totalFiles: update.totalFiles, currentStage: .generating, speed: nil, elapsedTime: nil, averageSpeed: nil, peakSpeed: nil, diff --git a/Shared/Core/Services/LocalTransferJournal.swift b/Shared/Core/Services/LocalTransferJournal.swift index 454a23bd..f140d673 100644 --- a/Shared/Core/Services/LocalTransferJournal.swift +++ b/Shared/Core/Services/LocalTransferJournal.swift @@ -73,11 +73,11 @@ final class LocalTransferJournal: ObservableObject { @discardableResult func enqueue(sourceURL: URL, destinationURLs: [URL], verificationMode: VerificationMode, cameraSettings: CameraLabelSettings, reportSettings: ReportPrefs, generateASCMHL: Bool = true, - projectID: UUID? = nil, projectCardID: UUID? = nil) throws -> UUID { + projectID: UUID? = nil, projectCardID: UUID? = nil, estimatedSourceBytes: Int64? = nil) throws -> UUID { defer { refresh() } return try store.enqueue(sourceURL: sourceURL, destinationURLs: destinationURLs, verificationMode: verificationMode, cameraSettings: cameraSettings, reportSettings: reportSettings, - generateASCMHL: generateASCMHL, projectID: projectID, projectCardID: projectCardID) + generateASCMHL: generateASCMHL, projectID: projectID, projectCardID: projectCardID, estimatedSourceBytes: estimatedSourceBytes) } /// A retry is a new attempt, preserving the previous attempt and its evidence. @@ -117,10 +117,10 @@ final class LocalTransferJournal: ObservableObject { sourceURL: sourceURL, destinationURLs: destinationURLs, sessionRecordIDs: sessionRecordIDs) } - func recordPlan(id: UUID, fileCounts: [Int]) throws { + func recordPlan(id: UUID, fileCounts: [Int], selectedSourceBytes: Int64? = nil) throws { defer { refresh() } try beforeRecordPlan?(id) - try store.recordPlan(id: id, fileCounts: fileCounts) + try store.recordPlan(id: id, fileCounts: fileCounts, selectedSourceBytes: selectedSourceBytes) } func fail(id: UUID, summary: String) throws { @@ -140,7 +140,8 @@ final class LocalTransferJournal: ObservableObject { verificationMode: VerificationMode, cameraSettings: CameraLabelSettings, reportSettings: ReportPrefs, - generateASCMHL: Bool + generateASCMHL: Bool, + estimatedSourceBytes: Int64? = nil ) throws { defer { refresh() } try store.replaceQueued( @@ -150,7 +151,8 @@ final class LocalTransferJournal: ObservableObject { verificationMode: verificationMode, cameraSettings: cameraSettings, reportSettings: reportSettings, - generateASCMHL: generateASCMHL + generateASCMHL: generateASCMHL, + estimatedSourceBytes: estimatedSourceBytes ) } diff --git a/Shared/Core/Services/SharedAppCoordinator.swift b/Shared/Core/Services/SharedAppCoordinator.swift index f8d4ef5b..1b21aeb7 100644 --- a/Shared/Core/Services/SharedAppCoordinator.swift +++ b/Shared/Core/Services/SharedAppCoordinator.swift @@ -835,8 +835,8 @@ class SharedAppCoordinator: ObservableObject { presentation.setFileCountTotal(prog.totalFiles) let destinationCount = presentedDestinationCount ?? activeRunContext?.destinationURLs.count ?? destinationURLs.count // Time left is measured against the copy work actually planned: the - // scanned source once per backup. The engine's `totalBytes` covers - // one backup, and is a 1 GB guess when the source was not scanned. + // scanned source once per backup. The engine's `totalBytes` is the + // selected source manifest size, even when Setup had no scan yet. let plannedTotalBytes: Int64? if let context = activeRunContext { if context.cameraLabelSettings.excludedAppleDoublePaths?.isEmpty == false, @@ -1091,7 +1091,10 @@ class SharedAppCoordinator: ObservableObject { sourceURL: source, destinationURLs: destinations, verificationMode: verificationMode ?? self.verificationMode, cameraSettings: settings, reportSettings: reportSettings ?? self.reportSettings, - generateASCMHL: generateASCMHL ?? self.generateASCMHL + generateASCMHL: generateASCMHL ?? self.generateASCMHL, + estimatedSourceBytes: sourceFolderInfo.flatMap { + $0.url.standardizedFileURL == source.standardizedFileURL ? $0.totalSize : nil + } ) queueSessionEnded = false addQueueSessionRecord(id) @@ -1139,7 +1142,10 @@ class SharedAppCoordinator: ObservableObject { verificationMode: verificationMode, cameraSettings: cameraLabelSettings, reportSettings: reportSettings, - generateASCMHL: generateASCMHL + generateASCMHL: generateASCMHL, + estimatedSourceBytes: sourceFolderInfo.flatMap { + $0.url.standardizedFileURL == sourceURL.standardizedFileURL ? $0.totalSize : nil + } ) editingSetupTransferID = nil composerBeforeEditing = nil @@ -2032,7 +2038,7 @@ class SharedAppCoordinator: ObservableObject { self.progress = progressUpdate if !runResults.attemptedPlanSave, let counts = progressUpdate.perDestinationTotals { runResults.attemptedPlanSave = true - do { try self.transferJournal.recordPlan(id: recordID, fileCounts: counts) } + do { try self.transferJournal.recordPlan(id: recordID, fileCounts: counts, selectedSourceBytes: progressUpdate.totalBytes) } catch { runResults.planSaveError = error SharedLogger.warning("Could not save transfer plan: \(error)", category: .transfer) diff --git a/docs/TODO.md b/docs/TODO.md index 8e8324ef..017284c1 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -161,7 +161,7 @@ Remote scope excludes screen sharing, arbitrary filesystem browsing, remote shel - [ ] **Portable settings/workflow export and import.** Build on existing project recipes and saved settings. Include safe preferences, verification/report choices, naming recipes and reusable destination configurations; preview changes before import and revalidate destinations on the receiving machine. **Acceptance:** omit credentials, security-scoped bookmarks, private history and machine-specific identities. No facility configuration-management system. - [ ] **Simplified operator mode, only if users ask.** Investigate preconfigured destinations, naming and verification with an everyday source → start → review flow. Warnings stay visible and validation still runs; no facility administration or permission system. - [ ] **Organize and rename.** Extend existing project naming and folder recipes with remaining naming-pattern and organize-by-date workflows. Preserve original card contents and test destination collisions before adding new options. -- [ ] **Honest byte-weighted progress.** The engine currently counts finished file/stage units, and the queue bar averages cards, so a tiny sidecar or small card can move the percentage as much as a huge clip or card. Track measurable copy and required readback bytes per destination, including source rereads and ASC MHL destination reads when enabled; show an explicit finishing stage where remaining report work has no reliable byte total. Keep the main UI quiet and reserve 100% for the authoritative result. **Acceptance:** synthetic tiny-plus-huge files/cards, two destinations with different speeds, reuse/retry, optional MHL, failure and cancellation show monotonic, plausible progress without an early completed state. +- [ ] **Honest byte-weighted progress.** The shared implementation now budgets selected-file bytes for source reads, writes/reuse checks, required verification and optional ASC MHL work; live chunk callbacks advance large clips, and finishing reports never reset the bar or show active 100%. Destination bars use their own byte work. Queue cards use persisted size estimates, destination counts and verification mode; a queue with unknown sizes stays indeterminate. Estimates stay separate from copy evidence and remain stable per attempt. Focused synthetic tests cover tiny-plus-large files, uneven destinations, three verification modes, reuse, cancellation/failure, MHL mapping, duplicate planning entries and 10,000 small files; the tiny-file regression fails on the previous engine at 25%. **Remaining:** full integration/platform validation, signed-app walkthrough, and finer progress through extra source-hashing passes on reused files. The current retry budget remains deliberately logical rather than an exact tally of every physical read. Keep the main UI quiet and reserve 100% for the authoritative result. - [x] **Speed for each drive.** Expanded transfer details show each local destination's written and read-back bytes with separate live I/O rates. These are measured write/read calls, not a durability benchmark or verification verdict. The telemetry is shared with future storage-scheduling work; this change does not alter scheduling. - [x] **Reconcile missed Mac card events.** Mount notifications and manual rescan now reconcile the mounted filesystem's identity with published cards and drives. Deterministic event-gap tests cover unmount/reinsert, same-path replacement, stale asynchronous analysis and source selection; stale discovery cannot authorize auto-queue. This is a bounded event-triggered backstop, not a new polling service. A real removable-card walkthrough remains in the release gate below. - [ ] **Slow-cable hint.** On Mac, show a quiet hint when reliable connection information identifies a USB 2 bottleneck. Do not promise a speed multiplier without measurements.