RedisRaft 的日志一致性问题

以下单文件 CuTest:

/* Standalone regression for an empty AppendEntries committing an unmatched tail.
 * From the library root (output outside the source tree):
 * cc -std=gnu11 -O0 -g -Iinclude -Itests tests/test_commit_bound.c \
 *    tests/CuTest.c src/raft_server.c src/raft_server_properties.c \
 *    src/raft_node.c src/raft_log.c -o /tmp/test_commit_bound
 * /tmp/test_commit_bound
 * Every RPC below originates in a library callback; no Raft state is patched.
 * Original library: empty-first fails, repair-first passes.
 */
#include <assert.h>
#include <stdio.h>
#include <string.h>
#include "CuTest.h"
#include "raft.h"
#include "raft_private.h"

enum { VOTE, VOTE_REPLY, APPEND, APPEND_REPLY, MAX_ENTRIES = 3 };
struct packet {
    int kind, from, to;
    raft_requestvote_req_t vote;
    raft_requestvote_resp_t voted;
    raft_appendentries_req_t append;
    raft_appendentries_resp_t appended;
    raft_entry_t *entries[MAX_ENTRIES];
};
struct peer {
    int id;
    raft_server_t *server;
};
static struct peer peers[3];
static struct packet queue[128];
static int queued, hold_a, empty_once, empty_fetches;

static void enqueue(struct peer *peer, struct packet *packet)
{
    assert(queued < (int)(sizeof(queue) / sizeof(queue[0])));
    packet->from = peer->id;
    queue[queued++] = *packet;
}

static int send_vote(raft_server_t *server, void *user, raft_node_t *node,
                     raft_requestvote_req_t *request)
{
    (void)server;
    struct packet packet = { .kind = VOTE, .to = raft_node_get_id(node), .vote = *request };
    enqueue(user, &packet);
    return 0;
}

static int send_append(raft_server_t *server, void *user, raft_node_t *node,
                       raft_appendentries_req_t *request)
{
    (void)server;
    struct packet packet = { .kind = APPEND, .to = raft_node_get_id(node), .append = *request };
    assert(request->n_entries >= 0 && request->n_entries <= MAX_ENTRIES);
    packet.append.entries = NULL;
    for (int i = 0; i < request->n_entries; i++) {
        packet.entries[i] = request->entries[i];
        raft_entry_hold(packet.entries[i]);
    }
    enqueue(user, &packet);
    return 0;
}

static void release_packet(struct packet *packet)
{
    if (packet->kind == APPEND)
        for (int i = 0; i < packet->append.n_entries; i++)
            raft_entry_release(packet->entries[i]);
}

static void receive(struct packet packet)
{
    struct peer *peer = &peers[packet.to];
    raft_node_t *sender = raft_get_node(peer->server, packet.from);
    struct packet response = { .to = packet.from };
    switch (packet.kind) {
    case VOTE:
        response.kind = VOTE_REPLY;
        assert(!raft_recv_requestvote(peer->server, sender, &packet.vote, &response.voted));
        enqueue(peer, &response);
        break;
    case VOTE_REPLY:
        assert(!raft_recv_requestvote_response(peer->server, sender, &packet.voted));
        break;
    case APPEND:
        packet.append.entries = packet.entries;
        response.kind = APPEND_REPLY;
        assert(!raft_recv_appendentries(peer->server, sender, &packet.append, &response.appended));
        enqueue(peer, &response);
        break;
    case APPEND_REPLY:
        assert(!raft_recv_appendentries_response(peer->server, sender, &packet.appended));
        break;
    default:
        assert(0);
    }
    release_packet(&packet);
    assert(!raft_apply_all(peer->server));
}

static struct packet take(int from, int to, int kind)
{
    for (int i = 0; i < queued; i++) {
        if (queue[i].from != from || queue[i].to != to || queue[i].kind != kind) continue;
        struct packet packet = queue[i];
        memmove(queue + i, queue + i + 1, (size_t)(--queued - i) * sizeof(queue[0]));
        return packet;
    }
    assert(!"expected library-generated RPC missing");
    return (struct packet){0};
}

static void deliver(int from, int to, int kind)
{
    receive(take(from, to, kind));
}

static int backpressure(raft_server_t *server, void *user, raft_node_t *node)
{
    (void)server;
    (void)user;
    return hold_a && raft_node_get_id(node) == 0;
}

static raft_index_t fetch(raft_server_t *server, void *user, raft_node_t *node,
                          raft_index_t index, raft_index_t count, raft_entry_t **entries)
{
    (void)user;
    if (raft_node_get_id(node) == 0 && empty_once) {
        empty_once = 0;
        empty_fetches++;
        return 0;
    }
    raft_index_t n = 0;
    while (n < count && n < MAX_ENTRIES) {
        raft_entry_t *entry = raft_get_entry_from_idx(server, index + n);
        if (!entry) break;
        entries[n++] = entry; /* Sender releases these getter references. */
    }
    return n;
}

static void submit(const char *command, int id)
{
    raft_entry_t *entry = raft_entry_new(strlen(command) + 1);
    assert(entry);
    entry->id = id;
    entry->type = RAFT_LOGTYPE_NORMAL;
    memcpy(entry->data, command, strlen(command) + 1);
    raft_entry_resp_t response = {0};
    assert(!raft_recv_entry(peers[0].server, entry, &response));
    raft_entry_release(entry);
}

static void run_commit_bound(CuTest *tc, int repair_first)
{
    queued = empty_once = empty_fetches = 0;
    hold_a = 1;
    raft_cbs_t callbacks = { .send_requestvote = send_vote, .send_appendentries = send_append };
    for (int i = 0; i < 3; i++) {
        peers[i] = (struct peer){ .id = i, .server = raft_new() };
        assert(peers[i].server);
        raft_set_callbacks(peers[i].server, &callbacks, &peers[i]);
        assert(!raft_config(peers[i].server, 1, RAFT_CONFIG_ELECTION_TIMEOUT, 1000));
        for (int j = 0; j < 3; j++)
            assert(raft_add_node(peers[i].server, NULL, j, j == i));
    }
    raft_server_t *a = peers[0].server, *b = peers[1].server;
    /* A wins term 1; all nodes receive and commit the initial no-op at index 1. */
    assert(!raft_election_start(a, 0));
    deliver(0, 1, VOTE); deliver(1, 0, VOTE_REPLY);
    deliver(0, 1, VOTE); deliver(1, 0, VOTE_REPLY);
    int drained = 0;
    while (queued) {
        assert(++drained <= 100);
        struct packet packet = queue[0];
        memmove(queue, queue + 1, (size_t)--queued * sizeof(queue[0]));
        receive(packet);
    }
    assert(raft_is_leader(a) && raft_get_current_term(a) == 1);
    assert(raft_get_commit_idx(a) == 1);
    /* Delay A's term-1 writes at indices 2 and 3 in the transport queue. */
    submit("set alpha 17", 1);
    submit("set beta 29", 2);
    assert(raft_get_current_idx(a) == 3 && raft_get_commit_idx(a) == 1);
    callbacks.backpressure = backpressure;
    callbacks.get_entries_to_send = fetch;
    raft_set_callbacks(b, &callbacks, &peers[1]);
    assert(!raft_periodic_internal(b, 4001));
    assert(!raft_periodic_internal(peers[2].server, 4001));
    deliver(1, 2, VOTE); deliver(2, 1, VOTE_REPLY);
    deliver(1, 2, VOTE); deliver(2, 1, VOTE_REPLY);
    assert(raft_is_leader(b) && raft_get_current_term(b) == 2);
    deliver(1, 2, APPEND); deliver(2, 1, APPEND_REPLY);
    assert(raft_get_commit_idx(b) == 2);
    assert(raft_node_get_next_idx(raft_get_node(b, 0)) == 2);

    /* One empty fetch generates a heartbeat, followed by the real repair RPC. */
    hold_a = 0;
    empty_once = 1;
    assert(!raft_send_appendentries(b, raft_get_node(b, 0)));
    struct packet empty = take(1, 0, APPEND), repair = take(1, 0, APPEND);
    assert(empty_fetches == 1 && empty.append.n_entries == 0);
    assert(empty.append.prev_log_idx == 1 && empty.append.prev_log_term == 1);
    assert(empty.append.term == 2 && empty.append.leader_commit == 2);
    assert(repair.append.n_entries == 1 && repair.entries[0]->term == 2);
    if (repair_first) receive(repair);
    receive(empty);

    raft_index_t commit_a = raft_get_commit_idx(a), commit_b = raft_get_commit_idx(b);
    raft_entry_t *entry_a = raft_get_entry_from_idx(a, 2), *entry_b = raft_get_entry_from_idx(b, 2);
    assert(entry_a && entry_b);
    int conflict = commit_a >= 2 && commit_b >= 2 &&
        (entry_a->term != entry_b->term || entry_a->type != entry_b->type ||
         entry_a->data_len != entry_b->data_len ||
         memcmp(entry_a->data, entry_b->data, entry_a->data_len));
    printf("%s: commits=%ld/%ld index2 terms=%ld/%ld types=%d/%d conflict=%d\n",
           repair_first ? "repair-first" : "empty-first", (long)commit_a, (long)commit_b,
           (long)entry_a->term, (long)entry_b->term, entry_a->type, entry_b->type, conflict);
    raft_entry_release(entry_a);
    raft_entry_release(entry_b);
    if (!repair_first) release_packet(&repair);
    for (int i = 0; i < queued; i++) release_packet(&queue[i]);
    for (int i = 0; i < 3; i++) raft_destroy(peers[i].server);
    queued = 0;

    CuAssert(tc, "committed entries at index 2 must agree", !conflict);
    CuAssertIntEquals_Msg(tc, "empty RPC must not commit an unmatched local suffix",
                          repair_first ? 2 : 1, commit_a);
}

static void TestRaft_empty_append_must_not_commit_unmatched_suffix(CuTest *tc)
{
    run_commit_bound(tc, 0);
}

static void TestRaft_repair_before_empty_append_control(CuTest *tc)
{
    run_commit_bound(tc, 1);
}

int main(void)
{
    CuSuite *suite = CuSuiteNew();
    CuString *output = CuStringNew();
    SUITE_ADD_TEST(suite, TestRaft_empty_append_must_not_commit_unmatched_suffix);
    SUITE_ADD_TEST(suite, TestRaft_repair_before_empty_append_control);
    CuSuiteRun(suite);
    CuSuiteDetails(suite, output);
    printf("%s", output->buffer);
    int result = suite->failCount ? 1 : 0;
    CuStringFree(output);
    CuSuiteFree(suite);
    return result;
}