Skip to main content

wp_knowledge/
facade.rs

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
221/// 以命名 provider 安装 PostgreSQL 连接(不替换默认 provider)。
222pub 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
266/// 以命名 provider 安装 MySQL 连接(不替换默认 provider)。
267pub 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
315// ---------------------------------------------------------------------------
316// Redis provider API
317// ---------------------------------------------------------------------------
318
319/// Check whether an item exists in a Bloom filter.
320///
321/// Returns `true` if the item *may* exist, `false` if it definitely does not.
322/// Requires the RedisBloom module.
323pub 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
336/// Get the value of a hash field.
337///
338/// Returns `None` when the key or field does not exist.
339pub 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
352/// Get the value of a key.
353///
354/// Returns `None` when the key does not exist.
355pub 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
368/// Check whether a member exists in a set.
369pub 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
382/// Add items to a Bloom filter.
383///
384/// Returns a `Vec<bool>` indicating whether each item was *new* to the filter.
385/// Requires the RedisBloom module.
386pub 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
391/// Create a new Bloom filter.
392///
393/// `error_rate`: desired false positive rate (e.g. `0.01` for 1%).
394/// `capacity`: expected number of items to add.
395/// Requires the RedisBloom module.
396pub fn redis_bf_create(key: &str, error_rate: f64, capacity: i64) -> KnowledgeResult<()> {
397    crate::redis::bf_reserve("knowdb", key, error_rate, capacity)
398}
399
400// ---------------------------------------------------------------------------
401// External call API — named queries from [fun.<name>]
402// ---------------------------------------------------------------------------
403
404/// Register [fun] definitions from knowdb.toml. Called once during init.
405pub(crate) fn register_fun_map(map: HashMap<String, crate::loader::FunSpec>) {
406    crate::fun::register_fun_map(map);
407}
408
409/// Execute a [fun.<name>] named query where returns = "bool".
410pub 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
431/// Execute a [fun.<name>] named query where returns = "value".
432pub 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
486// ---------------------------------------------------------------------------
487
488pub 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
506/// 指定名称的 provider 是否已安装。
507pub fn provider_exists(name: &str) -> bool {
508    runtime().provider_exists(name)
509}
510
511/// 当前已安装的命名 provider 名称列表。
512pub fn provider_names() -> Vec<String> {
513    runtime().provider_names()
514}
515
516/// 对指定命名 provider 执行查询(全部行)。
517pub 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
526/// 对指定命名 provider 异步执行查询(全部行)。
527pub 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
537/// 对指定命名 provider 执行首行查询。
538pub 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
542/// 对指定命名 provider 异步执行首行查询。
543pub 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
652/// 本地缓存路由派发:`provider` 为 `Some(name)` 时走命名 provider,`None` 时走默认 provider。
653pub 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    // provider 粒度 scope:与默认 provider 区分,使本地缓存代际按 provider 独立跟踪,
680    // 混用命名库与默认库时不互相重置。
681    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
697/// 本地缓存命中检查(含 generation 准备)与 miss 记录。命中时返回 `Some(hit)`。
698fn 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
796/// 异步本地缓存路由派发:`provider` 为 `Some(name)` 时走命名 provider,`None` 时走默认 provider。
797pub 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    // provider 粒度 scope:与默认 provider 区分,本地缓存代际按 provider 独立跟踪。
828    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        // New-style [provider.sqldb] / [[provider.sqldb]]
912        if let Some(specs) = provider_cfg.sqldb
913            && !specs.is_empty()
914        {
915            let expected = SqlProviderSpec::validate_specs(&specs)?;
916            // 默认 provider:生效名为 `default`(历史单库)者优先,否则取第一个。
917            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            // 外部 SQL provider 只读(仅 SELECT),初始化中途失败时允许部分安装:
923            // 已连通的库照常可用、连不上的库不可查询,错误由调用方记录后继续。
924            // 因此不做"先全部建池再统一提交"的事务化处理。
925            for spec in &specs {
926                let name = spec.effective_name().to_string();
927                let is_default = name == default_name;
928                // 无显式 name 的历史单库保留 connection_uri 作为 datasource 种子
929                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        // New-style [provider.redis]
961        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            // redis-only 配置不注册任何 sqldb 命名 provider,清空过期命名注册
972            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    // 本地 authority 模式:仅保留默认 provider,清理残留的命名 provider
993    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            &params,
1479            &mut cache,
1480        )
1481        .await;
1482        let second = cache_query_fields_async(
1483            "SELECT value FROM t WHERE id=:id",
1484            &key,
1485            &params,
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            &params,
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            &params,
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            &params,
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            &params,
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            &params,
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            &params,
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            &params,
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            &params,
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            &params,
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    // 安装一个命名内存 provider(不替换默认 provider)。
1945    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        // 默认 provider:route_t.value = 'default'
1974        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        // 命名 provider "geo":route_t.value = 'geo'
1984        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        // 同一 cache 实例切到默认 provider,本地缓存必须按 provider 隔离
2003        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        // 再查命名 provider,仍返回 geo
2007        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        // 1) 命名 provider:miss → 执行
2043        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        // 2) 默认 provider:scope 不同,不应清空 geo 的本地缓存
2047        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        // 3) 再查命名 provider:应命中本地缓存(不重新执行)
2050        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        // 先安装命名 provider,模拟上一份配置的残留
2067        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        // 再用本地 authority 配置初始化 → 应清理 geo,仅保留 default
2076        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}