VCR WIP [do not merge]

Cargo.toml

@@ -26,6 +26,7 @@ waymark-ids = { path = "crates/lib/ids" }
 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-jsonlines = { path = "crates/lib/jsonlines" }
 waymark-message-conversions = { path = "crates/lib/message-conversions" }
 waymark-metrics-util = { path = "crates/lib/metrics-util" }
 waymark-nonzero-duration = { path = "crates/lib/nonzero-duration" }
@@ -57,6 +58,13 @@ waymark-timed-future = { path = "crates/lib/timed-future" }
 waymark-tokio-metrics-bringup = { path = "crates/lib/tokio-metrics-bringup" }
 waymark-utils-futures = { path = "crates/lib/utils-futures" }
 waymark-utils-tokio-channel = { path = "crates/lib/utils-tokio-channel" }
+waymark-vcr-core = { path = "crates/lib/vcr-core" }
+waymark-vcr-file = { path = "crates/lib/vcr-file" }
+waymark-vcr-playback-worker-pool-core = { path = "crates/lib/vcr-playback-worker-pool-core" }
+waymark-vcr-recorder = { path = "crates/lib/vcr-recorder" }
+waymark-vcr-recorder-backend = { path = "crates/lib/vcr-recorder-backend" }
+waymark-vcr-recorder-bringup = { path = "crates/lib/vcr-recorder-bringup" }
+waymark-vcr-recorder-worker-pool = { path = "crates/lib/vcr-recorder-worker-pool" }
 waymark-webapp-backend = { path = "crates/lib/webapp-backend" }
 waymark-webapp-bringup = { path = "crates/lib/webapp-bringup" }
 waymark-webapp-config = { path = "crates/lib/webapp-config" }
@@ -71,13 +79,15 @@ waymark-worker-status-core = { path = "crates/lib/worker-status-core" }
 waymark-worker-status-reporter = { path = "crates/lib/worker-status-reporter" }
 waymark-workflow-registry-backend = { path = "crates/lib/workflow-registry-backend" }
 
+
 anyhow = "1"
 axum = "0.8"
 chrono = { version = "0.4", default-features = false }
 clap = "4.5"
 color-eyre = "0.6"
 console-subscriber = "0.5"
 cron = "0.12"
+dashmap = "6"
 either = "1"
 envfury = "0.2"
 function_name = "0.3"

crates/bin/start-workers/Cargo.toml

@@ -19,6 +19,7 @@ waymark-runloop = { workspace = true }
 waymark-scheduler-loop = { workspace = true }
 waymark-scheduler-loop-core = { workspace = true }
 waymark-tokio-metrics-bringup = { workspace = true }
+waymark-vcr-recorder-bringup = { workspace = true }
 waymark-webapp-bringup = { workspace = true }
 waymark-worker-remote = { workspace = true }
 waymark-worker-status-reporter = { workspace = true }

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

@@ -224,10 +224,17 @@ async fn main() -> Result<()> {
         }
     });
 
+    let (recorder_task, recorder_pool_handle, recorder_backend_handle) =
+        waymark_vcr_recorder_bringup::core("instances.jsonl", 1024).await?;
+
     // Run the runloop.
+    let runloop_remote_pool =
+        waymark_vcr_recorder_bringup::pool(remote_pool.clone(), Some(recorder_pool_handle));
+    let runloop_backend =
+        waymark_vcr_recorder_bringup::backend(backend.clone(), Some(recorder_backend_handle));
     let runloop = waymark_runloop::RunLoop::new_with_shutdown(
-        remote_pool.clone(),
-        backend.clone(),
+        runloop_remote_pool,
+        runloop_backend,
         RunLoopConfig {
             max_concurrent_instances: config.max_concurrent_instances,
             executor_shards: config.executor_shards,
@@ -261,6 +268,7 @@ async fn main() -> Result<()> {
     let _ = tokio::time::timeout(Duration::from_secs(5), garbage_collector_handle).await;
     let _ = tokio::time::timeout(Duration::from_secs(2), status_reporter_handle).await;
     let _ = tokio::time::timeout(Duration::from_secs(2), expired_lock_reclaimer_handle).await;
+    let _ = tokio::time::timeout(Duration::from_secs(2), recorder_task).await;
 
     if let Err(err) = remote_pool.shutdown().await {
         warn!(error = %err, "worker pool shutdown failed");

crates/lib/core-backend/Cargo.toml

@@ -4,13 +4,17 @@ version.workspace = true
 publish.workspace = true
 edition = "2024"
 
+[features]
+default = ["either"]
+
 [dependencies]
 waymark-backends-core = { workspace = true }
 waymark-ids = { workspace = true }
 waymark-runner-executor-core = { workspace = true }
 waymark-runner-state = { workspace = true }
 
 chrono = { workspace = true }
+either = { workspace = true, optional = true }
 nonempty-collections = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 

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

@@ -0,0 +1,101 @@
+use either::Either;
+
+use crate::CoreBackend;
+
+impl<Left, Right> CoreBackend for Either<Left, Right>
+where
+    Left: CoreBackend,
+    Right: CoreBackend<PollQueuedInstancesError = Left::PollQueuedInstancesError>,
+{
+    fn save_graphs<'a>(
+        &'a self,
+        claim: crate::LockClaim,
+        graphs: &'a [crate::GraphUpdate],
+    ) -> impl Future<Output = waymark_backends_core::BackendResult<Vec<crate::InstanceLockStatus>>>
+    + Send
+    + 'a {
+        match self {
+            Either::Left(inner) => Either::Left(inner.save_graphs(claim, graphs)),
+            Either::Right(inner) => Either::Right(inner.save_graphs(claim, graphs)),
+        }
+    }
+
+    fn save_actions_done<'a>(
+        &'a self,
+        actions: &'a [crate::ActionDone],
+    ) -> impl Future<Output = waymark_backends_core::BackendResult<()>> + Send + 'a {
+        match self {
+            Either::Left(inner) => Either::Left(inner.save_actions_done(actions)),
+            Either::Right(inner) => Either::Right(inner.save_actions_done(actions)),
+        }
+    }
+
+    type PollQueuedInstancesError = Left::PollQueuedInstancesError;
+
+    fn poll_queued_instances(
+        &self,
+        size: std::num::NonZeroUsize,
+        claim: crate::LockClaim,
+    ) -> impl Future<
+        Output = Result<
+            nonempty_collections::NEVec<crate::QueuedInstance>,
+            Self::PollQueuedInstancesError,
+        >,
+    > + Send
+    + '_ {
+        match self {
+            Either::Left(inner) => Either::Left(inner.poll_queued_instances(size, claim)),
+            Either::Right(inner) => Either::Right(inner.poll_queued_instances(size, claim)),
+        }
+    }
+
+    fn refresh_instance_locks<'a>(
+        &'a self,
+        claim: crate::LockClaim,
+        instance_ids: &'a [waymark_ids::InstanceId],
+    ) -> impl Future<Output = waymark_backends_core::BackendResult<Vec<crate::InstanceLockStatus>>>
+    + Send
+    + 'a {
+        match self {
+            Either::Left(inner) => Either::Left(inner.refresh_instance_locks(claim, instance_ids)),
+            Either::Right(inner) => {
+                Either::Right(inner.refresh_instance_locks(claim, instance_ids))
+            }
+        }
+    }
+
+    fn release_instance_locks<'a>(
+        &'a self,
+        lock_uuid: waymark_ids::LockId,
+        instance_ids: &'a [waymark_ids::InstanceId],
+    ) -> impl Future<Output = waymark_backends_core::BackendResult<()>> + Send + 'a {
+        match self {
+            Either::Left(inner) => {
+                Either::Left(inner.release_instance_locks(lock_uuid, instance_ids))
+            }
+            Either::Right(inner) => {
+                Either::Right(inner.release_instance_locks(lock_uuid, instance_ids))
+            }
+        }
+    }
+
+    fn save_instances_done<'a>(
+        &'a self,
+        instances: &'a [crate::InstanceDone],
+    ) -> impl Future<Output = waymark_backends_core::BackendResult<()>> + Send + 'a {
+        match self {
+            Either::Left(inner) => Either::Left(inner.save_instances_done(instances)),
+            Either::Right(inner) => Either::Right(inner.save_instances_done(instances)),
+        }
+    }
+
+    fn queue_instances<'a>(
+        &'a self,
+        instances: &'a [crate::QueuedInstance],
+    ) -> impl Future<Output = waymark_backends_core::BackendResult<()>> + Send + 'a {
+        match self {
+            Either::Left(inner) => Either::Left(inner.queue_instances(instances)),
+            Either::Right(inner) => Either::Right(inner.queue_instances(instances)),
+        }
+    }
+}

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

@@ -1,6 +1,10 @@
 //! Core backend traits for waymark.
 
+#[cfg(feature = "either")]
+mod either;
+
 mod data;
+
 pub mod poll_queued_instances;
 
 use nonempty_collections::NEVec;

crates/lib/vcr-core/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "waymark-vcr-core"
+edition = "2024"
+version.workspace = true
+publish.workspace = true
+
+[dependencies]
+waymark-ids = { workspace = true }
+waymark-worker-core = { workspace = true }
+
+serde = { workspace = true, features = ["derive"] }