1use std::path::Path;
2use std::sync::Arc;
3use std::time::Duration;
4use std::{
5 collections::hash_map::DefaultHasher,
6 collections::{HashMap, HashSet},
7 hash::{Hash, Hasher},
8};
9
10use crate::error::KnowledgeResult;
11use async_trait::async_trait;
12use rusqlite::ToSql;
13use rusqlite::{Connection, OpenFlags};
14use wp_log::{info_ctrl, warn_kdb};
15use wp_model_core::model::DataField;
16
17use crate::cache::CacheAble;
18use crate::loader::{
19 DEFAULT_SQLDB_NAME, ProviderKind, SqlProviderKind, SqlProviderSpec, parse_knowdb_conf,
20};
21use crate::mem::RowData;
22use crate::mem::memdb::MemDB;
23use crate::mem::thread_clone::ThreadClonedMDB;
24use crate::mysql::{MySqlProvider, MySqlProviderConfig};
25use crate::param::named_params_to_fields;
26use crate::postgres::{PostgresProvider, PostgresProviderConfig};
27use crate::runtime::{
28 CachePolicy, CachedRedisValue, DatasourceId, Generation, MetadataCacheScope, ProviderExecutor,
29 QueryRequest, QueryResponse, RedisCacheKey, RedisCmdTag, RuntimeSnapshot, runtime,
30};
31use crate::telemetry::{
32 CacheLayer, CacheOutcome, CacheTelemetryEvent, KnowledgeTelemetry, install_telemetry,
33 telemetry, telemetry_enabled,
34};
35
36struct MemProvider {
37 db: MemDB,
38 metadata_scope: MetadataCacheScope,
39}
40
41#[async_trait]
42impl ProviderExecutor for ThreadClonedMDB {
43 fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
44 ThreadClonedMDB::query_with_scope(self, sql)
45 }
46
47 fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
48 ThreadClonedMDB::query_fields_with_scope(self, sql, params)
49 }
50
51 fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
52 ThreadClonedMDB::query_row_with_scope(self, sql)
53 }
54
55 fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
56 ThreadClonedMDB::query_named_fields_with_scope(self, sql, params)
57 }
58}
59
60#[async_trait]
61impl ProviderExecutor for MemProvider {
62 fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
63 self.db.query_with_scope(&self.metadata_scope, sql)
64 }
65
66 fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
67 self.db
68 .query_fields_with_scope(&self.metadata_scope, sql, params)
69 }
70
71 fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
72 self.db.query_row_with_scope(&self.metadata_scope, sql)
73 }
74
75 fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
76 self.db
77 .query_named_fields_with_scope(&self.metadata_scope, sql, params)
78 }
79}
80
81#[async_trait]
82impl ProviderExecutor for PostgresProvider {
83 fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
84 PostgresProvider::query(self, sql)
85 }
86
87 fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
88 PostgresProvider::query_fields(self, sql, params)
89 }
90
91 fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
92 PostgresProvider::query_row(self, sql)
93 }
94
95 fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
96 PostgresProvider::query_named_fields(self, sql, params)
97 }
98
99 async fn query_async(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
100 PostgresProvider::query_async(self, sql).await
101 }
102
103 async fn query_fields_async(
104 &self,
105 sql: &str,
106 params: &[DataField],
107 ) -> KnowledgeResult<Vec<RowData>> {
108 PostgresProvider::query_fields_async(self, sql, params).await
109 }
110
111 async fn query_row_async(&self, sql: &str) -> KnowledgeResult<RowData> {
112 PostgresProvider::query_row_async(self, sql).await
113 }
114
115 async fn query_named_fields_async(
116 &self,
117 sql: &str,
118 params: &[DataField],
119 ) -> KnowledgeResult<RowData> {
120 PostgresProvider::query_named_fields_async(self, sql, params).await
121 }
122}
123
124#[async_trait]
125impl ProviderExecutor for MySqlProvider {
126 fn query(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
127 MySqlProvider::query(self, sql)
128 }
129
130 fn query_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<Vec<RowData>> {
131 MySqlProvider::query_fields(self, sql, params)
132 }
133
134 fn query_row(&self, sql: &str) -> KnowledgeResult<RowData> {
135 MySqlProvider::query_row(self, sql)
136 }
137
138 fn query_named_fields(&self, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
139 MySqlProvider::query_named_fields(self, sql, params)
140 }
141
142 async fn query_async(&self, sql: &str) -> KnowledgeResult<Vec<RowData>> {
143 MySqlProvider::query_async(self, sql).await
144 }
145
146 async fn query_fields_async(
147 &self,
148 sql: &str,
149 params: &[DataField],
150 ) -> KnowledgeResult<Vec<RowData>> {
151 MySqlProvider::query_fields_async(self, sql, params).await
152 }
153
154 async fn query_row_async(&self, sql: &str) -> KnowledgeResult<RowData> {
155 MySqlProvider::query_row_async(self, sql).await
156 }
157
158 async fn query_named_fields_async(
159 &self,
160 sql: &str,
161 params: &[DataField],
162 ) -> KnowledgeResult<RowData> {
163 MySqlProvider::query_named_fields_async(self, sql, params).await
164 }
165}
166
167fn install_provider<F>(
168 kind: ProviderKind,
169 datasource_id: DatasourceId,
170 build: F,
171) -> KnowledgeResult<()>
172where
173 F: FnOnce(Generation) -> KnowledgeResult<Arc<dyn ProviderExecutor>>,
174{
175 runtime().install_provider(kind, datasource_id, build)?;
176 Ok(())
177}
178
179fn datasource_id_for(kind: ProviderKind, seed: &str) -> DatasourceId {
180 DatasourceId::from_seed(kind, seed)
181}
182
183pub fn init_thread_cloned_from_authority(authority_uri: &str) -> KnowledgeResult<()> {
184 let datasource_id = datasource_id_for(ProviderKind::SqliteAuthority, authority_uri);
185 install_provider(ProviderKind::SqliteAuthority, datasource_id, |generation| {
186 Ok(Arc::new(ThreadClonedMDB::from_authority_with_scope(
187 authority_uri,
188 datasource_id_for(ProviderKind::SqliteAuthority, authority_uri),
189 generation.0,
190 )))
191 })
192}
193
194pub fn init_mem_provider(memdb: MemDB) -> KnowledgeResult<()> {
195 install_provider(
196 ProviderKind::SqliteAuthority,
197 datasource_id_for(ProviderKind::SqliteAuthority, "memdb"),
198 |generation| {
199 Ok(Arc::new(MemProvider {
200 db: memdb,
201 metadata_scope: MetadataCacheScope {
202 datasource_id: datasource_id_for(ProviderKind::SqliteAuthority, "memdb"),
203 generation,
204 },
205 }))
206 },
207 )
208}
209
210pub fn init_postgres_provider(connection_uri: &str, pool_size: Option<u32>) -> KnowledgeResult<()> {
211 init_postgres_provider_with_config(
212 PostgresProviderConfig::new(connection_uri).with_pool_size(pool_size),
213 )
214}
215
216pub fn init_postgres_provider_with_config(config: PostgresProviderConfig) -> KnowledgeResult<()> {
217 let connection_uri = config.connection_uri().to_string();
218 install_postgres_provider(DEFAULT_SQLDB_NAME, config, &connection_uri, true)
219}
220
221pub fn init_postgres_provider_named(
223 name: &str,
224 config: PostgresProviderConfig,
225) -> KnowledgeResult<()> {
226 install_postgres_provider(name, config, name, false)
227}
228
229fn install_postgres_provider(
230 name: &str,
231 config: PostgresProviderConfig,
232 datasource_seed: &str,
233 is_default: bool,
234) -> KnowledgeResult<()> {
235 let datasource_id = datasource_id_for(ProviderKind::Postgres, datasource_seed);
236 runtime().install_provider_named(
237 name,
238 ProviderKind::Postgres,
239 datasource_id.clone(),
240 |generation| {
241 let provider = PostgresProvider::connect(
242 &config,
243 MetadataCacheScope {
244 datasource_id: datasource_id.clone(),
245 generation,
246 },
247 )?;
248 Ok(Arc::new(provider))
249 },
250 is_default,
251 )?;
252 Ok(())
253}
254
255pub fn init_mysql_provider(connection_uri: &str, pool_size: Option<u32>) -> KnowledgeResult<()> {
256 init_mysql_provider_with_config(
257 MySqlProviderConfig::new(connection_uri).with_pool_size(pool_size),
258 )
259}
260
261pub fn init_mysql_provider_with_config(config: MySqlProviderConfig) -> KnowledgeResult<()> {
262 let connection_uri = config.connection_uri().to_string();
263 install_mysql_provider(DEFAULT_SQLDB_NAME, config, &connection_uri, true)
264}
265
266pub fn init_mysql_provider_named(name: &str, config: MySqlProviderConfig) -> KnowledgeResult<()> {
268 install_mysql_provider(name, config, name, false)
269}
270
271fn install_mysql_provider(
272 name: &str,
273 config: MySqlProviderConfig,
274 datasource_seed: &str,
275 is_default: bool,
276) -> KnowledgeResult<()> {
277 let datasource_id = datasource_id_for(ProviderKind::Mysql, datasource_seed);
278 runtime().install_provider_named(
279 name,
280 ProviderKind::Mysql,
281 datasource_id.clone(),
282 |generation| {
283 let provider = MySqlProvider::connect(
284 &config,
285 MetadataCacheScope {
286 datasource_id: datasource_id.clone(),
287 generation,
288 },
289 )?;
290 Ok(Arc::new(provider))
291 },
292 is_default,
293 )?;
294 Ok(())
295}
296
297fn build_postgres_config(spec: &SqlProviderSpec) -> PostgresProviderConfig {
298 PostgresProviderConfig::new(spec.connection_uri.clone())
299 .with_pool_size(spec.pool_size)
300 .with_min_connections(spec.min_connections)
301 .with_acquire_timeout_ms(spec.acquire_timeout_ms)
302 .with_idle_timeout_ms(spec.idle_timeout_ms)
303 .with_max_lifetime_ms(spec.max_lifetime_ms)
304}
305
306fn build_mysql_config(spec: &SqlProviderSpec) -> MySqlProviderConfig {
307 MySqlProviderConfig::new(spec.connection_uri.clone())
308 .with_pool_size(spec.pool_size)
309 .with_min_connections(spec.min_connections)
310 .with_acquire_timeout_ms(spec.acquire_timeout_ms)
311 .with_idle_timeout_ms(spec.idle_timeout_ms)
312 .with_max_lifetime_ms(spec.max_lifetime_ms)
313}
314
315pub fn redis_bf_exists(key: &str, item: &str) -> KnowledgeResult<bool> {
324 if let Some(generation) = current_generation() {
325 let ck = redis_cache_key(RedisCmdTag::BfExists, generation, key, &[item]);
326 if let Some(CachedRedisValue::Bool(v)) = runtime().redis_cache_get(&ck) {
327 return Ok(v);
328 }
329 let result = crate::redis::bf_exists("knowdb", key, item)?;
330 runtime().redis_cache_put(ck, CachedRedisValue::Bool(result));
331 return Ok(result);
332 }
333 crate::redis::bf_exists("knowdb", key, item)
334}
335
336pub fn redis_hget(key: &str, field: &str) -> KnowledgeResult<Option<String>> {
340 if let Some(generation) = current_generation() {
341 let ck = redis_cache_key(RedisCmdTag::HGet, generation, key, &[field]);
342 if let Some(CachedRedisValue::OptString(ref v)) = runtime().redis_cache_get(&ck) {
343 return Ok(v.clone());
344 }
345 let result = crate::redis::hget("knowdb", key, field)?;
346 runtime().redis_cache_put(ck, CachedRedisValue::OptString(result.clone()));
347 return Ok(result);
348 }
349 crate::redis::hget("knowdb", key, field)
350}
351
352pub fn redis_get(key: &str) -> KnowledgeResult<Option<String>> {
356 if let Some(generation) = current_generation() {
357 let ck = redis_cache_key(RedisCmdTag::Get, generation, key, &[]);
358 if let Some(CachedRedisValue::OptString(ref v)) = runtime().redis_cache_get(&ck) {
359 return Ok(v.clone());
360 }
361 let result = crate::redis::get("knowdb", key)?;
362 runtime().redis_cache_put(ck, CachedRedisValue::OptString(result.clone()));
363 return Ok(result);
364 }
365 crate::redis::get("knowdb", key)
366}
367
368pub fn redis_set_exists(key: &str, member: &str) -> KnowledgeResult<bool> {
370 if let Some(generation) = current_generation() {
371 let ck = redis_cache_key(RedisCmdTag::SetExists, generation, key, &[member]);
372 if let Some(CachedRedisValue::Bool(v)) = runtime().redis_cache_get(&ck) {
373 return Ok(v);
374 }
375 let result = crate::redis::set_exists("knowdb", key, member)?;
376 runtime().redis_cache_put(ck, CachedRedisValue::Bool(result));
377 return Ok(result);
378 }
379 crate::redis::set_exists("knowdb", key, member)
380}
381
382pub fn redis_bf_add(key: &str, items: &[&str]) -> KnowledgeResult<Vec<bool>> {
387 let owned: Vec<String> = items.iter().map(|s| s.to_string()).collect();
388 crate::redis::bf_madd("knowdb", key, &owned)
389}
390
391pub fn redis_bf_create(key: &str, error_rate: f64, capacity: i64) -> KnowledgeResult<()> {
397 crate::redis::bf_reserve("knowdb", key, error_rate, capacity)
398}
399
400pub(crate) fn register_fun_map(map: HashMap<String, crate::loader::FunSpec>) {
406 crate::fun::register_fun_map(map);
407}
408
409pub fn external_exists(service: &str, arg: &str) -> KnowledgeResult<bool> {
411 let spec = crate::fun::resolve_spec(service, true)?;
412 let redis_key = spec.key.as_deref().unwrap_or(arg);
413 if spec.enabled()
414 && let Some(generation) = current_generation()
415 {
416 let ck = redis_cache_key(RedisCmdTag::BfExists, generation, redis_key, &[arg]);
417 if let Some(CachedRedisValue::Bool(v)) = runtime().redis_cache_get(&ck) {
418 return Ok(v);
419 }
420 let result = crate::fun::external_exists(service, arg)?;
421 if let Some(ttl) = spec.ttl_ms {
422 runtime().redis_cache_put_with_ttl(ck, CachedRedisValue::Bool(result), ttl);
423 } else {
424 runtime().redis_cache_put(ck, CachedRedisValue::Bool(result));
425 }
426 return Ok(result);
427 }
428 crate::fun::external_exists(service, arg)
429}
430
431pub fn external_value(service: &str, arg: &str) -> KnowledgeResult<Option<String>> {
433 let spec = crate::fun::resolve_spec(service, false)?;
434 let redis_key = spec.key.as_deref().unwrap_or(arg);
435 if spec.enabled()
436 && let Some(generation) = current_generation()
437 {
438 let ck = redis_cache_key(RedisCmdTag::Get, generation, redis_key, &[arg]);
439 if let Some(CachedRedisValue::OptString(ref v)) = runtime().redis_cache_get(&ck) {
440 return Ok(v.clone());
441 }
442 let result = crate::fun::external_value(service, arg)?;
443 if let Some(ttl) = spec.ttl_ms {
444 runtime().redis_cache_put_with_ttl(
445 ck,
446 CachedRedisValue::OptString(result.clone()),
447 ttl,
448 );
449 } else {
450 runtime().redis_cache_put(ck, CachedRedisValue::OptString(result.clone()));
451 }
452 return Ok(result);
453 }
454 crate::fun::external_value(service, arg)
455}
456
457#[allow(dead_code)]
458pub(crate) fn redis_ping() -> KnowledgeResult<bool> {
459 crate::redis::ping_blocking("knowdb")
460}
461
462#[allow(dead_code)]
463pub(crate) fn redis_close() -> KnowledgeResult<()> {
464 crate::redis::close(Some("knowdb"))
465}
466
467fn redis_cache_key(cmd: RedisCmdTag, generation: u64, key: &str, args: &[&str]) -> RedisCacheKey {
468 use std::collections::hash_map::DefaultHasher;
469 use std::hash::{Hash, Hasher};
470 let mut hasher = DefaultHasher::new();
471 key.hash(&mut hasher);
472 let key_hash = hasher.finish();
473 let mut hasher = DefaultHasher::new();
474 for arg in args {
475 arg.hash(&mut hasher);
476 }
477 let args_hash = hasher.finish();
478 RedisCacheKey {
479 generation,
480 cmd_tag: cmd,
481 key_hash,
482 args_hash,
483 }
484}
485
486pub fn current_generation() -> Option<u64> {
489 runtime()
490 .current_generation()
491 .map(|generation| generation.0)
492}
493
494pub fn runtime_snapshot() -> RuntimeSnapshot {
495 runtime().snapshot()
496}
497
498pub fn install_runtime_telemetry(
499 telemetry_impl: Arc<dyn KnowledgeTelemetry>,
500) -> Arc<dyn KnowledgeTelemetry> {
501 install_telemetry(telemetry_impl)
502}
503
504pub use crate::sql_route::route_provider_sql;
505
506pub fn provider_exists(name: &str) -> bool {
508 runtime().provider_exists(name)
509}
510
511pub fn provider_names() -> Vec<String> {
513 runtime().provider_names()
514}
515
516pub fn query_for(name: &str, sql: &str) -> KnowledgeResult<Vec<RowData>> {
518 runtime()
519 .execute_for(
520 name,
521 &QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass),
522 )
523 .map(QueryResponse::into_rows)
524}
525
526pub async fn query_async_for(name: &str, sql: &str) -> KnowledgeResult<Vec<RowData>> {
528 runtime()
529 .execute_for_async(
530 name,
531 &QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass),
532 )
533 .await
534 .map(QueryResponse::into_rows)
535}
536
537pub fn query_fields_for(name: &str, sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
539 runtime().execute_first_row_fields_for(name, sql, params, CachePolicy::Bypass)
540}
541
542pub async fn query_fields_for_async(
544 name: &str,
545 sql: &str,
546 params: &[DataField],
547) -> KnowledgeResult<RowData> {
548 runtime()
549 .execute_first_row_fields_for_async(name, sql, params, CachePolicy::Bypass)
550 .await
551}
552
553pub fn query(sql: &str) -> KnowledgeResult<Vec<RowData>> {
554 runtime()
555 .execute(&QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass))
556 .map(QueryResponse::into_rows)
557}
558
559pub async fn query_async(sql: &str) -> KnowledgeResult<Vec<RowData>> {
560 runtime()
561 .execute_async(&QueryRequest::many(sql, Vec::new(), CachePolicy::Bypass))
562 .await
563 .map(QueryResponse::into_rows)
564}
565
566pub fn query_row(sql: &str) -> KnowledgeResult<RowData> {
567 runtime()
568 .execute(&QueryRequest::first_row(
569 sql,
570 Vec::new(),
571 CachePolicy::Bypass,
572 ))
573 .map(QueryResponse::into_row)
574}
575
576pub async fn query_row_async(sql: &str) -> KnowledgeResult<RowData> {
577 runtime()
578 .execute_async(&QueryRequest::first_row(
579 sql,
580 Vec::new(),
581 CachePolicy::Bypass,
582 ))
583 .await
584 .map(QueryResponse::into_row)
585}
586
587pub fn query_fields(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
588 runtime().execute_first_row_fields(sql, params, CachePolicy::Bypass)
589}
590
591pub async fn query_fields_async(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
592 runtime()
593 .execute_first_row_fields_async(sql, params, CachePolicy::Bypass)
594 .await
595}
596
597pub fn query_named_fields(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
598 query_fields(sql, params)
599}
600
601pub async fn query_named_fields_async(sql: &str, params: &[DataField]) -> KnowledgeResult<RowData> {
602 query_fields_async(sql, params).await
603}
604
605pub fn query_named<'a>(
606 sql: &str,
607 params: &'a [(&'a str, &'a dyn ToSql)],
608) -> KnowledgeResult<RowData> {
609 let fields = named_params_to_fields(params)?;
610 query_fields(sql, &fields)
611}
612
613pub fn cache_query_fields<const N: usize>(
614 sql: &str,
615 c_params: &[DataField; N],
616 query_params: &[DataField],
617 cache: &mut impl CacheAble<DataField, RowData, N>,
618) -> RowData {
619 cache_query_fields_with_scope(sql, stable_hash(sql), c_params, query_params, cache)
620}
621
622pub fn cache_query_fields_with_scope<const N: usize>(
623 sql: &str,
624 local_cache_scope: u64,
625 c_params: &[DataField; N],
626 query_params: &[DataField],
627 cache: &mut impl CacheAble<DataField, RowData, N>,
628) -> RowData {
629 let generation = current_generation();
630 let provider_kind = runtime().current_provider_kind();
631 if let Some(hit) = local_cache_hit(
632 local_cache_scope,
633 c_params,
634 cache,
635 generation,
636 provider_kind,
637 ) {
638 return hit;
639 }
640 match runtime().execute_first_row_fields(sql, query_params, CachePolicy::UseGlobal) {
641 Ok(row) => {
642 cache.save_scoped(local_cache_scope, c_params, row.clone());
643 row
644 }
645 Err(err) => {
646 warn_kdb!("[kdb] query error: {}", err);
647 Vec::new()
648 }
649 }
650}
651
652pub fn cache_query_fields_route<const N: usize>(
654 provider: Option<&str>,
655 sql: &str,
656 local_cache_scope: u64,
657 c_params: &[DataField; N],
658 query_params: &[DataField],
659 cache: &mut impl CacheAble<DataField, RowData, N>,
660) -> RowData {
661 let Some(name) = provider else {
662 return cache_query_fields_with_scope(
663 sql,
664 local_cache_scope,
665 c_params,
666 query_params,
667 cache,
668 );
669 };
670 let handle = match runtime().provider_by_name(name) {
671 Ok(handle) => handle,
672 Err(err) => {
673 warn_kdb!("[kdb] named provider '{name}' unavailable: {}", err);
674 return Vec::new();
675 }
676 };
677 let generation = Some(handle.generation.0);
678 let provider_kind = Some(handle.kind.clone());
679 let scope = local_cache_scope ^ stable_hash(name);
682 if let Some(hit) = local_cache_hit(scope, c_params, cache, generation, provider_kind) {
683 return hit;
684 }
685 match runtime().execute_first_row_fields_for(name, sql, query_params, CachePolicy::UseGlobal) {
686 Ok(row) => {
687 cache.save_scoped(scope, c_params, row.clone());
688 row
689 }
690 Err(err) => {
691 warn_kdb!("[kdb] query error: {}", err);
692 Vec::new()
693 }
694 }
695}
696
697fn local_cache_hit<const N: usize>(
699 local_cache_scope: u64,
700 c_params: &[DataField; N],
701 cache: &mut impl CacheAble<DataField, RowData, N>,
702 generation: Option<u64>,
703 provider_kind: Option<ProviderKind>,
704) -> Option<RowData> {
705 if let Some(generation) = generation {
706 cache.prepare_generation(local_cache_scope, generation);
707 }
708 if let Some(hit) = cache.fetch_scoped(local_cache_scope, c_params) {
709 runtime().record_local_cache_hit();
710 if telemetry_enabled() {
711 telemetry().on_cache(&CacheTelemetryEvent {
712 layer: CacheLayer::Local,
713 outcome: CacheOutcome::Hit,
714 provider_kind,
715 });
716 }
717 return Some(hit.clone());
718 }
719 runtime().record_local_cache_miss();
720 if telemetry_enabled() {
721 telemetry().on_cache(&CacheTelemetryEvent {
722 layer: CacheLayer::Local,
723 outcome: CacheOutcome::Miss,
724 provider_kind,
725 });
726 }
727 None
728}
729
730pub async fn cache_query_fields_async<const N: usize>(
731 sql: &str,
732 c_params: &[DataField; N],
733 query_params: &[DataField],
734 cache: &mut impl CacheAble<DataField, RowData, N>,
735) -> RowData {
736 cache_query_fields_async_with_scope(
737 sql,
738 stable_hash(sql),
739 c_params,
740 || query_params.to_vec(),
741 cache,
742 )
743 .await
744}
745
746pub async fn cache_query_fields_async_with<const N: usize, F>(
747 sql: &str,
748 c_params: &[DataField; N],
749 build_query_params: F,
750 cache: &mut impl CacheAble<DataField, RowData, N>,
751) -> RowData
752where
753 F: FnOnce() -> Vec<DataField>,
754{
755 cache_query_fields_async_with_scope(sql, stable_hash(sql), c_params, build_query_params, cache)
756 .await
757}
758
759pub async fn cache_query_fields_async_with_scope<const N: usize, F>(
760 sql: &str,
761 local_cache_scope: u64,
762 c_params: &[DataField; N],
763 build_query_params: F,
764 cache: &mut impl CacheAble<DataField, RowData, N>,
765) -> RowData
766where
767 F: FnOnce() -> Vec<DataField>,
768{
769 let generation = current_generation();
770 let provider_kind = runtime().current_provider_kind();
771 if let Some(hit) = local_cache_hit(
772 local_cache_scope,
773 c_params,
774 cache,
775 generation,
776 provider_kind,
777 ) {
778 return hit;
779 }
780 let query_params = build_query_params();
781 match runtime()
782 .execute_first_row_fields_async(sql, &query_params, CachePolicy::UseGlobal)
783 .await
784 {
785 Ok(row) => {
786 cache.save_scoped(local_cache_scope, c_params, row.clone());
787 row
788 }
789 Err(err) => {
790 warn_kdb!("[kdb] query error: {}", err);
791 Vec::new()
792 }
793 }
794}
795
796pub async fn cache_query_fields_route_async<const N: usize, F>(
798 provider: Option<&str>,
799 sql: &str,
800 local_cache_scope: u64,
801 c_params: &[DataField; N],
802 build_query_params: F,
803 cache: &mut impl CacheAble<DataField, RowData, N>,
804) -> RowData
805where
806 F: FnOnce() -> Vec<DataField>,
807{
808 let Some(name) = provider else {
809 return cache_query_fields_async_with_scope(
810 sql,
811 local_cache_scope,
812 c_params,
813 build_query_params,
814 cache,
815 )
816 .await;
817 };
818 let handle = match runtime().provider_by_name(name) {
819 Ok(handle) => handle,
820 Err(err) => {
821 warn_kdb!("[kdb] named provider '{name}' unavailable: {}", err);
822 return Vec::new();
823 }
824 };
825 let generation = Some(handle.generation.0);
826 let provider_kind = Some(handle.kind.clone());
827 let scope = local_cache_scope ^ stable_hash(name);
829 if let Some(hit) = local_cache_hit(scope, c_params, cache, generation, provider_kind) {
830 return hit;
831 }
832 let query_params = build_query_params();
833 match runtime()
834 .execute_first_row_fields_for_async(name, sql, &query_params, CachePolicy::UseGlobal)
835 .await
836 {
837 Ok(row) => {
838 cache.save_scoped(scope, c_params, row.clone());
839 row
840 }
841 Err(err) => {
842 warn_kdb!("[kdb] query error: {}", err);
843 Vec::new()
844 }
845 }
846}
847
848pub fn cache_query<const N: usize>(
849 sql: &str,
850 c_params: &[DataField; N],
851 named_params: &[(&str, &dyn ToSql)],
852 cache: &mut impl CacheAble<DataField, RowData, N>,
853) -> RowData {
854 let query_fields = match named_params_to_fields(named_params) {
855 Ok(fields) => fields,
856 Err(err) => {
857 warn_kdb!("[kdb] query param conversion error: {}", err);
858 return Vec::new();
859 }
860 };
861
862 cache_query_fields(sql, c_params, &query_fields, cache)
863}
864
865pub async fn cache_query_async<const N: usize>(
866 sql: &str,
867 c_params: &[DataField; N],
868 named_params: &[(&str, &dyn ToSql)],
869 cache: &mut impl CacheAble<DataField, RowData, N>,
870) -> RowData {
871 let query_fields = match named_params_to_fields(named_params) {
872 Ok(fields) => fields,
873 Err(err) => {
874 warn_kdb!("[kdb] query param conversion error: {}", err);
875 return Vec::new();
876 }
877 };
878
879 cache_query_fields_async(sql, c_params, &query_fields, cache).await
880}
881
882fn ensure_wal(authority_uri: &str) -> KnowledgeResult<()> {
883 if let Ok(conn) = Connection::open_with_flags(
884 authority_uri,
885 OpenFlags::SQLITE_OPEN_READ_WRITE
886 | OpenFlags::SQLITE_OPEN_CREATE
887 | OpenFlags::SQLITE_OPEN_URI,
888 ) {
889 let _ = conn.execute_batch(
890 "PRAGMA journal_mode=WAL;\nPRAGMA synchronous=NORMAL;\nPRAGMA temp_store=MEMORY;",
891 );
892 }
893 Ok(())
894}
895
896pub fn init_wal_pool_from_authority(authority_uri: &str, pool_size: u32) -> KnowledgeResult<()> {
897 ensure_wal(authority_uri)?;
898 let flags = OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI;
899 let mem = MemDB::new_file(authority_uri, pool_size, flags)?;
900 init_mem_provider(mem)
901}
902
903pub fn init_thread_cloned_from_knowdb(
904 root: &Path,
905 knowdb_conf: &Path,
906 authority_uri: &str,
907 dict: &orion_variate::EnvDict,
908) -> KnowledgeResult<()> {
909 let (conf, conf_abs, _) = parse_knowdb_conf(root, knowdb_conf, dict)?;
910 if let Some(provider_cfg) = conf.provider() {
911 if let Some(specs) = provider_cfg.sqldb
913 && !specs.is_empty()
914 {
915 let expected = SqlProviderSpec::validate_specs(&specs)?;
916 let default_name = if expected.contains(DEFAULT_SQLDB_NAME) {
918 DEFAULT_SQLDB_NAME.to_string()
919 } else {
920 specs[0].effective_name().to_string()
921 };
922 for spec in &specs {
926 let name = spec.effective_name().to_string();
927 let is_default = name == default_name;
928 let seed = match &spec.name {
930 Some(_) => name.clone(),
931 None => spec.connection_uri.clone(),
932 };
933 match spec.kind {
934 SqlProviderKind::Postgres => {
935 info_ctrl!(
936 "init postgres knowdb provider({name}:{}) ",
937 conf_abs.display(),
938 );
939 install_postgres_provider(
940 &name,
941 build_postgres_config(spec),
942 &seed,
943 is_default,
944 )?;
945 }
946 SqlProviderKind::Mysql => {
947 info_ctrl!("init mysql knowdb provider({name}:{}) ", conf_abs.display(),);
948 install_mysql_provider(&name, build_mysql_config(spec), &seed, is_default)?;
949 }
950 }
951 }
952 runtime().configure_result_cache(
953 conf.cache.enabled,
954 conf.cache.capacity,
955 Duration::from_millis(conf.cache.ttl_ms.max(1)),
956 );
957 runtime().prune_named_providers_except(&expected);
958 return Ok(());
959 }
960 if let Some(redis_cfg) = provider_cfg.redis {
962 info_ctrl!("init redis knowdb provider({}) ", conf_abs.display(),);
963 crate::redis::init_with_opts(
964 "knowdb",
965 &redis_cfg.connection_uri,
966 redis_cfg.pool_size,
967 redis_cfg.command_timeout_ms,
968 )?;
969 runtime().configure_redis_cache(conf.cache.enabled, conf.cache.capacity);
970 register_fun_map(conf.fun.clone());
971 runtime().prune_named_providers_except(&HashSet::new());
973 return Ok(());
974 }
975 }
976
977 crate::loader::build_authority_from_knowdb(root, knowdb_conf, authority_uri, dict)?;
978 let ro_uri = if let Some(rest) = authority_uri.strip_prefix("file:") {
979 let path_part = rest.split('?').next().unwrap_or(rest);
980 format!("file:{}?mode=ro&uri=true", path_part)
981 } else {
982 authority_uri.to_string()
983 };
984
985 info_ctrl!("init authority knowdb success({}) ", knowdb_conf.display(),);
986 init_thread_cloned_from_authority(&ro_uri)?;
987 runtime().configure_result_cache(
988 conf.cache.enabled,
989 conf.cache.capacity,
990 Duration::from_millis(conf.cache.ttl_ms.max(1)),
991 );
992 runtime().prune_named_providers_except(&HashSet::from([DEFAULT_SQLDB_NAME.to_string()]));
994 Ok(())
995}
996
997fn stable_hash(value: &str) -> u64 {
998 let mut hasher = DefaultHasher::new();
999 value.hash(&mut hasher);
1000 hasher.finish()
1001}
1002
1003#[cfg(test)]
1004mod tests {
1005 use super::*;
1006 use crate::cache::FieldQueryCache;
1007 use crate::error::KnowReason;
1008 use crate::mem::memdb::MemDB;
1009 use crate::mem::query_util::{COLNAME_CACHE, metadata_cache_key_for_scope};
1010 use crate::runtime::fields_to_params;
1011 use crate::telemetry::{
1012 CacheLayer, CacheTelemetryEvent, KnowledgeTelemetry, QueryTelemetryEvent,
1013 ReloadTelemetryEvent, reset_telemetry,
1014 };
1015 use orion_error::conversion::ToStructError;
1016 use orion_variate::EnvDict;
1017 use std::fs;
1018 use std::hint::black_box;
1019 use std::path::PathBuf;
1020 use std::sync::atomic::{AtomicU64, Ordering};
1021 use std::time::{Duration, Instant};
1022
1023 #[derive(Default)]
1024 struct TestTelemetry {
1025 reload_success: AtomicU64,
1026 reload_failure: AtomicU64,
1027 local_hits: AtomicU64,
1028 local_misses: AtomicU64,
1029 result_hits: AtomicU64,
1030 result_misses: AtomicU64,
1031 metadata_hits: AtomicU64,
1032 metadata_misses: AtomicU64,
1033 query_success: AtomicU64,
1034 query_failure: AtomicU64,
1035 }
1036
1037 impl KnowledgeTelemetry for TestTelemetry {
1038 fn on_cache(&self, event: &CacheTelemetryEvent) {
1039 let counter = match (event.layer, event.outcome) {
1040 (CacheLayer::Local, CacheOutcome::Hit) => &self.local_hits,
1041 (CacheLayer::Local, CacheOutcome::Miss) => &self.local_misses,
1042 (CacheLayer::Result, CacheOutcome::Hit) => &self.result_hits,
1043 (CacheLayer::Result, CacheOutcome::Miss) => &self.result_misses,
1044 (CacheLayer::Metadata, CacheOutcome::Hit) => &self.metadata_hits,
1045 (CacheLayer::Metadata, CacheOutcome::Miss) => &self.metadata_misses,
1046 };
1047 counter.fetch_add(1, Ordering::Relaxed);
1048 }
1049
1050 fn on_reload(&self, event: &ReloadTelemetryEvent) {
1051 let counter = match event.outcome {
1052 crate::telemetry::ReloadOutcome::Success => &self.reload_success,
1053 crate::telemetry::ReloadOutcome::Failure => &self.reload_failure,
1054 };
1055 counter.fetch_add(1, Ordering::Relaxed);
1056 }
1057
1058 fn on_query(&self, event: &QueryTelemetryEvent) {
1059 let counter = if event.success {
1060 &self.query_success
1061 } else {
1062 &self.query_failure
1063 };
1064 counter.fetch_add(1, Ordering::Relaxed);
1065 }
1066 }
1067
1068 fn perf_env_usize(key: &str, default: usize) -> usize {
1069 std::env::var(key)
1070 .ok()
1071 .and_then(|value| value.parse::<usize>().ok())
1072 .unwrap_or(default)
1073 }
1074
1075 fn seed_perf_provider(rows: usize) {
1076 let db = MemDB::instance();
1077 db.execute("CREATE TABLE perf_kv (id INTEGER PRIMARY KEY, value TEXT)")
1078 .expect("create perf_kv");
1079 db.execute("BEGIN IMMEDIATE").expect("begin perf_kv load");
1080 for id in 1..=rows {
1081 let sql = format!("INSERT INTO perf_kv (id, value) VALUES ({id}, 'value_{id}')");
1082 db.execute(sql.as_str()).expect("insert perf_kv row");
1083 }
1084 db.execute("COMMIT").expect("commit perf_kv load");
1085 init_mem_provider(db).expect("init perf provider");
1086 }
1087
1088 #[tokio::test(flavor = "current_thread")]
1089 async fn query_async_works_with_mem_provider() {
1090 let _guard = crate::runtime::runtime_test_guard().lock_async().await;
1091 let db = MemDB::instance();
1092 db.execute("CREATE TABLE async_kv (id INTEGER PRIMARY KEY, value TEXT)")
1093 .expect("create async_kv");
1094 db.execute("INSERT INTO async_kv (id, value) VALUES (1, 'hello')")
1095 .expect("insert async_kv row");
1096 init_mem_provider(db).expect("init mem provider");
1097
1098 let row = query_fields_async(
1099 "SELECT value FROM async_kv WHERE id=:id",
1100 &[DataField::from_digit(":id", 1)],
1101 )
1102 .await
1103 .expect("query async row");
1104 assert_eq!(row.len(), 1);
1105 assert_eq!(row[0].to_string(), "chars(hello)");
1106 }
1107
1108 #[derive(Clone)]
1109 struct PerfQuery {
1110 cache_key: [DataField; 1],
1111 query_params: [DataField; 1],
1112 bypass_req: QueryRequest,
1113 global_req: QueryRequest,
1114 }
1115
1116 fn build_perf_workload(ops: usize, hotset: usize) -> Vec<PerfQuery> {
1117 (0..ops)
1118 .map(|idx| {
1119 let id = ((idx * 17) % hotset + 1) as i64;
1120 let cache_key = [DataField::from_digit("id", id)];
1121 let query_params = [DataField::from_digit(":id", id)];
1122 let bypass_req = QueryRequest::first_row(
1123 "SELECT value FROM perf_kv WHERE id=:id",
1124 fields_to_params(&query_params),
1125 CachePolicy::Bypass,
1126 );
1127 let global_req = QueryRequest::first_row(
1128 "SELECT value FROM perf_kv WHERE id=:id",
1129 fields_to_params(&query_params),
1130 CachePolicy::UseGlobal,
1131 );
1132 PerfQuery {
1133 cache_key,
1134 query_params,
1135 bypass_req,
1136 global_req,
1137 }
1138 })
1139 .collect()
1140 }
1141
1142 #[derive(Debug, Clone, Copy)]
1143 struct PerfCounters {
1144 result_hits: u64,
1145 result_misses: u64,
1146 local_hits: u64,
1147 local_misses: u64,
1148 metadata_hits: u64,
1149 metadata_misses: u64,
1150 }
1151
1152 #[derive(Debug, Clone)]
1153 struct PerfResult {
1154 name: &'static str,
1155 elapsed: Duration,
1156 ops: usize,
1157 counters: PerfCounters,
1158 }
1159
1160 impl PerfResult {
1161 fn qps(&self) -> f64 {
1162 let secs = self.elapsed.as_secs_f64();
1163 if secs == 0.0 {
1164 self.ops as f64
1165 } else {
1166 self.ops as f64 / secs
1167 }
1168 }
1169 }
1170
1171 fn snapshot_delta(before: &RuntimeSnapshot, after: &RuntimeSnapshot) -> PerfCounters {
1172 PerfCounters {
1173 result_hits: after
1174 .result_cache_hits
1175 .saturating_sub(before.result_cache_hits),
1176 result_misses: after
1177 .result_cache_misses
1178 .saturating_sub(before.result_cache_misses),
1179 local_hits: after
1180 .local_cache_hits
1181 .saturating_sub(before.local_cache_hits),
1182 local_misses: after
1183 .local_cache_misses
1184 .saturating_sub(before.local_cache_misses),
1185 metadata_hits: after
1186 .metadata_cache_hits
1187 .saturating_sub(before.metadata_cache_hits),
1188 metadata_misses: after
1189 .metadata_cache_misses
1190 .saturating_sub(before.metadata_cache_misses),
1191 }
1192 }
1193
1194 fn run_bypass_perf(workload: &[PerfQuery], rows: usize) -> PerfResult {
1195 seed_perf_provider(rows);
1196 let before = runtime_snapshot();
1197 let started = Instant::now();
1198 for item in workload {
1199 let row = runtime()
1200 .execute(&item.bypass_req)
1201 .expect("execute bypass request")
1202 .into_row();
1203 black_box(row);
1204 }
1205 let elapsed = started.elapsed();
1206 let after = runtime_snapshot();
1207 PerfResult {
1208 name: "bypass",
1209 elapsed,
1210 ops: workload.len(),
1211 counters: snapshot_delta(&before, &after),
1212 }
1213 }
1214
1215 fn run_global_cache_perf(workload: &[PerfQuery], rows: usize) -> PerfResult {
1216 seed_perf_provider(rows);
1217 let before = runtime_snapshot();
1218 let started = Instant::now();
1219 for item in workload {
1220 let row = runtime()
1221 .execute(&item.global_req)
1222 .expect("execute global-cache request")
1223 .into_row();
1224 black_box(row);
1225 }
1226 let elapsed = started.elapsed();
1227 let after = runtime_snapshot();
1228 PerfResult {
1229 name: "global_cache",
1230 elapsed,
1231 ops: workload.len(),
1232 counters: snapshot_delta(&before, &after),
1233 }
1234 }
1235
1236 fn run_local_cache_perf(workload: &[PerfQuery], rows: usize) -> PerfResult {
1237 seed_perf_provider(rows);
1238 let mut cache = FieldQueryCache::with_capacity(workload.len().max(1));
1239 let before = runtime_snapshot();
1240 let started = Instant::now();
1241 for item in workload {
1242 let row = cache_query_fields(
1243 "SELECT value FROM perf_kv WHERE id=:id",
1244 &item.cache_key,
1245 &item.query_params,
1246 &mut cache,
1247 );
1248 black_box(row);
1249 }
1250 let elapsed = started.elapsed();
1251 let after = runtime_snapshot();
1252 PerfResult {
1253 name: "local_cache",
1254 elapsed,
1255 ops: workload.len(),
1256 counters: snapshot_delta(&before, &after),
1257 }
1258 }
1259
1260 fn print_perf_result(result: &PerfResult) {
1261 eprintln!(
1262 "[wp-knowledge][cache-perf] scenario={} elapsed_ms={} qps={:.0} result_hit={} result_miss={} local_hit={} local_miss={} metadata_hit={} metadata_miss={}",
1263 result.name,
1264 result.elapsed.as_millis(),
1265 result.qps(),
1266 result.counters.result_hits,
1267 result.counters.result_misses,
1268 result.counters.local_hits,
1269 result.counters.local_misses,
1270 result.counters.metadata_hits,
1271 result.counters.metadata_misses,
1272 );
1273 }
1274
1275 fn uniq_cache_cfg_tmp_dir() -> PathBuf {
1276 use rand::{Rng, rng};
1277 let rnd: u64 = rng().next_u64();
1278 std::env::temp_dir().join(format!("wpk_cache_cfg_{}", rnd))
1279 }
1280
1281 fn write_minimal_knowdb_with_cache(
1282 root: &Path,
1283 enabled: bool,
1284 capacity: usize,
1285 ttl_ms: u64,
1286 ) -> std::path::PathBuf {
1287 let models = root.join("models").join("knowledge");
1288 let example_dir = models.join("example");
1289 fs::create_dir_all(&example_dir).expect("create knowdb models/example");
1290 fs::write(
1291 models.join("knowdb.toml"),
1292 format!(
1293 r#"
1294version = 2
1295base_dir = "."
1296
1297[cache]
1298enabled = {enabled}
1299capacity = {capacity}
1300ttl_ms = {ttl_ms}
1301
1302[csv]
1303has_header = false
1304
1305[[tables]]
1306name = "example"
1307columns.by_index = [0,1]
1308"#
1309 ),
1310 )
1311 .expect("write knowdb.toml");
1312 fs::write(
1313 example_dir.join("create.sql"),
1314 r#"
1315CREATE TABLE IF NOT EXISTS {table} (
1316 id INTEGER PRIMARY KEY,
1317 value TEXT NOT NULL
1318);
1319"#,
1320 )
1321 .expect("write create.sql");
1322 fs::write(
1323 example_dir.join("insert.sql"),
1324 "INSERT INTO {table} (id, value) VALUES (?1, ?2);\n",
1325 )
1326 .expect("write insert.sql");
1327 fs::write(example_dir.join("data.csv"), "1,alpha\n").expect("write data.csv");
1328 models.join("knowdb.toml")
1329 }
1330
1331 fn write_provider_only_knowdb_with_cache(
1332 root: &Path,
1333 provider_kind: &str,
1334 connection_uri: &str,
1335 enabled: bool,
1336 capacity: usize,
1337 ttl_ms: u64,
1338 ) -> std::path::PathBuf {
1339 let models = root.join("models").join("knowledge");
1340 fs::create_dir_all(&models).expect("create knowdb models");
1341 fs::write(
1342 models.join("knowdb.toml"),
1343 format!(
1344 r#"
1345version = 2
1346base_dir = "."
1347
1348[cache]
1349enabled = {enabled}
1350capacity = {capacity}
1351ttl_ms = {ttl_ms}
1352
1353[provider.sqldb]
1354kind = "{provider_kind}"
1355connection_uri = "{connection_uri}"
1356"#
1357 ),
1358 )
1359 .expect("write provider knowdb.toml");
1360 models.join("knowdb.toml")
1361 }
1362
1363 fn restore_default_result_cache_config() {
1364 runtime().configure_result_cache(true, 1024, Duration::from_millis(30_000));
1365 }
1366
1367 #[test]
1368 fn provider_can_be_replaced() {
1369 let _guard = crate::runtime::runtime_test_guard()
1370 .lock()
1371 .expect("provider test guard");
1372 let db1 = MemDB::instance();
1373 db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1374 .expect("create table in db1");
1375 db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1376 .expect("seed db1");
1377 init_mem_provider(db1).expect("init provider db1");
1378 let row = query_row("SELECT value FROM t WHERE id = 1").expect("query db1");
1379 assert_eq!(row[0].to_string(), "chars(first)");
1380
1381 let db2 = MemDB::instance();
1382 db2.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1383 .expect("create table in db2");
1384 db2.execute("INSERT INTO t (id, value) VALUES (1, 'second')")
1385 .expect("seed db2");
1386 init_mem_provider(db2).expect("replace provider with db2");
1387 let row = query_row("SELECT value FROM t WHERE id = 1").expect("query db2");
1388 assert_eq!(row[0].to_string(), "chars(second)");
1389 }
1390
1391 #[test]
1392 fn sqlite_metadata_cache_uses_provider_scope_after_reload() {
1393 let _guard = crate::runtime::runtime_test_guard()
1394 .lock()
1395 .expect("provider test guard");
1396 COLNAME_CACHE.write().expect("metadata cache lock").clear();
1397
1398 let db = MemDB::instance();
1399 db.execute("CREATE TABLE cache_scope_t (id INTEGER PRIMARY KEY, value TEXT)")
1400 .expect("create table");
1401 db.execute("INSERT INTO cache_scope_t (id, value) VALUES (1, 'scope-old')")
1402 .expect("seed table");
1403
1404 let old_scope = MetadataCacheScope {
1405 datasource_id: DatasourceId("sqlite:old".to_string()),
1406 generation: Generation(1),
1407 };
1408 let new_scope = MetadataCacheScope {
1409 datasource_id: DatasourceId("sqlite:new".to_string()),
1410 generation: Generation(2),
1411 };
1412 let old_provider = MemProvider {
1413 db: db.clone(),
1414 metadata_scope: old_scope.clone(),
1415 };
1416
1417 install_provider(
1418 ProviderKind::SqliteAuthority,
1419 new_scope.datasource_id.clone(),
1420 |_generation| {
1421 Ok(Arc::new(MemProvider {
1422 db: db.clone(),
1423 metadata_scope: new_scope.clone(),
1424 }))
1425 },
1426 )
1427 .expect("install new provider");
1428
1429 let row = old_provider
1430 .query_row("SELECT value FROM cache_scope_t WHERE id = 1")
1431 .expect("old provider query");
1432 assert_eq!(row[0].to_string(), "chars(scope-old)");
1433
1434 let cache = COLNAME_CACHE.read().expect("metadata cache lock");
1435 assert!(cache.contains(&metadata_cache_key_for_scope(
1436 &old_scope,
1437 "SELECT value FROM cache_scope_t WHERE id = 1",
1438 )));
1439 assert!(!cache.contains(&metadata_cache_key_for_scope(
1440 &new_scope,
1441 "SELECT value FROM cache_scope_t WHERE id = 1",
1442 )));
1443 }
1444
1445 #[tokio::test(flavor = "current_thread")]
1446 async fn async_query_uses_runtime_bridge() {
1447 let _guard = crate::runtime::runtime_test_guard().lock_async().await;
1448 let db = MemDB::instance();
1449 db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1450 .expect("create table");
1451 db.execute("INSERT INTO t (id, value) VALUES (1, 'async-first')")
1452 .expect("seed table");
1453 init_mem_provider(db).expect("init provider");
1454
1455 let row = query_row_async("SELECT value FROM t WHERE id = 1")
1456 .await
1457 .expect("async query row");
1458 assert_eq!(row[0].to_string(), "chars(async-first)");
1459 }
1460
1461 #[tokio::test(flavor = "current_thread")]
1462 async fn async_cache_query_fields_hits_local_cache() {
1463 let _guard = crate::runtime::runtime_test_guard().lock_async().await;
1464 let db = MemDB::instance();
1465 db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1466 .expect("create table");
1467 db.execute("INSERT INTO t (id, value) VALUES (1, 'async-cache')")
1468 .expect("seed table");
1469 init_mem_provider(db).expect("init provider");
1470
1471 let key = [DataField::from_digit("id", 1)];
1472 let params = [DataField::from_digit(":id", 1)];
1473 let mut cache = FieldQueryCache::default();
1474
1475 let first = cache_query_fields_async(
1476 "SELECT value FROM t WHERE id=:id",
1477 &key,
1478 ¶ms,
1479 &mut cache,
1480 )
1481 .await;
1482 let second = cache_query_fields_async(
1483 "SELECT value FROM t WHERE id=:id",
1484 &key,
1485 ¶ms,
1486 &mut cache,
1487 )
1488 .await;
1489
1490 assert_eq!(first[0].to_string(), "chars(async-cache)");
1491 assert_eq!(second[0].to_string(), "chars(async-cache)");
1492 }
1493
1494 #[test]
1495 fn local_cache_is_cleared_when_generation_changes() {
1496 let _guard = crate::runtime::runtime_test_guard()
1497 .lock()
1498 .expect("provider test guard");
1499 let db1 = MemDB::instance();
1500 db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1501 .expect("create table in db1");
1502 db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1503 .expect("seed db1");
1504 init_mem_provider(db1).expect("init provider db1");
1505
1506 let key = [DataField::from_digit("id", 1)];
1507 let params = [DataField::from_digit(":id", 1)];
1508 let mut cache = FieldQueryCache::default();
1509 let row = cache_query_fields(
1510 "SELECT value FROM t WHERE id=:id",
1511 &key,
1512 ¶ms,
1513 &mut cache,
1514 );
1515 assert_eq!(row[0].to_string(), "chars(first)");
1516
1517 let db2 = MemDB::instance();
1518 db2.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1519 .expect("create table in db2");
1520 db2.execute("INSERT INTO t (id, value) VALUES (1, 'second')")
1521 .expect("seed db2");
1522 init_mem_provider(db2).expect("replace provider with db2");
1523
1524 let row = cache_query_fields(
1525 "SELECT value FROM t WHERE id=:id",
1526 &key,
1527 ¶ms,
1528 &mut cache,
1529 );
1530 assert_eq!(row[0].to_string(), "chars(second)");
1531 }
1532
1533 #[test]
1534 fn local_cache_is_scoped_by_sql_text() {
1535 let _guard = crate::runtime::runtime_test_guard()
1536 .lock()
1537 .expect("provider test guard");
1538 let db = MemDB::instance();
1539 db.execute("CREATE TABLE t1 (id INTEGER PRIMARY KEY, value TEXT)")
1540 .expect("create t1");
1541 db.execute("CREATE TABLE t2 (id INTEGER PRIMARY KEY, value TEXT)")
1542 .expect("create t2");
1543 db.execute("INSERT INTO t1 (id, value) VALUES (1, 'first')")
1544 .expect("seed t1");
1545 db.execute("INSERT INTO t2 (id, value) VALUES (1, 'second')")
1546 .expect("seed t2");
1547 init_mem_provider(db).expect("init provider");
1548
1549 let key = [DataField::from_digit("id", 1)];
1550 let params = [DataField::from_digit(":id", 1)];
1551 let mut cache = FieldQueryCache::default();
1552
1553 let row = cache_query_fields(
1554 "SELECT value FROM t1 WHERE id=:id",
1555 &key,
1556 ¶ms,
1557 &mut cache,
1558 );
1559 assert_eq!(row[0].to_string(), "chars(first)");
1560
1561 let row = cache_query_fields(
1562 "SELECT value FROM t2 WHERE id=:id",
1563 &key,
1564 ¶ms,
1565 &mut cache,
1566 );
1567 assert_eq!(row[0].to_string(), "chars(second)");
1568 }
1569
1570 #[test]
1571 fn runtime_snapshot_tracks_generation_and_cache_size() {
1572 let _guard = crate::runtime::runtime_test_guard()
1573 .lock()
1574 .expect("provider test guard");
1575 let db = MemDB::instance();
1576 db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1577 .expect("create table");
1578 db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1579 .expect("seed table");
1580 init_mem_provider(db).expect("init provider");
1581
1582 let mut cache = FieldQueryCache::default();
1583 let key = [DataField::from_digit("id", 1)];
1584 let params = [DataField::from_digit(":id", 1)];
1585 let row = cache_query_fields(
1586 "SELECT value FROM t WHERE id=:id",
1587 &key,
1588 ¶ms,
1589 &mut cache,
1590 );
1591 assert_eq!(row[0].to_string(), "chars(first)");
1592
1593 let snapshot = runtime_snapshot();
1594 assert!(matches!(
1595 snapshot.provider_kind,
1596 Some(ProviderKind::SqliteAuthority)
1597 ));
1598 assert!(snapshot.generation.is_some());
1599 assert!(snapshot.result_cache_len >= 1);
1600 assert!(snapshot.result_cache_capacity >= snapshot.result_cache_len);
1601 assert!(snapshot.metadata_cache_capacity >= snapshot.metadata_cache_len);
1602 assert!(snapshot.reload_successes >= 1);
1603 }
1604
1605 #[test]
1606 fn metadata_cache_is_scoped_by_generation() {
1607 let _guard = crate::runtime::runtime_test_guard()
1608 .lock()
1609 .expect("provider test guard");
1610 let sql = "SELECT value FROM t WHERE id = 1";
1611 let before = runtime_snapshot().metadata_cache_len;
1612
1613 let db1 = MemDB::instance();
1614 db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1615 .expect("create table in db1");
1616 db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1617 .expect("seed db1");
1618 init_mem_provider(db1).expect("init provider db1");
1619 let row = query_row(sql).expect("query db1");
1620 assert_eq!(row[0].to_string(), "chars(first)");
1621 let after_first = runtime_snapshot().metadata_cache_len;
1622 assert!(
1623 after_first > before,
1624 "metadata cache did not record first generation entry: before={before} after_first={after_first}"
1625 );
1626
1627 let db2 = MemDB::instance();
1628 db2.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1629 .expect("create table in db2");
1630 db2.execute("INSERT INTO t (id, value) VALUES (1, 'second')")
1631 .expect("seed db2");
1632 init_mem_provider(db2).expect("replace provider with db2");
1633 let row = query_row(sql).expect("query db2");
1634 assert_eq!(row[0].to_string(), "chars(second)");
1635 let after_second = runtime_snapshot().metadata_cache_len;
1636 assert!(
1637 after_second > after_first,
1638 "metadata cache did not keep a distinct generation entry: after_first={after_first} after_second={after_second}"
1639 );
1640 }
1641
1642 #[test]
1643 fn failed_provider_reload_keeps_previous_provider() {
1644 let _guard = crate::runtime::runtime_test_guard()
1645 .lock()
1646 .expect("provider test guard");
1647 let db1 = MemDB::instance();
1648 db1.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1649 .expect("create table in db1");
1650 db1.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1651 .expect("seed db1");
1652 init_mem_provider(db1).expect("init provider db1");
1653 let before_generation = current_generation();
1654
1655 let reload_err = install_provider(
1656 ProviderKind::SqliteAuthority,
1657 datasource_id_for(ProviderKind::SqliteAuthority, "reload-failure"),
1658 |_generation| {
1659 Err(KnowReason::from_logic()
1660 .to_err()
1661 .with_detail("expected reload failure"))
1662 },
1663 );
1664 assert!(reload_err.is_err());
1665
1666 let row = query_row("SELECT value FROM t WHERE id = 1").expect("query previous provider");
1667 assert_eq!(row[0].to_string(), "chars(first)");
1668 assert_eq!(current_generation(), before_generation);
1669 }
1670
1671 #[test]
1672 fn runtime_snapshot_records_cache_counters() {
1673 let _guard = crate::runtime::runtime_test_guard()
1674 .lock()
1675 .expect("provider test guard");
1676 let db = MemDB::instance();
1677 db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1678 .expect("create table");
1679 db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1680 .expect("seed table");
1681 init_mem_provider(db).expect("init provider");
1682
1683 let before = runtime_snapshot();
1684 let mut cache = FieldQueryCache::default();
1685 let key = [DataField::from_digit("id", 1)];
1686 let params = [DataField::from_digit(":id", 1)];
1687 let row = cache_query_fields(
1688 "SELECT value FROM t WHERE id=:id",
1689 &key,
1690 ¶ms,
1691 &mut cache,
1692 );
1693 assert_eq!(row[0].to_string(), "chars(first)");
1694 let row = cache_query_fields(
1695 "SELECT value FROM t WHERE id=:id",
1696 &key,
1697 ¶ms,
1698 &mut cache,
1699 );
1700 assert_eq!(row[0].to_string(), "chars(first)");
1701
1702 let after = runtime_snapshot();
1703 assert!(after.local_cache_hits > before.local_cache_hits);
1704 assert!(after.local_cache_misses > before.local_cache_misses);
1705 assert!(after.result_cache_misses > before.result_cache_misses);
1706 assert!(after.metadata_cache_misses > before.metadata_cache_misses);
1707 }
1708
1709 #[test]
1710 fn telemetry_receives_reload_cache_and_query_events() {
1711 let _guard = crate::runtime::runtime_test_guard()
1712 .lock()
1713 .expect("provider test guard");
1714 let telemetry_impl = Arc::new(TestTelemetry::default());
1715 let previous = install_runtime_telemetry(telemetry_impl.clone());
1716
1717 let db = MemDB::instance();
1718 db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1719 .expect("create table");
1720 db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1721 .expect("seed table");
1722 init_mem_provider(db).expect("init provider");
1723
1724 let mut cache = FieldQueryCache::default();
1725 let key = [DataField::from_digit("id", 1)];
1726 let params = [DataField::from_digit(":id", 1)];
1727 let row = cache_query_fields(
1728 "SELECT value FROM t WHERE id=:id",
1729 &key,
1730 ¶ms,
1731 &mut cache,
1732 );
1733 assert_eq!(row[0].to_string(), "chars(first)");
1734 let row = cache_query_fields(
1735 "SELECT value FROM t WHERE id=:id",
1736 &key,
1737 ¶ms,
1738 &mut cache,
1739 );
1740 assert_eq!(row[0].to_string(), "chars(first)");
1741
1742 let reload_err = install_provider(
1743 ProviderKind::SqliteAuthority,
1744 datasource_id_for(ProviderKind::SqliteAuthority, "telemetry-failure"),
1745 |_generation| {
1746 Err(KnowReason::from_logic()
1747 .to_err()
1748 .with_detail("expected telemetry reload failure"))
1749 },
1750 );
1751 assert!(reload_err.is_err());
1752
1753 install_runtime_telemetry(previous);
1754 reset_telemetry();
1755
1756 assert!(telemetry_impl.reload_success.load(Ordering::Relaxed) >= 1);
1757 assert!(telemetry_impl.reload_failure.load(Ordering::Relaxed) >= 1);
1758 assert!(telemetry_impl.local_hits.load(Ordering::Relaxed) >= 1);
1759 assert!(telemetry_impl.local_misses.load(Ordering::Relaxed) >= 1);
1760 assert!(telemetry_impl.result_misses.load(Ordering::Relaxed) >= 1);
1761 assert!(telemetry_impl.metadata_misses.load(Ordering::Relaxed) >= 1);
1762 assert!(telemetry_impl.query_success.load(Ordering::Relaxed) >= 1);
1763 }
1764
1765 #[test]
1766 #[ignore = "manual perf comparison; run with cargo test cache_perf_reports_cache_vs_no_cache -- --ignored --nocapture"]
1767 fn cache_perf_reports_cache_vs_no_cache() {
1768 let _guard = crate::runtime::runtime_test_guard()
1769 .lock()
1770 .expect("provider test guard");
1771 let rows = perf_env_usize("WP_KDB_PERF_ROWS", 10_000).max(1);
1772 let ops = perf_env_usize("WP_KDB_PERF_OPS", 120_000).max(1);
1773 let hotset = perf_env_usize("WP_KDB_PERF_HOTSET", 128).clamp(1, rows);
1774 let workload = build_perf_workload(ops, hotset);
1775
1776 eprintln!(
1777 "[wp-knowledge][cache-perf] rows={} ops={} hotset={} sql=SELECT value FROM perf_kv WHERE id=:id",
1778 rows, ops, hotset
1779 );
1780
1781 let bypass = run_bypass_perf(&workload, rows);
1782 let global = run_global_cache_perf(&workload, rows);
1783 let local = run_local_cache_perf(&workload, rows);
1784
1785 print_perf_result(&bypass);
1786 print_perf_result(&global);
1787 print_perf_result(&local);
1788
1789 eprintln!(
1790 "[wp-knowledge][cache-perf] speedup global_vs_bypass={:.2}x local_vs_bypass={:.2}x",
1791 bypass.elapsed.as_secs_f64() / global.elapsed.as_secs_f64(),
1792 bypass.elapsed.as_secs_f64() / local.elapsed.as_secs_f64(),
1793 );
1794
1795 assert_eq!(bypass.counters.result_hits, 0);
1796 assert_eq!(bypass.counters.result_misses, 0);
1797 assert_eq!(bypass.counters.local_hits, 0);
1798 assert_eq!(bypass.counters.local_misses, 0);
1799 assert!(global.counters.result_hits > 0);
1800 assert!(global.counters.result_misses > 0);
1801 assert_eq!(global.counters.local_hits, 0);
1802 assert_eq!(global.counters.local_misses, 0);
1803 assert!(local.counters.local_hits > 0);
1804 assert!(local.counters.local_misses > 0);
1805 assert!(local.counters.result_misses > 0);
1806 }
1807
1808 #[test]
1809 fn init_thread_cloned_from_knowdb_applies_result_cache_config() {
1810 let _guard = crate::runtime::runtime_test_guard()
1811 .lock()
1812 .expect("provider test guard");
1813 let root = uniq_cache_cfg_tmp_dir();
1814 let conf_path = write_minimal_knowdb_with_cache(&root, true, 7, 5);
1815 let auth_file = root.join(".run").join("authority.sqlite");
1816 fs::create_dir_all(auth_file.parent().expect("authority parent"))
1817 .expect("create authority parent");
1818 let authority_uri = format!("file:{}?mode=rwc&uri=true", auth_file.display());
1819
1820 init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default())
1821 .expect("init knowdb with cache config");
1822
1823 let snapshot = runtime_snapshot();
1824 assert!(snapshot.result_cache_enabled);
1825 assert_eq!(snapshot.result_cache_capacity, 7);
1826 assert_eq!(snapshot.result_cache_ttl_ms, 5);
1827
1828 let req = QueryRequest::first_row(
1829 "SELECT value FROM example WHERE id=:id",
1830 fields_to_params(&[DataField::from_digit(":id", 1)]),
1831 CachePolicy::UseGlobal,
1832 );
1833 let before = runtime_snapshot();
1834 let row = runtime()
1835 .execute(&req)
1836 .expect("first result-cache query")
1837 .into_row();
1838 assert_eq!(row[0].to_string(), "chars(alpha)");
1839 let row = runtime()
1840 .execute(&req)
1841 .expect("second result-cache query")
1842 .into_row();
1843 assert_eq!(row[0].to_string(), "chars(alpha)");
1844 std::thread::sleep(Duration::from_millis(12));
1845 let row = runtime()
1846 .execute(&req)
1847 .expect("expired result-cache query")
1848 .into_row();
1849 assert_eq!(row[0].to_string(), "chars(alpha)");
1850 let after = runtime_snapshot();
1851
1852 assert!(after.result_cache_hits > before.result_cache_hits);
1853 assert!(after.result_cache_misses >= before.result_cache_misses + 2);
1854
1855 restore_default_result_cache_config();
1856 let _ = fs::remove_dir_all(&root);
1857 }
1858
1859 #[test]
1860 fn disabled_result_cache_from_knowdb_config_forces_bypass() {
1861 let _guard = crate::runtime::runtime_test_guard()
1862 .lock()
1863 .expect("provider test guard");
1864 let root = uniq_cache_cfg_tmp_dir();
1865 let conf_path = write_minimal_knowdb_with_cache(&root, false, 3, 30_000);
1866 let auth_file = root.join(".run").join("authority.sqlite");
1867 fs::create_dir_all(auth_file.parent().expect("authority parent"))
1868 .expect("create authority parent");
1869 let authority_uri = format!("file:{}?mode=rwc&uri=true", auth_file.display());
1870
1871 init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default())
1872 .expect("init knowdb with cache disabled");
1873
1874 let snapshot = runtime_snapshot();
1875 assert!(!snapshot.result_cache_enabled);
1876 assert_eq!(snapshot.result_cache_capacity, 3);
1877 assert_eq!(snapshot.result_cache_ttl_ms, 30_000);
1878
1879 let req = QueryRequest::first_row(
1880 "SELECT value FROM example WHERE id=:id",
1881 fields_to_params(&[DataField::from_digit(":id", 1)]),
1882 CachePolicy::UseGlobal,
1883 );
1884 let before = runtime_snapshot();
1885 let _ = runtime()
1886 .execute(&req)
1887 .expect("first bypassed result-cache query");
1888 let _ = runtime()
1889 .execute(&req)
1890 .expect("second bypassed result-cache query");
1891 let after = runtime_snapshot();
1892
1893 assert_eq!(after.result_cache_hits, before.result_cache_hits);
1894 assert_eq!(after.result_cache_misses, before.result_cache_misses);
1895
1896 restore_default_result_cache_config();
1897 let _ = fs::remove_dir_all(&root);
1898 }
1899
1900 #[test]
1901 fn failed_knowdb_provider_init_does_not_apply_cache_config() {
1902 let _guard = crate::runtime::runtime_test_guard()
1903 .lock()
1904 .expect("provider test guard");
1905 restore_default_result_cache_config();
1906
1907 let db = MemDB::instance();
1908 db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
1909 .expect("create table");
1910 db.execute("INSERT INTO t (id, value) VALUES (1, 'first')")
1911 .expect("seed table");
1912 init_mem_provider(db).expect("init provider");
1913
1914 let before = runtime_snapshot();
1915 let root = uniq_cache_cfg_tmp_dir();
1916 let conf_path = write_provider_only_knowdb_with_cache(
1917 &root,
1918 "mysql",
1919 "not-a-valid-mysql-url",
1920 false,
1921 3,
1922 5,
1923 );
1924 let authority_uri = format!(
1925 "file:{}?mode=rwc&uri=true",
1926 root.join("unused.sqlite").display()
1927 );
1928
1929 let err =
1930 init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default());
1931 assert!(err.is_err());
1932
1933 let after = runtime_snapshot();
1934 assert_eq!(after.result_cache_enabled, before.result_cache_enabled);
1935 assert_eq!(after.result_cache_capacity, before.result_cache_capacity);
1936 assert_eq!(after.result_cache_ttl_ms, before.result_cache_ttl_ms);
1937
1938 let row = query_row("SELECT value FROM t WHERE id = 1").expect("query previous provider");
1939 assert_eq!(row[0].to_string(), "chars(first)");
1940
1941 let _ = fs::remove_dir_all(&root);
1942 }
1943
1944 fn install_named_mem_provider(name: &str, db: MemDB) {
1946 let datasource_id = DatasourceId::from_seed(ProviderKind::SqliteAuthority, name);
1947 let ds_for_scope = datasource_id.clone();
1948 runtime()
1949 .install_provider_named(
1950 name,
1951 ProviderKind::SqliteAuthority,
1952 datasource_id,
1953 |generation| {
1954 Ok(Arc::new(MemProvider {
1955 db,
1956 metadata_scope: MetadataCacheScope {
1957 datasource_id: ds_for_scope.clone(),
1958 generation,
1959 },
1960 }))
1961 },
1962 false,
1963 )
1964 .expect("install named mem provider");
1965 }
1966
1967 #[test]
1968 fn cache_query_fields_route_separates_named_and_default_providers() {
1969 let _guard = crate::runtime::runtime_test_guard()
1970 .lock()
1971 .expect("provider test guard");
1972
1973 let db_default = MemDB::instance();
1975 db_default
1976 .execute("CREATE TABLE route_t (id INTEGER PRIMARY KEY, value TEXT)")
1977 .expect("create default table");
1978 db_default
1979 .execute("INSERT INTO route_t (id, value) VALUES (1, 'default')")
1980 .expect("seed default");
1981 init_mem_provider(db_default).expect("init default provider");
1982
1983 let db_geo = MemDB::instance();
1985 db_geo
1986 .execute("CREATE TABLE route_t (id INTEGER PRIMARY KEY, value TEXT)")
1987 .expect("create geo table");
1988 db_geo
1989 .execute("INSERT INTO route_t (id, value) VALUES (1, 'geo')")
1990 .expect("seed geo");
1991 install_named_mem_provider("geo", db_geo);
1992
1993 let sql = "SELECT value FROM route_t WHERE id=:id";
1994 let c_params = [DataField::from_digit("id", 1)];
1995 let query_params = [DataField::from_digit(":id", 1)];
1996 let mut cache = FieldQueryCache::default();
1997
1998 let named =
1999 cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
2000 assert_eq!(named[0].to_string(), "chars(geo)");
2001
2002 let default = cache_query_fields_route(None, sql, 0, &c_params, &query_params, &mut cache);
2004 assert_eq!(default[0].to_string(), "chars(default)");
2005
2006 let named_again =
2008 cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
2009 assert_eq!(named_again[0].to_string(), "chars(geo)");
2010 }
2011
2012 #[test]
2013 fn mixed_provider_queries_keep_local_cache_per_provider() {
2014 let _guard = crate::runtime::runtime_test_guard()
2015 .lock()
2016 .expect("provider test guard");
2017
2018 let db_default = MemDB::instance();
2019 db_default
2020 .execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
2021 .expect("create default table");
2022 db_default
2023 .execute("INSERT INTO t (id, value) VALUES (1, 'default')")
2024 .expect("seed default");
2025 init_mem_provider(db_default).expect("init default provider");
2026
2027 let db_geo = MemDB::instance();
2028 db_geo
2029 .execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
2030 .expect("create geo table");
2031 db_geo
2032 .execute("INSERT INTO t (id, value) VALUES (1, 'geo')")
2033 .expect("seed geo");
2034 install_named_mem_provider("geo", db_geo);
2035
2036 let sql = "SELECT value FROM t WHERE id=:id";
2037 let c_params = [DataField::from_digit("id", 1)];
2038 let query_params = [DataField::from_digit(":id", 1)];
2039 let mut cache = FieldQueryCache::default();
2040
2041 let before = runtime_snapshot();
2042 let row =
2044 cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
2045 assert_eq!(row[0].to_string(), "chars(geo)");
2046 let row = cache_query_fields_route(None, sql, 0, &c_params, &query_params, &mut cache);
2048 assert_eq!(row[0].to_string(), "chars(default)");
2049 let row =
2051 cache_query_fields_route(Some("geo"), sql, 0, &c_params, &query_params, &mut cache);
2052 assert_eq!(row[0].to_string(), "chars(geo)");
2053 let after = runtime_snapshot();
2054 assert!(
2055 after.local_cache_hits > before.local_cache_hits,
2056 "命名 provider 二次查询应命中本地缓存(按 provider 隔离代际)"
2057 );
2058 }
2059
2060 #[test]
2061 fn local_authority_init_prunes_stale_named_providers() {
2062 let _guard = crate::runtime::runtime_test_guard()
2063 .lock()
2064 .expect("provider test guard");
2065
2066 let db = MemDB::instance();
2068 db.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, value TEXT)")
2069 .expect("create table");
2070 db.execute("INSERT INTO t (id, value) VALUES (1, 'x')")
2071 .expect("seed table");
2072 install_named_mem_provider("geo", db);
2073 assert!(provider_exists("geo"));
2074
2075 let root = uniq_cache_cfg_tmp_dir();
2077 let conf_path = write_minimal_knowdb_with_cache(&root, true, 16, 30_000);
2078 let auth_file = root.join(".run").join("authority.sqlite");
2079 fs::create_dir_all(auth_file.parent().expect("authority parent"))
2080 .expect("create authority parent");
2081 let authority_uri = format!("file:{}?mode=rwc&uri=true", auth_file.display());
2082
2083 init_thread_cloned_from_knowdb(&root, &conf_path, &authority_uri, &EnvDict::default())
2084 .expect("init local authority");
2085
2086 assert!(!provider_exists("geo"));
2087 assert!(provider_exists(DEFAULT_SQLDB_NAME));
2088
2089 restore_default_result_cache_config();
2090 let _ = fs::remove_dir_all(&root);
2091 }
2092
2093 #[test]
2094 fn cache_query_fields_route_unknown_provider_returns_empty() {
2095 let _guard = crate::runtime::runtime_test_guard()
2096 .lock()
2097 .expect("provider test guard");
2098 let c_params = [DataField::from_digit("id", 1)];
2099 let query_params = [DataField::from_digit(":id", 1)];
2100 let mut cache = FieldQueryCache::default();
2101 let out = cache_query_fields_route(
2102 Some("nope"),
2103 "SELECT value FROM t WHERE id=:id",
2104 0,
2105 &c_params,
2106 &query_params,
2107 &mut cache,
2108 );
2109 assert!(out.is_empty());
2110 }
2111
2112 #[test]
2113 fn query_for_named_provider_returns_rows() {
2114 let _guard = crate::runtime::runtime_test_guard()
2115 .lock()
2116 .expect("provider test guard");
2117 let db = MemDB::instance();
2118 db.execute("CREATE TABLE named_t (id INTEGER PRIMARY KEY, value TEXT)")
2119 .expect("create named table");
2120 db.execute("INSERT INTO named_t (id, value) VALUES (1, 'named')")
2121 .expect("seed named");
2122 install_named_mem_provider("geo", db);
2123
2124 let rows = query_for("geo", "SELECT value FROM named_t WHERE id = 1").expect("query geo");
2125 assert_eq!(rows.len(), 1);
2126 assert_eq!(rows[0][0].to_string(), "chars(named)");
2127 assert!(provider_exists("geo"));
2128 assert!(!provider_exists("nope"));
2129 assert!(provider_names().iter().any(|name| name == "geo"));
2130 }
2131
2132 #[tokio::test(flavor = "current_thread")]
2133 async fn query_async_for_named_provider_works() {
2134 let _guard = crate::runtime::runtime_test_guard().lock_async().await;
2135 let db = MemDB::instance();
2136 db.execute("CREATE TABLE named_t (id INTEGER PRIMARY KEY, value TEXT)")
2137 .expect("create named table");
2138 db.execute("INSERT INTO named_t (id, value) VALUES (1, 'async-named')")
2139 .expect("seed named");
2140 install_named_mem_provider("geo", db);
2141
2142 let rows = query_async_for("geo", "SELECT value FROM named_t WHERE id = 1")
2143 .await
2144 .expect("query geo");
2145 assert_eq!(rows[0][0].to_string(), "chars(async-named)");
2146 }
2147}