From 5d3a7cefa27f3fe20a9c6991fb044815b38aac01 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sun, 27 Sep 2026 08:09:40 +0100 Subject: Deliver an app's saved sends over HTTP, in order and in batches (083) pure-serve --send OUTPUT=http://host:port/path wires an output: after every save the owner queues the new sends as JSON (fields and groups) and a thread per output posts them in order with X-Pure-Source, X-Pure-Sequence and Idempotency-Key. A 2xx, or a Pure receiver's refusal, delivers; the owner then trims. 408, 429, 5xx and no answer are sent again after a pause growing to 10 s, other answers every 10 s: nothing is dropped. A Pure receiver applies a delivered message with pure_app_deliver (now taking placed fields) and advertises X-Pure-Batch: 256; the sender then posts up to that many numbers in a row as one JSON array (X-Pure-Count), applied in order in one burst with one save. Delivery kept up with orders once batched (391 deliveries/s one at a time; a backlog of 20,000 went at 60k/s). With four clients ordering, killing the receiver and then the sender (-9) left the receiver's count and total equal to what the sender saved, five times and under ASan. --- prototypes/connections/test-module.c | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'prototypes/connections/test-module.c') diff --git a/prototypes/connections/test-module.c b/prototypes/connections/test-module.c index bb59174..ccb3371 100644 --- a/prototypes/connections/test-module.c +++ b/prototypes/connections/test-module.c @@ -2581,7 +2581,7 @@ static PureResult deliver(PureApp *app, const char *source, uint64_t seq, const 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}); + PureResult result = pure_app_deliver(app, 0, source, seq, (PureSpan){payload, 12 + length}, NULL); pure_app_finish(app); return result; } @@ -2618,10 +2618,10 @@ static void links(void) { 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 && + CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent, NULL) == 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(pure_app_deliver(&b, 0, "orders.ordered", 1, sent, NULL) == 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); @@ -2630,7 +2630,7 @@ static void links(void) { 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(deliver(&b, "other", 0, "x", 1) == PURE_INVALID && pure_app_deliver(&a, 2, "s", 1, sent, NULL) == 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); -- cgit v1.2.3