cratestack_client_rust/client/
compact.rs

1//! Compact wire keys, client side (`docs/design/compact-wire-keys.md`,
2//! proposal C). Opt-in: a client without
3//! [`with_compact_keys`](CratestackClient::with_compact_keys) sends and
4//! reads exactly what it always did.
5//!
6//! With it on, every call the client can name an op for sends
7//! `Accept: <compact>, <plain>` and `Cratestack-Contract`, and a compact
8//! response is turned back into named bytes before the normal decode. The
9//! request body goes out compact only once a response for that op has come
10//! back compact (a per-op flag held in memory), so an old server is never
11//! sent a body it cannot read. `/rpc/batch` stays named.
12//!
13//! **Sealed (COSE) calls** are adaptive the same way: the compact types
14//! lead `Cratestack-Payload-Accept`, the request is sealed under
15//! `application/vnd.cratestack.compact+cbor` (or `+json`) only once the op
16//! answered compact, and a compact sealed answer is transcoded after it is
17//! opened. The server must opt the compact types in
18//! (`EnvelopeLayerBuilder::payload_media_types`); against one that has not,
19//! the answer is plain and the client stays named, and an unsigned `415`
20//! (the compact request type refused) turns the op back to named for the
21//! next call. That failed call is returned, not retried.
22
23use std::collections::{HashMap, HashSet, VecDeque};
24use std::sync::{Arc, Mutex, MutexGuard};
25
26use cratestack_core::compact::{Direction, KeyNode, KeyTables};
27use cratestack_core::{ContractSelector, OpContracts};
28use reqwest::Method;
29
30use crate::client::core::CratestackClient;
31use crate::codec::HttpClientCodec;
32use crate::error::{ClientError, HeaderPair};
33
34/// Encoding pins kept; past this the oldest first attempt's pin is evicted, so a
35/// prompt retry is covered and a long-lived instance keeps negotiating compact.
36const MAX_REMEMBERED_KEYS: usize = 1024;
37
38/// What a client opted into compact keys carries; shared by every clone.
39pub(crate) struct CompactClient {
40    pub(crate) tables: KeyTables,
41    by_op: HashMap<String, usize>,
42    /// Ops whose last 2xx response came back compact.
43    compact_ops: Mutex<HashSet<String>>,
44    /// `Idempotency-Key` -> whether its first attempt went out compact.
45    idempotent: Mutex<Pins>,
46}
47
48impl CompactClient {
49    pub(crate) fn new(mut tables: KeyTables, current: Option<OpContracts>) -> Self {
50        if let Some(current) = current {
51            // A server's tables also hold older accepted contracts; a
52            // client speaks only its own.
53            tables.ops.retain(|keys| {
54                current
55                    .iter()
56                    .any(|(op, digest)| *op == keys.op && *digest == keys.digest)
57            });
58        }
59        let by_op = tables
60            .ops
61            .iter()
62            .enumerate()
63            .map(|(index, keys)| (keys.op.clone(), index))
64            .collect();
65        Self {
66            tables,
67            by_op,
68            compact_ops: Mutex::default(),
69            idempotent: Mutex::default(),
70        }
71    }
72
73    pub(crate) fn node(&self, index: usize, direction: Direction) -> &KeyNode {
74        self.tables.node(&self.tables.ops[index], direction)
75    }
76
77    pub(crate) fn set_compact(&self, op: &str, on: bool) {
78        let mut ops = lock(&self.compact_ops);
79        if on {
80            ops.insert(op.to_owned());
81        } else {
82            ops.remove(op);
83        }
84    }
85
86    #[cfg(test)]
87    pub(crate) fn is_compact(&self, op: &str) -> bool {
88        lock(&self.compact_ops).contains(op)
89    }
90
91    fn decide(&self, op: &str, idempotency_key: Option<&str>) -> bool {
92        let wanted = lock(&self.compact_ops).contains(op);
93        let Some(key) = idempotency_key else {
94            return wanted;
95        };
96        let mut pins = lock(&self.idempotent);
97        if let Some(remembered) = pins.by_key.get(key) {
98            return *remembered;
99        }
100        if pins.by_key.len() >= MAX_REMEMBERED_KEYS
101            && let Some(oldest) = pins.order.pop_front()
102        {
103            pins.by_key.remove(&oldest);
104        }
105        pins.by_key.insert(key.to_owned(), wanted);
106        pins.order.push_back(key.to_owned());
107        wanted
108    }
109}
110
111/// Idempotency-Key -> first attempt went compact, in first-attempt order.
112#[derive(Default)]
113pub(crate) struct Pins {
114    pub(crate) by_key: HashMap<String, bool>,
115    order: VecDeque<String>,
116}
117
118fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
119    mutex
120        .lock()
121        .unwrap_or_else(|poisoned| poisoned.into_inner())
122}
123
124/// The compact plan of one call.
125pub(crate) struct CompactCall {
126    pub(crate) state: Arc<CompactClient>,
127    pub(crate) op: String,
128    pub(crate) index: usize,
129    pub(crate) send_compact: bool,
130    pub(crate) selector: String,
131}
132
133impl<C> CratestackClient<C>
134where
135    C: HttpClientCodec,
136{
137    /// Turn compact wire keys on, with `tables` holding one entry per op
138    /// (the client's own contracts). Off by default.
139    #[must_use]
140    pub fn with_compact_keys(mut self, tables: KeyTables) -> Self {
141        self.compact = Some(Arc::new(CompactClient::new(tables, None)));
142        self
143    }
144
145    /// Compact wire keys from the generated `KEY_TABLES` bytes, keeping the
146    /// entries whose digest is the op's current one in `contracts` (the
147    /// generated `OP_CONTRACTS`). Called by the generated
148    /// `Client::with_compact_keys`.
149    pub fn with_compact_key_bytes(
150        mut self,
151        tables: &[u8],
152        contracts: OpContracts,
153    ) -> Result<Self, ClientError> {
154        let tables = KeyTables::from_bytes(tables)
155            .map_err(|error| ClientError::BadInput(format!("invalid KEY_TABLES: {error}")))?;
156        self.compact = Some(Arc::new(CompactClient::new(tables, Some(contracts))));
157        Ok(self)
158    }
159
160    /// The plan for the call in flight, or `None` (compact off, no route named, or an op the tables lack such as `batch`).
161    pub(crate) fn compact_call(
162        &self,
163        method: &Method,
164        headers: &[HeaderPair<'_>],
165        has_body: bool,
166    ) -> Option<CompactCall> {
167        let state = self.compact.as_ref()?;
168        let template = self.compact_route.as_deref()?;
169        let (op, index) = match state.by_op.get(template) {
170            Some(index) => (template.to_owned(), *index),
171            None => {
172                let verb = if method == Method::HEAD {
173                    "GET"
174                } else {
175                    method.as_str()
176                };
177                let op = format!("{verb} {template}");
178                let index = *state.by_op.get(&op)?;
179                (op, index)
180            }
181        };
182        let key = headers
183            .iter()
184            .find(|(name, _)| name.eq_ignore_ascii_case("idempotency-key"))
185            .map(|(_, value)| *value);
186        let send_compact = has_body && state.decide(&op, key);
187        let selector = ContractSelector::of(&state.tables.ops[index].digest).to_header_value();
188        Some(CompactCall {
189            state: Arc::clone(state),
190            op,
191            index,
192            send_compact,
193            selector,
194        })
195    }
196}
197
198#[cfg(test)]
199#[path = "compact_idempotency_tests.rs"]
200mod idempotency_tests;
cratestack_client_rust/client/compact.rs

cratestack_client_rust/client/
compact.rs
