From cff25dd387376311b09bab3a58377fecbfe818e4 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Mon, 28 Sep 2026 10:43:45 +0100 Subject: Log an app's saves in reused segments and write list images whole beside the store --- Makefile | 11 +- core/log.h | 46 +++++ core/store.h | 2 +- store/store-log.c | 572 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ store/test-log.c | 235 ++++++++++++++++++++++ 5 files changed, 862 insertions(+), 4 deletions(-) create mode 100644 core/log.h create mode 100644 store/store-log.c create mode 100644 store/test-log.c diff --git a/Makefile b/Makefile index f03f07a..91cb191 100644 --- a/Makefile +++ b/Makefile @@ -17,11 +17,11 @@ BUILD ?= build # libpure.a is the app model with the tree store: what a host embeds (Vega's # surfaces link it). prototypes/ keeps the earlier document runtime (000-077). # Unit tests sit by their part (core/test-module.c, store/test-tree-store.c, -# serve/test-message-json.c); +# store/test-log.c, serve/test-message-json.c); # tests/check.py checks the hosts end to end. INCLUDES = -Icore MODEL = core/module.c core/module-list.c core/module-save.c core/module-text.c core/module-store.c core/flow.c -STORE = store/store-tree.c +STORE = store/store-tree.c store/store-log.c SERVE = serve/serve.c serve/deliver.c serve/message-json.c serve/crypto.c HEADERS = $(wildcard core/*.h) $(wildcard serve/*.h) $(wildcard store/*.h) OBJECTS = $(patsubst %.c,$(BUILD)/obj/%.o,$(MODEL) $(STORE)) @@ -65,6 +65,10 @@ $(BUILD)/test-message-json: serve/test-message-json.c serve/message-json.c $(BUI $(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -Iserve serve/test-message-json.c serve/message-json.c $(BUILD)/libpure.a \ $(LDFLAGS) -lm $(LDLIBS) -o $@ +$(BUILD)/test-log: store/test-log.c store/store-log.c $(HEADERS) + @mkdir -p $(BUILD) + $(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -DPURE_STORE_TEST store/test-log.c store/store-log.c $(LDFLAGS) $(LDLIBS) -o $@ + $(BUILD)/test-tree-store: store/test-tree-store.c $(STORE) $(HEADERS) @mkdir -p $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -DPURE_STORE_TEST store/test-tree-store.c $(STORE) $(LDFLAGS) $(LDLIBS) -o $@ @@ -72,8 +76,9 @@ $(BUILD)/test-tree-store: store/test-tree-store.c $(STORE) $(HEADERS) # The HTTP host's checks build their apps with aup (the sibling checkout) and # compare documents with the earlier document server. AUP ?= ../aup -check: all $(BUILD)/test-module $(BUILD)/test-tree-store $(BUILD)/test-message-json +check: all $(BUILD)/test-module $(BUILD)/test-tree-store $(BUILD)/test-log $(BUILD)/test-message-json $(BUILD)/test-tree-store + $(BUILD)/test-log $(BUILD)/test-module $(BUILD)/test-message-json $(MAKE) -C $(AUP) build/aup diff --git a/core/log.h b/core/log.h new file mode 100644 index 0000000..44dbdfb --- /dev/null +++ b/core/log.h @@ -0,0 +1,46 @@ +#ifndef PURE_LOG_H +#define PURE_LOG_H +#include "flow.h" +#include "result.h" +#include +#include +#include + +/* An app's log beside its store (store/store-log.c): an entry a save, its + * number (one more than the entry before), whether the save also committed + * the tree, and its payload. An entry counts once pure_log_append returns + * PURE_OK: it is on the device. One thread uses a log. */ +typedef struct PureLog PureLog; +/* The log in directory (made empty when it has none). The entries found are + * checked (digests, numbers following on) up to the first that is not. */ +PureResult pure_log_open(const char *directory, PureLog **out); +void pure_log_close(PureLog *); +/* The last entry's number (0: none). */ +uint64_t pure_log_last(const PureLog *); +/* The entries from number from on (0: every entry kept), in order; payloads + * stay valid until the next append. STORE_ERROR when an entry from `from` on + * is missing; a visit that returns false stops. Only before appending. */ +PureResult pure_log_read(PureLog *, uint64_t from, + bool (*visit)(void *, uint64_t seq, bool tree, PureSpan payload), void *context); +/* Entries after seq no longer count: the next append follows seq, over them. */ +void pure_log_end(PureLog *, uint64_t seq); +/* Append the entry numbered seq (one more than the last) and make it durable: + * UNCERTAIN when writing or syncing failed (it may or may not be there). */ +PureResult pure_log_append(PureLog *, uint64_t seq, bool tree, PureSpan payload); +/* The last appended entry no longer counts (its save failed after it was + * written): the next append takes its number and its place. */ +void pure_log_retract(PureLog *); +/* Entries up to seq are no longer read: their segments may be used again. */ +void pure_log_release(PureLog *, uint64_t seq); +/* Bytes of the segments that hold entries not released. */ +uint64_t pure_log_bytes(const PureLog *); + +/* A list's image: a file of directory named after the list, replaced whole + * (written beside it, synced, renamed over it): pure_image_write writes the + * bytes of parts one after another, pure_image_read gives them back (malloc'd; + * NOT_FOUND when there is none, STORE_ERROR when damaged). The bytes of the + * image are digested: a file cut short or changed is refused. */ +PureResult pure_image_write(const char *directory, const char *name, const PureSpan *parts, size_t count); +PureResult pure_image_read(const char *directory, const char *name, unsigned char **bytes, size_t *size); +PureResult pure_image_remove(const char *directory, const char *name); +#endif diff --git a/core/store.h b/core/store.h index 64b8d9e..7086d3a 100644 --- a/core/store.h +++ b/core/store.h @@ -51,7 +51,7 @@ PureResult pure_store_keys(const char *path, bool (*visit)(void *, const char *k /* Fault hooks exist only in the separately built unit-test binary. */ typedef enum { PURE_AFTER_HEADER, PURE_BEFORE_FILE_SYNC, PURE_BEFORE_RENAME, PURE_BEFORE_DIR_SYNC, - PURE_COPY_RANGE_UNSUPPORTED_AFTER_PARTIAL + PURE_COPY_RANGE_UNSUPPORTED_AFTER_PARTIAL, PURE_LOG_BEFORE_WRITE, PURE_LOG_BEFORE_SYNC, PURE_IMAGE_BEFORE_RENAME } PureStoreStage; #ifdef PURE_STORE_TEST int pure_store_test_fault(PureStoreStage stage); diff --git a/store/store-log.c b/store/store-log.c new file mode 100644 index 0000000..6138a84 --- /dev/null +++ b/store/store-log.c @@ -0,0 +1,572 @@ +/* The app log and list images (core/log.h). + * + * The log's entries go into segments, files log-0, log-1, ... in the store's + * directory, each allocated whole when made, so that an append neither grows + * a file nor takes new blocks and its sync writes the entry alone, and used + * again once every entry in it is released. A segment begins with a header + * naming its generation (higher than any before it) and the generation of the + * segment it continues; an entry is its save's number, the number and digest + * of the entry before it (0 for the first), its payload's length and flags, + * a digest of those and the payload, then the payload, padded to 8 bytes. + * Opening reads each segment's entries while each names the one before it, + * and chains the segments from the newest back by the generations and entries + * they name: a torn entry, or bytes a segment kept from an earlier use, end + * what is read. Save numbers may skip (a save that changed no list writes no + * entry) and are used again (for an entry taken back), so an entry is known + * by the digest of the one it names, not by numbers alone. + * + * An image is written to a temporary file in 256 KiB steps, each started to + * the device as the next is written, then synced and renamed over the image + * it replaces: a writer never has more than two steps waiting, so the log's + * syncs between them wait for little. */ +#if defined(__linux__) +#define _GNU_SOURCE +#else +#define _POSIX_C_SOURCE 200809L +#endif +#ifdef __APPLE__ +#define _DARWIN_C_SOURCE +#endif +#include "log.h" +#include "store.h" +#include "digest.h" +#include +#include +#include +#include +#include +#include +#include +#include + +enum { HEADER = 64, ENTRY = 40, SEGMENTS = 256, STEP = 256 << 10, TREE = 1 }; +#ifdef PURE_STORE_TEST +enum { SEGMENT = 16384 }; +#else +enum { SEGMENT = 8 << 20 }; +#endif +static const unsigned char log_magic[8] = {'P', 'U', 'R', 'E', 'L', 'O', 'G', '1'}; +static const unsigned char image_magic[8] = {'P', 'U', 'R', 'E', 'I', 'M', 'G', '1'}; + +static bool fault(PureStoreStage stage) { +#ifdef PURE_STORE_TEST + if (pure_store_test_fault(stage)) { errno = EIO; return true; } +#else + (void)stage; +#endif + return false; +} + +static uint64_t get64(const unsigned char *bytes) { + uint64_t value = 0; + for (unsigned i = 0; i < 8; ++i) value |= (uint64_t)bytes[i] << (8 * i); + return value; +} + +static void put64(unsigned char *bytes, uint64_t value) { + for (unsigned i = 0; i < 8; ++i) bytes[i] = (unsigned char)(value >> (8 * i)); +} + +static bool write_all(int fd, const unsigned char *bytes, size_t length, uint64_t offset) { + while (length) { + ssize_t n = pwrite(fd, bytes, length, (off_t)offset); + if (n < 0 && errno == EINTR) continue; + if (n <= 0) return false; + bytes += (size_t)n; + length -= (size_t)n; + offset += (uint64_t)n; + } + return true; +} + +static bool sync_whole(int fd) { + int rc; + do { rc = fsync(fd); } while (rc && errno == EINTR); + return rc == 0; +} + +static bool sync_data(int fd) { +#ifdef __APPLE__ + return sync_whole(fd); +#else + int rc; + do { rc = fdatasync(fd); } while (rc && errno == EINTR); + return rc == 0; +#endif +} + +/* Room for size bytes from the start of a file, allocated. */ +static bool allocate(int fd, uint64_t size) { +#if defined(__APPLE__) + return !ftruncate(fd, (off_t)size); +#else + int rc = posix_fallocate(fd, 0, (off_t)size); + return rc == 0 || ((rc == EOPNOTSUPP || rc == EINVAL) && !ftruncate(fd, (off_t)size)); +#endif +} + +/* ---- The log */ + +typedef struct { + int fd; + uint64_t generation; /* 0: no header */ + uint64_t predecessor; /* the generation of the segment it continues (0: none) */ + uint64_t size; /* bytes allocated */ + /* Its entries (first 0: none), the entry named before the first (and its + * digest), the last that counts (and its digest) and where that one ends. */ + uint64_t first, previous, named, last, digest, end; +} Segment; + +struct PureLog { + int directory; + Segment segment[SEGMENTS]; + unsigned count; + int current; /* the segment appended to, or -1 */ + uint64_t last, digest; /* the last entry's number and digest */ + uint64_t generation; /* the highest any header has had */ + uint64_t released; + /* The segments as opened, mapped until the first append (a start reads + * them; what is appended after is never read in the same run). */ + unsigned char *maps[SEGMENTS]; + size_t mapped[SEGMENTS]; + /* The last append, to take back: its segment as it was and the last entry before it. */ + int appended; + Segment before; + uint64_t last_before, digest_before; + unsigned char *buffer; + size_t capacity; +}; + +static uint64_t entry_digest(const unsigned char *head, const unsigned char *payload, size_t size) { + Digest digest = {0}; + digest_add(&digest, head, ENTRY - 8); + digest_add(&digest, payload, size); + return digest_end(&digest); +} + +static size_t padded(size_t size) { return (size + 7) & ~(size_t)7; } +static uint32_t entry_length(const unsigned char *head) { return (uint32_t)get64(head + 16); } + +/* The entries of a mapped segment from its header on, each naming the one + * before: the first (and the entry it names), the last, and their end. */ +static void scan(Segment *s, const unsigned char *map, size_t size) { + s->first = s->previous = s->named = s->last = s->digest = 0; + s->end = HEADER; + uint64_t at = HEADER; + while (at + ENTRY <= size) { + const unsigned char *head = map + at; + uint64_t seq = get64(head), previous = get64(head + 8), named = get64(head + 24); + uint32_t length = entry_length(head); + if (!seq || (s->first && (previous != s->last || named != s->digest || seq <= s->last)) || + (!s->first && previous >= seq) || length > size - at - ENTRY || + get64(head + 32) != entry_digest(head, head + ENTRY, length)) + break; + if (!s->first) s->first = seq, s->previous = previous, s->named = named; + s->last = seq; + s->digest = get64(head + 32); + at += ENTRY + padded(length); + s->end = at; + } +} + +/* Where entry seq of a segment as opened ends (0: not among its entries), + * and its digest. */ +static uint64_t end_of(const PureLog *log, unsigned i, uint64_t seq, uint64_t *digest) { + const Segment *s = &log->segment[i]; + const unsigned char *map = log->maps[i]; + if (!map || !s->first || seq < s->first || seq > s->last) return 0; + for (uint64_t at = HEADER; at < s->end;) { + const unsigned char *head = map + at; + at += ENTRY + padded(entry_length(head)); + if (get64(head) == seq) { + *digest = get64(head + 32); + return at; + } + } + return 0; +} + +static int generation_at(const PureLog *log, uint64_t generation) { + for (unsigned i = 0; generation && i < log->count; ++i) + if (log->segment[i].generation == generation) return (int)i; + return -1; +} + +/* Zeros over an entry's header: what follows it is no longer read. */ +static bool erase(const Segment *s, uint64_t at) { + static const unsigned char zeros[ENTRY]; + return at + ENTRY > s->size || (write_all(s->fd, zeros, ENTRY, at) && sync_data(s->fd)); +} + +static void unmap(PureLog *log) { + for (unsigned i = 0; i < log->count; ++i) + if (log->maps[i]) { + munmap(log->maps[i], log->mapped[i]); + log->maps[i] = NULL; + } +} + +PureResult pure_log_open(const char *directory, PureLog **out) { + *out = NULL; + PureLog *log = calloc(1, sizeof(*log)); + if (!log) return PURE_ALLOCATION; + log->current = log->appended = -1; + log->directory = open(directory, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (log->directory < 0) { free(log); return PURE_STORE_ERROR; } + PureResult result = PURE_OK; + for (unsigned n = 0; n < SEGMENTS; ++n) { + char name[24]; + snprintf(name, sizeof(name), "log-%u", n); + int fd = openat(log->directory, name, O_RDWR | O_NOFOLLOW | O_CLOEXEC); + if (fd < 0) { + if (errno != ENOENT) result = PURE_STORE_ERROR; + break; + } + struct stat file; + Segment *s = &log->segment[log->count++]; + *s = (Segment){.fd = fd, .end = HEADER}; + if (fstat(fd, &file)) { result = PURE_STORE_ERROR; break; } + s->size = (uint64_t)file.st_size; + if (s->size < HEADER) continue; + void *map = mmap(NULL, (size_t)s->size, PROT_READ, MAP_SHARED, fd, 0); + if (map == MAP_FAILED) { result = PURE_STORE_ERROR; break; } + log->maps[log->count - 1] = map; + log->mapped[log->count - 1] = (size_t)s->size; + const unsigned char *header = map; + if (memcmp(header, log_magic, 8) || get64(header + HEADER - 8) != digest_of(header, HEADER - 8)) continue; + s->generation = get64(header + 8); + s->predecessor = get64(header + 16); + if (s->generation > log->generation) log->generation = s->generation; + scan(s, map, (size_t)s->size); + } + if (result != PURE_OK) { pure_log_close(log); return result; } + /* The newest segment with entries holds the last; from it back, each + * segment names the one it continues, which counts up to the entry the + * later one names first. The rest hold nothing the log still has. */ + int top = -1; + for (unsigned i = 0; i < log->count; ++i) + if (log->segment[i].first && (top < 0 || log->segment[i].generation > log->segment[top].generation)) top = (int)i; + bool chained[SEGMENTS] = {false}; + for (int at = top, later = -1; at >= 0; later = at, at = generation_at(log, log->segment[at].predecessor)) { + Segment *s = &log->segment[at]; + if (chained[at] || s->generation >= (later >= 0 ? log->segment[later].generation : UINT64_MAX)) break; + if (later >= 0) { /* it counts up to the entry the later segment names */ + uint64_t digest, ends = end_of(log, (unsigned)at, log->segment[later].previous, &digest); + if (!ends || digest != log->segment[later].named) break; /* released: known from the later segment on */ + s->last = log->segment[later].previous; + s->digest = digest; + s->end = ends; + } + chained[at] = true; + if (!s->previous) break; + } + for (unsigned i = 0; i < log->count; ++i) + if (!chained[i]) { + Segment *s = &log->segment[i]; + s->first = s->previous = s->last = 0; + s->end = HEADER; + } + if (top >= 0) { + log->current = top; + log->last = log->segment[top].last; + log->digest = log->segment[top].digest; + } + *out = log; + return PURE_OK; +} + +void pure_log_close(PureLog *log) { + if (!log) return; + unmap(log); + for (unsigned i = 0; i < log->count; ++i) close(log->segment[i].fd); + close(log->directory); + free(log->buffer); + free(log); +} + +uint64_t pure_log_last(const PureLog *log) { return log->last; } + +uint64_t pure_log_bytes(const PureLog *log) { + uint64_t bytes = 0; + for (unsigned i = 0; i < log->count; ++i) { + const Segment *s = &log->segment[i]; + if (s->first && ((int)i == log->current || s->last > log->released)) bytes += s->end; + } + return bytes; +} + +PureResult pure_log_read(PureLog *log, uint64_t from, + bool (*visit)(void *, uint64_t seq, bool tree, PureSpan payload), void *context) { + if (log->current < 0 || log->last < from) return PURE_OK; + int chain[SEGMENTS]; + unsigned links = 0; + for (int at = log->current; at >= 0 && links < SEGMENTS;) { + const Segment *s = &log->segment[at]; + if (!s->first) break; + chain[links++] = at; + if (s->first <= from || !s->previous) break; + at = generation_at(log, s->predecessor); + } + if (!links) return PURE_OK; + const Segment *oldest = &log->segment[chain[links - 1]]; + if (from && oldest->first > from && oldest->previous >= from) return PURE_STORE_ERROR; /* entries from `from` on are gone */ + while (links--) { + const unsigned char *map = log->maps[chain[links]]; + const Segment *s = &log->segment[chain[links]]; + for (uint64_t at = HEADER; map && at < s->end;) { + const unsigned char *head = map + at; + uint64_t seq = get64(head); + if (seq >= from && !visit(context, seq, (get64(head + 16) >> 32) & TREE, (PureSpan){head + ENTRY, entry_length(head)})) + return PURE_OK; + at += ENTRY + padded(entry_length(head)); + } + } + return PURE_OK; +} + +void pure_log_end(PureLog *log, uint64_t seq) { + if (seq >= log->last) return; + /* The chained segment holding seq keeps its entries up to it; after it, + * there and in every newer segment, the device is erased too, so that + * opening again finds none of them. */ + int holder = -1; + for (int at = log->current; at >= 0; at = generation_at(log, log->segment[at].predecessor)) { + const Segment *s = &log->segment[at]; + if (!s->first) break; + if (s->first <= seq && seq <= s->last) { holder = at; break; } + if (!s->previous) break; + } + uint64_t kept = holder >= 0 ? log->segment[holder].generation : 0, digest = 0; + for (unsigned i = 0; i < log->count; ++i) { + Segment *s = &log->segment[i]; + if ((int)i == holder) { + s->end = end_of(log, i, seq, &digest); + s->last = seq; + s->digest = digest; + erase(s, s->end); + } else if (s->generation > kept) { + erase(s, HEADER); + s->first = s->previous = s->last = 0; + s->end = HEADER; + } + } + log->current = holder; + log->last = seq; + log->digest = digest; +} + +/* A segment to begin: one holding no entry still counted, else a new file. */ +static int fresh_segment(PureLog *log, uint64_t needed) { + int pick = -1; + for (unsigned i = 0; i < log->count; ++i) { + const Segment *s = &log->segment[i]; + if ((int)i == log->current || (s->first && s->last > log->released)) continue; + if (pick < 0 || s->size > log->segment[pick].size) pick = (int)i; + } + if (pick >= 0) { + Segment *s = &log->segment[pick]; + if (s->size < needed && !allocate(s->fd, needed)) return -1; + if (s->size < needed) s->size = needed; + return pick; + } + if (log->count == SEGMENTS) return -1; + char name[24]; + snprintf(name, sizeof(name), "log-%u", log->count); + int fd = openat(log->directory, name, O_RDWR | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC, 0600); + uint64_t size = needed > SEGMENT ? needed : SEGMENT; + /* The new file and its name are durable before an entry in it counts. */ + if (fd < 0 || !allocate(fd, size) || !sync_whole(fd) || !sync_whole(log->directory)) { + if (fd >= 0) { + close(fd); + unlinkat(log->directory, name, 0); + } + return -1; + } + log->segment[log->count] = (Segment){.fd = fd, .size = size, .end = HEADER}; + return (int)log->count++; +} + +PureResult pure_log_append(PureLog *log, uint64_t seq, bool tree, PureSpan payload) { + if (seq <= log->last || payload.size > UINT32_MAX) return PURE_INVALID; + unmap(log); + size_t entry = ENTRY + padded(payload.size); + int at = log->current; + Segment *s = at >= 0 ? &log->segment[at] : NULL; + /* A segment with no entry yet (a first one taken back) takes the next + * entry after its header, grown to hold it. */ + if (s && !s->first && s->end + entry > s->size) { + if (!allocate(s->fd, s->end + entry)) return PURE_STORE_ERROR; + s->size = s->end + entry; + } + bool begin = !s || s->end + entry > s->size; + if (begin && (at = fresh_segment(log, HEADER + entry)) < 0) return PURE_STORE_ERROR; + s = &log->segment[at]; + size_t size = (begin ? HEADER : 0) + entry; + if (log->capacity < size) { + size_t capacity = log->capacity ? log->capacity : 65536; + while (capacity < size) capacity *= 2; + unsigned char *grown = realloc(log->buffer, capacity); + if (!grown) return PURE_ALLOCATION; + log->buffer = grown; + log->capacity = capacity; + } + uint64_t generation = begin ? log->generation + 1 : s->generation, offset = begin ? 0 : s->end; + uint64_t predecessor = begin ? (log->current >= 0 ? log->segment[log->current].generation : 0) : s->predecessor; + unsigned char *out = log->buffer; + if (begin) { /* a segment's header goes with its first entry, in one write */ + memset(out, 0, HEADER); + memcpy(out, log_magic, 8); + put64(out + 8, generation); + put64(out + 16, predecessor); + put64(out + HEADER - 8, digest_of(out, HEADER - 8)); + out += HEADER; + } + put64(out, seq); + put64(out + 8, log->last); + put64(out + 16, (uint64_t)payload.size | (uint64_t)(tree ? TREE : 0) << 32); + put64(out + 24, log->digest); + if (payload.size) memcpy(out + ENTRY, payload.data, payload.size); + memset(out + ENTRY + payload.size, 0, padded(payload.size) - payload.size); + uint64_t digest = entry_digest(out, out + ENTRY, payload.size); + put64(out + 32, digest); + if (fault(PURE_LOG_BEFORE_WRITE)) return PURE_STORE_ERROR; /* nothing written: no entry */ + if (!write_all(s->fd, log->buffer, size, offset) || fault(PURE_LOG_BEFORE_SYNC) || !sync_data(s->fd)) + return PURE_UNCERTAIN; + if (begin) { + log->generation = generation; + *s = (Segment){.fd = s->fd, .size = s->size, .generation = generation, .predecessor = predecessor, .end = HEADER}; + } + log->before = *s; + log->last_before = log->last; + log->digest_before = log->digest; + log->appended = at; + if (!s->first) s->first = seq, s->previous = log->last, s->named = log->digest; + s->last = seq; + s->digest = digest; + s->end = offset + (begin ? HEADER : 0) + entry; + log->last = seq; + log->digest = digest; + log->current = at; + return PURE_OK; +} + +void pure_log_retract(PureLog *log) { + if (log->appended < 0) return; + /* The segment keeps its header even when the entry was its first: it + * stays the one appended to, and the next entry takes this one's place. */ + log->segment[log->appended] = log->before; + log->current = log->appended; + log->last = log->last_before; + log->digest = log->digest_before; + log->appended = -1; +} + +void pure_log_release(PureLog *log, uint64_t seq) { + if (seq > log->released) log->released = seq; +} + +/* ---- Images */ + +static bool image_name(char out[64], const char *name, const char *prefix) { + size_t length = strlen(name); + if (!length || length > 40) return false; + for (size_t i = 0; i < length; ++i) { + char c = name[i]; + if (!((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_')) return false; + } + snprintf(out, 64, "%simage-%s", prefix, name); + return true; +} + +PureResult pure_image_write(const char *directory, const char *name, const PureSpan *parts, size_t count) { + char final[64], temporary[96]; + if (!image_name(final, name, "")) return PURE_INVALID; + static uint64_t sequence; + snprintf(temporary, sizeof(temporary), ".%s-%ld-%llu", final, (long)getpid(), + (unsigned long long)__atomic_add_fetch(&sequence, 1, __ATOMIC_RELAXED)); + int dir = open(directory, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (dir < 0) return PURE_STORE_ERROR; + int fd = openat(dir, temporary, O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC, 0600); + if (fd < 0) { close(dir); return PURE_STORE_ERROR; } + uint64_t length = 0; + for (size_t i = 0; i < count; ++i) length += parts[i].size; + unsigned char head[16]; + memcpy(head, image_magic, 8); + put64(head + 8, length); + Digest digest = {0}; + bool ok = write_all(fd, head, sizeof(head), 0); + uint64_t at = sizeof(head), started = 0; + for (size_t i = 0; ok && i < count; ++i) + for (size_t done = 0; ok && done < parts[i].size;) { + size_t take = parts[i].size - done < STEP ? parts[i].size - done : STEP; + digest_add(&digest, parts[i].data + done, take); + ok = write_all(fd, parts[i].data + done, take, at); + done += take; + at += take; +#if defined(__linux__) + /* Written steps go to the device as the next are written; a + * writer waits for the step before the last. */ + if (ok && at - started >= STEP) { + sync_file_range(fd, (off_t)started, (off_t)(at - started), SYNC_FILE_RANGE_WRITE); + if (started >= STEP) + sync_file_range(fd, (off_t)(started - STEP), STEP, + SYNC_FILE_RANGE_WAIT_BEFORE | SYNC_FILE_RANGE_WRITE | SYNC_FILE_RANGE_WAIT_AFTER); + started = at; + } +#endif + } + unsigned char tail[8]; + put64(tail, digest_end(&digest)); + ok = ok && write_all(fd, tail, sizeof(tail), at) && !fault(PURE_IMAGE_BEFORE_RENAME) && sync_data(fd); + if (close(fd)) ok = false; + ok = ok && !renameat(dir, temporary, dir, final); + if (!ok) unlinkat(dir, temporary, 0); + ok = ok && sync_whole(dir); + close(dir); + return ok ? PURE_OK : PURE_STORE_ERROR; +} + +PureResult pure_image_read(const char *directory, const char *name, unsigned char **bytes, size_t *size) { + char final[64]; + *bytes = NULL; + *size = 0; + if (!image_name(final, name, "")) return PURE_INVALID; + int dir = open(directory, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (dir < 0) return PURE_STORE_ERROR; + int fd = openat(dir, final, O_RDONLY | O_NOFOLLOW | O_CLOEXEC); + close(dir); + if (fd < 0) return errno == ENOENT ? PURE_NOT_FOUND : PURE_STORE_ERROR; + struct stat file; + unsigned char *all = NULL; + PureResult result = PURE_STORE_ERROR; + if (!fstat(fd, &file) && file.st_size >= 24 && (all = malloc((size_t)file.st_size))) { + size_t have = 0, want = (size_t)file.st_size; + while (have < want) { + ssize_t n = pread(fd, all + have, want - have, (off_t)have); + if (n < 0 && errno == EINTR) continue; + if (n <= 0) break; + have += (size_t)n; + } + uint64_t length = have == want ? get64(all + 8) : UINT64_MAX; + if (have == want && !memcmp(all, image_magic, 8) && length == want - 24 && + get64(all + 16 + length) == digest_of(all + 16, (size_t)length)) { + memmove(all, all + 16, (size_t)length); + *bytes = all; + *size = (size_t)length; + result = PURE_OK; + } + } + if (result != PURE_OK) free(all); + close(fd); + return result; +} + +PureResult pure_image_remove(const char *directory, const char *name) { + char final[64]; + if (!image_name(final, name, "")) return PURE_INVALID; + int dir = open(directory, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (dir < 0) return PURE_STORE_ERROR; + bool ok = (!unlinkat(dir, final, 0) || errno == ENOENT) && sync_whole(dir); + close(dir); + return ok ? PURE_OK : PURE_STORE_ERROR; +} diff --git a/store/test-log.c b/store/test-log.c new file mode 100644 index 0000000..d4382e9 --- /dev/null +++ b/store/test-log.c @@ -0,0 +1,235 @@ +/* The app log (store/store-log.c) against a model of its entries: appends + * of every size (segments of 16 KiB in this build, so entries cross them and + * segments are used again), entries taken back, released, ended, torn on the + * device, and the log opened again after each: what it reads is what the + * model kept, and its files stay as many as the unreleased entries need. + * Then list images: written whole, read back, refused when damaged. */ +#define _POSIX_C_SOURCE 200809L +#include "log.h" +#include "store.h" +#include +#include +#include +#include +#include +#include +#include +#include + +#define CHECK(x) do { if (!(x)) { fprintf(stderr, "%s:%d: %s\n", __FILE__, __LINE__, #x); abort(); } } while (0) + +static int fail_stage = -1; +int pure_store_test_fault(PureStoreStage stage) { return (int)stage == fail_stage; } + +static uint64_t next_random(uint64_t *state) { + uint64_t x = *state; + x ^= x << 13, x ^= x >> 7, x ^= x << 17; + return *state = x; +} + +typedef struct { uint64_t seq; bool tree; unsigned char *bytes; size_t size; } Kept; +typedef struct { Kept *entries; size_t count, capacity; } Model; + +static void keep(Model *m, uint64_t seq, bool tree, const unsigned char *bytes, size_t size) { + if (m->count == m->capacity) { + m->capacity = m->capacity ? 2 * m->capacity : 64; + CHECK((m->entries = realloc(m->entries, m->capacity * sizeof(*m->entries)))); + } + Kept *k = &m->entries[m->count++]; + *k = (Kept){seq, tree, malloc(size ? size : 1), size}; + CHECK(k->bytes); + memcpy(k->bytes, bytes, size); +} + +static void drop_after(Model *m, uint64_t seq) { + while (m->count && m->entries[m->count - 1].seq > seq) free(m->entries[--m->count].bytes); +} + +typedef struct { const Model *model; size_t at; bool stop_after; uint64_t stop; } Reading; + +static bool visit(void *context, uint64_t seq, bool tree, PureSpan payload) { + Reading *r = context; + while (r->at < r->model->count && r->model->entries[r->at].seq < seq) ++r->at; + CHECK(r->at < r->model->count); + const Kept *k = &r->model->entries[r->at++]; + CHECK(k->seq == seq && k->tree == tree && k->size == payload.size && !memcmp(k->bytes, payload.data, payload.size)); + return !(r->stop_after && seq >= r->stop); +} + +static unsigned segment_files(const char *path) { + DIR *dir = opendir(path); + unsigned count = 0; + for (struct dirent *e; dir && (e = readdir(dir));) count += !strncmp(e->d_name, "log-", 4); + if (dir) closedir(dir); + return count; +} + +static char *new_directory(void) { + char *path = strdup("/tmp/pure-log-XXXXXX"); + CHECK(path && mkdtemp(path)); + return path; +} + +static void remove_directory(char *path) { + DIR *dir = opendir(path); + for (struct dirent *e; dir && (e = readdir(dir));) { + if (e->d_name[0] == '.' && (!e->d_name[1] || (e->d_name[1] == '.' && !e->d_name[2]))) continue; + char file[512]; + snprintf(file, sizeof(file), "%s/%s", path, e->d_name); + unlink(file); + } + if (dir) closedir(dir); + rmdir(path); + free(path); +} + +static void logs(void) { + char *path = new_directory(); + uint64_t seed = 0x6c6f6773u, seq = 0, released = 0, retracted_seq = 0; + Model model = {0}; + PureLog *log; + CHECK(pure_log_open(path, &log) == PURE_OK && pure_log_last(log) == 0); + unsigned char *payload = malloc(40000); + CHECK(payload); + unsigned reopened = 0, retracted = 0, ended = 0, most_files = 0; + for (unsigned step = 0; step < 6000; ++step) { + unsigned roll = (unsigned)(next_random(&seed) % 100); + if (roll < 70) { /* an entry, the number sometimes skipping (a save that wrote none) */ + seq += 1 + (next_random(&seed) % 4 == 0 ? next_random(&seed) % 3 : 0); + size_t size = next_random(&seed) % 10 == 0 ? next_random(&seed) % 40000 : next_random(&seed) % 900; + for (size_t i = 0; i < size; ++i) payload[i] = (unsigned char)next_random(&seed); + bool tree = next_random(&seed) % 2; + CHECK(pure_log_append(log, seq, tree, (PureSpan){payload, size}) == PURE_OK); + keep(&model, seq, tree, payload, size); + retracted_seq = 0; + /* Its save failed after it (the tree's commit): taken back, its + * number used again. Until an entry takes its place it is still + * on the device, and a start drops it (a tree entry past the + * tree's last save). */ + if (tree && next_random(&seed) % 12 == 0) { + pure_log_retract(log); + drop_after(&model, seq - 1); + retracted_seq = seq--; + ++retracted; + } + } else if (roll < 80 && model.count) { /* older entries released */ + if (model.entries[model.count / 2].seq > released) released = model.entries[model.count / 2].seq; + pure_log_release(log, released); + } else if (roll < 97) { /* opened again: the kept entries, read from the oldest one not released */ + pure_log_close(log); + CHECK(pure_log_open(path, &log) == PURE_OK); + uint64_t kept_last = model.count ? model.entries[model.count - 1].seq : 0; + if (retracted_seq) { /* found again, and dropped as a start would */ + CHECK(pure_log_last(log) == retracted_seq); + pure_log_end(log, kept_last); + retracted_seq = 0; + } + CHECK(pure_log_last(log) == kept_last); + size_t first = 0; + while (first < model.count && model.entries[first].seq <= released) ++first; + if (first < model.count) { + Reading reading = {&model, first, false, 0}; + CHECK(pure_log_read(log, model.entries[first].seq, visit, &reading) == PURE_OK); + CHECK(reading.at == model.count); + } + if (next_random(&seed) % 4 == 0 && first < model.count) { /* entries after one are dropped */ + uint64_t end = model.entries[first + next_random(&seed) % (model.count - first)].seq; + pure_log_end(log, end); + drop_after(&model, end); + seq = end; + ++ended; + } + pure_log_release(log, released); + ++reopened; + } + unsigned files = segment_files(path); + most_files = files > most_files ? files : most_files; + } + CHECK(reopened > 500 && retracted > 100 && ended > 100 && most_files < 64); + /* A write that fails before its sync is uncertain; an append numbered + * no higher than the last is refused. */ + fail_stage = PURE_LOG_BEFORE_SYNC; + CHECK(pure_log_append(log, seq + 1, false, (PureSpan){payload, 10}) == PURE_UNCERTAIN); + fail_stage = -1; + CHECK(pure_log_append(log, seq, false, (PureSpan){payload, 10}) == PURE_INVALID || !seq); + pure_log_close(log); + /* Bytes past the last entry, as an interrupted write leaves them, end the log there. */ + CHECK(pure_log_open(path, &log) == PURE_OK); + uint64_t last = pure_log_last(log); + CHECK(pure_log_append(log, last + 1, true, (PureSpan){payload, 3000}) == PURE_OK); + pure_log_close(log); + char file[512]; + unsigned damaged = 0; + for (unsigned n = 0; n < 256; ++n) { + snprintf(file, sizeof(file), "%s/log-%u", path, n); + int fd = open(file, O_RDWR); + if (fd < 0) break; + /* The entry just written is the one of the newest segment ending in its 3000 bytes: damage its payload. */ + unsigned char probe[8]; + struct stat st; + CHECK(!fstat(fd, &st)); + for (off_t at = 64; at + 32 + 3000 <= st.st_size; at += 8) { + CHECK(pread(fd, probe, 8, at) == 8); + uint64_t found = 0; + for (unsigned i = 0; i < 8; ++i) found |= (uint64_t)probe[i] << (8 * i); + if (found != last + 1) continue; + unsigned char byte = 0x5a; + CHECK(pwrite(fd, &byte, 1, at + 32 + 1500) == 1); + ++damaged; + break; + } + close(fd); + } + CHECK(damaged >= 1); + CHECK(pure_log_open(path, &log) == PURE_OK && pure_log_last(log) == last); + pure_log_close(log); + for (size_t i = 0; i < model.count; ++i) free(model.entries[i].bytes); + free(model.entries); + free(payload); + remove_directory(path); + printf("{\"log_steps\":6000,\"reopened\":%u,\"retracted\":%u,\"ended\":%u,\"most_segment_files\":%u}\n", reopened, + retracted, ended, most_files); +} + +static void images(void) { + char *path = new_directory(); + unsigned char *bytes; + size_t size; + CHECK(pure_image_read(path, "tasks", &bytes, &size) == PURE_NOT_FOUND); + unsigned char *big = malloc(3 << 20); + CHECK(big); + for (size_t i = 0; i < (3u << 20); ++i) big[i] = (unsigned char)(i * 131 + (i >> 11)); + PureSpan parts[3] = {{(const unsigned char *)"head", 4}, {big, 3u << 20}, {(const unsigned char *)"", 0}}; + CHECK(pure_image_write(path, "tasks", parts, 3) == PURE_OK); + CHECK(pure_image_read(path, "tasks", &bytes, &size) == PURE_OK && size == 4 + (3u << 20)); + CHECK(!memcmp(bytes, "head", 4) && !memcmp(bytes + 4, big, 3u << 20)); + free(bytes); + /* A failure before the rename leaves the image as it was, and no temporary file. */ + fail_stage = PURE_IMAGE_BEFORE_RENAME; + PureSpan other = {(const unsigned char *)"other", 5}; + CHECK(pure_image_write(path, "tasks", &other, 1) == PURE_STORE_ERROR); + fail_stage = -1; + CHECK(pure_image_read(path, "tasks", &bytes, &size) == PURE_OK && size == 4 + (3u << 20)); + free(bytes); + CHECK(pure_image_write(path, "tasks", &other, 1) == PURE_OK); + CHECK(pure_image_read(path, "tasks", &bytes, &size) == PURE_OK && size == 5 && !memcmp(bytes, "other", 5)); + free(bytes); + /* Damaged: refused. */ + char file[512]; + snprintf(file, sizeof(file), "%s/image-tasks", path); + int fd = open(file, O_RDWR); + CHECK(fd >= 0 && pwrite(fd, "X", 1, 17) == 1); + close(fd); + CHECK(pure_image_read(path, "tasks", &bytes, &size) == PURE_STORE_ERROR); + CHECK(pure_image_write(path, "bad/name", &other, 1) == PURE_INVALID); + CHECK(pure_image_remove(path, "tasks") == PURE_OK && pure_image_read(path, "tasks", &bytes, &size) == PURE_NOT_FOUND); + free(big); + remove_directory(path); + puts("{\"images\":\"written-read-kept-on-failure-refused-when-damaged\"}"); +} + +int main(void) { + logs(); + images(); + return 0; +} -- cgit v1.2.3