| 1 | import CryptoKit |
| 2 | import Foundation |
| 3 | import RastrilloApple |
| 4 | import RastrilloTestSupport |
| 5 | import XCTest |
| 6 | @testable import DocsKit |
| 7 | |
| 8 | final class OfflineKillTests: XCTestCase { |
| 9 | @MainActor func testTypingProcessAndRecovery() async throws { |
| 10 | let env = ProcessInfo.processInfo.environment |
| 11 | guard let directory = env["DOCS_KILL_DIR"], let mode = env["DOCS_KILL_MODE"] else { |
| 12 | throw XCTSkip("Run native/scripts/test-offline-kill.py") |
| 13 | } |
| 14 | let randomKill = env["DOCS_KILL_RANDOM"] == "1" |
| 15 | let root = URL(fileURLWithPath: directory) |
| 16 | let config = try JSONDecoder().decode([String: String].self, from: Data(contentsOf: root.appendingPathComponent("connection.json"))) |
| 17 | let client = NativeHTTP(origin: URL(string: config["origin"]!)!, credential: NativeSession(name: config["name"]!, value: config["value"]!)) |
| 18 | let path = ScriptedPathSource(mode == "write" ? .satisfied : .unsatisfied) |
| 19 | let connectivity = NativeConnectivity(source: path) |
| 20 | let keys = ScriptedKeyProvider(key: SymmetricKey(data: Data(repeating: 73, count: 32))) |
| 21 | let store = NativeSealedStore(root: root.appendingPathComponent("account"), accountID: "kill-test", keys: keys) |
| 22 | let account = DocsOfflineAccount(store: store, accountID: "kill-test", connectivity: connectivity, transport: { client }) |
| 23 | let model = DocsModel(transport: { client }, connectivity: connectivity, offlineAccount: account) |
| 24 | await model.open(config["sqid"]!) |
| 25 | XCTAssertNotNil(model.projection) |
| 26 | if mode == "write" { |
| 27 | guard await model.saveAll() else { |
| 28 | XCTFail("Initial save failed: \(model.syncStatus)"); return |
| 29 | } |
| 30 | path.send(.unsatisfied) |
| 31 | while connectivity.isPathSatisfied { try await Task.sleep(for: .milliseconds(5)) } |
| 32 | try String(ProcessInfo.processInfo.processIdentifier).write(to: root.appendingPathComponent("pid"), atomically: true, encoding: .utf8) |
| 33 | // Exercise I1a with writer ticks racing command/enqueue. No sleep |
| 34 | // or handshake freezes the typing process in random mode. |
| 35 | let writer: Task<Void, Never>? = randomKill ? Task { |
| 36 | while !Task.isCancelled { await model.persistReplica(); await Task.yield() } |
| 37 | } : nil |
| 38 | defer { writer?.cancel() } |
| 39 | for index in 1...(randomKill ? 10_000 : 200) { |
| 40 | model.didRender(model.projectionGeneration) |
| 41 | let text = model.projection?.children.map(\.plainText).joined() ?? "" |
| 42 | XCTAssertTrue(model.enqueue(["op": .string("display_edit"), "at": .number(Double(text.utf16.count)), "length": .number(0), "text": .string("k3-\(index);")])) |
| 43 | while model.hasPendingInput { await Task.yield() } |
| 44 | if randomKill || index == 12 { |
| 45 | let expected = model.projection?.children.map(\.plainText).joined() ?? "" |
| 46 | try expected.write(to: root.appendingPathComponent("expected"), atomically: true, encoding: .utf8) |
| 47 | try Data().write(to: root.appendingPathComponent("kill-now"), options: .atomic) |
| 48 | // The controller kills us in the middle of this typing run, |
| 49 | // before the two-second replica writer can run. |
| 50 | if !randomKill { try await Task.sleep(for: .seconds(1)) } |
| 51 | } |
| 52 | if !randomKill { try await Task.sleep(for: .milliseconds(15)) } |
| 53 | } |
| 54 | XCTFail("The controller did not kill the typing process") |
| 55 | } else { |
| 56 | let expected = try String(contentsOf: root.appendingPathComponent("expected"), encoding: .utf8) |
| 57 | let reopened = model.projection?.children.map(\.plainText).joined() ?? "" |
| 58 | if randomKill { |
| 59 | XCTAssertFalse(expected.isEmpty, "The controller must observe durable typing before killing") |
| 60 | XCTAssertTrue(reopened.hasPrefix(expected), "Offline reopen must preserve the last recorded durable prefix") |
| 61 | let tokens = reopened.split(separator: ";") |
| 62 | XCTAssertEqual(reopened, (1...max(1, tokens.count)).map { "k3-\($0);" }.joined(), "Recovered text must have no gaps, torn edits, or duplicates") |
| 63 | } else { |
| 64 | XCTAssertEqual(reopened, expected, "Offline reopen must include every durable keystroke exactly once") |
| 65 | } |
| 66 | path.send(.satisfied) |
| 67 | while !connectivity.isPathSatisfied { try await Task.sleep(for: .milliseconds(5)) } |
| 68 | await model.retry() |
| 69 | let deadline = ContinuousClock.now.advanced(by: .seconds(10)) |
| 70 | while model.hasUnsavedChanges && .now < deadline { await model.flush(); try await Task.sleep(for: .milliseconds(10)) } |
| 71 | XCTAssertFalse(model.hasUnsavedChanges) |
| 72 | let reply = try await client.responseDetails("/d/\(config["sqid"]!)/updates", maxBytes: 8 * 1024 * 1024) |
| 73 | try reply.requireSuccess() |
| 74 | let server = CRDTReplica(guid: model.detail!.guid) |
| 75 | let projected = try await server.open(update: reply.data) |
| 76 | XCTAssertEqual(projected.children.map(\.plainText).joined(), reopened, "Real server converges after the killed client's lane drains") |
| 77 | await server.close(); await model.stop(); account.drainer.stop() |
| 78 | } |
| 79 | } |
| 80 | } |
| 81 | |
amadan