Something went wrong. Try again.
A fork of @slices.network/slices forked from slices.network/slices
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965use anyhow::Result;use async_trait::async_trait;use atproto_jetstream::{ CancellationToken, Consumer, ConsumerTaskConfig, EventHandler, JetstreamEvent,};use chrono::Utc;use reqwest::Client;use std::collections::HashSet;use std::sync::Arc;use tokio::sync::{Mutex, RwLock};use tracing::{error, info, warn};
use crate::actor_resolver::resolve_actor_data;use crate::cache::{CacheBackend, CacheFactory, SliceCache};use crate::database::Database;use crate::errors::JetstreamError;use crate::jetstream_cursor::PostgresCursorHandler;use crate::logging::{LogLevel, Logger};use crate::models::{Actor, Record};
pub struct JetstreamConsumer { consumer: Consumer, database: Database, http_client: Client, actor_cache: Arc<Mutex<SliceCache>>, lexicon_cache: Arc<Mutex<SliceCache>>, domain_cache: Arc<Mutex<SliceCache>>, collections_cache: Arc<Mutex<SliceCache>>, pub event_count: Arc<std::sync::atomic::AtomicU64>, cursor_handler: Option<Arc<PostgresCursorHandler>>, slices_list: Arc<RwLock<Vec<String>>>,}
// Event handler that implements the EventHandler traitstruct SliceEventHandler { database: Database, http_client: Client, event_count: Arc<std::sync::atomic::AtomicU64>, actor_cache: Arc<Mutex<SliceCache>>, lexicon_cache: Arc<Mutex<SliceCache>>, domain_cache: Arc<Mutex<SliceCache>>, collections_cache: Arc<Mutex<SliceCache>>, cursor_handler: Option<Arc<PostgresCursorHandler>>, slices_list: Arc<RwLock<Vec<String>>>,}
#[async_trait]impl EventHandler for SliceEventHandler { async fn handle_event(&self, event: JetstreamEvent) -> Result<()> { let count = self .event_count .fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
if count.is_multiple_of(10000) { info!("Jetstream consumer has processed {} events", count); }
// Extract and update cursor position from event let time_us = match &event { JetstreamEvent::Commit { time_us, .. } => *time_us, JetstreamEvent::Delete { time_us, .. } => *time_us, JetstreamEvent::Identity { time_us, .. } => *time_us, JetstreamEvent::Account { time_us, .. } => *time_us, };
if let Some(cursor_handler) = &self.cursor_handler { cursor_handler.update_position(time_us);
// Periodically write cursor to DB (debounced by handler) if let Err(e) = cursor_handler.maybe_write_cursor().await { error!("Failed to write cursor: {}", e); } }
match event { JetstreamEvent::Commit { did, commit, .. } => { if let Err(e) = self.handle_commit_event(&did, commit).await { let message = format!("Error handling commit event: {}", e); error!("{}", message); Logger::global().log_jetstream( LogLevel::Error, &message, Some(serde_json::json!({ "error": e.to_string(), "did": did, "event_type": "commit" })), ); } } JetstreamEvent::Delete { did, commit, .. } => { if let Err(e) = self.handle_delete_event(&did, commit).await { let message = format!("Error handling delete event: {}", e); error!("{}", message); Logger::global().log_jetstream( LogLevel::Error, &message, Some(serde_json::json!({ "error": e.to_string(), "did": did, "event_type": "delete" })), ); } } _ => { // Ignore other event types (identity, account, etc.) } } Ok(()) }
fn handler_id(&self) -> String { "slice-records-indexer".to_string() }}
impl SliceEventHandler { /// Check if DID is an actor for the given slice async fn is_actor_cached( &self, did: &str, slice_uri: &str, ) -> Result<Option<bool>, anyhow::Error> { match self.actor_cache.lock().await.is_actor(did, slice_uri).await { Ok(result) => Ok(result), Err(e) => { warn!( error = ?e, did = did, slice_uri = slice_uri, "Actor cache error" ); Ok(None) } } }
/// Cache that an actor exists async fn cache_actor_exists(&self, did: &str, slice_uri: &str) { if let Err(e) = self .actor_cache .lock() .await .cache_actor_exists(did, slice_uri) .await { warn!( error = ?e, did = did, slice_uri = slice_uri, "Failed to cache actor exists" ); } }
/// Remove actor from cache async fn remove_actor_from_cache(&self, did: &str, slice_uri: &str) { if let Err(e) = self .actor_cache .lock() .await .remove_actor(did, slice_uri) .await { warn!( error = ?e, did = did, slice_uri = slice_uri, "Failed to remove actor from cache" ); } }
/// Get slice collections from cache with database fallback async fn get_slice_collections( &self, slice_uri: &str, ) -> Result<Option<HashSet<String>>, anyhow::Error> { // Try cache first let cache_result = { let mut cache = self.collections_cache.lock().await; cache.get_slice_collections(slice_uri).await };
match cache_result { Ok(Some(collections)) => Ok(Some(collections)), Ok(None) => { // Cache miss - load from database match self.database.get_slice_collections_list(slice_uri).await { Ok(collections) => { let collections_set: HashSet<String> = collections.into_iter().collect(); // Cache the result let _ = self .collections_cache .lock() .await .cache_slice_collections(slice_uri, &collections_set) .await; Ok(Some(collections_set)) } Err(e) => Err(e.into()), } } Err(e) => Err(e), } }
/// Get slice domain from cache with database fallback async fn get_slice_domain(&self, slice_uri: &str) -> Result<Option<String>, anyhow::Error> { // Try cache first let cache_result = { let mut cache = self.domain_cache.lock().await; cache.get_slice_domain(slice_uri).await };
match cache_result { Ok(Some(domain)) => Ok(Some(domain)), Ok(None) => { // Cache miss - load from database match self.database.get_slice_domain(slice_uri).await { Ok(Some(domain)) => { // Cache the result let _ = self .domain_cache .lock() .await .cache_slice_domain(slice_uri, &domain) .await; Ok(Some(domain)) } Ok(None) => Ok(None), Err(e) => Err(e.into()), } } Err(e) => Err(e), } }
/// Get slice lexicons from cache with database fallback async fn get_slice_lexicons( &self, slice_uri: &str, ) -> Result<Option<Vec<serde_json::Value>>, anyhow::Error> { // Try cache first let cache_result = { let mut cache = self.lexicon_cache.lock().await; cache.get_lexicons(slice_uri).await };
match cache_result { Ok(Some(lexicons)) => Ok(Some(lexicons)), Ok(None) => { // Cache miss - load from database match self.database.get_lexicons_by_slice(slice_uri).await { Ok(lexicons) if !lexicons.is_empty() => { // Cache the result let _ = self .lexicon_cache .lock() .await .cache_lexicons(slice_uri, &lexicons) .await; Ok(Some(lexicons)) } Ok(_) => Ok(None), // Empty lexicons Err(e) => Err(e.into()), } } Err(e) => Err(e), } } async fn handle_commit_event( &self, did: &str, commit: atproto_jetstream::JetstreamEventCommit, ) -> Result<()> { // Get all slices from cached list let slices = self.slices_list.read().await.clone();
// Process each slice for slice_uri in slices { // Get collections for this slice (with caching) let collections = match self.get_slice_collections(&slice_uri).await { Ok(Some(collections)) => collections, Ok(None) => continue, // No collections for this slice Err(e) => { error!("Failed to get collections for slice {}: {}", slice_uri, e); continue; } };
if collections.contains(&commit.collection) { // Special handling for network.slices.lexicon records // These should only be indexed to the slice specified in their JSON data if commit.collection == "network.slices.lexicon" { if let Some(target_slice_uri) = commit.record.get("slice").and_then(|v| v.as_str()) { // Skip this slice if it's not the target slice for this lexicon if slice_uri != target_slice_uri { continue; } } else { // No target slice specified, skip this lexicon record entirely continue; } } // Get the domain for this slice (with caching) let domain = match self.get_slice_domain(&slice_uri).await { Ok(Some(domain)) => domain, Ok(None) => continue, // No domain, skip Err(e) => { error!("Failed to get domain for slice {}: {}", slice_uri, e); continue; } };
// Check if this is a primary collection (starts with slice domain) let is_primary_collection = commit.collection.starts_with(&domain);
// For external collections, check actor status BEFORE expensive validation if !is_primary_collection { let is_actor = match self.is_actor_cached(did, &slice_uri).await { Ok(Some(cached_result)) => cached_result, Ok(None) => { // Cache miss means this DID is not an actor we've synced // For external collections, we only care about actors we've already added false } Err(e) => { error!("Error checking actor status: {}", e); continue; } };
if !is_actor { // Not an actor - skip validation entirely for external collections continue; } }
// Get lexicons for validation (after actor check for external collections) let lexicons = match self.get_slice_lexicons(&slice_uri).await { Ok(Some(lexicons)) => lexicons, Ok(None) => { info!( "No lexicons found for slice {} - skipping validation", slice_uri ); continue; } Err(e) => { error!("Failed to get lexicons for slice {}: {}", slice_uri, e); continue; } };
// Validate the record against the slice's lexicons let validation_result = match slices_lexicon::validate_record( lexicons.clone(), &commit.collection, commit.record.clone(), ) { Ok(_) => { info!( "Record validated for collection {} in slice {}", commit.collection, slice_uri ); true } Err(e) => { let message = format!( "Validation failed for collection {} in slice {}", commit.collection, slice_uri ); error!("{}: {}", message, e); Logger::global().log_jetstream_with_slice( LogLevel::Warn, &message, Some(serde_json::json!({ "collection": commit.collection, "slice_uri": slice_uri, "did": did })), Some(&slice_uri), ); false } };
if !validation_result { continue; // Skip this slice if validation fails }
if is_primary_collection { // Primary collection - ensure actor exists and index ALL records info!( "Primary collection {} for slice {} (domain: {}) - indexing record", commit.collection, slice_uri, domain );
// Ensure actor exists for primary collections let is_cached = matches!(self.is_actor_cached(did, &slice_uri).await, Ok(Some(_)));
if !is_cached { // Actor not in cache - create it info!("Creating new actor {} for slice {}", did, slice_uri);
// Resolve actor data (handle, PDS) match resolve_actor_data(&self.http_client, did).await { Ok(actor_data) => { let actor = Actor { did: actor_data.did.clone(), handle: actor_data.handle, slice_uri: slice_uri.clone(), indexed_at: Utc::now().to_rfc3339(), };
// Insert into database if let Err(e) = self.database.batch_insert_actors(&[actor]).await { error!("Failed to create actor {}: {}", did, e); } else { // Add to cache after successful database insert self.cache_actor_exists(did, &slice_uri).await; info!("Created actor {} for slice {}", did, slice_uri); } } Err(e) => { error!("Failed to resolve actor data for {}: {}", did, e); } } }
let uri = format!("at://{}/{}/{}", did, commit.collection, commit.rkey);
let record = Record { uri: uri.clone(), cid: commit.cid.clone(), did: did.to_string(), collection: commit.collection.clone(), json: commit.record.clone(), indexed_at: Utc::now(), slice_uri: Some(slice_uri.clone()), };
match self.database.upsert_record(&record).await { Ok(is_insert) => { let message = if is_insert { format!("Record inserted in {}", commit.collection) } else { format!("Record updated in {}", commit.collection) }; Logger::global().log_jetstream_with_slice( LogLevel::Info, &message, Some(serde_json::json!({ "did": did, "kind": "commit", "commit": { "rev": commit.rev, "operation": commit.operation, "collection": commit.collection, "rkey": commit.rkey, "record": commit.record, "cid": commit.cid }, "indexed_operation": if is_insert { "insert" } else { "update" }, "record_type": "primary" })), Some(&slice_uri), ); } Err(e) => { let message = "Failed to insert/update record"; Logger::global().log_jetstream_with_slice( LogLevel::Error, message, Some(serde_json::json!({ "did": did, "kind": "commit", "commit": { "rev": commit.rev, "operation": commit.operation, "collection": commit.collection, "rkey": commit.rkey, "record": commit.record, "cid": commit.cid }, "error": e.to_string(), "record_type": "primary" })), Some(&slice_uri), ); return Err(anyhow::anyhow!("Database error: {}", e)); } }
info!( "Successfully indexed {} record from primary collection: {}", commit.operation, uri ); break; } else { // External collection - we already checked actor status, so just index info!( "External collection {} - DID {} is actor in slice {} - indexing", commit.collection, did, slice_uri );
let uri = format!("at://{}/{}/{}", did, commit.collection, commit.rkey);
let record = Record { uri: uri.clone(), cid: commit.cid.clone(), did: did.to_string(), collection: commit.collection.clone(), json: commit.record.clone(), indexed_at: Utc::now(), slice_uri: Some(slice_uri.clone()), };
match self.database.upsert_record(&record).await { Ok(is_insert) => { let message = if is_insert { format!("Record inserted in {}", commit.collection) } else { format!("Record updated in {}", commit.collection) }; Logger::global().log_jetstream_with_slice( LogLevel::Info, &message, Some(serde_json::json!({ "did": did, "kind": "commit", "commit": { "rev": commit.rev, "operation": commit.operation, "collection": commit.collection, "rkey": commit.rkey, "record": commit.record, "cid": commit.cid }, "indexed_operation": if is_insert { "insert" } else { "update" }, "record_type": "external" })), Some(&slice_uri), ); } Err(e) => { let message = "Failed to insert/update record"; Logger::global().log_jetstream_with_slice( LogLevel::Error, message, Some(serde_json::json!({ "did": did, "kind": "commit", "commit": { "rev": commit.rev, "operation": commit.operation, "collection": commit.collection, "rkey": commit.rkey, "record": commit.record, "cid": commit.cid }, "error": e.to_string(), "record_type": "external" })), Some(&slice_uri), ); return Err(anyhow::anyhow!("Database error: {}", e)); } }
info!( "Successfully indexed {} record from external collection: {}", commit.operation, uri ); break; } } }
Ok(()) }
async fn handle_delete_event( &self, did: &str, commit: atproto_jetstream::JetstreamEventDelete, ) -> Result<()> { let uri = format!("at://{}/{}/{}", did, commit.collection, commit.rkey);
// Get all slices from cached list let slices = self.slices_list.read().await.clone();
let mut relevant_slices: Vec<String> = Vec::new();
for slice_uri in slices { // Get collections for this slice (with caching) let collections = match self.get_slice_collections(&slice_uri).await { Ok(Some(collections)) => collections, Ok(None) => continue, // No collections for this slice Err(e) => { error!("Failed to get collections for slice {}: {}", slice_uri, e); continue; } };
if !collections.contains(&commit.collection) { continue; }
// Get the domain for this slice (with caching) let domain = match self.get_slice_domain(&slice_uri).await { Ok(Some(domain)) => domain, Ok(None) => continue, // No domain, skip Err(e) => { error!("Failed to get domain for slice {}: {}", slice_uri, e); continue; } };
// Check if this is a primary collection (starts with slice domain) let is_primary_collection = commit.collection.starts_with(&domain);
if is_primary_collection { // Primary collection - always process deletes relevant_slices.push(slice_uri.clone()); } else { // External collection - only process if DID is an actor in this slice let is_actor = match self.is_actor_cached(did, &slice_uri).await { Ok(Some(cached_result)) => cached_result, _ => false, }; if is_actor { relevant_slices.push(slice_uri.clone()); } } }
if relevant_slices.is_empty() { // No relevant slices found, skip deletion return Ok(()); }
// Handle cascade deletion before deleting the record if let Err(e) = self .database .handle_cascade_deletion(&uri, &commit.collection) .await { warn!("Cascade deletion failed for {}: {}", uri, e); }
// Delete the record and log only for relevant slices match self.database.delete_record_by_uri(&uri, None).await { Ok(rows_affected) => { if rows_affected > 0 { info!( "Deleted record: {} ({} rows) for {} slice(s)", uri, rows_affected, relevant_slices.len() ); let message = format!("Record deleted from {}", commit.collection);
// Log to each relevant slice and check if actor cleanup is needed for slice_uri in &relevant_slices { Logger::global().log_jetstream_with_slice( LogLevel::Info, &message, Some(serde_json::json!({ "operation": "delete", "collection": commit.collection, "did": did, "uri": uri, "rows_affected": rows_affected })), Some(slice_uri), ); }
// Check if actor should be cleaned up (no more records) for slice_uri in &relevant_slices { match self.database.actor_has_records(did, slice_uri).await { Ok(has_records) => { if !has_records { // No more records for this actor in this slice - clean up match self.database.delete_actor(did, slice_uri).await { Ok(deleted) => { if deleted > 0 { info!( "Cleaned up actor {} from slice {} (no records remaining)", did, slice_uri ); // Remove from cache self.remove_actor_from_cache(did, slice_uri).await; } } Err(e) => { error!( "Failed to delete actor {} from slice {}: {}", did, slice_uri, e ); } } } } Err(e) => { error!( "Failed to check if actor {} has records in slice {}: {}", did, slice_uri, e ); } } } } } Err(e) => { let message = "Failed to delete record"; error!("{}: {}", message, e);
// Log error to each relevant slice for slice_uri in relevant_slices { Logger::global().log_jetstream_with_slice( LogLevel::Error, message, Some(serde_json::json!({ "operation": "delete", "collection": commit.collection, "did": did, "uri": uri, "error": e.to_string() })), Some(&slice_uri), ); } } }
Ok(()) }}
impl JetstreamConsumer { /// Create a new Jetstream consumer with optional cursor support and Redis cache /// /// # Arguments /// * `database` - Database connection for slice configurations and record storage /// * `jetstream_hostname` - Optional custom jetstream hostname /// * `cursor_handler` - Optional cursor handler for resumable event processing /// * `initial_cursor` - Optional starting cursor position (time_us) to resume from /// * `redis_url` - Optional Redis URL for caching (falls back to in-memory if not provided) pub async fn new( database: Database, jetstream_hostname: Option<String>, cursor_handler: Option<Arc<PostgresCursorHandler>>, initial_cursor: Option<i64>, redis_url: Option<String>, ) -> Result<Self, JetstreamError> { let config = ConsumerTaskConfig { user_agent: "slice-server/1.0".to_string(), compression: false, zstd_dictionary_location: String::new(), jetstream_hostname: jetstream_hostname .unwrap_or_else(|| "jetstream1.us-east.bsky.network".to_string()), collections: Vec::new(), dids: Vec::new(), max_message_size_bytes: None, cursor: initial_cursor, require_hello: true, };
let consumer = Consumer::new(config); let http_client = Client::new();
// Determine cache backend based on Redis URL let cache_backend = if let Some(redis_url) = redis_url { CacheBackend::Redis { url: redis_url, ttl_seconds: None, } } else { CacheBackend::InMemory { ttl_seconds: None } };
// Create cache instances let actor_cache = Arc::new(Mutex::new( CacheFactory::create_slice_cache(cache_backend.clone()) .await .map_err(|e| JetstreamError::ConnectionFailed { message: format!("Failed to create actor cache: {}", e), })?, ));
let lexicon_cache = Arc::new(Mutex::new( CacheFactory::create_slice_cache(cache_backend.clone()) .await .map_err(|e| JetstreamError::ConnectionFailed { message: format!("Failed to create lexicon cache: {}", e), })?, ));
let domain_cache = Arc::new(Mutex::new( CacheFactory::create_slice_cache(cache_backend.clone()) .await .map_err(|e| JetstreamError::ConnectionFailed { message: format!("Failed to create domain cache: {}", e), })?, ));
let collections_cache = Arc::new(Mutex::new( CacheFactory::create_slice_cache(cache_backend) .await .map_err(|e| JetstreamError::ConnectionFailed { message: format!("Failed to create collections cache: {}", e), })?, ));
Ok(Self { consumer, database, http_client, actor_cache, lexicon_cache, domain_cache, collections_cache, event_count: Arc::new(std::sync::atomic::AtomicU64::new(0)), cursor_handler, slices_list: Arc::new(RwLock::new(Vec::new())), }) }
/// Load slice configurations pub async fn load_slice_configurations(&self) -> Result<(), JetstreamError> { info!("Loading slice configurations...");
// Get all slices and update cached list let slices = self.database.get_all_slices().await?; *self.slices_list.write().await = slices.clone(); info!("Found {} total slices in database", slices.len());
Ok(()) }
/// Preload actor cache to avoid database hits during event processing async fn preload_actor_cache(&self) -> Result<(), JetstreamError> { info!("Preloading actor cache...");
let actors = self.database.get_all_actors().await?; info!("Found {} actors to cache", actors.len());
match self.actor_cache.lock().await.preload_actors(actors).await { Ok(_) => { info!("Actor cache preloaded successfully"); Ok(()) } Err(e) => { warn!(error = ?e, "Failed to preload actors to cache"); Ok(()) // Don't fail startup if preload fails } } }
/// Start consuming events from Jetstream pub async fn start_consuming( &self, cancellation_token: CancellationToken, ) -> Result<(), JetstreamError> { info!("Starting Jetstream consumer");
// Load initial slice configurations self.load_slice_configurations().await?;
// Preload actor cache self.preload_actor_cache().await?;
// Create and register the event handler let handler = Arc::new(SliceEventHandler { database: self.database.clone(), http_client: self.http_client.clone(), event_count: self.event_count.clone(), actor_cache: self.actor_cache.clone(), lexicon_cache: self.lexicon_cache.clone(), domain_cache: self.domain_cache.clone(), collections_cache: self.collections_cache.clone(), cursor_handler: self.cursor_handler.clone(), slices_list: self.slices_list.clone(), });
self.consumer.register_handler(handler).await.map_err(|e| { JetstreamError::ConnectionFailed { message: format!("Failed to register event handler: {}", e), } })?;
// Start periodic status reporting (with cancellation support) let event_count_for_status = self.event_count.clone(); let cancellation_token_for_status = cancellation_token.clone(); tokio::spawn(async move { let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(60)); // Every minute loop { tokio::select! { _ = interval.tick() => { let count = event_count_for_status.load(std::sync::atomic::Ordering::Relaxed); info!( "Jetstream consumer status: {} total events processed", count ); } _ = cancellation_token_for_status.cancelled() => { info!("Status reporting task cancelled"); break; } } } });
// Start the consumer info!("Starting Jetstream background consumer..."); let result = self .consumer .run_background(cancellation_token) .await .map_err(|e| JetstreamError::ConnectionFailed { message: format!("Consumer failed: {}", e), });
// Force write cursor on shutdown to ensure latest position is persisted if let Some(cursor_handler) = &self.cursor_handler { if let Err(e) = cursor_handler.force_write_cursor().await { error!("Failed to write final cursor position: {}", e); } else { info!("Final cursor position written to database"); } }
result?; Ok(()) }
/// Periodically reload slice configurations and actor cache to pick up new slices/collections/actors pub fn start_configuration_reloader(consumer: Arc<Self>) { tokio::spawn(async move { let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(300)); // Reload every 5 minutes interval.tick().await; // Skip first immediate tick
loop { interval.tick().await;
if let Err(e) = consumer.load_slice_configurations().await { error!("Failed to reload slice configurations: {}", e); }
if let Err(e) = consumer.preload_actor_cache().await { error!("Failed to reload actor cache: {}", e); } } }); }}