myxce/src/lib/raft/raft_sim.c

#ifdef XCE_TESTING

#include <stdio.h>
#include <string.h>

#include "raft.h"

/* ---------- no-op callbacks ---------- */

static void sim_noop_become_follower(void *ctx, uint64_t term, int leader_idx)
{
    (void)ctx; (void)term; (void)leader_idx;
}

static int sim_noop_persist(void *ctx, uint64_t term, int voted_for)
{
    (void)ctx; (void)term; (void)voted_for;
    return 0;
}

static void sim_noop_schedule_timeout(void *ctx, uint32_t delay_ms)
{
    (void)ctx; (void)delay_ms;
}

static void sim_noop_cancel_timeout(void *ctx)
{
    (void)ctx;
}

static void sim_noop_start_election_after_delay(void *ctx, uint32_t delay_ms)
{
    (void)ctx; (void)delay_ms;
}

/* ---------- wired callbacks ---------- */

/*
 * sim_send_vote_request: buffer a pending RPC into the cluster queue.
 *
 * ctx is raft_sim_node_t * (the source node).
 * The node's ctx field points to raft_sim_cluster_t *.
 */
static int sim_send_vote_request(void *ctx, int peer_idx, uint64_t term)
{
    raft_sim_node_t    *node = (raft_sim_node_t *)ctx;
    raft_sim_cluster_t *c    = (raft_sim_cluster_t *)node->ctx;

    if (c->pending_count >= 64)
        return -1; /* queue full — drop */

    raft_sim_rpc_t *rpc = &c->pending[c->pending_count++];
    rpc->src  = node->idx;
    rpc->dst  = peer_idx;
    rpc->term = term;
    return 0;
}

/*
 * sim_become_leader: increment cluster-level leader_count.
 *
 * ctx is raft_sim_node_t *.
 */
static void sim_become_leader(void *ctx, uint64_t term)
{
    raft_sim_node_t    *node = (raft_sim_node_t *)ctx;
    raft_sim_cluster_t *c    = (raft_sim_cluster_t *)node->ctx;
    (void)term;
    c->leader_count++;
}

/* ---------- log-replication callbacks (raft-log-replication) ----------
 *
 * These back the four required log callbacks with the node's in-memory log.
 * ctx is raft_sim_node_t * (same convention as the wired election callbacks).
 * The library holds no entry bytes, so the simulator owns storage here:
 *   - on_persist_entry: append/overwrite at 1-based index (term + bytes)
 *   - on_get_entry_term: read the stored term (0 = absent)
 *   - on_truncate_from: drop entries index..end
 *   - on_send_append_entries: buffer one AE (sourcing the single entry from the
 *     sender's own log at prev_log_index+1 when n_entries == 1)
 * No AppendEntries delivery happens here — that is raft_sim_deliver_append_entries
 * (task 8.2). These callbacks only do storage/term-read/truncate/buffer.
 */

/*
 * sim_persist_entry: durably append or overwrite the entry at 1-based `index`.
 * Sets the slot's term and copies up to RAFT_SIM_ENTRY_LEN bytes; extends
 * log_count when appending past the current tail.
 */
static int sim_persist_entry(void *ctx, uint64_t index, uint64_t term,
                             const void *entry, size_t len)
{
    raft_sim_node_t *node = (raft_sim_node_t *)ctx;

    if (index < 1 || index > RAFT_SIM_MAX_LOG)
        return -1; /* out of fixed sim bounds */

    raft_sim_entry_t *slot = &node->log[index - 1];
    slot->term = term;

    size_t n = len;
    if (n > RAFT_SIM_ENTRY_LEN)
        n = RAFT_SIM_ENTRY_LEN; /* bounded copy */
    if (entry && n > 0)
        memcpy(slot->bytes, entry, n);
    slot->len = n;

    if ((int)index > node->log_count)
        node->log_count = (int)index;

    return 0;
}

/*
 * sim_get_entry_term: return the stored term at 1-based `index`, or 0 if absent
 * (0-means-absent convention).
 */
static uint64_t sim_get_entry_term(void *ctx, uint64_t index)
{
    raft_sim_node_t *node = (raft_sim_node_t *)ctx;

    if (index < 1 || index > (uint64_t)node->log_count)
        return 0;

    return node->log[index - 1].term;
}

/*
 * sim_truncate_from: drop entries at `index`..end (clear term so they read as
 * absent) and shrink log_count.
 */
static int sim_truncate_from(void *ctx, uint64_t index)
{
    raft_sim_node_t *node = (raft_sim_node_t *)ctx;

    if (index < 1)
        return -1;

    for (uint64_t i = index; i <= (uint64_t)node->log_count; i++) {
        if (i <= RAFT_SIM_MAX_LOG) {
            raft_sim_entry_t *slot = &node->log[i - 1];
            memset(slot, 0, sizeof(*slot)); /* term=0 => absent */
        }
    }

    if (node->log_count >= (int)index)
        node->log_count = (int)index - 1;

    return 0;
}

/*
 * sim_send_append_entries: buffer one AppendEntries into the cluster AE queue.
 * For n_entries == 1 the single entry's index/term/bytes are sourced from the
 * sender's own in-memory log at prev_log_index + 1 (the library holds no bytes).
 */
static int sim_send_append_entries(void *ctx, int peer_idx,
                                   uint64_t prev_log_index,
                                   uint64_t prev_log_term,
                                   uint64_t leader_commit, int n_entries)
{
    raft_sim_node_t    *node = (raft_sim_node_t *)ctx;
    raft_sim_cluster_t *c    = (raft_sim_cluster_t *)node->ctx;

    if (c->ae_pending_count >= RAFT_SIM_MAX_AE)
        return -1; /* queue full — drop */

    raft_sim_ae_t *ae = &c->ae_pending[c->ae_pending_count++];
    memset(ae, 0, sizeof(*ae));
    ae->src            = node->idx;
    ae->dst            = peer_idx;
    ae->prev_log_index = prev_log_index;
    ae->prev_log_term  = prev_log_term;
    ae->leader_commit  = leader_commit;
    ae->n_entries      = n_entries;

    if (n_entries == 1) {
        uint64_t idx = prev_log_index + 1;
        ae->entry_index = idx;
        if (idx >= 1 && idx <= (uint64_t)node->log_count) {
            raft_sim_entry_t *slot = &node->log[idx - 1];
            ae->entry_term = slot->term;
            ae->entry_len  = slot->len;
            if (slot->len > 0)
                memcpy(ae->entry_bytes, slot->bytes, slot->len);
        }
    }

    return 0;
}

/*
 * sim_on_trace: append the event string to the node's trace_buf.
 *
 * ctx is raft_sim_node_t *.
 */
static void sim_on_trace(void *ctx, const char *event, const char *detail)
{
    raft_sim_node_t *node = (raft_sim_node_t *)ctx;
    (void)detail;

    if (!event)
        return;

    int remaining = (int)sizeof(node->trace_buf) - node->trace_len - 1;
    if (remaining <= 0)
        return;

    int written = snprintf(node->trace_buf + node->trace_len, (size_t)remaining,
                           "%s\n", event);
    if (written > 0 && written < remaining)
        node->trace_len += written;
    else if (written >= remaining)
        node->trace_len += remaining - 1; /* fill without null — snprintf already null-terminated */
}

/* ---------- public API ---------- */

int raft_sim_init(raft_sim_cluster_t *c, int n_nodes, uint32_t election_timeout_ms)
{
    memset(c, 0, sizeof(*c));

    if (n_nodes < 1 || n_nodes > RAFT_SIM_MAX_NODES)
        return -1;

    c->n             = n_nodes;
    c->leader_count  = 0;
    c->pending_count = 0;

    for (int i = 0; i < n_nodes; i++) {
        raft_sim_node_t *node = &c->nodes[i];

        /* Zero the node struct */
        memset(node, 0, sizeof(*node));

        node->idx       = i;
        node->partition = 0;
        node->ctx       = c; /* points back to the cluster */

        /* Wire callbacks */
        node->cbs.on_send_vote_request         = sim_send_vote_request;
        node->cbs.on_become_leader             = sim_become_leader;
        node->cbs.on_become_follower           = sim_noop_become_follower;
        node->cbs.on_persist                   = sim_noop_persist;
        node->cbs.on_schedule_timeout          = sim_noop_schedule_timeout;
        node->cbs.on_cancel_timeout            = sim_noop_cancel_timeout;
        node->cbs.on_start_election_after_delay = sim_noop_start_election_after_delay;
        /* log-replication callbacks backed by the node's in-memory log */
        node->cbs.on_persist_entry             = sim_persist_entry;
        node->cbs.on_get_entry_term            = sim_get_entry_term;
        node->cbs.on_send_append_entries       = sim_send_append_entries;
        node->cbs.on_truncate_from             = sim_truncate_from;
        node->cbs.on_vote_eligible             = NULL; /* optional — treat all as eligible */
        node->cbs.on_trace                     = sim_on_trace;

        /* Initialize raft state — pass node as ctx so callbacks can find both
         * node->idx (source) and node->ctx (cluster). */
        raft_config_t cfg;
        memset(&cfg, 0, sizeof(cfg));
        cfg.n_peers             = n_nodes;
        cfg.self_idx            = i;
        cfg.my_priority         = i; /* priority = index: node 0 has highest priority */
        cfg.election_timeout_ms = election_timeout_ms;
        cfg.initial_term        = 0;
        cfg.initial_voted_for   = -1;
        cfg.cbs                 = &node->cbs;
        cfg.ctx                 = node; /* node* as ctx — callbacks cast back to raft_sim_node_t* */

        int rc = raft_init(&node->state, &cfg);
        if (rc != 0)
            return rc;
    }

    return 0;
}

void raft_sim_dump_log(raft_sim_cluster_t *c, FILE *fp)
{
    for (int i = 0; i < c->n; i++) {
        raft_sim_node_t *node = &c->nodes[i];
        if (node->trace_len > 0) {
            fprintf(fp, "[node%d trace]\n%.*s\n",
                    i, node->trace_len, node->trace_buf);
        }
    }
}

void raft_sim_deliver_pending(raft_sim_cluster_t *c)
{
    /* Snapshot current count — callbacks may add more entries */
    int count = c->pending_count;
    c->pending_count = 0;

    for (int i = 0; i < count; i++) {
        raft_sim_rpc_t *rpc = &c->pending[i];
        int src = rpc->src;
        int dst = rpc->dst;
        uint64_t term = rpc->term;

        /* Drop if either endpoint is partitioned */
        if (c->nodes[src].partition || c->nodes[dst].partition)
            continue;

        /* Deliver vote request to dst */
        uint64_t reply_term = 0;
        int granted = raft_on_vote_request(&c->nodes[dst].state, src, term,
                                           &reply_term);

        /* Route reply back to src */
        raft_on_vote_reply(&c->nodes[src].state, dst, granted, reply_term);
    }
}

void raft_sim_start_all_elections(raft_sim_cluster_t *c)
{
    for (int i = 0; i < c->n; i++)
        raft_start_election(&c->nodes[i].state);
}

void raft_sim_fire_timeout(raft_sim_cluster_t *c, int node_idx)
{
    if (node_idx >= 0 && node_idx < c->n)
        raft_on_election_timeout(&c->nodes[node_idx].state);
}

void raft_sim_partition(raft_sim_cluster_t *c, int node_idx, int isolated)
{
    if (node_idx >= 0 && node_idx < c->n)
        c->nodes[node_idx].partition = isolated;
}

#endif /* XCE_TESTING */