use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use super::compile::testing::{AliasProjection, bootstrap, env};
use super::lkg::testing::{KEY, cache_path};
use super::status::testing::ManualClock;
use super::*;
use crate::backends::control_plane::{ControlPlaneError, ControlPlaneStore};
use crate::budget::NoBudget;
use crate::desired_state::oracle::InMemoryControlPlane;
use crate::desired_state::{DesiredState, ExpectedRevision, RevisionId, fixtures};
use crate::state::AppState;
use crate::telemetry;
use crate::usage::{UsageFanout, UsageSink};
use async_trait::async_trait;
struct Replica {
store: Arc<InMemoryControlPlane>,
state: AppState,
clock: ManualClock,
reconciler: Arc<Reconciler>,
ledger: Arc<MaterialLedger>,
}
struct ToggleCompiler {
delegate: RevisionCompiler<AliasProjection>,
refuse: Arc<AtomicBool>,
}
#[async_trait]
impl CandidateCompiler for ToggleCompiler {
async fn compile(
&self,
revision: &crate::desired_state::LoadedRevision,
generation: u64,
) -> Result<crate::state::ConfigSnapshot, CompileError> {
if self.refuse.load(Ordering::Acquire) {
return Err(CompileError::Projection {
revision: revision.id(),
source: ProjectionError::Incomplete {
detail: "the test compiler refused this refresh".to_owned(),
},
});
}
self.delegate.compile(revision, generation).await
}
}
impl Replica {
fn serving(store: &Arc<InMemoryControlPlane>) -> Self {
Self::build(store, "openai", None)
}
fn refusing(store: &Arc<InMemoryControlPlane>) -> Self {
Self::build(store, "nonexistent", None)
}
fn converging_policy(store: &Arc<InMemoryControlPlane>) -> Self {
let sinks: Vec<Box<dyn UsageSink>> = Vec::new();
let state = AppState::new(
bootstrap(),
&env(),
UsageFanout::new(sinks),
Box::new(NoBudget),
)
.expect("the bootstrap config is servable");
let clock = ManualClock::new();
let secrets = super::secrets::testing::permissive();
let ledger = Arc::clone(secrets.ledger());
let reconciler = Arc::new(Reconciler::new(
Arc::clone(store) as Arc<dyn ControlPlaneStore>,
Arc::new(RevisionCompiler::with_secrets(
bootstrap(),
env(),
PolicyProjection::over(TenancyProjection),
secrets,
)),
Arc::new(state.clone()),
settings(),
None,
Arc::new(clock.clone()),
));
Self {
store: Arc::clone(store),
state,
clock,
reconciler,
ledger,
}
}
fn with_cache(store: &Arc<InMemoryControlPlane>, cache: LastKnownGood) -> Self {
Self::build(store, "openai", Some(cache))
}
fn with_cache_and_unresolvable_secrets(
store: &Arc<InMemoryControlPlane>,
cache: LastKnownGood,
) -> Self {
Self::assembled(
store,
"openai",
Some(cache),
super::secrets::testing::unavailable(),
)
}
fn with_unresolvable_secrets(store: &Arc<InMemoryControlPlane>) -> Self {
Self::assembled(
store,
"openai",
None,
super::secrets::testing::unavailable(),
)
}
fn build(
store: &Arc<InMemoryControlPlane>,
provider: &'static str,
cache: Option<LastKnownGood>,
) -> Self {
Self::assembled(
store,
provider,
cache,
super::secrets::testing::permissive(),
)
}
fn assembled(
store: &Arc<InMemoryControlPlane>,
provider: &'static str,
cache: Option<LastKnownGood>,
secrets: Arc<SecretMaterialization>,
) -> Self {
let sinks: Vec<Box<dyn UsageSink>> = Vec::new();
let state = AppState::new(
bootstrap(),
&env(),
UsageFanout::new(sinks),
Box::new(NoBudget),
)
.expect("the bootstrap config is servable");
let clock = ManualClock::new();
let ledger = Arc::clone(secrets.ledger());
let compiler = Arc::new(RevisionCompiler::with_secrets(
bootstrap(),
env(),
AliasProjection { provider },
Arc::clone(&secrets),
));
let reconciler = Arc::new(Reconciler::new(
Arc::clone(store) as Arc<dyn ControlPlaneStore>,
compiler,
Arc::new(state.clone()),
settings(),
cache,
Arc::new(clock.clone()),
));
Self {
store: Arc::clone(store),
state,
clock,
reconciler,
ledger,
}
}
fn toggleable(store: &Arc<InMemoryControlPlane>) -> (Self, Arc<AtomicBool>) {
let sinks: Vec<Box<dyn UsageSink>> = Vec::new();
let state = AppState::new(
bootstrap(),
&env(),
UsageFanout::new(sinks),
Box::new(NoBudget),
)
.expect("the bootstrap config is servable");
let clock = ManualClock::new();
let secrets = super::secrets::testing::permissive();
let refuse = Arc::new(AtomicBool::new(false));
let compiler = Arc::new(ToggleCompiler {
delegate: RevisionCompiler::with_secrets(
bootstrap(),
env(),
AliasProjection { provider: "openai" },
secrets.clone(),
),
refuse: Arc::clone(&refuse),
});
let reconciler = Arc::new(Reconciler::new(
Arc::clone(store) as Arc<dyn ControlPlaneStore>,
compiler,
Arc::new(state.clone()),
settings(),
None,
Arc::new(clock.clone()),
));
(
Self {
store: Arc::clone(store),
state,
clock,
reconciler,
ledger: Arc::clone(secrets.ledger()),
},
refuse,
)
}
fn report(&self) -> RevisionReport {
self.reconciler.report()
}
fn served_aliases(&self) -> Vec<String> {
self.state
.config()
.config
.model
.iter()
.map(|model| model.name.clone())
.collect()
}
fn generation(&self) -> u64 {
self.state.config().generation
}
}
fn settings() -> ConvergenceSettings {
ConvergenceSettings {
poll_interval: Duration::from_millis(100),
target: Duration::from_secs(1),
backoff: BackoffPolicy {
initial: Duration::from_millis(100),
max: Duration::from_secs(4),
multiplier: 2,
},
}
}
fn control_plane() -> Arc<InMemoryControlPlane> {
Arc::new(InMemoryControlPlane::new())
}
async fn publish(
store: &InMemoryControlPlane,
key: &str,
expected: ExpectedRevision,
state: DesiredState,
) -> RevisionId {
store
.publish_revision(fixtures::candidate(expected, key, state))
.await
.expect("the candidate is valid")
.id
}
#[tokio::test]
async fn a_replica_converges_to_the_desired_revision_and_serves_it() {
let store = control_plane();
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::serving(&store);
assert_eq!(replica.generation(), 0, "the boot snapshot is generation 0");
let outcome = replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(
matches!(
outcome,
Outcome::Published { revision, generation, .. }
if revision == published && generation == 1
),
"{outcome:?}"
);
let report = replica.report();
assert!(report.converged());
assert_eq!(report.desired, Some(published));
assert_eq!(report.loaded, Some(published));
assert_eq!(report.active, Some(published));
assert_eq!(report.source, Some(SnapshotSource::ControlPlane));
assert_eq!(report.lag, Duration::ZERO);
assert_eq!(report.consecutive_failures, 0);
assert!(report.last_rejection.is_none());
assert!(replica.served_aliases().contains(&"fast".to_owned()));
assert_eq!(replica.generation(), 1);
}
#[tokio::test]
async fn a_converged_replica_does_not_republish_what_it_is_already_serving() {
let store = control_plane();
publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::serving(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
for _ in 0..5 {
let outcome = replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(
matches!(outcome, Outcome::AlreadyConverged { .. }),
"{outcome:?}"
);
}
assert_eq!(replica.generation(), 1, "no spurious republication");
}
#[tokio::test]
async fn a_force_refresh_recompiles_and_publishes_the_same_revision() {
let store = control_plane();
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::serving(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let outcome = replica
.reconciler
.force_refresh_once(telemetry::CONVERGENCE_NOTIFIED)
.await;
assert!(
matches!(
outcome,
Outcome::Published { revision, generation, .. }
if revision == published && generation == 2
),
"{outcome:?}"
);
assert_eq!(replica.report().active, Some(published));
assert_eq!(
replica.generation(),
2,
"the refreshed snapshot was published"
);
}
#[tokio::test]
async fn a_failed_force_refresh_does_not_publish() {
let store = control_plane();
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let (replica, refuse) = Replica::toggleable(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let before = replica.served_aliases();
refuse.store(true, Ordering::Release);
let outcome = replica
.reconciler
.force_refresh_once(telemetry::CONVERGENCE_NOTIFIED)
.await;
assert!(
matches!(
outcome,
Outcome::Rejected { revision, reason }
if revision == Some(published) && reason == "projection"
),
"{outcome:?}"
);
let report = replica.report();
assert_eq!(report.active, Some(published));
assert_eq!(report.generation, 1);
assert_eq!(replica.generation(), 1);
assert_eq!(replica.served_aliases(), before);
}
#[tokio::test]
async fn a_refused_force_refresh_is_retried_by_the_next_convergence() {
let store = control_plane();
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let (replica, refuse) = Replica::toggleable(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
refuse.store(true, Ordering::Release);
let refused = replica
.reconciler
.force_refresh_once(telemetry::CONVERGENCE_NOTIFIED)
.await;
assert!(matches!(refused, Outcome::Rejected { .. }), "{refused:?}");
refuse.store(false, Ordering::Release);
let outcome = replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(
matches!(
outcome,
Outcome::Published { revision, generation, .. }
if revision == published && generation == 2
),
"{outcome:?}"
);
assert_eq!(
replica.generation(),
2,
"the refresh eventually took effect"
);
}
#[tokio::test]
async fn a_newer_revision_replaces_the_previous_one_wholesale() {
let store = control_plane();
let first = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::serving(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(replica.served_aliases().contains(&"fast".to_owned()));
let second = publish(
&store,
"second",
ExpectedRevision::Exactly(first),
fixtures::state_with_renamed_alias(),
)
.await;
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert_eq!(replica.report().active, Some(second));
assert_eq!(replica.generation(), 2);
let aliases = replica.served_aliases();
assert!(aliases.contains(&"quick".to_owned()), "{aliases:?}");
assert!(
!aliases.contains(&"fast".to_owned()),
"the previous revision's alias is gone, not merged: {aliases:?}"
);
}
#[tokio::test(start_paused = true)]
async fn a_missed_notification_is_recovered_by_the_poll() {
let store = control_plane();
let replica = Replica::serving(&store);
let signal = Arc::new(ChangeSignal::new());
let (stop, stopped) = tokio::sync::oneshot::channel::<()>();
let loop_reconciler = Arc::clone(&replica.reconciler);
let task = tokio::spawn(async move {
loop_reconciler
.run(Arc::new(ChangeSignal::new()), async {
let _ = stopped.await;
})
.await;
});
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
assert_eq!(replica.report().active, None, "not converged yet");
advance_until(&replica, |report| report.active == Some(published)).await;
assert!(replica.served_aliases().contains(&"fast".to_owned()));
drop(signal);
let _ = stop.send(());
task.await.expect("the loop stops when shutdown completes");
}
#[tokio::test(start_paused = true)]
async fn a_notification_converges_before_the_next_poll() {
let store = control_plane();
let replica = Replica::serving(&store);
let signal = Arc::new(ChangeSignal::new());
let (stop, stopped) = tokio::sync::oneshot::channel::<()>();
let loop_reconciler = Arc::clone(&replica.reconciler);
let listener = Arc::clone(&signal);
let task = tokio::spawn(async move {
loop_reconciler
.run(listener, async {
let _ = stopped.await;
})
.await;
});
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
signal.notify();
for _ in 0..32 {
if replica.report().active == Some(published) {
break;
}
tokio::task::yield_now().await;
}
assert_eq!(
replica.report().active,
Some(published),
"a notification converges without the clock advancing"
);
let _ = stop.send(());
task.await.expect("the loop stops");
}
#[tokio::test(start_paused = true)]
async fn force_refresh_signals_coalesce_and_refresh_the_run_loop() {
let store = control_plane();
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::serving(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let signal = Arc::new(ChangeSignal::new());
let (stop, stopped) = tokio::sync::oneshot::channel::<()>();
let loop_reconciler = Arc::clone(&replica.reconciler);
let listener = Arc::clone(&signal);
let task = tokio::spawn(async move {
loop_reconciler
.run(listener, async {
let _ = stopped.await;
})
.await;
});
signal.force_refresh();
signal.force_refresh();
for _ in 0..32 {
if replica.generation() == 2 {
break;
}
tokio::task::yield_now().await;
}
assert_eq!(replica.report().active, Some(published));
assert_eq!(
replica.generation(),
2,
"the run loop performed one refresh"
);
for _ in 0..32 {
tokio::task::yield_now().await;
}
assert_eq!(
replica.generation(),
2,
"coalesced signals do not cause a second publication"
);
let _ = stop.send(());
task.await.expect("the loop stops");
}
#[tokio::test]
async fn a_revision_that_fails_the_boot_gate_publishes_nothing() {
let store = control_plane();
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::refusing(&store);
let before = replica.served_aliases();
let outcome = replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(
matches!(
outcome,
Outcome::Rejected { revision, reason }
if revision == Some(published) && reason == "validation"
),
"{outcome:?}"
);
let report = replica.report();
assert!(!report.converged());
assert_eq!(report.desired, Some(published));
assert_eq!(report.loaded, None, "a refused candidate never loaded");
assert_eq!(report.active, None);
assert_eq!(report.generation, 0);
assert_eq!(report.consecutive_failures, 1);
let rejection = report.last_rejection.expect("a reason is reported");
assert_eq!(rejection.reason, "validation");
assert!(
rejection.detail.contains("undefined provider"),
"{}",
rejection.detail
);
assert_eq!(replica.generation(), 0);
assert_eq!(replica.served_aliases(), before);
}
#[tokio::test]
async fn a_policy_these_backends_cannot_enforce_is_refused_before_anything_is_published() {
use crate::desired_state::Slug;
use crate::policy::fixtures::body;
let store = control_plane();
let mut state = fixtures::state();
let policy = body(
crate::desired_state::policy::PolicyScope::Project {
tenant: fixtures::tenant_id(1),
project: fixtures::project_id(2),
},
1,
1_000,
);
state
.insert(policy.version(Slug::parse("policy").expect("a valid slug")))
.expect("a policy resource");
let published = publish(&store, "first", ExpectedRevision::Empty, state).await;
let replica = Replica::converging_policy(&store);
let outcome = replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(
matches!(
outcome,
Outcome::Rejected { revision, reason }
if revision == Some(published) && reason == "unsupported"
),
"{outcome:?}"
);
let report = replica.report();
assert_eq!(report.active, None, "nothing was published");
assert_eq!(replica.generation(), 0);
assert!(
replica.state.policy().draining().is_empty(),
"a refused candidate leaves no generation half-installed"
);
assert_eq!(
replica
.state
.policy()
.active("platform")
.budget
.expect("the file's limits still govern")
.subject_microdollars,
bootstrap().budget.limit_microdollars
);
}
#[tokio::test]
async fn a_revision_this_build_cannot_read_is_refused_as_an_incompatibility() {
let store = control_plane();
let first = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::serving(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let serving = replica.served_aliases();
let second = publish(
&store,
"second",
ExpectedRevision::Exactly(first),
fixtures::state_with_renamed_alias(),
)
.await;
store.rewrite_version(fixtures::legacy_tenant(1, "acme"));
let outcome = replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(
matches!(
outcome,
Outcome::Rejected { revision, reason }
if revision == Some(second) && reason == "incompatible"
),
"{outcome:?}"
);
let report = replica.report();
let rejection = report.last_rejection.expect("a reason is reported");
assert_eq!(rejection.reason, "incompatible");
assert!(
rejection.detail.contains("not compatible with this build"),
"{}",
rejection.detail
);
assert_eq!(report.active, Some(first));
assert_eq!(report.desired, Some(second));
assert_eq!(replica.generation(), 1);
assert_eq!(replica.served_aliases(), serving);
}
#[tokio::test]
async fn a_price_book_this_build_cannot_bill_leaves_the_previous_pricing_active() {
let store = control_plane();
let body = fixtures::approved_price_book();
let first = publish(
&store,
"first",
ExpectedRevision::Empty,
fixtures::state_with_price_book(&body),
)
.await;
let replica = Replica::serving(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let priced = replica
.state
.config()
.pricing()
.expect("the converged snapshot carries pricing")
.clone();
assert!(priced.is_approved());
let second = publish(
&store,
"second",
ExpectedRevision::Exactly(first),
fixtures::state_with_price_book(&body),
)
.await;
store.rewrite_version(fixtures::unbillable_price_book(7, "baseline"));
let outcome = replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(
matches!(
outcome,
Outcome::Rejected { revision, reason }
if revision == Some(second) && reason == "incompatible"
),
"{outcome:?}"
);
let report = replica.report();
assert_eq!(report.active, Some(first));
assert_eq!(report.desired, Some(second));
assert_eq!(replica.generation(), 1);
let serving = replica
.state
.config()
.pricing()
.expect("pricing is retained, not dropped")
.clone();
assert_eq!(serving.book(), priced.book());
assert_eq!(serving.checksum(), priced.checksum());
assert_eq!(
serving.targets().collect::<Vec<_>>(),
priced.targets().collect::<Vec<_>>()
);
}
#[tokio::test]
async fn an_outage_keeps_the_previous_revision_serving_and_reports_the_lag() {
let store = control_plane();
let first = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::serving(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let serving = replica.served_aliases();
let second = publish(
&store,
"second",
ExpectedRevision::Exactly(first),
fixtures::state_with_renamed_alias(),
)
.await;
store.set_unavailable(true);
for attempt in 1..=3 {
let outcome = replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(
matches!(
outcome,
Outcome::Rejected {
reason: "unavailable",
..
}
),
"{outcome:?}"
);
assert_eq!(replica.report().consecutive_failures, attempt);
replica.clock.advance(Duration::from_secs(2));
}
let lagging = replica.report();
assert_eq!(
lagging.active,
Some(first),
"still serving the old revision"
);
assert_eq!(
lagging.desired,
Some(first),
"desired is unreadable, not lost"
);
assert_eq!(replica.generation(), 1);
assert_eq!(replica.served_aliases(), serving);
assert_eq!(
lagging.last_rejection.map(|rejection| rejection.reason),
Some("unavailable")
);
store.set_unavailable(false);
let outcome = replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(
matches!(outcome, Outcome::Published { revision, .. } if revision == second),
"{outcome:?}"
);
let recovered = replica.report();
assert!(recovered.converged());
assert_eq!(recovered.consecutive_failures, 0);
assert!(
recovered.last_rejection.is_none(),
"a success clears the reported failure"
);
assert_eq!(recovered.lag, Duration::ZERO);
}
#[tokio::test]
async fn lag_grows_while_a_replica_cannot_reach_the_desired_revision() {
let store = control_plane();
let first = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::refusing(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert_eq!(
replica.report().lag,
Duration::ZERO,
"measured, not guessed"
);
replica.clock.advance(Duration::from_secs(45));
let report = replica.report();
assert_eq!(report.desired, Some(first));
assert_eq!(report.lag, Duration::from_secs(45));
assert!(
report.lag > settings().target,
"past the documented convergence target, which is what an alert fires on"
);
}
#[tokio::test]
async fn repeated_failures_widen_the_retry_delay_up_to_the_ceiling() {
let store = control_plane();
publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
store.set_unavailable(true);
let replica = Replica::serving(&store);
let mut delays = Vec::new();
let mut backoff = Backoff::new(settings().backoff);
for _ in 0..8 {
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
delays.push(backoff.fail());
}
assert_eq!(
delays,
vec![
Duration::from_millis(100),
Duration::from_millis(200),
Duration::from_millis(400),
Duration::from_millis(800),
Duration::from_millis(1_600),
Duration::from_millis(3_200),
Duration::from_secs(4),
Duration::from_secs(4),
],
"exponential, then saturated at the ceiling"
);
assert_eq!(replica.report().consecutive_failures, 8);
}
#[tokio::test(start_paused = true)]
async fn a_replica_whose_control_plane_is_down_does_not_hot_loop() {
let store = control_plane();
publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
store.set_unavailable(true);
let replica = Replica::serving(&store);
let (stop, stopped) = tokio::sync::oneshot::channel::<()>();
let loop_reconciler = Arc::clone(&replica.reconciler);
let task = tokio::spawn(async move {
loop_reconciler
.run(Arc::new(ChangeSignal::new()), async {
let _ = stopped.await;
})
.await;
});
for _ in 0..30 {
tokio::time::advance(Duration::from_secs(1)).await;
tokio::task::yield_now().await;
}
let failures = replica.report().consecutive_failures;
assert!(
(2..=16).contains(&failures),
"bounded retries over 30s of outage, got {failures}"
);
assert_eq!(replica.generation(), 0, "nothing was published");
let _ = stop.send(());
task.await.expect("the loop stops");
}
#[tokio::test]
async fn a_cold_boot_during_an_outage_restores_the_signed_snapshot() {
let store = control_plane();
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let path = cache_path("cold-boot");
let warm = Replica::with_cache(
&store,
LastKnownGood::new(&path, KEY).expect("a long enough key"),
);
warm.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(path.exists(), "converging exported the cache");
store.set_unavailable(true);
let cold = Replica::with_cache(
&store,
LastKnownGood::new(&path, KEY).expect("a long enough key"),
);
let restored = cold.reconciler.bootstrap().await.expect("the cache serves");
assert_eq!(restored, published);
let report = cold.report();
assert_eq!(report.active, Some(published));
assert_eq!(report.source, Some(SnapshotSource::LastKnownGood));
assert!(cold.served_aliases().contains(&"fast".to_owned()));
assert_eq!(cold.generation(), 1);
store.set_unavailable(false);
let second = publish(
&store,
"second",
ExpectedRevision::Exactly(published),
fixtures::state_with_renamed_alias(),
)
.await;
cold.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let converged = cold.report();
assert_eq!(converged.active, Some(second));
assert_eq!(converged.source, Some(SnapshotSource::ControlPlane));
let _ = std::fs::remove_file(&path);
}
#[tokio::test]
async fn a_cold_boot_onto_an_unreadable_revision_restores_the_signed_snapshot() {
let store = control_plane();
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let path = cache_path("cold-boot-incompatible");
let warm = Replica::with_cache(
&store,
LastKnownGood::new(&path, KEY).expect("a long enough key"),
);
warm.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(path.exists(), "converging exported the cache");
let second = publish(
&store,
"second",
ExpectedRevision::Exactly(published),
fixtures::state_with_renamed_alias(),
)
.await;
store.rewrite_version(fixtures::legacy_tenant(1, "acme"));
let cold = Replica::with_cache(
&store,
LastKnownGood::new(&path, KEY).expect("a long enough key"),
);
let restored = cold
.reconciler
.bootstrap()
.await
.expect("an unreadable revision is not a reason to refuse to start");
assert_eq!(restored, published);
let report = cold.report();
assert_eq!(report.active, Some(published));
assert_eq!(report.source, Some(SnapshotSource::LastKnownGood));
assert_eq!(report.generation, 1);
assert!(cold.served_aliases().contains(&"fast".to_owned()));
let outcome = cold
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(
matches!(
outcome,
Outcome::Rejected { revision, reason }
if revision == Some(second) && reason == "incompatible"
),
"{outcome:?}"
);
assert_eq!(cold.report().active, Some(published));
let _ = std::fs::remove_file(&path);
}
#[tokio::test]
async fn a_cold_boot_onto_an_unreadable_revision_without_a_cache_refuses_to_start() {
let store = control_plane();
publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
store.rewrite_version(fixtures::legacy_tenant(1, "acme"));
let error = Replica::serving(&store)
.reconciler
.bootstrap()
.await
.expect_err("there is nothing to serve");
assert!(
matches!(
error,
BootstrapError::Store {
source: ControlPlaneError::Incompatible { .. }
}
),
"an unreadable revision is named as such, not as an outage: {error}"
);
}
#[tokio::test]
async fn a_cold_boot_whose_cache_a_newer_build_wrote_is_refused_as_a_skew() {
let store = control_plane();
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let path = cache_path("cold-boot-newer-cache");
let warm = Replica::with_cache(
&store,
LastKnownGood::new(&path, KEY).expect("a long enough key"),
);
warm.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let cached = LastKnownGood::new(&path, KEY).expect("a long enough key");
let readable = cached.load().expect("reads back").expect("a cache exists");
cached
.export_unassembled(readable.manifest(), &fixtures::state_with_legacy_tenant())
.expect("a newer build's export");
assert_eq!(readable.manifest().id, published);
store.rewrite_version(fixtures::legacy_tenant(1, "acme"));
let error = Replica::with_cache(
&store,
LastKnownGood::new(&path, KEY).expect("a long enough key"),
)
.reconciler
.bootstrap()
.await
.expect_err("there is nothing this build can serve");
assert!(
matches!(
error,
BootstrapError::Store {
source: ControlPlaneError::Incompatible { .. }
}
),
"the skew is named, not the cache that faithfully recorded it: {error}"
);
let _ = std::fs::remove_file(&path);
}
#[tokio::test]
async fn a_cold_boot_during_an_outage_without_a_cache_refuses_to_start() {
let store = control_plane();
publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
store.set_unavailable(true);
let replica = Replica::serving(&store);
let error = replica
.reconciler
.bootstrap()
.await
.expect_err("there is nothing to serve");
assert!(
matches!(error, BootstrapError::Unavailable { .. }),
"{error}"
);
assert_eq!(replica.generation(), 0);
}
#[tokio::test]
async fn a_cold_boot_against_an_empty_control_plane_refuses_to_start() {
let store = control_plane();
let replica = Replica::serving(&store);
let error = replica
.reconciler
.bootstrap()
.await
.expect_err("nothing has been published");
assert!(matches!(error, BootstrapError::Empty), "{error}");
}
#[tokio::test]
async fn a_boot_revision_that_does_not_compile_is_fatal_even_with_a_cache() {
let store = control_plane();
publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let path = cache_path("boot-rejected");
let replica = Replica::build(
&store,
"nonexistent",
Some(LastKnownGood::new(&path, KEY).expect("a long enough key")),
);
let error = replica
.reconciler
.bootstrap()
.await
.expect_err("the desired revision is unservable");
assert!(matches!(error, BootstrapError::Rejected { .. }), "{error}");
assert!(!path.exists(), "nothing unservable was ever cached");
}
#[tokio::test]
async fn an_in_flight_request_keeps_the_revision_it_started_under() {
let store = control_plane();
let first = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::serving(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let in_flight = replica.state.config();
assert_eq!(in_flight.generation, 1);
assert!(
in_flight
.config
.model
.iter()
.any(|model| model.name == "fast")
);
publish(
&store,
"second",
ExpectedRevision::Exactly(first),
fixtures::state_with_renamed_alias(),
)
.await;
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert_eq!(replica.generation(), 2);
assert!(replica.served_aliases().contains(&"quick".to_owned()));
assert_eq!(in_flight.generation, 1);
assert!(
in_flight
.config
.model
.iter()
.any(|model| model.name == "fast"),
"the alias the request resolved is still resolvable"
);
assert!(
!in_flight
.config
.model
.iter()
.any(|model| model.name == "quick"),
"and the newer revision has not leaked into it"
);
assert_eq!(replica.state.config().generation, 2);
}
async fn advance_until(replica: &Replica, predicate: impl Fn(&RevisionReport) -> bool) {
for _ in 0..64 {
if predicate(&replica.reconciler.report()) {
return;
}
tokio::time::advance(Duration::from_millis(50)).await;
tokio::task::yield_now().await;
}
panic!(
"the replica never reached the expected state: {:?}",
replica.reconciler.report()
);
}
#[tokio::test]
async fn a_published_revision_holds_the_material_it_was_compiled_against() {
let store = control_plane();
publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::serving(&store);
assert!(
replica.ledger.is_empty(),
"the boot snapshot has no typed credentials"
);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let held = replica.ledger.retained();
let expected = required_secrets(&fixtures::state());
assert_eq!(held, expected, "the published revision's versions are held");
assert_eq!(
replica.state.config().secrets().references(),
expected,
"and the snapshot serving requests is what holds them"
);
}
#[tokio::test]
async fn a_rotation_overlaps_versions_until_the_previous_snapshot_is_gone() {
let store = control_plane();
let first = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::serving(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let old = required_secrets(&fixtures::state());
let in_flight = replica.state.config();
publish(
&store,
"rotated",
ExpectedRevision::Exactly(first),
fixtures::state_with_rotated_credential(),
)
.await;
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let new = required_secrets(&fixtures::state_with_rotated_credential());
assert_ne!(new, old, "a rotation pins a different exact version");
assert_eq!(
replica.state.config().secrets().references(),
new,
"new requests authenticate with the rotated version"
);
for reference in &old {
assert!(
replica.ledger.holds(*reference),
"the in-flight request's version is still live: {reference}"
);
assert!(in_flight.secrets().get(*reference).is_some());
}
drop(in_flight);
for reference in &old {
assert!(
!replica.ledger.holds(*reference),
"the superseded version is zeroized once nothing serves it: {reference}"
);
}
assert_eq!(replica.ledger.retained(), new);
}
#[tokio::test]
async fn a_candidate_whose_material_does_not_resolve_leaves_the_previous_revision_serving() {
let store = control_plane();
let first = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let replica = Replica::serving(&store);
replica
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
let serving = required_secrets(&fixtures::state());
assert_eq!(replica.ledger.retained(), serving);
let unavailable = Replica::with_unresolvable_secrets(&store);
let rotated = publish(
&store,
"rotated",
ExpectedRevision::Exactly(first),
fixtures::state_with_rotated_credential(),
)
.await;
let outcome = unavailable
.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(matches!(outcome, Outcome::Rejected { .. }), "{outcome:?}");
let report = unavailable.report();
assert_eq!(report.desired, Some(rotated));
assert_ne!(report.active, Some(rotated), "the candidate is not active");
let rejection = report.last_rejection.expect("a recorded rejection");
assert_eq!(rejection.reason, "secret");
assert!(
!rejection.detail.contains(super::secrets::testing::MATERIAL),
"the rejection names references, not material: {}",
rejection.detail
);
assert!(
unavailable.ledger.is_empty(),
"a refused candidate retains nothing"
);
assert_eq!(replica.generation(), 1);
assert!(replica.served_aliases().contains(&"fast".to_owned()));
assert_eq!(replica.ledger.retained(), serving);
}
#[tokio::test]
async fn a_cold_boot_whose_material_does_not_resolve_refuses_to_start() {
let store = control_plane();
let published = publish(&store, "first", ExpectedRevision::Empty, fixtures::state()).await;
let path = cache_path("cold-boot-secrets");
let warm = Replica::with_cache(
&store,
LastKnownGood::new(&path, KEY).expect("a long enough key"),
);
warm.reconciler
.converge_once(telemetry::CONVERGENCE_POLLED)
.await;
assert!(path.exists(), "converging exported the cache");
let cold = Replica::with_cache_and_unresolvable_secrets(
&store,
LastKnownGood::new(&path, KEY).expect("a long enough key"),
);
let error = cold
.reconciler
.bootstrap()
.await
.expect_err("material this replica cannot unwrap is not servable");
assert!(
matches!(error, BootstrapError::Rejected { .. }),
"a refusal, not a restore: {error:?}"
);
let rendered = error.to_string();
assert!(
!rendered.contains(super::secrets::testing::MATERIAL),
"the refusal names references, not material: {rendered}"
);
assert_eq!(cold.generation(), 0, "nothing was published");
assert_eq!(cold.report().active, None);
assert!(cold.ledger.is_empty(), "a refused boot retains nothing");
let recovered = Replica::with_cache(
&store,
LastKnownGood::new(&path, KEY).expect("a long enough key"),
);
assert_eq!(
recovered.reconciler.bootstrap().await.expect("it boots"),
published
);
}
fn required_secrets(state: &DesiredState) -> Vec<crate::desired_state::secrets::SecretRef> {
let mut references: Vec<_> = crate::desired_state::credentials::Credentials::of(state)
.expect("readable fixture credentials")
.required_secrets()
.map(|(_, reference)| reference)
.collect();
references.sort_unstable();
references
}