From 1181a7077fd716307dd41614d265383d7aa5205e Mon Sep 17 00:00:00 2001 From: Minoru OSUKA Date: Mon, 13 Jul 2026 23:15:10 +0900 Subject: [PATCH] feat(engine): batch ingestion API put_documents / add_documents with one WAL fsync per batch (#863) Add Engine::put_documents / add_documents, batched forms of the singular write methods. Documents are applied sequentially in input order (WAL doc_id/seq allocation order is the replay order; in-batch duplicate-id dedup requires program order) and fail fast with the new LaurusError::BatchIngest { failed_index, failed_id, applied, source }. There is no batch rollback: the applied prefix stays in the WAL and NRT buffers, so retrying the batch or its suffix is idempotent. Under the default WalSyncPolicy::PerRecord the whole batch is made durable with a single WAL fsync at batch end instead of one per record: DocumentLog::defer_sync() opens an RAII sync-deferral scope that suppresses the per-record fsync, and the engine calls flush_wal() once on both the success and error paths. Group policy thresholds keep firing mid-batch, preserving that policy's bounded loss window. The deferral flag is global, so singular writes acknowledged during a concurrent batch would silently lose their per-record durability; put_document / add_document / delete_documents now re-assert it via DocumentLog::ensure_per_record_durability() before returning (a no-op in the common no-batch path). Verification: a cfg(test) fsync counter asserts the N->1 amortization exactly; integration tests cover uncommitted-batch recovery after reopen and applied-prefix recovery of a failed batch. On the FileStorage bench backend, ingest/bulk_batch_api/1000 runs ~36x faster than the per-record ingest/bulk/1000 (190 ms vs 6.9 s, fsync-bound). Refs #551, #572 --- docs/ja/src/laurus/api_reference.md | 4 + docs/ja/src/laurus/engine.md | 2 + docs/ja/src/laurus/persistence.md | 19 ++ docs/src/laurus/api_reference.md | 4 + docs/src/laurus/engine.md | 2 + docs/src/laurus/persistence.md | 19 ++ laurus/benches/lexical_indexing_bench.rs | 33 ++ laurus/src/engine.rs | 410 ++++++++++++++++++++++- laurus/src/error.rs | 25 ++ laurus/src/store/log.rs | 205 +++++++++++- laurus/tests/engine_batch_ingest_test.rs | 168 ++++++++++ 11 files changed, 875 insertions(+), 16 deletions(-) create mode 100644 laurus/tests/engine_batch_ingest_test.rs diff --git a/docs/ja/src/laurus/api_reference.md b/docs/ja/src/laurus/api_reference.md index 3b332f41..0b8f4238 100644 --- a/docs/ja/src/laurus/api_reference.md +++ b/docs/ja/src/laurus/api_reference.md @@ -15,6 +15,8 @@ cargo doc --open | `Engine::builder(storage, schema)` | `EngineBuilder` を作成 | | `engine.put_document(id, doc).await?` | ドキュメントのUpsert（IDが存在する場合は置き換え） | | `engine.add_document(id, doc).await?` | ドキュメントをチャンクとして追加（複数のチャンクが同一IDを共有可能） | +| `engine.put_documents(docs).await?` | `(id, doc)` ペアのバッチ Upsert — WAL fsync はバッチごとに 1 回（[バッチインジェスト](persistence.md#バッチインジェスト) 参照） | +| `engine.add_documents(docs).await?` | `(id, doc)` ペアのバッチチャンク追加 — 耐久性・エラー意味論は `put_documents` と同一 | | `engine.delete_documents(id).await?` | 外部IDによるすべてのドキュメント/チャンクの削除 | | `engine.get_documents(id).await?` | 外部IDによるすべてのドキュメント/チャンクの取得 | | `engine.search(request).await?` | 検索リクエストの実行 | @@ -26,6 +28,8 @@ cargo doc --open | `engine.stats()?` | インデックス統計の取得 | > **`put_document` と `add_document` の違い:** `put_document` はUpsertを実行します。同じ外部IDのドキュメントが既に存在する場合、削除して置き換えます。`add_document` は常に追加し、複数のドキュメントチャンクが同じ外部IDを共有できます。詳細は [Schema & Fields -- ドキュメントのインデキシング](../concepts/schema_and_fields.md#indexing-documents) を参照してください。 +> +> **バッチ形式:** `put_documents` / `add_documents` は `(id, doc)` ペアを入力順に逐次適用します（1 回の `put_documents` バッチ内で重複した ID は逐次 put と同じくデデュープされ、最後の出現が勝ちます）。適用できないドキュメントに遭遇すると fail-fast で `LaurusError::BatchIngest { failed_index, failed_id, applied, .. }` を返します。バッチロールバックはありません: 適用済みドキュメントは WAL と NRT バッファに残るため、バッチ（またはその suffix）の再試行は冪等です。耐久性の詳細は[バッチインジェスト](persistence.md#バッチインジェスト)を参照してください。 ### EngineBuilder diff --git a/docs/ja/src/laurus/engine.md b/docs/ja/src/laurus/engine.md index e00a134c..5aa1a070 100644 --- a/docs/ja/src/laurus/engine.md +++ b/docs/ja/src/laurus/engine.md @@ -151,6 +151,8 @@ graph LR | :--- | :--- | | `put_document(id, doc)` | Upsert -- 同じIDのドキュメントが既存の場合は置き換え | | `add_document(id, doc)` | 追加 -- 新しいチャンクとして追加（複数のチャンクが同一IDを共有可能） | +| `put_documents(docs)` | バッチ Upsert -- `(id, doc)` ペアを順序どおり適用し、WAL fsync はバッチごとに 1 回 | +| `add_documents(docs)` | バッチ追加 -- `put_documents` と同様だが既存チャンクを削除しない | | `get_documents(id)` | 外部IDによるすべてのドキュメント/チャンクの取得 | | `delete_documents(id)` | 外部IDによるすべてのドキュメント/チャンクの削除 | | `commit()` | 保留中の変更をストレージにフラッシュ（ドキュメントが検索可能になる） | diff --git a/docs/ja/src/laurus/persistence.md b/docs/ja/src/laurus/persistence.md index 42588800..bb2629ff 100644 --- a/docs/ja/src/laurus/persistence.md +++ b/docs/ja/src/laurus/persistence.md @@ -109,6 +109,25 @@ engine.add_document("doc-3", doc3).await?; > **ヒント:** コミットはセグメントをストレージにフラッシュするため比較的コストが高い操作です。バルクインデキシングでは、`commit()` を呼び出す前に多数のドキュメントをバッチ処理してください。 +## バッチインジェスト + +`put_documents` / `add_documents` は `put_document` / `add_document` のバッチ形式です。`(id, doc)` ペアを**入力順に逐次適用**し、既定の `PerRecord` ポリシー下ではレコードごとではなく**バッチ末尾の 1 回の WAL fsync** でバッチ全体を durable にします。 + +```rust +let docs: Vec<(String, Document)> = build_batch(); +engine.put_documents(docs).await?; // fsync 1 回で、全ドキュメントが単発 put と同等に durable +engine.commit().await?; // バッチ全体で 1 回のセグメントフラッシュ +``` + +留意すべき意味論: + +- **順序**: 1 回の `put_documents` バッチ内で重複した外部 ID は、同じ put を逐次発行した場合とまったく同じようにデデュープされます（最後の出現が勝ち）。`add_documents` での ID の繰り返しは正当なマルチチャンク追加です。 +- **fail-fast・ロールバックなし**: 適用できない最初のドキュメントでバッチは停止し、`LaurusError::BatchIngest { failed_index, failed_id, applied, .. }` を返します。失敗前に適用された `applied` 件はロールバック**されません** — WAL と NRT バッファに残り（即座に検索可能、次のコミットで永続化、クラッシュ時はリカバリで再生）、エラー経路でもバッチ末尾の WAL フラッシュは実行されます。バッチ全体、または `failed_index` からの suffix の再試行は put 意味論の下で冪等です。 +- **耐久性**: 呼び出しが `Ok` を返した時点で、バッチ内の全ドキュメントは成功した単発 put とまったく同等に durable です。呼び出し途中のクラッシュで失われるのは fsync 前の末尾のみで、リカバリは fsync 済み prefix をドキュメント単位で再生し、途中で切れた末尾レコードは通常どおり CRC フレーミングが破棄します。 +- **サイズ指針**: エンジンは各ドキュメントを WAL へ順次クローンするため、バッチのメモリは呼び出し側の `Vec` が支配的です。1 回の呼び出しあたり 1,000〜10,000 ドキュメントが良い既定値で、より大きなコーパスは複数回の呼び出しに分割してください（セグメントサイズを抑えるため定期的なコミットも推奨）。 +- **`Group` ポリシー下**: バッチ中もグループしきい値は発火し続けるため、同ポリシーの有界な損失ウィンドウは保たれます。バッチ末尾のフラッシュも実行されます。 +- **並行する単発書き込み**: 別タスクのバッチ実行中に完了した `put_document` / `add_document` / `delete_documents` は per-record の耐久性を完全に維持します — 単発書き込みは ack 前に fsync を再アサートするため、バッチが他の呼び出し元の保証を弱めることはありません。 + ## WAL 耐久性ポリシー 既定では、各 `add`/`delete` は返る前に WAL を fsync するため、成功した書き込みがクラッシュで失われることはありません。大量に取り込む場合、この書き込みごとの fsync がスループットのボトルネックになります。`WalSyncPolicy` により、書き込みごとの耐久性とスループットをトレードオフできます。 diff --git a/docs/src/laurus/api_reference.md b/docs/src/laurus/api_reference.md index 3e75a5e8..d0e0c29c 100644 --- a/docs/src/laurus/api_reference.md +++ b/docs/src/laurus/api_reference.md @@ -15,6 +15,8 @@ The central coordinator for all indexing and search operations. | `Engine::builder(storage, schema)` | Create an `EngineBuilder` | | `engine.put_document(id, doc).await?` | Upsert a document (replace if ID exists) | | `engine.add_document(id, doc).await?` | Add a document as a chunk (multiple chunks can share an ID) | +| `engine.put_documents(docs).await?` | Batched upsert of `(id, doc)` pairs — one WAL fsync per batch (see [Batch Ingestion](persistence.md#batch-ingestion)) | +| `engine.add_documents(docs).await?` | Batched chunk append of `(id, doc)` pairs — same durability and error semantics as `put_documents` | | `engine.delete_documents(id).await?` | Delete all documents/chunks by external ID | | `engine.get_documents(id).await?` | Get all documents/chunks by external ID | | `engine.search(request).await?` | Execute a search request | @@ -26,6 +28,8 @@ The central coordinator for all indexing and search operations. | `engine.stats()?` | Get index statistics | > **`put_document` vs `add_document`:** `put_document` performs an upsert — if a document with the same external ID already exists, it is deleted and replaced. `add_document` always appends, allowing multiple document chunks to share the same external ID. See [Schema & Fields — Indexing Documents](../concepts/schema_and_fields.md#indexing-documents) for details. +> +> **Batch forms:** `put_documents` / `add_documents` apply their `(id, doc)` pairs sequentially in input order (duplicate IDs within one `put_documents` batch dedup exactly like sequential puts — the last occurrence wins) and fail fast with `LaurusError::BatchIngest { failed_index, failed_id, applied, .. }` on the first document that cannot be applied. There is no batch rollback: applied documents stay in the WAL and NRT buffers, so retrying the batch (or its suffix) is idempotent. See [Batch Ingestion](persistence.md#batch-ingestion) for the durability details. ### EngineBuilder diff --git a/docs/src/laurus/engine.md b/docs/src/laurus/engine.md index a5fa6748..9fded212 100644 --- a/docs/src/laurus/engine.md +++ b/docs/src/laurus/engine.md @@ -150,6 +150,8 @@ graph LR | :--- | :--- | | `put_document(id, doc)` | Upsert -- replaces any existing document with the same ID | | `add_document(id, doc)` | Append -- adds as a new chunk (multiple chunks can share an ID) | +| `put_documents(docs)` | Batched upsert -- applies `(id, doc)` pairs in order with one WAL fsync per batch | +| `add_documents(docs)` | Batched append -- like `put_documents` but never deletes existing chunks | | `get_documents(id)` | Retrieve all documents/chunks by external ID | | `delete_documents(id)` | Delete all documents/chunks by external ID | | `commit()` | Flush pending changes to storage (makes documents searchable) | diff --git a/docs/src/laurus/persistence.md b/docs/src/laurus/persistence.md index 2c88ec54..90d7afcb 100644 --- a/docs/src/laurus/persistence.md +++ b/docs/src/laurus/persistence.md @@ -109,6 +109,25 @@ engine.add_document("doc-3", doc3).await?; > **Tip:** Commits are relatively expensive because they flush segments to storage. For bulk indexing, batch many documents before calling `commit()`. +## Batch Ingestion + +`put_documents` / `add_documents` are the batched forms of `put_document` / `add_document`. They apply their `(id, doc)` pairs **sequentially, in input order**, and under the default `PerRecord` policy they make the whole batch durable with a **single WAL fsync at batch end** instead of one per record: + +```rust +let docs: Vec<(String, Document)> = build_batch(); +engine.put_documents(docs).await?; // one fsync, all docs as durable as singular puts +engine.commit().await?; // one segment flush for the whole batch +``` + +Semantics to be aware of: + +- **Ordering**: duplicate external IDs within one `put_documents` batch dedup exactly like the same puts issued sequentially — the last occurrence wins. Repeating an ID in `add_documents` legitimately appends multiple chunks. +- **Fail-fast, no rollback**: the batch stops at the first document that cannot be applied and returns `LaurusError::BatchIngest { failed_index, failed_id, applied, .. }`. The `applied` documents before the failure are **not** rolled back — they stay in the WAL and NRT buffers (searchable immediately, durable at the next commit, replayed on crash recovery), and the batch-end WAL flush runs on the error path too. Retrying the batch, or its suffix starting at `failed_index`, is idempotent under put semantics. +- **Durability**: when the call returns `Ok`, every document in the batch is exactly as durable as a successful singular put. A crash mid-call loses at most the un-fsync'd tail; recovery replays the fsync'd prefix per document, and a torn trailing record is dropped by the CRC framing as usual. +- **Sizing**: the engine clones each document into the WAL as it goes, so batch memory is dominated by the caller's `Vec`. Batches of 1,000-10,000 documents per call are a good default; chunk larger corpora into multiple calls (and commit periodically to bound segment sizes). +- **Under `Group` policy**: the group thresholds keep firing mid-batch, so that policy's bounded loss window is preserved; the batch-end flush still runs. +- **Concurrent singular writes**: a `put_document` / `add_document` / `delete_documents` call that completes while another task's batch is in flight keeps its full per-record durability — singular writes re-assert the fsync before acknowledging, so a batch never weakens anyone else's guarantee. + ## WAL Durability Policy By default, each `add`/`delete` fsyncs the WAL before returning, so a successful write can never be lost to a crash. When ingesting at high volume, that per-write fsync becomes the throughput bottleneck. The `WalSyncPolicy` lets you trade per-write durability for throughput: diff --git a/laurus/benches/lexical_indexing_bench.rs b/laurus/benches/lexical_indexing_bench.rs index 52e3570e..0828d788 100644 --- a/laurus/benches/lexical_indexing_bench.rs +++ b/laurus/benches/lexical_indexing_bench.rs @@ -344,6 +344,38 @@ fn bench_bulk_ingest(c: &mut Criterion) { group.finish(); } +/// End-to-end `add_documents(N) + commit` — the batch-API twin of +/// `ingest/bulk` (#551). One WAL fsync per batch instead of one per record, +/// so the delta vs `ingest/bulk` is only visible on the FileStorage backend +/// (`LAURUS_BENCH_DISK=1`); on MemoryStorage the two should be equal. +fn bench_bulk_ingest_batch_api(c: &mut Criterion) { + let rt = Runtime::new().unwrap(); + let mut group = c.benchmark_group("ingest/bulk_batch_api"); + group.sample_size(SAMPLE_SIZE_SLOW); + + for &n in &ingest_corpus_sizes() { + group.throughput(Throughput::Elements(n as u64)); + group.bench_with_input(BenchmarkId::from_parameter(n), &n, |b, &n| { + b.iter_batched( + || { + let engine = rt.block_on(build_empty_engine()).unwrap(); + let docs = build_documents(n); + (engine, docs) + }, + |(engine, docs)| { + rt.block_on(async { + engine.add_documents(docs).await.unwrap(); + engine.commit().await.unwrap(); + }); + }, + BatchSize::SmallInput, + ); + }); + } + + group.finish(); +} + /// `(add × K + commit) × M` so M segments accumulate. M = 8 typically /// crosses the default `TieredMergePolicy::max_segments_per_tier = 4` /// threshold, so the M-sweep also exposes any auto-merge work. @@ -403,6 +435,7 @@ criterion_group!( bench_add_documents, bench_commit, bench_bulk_ingest, + bench_bulk_ingest_batch_api, bench_multi_segment_commit, ); criterion_main!(benches); diff --git a/laurus/src/engine.rs b/laurus/src/engine.rs index 88a26a21..46a6a030 100644 --- a/laurus/src/engine.rs +++ b/laurus/src/engine.rs @@ -284,6 +284,10 @@ impl Engine { /// or indexing into either the lexical or vector store fails. pub async fn put_document(&self, id: &str, doc: Document) -> Result<()> { let _ = self.index_internal(id, doc, false).await?; + // Re-assert per-record durability: a concurrent batch may hold a WAL + // sync-deferral scope, which would otherwise leave this acknowledged + // write unsynced. A no-op when the append already self-synced. + self.log.ensure_per_record_durability()?; Ok(()) } @@ -310,9 +314,126 @@ impl Engine { /// or vector store fails. pub async fn add_document(&self, id: &str, doc: Document) -> Result<()> { let _ = self.index_internal(id, doc, true).await?; + // Re-assert per-record durability: a concurrent batch may hold a WAL + // sync-deferral scope, which would otherwise leave this acknowledged + // write unsynced. A no-op when the append already self-synced. + self.log.ensure_per_record_durability()?; Ok(()) } + /// Index a batch of documents, replacing any existing documents that + /// share an external ID (batched form of [`Self::put_document`]). + /// + /// Documents are applied **sequentially, in input order** — the WAL + /// doc_id/seq allocation order is the crash-recovery replay order, and + /// duplicate external IDs within one batch must dedup exactly like the + /// equivalent sequence of [`Self::put_document`] calls (the last + /// occurrence wins). Both underlying stores serialize writes behind + /// mutexes, so unlike [`Self::search_batch`] there is nothing to gain + /// from processing batch entries concurrently. + /// + /// Under the default [`WalSyncPolicy::PerRecord`](crate::store::log::WalSyncPolicy) + /// the whole batch is made durable with a **single WAL fsync** at batch + /// end instead of one per record: per-record fsync is suppressed for the + /// duration of the call (via [`DocumentLog::defer_sync`](crate::store::log::DocumentLog::defer_sync)) + /// and [`Self::flush_wal`] runs once before returning, on both the + /// success and the error path. When the call returns `Ok`, every + /// document in the batch is as durable as a singular put. + /// + /// # Parameters + /// + /// - `docs` - `(external_id, document)` pairs, applied in order. + /// An empty batch is a no-op. + /// + /// # Errors + /// + /// Fails fast at the first document that cannot be applied, returning + /// [`LaurusError::BatchIngest`](crate::error::LaurusError::BatchIngest) with the failing position, its external + /// id, and the number of documents already applied. Applied documents + /// are **not** rolled back: they stay in the WAL and NRT buffers, + /// searchable immediately, durable at the next commit, and replayed on + /// crash recovery — so retrying the batch (or its suffix from + /// `failed_index`) is idempotent. The failing document inherits the + /// singular-put semantics: its WAL record (if written) is retried by + /// crash recovery and discarded by the next successful commit. + pub async fn put_documents(&self, docs: Vec<(String, Document)>) -> Result<()> { + self.index_batch_internal(docs, false).await + } + + /// Index a batch of document chunks, always appending (batched form of + /// [`Self::add_document`]). + /// + /// Unlike [`Self::put_documents`], existing documents with the same + /// external ID are **not** deleted, so a batch may legitimately repeat + /// an ID to add multiple chunks of the same logical document in one + /// call. + /// + /// Ordering, durability (single WAL fsync per batch under the default + /// per-record policy), and fail-fast error semantics are identical to + /// [`Self::put_documents`]. + /// + /// # Parameters + /// + /// - `docs` - `(external_id, document)` pairs, applied in order. + /// An empty batch is a no-op. + /// + /// # Errors + /// + /// Fails fast with [`LaurusError::BatchIngest`](crate::error::LaurusError::BatchIngest); see + /// [`Self::put_documents`] for the exact semantics. + pub async fn add_documents(&self, docs: Vec<(String, Document)>) -> Result<()> { + self.index_batch_internal(docs, true).await + } + + /// Shared sequential loop behind [`Self::put_documents`] / + /// [`Self::add_documents`]. + /// + /// Wraps per-document [`Self::index_internal`] calls in a WAL + /// sync-deferral scope and flushes the WAL exactly once at batch end on + /// both exit paths, converting the first per-document failure into + /// [`LaurusError::BatchIngest`](crate::error::LaurusError::BatchIngest). + /// + /// # Parameters + /// + /// - `docs` - `(external_id, document)` pairs, applied in order. + /// - `as_chunk` - `false` for put (delete-first) semantics, `true` for + /// add (append-chunk) semantics. + /// + /// # Errors + /// + /// Returns [`LaurusError::BatchIngest`](crate::error::LaurusError::BatchIngest) on the first failing document, + /// or the WAL flush error if the batch applied fully but the final + /// fsync failed (retrying the whole batch is idempotent). + async fn index_batch_internal( + &self, + docs: Vec<(String, Document)>, + as_chunk: bool, + ) -> Result<()> { + if docs.is_empty() { + return Ok(()); + } + let deferral = self.log.defer_sync(); + for (index, (id, doc)) in docs.into_iter().enumerate() { + if let Err(e) = self.index_internal(&id, doc, as_chunk).await { + drop(deferral); + // Make the applied prefix durable before reporting the + // failure, keeping the per-record durability contract at + // batch granularity. If the fsync itself fails the records + // stay marked dirty and are retried by the next flush_wal / + // commit, so the per-document error remains the primary one. + let _ = self.log.flush_wal(); + return Err(crate::error::LaurusError::BatchIngest { + failed_index: index, + failed_id: id, + applied: index, + source: Box::new(e), + }); + } + } + drop(deferral); + self.log.flush_wal() + } + async fn index_internal(&self, id: &str, mut doc: Document, as_chunk: bool) -> Result { // 1. Inject _id field use crate::data::DataValue; @@ -324,7 +445,7 @@ impl Engine { self.apply_dynamic_schema(&mut doc).await?; if !as_chunk { - self.delete_documents(id).await?; + self.delete_documents_internal(id).await?; } // 2. Write-Ahead Log: assign doc_id + persist (before any index updates) @@ -542,6 +663,28 @@ impl Engine { /// Returns an error if the WAL write, lexical deletion, or vector /// deletion fails for any matched document. pub async fn delete_documents(&self, id: &str) -> Result<()> { + self.delete_documents_internal(id).await?; + // Re-assert per-record durability: a concurrent batch may hold a WAL + // sync-deferral scope, which would otherwise leave these acknowledged + // deletes unsynced. A no-op when the appends already self-synced. + self.log.ensure_per_record_durability()?; + Ok(()) + } + + /// Body of [`Self::delete_documents`] without the per-record durability + /// re-assertion, so the batch-ingest path (whose put semantics + /// delete-first per document inside a WAL sync-deferral scope) does not + /// fsync once per deleted document. + /// + /// # Parameters + /// + /// - `id` - The external document identifier to delete. + /// + /// # Errors + /// + /// Returns an error if the WAL write, lexical deletion, or vector + /// deletion fails for any matched document. + async fn delete_documents_internal(&self, id: &str) -> Result<()> { let doc_ids = self.lexical.find_doc_ids_by_term("_id", id)?; for doc_id in doc_ids { // 1. Write to log @@ -2637,6 +2780,271 @@ mod tests { } } + /// Build a `(id, doc)` batch entry with a single `title` text field, for + /// the `put_documents` / `add_documents` tests below. + fn batch_entry(id: &str, title: &str) -> (String, crate::data::Document) { + use crate::data::DataValue; + let mut doc = crate::data::Document::new(); + doc.fields + .insert("title".into(), DataValue::Text(title.into())); + (id.to_string(), doc) + } + + /// Build an engine over `MemoryStorage` with a single `title` text field, + /// for the `put_documents` / `add_documents` tests below. + async fn title_only_engine() -> Engine { + use crate::engine::schema::FieldOption; + use crate::lexical::core::field::TextOption; + let storage: Arc = Arc::new(MemoryStorage::new(Default::default())); + let schema = Schema::builder() + .add_field("title", FieldOption::Text(TextOption::default())) + .build(); + Engine::new(storage, schema).await.unwrap() + } + + /// #551: an empty batch is a no-op — no WAL activity, no error. + #[tokio::test] + async fn test_put_documents_empty_batch_is_noop() { + let engine = title_only_engine().await; + engine.put_documents(Vec::new()).await.unwrap(); + engine.add_documents(Vec::new()).await.unwrap(); + assert!( + !engine.log.wal_is_dirty(), + "an empty batch must not leave unsynced WAL bytes" + ); + assert_eq!(engine.stats().unwrap().document_count, 0); + } + + /// #551: `put_documents` must produce exactly the same final state as the + /// equivalent sequence of singular `put_document` calls. + #[tokio::test] + async fn test_put_documents_matches_sequential_puts() { + let batch_engine = title_only_engine().await; + let sequential_engine = title_only_engine().await; + + let docs: Vec<_> = (0..20) + .map(|i| batch_entry(&format!("id{i}"), &format!("title-{i}"))) + .collect(); + + for (id, doc) in docs.clone() { + sequential_engine.put_document(&id, doc).await.unwrap(); + } + batch_engine.put_documents(docs).await.unwrap(); + + batch_engine.commit().await.unwrap(); + sequential_engine.commit().await.unwrap(); + + assert_eq!( + batch_engine.stats().unwrap().document_count, + sequential_engine.stats().unwrap().document_count, + ); + for i in 0..20 { + let batch_docs = batch_engine.get_documents(&format!("id{i}")).await.unwrap(); + let seq_docs = sequential_engine + .get_documents(&format!("id{i}")) + .await + .unwrap(); + assert_eq!(batch_docs.len(), 1, "id{i} must resolve to one doc"); + assert_eq!( + batch_docs[0].fields.get("title"), + seq_docs[0].fields.get("title"), + "id{i} must carry the same title on both paths" + ); + } + } + + /// #551: duplicate external ids **within one batch** must dedup exactly + /// like the same puts issued sequentially (last occurrence wins) — the + /// batch mirror of `test_put_document_dedupes_duplicate_ids_in_batch`. + #[tokio::test] + async fn test_put_documents_dedupes_duplicate_ids_in_batch() { + let engine = title_only_engine().await; + + // 10 unique ids × 3 revisions interleaved in a single batch call. + let mut docs = Vec::new(); + for rev in 0..3 { + for i in 0..10 { + docs.push(batch_entry(&format!("id{i}"), &format!("id{i}-rev{rev}"))); + } + } + engine.put_documents(docs).await.unwrap(); + engine.commit().await.unwrap(); + + assert_eq!( + engine.stats().unwrap().document_count, + 10, + "exactly 10 unique docs should be live after in-batch dedup" + ); + for i in 0..10 { + let docs = engine.get_documents(&format!("id{i}")).await.unwrap(); + assert_eq!(docs.len(), 1, "id{i} should resolve to a single doc"); + let title = docs[0] + .fields + .get("title") + .and_then(|v| v.as_text()) + .map(String::from); + assert_eq!( + title.as_deref(), + Some(format!("id{i}-rev2").as_str()), + "id{i} must retain the last occurrence in the batch" + ); + } + } + + /// #551: `add_documents` never delete-firsts, so repeating an id within + /// one batch legitimately accumulates chunks. + #[tokio::test] + async fn test_add_documents_same_id_creates_chunks() { + let engine = title_only_engine().await; + + let docs: Vec<_> = (0..4) + .map(|i| batch_entry("doc", &format!("chunk-{i}"))) + .collect(); + engine.add_documents(docs).await.unwrap(); + engine.commit().await.unwrap(); + + let chunks = engine.get_documents("doc").await.unwrap(); + assert_eq!( + chunks.len(), + 4, + "all four chunks sharing the external id must be live" + ); + } + + /// #551: fail-fast semantics — the batch stops at the first failing doc, + /// reports its position/id and the applied count, and the applied prefix + /// stays NRT-visible (no rollback). + #[tokio::test] + async fn test_put_documents_failfast_reports_index_and_applied() { + use crate::engine::schema::{DynamicFieldPolicy, FieldOption}; + use crate::lexical::core::field::TextOption; + + let storage: Arc = Arc::new(MemoryStorage::new(Default::default())); + let schema = Schema::builder() + .add_field("title", FieldOption::Text(TextOption::default())) + .dynamic_field_policy(DynamicFieldPolicy::Strict) + .build(); + let engine = Engine::new(storage, schema).await.unwrap(); + + // Docs 0-2 are fine; doc 3 carries an undeclared field, which the + // Strict policy rejects in apply_dynamic_schema (before any WAL / + // store mutation for that doc). + let mut docs: Vec<_> = (0..3) + .map(|i| batch_entry(&format!("ok{i}"), &format!("title-{i}"))) + .collect(); + let mut bad = crate::data::Document::new(); + bad.fields.insert( + "undeclared".into(), + crate::data::DataValue::Text("boom".into()), + ); + docs.push(("bad".to_string(), bad)); + docs.push(batch_entry("never", "never-applied")); + + let err = engine + .put_documents(docs) + .await + .expect_err("Strict policy must fail the batch at doc 3"); + match err { + crate::error::LaurusError::BatchIngest { + failed_index, + failed_id, + applied, + .. + } => { + assert_eq!(failed_index, 3); + assert_eq!(failed_id, "bad"); + assert_eq!(applied, 3); + } + other => panic!("expected BatchIngest, got: {other}"), + } + assert!( + !engine.log.wal_is_dirty(), + "the applied prefix must be flushed durable on the error path" + ); + + // The applied prefix must be live (NRT) — doc 4 must not have been + // attempted. + for i in 0..3 { + let docs = engine.get_documents(&format!("ok{i}")).await.unwrap(); + assert_eq!(docs.len(), 1, "applied doc ok{i} must stay visible"); + } + assert!( + engine.get_documents("never").await.unwrap().is_empty(), + "docs after the failing one must not be applied" + ); + } + + /// #551: under the default `PerRecord` policy a batch of N docs must fsync + /// the WAL exactly once (deferred-fsync scope), not once per record, and + /// leave nothing unsynced. + #[tokio::test] + async fn test_put_documents_single_wal_fsync_per_batch() { + let engine = title_only_engine().await; + + // Prime the WAL writer so the sync counter exists, then baseline it. + engine + .put_document("prime", batch_entry("prime", "prime").1) + .await + .unwrap(); + let baseline = engine.log.wal_sync_count(); + + let docs: Vec<_> = (0..50) + .map(|i| batch_entry(&format!("id{i}"), &format!("title-{i}"))) + .collect(); + engine.put_documents(docs).await.unwrap(); + + assert_eq!( + engine.log.wal_sync_count(), + baseline + 1, + "a 50-doc batch must amortize to exactly one WAL fsync" + ); + assert!( + !engine.log.wal_is_dirty(), + "the batch-end flush must leave no unsynced WAL bytes" + ); + } + + /// #551: singular writes acknowledged while a **concurrent batch** holds + /// the WAL sync-deferral scope must keep their per-record durability — + /// the deferral flag is global, so without the explicit re-assertion in + /// `put_document` / `add_document` / `delete_documents` their records + /// would be left unsynced until the batch ends. + #[tokio::test] + async fn test_singular_writes_stay_durable_during_concurrent_batch_deferral() { + let engine = title_only_engine().await; + engine + .put_document("seed", batch_entry("seed", "seed").1) + .await + .unwrap(); + + // Simulate an in-flight put_documents batch on another task. + let _foreign_batch = engine.log.defer_sync(); + + engine + .put_document("a", batch_entry("a", "a").1) + .await + .unwrap(); + assert!( + !engine.log.wal_is_dirty(), + "a singular put must be fsync'd before ack even during a batch" + ); + + engine + .add_document("b", batch_entry("b", "b").1) + .await + .unwrap(); + assert!( + !engine.log.wal_is_dirty(), + "a singular add must be fsync'd before ack even during a batch" + ); + + engine.delete_documents("seed").await.unwrap(); + assert!( + !engine.log.wal_is_dirty(), + "a singular delete must be fsync'd before ack even during a batch" + ); + } + /// #828: updating many docs that are still in the uncommitted buffer must /// stay correct under the deferred in-memory-index rebuild. Each update goes /// through `delete_documents` → `delete_document(old_buffered_id)` → diff --git a/laurus/src/error.rs b/laurus/src/error.rs index ebb3191e..6269abb9 100644 --- a/laurus/src/error.rs +++ b/laurus/src/error.rs @@ -94,6 +94,31 @@ pub enum LaurusError { #[error("Incompatible format: {0}")] IncompatibleFormat(String), + /// A batch ingestion call failed partway through (fail-fast semantics). + /// + /// Batch ingestion applies documents sequentially and stops at the first + /// failure. The `applied` documents that preceded the failure remain in + /// the WAL and NRT buffers — they are searchable, become durable at the + /// next commit, and are replayed on crash recovery. There is no batch + /// rollback; retrying the batch (or its suffix starting at + /// `failed_index`) is idempotent under put semantics. With fail-fast + /// sequential processing `applied == failed_index` always holds; both are + /// carried so callers get the count without knowing that invariant. + #[error( + "batch ingest failed at doc {failed_index} (id '{failed_id}') after {applied} documents were applied: {source}" + )] + BatchIngest { + /// Zero-based position of the failing document in the input batch. + failed_index: usize, + /// External id of the failing document. + failed_id: String, + /// Number of documents successfully applied before the failure. + applied: usize, + /// The underlying error that failed the document at `failed_index`. + #[source] + source: Box, + }, + /// JSON serialization/deserialization errors #[error("JSON error: {0}")] Json(#[from] serde_json::Error), diff --git a/laurus/src/store/log.rs b/laurus/src/store/log.rs index 8cf15aa0..18ffb73d 100644 --- a/laurus/src/store/log.rs +++ b/laurus/src/store/log.rs @@ -51,7 +51,7 @@ use std::io::{Read, Write}; use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::time::Duration; use parking_lot::{Mutex, RwLock}; @@ -250,6 +250,11 @@ struct WalWriterState { /// (framed length, including the CRC for v2). Drives the /// [`WalSyncPolicy::Group`] byte threshold; reset to 0 on flush. unsynced_bytes: usize, + /// Test-only: number of `flush_and_sync()` calls performed on this writer. + /// Lets tests assert fsync amortization exactly (e.g. the batch-ingest + /// path syncing once per batch instead of once per record). + #[cfg(test)] + sync_count: usize, } /// Unified document log providing WAL, doc_id generation, and document storage. @@ -277,6 +282,12 @@ pub struct DocumentLog { /// [`WalSyncPolicy::PerRecord`]; [`WalSyncPolicy::Group`] defers the fsync to /// amortize it over a batch. sync_policy: WalSyncPolicy, + /// Number of active [`WalSyncDeferral`] scopes (see [`Self::defer_sync`]). + /// While non-zero, the per-append fsync of [`WalSyncPolicy::PerRecord`] is + /// suppressed so a batch-ingest caller can amortize one fsync over the + /// whole batch via [`Self::flush_wal`]. [`WalSyncPolicy::Group`] thresholds + /// are unaffected (its loss-window contract is preserved mid-batch). + sync_deferral_depth: AtomicUsize, } impl DocumentLog { @@ -352,6 +363,7 @@ impl DocumentLog { next_seq: AtomicU64::new(1), doc_store: RwLock::new(doc_store), sync_policy, + sync_deferral_depth: AtomicUsize::new(0), }) } @@ -423,6 +435,8 @@ impl DocumentLog { dirty: false, unsynced_records: 0, unsynced_bytes: 0, + #[cfg(test)] + sync_count: 0, }); Ok(()) } @@ -456,7 +470,7 @@ impl DocumentLog { }, }; - Self::write_record(&mut writer_guard, &record, self.sync_policy)?; + self.write_record(&mut writer_guard, &record)?; Ok((doc_id, seq)) } @@ -481,7 +495,7 @@ impl DocumentLog { }, }; - Self::write_record(&mut writer_guard, &record, self.sync_policy)?; + self.write_record(&mut writer_guard, &record)?; Ok(seq) } @@ -603,25 +617,34 @@ impl DocumentLog { state.dirty = false; state.unsynced_records = 0; state.unsynced_bytes = 0; + #[cfg(test)] + { + state.sync_count += 1; + } } Ok(()) } - /// Append a record and fsync according to `policy`. + /// Append a record and fsync according to the effective sync policy. /// /// Under [`WalSyncPolicy::PerRecord`] the record is always fsync'd before - /// returning (per-record durability — the default contract). Under - /// [`WalSyncPolicy::Group`] the fsync is deferred and only happens once the - /// batch reaches `max_records` records or `max_bytes` bytes since the last - /// sync, amortizing one fsync over the batch; [`Self::flush_wal`] (called at - /// commit) forces any trailing partial batch durable. - fn write_record( - state: &mut Option, - record: &LogRecord, - policy: WalSyncPolicy, - ) -> Result<()> { + /// returning (per-record durability — the default contract), unless a + /// [`WalSyncDeferral`] scope is active (see [`Self::defer_sync`]), in which + /// case the fsync is left to the scope owner's batch-end + /// [`Self::flush_wal`]. Under [`WalSyncPolicy::Group`] the fsync is + /// deferred and only happens once the batch reaches `max_records` records + /// or `max_bytes` bytes since the last sync, amortizing one fsync over the + /// batch — deferral scopes do not suppress these threshold flushes, so the + /// Group policy's bounded loss window holds even mid-batch; + /// [`Self::flush_wal`] (called at commit) forces any trailing partial + /// batch durable. + fn write_record(&self, state: &mut Option, record: &LogRecord) -> Result<()> { Self::append_record_bytes(state, record)?; - if Self::batch_ready(state, policy) { + let flush_now = match self.sync_policy { + WalSyncPolicy::PerRecord => self.sync_deferral_depth.load(Ordering::Acquire) == 0, + group @ WalSyncPolicy::Group { .. } => Self::batch_ready(state, group), + }; + if flush_now { Self::flush_writer(state)?; } Ok(()) @@ -664,6 +687,53 @@ impl DocumentLog { Self::flush_writer(&mut writer_guard) } + /// Re-assert the per-record durability contract after a singular append. + /// + /// Under [`WalSyncPolicy::PerRecord`] this forces any appended-but-unsynced + /// WAL bytes durable. It is a no-op when the append already self-synced + /// (the common case — the dirty guard skips the fsync), but it is the + /// load-bearing fsync when a **concurrent batch** holds a + /// [`Self::defer_sync`] scope: the scope suppresses the per-record fsync + /// globally, so a singular write acknowledged during a batch would + /// otherwise silently lose its durability guarantee. Singular write entry + /// points call this before returning. Under [`WalSyncPolicy::Group`] it is + /// a no-op by design — that policy's bounded loss window applies to + /// singular writes too. + /// + /// # Errors + /// + /// Returns an error if flushing or fsyncing the open WAL writer fails. + pub fn ensure_per_record_durability(&self) -> Result<()> { + match self.sync_policy { + WalSyncPolicy::PerRecord => self.flush_wal(), + WalSyncPolicy::Group { .. } => Ok(()), + } + } + + /// Open a WAL sync-deferral scope for a batch of appends. + /// + /// While the returned [`WalSyncDeferral`] guard is alive, appends under + /// [`WalSyncPolicy::PerRecord`] skip their per-record fsync; the scope + /// owner is responsible for calling [`Self::flush_wal`] once at batch end + /// (on both success and error paths) so the whole batch becomes durable + /// with a single fsync. Under [`WalSyncPolicy::Group`] the scope changes + /// nothing: the group thresholds keep firing, preserving that policy's + /// bounded loss window. + /// + /// Scopes may nest (e.g. concurrent batches on different tasks); the + /// per-record fsync resumes once every guard has been dropped. Dropping + /// the guard does **not** flush — records appended inside an abandoned + /// scope stay buffered until the next append, [`Self::flush_wal`], or + /// commit makes them durable. + /// + /// # Returns + /// + /// An RAII guard that re-enables per-record fsync when dropped. + pub fn defer_sync(&self) -> WalSyncDeferral<'_> { + self.sync_deferral_depth.fetch_add(1, Ordering::AcqRel); + WalSyncDeferral { log: self } + } + /// Test-only: whether the open writer has appended-but-unsynced bytes. /// /// Lets tests observe that a deferred (group-commit) batch has been flushed @@ -677,6 +747,20 @@ impl DocumentLog { .is_some_and(|state| state.dirty) } + /// Test-only: number of fsyncs performed by the currently open WAL writer. + /// + /// The counter belongs to the open [`WalWriterState`] and therefore resets + /// when the writer is recreated (e.g. after [`Self::truncate`]). Used to + /// assert fsync amortization exactly — e.g. that a deferred batch of N + /// appends syncs once, not N times. + #[cfg(test)] + pub(crate) fn wal_sync_count(&self) -> usize { + self.wal_writer + .lock() + .as_ref() + .map_or(0, |state| state.sync_count) + } + /// Read all records from the WAL. /// /// Also updates internal counters (`next_seq`, `next_doc_id`) to be @@ -914,6 +998,23 @@ impl DocumentLog { } } +/// RAII guard for a WAL sync-deferral scope (see [`DocumentLog::defer_sync`]). +/// +/// While alive, appends under [`WalSyncPolicy::PerRecord`] skip their +/// per-record fsync so a batch caller can amortize one fsync over the whole +/// batch. Dropping the guard only re-enables per-record fsync; it does not +/// flush — the scope owner must call [`DocumentLog::flush_wal`] at batch end. +#[derive(Debug)] +pub struct WalSyncDeferral<'a> { + log: &'a DocumentLog, +} + +impl Drop for WalSyncDeferral<'_> { + fn drop(&mut self) { + self.log.sync_deferral_depth.fetch_sub(1, Ordering::AcqRel); + } +} + #[cfg(test)] mod tests { use super::*; @@ -1150,6 +1251,80 @@ mod tests { assert_eq!(records[0].seq, 1); } + /// #551: while a `defer_sync` scope is alive, per-record appends skip + /// their fsync; the batch-end `flush_wal` makes everything durable with a + /// single sync, and the records round-trip. + #[test] + fn defer_sync_suppresses_per_record_fsync_until_flush() { + let log = make_log(); + + let deferral = log.defer_sync(); + for i in 0..5 { + log.append(&format!("ext_{i}"), small_doc()).unwrap(); + } + assert!( + log.wal_is_dirty(), + "deferred appends must stay unsynced until the batch-end flush" + ); + assert_eq!( + log.wal_sync_count(), + 0, + "no per-record fsync may run inside the deferral scope" + ); + + drop(deferral); + log.flush_wal().unwrap(); + assert!(!log.wal_is_dirty()); + assert_eq!( + log.wal_sync_count(), + 1, + "the whole batch must amortize to exactly one fsync" + ); + assert_eq!(log.read_all().unwrap().len(), 5); + } + + /// #551: dropping the deferral guard restores the per-record contract — + /// the next append self-syncs (including any bytes left over from an + /// abandoned scope). + #[test] + fn defer_sync_drop_restores_per_record_fsync() { + let log = make_log(); + + { + let _deferral = log.defer_sync(); + log.append("ext_0", small_doc()).unwrap(); + assert!(log.wal_is_dirty()); + // Abandoned scope: no flush_wal before drop. + } + + log.append("ext_1", small_doc()).unwrap(); + assert!( + !log.wal_is_dirty(), + "the first append after the scope must sync itself and the leftover bytes" + ); + assert_eq!(log.read_all().unwrap().len(), 2); + } + + /// #551: a deferral scope does not suppress the `Group` policy's batch + /// thresholds — its bounded loss window holds even mid-batch. + #[test] + fn defer_sync_keeps_group_thresholds_firing() { + let log = make_group_log(2, usize::MAX); + + let _deferral = log.defer_sync(); + log.append("ext_0", small_doc()).unwrap(); + assert!( + log.wal_is_dirty(), + "below the record threshold the group batch stays unsynced" + ); + log.append("ext_1", small_doc()).unwrap(); + assert!( + !log.wal_is_dirty(), + "the group record threshold must fire despite the deferral scope" + ); + assert_eq!(log.wal_sync_count(), 1); + } + /// The default policy is per-record, and `group_with_defaults` uses the /// documented batch thresholds with no flush timer (Issue #542, Phase 4). #[test] diff --git a/laurus/tests/engine_batch_ingest_test.rs b/laurus/tests/engine_batch_ingest_test.rs new file mode 100644 index 00000000..23bbcb25 --- /dev/null +++ b/laurus/tests/engine_batch_ingest_test.rs @@ -0,0 +1,168 @@ +//! Integration tests for the batch ingestion API (#551): +//! [`Engine::put_documents`] / [`Engine::add_documents`] durability across an +//! uncommitted reopen. +//! +//! The batch call defers the per-record WAL fsync and flushes once at batch +//! end, so an acknowledged batch — even without a commit — must survive a +//! process restart exactly like the equivalent singular puts, and a +//! fail-fast batch must recover exactly its applied prefix. + +use std::sync::Arc; + +use laurus::lexical::TextOption; +use laurus::storage::Storage; +use laurus::storage::memory::{MemoryStorage, MemoryStorageConfig}; +use laurus::{DataValue, Document, Engine, FieldOption, LaurusError, Schema}; + +/// Build a `(id, doc)` batch entry with a single `title` text field. +fn batch_entry(id: &str, title: &str) -> (String, Document) { + let doc = Document::builder() + .add_field("title", DataValue::Text(title.into())) + .build(); + (id.to_string(), doc) +} + +/// A schema with a single `title` text field. +fn title_schema() -> Schema { + Schema::builder() + .add_field("title", FieldOption::Text(TextOption::default())) + .build() +} + +/// An acknowledged-but-uncommitted batch must be fully durable: reopening the +/// engine on the same storage replays every batched doc from the WAL. +#[tokio::test(flavor = "multi_thread")] +async fn test_put_documents_uncommitted_batch_recovers_after_reopen() -> laurus::Result<()> { + let storage: Arc = Arc::new(MemoryStorage::new(MemoryStorageConfig::default())); + let schema = title_schema(); + + // Round 1: batch-ingest WITHOUT commit, then drop the engine. + { + let engine = Engine::new(storage.clone(), schema.clone()).await?; + let docs: Vec<_> = (0..25) + .map(|i| batch_entry(&format!("id{i}"), &format!("title-{i}"))) + .collect(); + engine.put_documents(docs).await?; + // Drop without commit — durability must come from the batch-end + // WAL flush alone. + } + + // Round 2: reopen on the SAME storage; recovery replays the WAL. + { + let engine = Engine::new(storage.clone(), schema.clone()).await?; + engine.commit().await?; + + let stats = engine.stats()?; + assert_eq!( + stats.document_count, 25, + "every batched doc must be replayed from the WAL after reopen" + ); + for i in 0..25 { + let docs = engine.get_documents(&format!("id{i}")).await?; + assert_eq!(docs.len(), 1, "id{i} must survive the reopen"); + } + } + + Ok(()) +} + +/// A fail-fast batch recovers exactly its applied prefix: the docs before the +/// failing one are durable (batch-end flush runs on the error path too), the +/// failing doc and its successors never existed. +#[tokio::test(flavor = "multi_thread")] +async fn test_put_documents_failed_batch_recovers_applied_prefix() -> laurus::Result<()> { + use laurus::DynamicFieldPolicy; + + let storage: Arc = Arc::new(MemoryStorage::new(MemoryStorageConfig::default())); + let schema = Schema::builder() + .add_field("title", FieldOption::Text(TextOption::default())) + .dynamic_field_policy(DynamicFieldPolicy::Strict) + .build(); + + // Round 1: batch fails at position 3 (undeclared field under Strict); + // drop the engine without commit. + { + let engine = Engine::new(storage.clone(), schema.clone()).await?; + let mut docs: Vec<_> = (0..3) + .map(|i| batch_entry(&format!("ok{i}"), &format!("title-{i}"))) + .collect(); + docs.push(( + "bad".to_string(), + Document::builder() + .add_field("undeclared", DataValue::Text("boom".into())) + .build(), + )); + docs.push(batch_entry("never", "never-applied")); + + let err = engine + .put_documents(docs) + .await + .expect_err("Strict policy must fail the batch"); + assert!( + matches!( + err, + LaurusError::BatchIngest { + failed_index: 3, + applied: 3, + .. + } + ), + "expected BatchIngest at index 3, got: {err}" + ); + } + + // Round 2: reopen — exactly the applied prefix must be recovered. + { + let engine = Engine::new(storage.clone(), schema.clone()).await?; + engine.commit().await?; + + assert_eq!( + engine.stats()?.document_count, + 3, + "exactly the applied prefix must survive the reopen" + ); + for i in 0..3 { + let docs = engine.get_documents(&format!("ok{i}")).await?; + assert_eq!(docs.len(), 1, "applied doc ok{i} must be recovered"); + } + assert!( + engine.get_documents("bad").await?.is_empty(), + "the failing doc must not exist after recovery" + ); + assert!( + engine.get_documents("never").await?.is_empty(), + "docs after the failing one must not exist after recovery" + ); + } + + Ok(()) +} + +/// `add_documents` chunks sharing one external id survive an uncommitted +/// reopen as distinct chunks (no delete-first on replay). +#[tokio::test(flavor = "multi_thread")] +async fn test_add_documents_chunks_recover_after_reopen() -> laurus::Result<()> { + let storage: Arc = Arc::new(MemoryStorage::new(MemoryStorageConfig::default())); + let schema = title_schema(); + + { + let engine = Engine::new(storage.clone(), schema.clone()).await?; + let docs: Vec<_> = (0..4) + .map(|i| batch_entry("doc", &format!("chunk-{i}"))) + .collect(); + engine.add_documents(docs).await?; + } + + { + let engine = Engine::new(storage.clone(), schema.clone()).await?; + engine.commit().await?; + let chunks = engine.get_documents("doc").await?; + assert_eq!( + chunks.len(), + 4, + "all four chunks must be replayed as chunks, not deduped" + ); + } + + Ok(()) +}