i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 09:20:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

Revert "fix(admin): force manual oauth refresh to bypass local cache"

Browse Source 
This reverts commit 85a630cfa9.
This commit is contained in:
fawney19
2026-04-17 13:36:51 +08:00
parent c6af4791f8
commit 654a41c3b0
3 changed files with 24 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
apps/aether-gateway/src/handlers/admin/provider/oauth/dispatch/refresh/execution.rs
View File

@@ -29,7 +29,7 @@ pub(super) async fn execute_admin_provider_oauth_refresh(
Ok(None) => { Ok(None) => {
return Ok(RefreshDispatch::Respond(response::control_error_response( return Ok(RefreshDispatch::Respond(response::control_error_response(
http::StatusCode::BAD_REQUEST, http::StatusCode::BAD_REQUEST,
"Token 刷新未执行，请检查授权配置", "缺少 refresh_token，需要重新授权",
))); )));
} }
Err(AdminLocalOAuthRefreshError::HttpStatus { Err(AdminLocalOAuthRefreshError::HttpStatus {

2
apps/aether-gateway/src/state/oauth.rs
View File

@@ -619,7 +619,7 @@ impl AppState {
for _ in 0..2 { for _ in 0..2 {
let resolution = self let resolution = self
.oauth_refresh .oauth_refresh
.force_refresh_with_result( .resolve_with_result(
&executor, &executor,
&current_transport, &current_transport,
distributed_lock.as_ref(), distributed_lock.as_ref(),

107
crates/aether-provider-transport/src/oauth_refresh.rs
View File

@@ -237,41 +237,6 @@ impl LocalOAuthRefreshCoordinator {
transport: &GatewayProviderTransportSnapshot, transport: &GatewayProviderTransportSnapshot,
distributed_lock: Option<&RedisLockRunner>, distributed_lock: Option<&RedisLockRunner>,
distributed_owner: Option<&str>, distributed_owner: Option<&str>,
) -> Result<Option<LocalOAuthResolution>, LocalOAuthRefreshError> {
self.resolve_with_result_mode(
executor,
transport,
distributed_lock,
distributed_owner,
false,
)
.await
}
pub async fn force_refresh_with_result(
&self,
executor: &dyn LocalOAuthHttpExecutor,
transport: &GatewayProviderTransportSnapshot,
distributed_lock: Option<&RedisLockRunner>,
distributed_owner: Option<&str>,
) -> Result<Option<LocalOAuthResolution>, LocalOAuthRefreshError> {
self.resolve_with_result_mode(
executor,
transport,
distributed_lock,
distributed_owner,
true,
)
.await
}
async fn resolve_with_result_mode(
&self,
executor: &dyn LocalOAuthHttpExecutor,
transport: &GatewayProviderTransportSnapshot,
distributed_lock: Option<&RedisLockRunner>,
distributed_owner: Option<&str>,
force_refresh: bool,
) -> Result<Option<LocalOAuthResolution>, LocalOAuthRefreshError> { ) -> Result<Option<LocalOAuthResolution>, LocalOAuthRefreshError> {
let Some(adapter) = self let Some(adapter) = self
.adapters .adapters
@@ -287,19 +252,17 @@ impl LocalOAuthRefreshCoordinator {
} else { } else {
self.cached_entry(key_id).await self.cached_entry(key_id).await
}; };
if !force_refresh { if let Some(auth) = cached_entry
if let Some(auth) = cached_entry .as_ref()
.as_ref() .and_then(|entry| adapter.resolve_cached(transport, entry))
.and_then(|entry| adapter.resolve_cached(transport, entry)) {
{ return Ok(Some(LocalOAuthResolution::resolved(auth, None)));
return Ok(Some(LocalOAuthResolution::resolved(auth, None))); }
} if let Some(auth) = adapter.resolve_without_refresh(transport) {
if let Some(auth) = adapter.resolve_without_refresh(transport) { return Ok(Some(LocalOAuthResolution::resolved(auth, None)));
return Ok(Some(LocalOAuthResolution::resolved(auth, None))); }
} if !adapter.should_refresh(transport, cached_entry.as_ref()) {
if !adapter.should_refresh(transport, cached_entry.as_ref()) { return Ok(None);
return Ok(None);
}
} }
if key_id.is_empty() { if key_id.is_empty() {
return Ok(None); return Ok(None);
@@ -309,19 +272,17 @@ impl LocalOAuthRefreshCoordinator {
let _key_guard = key_lock.lock().await; let _key_guard = key_lock.lock().await;
let cached_entry = self.cached_entry(key_id).await; let cached_entry = self.cached_entry(key_id).await;
if !force_refresh { if let Some(auth) = cached_entry
if let Some(auth) = cached_entry .as_ref()
.as_ref() .and_then(|entry| adapter.resolve_cached(transport, entry))
.and_then(|entry| adapter.resolve_cached(transport, entry)) {
{ return Ok(Some(LocalOAuthResolution::resolved(auth, None)));
return Ok(Some(LocalOAuthResolution::resolved(auth, None))); }
} if let Some(auth) = adapter.resolve_without_refresh(transport) {
if let Some(auth) = adapter.resolve_without_refresh(transport) { return Ok(Some(LocalOAuthResolution::resolved(auth, None)));
return Ok(Some(LocalOAuthResolution::resolved(auth, None))); }
} if !adapter.should_refresh(transport, cached_entry.as_ref()) {
if !adapter.should_refresh(transport, cached_entry.as_ref()) { return Ok(None);
return Ok(None);
}
} }
let distributed_lease = match (distributed_lock, distributed_owner) { let distributed_lease = match (distributed_lock, distributed_owner) {
@@ -589,28 +550,4 @@ mod tests {
}) })
); );
} }
#[tokio::test]
async fn coordinator_force_refresh_bypasses_runtime_cache() {
let refresh_hits = Arc::new(AtomicUsize::new(0));
let coordinator =
LocalOAuthRefreshCoordinator::with_adapters_for_tests(vec![Arc::new(TestAdapter {
refresh_hits: Arc::clone(&refresh_hits),
})]);
let transport = sample_transport();
let executor = ReqwestLocalOAuthHttpExecutor::new(reqwest::Client::new());
let first = coordinator
.resolve_with_result(&executor, &transport, None, None)
.await
.expect("initial resolve should succeed");
let forced = coordinator
.force_refresh_with_result(&executor, &transport, None, None)
.await
.expect("forced refresh should succeed");
assert!(first.and_then(|result| result.refreshed_entry).is_some());
assert!(forced.and_then(|result| result.refreshed_entry).is_some());
assert_eq!(refresh_hits.load(Ordering::SeqCst), 2);
}
} }