Make stacked content pruning reference-aware (#1313)

Sources/tart/Commands/Prune.swift

@@ -81,27 +81,34 @@ struct Prune: AsyncParsableCommand {
   }
 
   static func pruneSpaceBudget(prunableStorages: [PrunableStorage], spaceBudgetBytes: UInt64) throws {
-    let prunables: [Prunable] = try prunableStorages
-      .flatMap { try $0.prunables() }
-      .sorted { try $0.accessDate() > $1.accessDate() }
+    while true {
+      let prunables: [Prunable] = try prunableStorages
+        .flatMap { try $0.prunables() }
+        .sorted { try $0.accessDate() > $1.accessDate() }
 
-    var spaceBudgetBytes = spaceBudgetBytes
-    var prunablesToDelete: [Prunable] = []
+      var remainingBudgetBytes = spaceBudgetBytes
+      var prunableToDelete: Prunable?
 
-    for prunable in prunables {
-      let prunableSizeBytes = UInt64(try prunable.allocatedSizeBytes())
+      for prunable in prunables {
+        let prunableSizeBytes = UInt64(try prunable.allocatedSizeBytes())
 
-      if prunableSizeBytes <= spaceBudgetBytes {
-        // Don't mark for deletion as
-        // there's a budget available
-        spaceBudgetBytes -= prunableSizeBytes
-      } else {
-        // Mark for deletion
-        prunablesToDelete.append(prunable)
+        if prunableSizeBytes <= remainingBudgetBytes {
+          // Don't mark for deletion as there is budget available
+          remainingBudgetBytes -= prunableSizeBytes
+        } else {
+          prunableToDelete = prunable
+          break
+        }
       }
-    }
 
-    try prunablesToDelete.forEach { try $0.delete() }
+      guard let prunableToDelete else {
+        return
+      }
+
+      // Deleting one cached stacked image can change which remaining image
+      // owns shared immutable content. Rebuild before choosing another.
+      try prunableToDelete.delete()
+    }
   }
 
   static func reclaimIfNeeded(_ requiredBytes: UInt64, _ initiator: Prunable? = nil) throws {
@@ -145,46 +152,51 @@ struct Prune: AsyncParsableCommand {
     try Prune.reclaimIfPossible(requiredBytes - volumeAvailableCapacityCalculated, initiator)
   }
 
-  private static func reclaimIfPossible(_ reclaimBytes: UInt64, _ initiator: Prunable? = nil) throws {
+  static func reclaimIfPossible(_ reclaimBytes: UInt64, _ initiator: Prunable? = nil) throws {
     let span = OTel.shared.tracer.spanBuilder(spanName: "prune").startSpan()
     defer { span.end() }
 
     let prunableStorages: [PrunableStorage] = [try VMStorageOCI(), try IPSWCache()]
-    let prunables: [Prunable] = try prunableStorages
-      .flatMap { try $0.prunables() }
-      .sorted { try $0.accessDate() < $1.accessDate() }
+    let prunables = {
+      try prunableStorages
+        .flatMap { try $0.prunables() }
+        .sorted { try $0.accessDate() < $1.accessDate() }
+    }
 
     // Does it even make sense to start?
-    let cacheUsedBytes = try prunables.map { try $0.allocatedSizeBytes() }.reduce(0, +)
-    if cacheUsedBytes < reclaimBytes {
+    let initialPrunables = try prunables()
+    let initialCacheUsedBytes = try initialPrunables.map { try $0.allocatedSizeBytes() }.reduce(0, +)
+    guard let reclaimBytes = Int(exactly: reclaimBytes), initialCacheUsedBytes >= reclaimBytes else {
       return
     }
 
-    var cacheReclaimedBytes: Int = 0
+    let targetCacheUsedBytes = initialCacheUsedBytes - reclaimBytes
+    var currentCacheUsedBytes = initialCacheUsedBytes
+    let initiatorPath = initiator.map {
+      $0.url.resolvingSymlinksInPath().standardizedFileURL.path
+    }
 
-    var it = prunables.makeIterator()
-
-    while cacheReclaimedBytes <= reclaimBytes {
-      guard let prunable = it.next() else {
+    while currentCacheUsedBytes > targetCacheUsedBytes {
+      // Deleting one cached stacked image can transfer ownership of shared
+      // immutable content to another record without reclaiming those bytes.
+      // Rebuild the candidates after every deletion so automatic pruning
+      // measures the cache that remains rather than a stale ownership snapshot.
+      guard let prunable = try prunables().first(where: {
+        $0.url.resolvingSymlinksInPath().standardizedFileURL.path != initiatorPath
+      }) else {
         break
       }
 
-      if prunable.url == initiator?.url.resolvingSymlinksInPath() {
-        // do not prune the initiator
-        continue
-      }
-
       let allocatedSizeBytes = try prunable.allocatedSizeBytes()
 
       OpenTelemetry.instance.contextProvider.activeSpan?
         .addEvent(name: "Pruned \(allocatedSizeBytes) bytes for \(prunable.url.path)")
 
-      cacheReclaimedBytes += allocatedSizeBytes
-
       try prunable.delete()
+      currentCacheUsedBytes = try prunables().map { try $0.allocatedSizeBytes() }.reduce(0, +)
     }
 
     OpenTelemetry.instance.contextProvider.activeSpan?
-      .addEvent(name: "Reclaimed \(cacheReclaimedBytes) bytes")
+      .addEvent(name: "Reclaimed \(initialCacheUsedBytes - currentCacheUsedBytes) bytes")
   }
 }

Sources/tart/Commands/Run.swift

@@ -1041,20 +1041,24 @@ struct AdditionalDisk {
         // A cached stacked image has no writable top overlay. Create one in a
         // disposable directory for this additional-disk attachment.
         let temporaryVMDir = try VMDirectory.temporary()
-        try FileManager.default.copyItem(at: vmDir.configURL, to: temporaryVMDir.configURL)
-        try FileManager.default.copyItem(at: vmDir.nvramURL, to: temporaryVMDir.nvramURL)
-        try FileManager.default.copyItem(at: vmDir.manifestURL, to: temporaryVMDir.manifestURL)
-        let lock = try FileLock(lockURL: temporaryVMDir.baseURL)
-        try lock.lock()
+        let temporaryVMDirLock = try FileLock(lockURL: temporaryVMDir.baseURL)
+        try temporaryVMDirLock.lock()
+        try vmDir.cloneStacked(
+          to: temporaryVMDir,
+          copyWritableOverlay: false,
+          generateMAC: false
+        )
         let stack = try temporaryVMDir.diskImageStack()
-        try stack.createWritableOverlay()
         let attachment = try stack.makeAttachment(
           readOnly: diskReadOnly,
           cachingMode: try VZDiskImageCachingMode(cachingModeRaw) ?? .automatic,
           synchronizationMode: try VZDiskImageSynchronizationMode(syncModeRaw)
         )
 
-        return AdditionalDisk(configuration: VZVirtioBlockDeviceConfiguration(attachment: attachment), temporaryDiskLock: lock)
+        return AdditionalDisk(
+          configuration: VZVirtioBlockDeviceConfiguration(attachment: attachment),
+          temporaryDiskLock: temporaryVMDirLock
+        )
       }
 
       // Unfortunately, VZDiskImageStorageDeviceAttachment does not support

Sources/tart/Config.swift

@@ -33,7 +33,7 @@ struct Config {
         continue
       }
 
-      try FileManager.default.removeItem(at: entry)
+      try VMDirectory(baseURL: entry).removeFromDisk()
 
       try lock.unlock()
     }

Sources/tart/ContentStore.swift

@@ -16,6 +16,7 @@ struct ContentStore {
 
   let baseURL: URL
   private let digestDirectoryURL: URL
+  private let pruneLockURL: URL
 
   init() throws {
     try self.init(baseURL: Config().tartCacheDir.appendingPathComponent("content", isDirectory: true))
@@ -24,13 +25,41 @@ struct ContentStore {
   init(baseURL: URL) throws {
     self.baseURL = baseURL
     self.digestDirectoryURL = baseURL.appendingPathComponent(Self.digestAlgorithm, isDirectory: true)
+    self.pruneLockURL = baseURL.appendingPathComponent(".gc.lock")
     try FileManager.default.createDirectory(at: digestDirectoryURL, withIntermediateDirectories: true)
+    if !FileManager.default.fileExists(atPath: pruneLockURL.path) {
+      _ = FileManager.default.createFile(atPath: pruneLockURL.path, contents: Data())
+    }
+  }
+
+  /// Serializes reference publication with the final reference check and
+  /// deletion of immutable cache entries across Tart processes.
+  func withPruneLock<T>(_ body: () throws -> T) throws -> T {
+    let lock = try FileLock(lockURL: pruneLockURL)
+    try lock.lock()
+    defer { try? lock.unlock() }
+
+    return try body()
+  }
+
+  /// Waits for any prune already scanning references to finish. After this
+  /// returns, later prune runs can see a reference the caller already wrote.
+  func synchronizePublishedReferences() throws {
+    try withPruneLock {}
   }
 
   func contentURL(for contentDigest: String) throws -> URL {
+    try contentURL(for: contentDigest, under: baseURL)
+  }
+
+  /// Returns the canonical path for a digest under an arbitrary content-store
+  /// root without creating directories or lock files.
+  func contentURL(for contentDigest: String, under baseURL: URL) throws -> URL {
     let digestHex = try validatedDigestHex(contentDigest)
 
-    return digestDirectoryURL.appendingPathComponent(digestHex)
+    return baseURL
+      .appendingPathComponent(Self.digestAlgorithm, isDirectory: true)
+      .appendingPathComponent(digestHex)
   }
 
   func temporaryContentURL(for contentDigest: String) throws -> URL {
@@ -61,9 +90,9 @@ struct ContentStore {
     return lockURL
   }
 
-  /// Returns a digest-addressed entry without rereading it. Pull verifies
-  /// content hashes before accepting a cache hit; clone only needs a cheap
-  /// structural check, like Tart's existing disk.img path.
+  /// Returns an immutable digest-addressed entry without rereading it. Files
+  /// are verified when installed and when deciding whether a pull is a cache
+  /// hit; normal clone/run/push paths trust the store like Tart's disk.img.
   func contentURLIfPresent(for contentDigest: String) throws -> URL? {
     let url = try contentURL(for: contentDigest)
 
@@ -71,6 +100,8 @@ struct ContentStore {
       return nil
     }
 
+    try url.updateAccessDate()
+
     return url
   }
 
@@ -88,6 +119,33 @@ struct ContentStore {
     return url
   }
 
+  /// Returns immutable content files that no retained cached image or local VM
+  /// references. Callers may prune these like other cache entries.
+  func prunables(excluding referencedContentDigests: Swift.Set<String>) throws -> [URL] {
+    guard let enumerator = FileManager.default.enumerator(
+      at: digestDirectoryURL,
+      includingPropertiesForKeys: [.isRegularFileKey],
+      options: [.skipsSubdirectoryDescendants]
+    ) else {
+      return []
+    }
+
+    return try enumerator.compactMap { element in
+      guard let url = element as? URL,
+            try url.resourceValues(forKeys: [.isRegularFileKey]).isRegularFile == true else {
+        return nil
+      }
+
+      let contentDigest = "\(Self.digestPrefix)\(url.lastPathComponent)"
+      guard (try? validatedDigestHex(contentDigest)) != nil,
+            !referencedContentDigests.contains(contentDigest) else {
+        return nil
+      }
+
+      return url
+    }
+  }
+
   /// Move a fully reconstructed temporary file into the cache after verifying
   /// its semantic identity. The caller should create the temporary file with
   /// temporaryContentURL(for:) or resumableContentURL(for:) so rename stays on

Sources/tart/OCI/Manifest.swift

@@ -37,6 +37,24 @@ struct TartDiskFileGroup: Equatable {
   /// Whole reconstructed-file digest. Existing flat manifests do not have
   /// this until a macOS 27 clone normalizes its local manifest copy.
   var contentDigest: String?
+
+  /// Expected size of the complete disk file reconstructed from these chunks.
+  func uncompressedSize() -> UInt64? {
+    var result: UInt64 = 0
+    for chunk in chunks {
+      guard let size = chunk.uncompressedSize() else {
+        return nil
+      }
+
+      let addition = result.addingReportingOverflow(size)
+      guard !addition.overflow else {
+        return nil
+      }
+      result = addition.partialValue
+    }
+
+    return result
+  }
 }
 
 enum TartDiskRepresentation: Equatable {
@@ -172,6 +190,16 @@ struct OCIManifest: Codable, Equatable {
     return .stacked(base: base, overlays: overlays)
   }
 
+  /// Returns content-store digests needed to reconstruct this disk stack.
+  func diskContentDigests() throws -> [String] {
+    switch try tartDiskRepresentation() {
+    case .flat(let base):
+      return base.contentDigest.map { [$0] } ?? []
+    case .stacked(let base, let overlays):
+      return ([base] + overlays).compactMap(\.contentDigest)
+    }
+  }
+
   private func validateChunkMetadata(_ chunks: [OCIManifestLayer]) throws {
     guard chunks.allSatisfy({ $0.uncompressedSize() != nil && $0.uncompressedContentDigest() != nil }) else {
       throw OCIManifestValidationError.invalidDiskMetadata("disk chunks need uncompressed size and content digest")

Sources/tart/VMDirectory+DiskImageStack.swift

@@ -1,6 +1,12 @@
 import Foundation
 
 extension VMDirectory {
+  /// Returns content-store digests needed to reconstruct this VM's disk stack.
+  func diskContentDigests() throws -> [String] {
+    let manifest = try OCIManifest(fromJSON: Data(contentsOf: manifestURL))
+    return try manifest.diskContentDigests()
+  }
+
   func diskImageStack(contentStore providedStore: ContentStore? = nil) throws -> DiskImageStack {
     let manifest = try OCIManifest(fromJSON: Data(contentsOf: manifestURL))
     let base: TartDiskFileGroup
@@ -42,13 +48,18 @@ extension VMDirectory {
     generateMAC: Bool,
     contentStore: ContentStore? = nil
   ) throws {
-    try FileManager.default.copyItem(at: configURL, to: destination.configURL)
-    try FileManager.default.copyItem(at: nvramURL, to: destination.nvramURL)
-    try FileManager.default.copyItem(at: manifestURL, to: destination.manifestURL)
+    let contentStore = try contentStore ?? ContentStore()
+    try contentStore.withPruneLock {
+      try FileManager.default.copyItem(at: configURL, to: destination.configURL)
+      try FileManager.default.copyItem(at: nvramURL, to: destination.nvramURL)
+      try FileManager.default.copyItem(at: manifestURL, to: destination.manifestURL)
 
-    if copyWritableOverlay {
-      try FileManager.default.copyItem(at: overlayURL, to: destination.overlayURL)
-    } else {
+      if copyWritableOverlay {
+        try FileManager.default.copyItem(at: overlayURL, to: destination.overlayURL)
+      }
+    }
+
+    if !copyWritableOverlay {
       try destination.diskImageStack(contentStore: contentStore).createWritableOverlay()
     }
 
@@ -67,17 +78,6 @@ extension VMDirectory {
     let contentDigest = try Digest.hash(diskURL)
     let contentStore = try providedStore ?? ContentStore()
 
-    if try contentStore.existingContentURL(for: contentDigest) == nil {
-      let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
-      do {
-        try FileManager.default.copyItem(at: diskURL, to: temporaryURL)
-        _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
-      } catch {
-        try? FileManager.default.removeItem(at: temporaryURL)
-        throw error
-      }
-    }
-
     var manifest = try OCIManifest(fromJSON: Data(contentsOf: manifestURL))
     guard case .flat = try manifest.tartDiskRepresentation() else {
       throw RuntimeError.VMConfigurationError("--stacked cannot use an image that already has a stacked disk")
@@ -101,7 +101,24 @@ extension VMDirectory {
 
     try FileManager.default.copyItem(at: configURL, to: destination.configURL)
     try FileManager.default.copyItem(at: nvramURL, to: destination.nvramURL)
-    try manifest.toJSON().write(to: destination.manifestURL)
+    try contentStore.withPruneLock {
+      try manifest.toJSON().write(to: destination.manifestURL)
+    }
+
+    // Publish the temporary VM's manifest before installing the shared base.
+    // Reference-aware pruning includes in-progress manifests, so the content
+    // cannot be collected in the window before this VM is moved into place.
+    if try contentStore.contentURLIfPresent(for: contentDigest) == nil {
+      let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
+      do {
+        try FileManager.default.copyItem(at: diskURL, to: temporaryURL)
+        _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
+      } catch {
+        try? FileManager.default.removeItem(at: temporaryURL)
+        throw error
+      }
+    }
+
     try destination.diskImageStack(contentStore: contentStore).createWritableOverlay()
 
     if generateMAC {

Sources/tart/VMDirectory.swift

@@ -373,11 +373,27 @@ struct VMDirectory: Prunable {
       throw RuntimeError.VMIsRunning(name)
     }
 
-    try FileManager.default.removeItem(at: baseURL)
+    // Standalone local VMs do not reference the shared content store. Delete
+    // them directly so a full disk can still be recovered before the content
+    // store has ever been initialized.
+    if isStandalone {
+      try FileManager.default.removeItem(at: baseURL)
+    } else {
+      try removeFromDisk()
+    }
 
     try lock.unlock()
   }
 
+  /// Removes a VM directory while preserving the content-store reference
+  /// protocol for any complete or partially published manifest it contains.
+  func removeFromDisk() throws {
+    let contentStore = try ContentStore()
+    try contentStore.withPruneLock {
+      try FileManager.default.removeItem(at: baseURL)
+    }
+  }
+
   func accessDate() throws -> Date {
     try baseURL.accessDate()
   }

Sources/tart/VMStorageLocal.swift

@@ -35,11 +35,27 @@ class VMStorageLocal: PrunableStorage {
 
   func move(_ name: String, from: VMDirectory) throws {
     _ = try FileManager.default.createDirectory(at: baseURL, withIntermediateDirectories: true)
-    _ = try FileManager.default.replaceItemAt(vmURL(name), withItemAt: from.baseURL)
+    try replace(VMDirectory(baseURL: vmURL(name)), with: from)
   }
 
   func rename(_ name: String, _ newName: String) throws {
-    _ = try FileManager.default.replaceItemAt(vmURL(newName), withItemAt: vmURL(name))
+    let source = VMDirectory(baseURL: vmURL(name))
+    let destination = VMDirectory(baseURL: vmURL(newName))
+    try replace(destination, with: source)
+  }
+
+  /// References in a manifest must not disappear while content GC is deciding
+  /// whether their immutable disk files are still in use.
+  private func replace(_ destination: VMDirectory, with source: VMDirectory) throws {
+    if FileManager.default.fileExists(atPath: source.manifestURL.path) ||
+      FileManager.default.fileExists(atPath: destination.manifestURL.path) {
+      let contentStore = try ContentStore()
+      try contentStore.withPruneLock {
+        _ = try FileManager.default.replaceItemAt(destination.baseURL, withItemAt: source.baseURL)
+      }
+    } else {
+      _ = try FileManager.default.replaceItemAt(destination.baseURL, withItemAt: source.baseURL)
+    }
   }
 
   func delete(_ name: String) throws {

Sources/tart/VMStorageOCI.swift

@@ -22,9 +22,8 @@ class VMStorageOCI: PrunableStorage {
   }
 
   /// Whether clone can use a cached image without pulling. Standalone images keep
-  /// Tart's existing structural check. Stacked cached images cheaply require every
-  /// immutable file with its expected length; explicit pull remains the path
-  /// that hashes content and repairs same-sized corruption.
+  /// Tart's existing structural check. Stacked cached images require every
+  /// immutable file with its expected length.
   func hasUsableCachedImageForClone(_ name: RemoteName, requireManifest: Bool = false) throws -> Bool {
     guard exists(name) else {
       return false
@@ -45,25 +44,7 @@ class VMStorageOCI: PrunableStorage {
 
     let contentStore = try ContentStore()
     for group in [base] + overlays {
-      guard let contentDigest = group.contentDigest,
-            let contentURL = try contentStore.contentURLIfPresent(for: contentDigest) else {
-        return false
-      }
-
-      var expectedSize: UInt64 = 0
-      for chunk in group.chunks {
-        guard let uncompressedSize = chunk.uncompressedSize() else {
-          return false
-        }
-        let addition = expectedSize.addingReportingOverflow(uncompressedSize)
-        guard !addition.overflow else {
-          return false
-        }
-        expectedSize = addition.partialValue
-      }
-
-      guard let actualSize = UInt64(exactly: try contentURL.sizeBytes()),
-            actualSize == expectedSize else {
+      guard try hasUsableCachedDiskFile(group, contentStore: contentStore) else {
         return false
       }
     }
@@ -164,35 +145,44 @@ class VMStorageOCI: PrunableStorage {
 
     let vmDir = try create(name, overwrite: exists(name))
 
-    if source.isStackedVM {
-      guard case .stacked(_, let overlays) = try manifest.tartDiskRepresentation(),
-            let contentDigest = overlays.last?.contentDigest else {
-        throw RuntimeError.VMConfigurationError("pushed image is missing its writable ASIF overlay")
-      }
-
-      // The pushed top overlay becomes immutable in the cached image. Keep a
-      // semantic copy so later clones do not need to fetch it back.
-      let contentStore = try ContentStore()
-      if try contentStore.existingContentURL(for: contentDigest) == nil {
-        let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
-        do {
-          try FileManager.default.copyItem(at: source.overlayURL, to: temporaryURL)
-          _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
-        } catch {
-          try? FileManager.default.removeItem(at: temporaryURL)
-          throw error
+    do {
+      if source.isStackedVM {
+        guard case .stacked(_, let overlays) = try manifest.tartDiskRepresentation(),
+              let contentDigest = overlays.last?.contentDigest else {
+          throw RuntimeError.VMConfigurationError("pushed image is missing its writable ASIF overlay")
         }
+
+        // The pushed top overlay becomes immutable in the cached image. Keep a
+        // semantic copy so later clones do not need to fetch it back.
+        let contentStore = try ContentStore()
+        try contentStore.withPruneLock {
+          try FileManager.default.copyItem(at: source.configURL, to: vmDir.configURL)
+          try FileManager.default.copyItem(at: source.nvramURL, to: vmDir.nvramURL)
+          // Publish the reference before installing the immutable top overlay,
+          // so reference-aware pruning cannot collect it in between.
+          try manifest.toJSON().write(to: vmDir.manifestURL)
+        }
+
+        if try contentStore.contentURLIfPresent(for: contentDigest) == nil {
+          let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
+          do {
+            try FileManager.default.copyItem(at: source.overlayURL, to: temporaryURL)
+            _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
+          } catch {
+            try? FileManager.default.removeItem(at: temporaryURL)
+            throw error
+          }
+        }
+      } else {
+        try source.clone(to: vmDir, generateMAC: false)
+        // Keep the exact manifest Tart submitted so tag links and later pushes
+        // refer to the same digest-addressed cached image.
+        try manifest.toJSON().write(to: vmDir.manifestURL)
       }
-
-      try FileManager.default.copyItem(at: source.configURL, to: vmDir.configURL)
-      try FileManager.default.copyItem(at: source.nvramURL, to: vmDir.nvramURL)
-    } else {
-      try source.clone(to: vmDir, generateMAC: false)
+    } catch {
+      try? vmDir.removeFromDisk()
+      throw error
     }
-
-    // Keep the exact manifest Tart submitted so tag links and later pushes
-    // refer to the same digest-addressed cached image.
-    try manifest.toJSON().write(to: vmDir.manifestURL)
   }
 
   func move(_ name: RemoteName, from: VMDirectory) throws{
@@ -203,11 +193,20 @@ class VMStorageOCI: PrunableStorage {
     try FileManager.default.createDirectory(at: targetURL.deletingLastPathComponent(),
                                             withIntermediateDirectories: true)
 
-    _ = try FileManager.default.replaceItemAt(targetURL, withItemAt: from.baseURL)
+    let target = VMDirectory(baseURL: targetURL)
+    if FileManager.default.fileExists(atPath: from.manifestURL.path) ||
+      FileManager.default.fileExists(atPath: target.manifestURL.path) {
+      let contentStore = try ContentStore()
+      try contentStore.withPruneLock {
+        _ = try FileManager.default.replaceItemAt(targetURL, withItemAt: from.baseURL)
+      }
+    } else {
+      _ = try FileManager.default.replaceItemAt(targetURL, withItemAt: from.baseURL)
+    }
   }
 
   func delete(_ name: RemoteName) throws {
-    try FileManager.default.removeItem(at: vmURL(name))
+    try removeRecord(at: vmURL(name))
     try gc()
   }
 
@@ -216,6 +215,7 @@ class VMStorageOCI: PrunableStorage {
 
     guard let enumerator = FileManager.default.enumerator(at: baseURL,
                                                           includingPropertiesForKeys: [.isSymbolicLinkKey]) else {
+      try gcContent()
       return
     }
 
@@ -243,7 +243,28 @@ class VMStorageOCI: PrunableStorage {
       let vmDir = VMDirectory(baseURL: baseURL)
 
       if !vmDir.isExplicitlyPulled() && incRefCount == 0 {
-        try FileManager.default.removeItem(at: baseURL)
+        try removeRecord(at: baseURL)
+      }
+    }
+
+    try gcContent()
+  }
+
+  /// Cached images with a manifest publish references into the shared content
+  /// store. Remove them through VMDirectory so reference removal is serialized
+  /// with clone, export, pull, and content GC, even if a record is incomplete.
+  private func removeRecord(at url: URL) throws {
+    try VMDirectory(baseURL: url).removeFromDisk()
+  }
+
+  /// Remove immutable files whose final published or in-progress reference
+  /// has disappeared, without collecting unrelated cached images.
+  fileprivate func gcContent() throws {
+    let contentStore = try ContentStore()
+    try contentStore.withPruneLock {
+      let referencedContentDigests = try referencedContentDigests(includeCachedImages: true)
+      for contentURL in try contentStore.prunables(excluding: referencedContentDigests) {
+        try FileManager.default.removeItem(at: contentURL)
       }
     }
   }
@@ -287,9 +308,57 @@ class VMStorageOCI: PrunableStorage {
   }
 
   func prunables() throws -> [Prunable] {
-    try list().filter { (_, vmDir, isSymlink) in
-      !isSymlink && vmDir.isStandalone
+    let records = try list().filter { (_, _, isSymlink) in
+      !isSymlink
     }.map { (_, vmDir, _) in vmDir }
+
+    // Attribute shared content to the newest cached image that references it.
+    // This counts each file once while charging it to the last record that
+    // normally needs to be removed before the file becomes reclaimable.
+    let nonCacheContentDigests = try referencedContentDigests(includeCachedImages: false)
+    var contentOwners = [String: VMDirectory]()
+    for record in records where record.isStackedCachedImage {
+      // Interrupted cache population can leave a truncated manifest in an
+      // otherwise recognizable cached record. It has no reliable content
+      // references, but it must not prevent pruning other cache entries.
+      for contentDigest in (try? record.diskContentDigests()) ?? []
+        where !nonCacheContentDigests.contains(contentDigest) {
+        guard let currentOwner = contentOwners[contentDigest] else {
+          contentOwners[contentDigest] = record
+          continue
+        }
+
+        let recordAccessDate = try record.accessDate()
+        let currentAccessDate = try currentOwner.accessDate()
+        if recordAccessDate > currentAccessDate ||
+          (recordAccessDate == currentAccessDate && record.url.path > currentOwner.url.path) {
+          contentOwners[contentDigest] = record
+        }
+      }
+    }
+
+    let contentStore = try ContentStore()
+    var ownedContentURLs = [URL: [URL]]()
+    for (contentDigest, owner) in contentOwners {
+      let contentURL = try contentStore.contentURL(for: contentDigest)
+      guard FileManager.default.fileExists(atPath: contentURL.path) else {
+        continue
+      }
+
+      ownedContentURLs[owner.url, default: []].append(contentURL)
+    }
+
+    var result: [Prunable] = records.map { record in
+      CachedImagePrunable(
+        vmDir: record,
+        ownedContentURLs: ownedContentURLs[record.url] ?? []
+      )
+    }
+
+    result += try contentStore.prunables(excluding: referencedContentDigests(includeCachedImages: true))
+      .map(ContentPrunable.init)
+
+    return result
   }
 
   func pull(_ name: RemoteName, registry: Registry, concurrency: UInt, deduplicate: Bool) async throws {
@@ -345,6 +414,12 @@ class VMStorageOCI: PrunableStorage {
       let tmpVMDirLock = try FileLock(lockURL: tmpVMDir.baseURL)
       try tmpVMDirLock.lock()
 
+      // Make in-progress stacked content references visible before reclaiming
+      // space or reconstructing immutable files.
+      try ContentStore().withPruneLock {
+        try manifestData.write(to: tmpVMDir.manifestURL)
+      }
+
       // A previously pulled standalone image already has the complete base
       // disk locally as disk.img. Promote that file into the content store
       // before sizing or pulling so a stacked child only fetches overlays.
@@ -402,16 +477,13 @@ class VMStorageOCI: PrunableStorage {
           return .throw
         }
 
-        // Preserve the exact manifest bytes received from the registry. Its
-        // digest identifies this cached image and stacked VMs pin it.
-        try manifestData.write(to: tmpVMDir.manifestURL)
         if preserveExplicitlyPulledMark {
           tmpVMDir.markExplicitlyPulled()
         }
 
         try move(digestName, from: tmpVMDir)
       }, onCancel: {
-        try? FileManager.default.removeItem(at: tmpVMDir.baseURL)
+        try? tmpVMDir.removeFromDisk()
       })
     } else {
       defaultLogger.appendNewLine("\(digestName) image is already cached! creating a symlink...")
@@ -430,9 +502,10 @@ class VMStorageOCI: PrunableStorage {
     _ = try VMStorageOCI().open(name)
   }
 
-  /// Returns `nil` for standalone images and the missing immutable disk-file groups
-  /// for stacked images. `ContentStore.existingContentURL()` intentionally
-  /// validates the digest so corrupt entries are repaired by a normal pull.
+  /// Returns nil for standalone images and the missing immutable disk-file
+  /// groups for stacked images. Like existing standalone cached images, cache hits trust
+  /// already-installed files; checking size still repairs truncated entries
+  /// without hashing a large prewarmed base on every pull.
   private func missingStackedDiskFileGroups(for manifest: OCIManifest) throws -> [TartDiskFileGroup]? {
     guard case .stacked(let base, let overlays) = try manifest.tartDiskRepresentation() else {
       return nil
@@ -441,10 +514,7 @@ class VMStorageOCI: PrunableStorage {
     let contentStore = try ContentStore()
     var missingGroups: [TartDiskFileGroup] = []
     for group in [base] + overlays {
-      guard let contentDigest = group.contentDigest else {
-        throw OCIManifestValidationError.invalidDiskMetadata("stacked disk files need a whole-file content digest")
-      }
-      if try contentStore.existingContentURL(for: contentDigest) == nil {
+      if try !hasUsableCachedDiskFile(group, contentStore: contentStore) {
         missingGroups.append(group)
       }
     }
@@ -462,23 +532,47 @@ class VMStorageOCI: PrunableStorage {
     }
 
     let contentStore = try ContentStore()
-    // Content-store entries are verified when installed. Avoid hashing a
-    // potentially large prewarmed base again on every stacked pull.
-    guard try contentStore.contentURLIfPresent(for: contentDigest) == nil else {
-      return
-    }
+    var attemptedCandidates = Swift.Set<String>()
+    while true {
+      // Keep the source record alive only while cloning its disk. The pull's
+      // in-progress manifest already protects the destination content digest,
+      // so hashing and installing the staged clone need not hold the global
+      // prune lock.
+      let temporaryURL = try contentStore.withPruneLock { () -> URL? in
+        // Content-store entries are verified when installed. Avoid hashing a
+        // potentially large prewarmed base again on every stacked pull.
+        guard try contentStore.contentURLIfPresent(for: contentDigest) == nil else {
+          return nil
+        }
 
-    for (_, vmDir, isSymlink) in try list() where !isSymlink && vmDir.isStandalone {
-      guard let manifestData = try? Data(contentsOf: vmDir.manifestURL),
-            let candidateManifest = try? OCIManifest(fromJSON: manifestData),
-            case .flat(let candidateBase) = try? candidateManifest.tartDiskRepresentation(),
-            diskChunksMatch(candidateBase.chunks, base.chunks) else {
-        continue
+        for (_, vmDir, isSymlink) in try list() where !isSymlink && vmDir.isStandalone {
+          guard !attemptedCandidates.contains(vmDir.baseURL.path),
+                let manifestData = try? Data(contentsOf: vmDir.manifestURL),
+                let candidateManifest = try? OCIManifest(fromJSON: manifestData),
+                case .flat(let candidateBase) = try? candidateManifest.tartDiskRepresentation(),
+                diskChunksMatch(candidateBase.chunks, base.chunks) else {
+            continue
+          }
+
+          attemptedCandidates.insert(vmDir.baseURL.path)
+          let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
+          do {
+            try FileManager.default.copyItem(at: vmDir.diskURL, to: temporaryURL)
+            return temporaryURL
+          } catch {
+            try? FileManager.default.removeItem(at: temporaryURL)
+            throw error
+          }
+        }
+
+        return nil
+      }
+
+      guard let temporaryURL else {
+        return
       }
 
-      let temporaryURL = try contentStore.temporaryContentURL(for: contentDigest)
       do {
-        try FileManager.default.copyItem(at: vmDir.diskURL, to: temporaryURL)
         _ = try contentStore.install(temporaryURL, contentDigest: contentDigest)
         return
       } catch ContentStoreError.contentDigestMismatch {
@@ -506,6 +600,17 @@ class VMStorageOCI: PrunableStorage {
     }
   }
 
+  private func hasUsableCachedDiskFile(_ group: TartDiskFileGroup, contentStore: ContentStore) throws -> Bool {
+    guard let contentDigest = group.contentDigest,
+          let contentURL = try contentStore.contentURLIfPresent(for: contentDigest),
+          let actualSize = UInt64(exactly: try contentURL.sizeBytes()),
+          let expectedSize = group.uncompressedSize() else {
+      return false
+    }
+
+    return actualSize == expectedSize
+  }
+
   func linked(from: RemoteName, to: RemoteName) -> Bool {
     do {
       let resolvedFrom = try FileManager.default.destinationOfSymbolicLink(atPath: vmURL(from).path)
@@ -516,9 +621,13 @@ class VMStorageOCI: PrunableStorage {
   }
 
   func link(from: RemoteName, to: RemoteName) throws {
-    try? FileManager.default.removeItem(at: vmURL(from))
-
-    try FileManager.default.createSymbolicLink(at: vmURL(from), withDestinationURL: vmURL(to))
+    // Export resolves mutable tags while holding this same lock, so replace
+    // the symlink atomically with respect to stacked archive staging.
+    let contentStore = try ContentStore()
+    try contentStore.withPruneLock {
+      try? FileManager.default.removeItem(at: vmURL(from))
+      try FileManager.default.createSymbolicLink(at: vmURL(from), withDestinationURL: vmURL(to))
+    }
 
     try gc()
   }
@@ -594,6 +703,108 @@ class VMStorageOCI: PrunableStorage {
       try LocalLayerCache(choosen.name, choosen.deduplicatedBytes, choosen.vmDir.diskURL, choosen.manifest)
     })
   }
+
+  /// Returns content referenced outside the OCI cache, optionally including
+  /// references published by retained cached images.
+  private func referencedContentDigests(includeCachedImages: Bool) throws -> Swift.Set<String> {
+    var result = Swift.Set<String>()
+
+    for (_, vmDir) in try VMStorageLocal().list() where vmDir.isStackedVM {
+      result.formUnion(try vmDir.diskContentDigests())
+    }
+
+    // Clone, pull, and import publish their manifest before installing
+    // immutable content. Include partially populated temporary directories so
+    // pruning cannot race those operations.
+    for url in try FileManager.default.contentsOfDirectory(
+      at: Config().tartTmpDir,
+      includingPropertiesForKeys: [],
+      options: .skipsHiddenFiles
+    ) {
+      let vmDir = VMDirectory(baseURL: url)
+      guard FileManager.default.fileExists(atPath: vmDir.manifestURL.path),
+            let contentDigests = try? vmDir.diskContentDigests() else {
+        continue
+      }
+
+      result.formUnion(contentDigests)
+    }
+
+    if includeCachedImages {
+      for (_, vmDir, isSymlink) in try list() where !isSymlink && vmDir.isStackedCachedImage {
+        // Malformed cached records are invalid references. Keep scanning so
+        // one interrupted population does not disable content GC globally.
+        if let contentDigests = try? vmDir.diskContentDigests() {
+          result.formUnion(contentDigests)
+        }
+      }
+    }
+
+    return result
+  }
+
+  fileprivate func deleteContentIfUnused(_ url: URL) throws {
+    let contentStore = try ContentStore()
+    try contentStore.withPruneLock {
+      let referencedContentDigests = try referencedContentDigests(includeCachedImages: true)
+      let stillPrunable = try contentStore.prunables(excluding: referencedContentDigests).contains {
+        $0.resolvingSymlinksInPath() == url.resolvingSymlinksInPath()
+      }
+      if stillPrunable {
+        try FileManager.default.removeItem(at: url)
+      }
+    }
+  }
+}
+
+private struct ContentPrunable: Prunable {
+  let url: URL
+
+  func delete() throws {
+    try VMStorageOCI().deleteContentIfUnused(url)
+  }
+
+  func accessDate() throws -> Date {
+    try url.accessDate()
+  }
+
+  func sizeBytes() throws -> Int {
+    try url.sizeBytes()
+  }
+
+  func allocatedSizeBytes() throws -> Int {
+    try url.allocatedSizeBytes()
+  }
+}
+
+/// A digest-addressed cached image plus immutable content attributed to the
+/// final remote reference that can release it.
+private struct CachedImagePrunable: Prunable {
+  let vmDir: VMDirectory
+  let ownedContentURLs: [URL]
+
+  var url: URL {
+    vmDir.url
+  }
+
+  func delete() throws {
+    try vmDir.delete()
+    // Deleting a record can make attributed content unreferenced. Run GC now
+    // so one prune invocation reclaims those bytes.
+    try VMStorageOCI().gcContent()
+  }
+
+  func accessDate() throws -> Date {
+    try vmDir.accessDate()
+  }
+
+  func sizeBytes() throws -> Int {
+    try vmDir.sizeBytes() + ownedContentURLs.map { try $0.sizeBytes() }.reduce(0, +)
+  }
+
+  func allocatedSizeBytes() throws -> Int {
+    try vmDir.allocatedSizeBytes() + ownedContentURLs.map { try $0.allocatedSizeBytes() }.reduce(0, +)
+  }
 }
 
 extension URL {