use crate::catalog::types::{Row, Value};
use crate::catalog::{
REACTIVE_PROCESSOR_CHECKPOINTS_TABLE, REACTIVE_PROCESSOR_DLQ_TABLE,
REACTIVE_PROCESSOR_REGISTRY_TABLE, SYSTEM_SCOPE_ID,
};
use crate::commit::validation::Mutation;
use crate::error::AedbError;
use crate::system_now_micros;
use crate::{AedbInstance, EventOutboxRecord};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Instant;
use tokio::sync::Mutex as AsyncMutex;
use tokio::task::JoinHandle;
pub(crate) fn reactive_processor_checkpoint_mutation(
processor_name: &str,
checkpoint_seq: u64,
) -> Mutation {
Mutation::Upsert {
project_id: crate::catalog::SYSTEM_PROJECT_ID.to_string(),
scope_id: SYSTEM_SCOPE_ID.to_string(),
table_name: REACTIVE_PROCESSOR_CHECKPOINTS_TABLE.to_string(),
primary_key: vec![Value::Text(processor_name.to_string().into())],
row: Row::from_values(vec![
Value::Text(processor_name.to_string().into()),
Value::Integer(checkpoint_seq as i64),
Value::Timestamp(system_now_micros() as i64),
]),
}
}
pub(crate) fn reactive_processor_registry_mutation(
processor_name: &str,
options: &ReactiveProcessorOptions,
enabled: bool,
) -> Result<Mutation, AedbError> {
let options_json =
serde_json::to_string(options).map_err(|e| AedbError::Encode(e.to_string()))?;
Ok(Mutation::Upsert {
project_id: crate::catalog::SYSTEM_PROJECT_ID.to_string(),
scope_id: SYSTEM_SCOPE_ID.to_string(),
table_name: REACTIVE_PROCESSOR_REGISTRY_TABLE.to_string(),
primary_key: vec![Value::Text(processor_name.to_string().into())],
row: Row::from_values(vec![
Value::Text(processor_name.to_string().into()),
Value::Json(options_json.into()),
Value::Boolean(enabled),
Value::Timestamp(system_now_micros() as i64),
]),
})
}
pub(crate) fn reactive_processor_dlq_mutation(
processor_name: &str,
events: &[EventOutboxRecord],
error: &str,
attempts: u32,
) -> Mutation {
let failed_at = system_now_micros() as i64;
let rows = events
.iter()
.map(|event| {
Row::from_values(vec![
Value::Text(processor_name.to_string().into()),
Value::Integer(event.commit_seq as i64),
Value::Text(event.topic.clone().into()),
Value::Text(event.event_key.clone().into()),
Value::Json(event.payload_json.clone().into()),
Value::Text(error.to_string().into()),
Value::Integer(attempts as i64),
Value::Timestamp(failed_at),
])
})
.collect();
Mutation::UpsertBatch {
project_id: crate::catalog::SYSTEM_PROJECT_ID.to_string(),
scope_id: SYSTEM_SCOPE_ID.to_string(),
table_name: REACTIVE_PROCESSOR_DLQ_TABLE.to_string(),
rows,
}
}
#[derive(Debug, Clone, Default)]
pub(crate) struct ReactiveCheckpointAckState {
pub(crate) last_persisted_seq: u64,
pub(crate) last_touch_micros: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub(crate) struct ReactiveCheckpointAckCacheKey {
pub(crate) processor_name: String,
pub(crate) caller_id: Option<String>,
}
pub(crate) const REACTIVE_ACK_CACHE_MAX_ENTRIES: usize = 1_024;
const REACTIVE_ACK_CACHE_EVICT_BATCH: usize = 128;
pub(crate) fn prune_reactive_ack_cache(
cache: &mut HashMap<ReactiveCheckpointAckCacheKey, ReactiveCheckpointAckState>,
) {
if cache.len() <= REACTIVE_ACK_CACHE_MAX_ENTRIES {
return;
}
let excess = cache.len() - REACTIVE_ACK_CACHE_MAX_ENTRIES;
let prune_count = excess
.saturating_add(REACTIVE_ACK_CACHE_EVICT_BATCH)
.min(cache.len());
let mut oldest: Vec<(ReactiveCheckpointAckCacheKey, u64)> = cache
.iter()
.map(|(k, v)| (k.clone(), v.last_touch_micros))
.collect();
oldest.sort_by_key(|(_, ts)| *ts);
for (key, _) in oldest.into_iter().take(prune_count) {
cache.remove(&key);
}
}
pub type ReactiveProcessorHandlerFuture =
Pin<Box<dyn Future<Output = Result<(), AedbError>> + Send>>;
pub type ReactiveProcessorHandler = Arc<
dyn Fn(Arc<AedbInstance>, Vec<EventOutboxRecord>) -> ReactiveProcessorHandlerFuture
+ Send
+ Sync,
>;
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(default)]
pub struct ReactiveProcessorOptions {
pub caller_id: Option<String>,
pub topic_filter: Option<String>,
pub run_on_interval: bool,
pub max_allowed_lag_commits: Option<u64>,
pub max_allowed_stall_ms: Option<u64>,
pub max_events_per_run: usize,
pub max_bytes_per_run: usize,
pub max_run_duration_ms: u64,
pub run_interval_ms: u64,
pub idle_backoff_ms: u64,
pub checkpoint_watermark_commits: u64,
pub max_retries: u32,
pub retry_backoff_ms: u64,
}
impl Default for ReactiveProcessorOptions {
fn default() -> Self {
Self {
caller_id: None,
topic_filter: None,
run_on_interval: false,
max_allowed_lag_commits: None,
max_allowed_stall_ms: None,
max_events_per_run: 256,
max_bytes_per_run: 2 * 1024 * 1024,
max_run_duration_ms: 250,
run_interval_ms: 20,
idle_backoff_ms: 50,
checkpoint_watermark_commits: 64,
max_retries: 3,
retry_backoff_ms: 100,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ReactiveProcessorRuntimeStatus {
pub processor_name: String,
pub running: bool,
pub runs_total: u64,
pub processed_events_total: u64,
pub failures_total: u64,
pub retries_total: u64,
pub dead_lettered_total: u64,
pub last_processed_seq: u64,
pub last_error: Option<String>,
pub last_run_started_micros: Option<u64>,
pub last_run_completed_micros: Option<u64>,
pub last_success_micros: Option<u64>,
pub last_failure_micros: Option<u64>,
pub last_retry_micros: Option<u64>,
pub last_sleep_ms: u64,
pub last_batch_events: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReactiveProcessorInfo {
pub processor_name: String,
pub options: ReactiveProcessorOptions,
pub enabled: bool,
pub running: bool,
pub updated_at_micros: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReactiveProcessorHealth {
pub processor_name: String,
pub enabled: bool,
pub running: bool,
pub checkpoint_seq: u64,
pub head_seq: u64,
pub lag_commits: u64,
pub runs_total: u64,
pub processed_events_total: u64,
pub failures_total: u64,
pub retries_total: u64,
pub dead_lettered_total: u64,
pub last_processed_seq: u64,
pub last_error: Option<String>,
pub last_run_started_micros: Option<u64>,
pub last_run_completed_micros: Option<u64>,
pub last_success_micros: Option<u64>,
pub last_failure_micros: Option<u64>,
pub last_retry_micros: Option<u64>,
pub last_sleep_ms: u64,
pub last_batch_events: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReactiveProcessorSloStatus {
pub processor_name: String,
pub breached: bool,
pub enabled: bool,
pub running: bool,
pub lag_commits: u64,
pub max_allowed_lag_commits: Option<u64>,
pub stall_ms: Option<u64>,
pub max_allowed_stall_ms: Option<u64>,
pub reasons: Vec<String>,
}
#[derive(Debug)]
pub(crate) struct ReactiveProcessorRuntime {
pub(crate) stop: Arc<AtomicBool>,
pub(crate) join: JoinHandle<()>,
pub(crate) stats: Arc<AsyncMutex<ReactiveProcessorRuntimeStatus>>,
}
#[derive(Debug, Clone)]
pub(crate) struct ReactiveProcessorRetryState {
pub(crate) events: Vec<EventOutboxRecord>,
pub(crate) last_seq: u64,
pub(crate) attempts: u32,
pub(crate) next_retry_at: Instant,
pub(crate) last_error: String,
}
#[derive(Debug, Clone)]
pub(crate) struct ReactiveProcessorRegistration {
pub(crate) processor_name: String,
pub(crate) options: ReactiveProcessorOptions,
pub(crate) enabled: bool,
pub(crate) updated_at_micros: u64,
}