Limit CloudKit sync batches to 250 records

FrameworkTests/CloudSyncLifecycleTests.swift

@@ -8,6 +8,64 @@ import Testing
 
 @Suite @MainActor
 struct CloudSyncLifecycleTests {
+    @Test
+    func uploadBatchesNeverExceedCloudKitRequestLimit() async throws {
+        let fixture = LifecycleFixture()
+        fixture.driver.start()
+        await fixture.driver.operation?.value
+        let ids = (0..<505).map {
+            CKRecord.ID(recordName: "record-\($0)", zoneID: fixture.driver.zoneID)
+        }
+        let pending = ids.map(CKSyncEngine.PendingRecordZoneChange.saveRecord)
+
+        let first = try #require(
+            fixture.driver.makeChangeBatch(pending, engine: fixture.network.nextEngine) {
+                CKRecord(recordType: "Test", recordID: $0)
+            })
+        let second = try #require(
+            fixture.driver.makeChangeBatch(Array(pending.dropFirst(250)), engine: fixture.network.nextEngine) {
+                CKRecord(recordType: "Test", recordID: $0)
+            })
+        let third = try #require(
+            fixture.driver.makeChangeBatch(Array(pending.dropFirst(500)), engine: fixture.network.nextEngine) {
+                CKRecord(recordType: "Test", recordID: $0)
+            })
+
+        #expect(CloudSyncLifecycle.maximumRecordsPerRequest == 250)
+        #expect(first.recordsToSave.map(\.recordID) == Array(ids.prefix(250)))
+        #expect(second.recordsToSave.map(\.recordID) == Array(ids.dropFirst(250).prefix(250)))
+        #expect(third.recordsToSave.map(\.recordID) == Array(ids.suffix(5)))
+        #expect(first.recordIDsToDelete.isEmpty)
+    }
+
+    @Test
+    func invalidPendingChangesDoNotConsumeTheRequestLimit() async throws {
+        let fixture = LifecycleFixture()
+        fixture.driver.start()
+        await fixture.driver.operation?.value
+        let ids = (0..<280).map {
+            CKRecord.ID(recordName: "record-\($0)", zoneID: fixture.driver.zoneID)
+        }
+        fixture.journal.blocked = Set(ids.prefix(10).map(\.recordName))
+        let missing = Set(ids.dropFirst(10).prefix(10).map(\.recordName))
+        let pending = ids.flatMap {
+            [
+                CKSyncEngine.PendingRecordZoneChange.deleteRecord($0),
+                CKSyncEngine.PendingRecordZoneChange.saveRecord($0),
+            ]
+        }
+
+        let batch = try #require(
+            fixture.driver.makeChangeBatch(pending, engine: fixture.network.nextEngine) { id in
+                missing.contains(id.recordName) ? nil : CKRecord(recordType: "Test", recordID: id)
+            })
+
+        #expect(batch.recordsToSave.count == 250)
+        #expect(batch.recordsToSave.first?.recordID == ids[20])
+        #expect(batch.recordsToSave.last?.recordID == ids[269])
+        #expect(batch.recordIDsToDelete.isEmpty)
+    }
+
     @Test
     func firstFetchGatesSendsAndAuthorizationWhileLocalChangesAccumulate() async throws {
         let fixture = LifecycleFixture()

FrameworkTests/CloudSyncSafetyTests.swift

@@ -9,6 +9,23 @@ import Testing
 
 @Suite @MainActor
 struct CloudSyncSafetyTests {
+    @Test(arguments: SyncSafetyCategory.allCases)
+    func everyCoordinatorUsesTheSharedCloudKitRequestLimit(category: SyncSafetyCategory) async throws {
+        let fixture = try CloudSyncSafetyFixture(category: category)
+        defer { fixture.stop() }
+        try await fixture.start()
+        let pending = (0...CloudSyncLifecycle.maximumRecordsPerRequest).map { index in
+            CKSyncEngine.PendingRecordZoneChange.saveRecord(
+                fixture.recordIDs[index % fixture.recordIDs.count]
+            )
+        }
+
+        let batch = try #require(fixture.batch(pending))
+
+        #expect(batch.recordsToSave.count == CloudSyncLifecycle.maximumRecordsPerRequest)
+        #expect(batch.recordIDsToDelete.isEmpty)
+    }
+
     @Test(arguments: SyncSafetyCategory.allCases)
     func unsupportedRecordIsFencedAtJournalBatchAndProviderWithoutBlockingOtherRecords(
         category: SyncSafetyCategory

Frameworks/TermindApplication/FeatureModels/ConnectionHistoryModel.swift

@@ -7,7 +7,7 @@ import struct TermindDomain.Host
 @MainActor
 @Observable
 public final class ConnectionHistoryModel {
-    public static let limit = 300
+    public static let limit = 250
     public static let presentationLimit = 100
 
     private let repository: any ConnectionHistoryRepository

Frameworks/TermindInfrastructure/Cloud/CloudSyncLifecycle.swift

@@ -95,6 +95,10 @@ struct CloudSyncTransport {
 /// One instance per category: only implementation is shared, never account or engine state.
 @MainActor @Observable
 final class CloudSyncLifecycle {
+    /// CloudKit accepts at most 250 record changes in one network request.
+    /// Every CKSyncEngine delegate in the app builds its batches through this lifecycle.
+    nonisolated static let maximumRecordsPerRequest = 250
+
     let store: CloudSyncLifecycleStore
     let transport: CloudSyncTransport
     let zoneID: CKRecordZone.ID
@@ -399,7 +403,10 @@ extension CloudSyncLifecycle {
             for change in pending {
                 // The protocol uses tombstones. Never emit a physical deletion from a restored queue.
                 guard case .saveRecord(let id) = change, try canUpload(id, engine: engine) else { continue }
-                if let record = recordProvider(id) { records.append(record) }
+                if let record = recordProvider(id) {
+                    records.append(record)
+                    if records.count == Self.maximumRecordsPerRequest { break }
+                }
             }
             guard accepts(engine), isReadyToSend, !records.isEmpty else { return nil }
             return CKSyncEngine.RecordZoneChangeBatch(recordsToSave: records)

Frameworks/TermindPersistence/SwiftDataVaultJournal.swift

@@ -438,6 +438,45 @@ public final class VaultSyncJournal {
         return VaultMutationSet(keys: changed)
     }
 
+    /// Removes connection events that exceed local retention before launch-time
+    /// history bootstrap. Records that have never reached CloudKit can disappear
+    /// from the journal; records with CloudKit identity require ordinary logical
+    /// deletion so other devices converge on the same retained timeline.
+    public func discardOverflowConnectionHistory(
+        _ values: [ConnectionHistoryEntry],
+        now: Date = .now
+    ) throws -> VaultMutationSet {
+        guard !values.isEmpty else { return .empty }
+        let state = try state()
+        var entries = try entriesByName()
+        var changed = Set<VaultRecordKey>()
+        var discardedPendingChanges = false
+
+        for value in values {
+            let key = VaultRecordKey(kind: .connectionHistory, id: value.id)
+            guard let row = entries[key.recordName], let envelope = row.envelope else { continue }
+            if row.systemFieldsData == nil {
+                context.delete(row)
+                entries[key.recordName] = nil
+                discardedPendingChanges = true
+                continue
+            }
+            guard !envelope.isDeleted else { continue }
+            let tombstone = VaultEnvelope(
+                key: key,
+                version: .next(after: envelope.version, writerID: state.installationID, now: now),
+                isDeleted: true
+            )
+            row.apply(tombstone, needsUpload: true)
+            changed.insert(key)
+        }
+
+        // A serialized engine may still contain names removed above. A fresh
+        // engine restores only the journal's surviving pending changes.
+        if discardedPendingChanges { state.engineStateData = nil }
+        return VaultMutationSet(keys: changed)
+    }
+
     /// Reconciles the shared installation directory without deleting records
     /// for devices that are merely absent from one client's local store.
     public func recordCurrentDevices(

Frameworks/TermindPersistence/SwiftDataVaultStore.swift

@@ -16,6 +16,8 @@ public final class HostPersistence {
     public func load() throws -> VaultSnapshot {
         let snapshot = try fetchSnapshot()
         let journal = VaultSyncJournal(context: context)
+        let overflow = try pruneConnectionHistory()
+        _ = try journal.discardOverflowConnectionHistory(overflow)
         _ = try journal.bootstrapIfNeeded(snapshot)
         _ = try journal.recordCurrentConnectionHistory(fetchConnectionHistory())
         _ = try journal.recordCurrentDevices(fetchDevices())
@@ -85,7 +87,7 @@ public final class HostPersistence {
                     try snapshot.apply(envelope)
                 }
             }
-            try pruneConnectionHistory()
+            _ = try pruneConnectionHistory()
             try persist(snapshot)
             try journal.acceptRemote(decision.accepted)
             let retentionMutations = try journal.recordCurrentConnectionHistory(
@@ -221,10 +223,14 @@ public final class HostPersistence {
         }
     }
 
-    private func pruneConnectionHistory() throws {
+    @discardableResult
+    private func pruneConnectionHistory() throws -> [ConnectionHistoryEntry] {
         let rows = try context.fetch(FetchDescriptor<PersistentConnectionHistoryEntry>())
             .sorted { ConnectionHistoryEntry.isNewer($0.value, $1.value) }
-        for row in rows.dropFirst(ConnectionHistoryModel.limit) { context.delete(row) }
+        let overflow = rows.dropFirst(ConnectionHistoryModel.limit)
+        let values = overflow.map(\.value)
+        for row in overflow { context.delete(row) }
+        return values
     }
 
     private func reconcile<Model: PersistentModel, Value>(

TFMTests/ArchitectureBoundaryTests.swift

@@ -191,17 +191,31 @@ struct ArchitectureBoundaryTests {
     @Test
     func cloudSyncCoordinatorsUseSharedUploadAndQuarantineGates() throws {
         let root = sourceRoot.appending(path: "../Frameworks/TermindInfrastructure/Cloud")
-        for name in ["VaultSyncCoordinator", "SettingsSyncCoordinator", "MetricsSyncCoordinator"] {
-            let source = try String(contentsOf: root.appending(path: "\(name).swift"), encoding: .utf8)
+        let files = try FileManager.default.contentsOfDirectory(
+            at: root, includingPropertiesForKeys: nil
+        ).filter { $0.pathExtension == "swift" }
+        let delegates = try files.filter {
+            try String(contentsOf: $0, encoding: .utf8).contains(": NSObject, CKSyncEngineDelegate")
+        }
+        #expect(!delegates.isEmpty)
+        for file in delegates {
+            let source = try String(contentsOf: file, encoding: .utf8)
             #expect(source.contains("lifecycle.makeChangeBatch("))
             #expect(source.contains("lifecycle.canUpload("))
             #expect(source.contains("CloudSyncRecordCoding.quarantine("))
             #expect(source.contains("isConflict: true"))
             #expect(!source.contains("try? journal.outgoingEntry"))
+            #expect(!source.contains("CKSyncEngine.RecordZoneChangeBatch("))
         }
         let lifecycle = try String(contentsOf: root.appending(path: "CloudSyncLifecycle.swift"), encoding: .utf8)
+        #expect(lifecycle.contains("maximumRecordsPerRequest = 250"))
+        #expect(lifecycle.contains("records.count == Self.maximumRecordsPerRequest"))
         #expect(!lifecycle.split(separator: "\n").contains("import SwiftData"))
         #expect(!lifecycle.split(separator: "\n").contains("import TermindPersistence"))
+        for file in files where file.lastPathComponent != "CloudSyncLifecycle.swift" {
+            let source = try String(contentsOf: file, encoding: .utf8)
+            #expect(!source.contains("CKSyncEngine.RecordZoneChangeBatch("))
+        }
     }
 
     @Test

TFMTests/ConnectionHistoryTests.swift

@@ -17,7 +17,7 @@ import struct TermindDomain.Host
 struct ConnectionHistoryTests {
     @Test
     func connectionProtocolSymbolsAreDistinct() {
-        #expect(ConnectionHistoryModel.limit == 300)
+        #expect(ConnectionHistoryModel.limit == 250)
         #expect(ConnectionHistoryModel.presentationLimit == 100)
         #expect(ConnectionHistoryEntry.Kind.ssh.symbol == "terminal")
         #expect(ConnectionHistoryEntry.Kind.telnet.symbol == "text.rectangle")

TFMTests/VaultSyncTests.swift

@@ -171,6 +171,75 @@ struct VaultSyncTests {
                 .first(where: { $0.recordName == key.recordName })?.needsUpload == true)
     }
 
+    @Test
+    func launchPrunesLegacyConnectionHistoryBeforeSyncBootstrap() throws {
+        let container = try makeContainer("ConnectionHistoryLaunchPruning")
+        let context = container.mainContext
+        let values = (0..<505).map { index in
+            ConnectionHistoryEntry(
+                hostID: UUID(), kind: .ssh,
+                connectedAt: Date(timeIntervalSince1970: TimeInterval(index)),
+                sourceDeviceID: UUID()
+            )
+        }
+        for value in values { context.insert(PersistentConnectionHistoryEntry(value)) }
+        try context.save()
+
+        _ = try HostPersistence(context: context).load()
+
+        let retained = try context.fetch(FetchDescriptor<PersistentConnectionHistoryEntry>())
+            .map(\.value)
+            .sorted(by: ConnectionHistoryEntry.isNewer)
+        let historyEnvelopes = try context.fetch(FetchDescriptor<PersistentVaultSyncEntry>())
+            .compactMap(\.envelope)
+            .filter { $0.key.kind == .connectionHistory }
+        #expect(retained.count == ConnectionHistoryModel.limit)
+        #expect(retained.map(\.id) == Array(values.suffix(ConnectionHistoryModel.limit).reversed()).map(\.id))
+        #expect(historyEnvelopes.count == ConnectionHistoryModel.limit)
+        #expect(Set(historyEnvelopes.map(\.key.id)) == Set(retained.map(\.id)))
+    }
+
+    @Test
+    func launchDiscardsUnsentOverflowAndTombstonesUploadedOverflow() throws {
+        let container = try makeContainer("ConnectionHistoryOverflowJournal")
+        let context = container.mainContext
+        let values = (0...(ConnectionHistoryModel.limit + 1)).map { index in
+            ConnectionHistoryEntry(
+                hostID: UUID(), kind: .ssh,
+                connectedAt: Date(timeIntervalSince1970: TimeInterval(index)),
+                sourceDeviceID: UUID()
+            )
+        }
+        for value in values { context.insert(PersistentConnectionHistoryEntry(value)) }
+        let unsent = values[0]
+        let uploaded = values[1]
+        let unsentKey = VaultRecordKey(kind: .connectionHistory, id: unsent.id)
+        let uploadedKey = VaultRecordKey(kind: .connectionHistory, id: uploaded.id)
+        let journal = VaultSyncJournal(context: context)
+        for value in [unsent, uploaded] {
+            let envelope = VaultEnvelope(
+                key: VaultRecordKey(kind: .connectionHistory, id: value.id),
+                version: .baseline(writerID: UUID()), payload: .connectionHistory(value)
+            )
+            context.insert(
+                try PersistentVaultSyncEntry(envelope: envelope, needsUpload: value == unsent)
+            )
+        }
+        try journal.updateSystemFields(Data("cloud-fields".utf8), for: uploadedKey)
+        let state = try journal.state()
+        state.engineStateData = Data("stale-pending-history".utf8)
+        try context.save()
+
+        _ = try HostPersistence(context: context).load()
+
+        #expect(try journal.envelope(for: unsentKey) == nil)
+        #expect(try journal.envelope(for: uploadedKey)?.isDeleted == true)
+        #expect(try journal.state().engineStateData == nil)
+        #expect(
+            try context.fetchCount(FetchDescriptor<PersistentConnectionHistoryEntry>())
+                == ConnectionHistoryModel.limit)
+    }
+
     @Test
     func importedKnownHostsSyncWithoutHostsAndPropagateEditsAndDeletion() throws {
         let source = try makeContainer("KnownHostsSource")