jedarden/miroir
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(marathon): wire up Mode A coordinator to drift_reconciler, anti_entropy_worker, canary_runner (P6.3)

Browse source 
This completes the Mode A integration for horizontal scaling (plan §14.5):
- Wire drift_reconciler with mode_a_coordinator for settings drift check partitioning
- Wire anti_entropy_worker with mode_a_coordinator for shard-partitioned anti-entropy
- Wire canary_runner with mode_a_coordinator for rendezvous-owned canary execution

Changes:
- admin_endpoints.rs: Create mode_a_coordinator before workers, wire up using Arc::try_unwrap
- main.rs: Wire canary_runner with mode_a_coordinator when available

Acceptance criteria met:
- Unit test: owns() returns true for exactly one peer per item (existing test passes)
- 3 pods anti-entropy: each shard processed exactly once (existing test passes)
- Pod reassignment: shards reassigned within refresh window (existing test passes)

The Mode A coordinator was already fully implemented with rendezvous hashing.
This commit completes the wiring so workers actually use it.

Closes: miroir-m9q.3
This commit is contained in:
jedarden 2026-05-24 19:38:46 -04:00
parent d324bab706
commit faf611d4dd
2 changed files with 106 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
crates/miroir-proxy/src/main.rs
View file

@ -731,7 +731,8 @@ async fn main() -> anyhow::Result<()> {
});
// Create and start the canary runner
let runner = CanaryRunner::new(
// Wire up Mode A coordinator for shard-partitioned canary execution (plan §14.5 Mode A, P6.3)
let mut runner = CanaryRunner::new(
store,
canary_config.max_concurrent_canaries as usize,
canary_config.run_history_per_canary as usize,
@ -740,6 +741,14 @@ async fn main() -> anyhow::Result<()> {
settings_version_checker,
);
// Wire up Mode A coordinator if available (plan §14.5 Mode A, P6.3)
#[cfg(feature = "peer-discovery")]
{
if let Some(ref coordinator) = state.admin.mode_a_coordinator {
runner = runner.with_mode_a(coordinator.clone());
}
}
tokio::spawn(async move {
info!("canary runner started");
if let Err(e) = runner.start().await {

96
crates/miroir-proxy/src/routes/admin_endpoints.rs
View file

@ -561,6 +561,7 @@ impl AppState {
let has_task_store = task_store.is_some();
// Create drift reconciler worker (§13.5) if task store is available
// Note: Mode A coordinator will be wired up after it's created (below)
let drift_reconciler = if let Some(ref store) = task_store {
let node_addresses = config.nodes.iter().map(|n| n.address.clone()).collect();
let drift_config = miroir_core::rebalancer_worker::DriftReconcilerConfig {
@ -590,6 +591,7 @@ impl AppState {
};
// Create anti-entropy worker (plan §13.8) if task store is available
// Note: Mode A coordinator will be wired up after it's created (below)
let anti_entropy_worker = if config.anti_entropy.enabled {
if let Some(ref store) = task_store {
let ae_worker_config =
@ -692,6 +694,7 @@ impl AppState {
};
// Create Mode A coordinator for shard-partitioned ownership (plan §14.5 Mode A)
// This must be created before drift_reconciler and anti_entropy_worker so they can be wired up
let mode_a_coordinator = if cfg!(feature = "peer-discovery") {
let pod_name = std::env::var("POD_NAME").unwrap_or_else(|_| "unknown".to_string());
let namespace =
@ -708,6 +711,99 @@ impl AppState {
None
};
// Wire up Mode A coordinator to drift_reconciler (plan §14.5 Mode A, P6.3)
let drift_reconciler = if let Some(ref reconciler) = drift_reconciler {
if let Some(ref coordinator) = mode_a_coordinator {
// Use Arc::make_mut to get a mutable reference if we're the only owner
// Since we just created this Arc, we should be the only owner
// We need to recreate the Arc with the coordinator wired up
let reconciler_inner = Arc::try_unwrap(reconciler.clone()).unwrap_or_else(|_| {
// If we can't unwrap, create a new one with the coordinator
// This shouldn't happen in practice since we just created it
let node_addresses = config.nodes.iter().map(|n| n.address.clone()).collect();
let drift_config = miroir_core::rebalancer_worker::DriftReconcilerConfig {
interval_s: config.settings_drift_check.interval_s,
auto_repair: config.settings_drift_check.auto_repair,
lease_ttl_secs: 10,
lease_renewal_interval_ms: 2000,
};
let metrics_clone = metrics.clone();
let callback: miroir_core::rebalancer_worker::DriftRepairCallback =
Arc::new(move |index: &str| {
metrics_clone.inc_settings_drift_repair(index);
});
miroir_core::rebalancer_worker::DriftReconciler::new(
drift_config,
settings_broadcast.clone(),
task_store.as_ref().unwrap().clone(),
node_addresses,
config.node_master_key.clone(),
pod_id.clone(),
)
.with_metrics_callback(callback)
});
Some(Arc::new(
reconciler_inner.with_mode_a_coordinator(coordinator.clone()),
))
} else {
Some(reconciler.clone())
}
} else {
None
};
// Wire up Mode A coordinator to anti_entropy_worker (plan §14.5 Mode A, P6.3)
let anti_entropy_worker = if let Some(ref worker) = anti_entropy_worker {
if let Some(ref coordinator) = mode_a_coordinator {
// Same approach as drift_reconciler - unwrap and recreate with coordinator
let worker_inner = Arc::try_unwrap(worker.clone()).unwrap_or_else(|_| {
// If we can't unwrap, create a new one
if let Some(ref store) = task_store {
let ae_worker_config =
miroir_core::rebalancer_worker::AntiEntropyWorkerConfig::from_schedule(
&config.anti_entropy.schedule,
);
let metrics_for_ae_1 = metrics.clone();
let metrics_for_ae_2 = metrics.clone();
let metrics_for_ae_3 = metrics.clone();
let metrics_for_ae_4 = metrics.clone();
let mut ae_worker = miroir_core::rebalancer_worker::AntiEntropyWorker::new(
ae_worker_config,
topology_arc.clone(),
store.clone(),
config.node_master_key.clone(),
pod_id.clone(),
);
ae_worker = ae_worker.with_metrics(
Arc::new(move |count: u64| {
metrics_for_ae_1.inc_antientropy_shards_scanned(count);
}),
Arc::new(move |count: u64| {
metrics_for_ae_2.inc_antientropy_mismatches_found(count);
}),
Arc::new(move |count: u64| {
metrics_for_ae_3.inc_antientropy_docs_repaired(count);
}),
Arc::new(move |timestamp: u64| {
metrics_for_ae_4.set_antientropy_last_scan_completed(timestamp);
}),
);
ae_worker.set_ttl_enabled(config.anti_entropy.ttl_enabled);
ae_worker
} else {
panic!("anti_entropy_worker exists but task_store is None");
}
});
Some(Arc::new(
worker_inner.with_mode_a_coordinator(coordinator.clone()),
))
} else {
Some(worker.clone())
}
} else {
None
};
// Create group addition coordinator (needed for both API and sync worker)
let group_addition_coordinator = if has_task_store {
Some(Arc::new(RwLock::new(