Add metrics and logs from the bughunt

Cargo.toml

@@ -27,6 +27,7 @@ waymark-ir-conversions = { path = "crates/lib/ir-conversions" }
 waymark-ir-format = { path = "crates/lib/ir-format" }
 waymark-ir-parser = { path = "crates/lib/ir-parser" }
 waymark-message-conversions = { path = "crates/lib/message-conversions" }
+waymark-metrics-util = { path = "crates/lib/metrics-util" }
 waymark-nonzero-duration = { path = "crates/lib/nonzero-duration" }
 waymark-observability = { path = "crates/lib/observability" }
 waymark-observability-macros = { path = "crates/lib/observability-macros" }

crates/bin/start-workers/Cargo.toml

@@ -25,6 +25,7 @@ waymark-worker-status-reporter = { workspace = true }
 
 anyhow = { workspace = true }
 envfury = { workspace = true }
+metrics = { workspace = true }
 prost = { workspace = true }
 sqlx = { workspace = true }
 thiserror = { workspace = true }

crates/bin/start-workers/src/main.rs

@@ -65,7 +65,10 @@ async fn main() -> Result<()> {
     // Load configuration and announce startup.
     let config = WorkerConfig::from_env()?;
 
-    tracing::debug!(target: "raw-config", ?config, "raw config");
+    // Prepare a new Lock ID to use in the runloop.
+    let lock_uuid = LockId::new_uuid_v4();
+
+    tracing::debug!(target: "raw-config", ?config, %lock_uuid, "raw config");
 
     info!(
         worker_count = config.worker_count,
@@ -81,9 +84,44 @@ async fn main() -> Result<()> {
         garbage_collector_batch_size = config.garbage_collector.batch_size,
         garbage_collector_retention_secs = config.garbage_collector.retention.as_secs(),
         max_action_lifecycle = ?config.max_action_lifecycle,
+        %lock_uuid,
         "starting worker infrastructure"
     );
 
+    metrics::gauge!(
+        "waymark_start_workers_up",
+
+        "worker_count" => config.worker_count.to_string(),
+        "concurrent_per_worker" => config.concurrent_per_worker.to_string(),
+        "user_modules" => format!("{:?}", config.user_modules),
+
+        "lock_ttl_seconds" => config.lock_ttl.as_secs_f64().to_string(),
+        "lock_heartbeat_seconds" => config.lock_heartbeat.as_secs_f64().to_string(),
+
+        "evict_sleep_threshold_seconds" => config.evict_sleep_threshold.as_secs_f64().to_string(),
+
+        "expired_lock_reclaimer_interval_seconds" => config.expired_lock_reclaimer_interval.as_secs_f64().to_string(),
+        "expired_lock_reclaimer_interval_seconds" => config.expired_lock_reclaimer_interval.as_secs_f64().to_string(),
+        "expired_lock_reclaimer_batch_size" => config.expired_lock_reclaimer_batch_size.to_string(),
+
+        "garbage_collector_interval_seconds" => config.garbage_collector.interval.as_secs_f64().to_string(),
+        "garbage_collector_batch_size" => config.garbage_collector.batch_size.to_string(),
+        "garbage_collector_retention_seconds" => config.garbage_collector.retention.as_secs_f64().to_string(),
+
+        "max_action_lifecycle" => config.max_action_lifecycle.map(|val| val.to_string()).unwrap_or("no".into()),
+
+        "executor_shards" => config.executor_shards.to_string(),
+        "poll_interval_seconds" => config.poll_interval.map(|val| val.as_secs_f64().to_string()).unwrap_or("no".into()),
+        "max_concurrent_instances" => config.max_concurrent_instances.to_string(),
+        "instance_done_batch_size" => config.instance_done_batch_size.map(|val| val.to_string()).unwrap_or("no".into()),
+        "persistence_interval_seconds" => config.persistence_interval.map(|val| val.as_secs_f64().to_string()).unwrap_or("no".into()),
+
+        "profile_interval_seconds" => config.profile_interval.as_secs_f64().to_string(),
+
+        "lock_uuid" => lock_uuid.to_string(),
+    )
+    .set(1);
+
     // Wire shutdown coordination.
     let shutdown_token = tokio_util::sync::CancellationToken::new();
 
@@ -187,7 +225,6 @@ async fn main() -> Result<()> {
     });
 
     // Run the runloop.
-    let lock_uuid = LockId::new_uuid_v4();
     let runloop = waymark_runloop::RunLoop::new_with_shutdown(
         remote_pool.clone(),
         backend.clone(),

crates/lib/backend-postgres/Cargo.toml

@@ -13,6 +13,7 @@ waymark-dag-builder = { workspace = true }
 waymark-garbage-collector-backend = { workspace = true }
 waymark-ids = { workspace = true }
 waymark-ir-conversions = { workspace = true }
+waymark-metrics-util = { workspace = true }
 waymark-observability = { workspace = true }
 waymark-proto = { workspace = true }
 waymark-runner = { workspace = true }

crates/lib/backend-postgres/src/core.rs

@@ -586,6 +586,8 @@ impl PostgresBackend {
         runner_builder.push_values(
             payloads.iter(),
             |mut b, (instance_id, _scheduled_at, _lock_expires_at, payload)| {
+                metrics::histogram!("waymark_postgres_save_graphs_runner_instances_payload_size")
+                    .record(payload.len() as f64);
                 b.push_bind(*instance_id).push_bind(payload.as_slice());
             },
         );
@@ -779,6 +781,9 @@ impl PostgresBackend {
             return Err(PollQueuedInstancesError::EmptyRows);
         };
 
+        metrics::counter!("waymark_backend_postgres_query_poll_instances_rows_total")
+            .increment(rows.len().get() as _);
+
         let claimed_instance_ids: Vec<Uuid> =
             rows.iter().map(|row| row.get("instance_id")).collect();
         sqlx::query("UPDATE runner_instances SET current_status = $2 WHERE instance_id = ANY($1)")

crates/lib/backend-postgres/src/lib.rs

@@ -14,6 +14,7 @@ use std::sync::{Arc, Mutex};
 
 use sqlx::PgPool;
 use waymark_backends_core::{BackendError, BackendResult};
+use waymark_metrics_util::Val as MetricsVal;
 use waymark_observability::obs;
 use waymark_secret_string::SecretStr;
 use waymark_timed_future::TimedFutureExt as _;
@@ -95,19 +96,22 @@ impl PostgresBackend {
             .clone()
     }
 
-    pub(crate) fn count_query(counts: &Arc<Mutex<HashMap<String, usize>>>, label: &str) {
+    pub(crate) fn count_query(counts: &Arc<Mutex<HashMap<String, usize>>>, label: &'static str) {
+        metrics::counter!("waymark_postgres_queries_total", "label" => label).increment(1);
         let mut guard = counts.lock().expect("query counts poisoned");
         *guard.entry(label.to_string()).or_insert(0) += 1;
     }
 
     pub(crate) fn count_batch_size(
         counts: &Arc<Mutex<HashMap<String, HashMap<usize, usize>>>>,
-        label: &str,
+        label: &'static str,
         size: usize,
     ) {
         if size == 0 {
             return;
         }
+        metrics::histogram!("waymark_postgres_queries_batch_size", "label" => label)
+            .record(MetricsVal(size));
         let mut guard = counts.lock().expect("batch size counts poisoned");
         let entry = guard.entry(label.to_string()).or_default();
         *entry.entry(size).or_insert(0) += 1;

crates/lib/metrics-util/Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "waymark-metrics-util"
+edition = "2024"
+version.workspace = true
+publish.workspace = true
+
+[dependencies]
+metrics = { workspace = true }

crates/lib/metrics-util/src/lib.rs

@@ -0,0 +1,9 @@
+//! Utilities for [`metrics`] crate.
+
+pub struct Val<T>(pub T);
+
+impl metrics::IntoF64 for Val<usize> {
+    fn into_f64(self) -> f64 {
+        self.0 as _
+    }
+}

crates/lib/runloop/Cargo.toml

@@ -10,6 +10,7 @@ waymark-core-backend = { workspace = true }
 waymark-dag = { workspace = true }
 waymark-dag-builder = { workspace = true }
 waymark-ids = { workspace = true }
+waymark-metrics-util = { workspace = true }
 waymark-nonzero-duration = { workspace = true }
 waymark-observability = { workspace = true }
 waymark-proto = { workspace = true }
@@ -23,6 +24,7 @@ waymark-worker-core = { workspace = true }
 waymark-workflow-registry-backend = { workspace = true }
 
 chrono = { workspace = true, default-features = false }
+metrics = { workspace = true }
 nonempty-collections = { workspace = true }
 prost = { workspace = true }
 serde_json = { workspace = true }

crates/lib/runloop/src/available_instance_slots.rs

@@ -98,6 +98,11 @@ impl Tracker {
             .send(available_slots)
             .map_err(|_| UpdateError::NoReaders)
     }
+
+    /// Peek at the current value of the available slots.
+    pub fn peek_available(&self) -> usize {
+        *self.tx.borrow()
+    }
 }
 
 impl Reader {