feat(queue): add :sort_key runtime option to control ordering

Per-push :sort_key option on Tidefall.Queue (mirrors :partition_key),
validated via the new Tidefall.Queue.Options. ETS :ordered_set key is
{sort_key, ref}: sort_key defaults to System.monotonic_time() or is
derived from an arity-1/arity-0 function; ref is retained so distinct
items never collide. Ordering is per-partition. Includes tests, docs,
usage-rules, and a CHANGELOG entry.

Author: cabol

CHANGELOG.md

@@ -5,13 +5,24 @@ All notable changes to this project will be documented in this file.
 This project adheres to
 [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+> [Full Changelog](https://github.com/cabol/tidefall/compare/v1.0.0-rc.0...HEAD)
+
+### Enhancements
+
+- [Tidefall.Queue] Added the `:sort_key` runtime option on `push/3` to control
+  how buffered items are ordered within a partition. Defaults to insertion
+  order; accepts a function (arity 1 applied to each item, or arity 0 evaluated
+  per item) that returns the sort term. Ordering is per partition and no item
+  is ever dropped.
+
 ## [v1.0.0-rc.0](https://github.com/cabol/tidefall/tree/v1.0.0-rc.0) (2026-06-13)
 
 ### Enhancements
 
-- `Tidefall.Queue` — insertion-ordered ETS buffer (`:ordered_set`) that
+- [Tidefall.Queue] Insertion-ordered ETS buffer (`:ordered_set`) that
   accumulates items and drains them to a processor in periodic batches.
-- `Tidefall.HashMap` — coalescing key-value buffer (`:set`, last-write-wins)
+- [Tidefall.HashMap] Coalescing key-value buffer (`:set`, last-write-wins)
   with version-aware conditional writes (`put_newer/4`, `put_all_newer/3`)
   and an optional `:key_hasher` for complex keys.
 - Module-based buffers via `use Tidefall.Queue` / `use Tidefall.HashMap` —

lib/tidefall/buffer.ex

@@ -25,6 +25,10 @@ defmodule Tidefall.Buffer do
 
   #{Tidefall.Buffer.Options.runtime_options_docs()}
 
+  Each buffer type may accept additional runtime options of its own —
+  see `Tidefall.Queue` (e.g. `:sort_key`) and `Tidefall.HashMap`
+  (e.g. `:key_hasher`) for their full option docs.
+
   """
 
   alias Tidefall.Buffer.{Options, Partition}
@@ -165,7 +169,7 @@ defmodule Tidefall.Buffer do
 
   ## Private functions
 
-  @compile [inline: [lookup: 1]]
+  @compile inline: [lookup: 1]
   defp lookup(buffer) do
     Registry.lookup(Tidefall.Registry, buffer)
   end

lib/tidefall/hash_map.ex

@@ -639,7 +639,7 @@ defmodule Tidefall.HashMap do
   ## Private functions
 
   # Iniline common instructions
-  @compile [inline: [new_entry: 3, new_entry: 4]]
+  @compile inline: [new_entry: 3, new_entry: 4]
 
   defp new_entry(ets_key, raw_key, value) do
     entry(key: ets_key, raw_key: raw_key, value: value)

lib/tidefall/queue.ex

@@ -23,7 +23,7 @@ defmodule Tidefall.Queue do
      |         |         |
      v         v         v
   +-------+ +-------+ +-------+     ETS :ordered_set
-  | P 0   | | P 1   | | P N-1 |     Key: {monotonic_time, ref}
+  | P 0   | | P 1   | | P N-1 |     Key: {sort_key, ref}
   +-------+ +-------+ +-------+     Val: item
      |         |         |
      v         v         v
@@ -34,17 +34,20 @@ defmodule Tidefall.Queue do
   ```
 
   Items are routed to partitions via `phash2`, stored in
-  `:ordered_set` ETS tables keyed by `{monotonic_time, ref}`
-  (ensuring insertion-time ordering with uniqueness), and
-  periodically flushed to the processor in batches.
+  `:ordered_set` ETS tables keyed by `{sort_key, ref}`, and
+  periodically flushed to the processor in batches. `sort_key`
+  is `System.monotonic_time()` by default — giving insertion-time
+  order — or the term produced by the `:sort_key` runtime option;
+  `ref` keeps every key unique so no item is ever overwritten.
+  Ordering is **per partition** (see `:sort_key` under runtime options).
 
   ## Start options
 
   #{Tidefall.Buffer.Options.start_options_docs()}
 
   ## Runtime options
 
-  #{Tidefall.Buffer.Options.runtime_options_docs()}
+  #{Tidefall.Queue.Options.runtime_options_docs()}
 
   ## Examples
 
@@ -112,13 +115,18 @@ defmodule Tidefall.Queue do
   import Record, only: [defrecordp: 2]
 
   alias Tidefall.Buffer
-  alias Tidefall.Buffer.{Definition, Options, Partition}
-
-  # Queue-specific key record (ordered by insertion time).
-  # The `timestamp` ensures order by insertion time (asc) while the
-  # `ref` makes each entry unique since there may be multiple entries
-  # with the same timestamp.
-  defrecordp(:key, timestamp: nil, ref: nil)
+  alias Tidefall.Buffer.{Definition, Partition}
+  alias Tidefall.Queue.Options
+
+  # Queue-specific key record.
+  #
+  # `sort_key` is the primary ordering term — `System.monotonic_time()`
+  # by default (insertion order), or whatever the `:sort_key` runtime
+  # option resolves to. `ref` is always retained as the uniqueness
+  # tiebreaker: since `make_ref()` is unique, distinct items never
+  # collide in the `:ordered_set`, so nothing is overwritten even when
+  # two items share the same `sort_key`.
+  defrecordp(:key, sort_key: nil, ref: nil)
 
   # Entry record stored in ETS. Queue only needs key/value; the
   # match spec returns just the value to the processor.
@@ -223,20 +231,24 @@ defmodule Tidefall.Queue do
       # Custom partition routing with fixed key (all items to same partition)
       push(:my_buffer, log_entry, partition_key: :logs)
 
+      # Custom ordering: drain by a value-derived sort key (per partition)
+      push(:my_buffer, event, sort_key: & &1.priority)
+
   """
   @spec push(buffer(), item() | [item()], keyword()) :: :ok
   def push(buffer, item_or_batch, opts \\ [])
 
   def push(buffer, batch, opts) when is_list(batch) do
     opts = Options.validate_runtime_options!(opts)
     partition_key = Keyword.fetch!(opts, :partition_key)
+    sort_key = Keyword.get(opts, :sort_key)
 
     batch
     |> Enum.group_by(&Buffer.get_partition(buffer, partition_key, &1))
     |> Enum.each(fn {partition, items} ->
       partition
       |> Partition.current_table()
-      |> :ets.insert(Enum.map(items, &new_entry(build_key(), &1)))
+      |> :ets.insert(Enum.map(items, &new_entry(build_key(sort_key, &1), &1)))
     end)
   end
 
@@ -300,9 +312,22 @@ defmodule Tidefall.Queue do
   ## Private functions
 
   # Iniline common instructions
-  @compile [inline: [build_key: 0, new_entry: 2]]
+  @compile inline: [new_entry: 2]
+
+  # Default (no `:sort_key`): order by insertion time.
+  defp build_key(nil, _item) do
+    key(sort_key: System.monotonic_time(), ref: make_ref())
+  end
 
-  defp build_key, do: key(timestamp: System.monotonic_time(), ref: make_ref())
+  # Arity-1: derive the sort term from the item.
+  defp build_key(fun, item) when is_function(fun, 1) do
+    key(sort_key: fun.(item), ref: make_ref())
+  end
+
+  # Arity-0: generate the sort term at push time.
+  defp build_key(fun, _item) when is_function(fun, 0) do
+    key(sort_key: fun.(), ref: make_ref())
+  end
 
   defp new_entry(key, value), do: entry(key: key, value: value)
 end

lib/tidefall/queue/options.ex

@@ -0,0 +1,59 @@
+defmodule Tidefall.Queue.Options do
+  @moduledoc false
+
+  alias Tidefall.Buffer
+
+  # Queue-specific runtime options, layered on top of the shared runtime
+  # options exposed by `Tidefall.Buffer.Options` (currently `:partition_key`).
+  # `:sort_key` is resolved per call (like `:partition_key`), so the client
+  # builds the ETS key without any per-buffer lookup.
+  queue_runtime_opts = [
+    sort_key: [
+      type: {:or, [{:fun, 1}, {:fun, 0}]},
+      required: false,
+      doc: """
+      Controls the term used to order buffered items within a partition.
+      The ETS key is `{sort_key_term, ref}`; `ref` is always retained for
+      uniqueness, so distinct items are never overwritten regardless of the
+      `:sort_key` value.
+
+      Can be one of:
+
+        * Omitted (default) — items order by insertion time
+          (`System.monotonic_time/0`), i.e. the order they were pushed.
+        * A function of arity 1 — applied to each item to derive its sort
+          term (e.g. `& &1.priority`, or an event timestamp carried in the
+          payload).
+        * A function of arity 0 — evaluated per item to generate the sort
+          term at push time (e.g. a custom clock or sequence).
+
+      > #### Ordering scope and ties {: .info}
+      >
+      > Ordering is **per partition** — items are routed across partitions
+      > first, and each partition's batch is ordered independently. For a
+      > single global order use `partitions: 1` or a `:partition_key` that
+      > co-locates the items you need ordered together. Order among items
+      > with the **same** sort term is unspecified (broken by `ref`),
+      > consistent with how same-timestamp items already drain today. Sort
+      > terms are compared with Erlang's total term order, so prefer
+      > integers, atoms, or binaries; complex terms (maps, tuples) sort in
+      > non-obvious ways.
+      """
+    ]
+  ]
+
+  # Runtime schema: shared opts + Queue opts.
+  @runtime_opts_schema NimbleOptions.new!(Buffer.Options.runtime_opts() ++ queue_runtime_opts)
+
+  ## API
+
+  @spec runtime_options_docs() :: binary()
+  def runtime_options_docs do
+    NimbleOptions.docs(@runtime_opts_schema)
+  end
+
+  @spec validate_runtime_options!(keyword()) :: keyword()
+  def validate_runtime_options!(opts) do
+    NimbleOptions.validate!(opts, @runtime_opts_schema)
+  end
+end

test/tidefall/queue/options_test.exs

@@ -0,0 +1,49 @@
+defmodule Tidefall.Queue.OptionsTest do
+  use ExUnit.Case, async: true
+
+  alias Tidefall.Queue.Options
+
+  describe "validate_runtime_options!/1 — :sort_key" do
+    test "ok: accepts an arity-1 function" do
+      fun = fn item -> item end
+
+      assert Options.validate_runtime_options!(sort_key: fun)[:sort_key] == fun
+    end
+
+    test "ok: accepts an arity-0 function" do
+      fun = fn -> 1 end
+
+      assert Options.validate_runtime_options!(sort_key: fun)[:sort_key] == fun
+    end
+
+    test "ok: omitting :sort_key validates and leaves it unset" do
+      refute Keyword.has_key?(Options.validate_runtime_options!([]), :sort_key)
+    end
+
+    test "ok: the shared :partition_key option still validates alongside :sort_key" do
+      opts = Options.validate_runtime_options!(sort_key: fn -> 1 end, partition_key: :p)
+
+      assert opts[:partition_key] == :p
+    end
+
+    test "error: rejects a non-function term" do
+      for bad <- [:high, 123, "x", %{}] do
+        assert_raise NimbleOptions.ValidationError, fn ->
+          Options.validate_runtime_options!(sort_key: bad)
+        end
+      end
+    end
+
+    test "error: rejects a function of the wrong arity" do
+      assert_raise NimbleOptions.ValidationError, fn ->
+        Options.validate_runtime_options!(sort_key: fn _a, _b -> :x end)
+      end
+    end
+  end
+
+  describe "runtime_options_docs/0" do
+    test "ok: renders the :sort_key option" do
+      assert Options.runtime_options_docs() =~ "sort_key"
+    end
+  end
+end