lodestone_wire/
bulk_batch_id.rs

1use std::collections::BTreeMap;
2
3use crate::{BULK_BATCH_ID_DOMAIN, BULK_BATCH_ID_FORMAT_VERSION, Digest, Identity, Resource};
4const MAXIMUM_ITEM_ACTION_IDS: usize = 1_024;
5
6#[derive(Clone, Debug, Eq, PartialEq)]
7pub enum BulkBatchIdError {
8    InvalidCollectionIdentity,
9    ZeroCollectionIdentity,
10    InvalidIndexKind,
11    InvalidIndexName,
12    InvalidIndexIncarnation,
13    ZeroIndexIncarnation,
14    TooManyItemActionIds,
15    InvalidItemActionId { position: usize },
16    ZeroItemActionId { position: usize },
17    DuplicateItemActionId { first: usize, duplicate: usize },
18}
19
20impl std::fmt::Display for BulkBatchIdError {
21    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
22        write!(formatter, "invalid Lodestone API bulk batch identity input: {self:?}")
23    }
24}
25
26impl std::error::Error for BulkBatchIdError {}
27
28/// Derives the sole current Lodestone API batch identity used by document bulk
29/// and retained-append bulk requests.
30///
31/// The BLAKE3-256 input is the versioned domain followed by this canonical
32/// binary payload, in order: format version as `u16` big-endian, collection as raw
33/// `u128` big-endian, index-name byte length as `u16` big-endian, exact UTF-8
34/// index-name bytes, index incarnation as raw `u128` big-endian, item count as
35/// `u32` big-endian, then each ordered action identity as 32 raw bytes.
36pub fn bulk_batch_id(
37    collection: &Identity,
38    index: &Resource,
39    item_action_ids: &[Digest],
40) -> Result<Digest, BulkBatchIdError> {
41    let payload = canonical_bulk_batch_payload_v1(collection, index, item_action_ids)?;
42    let mut hasher = blake3::Hasher::new();
43    hasher.update(BULK_BATCH_ID_DOMAIN);
44    hasher.update(&payload);
45    Ok(Digest {
46        value: hasher.finalize().to_hex().to_string(),
47    })
48}
49
50fn canonical_bulk_batch_payload_v1(
51    collection: &Identity,
52    index: &Resource,
53    item_action_ids: &[Digest],
54) -> Result<Vec<u8>, BulkBatchIdError> {
55    let collection = decode_identity(&collection.value).ok_or(BulkBatchIdError::InvalidCollectionIdentity)?;
56    if collection == 0 {
57        return Err(BulkBatchIdError::ZeroCollectionIdentity);
58    }
59    if index.kind != "index" {
60        return Err(BulkBatchIdError::InvalidIndexKind);
61    }
62    let index_name = index.name.as_bytes();
63    if !(1..=255).contains(&index_name.len()) || index_name.iter().any(|byte| *byte <= 0x1f || *byte == 0x7f) {
64        return Err(BulkBatchIdError::InvalidIndexName);
65    }
66    let incarnation = decode_identity(&index.incarnation.value).ok_or(BulkBatchIdError::InvalidIndexIncarnation)?;
67    if incarnation == 0 {
68        return Err(BulkBatchIdError::ZeroIndexIncarnation);
69    }
70    if item_action_ids.len() > MAXIMUM_ITEM_ACTION_IDS {
71        return Err(BulkBatchIdError::TooManyItemActionIds);
72    }
73
74    let name_length = u16::try_from(index_name.len()).map_err(|_| BulkBatchIdError::InvalidIndexName)?;
75    let item_count = u32::try_from(item_action_ids.len()).map_err(|_| BulkBatchIdError::TooManyItemActionIds)?;
76    let mut payload = Vec::with_capacity(40 + index_name.len() + item_action_ids.len() * 32);
77    payload.extend_from_slice(&BULK_BATCH_ID_FORMAT_VERSION.to_be_bytes());
78    payload.extend_from_slice(&collection.to_be_bytes());
79    payload.extend_from_slice(&name_length.to_be_bytes());
80    payload.extend_from_slice(index_name);
81    payload.extend_from_slice(&incarnation.to_be_bytes());
82    payload.extend_from_slice(&item_count.to_be_bytes());
83
84    let mut positions = BTreeMap::new();
85    for (position, action) in item_action_ids.iter().enumerate() {
86        let bytes = decode_digest(&action.value).ok_or(BulkBatchIdError::InvalidItemActionId { position })?;
87        if bytes == [0; 32] {
88            return Err(BulkBatchIdError::ZeroItemActionId { position });
89        }
90        if let Some(first) = positions.insert(bytes, position) {
91            return Err(BulkBatchIdError::DuplicateItemActionId {
92                first,
93                duplicate: position,
94            });
95        }
96        payload.extend_from_slice(&bytes);
97    }
98    Ok(payload)
99}
100
101fn decode_identity(value: &str) -> Option<u128> {
102    if value.len() != 32 || !value.bytes().all(is_lower_hex) {
103        return None;
104    }
105    u128::from_str_radix(value, 16).ok()
106}
107
108fn decode_digest(value: &str) -> Option<[u8; 32]> {
109    if value.len() != 64 || !value.bytes().all(is_lower_hex) {
110        return None;
111    }
112    let mut decoded = [0_u8; 32];
113    for (output, pair) in decoded.iter_mut().zip(value.as_bytes().chunks_exact(2)) {
114        *output = (lower_hex_value(pair[0])? << 4) | lower_hex_value(pair[1])?;
115    }
116    Some(decoded)
117}
118
119const fn is_lower_hex(value: u8) -> bool {
120    matches!(value, b'0'..=b'9' | b'a'..=b'f')
121}
122
123const fn lower_hex_value(value: u8) -> Option<u8> {
124    match value {
125        b'0'..=b'9' => Some(value - b'0'),
126        b'a'..=b'f' => Some(value - b'a' + 10),
127        _ => None,
128    }
129}
130
131#[cfg(test)]
132mod tests {
133    use super::*;
134
135    const VECTORS: &str = include_str!("../../../clients/native/fixtures/lodestone-bulk-batch-id.json");
136
137    fn identity(value: u128) -> Identity {
138        Identity {
139            value: format!("{value:032x}"),
140        }
141    }
142
143    fn digest(byte: u8) -> Digest {
144        Digest {
145            value: format!("{byte:02x}").repeat(32),
146        }
147    }
148
149    fn index(name: &str, incarnation: u128) -> Resource {
150        Resource {
151            kind: "index".to_owned(),
152            name: name.to_owned(),
153            incarnation: identity(incarnation),
154        }
155    }
156
157    #[test]
158    fn bulk_batch_identity_is_deterministic_ordered_and_scope_bound() -> Result<(), Box<dyn std::error::Error>> {
159        let collection = identity(1);
160        let resource = index("audit-events", 2);
161        let actions = [digest(0x11), digest(0x22)];
162        let first = bulk_batch_id(&collection, &resource, &actions)?;
163        assert_eq!(
164            first.value,
165            "25b1694f4d2883cc7c475d51ef6e9fc6ce1d42ef09a9283db30a4acd7ebd0f06"
166        );
167        let escaped = bulk_batch_id(
168            &Identity {
169                value: "0123456789abcdef0123456789abcdef".to_owned(),
170            },
171            &Resource {
172                kind: "index".to_owned(),
173                name: "quote\"\\雪".to_owned(),
174                incarnation: Identity {
175                    value: "fedcba9876543210fedcba9876543210".to_owned(),
176                },
177            },
178            &[
179                Digest {
180                    value: format!("{}01", "00".repeat(31)),
181                },
182                digest(0xff),
183            ],
184        )?;
185        assert_eq!(
186            escaped.value,
187            "619b49a49aa5a91437679522dbc272e3344118eea59f6ccd1fda7655b78d5f29"
188        );
189        assert_eq!(bulk_batch_id(&collection, &resource, &actions)?, first);
190        assert_ne!(
191            bulk_batch_id(&collection, &resource, &[actions[1].clone(), actions[0].clone()])?,
192            first
193        );
194        assert_ne!(bulk_batch_id(&identity(3), &resource, &actions)?, first);
195        assert_ne!(
196            bulk_batch_id(&collection, &index("audit-events-2", 2), &actions)?,
197            first
198        );
199        assert_ne!(bulk_batch_id(&collection, &index("audit-events", 3), &actions)?, first);
200        Ok(())
201    }
202
203    #[test]
204    fn bulk_batch_identity_rejects_duplicate_malformed_and_zero_actions() {
205        let collection = identity(1);
206        let resource = index("audit-events", 2);
207        assert_eq!(
208            bulk_batch_id(&collection, &resource, &[digest(0x11), digest(0x11)]),
209            Err(BulkBatchIdError::DuplicateItemActionId { first: 0, duplicate: 1 })
210        );
211        assert_eq!(
212            bulk_batch_id(&collection, &resource, &[Digest { value: "GG".repeat(32) }]),
213            Err(BulkBatchIdError::InvalidItemActionId { position: 0 })
214        );
215        assert_eq!(
216            bulk_batch_id(&collection, &resource, &[digest(0)]),
217            Err(BulkBatchIdError::ZeroItemActionId { position: 0 })
218        );
219    }
220
221    #[test]
222    fn bulk_batch_identity_rejects_invalid_scope_and_over_bound_inputs() {
223        let resource = index("audit-events", 2);
224        assert_eq!(
225            bulk_batch_id(&Identity { value: "01".to_owned() }, &resource, &[digest(0x11)],),
226            Err(BulkBatchIdError::InvalidCollectionIdentity)
227        );
228        assert_eq!(
229            bulk_batch_id(&identity(0), &resource, &[digest(0x11)]),
230            Err(BulkBatchIdError::ZeroCollectionIdentity)
231        );
232        assert_eq!(
233            bulk_batch_id(
234                &identity(1),
235                &Resource {
236                    kind: "alias".to_owned(),
237                    name: "audit-events".to_owned(),
238                    incarnation: identity(2),
239                },
240                &[digest(0x11)],
241            ),
242            Err(BulkBatchIdError::InvalidIndexKind)
243        );
244        assert_eq!(
245            bulk_batch_id(&identity(1), &index("", 2), &[digest(0x11)]),
246            Err(BulkBatchIdError::InvalidIndexName)
247        );
248        assert_eq!(
249            bulk_batch_id(&identity(1), &index("audit-events", 0), &[digest(0x11)]),
250            Err(BulkBatchIdError::ZeroIndexIncarnation)
251        );
252        assert_eq!(
253            bulk_batch_id(&identity(1), &resource, &vec![digest(0x11); 1_025]),
254            Err(BulkBatchIdError::TooManyItemActionIds)
255        );
256    }
257
258    #[test]
259    fn fixed_language_neutral_vectors_match_payload_and_digest() -> Result<(), Box<dyn std::error::Error>> {
260        let document: serde_json::Value = serde_json::from_str(VECTORS)?;
261        assert_eq!(document["format_version"].as_u64(), Some(1));
262        assert_eq!(
263            document["batch_id_format_version"].as_u64(),
264            Some(u64::from(BULK_BATCH_ID_FORMAT_VERSION))
265        );
266        assert_eq!(document["algorithm"].as_str(), Some("blake3-256"));
267        assert_eq!(
268            document["domain_separator_utf8"].as_str().map(str::as_bytes),
269            Some(BULK_BATCH_ID_DOMAIN)
270        );
271        let vectors = document["vectors"].as_array().ok_or("bulk batch vectors are absent")?;
272        for vector in vectors {
273            let collection = Identity {
274                value: vector["collection_identity"]
275                    .as_str()
276                    .ok_or("bulk batch vector collection is absent")?
277                    .to_owned(),
278            };
279            let index: Resource = serde_json::from_value(vector["index"].clone())?;
280            let action_values = vector["item_action_ids"]
281                .as_array()
282                .ok_or("bulk batch vector actions are absent")?;
283            let actions = action_values
284                .iter()
285                .map(|value| {
286                    value
287                        .as_str()
288                        .map(|value| Digest {
289                            value: value.to_owned(),
290                        })
291                        .ok_or("bulk batch vector action is invalid")
292                })
293                .collect::<Result<Vec<_>, _>>()?;
294            let expected_payload = decode_test_hex(
295                vector["canonical_payload_hex"]
296                    .as_str()
297                    .ok_or("bulk batch vector payload is absent")?,
298            )?;
299            assert_eq!(
300                canonical_bulk_batch_payload_v1(&collection, &index, &actions)?,
301                expected_payload
302            );
303            assert_eq!(
304                bulk_batch_id(&collection, &index, &actions)?.value,
305                vector["batch_id"]
306                    .as_str()
307                    .ok_or("bulk batch vector result is absent")?
308            );
309        }
310        Ok(())
311    }
312
313    fn decode_test_hex(value: &str) -> Result<Vec<u8>, Box<dyn std::error::Error>> {
314        if !value.len().is_multiple_of(2) {
315            return Err("bulk batch vector payload has odd hex width".into());
316        }
317        value
318            .as_bytes()
319            .chunks_exact(2)
320            .map(|pair| Ok(u8::from_str_radix(std::str::from_utf8(pair)?, 16)?))
321            .collect()
322    }
323}
lodestone_wire/bulk_batch_id.rs

lodestone_wire/
bulk_batch_id.rs
