lfg2025/archy
1
0
Fork 0
Code Issues 24 Pull Requests 4 Actions Packages Projects Releases 71 Wiki Activity

refactor: replace blocking std::fs and TCP I/O with async tokio equivalents

Browse Source 
- R6: Convert 6 std::fs calls in session.rs to tokio::fs async
- R7: Convert std::fs::read_to_string in docker_packages.rs to async
- R8: Convert 3 std::fs calls in port_allocator.rs to async, switch to tokio::sync::Mutex
- R9+R10+R11: Fix blocking I/O in node_message.rs and nostr_discovery.rs
- R12: Convert electrs_status.rs from sync TCP to async tokio::net with 5s timeouts
- R4+R5: Spawn periodic cleanup tasks for endpoint and login rate limiters

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dorian
2026-03-21 01:21:08 +00:00
co-authored by Claude Opus 4.6
parent 38dc845f57
commit 4d17c60da7
12 changed files with 161 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/archipelago/src/api/handler.rs
+1 -1
View File
@@ -31,7 +31,7 @@ impl ApiHandler {
state_manager: Arc<StateManager>,
metrics_store: Arc<MetricsStore>,
) -> Result<Self> {
let session_store = SessionStore::new();
let session_store = SessionStore::new().await;
let rpc_handler = Arc::new(
RpcHandler::new(
config.clone(),
core/archipelago/src/api/rpc/mod.rs
+41 -16
View File
@@ -42,7 +42,7 @@ use anyhow::{Context, Result};
use hyper::{Request, Response, StatusCode};
use serde::{Deserialize, Serialize};
use std::net::IpAddr;
use std::sync::{Arc, Mutex};
use std::sync::Arc;
use tracing::{debug, error};
#[derive(Debug, Deserialize)]
@@ -164,7 +164,7 @@ pub struct RpcHandler {
orchestrator: Option<Arc<DevContainerOrchestrator>>,
state_manager: Arc<StateManager>,
pub(crate) metrics_store: Arc<MetricsStore>,
port_allocator: Arc<Mutex<PortAllocator>>,
port_allocator: Arc<tokio::sync::Mutex<PortAllocator>>,
pub session_store: SessionStore,
login_rate_limiter: LoginRateLimiter,
endpoint_rate_limiter: EndpointRateLimiter,
@@ -188,7 +188,32 @@ impl RpcHandler {
} else {
None
};
let port_allocator = Arc::new(Mutex::new(PortAllocator::new(&config.data_dir)?));
let port_allocator = Arc::new(tokio::sync::Mutex::new(PortAllocator::new(&config.data_dir).await?));
let login_rate_limiter = LoginRateLimiter::new();
let endpoint_rate_limiter = EndpointRateLimiter::new();
// Spawn periodic rate limiter cleanup (every 5 minutes)
{
let limiter = endpoint_rate_limiter.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(300));
loop {
interval.tick().await;
limiter.cleanup().await;
}
});
}
{
let limiter = login_rate_limiter.clone();
tokio::spawn(async move {
let mut interval = tokio::time::interval(std::time::Duration::from_secs(300));
loop {
interval.tick().await;
limiter.cleanup().await;
}
});
}
Ok(Self {
config,
@@ -198,8 +223,8 @@ impl RpcHandler {
metrics_store,
port_allocator,
session_store,
login_rate_limiter: LoginRateLimiter::new(),
endpoint_rate_limiter: EndpointRateLimiter::new(),
login_rate_limiter,
endpoint_rate_limiter,
response_cache: ResponseCache::new(5),
mesh_service: Arc::new(tokio::sync::RwLock::new(None)),
transport_router: Arc::new(tokio::sync::RwLock::new(None)),
@@ -253,10 +278,10 @@ impl RpcHandler {
// If session invalid, try remember-me token to auto-restore session
if !authenticated {
if let Some(remember) = extract_cookie(&parts.headers, "remember") {
if crate::session::SessionStore::validate_remember_token(&remember) {
if crate::session::SessionStore::validate_remember_token(&remember).await {
// Auto-create a new session from the remember-me token
let new_token = self.session_store.create().await;
let new_csrf = derive_csrf_token(&new_token);
let new_csrf = derive_csrf_token(&new_token).await;
tracing::info!("Auto-restored session from remember-me token");
new_session_cookies = Some((new_token, new_csrf));
authenticated = true;
@@ -325,7 +350,7 @@ impl RpcHandler {
use hmac::{Hmac, Mac};
use sha2::Sha256;
type HmacSha256 = Hmac<Sha256>;
let secret = SessionStore::load_or_create_remember_secret();
let secret = SessionStore::load_or_create_remember_secret().await;
let mut mac = match HmacSha256::new_from_slice(&secret) {
Ok(m) => m,
Err(_) => { return Ok(Response::builder().status(500).body(hyper::Body::empty()).unwrap()); }
@@ -858,7 +883,7 @@ impl RpcHandler {
if let Ok(Some(totp_data)) = self.auth_manager.get_totp_data().await {
if let Ok(secret) = crate::totp::decrypt_secret(&totp_data, password) {
let token = self.session_store.create_pending(secret).await;
let csrf_token = derive_csrf_token(&token);
let csrf_token = derive_csrf_token(&token).await;
response.headers_mut().append(
"Set-Cookie",
format!("session={}; HttpOnly; SameSite=Lax; Path=/{}", token, self.cookie_suffix())
@@ -884,8 +909,8 @@ impl RpcHandler {
} else {
// No 2FA: create a full session immediately
let token = self.session_store.create().await;
let csrf_token = derive_csrf_token(&token);
let remember_token = self.session_store.create_remember_token();
let csrf_token = derive_csrf_token(&token).await;
let remember_token = self.session_store.create_remember_token().await;
response.headers_mut().append(
"Set-Cookie",
format!("session={}; HttpOnly; SameSite=Lax; Path=/{}", token, self.cookie_suffix())
@@ -921,8 +946,8 @@ impl RpcHandler {
.map(|s| s.to_string());
if let Some(new_token) = new_token_opt {
let csrf_token = derive_csrf_token(&new_token);
let remember_token = self.session_store.create_remember_token();
let csrf_token = derive_csrf_token(&new_token).await;
let remember_token = self.session_store.create_remember_token().await;
response.headers_mut().append(
"Set-Cookie",
format!(
@@ -969,7 +994,7 @@ impl RpcHandler {
if rpc_req.method == "auth.changePassword" && rpc_resp.error.is_none() {
if let Some(token) = &session_token {
let new_token = self.session_store.rotate(token).await;
let csrf_token = derive_csrf_token(&new_token);
let csrf_token = derive_csrf_token(&new_token).await;
response.headers_mut().append(
"Set-Cookie",
format!(
@@ -1059,11 +1084,11 @@ impl RpcHandler {
/// Deterministic: same session token always produces the same CSRF token.
/// Survives backend restarts because it depends only on the session token
/// and the on-disk remember secret (not ephemeral state).
fn derive_csrf_token(session_token: &str) -> String {
async fn derive_csrf_token(session_token: &str) -> String {
use hmac::{Hmac, Mac};
use sha2::Sha256;
type HmacSha256 = Hmac<Sha256>;
let secret = SessionStore::load_or_create_remember_secret();
let secret = SessionStore::load_or_create_remember_secret().await;
let mut mac = HmacSha256::new_from_slice(&secret).expect("HMAC key");
mac.update(format!("csrf:{}", session_token).as_bytes());
hex::encode(mac.finalize().into_bytes())
core/archipelago/src/api/rpc/network.rs
+2 -2
View File
@@ -51,7 +51,7 @@ impl RpcHandler {
/// Get the current node visibility setting.
pub(super) async fn handle_network_get_visibility(&self) -> Result<serde_json::Value> {
let vis = self.load_visibility().await;
let tor_address = docker_packages::read_tor_address("archipelago");
let tor_address = docker_packages::read_tor_address("archipelago").await;
Ok(serde_json::json!({
"visibility": vis.as_str(),
"tor_address": tor_address,
@@ -106,7 +106,7 @@ impl RpcHandler {
let (data, _) = self.state_manager.get_snapshot().await;
let my_pubkey = &data.server_info.pubkey;
let my_did = identity::did_key_from_pubkey_hex(my_pubkey)?;
let my_onion = docker_packages::read_tor_address("archipelago")
let my_onion = docker_packages::read_tor_address("archipelago").await
.unwrap_or_default();
let req_msg = serde_json::json!({
core/archipelago/src/api/rpc/node.rs
+1 -1
View File
@@ -71,7 +71,7 @@ impl RpcHandler {
}
pub(super) async fn handle_node_tor_address(&self) -> Result<serde_json::Value> {
let tor_address = docker_packages::read_tor_address("archipelago");
let tor_address = docker_packages::read_tor_address("archipelago").await;
Ok(serde_json::json!({ "tor_address": tor_address }))
}
core/archipelago/src/api/rpc/package.rs
+9 -10
View File
@@ -198,10 +198,8 @@ impl RpcHandler {
// App-specific configuration (should come from manifest)
let (mut ports, mut volumes, env_vars, custom_command, mut custom_args) = {
let mut allocator = self.port_allocator.lock().map_err(|e| {
anyhow::anyhow!("Port allocator lock poisoned: {}", e)
})?;
get_app_config(package_id, &self.config.host_ip, &mut allocator, &rpc_user, &rpc_pass)
let mut allocator = self.port_allocator.lock().await;
get_app_config(package_id, &self.config.host_ip, &mut allocator, &rpc_user, &rpc_pass).await
};
// Fedimint Gateway: auto-detect LND and switch to lnd mode
@@ -846,8 +844,9 @@ printtoconsole=1\n", rpcauth_line);
}
// Release port allocation
if let Ok(mut allocator) = self.port_allocator.lock() {
let _ = allocator.release(package_id);
{
let mut allocator = self.port_allocator.lock().await;
let _ = allocator.release(package_id).await;
}
// Clean data directories unless preserve_data
@@ -1487,7 +1486,7 @@ fn get_memory_limit(app_id: &str) -> &'static str {
/// Get app-specific configuration
/// Returns: (ports, volumes, env_vars, custom_command, custom_args)
fn get_app_config(
async fn get_app_config(
app_id: &str,
host_ip: &str,
allocator: &mut PortAllocator,
@@ -1625,7 +1624,7 @@ fn get_app_config(
),
"nextcloud" => {
let host_port = allocator
.allocate_or_get(app_id, 8085, 80)
.allocate_or_get(app_id, 8085, 80).await
.unwrap_or(8085);
(
vec![format!("{}:80", host_port)],
@@ -1637,7 +1636,7 @@ fn get_app_config(
}
"vaultwarden" => {
let host_port = allocator
.allocate_or_get(app_id, 8082, 80)
.allocate_or_get(app_id, 8082, 80).await
.unwrap_or(8082);
(
vec![format!("{}:80", host_port)],
@@ -1677,7 +1676,7 @@ fn get_app_config(
),
"filebrowser" => {
let host_port = allocator
.allocate_or_get(app_id, 8083, 80)
.allocate_or_get(app_id, 8083, 80).await
.unwrap_or(8083);
(
vec![format!("{}:80", host_port)],