1/* jobd.rs's `a_torn_write_is_cut_back` (review finding 4): a write that fails half-way is cut back to
2 * the last whole record, and a torn tail found at start (a crash mid-append) is dropped; every record
3 * before it replays. Also compaction and the in-memory store. */
4#include "mesh/jobd/store.h"
5#include "mesh/check.h"
6#include "mesh/strbuf.h"
7
8#include <stdio.h>
9#include <stdlib.h>
10#include <string.h>
11#include <sys/stat.h>
12#include <unistd.h>
13
14static mj_log e(uint64_t l, uint64_t n) {
15    mj_ops ops = {0};
16    mj_ops_push(&ops, mj_op_dupes(1, n));
17    return mj_log_entry(l, n, false, ops);
18}
19
20static uint64_t size_of(const char *p) {
21    struct stat st;
22    return stat(p, &st) == 0 ? (uint64_t)st.st_size : 0;
23}
24
25static int logged;
26static void logline(void *c, const char *line) {
27    (void)c;
28    if (strstr(line, "torn last record")) logged++;
29}
30
31int main(void) {
32    char *d = mfmt("%s/meshd-jobd-torn-%d", getenv("TMPDIR") ? getenv("TMPDIR") : "/tmp", (int)getpid());
33    mkdir(d, 0777);
34    char *p = mfmt("%s/jobs", d);
35    uint64_t l = (uint64_t)1 << 40;
36    mj_replica r;
37    char *err = NULL;
38    mjd_store *s = mjd_store_open(p, &r, NULL, NULL, &err);
39    mj_replica empty;
40    mj_replica_init(&empty);
41    CHECK(s && mj_replica_eq(&r, &empty));
42    mj_replica_free(&r);
43    mj_disk dk = mjd_store_disk(s);
44    mj_log recs[3] = {mj_log_lid(l), e(l, 1), e(l, 2)};
45    for (int i = 0; i < 3; i++) CHECK(dk.append(dk.ctx, &recs[i], &err));
46    CHECK(dk.sync(dk.ctx, &err));
47    uint64_t whole = size_of(p);
48    mjd_store_tear_next(s, 7);
49    mj_log e3 = e(l, 3);
50    CHECK(!dk.append(dk.ctx, &e3, &err) && strstr(err, "torn"));
51    free(err);
52    CHECK_U(size_of(p), whole); /* cut back to the last whole record */
53    mjd_store_free(s);
54    s = mjd_store_open(p, &r, NULL, NULL, &err);
55    CHECK(s && r.lid == l && r.npending == 2);
56    mjd_store_free(s);
57    /* A torn tail left by a crash: dropped at start, and the file rewritten without it. */
58    msb b = {0};
59    mj_log_encode(&e3, &b);
60    FILE *f = fopen(p, "ab");
61    fwrite(b.p, 1, 9, f);
62    fclose(f);
63    msb_free(&b);
64    mj_replica r2;
65    s = mjd_store_open(p, &r2, logline, NULL, &err);
66    CHECK(s && mj_replica_eq(&r2, &r) && logged == 1);
67    FILE *g = fopen(p, "rb");
68    msb all = {0};
69    char buf[4096];
70    size_t n;
71    while ((n = fread(buf, 1, sizeof buf, g)) > 0) msb_put(&all, buf, n);
72    fclose(g);
73    mj_replica r3;
74    size_t recs_n, at;
75    CHECK(mj_replay(all.p, all.n, &r3, &recs_n, &at, NULL) && at == all.n);
76    mj_replica_free(&r3);
77    msb_free(&all);
78    /* A compaction rewrites it as the replica's own log. */
79    CHECK(!mjd_store_wants_compact(s));
80    CHECK(mjd_store_rewrite(s, &r2, &err));
81    CHECK(mjd_store_persisted(s));
82    mjd_store_free(s);
83    /* In memory: nothing written, appends succeed. */
84    mj_replica rm;
85    s = mjd_store_open(NULL, &rm, NULL, NULL, &err);
86    dk = mjd_store_disk(s);
87    CHECK(!mjd_store_persisted(s) && dk.append(dk.ctx, &e3, &err) && dk.sync(dk.ctx, &err));
88    mjd_store_free(s);
89    mj_replica_free(&rm);
90    for (int i = 0; i < 3; i++) mj_log_free(&recs[i]);
91    mj_log_free(&e3);
92    mj_replica_free(&r);
93    mj_replica_free(&r2);
94    mj_replica_free(&empty);
95    unlink(p);
96    char *tmp = mfmt("%s.jobs-tmp", p);
97    unlink(tmp);
98    rmdir(d);
99    free(tmp);
100    free(p);
101    free(d);
102    return mesh_check_done();
103}