diff --git a/AGENTS.md b/AGENTS.md index f0b7f74b..dd2dbce7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -8,7 +8,7 @@ | SwiftFormat | 0.60.1 | `.mise.toml` | | Swift PM | 6.2 | `Package.swift` (`swift-tools-version`) | -**Libraries** (**StuffCore**, **LifecycleKit**, **LogKit**, **LogViewerUI**, **PeriscopeCore**, **PeriscopeUI**, **PeriscopeTools**, **TestHostSupport**, **RegionKit**, **WhereCore**, **WhereUI**) are defined in the root [`Package.swift`](Package.swift) — local package for libraries, Tuist for apps and test bundles. +**Libraries** (**StuffCore**, **LifecycleKit**, **JournalKit**, **LogKit**, **LogViewerUI**, **PeriscopeCore**, **PeriscopeUI**, **PeriscopeTools**, **TestHostSupport**, **RegionKit**, **WhereCore**, **WhereUI**) are defined in the root [`Package.swift`](Package.swift) — local package for libraries, Tuist for apps and test bundles. Tuist manifests live at the repo root ([`Project.swift`](Project.swift), [`Tuist.swift`](Tuist.swift)). `Project.swift` references `Package.local(path: .relativeToRoot("."))` and declares the **Where** app, **StuffTestHost**, and unit-test targets that depend on package products. @@ -53,8 +53,8 @@ by `./sync-agents`. ## Targets -- **Package products** ([`Package.swift`](Package.swift)) — **StuffCore** ([`Shared/StuffCore/Sources/`](Shared/StuffCore/Sources/)), **LifecycleKit** ([`Shared/LifecycleKit/Sources/`](Shared/LifecycleKit/Sources/)), **LogKit** ([`Shared/LogKit/Sources/`](Shared/LogKit/Sources/), the logging facade), **LogViewerUI** ([`Shared/LogViewerUI/Sources/`](Shared/LogViewerUI/Sources/), the generic SwiftUI log viewer), **PeriscopeCore** / **PeriscopeUI** / **PeriscopeTools** ([`Shared/Periscope/`](Shared/Periscope/), the typed hierarchical observability framework — core + SwiftData store, SwiftUI environment integration, and on-device exploration tooling), **SwiftDataInspector** ([`Shared/SwiftDataInspector/Sources/`](Shared/SwiftDataInspector/Sources/), the generic SwiftData browser), and **TestHostSupport** ([`Shared/TestHostSupport/Sources/`](Shared/TestHostSupport/Sources/), the shared UIKit hosting + run-loop helpers for hosted test bundles) under [`Shared/`](Shared/); the **Broadway** design-system libraries **BroadwayCore** / **BroadwayUI** under [`Shared/Broadway/`](Shared/Broadway/) (see [`Shared/Broadway/AGENTS.md`](Shared/Broadway/AGENTS.md)); **RegionKit** ([`Where/RegionKit/Sources/`](Where/RegionKit/Sources/), the geometry + GeoJSON + region-lookup engine WhereCore builds on) / **WhereCore** / **WhereUI** under [`Where/`](Where/). -- **Tuist targets** ([`Project.swift`](Project.swift)) — **Where** app ([`Where/Where/`](Where/Where/)) and its embedded app extensions **WhereWidgets** ([`Where/WhereWidgets/`](Where/WhereWidgets/), home/lock-screen widgets) and **WhereShareExtension** ([`Where/WhereShareExtension/`](Where/WhereShareExtension/), a Share-sheet action that saves shared content as evidence into the App Group store), **RegionViewer** ([`Where/RegionViewer/`](Where/RegionViewer/), a thin standalone **Mac Catalyst** host for the WhereUI region-map developer tool — the only target with a `.macCatalyst` destination), **BroadwayCatalog** ([`Shared/Broadway/BroadwayCatalog/`](Shared/Broadway/BroadwayCatalog/), the Broadway component showcase app), **StuffTestHost** ([`Shared/StuffTestHost/`](Shared/StuffTestHost/)), **WhereTests** (app tests, no host), and hosted **\*Tests** bundles (**StuffCoreTests**, **LifecycleKitTests**, **LogKitTests**, **LogViewerUITests**, **PeriscopeCoreTests**, **PeriscopeUITests**, **PeriscopeToolsTests**, **SwiftDataInspectorTests**, **RegionKitTests**, **WhereCoreTests**, **WhereUITests**, **BroadwayCoreTests**, **BroadwayUITests**, **BroadwayCatalogTests**) that depend on **StuffTestHost** + **TestHostSupport** + the relevant package product (most via the shared `unitTests` helper). +- **Package products** ([`Package.swift`](Package.swift)) — **StuffCore** ([`Shared/StuffCore/Sources/`](Shared/StuffCore/Sources/)), **LifecycleKit** ([`Shared/LifecycleKit/Sources/`](Shared/LifecycleKit/Sources/)), **JournalKit** ([`Shared/JournalKit/Sources/`](Shared/JournalKit/Sources/), the generic append-only crash-durable journal), **LogKit** ([`Shared/LogKit/Sources/`](Shared/LogKit/Sources/), the logging facade), **LogViewerUI** ([`Shared/LogViewerUI/Sources/`](Shared/LogViewerUI/Sources/), the generic SwiftUI log viewer), **PeriscopeCore** / **PeriscopeUI** / **PeriscopeTools** ([`Shared/Periscope/`](Shared/Periscope/), the typed hierarchical observability framework — core + SwiftData store, SwiftUI environment integration, and on-device exploration tooling), **SwiftDataInspector** ([`Shared/SwiftDataInspector/Sources/`](Shared/SwiftDataInspector/Sources/), the generic SwiftData browser), and **TestHostSupport** ([`Shared/TestHostSupport/Sources/`](Shared/TestHostSupport/Sources/), the shared UIKit hosting + run-loop helpers for hosted test bundles) under [`Shared/`](Shared/); the **Broadway** design-system libraries **BroadwayCore** / **BroadwayUI** under [`Shared/Broadway/`](Shared/Broadway/) (see [`Shared/Broadway/AGENTS.md`](Shared/Broadway/AGENTS.md)); **RegionKit** ([`Where/RegionKit/Sources/`](Where/RegionKit/Sources/), the geometry + GeoJSON + region-lookup engine WhereCore builds on) / **WhereCore** / **WhereUI** under [`Where/`](Where/). +- **Tuist targets** ([`Project.swift`](Project.swift)) — **Where** app ([`Where/Where/`](Where/Where/)) and its embedded app extensions **WhereWidgets** ([`Where/WhereWidgets/`](Where/WhereWidgets/), home/lock-screen widgets) and **WhereShareExtension** ([`Where/WhereShareExtension/`](Where/WhereShareExtension/), a Share-sheet action that saves shared content as evidence into the App Group store), **RegionViewer** ([`Where/RegionViewer/`](Where/RegionViewer/), a thin standalone **Mac Catalyst** host for the WhereUI region-map developer tool — the only target with a `.macCatalyst` destination), **BroadwayCatalog** ([`Shared/Broadway/BroadwayCatalog/`](Shared/Broadway/BroadwayCatalog/), the Broadway component showcase app), **StuffTestHost** ([`Shared/StuffTestHost/`](Shared/StuffTestHost/)), **WhereTests** (app tests, no host), and hosted **\*Tests** bundles (**StuffCoreTests**, **LifecycleKitTests**, **JournalKitTests**, **LogKitTests**, **LogViewerUITests**, **PeriscopeCoreTests**, **PeriscopeUITests**, **PeriscopeToolsTests**, **SwiftDataInspectorTests**, **RegionKitTests**, **WhereCoreTests**, **WhereUITests**, **BroadwayCoreTests**, **BroadwayUITests**, **BroadwayCatalogTests**) that depend on **StuffTestHost** + **TestHostSupport** + the relevant package product (most via the shared `unitTests` helper). - Add SPM library targets in `Package.swift` and wire apps/tests in `Project.swift` (see existing `unitTests` helper). A new module also ships a root `README.md` and `AGENTS.md` — see [Per-module docs](#per-module-docs). - **CI scheme**: CI runs the explicit shared **Stuff-iOS-Tests** scheme (all test bundles) rather than the autogenerated `Stuff-Workspace` scheme. New test bundles must be added to the `Stuff-iOS-Tests` scheme in `Project.swift` or CI won't run them. - **Never double-link a package product into a test bundle that already gets it through a dynamic-framework dependency.** Xcode's default SPM integration (how `Package.local` is wired here) embeds a product's code into *every* image that links it. **WhereUI** is a dynamic framework that statically embeds its own dependencies (WhereCore, BroadwayCore/BroadwayUI, LifecycleKit, LogViewerUI, SwiftDataInspector, …), so a `*Tests` bundle that depends on **WhereUI** *and* re-lists one of those in `extraPackageProducts` ends up with a **second copy** of it. When the shared **StuffTestHost** loads several `.xctest` bundles into one process, that leaves duplicate **type metadata** for the module, and any *type-keyed runtime lookup that crosses the WhereUI boundary* — SwiftUI `EnvironmentKey`s, `UITraitBridgedEnvironmentKey` bridging, the type-keyed `BTraits`/`BThemes`/`BStylesheets` containers — silently resolves against the wrong copy and returns the default (the writer stores under one copy's key *type*, the reader looks it up under another's). It only reproduces in the **full multi-bundle scheme** (not isolated `tuist test WhereUITests` runs) and is papered over by newer Xcode linkers, so it is brutal to diagnose (it cost ~2 hours once). **Depend on such products only transitively via `WhereUI`; keep them out of `extraPackageProducts`.** `WhereStylesheetTests.resolvesTraitAwareTokensFromTheBroadwayRoot` is the regression guard — it exercises a `\.stylesheet` (→ `\.bContext`) read across the boundary and fails if a duplicate copy returns. diff --git a/Package.swift b/Package.swift index e8d62bf2..119946f9 100644 --- a/Package.swift +++ b/Package.swift @@ -10,6 +10,7 @@ let package = Package( products: [ .library(name: "StuffCore", targets: ["StuffCore"]), .library(name: "LifecycleKit", targets: ["LifecycleKit"]), + .library(name: "JournalKit", targets: ["JournalKit"]), .library(name: "LogKit", targets: ["LogKit"]), .library(name: "LogViewerUI", targets: ["LogViewerUI"]), .library(name: "PeriscopeCore", targets: ["PeriscopeCore"]), @@ -39,6 +40,10 @@ let package = Package( .process("Resources"), ], ), + .target( + name: "JournalKit", + path: "Shared/JournalKit/Sources", + ), .target( name: "LogKit", path: "Shared/LogKit/Sources", @@ -52,6 +57,9 @@ let package = Package( ), .target( name: "PeriscopeCore", + dependencies: [ + .target(name: "JournalKit"), + ], path: "Shared/Periscope/PeriscopeCore/Sources", ), .target( diff --git a/Project.swift b/Project.swift index d03d736d..1fa30eee 100644 --- a/Project.swift +++ b/Project.swift @@ -281,11 +281,18 @@ let project = Project( productDependency: "LogViewerUI", sources: ["Shared/LogViewerUI/Tests/**"], ), + unitTests( + name: "JournalKitTests", + bundleIdSuffix: "journalkit", + productDependency: "JournalKit", + sources: ["Shared/JournalKit/Tests/**"], + ), unitTests( name: "PeriscopeCoreTests", bundleIdSuffix: "periscopecore", productDependency: "PeriscopeCore", sources: ["Shared/Periscope/PeriscopeCore/Tests/**"], + extraPackageProducts: ["JournalKit"], ), unitTests( name: "PeriscopeUITests", @@ -427,6 +434,7 @@ let project = Project( "LifecycleKitTests", "LogKitTests", "LogViewerUITests", + "JournalKitTests", "PeriscopeCoreTests", "PeriscopeUITests", "PeriscopeToolsTests", @@ -446,6 +454,7 @@ let project = Project( "LifecycleKitTests", "LogKitTests", "LogViewerUITests", + "JournalKitTests", "PeriscopeCoreTests", "PeriscopeUITests", "PeriscopeToolsTests", @@ -464,6 +473,7 @@ let project = Project( testScheme(name: "LifecycleKitTests"), testScheme(name: "LogKitTests"), testScheme(name: "LogViewerUITests"), + testScheme(name: "JournalKitTests"), testScheme(name: "PeriscopeCoreTests"), testScheme(name: "PeriscopeUITests"), testScheme(name: "PeriscopeToolsTests"), diff --git a/Shared/JournalKit/AGENTS.md b/Shared/JournalKit/AGENTS.md new file mode 100644 index 00000000..9c2085ba --- /dev/null +++ b/Shared/JournalKit/AGENTS.md @@ -0,0 +1,40 @@ +# JournalKit – Module Shape + +JournalKit is the generic append-only, crash-durable journal: synchronous +`Data` appends that survive process death, segment rotation under a byte +budget, and torn-tail-tolerant recovery. See [`README.md`](README.md) for +the API and durability model. + +This file complements the root [`AGENTS.md`](../../AGENTS.md), which owns +the build system, formatting, and global conventions. Read that first. + +## Scope & dependencies + +- **Foundation + os only.** No logging types, no Periscope imports — the + journal is payload-agnostic by design (PeriscopeCore layers log semantics + on top). Keep it that way. + +## Invariants + +- **`append` returning means the entry survives process death.** The write + reaches the kernel page cache synchronously; `.full` extends coverage to + kernel panics via `F_FULLFSYNC`. Nothing may buffer entries in user space. +- **Recovery never throws over a torn tail.** A crash can cut the file at + any byte; recovery yields every wholly-written entry and flags the tear. + The truncation fuzz in `JournalRecoveryTests` pins this at every cut + point — keep it passing. +- **A torn write poisons only its own segment.** A partial `write(2)` + (disk-full's shape) leaves bytes recovery stops at, so the segment is + marked poisoned and the next append rotates to a fresh one — later + entries must never land behind a tear. +- **Drops are whole segments, oldest first,** and always observable + (`droppedSegmentCount`, `droppedOlderEntries`) — the newest entries are + never sacrificed. A segment that fails to delete stays in the byte + accounting (later rotations retry it) and the drop loop moves to the + next-oldest, so the budget still wins. + +## Testing + +Swift Testing in [`Tests/`](Tests), hosted in `StuffTestHost` +(`JournalKitTests`). Tests journal into per-test temporary directories and +construct crashed-journal states (truncation, corruption) directly on disk. diff --git a/Shared/JournalKit/README.md b/Shared/JournalKit/README.md new file mode 100644 index 00000000..b9e40b49 --- /dev/null +++ b/Shared/JournalKit/README.md @@ -0,0 +1,64 @@ +# JournalKit + +An append-only, crash-durable journal of opaque `Data` entries — the +write-ahead net for anything that must survive the process dying +mid-flight. Periscope uses it as its log journal; the implementation is +payload-agnostic and has no logging knowledge. + +## Quick start + +```swift +import JournalKit + +// Writing (synchronous, any thread; ~microseconds per append): +let journal = try Journal( + directory: journalDirectory, + configuration: Journal.Configuration(maximumByteCount: 8 * 1024 * 1024), +) +try journal.append(encodedEntry, sync: .processDeath) +try journal.append(direEntry, sync: .full) // F_FULLFSYNC — milliseconds + +// Recovery (the next launch): +let recovered = try JournalRecovery.recover(directory: journalDirectory) +for payload in recovered.payloads { ingest(payload) } +try JournalRecovery.remove(directory: journalDirectory) +``` + +## Durability model + +- **`.processDeath`** (the default posture): once `append` returns, the + entry is in the kernel page cache and survives the *process* dying by any + means — crash, `SIGKILL`, jetsam. Cost: single-digit microseconds. + Verified empirically by the SIGKILL harness in + [`Shared/Periscope/Prototypes/JournalBenchmark`](../Periscope/Prototypes/JournalBenchmark). +- **`.full`**: `F_FULLFSYNC` before returning — survives kernel panics and + power loss. Milliseconds; reserve for entries that warrant it. + +## How it works + +Entries are framed `[UInt32 length][UInt32 crc32][payload]` and appended to +numbered segment files. Segments rotate at half the byte budget, and the +oldest segment drops whole when the budget overflows — the journal favors +the newest entries, like a flight recorder, and reports +`droppedSegmentCount` / `droppedOlderEntries` so callers can surface the +gap. An optional `segmentHeader` re-writes as the first entry of every +segment, so identity/context entries survive any amount of rotation +(recovery returns one copy per surviving segment). A *partial* write +poisons its segment and the next append rotates, so torn bytes never +strand later entries. + +Recovery reads segments in order and validates each entry's length and CRC. +A torn or corrupt entry ends that segment's recovery and sets +`foundTornEntry` — the partial final entry a crash leaves is expected, and +recovery never throws over it. + +## Contracts & limitations + +- Entry ordering across concurrent appenders is append order (writers + serialize on an internal lock); per-writer order is preserved. Any + cross-entry semantics — sequencing, deduplication, schemas — belong to + the caller's payloads. +- One `Journal` instance per directory at a time; reopening continues after + the existing segments rather than overwriting them. +- CRC-32 catches torn and corrupt entries; it is integrity checking, not + authentication. diff --git a/Shared/JournalKit/Sources/Journal.swift b/Shared/JournalKit/Sources/Journal.swift new file mode 100644 index 00000000..2b4bf6db --- /dev/null +++ b/Shared/JournalKit/Sources/Journal.swift @@ -0,0 +1,241 @@ +import Foundation +import os + +/// An append-only, crash-durable journal of opaque `Data` entries. +/// +/// `append(_:sync:)` is synchronous and safe from any thread: once it +/// returns, the entry has reached the kernel page cache and survives the +/// *process* dying by any means (crash, kill, jetsam). `.full` sync extends +/// that to kernel panics and power loss via `F_FULLFSYNC`, at +/// millisecond cost — reserve it for entries that warrant it. +/// +/// Entries are framed as `[length][crc32][payload]` and written to numbered +/// segment files in the journal's directory. When the journal exceeds its +/// byte budget, the oldest segment is dropped whole (the journal favors the +/// newest entries, like a flight recorder). ``JournalRecovery`` reads a +/// directory back tolerating the torn final entry a crash can leave. +/// +/// The journal is payload-agnostic: what the bytes mean, and any ordering +/// or deduplication semantics, belong to the caller. +public final class Journal: @unchecked Sendable { + public struct Configuration: Sendable { + /// Total byte budget across segments; on overflow the oldest + /// segment is dropped whole. + public var maximumByteCount: Int + + /// Re-written as the first entry of every segment, so rotation can + /// never drop it: identity/context entries (say, a session header) + /// stay recoverable however far the journal rotates. Recovery + /// returns it like any other entry — expect one copy per surviving + /// segment. + public var segmentHeader: Data? + + public init(maximumByteCount: Int, segmentHeader: Data? = nil) { + self.maximumByteCount = maximumByteCount + self.segmentHeader = segmentHeader + } + } + + /// How durable an individual append must be before it returns. + public enum Sync: Sendable { + /// Page cache — survives process death; lost in a kernel panic. + case processDeath + /// `F_FULLFSYNC` — survives kernel panics and power loss. + /// Milliseconds; reserve for the direst entries. + case full + } + + private struct State { + var segmentIndex: Int + var descriptor: Int32 + var segmentByteCount: Int + var totalByteCount: Int + var droppedSegmentCount = 0 + var isClosed = false + /// A short write left torn bytes at the segment's tail; recovery + /// would stop there, so nothing more may append to it — the next + /// append rotates to a fresh segment first. + var segmentIsPoisoned = false + #if DEBUG + var pendingShortWrite = false + var pendingRemovalFailure = false + #endif + } + + /// Segments rotate at half the budget so at most two exist: dropping + /// the older one can never discard the newest entries. + private var segmentBudget: Int { + configuration.maximumByteCount / 2 + } + + public let directory: URL + private let configuration: Configuration + private let state: OSAllocatedUnfairLock + + /// Opens a journal in `directory` (created if needed), continuing after + /// the highest existing segment so prior entries are never overwritten. + public init(directory: URL, configuration: Configuration) throws { + self.directory = directory + self.configuration = configuration + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + let existing = try JournalSegments.indexes(in: directory) + let index = (existing.last ?? 0) + 1 + let descriptor = try JournalSegments.open(index: index, in: directory) + let existingBytes = try existing.reduce(0) { total, index in + try total + (JournalSegments.byteCount(of: index, in: directory)) + } + state = OSAllocatedUnfairLock(initialState: State( + segmentIndex: index, + descriptor: descriptor, + segmentByteCount: 0, + totalByteCount: existingBytes, + )) + try state.withLock { state in + try writeSegmentHeader(&state) + } + } + + deinit { + close() + } + + /// Append one entry. Synchronous; safe from any thread. Throws when the + /// journal is closed or the write fails (disk full). A *partial* write + /// (some bytes landed, then failure) poisons the segment: recovery + /// stops at torn bytes, so the next append rotates to a fresh segment + /// rather than stranding everything written after the tear. + public func append(_ payload: Data, sync: Sync) throws { + let framed = JournalFraming.frame(payload) + try state.withLock { state in + guard !state.isClosed else { throw JournalError.closed } + if state.segmentIsPoisoned + || + (state.segmentByteCount + framed.count > segmentBudget && state + .segmentByteCount > 0) + { + try rotate(&state) + state.segmentIsPoisoned = false + } + + var bytesToWrite = framed[...] + #if DEBUG + if state.pendingShortWrite { + state.pendingShortWrite = false + bytesToWrite = framed.prefix(framed.count / 2) + } + #endif + let written = bytesToWrite.withUnsafeBytes { buffer in + write(state.descriptor, buffer.baseAddress, buffer.count) + } + + if written == framed.count { + state.segmentByteCount += framed.count + state.totalByteCount += framed.count + } else if written > 0 { + // Torn bytes are on disk (disk-full's shape): account for + // them and poison the segment so later entries land + // parseably in the next one. + state.segmentByteCount += written + state.totalByteCount += written + state.segmentIsPoisoned = true + throw JournalError.writeFailed(errno: errno) + } else { + // Nothing landed; the segment is still clean. + throw JournalError.writeFailed(errno: errno) + } + if case .full = sync { + _ = fcntl(state.descriptor, F_FULLFSYNC) + } + } + } + + #if DEBUG + /// Testing seam: the next append writes only half its frame and + /// reports failure — the shape of a disk-full torn write. + @_spi(Testing) public func injectShortWriteOnNextAppend() { + state.withLock { $0.pendingShortWrite = true } + } + + /// Testing seam: the next segment removal during rotation fails — + /// the accounting must not pretend it succeeded. + @_spi(Testing) public func injectRemovalFailureOnNextRotation() { + state.withLock { $0.pendingRemovalFailure = true } + } + #endif + + /// Segments dropped so far to stay within the byte budget — each one + /// took its entries with it. + public var droppedSegmentCount: Int { + state.withLock(\.droppedSegmentCount) + } + + /// Close the journal; further appends throw. Idempotent. + public func close() { + state.withLock { state in + guard !state.isClosed else { return } + Darwin.close(state.descriptor) + state.isClosed = true + } + } + + private func rotate(_ state: inout State) throws { + Darwin.close(state.descriptor) + let next = state.segmentIndex + 1 + state.descriptor = try JournalSegments.open(index: next, in: directory) + state.segmentIndex = next + state.segmentByteCount = 0 + try writeSegmentHeader(&state) + // Drop oldest segments until the (pre-append) total fits the budget. + var indexes = try JournalSegments.indexes(in: directory) + while state.totalByteCount > configuration.maximumByteCount, indexes.count > 1 { + let oldest = indexes.removeFirst() + let bytes = try JournalSegments.byteCount(of: oldest, in: directory) + // Only account for what actually leaves the disk: a segment + // that fails to delete is still there (a later rotation + // retries it), and the loop moves to the next-oldest so the + // byte budget still wins. + guard removeSegment(oldest, state: &state) else { continue } + state.totalByteCount -= bytes + state.droppedSegmentCount += 1 + } + } + + private func removeSegment(_ index: Int, state: inout State) -> Bool { + #if DEBUG + if state.pendingRemovalFailure { + state.pendingRemovalFailure = false + return false + } + #endif + do { + try FileManager.default.removeItem(at: JournalSegments.url(for: index, in: directory)) + return true + } catch { + return false + } + } + + /// Write the configured header as the segment's first entry — every + /// segment must be self-describing, since rotation drops whole + /// segments from the oldest end. + private func writeSegmentHeader(_ state: inout State) throws { + guard let header = configuration.segmentHeader else { return } + let framed = JournalFraming.frame(header) + let written = framed.withUnsafeBytes { buffer in + write(state.descriptor, buffer.baseAddress, buffer.count) + } + guard written == framed.count else { + if written > 0 { + state.segmentIsPoisoned = true + } + throw JournalError.writeFailed(errno: errno) + } + state.segmentByteCount += framed.count + state.totalByteCount += framed.count + } +} + +public enum JournalError: Error, Equatable { + case closed + case writeFailed(errno: Int32) +} diff --git a/Shared/JournalKit/Sources/JournalRecovery.swift b/Shared/JournalKit/Sources/JournalRecovery.swift new file mode 100644 index 00000000..4c238301 --- /dev/null +++ b/Shared/JournalKit/Sources/JournalRecovery.swift @@ -0,0 +1,148 @@ +import Foundation + +/// Reads a journal directory back — the "next launch" half of ``Journal``. +/// +/// Segments are read in write order; within each, entries are validated by +/// length and CRC. An invalid or incomplete entry ends that segment's +/// recovery (the torn tail a crash leaves is expected, and it can only be +/// the final entry of the final write). +public enum JournalRecovery { + public struct Recovered: Sendable { + /// Every intact entry, in append order. + public let payloads: [Data] + /// Whether a segment ended in a torn or corrupt entry. + public let foundTornEntry: Bool + /// Whether the writer dropped older segments to stay in budget — + /// entries older than `payloads.first` existed and are gone. + public let droppedOlderEntries: Bool + } + + /// Recover every intact entry from the journal in `directory`. A + /// missing directory recovers as empty. + public static func recover(directory: URL) throws -> Recovered { + let indexes = try JournalSegments.indexes(in: directory) + var payloads: [Data] = [] + var foundTornEntry = false + for index in indexes { + let data = try Data(contentsOf: JournalSegments.url(for: index, in: directory)) + let segment = JournalFraming.parse(data) + payloads.append(contentsOf: segment.payloads) + foundTornEntry = foundTornEntry || segment.foundTornEntry + } + return Recovered( + payloads: payloads, + foundTornEntry: foundTornEntry, + // Segment numbering starts at 1; a higher floor means the + // writer rotated older segments away. + droppedOlderEntries: (indexes.first ?? 1) > 1, + ) + } + + /// Delete the journal directory — call after ingesting a recovery. + public static func remove(directory: URL) throws { + guard FileManager.default.fileExists(atPath: directory.path) else { return } + try FileManager.default.removeItem(at: directory) + } +} + +/// Segment file naming and bookkeeping, shared by writer and recovery. +enum JournalSegments { + private static let fileExtension = "journalsegment" + + static func url(for index: Int, in directory: URL) -> URL { + directory.appendingPathComponent("\(index).\(fileExtension)") + } + + /// Present segment indexes, ascending. A missing directory is empty. + static func indexes(in directory: URL) throws -> [Int] { + guard FileManager.default.fileExists(atPath: directory.path) else { return [] } + return try FileManager.default.contentsOfDirectory(atPath: directory.path) + .compactMap { name -> Int? in + let parts = name.split(separator: ".") + guard parts.count == 2, parts[1] == fileExtension else { return nil } + return Int(parts[0]) + } + .sorted() + } + + static func open(index: Int, in directory: URL) throws -> Int32 { + let descriptor = Darwin.open( + url(for: index, in: directory).path, + O_WRONLY | O_CREAT | O_APPEND, + 0o644, + ) + guard descriptor >= 0 else { throw JournalError.writeFailed(errno: errno) } + return descriptor + } + + static func byteCount(of index: Int, in directory: URL) throws -> Int { + let attributes = try FileManager.default.attributesOfItem( + atPath: url(for: index, in: directory).path, + ) + return (attributes[.size] as? Int) ?? 0 + } +} + +/// The on-disk entry framing: `[UInt32 length][UInt32 crc32][payload]`, +/// little-endian. CRC validation rejects torn and corrupt entries. +enum JournalFraming { + /// Refuse absurd lengths (torn length bytes can decode as garbage). + static let maximumEntryByteCount = 64 * 1024 * 1024 + + static func frame(_ payload: Data) -> Data { + var framed = Data(capacity: payload.count + 8) + withUnsafeBytes(of: UInt32(payload.count).littleEndian) { framed.append(contentsOf: $0) } + withUnsafeBytes(of: CRC32.checksum(payload).littleEndian) { framed.append(contentsOf: $0) } + framed.append(payload) + return framed + } + + struct ParsedSegment { + let payloads: [Data] + let foundTornEntry: Bool + } + + static func parse(_ data: Data) -> ParsedSegment { + var payloads: [Data] = [] + var offset = 0 + while offset + 8 <= data.count { + let length = Int(data.subdata(in: offset ..< offset + 4) + .withUnsafeBytes { $0.loadUnaligned(as: UInt32.self).littleEndian }) + let crc = data.subdata(in: offset + 4 ..< offset + 8) + .withUnsafeBytes { $0.loadUnaligned(as: UInt32.self).littleEndian } + let end = offset + 8 + length + guard length <= maximumEntryByteCount, end <= data.count else { + return ParsedSegment(payloads: payloads, foundTornEntry: true) + } + let payload = data.subdata(in: offset + 8 ..< end) + guard CRC32.checksum(payload) == crc else { + return ParsedSegment(payloads: payloads, foundTornEntry: true) + } + payloads.append(payload) + offset = end + } + return ParsedSegment(payloads: payloads, foundTornEntry: offset != data.count) + } +} + +/// Table-driven CRC-32 (IEEE 802.3 polynomial) — entry integrity checking +/// without linking zlib. +enum CRC32 { + private static let table: [UInt32] = (0 ..< 256).map { index -> UInt32 in + var crc = UInt32(index) + for _ in 0 ..< 8 { + crc = (crc & 1) == 1 ? (crc >> 1) ^ 0xEDB8_8320 : crc >> 1 + } + return crc + } + + static func checksum(_ data: Data) -> UInt32 { + var crc: UInt32 = 0xFFFF_FFFF + data.withUnsafeBytes { buffer in + for byte in buffer { + crc = table[Int((crc ^ UInt32(byte)) & 0xFF)] ^ (crc >> 8) + } + } + return crc ^ 0xFFFF_FFFF + } +} diff --git a/Shared/JournalKit/Tests/JournalKitTestSupport.swift b/Shared/JournalKit/Tests/JournalKitTestSupport.swift new file mode 100644 index 00000000..32e207ca --- /dev/null +++ b/Shared/JournalKit/Tests/JournalKitTestSupport.swift @@ -0,0 +1,17 @@ +import Foundation +import JournalKit + +/// A fresh journal directory in the temporary directory; callers clean up +/// via `defer { try? FileManager.default.removeItem(at: url) }`. +func makeJournalDirectory() -> URL { + URL(fileURLWithPath: NSTemporaryDirectory()) + .appendingPathComponent("journalkit-tests-\(UUID().uuidString)", isDirectory: true) +} + +func payload(_ text: String) -> Data { + Data(text.utf8) +} + +func texts(_ payloads: [Data]) -> [String] { + payloads.map { String(decoding: $0, as: UTF8.self) } +} diff --git a/Shared/JournalKit/Tests/JournalRecoveryTests.swift b/Shared/JournalKit/Tests/JournalRecoveryTests.swift new file mode 100644 index 00000000..2027e837 --- /dev/null +++ b/Shared/JournalKit/Tests/JournalRecoveryTests.swift @@ -0,0 +1,99 @@ +import Foundation +@testable import JournalKit +import Testing + +struct JournalRecoveryTests { + @Test func missingDirectoriesRecoverEmpty() throws { + let recovered = try JournalRecovery.recover(directory: makeJournalDirectory()) + #expect(recovered.payloads.isEmpty) + #expect(!recovered.foundTornEntry) + #expect(!recovered.droppedOlderEntries) + } + + @Test func truncationAtEveryByteRecoversEveryWholeEntry() throws { + // The fuzz that pins torn-tail behavior: a crash can cut the file + // at any byte; recovery must yield exactly the entries wholly + // written before the cut, flag the tear, and never throw. + let directory = makeJournalDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let journal = try Journal( + directory: directory, + configuration: Journal.Configuration(maximumByteCount: 1_000_000), + ) + let entries = ["alpha", "beta-longer", "gamma-longest-of-all"] + for entry in entries { + try journal.append(payload(entry), sync: .processDeath) + } + journal.close() + + let segment = JournalSegments.url(for: 1, in: directory) + let full = try Data(contentsOf: segment) + let wholeLengths = entries.map { 8 + $0.utf8.count } + for cut in 0 ... full.count { + try full.prefix(cut).write(to: segment) + let recovered = try JournalRecovery.recover(directory: directory) + + var survivors = 0 + var consumed = 0 + for length in wholeLengths { + guard consumed + length <= cut else { break } + consumed += length + survivors += 1 + } + #expect(texts(recovered.payloads) == Array(entries.prefix(survivors))) + #expect(recovered.foundTornEntry == (cut != consumed)) + } + } + + @Test func corruptEntriesEndRecoveryAtTheLastGoodOne() throws { + let directory = makeJournalDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let journal = try Journal( + directory: directory, + configuration: Journal.Configuration(maximumByteCount: 1_000_000), + ) + try journal.append(payload("good"), sync: .processDeath) + try journal.append(payload("corrupted"), sync: .processDeath) + journal.close() + + // Flip one payload byte of the second entry; its CRC must reject it. + let segment = JournalSegments.url(for: 1, in: directory) + var bytes = try Data(contentsOf: segment) + bytes[bytes.count - 1] ^= 0xFF + try bytes.write(to: segment) + + let recovered = try JournalRecovery.recover(directory: directory) + #expect(texts(recovered.payloads) == ["good"]) + #expect(recovered.foundTornEntry) + } + + @Test func absurdLengthPrefixesAreRejectedNotAllocated() throws { + let directory = makeJournalDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + // A "length" of UInt32.max from torn bytes must not be trusted. + var bogus = Data() + withUnsafeBytes(of: UInt32.max.littleEndian) { bogus.append(contentsOf: $0) } + withUnsafeBytes(of: UInt32.zero.littleEndian) { bogus.append(contentsOf: $0) } + bogus.append(Data(repeating: 0xAB, count: 32)) + try bogus.write(to: JournalSegments.url(for: 1, in: directory)) + + let recovered = try JournalRecovery.recover(directory: directory) + #expect(recovered.payloads.isEmpty) + #expect(recovered.foundTornEntry) + } + + @Test func removeDeletesTheDirectoryAndToleratesAbsence() throws { + let directory = makeJournalDirectory() + let journal = try Journal( + directory: directory, + configuration: Journal.Configuration(maximumByteCount: 1_000_000), + ) + try journal.append(payload("entry"), sync: .processDeath) + journal.close() + + try JournalRecovery.remove(directory: directory) + #expect(!FileManager.default.fileExists(atPath: directory.path)) + try JournalRecovery.remove(directory: directory) // no throw on absence + } +} diff --git a/Shared/JournalKit/Tests/JournalTests.swift b/Shared/JournalKit/Tests/JournalTests.swift new file mode 100644 index 00000000..ba96e68a --- /dev/null +++ b/Shared/JournalKit/Tests/JournalTests.swift @@ -0,0 +1,203 @@ +import Foundation +@_spi(Testing) import JournalKit +import Testing + +struct JournalTests { + @Test func appendsRoundTripInOrder() throws { + let directory = makeJournalDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let journal = try Journal( + directory: directory, + configuration: Journal.Configuration(maximumByteCount: 1_000_000), + ) + try journal.append(payload("first"), sync: .processDeath) + try journal.append(payload("second"), sync: .processDeath) + try journal.append(payload("third"), sync: .full) + journal.close() + + let recovered = try JournalRecovery.recover(directory: directory) + #expect(texts(recovered.payloads) == ["first", "second", "third"]) + #expect(!recovered.foundTornEntry) + #expect(!recovered.droppedOlderEntries) + } + + @Test func closedJournalsRefuseAppends() throws { + let directory = makeJournalDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let journal = try Journal( + directory: directory, + configuration: Journal.Configuration(maximumByteCount: 1_000_000), + ) + journal.close() + #expect(throws: JournalError.closed) { + try journal.append(payload("late"), sync: .processDeath) + } + } + + @Test func overBudgetJournalsDropOldestSegmentsWhole() throws { + let directory = makeJournalDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + // ~100-byte entries against a 1KB budget: segments rotate at 512 + // bytes and the oldest drop as the budget overflows. + let journal = try Journal( + directory: directory, + configuration: Journal.Configuration(maximumByteCount: 1024), + ) + for index in 0 ..< 40 { + try journal.append( + payload("entry-\(index)-" + String(repeating: "x", count: 90)), + sync: .processDeath, + ) + } + journal.close() + + let recovered = try JournalRecovery.recover(directory: directory) + #expect(journal.droppedSegmentCount > 0) + #expect(recovered.droppedOlderEntries) + #expect(!recovered.payloads.isEmpty) + // The newest entry always survives — the journal drops from the + // oldest end, like a flight recorder. + #expect(texts(recovered.payloads).last?.hasPrefix("entry-39-") == true) + } + + @Test func tornWritesPoisonOnlyTheirSegment() throws { + // A short write (disk-full's shape) leaves torn bytes recovery + // stops at. The segment poisons; the next append rotates to a + // fresh one — so a moment of disk pressure loses one entry, not + // everything after it. + let directory = makeJournalDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let journal = try Journal( + directory: directory, + configuration: Journal.Configuration(maximumByteCount: 1_000_000), + ) + try journal.append(payload("before"), sync: .processDeath) + journal.injectShortWriteOnNextAppend() + #expect(throws: (any Error).self) { + try journal.append(payload("torn"), sync: .processDeath) + } + try journal.append(payload("after"), sync: .processDeath) + journal.close() + + let recovered = try JournalRecovery.recover(directory: directory) + #expect(texts(recovered.payloads) == ["before", "after"]) + #expect(recovered.foundTornEntry) + // The poisoned segment stayed behind; "after" landed in a new one. + let segments = try FileManager.default.contentsOfDirectory(atPath: directory.path) + #expect(segments.count == 2) + } + + @Test func failedSegmentRemovalsKeepTheBudgetHonest() throws { + // A removal that fails must not be counted as if it succeeded — + // the old accounting double-subtracted the segment's bytes on the + // retry, drifting the total low until the budget stopped enforcing + // and the journal grew without bound. + let directory = makeJournalDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let journal = try Journal( + directory: directory, + configuration: Journal.Configuration(maximumByteCount: 1024), + ) + for index in 0 ..< 60 { + if index.isMultiple(of: 5) { + journal.injectRemovalFailureOnNextRotation() + } + try journal.append( + payload("entry-\(index)-" + String(repeating: "x", count: 90)), + sync: .processDeath, + ) + } + journal.close() + + // Transient removal failures retried; the disk footprint stays + // near the budget instead of growing with the append count. + let onDisk = try FileManager.default.contentsOfDirectory(atPath: directory.path) + .reduce(0) { total, name in + let path = directory.appendingPathComponent(name).path + let size = ( + try? FileManager.default.attributesOfItem(atPath: path)[.size] as? Int, + ) ?? + 0 + return total + (size ?? 0) + } + #expect(onDisk <= 2048) + + let recovered = try JournalRecovery.recover(directory: directory) + #expect(texts(recovered.payloads).last?.hasPrefix("entry-59-") == true) + #expect(recovered.droppedOlderEntries) + } + + @Test func segmentHeadersSurviveRotation() throws { + // The header re-writes at every segment's start, so however far + // the budget rotates, the newest segment self-describes — dropping + // old segments can never orphan the journal's identity. + let directory = makeJournalDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let journal = try Journal( + directory: directory, + configuration: Journal.Configuration( + maximumByteCount: 1024, + segmentHeader: payload("header"), + ), + ) + for index in 0 ..< 40 { + try journal.append( + payload("entry-\(index)-" + String(repeating: "x", count: 90)), + sync: .processDeath, + ) + } + journal.close() + + let recovered = try JournalRecovery.recover(directory: directory) + #expect(recovered.droppedOlderEntries) + #expect(texts(recovered.payloads).first == "header") + #expect(texts(recovered.payloads).last?.hasPrefix("entry-39-") == true) + } + + @Test func reopeningContinuesAfterExistingSegments() throws { + let directory = makeJournalDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let configuration = Journal.Configuration(maximumByteCount: 1_000_000) + let first = try Journal(directory: directory, configuration: configuration) + try first.append(payload("before"), sync: .processDeath) + first.close() + + let second = try Journal(directory: directory, configuration: configuration) + try second.append(payload("after"), sync: .processDeath) + second.close() + + let recovered = try JournalRecovery.recover(directory: directory) + #expect(texts(recovered.payloads) == ["before", "after"]) + } + + @Test func concurrentAppendsAllSurvive() async throws { + let directory = makeJournalDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let journal = try Journal( + directory: directory, + configuration: Journal.Configuration(maximumByteCount: 10_000_000), + ) + let threads = 4 + let perThread = 250 + await withTaskGroup(of: Void.self) { group in + for thread in 0 ..< threads { + group.addTask { + for index in 0 ..< perThread { + try? journal.append(payload("t\(thread)-\(index)"), sync: .processDeath) + } + } + } + } + journal.close() + + let recovered = try JournalRecovery.recover(directory: directory) + #expect(recovered.payloads.count == threads * perThread) + #expect(Set(texts(recovered.payloads)).count == threads * perThread) + // Per-writer order is preserved even though writers interleave. + for thread in 0 ..< threads { + let mine = texts(recovered.payloads).filter { $0.hasPrefix("t\(thread)-") } + let expected = (0 ..< perThread).map { "t\(thread)-\($0)" } + #expect(mine == expected) + } + } +} diff --git a/Shared/Periscope/PeriscopeCore/AGENTS.md b/Shared/Periscope/PeriscopeCore/AGENTS.md index e824fdbd..54fee80c 100644 --- a/Shared/Periscope/PeriscopeCore/AGENTS.md +++ b/Shared/Periscope/PeriscopeCore/AGENTS.md @@ -14,15 +14,17 @@ the build system, formatting, and global conventions. Read that first. tags, attachments), `Loggers/` (`Log` and the scope tree), `Context/` (task-local and per-instance context derivation), `Spans/` (timing and exits), `Pipeline/` (the `Periscope` system, records, sinks), -`Store/` (SwiftData persistence and queries), and `Ambient/` (environmental -event sources). Tests stay flat, named 1:1 with their source files. +`Store/` (SwiftData persistence and queries), `Journal/` (the crash +journal's log-layer envelope and writer, over +[`JournalKit`](../../JournalKit)), and `Ambient/` (environmental event +sources). Tests stay flat, named 1:1 with their source files. ## Scope & dependencies -- **Foundation + os + SwiftData + Network only** (plus the ObjectiveC - runtime, solely for `LogContextProviding`'s deallocation trackers). No - SwiftUI, no app code, no LogKit. UIKit is allowed **only** inside - `#if canImport(UIKit)` (ambient sources, the image-attachment +- **Foundation + os + SwiftData + Network + JournalKit only** (plus the + ObjectiveC runtime, solely for `LogContextProviding`'s deallocation + trackers). No SwiftUI, no app code, no LogKit. UIKit is allowed **only** + inside `#if canImport(UIKit)` (ambient sources, the image-attachment convenience). - Layering: `PeriscopeUI` and `PeriscopeTools` depend on this module — never the reverse. @@ -50,6 +52,18 @@ event sources). Tests stay flat, named 1:1 with their source files. the context is discarded, row caches drop, and the session row refetches by identity — one poisoned batch must never wedge subsequent saves or fork the session. +- **The crash journal is synchronous at emit and silent on failure.** + Every buffered record appends to the journal (when an on-disk store is + attached) *before* `record()` returns — sequence stamped under the state + lock, file I/O outside it, fault+ records `F_FULLFSYNC`. Journal failures + count and log to OSLog but never throw into the emit path. Ingest runs + *before* `startSession` so recovered begans join the orphan sweep; a + journal that fails ingest stays for the next launch. +- **Only app processes ingest journals.** App extensions journal their own + sessions but skip ingest (ingest deletes journals; an extension launch + must not eat the live app's) — the app's next launch recovers everyone's. + Sharing one on-disk store across *concurrently live* processes is not + yet supported; see [`TODOs.md`](../TODOs.md). - **Payloads persist as versioned JSON** (`eventName` + `eventVersion`), not per-event schemas — changing an event's shape must not require a SwiftData migration. diff --git a/Shared/Periscope/PeriscopeCore/README.md b/Shared/Periscope/PeriscopeCore/README.md index 26670158..5e4dee79 100644 --- a/Shared/Periscope/PeriscopeCore/README.md +++ b/Shared/Periscope/PeriscopeCore/README.md @@ -118,6 +118,19 @@ the gap (scope definitions and span began/ended pairs are exempt). Event payload so old rows outlive their Swift types — `StoredLogEvent.decode(_:)` recovers the type, and tooling degrades to raw JSON when it can't. +**Crash durability**: on-disk stores also open a per-session journal +([JournalKit](../../JournalKit)) beside the database, and once the store is +added as a sink, every record appends to it *synchronously* at emit +(microseconds — a page-cache write that survives the process dying by any +means; fault-level records `F_FULLFSYNC` for kernel-panic coverage). At the +next launch the store ingests prior journals before the session starts: +undelivered records persist (deduplicated by event ID), recovered span +begans join the orphan sweep, a `.notice` marks the recovery, and the +journal is deleted. The loss window at a hard crash is the microseconds +between a record buffering and its append returning. In-memory stores +never journal, and only app processes ingest — app extensions journal +their own sessions and leave recovery to the app's next launch. + ## Contracts & limitations - Messages mirror to OSLog as `.public` — keep PII out of messages, or scrub diff --git a/Shared/Periscope/PeriscopeCore/Sources/Journal/LogJournal.swift b/Shared/Periscope/PeriscopeCore/Sources/Journal/LogJournal.swift new file mode 100644 index 00000000..7ee02297 --- /dev/null +++ b/Shared/Periscope/PeriscopeCore/Sources/Journal/LogJournal.swift @@ -0,0 +1,100 @@ +import Foundation +import JournalKit +import os + +/// The live crash journal: encodes log-layer envelopes and appends them +/// through JournalKit synchronously on the emitting thread, so every +/// record survives the process dying before the async pipeline drains. +/// +/// Owned by `PeriscopeStore` (which knows the session and where journals +/// live) and installed into `Periscope` when the store is added as a sink. +/// Append failures never propagate into the pipeline — they count and log +/// to OSLog, and the async path keeps delivering. +@_spi(Testing) public final class LogJournal: Sendable { + /// Total on-disk budget per session journal; beyond it the oldest + /// segment drops whole (flight-recorder posture). Generous on purpose: + /// journals live only until the next launch ingests and deletes them, + /// so the budget is a ceiling on pathological sessions, not a steady + /// state. + static let byteBudget = 300 * 1024 * 1024 + + /// Records at this level or above `F_FULLFSYNC` before returning — + /// kernel-panic durability for the direst records, milliseconds each, + /// rare by definition. + static let fullSyncFloor = LogLevel.fault + + private static let failureLogger = os.Logger( + subsystem: "com.stuff.periscope", + category: "LogJournal", + ) + + @_spi(Testing) public let directory: URL + private let journal: Journal + private let failureCount = OSAllocatedUnfairLock(initialState: 0) + + /// Opens the journal with the session entry as every segment's header, + /// so a recovered journal attributes itself however far the budget's + /// rotation went — dropping old segments must never orphan the rest. + @_spi(Testing) public init(directory: URL, session: LogSession) throws { + self.directory = directory + journal = try Journal( + directory: directory, + configuration: Journal.Configuration( + maximumByteCount: Self.byteBudget, + segmentHeader: LogJournalEntry.session(session).encoded(), + ), + ) + } + + /// Journal one emitted record. `sequence` (stamped under the pipeline's + /// state lock) orders replay, so appends may race here without + /// reordering recovery. + @_spi(Testing) public func append(_ record: LogRecord, sequence: Int) { + do { + let entry = try LogJournalEntry.record(LogJournalRecord( + record: record, + sequence: sequence, + )) + try journal.append( + entry.encoded(), + sync: record.level >= Self.fullSyncFloor ? .full : .processDeath, + ) + } catch { + noteFailure(error) + } + } + + /// Journal a scope definition — recovered records need the hierarchy. + func append(scope: LogScope) { + do { + try journal.append(LogJournalEntry.scope(scope).encoded(), sync: .processDeath) + } catch { + noteFailure(error) + } + } + + /// Close the underlying journal (tests; production journals live for + /// the process, and `Journal`'s own `deinit` closes the descriptor + /// when the last reference drops — no explicit teardown needed here). + @_spi(Testing) public func close() { + journal.close() + } + + /// Appends that failed (encoding or I/O) — observable telemetry, since + /// journal failures must not throw into the emit path. + @_spi(Testing) public var appendFailureCount: Int { + failureCount.withLock { $0 } + } + + private func noteFailure(_ error: any Error) { + let count = failureCount.withLock { count -> Int in + count += 1 + return count + } + // Log the first failure and then every 100th — an unwritable disk + // would otherwise spam OSLog once per record. + if count == 1 || count.isMultiple(of: 100) { + Self.failureLogger.warning("Journal append failed (\(count)): \(error)") + } + } +} diff --git a/Shared/Periscope/PeriscopeCore/Sources/Journal/LogJournalEntry.swift b/Shared/Periscope/PeriscopeCore/Sources/Journal/LogJournalEntry.swift new file mode 100644 index 00000000..2d73b42f --- /dev/null +++ b/Shared/Periscope/PeriscopeCore/Sources/Journal/LogJournalEntry.swift @@ -0,0 +1,187 @@ +import Foundation + +/// One entry in the crash-durability journal: the log-layer envelope that +/// PeriscopeCore writes through JournalKit's payload-agnostic `Journal`. +/// +/// Entries encode as versioned JSON (`{"v":1,"kind":"record","payload":…}`) +/// so a future build can evolve the shape; decoding an unknown kind yields +/// `nil` rather than an error, and ingest skips it. +@_spi(Testing) public enum LogJournalEntry: Sendable, Equatable { + /// The session the journal's records belong to — written first, so a + /// recovered journal is attributable on its own. + case session(LogSession) + /// A scope definition; without these a recovered record has IDs but no + /// hierarchy. + case scope(LogScope) + /// One emitted record. + case record(LogJournalRecord) + + private static let version = 1 + + /// One encoding pass: the payload nests as a JSON *object*, not + /// re-encoded bytes — base64-wrapping the payload would inflate every + /// entry ~33% and pay an extra encode on the emit path. + @_spi(Testing) public func encoded() throws -> Data { + switch self { + case let .session(session): + try JSONEncoder().encode( + Envelope(version: Self.version, kind: "session", payload: session), + ) + case let .scope(scope): + try JSONEncoder().encode( + Envelope(version: Self.version, kind: "scope", payload: scope), + ) + case let .record(record): + try JSONEncoder().encode( + Envelope(version: Self.version, kind: "record", payload: record), + ) + } + } + + /// Decode one journal payload. Returns `nil` for a kind this build + /// doesn't know (a newer build wrote it — skip, don't fail); throws + /// for malformed data. Two passes — a cheap kind peek, then the typed + /// decode — which only ingest pays, once per launch. + @_spi(Testing) public static func decoded(from data: Data) throws -> LogJournalEntry? { + switch try JSONDecoder().decode(EnvelopeHeader.self, from: data).kind { + case "session": + try .session(JSONDecoder().decode(Envelope.self, from: data).payload) + case "scope": + try .scope(JSONDecoder().decode(Envelope.self, from: data).payload) + case "record": + try .record(JSONDecoder().decode(Envelope.self, from: data) + .payload) + default: + nil + } + } + + /// The wire shape: version + kind discriminator + the payload as a + /// nested JSON object. + private struct Envelope: Codable { + var version: Int + var kind: String + var payload: Payload + + private enum CodingKeys: String, CodingKey { + case version = "v" + case kind + case payload + } + } + + /// The discriminator alone — decoded first to pick the typed decode. + private struct EnvelopeHeader: Codable { + var kind: String + } +} + +/// A `LogRecord` as journaled: the same durable mirror the store persists +/// (versioned payload JSON, columns, tags, call site), plus the journal +/// sequence that orders replay. +@_spi(Testing) public struct LogJournalRecord: Codable, Sendable, Equatable { + @_spi(Testing) public var id: UUID + @_spi(Testing) public var sequence: Int + @_spi(Testing) public var date: Date + @_spi(Testing) public var levelName: String + @_spi(Testing) public var severity: Int + @_spi(Testing) public var eventName: String + @_spi(Testing) public var eventVersion: Int + @_spi(Testing) public var message: String + @_spi(Testing) public var payload: Data + @_spi(Testing) public var scopes: [UUID] + @_spi(Testing) public var tags: [LogTag] + @_spi(Testing) public var spanID: UUID? + @_spi(Testing) public var spanExitMode: String? + @_spi(Testing) public var callFunction: String? + @_spi(Testing) public var callFileID: String? + @_spi(Testing) public var externalID: String? + @_spi(Testing) public var attachments: [LogJournalAttachment] + + /// Attachment blobs above this size are omitted from the journal (the + /// async path still delivers them when the process survives) — inlining + /// screenshots would blow the emit-latency budget. + @_spi(Testing) public static let maximumInlineAttachmentBytes = 64 * 1024 + + @_spi(Testing) public init(record: LogRecord, sequence: Int) throws { + id = record.id + self.sequence = sequence + date = record.date + levelName = record.level.name + severity = record.level.severity + eventName = record.eventName + eventVersion = record.eventVersion + message = record.message + payload = try JSONEncoder().encode(record.event) + scopes = record.scopes.map(\.rawValue) + tags = record.tags + spanID = record.spanID?.rawValue + spanExitMode = record.spanExit?.mode.rawValue + callFunction = record.callSite?.function + callFileID = record.callSite?.fileID + externalID = record.externalID + attachments = record.attachments.map { attachment in + attachment.data.count <= Self.maximumInlineAttachmentBytes + ? .inline(attachment) + : .omitted( + name: attachment.name, + contentType: attachment.contentType, + byteCount: attachment.data.count, + ) + } + } +} + +/// A journaled attachment: inlined when small, or a marker recording that +/// an oversized blob existed (its bytes only survive if the async path +/// delivered them before the crash). +@_spi(Testing) public enum LogJournalAttachment: Sendable, Equatable { + case inline(LogAttachment) + case omitted(name: String, contentType: LogAttachment.ContentType, byteCount: Int) +} + +// Hand-written rather than synthesized Codable, for two load-bearing +// reasons: `LogAttachment`/`ContentType` don't conform (the content type +// persists via its MIME string, matching the store), and synthesized enum +// coding would freeze the *case names* into the wire format as nesting +// (`{"inline":{"_0":…}}`) — a rename would silently break old journals. +// The flat shape keeps the format independent of Swift-side naming. +extension LogJournalAttachment: Codable { + private enum CodingKeys: String, CodingKey { + case name + case mimeType + case data + case omittedByteCount + } + + public func encode(to encoder: any Encoder) throws { + var container = encoder.container(keyedBy: CodingKeys.self) + switch self { + case let .inline(attachment): + try container.encode(attachment.name, forKey: .name) + try container.encode(attachment.contentType.mimeType, forKey: .mimeType) + try container.encode(attachment.data, forKey: .data) + case let .omitted(name, contentType, byteCount): + try container.encode(name, forKey: .name) + try container.encode(contentType.mimeType, forKey: .mimeType) + try container.encode(byteCount, forKey: .omittedByteCount) + } + } + + public init(from decoder: any Decoder) throws { + let container = try decoder.container(keyedBy: CodingKeys.self) + let name = try container.decode(String.self, forKey: .name) + let contentType = try LogAttachment.ContentType( + mimeType: container.decode(String.self, forKey: .mimeType), + ) + if let data = try container.decodeIfPresent(Data.self, forKey: .data) { + self = .inline(LogAttachment(name: name, contentType: contentType, data: data)) + } else { + self = try .omitted( + name: name, + contentType: contentType, + byteCount: container.decode(Int.self, forKey: .omittedByteCount), + ) + } + } +} diff --git a/Shared/Periscope/PeriscopeCore/Sources/Pipeline/Periscope.swift b/Shared/Periscope/PeriscopeCore/Sources/Pipeline/Periscope.swift index 8bcd2af1..b8c4d115 100644 --- a/Shared/Periscope/PeriscopeCore/Sources/Pipeline/Periscope.swift +++ b/Shared/Periscope/PeriscopeCore/Sources/Pipeline/Periscope.swift @@ -127,6 +127,12 @@ public final class Periscope: LogRecorder, Sendable { var ambientSources: [any AmbientEventSource] = [] var inspectModeEnabled = false var inspectObservers: [UUID: AsyncStream.Continuation] = [:] + /// The crash journal, installed when a `PeriscopeStore` sink is + /// added; every buffered record appends to it synchronously. + var journal: LogJournal? + /// Stamped under this lock so journal replay can restore buffer + /// order even though the file appends happen outside it. + var journalSequence = 0 /// The active drain task; `nil` exactly when nothing is draining. var drainTask: Task? /// The span watchdog. Generation-tagged: respawning with an earlier @@ -183,9 +189,28 @@ public final class Periscope: LogRecorder, Sendable { at: 0, ) } + // Journaling requires a store: the store owns the journal (its + // directory and session), and recovery is persistence. Adding one + // turns the emit-side tap on. + if let store = sink as? PeriscopeStore, let journal = store.journal { + install(journal: journal) + } scheduleDrainIfNeeded() } + /// Install the crash journal: every subsequent buffered record appends + /// to it synchronously, and the scopes seen so far replay into it (a + /// recovered record needs its hierarchy). + @_spi(Testing) public func install(journal: LogJournal) { + let scopes = state.withLock { state in + state.journal = journal + return Array(state.scopes.values) + } + for scope in scopes { + journal.append(scope: scope) + } + } + // MARK: Level floors /// The global minimum level; records below it are discarded at emit. @@ -244,13 +269,14 @@ public final class Periscope: LogRecorder, Sendable { // MARK: LogRecorder public func defineScope(_ scope: LogScope) { - let isNew = state.withLock { state in - guard state.scopes[scope.id] == nil else { return false } + let (isNew, journal) = state.withLock { state -> (Bool, LogJournal?) in + guard state.scopes[scope.id] == nil else { return (false, nil) } state.scopes[scope.id] = scope state.pending.append(.scope(scope)) - return true + return (true, state.journal) } guard isNew else { return } + journal?.append(scope: scope) scheduleDrainIfNeeded() } @@ -263,12 +289,27 @@ public final class Periscope: LogRecorder, Sendable { guard shouldRecord(level: original.level, scopes: original.scopes) else { return } } guard let record = redacted(original) else { return } - state.withLock { state in + let journaled = state.withLock { state -> (LogJournal, Int)? in Self.buffer(record, into: &state, configuration: configuration) + return Self.stampForJournal(&state) + } + // The file append happens outside the lock (I/O must not serialize + // every emitter); the sequence stamped inside it restores buffer + // order at replay. + if let (journal, sequence) = journaled { + journal.append(record, sequence: sequence) } scheduleFollowUp(for: record) } + /// Reserve the next journal sequence, when a journal is installed — + /// call under the state lock, append outside it. + private static func stampForJournal(_ state: inout State) -> (LogJournal, Int)? { + guard let journal = state.journal else { return nil } + state.journalSequence += 1 + return (journal, state.journalSequence) + } + /// Append `record` to the buffers and yield it to live observers — the /// in-lock half of delivery, shared by ``record(_:)`` and /// ``beginSpan(key:span:began:)``. Yields happen *inside* the lock so @@ -445,15 +486,18 @@ public final class Periscope: LogRecorder, Sendable { // One lock acquisition for registry + buffer: the span becomes // visible for closing only with its began already in the pipeline, // so no interleaving can record this span's end first. - let superseded = state.withLock { state -> OpenSpan? in + let (superseded, journaled) = state.withLock { + state -> (OpenSpan?, (LogJournal, Int)?) in let prior = state.openSpans.removeValue(forKey: key) state.openSpans[key] = span - if let buffered { - Self.buffer(buffered, into: &state, configuration: configuration) - } - return prior + guard let buffered else { return (prior, nil) } + Self.buffer(buffered, into: &state, configuration: configuration) + return (prior, Self.stampForJournal(&state)) } if let buffered { + if let (journal, sequence) = journaled { + journal.append(buffered, sequence: sequence) + } scheduleFollowUp(for: buffered) } scheduleWatchdogIfNeeded() diff --git a/Shared/Periscope/PeriscopeCore/Sources/Store/PeriscopeStore.swift b/Shared/Periscope/PeriscopeCore/Sources/Store/PeriscopeStore.swift index 5b1c7f12..57632b7a 100644 --- a/Shared/Periscope/PeriscopeCore/Sources/Store/PeriscopeStore.swift +++ b/Shared/Periscope/PeriscopeCore/Sources/Store/PeriscopeStore.swift @@ -37,7 +37,7 @@ public actor PeriscopeStore: LogSink { /// Internal-failure telemetry: logging must never crash or throw into /// the pipeline, so persistence problems land here and in OSLog. - private static let failureLogger = os.Logger( + static let failureLogger = os.Logger( subsystem: "com.stuff.periscope", category: "PeriscopeStore", ) @@ -49,13 +49,23 @@ public actor PeriscopeStore: LogSink { private var scopeRowCache: [UUID: SDLogScope] = [:] private var tagRowCache: [LogTag: SDLogTag] = [:] private var changeObservers: [UUID: AsyncStream.Continuation] = [:] - private var writeFailures = 0 + var writeFailures = 0 private var nextSequence: Int? #if DEBUG private var pendingWriteFailure: (any Error)? #endif + /// The crash journal for this launch — created for on-disk stores and + /// installed into `Periscope` when the store is added as a sink. + /// Boxed so the `@ModelActor`-synthesized init stays usable and the + /// pipeline can read it without an actor hop. + private nonisolated let journalBox = OSAllocatedUnfairLock(initialState: nil) + + @_spi(Testing) public nonisolated var journal: LogJournal? { + journalBox.withLock { $0 } + } + public static func makeContainer(storage: Storage) throws -> ModelContainer { let schema = Schema(PeriscopeSchema.models) let configuration = ModelConfiguration( @@ -67,21 +77,83 @@ public actor PeriscopeStore: LogSink { } /// App-wiring factory: opens the store and starts `session` so every - /// subsequent event is attributed to this launch. + /// subsequent event is attributed to this launch. On-disk stores also + /// open this launch's crash journal beside the database — the + /// synchronous net `Periscope` writes through once the store is added + /// as a sink. public static func make(storage: Storage, session: LogSession) async throws -> PeriscopeStore { let container = try makeContainer(storage: storage) let store = PeriscopeStore(modelContainer: container) + // Ingest before the session starts: recovered span begans must be + // in the store when startSession's orphan sweep runs, so a crashed + // flow closes as .orphaned instead of vanishing. + if storage == .onDisk { + await store.ingestRecoveredJournals() + } try await store.startSession(session) + if storage == .onDisk { + await store.openJournal(session: session) + } return store } - /// Test/preview factory: a fresh in-memory store per call. + /// Test/preview factory: a fresh in-memory store per call. In-memory + /// stores never journal — journaling is a durability feature, and + /// recovery *is* persistence. @_spi(Testing) public static func inMemory( session: LogSession, ) async throws -> PeriscopeStore { try await make(storage: .inMemory, session: session) } + /// Test factory: a fully journaling on-disk store rooted at an explicit + /// URL, so tests exercise the durability path in isolated temporary + /// directories. + @_spi(Testing) public static func onDisk( + databaseURL: URL, + session: LogSession, + ) async throws -> PeriscopeStore { + let schema = Schema(PeriscopeSchema.models) + let configuration = ModelConfiguration(schema: schema, url: databaseURL) + let container = try ModelContainer(for: schema, configurations: [configuration]) + let store = PeriscopeStore(modelContainer: container) + await store.ingestRecoveredJournals() + try await store.startSession(session) + await store.openJournal(session: session) + return store + } + + // MARK: Journal + + /// Where session journals live: `Periscope-Journals//` + /// beside the database. + @_spi(Testing) public nonisolated func journalDirectory(forSession id: UUID) -> URL { + journalsRoot.appendingPathComponent(id.uuidString, isDirectory: true) + } + + nonisolated var journalsRoot: URL { + let databaseDirectory = modelContainer.configurations.first?.url + .deletingLastPathComponent() + ?? URL.applicationSupportDirectory + return databaseDirectory.appendingPathComponent("Periscope-Journals", isDirectory: true) + } + + /// Open this launch's journal. Degraded-but-handled on failure: an + /// unjournalable disk must not take down logging, so the error logs + /// and counts while the async pipeline keeps delivering. + private func openJournal(session: LogSession) { + do { + let journal = try LogJournal( + directory: journalDirectory(forSession: session.id), + session: session, + ) + journalBox.withLock { $0 = journal } + } catch { + writeFailures += 1 + Self.failureLogger.warning("Failed to open the crash journal: \(error)") + } + } + // MARK: Sessions /// Record `session` as this launch's resource metadata; every event @@ -286,7 +358,7 @@ public actor PeriscopeStore: LogSink { /// save after it, and drop state that may reference rolled-back rows: /// the row caches, and the session-row reference (`ensureActiveSession` /// refetches or reinserts the same session identity on the next write). - private func recoverFromFailedWrite() { + func recoverFromFailedWrite() { modelContext.rollback() scopeRowCache.removeAll() tagRowCache.removeAll() @@ -295,7 +367,7 @@ public actor PeriscopeStore: LogSink { /// Throws the injected test failure, if any (DEBUG-only seam; a no-op /// in release). - private func throwInjectedFailureIfPending() throws { + func throwInjectedFailureIfPending() throws { #if DEBUG if let pendingWriteFailure { self.pendingWriteFailure = nil @@ -306,7 +378,7 @@ public actor PeriscopeStore: LogSink { /// The next monotonic insertion sequence, resuming past the largest /// stored value on the first write of a launch. - private func takeSequence() throws -> Int { + func takeSequence() throws -> Int { if let nextSequence { self.nextSequence = nextSequence + 1 return nextSequence @@ -407,7 +479,7 @@ public actor PeriscopeStore: LogSink { /// Fetch-or-create a scope row. Records normally arrive after their /// scope definitions, but an unknown scope still gets a placeholder row /// (empty name) that a later definition fills in. - private func scopeRow(for id: UUID) throws -> SDLogScope { + func scopeRow(for id: UUID) throws -> SDLogScope { if let cached = scopeRowCache[id] { return cached } @@ -445,7 +517,7 @@ public actor PeriscopeStore: LogSink { .sorted { $0.key.rawValue < $1.key.rawValue } } - private func tagRow(for tag: LogTag) throws -> SDLogTag { + func tagRow(for tag: LogTag) throws -> SDLogTag { if let cached = tagRowCache[tag] { return cached } @@ -956,7 +1028,7 @@ public actor PeriscopeStore: LogSink { changeObservers[id] = nil } - private func notifyChanged() { + func notifyChanged() { for continuation in changeObservers.values { continuation.yield(()) } diff --git a/Shared/Periscope/PeriscopeCore/Sources/Store/PeriscopeStoreJournalIngest.swift b/Shared/Periscope/PeriscopeCore/Sources/Store/PeriscopeStoreJournalIngest.swift new file mode 100644 index 00000000..44f0a62e --- /dev/null +++ b/Shared/Periscope/PeriscopeCore/Sources/Store/PeriscopeStoreJournalIngest.swift @@ -0,0 +1,243 @@ +import Foundation +import JournalKit +import SwiftData + +/// Journal ingest — the recovery half of crash durability. At launch, +/// before the new session starts, the store replays every prior session's +/// journal: records the async pipeline never delivered (the process died +/// first) are persisted, deduplicated against what did arrive, and the +/// journal is deleted. A journal that fails to ingest stays on disk for +/// the next launch to retry. +extension PeriscopeStore { + /// Whether this process is an app extension (widget, share sheet, …). + /// Extensions journal their own sessions but never ingest: ingest + /// deletes journals, and an extension launching mid-app-session must + /// not eat the live app's journal. The app's next launch ingests + /// everyone's. (Full multi-process coordination — the reverse case of + /// an app launch during a live extension session — is tracked in + /// `Shared/Periscope/TODOs.md`.) + private static let isAppExtension = Bundle.main.bundleURL.pathExtension == "appex" + + /// Ingest every prior session's journal under `Periscope-Journals/`. + /// Runs before `startSession`, so recovered span begans participate in + /// the orphan sweep. App processes only — see ``isAppExtension``. + func ingestRecoveredJournals() async { + guard !Self.isAppExtension else { + Self.failureLogger.debug("Skipping crash journal ingest in an app extension") + return + } + let root = journalsRoot + guard FileManager.default.fileExists(atPath: root.path) else { return } + let directories: [URL] + do { + directories = try FileManager.default.contentsOfDirectory( + at: root, + includingPropertiesForKeys: nil, + ) + } catch { + writeFailures += 1 + Self.failureLogger.warning("Failed to list crash journals: \(error)") + return + } + for directory in directories { + await ingestJournal(at: directory) + } + } + + private func ingestJournal(at directory: URL) async { + do { + let recovered = try JournalRecovery.recover(directory: directory) + let journal = Self.parse(recovered.payloads) + + guard let session = journal.session else { + // Without a session entry (a crash tore the very first + // write) the records are unattributable; the journal can + // only be discarded — but audibly. + if !recovered.payloads.isEmpty { + Self.failureLogger.warning( + "Discarding a crash journal with no session entry (\(recovered.payloads.count) entries)", + ) + } + try JournalRecovery.remove(directory: directory) + return + } + + await defineScopes(journal.scopes) + let inserted = try persistRecovered( + records: journal.records.sorted { $0.sequence < $1.sequence }, + session: session, + recovery: recovered, + ) + try JournalRecovery.remove(directory: directory) + let hasGaps = recovered.foundTornEntry || recovered.droppedOlderEntries + if inserted > 0 || hasGaps { + notifyChanged() + Self.failureLogger.info( + """ + Recovered \(inserted) journaled event(s) from session \ + \(session.id) (torn: \(recovered.foundTornEntry), \ + rotated: \(recovered.droppedOlderEntries)) + """, + ) + } + } catch { + // Leave the journal for the next launch to retry. + recoverFromFailedWrite() + writeFailures += 1 + Self.failureLogger.warning( + "Crash journal ingest failed for \(directory.lastPathComponent): \(error)", + ) + } + } + + private struct ParsedJournal { + var session: LogSession? + var scopes: [LogScope] = [] + var records: [LogJournalRecord] = [] + } + + private static func parse(_ payloads: [Data]) -> ParsedJournal { + var parsed = ParsedJournal() + for payload in payloads { + // Undecodable entries skip: an unknown kind is a newer build's + // (nil), and a corrupt payload can't be recovered — the + // journal-level CRC already dropped torn entries. + switch try? LogJournalEntry.decoded(from: payload) { + case let .session(session): + parsed.session = parsed.session ?? session + case let .scope(scope): + parsed.scopes.append(scope) + case let .record(record): + parsed.records.append(record) + case nil: + continue + } + } + return parsed + } + + /// Persist the recovered records the store doesn't already have, plus + /// a marker recording the recovery — and, honestly, its gaps — in one + /// save. Returns how many records were inserted. + private func persistRecovered( + records: [LogJournalRecord], + session: LogSession, + recovery: JournalRecovery.Recovered, + ) throws -> Int { + let existing = try existingEventIDs(among: records) + let missing = records.filter { !existing.contains($0.id) } + let hasGaps = recovery.foundTornEntry || recovery.droppedOlderEntries + // A complete journal whose every record already arrived needs no + // marker; a gappy one tells its story even when nothing inserts. + guard !missing.isEmpty || hasGaps else { return 0 } + + let sessionRow = try fetchOrInsertSession(session) + for record in missing { + let scopeRows = try record.scopes.map { try scopeRow(for: $0) } + let tagRows = try record.tags.map { try tagRow(for: $0) } + let attachmentRows = record.attachments.enumerated() + .compactMap { index, attachment -> SDLogAttachment? in + guard case let .inline(inline) = attachment else { return nil } + return SDLogAttachment( + name: inline.name, + contentType: inline.contentType.mimeType, + index: index, + data: inline.data, + ) + } + let row = try SDLogEvent( + eventID: record.id, + date: record.date, + sequence: takeSequence(), + severity: record.severity, + levelName: record.levelName, + eventName: record.eventName, + eventVersion: record.eventVersion, + message: record.message, + payload: record.payload, + orderedScopeIDs: record.scopes, + sessionID: sessionRow.sessionID, + spanID: record.spanID, + spanExitMode: record.spanExitMode, + callFunction: record.callFunction, + callFileID: record.callFileID, + externalID: record.externalID, + scopes: scopeRows, + tags: tagRows, + attachments: attachmentRows, + ) + modelContext.insert(row) + } + + // The recovery is itself diagnostic gold — mark it in the story, + // attributed to the crashed session, and be honest about what the + // journal could *not* preserve. Gaps escalate the marker to + // .warning (degraded but handled). + var text = "Recovered \(missing.count) event(s) from this session's crash journal" + if recovery.foundTornEntry { + text += "; the journal ended in a torn entry (its record is lost)" + } + if recovery.droppedOlderEntries { + text += "; older entries were dropped by the journal's byte budget" + } + let notice = Message(level: hasGaps ? .warning : .notice, text) + let marker = try SDLogEvent( + eventID: UUID(), + date: Date(), + sequence: takeSequence(), + severity: notice.level.severity, + levelName: notice.level.name, + eventName: Message.eventName, + eventVersion: Message.eventVersion, + message: notice.message, + payload: JSONEncoder().encode(notice), + orderedScopeIDs: [], + sessionID: sessionRow.sessionID, + spanID: nil, + spanExitMode: nil, + callFunction: nil, + callFileID: nil, + externalID: nil, + scopes: [], + tags: [], + attachments: [], + ) + modelContext.insert(marker) + + try throwInjectedFailureIfPending() + try modelContext.save() + return missing.count + } + + /// The recovered IDs the store already persisted (the drain delivered + /// them before the crash), fetched in chunks. + private func existingEventIDs(among records: [LogJournalRecord]) throws -> Set { + var existing: Set = [] + let ids = records.map(\.id) + for start in stride(from: 0, to: ids.count, by: 500) { + let chunk = Array(ids[start ..< min(start + 500, ids.count)]) + let descriptor = FetchDescriptor( + predicate: #Predicate { chunk.contains($0.eventID) }, + ) + try existing.formUnion(modelContext.fetch(descriptor).map(\.eventID)) + } + return existing + } + + /// The crashed session's row — usually already saved by that launch's + /// `startSession`; inserted from the journal's copy when the crash + /// preceded even that. + private func fetchOrInsertSession(_ session: LogSession) throws -> SDLogSession { + let id = session.id + var descriptor = FetchDescriptor( + predicate: #Predicate { $0.sessionID == id }, + ) + descriptor.fetchLimit = 1 + if let existing = try modelContext.fetch(descriptor).first { + return existing + } + let row = SDLogSession(session: session) + modelContext.insert(row) + return row + } +} diff --git a/Shared/Periscope/PeriscopeCore/Tests/LogJournalEntryTests.swift b/Shared/Periscope/PeriscopeCore/Tests/LogJournalEntryTests.swift new file mode 100644 index 00000000..8d58ec3e --- /dev/null +++ b/Shared/Periscope/PeriscopeCore/Tests/LogJournalEntryTests.swift @@ -0,0 +1,97 @@ +import Foundation +@_spi(Testing) import PeriscopeCore +import Testing + +struct LogJournalEntryTests { + @Test func sessionEntriesRoundTrip() throws { + let entry = LogJournalEntry.session(.fixture()) + let decoded = try LogJournalEntry.decoded(from: entry.encoded()) + #expect(decoded == entry) + } + + @Test func scopeEntriesRoundTrip() throws { + let scope = LogScope.root(named: "app").child(named: "photos") + let entry = LogJournalEntry.scope(scope) + let decoded = try LogJournalEntry.decoded(from: entry.encoded()) + #expect(decoded == entry) + } + + @Test func recordEntriesRoundTripEveryField() throws { + let key = LogTagKey("payment-id") + let scope = LogScope.root(named: "app") + let record = LogRecord( + date: Date(timeIntervalSinceReferenceDate: 123), + event: PhotoLogs(photoID: "p1"), + scopes: [scope.id], + tags: [LogTag(key: key, value: .int(3))], + attachments: [ + LogAttachment(name: "response", contentType: .json, data: Data([1, 2])), + ], + callSite: LogCallSite(function: "upload(_:)", fileID: "App/Uploader.swift"), + ) + let journaled = try LogJournalRecord(record: record, sequence: 42) + let entry = LogJournalEntry.record(journaled) + let decoded = try LogJournalEntry.decoded(from: entry.encoded()) + + guard case let .record(back) = try #require(decoded) else { + Issue.record("expected a record entry") + return + } + #expect(back == journaled) + #expect(back.sequence == 42) + #expect(back.eventName == PhotoLogs.eventName) + #expect(back.scopes == [scope.id.rawValue]) + #expect(back.tags[key] == .int(3)) + #expect(back.callFunction == "upload(_:)") + // The payload decodes back to the typed event. + let event = try JSONDecoder().decode(PhotoLogs.self, from: back.payload) + #expect(event.photoID == "p1") + } + + @Test func oversizedAttachmentsAreOmittedWithAMarker() throws { + let scope = LogScope.root(named: "app") + let big = Data(repeating: 0xAB, count: LogJournalRecord.maximumInlineAttachmentBytes + 1) + let record = LogRecord( + date: Date(timeIntervalSinceReferenceDate: 1), + event: Message(level: .info, "screenshotted"), + scopes: [scope.id], + attachments: [ + LogAttachment(name: "small", contentType: .json, data: Data([1])), + LogAttachment(name: "screenshot", contentType: .png, data: big), + ], + ) + let journaled = try LogJournalRecord(record: record, sequence: 1) + + #expect(journaled.attachments == [ + .inline(LogAttachment(name: "small", contentType: .json, data: Data([1]))), + .omitted(name: "screenshot", contentType: .png, byteCount: big.count), + ]) + // And the omission survives the wire format. + let decoded = try LogJournalEntry.decoded(from: LogJournalEntry.record(journaled).encoded()) + #expect(decoded == .record(journaled)) + } + + @Test func envelopesNestPayloadsAsJSONObjectsNotBase64() throws { + // The wire format pins the single-pass encoding: a base64 string + // payload would mean re-encoded bytes and ~33% inflation. + let entry = LogJournalEntry.session(.fixture()) + let object = try #require( + try JSONSerialization.jsonObject(with: entry.encoded()) as? [String: Any], + ) + #expect(object["v"] as? Int == 1) + #expect(object["kind"] as? String == "session") + #expect(object["payload"] is [String: Any]) + } + + @Test func unknownKindsDecodeAsNilNotErrors() throws { + // A newer build's entry kind must be skippable, not fatal. + let future = Data(#"{"v":9,"kind":"hologram","payload":""}"#.utf8) + #expect(try LogJournalEntry.decoded(from: future) == nil) + } + + @Test func malformedEntriesThrow() { + #expect(throws: (any Error).self) { + try LogJournalEntry.decoded(from: Data([0xFF, 0x00, 0x01])) + } + } +} diff --git a/Shared/Periscope/PeriscopeCore/Tests/LogJournalTests.swift b/Shared/Periscope/PeriscopeCore/Tests/LogJournalTests.swift new file mode 100644 index 00000000..4a20d700 --- /dev/null +++ b/Shared/Periscope/PeriscopeCore/Tests/LogJournalTests.swift @@ -0,0 +1,181 @@ +import Foundation +import JournalKit +@_spi(Testing) import PeriscopeCore +import Testing + +struct LogJournalTests { + private func makeDirectory() -> URL { + URL(fileURLWithPath: NSTemporaryDirectory()) + .appendingPathComponent("log-journal-tests-\(UUID().uuidString)", isDirectory: true) + } + + /// Decode every recovered payload this build understands. + private func entries(in directory: URL) throws -> [LogJournalEntry] { + try JournalRecovery.recover(directory: directory).payloads + .compactMap { try LogJournalEntry.decoded(from: $0) } + } + + @Test func opensWithTheSessionAsItsFirstEntry() throws { + let directory = makeDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let session = LogSession.fixture() + let journal = try LogJournal(directory: directory, session: session) + journal.close() + + #expect(try entries(in: directory) == [.session(session)]) + } + + @Test func journalCapturesRecordsBeforeAnyDrain() throws { + // The whole point: records reach disk synchronously at emit, even + // when the async pipeline never runs — a gated sink stands in for + // "the process died before draining". + let directory = makeDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let gate = GateSink() + let system = Periscope(configuration: Periscope.Configuration(), sinks: [gate]) + let journal = try LogJournal(directory: directory, session: .fixture()) + system.install(journal: journal) + + let log = Log(system: system)(for: "checkout") + log.info("about to crash") + log.error("crashed") + + let recovered = try entries(in: directory) + let records = recovered.compactMap { entry -> LogJournalRecord? in + guard case let .record(record) = entry else { return nil } + return record + } + #expect(records.map(\.message) == ["about to crash", "crashed"]) + #expect(records.map(\.sequence) == records.map(\.sequence).sorted()) + + // Scope definitions rode along, so the hierarchy is recoverable. + let scopes = recovered.compactMap { entry -> LogScope? in + guard case let .scope(scope) = entry else { return nil } + return scope + } + #expect(scopes.contains { $0.name == "checkout" }) + gate.open() + } + + @Test func spanPairsJournalInOrder() throws { + let directory = makeDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let system = Periscope(configuration: Periscope.Configuration(), sinks: [CapturingSink()]) + let journal = try LogJournal(directory: directory, session: .fixture()) + system.install(journal: journal) + + let log = Log(system: system) + log.begin(for: "checkout", lifetime: .indefinite) + log.end(for: "checkout", exit: .success) + + let records = try entries(in: directory).compactMap { entry -> LogJournalRecord? in + guard case let .record(record) = entry else { return nil } + return record + } + #expect(records.map(\.eventName) == [SpanBegan.eventName, SpanEnded.eventName]) + #expect(records[0].sequence < records[1].sequence) + #expect(records[0].spanID == records[1].spanID) + } + + @Test func journalWritesOnlyRedactedContent() throws { + // The journal taps in *after* redaction — the highest-consequence + // placement in the pipeline, since journal files outlive the + // process. The raw segment bytes are the tripwire: the secret must + // not appear anywhere on disk, and a suppressed record must not + // appear at all. + let directory = makeDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let system = Periscope( + configuration: Periscope.Configuration(redact: { record in + if record.message.contains("suppress-me") { + return nil + } + return LogRecord( + id: record.id, + date: record.date, + event: Message(level: record.level, "[redacted]"), + scopes: record.scopes, + ) + }), + sinks: [CapturingSink()], + ) + let journal = try LogJournal(directory: directory, session: .fixture()) + system.install(journal: journal) + + let log = Log(system: system) + log.info("card number 4242-secret") + log.info("suppress-me entirely") + + let records = try entries(in: directory).compactMap { entry -> LogJournalRecord? in + guard case let .record(record) = entry else { return nil } + return record + } + #expect(records.map(\.message) == ["[redacted]"]) + + // The leak vectors are the journaled payload JSON, message, and + // tags — check the decoded content (the envelope base64-wraps + // nested payloads, so raw bytes can't be grepped for them). + let payloadJSON = records.map { String(decoding: $0.payload, as: UTF8.self) }.joined() + #expect(!payloadJSON.contains("4242-secret")) + #expect(payloadJSON.contains("[redacted]")) + + // And nothing leaks in plaintext at the envelope level either. + let segmentBytes = try FileManager.default + .contentsOfDirectory(at: directory, includingPropertiesForKeys: nil) + .map { try Data(contentsOf: $0) } + .reduce(Data(), +) + let onDisk = String(decoding: segmentBytes, as: UTF8.self) + #expect(!onDisk.contains("4242-secret")) + #expect(!onDisk.contains("suppress-me")) + } + + @Test func journalFailuresCountInsteadOfThrowingIntoTheEmitPath() throws { + let directory = makeDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let system = Periscope(configuration: Periscope.Configuration(), sinks: [CapturingSink()]) + let journal = try LogJournal(directory: directory, session: .fixture()) + system.install(journal: journal) + journal.close() // every append now fails at the file layer + + let log = Log(system: system) + log.info("undeterred") + + // Two failed appends: the root scope definition (Log init) and the + // record — both swallowed into telemetry, neither thrown. + #expect(journal.appendFailureCount == 2) + } + + @Test func onDiskStoresOpenAndInstallTheJournal() async throws { + // No deferred cleanup: the ModelContainer outlives the test body, + // and deleting the database under it logs CoreData I/O errors. + // The directory lives in the simulator's ephemeral tmp. + let root = makeDirectory() + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + let session = LogSession.fixture() + let store = try await PeriscopeStore.onDisk( + databaseURL: root.appendingPathComponent("Periscope.store"), + session: session, + ) + let system = Periscope(configuration: Periscope.Configuration(), sinks: []) + system.add(sink: store) + + Log(system: system).warning("journaled via the store") + + let directory = store.journalDirectory(forSession: session.id) + let recovered = try entries(in: directory) + #expect(recovered.first == .session(session)) + #expect(recovered.contains { entry in + guard case let .record(record) = entry else { return false } + return record.message == "journaled via the store" + }) + } + + @Test func inMemoryStoresNeverJournal() async throws { + let store = try await PeriscopeStore.inMemory(session: .fixture()) + let system = Periscope(configuration: Periscope.Configuration(), sinks: []) + system.add(sink: store) + Log(system: system).info("ephemeral") + // No journal directory exists; nothing to recover, nothing crashed. + #expect(store.journal == nil) + } +} diff --git a/Shared/Periscope/PeriscopeCore/Tests/PeriscopeCoreTestSupport.swift b/Shared/Periscope/PeriscopeCore/Tests/PeriscopeCoreTestSupport.swift index 35615cc5..c847524f 100644 --- a/Shared/Periscope/PeriscopeCore/Tests/PeriscopeCoreTestSupport.swift +++ b/Shared/Periscope/PeriscopeCore/Tests/PeriscopeCoreTestSupport.swift @@ -198,6 +198,12 @@ extension LogSession { } } +/// A deterministic instant `offset` seconds into the reference era, for +/// store and journal tests. +func date(_ offset: TimeInterval) -> Date { + Date(timeIntervalSinceReferenceDate: offset) +} + /// A freeform record with an explicit date, for deterministic store tests. func makeRecord( _ text: String, diff --git a/Shared/Periscope/PeriscopeCore/Tests/PeriscopeStoreJournalIngestTests.swift b/Shared/Periscope/PeriscopeCore/Tests/PeriscopeStoreJournalIngestTests.swift new file mode 100644 index 00000000..bd3fe011 --- /dev/null +++ b/Shared/Periscope/PeriscopeCore/Tests/PeriscopeStoreJournalIngestTests.swift @@ -0,0 +1,270 @@ +import Foundation +import JournalKit +@_spi(Testing) import PeriscopeCore +import Testing + +struct PeriscopeStoreJournalIngestTests { + /// A root directory standing in for the app's storage across + /// "launches". Lives in the simulator's ephemeral tmp; not deleted in + /// the test body because ModelContainers outlive it. + private func makeRoot() throws -> URL { + let root = URL(fileURLWithPath: NSTemporaryDirectory()) + .appendingPathComponent("ingest-tests-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: root, withIntermediateDirectories: true) + return root + } + + private func databaseURL(in root: URL) -> URL { + root.appendingPathComponent("Periscope.store") + } + + /// Simulate a crashed launch: journal a session, scopes, and records + /// into the directory a store at `databaseURL` will look in — then + /// "crash" (close without any pipeline delivery). + private func writeCrashedJournal( + root: URL, + session: LogSession, + scopes: [LogScope], + records: [LogRecord], + ) throws { + let journalDirectory = root + .appendingPathComponent("Periscope-Journals", isDirectory: true) + .appendingPathComponent(session.id.uuidString, isDirectory: true) + let journal = try LogJournal(directory: journalDirectory, session: session) + let system = Periscope(configuration: Periscope.Configuration(), sinks: []) + system.install(journal: journal) + for scope in scopes { + system.defineScope(scope) + } + for record in records { + system.record(record) + } + journal.close() + } + + @Test func crashedSessionsRecoverIntoTheStore() async throws { + let root = try makeRoot() + let crashed = LogSession.fixture(startedAt: date(0)) + let scope = LogScope.root(named: "app").child(named: "checkout") + let key = LogTagKey("payment-id") + try writeCrashedJournal( + root: root, + session: crashed, + scopes: [LogScope.root(named: "app"), scope], + records: [ + LogRecord( + date: date(1), + event: Message(level: .error, "about to die"), + scopes: [scope.id], + tags: [LogTag(key: key, value: "pay_1")], + callSite: LogCallSite(function: "buy()", fileID: "App/Checkout.swift"), + ), + ], + ) + + // The "next launch": the store ingests before its session starts. + let store = try await PeriscopeStore.onDisk( + databaseURL: databaseURL(in: root), + session: .fixture(startedAt: date(100)), + ) + + let events = try await store.events(matching: LogQuery()) + let recovered = try #require(events.first { $0.message == "about to die" }) + #expect(recovered.sessionID == crashed.id) + #expect(recovered.level == .error) + #expect(recovered.tags == [LogTag(key: key, value: "pay_1")]) + #expect(recovered.callSite?.function == "buy()") + #expect(recovered.scopes == [scope.id]) + // The hierarchy came along. + #expect(try await store.scope(for: scope.id) == scope) + // The recovery is marked in the story, on the crashed session. + let marker = try #require(events.first { $0.message.contains("crash journal") }) + #expect(marker.sessionID == crashed.id) + #expect(marker.level == .notice) + // And both launches' sessions exist. + #expect(try await store.sessions().count == 2) + // The crashed session's journal is gone once ingested (the new + // launch's own journal directory remains, as it should). + let crashedJournal = store.journalDirectory(forSession: crashed.id) + #expect(!FileManager.default.fileExists(atPath: crashedJournal.path)) + } + + @Test func alreadyDeliveredRecordsAreNotDuplicated() async throws { + let root = try makeRoot() + let crashed = LogSession.fixture(startedAt: date(0)) + let scope = LogScope.root(named: "app") + + // First launch: one record made it through the pipeline to the + // store before the crash; a second only reached the journal. + let firstStore = try await PeriscopeStore.onDisk( + databaseURL: databaseURL(in: root), + session: crashed, + ) + let delivered = LogRecord( + date: date(1), + event: Message(level: .info, "delivered"), + scopes: [scope.id], + ) + await firstStore.defineScopes([scope]) + await firstStore.write([delivered]) + let journal = try #require(firstStore.journal) + journal.append(delivered, sequence: 1) + journal.append( + LogRecord( + date: date(2), + event: Message(level: .info, "journal only"), + scopes: [scope.id], + ), + sequence: 2, + ) + journal.close() + + // Second launch, same database: dedupe by event ID. + let store = try await PeriscopeStore.onDisk( + databaseURL: databaseURL(in: root), + session: .fixture(startedAt: date(100)), + ) + let events = try await store.events(matching: LogQuery()) + #expect(events.count(where: { $0.message == "delivered" }) == 1) + #expect(events.count(where: { $0.message == "journal only" }) == 1) + } + + @Test func recoveredSpanBegansJoinTheOrphanSweep() async throws { + let root = try makeRoot() + let crashed = LogSession.fixture(startedAt: date(0)) + let scope = LogScope.root(named: "app") + let began = SpanBegan( + spanID: SpanID(), + name: "checkout", + lifetime: .indefinite, + relaunchPolicy: .endsWithProcess, + ) + // No floors are set on the crashed system, so the raw SpanBegan + // record passes straight through the normal record path. + let record = LogRecord(date: date(1), event: began, scopes: [scope.id]) + try writeCrashedJournal(root: root, session: crashed, scopes: [scope], records: [record]) + + // The next launch ingests the began, then the session start's + // orphan sweep closes the flow the crash abandoned. + let store = try await PeriscopeStore.onDisk( + databaseURL: databaseURL(in: root), + session: .fixture(startedAt: date(100)), + ) + let pair = try await store.events(inSpan: began.spanID) + #expect(pair.count == 2) + let ended = try #require(pair.first { $0.eventName == SpanEnded.eventName }) + #expect(try ended.decode(SpanEnded.self).exit == .orphaned) + } + + @Test func tornJournalsEscalateTheRecoveryMarker() async throws { + let root = try makeRoot() + let crashed = LogSession.fixture(startedAt: date(0)) + let scope = LogScope.root(named: "app") + try writeCrashedJournal( + root: root, + session: crashed, + scopes: [scope], + records: [ + LogRecord( + date: date(1), + event: Message(level: .info, "intact"), + scopes: [scope.id], + ), + LogRecord(date: date(2), event: Message(level: .info, "torn"), scopes: [scope.id]), + ], + ) + // Tear the final entry, as a crash mid-append would. + let journalDirectory = root + .appendingPathComponent("Periscope-Journals", isDirectory: true) + .appendingPathComponent(crashed.id.uuidString, isDirectory: true) + let segment = try FileManager.default + .contentsOfDirectory(at: journalDirectory, includingPropertiesForKeys: nil) + .first { $0.pathExtension == "journalsegment" } + let segmentURL = try #require(segment) + let bytes = try Data(contentsOf: segmentURL) + try bytes.prefix(bytes.count - 5).write(to: segmentURL) + + let store = try await PeriscopeStore.onDisk( + databaseURL: databaseURL(in: root), + session: .fixture(startedAt: date(100)), + ) + let events = try await store.events(matching: LogQuery()) + #expect(events.contains { $0.message == "intact" }) + #expect(!events.contains { $0.message == "torn" }) + let marker = try #require(events.first { $0.message.contains("crash journal") }) + #expect(marker.level == .warning) + #expect(marker.message.contains("torn entry")) + } + + @Test func rotatedJournalsMarkTheirDroppedEntries() async throws { + // A journal that rotated away old segments still attributes itself + // (the session entry heads every segment) and its marker admits + // the loss. + let root = try makeRoot() + let crashed = LogSession.fixture(startedAt: date(0)) + let scope = LogScope.root(named: "app") + let journalDirectory = root + .appendingPathComponent("Periscope-Journals", isDirectory: true) + .appendingPathComponent(crashed.id.uuidString, isDirectory: true) + let journal = try Journal( + directory: journalDirectory, + configuration: Journal.Configuration( + maximumByteCount: 4096, + segmentHeader: LogJournalEntry.session(crashed).encoded(), + ), + ) + try journal.append(LogJournalEntry.scope(scope).encoded(), sync: .processDeath) + for index in 0 ..< 40 { + let record = LogRecord( + date: date(TimeInterval(index)), + event: Message(level: .info, "storm-\(index)"), + scopes: [scope.id], + ) + try journal.append( + LogJournalEntry.record(LogJournalRecord(record: record, sequence: index)).encoded(), + sync: .processDeath, + ) + } + journal.close() + + let store = try await PeriscopeStore.onDisk( + databaseURL: databaseURL(in: root), + session: .fixture(startedAt: date(100)), + ) + let events = try await store.events(matching: LogQuery()) + // The newest records survived; the oldest rotated away. + #expect(events.contains { $0.message == "storm-39" }) + #expect(!events.contains { $0.message == "storm-0" }) + let marker = try #require(events.first { $0.message.contains("crash journal") }) + #expect(marker.level == .warning) + #expect(marker.message.contains("older entries were dropped")) + #expect(marker.sessionID == crashed.id) + } + + @Test func journalsWithoutASessionEntryAreDiscarded() async throws { + let root = try makeRoot() + // A torn first write: entries exist, none of them the session. + let orphanDirectory = root + .appendingPathComponent("Periscope-Journals", isDirectory: true) + .appendingPathComponent(UUID().uuidString, isDirectory: true) + let raw = try Journal( + directory: orphanDirectory, + configuration: Journal.Configuration(maximumByteCount: 1_000_000), + ) + try raw.append( + LogJournalEntry.scope(LogScope.root(named: "app")).encoded(), + sync: .processDeath, + ) + raw.close() + + let store = try await PeriscopeStore.onDisk( + databaseURL: databaseURL(in: root), + session: .fixture(), + ) + #expect(try await store.events(matching: LogQuery()).isEmpty) + let journals = root.appendingPathComponent("Periscope-Journals") + // The unattributable journal is gone, not retried forever. + let remaining = try FileManager.default.contentsOfDirectory(atPath: journals.path) + #expect(!remaining.contains(orphanDirectory.lastPathComponent)) + } +} diff --git a/Shared/Periscope/PeriscopeCore/Tests/PeriscopeStoreTests.swift b/Shared/Periscope/PeriscopeCore/Tests/PeriscopeStoreTests.swift index 614e4237..bfc01573 100644 --- a/Shared/Periscope/PeriscopeCore/Tests/PeriscopeStoreTests.swift +++ b/Shared/Periscope/PeriscopeCore/Tests/PeriscopeStoreTests.swift @@ -5,10 +5,6 @@ import Testing private struct InjectedSaveFailure: Error {} struct PeriscopeStoreTests { - private func date(_ offset: TimeInterval) -> Date { - Date(timeIntervalSinceReferenceDate: offset) - } - /// A store with a small defined hierarchy: app → photos → album-1. private func makeStore() async throws -> ( store: PeriscopeStore, diff --git a/Shared/Periscope/Prototypes/JournalBenchmark/AGENTS.md b/Shared/Periscope/Prototypes/JournalBenchmark/AGENTS.md new file mode 100644 index 00000000..1cefb18d --- /dev/null +++ b/Shared/Periscope/Prototypes/JournalBenchmark/AGENTS.md @@ -0,0 +1,9 @@ +# JournalBenchmark – Module Shape + +A standalone macOS benchmark prototype (see [`README.md`](README.md)) +comparing journal implementations for Periscope's crash-durability design — +it is **not** wired into the root `Package.swift`, any Tuist target, or CI, +and never ships. Build and run it directly with SwiftPM (`swift build -c +release`). Results and caveats live in the README; keep them updated if the +harness changes. Repo-wide rules live in the root +[`AGENTS.md`](../../../../AGENTS.md). diff --git a/Shared/Periscope/Prototypes/JournalBenchmark/Package.resolved b/Shared/Periscope/Prototypes/JournalBenchmark/Package.resolved new file mode 100644 index 00000000..6667d0ea --- /dev/null +++ b/Shared/Periscope/Prototypes/JournalBenchmark/Package.resolved @@ -0,0 +1,15 @@ +{ + "originHash" : "673e70f1a98448affffd0335dbe54d2a9c17b8ddf15f28269507190ad48d341f", + "pins" : [ + { + "identity" : "grdb.swift", + "kind" : "remoteSourceControl", + "location" : "https://github.com/groue/GRDB.swift.git", + "state" : { + "revision" : "b83108d10f42680d78f23fe4d4d80fc88dab3212", + "version" : "7.11.1" + } + } + ], + "version" : 3 +} diff --git a/Shared/Periscope/Prototypes/JournalBenchmark/Package.swift b/Shared/Periscope/Prototypes/JournalBenchmark/Package.swift new file mode 100644 index 00000000..f1926e72 --- /dev/null +++ b/Shared/Periscope/Prototypes/JournalBenchmark/Package.swift @@ -0,0 +1,18 @@ +// swift-tools-version: 6.2 + +import PackageDescription + +let package = Package( + name: "JournalBenchmark", + platforms: [.macOS(.v15)], + dependencies: [ + .package(url: "https://github.com/groue/GRDB.swift.git", from: "7.0.0"), + ], + targets: [ + .executableTarget( + name: "JournalBenchmark", + dependencies: [.product(name: "GRDB", package: "GRDB.swift")], + swiftSettings: [.swiftLanguageMode(.v5)], + ), + ], +) diff --git a/Shared/Periscope/Prototypes/JournalBenchmark/README.md b/Shared/Periscope/Prototypes/JournalBenchmark/README.md new file mode 100644 index 00000000..5bec34c1 --- /dev/null +++ b/Shared/Periscope/Prototypes/JournalBenchmark/README.md @@ -0,0 +1,91 @@ +# JournalBenchmark + +A standalone prototype measuring the candidate implementations for +Periscope's crash-durability journal — the synchronous write-ahead net that +closes the emit-to-sink loss window. **Not a shipping target**: it is not +wired into the root `Package.swift` or any scheme. + +## Candidates + +| Variant | What it is | +|---|---| +| `file` / `file-fsync` | Append-only file, length-prefixed entries, `write(2)` under a lock (± `fsync` per append) | +| `sqlite-normal` / `sqlite-full` | Raw SQLite C API, WAL, one autocommit insert per record (`synchronous=NORMAL` / `FULL`) | +| `grdb` | GRDB `DatabaseQueue`, WAL + `synchronous=NORMAL`, one write per record | +| `coredata` / `coredata-batched` | `NSManagedObjectContext.performAndWait` insert, `save()` per record / per 100 | +| `swiftdata` / `swiftdata-batched` | `ModelContext` under a lock, `save()` per record / per 100 | + +Each run measures per-append latency distributions (single-threaded and +4-thread contended), throughput, and *actual* crash durability: a child +process appends 1050 records then `SIGKILL`s itself with no teardown; the +parent reopens the journal and counts what survived. + +## Running + +```bash +swift build -c release && ./.build/release/JournalBenchmark +``` + +## Results (2026-07-15, M-series macOS, release build) + +Caveats: macOS NVMe/APFS, not iPhone storage — absolute numbers will shift +on device, relative ordering should not. Darwin `fsync` does not force +platter durability (that's `F_FULLFSYNC`); every variant here is measured +at its app-crash-durable configuration, which is the design target. + +### Emit-path latency — 5000 records, ~350-byte JSON entries, single thread + +| variant | p50 µs | p90 µs | p99 µs | p99.9 µs | max µs | ops/s | +|------------------------|--------|--------|--------|----------|--------|-------| +| file | 1.3 | 3.0 | 4.7 | 8.1 | 40 | 559501 | +| file-fsync | 18.2 | 27.0 | 38.1 | 78.9 | 229 | 51065 | +| sqlite-normal | 6.7 | 13.0 | 15.9 | 2097.2 | 5697 | 80360 | +| sqlite-full | 34.2 | 51.7 | 89.3 | 11326.1 | 15513 | 16512 | +| grdb | 12.9 | 18.9 | 22.5 | 2682.7 | 10480 | 49893 | +| coredata | 53.8 | 66.0 | 166.0 | 3367.7 | 3857 | 15189 | +| coredata-batched | 1.0 | 1.6 | 327.5 | 406.2 | 6409 | 150606 | +| swiftdata | 166.9 | 184.2 | 2536.1 | 3465.8 | 4439 | 4753 | +| swiftdata-batched | 3.5 | 4.4 | 2606.0 | 2920.4 | 7729 | 30253 | + +### Contended — 4 threads emitting concurrently + +| variant | p50 µs | p90 µs | p99 µs | p99.9 µs | max µs | ops/s | +|------------------------|--------|--------|--------|----------|--------|-------| +| file ×4 | 3.0 | 38.2 | 134.8 | 258.2 | 363 | 278104 | +| file-fsync ×4 | 20.0 | 36.1 | 780.3 | 5041.2 | 39519 | 46297 | +| sqlite-normal ×4 | 15.8 | 22.4 | 37.3 | 2839.5 | 82845 | 43289 | +| sqlite-full ×4 | 36.5 | 54.2 | 103.0 | 16207.3 | 231106 | 15231 | +| grdb ×4 | 90.2 | 110.2 | 991.1 | 3473.9 | 13602 | 30710 | +| coredata ×4 | 206.5 | 217.1 | 996.6 | 3730.0 | 3880 | 16796 | +| coredata-batched ×4 | 17.0 | 20.2 | 385.4 | 1519.2 | 5581 | 108121 | +| swiftdata ×4 | 168.2 | 189.0 | 2443.3 | 3430.8 | 790598 | 4825 | +| swiftdata-batched ×4 | 5.5 | 56.4 | 2948.5 | 14690.2 | 28140 | 28980 | + +### Durability — child SIGKILLs itself mid-stream (no teardown) + +| variant | recovered | +|------------------------|-----------| +| file | all 1050 | +| file-fsync | all 1050 | +| sqlite-normal | all 1050 | +| sqlite-full | all 1050 | +| grdb | all 1050 | +| coredata | all 1050 | +| coredata-batched | 1000 of 1050 (lost 50) | +| swiftdata | all 1050 | +| swiftdata-batched | 1000 of 1050 (lost 50) | + +## Reading + +- Every **per-record-commit** variant survives SIGKILL fully — including + unfsynced appends and `synchronous=NORMAL` WAL: page-cache writes survive + process death, empirically. Both **batched** variants lose exactly their + unsaved tail: the reopened window, made visible. +- The **file** append is the only variant whose worst case stays in + microseconds (max 40µs single-threaded, 363µs contended). Every + SQLite-backed variant — raw, GRDB, Core Data, SwiftData — has + multi-millisecond tails on the emit path (WAL checkpoints, save + machinery), i.e. an occasional log call that stalls for milliseconds. +- **SwiftData** per-record is the slowest by far (167µs median, 2.5ms p99, + 790ms contended max) and its context is not thread-safe; **Core Data** + per-record is ~40× the file's median with millisecond tails. diff --git a/Shared/Periscope/Prototypes/JournalBenchmark/Sources/JournalBenchmark/Journals.swift b/Shared/Periscope/Prototypes/JournalBenchmark/Sources/JournalBenchmark/Journals.swift new file mode 100644 index 00000000..e6bf4947 --- /dev/null +++ b/Shared/Periscope/Prototypes/JournalBenchmark/Sources/JournalBenchmark/Journals.swift @@ -0,0 +1,356 @@ +import CoreData +import Foundation +import GRDB +import SQLite3 +import SwiftData + +/// One journal candidate: appends length-prefixed/row entries synchronously, +/// and can count what a *fresh* open recovers (the durability check). +protocol JournalStore { + /// Append one entry. Must be safe to call from multiple threads. + func append(seq: Int, payload: Data) throws + /// Commit anything a batched variant is still holding (end of run). + func finish() throws +} + +enum Variant: String, CaseIterable { + case file + case fileFsync = "file-fsync" + case sqliteNormal = "sqlite-normal" + case sqliteFull = "sqlite-full" + case grdb + case coreData = "coredata" + case coreDataBatched = "coredata-batched" + case swiftData = "swiftdata" + case swiftDataBatched = "swiftdata-batched" + + func make(at url: URL) throws -> JournalStore { + switch self { + case .file: try FileJournal(url: url, fsyncEachAppend: false) + case .fileFsync: try FileJournal(url: url, fsyncEachAppend: true) + case .sqliteNormal: try SQLiteJournal(url: url, synchronous: "NORMAL") + case .sqliteFull: try SQLiteJournal(url: url, synchronous: "FULL") + case .grdb: try GRDBJournal(url: url) + case .coreData: try CoreDataJournal(url: url, saveEvery: 1) + case .coreDataBatched: try CoreDataJournal(url: url, saveEvery: 100) + case .swiftData: try SwiftDataJournal(url: url, saveEvery: 1) + case .swiftDataBatched: try SwiftDataJournal(url: url, saveEvery: 100) + } + } + + /// Open the journal fresh (as the next launch would) and count entries. + func recoveredCount(at url: URL) throws -> Int { + switch self { + case .file, .fileFsync: try FileJournal.recover(url: url) + case .sqliteNormal, .sqliteFull: try SQLiteJournal.recover(url: url) + case .grdb: try GRDBJournal.recover(url: url) + case .coreData, .coreDataBatched: try CoreDataJournal.recover(url: url) + case .swiftData, .swiftDataBatched: try SwiftDataJournal.recover(url: url) + } + } +} + +// MARK: - Append-only file + +final class FileJournal: JournalStore { + private let fd: Int32 + private let fsyncEachAppend: Bool + private let lock = NSLock() + + init(url: URL, fsyncEachAppend: Bool) throws { + fd = open(url.path, O_WRONLY | O_CREAT | O_APPEND, 0o644) + guard fd >= 0 else { throw Failure("open failed: \(errno)") } + self.fsyncEachAppend = fsyncEachAppend + } + + func append(seq: Int, payload: Data) throws { + var buffer = Data(capacity: payload.count + 12) + withUnsafeBytes(of: UInt32(payload.count).littleEndian) { buffer.append(contentsOf: $0) } + withUnsafeBytes(of: Int64(seq).littleEndian) { buffer.append(contentsOf: $0) } + buffer.append(payload) + lock.lock() + defer { lock.unlock() } + let written = buffer.withUnsafeBytes { write(fd, $0.baseAddress, $0.count) } + guard written == buffer.count else { throw Failure("short write") } + if fsyncEachAppend { + fsync(fd) + } + } + + func finish() throws {} + + static func recover(url: URL) throws -> Int { + guard let data = try? Data(contentsOf: url) else { return 0 } + var count = 0 + var offset = 0 + while offset + 12 <= data.count { + let length = data.subdata(in: offset ..< offset + 4) + .withUnsafeBytes { $0.loadUnaligned(as: UInt32.self).littleEndian } + let end = offset + 12 + Int(length) + guard end <= data.count else { break } // torn tail + count += 1 + offset = end + } + return count + } +} + +// MARK: - Raw SQLite (WAL) + +final class SQLiteJournal: JournalStore { + private let db: OpaquePointer + private let insert: OpaquePointer + private let lock = NSLock() + + init(url: URL, synchronous: String) throws { + var handle: OpaquePointer? + guard sqlite3_open(url.path, &handle) == SQLITE_OK, let handle else { + throw Failure("sqlite open failed") + } + db = handle + try Self.exec(db, "PRAGMA journal_mode=WAL") + try Self.exec(db, "PRAGMA synchronous=\(synchronous)") + try Self.exec( + db, + "CREATE TABLE IF NOT EXISTS journal(seq INTEGER PRIMARY KEY, payload BLOB NOT NULL)", + ) + var statement: OpaquePointer? + guard sqlite3_prepare_v2( + db, + "INSERT INTO journal(seq, payload) VALUES(?, ?)", + -1, + &statement, + nil, + ) == SQLITE_OK, + let statement + else { throw Failure("prepare failed") } + insert = statement + } + + func append(seq: Int, payload: Data) throws { + lock.lock() + defer { lock.unlock() } + sqlite3_bind_int64(insert, 1, Int64(seq)) + _ = payload.withUnsafeBytes { + sqlite3_bind_blob( + insert, + 2, + $0.baseAddress, + Int32($0.count), + unsafeBitCast(-1, to: sqlite3_destructor_type.self), + ) + } + guard sqlite3_step(insert) == SQLITE_DONE else { throw Failure("insert failed") } + sqlite3_reset(insert) + sqlite3_clear_bindings(insert) + } + + func finish() throws {} + + static func recover(url: URL) throws -> Int { + var handle: OpaquePointer? + guard sqlite3_open(url.path, &handle) == SQLITE_OK, let handle else { return 0 } + defer { sqlite3_close(handle) } + var statement: OpaquePointer? + guard sqlite3_prepare_v2(handle, "SELECT COUNT(*) FROM journal", -1, &statement, nil) == + SQLITE_OK + else { + return 0 + } + defer { sqlite3_finalize(statement) } + guard sqlite3_step(statement) == SQLITE_ROW else { return 0 } + return Int(sqlite3_column_int64(statement, 0)) + } + + private static func exec(_ db: OpaquePointer, _ sql: String) throws { + guard sqlite3_exec(db, sql, nil, nil, nil) == SQLITE_OK else { + throw Failure("exec failed: \(sql)") + } + } +} + +// MARK: - GRDB + +final class GRDBJournal: JournalStore { + private let queue: DatabaseQueue + + init(url: URL) throws { + var configuration = Configuration() + configuration.journalMode = .wal + configuration.prepareDatabase { db in + try db.execute(sql: "PRAGMA synchronous=NORMAL") + } + queue = try DatabaseQueue(path: url.path, configuration: configuration) + try queue.write { db in + try db + .execute( + sql: "CREATE TABLE IF NOT EXISTS journal(seq INTEGER PRIMARY KEY, payload BLOB NOT NULL)", + ) + } + } + + func append(seq: Int, payload: Data) throws { + try queue.write { db in + try db.execute( + sql: "INSERT INTO journal(seq, payload) VALUES(?, ?)", + arguments: [seq, payload], + ) + } + } + + func finish() throws {} + + static func recover(url: URL) throws -> Int { + let queue = try DatabaseQueue(path: url.path) + return try queue.read { db in + try Int.fetchOne(db, sql: "SELECT COUNT(*) FROM journal") ?? 0 + } + } +} + +// MARK: - Core Data + +final class CoreDataJournal: JournalStore { + private let container: NSPersistentContainer + private let context: NSManagedObjectContext + private let saveEvery: Int + private var pending = 0 + + static let model: NSManagedObjectModel = { + let entity = NSEntityDescription() + entity.name = "Entry" + let seq = NSAttributeDescription() + seq.name = "seq" + seq.attributeType = .integer64AttributeType + let payload = NSAttributeDescription() + payload.name = "payload" + payload.attributeType = .binaryDataAttributeType + entity.properties = [seq, payload] + let model = NSManagedObjectModel() + model.entities = [entity] + return model + }() + + init(url: URL, saveEvery: Int) throws { + container = NSPersistentContainer(name: "Journal", managedObjectModel: Self.model) + let description = NSPersistentStoreDescription(url: url) + description.shouldAddStoreAsynchronously = false + container.persistentStoreDescriptions = [description] + var loadError: Error? + container.loadPersistentStores { _, error in loadError = error } + if let loadError { throw loadError } + context = container.newBackgroundContext() + self.saveEvery = saveEvery + } + + func append(seq: Int, payload: Data) throws { + var thrown: Error? + context.performAndWait { + let entry = NSManagedObject( + entity: Self.model.entitiesByName["Entry"]!, + insertInto: context, + ) + entry.setValue(seq, forKey: "seq") + entry.setValue(payload, forKey: "payload") + pending += 1 + if pending >= saveEvery { + do { + try context.save() + pending = 0 + } catch { + thrown = error + } + } + } + if let thrown { throw thrown } + } + + func finish() throws { + var thrown: Error? + context.performAndWait { + if context.hasChanges { + do { try context.save() } catch { thrown = error } + } + } + if let thrown { throw thrown } + } + + static func recover(url: URL) throws -> Int { + let container = NSPersistentContainer(name: "Journal", managedObjectModel: model) + let description = NSPersistentStoreDescription(url: url) + description.shouldAddStoreAsynchronously = false + container.persistentStoreDescriptions = [description] + var loadError: Error? + container.loadPersistentStores { _, error in loadError = error } + if let loadError { throw loadError } + let request = NSFetchRequest(entityName: "Entry") + return try container.newBackgroundContext().performAndWait { + try container.newBackgroundContext().count(for: request) + } + } +} + +// MARK: - SwiftData + +@Model +final class SDEntry { + var seq: Int + var payload: Data + + init(seq: Int, payload: Data) { + self.seq = seq + self.payload = payload + } +} + +final class SwiftDataJournal: JournalStore { + private let container: ModelContainer + private let context: ModelContext + private let saveEvery: Int + private var pending = 0 + private let lock = NSLock() + + init(url: URL, saveEvery: Int) throws { + let configuration = ModelConfiguration(url: url) + container = try ModelContainer(for: SDEntry.self, configurations: configuration) + context = ModelContext(container) + context.autosaveEnabled = false + self.saveEvery = saveEvery + } + + func append(seq: Int, payload: Data) throws { + // ModelContext is not thread-safe; exclusive access via lock is the + // most charitable synchronous shape available to a journal. + lock.lock() + defer { lock.unlock() } + context.insert(SDEntry(seq: seq, payload: payload)) + pending += 1 + if pending >= saveEvery { + try context.save() + pending = 0 + } + } + + func finish() throws { + lock.lock() + defer { lock.unlock() } + if context.hasChanges { + try context.save() + } + } + + static func recover(url: URL) throws -> Int { + let configuration = ModelConfiguration(url: url) + let container = try ModelContainer(for: SDEntry.self, configurations: configuration) + let context = ModelContext(container) + return try context.fetchCount(FetchDescriptor()) + } +} + +struct Failure: Error, CustomStringConvertible { + let description: String + + init(_ description: String) { + self.description = description + } +} diff --git a/Shared/Periscope/Prototypes/JournalBenchmark/Sources/JournalBenchmark/main.swift b/Shared/Periscope/Prototypes/JournalBenchmark/Sources/JournalBenchmark/main.swift new file mode 100644 index 00000000..57f9516b --- /dev/null +++ b/Shared/Periscope/Prototypes/JournalBenchmark/Sources/JournalBenchmark/main.swift @@ -0,0 +1,154 @@ +import Foundation + +/// A representative journal entry: what a real `LogRecord` serializes to +/// (name, level, date, scopes, tags, call site, payload) — ~350 bytes. +func makePayload(seq: Int) -> Data { + let entry: [String: Any] = [ + "id": UUID().uuidString, + "seq": seq, + "date": Date().timeIntervalSinceReferenceDate, + "event": "photo-uploaded", + "version": 1, + "level": ["name": "info", "severity": 200], + "message": "Uploaded photo p\(seq) (48211 bytes) after retry", + "scopes": [UUID().uuidString, UUID().uuidString], + "tags": ["payment-id": "pay_\(seq)", "retry": 2], + "function": "uploadPhoto(_:)", + "file": "Where/PhotoUploader.swift", + "payload": ["photoID": "p\(seq)", "byteCount": 48211], + ] + return try! JSONSerialization.data(withJSONObject: entry, options: [.sortedKeys]) +} + +struct Stats { + let label: String + let microseconds: [Double] + let totalSeconds: Double + + var report: String { + let sorted = microseconds.sorted() + func percentile(_ p: Double) -> Double { + sorted[min(sorted.count - 1, Int(Double(sorted.count) * p))] + } + let throughput = Double(sorted.count) / totalSeconds + return String( + format: "| %@ | %.1f | %.1f | %.1f | %.1f | %.0f | %.0f |", + label.padding(toLength: 22, withPad: " ", startingAt: 0), + percentile(0.5), + percentile(0.9), + percentile(0.99), + percentile(0.999), + sorted.last ?? 0, + throughput, + ) + } +} + +func bench(_ variant: Variant, records: Int, threads: Int, directory: URL) throws -> Stats { + let url = directory.appendingPathComponent("\(variant.rawValue)-\(threads)t.journal") + let store = try variant.make(at: url) + let payloads = (0 ..< records).map(makePayload(seq:)) + let clock = ContinuousClock() + let latencyLock = NSLock() + var microseconds: [Double] = [] + microseconds.reserveCapacity(records) + + let start = clock.now + if threads == 1 { + for seq in 0 ..< records { + let opStart = clock.now + try store.append(seq: seq, payload: payloads[seq]) + let elapsed = clock.now - opStart + microseconds.append(Double(elapsed.components.attoseconds) / 1e12) + } + } else { + let perThread = records / threads + DispatchQueue.concurrentPerform(iterations: threads) { thread in + var local: [Double] = [] + local.reserveCapacity(perThread) + for i in 0 ..< perThread { + let seq = thread * perThread + i + let opStart = clock.now + try? store.append(seq: seq, payload: payloads[seq]) + let elapsed = clock.now - opStart + local.append(Double(elapsed.components.attoseconds) / 1e12) + } + latencyLock.lock() + microseconds.append(contentsOf: local) + latencyLock.unlock() + } + } + try store.finish() + let total = clock.now - start + let bytes = (try? FileManager.default.attributesOfItem(atPath: url.path)[.size] as? Int) ?? 0 + let label = threads == 1 ? variant.rawValue : "\(variant.rawValue) ×\(threads)" + _ = bytes + return Stats( + label: label, + microseconds: microseconds, + totalSeconds: Double(total.components.attoseconds) / 1e18 + + Double(total.components.seconds), + ) +} + +/// Child-process mode: append `killAfter` records, then SIGKILL ourselves — +/// no teardown, no atexit, exactly what a crash does. +func runVictim(variant: Variant, url: URL, killAfter: Int) throws -> Never { + let store = try variant.make(at: url) + for seq in 0 ..< killAfter { + try store.append(seq: seq, payload: makePayload(seq: seq)) + } + raise(SIGKILL) + fatalError("unreachable") +} + +func runDurability(variant: Variant, directory: URL, killAfter: Int) throws -> String { + let url = directory.appendingPathComponent("victim-\(variant.rawValue).journal") + let process = Process() + process.executableURL = URL(fileURLWithPath: CommandLine.arguments[0]) + process.arguments = ["victim", variant.rawValue, url.path, String(killAfter)] + try process.run() + process.waitUntilExit() + let recovered = try variant.recoveredCount(at: url) + let verdict = recovered == killAfter + ? "all \(killAfter)" + : "\(recovered) of \(killAfter) (lost \(killAfter - recovered))" + return "| \(variant.rawValue.padding(toLength: 22, withPad: " ", startingAt: 0)) | \(verdict) |" +} + +// MARK: - Entry + +let arguments = CommandLine.arguments +if arguments.count >= 5, arguments[1] == "victim" { + guard let variant = Variant(rawValue: arguments[2]), let killAfter = Int(arguments[4]) else { + exit(2) + } + try runVictim(variant: variant, url: URL(fileURLWithPath: arguments[3]), killAfter: killAfter) +} + +let directory = URL(fileURLWithPath: NSTemporaryDirectory()) + .appendingPathComponent("journal-benchmark-\(UUID().uuidString)") +try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) +defer { try? FileManager.default.removeItem(at: directory) } + +let records = 5000 +print("## Emit-path latency — \(records) records, ~350-byte JSON entries, single thread\n") +print("| variant | p50 µs | p90 µs | p99 µs | p99.9 µs | max µs | ops/s |") +print("|------------------------|--------|--------|--------|----------|--------|-------|") +for variant in Variant.allCases { + try print(bench(variant, records: records, threads: 1, directory: directory).report) +} + +print("\n## Contended — 4 threads emitting concurrently\n") +print("| variant | p50 µs | p90 µs | p99 µs | p99.9 µs | max µs | ops/s |") +print("|------------------------|--------|--------|--------|----------|--------|-------|") +for variant in Variant.allCases { + try print(bench(variant, records: records, threads: 4, directory: directory).report) +} + +print("\n## Durability — child process SIGKILLs itself mid-stream (no teardown)\n") +print("| variant | recovered |") +print("|------------------------|-----------|") +for variant in Variant.allCases { + try print(runDurability(variant: variant, directory: directory, killAfter: 1050)) +} diff --git a/Shared/Periscope/TODOs.md b/Shared/Periscope/TODOs.md index 7b5525f9..cd1960ff 100644 --- a/Shared/Periscope/TODOs.md +++ b/Shared/Periscope/TODOs.md @@ -9,7 +9,6 @@ # Open issues ## P0s (Must do) -- design: Crash durability — the async pipeline can drop events exactly at crash time. Options from PR review: a synchronous per-event journal the async queue ingests, or writing straight to the store. Needs a plan/build loop. - design: Span record modeling — `spanID`/`spanExit` bolted onto every `LogRecord` (and `bypassesFloors` as a one-off flag) feels wrong; consider `enum { case span(Span), case event(Event) }` or a dedicated span record type. Plan/build loop. - design: Decompose `Periscope` (the type and its flat `State` — group watchdog/inspect/ambient/live-observer state into sub-structs) and `PeriscopeStore` into children per behavioral area. Plan/build loop. - design: `ScopeID` derivation — hash-derived vs a concatenated, human-readable path that preserves the input for debugging. Plan/build loop. @@ -20,6 +19,8 @@ ## P1s (Should do) - fix: After initial merge, we should come back and update the UI to consume the Shared/Broadway design system tooling, eg a PeriscopeStylesheet for components and other recommendations. +- feat: Journal attachments via external storage (PR #86 review). Instead of inlining blobs ≤64KB and omitting larger ones, write attachment bytes as files beside the journal segments (the entry referencing them by filename), clean them up with segment rotation and journal removal, and re-attach them at ingest. Removes the size cliff entirely — screenshots and payloads survive crashes too. Follow-up PR after #86. +- feat: Multi-process store + journal coordination. Today only app processes ingest journals (extensions journal but never ingest, so an extension launch can't delete the live app's journal) — but the reverse hole remains: an app launching while an extension session is live would ingest and delete that *live* journal out from under its open descriptor, silently ending its recoverability. Needs a claim mechanism (e.g. a claim file the writer holds, or skip-directories-with-live-claims) designed alongside App Group store sharing — which the store doesn't support yet either (exclusive sequence counters, SwiftData container coordination). ## P2s (Nice to have) @@ -27,7 +28,8 @@ # Completed issues -## PR review pass (bugs + mechanical reshapes) +## Crash durability (design loop 1) +- feat: Crash journal — every emitted record appends synchronously to a per-session append-only journal (**JournalKit**, a new generic module: CRC-framed segments, torn-tail-tolerant recovery, flight-recorder rotation) once an on-disk store is attached; fault+ records `F_FULLFSYNC`. At the next launch the store ingests prior journals before the session starts (dedupe by event ID, recovered begans join the orphan sweep, `.notice` recovery marker), then deletes them. Decision made on measured data: the benchmark prototype (`Prototypes/JournalBenchmark`) showed the file append is the only candidate with a microseconds-bounded worst case, and SIGKILL children proved page-cache appends survive process death while batched ORM saves lose their unsaved tail. In-memory stores never journal. Remaining refinement: journal the flush-on-background trigger is unnecessary now (the journal covers background kills), and attachment blobs >64KB journal as omitted markers. - fix: Ambient sources retain their NotificationCenter observer tokens (`AmbientObserverTokens`) — dropped tokens made observations unremovable and immortalized the captured system; restarts now replace instead of doubling. `AmbientEventSource` gains `stop()` and `Periscope.stopAmbientSources()`. - feat: `LogAttachment.ContentType` enum (json/png/jpeg/plainText + `.other(mime)`); `LogLevel.osLogType` is a stored, overridable property (identity and Codable stay name + severity); `SpanExit` factories for every mode. - feat: Tags reshaped — `LogTagValue` (typed values incl. any Codable), `[LogTag]` lists everywhere, multi-tag AND queries via filter-count predicate; SDLogTag gains `valueKind`.