From 9aa123ec5e8827a6101fa7c6db139397b39e04f2 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Wed, 30 Sep 2026 05:33:32 +0100 Subject: Recast waiting sends by field name when an output's fields change Review 004's F3 made a module refuse to start while sends waited on an output whose fields it had changed: their bytes would have meant something else. Safe, but it deadlocks a rolling upgrade. Growing a board shared by two nodes: one upgraded to a card with a new field refuses the other's old sends (another shape), and the other, its old sends waiting, cannot start the new module at all ("Cannot start the module on the store", saying nothing more). Waiting sends now move to the new fields by name, as a list's items do: a value goes over when the new type can hold it (u32 and number are numbers, bytes and text are bytes; a number that is not a u32, or bytes that are not UTF-8 for a text, cannot), the rest start at 0 or empty (pure_send_recast, with the payload layout load uses, pure_fields_lay_out). A start recasts them and a replace does into a new outbox. Their meaning must survive a crash between the recast and the save that names the new fields in the links. So the outbox's image now names the fields its sends were laid out with ("PLO1": the module's outputs after the save it holds), and is written again before any send is logged with other fields; a start reads those names over the links'. Checkpoints of the outbox name them too. A module without an output that sends wait on is still refused, now saying so in PureApp.why, which pure-serve prints ("sends wait on the output 'ordered', which this module does not have"); a failed start keeps it. test-module send_recast (widened, narrowed, UTF-8, a payload not laid out, there and back) and output_meaning (the swapped fields recast on a replace, read right after a restart that never saved, back again, sends logged after the image, the refusal and its reason). check.py server_upgrade: five sends wait on the old output, the sender starts with the new one, the old receiver refuses the new shape, the new one takes all eight. --- core/module-save.c | 191 ++++++++++++++++++++++++++++++++++++++++---- core/module-save.h | 28 ++++++- core/module.c | 230 +++++++++++++++++++++++++++++++++++++++++++++-------- core/module.h | 1 + core/test-module.c | 107 +++++++++++++++++++++---- serve/serve.c | 2 +- tests/check.py | 123 ++++++++++++++++++++++++++++ 7 files changed, 616 insertions(+), 66 deletions(-) diff --git a/core/module-save.c b/core/module-save.c index e737276..414d6b7 100644 --- a/core/module-save.c +++ b/core/module-save.c @@ -30,6 +30,13 @@ * its image and empties it. */ static const unsigned char values_magic[4] = {'P', 'V', 'S', '1'}, image_magic[4] = {'P', 'L', 'I', '1'}, list_magic[4] = {'P', 'L', 'S', '2'}, list_magic_1[4] = {'P', 'L', 'S', '1'}; +/* The outbox's image names the fields its sends were laid out with: "PLO1" + * in the place of "PLI1", and after the save it holds u8 outputs, each {u8 + * name length, name, u8 field count, each field's {u8 type, u8 name length, + * name}} (the module's outputs when it was written). The sends logged after + * it are laid out so too: a module whose outputs differ writes it again + * first. So what they mean moves with them, whatever the links hold. */ +static const unsigned char sends_magic[4] = {'P', 'L', 'O', '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 as many bytes as all their items (and LOG_KEPT) by @@ -149,7 +156,8 @@ PURE_COLD PureResult pure_save_open_text(PureApp *app, const PureModule *m, uint /* ---- Lists: images and the log */ -enum { IMAGE_HEADER = 5 + PURE_RECORD_FIELDS * (2 + PURE_MODULE_NAME) + 8 }; +enum { OUTPUTS_NAMED = 1 + PURE_MODULE_INPUTS * (2 + PURE_MODULE_NAME) + PURE_MODULE_FIELDS * (2 + PURE_MODULE_NAME), + IMAGE_HEADER = 5 + PURE_RECORD_FIELDS * (2 + PURE_MODULE_NAME) + 8 + OUTPUTS_NAMED }; static size_t header_of(unsigned count, const uint8_t *text, const char (*names)[PURE_MODULE_NAME], uint64_t seq, unsigned char *out) { @@ -181,6 +189,71 @@ static size_t runtime_header(uint32_t place, uint64_t seq, unsigned char *out) { return header_of(r->fields, r->kinds, r->names, seq, out); } +/* The outbox's image header: m's outputs named after the save it holds. */ +static size_t outbox_header(const PureModule *m, uint64_t seq, unsigned char *out) { + size_t at = runtime_header(PURE_LIST_OUTBOX, seq, out); + memcpy(out, sends_magic, 4); + unsigned char *count = out + at++; + *count = 0; + for (uint32_t i = 0; i < m->inputs; ++i) { + if (m->input[i].kind != PURE_KIND_OUTPUT) continue; + ++*count; + size_t length = strlen(m->input_names[i]); + out[at++] = (unsigned char)length; + memcpy(out + at, m->input_names[i], length); + at += length; + out[at++] = (unsigned char)m->input[i].field_count; + for (uint32_t f = 0; f < m->input[i].field_count; ++f) { + const PureField *field = &m->field[m->input[i].field_start + f]; + size_t named = strlen(field->name); + out[at++] = field->type; + out[at++] = (unsigned char)named; + memcpy(out + at, field->name, named); + at += named; + } + } + return at; +} + +/* What an output's sends were laid out with, as a store names it: its + * fields (laid out), or their types alone (named false: links before PLK4). */ +typedef struct { char name[PURE_MODULE_NAME]; PureField fields[PURE_MODULE_FIELDS]; uint32_t count; bool named; } SentShape; + +/* The outputs an outbox image names (at, size bytes): how many, or -1 when + * they are not well formed; shapes (when not NULL) takes them. */ +static int read_outputs(const unsigned char *at, size_t size, SentShape *shapes, size_t *used) { + size_t k = 0; + if (size < 1) return -1; + unsigned outputs = at[k++]; + for (unsigned t = 0; t < outputs; ++t) { + size_t length = k < size ? at[k] : 0; + if (!length || length >= PURE_MODULE_NAME || size - k < 2 + length) return -1; + unsigned fields = at[k + 1 + length]; + if (fields > PURE_MODULE_FIELDS) return -1; + if (shapes) { + memcpy(shapes[t].name, at + k + 1, length); + shapes[t].name[length] = 0; + shapes[t].count = fields; + shapes[t].named = true; + } + k += 2 + length; + for (unsigned f = 0; f < fields; ++f) { + size_t named = size - k >= 2 ? at[k + 1] : 0; + if (size - k < 2 || !named || named >= PURE_MODULE_NAME || size - k < 2 + named || !at[k] || + at[k] >= PURE_FIELD_ROWS) + return -1; + if (shapes) { + shapes[t].fields[f] = (PureField){.type = at[k]}; + memcpy(shapes[t].fields[f].name, at + k + 2, named); + } + k += 2 + named; + } + if (shapes) pure_fields_lay_out(shapes[t].fields, fields); + } + *used = k; + return (int)outputs; +} + /* A stored list by its place: a list state, or one of the runtime's. */ static PureList *stored_list(const PureApp *app, uint32_t i) { if (i == PURE_LIST_HEARD) return app->heard; @@ -204,6 +277,8 @@ struct PureStartLog { size_t count, capacity; uint64_t tree, last; /* the tree's last save; the last entry that counts */ bool stopped, damaged; + SentShape *sent; /* the outputs the outbox's image names (PLO1), sent_count of them */ + int sent_count; /* -1: its image names none */ }; static struct Named *named_list(struct PureStartLog *index, const char *name, size_t length, bool add) { @@ -259,6 +334,7 @@ void pure_save_close_start(PureApp *app) { if (!index) return; for (size_t i = 0; i < index->count; ++i) free(index->lists[i].changes); free(index->lists); + free(index->sent); free(index); app->start_log = NULL; } @@ -270,6 +346,7 @@ PURE_COLD PureResult pure_save_open_log(PureApp *app) { if (result == PURE_OK) result = open_state(app, "", "seq", &app->stored_seq); struct PureStartLog *index = result == PURE_OK ? calloc(1, sizeof(*index)) : NULL; if (result == PURE_OK && !index) result = PURE_ALLOCATION; + if (index) index->sent_count = -1; if (result == PURE_OK && pure_draft_revision(app->stored_seq)) { unsigned char *bytes; size_t size; @@ -455,14 +532,31 @@ PURE_COLD PureResult pure_save_open_runtime(PureApp *app, uint32_t place, PureLi *out = NULL; if (!directory_of(app)) return PURE_STORE_ERROR; unsigned char *bytes = NULL, header[IMAGE_HEADER]; - size_t size = 0, length = runtime_header(place, 0, header) - 8; + size_t size = 0, length = runtime_header(place, 0, header) - 8, at = length + 8; PureResult result = pure_image_read(directory_of(app), name, &bytes, &size); bool image = result == PURE_OK; - uint64_t holds = image && size >= length + 8 && !memcmp(bytes, header, length) ? get_le(bytes + length, 8) : 0; - if (image && (size < length + 8 || memcmp(bytes, header, length))) result = PURE_STORE_ERROR; + bool named = image && place == PURE_LIST_OUTBOX && size >= 4 && !memcmp(bytes, sends_magic, 4); + bool formed = image && size >= at && (named || !memcmp(bytes, image_magic, 4)) && + !memcmp(bytes + 4, header + 4, length - 4); + uint64_t holds = formed ? get_le(bytes + length, 8) : 0; + if (formed && named) { /* the outputs its sends were laid out with */ + size_t used = 0; + int outputs = read_outputs(bytes + at, size - at, NULL, &used); + SentShape *shapes = outputs > 0 ? calloc((size_t)outputs, sizeof(*shapes)) : NULL; + if (outputs < 0 || (outputs > 0 && !shapes)) formed = false; + else if (app->start_log) { + read_outputs(bytes + at, size - at, shapes, &used); + free(app->start_log->sent); + app->start_log->sent = shapes, app->start_log->sent_count = outputs; + shapes = NULL; + } + free(shapes); + at += used; + } + if (image && !formed) result = PURE_STORE_ERROR; else if (image || result == PURE_NOT_FOUND) { result = (*out = pure_runtime_list_new(place)) ? PURE_OK : PURE_ALLOCATION; - if (result == PURE_OK && image) result = pure_list_read(*out, bytes + length + 8, size - length - 8); + if (result == PURE_OK && image) result = pure_list_read(*out, bytes + at, size - at); } free(bytes); /* Then the changes logged after the save it holds (an image is written @@ -487,11 +581,12 @@ PURE_COLD PureResult pure_save_open_runtime(PureApp *app, uint32_t place, PureLi return PURE_OK; } -PureResult pure_save_runtime_image(PureApp *app, uint32_t place) { +PureResult pure_save_runtime_image(PureApp *app, const PureModule *m, uint32_t place, const PureList *list) { unsigned char header[IMAGE_HEADER]; - size_t header_size = runtime_header(place, app->seq, header); + size_t header_size = place == PURE_LIST_OUTBOX ? outbox_header(m, app->seq, header) : + runtime_header(place, app->seq, header); return write_image(directory_of(app), runtime_lists[place - PURE_LIST_HEARD].name, header, header_size, - stored_list(app, place)); + list ? list : stored_list(app, place)); } /* ---- Checkpoints */ @@ -536,7 +631,8 @@ PureCheckpoint *pure_app_checkpoint(PureApp *app) { c->state = (uint32_t)pick; snprintf(c->name, sizeof(c->name), "%s", stored_name(app, (uint32_t)pick)); c->seq = app->seq; - c->header_size = pick >= PURE_LIST_HEARD ? runtime_header((uint32_t)pick, c->seq, c->header) : + c->header_size = pick == PURE_LIST_OUTBOX ? outbox_header(m, c->seq, c->header) : + pick >= PURE_LIST_HEARD ? runtime_header((uint32_t)pick, c->seq, c->header) : image_header(m, m->state_ref[pick], c->seq, c->header); c->logged = app->logged[pick]; c->since = app->logged_since[pick]; @@ -722,6 +818,24 @@ PURE_COLD static PureResult sent_before(PureApp *app, const char *output, uint64 * for later); the sources and sends links kept before the runtime's lists * go to them (heard_moved, sent_moved: they need images). Links kept * before PLK4 have their fields' types alone. */ +/* The shape of the output of that name among shapes, or NULL. */ +static const SentShape *sent_shape(const SentShape *shapes, unsigned count, const char *name) { + for (unsigned t = 0; t < count; ++t) + if (!strcmp(shapes[t].name, name)) return &shapes[t]; + return NULL; +} + +/* Whether sends laid out as was are laid out as m's output i: the same + * fields, names and types in order (types alone when was names none). */ +static bool same_shape(const SentShape *was, const PureModule *m, uint32_t i) { + if (was->count != m->input[i].field_count) return false; + for (uint32_t f = 0; f < was->count; ++f) { + const PureField *field = &m->field[m->input[i].field_start + f]; + if (field->type != was->fields[f].type || (was->named && strcmp(field->name, was->fields[f].name))) return false; + } + return true; +} + PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m, bool *heard_moved, bool *sent_moved) { *heard_moved = *sent_moved = false; PureResult result = app->stored_links ? PURE_OK : open_state(app, "", "links", &app->stored_links); @@ -738,6 +852,8 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m, boo known = first || second || named || (size >= 5 && !memcmp(bytes, links_magic_3, 4)); unsigned outputs = known ? bytes[4] : 0; result = known ? PURE_OK : PURE_STORE_ERROR; + SentShape *linked = calloc(outputs ? outputs : 1, sizeof(*linked)); /* what each was sent with, as the links say */ + if (!linked) result = PURE_ALLOCATION; for (unsigned t = 0; t < outputs && result == PURE_OK; ++t) { char *name = names[t]; size_t length = at < size ? bytes[at] : 0; @@ -749,14 +865,24 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m, boo 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; + memcpy(linked[t].name, name, length + 1); + linked[t].count = fields; + linked[t].named = named; + if (fields > PURE_MODULE_FIELDS) result = PURE_STORE_ERROR; for (unsigned f = 0; f < fields && result == PURE_OK; ++f) { /* its type, then (PLK4) its name */ size_t named_length = named && size - at >= 2 ? bytes[at + 1] : 0; - if (size - at < (named ? 2 + named_length : 1)) { result = PURE_STORE_ERROR; break; } + if (size - at < (named ? 2 + named_length : 1) || named_length >= PURE_MODULE_NAME) { + result = PURE_STORE_ERROR; + break; + } const PureField *field = fits[t] ? &m->field[m->input[i].field_start + f] : NULL; fits[t] = field && field->type == bytes[at] && (!named || (strlen(field->name) == named_length && !memcmp(field->name, bytes + at + 2, named_length))); + linked[t].fields[f] = (PureField){.type = bytes[at]}; + if (named) memcpy(linked[t].fields[f].name, bytes + at + 2, named_length); at += named ? 2 + named_length : 1; } + if (result == PURE_OK) pure_fields_lay_out(linked[t].fields, fields); if (result != PURE_OK || size - at < 8) { result = PURE_STORE_ERROR; break; } uint64_t last = get_le(bytes + at, 8); at += 8; @@ -795,13 +921,48 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m, boo app->links_changed = true; } if (result == PURE_OK && !first && !second && at != size) result = PURE_STORE_ERROR; - /* Every send left goes out on an output of this module with its fields. */ + /* Every send left goes out on an output of this module, laid out with + * its fields: one laid out otherwise (as the outbox's image names them, + * else as the links do) is recast by name. */ + bool imaged = app->start_log && app->start_log->sent_count >= 0; + const SentShape *shapes = imaged ? app->start_log->sent : linked; + unsigned shape_count = imaged ? (unsigned)app->start_log->sent_count : outputs; for (size_t k = 0, count = pure_list_count(app->outbox); result == PURE_OK && k < count; ++k) { - PureCell name = pure_list_get(app->outbox, k, 0); - unsigned t = 0; - while (t < outputs && (strlen(names[t]) != name.size || memcmp(names[t], name.text, name.size))) ++t; - if (t == outputs || !fits[t]) result = PURE_CONFLICT; + PureCell cell = pure_list_get(app->outbox, k, 0); + char name[PURE_MODULE_NAME] = ""; + if (cell.size < PURE_MODULE_NAME) memcpy(name, cell.text, cell.size); + int i = pure_module_input(m, name); + const SentShape *was = sent_shape(shapes, shape_count, name); + if (i < 0 || m->input[i].kind != PURE_KIND_OUTPUT) { + snprintf(app->why, sizeof(app->why), "sends wait on the output '%s', which this module does not have", name); + result = PURE_CONFLICT; + } else if (!was) result = PURE_STORE_ERROR; /* sent on an output its store does not name */ + else if (!same_shape(was, m, (uint32_t)i)) { + PureCell payload = pure_list_get(app->outbox, k, 2); + unsigned char *made = NULL; + size_t made_size = 0; + if (!was->named) { + snprintf(app->why, sizeof(app->why), + "sends wait on '%s', laid out with fields its store names by type alone: start the old module once", + name); + result = PURE_CONFLICT; + } else if ((result = pure_send_recast(was->fields, was->count, &m->field[m->input[i].field_start], + m->input[i].field_count, (PureSpan){payload.text, payload.size}, + &made, &made_size)) == PURE_OK) { + result = pure_list_set(app->outbox, k, 2, (PureCell){.text = made, .size = made_size}, NULL); + *sent_moved = true; + } + free(made); + } + } + /* Sends from now on are laid out with this module's fields: an image + * naming others is written again, naming these. */ + for (uint32_t i = 0; result == PURE_OK && i < m->inputs && !*sent_moved; ++i) { + if (m->input[i].kind != PURE_KIND_OUTPUT) continue; + const SentShape *was = sent_shape(shapes, shape_count, m->input_names[i]); + *sent_moved = !was || !same_shape(was, m, i); } + free(linked); free(bytes); app->sent_saved = pure_list_count(app->outbox); memcpy(app->saved_sent, app->last_sent, sizeof(app->saved_sent)); diff --git a/core/module-save.h b/core/module-save.h index 1d6440d..d17cace 100644 --- a/core/module-save.h +++ b/core/module-save.h @@ -34,7 +34,10 @@ PureSnapshot *pure_save_snapshot(PureApp *, uint32_t runtime); * changes made again, or new (saved.rewrite: it needs an image, which * pure_save_runtime_image writes). */ PureResult pure_save_open_runtime(PureApp *, uint32_t place, PureList **out, PureSaved *saved); -PureResult pure_save_runtime_image(PureApp *, uint32_t place); +/* A runtime list's image, holding the last save: list, or the app's when + * NULL. The outbox's names the fields its sends were laid out with, m's + * outputs' ("PLO1"): what they mean whatever the links hold. */ +PureResult pure_save_runtime_image(PureApp *, const PureModule *m, uint32_t place, const PureList *list); /* An empty runtime list. */ PureList *pure_runtime_list_new(uint32_t place); /* Open the stored numbers, cursors and indices, and put each stored one of the @@ -43,15 +46,32 @@ PureList *pure_runtime_list_new(uint32_t place); PureResult pure_save_open_values(PureApp *, const PureModule *m, double *values, const int *kept, PureDraft *const *texts, PureList *const *lists); /* The stored links of a starting app (module.h, PureApp), after its runtime - * lists; CONFLICT when an output with sends left in the outbox is missing or - * has other fields. Links stored before the runtime lists kept the numbers - * heard and the sends: they go to them (moved: they need images). */ + * lists. The sends left in the outbox of an output whose fields changed are + * recast to m's by name (pure_send_recast; sent_moved: the outbox needs an + * image, as it does when m's outputs are not those its image names); + * CONFLICT (app->why says) when one waits on an output m lacks, or on one + * whose fields its store does not name (PLK3). Links stored before the + * runtime lists kept the numbers heard and the sends: they go to them + * (moved: they need images). */ PureResult pure_save_open_links(PureApp *, const PureModule *m, bool *heard_moved, bool *sent_moved); /* Publish the documents and every state changed since the last save * (values are the states' numbers); revisions are the documents'. */ PureResult pure_save_commit(PureApp *, const double *values, void *const *documents, size_t document_count, int64_t *revisions); +/* A message's fields laid out as load lays them: in order, the fixed ones + * and the variable ones' lengths in the prefix, but the last variable + * field's, which takes what follows the others. The prefix's size. */ +uint32_t pure_fields_lay_out(PureField *fields, uint32_t count); +/* A send's payload, laid out for from's fields, made anew for to's (both + * laid out): each field takes the value of from's of the same name when + * its type can hold it (u32 and number are numbers, bytes and text are + * bytes; a number that is not a u32, or bytes that are not UTF-8 for a + * text, cannot), else it is 0 or empty. The same rule moves a list's items + * (pure_list_migrate). STORE_ERROR when payload is not laid out by from. */ +PureResult pure_send_recast(const PureField *from, uint32_t from_count, const PureField *to, uint32_t to_count, + PureSpan payload, unsigned char **out, size_t *size); + /* A list with record's fields, taking from `from` each field of the same name * and kind and giving the others their defaults (module.c). */ PureResult pure_list_migrate(PureList **out, const PureList *from, const char *const *names, diff --git a/core/module.c b/core/module.c index b370f4b..8a57595 100644 --- a/core/module.c +++ b/core/module.c @@ -133,6 +133,101 @@ static double read_double(const unsigned char *p) { return value; } +static void write_double(unsigned char *p, double value) { + uint64_t bits; + memcpy(&bits, &value, sizeof(bits)); + write32(p, (uint32_t)bits); + write32(p + 4, (uint32_t)(bits >> 32)); +} + +static bool variable_field(uint8_t type) { + return type == PURE_FIELD_BYTES || type == PURE_FIELD_TEXT || type == PURE_FIELD_ROWS; +} + +uint32_t pure_fields_lay_out(PureField *fields, uint32_t count) { + uint32_t last = UINT32_MAX, prefix = 0; + for (uint32_t f = 0; f < count; ++f) + if (variable_field(fields[f].type)) last = f; + for (uint32_t f = 0; f < count; ++f) { + fields[f].offset = f == last ? PURE_FIELD_REST : prefix; + if (f != last) prefix += fields[f].type == PURE_FIELD_NUMBER ? 8 : 4; + } + return prefix; +} + +/* A field's value in a payload: a number, or bytes (of the payload). */ +typedef struct { double num; PureSpan bytes; } FieldValue; + +/* Each field's value in payload, laid out by fields (their offsets set): + * false when the payload is not so laid out. */ +static bool read_fields(const PureField *fields, uint32_t count, PureSpan payload, FieldValue *values) { + uint32_t prefix = 0; + for (uint32_t f = 0; f < count; ++f) + if (fields[f].offset != PURE_FIELD_REST) prefix += fields[f].type == PURE_FIELD_NUMBER ? 8 : 4; + if (payload.size < prefix) return false; + size_t at = prefix; + for (uint32_t f = 0; f < count; ++f) { + const PureField *field = &fields[f]; + values[f] = (FieldValue){0}; + if (field->type == PURE_FIELD_U32) values[f].num = read32(payload.data + field->offset); + else if (field->type == PURE_FIELD_NUMBER) values[f].num = read_double(payload.data + field->offset); + else if (field->type == PURE_FIELD_BYTES || field->type == PURE_FIELD_TEXT) { + size_t size = field->offset == PURE_FIELD_REST ? payload.size - at : read32(payload.data + field->offset); + if (size > payload.size - at) return false; + values[f].bytes = (PureSpan){payload.data + at, size}; + at += size; + } else return false; /* rows: an output sends none */ + } + return at == payload.size; +} + +PureResult pure_send_recast(const PureField *from, uint32_t from_count, const PureField *to, uint32_t to_count, + PureSpan payload, unsigned char **out, size_t *size) { + FieldValue was[PURE_MODULE_FIELDS], now[PURE_MODULE_FIELDS]; + *out = NULL; + *size = 0; + if (from_count > PURE_MODULE_FIELDS || to_count > PURE_MODULE_FIELDS || !read_fields(from, from_count, payload, was)) + return PURE_STORE_ERROR; + size_t total = 0; + for (uint32_t g = 0; g < to_count; ++g) { + const PureField *field = &to[g]; + now[g] = (FieldValue){0}; + uint32_t f = 0; + while (f < from_count && strcmp(from[f].name, field->name)) ++f; + bool number = field->type == PURE_FIELD_U32 || field->type == PURE_FIELD_NUMBER; + if (f < from_count && number && (from[f].type == PURE_FIELD_U32 || from[f].type == PURE_FIELD_NUMBER)) { + double v = was[f].num; + bool fits = field->type == PURE_FIELD_NUMBER || (v >= 0 && v <= UINT32_MAX && v == (double)(uint32_t)v); + now[g].num = fits ? v : 0; + } else if (f < from_count && !number && (from[f].type == PURE_FIELD_BYTES || from[f].type == PURE_FIELD_TEXT)) { + bool fits = field->type == PURE_FIELD_BYTES || from[f].type == PURE_FIELD_TEXT || pure_utf8_valid(was[f].bytes); + if (fits) now[g].bytes = was[f].bytes; + } + total += field->offset == PURE_FIELD_REST || number ? 0 : 4; + total += number ? (field->type == PURE_FIELD_NUMBER ? 8 : 4) : now[g].bytes.size; + } + unsigned char *made = malloc(total ? total : 1); + if (!made) return PURE_ALLOCATION; + uint32_t prefix = 0; + for (uint32_t g = 0; g < to_count; ++g) + if (to[g].offset != PURE_FIELD_REST) prefix += to[g].type == PURE_FIELD_NUMBER ? 8 : 4; + size_t at = prefix; + for (uint32_t g = 0; g < to_count; ++g) { + const PureField *field = &to[g]; + unsigned char *place = field->offset == PURE_FIELD_REST ? made + at : made + field->offset; + if (field->type == PURE_FIELD_U32) write32(place, (uint32_t)now[g].num); + else if (field->type == PURE_FIELD_NUMBER) write_double(place, now[g].num); + else { + if (field->offset != PURE_FIELD_REST) write32(place, (uint32_t)now[g].bytes.size); + if (now[g].bytes.size) memcpy(made + at, now[g].bytes.data, now[g].bytes.size); + at += now[g].bytes.size; + } + } + *out = made; + *size = at; + return PURE_OK; +} + PURE_COLD static bool key_name(const char *key) { static const char *const named[] = {"Space", "Enter", "ArrowLeft", "ArrowRight", "ArrowUp", "ArrowDown", "Backspace", "Delete", "Home", "End", "Tab", "Escape", "PageUp", "PageDown"}; @@ -914,13 +1009,6 @@ PURE_COLD PureResult pure_module_load(const void *bytes, size_t size, PureModule ((shape->kind == PURE_KIND_OUTPUT || arguments) && shape->keyed) || shape->field_start != next_field || shape->field_count > out->fields - next_field) REFUSE("message shape", i); shape->view = arguments ? read32(at + 8) : 0; - /* The prefix holds the fixed fields and the variable fields' lengths - * in order, but the last variable field's: it takes the rest. */ - uint32_t last = UINT32_MAX; - for (uint32_t f = 0; f < shape->field_count; ++f) { - uint32_t type = read32(fields + (size_t)(next_field + f) * FIELD_BYTES + 4) & 0xFF; - if (type == PURE_FIELD_BYTES || type == PURE_FIELD_TEXT || type == PURE_FIELD_ROWS) last = f; - } for (uint32_t f = 0; f < shape->field_count; ++f) { const unsigned char *entry = fields + (size_t)(next_field + f) * FIELD_BYTES; PureField *field = &out->field[next_field + f]; @@ -947,9 +1035,8 @@ PURE_COLD PureResult pure_module_load(const void *bytes, size_t size, PureModule if ((la < lb && !memcmp(a, b, la) && b[la] == '.') || (lb < la && !memcmp(a, b, lb) && a[lb] == '.')) REFUSE("field and group", next_field + f); } - field->offset = f == last ? PURE_FIELD_REST : shape->prefix; - if (f != last) shape->prefix += field->type == PURE_FIELD_NUMBER ? 8 : 4; } + shape->prefix = pure_fields_lay_out(&out->field[next_field], shape->field_count); next_field += shape->field_count; } if (next_field != out->fields) REFUSE("unused message fields", next_field); @@ -3325,31 +3412,85 @@ PURE_COLD static PureResult keep_list(PureList **out, PureList *from, const Pure * again, and the next event or save publishes the module's states; a state * the module drops stays stored as it was. Starting an app opens its module * over none. */ -/* Outputs of old that module has by name and with the same fields (their - * names and types, in order: what a send's bytes mean) keep their sends and - * numbers; one it lacks or changed refuses the module while sends are left - * on it (CONFLICT), and otherwise keeps its number for later. */ -PURE_COLD static PureResult remap_outputs(PureApp *app, const PureModule *old, const PureModule *module, bool apply) { +/* Whether module's output `to` lays out its fields as old's `from` does: + * the same names and types, in order (what a send's bytes mean). */ +static bool same_fields(const PureModule *old, uint32_t from, const PureModule *module, uint32_t to) { + if (module->input[to].field_count != old->input[from].field_count) return false; + for (uint32_t f = 0; f < old->input[from].field_count; ++f) { + const PureField *now = &module->field[module->input[to].field_start + f], + *was = &old->field[old->input[from].field_start + f]; + if (now->type != was->type || strcmp(now->name, was->name)) return false; + } + return true; +} + +/* Outputs of old that module has by name keep their sends and numbers; + * the sends of one whose fields changed are recast to its new fields by + * name (pure_send_recast), into a new outbox (*recast; NULL when none is). + * One it lacks refuses the module while sends are left on it (CONFLICT, + * app->why says), and otherwise keeps its number for later. *changed: an + * output of module is laid out otherwise than old's (or old has none of + * that name), so the outbox's image, which names them, is written again. */ +PURE_COLD static PureResult remap_outputs(PureApp *app, const PureModule *old, const PureModule *module, bool apply, + PureList **recast, bool *changed) { int map[PURE_MODULE_INPUTS]; + bool differs[PURE_MODULE_INPUTS] = {false}, recasting = false; for (uint32_t i = 0; i < old->inputs; ++i) { map[i] = -1; if (old->input[i].kind != PURE_KIND_OUTPUT) continue; int to = pure_module_input(module, old->input_names[i]); - bool same = to >= 0 && module->input[to].kind == PURE_KIND_OUTPUT && - module->input[to].field_count == old->input[i].field_count; - for (uint32_t f = 0; same && f < old->input[i].field_count; ++f) { - const PureField *now = &module->field[module->input[to].field_start + f], - *was = &old->field[old->input[i].field_start + f]; - same = now->type == was->type && !strcmp(now->name, was->name); + if (to < 0 || module->input[to].kind != PURE_KIND_OUTPUT) continue; + map[i] = to; + differs[i] = !same_fields(old, i, module, (uint32_t)to); + } + if (!apply) { + *recast = NULL; + *changed = false; + for (uint32_t k = 0; k < module->inputs && !*changed; ++k) { + if (module->input[k].kind != PURE_KIND_OUTPUT) continue; + int from = pure_module_input(old, module->input_names[k]); + *changed = from < 0 || old->input[from].kind != PURE_KIND_OUTPUT || differs[from]; + } + size_t count = app->outbox ? pure_list_count(app->outbox) : 0; + for (size_t s = 0; s < count; ++s) { + PureCell cell = pure_list_get(app->outbox, s, 0); + char name[PURE_MODULE_NAME] = ""; + if (cell.size < PURE_MODULE_NAME) memcpy(name, cell.text, cell.size); + int i = pure_module_input(old, name); + if (i < 0 || map[i] < 0) { + snprintf(app->why, sizeof(app->why), "sends wait on the output '%s', which this module does not have", + name); + return PURE_CONFLICT; + } + recasting = recasting || differs[i]; + } + if (!recasting) return PURE_OK; + PureList *made = pure_runtime_list_new(PURE_LIST_OUTBOX); + PureResult result = made ? PURE_OK : PURE_ALLOCATION; + for (size_t s = 0; result == PURE_OK && s < count; ++s) { + PureCell item[3] = {pure_list_get(app->outbox, s, 0), pure_list_get(app->outbox, s, 1), + pure_list_get(app->outbox, s, 2)}; + char name[PURE_MODULE_NAME] = ""; + memcpy(name, item[0].text, item[0].size); + int i = pure_module_input(old, name); + unsigned char *bytes = NULL; + size_t size = 0; + if (differs[i]) { + const PureInputShape *was = &old->input[i], *now = &module->input[map[i]]; + result = pure_send_recast(&old->field[was->field_start], was->field_count, &module->field[now->field_start], + now->field_count, (PureSpan){item[2].text, item[2].size}, &bytes, &size); + item[2] = (PureCell){.text = bytes, .size = size}; + } + if (result == PURE_OK) result = pure_list_insert(made, s, item, NULL); + free(bytes); } - map[i] = same ? to : -1; - size_t length = strlen(old->input_names[i]); - for (size_t s = 0, count = app->outbox ? pure_list_count(app->outbox) : 0; !same && s < count; ++s) { - PureCell name = pure_list_get(app->outbox, s, 0); - if (name.size == length && !memcmp(name.text, old->input_names[i], length)) return PURE_CONFLICT; + if (result != PURE_OK) { + pure_list_free(made); + return result; } + *recast = made; + return PURE_OK; } - if (!apply) return PURE_OK; uint64_t last[PURE_MODULE_INPUTS] = {0}, saved[PURE_MODULE_INPUTS] = {0}; for (uint32_t i = 0; i < old->inputs; ++i) { if (old->input[i].kind != PURE_KIND_OUTPUT || !app->last_sent[i]) continue; @@ -3389,6 +3530,7 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) bool rewrite[PURE_MODULE_STATES] = {false}; /* Lists are stored where the host names a directory for them (module-save.c). */ bool lists_stored = stored && app->data.directory && app->data.directory(app->data.context); + app->why[0] = 0; if (old && app->checkpoint) return PURE_CONFLICT; /* the host has a checkpoint of the running module out */ PureResult result = !old && lists_stored ? pure_save_open_log(app) : PURE_OK; /* The numbers heard and the outbox belong to the app, whatever its module. */ @@ -3455,8 +3597,10 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) /* Outputs keep their sends and numbers by name. */ bool heard_moved = false, sent_moved = false; if (result == PURE_OK && !old && !stored) app->sent_saved = pure_list_count(app->outbox); + PureList *recast = NULL; /* the outbox, its sends laid out with module's fields */ + bool outputs_changed = false; if (result == PURE_OK) - result = old ? remap_outputs(app, old, module, false) : + result = old ? remap_outputs(app, old, module, false, &recast, &outputs_changed) : stored ? pure_save_open_links(app, module, &heard_moved, &sent_moved) : PURE_OK; heard.rewrite = heard.rewrite || (heard_moved && lists_stored); sent.rewrite = sent.rewrite || (sent_moved && lists_stored); @@ -3481,12 +3625,20 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) for (uint32_t i = 0; i < module->states && result == PURE_OK; ++i) if (rewrite[i] && (result = pure_save_image(app, module, i, lists[i])) == PURE_OK) listed[i] = (PureSaved){.image = app->seq}; - if (result == PURE_OK && heard.rewrite && (result = pure_save_runtime_image(app, PURE_LIST_HEARD)) == PURE_OK) + if (result == PURE_OK && heard.rewrite && + (result = pure_save_runtime_image(app, module, PURE_LIST_HEARD, NULL)) == PURE_OK) heard = (PureSaved){.image = app->seq}; - if (result == PURE_OK && sent.rewrite && (result = pure_save_runtime_image(app, PURE_LIST_OUTBOX)) == PURE_OK) + if (result == PURE_OK && sent.rewrite && + (result = pure_save_runtime_image(app, module, PURE_LIST_OUTBOX, NULL)) == PURE_OK) sent = (PureSaved){.image = app->seq}; + /* A replace that changes how sends are laid out names it in the outbox's image first. */ + bool outbox_imaged = false; + if (result == PURE_OK && old && lists_stored && (recast || outputs_changed) && + (result = pure_save_runtime_image(app, module, PURE_LIST_OUTBOX, recast)) == PURE_OK) + outbox_imaged = true; if (imaged && result != PURE_OK && old) app->uncertain = true; if (result != PURE_OK) { + pure_list_free(recast); for (uint32_t i = 0; i < module->states; ++i) { if (kept[i] >= 0) { if (lists[i] && lists[i] != app->lists[kept[i]]) pure_list_free(lists[i]); @@ -3538,7 +3690,18 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) if (module->state_type[i] == PURE_STATE_LIST) (void)pure_list_keyed(lists[i], module->state_key[i] ? (int)module->state_key[i] - 1 : -1); } - if (old) remap_outputs(app, old, module, true); + if (old) remap_outputs(app, old, module, true, NULL, NULL); + if (recast) { /* in the place of the outbox, its sends now the new module's */ + pure_list_free(app->outbox); + app->outbox = recast; + app->durable_outbox = 0; + if (lists_stored) pure_list_journal(recast, true); + } + if (outbox_imaged) { + app->image_seq[PURE_LIST_OUTBOX] = app->seq; + app->logged_since[PURE_LIST_OUTBOX] = 0; + app->logged[PURE_LIST_OUTBOX] = 0; + } drop_passes(app); /* kept by instruction of the module they were made for */ app->module = module; app->counted = 0; @@ -3554,7 +3717,12 @@ PURE_COLD PureResult pure_app_start(PureApp *app, const PureModule *module, Pure *app = (PureApp){.data = data, .fault_pc = UINT32_MAX}; PureResult result = open_module(app, module); if (result == PURE_OK) result = pure_app_save(app); - if (result != PURE_OK) pure_app_stop(app); + if (result != PURE_OK) { /* stopped, saying why when it can */ + char why[sizeof(app->why)]; + memcpy(why, app->why, sizeof(why)); + pure_app_stop(app); + memcpy(app->why, why, sizeof(why)); + } return result; } diff --git a/core/module.h b/core/module.h index 7034b0e..ad31821 100644 --- a/core/module.h +++ b/core/module.h @@ -315,6 +315,7 @@ typedef struct { PureFault fault; uint32_t fault_pc; size_t memory; /* the budget of what the states hold (0: none): pure_app_memory */ + char why[128]; /* why the last start or replace was refused, when it says (else empty; a failed start keeps it) */ uint32_t counted; /* the states it counts, a bit each: lists (not a message's rows) and texts */ uint64_t ticks, events; /* The time events see as now, in milliseconds since 1970 (UTC). The host diff --git a/core/test-module.c b/core/test-module.c index 70cc02d..b318aee 100644 --- a/core/test-module.c +++ b/core/test-module.c @@ -1,5 +1,6 @@ #define _POSIX_C_SOURCE 200809L #include "module.h" +#include "module-save.h" #include "module-store.h" #include "module-text.h" #include "store.h" @@ -4307,40 +4308,115 @@ static void memory_budget(void) { remove_store(path); } -/* An output whose fields are named otherwise (here two numbers swapped) is - * another output: its waiting sends' bytes would mean something else. A - * module that has it so is refused while sends wait, replacing or starting - * over the store; once they are trimmed it replaces (review 004, F3). */ +/* A send laid out for some fields, made anew for others by name: values + * the new types can hold go over (a u32 widened, a number narrowed when it + * is one; text to bytes, and bytes that are UTF-8 to text), the rest start + * at 0 or empty. There and back again, nothing changes. */ +static void send_recast(void) { + PureField ordered[3] = {{"item", PURE_FIELD_TEXT, 0, 0}, {"count", PURE_FIELD_U32, 0, 0}, + {"total", PURE_FIELD_NUMBER, 0, 0}}; + PureField kept[5] = {{"total", PURE_FIELD_NUMBER, 0, 0}, {"item", PURE_FIELD_TEXT, 0, 0}, + {"count", PURE_FIELD_NUMBER, 0, 0}, {"memo", PURE_FIELD_BYTES, 0, 0}, + {"extra", PURE_FIELD_U32, 0, 0}}; + CHECK(pure_fields_lay_out(ordered, 3) == 12 && ordered[0].offset == PURE_FIELD_REST); + CHECK(pure_fields_lay_out(kept, 5) == 24 && kept[1].offset == 8 && kept[3].offset == PURE_FIELD_REST); + unsigned char payload[15] = {2, 0, 0, 0}; /* count 2, total 7.5, then the item */ + double total = 7.5; + memcpy(payload + 4, &total, 8); + memcpy(payload + 12, "pen", 3); + unsigned char *made, *back; + size_t size, back_size; + CHECK(pure_send_recast(ordered, 3, kept, 5, (PureSpan){payload, 15}, &made, &size) == PURE_OK && size == 27); + double count, first; + memcpy(&first, made, 8); + memcpy(&count, made + 12, 8); + CHECK(first == 7.5 && made[8] == 3 && count == 2 && !memcmp(made + 20, "\0\0\0\0", 4) && !memcmp(made + 24, "pen", 3)); + CHECK(pure_send_recast(kept, 5, ordered, 3, (PureSpan){made, size}, &back, &back_size) == PURE_OK && + back_size == 15 && !memcmp(back, payload, 15)); + free(back); + /* A number that is not a u32, and bytes that are not UTF-8, start empty. */ + PureField narrow[2] = {{"count", PURE_FIELD_U32, 0, 0}, {"memo", PURE_FIELD_TEXT, 0, 0}}; + pure_fields_lay_out(narrow, 2); + count = 2.5; + memcpy(made + 12, &count, 8); + memcpy(made + 24, "p\xffn", 3); /* the memo (the rest) is those three bytes now, the item none */ + made[8] = 0; + CHECK(pure_send_recast(kept, 5, narrow, 2, (PureSpan){made, 24 + 3}, &back, &back_size) == PURE_OK && + back_size == 4 && !memcmp(back, "\0\0\0\0", 4)); + free(back); + PureField bytes_memo[1] = {{"memo", PURE_FIELD_BYTES, 0, 0}}; + pure_fields_lay_out(bytes_memo, 1); + CHECK(pure_send_recast(kept, 5, bytes_memo, 1, (PureSpan){made, 24 + 3}, &back, &back_size) == PURE_OK && + back_size == 3 && !memcmp(back, "p\xffn", 3)); + free(back); + /* A payload not laid out by its fields is the store's error. */ + CHECK(pure_send_recast(kept, 5, ordered, 3, (PureSpan){made, 10}, &back, &back_size) == PURE_STORE_ERROR && !back); + free(made); +} + +/* Two numbers of a waiting send, as laid out (pay's fields in the app's module). */ +static bool pay_reads(const PureApp *app, size_t k, double first, double second) { + Outbox o = outbox_of(app); + double a, b; + if (k >= o.count || o.items[k].size != 16) return false; + memcpy(&a, o.items[k].payload, 8); + memcpy(&b, o.items[k].payload + 8, 8); + return a == first && b == second; +} + +/* An output whose fields are named otherwise (here two numbers swapped) + * would mean something else of the bytes its sends wait with: replacing or + * starting the module recasts them by field name, as a list's items move. + * The outbox's image names the fields it holds its sends in, so a start + * after a replace whose save never came (the links still naming the old + * fields) reads them right. A module without the output is refused while + * sends wait on it, and says so. */ static void output_meaning(void) { static const S nums[] = {{"n", 0}}; static const F was_fields[] = {{"debit", PURE_FIELD_NUMBER}, {"credit", PURE_FIELD_NUMBER}}; static const F now_fields[] = {{"credit", PURE_FIELD_NUMBER}, {"debit", PURE_FIELD_NUMBER}}; static const I was_inputs[] = {KEY("tap", "KeyT"), {"pay", NULL, PURE_KIND_OUTPUT, false, was_fields, 2}}; static const I now_inputs[] = {KEY("tap", "KeyT"), {"pay", NULL, PURE_KIND_OUTPUT, false, now_fields, 2}}; + static const I none_inputs[] = {KEY("tap", "KeyT")}; static const O was_code[] = {{C, 0, 10}, {C, 0, 20}, {SEND, 1, 0}}; /* pay { debit: 10, credit: 20 } */ static const O now_code[] = {{C, 0, 20}, {C, 0, 10}, {SEND, 1, 0}}; /* the same, fields in the new order */ - static const H handlers[] = {{1, 0, 3}}; + static const O none_code[] = {{C, 0, 1}, {ST, 0, 0}}; + static const H handlers[] = {{1, 0, 3}}, none_handlers[] = {{1, 0, 2}}; Spec was = {nums, 1, was_inputs, 2, NULL, 0, handlers, 1, was_code, 3, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0}; Spec now = {nums, 1, now_inputs, 2, NULL, 0, handlers, 1, now_code, 3, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0}; - CHECK(load(was, &module_a) == PURE_OK && load(now, &module_b) == PURE_OK); + Spec none = {nums, 1, none_inputs, 1, NULL, 0, none_handlers, 1, none_code, 2, NULL, 0, NULL, 0, NULL, 0, NULL, 0, + NULL, 0, NULL, 0}; + PureModule without; + CHECK(load(was, &module_a) == PURE_OK && load(now, &module_b) == PURE_OK && load(none, &without) == PURE_OK); CHECK(pure_module_shape(&module_a, 1) != pure_module_shape(&module_b, 1)); char *path = new_store(); PureApp app; CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK); - CHECK(pure_app_input(&app, 0, PURE_EDGE_DOWN) == PURE_OK && outbox_of(&app).count == 1); - CHECK(pure_app_replace(&app, &module_b) == PURE_CONFLICT && app.module == &module_a); + CHECK(pure_app_input(&app, 0, PURE_EDGE_DOWN) == PURE_OK && pay_reads(&app, 0, 10, 20)); + CHECK(pure_app_replace(&app, &module_b) == PURE_OK && app.module == &module_b && pay_reads(&app, 0, 20, 10)); + pure_app_stop(&app); /* no save since: the links name a's fields, the outbox's image b's */ + CHECK(pure_app_start(&app, &module_b, pure_tree_data(path)) == PURE_OK && pay_reads(&app, 0, 20, 10)); pure_app_stop(&app); - CHECK(pure_app_start(&app, &module_b, pure_tree_data(path)) == PURE_CONFLICT); - CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && outbox_of(&app).count == 1); - pure_app_trim(&app, 1, 1); - CHECK(pure_app_save(&app) == PURE_OK && pure_app_replace(&app, &module_b) == PURE_OK); - CHECK(pure_app_input(&app, 0, PURE_EDGE_DOWN) == PURE_OK && outbox_of(&app).items[0].seq == 2); + CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && pay_reads(&app, 0, 10, 20)); + CHECK(pure_app_input(&app, 0, PURE_EDGE_DOWN) == PURE_OK && pay_reads(&app, 1, 10, 20)); + CHECK(outbox_of(&app).items[1].seq == 2); pure_app_stop(&app); - CHECK(pure_app_start(&app, &module_b, pure_tree_data(path)) == PURE_OK && outbox_of(&app).count == 1); + CHECK(pure_app_start(&app, &module_b, pure_tree_data(path)) == PURE_OK && pay_reads(&app, 0, 20, 10) && + pay_reads(&app, 1, 20, 10) && outbox_of(&app).items[1].seq == 2); + /* Sent with b's fields, logged after the image named them. */ + CHECK(pure_app_input(&app, 0, PURE_EDGE_DOWN) == PURE_OK && pay_reads(&app, 2, 20, 10)); pure_app_stop(&app); - CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_CONFLICT); + CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && pay_reads(&app, 2, 10, 20)); + pure_app_stop(&app); + CHECK(pure_app_start(&app, &without, pure_tree_data(path)) == PURE_CONFLICT && strstr(app.why, "'pay'")); + CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && !app.why[0]); + CHECK(pure_app_replace(&app, &without) == PURE_CONFLICT && app.module == &module_a && strstr(app.why, "'pay'")); + pure_app_trim(&app, 1, 3); + CHECK(pure_app_save(&app) == PURE_OK && pure_app_replace(&app, &without) == PURE_OK && !app.why[0]); + pure_app_stop(&app); + pure_module_release(&without); remove_store(path); } @@ -4457,6 +4533,7 @@ int main(void) { keyed_mutations(); snapshots(); links(); + send_recast(); output_meaning(); memory_budget(); fold_mutations(); diff --git a/serve/serve.c b/serve/serve.c index a9dd9d4..d06f430 100644 --- a/serve/serve.c +++ b/serve/serve.c @@ -2230,7 +2230,7 @@ static bool start_executor(Executor *ex, const PureModule *module, uint32_t show ex->edit = edit; if (pure_store_open(store, &ex->store) != PURE_OK) { fputs("Cannot open the store\n", stderr); return false; } if (pure_app_start(&ex->app, module, pure_store_data(ex->store)) != PURE_OK || !pure_app_defer(&ex->app, true)) { - fputs("Cannot start the module on the store\n", stderr); + fprintf(stderr, "Cannot start the module on the store%s%s\n", ex->app.why[0] ? ": " : "", ex->app.why); return false; } pure_app_background_checkpoints(&ex->app, true); diff --git a/tests/check.py b/tests/check.py index ea4f9f6..b377e20 100644 --- a/tests/check.py +++ b/tests/check.py @@ -1317,6 +1317,128 @@ def server_windows(aup, pure, work): return {'windows_full': stats['a']['windows_full'], 'wire_deliveries': stats['a']['delivered'], 'link_records': stats['a']['link_records']} + +UPGRADED_SENDER = """state n = 0 +input buy = message { + item: text + count: u32 +} +output ordered = message { + note: text + item: text + count: u32 +} +on buy { + n = n + 1 + send ordered { note: "second", item: buy.item, count: buy.count } +} +view sent = n +""" + +UPGRADED_RECEIVER = """state orders = 0 +state noted = 0 +state stock = 0 +input order = message { + note: text + item: text + count: u32 +} +on order { + orders = orders + 1 + stock = stock + order.count + if order.note == "second" { noted = noted + 1 } +} +view count = orders +view noted_count = noted +view total = stock +""" + + +def server_upgrade(aup, pure, work): + """A rolling upgrade of an output's fields between two linked nodes. The + sender, its receiver down, holds five sends of ordered {item, count, + to.city}; started again with ordered {note, item, count} it recasts them + by name (note empty, to.city gone) where it was refused before (review + 004's F3), and a module without the output is refused naming it. The old + receiver refuses the new shape, so the sends wait; the upgraded receiver + takes all eight, once each.""" + import select + import socket + import time + modules = {} + for name, text in (('sender-1', LINK_SENDER), ('sender-2', UPGRADED_SENDER), ('receiver-1', LINK_RECEIVER), + ('receiver-2', UPGRADED_RECEIVER), ('sender-0', 'state n = 0\ninput tap = key KeyT\non tap.down { n = 1 }\n')): + source = work / f'upgrade-{name}.aup' + source.write_text(text) + modules[name] = compile_module(aup, source, source.with_suffix('.pmod')) and source.with_suffix('.pmod') + + def free_port(kind=socket.SOCK_STREAM): + with socket.socket(socket.AF_INET, kind) as probe: + probe.bind(('127.0.0.1', 0)) + return probe.getsockname()[1] + + def start(args, port): + process = subprocess.Popen([str(a) for a in args] + ['--port', str(port)], stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + ready, _, _ = select.select([process.stdout], [], [], 10) + line = process.stdout.readline() if ready else b'' + assert line, f'host did not start: {process.stderr.read(2000) if process.poll() is not None else ""}' + return process, json.loads(line) + + def stop(process): + process.terminate() + _, error = process.communicate(timeout=30) + assert process.returncode == 0, error[-1000:] + return [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1] + + def number(port, view): + return int(http(port, 'GET', f'/views/{view}')[3]) + + keys = {} + for name in ('sender', 'receiver'): + process, info = start([pure / 'build/pure-serve', '--module', modules['receiver-1'], '--store', + work / 'upgrade-key', '--key', work / f'upgrade-{name}.key'], 0) + keys[name] = info['public_key'] + stop(process) + port_a, port_b, link_port = free_port(), free_port(), free_port(socket.SOCK_DGRAM) + + def sender_args(module): + return [pure / 'build/pure-serve', '--module', modules[module], '--store', work / 'upgrade-a', '--key', + work / 'upgrade-sender.key', '--name', 'up', '--send', f'ordered=pure://127.0.0.1:{link_port}/order', + '--seal', f'ordered={keys["receiver"]}'] + + def receiver_args(module): + return [pure / 'build/pure-serve', '--module', modules[module], '--store', work / 'upgrade-b', '--key', + work / 'upgrade-receiver.key', '--link-port', link_port, '--sender', f'up.ordered={keys["sender"]}'] + + running, stats, said = {}, {}, '' + try: + running['a'] = start(sender_args('sender-1'), port_a)[0] + for i in range(5): + assert post(port_a, '/inputs/buy', {'item': f'first-{i}', 'count': 1})[0] == 200 + stop(running.pop('a')) + refused = subprocess.run([str(a) for a in sender_args('sender-0')] + ['--port', str(port_a)], + capture_output=True, timeout=30) + assert refused.returncode != 0 and b"sends wait on the output 'ordered'" in refused.stderr, refused.stderr + running['a'] = start(sender_args('sender-2'), port_a)[0] + for i in range(3): + assert post(port_a, '/inputs/buy', {'item': f'second-{i}', 'count': 10})[0] == 200 + running['b'] = start(receiver_args('receiver-1'), port_b)[0] + time.sleep(1.5) + assert number(port_b, 'count') == 0 # the old input takes another shape: the sends wait + stop(running.pop('b')) + running['b'] = start(receiver_args('receiver-2'), port_b)[0] + deadline = time.time() + 20 + while number(port_b, 'count') < 8 and time.time() < deadline: + time.sleep(0.05) + got = (number(port_b, 'count'), number(port_b, 'noted_count'), number(port_b, 'total')) + assert got == (8, 3, 35), got + finally: + for name, process in running.items(): + stats[name] = stop(process) + assert stats['a']['link_refused'] >= 1, stats['a'] + return {'recast_waiting': 5, 'delivered': 8, 'refused_while_old': stats['a']['link_refused']} + def server_links_independent(aup, pure, work): """A receiver written from docs/design/012-link-protocol.md alone, in Python with the cryptography package and no code of link/: a Pure node @@ -3023,6 +3145,7 @@ def main(): report['server_links_refused'] = server_links_refused(aup, pure, work) report['server_memory'] = server_memory(aup, pure, work) report['server_windows'] = server_windows(aup, pure, work) + report['server_upgrade'] = server_upgrade(aup, pure, work) report['server_links_independent'] = server_links_independent(aup, pure, work) report['server_links_moved'] = server_links_moved(aup, pure, work) report['server_webhook'] = server_webhook(aup, pure, work) -- cgit v1.2.3