oxcache 0.4.1

A high-performance multi-level cache library for Rust with L1 (memory) and L2 (Redis) caching.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
// Copyright (c) 2025-2026 Kirky.X
// SPDX-License-Identifier: MIT
// 链式缓存核心实现
//
// ChainCache 提供多后端链式访问,按分数从高到低遍历后端。
// 读取时从高分后端开始,写入时写入所有后端。

use crate::backend::BackendScore;
use crate::backend::{
    AtomicCacheWriter, BackendKind, CacheBackend, CacheConnector, CacheReader, CacheWriter, SyncCacheBackend,
};
use crate::core::EventPublisher;
use crate::error::{OxCacheError, OxCacheResult};
use crate::infra::metrics::unified::GLOBAL_UNIFIED_METRICS;
use async_trait::async_trait;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::OnceLock;
use std::time::Duration;

// Submodules
mod builder;
#[cfg(test)]
mod tests;

// Re-exports from submodules
pub use self::builder::ChainCacheBuilder;

/// 链式缓存中的一个后端链接
///
/// ChainLink 封装了一个后端实例及其分数信息。
/// 分数用于确定链式访问的顺序。
#[derive(Clone)]
pub struct ChainLink {
    /// 后端实例(async trait object)
    backend: Arc<dyn CacheBackend>,
    /// 同步后端实例(可选,当后端实现 SyncCacheBackend 时填充)
    backend_sync: Option<Arc<dyn SyncCacheBackend>>,
    /// 后端分数(越高越快)
    score: u8,
    /// 是否为持久化后端
    is_persistent: bool,
    /// 后端名称
    name: &'static str,
}

impl ChainLink {
    /// 创建新的链式链接
    pub fn new<B>(backend: B, score: u8, is_persistent: bool, name: &'static str) -> Self
    where
        B: CacheBackend + BackendScore + 'static,
    {
        Self {
            backend: Arc::new(backend),
            backend_sync: None,
            score,
            is_persistent,
            name,
        }
    }

    /// 从实现了 BackendScore 的后端创建链接
    pub fn from_backend<B>(backend: B) -> Self
    where
        B: CacheBackend + BackendScore + 'static,
    {
        let score = backend.score();
        let is_persistent = backend.is_persistent();
        let name = backend.backend_name();
        Self {
            backend: Arc::new(backend),
            backend_sync: None,
            score,
            is_persistent,
            name,
        }
    }

    /// 从实现了 SyncCacheBackend 的后端创建链接(同时支持 async 与 sync API)
    ///
    /// 与 `from_backend` 不同,此构造函数要求后端同时实现 `SyncCacheBackend`,
    /// 会同时填充 `backend`(async trait object)和 `backend_sync`(sync trait object),
    /// 使该链接可参与 `ChainCache` 的 sync API。
    pub fn from_sync_backend<B>(backend: B) -> Self
    where
        B: CacheBackend + BackendScore + SyncCacheBackend + 'static,
    {
        let score = backend.score();
        let is_persistent = backend.is_persistent();
        let name = backend.backend_name();
        let arc = Arc::new(backend);
        // 显式标注 sync_arc 类型以触发 unsized coercion(Option 不传播 coercion)
        let sync_arc: Arc<dyn SyncCacheBackend> = arc.clone();
        Self {
            backend: arc,
            backend_sync: Some(sync_arc),
            score,
            is_persistent,
            name,
        }
    }

    /// 获取后端实例引用
    pub fn backend(&self) -> &Arc<dyn CacheBackend> {
        &self.backend
    }

    /// 尝试获取同步后端实例
    ///
    /// 返回 `Some` 当且仅当该链接通过 `from_sync_backend` 创建(或后端同时实现
    /// `SyncCacheBackend`)。返回 `None` 表示该链接不支持 sync API。
    pub fn try_as_sync_backend(&self) -> Option<Arc<dyn SyncCacheBackend>> {
        self.backend_sync.clone()
    }

    /// 获取后端分数
    pub fn score(&self) -> u8 {
        self.score
    }

    /// 是否为持久化后端
    pub fn is_persistent(&self) -> bool {
        self.is_persistent
    }

    /// 获取后端名称
    pub fn name(&self) -> &'static str {
        self.name
    }
}

impl std::fmt::Debug for ChainLink {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ChainLink")
            .field("score", &self.score)
            .field("is_persistent", &self.is_persistent)
            .field("name", &self.name)
            .finish()
    }
}

/// 链式缓存
///
/// ChainCache 管理多个后端,按分数从高到低排序。
/// 读取时从高分后端开始,找到即返回;写入时写入所有后端。
///
/// # TTL 行为契约
///
/// ChainCache 不存储 TTL,所有 TTL 操作透传给链中后端。契约如下:
///
/// - **`set(key, value, ttl=Some(d))`**:所有链接用同一 TTL `d`(透传)
/// - **`set(key, value, ttl=None)`**:所有链接用 `default_ttl.or(None)`;
///   `default_ttl=None` 时各链接使用自己的全局 TTL(如 Moka 的 `time_to_live`)
/// - **`ttl(key)`**:遍历链接(按分数从高到低),返回首个 `Some(ttl)`;
///   即"最高分链接的剩余 TTL"。所有链接都返回 `None` 时(key 不存在或无
///   per-entry TTL)返回 `None`
/// - **`expire(key, ttl)`**:透传给所有链接,任一返回 `Ok(true)` 则返回
///   `Ok(true)`;所有链接都返回 `Ok(false)`(key 不存在)才返回 `Ok(false)`
///
/// `default_ttl` 优先级:`set(ttl=Some) > default_ttl > 各后端自己的全局 TTL`
///
/// # Example
///
/// ```rust,ignore
/// use oxcache::cache::{ChainCache, ChainLink};
/// use oxcache::backend::MokaMemoryBackend;
///
/// let l1 = MokaMemoryBackend::builder().capacity(10000).ttl(Duration::from_secs(300)).build();
/// let l2 = oxcache::backend::RedisBackend::builder().ttl(Duration::from_secs(3600)).build().await?;
///
/// let chain = ChainCache::builder()
///     .link(ChainLink::from_backend(l1))  // L1: 5分钟 TTL
///     .link(ChainLink::from_backend(l2))  // L2: 1小时 TTL
///     .enable_backfill()
///     .build();
///
/// chain.set("key", value, None).await?;  // L1 用 5分钟,L2 用 1小时
/// ```
pub struct ChainCache {
    /// 后端链接列表(按分数降序排列)
    links: Vec<ChainLink>,
    /// 是否启用回填
    backfill_enabled: bool,
    /// 是否启用竞速读(并发查询所有后端,返回最先命中者)
    race_read_enabled: bool,
    /// 默认 TTL
    default_ttl: Option<Duration>,
    /// 懒缓存的 sync backend 收集结果(问题 4.4):
    /// 链构建后 links 不可变,首次收集后复用,避免每次 sync 调用重复 clone 所有 Arc。
    sync_backends: OnceLock<Option<Vec<Arc<dyn SyncCacheBackend>>>>,
    /// 事件发布器(可选),用于抛出后端错误事件而非日志输出
    event_publisher: Option<Arc<dyn EventPublisher>>,
}

impl ChainCache {
    /// 创建新的链式缓存
    pub fn new(links: Vec<ChainLink>) -> Self {
        Self::builder().links(links).build()
    }

    /// 创建链式缓存构建器
    pub fn builder() -> ChainCacheBuilder {
        ChainCacheBuilder::default()
    }

    /// 获取后端链接列表
    pub fn links(&self) -> &[ChainLink] {
        &self.links
    }

    /// 获取后端数量
    pub fn len(&self) -> usize {
        self.links.len()
    }

    /// 检查是否为空
    pub fn is_empty(&self) -> bool {
        self.links.is_empty()
    }

    /// 获取指定分数的后端
    pub fn get_by_score(&self, score: u8) -> Option<&ChainLink> {
        self.links.iter().find(|link| link.score() == score)
    }

    /// 获取最高分后端
    pub fn highest_score_backend(&self) -> Option<&ChainLink> {
        self.links.first()
    }

    /// 抛出后端错误事件(不中断流程)
    ///
    /// 通过 `EventPublisher` 发射后端操作失败事件,调用方自行决定处理方式
    /// (日志、metrics、告警或忽略)。未配置 publisher 时为零开销 no-op。
    fn emit_backend_error(&self, key: &str, backend: &str, error: &OxCacheError) {
        if let Some(publisher) = &self.event_publisher {
            let _ = publisher.publish_error(Some(key.to_string()), format!("backend {}: {}", backend, error));
        }
    }

    /// 获取最低分后端
    pub fn lowest_score_backend(&self) -> Option<&ChainLink> {
        self.links.last()
    }

    /// 异步写入:写入所有链接,透传 TTL(公开 API,内部转为 `Arc` 零拷贝分发)。
    ///
    /// key/value 在 `CacheWriter::set` trait 层以 `Arc` 共享所有权(问题 2.2 / 2.3),
    /// 本方法作为用户入口保持 `&str` / `Vec<u8>` 签名,仅做一次 Arc 装箱。
    pub async fn set(&self, key: &str, value: Vec<u8>, ttl: Option<Duration>) -> OxCacheResult<()> {
        let key = Arc::from(key);
        let value = Arc::new(value);
        CacheWriter::set(self, key, value, ttl).await
    }

    /// 获取所有持久化后端
    pub fn persistent_backends(&self) -> Vec<&ChainLink> {
        self.links.iter().filter(|link| link.is_persistent()).collect()
    }

    /// 获取所有非持久化后端
    pub fn non_persistent_backends(&self) -> Vec<&ChainLink> {
        self.links.iter().filter(|link| !link.is_persistent()).collect()
    }

    /// 从链中读取数据
    ///
    /// 单个后端失败时记录 warn 日志(问题 5.1),并继续尝试下一个后端(L1 失败降级到 L2)。
    /// 若启用了竞速读(race_read),则并发查询所有后端并返回最先命中者。
    /// 所有后端都失败时返回 `Err`(与竞速读语义一致)。
    async fn read_from_chain(&self, key: &str) -> OxCacheResult<Option<Vec<u8>>> {
        if self.race_read_enabled {
            return self.race_read_from_chain(key).await;
        }

        let mut all_failed = true;
        let mut last_err: Option<OxCacheError> = None;

        for (index, link) in self.links.iter().enumerate() {
            match link.backend().get(key).await {
                Ok(Some(value)) => {
                    // 回填到更高分后端
                    if self.backfill_enabled && index > 0 {
                        // 查询原始 TTL 并在回填时保留
                        let original_ttl = self.links[index].backend().ttl(key).await.ok().flatten();
                        // 将 value 移入 Arc 后直接用于回填和返回,避免额外 clone(OCR #36)
                        let value = Arc::new(value);
                        self.backfill_to_higher_backends(Arc::from(key), value.clone(), index, original_ttl)
                            .await;
                        return Ok(Some(Arc::try_unwrap(value).unwrap_or_else(|arc| (*arc).clone())));
                    }
                    return Ok(Some(value));
                }
                Ok(None) => {
                    all_failed = false; // 至少有一个后端明确返回 miss
                    continue;
                }
                Err(e) => {
                    self.emit_backend_error(key, link.name(), &e);
                    last_err = Some(e);
                    continue;
                }
            }
        }

        // 所有后端都返回 Err 时传播错误,与竞速读语义一致
        if all_failed && self.links.is_empty() {
            return Ok(None);
        }
        if all_failed {
            return Err(last_err
                .unwrap_or_else(|| OxCacheError::Operation("All backends failed during sequential read".to_string())));
        }
        Ok(None)
    }

    /// 竞速读:并发向所有后端发起 `get`,返回最先命中者(问题 4.1)。
    ///
    /// 适用于 L1/L2 延迟差异小但可用性要求高的场景。最先返回命中值的后端
    /// 获胜;若全部未命中则返回 `None`;若全部失败则返回 `Err`。命中时若有
    /// 回填开启,异步回填到更高分后端。
    async fn race_read_from_chain(&self, key: &str) -> OxCacheResult<Option<Vec<u8>>> {
        if self.links.is_empty() {
            return Ok(None);
        }

        let mut set = tokio::task::JoinSet::new();
        for (index, link) in self.links.iter().enumerate() {
            let backend = link.backend().clone();
            let key = key.to_string();
            set.spawn(async move { (index, backend.get(&key).await) });
        }

        let mut errs: Vec<(&'static str, OxCacheError)> = Vec::new();
        let mut hits: Vec<(usize, Vec<u8>)> = Vec::new();

        while let Some(joined) = set.join_next().await {
            match joined {
                Ok((index, Ok(Some(value)))) => hits.push((index, value)),
                Ok((_index, Ok(None))) => {}
                Ok((index, Err(e))) => {
                    self.emit_backend_error(key, self.links[index].name(), &e);
                    errs.push((self.links[index].name(), e));
                }
                Err(e) => errs.push(("unknown", OxCacheError::Operation(e.to_string()))),
            }
        }

        // 取最先命中(分数最高,即 index 最小)的结果
        if let Some((index, value)) = hits.into_iter().min_by_key(|(i, _)| *i) {
            if self.backfill_enabled && index > 0 {
                // 查询原始 TTL 并在回填时保留
                let original_ttl = self.links[index].backend().ttl(key).await.ok().flatten();
                self.backfill_to_higher_backends(Arc::from(key), Arc::new(value.clone()), index, original_ttl)
                    .await;
            }
            return Ok(Some(value));
        }

        if errs.len() == self.links.len() {
            return Err(OxCacheError::Operation(
                "All backends failed during race read".to_string(),
            ));
        }

        Ok(None)
    }

    /// 回填数据到更高分后端,保留原始 TTL
    ///
    /// 顺序写入更高分后端(问题 4.2)。key/value 以 `Arc` 共享所有权(问题 2.2 / 2.3),
    /// 各后端 `Arc::clone` 仅增加引用计数,无堆拷贝。失败时记录 warn 日志,不回滚读取结果。
    /// `ttl` 为从源后端查询到的原始 TTL,传递给目标后端以保持过期语义一致。
    async fn backfill_to_higher_backends(
        &self,
        key: Arc<str>,
        value: Arc<Vec<u8>>,
        from_index: usize,
        ttl: Option<Duration>,
    ) {
        for link in &self.links[..from_index] {
            let backend = link.backend().clone();
            match backend.set(key.clone(), value.clone(), ttl).await {
                Ok(()) => {
                    GLOBAL_UNIFIED_METRICS.record_backfill_success();
                }
                Err(e) => {
                    GLOBAL_UNIFIED_METRICS.record_backfill_failed();
                    self.emit_backend_error(&key, link.name(), &e);
                }
            }
        }
    }

    /// 写入数据到所有后端
    /// ttl=None 时各 backend 用自己的默认 TTL
    /// ttl=Some 时所有 backend 用同一个 TTL
    ///
    /// 并发写入所有后端(JoinSet),写入延迟从 O(Σbackend) 降至 O(max(backend))(问题 4.3)。
    /// key/value 以 `Arc` 共享所有权传入,各后端 `Arc::clone` 零拷贝(问题 2.2 / 2.3)。
    async fn write_to_all_backends(
        &self,
        key: &Arc<str>,
        value: &Arc<Vec<u8>>,
        ttl: Option<Duration>,
    ) -> OxCacheResult<()> {
        let count = self.links.len();

        if count == 0 {
            return Ok(());
        }

        let effective_ttl = ttl.or(self.default_ttl);

        // 并发写入所有后端(问题 4.3)
        // 所有后端共享同一份 Arc 分配,`Arc::clone` 仅增加引用计数,无堆拷贝(问题 2.3)
        let mut errors: Vec<(&'static str, OxCacheError)> = Vec::new();
        let mut set = tokio::task::JoinSet::new();
        for link in &self.links {
            let backend = link.backend().clone();
            let name = link.name();
            let key = key.clone();
            let value = value.clone();
            set.spawn(async move { (name, backend.set(key, value, effective_ttl).await) });
        }

        while let Some(joined) = set.join_next().await {
            match joined {
                Ok((_name, Ok(()))) => {}
                Ok((name, Err(e))) => errors.push((name, e)),
                Err(e) => errors.push(("unknown", OxCacheError::Operation(e.to_string()))),
            }
        }

        // 抛出后端写入失败事件
        for (name, e) in &errors {
            self.emit_backend_error(key, name, e);
        }

        if errors.len() == self.links.len() {
            return Err(OxCacheError::Operation("All backends failed to write".to_string()));
        }

        Ok(())
    }

    /// 从所有后端并发删除数据
    ///
    /// 使用 JoinSet 并发执行,延迟从 O(Σbackend) 降至 O(max(backend)),
    /// 与 `write_to_all_backends()` 语义对称。部分后端失败时记录 warn 日志,
    /// 仅当所有后端都失败才返回 Err。
    async fn delete_from_all_backends(&self, key: &str) -> OxCacheResult<()> {
        let count = self.links.len();

        if count == 0 {
            return Ok(());
        }

        let mut set = tokio::task::JoinSet::new();
        for link in &self.links {
            let backend = link.backend().clone();
            let name = link.name();
            let key = key.to_string();
            set.spawn(async move { (name, backend.delete(&key).await) });
        }

        let mut errors: Vec<(&'static str, OxCacheError)> = Vec::new();
        while let Some(joined) = set.join_next().await {
            match joined {
                Ok((_name, Ok(()))) => {}
                Ok((name, Err(e))) => {
                    self.emit_backend_error(key, name, &e);
                    errors.push((name, e));
                }
                Err(e) => errors.push(("unknown", OxCacheError::Operation(e.to_string()))),
            }
        }

        if errors.len() == self.links.len() {
            return Err(OxCacheError::Operation(format!(
                "All backends failed to delete: {:?}",
                errors
            )));
        }

        Ok(())
    }

    // ========================================================================
    // Sync API (任务组 15)
    //
    // 同步版链式访问。语义与 async 版一致:get_sync 按分数从高到低遍历,
    // set_sync 写入所有链接(透传 TTL),delete_sync 从所有链接删除。
    //
    // 契约:链中任一链接未实现 SyncCacheBackend(`try_as_sync_backend` 返回
    // None)时,所有 sync API 返回 `Err(NotSupported)`。这避免部分链接静默
    // 跳过导致的写丢失风险。
    // ========================================================================

    /// 收集链中所有链接的 sync backend。
    ///
    /// 链中任一链接未实现 `SyncCacheBackend` 时返回 `Err(NotSupported)`。
    /// links 已按分数降序排列,故返回的 Vec 也是降序。结果懒缓存(问题 4.4):
    /// 首次调用后复用,避免每次 sync 操作重复 clone 所有 Arc。
    fn collect_sync_backends(&self) -> OxCacheResult<&[Arc<dyn SyncCacheBackend>]> {
        let cached = self.sync_backends.get_or_init(|| {
            self.links
                .iter()
                .map(|link| link.try_as_sync_backend())
                .collect::<Option<Vec<_>>>()
        });
        cached.as_deref().ok_or_else(|| {
            OxCacheError::NotSupported("chain sync API requires all links to support SyncCacheBackend".to_string())
        })
    }

    /// 同步读取:按分数从高到低遍历 sync backends,返回首个命中
    pub fn get_sync(&self, key: &str) -> OxCacheResult<Option<Vec<u8>>> {
        let sync_backends = self.collect_sync_backends()?;
        for backend in sync_backends {
            match backend.get(key) {
                Ok(Some(value)) => return Ok(Some(value)),
                Ok(None) => continue,
                Err(_) => continue,
            }
        }
        Ok(None)
    }

    /// 同步写入:写入所有 sync backends,透传 TTL
    pub fn set_sync(&self, key: &str, value: Vec<u8>, ttl: Option<Duration>) -> OxCacheResult<()> {
        let sync_backends = self.collect_sync_backends()?;

        if sync_backends.is_empty() {
            return Err(OxCacheError::Operation("Chain has no backends".to_string()));
        }

        let effective_ttl = ttl.or(self.default_ttl);
        let key_arc: Arc<str> = Arc::from(key);
        let value_arc: Arc<Vec<u8>> = Arc::new(value);
        let mut errors = Vec::new();

        // 所有后端共享同一份 Arc 分配,`Arc::clone` 仅增加引用计数,无堆拷贝(问题 2.2 / 2.3)
        for backend in sync_backends.iter() {
            if let Err(e) = backend.set(key_arc.clone(), value_arc.clone(), effective_ttl) {
                errors.push(e);
            }
        }

        if errors.len() == sync_backends.len() {
            return Err(OxCacheError::Operation("All backends failed to write".to_string()));
        }

        Ok(())
    }

    /// 同步删除:从所有 sync backends 删除
    pub fn delete_sync(&self, key: &str) -> OxCacheResult<()> {
        let sync_backends = self.collect_sync_backends()?;

        let mut errors = Vec::new();

        for backend in sync_backends {
            if let Err(e) = backend.delete(key) {
                errors.push(e);
            }
        }

        if errors.len() == sync_backends.len() && !sync_backends.is_empty() {
            return Err(OxCacheError::Operation(format!(
                "All backends failed to delete: {:?}",
                errors
            )));
        }

        Ok(())
    }
}

#[async_trait]
impl CacheReader for ChainCache {
    async fn get(&self, key: &str) -> OxCacheResult<Option<Vec<u8>>> {
        if self.links.is_empty() {
            return Ok(None);
        }
        self.read_from_chain(key).await
    }

    async fn exists(&self, key: &str) -> OxCacheResult<bool> {
        for link in &self.links {
            match link.backend().exists(key).await {
                Ok(true) => return Ok(true),
                Ok(false) => continue,
                Err(_) => continue,
            }
        }
        Ok(false)
    }

    async fn ttl(&self, key: &str) -> OxCacheResult<Option<Duration>> {
        for link in &self.links {
            match link.backend().ttl(key).await {
                Ok(Some(ttl)) => return Ok(Some(ttl)),
                Ok(None) => continue,
                Err(_) => continue,
            }
        }
        Ok(None)
    }

    async fn len(&self) -> OxCacheResult<u64> {
        if let Some(link) = self.links.first() {
            link.backend().len().await
        } else {
            Ok(0)
        }
    }

    async fn is_empty(&self) -> OxCacheResult<bool> {
        if let Some(link) = self.links.first() {
            link.backend().is_empty().await
        } else {
            Ok(true)
        }
    }

    async fn capacity(&self) -> OxCacheResult<u64> {
        if let Some(link) = self.links.first() {
            link.backend().capacity().await
        } else {
            Ok(0)
        }
    }

    async fn stats(&self) -> OxCacheResult<HashMap<String, String>> {
        let mut stats = HashMap::new();
        stats.insert("type".to_string(), "chain".to_string());
        stats.insert("backend_count".to_string(), self.links.len().to_string());

        for (index, link) in self.links.iter().enumerate() {
            stats.insert(format!("backend_{}_name", index), link.name().to_string());
            stats.insert(format!("backend_{}_score", index), link.score().to_string());
        }

        Ok(stats)
    }

    async fn keys(&self, pattern: &str) -> OxCacheResult<Vec<String>> {
        let mut seen = std::collections::HashSet::new();
        let mut result = Vec::new();
        for link in &self.links {
            if let Ok(keys) = link.backend().keys(pattern).await {
                for k in keys {
                    if seen.insert(k.clone()) {
                        result.push(k);
                    }
                }
            }
        }
        Ok(result)
    }
}

#[async_trait]
impl CacheWriter for ChainCache {
    async fn set(&self, key: Arc<str>, value: Arc<Vec<u8>>, ttl: Option<Duration>) -> OxCacheResult<()> {
        if self.links.is_empty() {
            return Err(OxCacheError::Operation("Chain has no backends".to_string()));
        }
        self.write_to_all_backends(&key, &value, ttl).await
    }

    async fn delete(&self, key: &str) -> OxCacheResult<()> {
        if self.links.is_empty() {
            return Ok(());
        }
        self.delete_from_all_backends(key).await
    }

    async fn clear(&self) -> OxCacheResult<()> {
        let mut errors = Vec::new();

        for link in &self.links {
            if let Err(e) = link.backend().clear().await {
                errors.push((link.name(), e));
            }
        }

        if errors.len() == self.links.len() && !self.links.is_empty() {
            return Err(OxCacheError::Operation(format!(
                "All backends failed to clear: {:?}",
                errors
            )));
        }

        Ok(())
    }

    async fn expire(&self, key: &str, ttl: Duration) -> OxCacheResult<bool> {
        let mut any_success = false;

        for link in &self.links {
            match link.backend().expire(key, ttl).await {
                Ok(true) => any_success = true,
                _ => continue,
            }
        }

        Ok(any_success)
    }
}

#[async_trait]
impl CacheConnector for ChainCache {
    /// 并发检查所有后端健康状态,每个后端有独立 5s 超时(问题 5.2)。
    ///
    /// 任一后端失败(含超时)即返回错误;所有后端健康才返回 Ok。
    async fn health_check(&self) -> OxCacheResult<()> {
        if self.links.is_empty() {
            return Ok(());
        }

        const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(5);

        let mut set = tokio::task::JoinSet::new();
        for link in &self.links {
            let backend = link.backend().clone();
            let name = link.name();
            set.spawn(async move {
                let result = tokio::time::timeout(HEALTH_CHECK_TIMEOUT, backend.health_check()).await;
                (name, result)
            });
        }

        let mut failures = Vec::new();
        while let Some(joined) = set.join_next().await {
            match joined {
                Ok((_name, Ok(Ok(())))) => {}
                Ok((name, Ok(Err(e)))) => failures.push((name, e)),
                Ok((name, Err(_))) => failures.push((
                    name,
                    OxCacheError::Timeout("health_check timed out after 5s".to_string()),
                )),
                Err(e) => failures.push(("unknown", OxCacheError::Operation(e.to_string()))),
            }
        }

        if failures.is_empty() {
            Ok(())
        } else {
            Err(OxCacheError::Operation(format!(
                "health_check failed for {} backend(s): {:?}",
                failures.len(),
                failures
            )))
        }
    }

    async fn shutdown(&self) {
        for link in &self.links {
            link.backend().shutdown().await;
        }
    }

    fn backend_kind(&self) -> BackendKind {
        BackendKind::Chain
    }
}

// ============================================================================
// AtomicCacheWriter for ChainCache (T028)
// ============================================================================

#[async_trait]
impl AtomicCacheWriter for ChainCache {
    /// 委托最高分后端的 `AtomicCacheWriter`。无原子后端时返回 `Err(NotSupported)`。
    async fn incr(&self, key: &str, delta: i64, ttl: Option<Duration>) -> OxCacheResult<i64> {
        let writer = self
            .links
            .first()
            .and_then(|link| link.backend().as_atomic_writer())
            .ok_or_else(|| {
                OxCacheError::NotSupported("incr: no link in chain implements AtomicCacheWriter".to_string())
            })?;
        writer.incr(key, delta, ttl).await
    }

    async fn compare_and_swap(
        &self,
        key: &str,
        expected: Option<&[u8]>,
        new: Vec<u8>,
        ttl: Option<Duration>,
    ) -> OxCacheResult<bool> {
        let writer = self
            .links
            .first()
            .and_then(|link| link.backend().as_atomic_writer())
            .ok_or_else(|| {
                OxCacheError::NotSupported(
                    "compare_and_swap: no link in chain implements AtomicCacheWriter".to_string(),
                )
            })?;
        writer.compare_and_swap(key, expected, new, ttl).await
    }

    async fn set_if_absent(&self, key: &str, value: Vec<u8>, ttl: Option<Duration>) -> OxCacheResult<bool> {
        let writer = self
            .links
            .first()
            .and_then(|link| link.backend().as_atomic_writer())
            .ok_or_else(|| {
                OxCacheError::NotSupported("set_if_absent: no link in chain implements AtomicCacheWriter".to_string())
            })?;
        writer.set_if_absent(key, value, ttl).await
    }
}

// ============================================================================
// keys() override for ChainCache (T029)
// ============================================================================