diff --git a/.agents/skills/moa-100-session-sweep/SKILL.md b/.agents/skills/moa-100-session-sweep/SKILL.md
index e6bc90d6c..ca454489c 100644
--- a/.agents/skills/moa-100-session-sweep/SKILL.md
+++ b/.agents/skills/moa-100-session-sweep/SKILL.md
@@ -99,6 +99,29 @@ Useful env overrides:
- `MOA_SWEEP_IDS`: comma-separated session ids for focused lanes
- `MOA_SWEEP_MODEL`: pinned model, usually `gpt-5.4-mini`
- `MOA_SWEEP_REDIS_URL`: runtime-cache Redis URL, default `redis://127.0.0.1:10051/0`
+- `MOA_SWEEP_PROVIDER_MAX_IN_FLIGHT`: per-credential provider in-flight budget for the
+ sweep orchestrator, default `64`. Chat calls are bounded per provider credential by
+ default (16 unless configured), which a full sweep can saturate; the runner passes this
+ through as `MOA_OPENAI_MAX_CONCURRENT_REQUESTS` so the budget never shapes outcomes.
+ The runner also forces `MOA_PROVIDERS_CONCURRENCY_SCOPE=local` so the sweep never
+ shares Redis lease budgets with the long-running compose orchestrator.
+
+## Runtime-Behavior Notes (defaults changed 2026-07-10)
+
+Interpretation guidance for sweeps at or after the 2026-07-10 defaults changes:
+
+- **Non-builtin (MCP) tools default to admin review.** Sweep personas use builtin tools
+ only, so current cases are unaffected — but any future persona that registers an MCP
+ tool will stall on an approval unless the case seeds an operator allow rule or handles
+ the review flow. Do not read such a stall as a delegation/fan-in regression.
+- **Skill distillation now requires 8+ tool calls per segment** (was 5). This gates
+ skill *learning*, not skill *activation* — `F-SKILL-INJECT` reads persisted
+ `skills_activated` segment evidence and is unaffected. Expect fewer
+ distillation proposals from short sweep sessions; that is intended, not a regression.
+- **Provider chat concurrency is bounded per credential by default.** The runner sizes
+ the sweep budget via `MOA_SWEEP_PROVIDER_MAX_IN_FLIGHT` (see above). If a sweep shows
+ clustered latency spikes or rate-limit failovers, check that override before suspecting
+ a provider or orchestrator regression.
## Baseline Rules
diff --git a/.agents/skills/moa-100-session-sweep/scripts/run_100_session_sweep.py b/.agents/skills/moa-100-session-sweep/scripts/run_100_session_sweep.py
index a6391247c..c07ee9dcc 100755
--- a/.agents/skills/moa-100-session-sweep/scripts/run_100_session_sweep.py
+++ b/.agents/skills/moa-100-session-sweep/scripts/run_100_session_sweep.py
@@ -39,6 +39,11 @@
# Minimum contiguous match length (chars) that counts as the final reply leaking raw worker output.
RAW_LEAK_MIN_CHARS = 120
MAX_WORKERS = int(os.environ.get("MOA_SWEEP_CONCURRENCY", "4"))
+# Provider in-flight budget for the sweep's single live key. Chat calls are
+# bounded per provider credential by default (16 unless configured), which a
+# full sweep (MOA_SWEEP_CONCURRENCY sessions x coordinator + spawned workers)
+# can saturate; size generously so the budget never shapes sweep outcomes.
+PROVIDER_MAX_IN_FLIGHT = os.environ.get("MOA_SWEEP_PROVIDER_MAX_IN_FLIGHT", "64")
SESSION_TIMEOUT_S = int(os.environ.get("MOA_SWEEP_SESSION_TIMEOUT_S", "260"))
TURN_LIMIT = int(os.environ.get("MOA_SWEEP_MAX_TURNS", "6"))
CASE_LIMIT = int(os.environ.get("MOA_SWEEP_LIMIT", "0"))
@@ -279,6 +284,12 @@ def start_orchestrator(env):
"MOA_LOCAL_DOCKER_ENABLED": "false",
"MOA_RUNTIME_CACHE_BACKEND": "redis",
"MOA_RUNTIME_CACHE_REDIS_URL": SWEEP_REDIS_URL,
+ # The sweep shares the compose Redis; force local concurrency scope
+ # so an inherited global scope can never make the sweep orchestrator
+ # share provider lease budgets with the long-running compose stack.
+ "MOA_PROVIDERS_CONCURRENCY_SCOPE": "local",
+ # Sweep model is routed through the OpenAI provider credential.
+ "MOA_OPENAI_MAX_CONCURRENT_REQUESTS": PROVIDER_MAX_IN_FLIGHT,
"RUST_LOG": os.environ.get("RUST_LOG", "info,moa_orchestrator=info"),
}
)
diff --git a/Cargo.lock b/Cargo.lock
index e608ff761..ae73a034e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3320,6 +3320,7 @@ dependencies = [
"insta",
"moa-core",
"moa-db",
+ "moa-test-support",
"serde_json",
"sqlx",
"thiserror 2.0.18",
@@ -3414,6 +3415,7 @@ version = "0.1.0"
dependencies = [
"async-trait",
"httpmock",
+ "metrics",
"moa-authz-schema",
"moa-core",
"moa-migrations",
@@ -3728,6 +3730,7 @@ dependencies = [
"blake3",
"bytes",
"chrono",
+ "futures-util",
"hex",
"hmac",
"liteparse",
@@ -4117,6 +4120,7 @@ dependencies = [
"constant_time_eq 0.3.1",
"hex",
"hmac",
+ "metrics",
"moa-core",
"moa-migrations",
"moka",
@@ -4224,6 +4228,7 @@ dependencies = [
"metrics",
"moa-core",
"moa-observability",
+ "moa-runtime-store",
"moa-test-support",
"opentelemetry",
"reqwest 0.13.4",
@@ -4272,7 +4277,6 @@ dependencies = [
"chrono",
"globset",
"moa-core",
- "moka",
"serde_json",
"shell-words",
"tokio",
diff --git a/SEQUENCE-DIAGRAMS.md b/SEQUENCE-DIAGRAMS.md
index 49b4d1fd9..bd6288d25 100644
--- a/SEQUENCE-DIAGRAMS.md
+++ b/SEQUENCE-DIAGRAMS.md
@@ -57,7 +57,7 @@ sequenceDiagram
Pipe-->>Brain: WorkingContext (cache_breakpoint marked)
Brain->>LLM: complete(compiled_context)
- LLM-->>Brain: ToolCall { bash: "fly deploy --app staging" }
+ LLM-->>Brain: ToolCall { bash: "deploy --app staging" }
Brain->>Log: emit ApprovalRequested
Orch->>Messaging: observe event
@@ -70,7 +70,7 @@ sequenceDiagram
Orch->>Brain: deliver signal
Brain->>Log: emit ApprovalDecided
- Brain->>Router: execute("bash", "fly deploy ...")
+ Brain->>Router: execute("bash", "deploy ...")
Router->>Hand: provision (lazy, first call)
Hand-->>Router: HandHandle
Router->>Hand: execute(bash, input)
@@ -384,7 +384,7 @@ sequenceDiagram
OldBrain->>Log: emit ToolCall
OldBrain->>Log: emit ToolResult
- Note over OldBrain: 💥 process crash / machine restart
(panic, OOM, Fly.io suspend, Ctrl+C)
+ Note over OldBrain: 💥 process crash / machine restart
(panic, OOM, host suspend, Ctrl+C)
Note over Orch: On next signal OR startup scan
Orch->>Log: list_sessions(status in [Running, WaitingApproval])
diff --git a/crates/moa-analytics/Cargo.toml b/crates/moa-analytics/Cargo.toml
index e68b7755c..6f4630760 100644
--- a/crates/moa-analytics/Cargo.toml
+++ b/crates/moa-analytics/Cargo.toml
@@ -19,4 +19,5 @@ workspace-hack.workspace = true
[dev-dependencies]
clickhouse = { workspace = true, features = ["test-util"] }
insta.workspace = true
+moa-test-support.workspace = true
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
diff --git a/crates/moa-analytics/src/clickhouse_exec.rs b/crates/moa-analytics/src/clickhouse_exec.rs
index 48acc4c98..0e1290a2e 100644
--- a/crates/moa-analytics/src/clickhouse_exec.rs
+++ b/crates/moa-analytics/src/clickhouse_exec.rs
@@ -34,8 +34,17 @@ fn is_unknown_table(error: &clickhouse::error::Error) -> bool {
#[derive(Clone)]
pub struct AnalyticsClickHouseClient {
client: Client,
+ max_execution_time_secs: u64,
+ max_rows_to_read: u64,
+ max_bytes_to_read: u64,
}
+/// Default ClickHouse per-query budgets applied when the caller does not
+/// override them from config. Mirror [`moa_core::config::AnalyticsConfig`].
+const DEFAULT_MAX_EXECUTION_TIME_SECS: u64 = 10;
+const DEFAULT_MAX_ROWS_TO_READ: u64 = 1_000_000_000;
+const DEFAULT_MAX_BYTES_TO_READ: u64 = 10_000_000_000;
+
impl std::fmt::Debug for AnalyticsClickHouseClient {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AnalyticsClickHouseClient")
@@ -56,13 +65,33 @@ impl AnalyticsClickHouseClient {
if let Some(password) = config.password.as_deref() {
client = client.with_password(password);
}
- Self { client }
+ Self::from_client(client)
}
/// Wraps an existing client, used by tests to point at a mock server.
#[must_use]
pub fn from_client(client: Client) -> Self {
- Self { client }
+ Self {
+ client,
+ max_execution_time_secs: DEFAULT_MAX_EXECUTION_TIME_SECS,
+ max_rows_to_read: DEFAULT_MAX_ROWS_TO_READ,
+ max_bytes_to_read: DEFAULT_MAX_BYTES_TO_READ,
+ }
+ }
+
+ /// Overrides the per-query ClickHouse budgets (`max_execution_time`,
+ /// `max_rows_to_read`, `max_bytes_to_read`) so a runaway read is bounded.
+ #[must_use]
+ pub fn with_query_budgets(
+ mut self,
+ max_execution_time_secs: u64,
+ max_rows_to_read: u64,
+ max_bytes_to_read: u64,
+ ) -> Self {
+ self.max_execution_time_secs = max_execution_time_secs;
+ self.max_rows_to_read = max_rows_to_read;
+ self.max_bytes_to_read = max_bytes_to_read;
+ self
}
/// Deletes a tenant's rows from every analytics table during offboarding.
@@ -107,7 +136,17 @@ impl AnalyticsClickHouseClient {
&self,
compiled: &CompiledAnalyticsQuery,
) -> Result>> {
- let mut query = self.client.query(&compiled.sql);
+ // Bound the read: a runaway scan is cancelled by ClickHouse instead of
+ // reading a tenant's full history to satisfy an exact ordered percentile.
+ let mut query = self
+ .client
+ .query(&compiled.sql)
+ .with_option(
+ "max_execution_time",
+ self.max_execution_time_secs.to_string(),
+ )
+ .with_option("max_rows_to_read", self.max_rows_to_read.to_string())
+ .with_option("max_bytes_to_read", self.max_bytes_to_read.to_string());
for bind in &compiled.bind_values {
query = match bind {
AnalyticsBindValue::String(value) => query.bind(value),
diff --git a/crates/moa-analytics/src/compiler.rs b/crates/moa-analytics/src/compiler.rs
index 59ad99fb5..b38f70cf7 100644
--- a/crates/moa-analytics/src/compiler.rs
+++ b/crates/moa-analytics/src/compiler.rs
@@ -729,7 +729,15 @@ mod tests {
aggregation: AnalyticsAggregation::P95,
alias: Some("p95_cost".to_string()),
}],
- filters: Vec::new(),
+ // Time-series datasets require a bounded window; a recent lower bound
+ // keeps the shared fixture valid against the wall clock.
+ filters: vec![AnalyticsFilter {
+ field: "finished_at".to_string(),
+ operator: AnalyticsFilterOperator::Gte,
+ value: Some(AnalyticsCell::String(
+ (chrono::Utc::now() - chrono::Duration::days(1)).to_rfc3339(),
+ )),
+ }],
order_by: vec![AnalyticsOrderBy {
field: "p95_cost".to_string(),
direction: AnalyticsSortDirection::Desc,
@@ -748,7 +756,8 @@ mod tests {
assert!(compiled.sql.contains("d.tenant_id = $1::UUID"));
assert!(compiled.sql.contains("PERCENTILE_CONT(0.95)"));
assert!(compiled.sql.contains("ORDER BY c1 DESC"));
- assert_eq!(compiled.bind_values.len(), 1);
+ // Tenant scope bind plus the required time-window lower bound.
+ assert_eq!(compiled.bind_values.len(), 2);
assert_eq!(compiled.limit, 25);
}
diff --git a/crates/moa-analytics/src/error.rs b/crates/moa-analytics/src/error.rs
index 965866219..3844ff662 100644
--- a/crates/moa-analytics/src/error.rs
+++ b/crates/moa-analytics/src/error.rs
@@ -120,6 +120,20 @@ pub enum AnalyticsError {
/// Maximum allowed time window in whole days.
max_days: i64,
},
+ /// A time-series dataset was queried without a bounded time window.
+ #[error(
+ "analytics dataset `{dataset}` requires a bounded time filter on `{time_field}` \
+ (a two-sided `between` within {max_days} days, or a lower bound `gte`/`gt`/`eq` \
+ no older than {max_days} days)"
+ )]
+ MissingTimeWindow {
+ /// Dataset that requires a bounded time window.
+ dataset: String,
+ /// Designated time field that must be bounded.
+ time_field: String,
+ /// Maximum allowed time window in whole days.
+ max_days: i64,
+ },
/// A query tried to order by a field or alias that is not selected.
#[error("order field `{field}` is not selected by the query")]
UnknownOrderField {
diff --git a/crates/moa-analytics/src/executor.rs b/crates/moa-analytics/src/executor.rs
index 20a5e9551..38569b5de 100644
--- a/crates/moa-analytics/src/executor.rs
+++ b/crates/moa-analytics/src/executor.rs
@@ -21,9 +21,23 @@ use crate::error::{AnalyticsError, Result};
/// [`AnalyticsService::clickhouse`] compiles and executes against the ClickHouse
/// read models. The edge selects the constructor by `[clickhouse]` presence and
/// calls the matching `query` / `query_clickhouse` entrypoint.
-#[derive(Debug, Clone, Default)]
+/// Default Postgres `statement_timeout` applied to each analytics query when
+/// the caller does not override it from config.
+pub const DEFAULT_STATEMENT_TIMEOUT_MS: u64 = 10_000;
+
+#[derive(Debug, Clone)]
pub struct AnalyticsService {
compiler: AnalyticsCompiler,
+ statement_timeout_ms: u64,
+}
+
+impl Default for AnalyticsService {
+ fn default() -> Self {
+ Self {
+ compiler: AnalyticsCompiler::default(),
+ statement_timeout_ms: DEFAULT_STATEMENT_TIMEOUT_MS,
+ }
+ }
}
impl AnalyticsService {
@@ -39,9 +53,18 @@ impl AnalyticsService {
analytics_catalog(),
AnalyticsBackend::ClickHouse,
),
+ statement_timeout_ms: DEFAULT_STATEMENT_TIMEOUT_MS,
}
}
+ /// Overrides the Postgres `statement_timeout` applied to each query, in
+ /// milliseconds, so a runaway scan is cancelled server-side.
+ #[must_use]
+ pub fn with_statement_timeout_ms(mut self, statement_timeout_ms: u64) -> Self {
+ self.statement_timeout_ms = statement_timeout_ms;
+ self
+ }
+
/// Returns the backend this service compiles and executes against.
pub fn backend(&self) -> AnalyticsBackend {
self.compiler.backend()
@@ -67,6 +90,14 @@ impl AnalyticsService {
let mut conn = ScopedConn::begin_tenant(pool, compiled.effective_tenant_id)
.await
.map_err(|error| AnalyticsError::Execution(error.to_string()))?;
+ // Bound the database work: an unbounded ordered-percentile scan is
+ // cancelled server-side rather than holding a connection open. `SET LOCAL`
+ // scopes the timeout to this transaction only.
+ sqlx::query("SELECT set_config('statement_timeout', $1, true)")
+ .bind(self.statement_timeout_ms.to_string())
+ .execute(conn.as_mut())
+ .await
+ .map_err(|error| AnalyticsError::Execution(error.to_string()))?;
let rows = execute_compiled(&compiled, conn.as_mut()).await?;
conn.commit()
.await
diff --git a/crates/moa-analytics/src/query.rs b/crates/moa-analytics/src/query.rs
index dafb9f230..ba1c00300 100644
--- a/crates/moa-analytics/src/query.rs
+++ b/crates/moa-analytics/src/query.rs
@@ -3,8 +3,9 @@
use chrono::{DateTime, Utc};
use moa_core::TenantId;
use moa_core::wire::analytics::{
- AnalyticsAggregation, AnalyticsCatalogResponse, AnalyticsCell, AnalyticsField,
- AnalyticsFieldKind, AnalyticsFieldRole, AnalyticsFilterOperator, AnalyticsQueryRequest,
+ AnalyticsAggregation, AnalyticsCatalogResponse, AnalyticsCell, AnalyticsDataset,
+ AnalyticsField, AnalyticsFieldKind, AnalyticsFieldRole, AnalyticsFilter,
+ AnalyticsFilterOperator, AnalyticsQueryRequest,
};
use crate::catalog::find_dataset;
@@ -38,6 +39,16 @@ pub struct ValidatedAnalyticsQuery {
pub fn validate_query(
catalog: &AnalyticsCatalogResponse,
request: AnalyticsQueryRequest,
+) -> Result {
+ validate_query_at(catalog, request, Utc::now())
+}
+
+/// Validates a query against a caller-supplied `now`, used by tests to pin the
+/// time-window bound against a fixed clock.
+pub fn validate_query_at(
+ catalog: &AnalyticsCatalogResponse,
+ request: AnalyticsQueryRequest,
+ now: DateTime,
) -> Result {
let dataset =
find_dataset(catalog, &request.dataset).ok_or_else(|| AnalyticsError::UnknownDataset {
@@ -118,6 +129,8 @@ pub fn validate_query(
validate_time_window(field, filter.operator, filter.value.as_ref())?;
}
+ enforce_required_time_window(dataset, &request.filters, now)?;
+
for order in &request.order_by {
if !selected_field_or_alias(&request, &order.field) {
return Err(AnalyticsError::UnknownOrderField {
@@ -264,6 +277,74 @@ fn validate_time_window(
Ok(())
}
+/// Requires a time-series dataset to carry a filter that bounds its scan to a
+/// window no wider than [`MAX_TIME_WINDOW_DAYS`].
+///
+/// A dataset opts out of the requirement by declaring no `default_time_field`.
+/// Otherwise at least one filter on that field must close the window: a
+/// two-sided `between` (span already checked by [`validate_time_window`]), an
+/// `eq` point, or a lower bound (`gte`/`gt`) whose start is no older than the
+/// limit. An upper bound alone (`lt`/`lte`) leaves history unbounded and does
+/// not satisfy the requirement.
+fn enforce_required_time_window(
+ dataset: &AnalyticsDataset,
+ filters: &[AnalyticsFilter],
+ now: DateTime,
+) -> Result<()> {
+ let Some(time_field) = dataset.default_time_field.as_deref() else {
+ return Ok(());
+ };
+
+ let mut bounded = false;
+ for filter in filters {
+ if filter.field != time_field {
+ continue;
+ }
+ match filter.operator {
+ AnalyticsFilterOperator::Between | AnalyticsFilterOperator::Eq => {
+ bounded = true;
+ }
+ AnalyticsFilterOperator::Gte | AnalyticsFilterOperator::Gt => {
+ let start = lower_bound_timestamp(time_field, filter.value.as_ref())?;
+ let days = now.signed_duration_since(start).num_days();
+ if days > MAX_TIME_WINDOW_DAYS {
+ return Err(AnalyticsError::TimeWindowTooLarge {
+ days,
+ max_days: MAX_TIME_WINDOW_DAYS,
+ });
+ }
+ bounded = true;
+ }
+ _ => {}
+ }
+ }
+
+ if bounded {
+ Ok(())
+ } else {
+ Err(AnalyticsError::MissingTimeWindow {
+ dataset: dataset.id.clone(),
+ time_field: time_field.to_string(),
+ max_days: MAX_TIME_WINDOW_DAYS,
+ })
+ }
+}
+
+fn lower_bound_timestamp(field: &str, value: Option<&AnalyticsCell>) -> Result> {
+ let raw = value
+ .and_then(cell_as_string)
+ .ok_or_else(|| AnalyticsError::InvalidFilterValue {
+ field: field.to_string(),
+ reason: "timestamp lower bound must be an RFC3339 string".to_string(),
+ })?;
+ DateTime::parse_from_rfc3339(raw)
+ .map(|timestamp| timestamp.with_timezone(&Utc))
+ .map_err(|error| AnalyticsError::InvalidFilterValue {
+ field: field.to_string(),
+ reason: error.to_string(),
+ })
+}
+
fn parse_timestamp_json(field: &str, value: &serde_json::Value) -> Result> {
let Some(value) = value.as_str() else {
return Err(AnalyticsError::InvalidFilterValue {
@@ -335,3 +416,134 @@ fn role_name(role: AnalyticsFieldRole) -> &'static str {
AnalyticsFieldRole::FilterOnly => "filter",
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::catalog::analytics_catalog;
+ use chrono::TimeZone;
+ use moa_core::wire::analytics::{AnalyticsDimension, AnalyticsMeasure};
+
+ fn fixed_now() -> DateTime {
+ Utc.with_ymd_and_hms(2026, 7, 10, 12, 0, 0)
+ .single()
+ .expect("now")
+ }
+
+ fn sessions_request(filters: Vec) -> AnalyticsQueryRequest {
+ AnalyticsQueryRequest {
+ dataset: "sessions".to_string(),
+ tenant_id: Some(TenantId::new()),
+ dimensions: vec![AnalyticsDimension {
+ field: "channel".to_string(),
+ alias: None,
+ }],
+ measures: vec![AnalyticsMeasure {
+ field: None,
+ aggregation: AnalyticsAggregation::Count,
+ alias: None,
+ }],
+ filters,
+ order_by: Vec::new(),
+ limit: Some(10),
+ }
+ }
+
+ fn created_at(op: AnalyticsFilterOperator, value: &str) -> AnalyticsFilter {
+ AnalyticsFilter {
+ field: "created_at".to_string(),
+ operator: op,
+ value: Some(AnalyticsCell::String(value.to_string())),
+ }
+ }
+
+ #[test]
+ fn rejects_query_without_time_window() {
+ // Pins: a time-series dataset queried with no time filter is rejected so it
+ // cannot scan a tenant's full event history.
+ let error = validate_query_at(
+ &analytics_catalog(),
+ sessions_request(Vec::new()),
+ fixed_now(),
+ )
+ .expect_err("unbounded query must be rejected");
+ assert!(matches!(error, AnalyticsError::MissingTimeWindow { .. }));
+ }
+
+ #[test]
+ fn rejects_upper_bound_only_time_window() {
+ // Pins: an upper bound alone (`lt`/`lte`) leaves history unbounded.
+ let filters = vec![created_at(
+ AnalyticsFilterOperator::Lt,
+ "2026-07-01T00:00:00Z",
+ )];
+ let error = validate_query_at(&analytics_catalog(), sessions_request(filters), fixed_now())
+ .expect_err("upper-bound-only must be rejected");
+ assert!(matches!(error, AnalyticsError::MissingTimeWindow { .. }));
+ }
+
+ #[test]
+ fn rejects_lower_bound_older_than_max_window() {
+ // Pins: a lower bound older than the limit is a too-wide window, not a pass.
+ let filters = vec![created_at(
+ AnalyticsFilterOperator::Gte,
+ "2024-01-01T00:00:00Z",
+ )];
+ let error = validate_query_at(&analytics_catalog(), sessions_request(filters), fixed_now())
+ .expect_err("too-wide lower bound must be rejected");
+ assert!(matches!(error, AnalyticsError::TimeWindowTooLarge { .. }));
+ }
+
+ #[test]
+ fn accepts_recent_lower_bound() {
+ let filters = vec![created_at(
+ AnalyticsFilterOperator::Gte,
+ "2026-06-10T00:00:00Z",
+ )];
+ validate_query_at(&analytics_catalog(), sessions_request(filters), fixed_now())
+ .expect("a recent lower bound is accepted");
+ }
+
+ #[test]
+ fn accepts_between_within_window() {
+ let between = AnalyticsFilter {
+ field: "created_at".to_string(),
+ operator: AnalyticsFilterOperator::Between,
+ value: Some(AnalyticsCell::Json(serde_json::json!([
+ "2026-06-01T00:00:00Z",
+ "2026-07-01T00:00:00Z"
+ ]))),
+ };
+ validate_query_at(
+ &analytics_catalog(),
+ sessions_request(vec![between]),
+ fixed_now(),
+ )
+ .expect("a bounded between is accepted");
+ }
+
+ #[test]
+ fn accepts_eq_point_bound() {
+ let filters = vec![created_at(
+ AnalyticsFilterOperator::Eq,
+ "2026-06-10T00:00:00Z",
+ )];
+ validate_query_at(&analytics_catalog(), sessions_request(filters), fixed_now())
+ .expect("an eq point bound is accepted");
+ }
+
+ #[test]
+ fn dataset_without_time_field_is_exempt() {
+ // Pins: a dataset that declares no default_time_field opts out of the window
+ // requirement instead of being unqueryable.
+ let dataset = AnalyticsDataset {
+ id: "catalog_meta".to_string(),
+ label: "Meta".to_string(),
+ description: String::new(),
+ default_time_field: None,
+ fields: Vec::new(),
+ };
+ enforce_required_time_window(&dataset, &[], fixed_now())
+ .expect("a dataset without a time field is exempt");
+ }
+}
diff --git a/crates/moa-analytics/tests/clickhouse_executor_offline.rs b/crates/moa-analytics/tests/clickhouse_executor_offline.rs
index d76078e10..cf916d666 100644
--- a/crates/moa-analytics/tests/clickhouse_executor_offline.rs
+++ b/crates/moa-analytics/tests/clickhouse_executor_offline.rs
@@ -12,7 +12,8 @@ use clickhouse::test::{Mock, handlers};
use moa_analytics::{AnalyticsClickHouseClient, AnalyticsService};
use moa_core::TenantId;
use moa_core::wire::analytics::{
- AnalyticsAggregation, AnalyticsDimension, AnalyticsMeasure, AnalyticsQueryRequest,
+ AnalyticsAggregation, AnalyticsCell, AnalyticsDimension, AnalyticsFilter,
+ AnalyticsFilterOperator, AnalyticsMeasure, AnalyticsQueryRequest,
};
#[tokio::test]
@@ -38,7 +39,15 @@ async fn clickhouse_executor_returns_empty_result_and_metadata_offline() {
aggregation: AnalyticsAggregation::P95,
alias: None,
}],
- filters: Vec::new(),
+ // Time-series datasets require a bounded window; a recent lower bound
+ // satisfies the validator against the wall clock.
+ filters: vec![AnalyticsFilter {
+ field: "created_at".to_string(),
+ operator: AnalyticsFilterOperator::Gte,
+ value: Some(AnalyticsCell::String(
+ (chrono::Utc::now() - chrono::Duration::days(1)).to_rfc3339(),
+ )),
+ }],
order_by: Vec::new(),
limit: Some(10),
};
diff --git a/crates/moa-analytics/tests/executor_statement_timeout_db.rs b/crates/moa-analytics/tests/executor_statement_timeout_db.rs
new file mode 100644
index 000000000..ea193082a
--- /dev/null
+++ b/crates/moa-analytics/tests/executor_statement_timeout_db.rs
@@ -0,0 +1,102 @@
+//! DB coverage for the analytics executor's per-query Postgres budget.
+
+use moa_analytics::AnalyticsService;
+use moa_core::TenantId;
+use moa_core::wire::analytics::{
+ AnalyticsCell, AnalyticsDimension, AnalyticsFilter, AnalyticsFilterOperator, AnalyticsMeasure,
+ AnalyticsQueryRequest,
+};
+use moa_test_support::postgres::{TestDb, bootstrap_test_db};
+
+async fn configured_test_db() -> Option {
+ std::env::var_os("MOA_DATABASE_URL")?;
+ Some(
+ bootstrap_test_db()
+ .await
+ .expect("bootstrap Postgres test database"),
+ )
+}
+
+#[tokio::test]
+#[ignore = "requires MOA_DATABASE_URL and a reachable Postgres instance"]
+async fn analytics_query_applies_statement_timeout_and_runs_db() {
+ // Pins: the Postgres analytics executor runs each query inside a tenant-scoped
+ // transaction with a bounded statement_timeout set, and the query path
+ // succeeds against the empty read models.
+ let Some(test_db) = configured_test_db().await else {
+ return;
+ };
+ let tenant = TenantId::new();
+ let request = AnalyticsQueryRequest {
+ dataset: "sessions".to_string(),
+ tenant_id: Some(tenant),
+ dimensions: vec![AnalyticsDimension {
+ field: "channel".to_string(),
+ alias: None,
+ }],
+ measures: vec![AnalyticsMeasure {
+ field: None,
+ aggregation: moa_core::wire::analytics::AnalyticsAggregation::Count,
+ alias: None,
+ }],
+ filters: vec![AnalyticsFilter {
+ field: "created_at".to_string(),
+ operator: AnalyticsFilterOperator::Gte,
+ value: Some(AnalyticsCell::String(
+ (chrono::Utc::now() - chrono::Duration::days(1)).to_rfc3339(),
+ )),
+ }],
+ order_by: Vec::new(),
+ limit: Some(10),
+ };
+
+ let response = AnalyticsService::new()
+ .with_statement_timeout_ms(5_000)
+ .query(test_db.store().pool(), request)
+ .await
+ .expect("analytics query runs under the statement-timeout budget");
+ assert_eq!(response.metadata.row_count, 0, "empty read model");
+}
+
+#[tokio::test]
+#[ignore = "requires MOA_DATABASE_URL and a reachable Postgres instance"]
+async fn statement_timeout_cancels_a_slow_query_db() {
+ // Pins: the per-transaction statement_timeout the executor sets actually
+ // cancels a runaway query server-side (SQLSTATE 57014), so an unbounded
+ // ordered-percentile scan cannot hold a connection indefinitely.
+ let Some(test_db) = configured_test_db().await else {
+ return;
+ };
+ let mut tx = test_db
+ .store()
+ .pool()
+ .begin()
+ .await
+ .expect("begin transaction");
+ // Same mechanism as executor.rs: SET LOCAL statement_timeout for this tx.
+ sqlx::query("SELECT set_config('statement_timeout', $1, true)")
+ .bind("100")
+ .execute(tx.as_mut())
+ .await
+ .expect("set statement timeout");
+
+ let error = sqlx::query("SELECT pg_sleep(3)")
+ .execute(tx.as_mut())
+ .await
+ .expect_err("a query longer than the timeout must be cancelled");
+ let cancelled = error
+ .as_database_error()
+ .and_then(|db_error| db_error.code().map(|code| code.into_owned()))
+ .as_deref()
+ == Some("57014")
+ || error
+ .to_string()
+ .to_lowercase()
+ .contains("statement timeout");
+ assert!(
+ cancelled,
+ "expected a statement-timeout cancellation, got: {error}"
+ );
+
+ let _ = tx.rollback().await;
+}
diff --git a/crates/moa-auth/auth0/src/auth0_provider.rs b/crates/moa-auth/auth0/src/auth0_provider.rs
index 907e14a47..319b98aa3 100644
--- a/crates/moa-auth/auth0/src/auth0_provider.rs
+++ b/crates/moa-auth/auth0/src/auth0_provider.rs
@@ -4,7 +4,7 @@
//! The provider maps the external `sub` claim to MOA's internal UUID through
//! the `auth0_user_map` table, creating a local row on first login.
-use crate::jwks_cache::JwksCache;
+use crate::jwks_cache::{JwksCache, JwksError};
use async_trait::async_trait;
use jsonwebtoken::{Algorithm, Validation, decode, decode_header};
use moa_core::TenantId;
@@ -95,11 +95,7 @@ impl AuthProvider for Auth0AuthProvider {
return Err(AuthError::Rejected);
}
let kid = header.kid.ok_or(AuthError::InvalidFormat)?;
- let key = self
- .jwks
- .key_for(&kid)
- .await
- .map_err(|error| AuthError::Unavailable(format!("jwks: {error}")))?;
+ let key = self.jwks.key_for(&kid).await.map_err(map_jwks_error)?;
let mut validation = Validation::new(Algorithm::RS256);
validation.set_issuer(&[&self.issuer]);
@@ -229,6 +225,21 @@ async fn provision_static(
Ok(resolved_id)
}
+/// Maps a JWKS key-resolution failure to an authentication outcome.
+///
+/// A `kid` a successful refresh proved the provider does not publish is an
+/// invalid credential ([`AuthError::Rejected`]); an unreachable or unparseable
+/// JWKS endpoint is a provider-availability failure ([`AuthError::Unavailable`])
+/// so callers do not treat a transient outage as a bad token.
+pub(crate) fn map_jwks_error(error: JwksError) -> AuthError {
+ match error {
+ JwksError::UnknownKid(_) => AuthError::Rejected,
+ JwksError::Http(_) | JwksError::Parse(_) => {
+ AuthError::Unavailable(format!("jwks: {error}"))
+ }
+ }
+}
+
pub(crate) fn parse_identity_type(value: Option<&str>) -> Result {
match value.unwrap_or("operator") {
"operator" => Ok(IdentityType::Operator),
diff --git a/crates/moa-auth/auth0/src/jwks_cache.rs b/crates/moa-auth/auth0/src/jwks_cache.rs
index 932b41ebf..33f6845c1 100644
--- a/crates/moa-auth/auth0/src/jwks_cache.rs
+++ b/crates/moa-auth/auth0/src/jwks_cache.rs
@@ -1,74 +1,267 @@
//! In-memory JWKS cache for JWT signature validation.
//!
-//! The cache refreshes on `kid` miss and after its configured TTL. It stores
-//! only public decoding keys and is safe to clone across request handlers.
+//! The cache serves known `kid`s from a last-known-good key set and refreshes
+//! that set — never discarding it — when a `kid` is unknown or its keys age past
+//! the configured TTL. It stores only public decoding keys and is safe to clone
+//! across request handlers.
+//!
+//! Because the JWT header (and therefore its `kid`) is attacker-controllable
+//! before signature validation, refreshes are hardened against amplification:
+//!
+//! * A miss never evicts the last-known-good set; the set is replaced only when
+//! a refresh succeeds, so known keys keep validating during a provider outage.
+//! * A single-flight refresh gate collapses concurrent misses into one fetch and
+//! enforces a cooldown, so a flood of distinct random `kid`s cannot drive one
+//! outbound JWKS request each.
+//! * Unknown `kid`s confirmed absent by a successful refresh are negative-cached
+//! for a short, bounded window, so repeated probes of the same `kid` do not
+//! refetch.
+//! * The JWKS HTTP client carries tight connect/request/body-size budgets.
use std::collections::HashMap;
use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
use jsonwebtoken::{DecodingKey, jwk::JwkSet};
use moka::future::Cache;
use thiserror::Error;
+use tokio::sync::{Mutex, RwLock};
/// Decoding keys from one JWKS fetch, cached as a single unit so that a `kid`
/// rotation refreshes every key together rather than per entry.
type KeySet = Arc>;
+/// Connect-phase budget for one JWKS fetch.
+const HTTP_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);
+/// Total wall-clock budget for one JWKS fetch (connect, send, and read).
+const HTTP_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
+/// Maximum accepted JWKS response body. Real documents are a few kilobytes; this
+/// bounds memory against a misbehaving or compromised provider endpoint.
+const MAX_JWKS_BODY_BYTES: u64 = 1024 * 1024;
+/// Minimum interval between JWKS refresh attempts. Rate-limits refreshes so that
+/// a burst of distinct unknown `kid`s cannot amplify into one fetch each.
+const DEFAULT_REFRESH_COOLDOWN: Duration = Duration::from_secs(10);
+/// How long a `kid` confirmed absent by a successful refresh is remembered as
+/// unknown so repeated probes are rejected without refetching.
+const DEFAULT_NEGATIVE_TTL: Duration = Duration::from_secs(60);
+/// Bound on the negative cache; the keyspace is attacker-controlled.
+const DEFAULT_NEGATIVE_CAPACITY: u64 = 1024;
+
+/// Mutable last-known-good state guarded by [`JwksCache::state`].
+#[derive(Default)]
+struct KeyState {
+ /// Most recent successfully fetched key set (empty before the first fetch).
+ keys: KeySet,
+ /// When [`Self::keys`] was last successfully refreshed; drives TTL freshness.
+ refreshed_at: Option,
+ /// When a refresh was last attempted (success or failure); drives the
+ /// cooldown so a failing provider is not hammered.
+ last_attempt: Option,
+}
+
+impl KeyState {
+ /// Whether the current key set is within its freshness TTL.
+ fn is_fresh(&self, ttl: Duration) -> bool {
+ self.refreshed_at
+ .is_some_and(|refreshed| refreshed.elapsed() < ttl)
+ }
+
+ /// Whether a refresh was attempted within the cooldown window.
+ fn within_cooldown(&self, cooldown: Duration) -> bool {
+ self.last_attempt
+ .is_some_and(|attempt| attempt.elapsed() < cooldown)
+ }
+}
+
/// Thread-safe cache of public JWT decoding keys fetched from one JWKS URL.
#[derive(Clone)]
pub struct JwksCache {
- keys: Cache<(), KeySet>,
+ inner: Arc,
+}
+
+struct Inner {
jwks_url: String,
http: reqwest::Client,
+ /// Last-known-good key set and refresh bookkeeping.
+ state: RwLock,
+ /// Single-flight refresh gate: held across the miss re-check, cooldown
+ /// decision, fetch, and replace so concurrent misses collapse into one
+ /// fetch and observe each other's cooldown.
+ refresh: Mutex<()>,
+ /// Bounded, short-TTL memory of `kid`s a successful refresh proved absent.
+ negative: Cache,
+ ttl: Duration,
+ cooldown: Duration,
}
impl JwksCache {
- /// Create a JWKS cache for `jwks_url` with entries valid for `ttl`.
+ /// Create a JWKS cache for `jwks_url` whose keys are considered fresh for
+ /// `ttl`, with default refresh cooldown and negative-cache policy.
#[must_use]
pub fn new(jwks_url: impl Into, ttl: Duration) -> Self {
+ Self::with_policy(
+ jwks_url,
+ ttl,
+ DEFAULT_REFRESH_COOLDOWN,
+ DEFAULT_NEGATIVE_TTL,
+ DEFAULT_NEGATIVE_CAPACITY,
+ )
+ }
+
+ /// Create a JWKS cache with explicit refresh cooldown and negative-cache
+ /// policy. Exposed for tests that must exercise cooldown and negative-TTL
+ /// windows without real-time waits.
+ #[must_use]
+ fn with_policy(
+ jwks_url: impl Into,
+ ttl: Duration,
+ cooldown: Duration,
+ negative_ttl: Duration,
+ negative_capacity: u64,
+ ) -> Self {
+ // Tight budgets: an unverified attacker-supplied `kid` can drive a fetch,
+ // so a slow or hostile endpoint must not tie up an auth request.
+ let http = reqwest::Client::builder()
+ .connect_timeout(HTTP_CONNECT_TIMEOUT)
+ .timeout(HTTP_REQUEST_TIMEOUT)
+ .build()
+ .unwrap_or_else(|error| {
+ tracing::warn!(?error, "falling back to default JWKS HTTP client");
+ reqwest::Client::new()
+ });
Self {
- keys: Cache::builder().time_to_live(ttl).build(),
- jwks_url: jwks_url.into(),
- http: reqwest::Client::new(),
+ inner: Arc::new(Inner {
+ jwks_url: jwks_url.into(),
+ http,
+ state: RwLock::new(KeyState::default()),
+ refresh: Mutex::new(()),
+ negative: Cache::builder()
+ .max_capacity(negative_capacity)
+ .time_to_live(negative_ttl)
+ .build(),
+ ttl,
+ cooldown,
+ }),
}
}
- /// Return the decoding key for `kid`, refreshing the cache when needed.
+ /// Return the decoding key for `kid`.
+ ///
+ /// Serves a fresh known `kid` from cache without any network call. Otherwise
+ /// takes the single-flight refresh path, which — subject to the cooldown —
+ /// refreshes the key set and either resolves `kid`, serves a still-valid
+ /// last-known-good key during a provider outage, negative-caches a confirmed
+ /// unknown `kid` ([`JwksError::UnknownKid`]), or surfaces the provider
+ /// failure ([`JwksError::Http`]/[`JwksError::Parse`]) without discarding
+ /// valid keys.
pub async fn key_for(&self, kid: &str) -> Result {
- // Fast path: an unexpired key set that already contains `kid`.
- if let Some(keys) = self.keys.get(&()).await
- && let Some(key) = keys.get(kid)
- {
- return Ok(key.clone());
+ let inner = self.inner.as_ref();
+
+ // Fast path: a known `kid` whose key set is still within its TTL.
+ if let Some(key) = inner.fresh_known_key(kid).await {
+ return Ok(key);
+ }
+ // A `kid` a recent successful refresh proved absent: reject without a
+ // fetch. Checked before taking the refresh gate so a repeated probe is
+ // cheap.
+ if inner.negative.get(kid).await.is_some() {
+ return Err(JwksError::UnknownKid(kid.to_string()));
}
- // Slow path: cold cache, expired TTL, or a `kid` that may have rotated
- // in. Drop any stale set and refetch the whole JWKS (coalescing
- // concurrent refreshes) before looking `kid` up again.
- self.keys.invalidate(&()).await;
- self.keys
- .try_get_with((), self.fetch_keys())
- .await
- .map_err(|error: Arc| (*error).clone())?
- .get(kid)
- .cloned()
- .ok_or_else(|| JwksError::UnknownKid(kid.to_string()))
+ // Slow path: single-flight. Concurrent misses queue here so at most one
+ // fetch runs; each re-checks the state the winner just published.
+ let _guard = inner.refresh.lock().await;
+
+ if let Some(key) = inner.fresh_known_key(kid).await {
+ return Ok(key);
+ }
+ if inner.negative.get(kid).await.is_some() {
+ return Err(JwksError::UnknownKid(kid.to_string()));
+ }
+
+ if inner.state.read().await.within_cooldown(inner.cooldown) {
+ // Rate-limited: no fetch. Serve a stale-but-known key (availability)
+ // or reject an unknown `kid` as an invalid credential. A `kid` not
+ // fetched here is not negative-cached — only a successful refresh may
+ // confirm absence.
+ return inner
+ .known_key(kid)
+ .await
+ .ok_or_else(|| JwksError::UnknownKid(kid.to_string()));
+ }
+
+ inner.state.write().await.last_attempt = Some(Instant::now());
+ match inner.fetch_keys().await {
+ Ok(new_keys) => {
+ {
+ let mut state = inner.state.write().await;
+ state.keys = new_keys.clone();
+ state.refreshed_at = Some(Instant::now());
+ }
+ match new_keys.get(kid) {
+ Some(key) => Ok(key.clone()),
+ None => {
+ inner.negative.insert(kid.to_string(), ()).await;
+ Err(JwksError::UnknownKid(kid.to_string()))
+ }
+ }
+ }
+ // Provider failure must not discard still-valid keys: serve the
+ // last-known-good key for `kid` if we have one, else surface the
+ // provider error.
+ Err(error) => inner.known_key(kid).await.ok_or(error),
+ }
+ }
+}
+
+impl Inner {
+ /// Returns the decoding key for `kid` only if the key set is fresh.
+ async fn fresh_known_key(&self, kid: &str) -> Option {
+ let state = self.state.read().await;
+ state
+ .is_fresh(self.ttl)
+ .then(|| state.keys.get(kid).cloned())
+ .flatten()
+ }
+
+ /// Returns the decoding key for `kid` from the last-known-good set,
+ /// regardless of freshness.
+ async fn known_key(&self, kid: &str) -> Option {
+ self.state.read().await.keys.get(kid).cloned()
}
async fn fetch_keys(&self) -> Result {
tracing::info!(jwks_url = %self.jwks_url, "refreshing JWKS");
- let jwks: JwkSet = self
+ let response = self
.http
.get(&self.jwks_url)
.send()
.await
.map_err(|error| JwksError::Http(error.to_string()))?
.error_for_status()
- .map_err(|error| JwksError::Http(error.to_string()))?
- .json()
+ .map_err(|error| JwksError::Http(error.to_string()))?;
+
+ // Body-size budget: reject an over-large declared length before reading,
+ // and cap the actually-read body as a backstop.
+ if response
+ .content_length()
+ .is_some_and(|len| len > MAX_JWKS_BODY_BYTES)
+ {
+ return Err(JwksError::Http(format!(
+ "JWKS body exceeds {MAX_JWKS_BODY_BYTES} byte budget"
+ )));
+ }
+ let body = response
+ .bytes()
.await
- .map_err(|error| JwksError::Parse(error.to_string()))?;
+ .map_err(|error| JwksError::Http(error.to_string()))?;
+ if body.len() as u64 > MAX_JWKS_BODY_BYTES {
+ return Err(JwksError::Http(format!(
+ "JWKS body exceeds {MAX_JWKS_BODY_BYTES} byte budget"
+ )));
+ }
+ let jwks: JwkSet =
+ serde_json::from_slice(&body).map_err(|error| JwksError::Parse(error.to_string()))?;
let mut keys = HashMap::new();
for jwk in jwks.keys {
@@ -113,39 +306,52 @@ mod tests {
use rsa::traits::PublicKeyParts;
use rsa::{RsaPrivateKey, rand_core::OsRng};
- /// Builds a one-key JWKS document for `kid` backed by a fresh RSA key.
- fn jwks_json(kid: &str) -> serde_json::Value {
+ /// Builds one JWK entry for `kid` backed by a fresh RSA key.
+ fn jwk_entry(kid: &str) -> serde_json::Value {
let key = RsaPrivateKey::new(&mut OsRng, 2048).expect("generate RSA test key");
let public = key.to_public_key();
serde_json::json!({
- "keys": [{
- "kty": "RSA",
- "kid": kid,
- "use": "sig",
- "alg": "RS256",
- "n": URL_SAFE_NO_PAD.encode(public.n().to_bytes_be()),
- "e": URL_SAFE_NO_PAD.encode(public.e().to_bytes_be()),
- }]
+ "kty": "RSA",
+ "kid": kid,
+ "use": "sig",
+ "alg": "RS256",
+ "n": URL_SAFE_NO_PAD.encode(public.n().to_bytes_be()),
+ "e": URL_SAFE_NO_PAD.encode(public.e().to_bytes_be()),
+ })
+ }
+
+ /// Builds a JWKS document covering every `kid` in `kids`.
+ fn jwks_body(kids: &[&str]) -> serde_json::Value {
+ serde_json::json!({
+ "keys": kids.iter().map(|kid| jwk_entry(kid)).collect::>(),
})
}
+ /// JWKS cache with explicit cooldown/negative windows for deterministic
+ /// timing without real-time waits.
+ fn cache_with(
+ url: String,
+ ttl: Duration,
+ cooldown: Duration,
+ negative_ttl: Duration,
+ ) -> JwksCache {
+ JwksCache::with_policy(url, ttl, cooldown, negative_ttl, 1024)
+ }
+
#[tokio::test]
- async fn caches_within_ttl_and_refetches_whole_jwks_on_unknown_kid() {
- // Pins: a `kid` hit inside the TTL is served from cache without a refetch,
- // while an unknown `kid` forces a single whole-JWKS refresh (rotation).
+ async fn known_kid_served_from_cache_within_ttl_without_refetch() {
+ // Pins: a `kid` hit inside the TTL is served from cache with no refetch.
let server = MockServer::start_async().await;
let mock = server
.mock_async(|when, then| {
when.method(GET).path("/jwks.json");
- then.status(200).json_body(jwks_json("kid-1"));
+ then.status(200).json_body(jwks_body(&["kid-1"]));
})
.await;
-
let cache = JwksCache::new(server.url("/jwks.json"), Duration::from_secs(3600));
cache.key_for("kid-1").await.expect("known kid resolves");
assert_eq!(mock.hits_async().await, 1, "first lookup fetches the JWKS");
-
cache
.key_for("kid-1")
.await
@@ -155,15 +361,197 @@ mod tests {
1,
"second lookup is served from cache"
);
+ }
+
+ #[tokio::test]
+ async fn unknown_kid_probe_during_outage_does_not_evict_known_keys() {
+ // Pins: F12 — a miss that triggers a failing refresh must not discard the
+ // last-known-good set; known keys keep validating during a provider outage.
+ let server = MockServer::start_async().await;
+ let ok = server
+ .mock_async(|when, then| {
+ when.method(GET).path("/jwks.json");
+ then.status(200).json_body(jwks_body(&["kid-1"]));
+ })
+ .await;
+ let cache = cache_with(
+ server.url("/jwks.json"),
+ Duration::from_secs(3600),
+ Duration::from_millis(1),
+ Duration::from_secs(60),
+ );
+ cache.key_for("kid-1").await.expect("known kid resolves");
+ assert_eq!(ok.hits_async().await, 1);
+
+ // The endpoint now errors for every fetch.
+ ok.delete_async().await;
+ let outage = server
+ .mock_async(|when, then| {
+ when.method(GET).path("/jwks.json");
+ then.status(500);
+ })
+ .await;
+ tokio::time::sleep(Duration::from_millis(5)).await; // exit the refresh cooldown
+
+ // An unknown kid drives a refresh that fails; it surfaces as a provider
+ // error (not eviction), because the fetched set never arrived.
+ match cache.key_for("kid-2").await {
+ Err(JwksError::Http(_)) => {}
+ Err(other) => panic!("expected a provider error, got {other:?}"),
+ Ok(_) => panic!("unknown kid must not resolve against the retained set"),
+ }
assert!(
- matches!(cache.key_for("kid-2").await, Err(JwksError::UnknownKid(_))),
- "unknown kid is rejected"
+ outage.hits_async().await >= 1,
+ "unknown kid attempted a refresh"
);
+
+ // The known kid still validates from the retained last-known-good set.
+ cache
+ .key_for("kid-1")
+ .await
+ .expect("known kid still served despite the outage");
+ }
+
+ #[tokio::test]
+ async fn repeated_unknown_kid_probes_within_negative_ttl_fetch_at_most_once() {
+ // Pins: F12 — an unknown kid confirmed absent by a successful refresh is
+ // negative-cached, so repeated probes do not refetch even once the
+ // cooldown has elapsed between them.
+ let server = MockServer::start_async().await;
+ let mock = server
+ .mock_async(|when, then| {
+ when.method(GET).path("/jwks.json");
+ then.status(200).json_body(jwks_body(&["kid-1"]));
+ })
+ .await;
+ let cache = cache_with(
+ server.url("/jwks.json"),
+ Duration::from_secs(3600),
+ Duration::from_millis(1),
+ Duration::from_secs(60),
+ );
+
+ cache.key_for("kid-1").await.expect("prime the cache");
+ assert_eq!(mock.hits_async().await, 1);
+ tokio::time::sleep(Duration::from_millis(5)).await; // exit cooldown
+
+ // First probe of the unknown kid refreshes once and confirms it absent.
+ assert!(matches!(
+ cache.key_for("kid-x").await,
+ Err(JwksError::UnknownKid(_))
+ ));
assert_eq!(
mock.hits_async().await,
2,
- "unknown kid forces a whole-JWKS refresh"
+ "first unknown probe refreshes once"
+ );
+
+ // Repeated probes, each after the cooldown has elapsed, are served from
+ // the negative cache without any further fetch.
+ for _ in 0..3 {
+ tokio::time::sleep(Duration::from_millis(3)).await;
+ assert!(matches!(
+ cache.key_for("kid-x").await,
+ Err(JwksError::UnknownKid(_))
+ ));
+ }
+ assert_eq!(
+ mock.hits_async().await,
+ 2,
+ "repeated unknown probes are served from the negative cache"
+ );
+ }
+
+ #[tokio::test]
+ async fn stale_key_set_serves_last_known_good_when_refresh_fails() {
+ // Pins: F12 — once the set is past its TTL and the provider is down, a
+ // known kid is still served from the retained last-known-good set rather
+ // than failing authentication.
+ let server = MockServer::start_async().await;
+ let ok = server
+ .mock_async(|when, then| {
+ when.method(GET).path("/jwks.json");
+ then.status(200).json_body(jwks_body(&["kid-1"]));
+ })
+ .await;
+ let cache = cache_with(
+ server.url("/jwks.json"),
+ Duration::from_millis(20),
+ Duration::from_millis(1),
+ Duration::from_secs(60),
+ );
+
+ cache.key_for("kid-1").await.expect("known kid resolves");
+
+ ok.delete_async().await;
+ let _outage = server
+ .mock_async(|when, then| {
+ when.method(GET).path("/jwks.json");
+ then.status(500);
+ })
+ .await;
+ // Exceed both the freshness TTL and the cooldown so a refresh is attempted.
+ tokio::time::sleep(Duration::from_millis(30)).await;
+
+ cache
+ .key_for("kid-1")
+ .await
+ .expect("stale known kid served from last-known-good during the outage");
+ }
+
+ #[tokio::test]
+ async fn new_key_is_rate_limited_within_cooldown_then_picked_up_after() {
+ // Pins: F12 — a newly published kid is rejected without a fetch while the
+ // cooldown is active (rate limit), then picked up by the first refresh
+ // after the cooldown elapses.
+ let server = MockServer::start_async().await;
+ // Precompute both bodies so no slow RSA work runs inside the cooldown window.
+ let body_one = jwks_body(&["kid-1"]);
+ let body_both = jwks_body(&["kid-1", "kid-2"]);
+ let first = server
+ .mock_async(|when, then| {
+ when.method(GET).path("/jwks.json");
+ then.status(200).json_body(body_one);
+ })
+ .await;
+ let cache = cache_with(
+ server.url("/jwks.json"),
+ Duration::from_secs(3600),
+ Duration::from_millis(300),
+ Duration::from_secs(60),
+ );
+
+ cache.key_for("kid-1").await.expect("known kid resolves");
+ assert_eq!(first.hits_async().await, 1);
+
+ // Publish the new key.
+ first.delete_async().await;
+ let second = server
+ .mock_async(|when, then| {
+ when.method(GET).path("/jwks.json");
+ then.status(200).json_body(body_both);
+ })
+ .await;
+
+ // Within the cooldown, the new kid is rejected as an invalid credential
+ // with no refetch.
+ assert!(matches!(
+ cache.key_for("kid-2").await,
+ Err(JwksError::UnknownKid(_))
+ ));
+ assert_eq!(
+ second.hits_async().await,
+ 0,
+ "no refetch within the cooldown window"
);
+
+ // After the cooldown elapses, the next probe refreshes and picks it up.
+ tokio::time::sleep(Duration::from_millis(350)).await;
+ cache
+ .key_for("kid-2")
+ .await
+ .expect("new key picked up after the cooldown");
+ assert_eq!(second.hits_async().await, 1);
}
}
diff --git a/crates/moa-auth/auth0/src/oidc_provider.rs b/crates/moa-auth/auth0/src/oidc_provider.rs
index 046e4d5f2..9f1bcd64a 100644
--- a/crates/moa-auth/auth0/src/oidc_provider.rs
+++ b/crates/moa-auth/auth0/src/oidc_provider.rs
@@ -4,7 +4,7 @@
//! audience, and JWKS URL. Claim names for tenant and identity type are
//! intentionally simple for P1.7 and can be extended by config later.
-use crate::auth0_provider::{parse_identity_type, resolve_or_provision_static};
+use crate::auth0_provider::{map_jwks_error, parse_identity_type, resolve_or_provision_static};
use crate::jwks_cache::JwksCache;
use async_trait::async_trait;
use jsonwebtoken::{Algorithm, Validation, decode, decode_header};
@@ -70,11 +70,7 @@ impl AuthProvider for OidcAuthProvider {
return Err(AuthError::Rejected);
}
let kid = header.kid.ok_or(AuthError::InvalidFormat)?;
- let key = self
- .jwks
- .key_for(&kid)
- .await
- .map_err(|error| AuthError::Unavailable(format!("jwks: {error}")))?;
+ let key = self.jwks.key_for(&kid).await.map_err(map_jwks_error)?;
let mut validation = Validation::new(Algorithm::RS256);
validation.set_issuer(&[&self.issuer]);
diff --git a/crates/moa-auth/authz-schema/src/lib.rs b/crates/moa-auth/authz-schema/src/lib.rs
index 38680013f..8f575a574 100644
--- a/crates/moa-auth/authz-schema/src/lib.rs
+++ b/crates/moa-auth/authz-schema/src/lib.rs
@@ -23,4 +23,4 @@ pub const SCHEMA_V1_JSON: &str = include_str!("schema_v1.json");
/// Increment this on any change that adds, removes, or restructures relations.
/// Outbox idempotency keys include this version so a tuple written under v1
/// cannot be silently re-applied under v2.
-pub const MODEL_VERSION: u32 = 4;
+pub const MODEL_VERSION: u32 = 5;
diff --git a/crates/moa-auth/authz-schema/src/schema_v1.json b/crates/moa-auth/authz-schema/src/schema_v1.json
index a3223cef6..39a98f633 100644
--- a/crates/moa-auth/authz-schema/src/schema_v1.json
+++ b/crates/moa-auth/authz-schema/src/schema_v1.json
@@ -310,13 +310,8 @@
}
},
{
- "tupleToUserset": {
- "tupleset": {
- "relation": "contact"
- },
- "computedUserset": {
- "relation": "contact"
- }
+ "computedUserset": {
+ "relation": "contact"
}
}
]
diff --git a/crates/moa-auth/authz-schema/src/tuple.rs b/crates/moa-auth/authz-schema/src/tuple.rs
index 176ec0817..d6cc05728 100644
--- a/crates/moa-auth/authz-schema/src/tuple.rs
+++ b/crates/moa-auth/authz-schema/src/tuple.rs
@@ -126,23 +126,6 @@ impl TupleKey {
object: self.object_wire(),
}
}
-
- /// Build the deterministic outbox idempotency key for this tuple.
- ///
- /// Format:
- /// `{op}-{object_type}-{object_id}-{relation}-{user_type}-{user_id}-v{model_version}`.
- pub fn idempotency_key(&self, op: TupleOp, model_version: u32) -> String {
- format!(
- "{}-{}-{}-{}-{}-{}-v{}",
- op,
- self.object_type,
- self.object_id,
- self.relation,
- self.user_type,
- self.user_id,
- model_version,
- )
- }
}
/// Serializable OpenFGA tuple key shape.
@@ -271,33 +254,6 @@ mod tests {
assert_eq!(tuple.to_wire().relation, "workspace");
}
- #[test]
- fn idempotency_key_is_deterministic_and_includes_model_version() {
- // Pins: outbox idempotency keys include operation, tuple identity, and model version.
- let user_id = Uuid::parse_str("11111111-1111-1111-1111-111111111111")
- .expect("fixture operator UUID should parse");
- let tenant_id = Uuid::parse_str("22222222-2222-2222-2222-222222222222")
- .expect("fixture tenant UUID should parse");
- let tuple = TupleKey::new(
- UserType::Operator,
- user_id,
- Relation::Operator,
- ObjectType::Tenant,
- tenant_id,
- );
-
- let first = tuple.idempotency_key(TupleOp::Write, 1);
- let second = tuple.idempotency_key(TupleOp::Write, 1);
- assert_eq!(first, second);
- assert_eq!(
- first,
- "write-tenant-22222222-2222-2222-2222-222222222222-operator-operator-11111111-1111-1111-1111-111111111111-v1"
- );
-
- let v2 = tuple.idempotency_key(TupleOp::Write, 2);
- assert_ne!(first, v2);
- }
-
#[test]
fn schema_v1_json_contains_security_contract_types_and_delegation() {
// Pins: the deployed OpenFGA JSON model includes the auth object set and agent delegation.
@@ -380,5 +336,20 @@ mod tests {
"session must define relation {relation}"
);
}
+
+ // Pins: the session's bound contact is a participant via a same-object
+ // computed userset, not a reflexive tuple-to-userset that OpenFGA can
+ // never satisfy. A `tupleToUserset` on `contact->contact` would require
+ // a (contact:X, contact, contact:X) tuple nothing writes.
+ let participant_children = session_relations["participant"]["union"]["child"]
+ .as_array()
+ .expect("participant must be a union of children");
+ assert!(
+ participant_children.iter().any(|child| {
+ child["computedUserset"]["relation"] == "contact"
+ && child.get("tupleToUserset").is_none()
+ }),
+ "participant must grant the session contact via a same-object computed userset"
+ );
}
}
diff --git a/crates/moa-auth/authz/Cargo.toml b/crates/moa-auth/authz/Cargo.toml
index d8f080bc0..0b909eb35 100644
--- a/crates/moa-auth/authz/Cargo.toml
+++ b/crates/moa-auth/authz/Cargo.toml
@@ -7,6 +7,7 @@ publish = false
[dependencies]
async-trait.workspace = true
+metrics.workspace = true
moa-authz-schema.workspace = true
moa-core.workspace = true
moa-ocsf.workspace = true
diff --git a/crates/moa-auth/authz/src/outbox.rs b/crates/moa-auth/authz/src/outbox.rs
index 7586af7bc..9a4f22cb3 100644
--- a/crates/moa-auth/authz/src/outbox.rs
+++ b/crates/moa-auth/authz/src/outbox.rs
@@ -5,11 +5,16 @@ use moa_authz_schema::{MODEL_VERSION, TupleKey, TupleOp};
use sqlx::PgExecutor;
use uuid::Uuid;
-/// Enqueue a tuple operation into `authz_outbox`.
+/// Enqueue the desired state of a tuple into `authz_outbox`.
///
/// Callers should execute this inside the same Postgres transaction as the
-/// state mutation that requires the tuple. Enqueue is idempotent: if the
-/// deterministic key already exists, the existing row is left unchanged.
+/// state mutation that requires the tuple. Enqueue models the *latest desired
+/// state* of the tuple identity `(user, relation, object, model_version)`: the
+/// single row for that identity is upserted to `op`, its generation is bumped,
+/// and it is reset to `pending`. Re-enqueuing the desired op that a row already
+/// carries in a non-terminal state is a no-op, but any change of desired op —
+/// or re-enqueuing a dead-lettered tuple — reactivates the row so the newest
+/// intent is applied. This lets `write -> delete -> write` converge to `write`.
pub async fn enqueue<'executor, Executor>(
exec: Executor,
op: TupleOp,
@@ -19,11 +24,9 @@ pub async fn enqueue<'executor, Executor>(
where
Executor: PgExecutor<'executor>,
{
- let idempotency_key = tuple.idempotency_key(op, MODEL_VERSION);
- insert_outbox(
+ upsert_desired_state(
exec,
op,
- &idempotency_key,
&tuple.user_wire(),
&tuple.relation.to_string(),
&tuple.object_wire(),
@@ -32,11 +35,12 @@ where
.await
}
-/// Enqueue a tuple operation using OpenFGA wire strings directly.
+/// Enqueue the desired state of a tuple using OpenFGA wire strings directly.
///
/// This is used for parent-edge tuples whose subject is another object, such
/// as `tenant: tenant session:`, which cannot be represented by
-/// the typed subject enum in [`TupleKey`].
+/// the typed subject enum in [`TupleKey`]. It shares the desired-state upsert
+/// semantics of [`enqueue`].
pub async fn enqueue_raw<'executor, Executor>(
exec: Executor,
op: TupleOp,
@@ -48,27 +52,20 @@ pub async fn enqueue_raw<'executor, Executor>(
where
Executor: PgExecutor<'executor>,
{
- let idempotency_key = format!("{op}-{object_wire}-{relation}-{user_wire}-v{MODEL_VERSION}");
- insert_outbox(
- exec,
- op,
- &idempotency_key,
- user_wire,
- relation,
- object_wire,
- tenant_id,
- )
- .await
+ upsert_desired_state(exec, op, user_wire, relation, object_wire, tenant_id).await
}
-/// Insert one `authz_outbox` row, leaving an existing row untouched on conflict.
+/// Upsert one tuple identity to its latest desired operation.
///
/// Shared by [`enqueue`] and [`enqueue_raw`], which differ only in how they
-/// derive the idempotency key and wire strings.
-async fn insert_outbox<'executor, Executor>(
+/// derive the wire strings. On conflict the row is reactivated (op set,
+/// generation incremented, status reset to `pending`, retry state cleared) only
+/// when the desired op differs from the stored op or the stored row is
+/// dead-lettered. A same-op enqueue against a non-terminal row leaves the row —
+/// including any in-flight lease — untouched.
+async fn upsert_desired_state<'executor, Executor>(
exec: Executor,
op: TupleOp,
- idempotency_key: &str,
user_wire: &str,
relation: &str,
object_wire: &str,
@@ -80,12 +77,23 @@ where
sqlx::query(
r#"
INSERT INTO authz_outbox
- (idempotency_key, op, tuple_user, tuple_relation, tuple_object, model_version, tenant_id)
- VALUES ($1, $2, $3, $4, $5, $6, $7)
- ON CONFLICT (idempotency_key) DO NOTHING
+ (op, tuple_user, tuple_relation, tuple_object, model_version, tenant_id,
+ generation, status, attempts, next_attempt_at)
+ VALUES ($1, $2, $3, $4, $5, $6, 1, 'pending', 0, NOW())
+ ON CONFLICT (tuple_user, tuple_relation, tuple_object, model_version) DO UPDATE
+ SET op = EXCLUDED.op,
+ generation = authz_outbox.generation + 1,
+ status = 'pending',
+ attempts = 0,
+ last_error = NULL,
+ lease_token = NULL,
+ lease_expires_at = NULL,
+ next_attempt_at = NOW(),
+ updated_at = NOW()
+ WHERE authz_outbox.op IS DISTINCT FROM EXCLUDED.op
+ OR authz_outbox.status = 'dead_letter'
"#,
)
- .bind(idempotency_key)
.bind(op.to_string())
.bind(user_wire)
.bind(relation)
@@ -97,13 +105,14 @@ where
Ok(())
}
+/// One claimed `authz_outbox` row projected for OpenFGA application.
#[derive(Debug, Clone, sqlx::FromRow)]
pub(crate) struct OutboxRow {
pub id: Uuid,
- pub idempotency_key: String,
pub op: String,
pub tuple_user: String,
pub tuple_relation: String,
pub tuple_object: String,
pub attempts: i32,
+ pub generation: i64,
}
diff --git a/crates/moa-auth/authz/src/poller.rs b/crates/moa-auth/authz/src/poller.rs
index 6eace4cad..d89f0927c 100644
--- a/crates/moa-auth/authz/src/poller.rs
+++ b/crates/moa-auth/authz/src/poller.rs
@@ -91,6 +91,9 @@ impl OutboxPoller {
/// Run one drain pass and return the number of rows applied successfully.
pub async fn tick(&self) -> Result {
+ // Sample the backlog age every tick (including idle ticks) so a stalled
+ // drain surfaces as a rising gauge rather than only in logs.
+ self.record_backlog_gauge().await;
let claimed = self.claim_batch().await?;
if claimed.is_empty() {
return Ok(0);
@@ -143,9 +146,9 @@ impl OutboxPoller {
updated_at = NOW()
FROM candidate
WHERE outbox.id = candidate.id
- RETURNING outbox.id, outbox.idempotency_key, outbox.op,
+ RETURNING outbox.id, outbox.op,
outbox.tuple_user, outbox.tuple_relation, outbox.tuple_object,
- outbox.attempts, outbox.lease_token
+ outbox.attempts, outbox.generation, outbox.lease_token
"#,
)
.bind(limit_i64(self.cfg.batch_size))
@@ -178,6 +181,10 @@ impl OutboxPoller {
}
async fn record_success(&self, claim: &ClaimedOutboxRow) -> Result {
+ // Compare-and-set on generation: only mark succeeded if the desired state
+ // this poller applied is still current. A concurrent enqueue that changed
+ // the desired op reset the row to pending and bumped its generation, so
+ // this update matches zero rows and the newer state is applied next tick.
let result = sqlx::query(
r#"
UPDATE authz_outbox
@@ -188,18 +195,21 @@ impl OutboxPoller {
WHERE id=$1
AND status = 'in_flight'
AND lease_token = $2
+ AND generation = $3
"#,
)
.bind(claim.row.id)
.bind(claim.lease_token)
+ .bind(claim.row.generation)
.execute(&self.pool)
.await?;
let completed = result.rows_affected() == 1;
if !completed {
tracing::debug!(
id = %claim.row.id,
- key = %claim.row.idempotency_key,
- "outbox row lease was lost before success could be recorded"
+ tuple = %claim.tuple_display(),
+ generation = claim.row.generation,
+ "outbox row desired state changed before success could be recorded"
);
}
Ok(completed)
@@ -212,10 +222,14 @@ impl OutboxPoller {
) -> Result<(), AuthzError> {
let row = &claim.row;
let next_attempts = row.attempts + 1;
+ // Every failure update is fenced on generation as well as the lease, so a
+ // concurrent desired-state change (which reactivated the row at a higher
+ // generation) is never overwritten with stale retry/dead-letter state.
if next_attempts >= self.cfg.max_attempts {
tracing::error!(
id = %row.id,
- key = %row.idempotency_key,
+ tuple = %claim.tuple_display(),
+ generation = row.generation,
attempts = next_attempts,
error = %error,
"outbox row exhausted retries; moving to dead_letter"
@@ -232,21 +246,25 @@ impl OutboxPoller {
WHERE id=$1
AND status = 'in_flight'
AND lease_token = $4
+ AND generation = $5
"#,
)
.bind(row.id)
.bind(next_attempts)
.bind(error.to_string())
.bind(claim.lease_token)
+ .bind(row.generation)
.execute(&self.pool)
.await?;
+ metrics::counter!("moa_authz_outbox_dead_letters_total").increment(1);
return Ok(());
}
let backoff = self.backoff_for(next_attempts);
tracing::warn!(
id = %row.id,
- key = %row.idempotency_key,
+ tuple = %claim.tuple_display(),
+ generation = row.generation,
attempts = next_attempts,
backoff_ms = backoff.as_millis() as u64,
error = %error,
@@ -265,6 +283,7 @@ impl OutboxPoller {
WHERE id=$1
AND status = 'in_flight'
AND lease_token = $5
+ AND generation = $6
"#,
)
.bind(row.id)
@@ -272,11 +291,41 @@ impl OutboxPoller {
.bind(error.to_string())
.bind(duration_millis_string(backoff))
.bind(claim.lease_token)
+ .bind(row.generation)
.execute(&self.pool)
.await?;
+ metrics::counter!("moa_authz_outbox_retries_total").increment(1);
Ok(())
}
+ /// Samples the drain-backlog age gauge: how long the oldest ready-to-apply
+ /// pending row has waited. Best-effort; a sampling error never fails the tick.
+ async fn record_backlog_gauge(&self) {
+ match self.oldest_ready_pending_age_seconds().await {
+ Ok(age) => {
+ metrics::gauge!("moa_authz_outbox_oldest_pending_age_seconds").set(age);
+ }
+ Err(error) => {
+ tracing::debug!(error = %error, "failed to sample authz outbox backlog age");
+ }
+ }
+ }
+
+ /// Returns the age in seconds of the oldest pending row whose next attempt is
+ /// due, or `0.0` when nothing is waiting to be applied.
+ async fn oldest_ready_pending_age_seconds(&self) -> Result {
+ let age: Option = sqlx::query_scalar(
+ r#"
+ SELECT EXTRACT(EPOCH FROM (NOW() - MIN(next_attempt_at)))
+ FROM authz_outbox
+ WHERE status = 'pending' AND next_attempt_at <= NOW()
+ "#,
+ )
+ .fetch_one(&self.pool)
+ .await?;
+ Ok(age.unwrap_or(0.0).max(0.0))
+ }
+
fn backoff_for(&self, attempt: i32) -> Duration {
let pow = (attempt as u32).saturating_sub(1).min(20);
let multiplier = 1u64 << pow;
@@ -288,12 +337,12 @@ impl OutboxPoller {
#[derive(Debug, Clone, sqlx::FromRow)]
struct ClaimedOutboxRecord {
id: Uuid,
- idempotency_key: String,
op: String,
tuple_user: String,
tuple_relation: String,
tuple_object: String,
attempts: i32,
+ generation: i64,
lease_token: Uuid,
}
@@ -301,12 +350,12 @@ impl ClaimedOutboxRecord {
fn row(&self) -> OutboxRow {
OutboxRow {
id: self.id,
- idempotency_key: self.idempotency_key.clone(),
op: self.op.clone(),
tuple_user: self.tuple_user.clone(),
tuple_relation: self.tuple_relation.clone(),
tuple_object: self.tuple_object.clone(),
attempts: self.attempts,
+ generation: self.generation,
}
}
}
@@ -326,6 +375,16 @@ struct ClaimedOutboxRow {
lease_token: Uuid,
}
+impl ClaimedOutboxRow {
+ /// Render the tuple identity for structured logs, e.g. `write user rel object`.
+ fn tuple_display(&self) -> String {
+ format!(
+ "{} {} {} {}",
+ self.row.op, self.row.tuple_user, self.row.tuple_relation, self.row.tuple_object
+ )
+ }
+}
+
/// Handle for cleanly shutting down a spawned outbox poller.
pub struct PollerHandle {
shutdown: Option>,
diff --git a/crates/moa-auth/authz/tests/authz_db/authz_poller_db.rs b/crates/moa-auth/authz/tests/authz_db/authz_poller_db.rs
index 671b2551b..ba127f434 100644
--- a/crates/moa-auth/authz/tests/authz_db/authz_poller_db.rs
+++ b/crates/moa-auth/authz/tests/authz_db/authz_poller_db.rs
@@ -174,16 +174,15 @@ async fn insert_outbox_row(
let sql = format!(
r#"
INSERT INTO authz_outbox
- (id, idempotency_key, op, tuple_user, tuple_relation, tuple_object,
+ (id, op, tuple_user, tuple_relation, tuple_object,
model_version, status, next_attempt_at, lease_token, lease_expires_at)
VALUES
- ($1, $2, 'write', $3, 'operator', $4, 1, $5,
- NOW() - INTERVAL '1 minute', $6, {lease_expires_at_sql})
+ ($1, 'write', $2, 'operator', $3, 1, $4,
+ NOW() - INTERVAL '1 minute', $5, {lease_expires_at_sql})
"#
);
sqlx::query(&sql)
.bind(row_id)
- .bind(format!("test-key-{row_id}"))
.bind(format!("operator:{}", Uuid::new_v4()))
.bind(format!("tenant:{}", Uuid::new_v4()))
.bind(status)
diff --git a/crates/moa-auth/authz/tests/authz_db/outbox_basic_db.rs b/crates/moa-auth/authz/tests/authz_db/outbox_basic_db.rs
index 583a888a2..3899282c7 100644
--- a/crates/moa-auth/authz/tests/authz_db/outbox_basic_db.rs
+++ b/crates/moa-auth/authz/tests/authz_db/outbox_basic_db.rs
@@ -1,7 +1,8 @@
-//! Outbox enqueue and retry behavior against Postgres.
+//! Outbox desired-state, convergence, and retry behavior against Postgres.
+use httpmock::{Method::POST, MockServer};
use moa_authz::{FgaClient, FgaConfig, OutboxPoller, PollerConfig, enqueue};
-use moa_authz_schema::{ObjectType, Relation, TupleKey, TupleOp, UserType};
+use moa_authz_schema::{MODEL_VERSION, ObjectType, Relation, TupleKey, TupleOp, UserType};
use sqlx::{PgPool, postgres::PgPoolOptions};
use std::time::Duration;
use uuid::Uuid;
@@ -37,58 +38,100 @@ fn quote_identifier(identifier: &str) -> String {
format!("\"{}\"", identifier.replace('"', "\"\""))
}
-#[tokio::test]
-async fn outbox_basic_enqueue_is_idempotent_on_same_key() {
- // Pins: repeated enqueue of the same tuple operation creates exactly one outbox row.
- let pool = test_pool().await;
- let user_id = Uuid::new_v4();
+fn test_tuple() -> (TupleKey, Uuid) {
let tenant_id = Uuid::new_v4();
let tuple = TupleKey::new(
UserType::Operator,
- user_id,
+ Uuid::new_v4(),
Relation::Operator,
ObjectType::Tenant,
tenant_id,
);
+ (tuple, tenant_id)
+}
- enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
- .await
- .expect("first enqueue should succeed");
- enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
- .await
- .expect("second enqueue should be idempotent");
- enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
- .await
- .expect("third enqueue should be idempotent");
+/// The single desired-state row for a tuple identity: `(op, generation, status)`.
+async fn desired_state(pool: &PgPool, tuple: &TupleKey) -> Option<(String, i64, String)> {
+ sqlx::query_as(
+ "SELECT op, generation, status FROM authz_outbox
+ WHERE tuple_user=$1 AND tuple_relation=$2 AND tuple_object=$3",
+ )
+ .bind(tuple.user_wire())
+ .bind(tuple.relation.to_string())
+ .bind(tuple.object_wire())
+ .fetch_optional(pool)
+ .await
+ .expect("desired-state query should succeed")
+}
- let count: i64 = sqlx::query_scalar(
- "SELECT COUNT(*) FROM authz_outbox WHERE tuple_user=$1 AND tuple_relation=$2 AND tuple_object=$3",
+async fn row_count(pool: &PgPool, tuple: &TupleKey) -> i64 {
+ sqlx::query_scalar(
+ "SELECT COUNT(*) FROM authz_outbox
+ WHERE tuple_user=$1 AND tuple_relation=$2 AND tuple_object=$3",
)
.bind(tuple.user_wire())
.bind(tuple.relation.to_string())
.bind(tuple.object_wire())
- .fetch_one(&pool)
+ .fetch_one(pool)
.await
- .expect("count query should succeed");
+ .expect("count query should succeed")
+}
+
+fn failing_poller(pool: PgPool, max_attempts: i32) -> OutboxPoller {
+ // 127.0.0.1:9 (discard) makes every FGA apply fail, driving the retry path.
+ poller_with_url(pool, "http://127.0.0.1:9".to_string(), max_attempts)
+}
+
+fn poller_with_url(pool: PgPool, url: String, max_attempts: i32) -> OutboxPoller {
+ let client = FgaClient::new(FgaConfig {
+ url,
+ preshared_key: "test".to_string(),
+ store_id: "store".to_string(),
+ model_id: "model".to_string(),
+ timeout_ms: 5_000,
+ })
+ .expect("client config should be valid");
+ OutboxPoller::new(
+ pool,
+ client,
+ PollerConfig {
+ batch_size: 8,
+ poll_interval: Duration::from_millis(10),
+ max_attempts,
+ backoff_base: Duration::from_millis(1),
+ backoff_cap: Duration::from_millis(1),
+ lease_duration: PollerConfig::default().lease_duration,
+ },
+ )
+}
+
+#[tokio::test]
+async fn outbox_same_op_enqueue_is_idempotent_and_holds_generation_db() {
+ // Pins: repeated enqueue of the same desired op keeps exactly one row at its
+ // initial generation, so unchanged intent never busts the row or resets retry.
+ let pool = test_pool().await;
+ let (tuple, tenant_id) = test_tuple();
+
+ for _ in 0..3 {
+ enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
+ .await
+ .expect("same-op enqueue should be idempotent");
+ }
+
+ assert_eq!(row_count(&pool, &tuple).await, 1);
assert_eq!(
- count, 1,
- "three enqueues of the same key must produce exactly one row"
+ desired_state(&pool, &tuple).await,
+ Some(("write".to_string(), 1, "pending".to_string()))
);
}
#[tokio::test]
-async fn outbox_basic_enqueue_separates_write_and_delete() {
- // Pins: write and delete use distinct idempotency keys for the same tuple.
+async fn outbox_alternating_ops_collapse_to_latest_desired_state_db() {
+ // Pins: write then delete for one tuple identity yields a single row whose
+ // desired op is the latest (delete) with a bumped generation — not two rows,
+ // and not a write that permanently owns the identity.
let pool = test_pool().await;
- let user_id = Uuid::new_v4();
- let tenant_id = Uuid::new_v4();
- let tuple = TupleKey::new(
- UserType::Operator,
- user_id,
- Relation::Operator,
- ObjectType::Tenant,
- tenant_id,
- );
+ let (tuple, tenant_id) = test_tuple();
enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
.await
@@ -97,69 +140,204 @@ async fn outbox_basic_enqueue_separates_write_and_delete() {
.await
.expect("delete enqueue should succeed");
- let count: i64 = sqlx::query_scalar(
- "SELECT COUNT(*) FROM authz_outbox WHERE tuple_user=$1 AND tuple_relation=$2 AND tuple_object=$3",
+ assert_eq!(row_count(&pool, &tuple).await, 1);
+ assert_eq!(
+ desired_state(&pool, &tuple).await,
+ Some(("delete".to_string(), 2, "pending".to_string()))
+ );
+}
+
+#[tokio::test]
+async fn outbox_grant_revoke_grant_converges_to_write_db() {
+ // Pins: the F03 regression — grant, revoke, then re-grant must leave the tuple
+ // in the granted (write) desired state, not suppressed by the first write.
+ let pool = test_pool().await;
+ let (tuple, tenant_id) = test_tuple();
+
+ enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
+ .await
+ .expect("grant should succeed");
+ enqueue(&pool, TupleOp::Delete, &tuple, Some(tenant_id))
+ .await
+ .expect("revoke should succeed");
+ enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
+ .await
+ .expect("re-grant should succeed");
+
+ assert_eq!(row_count(&pool, &tuple).await, 1);
+ assert_eq!(
+ desired_state(&pool, &tuple).await,
+ Some(("write".to_string(), 3, "pending".to_string()))
+ );
+}
+
+#[tokio::test]
+async fn outbox_revoke_grant_revoke_converges_to_delete_db() {
+ // Pins: the inverse sequence — revoke, grant, then re-revoke must converge to
+ // the revoked (delete) desired state.
+ let pool = test_pool().await;
+ let (tuple, tenant_id) = test_tuple();
+
+ enqueue(&pool, TupleOp::Delete, &tuple, Some(tenant_id))
+ .await
+ .expect("revoke should succeed");
+ enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
+ .await
+ .expect("grant should succeed");
+ enqueue(&pool, TupleOp::Delete, &tuple, Some(tenant_id))
+ .await
+ .expect("re-revoke should succeed");
+
+ assert_eq!(row_count(&pool, &tuple).await, 1);
+ assert_eq!(
+ desired_state(&pool, &tuple).await,
+ Some(("delete".to_string(), 3, "pending".to_string()))
+ );
+}
+
+#[tokio::test]
+async fn outbox_poller_applies_latest_desired_state_after_change_db() {
+ // Pins: when the desired op changes after the row was picked up (simulated by
+ // an in_flight row plus an enqueue of the opposite op), the poller applies the
+ // NEW op and converges the row to succeeded at the bumped generation.
+ let pool = test_pool().await;
+ let (tuple, tenant_id) = test_tuple();
+
+ // Simulate a poller that claimed a write (in_flight, generation 1). The
+ // model_version must match what `enqueue` uses, since it is part of the
+ // tuple identity the reactivating upsert conflicts on.
+ sqlx::query(
+ "INSERT INTO authz_outbox
+ (op, tuple_user, tuple_relation, tuple_object, model_version, tenant_id,
+ generation, status, lease_token, lease_expires_at, next_attempt_at)
+ VALUES ('write', $1, $2, $3, $4, $5, 1, 'in_flight', $6,
+ NOW() + INTERVAL '5 minutes', NOW())",
)
.bind(tuple.user_wire())
.bind(tuple.relation.to_string())
.bind(tuple.object_wire())
- .fetch_one(&pool)
+ .bind(MODEL_VERSION as i32)
+ .bind(tenant_id)
+ .bind(Uuid::new_v4())
+ .execute(&pool)
.await
- .expect("count query should succeed");
+ .expect("seed in_flight row should succeed");
+
+ // A concurrent revoke changes the desired op: the row reactivates as a
+ // pending delete at generation 2, releasing the stale lease.
+ enqueue(&pool, TupleOp::Delete, &tuple, Some(tenant_id))
+ .await
+ .expect("revoke should reactivate the row");
assert_eq!(
- count, 2,
- "write and delete must produce separate outbox rows"
+ desired_state(&pool, &tuple).await,
+ Some(("delete".to_string(), 2, "pending".to_string()))
+ );
+
+ let server = MockServer::start();
+ let delete = server.mock(|when, then| {
+ when.method(POST)
+ .path("/stores/store/write")
+ .body_contains("deletes");
+ then.status(200).json_body(serde_json::json!({}));
+ });
+ let poller = poller_with_url(pool.clone(), server.base_url(), 4);
+
+ let drained = poller.tick().await.expect("poller tick should complete");
+
+ assert_eq!(drained, 1, "the latest desired op should be applied once");
+ delete.assert_hits(1);
+ assert_eq!(
+ desired_state(&pool, &tuple).await,
+ Some(("delete".to_string(), 2, "succeeded".to_string())),
+ "poller must converge the row to the latest desired op"
);
}
#[tokio::test]
-async fn outbox_basic_failed_row_moves_to_dead_letter_at_max_attempts() {
- // Pins: a non-retryable poller batch stops at dead_letter after max_attempts.
+async fn outbox_dead_letter_reactivates_on_new_desired_state_db() {
+ // Pins: a dead-lettered tuple still accepts a new desired op and returns to
+ // pending, so a redrive after terminal failure is not permanently suppressed.
let pool = test_pool().await;
- let tenant_id = Uuid::new_v4();
- let user_id = Uuid::new_v4();
- let tuple = TupleKey::new(
- UserType::Operator,
- user_id,
- Relation::Operator,
- ObjectType::Tenant,
- tenant_id,
+ let (tuple, tenant_id) = test_tuple();
+
+ enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
+ .await
+ .expect("initial grant should succeed");
+
+ // Exhaust retries against an unreachable FGA to reach dead_letter.
+ let poller = failing_poller(pool.clone(), 1);
+ poller.tick().await.expect("poller tick should complete");
+ assert_eq!(
+ desired_state(&pool, &tuple).await,
+ Some(("write".to_string(), 1, "dead_letter".to_string()))
);
+
+ // A new desired op (revoke) must reactivate the dead-lettered identity.
+ enqueue(&pool, TupleOp::Delete, &tuple, Some(tenant_id))
+ .await
+ .expect("revoke should reactivate a dead-lettered tuple");
+
+ assert_eq!(row_count(&pool, &tuple).await, 1);
+ assert_eq!(
+ desired_state(&pool, &tuple).await,
+ Some(("delete".to_string(), 2, "pending".to_string()))
+ );
+}
+
+#[tokio::test]
+async fn outbox_dead_letter_reactivates_on_same_op_redrive_db() {
+ // Pins: re-enqueuing the SAME op that dead-lettered still reactivates the row,
+ // so a transient outage that exhausted retries can be redriven.
+ let pool = test_pool().await;
+ let (tuple, tenant_id) = test_tuple();
+
enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
.await
- .expect("enqueue should succeed");
+ .expect("initial grant should succeed");
+ failing_poller(pool.clone(), 1)
+ .tick()
+ .await
+ .expect("poller tick should complete");
+ assert_eq!(
+ desired_state(&pool, &tuple).await,
+ Some(("write".to_string(), 1, "dead_letter".to_string()))
+ );
- let client = FgaClient::new(FgaConfig {
- url: "http://127.0.0.1:9".to_string(),
- preshared_key: "test".to_string(),
- store_id: "store".to_string(),
- model_id: "model".to_string(),
- timeout_ms: 200,
- })
- .expect("client config should be valid");
- let poller = OutboxPoller::new(
- pool.clone(),
- client,
- PollerConfig {
- batch_size: 8,
- poll_interval: Duration::from_millis(10),
- max_attempts: 1,
- backoff_base: Duration::from_millis(1),
- backoff_cap: Duration::from_millis(1),
- lease_duration: PollerConfig::default().lease_duration,
- },
+ enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
+ .await
+ .expect("same-op redrive should reactivate the row");
+
+ assert_eq!(
+ desired_state(&pool, &tuple).await,
+ Some(("write".to_string(), 2, "pending".to_string()))
);
+}
- let drained = poller.tick().await.expect("poller tick should complete");
+#[tokio::test]
+async fn outbox_failed_row_moves_to_dead_letter_at_max_attempts_db() {
+ // Pins: a non-retryable poller batch stops at dead_letter after max_attempts.
+ let pool = test_pool().await;
+ let (tuple, tenant_id) = test_tuple();
+ enqueue(&pool, TupleOp::Write, &tuple, Some(tenant_id))
+ .await
+ .expect("enqueue should succeed");
+
+ let drained = failing_poller(pool.clone(), 1)
+ .tick()
+ .await
+ .expect("poller tick should complete");
assert_eq!(drained, 0, "failed rows are not counted as drained");
- let (status, attempts): (String, i32) = sqlx::query_as(
- "SELECT status, attempts FROM authz_outbox WHERE tuple_object=$1 ORDER BY created_at DESC LIMIT 1",
+ assert_eq!(
+ desired_state(&pool, &tuple).await,
+ Some(("write".to_string(), 1, "dead_letter".to_string()))
+ );
+ let attempts: i32 = sqlx::query_scalar(
+ "SELECT attempts FROM authz_outbox WHERE tuple_object=$1 ORDER BY created_at DESC LIMIT 1",
)
.bind(tuple.object_wire())
.fetch_one(&pool)
.await
- .expect("status query should succeed");
- assert_eq!(status, "dead_letter");
+ .expect("attempts query should succeed");
assert_eq!(attempts, 1);
}
diff --git a/crates/moa-auth/providers/src/api_keys.rs b/crates/moa-auth/providers/src/api_keys.rs
index 29ddb9a9d..867faada2 100644
--- a/crates/moa-auth/providers/src/api_keys.rs
+++ b/crates/moa-auth/providers/src/api_keys.rs
@@ -292,7 +292,12 @@ where
{
let key = generate(new.env);
let prefix = prefix_of(key.expose_secret())?;
- let hash = hash_key(key.expose_secret())?;
+ // Argon2 hashing is CPU-heavy; run it off the async runtime so key creation
+ // does not stall other tasks on the worker thread (mirrors the verify path).
+ let key_secret = key.expose_secret().to_string();
+ let hash = tokio::task::spawn_blocking(move || hash_key(&key_secret))
+ .await
+ .map_err(|error| ApiKeyError::Hash(format!("hash task failed: {error}")))??;
let id = Uuid::new_v4();
let (owner_user_id, owner_agent_id) = match new.owner {
KeyOwner::User(user_id) => (Some(user_id), None),
diff --git a/crates/moa-auth/providers/src/contact_tokens.rs b/crates/moa-auth/providers/src/contact_tokens.rs
index a2a1a0394..05d0d5b25 100644
--- a/crates/moa-auth/providers/src/contact_tokens.rs
+++ b/crates/moa-auth/providers/src/contact_tokens.rs
@@ -85,6 +85,11 @@ impl ContactTokenVerifier {
validation.set_required_spec_claims(&["exp", "nbf", "iat", "sub", "iss", "aud", "jti"]);
validation.leeway = 30;
+ // Verification is stateless: `jti` is required for presence but is not
+ // checked against a revocation denylist, so the token's TTL is the only
+ // revocation window. A `jti` denylist (checked here) would let operators
+ // revoke a token before it expires — follow-up, not implemented.
+
decode::(token, &self.decoding_key, &validation)
.map(|data| data.claims)
.map_err(|error| match error.kind() {
diff --git a/crates/moa-auth/providers/src/user_sessions.rs b/crates/moa-auth/providers/src/user_sessions.rs
index 9df3cc7fd..67dc9114a 100644
--- a/crates/moa-auth/providers/src/user_sessions.rs
+++ b/crates/moa-auth/providers/src/user_sessions.rs
@@ -83,7 +83,12 @@ where
{
let token = generate();
let prefix = prefix_of(&token)?;
- let hash = hash_token(&token)?;
+ // Argon2 hashing is CPU-heavy; run it off the async runtime so session-token
+ // creation does not stall other tasks on the worker thread (mirrors verify).
+ let token_for_hash = token.clone();
+ let hash = tokio::task::spawn_blocking(move || hash_token(&token_for_hash))
+ .await
+ .map_err(|error| UserSessionTokenError::Hash(format!("hash task failed: {error}")))??;
let id = Uuid::new_v4();
sqlx::query(
r#"
diff --git a/crates/moa-brain/assets/ner-gazetteer.txt b/crates/moa-brain/assets/ner-gazetteer.txt
index 82947328c..de85660af 100644
--- a/crates/moa-brain/assets/ner-gazetteer.txt
+++ b/crates/moa-brain/assets/ner-gazetteer.txt
@@ -4,7 +4,7 @@ postgres
postgresql
restate
cohere
-fly.io
+railway
aws
pgvector
memory graph
diff --git a/crates/moa-brain/examples/replay_corpus.rs b/crates/moa-brain/examples/replay_corpus.rs
index 598749359..d4d55365f 100644
--- a/crates/moa-brain/examples/replay_corpus.rs
+++ b/crates/moa-brain/examples/replay_corpus.rs
@@ -34,7 +34,7 @@ use moa_session::testing;
use serde::Serialize;
use serde_json::json;
-/// Per-turn compile budget, matching `CompactionConfig::max_input_tokens_per_turn`.
+/// Per-turn compile budget used by this offline replay-cost example.
const TURN_BUDGET: usize = 160_000;
/// Reference input price for uncached tokens, dollars per MTok.
const FRESH_PER_MTOK: f64 = 3.0;
diff --git a/crates/moa-brain/src/pipeline/memory.rs b/crates/moa-brain/src/pipeline/memory.rs
index 7e936c925..0bab6ec05 100644
--- a/crates/moa-brain/src/pipeline/memory.rs
+++ b/crates/moa-brain/src/pipeline/memory.rs
@@ -39,7 +39,7 @@ mod lineage;
mod rendering;
use lineage::lineage_context_from_context;
-use rendering::{render_memory_context, render_memory_context_with_budget};
+use rendering::render_memory_context_with_budget;
const MEMORY_BUDGET_DIVISOR: usize = 5;
const GRAPH_MEMORY_RESULTS: usize = 4;
@@ -232,8 +232,11 @@ pub trait ScopedRetrievalRuntimeFactory: Send + Sync {
) -> Result;
}
-#[derive(Debug, Default)]
-struct PostgresScopedRetrievalRuntimeFactory;
+struct PostgresScopedRetrievalRuntimeFactory {
+ /// Shared bounded enrichment worker handle, cloned into each scoped
+ /// [`HybridRetriever`]. `None` when constructed outside a Tokio runtime.
+ enrichment: Option,
+}
#[async_trait]
impl ScopedRetrievalRuntimeFactory for PostgresScopedRetrievalRuntimeFactory {
@@ -264,7 +267,8 @@ impl ScopedRetrievalRuntimeFactory for PostgresScopedRetrievalRuntimeFactory {
graph.clone(),
pgvector_source,
)
- .with_assume_app_role(assume_app_role),
+ .with_assume_app_role(assume_app_role)
+ .with_enrichment(self.enrichment.clone()),
);
let cached: Arc = if assume_app_role {
Arc::new(crate::retrieval::CachedHybridRetriever::new_for_app_role(
@@ -295,6 +299,12 @@ impl GraphMemoryRetriever {
pool: PgPool,
embedder: Option>,
) -> Self {
+ // F20: spawn the single shared bounded enrichment worker when a Tokio
+ // runtime is available (production and async tests). Sync callers
+ // (eval/unit setup) get no worker and skip best-effort enrichment.
+ let enrichment = tokio::runtime::Handle::try_current()
+ .ok()
+ .map(|_| crate::retrieval::enrichment::spawn_enrichment_worker(pool.clone()));
Self {
pool,
embedder,
@@ -304,7 +314,7 @@ impl GraphMemoryRetriever {
result_limit: GRAPH_MEMORY_RESULTS,
planner: crate::planning::QueryPlanner::new(),
scoped_runtimes: build_scoped_runtime_cache(),
- runtime_factory: Arc::new(PostgresScopedRetrievalRuntimeFactory),
+ runtime_factory: Arc::new(PostgresScopedRetrievalRuntimeFactory { enrichment }),
}
}
@@ -464,7 +474,25 @@ impl GraphMemoryRetriever {
let needs_backend = probes.iter().any(|probe| probe.cached_hits.is_none());
let query_embedding = if needs_backend {
match self.embedder.as_deref() {
- Some(embedder) => embed_query(embedder, query_str).await?,
+ // A query-embedding failure must degrade to lexical-only
+ // retrieval rather than abort the turn: an empty embedding makes
+ // the vector leg return nothing while the lexical leg still runs.
+ Some(embedder) => match embed_query(embedder, query_str).await {
+ Ok(embedding) => embedding,
+ Err(error) => {
+ tracing::warn!(
+ error = %error,
+ "query embedding failed; degrading to lexical-only retrieval"
+ );
+ metrics::counter!(
+ "moa_retrieval_leg_degraded_total",
+ "leg" => "embedding",
+ "reason" => "error",
+ )
+ .increment(1);
+ Vec::new()
+ }
+ },
None => Vec::new(),
}
} else {
@@ -526,12 +554,17 @@ impl GraphMemoryRetriever {
let result_limit = requested_result_limit;
- let mut fused = Vec::new();
- for sub_query in sub_queries {
- fused.extend(
+ // F23: run the (≤2) decomposed sub-queries concurrently instead of
+ // serially. `try_join_all` preserves the decomposition input order in its
+ // output, so fusion stays deterministic before `dedupe_and_rank_hits`.
+ let sub_results =
+ try_join_all(sub_queries.into_iter().map(|sub_query| {
self.retrieve_hits(ctx, sub_query, policy, requested_result_limit)
- .await?,
- );
+ }))
+ .await?;
+ let mut fused = Vec::new();
+ for hits in sub_results {
+ fused.extend(hits);
}
Ok(dedupe_and_rank_hits(fused, result_limit))
}
@@ -801,8 +834,38 @@ impl ContextProcessor for GraphMemoryRetriever {
let tokens_before = ctx.token_count;
let memory_budget =
(ctx.token_budget / MEMORY_BUDGET_DIVISOR).max(MIN_PAGE_EXCERPT_TOKENS);
- let per_hit_budget = (memory_budget / hits.len().max(1)).max(MIN_PAGE_EXCERPT_TOKENS);
- let rendered = render_memory_context(&hits, per_hit_budget);
+ // F23: cap the whole injected block by the memory budget rather than a
+ // per-hit floor (which let N hits reach 96·N > memory_budget). Reserve
+ // the outer memory-reminder wrapper so the injected message — not just
+ // the inner rendered section — stays within budget, then render the
+ // largest ranked-hit prefix that fits.
+ let evidence_budget = memory_budget.saturating_sub(memory_reminder_wrapper_tokens());
+ let budgeted = render_memory_context_with_budget(&hits, evidence_budget);
+ let omitted = hits.len().saturating_sub(budgeted.hit_count);
+ if omitted > 0 {
+ tracing::debug!(
+ omitted_hits = omitted,
+ rendered_hits = budgeted.hit_count,
+ evidence_budget,
+ "graph-memory injection omitted ranked hits to fit the memory budget"
+ );
+ metrics::counter!("moa_memory_injection_omitted_hits_total")
+ .increment(omitted as u64);
+ }
+ if budgeted.hit_count == 0 {
+ // The budget could not fit even one hit; inject nothing rather than
+ // an empty reminder wrapper.
+ return Ok(ProcessorOutput {
+ items_excluded: vec!["graph_memory".to_string()],
+ excluded_items: vec![ExcludedItem {
+ item: "graph_memory".to_string(),
+ reason: "memory budget too small to render any ranked hit".to_string(),
+ }],
+ metadata: router_strategy_metadata(strategy_label),
+ ..ProcessorOutput::default()
+ });
+ }
+ let rendered = budgeted.rendered;
let reminder = format!(
"{MEMORY_REMINDER_PREFIX}\n{}\n",
@@ -838,6 +901,13 @@ async fn embed_query(embedder: &dyn EmbeddingProvider, query: &str) -> Result` wrapper added around the rendered
+/// memory section, reserved from the memory budget so the whole injected message
+/// stays within budget.
+fn memory_reminder_wrapper_tokens() -> usize {
+ moa_core::estimate_text_tokens(&format!("{MEMORY_REMINDER_PREFIX}\n\n"))
+}
+
/// Builds the processor-output metadata that records the router's chosen
/// strategy, so evals and tests can observe the routing decision for a turn.
fn router_strategy_metadata(
diff --git a/crates/moa-brain/src/pipeline/memory/rendering.rs b/crates/moa-brain/src/pipeline/memory/rendering.rs
index 1bd030063..d5f121c7a 100644
--- a/crates/moa-brain/src/pipeline/memory/rendering.rs
+++ b/crates/moa-brain/src/pipeline/memory/rendering.rs
@@ -347,7 +347,38 @@ mod tests {
KnowledgeChunkHydration, KnowledgeChunkWindowPart, LegSources, RetrievalHit, SourceTier,
};
- use super::{MATCHED_CLOSE, MATCHED_OPEN, render_memory_context};
+ use super::{
+ MATCHED_CLOSE, MATCHED_OPEN, render_memory_context, render_memory_context_with_budget,
+ };
+
+ #[test]
+ fn budgeted_render_caps_aggregate_tokens_and_omits_excess_hits() {
+ // Pins: F23 — the aggregate renderer caps the whole rendered section at
+ // the token budget and drops ranked hits that do not fit, instead of a
+ // per-hit floor that let N hits reach 96·N tokens well past the budget.
+ let long_excerpt = "escalation ".repeat(60);
+ let hits = (0..8_u128)
+ .map(|index| fact_hit(Uuid::from_u128(index + 1), &long_excerpt))
+ .collect::>();
+ let token_budget = 200;
+
+ let budgeted = render_memory_context_with_budget(&hits, token_budget);
+
+ assert!(budgeted.hit_count >= 1, "at least one hit fits the budget");
+ assert!(
+ budgeted.hit_count < hits.len(),
+ "excess ranked hits are omitted under the aggregate budget"
+ );
+ assert!(
+ moa_core::estimate_text_tokens(&budgeted.rendered.section) <= token_budget,
+ "the rendered section stays within the aggregate token budget"
+ );
+ assert_eq!(
+ budgeted.consumed_tokens,
+ moa_core::estimate_text_tokens(&budgeted.rendered.section),
+ "reported consumed tokens match the rendered section"
+ );
+ }
fn fact_hit(uid: Uuid, summary: &str) -> RetrievalHit {
RetrievalHit {
diff --git a/crates/moa-brain/src/retrieval/enrichment.rs b/crates/moa-brain/src/retrieval/enrichment.rs
new file mode 100644
index 000000000..b86fc198c
--- /dev/null
+++ b/crates/moa-brain/src/retrieval/enrichment.rs
@@ -0,0 +1,459 @@
+//! Bounded, batching background worker for best-effort post-retrieval enrichment.
+//!
+//! After a successful retrieval the hybrid retriever needs two best-effort side
+//! writes — a `last_accessed_at` bump (memory decay input) and sampled retrieval
+//! lineage (quality-score input). Previously each retrieval spawned detached
+//! `tokio` tasks for these against the same Postgres pool that backs sessions,
+//! authz, and graph work, with no queue bound, backpressure, or shutdown drain,
+//! so under load they competed with durable writes and on shutdown were silently
+//! dropped.
+//!
+//! This module funnels both through one bounded, instrumented worker:
+//!
+//! * Enqueue is a non-blocking `try_send`; when the fixed-capacity queue is full
+//! the job is dropped with a metric (drop-on-overflow is correct for
+//! best-effort enrichment — the retrieval path must never backpressure on it).
+//! * Access bumps are coalesced within a batch window: one `UPDATE ... WHERE uid
+//! = ANY($1)` per `(scope, role)` instead of one statement per retrieval.
+//! * The worker drains its channel when all senders drop (shutdown), so queued
+//! work is flushed rather than lost.
+//! * Queue depth, drops, batch sizes, and flush latency are exported as metrics.
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::Duration;
+
+use async_trait::async_trait;
+use chrono::{DateTime, Utc};
+use moa_memory_types::MemoryScope;
+use sqlx::PgPool;
+use tokio::sync::mpsc;
+use tokio::task::JoinHandle;
+use uuid::Uuid;
+
+use crate::retrieval::legs::{bump_last_accessed, write_retrieval_lineage};
+use crate::retrieval::types::{LineageContext, RetrievalLineageHit};
+
+/// Fixed bound on queued enrichment jobs. Drop-on-overflow keeps the retrieval
+/// path from ever backpressuring on best-effort enrichment.
+const DEFAULT_QUEUE_CAPACITY: usize = 1024;
+/// Maximum time a partially filled batch waits before flushing.
+const DEFAULT_BATCH_WINDOW: Duration = Duration::from_millis(50);
+/// Maximum number of jobs coalesced into one flush.
+const DEFAULT_MAX_BATCH: usize = 256;
+
+/// A best-effort enrichment job produced after a successful retrieval.
+enum EnrichmentJob {
+ /// Bump `last_accessed_at` for retrieved nodes in one scope.
+ AccessBump {
+ scope: MemoryScope,
+ uids: Vec,
+ assume_app_role: bool,
+ },
+ /// Write sampled retrieval-lineage rows for one turn.
+ Lineage {
+ scope: MemoryScope,
+ lineage: LineageContext,
+ hits: Vec,
+ retrieved_at: DateTime,
+ assume_app_role: bool,
+ },
+}
+
+/// Applies enrichment writes. Abstracted behind a trait so the worker's
+/// batching, coalescing, and drain logic is unit-testable without a database.
+#[async_trait]
+pub(crate) trait EnrichmentSink: Send + Sync {
+ /// Bumps `last_accessed_at` for `uids` within one scope.
+ async fn bump_access(&self, scope: MemoryScope, uids: Vec, assume_app_role: bool);
+ /// Writes sampled retrieval-lineage rows for one turn.
+ async fn write_lineage(
+ &self,
+ scope: MemoryScope,
+ lineage: LineageContext,
+ hits: Vec,
+ retrieved_at: DateTime,
+ assume_app_role: bool,
+ );
+}
+
+/// Production sink backed by the shared runtime Postgres pool.
+struct PgEnrichmentSink {
+ pool: PgPool,
+}
+
+#[async_trait]
+impl EnrichmentSink for PgEnrichmentSink {
+ async fn bump_access(&self, scope: MemoryScope, uids: Vec, assume_app_role: bool) {
+ if let Err(error) =
+ bump_last_accessed(self.pool.clone(), scope, uids, assume_app_role).await
+ {
+ tracing::debug!(
+ error = %error,
+ "enrichment worker failed to bump graph-memory access timestamps"
+ );
+ }
+ }
+
+ async fn write_lineage(
+ &self,
+ scope: MemoryScope,
+ lineage: LineageContext,
+ hits: Vec,
+ retrieved_at: DateTime,
+ assume_app_role: bool,
+ ) {
+ if let Err(error) = write_retrieval_lineage(
+ self.pool.clone(),
+ scope,
+ lineage,
+ hits,
+ retrieved_at,
+ assume_app_role,
+ )
+ .await
+ {
+ tracing::debug!(
+ error = %error,
+ "enrichment worker failed to write graph-memory retrieval lineage"
+ );
+ }
+ }
+}
+
+/// Non-blocking handle used by the retrieval path to enqueue enrichment.
+#[derive(Clone)]
+pub(crate) struct EnrichmentHandle {
+ tx: mpsc::Sender,
+ dropped: Arc,
+}
+
+impl EnrichmentHandle {
+ /// Enqueues an access-timestamp bump; a no-op when `uids` is empty.
+ pub(crate) fn enqueue_access_bump(
+ &self,
+ scope: MemoryScope,
+ uids: Vec,
+ assume_app_role: bool,
+ ) {
+ if uids.is_empty() {
+ return;
+ }
+ self.try_send(EnrichmentJob::AccessBump {
+ scope,
+ uids,
+ assume_app_role,
+ });
+ }
+
+ /// Enqueues a retrieval-lineage write; a no-op when `hits` is empty.
+ pub(crate) fn enqueue_lineage(
+ &self,
+ scope: MemoryScope,
+ lineage: LineageContext,
+ hits: Vec,
+ retrieved_at: DateTime,
+ assume_app_role: bool,
+ ) {
+ if hits.is_empty() {
+ return;
+ }
+ self.try_send(EnrichmentJob::Lineage {
+ scope,
+ lineage,
+ hits,
+ retrieved_at,
+ assume_app_role,
+ });
+ }
+
+ fn try_send(&self, job: EnrichmentJob) {
+ let depth = self.tx.max_capacity().saturating_sub(self.tx.capacity());
+ metrics::gauge!("moa_retrieval_enrichment_queue_depth").set(depth as f64);
+ match self.tx.try_send(job) {
+ Ok(()) => {}
+ Err(mpsc::error::TrySendError::Full(_)) => {
+ let dropped = self.dropped.fetch_add(1, Ordering::Relaxed) + 1;
+ metrics::counter!("moa_retrieval_enrichment_dropped_total").increment(1);
+ tracing::warn!(
+ dropped,
+ "retrieval enrichment queue full; dropping best-effort job"
+ );
+ }
+ Err(mpsc::error::TrySendError::Closed(_)) => {
+ // The worker has shut down; there is nothing left to enrich.
+ }
+ }
+ }
+
+ #[cfg(test)]
+ fn dropped(&self) -> u64 {
+ self.dropped.load(Ordering::Relaxed)
+ }
+}
+
+/// Spawns the single production enrichment worker over the shared pool.
+///
+/// Must be called from within a Tokio runtime; callers without one skip
+/// enrichment entirely (the handle is optional on the retriever).
+pub(crate) fn spawn_enrichment_worker(pool: PgPool) -> EnrichmentHandle {
+ let (handle, _join) = spawn_with_sink(
+ Arc::new(PgEnrichmentSink { pool }),
+ DEFAULT_QUEUE_CAPACITY,
+ DEFAULT_BATCH_WINDOW,
+ DEFAULT_MAX_BATCH,
+ );
+ handle
+}
+
+/// Spawns a worker over an explicit sink and policy. The returned [`JoinHandle`]
+/// completes once every sender has dropped and the channel has been drained.
+fn spawn_with_sink(
+ sink: Arc,
+ capacity: usize,
+ batch_window: Duration,
+ max_batch: usize,
+) -> (EnrichmentHandle, JoinHandle<()>) {
+ let (tx, rx) = mpsc::channel(capacity);
+ let worker = EnrichmentWorker {
+ sink,
+ rx,
+ batch_window,
+ max_batch,
+ };
+ let join = tokio::spawn(worker.run());
+ (
+ EnrichmentHandle {
+ tx,
+ dropped: Arc::new(AtomicU64::new(0)),
+ },
+ join,
+ )
+}
+
+struct EnrichmentWorker {
+ sink: Arc,
+ rx: mpsc::Receiver,
+ batch_window: Duration,
+ max_batch: usize,
+}
+
+impl EnrichmentWorker {
+ async fn run(mut self) {
+ // `recv` yields `None` only once every sender has dropped and the buffer
+ // is empty, so this loop drains queued work on shutdown before exiting.
+ while let Some(first) = self.rx.recv().await {
+ let mut batch = vec![first];
+ let deadline = tokio::time::Instant::now() + self.batch_window;
+ let mut senders_closed = false;
+ while batch.len() < self.max_batch {
+ match tokio::time::timeout_at(deadline, self.rx.recv()).await {
+ Ok(Some(job)) => batch.push(job),
+ Ok(None) => {
+ senders_closed = true;
+ break;
+ }
+ Err(_elapsed) => break,
+ }
+ }
+ metrics::histogram!("moa_retrieval_enrichment_batch_size").record(batch.len() as f64);
+ self.flush(batch).await;
+ if senders_closed {
+ break;
+ }
+ }
+ }
+
+ async fn flush(&self, batch: Vec) {
+ let started = std::time::Instant::now();
+ // Coalesce access bumps per (scope, role): the deduped union of uids is
+ // written as a single `UPDATE ... WHERE uid = ANY($1)` for each scope.
+ let mut bumps: HashMap<(MemoryScope, bool), HashSet> = HashMap::new();
+ let mut lineage_jobs = Vec::new();
+ for job in batch {
+ match job {
+ EnrichmentJob::AccessBump {
+ scope,
+ uids,
+ assume_app_role,
+ } => {
+ bumps
+ .entry((scope, assume_app_role))
+ .or_default()
+ .extend(uids);
+ }
+ lineage @ EnrichmentJob::Lineage { .. } => lineage_jobs.push(lineage),
+ }
+ }
+ for ((scope, assume_app_role), uids) in bumps {
+ self.sink
+ .bump_access(scope, uids.into_iter().collect(), assume_app_role)
+ .await;
+ }
+ for job in lineage_jobs {
+ if let EnrichmentJob::Lineage {
+ scope,
+ lineage,
+ hits,
+ retrieved_at,
+ assume_app_role,
+ } = job
+ {
+ self.sink
+ .write_lineage(scope, lineage, hits, retrieved_at, assume_app_role)
+ .await;
+ }
+ }
+ metrics::histogram!("moa_retrieval_enrichment_flush_seconds")
+ .record(started.elapsed().as_secs_f64());
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Mutex;
+
+ use moa_core::SessionId;
+
+ use super::*;
+
+ #[derive(Default)]
+ struct CountingSink {
+ bumps: Mutex, bool)>>,
+ lineage_writes: Mutex,
+ }
+
+ #[async_trait]
+ impl EnrichmentSink for CountingSink {
+ async fn bump_access(
+ &self,
+ scope: MemoryScope,
+ mut uids: Vec,
+ assume_app_role: bool,
+ ) {
+ uids.sort();
+ self.bumps.lock().expect("counting sink bumps lock").push((
+ scope,
+ uids,
+ assume_app_role,
+ ));
+ }
+
+ async fn write_lineage(
+ &self,
+ _scope: MemoryScope,
+ _lineage: LineageContext,
+ _hits: Vec,
+ _retrieved_at: DateTime,
+ _assume_app_role: bool,
+ ) {
+ *self
+ .lineage_writes
+ .lock()
+ .expect("counting sink lineage lock") += 1;
+ }
+ }
+
+ fn scope() -> MemoryScope {
+ MemoryScope::Tenant {
+ tenant_id: moa_core::TenantId::new(),
+ }
+ }
+
+ fn lineage_context() -> LineageContext {
+ LineageContext {
+ session_id: SessionId::new(),
+ turn_id: None,
+ turn_seq: 1,
+ }
+ }
+
+ #[tokio::test]
+ async fn enqueue_never_blocks_and_drops_with_metric_when_queue_is_full() {
+ // Pins: F20 — enqueue is non-blocking; once the fixed-capacity queue is
+ // full, further jobs are dropped (counted) rather than backpressuring the
+ // retrieval path. Holds the receiver unconsumed so the queue stays full.
+ let (tx, _rx) = mpsc::channel::(2);
+ let handle = EnrichmentHandle {
+ tx,
+ dropped: Arc::new(AtomicU64::new(0)),
+ };
+ let uid = Uuid::now_v7();
+
+ // Fill the two-slot queue.
+ handle.enqueue_access_bump(scope(), vec![uid], false);
+ handle.enqueue_access_bump(scope(), vec![uid], false);
+ assert_eq!(handle.dropped(), 0, "enqueues within capacity are accepted");
+
+ // The next two enqueues overflow and are dropped, not blocked.
+ handle.enqueue_access_bump(scope(), vec![uid], false);
+ handle.enqueue_access_bump(scope(), vec![uid], false);
+ assert_eq!(
+ handle.dropped(),
+ 2,
+ "overflow jobs are dropped with a counter"
+ );
+ }
+
+ #[tokio::test]
+ async fn access_bumps_coalesce_into_one_deduped_batched_update() {
+ // Pins: F20 — access bumps within one batch coalesce to a single sink call
+ // per scope carrying the deduped union of uids, instead of one write each.
+ let sink = Arc::new(CountingSink::default());
+ let (handle, join) = spawn_with_sink(sink.clone(), 1024, Duration::from_secs(10), 256);
+ let scope = scope();
+ let u1 = Uuid::from_u128(1);
+ let u2 = Uuid::from_u128(2);
+ let u3 = Uuid::from_u128(3);
+
+ handle.enqueue_access_bump(scope.clone(), vec![u1, u2], false);
+ handle.enqueue_access_bump(scope.clone(), vec![u2, u3], false);
+ // Dropping the handle closes the channel, so the worker drains both queued
+ // bumps into one final batch and flushes deterministically.
+ drop(handle);
+ join.await.expect("worker joins after draining");
+
+ let bumps = sink.bumps.lock().expect("bumps");
+ assert_eq!(bumps.len(), 1, "the two bumps coalesce into one sink call");
+ let (bumped_scope, mut uids, role) = bumps[0].clone();
+ uids.sort();
+ assert_eq!(bumped_scope, scope);
+ assert!(!role);
+ assert_eq!(
+ uids,
+ vec![u1, u2, u3],
+ "coalesced uids are the deduped union"
+ );
+ }
+
+ #[tokio::test]
+ async fn shutdown_drains_queued_work() {
+ // Pins: F20 — queued enrichment is flushed on shutdown (all senders drop),
+ // not silently discarded.
+ let sink = Arc::new(CountingSink::default());
+ let (handle, join) = spawn_with_sink(sink.clone(), 1024, Duration::from_secs(10), 256);
+
+ for _ in 0..5 {
+ handle.enqueue_lineage(
+ scope(),
+ lineage_context(),
+ vec![RetrievalLineageHit {
+ uid: Uuid::now_v7(),
+ chunk_uid: None,
+ document_version_uid: None,
+ }],
+ Utc::now(),
+ false,
+ );
+ }
+ drop(handle);
+ join.await.expect("worker drains and joins on shutdown");
+
+ assert_eq!(
+ *sink.lineage_writes.lock().expect("lineage writes"),
+ 5,
+ "all queued lineage writes are flushed on shutdown"
+ );
+ }
+}
diff --git a/crates/moa-brain/src/retrieval/hybrid.rs b/crates/moa-brain/src/retrieval/hybrid.rs
index 721f3625d..230559c4b 100644
--- a/crates/moa-brain/src/retrieval/hybrid.rs
+++ b/crates/moa-brain/src/retrieval/hybrid.rs
@@ -11,7 +11,7 @@ use chrono::{DateTime, Utc};
use moa_core::MoaConfig;
use moa_core::RlsContext;
use moa_db::ScopedConn;
-use moa_memory_graph::{GraphStore, NodeIndexRow, NodeLabel};
+use moa_memory_graph::{GraphError, GraphStore, NodeIndexRow, NodeLabel};
use moa_memory_types::MemoryScope;
use moa_memory_vector::{Error as VectorError, PgvectorStore, TurbopufferStore};
use moa_providers::{ConfiguredReranker, Reranker, build_reranker_from_config};
@@ -20,14 +20,14 @@ use sqlx::PgPool;
use uuid::Uuid;
use crate::planning::Strategy;
+use crate::retrieval::enrichment::EnrichmentHandle;
use crate::retrieval::graph_seed::{
GraphSeedPlan, hydrate_graph_seed_rows, interim_graph_seed_plan, semantic_entity_seed_uids,
};
use crate::retrieval::legs::{
GRAPH_BUDGET, GRAPH_WEIGHT, LEXICAL_BUDGET, LEXICAL_WEIGHT, LegCandidate, RRF_K, VECTOR_BUDGET,
- VECTOR_WEIGHT, bump_last_accessed, graph_expansion_leg_with_diagnostics, hydrate_nodes,
- lexical_leg, rrf_fuse, timed_leg, turbopuffer_bm25_leg, vector_leg as run_vector_leg,
- walk_scoring, write_retrieval_lineage,
+ VECTOR_WEIGHT, graph_expansion_leg_with_diagnostics, hydrate_nodes, lexical_leg, rrf_fuse,
+ timed_leg, turbopuffer_bm25_leg, vector_leg as run_vector_leg, walk_scoring,
};
use crate::retrieval::policy::{GraphRetrievalPolicy, effective_graph_policy};
use crate::retrieval::ranking::{
@@ -61,6 +61,7 @@ pub struct HybridRetriever {
lineage_enabled: bool,
lineage_sample_rate: f64,
graph_policy: GraphRetrievalPolicy,
+ enrichment: Option,
}
/// Deterministically decides whether one turn's retrieval writes lineage rows.
@@ -108,9 +109,19 @@ impl HybridRetriever {
lineage_enabled: false,
lineage_sample_rate: 1.0,
graph_policy: GraphRetrievalPolicy::default(),
+ enrichment: None,
}
}
+ /// Attaches the shared bounded enrichment worker used for post-retrieval
+ /// `last_accessed_at` bumps and sampled lineage writes. When absent, those
+ /// best-effort writes are skipped.
+ #[must_use]
+ pub(crate) fn with_enrichment(mut self, enrichment: Option) -> Self {
+ self.enrichment = enrichment;
+ self
+ }
+
/// Creates a hybrid retriever from shared config.
#[must_use]
pub fn from_config(
@@ -241,9 +252,13 @@ impl HybridRetriever {
LEXICAL_BUDGET,
self.lexical_leg(&req, &backend_state),
);
- let (vector_hits, lexical_hits) = tokio::join!(vector_future, lexical_future);
- let mut vector_hits = vector_hits?;
- let mut lexical_hits = lexical_hits?;
+ let (vector_outcome, lexical_outcome) = tokio::join!(vector_future, lexical_future);
+ // Reduce each leg to its usable hits: a fatal error aborts, a transient
+ // error or timeout degrades to empty while the peer leg's hits are kept.
+ let vector_leg = reduce_leg("vector", vector_outcome)?;
+ let vector_timed_out = vector_leg.timed_out;
+ let mut vector_hits = vector_leg.value;
+ let mut lexical_hits = reduce_leg("lexical", lexical_outcome)?.value;
apply_lexical_boost_only_policy(&req, &vector_hits, &mut lexical_hits);
let interim = rrf_fuse(
&[],
@@ -284,7 +299,7 @@ impl HybridRetriever {
Default::default()
} else {
let graph_started = std::time::Instant::now();
- let mut output = run_leg(
+ let graph_outcome = run_leg(
req.disable_leg_timeouts,
"graph",
GRAPH_BUDGET,
@@ -299,7 +314,8 @@ impl HybridRetriever {
self.ranking_config.graph_rescue_evidence_floor,
),
)
- .await?;
+ .await;
+ let mut output = reduce_leg("graph", graph_outcome)?.value;
output.diagnostics.graph_latency_ms = duration_ms_u64(graph_started.elapsed());
output
};
@@ -320,12 +336,21 @@ impl HybridRetriever {
if fused.is_empty()
&& should_retry_vector_after_empty_fusion(
&req,
- &vector_hits,
+ vector_timed_out,
&lexical_hits.candidates,
&graph_hits,
)
{
- vector_hits = self.vector_leg(&req, &backend_state).await?;
+ // Retry only an observed timeout, and under the same leg budget so a
+ // slow backend cannot turn one bounded timeout into an uncapped tail.
+ let retry_outcome = run_leg(
+ req.disable_leg_timeouts,
+ "vector",
+ VECTOR_BUDGET,
+ self.vector_leg(&req, &backend_state),
+ )
+ .await;
+ vector_hits = reduce_leg("vector", retry_outcome)?.value;
fused = rrf_fuse(
&graph_hits,
&vector_hits,
@@ -379,45 +404,35 @@ impl HybridRetriever {
metrics::histogram!("moa_retrieval_rrf_rerank_seconds")
.record(fusion_started.elapsed().as_secs_f64());
- if req.ranking_reference_time.is_none() {
- let touched_uids = final_hits.iter().map(|hit| hit.uid).collect::>();
- let pool = self.pool.clone();
- let scope = req.scope.clone();
- let assume_app_role = self.assume_app_role;
- tokio::spawn(async move {
- if let Err(error) =
- bump_last_accessed(pool, scope, touched_uids, assume_app_role).await
- {
- tracing::debug!(error = %error, "failed to bump graph-memory access timestamps");
- }
- });
- }
- if self.lineage_enabled
- && let Some(lineage) = req.lineage
- && lineage_turn_sampled(&lineage, self.lineage_sample_rate)
- {
- let ranked_hits = final_hits
- .iter()
- .map(RetrievalLineageHit::from_hit)
- .collect::>();
- let pool = self.pool.clone();
- let scope = req.scope.clone();
- let assume_app_role = self.assume_app_role;
- let retrieved_at = Utc::now();
- tokio::spawn(async move {
- if let Err(error) = write_retrieval_lineage(
- pool,
- scope,
+ // F20: post-retrieval enrichment is best-effort and is routed through the
+ // shared bounded worker (coalesced, drop-on-overflow, drained on shutdown)
+ // instead of an unbounded detached task per retrieval. When no worker is
+ // configured (unit/eval retrievers) the writes are simply skipped.
+ if let Some(enrichment) = &self.enrichment {
+ if req.ranking_reference_time.is_none() {
+ let touched_uids = final_hits.iter().map(|hit| hit.uid).collect::>();
+ enrichment.enqueue_access_bump(
+ req.scope.clone(),
+ touched_uids,
+ self.assume_app_role,
+ );
+ }
+ if self.lineage_enabled
+ && let Some(lineage) = req.lineage
+ && lineage_turn_sampled(&lineage, self.lineage_sample_rate)
+ {
+ let ranked_hits = final_hits
+ .iter()
+ .map(RetrievalLineageHit::from_hit)
+ .collect::>();
+ enrichment.enqueue_lineage(
+ req.scope.clone(),
lineage,
ranked_hits,
- retrieved_at,
- assume_app_role,
- )
- .await
- {
- tracing::debug!(error = %error, "failed to write graph-memory retrieval lineage");
- }
- });
+ Utc::now(),
+ self.assume_app_role,
+ );
+ }
}
Ok(RetrievalOutput {
@@ -562,11 +577,24 @@ impl HybridRetriever {
hits: &[RetrievalHit],
) -> Result> {
let documents = hits.iter().map(rerank_document).collect::>();
- let reranked = self
+ let reranked = match self
.reranker
.rerank(&self.rerank_model, &req.query_text, &documents, req.k_final)
.await
- .map_err(|error| RetrievalError::Rerank(error.to_string()))?;
+ {
+ Ok(reranked) => reranked,
+ // Reranking is a best-effort refinement over already-fused hits: a
+ // provider failure must not abort otherwise-usable results. Fall back
+ // to the fused pre-rerank order, mirroring the empty-output fallback.
+ Err(error) => {
+ record_leg_degraded("rerank", "error");
+ tracing::warn!(
+ error = %error,
+ "reranker failed; falling back to fused pre-rerank order"
+ );
+ return Ok(hits.iter().take(req.k_final).cloned().collect());
+ }
+ };
let mut out = Vec::with_capacity(req.k_final.min(reranked.len()));
for hit in reranked {
if let Some(candidate) = hits.get(hit.index) {
@@ -592,13 +620,16 @@ fn fused_candidate_limit(k_final: usize) -> usize {
fn should_retry_vector_after_empty_fusion(
req: &RetrievalRequest,
- vector_hits: &[LegCandidate],
+ vector_timed_out: bool,
lexical_hits: &[LegCandidate],
graph_hits: &[LegCandidate],
) -> bool {
+ // Retry only when the vector leg specifically timed out: a genuinely empty
+ // vector result (or a degraded transient failure) is not masking candidates,
+ // so re-running it would only duplicate work.
!req.disable_leg_timeouts
&& !req.query_embedding.is_empty()
- && vector_hits.is_empty()
+ && vector_timed_out
&& lexical_hits.is_empty()
&& graph_hits.is_empty()
}
@@ -777,20 +808,160 @@ fn weights_for(strategy: Strategy) -> (f64, f64, f64) {
}
}
-fn leg_or_empty(
- name: &'static str,
- result: std::result::Result, tokio::time::error::Elapsed>,
-) -> Result
-where
- T: Default,
-{
- match result {
- Ok(Ok(value)) => Ok(value),
- Ok(Err(error)) => Err(error),
- Err(_) => {
- tracing::debug!(leg = name, "hybrid retrieval leg exceeded budget");
- Ok(T::default())
+/// Classified outcome of one retrieval leg.
+///
+/// Threads the *reason* a leg produced no usable hits — instead of collapsing
+/// every non-success into an empty default — so the caller can keep a
+/// successful peer leg, decide whether a timeout is worth one bounded retry,
+/// and abort only on a fatal (RLS/privacy/scope/invalid-config) error.
+enum LegOutcome {
+ /// The leg completed; `value` may be empty (a genuine empty result).
+ Completed(T),
+ /// The leg exceeded its time budget.
+ Timeout,
+ /// The leg hit a transient optional-backend/provider error; degrade it.
+ Transient(RetrievalError),
+ /// The leg hit an RLS/privacy/scope/invalid-config error; abort retrieval.
+ Fatal(RetrievalError),
+}
+
+/// One leg reduced to its usable value plus degradation signals.
+struct LegDegradation {
+ /// Usable hits (empty when the leg degraded or genuinely returned nothing).
+ value: T,
+ /// Whether the leg degraded (a timeout or transient error, not a real empty).
+ #[allow(dead_code)]
+ degraded: bool,
+ /// Whether the leg specifically timed out; gates the bounded vector retry.
+ timed_out: bool,
+}
+
+/// Reduces a [`LegOutcome`] to usable hits.
+///
+/// A completed leg passes through. A timeout or transient error degrades to an
+/// empty result (recording a metric and warning) so the peer leg's hits are
+/// kept. A fatal error aborts the whole retrieval.
+fn reduce_leg(name: &'static str, outcome: LegOutcome) -> Result> {
+ match outcome {
+ LegOutcome::Completed(value) => Ok(LegDegradation {
+ value,
+ degraded: false,
+ timed_out: false,
+ }),
+ LegOutcome::Timeout => {
+ record_leg_degraded(name, "timeout");
+ tracing::warn!(
+ leg = name,
+ "hybrid retrieval leg exceeded budget; degrading to empty and keeping peer legs"
+ );
+ Ok(LegDegradation {
+ value: T::default(),
+ degraded: true,
+ timed_out: true,
+ })
}
+ LegOutcome::Transient(error) => {
+ record_leg_degraded(name, "transient");
+ tracing::warn!(
+ leg = name,
+ error = %error,
+ "hybrid retrieval leg failed transiently; degrading to empty and keeping peer legs"
+ );
+ Ok(LegDegradation {
+ value: T::default(),
+ degraded: true,
+ timed_out: false,
+ })
+ }
+ LegOutcome::Fatal(error) => Err(error),
+ }
+}
+
+fn record_leg_degraded(leg: &'static str, reason: &'static str) {
+ metrics::counter!("moa_retrieval_leg_degraded_total", "leg" => leg, "reason" => reason)
+ .increment(1);
+}
+
+fn classify_leg_result(error: RetrievalError) -> LegOutcome {
+ if is_fatal_retrieval_error(&error) {
+ LegOutcome::Fatal(error)
+ } else {
+ LegOutcome::Transient(error)
+ }
+}
+
+/// Classifies a retrieval error as fatal (abort) or transient (degrade).
+///
+/// Fatal covers RLS/privacy/scope failures and invalid-configuration/contract
+/// errors that will not self-heal and must not be silently degraded. Transient
+/// covers ordinary optional backend, provider, network, and query failures that
+/// should degrade one leg while its peers are kept.
+fn is_fatal_retrieval_error(error: &RetrievalError) -> bool {
+ match error {
+ // Scope/RLS transaction setup failed: privacy boundary, abort.
+ RetrievalError::Scope(_) => true,
+ // A direct sidecar query failure in the fusion/hydration path is an
+ // ordinary backend error.
+ RetrievalError::Sqlx(_) => false,
+ RetrievalError::Graph(error) => is_fatal_graph_error(error),
+ RetrievalError::Vector(error) => is_fatal_vector_error(error),
+ }
+}
+
+fn is_fatal_graph_error(error: &GraphError) -> bool {
+ match error {
+ // RLS/privacy/scope invariants.
+ GraphError::RlsDenied | GraphError::MissingScope | GraphError::Scope(_) => true,
+ // Invalid data/config invariants that will not self-heal.
+ GraphError::MissingEmbeddingMetadata
+ | GraphError::Conflict(_)
+ | GraphError::BiTemporal(_)
+ | GraphError::UnknownNodeLabel(_)
+ | GraphError::UnknownEdgeLabel(_)
+ | GraphError::UnknownPiiClass(_)
+ | GraphError::ChangelogScopeMismatch { .. }
+ | GraphError::InvalidChangelogScope => true,
+ // Ordinary backend/query failures degrade this leg.
+ GraphError::GraphQuery(_)
+ | GraphError::Sidecar(_)
+ | GraphError::NotFound(_)
+ | GraphError::Json(_) => false,
+ // Delegate to the vector classification.
+ GraphError::Vector(error) => is_fatal_vector_error(error),
+ }
+}
+
+fn is_fatal_vector_error(error: &VectorError) -> bool {
+ match error {
+ // Embedding/index shape, privacy, backend-capability, and configuration
+ // invariants: retrying or degrading would hide a real misconfiguration.
+ VectorError::DimensionMismatch { .. }
+ | VectorError::UnknownPiiClass(_)
+ | VectorError::EmbedderConfig(_)
+ | VectorError::EmbedderMismatch { .. }
+ | VectorError::StoragePartitionEmbedderStateMissing { .. }
+ | VectorError::EmbedderModelMismatch { .. }
+ | VectorError::QueryLimitTooLarge(_)
+ | VectorError::StoragePartitionRequired { .. }
+ | VectorError::TurbopufferUnavailable { .. }
+ | VectorError::TurbopufferBaaRequired { .. }
+ | VectorError::TurbopufferConfig(_)
+ | VectorError::PromotionValidationFailed { .. }
+ | VectorError::InvalidPromotionState { .. }
+ | VectorError::TransactionalWritesUnsupported(_)
+ | VectorError::InvalidVectorSyncOperation(_)
+ | VectorError::UnsupportedVectorBackend { .. }
+ | VectorError::UnsupportedQueryFeature { .. } => true,
+ // Transient backend/provider/network/response failures degrade the leg.
+ VectorError::EmbeddingResponseLength { .. }
+ | VectorError::ProviderStatus { .. }
+ | VectorError::VectorProviderStatus { .. }
+ | VectorError::ReembedInProgress { .. }
+ | VectorError::TurbopufferResponse(_)
+ | VectorError::Core(_)
+ | VectorError::Sqlx(_)
+ | VectorError::Reqwest(_)
+ | VectorError::SerdeJson(_) => false,
}
}
@@ -799,15 +970,22 @@ async fn run_leg(
name: &'static str,
budget: std::time::Duration,
future: F,
-) -> Result
+) -> LegOutcome
where
T: Default,
F: std::future::Future