neiam /apollos_crib

It's Apollo's Crib

public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages

GitHub: each request has a connection of its own, so responses keep their ids

95bda40 · Gabe Morell · 2026-11-05 00:30

9 files +141 -4

Message

{commit_body(@commit)}

Files changed

modified apps/room_github/lib/room_github/worker.ex

+28 −3

@@ -231,8 +231,17 @@ defmodule RoomGithub.Worker do

231	231	end
232	232
233	233	# HTTPoison async stream handlers
234		− def handle_info(%HTTPoison.Error{reason: {:closed, :timeout}}, state) do
235		− {:noreply, state}
	234	+ def handle_info(%HTTPoison.Error{id: rid}, state) do
	235	+ close_conn(rid)
	236	+
	237	+ case Map.pop(state.refs, rid) do
	238	+ {nil, _refs} ->
	239	+ {:noreply, state}
	240	+
	241	+ {ref, refs} ->
	242	+ close_buffer(ref.fd)
	243	+ {:noreply, %{state \| refs: refs}}
	244	+ end
236	245	end
237	246
238	247	def handle_info(%HTTPoison.AsyncStatus{code: 200, id: _id}, state) do

@@ -269,6 +278,8 @@ defmodule RoomGithub.Worker do

269	278	end
270	279
271	280	def handle_info(%HTTPoison.AsyncEnd{id: rid}, state) do
	281	+ close_conn(rid)
	282	+
272	283	case state.refs[rid] do
273	284	nil ->
274	285	{:noreply, state}

@@ -302,6 +313,15 @@ defmodule RoomGithub.Worker do

302	313
303	314	defp close_buffer(fd), do: StringIO.close(fd)
304	315
	316	+ # Each request has a connection of its own, which nothing else will use once
	317	+ # its response is over; left open, an HTTP/2 one would stay up for good.
	318	+ defp close_conn(conn) do
	319	+ :hackney.close(conn)
	320	+ catch
	321	+ # Already gone: an HTTP/1 one stops with the process HTTPoison streams through.
	322	+ :exit, _ -> :ok
	323	+ end
	324	+
305	325	@max_pages 20
306	326
307	327	defp do_github_req(api_url, pat, path, params) do

@@ -320,7 +340,12 @@ defmodule RoomGithub.Worker do

320	340
321	341	case HTTPoison.get(url, headers,
322	342	follow_redirect: true,
323		− stream_to: self()
	343	+ stream_to: self(),
	344	+ # hackney 4 tags async messages with the connection's pid, and a
	345	+ # pooled HTTP/2 connection carries every concurrent request to the
	346	+ # host, so their responses would all arrive under one id. A
	347	+ # connection of its own gives each request an id of its own.
	348	+ hackney: [pool: false]
324	349	) do
325	350	{:ok, ref} ->
326	351	{:ok, ref}

modified apps/room_github/mix.exs

+4 −1

@@ -28,7 +28,10 @@ defmodule RoomGithub.MixProject do

28	28	{:decorator, "~> 1.4"},
29	29	{:telemetry, "~> 1.0"},
30	30	{:httpoison, "~> 3.0", override: true},
31		− {:jason, "~> 1.4"}
	31	+ {:jason, "~> 1.4"},
	32	+ # The worker's source config lives in RoomSanctum; its tests build one,
	33	+ # and stand in for GitHub with a Plug.Cowboy server.
	34	+ {:room_sanctum, in_umbrella: true, only: :test}
32	35	]
33	36	end
34	37	end

added apps/room_github/test/test_helper.exs

+1 −0

@@ -0,0 +1 @@

+ExUnit.start()

added apps/room_github/test/worker_test.exs

+108 −0

@@ -0,0 +1,108 @@

	1	+defmodule RoomGithub.WorkerTest do
	2	+ @moduledoc """
	3	+ The GitHub poll, against a stand-in for the API.
	4	+
	5	+ Responses come back as HTTPoison async messages tagged with the request's id,
	6	+ and the worker buffers each body under that id. On hackney 4 the id is the
	7	+ connection's pid, and a connection from hackney's shared pool carries every
	8	+ concurrent request to the same HTTP/2 host, so all of them arrive under one
	9	+ id.
	10	+ """
	11	+ use ExUnit.Case, async: false
	12	+
	13	+ alias RoomGithub.Worker
	14	+ alias RoomSanctum.Configuration.Configs.Github
	15	+
	16	+ defmodule FakeGithub do
	17	+ @moduledoc false
	18	+ # Answers a repo's workflow runs with one run, once the test lets the
	19	+ # request through.
	20	+ import Plug.Conn
	21	+
	22	+ def init(opts), do: opts
	23	+
	24	+ def call(%{path_info: ["repos", _owner, repo, "actions", "runs"]} = conn, opts) do
	25	+ send(opts[:test], {:holding, self()})
	26	+
	27	+ receive do
	28	+ :go -> :ok
	29	+ end
	30	+
	31	+ conn
	32	+ \|> put_resp_content_type("application/json")
	33	+ \|> send_resp(200, Jason.encode!(%{"workflow_runs" => [%{"id" => repo}]}))
	34	+ end
	35	+ end
	36	+
	37	+ setup do
	38	+ start_supervised!(
	39	+ {Plug.Cowboy, scheme: :http, plug: {FakeGithub, test: self()}, options: [port: 0]}
	40	+ )
	41	+
	42	+ port = :ranch.get_port(FakeGithub.HTTP)
	43	+ :ok = :hackney_pool.start_pool(:default, [])
	44	+
	45	+ state = %{
	46	+ id: "1",
	47	+ inst: %{
	48	+ id: 1,
	49	+ enabled: true,
	50	+ config: %Github{api_url: "http://127.0.0.1:#{port}", pat: "token"}
	51	+ },
	52	+ repos: [%{"full_name" => "octo/a"}, %{"full_name" => "octo/b"}],
	53	+ available_repos: [],
	54	+ runs: %{},
	55	+ jobs: %{},
	56	+ refs: %{}
	57	+ }
	58	+
	59	+ %{state: state}
	60	+ end
	61	+
	62	+ test "run requests each open a connection of their own", %{state: state} do
	63	+ {:noreply, state} = Worker.handle_cast(:query_runs, state)
	64	+ held = for _ <- 1..2, do: assert_receive({:holding, _pid}, 5_000)
	65	+
	66	+ # Both requests are in flight, and neither came out of the shared pool.
	67	+ assert :hackney_pool.get_stats(:default)[:in_use_count] == 0
	68	+
	69	+ for {:holding, pid} <- held, do: send(pid, :go)
	70	+ responses = receive_responses(2)
	71	+
	72	+ state =
	73	+ Enum.reduce(responses, state, fn msg, state ->
	74	+ {:noreply, state} = Worker.handle_info(msg, state)
	75	+ state
	76	+ end)
	77	+
	78	+ assert state.runs == %{
	79	+ "octo/a" => [%{"id" => "a", "repository" => %{"full_name" => "octo/a"}}],
	80	+ "octo/b" => [%{"id" => "b", "repository" => %{"full_name" => "octo/b"}}]
	81	+ }
	82	+
	83	+ assert state.refs == %{}
	84	+
	85	+ # Nor do those connections outlast their responses.
	86	+ for conn <- responses \|> Enum.map(& &1.id) \|> Enum.uniq() do
	87	+ ref = Process.monitor(conn)
	88	+ assert_receive {:DOWN, ^ref, :process, ^conn, _}
	89	+ end
	90	+ end
	91	+
	92	+ # Every message up to the count-th request's end.
	93	+ defp receive_responses(count, acc \\ [])
	94	+ defp receive_responses(0, acc), do: Enum.reverse(acc)
	95	+
	96	+ defp receive_responses(count, acc) do
	97	+ receive do
	98	+ %HTTPoison.AsyncEnd{} = msg ->
	99	+ receive_responses(count - 1, [msg \| acc])
	100	+
	101	+ %struct{} = msg
	102	+ when struct in [HTTPoison.AsyncStatus, HTTPoison.AsyncHeaders, HTTPoison.AsyncChunk] ->
	103	+ receive_responses(count, [msg \| acc])
	104	+ after
	105	+ 5_000 -> flunk("no complete response from the fake GitHub")
	106	+ end
	107	+ end
	108	+end

Parents: 2d2cb58