aboutsummaryrefslogtreecommitdiffstats
path: root/prototypes
diff options
context:
space:
mode:
Diffstat (limited to 'prototypes')
-rw-r--r--prototypes/connections/module.c46
-rw-r--r--prototypes/connections/module.h9
-rw-r--r--prototypes/connections/serve.c286
-rw-r--r--prototypes/connections/test-module.c75
4 files changed, 368 insertions, 48 deletions
diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c
index 3490ba9..7f22c49 100644
--- a/prototypes/connections/module.c
+++ b/prototypes/connections/module.c
@@ -315,7 +315,10 @@ PURE_COLD static const char *verify_block(const PureModule *m, PureBlock block,
of = O_ARG;
owner = arg >> 8 & 0xFF;
}
- if (out == T_FIELD) out = m->field[shape->field_start + arg].type == PURE_FIELD_U32 ? T_NUM : T_BYTES;
+ if (out == T_FIELD) {
+ uint8_t type = m->field[shape->field_start + arg].type;
+ out = type == PURE_FIELD_TEXT ? T_TEXT : type == PURE_FIELD_BYTES ? T_BYTES : T_NUM;
+ }
if (out == T_ITEM) out = field_type(m, list_of(arg), item_field(arg));
if (out) {
if (height == PURE_MODULE_STACK) REFUSE("stack overflow");
@@ -445,19 +448,24 @@ PURE_COLD PureResult pure_module_load(const void *bytes, size_t size, PureModule
}
if (shape->kind != PURE_KIND_MESSAGE || read32(at + 8) || shape->field_start != next_field ||
shape->field_count > out->fields - next_field) REFUSE("message shape", i);
- /* Fixed u32le fields in order, then at most one bytes tail. */
+ /* 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);
+ if (type == PURE_FIELD_BYTES || type == PURE_FIELD_TEXT) 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];
field->type = (uint8_t)read32(entry + 4);
- if (!name_at(text, strings, read32(entry), true, field->name) ||
- (field->type != PURE_FIELD_U32 && field->type != PURE_FIELD_BYTES) || read32(entry + 4) > 2 ||
- (field->type == PURE_FIELD_BYTES && f + 1 != shape->field_count))
+ if (!name_at(text, strings, read32(entry), true, field->name) || field->type < PURE_FIELD_U32 ||
+ read32(entry + 4) > PURE_FIELD_TEXT)
REFUSE("message field", next_field + f);
for (uint32_t k = 0; k < f; ++k)
if (!strcmp(out->field[next_field + k].name, field->name)) REFUSE("duplicate field", next_field + f);
- field->offset = shape->prefix;
- if (field->type == PURE_FIELD_U32) shape->prefix += 4;
+ field->offset = f == last ? PURE_FIELD_REST : shape->prefix;
+ if (f != last) shape->prefix += field->type == PURE_FIELD_NUMBER ? 8 : 4;
}
next_field += shape->field_count;
}
@@ -610,6 +618,7 @@ typedef struct {
double dt;
size_t row; /* the row a column is evaluated for */
bool modified, exists;
+ uint32_t (*spans)[2]; /* a message's variable fields: start and size in the payload */
} Event;
static PureResult fault(PureApp *app, PureFault kind, uint32_t pc) {
@@ -776,7 +785,8 @@ static PureResult run(const PureModule *m, double *values, PureBlock block, Even
case PURE_OP_FIELD: {
const PureField *field = &m->field[ev->shape->field_start + arg];
if (field->type == PURE_FIELD_U32) stack[sp++].num = read32(app->payload.data + field->offset);
- else stack[sp++].bytes = (PureBytes){FROM_PAYLOAD, 0, {0, 0}, field->offset, app->payload.size - field->offset};
+ else if (field->type == PURE_FIELD_NUMBER) stack[sp++].num = read_double(app->payload.data + field->offset);
+ else stack[sp++].bytes = (PureBytes){FROM_PAYLOAD, 0, {0, 0}, ev->spans[arg][0], ev->spans[arg][1]};
continue;
}
case PURE_OP_DOC: case PURE_OP_DOC_SIZE: {
@@ -1313,6 +1323,23 @@ PureResult pure_app_message(PureApp *app, uint32_t input, const char *key, int64
return PURE_INVALID;
const PureInputShape *shape = &m->input[input];
if (payload.size < shape->prefix) return fault(app, PURE_FAULT_LIMIT, UINT32_MAX);
+ /* Where each variable field is; a length past the payload, a text that
+ * is not UTF-8 or a number that is not finite fails the message. */
+ uint32_t spans[PURE_MODULE_FIELDS][2];
+ size_t at = 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_NUMBER && !isfinite(read_double(payload.data + field->offset)))
+ return fault(app, PURE_FAULT_NON_FINITE, UINT32_MAX);
+ if (field->type != PURE_FIELD_BYTES && field->type != PURE_FIELD_TEXT) continue;
+ size_t size = field->offset == PURE_FIELD_REST ? payload.size - at : read32(payload.data + field->offset);
+ if (size > payload.size - at) return fault(app, PURE_FAULT_LIMIT, UINT32_MAX);
+ if (field->type == PURE_FIELD_TEXT && !pure_utf8_valid((PureSpan){size ? payload.data + at : NULL, size}))
+ return fault(app, PURE_FAULT_ENCODING, UINT32_MAX);
+ spans[f][0] = (uint32_t)at;
+ spans[f][1] = (uint32_t)size;
+ at += size;
+ }
if (shape->keyed && m->collections) {
if (!app->data.open) return PURE_INVALID;
/* A key joins a burst once, so its revision moves once per save. */
@@ -1334,7 +1361,8 @@ PureResult pure_app_message(PureApp *app, uint32_t input, const char *key, int64
if (base >= 0 && base != app->revision) { pure_app_finish(app); return PURE_CONFLICT; }
}
app->payload = payload;
- Event ev = {.app = app, .shape = shape, .key = key, .base = base, .arena = &app->arena, .exists = app->revision > 0};
+ Event ev = {.app = app, .shape = shape, .key = key, .base = base, .arena = &app->arena, .exists = app->revision > 0,
+ .spans = spans};
PureResult result = dispatch(app, 1 + 2 * input, &ev);
if (result != PURE_OK) pure_app_finish(app);
return result;
diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h
index 09a8540..52671f0 100644
--- a/prototypes/connections/module.h
+++ b/prototypes/connections/module.h
@@ -51,7 +51,14 @@ typedef enum { PURE_FOLD_COUNT, PURE_FOLD_SUM, PURE_FOLD_MAX, PURE_FOLD_MIN, PUR
typedef enum { PURE_LESS, PURE_LESS_EQUAL, PURE_GREATER, PURE_GREATER_EQUAL, PURE_EQUAL, PURE_NOT_EQUAL } PureRelation;
typedef enum { PURE_KIND_KEY = 1, PURE_KIND_MESSAGE = 2, PURE_KIND_TEXT = 3 } PureInputKind;
-typedef enum { PURE_FIELD_U32 = 1, PURE_FIELD_BYTES = 2 } PureFieldType;
+/* 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
+ * PURE_FIELD_REST), which takes what follows the others; the variable
+ * fields' bytes follow the prefix in declaration order. A text is UTF-8 and
+ * a number finite, or the message fails. */
+typedef enum { PURE_FIELD_U32 = 1, PURE_FIELD_BYTES = 2, PURE_FIELD_NUMBER = 3, PURE_FIELD_TEXT = 4 } PureFieldType;
+enum { PURE_FIELD_REST = UINT32_MAX };
typedef enum {
PURE_STATE_NUM = 0, PURE_STATE_TEXT = 1, PURE_STATE_CURSOR = 2, PURE_STATE_LIST = 3, PURE_STATE_INDEX = 4
} PureStateType;
diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c
index 2a1587d..2cb69c1 100644
--- a/prototypes/connections/serve.c
+++ b/prototypes/connections/serve.c
@@ -38,6 +38,7 @@
#include <fcntl.h>
#include <inttypes.h>
#include <limits.h>
+#include <math.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <pthread.h>
@@ -77,8 +78,14 @@ typedef struct Connection {
uint64_t deadline;
char key[65]; /* the document's key, or the view's or input's name */
RouteKind route;
- unsigned char fields[4 * PURE_MODULE_FIELDS]; /* a message's u32 fields, from the query */
+ /* An app input's message: its fixed fields from the query, which fields
+ * were given there, and whether the body is a JSON object of the rest.
+ * Its ingress is [u32le payload size][payload][the JSON as it came]. */
+ unsigned char fields[8 * PURE_MODULE_FIELDS];
size_t field_bytes;
+ bool given[PURE_MODULE_FIELDS], json;
+ uint32_t input;
+ unsigned char *ingress;
int64_t expected;
size_t body_size, received, in_used;
unsigned char *body;
@@ -364,6 +371,7 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
(c->route == ROUTE_INPUT && !c->post)) return false;
bool length_seen = false, condition_seen = false, host_seen = false;
c->body_size = 0;
+ c->json = false;
c->expected = -1;
c->keep_alive = true;
line = next + 2;
@@ -398,6 +406,8 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
if (condition_seen || strcmp(value, "*")) return false;
c->expected = 0;
condition_seen = true;
+ } else if (!strcasecmp(line, "Content-Type")) {
+ c->json = !strncasecmp(value, "application/json", 16);
} else if (!strcasecmp(line, "Connection")) {
if (!strcasecmp(value, "close")) c->keep_alive = false;
else if (strcasecmp(value, "keep-alive")) return false;
@@ -409,13 +419,59 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
return host_seen && !condition_seen && c->body_size == 0;
}
-/* What an app route names in the module, with a message's u32 fields taken
- * from the query into c->fields. NULL, or the error to answer (404 for a
- * name the module lacks, 400 otherwise). */
+static void put_u32(unsigned char *p, uint32_t v) { for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> (8 * i)); }
+static uint32_t get_u32(const unsigned char *p) { return p[0] | p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24; }
+
+/* A JSON number's text: -?(0|[1-9][0-9]*)(.[0-9]+)?([eE][+-]?[0-9]+)?, its
+ * length (0 if there is none). An integer has neither part after it. */
+static size_t json_number(const unsigned char *p, size_t size, bool *integer) {
+ size_t i = 0;
+ *integer = true;
+ if (i < size && p[i] == '-') ++i;
+ if (i == size || p[i] < '0' || p[i] > '9') return 0;
+ if (p[i++] != '0') while (i < size && p[i] >= '0' && p[i] <= '9') ++i;
+ if (i < size && p[i] == '.') {
+ *integer = false;
+ if (++i == size || p[i] < '0' || p[i] > '9') return 0;
+ while (i < size && p[i] >= '0' && p[i] <= '9') ++i;
+ }
+ if (i < size && (p[i] == 'e' || p[i] == 'E')) {
+ *integer = false;
+ if (++i < size && (p[i] == '+' || p[i] == '-')) ++i;
+ if (i == size || p[i] < '0' || p[i] > '9') return 0;
+ while (i < size && p[i] >= '0' && p[i] <= '9') ++i;
+ }
+ return i;
+}
+
+/* A fixed field's value from its text (a query value or a JSON number) into
+ * the prefix: a u32 in range, or a finite number. */
+static bool fixed_field(const PureField *field, const unsigned char *text, size_t size, unsigned char *prefix) {
+ bool integer;
+ if (!size || json_number(text, size, &integer) != size || size > 40) return false;
+ char digits[48];
+ memcpy(digits, text, size);
+ digits[size] = 0;
+ if (field->type == PURE_FIELD_U32) {
+ uint64_t n;
+ if (!integer || !number(digits, UINT32_MAX, &n)) return false;
+ put_u32(prefix + field->offset, (uint32_t)n);
+ return true;
+ }
+ double value = strtod(digits, NULL);
+ if (!isfinite(value)) return false;
+ memcpy(prefix + field->offset, &value, 8);
+ return true;
+}
+
+/* What an app route names in the module, with a message's fixed fields
+ * taken from the query into c->fields. NULL, or the error to answer (404
+ * for a name the module lacks, 400 otherwise). */
static const char *resolve(const Host *host, Connection *c, char *query, Route *route, int *status) {
const PureModule *m = host->module;
*status = 404;
c->field_bytes = 0;
+ memset(c->given, 0, sizeof(c->given));
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);
@@ -428,39 +484,195 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route *
bool message = shape && shape->kind == PURE_KIND_MESSAGE;
if (message && shape->keyed) return "keyed";
if (shape) {
- const PureField *last = message && shape->field_count ? &m->field[shape->field_start + shape->field_count - 1] : NULL;
- if (c->body_size && shape->kind != PURE_KIND_TEXT && !(last && last->type == PURE_FIELD_BYTES)) return "body";
+ /* A body is a text input's text, a message's one variable field, or a
+ * JSON object of a message's fields (two variable fields need one). */
+ unsigned variables = 0;
+ for (uint32_t f = 0; message && f < shape->field_count; ++f)
+ variables += m->field[shape->field_start + f].type == PURE_FIELD_BYTES ||
+ m->field[shape->field_start + f].type == PURE_FIELD_TEXT;
+ if (c->json && !message) return "json";
+ if (!c->json && (variables > 1 || (c->body_size && shape->kind == PURE_KIND_KEY) ||
+ (c->body_size && message && !variables))) return variables > 1 ? "json" : "body";
+ c->input = (uint32_t)found;
c->field_bytes = message ? shape->prefix : 0;
}
- /* name=value pairs: a table's from and count, a message's u32 fields. */
- bool given[PURE_MODULE_FIELDS] = {false};
+ /* name=value pairs: a table's from and count, a message's fixed fields. */
for (char *pair = query; pair && *pair;) {
char *end = strchr(pair, '&'), *value = strchr(pair, '=');
if (end) *end = 0;
if (!value || (end && value > end)) return "query";
*value++ = 0;
- uint64_t n;
- if (!number(value, route->kind == ROUTE_VIEW ? SIZE_MAX : UINT32_MAX, &n)) return "query";
if (route->kind == ROUTE_VIEW) {
+ uint64_t n;
bool from = !strcmp(pair, "from");
- if (m->view_type[found] != PURE_VIEW_TABLE || (!from && strcmp(pair, "count"))) return "query";
+ if (!number(value, SIZE_MAX, &n) || m->view_type[found] != PURE_VIEW_TABLE || (!from && strcmp(pair, "count")))
+ return "query";
if (from) route->from = (size_t)n;
else route->count = (size_t)n;
} else {
uint32_t f = 0;
while (f < (message ? shape->field_count : 0) && strcmp(m->field[shape->field_start + f].name, pair)) ++f;
const PureField *field = message && f < shape->field_count ? &m->field[shape->field_start + f] : NULL;
- if (!field || field->type != PURE_FIELD_U32 || given[f]) return "query";
- given[f] = true;
- for (unsigned i = 0; i < 4; ++i) c->fields[field->offset + i] = (unsigned char)(n >> (8 * i));
+ if (!field || c->given[f] || (field->type != PURE_FIELD_U32 && field->type != PURE_FIELD_NUMBER) ||
+ !fixed_field(field, (const unsigned char *)value, strlen(value), c->fields)) return "query";
+ c->given[f] = true;
}
pair = end ? end + 1 : NULL;
}
- for (uint32_t f = 0; message && f < shape->field_count; ++f)
- if (m->field[shape->field_start + f].type == PURE_FIELD_U32 && !given[f]) return "missing_field";
+ for (uint32_t f = 0; message && !c->json && f < shape->field_count; ++f) {
+ uint8_t type = m->field[shape->field_start + f].type;
+ if ((type == PURE_FIELD_U32 || type == PURE_FIELD_NUMBER) && !c->given[f]) return "missing_field";
+ }
+ return NULL;
+}
+
+/* A JSON string's contents (after its opening quote) up to the closing one:
+ * its length, or 0 when it does not close or holds a control byte. */
+static size_t json_string_end(const unsigned char *p, size_t size) {
+ for (size_t i = 0; i < size; ++i) {
+ if (p[i] == '"') return i + 1;
+ if (p[i] < 0x20) return 0;
+ if (p[i] == '\\' && ++i == size) return 0;
+ }
+ return 0;
+}
+
+static int hex_digit(unsigned char c) {
+ return c >= '0' && c <= '9' ? c - '0' : c >= 'a' && c <= 'f' ? c - 'a' + 10 : c >= 'A' && c <= 'F' ? c - 'A' + 10 : -1;
+}
+
+static bool json_code(const unsigned char *p, size_t left, uint32_t *code) {
+ if (left < 4) return false;
+ *code = 0;
+ for (int i = 0; i < 4; ++i) {
+ int d = hex_digit(p[i]);
+ if (d < 0) return false;
+ *code = *code << 4 | (uint32_t)d;
+ }
+ return true;
+}
+
+/* Unescape a JSON string's contents into out (never longer); false for a
+ * bad escape or a lone surrogate. The result is checked as UTF-8 after. */
+static bool json_unescape(const unsigned char *in, size_t size, unsigned char *out, size_t *written) {
+ size_t o = 0;
+ for (size_t i = 0; i < size;) {
+ unsigned char b = in[i++];
+ if (b != '\\') { out[o++] = b; continue; }
+ if (i == size) return false;
+ b = in[i++];
+ const char *plain = strchr("\"\\/bfnrt", b);
+ if (b && plain) { out[o++] = (unsigned char)"\"\\/\b\f\n\r\t"[plain - "\"\\/bfnrt"]; continue; }
+ uint32_t code, low;
+ if (b != 'u' || !json_code(in + i, size - i, &code)) return false;
+ i += 4;
+ if (code >= 0xDC00 && code <= 0xDFFF) return false;
+ if (code >= 0xD800 && code <= 0xDBFF) {
+ if (size - i < 6 || in[i] != '\\' || in[i + 1] != 'u' || !json_code(in + i + 2, size - i - 2, &low) ||
+ low < 0xDC00 || low > 0xDFFF) return false;
+ i += 6;
+ code = 0x10000 + ((code - 0xD800) << 10) + (low - 0xDC00);
+ }
+ if (code < 0x80) out[o++] = (unsigned char)code;
+ else if (code < 0x800) { out[o++] = (unsigned char)(0xC0 | code >> 6); out[o++] = (unsigned char)(0x80 | (code & 0x3F)); }
+ else if (code < 0x10000) {
+ out[o++] = (unsigned char)(0xE0 | code >> 12);
+ out[o++] = (unsigned char)(0x80 | (code >> 6 & 0x3F));
+ out[o++] = (unsigned char)(0x80 | (code & 0x3F));
+ } else {
+ out[o++] = (unsigned char)(0xF0 | code >> 18);
+ out[o++] = (unsigned char)(0x80 | (code >> 12 & 0x3F));
+ out[o++] = (unsigned char)(0x80 | (code >> 6 & 0x3F));
+ out[o++] = (unsigned char)(0x80 | (code & 0x3F));
+ }
+ }
+ *written = o;
+ return true;
+}
+
+static size_t json_space(const unsigned char *p, size_t at, size_t size) {
+ while (at < size && (p[at] == ' ' || p[at] == '\t' || p[at] == '\n' || p[at] == '\r')) ++at;
+ return at;
+}
+
+/* An app input's JSON body, as it came after the payload's room, becomes its
+ * message's payload: an object with a member for each field the query did
+ * not give (a u32, a number, or a string for bytes and text), and no other.
+ * NULL, or the error to answer. */
+static const char *convert(Connection *c) {
+ const PureModule *m = c->host->module;
+ const PureInputShape *shape = &m->input[c->input];
+ const unsigned char *json = c->body;
+ size_t size = c->body_size;
+ unsigned char *payload = c->ingress + 4;
+ struct { const unsigned char *value; size_t size; bool string; } member[PURE_MODULE_FIELDS];
+ bool seen[PURE_MODULE_FIELDS] = {false};
+ size_t at = json_space(json, 0, size);
+ if (at == size || json[at++] != '{') return "json";
+ at = json_space(json, at, size);
+ for (bool first = true; at < size && json[at] != '}'; first = false) {
+ if (!first) {
+ if (json[at++] != ',') return "json";
+ at = json_space(json, at, size);
+ }
+ size_t key_end = at < size && json[at] == '"' ? json_string_end(json + at + 1, size - at - 1) : 0;
+ if (!key_end) return "json";
+ const unsigned char *key = json + at + 1;
+ size_t key_size = key_end - 1;
+ at = json_space(json, at + 1 + key_end, size);
+ if (at == size || json[at++] != ':') return "json";
+ at = json_space(json, at, size);
+ uint32_t f = 0;
+ while (f < shape->field_count && (strlen(m->field[shape->field_start + f].name) != key_size ||
+ memcmp(m->field[shape->field_start + f].name, key, key_size))) ++f;
+ if (f == shape->field_count || seen[f] || c->given[f]) return "json_field";
+ seen[f] = true;
+ bool integer, string = at < size && json[at] == '"';
+ size_t length = string ? json_string_end(json + at + 1, size - at - 1) : json_number(json + at, size - at, &integer);
+ if (!length) return "json";
+ member[f].value = json + at + string;
+ member[f].size = length - string;
+ member[f].string = string;
+ at = json_space(json, at + string + length, size);
+ }
+ if (at == size || json_space(json, at + 1, size) != size) return "json";
+ /* The fields in order: fixed ones into the prefix, strings after it. */
+ size_t used = shape->prefix;
+ for (uint32_t f = 0; f < shape->field_count; ++f) {
+ const PureField *field = &m->field[shape->field_start + f];
+ bool variable = field->type == PURE_FIELD_BYTES || field->type == PURE_FIELD_TEXT;
+ if (!seen[f]) {
+ if (!c->given[f]) return "missing_field";
+ continue;
+ }
+ if (member[f].string != variable) return "json_type";
+ if (!variable) {
+ if (!fixed_field(field, member[f].value, member[f].size, payload)) return "json_type";
+ continue;
+ }
+ size_t written;
+ if (!json_unescape(member[f].value, member[f].size, payload + used, &written) ||
+ !pure_utf8_valid((PureSpan){payload + used, written})) return "json";
+ if (field->offset != PURE_FIELD_REST) put_u32(payload + field->offset, (uint32_t)written);
+ used += written;
+ }
+ put_u32(c->ingress, (uint32_t)used);
return NULL;
}
+/* A request's body is in: an app input's JSON becomes its payload, and the
+ * request is ready, or it is answered 400 and never runs. */
+static void body_done(Connection *c) {
+ const char *why = c->route == ROUTE_INPUT && c->json ? convert(c) : NULL;
+ if (why) {
+ release_ticket(c); /* never made ready: it never runs */
+ respond(c, 400, "Bad Request", why, 0, (PureSpan){0});
+ return;
+ }
+ if (!pure_runtime_ready(c->ticket)) abort();
+ c->phase = WAITING;
+}
+
/* A whole head is in `in`: admit the request and take any body bytes that
* came with it; later bytes stay for the next request. */
static void start_request(Connection *c, size_t head_size) {
@@ -477,11 +689,13 @@ static void start_request(Connection *c, size_t head_size) {
int status;
const char *why = resolve(host, c, query, &route, &status);
if (why) { reject(c, status, status == 404 ? "Not Found" : "Bad Request", why); return; }
- /* A document's PUT and an input's event carry a payload: the message's fields, then the body. */
+ /* A document's PUT carries its body; an input's event its payload's size,
+ * the payload (the fixed fields, then the body), and room for its JSON. */
bool write = route.kind == ROUTE_INPUT || c->put;
+ size_t input_size = route.kind == ROUTE_INPUT ? 4 + c->field_bytes + c->body_size * (c->json ? 2 : 1) :
+ write ? c->body_size : 0;
PureResult admitted = pure_runtime_accept(&host->runtime, write ? PURE_INPUT_WRITE : PURE_INPUT_READ, c->key,
- route.kind == ROUTE_INPUT ? 0 : c->expected,
- write ? c->field_bytes + c->body_size : 0, &c->ticket);
+ route.kind == ROUTE_INPUT ? 0 : c->expected, input_size, &c->ticket);
if (admitted != PURE_OK) {
c->ticket = (PureTicket){0};
/* An unread body would desynchronize the stream. */
@@ -492,19 +706,20 @@ static void start_request(Connection *c, size_t head_size) {
host->by_slot[c->ticket.slot] = c;
host->routes[c->ticket.slot] = route;
if (!write) { c->phase = WAITING; return; }
- unsigned char *ingress = pure_runtime_ingress(c->ticket, NULL); /* NULL for no bytes, as a key press */
- if (c->field_bytes) memcpy(ingress, c->fields, c->field_bytes);
- c->body = ingress ? ingress + c->field_bytes : NULL;
+ unsigned char *ingress = c->ingress = pure_runtime_ingress(c->ticket, NULL); /* NULL for no bytes */
+ c->body = ingress;
+ if (route.kind == ROUTE_INPUT) {
+ put_u32(ingress, (uint32_t)(c->field_bytes + c->body_size));
+ if (c->field_bytes) memcpy(ingress + 4, c->fields, c->field_bytes);
+ c->body = ingress + 4 + c->field_bytes + (c->json ? c->body_size : 0);
+ }
size_t take = c->in_used < c->body_size ? c->in_used : c->body_size;
if (take) memcpy(c->body, c->in, take);
memmove(c->in, c->in + take, c->in_used - take);
c->in_used -= take;
c->received = take;
c->phase = READING_BODY;
- if (c->received == c->body_size) {
- if (!pure_runtime_ready(c->ticket)) abort();
- c->phase = WAITING;
- }
+ if (c->received == c->body_size) body_done(c);
}
static size_t head_end(const Connection *c) {
@@ -534,10 +749,7 @@ static void drive(Connection *c) {
if (n > 0) {
c->received += (size_t)n;
c->deadline = now_ms() + c->host->timeout_ms;
- if (c->received == c->body_size) {
- if (!pure_runtime_ready(c->ticket)) abort();
- c->phase = WAITING;
- }
+ if (c->received == c->body_size) body_done(c);
continue;
}
if (n < 0 && errno == EAGAIN) { c->readable = false; return; }
@@ -741,14 +953,15 @@ static PureResult show_view(PureApp *app, const Route *route, size_t available,
* whose reply (a field of it) is borrowed from the request. */
static PureResult apply_input(PureApp *app, const Route *route, const PureEffect *effect, Applied *out) {
const PureInputShape *shape = &app->module->input[route->target];
+ PureSpan payload = {effect->value.data + 4, get_u32(effect->value.data)};
PureResult status;
if (shape->kind == PURE_KIND_TEXT) {
- status = pure_app_text(app, route->target, effect->value);
+ status = pure_app_text(app, route->target, payload);
} else if (shape->kind == PURE_KIND_KEY) {
status = pure_app_input(app, route->target, PURE_EDGE_DOWN);
if (status == PURE_OK) status = pure_app_input(app, route->target, PURE_EDGE_UP);
} else {
- status = pure_app_message(app, route->target, NULL, -1, effect->value);
+ status = pure_app_message(app, route->target, NULL, -1, payload);
Reply reply = {0};
if (status == PURE_OK) pure_app_reply(app, measure_reply, &reply);
out->reply = (PureSpan){reply.size ? reply.first : NULL, reply.size};
@@ -994,9 +1207,12 @@ static bool replies_borrow(const PureModule *m, uint32_t input) {
for (uint32_t h = 0; h < m->handlers; ++h) {
if (m->handler[h].event != 1 + 2 * input) continue;
const PureBlock *code = &m->handler[h].code;
- for (uint32_t pc = code->start; pc < code->start + code->length; ++pc)
- if (m->code[pc].op == PURE_OP_REPLY && (pc == code->start || m->code[pc - 1].op != PURE_OP_FIELD))
- return false;
+ /* A field replied as it is, or a text field as its bytes. */
+ for (uint32_t pc = code->start; pc < code->start + code->length; ++pc) {
+ if (m->code[pc].op != PURE_OP_REPLY) continue;
+ uint32_t from = pc > code->start && m->code[pc - 1].op == PURE_OP_BYTES_FROM ? pc - 1 : pc;
+ if (from == code->start || m->code[from - 1].op != PURE_OP_FIELD) return false;
+ }
}
return true;
}
diff --git a/prototypes/connections/test-module.c b/prototypes/connections/test-module.c
index 11d0660..c881678 100644
--- a/prototypes/connections/test-module.c
+++ b/prototypes/connections/test-module.c
@@ -627,8 +627,8 @@ static void messages(void) {
static const H note_handler[] = {{1, 0, 2}};
bad = (Spec){states, 1, unkeyed, 1, views, 0, note_handler, 1, doc_read, 2, collections, 1, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0};
CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "document operation outside"));
- static const F tail_first[] = {{"insert", 2}, {"start", 1}};
- static const I misplaced[] = {{"edit", NULL, PURE_KIND_MESSAGE, true, tail_first, 2}};
+ static const F unknown_type[] = {{"insert", 5}, {"start", 1}};
+ static const I misplaced[] = {{"edit", NULL, PURE_KIND_MESSAGE, true, unknown_type, 2}};
bad = (Spec){states, 1, misplaced, 1, views, 0, NULL, 0, NULL, 0, collections, 1, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0};
CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "message field"));
static const H even[] = {{2, 0, 0}};
@@ -2191,12 +2191,81 @@ static void keyed_mutations(void) {
printf("{\"keyed_module_mutations\":20000,\"accepted_after_damage\":%u}\n", accepted);
}
+/* A message of a text, a number, bytes and a u32, in that order: the prefix
+ * holds the text's length, the number and the u32; the text's bytes come
+ * first after it and the bytes (the last variable field) take the rest. A
+ * length past the payload, a text that is not UTF-8 or a number that is not
+ * finite fails the message. */
+enum { TSET = PURE_OP_TEXT_SET };
+static void message_fields(void) {
+ static const S nums[] = {{"n", 0}};
+ static const T texts[] = {{"title", 1, 0}};
+ static const char *const constants[] = {""};
+ static const F fields[] = {{"title", PURE_FIELD_TEXT}, {"points", PURE_FIELD_NUMBER}, {"note", PURE_FIELD_BYTES},
+ {"count", PURE_FIELD_U32}};
+ static const I inputs[] = {{"post", NULL, PURE_KIND_MESSAGE, false, fields, 4}};
+ static const O code[] = {{FIELD, 1, 0}, {FIELD, 3, 0}, {ADD, 0, 0}, {ST, 0, 0}, {FIELD, 0, 0}, {TSET, 1, 0},
+ {FIELD, 2, 0}, {REPLY, 0, 0}};
+ static const H handlers[] = {{1, 0, 8}};
+ Spec spec = {nums, 1, inputs, 1, NULL, 0, handlers, 1, code, LEN(code), NULL, 0, texts, 1, constants, 1, NULL, 0,
+ NULL, 0, NULL, 0};
+ if (load(spec, &module_a) != PURE_OK) { fprintf(stderr, "%s\n", error); abort(); }
+ CHECK(module_a.input[0].prefix == 16 && module_a.field[0].offset == 0 && module_a.field[1].offset == 4 &&
+ module_a.field[2].offset == PURE_FIELD_REST && module_a.field[3].offset == 12);
+ PureApp app;
+ CHECK(pure_app_start(&app, &module_a, (PureData){0}) == PURE_OK);
+ unsigned char payload[64];
+ /* title "가나" (6 bytes), points 2.5, count 3, then the title's bytes and note "xyz". */
+ size_t size = 0;
+ uint32_t title_size = 6, count = 3;
+ double points = 2.5;
+ memcpy(payload, &title_size, 4);
+ memcpy(payload + 4, &points, 8);
+ memcpy(payload + 12, &count, 4);
+ memcpy(payload + 16, "\xea\xb0\x80\xeb\x82\x98" "xyz", 9);
+ size = 25;
+ CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, size}) == PURE_OK);
+ CHECK(app.values[0] == 5.5 && !strcmp(state_text(&app, 1), "\xea\xb0\x80\xeb\x82\x98") && replied(&app, "xyz"));
+ pure_app_finish(&app);
+ /* Empty variable fields. */
+ title_size = 0;
+ memcpy(payload, &title_size, 4);
+ CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, 16}) == PURE_OK && !*state_text(&app, 1) && replied(&app, ""));
+ pure_app_finish(&app);
+ /* A title longer than what is there, a title that is not UTF-8, a number that is not finite, a short prefix. */
+ title_size = 10;
+ memcpy(payload, &title_size, 4);
+ CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, 25}) == PURE_LIMIT && app.fault == PURE_FAULT_LIMIT);
+ title_size = 2;
+ memcpy(payload, &title_size, 4);
+ memcpy(payload + 16, "\xff\xfe", 2);
+ CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, 18}) == PURE_INVALID && app.fault == PURE_FAULT_ENCODING);
+ title_size = 0;
+ points = NAN;
+ memcpy(payload, &title_size, 4);
+ memcpy(payload + 4, &points, 8);
+ CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, 16}) == PURE_INVALID && app.fault == PURE_FAULT_NON_FINITE);
+ CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, 15}) == PURE_LIMIT);
+ CHECK(app.values[0] == 5.5 && !*state_text(&app, 1)); /* failed messages changed nothing */
+ pure_app_stop(&app);
+ /* A text field is text and a bytes field bytes. */
+ static const O text_as_bytes[] = {{FIELD, 0, 0}, {LEN_, 0, 0}, {ST, 0, 0}};
+ static const O bytes_as_text[] = {{FIELD, 2, 0}, {TSET, 1, 0}};
+ static const H three[] = {{1, 0, 3}}, two[] = {{1, 0, 2}};
+ Spec bad = spec;
+ bad.o = text_as_bytes; bad.no = 3; bad.h = three;
+ CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "operand type"));
+ bad.o = bytes_as_text; bad.no = 2; bad.h = two;
+ CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "operand type"));
+}
+
int main(void) {
behaviour();
faults();
rejections();
mutations();
messages();
+ message_fields();
message_mutations();
texts();
marks();
@@ -2214,6 +2283,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\"}");
+ 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\"}");
return 0;
}
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 18:46:52 +0000