Use MessagePack for threaded execution channels

lib/protocol/http/executor.rb

@@ -5,6 +5,7 @@
 
 require_relative "executor/version"
 require_relative "executor/error"
+require_relative "executor/codec"
 require_relative "executor/channel"
 require_relative "executor/transport"
 require_relative "executor/request"

lib/protocol/http/executor/channel.rb

@@ -5,6 +5,8 @@
 
 require "socket"
 
+require_relative "codec"
+
 module Protocol
 	module HTTP
 		module Executor
@@ -17,8 +19,10 @@ class Channel
 				# Initialize a channel over the given IO object.
 				#
 				# @parameter io [IO] The connected stream.
-				def initialize(io)
+				# @parameter codec [Class] The message serialization codec.
+				def initialize(io, codec = Codec::MessagePack)
 					@io = io
+					@codec = codec.new
 					@write_mutex = Mutex.new
 					@read_closed = false
 					@write_closed = false
@@ -29,7 +33,7 @@ def initialize(io)
 				# @parameter type [Symbol] The message type.
 				# @parameter payload [Object] The message payload.
 				def write(type, payload = nil)
-					data = Marshal.dump([type, payload])
+					data = @codec.dump([type, payload])
 
 					if data.bytesize > MAXIMUM_FRAME_SIZE
 						raise ArgumentError, "Frame is too large: #{data.bytesize} bytes!"
@@ -65,7 +69,7 @@ def read
 						raise ClosedError, "Invalid frame size: #{length} bytes!"
 					end
 
-					return Marshal.load(read_exactly(length))
+					return @codec.load(read_exactly(length))
 				rescue EOFError, IOError, SystemCallError => error
 					raise ClosedError, error.message
 				end

lib/protocol/http/executor/codec.rb

@@ -0,0 +1,69 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "msgpack"
+require "socket"
+
+module Protocol
+	module HTTP
+		module Executor
+			# Serialization codecs for execution channel messages.
+			module Codec
+				# MessagePack serialization for thread execution.
+				class MessagePack
+					SYMBOL_TYPE = 0
+					ADDRESS_TYPE = 1
+
+					# Initialize a MessagePack factory with the protocol's extension types.
+					def initialize
+						@factory = ::MessagePack::Factory.new
+						@factory.register_type(SYMBOL_TYPE, Symbol, packer: ->(symbol){symbol.to_s}, unpacker: ->(string){string.to_sym})
+						@factory.register_type(
+							ADDRESS_TYPE,
+							Addrinfo,
+							packer: ->(address){::MessagePack.pack([address.to_sockaddr, address.pfamily, address.socktype, address.protocol])},
+							unpacker: ->(data){Addrinfo.new(*::MessagePack.unpack(data))},
+						)
+					end
+
+					# Serialize a protocol message.
+					#
+					# @parameter message [Object] The message to serialize.
+					# @returns [String] The serialized message.
+					def dump(message)
+						@factory.dump(message)
+					end
+
+					# Deserialize a protocol message.
+					#
+					# @parameter data [String] The serialized message.
+					# @returns [Object] The deserialized message.
+					def load(data)
+						@factory.load(data)
+					end
+				end
+
+				# Marshal serialization for Ractor execution, where MessagePack's native extension is not available.
+				class Marshal
+					# Serialize a protocol message.
+					#
+					# @parameter message [Object] The message to serialize.
+					# @returns [String] The serialized message.
+					def dump(message)
+						::Marshal.dump(message)
+					end
+
+					# Deserialize a protocol message.
+					#
+					# @parameter data [String] The serialized message.
+					# @returns [Object] The deserialized message.
+					def load(data)
+						::Marshal.load(data)
+					end
+				end
+			end
+		end
+	end
+end

lib/protocol/http/executor/generic.rb

@@ -11,14 +11,17 @@ module HTTP
 		module Executor
 			# The common middleware implementation for isolated execution contexts.
 			class Generic < ::Protocol::HTTP::Middleware
+				CODEC = Codec::MessagePack
+
 				# Execute a request using a new isolated execution context.
 				#
 				# @parameter request [Protocol::HTTP::Request] The request to execute.
 				# @returns [Protocol::HTTP::Response] The remote response.
 				def call(request)
 					parent = ::Async::Task.current
-					endpoint, worker_streams = Transport.pair
-					backend = spawn(worker_streams)
+					codec = self.class::CODEC
+					endpoint, worker_streams = Transport.pair(codec)
+					backend = spawn(worker_streams, codec)
 
 					Execution.new(endpoint, backend, request, parent).call
 				rescue
@@ -31,8 +34,9 @@ def call(request)
 				# Spawn the isolated execution context.
 				#
 				# @parameter worker_streams [Array(IO)] The worker transport streams.
+				# @parameter codec [Class] The message serialization codec.
 				# @returns [Thread | Ractor] The isolated execution context.
-				def spawn(worker_streams)
+				def spawn(worker_streams, codec)
 					raise NotImplementedError
 				end
 			end

lib/protocol/http/executor/ractored.rb

@@ -8,6 +8,8 @@ module HTTP
 		module Executor
 			# Executes each request in a dedicated Ruby 4.1 Ractor.
 			class Ractored < Generic
+				CODEC = Codec::Marshal
+
 				# Initialize a Ractored executor.
 				#
 				# @parameter delegate [Interface(:call)] A shareable HTTP application.
@@ -34,9 +36,11 @@ def self.supported?
 
 				private
 
-				def spawn(worker_streams)
+				def spawn(worker_streams, codec)
 					descriptors = worker_streams.map(&:fileno).freeze
-					worker = ::Ractor.new(@delegate, descriptors, name: "protocol-http-executor"){|application, descriptors| Protocol::HTTP::Executor::Worker.run(application, descriptors.map{|descriptor| ::Socket.for_fd(descriptor)})}
+					worker = ::Ractor.new(@delegate, descriptors, codec, name: "protocol-http-executor") do |application, descriptors, codec|
+						Protocol::HTTP::Executor::Worker.run(application, descriptors.map{|descriptor| ::Socket.for_fd(descriptor)}, codec)
+					end
 
 					# The worker now owns the descriptors. Close these wrappers without closing
 					# the underlying descriptors:

lib/protocol/http/executor/threaded.rb

@@ -10,11 +10,11 @@ module Executor
 			class Threaded < Generic
 				private
 
-				def spawn(worker_streams)
+				def spawn(worker_streams, codec)
 					application = @delegate
 
-					return Thread.new(application, worker_streams) do |delegate, streams|
-						Worker.run(delegate, streams)
+					return Thread.new(application, worker_streams, codec) do |delegate, streams, codec|
+						Worker.run(delegate, streams, codec)
 					end
 				end
 			end

lib/protocol/http/executor/transport.rb

@@ -16,9 +16,10 @@ class Endpoint
 					#
 					# @parameter control [IO] The bidirectional control stream.
 					# @parameter body [IO] The bidirectional request and response body stream.
-					def initialize(control, body)
-						@control = Channel.new(control)
-						@body = Channel.new(body)
+					# @parameter codec [Class] The message serialization codec.
+					def initialize(control, body, codec = Codec::MessagePack)
+						@control = Channel.new(control, codec)
+						@body = Channel.new(body, codec)
 					end
 
 					# @attribute [Channel] The bidirectional control channel.
@@ -36,12 +37,13 @@ def close
 
 				# Create connected client and worker transport endpoints.
 				#
+				# @parameter codec [Class] The message serialization codec.
 				# @returns [Array(Endpoint, Array(IO))] The client endpoint and worker IO objects.
-				def self.pair
+				def self.pair(codec = Codec::MessagePack)
 					client_control, worker_control = Socket.pair(:UNIX, :STREAM, 0)
 					client_body, worker_body = Socket.pair(:UNIX, :STREAM, 0)
 
-					client = Endpoint.new(client_control, client_body)
+					client = Endpoint.new(client_control, client_body, codec)
 					worker = [worker_control, worker_body]
 
 					return client, worker

lib/protocol/http/executor/worker.rb

@@ -17,8 +17,9 @@ module Worker
 				#
 				# @parameter application [Interface(:call)] The HTTP application.
 				# @parameter streams [Array(IO)] The control and bidirectional body streams.
-				def self.run(application, streams)
-					endpoint = Transport::Endpoint.new(*streams)
+				# @parameter codec [Class] The message serialization codec.
+				def self.run(application, streams, codec = Codec::MessagePack)
+					endpoint = Transport::Endpoint.new(*streams, codec)
 
 					Sync do
 						execute(application, endpoint)

protocol-http-executor.gemspec

@@ -24,5 +24,6 @@ Gem::Specification.new do |spec|
 	spec.required_ruby_version = ">= 3.3"
 
 	spec.add_dependency "async"
+	spec.add_dependency "msgpack", "~> 1.0"
 	spec.add_dependency "protocol-http", "~> 0.70"
 end

releases.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+  - Use MessagePack serialization for threaded execution channels, avoiding application use of `Marshal`.
   - Add a Rack example using Falcon's `config/serve.rb`, Protocol::Rack, and Rack's opt-in Ractor support.
   - Add a Falcon integration example for Shopify's Ractor-safe Writebook experiment.
 

test/protocol/http/executor/channel.rb

@@ -6,9 +6,21 @@
 require "protocol/http/executor"
 
 describe Protocol::HTTP::Executor::Channel do
-	def channel_pair
+	class OversizedCodec
+		def dump(message)
+			return Data.new
+		end
+
+		class Data
+			def bytesize
+				return Protocol::HTTP::Executor::Channel::MAXIMUM_FRAME_SIZE + 1
+			end
+		end
+	end
+
+	def channel_pair(codec = Protocol::HTTP::Executor::Codec::MessagePack)
 		left, right = Socket.pair(:UNIX, :STREAM, 0)
-		return subject.new(left), right
+		return subject.new(left, codec), right
 	end
 
 	it "reports whether both directions are closed" do
@@ -25,20 +37,33 @@ def channel_pair
 	end
 
 	it "rejects oversized outgoing frames" do
+		channel, peer = channel_pair(OversizedCodec)
+
+		expect do
+			channel.write(:oversized)
+		end.to raise_exception(ArgumentError, message: be =~ /too large/)
+	ensure
+		channel&.close
+		peer&.close
+	end
+
+	it "preserves protocol message types and values" do
 		channel, peer = channel_pair
-		data = Object.new
-		maximum_frame_size = subject::MAXIMUM_FRAME_SIZE
-		data.define_singleton_method(:bytesize){maximum_frame_size + 1}
+		remote = subject.new(peer)
+		message = [:response, {status: 200, peer: Addrinfo.tcp("127.0.0.1", 443), body: "\x00\xFF".b}]
 
-		mock(Marshal) do |marshal|
-			marshal.replace(:dump){data}
-
-			expect do
-				channel.write(:oversized)
-			end.to raise_exception(ArgumentError, message: be =~ /too large/)
-		end
+		channel.write(*message)
+
+		type, payload = remote.read
+		expect(type).to be == :response
+		expect(payload[:status]).to be == 200
+		expect(payload[:body]).to be == "\x00\xFF".b
+		expect(payload[:peer].to_sockaddr).to be == message.last[:peer].to_sockaddr
+		expect(payload[:peer].socktype).to be == message.last[:peer].socktype
+		expect(payload[:peer].protocol).to be == message.last[:peer].protocol
 	ensure
 		channel&.close
+		remote&.close
 		peer&.close
 	end
 

test/protocol/http/executor/generic.rb

@@ -10,7 +10,7 @@
 		executor = subject.new(Protocol::HTTP::Middleware::Okay)
 
 		expect do
-			executor.send(:spawn, [])
+			executor.send(:spawn, [], Protocol::HTTP::Executor::Codec::MessagePack)
 		end.to raise_exception(NotImplementedError)
 	end
 end