55349845d0
Break monolith source files into focused modules: - stemedb-core/types.rs → types/ directory (assertion, source, gold_standard, etc.) - stemedb-storage: audit_store, quota_store, trust_rank_store, vector_index, vote_store → module directories - stemedb-ingest/worker.rs → worker/ with separate test modules - stemedb-query: engine, materializer, query → module directories - stemedb-lens: epoch_aware, skeptic → module directories - stemedb-sim/lib.rs → agent, arenas/, helpers, runner, strategy, types - stemedb-api/tests: integration_tests → http_basic, http_validation, http_epoch, http_pipeline - stemedb-api/tests: e2e_flow_test → e2e_full_pipeline, e2e_lens_resolution - stemedb-query/tests: e2e_pipeline → e2e_pipeline + e2e_decay Also adds new features: gold standard verification, escalation handlers, admin endpoints, concept hierarchy spec, arena roadmap, and Go SDK. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
54 lines
1.9 KiB
Rust
54 lines
1.9 KiB
Rust
//! Main run loop for the IngestWorker.
|
|
//!
|
|
//! Contains the continuous ingestion loop that tails the WAL.
|
|
|
|
use super::IngestWorker;
|
|
use crate::error::IngestError;
|
|
use std::sync::atomic::Ordering;
|
|
use stemedb_storage::KVStore;
|
|
use tracing::{debug, error, info, warn};
|
|
|
|
impl<S: KVStore + 'static> IngestWorker<S> {
|
|
/// Run the continuous ingestion loop, tailing the WAL until shutdown.
|
|
pub async fn run(&mut self) {
|
|
info!("Starting ingestion loop...");
|
|
loop {
|
|
// Check for shutdown signal
|
|
if self.shutdown.load(Ordering::Relaxed) {
|
|
info!("Shutdown signal received, stopping ingestion loop");
|
|
break;
|
|
}
|
|
|
|
match self.step().await {
|
|
Ok(0) => {
|
|
// No new data, sleep briefly
|
|
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
|
|
}
|
|
Ok(_) => {
|
|
// Processed data, continue immediately
|
|
}
|
|
Err(e) => {
|
|
// On shutdown, WAL errors are expected (files may be deleted)
|
|
if self.shutdown.load(Ordering::Relaxed) {
|
|
debug!("Error during shutdown (expected): {:?}", e);
|
|
break;
|
|
}
|
|
match &e {
|
|
IngestError::InputValidation(msg) => {
|
|
warn!("Rejected invalid input: {}", msg);
|
|
}
|
|
IngestError::InvalidSignature(msg) => {
|
|
warn!("Rejected invalid signature: {}", msg);
|
|
}
|
|
_ => {
|
|
error!("Ingestion error: {:?}", e);
|
|
}
|
|
}
|
|
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
|
|
}
|
|
}
|
|
}
|
|
info!("Ingestion loop stopped");
|
|
}
|
|
}
|