From 4a9739b0034c17a724f95d221a5d0dbaed69c2e3 Mon Sep 17 00:00:00 2001 From: AstroQore <69107895+AstroQore@users.noreply.github.com> Date: Thu, 20 Aug 2026 13:54:52 +0800 Subject: [PATCH] fix(session-index): bound FTS rebuild memory Co-Authored-By: Codex --- CHANGELOG.md | 11 ++ .../Sessions/SessionIndexService.swift | 81 ++++++++--- .../Sessions/SessionIndexStore.swift | 128 +++++++++++++++--- .../SessionIndexStoreTests.swift | 43 +++++- 4 files changed, 219 insertions(+), 44 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d752910..353be12 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ## [Unreleased] +### Fixed +- **Full-text session rebuilds now have a bounded working set.** Schema v5 + still indexes every eligible session and the same title, user, assistant, + system, tool, and file-operation excerpts, but commits them in transactions + capped at 4 MiB or 128 sessions instead of one FTS5 segment per session. + Each batch releases SQLite caches and applies malloc-zone pressure relief, + preventing a large first-run rebuild from retaining a multi-gigabyte heap. + SQLite temporary storage is file-backed and its page cache is capped at + 32 MiB; search semantics, snippets, roles, and the existing 512 KiB + per-session body policy are unchanged. + ## [0.4.1] - 2026-08-20 Release validation no longer races an FSEvents stream rearm when a watched diff --git a/Sources/AgentSessionKit/Sessions/SessionIndexService.swift b/Sources/AgentSessionKit/Sessions/SessionIndexService.swift index 3518834..4a3eb8f 100644 --- a/Sources/AgentSessionKit/Sessions/SessionIndexService.swift +++ b/Sources/AgentSessionKit/Sessions/SessionIndexService.swift @@ -18,6 +18,12 @@ public actor SessionIndexService { private let registry: SessionProviderRegistry private let bodyIndexing: @Sendable () -> Bool + /// Body bytes committed in one SQLite/FTS5 transaction. The cap is on + /// source excerpt bytes (not session count), so a batch remains bounded + /// even when several sessions hit the existing 512 KiB per-session cap. + static let indexBatchExcerptByteLimit = 4 * 1024 * 1024 + static let indexBatchEntryLimit = 128 + public init( homeDirectory: String = RealHomeDirectory.path, store: SessionIndexStore, @@ -60,13 +66,32 @@ public actor SessionIndexService { var seen: Set = [] let total = files.count var done = 0 + var batch: [SessionIndexStore.IndexBatchEntry] = [] + var batchExcerptBytes = 0 for file in files { let hash = Self.pathHash(file.url.path) seen.insert(hash) - await index(file: file.url, adapter: file.adapter, pathHash: hash, bodies: indexBodies) + if let entry = await prepareIndexEntry( + file: file.url, + adapter: file.adapter, + pathHash: hash, + bodies: indexBodies + ) { + let entryBytes = entry.excerpts?.reduce(0) { $0 + $1.excerpt.utf8.count } ?? 0 + if !batch.isEmpty, + batchExcerptBytes + entryBytes > Self.indexBatchExcerptByteLimit + || batch.count >= Self.indexBatchEntryLimit { + await commit(batch) + batch.removeAll(keepingCapacity: true) + batchExcerptBytes = 0 + } + batch.append(entry) + batchExcerptBytes += entryBytes + } done += 1 progress?(done, total) } + await commit(batch) do { try await store.pruneMissing(existingPathHashes: seen) @@ -147,38 +172,40 @@ public actor SessionIndexService { ) } - // MARK: - One file + // MARK: - Bounded rebuild batches - private func index( + private func prepareIndexEntry( file url: URL, adapter: any SessionProviderAdapter, pathHash: String, bodies: Bool - ) async { - guard let fingerprint = Self.fingerprint(url) else { return } + ) async -> SessionIndexStore.IndexBatchEntry? { + guard let fingerprint = Self.fingerprint(url) else { return nil } do { if let cursor = try await store.fileCursor(pathHash: pathHash), cursor.mtimeNanos == fingerprint.mtimeNanos, cursor.size == fingerprint.size { - return + return nil } - let summary = try adapter.extractMetadata(fileURL: url) - let row = try await store.upsertSession(summary) - if bodies { - let document = try adapter.parseTranscript(fileURL: url, range: nil) - try await store.replaceMessages( - sessionRow: row, - excerpts: Self.excerpts(from: document, provider: summary.provider) + return try autoreleasepool { + let summary = try adapter.extractMetadata(fileURL: url) + let excerpts: [SessionIndexStore.MessageExcerpt]? + if bodies { + let document = try adapter.parseTranscript(fileURL: url, range: nil) + excerpts = Self.excerpts(from: document, provider: summary.provider) + } else { + excerpts = nil + } + return SessionIndexStore.IndexBatchEntry( + summary: summary, + pathHash: pathHash, + path: url.path, + provider: adapter.provider, + mtimeNanos: fingerprint.mtimeNanos, + size: fingerprint.size, + excerpts: excerpts ) } - try await store.saveFileCursor( - pathHash: pathHash, - path: url.path, - provider: adapter.provider, - mtimeNanos: fingerprint.mtimeNanos, - size: fingerprint.size, - sessionRow: row - ) } catch { // The path is the user's own filesystem; log the file name // only, and let the next refresh retry. @@ -186,7 +213,19 @@ public actor SessionIndexService { "Session index: skipped \(adapter.provider.rawValue) file " + "\(KitLog.sanitize(url.lastPathComponent))." ) + return nil + } + } + + private func commit(_ entries: [SessionIndexStore.IndexBatchEntry]) async { + guard !entries.isEmpty else { return } + do { + try await store.applyIndexBatch(entries) + } catch { + KitLog.warn("Session index: committing a bounded batch failed.") } + await store.releaseTransientMemory() + _ = malloc_zone_pressure_relief(nil, 0) } static func pathHash(_ path: String) -> String { diff --git a/Sources/AgentSessionKit/Sessions/SessionIndexStore.swift b/Sources/AgentSessionKit/Sessions/SessionIndexStore.swift index 57a335c..f297f2a 100644 --- a/Sources/AgentSessionKit/Sessions/SessionIndexStore.swift +++ b/Sources/AgentSessionKit/Sessions/SessionIndexStore.swift @@ -65,7 +65,12 @@ public actor SessionIndexStore { /// /// v4 also changes no columns; it backfills system and tool excerpts now /// that callers can choose the message roles a search is allowed to hit. - static let schemaVersion = 4 + /// + /// v5 rebuilds the same logical index through bounded bulk transactions. + /// The body/search result set is unchanged; only the construction path is + /// different, so an interrupted v4 rebuild cannot keep its pathological + /// one-session-per-FTS-transaction layout. + static let schemaVersion = 5 /// Open (or create) the index at `url`. /// @@ -114,6 +119,8 @@ public actor SessionIndexStore { PRAGMA journal_mode=WAL; PRAGMA synchronous=NORMAL; PRAGMA foreign_keys=ON; + PRAGMA temp_store=FILE; + PRAGMA cache_size=-32768; """ guard sqlite3_exec(database, preamble, nil, nil, nil) == SQLITE_OK else { throw SessionIndexError.open @@ -285,32 +292,72 @@ public actor SessionIndexStore { } } + public struct IndexBatchEntry: Sendable { + public let summary: SessionSummary + public let pathHash: String + public let path: String + public let provider: SessionProvider + public let mtimeNanos: Int64 + public let size: Int64 + /// `nil` means metadata-only indexing. An empty array deliberately + /// clears a session whose readable transcript became empty. + public let excerpts: [MessageExcerpt]? + + public init( + summary: SessionSummary, + pathHash: String, + path: String, + provider: SessionProvider, + mtimeNanos: Int64, + size: Int64, + excerpts: [MessageExcerpt]? + ) { + self.summary = summary + self.pathHash = pathHash + self.path = path + self.provider = provider + self.mtimeNanos = mtimeNanos + self.size = size + self.excerpts = excerpts + } + } + + /// Commits a bounded group of sessions as one FTS transaction. Building a + /// large index one session per transaction creates thousands of tiny FTS5 + /// segments and repeatedly merges them; batching keeps both segment count + /// and merge working sets bounded without dropping a single excerpt. + public func applyIndexBatch(_ entries: [IndexBatchEntry]) throws { + guard !entries.isEmpty else { return } + try execute("BEGIN IMMEDIATE") + do { + for entry in entries { + let row = try upsertSession(entry.summary) + if let excerpts = entry.excerpts { + try replaceMessagesUncommitted(sessionRow: row, excerpts: excerpts) + } + try saveFileCursor( + pathHash: entry.pathHash, + path: entry.path, + provider: entry.provider, + mtimeNanos: entry.mtimeNanos, + size: entry.size, + sessionRow: row + ) + } + try execute("COMMIT") + } catch { + try? execute("ROLLBACK") + throw error + } + } + /// Replace every indexed excerpt for one session. The delete trigger /// clears the old FTS rows, so this is also how a shrinking /// transcript stops matching its removed text. public func replaceMessages(sessionRow: Int64, excerpts: [MessageExcerpt]) throws { try execute("BEGIN IMMEDIATE") do { - try run("DELETE FROM session_messages WHERE session_row = ?", [.integer(sessionRow)]) - if !excerpts.isEmpty { - let statement = try prepare( - "INSERT INTO session_messages(session_row, seq, role, excerpt) VALUES(?, ?, ?, ?)" - ) - defer { sqlite3_finalize(statement) } - for excerpt in excerpts { - sqlite3_reset(statement) - sqlite3_clear_bindings(statement) - bindAll([ - .integer(sessionRow), - .integer(Int64(excerpt.seq)), - .text(excerpt.role.rawValue), - .text(excerpt.excerpt) - ], to: statement) - guard sqlite3_step(statement) == SQLITE_DONE else { - throw SessionIndexError.statement - } - } - } + try replaceMessagesUncommitted(sessionRow: sessionRow, excerpts: excerpts) try execute("COMMIT") } catch { try? execute("ROLLBACK") @@ -318,6 +365,45 @@ public actor SessionIndexStore { } } + private func replaceMessagesUncommitted( + sessionRow: Int64, + excerpts: [MessageExcerpt] + ) throws { + try run("DELETE FROM session_messages WHERE session_row = ?", [.integer(sessionRow)]) + guard !excerpts.isEmpty else { return } + let statement = try prepare( + "INSERT INTO session_messages(session_row, seq, role, excerpt) VALUES(?, ?, ?, ?)" + ) + defer { sqlite3_finalize(statement) } + for excerpt in excerpts { + sqlite3_reset(statement) + sqlite3_clear_bindings(statement) + bindAll([ + .integer(sessionRow), + .integer(Int64(excerpt.seq)), + .text(excerpt.role.rawValue), + .text(excerpt.excerpt) + ], to: statement) + guard sqlite3_step(statement) == SQLITE_DONE else { + throw SessionIndexError.statement + } + } + } + + /// Asks SQLite to release connection-local caches after each bulk batch. + /// The host additionally applies malloc-zone pressure relief, so neither + /// layer retains a rebuild-sized high-water mark. + public func releaseTransientMemory() { + guard let database else { return } + _ = sqlite3_db_release_memory(database) + var logFrames: Int32 = 0 + var checkpointedFrames: Int32 = 0 + _ = sqlite3_wal_checkpoint_v2( + database, nil, SQLITE_CHECKPOINT_PASSIVE, + &logFrames, &checkpointedFrames + ) + } + /// Fingerprint of a file as of its last successful index pass. public struct FileCursor: Hashable, Sendable { public let mtimeNanos: Int64 diff --git a/Tests/AgentSessionKitTests/SessionIndexStoreTests.swift b/Tests/AgentSessionKitTests/SessionIndexStoreTests.swift index 99f1566..731a7d3 100644 --- a/Tests/AgentSessionKitTests/SessionIndexStoreTests.swift +++ b/Tests/AgentSessionKitTests/SessionIndexStoreTests.swift @@ -97,6 +97,45 @@ final class SessionIndexStoreTests: XCTestCase { XCTAssertEqual(count, 2) } + func testBoundedIndexBatchPreservesSessionsBodiesAndCursors() async throws { + let store = try makeStore() + let first = summary(id: "first", path: "/first.jsonl") + let second = summary(id: "second", path: "/second.jsonl") + try await store.applyIndexBatch([ + .init( + summary: first, + pathHash: "hash-first", + path: first.sourcePath, + provider: first.provider, + mtimeNanos: 11, + size: 22, + excerpts: [.init(seq: 0, role: .user, excerpt: "first complete body")] + ), + .init( + summary: second, + pathHash: "hash-second", + path: second.sourcePath, + provider: second.provider, + mtimeNanos: 33, + size: 44, + excerpts: [.init(seq: 0, role: .tool, excerpt: "second file operation")] + ) + ]) + + let sessionCount = try await store.sessionCount() + let messageCount = try await store.messageCount() + let firstHits = try await store.search(text: "complete body", scopes: [.user]) + let secondHits = try await store.search(text: "file operation", scopes: [.tool]) + XCTAssertEqual(sessionCount, 2) + XCTAssertEqual(messageCount, 2) + XCTAssertEqual(firstHits.map(\.summary.sessionID), ["first"]) + XCTAssertEqual(secondHits.map(\.summary.sessionID), ["second"]) + let cursor = try await store.fileCursor(pathHash: "hash-second") + XCTAssertEqual(cursor?.mtimeNanos, 33) + XCTAssertEqual(cursor?.size, 44) + XCTAssertNotNil(cursor?.sessionRow) + } + func testSummariesComeBackMostRecentlyActiveFirst() async throws { let store = try makeStore() try await store.upsertSession(summary( @@ -618,9 +657,9 @@ final class SessionIndexStoreTests: XCTestCase { let rebuilt = try makeStore() let remaining = try await rebuilt.sessionCount() - XCTAssertEqual(SessionIndexStore.schemaVersion, 4) + XCTAssertEqual(SessionIndexStore.schemaVersion, 5) XCTAssertEqual(remaining, 0) - XCTAssertEqual(try userVersion(), 4) + XCTAssertEqual(try userVersion(), 5) try await rebuilt.upsertSession(summary(harness: .claudeCode, model: "claude-fable-5")) let model = try await rebuilt.allSummaries().first?.model