feat: v1 — operator controls, observability and dashboard redesign

.github/workflows/ci.yml

@@ -106,6 +106,22 @@ jobs:
     if: needs.changes.outputs.dashboard == 'true' || needs.changes.outputs.ci == 'true' || github.event_name == 'push'
     runs-on: ubuntu-latest
 
+    # Page tests (DurableDashboard.LiveCase) mount the LiveViews against a
+    # real Durable instance.
+    services:
+      postgres:
+        image: postgres:18-alpine
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_PASSWORD: postgres
+        ports:
+          - 54321:5432
+        options: >-
+          --health-cmd="pg_isready -U postgres"
+          --health-interval=10s
+          --health-timeout=5s
+          --health-retries=5
+
     defaults:
       run:
         working-directory: durable_dashboard

.gitignore

@@ -11,3 +11,6 @@ erl_crash.dump
 /doc/
 /_build/
 /deps/
+
+# macOS Finder metadata
+.DS_Store

docker-compose.yml

@@ -7,12 +7,15 @@ services:
     ports:
       - "54321:5432"
     volumes:
-      - postgres_data:/var/lib/postgresql/data
+      # Postgres 18+ images keep data under /var/lib/postgresql/<major>/, so the
+      # volume mounts one level up. A new volume name keeps older-major data
+      # (the old `postgres_data`) from blocking startup.
+      - postgres18_data:/var/lib/postgresql
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U postgres"]
       interval: 5s
       timeout: 5s
       retries: 5
 
 volumes:
-  postgres_data:
+  postgres18_data:

durable/README.md

@@ -483,6 +483,37 @@ parallel do
 end
 ```
 
+### Workflow Options
+
+```elixir
+workflow "process_order", timeout: hours(2), queue: :orders do
+```
+
+- `:timeout` — how long a run may take from when it's due to start, waits and
+  sleeps included. A run past it fails with a `"timeout"` error before its next
+  step and runs its compensations; a waiting run is woken to do so (within the
+  timeout worker's one-minute sweep). It can't interrupt a step that's running.
+- `:queue` — the queue runs start on unless `Durable.start/3` names one.
+
+Unknown options fail the compile.
+
+### Errors
+
+When a step fails, its execution keeps what went wrong as `error`:
+
+- **Raised** (`raise`, a crash in a library) — `type` (the exception
+  module), `message`, and `stacktrace`: one frame per line with the app,
+  file and line, e.g. `(my_app 0.1.0) lib/my_app/orders.ex:28: ...`.
+- **Exited** (a `GenServer.call` timeout, a linked process going down) —
+  `type: "exit"`, a readable `message` (`Exception.format_exit/1`) and the
+  `stacktrace`. Throws are kept the same way.
+- **Returned** — `{:error, reason}` from the step: `reason` itself (a map is
+  stored as-is), with no stack, because nothing was raised.
+
+The same applies to compensations and to a branch's `on:` function. A
+failed run's `error` is its failing step's. The stack is as deep as the
+VM's `:backtrace_depth` (8 frames by default; Phoenix dev raises it to 20).
+
 ### API
 
 ```elixir
@@ -494,8 +525,23 @@ Durable.cancel(id, "reason")
 Durable.send_event(id, "event", payload)
 Durable.provide_input(id, "input_name", data)
 Durable.list_children(parent_id)
+
+# Operating
+Durable.retry(id)                        # failed run: resume from the failed step
+Durable.rerun(id, input: new_input)      # finished run: start a fresh copy
+Durable.pause_queue(:default, by: user)  # on every node, until resume_queue/2
+Durable.list_nodes()                     # heartbeating instances and what they run
+Durable.prune()                          # apply the :prune retention now
+
+# Observing
+Durable.Stats.overview(since: dt)        # also timeline/1, workflows/1, step_performance/1,
+                                         # queues/1, queue_history/1, error_groups/1
+Durable.Activity.list()                  # recovered runs, fenced workers, queue pauses…
 ```
 
+Telemetry events and `Telemetry.Metrics` definitions are documented in
+`Durable.Telemetry`.
+
 ## Mix Tasks
 
 Durable includes mix tasks for managing workflows from the command line.

durable/lib/durable.ex

@@ -94,6 +94,8 @@ defmodule Durable do
 
   """
 
+  alias Durable.Config
+  alias Durable.Queue.Control, as: QueueControl
   alias Durable.Scheduler.API, as: SchedulerAPI
 
   defmacro __using__(_opts) do
@@ -245,7 +247,15 @@ defmodule Durable do
   end
 
   @doc """
-  Cancels a running or pending workflow.
+  Cancels a pending, running or waiting workflow, and its child workflows.
+
+  The cancel is final: once it returns `:ok` the run stays `:cancelled`. A
+  running workflow stops at its next step boundary: the step already in
+  flight finishes and is recorded, but its output isn't merged into the run's
+  context and no later step runs. Cancelling does not run compensations.
+
+  Returns `{:error, :already_completed}` if the run has already finished, and
+  `{:error, :not_found}` for an unknown id.
 
   ## Examples
 
@@ -258,6 +268,104 @@ defmodule Durable do
     Durable.Executor.cancel_workflow(workflow_id, reason)
   end
 
+  @doc """
+  Resumes a failed workflow at the step that failed.
+
+  Steps that already completed are not re-run, and the failed step gets its
+  full retry budget again. Inside a parallel block only the children that
+  failed are re-queued.
+
+  Returns `{:error, :rolled_back}` for runs whose compensations ran — use
+  `rerun/2` for those — and `{:error, :not_failed}` for runs that aren't failed.
+
+  ## Options
+
+  - `:durable` - The Durable instance name (default: Durable)
+
+  ## Examples
+
+      {:ok, ^workflow_id} = Durable.retry(workflow_id)
+
+  """
+  @spec retry(String.t(), keyword()) :: {:ok, String.t()} | {:error, term()}
+  def retry(workflow_id, opts \\ []) do
+    Durable.Executor.retry_workflow(workflow_id, opts)
+  end
+
+  @doc """
+  Starts a fresh run of a finished workflow, with its original input or a new
+  one. The new run's `trigger` is `:rerun` and its `trigger_ref` is the
+  original run's id.
+
+  ## Options
+
+  - `:input` - Input for the new run (default: the original run's input)
+  - `:durable` - The Durable instance name (default: Durable)
+
+  ## Examples
+
+      {:ok, new_id} = Durable.rerun(workflow_id)
+      {:ok, new_id} = Durable.rerun(workflow_id, input: %{"order_id" => "ORD-1043"})
+
+  """
+  @spec rerun(String.t(), keyword()) :: {:ok, String.t()} | {:error, term()}
+  def rerun(workflow_id, opts \\ []) do
+    Durable.Executor.rerun_workflow(workflow_id, opts)
+  end
+
+  @doc """
+  Pauses a queue on every node. Runs already executing finish; no node claims
+  new work from the queue until `resume_queue/2`. The pause survives restarts.
+
+  Recorded in `Durable.Activity` as `"queue_paused"`.
+
+  ## Options
+
+  - `:durable` - The Durable instance name (default: Durable)
+  - `:by` - who paused it, kept with the activity record
+  """
+  @spec pause_queue(atom() | String.t(), keyword()) :: :ok | {:error, term()}
+  def pause_queue(queue, opts \\ []) do
+    QueueControl.pause(Config.get(Keyword.get(opts, :durable, Durable)), queue, opts)
+  end
+
+  @doc """
+  Resumes a queue paused with `pause_queue/2`, on every node.
+  """
+  @spec resume_queue(atom() | String.t(), keyword()) :: :ok | {:error, term()}
+  def resume_queue(queue, opts \\ []) do
+    QueueControl.resume(Config.get(Keyword.get(opts, :durable, Durable)), queue, opts)
+  end
+
+  @doc """
+  Returns the names of queues currently paused with `pause_queue/2`.
+  """
+  @spec paused_queues(keyword()) :: [String.t()]
+  def paused_queues(opts \\ []) do
+    QueueControl.paused(Config.get(Keyword.get(opts, :durable, Durable)))
+  end
+
+  @doc """
+  Lists the Durable nodes that have heartbeated recently, newest first. Each
+  has `:id`, `:queues`, `:version`, `:started_at`, `:heartbeat_at`, a
+  `:status` of `:alive` or `:missing` (no heartbeat for 30 s), and `:running`,
+  the number of runs it is executing.
+  """
+  @spec list_nodes(keyword()) :: [map()]
+  def list_nodes(opts \\ []) do
+    Durable.NodeRegistry.list(Config.get(Keyword.get(opts, :durable, Durable)))
+  end
+
+  @doc """
+  Deletes finished runs past the `:prune` retention now, instead of waiting
+  for the hourly sweep. Returns the number deleted per group, or `%{}` when
+  `:prune` isn't configured.
+  """
+  @spec prune(keyword()) :: %{atom() => non_neg_integer()}
+  def prune(opts \\ []) do
+    Durable.Pruner.prune(Config.get(Keyword.get(opts, :durable, Durable)))
+  end
+
   @doc """
   Provides input for a waiting workflow (human-in-the-loop).
 

durable/lib/durable/activity.ex

@@ -0,0 +1,76 @@
+defmodule Durable.Activity do
+  @moduledoc """
+  Operational events worth an operator's attention, kept for a week in
+  `durable.activity` — the record behind "what happened to my queues?".
+
+  | kind                  | recorded when                                                  |
+  | --------------------- | -------------------------------------------------------------- |
+  | `"stale_recovered"`   | runs whose node stopped heartbeating went back on the queue    |
+  | `"zombie_recovered"`  | runs left `running` with no live worker were failed            |
+  | `"worker_fenced"`     | a worker found its run reclaimed by another node and stopped   |
+  | `"ack_failed"`        | a worker couldn't release a finished run's lock                |
+  | `"queue_paused"`      | `Durable.pause_queue/2` (`details.by` when given)               |
+  | `"queue_resumed"`     | `Durable.resume_queue/2`                                       |
+  | `"schedule_disabled"` | a schedule was turned off after failing repeatedly            |
+  | `"pruned"`            | the retention pruner deleted finished runs                     |
+
+  Each row has `:at`, `:kind`, the `:node_id` that recorded it, and where it
+  applies `:queue`, `:workflow_id`, a `:count` and `:details`.
+
+  Recording never raises: losing an activity row must not break the recovery
+  or the action it describes.
+  """
+
+  import Ecto.Query
+
+  require Logger
+
+  alias Durable.Config
+  alias Durable.Repo
+  alias Durable.Storage.Schemas.ActivityEvent
+
+  @doc false
+  @spec record(Config.t(), String.t(), keyword()) :: :ok
+  def record(%Config{} = config, kind, attrs \\ []) do
+    event = %ActivityEvent{
+      at: DateTime.utc_now(),
+      kind: kind,
+      node_id: config.node_id,
+      queue: attrs[:queue] && to_string(attrs[:queue]),
+      workflow_id: attrs[:workflow_id],
+      count: Keyword.get(attrs, :count, 1),
+      details: attrs |> Keyword.get(:details, %{}) |> Durable.Executor.sanitize_for_json()
+    }
+
+    Repo.insert(event, config)
+    :ok
+  rescue
+    e ->
+      Logger.debug("[Durable] activity #{kind} not recorded: #{Exception.message(e)}")
+      :ok
+  end
+
+  @doc """
+  Recent activity, newest first.
+
+  ## Options
+
+  - `:since` - only events at or after this time (default: 24 hours ago)
+  - `:limit` - at most this many (default: 100)
+  - `:durable` - the Durable instance (default: `Durable`)
+  """
+  @spec list(keyword()) :: [ActivityEvent.t()]
+  def list(opts \\ []) do
+    config = Config.get(Keyword.get(opts, :durable, Durable))
+    since = Keyword.get(opts, :since, DateTime.add(DateTime.utc_now(), -86_400, :second))
+
+    Repo.all(
+      config,
+      from(a in ActivityEvent,
+        where: a.at >= ^since,
+        order_by: [desc: a.at],
+        limit: ^Keyword.get(opts, :limit, 100)
+      )
+    )
+  end
+end

durable/lib/durable/config.ex

@@ -49,6 +49,8 @@ defmodule Durable.Config do
           sleep_waker_batch_size: pos_integer(),
           log_level: false | :debug | :info | :warning | :error,
           pubsub: atom() | nil,
+          prune: keyword() | nil,
+          node_id: String.t() | nil,
           owns_pubsub?: boolean()
         }
 
@@ -66,6 +68,8 @@ defmodule Durable.Config do
     :sleep_waker_batch_size,
     :log_level,
     :pubsub,
+    :prune,
+    :node_id,
     owns_pubsub?: false
   ]
 
@@ -140,9 +144,39 @@ defmodule Durable.Config do
           "Pass an atom like `MyApp.PubSub` to reuse a PubSub started by the host app. " <>
           "Pass `:start` to have Durable start its own (named after the instance). " <>
           "Leave as `nil` to disable broadcasting (default)."
+    ],
+    prune: [
+      type:
+        {:or,
+         [
+           nil,
+           {:keyword_list,
+            [
+              completed: [type: :pos_integer, doc: "Keep completed runs this many ms."],
+              failed: [type: :pos_integer, doc: "Keep failed and rolled-back runs this many ms."],
+              cancelled: [type: :pos_integer, doc: "Keep cancelled runs this many ms."]
+            ]}
+         ]},
+      default: nil,
+      doc:
+        "Delete finished runs (with their steps, inputs and events) once they are older " <>
+          "than the given age in ms, e.g. `[completed: :timer.hours(24 * 14)]`. " <>
+          "Statuses left out are kept forever. Off by default."
     ]
   ]
 
+  # One id per running instance: the Erlang node name when distributed, else
+  # the hostname plus a per-boot suffix. Pollers stamp it into `locked_by`,
+  # and the node registry keys its heartbeat row on it.
+  defp node_id do
+    if node() == :nonode@nohost do
+      {:ok, host} = :inet.gethostname()
+      "#{host}-#{System.unique_integer([:positive])}"
+    else
+      Atom.to_string(node())
+    end
+  end
+
   @doc """
   Creates a new validated configuration from options.
 
@@ -160,7 +194,7 @@ defmodule Durable.Config do
     case NimbleOptions.validate(opts, @schema) do
       {:ok, validated} ->
         validated = resolve_pubsub(validated)
-        {:ok, struct(__MODULE__, validated)}
+        {:ok, %{struct(__MODULE__, validated) | node_id: node_id()}}
 
       {:error, %NimbleOptions.ValidationError{}} = error ->
         error

durable/lib/durable/definition.ex

@@ -116,8 +116,7 @@ defmodule Durable.Definition do
             compensations: %{atom() => Compensation.t()},
             opts: %{
               optional(:timeout) => pos_integer(),
-              optional(:max_retries) => pos_integer(),
-              optional(:queue) => atom()
+              optional(:queue) => atom() | String.t()
             }
           }