feat: SIGNAL-5811 add time interval bucket feature to ThrottleAndTimed and make loop_interval optional

[seungjinstord]

Change Buffy.ThrottleAndTimed to be able to modify data in state for each throttle() invoked.
This is additive with additional defoverridable functions - original API is intact.
:loop_interval option has looser requirement as it became optional.

Example use case: let's say for some event handler operation, we process an id by querying into the DB.

If that happens a thousand times per second into a connection that takes a lot of CPU/memory, like Postgres DB,
then making a thousand query connections per second would be significantly expensive compared to one connection with a list of thousand ids - let's say the same query ends up returning a list, regardless of one id or list of ids (logic only differs by input size of ids list).

In that case we need a way to collect ids across some set of events.

Buffy.ThrottleAndTimed already has bulk of the timing mechanism in place:

• it is a process that will not get killed once throttle work is done
• it has throttling logic
• it can be modified to have three additional defoverridable functions for key generating, args updating, and state update upon successful work operations.
• it can be modified so the loop interval feature is optional.

Example use (also noted in moduledoc):

    defmodule MyElasticEventThrottler do
      use Buffy.ThrottleAndTimed,
        throttle: 100,
        supervisor_module: DynamicSupervisor,
        supervisor_name: MyDynamicSupervisor

      def handle_throttle(%{test_pid: test_pid, values: values} = args) do
        Process.sleep(200)
        send(test_pid, {:ok, args, System.monotonic_time()})
        values
      end

      def args_to_key(%{key: key}), do: key |> :erlang.term_to_binary() |> :erlang.phash2()

      def update_args(%{values: values} = old_arg, %{values: new_values} = _new_arg)
          when is_list(values) and is_list(new_values) do
        %{old_arg | values: Enum.sort(values ++ new_values)}
      end

      def update_state_with_work_result(%{args: %{values: state_values} = args} = state, result_values) do
        # because `handle_throttle()` runs in the `:continue` lifecycle of GenServer,
        # inbox processing is paused until the logic completes. Inbox will continually get new messages,
        # from calling `throttle()` and will be processed only after completion of `handle_throttle()`.
        %{state | args: %{args | values: []}}
      end
    end

# ... and using it:
MyElasticEventThrottler.throttle(%{key: "my_key", test_pid: test_pid, values: [0,1,2,3]})

Also see the added test for a more thorough use case.

[kinson]

A few suggestions and one question about avoiding a potential race condition

[kinson]

Something I'm thinking about is whether there might be issues with using handle_cast in a situation where were continuously getting throttle messages and the timeout occurs in the middle of a burst of them. Is there a chance we could drop some of the new things being added to state or overwrite them when we call update_state_with_work_result/2.

I'd have to think about it and maybe test it out to know for sure if that's possible (or not possible), but maybe you've already considered this 🙂 and done they heavy mental lifting for me.

[seungjinstord]

I was trying to test it and was wondering why I wasn't getting any new data into the state.

Then I realized that the beauty of handle_continue is that incoming messages will be kept getting stored in the inbox until handle_continue finishes running.

After the handle_continue is finished, the GenServer will then start ingesting the backed up messages in the inbox - this was confirmed by checking the Process.info(pid, :message).

[kinson]

Then I realized that the beauty of handle_continue is that incoming messages will be kept getting stored in the inbox until handle_continue finishes running.

❤️

[kinson]

After the handle_continue is finished, the GenServer will then start ingesting the backed up messages in the inbox - this was confirmed by checking the Process.info(pid, :message).

Could we measure/monitor the size of the inbox queue? That could be valuable lest we run into an issue where messages get dropped. I'm trying to find documentation on the behavior as the mailbox size grows but having trouble for some reason 🙃

[kinson]

Thinking about it more, we could probably implement this in the handler itself during handle_throttle if we want to capture it

[kinson]

💥 I'll try plugging this into the wms-service and test it on a branch unless you already have one you're working on that I could use

[seungjinstord]

@kinson I'll set one up gimme a sec

UPDATE: here it is: https://github.com/stordco/wms-service/tree/SIGNAL-5811-with-buffy-pr - I tweaked mix.exs to use this branch

[kinson]

Thinking about it more, we could probably implement this in the handler itself during handle_throttle if we want to capture it

[kinson]

@seungjinstord this worked like a charm ✨