| 1 | import Foundation |
| 2 | import RastrilloNative |
| 3 | |
| 4 | struct PersistedReplica: Codable, Sendable { |
| 5 | let format: Int |
| 6 | let sequence: Int64 |
| 7 | let guid: String |
| 8 | let savedAt: Date |
| 9 | let state: Data |
| 10 | } |
| 11 | |
| 12 | /// One document's durability boundary. The writer and editor share this |
| 13 | /// watermark, but only a successful outbox write may advance it. |
| 14 | @MainActor final class DocsPersistence { |
| 15 | let store: BlobStore |
| 16 | let outbox: Outbox |
| 17 | let key: String |
| 18 | let replica: CRDTReplica |
| 19 | private(set) var commandsDurable = 0 |
| 20 | private var active = true |
| 21 | let cache: DocsReplicaCache |
| 22 | private let generation: Int |
| 23 | var lane: String { "doc:" + key } |
| 24 | var path: String { "cache/docs/" + key + "/replica" } |
| 25 | init(store: BlobStore, outbox: Outbox, key: String, replica: CRDTReplica, cache: DocsReplicaCache? = nil) { |
| 26 | self.store = store; self.outbox = outbox; self.key = key; self.replica = replica |
| 27 | self.cache = cache ?? DocsReplicaCache(store: store) |
| 28 | generation = self.cache.generation(key) |
| 29 | } |
| 30 | func reopen(fallback: Data, sequence: Int64) async throws { |
| 31 | try await outbox.load() |
| 32 | let entries = await outbox.entries(lane: lane) |
| 33 | let saved = try store.read(path).map { try JSONDecoder().decode(PersistedReplica.self, from: $0) } |
| 34 | if let saved, saved.format != 1 { throw BlobStoreError.unavailable } |
| 35 | _ = try await replica.open(update: saved?.state ?? fallback) |
| 36 | await replica.setSequence(saved?.sequence ?? sequence) |
| 37 | // Replay includes held entries: the editor must show everything kept |
| 38 | // here, even when it cannot currently send it. |
| 39 | for entry in entries where entry.kind == "update" { _ = try await replica.apply(entry.payload) } |
| 40 | commandsDurable = 0 |
| 41 | } |
| 42 | func enqueue(_ update: Data) async throws { |
| 43 | try await outbox.enqueue(lane: lane, kind: "update", payload: update) |
| 44 | commandsDurable += 1 |
| 45 | } |
| 46 | @discardableResult func persist() async throws -> Bool { |
| 47 | let captured = try await replica.persistentState() |
| 48 | // No suspension between checking the watermark and the atomic write. |
| 49 | // A writer running between command and enqueue must defer its capture. |
| 50 | guard active, captured.commandsApplied == commandsDurable else { return false } |
| 51 | let saved = PersistedReplica(format: 1, sequence: captured.sequence, guid: captured.guid, savedAt: Date(), state: captured.state) |
| 52 | return try await cache.commit(saved, key: key, generation: generation) { |
| 53 | self.active && captured.commandsApplied == self.commandsDurable |
| 54 | } |
| 55 | } |
| 56 | func compact() async throws { |
| 57 | let entries = await outbox.sendable(lane: lane, maxBytes: nil) |
| 58 | guard entries.count > 256 || entries.reduce(0, { $0 + $1.payload.count }) > 1024 * 1024 else { return } |
| 59 | var batch: [OutboxEntry] = []; var bytes = 0 |
| 60 | for entry in entries { |
| 61 | if entry.kind != "update" { if batch.isEmpty { continue }; break } |
| 62 | if !batch.isEmpty && bytes + entry.payload.count > 512 * 1024 { break } |
| 63 | batch.append(entry); bytes += entry.payload.count |
| 64 | } |
| 65 | guard batch.count > 1 else { return } |
| 66 | let merged = try await replica.mergeUpdates(batch.map(\.payload)) |
| 67 | guard merged.count <= 512 * 1024 else { return } |
| 68 | // A send can reserve the batch while mergeUpdates suspends. Refusing |
| 69 | // that compaction is expected; it must never delete the reservation. |
| 70 | do { try await outbox.replace(batch.map(\.id), with: merged) } |
| 71 | catch OutboxError.inFlight { } |
| 72 | catch OutboxError.unknownEntry { } |
| 73 | } |
| 74 | func discard() async throws { |
| 75 | active = false |
| 76 | cache.invalidate(key) |
| 77 | // Remove the replica first: a crash may replay lane entries again, |
| 78 | // but can never leave local text whose lane has already been removed. |
| 79 | try store.remove(path) |
| 80 | try await outbox.discard(lane: lane) |
| 81 | } |
| 82 | } |
| 83 | |
amadan