rzen/workouts
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Queue document writes durably and surface sync trouble in the UI

Browse Source 
iCloud Drive writes now flow through a persistent WriteBacklog sidecar
(drained with backoff, flushed on backgrounding, wiped with the cache on
account change), so a save can never be lost to a transient coordinator
error. A status banner on the workout list surfaces stuck syncing.
Also: the split picker gains a Recent section with day labels, split
rows fold SplitItem into SplitListView, and list rows dim the multiply
sign in sets-by-reps.

Claude-Session: https://claude.ai/code/session_01HJDQQDA9QdP8zByg43H5v3
This commit is contained in:
rzen
2026-07-08 12:48:12 -04:00
parent 495fce1e5a
commit c05e83cff7
13 changed files with 1175 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Workouts/Sync/SyncEngine.swift
+256 -49
View File
@@ -11,12 +11,16 @@ enum ICloudStatus: Equatable {
}
/// Orchestrates the iCloud Drive file layer and the SwiftData cache. iCloud is
/// the sole source of truth: every save/delete writes the file first, then
/// mirrors the change into the cache immediately (a same-process write doesn't
/// reliably wake the `NSMetadataQuery` observer — and never does in the
/// simulator — so waiting on it leaves the UI blind to the user's own action).
/// the sole source of truth. Every save/delete mirrors into the cache
/// immediately (a same-process write doesn't reliably wake the `NSMetadataQuery`
/// observer — and never does in the simulator — so waiting on it leaves the UI
/// blind to the user's own action) and queues the file write in a per-document
/// slot backlog (`WriteBacklog`), drained serially with backoff. A failed write
/// therefore never silently evaporates an edit: it stays in its slot, the cache
/// keeps showing it, and `writeQueueState` surfaces the stall to the UI.
/// The observer and the connect-time reconcile re-apply idempotently and remain
/// the sole channel for *remote* changes.
/// the sole channel for *remote* changes; both skip documents superseded by a
/// queued-but-unwritten local change.
@Observable
@MainActor
final class SyncEngine {
@@ -42,6 +46,12 @@ final class SyncEngine {
/// next successful write or full reconcile.
private(set) var lastSyncError: String?
/// Rollup of the write backlog for the UI banner tiers (see `WriteBacklog`).
private(set) var writeQueueState: WriteQueueState = .idle
/// Number of queued-but-unwritten document writes.
var pendingWriteCount: Int { backlog.count }
/// Maps a seed's id to its clone's id after a clone-on-edit fork, so a view still
/// holding the seed's id resolves to the live clone. In-memory only (@Observable
/// notifies on change); not persisted — durability comes from `repointWorkouts`,
@@ -71,6 +81,9 @@ final class SyncEngine {
private var monitor: MetadataObserver?
private var monitorTask: Task<Void, Never>?
private var connectAttempt = 0
private var backlog: WriteBacklog
private let backlogURL: URL
private var drainTask: Task<Void, Never>?
/// How long `connect()` keeps polling for a still-provisioning iCloud
/// container before falling to the end-of-the-line gate. Deliberately long
@@ -81,8 +94,17 @@ final class SyncEngine {
private var context: ModelContext { modelContainer.mainContext }
init(container: ModelContainer) {
init(container: ModelContainer, backlogURL: URL = WorkoutsModelContainer.pendingWritesURL) {
self.modelContainer = container
self.backlogURL = backlogURL
// Writes queued in a previous run survive relaunch; anything stuck past
// the TTL is dropped rather than drained stale into a container that has
// moved on. (An account change wipes the sidecar file before this load —
// see `WorkoutsModelContainer.wipeIfAccountChanged`.)
var loaded = WriteBacklogFile.load(from: backlogURL)
loaded.pruneExpired(now: Date())
self.backlog = loaded
refreshQueueState()
}
// MARK: - Connection (deferred, patient)
@@ -165,6 +187,10 @@ final class SyncEngine {
log.info("connect[\(attempt)]: directories ready → available")
WorkoutsModelContainer.persistCurrentIdentityToken()
// Deliver any backlog surviving from a previous run before reconcile
// lists the tree — reconcile skips pending ids anyway, but the sooner
// queued edits hit disk the smaller that window is.
kickDrain()
await reconcile()
startMonitoring(documentsURL: store.rootURL)
cleanupOldStubs()
@@ -204,7 +230,14 @@ final class SyncEngine {
monitorTask = nil
monitor?.stop()
monitor = nil
log.info("restore: began — metadata observer suspended")
// Queued writes reference pre-restore state; draining them into the
// restored tree would corrupt it with edits the user chose to roll back.
drainTask?.cancel()
drainTask = nil
backlog.removeAll()
persistBacklog()
refreshQueueState()
log.info("restore: began — metadata observer suspended, write backlog cleared")
}
/// Resume live file-watching after a restore. Starts a FRESH metadata observer
@@ -257,8 +290,8 @@ final class SyncEngine {
onCacheChanged?()
}
/// Apply a workout received from the watch — `save(workout:)` writes the file
/// and mirrors it into the cache, same as a local edit.
/// Apply a workout received from the watch — `save(workout:)` mirrors it into
/// the cache and queues the file write, same as a local edit.
func ingestFromWatch(_ doc: WorkoutDocument) async {
// A workout deleted on the phone leaves a tombstone; a watch that missed the
// delete may still push its stale copy. Honor the veto — never resurrect it —
@@ -268,18 +301,50 @@ final class SyncEngine {
onCacheChanged?()
return
}
// A queued-but-unwritten delete is the same veto — its stub just hasn't
// landed on disk yet.
if case .delete = backlog.pendingWrite(for: doc.id)?.payload {
onCacheChanged?()
return
}
// `updatedAt` intake gate: the watch mixes `sendMessage` with a queued
// `transferUserInfo` fallback, which are unordered — a failed-then-queued
// older edit can arrive after a newer one. Accept only what's strictly
// newer than the cache; the cache mirrors every accepted write
// immediately, so it's always at least as new as any pending slot.
// Strictly older means the watch is behind — re-push authoritative state
// so it corrects. Equal is the duplicate/echo case — ignore silently.
//
// KNOWN LIMIT (deliberate, revisit if users report vanished sets): this
// gate arbitrates *timestamps*, not *content*. A doc edited on a stale
// watch snapshot carries a fresh stamp and replaces the whole workout —
// concurrent edits inside one push round-trip (or a disconnection) lose
// one side wholesale. The watch-side absorb (WorkoutLogListView's
// onChange re-seed) narrows the stale window to one push latency; the
// durable fix is per-log merge: give WorkoutLogDocument its own optional
// `updatedAt` (same additive pattern as startedAt/completedAt), merge
// incoming docs log-by-log (newer log wins per id) instead of replacing,
// and carry log add/remove as explicit intents so an absent log is never
// ambiguous between "deleted" and "not seen yet". That makes edits to
// different logs commute, so delivery order and offline gaps stop
// mattering entirely.
if let cached = CacheMapper.fetchWorkout(id: doc.id, in: context) {
if doc.updatedAt < cached.updatedAt {
onCacheChanged?()
return
}
if doc.updatedAt == cached.updatedAt { return }
}
await save(workout: doc)
}
// MARK: - Public CRUD (write path: files only)
// MARK: - Public CRUD (mirror-first: cache now, file via the write queue)
/// Returns the *effective* id of the split that was written — normally `doc.id`,
/// but the clone's fresh id when an edited seed forks. Open views follow the swap
/// via `currentSplitID(for:)`.
@discardableResult
func save(split doc: SplitDocument) async -> String {
guard let store else { return doc.id }
// Seeds are immutable: a real edit forks the seed into a user-owned split and
// soft-deletes the original, keeping the curated seed intact and restorable. A
// pristine (no-op) save must NOT fork — the edit sheets stamp `updatedAt`
@@ -289,14 +354,9 @@ final class SyncEngine {
return await cloneSeedOnEdit(doc)
}
do {
try await store.write(doc, to: doc.relativePath)
CacheMapper.upsertSplit(doc, relativePath: doc.relativePath, into: context)
saveCacheAndNotify()
lastSyncError = nil
} catch {
report("Failed to save split", error)
}
CacheMapper.upsertSplit(doc, relativePath: doc.relativePath, into: context)
saveCacheAndNotify()
enqueueWrite(.split(doc), timestamp: doc.updatedAt)
return doc.id
}
@@ -377,41 +437,33 @@ final class SyncEngine {
}
func save(workout doc: WorkoutDocument) async {
guard let store else { return }
// The month bucket in a workout's path derives from `start`, so editing the
// start date (or a device in a different time zone) can move the file to a new
// path. Capture the previously-written path before the upsert overwrites it, so
// we can remove the stale file below — otherwise the same id would live at two
// paths and the old copy would re-import on the next reconcile.
// path. Capture the previously-written path before the upsert overwrites it —
// the queued write removes it after landing, otherwise the same id would live
// at two paths and the old copy would re-import on the next reconcile.
let previousPath = CacheMapper.fetchWorkout(id: doc.id, in: context)?.jsonRelativePath
do {
try await store.write(doc, to: doc.relativePath)
CacheMapper.upsertWorkout(doc, relativePath: doc.relativePath, into: context)
// Same id, new path: drop the orphaned file at the old bucket. The id lives
// on at the new path, so this is a plain removal — no tombstone (a tombstone
// would veto the record that just moved). Phone stays the sole writer.
if let previousPath, previousPath != doc.relativePath {
try? await store.remove(at: previousPath)
}
saveCacheAndNotify()
lastSyncError = nil
} catch {
report("Failed to save workout", error)
}
CacheMapper.upsertWorkout(doc, relativePath: doc.relativePath, into: context)
saveCacheAndNotify()
enqueueWrite(
.workout(doc),
timestamp: doc.updatedAt,
stalePath: previousPath != doc.relativePath ? previousPath : nil
)
}
func delete(split: Split) async {
let id = split.id, livePath = split.jsonRelativePath
await softDelete(id: id, kind: "split", livePath: livePath)
deleteCachedEntity(id: id)
saveCacheAndNotify()
enqueueWrite(.delete(id: id, kind: "split", livePath: livePath), timestamp: Date())
}
func delete(workout: Workout) async {
let id = workout.id, livePath = workout.jsonRelativePath
await softDelete(id: id, kind: "workout", livePath: livePath)
deleteCachedEntity(id: id)
saveCacheAndNotify()
enqueueWrite(.delete(id: id, kind: "workout", livePath: livePath), timestamp: Date())
}
/// Persist pending cache mutations and fan out the change notification — the
@@ -421,8 +473,10 @@ final class SyncEngine {
onCacheChanged?()
}
/// Writes a tombstone stub then removes the live file. Other devices learn of
/// the delete via the stub even if they were offline for the file removal.
/// Writes a tombstone stub then removes the live file, immediately and outside
/// the write queue. Used only by the multi-step flows that need their file ops
/// ordered within a larger transaction (`cloneSeedOnEdit`, `performCleanup`);
/// plain user deletes go through the queue as `.delete` payloads.
private func softDelete(id: String, kind: String, livePath: String) async {
guard let store, let tombstones else { return }
do {
@@ -434,6 +488,145 @@ final class SyncEngine {
}
}
// MARK: - Write queue
/// Queue a file write whose cache mirror has already been applied, persist the
/// backlog, and make sure the drain loop is running. The per-id slot keeps only
/// the newest version (`WriteBacklog.enqueue` is newer-wins on `timestamp`).
private func enqueueWrite(_ payload: PendingWrite.Payload, timestamp: Date, stalePath: String? = nil) {
var write = PendingWrite(payload: payload, timestamp: timestamp, enqueuedAt: Date())
if let stalePath { write.stalePaths.insert(stalePath) }
backlog.enqueue(write)
persistBacklog()
refreshQueueState()
kickDrain()
}
private func kickDrain() {
guard drainTask == nil, store != nil, !isRestoring else { return }
drainTask = Task { [weak self] in
await self?.drainBacklog()
guard let self else { return }
self.drainTask = nil
// A write enqueued in the same beat the loop was exiting finds
// `drainTask` still set and skips its kick — re-check. No spin: the
// loop only exits non-empty when the kick guard blocks anyway.
if !self.backlog.isEmpty { self.kickDrain() }
}
}
/// Serial delivery of the backlog: one write at a time, oldest due slot first,
/// exponential backoff per entry. Sleeps in short quanta while nothing is due
/// so a fresh enqueue never waits out a long backoff.
private func drainBacklog() async {
while store != nil, !isRestoring, !backlog.isEmpty {
guard let entry = backlog.nextDue(at: Date()) else {
try? await Task.sleep(for: .seconds(1))
continue
}
await attempt(entry)
}
}
/// Try every pending write immediately, ignoring backoff — the app is about to
/// background (last chance before suspension), or a test needs determinism.
func flushPendingWrites() async {
guard store != nil, !isRestoring else { return }
for entry in backlog.entries {
await attempt(entry)
}
}
private func attempt(_ entry: PendingWrite) async {
switch await perform(entry) {
case .success:
backlog.resolve(id: entry.documentID, ifTimestampAtMost: entry.timestamp)
case .retry:
backlog.markFailed(id: entry.documentID, at: Date())
case .fault(let message):
backlog.markFailed(id: entry.documentID, at: Date(), fault: message)
}
persistBacklog()
refreshQueueState()
}
private enum WriteOutcome {
case success
case retry
case fault(String)
}
private func perform(_ entry: PendingWrite) async -> WriteOutcome {
guard let store, let tombstones else { return .retry }
do {
switch entry.payload {
case .split(let doc):
try await store.write(doc, to: doc.relativePath)
case .workout(let doc):
try await store.write(doc, to: doc.relativePath)
case .delete(let id, let kind, let livePath):
// The stub is the authoritative delete record (it vetoes
// resurrection everywhere); the live-file removal is best-effort —
// the observer's stub handling and reconcile both reap a straggler.
try await tombstones.writeTombstone(Tombstone(id: id, deletedAt: entry.timestamp, kind: kind))
try? await store.remove(at: livePath)
}
for stale in entry.stalePaths where stale != entry.targetPath {
try? await store.remove(at: stale)
}
return .success
} catch {
log.error("write queue: \(entry.documentID, privacy: .public) failed (attempt \(entry.attempts + 1)): \(error)")
if let message = Self.unrecoverableDescription(for: error) { return .fault(message) }
return .retry
}
}
/// Errors no amount of retrying fixes — escalate the banner immediately.
/// (The drain still retries them on the long cadence; conditions can change.)
private static func unrecoverableDescription(for error: Error) -> String? {
let ns = error as NSError
guard ns.domain == NSCocoaErrorDomain else { return nil }
switch ns.code {
case NSFileWriteOutOfSpaceError:
return "iPhone storage is full — changes can't be saved to iCloud."
case NSFileWriteVolumeReadOnlyError:
return "iCloud storage is read-only — changes can't be saved."
default:
return nil
}
}
private func persistBacklog() {
WriteBacklogFile.save(backlog, to: backlogURL)
}
private func refreshQueueState() {
let state: WriteQueueState
if backlog.isEmpty {
state = .idle
} else if let fault = backlog.firstFaultMessage {
state = .fault(fault)
} else if backlog.maxAttempts >= WriteBacklog.tier2AttemptThreshold {
state = .fault("Changes aren't reaching iCloud.")
} else if backlog.maxAttempts >= WriteBacklog.tier1AttemptThreshold {
state = .retrying
} else {
state = .pending
}
if state != writeQueueState { writeQueueState = state }
}
/// True when a queued-but-unwritten local change for `id` is at least as new
/// as the file content just read — the cache already reflects the newer
/// pending version, so importing the older file would regress it until the
/// drain rewrites. (A pending `.delete` compares via its deletion stamp and
/// keeps the import from resurrecting the entity.)
private func supersededByPendingWrite(id: String, fileTimestamp: Date) -> Bool {
guard let pending = backlog.pendingWrite(for: id) else { return false }
return pending.timestamp >= fileTimestamp
}
// MARK: - Import / reconcile
private func importFile(relativePath: String) async {
@@ -453,10 +646,12 @@ final class SyncEngine {
if relativePath.hasPrefix("Splits/") {
guard let doc = try? DocumentCoder.decode(SplitDocument.self, from: data), doc.isReadable else { return }
if await tombstones.stubExists(id: doc.id) { try? await store.remove(at: relativePath); return }
if supersededByPendingWrite(id: doc.id, fileTimestamp: doc.updatedAt) { return }
CacheMapper.upsertSplit(doc, relativePath: relativePath, into: context)
} else if relativePath.hasPrefix("Workouts/") {
guard let doc = try? DocumentCoder.decode(WorkoutDocument.self, from: data), doc.isReadable else { return }
if await tombstones.stubExists(id: doc.id) { try? await store.remove(at: relativePath); return }
if supersededByPendingWrite(id: doc.id, fileTimestamp: doc.updatedAt) { return }
CacheMapper.upsertWorkout(doc, relativePath: relativePath, into: context)
}
}
@@ -496,26 +691,34 @@ final class SyncEngine {
if path.hasPrefix("Splits/") {
guard let doc = try? DocumentCoder.decode(SplitDocument.self, from: data), doc.isReadable else { continue }
if tombstoned.contains(doc.id) { try? await store.remove(at: path); continue }
CacheMapper.upsertSplit(doc, relativePath: path, into: context)
liveSplitIDs.insert(doc.id)
if supersededByPendingWrite(id: doc.id, fileTimestamp: doc.updatedAt) { continue }
CacheMapper.upsertSplit(doc, relativePath: path, into: context)
} else if path.hasPrefix("Workouts/") {
guard let doc = try? DocumentCoder.decode(WorkoutDocument.self, from: data), doc.isReadable else { continue }
if tombstoned.contains(doc.id) { try? await store.remove(at: path); continue }
CacheMapper.upsertWorkout(doc, relativePath: path, into: context)
liveWorkoutIDs.insert(doc.id)
if supersededByPendingWrite(id: doc.id, fileTimestamp: doc.updatedAt) { continue }
CacheMapper.upsertWorkout(doc, relativePath: path, into: context)
}
}
// Prune cache entities no longer backed by a live file — but never for
// a path that failed to read this pass (it may just be un-downloadable
// right now; pruning would make an eviction look like a deletion).
// right now; pruning would make an eviction look like a deletion), and
// never for an id with a queued-but-unwritten write (its file simply
// hasn't landed yet; pruning would evaporate the pending edit's mirror).
if let splits = try? context.fetch(FetchDescriptor<Split>()) {
for s in splits where !liveSplitIDs.contains(s.id) && !unreadablePaths.contains(s.jsonRelativePath) {
for s in splits where !liveSplitIDs.contains(s.id)
&& !unreadablePaths.contains(s.jsonRelativePath)
&& backlog.pendingWrite(for: s.id) == nil {
context.delete(s)
}
}
if let workouts = try? context.fetch(FetchDescriptor<Workout>()) {
for w in workouts where !liveWorkoutIDs.contains(w.id) && !unreadablePaths.contains(w.jsonRelativePath) {
for w in workouts where !liveWorkoutIDs.contains(w.id)
&& !unreadablePaths.contains(w.jsonRelativePath)
&& backlog.pendingWrite(for: w.id) == nil {
context.delete(w)
}
}
@@ -537,12 +740,16 @@ final class SyncEngine {
if let w = CacheMapper.fetchWorkout(id: id, in: context) { context.delete(w) }
}
/// Observer `.removed` handling. Skips entities with a queued-but-unwritten
/// write: their file may simply not have landed yet (or an old-path removal
/// raced the rewrite), and the pending edit's mirror must survive until the
/// drain delivers it.
private func deleteCachedEntity(jsonRelativePath path: String) {
if let splits = try? context.fetch(FetchDescriptor<Split>(predicate: #Predicate { $0.jsonRelativePath == path })) {
splits.forEach(context.delete)
splits.filter { backlog.pendingWrite(for: $0.id) == nil }.forEach(context.delete)
}
if let workouts = try? context.fetch(FetchDescriptor<Workout>(predicate: #Predicate { $0.jsonRelativePath == path })) {
workouts.forEach(context.delete)
workouts.filter { backlog.pendingWrite(for: $0.id) == nil }.forEach(context.delete)
}
}