Better observer state handling.

Author: samuel-williams-shopify

lib/async/utilization/metric.rb

@@ -14,13 +14,12 @@ class Metric
 			# Initialize a new metric.
 			#
 			# @parameter name [Symbol] The field name for this metric.
-			# @parameter registry [Registry] The registry instance to use.
-			def initialize(name, registry)
+			def initialize(name)
 				@name = name.to_sym
-				@registry = registry
 				@value = 0
-				@cache_valid = false
-				@cached_field_info = nil
+				
+				@observer = nil
+				@cached_field = nil
 				@cached_buffer = nil
 				@guard = Mutex.new
 			end
@@ -34,19 +33,43 @@ def initialize(name, registry)
 			# @attribute [Mutex] The mutex for thread safety.
 			attr :guard
 
-			# Invalidate the cached field information.
+			# Set the observer and rebuild cache.
 			#
-			# Called when the observer changes to force cache rebuild.
-			def invalidate
-				@cache_valid = false
-				@cached_field_info = nil
-				@cached_buffer = nil
+			# This is called when the registry assigns a new observer (or removes it).
+			# The cache is invalidated and then immediately recomputed so that the
+			# fast write path doesn't need to re-check the observer on the first write.
+			#
+			# @parameter observer [#set] The new observer (or nil).
+			def observer=(observer)
+				@guard.synchronize do
+					@observer = observer
+					
+					# Eagerly validate so the first write is fast.
+					outcome = :no_observer
+					if @observer
+						if field = @observer.schema[@name]
+							if buffer = @observer.buffer
+								@cached_field = field
+								@cached_buffer = buffer
+								outcome = :cached
+							else
+								outcome = :no_buffer
+							end
+						else
+							outcome = :missing_field_in_schema
+						end
+					else
+						outcome = :unsupported_observer
+					end
+					
+					# Console.info(self, "Cache validation", metric: @name, outcome: outcome)
+					
+					write_direct(@value)
+				end
 			end
 
 			# Increment the metric value.
 			#
-			# Uses the fast path (direct buffer write) when cache is valid and observer is available.
-			#
 			# @returns [Integer] The new value of the field.
 			def increment
 				@guard.synchronize do
@@ -103,28 +126,6 @@ def set(value)
 
 			protected
 
-			# Check if the cache is valid and rebuild if necessary.
-			#
-			# Always attempts to build the cache if it's invalid. Returns true if cache
-			# is now valid (observer exists, field is in schema, and buffer is available), false otherwise.
-			#
-			# @returns [bool] True if cache is valid, false otherwise.
-			def ensure_cache_valid!
-				unless @cache_valid
-					if observer = @registry.observer
-						if field = observer.schema[@name]
-							if buffer = observer.buffer
-								@cached_field_info = field
-								@cached_buffer = buffer
-							end
-						end
-					end
-					
-					# Once we've validated the cache, even if there was no observer or buffer, we mark it as valid, so that we don't try to revalidate it again:
-					@cache_valid = true
-				end
-			end
-			
 			# Write directly to the cached buffer if available.
 			#
 			# This is the fast path that avoids hash lookups. Always ensures cache is valid
@@ -133,16 +134,12 @@ def ensure_cache_valid!
 			# @parameter value [Numeric] The value to write.
 			# @returns [Boolean] Whether the write succeeded.
 			def write_direct(value)
-				self.ensure_cache_valid!
-				
 				if @cached_buffer
-					@cached_buffer.set_value(@cached_field_info.type, @cached_field_info.offset, value)
+					@cached_buffer.set_value(@cached_field.type, @cached_field.offset, value)
 				end
 
 				return true
 			rescue => error
-				# If write fails, log warning but don't invalidate cache
-				# The error might be transient, and invalidating would force hash lookups
 				Console.warn(self, "Failed to write metric value!", metric: {name: @name, value: value}, exception: error)
 
 				return false

lib/async/utilization/observer.rb

@@ -80,8 +80,9 @@ def initialize(schema, buffer)
 			# @parameter field [Symbol] The field name to set.
 			# @parameter value [Numeric] The value to set.
 			def set(field, value)
-				if field = @schema[field]
-					@buffer.set_value(field.type, field.offset, value)
+				if entry = @schema[field]
+					@buffer.set_value(entry.type, entry.offset, value)
+					# Console.info(self, "Wrote utilization metric", field: field, value: value, offset: entry.offset)
 				end
 			rescue => error
 				Console.warn(self, "Failed to set field in shared memory!", field: field, exception: error)

lib/async/utilization/registry.rb

@@ -3,6 +3,8 @@
 # Released under the MIT License.
 # Copyright, 2026, by Samuel Williams.
 
+require "console"
+
 module Async
 	module Utilization
 		# Registry for emitting utilization metrics.
@@ -68,19 +70,16 @@ def values
 			# @parameter observer [#set] The observer to set.
 			def observer=(observer)
 				@guard.synchronize do
-					# Invalidate all cached metrics
+					@observer = observer
+					
+					# Invalidate all cached metrics with new observer (or nil)
 					@metrics.each_value do |metric|
-						metric.invalidate
+						metric.observer = observer
 					end
 
-					@observer = observer
+					# Console.info(self, "Observer assigned", observer: observer, metric_count: @metrics.size)
 				end
 
-				# Notify observer of all current metric values (outside guard to avoid deadlock)
-				@metrics.each do |name, metric|
-					value = metric.guard.synchronize{metric.value}
-					observer.set(name, value)
-				end
 			end
 
 			# Set a field value.
@@ -135,7 +134,7 @@ def metric(field)
 				field = field.to_sym
 
 				@guard.synchronize do
-					@metrics[field] ||= Metric.new(field, self)
+					@metrics[field] ||= Metric.new(field)
 				end
 			end
 		end

test/async/utilization/metric.rb

@@ -154,9 +154,9 @@
 	end
 
 	it "writes directly to shared memory when observer is set" do
-		registry.observer = observer
 		metric = registry.metric(:total_requests)
-		
+		registry.observer = observer
+
 		metric.set(42)
 
 		# Read back from file to verify
@@ -216,33 +216,13 @@
 		expect(metric1).to be == metric2
 	end
 
-	it "falls back to observer.set when write_direct fails" do
-		registry.observer = observer
-		metric = registry.metric(:total_requests)
-		
-		# Force cache to be invalid by invalidating it
-		metric.invalidate
-		
-		# Set a value - should fall back to observer.set
-		metric.set(42)
-		expect(metric.value).to be == 42
-		
-		# Verify it was written to shared memory
-		buffer = IO::Buffer.map(File.open(shm_path, "r+b"), file_size, 0)
-		expect(buffer.get_value(:u64, 0)).to be == 42
-	end
-	
 	it "handles write errors gracefully" do
 		registry.observer = observer
 		metric = registry.metric(:total_requests)
 
 		# Set a value first to build the cache
 		metric.set(10)
-		
-		# Verify cache is built
-		expect(metric.instance_variable_get(:@cache_valid)).to be == true
-		cached_buffer = metric.instance_variable_get(:@cached_buffer)
-		
+
 		# Create an invalid buffer that will raise an error
 		invalid_buffer = Object.new
 		def invalid_buffer.set_value(type, offset, value)
@@ -251,13 +231,10 @@ def invalid_buffer.set_value(type, offset, value)
 
 		metric.instance_variable_set(:@cached_buffer, invalid_buffer)
 
-		# Should not raise, but log warning and keep cache valid
+		# Should not raise, but log warning
 		metric.set(42)
 		expect(metric.value).to be == 42
-		
-		# Cache should remain valid (not invalidated on error)
-		expect(metric.instance_variable_get(:@cache_valid)).to be == true
-		
+
 		# Assert that a warning was logged
 		expect_console.to have_logged(
 			severity: be == :warn,

test/async/utilization/registry.rb

@@ -56,44 +56,24 @@
 		expect(registry.values[:test_field]).to be == 0
 	end
 
-	it "can set an observer" do
-		observer = Object.new
-		def observer.set(field, value); end
-		
-		registry.set(:test_field, 10)
-		registry.observer = observer
-		
-		expect(registry.observer).to be == observer
-	end
-	
 	it "notifies observer when values change" do
-		values_set = []
-		
-		# Create a proper observer with schema
 		schema = Async::Utilization::Schema.build(test_field: :u64)
+		buffer = IO::Buffer.new(8)
+
 		observer = Object.new
-		
-		# Define methods on observer
-		observer.define_singleton_method(:set) do |field, value|
-			values_set << [field, value]
-		end
-		observer.define_singleton_method(:schema){schema}
-		observer.define_singleton_method(:buffer){nil}  # No buffer, so write_direct will return false
-		
+		observer.define_singleton_method(:schema) { schema }
+		observer.define_singleton_method(:buffer) { buffer }
+
 		registry.set(:test_field, 5)
 		registry.observer = observer
-		
-		# Observer should be notified of existing values
-		expect(values_set).to be(:include?, [:test_field, 5])
-		
-		# Clear and test new changes
-		# Note: Since observer has no buffer, write_direct will return false
-		# and no notification will occur (as per new design)
-		values_set.clear
+
+		# Buffer should be synced with the existing value on observer assignment
+		expect(buffer.get_value(:u64, 0)).to be == 5
+
 		registry.increment(:test_field)
-		
-		# With no buffer, write_direct fails silently, so no notification
-		expect(values_set).to be == []
+
+		# Buffer should reflect the incremented value
+		expect(buffer.get_value(:u64, 0)).to be == 6
 	end
 
 	it "uses metric method for fast path" do