🗑️ Remove storage param from SlotMessageStore

Author: acco

config/runtime.exs

@@ -232,8 +232,7 @@ if config_env() == :prod and self_hosted do
     api_base_url: "http://#{server_host}:#{server_port}",
     release_version: System.get_env("RELEASE_VERSION"),
     backfill_max_pending_messages: backfill_max_pending_messages,
-    max_memory_bytes: ConfigParser.max_memory_bytes(env_vars),
-    default_max_storage_bytes: ConfigParser.default_max_storage_bytes(env_vars)
+    max_memory_bytes: ConfigParser.max_memory_bytes(env_vars)
 end
 
 if config_env() == :prod and not self_hosted do

lib/sequin/config_parser.ex

@@ -89,6 +89,7 @@ defmodule Sequin.ConfigParser do
 
   defp validate_log_level(level, default_level) do
     Logger.warning("[ConfigParser] Invalid log level: #{inspect(level)}. Using default #{inspect(default_level)} level.")
+
     default_level
   end
 
@@ -181,18 +182,6 @@ defmodule Sequin.ConfigParser do
     end
   end
 
-  def default_max_storage_bytes(env) do
-    case env["DEFAULT_MAX_STORAGE_MB"] do
-      nil ->
-        nil
-
-      mb_str ->
-        mb_str
-        |> String.to_integer()
-        |> Sequin.Size.mb()
-    end
-  end
-
   defp parse_buffer_percent(env) do
     env
     |> Map.get("MEMORY_BUFFER_PERCENT", "20")

lib/sequin/consumers/consumers.ex

@@ -284,17 +284,6 @@ defmodule Sequin.Consumers do
     round(Sequin.Size.mb(consumer.max_memory_mb) * 0.8)
   end
 
-  @doc """
-  Calculates the maximum storage bytes allowed for a consumer.
-  """
-  @spec max_storage_bytes_for_consumer(SinkConsumer.t()) ::
-          non_neg_integer() | nil
-  def max_storage_bytes_for_consumer(%SinkConsumer{max_storage_mb: nil}), do: nil
-
-  def max_storage_bytes_for_consumer(%SinkConsumer{max_storage_mb: max_storage_mb}) do
-    round(Sequin.Size.mb(max_storage_mb) * 0.8)
-  end
-
   def earliest_sink_consumer_inserted_at_for_account(account_id) do
     account_id
     |> SinkConsumer.where_account_id()

lib/sequin/consumers/sink_consumer.ex

@@ -48,7 +48,6 @@ defmodule Sequin.Consumers.SinkConsumer do
              :status,
              :health,
              :max_memory_mb,
-             :max_storage_mb,
              :legacy_transform,
              :timestamp_format,
              :batch_timeout_ms,
@@ -70,7 +69,6 @@ defmodule Sequin.Consumers.SinkConsumer do
     field :batch_timeout_ms, :integer, default: nil
     field :annotations, :map, default: %{}
     field :max_memory_mb, :integer, default: 128
-    field :max_storage_mb, :integer, default: nil
     field :partition_count, :integer, default: 1
     field :legacy_transform, Ecto.Enum, values: [:none, :record_only], default: :none
     field :timestamp_format, Ecto.Enum, values: [:iso8601, :unix_microsecond], default: :iso8601
@@ -142,7 +140,6 @@ defmodule Sequin.Consumers.SinkConsumer do
       :sequence_id,
       :message_kind,
       :max_memory_mb,
-      :max_storage_mb,
       :transform_id,
       :routing_id,
       :filter_id
@@ -186,7 +183,6 @@ defmodule Sequin.Consumers.SinkConsumer do
       :status,
       :annotations,
       :max_memory_mb,
-      :max_storage_mb,
       :partition_count,
       :legacy_transform,
       :transform_id,
@@ -205,7 +201,6 @@ defmodule Sequin.Consumers.SinkConsumer do
     |> validate_number(:batch_size, less_than_or_equal_to: 1_000)
     |> validate_number(:batch_timeout_ms, greater_than: 0)
     |> validate_number(:max_memory_mb, greater_than_or_equal_to: 128)
-    |> validate_number(:max_storage_mb, greater_than_or_equal_to: 256)
     |> validate_number(:partition_count, greater_than_or_equal_to: 1)
     |> validate_number(:max_retry_count, greater_than: 0)
     |> validate_inclusion(:legacy_transform, [:none, :record_only])
@@ -243,7 +238,6 @@ defmodule Sequin.Consumers.SinkConsumer do
     |> put_change(:max_waiting, get_field(changeset, :max_waiting) || 20)
     |> put_change(:max_ack_pending, get_field(changeset, :max_ack_pending) || 10_000)
     |> put_change(:max_memory_mb, get_field(changeset, :max_memory_mb) || 128)
-    |> put_change(:max_storage_mb, get_field(changeset, :max_storage_mb) || default_max_storage_mb())
     |> put_change(:partition_count, get_field(changeset, :partition_count) || 1)
     |> put_change(:legacy_transform, get_field(changeset, :legacy_transform) || :none)
     |> put_change(:message_kind, get_field(changeset, :message_kind) || :event)
@@ -359,11 +353,4 @@ defmodule Sequin.Consumers.SinkConsumer do
   end
 
   def preload_cached_http_endpoint(consumer), do: consumer
-
-  defp default_max_storage_mb do
-    case Application.get_env(:sequin, :default_max_storage_bytes) do
-      nil -> nil
-      bytes -> round(bytes / 1024 / 1024)
-    end
-  end
 end

lib/sequin/runtime/consumer_lifecycle_event_worker.ex

@@ -74,9 +74,11 @@ defmodule Sequin.Runtime.ConsumerLifecycleEventWorker do
           consumer = Repo.preload(consumer, :replication_slot)
           CheckSinkConfigurationWorker.enqueue(consumer.id)
 
-          # Restart the entire supervision tree for replication, including slot processor, smss, etc.
-          # This is safest- later we can be a bit more intelligent about when to restart (ie. when name changes we don't have to restart)
-          {:ok, _} = RuntimeSupervisor.restart_replication(consumer.replication_slot)
+          unless consumer.status == :disabled do
+            # Restart the entire supervision tree for replication, including slot processor, smss, etc.
+            # This is safest- later we can be a bit more intelligent about when to restart (ie. when name changes we don't have to restart)
+            {:ok, _} = RuntimeSupervisor.restart_replication(consumer.replication_slot)
+          end
         end
 
       "delete" ->