Support multi-process debugging: sync breakpoints and coordinate instances

lib/debug/breakpoint.rb

@@ -6,7 +6,7 @@ module DEBUGGER__
   class Breakpoint
     include SkipPathHelper
 
-    attr_reader :key, :skip_src
+    attr_reader :key, :skip_src, :cond
 
     def initialize cond, command, path, do_enable: true
       @deleted = false
@@ -19,6 +19,16 @@ def initialize cond, command, path, do_enable: true
       enable if do_enable
     end
 
+    # Returns a serializable hash for cross-process breakpoint sync,
+    # or nil if this breakpoint type is not syncable.
+    def to_sync_data
+      nil
+    end
+
+    def syncable?
+      false
+    end
+
     def safe_eval b, expr
       b.eval(expr)
     rescue Exception => e
@@ -221,6 +231,16 @@ def activate_exact iseq, events, line
       end
     end
 
+    def to_sync_data
+      { 'type' => 'line', 'path' => @path, 'line' => @line,
+        'cond' => @cond, 'oneshot' => @oneshot,
+        'hook_call' => @hook_call, 'command' => @command }
+    end
+
+    def syncable?
+      true
+    end
+
     def duplicable?
       @oneshot
     end
@@ -302,6 +322,15 @@ def path_is? path
   class CatchBreakpoint < Breakpoint
     attr_reader :last_exc
 
+    def to_sync_data
+      { 'type' => 'catch', 'pat' => @pat, 'cond' => @cond,
+        'command' => @command, 'path' => @path }
+    end
+
+    def syncable?
+      true
+    end
+
     def initialize pat, cond: nil, command: nil, path: nil
       @pat = pat.freeze
       @key = [:catch, @pat].freeze
@@ -427,6 +456,16 @@ def to_s
   class MethodBreakpoint < Breakpoint
     attr_reader :sig_method_name, :method, :klass
 
+    def to_sync_data
+      { 'type' => 'method', 'klass' => @sig_klass_name,
+        'op' => @sig_op, 'method' => @sig_method_name,
+        'cond' => @cond, 'command' => @command }
+    end
+
+    def syncable?
+      true
+    end
+
     def initialize b, klass_name, op, method_name, cond: nil, command: nil, path: nil
       @sig_klass_name = klass_name
       @sig_op = op

lib/debug/irb_integration.rb

@@ -21,6 +21,9 @@ def activate_irb_integration
       irb = IRB::Irb.new(workspace)
       IRB.conf[:MAIN_CONTEXT] = irb.context
       IRB::Debug.setup(irb)
+      if (pi = SESSION.process_info)
+        irb.context.irb_name = "irb:rdbg@#{pi}"
+      end
       IRB::Context.prepend(IrbPatch)
     end
   end

lib/debug/server.rb

@@ -186,6 +186,9 @@ def process
         line = @session.process_group.sync do
           unless IO.select([@sock], nil, nil, 0)
             DEBUGGER__.debug{ "UI_Server can not read" }
+            # Wait briefly for the consuming process to publish breakpoint changes
+            sleep 0.05
+            @session.bp_sync_check
             break :can_not_read
           end
           @sock.gets&.chomp.tap{|line|

lib/debug/server_dap.rb

@@ -271,6 +271,10 @@ def recv_request
         end
       end
     rescue RetryBecauseCantRead
+      # Another process consumed the message. Wait briefly for it to
+      # process and publish any breakpoint changes, then sync.
+      sleep 0.05
+      @session.bp_sync_check
       retry
     end
 
@@ -356,6 +360,7 @@ def process_request req
               bps << SESSION.add_line_breakpoint(path, line)
             end
           }
+          SESSION.bp_sync_publish
           send_response req, breakpoints: (bps.map do |bp| {verified: true,} end)
         else
           send_response req, breakpoints: (args['breakpoints'].map do |bp| {verified: false, message: "#{req_path} could not be located; specify source location in launch.json with \"localfsMap\" or \"localfs\""} end)
@@ -391,12 +396,14 @@ def process_request req
           process_filter.call(bp_info['filterId'], bp_info['condition'])
         }
 
+        SESSION.bp_sync_publish
         send_response req, breakpoints: filters
 
       when 'disconnect'
         terminate = args.fetch("terminateDebuggee", false)
 
         SESSION.clear_all_breakpoints
+        SESSION.bp_sync_publish
         send_response req
 
         if SESSION.in_subsession?

lib/debug/session.rb

@@ -90,10 +90,86 @@ module DEBUGGER__
 
   class PostmortemError < RuntimeError; end
 
+  module BreakpointSync
+    def bp_sync_publish
+      return unless @process_group.multi?
+      @process_group.write_breakpoint_state(serialize_sync_breakpoints)
+    end
+
+    def bp_sync_check
+      return false unless @process_group.multi?
+      specs = @process_group.read_breakpoint_state
+      return false unless specs
+      reconcile_breakpoints(specs)
+      true
+    end
+
+    private
+
+    def serialize_sync_breakpoints
+      @bps.filter_map { |_key, bp| bp.to_sync_data }
+    end
+
+    def reconcile_breakpoints(specs)
+      remote_keys = {}
+
+      specs.each do |spec|
+        key = bp_key_from_spec(spec)
+        next unless key
+        remote_keys[key] = true
+        unless @bps.key?(key)
+          create_bp_from_spec(spec)
+        end
+      end
+
+      @bps.delete_if do |key, bp|
+        if syncable_bp?(bp) && !remote_keys.key?(key)
+          bp.delete
+          true
+        end
+      end
+    end
+
+    def bp_key_from_spec(spec)
+      case spec['type']
+      when 'line'   then [spec['path'], spec['line']]
+      when 'catch'  then [:catch, spec['pat']]
+      when 'method' then "#{spec['klass']}#{spec['op']}#{spec['method']}"
+      end
+    end
+
+    def create_bp_from_spec(spec)
+      bp = case spec['type']
+           when 'line'
+             return unless spec['path'].is_a?(String) && spec['line'].is_a?(Integer)
+             LineBreakpoint.new(spec['path'], spec['line'],
+               cond: spec['cond'], oneshot: spec['oneshot'],
+               hook_call: spec['hook_call'] != false,
+               command: spec['command'])
+           when 'catch'
+             return unless spec['pat'].is_a?(String)
+             CatchBreakpoint.new(spec['pat'],
+               cond: spec['cond'], command: spec['command'],
+               path: spec['path'])
+           when 'method'
+             return unless spec['klass'].is_a?(String) && spec['op'].is_a?(String) && spec['method'].is_a?(String)
+             MethodBreakpoint.new(TOPLEVEL_BINDING, spec['klass'], spec['op'], spec['method'],
+               cond: spec['cond'], command: spec['command'])
+           end
+
+      add_bp(bp) if bp
+    end
+
+    def syncable_bp?(bp)
+      bp.syncable?
+    end
+  end
+
   class Session
     attr_reader :intercepted_sigint_cmd, :process_group, :subsession_id
 
     include Color
+    include BreakpointSync
 
     def initialize
       @ui = nil
@@ -417,8 +493,8 @@ def wait_command_loop
     def prompt
       if @postmortem
         '(rdbg:postmortem) '
-      elsif @process_group.multi?
-        "(rdbg@#{process_info}) "
+      elsif (pi = process_info)
+        "(rdbg@#{pi}) "
       else
         '(rdbg) '
       end
@@ -1711,8 +1787,10 @@ def get_thread_client th = Thread.current
         DEBUGGER__.debug{ "Enter subsession (nested #{@subsession_stack.size})" }
       else
         DEBUGGER__.debug{ "Enter subsession" }
+        @process_group.wk_lock  # blocks until no other debugger is active
         stop_all_threads
         @process_group.lock
+        bp_sync_check  # sync breakpoints from other processes
       end
 
       @subsession_stack << true
@@ -1724,7 +1802,12 @@ def get_thread_client th = Thread.current
 
       if @subsession_stack.empty?
         DEBUGGER__.debug{ "Leave subsession" }
+        bp_sync_publish  # publish breakpoint changes to other processes
         @process_group.unlock
+        # Keep wk_lock held during step commands so the same worker
+        # re-enters the subsession without yielding to a sibling.
+        # Release on :continue so long-lived workers don't starve others.
+        @process_group.unlock_wk_lock if type == :continue
         restart_all_threads
       else
         DEBUGGER__.debug{ "Leave subsession (nested #{@subsession_stack.size})" }
@@ -2002,7 +2085,7 @@ def intercept_trap_sigint_end
     end
 
     def process_info
-      if @process_group.multi?
+      if @process_group.multi? || @process_group.wk_locked?
         "#{$0}\##{Process.pid}"
       end
     end
@@ -2028,6 +2111,7 @@ def extend_feature session: nil, thread_client: nil, ui: nil
   class ProcessGroup
     def initialize
       @lock_file = nil
+      @wk_lock_file = nil
     end
 
     def locked?
@@ -2057,10 +2141,45 @@ def multi?
       @lock_file
     end
 
+    # No-ops for single-process mode; overridden by MultiProcessGroup
+    def write_breakpoint_state(specs); end
+    def read_breakpoint_state; nil; end
+
+    # Well-known lock for coordinating independent debugger instances
+    # (e.g., parallel test workers that each load the debugger independently).
+    # Uses process group ID so sibling processes from the same command share the lock.
+    # Blocks until the lock is acquired — other workers wait in line.
+    def wk_lock
+      return if multi?  # MultiProcessGroup handles its own locking
+      ensure_wk_lock!
+      @wk_lock_file&.flock(File::LOCK_EX)
+    end
+
+    def wk_locked?
+      !multi? && @wk_lock_file
+    end
+
+    def unlock_wk_lock
+      return if multi?
+      @wk_lock_file&.flock(File::LOCK_UN)
+    end
+
+    private def ensure_wk_lock!
+      return if @wk_lock_file
+      require 'tmpdir'
+      path = File.join(Dir.tmpdir, "ruby-debug-#{Process.uid}-pgrp-#{Process.getpgrp}.lock")
+      @wk_lock_file = File.open(path, File::WRONLY | File::CREAT, 0600)
+    rescue SystemCallError => e
+      DEBUGGER__.warn "Failed to create well-known lock file: #{e.message}"
+    end
+
     def multi_process!
       require 'tempfile'
+      require 'json'
       @lock_tempfile = Tempfile.open("ruby-debug-lock-")
       @lock_tempfile.close
+      @state_tempfile = Tempfile.open("ruby-debug-state-")
+      @state_tempfile.close
       extend MultiProcessGroup
     end
   end
@@ -2076,6 +2195,7 @@ def after_fork child: true
           @lock_level = 0
           @lock_file = open(@lock_tempfile.path, 'w')
         end
+        @bp_sync_version = 0
       end
     end
 
@@ -2146,6 +2266,34 @@ def unlock
       end
     end
 
+    def write_breakpoint_state(specs)
+      # Read current file version to avoid drift between processes
+      current_v = begin
+        d = JSON.parse(File.read(@state_tempfile.path))
+        d['v']
+      rescue
+        0
+      end
+      @bp_sync_version = [current_v, @bp_sync_version].max + 1
+      data = JSON.generate({ 'v' => @bp_sync_version, 'bps' => specs })
+      tmp = "#{@state_tempfile.path}.#{Process.pid}.tmp"
+      File.write(tmp, data, perm: 0600)
+      File.rename(tmp, @state_tempfile.path)
+    rescue SystemCallError => e
+      DEBUGGER__.warn "Failed to write breakpoint state: #{e.message}"
+    end
+
+    def read_breakpoint_state
+      return nil unless File.exist?(@state_tempfile.path)
+      data = JSON.parse(File.read(@state_tempfile.path))
+      remote_v = data['v']
+      return nil if remote_v <= @bp_sync_version
+      @bp_sync_version = remote_v
+      data['bps']
+    rescue JSON::ParserError, SystemCallError
+      nil
+    end
+
     def sync &b
       info "sync"
 
@@ -2547,6 +2695,7 @@ def daemon(*args)
         child_hook = -> {
           DEBUGGER__.info "Attaching after process #{parent_pid} fork to child process #{Process.pid}"
           SESSION.process_group.after_fork child: true
+          SESSION.bp_sync_check
           SESSION.activate on_fork: true
         }
       end