fix: bounded event bus, JSON tracing, graceful shutdown, embedded migrations

core/Cargo.toml

@@ -28,13 +28,13 @@ config = "0.14"
 axum = { version = "0.7", features = ["multipart", "ws"] }
 tempfile = "3"
 tracing = "0.1"
+tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
 opentelemetry = "0.27"
 opentelemetry_sdk = "0.27"
 tracing-opentelemetry = "0.28"
-tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 prometheus = "0.14"
 tower = "0.4"
-tower-http = { version = "0.5", features = ["cors", "trace"] }
+tower-http = { version = "0.5", features = ["cors", "trace", "request-id", "set-header"] }
 utoipa = { version = "4", features = ["axum_extras"] }
 utoipa-swagger-ui = { version = "7", features = ["axum", "vendored"] }
 base64 = "0.22"

core/src/main.rs

@@ -55,6 +55,7 @@ use crate::rpc_provider::{ProviderRegistry, RegistryConfig, RegistrySnapshot, Rp
 use crate::simulation::{SimulationEngine, SimulationMode, SimulationResult};
 use crate::ws::SimulationBus;
 use tower_http::cors::{Any, CorsLayer};
+use tower_http::request_id::{MakeRequestUuid, PropagateRequestIdLayer, SetRequestIdLayer};
 use tower_http::trace::TraceLayer;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter};
 use utoipa::{OpenApi, ToSchema};
@@ -140,6 +141,16 @@ struct AppConfig {
     /// L2 treats it as stale. Default 100 ≈ 8 minutes at 5 s/ledger.
     #[serde(default = "default_max_ledger_age")]
     max_ledger_age: u32,
+    /// Broadcast channel capacity for the WebSocket event bus (issue #565).
+    /// Controls the per-subscriber in-flight event buffer; slow consumers
+    /// that fall behind receive `RecvError::Lagged` (backpressure via drop).
+    /// Clamped to [16, 65536]. Default 256.
+    #[serde(default = "default_event_bus_capacity")]
+    event_bus_capacity: usize,
+    /// Emit structured JSON log lines instead of the default human-readable
+    /// format (issue #572). Set `LOG_FORMAT=json` to enable.
+    #[serde(default)]
+    log_format_json: bool,
 }
 
 fn default_health_check_interval() -> u64 {
@@ -196,6 +207,10 @@ fn default_max_ledger_age() -> u32 {
     100
 }
 
+fn default_event_bus_capacity() -> usize {
+    256
+}
+
 fn load_config() -> Result<AppConfig, ConfigError> {
     dotenvy::dotenv().ok();
 
@@ -223,6 +238,8 @@ fn load_config() -> Result<AppConfig, ConfigError> {
         .set_default("emergency_verification_paused", false)?
         .set_default("disk_cache_path", "")?
         .set_default("max_ledger_age", 100)?
+        .set_default("event_bus_capacity", 256)?
+        .set_default("log_format_json", false)?
         .build()?;
 
     settings.try_deserialize()
@@ -1715,10 +1732,20 @@ async fn main() {
         env::set_var("RUST_LOG", "info");
     }
 
-    tracing_subscriber::registry()
-        .with(EnvFilter::from_default_env())
-        .with(tracing_subscriber::fmt::layer())
-        .init();
+    // ── Tracing init (#572: JSON format + x-request-id correlation) ────
+    let log_json = env::var("LOG_FORMAT").map(|v| v.to_lowercase() == "json").unwrap_or(false);
+    let filter = EnvFilter::from_default_env();
+    if log_json {
+        tracing_subscriber::registry()
+            .with(filter)
+            .with(tracing_subscriber::fmt::layer().json())
+            .init();
+    } else {
+        tracing_subscriber::registry()
+            .with(filter)
+            .with(tracing_subscriber::fmt::layer())
+            .init();
+    }
 
     tracing::info!("SoroScope Starting...");
 
@@ -2067,8 +2094,8 @@ async fn main() {
     let job_queue = JobQueue::new(database_url, &config.redis_url, job_queue_config.clone())
         .await
         .expect("Failed to initialize job queue");
-    // ── WebSocket event bus ─────────────────────────────────────────────
-    let simulation_bus = SimulationBus::new();
+    // ── WebSocket event bus (#565: configurable bounded channel) ───────
+    let simulation_bus = SimulationBus::with_capacity(config.event_bus_capacity);
 
     let job_worker = JobWorker::new(
         job_queue.clone(),
@@ -2215,6 +2242,12 @@ async fn main() {
         .layer(Extension(auth_state))
         .layer(cors)
         .layer(TraceLayer::new_for_http())
+        // ── x-request-id (#572) ───────────────────────────────────────
+        // Assigns a UUID to every inbound request under the `x-request-id`
+        // header and propagates it to outbound responses so clients can
+        // correlate log lines with specific requests.
+        .layer(PropagateRequestIdLayer::x_request_id())
+        .layer(SetRequestIdLayer::x_request_id(MakeRequestUuid))
         .with_state(app_state); // ← thread AppState through all handlers
 
     let bind_addr = format!("0.0.0.0:{}", config.server_port);
@@ -2231,9 +2264,37 @@ async fn main() {
         listener.local_addr().unwrap()
     );
 
+    // ── Graceful shutdown (#573: SIGTERM / SIGINT) ────────────────────
     axum::serve(listener, app)
+        .with_graceful_shutdown(shutdown_signal())
         .await
         .expect("Server failed to start");
+
+    tracing::info!("Server shut down gracefully.");
+}
+
+/// Waits for SIGTERM (Unix) or Ctrl-C (all platforms) and resolves once either
+/// signal is received, allowing axum to finish in-flight requests before exit.
+async fn shutdown_signal() {
+    #[cfg(unix)]
+    let sigterm = async {
+        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+            .expect("failed to install SIGTERM handler")
+            .recv()
+            .await;
+    };
+
+    #[cfg(not(unix))]
+    let sigterm = std::future::pending::<()>();
+
+    tokio::select! {
+        _ = tokio::signal::ctrl_c() => {
+            tracing::info!("Received SIGINT (Ctrl-C), shutting down…");
+        }
+        _ = sigterm => {
+            tracing::info!("Received SIGTERM, shutting down…");
+        }
+    }
 }
 
 // ─────────────────────────────────────────────────────────────────────────────

core/src/ws.rs

@@ -50,10 +50,16 @@ use crate::trace_propagation::TracedMessage;
 
 // ── Channel capacity ─────────────────────────────────────────────────────────
 
-/// Number of events that can be buffered per broadcast channel slot before
-/// slow consumers are forced to drop events via `RecvError::Lagged`.
+/// Default number of events buffered per broadcast channel slot before slow
+/// consumers are forced to drop events via `RecvError::Lagged`.
 const BUS_CAPACITY: usize = 256;
 
+/// Minimum allowed channel capacity (prevents degenerate single-slot configs).
+const BUS_CAPACITY_MIN: usize = 16;
+
+/// Maximum allowed channel capacity (guards against OOM from untrusted config).
+const BUS_CAPACITY_MAX: usize = 65_536;
+
 // ── Event types ──────────────────────────────────────────────────────────────
 
 /// Progress update emitted at each stage of job execution.
@@ -166,9 +172,20 @@ pub struct SimulationBus {
 }
 
 impl SimulationBus {
-    /// Create a new bus with the default channel capacity.
+    /// Create a new bus with the default channel capacity (`BUS_CAPACITY`).
     pub fn new() -> Arc<Self> {
-        let (sender, _) = broadcast::channel(BUS_CAPACITY);
+        Self::with_capacity(BUS_CAPACITY)
+    }
+
+    /// Create a new bus with an explicit channel capacity.
+    ///
+    /// `capacity` is clamped to `[BUS_CAPACITY_MIN, BUS_CAPACITY_MAX]`.
+    /// Slow subscribers that fall more than `capacity` events behind receive
+    /// [`RecvError::Lagged`] on the next receive call — this is the intended
+    /// backpressure mechanism (drop the stale event, catch up on the next tick).
+    pub fn with_capacity(capacity: usize) -> Arc<Self> {
+        let clamped = capacity.clamp(BUS_CAPACITY_MIN, BUS_CAPACITY_MAX);
+        let (sender, _) = broadcast::channel(clamped);
         Arc::new(Self { sender })
     }