From 262b0c41825174d8201157df827d1ef63ba647f9 Mon Sep 17 00:00:00 2001 From: George Payne Date: Tue, 7 Jul 2026 11:10:53 +0200 Subject: [PATCH 1/2] Make debug command enqueue atomic with the engine's resume The debug verbs did an unsynchronized paused check-then-enqueue while the command loop exits on the first Continue/Step without draining the queue. Two failure modes for a verb that lost the race with a resume (UI-1822): its command stranded - Done never signalled, the caller blocked forever (which since the Runner teardown gate hangs Destroy and wedges the MCP server's closeSession under s.mu) - and the stale command lingered in the queue and replayed at the NEXT pause, silently resuming a breakpoint the user never continued. Drain-at-exit alone can't fix it: a verb that passed the check can enqueue after any drain. - _debugSync makes each verb's check+enqueue atomic with the loop's resume+drain. Every loop exit (resume, exception, disposed-while- paused) goes through ResumeAndDrainDebugCommands: clear state, then fail anything still queued with an error, under the same lock. A command observed-paused-and-enqueued is therefore always either consumed or failed - never stranded, never carried across a pause. - All eight verbs check cmd.Error after their wait, so a drained resume surfaces as an error to the caller instead of a silent no-op. - Dispose drains as a belt-and-braces, and an enqueue that races CompleteAdding gets the verb's own refusal message. - Tests: 25x3 racing Continues with Join timeouts as strand detectors, and a stale-replay guard asserting the next pause waits for its own resume. Validated end-to-end: published runtime + cli dap/mcpserver suites and bindings under -race. - Go-side comments that documented the hazard now state the guarantee. --- .changeset/debug-command-strand.md | 5 + cli/internal/engine/runner_debug.go | 21 ++-- runtime/Gaffer.Runtime.Tests/DebugTests.cs | 109 ++++++++++++++++++ .../Projection/JintProjectionHandler.cs | 100 +++++++++++----- 4 files changed, 196 insertions(+), 39 deletions(-) create mode 100644 .changeset/debug-command-strand.md diff --git a/.changeset/debug-command-strand.md b/.changeset/debug-command-strand.md new file mode 100644 index 00000000..243048a6 --- /dev/null +++ b/.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. diff --git a/cli/internal/engine/runner_debug.go b/cli/internal/engine/runner_debug.go index f089ce0c..908bcac7 100644 --- a/cli/internal/engine/runner_debug.go +++ b/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. diff --git a/runtime/Gaffer.Runtime.Tests/DebugTests.cs b/runtime/Gaffer.Runtime.Tests/DebugTests.cs index 65fba317..e23c4bf6 100644 --- a/runtime/Gaffer.Runtime.Tests/DebugTests.cs +++ b/runtime/Gaffer.Runtime.Tests/DebugTests.cs @@ -805,4 +805,113 @@ 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; + var feedThread = new Thread(() => { + try { + session.Feed(MakeEvent()); + } catch (Exception ex) { + feedEx = ex; + } + }); + 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. + } + }); + 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); + + Exception? feedEx = null; + var feed1 = new Thread(() => { + try { + session.Feed(MakeEvent()); + } catch (Exception ex) { + feedEx = ex; + } + }); + 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) { + } + }); + 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; + } + }); + 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()!); + } } diff --git a/runtime/Gaffer.Runtime/Projection/JintProjectionHandler.cs b/runtime/Gaffer.Runtime/Projection/JintProjectionHandler.cs index c177bb03..bc40479b 100644 --- a/runtime/Gaffer.Runtime/Projection/JintProjectionHandler.cs +++ b/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 _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 _variableStore = new(); private List? _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; } + /// + /// Flips the engine back to running and fails any command that lost the + /// race with the resume, atomically with respect to + /// . 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. + /// + 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(); + } + } + + /// + /// 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. + /// + 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,42 +1006,40 @@ 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; @@ -997,10 +1047,8 @@ public DebugCallFrame[] GetCallStack() { } 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; @@ -1008,10 +1056,8 @@ public DebugScopeInfo[] GetScopes(int frameIndex) { } 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; @@ -1019,10 +1065,8 @@ public DebugVariable[] GetVariables(int variablesReference) { } 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; From ea93a5289dd47dad80fb3adb9cf0c4947c57bb00 Mon Sep 17 00:00:00 2001 From: George Payne Date: Tue, 7 Jul 2026 11:19:37 +0200 Subject: [PATCH 2/2] Run debug-race test threads as background threads Qodo: on a regression, the strand leaves the racer/feed threads blocked past the failed Join assert, and foreground threads then hang the test host at exit - a CI job timeout instead of a reported test failure. On the fixed code Dispose's drain releases them either way; IsBackground only changes how a future regression fails. --- runtime/Gaffer.Runtime.Tests/DebugTests.cs | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/runtime/Gaffer.Runtime.Tests/DebugTests.cs b/runtime/Gaffer.Runtime.Tests/DebugTests.cs index e23c4bf6..06f416c3 100644 --- a/runtime/Gaffer.Runtime.Tests/DebugTests.cs +++ b/runtime/Gaffer.Runtime.Tests/DebugTests.cs @@ -824,13 +824,16 @@ public void Racing_resume_verbs_never_strand_a_caller() { 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))); @@ -844,7 +847,7 @@ public void Racing_resume_verbs_never_strand_a_caller() { // "resumed before the command ran". Both fine - // only hanging is a bug. } - }); + }) { IsBackground = true }; racers[r].Start(); } foreach (var racer in racers) @@ -869,6 +872,8 @@ public void Resume_race_does_not_replay_into_the_next_pause() { 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 { @@ -876,7 +881,7 @@ public void Resume_race_does_not_replay_into_the_next_pause() { } catch (Exception ex) { feedEx = ex; } - }); + }) { IsBackground = true }; feed1.Start(); Assert.True(paused.Wait(TimeSpan.FromSeconds(5))); @@ -887,7 +892,7 @@ public void Resume_race_does_not_replay_into_the_next_pause() { session.Continue(); } catch (InvalidOperationException) { } - }); + }) { IsBackground = true }; racers[r].Start(); } foreach (var racer in racers) @@ -903,7 +908,7 @@ public void Resume_race_does_not_replay_into_the_next_pause() { } 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");