From c08c570d465ffa22c772c2c4d0910eb8a17a379e Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sun, 27 Sep 2026 13:48:55 +0100 Subject: Check pure-serve in Pure, not in aup The HTTP host's end-to-end checks (documents against the earlier server, lending, kill -9 durability, JSON, timers, links, webhooks and answers) sat in aup's tests, so a change to the host was verified from the toolchain. They now live in tests/check.py; make check builds their apps with the sibling aup and runs them after the unit tests, which stay beside their parts. Design 009 states the rule for all three repositories. --- Makefile | 8 + README.md | 4 +- docs/design/008-module-format.md | 2 +- docs/design/009-layout.md | 5 + docs/research/077-apps-as-http-services.md | 2 +- docs/research/079-lent-views.md | 2 +- docs/research/080-snapshot-reads.md | 2 +- docs/research/082-json-in-place.md | 2 +- docs/research/083-delivery-over-http.md | 6 +- tests/check.py | 1231 ++++++++++++++++++++++++++++ 10 files changed, 1255 insertions(+), 9 deletions(-) create mode 100644 tests/check.py diff --git a/Makefile b/Makefile index aed243e..d9033a1 100644 --- a/Makefile +++ b/Makefile @@ -14,6 +14,8 @@ BUILD ?= build # cli/ command-line hosts. # libpure.a is the app model with the tree store: what a host embeds (Vega's # surfaces link it). prototypes/ keeps the earlier document runtime (000-077). +# Unit tests sit by their part (core/test-module.c, store/test-tree-store.c); +# tests/check.py checks the hosts end to end. INCLUDES = -Icore MODEL = core/module.c core/module-list.c core/module-save.c core/module-text.c core/module-store.c core/flow.c STORE = store/store-tree.c @@ -54,6 +56,12 @@ $(BUILD)/test-tree-store: store/test-tree-store.c $(STORE) $(HEADERS) @mkdir -p $(BUILD) $(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -DPURE_STORE_TEST store/test-tree-store.c $(STORE) $(LDFLAGS) $(LDLIBS) -o $@ +# The HTTP host's checks build their apps with aup (the sibling checkout) and +# compare documents with the earlier document server. +AUP ?= ../aup check: all $(BUILD)/test-module $(BUILD)/test-tree-store $(BUILD)/test-tree-store $(BUILD)/test-module + $(MAKE) -C $(AUP) build/aup + $(MAKE) -C prototypes/connections BUILD=build/tree STORE=store-tree.c build/tree/pure-connections + python3 tests/check.py --aup $(AUP)/build/aup --aup-root $(AUP) diff --git a/README.md b/README.md index c7c7958..681c230 100644 --- a/README.md +++ b/README.md @@ -23,8 +23,10 @@ The source follows the principle below, a small core with the rest selected arou | `serve/` | `pure-serve`: the HTTP host, outbound delivery, JSON read in place | `build/pure-serve` | | `cli/` | command-line hosts | `build/pure-run`, `build/pure-edit-module` | | `prototypes/` | the earlier document runtime (000-077), built against `core/` and `store/` | `make -C prototypes/connections` | +| `tests/` | the hosts checked end to end, with apps built by aup (the sibling checkout) | `make check` | -`make` builds the library and the hosts, `make check` runs the store's and the app model's tests. Interfaces for +`make` builds the library and the hosts; `make check` runs the store's and the app model's unit tests (beside +their parts) and then `tests/check.py`. Interfaces for people are [Vega](/srv/git/vega.git), which links `libpure.a`. The text below records the earlier document runtime this grew from. diff --git a/docs/design/008-module-format.md b/docs/design/008-module-format.md index 84aa97b..b70b8f0 100644 --- a/docs/design/008-module-format.md +++ b/docs/design/008-module-format.md @@ -358,7 +358,7 @@ text·리스트가 남을 때만 위치를 유지한다. 응답만 복사한다. PUT 응답은 요청 본문에서 빌려야 하므로 `edit` 핸들러는 메시지 필드만 응답할 수 있다(`reply edit.insert`). 시작할 때 확인한다. -성능과 비교는 [연구 075](../research/075-http-host-against-axum.md). 검증(aup `make check`): 무작위 요청에 기존 서버와 +성능과 비교는 [연구 075](../research/075-http-host-against-axum.md). 검증(Pure `make check`의 `tests/check.py`, 앱은 aup가 빌드): 무작위 요청에 기존 서버와 상태 코드·ETag·오류·본문이 같고, 동시 편집이 커밋을 나누어 쓰며, `kill -9` 뒤에도 확인 응답한 편집이 모두 남는다. 앱 라우트는 키 있는 리스트 앱에 무작위 POST(쿼리 필드와 본문을 가진 메시지, text, 키)와 GET(수, text, 표의 쪽)을 800번 보내 상태 코드·오류·본문·JSON 행이 파이썬 모델과 같고, 멈췄다 다시 띄워도, 여섯 클라이언트가 보내는 중에 diff --git a/docs/design/009-layout.md b/docs/design/009-layout.md index 17d6a48..bcaf64e 100644 --- a/docs/design/009-layout.md +++ b/docs/design/009-layout.md @@ -33,6 +33,11 @@ | `cli/` | 명령행 호스트 | `build/pure-run`, `build/pure-edit-module` | | `prototypes/` | 이전 문서 런타임. `core/`와 `store/`를 가져다 빌드한다 | `make -C prototypes/connections` | +검사도 부품을 따른다. 부품의 단위 검사는 그 옆에(`core/test-module.c`, `store/test-tree-store.c`), 저장소 전체를 끝에서 +끝까지 보는 검사는 `tests/check.py`에 둔다. aup와 Vega도 같다(aup `compiler/`, Vega `lib/`·`surfaces/`·`symbols/`, 둘 다 +`tests/check.py`). pure-serve의 HTTP 검사는 전에 aup의 검사 안에 있었다. 호스트를 바꾸면 aup에서 확인해야 했던 것이다. +이제 Pure의 `make check`가 aup로 앱을 빌드해 돌린다(Vega가 aup와 pure-serve를 쓰는 것과 같은 방향이다). + 저장소 계약(`state.h`, `store.h`)은 코어 쪽에 둔다. 코어가 필요한 것을 정하고 저장소 구현이 그것을 채운다. 트리 저장소는 그 계약의 한 구현이다(이전 세대의 `store.c`, `store-paged.c`도 `store.h`의 구현이다). diff --git a/docs/research/077-apps-as-http-services.md b/docs/research/077-apps-as-http-services.md index 27e69fd..63904a5 100644 --- a/docs/research/077-apps-as-http-services.md +++ b/docs/research/077-apps-as-http-services.md @@ -68,7 +68,7 @@ Raspberry Pi 5, 서버는 CPU 0–2, 부하는 CPU 3, 저장소는 SD 카드의 ## 검증 -aup `make check`: 키 있는 리스트 앱에 무작위 요청 800개(쿼리 필드와 본문을 가진 메시지, text, 키 누름, 수·text 보기, 표의 +aup `make check`(지금은 Pure `tests/check.py`의 `server_app`): 키 있는 리스트 앱에 무작위 요청 800개(쿼리 필드와 본문을 가진 메시지, text, 키 누름, 수·text 보기, 표의 쪽, 없는 이름과 잘못된 쿼리)를 보내 상태 코드·오류 이름·본문·JSON 행이 파이썬 모델과 같다. 한 스레드와 연결 스레드 셋 + 소유자에서 모두 확인한다. 여섯 클라이언트가 보내는 중에 `kill -9`로 죽여도, 확인 응답한 이벤트(각 3,500–4,200개)가 다시 시작한 뒤 모두 남는다. ASan·UBSan 빌드에서 같은 검사와, 클라이언트 열둘이 쓰기·읽기·실패를 섞어 보내는 부하(요청 3,600개)를 diff --git a/docs/research/079-lent-views.md b/docs/research/079-lent-views.md index 1cb8453..668a832 100644 --- a/docs/research/079-lent-views.md +++ b/docs/research/079-lent-views.md @@ -38,7 +38,7 @@ Raspberry Pi 5, 서버는 CPU 0–2, 부하는 CPU 3. 1 MiB 메모를 `GET /view ## 회수 확인 -aup `make check`의 `server_lent_views`(스레드 1, 연결 스레드 3 + 소유자): +aup `make check`의 `server_lent_views`(지금은 Pure `tests/check.py`)(스레드 1, 연결 스레드 3 + 소유자): - 한 연결에 1 MiB 보기 32개를 이어 보내면 커널 버퍼가 받지 못한 답이 빌린 채 기다린다. 그동안 메모를 다섯 번 바꿔도 32개 답은 각각 한 버전 전체로 나온다(첫 답은 바꾸기 전 버전). diff --git a/docs/research/080-snapshot-reads.md b/docs/research/080-snapshot-reads.md index 1300632..82bf7f7 100644 --- a/docs/research/080-snapshot-reads.md +++ b/docs/research/080-snapshot-reads.md @@ -117,7 +117,7 @@ Raspberry Pi 5, 서버는 CPU 0–2, 부하는 CPU 3, 저장소는 SD 카드의 저장을 기다리는 변경이 있으면 스냅숏이 없고, 실패한 저장 뒤의 스냅숏은 실패 전과 같다. 상태를 메모리에만 두는 앱은 스냅숏을 만들지 않는다. 두 스레드가 최신 스냅숏을 읽는 동안(2만–4만 번) 앱이 계속 적용하고 게시한다. ASan·UBSan· LeakSanitizer 빌드에서 같다. -- aup `make check`: 무작위 요청 800개의 파이썬 모델 대조가 연결 스레드 하나와 셋에서 같다. 여섯 클라이언트가 쓰는 동안 +- aup `make check`(지금은 Pure `tests/check.py`의 `server_app`): 무작위 요청 800개의 파이썬 모델 대조가 연결 스레드 하나와 셋에서 같다. 여섯 클라이언트가 쓰는 동안 `kill -9`하는 검사에 다른 연결의 읽기 둘을 더했다. 읽기마다 요청 전에 확인 응답된 점수를 모두 보고, 합계가 뒤로 가지 않으며, 다시 띄운 서버의 합계는 읽기가 본 어떤 합계 이상이다(저장되지 않은 것을 보지 않았다). 빌려준 보기의 회수(079)도 그대로다. `pure-serve`의 ASan·UBSan 빌드로 같은 서버 검사를 통과하고, 섞인 부하 뒤 정상 종료에서 누수가 diff --git a/docs/research/082-json-in-place.md b/docs/research/082-json-in-place.md index 6a456b8..aab6afb 100644 --- a/docs/research/082-json-in-place.md +++ b/docs/research/082-json-in-place.md @@ -61,7 +61,7 @@ Raspberry Pi 5, 서버는 CPU 0–2(연결 스레드 하나), 부하는 CPU 3. 시작하거나 끝나는 자리는 `limit`, UTF-8이 아닌 자리는 `encoding`으로 실패하고 아무것도 바꾸지 않는다. 경로 이름 `user.name`은 받고, `user..name`, `.user`, `user.`, `user.1st`, `user name`, 그리고 `user`와 `user.name`을 함께 가진 모듈은 거절한다. -- aup `make check`의 `server_json`: 전의 무작위 평평한 JSON 300개와 틀린 요청 14가지에 더해, 그룹이 두 단계인 메시지에 +- aup `make check`의 `server_json`(지금은 Pure `tests/check.py`): 전의 무작위 평평한 JSON 300개와 틀린 요청 14가지에 더해, 그룹이 두 단계인 메시지에 무작위 JSON 100개(멤버 순서를 단계마다 섞고, 문자열에 이스케이프·제어 바이트·한글·이모지·U+2028)를 보내 답과 표가 파이썬 모델과 같다. 그룹의 고정 필드를 쿼리 경로로 준 요청, 틀린 그룹 요청 10가지(모르는 멤버, 그룹 자리에 값, 필드 자리에 객체, 빈 그룹, 두 번 온 필드, 배열, `null`, 외톨이 서로게이트, 닫히지 않은 객체), 쿼리와 본문에 함께 준 필드도 diff --git a/docs/research/083-delivery-over-http.md b/docs/research/083-delivery-over-http.md index eedc5f8..5f51b2b 100644 --- a/docs/research/083-delivery-over-http.md +++ b/docs/research/083-delivery-over-http.md @@ -74,14 +74,14 @@ Raspberry Pi 5. 받는 서버는 CPU 0–1(연결 스레드 2), 보내는 서버 ## 검증 -- aup `make check`의 `server_links`: 클라이언트 넷이 보내는 앱에 주문하는 동안 받는 서버를, 그다음 보내는 서버를 +- aup `make check`의 `server_links`(지금은 Pure `tests/check.py`): 클라이언트 넷이 보내는 앱에 주문하는 동안 받는 서버를, 그다음 보내는 서버를 `kill -9`로 죽이고 같은 저장소로 다시 띄운다. 그 뒤 수량 150(받는 앱이 거절한다)과 7의 주문을 더 넣는다. 받는 앱의 건수와 합계가 보내는 앱이 저장한 것(거절한 한 건을 빼고)과 같아지고, 확인 응답한 주문은 모두 그 안에 있다. 다섯 번(각 약 7,300건)과 ASan·UBSan 빌드로 한 번 통과했다. 거절한 주문 뒤의 주문도 전해졌다(막히지 않는다). -- aup `make check`의 `server_webhook`: 파이썬 HTTPS 서버(자체 인증서)를 제3자 웹훅으로 두고, 주문 다섯 건이 순서대로 +- aup `make check`의 `server_webhook`(지금은 Pure `tests/check.py`): 파이썬 HTTPS 서버(자체 인증서)를 제3자 웹훅으로 두고, 주문 다섯 건이 순서대로 출력의 JSON 모양과 헤더(`X-Pure-Source`, `Idempotency-Key`, 묶음 헤더 없음)로 도착한다. 처음에 503으로 답한 2번은 다시 온다. 그 인증서를 믿지 않는 보내는 쪽에서는 아무것도 도착하지 않고 기록이 남는다. ASan·UBSan 빌드로도 통과했다. -- aup `make check`의 `server_answers`: 파이썬 "결제 API"(1000 이하는 200 `succeeded`, 넘으면 402 `declined`, 같은 +- aup `make check`의 `server_answers`(지금은 Pure `tests/check.py`): 파이썬 "결제 API"(1000 이하는 200 `succeeded`, 넘으면 402 `declined`, 같은 `Idempotency-Key`에는 첫 답을 되풀이)에 출력을 잇고 응답을 `charged` 입력으로 받는다. 클라이언트 셋이 사는 동안 보내는 서버를 `kill -9`로 죽였다 살린다. 끝에 앱의 결제·거절 수의 합이 저장된 구매 수와 API가 받은 키 수와 같고, 결제 수와 결제 합계가 API의 200 응답과 같다(각 3,774–3,979건, 네 번, ASan·UBSan 빌드로 한 번 더). diff --git a/tests/check.py b/tests/check.py new file mode 100644 index 0000000..8352578 --- /dev/null +++ b/tests/check.py @@ -0,0 +1,1231 @@ +#!/usr/bin/env python3 +"""Pure's end-to-end checks: pure-serve, the HTTP host, running modules that +aup builds (the sibling checkout, --aup-root). The core's and the store's own +tests are core/test-module.c and store/test-tree-store.c. + +1. Documents: pure-serve running the document editor written in aup + (aup's examples/editor.aup) answers random requests exactly as the earlier + document server running editor.pure, with one and with three connection + threads, on persistent and pipelined connections too. +2. Lending: a read lends the stored extents and still carries its version + while the document changes; lent answers count against the payload budget; + a text view is lent while its answer waits and let go when it is read or + its connection closes, the store keeping one mapping again. +3. Durability: concurrent edits share commits, and every acknowledged edit + survives kill -9, with one and with three threads; so do the app's + acknowledged posts (keyed lists, JSON bodies with groups). +4. JSON: request bodies are read in place, fields by path, and malformed or + mistyped ones are refused with the field named. +5. Timers run on the owner and their beats are saved like events. +6. Links: sends are saved with their event and delivered in order and in + batches to another app, which applies each once, across kill -9 of either + side; webhooks receive them over HTTPS only from a trusted receiver; an + API's answers become the app's events (--reply). +""" +import argparse +import json +import random +import subprocess +import sys +import tempfile +from pathlib import Path + +HERE = Path(__file__).resolve().parent +ROOT = HERE.parent +# The app that exercises documents: the editor written in aup, from aup's examples. +EDITOR = ROOT.parent / 'aup' / 'examples' / 'editor.aup' + + +def run(*args, stdin=None, check=True): + return subprocess.run([str(a) for a in args], input=stdin, capture_output=True, text=True, + check=check, timeout=60) + + +def compile_module(aup, source_path, out): + run(aup, 'build', source_path, '-o', out) + return out.read_bytes() + + +class HttpHost: + """A Pure HTTP host on a fresh port; stops with its statistics.""" + + def __init__(self, args): + import select + self.process = subprocess.Popen([str(a) for a in args] + ['--port', '0'], stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + ready, _, _ = select.select([self.process.stdout], [], [], 10) + assert ready, 'HTTP host did not start' + line = self.process.stdout.readline() + assert line, self.process.communicate(timeout=5) + self.port = json.loads(line)['port'] + + def request(self, method, key, body=b'', expected=None): + import http.client + headers = {'Content-Length': str(len(body))} + if expected is not None: + headers['If-None-Match' if expected == 0 else 'If-Match'] = '*' if expected == 0 else f'"{expected}"' + connection = http.client.HTTPConnection('127.0.0.1', self.port, timeout=30) + try: + connection.request(method, f'/objects/{key}', body, headers) + response = connection.getresponse() + return response.status, response.getheader('ETag'), response.getheader('X-Pure-Error'), response.read() + finally: + connection.close() + + def stop(self): + self.process.terminate() + _, error = self.process.communicate(timeout=30) + assert self.process.returncode == 0, error[-1000:] + return [json.loads(line) for line in error.decode().splitlines() if line.startswith('{')] + + +def server_parity(aup, pure, work, seeds, operations=150, threads=1): + """The document server running editor.aup answers every request exactly + as the server running editor.pure: status, revision, error and body.""" + import struct + module = work / 'editor-server.pmod' + compile_module(aup, EDITOR, module) + compared = 0 + for seed in seeds: + rng = random.Random(seed + 13000) + old = HttpHost([pure / 'prototypes/connections/build/tree/pure-connections', '--program', pure / 'prototypes/connections/examples/editor.pure', + '--store', work / f'http-old-{seed}-{threads}']) + new = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / f'http-new-{seed}-{threads}', + '--threads', str(threads)]) + try: + revision, size = {}, {} + for _ in range(operations): + key = rng.choice(['a', 'b', 'long-key_1']) + current, length = revision.get(key, 0), size.get(key, 0) + if rng.random() < 0.35: + request = ('GET', key, b'', None) + else: + pick = rng.random() + base = current if pick < 0.75 else max(current - 1, 0) if pick < 0.9 else current + 3 + if rng.random() < 0.05: + body = bytes(rng.randrange(256) for _ in range(rng.randint(0, 7))) # too short + else: + insert = bytes(rng.randrange(256) for _ in range(rng.choice([0, 1, 7, 300, 5000]))) + body = struct.pack(' alone and held_again > alone and after_reading <= alone and after_leaving <= alone, \ + (alone, held, after_reading, held_again, after_leaving) + assert http(host.port, 'GET', '/views/page')[3] == b'c' * size + finally: + stats = host.stop() + assert stats[0]['active'] == 0 and stats[0]['retained_payload_bytes'] == 0, stats + return {'alone': alone, 'while_lent': max(held, held_again), 'after': after_leaving} + + +def server_bursts(aup, pure, work, clients=8, edits=30, shards=1): + """Concurrent edits share commits, and every acknowledged edit survives a + kill -9 of the server.""" + import http.client + import signal + import struct + import threading + import time + module = work / 'editor-bursts.pmod' + compile_module(aup, EDITOR, module) + + def load(host, rounds, acknowledged): + def client(number): + key, base = f'c{number}', acknowledged[number][1] + for _ in range(rounds): + try: + status, tag, _, _ = host.request('PUT', key, struct.pack(' deadline: + break + time.sleep(0.1) + assert (count, total) == (sent - 1, sent_total - 150), (count, total, sent, sent_total) + assert sent_total >= sum(acknowledged) + 157 and sum(acknowledged) > 0, (sent_total, acknowledged) + finally: + stats = {} + for name, process in hosts.items(): + process.terminate() + _, error = process.communicate(timeout=30) + assert process.returncode == 0, error[-1000:] + stats[name] = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1] + return {'orders': sent, 'acknowledged_total': sum(acknowledged), 'delivered_after_restart': stats['a']['delivered'], + 'attempts_after_restart': stats['a']['delivery_attempts']} + + +def server_webhook(aup, pure, work): + """An output wired to a third-party HTTPS endpoint (a Python server with + a certificate of its own): five orders arrive in order as JSON with the + sender's headers, one answered 503 at first arrives again, and none goes + to a receiver whose certificate the sender does not trust.""" + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + import ssl + import threading + import time + module = work / 'hook-sender.pmod' + module.with_suffix('.aup').write_text(LINK_SENDER) + compile_module(aup, module.with_suffix('.aup'), module) + for name in ('trusted', 'other'): + subprocess.run(['openssl', 'req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:P-256', '-nodes', + '-keyout', work / f'{name}.key', '-out', work / f'{name}.pem', '-days', '1', + '-subj', '/CN=127.0.0.1', '-addext', 'subjectAltName=IP:127.0.0.1'], + check=True, capture_output=True) + received, refused_once = [], set() + + class Hook(BaseHTTPRequestHandler): + protocol_version = 'HTTP/1.1' + + def do_POST(self): + body = self.rfile.read(int(self.headers['Content-Length'])) + received.append((dict(self.headers), json.loads(body))) + status = 200 + if self.headers['X-Pure-Sequence'] == '2' and '2' not in refused_once: + refused_once.add('2') + status = 503 + self.send_response(status) + self.send_header('Content-Length', '0') + self.end_headers() + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Hook) + context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + context.load_cert_chain(work / 'trusted.pem', work / 'trusted.key') + server.socket = context.wrap_socket(server.socket, server_side=True) + threading.Thread(target=server.serve_forever, daemon=True).start() + url = f'ordered=https://127.0.0.1:{server.server_address[1]}/hook' + try: + def buy(host, count): + import http.client + connection = http.client.HTTPConnection('127.0.0.1', host.port, timeout=10) + data = json.dumps({'item': f'i{count}', 'count': count}).encode() + connection.request('POST', '/inputs/buy', data, {'Content-Length': str(len(data)), 'Content-Type': 'application/json'}) + assert connection.getresponse().status == 200 + connection.close() + + host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'hook-a', '--send', url, + '--send-ca', work / 'trusted.pem']) + try: + for count in range(1, 6): + buy(host, count) + deadline = time.time() + 20 + while len({h['X-Pure-Sequence'] for h, _ in received}) < 5 and time.time() < deadline: + time.sleep(0.05) + finally: + stats = host.stop()[-1] + firsts = [] + for headers, body in received: + seq = int(headers['X-Pure-Sequence']) + assert headers['X-Pure-Source'] == 'hook-sender.ordered' and headers['Content-Type'] == 'application/json' + assert headers['Idempotency-Key'] == f'hook-sender.ordered-{seq}' and 'X-Pure-Count' not in headers + assert body == {'item': f'i{seq}', 'count': seq, 'to': {'city': 'Seoul'}}, body + if seq not in firsts: + firsts.append(seq) + assert firsts == [1, 2, 3, 4, 5] and len(received) == 6 and stats['delivered'] == 5, (firsts, len(received), stats) + # A receiver the sender cannot trust gets nothing; the send waits in the outbox. + before = len(received) + host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'hook-b', '--send', url, + '--send-ca', work / 'other.pem']) + try: + buy(host, 1) + time.sleep(1) + finally: + host.process.terminate() + _, error = host.process.communicate(timeout=30) + assert len(received) == before and b'not trusted' in error, error[-500:] + finally: + server.shutdown() + return {'hooks': len(received), 'delivered': stats['delivered']} + + +PAY_SOURCE = """state sent = 0 +state paid = 0 +state declined = 0 +state paid_total = 0 +input buy = message { + order: text + amount: u32 +} +output charge = message { + order: text + amount: u32 +} +input charged = message { + order: text + amount: u32 + status: text + http_status: u32 +} +on buy { + sent = sent + 1 + send charge { order: buy.order, amount: buy.amount } +} +on charged { + if charged.http_status == 200 { + paid = paid + 1 + paid_total = paid_total + charged.amount + } else { + declined = declined + 1 + } +} +view sent_view = sent +view paid_view = paid +view declined_view = declined +view paid_total_view = paid_total +""" + + +def server_answers(aup, pure, work): + """An output wired to a payments API whose answers come back as events + (--reply): the API answers 200 or 402 and gives a key's first answer to + every request with that Idempotency-Key. While clients buy, the sender is + killed (-9) and started again; in the end every saved order was answered + exactly once: the app's paid and declined counts and paid total match the + API's answers, one per key.""" + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + import select + import signal + import socket + import threading + import time + module = work / 'pay.pmod' + module.with_suffix('.aup').write_text(PAY_SOURCE) + compile_module(aup, module.with_suffix('.aup'), module) + answers, lock = {}, threading.Lock() + + class Payments(BaseHTTPRequestHandler): + protocol_version = 'HTTP/1.1' + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers['Content-Length']))) + with lock: + key = self.headers['Idempotency-Key'] + if key not in answers: + ok = body['amount'] <= 1000 + answers[key] = (body, 200 if ok else 402, + {'status': 'succeeded' if ok else 'declined', 'id': f'ch_{len(answers)}', 'x': [1, {}]}) + _, status, answer = answers[key] + data = json.dumps(answer).encode() + self.send_response(status) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Payments) + server.handle_error = lambda request, address: None # a sender killed mid-request + threading.Thread(target=server.serve_forever, daemon=True).start() + with socket.socket() as probe: + probe.bind(('127.0.0.1', 0)) + port = probe.getsockname()[1] + args = [pure / 'build/pure-serve', '--module', module, '--store', work / 'pay-store', '--port', port, + '--send', f'charge=http://127.0.0.1:{server.server_address[1]}/charge', '--reply', 'charge=charged'] + + def start(): + process = subprocess.Popen([str(a) for a in args], stdout=subprocess.PIPE, stderr=subprocess.PIPE) + ready, _, _ = select.select([process.stdout], [], [], 10) + assert ready and process.stdout.readline(), 'host did not start' + return process + + def number(view): + return int(http(port, 'GET', f'/views/{view}')[3]) + + host = start() + stop = threading.Event() + + def client(which): + import http.client + rng = random.Random(which + 8800) + while not stop.is_set(): + data = json.dumps({'order': f'o{which}-{rng.randrange(10**6)}', 'amount': rng.randint(1, 1500)}).encode() + try: + connection = http.client.HTTPConnection('127.0.0.1', port, timeout=10) + connection.request('POST', '/inputs/buy', data, {'Content-Length': str(len(data)), 'Content-Type': 'application/json'}) + connection.getresponse().read() + connection.close() + except OSError: + time.sleep(0.02) + + clients = [threading.Thread(target=client, args=(n,)) for n in range(3)] + for thread in clients: + thread.start() + try: + time.sleep(0.8) + host.send_signal(signal.SIGKILL) + host.wait() + host = start() + time.sleep(0.8) + finally: + stop.set() + for thread in clients: + thread.join() + try: + deadline = time.time() + 30 + while number('paid_view') + number('declined_view') < number('sent_view') and time.time() < deadline: + time.sleep(0.1) + sent, paid, declined, total = (number(v) for v in ('sent_view', 'paid_view', 'declined_view', 'paid_total_view')) + with lock: + keys = list(answers.values()) + assert paid + declined == sent == len(keys), (paid, declined, sent, len(keys)) + assert paid == sum(1 for _, status, _ in keys if status == 200), (paid, len(keys)) + assert total == sum(body['amount'] for body, status, _ in keys if status == 200) + finally: + host.terminate() + _, error = host.communicate(timeout=30) + server.shutdown() + stats = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1] + return {'orders': sent, 'paid': paid, 'declined': declined, 'answers_after_restart': stats['answers']} + + +BEAT_SOURCE = """state beats = 0 +state last = 0 +state stamped = 0 +input beat = every 50 ms +input stamp = message { + note: text +} +on beat { + beats = beats + 1 + last = now +} +on stamp { stamped = now } +view beats_view = beats +view last_view = last +view stamped_view = stamped +""" + + +def server_timers(aup, pure, work): + """A served app's timers tick on the server's clock and its events see + the time they were applied, on one thread and with an owner; the ticks + are saved, so a restarted server goes on from them.""" + import time + module, source = work / 'beat.pmod', work / 'beat.aup' + source.write_text(BEAT_SOURCE) + compile_module(aup, source, module) + seen = {} + for threads in (1, 3): + store = work / f'beat-{threads}' + beats = 0 + for run_number in range(2): + host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', store, '--threads', threads]) + try: + posted = time.time() * 1000 + assert http(host.port, 'POST', '/inputs/stamp', b'x')[0] == 200 + time.sleep(0.6) + before = time.time() * 1000 + read = lambda view: float(http(host.port, 'GET', f'/views/{view}')[3]) + count, last, stamped = read('beats_view'), read('last_view'), read('stamped_view') + after = time.time() * 1000 + 1 # the server's clock reads whole milliseconds + assert count >= beats + 5, (threads, run_number, count, beats) + assert before - 1000 < last <= after and posted - 50 <= stamped <= before, (last, stamped, posted, before) + beats = count + finally: + stats = host.stop() + assert stats[0]['ticks'] >= 5, stats + seen[threads] = beats + return seen + + +def server_app(aup, pure, work, seeds, operations=200, threads=1): + """An app served by its views and inputs, no routes written: random + POSTs (a message with query fields and a body, a text, a key) and GETs + (a number, a text, a table's pages) answer exactly as a Python model + says, failures with their fault; what was acknowledged is there after + the server stops, and after it is killed while clients post. The app's + owner thread applies the inputs; connection threads (one or several) + show views from the snapshot of its last save.""" + import signal + import threading + import time + module = work / 'served.pmod' + source = work / 'served.aup' + source.write_text(SERVED_SOURCE) + compile_module(aup, source, module) + names = ['ann', 'bob', 'zed', '가', 'a"b', 'back\\slash', 'tab\t', '', 'bob2'] + compared = 0 + for seed in seeds: + rng, board, at = random.Random(seed + 9100), {}, 0 + store = work / f'served-{seed}-{threads}' + serve = [pure / 'build/pure-serve', '--module', module, '--threads', threads] + host = HttpHost(serve + ['--store', store]) + order = lambda: sorted(board, key=lambda name: name.encode()) + + def place(name): + nonlocal at + keys = order() + if name not in board: + position = sum(1 for k in keys if k.encode() < name.encode()) + board[name] = 0 + if len(board) > 1 and at >= position: at += 1 + + try: + for _ in range(operations): + roll, keys = rng.random(), order() + if roll < 0.35: + name, points = rng.choice(names), rng.randint(0, 9) + if rng.random() < 0.05: + request = ('POST', '/inputs/score', b'x') # no points + want = (400, 'missing_field', b'') + elif rng.random() < 0.05: + request = ('POST', f'/inputs/score?points={points}', b'\xff') + want = (422, 'encoding', b'') + else: + request = ('POST', f'/inputs/score?points={points}', name.encode()) + place(name) + board[name] += points + at = order().index(name) + want = (200, 'none', name.encode()) + elif roll < 0.5: + name = rng.choice(names) + request = ('POST', '/inputs/join', name.encode()) + if name in board: want = (409, 'exists', b'') + else: + place(name) + at = order().index(name) + want = (200, 'none', b'') + elif roll < 0.58: + request = ('POST', '/inputs/drop', b'') + if not board: want = (404, 'not_found', b'') + else: + del board[keys[at]] + at = min(at, max(len(board) - 1, 0)) + want = (200, 'none', b'') + elif roll < 0.68: + request = ('GET', '/views/total', b'') + want = (200, 'none', str(sum(board.values())).encode()) + elif roll < 0.78: + request = ('GET', '/views/current', b'') + want = (200, 'none', keys[at].encode()) if board else (404, 'no_value', b'') + elif roll < 0.96: + first, count = rng.randint(0, 4), rng.choice([1, 2, 5, 100]) + request = ('GET', f'/views/leaders?from={first}&count={count}', b'') + shown = min(first, len(keys)) + rows = [{'name': k, 'points': board[k]} for k in keys[shown:shown + count]] + want = (200, 'none', {'rows': len(keys), 'from': shown, 'items': rows}) + else: + request, want = rng.choice([(('GET', '/views/nope', b''), (404, 'no_view', b'')), + (('POST', '/inputs/nope', b''), (404, 'no_input', b'')), + (('POST', '/inputs/join?x=1', b'q'), (400, 'query', b'')), + (('GET', '/views/total?from=1', b''), (400, 'query', b''))]) + status, error, kind, body = http(host.port, *request) + if isinstance(want[2], dict): + assert kind == 'application/json', kind + body = json.loads(body) + assert (status, error, body) == want, (seed, request, (status, error, body), want) + compared += 1 + finally: + stats = host.stop() + assert stats[0]['bursts'] <= stats[0]['requests'] and stats[0]['owner'], stats + assert stats[0]['views_from_snapshots'] > 0, stats + # A graceful stop kept every acknowledged event. + host = HttpHost(serve + ['--store', store]) + try: + got = json.loads(http(host.port, 'GET', '/views/leaders')[3]) + assert got['items'] == [{'name': k, 'points': board[k]} for k in order()], (seed, got) + finally: + host.stop() + + # Clients post while the server is killed: each acknowledged point is + # stored. Readers on other connections meanwhile see every point + # acknowledged before they asked, never a total going back, and never one + # that was not saved: the total after the kill is at least any they saw. + store = work / f'served-kill-{threads}' + serve = [pure / 'build/pure-serve', '--module', module, '--threads', threads] + host = HttpHost(serve + ['--store', store]) + acknowledged = [0] * 6 + seen = [0, 0] + + def client(number): + for _ in range(10000): + try: + status, _, _, _ = http(host.port, 'POST', '/inputs/score?points=1', f'c{number}'.encode()) + except OSError: + return + if status != 200: + return + acknowledged[number] += 1 + + def reader(number): + for _ in range(100000): + floor = sum(acknowledged) + try: + status, _, _, body = http(host.port, 'GET', '/views/total') + except OSError: + return + if status != 200: + return + total = int(body) + assert floor <= total and seen[number] <= total, (floor, seen[number], total) + seen[number] = total + + clients = [threading.Thread(target=client, args=(n,)) for n in range(len(acknowledged))] + clients += [threading.Thread(target=reader, args=(n,)) for n in range(len(seen))] + for thread in clients: + thread.start() + time.sleep(0.8) + host.process.send_signal(signal.SIGKILL) + host.process.wait() + for thread in clients: + thread.join() + host = HttpHost(serve + ['--store', store]) + try: + stored = {row['name']: row['points'] for row in json.loads(http(host.port, 'GET', '/views/leaders')[3])['items']} + for number, count in enumerate(acknowledged): + assert count > 0 and stored.get(f'c{number}', 0) in (count, count + 1), (number, count, stored) + assert min(seen) > 0 and max(seen) <= sum(stored.values()), (seen, stored) + finally: + host.stop() + return compared, sum(acknowledged) + + +def main(): + global EDITOR + parser = argparse.ArgumentParser() + parser.add_argument('--aup', type=Path, required=True, help='the aup compiler') + parser.add_argument('--aup-root', type=Path, default=ROOT.parent / 'aup', help='its checkout, for examples/editor.aup') + parser.add_argument('--seeds', type=int, default=4) + args = parser.parse_args() + aup, pure = args.aup.resolve(), ROOT + EDITOR = args.aup_root.resolve() / 'examples' / 'editor.aup' + report = {} + with tempfile.TemporaryDirectory(prefix='pure-check-') as directory: + work = Path(directory) + report['server_parity_requests'] = sum(server_parity(aup, pure, work, range(args.seeds), threads=threads) + for threads in (1, 3)) + report['server_pipelined_answers'] = server_pipelining(aup, pure, work) + report['server_lent_read_bytes'] = server_lent_reads(aup, pure, work) + report['server_lent_budget_bytes'] = server_lent_budget(aup, pure, work) + report['server_lent_views'] = [server_lent_views(aup, pure, work, threads) for threads in (1, 3)] + report['server_bursts'] = server_bursts(aup, pure, work) + report['server_bursts_three_threads'] = server_bursts(aup, pure, work, shards=3) + report['server_app_requests'], report['server_app_acknowledged_before_kill'] = \ + server_app(aup, pure, work, range(args.seeds)) + report['server_app_owner_requests'], report['server_app_owner_acknowledged_before_kill'] = \ + server_app(aup, pure, work, range(args.seeds), threads=3) + report['server_json_requests'] = server_json(aup, pure, work) + report['server_timer_beats'] = server_timers(aup, pure, work) + report['server_links'] = server_links(aup, pure, work) + report['server_webhook'] = server_webhook(aup, pure, work) + report['server_answers'] = server_answers(aup, pure, work) + print(json.dumps(report)) + + +if __name__ == '__main__': + sys.exit(main()) -- cgit v1.2.3