diff --git a/DEVELOPER_GUIDE.md b/DEVELOPER_GUIDE.md index 693a24c7f..8e1ad352e 100644 --- a/DEVELOPER_GUIDE.md +++ b/DEVELOPER_GUIDE.md @@ -412,7 +412,7 @@ Interactive terminal controls are allowed only when stdout is not redirected or Query commands that accept path filters (`search`, `definition`, `references`, `callers`, `callees`, `symbols`, `files`, `find`, `map`, `inspect`, `deps`, `impact`, `unused`, `hotspots`, and `validate`) expand `--project` into the matching project directory glob before hitting `DbReader`, so all existing SQL path predicates keep working. When the indexed project root cannot be resolved and project expansion falls back to the process current directory, CLI query context and MCP structured payloads include `project_filter_root` and `project_filter_root_fallback_reason`. `index --project` expands to the files under the selected project directory and reuses the existing `--files` update path, but rejects expansions above 65,536 files for one project or 131,072 unique files across all requested projects with an explicit-files recovery hint. -`cdidx batch` is a CLI-side query loop for editor integrations and scripts that need several query commands against the same DB without spawning `cdidx` repeatedly. Each newline-delimited stdin record may use the established JSON string-array form or the validated `{"command": "...", "args": [...]}` object form. Object input rejects duplicate/unknown properties, missing or blank commands, non-array `args`, non-string values, and the same argument count/length violations as array input. Serial mode opens one `DbContext` / `DbReader`; `--parallel ` requires `--json-summary`, is capped at 16 workers, and opens one isolated query-only context per active worker. Every form dispatches only commands in the side-effect-free allowlist owned by `CliCommandCatalog`. That schema includes query and read-only discovery surfaces such as `goto` and `audit`; adding a top-level command or a dispatcher arm alone cannot cross the batch safety boundary. +`cdidx batch` is a CLI-side query loop for editor integrations and scripts that need several query commands against the same DB without spawning `cdidx` repeatedly. Each newline-delimited stdin record may use the established JSON string-array form or the validated `{"command": "...", "args": [...]}` object form. Object input rejects duplicate/unknown properties, missing or blank commands, non-array `args`, non-string values, and the same argument count/length violations as array input. Serial mode opens one `DbContext` / `DbReader`; `--parallel ` requires `--json-summary`, is capped at 16 workers, and lazily retains at most one isolated query-only context per active worker slot for reuse across that batch invocation. A retained context is leased to only one command at a time, so SQLite readers are never used concurrently while repeated database open, schema probe, and reader setup costs are avoided. Between items, each retained direct connection or detached snapshot verifies the source SQLite header/WAL generation and file identity, replacing the session whenever freshness cannot be proved so long-lived batches observe later completed index updates. Every form dispatches only commands in the side-effect-free allowlist owned by `CliCommandCatalog`. That schema includes query and read-only discovery surfaces such as `goto` and `audit`; adding a top-level command or a dispatcher arm alone cannot cross the batch safety boundary. The default input budget remains 1,024 lines and is configurable through `--max-input-lines ` up to 65,536. Each decoded string argument remains capped at 8,192 characters. The JSON-summary output budget defaults to 10,485,760 characters and `--max-output-chars ` accepts 4,096 through 67,108,864. Immediate EOF with no commands remains exit 0 with no output by default; `--json-summary` appends a final JSON object with `commands_processed`, `line_errors`, `command_failures`, and `exit_code` for non-interactive callers that need an explicit empty-input signal. By default, child query commands stream their normal stdout/stderr directly. In @@ -438,9 +438,9 @@ final `record: "batch_summary"` retains `commands_processed`, `line_errors`, `command_failures`, and `exit_code`, and publishes `output_chars`, `output_char_limit`, `input_line_limit`, `parallelism`, and input/output limit state for empty-input, failure, and budget accounting. Parallel workers route -stdout/stderr through per-command bounded writers, keep a separate read-only -SQLite connection and thread-local batch reader, and buffer only the active -worker window. `ScopedConsoleOutput` keeps nested JSON-envelope capture on the +stdout/stderr through per-command bounded writers, reuse a separate read-only +SQLite connection and thread-local batch reader per active worker slot, and +buffer only the active worker window. `ScopedConsoleOutput` keeps nested JSON-envelope capture on the current worker's routed stdout instead of replacing another worker's process-wide writer. Completed records are committed to the shared output writer in input order; an ordinary item failure remains isolated. Caller cancellation is @@ -3776,7 +3776,7 @@ override が文書化されていない限り ANSI/progress control を抑止す path filter を受け付ける query コマンド（`search`, `definition`, `references`, `callers`, `callees`, `symbols`, `files`, `find`, `map`, `inspect`, `deps`, `impact`, `unused`, `hotspots`, `validate`）は、`--project` を対応する project directory glob に展開してから `DbReader` に渡す。これにより既存の SQL path predicate をそのまま利用できる。indexed project root を解決できず process current directory に fallback して project expansion する場合、CLI query context と MCP structured payload は `project_filter_root` と `project_filter_root_fallback_reason` を含める。`index --project` は選択された project directory 配下のファイルに展開し、既存の `--files` 更新経路を再利用する。ただし 1 project で 65,536 files、requested projects 全体で 131,072 unique files を超える展開は拒否し、明示的な `--files` を使う recovery hint を返す。 -`cdidx batch` は、同じ DB に複数の query command を投げる editor integration や script 向けの CLI 側 query loop である。newline-delimited な stdin record は従来の JSON 文字列配列 form、または検証済みの `{"command": "...", "args": [...]}` object form を使用できる。object input は重複/未知 property、欠落または空白 command、array でない `args`、文字列でない値、array input と同じ引数数/長さ違反を拒否する。serial mode は 1 つの `DbContext` / `DbReader` を開く。`--parallel ` は `--json-summary` を必須とし、最大 16 workers に制限し、active worker ごとに分離した query-only context を開く。すべての form は `CliCommandCatalog` が正本となる副作用なし allowlist の command だけを dispatch する。この schema には `goto` や `audit` などの query / read-only discovery surface が含まれ、top-level command や dispatcher arm を追加しただけでは batch の安全境界を越えられない。 +`cdidx batch` は、同じ DB に複数の query command を投げる editor integration や script 向けの CLI 側 query loop である。newline-delimited な stdin record は従来の JSON 文字列配列 form、または検証済みの `{"command": "...", "args": [...]}` object form を使用できる。object input は重複/未知 property、欠落または空白 command、array でない `args`、文字列でない値、array input と同じ引数数/長さ違反を拒否する。serial mode は 1 つの `DbContext` / `DbReader` を開く。`--parallel ` は `--json-summary` を必須とし、最大 16 workers に制限する。parallel mode は active worker slot ごとに分離した query-only context を最大 1 つまで遅延作成し、同じ batch invocation 内で再利用する。保持した context は一度に 1 command だけへ貸し出すため、SQLite reader を並行利用せず、database open、schema probe、reader setup の反復コストを避ける。各 item の間では、保持中の direct connection と分離 snapshot の両方で source SQLite header / WAL generation と file identity を検証し、freshness を証明できない場合は session を置き換えるため、長時間動作する batch も後から完了した index update を観測できる。すべての form は `CliCommandCatalog` が正本となる副作用なし allowlist の command だけを dispatch する。この schema には `goto` や `audit` などの query / read-only discovery surface が含まれ、top-level command や dispatcher arm を追加しただけでは batch の安全境界を越えられない。 既定の入力 budget は 1,024 行のままで、`--max-input-lines ` により最大 65,536 まで設定できる。デコード後の各文字列引数は引き続き 8,192 文字に制限する。JSON-summary 出力 budget は既定で 10,485,760 文字であり、`--max-output-chars ` は 4,096 から 67,108,864 までを受け付ける。command がない即時 EOF は既定で exit 0 かつ無出力のまま維持される。非対話の呼び出し元が空入力を明示的に判定したい場合は、`--json-summary` が `commands_processed`、`line_errors`、`command_failures`、`exit_code` を含む最終 JSON オブジェクトを追加する。 既定では child query command の通常の stdout / stderr を直接 stream する。`--json-summary` @@ -3798,8 +3798,8 @@ arguments、escape 展開、terminal error、final summary を含む serialized empty input、failure、budget accounting のために `commands_processed`、`line_errors`、 `command_failures`、`exit_code`、`output_chars`、`output_char_limit`、`input_line_limit`、 `parallelism` と input / output limit state を保持する。parallel worker は stdout / stderr を -command ごとの bounded writer へ route し、分離した read-only SQLite connection と thread-local -batch reader を使い、active worker window だけを buffer する。`ScopedConsoleOutput` は nested +command ごとの bounded writer へ route し、active worker slot ごとに分離した read-only SQLite +connection と thread-local batch reader を再利用し、active worker window だけを buffer する。`ScopedConsoleOutput` は nested JSON-envelope capture を現在の worker の routed stdout に保ち、他 worker の process-wide writer を 置き換えない。完了 record は入力順で共有 output writer へ commit する。通常の item failure は 他 item から隔離する。caller cancellation は、消費済み input item と final summary に diff --git a/TESTING_GUIDE.md b/TESTING_GUIDE.md index 74c1f2af2..334374973 100644 --- a/TESTING_GUIDE.md +++ b/TESTING_GUIDE.md @@ -664,6 +664,8 @@ Use `docs/test-doc-maintenance-plan.md` before moving oversized suites or adding Golden-file regression fixtures for the CLI `--json` output contracts (issue #1548). Each test runs one command (`status`, `search`, `references`, `impact`, `excerpt`) against a deterministic in-memory fixture, normalizes volatile fields (timestamps, absolute paths, commit SHAs, FTS5 scores, SQLite page counts), and diffs against the matching file under `tests/CodeIndex.Tests/golden/`. Renames, removals, reordered arrays, or new keys fail the snapshot so the contract change is forced to land alongside an intentional golden update. See "JSON `--json` output snapshots" below for the update procedure. - `QueryCommandRunnerBatchIssue4723Tests.cs` CLI batch coverage for structured command objects, configurable input/output budgets, bounded parallel read overlap, input-order result emission, per-item failure isolation, and cancellation/console restoration. The deterministic overlap test blocks the first worker until the second finishes through batch-only test seams; keep those seams reset in `finally` and do not replace the signal with timing assertions. +- `QueryCommandRunnerBatchIssue4872Tests.cs` + Parallel batch session-reuse coverage. Keep the exact worker-slot session bound, serial/parallel result and input-order parity, between-item checkpointed-WAL snapshot and direct-session refresh, validation/reader-construction failure cleanup, hot-WAL snapshot fixture, warmup, and the generous 12-item/3-item ratio guard together. The benchmark injects deterministic database-open/schema work through a test seam; its fixed delay models phase cost only, while worker coordination remains signal-driven. - `PropertyBasedParserTests.cs` FsCheck-driven property tests for parser-heavy paths called out in issue #1572: `ArgHelper.WantsHelp` and `ProgramRunner.IsProjectPathArg` never throw on arbitrary inputs; `FileIndexer.NormalizePathSeparators` is idempotent under double application; the literal-safe FTS5 sanitizer (`DbReader.SanitizeFtsQuery`) always emits a query that a real in-memory FTS5 virtual table can parse. They complement, not replace, the example-based tests in `ArgHelperTests.cs` / `QueryCommandRunnerTests.cs`. - `TestProjectHelper.cs`, `TestDeterminism.cs`, `RepositoryTestPaths.cs`, `TestConsoleLock.cs` @@ -1580,6 +1582,8 @@ dotnet test --filter "FullyQualifiedName~GitHelperTests" CLI の `--json` 出力契約に対するゴールデンファイル回帰フィクスチャ (issue #1548)。各テストは `status` / `search` / `references` / `impact` / `excerpt` を決定的なインメモリ fixture に対して実行し、揺らぐフィールド（timestamp、絶対パス、commit SHA、FTS5 score、SQLite page count など）を正規化したうえで `tests/CodeIndex.Tests/golden/` 配下のファイルと差分比較します。フィールドの rename / 削除 / 並び替え / 新規追加が起きると snapshot が失敗するため、契約変更は意図的な golden 更新と同じ PR で揃えざるを得ません。更新手順は下記「JSON `--json` 出力 snapshot」を参照してください。 - `QueryCommandRunnerBatchIssue4723Tests.cs` structured command object、設定可能な input / output budget、上限付き parallel read の重複実行、入力順の result 出力、item ごとの failure isolation、cancellation / console 復元を対象とする CLI batch test です。決定的な overlap test は batch 専用 test seam を通じて第 1 worker を第 2 worker の完了まで block します。seam は `finally` で必ず reset し、signal を timing assertion に置き換えないでください。 +- `QueryCommandRunnerBatchIssue4872Tests.cs` + parallel batch の session 再利用を検証します。worker slot 数と一致する厳密な session 上限、serial / parallel の result と入力順 parity、item 間の checkpoint 済み WAL snapshot / direct session 更新、validation / reader 構築失敗時の cleanup、hot-WAL snapshot fixture、warmup、十分に余裕を持たせた 12-item / 3-item ratio guard を一緒に維持してください。benchmark は test seam から決定的な database-open / schema 作業を注入します。固定 delay は phase cost のモデル化だけに使い、worker coordination は引き続き signal で制御します。 - `PropertyBasedParserTests.cs` issue #1572 で挙げられたパーサー系経路に対する FsCheck 駆動の property テスト: `ArgHelper.WantsHelp` と `ProgramRunner.IsProjectPathArg` が任意入力で例外を投げないこと、`FileIndexer.NormalizePathSeparators` が二重適用で idempotent であること、literal-safe な FTS5 サニタイザ (`DbReader.SanitizeFtsQuery`) が常にインメモリ FTS5 仮想テーブルで parse 可能なクエリを出力すること。`ArgHelperTests.cs` / `QueryCommandRunnerTests.cs` の例ベーステストを置き換えるものではなく補完します。 - `TestProjectHelper.cs`、`TestDeterminism.cs`、`RepositoryTestPaths.cs`、`TestConsoleLock.cs` diff --git a/changelog.d/unreleased/4872.fixed.md b/changelog.d/unreleased/4872.fixed.md new file mode 100644 index 000000000..70fb48c7c --- /dev/null +++ b/changelog.d/unreleased/4872.fixed.md @@ -0,0 +1,20 @@ +--- +category: fixed +issues: + - 4872 +affected: + - src/CodeIndex/Cli/QueryCommandRunner.Batch.cs + - src/CodeIndex/Database/DbConnectionFactory.cs + - src/CodeIndex/Database/DbContext.cs + - tests/CodeIndex.Tests/QueryCommandRunnerBatchIssue4872Tests.cs + - DEVELOPER_GUIDE.md + - TESTING_GUIDE.md +--- + +## English + +- **Parallel batch queries now reuse bounded read-only sessions (#4872)** — `cdidx batch --parallel` retains at most one isolated SQLite context per active worker slot instead of reopening and reproving the database for every item, verifies source generation and file identity for direct connections and detached snapshots between items, and cleans up validation/reader failures while preserving input-order envelopes, cancellation, failure isolation, and the configured worker bound. + +## 日本語 + +- **parallel batch query が上限付き read-only session を再利用するようになりました (#4872)** — `cdidx batch --parallel` は item ごとに database を再open・再検証せず、active worker slot ごとに最大 1 つの分離した SQLite context を保持します。item 間で direct connection と分離 snapshot の source generation / file identity を検証し、validation / reader failure を cleanup しながら、入力順 envelope、cancellation、failure isolation、設定された worker 上限を維持します。 diff --git a/src/CodeIndex/Cli/QueryCommandRunner.Batch.cs b/src/CodeIndex/Cli/QueryCommandRunner.Batch.cs index 796976d44..504be07b8 100644 --- a/src/CodeIndex/Cli/QueryCommandRunner.Batch.cs +++ b/src/CodeIndex/Cli/QueryCommandRunner.Batch.cs @@ -17,6 +17,9 @@ public static partial class QueryCommandRunner internal static Action? BatchParallelCommandCompletedForTesting { get; set; } internal static Action? BatchInputLineReadForTesting { get; set; } internal static Action? BatchParallelItemPreparedForTesting { get; set; } + internal static Action? BatchParallelSessionOpenedForTesting { get; set; } + internal static Func? BatchParallelReaderFactoryForTesting { get; set; } + internal static Action? BatchParallelDatabaseValidatingForTesting { get; set; } public static int RunBatch( string[] cmdArgs, @@ -156,31 +159,59 @@ public static int RunBatch( if (parallelism > 1) { + BatchParallelSession? firstSession = null; + DbContext? validationDb = null; try { - using var validationDb = new DbContext(DbOpenIntent.QueryOnly, dbPath, cancellationToken); + validationDb = new DbContext(DbOpenIntent.QueryOnly, dbPath, cancellationToken); + BatchParallelDatabaseValidatingForTesting?.Invoke(); if (!validationDb.TryValidateIsCodeIndexDb(out var validationReason)) return WriteInvalidCodeIndexDbError(dbPath, validationReason, json: false, jsonOptions); + var transferredDb = validationDb; + validationDb = null; + firstSession = BatchParallelSession.FromValidated( + dbPath, + transferredDb, + cancellationToken); } catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { + firstSession?.Dispose(); return WriteBatchSetupCancellationSummary( maxInputLines, maxOutputChars, parallelism, jsonOptions); } + catch + { + firstSession?.Dispose(); + throw; + } + finally + { + validationDb?.Dispose(); + } - return RunBatchParallel( - dbPath, - dbPathExplicit, - maxInputLines, - maxOutputChars, - parallelism, - includeRawStreams, - jsonOptions, - appVersion, - cancellationToken); + try + { + return RunBatchParallel( + dbPath, + dbPathExplicit, + maxInputLines, + maxOutputChars, + parallelism, + includeRawStreams, + jsonOptions, + appVersion, + firstSession!, + cancellationToken); + } + catch + { + firstSession?.Dispose(); + throw; + } } try @@ -454,8 +485,14 @@ private static int RunBatchParallel( bool includeRawStreams, JsonSerializerOptions jsonOptions, string appVersion, + BatchParallelSession firstSession, CancellationToken cancellationToken) { + var sessions = new BatchParallelSession[parallelism]; + sessions[0] = firstSession; + for (var index = 1; index < sessions.Length; index++) + sessions[index] = new BatchParallelSession(dbPath); + var availableSessions = new Queue(sessions); using var consoleOwnership = ConsoleStreamOwnership.Enter(); var originalOut = Console.Out; var originalError = Console.Error; @@ -618,7 +655,8 @@ await input.Writer.WriteAsync(item, stopProducing.Token) }); var active = new Queue<( BatchPendingItem Item, - Task Result)>(); + Task Result, + BatchParallelSession? Session)>(); try { @@ -626,21 +664,31 @@ await input.Writer.WriteAsync(item, stopProducing.Token) { while (active.Count < parallelism && input.Reader.TryRead(out var item)) { - var result = item.Error is not null - ? Task.FromResult(null) - : Task.Run( + BatchParallelSession? session = null; + Task result; + if (item.Error is not null) + { + result = Task.FromResult(null); + } + else + { + session = availableSessions.Dequeue(); + var assignedSession = session; + result = Task.Run( () => RunBatchParallelCommand( item.LineNumber, item.CommandName!, item.Arguments, dbPath, dbPathExplicit, + assignedSession, stdoutRouter, stderrRouter, jsonOptions, appVersion, cancellationToken)); - active.Enqueue((item, result)); + } + active.Enqueue((item, result, session)); } if (active.Count > 0 @@ -648,8 +696,10 @@ await input.Writer.WriteAsync(item, stopProducing.Token) || active.Count == parallelism || input.Reader.Completion.IsCompleted)) { - var (item, resultTask) = active.Dequeue(); + var (item, resultTask, session) = active.Dequeue(); var result = resultTask.GetAwaiter().GetResult(); + if (session is not null) + availableSessions.Enqueue(session); if (item.Error is not null) { if (item.Error.ExitCode is CommandExitCodes.CancelledBySignal @@ -818,7 +868,15 @@ await input.Writer.WriteAsync(item, stopProducing.Token) } finally { - ConsoleStreamOwnership.Restore(originalOut, originalError); + try + { + foreach (var session in sessions) + session.Dispose(); + } + finally + { + ConsoleStreamOwnership.Restore(originalOut, originalError); + } } } @@ -828,6 +886,7 @@ private static BatchParallelCommandResult RunBatchParallelCommand( string[] subArgs, string dbPath, bool dbPathExplicit, + BatchParallelSession session, BatchConsoleRouter stdoutRouter, BatchConsoleRouter stderrRouter, JsonSerializerOptions jsonOptions, @@ -844,14 +903,13 @@ private static BatchParallelCommandResult RunBatchParallelCommand( try { cancellationToken.ThrowIfCancellationRequested(); - using var db = new DbContext(DbOpenIntent.QueryOnly, dbPath, cancellationToken); - if (!db.TryValidateIsCodeIndexDb(out var validationReason)) + if (!session.TryGetReader(cancellationToken, out var reader, out var validationReason)) { exitCode = WriteInvalidCodeIndexDbError(dbPath, validationReason, json: false, jsonOptions); } else { - s_batchReader = new DbReader(db); + s_batchReader = reader; s_batchDbPath = dbPath; s_batchDbPathExplicit = dbPathExplicit; BatchParallelCommandStartedForTesting?.Invoke(lineNumber); @@ -1942,6 +2000,159 @@ public void Dispose() } } + private sealed class BatchParallelSession : IDisposable + { + private readonly string _dbPath; + private DbContext? _db; + private DbReader? _reader; + private DbConnectionFactory.QueryOnlySnapshotSourceState? _sourceState; + private bool _readerUsed; + private bool _disposed; + + public BatchParallelSession(string dbPath) + { + _dbPath = dbPath; + } + + private BatchParallelSession( + string dbPath, + DbContext validatedDb, + DbReader reader, + DbConnectionFactory.QueryOnlySnapshotSourceState? sourceState) + { + _dbPath = dbPath; + _db = validatedDb; + _reader = reader; + _sourceState = sourceState; + } + + public static BatchParallelSession FromValidated( + string dbPath, + DbContext validatedDb, + CancellationToken cancellationToken) + { + DbReader? reader = null; + try + { + reader = CreateReader(validatedDb); + var sourceState = CaptureSourceState(dbPath, validatedDb, cancellationToken); + BatchParallelSessionOpenedForTesting?.Invoke(); + return new BatchParallelSession(dbPath, validatedDb, reader, sourceState); + } + catch + { + reader?.Dispose(); + validatedDb.Dispose(); + throw; + } + } + + public bool TryGetReader( + CancellationToken cancellationToken, + out DbReader reader, + out string? validationReason) + { + ObjectDisposedException.ThrowIf(_disposed, this); + cancellationToken.ThrowIfCancellationRequested(); + if (_reader is not null + && _db is not null + && (!_readerUsed + || (_sourceState is { } sourceState + && DbConnectionFactory.IsQuerySourceStateCurrent( + _dbPath, + sourceState, + cancellationToken)))) + { + _readerUsed = true; + reader = _reader; + validationReason = null; + return true; + } + + DbContext? replacementDb = null; + DbReader? replacementReader = null; + try + { + replacementDb = new DbContext( + DbOpenIntent.QueryOnly, + _dbPath, + cancellationToken); + if (!replacementDb.TryValidateIsCodeIndexDb(out validationReason)) + { + replacementDb.Dispose(); + replacementDb = null; + reader = null!; + return false; + } + + replacementReader = CreateReader(replacementDb); + var replacementSourceState = CaptureSourceState( + _dbPath, + replacementDb, + cancellationToken); + BatchParallelSessionOpenedForTesting?.Invoke(); + + var previousReader = _reader; + var previousDb = _db; + _reader = replacementReader; + _db = replacementDb; + _sourceState = replacementSourceState; + _readerUsed = true; + replacementReader = null; + replacementDb = null; + try + { + previousReader?.Dispose(); + } + finally + { + previousDb?.Dispose(); + } + + reader = _reader; + return true; + } + catch + { + replacementReader?.Dispose(); + replacementDb?.Dispose(); + throw; + } + } + + public void Dispose() + { + if (_disposed) + return; + _disposed = true; + try + { + _reader?.Dispose(); + _reader = null; + } + finally + { + _db?.Dispose(); + _db = null; + } + } + + private static DbConnectionFactory.QueryOnlySnapshotSourceState? CaptureSourceState( + string dbPath, + DbContext db, + CancellationToken cancellationToken) + => db.QueryOnlySnapshotSourceState + ?? (DbConnectionFactory.TryCaptureQuerySourceState( + dbPath, + cancellationToken, + out var sourceState) + ? sourceState + : null); + + private static DbReader CreateReader(DbContext db) + => BatchParallelReaderFactoryForTesting?.Invoke(db) ?? new DbReader(db); + } + private sealed class BatchJsonOutputWriter( TextWriter output, int maxChars, diff --git a/src/CodeIndex/Database/DbConnectionFactory.cs b/src/CodeIndex/Database/DbConnectionFactory.cs index ca828a39a..d24799ecf 100644 --- a/src/CodeIndex/Database/DbConnectionFactory.cs +++ b/src/CodeIndex/Database/DbConnectionFactory.cs @@ -316,6 +316,37 @@ internal static bool IsQueryOnlySnapshotCurrent( && currentState == snapshotSourceState; } + internal static bool TryCaptureQuerySourceState( + string dbPath, + CancellationToken cancellationToken, + out QueryOnlySnapshotSourceState state) + { + cancellationToken.ThrowIfCancellationRequested(); + var localDbPath = dbPath; + if (SqliteFileUri.StartsWithFileScheme(dbPath)) + { + if (!TryGetLocalPath(dbPath, out var parsedPath, out _) || parsedPath == null) + { + state = default; + return false; + } + localDbPath = parsedPath; + } + + return TryCaptureQueryOnlySnapshotState( + localDbPath, + cancellationToken, + out state, + requireWalMode: false); + } + + internal static bool IsQuerySourceStateCurrent( + string dbPath, + QueryOnlySnapshotSourceState sourceState, + CancellationToken cancellationToken = default) + => TryCaptureQuerySourceState(dbPath, cancellationToken, out var currentState) + && currentState == sourceState; + private static SqliteConnection CreateStableWalSnapshotConnection( string localDbPath, out bool copiedHotWal, @@ -365,7 +396,7 @@ private static SqliteConnection CreateStableWalSnapshotConnection( if (TryCaptureQueryOnlySnapshotState(snapshotDbPath, cancellationToken, out var copied) && TryCaptureQueryOnlySnapshotState(normalizedDbPath, cancellationToken, out var after) - && before == copied + && before.ContentEquals(copied) && before == after) { copiedHotWal = before.WalLength > 0; @@ -444,7 +475,8 @@ private static void CopySnapshotFile(string sourcePath, string destinationPath, private static bool TryCaptureQueryOnlySnapshotState( string localDbPath, CancellationToken cancellationToken, - out QueryOnlySnapshotSourceState state) + out QueryOnlySnapshotSourceState state, + bool requireWalMode = true) { try { @@ -465,8 +497,7 @@ private static bool TryCaptureQueryOnlySnapshotState( if (dbHeaderLength < 20 || !dbHeader[..16].SequenceEqual("SQLite format 3\0"u8) - || dbHeader[18] != 2 - || dbHeader[19] != 2) + || (requireWalMode && (dbHeader[18] != 2 || dbHeader[19] != 2))) { state = default; return false; @@ -513,12 +544,24 @@ private static bool TryCaptureQueryOnlySnapshotState( walLength = 0; } + var dbFile = TryReadSourceFileIdentity(normalizedDbPath); + var walFile = TryReadSourceFileIdentity(walPath); + if (dbFile is null + || dbFile.Value.Length != dbLength + || (walFile?.Length ?? 0) != walLength) + { + state = default; + return false; + } + state = new QueryOnlySnapshotSourceState( dbLength, Fingerprint(dbHeader[..dbHeaderLength]), walLength, walHeaderFingerprint, - walLastFrameFingerprint); + walLastFrameFingerprint, + dbFile.Value, + walFile); return true; } catch (Exception ex) when (ex is IOException or UnauthorizedAccessException or NotSupportedException) @@ -528,6 +571,25 @@ private static bool TryCaptureQueryOnlySnapshotState( } } + private static QuerySourceFileIdentity? TryReadSourceFileIdentity(string path) + { + try + { + var file = new FileInfo(path); + file.Refresh(); + return file.Exists + ? new QuerySourceFileIdentity( + file.Length, + file.CreationTimeUtc.Ticks, + file.LastWriteTimeUtc.Ticks) + : null; + } + catch (Exception ex) when (ex is IOException or UnauthorizedAccessException or NotSupportedException) + { + return null; + } + } + private static int ReadAtMost(Stream stream, Span destination, CancellationToken cancellationToken) { var total = 0; @@ -621,7 +683,22 @@ internal readonly record struct QueryOnlySnapshotSourceState( string DbHeaderFingerprint, long WalLength, string? WalHeaderFingerprint, - string? WalLastFrameFingerprint); + string? WalLastFrameFingerprint, + QuerySourceFileIdentity DatabaseFile, + QuerySourceFileIdentity? WalFile) + { + internal bool ContentEquals(QueryOnlySnapshotSourceState other) + => DbLength == other.DbLength + && DbHeaderFingerprint == other.DbHeaderFingerprint + && WalLength == other.WalLength + && WalHeaderFingerprint == other.WalHeaderFingerprint + && WalLastFrameFingerprint == other.WalLastFrameFingerprint; + } + + internal readonly record struct QuerySourceFileIdentity( + long Length, + long CreationTimeUtcTicks, + long LastWriteTimeUtcTicks); private sealed class QueryOnlySnapshotSourceChangedException(Exception innerException) : IOException("The query-only snapshot source disappeared while it was being copied.", innerException); diff --git a/src/CodeIndex/Database/DbContext.cs b/src/CodeIndex/Database/DbContext.cs index 0c0f41b9f..4726abb34 100644 --- a/src/CodeIndex/Database/DbContext.cs +++ b/src/CodeIndex/Database/DbContext.cs @@ -361,6 +361,8 @@ internal static Func? ReadMigrationTransact internal bool ImmutableReadOnlyWalRisk => _immutableReadOnlyWalRisk; internal bool ConnectionPooling => _connectionPooling; internal bool QueryOnlySnapshotRequiresRefresh => _queryOnlySnapshotRequiresRefresh; + internal DbConnectionFactory.QueryOnlySnapshotSourceState? QueryOnlySnapshotSourceState + => _queryOnlySnapshotSourceState; internal bool IsQueryOnlySnapshotCurrent(CancellationToken cancellationToken = default) => !_queryOnlySnapshotRequiresRefresh || (_queryOnlySnapshotSourcePath != null diff --git a/tests/CodeIndex.Tests/QueryCommandRunnerBatchIssue4872Tests.cs b/tests/CodeIndex.Tests/QueryCommandRunnerBatchIssue4872Tests.cs new file mode 100644 index 000000000..08ddae46c --- /dev/null +++ b/tests/CodeIndex.Tests/QueryCommandRunnerBatchIssue4872Tests.cs @@ -0,0 +1,531 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using CodeIndex.Cli; +using CodeIndex.Database; +using Microsoft.Data.Sqlite; + +namespace CodeIndex.Tests; + +public partial class QueryCommandRunnerTests +{ + [Fact] + public void RunBatch_ParallelReusesBoundedSessionsAndMatchesSerialRecords_Issue4872() + { + using var project = TestProjectHelper.CreateTempProjectScope("cdidx_batch_session_reuse_4872"); + var dbPath = TestProjectHelper.CreateProjectDb(project.Root); + var input = BuildIssue4872BatchInput(9); + var (serialExitCode, serialStdout, serialStderr) = CaptureConsoleWithInput( + input, + () => QueryCommandRunner.RunBatch( + ["--db", dbPath, "--json-summary"], + _jsonOptions)); + + using var firstWaveStarted = new CountdownEvent(3); + var openedSessions = 0; + QueryCommandRunner.BatchParallelSessionOpenedForTesting = + () => Interlocked.Increment(ref openedSessions); + QueryCommandRunner.BatchParallelCommandStartedForTesting = lineNumber => + { + if (lineNumber > 3) + return; + firstWaveStarted.Signal(); + if (!firstWaveStarted.Wait(TimeSpan.FromSeconds(5))) + throw new TimeoutException("The first parallel batch wave did not start."); + }; + + string parallelStdout; + string parallelStderr; + int parallelExitCode; + try + { + (parallelExitCode, parallelStdout, parallelStderr) = CaptureConsoleWithInput( + input, + () => QueryCommandRunner.RunBatch( + ["--db", dbPath, "--json-summary", "--parallel", "3"], + _jsonOptions)); + } + finally + { + QueryCommandRunner.BatchParallelSessionOpenedForTesting = null; + QueryCommandRunner.BatchParallelCommandStartedForTesting = null; + } + + Assert.Equal(CommandExitCodes.Success, serialExitCode); + Assert.Equal(CommandExitCodes.Success, parallelExitCode); + Assert.Equal(string.Empty, serialStderr); + Assert.Equal(string.Empty, parallelStderr); + Assert.Equal(3, Volatile.Read(ref openedSessions)); + + var serialLines = ParseJsonLines(serialStdout); + var parallelLines = ParseJsonLines(parallelStdout); + try + { + Assert.Equal(10, serialLines.Count); + Assert.Equal(serialLines.Count, parallelLines.Count); + for (var index = 0; index < 9; index++) + { + var serialRecord = serialLines[index].RootElement; + var parallelRecord = parallelLines[index].RootElement; + Assert.Equal(index + 1, parallelRecord.GetProperty("line").GetInt32()); + Assert.Equal( + serialRecord.GetProperty("command").GetString(), + parallelRecord.GetProperty("command").GetString()); + Assert.Equal( + serialRecord.GetProperty("arguments").GetRawText(), + parallelRecord.GetProperty("arguments").GetRawText()); + Assert.Equal( + serialRecord.GetProperty("exit_code").GetInt32(), + parallelRecord.GetProperty("exit_code").GetInt32()); + Assert.Equal( + serialRecord.GetProperty("result").GetRawText(), + parallelRecord.GetProperty("result").GetRawText()); + } + + Assert.Equal(1, serialLines[^1].RootElement.GetProperty("parallelism").GetInt32()); + Assert.Equal(3, parallelLines[^1].RootElement.GetProperty("parallelism").GetInt32()); + } + finally + { + foreach (var document in serialLines) + document.Dispose(); + foreach (var document in parallelLines) + document.Dispose(); + } + } + + [Fact] + public async Task RunBatch_ParallelRefreshesDetachedSnapshotsBetweenItems_Issue4872() + { + using var project = TestProjectHelper.CreateTempProjectScope("cdidx_batch_snapshot_refresh_4872"); + var dbPath = TestProjectHelper.CreateProjectDb(project.Root); + using var writer = new SqliteConnection( + $"Data Source={dbPath};Mode=ReadWrite;Pooling=False"); + writer.Open(); + using (var setup = writer.CreateCommand()) + { + setup.CommandText = """ + PRAGMA journal_mode=WAL; + PRAGMA wal_autocheckpoint=0; + INSERT INTO files(path, lang, size, lines, checksum, modified) + VALUES ('src/Initial.cs', 'csharp', 1, 1, 'initial', CURRENT_TIMESTAMP); + PRAGMA wal_checkpoint(TRUNCATE); + UPDATE files SET checksum = 'initial-hot' WHERE path = 'src/Initial.cs'; + """; + setup.ExecuteNonQuery(); + } + Assert.True(new FileInfo(dbPath + "-wal").Length > 0); + var initialDbLength = new FileInfo(dbPath).Length; + + using var input = new InteractiveBatchTextReader(); + using var stdout = new StringWriter(); + using var stderr = new StringWriter(); + using var firstWaveCompleted = new CountdownEvent(2); + using var cancellation = new CancellationTokenSource(); + Task? runTask = null; + var openedSessions = 0; + QueryCommandRunner.BatchParallelSessionOpenedForTesting = + () => Interlocked.Increment(ref openedSessions); + QueryCommandRunner.BatchParallelCommandCompletedForTesting = lineNumber => + { + if (lineNumber <= 2) + firstWaveCompleted.Signal(); + }; + + try + { + runTask = Task.Run(() => + { + using var capture = ConsoleCapture.Start(stdout, stderr, input); + return QueryCommandRunner.RunBatch( + ["--db", dbPath, "--json-summary", "--parallel", "2"], + _jsonOptions, + cancellationToken: cancellation.Token); + }); + input.WriteLine("""{"command":"status","args":["--json"]}"""); + input.WriteLine("""{"command":"status","args":["--json"]}"""); + Assert.True( + firstWaveCompleted.Wait(TimeSpan.FromSeconds(60)), + "The first parallel batch wave did not complete."); + + using (var update = writer.CreateCommand()) + { + update.CommandText = """ + INSERT INTO files(path, lang, size, lines, checksum, modified) + VALUES ('src/Updated.cs', 'csharp', 1, 1, 'updated', CURRENT_TIMESTAMP); + PRAGMA wal_checkpoint(TRUNCATE); + """; + update.ExecuteNonQuery(); + } + Assert.Equal(initialDbLength, new FileInfo(dbPath).Length); + Assert.Equal(0, new FileInfo(dbPath + "-wal").Length); + + input.WriteLine("""{"command":"status","args":["--json"]}"""); + input.WriteLine("""{"command":"status","args":["--json"]}"""); + input.Complete(); + + var exitCode = await runTask.WaitAsync(TimeSpan.FromSeconds(60)); + Assert.Equal(CommandExitCodes.Success, exitCode); + Assert.Equal(string.Empty, stderr.ToString()); + + var lines = ParseJsonLines(stdout.ToString()); + try + { + Assert.Equal(5, lines.Count); + Assert.Equal(1, lines[0].RootElement.GetProperty("result").GetProperty("files").GetInt32()); + Assert.Equal(1, lines[1].RootElement.GetProperty("result").GetProperty("files").GetInt32()); + Assert.Equal(2, lines[2].RootElement.GetProperty("result").GetProperty("files").GetInt32()); + Assert.Equal(2, lines[3].RootElement.GetProperty("result").GetProperty("files").GetInt32()); + Assert.Equal(4, Volatile.Read(ref openedSessions)); + } + finally + { + foreach (var document in lines) + document.Dispose(); + } + } + finally + { + QueryCommandRunner.BatchParallelSessionOpenedForTesting = null; + QueryCommandRunner.BatchParallelCommandCompletedForTesting = null; + input.Complete(); + if (runTask is { IsCompleted: false }) + { + cancellation.Cancel(); + try + { + await runTask.WaitAsync(TimeSpan.FromSeconds(15)); + } + catch (OperationCanceledException) when (cancellation.IsCancellationRequested) + { + } + } + } + } + + [Fact] + public async Task RunBatch_ParallelRefreshesDirectSessionsBetweenItems_Issue4872() + { + using var project = TestProjectHelper.CreateTempProjectScope("cdidx_batch_direct_refresh_4872"); + var sourceDbPath = TestProjectHelper.CreateProjectDb(project.Root); + using (var source = new SqliteConnection( + $"Data Source={sourceDbPath};Mode=ReadWrite;Pooling=False")) + { + source.Open(); + using var checkpoint = source.CreateCommand(); + checkpoint.CommandText = "PRAGMA wal_checkpoint(TRUNCATE)"; + checkpoint.ExecuteNonQuery(); + } + + var dbPath = Path.Combine(project.Root, "direct.db"); + File.Copy(sourceDbPath, dbPath); + using var writer = new SqliteConnection( + $"Data Source={dbPath};Mode=ReadWrite;Pooling=False"); + writer.Open(); + using (var journalMode = writer.CreateCommand()) + { + journalMode.CommandText = "PRAGMA journal_mode=DELETE"; + Assert.Equal("delete", Assert.IsType(journalMode.ExecuteScalar())); + } + using (var setup = writer.CreateCommand()) + { + setup.CommandText = """ + INSERT INTO files(path, lang, size, lines, checksum, modified) + VALUES ('src/Initial.cs', 'csharp', 1, 1, 'initial', CURRENT_TIMESTAMP); + """; + setup.ExecuteNonQuery(); + } + Assert.Equal("delete", ReadJournalMode(writer)); + + using var input = new InteractiveBatchTextReader(); + using var stdout = new StringWriter(); + using var stderr = new StringWriter(); + using var firstWaveCompleted = new CountdownEvent(2); + using var cancellation = new CancellationTokenSource(); + Task? runTask = null; + var openedSessions = 0; + QueryCommandRunner.BatchParallelSessionOpenedForTesting = + () => Interlocked.Increment(ref openedSessions); + QueryCommandRunner.BatchParallelCommandCompletedForTesting = lineNumber => + { + if (lineNumber <= 2) + firstWaveCompleted.Signal(); + }; + + try + { + runTask = Task.Run(() => + { + using var capture = ConsoleCapture.Start(stdout, stderr, input); + return QueryCommandRunner.RunBatch( + ["--db", dbPath, "--json-summary", "--parallel", "2"], + _jsonOptions, + cancellationToken: cancellation.Token); + }); + input.WriteLine("""{"command":"status","args":["--json"]}"""); + input.WriteLine("""{"command":"status","args":["--json"]}"""); + Assert.True( + firstWaveCompleted.Wait(TimeSpan.FromSeconds(60)), + "The first parallel batch wave did not complete."); + + using (var update = writer.CreateCommand()) + { + update.CommandText = """ + INSERT INTO files(path, lang, size, lines, checksum, modified) + VALUES ('src/Updated.cs', 'csharp', 1, 1, 'updated', CURRENT_TIMESTAMP); + """; + Assert.Equal(1, update.ExecuteNonQuery()); + } + + input.WriteLine("""{"command":"status","args":["--json"]}"""); + input.WriteLine("""{"command":"status","args":["--json"]}"""); + input.Complete(); + + var exitCode = await runTask.WaitAsync(TimeSpan.FromSeconds(60)); + Assert.Equal(CommandExitCodes.Success, exitCode); + Assert.Equal(string.Empty, stderr.ToString()); + + var lines = ParseJsonLines(stdout.ToString()); + try + { + Assert.Equal(5, lines.Count); + Assert.Equal(1, lines[0].RootElement.GetProperty("result").GetProperty("files").GetInt32()); + Assert.Equal(1, lines[1].RootElement.GetProperty("result").GetProperty("files").GetInt32()); + Assert.Equal(2, lines[2].RootElement.GetProperty("result").GetProperty("files").GetInt32()); + Assert.Equal(2, lines[3].RootElement.GetProperty("result").GetProperty("files").GetInt32()); + Assert.Equal(4, Volatile.Read(ref openedSessions)); + } + finally + { + foreach (var document in lines) + document.Dispose(); + } + } + finally + { + QueryCommandRunner.BatchParallelSessionOpenedForTesting = null; + QueryCommandRunner.BatchParallelCommandCompletedForTesting = null; + input.Complete(); + if (runTask is { IsCompleted: false }) + { + cancellation.Cancel(); + try + { + await runTask.WaitAsync(TimeSpan.FromSeconds(15)); + } + catch (OperationCanceledException) when (cancellation.IsCancellationRequested) + { + } + } + } + } + + [Fact] + public void RunBatch_InitialValidationFailuresDisposeDetachedSnapshots_Issue4872() + { + using var project = TestProjectHelper.CreateTempProjectScope("cdidx_batch_validation_failure_4872"); + var dbPath = TestProjectHelper.CreateProjectDb(project.Root); + using var writer = new SqliteConnection( + $"Data Source={dbPath};Mode=ReadWrite;Pooling=False"); + writer.Open(); + using (var setup = writer.CreateCommand()) + { + setup.CommandText = """ + PRAGMA journal_mode=WAL; + PRAGMA wal_autocheckpoint=0; + INSERT INTO files(path, lang, size, lines, checksum, modified) + VALUES ('src/Initial.cs', 'csharp', 1, 1, 'initial', CURRENT_TIMESTAMP); + """; + setup.ExecuteNonQuery(); + } + + var snapshotDirectories = new ConcurrentBag(); + var originalDirectoryHook = DbConnectionFactory.QueryOnlySnapshotDirectoryCreatedForTesting; + var originalValidationHook = QueryCommandRunner.BatchParallelDatabaseValidatingForTesting; + DbConnectionFactory.QueryOnlySnapshotDirectoryCreatedForTesting = snapshotDirectories.Add; + QueryCommandRunner.BatchParallelDatabaseValidatingForTesting = + () => throw new InvalidDataException("Injected parallel batch validation failure."); + + try + { + Assert.Throws(() => CaptureConsoleWithInput( + BuildIssue4872BatchInput(1), + () => QueryCommandRunner.RunBatch( + ["--db", dbPath, "--json-summary", "--parallel", "2"], + _jsonOptions))); + } + finally + { + QueryCommandRunner.BatchParallelDatabaseValidatingForTesting = originalValidationHook; + DbConnectionFactory.QueryOnlySnapshotDirectoryCreatedForTesting = originalDirectoryHook; + } + + Assert.NotEmpty(snapshotDirectories); + Assert.All(snapshotDirectories, path => Assert.False(Directory.Exists(path), path)); + } + + [Fact] + public void RunBatch_ReaderConstructionFailuresDisposeDetachedSnapshots_Issue4872() + { + using var project = TestProjectHelper.CreateTempProjectScope("cdidx_batch_reader_failure_4872"); + var dbPath = TestProjectHelper.CreateProjectDb(project.Root); + using var writer = new SqliteConnection( + $"Data Source={dbPath};Mode=ReadWrite;Pooling=False"); + writer.Open(); + using (var setup = writer.CreateCommand()) + { + setup.CommandText = """ + PRAGMA journal_mode=WAL; + PRAGMA wal_autocheckpoint=0; + INSERT INTO files(path, lang, size, lines, checksum, modified) + VALUES ('src/Initial.cs', 'csharp', 1, 1, 'initial', CURRENT_TIMESTAMP); + """; + setup.ExecuteNonQuery(); + } + Assert.True(new FileInfo(dbPath + "-wal").Length > 0); + + var snapshotDirectories = new ConcurrentBag(); + var readerConstructionAttempts = 0; + var originalDirectoryHook = DbConnectionFactory.QueryOnlySnapshotDirectoryCreatedForTesting; + var originalReaderFactory = QueryCommandRunner.BatchParallelReaderFactoryForTesting; + DbConnectionFactory.QueryOnlySnapshotDirectoryCreatedForTesting = snapshotDirectories.Add; + QueryCommandRunner.BatchParallelReaderFactoryForTesting = db => + { + if (Interlocked.Increment(ref readerConstructionAttempts) == 1) + return new DbReader(db); + throw new InvalidDataException("Injected parallel batch reader construction failure."); + }; + + int exitCode; + string stdout; + string stderr; + try + { + (exitCode, stdout, stderr) = CaptureConsoleWithInput( + BuildIssue4872BatchInput(6), + () => QueryCommandRunner.RunBatch( + ["--db", dbPath, "--json-summary", "--parallel", "2"], + _jsonOptions)); + } + finally + { + QueryCommandRunner.BatchParallelReaderFactoryForTesting = originalReaderFactory; + DbConnectionFactory.QueryOnlySnapshotDirectoryCreatedForTesting = originalDirectoryHook; + } + + Assert.Equal(CommandExitCodes.RuntimeError, exitCode); + Assert.Equal(string.Empty, stderr); + Assert.True(Volatile.Read(ref readerConstructionAttempts) >= 3); + Assert.True(snapshotDirectories.Count >= 3); + Assert.All(snapshotDirectories, path => Assert.False(Directory.Exists(path), path)); + + var lines = ParseJsonLines(stdout); + try + { + Assert.Equal(7, lines.Count); + Assert.Equal(3, lines.Take(6).Count( + line => line.RootElement.GetProperty("exit_code").GetInt32() == CommandExitCodes.RuntimeError)); + } + finally + { + foreach (var document in lines) + document.Dispose(); + } + } + + [Fact] + public void RunBatch_ParallelSessionReuseKeepsTwelveItemCostWithinThreeItemRatio_Issue4872() + { + using var project = TestProjectHelper.CreateTempProjectScope("cdidx_batch_ratio_4872"); + var dbPath = TestProjectHelper.CreateProjectDb(project.Root); + using var writer = new SqliteConnection( + $"Data Source={dbPath};Mode=ReadWrite;Pooling=False"); + writer.Open(); + using (var command = writer.CreateCommand()) + { + command.CommandText = """ + PRAGMA journal_mode=WAL; + PRAGMA wal_autocheckpoint=0; + CREATE TABLE batch_ratio_filler(payload BLOB NOT NULL); + INSERT INTO batch_ratio_filler(payload) VALUES (zeroblob(16777216)); + """; + command.ExecuteNonQuery(); + } + Assert.True(new FileInfo(dbPath + "-wal").Length >= 16_777_216); + + _ = MeasureIssue4872ParallelBatch(dbPath, commandCount: 3); + var shortRuns = new TimeSpan[3]; + var longRuns = new TimeSpan[3]; + for (var iteration = 0; iteration < shortRuns.Length; iteration++) + { + shortRuns[iteration] = MeasureIssue4872ParallelBatch(dbPath, commandCount: 3); + longRuns[iteration] = MeasureIssue4872ParallelBatch(dbPath, commandCount: 12); + } + + var shortBaseline = shortRuns.Min(); + var longBaseline = longRuns.Min(); + var ratio = longBaseline.TotalMilliseconds / shortBaseline.TotalMilliseconds; + Assert.True( + ratio <= 3.0, + $"Parallel batch session reuse regressed: 12-item/3-item ratio was {ratio:F2} " + + $"({longBaseline.TotalMilliseconds:F1} ms / {shortBaseline.TotalMilliseconds:F1} ms)."); + } + + private TimeSpan MeasureIssue4872ParallelBatch(string dbPath, int commandCount) + { + using var firstWaveStarted = new CountdownEvent(3); + QueryCommandRunner.BatchParallelSessionOpenedForTesting = static () => + { + // Model an expensive large-index open/schema phase. This is workload cost, + // not a synchronization delay; the countdown below owns worker coordination. + Thread.Sleep(TimeSpan.FromMilliseconds(100)); + }; + QueryCommandRunner.BatchParallelCommandStartedForTesting = lineNumber => + { + if (lineNumber > 3) + return; + firstWaveStarted.Signal(); + if (!firstWaveStarted.Wait(TimeSpan.FromSeconds(5))) + throw new TimeoutException("The parallel batch benchmark wave did not start."); + }; + + try + { + var stopwatch = Stopwatch.StartNew(); + var (exitCode, _, stderr) = CaptureConsoleWithInput( + BuildIssue4872RejectedBatchInput(commandCount), + () => QueryCommandRunner.RunBatch( + ["--db", dbPath, "--json-summary", "--parallel", "3"], + _jsonOptions)); + stopwatch.Stop(); + Assert.Equal(CommandExitCodes.UsageError, exitCode); + Assert.Equal(string.Empty, stderr); + return stopwatch.Elapsed; + } + finally + { + QueryCommandRunner.BatchParallelSessionOpenedForTesting = null; + QueryCommandRunner.BatchParallelCommandStartedForTesting = null; + } + } + + private static string BuildIssue4872BatchInput(int commandCount) + => string.Join( + '\n', + Enumerable.Repeat( + """{"command":"languages","args":["--format","count"]}""", + commandCount)) + "\n"; + + private static string BuildIssue4872RejectedBatchInput(int commandCount) + => string.Join( + '\n', + Enumerable.Repeat( + """{"command":"unknown"}""", + commandCount)) + "\n"; + + private static string ReadJournalMode(SqliteConnection connection) + { + using var command = connection.CreateCommand(); + command.CommandText = "PRAGMA journal_mode"; + return Assert.IsType(command.ExecuteScalar()); + } +}