Serve modules through the runtime core with group commits

pure-connections-module offers the document server's HTTP contract with an
aup module: GET is its show message and PUT its edit message. Requests go
through the runtime core as before (work slots, payload budget, ingress,
replies borrowed from the request), and effects are taken on the owner
thread with the core's asynchronous interface.

Each poll round is one burst: every ready request is applied to the app
with deferred saving, one save stores them, and only then are the tickets
completed and answered. A failed save fails the whole burst, reads too. A
key seen again ends the burst first, and the module server accepts every
waiting connection at once so concurrent requests share a burst. On the
Pi's SD card eight clients' 240 edits take 0.59 s instead of 2.14 s.

3 files changed, 230 insertions, 8 deletions

diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md
index 4c5a9ac..2d30003 100644
--- a/docs/design/008-module-format.md
+++ b/docs/design/008-module-format.md
@@ -197,6 +197,25 @@ text·리스트가 남을 때만 위치를 유지한다.
 실패하면 묶음 전체가 되돌려지고, 저장 중 프로세스가 죽으면 묶음 전체가 남거나 전혀 남지 않는다. 비용(Pi 5, SD의 ext4): 저장하는 이벤트 하나에 약 10.6 ms가
 더 든다. 대부분 두 번의 `fsync`다. 300항목 리스트에서 인덱스 이동 한 번은 파일을 약 220 B 키운다(압축 전).
 
-아직 런타임 코어(`runtime.c`)의 작업·예산 모델과 연결하지 않았다. 호스트가 직접 호출한다. 실행기 코드는
+## 런타임 코어 위의 HTTP 서버
+
+`pure-connections-module --module 앱.pmod --store 경로`는 기존 문서 서버(`server.c`)와 같은 HTTP 계약을 모듈로
+제공한다. `GET /objects/키`는 모듈의 `show` 메시지, `PUT`(`If-Match`/`If-None-Match`)은 `edit` 메시지다. 요청은
+런타임 코어(`runtime.c`)를 그대로 거친다. 작업 슬롯 8개, 입력 바이트 예산, 코어가 소유한 ingress, 입력에서 빌린
+PUT 응답이 모두 기존과 같다. 효과는 비동기 인터페이스(`pure_runtime_take`/`pure_runtime_complete`)로 소유 스레드에서
+처리한다.
+
+- **poll 한 바퀴가 묶음 하나다.** 준비된 요청을 모두 꺼내 지연 저장 모드의 앱에 차례로 적용하고, `pure_app_save`
+  한 번으로 저장한 **뒤에** 티켓을 완료하고 응답한다. 저장이 실패하면 그 묶음의 읽기까지 모두 실패한다. 저장되지
+  않은 상태를 본 응답은 나가지 않는다. 같은 키가 다시 오면 묶음을 먼저 끝낸다. 모듈 서버는 기다리는 연결을 한 번에
+  모두 받아 동시 요청이 같은 묶음에 들어가게 한다.
+- GET 응답은 코어 예산에 계상되는 자원으로 복사한다. PUT 응답은 코어 계약대로 요청 본문에서 빌려야 하므로 `edit`
+  핸들러는 메시지 필드만 응답할 수 있다(`reply edit.insert`). 시작할 때 확인한다.
+- 검증(aup `make check`): `editor.aup`를 실행하는 모듈 서버와 `editor.pure`를 실행하는 기존 서버가 무작위 요청
+  (새 키, 오래된 리비전, 범위 밖, 짧은 본문, 5,000 B 삽입)에 상태 코드·ETag·오류·본문까지 같게 답한다. 동시 편집 중
+  `kill -9`로 죽여도 확인 응답을 받은 편집은 다시 시작한 서버에 모두 있다. Pi 5 SD(ext4)에서 동시 클라이언트 8개의
+  PUT 240건: 기존 서버 2.14 s, 모듈 서버 0.59 s(묶음 61개).
+
+아직 모듈 앱을 런타임 코어의 효과 단위로 세분하지는 않았다(`runtime.c`의 `definition`은 앱 전체다). 호스트가 직접 호출한다. 실행기 코드는
 `module.c` 33.1 KB, `module-text.c` 3.6 KB, `module-list.c` 5.7 KB, `module-save.c` 5.3 KB(`-O2` text)로 v1(7.4 KB)보다 크다. 줄이는 것이
 남은 일이다. 항목 text를 읽을 때마다 복사하는 비용도 남아 있다.
diff --git a/prototypes/connections/Makefile b/prototypes/connections/Makefile
index e78fdbd..83af085 100644
--- a/prototypes/connections/Makefile
+++ b/prototypes/connections/Makefile
@@ -40,6 +40,10 @@ $(BUILD)/pure-connections-async: server.c $(HOST) flow.c program.c $(CORE) $(HEA
 $(BUILD)/pure-connections-async-gated: server.c $(HOST) flow.c program.c $(CORE) test-async-server-gate.c $(HEADERS) | $(BUILD)
 	$(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_ASYNC_EFFECTS -DPURE_HTTP_ASYNC -DPURE_STORE_TEST server.c $(HOST) flow.c program.c $(CORE) test-async-server-gate.c $(LDFLAGS) -pthread $(LDLIBS) -o $@
 
+# The same HTTP host running an aup module (show and edit) with group commits.
+$(BUILD)/pure-connections-module: server.c flow.c module-store.c $(MODULE) $(CORE) module-store.h $(MODULE_HEADERS) | $(BUILD)
+	$(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_ASYNC_EFFECTS -DPURE_MODULE_SERVER server.c flow.c module-store.c $(MODULE) $(CORE) $(LDFLAGS) -lm $(LDLIBS) -o $@
+
 $(BUILD)/pure-connections-fixed: server.c $(HOST) flow.c $(CORE) $(HEADERS) $(BUILD)/embedded-plan.h | $(BUILD)
 	$(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_EMBEDDED_PROGRAM -I$(BUILD) server.c $(HOST) flow.c $(CORE) $(LDFLAGS) $(LDLIBS) -o $@
 
@@ -130,7 +134,7 @@ $(BUILD)/pure-run: pure-run.c module-store.c $(MODULE) module-store.h $(MODULE_H
 $(BUILD)/pure-edit-module: edit-module.c module-store.c $(MODULE) module-store.h $(MODULE_HEADERS) | $(BUILD)
 	$(CC) $(CPPFLAGS) $(CFLAGS) edit-module.c module-store.c $(MODULE) $(LDFLAGS) -lm $(LDLIBS) -o $@
 
-check-module: $(BUILD)/test-module $(BUILD)/pure-run $(BUILD)/pure-edit-module
+check-module: $(BUILD)/test-module $(BUILD)/pure-run $(BUILD)/pure-edit-module $(BUILD)/pure-connections-module
 	$(BUILD)/test-module
 
 $(BUILD)/test-tree-store: test-tree-store.c store-tree.c $(HEADERS) | $(BUILD)
diff --git a/prototypes/connections/server.c b/prototypes/connections/server.c
index 505bb71..448a915 100644
--- a/prototypes/connections/server.c
+++ b/prototypes/connections/server.c
@@ -2,6 +2,9 @@
 #include "store.h"
 #include "program.h"
 #include "store-port.h"
+#ifdef PURE_MODULE_SERVER
+#include "module-store.h"
+#endif
 #ifdef PURE_EMBEDDED_PROGRAM
 #include "embedded-plan.h"
 #endif
@@ -27,6 +30,14 @@
 #if defined(PURE_HTTP_ASYNC) && !defined(PURE_ASYNC_EFFECTS)
 #error PURE_HTTP_ASYNC requires PURE_ASYNC_EFFECTS
 #endif
+/* A module server runs an aup module instead of a document program: GET is
+ * its show message, PUT its edit message. Effects are taken on the owner
+ * thread, applied with deferred saving, saved once per burst and only then
+ * completed, so concurrent requests share one commit. */
+#if defined(PURE_MODULE_SERVER) && (!defined(PURE_ASYNC_EFFECTS) || defined(PURE_HTTP_ASYNC) || \
+                                     defined(PURE_EMBEDDED_PROGRAM))
+#error PURE_MODULE_SERVER takes effects on the owner thread and has no program
+#endif
 
 enum { SLOTS = 8, HEADER_LIMIT = 4096, CHUNK = 65536 };
 typedef enum { READING_HEADER, READING_BODY, WRITING, WAITING_EFFECT } Phase;
@@ -58,7 +69,12 @@ typedef struct {
 } Connection;
 struct Server {
     const char *path;
-#ifndef PURE_EMBEDDED_PROGRAM
+#ifdef PURE_MODULE_SERVER
+    PureModule module;
+    PureApp app;
+    uint32_t show, edit;
+    uint64_t saves, burst_events; /* saves is the number of bursts */
+#elif !defined(PURE_EMBEDDED_PROGRAM)
     PureProgram program;
 #endif
     PureRuntime runtime;
@@ -252,8 +268,136 @@ static void finish_effects(Server *server) {
 }
 #endif
 
+#ifdef PURE_MODULE_SERVER
+static void release_copy(void *allocation, size_t size) {
+    (void)size;
+    free(allocation);
+}
+
+static bool allocate_ingress(void *context, size_t size, PureResource *out) {
+    (void)context;
+    void *bytes = malloc(size);
+    if (!bytes) return false;
+    *out = (PureResource){.body = {bytes, size}, .allocation = bytes, .allocation_size = size, .release = release_copy};
+    return true;
+}
+
+typedef struct { unsigned char *bytes; size_t size; const unsigned char *first; size_t spans; } Reply;
+
+static bool measure_reply(void *context, PureSpan span) {
+    Reply *r = context;
+    if (!r->spans++) r->first = span.data;
+    r->size += span.size;
+    return true;
+}
+
+static bool copy_reply(void *context, PureSpan span) {
+    Reply *r = context;
+    if (span.size) memcpy(r->bytes + r->size, span.data, span.size);
+    r->size += span.size;
+    return true;
+}
+
+/* One burst of requests: applied in order, saved with one commit, then
+ * completed. If the save fails every request of the burst fails, reads too:
+ * they saw changes that were never stored. */
+typedef struct {
+    PureTicket ticket;
+    PureInput input;
+    PureResult status;
+    int64_t revision;
+    PureResource resource;
+    PureSpan reply;
+    char key[65];
+} Applied;
+
+static void apply_effect(Server *server, const PureEffect *effect, Applied *out) {
+    PureApp *app = &server->app;
+    *out = (Applied){.ticket = effect->ticket, .input = effect->input};
+    memcpy(out->key, effect->key, strlen(effect->key) + 1);
+    bool read = effect->input == PURE_INPUT_READ;
+    out->status = pure_app_message(app, read ? server->show : server->edit, effect->key,
+                                   read ? -1 : effect->expected, read ? (PureSpan){0} : effect->value);
+    out->revision = app->revision;
+    if (out->status != PURE_OK) return;
+    ++server->burst_events;
+    Reply reply = {0};
+    pure_app_reply(app, measure_reply, &reply);
+    if (!read) {
+        /* The module was checked at start: an edit replies part of its request. */
+        out->reply = (PureSpan){reply.size ? reply.first : NULL, reply.size};
+        return;
+    }
+    if (reply.size > effect->available) { out->status = PURE_BUDGET; return; }
+    if (!reply.size) return;
+    unsigned char *copy = malloc(reply.size);
+    if (!copy) { out->status = PURE_ALLOCATION; return; }
+    reply = (Reply){.bytes = copy};
+    pure_app_reply(app, copy_reply, &reply);
+    out->resource = (PureResource){.body = {copy, reply.size}, .allocation = copy,
+                                   .allocation_size = reply.size, .release = release_copy};
+}
+
+static void finish_burst(Server *server, Applied *applied, size_t count) {
+    if (!count) return;
+    PureResult saved = pure_app_save(&server->app);
+    ++server->saves;
+    for (size_t i = 0; i < count; ++i) {
+        Applied *a = &applied[i];
+        PureResult status = a->status == PURE_OK ? saved : a->status;
+        if (!pure_runtime_complete(a->ticket, status, a->input == PURE_INPUT_READ ? &a->resource : NULL,
+                                   status == PURE_OK ? a->reply : (PureSpan){0}, a->revision)) abort();
+        for (size_t j = 0; j < SLOTS; ++j) {
+            Connection *c = &server->connections[j];
+            if (c->fd < 0 || c->phase != WAITING_EFFECT ||
+                c->ticket.slot != a->ticket.slot || c->ticket.generation != a->ticket.generation) continue;
+            PureOutcome result;
+            if (!pure_runtime_result(a->ticket, &result)) abort();
+            c->body = NULL;
+            answer(c, result);
+            break;
+        }
+    }
+}
+
+/* Every request that became ready since the last burst; a key seen again
+ * ends the burst first, so a document's revision moves once per save. */
+static void serve_bursts(Server *server) {
+    Applied applied[SLOTS];
+    size_t count = 0;
+    PureEffect effect;
+    while (pure_runtime_take(&server->runtime, &effect) == PURE_TAKE_EFFECT) {
+        bool repeated = false;
+        for (size_t i = 0; i < count && !repeated; ++i) repeated = !strcmp(applied[i].key, effect.key);
+        if (repeated) {
+            finish_burst(server, applied, count);
+            count = 0;
+        }
+        apply_effect(server, &effect, &applied[count++]);
+    }
+    finish_burst(server, applied, count);
+}
+
+/* PUT replies must be borrowed from the request, so an edit handler may
+ * reply only a field of its message. */
+static bool edit_replies_borrow(const PureModule *m, uint32_t edit) {
+    for (uint32_t h = 0; h < m->handlers; ++h) {
+        if (m->handler[h].event != 1 + 2 * edit) 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;
+    }
+    return true;
+}
+#endif
+
 static void run_request(Connection *c) {
     if (c->put && !pure_runtime_ready(c->ticket)) abort();
+#ifdef PURE_MODULE_SERVER
+    c->phase = WAITING_EFFECT; /* the burst after this poll round serves it */
+    return;
+#endif
 #ifdef PURE_HTTP_ASYNC
     Server *server = c->server;
     PureEffect effect;
@@ -411,7 +555,10 @@ static bool close_on_exec(int fd) {
 int main(int argc, char **argv) {
     Server server = {.budget = 4u * 1024u * 1024u, .timeout_ms = 10000};
     unsigned port = 8767;
-#ifndef PURE_EMBEDDED_PROGRAM
+#ifdef PURE_MODULE_SERVER
+    const char *module_path = NULL;
+    bool explain = false;
+#elif !defined(PURE_EMBEDDED_PROGRAM)
     const char *program_path = NULL;
     bool explain = argc == 3 && !strcmp(argv[1], "--check-program");
     char program_error[160];
@@ -423,7 +570,9 @@ int main(int argc, char **argv) {
         if (i + 1 == argc) goto usage;
         uint64_t value;
         if (!strcmp(argv[i], "--store")) server.path = argv[i + 1];
-#ifndef PURE_EMBEDDED_PROGRAM
+#ifdef PURE_MODULE_SERVER
+        else if (!strcmp(argv[i], "--module")) module_path = argv[i + 1];
+#elif !defined(PURE_EMBEDDED_PROGRAM)
         else if (!strcmp(argv[i], "--program")) program_path = argv[i + 1];
 #endif
         else if (!strcmp(argv[i], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value;
@@ -433,7 +582,30 @@ int main(int argc, char **argv) {
         else goto usage;
     }
     if (!server.path && !explain) goto usage;
-#ifndef PURE_EMBEDDED_PROGRAM
+#ifdef PURE_MODULE_SERVER
+    if (!module_path) goto usage;
+    {
+        static unsigned char bytes[1 << 17];
+        char error[128];
+        FILE *file = fopen(module_path, "rb");
+        size_t size = file ? fread(bytes, 1, sizeof(bytes), file) : 0;
+        bool read_ok = file && !ferror(file) && size < sizeof(bytes);
+        if (file) fclose(file);
+        if (!read_ok || pure_module_load(bytes, size, &server.module, error, sizeof(error)) != PURE_OK) {
+            fprintf(stderr, "Module: %s\n", read_ok ? error : "cannot read"); return 2;
+        }
+        int show = pure_module_input(&server.module, "show"), edit = pure_module_input(&server.module, "edit");
+        if (show < 0 || edit < 0 || server.module.input[show].kind != PURE_KIND_MESSAGE || !server.module.input[show].keyed ||
+            server.module.input[edit].kind != PURE_KIND_MESSAGE || !server.module.input[edit].keyed) {
+            fputs("Module: needs keyed messages named show and edit\n", stderr); return 2;
+        }
+        if (!edit_replies_borrow(&server.module, (uint32_t)edit)) {
+            fputs("Module: edit may reply only a field of its message\n", stderr); return 2;
+        }
+        server.show = (uint32_t)show;
+        server.edit = (uint32_t)edit;
+    }
+#elif !defined(PURE_EMBEDDED_PROGRAM)
     if (!program_path) goto usage;
     if (!pure_program_load(program_path, &server.program, program_error, sizeof(program_error))) {
         fprintf(stderr, "Program: %s\n", program_error); return 2;
@@ -464,6 +636,13 @@ int main(int argc, char **argv) {
         return 0;
     }
 #endif
+#ifdef PURE_MODULE_SERVER
+    const void *program = &server.app;
+    if (pure_store_init(server.path) != PURE_OK) { fputs("Cannot initialize byte store directory\n", stderr); return 1; }
+    if (pure_app_start(&server.app, &server.module, pure_tree_data(server.path)) != PURE_OK ||
+        !pure_app_defer(&server.app, true)) { fputs("Cannot start the module on the store\n", stderr); return 1; }
+    PureHost host = {.allocate = allocate_ingress};
+#else
     const PureProgram *program =
 #ifdef PURE_EMBEDDED_PROGRAM
         &pure_embedded_program;
@@ -472,6 +651,7 @@ int main(int argc, char **argv) {
 #endif
     if (pure_store_init(server.path) != PURE_OK) { fputs("Cannot initialize byte store directory\n", stderr); return 1; }
     PureHost host = pure_store_host(server.path);
+#endif
 #ifdef PURE_HTTP_ASYNC
     server.effect_host = host;
     host.read = NULL;
@@ -479,7 +659,7 @@ int main(int argc, char **argv) {
 #endif
     if (!pure_runtime_init(&server.runtime, server.work, SLOTS, program,
                            host, server.budget)) return 1;
-#ifdef PURE_HTTP_ASYNC
+#if defined(PURE_HTTP_ASYNC) || defined(PURE_MODULE_SERVER)
     if (!pure_runtime_enable_async(&server.runtime)) abort();
 #endif
     signal(SIGPIPE, SIG_IGN);
@@ -589,8 +769,19 @@ int main(int argc, char **argv) {
                 if (c->phase == WRITING) write_connection(c); else read_connection(c);
             }
         }
+#ifdef PURE_MODULE_SERVER
+        serve_bursts(&server);
+#endif
+        /* A module server takes every waiting connection at once, so their
+         * requests can share the next burst. */
+#ifdef PURE_MODULE_SERVER
+        for (bool more = true; more && !stopping && (polls[0].revents & POLLIN) && server.active < SLOTS;) {
+#else
         if (!stopping && (polls[0].revents & POLLIN)) {
+            bool more;
+#endif
             int fd = accept(listener, NULL, NULL);
+            more = fd >= 0;
             if (fd < 0) continue;
             if (!nonblocking(fd)) { close(fd); continue; }
             if (server.send_buffer) setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &server.send_buffer, sizeof(server.send_buffer));
@@ -601,6 +792,7 @@ int main(int argc, char **argv) {
                 if (server.active > server.peak_active) server.peak_active = server.active;
                 break;
             }
+            (void)more;
         }
     }
     close(listener);
@@ -611,6 +803,10 @@ int main(int argc, char **argv) {
 #endif
     pure_runtime_stop(&server.runtime);
     for (size_t i = 0; i < SLOTS; ++i) close_connection(&server.connections[i]);
+#ifdef PURE_MODULE_SERVER
+    fprintf(stderr, "{\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 "}\n", server.saves, server.burst_events);
+    pure_app_stop(&server.app);
+#endif
 #ifdef PURE_HTTP_ASYNC
     pthread_mutex_lock(&server.job_mutex);
     server.worker_stop = true;
@@ -635,7 +831,10 @@ int main(int argc, char **argv) {
         server.expired, server.runtime.releases, server.prefix_copy_bytes, server.runtime.borrowed_output_bytes, sizeof(server));
     return 0;
 usage:
-#ifdef PURE_EMBEDDED_PROGRAM
+#ifdef PURE_MODULE_SERVER
+    fputs("pure-connections-module --module FILE.pmod --store DIRECTORY [--port 8767] [--budget BYTES] "
+          "[--timeout-ms 10000] [--send-buffer BYTES]\n", stderr);
+#elif defined(PURE_EMBEDDED_PROGRAM)
     fputs("pure-connections-fixed --store DIRECTORY [--port 8767] [--budget BYTES] [--timeout-ms 10000] [--send-buffer BYTES]\n", stderr);
 #else
     fputs("pure-connections --program FILE --store DIRECTORY [--port 8767] [--budget BYTES] [--timeout-ms 10000] [--send-buffer BYTES]\n"