somatize_core/
codec.rs

1//! Who writes down what only exists in one process, so it can be kept or sent.
2//!
3//! A hole of the core, and it took a third tenant to see it: it was written next
4//! to the wire when that was its only consumer, but a `Store` asks an opaque
5//! value the same question a socket does, and `data/` asks it a third time for
6//! Arrow IPC. What decides where a hole lives is what it serves.
7//!
8//! It does not move the frontier, it moves what falls on which side.
9//! [`Value::travels`] stays true — what comes out of `packed` **does** travel,
10//! being maps and bytes. The frontier goes from *the variant* to *the variant
11//! nobody registered a codec for*.
12//!
13//! The same pair sits at both ends in mirror image: the client packs the input
14//! and unpacks what came back, the worker unpacks the input and packs what it
15//! produced. Packing happens **before** a message is built, so `Answer`'s
16//! refusal is untouched and still guards.
17//!
18//! Failing is not symmetric. Going out, a value nobody can write down is an
19//! error: somebody over there is waiting. Coming back it is left behind and
20//! named if anybody reads it. And unlike a [`Keeper`](crate::Keeper) a codec
21//! that fails **does** stop the run — a cache recomputes, a wire has nothing to
22//! fall back on.
23
24use crate::{NodeId, Value};
25use std::fmt;
26use std::sync::Arc;
27
28/// The reserved key that says a map is not a map.
29///
30/// Here and not in an implementor: this shape is what makes two codecs the same
31/// codec, and two copies of it would drift the day one changed.
32const PACKED: &str = "__soma_opaque__";
33
34/// Where the bytes are, next to it.
35const BYTES: &str = "bytes";
36
37/// Something written down: what kind it was, and the bytes it became. The
38/// `kind` is named after the **type or the format** and never after the run —
39/// `torch.Tensor`, `arrow.RecordBatch` — since it is also how the far end knows
40/// who to ask to read it back.
41pub fn written_down(kind: impl Into<String>, bytes: Vec<u8>) -> Value {
42    Value::map(vec![
43        (PACKED.to_string(), Value::text(kind.into())),
44        (BYTES.to_string(), Value::Bytes(Arc::new(bytes))),
45    ])
46}
47
48/// What was written down in there, if that is what it is.
49pub fn as_written(value: &Value) -> Option<(&str, &[u8])> {
50    let Value::Map(pairs) = value else {
51        return None;
52    };
53    let kind = pairs.iter().find(|(key, _)| key == PACKED)?;
54    let bytes = pairs.iter().find(|(key, _)| key == BYTES)?;
55    match (&kind.1, &bytes.1) {
56        (Value::Text(kind), Value::Bytes(bytes)) => Some((kind, bytes)),
57        _ => None,
58    }
59}
60
61/// Whether there is anything written down in there at all, at any depth. Asked
62/// before the walk, so the ordinary value costs one look.
63pub fn anything_written(value: &Value) -> bool {
64    if as_written(value).is_some() {
65        return true;
66    }
67    match value {
68        Value::Map(pairs) => pairs.iter().any(|(_, value)| anything_written(value)),
69        Value::List(items) => items.iter().any(anything_written),
70        _ => false,
71    }
72}
73
74/// Writes down what cannot leave a process, and reads it back.
75pub trait Codec: Send + Sync {
76    /// This value in a shape that can leave the process, at any depth. Whatever
77    /// carries nothing opaque comes back as it was: this is asked of every value
78    /// that crosses.
79    fn packed(&self, value: &Value) -> Result<Value, CodecError>;
80
81    /// And the live one back, on the side that will use it.
82    fn unpacked(&self, value: &Value) -> Result<Value, CodecError>;
83}
84
85/// Every one of these packed, or the first that cannot be.
86pub fn packed_all(
87    codec: &dyn Codec,
88    values: &[(NodeId, Value)],
89) -> Result<Vec<(NodeId, Value)>, CodecError> {
90    values
91        .iter()
92        .map(|(id, value)| Ok((id.clone(), codec.packed(value)?)))
93        .collect()
94}
95
96/// And every one of these unpacked.
97pub fn unpacked_all(
98    codec: &dyn Codec,
99    values: &[(NodeId, Value)],
100) -> Result<Vec<(NodeId, Value)>, CodecError> {
101    values
102        .iter()
103        .map(|(id, value)| Ok((id.clone(), codec.unpacked(value)?)))
104        .collect()
105}
106
107/// Why something could not be written down, or read back.
108#[derive(Debug, Clone, PartialEq, Eq)]
109pub struct CodecError(String);
110
111impl CodecError {
112    /// A failure described by a message.
113    pub fn new(message: impl Into<String>) -> Self {
114        Self(message.into())
115    }
116
117    /// The message.
118    pub fn message(&self) -> &str {
119        &self.0
120    }
121}
122
123impl fmt::Display for CodecError {
124    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
125        f.write_str(&self.0)
126    }
127}
128
129impl std::error::Error for CodecError {}
somatize_core/codec.rs

somatize_core/
codec.rs
