reddb_server/runtime/
batch_insert.rs

1//! Issue #582 — Analytics slice 4: `BatchInsertEndpoint` HTTP support.
2//!
3//! Owns the config + durable idempotency receipts that the
4//! `POST /collections/:name/batch` handler uses to:
5//!
6//! * reject batches exceeding `red.batch.max_rows` with `413
7//!   BatchTooLarge` before any storage work,
8//! * deduplicate requests carrying the same `Idempotency-Key` within
9//!   `red.batch.idempotency_window_secs`, returning the previously
10//!   committed response body verbatim across process failover,
11//! * report row-level validation failures by index so the caller can
12//!   pinpoint the row that broke the all-or-nothing commit.
13//!
14//! The actual commit + AnalyticsSchemaRegistry validation runs in the
15//! handler (`server::handlers_entity::handle_batch_insert`). This
16//! module only owns the pieces that need to be unit-testable without
17//! booting a full HTTP server.
18
19use std::sync::{Mutex, MutexGuard, OnceLock};
20use std::time::{Duration, SystemTime, UNIX_EPOCH};
21
22use reddb_types::Value;
23
24use crate::storage::unified::entity::{EntityData, EntityId, EntityKind, RowData, UnifiedEntity};
25use crate::storage::unified::{Metadata, MetadataValue};
26use crate::{RedDBError, RedDBResult, RedDBRuntime};
27
28const DEFAULT_MAX_ROWS: usize = 10_000;
29const DEFAULT_IDEMPOTENCY_WINDOW_SECS: u64 = 300;
30const ENV_MAX_ROWS: &str = "RED_BATCH_MAX_ROWS";
31const ENV_IDEMPOTENCY_WINDOW_SECS: &str = "RED_BATCH_IDEMPOTENCY_WINDOW_SECS";
32const RECEIPT_COLLECTION: &str = "red.batch.idempotency";
33
34/// Knobs for the batch endpoint. Read from `RED_BATCH_*` env vars at
35/// process start; the `red.batch.*` config keys in the brief route to
36/// the same defaults until the broader config-overlay binding lands.
37#[derive(Debug, Clone, Copy)]
38pub struct BatchInsertConfig {
39    pub max_rows: usize,
40    pub idempotency_window: Duration,
41}
42
43impl BatchInsertConfig {
44    pub fn from_env() -> Self {
45        let max_rows = std::env::var(ENV_MAX_ROWS)
46            .ok()
47            .and_then(|v| v.parse::<usize>().ok())
48            .filter(|n| *n > 0)
49            .unwrap_or(DEFAULT_MAX_ROWS);
50        let window_secs = std::env::var(ENV_IDEMPOTENCY_WINDOW_SECS)
51            .ok()
52            .and_then(|v| v.parse::<u64>().ok())
53            .unwrap_or(DEFAULT_IDEMPOTENCY_WINDOW_SECS);
54        Self {
55            max_rows,
56            idempotency_window: Duration::from_secs(window_secs),
57        }
58    }
59}
60
61/// A committed response for a `(collection, idempotency-key)` pair.
62#[derive(Debug, Clone)]
63pub struct DurableReceipt {
64    pub status: u16,
65    pub body: Vec<u8>,
66}
67
68pub fn unix_time_ms() -> u64 {
69    SystemTime::now()
70        .duration_since(UNIX_EPOCH)
71        .map(|duration| u64::try_from(duration.as_millis()).unwrap_or(u64::MAX))
72        .unwrap_or(0)
73}
74
75fn receipt_key(collection: &str, idempotency_key: &str) -> String {
76    let principal = crate::runtime::execution_context::current_auth_principal()
77        .map(|principal| principal.to_string())
78        .unwrap_or_default();
79    let tenant = crate::runtime::execution_context::current_tenant().unwrap_or_default();
80    let parts = [
81        principal.as_str(),
82        tenant.as_str(),
83        collection,
84        idempotency_key,
85    ];
86    let mut material = Vec::new();
87    for part in parts {
88        material.extend_from_slice(&(part.len() as u64).to_be_bytes());
89        material.extend_from_slice(part.as_bytes());
90    }
91    hex::encode(crate::crypto::sha256(&material))
92}
93
94pub fn lock_durable_commit(collection: &str, idempotency_key: &str) -> MutexGuard<'static, ()> {
95    const STRIPES: usize = 64;
96    static LOCKS: OnceLock<Vec<Mutex<()>>> = OnceLock::new();
97    let locks = LOCKS.get_or_init(|| (0..STRIPES).map(|_| Mutex::new(())).collect());
98    let digest = crate::crypto::sha256(receipt_key(collection, idempotency_key).as_bytes());
99    let stripe = usize::from(digest[0]) % STRIPES;
100    locks[stripe]
101        .lock()
102        .unwrap_or_else(|poisoned| poisoned.into_inner())
103}
104
105fn row_u64(row: &RowData, field: &str) -> Option<u64> {
106    match row.get_field(field)? {
107        Value::UnsignedInteger(value) => Some(*value),
108        Value::Integer(value) if *value >= 0 => Some(*value as u64),
109        _ => None,
110    }
111}
112
113fn row_text(row: &RowData, field: &str) -> Option<String> {
114    match row.get_field(field)? {
115        Value::Text(value) => Some(value.to_string()),
116        _ => None,
117    }
118}
119
120fn receipt_is_committed(runtime: &RedDBRuntime, entity: &UnifiedEntity) -> bool {
121    let snapshots = runtime.snapshot_manager();
122    let inserted = entity.xmin == 0
123        || (!snapshots.is_active(entity.xmin) && !snapshots.is_aborted(entity.xmin));
124    let deleted =
125        entity.xmax != 0 && !snapshots.is_active(entity.xmax) && !snapshots.is_aborted(entity.xmax);
126    inserted && !deleted
127}
128
129pub fn lookup_durable_receipt(
130    runtime: &RedDBRuntime,
131    collection: &str,
132    idempotency_key: &str,
133    now_ms: u64,
134) -> RedDBResult<Option<DurableReceipt>> {
135    let store = runtime.db().store();
136    let Some(manager) = store.get_collection(RECEIPT_COLLECTION) else {
137        return Ok(None);
138    };
139    let _publication = manager
140        .has_backing()
141        .then(|| store.vault_publication_lock.lock());
142    let key = receipt_key(collection, idempotency_key);
143    let mut latest = None;
144    manager
145        .try_for_each_key_entity(&key, |entity| {
146            if !receipt_is_committed(runtime, entity) {
147                return true;
148            }
149            let Some(row) = entity.data.as_row() else {
150                return true;
151            };
152            let Some(expires_at_ms) = row_u64(row, "expires_at_ms") else {
153                return true;
154            };
155            if expires_at_ms <= now_ms {
156                return true;
157            }
158            let Some(status) = row_u64(row, "status").and_then(|value| u16::try_from(value).ok())
159            else {
160                return true;
161            };
162            let Some(body) = row_text(row, "body") else {
163                return true;
164            };
165            if latest
166                .as_ref()
167                .is_none_or(|(latest_id, _): &(EntityId, DurableReceipt)| entity.id > *latest_id)
168            {
169                latest = Some((
170                    entity.id,
171                    DurableReceipt {
172                        status,
173                        body: body.into_bytes(),
174                    },
175                ));
176            }
177            true
178        })
179        .map_err(|error| RedDBError::Internal(error.to_string()))?;
180    Ok(latest.map(|(_, receipt)| receipt))
181}
182
183pub fn insert_durable_receipt(
184    runtime: &RedDBRuntime,
185    collection: &str,
186    idempotency_key: &str,
187    status: u16,
188    body: &[u8],
189    expires_at_ms: u64,
190) -> RedDBResult<()> {
191    let body = std::str::from_utf8(body)
192        .map_err(|_| RedDBError::Internal("batch response is not UTF-8".to_string()))?;
193    let store = runtime.db().store();
194    let key = receipt_key(collection, idempotency_key);
195    let fields = [
196        ("key".to_string(), Value::text(key)),
197        ("collection".to_string(), Value::text(collection)),
198        (
199            "status".to_string(),
200            Value::UnsignedInteger(u64::from(status)),
201        ),
202        ("body".to_string(), Value::text(body)),
203        (
204            "expires_at_ms".to_string(),
205            Value::UnsignedInteger(expires_at_ms),
206        ),
207    ]
208    .into_iter()
209    .collect();
210    let mut entity = UnifiedEntity::new(
211        EntityId::new(0),
212        EntityKind::TableRow {
213            table: RECEIPT_COLLECTION.into(),
214            row_id: 0,
215        },
216        EntityData::Row(RowData {
217            columns: Vec::new(),
218            named: Some(fields),
219            schema: None,
220        }),
221    );
222    if let Some(xid) = runtime.current_xid() {
223        entity.set_xmin(xid);
224    }
225    let id = store
226        .insert_auto(RECEIPT_COLLECTION, entity)
227        .map_err(|error| RedDBError::Internal(error.to_string()))?;
228    store
229        .set_metadata(
230            RECEIPT_COLLECTION,
231            id,
232            Metadata::with_fields(
233                [(
234                    "_expires_at".to_string(),
235                    MetadataValue::Timestamp(expires_at_ms),
236                )]
237                .into_iter()
238                .collect(),
239            ),
240        )
241        .map_err(|error| RedDBError::Internal(error.to_string()))?;
242    Ok(())
243}
244
245#[derive(Debug, Clone, PartialEq)]
246pub enum BatchInsertError {
247    /// Body did not deserialize to a JSON array.
248    BodyNotJsonArray,
249    /// `rows.len() > config.max_rows`.
250    BatchTooLarge { limit: usize, got: usize },
251    /// Parse / shape failure for row `index`.
252    RowParseFailure { index: usize, reason: String },
253    /// AnalyticsSchemaRegistry rejected the row at `index`.
254    RowSchemaRejected { index: usize, reason: String },
255}
256
257impl BatchInsertError {
258    pub fn http_status(&self) -> u16 {
259        match self {
260            Self::BatchTooLarge { .. } => 413,
261            _ => 400,
262        }
263    }
264
265    pub fn code(&self) -> &'static str {
266        match self {
267            Self::BodyNotJsonArray => "BadRequest",
268            Self::BatchTooLarge { .. } => "BatchTooLarge",
269            Self::RowParseFailure { .. } => "RowParseFailure",
270            Self::RowSchemaRejected { .. } => "RowSchemaRejected",
271        }
272    }
273
274    pub fn row_index(&self) -> Option<usize> {
275        match self {
276            Self::RowParseFailure { index, .. } | Self::RowSchemaRejected { index, .. } => {
277                Some(*index)
278            }
279            _ => None,
280        }
281    }
282
283    pub fn message(&self) -> String {
284        match self {
285            Self::BodyNotJsonArray => "request body must be a JSON array of rows".to_string(),
286            Self::BatchTooLarge { limit, got } => {
287                format!("batch size {got} exceeds red.batch.max_rows={limit}")
288            }
289            Self::RowParseFailure { index, reason } => {
290                format!("row {index} failed validation: {reason}")
291            }
292            Self::RowSchemaRejected { index, reason } => {
293                format!("row {index} rejected by AnalyticsSchemaRegistry: {reason}")
294            }
295        }
296    }
297}
298
299#[cfg(test)]
300mod tests {
301    use super::*;
302
303    #[test]
304    fn default_config_matches_brief() {
305        let cfg = BatchInsertConfig {
306            max_rows: DEFAULT_MAX_ROWS,
307            idempotency_window: Duration::from_secs(DEFAULT_IDEMPOTENCY_WINDOW_SECS),
308        };
309        assert_eq!(cfg.max_rows, 10_000);
310        assert_eq!(cfg.idempotency_window, Duration::from_secs(300));
311    }
312
313    #[test]
314    fn batch_too_large_maps_to_413() {
315        let err = BatchInsertError::BatchTooLarge { limit: 10, got: 11 };
316        assert_eq!(err.http_status(), 413);
317        assert_eq!(err.code(), "BatchTooLarge");
318        assert!(err.row_index().is_none());
319        assert!(err.message().contains("10"));
320        assert!(err.message().contains("11"));
321    }
322
323    #[test]
324    fn row_schema_rejected_carries_index() {
325        let err = BatchInsertError::RowSchemaRejected {
326            index: 3,
327            reason: "AnalyticsSchemaError:TypeMismatch:click:v1:user_id".to_string(),
328        };
329        assert_eq!(err.http_status(), 400);
330        assert_eq!(err.code(), "RowSchemaRejected");
331        assert_eq!(err.row_index(), Some(3));
332        assert!(err.message().contains("row 3"));
333        assert!(err.message().contains("TypeMismatch"));
334    }
335
336    #[test]
337    fn row_parse_failure_carries_index() {
338        let err = BatchInsertError::RowParseFailure {
339            index: 7,
340            reason: "missing fields".to_string(),
341        };
342        assert_eq!(err.http_status(), 400);
343        assert_eq!(err.row_index(), Some(7));
344        assert!(err.message().contains("row 7"));
345    }
346
347    #[test]
348    fn body_not_json_array_is_400() {
349        let err = BatchInsertError::BodyNotJsonArray;
350        assert_eq!(err.http_status(), 400);
351        assert!(err.row_index().is_none());
352    }
353}
reddb_server/runtime/batch_insert.rs

reddb_server/runtime/
batch_insert.rs
