clawft_kernel/cog_ingest/
types.rs

1//! The `POST /api/v1/store/ingest` request contract and its validation.
2//!
3//! Body: `{"vectors":[[id,[8 floats]],...],"dedup":true}`. Anything else is
4//! rejected: wrong shape, wrong dimensionality, non-finite floats, an empty
5//! or oversize batch, unknown keys.
6
7use serde_json::Value;
8
9/// Dimensionality of an ingested vector (the ADR-069 feature packet).
10pub const DIMS: usize = 8;
11/// Most vectors in one request.
12pub const MAX_VECTORS_PER_BATCH: usize = 256;
13/// Largest request body the bridge reads.
14pub const MAX_BODY_BYTES: usize = 64 * 1024;
15/// Path every cog posts to.
16pub const INGEST_PATH: &str = "/api/v1/store/ingest";
17
18/// One vector of a batch.
19#[derive(Debug, Clone, Copy, PartialEq, serde::Serialize, serde::Deserialize)]
20pub struct IngestVector {
21    /// Caller-chosen id.
22    pub id: u64,
23    /// The 8 values.
24    pub values: [f32; DIMS],
25}
26
27/// A validated batch.
28#[derive(Debug, Clone, PartialEq)]
29pub struct IngestBatch {
30    /// Vectors (1..=[`MAX_VECTORS_PER_BATCH`]).
31    pub vectors: Vec<IngestVector>,
32    /// Skip vectors the store already holds (same id, or bit-identical values).
33    pub dedup: bool,
34}
35
36/// Why a request was refused. Maps to an HTTP status.
37#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
38pub enum IngestError {
39    /// Malformed request (400).
40    #[error("malformed request: {0}")]
41    Malformed(String),
42    /// Body or batch over a limit (413).
43    #[error("too large: {0}")]
44    TooLarge(String),
45    /// Missing or unknown token (401).
46    #[error("unauthorized")]
47    Unauthorized,
48    /// Token belongs to another instance (403).
49    #[error("forbidden")]
50    Forbidden,
51    /// Over the per-instance rate (429).
52    #[error("rate limited")]
53    RateLimited,
54    /// The store owner could not take the batch (502).
55    #[error("store unavailable: {0}")]
56    Unavailable(String),
57    /// No store owner is known for the placement (502).
58    #[error("not routed: {0}")]
59    NotRouted(String),
60}
61
62impl IngestError {
63    /// HTTP status code.
64    pub fn status(&self) -> u16 {
65        match self {
66            Self::Malformed(_) => 400,
67            Self::Unauthorized => 401,
68            Self::Forbidden => 403,
69            Self::TooLarge(_) => 413,
70            Self::RateLimited => 429,
71            Self::Unavailable(_) | Self::NotRouted(_) => 502,
72        }
73    }
74}
75
76fn bad(m: impl Into<String>) -> IngestError {
77    IngestError::Malformed(m.into())
78}
79
80/// Parse and validate a request body.
81pub fn parse_batch(body: &[u8]) -> Result<IngestBatch, IngestError> {
82    if body.len() > MAX_BODY_BYTES {
83        return Err(IngestError::TooLarge(format!(
84            "body over {MAX_BODY_BYTES} bytes"
85        )));
86    }
87    let v: Value = serde_json::from_slice(body).map_err(|e| bad(format!("not JSON: {e}")))?;
88    let obj = v.as_object().ok_or_else(|| bad("body must be an object"))?;
89    if let Some(k) = obj.keys().find(|k| *k != "vectors" && *k != "dedup") {
90        return Err(bad(format!("unknown key `{k}`")));
91    }
92    let dedup = match obj.get("dedup") {
93        None => false,
94        Some(Value::Bool(b)) => *b,
95        Some(_) => return Err(bad("`dedup` must be a boolean")),
96    };
97    let arr = obj
98        .get("vectors")
99        .and_then(Value::as_array)
100        .ok_or_else(|| bad("`vectors` must be an array"))?;
101    if arr.is_empty() {
102        return Err(bad("`vectors` is empty"));
103    }
104    if arr.len() > MAX_VECTORS_PER_BATCH {
105        return Err(IngestError::TooLarge(format!(
106            "{} vectors, at most {MAX_VECTORS_PER_BATCH}",
107            arr.len()
108        )));
109    }
110    let vectors = arr
111        .iter()
112        .enumerate()
113        .map(|(i, e)| parse_vector(e).map_err(|m| bad(format!("vectors[{i}]: {m}"))))
114        .collect::<Result<Vec<_>, _>>()?;
115    Ok(IngestBatch { vectors, dedup })
116}
117
118fn parse_vector(e: &Value) -> Result<IngestVector, String> {
119    let pair = e.as_array().filter(|p| p.len() == 2);
120    let [id, vals] = pair.map(|p| [&p[0], &p[1]]).ok_or("expected [id, [values]]")?;
121    let id = id.as_u64().ok_or("id must be a non-negative integer")?;
122    let vals = vals.as_array().ok_or("values must be an array")?;
123    if vals.len() != DIMS {
124        return Err(format!("{} values, expected {DIMS}", vals.len()));
125    }
126    let mut values = [0f32; DIMS];
127    for (slot, x) in values.iter_mut().zip(vals) {
128        let f = x.as_f64().ok_or("values must be numbers")? as f32;
129        if !f.is_finite() {
130            return Err("values must be finite f32".into());
131        }
132        *slot = f;
133    }
134    Ok(IngestVector { id, values })
135}
136
137/// True for a project id: a canonical ULID, the same check every other
138/// project id passes ([`clawft_types::project::validate_id`]).
139pub fn valid_project_id(s: &str) -> bool {
140    clawft_types::project::validate_id(s).is_ok()
141}
clawft_kernel/cog_ingest/types.rs

clawft_kernel/cog_ingest/
types.rs
