Make debug command enqueue atomic with the engine's resume

.changeset/debug-command-strand.md

@@ -0,0 +1,5 @@
+---
+"@kurrent/gaffer": patch
+---
+
+Racing debug commands can no longer wedge a debug session. When two resume verbs raced on a paused projection (a double-clicked continue, or the MCP auto-step racing a user step), the loser's command could be queued just as the engine resumed. Its caller then blocked forever, and the stale command silently resumed the next breakpoint instead. The runtime now makes the enqueue atomic with the resume, fails commands that lost the race with an error instead of stranding them, and never carries queued commands across a pause.

cli/internal/engine/runner_debug.go

@@ -68,11 +68,12 @@ func (r *Runner) doStep(fn func() error) error {
 	if r.debug == nil {
 		return nil
 	}
-	// Flip paused optimistically, before issuing the command. The only error
-	// fn can return is the runtime's "not paused" - which means the engine
-	// wasn't paused anyway, so false is the correct resulting state. Leaving it
-	// false on error also lets the caller's next Paused() guard reject cleanly
-	// rather than re-entering here and failing again.
+	// Flip paused optimistically, before issuing the command. The only errors
+	// fn can return are the runtime's "not paused" and its lost-the-race-
+	// with-a-resume refusal (UI-1822) - both mean the engine ended up
+	// running, so false is the correct resulting state. Leaving it false on
+	// error also lets the caller's next Paused() guard reject cleanly rather
+	// than re-entering here and failing again.
 	r.mu.Lock()
 	if r.closed {
 		// Teardown has begun: the session may already be freed. Refuse
@@ -177,12 +178,10 @@ func (r *Runner) drain() {
 // Front-ends should still stop their source loop first as a matter of
 // hygiene, but a straggling feed no longer races the teardown.
 //
-// The wait trusts every registered op to finish. The runtime's debug-command
-// queue can strand a command (an unsynchronized paused check-then-add racing
-// the command loop's exit leaves the command's Done unsignalled), and a
-// stranded verb now blocks Destroy where it previously leaked a goroutine.
-// The window needs concurrent debug verbs on a paused engine; the fix is
-// runtime-side, in the command loop.
+// The wait trusts every registered op to finish, which the runtime
+// guarantees: a debug verb that loses the race with a resume is failed with
+// an error rather than stranded (UI-1822), so no verb can hold an op - and
+// this wait - open indefinitely.
 //
 // The history store is NOT closed here: the Runner only borrows it, and in
 // the dev debug loop one store serves every restart iteration's runner.

runtime/Gaffer.Runtime.Tests/DebugTests.cs

@@ -805,4 +805,118 @@ public void Timeout_does_not_fire_during_pause() {
 		Assert.Null(feedEx);
 		Assert.NotNull(result);
 	}
+
+	[Fact]
+	public void Racing_resume_verbs_never_strand_a_caller() {
+		// UI-1822: a verb that passed the paused check could enqueue its
+		// command just as the command loop consumed another caller's resume
+		// and exited - the command's Done was never signalled and the caller
+		// blocked forever. The enqueue and the resume-drain are now atomic:
+		// every racer either wins the resume or gets an error, and always
+		// returns. The Join timeouts are the strand detectors.
+		var source = "fromAll().when({\n$init: function() { return { count: 0 }; },\nItemAdded: function(s, e) {\ns.count++;\nreturn s;\n}\n})";
+		using var session = new ProjectionSession(source, new ProjectionSessionOptions { EngineVersion = ProjectionVersion.V2, Debug = true });
+
+		var paused = new ManualResetEventSlim(false);
+		session.OnBreak = _ => paused.Set();
+		session.SetBreakpoint(4);
+
+		for (var i = 0; i < 25; i++) {
+			paused.Reset();
+			Exception? feedEx = null;
+			// Background threads: on a regression the strand leaves them
+			// blocked past the failed Join assert, and a foreground thread
+			// would then hang the test host instead of reporting the failure.
+			var feedThread = new Thread(() => {
+				try {
+					session.Feed(MakeEvent());
+				} catch (Exception ex) {
+					feedEx = ex;
+				}
+			}) { IsBackground = true };
+			feedThread.Start();
+			Assert.True(paused.Wait(TimeSpan.FromSeconds(5)));
+
+			var racers = new Thread[3];
+			for (var r = 0; r < racers.Length; r++) {
+				racers[r] = new Thread(() => {
+					try {
+						session.Continue();
+					} catch (InvalidOperationException) {
+						// Lost the race: "not paused", or drained with
+						// "resumed before the command ran". Both fine -
+						// only hanging is a bug.
+					}
+				}) { IsBackground = true };
+				racers[r].Start();
+			}
+			foreach (var racer in racers)
+				Assert.True(racer.Join(TimeSpan.FromSeconds(10)), "a resume verb stranded: its Done was never signalled");
+			Assert.True(feedThread.Join(TimeSpan.FromSeconds(10)));
+			Assert.Null(feedEx);
+		}
+
+		Assert.Contains("\"count\":25", session.GetState()!);
+	}
+
+	[Fact]
+	public void Resume_race_does_not_replay_into_the_next_pause() {
+		// The other half of UI-1822: a command left in the queue when the
+		// loop exited replayed at the NEXT pause - a stale Continue from a
+		// lost race would silently resume a breakpoint the user never
+		// continued. The resume-drain must leave the queue empty.
+		var source = "fromAll().when({\n$init: function() { return { count: 0 }; },\nItemAdded: function(s, e) {\ns.count++;\nreturn s;\n}\n})";
+		using var session = new ProjectionSession(source, new ProjectionSessionOptions { EngineVersion = ProjectionVersion.V2, Debug = true });
+
+		var paused = new ManualResetEventSlim(false);
+		session.OnBreak = _ => paused.Set();
+		session.SetBreakpoint(4);
+
+		// Background threads for the same reason as the racing test above:
+		// a regression must fail the test, not hang the test host.
+		Exception? feedEx = null;
+		var feed1 = new Thread(() => {
+			try {
+				session.Feed(MakeEvent());
+			} catch (Exception ex) {
+				feedEx = ex;
+			}
+		}) { IsBackground = true };
+		feed1.Start();
+		Assert.True(paused.Wait(TimeSpan.FromSeconds(5)));
+
+		var racers = new Thread[2];
+		for (var r = 0; r < racers.Length; r++) {
+			racers[r] = new Thread(() => {
+				try {
+					session.Continue();
+				} catch (InvalidOperationException) {
+				}
+			}) { IsBackground = true };
+			racers[r].Start();
+		}
+		foreach (var racer in racers)
+			Assert.True(racer.Join(TimeSpan.FromSeconds(10)), "a resume verb stranded: its Done was never signalled");
+		Assert.True(feed1.Join(TimeSpan.FromSeconds(10)));
+		Assert.Null(feedEx);
+
+		// The second pause must wait for its own resume.
+		paused.Reset();
+		var feed2 = new Thread(() => {
+			try {
+				session.Feed(MakeEvent());
+			} catch (Exception ex) {
+				feedEx = ex;
+			}
+		}) { IsBackground = true };
+		feed2.Start();
+		Assert.True(paused.Wait(TimeSpan.FromSeconds(5)));
+		Assert.False(feed2.Join(TimeSpan.FromMilliseconds(300)), "second pause auto-resumed: a stale command replayed");
+		Assert.True(session.IsPaused);
+
+		session.Continue();
+		Assert.True(feed2.Join(TimeSpan.FromSeconds(10)));
+		Assert.Null(feedEx);
+		Assert.Contains("\"count\":2", session.GetState()!);
+	}
 }

runtime/Gaffer.Runtime/Projection/JintProjectionHandler.cs

@@ -70,6 +70,9 @@ internal sealed class JintProjectionHandler : IDisposable {
 	private readonly ProjectionVersion _engineVersion;
 	private readonly TimeConstraint _timeConstraint;
 	private readonly BlockingCollection<DebugCommand> _debugCommands = new();
+	// Makes a verb's paused-check-then-enqueue atomic with the command loop's
+	// resume-then-drain (see EnqueueDebugCommand / ResumeAndDrainDebugCommands).
+	private readonly object _debugSync = new();
 	private readonly Dictionary<int, object> _variableStore = new();
 	private List<BreakablePosition>? _breakablePositions;
 	private int _nextVariableRef = 1;
@@ -190,6 +193,11 @@ public JintProjectionHandler(
 
 	public void Dispose() {
 		_debugCommands.CompleteAdding();
+		// Fail anything still queued so no caller is left blocked on a Done
+		// that will never be signalled. With the enqueue/resume lock this
+		// should be unreachable - a queued command implies a consuming loop -
+		// but Dispose mustn't bet on it.
+		DrainDebugCommands();
 		_engine.Dispose();
 	}
 
@@ -619,11 +627,11 @@ private StepMode EnterDebugCommandLoop(string reason, DebugInformation info) {
 			foreach (var cmd in _debugCommands.GetConsumingEnumerable()) {
 				switch (cmd) {
 					case ContinueCommand cc:
-						ClearDebugState();
+						ResumeAndDrainDebugCommands();
 						cc.Done.Set();
 						return StepMode.None;
 					case StepCommand sc:
-						ClearDebugState();
+						ResumeAndDrainDebugCommands();
 						sc.Done.Set();
 						return sc.Mode;
 					case GetCallStackCommand gc:
@@ -648,11 +656,11 @@ private StepMode EnterDebugCommandLoop(string reason, DebugInformation info) {
 				}
 			}
 		} catch {
-			ClearDebugState();
+			ResumeAndDrainDebugCommands();
 			throw;
 		}
 
-		ClearDebugState();
+		ResumeAndDrainDebugCommands();
 		throw new OperationCanceledException("Debug session disposed while paused");
 	}
 
@@ -663,6 +671,50 @@ private void ClearDebugState() {
 		_nextVariableRef = 1;
 	}
 
+	/// <summary>
+	/// Flips the engine back to running and fails any command that lost the
+	/// race with the resume, atomically with respect to
+	/// <see cref="EnqueueDebugCommand"/>. Every exit from the command loop
+	/// must come through here: a command left in the queue when the loop
+	/// exits blocks its caller forever (its Done is never set) and then
+	/// replays as a stale command at the next pause.
+	/// </summary>
+	private void ResumeAndDrainDebugCommands() {
+		lock (_debugSync) {
+			ClearDebugState();
+			DrainDebugCommands();
+		}
+	}
+
+	private void DrainDebugCommands() {
+		while (_debugCommands.TryTake(out var stale)) {
+			stale.Error = new InvalidOperationException("Engine resumed before the command ran");
+			stale.Done.Set();
+		}
+	}
+
+	/// <summary>
+	/// The verbs' shared enqueue: the paused check and the enqueue are one
+	/// atomic step, so a command can only enter the queue while the command
+	/// loop is still consuming (or is guaranteed to drain it on exit).
+	/// Without the lock, a verb that passed the check could add after the
+	/// loop consumed a resume and exited - the strand UI-1822 describes.
+	/// </summary>
+	private void EnqueueDebugCommand(DebugCommand cmd, string notPausedMessage) {
+		lock (_debugSync) {
+			if (!_paused)
+				throw new InvalidOperationException(notPausedMessage);
+			try {
+				_debugCommands.Add(cmd);
+			} catch (InvalidOperationException) {
+				// Dispose completed the collection between the check and the
+				// add. Surface the same refusal the caller would have gotten
+				// a beat later, not the framework's marked-as-complete text.
+				throw new InvalidOperationException(notPausedMessage);
+			}
+		}
+	}
+
 	private DebugCallFrame[] ReadCallStack(DebugInformation info) {
 		var stack = info.CallStack;
 		var frames = new DebugCallFrame[stack.Count];
@@ -954,75 +1006,67 @@ private void RemovePauseBreakPoints() {
 	}
 
 	public void Continue() {
-		if (!_paused)
-			throw new InvalidOperationException("Cannot continue when not paused");
 		using var cmd = new ContinueCommand();
-		_debugCommands.Add(cmd);
+		EnqueueDebugCommand(cmd, "Cannot continue when not paused");
 		cmd.Done.Wait();
+		if (cmd.Error != null)
+			throw cmd.Error;
 	}
 
 	public void StepInto() {
-		if (!_paused)
-			throw new InvalidOperationException("Cannot step when not paused");
 		using var cmd = new StepCommand { Mode = StepMode.Into };
-		_debugCommands.Add(cmd);
+		EnqueueDebugCommand(cmd, "Cannot step when not paused");
 		cmd.Done.Wait();
+		if (cmd.Error != null)
+			throw cmd.Error;
 	}
 
 	public void StepOver() {
-		if (!_paused)
-			throw new InvalidOperationException("Cannot step when not paused");
 		using var cmd = new StepCommand { Mode = StepMode.Over };
-		_debugCommands.Add(cmd);
+		EnqueueDebugCommand(cmd, "Cannot step when not paused");
 		cmd.Done.Wait();
+		if (cmd.Error != null)
+			throw cmd.Error;
 	}
 
 	public void StepOut() {
-		if (!_paused)
-			throw new InvalidOperationException("Cannot step when not paused");
 		using var cmd = new StepCommand { Mode = StepMode.Out };
-		_debugCommands.Add(cmd);
+		EnqueueDebugCommand(cmd, "Cannot step when not paused");
 		cmd.Done.Wait();
+		if (cmd.Error != null)
+			throw cmd.Error;
 	}
 
 	public DebugCallFrame[] GetCallStack() {
-		if (!_paused)
-			throw new InvalidOperationException("Cannot inspect when not paused");
 		using var cmd = new GetCallStackCommand();
-		_debugCommands.Add(cmd);
+		EnqueueDebugCommand(cmd, "Cannot inspect when not paused");
 		cmd.Done.Wait();
 		if (cmd.Error != null)
 			throw cmd.Error;
 		return cmd.Result!;
 	}
 
 	public DebugScopeInfo[] GetScopes(int frameIndex) {
-		if (!_paused)
-			throw new InvalidOperationException("Cannot inspect when not paused");
 		using var cmd = new GetScopesCommand { FrameIndex = frameIndex };
-		_debugCommands.Add(cmd);
+		EnqueueDebugCommand(cmd, "Cannot inspect when not paused");
 		cmd.Done.Wait();
 		if (cmd.Error != null)
 			throw cmd.Error;
 		return cmd.Result!;
 	}
 
 	public DebugVariable[] GetVariables(int variablesReference) {
-		if (!_paused)
-			throw new InvalidOperationException("Cannot inspect when not paused");
 		using var cmd = new GetVariablesCommand { VariablesReference = variablesReference };
-		_debugCommands.Add(cmd);
+		EnqueueDebugCommand(cmd, "Cannot inspect when not paused");
 		cmd.Done.Wait();
 		if (cmd.Error != null)
 			throw cmd.Error;
 		return cmd.Result!;
 	}
 
 	public DebugVariable Evaluate(string expression) {
-		if (!_paused)
-			throw new InvalidOperationException("Cannot evaluate when not paused");
 		using var cmd = new EvaluateCommand { Expression = expression };
-		_debugCommands.Add(cmd);
+		EnqueueDebugCommand(cmd, "Cannot evaluate when not paused");
 		cmd.Done.Wait();
 		if (cmd.Error != null)
 			throw cmd.Error;