The native bus

The library sends built-ins

# echo_mq — EchoMQ.Bus: the two publish tiers, one Valkey command each
defp do_publish(:ephemeral, conn, topic, env) do
  payload = Envelope.encode_payload(env)

  case EchoMQ.Connector.command(conn, ["PUBLISH", Topic.channel(topic), payload]) do
    {:ok, n} when is_integer(n) -> :ok
    {:ok, {:error_reply, msg}} -> {:error, {:error_reply, msg}}
    {:error, _} = err -> err
  end
end

# … the durable tier: one XADD through the stream writer
defp do_publish(:durable, conn, topic, env) do
  Stream.append(conn, topic, Topic.stream_name(), Envelope.encode_fields(env))
end

Most bus operations are one command already, the reclaim among them; the failure count, the batch append and the park are compositions, and the next section breaks them.

Where a composition spans commands

# echo_mq — EchoMQ.Bus.DLQ.record_failure/3: the failure count, two commands
def record_failure(conn, key, fail_ttl_ms)
    when is_binary(key) and is_integer(fail_ttl_ms) and fail_ttl_ms > 0 do
  case Connector.command(conn, ["INCR", key]) do
    {:ok, n} when is_integer(n) ->
      # … the TTL in whole seconds, rounded up; the EXPIRE reply is bound to _ and dropped
      ttl_s = div(fail_ttl_ms + 999, 1000)
      _ = Connector.command(conn, ["EXPIRE", key, ttl_s])
      {:ok, n}

    {:ok, other} ->
      {:error, {:unexpected_incr_reply, other}}

    {:error, _} = err ->
      err
  end
end

# echo_mq — EchoMQ.Stream.append_batch/4: N XADDs in one pipeline, the replies matched as _replies
def append_batch(conn, queue, name, records)
    when is_binary(queue) and is_binary(name) and is_list(records) do
  key = stream_key(queue, name)

  {cmds, brandeds} =
    Enum.map_reduce(records, [], fn fields, acc ->
      branded = EchoData.Snowflake.next_branded(Id.kind())
      {:ok, xadd_id} = Id.xadd_id(branded)
      {["XADD", key, xadd_id, "id", branded | flatten(fields)], [branded | acc]}
    end)

  case Connector.pipeline(conn, cmds) do
    {:ok, _replies} -> {:ok, Enum.reverse(brandeds)}
    {:error, _} = err -> err
  end
end

# echo_mq — EchoMQ.Bus.DLQ.park/6: one XADD on the DLQ topic's stream; :ok lets the loop XACK the source
def park(conn, env, topic, sub_id, failures, reason)
    when is_map(env) and is_binary(topic) and is_binary(sub_id) and is_integer(failures) do
  dlq_topic = Topic.dlq_topic(topic)
  parked = merge_provenance(env, topic, sub_id, failures, reason)

  case Stream.append(conn, dlq_topic, Topic.stream_name(), Envelope.encode_fields(parked)) do
    {:ok, _evt} -> :ok
    {:error, reason} -> {:error, {:park_failed, reason}}
  end
end

Each composition is correct while every command lands; its windows open between commands, where a stop or a dropped reply leaves the server holding something the caller was not told.

Three verbs, one server turn

// trafikey — registry::TABLE, the three bus rows after EMQ.STALLED.CHECK: one declared key, one token each
    emq_append => crate::emq::streams::append, Row { name: c"EMQ.APPEND", family: Family::Emq, arity: -5,
        flags: "write deny-oom", keys: EVAL_KEYS, acl: c"write slow", caps: &["append"],
    // …
    emq_failures_incr => crate::emq::bus::failures_incr, Row { name: c"EMQ.FAILURES.INCR", family: Family::Emq, arity: 4,
        flags: "write deny-oom fast", keys: EVAL_KEYS, acl: c"write fast", caps: &["failures.incr"],
    // …
    emq_publish => crate::emq::bus::publish, Row { name: c"EMQ.PUBLISH", family: Family::Emq, arity: -5,
        flags: "write deny-oom fast", keys: EVAL_KEYS, acl: c"write fast", caps: &["publish"],
    // …

// trafikey — emq::streams: the core EMQ.APPEND and EMQ.PUBLISH share, then EMQ.APPEND itself
pub(super) fn append_core(
    ctx: &Context,
    key: &[u8],
    entries: &[Entry<'_>],
    domain: impl Fn(&Entry<'_>) -> Result<(), ValkeyError>,
) -> ValkeyResult {
    // the fence, then each <evt> mapped to its stream ID: an id of namespace EVT, or a refusal
    let _caller = host::fence(ctx)?;
    let ids = entries
        .iter()
        .map(|entry| evt_id(entry.evt).ok_or_else(|| errors::emq(errors::KIND_RECORD)))
        .collect::<Result<Vec<_>, _>>()?;
    // the caller's own check per entry, then the key's type and the stream's top
    entries.iter().try_for_each(domain)?;
    after_top(stream_top(ctx, key)?, &ids)?;
    // the write guard, then one XADD per entry; the reply is the array of their stream IDs
    let w = host::begin_writes(ctx)?;
    apply(&w, &append_plan(key, entries, &ids))?;
    Ok(ValkeyValue::Array(
        ids.iter()
            .map(|id| host::bulk(id.text().as_bytes()))
            .collect(),
    ))
}
// …
pub(crate) fn append(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // numkeys 1, then the shape: the stream layout, n, the split into n entries, no empty field name
    let s = host::shape("EMQ.APPEND", &args, 1)?;
    let entries = append_shape(&s)?;
    append_core(ctx, s.key(1), &entries, |_| Ok(()))
}

// trafikey — emq::bus: EMQ.PUBLISH, the envelope's shape checks before the shared core
fn publish_shape<'a, S: Bytes>(s: &Shape<'a, S>) -> Result<Entry<'a>, ValkeyError> {
    if !matches!(
        keyspace::split_prefixed(s.key(1), b"stream:"),
        Some((_, b"msgs"))
    ) {
        return Err(s.layout_error());
    }
    if parse_count(s.arg(1)) != Some(1) {
        return Err(errors::one_entry(s.verb));
    }
    let entry = parse_entries(s, 1)?
        .into_iter()
        .next()
        .ok_or_else(|| errors::entries(s.verb))?;
    if !is_envelope(&entry) {
        return Err(errors::not_envelope(s.verb));
    }
    Ok(entry)
}
// …
pub(crate) fn publish(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    let s = host::shape("EMQ.PUBLISH", &args, 1)?;
    let entry = publish_shape(&s)?;
    // the core, with one domain check per entry: the msg field holds an id of namespace MSG
    append_core(ctx, s.key(1), &[entry], msg_branded)
}

// trafikey — emq::bus: EMQ.FAILURES.INCR, the count and its TTL in one turn
pub(crate) fn failures_incr(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // shape: numkeys 1, the fail:<sub>:<msg> layout, then fail_ttl_ms as EXPIRE seconds
    let s = host::shape("EMQ.FAILURES.INCR", &args, 1)?;
    let (sub, msg, seconds) = failures_shape(&s, host::now_ms()?)?;
    // the fence, the two brands, the key's type, then what INCR will reply, read before any write
    let _caller = host::fence(ctx)?;
    brands(sub, msg)?;
    let key = s.key(1);
    host::expect_type(ctx, key, b"string")?;
    let stored = match host::read(ctx, cmd::GET, &[key])? {
        Reply::Null => None,
        Reply::Bulk(value) => Some(value),
        _ => return Err(errors::emq(errors::HOST_REPLY)),
    };
    let count = next_count(stored.as_deref()).map_err(errors::emq)?;
    // the write guard, then INCR and EXPIRE <seconds>; the reply is the count
    let w = host::begin_writes(ctx)?;
    apply(&w, &failures_plan(key, seconds))?;
    Ok(ValkeyValue::Integer(count))
}

Natively a failure count and its TTL land in one turn and a batch lands whole or not at all; the park stays two commands, and the library still sends the built-ins.

Pattern & implementation

This dive closes the module; its hub sets the three dives side by side.

References