diff --git a/src/context/manager.rs b/src/context/manager.rs index 764f189a..6eb63260 100644 --- a/src/context/manager.rs +++ b/src/context/manager.rs @@ -46,11 +46,17 @@ impl ContextManager { description: impl Into, ) -> Result { // Hold write lock for the entire check-insert to prevent TOCTOU races - // where two concurrent calls both pass the active_count check. + // where two concurrent calls both pass the parallel_count check. let mut contexts = self.contexts.write().await; - let active_count = contexts.values().filter(|c| c.state.is_active()).count(); + // Only count jobs that consume execution slots (Pending, InProgress, Stuck). + // Completed and Submitted jobs are no longer actively executing and shouldn't + // block new job creation. + let parallel_count = contexts + .values() + .filter(|c| c.state.is_parallel_blocking()) + .count(); - if active_count >= self.max_jobs { + if parallel_count >= self.max_jobs { return Err(JobError::MaxJobsExceeded { max: self.max_jobs }); } @@ -965,4 +971,218 @@ mod tests { // And it's in the initial state (Pending), not modified by concurrent workers assert_eq!(returned_ctx.state, crate::context::JobState::Pending); // safety: test code } + + #[tokio::test] + async fn sequential_routines_unlimited_completed_not_counted() { + // TEST: Sequential (non-parallel) routines should NOT be limited by max_jobs. + // + // Completed/Submitted jobs should NOT count toward the parallel job limit, + // since they're no longer actively consuming execution resources. + // + // Scenario: Create 10 sequential routines, each completing before the next starts. + // Currently FAILS because Completed jobs still count as "active". + // After fix, should PASS because only Pending/InProgress/Stuck count. + + let manager = ContextManager::new(5); // max 5 truly parallel jobs + + // Try to create and complete 10 sequential routines + for i in 0..10 { + let result = manager + .create_job(format!("Sequential Routine {}", i), "one at a time") + .await; + + match result { + Ok(job_id) => { + // Simulate execution: Pending -> InProgress -> Completed + manager + .update_context(job_id, |ctx| { + ctx.transition_to(crate::context::JobState::InProgress, None) + }) + .await + .unwrap() + .unwrap(); + + manager + .update_context(job_id, |ctx| { + ctx.transition_to(crate::context::JobState::Completed, None) + }) + .await + .unwrap() + .unwrap(); + + println!("✓ Routine {} created and completed", i); + } + Err(JobError::MaxJobsExceeded { max }) => { + panic!( + "✗ Routine {} FAILED to create: MaxJobsExceeded (max={}).\n\ + This shows the bug: Completed jobs from routines 0-4 are still counting \ + toward the limit even though they're not running.\n\ + After the fix, this test should pass because Completed jobs won't count.", + i, max + ); + } + Err(e) => { + panic!("Unexpected error for routine {}: {:?}", i, e); + } + } + } + + // If we reach here, all 10 routines succeeded (bug is fixed) + assert_eq!(manager.all_jobs().await.len(), 10); + println!("✓ SUCCESS: All 10 sequential routines created despite max_jobs=5 limit"); + println!(" This is correct: Completed jobs don't count toward parallel limit"); + } + + #[tokio::test] + async fn parallel_jobs_limit_enforced_for_active_jobs() { + // TEST: Parallel (simultaneous) jobs ARE limited by max_jobs. + // + // Jobs in Pending/InProgress/Stuck states consume execution slots. + // The 6th truly-active job should fail because the limit is 5. + // + // This test verifies the limit DOES work correctly for parallel execution. + + let manager = ContextManager::new(5); // max 5 parallel jobs + + // Create 5 jobs and make them InProgress (simulating parallel execution) + let mut job_ids = Vec::new(); + for i in 0..5 { + let job_id = manager + .create_job(format!("Parallel Job {}", i), "running in parallel") + .await + .expect("First 5 jobs should create successfully"); + job_ids.push(job_id); + + // Transition to InProgress (simulating active execution) + manager + .update_context(job_id, |ctx| { + ctx.transition_to(crate::context::JobState::InProgress, None) + }) + .await + .unwrap() + .unwrap(); + } + + // Verify all 5 jobs are InProgress + for job_id in &job_ids { + let ctx = manager.get_context(*job_id).await.unwrap(); + assert_eq!( + ctx.state, + crate::context::JobState::InProgress, + "All jobs should be InProgress" + ); + } + + // Check active count - should be 5 (all InProgress) + let active_count = manager.active_count().await; + assert_eq!( + active_count, 5, + "Active count should be 5 (all InProgress jobs count)" + ); + + // Try to create a 6th job - should FAIL because limit is reached + let result = manager.create_job("Parallel Job 6", "sixth job").await; + + match result { + Err(JobError::MaxJobsExceeded { max: 5 }) => { + println!("✓ SUCCESS: Parallel job limit correctly enforced at 5 active jobs"); + println!("✓ 6th InProgress job correctly blocked when 5 are already running"); + } + Ok(_) => { + panic!( + "FAILED: 6th parallel job should have been blocked \ + but was created. Limit enforcement is broken." + ); + } + Err(e) => { + panic!( + "UNEXPECTED ERROR: Expected MaxJobsExceeded but got: {:?}", + e + ); + } + } + } + + #[tokio::test] + async fn completed_jobs_should_free_slots_after_fix() { + // TEST: After the fix, Completed jobs should NOT count toward the limit. + // + // This test demonstrates that when a job transitions from InProgress -> Completed, + // it should free up a slot in the parallel execution limit. + // + // Currently FAILS (bug not fixed), proving Completed jobs incorrectly stay in the limit. + // After fix, this will PASS (Completed jobs freed their slot). + + let manager = ContextManager::new(5); // max 5 parallel jobs + + // Create 5 InProgress jobs (fill the limit) + let mut job_ids = Vec::new(); + for i in 0..5 { + let job_id = manager + .create_job(format!("Job {}", i), "parallel") + .await + .unwrap(); + job_ids.push(job_id); + + manager + .update_context(job_id, |ctx| { + ctx.transition_to(crate::context::JobState::InProgress, None) + }) + .await + .unwrap() + .unwrap(); + } + + // Verify limit is hit + let result = manager.create_job("Job 5", "should fail").await; + assert!( + matches!(result, Err(JobError::MaxJobsExceeded { max: 5 })), + "Limit should be hit with 5 InProgress jobs" + ); + println!("✓ Limit enforced: 5 InProgress jobs block 6th creation"); + + // Now transition job 0 from InProgress -> Completed + manager + .update_context(job_ids[0], |ctx| { + ctx.transition_to(crate::context::JobState::Completed, None) + }) + .await + .unwrap() + .unwrap(); + + println!("✓ Job 0 transitioned: InProgress -> Completed"); + + // Try to create a 6th job - this will FAIL until the bug is fixed + let result = manager + .create_job("Job 5 (retry)", "after 1 Completed") + .await; + + match result { + Ok(job_6) => { + println!("✓ SUCCESS: 6th job created after job 0 completed"); + println!("✓ This proves Completed jobs don't count toward the limit (BUG FIXED)"); + + // Verify we can transition it to InProgress + manager + .update_context(job_6, |ctx| { + ctx.transition_to(crate::context::JobState::InProgress, None) + }) + .await + .unwrap() + .unwrap(); + println!("✓ 6th job now InProgress: 4 remaining + 1 new = 5 limit reached"); + } + Err(JobError::MaxJobsExceeded { max: 5 }) => { + panic!( + "✗ BUG NOT FIXED: 6th job creation still blocked after freeing slot.\n\ + State: 1 Completed (job 0) + 4 InProgress (jobs 1-4) = 5 active\n\ + BUG: Completed job 0 still counts toward limit\n\ + EXPECTED: Only 4 InProgress count, 1 slot free" + ); + } + Err(e) => { + panic!("Unexpected error: {:?}", e); + } + } + } } diff --git a/src/context/state.rs b/src/context/state.rs index 2402fd66..f5307947 100644 --- a/src/context/state.rs +++ b/src/context/state.rs @@ -81,6 +81,15 @@ impl JobState { pub fn is_active(&self) -> bool { !self.is_terminal() } + + /// Check if this job consumes a parallel execution slot. + /// + /// Only jobs in Pending, InProgress, or Stuck states consume execution resources + /// and should count toward the parallel job limit. Completed and Submitted jobs + /// are in the state machine but are no longer actively executing. + pub fn is_parallel_blocking(&self) -> bool { + matches!(self, Self::Pending | Self::InProgress | Self::Stuck) + } } impl std::fmt::Display for JobState {