perf(ScheduleController): Fetch schedules by individual stop for more cache hits

config/config.exs

@@ -20,6 +20,10 @@ config :mobile_app_backend, MBTAV3API.ResponseCache,
   gc_interval: :timer.hours(1),
   allocated_memory: 250_000_000
 
+config :mobile_app_backend, MBTAV3API.RepositoryCache,
+  gc_interval: :timer.hours(2),
+  allocated_memory: 250_000_000
+
 # Configures the endpoint
 config :mobile_app_backend, MobileAppBackendWeb.Endpoint,
   url: [host: "localhost"],

lib/mbta_v3_api/repository.ex

@@ -67,24 +67,34 @@ end
 
 defmodule MBTAV3API.Repository.Impl do
   @behaviour MBTAV3API.Repository
+
+  use Nebulex.Caching.Decorators
+
   alias MBTAV3API.JsonApi
 
+  @ttl :timer.hours(1)
+
   @impl true
   def alerts(params, opts \\ []), do: all(MBTAV3API.Alert, params, opts)
 
   @impl true
+  @decorate cacheable(cache: MBTAV3API.RepositoryCache, on_error: :nothing, opts: [ttl: @ttl])
   def route_patterns(params, opts \\ []), do: all(MBTAV3API.RoutePattern, params, opts)
 
   @impl true
+  @decorate cacheable(cache: MBTAV3API.RepositoryCache, on_error: :nothing, opts: [ttl: @ttl])
   def routes(params, opts \\ []), do: all(MBTAV3API.Route, params, opts)
 
   @impl true
+  @decorate cacheable(cache: MBTAV3API.RepositoryCache, on_error: :nothing, opts: [ttl: @ttl])
   def schedules(params, opts \\ []), do: all(MBTAV3API.Schedule, params, opts)
 
   @impl true
+  @decorate cacheable(cache: MBTAV3API.RepositoryCache, on_error: :nothing, opts: [ttl: @ttl])
   def stops(params, opts \\ []), do: all(MBTAV3API.Stop, params, opts)
 
   @impl true
+  @decorate cacheable(cache: MBTAV3API.RepositoryCache, on_error: :nothing, opts: [ttl: @ttl])
   def trips(params, opts \\ []), do: all(MBTAV3API.Trip, params, opts)
 
   @spec all(module(), JsonApi.Params.t(), Keyword.t()) ::

lib/mbta_v3_api/repository_cache.ex

@@ -0,0 +1,6 @@
+defmodule MBTAV3API.RepositoryCache do
+  @moduledoc """
+  Cache used to reduce the number of calls to the V3 API.
+  """
+  use Nebulex.Cache, otp_app: :mobile_app_backend, adapter: Nebulex.Adapters.Local
+end

lib/mobile_app_backend/application.ex

@@ -26,6 +26,7 @@ defmodule MobileAppBackend.Application do
            :default => [size: 200, count: 10, start_pool_metrics?: true]
          }},
         {MBTAV3API.ResponseCache, []},
+        {MBTAV3API.RepositoryCache, []},
         MBTAV3API.Supervisor,
         {MobileAppBackend.FinchPoolHealth, pool_name: Finch.CustomPool},
         MobileAppBackend.MapboxTokenRotator,

lib/mobile_app_backend_web/controllers/schedule_controller.ex

@@ -1,17 +1,34 @@
 defmodule MobileAppBackendWeb.ScheduleController do
   use MobileAppBackendWeb, :controller
+  require Logger
   alias MBTAV3API.JsonApi
   alias MBTAV3API.Repository
 
-  def schedules(conn, %{"stop_ids" => stop_ids, "date_time" => date_time}) do
-    if stop_ids == "" do
+  def schedules(conn, %{"stop_ids" => stop_ids_concat, "date_time" => date_time}) do
+    if stop_ids_concat == "" do
       json(conn, %{schedules: [], trips: %{}})
     else
-      {:ok, data} =
-        get_filter(stop_ids, date_time)
-        |> fetch_schedules()
+      stop_ids = String.split(stop_ids_concat, ",")
 
-      json(conn, data)
+      service_date = parse_service_date(date_time)
+
+      filters = Enum.map(stop_ids, &get_filter(&1, service_date))
+
+      data =
+        case filters do
+          [filter] -> fetch_schedules(filter)
+          filters -> fetch_schedules_parallel(filters)
+        end
+
+      case data do
+        :error ->
+          conn
+          |> put_status(:internal_server_error)
+          |> json(%{error: "fetch_failed"})
+
+        data ->
+          json(conn, data)
+      end
     end
   end
 
@@ -39,20 +56,50 @@ defmodule MobileAppBackendWeb.ScheduleController do
     json(conn, response)
   end
 
-  @spec get_filter(String.t(), String.t()) :: [JsonApi.Params.filter_param()]
-  defp get_filter(stop_ids, date_time) do
-    date_time = Util.parse_datetime!(date_time)
-    service_date = Util.datetime_to_gtfs(date_time)
-    [stop: stop_ids, date: service_date]
+  @spec parse_service_date(String.t()) :: Date.t()
+  defp parse_service_date(date_string) do
+    date_string
+    |> Util.parse_datetime!()
+    |> Util.datetime_to_gtfs()
+  end
+
+  @spec get_filter(String.t(), Date.t()) :: [JsonApi.Params.filter_param()]
+  defp get_filter(stop_id, service_date) do
+    [stop: stop_id, date: service_date]
+  end
+
+  @spec fetch_schedules_parallel([[JsonApi.Params.filter_param()]]) ::
+          %{schedules: [MBTAV3API.Schedule.t()], trips: JsonApi.Object.trip_map()} | :error
+  defp fetch_schedules_parallel(filters) do
+    filters
+    |> Task.async_stream(fn filter_params ->
+      {filter_params, fetch_schedules(filter_params)}
+    end)
+    |> Enum.reduce_while(%{schedules: [], trips: %{}}, fn result, acc ->
+      case result do
+        {:ok, {_params, %{schedules: schedules, trips: trips}}} ->
+          {:cont, %{schedules: acc.schedules ++ schedules, trips: Map.merge(acc.trips, trips)}}
+
+        {_result_type, {params, _response}} ->
+          Logger.warning(
+            "#{__MODULE__} skipped returning schedules due to error. params=#{inspect(params)}"
+          )
+
+          {:halt, :error}
+      end
+    end)
   end
 
   @spec fetch_schedules([JsonApi.Params.filter_param()]) ::
-          {:ok, %{schedules: [MBTAV3API.Schedule.t()], trips: JsonApi.Object.trip_map()}}
-          | {:error, term()}
+          %{schedules: [MBTAV3API.Schedule.t()], trips: JsonApi.Object.trip_map()}
+          | :error
   defp fetch_schedules(filter) do
-    with {:ok, %{data: schedules, included: %{trips: trips}}} <-
-           Repository.schedules(filter: filter, include: :trip, sort: {:departure_time, :asc}) do
-      {:ok, %{schedules: schedules, trips: trips}}
+    case Repository.schedules(filter: filter, include: :trip, sort: {:departure_time, :asc}) do
+      {:ok, %{data: schedules, included: %{trips: trips}}} ->
+        %{schedules: schedules, trips: trips}
+
+      _ ->
+        :error
     end
   end
 end

mix.exs

@@ -55,6 +55,7 @@ defmodule MobileAppBackend.MixProject do
       {:esbuild, "~> 0.7", runtime: Mix.env() == :dev},
       {:tailwind, "~> 0.2.0", runtime: Mix.env() == :dev},
       {:logster, "~> 1.1"},
+      {:decorator, "~> 1.4"},
       {:diskusage_logger, "~> 0.2", only: :prod},
       {:ehmon, github: "mbta/ehmon", only: :prod},
       {:sobelow, "~> 0.13", only: [:dev, :test], runtime: false},

mix.lock

@@ -8,6 +8,7 @@
   "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.4.0", "f239f68b588efa7707abce16a84d0d2acf3a0f50571f8bb7f56a15865aae820c", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7d98bac1ee4565d31b62d59f8823dfd8356a169e7fcbb83831b8a5397404c9de"},
   "cowlib": {:hex, :cowlib, "2.13.0", "db8f7505d8332d98ef50a3ef34b34c1afddec7506e4ee4dd4a3a266285d282ca", [:make, :rebar3], [], "hexpm", "e1e1284dc3fc030a64b1ad0d8382ae7e99da46c3246b815318a4b848873800a4"},
   "credo": {:hex, :credo, "1.7.9", "07bb31907746ae2b5e569197c9e16c0d75c8578a22f01bee63f212047efb2647", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "f87c11c34ba579f7c5044f02b2a807e1ed2fa5fdbb24dc7eb4ad59c1904887f3"},
+  "decorator": {:hex, :decorator, "1.4.0", "a57ac32c823ea7e4e67f5af56412d12b33274661bb7640ec7fc882f8d23ac419", [:mix], [], "hexpm", "0a07cedd9083da875c7418dea95b78361197cf2bf3211d743f6f7ce39656597f"},
   "dialyxir": {:hex, :dialyxir, "1.4.4", "fb3ce8741edeaea59c9ae84d5cec75da00fa89fe401c72d6e047d11a61f65f70", [:mix], [{:erlex, ">= 0.2.7", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "cd6111e8017ccd563e65621a4d9a4a1c5cd333df30cebc7face8029cacb4eff6"},
   "diskusage_logger": {:hex, :diskusage_logger, "0.2.0", "04fc48b538fe4de43153542a71ea94f623d54707d85844123baacfceedf625c3", [:mix], [], "hexpm", "e3f2aed1b0fc4590931c089a6453a4c4eb4c945912aa97bcabcc0cff7851f34d"},
   "dns_cluster": {:hex, :dns_cluster, "0.1.3", "0bc20a2c88ed6cc494f2964075c359f8c2d00e1bf25518a6a6c7fd277c9b0c66", [:mix], [], "hexpm", "46cb7c4a1b3e52c7ad4cbe33ca5079fbde4840dedeafca2baf77996c2da1bc33"},

test/mbta_v3_api/repository_test.exs

@@ -3,12 +3,18 @@ defmodule MBTAV3API.RepositoryTest do
 
   import Mox
 
-  alias MBTAV3API.Route
-  alias MBTAV3API.{Alert, Repository, RoutePattern, Stop}
+  alias MBTAV3API.JsonApi
+  alias MBTAV3API.JsonApi.Object
+  alias MBTAV3API.{Alert, Repository, Route, RoutePattern, Schedule, Stop}
   import Test.Support.Sigils
 
   setup :verify_on_exit!
 
+  setup do
+    MBTAV3API.RepositoryCache.flush()
+    :ok
+  end
+
   test "alerts/2" do
     expect(
       MobileAppBackend.HTTPMock,
@@ -428,4 +434,150 @@ defmodule MBTAV3API.RepositoryTest do
               ]
             }} = Repository.stops([])
   end
+
+  describe "schedules/2" do
+    test "returns cached response when given same request twice" do
+      expect(
+        MobileAppBackend.HTTPMock,
+        :request,
+        # Only called once because first response is cached
+        1,
+        fn %Req.Request{url: %URI{path: "/schedules"}, options: %{params: _params}} ->
+          {:ok,
+           Req.Response.json(%{
+             data: [
+               %{
+                 "attributes" => %{
+                   "arrival_time" => "2024-03-13T01:07:00-04:00",
+                   "departure_time" => "2024-03-13T01:07:00-04:00",
+                   "drop_off_type" => 0,
+                   "id" => "schedule-60565179-70159-90",
+                   "pickup_type" => 0,
+                   "route_id" => "Green-B",
+                   "stop_id" => "70159",
+                   "stop_sequence" => 90,
+                   "trip_id" => "trip_1"
+                 },
+                 "id" => "sched_1",
+                 "relationships" => %{
+                   "trip" => %{
+                     "data" => %{
+                       "id" => "trip_1",
+                       "type" => "trip"
+                     }
+                   }
+                 },
+                 "type" => "schedule"
+               }
+             ],
+             included: [
+               %{
+                 "attributes" => %{
+                   "headsign" => "Headsign",
+                   "direction_id" => 1
+                 },
+                 "id" => "trip_1",
+                 "type" => "trip"
+               }
+             ]
+           })}
+        end
+      )
+
+      assert {:ok,
+              %{
+                data: [
+                  %Schedule{
+                    id: "sched_1"
+                  }
+                ],
+                included: %{trips: %{"trip_1" => %{id: "trip_1"}}}
+              }} = Repository.schedules([])
+
+      assert {:ok,
+              %{
+                data: [
+                  %Schedule{
+                    id: "sched_1"
+                  }
+                ],
+                included: %{trips: %{"trip_1" => %{id: "trip_1"}}}
+              }} = Repository.schedules([])
+    end
+
+    test "makes new request when new params passed" do
+      expect(
+        MobileAppBackend.HTTPMock,
+        :request,
+        fn %Req.Request{url: %URI{path: "/schedules"}, options: %{params: _params}} ->
+          {:ok,
+           Req.Response.json(%{
+             data: [
+               %{
+                 "attributes" => %{
+                   "arrival_time" => "2024-03-13T01:07:00-04:00",
+                   "departure_time" => "2024-03-13T01:07:00-04:00",
+                   "drop_off_type" => 0,
+                   "id" => "schedule-60565179-70159-90",
+                   "pickup_type" => 0,
+                   "route_id" => "Green-B",
+                   "stop_id" => "70159",
+                   "stop_sequence" => 90,
+                   "trip_id" => "trip_1"
+                 },
+                 "id" => "sched_1",
+                 "relationships" => %{
+                   "trip" => %{
+                     "data" => %{
+                       "id" => "trip_1",
+                       "type" => "trip"
+                     }
+                   }
+                 },
+                 "type" => "schedule"
+               }
+             ],
+             included: [
+               %{
+                 "attributes" => %{
+                   "headsign" => "Headsign",
+                   "direction_id" => 1
+                 },
+                 "id" => "trip_1",
+                 "type" => "trip"
+               }
+             ]
+           })}
+        end
+      )
+
+      expect(
+        MobileAppBackend.HTTPMock,
+        :request,
+        fn %Req.Request{url: %URI{path: "/schedules"}, options: %{params: _params}} ->
+          {:ok,
+           Req.Response.json(%{
+             data: [],
+             included: []
+           })}
+        end
+      )
+
+      assert {:ok,
+              %{
+                data: [
+                  %Schedule{
+                    id: "sched_1"
+                  }
+                ],
+                included: %{trips: %{"trip_1" => %{id: "trip_1"}}}
+              }} = Repository.schedules([])
+
+      assert {:ok,
+              %JsonApi.Response{
+                data: [],
+                included: Object.to_full_map([])
+              }} == Repository.schedules(filter: [stop: "fake_stop"])
+    end
+  end
 end