diff --git a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Database/FilesDatabaseManager+ChangeDelivery.swift b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Database/FilesDatabaseManager+ChangeDelivery.swift index 78c54d38b5db8..7d21779c301a2 100644 --- a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Database/FilesDatabaseManager+ChangeDelivery.swift +++ b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Database/FilesDatabaseManager+ChangeDelivery.swift @@ -2,6 +2,22 @@ // SPDX-License-Identifier: GPL-2.0-or-later import Foundation +import RealmSwift + +/// The stored state of one change-delivery session. +typealias ChangeDeliverySessionState = ( + sessionId: String, + containerKey: String, + currentAnchorKey: String, + nextSequence: Int, + finalAnchorRawValue: Data, + incomplete: Bool, + pendingEndSequence: Int, + pendingAnchorKey: String?, + pendingMoreComing: Bool, + pendingReported: Bool, + hardRemoveDeleted: Bool +) extension FilesDatabaseManager { /// Create a durable change-delivery session and store the complete ordered change snapshot. @@ -60,216 +76,188 @@ extension FilesDatabaseManager { } } - /// Return the active delivery session whose committed anchor matches `anchorKey`. - func changeDeliverySession(forAnchorKey anchorKey: String, containerKey: String) -> ( - sessionId: String, - containerKey: String, - currentAnchorKey: String, - nextSequence: Int, - finalAnchorRawValue: Data, - incomplete: Bool, - pendingEndSequence: Int, - pendingAnchorKey: String?, - pendingMoreComing: Bool, - pendingReported: Bool, - hardRemoveDeleted: Bool - )? { - let sessions = ncDatabase() - .objects(RealmChangeDeliverySession.self) - .filter("currentAnchorKey == %@ AND containerKey == %@ AND completed == false", anchorKey, containerKey) - guard let session = sessions.first - else { - return nil - } - - return changeDeliverySession(sessionId: session.sessionId) - } - - /// Return an active delivery session whose pending continuation anchor matches `anchorKey`. - func changeDeliverySession(forPendingAnchorKey anchorKey: String, containerKey: String) -> ( - sessionId: String, - containerKey: String, - currentAnchorKey: String, - nextSequence: Int, - finalAnchorRawValue: Data, - incomplete: Bool, - pendingEndSequence: Int, - pendingAnchorKey: String?, - pendingMoreComing: Bool, - pendingReported: Bool, - hardRemoveDeleted: Bool - )? { - let sessions = ncDatabase() + /// + /// Return the active delivery session which `anchorKey` continues, either as its committed anchor or as the + /// anchor of its reported but not yet acknowledged batch. + /// + /// Both are matched in one query and read from the same snapshot, because a concurrent acknowledgement moves + /// the anchor from one to the other and two separate queries could miss it in both. + /// + func changeDeliverySession(forContinuationKey anchorKey: String, containerKey: String) -> ChangeDeliverySessionState? { + let session = ncDatabase() .objects(RealmChangeDeliverySession.self) - .filter("pendingAnchorKey == %@ AND containerKey == %@ AND pendingReported == true AND completed == false", anchorKey, containerKey) - guard let session = sessions.first else { - return nil - } + .filter( + "containerKey == %@ AND completed == false AND (currentAnchorKey == %@ OR (pendingAnchorKey == %@ AND pendingReported == true))", + containerKey, + anchorKey, + anchorKey + ) + .first - return changeDeliverySession(sessionId: session.sessionId) + return session.map(state(of:)) } /// Return an active delivery session by its durable identifier. - func changeDeliverySession(sessionId: String) -> ( - sessionId: String, - containerKey: String, - currentAnchorKey: String, - nextSequence: Int, - finalAnchorRawValue: Data, - incomplete: Bool, - pendingEndSequence: Int, - pendingAnchorKey: String?, - pendingMoreComing: Bool, - pendingReported: Bool, - hardRemoveDeleted: Bool - )? { + func changeDeliverySession(sessionId: String) -> ChangeDeliverySessionState? { guard let session = ncDatabase().object(ofType: RealmChangeDeliverySession.self, forPrimaryKey: sessionId), !session.completed else { return nil } - return ( - session.sessionId, - session.containerKey, - session.currentAnchorKey, - session.nextSequence, - session.finalAnchorRawValue, - session.incomplete, - session.pendingEndSequence, - session.pendingAnchorKey, - session.pendingMoreComing, - session.pendingReported, - session.hardRemoveDeleted - ) + return state(of: session) } - /// Return the next ordered range of an active change-delivery session. - func changeDeliveryItems( + /// + /// Mark the next batch of at most `maxItems` items as reported and return it. If a batch is already reported + /// but not acknowledged, return that batch again so a repeated request replays it. + /// + /// The session is read and changed in one write transaction, so a concurrent acknowledgement cannot leave the + /// batch starting at a stale position. + /// + /// - Parameter continuationAnchorKey: The anchor key for a batch which ends before the given sequence and is + /// followed by more. + /// - Returns: `nil` if the session no longer exists or the write fails. + /// + func prepareChangeDeliveryBatch( sessionId: String, - fromSequence sequence: Int, - limit: Int - ) -> [(sequence: Int, metadataData: Data, deleted: Bool)] { - ncDatabase() - .objects(RealmChangeDeliveryItem.self) - .filter("sessionId == %@ AND sequence >= %@", sessionId, sequence) - .sorted(byKeyPath: "sequence") - .prefix(limit) - .map { ($0.sequence, $0.metadataData, $0.deleted) } - } + maxItems: Int, + continuationAnchorKey: (Int) -> String + ) -> (items: [(metadataData: Data, deleted: Bool)], endSequence: Int, nextAnchorKey: String?, moreComing: Bool)? { + let database = ncDatabase() + var batch: (items: [(metadataData: Data, deleted: Bool)], endSequence: Int, nextAnchorKey: String?, moreComing: Bool)? - /// Return the deleted item identifiers in the currently prepared batch. - func pendingChangeDeliveryDeletedOcIds(sessionId: String) -> [String] { - guard let session = changeDeliverySession(sessionId: sessionId), - session.pendingReported, - session.pendingEndSequence > session.nextSequence - else { - return [] - } + do { + try database.write { + guard let session = database.object(ofType: RealmChangeDeliverySession.self, forPrimaryKey: sessionId), + !session.completed + else { + return + } - let decoder = JSONDecoder() - return changeDeliveryItems( - sessionId: sessionId, - fromSequence: session.nextSequence, - limit: session.pendingEndSequence - session.nextSequence - ).compactMap { item in - guard item.deleted, - let metadata = try? decoder.decode(SendableItemMetadata.self, from: item.metadataData) - else { - return nil + if !session.pendingReported { + let candidates = database.objects(RealmChangeDeliveryItem.self) + .filter("sessionId == %@ AND sequence >= %@", sessionId, session.nextSequence) + .sorted(byKeyPath: "sequence") + .prefix(maxItems + 1) + let moreComing = candidates.count > maxItems + let endSequence = candidates.prefix(maxItems).last.map { $0.sequence + 1 } ?? session.nextSequence + + session.pendingEndSequence = endSequence + session.pendingAnchorKey = moreComing ? continuationAnchorKey(endSequence) : nil + session.pendingMoreComing = moreComing + session.pendingReported = true + } + + let items = database.objects(RealmChangeDeliveryItem.self) + .filter("sessionId == %@ AND sequence >= %@ AND sequence < %@", sessionId, session.nextSequence, session.pendingEndSequence) + .sorted(byKeyPath: "sequence") + .map { (metadataData: $0.metadataData, deleted: $0.deleted) } + + batch = (Array(items), session.pendingEndSequence, session.pendingAnchorKey, session.pendingMoreComing) } - return metadata.ocId + } catch { + return nil } + + return batch } - func prepareChangeDeliveryBatch( - sessionId: String, - endSequence: Int, - nextAnchorKey: String?, - moreComing: Bool - ) -> Bool { + /// + /// Acknowledge the reported batch which ends before `endSequence` after the observer accepted it. + /// + /// The batch is identified explicitly because the next request can acknowledge it first and prepare the + /// following batch before the enumerator which reported it gets here. The session is read inside the write + /// transaction so such a concurrent change is seen. + /// + /// - Returns: `false` if the batch cannot be acknowledged or the write fails. A batch which is no longer + /// reported, for example because it was acknowledged already, is ignored and `true` is returned. + /// + func acknowledgeChangeDeliveryBatch(sessionId: String, endSequence: Int) -> Bool { let database = ncDatabase() - guard let session = database.object(ofType: RealmChangeDeliverySession.self, forPrimaryKey: sessionId), - !session.completed - else { - return false - } - - if session.pendingReported { - return session.pendingEndSequence == endSequence - && session.pendingAnchorKey == nextAnchorKey - && session.pendingMoreComing == moreComing - } + var acknowledged = true do { try database.write { - session.pendingEndSequence = endSequence - session.pendingAnchorKey = nextAnchorKey - session.pendingMoreComing = moreComing - session.pendingReported = true + guard let session = database.object(ofType: RealmChangeDeliverySession.self, forPrimaryKey: sessionId), + !session.completed, + session.pendingReported, + session.pendingEndSequence == endSequence + else { + return + } + + acknowledged = acknowledgeReportedBatch(of: session, in: database) } - return true } catch { return false } - } - /// Acknowledge the prepared batch after the observer accepted it. - func acknowledgeChangeDeliveryBatch(sessionId: String, deletedOcIds: [String]) -> Bool { - let database = ncDatabase() - guard let session = database.object(ofType: RealmChangeDeliverySession.self, forPrimaryKey: sessionId), - !session.completed - else { - return true - } - - guard session.pendingReported else { - return false - } + return acknowledged + } + /// + /// Apply the deletions of the reported batch of `session`, then advance the session past it or remove the + /// session after its final batch. Must be called inside a write transaction. + /// + /// - Returns: `false` if the session state is inconsistent. + /// + private func acknowledgeReportedBatch(of session: RealmChangeDeliverySession, in database: Realm) -> Bool { if session.pendingMoreComing, session.pendingAnchorKey == nil { return false } - let pendingEndSequence = session.pendingEndSequence - let pendingAnchorKey = session.pendingAnchorKey - let pendingMoreComing = session.pendingMoreComing - - do { - try database.write { - let deletedItems = deletedOcIds.isEmpty - ? database.objects(RealmItemMetadata.self).filter("ocId == %@", "") - : database.objects(RealmItemMetadata.self).filter("ocId IN %@", deletedOcIds) - if session.hardRemoveDeleted { - database.delete(deletedItems) - } else { - deletedItems.forEach { $0.deleted = true } - } + let sessionId = session.sessionId + let endSequence = session.pendingEndSequence + let decoder = JSONDecoder() + // An explicit array type selects the eager `compactMap`, Realm needs an array for IN below. + let deletedOcIds: [String] = database.objects(RealmChangeDeliveryItem.self) + .filter("sessionId == %@ AND sequence >= %@ AND sequence < %@ AND deleted == true", sessionId, session.nextSequence, endSequence) + .compactMap { try? decoder.decode(SendableItemMetadata.self, from: $0.metadataData).ocId } - if pendingMoreComing, let pendingAnchorKey { - session.nextSequence = pendingEndSequence - session.currentAnchorKey = pendingAnchorKey - database.objects(RealmChangeDeliveryItem.self) - .filter("sessionId == %@ AND sequence < %@", sessionId, pendingEndSequence) - .forEach { database.delete($0) } - } else { - database.objects(RealmChangeDeliveryItem.self) - .filter("sessionId == %@", sessionId) - .forEach { database.delete($0) } - database.delete(session) - } + if deletedOcIds.isEmpty == false { + let deletedItems = database.objects(RealmItemMetadata.self).filter("ocId IN %@", deletedOcIds) - if database.object(ofType: RealmChangeDeliverySession.self, forPrimaryKey: sessionId) != nil { - session.pendingEndSequence = 0 - session.pendingAnchorKey = nil - session.pendingMoreComing = false - session.pendingReported = false - } + if session.hardRemoveDeleted { + database.delete(deletedItems) + } else { + deletedItems.forEach { $0.deleted = true } } - return true - } catch { - return false } + + if session.pendingMoreComing, let pendingAnchorKey = session.pendingAnchorKey { + database.delete( + database.objects(RealmChangeDeliveryItem.self) + .filter("sessionId == %@ AND sequence < %@", sessionId, endSequence) + ) + session.nextSequence = endSequence + session.currentAnchorKey = pendingAnchorKey + session.pendingEndSequence = 0 + session.pendingAnchorKey = nil + session.pendingMoreComing = false + session.pendingReported = false + } else { + database.delete(database.objects(RealmChangeDeliveryItem.self).filter("sessionId == %@", sessionId)) + database.delete(session) + } + + return true + } + + private func state(of session: RealmChangeDeliverySession) -> ChangeDeliverySessionState { + ( + session.sessionId, + session.containerKey, + session.currentAnchorKey, + session.nextSequence, + session.finalAnchorRawValue, + session.incomplete, + session.pendingEndSequence, + session.pendingAnchorKey, + session.pendingMoreComing, + session.pendingReported, + session.hardRemoveDeleted + ) } /// Remove all active delivery sessions for one enumerated container. @@ -279,7 +267,8 @@ extension FilesDatabaseManager { .filter("containerKey == %@ AND completed == false", containerKey) try? database.write { - let sessionIds = sessions.map(\.sessionId) + // Realm requires an array for IN, not the lazy sequence `map` returns on its results. + let sessionIds = Array(sessions.map(\.sessionId)) if sessionIds.isEmpty == false { database.objects(RealmChangeDeliveryItem.self) .filter("sessionId IN %@", sessionIds) diff --git a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Documentation.docc/ChangeEnumeration.md b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Documentation.docc/ChangeEnumeration.md index 96547e65f8a19..a7ce3ab5ad416 100644 --- a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Documentation.docc/ChangeEnumeration.md +++ b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Documentation.docc/ChangeEnumeration.md @@ -93,10 +93,14 @@ limits it to 4,000. 3. Uses the extra item to determine `moreComing`. 4. Stores a pending acknowledgement boundary without advancing the committed cursor. +These steps run in one database transaction. If a batch is already pending, the same batch is returned again instead. + After `finishEnumeratingChanges(upTo:moreComing:)` returns, the enumerator acknowledges the batch and advances the committed cursor. If delivery is interrupted before that acknowledgement, the same batch is replayed from its original anchor. +File Provider may request the next batch before that acknowledgement runs, also on a new enumerator. A request with the continuation anchor of a batch means that File Provider accepted that batch, so such a request acknowledges it first. Each acknowledgement names its batch by session and end position and is ignored when that batch is no longer pending, so a late acknowledgement never commits a later batch. + Updates and deletions share the same limit. For a limit of three, two updates and one deletion consume the whole batch. @@ -128,6 +132,8 @@ validation. When the saved list is found, the enumerator reads the next batch without contacting the server again. The final batch returns the container's final sync anchor instead of a continuation anchor. +The saved list is found by its committed anchor or by the anchor of its pending batch, read in one query so a concurrent acknowledgement cannot hide it. + The framework may invalidate the current `Enumerator` after an intermediate batch and create another one for the next request. The extension process may also be restarted. Pending changes must therefore be stored outside the @@ -191,5 +197,7 @@ The change-enumeration tests are in - `testContainerChangesDeliveredInBatches` checks regular-directory changes. - `testTrashChangesDeliveredInBatches` checks batched trash deletions. +`Tests/NextcloudFileProviderKitTests/ChangeDeliveryBufferTests.swift` checks that a late acknowledgement from another enumerator or an earlier session never commits a later batch. The change-delivery cases in `FilesDatabaseManagerTests.swift` cover preparation, replay, acknowledgement by session and end position, and removal of sessions. + These tests exercise the delivery logic in the package. A signed macOS build is still required to verify the interaction with the real `fileproviderd`. diff --git a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/ChangeDeliveryBuffer.swift b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/ChangeDeliveryBuffer.swift index 7e971c3677b19..e4fb9768eb83f 100644 --- a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/ChangeDeliveryBuffer.swift +++ b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/ChangeDeliveryBuffer.swift @@ -41,49 +41,26 @@ final class ChangeDeliveryBuffer: @unchecked Sendable { logger = FileProviderLogger(category: "ChangeDeliveryBuffer", log: log) } + /// /// Whether an active durable session is positioned at the given anchor. + /// + /// The framework asks for a continuation anchor only after it accepted the batch which ended there, so a batch + /// which was reported up to `key` but not acknowledged yet is acknowledged here. + /// func isPrimed(forKey key: String) -> Bool { lock.lock() defer { lock.unlock() } - if let sessionId { - guard let session = dbManager.changeDeliverySession(sessionId: sessionId) else { - self.sessionId = nil - return false - } - - if session.currentAnchorKey == key { - return true - } - - guard session.pendingReported, - session.pendingAnchorKey == key - else { - return false - } + guard let session = dbManager.changeDeliverySession(forContinuationKey: key, containerKey: containerKey) else { + sessionId = nil + return false + } - let deletedOcIds = dbManager.pendingChangeDeliveryDeletedOcIds(sessionId: sessionId) - guard dbManager.acknowledgeChangeDeliveryBatch(sessionId: sessionId, deletedOcIds: deletedOcIds) else { + if session.currentAnchorKey != key { + guard dbManager.acknowledgeChangeDeliveryBatch(sessionId: session.sessionId, endSequence: session.pendingEndSequence) else { logger.error("Could not acknowledge the previously reported change delivery batch.") return false } - - return true - } - - if let session = dbManager.changeDeliverySession(forAnchorKey: key, containerKey: containerKey) { - sessionId = session.sessionId - return true - } - - guard let session = dbManager.changeDeliverySession(forPendingAnchorKey: key, containerKey: containerKey) else { - return false - } - - let deletedOcIds = dbManager.pendingChangeDeliveryDeletedOcIds(sessionId: session.sessionId) - guard dbManager.acknowledgeChangeDeliveryBatch(sessionId: session.sessionId, deletedOcIds: deletedOcIds) else { - logger.error("Could not acknowledge the previously reported change delivery batch.") - return false } sessionId = session.sessionId @@ -146,8 +123,11 @@ final class ChangeDeliveryBuffer: @unchecked Sendable { logger.debug("Persisted change delivery session.") } + /// /// Prepare the next update/delete batch without advancing the committed cursor. - /// Returns a continuation anchor for intermediate batches and the session's final anchor for the last. + /// Returns a continuation anchor for intermediate batches and the session's final anchor for the last, and + /// the identity of the batch to pass to ``acknowledgeBatch(_:)``. + /// func prepareChangeDeliveryBatch( maxItems: Int ) -> ( @@ -155,33 +135,38 @@ final class ChangeDeliveryBuffer: @unchecked Sendable { deleted: [SendableItemMetadata], moreComing: Bool, continuationAnchorRawValue: Data?, - finalAnchorRawValue: Data? + finalAnchorRawValue: Data?, + acknowledgement: (sessionId: String, endSequence: Int)? ) { lock.lock() defer { lock.unlock() } - let budget = max(1, maxItems) guard let sessionId, let session = dbManager.changeDeliverySession(sessionId: sessionId) else { - return ([], [], false, nil, nil) + return ([], [], false, nil, nil, nil) } - let storedItems = dbManager.changeDeliveryItems( + guard let batch = dbManager.prepareChangeDeliveryBatch( sessionId: sessionId, - fromSequence: session.nextSequence, - limit: budget + 1 - ) - let batchItems = Array(storedItems.prefix(budget)) - let moreComing = storedItems.count > budget + maxItems: max(1, maxItems), + continuationAnchorKey: { "\(Self.continuationPrefix)\(sessionId)|\($0)" } + ) else { + logger.error("Could not prepare change delivery batch.") + return ([], [], false, nil, nil, nil) + } + let decoder = JSONDecoder() var updated = [SendableItemMetadata]() var deleted = [SendableItemMetadata]() - for item in batchItems { + for item in batch.items { guard let metadata = try? decoder.decode(SendableItemMetadata.self, from: item.metadataData) else { + // The batch is not delivered, so do not leave it reported. logger.error("Could not decode change delivery item.") - return ([], [], false, nil, session.finalAnchorRawValue) + dbManager.removeChangeDeliverySessions(containerKey: containerKey) + self.sessionId = nil + return ([], [], false, nil, session.finalAnchorRawValue, nil) } if item.deleted { @@ -191,54 +176,36 @@ final class ChangeDeliveryBuffer: @unchecked Sendable { } } - let nextSequence = batchItems.last.map { $0.sequence + 1 } ?? session.nextSequence - let continuationAnchor: Data? - let nextAnchorKey: String? - if moreComing { - let key = "\(Self.continuationPrefix)\(sessionId)|\(nextSequence)" - continuationAnchor = Data(key.utf8) - nextAnchorKey = key - } else { - continuationAnchor = nil - nextAnchorKey = nil - } - - guard dbManager.prepareChangeDeliveryBatch( - sessionId: sessionId, - endSequence: nextSequence, - nextAnchorKey: nextAnchorKey, - moreComing: moreComing - ) else { - logger.error("Could not prepare change delivery batch.") - return ([], [], false, nil, nil) - } - logger.debug("Prepared change delivery batch.") return ( updated, deleted, - moreComing, - continuationAnchor, - session.finalAnchorRawValue + batch.moreComing, + batch.nextAnchorKey.map { Data($0.utf8) }, + session.finalAnchorRawValue, + (sessionId, batch.endSequence) ) } - /// Commit the prepared batch after `finishEnumeratingChanges` returns. - func acknowledgeBatch(deletedOcIds: [String]) { - lock.lock() - defer { lock.unlock() } - - guard let sessionId else { + /// + /// Commit a batch prepared by ``prepareChangeDeliveryBatch(maxItems:)`` after `finishEnumeratingChanges` + /// returns. Does nothing for `nil`, which stands for a batch that was not prepared. + /// + func acknowledgeBatch(_ batch: (sessionId: String, endSequence: Int)?) { + guard let batch else { return } - guard dbManager.acknowledgeChangeDeliveryBatch(sessionId: sessionId, deletedOcIds: deletedOcIds) else { + lock.lock() + defer { lock.unlock() } + + guard dbManager.acknowledgeChangeDeliveryBatch(sessionId: batch.sessionId, endSequence: batch.endSequence) else { logger.error("Could not acknowledge change delivery batch.") return } - if dbManager.changeDeliverySession(sessionId: sessionId) == nil { - self.sessionId = nil + if sessionId == batch.sessionId, dbManager.changeDeliverySession(sessionId: batch.sessionId) == nil { + sessionId = nil } } diff --git a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+ChangeEnumeration.swift b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+ChangeEnumeration.swift index c7231de890057..81e7fdd785361 100644 --- a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+ChangeEnumeration.swift +++ b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+ChangeEnumeration.swift @@ -83,6 +83,7 @@ extension Enumerator { deleted: [itemMetadata], anchor: anchor, moreComing: false, + acknowledgement: nil, account: account, remoteInterface: remoteInterface, dbManager: dbManager diff --git a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+ObserverReporting.swift b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+ObserverReporting.swift index 6f1bbd2dc7bce..d64a64773ba36 100644 --- a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+ObserverReporting.swift +++ b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+ObserverReporting.swift @@ -105,6 +105,7 @@ extension Enumerator { deleted: batch.deleted, anchor: reportedAnchor, moreComing: batch.moreComing, + acknowledgement: batch.acknowledgement, account: account, remoteInterface: remoteInterface, dbManager: dbManager @@ -113,13 +114,15 @@ extension Enumerator { /// Report one batch of changes and acknowledge it after the observer finishes. /// The caller keeps the batch within the framework's size limit and orders updates parents-first. - /// Deletions remain available for replay until acknowledgement. + /// Deletions remain available for replay until acknowledgement. A `nil` acknowledgement means the batch was not + /// prepared from the change buffer and has nothing to acknowledge. func completeChangesBatch( _ observer: NSFileProviderChangeObserver, updated: [SendableItemMetadata], deleted: [SendableItemMetadata], anchor: NSFileProviderSyncAnchor, moreComing: Bool, + acknowledgement: (sessionId: String, endSequence: Int)?, account: Account, remoteInterface: RemoteInterface, dbManager: FilesDatabaseManager @@ -144,9 +147,9 @@ extension Enumerator { reportBatchUpdates( observer, updated: updated, - deletedToRemove: deleted, anchor: anchor, moreComing: moreComing, + acknowledgement: acknowledgement, account: account, remoteInterface: remoteInterface, dbManager: dbManager @@ -160,15 +163,15 @@ extension Enumerator { private func reportBatchUpdates( _ observer: NSFileProviderChangeObserver, updated: [SendableItemMetadata], - deletedToRemove: [SendableItemMetadata], anchor: NSFileProviderSyncAnchor, moreComing: Bool, + acknowledgement: (sessionId: String, endSequence: Int)?, account: Account, remoteInterface: RemoteInterface, dbManager: FilesDatabaseManager, handleInvalidParent: Bool = true ) { - Task { [updated, deletedToRemove] in + Task { [updated] in do { let updatedItems = try await updated.toFileProviderItems(account: account, remoteInterface: remoteInterface, dbManager: dbManager, log: self.logger.log) @@ -178,7 +181,7 @@ extension Enumerator { } observer.finishEnumeratingChanges(upTo: anchor, moreComing: moreComing) - changeBuffer.acknowledgeBatch(deletedOcIds: deletedToRemove.map(\.ocId)) + changeBuffer.acknowledgeBatch(acknowledgement) } } catch let error as NSError { // This error can only mean a missing parent item identifier guard handleInvalidParent else { @@ -202,9 +205,9 @@ extension Enumerator { reportBatchUpdates( observer, updated: [metadata] + updated, - deletedToRemove: deletedToRemove, anchor: anchor, moreComing: moreComing, + acknowledgement: acknowledgement, account: account, remoteInterface: remoteInterface, dbManager: dbManager, diff --git a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+Trash.swift b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+Trash.swift index afbd52453f4e0..0355bfc4bfb90 100644 --- a/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+Trash.swift +++ b/shell_integration/MacOSX/NextcloudFileProviderKit/Sources/NextcloudFileProviderKit/Enumeration/Enumerator+Trash.swift @@ -35,7 +35,7 @@ extension Enumerator { } observer.finishEnumeratingChanges(upTo: reportedAnchor, moreComing: batch.moreComing) - changeBuffer.acknowledgeBatch(deletedOcIds: batch.deleted.map(\.ocId)) + changeBuffer.acknowledgeBatch(batch.acknowledgement) logger.debug("Reported trash deletion batch. deleted: \(batch.deleted.count), moreComing: \(batch.moreComing)") } diff --git a/shell_integration/MacOSX/NextcloudFileProviderKit/Tests/NextcloudFileProviderKitTests/ChangeDeliveryBufferTests.swift b/shell_integration/MacOSX/NextcloudFileProviderKit/Tests/NextcloudFileProviderKitTests/ChangeDeliveryBufferTests.swift new file mode 100644 index 0000000000000..6e71a33d7d39f --- /dev/null +++ b/shell_integration/MacOSX/NextcloudFileProviderKit/Tests/NextcloudFileProviderKitTests/ChangeDeliveryBufferTests.swift @@ -0,0 +1,99 @@ +// SPDX-FileCopyrightText: 2026 Nextcloud GmbH and Nextcloud contributors +// SPDX-License-Identifier: LGPL-3.0-or-later + +@preconcurrency import FileProvider +@testable import NextcloudFileProviderKit +import NextcloudFileProviderKitMocks +import RealmSwift +import TestInterface +import XCTest + +/// +/// Coverage for an enumerator acknowledging its batch only after the framework already asked for the next one, +/// possibly on a new enumerator or in a new session. +/// +/// XCTest instead of Swift Testing because the database works through the process-wide Realm default +/// configuration, which concurrently running Swift Testing suites would share. +/// +final class ChangeDeliveryBufferTests: NextcloudFileProviderKitTestCase { + static let account = Account(user: "testUser", id: "testUserId", serverUrl: "https://mock.nc.com", password: "abcd") + static let dbManager = FilesDatabaseManager(account: account, databaseDirectory: makeDatabaseDirectory(), fileProviderDomainIdentifier: NSFileProviderDomainIdentifier("test"), log: FileProviderLogMock()) + static let containerKey = "working-set" + static let batchSize = 2 + + override func setUp() { + super.setUp() + Realm.Configuration.defaultConfiguration.inMemoryIdentifier = name + } + + private func makeBuffer() -> ChangeDeliveryBuffer { + ChangeDeliveryBuffer(dbManager: Self.dbManager, containerKey: Self.containerKey, hardRemoveDeleted: true, log: FileProviderLogMock()) + } + + private func storeDeletedItems(_ ocIds: [String]) -> [SendableItemMetadata] { + ocIds.map { ocId in + var metadata = SendableItemMetadata(ocId: ocId, fileName: "\(ocId).txt", account: Self.account) + metadata.deleted = true + Self.dbManager.addItemMetadata(metadata) + return metadata + } + } + + private func anchorKey(_ rawValue: Data?) throws -> String { + try XCTUnwrap(rawValue.flatMap { String(data: $0, encoding: .utf8) }) + } + + func testLateAcknowledgementOfPreviousBatchDoesNotCommitTheNextBatch() throws { + let db = Self.dbManager.ncDatabase() // Strong reference for the in-memory database. + debugPrint(db) + let deleted = storeDeletedItems(["d0", "d1", "d2", "d3"]) + let updated = [SendableItemMetadata(ocId: "u0", fileName: "u0.txt", account: Self.account)] + + let first = makeBuffer() + first.prime(key: "initial", finalAnchorRawValue: Data("final".utf8), updated: updated, deleted: deleted) + let firstBatch = first.prepareChangeDeliveryBatch(maxItems: Self.batchSize) + let firstAnchor = try anchorKey(firstBatch.continuationAnchorRawValue) + + // The next request arrives on a new enumerator before the first one acknowledged its batch. + let second = makeBuffer() + XCTAssertTrue(second.isContinuation(forKey: firstAnchor)) + let secondBatch = second.prepareChangeDeliveryBatch(maxItems: Self.batchSize) + XCTAssertEqual(secondBatch.deleted.map(\.ocId), ["d1", "d2"]) + let secondAnchor = try anchorKey(secondBatch.continuationAnchorRawValue) + + first.acknowledgeBatch(firstBatch.acknowledgement) + second.acknowledgeBatch(secondBatch.acknowledgement) + + XCTAssertNil(Self.dbManager.itemMetadata(ocId: "d0")) + XCTAssertNil(Self.dbManager.itemMetadata(ocId: "d1"), "The second batch is acknowledged with its own deletions.") + XCTAssertNil(Self.dbManager.itemMetadata(ocId: "d2"), "The second batch is acknowledged with its own deletions.") + + let third = makeBuffer() + XCTAssertTrue(third.isContinuation(forKey: secondAnchor)) + let thirdBatch = third.prepareChangeDeliveryBatch(maxItems: Self.batchSize) + XCTAssertEqual(thirdBatch.deleted.map(\.ocId), ["d3"]) + XCTAssertFalse(thirdBatch.moreComing) + } + + func testLateAcknowledgementFromPreviousSessionDoesNotCommitTheNextSession() throws { + let db = Self.dbManager.ncDatabase() // Strong reference for the in-memory database. + debugPrint(db) + let firstFinalAnchor = Data("first-final".utf8) + + // A reused enumerator starts the next session before the previous session's acknowledgement ran. + let buffer = makeBuffer() + buffer.prime(key: "first", finalAnchorRawValue: firstFinalAnchor, updated: [], deleted: storeDeletedItems(["d0"])) + let firstBatch = buffer.prepareChangeDeliveryBatch(maxItems: Self.batchSize) + XCTAssertFalse(try buffer.isPrimed(forKey: anchorKey(firstFinalAnchor))) + buffer.reset() + buffer.prime(key: "second", finalAnchorRawValue: Data("second-final".utf8), updated: [], deleted: storeDeletedItems(["d1"])) + let secondBatch = buffer.prepareChangeDeliveryBatch(maxItems: Self.batchSize) + XCTAssertEqual(firstBatch.acknowledgement?.endSequence, secondBatch.acknowledgement?.endSequence) + + buffer.acknowledgeBatch(firstBatch.acknowledgement) + + XCTAssertNotNil(Self.dbManager.itemMetadata(ocId: "d1"), "The next session's batch stays unacknowledged.") + buffer.acknowledgeBatch(secondBatch.acknowledgement) + XCTAssertNil(Self.dbManager.itemMetadata(ocId: "d1")) + } +} diff --git a/shell_integration/MacOSX/NextcloudFileProviderKit/Tests/NextcloudFileProviderKitTests/EnumeratorTests.swift b/shell_integration/MacOSX/NextcloudFileProviderKit/Tests/NextcloudFileProviderKitTests/EnumeratorTests.swift index d775d2cf45fd6..daef8257eb7bb 100644 --- a/shell_integration/MacOSX/NextcloudFileProviderKit/Tests/NextcloudFileProviderKitTests/EnumeratorTests.swift +++ b/shell_integration/MacOSX/NextcloudFileProviderKit/Tests/NextcloudFileProviderKitTests/EnumeratorTests.swift @@ -26,6 +26,7 @@ final class EnumeratorTests: NextcloudFileProviderKitTestCase { var remoteTrashItemC: MockRemoteItem! static let dbManager = FilesDatabaseManager(account: account, databaseDirectory: makeDatabaseDirectory(), fileProviderDomainIdentifier: NSFileProviderDomainIdentifier("test"), log: FileProviderLogMock()) + static let acknowledgementTimeout: TimeInterval = 5 override func setUp() { super.setUp() @@ -1013,7 +1014,10 @@ final class EnumeratorTests: NextcloudFileProviderKitTestCase { let dbItemCMetadata = try XCTUnwrap( Self.dbManager.itemMetadata(ocId: remoteItemC.identifier) ) - XCTAssertNil(Self.dbManager.itemMetadata(ocId: remoteItemB.identifier)) + let itemBIdentifier = remoteItemB.identifier + await waitForAcknowledgement("The deleted item is removed.") { + Self.dbManager.itemMetadata(ocId: itemBIdentifier) == nil + } XCTAssertEqual(dbFolderMetadata.etag, remoteFolder.versionIdentifier) XCTAssertTrue(dbFolderMetadata.downloaded) XCTAssertEqual(dbItemAMetadata.etag, remoteItemA.versionIdentifier) @@ -1347,7 +1351,10 @@ final class EnumeratorTests: NextcloudFileProviderKitTestCase { try await observer.enumerateChanges() XCTAssertEqual(observer.changedItems.count, 0) XCTAssertEqual(observer.deletedItemIdentifiers.count, 1) - XCTAssertEqual(Self.dbManager.itemMetadata(ocId: remoteTrashItemA.identifier)?.deleted, true) + let trashItemIdentifier = remoteTrashItemA.identifier + await waitForAcknowledgement("The orphaned trash item is marked deleted.") { + Self.dbManager.itemMetadata(ocId: trashItemIdentifier)?.deleted == true + } } func testTrashItemEnumerationFailWhenNoTrashInCapabilities() async throws { @@ -2494,9 +2501,10 @@ final class EnumeratorTests: NextcloudFileProviderKitTestCase { previousTotal = total } - // Every delivered deletion's row is hard-removed by the time its (possibly later) batch finishes. - for deletedOcId in expectedDeleted { - XCTAssertNil(Self.dbManager.itemMetadata(ocId: deletedOcId), "Delivered deletions must be hard-removed across batch boundaries.") + // Every delivered deletion's row is hard-removed across batch boundaries. + let deletedOcIds = expectedDeleted + await waitForAcknowledgement("Delivered deletions are hard-removed.") { + deletedOcIds.allSatisfy { Self.dbManager.itemMetadata(ocId: $0) == nil } } // Anchors stay within the framework's 500-byte page/anchor limit. @@ -2579,9 +2587,20 @@ final class EnumeratorTests: NextcloudFileProviderKitTestCase { } XCTAssertEqual(observer.finishes.last?.moreComing, false) - // Delivered orphans are soft-deleted by the time their batch finishes. - for ocId in expectedOcIds { - XCTAssertEqual(Self.dbManager.itemMetadata(ocId: ocId)?.deleted, true, "Delivered trash orphans must be soft-deleted.") + // Delivered orphans are soft-deleted. + let orphanOcIds = expectedOcIds + await waitForAcknowledgement("Delivered trash orphans are soft-deleted.") { + orphanOcIds.allSatisfy { Self.dbManager.itemMetadata(ocId: $0)?.deleted == true } } } + + /// + /// Wait for database state which the enumerator writes when it acknowledges a batch. That happens only after + /// the observer received the batch, so it can still be pending when `enumerateChanges()` returns. + /// + private func waitForAcknowledgement(_ description: String, _ condition: @escaping @Sendable () -> Bool) async { + let expectation = XCTNSPredicateExpectation(predicate: NSPredicate { _, _ in condition() }, object: NSObject()) + expectation.expectationDescription = description + await fulfillment(of: [expectation], timeout: Self.acknowledgementTimeout) + } } diff --git a/shell_integration/MacOSX/NextcloudFileProviderKit/Tests/NextcloudFileProviderKitTests/FilesDatabaseManagerTests.swift b/shell_integration/MacOSX/NextcloudFileProviderKit/Tests/NextcloudFileProviderKitTests/FilesDatabaseManagerTests.swift index 20c00301e5e96..831cfbefe33e0 100644 --- a/shell_integration/MacOSX/NextcloudFileProviderKit/Tests/NextcloudFileProviderKitTests/FilesDatabaseManagerTests.swift +++ b/shell_integration/MacOSX/NextcloudFileProviderKit/Tests/NextcloudFileProviderKitTests/FilesDatabaseManagerTests.swift @@ -15,6 +15,7 @@ final class FilesDatabaseManagerTests: NextcloudFileProviderKitTestCase { ) static let dbManager = FilesDatabaseManager(account: account, databaseDirectory: makeDatabaseDirectory(), fileProviderDomainIdentifier: NSFileProviderDomainIdentifier("test"), log: FileProviderLogMock()) + static let changeDeliveryContainerKey = "working-set" override func setUp() { super.setUp() @@ -36,7 +37,7 @@ final class FilesDatabaseManagerTests: NextcloudFileProviderKitTestCase { XCTAssertTrue( Self.dbManager.createChangeDeliverySession( sessionId: sessionId, - containerKey: "working-set", + containerKey: Self.changeDeliveryContainerKey, anchorKey: "initial-anchor-\(name)", finalAnchorRawValue: Data("final-anchor".utf8), updated: [metadata], @@ -55,28 +56,113 @@ final class FilesDatabaseManagerTests: NextcloudFileProviderKitTestCase { 1 ) + let batch = Self.dbManager.prepareChangeDeliveryBatch(sessionId: sessionId, maxItems: 1) { "continuation|\($0)" } + XCTAssertEqual(batch?.endSequence, 1) + XCTAssertEqual(batch?.moreComing, false) + XCTAssertTrue(Self.dbManager.acknowledgeChangeDeliveryBatch(sessionId: sessionId, endSequence: 1)) + + let cleanedDatabase = Self.dbManager.ncDatabase() + XCTAssertNil(cleanedDatabase.object(ofType: RealmChangeDeliverySession.self, forPrimaryKey: sessionId)) XCTAssertTrue( - Self.dbManager.prepareChangeDeliveryBatch( - sessionId: sessionId, - endSequence: 1, - nextAnchorKey: nil, - moreComing: false - ) + cleanedDatabase.objects(RealmChangeDeliveryItem.self) + .where { $0.sessionId == sessionId } + .isEmpty ) + } + + /// + /// Store a change-delivery session with the given updates followed by deletions of stored items. + /// + private func createChangeDeliverySession(_ sessionId: String, updatedCount: Int, deletedOcIds: [String]) { + let updated = (0 ..< updatedCount).map { + SendableItemMetadata(ocId: "\(sessionId)-updated-\($0)", fileName: "updated.txt", account: Self.account) + } + let deleted = deletedOcIds.map { ocId in + var metadata = SendableItemMetadata(ocId: ocId, fileName: "\(ocId).txt", account: Self.account) + metadata.deleted = true + Self.dbManager.addItemMetadata(metadata) + return metadata + } + XCTAssertTrue( - Self.dbManager.acknowledgeChangeDeliveryBatch( + Self.dbManager.createChangeDeliverySession( sessionId: sessionId, - deletedOcIds: [] + containerKey: Self.changeDeliveryContainerKey, + anchorKey: "initial-anchor-\(sessionId)", + finalAnchorRawValue: Data("final-anchor-\(sessionId)".utf8), + updated: updated, + deleted: deleted, + incomplete: false, + hardRemoveDeleted: true ) ) + } - let cleanedDatabase = Self.dbManager.ncDatabase() - XCTAssertNil(cleanedDatabase.object(ofType: RealmChangeDeliverySession.self, forPrimaryKey: sessionId)) + func testRemovingChangeDeliverySessionsRemovesTheirItems() { + let sessionId = "removed-\(name)" XCTAssertTrue( - cleanedDatabase.objects(RealmChangeDeliveryItem.self) - .where { $0.sessionId == sessionId } - .isEmpty + Self.dbManager.createChangeDeliverySession( + sessionId: sessionId, + containerKey: Self.changeDeliveryContainerKey, + anchorKey: "initial-anchor-\(name)", + finalAnchorRawValue: Data("final-anchor".utf8), + updated: [SendableItemMetadata(ocId: "removed-item-\(name)", fileName: "item.txt", account: Self.account)], + deleted: [], + incomplete: false, + hardRemoveDeleted: true + ) ) + + Self.dbManager.removeChangeDeliverySessions(containerKey: Self.changeDeliveryContainerKey) + + let database = Self.dbManager.ncDatabase() + XCTAssertNil(database.object(ofType: RealmChangeDeliverySession.self, forPrimaryKey: sessionId)) + XCTAssertTrue(database.objects(RealmChangeDeliveryItem.self).where { $0.sessionId == sessionId }.isEmpty) + } + + func testChangeDeliveryContinuationKeyMatchesReportedAndAcknowledgedBatch() throws { + let sessionId = "continuation-\(name)" + createChangeDeliverySession(sessionId, updatedCount: 3, deletedOcIds: []) + let batch = try XCTUnwrap(Self.dbManager.prepareChangeDeliveryBatch(sessionId: sessionId, maxItems: 1) { "continuation|\($0)" }) + let anchorKey = try XCTUnwrap(batch.nextAnchorKey) + + let reported = try XCTUnwrap(Self.dbManager.changeDeliverySession(forContinuationKey: anchorKey, containerKey: Self.changeDeliveryContainerKey)) + XCTAssertEqual(reported.pendingAnchorKey, anchorKey) + XCTAssertNotEqual(reported.currentAnchorKey, anchorKey) + + XCTAssertTrue(Self.dbManager.acknowledgeChangeDeliveryBatch(sessionId: sessionId, endSequence: batch.endSequence)) + + let acknowledged = try XCTUnwrap(Self.dbManager.changeDeliverySession(forContinuationKey: anchorKey, containerKey: Self.changeDeliveryContainerKey)) + XCTAssertEqual(acknowledged.currentAnchorKey, anchorKey) + XCTAssertFalse(acknowledged.pendingReported) + } + + func testStaleChangeDeliveryAcknowledgementLeavesNewerBatchReported() throws { + let sessionId = "stale-acknowledgement-\(name)" + createChangeDeliverySession(sessionId, updatedCount: 1, deletedOcIds: ["\(name)-d0", "\(name)-d1"]) + let first = try XCTUnwrap(Self.dbManager.prepareChangeDeliveryBatch(sessionId: sessionId, maxItems: 1) { "continuation|\($0)" }) + XCTAssertTrue(Self.dbManager.acknowledgeChangeDeliveryBatch(sessionId: sessionId, endSequence: first.endSequence)) + let second = try XCTUnwrap(Self.dbManager.prepareChangeDeliveryBatch(sessionId: sessionId, maxItems: 1) { "continuation|\($0)" }) + + XCTAssertTrue(Self.dbManager.acknowledgeChangeDeliveryBatch(sessionId: sessionId, endSequence: first.endSequence)) + + let session = try XCTUnwrap(Self.dbManager.changeDeliverySession(sessionId: sessionId)) + XCTAssertTrue(session.pendingReported) + XCTAssertEqual(session.pendingEndSequence, second.endSequence) + XCTAssertEqual(session.currentAnchorKey, first.nextAnchorKey) + XCTAssertNotNil(Self.dbManager.itemMetadata(ocId: "\(name)-d0"), "The deletion of the unacknowledged batch is kept.") + } + + func testRepeatedChangeDeliveryPreparationReplaysReportedBatch() throws { + let sessionId = "replay-\(name)" + createChangeDeliverySession(sessionId, updatedCount: 3, deletedOcIds: []) + let first = try XCTUnwrap(Self.dbManager.prepareChangeDeliveryBatch(sessionId: sessionId, maxItems: 2) { "continuation|\($0)" }) + + let replayed = try XCTUnwrap(Self.dbManager.prepareChangeDeliveryBatch(sessionId: sessionId, maxItems: 1) { "continuation|\($0)" }) + + XCTAssertEqual(replayed.endSequence, first.endSequence) + XCTAssertEqual(replayed.nextAnchorKey, first.nextAnchorKey) + XCTAssertEqual(replayed.items.map(\.metadataData), first.items.map(\.metadataData)) } func testSchema203MigrationBackfillsCanonicalPathKeys() throws {