Send messages as part of events and apply delivered ones once (078)

An output (input kind 5) is a message the app sends; SEND (72) takes
an output's field values in order in a handler, lays them out as a
message's payload and keeps it, numbered per output from 1, with the
event's changes: a failed event sent nothing, and a failed save takes
back every send since the last save with its numbers. A host sees only
saved sends (pure_app_outbox) and trims what its receiver has saved
(pure_app_trim, gone with the next save). pure_app_deliver applies a
message from a named source once: a number heard before succeeds
without applying, a refused message counts as heard, a failed save
forgets what it heard.

Sends, numbers and heard numbers are the app's links state, stored
whole in one runtime document (PLK1) with the states, in the same
commit. Outputs follow their names across module changes; a module
that drops or retypes an output with sends left is refused, and a
dropped output's number is kept for when it returns. Killing the
process at every commit stage leaves an event and its send, or a
delivery and its effect, together or neither. pure-serve answers a
post to an output or a timer with 404 no_input.

1 files changed, 186 insertions, 2 deletions

diff --git a/prototypes/connections/module-save.c b/prototypes/connections/module-save.c
index a5eb95a..4149316 100644
--- a/prototypes/connections/module-save.c
+++ b/prototypes/connections/module-save.c
@@ -10,6 +10,12 @@
  * field {u8 0 number | 1 text, u8 name length, name}, then its items
  * (module-list.h). A text state: its UTF-8 bytes. */
 static const unsigned char values_magic[4] = {'P', 'V', 'S', '1'}, list_magic[4] = {'P', 'L', 'S', '1'};
+/* links: "PLK1"; u8 outputs, each {u8 name length, name, u8 field count, the
+ * fields' types, u64le last number} (the module's outputs, then those it
+ * dropped, without fields); u16le sources, each {u8 length, the name, u64le
+ * last number heard}; then the sent messages to the end, each {u8 output (in
+ * the table), u64le number, u32le size, payload}. */
+static const unsigned char links_magic[4] = {'P', 'L', 'K', '1'};
 static const PureVersion unsaved = {UINT64_MAX, UINT64_MAX};
 
 bool pure_save_enabled(const PureApp *app) { return app->data.state && app->data.save; }
@@ -252,6 +258,154 @@ PURE_COLD PureResult pure_save_open_values(PureApp *app, const PureModule *m, do
 
 static bool same_version(PureVersion a, PureVersion b) { return a.ref == b.ref && a.bytes == b.bytes; }
 
+static unsigned char *put_le(unsigned char *at, uint64_t value, unsigned width) {
+    for (unsigned i = 0; i < width; ++i) *at++ = (unsigned char)(value >> (8 * i));
+    return at;
+}
+
+static uint64_t get_le(const unsigned char *at, unsigned width) {
+    uint64_t value = 0;
+    for (unsigned i = 0; i < width; ++i) value |= (uint64_t)at[i] << (8 * i);
+    return value;
+}
+
+/* The links as stored; NULL when memory runs out. */
+static unsigned char *encode_links(const PureApp *app, size_t *size) {
+    const PureModule *m = app->module;
+    unsigned outputs = 0;
+    uint8_t table[PURE_MODULE_INPUTS];
+    size_t total = 4 + 1 + 2;
+    for (uint32_t i = 0; i < m->inputs; ++i)
+        if (m->input[i].kind == PURE_KIND_OUTPUT) {
+            table[i] = (uint8_t)outputs++;
+            total += 2 + strlen(m->input_names[i]) + m->input[i].field_count + 8;
+        }
+    for (unsigned c = 0; c < app->carried_count; ++c) total += 2 + strlen(app->carried[c].name) + 8;
+    for (unsigned h = 0; h < app->heard_count; ++h) total += 1 + strlen(app->heard[h].source) + 8;
+    for (size_t s = 0; s < app->sent_count; ++s)
+        if (!app->sent[s].trimmed) total += 1 + 8 + 4 + app->sent[s].size;
+    unsigned char *bytes = malloc(total), *at = bytes;
+    if (!bytes) return NULL;
+    memcpy(at, links_magic, 4);
+    at += 4;
+    *at++ = (unsigned char)(outputs + app->carried_count);
+    for (uint32_t i = 0; i < m->inputs; ++i) {
+        if (m->input[i].kind != PURE_KIND_OUTPUT) continue;
+        size_t length = strlen(m->input_names[i]);
+        *at++ = (unsigned char)length;
+        memcpy(at, m->input_names[i], length);
+        at += length;
+        *at++ = (unsigned char)m->input[i].field_count;
+        for (uint32_t f = 0; f < m->input[i].field_count; ++f) *at++ = m->field[m->input[i].field_start + f].type;
+        at = put_le(at, app->last_sent[i], 8);
+    }
+    for (unsigned c = 0; c < app->carried_count; ++c) {
+        size_t length = strlen(app->carried[c].name);
+        *at++ = (unsigned char)length;
+        memcpy(at, app->carried[c].name, length);
+        at += length;
+        *at++ = 0;
+        at = put_le(at, app->carried[c].last, 8);
+    }
+    at = put_le(at, app->heard_count, 2);
+    for (unsigned h = 0; h < app->heard_count; ++h) {
+        size_t length = strlen(app->heard[h].source);
+        *at++ = (unsigned char)length;
+        memcpy(at, app->heard[h].source, length);
+        at = put_le(at + length, app->heard[h].seq, 8);
+    }
+    for (size_t s = 0; s < app->sent_count; ++s) {
+        const struct PureSent *sent = &app->sent[s];
+        if (sent->trimmed) continue;
+        *at++ = table[sent->output];
+        at = put_le(at, sent->seq, 8);
+        at = put_le(at, sent->size, 4);
+        if (sent->size) memcpy(at, sent->payload, sent->size);
+        at += sent->size;
+    }
+    *size = (size_t)(at - bytes);
+    return bytes;
+}
+
+/* The stored links of a starting app: outputs by name (an output whose
+ * fields changed or that the module lacks refuses the start while it has
+ * sends left: CONFLICT; one without sends keeps its number for later),
+ * sources and sent messages. */
+PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) {
+    PureResult result = app->stored_links ? PURE_OK : open_state(app, "", "links", &app->stored_links);
+    if (result != PURE_OK || !pure_draft_revision(app->stored_links)) return result;
+    unsigned char *bytes;
+    size_t size;
+    if ((result = read_whole(app->stored_links, &bytes, &size)) != PURE_OK) return result;
+    int output[256];
+    bool fits[256];
+    size_t at = 5;
+    unsigned outputs = size >= 5 && !memcmp(bytes, links_magic, 4) ? bytes[4] : 0;
+    result = size >= 5 && !memcmp(bytes, links_magic, 4) ? PURE_OK : PURE_STORE_ERROR;
+    for (unsigned t = 0; t < outputs && result == PURE_OK; ++t) {
+        char name[PURE_MODULE_NAME];
+        size_t length = at < size ? bytes[at] : 0;
+        if (!length || length >= PURE_MODULE_NAME || size - at < 2 + length) { result = PURE_STORE_ERROR; break; }
+        memcpy(name, bytes + at + 1, length);
+        name[length] = 0;
+        unsigned fields = bytes[at + 1 + length];
+        at += 2 + length;
+        if (size - at < fields + 8) { result = PURE_STORE_ERROR; break; }
+        int i = pure_module_input(m, name);
+        output[t] = i >= 0 && m->input[i].kind == PURE_KIND_OUTPUT ? i : -1;
+        fits[t] = output[t] >= 0 && m->input[i].field_count == fields;
+        for (unsigned f = 0; fits[t] && f < fields; ++f) fits[t] = m->field[m->input[i].field_start + f].type == bytes[at + f];
+        uint64_t last = get_le(bytes + at + fields, 8);
+        at += fields + 8;
+        if (output[t] >= 0) app->last_sent[output[t]] = last;
+        else if (last) {
+            struct PureCarried *grown = realloc(app->carried, (app->carried_count + 1) * sizeof(*grown));
+            if (!grown) { result = PURE_ALLOCATION; break; }
+            app->carried = grown;
+            memcpy(grown[app->carried_count].name, name, length + 1);
+            grown[app->carried_count++].last = last;
+        }
+    }
+    unsigned sources = result == PURE_OK && size - at >= 2 ? (unsigned)get_le(bytes + at, 2) : 0;
+    if (result == PURE_OK && size - at < 2) result = PURE_STORE_ERROR;
+    at += 2;
+    if (result == PURE_OK && sources && !(app->heard = calloc(sources, sizeof(*app->heard)))) result = PURE_ALLOCATION;
+    app->heard_capacity = sources;
+    for (unsigned h = 0; h < sources && result == PURE_OK; ++h) {
+        size_t length = at < size ? bytes[at] : 0;
+        if (!length || length > PURE_SOURCE || size - at < 1 + length + 8) { result = PURE_STORE_ERROR; break; }
+        struct PureHeard *heard = &app->heard[app->heard_count++];
+        memcpy(heard->source, bytes + at + 1, length);
+        heard->seq = heard->saved = get_le(bytes + at + 1 + length, 8);
+        at += 1 + length + 8;
+    }
+    while (result == PURE_OK && at < size) {
+        if (size - at < 13 || bytes[at] >= outputs) { result = PURE_STORE_ERROR; break; }
+        unsigned t = bytes[at];
+        uint64_t seq = get_le(bytes + at + 1, 8);
+        size_t length = (size_t)get_le(bytes + at + 9, 4);
+        at += 13;
+        if (size - at < length) { result = PURE_STORE_ERROR; break; }
+        if (!fits[t]) { result = PURE_CONFLICT; break; }
+        if (app->sent_count == app->sent_capacity) {
+            size_t capacity = app->sent_capacity ? 2 * app->sent_capacity : 16;
+            struct PureSent *grown = realloc(app->sent, capacity * sizeof(*grown));
+            if (!grown) { result = PURE_ALLOCATION; break; }
+            app->sent = grown;
+            app->sent_capacity = capacity;
+        }
+        unsigned char *payload = malloc(length ? length : 1);
+        if (!payload) { result = PURE_ALLOCATION; break; }
+        if (length) memcpy(payload, bytes + at, length);
+        app->sent[app->sent_count++] = (struct PureSent){(uint32_t)output[t], seq, payload, length, false};
+        at += length;
+    }
+    free(bytes);
+    app->sent_durable = app->sent_count;
+    memcpy(app->saved_sent, app->last_sent, sizeof(app->saved_sent));
+    return result;
+}
+
 PureResult pure_save_commit(PureApp *app, const double *values, void *const *documents, size_t document_count,
                             int64_t *revisions) {
     if (!pure_save_enabled(app)) {
@@ -262,8 +416,8 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc
         return result;
     }
     const PureModule *m = app->module;
-    PureDraft *drafts[2 * PURE_MODULE_STATES + 1];
-    uint32_t states[2 * PURE_MODULE_STATES + 1];
+    PureDraft *drafts[2 * PURE_MODULE_STATES + 2];
+    uint32_t states[2 * PURE_MODULE_STATES + 2];
     size_t count = 0;
     for (uint32_t i = 0; i < m->states; ++i) {
         PureDraft *draft = m->state_type[i] == PURE_STATE_TEXT ? app->texts[i] :
@@ -292,6 +446,32 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc
             drafts[count++] = app->stored_values;
         }
     }
+    /* The links, written whole when they changed. */
+    PureVersion links_before = {0, 0};
+    bool links_written = false;
+    if (app->links_changed) {
+        PureResult result = app->stored_links ? PURE_OK : open_state(app, "", "links", &app->stored_links);
+        size_t length;
+        unsigned char *encoded = result == PURE_OK ? encode_links(app, &length) : NULL;
+        if (result == PURE_OK && !encoded) result = PURE_ALLOCATION;
+        if (result == PURE_OK) {
+            links_before = pure_draft_version(app->stored_links);
+            pure_draft_hold(app->stored_links, true);
+            result = pure_draft_splice(app->stored_links, 0, pure_draft_size(app->stored_links), (PureSpan){encoded, length});
+            if (result != PURE_OK) pure_draft_hold(app->stored_links, false);
+        }
+        free(encoded);
+        if (result != PURE_OK) {
+            if (values_changed) {
+                pure_draft_restore(app->stored_values, values_before);
+                pure_draft_hold(app->stored_values, false);
+            }
+            return result;
+        }
+        links_written = true;
+        states[count] = UINT32_MAX;
+        drafts[count++] = app->stored_links;
+    }
     if (!count && !document_count) return PURE_OK;
     PureResult result = app->data.save(app->data.context, documents, document_count, drafts, count, revisions);
     for (size_t k = 0; result == PURE_OK && k < count; ++k)
@@ -300,5 +480,9 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc
         if (result != PURE_OK) pure_draft_restore(app->stored_values, values_before);
         pure_draft_hold(app->stored_values, false);
     }
+    if (links_written) {
+        if (result != PURE_OK) pure_draft_restore(app->stored_links, links_before);
+        pure_draft_hold(app->stored_links, false);
+    }
     return result;
 }