fix stream on deploys

eigr · Feb 24, 2025 · 2c638aa · 2c638aa
1 parent d1a5941
commit 2c638aa
Show file tree

Hide file tree

Showing 7 changed files with 28 additions and 46 deletions.
diff --git a/lib/actors/actor/caller_consumer.ex b/lib/actors/actor/caller_consumer.ex
@@ -956,10 +956,7 @@ defmodule Actors.Actor.CallerConsumer do
       kind == :UNNAMED ->
         false
 
-      match?(true, stateful) ->
-        true
-
-      not is_nil(channel_group) and length(channel_group) > 0 ->
+      stateful == true ->
         true
 
       true ->

diff --git a/lib/actors/actor/entity/lifecycle.ex b/lib/actors/actor/entity/lifecycle.ex
@@ -78,7 +78,7 @@ defmodule Actors.Actor.Entity.Lifecycle do
         actor.id
       )
 
-    schedule_deactivate(deactivation_strategy, get_jitter())
+    schedule_deactivate(actor, deactivation_strategy, get_jitter())
 
     state =
       case maybe_schedule_snapshot_advance(snapshot_strategy) do
@@ -252,7 +252,7 @@ defmodule Actors.Actor.Entity.Lifecycle do
                   %ActorDeactivationStrategy{strategy: deactivation_strategy} =
                     _actor_deactivation_strategy
               }
-            } = _actor
+            } = actor
         } = state
       ) do
     queue_length = Process.info(self(), :message_queue_len)
@@ -265,7 +265,7 @@ defmodule Actors.Actor.Entity.Lifecycle do
         {:stop, :shutdown, state}
 
       _ ->
-        schedule_deactivate(deactivation_strategy)
+        schedule_deactivate(actor, deactivation_strategy)
         {:noreply, state}
     end
   end
@@ -423,7 +423,16 @@ defmodule Actors.Actor.Entity.Lifecycle do
 
   defp maybe_schedule_snapshot_advance(_), do: :ok
 
-  defp schedule_deactivate(deactivation_strategy, timeout_factor \\ 0) do
+  defp schedule_deactivate(actor, deactivation_strategy, timeout_factor \\ 0)
+
+  defp schedule_deactivate(
+         %Actor{settings: %ActorSettings{kind: :PROJECTION}},
+         deactivation_strategy,
+         timeout_factor
+       ),
+       do: :ok
+
+  defp schedule_deactivate(_actor, deactivation_strategy, timeout_factor) do
     strategy = maybe_get_default_deactivation_strategy(deactivation_strategy)
 
     Process.send_after(

diff --git a/lib/actors/actor/entity/lifecycle/projection_consumers.ex b/lib/actors/actor/entity/lifecycle/projection_consumers.ex
diff --git a/lib/actors/actor/entity/lifecycle/stream_consumer.ex b/lib/actors/actor/entity/lifecycle/stream_consumer.ex
@@ -6,6 +6,7 @@ defmodule Actors.Actor.Entity.Lifecycle.StreamConsumer do
   alias Spawn.Utils.Nats
   alias Spawn.Fact
   alias Google.Protobuf.Timestamp
+  alias Sidecar.GracefulShutdown
 
   @type fact :: %Fact{}
 
@@ -47,9 +48,14 @@ defmodule Actors.Actor.Entity.Lifecycle.StreamConsumer do
 
   @spec handle_message(any(), Message.t(), any()) :: Message.t()
   def handle_message(_processor_name, message, _context) do
-    message
-    |> build_fact()
-    |> Message.configure_ack(on_success: :term)
+    if GracefulShutdown.running?() do
+      message
+      |> build_fact()
+      |> Message.configure_ack(on_success: :term)
+    else
+      message
+      |> Message.failed("Failed to deliver because app is draining")
+    end
   end
 
   @spec handle_batch(any(), Message.t(), any(), opts()) :: list(Message.t())

diff --git a/lib/actors/actor/entity/lifecycle/stream_initiator.ex b/lib/actors/actor/entity/lifecycle/stream_initiator.ex
@@ -4,7 +4,7 @@ defmodule Actors.Actor.Entity.Lifecycle.StreamInitiator do
   """
   require Logger
 
-  alias Actors.Actor.Entity.Lifecycle.ProjectionConsumers
+  alias Actors.Actor.Entity.Lifecycle.StreamConsumer
 
   alias Spawn.Actors.Actor
   alias Spawn.Actors.ProjectionSettings
@@ -245,7 +245,7 @@ defmodule Actors.Actor.Entity.Lifecycle.StreamInitiator do
   end
 
   defp start_pipeline(actor) do
-    ProjectionConsumers.new(%{
+    StreamConsumer.start_link(%{
       actor_name: stream_name(actor),
       projection_pid: self(),
       strict_ordering: actor.settings.projection_settings.strict_events_ordering

diff --git a/lib/actors/supervisors/actor_supervisor.ex b/lib/actors/supervisors/actor_supervisor.ex
@@ -43,11 +43,9 @@ defmodule Actors.Supervisors.ActorSupervisor do
         supervisor_process_logger(__MODULE__),
         get_pubsub_adapter(opts),
         Actors.Actor.Pubsub,
-        Actors.Actor.Entity.Lifecycle.ProjectionConsumers,
         Actors.Actor.Entity.Supervisor.child_spec(opts)
       ] ++
-        maybe_add_invocation_scheduler(opts) ++
-        [{CallerProducer, []}] ++ consumers
+        maybe_add_invocation_scheduler(opts) ++ [{CallerProducer, []}] ++ consumers
 
     Supervisor.init(children, strategy: :one_for_one)
   end

diff --git a/lib/sidecar/supervisor.ex b/lib/sidecar/supervisor.ex
@@ -17,8 +17,8 @@ defmodule Sidecar.Supervisor do
       [
         supervisor_process_logger(__MODULE__),
         {Sidecar.GracefulShutdown, opts},
-        {Sidecar.ProcessSupervisor, opts},
-        {Sidecar.GRPC.Supervisor, opts}
+        {Sidecar.GRPC.Supervisor, opts},
+        {Sidecar.ProcessSupervisor, opts}
       ]
       |> Enum.reject(&is_nil/1)