helix_db/
lifecycle.rs

1//! Index lifecycle receipt and operation-status response contracts.
2//!
3//! ```
4//! use helix_db::{IndexDdlReceipt, IndexOperationStatus};
5//!
6//! let receipt: IndexDdlReceipt = sonic_rs::from_str(
7//!     r#"{"kind":"accepted","operation_id":"07070707-0707-0707-0707-070707070707","index_id":"42","generation":"3"}"#,
8//! ).unwrap();
9//! assert!(matches!(receipt, IndexDdlReceipt::Accepted { index_id: 42, .. }));
10//!
11//! let status: IndexOperationStatus = sonic_rs::from_str(
12//!     r#"{"status":"queued","operation_id":"07070707-0707-0707-0707-070707070707","index_id":"42","generation":"3","operation_kind":"build","family":"secondary","stage":"scan","attempt":0,"progress":{"entities":"0","input_bytes":"0","output_operations":"0","output_bytes":"0"},"future":true}"#,
13//! ).unwrap();
14//! assert!(matches!(status, IndexOperationStatus::Queued { .. }));
15//! ```
16
17use serde::{Deserialize, Deserializer, Serialize, Serializer};
18use uuid::Uuid;
19
20/// Result returned by CREATE or DROP.
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22#[serde(tag = "kind", rename_all = "snake_case")]
23pub enum IndexDdlReceipt {
24    /// A new durable operation was accepted.
25    Accepted {
26        /// Stable operation UUID.
27        #[serde(with = "uuid_string")]
28        operation_id: Uuid,
29        /// Logical index ID.
30        #[serde(with = "positive_u64_string")]
31        index_id: u64,
32        /// Physical generation.
33        #[serde(with = "positive_u64_string")]
34        generation: u64,
35    },
36    /// The request converged on existing work.
37    ExistingOperation {
38        /// Stable operation UUID.
39        #[serde(with = "uuid_string")]
40        operation_id: Uuid,
41    },
42    /// An identical index is already active.
43    AlreadyActive {
44        /// Logical index ID.
45        #[serde(with = "positive_u64_string")]
46        index_id: u64,
47        /// Active physical generation.
48        #[serde(with = "positive_u64_string")]
49        generation: u64,
50    },
51}
52
53/// BUILD or DROP.
54#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
55#[serde(rename_all = "snake_case")]
56pub enum IndexOperationKind {
57    /// Build and activate.
58    Build,
59    /// Drain and remove.
60    Drop,
61}
62
63/// Physical index family.
64#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
65#[serde(rename_all = "snake_case")]
66pub enum IndexFamily {
67    /// Equality/range secondary indexes.
68    Secondary,
69    /// Vector indexes.
70    Vector,
71    /// Text indexes.
72    Text,
73}
74
75/// Frozen lifecycle stage returned by operation status.
76#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
77#[serde(rename_all = "snake_case")]
78pub enum IndexOperationStage {
79    /// Scan authoritative secondary/vector entities.
80    Scan,
81    /// Scan authoritative text partitions.
82    ScanPartitions,
83    /// Apply mutation deltas captured during the scan.
84    CatchUp,
85    /// Validate secondary ownership and uniqueness.
86    Validate,
87    /// Validate vector metadata against its canonical descriptor.
88    ValidateDescriptor,
89    /// Validate an unchanged legacy vector namespace.
90    ValidateLegacyPhysical,
91    /// Compact text split state.
92    Compact,
93    /// Construct bounded text manifest pages.
94    PrepareManifests,
95    /// Validate manifest topology and remaining build ownership.
96    ValidateManifests,
97    /// Publish the hidden generation.
98    Activate,
99    /// Delete secondary entries.
100    DeleteEntries,
101    /// Retire vector memory.
102    RetireCache,
103    /// Delete vector physical rows.
104    DeletePhysical,
105    /// Delete retained mutation deltas.
106    DeleteDeltas,
107    /// Delete text generation metadata while retaining immutable blobs.
108    DeleteMetadata,
109    /// Finalize ordinary DROP cleanup.
110    Finalize,
111    /// Delete secondary entries during BUILD abort.
112    AbortingDeleteEntries,
113    /// Retire vector memory during BUILD abort.
114    AbortingRetireCache,
115    /// Delete vector physical rows during BUILD abort.
116    AbortingDeletePhysical,
117    /// Delete mutation deltas during BUILD abort.
118    AbortingDeleteDeltas,
119    /// Delete text metadata during BUILD abort.
120    AbortingDeleteMetadata,
121    /// Finalize BUILD abort cleanup.
122    AbortingFinalize,
123}
124
125impl IndexOperationStage {
126    /// Returns whether this stage belongs to BUILD abort cleanup.
127    pub const fn is_aborting(self) -> bool {
128        matches!(
129            self,
130            Self::AbortingDeleteEntries
131                | Self::AbortingRetireCache
132                | Self::AbortingDeletePhysical
133                | Self::AbortingDeleteDeltas
134                | Self::AbortingDeleteMetadata
135                | Self::AbortingFinalize
136        )
137    }
138}
139
140/// Stable blocked-operation reason.
141#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
142#[serde(rename_all = "snake_case")]
143pub enum IndexOperationBlockerCode {
144    /// Authoritative source data cannot satisfy the index contract.
145    InvalidSourceData,
146    /// A unique index found more than one owner for a value.
147    UniquenessViolation,
148    /// One source entity exceeds the configured bounded-step limit.
149    OversizedEntity,
150    /// A text manifest exceeds its configured bound.
151    ManifestLimit,
152    /// Text object-store configuration is unavailable.
153    ObjectStoreConfigurationUnavailable,
154    /// The runtime could not prove an internal lifecycle invariant.
155    InvariantViolation,
156}
157
158/// Stable index API error identifiers.
159#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
160#[serde(rename_all = "snake_case")]
161pub enum IndexErrorCode {
162    /// The family has not reached its safe public capability state.
163    IndexLifecycleUnavailable,
164    /// The logical index already exists.
165    IndexAlreadyExists,
166    /// The requested definition conflicts with the canonical definition.
167    IndexDefinitionConflict,
168    /// The logical index is already changing lifecycle state.
169    IndexBusy,
170    /// The logical index does not exist in the request scope.
171    IndexNotFound,
172    /// The retained operation does not exist in the request scope.
173    IndexOperationNotFound,
174    /// The retained operation cannot legally be aborted.
175    IndexOperationNotAbortable,
176    /// The logical index ID namespace is exhausted.
177    IndexIdExhausted,
178    /// The vector physical ID namespace is exhausted.
179    VectorPhysicalIdExhausted,
180    /// The physical generation namespace is exhausted.
181    IndexGenerationExhausted,
182    /// The canonical index-record revision is exhausted.
183    IndexRevisionExhausted,
184    /// The operation-record revision is exhausted.
185    IndexOperationRevisionExhausted,
186    /// A retained active handle no longer names the canonical generation.
187    StaleIndexGeneration,
188    /// Writer fencing prevented the runtime from proving commit outcome.
189    WriterFencedCommitOutcomeUnknown,
190}
191
192/// Monotonic public progress counters.
193#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
194pub struct IndexOperationProgress {
195    /// Authoritative source entities visited.
196    #[serde(with = "u64_string")]
197    pub entities: u64,
198    /// Source bytes consumed.
199    #[serde(with = "u64_string")]
200    pub input_bytes: u64,
201    /// Physical operations staged.
202    #[serde(with = "u64_string")]
203    pub output_operations: u64,
204    /// Physical output bytes staged.
205    #[serde(with = "u64_string")]
206    pub output_bytes: u64,
207}
208
209/// Fields common to every operation status.
210#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
211pub struct IndexOperationStatusCommon {
212    /// Stable operation UUID.
213    #[serde(with = "uuid_string")]
214    pub operation_id: Uuid,
215    /// Logical index ID.
216    #[serde(with = "positive_u64_string")]
217    pub index_id: u64,
218    /// Physical generation affected by this operation.
219    #[serde(with = "positive_u64_string")]
220    pub generation: u64,
221    /// BUILD or DROP.
222    pub operation_kind: IndexOperationKind,
223    /// Secondary, vector, or text physical lane.
224    pub family: IndexFamily,
225    /// Frozen family stage name.
226    pub stage: IndexOperationStage,
227    /// Number of durable claim attempts.
228    pub attempt: u32,
229    /// Monotonic bounded-work counters.
230    pub progress: IndexOperationProgress,
231}
232
233/// Status returned by get/retry/abort. Unknown additive fields are ignored.
234#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
235#[serde(tag = "status", rename_all = "snake_case")]
236pub enum IndexOperationStatus {
237    /// Runnable, including work waiting for a bounded retry deadline.
238    Queued {
239        /// Fields shared by every status.
240        #[serde(flatten)]
241        common: IndexOperationStatusCommon,
242    },
243    /// Claimed by a fenced writer.
244    Running {
245        /// Fields shared by every status.
246        #[serde(flatten)]
247        common: IndexOperationStatusCommon,
248    },
249    /// Paused until an explicit retry or abort.
250    Blocked {
251        /// Fields shared by every status.
252        #[serde(flatten)]
253        common: IndexOperationStatusCommon,
254        /// Stable machine-readable blocker.
255        blocker_code: IndexOperationBlockerCode,
256        /// Optional non-contractual diagnostic.
257        #[serde(default)]
258        message: Option<String>,
259    },
260    /// Build or drop completed successfully.
261    Succeeded {
262        /// Fields shared by every status.
263        #[serde(flatten)]
264        common: IndexOperationStatusCommon,
265    },
266    /// Build cleanup completed after an explicit abort.
267    Aborted {
268        /// Fields shared by every status.
269        #[serde(flatten)]
270        common: IndexOperationStatusCommon,
271    },
272}
273
274#[derive(Deserialize)]
275#[serde(tag = "status", rename_all = "snake_case")]
276enum IndexOperationStatusWire {
277    Queued {
278        #[serde(flatten)]
279        common: IndexOperationStatusCommon,
280    },
281    Running {
282        #[serde(flatten)]
283        common: IndexOperationStatusCommon,
284    },
285    Blocked {
286        #[serde(flatten)]
287        common: IndexOperationStatusCommon,
288        blocker_code: IndexOperationBlockerCode,
289        #[serde(default)]
290        message: Option<String>,
291    },
292    Succeeded {
293        #[serde(flatten)]
294        common: IndexOperationStatusCommon,
295    },
296    Aborted {
297        #[serde(flatten)]
298        common: IndexOperationStatusCommon,
299    },
300}
301
302impl<'de> Deserialize<'de> for IndexOperationStatus {
303    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
304    where
305        D: Deserializer<'de>,
306    {
307        let status = IndexOperationStatusWire::deserialize(deserializer)?;
308        Ok(match status {
309            IndexOperationStatusWire::Queued { common } => Self::Queued { common },
310            IndexOperationStatusWire::Running { common } => Self::Running { common },
311            IndexOperationStatusWire::Blocked {
312                common,
313                blocker_code,
314                message,
315            } => Self::Blocked {
316                common,
317                blocker_code,
318                message,
319            },
320            IndexOperationStatusWire::Succeeded { common } => Self::Succeeded { common },
321            IndexOperationStatusWire::Aborted { common } => {
322                if common.operation_kind != IndexOperationKind::Build || !common.stage.is_aborting()
323                {
324                    return Err(serde::de::Error::custom(
325                        "aborted status must describe build cleanup",
326                    ));
327                }
328                Self::Aborted { common }
329            }
330        })
331    }
332}
333
334mod u64_string {
335    use super::*;
336
337    pub(super) fn serialize<S>(value: &u64, serializer: S) -> Result<S::Ok, S::Error>
338    where
339        S: Serializer,
340    {
341        serializer.serialize_str(&value.to_string())
342    }
343
344    pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<u64, D::Error>
345    where
346        D: Deserializer<'de>,
347    {
348        let value = String::deserialize(deserializer)?;
349        let parsed = value.parse::<u64>().map_err(serde::de::Error::custom)?;
350        if parsed.to_string() != value {
351            return Err(serde::de::Error::custom(
352                "expected a canonical unsigned decimal string",
353            ));
354        }
355        Ok(parsed)
356    }
357}
358
359mod positive_u64_string {
360    use super::*;
361
362    pub(super) fn serialize<S>(value: &u64, serializer: S) -> Result<S::Ok, S::Error>
363    where
364        S: Serializer,
365    {
366        serializer.serialize_str(&value.to_string())
367    }
368
369    pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<u64, D::Error>
370    where
371        D: Deserializer<'de>,
372    {
373        let value = u64_string::deserialize(deserializer)?;
374        if value == 0 {
375            return Err(serde::de::Error::custom("identifier must be non-zero"));
376        }
377        Ok(value)
378    }
379}
380
381mod uuid_string {
382    use super::*;
383
384    pub(super) fn serialize<S>(value: &Uuid, serializer: S) -> Result<S::Ok, S::Error>
385    where
386        S: Serializer,
387    {
388        serializer.serialize_str(&value.to_string())
389    }
390
391    pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<Uuid, D::Error>
392    where
393        D: Deserializer<'de>,
394    {
395        let value = String::deserialize(deserializer)?;
396        let parsed = Uuid::parse_str(&value).map_err(serde::de::Error::custom)?;
397        if parsed.is_nil() || parsed.to_string() != value {
398            return Err(serde::de::Error::custom(
399                "operation ID must be a canonical lowercase non-nil UUID",
400            ));
401        }
402        Ok(parsed)
403    }
404}
405
406#[cfg(test)]
407mod tests {
408    use super::*;
409
410    #[test]
411    fn response_decoders_accept_additive_fields_and_reject_invalid_required_fields() {
412        let receipt: IndexDdlReceipt = sonic_rs::from_str(
413            r#"{"kind":"accepted","operation_id":"018f0c58-6bc7-7c56-8d3d-9c5f18a0f001","index_id":"42","generation":"3","future":true}"#,
414        )
415        .unwrap();
416        assert!(matches!(
417            receipt,
418            IndexDdlReceipt::Accepted { index_id: 42, .. }
419        ));
420
421        let status: IndexOperationStatus = sonic_rs::from_str(
422            r#"{"status":"blocked","operation_id":"018f0c58-6bc7-7c56-8d3d-9c5f18a0f001","index_id":"42","generation":"3","operation_kind":"build","family":"secondary","stage":"scan","attempt":2,"progress":{"entities":"9","input_bytes":"10","output_operations":"11","output_bytes":"12","future":true},"blocker_code":"uniqueness_violation","future":true}"#,
423        )
424        .unwrap();
425        assert!(matches!(status, IndexOperationStatus::Blocked { .. }));
426        for (stage, expected) in [
427            (
428                "validate_legacy_physical",
429                IndexOperationStage::ValidateLegacyPhysical,
430            ),
431            ("validate_manifests", IndexOperationStage::ValidateManifests),
432        ] {
433            let status: IndexOperationStatus = sonic_rs::from_str(&format!(
434                r#"{{"status":"queued","operation_id":"018f0c58-6bc7-7c56-8d3d-9c5f18a0f001","index_id":"42","generation":"3","operation_kind":"build","family":"text","stage":"{stage}","attempt":0,"progress":{{"entities":"0","input_bytes":"0","output_operations":"0","output_bytes":"0"}}}}"#,
435            ))
436            .unwrap();
437            let IndexOperationStatus::Queued { common } = status else {
438                panic!("valid text build stage must decode as queued");
439            };
440            assert_eq!(common.stage, expected);
441        }
442        let aborted: IndexOperationStatus = sonic_rs::from_str(
443            r#"{"status":"aborted","operation_id":"018f0c58-6bc7-7c56-8d3d-9c5f18a0f001","index_id":"42","generation":"3","operation_kind":"build","family":"secondary","stage":"aborting_finalize","attempt":2,"progress":{"entities":"9","input_bytes":"10","output_operations":"11","output_bytes":"12"}}"#,
444        )
445        .unwrap();
446        assert!(matches!(aborted, IndexOperationStatus::Aborted { .. }));
447
448        assert!(sonic_rs::from_str::<IndexDdlReceipt>(
449            r#"{"kind":"accepted","operation_id":"018F0C58-6BC7-7C56-8D3D-9C5F18A0F001","index_id":"42","generation":"3"}"#,
450        )
451        .is_err());
452        assert!(sonic_rs::from_str::<IndexDdlReceipt>(
453            r#"{"kind":"already_active","index_id":"0","generation":"03"}"#,
454        )
455        .is_err());
456        assert!(sonic_rs::from_str::<IndexOperationStatus>(r#"{"status":"future"}"#).is_err());
457        assert!(sonic_rs::from_str::<IndexOperationStatus>(
458            r#"{"status":"queued","operation_id":"018f0c58-6bc7-7c56-8d3d-9c5f18a0f001","index_id":"42","generation":"3","operation_kind":"build","family":"secondary","stage":"future","attempt":0,"progress":{"entities":"0","input_bytes":"0","output_operations":"0","output_bytes":"0"}}"#,
459        )
460        .is_err());
461        assert!(sonic_rs::from_str::<IndexOperationStatus>(
462            r#"{"status":"aborted","operation_id":"018f0c58-6bc7-7c56-8d3d-9c5f18a0f001","index_id":"42","generation":"3","operation_kind":"drop","family":"secondary","stage":"finalize","attempt":0,"progress":{"entities":"0","input_bytes":"0","output_operations":"0","output_bytes":"0"}}"#,
463        )
464        .is_err());
465    }
466}
helix_db/lifecycle.rs

helix_db/
lifecycle.rs
