iris-chat 0.1.49

Iris Chat command line client and shared encrypted chat core
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

use super::*;

struct PausedCachedRead {
    local: Arc<AttachmentChunkStore>,
    root: Hash,
    root_read: tokio::sync::Notify,
    resume: tokio::sync::Notify,
}

#[async_trait]
impl Store for PausedCachedRead {
    async fn get(&self, hash: &Hash) -> Result<Option<Vec<u8>>, StoreError> {
        let bytes = self.local.get(hash).await?;
        if *hash == self.root {
            self.root_read.notify_one();
            self.resume.notified().await;
        }
        Ok(bytes)
    }
    async fn has(&self, hash: &Hash) -> Result<bool, StoreError> {
        self.local.has(hash).await
    }
    async fn put(&self, _: Hash, _: Vec<u8>) -> Result<bool, StoreError> {
        Err(StoreError::Other("read-only fixture".into()))
    }
    async fn delete(&self, _: &Hash) -> Result<bool, StoreError> {
        Err(StoreError::Other("read-only fixture".into()))
    }
}

#[tokio::test]
async fn cached_tree_read_survives_concurrent_quota_eviction_without_network() {
    let dir = tempfile::tempdir().unwrap();
    let local = AttachmentChunkStore::open(dir.path(), 32 * 1024 * 1024).unwrap();
    let expected: Vec<u8> = (0..7 * 1024 * 1024)
        .map(|i| (9 + i / (512 * 1024)) as u8)
        .collect();
    // Received/recoverable cache: no pending outgoing tree protects these blocks.
    let tree = HashTree::new(HashTreeConfig::new(local.clone()));
    let (cid, _) = tree.put(&expected).await.unwrap();
    let root = cid.hash;
    let mut old = Vec::new();
    for value in 40..44 {
        let bytes = vec![value; 1024 * 1024];
        let hash = hashtree_core::sha256(&bytes);
        local.put(hash, bytes).await.unwrap();
        old.push(hash);
    }
    let source = Arc::new(PausedCachedRead {
        local: local.clone(),
        root,
        root_read: tokio::sync::Notify::new(),
        resume: tokio::sync::Notify::new(),
    });
    let reading_local = local.clone();
    let reading_source = source.clone();
    let read = tokio::spawn(async move {
        readiness::read(
            &cid,
            Some(reading_local),
            reading_source,
            MAX_ATTACHMENT_BYTES,
            Duration::ZERO,
        )
        .await
    });
    tokio::time::timeout(Duration::from_secs(5), source.root_read.notified())
        .await
        .unwrap();
    // Each new block must be admitted by evicting unrelated eligible cache.
    // A global read veto fails this; unprotected tree leaves fail the final read.
    for value in 80..88 {
        let bytes = vec![value; 1024 * 1024];
        local
            .put(hashtree_core::sha256(&bytes), bytes)
            .await
            .unwrap();
    }
    assert!(local.get(&old[0]).await.unwrap().is_none());
    source.resume.notify_one();
    assert_eq!(
        tokio::time::timeout(Duration::from_secs(5), read)
            .await
            .unwrap()
            .unwrap()
            .unwrap(),
        expected
    );
    assert!(
        local.delete(&root).await.unwrap(),
        "read protection is released on completion"
    );
}