oxirs_vec/real_time_embedding_pipeline/
mod.rs

1//! Real-time embedding pipeline module
2//!
3//! This module provides a comprehensive real-time embedding update system that can handle
4//! high-throughput streaming data with low-latency updates while maintaining consistency
5//! and performance guarantees.
6//!
7//! ## Architecture
8//!
9//! The real-time embedding pipeline is organized into several components:
10//!
11//! - **Configuration**: Pipeline configuration and settings
12//! - **Traits**: Core traits for extensibility
13//! - **Types**: Common data structures and types
14//! - **Pipeline**: Main pipeline implementation
15//! - **Streaming**: Stream processing components
16//! - **Coordination**: Update coordination and synchronization
17//! - **Monitoring**: Performance monitoring and alerting
18//! - **Versioning**: Version management and storage
19//! - **Consistency**: Consistency management and repair
20//!
21//! ## Usage
22//!
23//! ```rust,no_run
24//! # use tokio::runtime::Runtime;
25//! use oxirs_vec::real_time_embedding_pipeline::{
26//!     RealTimeEmbeddingPipeline, PipelineConfig, ConsistencyLevel
27//! };
28//!
29//! // Create pipeline configuration
30//! let config = PipelineConfig {
31//!     max_batch_size: 1000,
32//!     consistency_level: ConsistencyLevel::Session,
33//!     ..Default::default()
34//! };
35//!
36//! # let runtime = Runtime::new().expect("should succeed");
37//! # runtime.block_on(async {
38//! // Create and start pipeline
39//! let mut pipeline = RealTimeEmbeddingPipeline::new(config).expect("should succeed");
40//! pipeline.start().await.expect("should succeed");
41//! pipeline.stop().await.expect("should succeed");
42//! # });
43//! ```
44
45pub mod config;
46pub mod consistency;
47pub mod coordination;
48pub mod monitoring;
49pub mod pipeline;
50pub mod streaming;
51pub mod traits;
52pub mod types;
53pub mod versioning;
54
55// Re-export commonly used types
56pub use config::{
57    AlertThresholds, AutoScalingConfig, BackpressureStrategy, CompressionConfig, CompressionMethod,
58    ConsistencyLevel, MonitoringConfig, PipelineConfig, QualityAssuranceConfig, RetryConfig,
59    VersionControlConfig,
60};
61
62pub use traits::{
63    Alert, AlertCategory, AlertConfig, AlertHandler, AlertSeverity, AlertThrottling,
64    ConflictResolutionFunction, ConsistencyRepairStrategy, ContentItem, EmbeddingGenerator,
65    GeneratorStatistics, HealthStatus, Inconsistency, InconsistencyDetectionAlgorithm,
66    InconsistencySeverity, InconsistencyType, IncrementalVectorIndex, IndexStatistics, MetricPoint,
67    MetricsStorage, ProcessingPriority, ProcessingResult, ProcessingStatus, RepairResult,
68    RepairStatus, Transaction, TransactionLog, TransactionStatus, TransactionType, Version,
69    VersionStorage,
70};
71
72pub use types::{
73    BackpressureState, CircuitBreakerConfig, CircuitBreakerState, CoordinationState,
74    HealthCheckResult, NodeStatus, PerformanceMetrics, PipelineStatistics, QualityMetrics,
75    RealTimeConfig, ResourceUtilization, StreamState, StreamStatus, UpdateBatch, UpdateOperation,
76    UpdatePriority, UpdateStats, VersioningStrategy,
77};
78
79pub use consistency::{
80    ConsistencyConfig, ConsistencyManager, ConsistencyStatistics, DefaultRepairStrategy,
81    InconsistencyRepairEngine,
82};
83pub use coordination::{CoordinationConfig, CoordinationStatistics, UpdateCoordinator};
84pub use monitoring::{
85    AlertManager, ConsoleAlertHandler, InMemoryMetricsStorage, MetricsCollector,
86    PipelinePerformanceMonitor,
87};
88pub use pipeline::RealTimeEmbeddingPipeline;
89pub use streaming::{StreamProcessor, StreamProcessorConfig};
90pub use versioning::{
91    InMemoryVersionStorage, VersionManager, VersionManagerConfig, VersionManagerStatistics,
92};
93
94// Re-export error types
95pub use anyhow::{Context, Error, Result};
96
97// Pipeline-specific error types
98#[derive(Debug, thiserror::Error)]
99pub enum PipelineError {
100    #[error("Pipeline not initialized")]
101    NotInitialized,
102
103    #[error("Pipeline already running")]
104    AlreadyRunning,
105
106    #[error("Pipeline not running")]
107    NotRunning,
108
109    #[error("Configuration error: {message}")]
110    ConfigurationError { message: String },
111
112    #[error("Stream processing error: {message}")]
113    StreamProcessingError { message: String },
114
115    #[error("Coordination error: {message}")]
116    CoordinationError { message: String },
117
118    #[error("Consistency error: {message}")]
119    ConsistencyError { message: String },
120
121    #[error("Monitoring error: {message}")]
122    MonitoringError { message: String },
123
124    #[error("Version management error: {message}")]
125    VersionError { message: String },
126
127    #[error("Resource exhausted: {resource}")]
128    ResourceExhausted { resource: String },
129
130    #[error("Timeout error: {operation}")]
131    Timeout { operation: String },
132
133    #[error("Backpressure limit exceeded")]
134    BackpressureExceeded,
135
136    #[error("Quality check failed: {reason}")]
137    QualityCheckFailed { reason: String },
138
139    #[error("Circuit breaker open for: {component}")]
140    CircuitBreakerOpen { component: String },
141}
142
143/// Result type alias for pipeline operations
144pub type PipelineResult<T> = std::result::Result<T, PipelineError>;
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149
150    #[test]
151    fn test_pipeline_config_default() {
152        let config = PipelineConfig::default();
153        assert_eq!(config.max_batch_size, 1000);
154        assert_eq!(config.consistency_level, ConsistencyLevel::Session);
155        assert_eq!(config.backpressure_strategy, BackpressureStrategy::Adaptive);
156    }
157
158    #[test]
159    fn test_processing_priority_ordering() {
160        use ProcessingPriority::*;
161        assert!(Critical > High);
162        assert!(High > Normal);
163        assert!(Normal > Low);
164    }
165
166    #[test]
167    fn test_update_priority_ordering() {
168        use UpdatePriority::*;
169        assert!(Urgent > High);
170        assert!(High > Normal);
171        assert!(Normal > Background);
172    }
173
174    #[test]
175    fn test_alert_severity_ordering() {
176        use AlertSeverity::*;
177        assert!(Critical > Error);
178        assert!(Error > Warning);
179        assert!(Warning > Info);
180    }
181
182    #[test]
183    fn test_inconsistency_severity_ordering() {
184        use InconsistencySeverity::*;
185        assert!(Critical > High);
186        assert!(High > Medium);
187        assert!(Medium > Low);
188    }
189}
oxirs_vec/real_time_embedding_pipeline/mod.rs

oxirs_vec/real_time_embedding_pipeline/
mod.rs
