diff --git a/services/api/src/handlers.rs b/services/api/src/handlers.rs index 47171e04..669e3d8f 100644 --- a/services/api/src/handlers.rs +++ b/services/api/src/handlers.rs @@ -320,7 +320,7 @@ pub async fn newsletter_subscribe( tracing::info!(request_id, email = %email, source = %source, ip = %ip, "newsletter subscription attempt"); Ok(( - StatusCode::OK, + StatusCode::ACCEPTED, Json(NewsletterResponse { success: true, message: "Please check your email to confirm your subscription.".to_string(), @@ -975,6 +975,8 @@ pub async fn email_queue_stats( .await .map_err(into_api_error)?; + state.metrics.set_dlq_size(stats.dead_letter as i64); + Ok((StatusCode::OK, Json(stats))) } diff --git a/services/api/src/metrics.rs b/services/api/src/metrics.rs index e363f740..e7f3bf87 100644 --- a/services/api/src/metrics.rs +++ b/services/api/src/metrics.rs @@ -1,7 +1,7 @@ use std::time::Duration; use anyhow::Context; -use prometheus::{Encoder, HistogramVec, IntCounterVec, Registry, TextEncoder}; +use prometheus::{Encoder, HistogramVec, IntCounterVec, IntGauge, Registry, TextEncoder}; #[derive(Clone)] pub struct Metrics { @@ -13,6 +13,7 @@ pub struct Metrics { rpc_errors: IntCounterVec, rpc_fallbacks: IntCounterVec, db_timeouts: IntCounterVec, + email_dlq_size: IntGauge, } impl Metrics { @@ -70,6 +71,12 @@ impl Metrics { ) .context("db_timeouts metric")?; + let email_dlq_size = IntGauge::new( + "email_dlq_size", + "Number of email jobs currently in the dead-letter queue", + ) + .context("email_dlq_size metric")?; + registry.register(Box::new(cache_hits.clone()))?; registry.register(Box::new(cache_misses.clone()))?; registry.register(Box::new(invalidations.clone()))?; @@ -77,6 +84,7 @@ impl Metrics { registry.register(Box::new(rpc_errors.clone()))?; registry.register(Box::new(rpc_fallbacks.clone()))?; registry.register(Box::new(db_timeouts.clone()))?; + registry.register(Box::new(email_dlq_size.clone()))?; Ok(Self { registry, @@ -87,6 +95,7 @@ impl Metrics { rpc_errors, rpc_fallbacks, db_timeouts, + email_dlq_size, }) } @@ -126,6 +135,10 @@ impl Metrics { self.db_timeouts.with_label_values(&[operation]).inc(); } + pub fn set_dlq_size(&self, n: i64) { + self.email_dlq_size.set(n); + } + pub fn observe_tx_eviction(&self, count: u64) { if count > 0 { self.invalidations diff --git a/services/api/src/migrations.rs b/services/api/src/migrations.rs index 672563f9..3fee3120 100644 --- a/services/api/src/migrations.rs +++ b/services/api/src/migrations.rs @@ -108,9 +108,49 @@ impl<'a> MigrationRunner<'a> { /// Ensure the tracking table exists, then apply every pending migration. /// Already-applied migrations are skipped. Returns the number of newly /// applied migrations. + /// + /// Uses a PostgreSQL session-level advisory lock to serialize concurrent + /// invocations (e.g. multiple instances starting simultaneously). If another + /// instance already holds the lock, this call aborts with an error so the + /// caller can surface it and halt startup cleanly. pub async fn run(&self) -> anyhow::Result { self.ensure_tracking_table().await?; + // Stable lock key — chosen to be unique to this codebase. + const MIGRATION_LOCK_KEY: i64 = 0x7072_6564_6963_7471_u64 as i64; + + let mut lock_conn = self + .pool + .acquire() + .await + .context("acquire advisory lock connection")?; + + let locked: bool = sqlx::query_scalar("SELECT pg_try_advisory_lock($1)") + .bind(MIGRATION_LOCK_KEY) + .fetch_one(&mut *lock_conn) + .await + .context("acquire migration advisory lock")?; + + if !locked { + bail!( + "another instance holds the migration advisory lock — \ + aborting to prevent concurrent migration execution" + ); + } + + let result = self.run_inner().await; + + // Always release the lock, even on failure, before the connection + // returns to the pool (session-level locks survive pool reuse). + let _ = sqlx::query("SELECT pg_advisory_unlock($1)") + .bind(MIGRATION_LOCK_KEY) + .execute(&mut *lock_conn) + .await; + + result + } + + async fn run_inner(&self) -> anyhow::Result { let mut applied = 0usize; for migration in MIGRATIONS { diff --git a/services/api/src/security.rs b/services/api/src/security.rs index b9cf843b..9079eab5 100644 --- a/services/api/src/security.rs +++ b/services/api/src/security.rs @@ -468,7 +468,7 @@ pub async fn sendgrid_webhook_middleware( ) -> Result { let is_dev = std::env::var("ENVIRONMENT") .map(|e| e == "development") - .unwrap_or(true); // default to dev if not set + .unwrap_or(false); // default to non-dev so signature verification is enforced if config.secret.is_none() && !is_dev { return Err(StatusCode::UNAUTHORIZED);