diff options
Diffstat (limited to 'prototypes/connections')
| -rw-r--r-- | prototypes/connections/module-save.c | 188 | ||||
| -rw-r--r-- | prototypes/connections/module-save.h | 3 | ||||
| -rw-r--r-- | prototypes/connections/module.c | 257 | ||||
| -rw-r--r-- | prototypes/connections/module.h | 42 | ||||
| -rw-r--r-- | prototypes/connections/serve.c | 3 | ||||
| -rw-r--r-- | prototypes/connections/test-module.c | 236 |
6 files changed, 715 insertions, 14 deletions
diff --git a/prototypes/connections/module-save.c b/prototypes/connections/module-save.c index a5eb95a..4149316 100644 --- a/prototypes/connections/module-save.c +++ b/prototypes/connections/module-save.c @@ -10,6 +10,12 @@ * field {u8 0 number | 1 text, u8 name length, name}, then its items * (module-list.h). A text state: its UTF-8 bytes. */ static const unsigned char values_magic[4] = {'P', 'V', 'S', '1'}, list_magic[4] = {'P', 'L', 'S', '1'}; +/* links: "PLK1"; u8 outputs, each {u8 name length, name, u8 field count, the + * fields' types, u64le last number} (the module's outputs, then those it + * dropped, without fields); u16le sources, each {u8 length, the name, u64le + * last number heard}; then the sent messages to the end, each {u8 output (in + * the table), u64le number, u32le size, payload}. */ +static const unsigned char links_magic[4] = {'P', 'L', 'K', '1'}; static const PureVersion unsaved = {UINT64_MAX, UINT64_MAX}; bool pure_save_enabled(const PureApp *app) { return app->data.state && app->data.save; } @@ -252,6 +258,154 @@ PURE_COLD PureResult pure_save_open_values(PureApp *app, const PureModule *m, do static bool same_version(PureVersion a, PureVersion b) { return a.ref == b.ref && a.bytes == b.bytes; } +static unsigned char *put_le(unsigned char *at, uint64_t value, unsigned width) { + for (unsigned i = 0; i < width; ++i) *at++ = (unsigned char)(value >> (8 * i)); + return at; +} + +static uint64_t get_le(const unsigned char *at, unsigned width) { + uint64_t value = 0; + for (unsigned i = 0; i < width; ++i) value |= (uint64_t)at[i] << (8 * i); + return value; +} + +/* The links as stored; NULL when memory runs out. */ +static unsigned char *encode_links(const PureApp *app, size_t *size) { + const PureModule *m = app->module; + unsigned outputs = 0; + uint8_t table[PURE_MODULE_INPUTS]; + size_t total = 4 + 1 + 2; + for (uint32_t i = 0; i < m->inputs; ++i) + if (m->input[i].kind == PURE_KIND_OUTPUT) { + table[i] = (uint8_t)outputs++; + total += 2 + strlen(m->input_names[i]) + m->input[i].field_count + 8; + } + for (unsigned c = 0; c < app->carried_count; ++c) total += 2 + strlen(app->carried[c].name) + 8; + for (unsigned h = 0; h < app->heard_count; ++h) total += 1 + strlen(app->heard[h].source) + 8; + for (size_t s = 0; s < app->sent_count; ++s) + if (!app->sent[s].trimmed) total += 1 + 8 + 4 + app->sent[s].size; + unsigned char *bytes = malloc(total), *at = bytes; + if (!bytes) return NULL; + memcpy(at, links_magic, 4); + at += 4; + *at++ = (unsigned char)(outputs + app->carried_count); + for (uint32_t i = 0; i < m->inputs; ++i) { + if (m->input[i].kind != PURE_KIND_OUTPUT) continue; + size_t length = strlen(m->input_names[i]); + *at++ = (unsigned char)length; + memcpy(at, m->input_names[i], length); + at += length; + *at++ = (unsigned char)m->input[i].field_count; + for (uint32_t f = 0; f < m->input[i].field_count; ++f) *at++ = m->field[m->input[i].field_start + f].type; + at = put_le(at, app->last_sent[i], 8); + } + for (unsigned c = 0; c < app->carried_count; ++c) { + size_t length = strlen(app->carried[c].name); + *at++ = (unsigned char)length; + memcpy(at, app->carried[c].name, length); + at += length; + *at++ = 0; + at = put_le(at, app->carried[c].last, 8); + } + at = put_le(at, app->heard_count, 2); + for (unsigned h = 0; h < app->heard_count; ++h) { + size_t length = strlen(app->heard[h].source); + *at++ = (unsigned char)length; + memcpy(at, app->heard[h].source, length); + at = put_le(at + length, app->heard[h].seq, 8); + } + for (size_t s = 0; s < app->sent_count; ++s) { + const struct PureSent *sent = &app->sent[s]; + if (sent->trimmed) continue; + *at++ = table[sent->output]; + at = put_le(at, sent->seq, 8); + at = put_le(at, sent->size, 4); + if (sent->size) memcpy(at, sent->payload, sent->size); + at += sent->size; + } + *size = (size_t)(at - bytes); + return bytes; +} + +/* The stored links of a starting app: outputs by name (an output whose + * fields changed or that the module lacks refuses the start while it has + * sends left: CONFLICT; one without sends keeps its number for later), + * sources and sent messages. */ +PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) { + PureResult result = app->stored_links ? PURE_OK : open_state(app, "", "links", &app->stored_links); + if (result != PURE_OK || !pure_draft_revision(app->stored_links)) return result; + unsigned char *bytes; + size_t size; + if ((result = read_whole(app->stored_links, &bytes, &size)) != PURE_OK) return result; + int output[256]; + bool fits[256]; + size_t at = 5; + unsigned outputs = size >= 5 && !memcmp(bytes, links_magic, 4) ? bytes[4] : 0; + result = size >= 5 && !memcmp(bytes, links_magic, 4) ? PURE_OK : PURE_STORE_ERROR; + for (unsigned t = 0; t < outputs && result == PURE_OK; ++t) { + char name[PURE_MODULE_NAME]; + size_t length = at < size ? bytes[at] : 0; + if (!length || length >= PURE_MODULE_NAME || size - at < 2 + length) { result = PURE_STORE_ERROR; break; } + memcpy(name, bytes + at + 1, length); + name[length] = 0; + unsigned fields = bytes[at + 1 + length]; + at += 2 + length; + if (size - at < fields + 8) { result = PURE_STORE_ERROR; break; } + int i = pure_module_input(m, name); + output[t] = i >= 0 && m->input[i].kind == PURE_KIND_OUTPUT ? i : -1; + fits[t] = output[t] >= 0 && m->input[i].field_count == fields; + for (unsigned f = 0; fits[t] && f < fields; ++f) fits[t] = m->field[m->input[i].field_start + f].type == bytes[at + f]; + uint64_t last = get_le(bytes + at + fields, 8); + at += fields + 8; + if (output[t] >= 0) app->last_sent[output[t]] = last; + else if (last) { + struct PureCarried *grown = realloc(app->carried, (app->carried_count + 1) * sizeof(*grown)); + if (!grown) { result = PURE_ALLOCATION; break; } + app->carried = grown; + memcpy(grown[app->carried_count].name, name, length + 1); + grown[app->carried_count++].last = last; + } + } + unsigned sources = result == PURE_OK && size - at >= 2 ? (unsigned)get_le(bytes + at, 2) : 0; + if (result == PURE_OK && size - at < 2) result = PURE_STORE_ERROR; + at += 2; + if (result == PURE_OK && sources && !(app->heard = calloc(sources, sizeof(*app->heard)))) result = PURE_ALLOCATION; + app->heard_capacity = sources; + for (unsigned h = 0; h < sources && result == PURE_OK; ++h) { + size_t length = at < size ? bytes[at] : 0; + if (!length || length > PURE_SOURCE || size - at < 1 + length + 8) { result = PURE_STORE_ERROR; break; } + struct PureHeard *heard = &app->heard[app->heard_count++]; + memcpy(heard->source, bytes + at + 1, length); + heard->seq = heard->saved = get_le(bytes + at + 1 + length, 8); + at += 1 + length + 8; + } + while (result == PURE_OK && at < size) { + if (size - at < 13 || bytes[at] >= outputs) { result = PURE_STORE_ERROR; break; } + unsigned t = bytes[at]; + uint64_t seq = get_le(bytes + at + 1, 8); + size_t length = (size_t)get_le(bytes + at + 9, 4); + at += 13; + if (size - at < length) { result = PURE_STORE_ERROR; break; } + if (!fits[t]) { result = PURE_CONFLICT; break; } + if (app->sent_count == app->sent_capacity) { + size_t capacity = app->sent_capacity ? 2 * app->sent_capacity : 16; + struct PureSent *grown = realloc(app->sent, capacity * sizeof(*grown)); + if (!grown) { result = PURE_ALLOCATION; break; } + app->sent = grown; + app->sent_capacity = capacity; + } + unsigned char *payload = malloc(length ? length : 1); + if (!payload) { result = PURE_ALLOCATION; break; } + if (length) memcpy(payload, bytes + at, length); + app->sent[app->sent_count++] = (struct PureSent){(uint32_t)output[t], seq, payload, length, false}; + at += length; + } + free(bytes); + app->sent_durable = app->sent_count; + memcpy(app->saved_sent, app->last_sent, sizeof(app->saved_sent)); + return result; +} + PureResult pure_save_commit(PureApp *app, const double *values, void *const *documents, size_t document_count, int64_t *revisions) { if (!pure_save_enabled(app)) { @@ -262,8 +416,8 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc return result; } const PureModule *m = app->module; - PureDraft *drafts[2 * PURE_MODULE_STATES + 1]; - uint32_t states[2 * PURE_MODULE_STATES + 1]; + PureDraft *drafts[2 * PURE_MODULE_STATES + 2]; + uint32_t states[2 * PURE_MODULE_STATES + 2]; size_t count = 0; for (uint32_t i = 0; i < m->states; ++i) { PureDraft *draft = m->state_type[i] == PURE_STATE_TEXT ? app->texts[i] : @@ -292,6 +446,32 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc drafts[count++] = app->stored_values; } } + /* The links, written whole when they changed. */ + PureVersion links_before = {0, 0}; + bool links_written = false; + if (app->links_changed) { + PureResult result = app->stored_links ? PURE_OK : open_state(app, "", "links", &app->stored_links); + size_t length; + unsigned char *encoded = result == PURE_OK ? encode_links(app, &length) : NULL; + if (result == PURE_OK && !encoded) result = PURE_ALLOCATION; + if (result == PURE_OK) { + links_before = pure_draft_version(app->stored_links); + pure_draft_hold(app->stored_links, true); + result = pure_draft_splice(app->stored_links, 0, pure_draft_size(app->stored_links), (PureSpan){encoded, length}); + if (result != PURE_OK) pure_draft_hold(app->stored_links, false); + } + free(encoded); + if (result != PURE_OK) { + if (values_changed) { + pure_draft_restore(app->stored_values, values_before); + pure_draft_hold(app->stored_values, false); + } + return result; + } + links_written = true; + states[count] = UINT32_MAX; + drafts[count++] = app->stored_links; + } if (!count && !document_count) return PURE_OK; PureResult result = app->data.save(app->data.context, documents, document_count, drafts, count, revisions); for (size_t k = 0; result == PURE_OK && k < count; ++k) @@ -300,5 +480,9 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc if (result != PURE_OK) pure_draft_restore(app->stored_values, values_before); pure_draft_hold(app->stored_values, false); } + if (links_written) { + if (result != PURE_OK) pure_draft_restore(app->stored_links, links_before); + pure_draft_hold(app->stored_links, false); + } return result; } diff --git a/prototypes/connections/module-save.h b/prototypes/connections/module-save.h index ded9ba9..bc7adcc 100644 --- a/prototypes/connections/module-save.h +++ b/prototypes/connections/module-save.h @@ -21,6 +21,9 @@ PureResult pure_save_open_values(PureApp *, const PureModule *m, double *values, PureDraft *const *texts, PureList *const *lists); /* Store a list whose record changed in its draft, header and items anew. */ PureResult pure_save_rewrite_list(const PureModule *m, uint32_t state, PureList *list, PureDraft *draft); +/* The stored links of a starting app (module.h, PureApp); CONFLICT when an + * output with sends left is missing or has other fields. */ +PureResult pure_save_open_links(PureApp *, const PureModule *m); /* 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, diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c index b1f58fe..ee72494 100644 --- a/prototypes/connections/module.c +++ b/prototypes/connections/module.c @@ -21,7 +21,7 @@ enum { READ = IN_VIEW | IN_HANDLER }; * collection, a constant, a unit, the column's row, a branch target, a stack * depth, a fold, a relation or a keyed list. */ enum { A_NONE, A_STATE, A_ITEM = A_STATE + PURE_STATE_INDEX + 1, A_FIELD, A_COLLECTION, A_CONSTANT, A_UNIT, - A_ROW, A_TARGET, A_DEPTH, A_FOLD, A_RELATION, A_KEYED }; + A_ROW, A_TARGET, A_DEPTH, A_FOLD, A_RELATION, A_KEYED, A_OUTPUT }; /* A fold's block is verified as a column that leaves one number (count) or * two (a value and whether to include it), a table's filter as a column that * leaves one: these stand for a view type. */ @@ -98,6 +98,7 @@ static const struct { unsigned char pops, in[3], out, where, arg, of; } signatur [PURE_OP_LIST_FIND] = {1, {T_KEY}, T_INDEX, READ, A_KEYED, O_ARG}, [PURE_OP_LIST_HAS] = {1, {T_KEY}, T_NUM, READ, A_KEYED, O_ANY}, [PURE_OP_NOW] = {0, {0}, T_NUM, IN_HANDLER, A_NONE, O_ANY}, + [PURE_OP_SEND] = {0, {0}, 0, IN_HANDLER, A_OUTPUT, O_ANY}, [PURE_OP_FLOOR] = NUMS(1), [PURE_OP_MOD] = NUMS(2), }; #undef NUMS @@ -109,6 +110,10 @@ static uint32_t read32(const unsigned char *p) { return (uint32_t)p[0] | (uint32_t)p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24; } +static void write32(unsigned char *p, uint32_t value) { + for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(value >> (8 * i)); +} + static double read_double(const unsigned char *p) { uint64_t bits = read32(p) | (uint64_t)read32(p + 4) << 32; double value; @@ -187,6 +192,7 @@ PURE_COLD static bool names_arg(const PureModule *m, unsigned kind, uint32_t arg case A_FOLD: return (arg & 0xFF) <= PURE_FOLD_FIRST && state_of(m, arg >> 8 & 0xFF, PURE_STATE_LIST) && arg >> 16; case A_RELATION: return arg <= PURE_NOT_EQUAL; case A_KEYED: return state_of(m, arg, PURE_STATE_LIST) && m->state_key[arg]; + case A_OUTPUT: return arg < m->inputs && m->input[arg].kind == PURE_KIND_OUTPUT; case A_NONE: case A_TARGET: case A_DEPTH: return true; default: return state_of(m, arg, (uint8_t)(kind - A_STATE)); } @@ -296,10 +302,17 @@ PURE_COLD static const char *verify_code(const PureModule *m, PureBlock block, i } } /* Operand types, bottom to top; record operations take their fields. */ - unsigned char need[PURE_RECORD_FIELDS + 2]; + unsigned char need[PURE_MODULE_FIELDS + 2]; unsigned pops = signature[op->op].pops; memcpy(need, signature[op->op].in, sizeof(signature[op->op].in)); - if (op->op == PURE_OP_LIST_APPEND || op->op == PURE_OP_LIST_INSERT) { + if (op->op == PURE_OP_SEND) { + const PureInputShape *output = &m->input[arg]; + for (unsigned f = 0; f < output->field_count; ++f) { + uint8_t type = m->field[output->field_start + f].type; + need[f] = type == PURE_FIELD_TEXT ? T_TEXT : type == PURE_FIELD_BYTES ? T_BYTES : T_NUM; + } + pops = output->field_count; + } else if (op->op == PURE_OP_LIST_APPEND || op->op == PURE_OP_LIST_INSERT) { const PureRecord *record = &m->record[m->state_ref[arg]]; for (unsigned f = 0; f < record->count; ++f) need[pops + f] = field_type(m, arg, f); pops += record->count; @@ -516,7 +529,8 @@ PURE_COLD PureResult pure_module_load(const void *bytes, size_t size, PureModule shape->period > PURE_TIMER_LONGEST) REFUSE("timer input", i); continue; } - if (shape->kind != PURE_KIND_MESSAGE || read32(at + 8) || shape->field_start != next_field || + if ((shape->kind != PURE_KIND_MESSAGE && shape->kind != PURE_KIND_OUTPUT) || read32(at + 8) || + (shape->kind == PURE_KIND_OUTPUT && shape->keyed) || shape->field_start != next_field || shape->field_count > out->fields - next_field) REFUSE("message shape", i); /* The prefix holds the fixed fields and the variable fields' lengths * in order, but the last variable field's: it takes the rest. */ @@ -609,7 +623,8 @@ PURE_COLD PureResult pure_module_load(const void *bytes, size_t size, PureModule uint32_t event = read32(at); out->handler[i] = (PureHandler){event, {read32(at + 4), read32(at + 8)}}; int input = event_input(event); - if (event > 2u * out->inputs || (input >= 0 && out->input[input].kind != PURE_KIND_KEY && !(event & 1))) + if (event > 2u * out->inputs || (input >= 0 && out->input[input].kind != PURE_KIND_KEY && !(event & 1)) || + (input >= 0 && out->input[input].kind == PURE_KIND_OUTPUT)) REFUSE("handler event", i); } for (uint32_t i = 0; i < out->code_count; ++i, at += OP_BYTES) { @@ -838,6 +853,49 @@ static PureResult keep_bytes(PureArena *arena, const void *bytes, size_t length, /* Run a verified block. Types and stack bounds were proven at load. A change * the data refuses (a size limit, a missing item) fails the event with its * fault; other failures pass through. */ +typedef struct { unsigned char *at; } Copy; + +static bool copy_out(void *context, PureSpan span) { + Copy *copy = context; + if (span.size) memcpy(copy->at, span.data, span.size); + copy->at += span.size; + return true; +} + +/* An output's message from its field values, laid out as a message's + * payload (size bytes), numbered and kept with the event's changes. */ +static PureResult send(PureApp *app, uint32_t output, const Slot *fields, const PureArena *arena, size_t size) { + const PureModule *m = app->module; + const PureInputShape *shape = &m->input[output]; + if (app->sent_count == app->sent_capacity) { + size_t capacity = app->sent_capacity ? 2 * app->sent_capacity : 16; + struct PureSent *grown = realloc(app->sent, capacity * sizeof(*grown)); + if (!grown) return PURE_ALLOCATION; + app->sent = grown; + app->sent_capacity = capacity; + } + unsigned char *payload = malloc(size ? size : 1); + if (!payload) return PURE_ALLOCATION; + Copy copy = {payload + shape->prefix}; + for (uint32_t f = 0; f < shape->field_count; ++f) { + const PureField *field = &m->field[shape->field_start + f]; + if (field->type == PURE_FIELD_U32) write32(payload + field->offset, (uint32_t)fields[f].num); + else if (field->type == PURE_FIELD_NUMBER) { + uint64_t bits; + memcpy(&bits, &fields[f].num, 8); + write32(payload + field->offset, (uint32_t)bits); + write32(payload + field->offset + 4, (uint32_t)(bits >> 32)); + } else { + if (field->offset != PURE_FIELD_REST) write32(payload + field->offset, (uint32_t)fields[f].bytes.length); + PureResult result = visit_value(app, arena, fields[f].bytes, copy_out, ©); + if (result != PURE_OK) { free(payload); return result; } + } + } + app->sent[app->sent_count++] = (struct PureSent){output, ++app->last_sent[output], payload, size, false}; + app->links_changed = true; + return PURE_OK; +} + static PureResult run(const PureModule *m, double *values, PureBlock block, Event *ev, Slot *result) { Slot stack[PURE_MODULE_STACK]; size_t sp = 0; @@ -995,6 +1053,23 @@ static PureResult run(const PureModule *m, double *values, PureBlock block, Even } case PURE_OP_BYTES_FROM: case PURE_OP_INDEX_NUM: continue; /* the same bytes or position, retyped */ case PURE_OP_LIST_COUNT: stack[sp++].num = (double)pure_list_count(app->lists[arg]); continue; + case PURE_OP_SEND: { + const PureInputShape *output = &m->input[arg]; + size_t base = sp - output->field_count, size = output->prefix; + sp = base; + for (uint32_t f = 0; f < output->field_count; ++f) { + uint8_t type = m->field[output->field_start + f].type; + double number = stack[base + f].num; + if (type == PURE_FIELD_BYTES || type == PURE_FIELD_TEXT) size += (size_t)stack[base + f].bytes.length; + else if (type == PURE_FIELD_U32 && !(number >= 0 && number <= UINT32_MAX && number == floor(number))) { + kind = PURE_FAULT_LIMIT; + goto failed; + } + } + if (size > UINT32_MAX) { kind = PURE_FAULT_LIMIT; goto failed; } + if ((status = send(app, arg, stack + base, arena, size)) != PURE_OK) return status; + continue; + } case PURE_OP_LIST_APPEND: case PURE_OP_LIST_INSERT: { PureList *list = app->lists[arg]; const PureRecord *record = &m->record[m->state_ref[arg]]; @@ -1253,6 +1328,37 @@ static void close_burst(PureApp *app, bool keep) { app->open = false; } +static void drop_sent(PureApp *app, size_t count) { + while (app->sent_count > count) free(app->sent[--app->sent_count].payload); +} + +/* After a save: what it stored is the links' saved state (trimmed sends go), + * or, when it failed, the links return to the last saved state. */ +static void settle_links(PureApp *app, bool saved) { + if (saved) { + size_t kept = 0; + for (size_t i = 0; i < app->sent_count; ++i) { + if (app->sent[i].trimmed) free(app->sent[i].payload); + else app->sent[kept++] = app->sent[i]; + } + app->sent_count = app->sent_durable = kept; + memcpy(app->saved_sent, app->last_sent, sizeof(app->saved_sent)); + for (unsigned i = 0; i < app->heard_count; ++i) app->heard[i].saved = app->heard[i].seq; + } else { + drop_sent(app, app->sent_durable); + for (size_t i = 0; i < app->sent_count; ++i) app->sent[i].trimmed = false; + memcpy(app->last_sent, app->saved_sent, sizeof(app->last_sent)); + unsigned kept = 0; + for (unsigned i = 0; i < app->heard_count; ++i) + if (app->heard[i].saved) { + app->heard[kept] = app->heard[i]; + app->heard[kept++].seq = app->heard[i].saved; + } + app->heard_count = kept; + } + app->links_changed = false; +} + PureResult pure_app_save(PureApp *app) { void *documents[PURE_APP_PENDING]; int64_t revisions[PURE_APP_PENDING]; @@ -1260,6 +1366,7 @@ PureResult pure_app_save(PureApp *app) { for (unsigned i = 0; i < app->pending_count; ++i) if (app->pending[i].edited) documents[count++] = app->pending[i].document; PureResult result = pure_save_commit(app, app->values, documents, count, revisions); + settle_links(app, result == PURE_OK); close_burst(app, result == PURE_OK); return result; } @@ -1315,6 +1422,11 @@ void pure_app_stop(PureApp *app) { } else pure_retired_free(&app->retired); pure_draft_close(app->stored_values); + drop_sent(app, 0); + free(app->sent); + free(app->heard); + free(app->carried); + pure_draft_close(app->stored_links); free(app->pending); pure_undo_free(&app->undo); for (unsigned i = 0; i < app->row_tables; ++i) free(app->rows[i].items); @@ -1325,7 +1437,12 @@ void pure_app_stop(PureApp *app) { } /* Where one event began inside the burst: a failed event returns there. */ -typedef struct { PureVersion texts[PURE_MODULE_STATES], lists[PURE_MODULE_STATES]; size_t undo; bool opened; } Mark; +typedef struct { + PureVersion texts[PURE_MODULE_STATES], lists[PURE_MODULE_STATES]; + size_t undo, sent; + uint64_t last_sent[PURE_MODULE_INPUTS]; + bool opened; +} Mark; static void mark_event(PureApp *app, Mark *mark) { mark->opened = !app->open; @@ -1335,6 +1452,8 @@ static void mark_event(PureApp *app, Mark *mark) { if (app->list_drafts[i]) mark->lists[i] = pure_draft_version(app->list_drafts[i]); } mark->undo = app->undo.count; + mark->sent = app->sent_count; + memcpy(mark->last_sent, app->last_sent, sizeof(mark->last_sent)); } static void undo_event(PureApp *app, const Mark *mark) { @@ -1343,9 +1462,30 @@ static void undo_event(PureApp *app, const Mark *mark) { if (app->list_drafts[i]) pure_draft_restore(app->list_drafts[i], mark->lists[i]); } pure_undo_rollback(&app->undo, mark->undo); + drop_sent(app, mark->sent); + memcpy(app->last_sent, mark->last_sent, sizeof(app->last_sent)); if (mark->opened) close_burst(app, true); /* nothing happened: release the holds */ } +/* The last number heard from source, or NULL. */ +static struct PureHeard *heard_from(PureApp *app, const char *source) { + for (unsigned i = 0; i < app->heard_count; ++i) + if (!strcmp(app->heard[i].source, source)) return &app->heard[i]; + return NULL; +} + +/* Hear seq from source (room for a new source was made before). */ +static void hear(PureApp *app, const char *source, uint64_t seq) { + struct PureHeard *heard = heard_from(app, source); + if (!heard) { + heard = &app->heard[app->heard_count++]; + *heard = (struct PureHeard){.saved = 0}; + memcpy(heard->source, source, strlen(source) + 1); + } + heard->seq = seq; + app->links_changed = true; +} + static PureResult keep_reply(PureApp *app); /* One event: every handler for it runs in module order against pending @@ -1384,6 +1524,10 @@ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) { memcpy(app->pending[app->pending_count++].key, ev->key, strlen(ev->key) + 1); app->revision += ev->modified; /* a save moves an edited document one revision */ } + if (app->delivering.pending) { + hear(app, app->delivering.source, app->delivering.seq); + app->delivering.pending = false; + } return event && !app->deferred ? pure_app_save(app) : PURE_OK; } @@ -1517,6 +1661,58 @@ PureResult pure_app_message_fields(PureApp *app, uint32_t input, const char *key return message(app, input, key, base, payload, placed); } +void pure_app_outbox(const PureApp *app, PureSentVisit visit, void *context) { + for (size_t i = 0; i < app->sent_durable; ++i) { + const struct PureSent *sent = &app->sent[i]; + if (!sent->trimmed && !visit(context, sent->output, sent->seq, (PureSpan){sent->payload, sent->size})) return; + } +} + +void pure_app_trim(PureApp *app, uint32_t output, uint64_t through) { + for (size_t i = 0; i < app->sent_durable; ++i) + if (app->sent[i].output == output && app->sent[i].seq <= through && !app->sent[i].trimmed) { + app->sent[i].trimmed = true; + app->links_changed = true; + } +} + +PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload) { + const PureModule *m = app->module; + size_t length = source ? strlen(source) : 0; + app->duplicate = false; + if (input >= m->inputs || m->input[input].kind != PURE_KIND_MESSAGE || m->input[input].keyed || !length || + length > PURE_SOURCE || !seq) return PURE_INVALID; + for (size_t i = 0; i < length; ++i) + if (source[i] <= ' ' || source[i] > '~') return PURE_INVALID; + struct PureHeard *heard = heard_from(app, source); + if (heard && seq <= heard->seq) { + pure_app_finish(app); + app->duplicate = true; + return PURE_OK; + } + if (!heard && app->heard_count == app->heard_capacity) { + unsigned capacity = app->heard_capacity ? 2 * app->heard_capacity : 4; + struct PureHeard *grown = realloc(app->heard, capacity * sizeof(*grown)); + if (!grown) return PURE_ALLOCATION; + app->heard = grown; + app->heard_capacity = capacity; + } + app->delivering.source = source; + app->delivering.seq = seq; + app->delivering.pending = true; + PureResult result = message(app, input, NULL, -1, payload, NULL); + if (!app->delivering.pending) return result; + /* Not applied. Refused by the app (a fault), the message was delivered + * all the same: its number is heard, saved like a key event's changes. */ + app->delivering.pending = false; + if (app->fault == PURE_FAULT_NONE) return result; + Mark mark; + mark_event(app, &mark); + hear(app, source, seq); + PureResult saved = app->deferred ? PURE_OK : pure_app_save(app); + return saved == PURE_OK ? result : saved; +} + PureResult pure_app_reply(const PureApp *app, bool (*visit)(void *, PureSpan), void *context) { if (!app->has_reply) return PURE_OK; return visit_value(app, &app->arena, app->reply, visit, context); @@ -1738,6 +1934,52 @@ 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 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) { + int map[PURE_MODULE_INPUTS]; + 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) + same = module->field[module->input[to].field_start + f].type == old->field[old->input[i].field_start + f].type; + map[i] = same ? to : -1; + for (size_t s = 0; !same && s < app->sent_count; ++s) + if (app->sent[s].output == i) return PURE_CONFLICT; + } + 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; + if (map[i] >= 0) { + last[map[i]] = app->last_sent[i]; + saved[map[i]] = app->saved_sent[i]; + continue; + } + struct PureCarried *grown = realloc(app->carried, (app->carried_count + 1) * sizeof(*grown)); + if (!grown) continue; /* its numbers would start again: kept only while memory allows */ + app->carried = grown; + memcpy(grown[app->carried_count].name, old->input_names[i], PURE_MODULE_NAME); + grown[app->carried_count++].last = app->last_sent[i]; + app->links_changed = true; + } + for (unsigned c = 0; c < app->carried_count;) { + int to = pure_module_input(module, app->carried[c].name); + if (to < 0 || module->input[to].kind != PURE_KIND_OUTPUT || last[to]) { ++c; continue; } + last[to] = saved[to] = app->carried[c].last; + app->carried[c] = app->carried[--app->carried_count]; + app->links_changed = true; + } + for (size_t s = 0; s < app->sent_count; ++s) app->sent[s].output = (uint32_t)map[app->sent[s].output]; + memcpy(app->last_sent, last, sizeof(last)); + memcpy(app->saved_sent, saved, sizeof(saved)); + return PURE_OK; +} + PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) { const PureModule *old = app->module; bool stored = pure_save_enabled(app); @@ -1797,6 +2039,8 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) values[i] = same ? app->values[kept[i]] : 0; } if (result == PURE_OK && stored) result = pure_save_open_values(app, module, values, kept, texts, lists); + /* Outputs keep their sends and numbers by name. */ + if (result == PURE_OK) result = old ? remap_outputs(app, old, module, false) : stored ? pure_save_open_links(app, module) : PURE_OK; for (uint32_t i = 0; i < module->states; ++i) { uint32_t list = module->state_ref[i]; if (module->state_type[i] == PURE_STATE_INDEX && moved[list] && values[i] < (double)pure_list_count(lists[list])) @@ -1846,6 +2090,7 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module) memcpy(app->lists, lists, sizeof(lists)); memcpy(app->list_drafts, list_drafts, sizeof(list_drafts)); memcpy(app->saved, saved, sizeof(saved)); + if (old) remap_outputs(app, old, module, true); app->module = module; ++app->changes; return PURE_OK; diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h index 430ccae..ed40c15 100644 --- a/prototypes/connections/module.h +++ b/prototypes/connections/module.h @@ -38,7 +38,7 @@ typedef enum { PURE_OP_ITEM_GET, PURE_OP_ITEM_SET, PURE_OP_INDEX_LOAD, PURE_OP_INDEX_STORE, PURE_OP_INDEX_MOVE, PURE_OP_INDEX_FIRST, PURE_OP_INDEX_LAST, PURE_OP_INDEX_NUM, PURE_OP_ROW, PURE_OP_PICK, PURE_OP_SLIDE, PURE_OP_FOLD, PURE_OP_TEXT_COMPARE, PURE_OP_LIST_FIND, PURE_OP_LIST_HAS, - PURE_OP_NOW, PURE_OP_FLOOR, PURE_OP_MOD, + PURE_OP_NOW, PURE_OP_FLOOR, PURE_OP_MOD, PURE_OP_SEND, PURE_OP_COUNT } PureOpcode; /* FOLD's argument: the kind (bits 0-7), the list state (8-15) and the length @@ -51,9 +51,12 @@ typedef enum { PURE_FOLD_COUNT, PURE_FOLD_SUM, PURE_FOLD_MAX, PURE_FOLD_MIN, PUR * comparisons (LT LE GT GE EQ NE); texts order by their bytes. */ typedef enum { PURE_LESS, PURE_LESS_EQUAL, PURE_GREATER, PURE_GREATER_EQUAL, PURE_EQUAL, PURE_NOT_EQUAL } PureRelation; -/* A timer ticks every `period` milliseconds while a host runs the app. */ -typedef enum { PURE_KIND_KEY = 1, PURE_KIND_MESSAGE = 2, PURE_KIND_TEXT = 3, PURE_KIND_TIMER = 4 } PureInputKind; -enum { PURE_TIMER_LONGEST = 86400000 }; +/* A timer ticks every `period` milliseconds while a host runs the app. An + * output is a message the app sends (SEND, in a handler; it has fields as a + * message has and no handler): what receives it is the host's wiring. */ +typedef enum { PURE_KIND_KEY = 1, PURE_KIND_MESSAGE = 2, PURE_KIND_TEXT = 3, PURE_KIND_TIMER = 4, + PURE_KIND_OUTPUT = 5 } PureInputKind; +enum { PURE_TIMER_LONGEST = 86400000, PURE_SOURCE = 64 }; /* A message field: a u32le or a binary64 (fixed, at its offset in the * payload's prefix), or bytes or text of any length. The prefix also holds * each variable field's u32le length but the last one's (its offset is @@ -228,6 +231,22 @@ typedef struct { unsigned char *copied; int64_t revision; PureArena arena, view_arena; + /* Links (runtime-owned, saved with the states): messages sent on outputs + * and not trimmed, in the order sent (the first sent_durable are saved; + * a trimmed one goes at the next save), each output's last number (and + * as saved), those of outputs a module dropped, and the last number + * delivered from each source (and as saved). */ + struct PureSent { uint32_t output; uint64_t seq; unsigned char *payload; size_t size; bool trimmed; } *sent; + size_t sent_count, sent_capacity, sent_durable; + uint64_t last_sent[PURE_MODULE_INPUTS], saved_sent[PURE_MODULE_INPUTS]; + struct PureCarried { char name[PURE_MODULE_NAME]; uint64_t last; } *carried; + unsigned carried_count; + struct PureHeard { char source[PURE_SOURCE + 1]; uint64_t seq, saved; } *heard; + unsigned heard_count, heard_capacity; + PureDraft *stored_links; + bool links_changed; /* since the last save */ + bool duplicate; /* the last delivery was heard before and not applied */ + struct { const char *source; uint64_t seq; bool pending; } delivering; struct PureSnapshot *snapshot; /* the newest snapshot made, or NULL */ PureRetired retired; /* list items let go since it was made */ bool reading; /* a reader: its states are a snapshot's */ @@ -285,6 +304,21 @@ PureResult pure_app_cell_text(PureApp *, uint32_t column, size_t row, bool (*vis * start at their initial value. The module must outlive the app. */ PureResult pure_app_replace(PureApp *, const PureModule *); +/* Connections between apps (docs/research/078-app-connections.md). A host + * sees only saved sends, delivers each at least once, in order per output, + * and trims what its receiver has saved; a receiver applies each once. */ +typedef bool (*PureSentVisit)(void *context, uint32_t output, uint64_t seq, PureSpan payload); +void pure_app_outbox(const PureApp *, PureSentVisit visit, void *context); +/* Forget output's saved sends numbered up to through (saved with the next save). */ +void pure_app_trim(PureApp *, uint32_t output, uint64_t through); +/* Deliver a message event from source (a sender's name, 1 to 64 printable + * ASCII bytes) numbered seq (from 1, growing per source). A number heard + * before is not applied again: it succeeds with app->duplicate set. The + * number counts as heard with the event's changes, and also when the app + * refuses the message (a fault: it was delivered, and refused); not when a + * save fails. */ +PureResult pure_app_deliver(PureApp *, uint32_t input, const char *source, uint64_t seq, PureSpan payload); + /* Readers on other threads. Right after a save an app can make a snapshot of * its saved states, for evaluating views anywhere while it keeps applying * events: numbers are copied; text states share their stored versions and diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c index 1eb5ca5..0029164 100644 --- a/prototypes/connections/serve.c +++ b/prototypes/connections/serve.c @@ -501,6 +501,9 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route * if (route->kind == ROUTE_NONE || (route->kind == ROUTE_OBJECT && !host->documents)) return "route"; if (route->kind == ROUTE_OBJECT) return NULL; int found = route->kind == ROUTE_VIEW ? pure_module_view(m, c->key) : pure_module_input(m, c->key); + /* An output is the app's to send and a timer ticks by itself: neither is posted. */ + if (found >= 0 && route->kind == ROUTE_INPUT && + (m->input[found].kind == PURE_KIND_OUTPUT || m->input[found].kind == PURE_KIND_TIMER)) found = -1; if (found < 0) return route->kind == ROUTE_VIEW ? "no_view" : "no_input"; *status = 400; route->target = (uint32_t)found; diff --git a/prototypes/connections/test-module.c b/prototypes/connections/test-module.c index ae82c69..bb59174 100644 --- a/prototypes/connections/test-module.c +++ b/prototypes/connections/test-module.c @@ -127,7 +127,7 @@ static size_t build(unsigned char *out, Spec spec) { *p++ = (unsigned char)(in->kind ? in->kind : 1); *p++ = in->keyed; *p++ = 0; *p++ = 0; bool timer = in->kind == PURE_KIND_TIMER; p = put32(p, timer ? (uint32_t)in->nfields : in->kind ? 0 : offset[n++]); - uint32_t first = in->kind == PURE_KIND_MESSAGE ? field_start : 0; /* only messages have fields */ + uint32_t first = in->kind == PURE_KIND_MESSAGE || in->kind == PURE_KIND_OUTPUT ? field_start : 0; /* only messages and outputs have fields */ *p++ = (unsigned char)first; *p++ = (unsigned char)(first >> 8); *p++ = (unsigned char)(timer ? 0 : in->nfields); *p++ = 0; for (size_t f = 0; !timer && f < in->nfields; ++f) field_offsets[nf++] = offset[n++]; @@ -2539,6 +2539,237 @@ static void message_fields(void) { CHECK(load(groups, &module_b) == PURE_INVALID && strstr(error, "field and group")); } +/* ---- Connections between apps (docs/research/078-app-connections.md): a + * send is part of its event and is seen once saved; a receiver applies each + * numbered message once, counts a refused one as heard, and forgets what a + * failed save undid. */ +enum { SEND = PURE_OP_SEND }; +typedef struct { uint32_t output; uint64_t seq; unsigned char payload[64]; size_t size; } Sent; +typedef struct { Sent items[16]; size_t count; } Outbox; + +static bool collect_sent(void *context, uint32_t output, uint64_t seq, PureSpan payload) { + Outbox *o = context; + CHECK(o->count < 16 && payload.size <= 64); + Sent *s = &o->items[o->count++]; + *s = (Sent){output, seq, {0}, payload.size}; + memcpy(s->payload, payload.data, payload.size); + return true; +} + +static Outbox outbox_of(const PureApp *app) { + Outbox o = {.count = 0}; + pure_app_outbox(app, collect_sent, &o); + return o; +} + +/* buy/fail {item: text, count: u32}: the count at 0, then the item. */ +static PureResult buy(PureApp *app, uint32_t input, const char *item, uint32_t count) { + unsigned char payload[64]; + size_t length = strlen(item); + memcpy(payload, &count, 4); + memcpy(payload + 4, item, length); + PureResult result = pure_app_message(app, input, NULL, -1, (PureSpan){payload, 4 + length}); + pure_app_finish(app); + return result; +} + +/* order {item: text, count: u32, total: number}: count, total, then item. */ +static PureResult deliver(PureApp *app, const char *source, uint64_t seq, const char *item, uint32_t count) { + unsigned char payload[64]; + size_t length = strlen(item); + double total = 0; + memcpy(payload, &count, 4); + memcpy(payload + 4, &total, 8); + memcpy(payload + 12, item, length); + PureResult result = pure_app_deliver(app, 0, source, seq, (PureSpan){payload, 12 + length}); + pure_app_finish(app); + return result; +} + +static void links(void) { + static const S sender_nums[] = {{"n", 0}}; + static const F order_fields[] = {{"item", PURE_FIELD_TEXT}, {"count", PURE_FIELD_U32}}; + static const F ordered_fields[] = {{"item", PURE_FIELD_TEXT}, {"count", PURE_FIELD_U32}, {"total", PURE_FIELD_NUMBER}}; + static const I sender_inputs[] = {{"buy", NULL, PURE_KIND_MESSAGE, false, order_fields, 2}, + {"fail", NULL, PURE_KIND_MESSAGE, false, order_fields, 2}, + {"ordered", NULL, PURE_KIND_OUTPUT, false, ordered_fields, 3}}; + static const O sender_code[] = {{L, 0, 0}, {C, 0, 1}, {ADD, 0, 0}, {ST, 0, 0}, {FIELD, 0, 0}, {FIELD, 1, 0}, + {L, 0, 0}, {SEND, 2, 0}, /* buy */ + {FIELD, 0, 0}, {FIELD, 1, 0}, {C, 0, 0}, {SEND, 2, 0}, {C, 0, 0}, {REQ, 0, 0}}; /* fail */ + static const H sender_handlers[] = {{1, 0, 8}, {3, 8, 6}}; + Spec sender = {sender_nums, 1, sender_inputs, 3, NULL, 0, sender_handlers, 2, sender_code, LEN(sender_code), + NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0}; + static const S receiver_nums[] = {{"stock", 0}}; + static const T receiver_texts[] = {{"last", 1, 0}}; + static const char *const receiver_constants[] = {""}; + static const I receiver_inputs[] = {{"order", NULL, PURE_KIND_MESSAGE, false, ordered_fields, 3}}; + static const O receiver_code[] = {{FIELD, 1, 0}, {L, 0, 0}, {ADD, 0, 0}, {ST, 0, 0}, {FIELD, 0, 0}, {TSET, 1, 0}, + {FIELD, 1, 0}, {C, 0, 100}, {LT, 0, 0}, {REQ, 0, 0}}; /* refuses 100 or more */ + static const H receiver_handlers[] = {{1, 0, 10}}; + Spec receiver = {receiver_nums, 1, receiver_inputs, 1, NULL, 0, receiver_handlers, 1, receiver_code, + LEN(receiver_code), NULL, 0, receiver_texts, 1, receiver_constants, 1, NULL, 0, NULL, 0, NULL, 0}; + if (load(sender, &module_a) != PURE_OK || load(receiver, &module_b) != PURE_OK) { fprintf(stderr, "%s\n", error); abort(); } + CHECK(module_a.input[2].prefix == 12); + + /* In memory: a send is its event's; the receiver applies it once per source. */ + PureApp a, b; + CHECK(pure_app_start(&a, &module_a, (PureData){0}) == PURE_OK && pure_app_start(&b, &module_b, (PureData){0}) == PURE_OK); + CHECK(buy(&a, 0, "pen", 2) == PURE_OK && buy(&a, 1, "ink", 3) == PURE_LIMIT); + Outbox o = outbox_of(&a); + CHECK(o.count == 1 && o.items[0].output == 2 && o.items[0].seq == 1 && o.items[0].size == 15 && a.values[0] == 1); + PureSpan sent = {o.items[0].payload, o.items[0].size}; + CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent) == PURE_OK && !b.duplicate && b.values[0] == 2 && + !strcmp(state_text(&b, 1), "pen")); + pure_app_finish(&b); + CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent) == PURE_OK && b.duplicate && b.values[0] == 2); + CHECK(deliver(&b, "other", 1, "cup", 5) == PURE_OK && !b.duplicate && b.values[0] == 7); + CHECK(deliver(&b, "orders.ordered", 3, "hat", 1) == PURE_OK && b.values[0] == 8); /* numbers may skip */ + CHECK(deliver(&b, "orders.ordered", 2, "old", 1) == PURE_OK && b.duplicate && b.values[0] == 8); + /* Refused (count 100): delivered all the same, never applied again. */ + CHECK(deliver(&b, "other", 2, "big", 100) == PURE_LIMIT && b.values[0] == 7 + 1); + CHECK(deliver(&b, "other", 2, "big", 1) == PURE_OK && b.duplicate && b.values[0] == 8); + /* Bad deliveries change nothing. */ + CHECK(deliver(&b, "", 4, "x", 1) == PURE_INVALID && deliver(&b, "a b", 4, "x", 1) == PURE_INVALID); + CHECK(deliver(&b, "other", 0, "x", 1) == PURE_INVALID && pure_app_deliver(&a, 2, "s", 1, sent) == PURE_INVALID); + CHECK(pure_app_message(&a, 2, NULL, -1, sent) == PURE_INVALID); + /* Trimmed, a send is gone at once for the host and from the store at the next save. */ + pure_app_trim(&a, 2, 1); + CHECK(outbox_of(&a).count == 0 && pure_app_save(&a) == PURE_OK && a.sent_count == 0); + CHECK(buy(&a, 0, "cap", 1) == PURE_OK && outbox_of(&a).items[0].seq == 2); + pure_app_stop(&a); + pure_app_stop(&b); + + /* Stored: sends and heard numbers survive restarts; a deferred burst's sends + * appear with its save, and a failed save takes them and their numbers back. */ + char path_a[64], path_b[64]; + snprintf(path_a, sizeof(path_a), "%s", new_store()); + snprintf(path_b, sizeof(path_b), "%s", new_store()); + CHECK(pure_app_start(&a, &module_a, pure_tree_data(path_a)) == PURE_OK); + CHECK(pure_app_start(&b, &module_b, pure_tree_data(path_b)) == PURE_OK); + CHECK(buy(&a, 0, "pen", 2) == PURE_OK && buy(&a, 0, "ink", 3) == PURE_OK); + pure_app_stop(&a); + CHECK(pure_app_start(&a, &module_a, pure_tree_data(path_a)) == PURE_OK); + o = outbox_of(&a); + CHECK(o.count == 2 && o.items[0].seq == 1 && o.items[1].seq == 2 && a.values[0] == 2); + CHECK(pure_app_defer(&a, true)); + for (int i = 0; i < 3; ++i) CHECK(buy(&a, 0, "box", 1) == PURE_OK); + CHECK(outbox_of(&a).count == 2 && pure_app_save(&a) == PURE_OK); + o = outbox_of(&a); + CHECK(o.count == 5 && o.items[4].seq == 5); + CHECK(buy(&a, 0, "lost", 1) == PURE_OK); + pure_app_trim(&a, 2, 3); + fail_stage = PURE_AFTER_HEADER; + CHECK(pure_app_save(&a) == PURE_STORE_ERROR); + fail_stage = -1; + o = outbox_of(&a); + CHECK(o.count == 5 && o.items[0].seq == 1 && a.values[0] == 5); /* the trim and the send undone */ + CHECK(buy(&a, 0, "next", 1) == PURE_OK && pure_app_save(&a) == PURE_OK); + o = outbox_of(&a); + CHECK(o.count == 6 && o.items[5].seq == 6); /* numbered on from 5 */ + pure_app_trim(&a, 2, 4); + CHECK(pure_app_save(&a) == PURE_OK); + pure_app_stop(&a); + CHECK(pure_app_start(&a, &module_a, pure_tree_data(path_a)) == PURE_OK); + o = outbox_of(&a); + CHECK(o.count == 2 && o.items[0].seq == 5 && o.items[1].seq == 6); + /* The receiver: heard numbers stay heard after a restart; a refused one too. */ + CHECK(deliver(&b, "orders.ordered", 1, "pen", 2) == PURE_OK && deliver(&b, "x", 1, "big", 100) == PURE_LIMIT); + pure_app_stop(&b); + CHECK(pure_app_start(&b, &module_b, pure_tree_data(path_b)) == PURE_OK && b.values[0] == 2); + CHECK(deliver(&b, "orders.ordered", 1, "pen", 2) == PURE_OK && b.duplicate && b.values[0] == 2); + CHECK(deliver(&b, "x", 1, "big", 1) == PURE_OK && b.duplicate); + /* A save that fails forgets what it heard. */ + CHECK(pure_app_defer(&b, true) && deliver(&b, "orders.ordered", 2, "ink", 3) == PURE_OK && b.values[0] == 5); + fail_stage = PURE_AFTER_HEADER; + CHECK(pure_app_save(&b) == PURE_STORE_ERROR && b.values[0] == 2); + fail_stage = -1; + CHECK(deliver(&b, "orders.ordered", 2, "ink", 3) == PURE_OK && !b.duplicate && b.values[0] == 5); + CHECK(pure_app_save(&b) == PURE_OK); + pure_app_stop(&b); + + /* Replacing: a module without the output is refused while sends are left + * on it; once they are trimmed it replaces, and the output's number is + * kept for a module that has it again. */ + static const I plain_inputs[] = {{"buy", NULL, PURE_KIND_MESSAGE, false, order_fields, 2}}; + static const O plain_code[] = {{L, 0, 0}, {C, 0, 1}, {ADD, 0, 0}, {ST, 0, 0}}; + static const H plain_handlers[] = {{1, 0, 4}}; + Spec plain = {sender_nums, 1, plain_inputs, 1, NULL, 0, plain_handlers, 1, plain_code, LEN(plain_code), + NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0}; + CHECK(load(plain, &module_b) == PURE_OK); + CHECK(pure_app_replace(&a, &module_b) == PURE_CONFLICT && a.module == &module_a); + pure_app_trim(&a, 2, 6); + CHECK(pure_app_save(&a) == PURE_OK && pure_app_replace(&a, &module_b) == PURE_OK); + CHECK(buy(&a, 0, "plain", 1) == PURE_OK && pure_app_save(&a) == PURE_OK); + pure_app_stop(&a); + CHECK(pure_app_start(&a, &module_b, pure_tree_data(path_a)) == PURE_OK); /* the number kept in the store */ + CHECK(pure_app_replace(&a, &module_a) == PURE_OK && buy(&a, 0, "back", 1) == PURE_OK && pure_app_save(&a) == PURE_OK); + o = outbox_of(&a); + CHECK(o.count == 1 && o.items[0].seq == 7); + /* A stored send whose output the module lacks refuses the start. */ + pure_app_stop(&a); + CHECK(pure_app_start(&a, &module_b, pure_tree_data(path_a)) == PURE_CONFLICT); + remove_store(path_a); + remove_store(path_b); + + /* A process that dies at each commit stage leaves an event and its send + * together or neither, and a delivery and its effect together or neither. */ + snprintf(path_a, sizeof(path_a), "%s", new_store()); + snprintf(path_b, sizeof(path_b), "%s", new_store()); + CHECK(load(receiver, &module_b) == PURE_OK); + uint64_t heard = 0; + size_t kept[2] = {0, 0}; /* sends and deliveries a death left in place */ + for (int round = 0; round < 2; ++round) + for (int stage = PURE_AFTER_HEADER; stage <= PURE_BEFORE_DIR_SYNC; ++stage) { + pid_t child = fork(); + CHECK(child >= 0); + if (!child) { + PureApp doomed, taker; + if (pure_app_start(&doomed, &module_a, pure_tree_data(path_a)) != PURE_OK || + pure_app_start(&taker, &module_b, pure_tree_data(path_b)) != PURE_OK) _exit(1); + crash = 1; + fail_stage = stage; + if (round) buy(&doomed, 0, "crash", 1); + else deliver(&taker, "orders.ordered", heard + 1, "crash", 1); + _exit(0); + } + int status; + CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 77); + CHECK(pure_app_start(&a, &module_a, pure_tree_data(path_a)) == PURE_OK); + CHECK(pure_app_start(&b, &module_b, pure_tree_data(path_b)) == PURE_OK); + o = outbox_of(&a); + CHECK(o.count == (size_t)a.values[0] && (!o.count || o.items[o.count - 1].seq == o.count)); + heard = b.heard_count ? b.heard[0].seq : 0; + CHECK(b.values[0] == (double)heard); + kept[0] = (size_t)heard; + kept[1] = o.count; + pure_app_stop(&a); + pure_app_stop(&b); + } + CHECK(kept[0] == 1 && kept[1] == 1); /* only a death after the head was written keeps it */ + remove_store(path_a); + remove_store(path_b); + + /* What an output and SEND may not be. */ + const struct { const O *code; size_t count; const char *reason; } refused[] = { + {(const O[]){{FIELD, 1, 0}, {FIELD, 0, 0}, {L, 0, 0}, {SEND, 2, 0}}, 4, "operand type"}, + {(const O[]){{FIELD, 0, 0}, {FIELD, 1, 0}, {L, 0, 0}, {SEND, 1, 0}}, 4, "no such state"}, + {(const O[]){{FIELD, 0, 0}, {FIELD, 1, 0}, {SEND, 2, 0}}, 3, "stack underflow"}, + }; + for (size_t i = 0; i < LEN(refused); ++i) { + Spec bad = sender; + H handler = {1, 0, (uint32_t)refused[i].count}; + bad.o = refused[i].code; bad.no = refused[i].count; bad.h = &handler; bad.nh = 1; + if (load(bad, &module_b) != PURE_INVALID || !strstr(error, refused[i].reason)) { + fprintf(stderr, "links case %zu: expected '%s', got '%s'\n", i, refused[i].reason, error); + abort(); + } + } + H on_output[] = {{1, 0, 8}, {5, 8, 6}}; + Spec bad = sender; + bad.h = on_output; + CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "handler event")); +} + /* A timer's tick sees the time the host set; floor and a floored modulo * take it apart. now is for handlers only. */ enum { NOW = PURE_OP_NOW, FLOOR = PURE_OP_FLOOR, MOD = PURE_OP_MOD }; @@ -2608,6 +2839,7 @@ int main(void) { keyed(); keyed_mutations(); snapshots(); + links(); fold_mutations(); pointing(); list_mutations(); @@ -2616,6 +2848,6 @@ int main(void) { stored_states(); deferred(); printf("{\"sizeof_module\":%zu,\"sizeof_app\":%zu}\n", sizeof(PureModule), sizeof(PureApp)); - puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration-deferred-bursts-pointing-marks-folds-filters-keyed-lists-text-order-message-fields-timers-now-snapshots\"}"); + puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration-deferred-bursts-pointing-marks-folds-filters-keyed-lists-text-order-message-fields-timers-now-snapshots-links\"}"); return 0; } |