diff --git a/lib/workflow_metal/application.ex b/lib/workflow_metal/application.ex index d01a29d..21e9719 100644 --- a/lib/workflow_metal/application.ex +++ b/lib/workflow_metal/application.ex @@ -1,5 +1,20 @@ defmodule WorkflowMetal.Application do - @moduledoc false + @moduledoc """ + A workflow_metal application is a collection of workflows, cases, tasks and workitems. + + In `config/config.exs` set default_lifespan_timeout for any `Case`, `Task` and `Workitem`: + + config :my_app, MyApp.Application, + case: [ + lifespan_timeout: 30000 + ], + task: [ + lifespan_timeout: 60000 + ], + workitem: [ + lifespan_timeout: 300000 + ] + """ alias WorkflowMetal.Application.Config diff --git a/lib/workflow_metal/case/case.ex b/lib/workflow_metal/case/case.ex index 5aecfc4..a8c9f7d 100644 --- a/lib/workflow_metal/case/case.ex +++ b/lib/workflow_metal/case/case.ex @@ -233,7 +233,7 @@ defmodule WorkflowMetal.Case.Case do {:noreply, data, {:continue, :stop_server}} _error -> - {:noreply, data} + {:noreply, data, get_timeout(data.application)} end end @@ -241,7 +241,7 @@ defmodule WorkflowMetal.Case.Case do def handle_continue({:revoke_tokens, locked_token_schemas, except_task_id}, %__MODULE__{state: :active} = data) do data = do_revoke_tokens(locked_token_schemas, except_task_id, data) - {:noreply, data} + {:noreply, data, get_timeout(data.application)} end @impl GenServer @@ -257,10 +257,10 @@ defmodule WorkflowMetal.Case.Case do "#{describe(data)}: tokens(#{Enum.join(token_ids, ", ")}) have been locked by the task(#{task_id})" end) - {:reply, {:ok, locked_token_schemas}, data} + {:reply, {:ok, locked_token_schemas}, data, get_timeout(data.application)} {:error, _reason} = error -> - {:reply, error, data} + {:reply, error, data, get_timeout(data.application)} end end @@ -290,19 +290,19 @@ defmodule WorkflowMetal.Case.Case do def handle_call({:offer_tokens_to_task, task_id}, _from, %{state: :active} = data) do {tokens, data} = do_offer_tokens_to_task(task_id, data) - {:reply, {:ok, tokens}, data} + {:reply, {:ok, tokens}, data, get_timeout(data.application)} end @impl GenServer def handle_call({:fetch_locked_tokens_from_task, task_id}, _from, %{state: :active} = data) do tokens = do_fetch_locked_tokens_from_task(task_id, data) - {:reply, {:ok, tokens}, data} + {:reply, {:ok, tokens}, data, get_timeout(data.application)} end @impl GenServer def handle_call(_msg, _from, %{} = data) do - {:reply, {:error, :case_not_available}, data} + {:reply, {:error, :case_not_available}, data, get_timeout(data.application)} end @impl GenServer @@ -332,6 +332,13 @@ defmodule WorkflowMetal.Case.Case do {:noreply, data, {:continue, :stop_server}} end + @impl GenServer + def handle_info(:timeout, %__MODULE__{} = data) do + Logger.debug(describe(data) <> " stopping due to inactivity timeout") + + {:stop, :normal, data} + end + @spec fetch_unconsumed_tokens(t()) :: t() defp fetch_unconsumed_tokens(%__MODULE__{} = data) do %{ @@ -829,6 +836,13 @@ defmodule WorkflowMetal.Case.Case do {:ok, data} end + defp get_timeout(application) do + application + |> WorkflowMetal.Application.Config.get(:case) + |> Kernel.||([]) + |> Keyword.get(:lifespan_timeout, :timer.minutes(5)) + end + defp describe(%__MODULE__{} = data) do %{ case_schema: %Schema.Case{ diff --git a/lib/workflow_metal/task/task.ex b/lib/workflow_metal/task/task.ex index 9456d09..6c10402 100644 --- a/lib/workflow_metal/task/task.ex +++ b/lib/workflow_metal/task/task.ex @@ -198,7 +198,7 @@ defmodule WorkflowMetal.Task.Task do |> request_tokens() |> start_created_workitems() - {:noreply, data} + {:noreply, data, get_timeout(data.application)} end def handle_continue(:after_start, %{state: :executing} = data) do @@ -221,10 +221,10 @@ defmodule WorkflowMetal.Task.Task do |> allocate_workitem() |> update_task(%{state: :allocated}) - {:noreply, data} + {:noreply, data, get_timeout(data.application)} _other -> - {:noreply, data} + {:noreply, data, get_timeout(data.application)} end end @@ -240,7 +240,7 @@ defmodule WorkflowMetal.Task.Task do {:noreply, data, {:continue, :stop_server}} _error -> - {:noreply, data} + {:noreply, data, get_timeout(data.application)} end end @@ -254,7 +254,7 @@ defmodule WorkflowMetal.Task.Task do {:noreply, data, {:continue, :stop_server}} _error -> - {:noreply, data} + {:noreply, data, get_timeout(data.application)} end end @@ -271,7 +271,7 @@ defmodule WorkflowMetal.Task.Task do {:noreply, data, {:continue, :stop_server}} _ -> - {:noreply, data} + {:noreply, data, get_timeout(data.application)} end end @@ -295,7 +295,7 @@ defmodule WorkflowMetal.Task.Task do def handle_cast({:receive_tokens, token_schemas}, %{state: :allocated} = data) do data = Enum.reduce(token_schemas, data, &upsert_ets_token/2) - {:noreply, data} + {:noreply, data, get_timeout(data.application)} end @impl GenServer @@ -325,7 +325,7 @@ defmodule WorkflowMetal.Task.Task do when state in [:allocated, :executing] do data = upsert_ets_workitem({workitem_id, :started}, data) - {:noreply, data} + {:noreply, data, get_timeout(data.application)} end @impl GenServer @@ -345,7 +345,7 @@ defmodule WorkflowMetal.Task.Task do @impl GenServer def handle_cast(_msg, %{} = data) do - {:noreply, data} + {:noreply, data, get_timeout(data.application)} end @impl GenServer @@ -357,7 +357,7 @@ defmodule WorkflowMetal.Task.Task do {:reply, {:ok, locked_token_schemas}, data} error -> - {:reply, error, data} + {:reply, error, data, get_timeout(data.application)} end end @@ -378,12 +378,19 @@ defmodule WorkflowMetal.Task.Task do acc end) - {:reply, {:ok, tokens}, data} + {:reply, {:ok, tokens}, data, get_timeout(data.application)} end @impl GenServer def handle_call(_msg, _from, %{} = data) do - {:reply, {:error, :task_not_available}, data} + {:reply, {:error, :task_not_available}, data, get_timeout(data.application)} + end + + @impl GenServer + def handle_info(:timeout, %__MODULE__{} = data) do + Logger.debug(describe(data) <> " stopping due to inactivity timeout") + + {:stop, :normal, data} end @spec fetch_workitems(t()) :: t() @@ -775,6 +782,13 @@ defmodule WorkflowMetal.Task.Task do data end + defp get_timeout(application) do + application + |> WorkflowMetal.Application.Config.get(:task) + |> Kernel.||([]) + |> Keyword.get(:lifespan_timeout, :timer.minutes(1)) + end + defp describe(%__MODULE__{} = data) do %{ task_schema: %Schema.Task{ diff --git a/lib/workflow_metal/workitem/workitem.ex b/lib/workflow_metal/workitem/workitem.ex index 84744ed..490f7cf 100644 --- a/lib/workflow_metal/workitem/workitem.ex +++ b/lib/workflow_metal/workitem/workitem.ex @@ -151,7 +151,8 @@ defmodule WorkflowMetal.Workitem.Workitem do application: application, state: :started, workitem_schema: workitem_schema - } + }, + get_timeout(application) } state when state in [:completed, :abandoned] -> @@ -167,7 +168,7 @@ defmodule WorkflowMetal.Workitem.Workitem do {:ok, :started, data} -> data = update_workitem(%{state: :started}, data) - {:noreply, data} + {:noreply, data, get_timeout(data.application)} {:ok, {:completed, workitem_output}, data} -> {:noreply, %{data | state: :started}, {:continue, {:complete, workitem_output}}} @@ -175,7 +176,7 @@ defmodule WorkflowMetal.Workitem.Workitem do {:ok, :abandoned, data} -> data = update_workitem(%{state: :abandoned}, data) - {:noreply, data} + {:noreply, data, get_timeout(data.application)} end end @@ -215,6 +216,13 @@ defmodule WorkflowMetal.Workitem.Workitem do {:noreply, data, {:continue, :stop_server}} end + @impl GenServer + def handle_info(:timeout, %__MODULE__{} = data) do + Logger.debug(describe(data) <> " stopping due to inactivity timeout") + + {:stop, :normal, data} + end + @spec update_workitem(map(), t()) :: t() defp update_workitem(params, %__MODULE__{} = data) do %{ @@ -330,6 +338,13 @@ defmodule WorkflowMetal.Workitem.Workitem do {:ok, data} end + defp get_timeout(application) do + application + |> WorkflowMetal.Application.Config.get(:case) + |> Kernel.||([]) + |> Keyword.get(:lifespan_timeout, :timer.seconds(30)) + end + defp describe(%__MODULE__{} = data) do %{ workitem_schema: %Schema.Workitem{