From e5997507799b99717040ad604bc83d720c7dd62d Mon Sep 17 00:00:00 2001 From: Kaede Akino Date: Tue, 22 Sep 2026 22:39:48 +0800 Subject: [PATCH] Limit CloudKit sync batches to 250 records --- FrameworkTests/CloudSyncLifecycleTests.swift | 58 ++++++++++++++++ FrameworkTests/CloudSyncSafetyTests.swift | 17 +++++ .../ConnectionHistoryModel.swift | 2 +- .../Cloud/CloudSyncLifecycle.swift | 9 ++- .../SwiftDataVaultJournal.swift | 39 +++++++++++ .../SwiftDataVaultStore.swift | 12 +++- TFMTests/ArchitectureBoundaryTests.swift | 18 ++++- TFMTests/ConnectionHistoryTests.swift | 2 +- TFMTests/VaultSyncTests.swift | 69 +++++++++++++++++++ 9 files changed, 218 insertions(+), 8 deletions(-) diff --git a/FrameworkTests/CloudSyncLifecycleTests.swift b/FrameworkTests/CloudSyncLifecycleTests.swift index 316c7024..9d0651c7 100644 --- a/FrameworkTests/CloudSyncLifecycleTests.swift +++ b/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() diff --git a/FrameworkTests/CloudSyncSafetyTests.swift b/FrameworkTests/CloudSyncSafetyTests.swift index 496be9fc..d096cc18 100644 --- a/FrameworkTests/CloudSyncSafetyTests.swift +++ b/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 diff --git a/Frameworks/TermindApplication/FeatureModels/ConnectionHistoryModel.swift b/Frameworks/TermindApplication/FeatureModels/ConnectionHistoryModel.swift index ee0621c2..0c25f806 100644 --- a/Frameworks/TermindApplication/FeatureModels/ConnectionHistoryModel.swift +++ b/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 diff --git a/Frameworks/TermindInfrastructure/Cloud/CloudSyncLifecycle.swift b/Frameworks/TermindInfrastructure/Cloud/CloudSyncLifecycle.swift index 41ad761b..5813944c 100644 --- a/Frameworks/TermindInfrastructure/Cloud/CloudSyncLifecycle.swift +++ b/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) diff --git a/Frameworks/TermindPersistence/SwiftDataVaultJournal.swift b/Frameworks/TermindPersistence/SwiftDataVaultJournal.swift index fd3ba935..ea88db8b 100644 --- a/Frameworks/TermindPersistence/SwiftDataVaultJournal.swift +++ b/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() + 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( diff --git a/Frameworks/TermindPersistence/SwiftDataVaultStore.swift b/Frameworks/TermindPersistence/SwiftDataVaultStore.swift index 126b49a0..70b93b91 100644 --- a/Frameworks/TermindPersistence/SwiftDataVaultStore.swift +++ b/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()) .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( diff --git a/TFMTests/ArchitectureBoundaryTests.swift b/TFMTests/ArchitectureBoundaryTests.swift index 2060dffc..3dd5b040 100644 --- a/TFMTests/ArchitectureBoundaryTests.swift +++ b/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 diff --git a/TFMTests/ConnectionHistoryTests.swift b/TFMTests/ConnectionHistoryTests.swift index f8256c81..195bc100 100644 --- a/TFMTests/ConnectionHistoryTests.swift +++ b/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") diff --git a/TFMTests/VaultSyncTests.swift b/TFMTests/VaultSyncTests.swift index fa125530..3f846930 100644 --- a/TFMTests/VaultSyncTests.swift +++ b/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()) + .map(\.value) + .sorted(by: ConnectionHistoryEntry.isNewer) + let historyEnvelopes = try context.fetch(FetchDescriptor()) + .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()) + == ConnectionHistoryModel.limit) + } + @Test func importedKnownHostsSyncWithoutHostsAndPropagateEditsAndDeletion() throws { let source = try makeContainer("KnownHostsSource")