1/* durable: persists before sends and answers, one sync per run of persists, and fail-stop (jobs.rs's
2 * `nothing_leaves_without_its_persist`, the durable half; the node half is in @ajax/mesh-jobs-node). */
3#include "mesh/jobs/out.h"
4#include "mesh/abi/peer_frames.h"
5#include "mesh/check.h"
6
7#include <stdlib.h>
8#include <string.h>
9
10typedef struct {
11    int appends, syncs, fail_append, fail_sync;
12    char trace[64];
13} disk;
14
15static bool app(void *c, const mj_log *l, char **err) {
16    disk *d = c;
17    (void)l;
18    d->appends++;
19    strcat(d->trace, "a");
20    if (d->appends == d->fail_append) {
21        *err = mstrdup("ENOSPC");
22        return false;
23    }
24    return true;
25}
26
27static bool syn(void *c, char **err) {
28    disk *d = c;
29    d->syncs++;
30    strcat(d->trace, "s");
31    if (d->fail_sync) {
32        *err = mstrdup("EIO");
33        return false;
34    }
35    return true;
36}
37
38static mj_outs sample(void) {
39    mj_outs v = {0};
40    mj_outs_push(&v, mj_out_persist(mj_log_lid(1)));
41    mj_outs_push(&v, mj_out_persist(mj_log_commit(1, 2)));
42    mj_outs_push(&v, mj_out_log(mstrdup("x")));
43    mj_outs_push(&v, mj_out_send(0, mj_frame_ack(1, 2, false)));
44    mj_outs_push(&v, mj_out_answer(7, true, mstrdup("ok")));
45    mj_outs_push(&v, mj_out_persist(mj_log_lid(2)));
46    mj_outs_push(&v, mj_out_cut(5, 0));
47    return v;
48}
49
50int main(void) {
51    disk d = {0};
52    mj_disk k = {&d, app, syn};
53    mj_outs in = sample();
54    bool ok;
55    char *err = NULL;
56    mj_outs got = mj_durable(&in, &k, &ok, &err);
57    CHECK(ok && got.n == 4);
58    CHECK_S(d.trace, "aasa" "s"); /* two appends, a sync before the send, the last append, a sync at the end */
59    CHECK(got.v[0].kind == MJ_OUT_LOG && got.v[1].kind == MJ_OUT_SEND && got.v[2].kind == MJ_OUT_ANSWER && got.v[3].kind == MJ_OUT_CUT);
60    CHECK(got.v[1].frame.ty == MESH_F_JOB_ACK && got.v[1].frame.n == 2);
61    mj_outs_free(&got);
62    /* The second append fails: nothing after the first append is acted on. */
63    disk d2 = {.fail_append = 2};
64    k.ctx = &d2;
65    in = sample();
66    got = mj_durable(&in, &k, &ok, &err);
67    CHECK(!ok && got.n == 0);
68    CHECK_S(err, "ENOSPC");
69    free(err);
70    mj_outs_free(&got);
71    /* The sync before the send fails: the log line before it went, the send did not. */
72    disk d3 = {.fail_sync = 1};
73    k.ctx = &d3;
74    in = sample();
75    got = mj_durable(&in, &k, &ok, &err);
76    CHECK(!ok && got.n == 1 && got.v[0].kind == MJ_OUT_LOG);
77    CHECK_S(err, "EIO");
78    free(err);
79    mj_outs_free(&got);
80    return mesh_check_done();
81}