The job core, natively · dive 02

The lease epoch

Every native lease records the server process that granted it. Only the stall sweep reads the record, to tell a server restart from a worker's stall.

Valkey gives each server process a run_id: forty lowercase hex characters in INFO server, generated when the process starts. A restart produces a new one. trafikey reads it once at load and holds it in memory. Three verbs write it on the job row as lease_epoch whenever they grant or extend a lease: EMQ.CLAIM, EMQ.EXTEND and EMQ.EXTENDS. One verb reads it: EMQ.STALLED.CHECK. A lease whose epoch is the current process's ran out under a live server and counts against the stall budget; any other lease crossed a restart and goes back to pending uncounted. Test the shape check and the exemption on the tests' own values, then follow one lease across a restart.

One value per process Three writers One reader One lease across a restart References

One value per process

The module's load function reads the epoch before it finishes loading. If the read fails, it logs a warning and refuses the load, so a server that serves the EMQ verbs always holds an epoch. The read itself goes through the three raw server-info functions of the module API: open the server section, take the run_id field, free the section.

// trafikey — init, the module's load function: the epoch read
fn init(ctx: &Context, _: &[ValkeyString]) -> Status {
    // …
    // the server epoch is read once at load and held in memory; no key is written
    if let Err(e) = emq::epoch::load(ctx) {
        ctx.log_warning(&format!(
            "trafikey: emq epoch not loaded: {}",
            emq::errors::text(&e)
        ));
        return Status::Err;
    }
    // …

// trafikey — host::server_run_id: INFO server's run_id through three raw server-info calls
pub fn server_run_id(ctx: &Context) -> Result<Vec<u8>, ValkeyError> {
    let get = api(unsafe { raw::RedisModule_GetServerInfo })?;
    let field = api(unsafe { raw::RedisModule_ServerInfoGetFieldC })?;
    let free = api(unsafe { raw::RedisModule_FreeServerInfo })?;
    let info = unsafe { get(ctx.ctx, c"server".as_ptr()) };
    if info.is_null() {
        return Err(errors::emq(errors::HOST_CALL));
    }
    let value = unsafe { field(info, c"run_id".as_ptr()) };
    let id = (!value.is_null()).then(|| {
        unsafe { std::ffi::CStr::from_ptr(value) }
            .to_bytes()
            .to_vec()
    });
    unsafe { free(ctx.ctx, info) };
    id.ok_or_else(|| errors::emq(errors::HOST_CALL))
}

emq::epoch holds the rest. The value must be exactly forty lowercase hex characters, the form Valkey prints; anything else refuses the load with ERR EMQ host call failed. A OnceLock caches it for the life of the process, and a second load keeps the first value. current() hands the cached bytes to a handler, or refuses before any write when nothing was loaded. exempt is the whole rule the sweep applies: a lease is exempt when its lease_epoch is missing or differs from the current run_id.

// trafikey — emq::epoch: the length, the cache, the shape check, the load, the read and the
// one rule the sweep applies
/// `CONFIG_RUN_ID_SIZE` (Valkey `server.h`).
const RUN_ID_LEN: usize = 40;

static RUN_ID: OnceLock<Vec<u8>> = OnceLock::new();

/// A `run_id` as Valkey prints it: exactly 40 lowercase hex characters (`getRandomHexChars`).
fn valid_run_id(id: &[u8]) -> bool {
    id.len() == RUN_ID_LEN
        && id
            .iter()
            .all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(c))
}

/// Reads `INFO server` `run_id` once (host.rs [`super::host::server_run_id`]) and caches it for
/// the process. A second call keeps the first value (one process, one `run_id`). A missing host
/// API or a malformed value refuses typed, and module init then refuses the load.
pub(crate) fn load(ctx: &Context) -> Result<(), ValkeyError> {
    let id = host::server_run_id(ctx)?;
    if !valid_run_id(&id) {
        return Err(errors::emq(errors::HOST_CALL));
    }
    let _ = RUN_ID.set(id);
    Ok(())
}

/// The cached `run_id`. Before [`load`] ran there is no epoch, and a caller that needs one must
/// refuse before any write rather than guess.
pub(crate) fn current() -> Result<&'static [u8], ValkeyError> {
    RUN_ID
        .get()
        .map(Vec::as_slice)
        .ok_or_else(|| errors::emq(errors::HOST_CALL))
}

/// Pure, Δ15: an expired lease is restart-caused (exempt from the stall count) when its
/// `lease_epoch` is missing or differs from the current `run_id`.
pub(crate) fn exempt(lease_epoch: Option<&[u8]>, current: &[u8]) -> bool {
    lease_epoch != Some(current)
}

The tests use one value as the current process, NOW, and two families of others: five strings the shape check must reject, and three stored fields the exemption must treat as foreign. Pick a value and apply both checks to it.

// trafikey — the epoch tests: what load accepts, and what the sweep counts
    const NOW: &[u8] = b"3f2c0d8e9b1a4c5d6e7f8091a2b3c4d5e6f70812";

    #[test]
    fn run_ids_are_40_lowercase_hex() {
        assert!(valid_run_id(NOW));
        for bad in [
            &b""[..],
            &NOW[..39],
            b"3F2C0D8E9B1A4C5D6E7F8091A2B3C4D5E6F70812",
            b"3f2c0d8e9b1a4c5d6e7f8091a2b3c4d5e6f7081g",
            b"3f2c0d8e9b1a4c5d6e7f8091a2b3c4d5e6f708123",
        ] {
            assert!(!valid_run_id(bad), "{bad:?}");
        }
    }

    #[test]
    fn exemption_is_missing_or_different() {
        assert!(!exempt(Some(NOW), NOW), "same process: counted");
        // … a missing field is exempt as well: exempt(None, NOW)
        assert!(
            exempt(Some(b"0000000000000000000000000000000000000000"), NOW),
            "another process: exempt"
        );
        assert!(exempt(Some(b""), NOW), "an empty field is not this process");
    }

Two checks on one value · pick a value from the tests

3f2c0d8e9b1a4c5d6e7f8091a2b3c4d5e6f70812 · 40 characters
as the server’s run_id at load: valid_run_id → true, so load caches it for the process
as a row’s lease_epoch, swept while run_id is NOW: exempt → false: counted, stalled + 1 against max_stalled
the test: “same process: counted”

The shape check guards the value at load; the exemption compares the bytes stored on the row. A stored epoch was a valid run_id when it was written, so the sweep never re-checks its shape.

Three writers

EMQ.CLAIM takes the epoch in its read pass, beside the predicted attempts and the deadline, and writes it in the same HSET that sets state to active. A claim never leaves an active row without an epoch.

// trafikey — EMQ.CLAIM, the stamp: the epoch is taken in the read pass, written with the state
        planned = Some((att, at, epoch::current()?));
    // …
    // the lease carries the server epoch: state and lease_epoch in one HSET
    w.call(
        cmd::HSET,
        &[&row, b"state", b"active", b"lease_epoch", run_id],
    )?;
    w.call(cmd::ZADD, &[active, at.to_string().as_bytes(), &id])?;

An extension re-scores the lease in active and writes the epoch again. EMQ.EXTEND does it for one row; EMQ.EXTENDS does it for every live pair after its read pass, under one now. A worker that extends its lease after a restart therefore moves the lease to the new process: the stamp always names the process that last granted the lease.

// trafikey — EMQ.EXTEND and EMQ.EXTENDS, the re-stamp on every extension
    let at = host::deadline(s.verb, host::now_ms()?, lease)?;
    let run_id = epoch::current()?;
    let w = host::begin_writes(ctx)?;
    w.call(cmd::ZADD, &[s.key(1), at.to_string().as_bytes(), s.arg(1)])?;
    // the verdict read attempts from the row, so the row is a hash and this write cannot fail
    w.call(cmd::HSET, &[s.key(2), b"lease_epoch", run_id])?;
    Ok(ValkeyValue::Integer(1))
}
// …
    if !live.is_empty() {
        for (_, row) in &live {
            caller.open(row, Access::Write)?;
        }
        let run_id = epoch::current()?;
        let score = at.to_string();
        let w = host::begin_writes(ctx)?;
        // each live lease is re-scored and stamped; the read pass proved each live row a hash
        for (id, row) in &live {
            w.call(cmd::ZADD, &[s.key(1), score.as_bytes(), id])?;
            w.call(cmd::HSET, &[row.as_slice(), b"lease_epoch", run_id])?;
        }
    }

Each writer calls epoch::current() before host::begin_writes, so a missing epoch refuses the command with nothing written. The scripts write no lease_epoch: @claim2 and the scripted extensions leave the row as wire 4 defined it.

Grant and extend write the stamp; nothing else does. The stamp is the process that last granted the lease.

One reader

EMQ.STALLED.CHECK is the only command that reads lease_epoch. It takes the current epoch right after the fence, selects the expired members of active, and asks decide_member for a verdict on each. The verdict reads lazily: a ghost stops at TYPE, a lane row stops at group, and an exempt member stops at lease_epoch, before stalled is read.

// trafikey — the one reader: decide_member's epoch step, and the sweep that feeds it
fn decide_member(
    run_id: &[u8],
    max_stalled: i64,
    seen: &MemberSeen,
) -> Step<MemberRead, Result<Swept, SweepRefusal>> {
    // …
    let Some(lease_epoch) = &seen.epoch else {
        return Step::Need(MemberRead::Epoch);
    };
    if epoch::exempt(lease_epoch.as_deref(), run_id) {
        return Step::Done(Ok(Swept::Exempt));
    }
    // …
pub(crate) fn stalled_check(ctx: &Context, args: Vec<ValkeyString>) -> ValkeyResult {
    // …
    let caller = host::fence(ctx)?;
    let run_id = epoch::current()?;
    // …
                Step::Need(MemberRead::Epoch) => {
                    seen.epoch = Some(hget(ctx, &row, b"lease_epoch")?)
                }
            // …

An exempt member is removed from active, added to pending and set to state pending; its stalled field is not written, and it is listed among the recovered ids. A counted member gets stalled plus one and is recovered or dead-lettered by max_stalled, as the next dive shows.

A lease from the current process counts; a lease from any other process, or with no stamp, goes back uncounted.

One lease across a restart

Four histories of one lease, each ending in the same sweep. The dataset survives the restart; what changes is the run_id. Choose a history and a max_stalled; the job's row holds no stalled field before the sweep.

One lease, one sweep · pick a history

history

max_stalled

sweep: run_id 0000…0000 · lease_epoch 3f2c…0812 → exempt = true
exempt: ZREM active, ZADD pending 0, HSET state pending; stalled is not written, and the job is in the recovered list
why: the lease was written by the process before the restart: the server stopped, not the worker

With max_stalled at 1, a counted expiry is a death: the first counted stall dead-letters the job. Without the epoch, every lease that ran out across a restart would count, and at that setting every one of them would be dead-lettered. With the epoch those leases go back to pending, and the budget is spent only on stalls that happened under a live server. A lease written before the stamp existed has no field, which is foreign by the same rule, so an upgrade restart dead-letters nothing.

The epoch moves the line between "the server went away" and "the worker went away" into the row itself, at the moment the lease is granted.

Pattern & implementation

the pattern (Redis Patterns Applied)

An expired lock is the signal that a holder died, and a recovery pass turns that signal into a retry. Lock expiry detection teaches it.

the implementation (trafikey)

The lease carries the identity of the process that granted it, so the sweep can tell an expiry the server caused, by restarting, from one a worker caused, by stopping, and charge only the second.

The next dive reads the budgets the sweep and the failure verb charge: failures and recovery.

References

Sources

Valkey — INFO — the server section and its run_id field.
Valkey — Modules API reference — the server-info calls behind host::server_run_id.
Valkey — HSET — the write that stamps lease_epoch.
Valkey — ZRANGE — the BYSCORE selection of expired leases.

Related in this course

The job core, natively — the module this dive belongs to.
One transition, two executors — the claim on both executors and the restart exemption beside it.
Loading the module — the load function the epoch read belongs to.
Claim and the lease — the scripted lease the native one extends.
redis-patterns · Lock expiry detection — the expiry as the signal of a dead holder.
Failures and recovery — the next dive.