i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-10-06 01:17:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/main' into codex/fix-antigravity-quota

Browse Source
This commit is contained in:
ZheFox
2026-09-03 13:24:39 +08:00
parent 40a5e1470d 058660ec2e
commit 587486ab0c
94 changed files with 1926 additions and 3765 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aether-data/runtime/src/lifecycle/backfill/mysql.rs
+12 -2
View File
@@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet};
use sqlx::{
migrate::{Migrate, MigrateError, Migrator},
query, Connection, MySqlConnection, Row,
query, query_scalar, Connection, MySqlConnection, Row,
};
use tracing::{error, info, warn};
@@ -11,6 +11,7 @@ use crate::driver::mysql::MysqlPool;
static BACKFILL_MIGRATOR: Migrator = sqlx::migrate!("./backfills/mysql");
const SCHEMA_BACKFILLS_TABLE_EXISTS_SQL: &str = "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = 'schema_backfills'";
const ENSURE_SCHEMA_BACKFILLS_TABLE_SQL: &str = r#"
CREATE TABLE IF NOT EXISTS schema_backfills (
version BIGINT NOT NULL,
@@ -161,12 +162,21 @@ async fn run_backfills_locked(conn: &mut MySqlConnection) -> Result<(), MigrateE
async fn pending_backfills_locked(
conn: &mut MySqlConnection,
) -> Result<Vec<PendingBackfillInfo>, MigrateError> {
ensure_schema_backfills_table(conn).await?;
if !schema_backfills_table_exists(conn).await? {
return Ok(pending_backfills_from_applied(&[]));
}
let applied_backfills = list_applied_backfills(conn).await?;
validate_applied_backfills(&applied_backfills)?;
Ok(pending_backfills_from_applied(&applied_backfills))
}
async fn schema_backfills_table_exists(conn: &mut MySqlConnection) -> Result<bool, MigrateError> {
let total: i64 = query_scalar(SCHEMA_BACKFILLS_TABLE_EXISTS_SQL)
.fetch_one(&mut *conn)
.await?;
Ok(total > 0)
}
async fn ensure_schema_backfills_table(conn: &mut MySqlConnection) -> Result<(), MigrateError> {
query(ENSURE_SCHEMA_BACKFILLS_TABLE_SQL)
.execute(&mut *conn)
crates/aether-data/runtime/src/lifecycle/backfill/postgres.rs
+11 -5
View File
@@ -157,17 +157,23 @@ async fn run_backfills_locked(conn: &mut PgConnection) -> Result<(), MigrateErro
async fn pending_backfills_locked(
conn: &mut PgConnection,
) -> Result<Vec<PendingBackfillInfo>, MigrateError> {
ensure_schema_backfills_table(conn).await?;
if !schema_backfills_table_exists(conn).await? {
return Ok(pending_backfills_from_applied(&[]));
}
let applied_backfills = list_applied_backfills(conn).await?;
validate_applied_backfills(&applied_backfills)?;
Ok(pending_backfills_from_applied(&applied_backfills))
}
async fn ensure_schema_backfills_table(conn: &mut PgConnection) -> Result<(), MigrateError> {
let exists: bool = query_scalar(SCHEMA_BACKFILLS_TABLE_EXISTS_SQL)
async fn schema_backfills_table_exists(conn: &mut PgConnection) -> Result<bool, MigrateError> {
query_scalar(SCHEMA_BACKFILLS_TABLE_EXISTS_SQL)
.fetch_one(&mut *conn)
.await?;
if exists {
.await
.map_err(Into::into)
}
async fn ensure_schema_backfills_table(conn: &mut PgConnection) -> Result<(), MigrateError> {
if schema_backfills_table_exists(conn).await? {
return Ok(());
}
query(ENSURE_SCHEMA_BACKFILLS_TABLE_SQL)
crates/aether-data/runtime/src/lifecycle/backfill/sqlite.rs
+13 -2
View File
@@ -2,7 +2,7 @@ use std::collections::{HashMap, HashSet};
use sqlx::{
migrate::{Migrate, MigrateError, Migrator},
query, Connection, Row, SqliteConnection,
query, query_scalar, Connection, Row, SqliteConnection,
};
use tracing::{error, info, warn};
@@ -11,6 +11,8 @@ use crate::driver::sqlite::SqlitePool;
static BACKFILL_MIGRATOR: Migrator = sqlx::migrate!("./backfills/sqlite");
const SCHEMA_BACKFILLS_TABLE_EXISTS_SQL: &str =
"SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'schema_backfills'";
const ENSURE_SCHEMA_BACKFILLS_TABLE_SQL: &str = r#"
CREATE TABLE IF NOT EXISTS schema_backfills (
version INTEGER NOT NULL PRIMARY KEY,
@@ -162,12 +164,21 @@ async fn run_backfills_locked(conn: &mut SqliteConnection) -> Result<(), Migrate
async fn pending_backfills_locked(
conn: &mut SqliteConnection,
) -> Result<Vec<PendingBackfillInfo>, MigrateError> {
ensure_schema_backfills_table(conn).await?;
if !schema_backfills_table_exists(conn).await? {
return Ok(pending_backfills_from_applied(&[]));
}
let applied_backfills = list_applied_backfills(conn).await?;
validate_applied_backfills(&applied_backfills)?;
Ok(pending_backfills_from_applied(&applied_backfills))
}
async fn schema_backfills_table_exists(conn: &mut SqliteConnection) -> Result<bool, MigrateError> {
let total: i64 = query_scalar(SCHEMA_BACKFILLS_TABLE_EXISTS_SQL)
.fetch_one(&mut *conn)
.await?;
Ok(total > 0)
}
async fn ensure_schema_backfills_table(conn: &mut SqliteConnection) -> Result<(), MigrateError> {
query(ENSURE_SCHEMA_BACKFILLS_TABLE_SQL)
.execute(&mut *conn)
crates/aether-data/runtime/src/lifecycle/backfill/tests.rs
+25
View File
@@ -310,6 +310,31 @@ INSERT INTO usage_settlement_snapshots (
}
}
#[tokio::test]
async fn pending_sqlite_backfills_does_not_create_tracking_table() {
let pool = sqlx::sqlite::SqlitePoolOptions::new()
.max_connections(1)
.connect("sqlite::memory:")
.await
.expect("sqlite backfill status pool should connect");
run_sqlite_migrations(&pool)
.await
.expect("sqlite schema should migrate");
let pending = pending_sqlite_backfills(&pool)
.await
.expect("sqlite pending backfills should load");
assert!(!pending.is_empty());
let tracking_tables: i64 = query_scalar(
"SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'schema_backfills'",
)
.fetch_one(&pool)
.await
.expect("sqlite tracking table state should load");
assert_eq!(tracking_tables, 0);
}
#[tokio::test]
async fn sqlite_backfills_apply_portable_repairs_and_record_versions() {
let pool = sqlx::sqlite::SqlitePoolOptions::new()
crates/aether-data/runtime/src/lifecycle/bootstrap/postgres.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use tracing::info;
// Generated by build.rs from schema/bootstrap/postgres.
pub(crate) static EMPTY_DATABASE_SNAPSHOT_SQL: &str =
include_str!(concat!(env!("OUT_DIR"), "/empty_database_snapshot.sql"));
pub(crate) const EMPTY_DATABASE_SNAPSHOT_CUTOFF_VERSION: i64 = 20260821000000;
pub(crate) const EMPTY_DATABASE_SNAPSHOT_CUTOFF_VERSION: i64 = 20260903000000;
const PUBLIC_BASE_TABLE_COUNT_SQL: &str = r#"
SELECT COUNT(*)::BIGINT
crates/aether-data/runtime/src/lifecycle/migrate/tests.rs
+4
View File
@@ -412,6 +412,7 @@ fn empty_database_snapshot_covers_current_cutoff_versions() {
20260727000000,
20260731000000,
20260821000000,
20260903000000,
]
);
}
@@ -1066,6 +1067,7 @@ fn mysql_and_sqlite_migrations_include_enabled_incrementals() {
20260727000000,
20260731000000,
20260821000000,
20260903000000,
]
);
assert_eq!(
@@ -1101,6 +1103,7 @@ fn mysql_and_sqlite_migrations_include_enabled_incrementals() {
20260727000000,
20260731000000,
20260821000000,
20260903000000,
]
);
}
@@ -2208,6 +2211,7 @@ fn pending_migrations_from_applied_skips_versions_already_applied() {
20260727000000,
20260731000000,
20260821000000,
20260903000000,
]
);
}
crates/aether-data/runtime/src/repository/routing_profiles/memory.rs
+53 -1
View File
@@ -59,7 +59,14 @@ impl RoutingGroupReadRepository for InMemoryRoutingGroupRepository {
.values()
.cloned()
.collect::<Vec<_>>();
groups.sort_by(|left, right| left.name.cmp(&right.name).then(left.id.cmp(&right.id)));
groups.sort_by(|left, right| {
right
.enabled
.cmp(&left.enabled)
.then(left.sort_order.cmp(&right.sort_order))
.then(left.name.cmp(&right.name))
.then(left.id.cmp(&right.id))
});
Ok(groups)
}
@@ -273,6 +280,7 @@ mod tests {
description: None,
enabled: true,
is_system_default: true,
sort_order: 0,
config_json: json!({}),
version: 1,
created_at: 1,
@@ -320,6 +328,49 @@ mod tests {
);
}
#[tokio::test]
async fn lists_enabled_groups_first_and_respects_sort_order() {
let repository = InMemoryRoutingGroupRepository::default();
for (id, enabled, sort_order) in [
("disabled-first", false, 0),
("enabled-second", true, 20),
("enabled-first", true, 10),
] {
repository
.create_routing_group(CreateRoutingGroupRecord {
id: id.to_string(),
name: id.to_string(),
description: None,
enabled,
is_system_default: false,
sort_order,
config_json: json!({}),
version: 1,
created_at: 1,
updated_at: 1,
published_at: None,
})
.await
.expect("group should store");
}
let ids = repository
.list_routing_groups()
.await
.expect("groups should list")
.into_iter()
.map(|group| group.id)
.collect::<Vec<_>>();
assert_eq!(
ids,
vec![
"enabled-first".to_string(),
"enabled-second".to_string(),
"disabled-first".to_string(),
]
);
}
#[tokio::test]
async fn keeps_system_and_subject_defaults_unique() {
let repository = InMemoryRoutingGroupRepository::default();
@@ -414,6 +465,7 @@ mod tests {
description: None,
enabled: true,
is_system_default,
sort_order: 0,
config_json: json!({}),
version: 1,
created_at: 1,