From 954e25c7f2d86b8be55405ab69027a32115b851a Mon Sep 17 00:00:00 2001 From: eunhhu Date: Sun, 27 Sep 2026 08:22:21 +0100 Subject: Check a payments API's answers coming back as events exactly once A Python payments API answers 200 or 402 and repeats a key's first answer; the sender's output is wired to it and its answers to the charged input. With the sender killed (-9) while clients buy, the app's paid and declined counts and paid total match the API's answers one per key. The language notes say how answers arrive. --- docs/design/001-language-v0.md | 3 + tests/check.py | 149 ++++++++++++++++++++++++++++++++++++++++- 2 files changed, 149 insertions(+), 3 deletions(-) diff --git a/docs/design/001-language-v0.md b/docs/design/001-language-v0.md index 0f95500..3fe9ff8 100644 --- a/docs/design/001-language-v0.md +++ b/docs/design/001-language-v0.md @@ -123,6 +123,9 @@ on edit { `send 이름 { 필드: 값, 그룹: { 필드: 값 } }`으로 모든 필드를 한 번씩 주어 보낸다. 보냄은 이벤트의 일부라서 이벤트가 실패하면 보내지 않았고, 호스트는 저장된 보냄만 전한다. 무엇이 받는지는 앱 밖(호스트의 배선)이 정한다. 받는 앱은 같은 메시지를 여러 번 받아도 한 번만 적용한다(Pure `docs/research/078-app-connections.md`). 출력에는 핸들러가 없다. +- 외부 API의 응답은 보통의 메시지 입력으로 받는다. 호스트가 출력의 응답을 그 입력에 이으면(`pure-serve --reply 출력=입력`), + 필드는 응답 JSON에서, 응답에 없는 필드는 보낸 메시지의 같은 이름 필드에서, `http_status: u32`는 상태 코드로 채워진다 + (Pure `docs/research/083-delivery-over-http.md`). - `이름[key]`는 그 메시지 키의 문서다. 읽으면 bytes, `.bytes`는 크기(없으면 0)다. 없는 문서를 읽으면 `not_found`. - `replace(a .. b, 값)`는 문서의 `[a, b)`를 바꾼다. 정수가 아니거나 범위 밖이면 이벤트가 `limit`으로 실패한다. - `require 조건`이 거짓이면 이벤트가 `limit`으로 실패한다. 크기 정책처럼 저장소의 특수 인자였던 것이 문장 하나가 된다. diff --git a/tests/check.py b/tests/check.py index 0cef00b..dfec9d5 100644 --- a/tests/check.py +++ b/tests/check.py @@ -1484,7 +1484,7 @@ def server_webhook(aup, pure, work): 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.""" - import http.server + from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import ssl import threading import time @@ -1498,7 +1498,7 @@ def server_webhook(aup, pure, work): check=True, capture_output=True) received, refused_once = [], set() - class Hook(http.server.BaseHTTPRequestHandler): + class Hook(BaseHTTPRequestHandler): protocol_version = 'HTTP/1.1' def do_POST(self): @@ -1515,7 +1515,7 @@ def server_webhook(aup, pure, work): def log_message(self, *args): pass - server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Hook) + 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) @@ -1565,6 +1565,148 @@ def server_webhook(aup, pure, work): 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 @@ -2271,6 +2413,7 @@ def main(): report['server_timer_beats'] = server_timers(aup, args.pure.resolve(), work) report['server_links'] = server_links(aup, args.pure.resolve(), work) report['server_webhook'] = server_webhook(aup, args.pure.resolve(), work) + report['server_answers'] = server_answers(aup, args.pure.resolve(), work) print(json.dumps(report)) -- cgit v1.2.3