mirror of
https://repo.dactyloidae.xyz/Dactyloidae/UXP.git
synced 2026-09-09 01:08:39 +09:00
161 lines
7.1 KiB
Swift
161 lines
7.1 KiB
Swift
/* This Source Code Form is subject to the terms of the Mozilla Public
|
|
* License, v. 2.0. If a copy of the MPL was not distributed with this
|
|
* file, You can obtain one at http://mozilla.org/MPL/2.0/. */
|
|
|
|
import Foundation
|
|
import Shared
|
|
import Storage
|
|
|
|
private let log = Logger.syncLogger
|
|
let BookmarksStorageVersion = 2
|
|
|
|
/**
|
|
* This is like a synchronizer, but it downloads records bit by bit, eventually
|
|
* notifying that the local storage is up to date with the server contents.
|
|
*
|
|
* Because batches might be separated over time, it's possible for the server
|
|
* state to change between calls. These state changes might include:
|
|
*
|
|
* 1. New changes arriving. This is fairly routine, but it's worth noting that
|
|
* changes might affect existing records that have been batched!
|
|
* 2. Wipes. The collection (or the server as a whole) might be deleted. This
|
|
* should be accompanied by a change in syncID in meta/global; it's the caller's
|
|
* responsibility to detect this.
|
|
* 3. A storage format change. This should be unfathomably rare, but if it happens
|
|
* we must also be prepared to discard our existing batched data.
|
|
* 4. TTL expiry. We need to do better about TTL handling in general, but here
|
|
* we might find that a downloaded record is no longer live by the time we
|
|
* come to apply it! This doesn't apply to bookmark records, so we will ignore
|
|
* it for the moment.
|
|
*
|
|
* Batch downloading without continuation tokens is achieved as follows:
|
|
*
|
|
* * A minimum timestamp is established. This starts as zero.
|
|
* * A fetch is issued against the server for records changed since that timestamp,
|
|
* ordered by modified time ascending, and limited to the batch size.
|
|
* * If the batch is complete, we flush it to storage and advance the minimum
|
|
* timestamp to just before the newest record in the batch. This ensures that
|
|
* a divided set of records with the same modified times will be downloaded
|
|
* entirely so long as the set is never larger than the batch size.
|
|
* * Iterate until we determine that there are no new records to fetch.
|
|
*
|
|
* Batch downloading with continuation tokens is much easier:
|
|
*
|
|
* * A minimum timestamp is established.
|
|
* * Make a request with limit=N.
|
|
* * Look for an X-Weave-Next-Offset header. Supply that in the next request.
|
|
* Also supply X-If-Unmodified-Since to avoid missed modifications.
|
|
*
|
|
* We do the latter, because we only support Sync 1.5. The use of the offset
|
|
* allows us to efficiently process batches, particularly those that contain
|
|
* large sets of records with the same timestamp. We still maintain the last
|
|
* modified timestamp to allow for resuming a batch in the case of a conflicting
|
|
* write, detected via X-I-U-S.
|
|
*/
|
|
|
|
public class BookmarksMirrorer {
|
|
private let downloader: BatchingDownloader<BookmarkBasePayload>
|
|
private let storage: BookmarkBufferStorage
|
|
private let batchSize: Int
|
|
private let statsSession: SyncEngineStatsSession
|
|
|
|
public init(storage: BookmarkBufferStorage, client: Sync15CollectionClient<BookmarkBasePayload>, basePrefs: Prefs, collection: String, statsSession: SyncEngineStatsSession, batchSize: Int=100) {
|
|
self.storage = storage
|
|
self.downloader = BatchingDownloader(collectionClient: client, basePrefs: basePrefs, collection: collection)
|
|
self.batchSize = batchSize
|
|
self.statsSession = statsSession
|
|
}
|
|
|
|
var lastModified: Timestamp {
|
|
get {
|
|
return self.downloader.lastModified
|
|
}
|
|
}
|
|
|
|
// TODO
|
|
public func storageFormatDidChange() {
|
|
}
|
|
|
|
// TODO
|
|
public func onWipeWasAppliedToStorage() {
|
|
}
|
|
|
|
private func applyRecordsFromBatcher() -> Success {
|
|
let retrieved = self.downloader.retrieve()
|
|
let invalid = retrieved.filter { !$0.payload.isValid() }
|
|
|
|
if !invalid.isEmpty {
|
|
// There's nothing we can do with invalid input. There's also no point in
|
|
// tracking failing GUIDs here yet: if another client reuploads those records
|
|
// correctly, we'll encounter them routinely due to a newer timestamp.
|
|
// The only improvement we could make is to drop records from the buffer if we
|
|
// happen to see a new, invalid one before somehow syncing again, but that's
|
|
// unlikely enough that it's not worth doing.
|
|
//
|
|
// Bug 1258801 tracks recording telemetry for these invalid items, which is
|
|
// why we don't simply drop them on the ground at the download stage.
|
|
//
|
|
// We might also choose to perform certain simple recovery actions here: for example,
|
|
// bookmarks with null URIs are clearly invalid, and could be treated as if they
|
|
// weren't present on the server, or transparently deleted.
|
|
log.warning("Invalid records: \(invalid.map { $0.id }.joined(separator: ", ")).")
|
|
}
|
|
|
|
let mirrorItems = retrieved.flatMap { record -> BookmarkMirrorItem? in
|
|
guard record.payload.isValid() else {
|
|
return nil
|
|
}
|
|
|
|
return (record.payload as MirrorItemable).toMirrorItem(record.modified)
|
|
}
|
|
|
|
if mirrorItems.isEmpty {
|
|
log.debug("Got empty batch.")
|
|
return succeed()
|
|
}
|
|
|
|
log.debug("Applying \(mirrorItems.count) downloaded bookmarks.")
|
|
return self.storage.applyRecords(mirrorItems)
|
|
}
|
|
|
|
public func go(info: InfoCollections, greenLight: @escaping () -> Bool) -> SyncResult {
|
|
if !greenLight() {
|
|
log.info("Green light turned red. Stopping mirror operation.")
|
|
return deferMaybe(SyncStatus.notStarted(.redLight))
|
|
}
|
|
|
|
log.debug("Downloading up to \(self.batchSize) records.")
|
|
return self.downloader.go(info, limit: self.batchSize)
|
|
.bind { result in
|
|
guard let end = result.successValue else {
|
|
log.warning("Got failure: \(result.failureValue!)")
|
|
return deferMaybe(result.failureValue!)
|
|
}
|
|
switch end {
|
|
case .complete:
|
|
log.info("Done with batched mirroring.")
|
|
return self.applyRecordsFromBatcher()
|
|
>>> effect(self.downloader.advance)
|
|
>>> self.storage.doneApplyingRecordsAfterDownload
|
|
>>> always(SyncStatus.completed(self.statsSession.end()))
|
|
case .incomplete:
|
|
log.debug("Running another batch.")
|
|
// This recursion is fine because Deferred always pushes callbacks onto a queue.
|
|
return self.applyRecordsFromBatcher()
|
|
>>> effect(self.downloader.advance)
|
|
>>> { self.go(info: info, greenLight: greenLight) }
|
|
case .interrupted:
|
|
log.info("Interrupted. Aborting batching this time.")
|
|
return deferMaybe(SyncStatus.partial(self.statsSession))
|
|
case .noNewData:
|
|
log.info("No new data. No need to continue batching.")
|
|
self.downloader.advance()
|
|
return deferMaybe(SyncStatus.completed(self.statsSession.end()))
|
|
}
|
|
}
|
|
}
|
|
|
|
func advanceNextDownloadTimestampTo(timestamp: Timestamp) {
|
|
self.downloader.advanceTimestampTo(timestamp)
|
|
}
|
|
}
|