==> src/relay.c <==1/* peer.rs: run, Relay (events, links, the core's outputs, beats, changes, the leader, policies,
2 * images). */
3#include "relay.h"
4#include "mesh/abi/backend.h"
5#include "mesh/abi/exchange.h"
6#include "mesh/abi/group_state.h"
7#include "mesh/abi/host_links.h"
8#include "mesh/abi/host_members.h"
9#include "mesh/abi/host_mload.h"
10#include "mesh/abi/host_status.h"
11#include "mesh/abi/peer_frames.h"
12#include "mesh/abi/peer_sizes.h"
13#include "mesh/abi/reconcile.h"
14#include "mesh/fleet_groups.h"
15#include "mesh/rust_fmt.h"
16
17#include <stdio.h>
18#include <stdlib.h>
19#include <string.h>
20#include <netdb.h>
21#include <sys/socket.h>
22#include <time.h>
23#include <unistd.h>
24
25#define CHANGE_WAIT_MS 60000u /* a change's reconcile on the reachable members, answered within */
26#define FIRST_DOC_RETRY_MS 1000u /* a leader's first document that did not commit is proposed again */
27#define FORWARD_WAIT_MS 70000u /* a change forwarded to the leader */
28#define IMAGE_WAIT_MS 30000u
29#define HELLO_RESEND_MS 1000u
30
31uint64_t mpr_mono_us(void) {
32 struct timespec ts;
33 clock_gettime(CLOCK_MONOTONIC, &ts);
34 return (uint64_t)ts.tv_sec * 1000000u + (uint64_t)ts.tv_nsec / 1000u;
35}
36
37uint64_t mpr_mono_ms(void) { return mpr_mono_us() / 1000u; }
38
39uint64_t relay_now(const mpr_relay *r) { return mpr_mono_ms() - r->t0_ms + 1; }
40
41void relay_log(mpr_relay *r, bool error, char *line) {
42 if (r->h.log) r->h.log(r->h.ctx, error, line);
43 free(line);
44}
45
46void mpr_recon_free(mpr_recon *x) {
47 free(x->reason);
48 for (size_t i = 0; i < x->nmoves; i++) free(x->moves[i].name);
49 free(x->moves);
50 *x = (mpr_recon){0};
51}
52
53void mpr_payload_free(mpr_payload *p) {
54 free(p->dir);
55 free(p->manifest);
56 mpw_bytes_free(&p->ptx);
57 *p = (mpr_payload){0};
58}
59
60void waiter_done(waiter *w, bool ok, char *err) {
61 pthread_mutex_lock(&w->mu);
62 w->ok = ok;
63 w->err = err;
64 w->done = true;
65 pthread_cond_signal(&w->cv);
66 pthread_mutex_unlock(&w->mu);
67}
68
69/* ---- events */
70
71static void post(mpr_relay *r, event e) {
72 pthread_mutex_lock(&r->ev_mu);
73 if (r->ev_n == r->ev_cap) {
74 r->ev_cap = r->ev_cap ? r->ev_cap * 2 : 64;
75 r->ev = mt_realloc(r->ev, r->ev_cap * sizeof *r->ev);
76 }
77 r->ev[r->ev_n++] = e;
78 pthread_cond_signal(&r->ev_cv);
79 pthread_mutex_unlock(&r->ev_mu);
80}
81
82static void event_free(event *e) {
83 free(e->text);
84 free(e->name);
85 free(e->dir);
86 mpw_bytes_free(&e->frame);
87}
88
89/* ---- links */
90
91uint32_t relay_linked_mask(mpr_relay *r) {
92 uint32_t m = 1u << r->me;
93 pthread_mutex_lock(&r->links_mu);
94 for (uint32_t k = 0; k < r->n; k++)
95 if (r->links[k]) m |= 1u << k;
96 pthread_mutex_unlock(&r->links_mu);
97 return m;
98}
99
100bool relay_linked(mpr_relay *r, uint32_t k) {
101 pthread_mutex_lock(&r->links_mu);
102 bool yes = k < MAXS && r->links[k] != NULL;
103 pthread_mutex_unlock(&r->links_mu);
104 return yes;
105}
106
107uint64_t relay_last_rx(mpr_relay *r, uint32_t k, bool *linked) {
108 pthread_mutex_lock(&r->links_mu);
109 uint64_t t = 0;
110 *linked = r->links[k] != NULL;
111 if (*linked) t = atomic_load(&r->links[k]->last_rx);
112 pthread_mutex_unlock(&r->links_mu);
113 return t;
114}
115
116void relay_send(mpr_relay *r, uint32_t k, const mpw_bytes *f) {
117 if (!f->n) return;
118 pthread_mutex_lock(&r->links_mu);
119 if (k < MAXS && r->links[k]) mpn_send(r->links[k]->l, f->b, f->n);
120 pthread_mutex_unlock(&r->links_mu);
121}
122
123void relay_broadcast(mpr_relay *r, const mpw_bytes *f) {
124 for (uint32_t k = 0; k < r->n; k++)
125 if (k != r->me) relay_send(r, k, f);
126}
127
128bool mpr_send(mpr_relay *r, uint32_t k, const uint8_t *f, size_t n, bool bulk) {
129 pthread_mutex_lock(&r->links_mu);
130 bool ok = k < MAXS && r->links[k] && (bulk ? mpn_send_bulk(r->links[k]->l, f, n) : mpn_send(r->links[k]->l, f, n));
131 pthread_mutex_unlock(&r->links_mu);
132 return ok;
133}
134
135void mpr_broadcast(mpr_relay *r, const uint8_t *f, size_t n) {
136 for (uint32_t k = 0; k < r->n; k++)
137 if (k != r->me) mpr_send(r, k, f, n, false);
138}
139
140/* ---- the reader's direct frames (on the link's reader thread) */
141
142static char *on_frame(void *ctx, uint32_t from, uint64_t lid, mpw_bytes f) {
143 linkrec *lr = ctx;
144 mpr_relay *r = lr->r;
145 uint64_t up = r->h.uptime_ms(r->h.ctx);
146 atomic_store(&lr->last_rx, up > 1 ? up : 1);
147 uint8_t ty = f.b[0];
148 const uint8_t *b = f.b + 1;
149 char *err = NULL;
150 switch (ty) {
151 case MESH_F_ROUND:
152 group_round(r, lr, b);
153 mpw_bytes_free(&f);
154 return NULL;
155 case MESH_F_PLACE: {
156 size_t rest;
157 mpw_place m;
158 if (!mpw_place_rest_len(b, &rest, &err) || !mpw_place_decode(b, b + MPW_PLACE_HEAD, rest, &m, &err)) break;
159 mpw_place_free(&m);
160 r->h.place_frame(r->h.ctx, from, f);
161 return NULL;
162 }
163 case MESH_F_FRAME: {
164 size_t rest;
165 mpw_frame m;
166 if (!mpw_frame_rest_len(b, &rest, &err) || !mpw_frame_decode(b, b + MPW_FRAME_HEAD, rest, &m, &err)) break;
167 uint32_t sys = m.sys;
168 mpw_frame_free(&m);
169 if (sys != from) {
170 err = mt_fmt("F_FRAME of system %u on system %u's link", sys, from);
171 break;
172 }
173 r->h.place_frame(r->h.ctx, from, f);
174 return NULL;
175 }
176 case MESH_F_INPUT: {
177 char name[MESH_NAME_BYTES + 1];
178 if (!mpw_name_of(b, MESH_NAME_BYTES, name, &err)) break;
179 r->h.place_frame(r->h.ctx, from, f);
180 return NULL;
181 }
182 case MESH_F_PERF: {
183 uint32_t n = mpw_u32_at(b, 0);
184 char *text = mt_strndup((const char *)b + 4, n);
185 board_report(r, from, text);
186 free(text);
187 mpw_bytes_free(&f);
188 return NULL;
189 }
190 case MESH_F_LOADREP: {
191 uint64_t at = up > 1 ? up : 1;
192 size_t base = mesh_mload(from);
193 r->h.host_bytes(r->h.ctx, base + MESH_ML_SHARE, b, MESH_LOADREP_BYTES);
194 uint8_t ab[8];
195 memcpy(ab, &at, 8);
196 r->h.host_bytes(r->h.ctx, base + MESH_ML_AT_MS, ab, 8);
197 mpw_bytes_free(&f);
198 return NULL;
199 }
200 default:
201 if (ty >= MESH_F_JOB_AVAIL && ty <= MESH_F_RESULT) {
202 r->h.jobs_frame(r->h.ctx, from, ty, b + MESH_JOB_HEAD, f.n - 1 - MESH_JOB_HEAD);
203 mpw_bytes_free(&f);
204 return NULL;
205 }
206 if (ty >= MESH_F_BLOB_PUT && ty <= MESH_F_BLOB) {
207 r->h.blobs_frame(r->h.ctx, from, ty, b + MESH_JOB_HEAD, f.n - 1 - MESH_JOB_HEAD);
208 mpw_bytes_free(&f);
209 return NULL;
210 }
211 post(r, (event){.kind = EV_FRAME, .from = from, .lid = lid, .frame = f});
212 return NULL;
213 }
214 mpw_bytes_free(&f);
215 return err;
216}
217
218static void on_closed(void *ctx, uint32_t from, uint64_t lid, const char *why) {
219 linkrec *lr = ctx;
220 post(lr->r, (event){.kind = EV_CLOSED, .from = from, .lid = lid, .text = mt_strdup(why)});
221}
222
223static void on_count(void *ctx, uint32_t k, bool tx, uint8_t type, size_t bytes) {
224 linkrec *lr = ctx;
225 if (lr->r->h.net_count) lr->r->h.net_count(lr->r->h.ctx, k, tx, type, bytes);
226}
227
228static void on_linked(void *ctx, int fd, const mpk_join *join, atomic_bool *alive) {
229 mpr_relay *r = ctx;
230 post(r, (event){.kind = EV_LINK, .fd = fd, .from = join->system, .join = *join, .alive = alive});
231}
232
233static void on_link_log(void *ctx, const char *line) { relay_log(ctx, true, mt_strdup(line)); }
234
235/* ---- the core's outputs */
236
237void relay_my_status(mpr_relay *r, mpk_beat_status *out) {
238 mpr_recon rs = {0};
239 r->h.recon(r->h.ctx, &rs);
240 uint32_t l;
241 bool hl = mfc_leader(r->core, &l);
242 *out = (mpk_beat_status){.has_leader = hl, .leader = hl ? l : 0, .applied = rs.applied, .status = rs.status, .up = mfc_alive(r->core, relay_now(r)),
243 .group_epoch = r->has_hello ? r->hello.epoch : 0, .group_state = atomic_load(&r->hello_state), .reason = rs.reason, .moves = rs.moves,
244 .nmoves = rs.nmoves};
245 rs.reason = NULL, rs.moves = NULL, rs.nmoves = 0;
246 mpr_recon_free(&rs);
247}
248
249static bool persist(void *ctx, char **err) {
250 mpr_relay *r = ctx;
251 if (!r->has_store) return true;
252 return mfs_write(&r->store, mfc_persist(r->core), err);
253}
254
255static void done(mpr_relay *r, uint64_t v, bool ok, const char *err);
256
257static void apply(mpr_relay *r, mfc_outs outs) {
258 if (atomic_load(&r->fatal)) {
259 mfc_outs_free(&outs);
260 return;
261 }
262 char *err = NULL;
263 if (!mfc_durable(&outs, persist, r, &err)) {
264 char *why = mt_fmt("fleet: cannot persist the fleet state (%s)", err);
265 atomic_store(&r->fatal, true);
266 r->h.fail_stop(r->h.ctx, why);
267 free(why), free(err);
268 mfc_outs_free(&outs);
269 return;
270 }
271 for (size_t i = 0; i < outs.n; i++) {
272 mfc_out *o = &outs.v[i];
273 switch (o->kind) {
274 case MFC_OUT_PERSIST: break;
275 case MFC_OUT_SEND: {
276 mpw_bytes f;
277 if (o->msg.kind == MFC_BEAT) {
278 mpk_beat_status st;
279 relay_my_status(r, &st);
280 f = mpk_beat_frame(&o->msg, &st);
281 mpk_beat_status_free(&st);
282 } else f = mpk_msg_frame(&o->msg);
283 relay_send(r, o->to, &f);
284 mpw_bytes_free(&f);
285 break;
286 }
287 case MFC_OUT_COMMITTED: {
288 const mfd_doc *d = mfc_committed(r->core);
289 relay_log(r, false, mt_fmt("fleet: version %llu committed (term %llu): %zu program(s), %zu group(s)", (unsigned long long)d->version,
290 (unsigned long long)d->term, d->nprogs, d->ngroups));
291 r->h.committed(r->h.ctx, d);
292 break;
293 }
294 case MFC_OUT_LEADER:
295 if (o->has_leader && o->leader == r->me) {
296 relay_log(r, false, mt_fmt("fleet: system %u is the leader (term %llu)", r->me, (unsigned long long)mfc_term(r->core)));
297 r->has_first_doc_at = false;
298 } else if (o->has_leader) {
299 relay_log(r, false, mt_fmt("fleet: leader is system %u (term %llu)", o->leader, (unsigned long long)mfc_term(r->core)));
300 } else {
301 relay_log(r, true, mt_fmt("fleet: no leader (this member hears %d of %u members; a majority is %u): changes are refused until there is one",
302 ml_popcount(mfc_alive(r->core, relay_now(r))), r->n, mfc_quorum(mfc_cfg(r->core))));
303 }
304 /* Not leading: the leader's memory goes (a later leadership takes a migration over from the document). */
305 if (!(o->has_leader && o->leader == r->me)) {
306 ms_free(&r->paused_since);
307 if (r->has_mig) mst_move_free(&r->mig.mv);
308 r->has_mig = false;
309 }
310 break;
311 case MFC_OUT_DONE: done(r, o->version, o->ok, o->text); break;
312 case MFC_OUT_LOG: relay_log(r, false, mt_strdup(o->text)); break;
313 }
314 }
315 mfc_outs_free(&outs);
316}
317
318void relay_beat_all(mpr_relay *r) {
319 mfc_msg b = mfc_beat(r->core);
320 mpk_beat_status st;
321 relay_my_status(r, &st);
322 mpw_bytes f = mpk_beat_frame(&b, &st);
323 relay_broadcast(r, &f);
324 mpw_bytes_free(&f);
325 mpk_beat_status_free(&st);
326 r->last_beat_ms = mpr_mono_ms();
327}
328
329static char *leader_word(mpr_relay *r) {
330 uint32_t l;
331 if (mfc_leader(r->core, &l)) return mt_fmt("system %u leads term %llu", l, (unsigned long long)mfc_term(r->core));
332 if (mfc_fixed(mfc_cfg(r->core))) return mt_strdup("no leader: system 0, the pair's fixed coordinator, is unreachable (a pair has no failover)");
333 return mt_fmt("no leader: this member hears %d of %u members, a majority is %u", ml_popcount(mfc_alive(r->core, relay_now(r))), r->n, mfc_quorum(mfc_cfg(r->core)));
334}
335
336void relay_reply(mpr_relay *r, replyto to, bool ok, const char *text) {
337 if (!to.remote) {
338 if (to.local.fn) to.local.fn(to.local.ctx, ok, text);
339 return;
340 }
341 size_t n = strlen(text);
342 if (n > MESH_F_CTL_REPLY_MAX) n = MESH_F_CTL_REPLY_MAX;
343 mpw_bytes f = {0};
344 uint8_t t = MESH_F_CTL_REPLY;
345 mpw_put(&f, &t, 1);
346 mpw_put32(&f, to.id);
347 mpw_put32(&f, ok);
348 mpw_put32(&f, (uint32_t)n);
349 mpw_put(&f, text, n);
350 relay_send(r, to.k, &f);
351 mpw_bytes_free(&f);
352}
353
354void mpr_ctl_reply(mpr_relay *r, uint32_t to, uint32_t id, bool ok, const char *text) { relay_reply(r, (replyto){true, to, id, {0}}, ok, text); }
355
356/* ---- links: install, unlink, drop */
357
358static void unlink_member(mpr_relay *r, uint32_t k);
359
360static void free_linkrec(linkrec *lr) {
361 if (!lr) return;
362 mpn_link_close(lr->l);
363 while (atomic_load(&lr->refs) > 0) {
364 struct timespec ts = {0, 100000};
365 nanosleep(&ts, NULL);
366 }
367 mpn_link_free(lr->l);
368 pthread_mutex_destroy(&lr->sent_mu);
369 free(lr);
370}
371
372static void install(mpr_relay *r, event *e) {
373 uint32_t from = e->from;
374 pthread_mutex_lock(&r->links_mu);
375 linkrec *old = r->links[from];
376 r->links[from] = NULL;
377 pthread_mutex_unlock(&r->links_mu);
378 if (old) {
379 relay_log(r, false, mt_fmt("peer: system %u linked again; dropping its old link", from));
380 mpn_link_close(old->l);
381 unlink_member(r, from);
382 free_linkrec(old);
383 }
384 linkrec *lr = mt_alloc(sizeof *lr);
385 lr->r = r;
386 lr->from = from;
387 lr->lid = ++r->next_lid;
388 uint64_t up = r->h.uptime_ms(r->h.ctx);
389 atomic_init(&lr->last_rx, up > 1 ? up : 1);
390 atomic_init(&lr->last_round, 0);
391 atomic_init(&lr->refs, 0);
392 pthread_mutex_init(&lr->sent_mu, NULL);
393 /* The link's local address (the route to this member). */
394 struct sockaddr_storage la;
395 socklen_t ll = sizeof la;
396 char local[64] = "?";
397 if (getsockname(e->fd, (struct sockaddr *)&la, &ll) == 0) {
398 char host[48] = "?", serv[16] = "?";
399 if (getnameinfo((struct sockaddr *)&la, ll, host, sizeof host, serv, sizeof serv, NI_NUMERICHOST | NI_NUMERICSERV) == 0) snprintf(local, sizeof local, "%s:%s", host, serv);
400 }
401 r->h.host_str(r->h.ctx, mesh_link(from) + MESH_L_LOCAL, 32, local);
402 /* Both JOINs (the handshake ran before the link existed). */
403 if (r->h.net_count) {
404 r->h.net_count(r->h.ctx, from, true, MESH_F_JOIN, 1 + MESH_JOIN_BYTES);
405 r->h.net_count(r->h.ctx, from, false, MESH_F_JOIN, 1 + MESH_JOIN_BYTES);
406 }
407 const mpk_caps *c = &e->join.caps;
408 if (mesh_backend_kind(c->backend) == MESH_BACKEND_CUDA)
409 relay_log(r, false, mt_fmt("peer: linked to system %u (%s, %s, cc %u.%u, %u of %u SMs, %u blocks/SM, %s)", from, c->host, c->gpu, c->cc / 10, c->cc % 10, c->sms,
410 c->sms_total, c->blocks_per_sm, c->arch));
411 else relay_log(r, false, mt_fmt("peer: linked to system %u (%s, %s, backend %u, %u cores, %s)", from, c->host, c->gpu, c->backend, c->sms_total, c->arch));
412 publish_member(r, from, MESH_MS_LINKED, c);
413 r->caps[from] = *c;
414 r->has_caps[from] = true;
415 mpn_sink sink = {lr, on_frame, on_closed, on_count};
416 lr->l = mpn_link_start(e->fd, from, lr->lid, (uint32_t)r->tun.dead_ms, &sink, e->alive);
417 pthread_mutex_lock(&r->links_mu);
418 r->links[from] = lr;
419 pthread_mutex_unlock(&r->links_mu);
420 r->h.host_u32(r->h.ctx, MESH_S_LINKED, relay_linked_mask(r));
421 apply(r, mfc_link(r->core, from, true, relay_now(r)));
422 /* It hears from us at once: our beat and, if it is in our group, our HELLO. */
423 mfc_msg b = mfc_beat(r->core);
424 mpk_beat_status st;
425 relay_my_status(r, &st);
426 mpw_bytes f = mpk_beat_frame(&b, &st);
427 relay_send(r, from, &f);
428 mpw_bytes_free(&f);
429 mpk_beat_status_free(&st);
430 if (r->has_hello && (r->hello.mask >> from & 1) && atomic_load(&r->hello_state) == MESH_GL_HELLO) {
431 mpw_bytes h = mpk_hello_frame(r->me, &r->hello);
432 relay_send(r, from, &h);
433 mpw_bytes_free(&h);
434 }
435}
436
437/* Forget a member's link-scoped state (its HELLO, forwarded requests, image fetches). */
438static void unlink_member(mpr_relay *r, uint32_t k) {
439 r->has_theirs[k] = false;
440 for (size_t i = 0; i < 64; i++)
441 if (r->fwd[i].used && r->fwd[i].leader == k) {
442 r->fwd[i].used = false;
443 if (r->fwd[i].reply.fn) {
444 char *t = mt_fmt("peer: the leader (system %u) was lost before it answered; the change may or may not have committed (`fleet` shows the version)", k);
445 r->fwd[i].reply.fn(r->fwd[i].reply.ctx, false, t);
446 free(t);
447 }
448 }
449 for (size_t i = 0; i < 32; i++)
450 if (r->images[i].used && r->images[i].from == k) {
451 r->images[i].used = false;
452 waiter_done(r->images[i].w, false, mt_fmt("system %u was lost before it sent the image", k));
453 }
454 sched_view_timeouts(r, true, k);
455 apply(r, mfc_link(r->core, k, false, relay_now(r)));
456 /* The group: a member lost pauses it here at once (the leader commits it paused). */
457 uint32_t hs = atomic_load(&r->hello_state);
458 if (r->has_hello && (r->hello.mask >> k & 1) && (hs == MESH_GL_RUNNING || hs == MESH_GL_HELLO)) {
459 char *why = mt_fmt("system %u lost", k);
460 group_pause(r, why);
461 free(why);
462 }
463}
464
465static void drop_link(mpr_relay *r, uint32_t from, const char *why) {
466 pthread_mutex_lock(&r->links_mu);
467 linkrec *lr = r->links[from];
468 r->links[from] = NULL;
469 pthread_mutex_unlock(&r->links_mu);
470 if (!lr) return;
471 mpn_link_close(lr->l);
472 relay_log(r, true, mt_fmt("peer: link to system %u down (%s)", from, why));
473 publish_member(r, from, MESH_MS_ABSENT, NULL);
474 r->h.host_u32(r->h.ctx, MESH_S_LINKED, relay_linked_mask(r));
475 unlink_member(r, from);
476 free_linkrec(lr);
477}
478
479static bool current(mpr_relay *r, uint32_t from, uint64_t lid) {
480 pthread_mutex_lock(&r->links_mu);
481 bool yes = from < MAXS && r->links[from] && r->links[from]->lid == lid;
482 pthread_mutex_unlock(&r->links_mu);
483 return yes;
484}
485
486/* ---- images */
487
488static void serve_image(mpr_relay *r, uint32_t k, uint32_t id, const char *name, uint64_t hash, const char *dir) {
489 mpr_payload p = {0};
490 char *err = NULL;
491 mpw_bytes f = {0};
492 uint8_t t = MESH_F_IMAGE;
493 mpw_put(&f, &t, 1);
494 mpw_put32(&f, id);
495 uint8_t hb[8];
496 memcpy(hb, &hash, 8);
497 mpw_put(&f, hb, 8);
498 if (r->h.image(r->h.ctx, dir, name, hash, &p, &err)) {
499 size_t dl = strlen(p.dir), ml = strlen(p.manifest);
500 mpw_put32(&f, 1);
501 mpw_put32(&f, (uint32_t)dl);
502 mpw_put32(&f, (uint32_t)ml);
503 mpw_put32(&f, (uint32_t)p.ptx.n);
504 mpw_put(&f, p.dir, dl);
505 mpw_put(&f, p.manifest, ml);
506 mpw_put(&f, p.ptx.b, p.ptx.n);
507 mpr_payload_free(&p);
508 } else {
509 mpw_put32(&f, 0);
510 mpw_put32(&f, 0);
511 mpw_put32(&f, (uint32_t)strlen(err));
512 mpw_put32(&f, 0);
513 mpw_put(&f, err, strlen(err));
514 free(err);
515 }
516 if (f.n > MESH_F_LOAD_MAX + MESH_IMAGE_HEAD + 1) relay_log(r, true, mt_fmt("peer: image %s for system %u is larger than F_LOAD_MAX", dir, k));
517 else relay_send(r, k, &f);
518 mpw_bytes_free(&f);
519}
520
521static void fetch_image(mpr_relay *r, const char *name, uint64_t hash, const char *dir, waiter *w) {
522 uint32_t l, from = UINT32_MAX;
523 if (mfc_leader(r->core, &l) && l != r->me && relay_linked(r, l)) from = l;
524 else
525 for (uint32_t k = 0; k < r->n && from == UINT32_MAX; k++)
526 if (k != r->me && relay_linked(r, k)) from = k;
527 if (from == UINT32_MAX) {
528 waiter_done(w, false, mt_strdup("no member is linked to fetch it from"));
529 return;
530 }
531 size_t slot = 0;
532 while (slot < 32 && r->images[slot].used) slot++;
533 if (slot == 32) {
534 waiter_done(w, false, mt_strdup("too many image fetches in flight"));
535 return;
536 }
537 uint32_t id = r->next_id++;
538 size_t dl = strlen(dir) > 64 ? 64 : strlen(dir);
539 mpw_bytes f = {0};
540 uint8_t t = MESH_F_IMAGE_REQ, nm[MESH_NAME_BYTES], hb[8];
541 mpw_put(&f, &t, 1);
542 mpw_put32(&f, id);
543 mpw_name16(name, nm);
544 mpw_put(&f, nm, sizeof nm);
545 memcpy(hb, &hash, 8);
546 mpw_put(&f, hb, 8);
547 mpw_put32(&f, (uint32_t)dl);
548 mpw_put(&f, dir, dl);
549 relay_send(r, from, &f);
550 mpw_bytes_free(&f);
551 r->images[slot] = (image_wait){true, id, w, from, mpr_mono_ms() + IMAGE_WAIT_MS};
552}
553
554static void image_reply(mpr_relay *r, uint32_t id, const uint8_t *b, size_t n) {
555 for (size_t i = 0; i < 32; i++) {
556 if (!r->images[i].used || r->images[i].id != id) continue;
557 r->images[i].used = false;
558 waiter *w = r->images[i].w;
559 size_t dl = mpw_u32_at(b, 16), ml = mpw_u32_at(b, 20), pl = mpw_u32_at(b, 24);
560 const uint8_t *body = b + MESH_IMAGE_HEAD;
561 if (MESH_IMAGE_HEAD + dl + ml + pl > n) {
562 waiter_done(w, false, mt_strdup("F_IMAGE: short"));
563 return;
564 }
565 if (mpw_u32_at(b, 12)) {
566 w->payload.dir = mt_strndup((const char *)body, dl);
567 w->payload.manifest = mt_strndup((const char *)body + dl, ml);
568 mpw_put(&w->payload.ptx, body + dl + ml, pl);
569 waiter_done(w, true, NULL);
570 } else waiter_done(w, false, mt_strndup((const char *)body + dl, ml));
571 return;
572 }
573}
574
575/* ---- changes */
576
577static void change(mpr_relay *r, char *line, mpr_reply reply) {
578 uint32_t l;
579 bool hl = mfc_leader(r->core, &l);
580 /* A job request here, and we lead: the jobs thread (another member's leader gets it by F_CTL). */
581 if (mt_starts(line, "job ") && mfc_role_of(r->core) == MFC_LEADER) {
582 r->h.jobs_ctl(r->h.ctx, line, false, 0, 0, reply);
583 free(line);
584 return;
585 }
586 if (hl && l == r->me && mfc_role_of(r->core) == MFC_LEADER) {
587 r->queue_line = mt_realloc(r->queue_line, (r->nqueue + 1) * sizeof *r->queue_line);
588 r->queue_to = mt_realloc(r->queue_to, (r->nqueue + 1) * sizeof *r->queue_to);
589 r->queue_line[r->nqueue] = line;
590 r->queue_to[r->nqueue++] = (replyto){.local = reply};
591 return;
592 }
593 if (hl && relay_linked(r, l)) {
594 size_t slot = 0;
595 while (slot < 64 && r->fwd[slot].used) slot++;
596 uint32_t id = r->next_id++;
597 size_t n = strlen(line) > MESH_F_CTL_MAX ? MESH_F_CTL_MAX : strlen(line);
598 mpw_bytes f = {0};
599 uint8_t t = MESH_F_CTL;
600 mpw_put(&f, &t, 1);
601 mpw_put32(&f, id);
602 mpw_put32(&f, (uint32_t)n);
603 mpw_put(&f, line, n);
604 relay_send(r, l, &f);
605 mpw_bytes_free(&f);
606 if (slot < 64) r->fwd[slot] = (fwd_t){true, id, reply, l, mpr_mono_ms() + FORWARD_WAIT_MS};
607 free(line);
608 return;
609 }
610 char *lw = leader_word(r), *q = rf_debug_str(line);
611 char *e = mt_fmt("peer: %s; the desired state is unchanged (%s refused)", lw, q);
612 relay_log(r, true, mt_fmt("fleet: %s", e));
613 if (reply.fn) reply.fn(reply.ctx, false, e);
614 free(lw), free(q), free(e), free(line);
615}
616
617/* The leader's catalog view of an image (its own catalog). */
618static bool catalog(void *ctx, const char *dir, mfe_image *out, char **err) {
619 mpr_relay *r = ctx;
620 return r->h.catalog(r->h.ctx, dir, out, err);
621}
622
623static bool backend_of(mpr_relay *r, uint32_t k, uint32_t *kind) {
624 if (k == r->me) return *kind = mesh_backend_kind(r->my_caps.backend), true;
625 if (!r->has_caps[k]) return false;
626 *kind = mesh_backend_kind(r->caps[k].backend);
627 return true;
628}
629
630static ml_list eligible_up(void *ctx, const mt_strs *backends) {
631 mpr_relay *r = ctx;
632 uint32_t alive = mfc_alive(r->core, relay_now(r)), kind;
633 ml_list l = {0};
634 for (uint32_t k = 0; k < r->n; k++)
635 if ((alive >> k & 1) && backend_of(r, k, &kind) && mfd_can_run(backends, kind)) ml_push(&l, k);
636 return l;
637}
638
639bool relay_propose(mpr_relay *r, mfd_doc doc, char *what, replyto to, uint64_t *version) {
640 uint64_t now = relay_now(r), v;
641 mfd_doc before = mfd_doc_clone(mfc_committed(r->core));
642 mfc_outs outs;
643 char *err = NULL;
644 if (mfc_propose(r->core, doc, now, &v, &outs, &err)) {
645 relay_log(r, false, mt_fmt("fleet: proposing version %llu: %s", (unsigned long long)v, what));
646 r->has_pending = true;
647 r->pending = (pending_t){.what = what, .reply = to, .version = v, .before = before};
648 apply(r, outs);
649 if (version) *version = v;
650 return true;
651 }
652 char *e = mt_fmt("peer: %s; the desired state is unchanged", err);
653 relay_log(r, true, mt_fmt("fleet: %s: %s", what, e));
654 relay_reply(r, to, false, e);
655 free(e), free(err), free(what);
656 mfd_doc_free(&before);
657 return false;
658}
659
660static void pending_free(pending_t *p) {
661 free(p->what);
662 mfd_doc_free(&p->before);
663 mfd_doc_free(&p->after);
664}
665
666/* A proposal ended (the core). */
667static void done(mpr_relay *r, uint64_t v, bool ok, const char *err) {
668 uint32_t alive = mfc_alive(r->core, relay_now(r));
669 if (!r->has_pending || r->pending.version != v) return;
670 pending_t p = r->pending;
671 r->has_pending = false;
672 if (ok) {
673 p.committed = true;
674 p.up = alive;
675 p.deadline = mpr_mono_ms() + CHANGE_WAIT_MS;
676 p.after = mfd_doc_clone(mfc_committed(r->core));
677 r->waiting = mt_realloc(r->waiting, (r->nwaiting + 1) * sizeof *r->waiting);
678 r->waiting[r->nwaiting++] = p;
679 return;
680 }
681 relay_log(r, true, mt_fmt("fleet: version %llu (%s) not committed: %s", (unsigned long long)v, p.what, err));
682 char *t = mt_fmt("peer: %s; the change may still apply if a later leader holds it (`fleet` shows the version)", err);
683 relay_reply(r, p.reply, false, t);
684 free(t);
685 pending_free(&p);
686}
687
688/* A waiting change's answer once every member that was up reconciled it (or CHANGE_WAIT passed), and
689 * whether to revert it (nothing applied it anywhere). */
690static bool waiting_verdict(mpr_relay *r, const pending_t *p, bool *ok, char **text, bool *revert) {
691 if (!p->committed) return false;
692 uint64_t v = p->version;
693 mpr_recon my = {0};
694 r->h.recon(r->h.ctx, &my);
695 uint32_t alive = mfc_alive(r->core, relay_now(r));
696 bool all = true;
697 for (uint32_t k = 0; k < r->n; k++) {
698 if (!((p->up & alive) >> k & 1)) continue;
699 uint64_t applied = k == r->me ? my.applied : r->st[k].applied;
700 if (applied < v) all = false;
701 }
702 if (!all && mpr_mono_ms() < p->deadline) {
703 mpr_recon_free(&my);
704 return false;
705 }
706 mt_strs parts = {0};
707 int good = 0, failed = 0;
708 for (uint32_t k = 0; k < r->n; k++) {
709 uint64_t applied = k == r->me ? my.applied : r->st[k].applied;
710 uint32_t st = k == r->me ? my.status : r->st[k].status;
711 const char *why = k == r->me ? (my.reason ? my.reason : "") : (r->st[k].reason ? r->st[k].reason : "");
712 if (!(p->up >> k & 1)) mt_push_own(&parts, mt_fmt("system %u absent: reconcile on join", k));
713 else if (applied < v && !(alive >> k & 1)) mt_push_own(&parts, mt_fmt("system %u lost before it reconciled: reconcile on join", k));
714 else if (applied < v) mt_push_own(&parts, mt_fmt("system %u pending: not reconciled in 60s", k));
715 else if (st == MESH_RS_FAILED) failed++, mt_push_own(&parts, mt_fmt("system %u err %s", k, why));
716 else {
717 good++;
718 if (!*why || !strcmp(why, "ok")) mt_push_own(&parts, mt_fmt("system %u ok", k));
719 else mt_push_own(&parts, mt_fmt("system %u %s", k, why));
720 }
721 }
722 mpr_recon_free(&my);
723 char *detail = mt_join(&parts, "; ");
724 mt_strs_free(&parts);
725 if (good == 0 && failed > 0 && !mt_starts(p->what, "revert")) {
726 *ok = false;
727 *text = mt_fmt("peer err (%s): %s; reverted", detail, p->what);
728 *revert = true;
729 } else {
730 *ok = failed == 0;
731 *text = mt_fmt("version %llu %s: %s (%s)", (unsigned long long)v, failed == 0 ? "peer ok" : "peer err", p->what, detail);
732 *revert = false;
733 }
734 free(detail);
735 return true;
736}
737
738/* Committed changes: each answered once every member that was up reconciled it (or after CHANGE_WAIT). */
739static void pending_check(mpr_relay *r) {
740 size_t i = 0;
741 while (i < r->nwaiting) {
742 bool ok, revert;
743 char *text;
744 if (!waiting_verdict(r, &r->waiting[i], &ok, &text, &revert)) {
745 i++;
746 continue;
747 }
748 bool leading = mfc_role_of(r->core) == MFC_LEADER;
749 /* A revert waits until nothing else is in flight here (nor before this term committed its first document). */
750 if (revert && leading && (r->has_pending || mfc_busy(r->core) || !mfc_term_committed(r->core))) {
751 free(text);
752 i++;
753 continue;
754 }
755 pending_t p = r->waiting[i];
756 memmove(r->waiting + i, r->waiting + i + 1, (r->nwaiting - i - 1) * sizeof *r->waiting);
757 r->nwaiting--;
758 if (revert && !leading) {
759 char *t = mt_fmt("%s -- not reverted: system %u no longer leads", text, r->me);
760 relay_reply(r, p.reply, ok, t);
761 free(t);
762 } else if (revert) {
763 relay_log(r, true, mt_fmt("fleet: version %llu (%s) failed on every member that tried it; reverting", (unsigned long long)p.version, p.what));
764 mfd_doc back = mfe_revert(mfc_committed(r->core), &p.before, &p.after);
765 relay_reply(r, p.reply, ok, text);
766 relay_propose(r, back, mt_fmt("revert version %llu", (unsigned long long)p.version), (replyto){0}, NULL);
767 } else {
768 relay_log(r, false, mt_fmt("fleet: %s", text));
769 relay_reply(r, p.reply, ok, text);
770 }
771 free(text);
772 pending_free(&p);
773 }
774}
775
776/* The first leader's programs as the first document. */
777static mfd_doc seed(mpr_relay *r, uint64_t now) {
778 size_t n = 0;
779 mfl_actual *inv = r->h.inventory(r->h.ctx, &n);
780 char **names = mt_alloc((n ? n : 1) * sizeof *names);
781 mfd_prog *specs = mt_alloc((n ? n : 1) * sizeof *specs);
782 size_t k = 0;
783 for (size_t i = 0; i < n; i++) {
784 mfe_image im;
785 char *err = NULL;
786 if (!r->h.catalog(r->h.ctx, inv[i].dir, &im, &err)) {
787 relay_log(r, true, mt_fmt("fleet: seeding: %s: %s; left out", inv[i].name, err));
788 free(err);
789 continue;
790 }
791 if (im.hash != inv[i].hash) {
792 relay_log(r, true, mt_fmt("fleet: seeding: %s runs %016llx here but the catalog's %s is %016llx; left out", inv[i].name, (unsigned long long)inv[i].hash,
793 inv[i].dir, (unsigned long long)im.hash));
794 mfe_image_free(&im);
795 continue;
796 }
797 mfd_prog p = mfd_prog_new(inv[i].dir, inv[i].hash, inv[i].run, &im.needs);
798 p.lockstep = im.lockstep;
799 p.backends = mt_strs_clone(&im.backends);
800 names[k] = inv[i].name;
801 specs[k++] = p;
802 mfe_image_free(&im);
803 }
804 ml_list up = ml_members(mfc_alive(r->core, now));
805 mfd_doc d = mfl_seed(names, specs, k, &up, NULL, NULL);
806 /* mfl_seed's backend lookup: the members' caps (none -> u32::MAX). */
807 for (size_t gi = 0; gi < d.ngroups; gi++) {
808 mfd_group *g = &d.groups[gi];
809 const mfd_prog *p = mfd_prog_get(&d, g->program);
810 ml_free(&g->members);
811 for (size_t j = 0; j < up.n; j++) {
812 uint32_t kind = UINT32_MAX;
813 backend_of(r, up.v[j], &kind);
814 if (mfd_can_run(&p->backends, kind)) ml_push(&g->members, up.v[j]);
815 }
816 }
817 mt_strs ns = {0};
818 for (size_t i = 0; i < d.nprogs; i++) mt_push(&ns, d.pname[i]);
819 char *j = mt_join(&ns, ", ");
820 relay_log(r, false, mt_fmt("fleet: seeding the desired state from system %u's programs: %s", r->me, j));
821 free(j);
822 mt_strs_free(&ns);
823 for (size_t i = 0; i < k; i++) mfd_prog_free(&specs[i]);
824 free(specs), free(names);
825 for (size_t i = 0; i < n; i++) mfl_actual_free(&inv[i]);
826 free(inv);
827 ml_free(&up);
828 return d;
829}
830
831/* The leader's policies: lockstep groups (pause, resume, auto-reform), then the scheduler. */
832static void policies(mpr_relay *r, uint64_t now) {
833 const mfd_doc *doc = mfc_committed(r->core);
834 if (doc->version == 0) return;
835 uint32_t alive = mfc_alive(r->core, now);
836 for (size_t gi = 0; gi < doc->ngroups; gi++) {
837 const mfd_group *g = &doc->groups[gi];
838 /* A member whose data plane paused this epoch (it lost a member we may still hear) counts as missing. */
839 uint32_t broken = 0;
840 for (size_t i = 0; i < g->members.n; i++) {
841 uint32_t k = g->members.v[i];
842 uint32_t e = k == r->me ? (r->has_hello ? r->hello.epoch : 0) : r->st[k].group_epoch;
843 uint32_t s = k == r->me ? atomic_load(&r->hello_state) : r->st[k].group_state;
844 if ((alive >> k & 1) && e == g->epoch && s == MESH_GL_PAUSED) broken |= 1u << k;
845 }
846 uint32_t up = g->state == MFD_RUNNING ? alive & ~broken : alive;
847 char tv[24];
848 if (g->state == MFD_PAUSED) {
849 if (!ms_get(&r->paused_since, g->name)) {
850 snprintf(tv, sizeof tv, "%llu", (unsigned long long)now);
851 ms_set(&r->paused_since, g->name, tv, true);
852 }
853 } else ms_del(&r->paused_since, g->name);
854 const char *ps = ms_get(&r->paused_since, g->name);
855 int64_t paused_ms = ps ? (int64_t)(now - strtoull(ps, NULL, 10)) : -1;
856 mfd_group ng;
857 if (!mfg_policy(g, up, paused_ms, &ng)) continue;
858 char *what, *ml = ml_text(&g->members);
859 if (g->state == MFD_RUNNING && ng.state == MFD_PAUSED) {
860 ml_list al = ml_members(alive), bl = ml_members(broken);
861 char *at = ml_text(&al), *bt = ml_text(&bl);
862 char *extra = broken ? mt_fmt("; paused here by %s", bt) : mt_strdup("");
863 what = mt_fmt("group %s paused (members %s; up %s%s)", g->name, ml, at, extra);
864 free(at), free(bt), free(extra);
865 ml_free(&al), ml_free(&bl);
866 } else if (ml_eq(&ng.members, &g->members)) what = mt_fmt("group %s resumed: epoch %u (every member back; reset)", g->name, ng.epoch);
867 else {
868 char *nm = ml_text(&ng.members);
869 what = mt_fmt("group %s re-formed with the survivors after %lld ms: epoch %u members %s", g->name, (long long)(paused_ms < 0 ? 0 : paused_ms), ng.epoch, nm);
870 free(nm);
871 }
872 free(ml);
873 mfd_doc d = mfd_doc_clone(doc);
874 mfd_group *x = mfd_group_named(&d, g->name);
875 mfd_group_free(x);
876 *x = ng;
877 relay_propose(r, d, what, (replyto){0}, NULL);
878 return;
879 }
880 /* The scheduler: placement, migrations, rebalancing, group proposals. */
881 sched_schedule(r, now);
882}
883
884static void lead(mpr_relay *r, uint64_t now) {
885 if (mfc_busy(r->core) || r->has_pending) return;
886 /* The first document of this leadership: the seed (a fresh fleet) or the accepted one re-stamped. */
887 if (!mfc_term_committed(r->core)) {
888 if (r->has_first_doc_at && mpr_mono_ms() - r->first_doc_at_ms < FIRST_DOC_RETRY_MS) return;
889 r->has_first_doc_at = true;
890 r->first_doc_at_ms = mpr_mono_ms();
891 mfd_doc d;
892 char *what;
893 if (mfc_first_doc(r->core, &d)) {
894 what = mt_fmt("version %llu (term %llu) re-committed in term %llu", (unsigned long long)d.version, (unsigned long long)d.term, (unsigned long long)mfc_term(r->core));
895 } else {
896 mfd_doc s = seed(r, now), empty = {0};
897 mt_strs ch;
898 d = sched_edit(r, &empty, s, now, &ch);
899 mt_strs_free(&ch);
900 what = mt_strdup("the first desired state (the first leader's programs)");
901 }
902 relay_propose(r, d, what, (replyto){0}, NULL);
903 return;
904 }
905 if (r->nqueue) {
906 char *line = r->queue_line[0];
907 replyto to = r->queue_to[0];
908 memmove(r->queue_line, r->queue_line + 1, (r->nqueue - 1) * sizeof *r->queue_line);
909 memmove(r->queue_to, r->queue_to + 1, (r->nqueue - 1) * sizeof *r->queue_to);
910 r->nqueue--;
911 mfe_req req;
912 char *err = NULL;
913 if (!mfe_parse(line, &req, &err)) {
914 relay_reply(r, to, false, err);
915 free(err), free(line);
916 return;
917 }
918 free(line);
919 mfd_doc base = mfd_doc_clone(mfc_committed(r->core)), doc;
920 mfe_ctx cx = {catalog, eligible_up, r, r->n};
921 char *what = NULL;
922 if (mfe_edit(&base, &req, &cx, &doc, &what, &err)) {
923 /* Placed by the scheduler in the same version (a load lands where it fits; a group gets its members). */
924 mt_strs ch;
925 mfd_doc placed = sched_edit(r, &base, doc, now, &ch);
926 if (ch.n) {
927 char *j = mt_join(&ch, "; "), *w2 = mt_fmt("%s; sched: %s", what, j);
928 free(j), free(what);
929 what = w2;
930 }
931 mt_strs_free(&ch);
932 relay_propose(r, placed, what, to, NULL);
933 } else {
934 relay_reply(r, to, false, err);
935 free(err);
936 }
937 mfe_req_free(&req);
938 mfd_doc_free(&base);
939 return;
940 }
941 policies(r, now);
942}
943
944/* ---- the driver's view */
945
946static void send_view(mpr_relay *r) {
947 uint64_t now = relay_now(r);
948 uint32_t up = mfc_alive(r->core, now), l;
949 bool hl = mfc_leader(r->core, &l);
950 r->h.jobs_view(r->h.ctx, mfc_term(r->core), hl, l, up, relay_linked_mask(r));
951 const mfd_doc *c = mfc_committed(r->core);
952 mt_buf pb = {0};
953 for (size_t i = 0; i < c->nplaces; i++) {
954 char *w = mt_join(&c->place_words[i], " ");
955 mt_printf(&pb, "%s %s\n", c->place_name[i], w);
956 free(w);
957 }
958 char *places = mt_take(&pb);
959 bool same = r->has_view && r->view_up == up && r->view_has_leader == hl && (!hl || r->view_leader == l) && !strcmp(r->view_places, places);
960 if (!same) {
961 r->has_view = true;
962 r->view_up = up, r->view_has_leader = hl, r->view_leader = hl ? l : 0;
963 free(r->view_places);
964 r->view_places = places;
965 r->h.driver_view(r->h.ctx, up, hl, l, c);
966 r->h.peer_state(r->h.ctx, hl ? MPR_PEER_UP : MPR_PEER_WAITING);
967 } else free(places);
968}
969
970/* ---- periodic work */
971
972static void timeouts(mpr_relay *r) {
973 uint64_t t = mpr_mono_ms();
974 for (size_t i = 0; i < 64; i++)
975 if (r->fwd[i].used && t >= r->fwd[i].deadline) {
976 r->fwd[i].used = false;
977 if (r->fwd[i].reply.fn) {
978 char *e = mt_fmt("peer: the leader (system %u) did not answer in 70s; `fleet` shows whether it committed", r->fwd[i].leader);
979 r->fwd[i].reply.fn(r->fwd[i].reply.ctx, false, e);
980 free(e);
981 }
982 }
983 for (size_t i = 0; i < 32; i++)
984 if (r->images[i].used && t >= r->images[i].deadline) {
985 r->images[i].used = false;
986 waiter_done(r->images[i].w, false, mt_fmt("system %u did not send the image in 30s", r->images[i].from));
987 }
988 sched_view_timeouts(r, false, 0);
989}
990
991static void tick(mpr_relay *r) {
992 uint64_t now = relay_now(r);
993 apply(r, mfc_tick(r->core, now));
994 if (mpr_mono_ms() - r->last_beat_ms >= r->tun.ping_ms) relay_beat_all(r);
995 /* The leader's work: its first document, queued changes, the policies. */
996 if (mfc_role_of(r->core) == MFC_LEADER) lead(r, now);
997 pending_check(r);
998 timeouts(r);
999 group_want(r);
1000 if (atomic_load(&r->hello_state) == MESH_GL_HELLO && mpr_mono_ms() - r->hello_sent_ms >= HELLO_RESEND_MS) group_hello_send(r);
1001 send_view(r);
1002 if (mpr_mono_ms() - r->last_publish_ms >= 100) publish_all(r, now);
1003 uint32_t l;
1004 atomic_store(&r->pub_version, mfc_committed(r->core)->version);
1005 atomic_store(&r->pub_leader, mfc_leader(r->core, &l) ? l : MESH_PEER_NONE);
1006 atomic_store(&r->pub_role, mfc_role_code(mfc_role_of(r->core)));
1007 atomic_store(&r->pub_alive, mfc_alive(r->core, now));
1008}
1009
1010/* ---- one event */
1011
1012static void frame_event(mpr_relay *r, event *e) {
1013 uint32_t from = e->from;
1014 const uint8_t *b = e->frame.b + 1;
1015 size_t n = e->frame.n - 1;
1016 uint64_t now = relay_now(r);
1017 char *err = NULL;
1018 switch (e->frame.b[0]) {
1019 case MESH_F_HELLO: {
1020 uint32_t sys, ver;
1021 mpk_hello h;
1022 if (!mpk_hello_decode(b, &sys, &ver, &h, &err)) break;
1023 if (sys != from) err = mt_fmt("HELLO from system %u on system %u's link", sys, from);
1024 else if (ver != MESH_PEER_VERSION) err = mt_fmt("HELLO of peer-link version %u", ver);
1025 else group_hello(r, from, &h);
1026 break;
1027 }
1028 case MESH_F_BEAT: {
1029 uint32_t rn = mpw_u32_at(b, MESH_BEAT_HEAD - 4);
1030 uint32_t k = mpw_u32_at(b, MESH_BEAT_HEAD + rn);
1031 mfc_msg m;
1032 mpk_beat_status st;
1033 if (!mpk_beat_decode(b, b + MESH_BEAT_HEAD, rn, b + MESH_BEAT_HEAD + rn + 4, (size_t)k * MESH_BEAT_MOVE_BYTES, &m, &st, &err)) break;
1034 if (st.status == MESH_RS_FAILED && (st.status != r->seen_status[from] || !mt_eq(st.reason, r->seen_reason[from])))
1035 relay_log(r, true, mt_fmt("fleet: system %u failed to reconcile version %llu: %s", from, (unsigned long long)st.applied, st.reason));
1036 r->seen_status[from] = st.status;
1037 mt_set(&r->seen_reason[from], st.reason);
1038 mpk_beat_status_free(&r->st[from]);
1039 r->st[from] = st;
1040 apply(r, mfc_recv(r->core, from, m, now));
1041 break;
1042 }
1043 case MESH_F_VOTE_REQ: apply(r, mfc_recv(r->core, from, mpk_vote_req_decode(b), now)); break;
1044 case MESH_F_VOTE: apply(r, mfc_recv(r->core, from, mpk_vote_decode(b), now)); break;
1045 case MESH_F_STATE_ACK:
1046 case MESH_F_COMMIT: apply(r, mfc_recv(r->core, from, mpk_ack_decode(e->frame.b[0], b), now)); break;
1047 case MESH_F_STATE: {
1048 uint32_t len = mpw_u32_at(b, 8);
1049 char *t = mt_strndup((const char *)b + MESH_STATE_HEAD, len);
1050 mfd_doc d;
1051 if (strlen(t) != len) err = mt_strdup("F_STATE: not UTF-8");
1052 else if (mfd_decode(t, &d, &err)) apply(r, mfc_recv(r->core, from, (mfc_msg){.kind = MFC_STATE, .term = mpw_u64_at(b, 0), .doc = d}, now));
1053 free(t);
1054 break;
1055 }
1056 case MESH_F_CTL: {
1057 uint32_t id = mpw_u32_at(b, 0), len = mpw_u32_at(b, 4);
1058 char *line = mt_strndup((const char *)b + MESH_CTL_HEAD, len);
1059 replyto to = {true, from, id, {0}};
1060 if (mt_starts(line, "job ")) {
1061 /* A job request forwarded to us: the jobs thread answers it (F_CTL_REPLY), not the desired state's queue. */
1062 r->h.jobs_ctl(r->h.ctx, line, true, from, id, (mpr_reply){0});
1063 free(line);
1064 } else if (mfc_role_of(r->core) == MFC_LEADER) {
1065 r->queue_line = mt_realloc(r->queue_line, (r->nqueue + 1) * sizeof *r->queue_line);
1066 r->queue_to = mt_realloc(r->queue_to, (r->nqueue + 1) * sizeof *r->queue_to);
1067 r->queue_line[r->nqueue] = line;
1068 r->queue_to[r->nqueue++] = to;
1069 } else {
1070 char *lw = leader_word(r), *t = mt_fmt("peer: system %u is not the leader (%s)", r->me, lw);
1071 relay_reply(r, to, false, t);
1072 free(lw), free(t), free(line);
1073 }
1074 break;
1075 }
1076 case MESH_F_CTL_REPLY: {
1077 uint32_t id = mpw_u32_at(b, 0), ok = mpw_u32_at(b, 4), len = mpw_u32_at(b, 8);
1078 for (size_t i = 0; i < 64; i++)
1079 if (r->fwd[i].used && r->fwd[i].id == id) {
1080 r->fwd[i].used = false;
1081 char *t = mt_strndup((const char *)b + MESH_CTL_REPLY_HEAD, len);
1082 if (r->fwd[i].reply.fn) r->fwd[i].reply.fn(r->fwd[i].reply.ctx, ok != 0, t);
1083 free(t);
1084 }
1085 break;
1086 }
1087 case MESH_F_IMAGE_REQ: {
1088 char name[MESH_NAME_BYTES + 1];
1089 if (!mpw_name_of(b + 4, MESH_NAME_BYTES, name, &err)) break;
1090 uint32_t dl = mpw_u32_at(b, MESH_IMAGE_REQ_HEAD - 4);
1091 char *dir = mt_strndup((const char *)b + MESH_IMAGE_REQ_HEAD, dl);
1092 serve_image(r, from, mpw_u32_at(b, 0), name, mpw_u64_at(b, 4 + MESH_NAME_BYTES), dir);
1093 free(dir);
1094 break;
1095 }
1096 case MESH_F_IMAGE: image_reply(r, mpw_u32_at(b, 0), b, n); break;
1097 case MESH_F_VIEW_REQ: {
1098 char name[MESH_NAME_BYTES + 1];
1099 if (!mpw_name_of(b + 4, MESH_NAME_BYTES, name, &err)) break;
1100 sched_serve_view(r, from, mpw_u32_at(b, 0), name, mpw_u64_at(b, 4 + MESH_NAME_BYTES));
1101 break;
1102 }
1103 case MESH_F_VIEW: sched_view_chunk(r, mpw_u32_at(b, 0), b, n); break;
1104 default: break;
1105 }
1106 if (err) {
1107 drop_link(r, from, err);
1108 free(err);
1109 }
1110}
1111
1112static void handle(mpr_relay *r, event *e) {
1113 switch (e->kind) {
1114 case EV_LINK: install(r, e); break;
1115 case EV_CLOSED:
1116 if (current(r, e->from, e->lid)) drop_link(r, e->from, e->text);
1117 break;
1118 case EV_FRAME:
1119 if (current(r, e->from, e->lid)) frame_event(r, e);
1120 break;
1121 case EV_CHANGE:
1122 change(r, e->text, e->reply);
1123 e->text = NULL;
1124 break;
1125 case EV_WAKE:
1126 relay_beat_all(r);
1127 group_want(r);
1128 break;
1129 case EV_FETCH_IMAGE: fetch_image(r, e->name, e->hash, e->dir, e->w); break;
1130 case EV_FETCH_VIEW: sched_fetch_view(r, e->name, e->hash, e->from, e->w); break;
1131 case EV_CTL_REPLY: relay_reply(r, (replyto){true, e->from, e->id, {0}}, e->ok, e->text); break;
1132 }
1133 event_free(e);
1134}
1135
1136static void *relay_main(void *arg) {
1137 mpr_relay *r = arg;
1138 while (!atomic_load(&r->quit)) {
1139 if (atomic_load(&r->fatal)) break;
1140 pthread_mutex_lock(&r->ev_mu);
1141 if (!r->ev_n) {
1142 struct timespec ts;
1143 clock_gettime(CLOCK_REALTIME, &ts);
1144 ts.tv_nsec += 10 * 1000000L;
1145 if (ts.tv_nsec >= 1000000000L) ts.tv_sec++, ts.tv_nsec -= 1000000000L;
1146 pthread_cond_timedwait(&r->ev_cv, &r->ev_mu, &ts);
1147 }
1148 event *batch = r->ev;
1149 size_t nb = r->ev_n;
1150 r->ev = NULL, r->ev_n = r->ev_cap = 0;
1151 pthread_mutex_unlock(&r->ev_mu);
1152 for (size_t i = 0; i < nb; i++) handle(r, &batch[i]);
1153 free(batch);
1154 tick(r);
1155 }
1156 return NULL;
1157}
1158
1159/* ---- the API */
1160
1161mpr_relay *mpr_start(const mcc_topology *topo, const mpk_caps *caps, const char *state_path, const mpr_host *host, char **err) {
1162 mpr_relay *r = mt_alloc(sizeof *r);
1163 r->h = *host;
1164 r->me = topo->me;
1165 r->n = topo->n;
1166 r->tun = topo->tun;
1167 r->my_caps = *caps;
1168 r->w = topo->tun.sched;
1169 mfp_persist p = {0};
1170 if (state_path) {
1171 if (!mfs_open(state_path, r->n, &r->store, &p, err)) {
1172 free(r);
1173 return NULL;
1174 }
1175 r->has_store = true;
1176 }
1177 pthread_mutex_init(&r->ev_mu, NULL);
1178 pthread_cond_init(&r->ev_cv, NULL);
1179 pthread_mutex_init(&r->links_mu, NULL);
1180 pthread_mutex_init(&r->group_mu, NULL);
1181 pthread_mutex_init(&r->send_mu, NULL);
1182 pthread_mutex_init(&r->tag_mu, NULL);
1183 pthread_mutex_init(&r->own_mu, NULL);
1184 board_init(&r->board);
1185 atomic_init(&r->quit, false);
1186 atomic_init(&r->fatal, false);
1187 atomic_init(&r->mirroring, false);
1188 atomic_init(&r->hello_state, MESH_GL_NONE);
1189 atomic_init(&r->last_sent, UINT64_MAX);
1190 atomic_init(&r->pub_version, 0);
1191 atomic_init(&r->pub_leader, MESH_PEER_NONE);
1192 atomic_init(&r->pub_role, 1);
1193 atomic_init(&r->pub_alive, 1u << topo->me);
1194 memset(r->tag.rank_of, 0xff, sizeof r->tag.rank_of);
1195 r->tag.me = 0xff;
1196 r->tracker = mst_new();
1197 r->next_id = 1;
1198 r->t0_ms = mpr_mono_ms();
1199 r->last_beat_ms = r->last_publish_ms = r->last_sched_text_ms = r->hello_sent_ms = mpr_mono_ms();
1200 r->h.peer_state(r->h.ctx, MPR_PEER_WAITING);
1201 publish_member(r, r->me, MESH_MS_UP, caps);
1202 if (p.committed.version > 0) {
1203 relay_log(r, false, mt_fmt("fleet: system %u of %u: term %llu, committed version %llu (term %llu) from --fleet-state; reconciling to it", r->me, r->n,
1204 (unsigned long long)p.term, (unsigned long long)p.committed.version, (unsigned long long)p.committed.term));
1205 r->h.committed(r->h.ctx, &p.committed);
1206 }
1207 uint64_t seed = (uint64_t)getpid() << 20 ^ mpr_mono_us() ^ ((uint64_t)r->me + 1) * 0x9e3779b97f4a7c15ull;
1208 r->core = mfc_new((mfc_config){r->me, r->n, r->tun.dead_ms, seed}, p, 1);
1209 const mfc_config *cfg = mfc_cfg(r->core);
1210 relay_log(r, false, mt_fmt("fleet: system %u of %u: %s; a majority is %u; beats every %llu ms, a member is gone after %llu ms", r->me, r->n,
1211 mfc_fixed(cfg) ? "a pair: system 0 coordinates (no failover)" : "leader by election", mfc_quorum(cfg), (unsigned long long)r->tun.ping_ms,
1212 (unsigned long long)r->tun.dead_ms));
1213 mpn_owner owner = {r, on_linked, on_link_log};
1214 if (topo->listen) {
1215 int fd = mpn_listen(topo->listen, err);
1216 if (fd < 0) {
1217 relay_log(r, true, mt_fmt("peer: %s", *err));
1218 mpr_stop(r);
1219 return NULL;
1220 }
1221 relay_log(r, false, mt_fmt("peer: system %u of %u listening on %s", r->me, r->n, topo->listen));
1222 r->acceptor = mpn_acceptor_start(fd, r->me, r->n, caps, (uint32_t)r->tun.dead_ms, &owner);
1223 }
1224 for (size_t i = 0; i < topo->ndial; i++) {
1225 relay_log(r, false, mt_fmt("peer: system %u dials system %u at %s", r->me, topo->dial_k[i], topo->dial_addr[i]));
1226 r->dialers[r->ndialers++] = mpn_dialer_start(topo->dial_k[i], topo->dial_addr[i], r->me, r->n, caps, (uint32_t)r->tun.dead_ms, &owner);
1227 }
1228 pthread_create(&r->sender, NULL, group_sender, r);
1229 pthread_create(&r->thread, NULL, relay_main, r);
1230 return r;
1231}
1232
1233void mpr_stop(mpr_relay *r) {
1234 if (!r) return;
1235 atomic_store(&r->quit, true);
1236 mpn_acceptor_stop(r->acceptor);
1237 if (r->thread) pthread_join(r->thread, NULL);
1238 if (r->sender) pthread_join(r->sender, NULL);
1239 /* The links before their dialers: closing a link clears its dialer's alive flag. */
1240 for (uint32_t k = 0; k < MAXS; k++) free_linkrec(r->links[k]);
1241 for (size_t i = 0; i < r->ndialers; i++) mpn_dialer_stop(r->dialers[i]);
1242 for (size_t i = 0; i < r->ev_n; i++) {
1243 if (r->ev[i].kind == EV_LINK) close(r->ev[i].fd);
1244 event_free(&r->ev[i]);
1245 }
1246 free(r->ev);
1247 mfc_free(r->core);
1248 if (r->has_store) mfs_close(&r->store);
1249 for (uint32_t k = 0; k < MAXS; k++) {
1250 mpk_beat_status_free(&r->st[k]);
1251 free(r->seen_reason[k]);
1252 }
1253 for (size_t i = 0; i < r->nqueue; i++) free(r->queue_line[i]);
1254 free(r->queue_line), free(r->queue_to);
1255 if (r->has_pending) pending_free(&r->pending);
1256 for (size_t i = 0; i < r->nwaiting; i++) pending_free(&r->waiting[i]);
1257 free(r->waiting);
1258 for (size_t i = 0; i < 32; i++) {
1259 if (r->images[i].used) waiter_done(r->images[i].w, false, mt_strdup("the fleet thread stopped"));
1260 if (r->views[i].used) {
1261 mpv_wait_free(&r->views[i].v);
1262 waiter_done(r->views[i].w, false, mt_strdup("the fleet thread stopped"));
1263 }
1264 }
1265 ms_free(&r->paused_since);
1266 free(r->view_places);
1267 free(r->hello_why);
1268 mst_free(r->tracker);
1269 if (r->has_mig) mst_move_free(&r->mig.mv);
1270 for (size_t i = 0; i < r->nperf_mem; i++) free(r->perf_mem[i].name);
1271 free(r->perf_mem);
1272 for (size_t i = 0; i < r->nfail_mem; i++) free(r->fail_mem[i].name), free(r->fail_mem[i].why);
1273 free(r->fail_mem);
1274 mt_strs_free(&r->sched_log);
1275 free(r->group_prop);
1276 msp_explains_free(&r->explains);
1277 board_free(&r->board);
1278 pthread_mutex_destroy(&r->ev_mu);
1279 pthread_cond_destroy(&r->ev_cv);
1280 pthread_mutex_destroy(&r->links_mu);
1281 pthread_mutex_destroy(&r->group_mu);
1282 pthread_mutex_destroy(&r->send_mu);
1283 pthread_mutex_destroy(&r->tag_mu);
1284 pthread_mutex_destroy(&r->own_mu);
1285 free(r);
1286}
1287
1288void mpr_change(mpr_relay *r, const char *line, mpr_reply reply) { post(r, (event){.kind = EV_CHANGE, .text = mt_strdup(line), .reply = reply}); }
1289void mpr_wake(mpr_relay *r) { post(r, (event){.kind = EV_WAKE}); }
1290
1291static bool wait_for(waiter *w, uint32_t ms, char **err) {
1292 struct timespec ts;
1293 clock_gettime(CLOCK_REALTIME, &ts);
1294 ts.tv_sec += ms / 1000;
1295 ts.tv_nsec += (long)(ms % 1000) * 1000000L;
1296 if (ts.tv_nsec >= 1000000000L) ts.tv_sec++, ts.tv_nsec -= 1000000000L;
1297 pthread_mutex_lock(&w->mu);
1298 while (!w->done)
1299 if (pthread_cond_timedwait(&w->cv, &w->mu, &ts) != 0) break;
1300 bool done = w->done, ok = w->ok;
1301 pthread_mutex_unlock(&w->mu);
1302 if (!done) {
1303 *err = mt_fmt("no answer in %us", ms / 1000);
1304 return false;
1305 }
1306 if (!ok) *err = w->err ? w->err : mt_strdup("failed");
1307 else free(w->err);
1308 return ok;
1309}
1310
1311bool mpr_fetch_image(mpr_relay *r, const char *name, uint64_t hash, const char *dir, uint32_t wait_ms, mpr_payload *out, char **err) {
1312 waiter *w = mt_alloc(sizeof *w);
1313 pthread_mutex_init(&w->mu, NULL);
1314 pthread_cond_init(&w->cv, NULL);
1315 post(r, (event){.kind = EV_FETCH_IMAGE, .name = mt_strdup(name), .hash = hash, .dir = mt_strdup(dir), .w = w});
1316 bool ok = wait_for(w, wait_ms, err);
1317 if (ok) *out = w->payload;
1318 /* A late answer would write into w: it is leaked rather than freed under the relay (rare: a
1319 * timeout); the relay drops its slot at IMAGE_WAIT. */
1320 if (ok || w->done) {
1321 if (!ok) mpr_payload_free(&w->payload);
1322 pthread_mutex_destroy(&w->mu);
1323 pthread_cond_destroy(&w->cv);
1324 free(w);
1325 }
1326 return ok;
1327}
1328
1329bool mpr_fetch_view(mpr_relay *r, const char *name, uint64_t hash, uint32_t from, uint32_t wait_ms, uint32_t *view_abi, mpw_bytes *view, char **err) {
1330 waiter *w = mt_alloc(sizeof *w);
1331 pthread_mutex_init(&w->mu, NULL);
1332 pthread_cond_init(&w->cv, NULL);
1333 post(r, (event){.kind = EV_FETCH_VIEW, .name = mt_strdup(name), .hash = hash, .from = from, .w = w});
1334 bool ok = wait_for(w, wait_ms, err);
1335 if (ok) *view_abi = w->view_abi, *view = w->view;
1336 if (ok || w->done) {
1337 if (!ok) mpw_bytes_free(&w->view);
1338 pthread_mutex_destroy(&w->mu);
1339 pthread_cond_destroy(&w->cv);
1340 free(w);
1341 }
1342 return ok;
1343}
1344
1345uint64_t mpr_version(mpr_relay *r) { return atomic_load(&r->pub_version); }
1346bool mpr_leader(mpr_relay *r, uint32_t *leader) {
1347 uint32_t l = atomic_load(&r->pub_leader);
1348 if (l == MESH_PEER_NONE) return false;
1349 if (leader) *leader = l;
1350 return true;
1351}
1352uint32_t mpr_role(mpr_relay *r) { return atomic_load(&r->pub_role); }
1353uint32_t mpr_alive(mpr_relay *r) { return atomic_load(&r->pub_alive); }