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.

Read 511's regional feed: a poll interval, and per-agency filtering

55e7f46 · Gabriel Morell · 2026-08-25 21:13

20 files +253 -5
Message 
{commit_body(@commit)}

Files changed

modified ?
+0 −0

Binary file — not rendered.

modified ?
+0 −0

Binary file — not rendered.

modified apps/room_gtfs/lib/room_gtfs/gtfs_worker.ex
+152 −4
@@ -331,7 +331,16 @@ defmodule RoomGtfs.Worker.RT do
331 331 # that looks exactly like silence. See RoomGtfs.FeedHealth.
332 332 defp fetch_recorded(state, kind, url) do
333 333 started = System.monotonic_time(:millisecond)
334 − result = FeedCache.get(url)
334 +
335 + # A source polling every three minutes can happily use data up to three
336 + # minutes old, and several sources sharing one URL -- which is the whole
337 + # point of the regional feed -- then fetch it once between them instead of
338 + # once each.
339 + result =
340 + case rt_period_ms(state) do
341 + nil -> FeedCache.get(url)
342 + period -> FeedCache.get(url, period - :timer.seconds(5))
343 + end
335 344 took = System.monotonic_time(:millisecond) - started
336 345
337 346 RoomGtfs.FeedHealth.record(state.id, state.inst && state.inst.name, kind, url, result, took)
@@ -339,6 +348,63 @@ defmodule RoomGtfs.Worker.RT do
339 348 result
340 349 end
341 350
351 + @doc """
352 + Narrow a multi-agency feed to one source, and unprefix what is left.
353 +
354 + 511's regional feed carries 28 operators in one message, which is how three
355 + Bay Area sources share a single request against a 60-per-hour quota. Trips
356 + there are keyed `SF:12074394_M21`, and the static feed calls the same trip
357 + `12074394_M21`, so a source reading the regional feed matches nothing until
358 + the prefix comes off.
359 +
360 + Stripping alone would be wrong. Two operators can use the same trip id --
361 + short numeric ids like "1100" are common -- so entities belonging to other
362 + agencies are dropped rather than unprefixed into a namespace where they can
363 + collide. Only stop ids are left alone, because the regional feed does not
364 + prefix those.
365 +
366 + A source with no `rt_agency` gets its feed back untouched, which is every
367 + single-agency feed.
368 + """
369 + def for_this_agency(feed, %{inst: %{config: %{rt_agency: agency}}})
370 + when is_binary(agency) and agency != "" do
371 + prefix = agency <> ":"
372 +
373 + entities =
374 + feed.entity
375 + |> Enum.filter(&belongs_to?(&1, prefix))
376 + |> Enum.map(&unprefix(&1, prefix))
377 +
378 + %{feed | entity: entities}
379 + end
380 +
381 + def for_this_agency(feed, _state), do: feed
382 +
383 + defp belongs_to?(entity, prefix) do
384 + String.starts_with?(entity_trip_id(entity) || "", prefix)
385 + end
386 +
387 + defp entity_trip_id(%{trip_update: %{trip: %{trip_id: id}}}) when is_binary(id), do: id
388 + defp entity_trip_id(%{vehicle: %{trip: %{trip_id: id}}}) when is_binary(id), do: id
389 + defp entity_trip_id(_entity), do: nil
390 +
391 + defp unprefix(entity, prefix) do
392 + entity
393 + |> update_trip(entity.trip_update, prefix, :trip_update)
394 + |> then(fn e -> update_trip(e, e.vehicle, prefix, :vehicle) end)
395 + end
396 +
397 + defp update_trip(entity, nil, _prefix, _key), do: entity
398 +
399 + defp update_trip(entity, %{trip: nil}, _prefix, _key), do: entity
400 +
401 + defp update_trip(entity, %{trip: trip} = message, prefix, key) do
402 + trip = %{trip | trip_id: String.replace_prefix(trip.trip_id || "", prefix, "")}
403 + Map.put(entity, key, %{message | trip: trip})
404 + end
405 +
406 + defp update_trip(entity, _message, _prefix, _key), do: entity
407 +
342 408 defp via_tuple(name), do: {:via, Registry, {@registry, name}}
343 409
344 410 def handle_call({:query_alerts, stop, route_ids}, _from, state) do
@@ -687,6 +753,36 @@ defmodule RoomGtfs.Worker.RT do
687 753 end
688 754
689 755 def handle_cast(:update_realtime, state) do
756 + if rt_due?(state) do
757 + do_update_realtime(Map.put(state, :rt_polled_at, System.monotonic_time(:millisecond)))
758 + else
759 + {:noreply, state}
760 + end
761 + end
762 +
763 + # The poller ticks every 30s for everyone; a source with its own interval is
764 + # gated here rather than by rescheduling the Periodic, which cannot be changed
765 + # once started and would have to be torn down and rebuilt on every config
766 + # refresh.
767 + defp rt_due?(state) do
768 + case rt_period_ms(state) do
769 + nil ->
770 + true
771 +
772 + period ->
773 + case Map.get(state, :rt_polled_at) do
774 + nil -> true
775 + at -> System.monotonic_time(:millisecond) - at >= period
776 + end
777 + end
778 + end
779 +
780 + defp rt_period_ms(%{inst: %{config: %{rt_period: seconds}}}) when is_integer(seconds),
781 + do: :timer.seconds(seconds)
782 +
783 + defp rt_period_ms(_state), do: nil
784 +
785 + defp do_update_realtime(state) do
690 786 state =
691 787 case state.inst.enabled do
692 788 true ->
@@ -698,7 +794,7 @@ defmodule RoomGtfs.Worker.RT do
698 794 val ->
699 795 case fetch_recorded(state, :sa, val) do
700 796 {:ok, data_sa} when is_struct(data_sa, TransitRealtime.FeedMessage) ->
701 − state |> Map.put(:rt_sa, data_sa)
797 + state |> Map.put(:rt_sa, for_this_agency(data_sa, state))
702 798
703 799 {:error, error} ->
704 800 Logger.info(
@@ -717,7 +813,7 @@ defmodule RoomGtfs.Worker.RT do
717 813 val ->
718 814 case fetch_recorded(state, :tu, val) do
719 815 {:ok, data_tu} when is_struct(data_tu, TransitRealtime.FeedMessage) ->
720 − state |> Map.put(:rt_tu, data_tu)
816 + state |> Map.put(:rt_tu, for_this_agency(data_tu, state))
721 817
722 818 {:error, error} ->
723 819 Logger.info(
@@ -736,6 +832,7 @@ defmodule RoomGtfs.Worker.RT do
736 832 val ->
737 833 case fetch_recorded(state, :vp, val) do
738 834 {:ok, data_vp} when is_struct(data_vp, TransitRealtime.FeedMessage) ->
835 + data_vp = for_this_agency(data_vp, state)
739 836 new_state = state |> Map.put(:rt_vp, data_vp) |> ensure_stops()
740 837
741 838 # Broadcast vehicle position updates
@@ -1420,6 +1517,57 @@ end
1420 1517 end
1421 1518 end
1422 1519
1520 + # Bytes after the end-of-central-directory record, which some publishers add
1521 + # and the zip format does not allow undeclared.
1522 + #
1523 + # 511.org appends an HTML fragment to every GTFS download -- 680 bytes of
1524 + # `<!DOCTYPE html ...>` -- while declaring a comment length of zero. The
1525 + # `unzip` command shrugs and reads the archive anyway; Unzip looks for the
1526 + # EOCD at the tail of the blob, does not find it there, and reports "missing
1527 + # EOCD record". A ten megabyte feed then fails to import over 680 bytes of
1528 + # markup, and the error names neither the cause nor the culprit.
1529 + #
1530 + # So the record is found and anything past it dropped. A well-formed archive
1531 + # ends exactly at its EOCD and comes back untouched.
1532 + def trim_zip_tail(body) when is_binary(body) do
1533 + case last_eocd_offset(body) do
1534 + nil ->
1535 + body
1536 +
1537 + offset ->
1538 + <<_::binary-size(offset), _::binary-size(20), comment_len::16-little, _rest::binary>> = body
1539 + declared_end = offset + 22 + comment_len
1540 +
1541 + if byte_size(body) > declared_end do
1542 + Logger.info(
1543 + "gtfs zip carried #{byte_size(body) - declared_end} bytes after its EOCD; trimming"
1544 + )
1545 +
1546 + binary_part(body, 0, declared_end)
1547 + else
1548 + body
1549 + end
1550 + end
1551 + end
1552 +
1553 + def trim_zip_tail(body), do: body
1554 +
1555 + # Searched from the end, and only over the tail: the signature is four bytes
1556 + # and can occur by chance inside compressed data, so the last plausible
1557 + # occurrence is the one that means anything. 128KB covers a legal 64KB comment
1558 + # and then some junk on top of it.
1559 + defp last_eocd_offset(body) do
1560 + window = min(byte_size(body), 128 * 1024)
1561 + start = byte_size(body) - window
1562 +
1563 + body
1564 + |> binary_part(start, window)
1565 + |> then(&:binary.matches(&1, <<0x50, 0x4B, 0x05, 0x06>>))
1566 + |> Enum.map(fn {at, _len} -> start + at end)
1567 + |> Enum.filter(&(&1 + 22 <= byte_size(body)))
1568 + |> List.last()
1569 + end
1570 +
1423 1571 defp do_import_static(id, cfg) do
1424 1572 Logger.info("GTFS::#{id} updating static info")
1425 1573 bcast(id, :downloading, 1, 11)
@@ -1428,7 +1576,7 @@ end
1428 1576 {:ok, result} ->
1429 1577 bcast(id, :extracting, 2, 11)
1430 1578
1431 − case result.body |> Unzip.InMem.new() |> Unzip.new() do
1579 + case result.body |> trim_zip_tail() |> Unzip.InMem.new() |> Unzip.new() do
1432 1580 {:ok, unzip} ->
1433 1581 files = Unzip.list_entries(unzip)
1434 1582
added apps/room_gtfs/test/zip_tail_test.exs
+62 −0
@@ -0,0 +1,62 @@
1 +defmodule RoomGtfs.ZipTailTest do
2 + @moduledoc """
3 + 511.org appends an HTML fragment to every GTFS download -- 680 bytes of
4 + `<!DOCTYPE html ...>` after the end-of-central-directory record, while
5 + declaring a comment length of zero. `unzip` reads the archive anyway; Unzip
6 + looks for the EOCD at the tail of the blob and reports "missing EOCD record",
7 + so a ten megabyte feed fails to import over 680 bytes of markup.
8 + """
9 + use ExUnit.Case, async: true
10 +
11 + alias RoomGtfs.Worker.Static
12 +
13 + # A real archive, built rather than checked in, so the fixture cannot drift
14 + # from what the zip format actually produces.
15 + defp zip_bytes do
16 + {:ok, {_name, bytes}} =
17 + :zip.create(~c"feed.zip", [{~c"agency.txt", "agency_id,agency_name\nX,Example\n"}], [:memory])
18 +
19 + bytes
20 + end
21 +
22 + defp readable?(blob) do
23 + match?({:ok, _}, blob |> Unzip.InMem.new() |> Unzip.new())
24 + end
25 +
26 + test "a well-formed archive is returned untouched" do
27 + blob = zip_bytes()
28 + assert Static.trim_zip_tail(blob) == blob
29 + assert readable?(blob)
30 + end
31 +
32 + test "html appended after the EOCD is trimmed, and the archive then reads" do
33 + blob = zip_bytes()
34 + junk = "\r\n\r\n<!DOCTYPE html PUBLIC \"-//W3C//DTD XHTML 1.0 Transitional//EN\">\n</html>\r\n"
35 + polluted = blob <> junk
36 +
37 + refute readable?(polluted), "precondition: the polluted blob is what Unzip rejects"
38 +
39 + trimmed = Static.trim_zip_tail(polluted)
40 + assert trimmed == blob
41 + assert readable?(trimmed)
42 + end
43 +
44 + test "a declared comment is kept, since it is part of the archive" do
45 + # Trimming to EOCD+22 unconditionally would truncate a legal comment and
46 + # leave the record claiming bytes that are no longer there.
47 + blob = zip_bytes()
48 + comment = "built by a publisher that uses the comment field"
49 + len = byte_size(comment)
50 +
51 + # The comment length lives in the last two bytes of the EOCD record.
52 + with_comment = binary_part(blob, 0, byte_size(blob) - 2) <> <<len::16-little>> <> comment
53 +
54 + assert Static.trim_zip_tail(with_comment) == with_comment
55 + end
56 +
57 + test "a body that is not a zip at all is left alone to fail downstream" do
58 + # The decoder's own error is more informative than anything this could say.
59 + assert Static.trim_zip_tail("<html>not a zip</html>") == "<html>not a zip</html>"
60 + assert Static.trim_zip_tail(<<>>) == <<>>
61 + end
62 +end
modified apps/room_sanctum/lib/room_sanctum/configuration/configs/gtfs.ex
+19 −1
@@ -10,11 +10,29 @@ defmodule RoomSanctum.Configuration.Configs.GTFS do
10 10 field :url_rt_tu, :string
11 11 field :url_rt_vp, :string
12 12 field :tz, :string
13 +
14 + # Seconds between realtime polls. Nil polls at the default 30s, which is
15 + # what an unmetered feed wants. A metered one does not: 511.org allows 60
16 + # requests an hour on a default token, and three feed kinds polled every 30
17 + # seconds is 360 -- so those sources set this and trade freshness for
18 + # staying inside the budget.
19 + field :rt_period, :integer
20 +
21 + # Operator code for a feed that carries several agencies at once, e.g. "SF"
22 + # for a source reading 511's regional feed. Realtime entities there are
23 + # keyed `SF:trip_id`, so this both selects the ones belonging to this source
24 + # and says what prefix to strip. Left blank for a single-agency feed, which
25 + # is nearly all of them.
26 + field :rt_agency, :string
13 27 end
14 28
15 29 def changeset(source, params) do
16 30 source
17 − |> cast(params, ~w(url url_rt_sa url_rt_tu url_rt_vp tz)a)
31 + |> cast(params, ~w(url url_rt_sa url_rt_tu url_rt_vp tz rt_period rt_agency)a)
18 32 |> validate_required([:url, :tz])
33 + # Below about ten seconds the poller cannot keep up and the feed will not
34 + # have changed anyway; the upper bound is a day, past which it is not
35 + # realtime by any reading.
36 + |> validate_number(:rt_period, greater_than_or_equal_to: 10, less_than_or_equal_to: 86_400)
19 37 end
20 38 end
modified apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/form_component.html.heex
+20 −0
@@ -79,6 +79,26 @@
79 79 <div class="form-control">
80 80 <.input field={gtfs_form[:url_rt_vp]} type="text" label="Realtime URL Vehicle Positions"/>
81 81 </div>
82 + <div class="form-control">
83 + <.input field={gtfs_form[:rt_period]} type="number" label="Realtime poll interval (seconds)"/>
84 + <label class="label">
85 + <span class="label-text-alt opacity-70">
86 + Blank polls every 30s. Raise it for a feed with a request quota &mdash;
87 + 511.org allows 60 requests an hour, which three feed kinds exhaust in
88 + twenty minutes at the default.
89 + </span>
90 + </label>
91 + </div>
92 + <div class="form-control">
93 + <.input field={gtfs_form[:rt_agency]} type="text" label="Realtime operator code"/>
94 + <label class="label">
95 + <span class="label-text-alt opacity-70">
96 + Only for a feed carrying several agencies at once, such as 511's regional
97 + feed, where trips are keyed <code>SF:12074394_M21</code>. Selects this
98 + source's entities and strips the prefix. Leave blank otherwise.
99 + </span>
100 + </label>
101 + </div>
82 102 <% end %>
83 103 <div class="form-control mt-2">
84 104 <label class="label"><span class="label-text">Static refresh interval</span></label>

Parents: d446902