yog 0.0.62

yog: the standalone server for litany loops — the world, the balls and the conversations, behind one wire
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

//! BEP 44 traffic: `get` reads the newest verified item under a key and salt,
//! `put` stores one at the nodes closest to its target. Both are one walk
//! (`lookup`) asking `get` — the walk that finds the closest nodes is the
//! walk that collects their write tokens — and `put` then spends those tokens
//! in one more round. Nothing here trusts the commons: an item is a value
//! only once its signature verifies under the key the caller asked for.

use super::Dht;
use super::bencode::{Dict, bytes, entry};
use super::krpc::{Message, Node};
use super::lookup::Pending;
use super::mutable::{Mutable, target_of};

impl Dht {
    /// The newest item signed under `key` for `salt`, or nothing: no node
    /// held one, or none held one that verifies.
    pub fn get(&mut self, key: [u8; 32], salt: Vec<u8>) -> Result<Option<Mutable>, String> {
        let target = target_of(&key, &salt);
        let out = self.search(
            target,
            "get",
            Dict::from([entry("target", bytes(&target.0))]),
        )?;
        Ok(out
            .replies
            .iter()
            .filter_map(|(_, r)| Mutable::from_reply(r, key, &salt))
            .max_by_key(|m| m.seq))
    }

    /// Store `item` at the K closest nodes that offered a write token; answers
    /// how many acknowledged. Zero is an error naming the first refusal —
    /// a stale sequence number, say — or the silence.
    pub fn put(&mut self, item: Mutable) -> Result<usize, String> {
        let target = item.target();
        let out = self.search(
            target,
            "get",
            Dict::from([entry("target", bytes(&target.0))]),
        )?;
        let holders: Vec<(Node, Vec<u8>)> = out
            .replies
            .iter()
            .filter_map(|(n, r)| Some((*n, r.get(b"token".as_slice())?.as_bytes()?.to_vec())))
            .take(self.config.k)
            .collect();
        let mut pending = Pending::new();
        for (node, token) in holders {
            self.ask(&mut pending, node.addr, "put", item.put_args(&token));
        }
        let mut acks = 0usize;
        let mut refusals = Vec::new();
        self.collect(&mut pending, &mut |addr, message| match message {
            Message::Reply { .. } => acks += 1,
            Message::Error { code, message, .. } => {
                refusals.push(format!("{addr}: {code} {message}"));
            }
        })?;
        if acks == 0 {
            return Err(refusals
                .into_iter()
                .next()
                .unwrap_or_else(|| format!("no DHT node stored the item at {target}")));
        }
        Ok(acks)
    }
}