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