Fix handling of timeouts/deadlines.

Author: ioquatix

fixtures/async/grpc/test_interface.rb

@@ -22,6 +22,7 @@ class TestInterface < Protocol::GRPC::Interface
 					response_class: Protocol::GRPC::Fixtures::TestMessage, streaming: :bidirectional
 				rpc :ClientStreamingCall, request_class: Protocol::GRPC::Fixtures::TestMessage,
 					response_class: Protocol::GRPC::Fixtures::TestMessage, streaming: :client_streaming
+				rpc :SlowCall
 			end
 
 			# Test service implementation
@@ -53,7 +54,6 @@ def bidirectional_call(input, output, _call)
 						response = Protocol::GRPC::Fixtures::TestMessage.new(value: "Echo: #{request.value}")
 						output.write(response)
 					end
-					puts "Closing write"
 					output.close_write
 				end
 
@@ -66,6 +66,13 @@ def client_streaming_call(input, output, _call)
 					response = Protocol::GRPC::Fixtures::TestMessage.new(value: "Received: #{values.join(', ')}")
 					output.write(response)
 				end
+				
+				def slow_call(input, output, _call)
+					request = input.read
+					sleep 1 # Simulate a slow operation
+					response = Protocol::GRPC::Fixtures::TestMessage.new(value: "Slow response: #{request.value}")
+					output.write(response)
+				end
 			end
 		end
 	end

lib/async/grpc/dispatcher.rb

@@ -56,24 +56,31 @@ def invoke_service(service, handler_method, input, output, call)
 				end
 
 				# Mark trailers and add status (if not already set by handler):
-				if call.response&.headers
-					call.response.headers.trailer!
-					
+				if headers = call.response&.headers
 					# Only add OK status if grpc-status hasn't been set by the handler:
-					unless call.response.headers["grpc-status"]
-						Protocol::GRPC::Metadata.add_status!(call.response.headers, status: Protocol::GRPC::Status::OK)
+					unless headers["grpc-status"]
+						Protocol::GRPC::Metadata.add_status!(headers, status: Protocol::GRPC::Status::OK)
 					end
 				end
 			end
 
 			def dispatch_to_service(service, handler_method, input, output, call, deadline, parent: Async::Task.current)
 				if deadline
-					parent.with_timeout(deadline) do
+					parent.with_timeout(deadline.remaining) do
 						invoke_service(service, handler_method, input, output, call)
 					end
 				else
 					invoke_service(service, handler_method, input, output, call)
 				end
+			rescue Async::TimeoutError
+				# Close input and output streams:
+				input.close
+				output.close_write unless output.closed?
+				
+				# Set DEADLINE_EXCEEDED status in trailers:
+				if headers = call.response&.headers
+					Protocol::GRPC::Metadata.add_status!(headers, status: Protocol::GRPC::Status::DEADLINE_EXCEEDED, message: "Deadline exceeded")
+				end
 			end
 
 			# Dispatch the request to the appropriate service.

releases.md

@@ -1,5 +1,9 @@
 # Releases
 
+## Unreleased
+
+  - Fix handling of timeouts/deadlines.
+
 ## v0.4.0
 
   - Fix handling of trailers.

test/async/grpc/dispatcher.rb

@@ -3,6 +3,8 @@
 # Released under the MIT License.
 # Copyright, 2025-2026, by Samuel Williams.
 
+require "sus/fixtures/async/scheduler_context"
+
 require "async/grpc/dispatcher"
 require "async/grpc/service"
 require "protocol/http"
@@ -102,5 +104,26 @@
 
 			expect(response.status).to be == 404
 		end
+		
+		it "handles timeout correctly" do
+			path = Protocol::GRPC::Methods.build_path(service_name, "SlowCall")
+			request = Protocol::HTTP::Request.new("http", "localhost", "POST", path, nil, headers, request_body)
+			request.headers["grpc-timeout"] = "100m" # 100 milliseconds
+			
+			response = dispatcher.call(request)
+			
+			expect(response.status).to be == 200
+			
+			# The response body should be consumed to access trailers:
+			response_body = Protocol::GRPC::Body::ReadableBody.wrap(response, message_class: Protocol::GRPC::Fixtures::TestMessage)
+			response_body.finish
+			
+			# Check that grpc-status is DEADLINE_EXCEEDED (4):
+			status = Protocol::GRPC::Metadata.extract_status(response.headers)
+			expect(status).to be == Protocol::GRPC::Status::DEADLINE_EXCEEDED
+			
+			message = Protocol::GRPC::Metadata.extract_message(response.headers)
+			expect(message).to be == "Deadline exceeded"
+		end
 	end
 end