ms2k/remora
SHA256
Watch
0
Code Releases Activity

Ship production deployment and runtime observability

Browse source
This commit is contained in:
ms2k 2026-07-18 04:22:14 +09:00
commit 775dd61963
26 changed files with 3712 additions and 175 deletions
Download patch file Download diff file Expand all files Collapse all files

6
crates/remora-api/Cargo.toml
View file

@ -14,13 +14,17 @@ getrandom = "0.3.4"
remora-admission = { path = "../remora-admission" }
remora-attempt = { path = "../remora-attempt" }
remora-contracts = { path = "../remora-contracts" }
remora-coordination = { path = "../remora-coordination" }
remora-credentials = { path = "../remora-credentials" }
remora-deletion = { path = "../remora-deletion" }
remora-domain.workspace = true
remora-durable.workspace = true
remora-identity = { path = "../remora-identity" }
remora-result = { path = "../remora-result" }
remora-runtime = { path = "../remora-runtime" }
remora-target = { path = "../remora-target" }
rustls = "=0.23.42"
rustls-pemfile = "=2.2.0"
serde.workspace = true
serde_json.workspace = true
sha2.workspace = true
@ -28,6 +32,7 @@ sqlx.workspace = true
thiserror.workspace = true
time.workspace = true
tokio.workspace = true
tracing.workspace = true
[dev-dependencies]
base64 = "=0.22.1"
@ -35,5 +40,4 @@ jsonwebtoken = "=9.3.1"
rcgen = "=0.14.8"
remora-coordination = { path = "../remora-coordination" }
reqwest = { version = "=0.13.4", default-features = false, features = ["form", "json", "rustls"] }
rustls = "=0.23.42"
tokio-tungstenite = "=0.30.0"

675
crates/remora-api/src/application.rs Normal file
View file

@ -0,0 +1,675 @@
use std::{
fs,
io::{self, BufReader},
net::{SocketAddr, TcpListener},
path::{Path, PathBuf},
sync::Arc,
time::Duration,
};
use actix_web::{App, HttpServer, middleware, web};
use remora_admission::{AdmissionConfig, AdmissionService};
use remora_attempt::{
AttemptConfig, AttemptService, CleanupIdentity, RecoveryConfig, StagingMultipart, StagingObject,
};
use remora_coordination::{CoordinationStore, StoreConfig};
use remora_credentials::{CredentialRegistry, VaultClient, VaultConfig};
use remora_deletion::{DeletionConfig, DeletionService, DrainOutcome, TargetRegistry};
use remora_domain::{BucketName, OpaqueId, TargetPrefix, TargetStoreId};
use remora_durable::{PoolConfig, connect};
use remora_identity::{DefaultTarget, OidcConfig, OidcIdentityProvider};
use remora_runtime::{
ManagementConfig, RuntimeError, RuntimeStatus, env_duration, env_parse, env_string,
required_env, spawn_periodic, start_management_server, trace_request,
};
use remora_target::{MultipartIdentity, ObjectIdentity, TargetStore, TargetStoreConfig};
use rustls::{
RootCertStore, ServerConfig,
pki_types::{CertificateDer, PrivateKeyDer},
server::WebPkiClientVerifier,
};
use sqlx::PgPool;
use tokio::task::JoinHandle;
use crate::{
PublicApiConfig, PublicApiServices, PublicApiState, WorkerApiState, authenticate_rustls_worker,
configure_public_api, configure_worker_api,
};
#[derive(Clone)]
pub struct ApiApplicationConfig {
pub public_bind: SocketAddr,
pub worker_bind: SocketAddr,
pub management: ManagementConfig,
pub database_url: String,
pub database_connections: u32,
pub coordination_url: String,
pub coordination_namespace: String,
pub coordination: StoreConfig,
pub attempt: AttemptConfig,
pub recovery: RecoveryConfig,
pub recovery_interval: Duration,
pub credential_reconcile_interval: Duration,
pub deletion_interval: Duration,
pub default_target_prefix: String,
pub reserved_target_segments: Vec<String>,
pub admission_policy_version: String,
pub public_api: PublicApiConfig,
pub oidc: OidcConfig,
pub target: TargetStoreConfig,
pub vault: VaultConfig,
pub worker_tls: WorkerTlsConfig,
}
#[derive(Clone, Debug)]
pub struct WorkerTlsConfig {
pub certificate: PathBuf,
pub private_key: PathBuf,
pub client_ca: PathBuf,
}
pub struct ApiApplication {
config: ApiApplicationConfig,
}
impl ApiApplicationConfig {
pub fn from_env() -> Result<Self, RuntimeError> {
let reserved_target_segments = comma_values(
"REMORA_RESERVED_TARGET_SEGMENTS",
&env_string("REMORA_RESERVED_TARGET_SEGMENTS", "_remora"),
)?;
let oidc_audience = required_env("REMORA_OIDC_AUDIENCE")?;
let oidc_resource_client = env_string("REMORA_OIDC_RESOURCE_CLIENT", &oidc_audience);
let mut oidc = OidcConfig::new(required_env("REMORA_OIDC_ISSUER")?, oidc_audience)
.map_err(runtime_error)?
.with_resource_access_client(oidc_resource_client)
.with_reserved_segments(reserved_target_segments.clone());
for claim in comma_values(
"REMORA_OIDC_PROJECT_ROLE_CLAIMS",
&env_string(
"REMORA_OIDC_PROJECT_ROLE_CLAIMS",
remora_identity::ZITADEL_PROJECT_ROLES_CLAIM,
),
)? {
oidc = oidc.with_project_role_claim(claim);
}
let event_origins = comma_values(
"REMORA_EVENT_ALLOWED_ORIGINS",
&required_env("REMORA_EVENT_ALLOWED_ORIGINS")?,
)?;
let cursor_secret_path = PathBuf::from(required_env("REMORA_CURSOR_SECRET_FILE")?);
let cursor_secret = fs::read(&cursor_secret_path)
.map_err(|error| file_error("REMORA_CURSOR_SECRET_FILE", error))?;
let global_limit = env_parse("REMORA_GLOBAL_ATTEMPT_LIMIT", "64")?;
let lease_ttl = env_duration("REMORA_LEASE_TTL_SECONDS", 30)?;
let target = TargetStoreConfig::from_env().map_err(runtime_error)?;
let vault = VaultConfig::from_env().map_err(runtime_error)?;
Ok(Self {
public_bind: env_parse("REMORA_PUBLIC_BIND", "0.0.0.0:8080")?,
worker_bind: env_parse("REMORA_WORKER_BIND", "0.0.0.0:8443")?,
management: ManagementConfig::from_env(crate::COMPONENT_NAME)?,
database_url: required_env("DATABASE_URL")?,
database_connections: env_parse("REMORA_DATABASE_MAX_CONNECTIONS", "32")?,
coordination_url: required_env("REMORA_COORDINATION_URL")?,
coordination_namespace: required_env("REMORA_COORDINATION_NAMESPACE")?,
coordination: StoreConfig {
global_limit,
lease_ttl,
reserve_ttl: env_duration("REMORA_RESERVE_TTL_SECONDS", 30)?,
progress_ttl: env_duration("REMORA_PROGRESS_TTL_SECONDS", 60)?,
cancel_ttl: None,
},
attempt: AttemptConfig {
max_attempts: env_parse("REMORA_MAX_ATTEMPTS", "3")?,
},
recovery: RecoveryConfig {
retry_delay: env_duration("REMORA_RETRY_DELAY_SECONDS", 1)?,
cleanup_claim_ttl: env_duration("REMORA_CLEANUP_CLAIM_TTL_SECONDS", 30)?,
batch_size: env_parse("REMORA_RECOVERY_BATCH_SIZE", "128")?,
},
recovery_interval: env_duration("REMORA_RECOVERY_INTERVAL_SECONDS", 2)?,
credential_reconcile_interval: env_duration(
"REMORA_CREDENTIAL_RECONCILE_INTERVAL_SECONDS",
10,
)?,
deletion_interval: env_duration("REMORA_DELETION_INTERVAL_SECONDS", 2)?,
default_target_prefix: env_string("REMORA_DEFAULT_TARGET_PREFIX", ""),
reserved_target_segments,
admission_policy_version: env_string(
"REMORA_ADMISSION_POLICY_VERSION",
"deployment-v1",
),
public_api: PublicApiConfig {
cursor_secret,
member_access_ttl: env_duration("REMORA_MEMBER_ACCESS_TTL_SECONDS", 60)?,
event_retention: env_duration("REMORA_EVENT_RETENTION_SECONDS", 3600)?,
event_poll_interval: duration_millis("REMORA_EVENT_POLL_MILLIS", 250)?,
event_heartbeat_interval: env_duration("REMORA_EVENT_HEARTBEAT_SECONDS", 15)?,
event_ticket_ttl: env_duration("REMORA_EVENT_TICKET_TTL_SECONDS", 30)?,
event_allowed_origins: event_origins,
},
oidc,
target,
vault,
worker_tls: WorkerTlsConfig {
certificate: PathBuf::from(required_env("REMORA_WORKER_TLS_CERT_FILE")?),
private_key: PathBuf::from(required_env("REMORA_WORKER_TLS_KEY_FILE")?),
client_ca: PathBuf::from(required_env("REMORA_WORKER_TLS_CLIENT_CA_FILE")?),
},
})
}
}
impl ApiApplication {
#[must_use]
pub fn new(config: ApiApplicationConfig) -> Self {
Self { config }
}
pub async fn run(self) -> Result<(), RuntimeError> {
let _ = rustls::crypto::ring::default_provider().install_default();
let pool = connect(
&self.config.database_url,
PoolConfig {
max_connections: self.config.database_connections,
acquire_timeout: Duration::from_secs(10),
idle_timeout: Some(Duration::from_secs(60)),
},
)
.await
.map_err(runtime_error)?;
migrate_all(&pool).await?;
let coordination = CoordinationStore::connect(
&self.config.coordination_url,
&self.config.coordination_namespace,
self.config.coordination.clone(),
)
.await
.map_err(runtime_error)?;
let target = TargetStore::connect(self.config.target.clone()).map_err(runtime_error)?;
let readiness_target = target.clone();
let target_registry = TargetRegistry::new([target.clone()]);
let vault = VaultClient::new(self.config.vault.clone()).map_err(runtime_error)?;
let readiness_vault = vault.clone();
let credentials = CredentialRegistry::new(pool.clone(), vault);
let attempts = AttemptService::new(
pool.clone(),
coordination.clone(),
self.config.attempt.clone(),
)
.map_err(runtime_error)?;
let deletion = DeletionService::new(
pool.clone(),
target_registry.clone(),
DeletionConfig::default(),
)
.map_err(runtime_error)?;
let state = PublicApiState::new(
pool.clone(),
PublicApiServices {
identity: OidcIdentityProvider::new(self.config.oidc.clone())
.map_err(runtime_error)?,
admission: admission_service(&self.config, &pool)?,
attempts: attempts.clone(),
credentials: credentials.clone(),
deletion: deletion.clone(),
targets: target_registry.clone(),
},
self.config.public_api.clone(),
)
.map_err(runtime_error)?;
let status = RuntimeStatus::new(crate::COMPONENT_NAME, pool.clone(), coordination.clone())
.with_readiness_check(move || {
let target = readiness_target.clone();
async move { target.check().await.is_ok() }
})
.with_readiness_check(move || {
let vault = readiness_vault.clone();
async move { vault.check().await.is_ok() }
});
let maintenance = start_maintenance(
&self.config,
&pool,
&status,
attempts.clone(),
credentials,
deletion,
target_registry,
);
let public = public_server(self.config.public_bind, state)?;
let worker = worker_server(
self.config.worker_bind,
WorkerApiState::new(attempts),
&self.config.worker_tls,
)?;
let management = start_management_server(self.config.management, status.clone())
.map_err(runtime_error)?;
status.set_accepting_traffic(true);
tracing::info!(
public_bind = %self.config.public_bind,
worker_bind = %self.config.worker_bind,
"api application ready"
);
wait_for_shutdown().await?;
status.set_accepting_traffic(false);
public.stop(true).await;
worker.stop(true).await;
management.stop(true).await;
for task in maintenance {
task.abort();
}
pool.close().await;
Ok(())
}
}
fn admission_service(
config: &ApiApplicationConfig,
pool: &PgPool,
) -> Result<AdmissionService, RuntimeError> {
let store =
TargetStoreId::parse(config.target.target_store_id.clone()).map_err(runtime_error)?;
let bucket = BucketName::parse(config.target.bucket.clone()).map_err(runtime_error)?;
let prefix = TargetPrefix::parse(
&config.default_target_prefix,
config.reserved_target_segments.iter().map(String::as_str),
)
.map_err(runtime_error)?;
AdmissionService::new(
pool.clone(),
AdmissionConfig {
default_target: DefaultTarget {
store,
bucket,
prefix,
},
reserved_target_segments: config.reserved_target_segments.clone(),
policy_version: config.admission_policy_version.clone(),
},
)
.map_err(runtime_error)
}
async fn migrate_all(pool: &PgPool) -> Result<(), RuntimeError> {
remora_durable::migrate(pool).await.map_err(runtime_error)?;
remora_identity::migrate(pool)
.await
.map_err(runtime_error)?;
remora_credentials::migrate(pool)
.await
.map_err(runtime_error)?;
remora_admission::migrate(pool)
.await
.map_err(runtime_error)?;
remora_attempt::migrate(pool).await.map_err(runtime_error)?;
remora_result::migrate(pool).await.map_err(runtime_error)?;
remora_deletion::migrate(pool)
.await
.map_err(runtime_error)?;
crate::migrate(pool).await.map_err(runtime_error)
}
fn start_maintenance(
config: &ApiApplicationConfig,
pool: &PgPool,
status: &RuntimeStatus,
attempts: AttemptService,
credentials: CredentialRegistry,
deletion: DeletionService,
targets: TargetRegistry,
) -> Vec<JoinHandle<()>> {
let recovery = config.recovery.clone();
let cleanup_attempts = attempts.clone();
let cleanup_recovery = recovery.clone();
let cleanup_targets = targets.clone();
let attempt_task = spawn_periodic(
"attempt_recovery",
config.recovery_interval,
status.clone(),
move || {
let attempts = attempts.clone();
let recovery = recovery.clone();
async move {
attempts
.reconcile_once(&recovery)
.await
.map(|_| ())
.map_err(|error| error.to_string())
}
},
);
let credential_task = spawn_periodic(
"credential_recovery",
config.credential_reconcile_interval,
status.clone(),
move || {
let credentials = credentials.clone();
async move {
credentials
.reconcile_secret_write_intents()
.await
.map(|_| ())
.map_err(|error| error.to_string())
}
},
);
let deletion_pool = pool.clone();
let deletion_task = spawn_periodic(
"result_deletion",
config.deletion_interval,
status.clone(),
move || {
let deletion = deletion.clone();
let pool = deletion_pool.clone();
async move { process_deletions(&pool, &deletion).await }
},
);
let cleanup_task = spawn_periodic(
"staging_cleanup",
config.recovery_interval,
status.clone(),
move || {
let attempts = cleanup_attempts.clone();
let recovery = cleanup_recovery.clone();
let targets = cleanup_targets.clone();
async move { process_staging_cleanup(&attempts, &recovery, &targets).await }
},
);
let event_pool = pool.clone();
let retention = config.public_api.event_retention;
let event_task = spawn_periodic(
"event_history_prune",
Duration::from_secs(60),
status.clone(),
move || {
let pool = event_pool.clone();
async move {
let retention =
time::Duration::try_from(retention).map_err(|error| error.to_string())?;
crate::prune_live_event_history(&pool, time::OffsetDateTime::now_utc() - retention)
.await
.map(|_| ())
.map_err(|error| error.to_string())
}
},
);
vec![
attempt_task,
credential_task,
deletion_task,
cleanup_task,
event_task,
]
}
async fn process_deletions(pool: &PgPool, deletion: &DeletionService) -> Result<(), String> {
let operation_ids = sqlx::query_scalar::<_, String>(
"SELECT operation_id FROM deletion_operations
WHERE state IN ('queued', 'running', 'retry_wait', 'blocked')
ORDER BY updated_at, operation_id LIMIT 32",
)
.fetch_all(pool)
.await
.map_err(|error| error.to_string())?;
for value in operation_ids {
let operation_id = OpaqueId::parse(value).map_err(|error| error.to_string())?;
match deletion.process_next(&operation_id).await {
Ok(
DrainOutcome::Idle
| DrainOutcome::ItemCompleted
| DrainOutcome::Blocked
| DrainOutcome::Completed,
) => {}
Err(error) => return Err(error.to_string()),
}
}
Ok(())
}
async fn process_staging_cleanup(
attempts: &AttemptService,
recovery: &RecoveryConfig,
targets: &TargetRegistry,
) -> Result<(), String> {
let Some(claim) = attempts
.claim_cleanup(recovery)
.await
.map_err(|error| error.to_string())?
else {
return Ok(());
};
let (target_store, bucket) = cleanup_target(&claim.identity);
let Some(target) = targets.resolve(target_store, bucket) else {
attempts
.retry_cleanup(&claim, "target_configuration")
.await
.map_err(|error| error.to_string())?;
return Err("cleanup target configuration is unavailable".into());
};
let effect = match &claim.identity {
CleanupIdentity::Object(object) => target
.delete_exact(&ObjectIdentity {
target_store_id: object.target_store.clone(),
bucket: object.bucket.clone(),
key: object.key.clone(),
version_id: object.version_id.clone(),
})
.await
.map(|_| None),
CleanupIdentity::Multipart(multipart) => target
.abort_multipart_exact(&MultipartIdentity {
target_store_id: multipart.target_store.clone(),
bucket: multipart.bucket.clone(),
key: multipart.key.clone(),
upload_id: multipart.upload_id.clone(),
})
.await
.map(|_| None),
CleanupIdentity::WriteIntent(intent) => {
target.discover_staging_write(&intent.key).await.map(Some)
}
};
match effect {
Ok(Some(discovered)) => {
let objects = discovered
.objects
.into_iter()
.map(|object| StagingObject {
target_store: object.target_store_id,
bucket: object.bucket,
key: object.key,
version_id: object.version_id,
})
.collect();
let multiparts = discovered
.multiparts
.into_iter()
.map(|multipart| StagingMultipart {
target_store: multipart.target_store_id,
bucket: multipart.bucket,
key: multipart.key,
upload_id: multipart.upload_id,
})
.collect();
attempts
.reconcile_write_intent(&claim, objects, multiparts)
.await
.map(|_| ())
.map_err(|error| error.to_string())
}
Ok(None) => attempts
.complete_cleanup(&claim)
.await
.map(|_| ())
.map_err(|error| error.to_string()),
Err(_) => {
attempts
.retry_cleanup(&claim, "target_unavailable")
.await
.map_err(|error| error.to_string())?;
Err("cleanup target operation failed".into())
}
}
}
fn cleanup_target(identity: &CleanupIdentity) -> (&str, &str) {
match identity {
CleanupIdentity::WriteIntent(intent) => (&intent.target_store, &intent.bucket),
CleanupIdentity::Object(object) => (&object.target_store, &object.bucket),
CleanupIdentity::Multipart(multipart) => (&multipart.target_store, &multipart.bucket),
}
}
fn public_server(
bind: SocketAddr,
state: PublicApiState,
) -> Result<actix_web::dev::ServerHandle, RuntimeError> {
let listener = TcpListener::bind(bind).map_err(runtime_error)?;
let server = HttpServer::new(move || {
App::new()
.wrap(middleware::Logger::new("%a \"%m %U\" %s %b %D"))
.wrap(middleware::from_fn(trace_request))
.app_data(web::Data::new(state.clone()))
.configure(configure_public_api)
})
.disable_signals()
.listen(listener)
.map_err(runtime_error)?
.run();
let handle = server.handle();
tokio::spawn(server);
Ok(handle)
}
fn worker_server(
bind: SocketAddr,
state: WorkerApiState,
tls: &WorkerTlsConfig,
) -> Result<actix_web::dev::ServerHandle, RuntimeError> {
let listener = TcpListener::bind(bind).map_err(runtime_error)?;
let server = HttpServer::new(move || {
App::new()
.wrap(middleware::Logger::new("%a \"%m %U\" %s %b %D"))
.wrap(middleware::from_fn(trace_request))
.app_data(web::Data::new(state.clone()))
.configure(configure_worker_api)
})
.disable_signals()
.on_connect(authenticate_rustls_worker)
.listen_rustls_0_23(listener, server_tls(tls)?)
.map_err(runtime_error)?
.run();
let handle = server.handle();
tokio::spawn(server);
Ok(handle)
}
fn server_tls(config: &WorkerTlsConfig) -> Result<ServerConfig, RuntimeError> {
let server_certificates = certificates(&config.certificate)?;
let key = private_key(&config.private_key)?;
let mut roots = RootCertStore::empty();
for certificate in certificates(&config.client_ca)? {
roots.add(certificate).map_err(runtime_error)?;
}
let verifier = WebPkiClientVerifier::builder(Arc::new(roots))
.build()
.map_err(runtime_error)?;
ServerConfig::builder()
.with_client_cert_verifier(verifier)
.with_single_cert(server_certificates, key)
.map_err(runtime_error)
}
fn certificates(path: &Path) -> Result<Vec<CertificateDer<'static>>, RuntimeError> {
let file = fs::File::open(path).map_err(|error| path_error(path, error))?;
let certificates = rustls_pemfile::certs(&mut BufReader::new(file))
.collect::<Result<Vec<_>, _>>()
.map_err(runtime_error)?;
if certificates.is_empty() {
return Err(RuntimeError::message(format!(
"{} contains no certificates",
path.display()
)));
}
Ok(certificates)
}
fn private_key(path: &Path) -> Result<PrivateKeyDer<'static>, RuntimeError> {
let file = fs::File::open(path).map_err(|error| path_error(path, error))?;
rustls_pemfile::private_key(&mut BufReader::new(file))
.map_err(runtime_error)?
.ok_or_else(|| RuntimeError::message(format!("{} contains no private key", path.display())))
}
async fn wait_for_shutdown() -> Result<(), RuntimeError> {
#[cfg(unix)]
{
let mut terminate =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
.map_err(runtime_error)?;
tokio::select! {
result = tokio::signal::ctrl_c() => result.map_err(runtime_error),
_ = terminate.recv() => Ok(()),
}
}
#[cfg(not(unix))]
tokio::signal::ctrl_c().await.map_err(runtime_error)
}
fn comma_values(name: &str, input: &str) -> Result<Vec<String>, RuntimeError> {
let values = input
.split(',')
.map(str::trim)
.map(str::to_owned)
.collect::<Vec<_>>();
if values.is_empty() || values.iter().any(String::is_empty) {
return Err(RuntimeError::message(format!(
"{name} must contain non-empty comma-separated values"
)));
}
Ok(values)
}
fn duration_millis(name: &str, default: u64) -> Result<Duration, RuntimeError> {
let milliseconds = env_parse(name, &default.to_string())?;
if milliseconds == 0 {
return Err(RuntimeError::message(format!(
"{name} must be greater than zero"
)));
}
Ok(Duration::from_millis(milliseconds))
}
fn runtime_error(error: impl std::fmt::Display) -> RuntimeError {
RuntimeError::message(error.to_string())
}
fn file_error(name: &str, error: io::Error) -> RuntimeError {
RuntimeError::message(format!("cannot read {name}: {}", error.kind()))
}
fn path_error(path: &Path, error: io::Error) -> RuntimeError {
RuntimeError::message(format!("cannot read {}: {}", path.display(), error.kind()))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn rejects_empty_comma_values() {
assert!(comma_values("TEST", "one,,two").is_err());
}
#[test]
fn file_errors_do_not_include_file_contents() {
let error = file_error(
"REMORA_CURSOR_SECRET_FILE",
io::Error::from(io::ErrorKind::NotFound),
);
assert_eq!(
error.to_string(),
"cannot read REMORA_CURSOR_SECRET_FILE: entity not found"
);
}
}

2
crates/remora-api/src/lib.rs
View file

@ -1,7 +1,9 @@
mod application;
mod migrate;
mod public;
mod worker;
pub use application::{ApiApplication, ApiApplicationConfig};
pub use migrate::{migrate, migrator};
pub use public::{
PublicApiConfig, PublicApiServices, PublicApiState, configure_public_api,

10
crates/remora-api/src/main.rs
View file

@ -1,3 +1,9 @@
fn main() {
println!("{}", remora_api::COMPONENT_NAME);
use remora_api::{ApiApplication, ApiApplicationConfig};
#[actix_web::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
remora_runtime::init_logging()?;
let config = ApiApplicationConfig::from_env()?;
ApiApplication::new(config).run().await?;
Ok(())
}

5
crates/remora-api/src/worker.rs
View file

@ -96,6 +96,7 @@ pub fn configure_worker_api(config: &mut web::ServiceConfig) {
.limit(1024 * 1024)
.error_handler(|_, _| WorkerApiError::invalid_request().into()),
)
.route("/health/ready", web::get().to(worker_ready))
.route("/jobs/{job_id}/claim", web::post().to(claim))
.route("/attempts/{attempt_id}/renew", web::post().to(renew))
.route("/attempts/{attempt_id}/activate", web::post().to(activate))
@ -131,6 +132,10 @@ pub fn configure_worker_api(config: &mut web::ServiceConfig) {
);
}
async fn worker_ready(_: AuthenticatedWorker) -> HttpResponse {
HttpResponse::Ok().finish()
}
async fn register_staging_object(
_: AuthenticatedWorker,
state: web::Data<WorkerApiState>,

45
crates/remora-api/tests/deployment.rs Normal file
View file

@ -0,0 +1,45 @@
#[allow(dead_code)]
mod support;
use remora_contracts::JobResourceDto;
use reqwest::StatusCode;
use serde_json::{Value, json};
use support::{mint_provider_token, required_env};
#[tokio::test]
async fn deployed_api_accepts_an_authorized_job_for_coordination_rebuild() {
if std::env::var("REMORA_DEPLOYMENT_REQUIRE").is_err() {
return;
}
let base = required_env("REMORA_DEPLOYMENT_API_URL")
.trim_end_matches('/')
.to_owned();
let client = reqwest::Client::builder()
.redirect(reqwest::redirect::Policy::none())
.build()
.unwrap();
let unauthenticated = client.get(format!("{base}/jobs")).send().await.unwrap();
assert_eq!(unauthenticated.status(), StatusCode::UNAUTHORIZED);
let token = mint_provider_token().await.unwrap();
let submitted = client
.post(format!("{base}/jobs"))
.bearer_auth(token)
.header("Idempotency-Key", "deployment-coordination-rebuild")
.json(&json!({
"source": {
"kind": "http",
"url": "https://example.com/deployment-smoke",
"result_path": "artifact.bin"
},
"label": "deployment-coordination-rebuild"
}))
.send()
.await
.unwrap();
assert_eq!(submitted.status(), StatusCode::ACCEPTED);
let body: Value = submitted.json().await.unwrap();
serde_json::from_value::<JobResourceDto>(body.clone()).unwrap();
assert_eq!(body["phase"], "Queued");
}

5
crates/remora-api/tests/live_events.rs
View file

@ -115,10 +115,7 @@ impl Harness {
let attempts = AttemptService::new(
pool.clone(),
coordination.clone(),
AttemptConfig {
max_attempts: 3,
local_limit: 8,
},
AttemptConfig { max_attempts: 3 },
)
.unwrap();
let admission = AdmissionService::new(

5
crates/remora-api/tests/public_api.rs
View file

@ -117,10 +117,7 @@ impl Harness {
let attempts = AttemptService::new(
pool.clone(),
coordination.clone(),
AttemptConfig {
max_attempts: 3,
local_limit: 8,
},
AttemptConfig { max_attempts: 3 },
)
.unwrap();
let vault = VaultClient::new(VaultConfig::from_env().unwrap()).unwrap();

5
crates/remora-api/tests/worker_api.rs
View file

@ -88,10 +88,7 @@ impl Harness {
AttemptService::new(
self.pool.clone(),
self.coordination.clone(),
AttemptConfig {
max_attempts: 3,
local_limit: 8,
},
AttemptConfig { max_attempts: 3 },
)
.unwrap()
}

29
crates/remora-attempt/src/recovery.rs
View file

@ -97,7 +97,6 @@ impl AttemptService {
.coordination
.release(authority.job_id.as_str(), &lease_owner(authority))
.await;
self.release_local(&authority.job_id);
}
Ok(outcome)
}
@ -154,7 +153,6 @@ impl AttemptService {
.coordination
.release(authority.job_id.as_str(), &lease_owner(authority))
.await;
self.release_local(&authority.job_id);
}
Ok(decision)
}
@ -649,7 +647,6 @@ impl AttemptService {
};
let mut report = ReconcileReport::default();
if row.phase == "retry_wait" {
self.release_local_if_stopped(job_id).await?;
if row.retry_not_before.is_none() {
let due =
OffsetDateTime::now_utc() + time_duration(config.retry_delay, "retry_delay")?;
@ -731,7 +728,6 @@ impl AttemptService {
}
}
if parse_phase(&row.phase)?.is_terminal() {
self.release_local_if_stopped(job_id).await?;
let expired = self.coordination.expire_lease(job_id.as_str()).await?;
self.coordination.clean_terminal(job_id.as_str()).await?;
if expired.code.is_success() {
@ -774,9 +770,6 @@ impl AttemptService {
})
.await?;
let _ = self.coordination.expire_lease(job_id.as_str()).await?;
if reaped.release_local {
self.release_local(job_id);
}
Ok(reaped.changed)
}
@ -861,22 +854,6 @@ impl AttemptService {
.await?;
Ok(members.rows_affected() + manifests.rows_affected())
}
async fn release_local_if_stopped(&self, job_id: &OpaqueId) -> Result<(), AttemptError> {
let stopped = sqlx::query_scalar::<_, bool>(
"SELECT NOT EXISTS (
SELECT 1 FROM attempts
WHERE job_id = $1 AND state = 'executing' AND worker_down_at IS NULL
)",
)
.bind(job_id.as_str())
.fetch_one(&self.pool)
.await?;
if stopped {
self.release_local(job_id);
}
Ok(())
}
}
async fn cooperative_stop_transaction(
@ -1242,10 +1219,7 @@ async fn reap_lost_attempt(
.await?;
enqueue(tx, "job.updated", json!({ "job_id": job_id })).await?;
}
Ok(ReapEffect {
changed: true,
release_local: row.state == "authorized" || row.worker_down_at.is_some(),
})
Ok(ReapEffect { changed: true })
}
async fn validate_staging_key(
@ -1393,7 +1367,6 @@ struct LockedReapRow {
#[derive(Default)]
struct ReapEffect {
changed: bool,
release_local: bool,
}
#[derive(FromRow)]

37
crates/remora-attempt/src/service.rs
View file

@ -2,7 +2,7 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use remora_coordination::{
ActivateOptions, CoordinationStore, LeaseOwner, LocalPermitGate, MAX_SAFE_INTEGER, OpCode,
ActivateOptions, CoordinationStore, LeaseOwner, MAX_SAFE_INTEGER, OpCode,
};
use remora_domain::{OpaqueId, fingerprint, ports::MutationScope};
use remora_durable::{
@ -29,15 +29,11 @@ impl AttemptFaultInjector for NoFaults {
#[derive(Clone, Debug)]
pub struct AttemptConfig {
pub max_attempts: u64,
pub local_limit: u64,
}
impl Default for AttemptConfig {
fn default() -> Self {
Self {
max_attempts: 3,
local_limit: 4,
}
Self { max_attempts: 3 }
}
}
@ -45,7 +41,6 @@ impl Default for AttemptConfig {
pub struct AttemptService {
pub(crate) pool: PgPool,
pub(crate) coordination: CoordinationStore,
local_permits: LocalPermitGate,
pub(crate) operation_subject: OpaqueId,
pub(crate) faults: Arc<dyn AttemptFaultInjector>,
pub(crate) config: AttemptConfig,
@ -71,15 +66,9 @@ impl AttemptService {
"max_attempts must fit the coordination safe-integer range".into(),
));
}
if config.local_limit == 0 {
return Err(AttemptError::Invalid(
"local_limit must be greater than zero".into(),
));
}
Ok(Self {
pool,
coordination,
local_permits: LocalPermitGate::new(config.local_limit),
operation_subject: OpaqueId::parse("remora_attempt_service")
.map_err(|error| AttemptError::Invalid(error.to_string()))?,
faults,
@ -114,12 +103,6 @@ impl AttemptService {
worker_incarnation: &OpaqueId,
operation_id: &OpaqueId,
) -> Result<ClaimOutcome, AttemptError> {
if self.local_permits.acquire(job_id.as_str()).is_err() {
return Ok(ClaimOutcome::Rejected {
reason: ClaimRejection::LocalCapacity,
disposition: None,
});
}
let request = ClaimRequest {
job_id,
worker_incarnation,
@ -177,22 +160,16 @@ impl AttemptService {
if reserved.load(Ordering::Acquire) {
self.compensate_reservation(job_id, worker_incarnation)
.await;
} else {
self.local_permits.release(job_id.as_str());
}
return Err(error.into());
}
};
self.faults.hit(AttemptFaultPoint::ClaimAfterDurable);
if outcome.disposition == MutationDisposition::Conflict {
self.local_permits.release(job_id.as_str());
return Err(AttemptError::OperationConflict);
}
let decision: ClaimOutcome = serde_json::from_value(outcome.response)
.map_err(|error| AttemptError::Invalid(error.to_string()))?;
if matches!(decision, ClaimOutcome::Rejected { .. }) {
self.local_permits.release(job_id.as_str());
}
Ok(decision.with_disposition(outcome.disposition))
}
@ -564,21 +541,11 @@ impl AttemptService {
}))
}
pub fn release_local(&self, job_id: &OpaqueId) -> bool {
self.local_permits.release_all(job_id.as_str())
}
#[must_use]
pub fn local_used(&self) -> u64 {
self.local_permits.used()
}
async fn compensate_reservation(&self, job_id: &OpaqueId, worker: &OpaqueId) {
let _ = self
.coordination
.free_reserve(job_id.as_str(), Some(worker.as_str()))
.await;
self.local_permits.release(job_id.as_str());
}
pub(crate) async fn release_exact(&self, authority: &AttemptAuthority) {

1
crates/remora-attempt/src/types.rs
View file

@ -58,7 +58,6 @@ pub struct AttemptAuthority {
pub enum ClaimRejection {
JobUnavailable,
AttemptsExhausted,
LocalCapacity,
GlobalCapacity,
ReservationHeld,
}

62
crates/remora-attempt/tests/authority.rs
View file

@ -61,31 +61,20 @@ impl Harness {
Some(Self { pool, coordination })
}
fn service(&self, max_attempts: u64, local_limit: u64) -> AttemptService {
fn service(&self, max_attempts: u64) -> AttemptService {
AttemptService::new(
self.pool.clone(),
self.coordination.clone(),
AttemptConfig {
max_attempts,
local_limit,
},
AttemptConfig { max_attempts },
)
.expect("attempt service")
}
fn service_with_fault(
&self,
point: AttemptFaultPoint,
max_attempts: u64,
local_limit: u64,
) -> AttemptService {
fn service_with_fault(&self, point: AttemptFaultPoint, max_attempts: u64) -> AttemptService {
AttemptService::new_with_fault_injector(
self.pool.clone(),
self.coordination.clone(),
AttemptConfig {
max_attempts,
local_limit,
},
AttemptConfig { max_attempts },
Arc::new(PanicOnce::new(point)),
)
.expect("fault-injected attempt service")
@ -154,7 +143,7 @@ async fn claim_replay_conflict_and_fence_allocation_are_durable() {
let job = harness.job().await;
let worker = unique_id("worker");
let operation = unique_id("claimop");
let service = harness.service(2, 2);
let service = harness.service(2);
let first = claimed(
service
@ -189,7 +178,6 @@ async fn claim_replay_conflict_and_fence_allocation_are_durable() {
service.recover_inactive(&authority).await.unwrap(),
RecoveryOutcome::Reaped
);
service.release_local(&job);
harness.coordination.enqueue(job.as_str()).await.unwrap();
let second = claimed(
service
@ -201,7 +189,6 @@ async fn claim_replay_conflict_and_fence_allocation_are_durable() {
sleep(Duration::from_millis(2200)).await;
let second_authority = service.authority(&second.2).await.unwrap().unwrap();
service.recover_inactive(&second_authority).await.unwrap();
service.release_local(&job);
harness.coordination.enqueue(job.as_str()).await.unwrap();
let exhausted = service
.claim(&job, &worker, &unique_id("claimop"))
@ -217,7 +204,7 @@ async fn activation_replay_requires_the_complete_live_authority_tuple() {
};
let job = harness.job().await;
let worker = unique_id("worker");
let service = harness.service(3, 2);
let service = harness.service(3);
let claim_operation = unique_id("claimop");
let (_, _, attempt) = claimed(
service
@ -240,10 +227,7 @@ async fn activation_replay_requires_the_complete_live_authority_tuple() {
let unavailable = AttemptService::new(
unavailable_pool,
harness.coordination.clone(),
AttemptConfig {
max_attempts: 3,
local_limit: 2,
},
AttemptConfig { max_attempts: 3 },
)
.unwrap();
assert!(
@ -338,7 +322,7 @@ async fn activation_and_inactive_recovery_serialize_on_durable_authority() {
};
let job = harness.job().await;
let worker = unique_id("worker");
let service = harness.service(3, 2);
let service = harness.service(3);
let (_, _, attempt) = claimed(
service
.claim(&job, &worker, &unique_id("claimop"))
@ -380,11 +364,11 @@ async fn activation_and_inactive_recovery_serialize_on_durable_authority() {
}
#[tokio::test]
async fn local_and_global_permits_bound_claims_and_survive_lease_expiry() {
async fn global_permits_bound_claims_and_survive_lease_expiry() {
let Some(harness) = Harness::new("permits", Duration::from_millis(500), 1).await else {
return;
};
let service = harness.service(3, 1);
let service = harness.service(3);
let worker = unique_id("worker");
let first_job = harness.job().await;
let (_, _, attempt) = claimed(
@ -399,8 +383,6 @@ async fn local_and_global_permits_bound_claims_and_survive_lease_expiry() {
.await
.unwrap();
sleep(Duration::from_millis(600)).await;
assert_eq!(service.local_used(), 1);
let second_job = harness.job().await;
assert!(matches!(
service
@ -409,13 +391,11 @@ async fn local_and_global_permits_bound_claims_and_survive_lease_expiry() {
.unwrap(),
ClaimOutcome::Rejected { .. }
));
service.release_local(&first_job);
harness
.coordination
.expire_lease(first_job.as_str())
.await
.unwrap();
assert_eq!(service.local_used(), 0);
}
#[tokio::test]
@ -425,7 +405,7 @@ async fn precommit_lease_is_compensated_when_durable_authority_was_reaped() {
};
let job = harness.job().await;
let worker = unique_id("worker");
let service = harness.service(3, 2);
let service = harness.service(3);
let (_, _, attempt) = claimed(
service
.claim(&job, &worker, &unique_id("claimop"))
@ -479,11 +459,11 @@ async fn crash_injection_covers_every_cross_store_and_lost_response_boundary() {
return;
};
let worker = unique_id("worker");
let normal = harness.service(32, 32);
let normal = harness.service(32);
let before_reserve_job = harness.job().await;
crash_claim(
harness.service_with_fault(AttemptFaultPoint::ClaimBeforeReserve, 32, 32),
harness.service_with_fault(AttemptFaultPoint::ClaimBeforeReserve, 32),
before_reserve_job.clone(),
worker.clone(),
unique_id("claimop"),
@ -500,7 +480,7 @@ async fn crash_injection_covers_every_cross_store_and_lost_response_boundary() {
let after_reserve_job = harness.job().await;
crash_claim(
harness.service_with_fault(AttemptFaultPoint::ClaimAfterReserve, 32, 32),
harness.service_with_fault(AttemptFaultPoint::ClaimAfterReserve, 32),
after_reserve_job.clone(),
worker.clone(),
unique_id("claimop"),
@ -523,7 +503,7 @@ async fn crash_injection_covers_every_cross_store_and_lost_response_boundary() {
let lost_claim_job = harness.job().await;
let lost_claim_operation = unique_id("claimop");
crash_claim(
harness.service_with_fault(AttemptFaultPoint::ClaimAfterDurable, 32, 32),
harness.service_with_fault(AttemptFaultPoint::ClaimAfterDurable, 32),
lost_claim_job.clone(),
worker.clone(),
lost_claim_operation.clone(),
@ -544,7 +524,7 @@ async fn crash_injection_covers_every_cross_store_and_lost_response_boundary() {
let before_lease_operation = unique_id("activateop");
crash_activate(
harness.service_with_fault(AttemptFaultPoint::ActivationBeforeLease, 32, 32),
harness.service_with_fault(AttemptFaultPoint::ActivationBeforeLease, 32),
lost_claim_authority.clone(),
before_lease_operation.clone(),
)
@ -579,7 +559,7 @@ async fn crash_injection_covers_every_cross_store_and_lost_response_boundary() {
.unwrap();
let after_lease_operation = unique_id("activateop");
crash_activate(
harness.service_with_fault(AttemptFaultPoint::ActivationAfterLease, 32, 32),
harness.service_with_fault(AttemptFaultPoint::ActivationAfterLease, 32),
after_lease_authority.clone(),
after_lease_operation.clone(),
)
@ -614,7 +594,7 @@ async fn crash_injection_covers_every_cross_store_and_lost_response_boundary() {
.unwrap();
let lost_activation_operation = unique_id("activateop");
crash_activate(
harness.service_with_fault(AttemptFaultPoint::ActivationAfterDurable, 32, 32),
harness.service_with_fault(AttemptFaultPoint::ActivationAfterDurable, 32),
lost_activation_authority.clone(),
lost_activation_operation.clone(),
)
@ -633,12 +613,12 @@ async fn crash_injection_covers_every_cross_store_and_lost_response_boundary() {
);
crash_renew(
harness.service_with_fault(AttemptFaultPoint::RenewalBeforeLease, 32, 32),
harness.service_with_fault(AttemptFaultPoint::RenewalBeforeLease, 32),
executing.clone(),
)
.await;
crash_renew(
harness.service_with_fault(AttemptFaultPoint::RenewalAfterLease, 32, 32),
harness.service_with_fault(AttemptFaultPoint::RenewalAfterLease, 32),
executing.clone(),
)
.await;
@ -666,7 +646,7 @@ async fn crash_injection_covers_every_cross_store_and_lost_response_boundary() {
sleep(Duration::from_millis(600)).await;
for (point, authority) in recovery_cases {
let used_before = harness.coordination.global_used().await.unwrap();
crash_recovery(harness.service_with_fault(point, 32, 32), authority.clone()).await;
crash_recovery(harness.service_with_fault(point, 32), authority.clone()).await;
let expected = if point == AttemptFaultPoint::RecoveryAfterDurable {
RecoveryOutcome::AlreadyInactive
} else {

6
crates/remora-attempt/tests/mutations.rs
View file

@ -67,10 +67,7 @@ impl Harness {
AttemptService::new(
self.pool.clone(),
self.coordination.clone(),
AttemptConfig {
max_attempts,
local_limit: 8,
},
AttemptConfig { max_attempts },
)
.expect("attempt service")
}
@ -637,7 +634,6 @@ async fn failure_disposition_retries_with_budget_and_becomes_terminal_when_exhau
)
.await
.unwrap();
service.release_local(&job);
harness.coordination.enqueue(job.as_str()).await.unwrap();
let second = harness.executing(&service, &job).await;
assert_eq!(second.fence, 2);

25
crates/remora-attempt/tests/recovery.rs
View file

@ -163,10 +163,7 @@ impl Harness {
AttemptService::new(
self.pool.clone(),
self.coordination.clone(),
AttemptConfig {
max_attempts,
local_limit: 16,
},
AttemptConfig { max_attempts },
)
.expect("attempt service")
}
@ -304,10 +301,7 @@ async fn activation_and_recovery_share_authority_serialization() {
let service = AttemptService::new_with_fault_injector(
harness.pool.clone(),
harness.coordination.clone(),
AttemptConfig {
max_attempts: 3,
local_limit: 16,
},
AttemptConfig { max_attempts: 3 },
activation_gate.clone(),
)
.unwrap();
@ -371,10 +365,7 @@ async fn activation_and_recovery_share_authority_serialization() {
let service = AttemptService::new_with_fault_injector(
harness.pool.clone(),
harness.coordination.clone(),
AttemptConfig {
max_attempts: 3,
local_limit: 16,
},
AttemptConfig { max_attempts: 3 },
recovery_gate.clone(),
)
.unwrap();
@ -485,10 +476,7 @@ async fn coordination_partition_rolls_back_and_restart_converges() {
let service = AttemptService::new(
harness.pool.clone(),
store.clone(),
AttemptConfig {
max_attempts: 3,
local_limit: 16,
},
AttemptConfig { max_attempts: 3 },
)
.unwrap();
let job = harness.job().await;
@ -529,10 +517,7 @@ async fn coordination_partition_rolls_back_and_restart_converges() {
let restarted = AttemptService::new(
harness.pool.clone(),
resumed_store.clone(),
AttemptConfig {
max_attempts: 3,
local_limit: 16,
},
AttemptConfig { max_attempts: 3 },
)
.unwrap();
assert_eq!(

8
crates/remora-coordination/src/store.rs
View file

@ -1,6 +1,6 @@
use std::time::Duration;
use redis::aio::ConnectionManager;
use redis::aio::{ConnectionManager, ConnectionManagerConfig};
use redis::{AsyncCommands, Client, FromRedisValue, RedisResult, Value};
use crate::error::CoordinationError;
@ -51,7 +51,11 @@ impl CoordinationStore {
let client = Client::open(url).map_err(|error| {
CoordinationError::config(format!("invalid coordination URL: {error}"))
})?;
let conn = ConnectionManager::new(client).await?;
let connection_config = ConnectionManagerConfig::new()
.set_connection_timeout(Duration::from_secs(3))
.set_response_timeout(Duration::from_secs(5))
.set_max_delay(1_000);
let conn = ConnectionManager::new_with_config(client, connection_config).await?;
Ok(Self {
conn,
keys: KeySpace::new(namespace),

32
crates/remora-credentials/src/vault.rs
View file

@ -53,7 +53,7 @@ impl fmt::Debug for VaultConfig {
impl VaultConfig {
pub fn from_env() -> Result<Self, CredentialError> {
let address = required_env("REMORA_VAULT_ADDR")?;
let token = required_env("REMORA_VAULT_TOKEN")?;
let token = vault_token()?;
let full_path = required_env("REMORA_VAULT_PATH")?;
let (mount, path_prefix) = split_vault_path(&full_path)?;
Ok(Self {
@ -95,6 +95,21 @@ impl VaultConfig {
}
}
fn vault_token() -> Result<String, CredentialError> {
if let Ok(token) = required_env("REMORA_VAULT_TOKEN") {
return Ok(token);
}
let path = required_env("REMORA_VAULT_TOKEN_FILE")?;
let token = std::fs::read_to_string(path).map_err(|error| {
CredentialError::Config(format!("cannot read Vault token: {}", error.kind()))
})?;
let token = token.trim().to_owned();
if token.is_empty() {
return Err(CredentialError::Config("Vault token file is empty".into()));
}
Ok(token)
}
#[derive(Clone)]
pub struct VaultClient {
http: Client,
@ -131,6 +146,21 @@ impl VaultClient {
&self.config
}
pub async fn check(&self) -> Result<(), CredentialError> {
let url = format!(
"{}/v1/sys/health?standbyok=true&perfstandbyok=true",
self.config.address.trim_end_matches('/')
);
let response = self.http.get(url).send().await.map_err(map_transport)?;
let status = response.status();
let _ = response.bytes().await;
if status.is_success() {
Ok(())
} else {
Err(map_http_status(status, "check vault health"))
}
}
/// Create secret material and return the immutable KV v2 version number.
///
/// The write uses CAS 0 so an existing intent-bound path is never overwritten.

16
crates/remora-runtime/Cargo.toml Normal file
View file

@ -0,0 +1,16 @@
[package]
name = "remora-runtime"
edition.workspace = true
publish = false
rust-version.workspace = true
version.workspace = true
[dependencies]
actix-web = "=4.14.0"
getrandom = "0.3.4"
remora-coordination = { path = "../remora-coordination" }
serde.workspace = true
sqlx.workspace = true
tokio.workspace = true
tracing.workspace = true
tracing-subscriber.workspace = true

358
crates/remora-runtime/src/lib.rs Normal file
View file

@ -0,0 +1,358 @@
use std::{
env, fmt,
future::Future,
net::SocketAddr,
pin::Pin,
str::FromStr,
sync::{
Arc,
atomic::{AtomicBool, AtomicU64, Ordering},
},
time::Duration,
};
use actix_web::{
App, Error, HttpResponse, HttpServer,
body::MessageBody,
dev::{ServiceRequest, ServiceResponse},
http::header::{HeaderName, HeaderValue},
middleware::{Next, from_fn},
web,
};
use remora_coordination::CoordinationStore;
use serde::Serialize;
use sqlx::PgPool;
use tokio::task::JoinHandle;
use tracing::Instrument as _;
use tracing_subscriber::EnvFilter;
#[derive(Clone, Debug)]
pub struct ManagementConfig {
pub bind: SocketAddr,
pub component: String,
}
impl ManagementConfig {
pub fn from_env(component: impl Into<String>) -> Result<Self, RuntimeError> {
Ok(Self {
bind: env_parse("REMORA_MANAGEMENT_BIND", "0.0.0.0:9090")?,
component: component.into(),
})
}
}
#[derive(Clone)]
pub struct RuntimeStatus {
component: Arc<str>,
accepting_traffic: Arc<AtomicBool>,
background_runs: Arc<AtomicU64>,
background_failures: Arc<AtomicU64>,
pool: PgPool,
coordination: CoordinationStore,
readiness_checks: Vec<ReadinessCheck>,
}
type ReadinessCheck = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = bool> + Send>> + Send + Sync>;
impl RuntimeStatus {
#[must_use]
pub fn new(
component: impl Into<String>,
pool: PgPool,
coordination: CoordinationStore,
) -> Self {
Self {
component: Arc::from(component.into()),
accepting_traffic: Arc::new(AtomicBool::new(false)),
background_runs: Arc::new(AtomicU64::new(0)),
background_failures: Arc::new(AtomicU64::new(0)),
pool,
coordination,
readiness_checks: Vec::new(),
}
}
#[must_use]
pub fn with_readiness_check<F, Fut>(mut self, check: F) -> Self
where
F: Fn() -> Fut + Send + Sync + 'static,
Fut: Future<Output = bool> + Send + 'static,
{
self.readiness_checks
.push(Arc::new(move || Box::pin(check())));
self
}
pub fn set_accepting_traffic(&self, value: bool) {
self.accepting_traffic.store(value, Ordering::Release);
}
pub fn record_background_run(&self, succeeded: bool) {
self.background_runs.fetch_add(1, Ordering::Relaxed);
if !succeeded {
self.background_failures.fetch_add(1, Ordering::Relaxed);
}
}
async fn ready(&self) -> bool {
if !self.accepting_traffic.load(Ordering::Acquire) {
return false;
}
let database = sqlx::query_scalar::<_, i32>("SELECT 1")
.fetch_one(&self.pool)
.await
.is_ok();
let coordination = self.coordination.global_used().await.is_ok();
if !database || !coordination {
return false;
}
for check in &self.readiness_checks {
if !check().await {
return false;
}
}
true
}
fn metrics(&self, ready: bool) -> String {
let component = prometheus_label(&self.component);
format!(
"# TYPE remora_build_info gauge\n\
remora_build_info{{component=\"{component}\"}} 1\n\
# TYPE remora_ready gauge\n\
remora_ready{{component=\"{component}\"}} {}\n\
# TYPE remora_background_runs_total counter\n\
remora_background_runs_total{{component=\"{component}\"}} {}\n\
# TYPE remora_background_failures_total counter\n\
remora_background_failures_total{{component=\"{component}\"}} {}\n",
u8::from(ready),
self.background_runs.load(Ordering::Relaxed),
self.background_failures.load(Ordering::Relaxed),
)
}
}
#[derive(Debug)]
pub struct RuntimeError(String);
impl RuntimeError {
#[must_use]
pub fn message(message: impl Into<String>) -> Self {
Self(message.into())
}
}
impl fmt::Display for RuntimeError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.0)
}
}
impl std::error::Error for RuntimeError {}
pub fn init_logging() -> Result<(), RuntimeError> {
let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
tracing_subscriber::fmt()
.with_env_filter(filter)
.json()
.with_current_span(true)
.with_span_list(true)
.try_init()
.map_err(|error| RuntimeError::message(format!("initialize logging: {error}")))
}
pub fn required_env(name: &str) -> Result<String, RuntimeError> {
env::var(name)
.ok()
.filter(|value| !value.is_empty())
.ok_or_else(|| RuntimeError::message(format!("{name} is required")))
}
pub fn env_string(name: &str, default: &str) -> String {
env::var(name)
.ok()
.filter(|value| !value.is_empty())
.unwrap_or_else(|| default.to_owned())
}
pub fn env_parse<T>(name: &str, default: &str) -> Result<T, RuntimeError>
where
T: FromStr,
T::Err: fmt::Display,
{
env_string(name, default)
.parse()
.map_err(|error| RuntimeError::message(format!("invalid {name}: {error}")))
}
pub fn env_duration(name: &str, default_seconds: u64) -> Result<Duration, RuntimeError> {
let seconds = env_parse(name, &default_seconds.to_string())?;
if seconds == 0 {
return Err(RuntimeError::message(format!(
"{name} must be greater than zero"
)));
}
Ok(Duration::from_secs(seconds))
}
pub fn spawn_periodic<F, Fut>(
name: &'static str,
interval: Duration,
status: RuntimeStatus,
mut operation: F,
) -> JoinHandle<()>
where
F: FnMut() -> Fut + Send + 'static,
Fut: Future<Output = Result<(), String>> + Send,
{
tokio::spawn(async move {
let mut ticker = tokio::time::interval(interval);
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
ticker.tick().await;
let result = operation().await;
status.record_background_run(result.is_ok());
match result {
Ok(()) => tracing::debug!(background_operation = name, "background pass completed"),
Err(error) => {
tracing::error!(background_operation = name, error, "background pass failed")
}
}
}
})
}
pub fn start_management_server(
config: ManagementConfig,
status: RuntimeStatus,
) -> std::io::Result<actix_web::dev::ServerHandle> {
let server = HttpServer::new(move || {
App::new()
.wrap(from_fn(trace_request))
.app_data(web::Data::new(status.clone()))
.route("/health/live", web::get().to(live))
.route("/health/ready", web::get().to(ready))
.route("/metrics", web::get().to(metrics))
})
.disable_signals()
.bind(config.bind)?
.run();
let handle = server.handle();
actix_web::rt::spawn(server);
Ok(handle)
}
pub async fn trace_request(
request: ServiceRequest,
next: Next<impl MessageBody>,
) -> Result<ServiceResponse<impl MessageBody>, Error> {
let trace_id = request
.headers()
.get("traceparent")
.and_then(|value| value.to_str().ok())
.and_then(parent_trace_id)
.unwrap_or_else(random_hex::<16>);
let span_id = random_hex::<8>();
let response_trace = format!("00-{trace_id}-{span_id}-01");
let span = tracing::info_span!(
"http_request",
trace_id,
method = %request.method(),
path = request.path(),
);
let mut response = next.call(request).instrument(span).await?;
response.headers_mut().insert(
HeaderName::from_static("traceparent"),
HeaderValue::from_str(&response_trace).expect("generated traceparent is valid"),
);
Ok(response)
}
async fn live() -> HttpResponse {
HttpResponse::Ok().json(Health { status: "live" })
}
async fn ready(status: web::Data<RuntimeStatus>) -> HttpResponse {
if status.ready().await {
HttpResponse::Ok().json(Health { status: "ready" })
} else {
HttpResponse::ServiceUnavailable().json(Health {
status: "not_ready",
})
}
}
async fn metrics(status: web::Data<RuntimeStatus>) -> HttpResponse {
HttpResponse::Ok()
.content_type("text/plain; version=0.0.4")
.body(status.metrics(status.ready().await))
}
#[derive(Serialize)]
struct Health {
status: &'static str,
}
fn prometheus_label(value: &str) -> String {
value.replace('\\', "\\\\").replace('"', "\\\"")
}
fn parent_trace_id(value: &str) -> Option<String> {
let mut parts = value.split('-');
let version = parts.next()?;
let trace_id = parts.next()?;
let parent_id = parts.next()?;
let flags = parts.next()?;
if parts.next().is_some()
|| version != "00"
|| trace_id.len() != 32
|| parent_id.len() != 16
|| flags.len() != 2
|| trace_id.bytes().all(|byte| byte == b'0')
|| !trace_id.bytes().all(|byte| byte.is_ascii_hexdigit())
|| !parent_id.bytes().all(|byte| byte.is_ascii_hexdigit())
|| !flags.bytes().all(|byte| byte.is_ascii_hexdigit())
{
return None;
}
Some(trace_id.to_ascii_lowercase())
}
fn random_hex<const N: usize>() -> String {
let mut bytes = [0_u8; N];
getrandom::fill(&mut bytes).expect("operating system randomness is required");
let mut value = String::with_capacity(N * 2);
for byte in bytes {
use std::fmt::Write as _;
write!(value, "{byte:02x}").expect("writing to String cannot fail");
}
value
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn escapes_prometheus_labels() {
assert_eq!(prometheus_label("api\"\\node"), "api\\\"\\\\node");
}
#[test]
fn runtime_errors_do_not_capture_environment_values() {
let error = RuntimeError::message("REMORA_SECRET is required");
assert_eq!(error.to_string(), "REMORA_SECRET is required");
}
#[test]
fn accepts_only_w3c_trace_identifiers() {
assert_eq!(
parent_trace_id("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"),
Some("4bf92f3577b34da6a3ce929d0e0e4736".into())
);
assert!(
parent_trace_id("00-00000000000000000000000000000000-00f067aa0ba902b7-01").is_none()
);
assert!(parent_trace_id("invalid").is_none());
}
}

4
crates/remora-target/src/store.rs
View file

@ -190,6 +190,10 @@ impl TargetStore {
&self.config
}
pub async fn check(&self) -> Result<(), TargetError> {
self.operator.check().await.map_err(map_opendal)
}
/// Open a create-only streaming writer for one attempt object.
pub async fn open_create_only(
&self,

Some files were not shown because too many files have changed in this diff Show more