P5.3 §13.3: Adaptive replica selection (EWMA-based)
Implemented EWMA-scored replica selection replacing round-robin:
- score(node) = α · latency_p95_ms + β · in_flight_count + γ · error_rate
- Router picks lowest-scoring node with probability 1-ε
- With ε (default 0.05) picks uniformly random for exploration
Config (plan §13.3):
replica_selection:
strategy: adaptive | round_robin | random
latency_weight: 1.0
inflight_weight: 2.0
error_weight: 10.0
ewma_half_life_ms: 5000
exploration_epsilon: 0.05
Metrics:
- miroir_replica_selection_score{node_id} gauge
- miroir_replica_selection_exploration_total counter
Acceptance tests pass:
- Degraded node traffic drops within 2× half-life
- Node recovers after latency clears
- Exploration samples degraded node (~1.7% with ε=0.05)
- Round-robin fallback works identically to Phase 1
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
jedarden
parent
c5cd8b91c0
commit
99767d95c7
7 changed files with 879 additions and 79 deletions
|
|
@ -148,6 +148,21 @@ impl Default for NodeMetrics {
|
|||
}
|
||||
}
|
||||
|
||||
/// Callback for reporting selection events.
|
||||
pub trait SelectionObserver: Send + Sync {
|
||||
/// Called when a node is selected with its score.
|
||||
fn report_selection(&self, node_id: &str, score: f64);
|
||||
/// Called when exploration selects a random node.
|
||||
fn report_exploration(&self);
|
||||
}
|
||||
|
||||
/// No-op observer for when metrics aren't needed.
|
||||
struct NoOpObserver;
|
||||
impl SelectionObserver for NoOpObserver {
|
||||
fn report_selection(&self, _node_id: &str, _score: f64) {}
|
||||
fn report_exploration(&self) {}
|
||||
}
|
||||
|
||||
/// Replica selector.
|
||||
pub struct ReplicaSelector {
|
||||
/// Configuration.
|
||||
|
|
@ -158,19 +173,27 @@ pub struct ReplicaSelector {
|
|||
rr_counter: Arc<RwLock<HashMap<String, u64>>>,
|
||||
/// Random number generator.
|
||||
rng: Arc<std::sync::Mutex<StdRng>>,
|
||||
/// Observer for selection events.
|
||||
observer: Arc<dyn SelectionObserver>,
|
||||
}
|
||||
|
||||
impl ReplicaSelector {
|
||||
/// Create a new replica selector.
|
||||
pub fn new(config: ReplicaSelectionConfig) -> Self {
|
||||
/// Create a new replica selector with a metrics observer.
|
||||
pub fn new_with_observer(config: ReplicaSelectionConfig, observer: Arc<dyn SelectionObserver>) -> Self {
|
||||
Self {
|
||||
config,
|
||||
metrics: Arc::new(RwLock::new(HashMap::new())),
|
||||
rr_counter: Arc::new(RwLock::new(HashMap::new())),
|
||||
rng: Arc::new(std::sync::Mutex::new(StdRng::from_entropy())),
|
||||
observer,
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a new replica selector without metrics.
|
||||
pub fn new(config: ReplicaSelectionConfig) -> Self {
|
||||
Self::new_with_observer(config, Arc::new(NoOpObserver))
|
||||
}
|
||||
|
||||
/// Select a node from the given candidates.
|
||||
///
|
||||
/// Returns the selected node ID, or None if candidates is empty.
|
||||
|
|
@ -194,12 +217,21 @@ impl ReplicaSelector {
|
|||
|
||||
// Exploration: with probability epsilon, pick randomly
|
||||
if self.should_explore() {
|
||||
return self.select_random(candidates);
|
||||
self.observer.report_exploration();
|
||||
let selected = self.select_random(candidates);
|
||||
if let Some(ref node) = selected {
|
||||
let score = metrics
|
||||
.get(node)
|
||||
.map(|m| m.score(&self.config))
|
||||
.unwrap_or(1000.0);
|
||||
self.observer.report_selection(node.as_str(), score);
|
||||
}
|
||||
return selected;
|
||||
}
|
||||
|
||||
// Compute scores and find the minimum
|
||||
let mut best_node = None;
|
||||
// Compute scores and collect all nodes with the minimum score
|
||||
let mut best_score = f64::INFINITY;
|
||||
let mut best_nodes: Vec<NodeId> = Vec::new();
|
||||
|
||||
for node in candidates {
|
||||
let score = metrics
|
||||
|
|
@ -209,11 +241,27 @@ impl ReplicaSelector {
|
|||
|
||||
if score < best_score {
|
||||
best_score = score;
|
||||
best_node = Some(node.clone());
|
||||
best_nodes.clear();
|
||||
best_nodes.push(node.clone());
|
||||
} else if (score - best_score).abs() < 1e-10 {
|
||||
// Scores are essentially equal - add to tie list
|
||||
best_nodes.push(node.clone());
|
||||
}
|
||||
}
|
||||
|
||||
best_node
|
||||
// If multiple nodes have the same best score, pick randomly
|
||||
let selected = if best_nodes.len() == 1 {
|
||||
best_nodes.into_iter().next()
|
||||
} else {
|
||||
let idx = self.rng.lock().unwrap().gen_range(0..best_nodes.len());
|
||||
best_nodes.get(idx).cloned()
|
||||
};
|
||||
|
||||
if let Some(ref node) = selected {
|
||||
self.observer.report_selection(node.as_str(), best_score);
|
||||
}
|
||||
|
||||
selected
|
||||
}
|
||||
|
||||
/// Round-robin selection.
|
||||
|
|
@ -293,6 +341,18 @@ impl ReplicaSelector {
|
|||
}
|
||||
}
|
||||
|
||||
impl Clone for ReplicaSelector {
|
||||
fn clone(&self) -> Self {
|
||||
Self {
|
||||
config: self.config.clone(),
|
||||
metrics: Arc::clone(&self.metrics),
|
||||
rr_counter: Arc::clone(&self.rr_counter),
|
||||
rng: Arc::clone(&self.rng),
|
||||
observer: Arc::clone(&self.observer),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for ReplicaSelector {
|
||||
fn default() -> Self {
|
||||
Self::new(ReplicaSelectionConfig::default())
|
||||
|
|
@ -348,30 +408,9 @@ mod tests {
|
|||
let node1 = NodeId::new("node-1".to_string());
|
||||
let node2 = NodeId::new("node-2".to_string());
|
||||
|
||||
// Record some metrics
|
||||
{
|
||||
let mut metrics = selector.metrics.write().await;
|
||||
metrics.insert(
|
||||
node1.clone(),
|
||||
NodeMetrics {
|
||||
latency_p95_ms: 10.0,
|
||||
in_flight: 0,
|
||||
error_rate: 0.0,
|
||||
half_life_ms: 5000,
|
||||
last_updated: Instant::now(),
|
||||
},
|
||||
);
|
||||
metrics.insert(
|
||||
node2.clone(),
|
||||
NodeMetrics {
|
||||
latency_p95_ms: 100.0,
|
||||
in_flight: 0,
|
||||
error_rate: 0.0,
|
||||
half_life_ms: 5000,
|
||||
last_updated: Instant::now(),
|
||||
},
|
||||
);
|
||||
}
|
||||
// Seed metrics by recording successful requests
|
||||
selector.record_success(&node1, 10.0).await;
|
||||
selector.record_success(&node2, 100.0).await;
|
||||
|
||||
// Should select node-1 (lower score)
|
||||
let candidates = vec![node2.clone(), node1.clone()];
|
||||
|
|
@ -416,10 +455,12 @@ mod tests {
|
|||
assert!(metrics.is_some());
|
||||
assert_eq!(metrics.unwrap().in_flight, 1);
|
||||
|
||||
// Record success decrements in-flight and updates latency
|
||||
selector.record_success(&node, 50.0).await;
|
||||
|
||||
let metrics = selector.get_metrics(&node).await;
|
||||
assert!(metrics.is_some());
|
||||
// In-flight should be decremented (from 1 to 0)
|
||||
assert_eq!(metrics.unwrap().in_flight, 0);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -384,11 +384,12 @@ pub struct ScatterResult {
|
|||
}
|
||||
|
||||
#[instrument(skip_all, fields(query_seq, rf, shard_count))]
|
||||
pub fn plan_search_scatter(
|
||||
pub async fn plan_search_scatter(
|
||||
topology: &Topology,
|
||||
query_seq: u64,
|
||||
rf: usize,
|
||||
shard_count: u32,
|
||||
replica_selector: Option<&ReplicaSelector>,
|
||||
) -> ScatterPlan {
|
||||
let chosen_group = query_group(query_seq, topology.replica_group_count());
|
||||
|
||||
|
|
@ -407,7 +408,16 @@ pub fn plan_search_scatter(
|
|||
let mut shard_to_node = HashMap::new();
|
||||
for shard_id in 0..shard_count {
|
||||
let replicas = crate::router::assign_shard_in_group(shard_id, group.nodes(), rf);
|
||||
let selected = replicas[(query_seq as usize) % replicas.len()].clone();
|
||||
|
||||
let selected = if let Some(selector) = replica_selector {
|
||||
match selector.select(&replicas, chosen_group).await {
|
||||
Some(node) => node,
|
||||
None => replicas[(query_seq as usize) % replicas.len()].clone(),
|
||||
}
|
||||
} else {
|
||||
replicas[(query_seq as usize) % replicas.len()].clone()
|
||||
};
|
||||
|
||||
shard_to_node.insert(shard_id, selected);
|
||||
}
|
||||
|
||||
|
|
@ -424,7 +434,7 @@ pub fn plan_search_scatter(
|
|||
///
|
||||
/// Excludes nodes whose settings version for the given index is below `floor`.
|
||||
/// Returns None if no covering set can be assembled (caller should return 503).
|
||||
pub fn plan_search_scatter_with_version_floor(
|
||||
pub async fn plan_search_scatter_with_version_floor(
|
||||
topology: &Topology,
|
||||
query_seq: u64,
|
||||
rf: usize,
|
||||
|
|
@ -432,6 +442,7 @@ pub fn plan_search_scatter_with_version_floor(
|
|||
index: &str,
|
||||
floor: u64,
|
||||
version_checker: &impl Fn(&str, &str) -> u64,
|
||||
replica_selector: Option<&ReplicaSelector>,
|
||||
) -> Option<ScatterPlan> {
|
||||
let chosen_group = query_group(query_seq, topology.replica_group_count());
|
||||
|
||||
|
|
@ -450,7 +461,7 @@ pub fn plan_search_scatter_with_version_floor(
|
|||
let mut shard_to_node = HashMap::new();
|
||||
for shard_id in 0..shard_count {
|
||||
let replicas = crate::router::assign_shard_in_group(shard_id, group.nodes(), rf);
|
||||
// Filter by version floor, then rotate by query_seq
|
||||
// Filter by version floor
|
||||
let eligible: Vec<_> = replicas
|
||||
.iter()
|
||||
.filter(|node_id| {
|
||||
|
|
@ -463,7 +474,17 @@ pub fn plan_search_scatter_with_version_floor(
|
|||
return None;
|
||||
}
|
||||
|
||||
let selected = eligible[query_seq as usize % eligible.len()];
|
||||
let selected = if let Some(selector) = replica_selector {
|
||||
// Convert Vec<&NodeId> to Vec<NodeId> for selector
|
||||
let eligible_owned: Vec<NodeId> = eligible.iter().map(|&n| n.clone()).collect();
|
||||
match selector.select(&eligible_owned, chosen_group).await {
|
||||
Some(node) => node,
|
||||
None => eligible[query_seq as usize % eligible.len()].clone(),
|
||||
}
|
||||
} else {
|
||||
eligible[query_seq as usize % eligible.len()].clone()
|
||||
};
|
||||
|
||||
shard_to_node.insert(shard_id, selected.clone());
|
||||
}
|
||||
|
||||
|
|
@ -480,12 +501,13 @@ pub fn plan_search_scatter_with_version_floor(
|
|||
///
|
||||
/// Used when a session has a pending write and needs to read from the pinned group
|
||||
/// to ensure read-your-writes consistency.
|
||||
pub fn plan_search_scatter_for_group(
|
||||
pub async fn plan_search_scatter_for_group(
|
||||
topology: &Topology,
|
||||
query_seq: u64,
|
||||
rf: usize,
|
||||
shard_count: u32,
|
||||
pinned_group: u32,
|
||||
replica_selector: Option<&ReplicaSelector>,
|
||||
) -> Option<ScatterPlan> {
|
||||
let group = topology.group(pinned_group)?;
|
||||
|
||||
|
|
@ -495,7 +517,16 @@ pub fn plan_search_scatter_for_group(
|
|||
if replicas.is_empty() {
|
||||
continue;
|
||||
}
|
||||
let selected = replicas[query_seq as usize % replicas.len()].clone();
|
||||
|
||||
let selected = if let Some(selector) = replica_selector {
|
||||
match selector.select(&replicas, pinned_group).await {
|
||||
Some(node) => node,
|
||||
None => replicas[query_seq as usize % replicas.len()].clone(),
|
||||
}
|
||||
} else {
|
||||
replicas[query_seq as usize % replicas.len()].clone()
|
||||
};
|
||||
|
||||
shard_to_node.insert(shard_id, selected);
|
||||
}
|
||||
|
||||
|
|
@ -979,43 +1010,43 @@ mod tests {
|
|||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_plan_pure_function() {
|
||||
#[tokio::test]
|
||||
async fn test_plan_pure_function() {
|
||||
let topo = make_test_topology();
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
|
||||
assert_eq!(plan.chosen_group, 0);
|
||||
assert_eq!(plan.target_shards.len(), 64);
|
||||
assert!(plan.hedging_eligible);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_plan_group_rotation() {
|
||||
#[tokio::test]
|
||||
async fn test_plan_group_rotation() {
|
||||
let topo = make_test_topology();
|
||||
assert_eq!(plan_search_scatter(&topo, 0, 2, 64).chosen_group, 0);
|
||||
assert_eq!(plan_search_scatter(&topo, 1, 2, 64).chosen_group, 1);
|
||||
assert_eq!(plan_search_scatter(&topo, 0, 2, 64, None).await.chosen_group, 0);
|
||||
assert_eq!(plan_search_scatter(&topo, 1, 2, 64, None).await.chosen_group, 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_plan_shard_mapping() {
|
||||
#[tokio::test]
|
||||
async fn test_plan_shard_mapping() {
|
||||
let topo = make_test_topology();
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
|
||||
for s in 0..64 { assert!(plan.shard_to_node.contains_key(&s)); }
|
||||
let g0 = topo.group(0).unwrap();
|
||||
for (_, nid) in &plan.shard_to_node { assert!(g0.nodes().contains(nid)); }
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_plan_hedging() {
|
||||
#[tokio::test]
|
||||
async fn test_plan_hedging() {
|
||||
let mut topo = Topology::new(64, 1, 1);
|
||||
topo.add_node(Node::new(NodeId::new("n0".into()), "http://n0:7700".into(), 0));
|
||||
assert!(!plan_search_scatter(&topo, 0, 1, 64).hedging_eligible);
|
||||
assert!(plan_search_scatter(&make_test_topology(), 0, 2, 64).hedging_eligible);
|
||||
assert!(!plan_search_scatter(&topo, 0, 1, 64, None).await.hedging_eligible);
|
||||
assert!(plan_search_scatter(&make_test_topology(), 0, 2, 64, None).await.hedging_eligible);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_scatter_mock() {
|
||||
let topo = make_test_topology();
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
|
||||
let mut c = MockNodeClient::default();
|
||||
c.responses.insert(NodeId::new("node-0".into()), serde_json::json!({"hits": [{"id": "doc1"}], "estimatedTotalHits": 1, "processingTimeMs": 5}));
|
||||
let r = execute_scatter(plan, &c, make_req(), &topo, UnavailableShardPolicy::Partial).await.unwrap();
|
||||
|
|
@ -1026,7 +1057,7 @@ mod tests {
|
|||
#[tokio::test]
|
||||
async fn test_scatter_partial() {
|
||||
let topo = make_test_topology();
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
|
||||
let mut c = MockNodeClient::default();
|
||||
c.errors.insert(NodeId::new("node-0".into()), NodeError::Timeout);
|
||||
let r = execute_scatter(plan, &c, make_req(), &topo, UnavailableShardPolicy::Partial).await.unwrap();
|
||||
|
|
@ -1036,21 +1067,21 @@ mod tests {
|
|||
#[tokio::test]
|
||||
async fn test_scatter_error_policy() {
|
||||
let topo = make_test_topology();
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
|
||||
let mut c = MockNodeClient::default();
|
||||
c.errors.insert(NodeId::new("node-0".into()), NodeError::Timeout);
|
||||
assert!(execute_scatter(plan, &c, make_req(), &topo, UnavailableShardPolicy::Error).await.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_plan_invalid_group() {
|
||||
assert!(plan_search_scatter(&Topology::new(64, 0, 1), 0, 1, 64).shard_to_node.is_empty());
|
||||
#[tokio::test]
|
||||
async fn test_plan_invalid_group() {
|
||||
assert!(plan_search_scatter(&Topology::new(64, 0, 1), 0, 1, 64, None).await.shard_to_node.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_scatter_node_not_in_topo() {
|
||||
let topo = make_test_topology();
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
|
||||
let r = execute_scatter(plan, &MockNodeClient::default(), make_req(), &Topology::new(64, 2, 2), UnavailableShardPolicy::Partial).await.unwrap();
|
||||
assert!(r.partial);
|
||||
}
|
||||
|
|
@ -1058,7 +1089,7 @@ mod tests {
|
|||
#[tokio::test]
|
||||
async fn test_sg_rrf() {
|
||||
let topo = make_test_topology();
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
|
||||
let mut c = MockNodeClient::default();
|
||||
c.responses.insert(NodeId::new("node-0".into()), serde_json::json!({"hits": [{"id": "a", "_rankingScore": 0.9}], "estimatedTotalHits": 1, "processingTimeMs": 5}));
|
||||
let s = crate::merger::RrfStrategy::default_strategy();
|
||||
|
|
@ -1069,7 +1100,7 @@ mod tests {
|
|||
#[tokio::test]
|
||||
async fn test_sg_degraded() {
|
||||
let topo = make_test_topology();
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
|
||||
let mut c = MockNodeClient::default();
|
||||
c.responses.insert(NodeId::new("node-0".into()), serde_json::json!({"hits": [{"id": "a"}], "estimatedTotalHits": 1, "processingTimeMs": 5}));
|
||||
c.errors.insert(NodeId::new("node-2".into()), NodeError::Timeout);
|
||||
|
|
@ -1105,7 +1136,7 @@ mod tests {
|
|||
#[tokio::test]
|
||||
async fn test_execute_preflight() {
|
||||
let topo = make_test_topology();
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
|
||||
let mut c = MockNodeClient::default();
|
||||
c.preflight_responses.insert(NodeId::new("node-0".into()), PreflightResponse {
|
||||
total_docs: 30000, avg_doc_length: 50.0,
|
||||
|
|
@ -1128,7 +1159,7 @@ mod tests {
|
|||
#[tokio::test]
|
||||
async fn test_dfs_query_then_fetch() {
|
||||
let topo = make_test_topology();
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
|
||||
let mut c = MockNodeClient::default();
|
||||
c.responses.insert(NodeId::new("node-0".into()), serde_json::json!({"hits": [{"id": "a", "_rankingScore": 0.9}], "estimatedTotalHits": 1, "processingTimeMs": 5}));
|
||||
c.preflight_responses.insert(NodeId::new("node-0".into()), PreflightResponse {
|
||||
|
|
@ -1165,7 +1196,7 @@ mod tests {
|
|||
topo.add_node(Node::new(NodeId::new("node-1".into()), "http://node-1:7700".into(), 0));
|
||||
topo.add_node(Node::new(NodeId::new("node-2".into()), "http://node-2:7700".into(), 0));
|
||||
|
||||
let plan = plan_search_scatter(&topo, 0, 1, 3);
|
||||
let plan = plan_search_scatter(&topo, 0, 1, 3, None).await;
|
||||
|
||||
// Simulate severely skewed shard distribution
|
||||
let mut c = MockNodeClient::default();
|
||||
|
|
@ -1239,7 +1270,7 @@ mod tests {
|
|||
#[tokio::test]
|
||||
async fn test_dfs_empty_query_terms() {
|
||||
let topo = make_test_topology();
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 64, None).await;
|
||||
let c = MockNodeClient::default();
|
||||
|
||||
let preflight_req = PreflightRequest {
|
||||
|
|
@ -1261,7 +1292,7 @@ mod tests {
|
|||
topo.add_node(Node::new(NodeId::new("node-1".into()), "http://node-1:7700".into(), 0));
|
||||
topo.add_node(Node::new(NodeId::new("node-2".into()), "http://node-2:7700".into(), 0));
|
||||
|
||||
let plan = plan_search_scatter(&topo, 0, 1, 3);
|
||||
let plan = plan_search_scatter(&topo, 0, 1, 3, None).await;
|
||||
let mut c = MockNodeClient::default();
|
||||
|
||||
// Node 0 returns valid data
|
||||
|
|
@ -1394,7 +1425,7 @@ mod tests {
|
|||
topo.add_node(Node::new(NodeId::new("node-g1-0".into()), "http://g1-0:7700".into(), 1));
|
||||
topo.add_node(Node::new(NodeId::new("node-g1-1".into()), "http://g1-1:7700".into(), 1));
|
||||
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 16); // query_seq=0 → group 0
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 16, None).await; // query_seq=0 → group 0
|
||||
assert_eq!(plan.chosen_group, 0);
|
||||
|
||||
let mut c = MockNodeClient::default();
|
||||
|
|
@ -1432,7 +1463,7 @@ mod tests {
|
|||
topo.add_node(Node::new(NodeId::new("node-g1-0".into()), "http://g1-0:7700".into(), 1));
|
||||
topo.add_node(Node::new(NodeId::new("node-g1-1".into()), "http://g1-1:7700".into(), 1));
|
||||
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 16);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 16, None).await;
|
||||
let mut c = MockNodeClient::default();
|
||||
|
||||
// All nodes fail
|
||||
|
|
@ -1459,7 +1490,7 @@ mod tests {
|
|||
topo.add_node(Node::new(NodeId::new("node-g1-0".into()), "http://g1-0:7700".into(), 1));
|
||||
topo.add_node(Node::new(NodeId::new("node-g1-1".into()), "http://g1-1:7700".into(), 1));
|
||||
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 16);
|
||||
let plan = plan_search_scatter(&topo, 0, 2, 16, None).await;
|
||||
let mut c = MockNodeClient::default();
|
||||
|
||||
// Group 1 nodes are healthy but partial policy shouldn't use them
|
||||
|
|
|
|||
475
crates/miroir-core/tests/p13_3_adaptive_replica_selection.rs
Normal file
475
crates/miroir-core/tests/p13_3_adaptive_replica_selection.rs
Normal file
|
|
@ -0,0 +1,475 @@
|
|||
//! P5.3 §13.3 Adaptive replica selection (EWMA-based) acceptance tests.
|
||||
//!
|
||||
//! Tests the EWMA-scored replica selection that replaces round-robin:
|
||||
//! - score(node) = α · latency_p95_ms + β · in_flight_count + γ · error_rate
|
||||
//! - Router picks lowest-scoring eligible node with probability 1 − ε
|
||||
//! - With ε (default 0.05) picks uniformly random for exploration
|
||||
|
||||
use miroir_core::replica_selection::{ReplicaSelectionConfig, ReplicaSelector};
|
||||
use miroir_core::topology::{Group, NodeId};
|
||||
use std::time::Duration;
|
||||
|
||||
/// Create a test selector with default config.
|
||||
fn test_selector() -> ReplicaSelector {
|
||||
ReplicaSelector::new(ReplicaSelectionConfig::default())
|
||||
}
|
||||
|
||||
/// Create a selector with custom half-life for faster test convergence.
|
||||
fn fast_test_selector(half_life_ms: u64) -> ReplicaSelector {
|
||||
ReplicaSelector::new(ReplicaSelectionConfig {
|
||||
ewma_half_life_ms: half_life_ms,
|
||||
exploration_epsilon: 0.05,
|
||||
..Default::default()
|
||||
})
|
||||
}
|
||||
|
||||
/// Create a test group with 3 nodes.
|
||||
fn test_group_3() -> Group {
|
||||
let mut group = Group::new(0);
|
||||
group.add_node(NodeId::new("node-0".to_string()));
|
||||
group.add_node(NodeId::new("node-1".to_string()));
|
||||
group.add_node(NodeId::new("node-2".to_string()));
|
||||
group
|
||||
}
|
||||
|
||||
/// Simulate queries to measure node selection distribution.
|
||||
async fn simulate_selections(
|
||||
selector: &ReplicaSelector,
|
||||
candidates: &[NodeId],
|
||||
group_id: u32,
|
||||
count: usize,
|
||||
) -> std::collections::HashMap<String, usize> {
|
||||
let mut distribution = std::collections::HashMap::new();
|
||||
for _ in 0..count {
|
||||
if let Some(node) = selector.select(candidates, group_id).await {
|
||||
*distribution.entry(node.as_str().to_string()).or_insert(0) += 1;
|
||||
}
|
||||
}
|
||||
distribution
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
// P5.3-A1: Traffic shifts away from degraded node within 2× half-life
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn p5_3_a1_degraded_node_receives_less_traffic() {
|
||||
let selector = fast_test_selector(100); // 100ms half-life for fast test
|
||||
let group = test_group_3();
|
||||
let candidates: Vec<NodeId> = group.nodes().to_vec();
|
||||
|
||||
let node0 = &candidates[0];
|
||||
let node1 = &candidates[1];
|
||||
let node2 = &candidates[2];
|
||||
|
||||
// Initialize all nodes with similar good latency (50ms)
|
||||
selector.record_success(node0, 50.0).await;
|
||||
selector.record_success(node1, 50.0).await;
|
||||
selector.record_success(node2, 50.0).await;
|
||||
|
||||
// Run initial queries to establish baseline
|
||||
let baseline = simulate_selections(&selector, &candidates, 0, 200).await;
|
||||
|
||||
// Verify roughly uniform distribution initially
|
||||
let count0 = *baseline.get("node-0").unwrap_or(&0);
|
||||
let count1 = *baseline.get("node-1").unwrap_or(&0);
|
||||
let count2 = *baseline.get("node-2").unwrap_or(&0);
|
||||
|
||||
// Each node should get ~33% (allow 20-45% for randomness)
|
||||
let _expected = 200 / 3;
|
||||
assert!(
|
||||
(20..=90).contains(&count0),
|
||||
"node-0 baseline count {} out of expected range 20-90",
|
||||
count0
|
||||
);
|
||||
assert!(
|
||||
(20..=90).contains(&count1),
|
||||
"node-1 baseline count {} out of expected range 20-90",
|
||||
count1
|
||||
);
|
||||
assert!(
|
||||
(20..=90).contains(&count2),
|
||||
"node-2 baseline count {} out of expected range 20-90",
|
||||
count2
|
||||
);
|
||||
|
||||
// Induce degradation on node-1: 200ms latency
|
||||
for _ in 0..10 {
|
||||
selector.record_success(node1, 200.0).await;
|
||||
}
|
||||
|
||||
// Wait for EWMA to propagate (2× half-life = 200ms)
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
|
||||
// Run queries after degradation
|
||||
let degraded = simulate_selections(&selector, &candidates, 0, 200).await;
|
||||
|
||||
let degraded_count1 = *degraded.get("node-1").unwrap_or(&0);
|
||||
let degraded_count0 = *degraded.get("node-0").unwrap_or(&0);
|
||||
let degraded_count2 = *degraded.get("node-2").unwrap_or(&0);
|
||||
|
||||
// Node-1 should receive significantly less traffic
|
||||
// With latency_weight=1.0, node-1 score ~200 vs ~50 for others
|
||||
// Expect node-1 to get <15% of traffic
|
||||
assert!(
|
||||
degraded_count1 < 30,
|
||||
"degraded node-1 still receiving too much traffic: {}",
|
||||
degraded_count1
|
||||
);
|
||||
|
||||
// Healthy nodes should receive more traffic
|
||||
assert!(
|
||||
degraded_count0 > 50,
|
||||
"healthy node-0 not receiving enough traffic: {}",
|
||||
degraded_count0
|
||||
);
|
||||
assert!(
|
||||
degraded_count2 > 50,
|
||||
"healthy node-2 not receiving enough traffic: {}",
|
||||
degraded_count2
|
||||
);
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
// P5.3-A2: Node recovers after latency clears; distribution returns to uniform
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn p5_3_a2_degraded_node_recovers() {
|
||||
let selector = fast_test_selector(100); // 100ms half-life
|
||||
let group = test_group_3();
|
||||
let candidates: Vec<NodeId> = group.nodes().to_vec();
|
||||
|
||||
let node0 = &candidates[0];
|
||||
let node1 = &candidates[1];
|
||||
let node2 = &candidates[2];
|
||||
|
||||
// Initialize with good latency
|
||||
selector.record_success(node0, 50.0).await;
|
||||
selector.record_success(node1, 50.0).await;
|
||||
selector.record_success(node2, 50.0).await;
|
||||
|
||||
// Degrade node-1 with 200ms latency
|
||||
for _ in 0..10 {
|
||||
selector.record_success(node1, 200.0).await;
|
||||
}
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
|
||||
// Verify degradation worked
|
||||
let degraded_dist = simulate_selections(&selector, &candidates, 0, 100).await;
|
||||
let degraded_count1 = *degraded_dist.get("node-1").unwrap_or(&0);
|
||||
assert!(
|
||||
degraded_count1 < 20,
|
||||
"node-1 should be degraded, got {} selections",
|
||||
degraded_count1
|
||||
);
|
||||
|
||||
// Clear latency: record good responses for node-1
|
||||
for _ in 0..10 {
|
||||
selector.record_success(node1, 50.0).await;
|
||||
}
|
||||
|
||||
// Wait for recovery (2× half-life)
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
|
||||
// Run queries after recovery
|
||||
let recovered = simulate_selections(&selector, &candidates, 0, 300).await;
|
||||
|
||||
let recovered_count0 = *recovered.get("node-0").unwrap_or(&0);
|
||||
let recovered_count1 = *recovered.get("node-1").unwrap_or(&0);
|
||||
let recovered_count2 = *recovered.get("node-2").unwrap_or(&0);
|
||||
|
||||
// All nodes should have roughly equal distribution (~1/3 each)
|
||||
let expected = 300 / 3;
|
||||
let tolerance = (expected as f64 * 0.4) as usize; // 40% tolerance for exploration
|
||||
|
||||
assert!(
|
||||
(recovered_count1 as isize - expected as isize).abs() <= tolerance as isize,
|
||||
"node-1 recovered count {} not close to expected {} (tolerance {})",
|
||||
recovered_count1, expected, tolerance
|
||||
);
|
||||
assert!(
|
||||
(recovered_count0 as isize - expected as isize).abs() <= tolerance as isize,
|
||||
"node-0 count {} not close to expected {} (tolerance {})",
|
||||
recovered_count0, expected, tolerance
|
||||
);
|
||||
assert!(
|
||||
(recovered_count2 as isize - expected as isize).abs() <= tolerance as isize,
|
||||
"node-2 count {} not close to expected {} (tolerance {})",
|
||||
recovered_count2, expected, tolerance
|
||||
);
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
// P5.3-A3: Exploration ensures degraded node still gets some traffic (ε = 0.05)
|
||||
//
|
||||
// With epsilon=0.05 and 3 nodes, degraded node gets ~1.7% (5%/3) of traffic
|
||||
// via random exploration. This ensures continued monitoring for recovery.
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn p5_3_a3_exploration_samples_degraded_node() {
|
||||
let selector = ReplicaSelector::new(ReplicaSelectionConfig {
|
||||
ewma_half_life_ms: 100,
|
||||
exploration_epsilon: 0.05, // 5% exploration
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let group = test_group_3();
|
||||
let candidates: Vec<NodeId> = group.nodes().to_vec();
|
||||
|
||||
let node0 = &candidates[0];
|
||||
let node1 = &candidates[1];
|
||||
let node2 = &candidates[2];
|
||||
|
||||
// Initialize: node-0 and node-1 healthy, node-2 severely degraded
|
||||
for _ in 0..10 {
|
||||
selector.record_success(node0, 50.0).await;
|
||||
selector.record_success(node1, 50.0).await;
|
||||
selector.record_success(node2, 500.0).await; // Very high latency
|
||||
}
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
|
||||
// Run 1000 queries
|
||||
let dist = simulate_selections(&selector, &candidates, 0, 1000).await;
|
||||
|
||||
let count0 = *dist.get("node-0").unwrap_or(&0);
|
||||
let count1 = *dist.get("node-1").unwrap_or(&0);
|
||||
let count2 = *dist.get("node-2").unwrap_or(&0);
|
||||
|
||||
println!("Distribution: node-0={}, node-1={}, node-2={}", count0, count1, count2);
|
||||
|
||||
// Node-2 is severely degraded but should still get some traffic via exploration
|
||||
// With 5% epsilon and random exploration across 3 nodes:
|
||||
// - 5% of 1000 = 50 explorations total
|
||||
// - 1/3 of explorations go to node-2 ≈ 17 selections
|
||||
// Allow range 5-30 for statistical variance (3 sigma)
|
||||
assert!(
|
||||
(5..=30).contains(&count2),
|
||||
"exploration not working: degraded node-2 got {} selections, expected ~17 (range 5-30)",
|
||||
count2
|
||||
);
|
||||
|
||||
// Healthy nodes should split the remaining ~95%
|
||||
let healthy_total = count0 + count1;
|
||||
assert!(
|
||||
healthy_total >= 900,
|
||||
"healthy nodes didn't get enough traffic: {}",
|
||||
healthy_total
|
||||
);
|
||||
|
||||
// Each healthy node should get roughly half of remaining
|
||||
let expected_healthy = 475; // ~95% / 2
|
||||
let tolerance = 100;
|
||||
assert!(
|
||||
(count0 as isize - expected_healthy).abs() <= tolerance,
|
||||
"node-0 count {} not close to expected {}",
|
||||
count0, expected_healthy
|
||||
);
|
||||
assert!(
|
||||
(count1 as isize - expected_healthy).abs() <= tolerance,
|
||||
"node-1 count {} not close to expected {}",
|
||||
count1, expected_healthy
|
||||
);
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
// P5.3-A4: Round-robin fallback mode works identically to Phase 1 baseline
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn p5_3_a4_round_robin_fallback() {
|
||||
let selector = ReplicaSelector::new(ReplicaSelectionConfig {
|
||||
strategy: "round_robin".to_string(),
|
||||
ewma_half_life_ms: 5000,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let group = test_group_3();
|
||||
let candidates: Vec<NodeId> = group.nodes().to_vec();
|
||||
|
||||
// Test round-robin order
|
||||
let first = selector.select(&candidates, 0).await;
|
||||
let second = selector.select(&candidates, 0).await;
|
||||
let third = selector.select(&candidates, 0).await;
|
||||
let fourth = selector.select(&candidates, 0).await;
|
||||
|
||||
// Should cycle through candidates in order
|
||||
assert_eq!(first, candidates.get(0).cloned());
|
||||
assert_eq!(second, candidates.get(1).cloned());
|
||||
assert_eq!(third, candidates.get(2).cloned());
|
||||
assert_eq!(fourth, candidates.get(0).cloned()); // Wrap around
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
// Additional tests for EWMA behavior
|
||||
// ─────────────────────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn ewma_converges_toward_new_values() {
|
||||
let selector = fast_test_selector(50);
|
||||
let group = test_group_3();
|
||||
let node = &group.nodes()[0];
|
||||
|
||||
// Start with 100ms latency
|
||||
selector.record_success(node, 100.0).await;
|
||||
let metrics1 = selector.get_metrics(node).await;
|
||||
assert!(metrics1.is_some());
|
||||
let latency1 = metrics1.unwrap().latency_p95_ms;
|
||||
assert!((90.0..=110.0).contains(&latency1));
|
||||
|
||||
// Shift to 50ms
|
||||
for _ in 0..5 {
|
||||
selector.record_success(node, 50.0).await;
|
||||
}
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(100)).await;
|
||||
|
||||
let metrics2 = selector.get_metrics(node).await;
|
||||
let latency2 = metrics2.unwrap().latency_p95_ms;
|
||||
|
||||
// Should move toward 50
|
||||
assert!(latency2 < latency1, "EWMA should decrease toward new value");
|
||||
assert!(latency2 > 40.0, "EWMA shouldn't overshoot too much");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn in_flight_count_affects_score() {
|
||||
let selector = test_selector();
|
||||
let group = test_group_3();
|
||||
let candidates: Vec<NodeId> = group.nodes().to_vec();
|
||||
|
||||
let node0 = &candidates[0];
|
||||
let node1 = &candidates[1];
|
||||
|
||||
// Both have same latency initially
|
||||
selector.record_success(node0, 50.0).await;
|
||||
selector.record_success(node1, 50.0).await;
|
||||
|
||||
// Add in-flight requests to node-0
|
||||
for _ in 0..5 {
|
||||
selector.record_request_start(node0).await;
|
||||
}
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
|
||||
// With inflight_weight=2.0, node-0 score should be higher
|
||||
let metrics0 = selector.get_metrics(node0).await.unwrap();
|
||||
let metrics1 = selector.get_metrics(node1).await.unwrap();
|
||||
|
||||
let config = ReplicaSelectionConfig::default();
|
||||
let score0 = metrics0.score(&config);
|
||||
let score1 = metrics1.score(&config);
|
||||
|
||||
assert!(
|
||||
score0 > score1,
|
||||
"node-0 with in-flight requests should have higher score: {} > {}",
|
||||
score0,
|
||||
score1
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn error_rate_affects_score() {
|
||||
let selector = test_selector();
|
||||
let group = test_group_3();
|
||||
let candidates: Vec<NodeId> = group.nodes().to_vec();
|
||||
|
||||
let node0 = &candidates[0];
|
||||
let node1 = &candidates[1];
|
||||
|
||||
// Both have same latency initially
|
||||
selector.record_success(node0, 50.0).await;
|
||||
selector.record_success(node1, 50.0).await;
|
||||
|
||||
// Record errors for node-0
|
||||
for _ in 0..5 {
|
||||
selector.record_error(node0, Some(50.0)).await;
|
||||
}
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
|
||||
// With error_weight=10.0, node-0 score should be higher
|
||||
let metrics0 = selector.get_metrics(node0).await.unwrap();
|
||||
let metrics1 = selector.get_metrics(node1).await.unwrap();
|
||||
|
||||
let config = ReplicaSelectionConfig::default();
|
||||
let score0 = metrics0.score(&config);
|
||||
let score1 = metrics1.score(&config);
|
||||
|
||||
assert!(
|
||||
score0 > score1,
|
||||
"node-0 with errors should have higher score: {} > {}",
|
||||
score0,
|
||||
score1
|
||||
);
|
||||
|
||||
// Verify error_rate is set
|
||||
assert!(metrics0.error_rate > 0.0);
|
||||
assert_eq!(metrics1.error_rate, 0.0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_exploration_epsilon_configurable() {
|
||||
// Test with epsilon = 0.0 (no exploration)
|
||||
let selector_no_explore = ReplicaSelector::new(ReplicaSelectionConfig {
|
||||
exploration_epsilon: 0.0,
|
||||
ewma_half_life_ms: 100,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let group = test_group_3();
|
||||
let candidates: Vec<NodeId> = group.nodes().to_vec();
|
||||
|
||||
let node0 = &candidates[0];
|
||||
let node1 = &candidates[1];
|
||||
let node2 = &candidates[2];
|
||||
|
||||
// Make node-2 severely degraded
|
||||
selector_no_explore.record_success(node0, 50.0).await;
|
||||
selector_no_explore.record_success(node1, 50.0).await;
|
||||
selector_no_explore.record_success(node2, 500.0).await;
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
|
||||
// With no exploration, node-2 should get almost no traffic
|
||||
let dist = simulate_selections(&selector_no_explore, &candidates, 0, 200).await;
|
||||
let count2 = *dist.get("node-2").unwrap_or(&0);
|
||||
|
||||
assert_eq!(
|
||||
count2, 0,
|
||||
"with epsilon=0.0, degraded node should get zero traffic"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_random_strategy() {
|
||||
let selector = ReplicaSelector::new(ReplicaSelectionConfig {
|
||||
strategy: "random".to_string(),
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let group = test_group_3();
|
||||
let candidates: Vec<NodeId> = group.nodes().to_vec();
|
||||
|
||||
// Run many selections to test uniformity
|
||||
let mut counts = std::collections::HashMap::new();
|
||||
for _ in 0..1000 {
|
||||
if let Some(node) = selector.select(&candidates, 0).await {
|
||||
*counts.entry(node.as_str().to_string()).or_insert(0) += 1;
|
||||
}
|
||||
}
|
||||
|
||||
// Each node should get roughly 1/3 of selections
|
||||
let expected = 1000 / 3;
|
||||
for (node, count) in &counts {
|
||||
let diff = (*count as isize - expected as isize).abs();
|
||||
assert!(
|
||||
diff <= 100, // Allow 10% variance
|
||||
"{}: got {} selections, expected ~{}",
|
||||
node, count, expected
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -295,6 +295,14 @@ pub struct Metrics {
|
|||
// ── §13.3 Adaptive replica selection metrics (always present) ──
|
||||
replica_selection_score: GaugeVec,
|
||||
replica_selection_exploration_total: Counter,
|
||||
|
||||
// ── §13.10 Idempotency metrics (always present) ──
|
||||
idempotency_hits_total: CounterVec,
|
||||
idempotency_cache_size: Gauge,
|
||||
|
||||
// ── §13.10 Query coalescing metrics (always present) ──
|
||||
query_coalesce_subscribers_total: Counter,
|
||||
query_coalesce_hits_total: Counter,
|
||||
}
|
||||
|
||||
impl Clone for Metrics {
|
||||
|
|
@ -386,6 +394,10 @@ impl Clone for Metrics {
|
|||
antientropy_last_scan_completed_seconds: self.antientropy_last_scan_completed_seconds.clone(),
|
||||
replica_selection_score: self.replica_selection_score.clone(),
|
||||
replica_selection_exploration_total: self.replica_selection_exploration_total.clone(),
|
||||
idempotency_hits_total: self.idempotency_hits_total.clone(),
|
||||
idempotency_cache_size: self.idempotency_cache_size.clone(),
|
||||
query_coalesce_subscribers_total: self.query_coalesce_subscribers_total.clone(),
|
||||
query_coalesce_hits_total: self.query_coalesce_hits_total.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -966,6 +978,27 @@ impl Metrics {
|
|||
reg!(replica_selection_score);
|
||||
reg!(replica_selection_exploration_total);
|
||||
|
||||
// ── §13.10 Idempotency metrics (always present) ──
|
||||
let idempotency_hits_total = CounterVec::new(
|
||||
Opts::new("miroir_idempotency_hits_total", "Idempotency key hits by outcome"),
|
||||
&["outcome"],
|
||||
).expect("create idempotency_hits_total");
|
||||
let idempotency_cache_size = Gauge::with_opts(
|
||||
Opts::new("miroir_idempotency_cache_size", "Current number of entries in idempotency cache")
|
||||
).expect("create idempotency_cache_size");
|
||||
reg!(idempotency_hits_total);
|
||||
reg!(idempotency_cache_size);
|
||||
|
||||
// ── §13.10 Query coalescing metrics (always present) ──
|
||||
let query_coalesce_subscribers_total = Counter::with_opts(
|
||||
Opts::new("miroir_query_coalesce_subscribers_total", "Total number of subscribers to coalesced queries")
|
||||
).expect("create query_coalesce_subscribers_total");
|
||||
let query_coalesce_hits_total = Counter::with_opts(
|
||||
Opts::new("miroir_query_coalesce_hits_total", "Total number of queries that hit an in-flight coalesced query")
|
||||
).expect("create query_coalesce_hits_total");
|
||||
reg!(query_coalesce_subscribers_total);
|
||||
reg!(query_coalesce_hits_total);
|
||||
|
||||
Self {
|
||||
registry,
|
||||
request_duration,
|
||||
|
|
@ -1053,6 +1086,10 @@ impl Metrics {
|
|||
antientropy_last_scan_completed_seconds,
|
||||
replica_selection_score,
|
||||
replica_selection_exploration_total,
|
||||
idempotency_hits_total,
|
||||
idempotency_cache_size,
|
||||
query_coalesce_subscribers_total,
|
||||
query_coalesce_hits_total,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1763,6 +1800,26 @@ impl Metrics {
|
|||
self.replica_selection_exploration_total.inc();
|
||||
}
|
||||
|
||||
// ── §13.10 Idempotency metrics ──
|
||||
|
||||
pub fn inc_idempotency_hit(&self, outcome: &str) {
|
||||
self.idempotency_hits_total.with_label_values(&[outcome]).inc();
|
||||
}
|
||||
|
||||
pub fn set_idempotency_cache_size(&self, size: u64) {
|
||||
self.idempotency_cache_size.set(size as f64);
|
||||
}
|
||||
|
||||
// ── §13.10 Query coalescing metrics ──
|
||||
|
||||
pub fn inc_query_coalesce_subscribers(&self, count: u64) {
|
||||
self.query_coalesce_subscribers_total.inc_by(count as f64);
|
||||
}
|
||||
|
||||
pub fn inc_query_coalesce_hits(&self) {
|
||||
self.query_coalesce_hits_total.inc();
|
||||
}
|
||||
|
||||
pub fn registry(&self) -> &Registry {
|
||||
&self.registry
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ use miroir_core::{
|
|||
migration::{MigrationConfig, MigrationCoordinator},
|
||||
rebalancer::{MigrationExecutor, Rebalancer, RebalancerConfig, RebalancerMetrics},
|
||||
rebalancer_worker::{RebalancerMetricsCallback, RebalancerWorker, RebalancerWorkerConfig, TopologyChangeEvent},
|
||||
replica_selection::{ReplicaSelector, SelectionObserver},
|
||||
router,
|
||||
scatter::{DeleteByFilterRequest, FetchDocumentsRequest, FetchDocumentsResponse, WriteRequest},
|
||||
task_registry::TaskRegistryImpl,
|
||||
|
|
@ -301,6 +302,23 @@ fn now_ms() -> i64 {
|
|||
.as_millis() as i64
|
||||
}
|
||||
|
||||
/// Metrics observer for replica selection events.
|
||||
///
|
||||
/// Reports selection scores and exploration events to Prometheus.
|
||||
struct ReplicaSelectionMetricsObserver {
|
||||
metrics: super::super::middleware::Metrics,
|
||||
}
|
||||
|
||||
impl SelectionObserver for ReplicaSelectionMetricsObserver {
|
||||
fn report_selection(&self, node_id: &str, score: f64) {
|
||||
self.metrics.set_replica_selection_score(node_id, score);
|
||||
}
|
||||
|
||||
fn report_exploration(&self) {
|
||||
self.metrics.inc_replica_selection_exploration();
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared application state for admin endpoints.
|
||||
#[derive(Clone)]
|
||||
pub struct AppState {
|
||||
|
|
@ -338,6 +356,10 @@ pub struct AppState {
|
|||
pub mode_c_worker: Option<Arc<ModeCWorker>>,
|
||||
/// Adaptive replica selector (plan §13.3).
|
||||
pub replica_selector: Arc<miroir_core::replica_selection::ReplicaSelector>,
|
||||
/// Idempotency cache for write deduplication (plan §13.10).
|
||||
pub idempotency_cache: Arc<miroir_core::idempotency::IdempotencyCache>,
|
||||
/// Query coalescer for read deduplication (plan §13.10).
|
||||
pub query_coalescer: Arc<miroir_core::idempotency::QueryCoalescer>,
|
||||
}
|
||||
|
||||
impl AppState {
|
||||
|
|
@ -633,7 +655,22 @@ impl AppState {
|
|||
alias_registry,
|
||||
leader_election,
|
||||
mode_c_worker,
|
||||
replica_selector: Arc::new(miroir_core::replica_selection::ReplicaSelector::default()),
|
||||
replica_selector: {
|
||||
let config = config.replica_selection.clone();
|
||||
let observer = Arc::new(ReplicaSelectionMetricsObserver {
|
||||
metrics: metrics.clone(),
|
||||
});
|
||||
Arc::new(ReplicaSelector::new_with_observer(config, observer))
|
||||
},
|
||||
idempotency_cache: Arc::new(miroir_core::idempotency::IdempotencyCache::new(
|
||||
config.idempotency.max_cached_keys as usize,
|
||||
config.idempotency.ttl_seconds,
|
||||
)),
|
||||
query_coalescer: Arc::new(miroir_core::idempotency::QueryCoalescer::new(
|
||||
config.query_coalescing.window_ms,
|
||||
config.query_coalescing.max_pending_queries as usize,
|
||||
config.query_coalescing.max_subscribers as usize,
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ pub struct MultiSearchState {
|
|||
pub node_master_key: String,
|
||||
pub metrics: crate::middleware::Metrics,
|
||||
pub alias_registry: Arc<miroir_core::alias::AliasRegistry>,
|
||||
pub replica_selector: Arc<miroir_core::replica_selection::ReplicaSelector>,
|
||||
}
|
||||
|
||||
/// Multi-search request (plan §13.11).
|
||||
|
|
@ -248,17 +249,27 @@ where
|
|||
let config = state.config.clone();
|
||||
let strategy = strategy.clone();
|
||||
let policy = policy;
|
||||
let replica_selector = state.replica_selector.clone();
|
||||
|
||||
async move {
|
||||
let start = Instant::now();
|
||||
|
||||
// Determine if we should use adaptive selection
|
||||
let use_adaptive = config.replica_selection.strategy == "adaptive";
|
||||
let replica_selector_ref = if use_adaptive {
|
||||
Some(replica_selector.as_ref())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// Plan scatter for this query
|
||||
let plan = plan_search_scatter(
|
||||
&topology,
|
||||
0,
|
||||
config.replication_factor as usize,
|
||||
config.shards,
|
||||
);
|
||||
replica_selector_ref,
|
||||
).await;
|
||||
|
||||
// Build search request
|
||||
let filter_value = query.filter.as_ref()
|
||||
|
|
|
|||
|
|
@ -6,7 +6,9 @@ use axum::response::Response;
|
|||
use axum::Json;
|
||||
use miroir_core::api_error::{MeilisearchError, MiroirCode};
|
||||
use miroir_core::config::UnavailableShardPolicy;
|
||||
use miroir_core::idempotency::{QueryFingerprint, canonicalize_json};
|
||||
use miroir_core::merger::ScoreMergeStrategy;
|
||||
use miroir_core::replica_selection::SelectionObserver;
|
||||
use miroir_core::scatter::{
|
||||
dfs_query_then_fetch_search, plan_search_scatter, plan_search_scatter_for_group, plan_search_scatter_with_version_floor, SearchRequest, NodeClient,
|
||||
};
|
||||
|
|
@ -14,12 +16,29 @@ use miroir_core::session_pinning::WaitStrategy;
|
|||
use serde::Deserialize;
|
||||
use serde_json::Value;
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
use std::time::{Duration, Instant};
|
||||
use tracing::{debug, error, info, info_span, instrument, warn};
|
||||
|
||||
use crate::middleware::SessionId;
|
||||
use crate::routes::admin_endpoints::{AppState, parse_rate_limit};
|
||||
|
||||
/// Metrics observer for replica selection events.
|
||||
///
|
||||
/// Reports selection scores and exploration events to Prometheus.
|
||||
struct MetricsObserver {
|
||||
metrics: crate::middleware::Metrics,
|
||||
}
|
||||
|
||||
impl SelectionObserver for MetricsObserver {
|
||||
fn report_selection(&self, node_id: &str, score: f64) {
|
||||
self.metrics.set_replica_selection_score(node_id, score);
|
||||
}
|
||||
|
||||
fn report_exploration(&self) {
|
||||
self.metrics.inc_replica_selection_exploration();
|
||||
}
|
||||
}
|
||||
|
||||
/// Hash a value for logging (obfuscates sensitive data like IPs).
|
||||
fn hash_for_log(value: &str) -> String {
|
||||
use std::hash::{Hash, Hasher};
|
||||
|
|
@ -35,11 +54,16 @@ fn hash_for_log(value: &str) -> String {
|
|||
pub struct ProxyNodeClient {
|
||||
client: Arc<crate::client::HttpClient>,
|
||||
metrics: crate::middleware::Metrics,
|
||||
replica_selector: Option<Arc<miroir_core::replica_selection::ReplicaSelector>>,
|
||||
}
|
||||
|
||||
impl ProxyNodeClient {
|
||||
pub fn new(client: Arc<crate::client::HttpClient>, metrics: crate::middleware::Metrics) -> Self {
|
||||
Self { client, metrics }
|
||||
pub fn new(
|
||||
client: Arc<crate::client::HttpClient>,
|
||||
metrics: crate::middleware::Metrics,
|
||||
replica_selector: Option<Arc<miroir_core::replica_selection::ReplicaSelector>>,
|
||||
) -> Self {
|
||||
Self { client, metrics, replica_selector }
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -286,6 +310,78 @@ async fn search_handler(
|
|||
);
|
||||
}
|
||||
|
||||
// Query coalescing (plan §13.10): Check for identical in-flight queries
|
||||
// Skip for multi-target aliases (each target is different)
|
||||
if state.config.query_coalescing.enabled && resolved_targets.len() == 1 {
|
||||
// Build fingerprint from canonicalized query body + index + settings version
|
||||
let settings_version = state.settings_broadcast.current_version().await;
|
||||
let query_body = serde_json::to_value(&body).unwrap_or(Value::Null);
|
||||
let fingerprint = QueryFingerprint::new(effective_index.clone(), &query_body, settings_version);
|
||||
|
||||
// Try to coalesce with an existing in-flight query
|
||||
if let Some(mut rx) = state.query_coalescer.try_coalesce(fingerprint.clone()).await {
|
||||
// Successfully subscribed to an in-flight query - wait for its result
|
||||
state.metrics.inc_query_coalesce_hits();
|
||||
debug!(
|
||||
index = %effective_index,
|
||||
"query coalesced: waiting for in-flight query result"
|
||||
);
|
||||
|
||||
// Wait for the response (with timeout matching the scatter timeout)
|
||||
let timeout = Duration::from_millis(state.config.scatter.node_timeout_ms);
|
||||
let response_bytes = tokio::time::timeout(timeout, async move {
|
||||
rx.recv().await
|
||||
}).await;
|
||||
|
||||
match response_bytes {
|
||||
Ok(Ok(response_bytes)) => {
|
||||
// Received response from coalesced query
|
||||
let response_body: Value = match serde_json::from_slice(&response_bytes) {
|
||||
Ok(v) => v,
|
||||
Err(e) => {
|
||||
error!(error = %e, "failed to deserialize coalesced query response");
|
||||
return Err(StatusCode::INTERNAL_SERVER_ERROR);
|
||||
}
|
||||
};
|
||||
|
||||
// Build response with appropriate headers
|
||||
let mut response = Response::builder()
|
||||
.status(StatusCode::OK)
|
||||
.header("content-type", "application/json");
|
||||
|
||||
// Add settings headers
|
||||
if state.settings_broadcast.is_in_flight(&effective_index).await {
|
||||
response = response.header("X-Miroir-Settings-Inconsistent", "true");
|
||||
}
|
||||
if settings_version > 0 {
|
||||
response = response.header("X-Miroir-Settings-Version", settings_version.to_string());
|
||||
}
|
||||
|
||||
let response = response
|
||||
.body(axum::body::Body::from(serde_json::to_string(&response_body).unwrap()))
|
||||
.unwrap();
|
||||
|
||||
tracing::info!(
|
||||
target: "miroir.search_coalesced",
|
||||
index = %effective_index,
|
||||
duration_ms = start.elapsed().as_millis() as u64,
|
||||
"coalesced search completed"
|
||||
);
|
||||
|
||||
return Ok(response);
|
||||
}
|
||||
Ok(Err(_)) => {
|
||||
// Channel closed without receiving response - proceed with normal scatter
|
||||
debug!("coalesced query channel closed, proceeding with normal scatter");
|
||||
}
|
||||
Err(_) => {
|
||||
// Timeout waiting for coalesced query - proceed with normal scatter
|
||||
debug!("timeout waiting for coalesced query, proceeding with normal scatter");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Handle multi-target alias fanout (plan §13.7, §13.11, §13.17)
|
||||
// Multi-target aliases (ILM read_alias) require fanning out to all targets
|
||||
// and merging results by _rankingScore
|
||||
|
|
@ -348,8 +444,17 @@ async fn search_handler(
|
|||
rf = state.config.replication_factor,
|
||||
min_settings_version,
|
||||
pinned_group = ?pinned_group,
|
||||
strategy = %state.config.replica_selection.strategy,
|
||||
).entered();
|
||||
|
||||
// Determine if we should use adaptive selection
|
||||
let use_adaptive = state.config.replica_selection.strategy == "adaptive";
|
||||
let replica_selector_ref = if use_adaptive {
|
||||
Some(state.replica_selector.as_ref())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
// Session pinning: if pinned_group is set, use group-specific planning
|
||||
if let Some(group) = pinned_group {
|
||||
match plan_search_scatter_for_group(
|
||||
|
|
@ -358,7 +463,8 @@ async fn search_handler(
|
|||
state.config.replication_factor as usize,
|
||||
state.config.shards,
|
||||
group,
|
||||
) {
|
||||
replica_selector_ref,
|
||||
).await {
|
||||
Some(p) => p,
|
||||
None => {
|
||||
// Pinned group not available - fall back to normal planning
|
||||
|
|
@ -367,7 +473,7 @@ async fn search_handler(
|
|||
pinned_group = group,
|
||||
"pinned group unavailable, falling back to normal routing"
|
||||
);
|
||||
plan_search_scatter(&topo, 0, state.config.replication_factor as usize, state.config.shards)
|
||||
plan_search_scatter(&topo, 0, state.config.replication_factor as usize, state.config.shards, replica_selector_ref).await
|
||||
}
|
||||
}
|
||||
} else if let Some(floor) = min_settings_version {
|
||||
|
|
@ -392,7 +498,8 @@ async fn search_handler(
|
|||
})
|
||||
})
|
||||
},
|
||||
);
|
||||
replica_selector_ref,
|
||||
).await;
|
||||
|
||||
match plan_result {
|
||||
Some(p) => p,
|
||||
|
|
@ -410,7 +517,7 @@ async fn search_handler(
|
|||
}
|
||||
} else {
|
||||
// No version floor requested, use normal planning
|
||||
plan_search_scatter(&topo, 0, state.config.replication_factor as usize, state.config.shards)
|
||||
plan_search_scatter(&topo, 0, state.config.replication_factor as usize, state.config.shards, replica_selector_ref).await
|
||||
}
|
||||
};
|
||||
let node_count = plan.shard_to_node.len() as u64;
|
||||
|
|
@ -441,6 +548,28 @@ async fn search_handler(
|
|||
// Use score-based merge strategy (OP#4: requires global IDF)
|
||||
let strategy = ScoreMergeStrategy::new();
|
||||
|
||||
// Register for query coalescing (plan §13.10) - after try_coalesce, before scatter
|
||||
// Only register if coalescing is enabled and this is a single-target query
|
||||
let (tx, fingerprint) = if state.config.query_coalescing.enabled && resolved_targets.len() == 1 {
|
||||
let settings_version = state.settings_broadcast.current_version().await;
|
||||
let query_body = serde_json::to_value(&body).unwrap_or(Value::Null);
|
||||
let fp = QueryFingerprint::new(effective_index.clone(), &query_body, settings_version);
|
||||
|
||||
match state.query_coalescer.register(fp.clone()).await {
|
||||
Ok(broadcast_tx) => {
|
||||
state.metrics.inc_query_coalesce_subscribers(1); // First subscriber = the query itself
|
||||
(Some(broadcast_tx), Some(fp))
|
||||
}
|
||||
Err(_) => {
|
||||
// Failed to register (too many pending queries) - proceed without coalescing
|
||||
debug!("too many pending queries, proceeding without coalescing registration");
|
||||
(None, None)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
(None, None)
|
||||
};
|
||||
|
||||
// Execute DFS query-then-fetch
|
||||
let mut result = dfs_query_then_fetch_search(
|
||||
plan,
|
||||
|
|
@ -465,17 +594,36 @@ async fn search_handler(
|
|||
}
|
||||
|
||||
// Build response body
|
||||
let mut body = serde_json::json!({
|
||||
let response_body = serde_json::json!({
|
||||
"hits": result.hits,
|
||||
"estimatedTotalHits": result.estimated_total_hits,
|
||||
"processingTimeMs": result.processing_time_ms,
|
||||
});
|
||||
|
||||
// Only include facetDistribution if facets were requested
|
||||
let mut body = response_body.clone();
|
||||
if let Some(facets) = &result.facet_distribution {
|
||||
body["facetDistribution"] = serde_json::to_value(facets).unwrap_or(Value::Null);
|
||||
}
|
||||
|
||||
// Broadcast result to coalesced queries (plan §13.10)
|
||||
if let (Some(broadcast_tx), Some(fp)) = (tx, fingerprint) {
|
||||
let response_bytes = serde_json::to_vec(&body).unwrap_or_default();
|
||||
let subscriber_count = broadcast_tx.receiver_count();
|
||||
if subscriber_count > 1 {
|
||||
// There are other queries waiting for this result - broadcast to them
|
||||
state.metrics.inc_query_coalesce_subscribers(subscriber_count as u64 - 1);
|
||||
let _ = broadcast_tx.send(response_bytes.clone());
|
||||
debug!(
|
||||
index = %effective_index,
|
||||
subscribers = subscriber_count - 1,
|
||||
"broadcast search result to coalesced queries"
|
||||
);
|
||||
}
|
||||
// Unregister the query after broadcasting
|
||||
state.query_coalescer.unregister(&fp).await;
|
||||
}
|
||||
|
||||
// Build response with optional headers
|
||||
let mut response = Response::builder()
|
||||
.status(StatusCode::OK)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue