haematite_cli/client/
write.rs

1//! `haematite write`: one change to a branch, and always whether it was applied.
2//!
3//! The change goes through `mutation.apply` under a request id. When the
4//! outcome is not known after sending, the client connects once more and asks
5//! for the receipt; a missing receipt is never read as "not applied".
6
7use std::io::Read;
8use std::path::Path;
9use std::process::ExitCode;
10use std::time::Duration;
11
12use serde::{Deserialize, Serialize};
13use serde_json::value::RawValue;
14
15use super::{Client, Common, Ended, Reply, Text, finish, one_value, socket_of};
16
17const USAGE: &str = "\
18usage: haematite write (--socket <path> | --config <serve.json>) [--timeout-ms N] --branch <name>
19                  [--expected <generation>.<sequence>] [--request-id <text>] <kind> <value | ->
20<kind> is a change kind as the service names it (document.put, vector.put, documents.put_many,
21vectors.put_many, edges.put_many, documents.delete_many and the rest); <value> is one JSON value, sent as the text given; - reads it from standard input.
22branch.fork: --branch is the source, and the value names the new branch and the source version
23it starts from: {\"name\":\"<new>\",\"from\":{\"generation\":\"<g>\",\"sequence\":\"<s>\"}}.
24branch.discard: the value names the --branch itself, never another: {\"name\":\"<branch>\"}.
25Without --expected, the change is applied to the branch as it stands when the command runs. A
26caller who read a record and is changing it gives the version it read as --expected, so a change
27made in between is refused rather than overwritten.
28Without --request-id, one is made and printed as `haematite: request <id>` before sending. If the
29outcome is not known, running the same command with that --request-id is safe to repeat.
30Repeated after the change was applied, the same command answers with the same receipt; with
31different content under the same --request-id it is refused.
32--timeout-ms bounds connecting, writing and reading; the default is 30000.
33Exit 0: applied, the receipt on standard output. Exit 1: could not ask; nothing was sent.
34Exit 2: refused by the service, its code and message on standard error. Exit 3: the outcome
35is not known; standard error gives the request id to repeat it with.";
36
37/// The most branches asked for on one page while finding a head, as the
38/// Studio asks; never more than the service's own page.
39const BRANCH_PAGE: u32 = 200;
40
41struct Write {
42    target: super::Target,
43    timeout: Duration,
44    branch: String,
45    expected: Option<Box<RawValue>>,
46    request_id: Option<String>,
47    kind: String,
48    value: Text,
49}
50
51#[derive(Serialize)]
52struct Apply<'a> {
53    request_id: &'a str,
54    branch: &'a str,
55    expected: &'a RawValue,
56    change: Change<'a>,
57}
58
59#[derive(Serialize)]
60struct Change<'a> {
61    kind: &'a str,
62    value: &'a RawValue,
63}
64
65#[derive(Serialize)]
66struct ReceiptQuery<'a> {
67    branch: &'a str,
68    request_id: &'a str,
69}
70
71#[derive(Serialize)]
72struct BranchQuery<'a> {
73    #[serde(skip_serializing_if = "Option::is_none")]
74    after: Option<&'a str>,
75    limit: u32,
76}
77
78#[derive(Deserialize)]
79struct BranchPage {
80    entries: Vec<Branch>,
81    next_after: Option<String>,
82}
83
84#[derive(Deserialize)]
85struct Branch {
86    branch: String,
87    version: Box<RawValue>,
88}
89
90/// `haematite write`, ending in its exit code.
91#[must_use]
92pub fn run(args: impl Iterator<Item = String>) -> ExitCode {
93    finish(write(args))
94}
95
96fn write(args: impl Iterator<Item = String>) -> Result<Reply, Ended> {
97    let asked = parse(args).map_err(|why| Ended::Usage(format!("{why}\n{USAGE}")))?;
98    let value = one_value(asked.value, "value")?;
99    let socket = socket_of(&asked.target)?;
100    // Only an ordinary change is applied on the version before its receipt's.
101    let ordinary = !matches!(asked.kind.as_str(), "branch.fork" | "branch.discard");
102    let replays = asked.request_id.is_some() && ordinary;
103    let request_id = asked.request_id.map_or_else(made_request_id, Ok)?;
104    let mut client = Client::connect(&socket, asked.timeout)?;
105    let branch = asked.branch.as_str();
106    let expected = match asked.expected {
107        Some(expected) => expected,
108        None => match origin(&mut client, branch, &request_id, replays)? {
109            Ok(origin) => origin,
110            Err(refused) => return Ok(refused),
111        },
112    };
113    let change = Change {
114        kind: &asked.kind,
115        value: &value,
116    };
117    let apply = Apply {
118        request_id: &request_id,
119        branch,
120        expected: &expected,
121        change,
122    };
123    let params = serde_json::value::to_raw_value(&apply)
124        .map_err(|error| Ended::NotAsked(format!("the request could not be written: {error}")))?;
125    match client.call("mutation.apply", Some(&params)) {
126        Err(Ended::Unknown(why)) => reconcile(&socket, asked.timeout, branch, &request_id, &why),
127        other => other,
128    }
129}
130
131fn parse(mut args: impl Iterator<Item = String>) -> Result<Write, String> {
132    let mut common = Common::default();
133    let mut branch = None;
134    let mut expected = None;
135    let mut request_id = None;
136    let mut rest = Vec::new();
137    while let Some(arg) = args.next() {
138        if common.take(&arg, &mut args)? {
139            continue;
140        }
141        match arg.as_str() {
142            "--branch" if branch.is_none() => branch = Some(value_of(&arg, &mut args)?),
143            "--expected" if expected.is_none() => {
144                expected = Some(version(&value_of(&arg, &mut args)?)?);
145            }
146            "--request-id" if request_id.is_none() => {
147                request_id = Some(value_of(&arg, &mut args)?);
148            }
149            flag if flag.starts_with("--") => {
150                return Err(format!("unknown or repeated flag {flag:?}"));
151            }
152            _ => rest.push(arg),
153        }
154    }
155    let (target, timeout) = common.finish()?;
156    let branch = branch.ok_or("name the branch with --branch")?;
157    let mut rest = rest.into_iter();
158    let kind = rest.next().ok_or("name the change kind")?;
159    let value = rest
160        .next()
161        .map(Text::of)
162        .ok_or("give the change's value, or -")?;
163    if let Some(extra) = rest.next() {
164        return Err(format!("unexpected argument {extra:?}"));
165    }
166    Ok(Write {
167        target,
168        timeout,
169        branch,
170        expected,
171        request_id,
172        kind,
173        value,
174    })
175}
176
177fn value_of(flag: &str, args: &mut impl Iterator<Item = String>) -> Result<String, String> {
178    args.next()
179        .filter(|value| !value.is_empty())
180        .ok_or_else(|| format!("{flag} needs a value"))
181}
182
183/// `<generation>.<sequence>` as the service writes a version.
184fn version(text: &str) -> Result<Box<RawValue>, String> {
185    let refused = || format!("--expected {text:?} is not <generation>.<sequence>");
186    let (generation, sequence) = text.split_once('.').ok_or_else(refused)?;
187    let generation: u64 = generation.parse().map_err(|_| refused())?;
188    let sequence: u64 = sequence.parse().map_err(|_| refused())?;
189    version_value(generation, sequence).map_err(|error| error.to_string())
190}
191
192fn version_value(generation: u64, sequence: u64) -> serde_json::Result<Box<RawValue>> {
193    RawValue::from_string(format!(
194        r#"{{"generation":"{generation}","sequence":"{sequence}"}}"#
195    ))
196}
197
198/// The version a change without --expected is sent against.
199///
200/// A given request id may name a change already applied. The service replays
201/// its receipt only for the same request, `expected` included, so the version
202/// that change was applied on is sent again and the service judges whether
203/// the content is the same: the receipt, or `request_conflict` with nothing
204/// applied. Otherwise, and for a made id, a fork or a discard, the branch's
205/// head, and the service answers.
206fn origin(
207    client: &mut Client,
208    branch: &str,
209    request_id: &str,
210    replays: bool,
211) -> Result<Result<Box<RawValue>, Reply>, Ended> {
212    if replays {
213        match earlier(client, branch, request_id)? {
214            Err(refused) => return Ok(Err(refused)),
215            Ok(Some(receipt)) => {
216                if let Some(origin) = applied_on(&receipt) {
217                    return Ok(Ok(origin));
218                }
219            }
220            Ok(None) => {}
221        }
222    }
223    head(client, branch, request_id)
224}
225
226/// The receipt the branch already holds for `request_id`, if any. Nothing that
227/// changes anything has been sent, so a failure here is could not ask.
228fn earlier(
229    client: &mut Client,
230    branch: &str,
231    request_id: &str,
232) -> Result<Result<Option<Box<RawValue>>, Reply>, Ended> {
233    let unread = |why: &dyn std::fmt::Display| {
234        Ended::NotAsked(format!(
235            "an earlier receipt for {request_id} could not be read: {why}"
236        ))
237    };
238    let query = ReceiptQuery { branch, request_id };
239    let params = serde_json::value::to_raw_value(&query).map_err(|error| unread(&error))?;
240    match client.call("mutation.receipt", Some(&params)) {
241        Ok(Reply::Answered(receipt)) if receipt.get() == "null" => Ok(Ok(None)),
242        Ok(Reply::Answered(receipt)) => Ok(Ok(Some(receipt))),
243        Ok(refused @ Reply::Refused(_)) => Ok(Err(refused)),
244        Err(ended) => Err(unread(&ended.why())),
245    }
246}
247
248#[derive(Deserialize)]
249struct Applied {
250    version: AppliedVersion,
251}
252
253#[derive(Deserialize)]
254struct AppliedVersion {
255    generation: String,
256    sequence: String,
257}
258
259/// The version a change was applied on. This depends on the service's rule
260/// for an ordinary change (storage/mutations.rs `mutate_with_import_bound`):
261/// one applied on `{generation, sequence}` is written as `{generation,
262/// sequence + 1}`; `tests/client_write.rs` pins that rule beside this client.
263/// It exists because the served receipt carries no `request_origin`; once the
264/// attributed admission path is served, its receipt does and is read instead.
265/// `None` for sequence 0, or a receipt that does not read as one.
266fn applied_on(receipt: &RawValue) -> Option<Box<RawValue>> {
267    let applied: Applied = serde_json::from_str(receipt.get()).ok()?;
268    let generation: u64 = applied.version.generation.parse().ok()?;
269    let sequence: u64 = applied.version.sequence.parse().ok()?;
270    version_value(generation, sequence.checked_sub(1)?).ok()
271}
272
273/// 128 bits from the operating system's random source, in hex, printed
274/// before anything is sent so the caller has it whatever happens next.
275fn made_request_id() -> Result<String, Ended> {
276    let mut bytes = [0_u8; 16];
277    std::fs::File::open("/dev/urandom")
278        .and_then(|mut source| source.read_exact(&mut bytes))
279        .map_err(|error| Ended::NotAsked(format!("no request id could be made: {error}")))?;
280    let made = format!("{:032x}", u128::from_be_bytes(bytes));
281    eprintln!("haematite: request {made}");
282    Ok(made)
283}
284
285/// The branch's head as the service lists it, read the way the Studio reads
286/// it. A branch that is not listed is asked for its receipt, a read the
287/// service refuses by name for a missing branch; that refusal is passed on.
288fn head(
289    client: &mut Client,
290    branch: &str,
291    request_id: &str,
292) -> Result<Result<Box<RawValue>, Reply>, Ended> {
293    let unread = |why: &dyn std::fmt::Display| {
294        Ended::NotAsked(format!("the head of {branch} could not be read: {why}"))
295    };
296    let limit = BRANCH_PAGE.min(client.max_page().get());
297    let mut after = None;
298    loop {
299        let query = BranchQuery {
300            after: after.as_deref(),
301            limit,
302        };
303        let params = serde_json::value::to_raw_value(&query).map_err(|error| unread(&error))?;
304        let page = match client.call("branches.list", Some(&params)) {
305            Ok(Reply::Answered(page)) => page,
306            Ok(refused @ Reply::Refused(_)) => return Ok(Err(refused)),
307            Err(ended) => return Err(unread(&ended.why())),
308        };
309        let page: BranchPage = serde_json::from_str(page.get()).map_err(|error| unread(&error))?;
310        if let Some(found) = page
311            .entries
312            .into_iter()
313            .find(|entry| entry.branch == branch)
314        {
315            return Ok(Ok(found.version));
316        }
317        let Some(next) = page.next_after else {
318            break;
319        };
320        after = Some(next);
321    }
322    let query = ReceiptQuery { branch, request_id };
323    let params = serde_json::value::to_raw_value(&query).map_err(|error| unread(&error))?;
324    match client.call("mutation.receipt", Some(&params)) {
325        Ok(refused @ Reply::Refused(_)) => Ok(Err(refused)),
326        Ok(Reply::Answered(receipt)) if receipt.get() == "null" => {
327            Err(unread(&format!("{branch} is not listed")))
328        }
329        // A discard that landed: the branch is gone and its receipt remains.
330        // Which change it was cannot be checked from here, so nothing is sent.
331        Ok(Reply::Answered(_)) => Err(Ended::Unknown(format!(
332            "{branch} is not listed, and the service holds a receipt for request \
333             {request_id} on it, so a change under this id was applied; whether it was this \
334             change is confirmed by running the same command with the --expected first sent"
335        ))),
336        Err(ended) => Err(unread(&ended.why())),
337    }
338}
339
340/// The outcome of a request that was sent and not answered: its receipt, on a
341/// new connection. A receipt found is the answer; anything else leaves the
342/// outcome not known, with the request id to repeat it under.
343fn reconcile(
344    socket: &Path,
345    timeout: Duration,
346    branch: &str,
347    request_id: &str,
348    why: &str,
349) -> Result<Reply, Ended> {
350    let unknown = |found: String| {
351        Ended::Unknown(format!(
352            "request {request_id}: {why}; {found}. Running the same command with --request-id \
353             {request_id} is safe to repeat"
354        ))
355    };
356    let query = ReceiptQuery { branch, request_id };
357    let asked = serde_json::value::to_raw_value(&query)
358        .map_err(|error| Ended::NotAsked(error.to_string()))
359        .and_then(|params| {
360            Client::connect(socket, timeout)?.call("mutation.receipt", Some(&params))
361        });
362    match asked {
363        Ok(Reply::Answered(receipt)) if receipt.get() != "null" => Ok(Reply::Answered(receipt)),
364        Ok(Reply::Answered(_)) => Err(unknown("the service has no receipt for it yet".to_owned())),
365        Ok(Reply::Refused(refusal)) => Err(unknown(format!(
366            "its receipt was refused: {}: {}",
367            refusal.code, refusal.message
368        ))),
369        Err(ended) => Err(unknown(format!(
370            "its receipt could not be asked: {}",
371            ended.why()
372        ))),
373    }
374}
haematite_cli/client/write.rs

haematite_cli/client/
write.rs
