diff options
| -rw-r--r-- | core/module-save.c | 191 | ||||
| -rw-r--r-- | core/module-save.h | 28 | ||||
| -rw-r--r-- | core/module.c | 230 | ||||
| -rw-r--r-- | core/module.h | 1 | ||||
| -rw-r--r-- | core/test-module.c | 107 | ||||
| -rw-r--r-- | serve/serve.c | 2 | ||||
| -rw-r--r-- | 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) |