ai/byre
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 15 Wiki Activity

Cache task events on disk per server; fix performance audit findings

Browse Source
This commit is contained in:
Ian Gulliver
2026-10-23 12:49:43 -07:00
parent dc2fcc6075
commit d6be98f9d8
13 changed files with 448 additions and 250 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mac/Sources/Byre/EventCache.swift
+139
View File
@@ -0,0 +1,139 @@
import Foundation
private let encoder: JSONEncoder = {
let e = JSONEncoder()
e.dateEncodingStrategy = .secondsSince1970
return e
}()
private let decoder: JSONDecoder = {
let d = JSONDecoder()
d.dateDecodingStrategy = .secondsSince1970
return d
}()
final class EventCache: @unchecked Sendable {
private let dir: URL
private let queue = DispatchQueue(label: "run.fc.byre.events")
private var lastSeqs: [String: Int64] = [:]
private var generation = 0
init(host: String) {
dir = URL.applicationSupportDirectory.appending(path: "run.fc.byre.mac/events/\(host)", directoryHint: .isDirectory)
}
func open(schema: String) {
queue.async {
self.generation += 1
let fm = FileManager.default
let marker = self.dir.appending(path: "schema")
if (try? String(contentsOf: marker, encoding: .utf8)) == schema {
return
}
self.lastSeqs = [:]
try? fm.removeItem(at: self.dir)
do {
try fm.createDirectory(at: self.dir, withIntermediateDirectories: true)
try schema.write(to: marker, atomically: true, encoding: .utf8)
} catch {
log.error("open event cache: \(errorText(error), privacy: .public)")
}
}
}
func load(_ taskId: String) async -> (events: [ByreEvent], generation: Int) {
await withCheckedContinuation { c in
queue.async {
let events = self.read(taskId)
self.lastSeqs[taskId] = events.last?.seq ?? 0
c.resume(returning: (events, self.generation))
}
}
}
func append(_ taskId: String, _ events: [ByreEvent], generation: Int) {
queue.async {
guard generation == self.generation, let last = self.lastSeqs[taskId] else {
return
}
let fresh = events.filter { $0.seq > last }
guard let newest = fresh.last else {
return
}
do {
try self.write(taskId, fresh)
self.lastSeqs[taskId] = newest.seq
} catch {
log.error("cache events of \(taskId, privacy: .public): \(errorText(error), privacy: .public)")
self.forget(taskId)
}
}
}
func drop(_ taskId: String) {
queue.async {
self.forget(taskId)
}
}
func keep(_ taskIds: Set<String>) {
queue.async {
let files = (try? FileManager.default.contentsOfDirectory(at: self.dir, includingPropertiesForKeys: nil)) ?? []
for url in files where url.pathExtension == "ndjson" {
let taskId = url.deletingPathExtension().lastPathComponent
if !taskIds.contains(taskId) {
self.forget(taskId)
}
}
}
}
private func file(_ taskId: String) -> URL {
dir.appending(path: "\(taskId).ndjson")
}
private func forget(_ taskId: String) {
lastSeqs[taskId] = nil
try? FileManager.default.removeItem(at: file(taskId))
}
private func read(_ taskId: String) -> [ByreEvent] {
let url = file(taskId)
guard let data = try? Data(contentsOf: url) else {
return []
}
var events: [ByreEvent] = []
var kept = 0
for line in data.split(separator: 0x0a, omittingEmptySubsequences: false) where !line.isEmpty {
guard let ev = try? decoder.decode(ByreEvent.self, from: line) else {
break
}
events.append(ev)
kept = min(line.endIndex + 1, data.endIndex) - data.startIndex
}
if kept < data.count, let h = try? FileHandle(forWritingTo: url) {
try? h.truncate(atOffset: UInt64(kept))
try? h.close()
}
return events
}
private func write(_ taskId: String, _ events: [ByreEvent]) throws {
var data = Data()
for ev in events {
data.append(try encoder.encode(ev))
data.append(0x0a)
}
let url = file(taskId)
let fm = FileManager.default
if !fm.fileExists(atPath: url.path), !fm.createFile(atPath: url.path, contents: nil) {
throw ApiError(status: 0, message: "create \(url.path)")
}
let h = try FileHandle(forWritingTo: url)
defer {
try? h.close()
}
try h.seekToEnd()
try h.write(contentsOf: data)
}
}
mac/Sources/Byre/Feed.swift
+4 -1
View File
@@ -42,6 +42,7 @@ final class Feed {
@ObservationIgnored private var scheduled = false
@ObservationIgnored private var toolNames: [String: String] = [:]
@ObservationIgnored private let shown: () -> Bool
@ObservationIgnored private var persist: ([ByreEvent]) -> Void = { _ in }
init(shown: @escaping () -> Bool) {
self.shown = shown
@@ -60,12 +61,13 @@ final class Feed {
}
}
func load(_ history: [ByreEvent], inherited: Int = 0) {
func load(_ history: [ByreEvent], inherited: Int = 0, persist: @escaping ([ByreEvent]) -> Void = { _ in }) {
if ready {
return
}
ready = true
self.inherited = inherited
self.persist = persist
buffer = history + buffer
flush()
}
@@ -131,6 +133,7 @@ final class Feed {
return
}
events.append(contentsOf: fresh)
persist(fresh)
let found = findCaptures(fresh)
if !found.isEmpty {
captures.append(contentsOf: found)
mac/Sources/Byre/Startup.swift
+5 -1
View File
@@ -173,6 +173,7 @@ final class Connection {
let overlays = Overlays()
let forwards: Forwards
let pages: Pages
let cache: EventCache
var phase: Phase = .starting
@ObservationIgnored private var stopped = false
@ObservationIgnored private var restarting = false
@@ -181,6 +182,7 @@ final class Connection {
self.origin = origin
forwards = Forwards(host: origin.host()!)
pages = Pages(api: API(origin: origin))
cache = EventCache(host: origin.host()!)
Task {
await start()
}
@@ -231,9 +233,10 @@ final class Connection {
}
private func attempt(_ old: Store?) async -> Bool {
let store = Store(api: API(origin: origin), images: old?.images ?? ImageDrafts(), forwards: forwards, pages: pages)
let store = Store(api: API(origin: origin), images: old?.images ?? ImageDrafts(), forwards: forwards, pages: pages, cache: cache)
do {
let status = try await store.api.client.getStatus().ok.body.json
cache.open(schema: status.schema)
async let me: Void = store.loadMe()
async let projects: Void = store.loadProjects()
async let inbox: Void = store.loadInbox()
@@ -254,6 +257,7 @@ final class Connection {
store.startGlobalStream(after: status.lastEventSeq) { [weak self] in
self?.reconnected(status.startedAt)
}
store.startFeeds()
store.remote.start()
forwards.sync(store.api, Set(store.activeTasks().map(\.task.id)))
Notifier.shared.deliver()
mac/Sources/Byre/Store.swift
+23 -5
View File
@@ -92,9 +92,11 @@ final class Store {
let remote: Remote
let forwards: Forwards
let pages: Pages
let cache: EventCache
private var taskFeeds: [String: Feed] = [:]
@ObservationIgnored private var historyLoads: Set<ObjectIdentifier> = []
@ObservationIgnored private var feedsStarted = false
@ObservationIgnored private var requestSeqs: [String: Int] = [:]
@ObservationIgnored private var taskReloads: Set<String> = []
@ObservationIgnored private var listReloads: Set<String> = []
@@ -109,11 +111,12 @@ final class Store {
let changesMemory = ChangesMemory()
@ObservationIgnored private var localSends: [String: LocalSends] = [:]
init(api: API, images: ImageDrafts, forwards: Forwards, pages: Pages) {
init(api: API, images: ImageDrafts, forwards: Forwards, pages: Pages, cache: EventCache) {
self.api = api
self.images = images
self.forwards = forwards
self.pages = pages
self.cache = cache
leadFeed = Feed(shown: { true })
paneFocus = PaneFocus(api: api)
remote = Remote(api: api)
@@ -443,8 +446,10 @@ final class Store {
private func applyTasks(_ projectId: String, _ list: [ByreTask]) async throws {
tasks[projectId] = list.sorted { $0.createdAt > $1.createdAt }
for t in list where t.status == .active && !t.background {
_ = taskFeed(t.id)
if feedsStarted {
for t in list where t.status == .active && !t.background {
_ = taskFeed(t.id)
}
}
dropClosedFeeds()
drafts?.prune(self)
@@ -622,7 +627,8 @@ final class Store {
}
private func loadHistory(_ taskId: String, _ feed: Feed) async throws {
var got: [ByreEvent] = []
let (cached, generation) = await cache.load(taskId)
var got = cached
while true {
let page = try await api.client.listTaskEvents(path: .init(taskId: taskId), query: .init(after: got.last?.seq ?? 0, limit: historyPage)).ok.body.json
got += page
@@ -630,7 +636,18 @@ final class Store {
break
}
}
feed.load(got, inherited: got.prefix { $0.taskId != taskId }.count)
feed.load(got, inherited: got.prefix { $0.taskId != taskId }.count, persist: { [cache] events in
cache.append(taskId, events, generation: generation)
})
}
func startFeeds() {
feedsStarted = true
let active = activeTasks()
cache.keep(Set(active.map(\.task.id)))
for (_, task) in active where !task.background {
_ = taskFeed(task.id)
}
}
private func taskFeed(_ taskId: String) -> Feed {
@@ -658,6 +675,7 @@ final class Store {
for taskId in taskFeeds.keys {
if taskId != selectedTaskId, let task = findTask(taskId), task.status != .active {
taskFeeds[taskId] = nil
cache.drop(taskId)
}
}
}