diff --git a/backend/crates/gateway-core/src/provider_ports.rs b/backend/crates/gateway-core/src/provider_ports.rs index 8b9bfb594..6255fba9b 100644 --- a/backend/crates/gateway-core/src/provider_ports.rs +++ b/backend/crates/gateway-core/src/provider_ports.rs @@ -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, diff --git a/backend/crates/gateway-store/src/postgres/runtime_settings.rs b/backend/crates/gateway-store/src/postgres/runtime_settings.rs index 2e568dd00..922171f4e 100644 --- a/backend/crates/gateway-store/src/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/src/postgres/runtime_settings.rs @@ -289,25 +289,20 @@ impl ProviderRuntimePolicyPort for PgRuntimeSettingsRepository { slot: chrono::NaiveDateTime, ) -> futures::future::BoxFuture<'a, Result> { 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) }) } diff --git a/backend/crates/gateway-store/tests/postgres/mod.rs b/backend/crates/gateway-store/tests/postgres/mod.rs index eac173a9c..e8773f573 100644 --- a/backend/crates/gateway-store/tests/postgres/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/mod.rs @@ -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", diff --git a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs index 31db79e8d..59c0c9b45 100644 --- a/backend/crates/gateway-store/tests/postgres/runtime_settings.rs +++ b/backend/crates/gateway-store/tests/postgres/runtime_settings.rs @@ -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 = + sqlx::query_scalar("select to_regclass('account_warmup_slots')::text") + .fetch_one(&database.pool) + .await + .unwrap(); + assert!(warmup_table.is_none()); + let cursor: Option> = + 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 = + 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::>().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; } diff --git a/backend/crates/providers/openai/src/provider/workers.rs b/backend/crates/providers/openai/src/provider/workers.rs index 7ede78590..adfd79d48 100644 --- a/backend/crates/providers/openai/src/provider/workers.rs +++ b/backend/crates/providers/openai/src/provider/workers.rs @@ -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(()) }) } diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 8a91c0b7b..b8d6bd210 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -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 diff --git a/backend/migrations/0021_account_warmup_slots.sql b/backend/migrations/0021_account_warmup_slots.sql deleted file mode 100644 index 46dbe6d00..000000000 --- a/backend/migrations/0021_account_warmup_slots.sql +++ /dev/null @@ -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) -); diff --git a/backend/migrations/0022_guardian_reserved_concurrency.sql b/backend/migrations/0021_runtime_scheduling.sql similarity index 78% rename from backend/migrations/0022_guardian_reserved_concurrency.sql rename to backend/migrations/0021_runtime_scheduling.sql index c0e8df9d9..0457216f9 100644 --- a/backend/migrations/0022_guardian_reserved_concurrency.sql +++ b/backend/migrations/0021_runtime_scheduling.sql @@ -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); diff --git a/docs/api.md b/docs/api.md index dc9f0c2d2..bb298647b 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1244,7 +1244,7 @@ accountWarmupModel 定时账号预热默认关闭。`accountWarmupScheduleTime` 使用部署时区中的 `HH:MM`, 多个时段以逗号分隔,默认 `08:00`;`accountWarmupModel` 默认 `null`,开启前必须显式选择模型。 任务面向可用的 OpenAI OAuth 账号,跳过周额度耗尽及五小时窗口距离重置仍超过 30 分钟的账号。 -不存在的本地时刻跳过,重复时刻只执行较早一次;同一本地分钟的去重跨进程重启保留。 +不存在的本地时刻跳过,重复时刻只执行较早一次;执行进度跨重启保留,时钟回拨不补跑已领取时刻之前的时段。 只有收到响应成功终态才记为预热成功;预热不计入客户端业务用量 `requestLocationEnabled` 是必填布尔值,默认 `false`:关闭时不覆盖客户端原有位置和时区;开启时使用已保存的 diff --git a/docs/architecture.md b/docs/architecture.md index ae6b5361e..6c40f48f1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -772,7 +772,8 @@ Worker 由各 Bundle 贡献、由 Host 统一监督: - Provider:credential refresh、quota/catalog 健康、账号预热和官方版本/etag 检查 预热与备份按部署时区解释本地时刻,缺失时刻跳过、重复时刻取较早一次。 -预热由 Store 持久领取本地分钟,防止多个时间槽、回拨或重启丢失去重状态;取消仍沿既有 worker lease 生命周期释放。 +预热由 Store 在运行设置记录中持久化 UTC 执行游标,只领取晚于游标的时刻,重启或回拨不重复执行历史时段。 +游标独立于管理配置的替换与版本发布;取消仍沿既有 worker lease 生命周期释放。 备份设置中的时区仅记录游标来源,切换后从当前时刻重算未来游标,不补跑旧计划。 游标重算核对启停、Cron、来源时区及旧游标;任务入队与游标推进同事务提交,冲突时只跳过任务