From 92290c5102e7fbb1248a0f215b1099de4d00c143 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Mon, 28 Sep 2026 04:31:43 +0100 Subject: Take a message's rows over HTTP A rows field's JSON value is an array of objects naming the record's fields (numbers, strings; a field left out takes its default). The connection thread encodes them as the list's stored form, which may be longer than the JSON, grows the request's input by that much (pure_runtime_grow: charged to the payload budget as admission charges an input, 503 payload_budget when it does not fit) and places them after the body, so the owner still applies a message it need not parse. A message whose only variable field is rows takes a body that is not JSON as those rows in their stored form: a bulk insert without JSON. An API's answer (--reply) gives rows the same way. A delivered batch holds no rows. server_rows loads 30,000 stock rows in one JSON request (some 250,000 rows a second, against 6,500 with a request a row) and 20,000 more as bytes, refuses a load with a key already there whole (409 exists), then checks 400 random orders against a model (294 applied, 106 refused with none of their rows applied), texts with escapes and Korean, a walk that removes, and a restart that keeps every state and no rows. The format's document (008) describes walks, rows fields and lists of rows, and OUTER_ROW. --- core/runtime.c | 29 +++++++++++++++++++++++++++++ core/runtime.h | 4 ++++ 2 files changed, 33 insertions(+) (limited to 'core') diff --git a/core/runtime.c b/core/runtime.c index 4a857c7..19ff215 100644 --- a/core/runtime.c +++ b/core/runtime.c @@ -201,6 +201,35 @@ unsigned char *pure_runtime_ingress(PureTicket ticket, size_t *size) { return (unsigned char *)w->resource.body.data; } +PureResult pure_runtime_grow(PureTicket ticket, size_t size) { + PureWork *w = lookup(ticket, true); + PureRuntime *r = ticket.runtime; + if (!w || r->running || w->state != RECEIVING || w->input != PURE_INPUT_WRITE || w->parent.runtime) + return PURE_INVALID; + size_t had = w->resource.body.size; + if (size <= had) return PURE_OK; + if (size - had > r->budget - r->retained) { + METRIC(++r->budget_rejections); + return PURE_BUDGET; + } + if (!r->host.allocate) return PURE_ALLOCATION; + PureResource grown = {0}; + r->running = true; + bool ok = r->host.allocate(r->host.context, size, &grown); + r->running = false; + if (!ok || !grown.body.data || grown.body.size != size || !grown.release) { + dispose(r, &grown); + return PURE_ALLOCATION; + } + if (had) memcpy((unsigned char *)grown.body.data, w->resource.body.data, had); + dispose(r, &w->resource); + w->resource = grown; + w->charged += size - had; + r->retained += size - had; + METRIC(if (r->retained > r->peak_retained) r->peak_retained = r->retained); + return PURE_OK; +} + bool pure_runtime_step(PureRuntime *r) { if (!r || r->running || r->ready_head == NONE) return false; #ifdef PURE_ASYNC_EFFECTS diff --git a/core/runtime.h b/core/runtime.h index 70f5d08..fd19525 100644 --- a/core/runtime.h +++ b/core/runtime.h @@ -110,6 +110,10 @@ bool pure_runtime_bind(PureRuntime *, PureWork *, size_t capacity); PureResult pure_runtime_accept(PureRuntime *, PureInput, const char *key, int64_t expected, size_t input_size, PureTicket *out); unsigned char *pure_runtime_ingress(PureTicket, size_t *size); +/* A WRITE input still being filled grows to size with its bytes kept (the + * ingress moves: take it again), charged to the budget as its admission was; + * BUDGET or ALLOCATION leave it as it was. */ +PureResult pure_runtime_grow(PureTicket, size_t size); bool pure_runtime_ready(PureTicket); bool pure_runtime_step(PureRuntime *); /* execute at most one ready job */ bool pure_runtime_result(PureTicket, PureOutcome *out); -- cgit v1.2.3