==> src/store.c <==1/* jobd.rs's Store. */
2#include "mesh/jobd/store.h"
3#include "mesh/strbuf.h"
4
5#include <errno.h>
6#include <fcntl.h>
7#include <stdio.h>
8#include <stdlib.h>
9#include <string.h>
10#include <sys/stat.h>
11#include <unistd.h>
12
13struct mjd_store {
14 char *path;
15 int fd;
16 uint64_t bytes, base;
17 bool dirty;
18 bool tear;
19 size_t tear_at;
20};
21
22static void say(mjd_log_fn log, void *ctx, char *line) {
23 if (log) log(ctx, line);
24 free(line);
25}
26
27/* Path::with_extension("jobs-tmp"): the file name's extension replaced (or added). */
28static char *tmp_of(const char *path) {
29 const char *slash = strrchr(path, '/');
30 const char *name = slash ? slash + 1 : path;
31 const char *dot = strrchr(name, '.');
32 if (dot && dot != name) return mfmt("%.*s.jobs-tmp", (int)(dot - path), path);
33 return mfmt("%s.jobs-tmp", path);
34}
35
36static bool write_all(int fd, const void *b, size_t n) {
37 const char *p = b;
38 while (n) {
39 ssize_t w = write(fd, p, n);
40 if (w < 0 && errno == EINTR) continue;
41 if (w <= 0) return false;
42 p += w;
43 n -= (size_t)w;
44 }
45 return true;
46}
47
48bool mjd_store_rewrite(mjd_store *s, const mj_replica *r, char **err) {
49 if (!s->path) return true;
50 char *tmp = tmp_of(s->path);
51 mj_logs lg = mj_replica_log(r);
52 msb b = {0};
53 for (size_t i = 0; i < lg.n; i++) mj_log_encode(&lg.v[i], &b);
54 mj_logs_free(&lg);
55 const char *what = NULL;
56 int fd = open(tmp, O_WRONLY | O_CREAT | O_TRUNC, 0644);
57 if (fd < 0) what = "create";
58 else {
59 if (!write_all(fd, b.p ? b.p : "", b.n)) what = "write";
60 else if (fsync(fd) != 0) what = "fsync";
61 close(fd);
62 }
63 if (!what && rename(tmp, s->path) != 0) what = "rename";
64 if (!what) {
65 char *dir = mstrdup(s->path), *slash = strrchr(dir, '/');
66 if (slash && slash != dir) {
67 *slash = 0;
68 int dfd = open(dir, O_RDONLY);
69 if (dfd < 0 || fsync(dfd) != 0) what = "fsync dir";
70 if (dfd >= 0) close(dfd);
71 }
72 free(dir);
73 }
74 int nfd = -1;
75 if (!what && (nfd = open(s->path, O_WRONLY | O_APPEND)) < 0) what = "open";
76 free(tmp);
77 if (what) {
78 if (err) *err = mfmt("--jobs-state %s: %s: %s", s->path, what, strerror(errno));
79 msb_free(&b);
80 return false;
81 }
82 if (s->fd >= 0) close(s->fd);
83 s->fd = nfd;
84 s->bytes = s->base = b.n;
85 s->dirty = false;
86 msb_free(&b);
87 return true;
88}
89
90mjd_store *mjd_store_open(const char *path, mj_replica *rep, mjd_log_fn log, void *ctx, char **err) {
91 mjd_store *s = mxalloc(sizeof *s);
92 s->fd = -1;
93 mj_replica_init(rep);
94 if (!path) return s;
95 s->path = mstrdup(path);
96 FILE *f = fopen(path, "rb");
97 msb b = {0};
98 if (f) {
99 char buf[65536];
100 size_t r;
101 while ((r = fread(buf, 1, sizeof buf, f)) > 0) msb_put(&b, buf, r);
102 fclose(f);
103 } else if (errno != ENOENT) {
104 if (err) *err = mfmt("--jobs-state %s: %s", path, strerror(errno));
105 mjd_store_free(s);
106 return NULL;
107 }
108 size_t n = 0, whole = 0;
109 char *e = NULL;
110 mj_replica_free(rep);
111 if (!mj_replay(b.p ? b.p : "", b.n, rep, &n, &whole, &e)) {
112 if (err) *err = mfmt("--jobs-state %s: %s", path, e);
113 free(e);
114 msb_free(&b);
115 mj_replica_init(rep);
116 mjd_store_free(s);
117 return NULL;
118 }
119 if (whole < b.n) say(log, ctx, mfmt("jobs: %s: a torn last record (%zu bytes) dropped", path, b.n - whole));
120 msb_free(&b);
121 if (!mjd_store_rewrite(s, rep, err)) {
122 mjd_store_free(s);
123 return NULL;
124 }
125 if (n > 0)
126 say(log, ctx, mfmt("jobs: %zu records of %s replayed: %zu job(s) committed, %zu entr(ies) pending", n, path, momap_len(rep->committed.jobs),
127 rep->npending));
128 return s;
129}
130
131void mjd_store_free(mjd_store *s) {
132 if (!s) return;
133 if (s->fd >= 0) close(s->fd);
134 free(s->path);
135 free(s);
136}
137
138bool mjd_store_persisted(const mjd_store *s) { return s->path != NULL; }
139
140bool mjd_store_wants_compact(const mjd_store *s) { return s->path && s->bytes > 2 * s->base + (8u << 20); }
141
142void mjd_store_tear_next(mjd_store *s, size_t k) {
143 s->tear = true;
144 s->tear_at = k;
145}
146
147static bool st_append(void *ctx, const mj_log *l, char **err) {
148 mjd_store *s = ctx;
149 if (s->fd < 0) return true;
150 msb b = {0};
151 mj_log_encode(l, &b);
152 bool ok;
153 const char *why;
154 if (s->tear) {
155 s->tear = false;
156 write_all(s->fd, b.p, s->tear_at < b.n ? s->tear_at : b.n);
157 ok = false;
158 why = "injected: torn write";
159 } else {
160 ok = write_all(s->fd, b.p, b.n);
161 why = strerror(errno);
162 }
163 if (!ok) {
164 /* Cut back to the last whole record (what a replay reads), then fail. */
165 char *w = mstrdup(why);
166 bool cut = ftruncate(s->fd, (off_t)s->bytes) == 0 && fsync(s->fd) == 0;
167 if (err) *err = cut ? mfmt("job log: %s", w) : mfmt("job log: %s (and cutting it back to its last whole record: %s)", w, strerror(errno));
168 free(w);
169 msb_free(&b);
170 return false;
171 }
172 s->bytes += b.n;
173 s->dirty = true;
174 msb_free(&b);
175 return true;
176}
177
178static bool st_sync(void *ctx, char **err) {
179 mjd_store *s = ctx;
180 if (!s->dirty) return true;
181 s->dirty = false;
182 if (s->fd < 0) return true;
183 /* fsync(2): the data reached the drive (not a drive cache flush, which the jobs thread can't afford every tick). */
184 if (fsync(s->fd) != 0) {
185 if (err) *err = mfmt("job log sync: %s", strerror(errno));
186 return false;
187 }
188 return true;
189}
190
191mj_disk mjd_store_disk(mjd_store *s) { return (mj_disk){s, st_append, st_sync}; }