Skip to main content

opendal_layer_foyer/
lib.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18#![doc = include_str!("../README.md")]
19#![cfg_attr(docsrs, feature(doc_cfg))]
20#![cfg_attr(docsrs, doc(auto_cfg))]
21#![deny(missing_docs)]
22mod deleter;
23mod error;
24mod full;
25mod writer;
26
27use std::{
28    collections::HashSet,
29    ops::{Bound, Deref, Range, RangeBounds},
30    sync::{Arc, Mutex},
31};
32
33use foyer::{Code, HybridCache, Result as FoyerResult};
34
35use opendal_core::raw::*;
36use opendal_core::*;
37
38pub use deleter::Deleter;
39pub use writer::Writer;
40
41/// [`FoyerKey`] is a key for the foyer cache. It implements foyer's [`Code`] trait
42/// directly, so the layer does not depend on foyer's `serde` feature (and its bincode
43/// dependency).
44///
45/// It's possible to specify a version in the [`OpRead`] args:
46///
47/// - If a version is given, the object is cached under that versioned key.
48/// - If version is not supplied, the object is cached exactly as returned by the backend,
49///   We do NOT interpret `None` as "latest" and we do not promote it to any other version.
50#[derive(Debug, Clone, PartialEq, Eq, Hash)]
51pub struct FoyerKey {
52    /// Object path used as the cache key.
53    pub path: String,
54    /// Object version, when the read targets a specific version.
55    pub version: Option<String>,
56}
57
58impl Code for FoyerKey {
59    fn encode(&self, writer: &mut impl std::io::Write) -> FoyerResult<()> {
60        write_bytes(writer, self.path.as_bytes())?;
61        match &self.version {
62            None => writer.write_all(&[0u8])?,
63            Some(version) => {
64                writer.write_all(&[1u8])?;
65                write_bytes(writer, version.as_bytes())?;
66            }
67        }
68        Ok(())
69    }
70
71    fn decode(reader: &mut impl std::io::Read) -> FoyerResult<Self> {
72        let path = read_string(reader)?;
73        let mut tag = [0u8; 1];
74        reader.read_exact(&mut tag)?;
75        let version = match tag[0] {
76            0 => None,
77            1 => Some(read_string(reader)?),
78            other => {
79                return Err(std::io::Error::new(
80                    std::io::ErrorKind::InvalidData,
81                    format!("invalid FoyerKey version tag: {other}"),
82                )
83                .into());
84            }
85        };
86        Ok(FoyerKey { path, version })
87    }
88
89    fn estimated_size(&self) -> usize {
90        8 + self.path.len() + 1 + self.version.as_ref().map_or(0, |v| 8 + v.len())
91    }
92}
93
94fn write_bytes(writer: &mut impl std::io::Write, bytes: &[u8]) -> FoyerResult<()> {
95    writer.write_all(&(bytes.len() as u64).to_le_bytes())?;
96    writer.write_all(bytes)?;
97    Ok(())
98}
99
100fn read_string(reader: &mut impl std::io::Read) -> FoyerResult<String> {
101    let mut len = [0u8; 8];
102    reader.read_exact(&mut len)?;
103    let len = u64::from_le_bytes(len) as usize;
104    let mut buf = vec![0u8; len];
105    reader.read_exact(&mut buf)?;
106    String::from_utf8(buf)
107        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e).into())
108}
109
110/// [`FoyerValue`] stores buffered object data in Foyer.
111#[derive(Debug)]
112pub struct FoyerValue(
113    /// Buffered object data.
114    pub Buffer,
115);
116
117impl Deref for FoyerValue {
118    type Target = Buffer;
119
120    fn deref(&self) -> &Self::Target {
121        &self.0
122    }
123}
124
125impl Code for FoyerValue {
126    fn encode(&self, writer: &mut impl std::io::Write) -> FoyerResult<()> {
127        let len = self.0.len() as u64;
128        writer.write_all(&len.to_le_bytes())?;
129        std::io::copy(&mut self.0.clone(), writer)?;
130        Ok(())
131    }
132
133    fn decode(reader: &mut impl std::io::Read) -> FoyerResult<Self>
134    where
135        Self: Sized,
136    {
137        let mut len_bytes = [0u8; 8];
138        reader.read_exact(&mut len_bytes)?;
139        let len = u64::from_le_bytes(len_bytes) as usize;
140        let mut buffer = vec![0u8; len];
141        reader.read_exact(&mut buffer[..len])?;
142        Ok(FoyerValue(buffer.into()))
143    }
144
145    fn estimated_size(&self) -> usize {
146        8 + self.0.len()
147    }
148}
149
150/// `FoyerLayer` caches OpenDAL data with [foyer](https://github.com/foyer-rs/foyer).
151///
152/// # Operation Behavior
153/// - `write`: [`FoyerLayer`] caches data after the service completes the write.
154/// - `read`: [`FoyerLayer`] checks the cache first. On a cache miss, it reads
155///   from the service and caches the result.
156/// - `delete`: [`FoyerLayer`] removes cached data after a successful delete when the deleter
157///   closes. A failed delete is not invalidated. Cache invalidation happens before the underlying
158///   deleter is closed, so the data remains invalidated if closing the deleter fails.
159/// - Other operations: [`FoyerLayer`] passes operations such as `list`, `copy`,
160///   and `rename` to the service without caching their results.
161///
162/// # Examples
163///
164/// ```no_run
165/// use opendal_core::{Operator, services::Memory};
166/// use opendal_layer_foyer::FoyerLayer;
167/// use foyer::HybridCacheBuilder;
168///
169/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
170/// let cache = HybridCacheBuilder::new()
171///     .memory(64 * 1024 * 1024) // 64MB memory cache
172///     .with_shards(4)
173///     .storage()
174///     .build()
175///     .await?;
176///
177/// let op = Operator::new(Memory::default())?
178///     .layer(FoyerLayer::new(cache));
179/// # Ok(())
180/// # }
181/// ```
182///
183/// # Note
184///
185/// When object versioning is enabled, `FoyerLayer` treats objects with the same
186/// path but different versions as different objects.
187#[derive(Debug)]
188pub struct FoyerLayer {
189    cache: HybridCache<FoyerKey, FoyerValue>,
190    size_limit: Range<usize>,
191    deleted_keys: Arc<Mutex<HashSet<FoyerKey>>>,
192}
193
194impl FoyerLayer {
195    /// Creates a new `FoyerLayer` with the given foyer hybrid cache.
196    pub fn new(cache: HybridCache<FoyerKey, FoyerValue>) -> Self {
197        FoyerLayer {
198            cache,
199            size_limit: 0..usize::MAX,
200            deleted_keys: Arc::default(),
201        }
202    }
203
204    /// Set the object-size range eligible for caching.
205    ///
206    /// The layer reads through or writes through objects outside this range
207    /// without retaining them in Foyer. The default range accepts every
208    /// representable object size below `usize::MAX`.
209    pub fn with_size_limit<R: RangeBounds<usize>>(mut self, size_limit: R) -> Self {
210        let start = match size_limit.start_bound() {
211            Bound::Included(v) => *v,
212            Bound::Excluded(v) => *v + 1,
213            Bound::Unbounded => 0,
214        };
215        let end = match size_limit.end_bound() {
216            Bound::Included(v) => *v + 1,
217            Bound::Excluded(v) => *v,
218            Bound::Unbounded => usize::MAX,
219        };
220        self.size_limit = start..end;
221        self
222    }
223}
224
225impl Layer for FoyerLayer {
226    fn apply_service(&self, inner: Servicer) -> Servicer {
227        Arc::new(self.layer(inner))
228    }
229}
230
231impl FoyerLayer {
232    fn layer(&self, inner: Servicer) -> FoyerService {
233        FoyerService {
234            inner,
235            cache: self.cache.clone(),
236            size_limit: self.size_limit.clone(),
237            deleted_keys: self.deleted_keys.clone(),
238        }
239    }
240}
241
242#[derive(Debug)]
243pub(crate) struct Inner {
244    pub(crate) srv: Servicer,
245    pub(crate) ctx: OperationContext,
246    pub(crate) cache: HybridCache<FoyerKey, FoyerValue>,
247    pub(crate) deleted_keys: Arc<Mutex<HashSet<FoyerKey>>>,
248}
249
250#[derive(Debug)]
251#[doc(hidden)]
252pub struct FoyerService {
253    inner: Servicer,
254    cache: HybridCache<FoyerKey, FoyerValue>,
255    size_limit: Range<usize>,
256    deleted_keys: Arc<Mutex<HashSet<FoyerKey>>>,
257}
258
259impl FoyerService {
260    fn operation_inner(&self, ctx: &OperationContext) -> Arc<Inner> {
261        Arc::new(Inner {
262            srv: self.inner.clone(),
263            ctx: ctx.clone(),
264            cache: self.cache.clone(),
265            deleted_keys: self.deleted_keys.clone(),
266        })
267    }
268}
269
270impl Service for FoyerService {
271    type Reader = full::FullReader;
272    type Writer = Writer<oio::Writer>;
273    type Lister = oio::Lister;
274    type Deleter = Deleter<oio::Deleter>;
275    type Copier = oio::Copier;
276
277    fn info(&self) -> ServiceInfo {
278        self.inner.info()
279    }
280
281    fn capability(&self) -> Capability {
282        self.inner.capability()
283    }
284
285    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
286        Ok(full::FullReader::new(
287            self.operation_inner(ctx),
288            self.size_limit.clone(),
289            path.to_string(),
290            args,
291        ))
292    }
293
294    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
295        let inner = self.operation_inner(ctx);
296        let size_limit = self.size_limit.clone();
297        let path = path.to_string();
298        let w = inner.srv.write(&inner.ctx, &path, args)?;
299        Ok(Writer::new(w, path, inner, size_limit))
300    }
301
302    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
303        let inner = self.operation_inner(ctx);
304        let d = inner.srv.delete(&inner.ctx)?;
305        Ok(Deleter::new(d, inner))
306    }
307
308    fn copy(
309        &self,
310        ctx: &OperationContext,
311        from: &str,
312        to: &str,
313        args: OpCopy,
314        opts: OpCopier,
315    ) -> Result<Self::Copier> {
316        self.inner.copy(ctx, from, to, args, opts)
317    }
318
319    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
320        self.inner.list(ctx, path, args)
321    }
322
323    async fn create_dir(
324        &self,
325        ctx: &OperationContext,
326        path: &str,
327        args: OpCreateDir,
328    ) -> Result<RpCreateDir> {
329        self.inner.create_dir(ctx, path, args).await
330    }
331
332    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
333        self.inner.stat(ctx, path, args).await
334    }
335
336    async fn rename(
337        &self,
338        ctx: &OperationContext,
339        from: &str,
340        to: &str,
341        args: OpRename,
342    ) -> Result<RpRename> {
343        self.inner.rename(ctx, from, to, args).await
344    }
345
346    async fn presign(
347        &self,
348        ctx: &OperationContext,
349        path: &str,
350        args: OpPresign,
351    ) -> Result<RpPresign> {
352        self.inner.presign(ctx, path, args).await
353    }
354
355    // TODO(MrCroxx): Implement copy, rename with foyer cache.
356}
357
358#[cfg(test)]
359mod tests {
360    use foyer::{
361        BlockEngineConfig, DeviceBuilder, Error as FoyerError, ErrorKind as FoyerErrorKind,
362        FsDeviceBuilder, HybridCache, HybridCacheBuilder, RecoverMode,
363    };
364    use opendal_core::raw::Layer as _;
365    use opendal_core::raw::oio::Read as _;
366    use opendal_core::raw::oio::ReadStream as _;
367    use opendal_core::{Buffer, Operator, services::Memory};
368    use size::consts::MiB;
369    use std::io::Cursor;
370    use std::sync::Arc;
371    use std::sync::atomic::{AtomicUsize, Ordering};
372
373    use super::*;
374    use crate::error::extract_err;
375
376    fn key(i: u8) -> String {
377        format!("obj-{i}")
378    }
379
380    fn value(i: u8) -> Vec<u8> {
381        // ~ 64KiB with metadata
382        vec![i; 63 * 1024]
383    }
384
385    async fn memory_cache() -> HybridCache<FoyerKey, FoyerValue> {
386        HybridCacheBuilder::new()
387            .memory(1024 * 1024)
388            .with_shards(1)
389            .storage()
390            .with_recover_mode(RecoverMode::None)
391            .build()
392            .await
393            .unwrap()
394    }
395
396    #[derive(Debug)]
397    struct MockReadState {
398        data: Buffer,
399        stat_calls: AtomicUsize,
400        open_calls: AtomicUsize,
401        read_calls: AtomicUsize,
402        last_stat_args: Mutex<Option<OpStat>>,
403        last_read_args: Mutex<Option<OpRead>>,
404    }
405
406    impl MockReadState {
407        fn new(data: Buffer) -> Self {
408            Self {
409                data,
410                stat_calls: AtomicUsize::new(0),
411                open_calls: AtomicUsize::new(0),
412                read_calls: AtomicUsize::new(0),
413                last_stat_args: Mutex::new(None),
414                last_read_args: Mutex::new(None),
415            }
416        }
417
418        fn metadata(&self) -> Metadata {
419            Metadata::new(EntryMode::FILE).with_content_length(self.data.len() as _)
420        }
421
422        fn rp_read(&self) -> RpRead {
423            RpRead::new(self.metadata())
424        }
425
426        fn read_range(&self, range: BytesRange) -> Buffer {
427            self.data.slice(range.to_range_as_usize())
428        }
429    }
430
431    #[derive(Debug, Clone)]
432    struct MockReadService {
433        state: Arc<MockReadState>,
434    }
435
436    impl MockReadService {
437        fn new(data: impl Into<Buffer>) -> Self {
438            Self {
439                state: Arc::new(MockReadState::new(data.into())),
440            }
441        }
442    }
443
444    #[derive(Debug)]
445    struct MockReadReader {
446        state: Arc<MockReadState>,
447    }
448
449    impl oio::Read for MockReadReader {
450        async fn open(&self, range: BytesRange) -> Result<(RpRead, Box<dyn oio::ReadStreamDyn>)> {
451            self.state.open_calls.fetch_add(1, Ordering::Relaxed);
452            let buffer = self.state.read_range(range);
453            Ok((
454                self.state.rp_read(),
455                Box::new(buffer) as Box<dyn oio::ReadStreamDyn>,
456            ))
457        }
458
459        async fn read(&self, range: BytesRange) -> Result<(RpRead, Buffer)> {
460            self.state.read_calls.fetch_add(1, Ordering::Relaxed);
461            if range.size().is_none() {
462                return Err(Error::new(
463                    ErrorKind::Unsupported,
464                    "mock reader requires a bounded read range",
465                ));
466            }
467
468            Ok((self.state.rp_read(), self.state.read_range(range)))
469        }
470    }
471
472    impl Service for MockReadService {
473        type Reader = MockReadReader;
474        type Writer = ();
475        type Lister = ();
476        type Deleter = ();
477        type Copier = ();
478
479        fn info(&self) -> ServiceInfo {
480            ServiceInfo::with_scheme("mock")
481        }
482
483        fn capability(&self) -> Capability {
484            Capability {
485                read: true,
486                stat: true,
487                ..Default::default()
488            }
489        }
490
491        async fn create_dir(
492            &self,
493            _: &OperationContext,
494            _: &str,
495            _: OpCreateDir,
496        ) -> Result<RpCreateDir> {
497            Err(Error::new(
498                ErrorKind::Unsupported,
499                "operation is not supported",
500            ))
501        }
502
503        async fn stat(&self, _: &OperationContext, _: &str, args: OpStat) -> Result<RpStat> {
504            self.state.stat_calls.fetch_add(1, Ordering::Relaxed);
505            *self.state.last_stat_args.lock().unwrap() = Some(args);
506            Ok(RpStat::new(self.state.metadata()))
507        }
508
509        fn read(&self, _ctx: &OperationContext, _: &str, args: OpRead) -> Result<Self::Reader> {
510            *self.state.last_read_args.lock().unwrap() = Some(args);
511            Ok(MockReadReader {
512                state: self.state.clone(),
513            })
514        }
515
516        fn write(&self, _ctx: &OperationContext, _: &str, _: OpWrite) -> Result<Self::Writer> {
517            Err(Error::new(
518                ErrorKind::Unsupported,
519                "operation is not supported",
520            ))
521        }
522
523        fn delete(&self, _ctx: &OperationContext) -> Result<Self::Deleter> {
524            Err(Error::new(
525                ErrorKind::Unsupported,
526                "operation is not supported",
527            ))
528        }
529
530        fn list(&self, _ctx: &OperationContext, _: &str, _: OpList) -> Result<Self::Lister> {
531            Err(Error::new(
532                ErrorKind::Unsupported,
533                "operation is not supported",
534            ))
535        }
536
537        fn copy(
538            &self,
539            _: &OperationContext,
540            _: &str,
541            _: &str,
542            _: OpCopy,
543            _: OpCopier,
544        ) -> Result<Self::Copier> {
545            Err(Error::new(
546                ErrorKind::Unsupported,
547                "operation is not supported",
548            ))
549        }
550
551        async fn rename(
552            &self,
553            _: &OperationContext,
554            _: &str,
555            _: &str,
556            _: OpRename,
557        ) -> Result<RpRename> {
558            Err(Error::new(
559                ErrorKind::Unsupported,
560                "operation is not supported",
561            ))
562        }
563
564        async fn presign(&self, _: &OperationContext, _: &str, _: OpPresign) -> Result<RpPresign> {
565            Err(Error::new(
566                ErrorKind::Unsupported,
567                "operation is not supported",
568            ))
569        }
570    }
571
572    fn service_context(_: &Servicer) -> OperationContext {
573        OperationContext::new()
574    }
575
576    #[tokio::test]
577    async fn test_full_reader_open_fallback_preserves_stream() {
578        let cache = memory_cache().await;
579        let source = Arc::new(MockReadService::new("0123456789"));
580        let state = source.state.clone();
581        let service = FoyerLayer::new(cache)
582            .with_size_limit(0..1)
583            .apply_service(source);
584        let ctx = service_context(&service);
585
586        let reader = service.read(&ctx, "test", OpRead::default()).unwrap();
587        let (_, mut stream) = reader.open(BytesRange::new(0, None)).await.unwrap();
588        let buffer = stream.read_all().await.unwrap();
589
590        assert_eq!(buffer.to_vec(), b"0123456789");
591        assert_eq!(state.open_calls.load(Ordering::Relaxed), 1);
592        assert_eq!(state.read_calls.load(Ordering::Relaxed), 0);
593        assert_eq!(state.stat_calls.load(Ordering::Relaxed), 1);
594    }
595
596    #[tokio::test]
597    async fn test_full_reader_open_fills_cache() {
598        let cache = memory_cache().await;
599        let source = Arc::new(MockReadService::new("0123456789"));
600        let state = source.state.clone();
601        let service = FoyerLayer::new(cache)
602            .with_size_limit(0..100)
603            .apply_service(source);
604        let ctx = service_context(&service);
605
606        let reader = service.read(&ctx, "test", OpRead::default()).unwrap();
607        let (_, mut stream) = reader.open(BytesRange::from(0_u64..2)).await.unwrap();
608        let buffer = stream.read_all().await.unwrap();
609
610        assert_eq!(buffer.to_vec(), b"01");
611        assert_eq!(state.stat_calls.load(Ordering::Relaxed), 1);
612        assert_eq!(state.open_calls.load(Ordering::Relaxed), 1);
613        assert_eq!(state.read_calls.load(Ordering::Relaxed), 0);
614
615        let reader = service.read(&ctx, "test", OpRead::default()).unwrap();
616        let (_, mut stream) = reader.open(BytesRange::from(4_u64..7)).await.unwrap();
617        let buffer = stream.read_all().await.unwrap();
618
619        assert_eq!(buffer.to_vec(), b"456");
620        assert_eq!(state.stat_calls.load(Ordering::Relaxed), 1);
621        assert_eq!(state.open_calls.load(Ordering::Relaxed), 1);
622        assert_eq!(state.read_calls.load(Ordering::Relaxed), 0);
623    }
624
625    #[tokio::test]
626    async fn test_cache_fill_preserves_read_args() {
627        let cache = memory_cache().await;
628        let source = Arc::new(MockReadService::new("0123456789"));
629        let state = source.state.clone();
630        let service = FoyerLayer::new(cache)
631            .with_size_limit(0..100)
632            .apply_service(source);
633        let ctx = service_context(&service);
634
635        let args = OpRead::default().with_version("v1").with_if_match("etag-1");
636        let reader = service.read(&ctx, "test", args).unwrap();
637        let (_, mut stream) = reader.open(BytesRange::new(0, None)).await.unwrap();
638        stream.read_all().await.unwrap();
639
640        let stat_args = state.last_stat_args.lock().unwrap().clone().unwrap();
641        assert_eq!(stat_args.version(), Some("v1"));
642        assert_eq!(stat_args.if_match(), Some("etag-1"));
643
644        let read_args = state.last_read_args.lock().unwrap().clone().unwrap();
645        assert_eq!(read_args.version(), Some("v1"));
646        assert_eq!(read_args.if_match(), Some("etag-1"));
647    }
648
649    #[tokio::test]
650    async fn test() {
651        let dir = tempfile::tempdir().unwrap();
652
653        let cache = HybridCacheBuilder::new()
654            .memory(10)
655            .with_shards(1)
656            .storage()
657            .with_engine_config(
658                BlockEngineConfig::new(
659                    FsDeviceBuilder::new(dir.path())
660                        .with_capacity(16 * MiB as usize)
661                        .build()
662                        .unwrap(),
663                )
664                .with_block_size(MiB as usize),
665            )
666            .with_recover_mode(RecoverMode::None)
667            .build()
668            .await
669            .unwrap();
670
671        let op = Operator::new(Memory::default())
672            .unwrap()
673            .layer(FoyerLayer::new(cache.clone()));
674
675        assert!(op.list("/").await.unwrap().is_empty());
676
677        for i in 0..64 {
678            op.write(&key(i), value(i)).await.unwrap();
679        }
680
681        assert_eq!(op.list("/").await.unwrap().len(), 64);
682
683        for i in 0..64 {
684            let buf = op.read(&key(i)).await.unwrap();
685            assert_eq!(buf.to_vec(), value(i));
686        }
687
688        cache.clear().await.unwrap();
689
690        for i in 0..64 {
691            let buf = op.read(&key(i)).await.unwrap();
692            assert_eq!(buf.to_vec(), value(i));
693        }
694
695        for i in 0..64 {
696            op.delete(&key(i)).await.unwrap();
697        }
698
699        assert!(op.list("/").await.unwrap().is_empty());
700
701        for i in 0..64 {
702            let res = op.read(&key(i)).await;
703            assert!(res.is_err(), "should fail to read deleted file");
704        }
705    }
706
707    #[tokio::test]
708    async fn test_size_limit() {
709        let dir = tempfile::tempdir().unwrap();
710
711        let cache = HybridCacheBuilder::new()
712            .memory(1024 * 1024)
713            .with_shards(1)
714            .storage()
715            .with_engine_config(
716                BlockEngineConfig::new(
717                    FsDeviceBuilder::new(dir.path())
718                        .with_capacity(16 * MiB as usize)
719                        .build()
720                        .unwrap(),
721                )
722                .with_block_size(MiB as usize),
723            )
724            .with_recover_mode(RecoverMode::None)
725            .build()
726            .await
727            .unwrap();
728
729        // Set size limit: only cache files between 1KB and 10KB
730        let op = Operator::new(Memory::default())
731            .unwrap()
732            .layer(FoyerLayer::new(cache.clone()).with_size_limit(1024..10 * 1024));
733
734        let small_data = vec![1u8; 5 * 1024]; // 5KB - should be cached
735        let large_data = vec![2u8; 20 * 1024]; // 20KB - should NOT be cached
736        let tiny_data = vec![3u8; 512]; // 512B - below size limit, should NOT be cached
737
738        // Write all files
739        op.write("small.txt", small_data.clone()).await.unwrap();
740        op.write("large.txt", large_data.clone()).await.unwrap();
741        op.write("tiny.txt", tiny_data.clone()).await.unwrap();
742
743        // All should be readable
744        let read_small = op.read("small.txt").await.unwrap();
745        assert_eq!(read_small.to_vec(), small_data);
746
747        let read_large = op.read("large.txt").await.unwrap();
748        assert_eq!(read_large.to_vec(), large_data);
749
750        let read_tiny = op.read("tiny.txt").await.unwrap();
751        assert_eq!(read_tiny.to_vec(), tiny_data);
752
753        // Clear the cache to test read-through behavior
754        cache.clear().await.unwrap();
755
756        // All files should still be readable from underlying storage
757        let read_small = op.read("small.txt").await.unwrap();
758        assert_eq!(read_small.to_vec(), small_data);
759
760        let read_large = op.read("large.txt").await.unwrap();
761        assert_eq!(read_large.to_vec(), large_data);
762
763        let read_tiny = op.read("tiny.txt").await.unwrap();
764        assert_eq!(read_tiny.to_vec(), tiny_data);
765
766        // After reading, small file should be cached, but large and tiny should not
767        // We can verify this by reading with range - cached files should support range reads
768        let read_small_range = op.read_with("small.txt").range(0..1024).await.unwrap();
769        assert_eq!(read_small_range.len(), 1024);
770        assert_eq!(read_small_range.to_vec(), small_data[0..1024]);
771    }
772
773    #[test]
774    fn test_error() {
775        let e = Error::new(ErrorKind::NotFound, "not found");
776        let fe = FoyerError::new(FoyerErrorKind::External, "external error").with_source(e);
777        let oe = extract_err(fe);
778        assert_eq!(oe.kind(), ErrorKind::NotFound);
779    }
780
781    #[test]
782    fn test_foyer_key_version_none_vs_empty() {
783        let key_none = FoyerKey {
784            path: "test/path".to_string(),
785            version: None,
786        };
787
788        let key_empty = FoyerKey {
789            path: "test/path".to_string(),
790            version: Some("".to_string()),
791        };
792
793        let mut buf_none = Vec::new();
794        key_none.encode(&mut buf_none).unwrap();
795
796        let mut buf_empty = Vec::new();
797        key_empty.encode(&mut buf_empty).unwrap();
798
799        assert_ne!(
800            buf_none, buf_empty,
801            "Serialization of version=None and version=\"\" should be different"
802        );
803
804        let decoded_none = FoyerKey::decode(&mut Cursor::new(&buf_none)).unwrap();
805        assert_eq!(decoded_none, key_none);
806        let decoded_empty = FoyerKey::decode(&mut Cursor::new(&buf_empty)).unwrap();
807        assert_eq!(decoded_empty, key_empty);
808    }
809
810    #[test]
811    fn test_foyer_key_serde() {
812        use std::io::Cursor;
813
814        let test_cases = vec![
815            FoyerKey {
816                path: "simple".to_string(),
817                version: None,
818            },
819            FoyerKey {
820                path: "with/slash/path".to_string(),
821                version: None,
822            },
823            FoyerKey {
824                path: "versioned".to_string(),
825                version: Some("v1.0.0".to_string()),
826            },
827            FoyerKey {
828                path: "empty-version".to_string(),
829                version: Some("".to_string()),
830            },
831            FoyerKey {
832                path: "".to_string(),
833                version: None,
834            },
835            FoyerKey {
836                path: "unicode/θ·―εΎ„/πŸš€".to_string(),
837                version: Some("η‰ˆζœ¬-1".to_string()),
838            },
839            FoyerKey {
840                path: "long/".to_string().repeat(100),
841                version: Some("long-version-".to_string().repeat(50)),
842            },
843        ];
844
845        for original in test_cases {
846            let mut buffer = Vec::new();
847            original
848                .encode(&mut buffer)
849                .expect("encoding should succeed");
850
851            let decoded =
852                FoyerKey::decode(&mut Cursor::new(&buffer)).expect("decoding should succeed");
853
854            assert_eq!(
855                decoded, original,
856                "decode(encode(key)) should equal original key"
857            );
858        }
859    }
860}