==> tests/node.c <==1/* jobs.rs's direct node tests: review findings 6 and 9, no snapshot before the merge commits, an
2 * answer packs its entries, and nothing leaves without its persist. (The fleet tests, and the node
3 * traced output for output against the Rust, are in @ajax/mesh-jobs-sim.) */
4#include "mesh/jobs/node.h"
5#include "mesh/abi/peer_frames.h"
6#include "mesh/check.h"
7
8#include <stdlib.h>
9#include <string.h>
10
11static mj_spec spec(mj_job_id id, uint64_t n) {
12 mj_setup su = {6, 0};
13 return (mj_spec){.id = id, .program = mstrdup("p"), .op = 254, .items = mj_ranges_of(0, n), .attempts = 3, .mode = MJ_MODE_COLLECT,
14 .setup = mj_setup_clone(&su, 1), .nsetup = 1, .submitted_ms = 5, .members = 5};
15}
16
17static mj_ops one(mj_op o) {
18 mj_ops v = {0};
19 mj_ops_push(&v, o);
20 return v;
21}
22
23static mj_ops submit(mj_job_id id, uint64_t n) {
24 mj_spec s = spec(id, n);
25 mj_ops v = one(mj_op_submit(&s));
26 mj_spec_free(&s);
27 return v;
28}
29
30static mj_cfg cfg(uint32_t me) { return (mj_cfg){me, 3, 2, 3000}; }
31
32/* Some Send(to, frame) in outs with type ty (and, for ACK, n and gap). */
33static bool sent(const mj_outs *o, uint32_t to, uint8_t ty, int64_t n, int gap) {
34 for (size_t i = 0; i < o->n; i++) {
35 const mj_out *x = &o->v[i];
36 if (x->kind != MJ_OUT_SEND || x->to != to || x->frame.ty != ty) continue;
37 if (n >= 0 && x->frame.n != (uint64_t)n) continue;
38 if (gap >= 0 && x->frame.gap != (gap != 0)) continue;
39 return true;
40 }
41 return false;
42}
43
44static bool sent_state(const mj_outs *o, uint32_t to, uint32_t kind, int64_t n) {
45 for (size_t i = 0; i < o->n; i++) {
46 const mj_out *x = &o->v[i];
47 if (x->kind == MJ_OUT_SEND && x->to == to && x->frame.ty == MESH_F_JOB_STATE && x->frame.kind == kind && (n < 0 || x->frame.n == (uint64_t)n))
48 return true;
49 }
50 return false;
51}
52
53static void merge_into(mj_outs *a, mj_outs b) { mj_outs_append(a, &b); }
54
55/* Review finding 6, the node half: the snapshot's acknowledgement says 10, not 8. */
56static void snapshot_ack(void) {
57 uint64_t l = (uint64_t)3 << 40;
58 mj_replica r;
59 mj_replica_init(&r);
60 mj_log seq[] = {mj_log_lid(l), mj_log_snapshot(l, 8, (mj_ops){0}), mj_log_entry(l, 9, false, one(mj_op_fails(1, 9, 1))),
61 mj_log_entry(l, 10, false, one(mj_op_fails(1, 10, 1)))};
62 for (int i = 0; i < 4; i++) {
63 mj_replica_apply(&r, &seq[i]);
64 mj_log_free(&seq[i]);
65 }
66 mj_log s8 = mj_log_snapshot(l, 8, submit(1, 100));
67 mj_replica_apply(&r, &s8);
68 mj_log_free(&s8);
69 mj_log s9 = mj_log_snapshot(l, 9, submit(1, 100));
70 mj_replica_apply(&r, &s9);
71 mj_log_free(&s9);
72 mj_node *f = mj_node_new(cfg(1), mj_replica_clone(&r));
73 mj_outs o = mj_node_recv(f, 0, mj_frame_state(l, MESH_JOBS_ST_PART, 9, 9, 0, submit(1, 100)), 1, 1);
74 merge_into(&o, mj_node_recv(f, 0, mj_frame_state(l, MESH_JOBS_ST_SNAP_END, 9, 9, 1, (mj_ops){0}), 2, 2));
75 CHECK(sent(&o, 0, MESH_F_JOB_ACK, 10, 0));
76 CHECK_U(mj_replica_last(&f->rep, l), 10);
77 mj_outs_free(&o);
78 mj_node_free(f);
79 mj_replica_free(&r);
80}
81
82static mj_frame part(uint64_t l, uint32_t seq, uint64_t item) {
83 if (item == 0) return mj_frame_state(l, MESH_JOBS_ST_PART, 1, 0, seq, submit(1, 100));
84 mj_res res = {.lease = 1, .member = 0, .result = 1};
85 return mj_frame_state(l, MESH_JOBS_ST_PART, 1, 0, seq, one(mj_op_done(1, item, &res)));
86}
87
88static mj_frame have(uint64_t lid, uint32_t kind, uint32_t seq, uint64_t src0) { return mj_frame_have(lid, kind, seq, src0, 1, (mj_ops){0}, NULL, 0); }
89
90/* Review finding 9: a table in parts with a part lost is not installed; a pull's answer with a part
91 * lost is asked again. */
92static void part_lost(void) {
93 uint64_t l = (uint64_t)5 << 40;
94 mj_replica r;
95 mj_replica_init(&r);
96 mj_node *f = mj_node_new(cfg(1), r);
97 mj_outs o = {0};
98 merge_into(&o, mj_node_recv(f, 0, part(l, 0, 0), 1000, 1));
99 merge_into(&o, mj_node_recv(f, 0, part(l, 2, 3), 1000, 1));
100 merge_into(&o, mj_node_recv(f, 0, mj_frame_state(l, MESH_JOBS_ST_REPLACE_END, 1, 0, 3, (mj_ops){0}), 1000, 1));
101 CHECK(f->rep.npending == 0 && sent(&o, 0, MESH_F_JOB_ACK, -1, 1));
102 mj_outs_free(&o);
103 /* Sent again whole: installed, without the earlier part 2's leftovers. */
104 merge_into(&o, mj_node_recv(f, 0, part(l, 0, 0), 2000, 2));
105 merge_into(&o, mj_node_recv(f, 0, part(l, 1, 2), 2000, 2));
106 mj_outs_free(&o);
107 o = mj_node_recv(f, 0, mj_frame_state(l, MESH_JOBS_ST_REPLACE_END, 1, 0, 2, (mj_ops){0}), 2000, 2);
108 CHECK(sent(&o, 0, MESH_F_JOB_ACK, 1, 0));
109 mj_outs_free(&o);
110 CHECK(f->rep.npending == 1 && f->rep.pending[0].replace && f->rep.pending[0].n == 1);
111 char *w = mj_ranges_word_new(&mj_table_get(&f->rep.pending[0].table, 1)->done);
112 CHECK_S(w, "2-3");
113 free(w);
114 mj_node_free(f);
115 /* A new leader's pull: an answer with a part lost is asked again, not merged. */
116 mj_replica r2;
117 mj_replica_init(&r2);
118 mj_node *ld = mj_node_new((mj_cfg){0, 3, 2, 3000}, r2);
119 o = mj_node_set_view(ld, (mj_view){6, true, 0, 0x7, 0x7}, 1, 1);
120 uint64_t lid = mj_node_lid(ld);
121 CHECK(sent(&o, 1, MESH_F_JOB_PULL, -1, -1));
122 mj_outs_free(&o);
123 merge_into(&o, mj_node_recv(ld, 1, have(lid, MESH_JOBS_HAVE_COMMITTED, 0, l), 2, 2));
124 merge_into(&o, mj_node_recv(ld, 1, have(lid, MESH_JOBS_HAVE_END, 2, l), 3, 3));
125 CHECK(!mj_node_active(ld) && sent(&o, 1, MESH_F_JOB_PULL, -1, -1));
126 mj_outs_free(&o);
127 merge_into(&o, mj_node_recv(ld, 1, have(lid, MESH_JOBS_HAVE_COMMITTED, 0, l), 4, 4));
128 merge_into(&o, mj_node_recv(ld, 1, have(lid, MESH_JOBS_HAVE_DELTA, 1, l), 4, 4));
129 merge_into(&o, mj_node_recv(ld, 1, have(lid, MESH_JOBS_HAVE_END, 2, l), 5, 5));
130 CHECK(mj_node_active(ld));
131 mj_outs_free(&o);
132 mj_node_free(ld);
133}
134
135/* A new leader whose merge (its first entry) is not committed yet sends no snapshot. */
136static void no_snapshot_before_merge(void) {
137 mj_replica r;
138 mj_replica_init(&r);
139 mj_node *ld = mj_node_new((mj_cfg){0, 3, 2, 3000}, r);
140 mj_outs o = mj_node_set_view(ld, (mj_view){6, true, 0, 0x3, 0x3}, 1, 1);
141 mj_outs_free(&o);
142 uint64_t lid = mj_node_lid(ld);
143 o = mj_node_recv(ld, 1, mj_frame_have(lid, MESH_JOBS_HAVE_COMMITTED, 0, 0, 0, (mj_ops){0}, NULL, 0), 2, 2);
144 mj_outs_free(&o);
145 o = mj_node_recv(ld, 1, mj_frame_have(lid, MESH_JOBS_HAVE_END, 1, 0, 0, (mj_ops){0}, NULL, 0), 2, 2);
146 mj_outs_free(&o);
147 CHECK(mj_node_active(ld) && mj_node_commit_n(ld) == 0);
148 /* System 2 links: it is due a snapshot, but gets only the merge entry while nothing is committed. */
149 o = mj_node_set_view(ld, (mj_view){6, true, 0, 0x7, 0x7}, 3, 3);
150 merge_into(&o, mj_node_tick(ld, 4, 4));
151 CHECK(!sent_state(&o, 2, MESH_JOBS_ST_SNAP_END, -1));
152 CHECK(sent_state(&o, 2, MESH_JOBS_ST_REPLACE_END, 1));
153 mj_outs_free(&o);
154 /* A follower never installs one at n = 0. */
155 mj_replica r2;
156 mj_replica_init(&r2);
157 mj_node *f = mj_node_new((mj_cfg){2, 3, 2, 3000}, r2);
158 mj_spec s = spec(1, 10);
159 mj_op sub = mj_op_submit(&s);
160 mj_table_apply(&f->rep.committed, &sub);
161 mj_op_free(&sub);
162 mj_spec_free(&s);
163 o = mj_node_recv(f, 0, mj_frame_state(lid, MESH_JOBS_ST_PART, 0, 0, 0, (mj_ops){0}), 5, 5);
164 merge_into(&o, mj_node_recv(f, 0, mj_frame_state(lid, MESH_JOBS_ST_SNAP_END, 0, 0, 1, (mj_ops){0}), 5, 5));
165 CHECK(mj_table_get(&f->rep.committed, 1) != NULL);
166 mj_outs_free(&o);
167 mj_node_free(f);
168 mj_node_free(ld);
169}
170
171/* A pull's answer packs a leadership's entries into parts by size, not one part per entry; the merge
172 * takes it whole. */
173static void answer_packs(void) {
174 uint64_t l = (uint64_t)4 << 40;
175 mj_replica r;
176 mj_replica_init(&r);
177 mj_log snap = mj_log_snapshot(l, 1, submit(1, 100000));
178 mj_replica_apply(&r, &snap);
179 mj_log_free(&snap);
180 for (uint64_t n = 2; n < 3002; n++) {
181 mj_res res = {.lease = 9, .member = 1, .result = (uint32_t)n};
182 mj_log e = mj_log_entry(l, n, false, one(mj_op_done(1, n, &res)));
183 mj_replica_apply(&r, &e);
184 mj_log_free(&e);
185 }
186 mj_node *f = mj_node_new(cfg(1), r);
187 mj_frame *a;
188 size_t na;
189 mj_node_answer(f, (uint64_t)5 << 40, &a, &na);
190 CHECK(na <= 4);
191 mj_answer ans;
192 mj_answer_init(&ans);
193 for (size_t i = 0; i < na; i++) {
194 if (a[i].kind == MESH_JOBS_HAVE_COMMITTED)
195 for (size_t k = 0; k < a[i].ops.n; k++) mj_table_apply(&ans.committed, &a[i].ops.v[k]);
196 else if (a[i].kind == MESH_JOBS_HAVE_DELTA) {
197 mj_ops *e = mj_answer_entry(&ans, a[i].src0, a[i].src1);
198 mj_ops c = mj_ops_clone(&a[i].ops);
199 mj_ops_append(e, &c);
200 }
201 mj_frame_free(&a[i]);
202 }
203 free(a);
204 mj_answer *in[1] = {&ans};
205 mj_table t = mj_merge(in, 1);
206 CHECK_U(momap_len(mj_table_get(&t, 1)->results), 3000);
207 mj_table_free(&t);
208 mj_answer_free(&ans);
209 mj_node_free(f);
210}
211
212typedef struct {
213 int appends, fail_append;
214 bool fail_sync, synced;
215} disk;
216static bool app(void *c, const mj_log *l, char **err) {
217 disk *d = c;
218 (void)l;
219 d->appends++;
220 d->synced = false;
221 if (d->appends == d->fail_append) {
222 *err = mstrdup("ENOSPC");
223 return false;
224 }
225 return true;
226}
227static bool syn(void *c, char **err) {
228 disk *d = c;
229 if (d->fail_sync) {
230 *err = mstrdup("EIO");
231 return false;
232 }
233 d->synced = true;
234 return true;
235}
236
237static size_t acks(const mj_outs *o) {
238 size_t n = 0;
239 for (size_t i = 0; i < o->n; i++) n += o->v[i].kind == MJ_OUT_SEND && o->v[i].frame.ty == MESH_F_JOB_ACK;
240 return n;
241}
242
243static mj_outs follow_two(void) {
244 uint64_t l = (uint64_t)2 << 40;
245 mj_replica r;
246 mj_replica_init(&r);
247 mj_node *f = mj_node_new(cfg(1), r);
248 mj_outs o = mj_node_recv(f, 0, mj_frame_state(l, MESH_JOBS_ST_PART, 1, 0, 0, submit(1, 10)), 1, 1);
249 merge_into(&o, mj_node_recv(f, 0, mj_frame_state(l, MESH_JOBS_ST_REPLACE_END, 1, 0, 1, (mj_ops){0}), 1, 1));
250 mj_node_free(f);
251 return o;
252}
253
254/* Review finding 4: no acknowledgement leaves without the append and sync before it. */
255static void persist_first(void) {
256 mj_outs all = follow_two();
257 CHECK_U(acks(&all), 1);
258 for (int k = 1; k <= 2; k++) {
259 disk d = {.fail_append = k};
260 mj_disk dk = {&d, app, syn};
261 mj_outs o = follow_two();
262 bool ok;
263 char *err = NULL;
264 mj_outs sent_ = mj_durable(&o, &dk, &ok, &err);
265 CHECK(!ok && acks(&sent_) == 0);
266 free(err);
267 mj_outs_free(&sent_);
268 }
269 disk d3 = {.fail_sync = true};
270 mj_disk dk3 = {&d3, app, syn};
271 mj_outs o = follow_two();
272 bool ok;
273 char *err = NULL;
274 mj_outs s3 = mj_durable(&o, &dk3, &ok, &err);
275 CHECK(!ok && acks(&s3) == 0);
276 free(err);
277 mj_outs_free(&s3);
278 disk d = {0};
279 mj_disk dk = {&d, app, syn};
280 mj_outs s4 = mj_durable(&all, &dk, &ok, &err);
281 CHECK(ok && acks(&s4) == 1 && d.synced && d.appends == 2);
282 mj_outs_free(&s4);
283}
284
285int main(void) {
286 snapshot_ack();
287 part_lost();
288 no_snapshot_before_merge();
289 answer_packs();
290 persist_first();
291 return mesh_check_done();
292}