From bdf3aa1d0ee5f83ba369cbc1090ca92d252a2c81 Mon Sep 17 00:00:00 2001 From: orionsutton Date: Wed, 12 Aug 2026 23:08:51 +0800 Subject: [PATCH 1/2] feat(server): spill oversized TUIndex results to a store tmp file --- src/server/compiler/indexer.cpp | 48 ++++++++++-- src/server/protocol/worker.h | 18 ++++- src/server/state/config.h | 8 ++ src/server/worker/stateless_worker.cpp | 15 ++++ src/support/filesystem.h | 8 ++ tests/integration/server/index_spill.test.ts | 67 ++++++++++++++++ tests/unit/server/stateless_worker_tests.cpp | 81 ++++++++++++++++++++ 7 files changed, 237 insertions(+), 8 deletions(-) create mode 100644 tests/integration/server/index_spill.test.ts diff --git a/src/server/compiler/indexer.cpp b/src/server/compiler/indexer.cpp index 4d90d8226..08a623788 100644 --- a/src/server/compiler/indexer.cpp +++ b/src/server/compiler/indexer.cpp @@ -3,6 +3,7 @@ #include #include #include +#include #include #include #include @@ -532,12 +533,48 @@ kota::task<> Indexer::index_one(std::uint32_t server_path_id, co_return; workspace.fill_pcm_deps(params.pcms); + params.index_inline_limit = workspace.config.project.index_inline_limit.value; + + // Deliberately never committed: the PendingEntry removes the transfer + // file on every master-side exit path, and a write orphaned by + // cancellation is swept with the instance tmp dir. + std::optional transfer; + if(workspace.store) { + transfer = workspace.store->begin_store("index", shard_key(file_path)); + params.index_output_path = transfer->tmp_path; + } LOG_INFO("[{}/{}] Indexing {}", index, total, file_path); ScopedTimer timer; auto result = co_await pool.send_stateless(params); - if(result.has_value() && result.value().success && !result.value().tu_index_data.empty()) { + + llvm::StringRef tu_index_bytes; + std::unique_ptr spilled; + if(result.has_value() && result.value().success) { + tu_index_bytes = result.value().tu_index_data; + if(tu_index_bytes.empty() && result.value().tu_index_file_size > 0 && transfer) { + auto buffer = llvm::MemoryBuffer::getFile(transfer->tmp_path); + if(!buffer || (*buffer)->getBufferSize() != result.value().tu_index_file_size || + llvm::xxh3_64bits((*buffer)->getBuffer()) != result.value().tu_index_hash) { + // A transfer loss is a transport failure, not a build + // verdict: the work itself is fine (a worker died + // mid-write, or the cache dir was wiped mid-flight), so + // requeue like a crash instead of serving the stale shard + // as fresh until the file's next edit. + LOG_WARN("[{}/{}] Requeueing {}: corrupt TUIndex transfer", + index, + total, + file_path); + note_dispatch_failure(server_path_id, ticket, false); + co_return; + } + spilled = std::move(*buffer); + tu_index_bytes = spilled->getBuffer(); + } + } + + if(result.has_value() && result.value().success && !tu_index_bytes.empty()) { auto index_ms = timer.ms(); // Merge guard: a newer content-level invalidation during this build // (or a removal clearing the entry) means this result describes text @@ -553,18 +590,19 @@ kota::task<> Indexer::index_one(std::uint32_t server_path_id, co_return; } ScopedTimer merge_timer; - merge(result.value().tu_index_data.data(), result.value().tu_index_data.size()); + merge(tu_index_bytes.data(), tu_index_bytes.size()); LOG_PERF("index", - "progress={}/{} file={} bytes={} index_ms={} merge_ms={}", + "progress={}/{} file={} bytes={} transfer={} index_ms={} merge_ms={}", index, total, file_path, - result.value().tu_index_data.size(), + tu_index_bytes.size(), + spilled ? "file" : "inline", index_ms, merge_timer.ms()); } else if(result.has_value() && !result.value().success) { LOG_WARN("[{}/{}] Index failed for {}: {}", index, total, file_path, result.value().error); - } else if(result.has_value() && result.value().tu_index_data.empty()) { + } else if(result.has_value()) { LOG_WARN("[{}/{}] Index returned empty TUIndex for {}", index, total, file_path); } else if(result.error().code == worker::dispatch_errc::cancelled || result.error().code == worker::dispatch_errc::worker_crashed || diff --git a/src/server/protocol/worker.h b/src/server/protocol/worker.h index 321774a07..012a36920 100644 --- a/src/server/protocol/worker.h +++ b/src/server/protocol/worker.h @@ -185,9 +185,9 @@ enum class BuildKind : uint8_t { /// Unified parameters for all stateless build/compilation tasks. /// Fields are used selectively based on `kind`: /// - All: file, directory, arguments -/// - BuildPCH: + content, preamble_bound, output_path +/// - BuildPCH: + content, preamble_bound, output_path, index_output_path /// - BuildPCM: + module_name, pcms, output_path -/// - Index: + pcms +/// - Index: + pcms, index_output_path, index_inline_limit /// - Completion: + text, version, offset, pch, pcms /// - SignatureHelp: + text, version, offset, pch, pcms /// - Format: + text, format_range (optional) @@ -214,8 +214,14 @@ struct BuildParams { /// `.pch.idx`), allocated by the master's store alongside output_path. /// The worker serializes the preamble's index and feature state into /// it; the master commits both blobs together. + /// Index: tmp path for a serialized TUIndex too large to travel inline + /// in the IPC frame (see BuildResult::tu_index_file_size). std::string index_output_path; + /// Index: serialized results above this many bytes spill to + /// index_output_path instead of travelling inline. + std::uint64_t index_inline_limit = 8 * 1024 * 1024; + std::string module_name; ///< BuildPCM uint32_t preamble_bound = UINT32_MAX; ///< BuildPCH LocalSourceRange format_range; ///< Format (default = full document) @@ -242,7 +248,13 @@ struct BuildResult { /// whose mtime is past this moment may differ from what the build read. std::int64_t build_at = 0; std::vector deps; - std::string tu_index_data; ///< Index: serialized TUIndex, merged by the master + std::string tu_index_data; ///< Index: serialized TUIndex, merged by the master + /// Index: set instead of tu_index_data when the serialized TUIndex was + /// spilled to BuildParams::index_output_path (too large for the IPC + /// frame): blob byte size and its xxh3_64, checked by the master before + /// use — the file may be torn if the worker died mid-write. + std::uint64_t tu_index_file_size = 0; + std::uint64_t tu_index_hash = 0; kota::codec::RawValue result_json; ///< Completion/SignatureHelp result }; diff --git a/src/server/state/config.h b/src/server/state/config.h index 275e53982..965d6c65b 100644 --- a/src/server/state/config.h +++ b/src/server/state/config.h @@ -76,6 +76,14 @@ struct ProjectConfig { "harness.") test_hooks = false; + /// Test-facing, like test_hooks: the default only matters for results + /// large enough that CI cannot produce them. + KOTATSU_ANNOTATE(defaulted = true, + description = + "Serialized TUIndex results above this many bytes are " + "spilled to a tmp file instead of sent inline.") + index_inline_limit = 8 * 1024 * 1024; + KOTATSU_ANNOTATE(defaulted = true, description = "Number of stateful workers.") stateful_worker_count = 2; diff --git a/src/server/worker/stateless_worker.cpp b/src/server/worker/stateless_worker.cpp index 032d2fd48..6ab8e6600 100644 --- a/src/server/worker/stateless_worker.cpp +++ b/src/server/worker/stateless_worker.cpp @@ -19,6 +19,7 @@ #include "kota/ipc/peer.h" #include "kota/ipc/transport.h" #include "llvm/Support/raw_ostream.h" +#include "llvm/Support/xxhash.h" namespace clice { @@ -278,6 +279,20 @@ static worker::BuildResult handle_index(const worker::BuildParams& params, timer.ms()); worker::BuildResult result; result.success = true; + + // The master verifies size+hash before use, so a torn write is detected + // there and needs no atomic-rename dance here. + if(serialized.size() > params.index_inline_limit && !params.index_output_path.empty()) { + auto written = fs::write(params.index_output_path, serialized); + if(written) { + result.tu_index_file_size = serialized.size(); + result.tu_index_hash = llvm::xxh3_64bits(serialized); + return result; + } + LOG_WARN("Index spill to {} failed ({}), sending inline", + params.index_output_path, + written.error().message()); + } result.tu_index_data = std::move(serialized); return result; } diff --git a/src/support/filesystem.h b/src/support/filesystem.h index 0632facb0..560bf92d7 100644 --- a/src/support/filesystem.h +++ b/src/support/filesystem.h @@ -125,6 +125,14 @@ inline std::expected write(llvm::StringRef path, llvm::St } os << content; os.flush(); + // A mid-write failure (disk full, EIO) lands in the stream's error + // flag, and an unchecked flag turns into a fatal error in the + // destructor — surface it as an ordinary error instead. + if(os.has_error()) { + EC = os.error(); + os.clear_error(); + return std::unexpected(EC); + } return std::expected(); } diff --git a/tests/integration/server/index_spill.test.ts b/tests/integration/server/index_spill.test.ts new file mode 100644 index 000000000..f269cce30 --- /dev/null +++ b/tests/integration/server/index_spill.test.ts @@ -0,0 +1,67 @@ +/// Oversized TUIndex results travel via a store tmp file instead of the +/// IPC frame. With a tiny inline limit every background-index result +/// spills; cross-file references prove the master verified and merged the +/// spilled bytes, and the transfer files must not outlive their requests. + +import * as fs from "node:fs"; +import * as path from "node:path"; + +import { sleep, type CliceClient } from "@clice/tools/client"; +import type { StatsResult } from "@clice/tools/protocol"; +import { expect, test } from "../fixtures.ts"; + +/// Poll clice/internal/stats until predicate(stats) holds. +async function waitStats( + client: CliceClient, + predicate: (stats: StatsResult) => boolean, + message = "", +): Promise { + const deadline = Date.now() + 30_000; + for (;;) { + const stats = await client.stats(); + if (predicate(stats)) { + return stats; + } + if (Date.now() > deadline) { + throw new Error(`${message || "stats condition"} not met: ${JSON.stringify(stats)}`); + } + await sleep(200); + } +} + +function masterLog(logsDir: string): string { + if (!fs.existsSync(logsDir)) { + return ""; + } + return fs + .readdirSync(logsDir, { recursive: true, encoding: "utf8" }) + .filter((name) => path.basename(name) === "master.log") + .map((name) => fs.readFileSync(path.join(logsDir, name), "utf8")) + .join("\n"); +} + +test("index spill roundtrip", async ({ session }) => { + const { client, workspace } = session.tmp(); + // The reference probe excludes declarations, so lib.cpp needs a real + // call site for it to show up in the reference list. + workspace.write("lib.cpp", "int shared_fn() { return 42; }\nint call_it() { return shared_fn(); }\n"); + workspace.write("main.cpp", "int shared_fn();\nint main() { return shared_fn(); }\n"); + workspace.writeCDB(["lib.cpp", "main.cpp"]); + workspace.pinCacheDir(); + await client.initialize(workspace, { + initializationOptions: { + project: { idle_timeout_ms: 0, index_inline_limit: 1 }, + }, + }); + + const [uri] = await client.openAndWait("main.cpp"); + expect( + await client.waitForReference(uri, 0, 4, workspace.uri("lib.cpp")), + "cross-file reference through spilled index", + ).toBe(true); + + expect(masterLog(workspace.path(".clice/logs"))).toContain("transfer=file"); + + await waitStats(client, (s) => s.pendingTmpFiles === 0, "transfer files leaked"); + expect(workspace.tmpFiles(), "tmp directory should be empty after settling").toEqual([]); +}); diff --git a/tests/unit/server/stateless_worker_tests.cpp b/tests/unit/server/stateless_worker_tests.cpp index 9f75d2280..faf9c37a6 100644 --- a/tests/unit/server/stateless_worker_tests.cpp +++ b/tests/unit/server/stateless_worker_tests.cpp @@ -4,8 +4,10 @@ #include "test/test.h" #include "server/protocol/worker.h" #include "server/worker_test_helpers.h" +#include "support/filesystem.h" #include "kota/codec/bincode/bincode.h" +#include "llvm/Support/xxhash.h" namespace clice::testing { @@ -133,9 +135,88 @@ TEST_CASE(IndexRequest) { params.file = src; params.directory = "/tmp"; params.arguments = make_args(src); + params.index_output_path = tmp.path("spill.bin"); auto result = co_await w.peer->send_request(params); EXPECT_TRUE(result.has_value()); + if(result.has_value()) { + // A small index stays inline even with a spill path offered. + EXPECT_TRUE(result->success); + EXPECT_FALSE(result->tu_index_data.empty()); + EXPECT_EQ(result->tu_index_file_size, std::uint64_t(0)); + } + test_done = true; + w.peer->close_output(); + }); + + ASSERT_TRUE(test_done); +} + +TEST_CASE(IndexSpillToFile) { + TempDir tmp; + tmp.touch("test_index.cpp", "int indexed_var = 1;\n"); + auto src = tmp.path("test_index.cpp"); + auto spill = tmp.path("spill.bin"); + + WorkerHandle w; + ASSERT_TRUE(w.spawn()); + + bool test_done = false; + + w.run([&]() -> kota::task<> { + worker::BuildParams params; + params.kind = worker::BuildKind::Index; + params.file = src; + params.directory = "/tmp"; + params.arguments = make_args(src); + params.index_output_path = spill; + params.index_inline_limit = 1; + + auto result = co_await w.peer->send_request(params); + EXPECT_TRUE(result.has_value()); + if(result.has_value()) { + EXPECT_TRUE(result->success); + EXPECT_TRUE(result->tu_index_data.empty()); + auto blob = fs::read(spill); + EXPECT_TRUE(blob.has_value()); + if(blob.has_value()) { + EXPECT_EQ(result->tu_index_file_size, blob->size()); + EXPECT_EQ(result->tu_index_hash, llvm::xxh3_64bits(*blob)); + } + } + test_done = true; + w.peer->close_output(); + }); + + ASSERT_TRUE(test_done); +} + +TEST_CASE(SpillFailureFallsInline) { + TempDir tmp; + tmp.touch("test_index.cpp", "int indexed_var = 1;\n"); + auto src = tmp.path("test_index.cpp"); + + WorkerHandle w; + ASSERT_TRUE(w.spawn()); + + bool test_done = false; + + w.run([&]() -> kota::task<> { + worker::BuildParams params; + params.kind = worker::BuildKind::Index; + params.file = src; + params.directory = "/tmp"; + params.arguments = make_args(src); + params.index_output_path = tmp.path("no_such_dir/spill.bin"); + params.index_inline_limit = 1; + + auto result = co_await w.peer->send_request(params); + EXPECT_TRUE(result.has_value()); + if(result.has_value()) { + EXPECT_TRUE(result->success); + EXPECT_FALSE(result->tu_index_data.empty()); + EXPECT_EQ(result->tu_index_file_size, std::uint64_t(0)); + } test_done = true; w.peer->close_output(); }); From ed18493e1d4f928f39be6e83b4e010fa0b74a83e Mon Sep 17 00:00:00 2001 From: orionsutton Date: Wed, 12 Aug 2026 23:51:12 +0800 Subject: [PATCH 2/2] style(tests): format index spill test --- tests/integration/server/index_spill.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/integration/server/index_spill.test.ts b/tests/integration/server/index_spill.test.ts index f269cce30..33fe54e97 100644 --- a/tests/integration/server/index_spill.test.ts +++ b/tests/integration/server/index_spill.test.ts @@ -44,7 +44,10 @@ test("index spill roundtrip", async ({ session }) => { const { client, workspace } = session.tmp(); // The reference probe excludes declarations, so lib.cpp needs a real // call site for it to show up in the reference list. - workspace.write("lib.cpp", "int shared_fn() { return 42; }\nint call_it() { return shared_fn(); }\n"); + workspace.write( + "lib.cpp", + "int shared_fn() { return 42; }\nint call_it() { return shared_fn(); }\n", + ); workspace.write("main.cpp", "int shared_fn();\nint main() { return shared_fn(); }\n"); workspace.writeCDB(["lib.cpp", "main.cpp"]); workspace.pinCacheDir();