1/* fleet.rs's catch-up tests (review finding 1) over a network delivered by hand, and review finding
2 * 4: nothing leaves without its persist. */
3#include "mesh/check.h"
4#include "mesh/fleet_core.h"
5#include "mesh/mt.h"
6
7#include <stdlib.h>
8#include <string.h>
9
10typedef struct {
11    uint32_t from, to;
12    mfc_msg m;
13} qitem;
14
15typedef struct {
16    mfc_core *c[5];
17    uint32_t n;
18    qitem *q;
19    size_t qh, qn, qcap;
20    bool down[5];
21    uint64_t now;
22    mfc_id high[5];
23} net_t;
24
25static void enqueue(net_t *n, uint32_t from, uint32_t to, mfc_msg m) {
26    if (n->qn == n->qcap) {
27        n->qcap = n->qcap ? n->qcap * 2 : 64;
28        n->q = mt_realloc(n->q, n->qcap * sizeof *n->q);
29    }
30    n->q[n->qn++] = (qitem){from, to, m};
31}
32
33static void out(net_t *n, uint32_t k, mfc_outs o) {
34    for (size_t i = 0; i < o.n; i++)
35        if (o.v[i].kind == MFC_OUT_SEND) {
36            enqueue(n, k, o.v[i].to, o.v[i].msg);
37            o.v[i].kind = MFC_OUT_LOG; /* moved */
38        }
39    mfc_outs_free(&o);
40    mfc_id id = mfc_doc_id(mfc_committed(n->c[k]));
41    CHECK(mfc_id_cmp(id, n->high[k]) >= 0);
42    n->high[k] = id;
43}
44
45static void net_init(net_t *n, uint32_t count) {
46    memset(n, 0, sizeof *n);
47    n->n = count;
48    n->now = 1;
49    for (uint32_t k = 0; k < count; k++) n->c[k] = mfc_new((mfc_config){k, count, 3000, 7 + k}, (mfp_persist){0}, 1);
50    for (uint32_t a = 0; a < count; a++)
51        for (uint32_t b = 0; b < count; b++)
52            if (a != b) out(n, a, mfc_link(n->c[a], b, true, 1));
53}
54
55static void net_free(net_t *n) {
56    for (uint32_t k = 0; k < n->n; k++) mfc_free(n->c[k]);
57    for (size_t i = n->qh; i < n->qn; i++) mfc_msg_free(&n->q[i].m);
58    free(n->q);
59}
60
61typedef bool (*keep_fn)(uint32_t from, uint32_t to, const mfc_msg *m);
62
63static bool all_keep(uint32_t f, uint32_t t, const mfc_msg *m) { return (void)f, (void)t, (void)m, true; }
64static bool none_keep(uint32_t f, uint32_t t, const mfc_msg *m) { return (void)f, (void)t, (void)m, false; }
65
66/* Delivers everything queued; keep false drops it. */
67static void run(net_t *n, keep_fn keep) {
68    while (n->qh < n->qn) {
69        qitem it = n->q[n->qh++];
70        if (n->down[it.from] || n->down[it.to] || !keep(it.from, it.to, &it.m)) {
71            mfc_msg_free(&it.m);
72            continue;
73        }
74        out(n, it.to, mfc_recv(n->c[it.to], it.from, it.m, n->now));
75    }
76}
77
78static void crash(net_t *n, uint32_t k) {
79    n->down[k] = true;
80    for (uint32_t j = 0; j < n->n; j++)
81        if (j != k && !n->down[j]) out(n, j, mfc_link(n->c[j], k, false, n->now));
82}
83
84static void elect(net_t *n, uint32_t k) {
85    n->now += 5000;
86    out(n, k, mfc_tick(n->c[k], n->now));
87    run(n, all_keep);
88    CHECK(mfc_role_of(n->c[k]) == MFC_LEADER);
89}
90
91static void beat(net_t *n, uint32_t k, uint32_t to) { enqueue(n, k, to, mfc_beat(n->c[k])); }
92
93static uint64_t propose(net_t *n, uint32_t k, const char *tag) {
94    mfd_doc d = mfd_doc_clone(mfc_committed(n->c[k]));
95    mt_strs w = {0};
96    mt_push(&w, "local");
97    mfd_place_put(&d, tag, w);
98    uint64_t v = 0;
99    mfc_outs o;
100    char *err = NULL;
101    CHECK(mfc_propose(n->c[k], d, n->now, &v, &o, &err));
102    out(n, k, o);
103    return v;
104}
105
106/* peer.rs `lead`: the term's first document (the accepted one re-stamped). */
107static uint64_t first(net_t *n, uint32_t k) {
108    CHECK(!mfc_term_committed(n->c[k]));
109    mfd_doc d = {0};
110    mfc_first_doc(n->c[k], &d);
111    uint64_t v = 0;
112    mfc_outs o;
113    char *err = NULL;
114    CHECK(mfc_propose(n->c[k], d, n->now, &v, &o, &err));
115    out(n, k, o);
116    return v;
117}
118
119static bool not_commit_0_to_1(uint32_t f, uint32_t t, const mfc_msg *m) { return !(f == 0 && t == 1 && m->kind == MFC_COMMIT); }
120static bool not_to_2(uint32_t f, uint32_t t, const mfc_msg *m) { return (void)f, (void)m, t != 2; }
121static bool n5_filter(uint32_t f, uint32_t t, const mfc_msg *m) {
122    return !(f == 0 && (t == 3 || t == 4) && m->kind == MFC_STATE) && !(f == 0 && t != 2 && m->kind == MFC_COMMIT);
123}
124
125/* N = 3: a new leader whose committed document is behind must not catch a follower "up" to it. */
126/* fleet.rs catch_up_never_regresses_a_committed_document */
127static void never_regresses(void) {
128    net_t n;
129    net_init(&n, 3);
130    elect(&n, 0);
131    first(&n, 0);
132    run(&n, all_keep);
133    propose(&n, 0, "a");
134    run(&n, all_keep);
135    uint64_t v2 = propose(&n, 0, "x");
136    run(&n, not_commit_0_to_1);
137    CHECK(mfc_committed(n.c[0])->version == v2 && mfc_committed(n.c[2])->version == v2 && mfc_committed(n.c[1])->version == v2 - 1);
138    crash(&n, 0);
139    elect(&n, 1);
140    beat(&n, 2, 1);
141    run(&n, all_keep);
142    beat(&n, 1, 2);
143    run(&n, all_keep);
144    CHECK_U(mfc_committed(n.c[2])->version, v2);
145    CHECK(mfd_place_get(mfc_committed(n.c[2]), "x") != NULL);
146    uint64_t v = first(&n, 1);
147    run(&n, all_keep);
148    for (uint32_t k = 1; k < 3; k++) CHECK(mfc_committed(n.c[k])->version == v && mfd_place_get(mfc_committed(n.c[k]), "x"));
149    net_free(&n);
150}
151
152/* A follower that missed a commit while away catches up even past an abandoned proposal. */
153/* fleet.rs a_follower_catches_up_past_an_abandoned_proposal */
154static void past_abandoned(void) {
155    net_t n;
156    net_init(&n, 3);
157    elect(&n, 0);
158    first(&n, 0);
159    run(&n, all_keep);
160    uint64_t v2 = propose(&n, 0, "a");
161    run(&n, not_to_2);
162    CHECK(mfc_committed(n.c[0])->version == v2 && mfc_committed(n.c[2])->version == v2 - 1);
163    propose(&n, 0, "b");
164    run(&n, none_keep);
165    n.now += 4000;
166    for (uint32_t k = 1; k < 3; k++) out(&n, 0, mfc_link(n.c[0], k, true, n.now));
167    out(&n, 0, mfc_tick(n.c[0], n.now));
168    CHECK(!mfc_busy(n.c[0]) && mfc_accepted(n.c[0])->version == v2 + 1 && mfc_committed(n.c[0])->version == v2);
169    for (int i = 0; i < 3; i++) {
170        n.now += 1100;
171        beat(&n, 2, 0);
172        run(&n, all_keep);
173        beat(&n, 0, 2);
174        run(&n, all_keep);
175    }
176    CHECK_U(mfc_committed(n.c[2])->version, v2);
177    net_free(&n);
178}
179
180/* N = 5: version 2 (x) committed on {0, 1, 2}; two leader changes lose nothing. */
181/* fleet.rs a_committed_version_survives_two_leader_changes_n5 */
182static void survives_two_changes(void) {
183    net_t n;
184    net_init(&n, 5);
185    elect(&n, 0);
186    first(&n, 0);
187    run(&n, all_keep);
188    propose(&n, 0, "a");
189    run(&n, all_keep);
190    uint64_t v2 = propose(&n, 0, "x");
191    run(&n, n5_filter);
192    CHECK(mfd_place_get(mfc_committed(n.c[0]), "x") && mfc_committed(n.c[2])->version == v2);
193    crash(&n, 0);
194    elect(&n, 1);
195    beat(&n, 2, 1);
196    run(&n, all_keep);
197    beat(&n, 1, 2);
198    run(&n, all_keep);
199    crash(&n, 1);
200    elect(&n, 2);
201    uint64_t v = first(&n, 2);
202    run(&n, all_keep);
203    for (uint32_t k = 2; k < 5; k++) {
204        const mfd_doc *d = mfc_committed(n.c[k]);
205        CHECK(d->version == v && mfd_place_get(d, "x") && mfd_place_get(d, "a"));
206    }
207    net_free(&n);
208}
209
210static bool fail_persist(void *ctx, char **err) {
211    *err = mt_strdup((const char *)ctx);
212    return false;
213}
214
215static bool count_persist(void *ctx, char **err) {
216    (void)err;
217    (*(int *)ctx)++;
218    return true;
219}
220
221static bool has_send(const mfc_outs *o, mfc_msg_kind kind, int to, int granted) {
222    for (size_t i = 0; i < o->n; i++)
223        if (o->v[i].kind == MFC_OUT_SEND && o->v[i].msg.kind == kind && (to < 0 || o->v[i].to == (uint32_t)to) &&
224            (granted < 0 || o->v[i].msg.granted == (bool)granted))
225            return true;
226    return false;
227}
228
229/* Review finding 4: a vote (or an acknowledgement) whose persist failed is never sent. */
230/* fleet.rs nothing_leaves_without_its_persist */
231static void nothing_leaves_without_persist(void) {
232    mfc_core *c = mfc_new((mfc_config){1, 3, 3000, 3}, (mfp_persist){0}, 1);
233    mfc_outs o = mfc_link(c, 0, true, 1);
234    mfc_outs_free(&o);
235    o = mfc_link(c, 2, true, 1);
236    mfc_outs_free(&o);
237    o = mfc_recv(c, 0, (mfc_msg){.kind = MFC_VOTE_REQ, .pre = false, .term = 5}, 10000);
238    CHECK(has_send(&o, MFC_VOTE, 0, 1));
239    char *err = NULL;
240    CHECK(!mfc_durable(&o, fail_persist, "ENOSPC", &err));
241    free(err);
242    CHECK(!has_send(&o, MFC_VOTE, -1, -1));
243    mfc_outs_free(&o);
244    mfc_free(c);
245    /* The disk holds what it held before the vote: no vote in term 5. */
246    c = mfc_new((mfc_config){1, 3, 3000, 3}, (mfp_persist){0}, 1);
247    o = mfc_link(c, 0, true, 1);
248    mfc_outs_free(&o);
249    o = mfc_link(c, 2, true, 1);
250    mfc_outs_free(&o);
251    o = mfc_recv(c, 2, (mfc_msg){.kind = MFC_VOTE_REQ, .pre = false, .term = 5}, 10000);
252    CHECK(has_send(&o, MFC_VOTE, 2, 1));
253    mfc_outs_free(&o);
254    /* An accepted document's acknowledgement likewise. */
255    mfd_doc d = {.version = 1, .term = 5};
256    mfd_place_put(&d, "x", (mt_strs){0});
257    o = mfc_recv(c, 2, (mfc_msg){.kind = MFC_STATE, .term = 5, .doc = d}, 10001);
258    CHECK(!mfc_durable(&o, fail_persist, "EIO", &err));
259    free(err);
260    CHECK(!has_send(&o, MFC_STATE_ACK, -1, -1));
261    mfc_outs_free(&o);
262    /* A disk that works: the persist precedes the sends, which all come out. */
263    o = mfc_recv(c, 2, (mfc_msg){.kind = MFC_COMMIT, .term = 5, .ver = {5, 1}}, 10002);
264    size_t want = 0, persists = 0;
265    for (size_t i = 0; i < o.n; i++) {
266        if (o.v[i].kind == MFC_OUT_PERSIST) persists++;
267        else want++;
268    }
269    int wrote = 0;
270    CHECK(mfc_durable(&o, count_persist, &wrote, &err));
271    CHECK(o.n == want && (size_t)wrote == persists && persists > 0);
272    mfc_outs_free(&o);
273    mfc_free(c);
274}
275
276int main(void) {
277    never_regresses();
278    past_abandoned();
279    survives_two_changes();
280    nothing_leaves_without_persist();
281    return mesh_check_done();
282}