1/* reconcile.rs a_cycle_is_hellod_only_with_its_image_and_task; and passes against a fake driver: loads,
2 * unsupported programs, failures and the retry, a migration (load stopped, view, start), a lockstep
3 * epoch and a cycle with its task, placement on the leader. */
4#include "mesh/abi/reconcile.h"
5#include "mesh/check.h"
6#include "mesh/reconcile.h"
7#include "mesh/mt.h"
8
9#include <stdlib.h>
10
11static mfd_group group(mfd_op op) {
12    return (mfd_group){.name = "train", .program = "train", .epoch = 2, .cycle = 3, .op = op, .members = ml_of((uint32_t[]){0, 1, 2}, 3), .state = MFD_RUNNING};
13}
14
15static void cycle_ready(void) {
16    mfl_actual a = {.name = "train", .dir = "train", .hash = 0xa, .lockstep = true}, b = a;
17    b.hash = 0xb;
18    char *why = NULL;
19    mfd_group up = group((mfd_op){MFD_OP_UPGRADE, 0, 0});
20    CHECK(!mrc_cycle_ready(&up, 0xb, &a, 1, NULL, NULL, &why));
21    CHECK(mt_contains(why, "runs 000000000000000a here, the group 000000000000000b") && mt_contains(why, "paused"));
22    free(why);
23    CHECK(mrc_cycle_ready(&up, 0xb, &b, 1, NULL, NULL, &why));
24    CHECK(!mrc_cycle_ready(&up, 0xb, NULL, 0, NULL, NULL, &why) && mt_contains(why, "not loaded"));
25    free(why);
26    mfd_group t = group((mfd_op){MFD_OP_TASK, 3, 7});
27    bool no = false, yes = true;
28    CHECK(!mrc_cycle_ready(&t, 0xb, &b, 1, &no, "no result in 60s", &why) && mt_contains(why, "until the group's next epoch"));
29    free(why);
30    CHECK(!mrc_cycle_ready(&t, 0xb, &b, 1, NULL, NULL, &why) && mt_ends(why, "not run here"));
31    free(why);
32    CHECK(mrc_cycle_ready(&t, 0xb, &b, 1, &yes, NULL, &why));
33    mfd_group rb = group((mfd_op){MFD_OP_ROLLBACK, 0, 0});
34    CHECK(mrc_cycle_ready(&rb, 0xa, &a, 1, NULL, NULL, &why));
35    ml_free(&up.members), ml_free(&t.members), ml_free(&rb.members);
36}
37
38/* ---- a fake driver */
39
40typedef struct {
41    char name[16], dir[16];
42    uint64_t hash;
43    bool run;
44} prog;
45
46typedef struct {
47    prog p[16];
48    int n;
49    mt_strs calls; /* what the reconciler asked, in order */
50    uint64_t applied;
51    uint32_t last_status;
52    char *last_reason;
53    bool leader, want;
54    mpk_hello hello;
55    bool helloed;
56    uint32_t words[3];
57    const char *no_image; /* this program has no image here: unsupported */
58    const char *fail_load;
59} fake;
60
61static prog *find(fake *f, const char *name) {
62    for (int i = 0; i < f->n; i++)
63        if (!strcmp(f->p[i].name, name)) return &f->p[i];
64    return NULL;
65}
66
67static mfl_actual *inventory(void *ctx, size_t *n) {
68    fake *f = ctx;
69    mfl_actual *a = calloc((size_t)f->n + 1, sizeof *a);
70    for (int i = 0; i < f->n; i++) a[i] = mfl_actual_new(f->p[i].name, f->p[i].dir, f->p[i].hash, false, 0, f->p[i].run);
71    *n = (size_t)f->n;
72    return a;
73}
74
75typedef struct {
76    char name[16], dir[16];
77    uint64_t hash;
78} src_t;
79
80static bool source(void *ctx, const char *name, const char *dir, uint64_t hash, void **src, char **err) {
81    fake *f = ctx;
82    if (f->no_image && !strcmp(name, f->no_image)) {
83        *err = mt_fmt("unsupported: no image of %s for this member", name);
84        return false;
85    }
86    src_t *s = calloc(1, sizeof *s);
87    snprintf(s->name, sizeof s->name, "%s", name), snprintf(s->dir, sizeof s->dir, "%s", dir);
88    s->hash = hash;
89    *src = s;
90    return true;
91}
92
93static void src_free(void *ctx, void *src) { (void)ctx, free(src); }
94
95static bool load(void *ctx, void *src, bool run, char **text, char **err) {
96    fake *f = ctx;
97    src_t *s = src;
98    if (f->fail_load && !strcmp(s->name, f->fail_load)) {
99        *err = mt_strdup("jit: boom");
100        return false;
101    }
102    prog *p = &f->p[f->n++];
103    snprintf(p->name, sizeof p->name, "%s", s->name), snprintf(p->dir, sizeof p->dir, "%s", s->dir);
104    p->hash = s->hash, p->run = run;
105    mt_push_own(&f->calls, mt_fmt("load %s %s", s->name, run ? "running" : "stopped"));
106    *text = mt_strdup("slot 1");
107    return true;
108}
109
110static bool upgrade(void *ctx, const char *name, void *src, char **text, char **err) {
111    (void)err;
112    fake *f = ctx;
113    find(f, name)->hash = ((src_t *)src)->hash;
114    mt_push_own(&f->calls, mt_fmt("upgrade %s", name));
115    *text = mt_strdup("jit 1.0 ms");
116    return true;
117}
118
119static bool unload(void *ctx, const char *name, char **text, char **err) {
120    (void)err;
121    fake *f = ctx;
122    prog *p = find(f, name);
123    *p = f->p[--f->n];
124    mt_push_own(&f->calls, mt_fmt("unload %s", name));
125    *text = mt_strdup("");
126    return true;
127}
128
129static bool simple(fake *f, const char *what, const char *name, char **text) {
130    mt_push_own(&f->calls, mt_fmt("%s %s", what, name));
131    *text = mt_strdup("ok");
132    return true;
133}
134
135static bool rollback(void *ctx, const char *name, char **text, char **err) { return (void)err, simple(ctx, "rollback", name, text); }
136static bool reset(void *ctx, const char *name, char **text, char **err) { return (void)err, simple(ctx, "reset", name, text); }
137
138static bool prog_run(void *ctx, const char *name, bool on, char **text, char **err) {
139    (void)err;
140    fake *f = ctx;
141    find(f, name)->run = on;
142    return simple(f, on ? "start" : "stop", name, text);
143}
144
145static bool task(void *ctx, const char *name, uint32_t op, uint32_t arg, uint32_t *result, char **err) {
146    (void)err;
147    fake *f = ctx;
148    mt_push_own(&f->calls, mt_fmt("task %s %u %u", name, op, arg));
149    *result = op + arg;
150    return true;
151}
152
153static bool place(void *ctx, const char *name, const mt_strs *words, char **text, char **err) {
154    (void)err;
155    fake *f = ctx;
156    char *j = mt_join(words, " ");
157    mt_push_own(&f->calls, mt_fmt("place %s %s", name, j));
158    free(j);
159    *text = mt_strdup("done");
160    return true;
161}
162
163static bool fetch_view(void *ctx, const char *name, uint64_t hash, uint32_t from, uint32_t *abi, mpw_bytes *b, char **err) {
164    (void)hash, (void)err;
165    fake *f = ctx;
166    mt_push_own(&f->calls, mt_fmt("fetch view %s from %u", name, from));
167    *abi = 1;
168    mpw_put(b, "view", 4);
169    return true;
170}
171
172static bool view_put(void *ctx, const char *name, uint32_t abi, const mpw_bytes *b, char **text, char **err) {
173    (void)err;
174    fake *f = ctx;
175    mt_push_own(&f->calls, mt_fmt("view put %s %u %zu", name, abi, b->n));
176    *text = mt_strdup("");
177    return true;
178}
179
180static void quiesce(void *ctx, bool mirror, bool member) {
181    fake *f = ctx;
182    f->want = false;
183    mt_push_own(&f->calls, mt_fmt("quiesce mirror %d member %d", mirror, member));
184}
185
186static bool want_set(void *ctx) { return ((fake *)ctx)->want; }
187
188static void group_words(void *ctx, uint32_t rank, uint32_t n, uint32_t epoch) {
189    fake *f = ctx;
190    f->words[0] = rank, f->words[1] = n, f->words[2] = epoch;
191}
192
193static void ready(void *ctx, const mpk_hello *spec) {
194    fake *f = ctx;
195    f->hello = *spec;
196    f->helloed = f->want = true;
197}
198
199static bool is_leader(void *ctx) { return ((fake *)ctx)->leader; }
200
201static void status(void *ctx, uint64_t applied, uint32_t st, const char *reason, const mst_outcome *m, size_t nm) {
202    (void)m, (void)nm;
203    fake *f = ctx;
204    f->applied = applied, f->last_status = st;
205    mt_set(&f->last_reason, reason);
206}
207
208static void fails(void *ctx, char *const *n, const uint64_t *h, const bool *u, char *const *w, size_t k) { (void)ctx, (void)n, (void)h, (void)u, (void)w, (void)k; }
209static uint64_t applied(void *ctx) { return ((fake *)ctx)->applied; }
210static void logline(void *ctx, bool error, const char *line) { (void)ctx, (void)error, (void)line; }
211static void sleep_ms(void *ctx, uint32_t ms) { (void)ctx, (void)ms; }
212
213static mrc_ops ops(fake *f, uint32_t me) {
214    return (mrc_ops){f, me, inventory, source, src_free, load, upgrade, unload, rollback, prog_run, reset, task, place, fetch_view, view_put,
215                     quiesce, want_set, group_words, ready, is_leader, status, fails, applied, logline, sleep_ms};
216}
217
218static mfd_doc doc(void) {
219    mfd_doc d = {.version = 3, .term = 1};
220    mesh_needs n = msn_default();
221    mfd_prog_put(&d, "hello", mfd_prog_new("hello", 0xabc, true, &n));
222    mfd_prog_put(&d, "tick", mfd_prog_new("tick", 0x123, false, &n));
223    mesh_needs_free(&n);
224    return d;
225}
226
227static bool called(fake *f, const char *what) { return mt_strs_has(&f->calls, what); }
228
229static void passes(void) {
230    fake f = {0};
231    mrc_ops o = ops(&f, 0);
232    mrc_rec *r = mrc_new(&o);
233    mfd_doc d = doc();
234    CHECK(!mrc_document(r, &d));
235    CHECK(called(&f, "load hello running") && called(&f, "load tick stopped") && f.last_status == MESH_RS_SYNC && f.applied == 3);
236    CHECK(mt_starts(f.last_reason, "ok: load hello (slot 1), load tick (slot 1)"));
237    /* In sync: nothing to do. */
238    mt_strs_free(&f.calls);
239    CHECK(!mrc_document(r, &d) && f.calls.n == 0 && !strcmp(f.last_reason, "ok"));
240    /* An upgrade by hash, a program placed elsewhere unloaded, and an unsupported one reported, not failed. */
241    mfd_doc e = mfd_doc_clone(&d);
242    e.version = 4;
243    mfd_prog_get(&e, "hello")->hash = 0xdef;
244    mfd_prog_get(&e, "tick")->at = ml_some((uint32_t[]){2}, 1);
245    mesh_needs n = msn_default();
246    mfd_prog_put(&e, "metal-only", mfd_prog_new("metal-only", 7, true, &n));
247    mesh_needs_free(&n);
248    f.no_image = "metal-only";
249    CHECK(!mrc_document(r, &e));
250    CHECK(called(&f, "upgrade hello") && called(&f, "unload tick") && f.last_status == MESH_RS_SYNC && mt_contains(f.last_reason, "unsupported here: metal-only"));
251    /* A failure: reported, skipped at that hash, and tried again on the retry. */
252    mfd_doc g = mfd_doc_clone(&e);
253    g.version = 5;
254    n = msn_default();
255    mfd_prog_put(&g, "flaky", mfd_prog_new("flaky", 9, true, &n));
256    mesh_needs_free(&n);
257    f.fail_load = "flaky";
258    CHECK(mrc_document(r, &g) && f.last_status == MESH_RS_FAILED && mt_contains(f.last_reason, "load flaky: jit: boom"));
259    mt_strs_free(&f.calls);
260    CHECK(!mrc_document(r, &g)); /* skipped (same version, same hash): nothing failed this pass */
261    CHECK(!called(&f, "load flaky running"));
262    f.fail_load = NULL;
263    CHECK(!mrc_retry(r) && called(&f, "load flaky running") && f.last_status == MESH_RS_SYNC);
264    /* Placement: the leader runs the requests; one dropped from the document goes back to local. */
265    f.leader = true;
266    mfd_doc h = mfd_doc_clone(&g);
267    h.version = 6;
268    mt_strs w = {0};
269    mt_push(&w, "pin");
270    mt_push(&w, "1");
271    mfd_place_put(&h, "hello", w);
272    mt_strs_free(&f.calls);
273    CHECK(!mrc_document(r, &h) && called(&f, "place hello pin 1"));
274    mfd_place_remove(&h, "hello");
275    h.version = 7;
276    mt_strs_free(&f.calls);
277    CHECK(!mrc_document(r, &h) && called(&f, "place hello local"));
278    mfd_doc_free(&d), mfd_doc_free(&e), mfd_doc_free(&g), mfd_doc_free(&h);
279    mrc_free(r);
280    mt_strs_free(&f.calls);
281    free(f.last_reason);
282}
283
284static void migration(void) {
285    fake f = {0};
286    mrc_ops o = ops(&f, 1);
287    mrc_rec *r = mrc_new(&o);
288    mfd_doc d = {.version = 8, .term = 1};
289    mesh_needs n = msn_default();
290    msn_strs_push(&n.feeds, &n.nfeeds, "pipehash");
291    mfd_prog src = mfd_prog_new("pipesrc", 1, true, &n);
292    mesh_needs_free(&n);
293    n = msn_default();
294    mfd_prog hash = mfd_prog_new("pipehash", 2, true, &n);
295    mesh_needs_free(&n);
296    src.at = ml_some((uint32_t[]){1}, 1), hash.at = ml_some((uint32_t[]){1}, 1);
297    src.has_from = hash.has_from = src.has_to = hash.has_to = true;
298    src.from = hash.from = 2, src.to = hash.to = 1, src.since = hash.since = 8;
299    mfd_prog_put(&d, "pipesrc", src);
300    mfd_prog_put(&d, "pipehash", hash);
301    CHECK(!mrc_document(r, &d));
302    /* Loaded stopped, views producers first, started consumers first. */
303    long ls = mt_strs_find(&f.calls, "load pipesrc stopped"), lh = mt_strs_find(&f.calls, "load pipehash stopped");
304    long vs = mt_strs_find(&f.calls, "fetch view pipesrc from 2"), vh = mt_strs_find(&f.calls, "fetch view pipehash from 2");
305    long ss = mt_strs_find(&f.calls, "start pipesrc"), sh = mt_strs_find(&f.calls, "start pipehash");
306    CHECK(ls >= 0 && lh > ls && vs > lh && vh > vs && sh > vh && ss > sh);
307    CHECK(called(&f, "view put pipehash 1 4") && mt_contains(f.last_reason, "migrate pipehash,pipesrc from 2: view 8 B"));
308    mfd_doc_free(&d);
309    mrc_free(r);
310    mt_strs_free(&f.calls);
311    free(f.last_reason);
312}
313
314static void lockstep(void) {
315    fake f = {0};
316    mrc_ops o = ops(&f, 2);
317    mrc_rec *r = mrc_new(&o);
318    mfd_doc d = {.version = 2, .term = 1};
319    mesh_needs n = msn_default();
320    mfd_prog t = mfd_prog_new("train", 0x77, true, &n);
321    t.lockstep = true;
322    mfd_prog_put(&d, "train", t);
323    mesh_needs_free(&n);
324    mfd_group_push(&d, (mfd_group){.name = mt_strdup("train"), .program = mt_strdup("train"), .epoch = 1, .members = ml_of((uint32_t[]){0, 2}, 2), .state = MFD_RUNNING});
325    CHECK(!mrc_document(r, &d));
326    CHECK(called(&f, "quiesce mirror 0 member 1") && called(&f, "load train running") && called(&f, "reset train"));
327    CHECK(f.words[0] == 1 && f.words[1] == 2 && f.words[2] == 1 && f.helloed && f.hello.epoch == 1 && f.hello.cycle == 0 && f.hello.mask == 5 && f.hello.hash == 0x77);
328    CHECK(f.hello.order[0] == 0 && f.hello.order[1] == 2 && f.hello.order[2] == 0xff);
329    /* A task cycle: quiesced with rounds mirrored, the task run, HELLOed. */
330    mfd_doc c = mfd_doc_clone(&d);
331    c.version = 3;
332    c.groups[0].cycle = 1;
333    c.groups[0].op = (mfd_op){MFD_OP_TASK, 3, 4};
334    f.helloed = false;
335    mt_strs_free(&f.calls);
336    CHECK(!mrc_document(r, &c));
337    CHECK(called(&f, "quiesce mirror 1 member 1") && called(&f, "task train 3 4") && f.helloed && f.hello.cycle == 1);
338    CHECK(mt_contains(f.last_reason, "task train -> 7 (cycle 1)"));
339    /* Paused: unfed, reported. */
340    mfd_doc p = mfd_doc_clone(&c);
341    p.version = 4;
342    p.groups[0].state = MFD_PAUSED;
343    f.want = true;
344    mt_strs_free(&f.calls);
345    CHECK(!mrc_document(r, &p) && called(&f, "quiesce mirror 0 member 1") && mt_contains(f.last_reason, "group train epoch 1: paused"));
346    /* The lockstep program leaves the document: its group goes too. */
347    mfd_doc z = {.version = 5, .term = 1};
348    f.want = false;
349    mt_strs_free(&f.calls);
350    CHECK(!mrc_document(r, &z));
351    mfd_doc_free(&d), mfd_doc_free(&c), mfd_doc_free(&p), mfd_doc_free(&z);
352    mrc_free(r);
353    mt_strs_free(&f.calls);
354    free(f.last_reason);
355}
356
357int main(void) {
358    cycle_ready();
359    passes();
360    migration();
361    lockstep();
362    return mesh_check_done();
363}