refactor(store): persist warmup progress in runtime settings

Replace warmup slot history with a monotonic UTC cursor and consolidate unreleased scheduling migrations.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-01 23:21:48 +08:00
co-authored by Codex
parent 90e241f74b
commit 45e72682e6
10 changed files with 120 additions and 51 deletions
@@ -936,7 +936,7 @@ fn invalid_refresh_policy(operation: &'static str) -> ProviderStoreError {
}
pub trait ProviderRuntimePolicyPort: Send + Sync {
/// 原子领取每日时间槽,跨进程重启与重复本地时刻保持幂等。
/// 原子推进预热执行游标,重启或时钟回拨后不重复领取已消费的时刻。
fn claim_warmup_slot<'a>(
&'a self,
_timezone: crate::time::DeploymentTimeZone,
@@ -289,25 +289,20 @@ impl ProviderRuntimePolicyPort for PgRuntimeSettingsRepository {
slot: chrono::NaiveDateTime,
) -> futures::future::BoxFuture<'a, Result<bool, ProviderStoreError>> {
Box::pin(async move {
let mut transaction = self
.pool
.begin()
.await
.map_err(|_| provider_unavailable("claim warmup slot"))?;
// 去重只需覆盖近期开关、重启与回拨,保留七天后由下一次领取回收。
sqlx::query(
"delete from account_warmup_slots where claimed_at < now() - interval '7 days'",
let slot = timezone.resolve_local(slot).ok_or_else(|| {
ProviderStoreError::new(ProviderStoreErrorKind::InvalidData, "resolve warmup slot")
})?;
// 执行游标只向前推进;设置保存不覆盖它,领取也不发布新的配置版本。
let claimed = sqlx::query(
"update runtime_settings set account_warmup_cursor = $1
where id = 1 and (account_warmup_cursor is null or account_warmup_cursor < $1)",
)
.execute(&mut *transaction)
.bind(slot)
.execute(&self.pool)
.await
.map_err(|_| provider_unavailable("cleanup warmup slots"))?;
let claimed = sqlx::query("insert into account_warmup_slots (timezone, local_slot) values ($1, $2) on conflict do nothing")
.bind(timezone.name()).bind(slot).execute(&mut *transaction).await
.map_err(|_| provider_unavailable("claim warmup slot"))?.rows_affected() == 1;
transaction
.commit()
.await
.map_err(|_| provider_unavailable("commit warmup slot"))?;
.map_err(|_| provider_unavailable("claim warmup slot"))?
.rows_affected()
== 1;
Ok(claimed)
})
}
@@ -234,7 +234,6 @@ async fn connect_and_migrate_should_apply_all_migrations_once_and_reopen_cleanly
"_sqlx_migrations",
"account_group_accounts",
"account_groups",
"account_warmup_slots",
"admin_audit_events",
"admin_users",
"authorization_receipts",
@@ -904,9 +904,9 @@ async fn control_plane_replacement_commits_one_writer_per_revision_and_preserves
}
#[tokio::test]
async fn warmup_slots_survive_restart_and_serialize_competing_claims() {
async fn warmup_cursor_survives_restart_and_settings_updates_without_rewinding() {
use gateway_core::{provider_ports::ProviderRuntimePolicyPort, time::DeploymentTimeZone};
let Some(database) = TestDatabase::create("warmup_slots").await else {
let Some(database) = TestDatabase::create("warmup_cursor").await else {
return;
};
let zone = DeploymentTimeZone::default();
@@ -917,11 +917,15 @@ async fn warmup_slots_survive_restart_and_serialize_competing_claims() {
.unwrap();
let first = PgRuntimeSettingsRepository::new(database.pool.clone());
let second = PgRuntimeSettingsRepository::new(database.pool.clone());
let before = first.load_runtime_settings().await.unwrap();
let (a, b) = tokio::join!(
first.claim_warmup_slot(zone, slot),
second.claim_warmup_slot(zone, slot)
);
assert_ne!(a.unwrap(), b.unwrap(), "one leader claims the local minute");
assert_ne!(a.unwrap(), b.unwrap(), "the local minute is claimed once");
let after = first.load_runtime_settings().await.unwrap();
assert_eq!(after.config_revision, before.config_revision);
assert_eq!(after.updated_at, before.updated_at);
let restarted = PgRuntimeSettingsRepository::new(database.pool.clone());
assert!(!restarted.claim_warmup_slot(zone, slot).await.unwrap());
assert!(
@@ -934,15 +938,104 @@ async fn warmup_slots_survive_restart_and_serialize_competing_claims() {
!restarted.claim_warmup_slot(zone, slot).await.unwrap(),
"another slot cannot erase prior deduplication"
);
sqlx::query("update account_warmup_slots set claimed_at = now() - interval '8 days'")
.execute(&database.pool)
restarted
.update_runtime_settings(settings_with_margin(3600))
.await
.unwrap();
assert!(restarted.claim_warmup_slot(zone, slot).await.unwrap());
let count: i64 = sqlx::query_scalar("select count(*) from account_warmup_slots")
.fetch_one(&database.pool)
assert!(!restarted.claim_warmup_slot(zone, slot).await.unwrap());
assert!(
!restarted
.claim_warmup_slot(zone, slot - TimeDelta::hours(1))
.await
.unwrap(),
"clock rollback cannot replay an earlier time slot"
);
assert!(
restarted
.claim_warmup_slot(zone, slot + TimeDelta::days(1))
.await
.unwrap()
);
database.close().await;
}
#[tokio::test]
async fn runtime_scheduling_upgrade_preserves_settings_without_a_warmup_table() {
let Some(database) = TestDatabase::create_through("runtime_scheduling_upgrade", 20).await
else {
return;
};
sqlx::query(
"update runtime_settings set rotation_strategy = 'sticky',
account_warmup_enabled = true, account_warmup_schedule_time = '08:00,13:00',
account_warmup_model = 'test-model' where id = 1",
)
.execute(&database.pool)
.await
.unwrap();
super::TEST_MIGRATOR.run(&database.pool).await.unwrap();
let warmup_table: Option<String> =
sqlx::query_scalar("select to_regclass('account_warmup_slots')::text")
.fetch_one(&database.pool)
.await
.unwrap();
assert!(warmup_table.is_none());
let cursor: Option<DateTime<Utc>> =
sqlx::query_scalar("select account_warmup_cursor from runtime_settings where id = 1")
.fetch_one(&database.pool)
.await
.unwrap();
assert!(cursor.is_none());
let settings = PgRuntimeSettingsRepository::new(database.pool.clone())
.load_runtime_settings()
.await
.unwrap();
assert_eq!(count, 1, "expired deduplication state is reclaimed");
assert_eq!(settings.rotation_strategy, "sticky");
assert!(settings.account_warmup_enabled);
assert_eq!(settings.account_warmup_schedule_time, "08:00,13:00");
assert_eq!(settings.account_warmup_model.as_deref(), Some("test-model"));
assert_eq!(settings.openai_guardian_reserved_concurrency, 0);
database.close().await;
}
#[tokio::test]
async fn warmup_cursor_resolves_dst_and_deduplicates_across_timezones() {
use gateway_core::{provider_ports::ProviderRuntimePolicyPort, time::DeploymentTimeZone};
let Some(database) = TestDatabase::create("warmup_cursor_dst").await else {
return;
};
let repository = PgRuntimeSettingsRepository::new(database.pool.clone());
let zone: DeploymentTimeZone = "America/New_York".parse().unwrap();
let slot = chrono::NaiveDate::from_ymd_opt(2026, 11, 1)
.unwrap()
.and_hms_opt(1, 30, 0)
.unwrap();
assert!(repository.claim_warmup_slot(zone, slot).await.unwrap());
let cursor: DateTime<Utc> =
sqlx::query_scalar("select account_warmup_cursor from runtime_settings where id = 1")
.fetch_one(&database.pool)
.await
.unwrap();
assert_eq!(
cursor,
"2026-11-01T05:30:00Z".parse::<DateTime<Utc>>().unwrap()
);
assert!(!repository.claim_warmup_slot(zone, slot).await.unwrap());
// 回拨中的重复本地时刻取较早一次,不能误挡后一分的正常调度。
let next = slot + TimeDelta::minutes(1);
assert!(repository.claim_warmup_slot(zone, next).await.unwrap());
let utc: DeploymentTimeZone = "UTC".parse().unwrap();
assert!(
!repository
.claim_warmup_slot(utc, zone.resolve_local(next).unwrap().naive_utc())
.await
.unwrap(),
"changing the deployment timezone cannot reclaim the same instant"
);
let missing = chrono::NaiveDate::from_ymd_opt(2026, 3, 8)
.unwrap()
.and_hms_opt(2, 30, 0)
.unwrap();
assert!(repository.claim_warmup_slot(zone, missing).await.is_err());
database.close().await;
}
@@ -429,18 +429,6 @@ impl ScheduledTask for OpenAiWarmupTask {
tracing::warn!(error = %error, "OpenAI account warmup cycle failed");
}
}
let local_after = self.timezone.local(chrono::Utc::now());
if local_after.date_naive() == local_now.date_naive()
&& local_after.hour() == hour
&& local_after.minute() == minute
{
// 持有本轮 leader lease 到时间槽结束,避免其他实例在同一分钟再次执行。
let hold = Duration::from_secs(u64::from(61 - local_after.second()));
tokio::select! {
() = context.cancellation().cancelled() => return Ok(()),
() = tokio::time::sleep(hold) => {}
}
}
Ok(())
})
}
+1 -2
View File
@@ -18,5 +18,4 @@ fcfae912ddefd54551ebf8e7e733442519d2862d2977134d548ebc1e75c1dc34 0017_unlimited
229c70296764b94fe69c3f19889e566fa290c6c44c64970efe96d7fd2a0008d5 0018_plugins.sql
40a4d4b16e6ec2c2c659980d416b089fc170acd2255aaaf751c4a04c3790a941 0019_plugin_resources.sql
e8c1339166ac02fb2d4a077424ee5f64f6dc20eb4d3f8ec4fcb09c0f2a0e61f6 0020_plugin_trust.sql
338037806358b23dc8b62a39e73b04ce4173c4556383e5df19b7e8c7123924ed 0021_account_warmup_slots.sql
3b6924605e7e70c074d409611d13f17625f98fae4e9dbe8012a6b3dc3aa1679f 0022_guardian_reserved_concurrency.sql
22faf2646168275150b1786c0f463f9ff2ca52463a272479e4fcc6f46c8cfa37 0021_runtime_scheduling.sql
@@ -1,7 +0,0 @@
-- 每日预激活槽在执行前领取,进程重启与夏令时回拨不会重复发送上游请求。
create table account_warmup_slots (
timezone text not null,
local_slot timestamp without time zone not null,
claimed_at timestamptz not null default now(),
primary key (timezone, local_slot)
);
@@ -1,3 +1,4 @@
alter table runtime_settings
add column account_warmup_cursor timestamptz,
add column openai_guardian_reserved_concurrency bigint not null default 0
check (openai_guardian_reserved_concurrency between 0 and 4294967295);
+1 -1
View File
@@ -1244,7 +1244,7 @@ accountWarmupModel
定时账号预热默认关闭。`accountWarmupScheduleTime` 使用部署时区中的 `HH:MM`,
多个时段以逗号分隔,默认 `08:00`;`accountWarmupModel` 默认 `null`,开启前必须显式选择模型。
任务面向可用的 OpenAI OAuth 账号,跳过周额度耗尽及五小时窗口距离重置仍超过 30 分钟的账号。
不存在的本地时刻跳过,重复时刻只执行较早一次;同一本地分钟的去重跨进程重启保留。
不存在的本地时刻跳过,重复时刻只执行较早一次;执行进度跨重启保留,时钟回拨不补跑已领取时刻之前的时段。
只有收到响应成功终态才记为预热成功;预热不计入客户端业务用量
`requestLocationEnabled` 是必填布尔值,默认 `false`:关闭时不覆盖客户端原有位置和时区;开启时使用已保存的
+2 -1
View File
@@ -772,7 +772,8 @@ Worker 由各 Bundle 贡献、由 Host 统一监督:
- Provider:credential refresh、quota/catalog 健康、账号预热和官方版本/etag 检查
预热与备份按部署时区解释本地时刻,缺失时刻跳过、重复时刻取较早一次。
预热由 Store 持久领取本地分钟,防止多个时间槽、回拨或重启丢失去重状态;取消仍沿既有 worker lease 生命周期释放。
预热由 Store 在运行设置记录中持久化 UTC 执行游标,只领取晚于游标的时刻,重启或回拨不重复执行历史时段。
游标独立于管理配置的替换与版本发布;取消仍沿既有 worker lease 生命周期释放。
备份设置中的时区仅记录游标来源,切换后从当前时刻重算未来游标,不补跑旧计划。
游标重算核对启停、Cron、来源时区及旧游标;任务入队与游标推进同事务提交,冲突时只跳过任务