atlas_rdma/
handshake.rs

1// SPDX-License-Identifier: AGPL-3.0-only
2//
3// The client side of the rail handshake as PURE byte functions, generic over
4// `Read`/`Write` and taking QP identities as plain `(qpn, psn, gid)` tuples —
5// no `Verbs` (and thus no ibverbs) required. `railset::RailSet` delegates to
6// these in production; tests drive them against a scripted fake stream to pin
7// the client's complete emitted byte sequence (`tests/transcript_golden.rs`),
8// which is the only test class that catches a write REORDER.
9//
10// Wire order per dialect (all little-endian, a frozen external contract):
11//   client: [u8 n_rails]                                   (`write_n_rails`)
12//   server: RO  → [u8 n]{VerbsServerParams}×n              (`wire::read_server_rails`)
13//           RW  → [u8 n echo]{CacheServerParams}×n         (`read_rw_server_params`)
14//   client: [u8 n_rails] {VerbsClientParams}×n             (`write_client_params`)
15//   server: [u8 ack] == STATUS_OK                          (`read_ack`)
16// The n_rails byte is deliberately sent TWICE by the client — that is the wire
17// format, not a redundancy to deduplicate.
18
19use anyhow::{Context, Result, bail};
20use std::io::{Read, Write};
21
22use crate::wire::{CacheServerParams, STATUS_OK, VerbsClientParams};
23
24/// Step 1: tell the peer how many rails we want to stripe across.
25pub fn write_n_rails<W: Write>(w: &mut W, n: usize) -> Result<()> {
26    w.write_all(&[n as u8]).context("send n_rails")?;
27    Ok(())
28}
29
30/// RW-blade dialect (KV overflow / snapshots): read the peer's `[u8 n]` echo
31/// (must equal the negotiated `want`; deliberately NOT bounded to 8 — the RO
32/// dialect's 1..=8 bound is its own, and unifying validation would change
33/// accepted wire inputs) followed by `n` `CacheServerParams`.
34pub fn read_rw_server_params<R: Read>(
35    r: &mut R,
36    want: usize,
37    peer: &str,
38) -> Result<Vec<CacheServerParams>> {
39    let mut b1 = [0u8; 1];
40    if let Err(e) = r.read_exact(&mut b1) {
41        // A clean EOF here is the peer REJECTING us, not a transport fault: it
42        // hangs up after logging its own reason (e.g. "paging blade cap") and
43        // the v2 wire has no error frame to carry that reason back. Bare
44        // `read_exact` context yields "failed to fill whole buffer", which
45        // sends operators hunting a network problem that does not exist.
46        if e.kind() == std::io::ErrorKind::UnexpectedEof {
47            bail!(
48                "{peer} closed the connection during the rail handshake without sending rail \
49                 params. The peer REJECTED this client — read ITS log for the reason. Most \
50                 common: the arena this client requested exceeds the peer's --max-blade-gb \
51                 cap (peer logs \"paging blade cap\"); also possible: a blob_bytes/kind \
52                 mismatch against an arena a prior client already fixed."
53            );
54        }
55        return Err(e).context("read peer n_rails");
56    }
57    if b1[0] as usize != want {
58        bail!("{peer} granted {} rails, wanted {want}", b1[0]);
59    }
60    let mut server = Vec::with_capacity(want);
61    for _ in 0..want {
62        server
63            .push(CacheServerParams::read_from(r).with_context(|| format!("read {peer} params"))?);
64    }
65    Ok(server)
66}
67
68/// Reply with our rail count (again — wire format) + each rail's QP identity.
69pub fn write_client_params<W: Write>(w: &mut W, ids: &[(u32, u32, [u8; 16])]) -> Result<()> {
70    w.write_all(&[ids.len() as u8])
71        .context("send client n_rails")?;
72    for &(qpn, psn, gid) in ids {
73        VerbsClientParams { qpn, psn, gid }
74            .write_to(w)
75            .context("send verbs client params")?;
76    }
77    Ok(())
78}
79
80/// Final step: the peer's one-byte ready ack, which must be `STATUS_OK`.
81pub fn read_ack<R: Read>(r: &mut R, peer: &str) -> Result<()> {
82    let mut ack = [0u8; 1];
83    r.read_exact(&mut ack).context("read verbs ready ack")?;
84    if ack[0] != STATUS_OK {
85        bail!("{peer} refused connection (ack {})", ack[0]);
86    }
87    Ok(())
88}
atlas_rdma/handshake.rs

atlas_rdma/
handshake.rs
