Test coverage for search embeddings cache: repository and service layer (#1593)

* Prepared repository testss plumbing for search cache

* Richer embeddings cache test suite

* Tests for EmbeddingsProvider logic

* changelog

Author: zaychenko-sergei

CHANGELOG.md

@@ -11,6 +11,10 @@ Recommendation: for ease of reading, use the following order:
 - Fixed
 -->
 
+## [Unreleased]
+### Changed
+- Test coverage for search embeddings cache: repository and service layer
+
 ## [0.259.1] - 2026-02-12
 ### Fixed
 - Sync odf schema to accept odf format

Cargo.lock

@@ -82,9 +82,10 @@ members = [
     "src/infra/search/openai",
     "src/infra/search/elasticsearch",
     "src/infra/search/elasticsearch-macros",
-    "src/infra/search/cache/inmem",
-    "src/infra/search/cache/postgres",
-    "src/infra/search/cache/sqlite",
+    "src/infra/search/cache-repo-tests",
+    "src/infra/search/cache-inmem",
+    "src/infra/search/cache-postgres",
+    "src/infra/search/cache-sqlite",
     ## Task System
     "src/infra/task-system/repo-tests",
     "src/infra/task-system/inmem",
@@ -232,9 +233,10 @@ kamu-datasets-repo-tests = { version = "0.259.1", path = "src/infra/datasets/rep
 kamu-search-openai = { version = "0.259.1", path = "src/infra/search/openai", default-features = false }
 kamu-search-elasticsearch = { version = "0.259.1", path = "src/infra/search/elasticsearch", default-features = false }
 kamu-search-elasticsearch-macros = { version = "0.259.1", path = "src/infra/search/elasticsearch-macros", default-features = false }
-kamu-search-cache-inmem = { version = "0.259.1", path = "src/infra/search/cache/inmem", default-features = false }
-kamu-search-cache-postgres = { version = "0.259.1", path = "src/infra/search/cache/postgres", default-features = false }
-kamu-search-cache-sqlite = { version = "0.259.1", path = "src/infra/search/cache/sqlite", default-features = false }
+kamu-search-cache-inmem = { version = "0.259.1", path = "src/infra/search/cache-inmem", default-features = false }
+kamu-search-cache-postgres = { version = "0.259.1", path = "src/infra/search/cache-postgres", default-features = false }
+kamu-search-cache-sqlite = { version = "0.259.1", path = "src/infra/search/cache-sqlite", default-features = false }
+kamu-search-cache-repo-tests = { version = "0.259.1", path = "src/infra/search/cache-repo-tests", default-features = false }
 ## Task System
 kamu-task-system-inmem = { version = "0.259.1", path = "src/infra/task-system/inmem", default-features = false }
 kamu-task-system-postgres = { version = "0.259.1", path = "src/infra/task-system/postgres", default-features = false }

Cargo.toml

@@ -11,7 +11,7 @@ POSTGRES_CRATES := \
 	./src/infra/datasets/postgres \
 	./src/infra/flow-system/postgres \
 	./src/infra/messaging-outbox/postgres \
-	./src/infra/search/cache/postgres \
+	./src/infra/search/cache-postgres \
 	./src/infra/task-system/postgres \
 	./src/infra/webhooks/postgres
 
@@ -27,7 +27,7 @@ SQLITE_CRATES := \
 	./src/infra/datasets/sqlite \
 	./src/infra/flow-system/sqlite \
 	./src/infra/messaging-outbox/sqlite \
-	./src/infra/search/cache/sqlite \
+	./src/infra/search/cache-sqlite \
 	./src/infra/task-system/sqlite \
 	./src/infra/webhooks/sqlite
 

Makefile

@@ -890,6 +890,7 @@ class "WsSmartTransferProtocolClient"
 "EmbeddingsProviderImpl"  --> "EmbeddingsCacheRepository"
 "EmbeddingsProviderImpl"  --> "EmbeddingsChunker"
 "EmbeddingsProviderImpl"  --> "EmbeddingsEncoder"
+"EmbeddingsProviderImpl"  --> "SystemTimeSource"
 "TaskEventStore" <|-- "SqliteTaskEventStore"
 "SqliteTaskEventStore"  --> "TransactionRefT"
 "WebhookDeliveryRepository" <|-- "SqliteWebhookDeliveryRepository"

resources/di.puml

@@ -34,6 +34,7 @@ internal-error = { workspace = true }
 kamu-core = { workspace = true }
 kamu-search = { workspace = true }
 odf = { workspace = true }
+time-source = { workspace = true }
 
 chrono = { version = "0.4", default-features = false }
 dill = { version = "0.15", default-features = false }
@@ -47,3 +48,8 @@ tracing = { version = "0.1", default-features = false }
 tokio = { version = "1", default-features = false }
 
 [dev-dependencies]
+kamu-search-cache-inmem = { workspace = true }
+time-source = { workspace = true }
+
+test-log = { version = "0.2", features = ["trace"] }
+tokio = { version = "1", default-features = false, features = ["rt", "macros"] }

src/domain/search/services/Cargo.toml

@@ -10,9 +10,9 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use chrono::Utc;
 use internal_error::*;
 use kamu_search::*;
+use time_source::SystemTimeSource;
 use tokio::sync::OnceCell;
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -21,6 +21,7 @@ pub struct EmbeddingsProviderImpl {
     embeddings_chunker: Arc<dyn EmbeddingsChunker>,
     embeddings_encoder: Arc<dyn EmbeddingsEncoder>,
     embeddings_cache_repo: Arc<dyn EmbeddingsCacheRepository>,
+    time_source: Arc<dyn SystemTimeSource>,
     cached_model: OnceCell<EmbeddingModelRow>,
 }
 
@@ -34,11 +35,13 @@ impl EmbeddingsProviderImpl {
         embeddings_chunker: Arc<dyn EmbeddingsChunker>,
         embeddings_encoder: Arc<dyn EmbeddingsEncoder>,
         embeddings_cache_repo: Arc<dyn EmbeddingsCacheRepository>,
+        time_source: Arc<dyn SystemTimeSource>,
     ) -> Self {
         Self {
             embeddings_chunker,
             embeddings_encoder,
             embeddings_cache_repo,
+            time_source,
             cached_model: OnceCell::new(),
         }
     }
@@ -145,12 +148,18 @@ impl EmbeddingsProvider for EmbeddingsProviderImpl {
 
         tracing::debug!(chunk_count = chunks.len(), "Content chunking completed");
 
-        // 3) Normalize chunks
+        // 3) Normalize chunks and filter out empty ones
         let normalized_chunks: Vec<String> = chunks
             .iter()
             .map(|s| Self::normalize_input_chunk(s))
+            .filter(|s| !s.is_empty())
             .collect();
 
+        if normalized_chunks.is_empty() {
+            tracing::warn!("No non-empty chunks after normalization");
+            return Ok(vec![]);
+        }
+
         tracing::debug!(
             normalized_chunk_count = normalized_chunks.len(),
             "Chunk normalization completed"
@@ -232,7 +241,7 @@ impl EmbeddingsProvider for EmbeddingsProviderImpl {
 
         // 8) Touch stats
         self.embeddings_cache_repo
-            .touch_embeddings(&cache_keys, Utc::now())
+            .touch_embeddings(&cache_keys, self.time_source.now())
             .await
             .int_err()?;
 
@@ -320,7 +329,7 @@ impl EmbeddingsProvider for EmbeddingsProviderImpl {
 
         // 6) Touch stats
         self.embeddings_cache_repo
-            .touch_embeddings(&[cache_key], Utc::now())
+            .touch_embeddings(&[cache_key], self.time_source.now())
             .await
             .int_err()?;
 

src/domain/search/services/src/embeddings_provider_impl.rs

@@ -0,0 +1,10 @@
+// Copyright Kamu Data, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+mod tests;

src/domain/search/services/tests/mod.rs

@@ -0,0 +1,10 @@
+// Copyright Kamu Data, Inc. and contributors. All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+mod test_embeddings_provider_impl;