From 3c755c77ec02063dfb01598d797bf597d6a3de41 Mon Sep 17 00:00:00 2001 From: Michael Cerisano Date: Sat, 17 Oct 2026 03:40:25 -0400 Subject: [PATCH 1/2] Show live write and readback speed per destination --- .../ProgressPresentationFeedTests.swift | 22 +++ .../TransferProgressPresentationTests.swift | 55 ++++++- .../BitMatchEngine/EngineProtocols.swift | 34 ++++ .../File/DestinationWriter.swift | 70 +++++--- .../Models/TransferIOProgress.swift | 128 +++++++++++++++ .../BitMatchEngine/TransferPipeline.swift | 40 ++++- .../BitMatchEngineTests/FanOutTests.swift | 151 ++++++++++++++++++ .../TransferIOMeterTests.swift | 54 +++++++ .../Models/TransferProgressPresentation.swift | 63 +++++++- Shared/Core/Services/CopyVerifyExecutor.swift | 24 +++ .../Core/Services/SharedAppCoordinator.swift | 16 ++ Shared/Core/ViewModels/LiveProgressFeed.swift | 3 + Shared/Views/Progress/ProgressScreen.swift | 33 +++- .../Views/TransferPresentationAdapters.swift | 2 + docs/TODO.md | 2 +- 15 files changed, 664 insertions(+), 33 deletions(-) create mode 100644 Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift create mode 100644 Packages/BitMatchEngine/Tests/BitMatchEngineTests/TransferIOMeterTests.swift diff --git a/BitMatchTests/ProgressPresentationFeedTests.swift b/BitMatchTests/ProgressPresentationFeedTests.swift index 20361706..08d3ad90 100644 --- a/BitMatchTests/ProgressPresentationFeedTests.swift +++ b/BitMatchTests/ProgressPresentationFeedTests.swift @@ -47,12 +47,19 @@ struct ProgressPresentationFeedTests { defer { folders.cleanup() } let coordinator = makeCoordinator(folders) coordinator.progress = progress(files: 2, of: 4, file: "B.ARW") + coordinator.liveProgress.ioProgress = TransferIOProgress(sequence: 1, destinations: [ + DestinationIOProgress(writtenBytes: 1, readbackBytes: 0, + writeBytesPerSecond: nil, readbackBytesPerSecond: nil, + lastWriteUptime: nil, lastReadbackUptime: nil) + ]) #expect(await waitUntil { coordinator.progressPresentation.fileCountTotal == 4 }) coordinator.resetForNewOperation() #expect(coordinator.progressPresentation.fileCountTotal == 0) #expect(coordinator.progressPresentation.currentFileName == nil) + #expect(coordinator.liveProgress.ioProgress == nil) + #expect(coordinator.liveProgress.rateFloorUptime == 0) } /// Plant: in `setupProgressPresentation`, delete @@ -69,4 +76,19 @@ struct ProgressPresentationFeedTests { coordinator.operationState = .cancelled #expect(!coordinator.progressPresentation.isTracking) } + + @Test func pauseInvalidatesTheLastPerDestinationRateSample() throws { + let folders = try CoordinatorFolders() + defer { folders.cleanup() } + let coordinator = makeCoordinator(folders) + coordinator.operationState = .inProgress + coordinator.operationState = .paused(PauseInfo( + pausedAt: Date(), currentFile: nil, filesProcessed: 0, totalFiles: 1, + bytesProcessed: 0, reason: .userRequested + )) + let pausedFloor = coordinator.liveProgress.rateFloorUptime + #expect(pausedFloor > 0) + coordinator.operationState = .resuming + #expect(coordinator.liveProgress.rateFloorUptime >= pausedFloor) + } } diff --git a/BitMatchTests/TransferProgressPresentationTests.swift b/BitMatchTests/TransferProgressPresentationTests.swift index 29d3bde0..eb2c8514 100644 --- a/BitMatchTests/TransferProgressPresentationTests.swift +++ b/BitMatchTests/TransferProgressPresentationTests.swift @@ -50,6 +50,7 @@ struct TransferProgressPresentationTests { state: OperationState, isRunning: Bool = true, progress: OperationProgress?, + ioProgress: TransferIOProgress? = nil, destinations: [URL]? = nil, speed: String? = "90 MB/s", timeRemaining: String? = "3 min" @@ -58,6 +59,7 @@ struct TransferProgressPresentationTests { state: state, isRunning: isRunning, progress: progress, + ioProgress: ioProgress, sourceName: "A001", destinations: destinations ?? [primary, secondary], speed: speed, @@ -128,6 +130,55 @@ struct TransferProgressPresentationTests { #expect(p.destinations.map(\.name) == ["Primary", "Secondary"]) } + @Test func destinationRowsKeepWriteAndReadbackMeasurementsSeparate() { + let sample = TransferIOProgress(sequence: 2, destinations: [ + DestinationIOProgress( + writtenBytes: 8_000_000, readbackBytes: 4_000_000, + writeBytesPerSecond: 16_000_000, readbackBytesPerSecond: 8_000_000, + lastWriteUptime: 100, lastReadbackUptime: 100), + DestinationIOProgress( + writtenBytes: 0, readbackBytes: 0, + writeBytesPerSecond: nil, readbackBytesPerSecond: nil, + lastWriteUptime: nil, lastReadbackUptime: nil) + ]) + let p = make(state: .inProgress, + progress: progress(stage: .copying, overall: 0.25, totals: [2, 2], completed: [1, 0]), + ioProgress: sample) + let first = p.destinations[0] + #expect(first.writeDetail(nowUptime: 100.5)?.contains("/s") == true) + #expect(first.readbackDetail(nowUptime: 100.5)?.contains("/s") == true) + #expect(first.writeDetail(nowUptime: 100.5) != first.readbackDetail(nowUptime: 100.5)) + #expect(first.writeDetail(nowUptime: 103)?.contains("/s") == false) + #expect(p.destinations[1].writeDetail(nowUptime: 100.5) == nil) + #expect(p.destinations[1].readbackDetail(nowUptime: 100.5) == nil) + + // Completed counts include failed files. A destination that has + // processed every file must not keep advertising a live write rate. + let processed = make( + state: .inProgress, + progress: progress(stage: .copying, overall: 0.5, totals: [1, 1], completed: [1, 1]), + ioProgress: sample + ) + #expect(processed.destinations[0].stateLabel == "Processed") + #expect(processed.destinations[0].writeDetail(nowUptime: 100.5)?.contains("/s") == false) + + let paused = TransferProgressPresentation.rows( + destinations: [primary, secondary], progress: nil, + ioProgress: sample, phase: .paused + ) + #expect(paused[0].writeDetail(nowUptime: 100.5)?.contains("/s") == false) + let resumed = TransferProgressPresentation.rows( + destinations: [primary, secondary], progress: nil, + ioProgress: sample, minimumRateUptime: 101, phase: .copying + ) + #expect(resumed[0].writeDetail(nowUptime: 101.5)?.contains("/s") == false) + let report = TransferProgressPresentation.rows( + destinations: [primary, secondary], progress: nil, + ioProgress: sample, phase: .writingReports + ) + #expect(report[0].io == nil) + } + @Test func destinationRowsReusePrecomputedVolumeNames() { let internalFolder = URL(fileURLWithPath: "/Users/mike/Backups/Project", isDirectory: true) let rows = TransferProgressPresentation.rows( @@ -162,7 +213,7 @@ struct TransferProgressPresentationTests { #expect(p.destinations.first?.helpPath == folder.path) } - /// Audit C1: a fully copied backup is "Copied", then "Verifying", and + /// Audit C1: a fully processed backup is "Processed", then "Verifying", and /// never "Done" or a verdict while the run is still going. /// Plant: in `TransferProgressPresentation.rows`, replace /// `state = phase == .verifying ? .verifying : .copied` with `state = .copied`. @@ -176,7 +227,7 @@ struct TransferProgressPresentationTests { progress: progress(stage: .verifying, overall: 0.75, totals: [4, 4], completed: [4, 4]) ) - #expect(copying.destinations.allSatisfy { $0.state == .copied && $0.stateLabel == "Copied" }) + #expect(copying.destinations.allSatisfy { $0.state == .copied && $0.stateLabel == "Processed" }) #expect(verifying.destinations.allSatisfy { $0.state == .verifying && $0.stateLabel == "Verifying" }) #expect(verifying.destinations.allSatisfy { !$0.symbol.contains("checkmark") }) } diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/EngineProtocols.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/EngineProtocols.swift index f4384d68..cf6bf0f2 100644 --- a/Packages/BitMatchEngine/Sources/BitMatchEngine/EngineProtocols.swift +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/EngineProtocols.swift @@ -29,6 +29,7 @@ public protocol ChecksumService: Sendable { // MARK: - File Operations Service Protocol public protocol FileOperationsService: Sendable { typealias ProgressCallback = @Sendable (OperationProgress) -> Void + typealias IOProgressCallback = @Sendable (TransferIOProgress) -> Void typealias FileResultCallback = @Sendable (FileOperationResult) async -> Void func performFileOperation( @@ -40,12 +41,45 @@ public protocol FileOperationsService: Sendable { progressCallback: @escaping ProgressCallback, onFileResult: FileResultCallback? ) async throws -> FileOperation + + /// Optional live telemetry channel. Existing service implementations can + /// keep the original operation method; telemetry never affects the result. + func performFileOperation( + sourceURL: URL, + destinationURLs: [URL], + verificationMode: VerificationMode, + settings: CameraLabelSettings, + estimatedTotalBytes: Int64?, + progressCallback: @escaping ProgressCallback, + ioProgressCallback: @escaping IOProgressCallback, + onFileResult: FileResultCallback? + ) async throws -> FileOperation func cancelOperation() func pauseOperation() async func resumeOperation() async } +public extension FileOperationsService { + func performFileOperation( + sourceURL: URL, + destinationURLs: [URL], + verificationMode: VerificationMode, + settings: CameraLabelSettings, + estimatedTotalBytes: Int64?, + progressCallback: @escaping ProgressCallback, + ioProgressCallback: @escaping IOProgressCallback, + onFileResult: FileResultCallback? + ) async throws -> FileOperation { + try await performFileOperation( + sourceURL: sourceURL, destinationURLs: destinationURLs, + verificationMode: verificationMode, settings: settings, + estimatedTotalBytes: estimatedTotalBytes, + progressCallback: progressCallback, onFileResult: onFileResult + ) + } +} + // MARK: - Shared Result Types public struct FileOperation: Sendable { diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift index c580f67b..b9b6af93 100644 --- a/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift @@ -404,6 +404,7 @@ public final class PinnedDestinationFile: @unchecked Sendable { #endif public final class DestinationWriter { + typealias IOByteCallback = @Sendable (Int, TransferIOPhase, Int, TimeInterval) -> Void struct SourceReadEvidence: Sendable { let digests: VerifiedDigests let identity: VerifiedFileIdentity @@ -533,6 +534,7 @@ public final class DestinationWriter { preEnumeratedFiles: [URL], pauseCheck: (@Sendable () async throws -> Void)? = nil, hooks: FanOutHooks? = nil, + onIOBytes: IOByteCallback? = nil, onSourceReadEvidence: @escaping @Sendable (String, SourceReadEvidence) async -> Void = { _, _ in }, onProgress: @escaping @Sendable (Int, String, Int64, Bool) async -> Void, onError: @escaping @Sendable (Int, String, Error) async -> Void @@ -605,6 +607,7 @@ public final class DestinationWriter { checksumService: checksumService, pauseCheck: pauseCheck, hooks: hooks, + onIOBytes: onIOBytes, onSourceReadEvidence: onSourceReadEvidence, onProgress: onProgress, onError: onError @@ -702,6 +705,7 @@ public final class DestinationWriter { checksumService: any ChecksumService, pauseCheck: (@Sendable () async throws -> Void)?, hooks: FanOutHooks?, + onIOBytes: IOByteCallback?, onSourceReadEvidence: @escaping @Sendable (String, SourceReadEvidence) async -> Void, onProgress: @escaping @Sendable (Int, String, Int64, Bool) async -> Void, onError: @escaping @Sendable (Int, String, Error) async -> Void @@ -734,7 +738,8 @@ public final class DestinationWriter { checksumService: checksumService, destinationIndex: destination.index, relativePath: relativePath, - hooks: hooks + hooks: hooks, + onIOBytes: onIOBytes ) pendingReuse.append(destination) } else { @@ -828,7 +833,10 @@ public final class DestinationWriter { file.destination.root.logicalRootURL, currentChunkIndex ) + let writeStarted = ProcessInfo.processInfo.systemUptime try writeAll(chunkData, to: file.temporaryFD) + onIOBytes?(file.destination.index, .writing, chunkData.count, + ProcessInfo.processInfo.systemUptime - writeStarted) return FanOutWriteResult(file: file, error: nil) } catch { return FanOutWriteResult(file: file, error: error) @@ -957,7 +965,8 @@ public final class DestinationWriter { try await validateReusableExistingDestinationFile( source: source, destination: existing, sourceSize: sourceSize, verificationMode: verificationMode, checksumService: checksumService, - destinationIndex: file.destination.index, relativePath: relativePath, hooks: hooks) + destinationIndex: file.destination.index, relativePath: relativePath, + hooks: hooks, onIOBytes: onIOBytes) file.cleanup() await onProgress(file.destination.index, relativePath, sourceSize, true) continue @@ -1084,7 +1093,8 @@ public final class DestinationWriter { checksumService: any ChecksumService, destinationIndex: Int, relativePath: String, - hooks: FanOutHooks? + hooks: FanOutHooks?, + onIOBytes: IOByteCallback? ) async throws { let destinationInfo = try destination.snapshot() guard Int64(destinationInfo.st_size) == sourceSize else { @@ -1105,7 +1115,8 @@ public final class DestinationWriter { pinnedDestination: destination, destinationIndex: destinationIndex, relativePath: relativePath, - hooks: hooks + hooks: hooks, + onIOBytes: onIOBytes ) else { throw reuseConflict("Existing destination file bytes differ; refusing to overwrite it", relativePath: relativePath) } @@ -1118,7 +1129,8 @@ public final class DestinationWriter { checksumService: checksumService, destinationIndex: destinationIndex, relativePath: relativePath, - hooks: hooks + hooks: hooks, + onIOBytes: onIOBytes ) else { throw reuseConflict("Existing destination file checksum differs; refusing to overwrite it", relativePath: relativePath) } @@ -1165,6 +1177,7 @@ public final class DestinationWriter { sourceReadEvidence: SourceReadEvidence? = nil, destinationIndex: Int = -1, hooks: FanOutHooks? = nil, + onIOBytes: IOByteCallback? = nil, inspection: @Sendable (PinnedDestinationFile) throws -> ClipIntegrityFinding = { try $0.inspectClipIntegrity() } @@ -1179,6 +1192,7 @@ public final class DestinationWriter { sourceReadEvidence: sourceReadEvidence, destinationIndex: destinationIndex, hooks: hooks, + onIOBytes: onIOBytes, inspection: inspection ) } @@ -1193,6 +1207,7 @@ public final class DestinationWriter { sourceReadEvidence: SourceReadEvidence? = nil, destinationIndex: Int = -1, hooks: FanOutHooks? = nil, + onIOBytes: IOByteCallback? = nil, inspection: @Sendable (PinnedDestinationFile) throws -> ClipIntegrityFinding = { try $0.inspectClipIntegrity() } @@ -1214,7 +1229,8 @@ public final class DestinationWriter { pinnedDestination: destination, destinationIndex: destinationIndex, relativePath: relativePath, - hooks: hooks + hooks: hooks, + onIOBytes: onIOBytes ) // Paranoid mode still byte-compares through the pinned handle, // and also computes a real SHA-256 digest (source via the stable @@ -1229,7 +1245,8 @@ public final class DestinationWriter { sourceReadEvidence: sourceReadEvidence, destinationIndex: destinationIndex, relativePath: relativePath, - hooks: hooks + hooks: hooks, + onIOBytes: onIOBytes ) verification = VerificationResult( sourceChecksum: digest.sourceChecksum, @@ -1251,7 +1268,8 @@ public final class DestinationWriter { sourceReadEvidence: sourceReadEvidence, destinationIndex: destinationIndex, relativePath: relativePath, - hooks: hooks + hooks: hooks, + onIOBytes: onIOBytes ) } @@ -1287,7 +1305,8 @@ public final class DestinationWriter { checksumService: any ChecksumService, destinationIndex: Int, relativePath: String, - hooks: FanOutHooks? + hooks: FanOutHooks?, + onIOBytes: IOByteCallback? ) async throws -> Bool { let types = verificationMode.checksumTypes guard !types.isEmpty else { return false } @@ -1299,7 +1318,8 @@ public final class DestinationWriter { sourceReadEvidence: nil, destinationIndex: destinationIndex, relativePath: relativePath, - hooks: hooks + hooks: hooks, + onIOBytes: onIOBytes ).matches } @@ -1315,7 +1335,8 @@ public final class DestinationWriter { sourceReadEvidence: SourceReadEvidence?, destinationIndex: Int, relativePath: String, - hooks: FanOutHooks? + hooks: FanOutHooks?, + onIOBytes: IOByteCallback? ) async throws -> VerificationResult { let startTime = Date() var sourceValues: [ChecksumAlgorithm: String] = [:] @@ -1353,7 +1374,8 @@ public final class DestinationWriter { algorithms: Array(Set(requiredTypes + [.md5])), destinationIndex: destinationIndex, relativePath: relativePath, - hooks: hooks + hooks: hooks, + onIOBytes: onIOBytes ) let primary = requiredTypes.contains(.sha256) ? ChecksumAlgorithm.sha256 : requiredTypes[0] let matches = requiredTypes.allSatisfy { type in @@ -1418,14 +1440,16 @@ public final class DestinationWriter { algorithms: [ChecksumAlgorithm], destinationIndex: Int, relativePath: String, - hooks: FanOutHooks? + hooks: FanOutHooks?, + onIOBytes: IOByteCallback? ) async throws -> PinnedDigestRead { var hasher = MultiDigestHasher(algorithms: algorithms) let identity = try await readPinnedDestination( destination, destinationIndex: destinationIndex, relativePath: relativePath, - hooks: hooks + hooks: hooks, + onIOBytes: onIOBytes ) { hasher.update($0) } let digests = hasher.finalize() if let ordinal = TransferDiagnostics.verifyOrdinal { @@ -1493,6 +1517,7 @@ public final class DestinationWriter { destinationIndex: Int, relativePath: String, hooks: FanOutHooks?, + onIOBytes: IOByteCallback?, consume: (Data) -> Void ) async throws -> VerifiedFileIdentity { let initial = try destination.snapshot() @@ -1512,13 +1537,16 @@ public final class DestinationWriter { try await PauseGate.waitIfCurrentIsPaused() // Bound temporary Foundation allocations to this chunk, even when // the pause gate returns without suspending the task. - let count = try autoreleasepool { () throws -> Int in + let (count, readSeconds) = try autoreleasepool { () throws -> (Int, TimeInterval) in + let readStarted = ProcessInfo.processInfo.systemUptime let data = try handle.read(upToCount: 4 * 1024 * 1024) ?? Data() + let readSeconds = ProcessInfo.processInfo.systemUptime - readStarted if !data.isEmpty { consume(data) } - return data.count + return (data.count, readSeconds) } if count == 0 { break } hooks?.destinationDidRead?(destinationIndex, relativePath, count) + onIOBytes?(destinationIndex, .readingBack, count, readSeconds) bytesRead += Int64(count) if ordinal != nil, bytesRead >= nextBreadcrumb { nextBreadcrumb += verifyReadBreadcrumbInterval @@ -1542,7 +1570,8 @@ public final class DestinationWriter { pinnedDestination: PinnedDestinationFile, destinationIndex: Int = -1, relativePath: String = "", - hooks: FanOutHooks? = nil + hooks: FanOutHooks? = nil, + onIOBytes: IOByteCallback? = nil ) async throws -> Bool { let sourceAttributes = try FileManager.default.attributesOfItem(atPath: source.path) let sourceSize = (sourceAttributes[.size] as? NSNumber)?.int64Value ?? -1 @@ -1566,18 +1595,21 @@ public final class DestinationWriter { try Task.checkCancellation() try await PauseGate.waitIfCurrentIsPaused() // Drain each chunk's buffers; see readPinnedDestination. - let (sourceCount, destinationCount, same) = try autoreleasepool { () throws -> (Int, Int, Bool) in + let (sourceCount, destinationCount, same, readSeconds) = try autoreleasepool { () throws -> (Int, Int, Bool, TimeInterval) in let sourceRead = try sourceHandle.read(upToCount: 4 * 1024 * 1024) ?? Data() let sourceData = try hooks?.transformVerificationSourceRead?(relativePath, sourceRead) ?? sourceRead guard sourceData.count == sourceRead.count else { throw sourceChangedError() } + let readStarted = ProcessInfo.processInfo.systemUptime let destinationData = try destinationHandle.read(upToCount: 4 * 1024 * 1024) ?? Data() - return (sourceData.count, destinationData.count, sourceData == destinationData) + let readSeconds = ProcessInfo.processInfo.systemUptime - readStarted + return (sourceData.count, destinationData.count, sourceData == destinationData, readSeconds) } guard sourceCount > 0, destinationCount > 0 else { throw NSError(domain: "DestinationWriter", code: -11, userInfo: [NSLocalizedDescriptionKey: "File changed while comparing bytes"]) } hooks?.verificationSourceDidRead?(relativePath, sourceCount) hooks?.destinationDidRead?(destinationIndex, relativePath, destinationCount) + onIOBytes?(destinationIndex, .readingBack, destinationCount, readSeconds) if !same { return false } bytesRead += Int64(sourceCount) } diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift new file mode 100644 index 00000000..e3997b2d --- /dev/null +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift @@ -0,0 +1,128 @@ +import Foundation +import Synchronization + +/// Live measurements only. These numbers are never verification evidence or a +/// destination-health verdict. Indices match the selected destinations. +public struct DestinationIOProgress: Equatable, Sendable { + public let writtenBytes: Int64 + public let readbackBytes: Int64 + public let writeBytesPerSecond: Double? + public let readbackBytesPerSecond: Double? + public let lastWriteUptime: TimeInterval? + public let lastReadbackUptime: TimeInterval? + + public init(writtenBytes: Int64, readbackBytes: Int64, + writeBytesPerSecond: Double?, readbackBytesPerSecond: Double?, + lastWriteUptime: TimeInterval?, lastReadbackUptime: TimeInterval?) { + self.writtenBytes = writtenBytes + self.readbackBytes = readbackBytes + self.writeBytesPerSecond = writeBytesPerSecond + self.readbackBytesPerSecond = readbackBytesPerSecond + self.lastWriteUptime = lastWriteUptime + self.lastReadbackUptime = lastReadbackUptime + } +} + +public struct TransferIOProgress: Equatable, Sendable { + /// Lets an asynchronous UI consumer discard an older callback. + public let sequence: UInt64 + public let destinations: [DestinationIOProgress] + + public init(sequence: UInt64, destinations: [DestinationIOProgress]) { + self.sequence = sequence + self.destinations = destinations + } +} + +enum TransferIOPhase: Sendable { + case writing + case readingBack +} + +/// Counts bytes only after an actual destination I/O call succeeds. A short +/// rolling window measures each destination's I/O service time, not the whole +/// fan-out wall time: the writer waits for its slowest destination after each +/// chunk, so dividing every destination's bytes by the same wall time would +/// misleadingly give fast and slow destinations the same rate. +final class TransferIOMeter: Sendable { + private struct Sample { + let at: TimeInterval + let bytes: Int64 + let seconds: TimeInterval + } + + private struct PhaseState { + var bytes: Int64 = 0 + var lastUptime: TimeInterval? + var samples: [Sample] = [] + + mutating func add(bytes newBytes: Int, duration: TimeInterval, now: TimeInterval) { + let (sum, overflow) = bytes.addingReportingOverflow(Int64(newBytes)) + bytes = overflow ? Int64.max : sum + lastUptime = now + samples.append(Sample(at: now, bytes: Int64(newBytes), seconds: max(0, duration))) + samples.removeAll { now - $0.at > 3 } + if samples.count > 64 { samples.removeFirst(samples.count - 64) } + } + + func rate(now: TimeInterval) -> Double? { + let recent = samples.filter { now - $0.at <= 3 } + let seconds = recent.reduce(0) { $0 + $1.seconds } + guard recent.count >= 2, seconds >= 0.1 else { return nil } + return Double(recent.reduce(Int64(0)) { $0 + $1.bytes }) / seconds + } + } + + private struct DestinationState { + var writing = PhaseState() + var readingBack = PhaseState() + } + + private struct State { + var destinations: [DestinationState] + var sequence: UInt64 = 0 + var lastPublishUptime: TimeInterval? + } + + private let state: Mutex + + init(destinationCount: Int) { + state = Mutex(State(destinations: Array(repeating: DestinationState(), count: max(0, destinationCount)))) + } + + /// Returns at most one UI snapshot per 250 ms, plus the first event. + func record(destination: Int, phase: TransferIOPhase, bytes: Int, + duration: TimeInterval, now: TimeInterval = ProcessInfo.processInfo.systemUptime) -> TransferIOProgress? { + guard bytes > 0, duration >= 0, duration.isFinite, now.isFinite else { return nil } + return state.withLock { state in + guard state.destinations.indices.contains(destination) else { return nil } + switch phase { + case .writing: + state.destinations[destination].writing.add(bytes: bytes, duration: duration, now: now) + case .readingBack: + state.destinations[destination].readingBack.add(bytes: bytes, duration: duration, now: now) + } + guard state.lastPublishUptime.map({ now - $0 >= 0.25 }) ?? true else { return nil } + state.lastPublishUptime = now + return Self.snapshot(&state, now: now) + } + } + + func finalSnapshot(now: TimeInterval = ProcessInfo.processInfo.systemUptime) -> TransferIOProgress { + state.withLock { state in Self.snapshot(&state, now: now) } + } + + private static func snapshot(_ state: inout State, now: TimeInterval) -> TransferIOProgress { + state.sequence &+= 1 + return TransferIOProgress(sequence: state.sequence, destinations: state.destinations.map { destination in + DestinationIOProgress( + writtenBytes: destination.writing.bytes, + readbackBytes: destination.readingBack.bytes, + writeBytesPerSecond: destination.writing.rate(now: now), + readbackBytesPerSecond: destination.readingBack.rate(now: now), + lastWriteUptime: destination.writing.lastUptime, + lastReadbackUptime: destination.readingBack.lastUptime + ) + }) + } +} diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift index 7aad3d7d..240cf344 100644 --- a/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift @@ -319,6 +319,25 @@ public final class TransferPipeline: FileOperationsService, Sendable { estimatedTotalBytes: Int64? = nil, progressCallback: @escaping ProgressCallback, onFileResult: FileResultCallback? + ) async throws -> FileOperation { + try await performFileOperation( + sourceURL: sourceURL, destinationURLs: destinationURLs, + verificationMode: verificationMode, settings: settings, + estimatedTotalBytes: estimatedTotalBytes, + progressCallback: progressCallback, ioProgressCallback: { _ in }, + onFileResult: onFileResult + ) + } + + public func performFileOperation( + sourceURL: URL, + destinationURLs: [URL], + verificationMode: VerificationMode, + settings: CameraLabelSettings, + estimatedTotalBytes: Int64?, + progressCallback: @escaping ProgressCallback, + ioProgressCallback: @escaping IOProgressCallback, + onFileResult: FileResultCallback? ) async throws -> FileOperation { let operationID = UUID() guard activeOperations.reserve(operationID) else { @@ -347,7 +366,8 @@ public final class TransferPipeline: FileOperationsService, Sendable { // tasks it starts, and nothing outside it (I9). let operationTask = Task { [pauseGate] in try await PauseGate.$current.withValue(pauseGate) { - try await executeOperation(operation, progressCallback: progressCallback, onFileResult: onFileResult) + try await executeOperation(operation, progressCallback: progressCallback, + ioProgressCallback: ioProgressCallback, onFileResult: onFileResult) } } activeOperations.attach(operationTask, to: operationID) @@ -371,6 +391,7 @@ public final class TransferPipeline: FileOperationsService, Sendable { settings: CameraLabelSettings, estimatedTotalBytes: Int64?, progressCallback: @escaping ProgressCallback, + ioProgressCallback: @escaping IOProgressCallback = { _ in }, onFileResult: FileResultCallback?, fanOutHooks: DestinationWriter.FanOutHooks? ) async throws -> FileOperation { @@ -388,6 +409,7 @@ public final class TransferPipeline: FileOperationsService, Sendable { settings: settings, estimatedTotalBytes: estimatedTotalBytes, progressCallback: progressCallback, + ioProgressCallback: ioProgressCallback, onFileResult: onFileResult ) } @@ -414,6 +436,7 @@ public final class TransferPipeline: FileOperationsService, Sendable { private func executeOperation( _ operation: FileOperation, progressCallback: @escaping ProgressCallback, + ioProgressCallback: @escaping IOProgressCallback, onFileResult: FileResultCallback? ) async throws -> FileOperation { @@ -498,6 +521,14 @@ public final class TransferPipeline: FileOperationsService, Sendable { 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 in + if let snapshot = ioMeter.record(destination: index, phase: phase, + bytes: bytes, duration: duration) { + 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() { @@ -596,7 +627,8 @@ public final class TransferPipeline: FileOperationsService, Sendable { clipURL: job.destination, sourceReadEvidence: job.sourceReadEvidence, destinationIndex: job.destinationIndex, - hooks: self.fanOutHooks + hooks: self.fanOutHooks, + onIOBytes: onIOBytes ) } } @@ -767,6 +799,7 @@ public final class TransferPipeline: FileOperationsService, Sendable { preEnumeratedFiles: sourceFileURLs, pauseCheck: { try await pauseGate.wait() }, hooks: fanOutHooks, + onIOBytes: onIOBytes, onSourceReadEvidence: { relativePath, evidence in await sourceReadEvidence.record(evidence, for: relativePath) }, @@ -880,7 +913,8 @@ public final class TransferPipeline: FileOperationsService, Sendable { clipURL: destinationFileURL, sourceReadEvidence: await sourceReadEvidence.evidence(for: relativePath), destinationIndex: destIndex, - hooks: fanOutHooks + hooks: fanOutHooks, + onIOBytes: onIOBytes ) } } diff --git a/Packages/BitMatchEngine/Tests/BitMatchEngineTests/FanOutTests.swift b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/FanOutTests.swift index 5f8a0aa8..ddeedc95 100644 --- a/Packages/BitMatchEngine/Tests/BitMatchEngineTests/FanOutTests.swift +++ b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/FanOutTests.swift @@ -8,6 +8,157 @@ import Darwin @testable import BitMatchEngine struct FanOutTests { + @Test + func liveIOCountsRealWritesAndReadbackWithoutChangingTheVerdict() async throws { + try await FileOperationsTestLock.shared.run { + #if os(macOS) + let fixture = try FanOutFixture(name: "live-io") + defer { fixture.cleanup() } + let bytes = Data(repeating: 0x42, count: 4 * 1024 * 1024 + 37) + try fixture.write(bytes, to: "clip.bin") + let updates = Mutex<[TransferIOProgress]>([]) + let settings = CameraLabelSettings() + let pipeline = fixture.pipeline(hooks: .init()) + let operation = try await pipeline.performFileOperation( + sourceURL: fixture.source, destinationURLs: fixture.destinations, + verificationMode: .standard, settings: settings, + estimatedTotalBytes: Int64(bytes.count), + progressCallback: { _ in }, + ioProgressCallback: { update in updates.withLock { $0.append(update) } }, + onFileResult: nil + ) + let snapshots = updates.withLock { $0 } + let latest = try #require(snapshots.max(by: { $0.sequence < $1.sequence })) + #expect(latest.destinations.count == 2) + #expect(latest.destinations.allSatisfy { + $0.writtenBytes == Int64(bytes.count) && $0.readbackBytes == Int64(bytes.count) + }) + #expect(operation.results.count == 2) + #expect(operation.results.allSatisfy { $0.success }) + #expect(try Data(contentsOf: fixture.source.appendingPathComponent("clip.bin")) == bytes) + + // Reuse independently reads the existing destination, but writes + // no new media bytes and must not inflate the copy-speed line. + let reusedUpdates = Mutex<[TransferIOProgress]>([]) + let reused = try await fixture.pipeline(hooks: .init()).performFileOperation( + sourceURL: fixture.source, destinationURLs: fixture.destinations, + verificationMode: .standard, settings: settings, + estimatedTotalBytes: Int64(bytes.count), + progressCallback: { _ in }, + ioProgressCallback: { update in reusedUpdates.withLock { $0.append(update) } }, + onFileResult: nil + ) + let reusedSnapshots = reusedUpdates.withLock { $0 } + let reusedLatest = try #require(reusedSnapshots.max(by: { $0.sequence < $1.sequence })) + #expect(reusedLatest.destinations.allSatisfy { + $0.writtenBytes == 0 && $0.readbackBytes >= Int64(bytes.count) + }) + #expect(reused.results.count == 2) + #expect(reused.results.allSatisfy { $0.success }) + #else + #expect(true) + #endif + } + } + + @Test + func quickCopyReportsWritesButNoReadback() async throws { + try await FileOperationsTestLock.shared.run { + #if os(macOS) + let fixture = try FanOutFixture(name: "quick-live-io") + defer { fixture.cleanup() } + let bytes = Data(repeating: 0x33, count: 4 * 1024 * 1024 + 1) + try fixture.write(bytes, to: "clip.bin") + let updates = Mutex<[TransferIOProgress]>([]) + _ = try await fixture.pipeline(hooks: .init()).performFileOperation( + sourceURL: fixture.source, destinationURLs: fixture.destinations, + verificationMode: .quick, settings: CameraLabelSettings(), + estimatedTotalBytes: Int64(bytes.count), + progressCallback: { _ in }, + ioProgressCallback: { update in updates.withLock { $0.append(update) } }, + onFileResult: nil + ) + let snapshots = updates.withLock { $0 } + let latest = try #require(snapshots.max(by: { $0.sequence < $1.sequence })) + #expect(latest.destinations.allSatisfy { + $0.writtenBytes == Int64(bytes.count) && $0.readbackBytes == 0 + }) + #else + #expect(true) + #endif + } + } + + @Test + func paranoidReportsBothPhysicalDestinationReadPasses() async throws { + try await FileOperationsTestLock.shared.run { + #if os(macOS) + let fixture = try FanOutFixture(name: "paranoid-live-io") + defer { fixture.cleanup() } + let bytes = Data(repeating: 0x54, count: 16_401) + try fixture.write(bytes, to: "clip.bin") + let updates = Mutex<[TransferIOProgress]>([]) + let operation = try await fixture.pipeline(hooks: .init()).performFileOperation( + sourceURL: fixture.source, destinationURLs: fixture.destinations, + verificationMode: .paranoid, settings: CameraLabelSettings(), + estimatedTotalBytes: Int64(bytes.count), + progressCallback: { _ in }, + ioProgressCallback: { update in updates.withLock { $0.append(update) } }, + onFileResult: nil + ) + let snapshots = updates.withLock { $0 } + let latest = try #require(snapshots.max(by: { $0.sequence < $1.sequence })) + #expect(latest.destinations.allSatisfy { + $0.writtenBytes == Int64(bytes.count) + && $0.readbackBytes == Int64(bytes.count * 2) + }) + #expect(operation.results.allSatisfy { $0.success }) + #else + #expect(true) + #endif + } + } + + @Test + func failedPartialWriteReportsOnlySuccessfulIOAndNoCompletedResult() async throws { + try await FileOperationsTestLock.shared.run { + #if os(macOS) + let fixture = try FanOutFixture(name: "partial-live-io") + defer { fixture.cleanup() } + let size = 4 * 1024 * 1024 + 1 + let sourceBytes = Data(repeating: 0x77, count: size) + try fixture.write(sourceBytes, to: "clip.bin") + let updates = Mutex<[TransferIOProgress]>([]) + let hooks = DestinationWriter.FanOutHooks(beforeWrite: { destination, _, chunk in + if destination == 1 && chunk == 1 { + throw NSError(domain: NSPOSIXErrorDomain, code: Int(EIO)) + } + }) + let operation = try await fixture.pipeline(hooks: hooks).performFileOperation( + sourceURL: fixture.source, destinationURLs: fixture.destinations, + verificationMode: .standard, settings: CameraLabelSettings(), + estimatedTotalBytes: Int64(size), + progressCallback: { _ in }, + ioProgressCallback: { update in updates.withLock { $0.append(update) } }, + onFileResult: nil + ) + let snapshots = updates.withLock { $0 } + let latest = try #require(snapshots.max(by: { $0.sequence < $1.sequence })) + #expect(latest.destinations[0].writtenBytes == Int64(size)) + #expect(latest.destinations[1].writtenBytes == 4 * 1024 * 1024) + #expect(latest.destinations[1].readbackBytes == 0) + #expect(operation.results.contains { !$0.success }) + let failedRoot = SafetyValidator.resolvedDestinationRoot( + source: fixture.source, destination: fixture.destinations[1], + settings: CameraLabelSettings()) + #expect(!FileManager.default.fileExists(atPath: failedRoot.appendingPathComponent("clip.bin").path)) + #expect(try Data(contentsOf: fixture.source.appendingPathComponent("clip.bin")) == sourceBytes) + #else + #expect(true) + #endif + } + } + @Test func destinationFaultsStayIsolatedAndFailCoverage() async throws { try await FileOperationsTestLock.shared.run { diff --git a/Packages/BitMatchEngine/Tests/BitMatchEngineTests/TransferIOMeterTests.swift b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/TransferIOMeterTests.swift new file mode 100644 index 00000000..99409faf --- /dev/null +++ b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/TransferIOMeterTests.swift @@ -0,0 +1,54 @@ +import Foundation +import Testing +@testable import BitMatchEngine + +struct TransferIOMeterTests { + @Test func ratesSeparateFastAndSlowDestinationsAndReadback() throws { + let meter = TransferIOMeter(destinationCount: 2) + let chunk = 4 * 1024 * 1024 + + let first = try #require(meter.record(destination: 0, phase: .writing, + bytes: chunk, duration: 0.1, now: 10)) + #expect(first.destinations[0].writtenBytes == Int64(chunk)) + #expect(first.destinations[0].writeBytesPerSecond == nil) + #expect(first.destinations[1].writtenBytes == 0) + + _ = meter.record(destination: 1, phase: .writing, bytes: chunk, duration: 0.5, now: 10.3) + _ = meter.record(destination: 0, phase: .writing, bytes: chunk, duration: 0.1, now: 10.6) + _ = meter.record(destination: 1, phase: .writing, bytes: chunk, duration: 0.5, now: 10.9) + let writes = meter.finalSnapshot(now: 10.9) + #expect(writes.destinations.map(\.writtenBytes) == [Int64(chunk * 2), Int64(chunk * 2)]) + #expect((writes.destinations[0].writeBytesPerSecond ?? 0) > + (writes.destinations[1].writeBytesPerSecond ?? 0) * 4) + #expect(writes.destinations.allSatisfy { $0.readbackBytes == 0 }) + + _ = meter.record(destination: 1, phase: .readingBack, bytes: chunk, duration: 0.2, now: 11.2) + _ = meter.record(destination: 1, phase: .readingBack, bytes: chunk, duration: 0.2, now: 11.5) + let read = meter.finalSnapshot(now: 11.5) + #expect(read.destinations[0].readbackBytes == 0) + #expect(read.destinations[1].readbackBytes == Int64(chunk * 2)) + #expect((read.destinations[1].readbackBytesPerSecond ?? 0) > 0) + #expect(read.sequence > writes.sequence) + } + + @Test func invalidSamplesAndSilentDestinationsCannotInventSpeed() { + let meter = TransferIOMeter(destinationCount: 2) + #expect(meter.record(destination: 5, phase: .writing, bytes: 100, duration: 1, now: 1) == nil) + #expect(meter.record(destination: 0, phase: .writing, bytes: 0, duration: 1, now: 1) == nil) + #expect(meter.record(destination: 0, phase: .writing, bytes: 100, duration: -1, now: 1) == nil) + let snapshot = meter.finalSnapshot(now: 1) + #expect(snapshot.destinations.allSatisfy { $0.writtenBytes == 0 && $0.writeBytesPerSecond == nil }) + } + + @Test func throttledUIUpdatesStillCountEverySuccessfulChunk() throws { + let meter = TransferIOMeter(destinationCount: 1) + let first = try #require(meter.record(destination: 0, phase: .writing, + bytes: 100, duration: 0.1, now: 10)) + #expect(meter.record(destination: 0, phase: .writing, + bytes: 200, duration: 0.1, now: 10.1) == nil) + let final = meter.finalSnapshot(now: 10.1) + #expect(final.sequence > first.sequence) + #expect(final.destinations[0].writtenBytes == 300) + #expect(abs((final.destinations[0].writeBytesPerSecond ?? 0) - 1_500) < 1) + } +} diff --git a/Shared/Core/Models/TransferProgressPresentation.swift b/Shared/Core/Models/TransferProgressPresentation.swift index 35df1a00..941b663b 100644 --- a/Shared/Core/Models/TransferProgressPresentation.swift +++ b/Shared/Core/Models/TransferProgressPresentation.swift @@ -43,15 +43,14 @@ struct ProgressControls: Equatable, Sendable { } } -/// One backup's line while a run is in progress. It reports copying only: -/// the engine counts a backup's files as they are copied (failures -/// included), and verification is judged in the finished transfer row. So a row is -/// never green and never says "Done" (accessibility audit C1). +/// One backup's line while a run is in progress. File counts include failed +/// attempts, so even a full count is only "Processed", never a copy verdict. +/// Verification is judged in the finished transfer row (accessibility C1). struct DestinationProgressRow: Equatable, Identifiable, Sendable { enum State: Equatable, Sendable { case waiting case copying - /// Every file was copied to this backup; not a verdict. + /// Every file was processed for this backup; failures may be included. case copied case verifying } @@ -67,16 +66,50 @@ struct DestinationProgressRow: Equatable, Identifiable, Sendable { /// reported per-backup counts yet. Never the overall fraction. let fraction: Double? let countText: String? + /// Observed I/O only; no path, verification verdict or drive-health claim. + let io: DestinationIOProgress? + let showsLiveRate: Bool + let minimumRateUptime: TimeInterval var stateLabel: String { switch state { case .waiting: "Waiting" case .copying: "Copying" - case .copied: "Copied" + case .copied: "Processed" case .verifying: "Verifying" } } + func writeDetail(nowUptime: TimeInterval) -> String? { + guard let io, io.writtenBytes > 0 else { return nil } + return Self.detail("Written", bytes: io.writtenBytes, + rate: io.writeBytesPerSecond, lastUptime: io.lastWriteUptime, + nowUptime: nowUptime, + showsRate: showsLiveRate && state == .copying, + minimumRateUptime: minimumRateUptime) + } + + func readbackDetail(nowUptime: TimeInterval) -> String? { + guard let io, io.readbackBytes > 0 else { return nil } + return Self.detail("Read back", bytes: io.readbackBytes, + rate: io.readbackBytesPerSecond, lastUptime: io.lastReadbackUptime, + nowUptime: nowUptime, showsRate: showsLiveRate, + minimumRateUptime: minimumRateUptime) + } + + private static func detail(_ label: String, bytes: Int64, rate: Double?, + lastUptime: TimeInterval?, nowUptime: TimeInterval, + showsRate: Bool, minimumRateUptime: TimeInterval) -> String { + let amount = ByteCountFormatter.string(fromByteCount: bytes, countStyle: .file) + guard showsRate, let rate, rate.isFinite, rate > 0, + let lastUptime, lastUptime >= minimumRateUptime, + nowUptime - lastUptime >= 0, + nowUptime - lastUptime < 2 else { return "\(label) \(amount)" } + let rateAmount = ByteCountFormatter.string( + fromByteCount: Int64(min(rate, Double(Int64.max / 2))), countStyle: .file) + return "\(label) \(amount) · \(rateAmount)/s" + } + var symbol: String { switch state { case .waiting: "clock" @@ -177,6 +210,8 @@ struct TransferProgressPresentation: Equatable, Sendable { state: OperationState, isRunning: Bool, progress: OperationProgress?, + ioProgress: TransferIOProgress? = nil, + minimumRateUptime: TimeInterval = 0, sourceName: String?, destinations: [URL], destinationNames: [String]? = nil, @@ -220,6 +255,8 @@ struct TransferProgressPresentation: Equatable, Sendable { destinations: destinations, destinationNames: destinationNames, progress: progress, + ioProgress: ioProgress, + minimumRateUptime: minimumRateUptime, phase: phase ), controls: controls, @@ -323,6 +360,8 @@ struct TransferProgressPresentation: Equatable, Sendable { destinations: [URL], destinationNames: [String]? = nil, progress: OperationProgress?, + ioProgress: TransferIOProgress? = nil, + minimumRateUptime: TimeInterval = 0, phase: ProgressPhase ) -> [DestinationProgressRow] { let totals = progress?.perDestinationTotals @@ -346,6 +385,13 @@ struct TransferProgressPresentation: Equatable, Sendable { let name = destinationNames.flatMap { names in names.indices.contains(index) ? names[index] : nil } ?? DestinationVolumeLabel.name(for: url) + let measured = ioProgress.flatMap { update in + update.destinations.indices.contains(index) ? update.destinations[index] : nil + } + let showsIO = phase == .copying || phase == .verifying + || phase == .paused || phase == .resuming + let hasMeasuredBytes = (measured?.writtenBytes ?? 0) > 0 + || (measured?.readbackBytes ?? 0) > 0 return DestinationProgressRow( id: url.path, name: name, @@ -353,7 +399,10 @@ struct TransferProgressPresentation: Equatable, Sendable { helpPath: url.path, state: state, fraction: fraction, - countText: countText + countText: countText, + io: showsIO && hasMeasuredBytes ? measured : nil, + showsLiveRate: phase == .copying || phase == .verifying, + minimumRateUptime: minimumRateUptime ) } } diff --git a/Shared/Core/Services/CopyVerifyExecutor.swift b/Shared/Core/Services/CopyVerifyExecutor.swift index 25ce444e..e935af0a 100644 --- a/Shared/Core/Services/CopyVerifyExecutor.swift +++ b/Shared/Core/Services/CopyVerifyExecutor.swift @@ -68,6 +68,21 @@ struct CopyVerifyCallbacks: Sendable { let onResult: @MainActor @Sendable (ResultRow) -> Void let onStateChange: @MainActor @Sendable (OperationState) -> Void let onAuthoritativeResults: @MainActor @Sendable ([ResultRow]) throws -> Void + let onIOProgress: @MainActor @Sendable (TransferIOProgress) -> Void + + init( + onProgress: @escaping @MainActor @Sendable (OperationProgress) -> Void, + onResult: @escaping @MainActor @Sendable (ResultRow) -> Void, + onStateChange: @escaping @MainActor @Sendable (OperationState) -> Void, + onAuthoritativeResults: @escaping @MainActor @Sendable ([ResultRow]) throws -> Void, + onIOProgress: @escaping @MainActor @Sendable (TransferIOProgress) -> Void = { _ in } + ) { + self.onProgress = onProgress + self.onResult = onResult + self.onStateChange = onStateChange + self.onAuthoritativeResults = onAuthoritativeResults + self.onIOProgress = onIOProgress + } } /// Service that executes copy/verify operations @@ -195,6 +210,12 @@ final class CopyVerifyExecutor { self.handleProgress(progressUpdate, callbacks: callbacks) } } + let ioProgressCallback: FileOperationsService.IOProgressCallback = { [weak self] update in + Task { @MainActor in + guard let self, self.diagnosticRun == config.operationId else { return } + callbacks.onIOProgress(update) + } + } let fileResultCallback: FileOperationsService.FileResultCallback? = { [weak self] fileResult in guard let self else { return } await self.handleFileResult( @@ -218,6 +239,7 @@ final class CopyVerifyExecutor { settings: config.cameraLabelSettings, estimatedTotalBytes: config.estimatedBytes, progressCallback: progressCallback, + ioProgressCallback: ioProgressCallback, onFileResult: fileResultCallback, fanOutHooks: demoHooks ) @@ -229,6 +251,7 @@ final class CopyVerifyExecutor { settings: config.cameraLabelSettings, estimatedTotalBytes: config.estimatedBytes, progressCallback: progressCallback, + ioProgressCallback: ioProgressCallback, onFileResult: fileResultCallback ) } @@ -240,6 +263,7 @@ final class CopyVerifyExecutor { settings: config.cameraLabelSettings, estimatedTotalBytes: config.estimatedBytes, progressCallback: progressCallback, + ioProgressCallback: ioProgressCallback, onFileResult: fileResultCallback ) #endif diff --git a/Shared/Core/Services/SharedAppCoordinator.swift b/Shared/Core/Services/SharedAppCoordinator.swift index e0bbba28..996fc133 100644 --- a/Shared/Core/Services/SharedAppCoordinator.swift +++ b/Shared/Core/Services/SharedAppCoordinator.swift @@ -769,6 +769,9 @@ class SharedAppCoordinator: ObservableObject { let presentation = self.progressPresentation switch state { case .inProgress, .copying, .verifying, .resuming: + if case .resuming = state { + self.liveProgress.rateFloorUptime = ProcessInfo.processInfo.systemUptime + } // One run is tracked once. Resume (`.resuming` then // `.inProgress`) and stage changes keep the byte totals // and speed samples; a new run stops tracking first @@ -784,9 +787,11 @@ class SharedAppCoordinator: ObservableObject { case .paused: // Paused time is left out of speed and time remaining. presentation.notePaused() + self.liveProgress.rateFloorUptime = ProcessInfo.processInfo.systemUptime case .completed, .failed, .cancelled: presentation.stopProgressTracking() self.lastPresentedBytes = 0 + self.liveProgress.rateFloorUptime = ProcessInfo.processInfo.systemUptime default: break } @@ -1830,6 +1835,8 @@ class SharedAppCoordinator: ObservableObject { let startID = UUID() activeStartID = startID startCancellationRequested = false + liveProgress.ioProgress = nil + liveProgress.rateFloorUptime = ProcessInfo.processInfo.systemUptime isOperationInProgress = true activeJournalRecordID = nil defer { @@ -2044,6 +2051,13 @@ class SharedAppCoordinator: ObservableObject { SharedLogger.transferEvent(.authoritativeAccepted, run: startID) runResults.replace(with: allResults) self.results = allResults + }, + onIOProgress: { [weak self] update in + guard let self, + self.activeStartID == startID, + !self.startCancellationRequested, + update.sequence > (self.liveProgress.ioProgress?.sequence ?? 0) else { return } + self.liveProgress.ioProgress = update } ) @@ -2791,6 +2805,8 @@ class SharedAppCoordinator: ObservableObject { func resetForNewOperation() { results = [] progress = nil + liveProgress.ioProgress = nil + liveProgress.rateFloorUptime = 0 progressPresentation.reset() lastPresentedBytes = 0 presentedDestinationCount = nil diff --git a/Shared/Core/ViewModels/LiveProgressFeed.swift b/Shared/Core/ViewModels/LiveProgressFeed.swift index 9ca5cdb2..c7e3c95e 100644 --- a/Shared/Core/ViewModels/LiveProgressFeed.swift +++ b/Shared/Core/ViewModels/LiveProgressFeed.swift @@ -12,6 +12,9 @@ import BitMatchEngine @MainActor final class LiveProgressFeed: ObservableObject { @Published var progress: OperationProgress? + @Published var ioProgress: TransferIOProgress? + /// A pause/resume or new run cannot reuse a pre-pause speed sample. + @Published var rateFloorUptime: TimeInterval = 0 init(progress: OperationProgress? = nil) { self.progress = progress diff --git a/Shared/Views/Progress/ProgressScreen.swift b/Shared/Views/Progress/ProgressScreen.swift index 0eda0d44..777fe4fa 100644 --- a/Shared/Views/Progress/ProgressScreen.swift +++ b/Shared/Views/Progress/ProgressScreen.swift @@ -326,6 +326,20 @@ private struct DestinationProgressRowView: View { let showsProgress: Bool var body: some View { + Group { + if row.io != nil { + // Refresh both the visible rate and its VoiceOver value when + // I/O stops, so a stale sample never sounds current. + TimelineView(.periodic(from: .now, by: 1)) { _ in + content(nowUptime: ProcessInfo.processInfo.systemUptime) + } + } else { + content(nowUptime: ProcessInfo.processInfo.systemUptime) + } + } + } + + private func content(nowUptime: TimeInterval) -> some View { VStack(alignment: .leading, spacing: 6) { HStack(spacing: 8) { Text(row.name) @@ -353,6 +367,19 @@ private struct DestinationProgressRowView: View { .tint(tint) .frame(minHeight: 8) } + if row.io != nil { + VStack(alignment: .leading, spacing: 2) { + if let detail = row.writeDetail(nowUptime: nowUptime) { + Text(detail) + } + if let detail = row.readbackDetail(nowUptime: nowUptime) { + Text(detail) + } + } + .font(.caption) + .foregroundStyle(.secondary) + .monospacedDigit() + } } .padding(.vertical, 9) .padding(.horizontal, 2) @@ -360,7 +387,11 @@ private struct DestinationProgressRowView: View { .help(row.helpPath) .accessibilityElement(children: .ignore) .accessibilityLabel("Destination \(row.name), \(row.stateLabel)") - .accessibilityValue(row.countText ?? "") + .accessibilityValue([ + row.countText, + row.writeDetail(nowUptime: nowUptime), + row.readbackDetail(nowUptime: nowUptime) + ].compactMap { $0 }.joined(separator: ", ")) } private var statusColor: Color { diff --git a/Shared/Views/TransferPresentationAdapters.swift b/Shared/Views/TransferPresentationAdapters.swift index 69ecc1f1..56c66949 100644 --- a/Shared/Views/TransferPresentationAdapters.swift +++ b/Shared/Views/TransferPresentationAdapters.swift @@ -10,6 +10,8 @@ extension TransferProgressPresentation { state: coordinator.operationState, isRunning: coordinator.isOperationInProgress, progress: coordinator.progress, + ioProgress: coordinator.liveProgress.ioProgress, + minimumRateUptime: coordinator.liveProgress.rateFloorUptime, sourceName: coordinator.presentedSourceURL?.lastPathComponent, destinations: coordinator.presentedDestinationURLs, destinationNames: coordinator.presentedDestinationNames, diff --git a/docs/TODO.md b/docs/TODO.md index 34b65d59..def15f66 100644 --- a/docs/TODO.md +++ b/docs/TODO.md @@ -160,7 +160,7 @@ Remote scope excludes screen sharing, arbitrary filesystem browsing, remote shel - [ ] **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. -- [ ] **Speed for each drive.** Show which destination is slow instead of only an average; share measured-throughput instrumentation with the storage-scheduling work, without requiring a scheduler change to display it. Put per-destination copy/readback progress and measured throughput in expanded details, not another default setup control; distinguish a slow write from slow verification. +- [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. - [ ] **Time estimate before Start.** Use card size and recent measured speeds; keep “estimating” when evidence is insufficient. From 06a5011f5bec27cd57309affa34a9eff70bbf961 Mon Sep 17 00:00:00 2001 From: Michael Cerisano Date: Sat, 17 Oct 2026 09:39:46 -0400 Subject: [PATCH 2/2] Correct first-file progress and active I/O rates --- .../ProgressPresentationFeedTests.swift | 1 + .../TransferProgressPresentationTests.swift | 47 +++++++++++++ .../File/DestinationWriter.swift | 37 ++++++---- .../Models/TransferIOProgress.swift | 67 +++++++++++++++---- .../BitMatchEngine/TransferPipeline.swift | 11 ++- .../BitMatchEngineTests/FanOutTests.swift | 26 +++++++ .../TransferIOMeterTests.swift | 23 +++++++ .../Models/TransferProgressPresentation.swift | 22 +++++- Shared/Views/Progress/ProgressScreen.swift | 3 +- 9 files changed, 204 insertions(+), 33 deletions(-) diff --git a/BitMatchTests/ProgressPresentationFeedTests.swift b/BitMatchTests/ProgressPresentationFeedTests.swift index 08d3ad90..913aef1d 100644 --- a/BitMatchTests/ProgressPresentationFeedTests.swift +++ b/BitMatchTests/ProgressPresentationFeedTests.swift @@ -50,6 +50,7 @@ struct ProgressPresentationFeedTests { coordinator.liveProgress.ioProgress = TransferIOProgress(sequence: 1, destinations: [ DestinationIOProgress(writtenBytes: 1, readbackBytes: 0, writeBytesPerSecond: nil, readbackBytesPerSecond: nil, + writeRateWindowStartUptime: nil, readbackRateWindowStartUptime: nil, lastWriteUptime: nil, lastReadbackUptime: nil) ]) #expect(await waitUntil { coordinator.progressPresentation.fileCountTotal == 4 }) diff --git a/BitMatchTests/TransferProgressPresentationTests.swift b/BitMatchTests/TransferProgressPresentationTests.swift index eb2c8514..d972924e 100644 --- a/BitMatchTests/TransferProgressPresentationTests.swift +++ b/BitMatchTests/TransferProgressPresentationTests.swift @@ -135,10 +135,12 @@ struct TransferProgressPresentationTests { DestinationIOProgress( writtenBytes: 8_000_000, readbackBytes: 4_000_000, writeBytesPerSecond: 16_000_000, readbackBytesPerSecond: 8_000_000, + writeRateWindowStartUptime: 99.5, readbackRateWindowStartUptime: 99.5, lastWriteUptime: 100, lastReadbackUptime: 100), DestinationIOProgress( writtenBytes: 0, readbackBytes: 0, writeBytesPerSecond: nil, readbackBytesPerSecond: nil, + writeRateWindowStartUptime: nil, readbackRateWindowStartUptime: nil, lastWriteUptime: nil, lastReadbackUptime: nil) ]) let p = make(state: .inProgress, @@ -179,6 +181,51 @@ struct TransferProgressPresentationTests { #expect(report[0].io == nil) } + @Test func firstLargeClipShowsActiveDestinationBeforeAnyFileCompletes() { + let sample = TransferIOProgress(sequence: 1, destinations: [ + DestinationIOProgress( + writtenBytes: 8_000_000, readbackBytes: 0, + writeBytesPerSecond: 16_000_000, readbackBytesPerSecond: nil, + writeRateWindowStartUptime: 99.5, readbackRateWindowStartUptime: nil, + lastWriteUptime: 100, lastReadbackUptime: nil) + ]) + let active = TransferProgressPresentation.rows( + destinations: [primary], + progress: progress(stage: .copying, overall: 0, files: 0, total: 1, + totals: [1], completed: [0]), + ioProgress: sample, nowUptime: 100.5, phase: .copying + ) + #expect(active[0].countText == "0 of 1") + #expect(active[0].stateLabel == "Copying") + #expect(active[0].writeDetail(nowUptime: 100.5)?.contains("/s") == true) + + let stale = TransferProgressPresentation.rows( + destinations: [primary], + progress: progress(stage: .copying, overall: 0, files: 0, total: 1, + totals: [1], completed: [0]), + ioProgress: sample, nowUptime: 103, phase: .copying + ) + #expect(stale[0].stateLabel == "Waiting") + #expect(stale[0].writeDetail(nowUptime: 103)?.contains("/s") == false) + } + + @Test func firstResumedSampleCannotReusePrePauseRateWindow() { + let mixed = TransferIOProgress(sequence: 2, destinations: [ + DestinationIOProgress( + writtenBytes: 12_000_000, readbackBytes: 0, + writeBytesPerSecond: 20_000_000, readbackBytesPerSecond: nil, + writeRateWindowStartUptime: 99.5, readbackRateWindowStartUptime: nil, + lastWriteUptime: 101.3, lastReadbackUptime: nil) + ]) + let rows = TransferProgressPresentation.rows( + destinations: [primary], progress: nil, ioProgress: mixed, + minimumRateUptime: 101, nowUptime: 101.5, phase: .copying + ) + #expect(rows[0].stateLabel == "Copying") + #expect(rows[0].writeDetail(nowUptime: 101.5)?.contains("/s") == false) + #expect(rows[0].writeDetail(nowUptime: 101.5)?.contains("Written") == true) + } + @Test func destinationRowsReusePrecomputedVolumeNames() { let internalFolder = URL(fileURLWithPath: "/Users/mike/Backups/Project", isDirectory: true) let rows = TransferProgressPresentation.rows( diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift index b9b6af93..add9f353 100644 --- a/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/File/DestinationWriter.swift @@ -404,7 +404,7 @@ public final class PinnedDestinationFile: @unchecked Sendable { #endif public final class DestinationWriter { - typealias IOByteCallback = @Sendable (Int, TransferIOPhase, Int, TimeInterval) -> Void + typealias IOByteCallback = @Sendable (Int, TransferIOPhase, Int, TimeInterval, TimeInterval) -> Void struct SourceReadEvidence: Sendable { let digests: VerifiedDigests let identity: VerifiedFileIdentity @@ -833,10 +833,9 @@ public final class DestinationWriter { file.destination.root.logicalRootURL, currentChunkIndex ) - let writeStarted = ProcessInfo.processInfo.systemUptime - try writeAll(chunkData, to: file.temporaryFD) - onIOBytes?(file.destination.index, .writing, chunkData.count, - ProcessInfo.processInfo.systemUptime - writeStarted) + try writeAll(chunkData, to: file.temporaryFD) { count, seconds, ended in + onIOBytes?(file.destination.index, .writing, count, seconds, ended) + } return FanOutWriteResult(file: file, error: nil) } catch { return FanOutWriteResult(file: file, error: error) @@ -1004,14 +1003,23 @@ public final class DestinationWriter { return FileHandle(fileDescriptor: fd, closeOnDealloc: true) } - private static func writeAll(_ data: Data, to fd: Int32) throws { + private static func writeAll( + _ data: Data, + to fd: Int32, + onWritten: (Int, TimeInterval, TimeInterval) -> Void + ) throws { try data.withUnsafeBytes { rawBuffer in guard let base = rawBuffer.baseAddress else { return } var written = 0 while written < rawBuffer.count { + let started = ProcessInfo.processInfo.systemUptime let count = Darwin.write(fd, base.advanced(by: written), rawBuffer.count - written) if count > 0 { written += count + // Count successful syscalls even if a later write for the + // same chunk fails and the temporary file is discarded. + let ended = ProcessInfo.processInfo.systemUptime + onWritten(count, ended - started, ended) } else if count < 0, errno == EINTR { continue } else { @@ -1537,16 +1545,16 @@ public final class DestinationWriter { try await PauseGate.waitIfCurrentIsPaused() // Bound temporary Foundation allocations to this chunk, even when // the pause gate returns without suspending the task. - let (count, readSeconds) = try autoreleasepool { () throws -> (Int, TimeInterval) in + let (count, readSeconds, readEnded) = try autoreleasepool { () throws -> (Int, TimeInterval, TimeInterval) in let readStarted = ProcessInfo.processInfo.systemUptime let data = try handle.read(upToCount: 4 * 1024 * 1024) ?? Data() - let readSeconds = ProcessInfo.processInfo.systemUptime - readStarted + let readEnded = ProcessInfo.processInfo.systemUptime if !data.isEmpty { consume(data) } - return (data.count, readSeconds) + return (data.count, readEnded - readStarted, readEnded) } if count == 0 { break } hooks?.destinationDidRead?(destinationIndex, relativePath, count) - onIOBytes?(destinationIndex, .readingBack, count, readSeconds) + onIOBytes?(destinationIndex, .readingBack, count, readSeconds, readEnded) bytesRead += Int64(count) if ordinal != nil, bytesRead >= nextBreadcrumb { nextBreadcrumb += verifyReadBreadcrumbInterval @@ -1595,21 +1603,22 @@ public final class DestinationWriter { try Task.checkCancellation() try await PauseGate.waitIfCurrentIsPaused() // Drain each chunk's buffers; see readPinnedDestination. - let (sourceCount, destinationCount, same, readSeconds) = try autoreleasepool { () throws -> (Int, Int, Bool, TimeInterval) in + let (sourceCount, destinationCount, same, readSeconds, readEnded) = try autoreleasepool { () throws -> (Int, Int, Bool, TimeInterval, TimeInterval) in let sourceRead = try sourceHandle.read(upToCount: 4 * 1024 * 1024) ?? Data() let sourceData = try hooks?.transformVerificationSourceRead?(relativePath, sourceRead) ?? sourceRead guard sourceData.count == sourceRead.count else { throw sourceChangedError() } let readStarted = ProcessInfo.processInfo.systemUptime let destinationData = try destinationHandle.read(upToCount: 4 * 1024 * 1024) ?? Data() - let readSeconds = ProcessInfo.processInfo.systemUptime - readStarted - return (sourceData.count, destinationData.count, sourceData == destinationData, readSeconds) + let readEnded = ProcessInfo.processInfo.systemUptime + return (sourceData.count, destinationData.count, sourceData == destinationData, + readEnded - readStarted, readEnded) } guard sourceCount > 0, destinationCount > 0 else { throw NSError(domain: "DestinationWriter", code: -11, userInfo: [NSLocalizedDescriptionKey: "File changed while comparing bytes"]) } hooks?.verificationSourceDidRead?(relativePath, sourceCount) hooks?.destinationDidRead?(destinationIndex, relativePath, destinationCount) - onIOBytes?(destinationIndex, .readingBack, destinationCount, readSeconds) + onIOBytes?(destinationIndex, .readingBack, destinationCount, readSeconds, readEnded) if !same { return false } bytesRead += Int64(sourceCount) } diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift index e3997b2d..40699811 100644 --- a/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/Models/TransferIOProgress.swift @@ -8,16 +8,23 @@ public struct DestinationIOProgress: Equatable, Sendable { public let readbackBytes: Int64 public let writeBytesPerSecond: Double? public let readbackBytesPerSecond: Double? + /// Earliest I/O call included in the displayed rate. A pause invalidates + /// a rate window that still contains pre-pause work. + public let writeRateWindowStartUptime: TimeInterval? + public let readbackRateWindowStartUptime: TimeInterval? public let lastWriteUptime: TimeInterval? public let lastReadbackUptime: TimeInterval? public init(writtenBytes: Int64, readbackBytes: Int64, writeBytesPerSecond: Double?, readbackBytesPerSecond: Double?, + writeRateWindowStartUptime: TimeInterval?, readbackRateWindowStartUptime: TimeInterval?, lastWriteUptime: TimeInterval?, lastReadbackUptime: TimeInterval?) { self.writtenBytes = writtenBytes self.readbackBytes = readbackBytes self.writeBytesPerSecond = writeBytesPerSecond self.readbackBytesPerSecond = readbackBytesPerSecond + self.writeRateWindowStartUptime = writeRateWindowStartUptime + self.readbackRateWindowStartUptime = readbackRateWindowStartUptime self.lastWriteUptime = lastWriteUptime self.lastReadbackUptime = lastReadbackUptime } @@ -40,15 +47,22 @@ enum TransferIOPhase: Sendable { } /// Counts bytes only after an actual destination I/O call succeeds. A short -/// rolling window measures each destination's I/O service time, not the whole -/// fan-out wall time: the writer waits for its slowest destination after each -/// chunk, so dividing every destination's bytes by the same wall time would -/// misleadingly give fast and slow destinations the same rate. +/// rolling window measures the union of each destination's active I/O calls. +/// It excludes fan-out waiting: the writer waits for its slowest destination +/// after each chunk, so dividing every destination's bytes by the same wall +/// time would give fast and slow destinations the same rate. final class TransferIOMeter: Sendable { private struct Sample { let at: TimeInterval let bytes: Int64 let seconds: TimeInterval + + var started: TimeInterval { at - seconds } + } + + private struct Rate { + let bytesPerSecond: Double + let windowStartUptime: TimeInterval } private struct PhaseState { @@ -59,17 +73,40 @@ final class TransferIOMeter: Sendable { mutating func add(bytes newBytes: Int, duration: TimeInterval, now: TimeInterval) { let (sum, overflow) = bytes.addingReportingOverflow(Int64(newBytes)) bytes = overflow ? Int64.max : sum - lastUptime = now + lastUptime = max(lastUptime ?? now, now) samples.append(Sample(at: now, bytes: Int64(newBytes), seconds: max(0, duration))) - samples.removeAll { now - $0.at > 3 } + let newest = lastUptime ?? now + samples.removeAll { newest - $0.at > 3 } if samples.count > 64 { samples.removeFirst(samples.count - 64) } } - func rate(now: TimeInterval) -> Double? { + func rate(now: TimeInterval) -> Rate? { let recent = samples.filter { now - $0.at <= 3 } - let seconds = recent.reduce(0) { $0 + $1.seconds } - guard recent.count >= 2, seconds >= 0.1 else { return nil } - return Double(recent.reduce(Int64(0)) { $0 + $1.bytes }) / seconds + guard recent.count >= 2 else { return nil } + // Copy workers and pipelined verifiers can issue calls to the + // same destination concurrently. Summing call durations would + // double-count overlap and understate that destination's rate. + let intervals = recent.sorted { $0.started < $1.started } + var intervalStart = intervals[0].started + var intervalEnd = intervals[0].at + var activeSeconds: TimeInterval = 0 + for sample in intervals.dropFirst() { + if sample.started <= intervalEnd { + intervalEnd = max(intervalEnd, sample.at) + } else { + activeSeconds += intervalEnd - intervalStart + intervalStart = sample.started + intervalEnd = sample.at + } + } + activeSeconds += intervalEnd - intervalStart + // Avoid timing noise on the first tiny calls, but let a sustained + // fast destination converge even when 64 samples take <100 ms. + guard activeSeconds.isFinite, + activeSeconds >= 0.01 || (recent.count >= 16 && activeSeconds > 0) else { return nil } + let bytes = recent.reduce(Int64(0)) { $0 + $1.bytes } + return Rate(bytesPerSecond: Double(bytes) / activeSeconds, + windowStartUptime: intervals[0].started) } } @@ -115,11 +152,15 @@ final class TransferIOMeter: Sendable { private static func snapshot(_ state: inout State, now: TimeInterval) -> TransferIOProgress { state.sequence &+= 1 return TransferIOProgress(sequence: state.sequence, destinations: state.destinations.map { destination in - DestinationIOProgress( + let writeRate = destination.writing.rate(now: now) + let readbackRate = destination.readingBack.rate(now: now) + return DestinationIOProgress( writtenBytes: destination.writing.bytes, readbackBytes: destination.readingBack.bytes, - writeBytesPerSecond: destination.writing.rate(now: now), - readbackBytesPerSecond: destination.readingBack.rate(now: now), + writeBytesPerSecond: writeRate?.bytesPerSecond, + readbackBytesPerSecond: readbackRate?.bytesPerSecond, + writeRateWindowStartUptime: writeRate?.windowStartUptime, + readbackRateWindowStartUptime: readbackRate?.windowStartUptime, lastWriteUptime: destination.writing.lastUptime, lastReadbackUptime: destination.readingBack.lastUptime ) diff --git a/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift b/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift index 240cf344..14172392 100644 --- a/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift +++ b/Packages/BitMatchEngine/Sources/BitMatchEngine/TransferPipeline.swift @@ -522,9 +522,9 @@ public final class TransferPipeline: FileOperationsService, Sendable { // 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 in + let onIOBytes: DestinationWriter.IOByteCallback = { index, phase, bytes, duration, ended in if let snapshot = ioMeter.record(destination: index, phase: phase, - bytes: bytes, duration: duration) { + bytes: bytes, duration: duration, now: ended) { ioProgressCallback(snapshot) } } @@ -790,6 +790,13 @@ public final class TransferPipeline: FileOperationsService, Sendable { ) } + // A first clip may take hours. Enter the copy stage before its + // 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)) + } + try await DestinationWriter.copyAllSafelyFanOut( from: operation.sourceURL, toPinnedRoots: fanOutDestinations, diff --git a/Packages/BitMatchEngine/Tests/BitMatchEngineTests/FanOutTests.swift b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/FanOutTests.swift index ddeedc95..df2000a7 100644 --- a/Packages/BitMatchEngine/Tests/BitMatchEngineTests/FanOutTests.swift +++ b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/FanOutTests.swift @@ -8,6 +8,32 @@ import Darwin @testable import BitMatchEngine struct FanOutTests { + @Test + func copyStageStartsBeforeFirstLargeFileFinishes() async throws { + try await FileOperationsTestLock.shared.run { + #if os(macOS) + let fixture = try FanOutFixture(name: "copy-stage-before-result") + defer { fixture.cleanup() } + try fixture.write(Data(repeating: 0x41, count: 4 * 1024 * 1024 + 1), to: "clip.bin") + let updates = Mutex<[OperationProgress]>([]) + _ = try await fixture.pipeline(hooks: .init()).performFileOperation( + sourceURL: fixture.source, destinationURLs: fixture.destinations, + verificationMode: .standard, settings: CameraLabelSettings(), + progressCallback: { update in updates.withLock { $0.append(update) } }, + onFileResult: nil + ) + let stages = updates.withLock { $0 } + #expect(stages.first?.currentStage == .preparing) + #expect(stages.contains { + $0.currentStage == .copying && $0.filesProcessed == 0 + && $0.perDestinationCompleted == [0, 0] + }) + #else + #expect(true) + #endif + } + } + @Test func liveIOCountsRealWritesAndReadbackWithoutChangingTheVerdict() async throws { try await FileOperationsTestLock.shared.run { diff --git a/Packages/BitMatchEngine/Tests/BitMatchEngineTests/TransferIOMeterTests.swift b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/TransferIOMeterTests.swift index 99409faf..e1e1bd90 100644 --- a/Packages/BitMatchEngine/Tests/BitMatchEngineTests/TransferIOMeterTests.swift +++ b/Packages/BitMatchEngine/Tests/BitMatchEngineTests/TransferIOMeterTests.swift @@ -51,4 +51,27 @@ struct TransferIOMeterTests { #expect(final.destinations[0].writtenBytes == 300) #expect(abs((final.destinations[0].writeBytesPerSecond ?? 0) - 1_500) < 1) } + + @Test func overlappingCallsCountConcurrentBytesOverOneActiveInterval() { + let meter = TransferIOMeter(destinationCount: 1) + let chunk = 4 * 1024 * 1024 + _ = meter.record(destination: 0, phase: .writing, bytes: chunk, duration: 1, now: 10) + _ = meter.record(destination: 0, phase: .writing, bytes: chunk, duration: 1, now: 10) + let snapshot = meter.finalSnapshot(now: 10) + #expect(snapshot.destinations[0].writtenBytes == Int64(chunk * 2)) + #expect(abs((snapshot.destinations[0].writeBytesPerSecond ?? 0) - Double(chunk * 2)) < 1) + #expect(snapshot.destinations[0].writeRateWindowStartUptime == 9) + } + + @Test func sustainedFastCallsStillProduceARate() { + let meter = TransferIOMeter(destinationCount: 1) + for index in 0..<64 { + _ = meter.record(destination: 0, phase: .writing, + bytes: 4 * 1024 * 1024, duration: 0.0005, + now: 10 + Double(index + 1) * 0.0005) + } + let snapshot = meter.finalSnapshot(now: 10.032) + #expect(snapshot.destinations[0].writtenBytes == 64 * 4 * 1024 * 1024) + #expect((snapshot.destinations[0].writeBytesPerSecond ?? 0) > 0) + } } diff --git a/Shared/Core/Models/TransferProgressPresentation.swift b/Shared/Core/Models/TransferProgressPresentation.swift index 941b663b..5af33263 100644 --- a/Shared/Core/Models/TransferProgressPresentation.swift +++ b/Shared/Core/Models/TransferProgressPresentation.swift @@ -83,7 +83,9 @@ struct DestinationProgressRow: Equatable, Identifiable, Sendable { func writeDetail(nowUptime: TimeInterval) -> String? { guard let io, io.writtenBytes > 0 else { return nil } return Self.detail("Written", bytes: io.writtenBytes, - rate: io.writeBytesPerSecond, lastUptime: io.lastWriteUptime, + rate: io.writeBytesPerSecond, + rateWindowStartUptime: io.writeRateWindowStartUptime, + lastUptime: io.lastWriteUptime, nowUptime: nowUptime, showsRate: showsLiveRate && state == .copying, minimumRateUptime: minimumRateUptime) @@ -92,22 +94,26 @@ struct DestinationProgressRow: Equatable, Identifiable, Sendable { func readbackDetail(nowUptime: TimeInterval) -> String? { guard let io, io.readbackBytes > 0 else { return nil } return Self.detail("Read back", bytes: io.readbackBytes, - rate: io.readbackBytesPerSecond, lastUptime: io.lastReadbackUptime, + rate: io.readbackBytesPerSecond, + rateWindowStartUptime: io.readbackRateWindowStartUptime, + lastUptime: io.lastReadbackUptime, nowUptime: nowUptime, showsRate: showsLiveRate, minimumRateUptime: minimumRateUptime) } private static func detail(_ label: String, bytes: Int64, rate: Double?, + rateWindowStartUptime: TimeInterval?, lastUptime: TimeInterval?, nowUptime: TimeInterval, showsRate: Bool, minimumRateUptime: TimeInterval) -> String { let amount = ByteCountFormatter.string(fromByteCount: bytes, countStyle: .file) guard showsRate, let rate, rate.isFinite, rate > 0, + let rateWindowStartUptime, rateWindowStartUptime >= minimumRateUptime, let lastUptime, lastUptime >= minimumRateUptime, nowUptime - lastUptime >= 0, nowUptime - lastUptime < 2 else { return "\(label) \(amount)" } let rateAmount = ByteCountFormatter.string( fromByteCount: Int64(min(rate, Double(Int64.max / 2))), countStyle: .file) - return "\(label) \(amount) · \(rateAmount)/s" + return "\(label) \(amount) · active \(rateAmount)/s" } var symbol: String { @@ -362,6 +368,7 @@ struct TransferProgressPresentation: Equatable, Sendable { progress: OperationProgress?, ioProgress: TransferIOProgress? = nil, minimumRateUptime: TimeInterval = 0, + nowUptime: TimeInterval = ProcessInfo.processInfo.systemUptime, phase: ProgressPhase ) -> [DestinationProgressRow] { let totals = progress?.perDestinationTotals @@ -388,6 +395,15 @@ struct TransferProgressPresentation: Equatable, Sendable { let measured = ioProgress.flatMap { update in update.destinations.indices.contains(index) ? update.destinations[index] : nil } + if state == .waiting, phase == .copying, + let measured, measured.writtenBytes > 0, + let lastWrite = measured.lastWriteUptime, + lastWrite >= minimumRateUptime, + nowUptime - lastWrite >= 0, nowUptime - lastWrite < 2 { + // The first large file has no completed-file count yet. + // Successful recent writes prove this destination is active. + state = .copying + } let showsIO = phase == .copying || phase == .verifying || phase == .paused || phase == .resuming let hasMeasuredBytes = (measured?.writtenBytes ?? 0) > 0 diff --git a/Shared/Views/Progress/ProgressScreen.swift b/Shared/Views/Progress/ProgressScreen.swift index 777fe4fa..5be18979 100644 --- a/Shared/Views/Progress/ProgressScreen.swift +++ b/Shared/Views/Progress/ProgressScreen.swift @@ -384,7 +384,8 @@ private struct DestinationProgressRowView: View { .padding(.vertical, 9) .padding(.horizontal, 2) .frame(maxWidth: .infinity, alignment: .leading) - .help(row.helpPath) + .help(row.io == nil ? row.helpPath : + "\(row.helpPath)\nRates measure active file reads and writes, not full-transfer or durable-storage speed.") .accessibilityElement(children: .ignore) .accessibilityLabel("Destination \(row.name), \(row.stateLabel)") .accessibilityValue([