1/* fleet.rs Core: the election and commit state machine (see the header). */
2#include "mesh/fleet_core.h"
3#include "mesh/member_list.h"
4#include "mesh/mt.h"
5
6#include <stdlib.h>
7#include <string.h>
8
9mfc_msg mfc_msg_clone(const mfc_msg *m) {
10    mfc_msg c = *m;
11    if (m->kind == MFC_STATE) c.doc = mfd_doc_clone(&m->doc);
12    return c;
13}
14
15void mfc_msg_free(mfc_msg *m) {
16    if (m->kind == MFC_STATE) mfd_doc_free(&m->doc);
17}
18
19const char *mfc_role_word(mfc_role r) {
20    static const char *const w[] = {"follower", "pre-candidate", "candidate", "leader"};
21    return w[r];
22}
23
24uint32_t mfc_role_code(mfc_role r) { return (uint32_t)r + 1; }
25
26bool mfc_fixed(const mfc_config *c) { return c->n == 2; }
27uint32_t mfc_quorum(const mfc_config *c) { return mfc_fixed(c) ? 1 : c->n / 2 + 1; }
28
29static void out_free(mfc_out *o) {
30    if (o->kind == MFC_OUT_SEND) mfc_msg_free(&o->msg);
31    free(o->text);
32}
33
34void mfc_outs_free(mfc_outs *o) {
35    for (size_t i = 0; i < o->n; i++) out_free(&o->v[i]);
36    free(o->v);
37    *o = (mfc_outs){0};
38}
39
40static void push(mfc_outs *o, mfc_out x) {
41    if (o->n == o->cap) {
42        o->cap = o->cap ? o->cap * 2 : 8;
43        o->v = mt_realloc(o->v, o->cap * sizeof *o->v);
44    }
45    o->v[o->n++] = x;
46}
47
48bool mfc_durable(mfc_outs *outs, mfc_persist_fn persist, void *ctx, char **err) {
49    size_t j = 0;
50    for (size_t i = 0; i < outs->n; i++) {
51        if (outs->v[i].kind != MFC_OUT_PERSIST) {
52            outs->v[j++] = outs->v[i];
53            continue;
54        }
55        if (!persist(ctx, err)) {
56            for (size_t k = i + 1; k < outs->n; k++) out_free(&outs->v[k]);
57            outs->n = j;
58            return false;
59        }
60    }
61    outs->n = j;
62    return true;
63}
64
65typedef struct {
66    bool on;
67    mfd_doc doc;
68    uint32_t acks;
69    uint64_t deadline;
70} inflight;
71
72struct mfc_core {
73    mfc_config cfg;
74    mfp_persist p;
75    mfc_role role;
76    bool has_leader;
77    uint32_t leader;
78    uint32_t linked;
79    uint64_t last_heard[MESH_MAX_SYSTEMS];
80    uint64_t leader_contact;
81    uint64_t election_at;
82    uint32_t votes;
83    uint64_t rng;
84    inflight inflight;
85    uint64_t caught_up_at[MESH_MAX_SYSTEMS]; /* catch-up sends per member, at most one a second */
86};
87
88static uint64_t rnd(mfc_core *c, uint64_t n) {
89    c->rng ^= c->rng << 13;
90    c->rng ^= c->rng >> 7;
91    c->rng ^= c->rng << 17;
92    return n == 0 ? 0 : c->rng % n;
93}
94
95/* 150 ms x id + 0..100 ms. */
96static uint64_t stagger(mfc_core *c) { return 150 * (uint64_t)c->cfg.me + rnd(c, 100); }
97/* The election timeout after the last leader contact. */
98static uint64_t timeout(mfc_core *c) { return c->cfg.dead_ms + stagger(c); }
99
100mfc_core *mfc_new(mfc_config cfg, mfp_persist p, uint64_t now) {
101    mfc_core *c = mt_alloc(sizeof *c);
102    c->cfg = cfg;
103    c->p = p;
104    c->role = MFC_FOLLOWER;
105    c->rng = cfg.seed | 1;
106    if (mfc_fixed(&cfg)) {
107        if (c->p.term < 1) c->p.term = 1;
108        if (cfg.me == 0) {
109            c->role = MFC_LEADER;
110            c->has_leader = true;
111            c->leader = 0;
112        }
113    }
114    /* At start nobody leads yet: a first pre-vote after a fifth of the dead time (a member that
115     * joins a working fleet hears its leader's beat sooner, and pre-votes are refused while a
116     * leader is heard, so this never disturbs one). */
117    c->election_at = now + c->cfg.dead_ms / 5 + stagger(c);
118    return c;
119}
120
121void mfc_free(mfc_core *c) {
122    if (!c) return;
123    mfp_free(&c->p);
124    if (c->inflight.on) mfd_doc_free(&c->inflight.doc);
125    free(c);
126}
127
128const mfc_config *mfc_cfg(const mfc_core *c) { return &c->cfg; }
129const mfp_persist *mfc_persist(const mfc_core *c) { return &c->p; }
130uint64_t mfc_term(const mfc_core *c) { return c->p.term; }
131mfc_role mfc_role_of(const mfc_core *c) { return c->role; }
132bool mfc_leader(const mfc_core *c, uint32_t *l) {
133    if (c->has_leader && l) *l = c->leader;
134    return c->has_leader;
135}
136const mfd_doc *mfc_committed(const mfc_core *c) { return &c->p.committed; }
137const mfd_doc *mfc_accepted(const mfc_core *c) { return &c->p.accepted; }
138bool mfc_busy(const mfc_core *c) { return c->inflight.on; }
139bool mfc_term_committed(const mfc_core *c) { return c->p.committed.version > 0 && c->p.committed.term == c->p.term; }
140uint64_t mfc_next_version(const mfc_core *c) {
141    uint64_t a = c->p.accepted.version, b = c->p.committed.version;
142    return (a > b ? a : b) + 1;
143}
144bool mfc_first_doc(const mfc_core *c, mfd_doc *out) {
145    if (c->p.accepted.version == 0) return false;
146    *out = mfd_doc_clone(&c->p.accepted);
147    return true;
148}
149uint32_t mfc_linked(const mfc_core *c) { return c->linked; }
150bool mfc_voted(const mfc_core *c, uint32_t *v) {
151    if (c->p.has_voted && v) *v = c->p.voted;
152    return c->p.has_voted;
153}
154
155uint32_t mfc_alive(const mfc_core *c, uint64_t now) {
156    uint32_t m = 0;
157    for (uint32_t k = 0; k < c->cfg.n; k++)
158        if (k == c->cfg.me || ((c->linked >> k & 1) && (now > c->last_heard[k] ? now - c->last_heard[k] : 0) < c->cfg.dead_ms)) m |= 1u << k;
159    return m;
160}
161
162bool mfc_has_quorum(const mfc_core *c, uint64_t now) { return (uint32_t)ml_popcount(mfc_alive(c, now)) >= mfc_quorum(&c->cfg); }
163
164static void send(mfc_outs *out, uint32_t k, mfc_msg m) { push(out, (mfc_out){.kind = MFC_OUT_SEND, .to = k, .msg = m}); }
165
166static void send_all(const mfc_core *c, const mfc_msg *m, mfc_outs *out) {
167    for (uint32_t k = 0; k < c->cfg.n; k++)
168        if (k != c->cfg.me && (c->linked >> k & 1)) send(out, k, mfc_msg_clone(m));
169}
170
171static void log_line(mfc_outs *out, char *text) { push(out, (mfc_out){.kind = MFC_OUT_LOG, .text = text}); }
172static void persist_out(mfc_outs *out) { push(out, (mfc_out){.kind = MFC_OUT_PERSIST}); }
173
174static void set_leader(mfc_core *c, bool has, uint32_t l, mfc_outs *out) {
175    if (c->has_leader == has && (!has || c->leader == l)) return;
176    c->has_leader = has;
177    c->leader = has ? l : 0;
178    push(out, (mfc_out){.kind = MFC_OUT_LEADER, .has_leader = has, .leader = l});
179}
180
181static void abandon(mfc_core *c, const char *why, mfc_outs *out) {
182    if (!c->inflight.on) return;
183    push(out, (mfc_out){.kind = MFC_OUT_DONE, .version = c->inflight.doc.version, .ok = false, .text = mt_strdup(why)});
184    mfd_doc_free(&c->inflight.doc);
185    c->inflight.on = false;
186}
187
188/* A higher term: follow it (no vote yet in it). */
189static void adopt(mfc_core *c, uint64_t term, uint64_t now, mfc_outs *out) {
190    if (term <= c->p.term) return;
191    if (c->role == MFC_LEADER)
192        log_line(out, mt_fmt("fleet: term %llu seen; stepping down (was leader of term %llu)", (unsigned long long)term, (unsigned long long)c->p.term));
193    c->p.term = term;
194    c->p.has_voted = false;
195    c->role = MFC_FOLLOWER;
196    abandon(c, "the leader changed (a higher term)", out);
197    set_leader(c, false, 0, out);
198    c->election_at = now + timeout(c);
199    persist_out(out);
200}
201
202mfc_outs mfc_link(mfc_core *c, uint32_t k, bool up, uint64_t now) {
203    mfc_outs out = {0};
204    if (k >= c->cfg.n || k == c->cfg.me) return out;
205    if (up) {
206        c->linked |= 1u << k;
207        c->last_heard[k] = now;
208        c->caught_up_at[k] = 0;
209        /* The pair: system 1's leader is system 0 whenever they are linked. */
210        if (mfc_fixed(&c->cfg) && k == 0) {
211            c->leader_contact = now;
212            set_leader(c, true, 0, &out);
213        }
214    } else {
215        c->linked &= ~(1u << k);
216        if (c->has_leader && c->leader == k) {
217            /* Its link closed: no leader here now, and no need to wait out dead_ms. */
218            set_leader(c, false, 0, &out);
219            if (!mfc_fixed(&c->cfg)) {
220                uint64_t t = now + stagger(c);
221                if (t < c->election_at) c->election_at = t;
222            }
223        }
224    }
225    return out;
226}
227
228mfc_outs mfc_tick(mfc_core *c, uint64_t now) {
229    mfc_outs out = {0};
230    if (c->inflight.on && now >= c->inflight.deadline) {
231        char *why = mt_fmt("not committed: %d of %u members acknowledged within %llu ms (a majority is %u)", ml_popcount(c->inflight.acks), c->cfg.n,
232                           (unsigned long long)c->cfg.dead_ms, mfc_quorum(&c->cfg));
233        abandon(c, why, &out);
234        free(why);
235    }
236    if (mfc_fixed(&c->cfg)) return out;
237    if (c->role == MFC_LEADER) {
238        if (!mfc_has_quorum(c, now)) {
239            log_line(&out, mt_fmt("fleet: leader of term %llu hears only %d of %u members (a majority is %u): stepping down; no changes until a majority is back",
240                                  (unsigned long long)c->p.term, ml_popcount(mfc_alive(c, now)), c->cfg.n, mfc_quorum(&c->cfg)));
241            c->role = MFC_FOLLOWER;
242            abandon(c, "the leader lost its majority", &out);
243            set_leader(c, false, 0, &out);
244            c->election_at = now + timeout(c);
245        }
246    } else if (now >= c->election_at) {
247        /* Pre-vote first: no term change unless a majority would elect us. */
248        c->role = MFC_PRE_CANDIDATE;
249        c->votes = 1u << c->cfg.me;
250        if (c->has_leader) set_leader(c, false, 0, &out);
251        c->election_at = now + c->cfg.dead_ms / 2 + stagger(c);
252        mfc_msg m = {.kind = MFC_VOTE_REQ, .pre = true, .term = c->p.term + 1, .acc = mfc_doc_id(&c->p.accepted)};
253        send_all(c, &m, &out);
254    }
255    return out;
256}
257
258static bool up_to_date(const mfc_core *c, mfc_id acc) { return mfc_id_cmp(acc, mfc_doc_id(&c->p.accepted)) >= 0; }
259
260static bool hears_leader(const mfc_core *c, uint64_t now) {
261    if (!c->has_leader) return false;
262    if (c->leader == c->cfg.me) return true;
263    return (c->linked >> c->leader & 1) && (now > c->leader_contact ? now - c->leader_contact : 0) < c->cfg.dead_ms;
264}
265
266/* The leader's commit point reached us: commit the accepted document if it is that one. Only
267 * forward: a commit point at or below ours is old news. */
268static void follow_commit(mfc_core *c, mfc_id commit, mfc_outs *out) {
269    if (mfc_id_cmp(commit, mfc_doc_id(&c->p.committed)) > 0 && mfc_id_cmp(commit, mfc_doc_id(&c->p.accepted)) == 0 && commit.version > 0) {
270        mfd_doc_free(&c->p.committed);
271        c->p.committed = mfd_doc_clone(&c->p.accepted);
272        persist_out(out);
273        push(out, (mfc_out){.kind = MFC_OUT_COMMITTED});
274    }
275}
276
277/* A follower whose beat shows it behind gets the leader's accepted document (at most once a
278 * second), or -- after an abandoned proposal of this term left accepted ahead of committed -- the
279 * committed one (fleet.rs catch_up says why). */
280static void catch_up(mfc_core *c, uint32_t k, mfc_id commit, mfc_id acc, uint64_t now, mfc_outs *out) {
281    bool stuck = !c->inflight.on && mfc_term_committed(c) && mfc_id_cmp(mfc_doc_id(&c->p.accepted), mfc_doc_id(&c->p.committed)) != 0 &&
282                 mfc_id_cmp(commit, mfc_doc_id(&c->p.committed)) < 0;
283    const mfd_doc *want = stuck ? &c->p.committed : &c->p.accepted;
284    if (mfc_id_cmp(acc, mfc_doc_id(want)) == 0 || want->version == 0) return;
285    uint64_t last = c->caught_up_at[k];
286    if (last != 0 && (now > last ? now - last : 0) < 1000) return;
287    c->caught_up_at[k] = now;
288    send(out, k, (mfc_msg){.kind = MFC_STATE, .term = c->p.term, .doc = mfd_doc_clone(want)});
289}
290
291static void commit_inflight(mfc_core *c, mfc_outs *out) {
292    mfc_id ver = mfc_doc_id(&c->inflight.doc);
293    mfd_doc_free(&c->p.committed);
294    c->p.committed = c->inflight.doc;
295    c->inflight.on = false;
296    c->inflight.doc = (mfd_doc){0};
297    persist_out(out);
298    push(out, (mfc_out){.kind = MFC_OUT_COMMITTED});
299    mfc_msg m = {.kind = MFC_COMMIT, .term = c->p.term, .ver = ver};
300    send_all(c, &m, out);
301    push(out, (mfc_out){.kind = MFC_OUT_DONE, .version = ver.version, .ok = true});
302}
303
304mfc_msg mfc_beat(const mfc_core *c) {
305    return (mfc_msg){.kind = MFC_BEAT, .term = c->p.term, .leader = c->role == MFC_LEADER, .commit = mfc_doc_id(&c->p.committed), .acc = mfc_doc_id(&c->p.accepted)};
306}
307
308mfc_outs mfc_recv(mfc_core *c, uint32_t from, mfc_msg m, uint64_t now) {
309    mfc_outs out = {0};
310    if (from >= c->cfg.n || from == c->cfg.me) {
311        mfc_msg_free(&m);
312        return out;
313    }
314    c->last_heard[from] = now;
315    bool fixed = mfc_fixed(&c->cfg);
316    switch (m.kind) {
317    case MFC_BEAT:
318        if (fixed) {
319            if (from == 0 && m.leader) {
320                c->leader_contact = now;
321                set_leader(c, true, 0, &out);
322                follow_commit(c, m.commit, &out);
323            }
324            if (c->role == MFC_LEADER) catch_up(c, from, m.commit, m.acc, now, &out);
325            break;
326        }
327        adopt(c, m.term, now, &out);
328        if (m.term < c->p.term) break;
329        if (m.leader) {
330            if (c->role == MFC_LEADER) break; /* same term, two leaders: impossible (one vote each) */
331            c->role = MFC_FOLLOWER;
332            c->leader_contact = now;
333            c->election_at = now + timeout(c);
334            set_leader(c, true, from, &out);
335            follow_commit(c, m.commit, &out);
336        } else if (c->role == MFC_LEADER) {
337            catch_up(c, from, m.commit, m.acc, now, &out);
338        }
339        break;
340    case MFC_VOTE_REQ:
341        if (fixed) break;
342        if (m.pre) {
343            bool granted = m.term > c->p.term && up_to_date(c, m.acc) && !hears_leader(c, now);
344            send(&out, from, (mfc_msg){.kind = MFC_VOTE, .pre = true, .term = m.term, .granted = granted});
345        } else {
346            /* A real election: only after a pre-vote majority, so a higher term here is legitimate. */
347            adopt(c, m.term, now, &out);
348            bool granted = m.term == c->p.term && c->role != MFC_LEADER && (!c->p.has_voted || c->p.voted == from) && up_to_date(c, m.acc);
349            if (granted) {
350                if (!c->p.has_voted) {
351                    c->p.has_voted = true;
352                    c->p.voted = from;
353                    persist_out(&out);
354                }
355                c->election_at = now + timeout(c);
356            }
357            send(&out, from, (mfc_msg){.kind = MFC_VOTE, .pre = false, .term = c->p.term, .granted = granted});
358        }
359        break;
360    case MFC_VOTE:
361        if (m.pre) {
362            if (c->role == MFC_PRE_CANDIDATE && m.term == c->p.term + 1 && m.granted) {
363                c->votes |= 1u << from;
364                if ((uint32_t)ml_popcount(c->votes) >= mfc_quorum(&c->cfg)) {
365                    /* A real election. */
366                    c->role = MFC_CANDIDATE;
367                    c->p.term += 1;
368                    c->p.has_voted = true;
369                    c->p.voted = c->cfg.me;
370                    c->votes = 1u << c->cfg.me;
371                    c->election_at = now + c->cfg.dead_ms / 2 + stagger(c);
372                    persist_out(&out);
373                    log_line(&out, mt_fmt("fleet: election for term %llu (pre-votes from 0x%x)", (unsigned long long)c->p.term, c->votes));
374                    mfc_msg r = {.kind = MFC_VOTE_REQ, .pre = false, .term = c->p.term, .acc = mfc_doc_id(&c->p.accepted)};
375                    send_all(c, &r, &out);
376                }
377            }
378        } else {
379            adopt(c, m.term, now, &out);
380            if (c->role == MFC_CANDIDATE && m.term == c->p.term && m.granted) {
381                c->votes |= 1u << from;
382                if ((uint32_t)ml_popcount(c->votes) >= mfc_quorum(&c->cfg)) {
383                    c->role = MFC_LEADER;
384                    log_line(&out, mt_fmt("fleet: system %u leads term %llu (votes 0x%x)", c->cfg.me, (unsigned long long)c->p.term, c->votes));
385                    set_leader(c, true, c->cfg.me, &out);
386                    mfc_msg b = mfc_beat(c);
387                    send_all(c, &b, &out);
388                }
389            }
390        }
391        break;
392    case MFC_STATE: {
393        if (!fixed) adopt(c, m.term, now, &out);
394        if (m.term < c->p.term || c->role == MFC_LEADER || (fixed && from != 0)) break;
395        /* The current leader's document is accepted as is (it never sends one below its committed). */
396        c->role = MFC_FOLLOWER;
397        c->leader_contact = now;
398        c->election_at = now + timeout(c);
399        set_leader(c, true, from, &out);
400        mfc_id ver = mfc_doc_id(&m.doc);
401        mfd_doc_free(&c->p.accepted);
402        c->p.accepted = m.doc;
403        m.doc = (mfd_doc){0};
404        persist_out(&out);
405        send(&out, from, (mfc_msg){.kind = MFC_STATE_ACK, .term = c->p.term, .ver = ver});
406        break;
407    }
408    case MFC_STATE_ACK:
409        if (m.term != c->p.term || c->role != MFC_LEADER) break;
410        if (c->inflight.on && mfc_id_cmp(mfc_doc_id(&c->inflight.doc), m.ver) == 0) {
411            c->inflight.acks |= 1u << from;
412            if ((uint32_t)ml_popcount(c->inflight.acks) >= mfc_quorum(&c->cfg)) commit_inflight(c, &out);
413            break;
414        }
415        /* A catch-up of the committed document landed. */
416        if (mfc_id_cmp(m.ver, mfc_doc_id(&c->p.committed)) == 0) send(&out, from, (mfc_msg){.kind = MFC_COMMIT, .term = c->p.term, .ver = m.ver});
417        break;
418    case MFC_COMMIT:
419        if (m.term == c->p.term && c->has_leader && c->leader == from) follow_commit(c, m.ver, &out);
420        break;
421    }
422    mfc_msg_free(&m);
423    return out;
424}
425
426bool mfc_propose(mfc_core *c, mfd_doc doc, uint64_t now, uint64_t *version, mfc_outs *outs, char **err) {
427    if (c->role != MFC_LEADER) {
428        *err = c->has_leader ? mt_fmt("not the leader (system %u is)", c->leader) : mt_strdup("no leader");
429        mfd_doc_free(&doc);
430        return false;
431    }
432    if (c->inflight.on) {
433        *err = mt_strdup("busy: another change is being committed");
434        mfd_doc_free(&doc);
435        return false;
436    }
437    uint32_t alive = mfc_alive(c, now);
438    if ((uint32_t)ml_popcount(alive) < mfc_quorum(&c->cfg)) {
439        ml_list l = ml_members(alive);
440        char *t = ml_text(&l);
441        *err = mt_fmt("no quorum: %d of %u members reachable (%s), a majority is %u", ml_popcount(alive), c->cfg.n, t, mfc_quorum(&c->cfg));
442        free(t);
443        ml_free(&l);
444        mfd_doc_free(&doc);
445        return false;
446    }
447    doc.term = c->p.term;
448    doc.version = mfc_next_version(c);
449    *version = doc.version;
450    mfd_doc_free(&c->p.accepted);
451    c->p.accepted = mfd_doc_clone(&doc);
452    mfc_outs out = {0};
453    persist_out(&out);
454    c->inflight = (inflight){true, mfd_doc_clone(&doc), 1u << c->cfg.me, now + c->cfg.dead_ms};
455    mfc_msg m = {.kind = MFC_STATE, .term = c->p.term, .doc = doc};
456    send_all(c, &m, &out);
457    mfd_doc_free(&m.doc);
458    /* The pair's coordinator commits alone (its member follows when it hears it). */
459    if (mfc_quorum(&c->cfg) <= 1) commit_inflight(c, &out);
460    *outs = out;
461    return true;
462}