Write list images a chunk at a time, and keep the log within the lists' size

6 files changed, 116 insertions, 37 deletions

diff --git a/core/log.h b/core/log.h
index 44dbdfb..3edf4e9 100644
--- a/core/log.h
+++ b/core/log.h
@@ -37,10 +37,11 @@ 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;
+ * chunks next gives, one after another, until an empty one (next false:
+ * ALLOCATION, nothing replaced); 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_write(const char *directory, const char *name, bool (*next)(void *, PureSpan *chunk), void *context);
 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/module-list.c b/core/module-list.c
index d69ad98..c2a831a 100644
--- a/core/module-list.c
+++ b/core/module-list.c
@@ -708,6 +708,29 @@ void pure_list_encode(const PureList *list, unsigned char *out) {
     visit_leaves(list->root, encode_leaf, &encoding);
 }
 
+size_t pure_list_encode_part(const PureList *list, size_t *at, unsigned char *out, size_t capacity, size_t *needed) {
+    const Layout *l = &list->layout;
+    PureRun run;
+    size_t used = 0;
+    *needed = 0;
+    if (!pure_list_run(list, *at, &run)) return 0;
+    do {
+        const Node *leaf = run.path[run.depth];
+        for (unsigned i = (unsigned)(*at - run.start); i < leaf->count; ++i) {
+            size_t size = item_bytes(l, leaf, i);
+            if (size > capacity - used) {
+                if (!used) *needed = size;
+                return used;
+            }
+            unsigned char *end = out + used;
+            for (unsigned f = 0; f < l->fields; ++f) end = encode_cell(l, f, cell_at(l, leaf, i, f), end);
+            used += size;
+            ++*at;
+        }
+    } while (pure_list_next(list, &run));
+    return used;
+}
+
 /* An item's cells in stored form at *at in bytes. */
 static bool decode_item(const Layout *l, const unsigned char *bytes, size_t size, size_t *at, PureCell *cells) {
     for (unsigned f = 0; f < l->fields; ++f)
diff --git a/core/module-list.h b/core/module-list.h
index ec7ee10..faacf4b 100644
--- a/core/module-list.h
+++ b/core/module-list.h
@@ -96,6 +96,10 @@ PureResult pure_list_sorted(const PureList *, unsigned field, PureList **out, si
  * pure_list_encode writes them to out. */
 size_t pure_list_size(const PureList *);
 void pure_list_encode(const PureList *, unsigned char *out);
+/* The items from position *at on, as many whole ones as capacity holds,
+ * into out: their bytes (0 at the end, or with *needed the bytes of the one
+ * item that does not fit), *at moved past them. */
+size_t pure_list_encode_part(const PureList *, size_t *at, unsigned char *out, size_t capacity, size_t *needed);
 /* Append the items stored in bytes; each field is checked. */
 PureResult pure_list_read(PureList *, const unsigned char *bytes, size_t size);
 /* A stored list journals its changes (from pure_list_journal(list, true)
diff --git a/core/module-save.c b/core/module-save.c
index f0befb9..c05394f 100644
--- a/core/module-save.c
+++ b/core/module-save.c
@@ -32,7 +32,8 @@ static const unsigned char values_magic[4] = {'P', 'V', 'S', '1'}, image_magic[4
                            list_magic[4] = {'P', 'L', 'S', '2'}, list_magic_1[4] = {'P', 'L', 'S', '1'};
 /* Logged changes a list keeps before its image is written again: at least
  * JOURNAL_MIN, else as many bytes as its items; and whatever the lists, the
- * log is kept within LOG_KEPT by writing the image that holds it back. */
+ * log is kept within as many bytes as all their items (and LOG_KEPT) by
+ * writing the image that holds it back. */
 #ifdef PURE_STORE_TEST
 enum { JOURNAL_MIN = 256 };
 #define LOG_KEPT (UINT64_C(64) << 10)
@@ -261,17 +262,50 @@ void pure_save_release(PureApp *app) {
     pure_log_release(app->log, kept - 1);
 }
 
-PureResult pure_save_image(PureApp *app, const PureModule *m, uint32_t state, const PureList *list) {
-    size_t items = pure_list_size(list);
-    unsigned char header[IMAGE_HEADER], *bytes = malloc(items ? items : 1);
-    if (!bytes) return PURE_ALLOCATION;
-    pure_list_encode(list, bytes);
-    PureSpan parts[2] = {{header, image_header(m, m->state_ref[state], app->seq, header)}, {bytes, items}};
-    PureResult result = pure_image_write(directory_of(app), m->state_names[state], parts, 2);
-    free(bytes);
+/* An image's bytes as the store writes them: its header, then its items a
+ * chunk at a time (one item at least, whatever its size). */
+typedef struct {
+    const PureList *list;
+    size_t at, capacity;
+    const unsigned char *header;
+    size_t header_size;
+    bool started;
+    unsigned char *buffer;
+} Imaging;
+
+static bool image_chunk(void *context, PureSpan *chunk) {
+    Imaging *i = context;
+    if (!i->started) {
+        i->started = true;
+        *chunk = (PureSpan){i->header, i->header_size};
+        return true;
+    }
+    size_t needed, length = pure_list_encode_part(i->list, &i->at, i->buffer, i->capacity, &needed);
+    if (!length && needed) {
+        unsigned char *grown = realloc(i->buffer, needed);
+        if (!grown) return false;
+        i->buffer = grown;
+        i->capacity = needed;
+        length = pure_list_encode_part(i->list, &i->at, i->buffer, i->capacity, &needed);
+    }
+    *chunk = (PureSpan){i->buffer, length};
+    return true;
+}
+
+static PureResult write_image(const char *directory, const char *name, const unsigned char *header, size_t header_size,
+                              const PureList *list) {
+    Imaging imaging = {list, 0, 256 << 10, header, header_size, false, malloc(256 << 10)};
+    PureResult result = imaging.buffer ? pure_image_write(directory, name, image_chunk, &imaging) : PURE_ALLOCATION;
+    free(imaging.buffer);
     return result;
 }
 
+PureResult pure_save_image(PureApp *app, const PureModule *m, uint32_t state, const PureList *list) {
+    unsigned char header[IMAGE_HEADER];
+    size_t header_size = image_header(m, m->state_ref[state], app->seq, header);
+    return write_image(directory_of(app), m->state_names[state], header, header_size, list);
+}
+
 PURE_COLD PureResult pure_save_open_list(PureApp *app, const PureModule *m, uint32_t state, PureList **out,
                                          PureDraft **legacy, PureSaved *saved) {
     uint32_t record = m->state_ref[state];
@@ -399,7 +433,10 @@ PureCheckpoint *pure_app_checkpoint(PureApp *app) {
         if (app->logged[i] > (items > JOURNAL_MIN ? items : JOURNAL_MIN)) due = (int)i;
         else if (oldest < 0 || app->logged_since[i] < app->logged_since[oldest]) oldest = (int)i;
     }
-    int pick = due >= 0 ? due : pure_log_bytes(app->log) > LOG_KEPT ? oldest : -1;
+    uint64_t items = 0;
+    for (uint32_t i = 0; due < 0 && i < m->states; ++i)
+        if (m->state_type[i] == PURE_STATE_LIST && app->lists[i]) items += pure_list_size(app->lists[i]);
+    int pick = due >= 0 ? due : pure_log_bytes(app->log) > (items > LOG_KEPT ? items : LOG_KEPT) ? oldest : -1;
     if (pick < 0) return NULL;
     PureCheckpoint *c = calloc(1, sizeof(*c));
     const char *directory = directory_of(app);
@@ -421,15 +458,7 @@ PureCheckpoint *pure_app_checkpoint(PureApp *app) {
 }
 
 PureResult pure_checkpoint_write(PureCheckpoint *c) {
-    const PureList *list = pure_snapshot_list(c->snapshot, c->state);
-    size_t items = pure_list_size(list);
-    unsigned char *bytes = malloc(items ? items : 1);
-    if (!bytes) return PURE_ALLOCATION;
-    pure_list_encode(list, bytes);
-    PureSpan parts[2] = {{c->header, c->header_size}, {bytes, items}};
-    PureResult result = pure_image_write(c->directory, c->name, parts, 2);
-    free(bytes);
-    return result;
+    return write_image(c->directory, c->name, c->header, c->header_size, pure_snapshot_list(c->snapshot, c->state));
 }
 
 void pure_app_checkpointed(PureApp *app, PureCheckpoint *c, PureResult result) {
diff --git a/store/store-log.c b/store/store-log.c
index 6138a84..f73d2bc 100644
--- a/store/store-log.c
+++ b/store/store-log.c
@@ -478,7 +478,7 @@ static bool image_name(char out[64], const char *name, const char *prefix) {
     return true;
 }
 
-PureResult pure_image_write(const char *directory, const char *name, const PureSpan *parts, size_t count) {
+PureResult pure_image_write(const char *directory, const char *name, bool (*next)(void *, PureSpan *chunk), void *context) {
     char final[64], temporary[96];
     if (!image_name(final, name, "")) return PURE_INVALID;
     static uint64_t sequence;
@@ -488,19 +488,17 @@ PureResult pure_image_write(const char *directory, const char *name, const PureS
     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);
+    put64(head + 8, 0); /* its length, once known */
     Digest digest = {0};
-    bool ok = write_all(fd, head, sizeof(head), 0);
+    bool ok = write_all(fd, head, sizeof(head), 0), produced = true;
     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);
+    for (PureSpan chunk; ok && (produced = next(context, &chunk)) && chunk.size;)
+        for (size_t done = 0; ok && done < chunk.size;) {
+            size_t take = chunk.size - done < STEP ? chunk.size - done : STEP;
+            digest_add(&digest, chunk.data + done, take);
+            ok = write_all(fd, chunk.data + done, take, at);
             done += take;
             at += take;
 #if defined(__linux__)
@@ -517,13 +515,15 @@ PureResult pure_image_write(const char *directory, const char *name, const PureS
         }
     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);
+    put64(head + 8, at - sizeof(head));
+    ok = ok && produced && write_all(fd, tail, sizeof(tail), at) && write_all(fd, head, sizeof(head), 0) &&
+         !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;
+    return ok ? PURE_OK : produced ? PURE_STORE_ERROR : PURE_ALLOCATION;
 }
 
 PureResult pure_image_read(const char *directory, const char *name, unsigned char **bytes, size_t *size) {
diff --git a/store/test-log.c b/store/test-log.c
index d4382e9..451f354 100644
--- a/store/test-log.c
+++ b/store/test-log.c
@@ -191,6 +191,25 @@ static void logs(void) {
            retracted, ended, most_files);
 }
 
+typedef struct { const PureSpan *parts; size_t count, at; } Parts;
+
+static bool next_part(void *context, PureSpan *chunk) {
+    Parts *p = context;
+    *chunk = p->at < p->count ? p->parts[p->at++] : (PureSpan){0};
+    if (chunk->data && !chunk->size) return next_part(context, chunk); /* an empty part is not the end */
+    return true;
+}
+
+static PureResult write_parts(const char *path, const char *name, const PureSpan *parts, size_t count) {
+    Parts p = {parts, count, 0};
+    return pure_image_write(path, name, next_part, &p);
+}
+
+static bool refuse(void *context, PureSpan *chunk) {
+    (void)context, (void)chunk;
+    return false;
+}
+
 static void images(void) {
     char *path = new_directory();
     unsigned char *bytes;
@@ -200,18 +219,18 @@ static void images(void) {
     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(write_parts(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);
+    CHECK(write_parts(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(write_parts(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. */
@@ -221,7 +240,10 @@ static void images(void) {
     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(write_parts(path, "bad/name", &other, 1) == PURE_INVALID);
+    /* A producer that fails replaces nothing. */
+    CHECK(pure_image_write(path, "tasks", refuse, NULL) == PURE_ALLOCATION);
+    CHECK(pure_image_read(path, "tasks", &bytes, &size) == PURE_STORE_ERROR); /* still the damaged one */
     CHECK(pure_image_remove(path, "tasks") == PURE_OK && pure_image_read(path, "tasks", &bytes, &size) == PURE_NOT_FOUND);
     free(big);
     remove_directory(path);