Make unpack methods async for deterministic progress delivery

Author: DePasqualeOrg

Sources/Containerization/Image/Unpacker/EXT4Unpacker.swift

@@ -42,19 +42,18 @@ public struct EXT4Unpacker: Unpacker {
         archive: URL,
         compression: ContainerizationArchive.Filter,
         at path: URL
-    ) throws {
+    ) async throws {
         let cleanedPath = try prepareUnpackPath(path: path)
         let filesystem = try EXT4.Formatter(
             FilePath(cleanedPath),
             minDiskSize: blockSizeInBytes
         )
         defer { try? filesystem.close() }
 
-        try filesystem.unpack(
+        try await filesystem.unpack(
             source: archive,
             format: .paxRestricted,
-            compression: compression,
-            progress: nil
+            compression: compression
         )
     }
     #endif
@@ -120,7 +119,7 @@ public struct EXT4Unpacker: Unpacker {
                 filter: resolved.filter,
                 file: resolved.file
             )
-            try filesystem.unpack(reader: reader, progress: progress)
+            try await filesystem.unpack(reader: reader, progress: progress)
         }
 
         return .block(

Sources/ContainerizationEXT4/Formatter+Unpack.swift

@@ -25,7 +25,26 @@ private typealias Hardlinks = [FilePath: FilePath]
 
 extension EXT4.Formatter {
     /// Unpack the provided archive on to the ext4 filesystem.
-    public func unpack(reader: ArchiveReader, progress: ProgressHandler? = nil) throws {
+    public func unpack(reader: ArchiveReader, progress: ProgressHandler? = nil) async throws {
+        try await self.unpackEntries(reader: reader, progress: progress)
+    }
+
+    /// Unpack an archive at the source URL on to the ext4 filesystem.
+    public func unpack(
+        source: URL,
+        format: ContainerizationArchive.Format = .paxRestricted,
+        compression: ContainerizationArchive.Filter = .gzip
+    ) async throws {
+        let reader = try ArchiveReader(
+            format: format,
+            filter: compression,
+            file: source
+        )
+        try await self.unpack(reader: reader)
+    }
+
+    /// Core unpack logic. When `progress` is nil the handler calls are skipped.
+    private func unpackEntries(reader: ArchiveReader, progress: ProgressHandler?) async throws {
         var hardlinks: Hardlinks = [:]
         // Allocate a single 128KiB reusable buffer for all files to minimize allocations
         // and reduce the number of read calls to libarchive.
@@ -39,35 +58,33 @@ extension EXT4.Formatter {
                 continue
             }
 
-            defer {
-                // Count the number of entries
-                if let progress {
-                    Task {
-                        await progress([
-                            ProgressEvent(event: "add-items", value: 1)
-                        ])
-                    }
-                }
-            }
-
             pathEntry = preProcessPath(s: pathEntry)
             let path = FilePath(pathEntry)
 
             if path.base.hasPrefix(".wh.") {
                 if path.base == ".wh..wh..opq" {  // whiteout directory
                     try self.unlink(path: path.dir, directoryWhiteout: true)
+                    if let progress {
+                        await progress([ProgressEvent(event: "add-items", value: 1)])
+                    }
                     continue
                 }
                 let startIndex = path.base.index(path.base.startIndex, offsetBy: ".wh.".count)
                 let filePath = String(path.base[startIndex...])
                 let dir: FilePath = path.dir
                 try self.unlink(path: dir.join(filePath))
+                if let progress {
+                    await progress([ProgressEvent(event: "add-items", value: 1)])
+                }
                 continue
             }
 
             if let hardlink = entry.hardlink {
                 let hl = preProcessPath(s: hardlink)
                 hardlinks[path] = FilePath(hl)
+                if let progress {
+                    await progress([ProgressEvent(event: "add-items", value: 1)])
+                }
                 continue
             }
             let ts = FileTimestamps(
@@ -84,13 +101,8 @@ extension EXT4.Formatter {
                     uid: entry.owner,
                     gid: entry.group, xattrs: entry.xattrs, fileBuffer: reusableBuffer)
 
-                // Count the size of files
                 if let progress, let size = entry.size {
-                    Task {
-                        await progress([
-                            ProgressEvent(event: "add-size", value: Int64(size))
-                        ])
-                    }
+                    await progress([ProgressEvent(event: "add-size", value: Int64(size))])
                 }
             case .symbolicLink:
                 var symlinkTarget: FilePath?
@@ -104,6 +116,10 @@ extension EXT4.Formatter {
             default:
                 continue
             }
+
+            if let progress {
+                await progress([ProgressEvent(event: "add-items", value: 1)])
+            }
         }
         guard hardlinks.acyclic else {
             throw UnpackError.circularLinks
@@ -115,13 +131,13 @@ extension EXT4.Formatter {
         }
     }
 
-    /// Unpack an archive at the source URL on to the ext4 filesystem.
+    /// Unpack an archive at the source URL on to the ext4 filesystem with progress reporting.
     public func unpack(
         source: URL,
         format: ContainerizationArchive.Format = .paxRestricted,
         compression: ContainerizationArchive.Filter = .gzip,
-        progress: ProgressHandler? = nil
-    ) throws {
+        progress: @escaping ProgressHandler
+    ) async throws {
         // For zstd, decompress once and reuse for both passes to avoid double decompression.
         let fileToRead: URL
         let readerFilter: ContainerizationArchive.Filter
@@ -141,34 +157,30 @@ extension EXT4.Formatter {
             }
         }
 
-        // Optional first pass: scan headers to get total size (fast, metadata only)
-        if let progress {
-            let sizeReader = try ArchiveReader(
-                format: format,
-                filter: readerFilter,
-                file: fileToRead
-            )
-            var totalSize: Int64 = 0
-            for (entry, _) in sizeReader.makeStreamingIterator() {
-                try Task.checkCancellation()
-                if entry.fileType == .regular, let size = entry.size {
-                    totalSize += Int64(size)
-                }
-            }
-            if totalSize > 0 {
-                Task {
-                    await progress([ProgressEvent(event: "add-total-size", value: totalSize)])
-                }
+        // First pass: scan headers to get total size (fast, metadata only)
+        let sizeReader = try ArchiveReader(
+            format: format,
+            filter: readerFilter,
+            file: fileToRead
+        )
+        var totalSize: Int64 = 0
+        for (entry, _) in sizeReader.makeStreamingIterator() {
+            try Task.checkCancellation()
+            if entry.fileType == .regular, let size = entry.size {
+                totalSize += Int64(size)
             }
         }
+        if totalSize > 0 {
+            await progress([ProgressEvent(event: "add-total-size", value: totalSize)])
+        }
 
         // Second pass: unpack
         let reader = try ArchiveReader(
             format: format,
             filter: readerFilter,
             file: fileToRead
         )
-        try self.unpack(reader: reader, progress: progress)
+        try await self.unpack(reader: reader, progress: progress)
     }
 
     private func preProcessPath(s: String) -> String {

Sources/cctl/RootfsCommand.swift

@@ -96,7 +96,7 @@ extension Application {
 
             private func outputExt4(archive: URL, to path: URL) async throws {
                 let unpacker = EXT4Unpacker(blockSizeInBytes: 256.mib())
-                try unpacker.unpack(archive: archive, compression: .gzip, at: path)
+                try await unpacker.unpack(archive: archive, compression: .gzip, at: path)
             }
 
             private func outputImage(path: URL, reference: String) async throws {

Tests/ContainerizationEXT4Tests/TestEXT4Unpacker.swift

@@ -28,7 +28,7 @@ struct Ext4UnpackerTests {
             .appendingPathComponent("ext4.unpacked.oci.img.delme", isDirectory: false))
 
     final class MockEXT4Unpacker {
-        static func Unpack(index: String, fsPath: FilePath) throws {
+        static func Unpack(index: String, fsPath: FilePath) async throws {
             let fs = try EXT4.Formatter(fsPath)
             let bundle = Bundle.module
             guard let indexPath = bundle.url(forResource: index, withExtension: nil) else {
@@ -67,14 +67,14 @@ struct Ext4UnpackerTests {
                 guard let layerPath = bundle.url(forResource: layerDigest, withExtension: nil) else {
                     throw NSError(domain: "layer \(layerDigest) not found", code: 1)
                 }
-                try fs.unpack(source: layerPath)
+                try await fs.unpack(source: layerPath)
             }
             try fs.close()
         }
     }
 
-    @Test func eXT4Unpacker() throws {
-        try MockEXT4Unpacker.Unpack(index: self.indexSHA, fsPath: self.fsPath)
+    @Test func eXT4Unpacker() async throws {
+        try await MockEXT4Unpacker.Unpack(index: self.indexSHA, fsPath: self.fsPath)
         let ext4 = try EXT4.EXT4Reader(blockDevice: self.fsPath)
         let children = try ext4.children(of: EXT4.RootInode)
         #expect(

Tests/ContainerizationEXT4Tests/TestFormatterUnpack.swift

@@ -42,10 +42,19 @@ struct Tar2EXT4Test: ~Copyable {
         "extendedattribute.test": Data([15, 26, 54, 1, 2, 4, 6, 7, 7]),
     ]
 
+    let layer1Path: URL
+    let layer2Path: URL
+
     init() throws {
-        // create layer1
+        // Compute paths before any throwing code to satisfy ~Copyable initialization rules.
         let layer1Path = FileManager.default.uniqueTemporaryDirectory()
             .appendingPathComponent("layer1.tar.gz", isDirectory: false)
+        let layer2Path = FileManager.default.uniqueTemporaryDirectory()
+            .appendingPathComponent("layer2.tar.gz", isDirectory: false)
+        self.layer1Path = layer1Path
+        self.layer2Path = layer2Path
+
+        // create layer1
         let layer1Archiver = try ArchiveWriter(
             configuration: ArchiveWriterConfiguration(format: .paxRestricted, filter: .gzip))
         try layer1Archiver.open(file: layer1Path)
@@ -57,8 +66,6 @@ struct Tar2EXT4Test: ~Copyable {
         try layer1Archiver.finishEncoding()
 
         // create layer2
-        let layer2Path = FileManager.default.uniqueTemporaryDirectory()
-            .appendingPathComponent("layer2.tar.gz", isDirectory: false)
         let layer2Archiver = try ArchiveWriter(
             configuration: ArchiveWriterConfiguration(format: .paxRestricted, filter: .gzip))
         try layer2Archiver.open(file: layer2Path)
@@ -82,18 +89,23 @@ struct Tar2EXT4Test: ~Copyable {
         // a new layer overwriting over an existing layer
         try layer2Archiver.writeEntry(entry: WriteEntry.file(path: "/dir2/file1", permissions: 0o644), data: nil)
         try layer2Archiver.finishEncoding()
+    }
 
-        let unpacker = try EXT4.Formatter(fsPath)
-        try unpacker.unpack(source: layer1Path)
-        try unpacker.unpack(source: layer2Path)
-        try unpacker.close()
+    private func unpackLayers() async throws {
+        let formatter = try EXT4.Formatter(fsPath)
+        try await formatter.unpack(source: layer1Path)
+        try await formatter.unpack(source: layer2Path)
+        try formatter.close()
     }
 
     deinit {
         try? FileManager.default.removeItem(at: fsPath.url)
+        try? FileManager.default.removeItem(at: layer1Path.deletingLastPathComponent())
+        try? FileManager.default.removeItem(at: layer2Path.deletingLastPathComponent())
     }
 
-    @Test func testUnpackBasic() throws {
+    @Test func testUnpackBasic() async throws {
+        try await unpackLayers()
         let ext4 = try EXT4.EXT4Reader(blockDevice: fsPath)
         // just a directory
         let dir1Inode = try ext4.getInode(number: 12)
@@ -196,12 +208,9 @@ struct UnpackProgressTest {
 
         // Unpack with progress tracking
         let formatter = try EXT4.Formatter(fsPath)
-        try formatter.unpack(source: archivePath, progress: progressHandler)
+        try await formatter.unpack(source: archivePath, progress: progressHandler)
         try formatter.close()
 
-        // Allow async progress tasks to complete
-        try await Task.sleep(for: .milliseconds(100))
-
         // Analyze collected events
         let allEvents = await collector.allEvents()
 
@@ -292,7 +301,7 @@ struct UnpackProgressTest {
         }
     }
 
-    @Test func progressHandlerIsOptional() throws {
+    @Test func progressHandlerIsOptional() async throws {
         // Verify that unpacking works without a progress handler (existing behavior)
         let tempDir = FileManager.default.uniqueTemporaryDirectory()
         let archivePath = tempDir.appendingPathComponent("test.tar.gz", isDirectory: false)
@@ -314,7 +323,7 @@ struct UnpackProgressTest {
 
         // Unpack without progress handler - should not throw
         let formatter = try EXT4.Formatter(fsPath)
-        try formatter.unpack(source: archivePath)
+        try await formatter.unpack(source: archivePath)
         try formatter.close()
 
         // Verify the file was unpacked correctly

Tests/ContainerizationTests/ImageTests/HeaderScanTimingTests.swift

@@ -69,7 +69,7 @@ struct ImageHeaderScanTimingTest {
                 let compression = compressionFilter(for: layer.mediaType)
                 let compressedSize = try FileManager.default.attributesOfItem(atPath: content.path.path)[.size] as? Int64 ?? 0
                 let label = "\(image.label) layer \(i + 1)/\(manifest.layers.count) (\(layer.mediaType), \(formatBytes(compressedSize)) compressed)"
-                try measureOverhead(url: content.path, compression: compression, label: label)
+                try await measureOverhead(url: content.path, compression: compression, label: label)
             }
         }
     }
@@ -87,7 +87,7 @@ struct ImageHeaderScanTimingTest {
         }
     }
 
-    private func measureOverhead(url: URL, compression: ContainerizationArchive.Filter, label: String) throws {
+    private func measureOverhead(url: URL, compression: ContainerizationArchive.Filter, label: String) async throws {
         let clock = ContinuousClock()
 
         print("\n--- \(label) ---\n")
@@ -133,9 +133,9 @@ struct ImageHeaderScanTimingTest {
         let fsPath1 = FilePath(tempDir1.appendingPathComponent("no-progress.ext4.img", isDirectory: false))
         defer { try? FileManager.default.removeItem(at: tempDir1) }
 
-        let unpackOnlyDuration = try clock.measure {
+        let unpackOnlyDuration = try await clock.measure {
             let formatter = try EXT4.Formatter(fsPath1)
-            try formatter.unpack(source: url, compression: compression)
+            try await formatter.unpack(source: url, compression: compression)
             try formatter.close()
         }
         print("  Unpack (no progress): \(unpackOnlyDuration)")
@@ -146,9 +146,9 @@ struct ImageHeaderScanTimingTest {
         defer { try? FileManager.default.removeItem(at: tempDir2) }
 
         let noopProgress: ProgressHandler = { _ in }
-        let withProgressDuration = try clock.measure {
+        let withProgressDuration = try await clock.measure {
             let formatter = try EXT4.Formatter(fsPath2)
-            try formatter.unpack(source: url, compression: compression, progress: noopProgress)
+            try await formatter.unpack(source: url, compression: compression, progress: noopProgress)
             try formatter.close()
         }
         print("  Unpack (w/ progress): \(withProgressDuration)")