Introduce Loop abstraction.

Author: ioquatix

lib/async/container/supervisor/loop.rb

@@ -0,0 +1,77 @@
+module Async
+	module Container
+		module Supervisor
+			# A robust loop that executes a block at regular intervals.
+			#
+			# If an error occurs during the execution of the block, it is logged and the loop continues
+			# (if `robust` is true). Otherwise, the error is raised and the loop stops.
+			#
+			# @parameter interval [Integer] The interval in seconds between executions of the block.
+			# @parameter robust [Boolean] Whether to continue the loop on errors.
+			class Loop
+				def self.run(...)
+					self.new.tap do |loop|
+						loop.run
+					end
+				end
+				
+				def initialize(interval: 60, robust: true, skip: true, &block)
+					@interval = interval
+					@robust = robust
+					@skip = skip
+					@block = block
+				end
+				
+				def running?
+					@running
+				end
+				
+				# Stop on the next iteration of the loop.
+				def stop!
+					@running = false
+				end
+				
+				def run
+					if @running
+						raise RuntimeError, "Loop is already running!"
+					end
+					
+					@running = true
+					while @running
+						clock = Async::Clock.start!
+						
+						begin
+							@block.call(self)
+						rescue => error
+							if @robust
+								Console.error(self, "Loop error:", error)
+							else
+								raise error
+							end
+						end
+						
+						break unless @running
+						
+						remaining = @interval - clock.elapsed
+						
+						if remaining <= 0
+							Console.warn(self, "Loop is behind schedule by #{-remaining.round(2)} seconds.", interval: @interval)
+							
+							# We are behind schedule, so we can catch up:
+							if @skip
+								# Remaining will now be positive
+								remaining = remaining % @interval
+							end
+						end
+						
+						if remaining.positive?
+							sleep(remaining)
+						end
+					end
+				ensure
+					@running = false
+				end
+			end
+		end
+	end
+end

lib/async/container/supervisor/memory_monitor.rb

@@ -6,6 +6,8 @@
 require "memory/leak/cluster"
 require "set"
 
+require_relative "loop"
+
 module Async
 	module Container
 		module Supervisor
@@ -103,8 +105,12 @@ def memory_leak_detected(process_id, monitor)
 					end
 
 					# Kill the process gently:
-					Console.info(self, "Killing process!", child: {process_id: process_id})
-					Process.kill(:INT, process_id)
+					begin
+						Console.info(self, "Killing process!", child: {process_id: process_id})
+						Process.kill(:INT, process_id)
+					rescue => error
+						Console.warn(self, "Failed to kill process!", child: {process_id: process_id}, exception: error)
+					end
 
 					true
 				end
@@ -114,7 +120,7 @@ def memory_leak_detected(process_id, monitor)
 				# @returns [Async::Task] The task that is running the memory monitor.
 				def run
 					Async do
-						while true
+						Loop.run(interval: @interval) do
 							# This block must return true if the process was killed.
 							@cluster.check! do |process_id, monitor|
 								Console.error(self, "Memory leak detected!", child: {process_id: process_id}, monitor: monitor)
@@ -125,8 +131,6 @@ def run
 									Console.error(self, "Failed to handle memory leak!", child: {process_id: process_id}, exception: error)
 								end
 							end
-							
-							sleep(@interval)
 						end
 					end
 				end

lib/async/container/supervisor/process_monitor.rb

@@ -4,6 +4,7 @@
 # Copyright, 2025, by Samuel Williams.
 
 require "process/metrics"
+require_relative "loop"
 
 module Async
 	module Container
@@ -70,15 +71,13 @@ def status(call)
 				# @returns [Async::Task] The task that is running the process monitor.
 				def run
 					Async do
-						while true
+						Loop.run(interval: @interval) do
 							metrics = self.metrics
 
 							# Log each process individually for better searchability in log platforms:
 							metrics.each do |process_id, general|
 								Console.info(self, "Process metrics captured.", general: general)
 							end
-							
-							sleep(@interval)
 						end
 					end
 				end