rustev_remote_http/
exchange.rs

1//! Exchange records and their delivery (spec 009, 3.9), and reconciliation
2//! offers for charges learned late (3.4.4, 3.10.4).
3//!
4//! The adapter hands every `rustev.remote-exchange/1` record to a host
5//! sink injected at construction. A sink failure never changes an attempt's
6//! result; it is counted. Retention is the host's (R-19): the record holds
7//! digests only, and request and response bytes reach the sink only when
8//! the host asked for them.
9
10use std::collections::BTreeMap;
11use std::panic::{AssertUnwindSafe, catch_unwind};
12use std::sync::Arc;
13use std::sync::atomic::{AtomicU64, Ordering};
14
15use rustev_contract::ids::ArtifactId;
16use rustev_contract::output::RawOutput;
17use rustev_contract::remote::{Attribution, ExchangeMember, ExchangeRecord, ServedIdentity};
18use rustev_contract::run::Charge;
19use rustev_contract::schema;
20
21use crate::wire::digest_bytes;
22
23/// Request and response bytes, retained only under explicit retention.
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub struct RetainedBytes {
26    pub request: Vec<u8>,
27    pub response: Option<Vec<u8>>,
28}
29
30/// A charge learned after the attempt's own report, offered to the host
31/// for reconciliation of the shared ledger (spec 003, 3.5.5). The host
32/// decides; an offer changes nothing by itself.
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct ReconciliationOffer {
35    pub attempt_id: String,
36    /// In deployment units, rounded up.
37    pub observed_units: u64,
38    /// The adapter binding the attempt ran against.
39    pub binding: ArtifactId,
40}
41
42/// The host's exchange sink.
43pub trait ExchangeSink: Send + Sync {
44    /// Receive one record, and its bytes when retention is on.
45    fn deliver(
46        &self,
47        record: &ExchangeRecord,
48        retained: Option<&RetainedBytes>,
49    ) -> Result<(), String>;
50
51    /// Receive a reconciliation offer. The default discards it.
52    fn offer(&self, offer: &ReconciliationOffer) -> Result<(), String> {
53        let _ = offer;
54        Ok(())
55    }
56
57    /// [`Self::deliver`], naming where the host keeps the record: an opaque
58    /// locator a `rustev.run/2` record may reference (spec 020, 3.2), never
59    /// a URL with credentials. The default delivers and names none.
60    fn deliver_located(
61        &self,
62        record: &ExchangeRecord,
63        retained: Option<&RetainedBytes>,
64    ) -> Result<Option<String>, String> {
65        self.deliver(record, retained).map(|()| None)
66    }
67}
68
69/// What delivering one record established, for remote evidence.
70#[derive(Debug, Clone, PartialEq, Eq)]
71pub struct DeliveryNote {
72    /// The delivered (bounded) record's record digest.
73    pub digest: Option<String>,
74    /// The locator the sink named; `Err` when delivery failed.
75    pub located: Result<Option<String>, ()>,
76}
77
78/// A sink that keeps nothing.
79pub struct DiscardExchanges;
80
81impl ExchangeSink for DiscardExchanges {
82    fn deliver(&self, _: &ExchangeRecord, _: Option<&RetainedBytes>) -> Result<(), String> {
83        Ok(())
84    }
85}
86
87/// Delivery counts. In memory only.
88#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
89pub struct ExchangeStats {
90    pub delivered: u64,
91    /// Refused or panicked deliveries; the attempts' results were unchanged.
92    pub failed: u64,
93    pub offers: u64,
94    pub offers_failed: u64,
95}
96
97/// Delivers records and offers to the host sink, counting failures.
98pub struct Exchanges {
99    sink: Arc<dyn ExchangeSink>,
100    retain_bytes: bool,
101    delivered: AtomicU64,
102    failed: AtomicU64,
103    offers: AtomicU64,
104    offers_failed: AtomicU64,
105}
106
107impl Exchanges {
108    pub fn new(sink: Arc<dyn ExchangeSink>, retain_bytes: bool) -> Self {
109        Exchanges {
110            sink,
111            retain_bytes,
112            delivered: AtomicU64::new(0),
113            failed: AtomicU64::new(0),
114            offers: AtomicU64::new(0),
115            offers_failed: AtomicU64::new(0),
116        }
117    }
118
119    pub fn retains_bytes(&self) -> bool {
120        self.retain_bytes
121    }
122
123    /// Bound the record and hand it over. Never fails: a refusal or a panic
124    /// in the sink is counted, and the note says so.
125    pub fn deliver(&self, record: ExchangeRecord, bytes: Option<RetainedBytes>) -> DeliveryNote {
126        use rustev_contract::Document;
127        let record = record.bounded();
128        let digest = record.record_digest().ok().map(|d| d.as_str().to_string());
129        let retained = if self.retain_bytes { bytes } else { None };
130        let located = match catch_unwind(AssertUnwindSafe(|| {
131            self.sink.deliver_located(&record, retained.as_ref())
132        })) {
133            Ok(Ok(locator)) => Ok(locator),
134            _ => Err(()),
135        };
136        let counter = if located.is_ok() {
137            &self.delivered
138        } else {
139            &self.failed
140        };
141        counter.fetch_add(1, Ordering::SeqCst);
142        DeliveryNote { digest, located }
143    }
144
145    /// Offer a charge for reconciliation. Never fails; failures are counted.
146    pub fn offer(&self, offer: ReconciliationOffer) {
147        let ok = matches!(
148            catch_unwind(AssertUnwindSafe(|| self.sink.offer(&offer))),
149            Ok(Ok(()))
150        );
151        let counter = if ok {
152            &self.offers
153        } else {
154            &self.offers_failed
155        };
156        counter.fetch_add(1, Ordering::SeqCst);
157    }
158
159    pub fn stats(&self) -> ExchangeStats {
160        ExchangeStats {
161            delivered: self.delivered.load(Ordering::SeqCst),
162            failed: self.failed.load(Ordering::SeqCst),
163            offers: self.offers.load(Ordering::SeqCst),
164            offers_failed: self.offers_failed.load(Ordering::SeqCst),
165        }
166    }
167}
168
169/// `sha256:` of an output's record canonical bytes (spec 004, 5.6); `None`
170/// for an output without that form (a non-finite number).
171pub fn output_digest(output: &RawOutput) -> Option<String> {
172    rustev_contract::canonical::record_canonical_bytes(output)
173        .ok()
174        .map(|b| digest_bytes(&b))
175}
176
177/// An exchange record with every optional member empty, for an adapter to
178/// fill in. `members` must be in attempt-id order.
179pub fn blank_record(
180    protocol: &str,
181    binding: ArtifactId,
182    members: Vec<ExchangeMember>,
183) -> ExchangeRecord {
184    ExchangeRecord {
185        schema: schema::REMOTE_EXCHANGE.to_string(),
186        protocol: protocol.to_string(),
187        binding,
188        members,
189        request_digest: None,
190        response_digest: None,
191        bytes_sent: false,
192        status: None,
193        retry_after: None,
194        requested_model: None,
195        served: ServedIdentity::unknown(),
196        route: vec![],
197        generation_id: None,
198        usage: BTreeMap::new(),
199        reported_cost: None,
200        exchange_charge: Charge::Unknown,
201        attribution: Attribution::Single,
202        privacy_requested: BTreeMap::new(),
203        provider_extras: BTreeMap::new(),
204        late: false,
205        truncated: false,
206        phase_expiry: None,
207        principal: None,
208        connection: None,
209    }
210}
211
212#[cfg(test)]
213mod tests {
214    use super::*;
215    use rustev_contract::remote::{MemberOutcome, RemoteCode};
216    use std::sync::Mutex;
217
218    const A: &str = "sha256:00000000000000000000000000000000000000000000000000000000000000aa";
219
220    #[derive(Default)]
221    struct Keep {
222        records: Mutex<Vec<(ExchangeRecord, Option<RetainedBytes>)>>,
223        fail: bool,
224        panic: bool,
225    }
226
227    impl ExchangeSink for Keep {
228        fn deliver(&self, r: &ExchangeRecord, b: Option<&RetainedBytes>) -> Result<(), String> {
229            if self.panic {
230                panic!("SYNTHETIC sink panic");
231            }
232            self.records.lock().unwrap().push((r.clone(), b.cloned()));
233            if self.fail {
234                Err("SYNTHETIC refusal".into())
235            } else {
236                Ok(())
237            }
238        }
239    }
240
241    fn record() -> ExchangeRecord {
242        blank_record(
243            "rustev.remote/1",
244            ArtifactId::parse(A).unwrap(),
245            vec![ExchangeMember {
246                attempt_id: "d/s//1".into(),
247                outcome: MemberOutcome::Failure {
248                    code: RemoteCode::RemoteError,
249                    detail: "x".repeat(1000),
250                },
251                charge: Charge::Unknown,
252            }],
253        )
254    }
255
256    fn bytes() -> Option<RetainedBytes> {
257        Some(RetainedBytes {
258            request: b"{}".to_vec(),
259            response: None,
260        })
261    }
262
263    #[test]
264    fn records_are_bounded_and_bytes_kept_only_under_retention() {
265        let keep = Arc::new(Keep::default());
266        let x = Exchanges::new(keep.clone(), false);
267        x.deliver(record(), bytes());
268        let (r, b) = keep.records.lock().unwrap()[0].clone();
269        assert!(r.is_bounded() && r.truncated);
270        assert!(b.is_none(), "digest-only by default");
271        let x = Exchanges::new(keep.clone(), true);
272        x.deliver(record(), bytes());
273        assert!(keep.records.lock().unwrap()[1].1.is_some());
274    }
275
276    #[test]
277    fn sink_failures_and_panics_are_counted() {
278        let x = Exchanges::new(
279            Arc::new(Keep {
280                fail: true,
281                ..Keep::default()
282            }),
283            false,
284        );
285        x.deliver(record(), None);
286        let p = Exchanges::new(
287            Arc::new(Keep {
288                panic: true,
289                ..Keep::default()
290            }),
291            false,
292        );
293        p.deliver(record(), None);
294        assert_eq!(x.stats().failed, 1);
295        assert_eq!(p.stats().failed, 1);
296        assert_eq!(p.stats().delivered, 0);
297        // The default offer handler accepts and discards.
298        p.offer(ReconciliationOffer {
299            attempt_id: "a".into(),
300            observed_units: 1,
301            binding: ArtifactId::parse(A).unwrap(),
302        });
303        assert_eq!(p.stats().offers, 1);
304    }
305}
rustev_remote_http/exchange.rs

rustev_remote_http/
exchange.rs
