Reclaim the snapshot a record stops naming

Sources/Services/ContainerImagesService/Server/ImagesService.swift

@@ -32,6 +32,17 @@ public actor ImagesService {
     private let imageStore: ImageStore
     private let snapshotStore: SnapshotStore
 
+    /// One landing, removal, or sweep at a time. The store operations
+    /// suspend mid-flight, and the actor admits other calls at every
+    /// suspension, so a garbage collection overlapping a landing computes
+    /// its keep set without the arriving image and prunes blobs and
+    /// snapshots the landing is about to reference. The lock is held
+    /// across each whole mutating operation, the way PodsService holds
+    /// its lock across lifecycle operations; containerd guards the same
+    /// window with leases.
+    /// https://github.com/containerd/containerd/blob/main/docs/garbage-collection.md
+    private let lock = AsyncLock()
+
     public init(
         contentStore: ContentStore,
         imageStore: ImageStore,
@@ -60,6 +71,39 @@ public actor ImagesService {
         return exists
     }
 
+    /// Sweep the snapshot directories no current image claims.
+    ///
+    /// An unpacked snapshot is owned by the record that names it, so a record
+    /// that stops naming it leaves it owned by nobody: registering an image
+    /// over an existing reference replaces the record, and removing a
+    /// reference takes it away. Only the store sees either happen, so every
+    /// path that lands or removes records runs this sweep afterwards. The
+    /// sweep is housekeeping: its failure is loud in the log and never fails
+    /// the operation that carried it.
+    private func sweepReplacedSnapshots(after function: String = #function) async {
+        do {
+            let kept = try await self._list()
+            let freedSnapshotBytes = try await self.snapshotStore.clean(keepingSnapshotsFor: kept)
+            if freedSnapshotBytes > 0 {
+                self.log.debug(
+                    "ImagesService: swept snapshots of replaced images",
+                    metadata: [
+                        "func": "\(function)",
+                        "freedBytes": "\(freedSnapshotBytes)",
+                    ]
+                )
+            }
+        } catch {
+            self.log.warning(
+                "ImagesService: snapshot sweep failed",
+                metadata: [
+                    "func": "\(function)",
+                    "error": "\(error)",
+                ]
+            )
+        }
+    }
+
     public func list() async throws -> [ImageDescription] {
         self.log.debug(
             "ImagesService: enter",
@@ -81,6 +125,16 @@ public actor ImagesService {
 
     public func pull(reference: String, platform: Platform?, insecure: Bool, progressUpdate: ProgressUpdateHandler?, maxConcurrentDownloads: Int = 3) async throws
         -> ImageDescription
+    {
+        try await lock.withLock { _ in
+            try await self._pull(
+                reference: reference, platform: platform, insecure: insecure,
+                progressUpdate: progressUpdate, maxConcurrentDownloads: maxConcurrentDownloads)
+        }
+    }
+
+    private func _pull(reference: String, platform: Platform?, insecure: Bool, progressUpdate: ProgressUpdateHandler?, maxConcurrentDownloads: Int) async throws
+        -> ImageDescription
     {
         self.log.debug(
             "ImagesService: enter",
@@ -111,10 +165,17 @@ public actor ImagesService {
         guard let img else {
             throw ContainerizationError(.internalError, message: "failed to pull image \(reference)")
         }
+        await self.sweepReplacedSnapshots()
         return img.description.fromCZ
     }
 
     public func push(reference: String, platform: Platform?, insecure: Bool, progressUpdate: ProgressUpdateHandler?) async throws {
+        try await lock.withLock { _ in
+            try await self._push(reference: reference, platform: platform, insecure: insecure, progressUpdate: progressUpdate)
+        }
+    }
+
+    private func _push(reference: String, platform: Platform?, insecure: Bool, progressUpdate: ProgressUpdateHandler?) async throws {
         self.log.debug(
             "ImagesService: enter",
             metadata: [
@@ -142,6 +203,12 @@ public actor ImagesService {
     }
 
     public func tag(old: String, new: String) async throws -> ImageDescription {
+        try await lock.withLock { _ in
+            try await self._tag(old: old, new: new)
+        }
+    }
+
+    private func _tag(old: String, new: String) async throws -> ImageDescription {
         self.log.debug(
             "ImagesService: enter",
             metadata: [
@@ -166,6 +233,12 @@ public actor ImagesService {
     }
 
     public func delete(reference: String, garbageCollect: Bool) async throws {
+        try await lock.withLock { _ in
+            try await self._delete(reference: reference, garbageCollect: garbageCollect)
+        }
+    }
+
+    private func _delete(reference: String, garbageCollect: Bool) async throws {
         self.log.debug(
             "ImagesService: enter",
             metadata: [
@@ -184,9 +257,16 @@ public actor ImagesService {
         }
 
         try await self.imageStore.delete(reference: reference, performCleanup: garbageCollect)
+        await self.sweepReplacedSnapshots()
     }
 
     public func save(references: [String], out: URL, platform: Platform?) async throws {
+        try await lock.withLock { _ in
+            try await self._save(references: references, out: out, platform: platform)
+        }
+    }
+
+    private func _save(references: [String], out: URL, platform: Platform?) async throws {
         self.log.debug(
             "ImagesService: enter",
             metadata: [
@@ -215,6 +295,12 @@ public actor ImagesService {
     }
 
     public func load(from tarFile: URL, force: Bool) async throws -> ([ImageDescription], [String]) {
+        try await lock.withLock { _ in
+            try await self._load(from: tarFile, force: force)
+        }
+    }
+
+    private func _load(from tarFile: URL, force: Bool) async throws -> ([ImageDescription], [String]) {
         let archivePathname = tarFile.absolutePath()
         self.log.debug(
             "ImagesService: enter",
@@ -244,6 +330,7 @@ public actor ImagesService {
         }
 
         let loaded = try await self.imageStore.load(from: tempDir)
+        await self.sweepReplacedSnapshots()
         var images: [ImageDescription] = []
         for image in loaded {
             images.append(image.description.fromCZ)
@@ -252,6 +339,12 @@ public actor ImagesService {
     }
 
     public func cleanUpOrphanedBlobs() async throws -> ([String], UInt64) {
+        try await lock.withLock { _ in
+            try await self._cleanUpOrphanedBlobs()
+        }
+    }
+
+    private func _cleanUpOrphanedBlobs() async throws -> ([String], UInt64) {
         self.log.debug(
             "ImagesService: enter",
             metadata: [
@@ -336,6 +429,12 @@ public actor ImagesService {
 
 extension ImagesService {
     public func unpack(description: ImageDescription, platform: Platform?, progressUpdate: ProgressUpdateHandler?) async throws {
+        try await lock.withLock { _ in
+            try await self._unpack(description: description, platform: platform, progressUpdate: progressUpdate)
+        }
+    }
+
+    private func _unpack(description: ImageDescription, platform: Platform?, progressUpdate: ProgressUpdateHandler?) async throws {
         self.log.debug(
             "ImagesService: enter",
             metadata: [
@@ -360,6 +459,12 @@ extension ImagesService {
     }
 
     public func deleteImageSnapshot(description: ImageDescription, platform: Platform?) async throws {
+        try await lock.withLock { _ in
+            try await self._deleteImageSnapshot(description: description, platform: platform)
+        }
+    }
+
+    private func _deleteImageSnapshot(description: ImageDescription, platform: Platform?) async throws {
         self.log.debug(
             "ImagesService: enter",
             metadata: [
@@ -384,6 +489,12 @@ extension ImagesService {
     }
 
     public func getImageSnapshot(description: ImageDescription, platform: Platform) async throws -> Filesystem {
+        try await lock.withLock { _ in
+            try await self._getImageSnapshot(description: description, platform: platform)
+        }
+    }
+
+    private func _getImageSnapshot(description: ImageDescription, platform: Platform) async throws -> Filesystem {
         self.log.debug(
             "ImagesService: enter",
             metadata: [

Sources/Services/ContainerImagesService/Server/SnapshotStore.swift

@@ -168,12 +168,32 @@ public actor SnapshotStore {
     public func clean(keepingSnapshotsFor images: [Containerization.Image] = []) async throws -> UInt64 {
         var toKeep: [String] = [Self.ingestDirName]
         for image in images {
-            for manifest in try await image.index().manifests {
-                guard let platform = manifest.platform else {
-                    continue
+            // An image is read to learn which snapshot it claims, and reading
+            // one whose content the store no longer holds fails. That is a
+            // fact about that image and not about the others, so it costs the
+            // sweep that image rather than the whole pass: a single record
+            // left behind by an interrupted pull would otherwise stop every
+            // snapshot from ever being reclaimed again, and the store fills
+            // with the unpacked filesystems of images nothing names. An image
+            // the store cannot read is one it cannot unpack or run either, so
+            // the snapshot that image would have claimed is not worth keeping
+            // the store full for.
+            do {
+                for manifest in try await image.index().manifests {
+                    guard let platform = manifest.platform else {
+                        continue
+                    }
+                    let desc = try await image.descriptor(for: platform)
+                    toKeep.append(desc.digest.trimmingDigestPrefix)
                 }
-                let desc = try await image.descriptor(for: platform)
-                toKeep.append(desc.digest.trimmingDigestPrefix)
+            } catch {
+                self.log?.warning(
+                    "snapshot sweep read no manifests for an image; its snapshots are not kept",
+                    metadata: [
+                        "image": "\(image.reference)",
+                        "error": "\(error)",
+                    ]
+                )
             }
         }
         let all = try self.fm.contentsOfDirectory(at: self.path, includingPropertiesForKeys: [.totalFileAllocatedSizeKey]).map {
@@ -186,8 +206,24 @@ public actor SnapshotStore {
             guard self.fm.fileExists(atPath: unpackedPath.absolutePath()) else {
                 continue
             }
-            deletedBytes += self.fm.allocatedSize(of: unpackedPath)
-            try self.fm.removeItem(at: unpackedPath)
+            let size = self.fm.allocatedSize(of: unpackedPath)
+            // A snapshot the store cannot remove costs one snapshot's worth of
+            // space, and every other snapshot in the set is still owed to the
+            // caller, as are the bytes already freed. The removal that failed
+            // is what the caller needs to act on, so the error goes to the log
+            // with the snapshot that raised it.
+            do {
+                try self.fm.removeItem(at: unpackedPath)
+                deletedBytes += size
+            } catch {
+                self.log?.warning(
+                    "snapshot sweep could not remove a snapshot",
+                    metadata: [
+                        "snapshot": "\(dir)",
+                        "error": "\(error)",
+                    ]
+                )
+            }
         }
         return deletedBytes
     }