lernie 0.1.60

lernie: the operator seat — the window and wire client for a yog server
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

//! 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 flight, every holder at once, each with its own deadline.
//! 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::flight::Flight;
use super::krpc::{Message, Node};
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")?;
        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; a walk that reached no
    /// token holder at all (every node near the target silent or refusing
    /// `get`) is its own error, sent nothing, and is not a silent `put`
    /// (yog bl-f519: the two read as one, and were mistaken for each other).
    pub fn put(&mut self, item: Mutable) -> Result<usize, String> {
        let target = item.target();
        let out = self.search(target, "get")?;
        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();
        if holders.is_empty() {
            return Err(format!("no DHT node near {target} offered a write token"));
        }
        let mut flight = Flight::new();
        for (node, token) in holders {
            self.ask(&mut flight, node.addr, false, "put", item.put_args(&token));
        }
        let mut acks = 0usize;
        let mut refusals = Vec::new();
        while !flight.is_empty() {
            match self.land(&mut flight)? {
                Some((_, Message::Reply { .. })) => acks += 1,
                Some((query, Message::Error { code, message, .. })) => {
                    refusals.push(format!("{}: {code} {message}", query.addr));
                }
                None => {}
            }
        }
        if acks == 0 {
            return Err(refusals
                .into_iter()
                .next()
                .unwrap_or_else(|| format!("no DHT node stored the item at {target}")));
        }
        Ok(acks)
    }
}