==> tests/cluster.c <==1/* A CPU-only fleet: three whole meshds (--backend wasm) in child processes on loopback, members of
2 * one cluster config, driven over their control ports like mesh-fleetcheck drives a fleet:
3 * - they elect one leader and agree on it, its term and the committed version (`fleet`);
4 * - `load` and `start` asked of a follower go through the leader, commit, and every member runs the
5 * program (the reconciler, its image from the local catalog);
6 * - `job submit hashjob items N` on a follower runs on the members' wasm lanes to the end, and
7 * `job results` holds every item exactly once, each result and job_out recomputed here;
8 * - `jobs` answers on every member; the HTTP relay answers;
9 * - a member killed (SIGKILL) is seen down by the others, the two left keep a leader and commit a
10 * change, and they exit 0 on SIGTERM. */
11#define _XOPEN_SOURCE 700 /* sexc compiles tests without the package's cflags (reported) */
12#include "meshd.h"
13#include "mesh/abi.h"
14#include "mesh/check.h"
15#include "mesh/exec_wasm_testkit.h"
16#include "mesh/ref/sha256d.h"
17
18#include <arpa/inet.h>
19#include <fcntl.h>
20#include <netinet/in.h>
21#include <signal.h>
22#include <stdio.h>
23#include <stdlib.h>
24#include <string.h>
25#include <sys/socket.h>
26#include <sys/stat.h>
27#include <sys/time.h>
28#include <sys/wait.h>
29#include <time.h>
30#include <unistd.h>
31
32#define N 3
33#define ITEMS 240
34
35static void sleep_ms(long ms) {
36 struct timespec t = {ms / 1000, (ms % 1000) * 1000000L};
37 nanosleep(&t, NULL);
38}
39
40static long now_ms(void) {
41 struct timespec t;
42 clock_gettime(CLOCK_MONOTONIC, &t);
43 return t.tv_sec * 1000L + t.tv_nsec / 1000000L;
44}
45
46static int free_port(void) {
47 int fd = socket(AF_INET, SOCK_STREAM, 0);
48 struct sockaddr_in sa = {0};
49 sa.sin_family = AF_INET;
50 sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
51 bind(fd, (struct sockaddr *)&sa, sizeof sa);
52 socklen_t sl = sizeof sa;
53 getsockname(fd, (struct sockaddr *)&sa, &sl);
54 close(fd);
55 return ntohs(sa.sin_port);
56}
57
58static int conn(int port) {
59 for (int i = 0; i < 500; i++) {
60 int fd = socket(AF_INET, SOCK_STREAM, 0);
61 struct sockaddr_in sa = {0};
62 sa.sin_family = AF_INET;
63 sa.sin_port = htons((uint16_t)port);
64 sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
65 if (connect(fd, (struct sockaddr *)&sa, sizeof sa) == 0) return fd;
66 close(fd);
67 sleep_ms(20);
68 }
69 return -1;
70}
71
72static int CTL[N], HTTP[N], PEER[N];
73static pid_t PID[N];
74static bool ALIVE[N];
75
76/* One request on a fresh connection to member k's control port: a line ("ok ..."/"err ..."), or a
77 * framed reply (framed). NULL when the member does not answer. */
78static char *request(int k, const char *req, bool framed) {
79 int fd = conn(CTL[k]);
80 if (fd < 0) return NULL;
81 struct timeval tv = {120, 0};
82 setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof tv);
83 if (write(fd, req, strlen(req)) < 0 || write(fd, "\n", 1) < 0) {
84 close(fd);
85 return NULL;
86 }
87 char *b = NULL;
88 if (framed) {
89 uint8_t h[4];
90 size_t got = 0;
91 while (got < 4) {
92 ssize_t r = read(fd, h + got, 4 - got);
93 if (r <= 0) break;
94 got += (size_t)r;
95 }
96 if (got == 4) {
97 size_t n = (size_t)h[0] | (size_t)h[1] << 8 | (size_t)h[2] << 16 | (size_t)h[3] << 24;
98 b = malloc(n + 1);
99 size_t at = 0;
100 while (at < n) {
101 ssize_t r = read(fd, b + at, n - at);
102 if (r <= 0) break;
103 at += (size_t)r;
104 }
105 b[at] = 0;
106 }
107 } else {
108 b = malloc(65536);
109 size_t i = 0;
110 while (i < 65535 && read(fd, b + i, 1) == 1)
111 if (b[i++] == '\n') break;
112 b[i] = 0;
113 }
114 close(fd);
115 return b;
116}
117
118/* The value after key in a line of words ("key value"), into out. */
119static bool kv(const char *line, const char *key, char *out, size_t cap) {
120 size_t kl = strlen(key);
121 for (const char *p = line; (p = strstr(p, key)); p += kl) {
122 if ((p == line || p[-1] == ' ') && p[kl] == ' ') {
123 const char *v = p + kl + 1;
124 size_t n = strcspn(v, " \n");
125 if (n >= cap) n = cap - 1;
126 memcpy(out, v, n);
127 out[n] = 0;
128 return true;
129 }
130 }
131 return false;
132}
133
134typedef struct {
135 char leader[8], up[32], role[16];
136 unsigned long long term, version, applied;
137} view;
138
139static bool fleet_view(int k, view *v) {
140 char *t = request(k, "fleet", true);
141 if (!t) return false;
142 unsigned sys, n, q;
143 unsigned long long acc;
144 bool ok = sscanf(t, "fleet system %u of %u role %15s leader %7s term %llu version %llu accepted %llu applied %llu quorum %u up %31s ", &sys, &n,
145 v->role, v->leader, &v->term, &v->version, &acc, &v->applied, &q, v->up) == 10;
146 free(t);
147 return ok;
148}
149
150/* The members in ks agree on a leader, its term and the committed version, all applied, every one up
151 * (mesh-fleetcheck's `agreed`): the leader, or -1. */
152static int agreed(const int *ks, int nk, const char *up) {
153 view v0, v;
154 if (!fleet_view(ks[0], &v0) || !strcmp(v0.leader, "-")) return -1;
155 for (int i = 0; i < nk; i++) {
156 if (!fleet_view(ks[i], &v)) return -1;
157 if (strcmp(v.leader, v0.leader) || v.term != v0.term || v.version != v0.version || v.applied != v.version || strcmp(v.up, up)) return -1;
158 }
159 return atoi(v0.leader);
160}
161
162static int wait_agreed(const int *ks, int nk, const char *up, long ms) {
163 long end = now_ms() + ms;
164 int l;
165 while ((l = agreed(ks, nk, up)) < 0 && now_ms() < end) sleep_ms(100);
166 return l;
167}
168
169/* A change through member k: "ok ..." (retried while the fleet is busy or the leader does not yet know
170 * the program runs: mesh-fleetcheck's submit). */
171static char *change(int k, const char *req) {
172 long end = now_ms() + 30000;
173 for (;;) {
174 char *r = request(k, req, false);
175 if (!r || strncmp(r, "ok", 2) == 0 || now_ms() > end) return r;
176 if (!strstr(r, "no member runs") && !strstr(r, "merging") && !strstr(r, "busy") && !strstr(r, "pending") && !strstr(r, "no leader")) return r;
177 free(r);
178 sleep_ms(300);
179 }
180}
181
182/* hashjob's item i: the fold over K = 1024 of sha256d(LE8(i * K + j))'s first LE u32, and the first 16
183 * bytes of sha256d(LE8(i)) (its job_out). */
184static uint32_t hashjob(uint64_t i, uint8_t out[16]) {
185 uint32_t fold = 0;
186 uint8_t h[32], m[8];
187 for (uint64_t j = 0; j < 1024; j++) {
188 uint64_t x = i * 1024 + j;
189 for (int b = 0; b < 8; b++) m[b] = (uint8_t)(x >> (8 * b));
190 mesh_sha256d(m, 8, h);
191 fold ^= (uint32_t)h[0] | (uint32_t)h[1] << 8 | (uint32_t)h[2] << 16 | (uint32_t)h[3] << 24;
192 }
193 if (fold == MESH_R_JOB_FAIL) fold--;
194 for (int b = 0; b < 8; b++) m[b] = (uint8_t)(i >> (8 * b));
195 mesh_sha256d(m, 8, h);
196 memcpy(out, h, 16);
197 return fold;
198}
199
200static void copy_file(const char *from, const char *to) {
201 FILE *in = fopen(from, "rb"), *out = fopen(to, "wb");
202 char buf[65536];
203 size_t k;
204 while ((k = fread(buf, 1, sizeof buf, in)) > 0) fwrite(buf, 1, k, out);
205 fclose(in);
206 fclose(out);
207}
208
209static pid_t spawn_member(int k, const char *root, const char *cat, const char *conf) {
210 char sys[8], ctl[32], http[32], fleet[256], logs[256], logf[256];
211 snprintf(sys, sizeof sys, "%d", k);
212 snprintf(ctl, sizeof ctl, "127.0.0.1:%d", CTL[k]);
213 snprintf(http, sizeof http, "127.0.0.1:%d", HTTP[k]);
214 snprintf(fleet, sizeof fleet, "%s/m%d.fleet", root, k);
215 snprintf(logs, sizeof logs, "%s/logs%d", root, k);
216 snprintf(logf, sizeof logf, "%s/meshd%d.log", root, k);
217 char *argv[] = {"meshd", "--backend", "wasm", "--programs", (char *)cat, "--cluster", (char *)conf, "--system", sys, "--fleet-state", fleet,
218 "--control", ctl, "--http", http, "--logs", logs, NULL};
219 pid_t pid = fork();
220 if (pid == 0) {
221 int fd = open(logf, O_WRONLY | O_CREAT | O_TRUNC, 0600);
222 dup2(fd, 2);
223 dup2(fd, 1);
224 meshd_main((int)(sizeof argv / sizeof *argv) - 1, argv);
225 }
226 return pid;
227}
228
229static int wait_exit(pid_t pid) {
230 int st = 0;
231 waitpid(pid, &st, 0);
232 return WIFEXITED(st) ? WEXITSTATUS(st) : 128 + WTERMSIG(st);
233}
234
235static void stop_all(void) {
236 for (int k = 0; k < N; k++)
237 if (ALIVE[k]) kill(PID[k], SIGKILL), waitpid(PID[k], NULL, 0);
238}
239
240/* The job's line in `jobs` on member k: its state word, or "" */
241static bool job_state(int k, const char *id, char *state, size_t cap, unsigned long long *done) {
242 char *t = request(k, "jobs", true);
243 if (!t) return false;
244 bool found = false;
245 for (char *l = t; l && *l; l = strchr(l, '\n') ? strchr(l, '\n') + 1 : NULL) {
246 char w[64];
247 if (strncmp(l, "job ", 4) || sscanf(l + 4, "%63s", w) != 1 || strcmp(w, id)) continue;
248 char *e = strchr(l, '\n');
249 if (e) *e = 0;
250 found = kv(l, "state", state, cap);
251 char d[32] = "0";
252 const char *items = strstr(l, " items ");
253 if (items) kv(items, "done", d, sizeof d);
254 *done = strtoull(d, NULL, 10);
255 break;
256 }
257 free(t);
258 return found;
259}
260
261int main(void) {
262 char *wasm = mesh_kit_artifact("ajax-mesh-prog-hashjob-wasm");
263 if (!wasm) {
264 fprintf(stderr, "meshd cluster test: programs/hashjob-wasm is not built (run `sexc test` for the workspace); skipped\n");
265 return 0;
266 }
267 char root[128], cat[192], path[320], conf[192];
268 snprintf(root, sizeof root, "/tmp/meshd-fleet-%d", (int)getpid());
269 snprintf(cat, sizeof cat, "%s/programs", root);
270 snprintf(conf, sizeof conf, "%s/cluster.conf", root);
271 mkdir(root, 0700);
272 mkdir(cat, 0700);
273 snprintf(path, sizeof path, "%s/hashjob", cat);
274 mkdir(path, 0700);
275 snprintf(path, sizeof path, "%s/hashjob/program.toml", cat);
276 FILE *f = fopen(path, "w");
277 fputs("name = \"hashjob\"\nabi = 6\nbackends = [\"wasm\"]\nview_bytes = 0x1000\nnodes = 4\npolicy = \"job\"\nfanout = 4\n"
278 "param = 1024\njob_base = 0x08\njob_out = \"0x100:16\"\ncounters = [\"items\", \"khashes\", \"failed\", \"\"]\n"
279 "fields = [\"items:u64:0x20\", \"hashes:u64:0x28\", \"last:u64:0x30\", \"fail_item:u32:0x10\", \"trap_item:u32:0x18\"]\n"
280 "task_ms = 2000\nautostart = false\n",
281 f);
282 fclose(f);
283 snprintf(path, sizeof path, "%s/hashjob/hashjob.wasm", cat);
284 copy_file(wasm, path);
285 free(wasm);
286
287 for (int k = 0; k < N; k++) CTL[k] = free_port(), HTTP[k] = free_port(), PEER[k] = free_port();
288 f = fopen(conf, "w");
289 fprintf(f, "# a CPU-only fleet on loopback\nsystems = %d\nping_ms = 100\ndead_ms = 1500\n", N);
290 for (int k = 0; k < N; k++) fprintf(f, "%d.peer = 127.0.0.1:%d\n", k, PEER[k]);
291 fclose(f);
292 for (int k = 0; k < N; k++) {
293 PID[k] = spawn_member(k, root, cat, conf);
294 ALIVE[k] = true;
295 }
296 atexit(stop_all);
297
298 /* One leader, agreed by all three, every member up. */
299 int all[N] = {0, 1, 2};
300 int leader = wait_agreed(all, N, "0,1,2", 60000);
301 CHECK(leader >= 0);
302 if (leader < 0) return 1;
303 int follower = (leader + 1) % N;
304 view v;
305 CHECK(fleet_view(follower, &v) && !strcmp(v.role, "follower"));
306 CHECK(fleet_view(leader, &v) && !strcmp(v.role, "leader"));
307 fprintf(stderr, "cluster: leader %d, term %llu\n", leader, v.term);
308
309 /* The program on the fleet, asked of a follower: committed, reconciled and running everywhere. */
310 char *r = change(follower, "load hashjob");
311 CHECK(r && strncmp(r, "ok", 2) == 0);
312 fprintf(stderr, "load hashjob: %s", r ? r : "(none)\n");
313 free(r);
314 r = change(follower, "start hashjob");
315 CHECK(r && strncmp(r, "ok", 2) == 0);
316 fprintf(stderr, "start hashjob: %s", r ? r : "(none)\n");
317 free(r);
318 CHECK(wait_agreed(all, N, "0,1,2", 30000) == leader);
319 for (int k = 0; k < N; k++) {
320 char *p = request(k, "programs", true);
321 CHECK(p && strstr(p, " hashjob running "));
322 free(p);
323 }
324
325 /* A job array through a follower: to the end, every item exactly once and right. */
326 char req[64];
327 snprintf(req, sizeof req, "job submit hashjob items %d", ITEMS);
328 r = change(follower, req);
329 CHECK(r && strncmp(r, "ok job ", 7) == 0);
330 fprintf(stderr, "%s: %s", req, r ? r : "(none)\n");
331 char id[32] = "";
332 if (r) sscanf(r, "ok job %31s", id);
333 free(r);
334 char state[32] = "";
335 unsigned long long done = 0;
336 long end = now_ms() + 300000;
337 while (now_ms() < end) {
338 if (job_state(leader, id, state, sizeof state, &done) && !strncmp(state, "done", 4)) break;
339 sleep_ms(250);
340 }
341 CHECK_S(state, "done");
342 CHECK_U(done, ITEMS);
343 for (int k = 0; k < N; k++) {
344 char s2[32] = "";
345 unsigned long long d2 = 0;
346 long e2 = now_ms() + 10000;
347 while (now_ms() < e2 && !(job_state(k, id, s2, sizeof s2, &d2) && !strncmp(s2, "done", 4))) sleep_ms(100);
348 CHECK_S(s2, "done");
349 }
350 snprintf(req, sizeof req, "job results %s 0 100000", id);
351 char *res = request(follower, req, true);
352 CHECK(res && strncmp(res, "results ", 8) == 0);
353 static uint8_t seen[ITEMS];
354 unsigned per[N] = {0};
355 int lines = 0, bad = 0, dup = 0;
356 for (char *l = res ? strchr(res, '\n') : NULL; l && l[1]; l = strchr(l + 1, '\n')) {
357 unsigned long long item;
358 unsigned result;
359 if (sscanf(l + 1, "%llu %u", &item, &result) != 2) continue;
360 char *e = strchr(l + 1, '\n');
361 if (e) *e = 0;
362 char out[64] = "", mem[16] = "";
363 kv(l + 1, "out", out, sizeof out);
364 kv(l + 1, "member", mem, sizeof mem);
365 if (e) *e = '\n';
366 lines++;
367 if (item >= ITEMS) {
368 bad++;
369 continue;
370 }
371 if (seen[item]++) dup++;
372 uint8_t want[16];
373 char hex[33];
374 uint32_t fold = hashjob(item, want);
375 mesh_sha_hex(want, 16, hex);
376 if (fold != result || strcmp(out, hex)) bad++;
377 unsigned m = (unsigned)atoi(mem);
378 if (m < N) per[m]++;
379 }
380 free(res);
381 int missing = 0;
382 for (int i = 0; i < ITEMS; i++) missing += !seen[i];
383 CHECK_I(lines, ITEMS);
384 CHECK_I(dup, 0);
385 CHECK_I(missing, 0);
386 CHECK_I(bad, 0);
387 fprintf(stderr, "job %s: %d items exactly once, every result and job_out recomputed: equal; per member %u / %u / %u\n", id, lines, per[0], per[1],
388 per[2]);
389
390 /* `jobs` on every member; the HTTP relay answers. */
391 for (int k = 0; k < N; k++) {
392 char *j = request(k, "jobs", true);
393 CHECK(j && strncmp(j, "jobs system ", 12) == 0);
394 free(j);
395 }
396 int hfd = conn(HTTP[follower]);
397 CHECK(hfd >= 0);
398 const char *get = "GET /status HTTP/1.0\r\nHost: x\r\n\r\n";
399 CHECK(write(hfd, get, strlen(get)) == (ssize_t)strlen(get));
400 char hb[64] = "";
401 struct timeval tv = {30, 0};
402 setsockopt(hfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof tv);
403 ssize_t hn = read(hfd, hb, sizeof hb - 1);
404 hb[hn > 0 ? hn : 0] = 0;
405 CHECK(strncmp(hb, "HTTP/1.", 7) == 0);
406 close(hfd);
407
408 /* A member killed: the others see it down, keep their leader (or elect one) and commit. */
409 int victim = leader == 2 ? 1 : 2;
410 kill(PID[victim], SIGKILL);
411 waitpid(PID[victim], NULL, 0);
412 ALIVE[victim] = false;
413 int rest[2], nr = 0;
414 for (int k = 0; k < N; k++)
415 if (k != victim) rest[nr++] = k;
416 char up2[16];
417 snprintf(up2, sizeof up2, "%d,%d", rest[0], rest[1]);
418 int l2 = wait_agreed(rest, 2, up2, 30000);
419 CHECK(l2 >= 0 && l2 != victim);
420 for (int i = 0; i < 2; i++) {
421 char *t = request(rest[i], "fleet", true);
422 char want[32];
423 snprintf(want, sizeof want, "\nmember %d down ", victim);
424 CHECK(t && strstr(t, want));
425 free(t);
426 }
427 r = change(rest[0], "stop hashjob");
428 CHECK(r && strncmp(r, "ok", 2) == 0);
429 fprintf(stderr, "member %d killed: seen down; leader %d; stop hashjob: %s", victim, l2, r ? r : "(none)\n");
430 free(r);
431 CHECK(wait_agreed(rest, 2, up2, 30000) == l2);
432
433 for (int i = 0; i < 2; i++) kill(PID[rest[i]], SIGTERM);
434 for (int i = 0; i < 2; i++) {
435 CHECK_I(wait_exit(PID[rest[i]]), 0);
436 ALIVE[rest[i]] = false;
437 }
438 char cmd[256];
439 snprintf(cmd, sizeof cmd, "rm -rf %s", root);
440 if (mesh_check_failures == 0 && system(cmd) != 0) fprintf(stderr, "could not remove %s\n", root);
441 else if (mesh_check_failures) fprintf(stderr, "logs kept in %s\n", root);
442 return mesh_check_done();
443}