The native fence

The fence, verbatim

// trafikey — the three constants the fence is built from
/// `{emq}:version`, the cross-queue fence key (`Keyspace.version_key/0`).
pub const VERSION_KEY: &[u8] = b"{emq}:version";
/// The braced reserve for cross-queue keys (`Keyspace.reserve/1`).
pub const RESERVE: &[u8] = b"{emq}:";
// the wire identity held in VERSION_KEY: the one value the fence accepts
pub const EPOCH: &[u8] = b"echomq:6.0.0";

// trafikey — the fence every write verb calls after its shape and layout checks
/// The epoch fence (G0.3 b, C1): `{emq}:version` must hold `echomq:6.0.0`. Read-only; a wrong
/// type passes the server's `WRONGTYPE` through. `{emq}:version` is outside every key-spec
/// (Δ13), so it is opened for read as the caller first (D5.5), and the [`Caller`] is returned for
/// the handler's other derived keys.
pub fn fence(ctx: &Context) -> Result<Caller, ValkeyError> {
    // a cluster-enabled server is refused here, before any key is read
    standalone()?;
    // the calling ACL user, resolved once per command
    let caller = Caller::of(ctx)?;
    // the fence key is outside the key-spec, so the caller's read right on it is checked here
    caller.open(keyspace::VERSION_KEY, Access::Read)?;
    // one GET decides; nothing is written on any path (a WRONGTYPE has already passed through the ?)
    match read(ctx, cmd::GET, &[keyspace::VERSION_KEY])? {
        // the module's wire identity: pass, and hand the checked caller back
        Reply::Bulk(b) if b == keyspace::EPOCH => Ok(caller),
        // any other string: EMQWIRE expected echomq:6.0.0 found <value>
        Reply::Bulk(b) => Err(errors::wire_found(&b)),
        // no value at all: EMQWIRE expected echomq:6.0.0 but {emq}:version is absent
        Reply::Null => Err(errors::emq(errors::WIRE_ABSENT)),
        // any other reply shape: a typed host error
        _ => Err(errors::emq(errors::HOST_REPLY)),
    }
}

// trafikey — the EMQWIRE texts, verbatim
pub const WIRE_ABSENT: &str = "EMQWIRE expected echomq:6.0.0 but {emq}:version is absent";
pub const WIRE_CLUSTER: &str = "EMQWIRE expected a standalone server but cluster mode is enabled";
// …
/// G0: `EMQWIRE expected echomq:6.0.0 found <brand>`; the brand comes from the dataset.
pub fn wire_found(brand: &[u8]) -> ValkeyError {
    // the stored value is printed through printable, so control bytes still read as ASCII
    ValkeyError::String(format!(
        "EMQWIRE expected echomq:6.0.0 found {}",
        printable(brand)
    ))
}

// trafikey — the server mode, cached once at load and consulted by every fence
static CLUSTER: OnceLock<bool> = OnceLock::new();

/// Module init: cache the server mode (D6.4). A second call keeps the first value.
pub(crate) fn load_mode(ctx: &Context) {
    let _ = CLUSTER.set(ctx.get_flags().contains(ContextFlags::CLUSTER));
}

/// Pure, D6.4: standalone admits; a cluster refuses with its `EMQWIRE` text; an unknown mode (init
/// never cached it) refuses as a host failure rather than guess.
fn mode_refusal(cluster: Option<bool>) -> Result<(), ValkeyError> {
    match cluster {
        Some(false) => Ok(()),
        Some(true) => Err(errors::emq(errors::WIRE_CLUSTER)),
        None => Err(errors::emq(errors::HOST_CALL)),
    }
}

/// D6.4: refuse on a cluster-enabled server, before any key is read or written.
pub fn standalone() -> Result<(), ValkeyError> {
    mode_refusal(CLUSTER.get().copied())
}

/// D6.4: EMQ is served only on a server known to be standalone; `EMQ.VERSION` advertises
/// capabilities only then, so R14 clients refuse on a cluster.
pub fn serving() -> bool {
    CLUSTER.get() == Some(&false)
}

The fence is one GET as the caller: the module's own value passes, anything else is named in the refusal, and nothing is written on any path.

Who writes the value

// trafikey — EMQ.INSTALL 1 {emq}:version: the one administrative writer of the fence value
/// H1, H2: numkeys is exactly `1` and `KEYS[1]` is exactly `{emq}:version`.
fn check_shape<S: Bytes>(args: &[S]) -> Result<(), ValkeyError> {
    let s = host::shape(VERB, args, 1)?;
    if s.key(1) == keyspace::VERSION_KEY {
        Ok(())
    } else {
        Err(s.layout_error())
    }
}

/// H3, H4 from the one `GET {emq}:version`: a non-string passes the server's `WRONGTYPE` through
/// (incompatible installed state), another string is a foreign brand. A plain `SET` follows an
/// `Install`, so this check alone protects existing state (the C1/C4 controls rely on it).
fn decide(stored: Result<Reply, ValkeyError>) -> Result<Step, ValkeyError> {
    match stored? {
        // absent: install it
        Reply::Null => Ok(Step::Install),
        // already the module's value: a no-op
        Reply::Bulk(brand) if brand == keyspace::EPOCH => Ok(Step::Equal),
        // a foreign value: refuse and leave it unchanged
        Reply::Bulk(brand) => Err(errors::wire_found(&brand)),
        _ => Err(errors::emq(errors::HOST_REPLY)),
    }
}

/// `EMQ.INSTALL 1 {emq}:version` (install.md#grammar).
pub(crate) fn install(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // shape: numkeys exactly 1 and the key exactly {emq}:version
    check_shape(&args)?;
    // a cluster-enabled server refuses after the shape check, before the read
    host::standalone()?;
    match decide(host::read(ctx, cmd::GET, &[keyspace::VERSION_KEY]))? {
        Step::Equal => Ok(ValkeyValue::Integer(0)),
        Step::Install => {
            // the write guard, then one plain SET, replicated as its effect
            let w = host::begin_writes(ctx)?;
            w.call(cmd::SET, &[keyspace::VERSION_KEY, keyspace::EPOCH])?;
            Ok(ValkeyValue::Integer(1))
        }
    }
}

# EMQ.INSTALL 1 {emq}:version — the four outcomes of its one GET
stored: absent          → SET {emq}:version echomq:6.0.0 → 1
stored: echomq:6.0.0    → 0, nothing written
stored: another string  → EMQWIRE expected echomq:6.0.0 found <value>, nothing written
stored: not a string    → WRONGTYPE …, nothing written

One administrative write, then only reads: the value is installed before the first ordinary client and never touched by one.

EMQ.VERSION and the client rule

// trafikey — EMQ.VERSION: six entries, no key read, no fence
const MODULE_VERSION: &str = env!("CARGO_PKG_VERSION");
const MODULE_MAJOR: i64 = decimal(env!("CARGO_PKG_VERSION_MAJOR"));
/// Reported, never compared (R14). The builder may export `TRAFIKEY_BUILD_ID`.
const BUILD: &str = match option_env!("TRAFIKEY_BUILD_ID") {
    Some(id) => id,
    None => "unset",
};
// …
// read-only, arity 1: any argument gets Valkey's arity error. A map with TRA.CAPABILITIES'
// encoding — a RESP3 map, a flat array of 12 under RESP2. It reads no key and runs no fence.
pub(crate) fn version(ctx: &Context, _: Vec<ValkeyString>) -> ValkeyResult {
    // G1 order is the EMQ rows' table order; a token is advertised only while its command is
    // registered (R14), so a capability always names a served command.
    // on a cluster-enabled server every EMQ command stays registered but refuses, so none is advertised
    let caps = if host::serving() {
        registry::capabilities(Family::Emq, |name| registry::registered(ctx, name))
            .collect::<Vec<_>>()
            .join(" ")
    } else {
        String::new()
    };
    let text = |v: &[u8]| raw::reply_with_string_buffer(ctx.ctx, v.as_ptr().cast(), v.len());
    raw::reply_with_map(ctx.ctx, 6);
    // the wire identity the fence requires
    text(b"wire");
    text(keyspace::EPOCH);
    text(b"module");
    text(b"trafikey");
    text(b"module_version");
    text(MODULE_VERSION.as_bytes());
    text(b"module_major");
    raw::reply_with_long_long(ctx.ctx, MODULE_MAJOR);
    // reported, never compared
    text(b"build");
    text(BUILD.as_bytes());
    // the registered write verbs' tokens, in table order (claim advertises two)
    text(b"capabilities");
    text(caps.as_bytes());
    Ok(ValkeyValue::NoReply)
}

Three comparisons and no key read: wire equal, major equal, the needed tokens listed. Build is reported and never compared.

One node, one fence value

# echo_wire — EchoMQ.Connector: the connect refusals that stop it
@fatal_connect [:version_fence, :auth_refused, :boot_refused, :emq_absent, :emq_version, :r14]

# echo_wire — EchoMQ.Connector, the version read at connect: only the module's
# brand runs; absent or any other value is {:version_fence, …}, fatal.
defp brand(sock, buf) do
  case sync(sock, ["GET", Keyspace.version_key()], buf) do
    {:ok, @brand, _} -> :ok
    {:ok, nil, _} -> {:error, {:version_fence, :absent}}
    {:ok, {:error_reply, msg}, _} -> {:error, {:version_fence, {:server, msg}}}
    {:ok, got, _} -> {:error, {:version_fence, got}}
    {:error, _} = err -> err
  end
end

# echo_wire — EchoMQ.Connector, opened with wire: 6: the module's client
# EMQ.VERSION, then the client rule (R14), then a read of the brand. It
# sends no SET and no EMQ.INSTALL, and every refusal is fatal.
defp fence(sock, buf, 6) do
  with {:ok, g1, buf2} <- emq_version(sock, buf),
       :ok <- r14(g1) do
    brand(sock, buf2)
  end
end
# …
# the client rule: wire equal, module_major 0, and the fourteen job-core
# tokens of @wire6_job_caps listed; every failed condition is reported,
# and build and module_version are never compared.
defp r14(g1) do
  listed = String.split(g1["capabilities"])
  missing = Enum.reject(@wire6_job_caps, &(&1 in listed))

  failures =
    for {key, got, held?} <- [
          {:wire, g1["wire"], g1["wire"] == @wire6_version},
          {:module_major, g1["module_major"], g1["module_major"] == @wire6_module_major},
          {:missing, missing, missing == []}
        ],
        not held?,
        do: {key, got}

  if failures == [], do: :ok, else: {:error, {:r14, failures}}
end

# the brand read: only the module's value passes; absent is refused, never claimed
defp brand(sock, buf) do
  case sync(sock, ["GET", Keyspace.version_key()], buf) do
    {:ok, @wire6_version, _} -> :ok
    {:ok, nil, _} -> {:error, {:version_fence, :absent}}
    {:ok, {:error_reply, msg}, _} -> {:error, {:version_fence, {:server, msg}}}
    {:ok, got, _} -> {:error, {:version_fence, got}}
    {:error, _} = err -> err
  end
end

Two wire identities, one key per node: the fence is why the executors never share a node, not a choice made after the fact.

Recap — the native fence

Next in the native module: One transition, two executors — the claim as the retired @claim2 script and as EMQ.CLAIM, the six differences, the restart exemption, and how parity was kept.

References