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
//! Studio's verbs and head follows, carried to their store's socket.
//!
//! The session's Lys pass is presented first (ACCESS-002 R2); a branch's
//! head is followed the same way, the store pushing each change
//! ([`follow`]).
//!
//! Each verb is one connection: hello, then `session.pass` with the
//! session's pass, then the verb. A pass the store refuses ends the request
//! there and the verb is never sent. `hello` touches no resource, so it is
//! answered with the store's own hello. `session.pass` belongs to the door,
//! so a browser can never present a pass of its own.
use std::path::Path;
use std::time::Duration;
use serde_json::value::RawValue;
use serde_json::{Value, json};
use crate::client::{Client, Ended, Refusal, Reply};
use crate::serve::pass::now;
/// What a carried verb came to.
#[derive(Debug)]
pub enum Carried {
/// The bytes the store sent for `result`.
Answered(Box<RawValue>),
/// The store's refusal, or the door's for a verb it keeps to itself.
Refused(Refusal),
}
/// Carry `method` with `params` to the store at `socket`, presenting
/// `token` first, each step bounded by `timeout`.
///
/// # Errors
///
/// [`Ended`] as the client names it: nothing asked, or, from the verb's
/// sending on, an outcome that is not known.
pub fn carry(
socket: &Path,
timeout: Duration,
token: &str,
method: &str,
params: Option<&RawValue>,
) -> Result<Carried, Ended> {
if method == "session.pass" {
return Ok(Carried::Refused(Refusal {
code: "invalid_operation".to_owned(),
message: "session.pass is the door's own and is never carried".to_owned(),
path: None,
// The door refuses this itself; no store sent a receipt.
verified: None,
}));
}
let mut client = Client::connect(socket, timeout)?;
if method == "hello" {
return Ok(Carried::Answered(client.hello().to_owned()));
}
let pass = serde_json::value::to_raw_value(&json!({ "pass": token }))
.map_err(|error| Ended::NotAsked(format!("the pass could not be written: {error}")))?;
if let Reply::Refused(refusal) = client.call("session.pass", Some(&pass))? {
return Ok(Carried::Refused(refusal));
}
Ok(match client.call(method, params)? {
Reply::Answered(answer) => Carried::Answered(answer),
Reply::Refused(refusal) => Carried::Refused(refusal),
})
}
/// What asking a store to follow a branch's head came to.
pub enum Followed {
/// The store took the follow and sent the head it starts from.
Following(Following),
/// The store refused the session's pass or the follow.
Refused(Refusal),
}
/// A follow a store took: the first head it sent, and the connection every
/// later frame arrives on.
pub struct Following {
/// The bytes the store sent for the first head's `result`.
pub first: Box<RawValue>,
client: Client,
/// When the presented pass expires, in seconds since the epoch, as the
/// store answered `session.pass`.
expires: Option<u64>,
timeout: Duration,
}
impl Following {
/// The store's next frame of the follow: a head, or the refusal that
/// ends it. The store sends a frame only when the head moves, so the
/// read waits as long as that takes, bounded by the pass's expiry and
/// `timeout` past it: the store ends every follow at that expiry, so a
/// read still waiting then means a store that stopped answering. With no
/// expiry known the read is unbounded.
///
/// # Errors
///
/// [`Ended`] as the client names it when no frame could be read.
pub fn next_frame(&mut self) -> Result<Reply, Ended> {
self.client.next_reply()
}
pub(super) fn disconnect_handle(&self) -> std::io::Result<std::os::unix::net::UnixStream> {
self.client.follow_socket()
}
/// Bound every later read by the pass's expiry and `timeout` past it,
/// set once when the follow starts. A store that has already closed the
/// connection refuses the bound on some systems (macOS answers EINVAL);
/// what it sent before closing is still read, and the read itself then
/// names the closed connection, so that refusal is not the follow's end.
fn bound_reads(&self) {
let bound = self.expires.map(|expires| {
let left = Duration::from_secs(expires.saturating_sub(now()));
left.saturating_add(self.timeout)
});
if let Err(ended) = self.client.read_within(bound) {
eprintln!("haematite door: a follow's read bound was not set: {ended}");
}
}
}
/// Ask the store at `socket` to follow `branch`'s head, presenting `token`
/// first, each step up to the first head bounded by `timeout`.
///
/// # Errors
///
/// [`Ended`] as the client names it. A follow changes nothing, so an
/// unknown outcome only means no head was read.
pub fn follow(
socket: &Path,
timeout: Duration,
token: &str,
branch: &str,
) -> Result<Followed, Ended> {
let mut client = Client::connect(socket, timeout)?;
let pass = raw(&json!({ "pass": token }))?;
let expires = match client.call("session.pass", Some(&pass))? {
Reply::Refused(refusal) => return Ok(Followed::Refused(refusal)),
Reply::Answered(answer) => expiry(&answer),
};
let params = raw(&json!({ "branch": branch }))?;
Ok(match client.call("head.follow", Some(¶ms))? {
Reply::Answered(first) => {
let following = Following {
first,
client,
expires,
timeout,
};
following.bound_reads();
Followed::Following(following)
}
Reply::Refused(refusal) => Followed::Refused(refusal),
})
}
fn raw(value: &Value) -> Result<Box<RawValue>, Ended> {
serde_json::value::to_raw_value(value)
.map_err(|error| Ended::NotAsked(format!("the request could not be written: {error}")))
}
/// The expiry, in seconds since the epoch, a store answered `session.pass`
/// with: `{holder, expires}`.
fn expiry(answer: &RawValue) -> Option<u64> {
let answer: Value = serde_json::from_str(answer.get()).ok()?;
answer.get("expires")?.as_u64()
}
#[cfg(test)]
#[path = "forward_tests.rs"]
mod tests;