mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-15 00:06:24 +00:00
feat(openai): check client profile releases hourly (#711)
This commit is contained in:
@@ -1,18 +1,22 @@
|
||||
//! CLI 稳定发布来自官方 npm 包及其平台依赖,不与 Desktop 的内嵌版本混用
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::sync::Arc;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use chrono::Utc;
|
||||
use futures::StreamExt as _;
|
||||
use futures::{StreamExt as _, future::BoxFuture};
|
||||
use gateway_core::provider_ports::{ProviderArtifactProfile, ProviderArtifactProfileCachePort};
|
||||
use gateway_core::routing::ProviderKind;
|
||||
use reqwest::{Client, redirect::Policy};
|
||||
use serde::Deserialize;
|
||||
use tokio::time::Instant;
|
||||
|
||||
use super::selection::{ClientKind, ClientPlatform, ClientRelease, object};
|
||||
use super::{ARTIFACT_PROFILE_CACHE_TTL, CodexWireProfileState, baseline};
|
||||
use super::{
|
||||
ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL, ARTIFACT_PROFILE_CACHE_TTL, CodexWireProfileState,
|
||||
baseline,
|
||||
};
|
||||
|
||||
const ENDPOINT: &str = "https://registry.npmjs.org/@openai%2Fcodex/latest";
|
||||
const MAX_BYTES: usize = 256 * 1024;
|
||||
@@ -67,11 +71,21 @@ fn sequence(version: &str) -> Result<u64, CliReleaseError> {
|
||||
Ok(1 + version.major * 1_000_000 + version.minor * 1_000 + version.patch)
|
||||
}
|
||||
|
||||
pub struct CliReleaseService {
|
||||
/// 官方 npm 发布读取边界
|
||||
pub trait CliReleaseTransport: Send + Sync {
|
||||
fn fetch(&self) -> BoxFuture<'_, Result<String, CliReleaseError>>;
|
||||
}
|
||||
|
||||
struct OfficialCliReleaseTransport {
|
||||
client: Client,
|
||||
}
|
||||
|
||||
pub struct CliReleaseService {
|
||||
transport: Arc<dyn CliReleaseTransport>,
|
||||
provider: ProviderKind,
|
||||
state: CodexWireProfileState,
|
||||
cache: Arc<dyn ProviderArtifactProfileCachePort>,
|
||||
cache_writes: Mutex<BTreeMap<String, Instant>>,
|
||||
}
|
||||
|
||||
impl CliReleaseService {
|
||||
@@ -88,12 +102,27 @@ impl CliReleaseService {
|
||||
.timeout(Duration::from_secs(30))
|
||||
.build()
|
||||
.map_err(|_| CliReleaseError::Fetch)?;
|
||||
Ok(Self {
|
||||
client,
|
||||
Ok(Self::with_transport(
|
||||
provider,
|
||||
state,
|
||||
cache,
|
||||
})
|
||||
Arc::new(OfficialCliReleaseTransport { client }),
|
||||
))
|
||||
}
|
||||
|
||||
pub fn with_transport(
|
||||
provider: ProviderKind,
|
||||
state: CodexWireProfileState,
|
||||
cache: Arc<dyn ProviderArtifactProfileCachePort>,
|
||||
transport: Arc<dyn CliReleaseTransport>,
|
||||
) -> Self {
|
||||
Self {
|
||||
transport,
|
||||
provider,
|
||||
state,
|
||||
cache,
|
||||
cache_writes: Mutex::default(),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn restore(&self) {
|
||||
@@ -135,7 +164,7 @@ impl CliReleaseService {
|
||||
}
|
||||
|
||||
pub async fn refresh(&self) -> Result<(), CliReleaseError> {
|
||||
let result = self.fetch().await;
|
||||
let result = self.transport.fetch().await;
|
||||
for (platform, arch, target) in TARGETS {
|
||||
let outcome = match &result {
|
||||
Ok(version) => self.publish(platform, arch, target, version).await,
|
||||
@@ -155,16 +184,29 @@ impl CliReleaseService {
|
||||
version: &str,
|
||||
) -> Result<ClientRelease, String> {
|
||||
let next = sequence(version).map_err(|error| error.to_string())?;
|
||||
if self
|
||||
.state
|
||||
.client_release(ClientKind::Cli, platform, arch)
|
||||
.is_some_and(|current| {
|
||||
sequence(¤t.codex_version).is_ok_and(|current| current > next)
|
||||
})
|
||||
{
|
||||
let current = self.state.client_release(ClientKind::Cli, platform, arch);
|
||||
if current.as_ref().is_some_and(|current| {
|
||||
sequence(¤t.codex_version).is_ok_and(|current| current > next)
|
||||
}) {
|
||||
return Err("官方 CLI 发布版本低于当前已核验版本".to_owned());
|
||||
}
|
||||
let verified_at = Utc::now();
|
||||
let unchanged = current.filter(|current| current.codex_version == version);
|
||||
if let Some(current) = &unchanged {
|
||||
let fresh = self
|
||||
.cache_writes
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.get(target)
|
||||
.is_some_and(|written| written.elapsed() < ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL);
|
||||
if fresh {
|
||||
return Ok(current.clone());
|
||||
}
|
||||
}
|
||||
// 同版探测和缓存续期保留实际核验时间,只有新版才产生新的核验时间
|
||||
let verified_at = unchanged
|
||||
.as_ref()
|
||||
.and_then(|release| release.verified_at)
|
||||
.unwrap_or_else(Utc::now);
|
||||
let payload =
|
||||
object(&serde_json::json!({"version": version})).map_err(|error| error.to_string())?;
|
||||
let profile = ProviderArtifactProfile::new(
|
||||
@@ -179,41 +221,51 @@ impl CliReleaseService {
|
||||
.replace_if_newer(profile, ARTIFACT_PROFILE_CACHE_TTL)
|
||||
.await
|
||||
{
|
||||
Ok(true) => Ok(ClientRelease {
|
||||
codex_version: version.to_owned(),
|
||||
desktop_version: None,
|
||||
desktop_build: None,
|
||||
verified_at: Some(verified_at),
|
||||
}),
|
||||
Ok(true) => {
|
||||
self.cache_writes
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.insert(target.to_owned(), Instant::now());
|
||||
Ok(ClientRelease {
|
||||
codex_version: version.to_owned(),
|
||||
desktop_version: None,
|
||||
desktop_build: None,
|
||||
verified_at: Some(verified_at),
|
||||
})
|
||||
}
|
||||
Ok(false) => Err("CLI 发布缓存已更新,等待下次同步".to_owned()),
|
||||
Err(_) => Err("CLI 发布版本缓存写入失败".to_owned()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn fetch(&self) -> Result<String, CliReleaseError> {
|
||||
let response = self
|
||||
.client
|
||||
.get(ENDPOINT)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|_| CliReleaseError::Fetch)?;
|
||||
if !response.status().is_success()
|
||||
|| response
|
||||
.content_length()
|
||||
.is_some_and(|length| length > MAX_BYTES as u64)
|
||||
{
|
||||
return Err(CliReleaseError::Fetch);
|
||||
}
|
||||
let mut bytes = Vec::new();
|
||||
let mut stream = response.bytes_stream();
|
||||
while let Some(chunk) = stream.next().await {
|
||||
let chunk = chunk.map_err(|_| CliReleaseError::Fetch)?;
|
||||
if bytes.len().saturating_add(chunk.len()) > MAX_BYTES {
|
||||
return Err(CliReleaseError::Invalid);
|
||||
impl CliReleaseTransport for OfficialCliReleaseTransport {
|
||||
fn fetch(&self) -> BoxFuture<'_, Result<String, CliReleaseError>> {
|
||||
Box::pin(async move {
|
||||
let response = self
|
||||
.client
|
||||
.get(ENDPOINT)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|_| CliReleaseError::Fetch)?;
|
||||
if !response.status().is_success()
|
||||
|| response
|
||||
.content_length()
|
||||
.is_some_and(|length| length > MAX_BYTES as u64)
|
||||
{
|
||||
return Err(CliReleaseError::Fetch);
|
||||
}
|
||||
bytes.extend_from_slice(&chunk);
|
||||
}
|
||||
parse_cli_release(&bytes)
|
||||
let mut bytes = Vec::new();
|
||||
let mut stream = response.bytes_stream();
|
||||
while let Some(chunk) = stream.next().await {
|
||||
let chunk = chunk.map_err(|_| CliReleaseError::Fetch)?;
|
||||
if bytes.len().saturating_add(chunk.len()) > MAX_BYTES {
|
||||
return Err(CliReleaseError::Invalid);
|
||||
}
|
||||
bytes.extend_from_slice(&chunk);
|
||||
}
|
||||
parse_cli_release(&bytes)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -34,11 +34,12 @@ use selection::{ClientKind, ClientPlatform, ClientRelease};
|
||||
pub const CODEX_DESKTOP_APPCAST_URL: &str =
|
||||
"https://persistent.oaistatic.com/codex-app-prod/appcast.xml";
|
||||
/// 官方发布元数据检查周期
|
||||
pub const APPCAST_POLL_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60);
|
||||
pub const APPCAST_POLL_INTERVAL: Duration = Duration::from_secs(60 * 60);
|
||||
|
||||
const APPCAST_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
const MAX_APPCAST_BYTES: usize = 1024 * 1024;
|
||||
const ARTIFACT_PROFILE_CACHE_TTL: Duration = Duration::from_secs(30 * 24 * 60 * 60);
|
||||
const ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL: Duration = Duration::from_secs(24 * 60 * 60);
|
||||
const ARTIFACT_PROFILE_SCHEMA_VERSION: u64 = 1;
|
||||
|
||||
/// 保留已有配置类型入口;位置规则由出口领域统一维护
|
||||
@@ -634,6 +635,7 @@ pub struct CodexDesktopReleaseService {
|
||||
profile: CodexWireProfileState,
|
||||
cache: CodexArtifactProfileCache,
|
||||
verified: Arc<RwLock<Option<CodexBundledReleaseProfile>>>,
|
||||
cache_written_at: Arc<RwLock<Option<tokio::time::Instant>>>,
|
||||
}
|
||||
|
||||
impl CodexDesktopReleaseService {
|
||||
@@ -668,6 +670,7 @@ impl CodexDesktopReleaseService {
|
||||
profile,
|
||||
cache,
|
||||
verified: Arc::new(RwLock::new(verified)),
|
||||
cache_written_at: Arc::default(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -732,10 +735,24 @@ impl CodexDesktopReleaseService {
|
||||
{
|
||||
return Err(CodexDesktopReleaseError::InconsistentRelease);
|
||||
}
|
||||
if !self.cache.replace_if_newer(&verified).await? {
|
||||
return Err(CodexDesktopReleaseError::ReleaseRollback);
|
||||
// 轻量检查只推进观察时间,缓存单独按天续期
|
||||
let renewal_due = self
|
||||
.cache_written_at
|
||||
.read()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.is_none_or(|written| {
|
||||
written.elapsed() >= ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL
|
||||
});
|
||||
if renewal_due {
|
||||
if !self.cache.replace_if_newer(&verified).await? {
|
||||
return Err(CodexDesktopReleaseError::ReleaseRollback);
|
||||
}
|
||||
*self
|
||||
.cache_written_at
|
||||
.write()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner) =
|
||||
Some(tokio::time::Instant::now());
|
||||
}
|
||||
self.profile.update_bundled_release(&verified);
|
||||
return Ok(release);
|
||||
}
|
||||
}
|
||||
@@ -766,6 +783,10 @@ impl CodexDesktopReleaseService {
|
||||
.verified
|
||||
.write()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(published);
|
||||
*self
|
||||
.cache_written_at
|
||||
.write()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner) = Some(tokio::time::Instant::now());
|
||||
Ok(release)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,10 @@
|
||||
use super::selection::{
|
||||
ClientKind, ClientPlatform, ClientProfileSelection, ClientRelease, VersionMode, object,
|
||||
};
|
||||
use super::{ARTIFACT_PROFILE_CACHE_TTL, CodexWireProfileState, baseline};
|
||||
use super::{
|
||||
ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL, ARTIFACT_PROFILE_CACHE_TTL, CodexWireProfileState,
|
||||
baseline,
|
||||
};
|
||||
use chrono::Utc;
|
||||
use futures::future::BoxFuture;
|
||||
use gateway_core::provider_ports::{ProviderArtifactProfile, ProviderArtifactProfileCachePort};
|
||||
@@ -135,6 +138,7 @@ pub struct PlatformDesktopReleaseService {
|
||||
cache: Arc<dyn ProviderArtifactProfileCachePort>,
|
||||
transport: Arc<dyn DesktopArtifactTransport>,
|
||||
verified: Mutex<BTreeMap<DesktopTarget, VerifiedArtifact>>,
|
||||
cache_written_at: Mutex<BTreeMap<DesktopTarget, tokio::time::Instant>>,
|
||||
}
|
||||
impl PlatformDesktopReleaseService {
|
||||
pub fn new(
|
||||
@@ -149,6 +153,7 @@ impl PlatformDesktopReleaseService {
|
||||
cache,
|
||||
transport,
|
||||
verified: Mutex::new(baseline_artifacts()),
|
||||
cache_written_at: Mutex::default(),
|
||||
}
|
||||
}
|
||||
pub async fn restore(&self) {
|
||||
@@ -206,14 +211,32 @@ impl PlatformDesktopReleaseService {
|
||||
.cloned();
|
||||
let mut artifact = self
|
||||
.transport
|
||||
.fetch(target, previous)
|
||||
.fetch(target, previous.clone())
|
||||
.await
|
||||
.map_err(|error| error.to_string())?;
|
||||
let sequence = sequence(target, &artifact.release).map_err(|error| error.to_string())?;
|
||||
if !self.accepts(target, &artifact.release) {
|
||||
return Err("官方 Desktop 发布低于当前版本或同构建版本不一致".to_owned());
|
||||
}
|
||||
let verified_at = Utc::now();
|
||||
let unchanged = previous
|
||||
.as_ref()
|
||||
.is_some_and(|previous| previous.identity == artifact.identity);
|
||||
if unchanged
|
||||
&& self
|
||||
.cache_written_at
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.get(&target)
|
||||
.is_some_and(|written| written.elapsed() < ARTIFACT_PROFILE_CACHE_RENEW_INTERVAL)
|
||||
{
|
||||
return Ok(artifact.release);
|
||||
}
|
||||
// 相同制品只续期缓存,保留实际提取版本时的核验时间
|
||||
let verified_at = if unchanged {
|
||||
artifact.release.verified_at.unwrap_or_else(Utc::now)
|
||||
} else {
|
||||
Utc::now()
|
||||
};
|
||||
// 核验时间属于缓存信封,避免同构建重复检查被判定为不同制品
|
||||
artifact.release.verified_at = None;
|
||||
let payload =
|
||||
@@ -235,6 +258,10 @@ impl PlatformDesktopReleaseService {
|
||||
Ok(false) => return Err("Desktop 发布缓存已更新,等待下次同步".to_owned()),
|
||||
Err(_) => return Err("Desktop 发布缓存写入失败".to_owned()),
|
||||
}
|
||||
self.cache_written_at
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.insert(target, tokio::time::Instant::now());
|
||||
artifact.release.verified_at = Some(verified_at);
|
||||
self.verified
|
||||
.lock()
|
||||
|
||||
@@ -282,7 +282,7 @@ impl ClientProfileSelection {
|
||||
verified_at: release.verified_at.unwrap_or(DateTime::UNIX_EPOCH),
|
||||
};
|
||||
if let Some(entry) = self.cli_entry {
|
||||
// 入口后缀使用同一次解析得到的 Core 版本,避免每日更新后头部与后缀混用
|
||||
// 入口后缀使用同一次解析得到的 Core 版本,避免版本更新后头部与后缀混用
|
||||
// originator 可单独覆盖;入口名与官方 clientInfo.name 的语义保持一致
|
||||
profile.exact_user_agent = Some(format!(
|
||||
"{} ({}; {})",
|
||||
|
||||
@@ -82,7 +82,7 @@ async fn official_cli_cache_updates_entry_headers_together_and_preserves_frozen_
|
||||
))
|
||||
.unwrap();
|
||||
|
||||
// 复用官方版本服务的恢复入口,模拟每日检查已核验并写入的下一份发布资料
|
||||
// 复用官方版本服务的恢复入口,模拟版本检查已核验并写入的下一份发布资料
|
||||
cache
|
||||
.replace_if_newer(
|
||||
ProviderArtifactProfile::new(
|
||||
@@ -142,3 +142,178 @@ async fn official_cli_cache_updates_entry_headers_together_and_preserves_frozen_
|
||||
automatic.resolve(&state).unwrap()
|
||||
);
|
||||
}
|
||||
|
||||
struct CliTransport {
|
||||
version: std::sync::Mutex<String>,
|
||||
}
|
||||
|
||||
impl provider_openai::transport::profile::cli_release::CliReleaseTransport for CliTransport {
|
||||
fn fetch(
|
||||
&self,
|
||||
) -> futures::future::BoxFuture<
|
||||
'_,
|
||||
Result<String, provider_openai::transport::profile::cli_release::CliReleaseError>,
|
||||
> {
|
||||
Box::pin(async move { Ok(self.version.lock().unwrap().clone()) })
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct CliCache {
|
||||
writes: std::sync::Mutex<std::collections::BTreeMap<String, usize>>,
|
||||
failing_target: std::sync::Mutex<Option<String>>,
|
||||
}
|
||||
|
||||
impl gateway_core::provider_ports::ProviderArtifactProfileCachePort for CliCache {
|
||||
fn replace_if_newer(
|
||||
&self,
|
||||
profile: gateway_core::provider_ports::ProviderArtifactProfile,
|
||||
_ttl: std::time::Duration,
|
||||
) -> futures::future::BoxFuture<
|
||||
'_,
|
||||
Result<bool, gateway_core::provider_ports::ProviderStoreError>,
|
||||
> {
|
||||
Box::pin(async move {
|
||||
*self
|
||||
.writes
|
||||
.lock()
|
||||
.unwrap()
|
||||
.entry(profile.artifact_key().to_owned())
|
||||
.or_default() += 1;
|
||||
if self.failing_target.lock().unwrap().as_deref() == Some(profile.artifact_key()) {
|
||||
return Err(gateway_core::provider_ports::ProviderStoreError::new(
|
||||
gateway_core::provider_ports::ProviderStoreErrorKind::Unavailable,
|
||||
"write CLI fixture",
|
||||
));
|
||||
}
|
||||
Ok(true)
|
||||
})
|
||||
}
|
||||
|
||||
fn read<'a>(
|
||||
&'a self,
|
||||
_provider: &'a gateway_core::routing::ProviderKind,
|
||||
_key: &'a str,
|
||||
) -> futures::future::BoxFuture<
|
||||
'a,
|
||||
Result<
|
||||
Option<gateway_core::provider_ports::ProviderArtifactProfile>,
|
||||
gateway_core::provider_ports::ProviderStoreError,
|
||||
>,
|
||||
> {
|
||||
Box::pin(async { Ok(None) })
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn unchanged_cli_release_only_renews_cache_daily_and_preserves_verification_time() {
|
||||
use provider_openai::transport::profile::{
|
||||
CodexWireProfileState,
|
||||
cli_release::CliReleaseService,
|
||||
selection::{ClientKind, ClientPlatform},
|
||||
};
|
||||
use std::{sync::Arc, time::Duration};
|
||||
let state = CodexWireProfileState::new(super::wire_profile());
|
||||
let cache = Arc::new(CliCache::default());
|
||||
let transport = Arc::new(CliTransport {
|
||||
version: std::sync::Mutex::new("1.0.0".to_owned()),
|
||||
});
|
||||
let service = CliReleaseService::with_transport(
|
||||
gateway_core::routing::ProviderKind::new("openai").unwrap(),
|
||||
state.clone(),
|
||||
cache.clone(),
|
||||
transport,
|
||||
);
|
||||
service.refresh().await.unwrap();
|
||||
let verified = state
|
||||
.client_release(ClientKind::Cli, ClientPlatform::Linux, "x86_64")
|
||||
.unwrap();
|
||||
tokio::time::advance(Duration::from_secs(3600)).await;
|
||||
service.refresh().await.unwrap();
|
||||
assert!(
|
||||
cache
|
||||
.writes
|
||||
.lock()
|
||||
.unwrap()
|
||||
.values()
|
||||
.all(|count| *count == 1)
|
||||
);
|
||||
tokio::time::advance(Duration::from_secs(23 * 3600)).await;
|
||||
service.refresh().await.unwrap();
|
||||
assert!(
|
||||
cache
|
||||
.writes
|
||||
.lock()
|
||||
.unwrap()
|
||||
.values()
|
||||
.all(|count| *count == 2)
|
||||
);
|
||||
assert_eq!(
|
||||
state
|
||||
.client_release(ClientKind::Cli, ClientPlatform::Linux, "x86_64")
|
||||
.unwrap(),
|
||||
verified
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn cli_partial_cache_failure_retries_the_same_release_without_republishing_other_targets() {
|
||||
use provider_openai::transport::profile::{
|
||||
CodexWireProfileState,
|
||||
cli_release::CliReleaseService,
|
||||
selection::{ClientKind, ClientPlatform},
|
||||
};
|
||||
use std::{sync::Arc, time::Duration};
|
||||
let state = CodexWireProfileState::new(super::wire_profile());
|
||||
let original = state
|
||||
.client_release(ClientKind::Cli, ClientPlatform::Linux, "x86_64")
|
||||
.unwrap();
|
||||
let cache = Arc::new(CliCache::default());
|
||||
*cache.failing_target.lock().unwrap() = Some("cli-linux-x64".to_owned());
|
||||
let transport = Arc::new(CliTransport {
|
||||
version: std::sync::Mutex::new("1.0.0".to_owned()),
|
||||
});
|
||||
let service = CliReleaseService::with_transport(
|
||||
gateway_core::routing::ProviderKind::new("openai").unwrap(),
|
||||
state.clone(),
|
||||
cache.clone(),
|
||||
transport,
|
||||
);
|
||||
service.refresh().await.unwrap();
|
||||
assert_eq!(
|
||||
state
|
||||
.client_release(ClientKind::Cli, ClientPlatform::Linux, "x86_64")
|
||||
.unwrap(),
|
||||
original
|
||||
);
|
||||
assert!(
|
||||
state
|
||||
.client_release_status(ClientKind::Cli, ClientPlatform::Linux, "x86_64")
|
||||
.1
|
||||
.is_some()
|
||||
);
|
||||
*cache.failing_target.lock().unwrap() = None;
|
||||
tokio::time::advance(Duration::from_secs(3600)).await;
|
||||
service.refresh().await.unwrap();
|
||||
assert_eq!(
|
||||
state
|
||||
.client_release(ClientKind::Cli, ClientPlatform::Linux, "x86_64")
|
||||
.unwrap()
|
||||
.codex_version,
|
||||
"1.0.0"
|
||||
);
|
||||
assert!(
|
||||
state
|
||||
.client_release_status(ClientKind::Cli, ClientPlatform::Linux, "x86_64")
|
||||
.1
|
||||
.is_none()
|
||||
);
|
||||
let writes = cache.writes.lock().unwrap();
|
||||
assert_eq!(writes["cli-linux-x64"], 2);
|
||||
assert!(
|
||||
writes
|
||||
.iter()
|
||||
.filter(|(key, _)| key.as_str() != "cli-linux-x64")
|
||||
.all(|(_, count)| *count == 1)
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
//! 客户端画像测试入口,以及 Desktop 发行解析和原子更新测试
|
||||
|
||||
use std::collections::VecDeque;
|
||||
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
@@ -75,14 +76,16 @@ impl CodexDesktopReleaseTransport for ReleaseTransport {
|
||||
#[derive(Default)]
|
||||
struct ArtifactProfiles {
|
||||
profile: Mutex<Option<ProviderArtifactProfile>>,
|
||||
fail_writes: bool,
|
||||
fail_writes: AtomicBool,
|
||||
writes: AtomicUsize,
|
||||
}
|
||||
|
||||
impl ArtifactProfiles {
|
||||
fn failing() -> Self {
|
||||
Self {
|
||||
profile: Mutex::new(None),
|
||||
fail_writes: true,
|
||||
fail_writes: AtomicBool::new(true),
|
||||
writes: AtomicUsize::new(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -94,7 +97,8 @@ impl ProviderArtifactProfileCachePort for ArtifactProfiles {
|
||||
_ttl: Duration,
|
||||
) -> BoxFuture<'_, Result<bool, ProviderStoreError>> {
|
||||
Box::pin(async move {
|
||||
if self.fail_writes {
|
||||
self.writes.fetch_add(1, Ordering::SeqCst);
|
||||
if self.fail_writes.load(Ordering::SeqCst) {
|
||||
return Err(ProviderStoreError::new(
|
||||
ProviderStoreErrorKind::Unavailable,
|
||||
"replace test artifact profile",
|
||||
@@ -272,19 +276,34 @@ async fn bundled_release_update_should_change_core_and_desktop_identity_atomical
|
||||
assert_eq!(transport.core_probes(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unchanged_verified_release_should_refresh_cache_without_downloading_core_again() {
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn unchanged_verified_release_renews_cache_daily_without_downloading_core_again() {
|
||||
let state = CodexWireProfileState::new(wire_profile());
|
||||
let latest = release("26.810.41047", "6570");
|
||||
let transport = Arc::new(ReleaseTransport::new(
|
||||
[Ok(latest.clone()), Ok(latest)],
|
||||
(0..5).map(|_| Ok(latest.clone())),
|
||||
[Ok("0.148.0-alpha.9".to_owned())],
|
||||
));
|
||||
let cache = Arc::new(ArtifactProfiles::default());
|
||||
let service = service(state, transport.clone(), cache, None);
|
||||
let service = service(state.clone(), transport.clone(), cache.clone(), None);
|
||||
|
||||
service.refresh().await.expect("first release refresh");
|
||||
let verified = state.snapshot();
|
||||
tokio::time::advance(Duration::from_secs(60 * 60)).await;
|
||||
service.refresh().await.expect("cached release refresh");
|
||||
assert_eq!(cache.writes.load(Ordering::SeqCst), 1);
|
||||
|
||||
tokio::time::advance(Duration::from_secs(23 * 60 * 60)).await;
|
||||
cache.fail_writes.store(true, Ordering::SeqCst);
|
||||
service.refresh().await.expect_err("renewal failure");
|
||||
assert_eq!(state.snapshot(), verified);
|
||||
cache.fail_writes.store(false, Ordering::SeqCst);
|
||||
service.refresh().await.expect("retry renewal");
|
||||
assert_eq!(cache.writes.load(Ordering::SeqCst), 3);
|
||||
service.refresh().await.expect("renewed cache");
|
||||
assert_eq!(cache.writes.load(Ordering::SeqCst), 3);
|
||||
assert_eq!(state.snapshot(), verified);
|
||||
assert!(service.status().snapshot().last_error.is_none());
|
||||
|
||||
assert_eq!(transport.core_probes(), 1);
|
||||
}
|
||||
@@ -329,15 +348,11 @@ async fn cache_failure_should_not_publish_a_partially_verified_profile() {
|
||||
let original = wire_profile();
|
||||
let state = CodexWireProfileState::new(original.clone());
|
||||
let transport = Arc::new(ReleaseTransport::new(
|
||||
[Ok(release("26.810.41047", "6570"))],
|
||||
[Ok("0.148.0-alpha.9".to_owned())],
|
||||
(0..2).map(|_| Ok(release("26.810.41047", "6570"))),
|
||||
(0..2).map(|_| Ok("0.148.0-alpha.9".to_owned())),
|
||||
));
|
||||
let service = service(
|
||||
state.clone(),
|
||||
transport,
|
||||
Arc::new(ArtifactProfiles::failing()),
|
||||
None,
|
||||
);
|
||||
let cache = Arc::new(ArtifactProfiles::failing());
|
||||
let service = service(state.clone(), transport, cache.clone(), None);
|
||||
|
||||
service
|
||||
.refresh()
|
||||
@@ -346,6 +361,9 @@ async fn cache_failure_should_not_publish_a_partially_verified_profile() {
|
||||
|
||||
assert_eq!(state.snapshot(), original);
|
||||
assert!(service.status().snapshot().latest.is_none());
|
||||
cache.fail_writes.store(false, Ordering::SeqCst);
|
||||
service.refresh().await.expect("retry same release");
|
||||
assert_eq!(state.snapshot().desktop_build, "6570");
|
||||
}
|
||||
|
||||
/// 滞后档位按发布观察序列解析:启动种子、历次核验发布与失败观察共同构成历史。
|
||||
|
||||
@@ -14,6 +14,7 @@ use provider_openai::transport::profile::selection::{ClientKind, ClientProfileSe
|
||||
use serde_json::json;
|
||||
use std::collections::BTreeMap;
|
||||
use std::io;
|
||||
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
@@ -24,7 +25,8 @@ mod xz;
|
||||
#[derive(Default)]
|
||||
struct Cache {
|
||||
profiles: Mutex<BTreeMap<String, ProviderArtifactProfile>>,
|
||||
reject: bool,
|
||||
reject: AtomicBool,
|
||||
writes: AtomicUsize,
|
||||
}
|
||||
impl ProviderArtifactProfileCachePort for Cache {
|
||||
fn read<'a>(
|
||||
@@ -40,7 +42,8 @@ impl ProviderArtifactProfileCachePort for Cache {
|
||||
_ttl: Duration,
|
||||
) -> BoxFuture<'_, Result<bool, ProviderStoreError>> {
|
||||
Box::pin(async move {
|
||||
if self.reject {
|
||||
self.writes.fetch_add(1, Ordering::Relaxed);
|
||||
if self.reject.load(Ordering::Relaxed) {
|
||||
return Ok(false);
|
||||
}
|
||||
self.profiles
|
||||
@@ -69,6 +72,9 @@ impl DesktopArtifactTransport for Transport {
|
||||
if round == 1 && target == TARGETS[0] {
|
||||
return Err(io::Error::other("fixture outage"));
|
||||
}
|
||||
if round == 3 {
|
||||
return Ok(previous);
|
||||
}
|
||||
let build = if round == 2 {
|
||||
1
|
||||
} else {
|
||||
@@ -136,22 +142,85 @@ async fn releases_are_independent_atomic_cached_and_preserved_on_failures_or_rol
|
||||
.await;
|
||||
assert_eq!(selection.resolve(&restored).unwrap(), frozen);
|
||||
}
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn unchanged_artifacts_renew_daily_and_keep_verification_time() {
|
||||
let state = CodexWireProfileState::new(super::wire_profile());
|
||||
let cache = Arc::new(Cache::default());
|
||||
let transport = Arc::new(Transport::default());
|
||||
*transport.round.lock().unwrap() = 3;
|
||||
let service = service(state.clone(), cache.clone(), transport);
|
||||
let target = TARGETS[0];
|
||||
let original = state.client_release(ClientKind::Desktop, target.platform, target.arch);
|
||||
service.refresh().await;
|
||||
assert_eq!(cache.writes.load(Ordering::Relaxed), 4);
|
||||
tokio::time::advance(Duration::from_secs(60 * 60)).await;
|
||||
service.refresh().await;
|
||||
assert_eq!(cache.writes.load(Ordering::Relaxed), 4);
|
||||
for day in 1..=31 {
|
||||
tokio::time::advance(Duration::from_secs(24 * 60 * 60)).await;
|
||||
service.refresh().await;
|
||||
assert_eq!(cache.writes.load(Ordering::Relaxed), 4 * (day + 1));
|
||||
assert_eq!(
|
||||
state.client_release(ClientKind::Desktop, target.platform, target.arch),
|
||||
original
|
||||
);
|
||||
}
|
||||
cache.reject.store(true, Ordering::Relaxed);
|
||||
tokio::time::advance(Duration::from_secs(24 * 60 * 60)).await;
|
||||
service.refresh().await;
|
||||
assert!(
|
||||
state
|
||||
.client_release_status(ClientKind::Desktop, target.platform, target.arch)
|
||||
.1
|
||||
.is_some()
|
||||
);
|
||||
cache.reject.store(false, Ordering::Relaxed);
|
||||
tokio::time::advance(Duration::from_secs(60 * 60)).await;
|
||||
service.refresh().await;
|
||||
assert_eq!(cache.writes.load(Ordering::Relaxed), 4 * 34);
|
||||
assert!(
|
||||
state
|
||||
.client_release_status(ClientKind::Desktop, target.platform, target.arch)
|
||||
.1
|
||||
.is_none()
|
||||
);
|
||||
assert_eq!(
|
||||
state.client_release(ClientKind::Desktop, target.platform, target.arch),
|
||||
original
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cache_rejection_keeps_the_previous_release() {
|
||||
async fn cache_rejection_preserves_previous_release_and_retries_same_artifact() {
|
||||
let state = CodexWireProfileState::new(super::wire_profile());
|
||||
let before = state.client_release(ClientKind::Desktop, TARGETS[0].platform, TARGETS[0].arch);
|
||||
service(
|
||||
state.clone(),
|
||||
Arc::new(Cache {
|
||||
reject: true,
|
||||
..Default::default()
|
||||
}),
|
||||
Arc::new(Transport::default()),
|
||||
)
|
||||
.refresh()
|
||||
.await;
|
||||
let cache = Arc::new(Cache {
|
||||
reject: AtomicBool::new(true),
|
||||
..Default::default()
|
||||
});
|
||||
let transport = Arc::new(Transport::default());
|
||||
let service = service(state.clone(), cache.clone(), transport);
|
||||
service.refresh().await;
|
||||
assert_eq!(
|
||||
state.client_release(ClientKind::Desktop, TARGETS[0].platform, TARGETS[0].arch),
|
||||
before
|
||||
);
|
||||
cache.reject.store(false, Ordering::Relaxed);
|
||||
service.refresh().await;
|
||||
assert_eq!(cache.writes.load(Ordering::Relaxed), 8);
|
||||
for target in TARGETS {
|
||||
assert_eq!(
|
||||
state
|
||||
.client_release(ClientKind::Desktop, target.platform, target.arch)
|
||||
.unwrap()
|
||||
.codex_version,
|
||||
"0.200.0"
|
||||
);
|
||||
assert!(
|
||||
state
|
||||
.client_release_status(ClientKind::Desktop, target.platform, target.arch)
|
||||
.1
|
||||
.is_none()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
+4
-2
@@ -2468,13 +2468,14 @@ TUI 默认标识为 `codex-tui`,Exec 为 `codex_exec`,入口后缀使用同
|
||||
因此 Desktop 预设保持默认即可;CLI 预设建议配置真实值,多 Key 可通过各自的画像覆盖差异化
|
||||
|
||||
`versionLag` 让 `latest` 模式跟随官方发布但滞后 N 个已观察版本采用,省略时采用最新版本。
|
||||
档位基于网关观察到的发布序列解析:启动种子与历次 24 小时检查的核验版本按时间倒序构成历史,
|
||||
档位基于网关观察到的发布序列解析:启动种子与历次检查的核验版本按时间倒序构成历史,
|
||||
取第 N 项;序列不足时回退最旧已核验版本,不会因此解析失败。历史仅保留在进程内,
|
||||
重启后从启动种子与制品缓存重新积累
|
||||
|
||||
六套预设均支持自动更新:macOS Desktop 支持 arm64,Windows/Linux Desktop 及三套 CLI 支持 arm64、x86_64。
|
||||
预设接口的 `automaticAvailable`、`reason` 表示当前组合的可用性;自定义架构可能使自动解析不可用。
|
||||
每 24 小时后台检查官方稳定发布,失败保留同组合上次有效版本;固定值不受后台更新影响。
|
||||
启动时立即检查官方稳定发布,之后每轮检查完成等待 1 小时再检查。
|
||||
版本变化后提取并更新画像,失败保留同组合上次有效版本,下一轮重试;固定值不受后台更新影响。
|
||||
Desktop 的应用版本、Core 和构建号来自同一平台、架构的官方制品:macOS ZIP、Windows MSIX、Linux DEB。
|
||||
Windows/Linux 通过 ETag 检查更新,未变化时复用已核验版本;CLI 依据官方 npm 稳定标签和对应平台依赖
|
||||
|
||||
@@ -2485,6 +2486,7 @@ Windows/Linux 通过 ETag 检查更新,未变化时复用已核验版本;CLI
|
||||
完整自定义配置不携带官方制品核验时间。
|
||||
客户端画像配置控制应用层请求字段,不切换操作系统的 TLS 实现。默认 HTTP 使用 native TLS,
|
||||
WebSocket 使用 rustls;配置自定义 CA 时 HTTP 也使用 rustls。TLS 指纹需按实际部署平台与传输路径核验。
|
||||
版本未变化时只更新 `checkedAt`,保留 `verifiedAt`;缓存续期不改写核验时间。
|
||||
未完成本次启动检查时 `checkedAt` 为 `null`。非法或当前不可用的选择返回 `400`,保存失败不提交其他修改
|
||||
|
||||
Responses 的自动 Desktop Git 上下文与请求头使用同一画像快照:已有 `appServerVersion` 同步为 Core 版本,
|
||||
|
||||
@@ -1297,6 +1297,10 @@ Worker 由各 Bundle 贡献、由 Host 统一监督:
|
||||
以及账号错误保护恢复 worker(到期解除或探测恢复)、重置卡自动使用和公共 CPR 互联同步
|
||||
- Provider:credential refresh、quota/catalog 健康、账号预热和官方版本/etag 检查
|
||||
|
||||
OpenAI 客户端画像由 Provider 每小时检查发布元数据,发布标识变化后才提取对应制品版本并发布画像。
|
||||
同版本的有效制品缓存每 24 小时续期,保留 30 天;续期保留原核验时间,失败不推进成功写入游标。
|
||||
版本及缓存更新失败保留当前画像,下轮继续尝试;Host 只监督调度与取消。
|
||||
|
||||
互联同步使用账号出口的受管 HTTP 客户端,不跟随跳转;网络等待不持有数据库事务。
|
||||
运行期排程、并发预算、连接失效和观测提交见 [CPR 互联观测](#cpr-互联观测),失败遵守退避及上游 `Retry-After`
|
||||
|
||||
|
||||
Reference in New Issue
Block a user