GITGUD

neiam 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: what arrives for a request after its end is dropped

98f9c28 · Gabe Morell · 2026-11-05 00:27

7 files +67 -29
Message 
{commit_body(@commit)}

Files changed

modified apps/room_gitlab/lib/room_gitlab/gitlab_worker.ex
+40 −29
@@ -344,54 +344,65 @@ defmodule RoomGitlab.Worker do
344 344 {:reply, [:ok], state}
345 345 end
346 346
347 − # IO.inspect({:handle_info, msg})
348 − def handle_info(%HTTPoison.Error{reason: {:closed, :timeout}, id: resp_id}, state) do
347 + # A request's buffer goes from state.refs when its response ends or fails,
348 + # so anything that still arrives for it afterwards is dropped.
349 + def handle_info(%HTTPoison.Error{id: resp_id}, state) do
349 350 close_conn(resp_id)
350 − {:noreply, state}
351 +
352 + case Map.pop(state.refs, resp_id) do
353 + {nil, _refs} ->
354 + {:noreply, state}
355 +
356 + {ref, refs} ->
357 + StringIO.close(ref.fd)
358 + {:noreply, %{state | refs: refs}}
359 + end
351 360 end
352 361
353 − def handle_info(%HTTPoison.AsyncStatus{code: status_code, id: resp_id}, state) do
354 − # IO.inspect(status_code)
362 + def handle_info(%HTTPoison.AsyncStatus{code: 200}, state), do: {:noreply, state}
355 363
356 − if status_code == 200 do
357 − {:noreply, state}
358 − else
359 − refs = state[:refs]
360 − r = state[:refs][resp_id]
361 − r = r |> Map.put(:write, false)
362 − refs = refs |> Map.put(resp_id, r)
363 − {:noreply, state |> Map.put(:refs, refs)}
364 + def handle_info(%HTTPoison.AsyncStatus{id: resp_id}, state) do
365 + case state.refs do
366 + %{^resp_id => ref} ->
367 + {:noreply, %{state | refs: Map.put(state.refs, resp_id, Map.put(ref, :write, false))}}
368 +
369 + _ ->
370 + {:noreply, state}
364 371 end
365 372 end
366 373
367 − def handle_info(%HTTPoison.AsyncHeaders{headers: headers, id: resp_id}, state) do
368 − # IO.inspect(headers)
374 + def handle_info(%HTTPoison.AsyncHeaders{}, state) do
369 375 {:noreply, state}
370 376 end
371 377
372 378 def handle_info(%HTTPoison.AsyncChunk{chunk: chunk, id: resp_id}, state) do
373 − IO.binwrite(state[:refs][resp_id][:fd], chunk)
379 + case state.refs do
380 + %{^resp_id => %{fd: fd}} -> IO.binwrite(fd, chunk)
381 + _ -> :ok
382 + end
383 +
374 384 {:noreply, state}
375 385 end
376 386
377 387 def handle_info(%HTTPoison.AsyncEnd{id: resp_id}, state) do
378 388 close_conn(resp_id)
379 − {_, str} = StringIO.contents(state[:refs][resp_id][:fd])
380 − StringIO.close(state[:refs][resp_id][:fd])
381 − t = state[:refs][resp_id][:type]
382 389
383 − case Jason.decode(str) do
384 − {:ok, decoded} ->
385 − case state[:refs][resp_id] |> Map.get(:write) do
386 − false ->
387 − {:noreply, state}
390 + case Map.pop(state.refs, resp_id) do
391 + {nil, _refs} ->
392 + {:noreply, state}
393 +
394 + {ref, refs} ->
395 + {:ok, {_input, body}} = StringIO.close(ref.fd)
396 + {:noreply, store_response(%{state | refs: refs}, ref, body)}
397 + end
398 + end
388 399
389 − nil ->
390 − {:noreply, store_decoded(state, t, state[:refs][resp_id][:id], decoded)}
391 − end
400 + defp store_response(state, %{write: false}, _body), do: state
392 401
393 − {:error, error} ->
394 − {:noreply, state}
402 + defp store_response(state, ref, body) do
403 + case Jason.decode(body) do
404 + {:ok, decoded} -> store_decoded(state, ref.type, ref[:id], decoded)
405 + {:error, _error} -> state
395 406 end
396 407 end
397 408
modified apps/room_gitlab/test/gitlab_worker_test.exs
+27 −0
@@ -84,6 +84,33 @@ defmodule RoomGitlab.WorkerTest do
84 84 end
85 85 end
86 86
87 + test "what still arrives for a request after its end is dropped", %{state: state} do
88 + # Its connection, already gone.
89 + conn = spawn(fn -> :ok end)
90 + {:ok, fd} = StringIO.open("")
91 + state = %{state | refs: %{conn => %{fd: fd, type: :jobs, id: 22}}}
92 +
93 + state =
94 + Enum.reduce(
95 + [
96 + %HTTPoison.AsyncStatus{id: conn, code: 200},
97 + %HTTPoison.AsyncChunk{id: conn, chunk: ~s([{"id":"22"}])},
98 + %HTTPoison.AsyncEnd{id: conn},
99 + # A second response under the same id, as two requests sharing a
100 + # connection used to bring.
101 + %HTTPoison.AsyncStatus{id: conn, code: 503},
102 + %HTTPoison.AsyncChunk{id: conn, chunk: "unavailable"},
103 + %HTTPoison.AsyncEnd{id: conn},
104 + %HTTPoison.Error{id: conn, reason: :closed}
105 + ],
106 + state,
107 + &handle_info/2
108 + )
109 +
110 + assert state.jobs == %{22 => [%{"id" => "22"}]}
111 + assert state.refs == %{}
112 + end
113 +
87 114 defp query_jobs(state, project_ids) do
88 115 projects = for id <- project_ids, do: %{"id" => id}
89 116 Worker.handle_cast({:query_jobs, %{}}, %{state | projects: projects})

Parents: 9b66bba