fix: check buffer for all queues, not only first over limit (#3062)

* fix: check buffer for all queues, not only first over limit

* perf: accomodate for spikes

* chore: fix failing test

Author: Ziinc

lib/logflare/backends.ex

@@ -816,15 +816,7 @@ defmodule Logflare.Backends do
   end
 
   def cached_local_pending_buffer_full?(%Source{id: source_id}) do
-    PubSubRates.Cache.get_local_buffer(source_id, nil)
-    |> Map.get(:queues, [])
-    |> case do
-      [] ->
-        false
-
-      queues ->
-        Enum.all?(queues, fn {_key, v} -> v > @max_pending_buffer_len_per_queue end)
-    end
+    buffer_full_for_backend?(source_id, nil)
   end
 
   @spec buffer_full_for_backend?(
@@ -835,7 +827,7 @@ defmodule Logflare.Backends do
   defp buffer_full_for_backend?(source_id, backend_id) do
     case PubSubRates.Cache.get_local_buffer(source_id, backend_id) do
       %{queues: [_ | _] = queues} ->
-        Enum.any?(queues, fn {_key, count} ->
+        Enum.all?(queues, fn {_key, count} ->
           count > @max_pending_buffer_len_per_queue
         end)
 

lib/logflare/backends/ingest_event_queue.ex

@@ -14,7 +14,7 @@ defmodule Logflare.Backends.IngestEventQueue do
 
   @ets_table_mapper :ingest_event_queue_mapping
   @ets_table :source_ingest_events
-  @max_queue_size 15_000
+  @max_queue_size 30_000
   @type source_backend_pid ::
           {Source.t() | pos_integer(), Backend.t() | pos_integer() | nil, pid() | nil}
   @type table_key :: {pos_integer(), pos_integer() | nil, pid() | nil}

lib/logflare/backends/ingest_event_queue/queue_janitor.ex

@@ -14,8 +14,8 @@ defmodule Logflare.Backends.IngestEventQueue.QueueJanitor do
   require Logger
   @default_interval 1_000
   @default_remainder 100
-  @default_max Logflare.Backends.max_buffer_queue_len()
   @default_purge_ratio 0.05
+  @default_max round(Logflare.Backends.max_buffer_queue_len() * 1.2)
 
   def start_link(opts) do
     GenServer.start_link(__MODULE__, opts)

test/logflare_web/plugs/buffer_limiter_test.exs

@@ -263,10 +263,12 @@ defmodule LogflareWeb.Plugs.BufferLimiterTest do
       assert json_response(conn, 429)
     end
 
-    test "returns 429 when user default queue is full even if system default is not", %{
-      conn: conn,
-      source: source
-    } do
+    test "returns 429 when user's backend default ingest queue is full even if system queue is not",
+         %{
+           conn: conn,
+           source: source,
+           backend1: backend1
+         } do
       system_queue_key = {source.id, nil, spawn(fn -> :ok end)}
       IngestEventQueue.upsert_tid(system_queue_key)
 
@@ -275,7 +277,7 @@ defmodule LogflareWeb.Plugs.BufferLimiterTest do
         IngestEventQueue.add_to_table(system_queue_key, [le])
       end
 
-      user_queue_key = {source.id, nil, self()}
+      user_queue_key = {source.id, backend1.id, spawn(fn -> :ok end)}
       IngestEventQueue.upsert_tid(user_queue_key)
 
       for _ <- 1..(Backends.max_buffer_queue_len() + 500) do
@@ -284,6 +286,7 @@ defmodule LogflareWeb.Plugs.BufferLimiterTest do
       end
 
       Backends.cache_local_buffer_lens(source.id, nil)
+      Backends.cache_local_buffer_lens(source.id, backend1.id)
 
       conn =
         conn