From 00c9b1b4d0dede6ed5939bf2bac9d40839566cfe Mon Sep 17 00:00:00 2001 From: "Corey N. Runkel" Date: Wed, 5 Aug 2026 14:23:40 -0400 Subject: [PATCH 1/4] feat(state,parse): Only update new stop events --- apps/model/lib/model/stop_event.ex | 7 +- apps/parse/lib/parse/stop_events.ex | 51 +++++-- apps/parse/test/parse/stop_events_test.exs | 93 +++++++++++- apps/state/lib/state/stop_event.ex | 61 ++++++++ apps/state/test/state/stop_event_test.exs | 157 +++++++++++++++++++++ 5 files changed, 354 insertions(+), 15 deletions(-) diff --git a/apps/model/lib/model/stop_event.ex b/apps/model/lib/model/stop_event.ex index ae4a1f557..c7b20ddb7 100644 --- a/apps/model/lib/model/stop_event.ex +++ b/apps/model/lib/model/stop_event.ex @@ -14,7 +14,8 @@ defmodule Model.StopEvent do :stop_id, :stop_sequence, :arrived, - :departed + :departed, + :timestamp ] @typedoc """ @@ -35,6 +36,7 @@ defmodule Model.StopEvent do See [GTFS `stop_times.txt` `stop_sequence`](https://gtfs.org/documentation/schedule/reference/#stop_timestxt). * `:arrived` - When the vehicle (`vehicle_id`) arrived at the stop (`stop_id`) as a time-zone aware [RFC 3339 datetime](https://datatracker.ietf.org/doc/html/rfc3339#page-10). `nil` if the stop is the first stop on the trip (`trip_id`). * `:departed` - When the vehicle (`vehicle_id`) departed from the stop (`stop_id`) as time-zone aware [RFC 3339 datetime](https://datatracker.ietf.org/doc/html/rfc3339#page-10). `nil` if the last stop (`stop_id`) on the trip (`trip_id`). + * `:timestamp` - Unix timestamp representing when this stop event record was last updated in the source system. """ @type t :: %__MODULE__{ @@ -48,6 +50,7 @@ defmodule Model.StopEvent do stop_sequence: Model.Schedule.stop_sequence(), revenue: :REVENUE | :NON_REVENUE, arrived: DateTime.t() | nil, - departed: DateTime.t() | nil + departed: DateTime.t() | nil, + timestamp: integer() | nil } end diff --git a/apps/parse/lib/parse/stop_events.ex b/apps/parse/lib/parse/stop_events.ex index ef597ff07..5fc89fbab 100644 --- a/apps/parse/lib/parse/stop_events.ex +++ b/apps/parse/lib/parse/stop_events.ex @@ -10,13 +10,28 @@ defmodule Parse.StopEvents do @behaviour Parse @impl Parse - @spec parse(binary()) :: [Model.StopEvent.t()] - def parse(body) do - body - |> decompress() - |> String.split("\n", trim: true) - |> Enum.map(&parse_line/1) - |> Enum.reject(&is_nil/1) + def parse(binary) when is_binary(binary) do + parse(binary, []) + end + + @spec parse(binary(), keyword()) :: [Model.StopEvent.t()] | {:partial, [Model.StopEvent.t()]} + def parse(body, opts) when is_binary(body) and is_list(opts) do + newer_than = Keyword.get(opts, :newer_than) + + events = + body + |> decompress() + |> String.split("\n", trim: true) + |> Enum.map(&parse_line(&1, newer_than)) + |> Enum.reject(&is_nil/1) + + # When filtering by timestamp, always return {:partial, events} tuple + # to distinguish filtered updates from full state replacements + if newer_than do + {:partial, events} + else + events + end end defp decompress(body) do @@ -25,10 +40,14 @@ defmodule Parse.StopEvents do _ -> body end - defp parse_line(line) do + defp parse_line(line, newer_than) do case Jason.decode(line) do {:ok, record} -> - parse_record(record) + if should_include_record?(record, newer_than) do + parse_record(record) + else + nil + end e -> Logger.error("#{__MODULE__} decode_error error=#{inspect(e)}") @@ -36,6 +55,17 @@ defmodule Parse.StopEvents do end end + defp should_include_record?(_record, nil), do: true + + defp should_include_record?(%{"timestamp" => timestamp}, newer_than) + when is_integer(timestamp) and is_integer(newer_than) do + timestamp > newer_than + end + + # Records without a timestamp field are excluded when filtering is active. + # This treats records without timestamps as stale/invalid when doing incremental updates. + defp should_include_record?(_record, _newer_than), do: false + defp parse_record( %{ "id" => id, @@ -64,7 +94,8 @@ defmodule Parse.StopEvents do stop_id: stop_id, stop_sequence: stop_sequence, arrived: arrived, - departed: departed + departed: departed, + timestamp: Map.get(record, "timestamp") } else {:error, reason} -> diff --git a/apps/parse/test/parse/stop_events_test.exs b/apps/parse/test/parse/stop_events_test.exs index 1036e588f..783b6bcb4 100644 --- a/apps/parse/test/parse/stop_events_test.exs +++ b/apps/parse/test/parse/stop_events_test.exs @@ -33,7 +33,8 @@ defmodule Parse.StopEventsTest do stop_id: "70512", stop_sequence: 4, arrived: DateTime.from_naive!(~N[2026-02-24T10:18:23], "America/New_York"), - departed: DateTime.from_naive!(~N[2026-02-24T10:21:19], "America/New_York") + departed: DateTime.from_naive!(~N[2026-02-24T10:21:19], "America/New_York"), + timestamp: 1771950045 }, # arrival only %StopEvent{ @@ -47,7 +48,8 @@ defmodule Parse.StopEventsTest do stop_id: "2231", stop_sequence: 1, arrived: DateTime.from_naive!(~N[2026-02-24T15:54:46], "America/New_York"), - departed: nil + departed: nil, + timestamp: 1771968343 }, # departure only %StopEvent{ @@ -61,7 +63,8 @@ defmodule Parse.StopEventsTest do stop_id: "12232", stop_sequence: 2, arrived: nil, - departed: DateTime.from_naive!(~N[2026-02-24T16:08:53], "America/New_York") + departed: DateTime.from_naive!(~N[2026-02-24T16:08:53], "America/New_York"), + timestamp: 1771968343 } ] @@ -152,4 +155,88 @@ defmodule Parse.StopEventsTest do assert [%StopEvent{id: "test-trip-1"}] = result end end + + describe "parse with timestamp filtering" do + test "returns {:partial, events} when newer_than option filters out old records" do + ndjson = """ + {"id":"old-1","timestamp":1771968300,"start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} + {"id":"old-2","timestamp":1771968340,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} + {"id":"new-1","timestamp":1771968350,"start_date":"20260224","trip_id":"test3","vehicle_id":"v3","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop3","stop_sequence":3,"arrived":1771966486,"departed":1771967246} + {"id":"new-2","timestamp":1771968360,"start_date":"20260224","trip_id":"test4","vehicle_id":"v4","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop4","stop_sequence":4,"arrived":1771966486,"departed":1771967246} + """ + + result = parse(ndjson, newer_than: 1771968343) + + assert {:partial, events} = result + assert length(events) == 2 + assert Enum.all?(events, fn e -> e.id in ["new-1", "new-2"] end) + end + + test "returns full list when newer_than option is not provided" do + ndjson = """ + {"id":"event-1","timestamp":1771968300,"start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} + {"id":"event-2","timestamp":1771968350,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} + """ + + result = parse(ndjson) + + assert is_list(result) + refute match?({:partial, _}, result) + assert length(result) == 2 + end + + test "returns {:partial, []} when all records are older than newer_than" do + ndjson = """ + {"id":"old-1","timestamp":1771968300,"start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} + {"id":"old-2","timestamp":1771968340,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} + """ + + result = parse(ndjson, newer_than: 1771968400) + + assert {:partial, []} = result + end + + test "excludes records at the boundary (timestamp == newer_than)" do + ndjson = """ + {"id":"at-boundary","timestamp":1771968343,"start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} + {"id":"after-boundary","timestamp":1771968344,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} + """ + + result = parse(ndjson, newer_than: 1771968343) + + # Only the record with timestamp > 1771968343 should be included + assert {:partial, events} = result + assert length(events) == 1 + assert [%StopEvent{id: "after-boundary"}] = events + end + + test "works with gzipped data and newer_than option" do + ndjson = """ + {"id":"old-1","timestamp":1771968300,"start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} + {"id":"new-1","timestamp":1771968350,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} + """ + + gzipped = :zlib.gzip(ndjson) + + result = parse(gzipped, newer_than: 1771968343) + + assert {:partial, events} = result + assert length(events) == 1 + assert [%StopEvent{id: "new-1"}] = events + end + + test "handles records with missing timestamp field by excluding them" do + ndjson = """ + {"id":"no-timestamp","start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} + {"id":"with-timestamp","timestamp":1771968350,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} + """ + + result = parse(ndjson, newer_than: 1771968343) + + # Records without timestamp are excluded when filtering is enabled + assert {:partial, events} = result + assert length(events) == 1 + assert [%StopEvent{id: "with-timestamp"}] = events + end + end end diff --git a/apps/state/lib/state/stop_event.ex b/apps/state/lib/state/stop_event.ex index c8ee688aa..21f3d6fd2 100644 --- a/apps/state/lib/state/stop_event.ex +++ b/apps/state/lib/state/stop_event.ex @@ -24,6 +24,67 @@ defmodule State.StopEvent do # Filter keys ordered by typical selectivity (most selective first) @index_keys [:trip_ids, :vehicle_ids, :stop_ids, :route_ids] + @impl State.Server + def handle_new_state(binary) when is_binary(binary) do + # Get the maximum timestamp from existing data + max_timestamp = get_max_timestamp() + + # Parse with timestamp filtering if we have existing data + opts = if max_timestamp, do: [newer_than: max_timestamp], else: [] + + parser = Parse.StopEvents + + parsed_data = + try do + parser.parse(binary, opts) + rescue + e -> + # log_parse_error returns nil, so we return nil on error to avoid passing + # invalid data to super/1. The caller (State.Server) expects nil to mean + # "no data to insert" and will handle it gracefully. + State.Server.log_parse_error(__MODULE__, e) + end + + # Only proceed with update if parsing succeeded + case parsed_data do + nil -> :ok + data -> super(data) + end + end + + def handle_new_state(data), do: super(data) + + # Get the maximum timestamp from existing data in the table. + # Uses a dynamic match spec based on the StopEvent struct to avoid brittleness + # from hardcoded field positions. + defp get_max_timestamp do + # Mnesia records are tuples of {RecordName, field1, field2, ...} where + # fields follow Recordable declaration order (from StopEvent.fields/0). + fields = StopEvent.fields() + timestamp_position = Enum.find_index(fields, &(&1 == :timestamp)) + + unless timestamp_position do + raise "timestamp field not found in StopEvent struct" + end + + # Build tuple pattern: {StopEvent, :_, :_, ..., :"$1"} with $1 at timestamp's position + wildcards = List.duplicate(:_, length(fields)) + pattern_list = [StopEvent | List.replace_at(wildcards, timestamp_position, :"$1")] + pattern = List.to_tuple(pattern_list) + + match_spec = [{pattern, [], [:"$1"]}] + + case :mnesia.dirty_select(__MODULE__, match_spec) do + [] -> + nil + + timestamps -> + timestamps + |> Enum.reject(&is_nil/1) + |> Enum.max(fn -> nil end) + end + end + @spec by_id(String.t()) :: StopEvent.t() | nil def by_id(id) do case super(id) do diff --git a/apps/state/test/state/stop_event_test.exs b/apps/state/test/state/stop_event_test.exs index cb9597448..31f3d5d36 100644 --- a/apps/state/test/state/stop_event_test.exs +++ b/apps/state/test/state/stop_event_test.exs @@ -391,4 +391,161 @@ defmodule State.StopEventTest do assert by_id("nonexistent") == nil end end + + describe "partial updates with timestamps" do + test "partial update only affects records with matching keys" do + # Initial full state + initial_events = [ + %StopEvent{ + id: "trip1-route1-v1-1", + vehicle_id: "v1", + start_date: ~D[2026-02-24], + trip_id: "trip1", + direction_id: 0, + route_id: "route1", + revenue: :REVENUE, + stop_id: "stop1", + stop_sequence: 1, + arrived: ~U[2026-02-24 15:28:06Z], + departed: ~U[2026-02-24 15:40:46Z] + }, + %StopEvent{ + id: "trip1-route1-v1-2", + vehicle_id: "v1", + start_date: ~D[2026-02-24], + trip_id: "trip1", + direction_id: 0, + route_id: "route1", + revenue: :REVENUE, + stop_id: "stop2", + stop_sequence: 2, + arrived: ~U[2026-02-24 15:41:26Z], + departed: ~U[2026-02-24 15:42:13Z] + } + ] + + State.StopEvent.new_state(initial_events) + assert State.StopEvent.size() == 2 + + # Partial update - replace event with id trip1-route1-v1-1 and add new event + updated_event = %StopEvent{ + id: "trip1-route1-v1-1", + vehicle_id: "v1", + start_date: ~D[2026-02-24], + trip_id: "trip1", + direction_id: 0, + route_id: "route1", + revenue: :REVENUE, + stop_id: "stop1", + stop_sequence: 1, + arrived: ~U[2026-02-24 15:28:06Z], + departed: ~U[2026-02-24 15:41:00Z] + } + + new_event = %StopEvent{ + id: "trip1-route1-v1-3", + vehicle_id: "v1", + start_date: ~D[2026-02-24], + trip_id: "trip1", + direction_id: 0, + route_id: "route1", + revenue: :REVENUE, + stop_id: "stop3", + stop_sequence: 3, + arrived: ~U[2026-02-24 15:43:00Z], + departed: ~U[2026-02-24 15:44:00Z] + } + + State.StopEvent.new_state({:partial, [updated_event, new_event]}) + + # Should now have 3 events + assert State.StopEvent.size() == 3 + + # Verify the update was applied + result = by_id("trip1-route1-v1-1") + assert result.departed == ~U[2026-02-24 15:41:00Z] + + # Verify the new event was added + result = by_id("trip1-route1-v1-3") + assert result.stop_sequence == 3 + + # Verify the unchanged event is still there + result = by_id("trip1-route1-v1-2") + assert result.departed == ~U[2026-02-24 15:42:13Z] + end + + test "accepts NDJSON string with timestamp filtering" do + # Create initial state with old events + ndjson_old = """ + {"id":"old-event-1","timestamp":1771968300,"start_date":"20260224","trip_id":"trip1","vehicle_id":"v1","direction_id":0,"route_id":"route1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} + {"id":"old-event-2","timestamp":1771968340,"start_date":"20260224","trip_id":"trip2","vehicle_id":"v2","direction_id":0,"route_id":"route2","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} + """ + + gzipped_old = :zlib.gzip(ndjson_old) + State.StopEvent.new_state(gzipped_old) + assert State.StopEvent.size() == 2 + + # Verify the timestamps were stored + event1 = by_id("old-event-1") + assert event1.timestamp == 1771968300 + + # Send an update with both old and new events + # The automatic timestamp filtering should only process the new ones + ndjson_update = """ + {"id":"old-event-1","timestamp":1771968300,"start_date":"20260224","trip_id":"trip1","vehicle_id":"v1","direction_id":0,"route_id":"route1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} + {"id":"old-event-2","timestamp":1771968340,"start_date":"20260224","trip_id":"trip2","vehicle_id":"v2","direction_id":0,"route_id":"route2","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} + {"id":"new-event-1","timestamp":1771968350,"start_date":"20260224","trip_id":"trip3","vehicle_id":"v3","direction_id":0,"route_id":"route3","revenue":true,"stop_id":"stop3","stop_sequence":3,"arrived":1771966486,"departed":1771967246} + {"id":"new-event-2","timestamp":1771968360,"start_date":"20260224","trip_id":"trip4","vehicle_id":"v4","direction_id":0,"route_id":"route4","revenue":true,"stop_id":"stop4","stop_sequence":4,"arrived":1771966486,"departed":1771967246} + """ + + gzipped_update = :zlib.gzip(ndjson_update) + + # This should automatically use timestamp filtering and only add the 2 new events + State.StopEvent.new_state(gzipped_update) + + # Should have 4 events total (2 old + 2 new) + assert State.StopEvent.size() == 4 + + # Verify the new events were added + new_event1 = by_id("new-event-1") + assert new_event1.timestamp == 1771968350 + + new_event2 = by_id("new-event-2") + assert new_event2.timestamp == 1771968360 + end + + test "timestamp filtering works correctly on subsequent updates" do + # Initial state + ndjson1 = """ + {"id":"event-1","timestamp":1771968300,"start_date":"20260224","trip_id":"trip1","vehicle_id":"v1","direction_id":0,"route_id":"route1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} + """ + + State.StopEvent.new_state(:zlib.gzip(ndjson1)) + assert State.StopEvent.size() == 1 + + # Second update - should filter based on timestamp 1771968300 + ndjson2 = """ + {"id":"event-1","timestamp":1771968300,"start_date":"20260224","trip_id":"trip1","vehicle_id":"v1","direction_id":0,"route_id":"route1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} + {"id":"event-2","timestamp":1771968350,"start_date":"20260224","trip_id":"trip2","vehicle_id":"v2","direction_id":0,"route_id":"route2","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} + """ + + State.StopEvent.new_state(:zlib.gzip(ndjson2)) + assert State.StopEvent.size() == 2 + + # Third update - should filter based on timestamp 1771968350 + ndjson3 = """ + {"id":"event-1","timestamp":1771968300,"start_date":"20260224","trip_id":"trip1","vehicle_id":"v1","direction_id":0,"route_id":"route1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} + {"id":"event-2","timestamp":1771968350,"start_date":"20260224","trip_id":"trip2","vehicle_id":"v2","direction_id":0,"route_id":"route2","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} + {"id":"event-3","timestamp":1771968400,"start_date":"20260224","trip_id":"trip3","vehicle_id":"v3","direction_id":0,"route_id":"route3","revenue":true,"stop_id":"stop3","stop_sequence":3,"arrived":1771966486,"departed":1771967246} + """ + + State.StopEvent.new_state(:zlib.gzip(ndjson3)) + assert State.StopEvent.size() == 3 + + # Verify all events are present + assert by_id("event-1").timestamp == 1771968300 + assert by_id("event-2").timestamp == 1771968350 + assert by_id("event-3").timestamp == 1771968400 + end + end end From 2794a99b9eded3391f6652e6a7449654bce8f854 Mon Sep 17 00:00:00 2001 From: "Corey N. Runkel" Date: Wed, 5 Aug 2026 17:15:07 -0400 Subject: [PATCH 2/4] Evict old records from stop_events --- apps/parse/test/parse/stop_events_test.exs | 16 +- apps/state/lib/state/stop_event.ex | 78 +++++++ apps/state/test/state/stop_event_test.exs | 231 ++++++++++++++++++++- 3 files changed, 311 insertions(+), 14 deletions(-) diff --git a/apps/parse/test/parse/stop_events_test.exs b/apps/parse/test/parse/stop_events_test.exs index 783b6bcb4..49cfdda33 100644 --- a/apps/parse/test/parse/stop_events_test.exs +++ b/apps/parse/test/parse/stop_events_test.exs @@ -34,7 +34,7 @@ defmodule Parse.StopEventsTest do stop_sequence: 4, arrived: DateTime.from_naive!(~N[2026-02-24T10:18:23], "America/New_York"), departed: DateTime.from_naive!(~N[2026-02-24T10:21:19], "America/New_York"), - timestamp: 1771950045 + timestamp: 1_771_950_045 }, # arrival only %StopEvent{ @@ -49,7 +49,7 @@ defmodule Parse.StopEventsTest do stop_sequence: 1, arrived: DateTime.from_naive!(~N[2026-02-24T15:54:46], "America/New_York"), departed: nil, - timestamp: 1771968343 + timestamp: 1_771_968_343 }, # departure only %StopEvent{ @@ -64,7 +64,7 @@ defmodule Parse.StopEventsTest do stop_sequence: 2, arrived: nil, departed: DateTime.from_naive!(~N[2026-02-24T16:08:53], "America/New_York"), - timestamp: 1771968343 + timestamp: 1_771_968_343 } ] @@ -165,7 +165,7 @@ defmodule Parse.StopEventsTest do {"id":"new-2","timestamp":1771968360,"start_date":"20260224","trip_id":"test4","vehicle_id":"v4","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop4","stop_sequence":4,"arrived":1771966486,"departed":1771967246} """ - result = parse(ndjson, newer_than: 1771968343) + result = parse(ndjson, newer_than: 1_771_968_343) assert {:partial, events} = result assert length(events) == 2 @@ -191,7 +191,7 @@ defmodule Parse.StopEventsTest do {"id":"old-2","timestamp":1771968340,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} """ - result = parse(ndjson, newer_than: 1771968400) + result = parse(ndjson, newer_than: 1_771_968_400) assert {:partial, []} = result end @@ -202,7 +202,7 @@ defmodule Parse.StopEventsTest do {"id":"after-boundary","timestamp":1771968344,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} """ - result = parse(ndjson, newer_than: 1771968343) + result = parse(ndjson, newer_than: 1_771_968_343) # Only the record with timestamp > 1771968343 should be included assert {:partial, events} = result @@ -218,7 +218,7 @@ defmodule Parse.StopEventsTest do gzipped = :zlib.gzip(ndjson) - result = parse(gzipped, newer_than: 1771968343) + result = parse(gzipped, newer_than: 1_771_968_343) assert {:partial, events} = result assert length(events) == 1 @@ -231,7 +231,7 @@ defmodule Parse.StopEventsTest do {"id":"with-timestamp","timestamp":1771968350,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} """ - result = parse(ndjson, newer_than: 1771968343) + result = parse(ndjson, newer_than: 1_771_968_343) # Records without timestamp are excluded when filtering is enabled assert {:partial, events} = result diff --git a/apps/state/lib/state/stop_event.ex b/apps/state/lib/state/stop_event.ex index 21f3d6fd2..05bb923b8 100644 --- a/apps/state/lib/state/stop_event.ex +++ b/apps/state/lib/state/stop_event.ex @@ -54,6 +54,78 @@ defmodule State.StopEvent do def handle_new_state(data), do: super(data) + @impl State.Server + def post_commit_hook do + evict_old_records() + :ok + end + + @doc """ + Evicts records older than the configured retention period. + + ## Parameters + + * `retention_seconds` - Number of seconds to retain records. Defaults to configured + value (default: 7200 seconds / 2 hours). Records with timestamps older than + `now - retention_seconds` will be deleted. + + Records without timestamps are not evicted. + + ## Examples + + # Use default retention period (2 hours) + evict_old_records() + + # Use custom retention period (1 hour) + evict_old_records(3600) + + """ + @spec evict_old_records(non_neg_integer()) :: :ok + def evict_old_records(retention_seconds \\ nil) do + retention_seconds = retention_seconds || get_retention_seconds() + now = System.system_time(:second) + cutoff = now - retention_seconds + + # Build match spec to find records with timestamp < cutoff + fields = StopEvent.fields() + timestamp_position = Enum.find_index(fields, &(&1 == :timestamp)) + id_position = Enum.find_index(fields, &(&1 == :id)) + + unless timestamp_position && id_position do + raise "timestamp or id field not found in StopEvent struct" + end + + # Build pattern with both timestamp and id as variables + wildcards = List.duplicate(:_, length(fields)) + + pattern_list = [ + StopEvent + | wildcards + |> List.replace_at(timestamp_position, :"$1") + |> List.replace_at(id_position, :"$2") + ] + + pattern = List.to_tuple(pattern_list) + + # Match spec: select ids where timestamp < cutoff and timestamp is not nil + # Guards: timestamp is not nil AND timestamp < cutoff + match_spec = [ + {pattern, + [ + {:andalso, {:"/=", :"$1", nil}, {:<, :"$1", cutoff}} + ], [:"$2"]} + ] + + ids_to_delete = :mnesia.dirty_select(__MODULE__, match_spec) + + # Delete each record by id + Enum.each(ids_to_delete, fn id -> + :mnesia.dirty_delete(__MODULE__, id) + end) + + :ok + end + # Get the maximum timestamp from existing data in the table. # Uses a dynamic match spec based on the StopEvent struct to avoid brittleness # from hardcoded field positions. @@ -85,6 +157,12 @@ defmodule State.StopEvent do end end + defp get_retention_seconds do + :state + |> Application.get_env(__MODULE__, []) + |> Keyword.get(:retention_seconds, 7200) + end + @spec by_id(String.t()) :: StopEvent.t() | nil def by_id(id) do case super(id) do diff --git a/apps/state/test/state/stop_event_test.exs b/apps/state/test/state/stop_event_test.exs index 31f3d5d36..949199826 100644 --- a/apps/state/test/state/stop_event_test.exs +++ b/apps/state/test/state/stop_event_test.exs @@ -4,6 +4,13 @@ defmodule State.StopEventTest do alias Model.StopEvent import State.StopEvent + # Disable automatic eviction by default for all tests except those explicitly testing eviction + setup do + Application.put_env(:state, State.StopEvent, retention_seconds: 999_999_999) + on_exit(fn -> Application.delete_env(:state, State.StopEvent) end) + :ok + end + describe "filter_by/1" do setup do stop_event1 = %StopEvent{ @@ -487,7 +494,7 @@ defmodule State.StopEventTest do # Verify the timestamps were stored event1 = by_id("old-event-1") - assert event1.timestamp == 1771968300 + assert event1.timestamp == 1_771_968_300 # Send an update with both old and new events # The automatic timestamp filtering should only process the new ones @@ -508,10 +515,10 @@ defmodule State.StopEventTest do # Verify the new events were added new_event1 = by_id("new-event-1") - assert new_event1.timestamp == 1771968350 + assert new_event1.timestamp == 1_771_968_350 new_event2 = by_id("new-event-2") - assert new_event2.timestamp == 1771968360 + assert new_event2.timestamp == 1_771_968_360 end test "timestamp filtering works correctly on subsequent updates" do @@ -543,9 +550,221 @@ defmodule State.StopEventTest do assert State.StopEvent.size() == 3 # Verify all events are present - assert by_id("event-1").timestamp == 1771968300 - assert by_id("event-2").timestamp == 1771968350 - assert by_id("event-3").timestamp == 1771968400 + assert by_id("event-1").timestamp == 1_771_968_300 + assert by_id("event-2").timestamp == 1_771_968_350 + assert by_id("event-3").timestamp == 1_771_968_400 + end + end + + describe "evicting old records" do + # Tests in this describe block use the global setup that disables eviction + + test "evicts records older than the configured retention period" do + now = System.system_time(:second) + two_hours_ago = now - 7200 + three_hours_ago = now - 10_800 + one_hour_ago = now - 3600 + + old_event = %StopEvent{ + id: "old-event", + vehicle_id: "v1", + start_date: ~D[2026-08-05], + trip_id: "trip1", + direction_id: 0, + route_id: "route1", + revenue: :REVENUE, + stop_id: "stop1", + stop_sequence: 1, + arrived: nil, + departed: nil, + timestamp: three_hours_ago + } + + borderline_event = %StopEvent{ + id: "borderline-event", + vehicle_id: "v2", + start_date: ~D[2026-08-05], + trip_id: "trip2", + direction_id: 0, + route_id: "route2", + revenue: :REVENUE, + stop_id: "stop2", + stop_sequence: 1, + arrived: nil, + departed: nil, + timestamp: two_hours_ago + } + + recent_event = %StopEvent{ + id: "recent-event", + vehicle_id: "v3", + start_date: ~D[2026-08-05], + trip_id: "trip3", + direction_id: 0, + route_id: "route3", + revenue: :REVENUE, + stop_id: "stop3", + stop_sequence: 1, + arrived: nil, + departed: nil, + timestamp: one_hour_ago + } + + State.StopEvent.new_state([old_event, borderline_event, recent_event]) + assert State.StopEvent.size() == 3 + + # Trigger eviction with 2-hour retention (7200 seconds) + State.StopEvent.evict_old_records(7200) + + # Old event should be evicted, borderline is exactly at boundary (should be kept), + # recent should remain + assert State.StopEvent.size() == 2 + assert by_id("old-event") == nil + assert by_id("borderline-event") != nil + assert by_id("recent-event") != nil + end + + test "evicts records using custom retention period" do + now = System.system_time(:second) + thirty_minutes_ago = now - 1800 + forty_five_minutes_ago = now - 2700 + + old_event = %StopEvent{ + id: "old-event", + vehicle_id: "v1", + start_date: ~D[2026-08-05], + trip_id: "trip1", + direction_id: 0, + route_id: "route1", + revenue: :REVENUE, + stop_id: "stop1", + stop_sequence: 1, + arrived: nil, + departed: nil, + timestamp: forty_five_minutes_ago + } + + recent_event = %StopEvent{ + id: "recent-event", + vehicle_id: "v2", + start_date: ~D[2026-08-05], + trip_id: "trip2", + direction_id: 0, + route_id: "route2", + revenue: :REVENUE, + stop_id: "stop2", + stop_sequence: 1, + arrived: nil, + departed: nil, + timestamp: thirty_minutes_ago + } + + State.StopEvent.new_state([old_event, recent_event]) + assert State.StopEvent.size() == 2 + + # Evict with 40 minute retention (2400 seconds) + State.StopEvent.evict_old_records(2400) + + assert State.StopEvent.size() == 1 + assert by_id("old-event") == nil + assert by_id("recent-event") != nil + end + + test "handles records without timestamps gracefully" do + now = System.system_time(:second) + one_hour_ago = now - 3600 + + event_with_timestamp = %StopEvent{ + id: "with-timestamp", + vehicle_id: "v1", + start_date: ~D[2026-08-05], + trip_id: "trip1", + direction_id: 0, + route_id: "route1", + revenue: :REVENUE, + stop_id: "stop1", + stop_sequence: 1, + arrived: nil, + departed: nil, + timestamp: one_hour_ago + } + + event_without_timestamp = %StopEvent{ + id: "without-timestamp", + vehicle_id: "v2", + start_date: ~D[2026-08-05], + trip_id: "trip2", + direction_id: 0, + route_id: "route2", + revenue: :REVENUE, + stop_id: "stop2", + stop_sequence: 1, + arrived: nil, + departed: nil, + timestamp: nil + } + + State.StopEvent.new_state([event_with_timestamp, event_without_timestamp]) + assert State.StopEvent.size() == 2 + + # Eviction should not crash on nil timestamps + State.StopEvent.evict_old_records(7200) + + # Both should remain (one is recent, one has no timestamp to compare) + assert State.StopEvent.size() == 2 + end + end + + describe "automatic eviction on new_state" do + test "evicts old records automatically when configured" do + # Set a short retention period for this test + Application.put_env(:state, State.StopEvent, retention_seconds: 7200) + on_exit(fn -> Application.delete_env(:state, State.StopEvent) end) + + now = System.system_time(:second) + three_hours_ago = now - 10_800 + one_hour_ago = now - 3600 + + old_event = %StopEvent{ + id: "old-event", + vehicle_id: "v1", + start_date: ~D[2026-08-05], + trip_id: "trip1", + direction_id: 0, + route_id: "route1", + revenue: :REVENUE, + stop_id: "stop1", + stop_sequence: 1, + arrived: nil, + departed: nil, + timestamp: three_hours_ago + } + + State.StopEvent.new_state([old_event]) + # Old event is automatically evicted by post_commit_hook + assert State.StopEvent.size() == 0 + + # Add a new event - should trigger automatic eviction + recent_event = %StopEvent{ + id: "recent-event", + vehicle_id: "v2", + start_date: ~D[2026-08-05], + trip_id: "trip2", + direction_id: 0, + route_id: "route2", + revenue: :REVENUE, + stop_id: "stop2", + stop_sequence: 1, + arrived: nil, + departed: nil, + timestamp: one_hour_ago + } + + State.StopEvent.new_state([recent_event]) + + # Recent event should be kept + assert by_id("recent-event") != nil + assert State.StopEvent.size() == 1 end end end From 173bd1059c9d2f2352dbd9ebfd1436195d82c399 Mon Sep 17 00:00:00 2001 From: "Corey N. Runkel" Date: Thu, 6 Aug 2026 14:49:09 -0400 Subject: [PATCH 3/4] Consolidate tests and clean docs --- apps/parse/test/parse/stop_events_test.exs | 17 ----- apps/state/config/config.exs | 5 ++ apps/state/lib/state/stop_event.ex | 78 +++++++++------------- apps/state/test/state/stop_event_test.exs | 66 +++++++++--------- 4 files changed, 72 insertions(+), 94 deletions(-) diff --git a/apps/parse/test/parse/stop_events_test.exs b/apps/parse/test/parse/stop_events_test.exs index 49cfdda33..28ddcbe77 100644 --- a/apps/parse/test/parse/stop_events_test.exs +++ b/apps/parse/test/parse/stop_events_test.exs @@ -204,27 +204,11 @@ defmodule Parse.StopEventsTest do result = parse(ndjson, newer_than: 1_771_968_343) - # Only the record with timestamp > 1771968343 should be included assert {:partial, events} = result assert length(events) == 1 assert [%StopEvent{id: "after-boundary"}] = events end - test "works with gzipped data and newer_than option" do - ndjson = """ - {"id":"old-1","timestamp":1771968300,"start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} - {"id":"new-1","timestamp":1771968350,"start_date":"20260224","trip_id":"test2","vehicle_id":"v2","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop2","stop_sequence":2,"arrived":1771966486,"departed":1771967246} - """ - - gzipped = :zlib.gzip(ndjson) - - result = parse(gzipped, newer_than: 1_771_968_343) - - assert {:partial, events} = result - assert length(events) == 1 - assert [%StopEvent{id: "new-1"}] = events - end - test "handles records with missing timestamp field by excluding them" do ndjson = """ {"id":"no-timestamp","start_date":"20260224","trip_id":"test1","vehicle_id":"v1","direction_id":0,"route_id":"1","revenue":true,"stop_id":"stop1","stop_sequence":1,"arrived":1771966486,"departed":1771967246} @@ -233,7 +217,6 @@ defmodule Parse.StopEventsTest do result = parse(ndjson, newer_than: 1_771_968_343) - # Records without timestamp are excluded when filtering is enabled assert {:partial, events} = result assert length(events) == 1 assert [%StopEvent{id: "with-timestamp"}] = events diff --git a/apps/state/config/config.exs b/apps/state/config/config.exs index 16a7f3806..d6c27a1fa 100644 --- a/apps/state/config/config.exs +++ b/apps/state/config/config.exs @@ -867,4 +867,9 @@ config :state, :stops_on_route, ] } +# Configuration for StopEvent state +config :state, State.StopEvent, + # Retention period in seconds (default: 2 hours = 7200 seconds) + retention_seconds: 7200 + import_config "#{config_env()}.exs" diff --git a/apps/state/lib/state/stop_event.ex b/apps/state/lib/state/stop_event.ex index 05bb923b8..941759d4c 100644 --- a/apps/state/lib/state/stop_event.ex +++ b/apps/state/lib/state/stop_event.ex @@ -40,7 +40,7 @@ defmodule State.StopEvent do rescue e -> # log_parse_error returns nil, so we return nil on error to avoid passing - # invalid data to super/1. The caller (State.Server) expects nil to mean + # invalid data to super/1. State.Server expects nil to mean # "no data to insert" and will handle it gracefully. State.Server.log_parse_error(__MODULE__, e) end @@ -65,11 +65,9 @@ defmodule State.StopEvent do ## Parameters - * `retention_seconds` - Number of seconds to retain records. Defaults to configured - value (default: 7200 seconds / 2 hours). Records with timestamps older than - `now - retention_seconds` will be deleted. - - Records without timestamps are not evicted. + * `retention_seconds` - Number of seconds to retain records. Defaults to + 7200 seconds / 2 hours). Records with timestamps older than `now - + retention_seconds` will be deleted. ## Examples @@ -80,52 +78,40 @@ defmodule State.StopEvent do evict_old_records(3600) """ - @spec evict_old_records(non_neg_integer()) :: :ok - def evict_old_records(retention_seconds \\ nil) do - retention_seconds = retention_seconds || get_retention_seconds() - now = System.system_time(:second) - cutoff = now - retention_seconds - - # Build match spec to find records with timestamp < cutoff - fields = StopEvent.fields() - timestamp_position = Enum.find_index(fields, &(&1 == :timestamp)) - id_position = Enum.find_index(fields, &(&1 == :id)) - - unless timestamp_position && id_position do - raise "timestamp or id field not found in StopEvent struct" - end - - # Build pattern with both timestamp and id as variables - wildcards = List.duplicate(:_, length(fields)) - - pattern_list = [ - StopEvent - | wildcards - |> List.replace_at(timestamp_position, :"$1") - |> List.replace_at(id_position, :"$2") - ] - - pattern = List.to_tuple(pattern_list) + @spec evict_old_records() :: :ok + def evict_old_records do + evict_old_records(get_retention_seconds()) + end - # Match spec: select ids where timestamp < cutoff and timestamp is not nil - # Guards: timestamp is not nil AND timestamp < cutoff - match_spec = [ - {pattern, - [ - {:andalso, {:"/=", :"$1", nil}, {:<, :"$1", cutoff}} - ], [:"$2"]} - ] + @spec evict_old_records(non_neg_integer()) :: :ok + def evict_old_records(retention_seconds) when retention_seconds >= 0 do + cutoff = System.system_time(:second) - retention_seconds + match_spec = build_eviction_match_spec(cutoff) - ids_to_delete = :mnesia.dirty_select(__MODULE__, match_spec) - - # Delete each record by id - Enum.each(ids_to_delete, fn id -> - :mnesia.dirty_delete(__MODULE__, id) - end) + __MODULE__ + |> :mnesia.dirty_select(match_spec) + |> Enum.each(&:mnesia.dirty_delete(__MODULE__, &1)) :ok end + defp build_eviction_match_spec(cutoff) do + fields = StopEvent.fields() + timestamp_pos = Enum.find_index(fields, &(&1 == :timestamp)) + id_pos = Enum.find_index(fields, &(&1 == :id)) + + pattern = + :_ + |> List.duplicate(length(fields)) + |> List.replace_at(timestamp_pos, :"$1") + |> List.replace_at(id_pos, :"$2") + |> then(&[StopEvent | &1]) + |> List.to_tuple() + + # Match spec format: {pattern, guards, result} + [{pattern, [{:andalso, {:"/=", :"$1", nil}, {:<, :"$1", cutoff}}], [:"$2"]}] + end + # Get the maximum timestamp from existing data in the table. # Uses a dynamic match spec based on the StopEvent struct to avoid brittleness # from hardcoded field positions. diff --git a/apps/state/test/state/stop_event_test.exs b/apps/state/test/state/stop_event_test.exs index 949199826..d22fcf979 100644 --- a/apps/state/test/state/stop_event_test.exs +++ b/apps/state/test/state/stop_event_test.exs @@ -400,38 +400,42 @@ defmodule State.StopEventTest do end describe "partial updates with timestamps" do - test "partial update only affects records with matching keys" do - # Initial full state - initial_events = [ - %StopEvent{ - id: "trip1-route1-v1-1", - vehicle_id: "v1", - start_date: ~D[2026-02-24], - trip_id: "trip1", - direction_id: 0, - route_id: "route1", - revenue: :REVENUE, - stop_id: "stop1", - stop_sequence: 1, - arrived: ~U[2026-02-24 15:28:06Z], - departed: ~U[2026-02-24 15:40:46Z] - }, - %StopEvent{ - id: "trip1-route1-v1-2", - vehicle_id: "v1", - start_date: ~D[2026-02-24], - trip_id: "trip1", - direction_id: 0, - route_id: "route1", - revenue: :REVENUE, - stop_id: "stop2", - stop_sequence: 2, - arrived: ~U[2026-02-24 15:41:26Z], - departed: ~U[2026-02-24 15:42:13Z] - } - ] + setup do + stop_event1 = %StopEvent{ + id: "trip1-route1-v1-1", + vehicle_id: "v1", + start_date: ~D[2026-02-24], + trip_id: "trip1", + direction_id: 0, + route_id: "route1", + revenue: :REVENUE, + stop_id: "stop1", + stop_sequence: 1, + arrived: ~U[2026-02-24 15:28:06Z], + departed: ~U[2026-02-24 15:40:46Z] + } + + stop_event2 = %StopEvent{ + id: "trip1-route1-v1-2", + vehicle_id: "v1", + start_date: ~D[2026-02-24], + trip_id: "trip1", + direction_id: 0, + route_id: "route1", + revenue: :REVENUE, + stop_id: "stop2", + stop_sequence: 2, + arrived: ~U[2026-02-24 15:41:26Z], + departed: ~U[2026-02-24 15:42:13Z] + } + + State.StopEvent.new_state([stop_event1, stop_event2]) + + {:ok, %{event1: stop_event1, event2: stop_event2}} + end - State.StopEvent.new_state(initial_events) + test "partial update only affects records with matching keys", %{event1: _e1, event2: _e2} do + # Initial state already set up with 2 events assert State.StopEvent.size() == 2 # Partial update - replace event with id trip1-route1-v1-1 and add new event From ffe2b7fa04a705055b7e60f5e994faeba71f0baf Mon Sep 17 00:00:00 2001 From: Corey Runkel <39202587+runkelcorey@users.noreply.github.com> Date: Thu, 6 Aug 2026 17:10:29 -0400 Subject: [PATCH 4/4] Apply suggestions from code review Co-authored-by: Eddie Maldonado --- apps/state/test/state/stop_event_test.exs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/apps/state/test/state/stop_event_test.exs b/apps/state/test/state/stop_event_test.exs index d22fcf979..7dc9c1821 100644 --- a/apps/state/test/state/stop_event_test.exs +++ b/apps/state/test/state/stop_event_test.exs @@ -623,9 +623,9 @@ defmodule State.StopEventTest do # Old event should be evicted, borderline is exactly at boundary (should be kept), # recent should remain assert State.StopEvent.size() == 2 - assert by_id("old-event") == nil - assert by_id("borderline-event") != nil - assert by_id("recent-event") != nil + assert is_nil(by_id("old-event")) + refute is_nil(by_id("borderline-event")) + refute is_nil(by_id("recent-event")) end test "evicts records using custom retention period" do @@ -670,8 +670,8 @@ defmodule State.StopEventTest do State.StopEvent.evict_old_records(2400) assert State.StopEvent.size() == 1 - assert by_id("old-event") == nil - assert by_id("recent-event") != nil + assert is_nil(by_id("old-event")) + refute is_nil(by_id("recent-event")) end test "handles records without timestamps gracefully" do