fix(vehicle): keep the stream connected while a drive is interrupted by an API outage

When a fetch reported the vehicle offline or asleep, the stream was always
torn down, including mid-drive. There the vehicle is usually only
unreachable for the API for a while, and the stream keeps delivering
positions; those were discarded until the next successful fetch, leaving a
hole in the drive. #5535 already reconnects the stream once the vehicle is
back online, but everything streamed in between was lost.

Keep the stream while the state machine is in a driving state and record
D/N/R frames into the open drive also while it is unavailable or offline.
The fetch scheduled by the offline handlers stays the only way back to
:available; other frames no longer schedule a fetch in those states, since
every one of them would fail the same way.

Author: onevcat

lib/teslamate/vehicles/vehicle.ex

@@ -482,10 +482,24 @@ defmodule TeslaMate.Vehicles.Vehicle do
             end
         end
 
-      {:ok, %Vehicle{state: state} = vehicle} when state in ["offline", "asleep"] ->
+      {:ok, %Vehicle{state: vehicle_state} = vehicle}
+      when vehicle_state in ["offline", "asleep"] ->
         # disconnect stream in case we started it to detect real online
         # (in that case we won't go through Start / :offline or Start / :asleep)
-        :ok = disconnect_stream(data)
+        #
+        # Not while driving: there the vehicle is usually only unreachable for
+        # the API for a while, and the stream keeps delivering positions that
+        # belong to the running drive. The driving handlers keep polling and
+        # decide when the drive is over.
+        data =
+          case state do
+            {:driving, _status, _drive} ->
+              data
+
+            _ ->
+              :ok = disconnect_stream(data)
+              %Data{data | stream_pid: nil}
+          end
 
         %Data{} =
           data =
@@ -494,10 +508,10 @@ defmodule TeslaMate.Vehicles.Vehicle do
             %Data{data | last_response: last_response, geofence: geofence}
           end
 
-        {:keep_state, %Data{data | pre_online_check: :idle, stream_pid: nil},
+        {:keep_state, %Data{data | pre_online_check: :idle},
          [
            broadcast_fetch(false),
-           {:next_event, :internal, {:update, {String.to_existing_atom(state), vehicle}}}
+           {:next_event, :internal, {:update, {String.to_existing_atom(vehicle_state), vehicle}}}
          ]}
 
       {:ok, %Vehicle{state: state} = vehicle} ->
@@ -718,7 +732,11 @@ defmodule TeslaMate.Vehicles.Vehicle do
         %Data{} = data
       ) do
     case {status, stream_data} do
-      {:available, %Stream.Data{shift_state: shift_state}} when shift_state in ~w(D N R) ->
+      # Also while the API reports the vehicle unavailable or offline: the
+      # stream stays connected then, and only a successful fetch brings the
+      # status back to :available.
+      {_status, %Stream.Data{shift_state: shift_state}}
+      when shift_state in ~w(D N R) and not is_nil(drv) ->
         {elevation, geofence} =
           Repo.checkout(fn ->
             {:ok, %{elevation: elevation} = position} =
@@ -740,8 +758,13 @@ defmodule TeslaMate.Vehicles.Vehicle do
              geofence: geofence
          }, broadcast_summary()}
 
-      {_status, %Stream.Data{}} ->
+      {:available, %Stream.Data{}} ->
         {:keep_state_and_data, schedule_fetch(0, data)}
+
+      {_status, %Stream.Data{}} ->
+        # The offline handlers have a fetch scheduled already; one per frame
+        # would only add vehicle_data requests that are known to fail.
+        :keep_state_and_data
     end
   end
 

test/teslamate/vehicles/vehicle/driving_test.exs

@@ -458,15 +458,16 @@ defmodule TeslaMate.Vehicles.Vehicle.DrivingTest do
       assert_receive :continue?
       refute_receive _
       send(:"api_#{name}", :continue)
-      assert_receive {:"$websockex_cast", :disconnect}
       assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :offline}}}
+      refute_received {:"$websockex_cast", :disconnect}
 
       # Logs previous drive
       assert_receive {:close_drive, ^drive, lookup_address: true}, 250
 
+      # The stream was kept through the offline period, nothing to reconnect
       d1 = DateTime.from_unix!(now_ts + 1 + :timer.minutes(15), :millisecond)
       assert_receive {:start_state, ^car, :online, date: ^d1}
-      assert_receive {ApiMock, {:stream, 1000, _}}
+      refute_received {ApiMock, {:stream, 1000, _}}
       assert_receive {:insert_position, ^car, %{}}
       assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :online}}}
 
@@ -537,7 +538,7 @@ defmodule TeslaMate.Vehicles.Vehicle.DrivingTest do
     end
 
     @tag :capture_log
-    test "reconnects the stream when a drive resumes after an offline period",
+    test "keeps the stream connected through an offline period mid-drive",
          %{test: name} do
       now = DateTime.utc_now()
       now_ts = DateTime.to_unix(now, :millisecond)
@@ -574,14 +575,13 @@ defmodule TeslaMate.Vehicles.Vehicle.DrivingTest do
       assert_receive {:insert_position, drive, %{longitude: 0.1, speed: 48}}
       assert_receive {:insert_position, ^drive, %{longitude: 0.1, speed: 48}}
 
-      # Vehicle goes offline: the stream is disconnected
-      assert_receive {:"$websockex_cast", :disconnect}
+      # Vehicle goes offline: the stream stays connected
       assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :offline}}}, 500
+      refute_received {:"$websockex_cast", :disconnect}
 
-      # Vehicle comes back online mid-drive: the stream is reconnected
-      assert_receive {ApiMock, {:stream, 1000, _}}, 500
-
-      assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :driving}}}
+      # Vehicle comes back online mid-drive: nothing to reconnect
+      assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :driving}}}, 500
+      refute_received {ApiMock, {:stream, 1000, _}}
       assert_receive {:insert_position, drive, %{longitude: 0.2, speed: 32}}
       assert_receive {:insert_position, ^drive, %{longitude: 0.3}}
       assert_receive {:close_drive, ^drive, lookup_address: true}
@@ -595,7 +595,7 @@ defmodule TeslaMate.Vehicles.Vehicle.DrivingTest do
     end
 
     @tag :capture_log
-    test "reconnects the stream when a drive resumes after a short unavailability",
+    test "keeps the stream connected through a short unavailability mid-drive",
          %{test: name} do
       now = DateTime.utc_now()
       now_ts = DateTime.to_unix(now, :millisecond)
@@ -632,13 +632,11 @@ defmodule TeslaMate.Vehicles.Vehicle.DrivingTest do
       assert_receive {:insert_position, drive, %{longitude: 0.1, speed: 48}}
       assert_receive {:insert_position, ^drive, %{longitude: 0.1, speed: 48}}
 
-      # Vehicle becomes unavailable: the stream is disconnected
-      assert_receive {:"$websockex_cast", :disconnect}
-
-      # Vehicle comes back online mid-drive: the stream is reconnected
-      assert_receive {ApiMock, {:stream, 1000, _}}, 500
-
-      assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :driving}}}
+      # Vehicle becomes unavailable and comes back online mid-drive: the stream
+      # stays connected, nothing to reconnect
+      assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :driving}}}, 500
+      refute_received {:"$websockex_cast", :disconnect}
+      refute_received {ApiMock, {:stream, 1000, _}}
       assert_receive {:insert_position, drive, %{longitude: 0.2, speed: 32}}
       assert_receive {:insert_position, ^drive, %{longitude: 0.3}}
       assert_receive {:close_drive, ^drive, lookup_address: true}

test/teslamate/vehicles/vehicle/streaming_test.exs

@@ -157,6 +157,82 @@ defmodule TeslaMate.Vehicles.Vehicle.StreamingTest do
       refute_receive _
     end
 
+    @tag :capture_log
+    test "keeps recording streamed positions while the vehicle is unavailable for the API",
+         %{test: name} do
+      me = self()
+      now = DateTime.utc_now()
+      now_ts = DateTime.to_unix(now, :millisecond)
+
+      wait_for_continue = fn result ->
+        fn ->
+          send(me, :continue?)
+
+          receive do
+            :continue -> result
+          after
+            5_000 -> raise "No :continue after 5s"
+          end
+        end
+      end
+
+      events = [
+        {:ok, online_event(now_ts)},
+        {:ok, online_event(now_ts, drive_state: %{timestamp: now_ts})},
+        wait_for_continue.({:ok, %TeslaApi.Vehicle{state: "offline"}}),
+        wait_for_continue.(
+          {:ok,
+           online_event(now_ts + 10,
+             drive_state: %{
+               timestamp: now_ts + 10,
+               latitude: 42.5,
+               longitude: 42.0,
+               shift_state: "D",
+               speed: 20,
+               power: 8
+             }
+           )}
+        ),
+        fn -> Process.sleep(10_000) end
+      ]
+
+      :ok = start_vehicle(name, events)
+
+      assert_receive {:start_state, car, :online, date: _}
+      assert_receive {:insert_position, ^car, %{}}
+      assert_receive {ApiMock, {:stream, _eid, func}} when is_function(func)
+      assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :online}}}
+      assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :online}}}
+
+      assert_receive :continue?
+      stream(name, %{shift_state: "D", speed: 10, power: 5, time: now})
+      assert_receive {:start_drive, ^car}
+      assert_receive {:insert_position, drive, %{speed: 16}}
+      assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :driving, speed: 16}}}
+
+      # The API loses the vehicle mid-drive: the stream stays connected
+      send(:"api_#{name}", :continue)
+      assert_receive :continue?
+      refute_received {:"$websockex_cast", :disconnect}
+
+      # Streamed positions still belong to the drive
+      stream(name, %{shift_state: "D", speed: 15, power: 10, est_lat: 42.31, time: now})
+      assert_receive {:insert_position, ^drive, %{speed: 24, latitude: 42.31}}
+      assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :driving, speed: 24}}}
+
+      # Everything else is left to the fetch that is already scheduled
+      stream(name, %{shift_state: "P", speed: nil, power: nil, time: now})
+      refute_receive _
+
+      # The API sees the vehicle again: the drive continues on the same stream
+      send(:"api_#{name}", :continue)
+      assert_receive {:insert_position, ^drive, %{latitude: 42.5, speed: 32}}
+      assert_receive {:pubsub, {:broadcast, _, _, %Summary{state: :driving}}}
+      refute_received {ApiMock, {:stream, _, _}}
+
+      refute_receive _
+    end
+
     test "updates the geofence from streaming positions", %{test: name} do
       now = DateTime.utc_now()
       now_ts = DateTime.to_unix(now, :millisecond)