jonnifyechomq

Retention & the archive · dive 01

Retention is a policy

Bound the live log by length or age — on its own beat, never coupled to a consumer being up.

An append-only log appends forever, so the live log has to be bounded. EchoMQ.Stream.trim/4 caps it — by length (MAXLEN) or by age (MINID), removing the entries outside a declared window over XTRIM issued direct. The trap the design refuses: coupling a safety property (bounded memory) to a liveness fact (a consumer is up). Fold the trim into a consumer's loop and a stream nobody drains grows forever. So the cadence lives on its own beat. Pick a window on the right to watch the trim bound the log — and read why the approximate ~ default can never delete inside it. It is all real code.

The two windows The safe default The opt-in driver Pattern & implementation References

A trim window · pick the bound

emq:{q}:stream:<name> · ordered by mint EVT EVT EVT EVT EVT EVT EVT EVT head — oldest tail — newest kept — the window removed — outside INV4 — a trim never deletes inside the window

The two windows

EchoMQ.Stream.trim(conn, queue, name, window) removes the entries outside a declared window over XTRIM issued direct — no new script, the same no-Lua discipline the whole stream tier follows. Two window forms, one bounds by length and one by age:

The {:maxlen, count, approx?} form keeps the count newest entries and removes the older. The {:minid, %DateTime{}, approx?} form removes every entry minted strictly before the instant — the floor id derived from minid_floor/1 (ms = DateTime.to_unix(dt, :millisecond), the rung's one piece of real id-math), never a raw snowflake integer to the wire. It answers {:ok, removed_count} or {:error, term} verbatim — a WRONGTYPE against a non-stream key is surfaced, not swallowed; and it raises before any wire on a malformed queue/stream name, policy before existence before write.

# echo_mq — EchoMQ.Stream
# trim/4 removes the entries OUTSIDE a declared window over XTRIM issued DIRECT
# (no new script — the no-Lua rung). Two window forms: by length, and by age.
def trim(conn, queue, name, {:maxlen, count, approx?})
    when is_integer(count) and count >= 0 and is_boolean(approx?) do
  key = stream_key(queue, name)
  # keep the `count` newest entries, remove the older
  xtrim(conn, ["XTRIM", key, "MAXLEN", approx_flag(approx?), Integer.to_string(count)])
end

def trim(conn, queue, name, {:minid, %DateTime{} = dt, approx?})
    when is_boolean(approx?) do
  key = stream_key(queue, name)
  # remove every entry minted strictly BEFORE `dt`; the floor id is "<ms>-0"
  # from minid_floor/1 — never a raw snowflake integer to the wire (INV6)
  xtrim(conn, ["XTRIM", key, "MINID", approx_flag(approx?), minid_floor(dt)])
end

One verb, two windows: MAXLEN bounds the log by how many, MINID bounds it by how old — and either way XTRIM removes only the entries outside the bound.

The safe default — ~ over =

The third element of the window, approx?, selects the trim mode — and the default is the safe one. true maps to ~ (approximate): XTRIM trims in whole macro-nodes, so it may under-trim (leave a few extra entries) but can never over-trim. Because it can never reach past the window edge, it can never delete an entry inside the window — that is INV4, true by construction. false maps to = (exact): it removes precisely to the window edge, the opt-in.

Either way the blast radius is bounded by the window: a trim can never delete an entry inside it. That is the guarantee that makes a default-on trim safe — it can only ever shrink the part of the log you declared expendable. The approximate form is also cheaper (whole macro-nodes, not a precise scan), so the safe choice is the fast one.

Approximate vs exact · what each removes

EVT EVT EVT EVT EVT EVT EVT EVT MAXLEN 4 — the window (always kept) ·

The default is ~ for a reason: it cannot delete inside the window, and it is cheaper. The exact = form is there when a precise length matters more than cost — but it buys precision at the edge, never safety inside it; both honour INV4.

The opt-in driver — EchoMQ.StreamRetention

A trim is a single call; continuous bounded memory is a cadence. EchoMQ.StreamRetention is the named, opt-in driver for that cadence: a :transient GenServer that beats on :tick_ms (default 1_000) and, on each beat, re-applies a declared per-stream :policy — a list of {queue, name, window} — via the public EchoMQ.Stream.trim/4. Three properties make it safe.

Decoupled from consumer liveness. Retention is a property of the stream, not of a consumer: a stream nobody drains still trims if its policy is declared here, and the EchoMQ.StreamConsumer loop is never touched. Opt-in, owner-started. There is no auto-start — a deployment that wants continuous bounded memory starts the driver; a stream you want unbounded is never declared, and so is never silently trimmed. BEAM-side policy, idempotent. The policy is process state, not a keyspace subkey, so there is no at-rest cleanup; and the trim is idempotent over the stream, so a :transient restart loses no guarantee and over-deletes nothing. A manual trim/4 is the equally-supported cadence — the driver is sugar over the verb.

# echo_mq — EchoMQ.StreamRetention
# One sweep, exposed for a direct-drive test: apply the DECLARED policy's trim
# calls via the public EchoMQ.Stream.trim/4, at the injected clock instant.
# Decoupled from any EchoMQ.StreamConsumer — a stream with no consumer still trims.
def sweep(%{conn: conn, policy: policy, clock: clock}) do
  case Core.decide(policy, clock.()) do
    :noop ->
      {:ok, %{trimmed: 0, calls: 0}}

    calls when is_list(calls) ->
      {removed, applied} =
        Enum.reduce(calls, {0, 0}, fn {queue, name, window}, {rem_acc, n_acc} ->
          # each trim is soft-matched — a wire hiccup on one stream is logged and
          # skipped, never crash-looping the cadence; the next tick re-applies the
          # idempotent window (a transient fault loses no retention guarantee).
          case apply_trim(conn, queue, name, window) do
            {:ok, removed} -> {rem_acc + removed, n_acc + 1}
            :skipped -> {rem_acc, n_acc + 1}
          end
        end)

      {:ok, %{trimmed: removed, calls: applied}}
  end
end

The driver is sugar over the verb: a declared, BEAM-side, idempotent policy re-applied on its own beat — so bounded memory is a property the stream owns, not a side effect of a consumer being up.

Worked on a codemojex activity feed (the GES guess events appended on the game lifecycle): keep the last 10_000 events resident and no more. Declare a length window and start the driver — every beat re-caps the feed, and the ~ default never deletes one of the newest 10_000.

# bound a codemojex activity feed to its last 10_000 events, on its own beat
{:ok, _} =
  EchoMQ.StreamRetention.start_link(
    conn: conn,
    tick_ms: 1_000,
    policy: [{"codemojex", "activity", {:maxlen, 10_000, true}}]
  )

Pattern & implementation

the pattern (Redis Patterns Applied)

A stream needs retention; XTRIM MAXLEN caps it by length and XTRIM MINID by age; the approximate ~ form is cheaper and safe. Streams & Events teaches stream retention.

the implementation (echo_mq)

EchoMQ.Stream.trim/4 issues XTRIM direct, mapping the two declared windows; ~ is the safe default that can never delete inside the window (INV4); and the EchoMQ.StreamRetention driver re-applies a declared BEAM-side policy on its own beat, decoupled from consumer liveness.

The pattern says: bound the log. The implementation makes the bound a policy the stream owns — safe by default, decoupled from who is reading, and idempotent across a restart.

Recap — a policy, not a default

Retention bounds the live log by length (MAXLEN) or age (MINID), on its own beat — never coupled to a consumer being up. The ~ approximate default may under-trim but can never over-trim, so it can never delete inside the window (INV4); the exact = is the opt-in. The EchoMQ.StreamRetention driver re-applies a declared, BEAM-side, idempotent policy, opt-in and owner-started, and a manual trim/4 is the equally-supported cadence. What this dive leaves open: where the trimmed records go. They are folded to the durable floor first, so nothing is lost — the next dive.

Next in the module: Nothing is lost — the fold into the Graft floor that runs before the trim, and the merge-read that serves the archived past beside the live tail.

References

Where the trim and its windows are grounded, and the approximate trim semantics they rest on.

Sources

Related in this course