Skip to content

Commit a56336e

Browse files
test: a cancel racing the exchange's end, a forged body, the watcher kind
Review round 2's R2-1, R2-2 and R2-3. R2-1: cancellation_test.rb gains the deterministic race -- an ordinary caller's hook registered before the adapter's parks the canceller between Source#cancel's flag flip and the adapter's hook, a fake client answers a 204 once the flag is up, check-after-resume settles the pivot cancelled and the exchange closes its queue, and only then does the adapter's hook run. Source#cancel must return true on the canceller's thread; against the pre-fix push it returned ClosedQueueError, and the canceller is unparked from an ensure so no assertion path leaves the thread parked. R2-2: request_mapper_test.rb's forged request takes method: and body:, and the body-forbidden clause is asserted over a forged GET and HEAD carrying a body -- the only shape that reaches the guard, since HTTP-7 makes a built GET's body nil -- with a forged POST as the control that shows the fixture carries its body through. Removing the guard now fails by name where it survived every suite. R2-3: the ASYNC-20 case reads the delivered response's watcher while the body is open -- one per response, transient -- and asserts it after the body's release, so a watcher spawned without transient: true fails in milliseconds by name instead of holding the reactor open until the run is killed. watcher_tasks joins the reactor helper beside assert_exchange_released, which now reads through it.
1 parent e1f65b0 commit a56336e

3 files changed

Lines changed: 111 additions & 18 deletions

File tree

‎gems/dexpace-transport-async_http/test/dexpace/transport/async_http/cancellation_test.rb‎

Lines changed: 68 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,59 @@ class DexpaceTransportAsyncHTTPCancellationTest < DexpaceTestCase
112112
adapter&.close
113113
server&.close
114114
end
115+
116+
# Review round 2's R2-1. `Cancellation::Source#cancel` steals its hooks under its mutex, flips
117+
# the flag and runs them OUTSIDE it, on the canceller's thread -- so the exchange can end
118+
# between the flip and the adapter's hook: check-after-resume sees the flag, the pivot
119+
# settles cancelled and the exchange closes its queue before the hook pushes onto it, and a
120+
# push onto a closed queue raises `ClosedQueueError`, which `Hooks.notify` would hand back to
121+
# the caller's `Source#cancel`. Deterministic through an ordinary caller's hook registered
122+
# FIRST, which parks the canceller across exactly that window; the same ordering needs no
123+
# slow hook when the canceller is merely descheduled there. A cancel that lost the race
124+
# against the exchange's own end is not the caller's failure.
125+
test "a token cancel in flight while the exchange finishes never raises out of " \
126+
"Source#cancel on the canceller's thread, and the future is cancelled" do
127+
source = Dexpace::Cancellation.source
128+
flagged = ::Thread::Queue.new # the canceller has flipped the flag and is in its hooks
129+
finished = ::Thread::Queue.new # the exchange has ended: let the adapter's hook run now
130+
outcome = ::Thread::Queue.new
131+
source.token.on_cancel do |_reason|
132+
flagged.push(true)
133+
finished.pop
134+
end
135+
client = Object.new
136+
client.define_singleton_method(:retries) { 0 }
137+
client.define_singleton_method(:pool) { Object.new.tap { |pool| def pool.close = nil } }
138+
client.define_singleton_method(:call) do |_native|
139+
flagged.pop # scheduler-aware: the reactor keeps turning until the cancel is in flight
140+
::Protocol::HTTP::Response.new("HTTP/1.1", 204, ::Protocol::HTTP::Headers.new, nil)
141+
end
142+
adapter = AsyncHTTP.using(client)
143+
canceller = nil
144+
145+
Sync do |task|
146+
future = adapter.call(request("http://example.test/"), nil, source.token)
147+
canceller = ::Thread.new do
148+
outcome.push(begin
149+
source.cancel(:racing)
150+
rescue ::StandardError => error
151+
error
152+
end)
153+
end
154+
155+
error = assert_raises(Dexpace::CancelledError) { value_within(future) }
156+
157+
assert_equal(:racing, error.reason)
158+
assert_predicate(future, :cancelled?)
159+
assert_exchange_released(task) # the exchange ended and closed its queue...
160+
ensure
161+
finished.push(true) # ...and only now does the adapter's hook run (on every path)
162+
end
163+
canceller.join
164+
result = outcome.pop
165+
166+
assert_same(true, result, "Source#cancel raised #{result.inspect} out of the adapter's hook")
167+
end
115168
end
116169

117170
# A cancel racing the delivery (TRANSPORT-9), landing after it (ASYNC-20), or reaching a blocked
@@ -146,12 +199,18 @@ class DexpaceTransportAsyncHTTPCancellationDeliveryTest < DexpaceTestCase
146199
end
147200

148201
# The negative twin (ASYNC-20): a response already delivered to the caller must not be closed
149-
# by a late cancellation of the future -- a cancel on a settled pivot is a no-op.
150-
test "ASYNC-20: cancelling the future after delivery does not close the delivered response" do
202+
# by a late cancellation of the future -- a cancel on a settled pivot is a no-op. The watcher
203+
# that would close it on a TOKEN cancel stays for the life of the delivered response, and
204+
# transient: a body a caller never closes must not hold the caller's `Sync` block open --
205+
# without `transient: true` that property fails as a hang and never by name (review round
206+
# 2's R2-3), so it is asserted here before the body is released and its release is asserted
207+
# to end the watcher.
208+
test "ASYNC-20: cancelling the future after delivery does not close the delivered response; " \
209+
"its watcher stays, transient, until the body is released" do
151210
server = AsyncHTTPHoldingServer.new(hold: :head)
152211
adapter = AsyncHTTP.build
153212

154-
reactor_over(server) do
213+
reactor_over(server) do |task|
155214
future = adapter.call(request("http://127.0.0.1:#{server.port}/"), nil, nil)
156215
server.wait_for_accept
157216
server.release("ok")
@@ -161,7 +220,13 @@ class DexpaceTransportAsyncHTTPCancellationDeliveryTest < DexpaceTestCase
161220

162221
refute_predicate(response.body, :closed?)
163222
refute_predicate(future, :cancelled?)
223+
# Read while the body is open and asserted only after its release: a non-transient
224+
# watcher under a still-open body would hold the reactor open on the failing assertion.
225+
transient = watcher_tasks(task).map(&:transient?)
226+
164227
assert_equal("ok", response.body_string)
228+
assert_equal([true], transient, "one watcher per delivered response, and transient")
229+
assert_exchange_released(task) # the body's release is what ends the watcher
165230
end
166231
ensure
167232
adapter&.close

‎gems/dexpace-transport-async_http/test/dexpace/transport/async_http/request_mapper_test.rb‎

Lines changed: 25 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -44,13 +44,14 @@ def map(request, logger: Dexpace::Instrumentation::Logger::NULL, policy: DropPol
4444
def names(native) = native.headers.to_a.map { |name, _| name.downcase }
4545

4646
# A request-shaped object answering the four readers the seam contract types nothing about,
47-
# carrying headers no Dexpace validation has seen -- design §10.10's admitted hole.
47+
# carrying headers no Dexpace validation has seen -- design §10.10's admitted hole -- and,
48+
# with `body:`, a body on a method the builder's HTTP-7 would have refused it on.
4849
def forged_request(name, value)
4950
forged_request_with([name, value])
5051
end
5152

52-
def forged_request_with(*pairs)
53-
template = request
53+
def forged_request_with(*pairs, method: "GET", body: nil)
54+
template = request(method: method)
5455
headers = Object.new
5556
headers.define_singleton_method(:each_entry) do |&block|
5657
pairs.each do |pair|
@@ -61,7 +62,7 @@ def forged_request_with(*pairs)
6162
forged.define_singleton_method(:method) { template.method }
6263
forged.define_singleton_method(:url) { template.url }
6364
forged.define_singleton_method(:headers) { headers }
64-
forged.define_singleton_method(:body) { nil }
65+
forged.define_singleton_method(:body) { body }
6566
forged
6667
end
6768
end
@@ -169,12 +170,30 @@ class MappingTest < DexpaceTestCase
169170
refute_includes(names(native), "content-type")
170171
end
171172

172-
test "a body-less request maps to a nil native body, and a body-forbidden method attaches " \
173-
"none" do
173+
test "a body-less request maps to a nil native body" do
174174
assert_nil(map(request(method: "POST")).body)
175175
assert_nil(map(request(method: "GET")).body)
176176
end
177177

178+
# Dispatch step 8: a body-forbidden method gets no body attached. Through the model the guard
179+
# is unreachable -- HTTP-7 already makes a GET's body nil at the builder -- so it is asserted
180+
# over the forged shape that met no builder (design §10.10's hole), where a body on a GET or
181+
# a HEAD would otherwise go out chunked; the forged POST is the control that shows the
182+
# fixture carries its body through (review round 2's R2-2).
183+
test "a body-forbidden method attaches none, even on a forged request carrying one" do
184+
smuggled = Dexpace::Body.bytes("smuggled".b)
185+
186+
%w[GET HEAD].each do |method|
187+
forged = forged_request_with(%w[X-Ok 1], method: method, body: smuggled)
188+
189+
assert_predicate(forged.method, :body_forbidden?, method)
190+
assert_nil(map(forged).body, method)
191+
end
192+
control = map(forged_request_with(%w[X-Ok 1], method: "POST", body: smuggled))
193+
194+
assert_kind_of(RequestBody, control.body)
195+
end
196+
178197
test "a body maps to a RequestBody the library pulls; framing is never copied from a header" do
179198
native = map(request(headers: { "Content-Length" => "999" },
180199
body: Dexpace::Body.bytes("abc".b), method: "POST",))

‎gems/dexpace-transport-async_http/test/support/async_http_reactor.rb‎

Lines changed: 18 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -28,20 +28,29 @@ def reactor_over(server)
2828
end
2929
end
3030

31-
# After a settlement that delivered no response, the exchange task and its transient watcher
32-
# are both gone from the reactor: a turn or two for them to finish, then no non-transient child
33-
# remains and no child is a watcher. The pool's own transient gardener is the one child that
34-
# legitimately stays for the client's life, which is why the watcher is found by its
35-
# annotation rather than by counting. The watcher's release is what keeps a long-lived reactor
36-
# from accumulating one parked task per failed exchange, and this is what turns its absence red.
31+
# The watcher tasks alive under `task`, found by the adapter's own annotation: one per exchange
32+
# whose queue is still open -- in flight, or delivered with its body not yet released.
3733
#
3834
# @param task [Async::Task] the task the exchange was spawned under
39-
def assert_exchange_released(task)
35+
# @return [Array<Async::Task>]
36+
def watcher_tasks(task)
4037
watcher = Dexpace::Transport::AsyncHTTP.const_get(:Exchange, false)::WATCHER_ANNOTATION
38+
Array(task.children).select { |child| child.annotation == watcher }
39+
end
40+
41+
# After a settlement that delivered no response, or after a delivered body's release, the
42+
# exchange task and its transient watcher are both gone from the reactor: a turn or two for
43+
# them to finish, then no non-transient child remains and no child is a watcher. The pool's own
44+
# transient gardener is the one child that legitimately stays for the client's life, which is
45+
# why the watcher is found by its annotation rather than by counting. The watcher's release is
46+
# what keeps a long-lived reactor from accumulating one parked task per failed exchange, and
47+
# this is what turns its absence red.
48+
#
49+
# @param task [Async::Task] the task the exchange was spawned under
50+
def assert_exchange_released(task)
4151
RELEASE_TURNS.times do
4252
task.yield
43-
children = Array(task.children)
44-
return if children.all?(&:transient?) && children.none? { |c| c.annotation == watcher }
53+
return if Array(task.children).all?(&:transient?) && watcher_tasks(task).empty?
4554
end
4655

4756
flunk("the exchange task or its watcher outlived the settlement by #{RELEASE_TURNS} turns")

0 commit comments

Comments
 (0)