From ddff7e15d1d0e90a5f5a52fcc93828c8ef0bbd5c Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sun, 27 Sep 2026 08:22:21 +0100 Subject: Make an API's answers the app's events (--reply OUTPUT=INPUT) The final answer to each send of a wired output (a 2xx, or a 4xx but 408 and 429) becomes a message event of the given input: fields from the answer's JSON object by name, other members skipped; a field the answer lacks from the send's field of that name and type, so the request's own id is there; a u32 http_status field is the status. It is applied with the send's number from the wire's source, before the send is trimmed and in the same save, so an answer that comes again after a restart (the API repeating itself for the Idempotency-Key) applies once. Sends that wait for answers go one per request; answers are saved with the next round, or at once when none waits. A server that writes an answer in pieces holds the second until the first is acknowledged, and a delayed acknowledgement stalled every answer 40 ms: the client now acknowledges at once (TCP_QUICKACK) after each read. Against a payments API that answers 200 or 402, with the sender killed (-9) midway, paid and declined orders matched the API's answers one to one. --- prototypes/connections/deliver.c | 191 +++++++++++++++++++++++++++++++++++---- 1 file changed, 171 insertions(+), 20 deletions(-) (limited to 'prototypes/connections/deliver.c') diff --git a/prototypes/connections/deliver.c b/prototypes/connections/deliver.c index c4136a2..0d3da64 100644 --- a/prototypes/connections/deliver.c +++ b/prototypes/connections/deliver.c @@ -1,5 +1,6 @@ #define _POSIX_C_SOURCE 200809L #include "deliver.h" +#include "message-json.h" #include "module-text.h" #include #include @@ -18,10 +19,14 @@ #include #include -enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256 }; -enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED }; +enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256, + ANSWER_MOST = 1 << 20 }; +enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED, ANSWER_UNREAD }; -typedef struct { char *body; size_t size; uint64_t seq; } Pending; +typedef struct { char *body; size_t size; uint64_t seq; unsigned char *sent; size_t sent_size; } Pending; +/* An answer made into its input's message: a table of the fields' starts and + * sizes (u32le), then the payload from its prefix on. */ +typedef struct { uint64_t seq; unsigned char *bytes; size_t size, table; } Answer; struct PureWire { const PureModule *module; @@ -40,6 +45,11 @@ struct PureWire { int fd; /* the connection kept for the next send, or -1; changed under the lock */ bool tls; /* https: the connection is TLS */ SSL *ssl; /* its TLS session */ + int answer_input; /* the input answers become events of, or -1 */ + Answer *answers; /* made, for the owner to take in order */ + size_t answer_head, answer_count, answer_capacity; + Answer taken; /* the one the owner took last, kept until it takes the next */ + uint32_t placed[2 * PURE_MODULE_FIELDS]; }; /* ---- TLS: one client context, verifying the receiver's certificate against @@ -226,12 +236,42 @@ PureWire *pure_wire_open(const PureModule *m, const char *output, const char *ur memcpy(w->source, source, length + 1); w->fd = -1; w->tls = tls; + w->answer_input = -1; pthread_mutex_init(&w->lock, NULL); pthread_cond_init(&w->changed, NULL); return w; } uint32_t pure_wire_output(const PureWire *w) { return w->output; } + +bool pure_wire_reply(PureWire *w, uint32_t input, char *error, size_t size) { + const PureModule *m = w->module; + if (input >= m->inputs || m->input[input].kind != PURE_KIND_MESSAGE || m->input[input].keyed) { + snprintf(error, size, "an answer goes to a message input without a key"); + return false; + } + w->answer_input = (int)input; + return true; +} + +bool pure_wire_answer(PureWire *w, PureWireAnswer *out) { + free(w->taken.bytes); + w->taken = (Answer){0}; + pthread_mutex_lock(&w->lock); + bool some = w->answer_count > 0; + if (some) { + w->taken = w->answers[w->answer_head++]; + --w->answer_count; + w->delivered = w->taken.seq; /* trimmed after the owner applies it */ + } + pthread_mutex_unlock(&w->lock); + if (!some) return false; + const PureInputShape *in = &w->module->input[w->answer_input]; + for (size_t i = 0; i < 2 * (size_t)in->field_count; ++i) w->placed[i] = read_u32(w->taken.bytes + 4 * i); + *out = (PureWireAnswer){(uint32_t)w->answer_input, w->taken.seq, + {w->taken.bytes + w->taken.table, w->taken.size - w->taken.table}, w->placed, w->source}; + return true; +} uint64_t pure_wire_queued(const PureWire *w) { return w->queued; } uint64_t pure_wire_delivered(PureWire *w) { @@ -275,7 +315,14 @@ bool pure_wire_queue(PureWire *w, uint64_t seq, PureSpan payload) { w->capacity = capacity; } } - w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq}; + unsigned char *sent = NULL; + if (w->answer_input >= 0 && !(sent = malloc(payload.size ? payload.size : 1))) { + pthread_mutex_unlock(&w->lock); + free(t.bytes); + return false; + } + if (sent && payload.size) memcpy(sent, payload.data, payload.size); + w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq, sent, payload.size}; w->queued = seq; pthread_cond_signal(&w->changed); pthread_mutex_unlock(&w->lock); @@ -294,6 +341,13 @@ static bool fill(Reader *r) { if (w->ssl) n = SSL_read(w->ssl, r->bytes + r->end, (int)(sizeof(r->bytes) - r->end)); else do n = recv(w->fd, r->bytes + r->end, sizeof(r->bytes) - r->end, 0); while (n < 0 && errno == EINTR); if (n <= 0) return false; +#ifdef TCP_QUICKACK + /* Acknowledge at once: a server that writes an answer in pieces holds the + * next piece until the first is acknowledged (Nagle), which a delayed + * acknowledgement would stall for 40 ms. The flag does not stay set. */ + int one = 1; + setsockopt(w->fd, IPPROTO_TCP, TCP_QUICKACK, &one, sizeof(one)); +#endif r->end += (size_t)n; return true; } @@ -320,14 +374,17 @@ static bool read_line(Reader *r, char *line, size_t size) { } } -static bool skip(Reader *r, uint64_t size) { +/* size bytes of the body: kept in into (up to ANSWER_MOST), or skipped when it is NULL. */ +static bool take(Reader *r, uint64_t size, Text *into) { + if (into && size > ANSWER_MOST - into->size) return false; while (size) { if (r->at == r->end && !fill(r)) return false; size_t n = r->end - r->at < size ? r->end - r->at : (size_t)size; + if (into) put(into, r->bytes + r->at, n); r->at += n; size -= n; } - return true; + return !into || !into->failed; } static bool send_all(PureWire *w, const char *bytes, size_t size) { @@ -397,7 +454,7 @@ static bool refusal(const char *error) { /* The answer to one request: its status, whether a Pure receiver refused it, * and whether the connection may carry the next. */ -static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, unsigned *batch) { +static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, unsigned *batch, Text *body) { Reader r = {.wire = w}; char line[1024]; if (!read_line(&r, line, sizeof(line)) || strncmp(line, "HTTP/1.", 7) || strlen(line) < 12) return false; @@ -414,15 +471,18 @@ static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, uns if (!read_line(&r, line, sizeof(line))) return false; chunk = strtoull(line, NULL, 16); if (!chunk) break; - if (!skip(&r, chunk) || !read_line(&r, line, sizeof(line))) return false; + if (!take(&r, chunk, body) || !read_line(&r, line, sizeof(line))) return false; } while (read_line(&r, line, sizeof(line)) && *line) {} return true; } if (*status == 204 || *status == 304 || *status < 200) return true; - if (sized) return skip(&r, length); + if (sized) return take(&r, length, body); *keep = false; /* the body runs to the close */ - while (fill(&r)) r.at = r.end; + while (fill(&r)) { + if (body && !take(&r, r.end - r.at, body)) return false; + r.at = r.end; + } return true; } char *value = strchr(line, ':'); @@ -438,9 +498,64 @@ static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, uns return false; } +static void write_u32(unsigned char *p, uint32_t v) { for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> (8 * i)); } + +/* The answer (status, body) to the send p as a message of the answer input: + * fields from the body's JSON object by name, others skipped; a field the + * body lacks from the send's field of that name and type; a u32 field named + * http_status from the status. False when a field has none of these. */ +static bool form_answer(const PureWire *w, int status, const Text *body, const Pending *p, Answer *out) { + const PureModule *m = w->module; + const PureInputShape *in = &m->input[w->answer_input], *sent = &m->input[w->output]; + size_t table = 8 * (size_t)in->field_count, total = table + in->prefix + body->size + p->sent_size; + unsigned char *bytes = calloc(1, total ? total : 1); + if (!bytes) return false; + unsigned char *prefix = bytes + table, *text = prefix + in->prefix, *copy = text + body->size; + if (body->size) memcpy(text, body->bytes, body->size); + if (p->sent_size) memcpy(copy, p->sent, p->sent_size); + PureJsonMessage message = {.module = m, .shape = in, .payload = prefix, .table = bytes, .lenient = true}; + size_t at = pure_json_space(text, 0, body->size); + if (at < body->size && text[at] == '{' && pure_json_message(&message, text, body->size, &at)) { + free(bytes); + return false; + } + for (uint32_t f = 0; f < in->field_count; ++f) { + if (message.seen[f]) continue; + const PureField *field = &m->field[in->field_start + f]; + if (field->type == PURE_FIELD_U32 && !strcmp(field->name, "http_status")) { + write_u32(prefix + field->offset, (uint32_t)status); + continue; + } + /* The send's field of the same name and type, and where its bytes are. */ + size_t start = sent->prefix, length = 0; + const PureField *from = NULL; + for (uint32_t g = 0; g < sent->field_count && !from; ++g) { + const PureField *other = &m->field[sent->field_start + g]; + bool variable = other->type == PURE_FIELD_BYTES || other->type == PURE_FIELD_TEXT; + if (variable) length = other->offset == PURE_FIELD_REST ? p->sent_size - start : read_u32(p->sent + other->offset); + if (!strcmp(other->name, field->name) && other->type == field->type) from = other; + else if (variable) start += length; + } + bool variable = field->type == PURE_FIELD_BYTES || field->type == PURE_FIELD_TEXT; + if (!from || (variable && (start > p->sent_size || length > p->sent_size - start))) { + free(bytes); + return false; + } + if (field->type == PURE_FIELD_U32 || field->type == PURE_FIELD_NUMBER) + memcpy(prefix + field->offset, p->sent + from->offset, field->type == PURE_FIELD_U32 ? 4 : 8); + else { + write_u32(bytes + 8 * f, (uint32_t)(copy + start - prefix)); + write_u32(bytes + 8 * f + 4, (uint32_t)length); + } + } + *out = (Answer){p->seq, bytes, total, table}; + return true; +} + /* One request for count sends from p (numbered in a row): one object, or - * an array of them for a receiver that takes batches. */ -static int post(PureWire *w, const Pending *p, unsigned count) { + * an array of them for a receiver that takes batches. A final answer to a + * send whose answers are wired is made into *answer. */ +static int post(PureWire *w, const Pending *p, unsigned count, Answer *answer) { size_t body = count > 1 ? count + 1 : 0; for (unsigned i = 0; i < count; ++i) body += p[i].size; char head[2048], batch[32] = ""; @@ -473,19 +588,30 @@ static int post(PureWire *w, const Pending *p, unsigned count) { int status; bool refused, keep, sent = send_all(w, head, (size_t)size); unsigned batch_size = 0; + Text body = {0}; for (unsigned i = 0; sent && i < count; ++i) sent = (!i || count == 1 || send_all(w, ",", 1)) && (i || count == 1 || send_all(w, "[", 1)) && send_all(w, p[i].body, p[i].size); if (sent && count > 1) sent = send_all(w, "]", 1); - if (!sent || !read_answer(w, &status, &refused, &keep, &batch_size)) { + if (!sent || !read_answer(w, &status, &refused, &keep, &batch_size, w->answer_input >= 0 ? &body : NULL)) { + free(body.bytes); drop_connection(w); if (fresh) return AGAIN; continue; } if (!keep) drop_connection(w); - if (batch_size) w->batch = batch_size; + if (batch_size && w->answer_input < 0) w->batch = batch_size; + if (status == 408 || status == 429 || status >= 500) { + free(body.bytes); + return AGAIN; + } + if (w->answer_input >= 0 && status >= 200 && status < 500) { /* a final answer: the app hears it */ + bool formed = form_answer(w, status, &body, p, answer); + free(body.bytes); + return formed ? DELIVERED : ANSWER_UNREAD; + } + free(body.bytes); if ((status >= 200 && status < 300) || (status >= 400 && status < 500 && refused)) return DELIVERED; - if (status == 408 || status == 429 || status >= 500) return AGAIN; return REFUSED_BY_PEER; } return AGAIN; @@ -507,13 +633,31 @@ static void *deliver(void *context) { memcpy(next, w->queue + w->head, count * sizeof(*next)); ++w->attempts; pthread_mutex_unlock(&w->lock); - int outcome = post(w, next, count); + Answer answer = {0}; + int outcome = post(w, next, count, &answer); pthread_mutex_lock(&w->lock); + if (outcome == DELIVERED && answer.bytes) { /* for the owner, who then counts it delivered */ + if (w->answer_head + w->answer_count == w->answer_capacity) { + if (w->answer_head) { + memmove(w->answers, w->answers + w->answer_head, w->answer_count * sizeof(*w->answers)); + w->answer_head = 0; + } else { + size_t capacity = w->answer_capacity ? 2 * w->answer_capacity : 16; + Answer *grown = realloc(w->answers, capacity * sizeof(*grown)); + if (!grown) { free(answer.bytes); outcome = AGAIN; } + else { w->answers = grown; w->answer_capacity = capacity; } + } + } + if (outcome == DELIVERED) w->answers[w->answer_head + w->answer_count++] = answer; + } if (outcome == DELIVERED) { - for (unsigned i = 0; i < count; ++i) free(w->queue[w->head + i].body); + for (unsigned i = 0; i < count; ++i) { + free(w->queue[w->head + i].body); + free(w->queue[w->head + i].sent); + } w->head += count; w->count -= count; - w->delivered = next[count - 1].seq; + if (!answer.bytes) w->delivered = next[count - 1].seq; w->deliveries += count; pause_ms = 0; pthread_mutex_unlock(&w->lock); @@ -521,9 +665,10 @@ static void *deliver(void *context) { pthread_mutex_lock(&w->lock); continue; } - if ((outcome == REFUSED_BY_PEER || outcome == TLS_FAILED) && reported != next[0].seq) { + if (outcome != AGAIN && reported != next[0].seq) { fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"%s\"}\n", w->source, (unsigned long long)next[0].seq, outcome == TLS_FAILED ? "the receiver's certificate is not trusted; trying again" : + outcome == ANSWER_UNREAD ? "its answer does not make the answer input's message; trying again" : "the receiver does not take it; trying again"); reported = next[0].seq; } @@ -557,7 +702,13 @@ void pure_wire_close(PureWire *w) { if (w->started) pthread_join(w->thread, NULL); if (w->ssl) SSL_free(w->ssl); if (w->fd >= 0) close(w->fd); - for (size_t i = 0; i < w->count; ++i) free(w->queue[w->head + i].body); + for (size_t i = 0; i < w->count; ++i) { + free(w->queue[w->head + i].body); + free(w->queue[w->head + i].sent); + } + for (size_t i = 0; i < w->answer_count; ++i) free(w->answers[w->answer_head + i].bytes); + free(w->answers); + free(w->taken.bytes); free(w->queue); pthread_cond_destroy(&w->changed); pthread_mutex_destroy(&w->lock); -- cgit v1.2.3