Skip to content

Fix nightly rolling-upgrade rehearsal #478

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hardbyte merged 1 commit into from
Aug 30, 2026
Merged

Fix nightly rolling-upgrade rehearsal #478

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion .github/workflows/nightly-chaos.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -374,7 +374,13 @@ jobs:
# ─── Notify on failure ───────────────────────────────────
notify-failure:
name: Create issue on failure
needs: [tla-storage, postgres-failover-smoke, rust-nightly, python-nightly]
needs:
- tla-storage
- compat-matrix
- postgres-failover-smoke
- rolling-upgrade-rehearsal
- rust-nightly
- python-nightly
if: failure()
runs-on: ubuntu-latest
timeout-minutes: 5
Expand All @@ -391,7 +397,9 @@ jobs:

**Failed jobs:**
- tla-storage: ${{ needs.tla-storage.result }}
- compat-matrix: ${{ needs.compat-matrix.result }}
- postgres-failover-smoke: ${{ needs.postgres-failover-smoke.result }}
- rolling-upgrade-rehearsal: ${{ needs.rolling-upgrade-rehearsal.result }}
- rust-nightly: ${{ needs.rust-nightly.result }}
- python-nightly: ${{ needs.python-nightly.result }}

Expand Down
39 changes: 27 additions & 12 deletions awa-worker/src/client.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1327,18 +1327,15 @@ impl Client {
/// when an active schema simply predates this binary's migrations (a
/// binary-first rolling upgrade). The operator fix differs, so name the
/// actual cause.
async fn queue_storage_not_ready_error(
&self,
fn queue_storage_not_ready_error(
schema: &str,
role_hint: &str,
schema_version: i32,
) -> awa_model::AwaError {
let version = awa_model::migrations::current_version_readonly(&self.pool)
.await
.unwrap_or(0);
if version > 0 && version < awa_model::migrations::CURRENT_VERSION {
if schema_version > 0 && schema_version < awa_model::migrations::CURRENT_VERSION {
return awa_model::AwaError::Validation(format!(
"queue storage schema '{schema}' is missing relations this runtime requires: \
schema version {version} is older than this binary's version {}; \
schema version {schema_version} is older than this binary's version {}; \
run `awa migrate` before starting this runtime",
awa_model::migrations::CURRENT_VERSION
));
Expand All @@ -1356,7 +1353,7 @@ impl Client {

let status = transition::status(&self.pool).await?;
let expected_schema = Self::expected_queue_storage_schema(&status)?;
let prepared_schema_ready = if let Some(schema) = expected_schema.as_deref() {
let mut prepared_schema_ready = if let Some(schema) = expected_schema.as_deref() {
if runtime.store.schema() != schema {
return Err(awa_model::AwaError::Validation(format!(
"queue storage runtime configured for schema '{}' but transition state requires '{}'",
Expand All @@ -1368,6 +1365,18 @@ impl Client {
} else {
false
};
let mut observed_schema_version = 0;
if let Some(schema) = expected_schema.as_deref() {
if !prepared_schema_ready {
observed_schema_version =
awa_model::migrations::current_version_readonly(&self.pool).await?;
if observed_schema_version == awa_model::migrations::CURRENT_VERSION {
// The migration may have committed between the readiness and version probes.
prepared_schema_ready =
transition::queue_storage_schema_ready(&self.pool, schema).await?;
}
}
}

match self.transition_role {
TransitionWorkerRole::CanonicalDrain => Ok(RuntimeStorage::Canonical),
Expand All @@ -1378,16 +1387,22 @@ impl Client {
)
})?;
if !prepared_schema_ready {
return Err(self
.queue_storage_not_ready_error(&schema, "queue-storage-target")
.await);
return Err(Self::queue_storage_not_ready_error(
&schema,
"queue-storage-target",
observed_schema_version,
));
}
Ok(RuntimeStorage::QueueStorage(runtime.clone()))
}
TransitionWorkerRole::Auto => {
if let Some(schema) = expected_schema.as_deref() {
if !prepared_schema_ready {
return Err(self.queue_storage_not_ready_error(schema, "0.6").await);
return Err(Self::queue_storage_not_ready_error(
schema,
"0.6",
observed_schema_version,
));
}
}

Expand Down
31 changes: 17 additions & 14 deletions awa/tests/rolling_upgrade_rehearsal_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
//! Released-artifact rolling-upgrade rehearsal (#427).
//!
//! This first cell exercises migrate-first operation from the latest released
//! 0.6 patch to the current v043 schema while traffic remains live. The old
//! These cells exercise migration and rolling operation from the latest
//! released 0.6 patch to the current schema while traffic remains live. The old
//! worker is a PyPI wheel installed by the workflow, not a source checkout.

use async_trait::async_trait;
Expand Down Expand Up @@ -829,22 +829,25 @@ async fn test_migrate_first_mixed_fleet_flip_and_fence() {
wait_for_accepted(&producer, 25, Duration::from_secs(20)).await;
let old_before_migration = old_worker.completed().len();

// Apply v041-v043 while the released runtime and producer remain live.
// Apply all current migrations while the released runtime and producer remain live.
migrations::run(&pool)
.await
.expect("current migrations must accept a live 0.6.3 runtime");
assert_eq!(schema_version(&pool).await, 43);
assert_eq!(schema_version(&pool).await, migrations::CURRENT_VERSION);
let authority: String =
sqlx::query_scalar("SELECT authority FROM awa.ring_cursor_authority WHERE singleton")
.fetch_one(&pool)
.await
.expect("read ring authority");
assert_eq!(authority, "columns");
report.phase("migrated_to_v043_columns_authority");
report.phase("migrated_to_current_columns_authority");

let current_completed = Arc::new(Mutex::new(HashSet::new()));
let current = current_client(pool.clone(), &queue, current_completed.clone());
current.start().await.expect("start current worker on v043");
current
.start()
.await
.expect("start current worker on current schema");
wait_for_mixed_fleet(
&old_worker,
old_before_migration,
Expand Down Expand Up @@ -981,8 +984,8 @@ async fn test_migrate_first_deadline_rescue_resumes_with_current_leader() {
migrations::run(&pool)
.await
.expect("migrate with released deadline worker live");
assert_eq!(schema_version(&pool).await, 43);
report.phase("migrated_to_v043");
assert_eq!(schema_version(&pool).await, migrations::CURRENT_VERSION);
report.phase("migrated_to_current");

let inserted = insert_with(
&pool,
Expand Down Expand Up @@ -1092,7 +1095,7 @@ async fn test_migrate_first_deadline_rescue_resumes_with_current_leader() {
}

/// Binary-first upgrade order: the current binary is deployed before
/// v041-v043 apply. The current runtime requires relations those migrations
/// the current migrations apply. The current runtime requires relations they
/// add, so on the released artifact's v040 schema it must refuse startup
/// loudly — naming pending migrations, not silently degrading — while the
/// released fleet keeps draining traffic. The roll completes only after the
Expand Down Expand Up @@ -1154,7 +1157,7 @@ async fn test_binary_first_current_refuses_v040_then_rolls_after_migration() {
migrations::run(&pool)
.await
.expect("migration must apply with the released fleet live");
assert_eq!(schema_version(&pool).await, 43);
assert_eq!(schema_version(&pool).await, migrations::CURRENT_VERSION);
let authority: String =
sqlx::query_scalar("SELECT authority FROM awa.ring_cursor_authority WHERE singleton")
.fetch_one(&pool)
Expand Down Expand Up @@ -1288,7 +1291,7 @@ async fn test_migrate_first_full_workload_reconciles_designed_outcomes() {
migrations::run(&pool)
.await
.expect("migration must apply with the released fleet and banked backlog live");
assert_eq!(schema_version(&pool).await, 43);
assert_eq!(schema_version(&pool).await, migrations::CURRENT_VERSION);

let simple_completed = Arc::new(Mutex::new(HashSet::new()));
let current = workload_client(
Expand All @@ -1300,7 +1303,7 @@ async fn test_migrate_first_full_workload_reconciles_designed_outcomes() {
current
.start()
.await
.expect("start current workload runtime on v043");
.expect("start current workload runtime on current schema");
wait_for_mixed_fleet(
&old_worker,
old_before_migration,
Expand Down Expand Up @@ -1496,7 +1499,7 @@ async fn test_migrate_first_full_workload_reconciles_designed_outcomes() {
/// Overlapped upgrade order: the current deployment and the migration race
/// under live released traffic. A rolling deployment keeps restarting its new
/// pods, so the current runtime start-retries through its schema refusal and
/// must come up unaided the moment v041-v043 commit. Both versions must then
/// must come up unaided the moment the current migrations commit. Both versions must then
/// claim and complete work concurrently before any flip.
#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore = "requires a released awa-pg 0.6.3 environment"]
Expand Down Expand Up @@ -1549,7 +1552,7 @@ async fn test_overlap_current_roll_races_live_migration() {
let (current, migration_result) = tokio::join!(start_retry, migrations::run(&pool));
migration_result.expect("migration must apply with both fleets and the producer live");
report.phase("migration_committed_during_roll");
assert_eq!(schema_version(&pool).await, 43);
assert_eq!(schema_version(&pool).await, migrations::CURRENT_VERSION);

// Both versions claim and complete concurrently after migration, before
// any flip.
Expand Down