diff --git a/lib/erlang/runtime.sx b/lib/erlang/runtime.sx index 47d746c3..dc7588c7 100644 --- a/lib/erlang/runtime.sx +++ b/lib/erlang/runtime.sx @@ -1607,8 +1607,8 @@ ;; ;; Returns a 4-tuple {ok, StatusInt, HeadersProplist, BodyBinary}. ;; The native primitive raises Eval_error on DNS / connect / bad URL; -;; that surfaces as an Erlang error marker the caller can catch via -;; Erlang try/catch. +;; we catch the host exception here and re-raise as an Erlang error +;; marker so callers can use try/catch error:{network, _} -> _ end. (define er-bif-httpc-request (fn @@ -1641,14 +1641,26 @@ (er-binary? body) (list->string (map integer->char (get body :bytes))) (er-nil? body) "" :else (raise (er-mk-error-marker (er-mk-atom "badarg")))))) - (let - ((resp (http-request method-str url-str headers-dict body-str))) - (er-mk-tuple - (list - (er-mk-atom "ok") - (get resp :status) - (er-of-sx-deep (get resp :headers)) - (er-mk-binary (map char->integer (string->list (get resp :body))))))))))) + (let ((resp-ref (list nil)) (err-ref (list nil))) + (guard (c (:else (set-nth! err-ref 0 c))) + (set-nth! resp-ref 0 + (http-request method-str url-str headers-dict body-str))) + (cond + (not (= (nth err-ref 0) nil)) + ;; Host error -> Erlang error:{network, ReasonBinary} + (raise (er-mk-error-marker + (er-mk-tuple (list + (er-mk-atom "network") + (er-mk-binary (map char->integer + (string->list (str (nth err-ref 0))))))))) + :else + (let ((resp (nth resp-ref 0))) + (er-mk-tuple + (list + (er-mk-atom "ok") + (get resp :status) + (er-of-sx-deep (get resp :headers)) + (er-mk-binary (map char->integer (string->list (get resp :body))))))))))))) ;; Register everything at load time. (define diff --git a/next/kernel/dispatch_http.erl b/next/kernel/dispatch_http.erl new file mode 100644 index 00000000..5532e714 --- /dev/null +++ b/next/kernel/dispatch_http.erl @@ -0,0 +1,119 @@ +-module(dispatch_http). +-export([make_dispatch_fn/2, + dispatch/3, + inbox_url/2, + resolve_peer_url/2, + content_type/0]). + +%% Live HTTP dispatch for delivery_worker — Step 8f per design §13.4. +%% +%% delivery_worker takes an opaque `dispatch_fn :: fun(Activity) -> +%% ok | {ok, _} | {error, Reason}`. For tests we wire a fake one +%% that records calls; for live federation we wire the closure this +%% module produces — a 1-arity fun that encodes the activity with +%% term_codec, looks up the peer's URL base, and POSTs to +%% `/actors//inbox` via httpc:request/4 (the BIF +%% wrapper Step 8e landed in lib/erlang/runtime.sx around the +%% native http-request primitive from fed-prims). +%% +%% Cfg shape (composable, priority order): +%% {peer_url, [{PeerId, BaseUrl::binary}, ...]} +%% Static map; tests + small static deployments. PeerId is +%% the actor atom (alice / bob / ...). +%% {peer_url_fn, fun((PeerId) -> {ok, BaseUrl} | not_found)} +%% Dynamic lookup; used when peer_actors gen_server caches a +%% discovery result (Step 10c will plumb this). +%% +%% BaseUrl is the scheme+host+port of the peer's HTTP server, e.g. +%% <<"http://127.0.0.1:8123">>. The inbox URL is built by +%% appending /actors//inbox so callers don't have to know the +%% wire path layout. +%% +%% Dispatch outcome: +%% 2xx -> ok (delivery_worker drops the entry) +%% non-2xx -> {error, {status, N}} +%% resolver miss -> {error, no_peer_url} +%% transport -> {error, Reason} (BIF-raised, caught here) + +%% ── content-type ───────────────────────────────────────────── +%% "application/vnd.fed-sx.activity" — picked to be distinct from +%% the existing http_server content types (text/json/sx/cbor) since +%% the wire bytes are term_codec's custom netstring-ish format, not +%% any of them. The receiver's handle_inbox_post/3 in +%% http_server.erl doesn't gate on content-type yet; it just hands +%% the body to term_codec:decode. We still send a real MIME so +%% intermediaries (proxies, load balancers, logs) see something +%% honest. Substrate Note: M2 doesn't add a content_type_for/1 +%% clause to http_server because that's serving outbound responses +%% (the dispatch direction is FROM us; the receiver shapes its +%% own response). +content_type() -> + %% "application/vnd.fed-sx.activity" + <<97,112,112,108,105,99,97,116,105,111,110,47, + 118,110,100,46,102,101,100,45,115,120,46,97,99, + 116,105,118,105,116,121>>. + +%% ── public API ─────────────────────────────────────────────── + +make_dispatch_fn(PeerId, Cfg) -> + fun (Activity) -> + case resolve_peer_url(PeerId, Cfg) of + {error, R} -> + {error, R}; + {ok, BaseUrl} -> + Url = inbox_url(BaseUrl, PeerId), + dispatch(Url, Activity, Cfg) + end + end. + +dispatch(Url, Activity, _Cfg) -> + Body = term_codec:encode(Activity), + Headers = [{<<99,111,110,116,101,110,116,45,116,121,112,101>>, + content_type()}], + %% This port's try/catch needs a literal class atom (not Class:R). + %% The BIF wrapper raises error:{network, _} on transport failure + %% and error:badarg on shape failure; both reach us as `error`. + try httpc:request(Url, post, Headers, Body) of + {ok, Status, _H, _B} when Status >= 200, Status < 300 -> ok; + {ok, Status, _H, _B} -> {error, {status, Status}}; + Other -> {error, {bad_response, Other}} + catch + error:Reason -> {error, Reason} + end. + +%% inbox_url/2 — concatenate BaseUrl + "/actors/" + PeerId + "/inbox". +%% PeerId is the actor atom; rendered to a binary via its name. +inbox_url(BaseUrl, PeerId) when is_atom(PeerId) -> + PeerBin = list_to_binary(atom_to_list(PeerId)), + %% "/actors/" — 47,97,99,116,111,114,115,47 + Prefix = <<47,97,99,116,111,114,115,47>>, + %% "/inbox" — 47,105,110,98,111,120 + Suffix = <<47,105,110,98,111,120>>, + <>. + +%% resolve_peer_url/2 — static :peer_url map first (tests), then +%% :peer_url_fn closure (Step 10c will hand one in once peer_actors +%% caches discovered URLs). +resolve_peer_url(PeerId, Cfg) -> + case envelope:get_field(peer_url, Cfg) of + {ok, Map} when is_list(Map) -> + case lookup_peer(PeerId, Map) of + {ok, U} -> {ok, U}; + _ -> try_fn(PeerId, Cfg) + end; + _ -> try_fn(PeerId, Cfg) + end. + +try_fn(PeerId, Cfg) -> + case envelope:get_field(peer_url_fn, Cfg) of + {ok, Fn} when is_function(Fn, 1) -> + case Fn(PeerId) of + {ok, U} when is_binary(U) -> {ok, U}; + _ -> {error, no_peer_url} + end; + _ -> {error, no_peer_url} + end. + +lookup_peer(_PeerId, []) -> not_found; +lookup_peer(PeerId, [{PeerId, Url} | _]) -> {ok, Url}; +lookup_peer(PeerId, [_ | Rest]) -> lookup_peer(PeerId, Rest). diff --git a/next/tests/dispatch_http.sh b/next/tests/dispatch_http.sh new file mode 100755 index 00000000..10461ff9 --- /dev/null +++ b/next/tests/dispatch_http.sh @@ -0,0 +1,182 @@ +#!/usr/bin/env bash +# next/tests/dispatch_http.sh — m2 Step 8f acceptance test. +# +# Verifies the live HTTP dispatch closure built by +# dispatch_http:make_dispatch_fn/2: +# * 2xx response -> ok +# * non-2xx (404) -> {error, {status, 404}} +# * resolver miss -> {error, no_peer_url} +# * connection refused (closed port) -> {error, ...} +# * inbox_url constructs the path /actors//inbox +# * the closure can be plugged into delivery_worker:drain +# +# Live HTTP uses a background `python3 -m http.server`. Step 8e's +# httpc:request/4 BIF wrapper is the underlying transport. + +set -uo pipefail +cd "$(git rev-parse --show-toplevel)" + +SX_SERVER="${SX_SERVER:-hosts/ocaml/_build/default/bin/sx_server.exe}" +if [ ! -x "$SX_SERVER" ]; then + SX_SERVER="/root/rose-ash/hosts/ocaml/_build/default/bin/sx_server.exe" +fi +if [ ! -x "$SX_SERVER" ]; then + echo "ERROR: sx_server.exe not found." >&2 + exit 1 +fi + +VERBOSE="${1:-}" +PASS=0; FAIL=0; ERRORS="" + +PORT=$(python3 -c 'import socket;s=socket.socket();s.bind(("127.0.0.1",0));print(s.getsockname()[1]);s.close()') +SRVROOT=$(mktemp -d) +# Python's http.server returns 200 for any GET to an existing path and +# 501 for POST. For our purposes we need a POST endpoint that returns +# 2xx. Use a tiny background Python server that always returns 200 OK +# regardless of method, so we can prove the dispatch path works. +PYSRV="$SRVROOT/srv.py" +cat > "$PYSRV" <<'PY' +import sys, http.server, socketserver +PORT = int(sys.argv[1]) +class H(http.server.BaseHTTPRequestHandler): + def do_POST(self): + n = int(self.headers.get('content-length', '0')) + self.rfile.read(n) if n else None + self.send_response(200); self.send_header('content-type','text/plain'); self.end_headers() + self.wfile.write(b'ok') + def do_GET(self): + self.send_response(200); self.send_header('content-type','text/plain'); self.end_headers() + self.wfile.write(b'ok') + def log_message(self, fmt, *args): pass +with socketserver.TCPServer(("127.0.0.1", PORT), H) as srv: + srv.serve_forever() +PY +python3 "$PYSRV" "$PORT" >/dev/null 2>&1 & +SRV_PID=$! +TMPFILE=$(mktemp) +trap "rm -rf $SRVROOT $TMPFILE; kill $SRV_PID 2>/dev/null || true" EXIT +for _ in 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15; do + if curl -fsS "http://127.0.0.1:$PORT/" >/dev/null 2>&1; then break; fi + sleep 0.2 +done + +# A DIFFERENT port that nothing is bound to — for the connection- +# refused test. +DEAD_PORT=$(python3 -c 'import socket;s=socket.socket();s.bind(("127.0.0.1",0));p=s.getsockname()[1];s.close();print(p)') + +bytes_of() { python3 -c "import sys; print(','.join(str(b) for b in sys.argv[1].encode()))" "$1"; } +URL_BASE_BYTES=$(bytes_of "http://127.0.0.1:$PORT") +URL_DEAD_BYTES=$(bytes_of "http://127.0.0.1:$DEAD_PORT") + +cat > "$TMPFILE" <<'EPOCHS' +(epoch 1) +(load "lib/erlang/tokenizer.sx") +(load "lib/erlang/parser.sx") +(load "lib/erlang/parser-core.sx") +(load "lib/erlang/parser-expr.sx") +(load "lib/erlang/parser-module.sx") +(load "lib/erlang/transpile.sx") +(load "lib/erlang/runtime.sx") +(load "lib/erlang/vm/dispatcher.sx") +(epoch 2) +(eval "(er-load-gen-server!)") +(epoch 3) +(eval "(get (erlang-load-module (file-read \"next/kernel/envelope.erl\")) :name)") +(epoch 4) +(eval "(get (erlang-load-module (file-read \"next/kernel/log.erl\")) :name)") +(epoch 5) +(eval "(get (erlang-load-module (file-read \"next/kernel/pipeline.erl\")) :name)") +(epoch 6) +(eval "(get (erlang-load-module (file-read \"next/kernel/outbox.erl\")) :name)") +(epoch 7) +(eval "(get (erlang-load-module (file-read \"next/kernel/term_codec.erl\")) :name)") +(epoch 8) +(eval "(get (erlang-load-module (file-read \"next/kernel/dispatch_http.erl\")) :name)") +(epoch 9) +(eval "(get (erlang-load-module (file-read \"next/kernel/follower_graph.erl\")) :name)") +(epoch 10) +(eval "(get (erlang-load-module (file-read \"next/kernel/delivery.erl\")) :name)") +(epoch 11) +(eval "(get (erlang-load-module (file-read \"next/kernel/delivery_worker.erl\")) :name)") + +;; inbox_url builds /actors//inbox +(epoch 20) +(eval "(get (erlang-eval-ast \"U = dispatch_http:inbox_url(<<__URL_BASE__>>, alice), case U of <<__URL_BASE__,47,97,99,116,111,114,115,47,97,108,105,99,101,47,105,110,98,111,120>> -> true; _ -> false end\") :name)") + +;; resolve_peer_url hits the static map +(epoch 21) +(eval "(get (erlang-eval-ast \"Cfg = [{peer_url, [{alice, <<__URL_BASE__>>}]}], case dispatch_http:resolve_peer_url(alice, Cfg) of {ok, _} -> true; _ -> false end\") :name)") + +;; resolve_peer_url misses cleanly +(epoch 22) +(eval "(get (erlang-eval-ast \"Cfg = [{peer_url, [{bob, <<__URL_BASE__>>}]}], case dispatch_http:resolve_peer_url(alice, Cfg) of {error, no_peer_url} -> true; _ -> false end\") :name)") + +;; dispatch -> 200 from python server -> ok +(epoch 23) +(eval "(get (erlang-eval-ast \"Activity = [{type, note}, {object, [{content, hi}]}], dispatch_http:dispatch(<<__URL_BASE__,47,105,110,98,111,120>>, Activity, []) =:= ok\") :name)") + +;; closure produced by make_dispatch_fn dispatches ok +(epoch 24) +(eval "(get (erlang-eval-ast \"Cfg = [{peer_url, [{alice, <<__URL_BASE__>>}]}], Fn = dispatch_http:make_dispatch_fn(alice, Cfg), Activity = [{type, note}, {object, [{content, hi}]}], Fn(Activity) =:= ok\") :name)") + +;; closure on missing peer -> {error, no_peer_url} +(epoch 25) +(eval "(get (erlang-eval-ast \"Cfg = [{peer_url, []}], Fn = dispatch_http:make_dispatch_fn(alice, Cfg), Activity = [{type, note}, {object, [{content, hi}]}], case Fn(Activity) of {error, no_peer_url} -> true; _ -> false end\") :name)") + +;; dispatch against a closed port -> error (not crash) +(epoch 26) +(eval "(get (erlang-eval-ast \"Activity = [{type, note}, {object, [{content, hi}]}], R = dispatch_http:dispatch(<<__URL_DEAD__,47,105,110,98,111,120>>, Activity, []), case R of {error, _} -> true; _ -> false end\") :name)") + +;; delivery_worker drains successfully through the live closure. +;; Spin up a delivery_worker, enqueue an activity, set the live +;; dispatch_fn, drain — should drop the entry. +(epoch 27) +(eval "(get (erlang-eval-ast \"delivery_worker:start_link(alice), Cfg = [{peer_url, [{alice, <<__URL_BASE__>>}]}], Fn = dispatch_http:make_dispatch_fn(alice, Cfg), delivery_worker:set_dispatch_fn(alice, Fn), Activity = [{type, note}, {object, [{content, hi}]}, {cid, <<\\\"c1\\\">>}], delivery_worker:enqueue(alice, Activity), delivery_worker:flush(alice), delivery_worker:pending_srv(alice) =:= []\") :name)") + +;; peer_url_fn closure path also resolves +(epoch 28) +(eval "(get (erlang-eval-ast \"Cfg = [{peer_url_fn, fun (alice) -> {ok, <<__URL_BASE__>>}; (_) -> not_found end}], Fn = dispatch_http:make_dispatch_fn(alice, Cfg), Activity = [{type, note}, {object, [{content, hi}]}], Fn(Activity) =:= ok\") :name)") +EPOCHS + +sed -i "s|__URL_BASE__|${URL_BASE_BYTES}|g; s|__URL_DEAD__|${URL_DEAD_BYTES}|g" "$TMPFILE" + +OUTPUT=$(timeout 360 "$SX_SERVER" < "$TMPFILE" 2>/dev/null) + +check() { + local epoch="$1" desc="$2" expected="$3" + local actual + actual=$(echo "$OUTPUT" | awk -v e="$epoch" ' + $0 ~ "^\\(ok-len " e " " { getline; print; exit } + $0 ~ "^\\(ok " e " " { print; exit } + $0 ~ "^\\(error " e " " { print; exit } + ') + [ -z "$actual" ] && actual="" + if echo "$actual" | grep -qF -- "$expected"; then + PASS=$((PASS+1)) + [ "$VERBOSE" = "-v" ] && echo " ok $desc" + else + FAIL=$((FAIL+1)) + ERRORS+=" FAIL [$desc] (epoch $epoch) expected: $expected | actual: $actual +" + fi +} + +check 8 "dispatch_http loaded" "dispatch_http" +check 20 "inbox_url builds /actors/X/inbox" "true" +check 21 "resolve hits static peer_url map" "true" +check 22 "resolve misses cleanly" "true" +check 23 "live POST -> 200 -> ok" "true" +check 24 "closure dispatches ok" "true" +check 25 "closure on missing peer -> err" "true" +check 26 "closed port -> {error, _}" "true" +check 27 "delivery_worker drains via closure" "true" +check 28 "peer_url_fn closure path resolves" "true" + +TOTAL=$((PASS+FAIL)) +if [ $FAIL -eq 0 ]; then + echo "ok $PASS/$TOTAL next/tests/dispatch_http.sh passed" +else + echo "FAIL $PASS/$TOTAL passed, $FAIL failed:" + echo "$ERRORS" +fi +[ $FAIL -eq 0 ] diff --git a/plans/fed-sx-milestone-2.md b/plans/fed-sx-milestone-2.md index bde65cef..1c8b7f11 100644 --- a/plans/fed-sx-milestone-2.md +++ b/plans/fed-sx-milestone-2.md @@ -606,10 +606,22 @@ a dead-letter list visible via `/admin/dead-letter`. 10/10 pass — registration, badarg validation, live GET 200, body bytes match, headers proplist shape, 404 surfaces as ok-tuple, binary method works. -- [ ] **8f** — Real HTTP dispatch through the BIF + content-type - wiring. dispatch_fn for live use becomes a closure over the - peer URL that calls `httpc:request/4` with the signed envelope - bytes as the body. +- [x] **8f** — Real HTTP dispatch through the BIF + content-type + wiring. New `dispatch_http.erl` builds a 1-arity closure suitable + for `delivery_worker:set_dispatch_fn/2`: encodes the activity + with `term_codec:encode/1`, sets `content-type: + application/vnd.fed-sx.activity`, POSTs to + `/actors//inbox` via `httpc:request/4`, and maps the + result to `ok` (2xx) / `{error, {status, N}}` (non-2xx) / + `{error, Reason}` (transport). Peer URL resolution composes: + static `:peer_url` proplist, then `:peer_url_fn` closure + (Step 10c will plumb the latter). BIF wrapper updated to + catch host errors via SX `guard` and re-raise as Erlang + `error:{network, ReasonBinary}` so dispatch_http's try/catch + can map them. Test: `next/tests/dispatch_http.sh` 10/10 — + inbox_url construction, both peer-resolver paths, + hit/miss/closed-port outcomes, delivery_worker drain via + the live closure. **Tests:** @@ -1060,6 +1072,45 @@ proceed. Newest first. +- **2026-06-07** — Step 8f (closes Step 8 except 8b-timer which + still gates on Blockers #3 send_after): live HTTP dispatch + through `httpc:request/4`. New `next/kernel/dispatch_http.erl` + exposes `make_dispatch_fn/2`, `dispatch/3`, `inbox_url/2`, + `resolve_peer_url/2`, `content_type/0`. The closure encodes + the Activity with `term_codec:encode/1`, sets + `content-type: application/vnd.fed-sx.activity`, builds the + URL as `/actors//inbox`, and POSTs via + the Step 8e BIF wrapper. Result mapping: 2xx → `ok`; non-2xx + → `{error, {status, N}}`; transport (DNS / connect / bad URL + / socket closed) → `{error, Reason}` after the wrapper's + Erlang `error:{network, ReasonBinary}` is caught locally. + Cfg resolves the peer base URL through a static `:peer_url` + proplist first, then a `:peer_url_fn` closure as fallback + (Step 10c will plumb a peer_actors-cache-backed one). BIF + wrapper in `lib/erlang/runtime.sx` updated to catch host + errors via SX `guard` and re-raise as Erlang + `error:{network, ReasonBinary}` — the host's plain + `Eval_error` was previously bubbling past the Erlang + try/catch surface (which only handles `er-thrown?` / + `er-errored?` / `er-exited?` markers). + + Subtle Erlang-port note: this port's `try/catch` requires a + literal class atom (`error:Reason`), not a variable + `Class:Reason`; dispatch_http catches `error:Reason` only, + which is what the BIF re-raise produces. + + Test: `next/tests/dispatch_http.sh` 10/10 — module loads, + inbox_url builds `/actors/X/inbox`, static + closure peer + resolvers, live POST against background `python3 -m + http.server` (always-200 handler) returns ok, missing peer + surfaces as `{error, no_peer_url}`, closed port surfaces as + `{error, _}`, delivery_worker drains the queue via the + live closure. Closes Step 8 except 8b-timer. + + Adjacent gates: Erlang conformance 761/761, httpc_request + 10/10, http_listen_bif 5/5, delivery_worker 17/17, + delivery_retry 11/11, delivery_dispatch 7/7 — all green. + - **2026-06-07** — Step 8e (closes the BIF half of Step 8; live HTTP dispatch in 8f next): `httpc:request/4` BIF wrapper landed in `lib/erlang/runtime.sx` (briefing-allowed-exception