From 22bfd6f0d3b42af13c16cd00c77193b4b5203a4a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bartosz=20B=C5=82aszk=C3=B3w?= Date: Tue, 29 Sep 2026 20:38:04 +0200 Subject: [PATCH 1/6] refactor: add Cache behaviour; stop assuming every cache is Cachex Generic code (health check, telemetry, test support, cache busting) called Cachex directly on every cache module. That breaks as soon as a cache moves to another backend, e.g. the Nebulex multilevel KeyValues cache in #4054. - Logflare.Cache: a behaviour for cache operations (healthy?/0, stats/0, reset/0). `use Logflare.Cache` injects overridable Cachex defaults; the default implementation module can be swapped with `impl:`. - `use Logflare.ContextCache` builds on it and makes bust_by/1 a required callback with a default primary-key bust. Every bust now goes through the cache's bust_by/1; the tuples CacheBuster emits are unchanged. - The health check, telemetry and test resets call these callbacks instead of Cachex. - Cachex configuration is shared (Logflare.Cache.CachexOps.child_spec/2 with named limit/ttl/purge_interval/warmer/compressed options), replacing the per-cache child_spec boilerplate and the Utils.cache_* helpers. Every cache keeps its current settings. Behaviour changes: - /health reports an unhealthy cache as "unhealthy" instead of "no_cache". - The cachex..purge and .stats metrics are removed; Cachex 4 never sets them, so they were always 0. - Cache child specs are typed :supervisor (Cachex starts a supervisor). - A {Rules, pkey} bust follows Rules.Cache's own `id:` busting instead of the generic scan; CacheBuster only sends Rules keyword busts. - Busting a cache by an unsupported keyword raises ArgumentError. --- .dialyzer_ignore.exs | 1 - AGENTS.md | 5 +- lib/logflare/auth/cache.ex | 28 ++-- lib/logflare/backends/cache.ex | 28 +--- lib/logflare/billing/cache.ex | 28 ++-- lib/logflare/cache.ex | 52 ++++++++ lib/logflare/cache/cachex_ops.ex | 122 ++++++++++++++++++ lib/logflare/config_cat/cache.ex | 26 +--- lib/logflare/context_cache.ex | 89 +++++++------ .../context_cache/tombstones/cache.ex | 28 ++-- lib/logflare/endpoints/cache.ex | 28 ++-- lib/logflare/key_values/cache.ex | 60 ++------- lib/logflare/logs/log_events_cache.ex | 36 ++---- lib/logflare/logs/rejected_log_events.ex | 28 ++-- lib/logflare/partners/cache.ex | 26 +--- lib/logflare/pubsub_rates/cache.ex | 29 ++--- lib/logflare/rules/cache.ex | 55 ++------ lib/logflare/saved_searches/cache.ex | 44 +------ lib/logflare/source_schemas/cache.ex | 40 ++---- lib/logflare/sources/cache.ex | 34 ++--- lib/logflare/system_cache.ex | 46 ++----- lib/logflare/team_users/cache.ex | 26 +--- lib/logflare/users/cache.ex | 33 ++--- lib/logflare/utils.ex | 44 ------- .../controllers/health_check_controller.ex | 6 +- lib/telemetry.ex | 29 +---- mix.exs | 2 +- test/logflare/cache/cachex_ops_test.exs | 106 +++++++++++++++ test/logflare/cache_test.exs | 81 ++++++++++++ test/logflare/context_cache_test.exs | 32 +++++ test/logflare/system_cache_test.exs | 10 +- test/logflare/telemetry_test.exs | 2 - .../health_check_controller_test.exs | 15 +++ .../context_cache_bust_by_kw_bench.exs | 4 +- test/support/conn_case.ex | 2 +- test/support/data_case.ex | 2 +- test/test_helper.exs | 1 + 37 files changed, 620 insertions(+), 608 deletions(-) create mode 100644 lib/logflare/cache.ex create mode 100644 lib/logflare/cache/cachex_ops.ex create mode 100644 test/logflare/cache/cachex_ops_test.exs create mode 100644 test/logflare/cache_test.exs diff --git a/.dialyzer_ignore.exs b/.dialyzer_ignore.exs index e4790b424f..8a86cd8e76 100644 --- a/.dialyzer_ignore.exs +++ b/.dialyzer_ignore.exs @@ -43,7 +43,6 @@ {"lib/logflare_web/controllers/api/endpoint_controller.ex", :pattern_match}, {"lib/logflare_web/controllers/billing_controller.ex", :unused_fun}, {"lib/logflare_web/controllers/billing_controller.ex", :pattern_match_cov}, - {"lib/logflare_web/controllers/health_check_controller.ex", :pattern_match}, {"lib/logflare_web/controllers/source_controller.ex", :pattern_match}, {"lib/logflare_web/live/monaco_editor_component.ex", :pattern_match}, {"lib/logflare_web/live/query_live.ex", :no_return}, diff --git a/AGENTS.md b/AGENTS.md index 553f9b6431..ddc85396b2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -84,8 +84,9 @@ gate to go green - fix the code. that have pre-existing findings. They are off so the gate is green on existing code. Fixing a backlog check's findings and removing it from that list is a welcome change on its own. Never add a check to the list to go green. -- `mix test.slop` runs `ex_dna --max-clones 29`, a duplication ratchet. A new - clone fails CI. When you remove clones, lower the number. Never raise it. +- `mix test.slop` runs `ex_dna --max-clones `, + a duplication ratchet. A new clone fails CI. When you remove clones, + lower the number. Never raise it. - `mix test.structure` runs `reach.check --smells` against `.reach.baseline.json` (194 accepted findings), so only **new** structural smells fail. To accept a diff --git a/lib/logflare/auth/cache.ex b/lib/logflare/auth/cache.ex index b10b79e0a2..1aec407aca 100644 --- a/lib/logflare/auth/cache.ex +++ b/lib/logflare/auth/cache.ex @@ -4,31 +4,19 @@ defmodule Logflare.Auth.Cache do Cachex `expiration`. """ + use Logflare.ContextCache + alias Logflare.Auth + alias Logflare.Cache.CachexOps alias Logflare.OauthAccessTokens.OauthAccessToken alias Logflare.User - alias Logflare.Utils def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - __MODULE__, - [ - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min(5, 2) - ] - ]} - } + CachexOps.child_spec(__MODULE__, + limit: 100_000, + ttl: to_timeout(minute: 5), + purge_interval: to_timeout(minute: 2) + ) end @spec verify_access_token(OauthAccessToken.t() | String.t()) :: diff --git a/lib/logflare/backends/cache.ex b/lib/logflare/backends/cache.ex index caaa98fbde..2e31effa8e 100644 --- a/lib/logflare/backends/cache.ex +++ b/lib/logflare/backends/cache.ex @@ -1,33 +1,13 @@ defmodule Logflare.Backends.Cache do @moduledoc false + use Logflare.ContextCache + alias Logflare.Backends - alias Logflare.Utils - import Cachex.Spec + alias Logflare.Cache.CachexOps def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - __MODULE__, - [ - warmers: [ - warmer(required: false, module: Backends.CacheWarmer, name: Backends.CacheWarmer) - ], - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min() - ] - ]} - } + CachexOps.child_spec(__MODULE__, limit: 100_000, warmer: Backends.CacheWarmer) end def list_backends(arg), do: apply_repo_fun(__ENV__.function, [arg]) diff --git a/lib/logflare/billing/cache.ex b/lib/logflare/billing/cache.ex index 31ac2889d7..41e119cfe4 100644 --- a/lib/logflare/billing/cache.ex +++ b/lib/logflare/billing/cache.ex @@ -1,29 +1,17 @@ defmodule Logflare.Billing.Cache do @moduledoc false + use Logflare.ContextCache + alias Logflare.Billing - alias Logflare.Utils + alias Logflare.Cache.CachexOps def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - __MODULE__, - [ - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min(180, 10) - ] - ]} - } + CachexOps.child_spec(__MODULE__, + limit: 100_000, + ttl: to_timeout(hour: 3), + purge_interval: to_timeout(minute: 10) + ) end def get_billing_account_by(keyword) do diff --git a/lib/logflare/cache.ex b/lib/logflare/cache.ex new file mode 100644 index 0000000000..7360bf5a38 --- /dev/null +++ b/lib/logflare/cache.ex @@ -0,0 +1,52 @@ +defmodule Logflare.Cache do + @moduledoc """ + Operational contract of an application cache, independent of its storage backend. + + `use Logflare.Cache` injects default implementations of every callback, delegating to + `Logflare.Cache.CachexOps`. A cache on another backend overrides them, or passes + `impl: module` whose `healthy?/1`, `stats/1` and `reset/1` take the cache module. + """ + + alias Logflare.Cache.CachexOps + + @typedoc """ + Counters since the last `c:reset/0`. Rates are percentages (0-100); `total_heap_size` is in words. + """ + @type stats() :: %{ + evictions: non_neg_integer(), + expirations: non_neg_integer(), + operations: non_neg_integer(), + hits: non_neg_integer(), + misses: non_neg_integer(), + hit_rate: number(), + miss_rate: number(), + total_heap_size: non_neg_integer() + } + + @doc "Whether the cache on this node can serve requests." + @callback healthy?() :: boolean() + + @callback stats() :: stats() + + @doc "Clears all entries and statistics." + @callback reset() :: :ok + + defmacro __using__(opts) do + impl = Keyword.get(opts, :impl, CachexOps) + + quote do + @behaviour Logflare.Cache + + @impl Logflare.Cache + def healthy?, do: unquote(impl).healthy?(__MODULE__) + + @impl Logflare.Cache + def stats, do: unquote(impl).stats(__MODULE__) + + @impl Logflare.Cache + def reset, do: unquote(impl).reset(__MODULE__) + + defoverridable healthy?: 0, stats: 0, reset: 0 + end + end +end diff --git a/lib/logflare/cache/cachex_ops.ex b/lib/logflare/cache/cachex_ops.ex new file mode 100644 index 0000000000..9b25a40172 --- /dev/null +++ b/lib/logflare/cache/cachex_ops.ex @@ -0,0 +1,122 @@ +defmodule Logflare.Cache.CachexOps do + @moduledoc """ + Cachex implementations of the `Logflare.Cache` callbacks, and builders for Cachex start options. + """ + + import Cachex.Spec + + @stat_keys [ + :evictions, + :expirations, + :operations, + :hits, + :misses, + :hit_rate, + :miss_rate + ] + + @type warmer() :: module() | {module(), interval: pos_integer()} + @type cache_opt() :: + {:limit, pos_integer() | nil} + | {:ttl, pos_integer()} + | {:purge_interval, pos_integer()} + | {:warmer, warmer() | nil} + | {:compressed, boolean()} + + @doc """ + Child spec for a Cachex cache registered as `name`, with the stats hook when `stats_enabled?/0`. + + Options: + * `:limit` (required) - maximum number of entries, `nil` for no limit + * `:ttl` - default entry time-to-live, a timeout in ms, defaults to 20 minutes + * `:purge_interval` - timeout in ms between purges of expired entries, defaults to 5 minutes + * `:warmer` - a non-required warmer module, or `{module, interval: ms}`, registered under the module name + * `:compressed` - defaults to `false` + """ + @spec child_spec(atom(), [cache_opt()]) :: Supervisor.child_spec() + def child_spec(name, opts) do + opts = Keyword.update!(cachex_opts(opts), :hooks, &(stats_hooks() ++ &1)) + Supervisor.child_spec({Cachex, [name, opts]}, id: name) + end + + @doc """ + Cachex start options for `t:cache_opt/0`, without the stats hook. + """ + @spec cachex_opts([cache_opt()]) :: keyword() + def cachex_opts(opts) do + [ + hooks: limit_hooks(Keyword.fetch!(opts, :limit)), + expiration: + expiration( + default: Keyword.get(opts, :ttl, to_timeout(minute: 20)), + interval: Keyword.get(opts, :purge_interval, to_timeout(minute: 5)), + lazy: true + ), + warmers: warmers(opts[:warmer]), + compressed: Keyword.get(opts, :compressed, false) + ] + end + + @spec stats_enabled?() :: boolean() + def stats_enabled?, do: Application.get_env(:logflare, :cache_stats, false) + + @spec healthy?(Cachex.t()) :: boolean() + def healthy?(cache), do: match?({:ok, _}, Cachex.size(cache)) + + @doc """ + Raises when the cache runs without the stats hook. + """ + @spec stats(Cachex.t()) :: Logflare.Cache.stats() + def stats(cache) do + {:ok, stats} = Cachex.stats(cache) + + {:total_heap_size, total_heap_size} = + cache + |> Process.whereis() + |> Process.info(:total_heap_size) + + for key <- @stat_keys, into: %{total_heap_size: total_heap_size} do + {key, Map.get(stats, key, 0)} + end + end + + @spec reset(Cachex.t()) :: :ok + def reset(cache) do + {:ok, true} = Cachex.reset(cache, hooks: [Cachex.Stats]) + :ok + end + + @doc """ + Deletes `keys` and returns how many of them were present. + """ + @spec delete_keys(Cachex.t(), Enumerable.t()) :: {:ok, non_neg_integer()} + def delete_keys(cache, keys) do + Cachex.execute(cache, fn worker -> + Enum.reduce(keys, 0, fn key, acc -> acc + take_count(worker, key) end) + end) + end + + defp take_count(cache, key) do + case Cachex.take(cache, key) do + {:ok, nil} -> 0 + {:ok, _value} -> 1 + end + end + + defp stats_hooks, do: if(stats_enabled?(), do: [hook(module: Cachex.Stats)], else: []) + + defp limit_hooks(nil), do: [] + + defp limit_hooks(limit) when is_integer(limit) do + [hook(module: Cachex.Limit.Scheduled, args: {limit, [], []})] + end + + defp warmers(nil), do: [] + + defp warmers({module, opts}) do + [interval: interval] = Keyword.validate!(opts, interval: nil) + [warmer(module: module, name: module, required: false, interval: interval)] + end + + defp warmers(module), do: warmers({module, []}) +end diff --git a/lib/logflare/config_cat/cache.ex b/lib/logflare/config_cat/cache.ex index 846d301214..ad5e643da5 100644 --- a/lib/logflare/config_cat/cache.ex +++ b/lib/logflare/config_cat/cache.ex @@ -4,28 +4,14 @@ defmodule Logflare.ConfigCatCache do Used to avoid repeated ConfigCat calls on hot paths like LogEvent processing. """ - alias Logflare.Utils + alias Logflare.Cache.CachexOps def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - __MODULE__, - [ - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min(10, 1) - ] - ]} - } + CachexOps.child_spec(__MODULE__, + limit: 100_000, + ttl: to_timeout(minute: 10), + purge_interval: to_timeout(minute: 1) + ) end @spec get(term()) :: {:ok, term()} | {:error, term()} diff --git a/lib/logflare/context_cache.ex b/lib/logflare/context_cache.ex index 172ed6835d..3b8b393f8b 100644 --- a/lib/logflare/context_cache.ex +++ b/lib/logflare/context_cache.ex @@ -12,8 +12,8 @@ defmodule Logflare.ContextCache do primary key checking within the matchspec. This approach queries across a narrower set of records, providing better performance compared to a reverse index approach. - If customization of busting is needed, cache module may implement `c:bust_by/1` callback expecting - a keyword list instead of primary key for entry. + `use Logflare.ContextCache` makes the module a `Logflare.Cache` and injects a default `c:bust_by/1` + that busts by `id:` (see `bust_by/2`). Caches that need other busting keys override it. ## List Busting @@ -36,13 +36,27 @@ defmodule Logflare.ContextCache do filter out stale incoming messages. """ + alias Logflare.Cache.CachexOps alias Logflare.ContextCache.Gossip @doc """ - Optional callback implementing custom cache key busting by a keyword of values + Busts cache entries by a keyword of values, returning the number of entries busted. """ @callback bust_by(keyword()) :: {:ok, non_neg_integer()} | {:error, term()} + defmacro __using__(opts) do + quote do + use Logflare.Cache, unquote(opts) + + @behaviour Logflare.ContextCache + + @impl Logflare.ContextCache + def bust_by(kw), do: Logflare.ContextCache.bust_by(__MODULE__, kw) + + defoverridable bust_by: 1 + end + end + @spec apply_fun(module(), tuple() | atom(), list()) :: any() def apply_fun(context, {fun, _arity}, args), do: apply_fun(context, fun, args) @@ -70,36 +84,30 @@ defmodule Logflare.ContextCache do It is intended for following a WAL for cache busting. When a new record comes in from the WAL, the CacheBuster process calls this function with either the primary keys extracted from those records - or a keyword list with fields useful for busting. - - For primary key, the function then: - - 1. Queries the relevant context cache using a matchspec to find entries to bust - 2. Handles both single records and lists of records containing matching IDs - 3. Deletes matching cache entries - - For keywords, it expects the cache to handle busting by implementing `c:bust_by/1` + or a keyword list with fields useful for busting. Both are passed to the context cache's `c:bust_by/1`, + a primary key as `[id: pkey]`. """ @spec bust_keys(list()) :: {:ok, non_neg_integer()} def bust_keys(values) when is_list(values) do busted = - for {context, primary_key} <- values, reduce: 0 do + for {context, pkey_or_kw} <- values, reduce: 0 do acc -> - {:ok, n} = bust_key({context, primary_key}) + {:ok, n} = bust_key(context, pkey_or_kw) acc + n end {:ok, busted} end - defp bust_key({context, kw}) when is_list(kw) do - context_cache = cache_name(context) - context_cache.bust_by(kw) - end - - defp bust_key({context, pkey}) do - context_cache = cache_name(context) + defp bust_key(context, kw) when is_list(kw), do: cache_name(context).bust_by(kw) + defp bust_key(context, pkey), do: cache_name(context).bust_by(id: pkey) + @doc """ + Default `c:bust_by/1`: with `[id: pkey]`, busts every entry whose cached value is a map with that `:id`, + an `{:ok, map}` with that `:id`, or a list containing such a map. Raises `ArgumentError` for any other keyword. + """ + @spec bust_by(Cachex.t(), keyword()) :: {:ok, non_neg_integer()} + def bust_by(cache, id: pkey) do filter = { # use orelse to prevent 2nd condition failing as value is not a map @@ -119,12 +127,22 @@ defmodule Logflare.ContextCache do {:==, {:map_get, :id, {:element, 2, :value}}, pkey}} } - query = - Cachex.Query.build(where: filter, output: {:key, :value}) + query = Cachex.Query.build(where: filter, output: {:key, :value}) - context_cache - |> Cachex.stream!(query) - |> delete_matching_entries(context_cache, pkey) + keys = + cache + |> Cachex.stream!(query) + |> Stream.filter(fn + {_k, {:cached, v}} when is_list(v) -> Enum.any?(v, &(&1.id == pkey)) + {_k, _v} -> true + end) + |> Stream.map(fn {k, _v} -> k end) + + CachexOps.delete_keys(cache, keys) + end + + def bust_by(cache, kw) do + raise ArgumentError, "#{inspect(cache)} does not support busting by #{inspect(kw)}" end @spec cache_name(atom()) :: atom() @@ -151,23 +169,4 @@ defmodule Logflare.ContextCache do value end end - - defp delete_matching_entries(entries, context_cache, pkey) do - to_delete = - entries - |> Stream.filter(fn - {_k, {:cached, v}} when is_list(v) -> - Enum.any?(v, &(&1.id == pkey)) - - {_k, _v} -> - true - end) - - Cachex.execute(context_cache, fn worker -> - Enum.reduce(to_delete, 0, fn {k, _v}, acc -> - Cachex.del(worker, k) - acc + 1 - end) - end) - end end diff --git a/lib/logflare/context_cache/tombstones/cache.ex b/lib/logflare/context_cache/tombstones/cache.ex index 2be676f9ad..34b489cf0c 100644 --- a/lib/logflare/context_cache/tombstones/cache.ex +++ b/lib/logflare/context_cache/tombstones/cache.ex @@ -1,28 +1,18 @@ defmodule Logflare.ContextCache.Tombstones.Cache do @moduledoc false - require Cachex.Spec - @name __MODULE__ - - def child_spec(_options) do - expiration = - Cachex.Spec.expiration( - interval: to_timeout(second: 30), - default: to_timeout(minute: 1), - lazy: true - ) + use Logflare.Cache - hooks = - if Application.get_env(:logflare, :cache_stats, false) do - [Cachex.Spec.hook(module: Cachex.Stats)] - end + alias Logflare.Cache.CachexOps - options = [ - expiration: expiration, - hooks: List.wrap(hooks) - ] + @name __MODULE__ - Supervisor.child_spec({Cachex, [@name, options]}, id: @name) + def child_spec(_options) do + CachexOps.child_spec(@name, + limit: nil, + ttl: to_timeout(minute: 1), + purge_interval: to_timeout(second: 30) + ) end def put_tombstone(cache, tombstone) do diff --git a/lib/logflare/endpoints/cache.ex b/lib/logflare/endpoints/cache.ex index 4a653589f4..d9975d58a9 100644 --- a/lib/logflare/endpoints/cache.ex +++ b/lib/logflare/endpoints/cache.ex @@ -3,29 +3,17 @@ defmodule Logflare.Endpoints.Cache do Cachex for Endpoints context. """ + use Logflare.ContextCache + + alias Logflare.Cache.CachexOps alias Logflare.Endpoints - alias Logflare.Utils def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - __MODULE__, - [ - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.reject(&is_nil/1), - expiration: Utils.cache_expiration_min(2, 1) - ] - ]} - } + CachexOps.child_spec(__MODULE__, + limit: 100_000, + ttl: to_timeout(minute: 2), + purge_interval: to_timeout(minute: 1) + ) end def get_endpoint_query(kw), do: apply_repo_fun(:get_endpoint_query, [kw]) diff --git a/lib/logflare/key_values/cache.ex b/lib/logflare/key_values/cache.ex index 3a55a9b0d3..7f2bc6bf3f 100644 --- a/lib/logflare/key_values/cache.ex +++ b/lib/logflare/key_values/cache.ex @@ -1,46 +1,21 @@ defmodule Logflare.KeyValues.Cache do @moduledoc false + use Logflare.ContextCache + + alias Logflare.Cache.CachexOps alias Logflare.ContextCache alias Logflare.KeyValues alias Logflare.Repo - alias Logflare.Utils - - import Cachex.Spec - - @behaviour ContextCache def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - start: { - Cachex, - :start_link, - [ - __MODULE__, - [ - compressed: true, - warmers: [ - warmer( - required: false, - module: KeyValues.CacheWarmer, - name: KeyValues.CacheWarmer, - interval: :timer.hours(1) - ) - ], - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(10_000_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min(1440, 60) - ] - ] - } - } + CachexOps.child_spec(__MODULE__, + limit: 10_000_000, + ttl: to_timeout(day: 1), + purge_interval: to_timeout(hour: 1), + compressed: true, + warmer: {KeyValues.CacheWarmer, interval: :timer.hours(1)} + ) end @spec count(integer()) :: non_neg_integer() @@ -87,13 +62,7 @@ defmodule Logflare.KeyValues.Cache do @impl ContextCache def bust_by(kw) do - entries = bust_entries(kw) - - Cachex.execute(__MODULE__, fn worker -> - Enum.reduce(entries, 0, fn k, acc -> - acc + delete_and_count(worker, k) - end) - end) + CachexOps.delete_keys(__MODULE__, bust_entries(kw)) end defp bust_entries(kw) do @@ -118,11 +87,4 @@ defmodule Logflare.KeyValues.Cache do _ -> false end) end - - defp delete_and_count(cache, key) do - case Cachex.take(cache, key) do - {:ok, nil} -> 0 - {:ok, _value} -> 1 - end - end end diff --git a/lib/logflare/logs/log_events_cache.ex b/lib/logflare/logs/log_events_cache.ex index 2eba752e90..ebc8dea61f 100644 --- a/lib/logflare/logs/log_events_cache.ex +++ b/lib/logflare/logs/log_events_cache.ex @@ -1,36 +1,22 @@ defmodule Logflare.Logs.LogEvents.Cache do @moduledoc false - alias Logflare.Logs.LogEvents + + use Logflare.ContextCache + + alias Logflare.Cache.CachexOps alias Logflare.ContextCache alias Logflare.LogEvent, as: LE - alias Logflare.Utils + alias Logflare.Logs.LogEvents @cache __MODULE__ def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - name: __MODULE__, - start: { - Cachex, - :start_link, - [ - @cache, - [ - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(15_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min(10), - compressed: true - ] - ] - } - } + CachexOps.child_spec(@cache, + limit: 15_000, + ttl: to_timeout(minute: 10), + purge_interval: to_timeout(minute: 5), + compressed: true + ) end @fetch_event_by_id {:fetch_event_by_id, 2} diff --git a/lib/logflare/logs/rejected_log_events.ex b/lib/logflare/logs/rejected_log_events.ex index 484036227d..343100ddfd 100644 --- a/lib/logflare/logs/rejected_log_events.ex +++ b/lib/logflare/logs/rejected_log_events.ex @@ -22,33 +22,21 @@ defmodule Logflare.Logs.RejectedLogEvents do ``` """ + use Logflare.Cache + + alias Logflare.Cache.CachexOps alias Logflare.Sources.Source alias Logflare.LogEvent, as: LE - alias Logflare.Utils alias Logflare.Sources @cache __MODULE__ def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: @cache, - start: - {Cachex, :start_link, - [ - @cache, - [ - expiration: Utils.cache_expiration_min(60), - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(10_000) - ] - |> Enum.filter(& &1) - ] - ]} - } + CachexOps.child_spec(@cache, + limit: 10_000, + ttl: to_timeout(hour: 1), + purge_interval: to_timeout(minute: 5) + ) end @spec get_by_source(Source.t()) :: list(LE.t()) diff --git a/lib/logflare/partners/cache.ex b/lib/logflare/partners/cache.ex index cb99da62db..b677412eb9 100644 --- a/lib/logflare/partners/cache.ex +++ b/lib/logflare/partners/cache.ex @@ -3,30 +3,12 @@ defmodule Logflare.Partners.Cache do Cache for Partners """ - alias Logflare.Partners - alias Logflare.Utils + use Logflare.ContextCache - def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) + alias Logflare.Cache.CachexOps + alias Logflare.Partners - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - __MODULE__, - [ - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min() - ] - ]} - } - end + def child_spec(_), do: CachexOps.child_spec(__MODULE__, limit: 100_000) def get_partner(id), do: apply_repo_fun(__ENV__.function, [id]) def get_user_by_uuid(partner, token), do: apply_repo_fun(__ENV__.function, [partner, token]) diff --git a/lib/logflare/pubsub_rates/cache.ex b/lib/logflare/pubsub_rates/cache.ex index af94238361..0bc767f83e 100644 --- a/lib/logflare/pubsub_rates/cache.ex +++ b/lib/logflare/pubsub_rates/cache.ex @@ -1,35 +1,24 @@ defmodule Logflare.PubSubRates.Cache do @moduledoc false + + use Logflare.Cache + require Logger + alias Logflare.Cache.CachexOps alias Logflare.Sources.Source alias Logflare.Cluster alias Logflare.Users.Cache, as: UsersCache - alias Logflare.Utils @cache __MODULE__ @default_bucket_width 60 def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - @cache, - [ - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min(5) - ] - ]} - } + CachexOps.child_spec(@cache, + limit: 100_000, + ttl: to_timeout(minute: 5), + purge_interval: to_timeout(minute: 5) + ) end def clear do diff --git a/lib/logflare/rules/cache.ex b/lib/logflare/rules/cache.ex index 47b365f555..58e08c4974 100644 --- a/lib/logflare/rules/cache.ex +++ b/lib/logflare/rules/cache.ex @@ -1,41 +1,22 @@ defmodule Logflare.Rules.Cache do @moduledoc false + use Logflare.ContextCache + alias Logflare.Backends.Backend + alias Logflare.Cache.CachexOps alias Logflare.ContextCache alias Logflare.Repo alias Logflare.Rules alias Logflare.Sources.Source - alias Logflare.Utils - import Cachex.Spec - - @behaviour ContextCache def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - start: { - Cachex, - :start_link, - [ - __MODULE__, - [ - warmers: [ - warmer(required: false, module: Rules.CacheWarmer, name: Rules.CacheWarmer) - ], - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min(60, 5) - ] - ] - } - } + CachexOps.child_spec(__MODULE__, + limit: 100_000, + ttl: to_timeout(hour: 1), + purge_interval: to_timeout(minute: 5), + warmer: Rules.CacheWarmer + ) end @spec list_rules(Source.t() | Backend.t()) :: [Rules.Rule.t()] @@ -57,9 +38,8 @@ defmodule Logflare.Rules.Cache do @impl ContextCache def bust_by(kw) do - entries = - kw - |> Enum.flat_map(fn + keys = + Enum.flat_map(kw, fn {:id, id} -> [{:get_rule, [id]}] @@ -70,11 +50,7 @@ defmodule Logflare.Rules.Cache do [{:list_by_backend_id, [backend_id]}] end) - Cachex.execute(Rules.Cache, fn worker -> - Enum.reduce(entries, 0, fn k, acc -> - acc + delete_and_count(worker, k) - end) - end) + CachexOps.delete_keys(__MODULE__, keys) end defp fetch_rule(cache, id) do @@ -83,13 +59,6 @@ defmodule Logflare.Rules.Cache do end) end - defp delete_and_count(cache, key) do - case Cachex.take(cache, key) do - {:ok, nil} -> 0 - {:ok, _value} -> 1 - end - end - defp apply_repo_fun(fun, args) do Logflare.ContextCache.apply_fun(Rules, fun, args) end diff --git a/lib/logflare/saved_searches/cache.ex b/lib/logflare/saved_searches/cache.ex index 0922ed1c48..1182dffb89 100644 --- a/lib/logflare/saved_searches/cache.ex +++ b/lib/logflare/saved_searches/cache.ex @@ -1,55 +1,23 @@ defmodule Logflare.SavedSearches.Cache do @moduledoc false - @behaviour Logflare.ContextCache + use Logflare.ContextCache + alias Logflare.Cache.CachexOps alias Logflare.SavedSearches - alias Logflare.Utils - def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - __MODULE__, - [ - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(10_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min() - ] - ]} - } - end + def child_spec(_), do: CachexOps.child_spec(__MODULE__, limit: 10_000) def list_saved_searches_by_source(source_id), do: apply_repo_fun(__ENV__.function, [source_id]) @impl Logflare.ContextCache def bust_by(kw) do - entries = - kw - |> Enum.map(fn + keys = + Enum.map(kw, fn {:source_id, source_id} -> {:list_saved_searches_by_source, [source_id]} end) - Cachex.execute(__MODULE__, fn cache -> - Enum.reduce(entries, 0, fn key, acc -> - acc + delete_and_count(cache, key) - end) - end) - end - - defp delete_and_count(cache, key) do - case Cachex.take(cache, key) do - {:ok, nil} -> 0 - {:ok, _value} -> 1 - end + CachexOps.delete_keys(__MODULE__, keys) end defp apply_repo_fun(arg1, arg2) do diff --git a/lib/logflare/source_schemas/cache.ex b/lib/logflare/source_schemas/cache.ex index 589d9c70f4..eba0f749f0 100644 --- a/lib/logflare/source_schemas/cache.ex +++ b/lib/logflare/source_schemas/cache.ex @@ -1,38 +1,18 @@ defmodule Logflare.SourceSchemas.Cache do @moduledoc false - alias Logflare.SourceSchemas - alias Logflare.Utils - import Cachex.Spec + use Logflare.ContextCache - def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) + alias Logflare.Cache.CachexOps + alias Logflare.SourceSchemas - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - __MODULE__, - [ - warmers: [ - warmer( - required: false, - module: SourceSchemas.CacheWarmer, - name: SourceSchemas.CacheWarmer - ) - ], - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.filter(& &1), - # shorter expiration for schemas - expiration: Utils.cache_expiration_min(10, 2) - ] - ]} - } + def child_spec(_) do + CachexOps.child_spec(__MODULE__, + limit: 100_000, + ttl: to_timeout(minute: 10), + purge_interval: to_timeout(minute: 2), + warmer: SourceSchemas.CacheWarmer + ) end def get_source_schema_by(kv), do: apply_fun(__ENV__.function, [kv]) diff --git a/lib/logflare/sources/cache.ex b/lib/logflare/sources/cache.ex index 0e8ce2c081..9ac2ec6e49 100644 --- a/lib/logflare/sources/cache.ex +++ b/lib/logflare/sources/cache.ex @@ -1,37 +1,21 @@ defmodule Logflare.Sources.Cache do @moduledoc false + use Logflare.ContextCache + + alias Logflare.Cache.CachexOps alias Logflare.Repo alias Logflare.Rules alias Logflare.Sources alias Logflare.Sources.Source - alias Logflare.Utils - - import Cachex.Spec def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - __MODULE__, - [ - warmers: [ - warmer(required: false, module: Sources.CacheWarmer, name: Sources.CacheWarmer) - ], - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min(60, 5) - ] - ]} - } + CachexOps.child_spec(__MODULE__, + limit: 100_000, + ttl: to_timeout(hour: 1), + purge_interval: to_timeout(minute: 5), + warmer: Sources.CacheWarmer + ) end # For ingest diff --git a/lib/logflare/system_cache.ex b/lib/logflare/system_cache.ex index 61daf8476a..311df1a43b 100644 --- a/lib/logflare/system_cache.ex +++ b/lib/logflare/system_cache.ex @@ -1,50 +1,26 @@ defmodule Logflare.SystemCache do @moduledoc false - import Cachex.Spec + use Logflare.Cache require Logger - alias Logflare.Utils + alias Logflare.Cache.CachexOps @cache __MODULE__ def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - env = Application.get_env(:logflare, :env) - - warmers = - if env == :test do - [] - else - [ - warmer( - required: false, - module: __MODULE__.Warmer, - name: __MODULE__.Warmer, - interval: :timer.seconds(3) - ) - ] + warmer = + if Application.get_env(:logflare, :env) != :test do + {__MODULE__.Warmer, interval: :timer.seconds(3)} end - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - @cache, - [ - warmers: warmers, - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_sec(5, 5) - ] - ]} - } + CachexOps.child_spec(@cache, + limit: 100, + ttl: to_timeout(second: 5), + purge_interval: to_timeout(second: 5), + warmer: warmer + ) end @spec memory_utilization() :: float() diff --git a/lib/logflare/team_users/cache.ex b/lib/logflare/team_users/cache.ex index 9baee55215..7e3fea5096 100644 --- a/lib/logflare/team_users/cache.ex +++ b/lib/logflare/team_users/cache.ex @@ -3,30 +3,12 @@ defmodule Logflare.TeamUsers.Cache do Cache for TeamUsers. """ - alias Logflare.TeamUsers - alias Logflare.Utils + use Logflare.ContextCache - def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) + alias Logflare.Cache.CachexOps + alias Logflare.TeamUsers - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - __MODULE__, - [ - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min() - ] - ]} - } - end + def child_spec(_), do: CachexOps.child_spec(__MODULE__, limit: 100_000) def get_team_user(id), do: apply_repo_fun({:get_team_user, 1}, [id]) def get_team_user!(id), do: apply_repo_fun(:get_team_user!, [id]) diff --git a/lib/logflare/users/cache.ex b/lib/logflare/users/cache.ex index a5231705e1..f14d047893 100644 --- a/lib/logflare/users/cache.ex +++ b/lib/logflare/users/cache.ex @@ -3,33 +3,18 @@ defmodule Logflare.Users.Cache do Cache for users. """ + use Logflare.ContextCache + + alias Logflare.Cache.CachexOps alias Logflare.Users - alias Logflare.Utils - import Cachex.Spec def child_spec(_) do - stats = Application.get_env(:logflare, :cache_stats, false) - - %{ - id: __MODULE__, - start: - {Cachex, :start_link, - [ - __MODULE__, - [ - warmers: [ - warmer(required: false, module: Users.CacheWarmer, name: Users.CacheWarmer) - ], - hooks: - [ - if(stats, do: Utils.cache_stats()), - Utils.cache_limit(100_000) - ] - |> Enum.filter(& &1), - expiration: Utils.cache_expiration_min(180, 10) - ] - ]} - } + CachexOps.child_spec(__MODULE__, + limit: 100_000, + ttl: to_timeout(hour: 3), + purge_interval: to_timeout(minute: 10), + warmer: Users.CacheWarmer + ) end def update(user), diff --git a/lib/logflare/utils.ex b/lib/logflare/utils.ex index 664f1b6749..3f0ff5ad76 100644 --- a/lib/logflare/utils.ex +++ b/lib/logflare/utils.ex @@ -8,7 +8,6 @@ defmodule Logflare.Utils do alias Logflare.Backends.Backend alias Logflare.OauthAccessTokens.OauthAccessToken alias Logflare.OauthAccessTokens.PartnerOauthAccessToken - import Cachex.Spec import Logflare.Utils.Guards, only: [is_atom_value: 1] @sensitive_header_names ["authorization", "x-api-key", "Authorization", "X-API-Key"] @@ -67,49 +66,6 @@ defmodule Logflare.Utils do @original_to_string String.Chars.impl_for(%Tesla.Env{}) def tesla_env_to_string, do: @original_to_string - def cache_stats do - hook(module: Cachex.Stats) - end - - def cache_limit(n) when is_integer(n) do - hook( - module: Cachex.Limit.Scheduled, - args: { - # setting cache max size - n, - # options for `Cachex.prune/3` - [], - # options for `Cachex.Limit.Scheduled` - [] - } - ) - end - - @doc """ - Builds a long Cachex expiration spec - Defaults to 20 min with 5 min cleanup intervals - """ - @spec cache_expiration_min(non_neg_integer(), non_neg_integer()) :: Cachex.Spec.expiration() - def cache_expiration_min(default \\ 20, interval \\ 5) do - cache_expiration_sec(default * 60, interval * 60) - end - - @doc """ - Builds a short Cachex expiration spec - Defaults to 60 sec with 20 sec cleanup intervals - """ - @spec cache_expiration_sec(non_neg_integer(), non_neg_integer()) :: Cachex.Spec.expiration() - def cache_expiration_sec(default \\ 60, interval \\ 20) do - expiration( - # default record expiration of 20 mins - default: :timer.seconds(default), - # how often cleanup should occur, 5 mins - interval: :timer.seconds(interval), - # whether to enable lazy checking - lazy: true - ) - end - @doc """ Stringifies an atom map to a string map. diff --git a/lib/logflare_web/controllers/health_check_controller.ex b/lib/logflare_web/controllers/health_check_controller.ex index 7ea5f126ca..fc27a21b73 100644 --- a/lib/logflare_web/controllers/health_check_controller.ex +++ b/lib/logflare_web/controllers/health_check_controller.ex @@ -106,11 +106,7 @@ defmodule LogflareWeb.HealthCheckController do Logflare.Logs.LogEvents.Cache ], into: %{} do - # call is O(1) - case Cachex.size(cache) do - {:ok, _} -> {cache, :ok} - {:error, :no_cache} -> {cache, :no_cache} - end + {cache, if(cache.healthy?(), do: :ok, else: :unhealthy)} end end end diff --git a/lib/telemetry.ex b/lib/telemetry.ex index fc3e5d6921..b64c3ac7b9 100644 --- a/lib/telemetry.ex +++ b/lib/telemetry.ex @@ -9,6 +9,7 @@ defmodule Logflare.Telemetry do alias Logflare.Backends.Adaptor.ClickHouseAdaptor alias Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryConnectionSup + alias Logflare.Cache.CachexOps def start_link(arg), do: Supervisor.start_link(__MODULE__, arg, name: __MODULE__) @@ -174,14 +175,12 @@ defmodule Logflare.Telemetry do # isn't the case in dev/test, so there'd otherwise be no way to exercise it. @doc false def metrics do - cache_stats? = Application.get_env(:logflare, :cache_stats, false) + cache_stats? = CachexOps.stats_enabled?() cache_metrics = if cache_stats? do Enum.flat_map(@caches, fn {_cache, metric} -> [ - last_value("cachex.#{metric}.purge"), - last_value("cachex.#{metric}.stats"), last_value("cachex.#{metric}.evictions"), last_value("cachex.#{metric}.expirations"), last_value("cachex.#{metric}.operations"), @@ -753,7 +752,7 @@ defmodule Logflare.Telemetry do end defp periodic_measurements do - cache_stats? = Application.get_env(:logflare, :cache_stats, false) + cache_stats? = CachexOps.stats_enabled?() cachex_metrics = if cache_stats? do @@ -864,27 +863,7 @@ defmodule Logflare.Telemetry do def cachex_metrics do Enum.each(@caches, fn {cache, metric} -> - {:ok, stats} = Cachex.stats(cache) - - {:total_heap_size, total_heap_size} = - cache - |> Process.whereis() - |> Process.info(:total_heap_size) - - metrics = %{ - purge: Map.get(stats, :purge, 0), - stats: Map.get(stats, :stats, 0), - evictions: Map.get(stats, :evictions, 0), - expirations: Map.get(stats, :expirations, 0), - operations: Map.get(stats, :operations, 0), - hits: Map.get(stats, :hits, 0), - misses: Map.get(stats, :misses, 0), - hit_rate: Map.get(stats, :hit_rate, 0), - miss_rate: Map.get(stats, :miss_rate, 0), - total_heap_size: total_heap_size - } - - :telemetry.execute([:cachex, metric], metrics) + :telemetry.execute([:cachex, metric], cache.stats()) end) end diff --git a/mix.exs b/mix.exs index 8073d08a8b..c5b92a719b 100644 --- a/mix.exs +++ b/mix.exs @@ -283,7 +283,7 @@ defmodule Logflare.Mixfile do "test.compile": ["compile --warnings-as-errors"], "test.format": ["format --check-formatted"], "test.security": ["sobelow --threshold high --ignore Config.HTTPS"], - "test.slop": ["ex_dna --max-clones 26"], + "test.slop": ["ex_dna --max-clones 25"], "test.structure": [ "reach.check --smells --strict --baseline .reach.baseline.json" ], diff --git a/test/logflare/cache/cachex_ops_test.exs b/test/logflare/cache/cachex_ops_test.exs new file mode 100644 index 0000000000..c88ba0da05 --- /dev/null +++ b/test/logflare/cache/cachex_ops_test.exs @@ -0,0 +1,106 @@ +defmodule Logflare.Cache.CachexOpsTest do + use ExUnit.Case, async: false + + import Cachex.Spec + + alias Logflare.Cache.CachexOps + + @cache __MODULE__.Cache + + describe "child_spec/2" do + test "started under a supervisor" do + start_cache(limit: 10) + + assert {:ok, true} = Cachex.put(@cache, :key, :value) + assert {:ok, :value} = Cachex.get(@cache, :key) + end + + for {stats_enabled?, expected} <- [{true, :ok}, {false, :error}] do + test "cache stats config set to #{stats_enabled?}" do + original = Application.get_env(:logflare, :cache_stats) + on_exit(fn -> Application.put_env(:logflare, :cache_stats, original) end) + Application.put_env(:logflare, :cache_stats, unquote(stats_enabled?)) + + start_cache(limit: 10) + Cachex.put(@cache, :key, :value) + + assert {unquote(expected), _} = Cachex.stats(@cache) + assert :ok = CachexOps.reset(@cache) + assert {:ok, 0} = Cachex.size(@cache) + end + end + end + + describe "cachex_opts/1" do + for {name, opts, expected} <- [ + {"only a nil limit", [limit: nil], + [ + hooks: [], + expiration: expiration(default: :timer.minutes(20), interval: :timer.minutes(5)), + warmers: [], + compressed: false + ]}, + {"an integer limit", [limit: 10], + [hooks: [hook(module: Cachex.Limit.Scheduled, args: {10, [], []})]]}, + {"a warmer module", [limit: nil, warmer: __MODULE__.Warmer], + [ + warmers: [ + warmer(module: __MODULE__.Warmer, name: __MODULE__.Warmer, required: false) + ] + ]}, + {"a warmer with interval", [limit: nil, warmer: {__MODULE__.Warmer, interval: 1_000}], + [ + warmers: [ + warmer( + module: __MODULE__.Warmer, + name: __MODULE__.Warmer, + required: false, + interval: 1_000 + ) + ] + ]}, + {"ttl and purge interval", [limit: nil, ttl: 10_000, purge_interval: 2_000], + [expiration: expiration(default: 10_000, interval: 2_000, lazy: true)]}, + {"compression", [limit: nil, compressed: true], [compressed: true]} + ] do + test name do + opts = CachexOps.cachex_opts(unquote(Macro.escape(opts))) + + for {key, value} <- unquote(Macro.escape(expected)) do + assert Keyword.fetch!(opts, key) == value + end + end + end + + test "missing limit" do + assert_raise KeyError, fn -> CachexOps.cachex_opts(compressed: true) end + end + end + + describe "delete_keys/2" do + setup do + start_cache(limit: nil) + Cachex.put_many(@cache, a: 1, b: 2) + :ok + end + + for {name, keys, expected_count} <- [ + {"existing keys", [:a, :b], 2}, + {"missing keys", [:x, :y], 0}, + {"existing and missing keys", [:a, :x], 1}, + {"no keys", [], 0} + ] do + test name do + keys = unquote(keys) + + assert {:ok, unquote(expected_count)} = CachexOps.delete_keys(@cache, keys) + + for key <- keys do + assert {:ok, false} = Cachex.exists?(@cache, key) + end + end + end + end + + defp start_cache(opts), do: start_supervised!(CachexOps.child_spec(@cache, opts)) +end diff --git a/test/logflare/cache_test.exs b/test/logflare/cache_test.exs new file mode 100644 index 0000000000..3ad1ac405f --- /dev/null +++ b/test/logflare/cache_test.exs @@ -0,0 +1,81 @@ +defmodule Logflare.CacheTest do + use ExUnit.Case, async: true + + alias Logflare.Cache.CachexOps + + defmodule TestCache do + use Logflare.Cache + end + + defmodule NotStartedCache do + use Logflare.Cache + end + + defmodule RecordingImpl do + def healthy?(cache), do: record({:healthy?, cache}, true) + def stats(cache), do: record({:stats, cache}, %{}) + def reset(cache), do: record({:reset, cache}, :ok) + + defp record(call, result) do + send(self(), call) + result + end + end + + defmodule CustomImplCache do + use Logflare.Cache, impl: RecordingImpl + end + + setup do + start_supervised!(CachexOps.child_spec(TestCache, limit: 100)) + :ok + end + + test "started cache" do + assert TestCache.healthy?() + end + + test "cache that is not started" do + refute NotStartedCache.healthy?() + end + + test "cache after reads and writes" do + Cachex.put(TestCache, :key, :value) + Cachex.get(TestCache, :key) + Cachex.get(TestCache, :missing) + + assert %{hits: 1, misses: 1, total_heap_size: heap} = stats = TestCache.stats() + assert heap > 0 + + assert stats |> Map.keys() |> Enum.sort() == [ + :evictions, + :expirations, + :hit_rate, + :hits, + :miss_rate, + :misses, + :operations, + :total_heap_size + ] + end + + test "cache with entries and stats, after reset" do + Cachex.put(TestCache, :key, :value) + Cachex.get(TestCache, :key) + + assert :ok = TestCache.reset() + + assert {:ok, nil} = Cachex.get(TestCache, :key) + assert %{hits: 0} = TestCache.stats() + end + + test "cache with a custom impl module" do + assert CustomImplCache.healthy?() + assert %{} = CustomImplCache.stats() + assert :ok = CustomImplCache.reset() + + assert_received {:healthy?, CustomImplCache} + assert_received {:stats, CustomImplCache} + assert_received {:reset, CustomImplCache} + end +end diff --git a/test/logflare/context_cache_test.exs b/test/logflare/context_cache_test.exs index d753a5263a..870d214290 100644 --- a/test/logflare/context_cache_test.exs +++ b/test/logflare/context_cache_test.exs @@ -9,6 +9,11 @@ defmodule Logflare.ContextCacheTest do alias Logflare.Backends alias Logflare.Backends.Backend alias Logflare.Auth + alias Logflare.Cache.CachexOps + + defmodule TestCache do + use Logflare.ContextCache + end describe "ContextCache" do setup do @@ -54,6 +59,33 @@ defmodule Logflare.ContextCacheTest do end end + describe "default bust_by/1" do + setup do + start_supervised!(CachexOps.child_spec(TestCache, limit: nil)) + :ok + end + + for {name, value, expected_count} <- [ + {"map with the id", %{id: 1}, 1}, + {":ok tuple with a map with the id", {:ok, %{id: 1}}, 1}, + {"list containing a map with the id", [%{id: 2}, %{id: 1}], 1}, + {"map with another id", %{id: 2}, 0}, + {"list without the id", [%{id: 2}], 0}, + {"nil", nil, 0} + ] do + test "cached #{name}" do + Cachex.put(TestCache, :key, {:cached, unquote(Macro.escape(value))}) + + assert {:ok, unquote(expected_count)} = TestCache.bust_by(id: 1) + assert {:ok, unquote(expected_count == 0)} = Cachex.exists?(TestCache, :key) + end + end + + test "keyword other than id" do + assert_raise ArgumentError, fn -> TestCache.bust_by(source_id: 1) end + end + end + describe "unboxed transaction" do setup do on_exit(fn -> diff --git a/test/logflare/system_cache_test.exs b/test/logflare/system_cache_test.exs index 69078c97ab..251941ef08 100644 --- a/test/logflare/system_cache_test.exs +++ b/test/logflare/system_cache_test.exs @@ -44,8 +44,7 @@ defmodule Logflare.SystemCacheTest do Application.put_env(:logflare, :env, :prod) on_exit(fn -> Application.put_env(:logflare, :env, previous_env) end) - assert %{start: {Cachex, :start_link, [SystemCache, options]}} = SystemCache.child_spec(nil) - assert [warmer_spec] = Keyword.fetch!(options, :warmers) + assert [warmer_spec] = Keyword.fetch!(cachex_options(), :warmers) refute warmer(warmer_spec, :required) end @@ -55,7 +54,7 @@ defmodule Logflare.SystemCacheTest do on_exit(fn -> Application.put_env(:logflare, :env, previous_env) end) set_mimic_global() - assert %{start: {Cachex, :start_link, [SystemCache, options]}} = SystemCache.child_spec(nil) + options = cachex_options() assert [warmer_spec] = Keyword.fetch!(options, :warmers) cache_name = Module.concat(__MODULE__, IntegrationCache) @@ -92,4 +91,9 @@ defmodule Logflare.SystemCacheTest do assert log =~ "SystemCache warmer failed" assert log =~ "monitor_unavailable" end + + defp cachex_options do + %{start: {Cachex, :start_link, [[_name, options]]}} = SystemCache.child_spec(nil) + options + end end diff --git a/test/logflare/telemetry_test.exs b/test/logflare/telemetry_test.exs index 0debc4968c..3cb8326de8 100644 --- a/test/logflare/telemetry_test.exs +++ b/test/logflare/telemetry_test.exs @@ -891,8 +891,6 @@ defmodule Logflare.TelemetryTest do :miss_rate, :misses, :operations, - :purge, - :stats, :total_heap_size ] diff --git a/test/logflare_web/controllers/health_check_controller_test.exs b/test/logflare_web/controllers/health_check_controller_test.exs index d533150992..9606235594 100644 --- a/test/logflare_web/controllers/health_check_controller_test.exs +++ b/test/logflare_web/controllers/health_check_controller_test.exs @@ -39,6 +39,21 @@ defmodule LogflareWeb.HealthCheckControllerTest do } = json_response(conn, 200) end + test "cache reporting unhealthy", %{conn: conn} do + start_supervised!(Source.Supervisor) + stub(Logflare.Partners.Cache, :healthy?, fn -> false end) + + conn = get(conn, "/health") + + assert %{ + "status" => "coming_up", + "caches" => %{ + "Elixir.Logflare.Partners.Cache" => "unhealthy", + "Elixir.Logflare.Auth.Cache" => "ok" + } + } = json_response(conn, 503) + end + test "readiness check", %{conn: conn} do start_supervised!(Source.Supervisor) :timer.sleep(1000) diff --git a/test/profiling/context_cache_bust_by_kw_bench.exs b/test/profiling/context_cache_bust_by_kw_bench.exs index e7446b44d9..f635b2c702 100644 --- a/test/profiling/context_cache_bust_by_kw_bench.exs +++ b/test/profiling/context_cache_bust_by_kw_bench.exs @@ -18,9 +18,9 @@ end Benchee.run( %{ + # Rules.Cache overrides bust_by/1, so call the default match-spec scan directly "bust_keys by primary key" => fn [{_source, rule} | _] -> - pkey = rule.id - ContextCache.bust_keys([{Rules, pkey}]) + ContextCache.bust_by(cache, id: rule.id) end, "bust_keys by relation key" => fn [{source, _rule} | _] -> ContextCache.bust_keys([{Rules, source_id: source.id}]) diff --git a/test/support/conn_case.ex b/test/support/conn_case.ex index e7640168a7..16c4ac15be 100644 --- a/test/support/conn_case.ex +++ b/test/support/conn_case.ex @@ -79,7 +79,7 @@ defmodule LogflareWeb.ConnCase do end) caches = Logflare.ContextCache.Supervisor.list_caches() - Enum.each(caches, &Cachex.reset(&1, hooks: [Cachex.Stats])) + Enum.each(caches, & &1.reset()) on_exit(fn -> IngestEventQueue.delete_all_mappings() diff --git a/test/support/data_case.ex b/test/support/data_case.ex index 53db5a9ae6..4d86aa2e84 100644 --- a/test/support/data_case.ex +++ b/test/support/data_case.ex @@ -45,7 +45,7 @@ defmodule Logflare.DataCase do end) caches = Logflare.ContextCache.Supervisor.list_caches() - Enum.each(caches, &Cachex.reset(&1, hooks: [Cachex.Stats])) + Enum.each(caches, & &1.reset()) on_exit(fn -> # Deterministic, not timer-based: IngestEventQueue's generation-store diff --git a/test/test_helper.exs b/test/test_helper.exs index 3b2545b16e..0f897fa99f 100644 --- a/test/test_helper.exs +++ b/test/test_helper.exs @@ -74,6 +74,7 @@ Mimic.copy(Logflare.Lql) Mimic.copy(Logflare.Mailer) Mimic.copy(Logflare.NaturalLanguageLql.AnthropicClient) Mimic.copy(Logflare.Networking.GrpcPool) +Mimic.copy(Logflare.Partners.Cache) Mimic.copy(Logflare.Rules) Mimic.copy(Logflare.SingleTenant) Mimic.copy(Logflare.Sources) From 62753e2796a689cce25be7884d0d8775fb0a2b5d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bartosz=20B=C5=82aszk=C3=B3w?= Date: Wed, 30 Sep 2026 09:50:44 +0200 Subject: [PATCH 2/6] fix: report cachex total_heap_size in bytes Process.info(pid, :total_heap_size) returns the heap size in words, but the cachex..total_heap_size metric is declared with unit {:byte, :megabyte}, so the exported value was too small by a factor of the word size (8x on 64-bit). Convert words to bytes with :erlang.system_info(:wordsize) before emitting it. --- lib/logflare/cache.ex | 2 +- lib/logflare/cache/cachex_ops.ex | 6 ++++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/lib/logflare/cache.ex b/lib/logflare/cache.ex index 7360bf5a38..c43785437e 100644 --- a/lib/logflare/cache.ex +++ b/lib/logflare/cache.ex @@ -10,7 +10,7 @@ defmodule Logflare.Cache do alias Logflare.Cache.CachexOps @typedoc """ - Counters since the last `c:reset/0`. Rates are percentages (0-100); `total_heap_size` is in words. + Counters since the last `c:reset/0`. Rates are percentages (0-100); `total_heap_size` is in bytes. """ @type stats() :: %{ evictions: non_neg_integer(), diff --git a/lib/logflare/cache/cachex_ops.ex b/lib/logflare/cache/cachex_ops.ex index 9b25a40172..1ce77234cd 100644 --- a/lib/logflare/cache/cachex_ops.ex +++ b/lib/logflare/cache/cachex_ops.ex @@ -70,12 +70,14 @@ defmodule Logflare.Cache.CachexOps do def stats(cache) do {:ok, stats} = Cachex.stats(cache) - {:total_heap_size, total_heap_size} = + {:total_heap_size, heap_words} = cache |> Process.whereis() |> Process.info(:total_heap_size) - for key <- @stat_keys, into: %{total_heap_size: total_heap_size} do + heap_bytes = heap_words * :erlang.system_info(:wordsize) + + for key <- @stat_keys, into: %{total_heap_size: heap_bytes} do {key, Map.get(stats, key, 0)} end end From 57ae000a77c3bdcd99812e12175ce0d1262998c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bartosz=20B=C5=82aszk=C3=B3w?= Date: Thu, 1 Oct 2026 10:55:15 +0200 Subject: [PATCH 3/6] refactor: route ContextCache reads and busting through an Ops implementation Logflare.ContextCache still called Cachex directly for the read path (apply_fun/3, update/4) and for the default primary-key bust. - ContextCache gains required fetch/2 and update/2 callbacks next to bust_by/1. `use Logflare.ContextCache` injects overridable defaults that delegate to the `impl:` module (Logflare.Cache.CachexOps by default), and apply_fun/3, update/4 and bust_keys/1 only call these callbacks. - The impl contract is explicit: Logflare.Cache.Ops (healthy?/1, stats/1, reset/1) and Logflare.ContextCache.Ops (bust_by/2, fetch/3, update/3). - The Cachex read path, the primary-key bust and the gossip of cache misses move to CachexOps. The `{:cached, value}` wrapper is now private to it. - ContextCache.fetch/3 and ContextCache.bust_by/2 are removed; callers use CachexOps directly. Still Cachex-only: ContextCache.Gossip's receive side, cache warmers, and cache modules that call Cachex themselves. Co-Authored-By: Claude Opus 5.5 --- lib/logflare/cache.ex | 2 +- lib/logflare/cache/cachex_ops.ex | 85 +++++++++++- lib/logflare/cache/ops.ex | 12 ++ lib/logflare/context_cache.ex | 127 +++++------------- lib/logflare/context_cache/ops.ex | 12 ++ lib/logflare/rules/cache.ex | 2 +- test/logflare/cache/cachex_ops_test.exs | 35 +++++ test/logflare/cache_test.exs | 5 + test/logflare/context_cache_test.exs | 50 +++++++ .../context_cache_bust_by_kw_bench.exs | 3 +- 10 files changed, 234 insertions(+), 99 deletions(-) create mode 100644 lib/logflare/cache/ops.ex create mode 100644 lib/logflare/context_cache/ops.ex diff --git a/lib/logflare/cache.ex b/lib/logflare/cache.ex index c43785437e..0929bc0752 100644 --- a/lib/logflare/cache.ex +++ b/lib/logflare/cache.ex @@ -4,7 +4,7 @@ defmodule Logflare.Cache do `use Logflare.Cache` injects default implementations of every callback, delegating to `Logflare.Cache.CachexOps`. A cache on another backend overrides them, or passes - `impl: module` whose `healthy?/1`, `stats/1` and `reset/1` take the cache module. + `impl: module` naming a `Logflare.Cache.Ops` implementation. """ alias Logflare.Cache.CachexOps diff --git a/lib/logflare/cache/cachex_ops.ex b/lib/logflare/cache/cachex_ops.ex index 1ce77234cd..67790d1a3b 100644 --- a/lib/logflare/cache/cachex_ops.ex +++ b/lib/logflare/cache/cachex_ops.ex @@ -1,10 +1,18 @@ defmodule Logflare.Cache.CachexOps do @moduledoc """ - Cachex implementations of the `Logflare.Cache` callbacks, and builders for Cachex start options. + Cachex implementations of the `Logflare.Cache` and `Logflare.ContextCache` callbacks, and builders + for Cachex start options. + + Context cache values are stored as `{:cached, value}`, because Cachex treats a stored `nil` as a miss. """ + @behaviour Logflare.Cache.Ops + @behaviour Logflare.ContextCache.Ops + import Cachex.Spec + alias Logflare.ContextCache.Gossip + @stat_keys [ :evictions, :expirations, @@ -60,12 +68,14 @@ defmodule Logflare.Cache.CachexOps do @spec stats_enabled?() :: boolean() def stats_enabled?, do: Application.get_env(:logflare, :cache_stats, false) + @impl Logflare.Cache.Ops @spec healthy?(Cachex.t()) :: boolean() def healthy?(cache), do: match?({:ok, _}, Cachex.size(cache)) @doc """ Raises when the cache runs without the stats hook. """ + @impl Logflare.Cache.Ops @spec stats(Cachex.t()) :: Logflare.Cache.stats() def stats(cache) do {:ok, stats} = Cachex.stats(cache) @@ -82,12 +92,85 @@ defmodule Logflare.Cache.CachexOps do end end + @impl Logflare.Cache.Ops @spec reset(Cachex.t()) :: :ok def reset(cache) do {:ok, true} = Cachex.reset(cache, hooks: [Cachex.Stats]) :ok end + @doc """ + With `[id: pkey]`, busts every entry whose cached value is a map with that `:id`, an `{:ok, map}` + with that `:id`, or a list containing such a map. Raises `ArgumentError` for any other keyword. + + It scans the cache with a match spec instead of keeping a reverse index of primary keys. + """ + @impl Logflare.ContextCache.Ops + @spec bust_by(Cachex.t(), keyword()) :: {:ok, non_neg_integer()} + def bust_by(cache, id: pkey) do + filter = + { + # use orelse to prevent 2nd condition failing as value is not a map + :orelse, + { + :orelse, + # handle lists + {:is_list, {:element, 2, :value}}, + # handle :ok tuples when struct with id is in 2nd element pos. + {:andalso, {:is_tuple, {:element, 2, :value}}, + {:andalso, {:==, {:element, 1, {:element, 2, :value}}, :ok}, + {:andalso, {:is_map, {:element, 2, {:element, 2, :value}}}, + {:==, {:map_get, :id, {:element, 2, {:element, 2, :value}}}, pkey}}}} + }, + # handle single maps + {:andalso, {:is_map, {:element, 2, :value}}, + {:==, {:map_get, :id, {:element, 2, :value}}, pkey}} + } + + query = Cachex.Query.build(where: filter, output: {:key, :value}) + + keys = + cache + |> Cachex.stream!(query) + |> Stream.filter(fn + {_k, {:cached, v}} when is_list(v) -> Enum.any?(v, &(&1.id == pkey)) + {_k, _v} -> true + end) + |> Stream.map(fn {k, _v} -> k end) + + delete_keys(cache, keys) + end + + def bust_by(cache, kw) do + raise ArgumentError, "#{inspect(cache)} does not support busting by #{inspect(kw)}" + end + + @doc """ + Returns the cached value for `key`, calling `getter` and caching its result on a miss. + + A miss is also multicast to peer nodes, see `Logflare.ContextCache.Gossip`. Accepts a Cachex + worker, so calls can be batched in `Cachex.execute/2`. + """ + @impl Logflare.ContextCache.Ops + @spec fetch(Cachex.t(), term(), (-> term())) :: term() + def fetch(cache, key, getter) do + case Cachex.fetch(cache, key, fn _key -> {:commit, {:cached, getter.()}} end) do + {:commit, {:cached, value}} -> + Gossip.multicast(cache, key, value) + value + + {:ok, {:cached, value}} -> + value + end + end + + @impl Logflare.ContextCache.Ops + @spec update(Cachex.t(), term(), term()) :: :ok + def update(cache, key, value) do + {:ok, _updated?} = Cachex.update(cache, key, {:cached, value}) + :ok + end + @doc """ Deletes `keys` and returns how many of them were present. """ diff --git a/lib/logflare/cache/ops.ex b/lib/logflare/cache/ops.ex new file mode 100644 index 0000000000..11ea1ac22c --- /dev/null +++ b/lib/logflare/cache/ops.ex @@ -0,0 +1,12 @@ +defmodule Logflare.Cache.Ops do + @moduledoc """ + Implementation of the `Logflare.Cache` callbacks on one storage backend. + + Selected with `use Logflare.Cache, impl: module`. Every function takes the cache module + the callback was called on. + """ + + @callback healthy?(cache :: module()) :: boolean() + @callback stats(cache :: module()) :: Logflare.Cache.stats() + @callback reset(cache :: module()) :: :ok +end diff --git a/lib/logflare/context_cache.ex b/lib/logflare/context_cache.ex index 3b8b393f8b..f03170ea3b 100644 --- a/lib/logflare/context_cache.ex +++ b/lib/logflare/context_cache.ex @@ -6,54 +6,59 @@ defmodule Logflare.ContextCache do e.g. `Logflare.Users.Cache` functions go through `apply_fun/3` and results of those functions are returned to the caller and cached in the respective cache. - ## Cache Implementation + ## Implementation - The cache implementation directly queries the relevant context cache to be busted and performs - primary key checking within the matchspec. This approach queries across a narrower set of records, - providing better performance compared to a reverse index approach. + `use Logflare.ContextCache` makes the module a `Logflare.Cache` and injects overridable defaults + for `c:fetch/2`, `c:update/2` and `c:bust_by/1`. They delegate to a `Logflare.ContextCache.Ops` + module, `Logflare.Cache.CachexOps` unless `impl: module` is given. - `use Logflare.ContextCache` makes the module a `Logflare.Cache` and injects a default `c:bust_by/1` - that busts by `id:` (see `bust_by/2`). Caches that need other busting keys override it. + ## Busting - ## List Busting - - The cache supports busting records within lists. If a struct in a non-empty list contains - the :id field, the record will get busted when that ID is encountered in the write-ahead log. + `bust_keys/1` busts entries by primary key or by a keyword of fields. Caches that need busting + keys other than `id:` override `c:bust_by/1`. ## Memoization This module can also be used to cache heavy functions or db calls hidden behind a 3rd party - library. See `Logflare.Auth.Cache` for an example. In this example, the `expiration` set in that - Cachex child_spec is handling the cache expiration. - - In the case functions don't return a response with a primary key, or something else we can - bust the cache on, it will get reverse indexed with `select_key/1` as `:unknown`. - - ## Gossip - - Cache misses are optionally multicast to peer nodes via `:erpc` to warm the cluster. - To prevent race conditions, WAL invalidations write short-lived tombstones that - filter out stale incoming messages. + library. See `Logflare.Auth.Cache` for an example, where the cache's own expiration is the only + invalidation. """ alias Logflare.Cache.CachexOps - alias Logflare.ContextCache.Gossip @doc """ Busts cache entries by a keyword of values, returning the number of entries busted. """ @callback bust_by(keyword()) :: {:ok, non_neg_integer()} | {:error, term()} + @doc """ + Returns the cached value for `key`, calling `getter` and caching its result on a miss. + """ + @callback fetch(key :: term(), getter :: (-> term())) :: term() + + @doc """ + Replaces the value cached for `key`. Does nothing when `key` is not cached. + """ + @callback update(key :: term(), value :: term()) :: :ok + defmacro __using__(opts) do + impl = Keyword.get(opts, :impl, CachexOps) + quote do use Logflare.Cache, unquote(opts) @behaviour Logflare.ContextCache @impl Logflare.ContextCache - def bust_by(kw), do: Logflare.ContextCache.bust_by(__MODULE__, kw) + def bust_by(kw), do: unquote(impl).bust_by(__MODULE__, kw) + + @impl Logflare.ContextCache + def fetch(key, getter), do: unquote(impl).fetch(__MODULE__, key, getter) + + @impl Logflare.ContextCache + def update(key, value), do: unquote(impl).update(__MODULE__, key, value) - defoverridable bust_by: 1 + defoverridable bust_by: 1, fetch: 2, update: 2 end end @@ -61,10 +66,7 @@ defmodule Logflare.ContextCache do def apply_fun(context, {fun, _arity}, args), do: apply_fun(context, fun, args) def apply_fun(context, fun, args) when is_atom(fun) do - cache = cache_name(context) - cache_key = {fun, args} - - fetch(cache, cache_key, fn -> + cache_name(context).fetch({fun, args}, fn -> Logflare.Repo.apply_with_replica(context, fun, args) end) end @@ -72,11 +74,9 @@ defmodule Logflare.ContextCache do @doc """ Updates cache entry to the given value """ + @spec update(module(), atom(), list(), term()) :: :ok def update(context, fun, args, value) when is_atom(fun) do - cache = cache_name(context) - cache_key = {fun, args} - - Cachex.update(cache, cache_key, {:cached, value}) + cache_name(context).update({fun, args}, value) end @doc """ @@ -102,71 +102,8 @@ defmodule Logflare.ContextCache do defp bust_key(context, kw) when is_list(kw), do: cache_name(context).bust_by(kw) defp bust_key(context, pkey), do: cache_name(context).bust_by(id: pkey) - @doc """ - Default `c:bust_by/1`: with `[id: pkey]`, busts every entry whose cached value is a map with that `:id`, - an `{:ok, map}` with that `:id`, or a list containing such a map. Raises `ArgumentError` for any other keyword. - """ - @spec bust_by(Cachex.t(), keyword()) :: {:ok, non_neg_integer()} - def bust_by(cache, id: pkey) do - filter = - { - # use orelse to prevent 2nd condition failing as value is not a map - :orelse, - { - :orelse, - # handle lists - {:is_list, {:element, 2, :value}}, - # handle :ok tuples when struct with id is in 2nd element pos. - {:andalso, {:is_tuple, {:element, 2, :value}}, - {:andalso, {:==, {:element, 1, {:element, 2, :value}}, :ok}, - {:andalso, {:is_map, {:element, 2, {:element, 2, :value}}}, - {:==, {:map_get, :id, {:element, 2, {:element, 2, :value}}}, pkey}}}} - }, - # handle single maps - {:andalso, {:is_map, {:element, 2, :value}}, - {:==, {:map_get, :id, {:element, 2, :value}}, pkey}} - } - - query = Cachex.Query.build(where: filter, output: {:key, :value}) - - keys = - cache - |> Cachex.stream!(query) - |> Stream.filter(fn - {_k, {:cached, v}} when is_list(v) -> Enum.any?(v, &(&1.id == pkey)) - {_k, _v} -> true - end) - |> Stream.map(fn {k, _v} -> k end) - - CachexOps.delete_keys(cache, keys) - end - - def bust_by(cache, kw) do - raise ArgumentError, "#{inspect(cache)} does not support busting by #{inspect(kw)}" - end - @spec cache_name(atom()) :: atom() def cache_name(context) do Module.concat(context, Cache) end - - @doc """ - Low level API for fetching from cache. Allows wrapping calls with - `Cachex.execute/2` and accessing arbitrary key or calling any getter function. - """ - @spec fetch(Cachex.t(), {atom(), list()}, fun()) :: term() - def fetch(cache, cache_key, getter_fn) do - case Cachex.fetch(cache, cache_key, fn _cache_key -> - # Use a `:cached` tuple here otherwise when an fn returns nil Cachex will miss - # the cache because it thinks ETS returned nil - {:commit, {:cached, getter_fn.()}} - end) do - {:commit, {:cached, value}} -> - Gossip.multicast(cache, cache_key, value) - value - - {:ok, {:cached, value}} -> - value - end - end end diff --git a/lib/logflare/context_cache/ops.ex b/lib/logflare/context_cache/ops.ex new file mode 100644 index 0000000000..5b552beb02 --- /dev/null +++ b/lib/logflare/context_cache/ops.ex @@ -0,0 +1,12 @@ +defmodule Logflare.ContextCache.Ops do + @moduledoc """ + Implementation of the `Logflare.ContextCache` callbacks on one storage backend. + + Selected with `use Logflare.ContextCache, impl: module`, which also requires the module to + implement `Logflare.Cache.Ops`. Every function takes the cache module the callback was called on. + """ + + @callback bust_by(cache :: module(), keyword()) :: {:ok, non_neg_integer()} | {:error, term()} + @callback fetch(cache :: module(), key :: term(), getter :: (-> term())) :: term() + @callback update(cache :: module(), key :: term(), value :: term()) :: :ok +end diff --git a/lib/logflare/rules/cache.ex b/lib/logflare/rules/cache.ex index 58e08c4974..64d36b6bad 100644 --- a/lib/logflare/rules/cache.ex +++ b/lib/logflare/rules/cache.ex @@ -54,7 +54,7 @@ defmodule Logflare.Rules.Cache do end defp fetch_rule(cache, id) do - ContextCache.fetch(cache, {:get_rule, [id]}, fn -> + CachexOps.fetch(cache, {:get_rule, [id]}, fn -> Repo.with_replica(fn -> Rules.get_rule(id) end) end) end diff --git a/test/logflare/cache/cachex_ops_test.exs b/test/logflare/cache/cachex_ops_test.exs index c88ba0da05..d013ee84fb 100644 --- a/test/logflare/cache/cachex_ops_test.exs +++ b/test/logflare/cache/cachex_ops_test.exs @@ -102,5 +102,40 @@ defmodule Logflare.Cache.CachexOpsTest do end end + describe "fetch/3" do + setup do + start_cache(limit: nil) + :ok + end + + for value <- [:value, nil] do + test "missing key with a getter returning #{inspect(value)}" do + value = unquote(value) + + assert CachexOps.fetch(@cache, :key, fn -> value end) == value + assert CachexOps.fetch(@cache, :key, fn -> flunk("getter called on a hit") end) == value + end + end + end + + describe "update/3" do + setup do + start_cache(limit: nil) + :ok + end + + test "cached key" do + CachexOps.fetch(@cache, :key, fn -> :old end) + + assert :ok = CachexOps.update(@cache, :key, :new) + assert :new = CachexOps.fetch(@cache, :key, fn -> flunk("getter called on a hit") end) + end + + test "missing key" do + assert :ok = CachexOps.update(@cache, :key, :new) + assert {:ok, false} = Cachex.exists?(@cache, :key) + end + end + defp start_cache(opts), do: start_supervised!(CachexOps.child_spec(@cache, opts)) end diff --git a/test/logflare/cache_test.exs b/test/logflare/cache_test.exs index 3ad1ac405f..7a04d18d8b 100644 --- a/test/logflare/cache_test.exs +++ b/test/logflare/cache_test.exs @@ -12,8 +12,13 @@ defmodule Logflare.CacheTest do end defmodule RecordingImpl do + @behaviour Logflare.Cache.Ops + + @impl true def healthy?(cache), do: record({:healthy?, cache}, true) + @impl true def stats(cache), do: record({:stats, cache}, %{}) + @impl true def reset(cache), do: record({:reset, cache}, :ok) defp record(call, result) do diff --git a/test/logflare/context_cache_test.exs b/test/logflare/context_cache_test.exs index 870d214290..912ceb8c30 100644 --- a/test/logflare/context_cache_test.exs +++ b/test/logflare/context_cache_test.exs @@ -15,6 +15,34 @@ defmodule Logflare.ContextCacheTest do use Logflare.ContextCache end + defmodule RecordingOps do + @behaviour Logflare.Cache.Ops + @behaviour Logflare.ContextCache.Ops + + @impl Logflare.Cache.Ops + def healthy?(cache), do: record({:healthy?, cache}, true) + @impl Logflare.Cache.Ops + def stats(cache), do: record({:stats, cache}, %{}) + @impl Logflare.Cache.Ops + def reset(cache), do: record({:reset, cache}, :ok) + + @impl Logflare.ContextCache.Ops + def bust_by(cache, kw), do: record({:bust_by, cache, kw}, {:ok, 1}) + @impl Logflare.ContextCache.Ops + def fetch(cache, key, _getter), do: record({:fetch, cache, key}, :fetched) + @impl Logflare.ContextCache.Ops + def update(cache, key, value), do: record({:update, cache, key, value}, :ok) + + defp record(call, result) do + send(self(), call) + result + end + end + + defmodule CustomImpl.Cache do + use Logflare.ContextCache, impl: RecordingOps + end + describe "ContextCache" do setup do insert(:plan, name: "Free") @@ -86,6 +114,28 @@ defmodule Logflare.ContextCacheTest do end end + describe "context cache with a custom impl module" do + test "apply_fun/3" do + assert :fetched = ContextCache.apply_fun(CustomImpl, :get, [1]) + assert_received {:fetch, CustomImpl.Cache, {:get, [1]}} + end + + test "update/4" do + assert :ok = ContextCache.update(CustomImpl, :get, [1], :value) + assert_received {:update, CustomImpl.Cache, {:get, [1]}, :value} + end + + for {name, bust, expected_kw} <- [ + {"primary key", 5, [id: 5]}, + {"keyword", [source_id: 5], [source_id: 5]} + ] do + test "bust_keys/1 with a #{name}" do + assert {:ok, 1} = ContextCache.bust_keys([{CustomImpl, unquote(bust)}]) + assert_received {:bust_by, CustomImpl.Cache, unquote(expected_kw)} + end + end + end + describe "unboxed transaction" do setup do on_exit(fn -> diff --git a/test/profiling/context_cache_bust_by_kw_bench.exs b/test/profiling/context_cache_bust_by_kw_bench.exs index f635b2c702..5ed44494fb 100644 --- a/test/profiling/context_cache_bust_by_kw_bench.exs +++ b/test/profiling/context_cache_bust_by_kw_bench.exs @@ -1,3 +1,4 @@ +alias Logflare.Cache.CachexOps alias Logflare.ContextCache alias Logflare.Rules @@ -20,7 +21,7 @@ Benchee.run( %{ # Rules.Cache overrides bust_by/1, so call the default match-spec scan directly "bust_keys by primary key" => fn [{_source, rule} | _] -> - ContextCache.bust_by(cache, id: rule.id) + CachexOps.bust_by(cache, id: rule.id) end, "bust_keys by relation key" => fn [{source, _rule} | _] -> ContextCache.bust_keys([{Rules, source_id: source.id}]) From bc0aaf96d62d59fce63e3c763f1f7787744d4b50 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bartosz=20B=C5=82aszk=C3=B3w?= Date: Thu, 1 Oct 2026 14:06:29 +0200 Subject: [PATCH 4/6] test: drop cache tests covered by the generic Cache and ContextCache tests --- test/logflare/cache/cachex_ops_test.exs | 35 ++++++++--- test/logflare/context_cache_test.exs | 81 +------------------------ 2 files changed, 30 insertions(+), 86 deletions(-) diff --git a/test/logflare/cache/cachex_ops_test.exs b/test/logflare/cache/cachex_ops_test.exs index d013ee84fb..ad56456a26 100644 --- a/test/logflare/cache/cachex_ops_test.exs +++ b/test/logflare/cache/cachex_ops_test.exs @@ -8,13 +8,6 @@ defmodule Logflare.Cache.CachexOpsTest do @cache __MODULE__.Cache describe "child_spec/2" do - test "started under a supervisor" do - start_cache(limit: 10) - - assert {:ok, true} = Cachex.put(@cache, :key, :value) - assert {:ok, :value} = Cachex.get(@cache, :key) - end - for {stats_enabled?, expected} <- [{true, :ok}, {false, :error}] do test "cache stats config set to #{stats_enabled?}" do original = Application.get_env(:logflare, :cache_stats) @@ -102,6 +95,34 @@ defmodule Logflare.Cache.CachexOpsTest do end end + describe "bust_by/2" do + setup do + start_cache(limit: nil) + :ok + end + + for {name, value, expected_count} <- [ + {"map with the id", %{id: 1}, 1}, + {":ok tuple with a map with the id", {:ok, %{id: 1}}, 1}, + {":ok 3-tuple with a map with the id", {:ok, %{id: 1}, :extra}, 1}, + {"list containing a map with the id", [%{id: 2}, %{id: 1}], 1}, + {"map with another id", %{id: 2}, 0}, + {"list without the id", [%{id: 2}], 0}, + {"nil", nil, 0} + ] do + test "cached #{name}" do + Cachex.put(@cache, :key, {:cached, unquote(Macro.escape(value))}) + + assert {:ok, unquote(expected_count)} = CachexOps.bust_by(@cache, id: 1) + assert {:ok, unquote(expected_count == 0)} = Cachex.exists?(@cache, :key) + end + end + + test "keyword other than id" do + assert_raise ArgumentError, fn -> CachexOps.bust_by(@cache, source_id: 1) end + end + end + describe "fetch/3" do setup do start_cache(limit: nil) diff --git a/test/logflare/context_cache_test.exs b/test/logflare/context_cache_test.exs index 912ceb8c30..7d519acdc8 100644 --- a/test/logflare/context_cache_test.exs +++ b/test/logflare/context_cache_test.exs @@ -4,16 +4,6 @@ defmodule Logflare.ContextCacheTest do alias Ecto.Adapters.SQL alias Logflare.ContextCache alias Logflare.ContextCache.TransactionBroadcaster - alias Logflare.Sources - alias Logflare.Sources.Source - alias Logflare.Backends - alias Logflare.Backends.Backend - alias Logflare.Auth - alias Logflare.Cache.CachexOps - - defmodule TestCache do - use Logflare.ContextCache - end defmodule RecordingOps do @behaviour Logflare.Cache.Ops @@ -43,75 +33,8 @@ defmodule Logflare.ContextCacheTest do use Logflare.ContextCache, impl: RecordingOps end - describe "ContextCache" do - setup do - insert(:plan, name: "Free") - user = insert(:user) - source = insert(:source, user: user) - %{source: source, user: user} - end - - test "bust_keys/1, does nothing for empty list" do - assert {:ok, 0} = ContextCache.bust_keys([]) - end - - test "apply_fun/3, bust_keys/1 by :id field of value", %{source: source} do - Sources.Cache.get_by(token: source.token) - cache_key = {:get_by, [[token: source.token]]} - assert {:cached, %Source{}} = Cachex.get!(Sources.Cache, cache_key) - - assert {:ok, 1} = ContextCache.bust_keys([{Sources, source.id}]) - assert is_nil(Cachex.get!(Sources.Cache, cache_key)) - end - - test "apply_fun/3, bust_keys/1 by :id field of value for :ok tuple", %{user: user} do - {:ok, key} = Auth.create_access_token(user) - assert {:ok, _token, _user} = Auth.Cache.verify_access_token(key.token) - cache_key = {:verify_access_token, [key.token]} - assert {:cached, {:ok, %_{}, _user}} = Cachex.get!(Auth.Cache, cache_key) - - assert {:ok, 1} = ContextCache.bust_keys([{Auth, key.id}]) - assert is_nil(Cachex.get!(Auth.Cache, cache_key)) - end - - test "apply_fun/3, bust_keys/1 if primary key is in list of returned structs", %{ - source: source - } do - backend = insert(:backend, sources: [source]) - Backends.Cache.list_backends(source_id: source.id) - cache_key = {:list_backends, [[source_id: source.id]]} - assert {:cached, [%Backend{}]} = Cachex.get!(Backends.Cache, cache_key) - - assert {:ok, 1} = ContextCache.bust_keys([{Backends, backend.id}]) - assert is_nil(Cachex.get!(Backends.Cache, cache_key)) - end - end - - describe "default bust_by/1" do - setup do - start_supervised!(CachexOps.child_spec(TestCache, limit: nil)) - :ok - end - - for {name, value, expected_count} <- [ - {"map with the id", %{id: 1}, 1}, - {":ok tuple with a map with the id", {:ok, %{id: 1}}, 1}, - {"list containing a map with the id", [%{id: 2}, %{id: 1}], 1}, - {"map with another id", %{id: 2}, 0}, - {"list without the id", [%{id: 2}], 0}, - {"nil", nil, 0} - ] do - test "cached #{name}" do - Cachex.put(TestCache, :key, {:cached, unquote(Macro.escape(value))}) - - assert {:ok, unquote(expected_count)} = TestCache.bust_by(id: 1) - assert {:ok, unquote(expected_count == 0)} = Cachex.exists?(TestCache, :key) - end - end - - test "keyword other than id" do - assert_raise ArgumentError, fn -> TestCache.bust_by(source_id: 1) end - end + test "bust_keys/1 with an empty list" do + assert {:ok, 0} = ContextCache.bust_keys([]) end describe "context cache with a custom impl module" do From cbdf877220f1764111fc4557baa09981fff99d33 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bartosz=20B=C5=82aszk=C3=B3w?= Date: Mon, 29 Jun 2026 12:32:47 +0200 Subject: [PATCH 5/6] WIP: Introduce Nebulex TODO: Address the performance impact on busting --- lib/logflare/key_values/cache.ex | 132 ++++++++++++------ lib/logflare/key_values/cache/l1.ex | 14 ++ lib/logflare/key_values/cache/multilevel.ex | 14 ++ lib/logflare/key_values/cache_warmer.ex | 6 +- mix.exs | 3 + mix.lock | 6 + test/logflare/key_values/cache_test.exs | 7 +- .../logflare/key_values/cache_warmer_test.exs | 13 +- 8 files changed, 140 insertions(+), 55 deletions(-) create mode 100644 lib/logflare/key_values/cache/l1.ex create mode 100644 lib/logflare/key_values/cache/multilevel.ex diff --git a/lib/logflare/key_values/cache.ex b/lib/logflare/key_values/cache.ex index 7f2bc6bf3f..c29771eb52 100644 --- a/lib/logflare/key_values/cache.ex +++ b/lib/logflare/key_values/cache.ex @@ -1,34 +1,34 @@ defmodule Logflare.KeyValues.Cache do - @moduledoc false + @moduledoc """ + Read-through cache for the `Logflare.KeyValues` context. + + Caller-facing facade over a Nebulex multi-level cache + (`Logflare.KeyValues.Cache.Multilevel`). Today the only level is a local, + Cachex-backed cache (`Logflare.KeyValues.Cache.L1`); a distributed level can + be added later (O11Y-1504) without touching callers. + """ use Logflare.ContextCache alias Logflare.Cache.CachexOps alias Logflare.ContextCache alias Logflare.KeyValues + alias Logflare.KeyValues.Cache.L1 + alias Logflare.KeyValues.Cache.Multilevel alias Logflare.Repo def child_spec(_) do - CachexOps.child_spec(__MODULE__, - limit: 10_000_000, - ttl: to_timeout(day: 1), - purge_interval: to_timeout(hour: 1), - compressed: true, - warmer: {KeyValues.CacheWarmer, interval: :timer.hours(1)} - ) + %{ + id: __MODULE__, + start: {Multilevel, :start_link, [multilevel_opts()]} + } end @spec count(integer()) :: non_neg_integer() def count(user_id) do - cache_key = {:count, user_id} - - Cachex.fetch(__MODULE__, cache_key, fn _key -> - {:commit, {:cached, Repo.apply_with_replica(KeyValues, :count_key_values, [user_id])}} + fetch_or_store({:count, user_id}, fn -> + Repo.apply_with_replica(KeyValues, :count_key_values, [user_id]) end) - |> case do - {:commit, {:cached, v}} -> v - {:ok, {:cached, v}} -> v - end end @spec lookup(integer(), String.t()) :: map() | nil @@ -39,52 +39,98 @@ defmodule Logflare.KeyValues.Cache do @spec lookup(integer(), String.t(), String.t() | nil) :: term() | nil def lookup(user_id, key, accessor_path) do key_args = [user_id, key, accessor_path] - cache_key = {:lookup, key_args} + fetch_or_store({:lookup, key_args}, fn -> lookup_fallback(key_args) end) + end - Cachex.execute!(__MODULE__, fn worker -> - Cachex.fetch(worker, cache_key, fn _key -> lookup_fallback(worker, key_args) end) - end) - |> case do - {:commit, {:cached, v}} -> v - {:ok, {:cached, v}} -> v - end + @doc """ + Writes entries through the cache. Used by `Logflare.KeyValues.CacheWarmer`. + """ + @spec put_all([{term(), term()}]) :: :ok | {:error, term()} + def put_all([]), do: :ok + def put_all(entries), do: Multilevel.put_all(entries) + + # Only L1 gates health: a remote level outage should degrade to L1 plus the DB, not fail every node. + @impl Logflare.Cache + def healthy? do + CachexOps.healthy?(L1.cache_name()) + rescue + Nebulex.CacheNotFoundError -> false end - defp lookup_fallback(worker, [user_id, key, accessor_path] = key_args) do - case Cachex.get(worker, {:lookup, [user_id, key, nil]}) do - {:ok, {:cached, full_value}} -> - {:commit, {:cached, KeyValues.extract_value(full_value, accessor_path)}} + @impl Logflare.Cache + def stats, do: CachexOps.stats(L1.cache_name()) + + @impl Logflare.Cache + def reset do + {:ok, _count} = Multilevel.delete_all() + CachexOps.reset(L1.cache_name()) + end - {:ok, nil} -> - {:commit, {:cached, Repo.apply_with_replica(KeyValues, :lookup, key_args)}} + defp lookup_fallback([user_id, key, accessor_path] = key_args) do + case Multilevel.fetch({:lookup, [user_id, key, nil]}) do + {:ok, full_value} -> KeyValues.extract_value(full_value, accessor_path) + {:error, %Nebulex.KeyError{}} -> Repo.apply_with_replica(KeyValues, :lookup, key_args) end end @impl ContextCache def bust_by(kw) do - CachexOps.delete_keys(__MODULE__, bust_entries(kw)) - end - - defp bust_entries(kw) do user_id = Keyword.get(kw, :user_id) key = Keyword.get(kw, :key) - entries = if user_id, do: [{:count, user_id}], else: [] + keys = bust_keys(user_id, key) + # delete by exact keys so the bust propagates to every cache level + Multilevel.delete_all(in: keys) + end + + defp fetch_or_store(cache_key, getter) do + case Multilevel.fetch(cache_key) do + {:ok, value} -> + value - if user_id && key do - lookup_keys = find_lookup_keys(user_id, key) - lookup_keys ++ entries - else - entries + {:error, %Nebulex.KeyError{}} -> + value = getter.() + Multilevel.put(cache_key, value) + value end end - defp find_lookup_keys(user_id, key) do - {:ok, keys} = Cachex.keys(__MODULE__) + defp bust_keys(nil, _key), do: [] + + defp bust_keys(user_id, nil), do: [{:count, user_id}] + + defp bust_keys(user_id, key) do + [{:count, user_id} | lookup_keys(user_id, key)] + end - Enum.filter(keys, fn + # Gathers `{:lookup, [user_id, key, _accessor]}` keys for every cached accessor + # variant. Streams keys (not values) from the local level and filters them with + # a compiled match, which benchmarks faster than an interpreted ETS guard. + defp lookup_keys(user_id, key) do + [select: :key] + |> L1.stream!() + |> Stream.filter(fn {:lookup, [^user_id, ^key | _]} -> true _ -> false end) + |> Enum.to_list() + end + + defp multilevel_opts do + [ + inclusion_policy: :inclusive, + levels: [{L1, l1_opts()}] + ] + end + + defp l1_opts do + [stats: CachexOps.stats_enabled?()] ++ + CachexOps.cachex_opts( + limit: 10_000_000, + ttl: to_timeout(day: 1), + purge_interval: to_timeout(hour: 1), + compressed: true, + warmer: {KeyValues.CacheWarmer, interval: :timer.hours(1)} + ) end end diff --git a/lib/logflare/key_values/cache/l1.ex b/lib/logflare/key_values/cache/l1.ex new file mode 100644 index 0000000000..e954fe0424 --- /dev/null +++ b/lib/logflare/key_values/cache/l1.ex @@ -0,0 +1,14 @@ +defmodule Logflare.KeyValues.Cache.L1 do + @moduledoc """ + Local (L1) level of the `Logflare.KeyValues.Cache` multi-level cache. + + Backed by Cachex through `Nebulex.Adapters.Cachex`, preserving the original + Cachex configuration (compression, warmers, size limit, stats, expiration). + Start options are provided by `Logflare.KeyValues.Cache` when the multi-level + cache boots its levels. + """ + + use Nebulex.Cache, + otp_app: :logflare, + adapter: Nebulex.Adapters.Cachex +end diff --git a/lib/logflare/key_values/cache/multilevel.ex b/lib/logflare/key_values/cache/multilevel.ex new file mode 100644 index 0000000000..0b9e5bb26b --- /dev/null +++ b/lib/logflare/key_values/cache/multilevel.ex @@ -0,0 +1,14 @@ +defmodule Logflare.KeyValues.Cache.Multilevel do + @moduledoc """ + Nebulex multi-level cache backing `Logflare.KeyValues.Cache`. + + Currently configured with a single local level (`Logflare.KeyValues.Cache.L1`, + Cachex-backed). It provides the seam for adding a distributed L2 level later + (see O11Y-1504) without changing callers, which always go through the + `Logflare.KeyValues.Cache` facade. + """ + + use Nebulex.Cache, + otp_app: :logflare, + adapter: Nebulex.Adapters.Multilevel +end diff --git a/lib/logflare/key_values/cache_warmer.ex b/lib/logflare/key_values/cache_warmer.ex index d82162376d..c3b0a41a99 100644 --- a/lib/logflare/key_values/cache_warmer.ex +++ b/lib/logflare/key_values/cache_warmer.ex @@ -36,7 +36,7 @@ defmodule Logflare.KeyValues.CacheWarmer do |> Stream.chunk_every(500) |> Enum.each(fn chunk -> entries = Enum.map(chunk, &to_cache_entry/1) - Cachex.put_many(Cache, entries) + Cache.put_all(entries) end) end) end @@ -48,11 +48,11 @@ defmodule Logflare.KeyValues.CacheWarmer do |> Repo.all() |> Enum.map(&to_cache_entry/1) - if entries != [], do: Cachex.put_many(Cache, entries) + Cache.put_all(entries) end defp to_cache_entry(%KeyValue{} = kv) do - {{:lookup, [kv.user_id, kv.key, nil]}, {:cached, kv.value}} + {{:lookup, [kv.user_id, kv.key, nil]}, kv.value} end defp initialized? do diff --git a/mix.exs b/mix.exs index c5b92a719b..737f1c21be 100644 --- a/mix.exs +++ b/mix.exs @@ -122,6 +122,9 @@ defmodule Logflare.Mixfile do {:typed_struct, "~> 0.1", runtime: false}, {:lqueue, "~> 1.1"}, {:cachex, "~> 4.0"}, + {:nebulex, "~> 3.0"}, + {:nebulex_distributed, "~> 3.0"}, + {:nebulex_adapters_cachex, "~> 3.0"}, {:ex_machina, "~> 2.3", only: [:dev, :test]}, {:iteraptor, "~> 1.15"}, {:decorator, "~> 1.3"}, diff --git a/mix.lock b/mix.lock index b270272dd0..3143c688da 100644 --- a/mix.lock +++ b/mix.lock @@ -108,6 +108,11 @@ "mix_audit": {:hex, :mix_audit, "2.1.5", "c0f77cee6b4ef9d97e37772359a187a166c7a1e0e08b50edf5bf6959dfe5a016", [:make, :mix], [{:jason, "~> 1.4", [hex: :jason, repo: "hexpm", optional: false]}, {:yaml_elixir, "~> 2.11", [hex: :yaml_elixir, repo: "hexpm", optional: false]}], "hexpm", "87f9298e21da32f697af535475860dc1d3617a010e0b418d2ec6142bc8b42d69"}, "mix_test_watch": {:hex, :mix_test_watch, "1.4.0", "d88bcc4fbe3198871266e9d2f00cd8ae350938efbb11d3fa1da091586345adbb", [:mix], [{:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}], "hexpm", "2b4693e17c8ead2ef56d4f48a0329891e8c2d0d73752c0f09272a2b17dc38d1b"}, "money": {:hex, :money, "1.14.0", "61c1e9d9ae1dd45dae7f72568987b3e7275031c3f5a0bf8a053bd74259555934", [:mix], [{:decimal, "~> 1.2 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}, {:ecto, "~> 2.1 or ~> 3.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:phoenix_html, "~> 2.0 or ~> 3.0 or ~> 4.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}], "hexpm", "b8691009e0c31715d2e5a3cca68ca2e1a46895d63c11257b317d8801ee2c54e3"}, + "nebulex": {:hex, :nebulex, "3.0.4", "b55f73e5cd455a57a4dd25df9099806b9bae2384653a76a2fa2fa3184b3abb57", [:mix], [{:decorator, "~> 1.4", [hex: :decorator, repo: "hexpm", optional: true]}, {:nimble_options, "~> 0.5 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm", "446afc6d3f701ba991f1fb0eee36c600f888530e52f30f80b308480ad65faab6"}, + "nebulex_adapters_cachex": {:hex, :nebulex_adapters_cachex, "3.0.0", "8b02b1dc171fb33cfa7527aa7dff4fde7a4484decc04ac143e28ae141713b30f", [:mix], [{:cachex, "~> 4.0", [hex: :cachex, repo: "hexpm", optional: false]}, {:nebulex, "~> 3.0", [hex: :nebulex, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm", "2f0d7c2e714e6be6ac2e74a501bc29cf704c6074c4140416037119311a9707cd"}, + "nebulex_distributed": {:hex, :nebulex_distributed, "3.2.3", "d45e28ee7d88d5a49dd4ee71aa7c2d0be3337b53875a78245bcb021bd46d53e1", [:mix], [{:ex_hash_ring, "~> 6.0 or ~> 7.0", [hex: :ex_hash_ring, repo: "hexpm", optional: false]}, {:nebulex, "~> 3.0", [hex: :nebulex, repo: "hexpm", optional: false]}, {:nebulex_local, "~> 3.0", [hex: :nebulex_local, repo: "hexpm", optional: false]}, {:nebulex_streams, "~> 0.2", [hex: :nebulex_streams, repo: "hexpm", optional: false]}, {:partitioned_buffer, "~> 0.4", [hex: :partitioned_buffer, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm", "67a353fb5d67506767667d816d9f8a46a813549ba74e74bb0a1a0d83a078abff"}, + "nebulex_local": {:hex, :nebulex_local, "3.0.0", "1aead00a59d2043f419440b4c55851d3585419ef8918c1f47148d8ae180e8179", [:mix], [{:ex2ms, "~> 1.7", [hex: :ex2ms, repo: "hexpm", optional: true]}, {:nebulex, "~> 3.0", [hex: :nebulex, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.5 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:shards, "~> 1.1", [hex: :shards, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm", "7a087d9f071369ba272cd688c2bc4b758926ab3a2e239bce1b529653a14bdad1"}, + "nebulex_streams": {:hex, :nebulex_streams, "0.2.0", "f8f6f073a19cccf244b4e8edbf0516f3649ddad38a24b6443d3a3b1b70416cf5", [:mix], [{:nebulex, "~> 3.0.1", [hex: :nebulex, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.5 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:phoenix_pubsub, "~> 2.1", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "4cd38507756d16fa314cadf4452ea6c501fa49a77bf9bf2c46719b84c83d8d20"}, "nimble_options": {:hex, :nimble_options, "1.1.1", "e3a492d54d85fc3fd7c5baf411d9d2852922f66e69476317787a7b2bb000a61b", [:mix], [], "hexpm", "821b2470ca9442c4b6984882fe9bb0389371b8ddec4d45a9504f00a66f650b44"}, "nimble_parsec": {:hex, :nimble_parsec, "1.4.2", "8efba0122db06df95bfaa78f791344a89352ba04baedd3849593bfce4d0dc1c6", [:mix], [], "hexpm", "4b21398942dda052b403bbe1da991ccd03a053668d147d53fb8c4e0efe09c973"}, "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"}, @@ -129,6 +134,7 @@ "otel_metric_exporter": {:git, "https://github.com/supabase/elixir-otel-metric-exporter", "2a6de9161c15e4236bf645287450e6c1beed1ad7", [ref: "2a6de91"]}, "paper_trail": {:hex, :paper_trail, "1.1.2", "a2f3a27a1165c424dbc418448d858baff6b3fa15985c2bec01731f8f4bbd08fa", [:mix], [{:ecto, ">= 3.12.2", [hex: :ecto, repo: "hexpm", optional: false]}, {:ecto_sql, ">= 3.12.0", [hex: :ecto_sql, repo: "hexpm", optional: false]}], "hexpm", "6f540fdc83e37eadb1e5e0e08f6d97d223daf9ecfe75448ebe7e13e4a5b65ace"}, "parse_trans": {:hex, :parse_trans, "3.4.1", "6e6aa8167cb44cc8f39441d05193be6e6f4e7c2946cb2759f015f8c56b76e5ff", [:rebar3], [], "hexpm", "620a406ce75dada827b82e453c19cf06776be266f5a67cff34e1ef2cbb60e49a"}, + "partitioned_buffer": {:hex, :partitioned_buffer, "0.4.3", "5afd9f9748e9dda78824f355689cfd781060df29e05acfcea8e809ebdf5fca9d", [:mix], [{:nimble_options, "~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "79d33f27d859b00f2e4e3a01261f8af01a0e8b96ea04d1ae76c971f3a920cc31"}, "pgoutput_decoder": {:hex, :pgoutput_decoder, "0.1.0", "d4ffae6e58a563f2e6de8a0495d9f9afbe2f4ac75d6805419cd4a0d05f414c00", [:mix], [], "hexpm", "4dbecbe4eb8de728178fd129ccba810bccafa9a8769c6714c8b7b22963081c27"}, "phoenix": {:hex, :phoenix, "1.8.9", "a63ed0962ed5b903b146dab0ae8eb8387fe478f8171a5e26d56a165f35996fe1", [:mix], [{:bandit, "~> 1.0", [hex: :bandit, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.1", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:phoenix_template, "~> 1.0", [hex: :phoenix_template, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 2.0", [hex: :phoenix_view, repo: "hexpm", optional: true]}, {:plug, "~> 1.14", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.7", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.2 or ~> 2.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:websock_adapter, "~> 0.5", [hex: :websock_adapter, repo: "hexpm", optional: false]}], "hexpm", "3477e2dd5a4f61820341169031bdfe21275f659923bea9c5c0ea2aa1c3fcc046"}, "phoenix_ecto": {:hex, :phoenix_ecto, "4.7.0", "75c4b9dfb3efdc42aec2bd5f8bccd978aca0651dbcbc7a3f362ea5d9d43153c6", [:mix], [{:ecto, "~> 3.5", [hex: :ecto, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.14.2 or ~> 3.0 or ~> 4.1", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:plug, "~> 1.9", [hex: :plug, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.16 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}], "hexpm", "1d75011e4254cb4ddf823e81823a9629559a1be93b4321a6a5f11a5306fbf4cc"}, diff --git a/test/logflare/key_values/cache_test.exs b/test/logflare/key_values/cache_test.exs index 0e67d2ad40..d8c55bcef2 100644 --- a/test/logflare/key_values/cache_test.exs +++ b/test/logflare/key_values/cache_test.exs @@ -3,6 +3,7 @@ defmodule Logflare.KeyValues.CacheTest do use Logflare.DataCase, async: false alias Logflare.KeyValues + alias Logflare.KeyValues.Cache.Multilevel setup do user = insert(:user) @@ -16,7 +17,7 @@ defmodule Logflare.KeyValues.CacheTest do # first call populates cache assert ^value = KeyValues.Cache.lookup(user.id, "proj1") cache_key = {:lookup, [user.id, "proj1", nil]} - assert {:cached, ^value} = Cachex.get!(KeyValues.Cache, cache_key) + assert {:ok, ^value} = Multilevel.fetch(cache_key) end test "lookup/3 caches extracted value with accessor path", %{user: user} do @@ -25,7 +26,7 @@ defmodule Logflare.KeyValues.CacheTest do assert "abc" = KeyValues.Cache.lookup(user.id, "proj1", "org.id") cache_key = {:lookup, [user.id, "proj1", "org.id"]} - assert {:cached, "abc"} = Cachex.get!(KeyValues.Cache, cache_key) + assert {:ok, "abc"} = Multilevel.fetch(cache_key) end test "lookup/3 on a warmed entry", %{user: user} do @@ -86,7 +87,7 @@ defmodule Logflare.KeyValues.CacheTest do KeyValues.Cache.bust_by(user_id: user.id, key: "k1") cache_key = {:count, user.id} - assert is_nil(Cachex.get!(KeyValues.Cache, cache_key)) + assert {:error, %Nebulex.KeyError{}} = Multilevel.fetch(cache_key) end end end diff --git a/test/logflare/key_values/cache_warmer_test.exs b/test/logflare/key_values/cache_warmer_test.exs index a2def0d5ac..93226523a7 100644 --- a/test/logflare/key_values/cache_warmer_test.exs +++ b/test/logflare/key_values/cache_warmer_test.exs @@ -4,7 +4,7 @@ defmodule Logflare.KeyValues.CacheWarmerTest do import ExUnit.CaptureLog - alias Logflare.KeyValues.Cache + alias Logflare.KeyValues.Cache.Multilevel alias Logflare.KeyValues.CacheWarmer @pt_key {CacheWarmer, :initialized} @@ -22,8 +22,8 @@ defmodule Logflare.KeyValues.CacheWarmerTest do CacheWarmer.execute(nil) - assert {:cached, kv1.value} == Cachex.get!(Cache, {:lookup, [user.id, "k1", nil]}) - assert {:cached, kv2.value} == Cachex.get!(Cache, {:lookup, [user.id, "k2", nil]}) + assert {:ok, kv1.value} == Multilevel.fetch({:lookup, [user.id, "k1", nil]}) + assert {:ok, kv2.value} == Multilevel.fetch({:lookup, [user.id, "k2", nil]}) end test "marks itself as initialized after first run" do @@ -78,10 +78,11 @@ defmodule Logflare.KeyValues.CacheWarmerTest do CacheWarmer.execute(nil) - assert {:cached, %{"v" => "new"}} == - Cachex.get!(Cache, {:lookup, [user.id, "new_key", nil]}) + assert {:ok, %{"v" => "new"}} == + Multilevel.fetch({:lookup, [user.id, "new_key", nil]}) - assert is_nil(Cachex.get!(Cache, {:lookup, [user.id, "old_key", nil]})) + assert {:error, %Nebulex.KeyError{}} = + Multilevel.fetch({:lookup, [user.id, "old_key", nil]}) end test "no-ops when no recent records exist" do From 39a2e6b7f7c8d91fab5096d0cedd2ed9f6a2cf6f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bartosz=20B=C5=82aszk=C3=B3w?= Date: Thu, 17 Sep 2026 12:34:01 +0200 Subject: [PATCH 6/6] perf: Fix bust_by regression under Nebulex --- lib/logflare/key_values/cache.ex | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/lib/logflare/key_values/cache.ex b/lib/logflare/key_values/cache.ex index c29771eb52..9343349086 100644 --- a/lib/logflare/key_values/cache.ex +++ b/lib/logflare/key_values/cache.ex @@ -78,9 +78,20 @@ defmodule Logflare.KeyValues.Cache do user_id = Keyword.get(kw, :user_id) key = Keyword.get(kw, :key) - keys = bust_keys(user_id, key) - # delete by exact keys so the bust propagates to every cache level - Multilevel.delete_all(in: keys) + busted = + user_id + |> bust_keys(key) + |> Enum.count(&delete_key/1) + + {:ok, busted} + end + + # Workaround for poor `delete_all(in: x)` performance + # until https://github.com/elixir-nebulex/nebulex_local/issues/8 is released + defp delete_key(cache_key) do + {:ok, present?} = Multilevel.has_key?(cache_key) + :ok = Multilevel.delete(cache_key) + present? end defp fetch_or_store(cache_key, getter) do