neiam
/apollos_crib
It's Apollo's Crib
public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages
⭐
Log in to mark this repository.
GitLab: each request has a connection of its own, so responses keep their ids
9b66bba · Gabe Morell · 2026-11-05 00:25
Message
{commit_body(@commit)}
Files changed
modified
apps/room_gitlab/lib/room_gitlab/gitlab_worker.ex
+17
−1
@@ -180,7 +180,12 @@ defmodule RoomGitlab.Worker do
| 180 | 180 | ["PRIVATE-TOKEN": "#{token}", Accept: "Application/json; Charset=utf-8"], |
| 181 | 181 | params: params, |
| 182 | 182 | follow_redirect: true, |
| 183 | − stream_to: self() | |
| 183 | + stream_to: self(), | |
| 184 | + # hackney 4 tags async messages with the connection's pid, and a pooled | |
| 185 | + # HTTP/2 connection carries every concurrent request to the host, so | |
| 186 | + # their responses would all arrive under one id. A connection of its | |
| 187 | + # own gives each request an id of its own. | |
| 188 | + hackney: [pool: false] | |
| 184 | 189 | ) |
| 185 | 190 | |
| 186 | 191 | # |> IO.inspect() |
@@ -341,6 +346,7 @@ defmodule RoomGitlab.Worker do
| 341 | 346 | |
| 342 | 347 | # IO.inspect({:handle_info, msg}) |
| 343 | 348 | def handle_info(%HTTPoison.Error{reason: {:closed, :timeout}, id: resp_id}, state) do |
| 349 | + close_conn(resp_id) | |
| 344 | 350 | {:noreply, state} |
| 345 | 351 | end |
| 346 | 352 |
@@ -369,6 +375,7 @@ defmodule RoomGitlab.Worker do
| 369 | 375 | end |
| 370 | 376 | |
| 371 | 377 | def handle_info(%HTTPoison.AsyncEnd{id: resp_id}, state) do |
| 378 | + close_conn(resp_id) | |
| 372 | 379 | {_, str} = StringIO.contents(state[:refs][resp_id][:fd]) |
| 373 | 380 | StringIO.close(state[:refs][resp_id][:fd]) |
| 374 | 381 | t = state[:refs][resp_id][:type] |
@@ -388,6 +395,15 @@ defmodule RoomGitlab.Worker do
| 388 | 395 | end |
| 389 | 396 | end |
| 390 | 397 | |
| 398 | + # Each request has a connection of its own, which nothing else will use once | |
| 399 | + # its response is over; left open, an HTTP/2 one would stay up for good. | |
| 400 | + defp close_conn(conn) do | |
| 401 | + :hackney.close(conn) | |
| 402 | + catch | |
| 403 | + # Already gone: an HTTP/1 one stops with the process HTTPoison streams through. | |
| 404 | + :exit, _ -> :ok | |
| 405 | + end | |
| 406 | + | |
| 391 | 407 | defp store_decoded(state, :projects, _id, decoded) do |
| 392 | 408 | filtered = filter_watched(decoded, state.inst) |
| 393 | 409 |
modified
apps/room_gitlab/mix.exs
+3
−0
@@ -31,6 +31,9 @@ defmodule RoomGitlab.MixProject do
| 31 | 31 | {:telemetry, "~> 1.0"}, #=> When using the Telemetry events (Nebulex stats) |
| 32 | 32 | {:httpoison, "~> 3.0", override: true}, |
| 33 | 33 | {:jason, "~> 1.4"}, |
| 34 | + # The worker's source config lives in RoomSanctum; its tests build one, | |
| 35 | + # and stand in for GitLab with a Plug.Cowboy server. | |
| 36 | + {:room_sanctum, in_umbrella: true, only: :test}, | |
| 34 | 37 | # {:dep_from_hexpm, "~> 0.3.0"}, |
| 35 | 38 | # {:dep_from_git, git: "https://github.com/elixir-lang/my_dep.git", tag: "0.1.0"}, |
| 36 | 39 | # {:sibling_app_in_umbrella, in_umbrella: true} |
added
apps/room_gitlab/test/gitlab_worker_test.exs
+113
−0
@@ -0,0 +1,113 @@
| 1 | +defmodule RoomGitlab.WorkerTest do | |
| 2 | + @moduledoc """ | |
| 3 | + The GitLab 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 files each body under whatever it recorded for that id. On | |
| 7 | + hackney 4 that id is the connection's pid, and a connection from hackney's | |
| 8 | + shared pool carries every concurrent request to the same HTTP/2 host, so all | |
| 9 | + of them arrive under one id. | |
| 10 | + """ | |
| 11 | + use ExUnit.Case, async: false | |
| 12 | + | |
| 13 | + alias RoomGitlab.Worker | |
| 14 | + alias RoomSanctum.Configuration.Configs.Gitlab | |
| 15 | + | |
| 16 | + defmodule FakeGitlab do | |
| 17 | + @moduledoc false | |
| 18 | + # Answers a project's jobs with one job that names the project, once the | |
| 19 | + # test lets the request through. | |
| 20 | + import Plug.Conn | |
| 21 | + | |
| 22 | + def init(opts), do: opts | |
| 23 | + | |
| 24 | + def call(%{path_info: ["api", "v4", "projects", id, "jobs"]} = 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!([%{"id" => id, "status" => "success"}])) | |
| 34 | + end | |
| 35 | + end | |
| 36 | + | |
| 37 | + setup do | |
| 38 | + start_supervised!( | |
| 39 | + {Plug.Cowboy, scheme: :http, plug: {FakeGitlab, test: self()}, options: [port: 0]} | |
| 40 | + ) | |
| 41 | + | |
| 42 | + port = :ranch.get_port(FakeGitlab.HTTP) | |
| 43 | + :ok = :hackney_pool.start_pool(:default, []) | |
| 44 | + | |
| 45 | + state = %{ | |
| 46 | + id: "1", | |
| 47 | + inst: %{ | |
| 48 | + id: 1, | |
| 49 | + enabled: true, | |
| 50 | + config: %Gitlab{url: "http://127.0.0.1:#{port}", pat: "token"} | |
| 51 | + }, | |
| 52 | + projects: [], | |
| 53 | + available_projects: [], | |
| 54 | + commits: %{}, | |
| 55 | + jobs: %{}, | |
| 56 | + refs: %{}, | |
| 57 | + refs_t: %{}, | |
| 58 | + refs_i: %{} | |
| 59 | + } | |
| 60 | + | |
| 61 | + %{state: state} | |
| 62 | + end | |
| 63 | + | |
| 64 | + test "job requests each open a connection of their own", %{state: state} do | |
| 65 | + {:noreply, state} = query_jobs(state, [11, 22]) | |
| 66 | + held = for _ <- 1..2, do: assert_receive({:holding, _pid}, 5_000) | |
| 67 | + | |
| 68 | + # Both requests are in flight, and neither came out of the shared pool. | |
| 69 | + assert :hackney_pool.get_stats(:default)[:in_use_count] == 0 | |
| 70 | + | |
| 71 | + for {:holding, pid} <- held, do: send(pid, :go) | |
| 72 | + responses = receive_responses(2) | |
| 73 | + state = Enum.reduce(responses, state, &handle_info/2) | |
| 74 | + | |
| 75 | + assert state.jobs == %{ | |
| 76 | + 11 => [%{"id" => "11", "status" => "success"}], | |
| 77 | + 22 => [%{"id" => "22", "status" => "success"}] | |
| 78 | + } | |
| 79 | + | |
| 80 | + # Nor do those connections outlast their responses. | |
| 81 | + for conn <- responses |> Enum.map(& &1.id) |> Enum.uniq() do | |
| 82 | + ref = Process.monitor(conn) | |
| 83 | + assert_receive {:DOWN, ^ref, :process, ^conn, _} | |
| 84 | + end | |
| 85 | + end | |
| 86 | + | |
| 87 | + defp query_jobs(state, project_ids) do | |
| 88 | + projects = for id <- project_ids, do: %{"id" => id} | |
| 89 | + Worker.handle_cast({:query_jobs, %{}}, %{state | projects: projects}) | |
| 90 | + end | |
| 91 | + | |
| 92 | + defp handle_info(msg, state) do | |
| 93 | + {:noreply, state} = Worker.handle_info(msg, state) | |
| 94 | + state | |
| 95 | + end | |
| 96 | + | |
| 97 | + # Every message up to the count-th request's end. | |
| 98 | + defp receive_responses(count, acc \\ []) | |
| 99 | + defp receive_responses(0, acc), do: Enum.reverse(acc) | |
| 100 | + | |
| 101 | + defp receive_responses(count, acc) do | |
| 102 | + receive do | |
| 103 | + %HTTPoison.AsyncEnd{} = msg -> | |
| 104 | + receive_responses(count - 1, [msg | acc]) | |
| 105 | + | |
| 106 | + %struct{} = msg | |
| 107 | + when struct in [HTTPoison.AsyncStatus, HTTPoison.AsyncHeaders, HTTPoison.AsyncChunk] -> | |
| 108 | + receive_responses(count, [msg | acc]) | |
| 109 | + after | |
| 110 | + 5_000 -> flunk("no complete response from the fake GitLab") | |
| 111 | + end | |
| 112 | + end | |
| 113 | +end |
Parents: 17c8ef1