@@ -19740,40 +19740,283 @@ endpoint = { kind = "unix", socket_path = "/run/agent-substrate/isolation.sock"
1974019740 }
1974119741
1974219742 #[tokio::test]
19743- async fn provisioning_reconcile_waits_for_a_local_workspace_writer() {
19743+ async fn provisioning_reconcile_skips_a_sandbox_behind_a_workspace_writer() {
19744+ use crate::gateway_metrics::MetricsCapture;
19745+ use crate::persistence::MutationLockKey;
19746+ let metrics = MetricsCapture::install();
1974419747 let runtime = test_runtime(ControlledDriver::new()).await;
19748+ // Keys are taken in ascending order. With the sandbox key first, the
19749+ // scan holds X(sandbox) while it waits for S(workspace).
19750+ let workspace_key = MutationLockKey::Workspace("default").advisory_key();
19751+ let sandbox_id = (0..1000)
19752+ .map(|i| format!("sb-ws-writer-{i}"))
19753+ .find(|id| MutationLockKey::Sandbox(id).advisory_key() < workspace_key)
19754+ .expect("a sandbox key that sorts before the workspace key");
19755+ let sandbox_key = MutationLockKey::Sandbox(&sandbox_id).advisory_key();
1974519756 runtime
1974619757 .store
1974719758 .put_message(&sandbox_record(
19748- "sb-ws-writer" ,
19759+ &sandbox_id ,
1974919760 "ws-writer",
1975019761 SandboxPhase::Provisioning,
1975119762 ))
1975219763 .await
1975319764 .unwrap();
19765+ let now = openshell_core::time::now_ms();
19766+ runtime.reconcile_provisioning_deadlines(now).await.unwrap();
19767+
1975419768 // Reconcile re-derives configuration from provider and profile
19755- // records, so it waits for their writers, which hold X(workspace).
19769+ // records, so it must not run under their writers, which hold
19770+ // X(workspace).
1975619771 let writer = runtime
1975719772 .mutation_guard(MutationScope::Workspace("default"))
1975819773 .await
1975919774 .unwrap();
19760- let mut reconcile = {
19775+ let reconcile = {
1976119776 let runtime = runtime.clone();
19762- let now = openshell_core::time::now_ms();
19763- tokio::spawn(async move { runtime.reconcile_provisioning_deadlines(now).await })
19777+ tokio::spawn(async move {
19778+ runtime
19779+ .reconcile_provisioning_deadlines(now + 300_000)
19780+ .await
19781+ })
1976419782 };
19765- assert!(
19766- tokio::time::timeout(Duration::from_millis(100), &mut reconcile)
19783+ tokio::time::timeout(Duration::from_secs(5), async {
19784+ while runtime.mutation_locks.key_references(sandbox_key) == 0
19785+ || runtime.mutation_locks.key_references(workspace_key) < 2
19786+ {
19787+ tokio::task::yield_now().await;
19788+ }
19789+ })
19790+ .await
19791+ .expect("the scan holds the sandbox key and waits for the workspace key");
19792+
19793+ // Start, stop, and delete take the sandbox key; they must not wait
19794+ // for the writer through the scan.
19795+ let lifecycle = {
19796+ let runtime = runtime.clone();
19797+ let sandbox_id = sandbox_id.clone();
19798+ tokio::spawn(async move { drop(runtime.lock_sandbox_local(&sandbox_id).await) })
19799+ };
19800+ tokio::time::timeout(Duration::from_secs(5), lifecycle)
19801+ .await
19802+ .expect("a lifecycle lock proceeds while the writer holds the workspace")
19803+ .unwrap();
19804+ tokio::time::timeout(Duration::from_secs(5), reconcile)
19805+ .await
19806+ .expect("the scan skips the busy sandbox instead of waiting for the writer")
19807+ .unwrap()
19808+ .unwrap();
19809+ assert_eq!(runtime.mutation_locks.key_references(sandbox_key), 0);
19810+ assert_eq!(
19811+ metrics.value("openshell_server_provisioning_deadline_skips_total{overdue=\"true\"}"),
19812+ Some(1),
19813+ "the sandbox was past its deadline when the scan skipped it"
19814+ );
19815+ let skipped = runtime
19816+ .store
19817+ .get_message::<Sandbox>(&sandbox_id)
19818+ .await
19819+ .unwrap()
19820+ .unwrap();
19821+ assert_eq!(skipped.phase(), i32::from(SandboxPhase::Provisioning));
19822+
19823+ drop(writer);
19824+ runtime
19825+ .reconcile_provisioning_deadlines(now + 300_000)
19826+ .await
19827+ .unwrap();
19828+ let expired = runtime
19829+ .store
19830+ .get_message::<Sandbox>(&sandbox_id)
19831+ .await
19832+ .unwrap()
19833+ .unwrap();
19834+ assert_eq!(expired.phase(), i32::from(SandboxPhase::Error));
19835+ }
19836+
19837+ #[tokio::test]
19838+ async fn provisioning_reconcile_continues_past_a_skipped_sandbox() {
19839+ let runtime = test_runtime(ControlledDriver::new()).await;
19840+ // The scan orders by insert time, then name: the busy sandbox is stored
19841+ // first and named first, so a skip that ended the pass would leave the
19842+ // other sandbox alone.
19843+ let busy = sandbox_record("sb-skip-busy", "a-busy", SandboxPhase::Provisioning);
19844+ let mut other = sandbox_record("sb-skip-other", "b-other", SandboxPhase::Provisioning);
19845+ other.metadata.as_mut().unwrap().workspace = "alpha".to_string();
19846+ runtime.store.put_message(&busy).await.unwrap();
19847+ runtime.store.put_message(&other).await.unwrap();
19848+ let now = openshell_core::time::now_ms();
19849+ runtime.reconcile_provisioning_deadlines(now).await.unwrap();
19850+
19851+ let writer = runtime
19852+ .mutation_guard(MutationScope::Workspace("default"))
19853+ .await
19854+ .unwrap();
19855+ // Spawned: the lock-order check forbids nesting under the writer in this task.
19856+ let reconcile = {
19857+ let runtime = runtime.clone();
19858+ tokio::spawn(async move {
19859+ runtime
19860+ .reconcile_provisioning_deadlines(now + 300_000)
19861+ .await
19862+ })
19863+ };
19864+ tokio::time::timeout(Duration::from_secs(5), reconcile)
19865+ .await
19866+ .expect("the scan skips the busy sandbox instead of waiting for the writer")
19867+ .unwrap()
19868+ .unwrap();
19869+ let skipped = runtime
19870+ .store
19871+ .get_message::<Sandbox>("sb-skip-busy")
19872+ .await
19873+ .unwrap()
19874+ .unwrap();
19875+ assert_eq!(skipped.phase(), i32::from(SandboxPhase::Provisioning));
19876+ let expired = runtime
19877+ .store
19878+ .get_message::<Sandbox>("sb-skip-other")
19879+ .await
19880+ .unwrap()
19881+ .unwrap();
19882+ assert_eq!(
19883+ expired.phase(),
19884+ i32::from(SandboxPhase::Error),
19885+ "a skipped sandbox must not end the pass"
19886+ );
19887+ drop(writer);
19888+ }
19889+
19890+ #[tokio::test]
19891+ async fn provisioning_reconcile_waits_once_per_pass_behind_a_busy_workspace() {
19892+ use super::provisioning_deadline::SCAN_LOCK_WAIT_BUDGET;
19893+ use crate::gateway_metrics::MetricsCapture;
19894+ let metrics = MetricsCapture::install();
19895+ let runtime = test_runtime(ControlledDriver::new()).await;
19896+ let ids: Vec<String> = (0..6).map(|i| format!("sb-budget-{i}")).collect();
19897+ for (i, id) in ids.iter().enumerate() {
19898+ runtime
19899+ .store
19900+ .put_message(&sandbox_record(
19901+ id,
19902+ &format!("budget-{i}"),
19903+ SandboxPhase::Provisioning,
19904+ ))
1976719905 .await
19768- .is_err(),
19769- "provisioning reconcile must wait for the workspace writer"
19906+ .unwrap();
19907+ }
19908+ let now = openshell_core::time::now_ms();
19909+ runtime.reconcile_provisioning_deadlines(now).await.unwrap();
19910+
19911+ // Every sandbox needs S(workspace), which the writer holds.
19912+ let writer = runtime
19913+ .mutation_guard(MutationScope::Workspace("default"))
19914+ .await
19915+ .unwrap();
19916+ let started = tokio::time::Instant::now();
19917+ // Spawned: the lock-order check forbids nesting under the writer in this task.
19918+ let reconcile = {
19919+ let runtime = runtime.clone();
19920+ tokio::spawn(async move { runtime.reconcile_provisioning_deadlines(now + 1_000).await })
19921+ };
19922+ tokio::time::timeout(Duration::from_secs(10), reconcile)
19923+ .await
19924+ .expect("the scan finishes while the writer holds the workspace")
19925+ .unwrap()
19926+ .unwrap();
19927+ let waited = started.elapsed();
19928+ assert!(
19929+ waited < SCAN_LOCK_WAIT_BUDGET * 4,
19930+ "one wait per sandbox would take {:?}; the pass took {waited:?}",
19931+ SCAN_LOCK_WAIT_BUDGET * 6
1977019932 );
19933+ assert_eq!(
19934+ metrics.value("openshell_server_provisioning_deadline_skips_total{overdue=\"false\"}"),
19935+ Some(6)
19936+ );
19937+ for id in &ids {
19938+ let skipped = runtime
19939+ .store
19940+ .get_message::<Sandbox>(id)
19941+ .await
19942+ .unwrap()
19943+ .unwrap();
19944+ assert_eq!(skipped.phase(), i32::from(SandboxPhase::Provisioning));
19945+ }
1977119946 drop(writer);
19947+ }
19948+
19949+ #[tokio::test]
19950+ async fn provisioning_reconcile_releases_the_workspace_behind_a_busy_sandbox() {
19951+ use crate::persistence::MutationLockKey;
19952+ let runtime = test_runtime(ControlledDriver::new()).await;
19953+ // With the workspace key first, the scan holds S(workspace) while it
19954+ // waits for X(sandbox), and a provider writer queues behind the scan.
19955+ let workspace_key = MutationLockKey::Workspace("default").advisory_key();
19956+ let sandbox_id = (0..1000)
19957+ .map(|i| format!("sb-busy-{i}"))
19958+ .find(|id| MutationLockKey::Sandbox(id).advisory_key() > workspace_key)
19959+ .expect("a sandbox key that sorts after the workspace key");
19960+ let sandbox_key = MutationLockKey::Sandbox(&sandbox_id).advisory_key();
19961+ runtime
19962+ .store
19963+ .put_message(&sandbox_record(
19964+ &sandbox_id,
19965+ "busy",
19966+ SandboxPhase::Provisioning,
19967+ ))
19968+ .await
19969+ .unwrap();
19970+ let now = openshell_core::time::now_ms();
19971+ runtime.reconcile_provisioning_deadlines(now).await.unwrap();
19972+
19973+ let lifecycle = runtime.lock_sandbox_local(&sandbox_id).await;
19974+ let reconcile = {
19975+ let runtime = runtime.clone();
19976+ tokio::spawn(async move {
19977+ runtime
19978+ .reconcile_provisioning_deadlines(now + 300_000)
19979+ .await
19980+ })
19981+ };
19982+ tokio::time::timeout(Duration::from_secs(5), async {
19983+ while runtime.mutation_locks.key_references(workspace_key) == 0
19984+ || runtime.mutation_locks.key_references(sandbox_key) < 2
19985+ {
19986+ tokio::task::yield_now().await;
19987+ }
19988+ })
19989+ .await
19990+ .expect("the scan holds the workspace key and waits for the sandbox key");
19991+
19992+ let writer = {
19993+ let runtime = runtime.clone();
19994+ tokio::spawn(async move {
19995+ drop(
19996+ runtime
19997+ .mutation_guard(MutationScope::Workspace("default"))
19998+ .await
19999+ .unwrap(),
20000+ );
20001+ })
20002+ };
20003+ tokio::time::timeout(Duration::from_secs(5), writer)
20004+ .await
20005+ .expect("a provider writer proceeds while the sandbox is busy")
20006+ .unwrap();
1977220007 tokio::time::timeout(Duration::from_secs(5), reconcile)
1977320008 .await
19774- .expect("provisioning reconcile proceeds once the writer releases")
20009+ .expect("the scan skips the busy sandbox instead of waiting for it")
20010+ .unwrap()
20011+ .unwrap();
20012+ drop(lifecycle);
20013+ let skipped = runtime
20014+ .store
20015+ .get_message::<Sandbox>(&sandbox_id)
20016+ .await
1977520017 .unwrap()
1977620018 .unwrap();
20019+ assert_eq!(skipped.phase(), i32::from(SandboxPhase::Provisioning));
1977720020 }
1977820021
1977920022 #[tokio::test]
0 commit comments