1/* @ajax/mesh-jobs-op: an edit of the job table (jobs.rs `Op`), an item's committed result (`Res`)
2 * and a payload's reference (`BlobRef`), and the binary encoding of ops (`put_op`, `get_op`,
3 * `encode_ops`, `decode_ops`, `parts`).
4 *
5 * Every op is idempotent, and applying two tables' ops in any order gives their union (an item's
6 * first result wins): see @ajax/mesh-jobs-table. On the wire an op is a u8 tag (its kind's number
7 * below) and its fields. */
8#ifndef MESH_JOBS_OP_H
9#define MESH_JOBS_OP_H
10
11#include "mesh/jobs/id.h"
12#include "mesh/jobs/ranges.h"
13#include "mesh/jobs/spec.h"
14#include "mesh/jobs/wire.h"
15#include "mesh/strbuf.h"
16
17#include <stdbool.h>
18#include <stddef.h>
19#include <stdint.h>
20
21/* A delta, a snapshot part: at most this many bytes of ops per frame. */
22#define MESH_JOBS_PART_BYTES ((size_t)1 << 20)
23
24/* An item's payload, off the log: its sha256, its length, and the members that acknowledged a
25 * durable copy (bit k; the leader that committed it too). */
26typedef struct {
27    mj_hash hash;
28    uint32_t len, holders;
29} mj_blob_ref;
30
31/* An item's result as committed. */
32typedef struct {
33    uint64_t lease;
34    uint32_t member, result;
35    uint8_t *out;
36    size_t out_len;
37    bool has_blob;
38    mj_blob_ref blob;
39} mj_res;
40
41void mj_res_free(mj_res *r);
42mj_res mj_res_clone(const mj_res *r);
43bool mj_res_eq(const mj_res *a, const mj_res *b);
44bool mj_blob_ref_eq(const mj_blob_ref *a, const mj_blob_ref *b);
45
46typedef struct {
47    uint32_t member;
48    uint64_t n;
49} mj_tally;
50
51typedef enum {
52    MJ_OP_SUBMIT = 1,
53    MJ_OP_CANCEL = 2,
54    MJ_OP_DONE = 3,
55    MJ_OP_DONE_RUN = 4, /* items [s, e) done with result 0 and no bytes, by one lease */
56    MJ_OP_FAIL = 5,
57    MJ_OP_FAILS = 6,
58    MJ_OP_DUPES = 7,
59    MJ_OP_FINISH = 8,
60    MJ_OP_FORGET = 9,
61    MJ_OP_FOLD = 10,    /* every item from the job's first to `upto` is done; tally: the folded runs' items per member */
62    MJ_OP_TONE = 11,    /* the job's presentation word := tone (change seq: the highest wins) */
63} mj_op_kind;
64
65/* One op; the fields its kind uses (the rest are zero). */
66typedef struct {
67    mj_op_kind kind;
68    mj_job_id job;     /* every kind (Submit: spec.id) */
69    mj_spec spec;      /* SUBMIT */
70    char *why;         /* CANCEL, FAIL */
71    uint64_t item;     /* DONE, FAIL, FAILS */
72    mj_res res;        /* DONE */
73    uint64_t s, e, lease; /* DONE_RUN */
74    uint32_t member;   /* DONE_RUN */
75    uint32_t tries;    /* FAIL, FAILS */
76    uint64_t n;        /* DUPES */
77    uint64_t at_ms;    /* FINISH */
78    uint32_t state;    /* FINISH */
79    uint64_t upto;     /* FOLD */
80    mj_tally *tally;   /* FOLD */
81    size_t ntally;
82    uint32_t seq, tone; /* TONE */
83} mj_op;
84
85void mj_op_free(mj_op *o);
86mj_op mj_op_clone(const mj_op *o);
87bool mj_op_eq(const mj_op *a, const mj_op *b);
88/* The job an op edits. */
89static inline mj_job_id mj_op_job(const mj_op *o) { return o->kind == MJ_OP_SUBMIT ? o->spec.id : o->job; }
90
91/* Constructors (they copy what they're given). */
92mj_op mj_op_submit(const mj_spec *s);
93mj_op mj_op_cancel(mj_job_id job, const char *why);
94mj_op mj_op_done(mj_job_id job, uint64_t item, const mj_res *res);
95mj_op mj_op_done_run(mj_job_id job, uint64_t s, uint64_t e, uint64_t lease, uint32_t member);
96mj_op mj_op_fail(mj_job_id job, uint64_t item, uint32_t tries, const char *why);
97mj_op mj_op_fails(mj_job_id job, uint64_t item, uint32_t tries);
98mj_op mj_op_dupes(mj_job_id job, uint64_t n);
99mj_op mj_op_finish(mj_job_id job, uint64_t at_ms, uint32_t state);
100mj_op mj_op_forget(mj_job_id job);
101mj_op mj_op_fold(mj_job_id job, uint64_t upto, const mj_tally *tally, size_t n);
102mj_op mj_op_tone(mj_job_id job, uint32_t seq, uint32_t tone);
103
104typedef struct {
105    mj_op *v;
106    size_t n, cap;
107} mj_ops;
108
109void mj_ops_push(mj_ops *v, mj_op o); /* takes o */
110void mj_ops_free(mj_ops *v);
111mj_ops mj_ops_clone(const mj_ops *v);
112bool mj_ops_eq(const mj_ops *a, const mj_ops *b);
113/* Moves every op of src onto the end of dst (src is empty after). */
114void mj_ops_append(mj_ops *dst, mj_ops *src);
115
116/* A payload reference: tag 0 (none) or 1, the hash, len, holders. */
117void mj_w_blob(msb *w, bool has, const mj_blob_ref *b);
118bool mj_rd_blob(mj_rd *r, bool *has, mj_blob_ref *b);
119
120void mj_put_op(msb *w, const mj_op *o);
121/* One op from r; on failure r->err says why ("job op 12?", "job op fold: 9 members", the spec's). */
122bool mj_get_op(mj_rd *r, mj_op *o);
123void mj_encode_ops(const mj_op *ops, size_t n, msb *out);
124/* Ops until the bytes end; false with *err (the caller frees it). */
125bool mj_decode_ops(const void *b, size_t n, mj_ops *out, char **err);
126/* Ops read from r until it ends (appended to out); false on the first bad one. */
127bool mj_rd_ops(mj_rd *r, mj_ops *out);
128
129/* Ops encoded in parts of at most PART_BYTES each (an op is never split, so a part may run over by
130 * its last op); always at least one part (empty when there are no ops). */
131typedef struct {
132    msb *v;
133    size_t n;
134} mj_parts;
135mj_parts mj_parts_of(const mj_op *ops, size_t n);
136void mj_parts_free(mj_parts *p);
137
138#endif