Improve server and monitor robustness.

Author: ioquatix

async-service-supervisor.gemspec

@@ -24,6 +24,7 @@ Gem::Specification.new do |spec|
 
 	spec.required_ruby_version = ">= 3.3"
 
+	spec.add_dependency "async", "~> 2.38"
 	spec.add_dependency "async-bus"
 	spec.add_dependency "async-service", "~> 0.15"
 	spec.add_dependency "async-utilization", "~> 0.3"

examples/resize_mmap/test.rb

@@ -0,0 +1,186 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+# High-frequency regression test: does supervisor-side ftruncate + mmap remap
+# invalidate worker-side mmap slices (as created by Observer.open)?
+#
+# The supervisor (parent process) holds @file open and calls:
+#   @file.truncate(new_size)
+#   @buffer.free
+#   @buffer = IO::Buffer.map(@file, new_size)
+#
+# The worker (child process) independently opens the same file, maps a
+# page-aligned window covering its segment, closes the fd, and holds a slice
+# of the resulting buffer — exactly what Observer.open does.
+#
+# This test runs both sides concurrently at maximum speed to expose any
+# platform/version scenario where ftruncate or munmap (via #free) on one
+# process's mapping invalidates another process's independent mapping.
+#
+# Usage:
+#   ruby examples/resize_mmap/test.rb
+
+require "tmpdir"
+
+PAGE_SIZE    = IO::Buffer::PAGE_SIZE
+SEGMENT_SIZE = 512
+ITERATIONS   = 200_000   # write iterations per worker
+RESIZES      = 20        # supervisor-side resizes per test run
+
+# Reproduce the exact mapping logic from Observer.open.
+# Returns [slice, parent_buffer] — caller must hold parent_buffer to prevent GC
+# (IO::Buffer#slice keeps a back-reference to the parent in Ruby's C layer, but
+# we return it explicitly to make the lifetime contract obvious in tests).
+def worker_map(path, segment_offset, segment_size)
+	page_size         = IO::Buffer::PAGE_SIZE
+	aligned_offset    = (segment_offset / page_size) * page_size
+	offset_adjustment = segment_offset - aligned_offset
+	aligned_end       = (((segment_offset + segment_size) + page_size - 1) / page_size) * page_size
+	map_size          = [aligned_end - aligned_offset, page_size].max
+
+	file = File.open(path, "r+b")
+	# Mirror Observer.open: fd is closed after mapping; mapping persists independently.
+	parent = IO::Buffer.map(file, map_size, aligned_offset)
+	file.close
+
+	slice = (offset_adjustment > 0 || map_size > segment_size) ?
+		parent.slice(offset_adjustment, segment_size) :
+		parent
+
+	[slice, parent]
+end
+
+# Run one test scenario.
+#
+# Layout within a segment (as seen by both sides):
+#   [0..7]  u64  iteration counter — final value must equal ITERATIONS
+#   [8..15] u64  write-error count — must be 0
+#
+# Synchronisation:
+#   mapped_w/r  child → parent: "I have mapped the file, you may start resizing"
+#   done_w/r    child → parent: "I have finished writing all ITERATIONS values"
+#   exit_w/r    parent → child: "you may exit now"
+def run_test(path, segment_offset, label)
+	File.open(path, "w+b") { |f| f.truncate(PAGE_SIZE) }
+
+	mapped_r, mapped_w = IO.pipe   # child signals: file mapped
+	done_r,   done_w   = IO.pipe   # child signals: writes complete
+	exit_r,   exit_w   = IO.pipe   # parent signals: child may exit
+
+	child_pid = fork do
+		[mapped_r, done_r, exit_w].each(&:close)
+
+		slice, _parent = worker_map(path, segment_offset, SEGMENT_SIZE)
+
+		mapped_w.write("1")
+		mapped_w.close   # unblock parent resize loop
+
+		errors = 0
+		ITERATIONS.times do |i|
+			begin
+				slice.set_value(:u64, 0, i + 1)
+			rescue => e
+				errors += 1
+			end
+		end
+		slice.set_value(:u64, 8, errors)
+
+		done_w.write("1")
+		done_w.close     # unblock parent result read
+
+		exit_r.read
+		exit_r.close
+		exit 0
+	end
+
+	[mapped_w, done_w, exit_r].each(&:close)
+
+	# Wait for the worker to have mapped the file before resizing.
+	mapped_r.read
+	mapped_r.close
+
+	# Supervisor side: resize at maximum speed, concurrently with the child writes.
+	# Use a thread so the resize loop and the done_r.read can overlap in time.
+	sup_file   = File.open(path, "r+b")
+	sup_buffer = IO::Buffer.map(sup_file, PAGE_SIZE)
+	curr_size  = PAGE_SIZE
+
+	resize_thread = Thread.new do
+		RESIZES.times do
+			curr_size *= 2
+			sup_file.truncate(curr_size)
+			sup_buffer.free
+			sup_buffer = IO::Buffer.map(sup_file, curr_size)
+		end
+	end
+
+	# Block until the child has finished all writes.
+	done_r.read
+	done_r.close
+
+	# Drain the resize thread before reading results.
+	resize_thread.join
+
+	final_counter = sup_buffer.get_value(:u64, segment_offset + 0)
+	write_errors  = sup_buffer.get_value(:u64, segment_offset + 8)
+
+	sup_buffer.free
+	sup_file.close
+
+	exit_w.write("1")
+	exit_w.close
+
+	Process.waitpid(child_pid)
+	File.unlink(path) rescue nil
+
+	[final_counter, write_errors]
+end
+
+# ── Main ──────────────────────────────────────────────────────────────────────
+
+puts "Platform:    #{RUBY_PLATFORM}"
+puts "Ruby:        #{RUBY_VERSION}"
+puts "PAGE_SIZE:   #{PAGE_SIZE}"
+puts "SEGMENT_SIZE: #{SEGMENT_SIZE}"
+puts "ITERATIONS:  #{ITERATIONS}"
+puts "RESIZES:     #{RESIZES}  (#{PAGE_SIZE} → #{PAGE_SIZE * 2**RESIZES} bytes)"
+puts
+
+failures = 0
+
+Dir.mktmpdir do |dir|
+	path = File.join(dir, "test.shm")
+
+	tests = {
+		"page-aligned offset (segment_offset=0, exercises map_size > segment_size slice path)" => 0,
+		"non-page-aligned offset (segment_offset=#{SEGMENT_SIZE}, exercises offset_adjustment > 0 slice path)" => SEGMENT_SIZE,
+	}
+
+	tests.each_with_index do |(label, segment_offset), index|
+		print "Test #{index + 1}: #{label}\n         ... "
+		$stdout.flush
+
+		final, errors = run_test(path, segment_offset, label)
+
+		if errors > 0 || final != ITERATIONS
+			puts "FAIL"
+			puts "  expected counter = #{ITERATIONS}, got #{final}"
+			puts "  write errors (IO::Buffer exceptions) = #{errors}"
+			if final != ITERATIONS
+				puts "  => writes went to stale/invalid pages and were not visible to the supervisor"
+			end
+			failures += 1
+		else
+			puts "PASS (counter=#{ITERATIONS}, errors=0)"
+		end
+	end
+end
+
+puts
+if failures == 0
+	puts "All tests passed — ftruncate + IO::Buffer.map remap does NOT invalidate existing worker mmap slices on this platform."
+	exit 0
+else
+	puts "#{failures} test(s) FAILED — existing mmap slices were invalidated by ftruncate/remap on this platform!"
+	exit 1
+end

lib/async/service/supervisor/loop.rb

@@ -6,23 +6,23 @@
 require "memory/leak/cluster"
 require "set"
 
-require_relative "loop"
+require_relative "monitor"
 
 module Async
 	module Service
 		module Supervisor
 			# Monitors worker memory usage and restarts workers that exceed limits.
 			#
 			# Uses the `memory` gem to track process memory and detect leaks.
-			class MemoryMonitor
+			class MemoryMonitor < Monitor
 				# Create a new memory monitor.
 				#
 				# @parameter interval [Integer] The interval at which to check for memory leaks.
 				# @parameter total_size_limit [Integer] The total size limit of all processes, or nil for no limit.
 				# @parameter free_size_minimum [Integer] The minimum free memory threshold, or nil for no threshold.
 				# @parameter options [Hash] Options to pass to the cluster when adding processes.
 				def initialize(interval: 10, total_size_limit: nil, free_size_minimum: nil, **options)
-					@interval = interval
+					super(interval: interval)
 					@cluster = Memory::Leak::Cluster.new(total_size_limit: total_size_limit, free_size_minimum: free_size_minimum)
 
 					# We use these options when adding processes to the cluster:
@@ -85,28 +85,11 @@ def remove(supervisor_controller)
 					end
 				end
 
-				# The key used when this monitor's status is aggregated with others.
-				def self.monitor_type
-					:memory_monitor
-				end
-				
 				# Serialize memory cluster data for JSON.
 				def as_json
 					@cluster.as_json
 				end
 
-				# Serialize to JSON string.
-				def to_json(...)
-					as_json.to_json(...)
-				end
-				
-				# Get status for the memory monitor.
-				#
-				# @returns [Hash] Hash with type and data keys.
-				def status
-					{type: self.class.monitor_type, data: as_json}
-				end
-				
 				# Invoked when a memory leak is detected.
 				#
 				# @parameter process_id [Integer] The process ID of the process that has a memory leak.
@@ -128,21 +111,15 @@ def memory_leak_detected(process_id, monitor)
 					true
 				end
 
-				# Run the memory monitor.
-				#
-				# @returns [Async::Task] The task that is running the memory monitor.
-				def run
-					Async do
-						Loop.run(interval: @interval) do
-							@guard.synchronize do
-								# This block must return true if the process was killed.
-								@cluster.check! do |process_id, monitor|
-									begin
-										memory_leak_detected(process_id, monitor)
-									rescue => error
-										Console.error(self, "Failed to handle memory leak!", child: {process_id: process_id}, exception: error)
-									end
-								end
+				# Run one iteration of the memory monitor.
+				def run_once
+					@guard.synchronize do
+						# This block must return true if the process was killed.
+						@cluster.check! do |process_id, monitor|
+							begin
+								memory_leak_detected(process_id, monitor)
+							rescue => error
+								Console.error(self, "Failed to handle memory leak!", child: {process_id: process_id}, exception: error)
 							end
 						end
 					end

lib/async/service/supervisor/memory_monitor.rb

@@ -0,0 +1,55 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async/loop"
+
+module Async
+	module Service
+		module Supervisor
+			class Monitor
+				def initialize(interval: 1.0)
+					@interval = interval
+				end
+				
+				def as_json(...)
+					{}
+				end
+				
+				# Serialize to JSON string.
+				def to_json(...)
+					as_json.to_json(...)
+				end
+				
+				# Get aggregated utilization status by service name.
+				#
+				# Reads utilization data from all registered workers and aggregates it
+				# by service name (from supervisor_controller.state[:name]).
+				#
+				# @returns [Hash] Hash with type and data keys.
+				def status
+					{type: self.class.name, data: as_json}
+				end
+				
+				# Run one iteration of the monitor.
+				def run_once
+					# This method can be overridden by subclasses to implement specific monitoring logic.
+				end
+				
+				# Run the utilization monitor.
+				#
+				# Periodically aggregates utilization data from all workers.
+				#
+				# @returns [Async::Task] The task that is running the utilization monitor.
+				def run(parent: Async::Task.current)
+					parent.async do
+						Loop.periodic(interval: @interval) do
+							self.run_once
+						end
+					end
+				end
+			end
+		end
+	end
+end