1/* @ajax/mesh-jobs-log: what a member persists (jobs.rs `Log`, `Entry`, `Replica`, `replay`), in an
2 * append-only log of records:
3 * - LID: the highest leadership it followed or led (a restarted member never follows an older one
4 *   again, and leads above it);
5 * - SNAPSHOT: the committed table at commit point (lid, n), which replaces what was before (a
6 *   follower's entries of that leadership past n are kept);
7 * - ENTRY: an entry (lid, n) received or made, not committed yet: a delta, or a replace (a new
8 *   leader's merge, which drops other leaderships' entries);
9 * - COMMIT: the entries of lid up to n are committed.
10 *
11 * Replaying the log rebuilds the member's replica exactly: entries that were not committed stay
12 * pending, never in the committed table (review finding 5). A torn tail (a crash mid-append, never
13 * acknowledged: review finding 4) ends the replay; a whole record that doesn't decode is corruption. */
14#ifndef MESH_JOBS_LOG_H
15#define MESH_JOBS_LOG_H
16
17#include "mesh/jobs/op.h"
18#include "mesh/jobs/table.h"
19#include "mesh/strbuf.h"
20
21#include <stdbool.h>
22#include <stddef.h>
23#include <stdint.h>
24
25typedef enum { MJ_LOG_LID = 1, MJ_LOG_SNAPSHOT = 2, MJ_LOG_ENTRY = 3, MJ_LOG_COMMIT = 4 } mj_log_kind;
26
27typedef struct {
28    mj_log_kind kind;
29    uint64_t lid;
30    uint64_t n;   /* SNAPSHOT, ENTRY, COMMIT */
31    bool replace; /* ENTRY */
32    mj_ops ops;   /* SNAPSHOT, ENTRY */
33} mj_log;
34
35void mj_log_free(mj_log *l);
36mj_log mj_log_clone(const mj_log *l);
37static inline mj_log mj_log_lid(uint64_t lid) { return (mj_log){.kind = MJ_LOG_LID, .lid = lid}; }
38static inline mj_log mj_log_commit(uint64_t lid, uint64_t n) { return (mj_log){.kind = MJ_LOG_COMMIT, .lid = lid, .n = n}; }
39/* These take the ops. */
40static inline mj_log mj_log_snapshot(uint64_t lid, uint64_t n, mj_ops ops) { return (mj_log){.kind = MJ_LOG_SNAPSHOT, .lid = lid, .n = n, .ops = ops}; }
41static inline mj_log mj_log_entry(uint64_t lid, uint64_t n, bool replace, mj_ops ops) {
42    return (mj_log){.kind = MJ_LOG_ENTRY, .lid = lid, .n = n, .replace = replace, .ops = ops};
43}
44
45/* A record: u8 kind, u64 length, the body. Appended to out. */
46void mj_log_encode(const mj_log *l, msb *out);
47bool mj_log_decode(uint8_t kind, const void *body, size_t n, mj_log *out, char **err);
48
49/* A received or made entry, not committed yet. */
50typedef struct {
51    uint64_t lid; /* its leadership */
52    uint64_t n;
53    bool replace;
54    mj_ops ops;      /* a delta's */
55    mj_table table;  /* a replace's (initialised only when replace) */
56} mj_entry;
57
58void mj_entry_free(mj_entry *e);
59mj_entry mj_entry_clone(const mj_entry *e);
60bool mj_entry_eq(const mj_entry *a, const mj_entry *b);
61
62/* A member's durable job state: what its log replays to. */
63typedef struct {
64    uint64_t lid;                   /* the highest leadership followed or led */
65    uint64_t commit_lid, commit_n;  /* the committed table's commit point */
66    mj_table committed;
67    mj_entry *pending;              /* entries not committed yet, in order */
68    size_t npending, cappending;
69} mj_replica;
70
71void mj_replica_init(mj_replica *r);
72void mj_replica_free(mj_replica *r);
73mj_replica mj_replica_clone(const mj_replica *r);
74bool mj_replica_eq(const mj_replica *a, const mj_replica *b);
75void mj_replica_apply(mj_replica *r, const mj_log *l);
76/* The last entry pending of leadership lid (0: none). */
77uint64_t mj_replica_last(const mj_replica *r, uint64_t lid);
78
79typedef struct {
80    mj_log *v;
81    size_t n, cap;
82} mj_logs;
83void mj_logs_push(mj_logs *v, mj_log l);
84void mj_logs_free(mj_logs *v);
85/* The log that replays to r (a compaction). */
86mj_logs mj_replica_log(const mj_replica *r);
87
88/* Replays a log file: the replica, the number of records and the bytes of whole records. A torn
89 * tail ends it there; a whole record that does not decode is an error (*err: "<why> (record n at
90 * byte b)"). */
91bool mj_replay(const void *bytes, size_t n, mj_replica *out, size_t *records, size_t *whole, char **err);
92
93#endif