diff --git a/.aspect/axl.axl b/.aspect/axl.axl index 76eb744ae..c5d7f53c5 100644 --- a/.aspect/axl.axl +++ b/.aspect/axl.axl @@ -16,7 +16,6 @@ load("@aspect//private/lib/artifacts.axl", "artifact_name", "build_log_header", load("@aspect//private/lib/bazel_results.axl", "bb_clientd_root", "compute_reproducer_command", "file_uri_to_path", "init_data", "render_check_output", "resolve_aspect_url", "summary_title") load("@aspect//private/lib/check_dispatch.axl", "SURFACE_PR_COMMENT", "SURFACE_STATUS_CHECK", "TEMPLATE_SCOPE_KEYS", "resolve_templates", "snippet_budget_for") load("@aspect//private/lib/ci.axl", "detect_build_url") -load("@aspect//private/lib/deliveryd.axl", deliveryd_parse_endpoint = "parse_endpoint") load("@aspect//private/lib/environment.axl", "color_enabled", "detect_ci", "parse_git_url_name", "sanitize_filename") load( "@aspect//private/lib/gazelle_results.axl", @@ -1820,25 +1819,6 @@ def test_sanitize_filename(tc: int) -> int: tc = test_case(tc, got == expected, "sanitize_filename: %s → %r (got %r)" % (desc, expected, got)) return tc -def test_deliveryd_parse_endpoint(tc: int) -> int: - """Coverage for deliveryd.parse_endpoint.""" - base, sock = deliveryd_parse_endpoint("unix:///run/deliveryd.sock") - tc = test_case(tc, base == "http://localhost", "parse_endpoint: unix → base = http://localhost") - tc = test_case(tc, sock == "/run/deliveryd.sock", "parse_endpoint: unix → socket path") - - base, sock = deliveryd_parse_endpoint("http://deliveryd.local:8080") - tc = test_case(tc, base == "http://deliveryd.local:8080", "parse_endpoint: http → base passthrough") - tc = test_case(tc, sock == None, "parse_endpoint: http → no socket") - - base, sock = deliveryd_parse_endpoint("https://deliveryd.example.com") - tc = test_case(tc, base == "https://deliveryd.example.com", "parse_endpoint: https → base passthrough") - tc = test_case(tc, sock == None, "parse_endpoint: https → no socket") - - # Invalid scheme should fail() — we can't catch it easily, so just don't test - # the failure path here. - - return tc - def test_delivery_should_upload_grpc_log(tc: int) -> int: """Coverage for the --upload-grpc-log tristate gating predicate.""" @@ -4112,7 +4092,6 @@ def impl(ctx: TaskContext) -> int: tc = test_repro_commands_lib(tc) tc = test_parse_git_url_name(tc) tc = test_sanitize_filename(tc) - tc = test_deliveryd_parse_endpoint(tc) tc = test_delivery_should_upload_grpc_log(tc) tc = test_delivery_count_unexplained_unresolved(tc) tc = test_delivery_should_surface_phase2_stderr(tc) diff --git a/.aspect/config.axl b/.aspect/config.axl index 5de9e26d9..e8f69b538 100644 --- a/.aspect/config.axl +++ b/.aspect/config.axl @@ -17,6 +17,7 @@ load("@aspect//private/lib/bazelrc_test.axl", "bazelrc_tests") load("@aspect//private/lib/ci_test.axl", "ci_tests") load("@aspect//private/lib/circleci_test.axl", "circleci_tests") load("@aspect//private/lib/delivery_results_test.axl", "delivery_results_unit_tests", "delivery_template_snapshot_tests") +load("@aspect//private/lib/deliveryd_test.axl", "deliveryd_tests") load("@aspect//private/lib/deployment_flags_test.axl", "deployment_flags_tests") load("@aspect//private/lib/format_results_test.axl", "format_template_snapshot_tests") load("@aspect//private/lib/format_spawn_test.axl", "format_spawn_tests") @@ -380,6 +381,10 @@ def config(ctx: ConfigContext): # Run with: aspect dev test-deployment-flags ctx.tasks.add(deployment_flags_tests) + # deliveryd.axl: v1/v2 negotiation and the wire shape of each verb. + # Run with: aspect dev test-deliveryd + ctx.tasks.add(deliveryd_tests) + # bazel/build_events.axl: the end-of-build BES upload summary line. # Run with: aspect dev test-bes-sinks ctx.tasks.add(bes_sinks_tests) diff --git a/crates/aspect-cli/src/builtins/aspect/delivery.axl b/crates/aspect-cli/src/builtins/aspect/delivery.axl index 546a9e3eb..1d05254d4 100644 --- a/crates/aspect-cli/src/builtins/aspect/delivery.axl +++ b/crates/aspect-cli/src/builtins/aspect/delivery.axl @@ -14,11 +14,12 @@ Up to three Bazel build phases plus a dispatch loop: a remote executor: use the user's `--remote_executor` if set, else an in-process dummy so a plain `--remote_cache` suffices — see `lib/remote_executor.axl`. - 3. deliveryd record/query — record (label, digest) pairs, query per-prefix state. + 3. deliveryd record/query — record (label, digest) pairs, query per-prefix + state. A v2 daemon needs no record step; `lib/deliveryd.axl` negotiates. 4. Phase 3 — download runfiles for targets that survived change detection, streaming BES into a run tracker to discover entrypoints. - 5. Dispatch — `bazel run` each pending target, then `deliveryd_deliver` - (or `deliveryd_delete_artifact` on failure). + 5. Dispatch — `bazel run` each pending target, then `dd.deliver()` + (or `dd.delete()` on failure). Flags @@ -158,13 +159,7 @@ load("./private/lib/artifacts.axl", "artifacts") load("./private/lib/bazel_results.axl", "BES_DRAIN_TICK_MS", "MIN_HEARTBEAT_INTERVAL_MS", "archive_bazel_attempt", "bes_get", "now_ms", "process_event", bazel_init_data = "init_data") load("./private/lib/ci.axl", "resolve_build_url") load("./private/lib/delivery_results.axl", delivery_add_result = "add_result", delivery_build_manifest = "build_manifest", delivery_conclusion = "conclusion", delivery_init_data = "init_data") -load( - "./private/lib/deliveryd.axl", - deliveryd_delete_artifact = "delete_artifact", - deliveryd_deliver = "deliver", - deliveryd_query = "query", - deliveryd_record = "record", -) +load("./private/lib/deliveryd.axl", "deliveryd") load("./private/lib/environment.axl", "apply_output_base_suffix", "color_enabled", "error", "info", "sanitize_filename", "warn") load("./private/lib/github.axl", "detect_commit_sha") load("./private/lib/health_check.axl", "HealthCheckTrait") @@ -258,12 +253,18 @@ def fmt_elapsed(secs): return "{}.{}s".format(int(secs), int(secs * 10) % 10) return "{}m{}s".format(int(secs) // 60, int(secs) % 60) -def _print_header(color, endpoint, ci_host, commit_sha, prefix, prefix_source, build_url, expanded_flags, targets, forced_targets, mode, dry_run, track_state): +def _print_header(color, dd, endpoint, ci_host, commit_sha, prefix, prefix_source, build_url, expanded_flags, targets, forced_targets, mode, dry_run, track_state): print(ansi.style("Delivery:", ansi.BOLD, color)) mode_str = mode + (" (dry-run)" if dry_run else "") print(" {}: {}".format(ansi.style("Mode", ansi.BOLD, color), mode_str)) if track_state: - print(" {}: {}".format(ansi.style("State", ansi.BOLD, color), endpoint)) + print(" {}: {} (API v{})".format(ansi.style("State", ansi.BOLD, color), endpoint, dd.version)) + + # The daemon's own storage stats, off the negotiation probe — a v1 + # daemon reports none, so the line is dropped rather than shown empty. + storage = deliveryd.describe_storage(dd.status) + if storage: + print(" {}: {}".format(ansi.style("Storage", ansi.BOLD, color), storage)) else: print(" {}: (untracked — --track-state=false)".format(ansi.style("State", ansi.BOLD, color))) print(" {}: {}".format(ansi.style("Host", ansi.BOLD, color), ci_host)) @@ -1146,10 +1147,10 @@ def _delivery_impl(ctx): preview_only = dry_run and not track_state # deliveryd state backend; Workflows runners expose its Unix socket here. - endpoint = ctx.std.env.var("ASPECT_WORKFLOWS_DELIVERY_API_ENDPOINT") + endpoint = ctx.std.env.var(deliveryd.ENDPOINT_ENV) if not endpoint and track_state: _emit_final(ph, exit_code = 1) - fail("ASPECT_WORKFLOWS_DELIVERY_API_ENDPOINT is not set. The delivery state backend must be running. (Pass --track-state=false to run without state tracking; allowed in combination with --dry-run or --mode=always.)") + fail("{} is not set. The delivery state backend must be running. (Pass --track-state=false to run without state tracking; allowed in combination with --dry-run or --mode=always.)".format(deliveryd.ENDPOINT_ENV)) # commit-sha / build-url auto-detect from CI env. commit_sha is only # consumed under track_state, so fail only when both apply. @@ -1181,6 +1182,23 @@ def _delivery_impl(ctx): # identical across salt namespaces, hiding the per-salt separation. data["delivery"]["prefix"] = prefix + # The state backend. Constructing it negotiates the deliveryd API version + # and probes reachability, so a dead daemon fails here rather than midway + # through the first record. + dd = None + if track_state: + dd = deliveryd.new( + ctx, + host = ci_host, + prefix = prefix, + commit = commit_sha, + signature = build_url, + endpoint = endpoint, + ) + if not dd.status.ok: + error(ctx.std, "Delivery state backend is unavailable at {}: {}".format(dd.endpoint, dd.status.error)) + return _emit_final(ph, exit_code = 1) + # iter handles are created per attempt inside the retry loop. The # CLI-streamed `--bes-backend` sinks stream phase-1 events and are waited # alongside the trait sinks (see `_get_output_shas`). Announced under the @@ -1289,7 +1307,7 @@ def _delivery_impl(ctx): _emit_final(ph, exit_code = 1) fail(msg) - _print_header(color, endpoint, ci_host, commit_sha, prefix, prefix_source, build_url, rc.expand(command = "build"), targets, forced_targets, mode, dry_run, track_state) + _print_header(color, dd, endpoint, ci_host, commit_sha, prefix, prefix_source, build_url, rc.expand(command = "build"), targets, forced_targets, mode, dry_run, track_state) # `rc=` opts into the runinfo aspect so the deliverables spawned in phase 3 # replay their `args` attribute + env the way `bazel run` does. The aspect @@ -1353,7 +1371,7 @@ def _delivery_impl(ctx): ) for label in targets: if label in output_shas: - deliveryd_record(ctx, endpoint, ci_host, commit_sha, prefix, label, output_shas[label]) + dd.record(label, output_shas[label]) # Change detection: query deliveryd for prior-delivery state, filtered # below to changed targets. Skipped when --track-state=false (no endpoint) @@ -1370,9 +1388,9 @@ def _delivery_impl(ctx): data = data, phase = phases.Phase(name = "detect", description = "Detect already-delivered targets", emoji = "🔬"), ) - delivery_state = deliveryd_query(ctx, endpoint, ci_host, commit_sha, prefix) + delivered = dd.query(output_shas) else: - delivery_state = {} + delivered = {} # Phase 3: fused download + delivery. Bazel lands a target's outputs + # runfiles on disk before its `target_completed` fires (current-Bazel @@ -1386,7 +1404,7 @@ def _delivery_impl(ctx): for label in targets if output_shas.get(label) and ( label in forced_targets or - not (delivery_state.get(label) and delivery_state.get(label).get("delivered")) + label not in delivered ) ] @@ -1422,7 +1440,6 @@ def _delivery_impl(ctx): for label in targets: # --mode=always behaves as if every target was --force-target'd. is_forced = label in forced_targets or mode == "always" - target_state = delivery_state.get(label) output_sha = output_shas.get(label) if not output_sha and phase_1_2_runs: @@ -1453,9 +1470,9 @@ def _delivery_impl(ctx): rows.append((label, "WARN", "warn", msg, "-", "-")) delivery_add_result(data, label, "warn", message = msg, is_forced = is_forced) _fire_target_hook(data, delivery_trait, run_tracker, label) - elif not is_forced and target_state and target_state.get("delivered"): + elif not is_forced and label in delivered: skipped_count += 1 - delivered_by = target_state.get("delivered_by") or "-" + delivered_by = delivered[label] or "-" rows.append((label, "SKIP", "skip", delivered_by, output_sha, "-")) delivery_add_result( data, @@ -1672,13 +1689,13 @@ def _delivery_impl(ctx): forced_marker = " (FORCED)" if is_forced else "" if exit_code == 0: if track_state: - deliveryd_deliver(ctx, endpoint, ci_host, output_sha, prefix, build_url) + dd.deliver(label, output_sha) success_count += 1 rows.append((label, "OK" + forced_marker, "ok", build_url, output_sha, duration)) delivery_add_result(data, label, "ok", message = build_url, output_sha = output_sha, is_forced = is_forced) else: if track_state: - deliveryd_delete_artifact(ctx, endpoint, ci_host, output_sha, prefix) + dd.delete(output_sha) failed_count += 1 rows.append((label, "FAIL" + forced_marker, "fail", "-", output_sha, duration)) delivery_add_result(data, label, "fail", message = "exit {}".format(exit_code), output_sha = output_sha, is_forced = is_forced) diff --git a/crates/aspect-cli/src/builtins/aspect/private/lib/deliveryd.axl b/crates/aspect-cli/src/builtins/aspect/private/lib/deliveryd.axl index 5c5cd7c65..55121eb41 100644 --- a/crates/aspect-cli/src/builtins/aspect/private/lib/deliveryd.axl +++ b/crates/aspect-cli/src/builtins/aspect/private/lib/deliveryd.axl @@ -1,140 +1,342 @@ -""" -Client library for communicating with deliveryd. +"""Client for `deliveryd`, the delivery-state daemon — speaks v2, falls back to v1. + +deliveryd tracks which artifacts have already been delivered so a re-run can skip +them. v2 keys that state by artifact digest and answers one batched +`QUERY /delivery`; v1 keys it by commit and needs a per-target `record` before +`POST /query` can answer. `new()` negotiates the dialect once, off the +`DeliveryD-Version` header on `/health`, and every verb below speaks whichever +one the daemon advertises — callers write v2-shaped code and never branch on the +version. -deliveryd is a Unix socket HTTP server that manages delivery state, -tracking which artifacts have been delivered and preventing re-delivery. + dd = deliveryd.new(ctx, host = "gh", prefix = ctx.task.name, commit = sha, signature = build_url) + delivered = dd.query(digests) # {label: digest} -> {label: signature} + dd.deliver(label, digest) # after the run succeeded + dd.delete(digest) # after it failed """ -def parse_endpoint(endpoint): - """ - Parse a deliveryd endpoint string. +load("./text.axl", "format_bytes_iec") - Accepts: - - unix:///path/to/socket → unix socket - - http://host:port → plain HTTP - - https://host:port → HTTPS - """ +# Where an Aspect Workflows runner publishes the daemon's Unix socket. +ENDPOINT_ENV = "ASPECT_WORKFLOWS_DELIVERY_API_ENDPOINT" + +# A daemon that answers /health without advertising a version predates v2 and +# speaks the `record` + commit-keyed `query` + `deliver` + `artifact/delete` API. +_V1 = 1 +_V2 = 2 + +# hyper and reqwest lower-case header names; the daemon spells it DeliveryD-Version. +_VERSION_HEADER = "deliveryd-version" + +_JSON_HEADERS = {"Content-Type": "application/json"} + +# Usable before the version is negotiated — the /health probe rides it. +_Conn = record( + http = field(typing.Any), + base_url = field(str), + socket = field(str | None), + endpoint = field(str), +) + +# `status = 0` means the request never completed (DNS, a missing socket, a +# refused connection); `body` then holds the transport error. +_Response = record( + status = field(int), + body = field(str), + headers = field(list), +) + +# Everything a delivery run holds constant: the connection, the negotiated +# dialect, the state key (`host`/`prefix`) and the provenance stored alongside a +# delivery (`commit`/`signature`). The verbs thread it, so no call re-passes ctx. +_Client = record( + conn = field(_Conn), + version = field(int), + host = field(str), + prefix = field(str), + commit = field(str), + signature = field(str), +) + +# The daemon's view of its backing store, as reported by /health. Absent on a +# v1 daemon, whose /health carries no body. +Storage = record( + reachable = field(bool, default = False), + ping_ms = field(int, default = 0), + used_bytes = field(int, default = 0), + max_bytes = field(int, default = 0), + used_pct = field(float, default = 0.0), + policy = field(str, default = ""), +) + +Health = record( + ok = field(bool), + version = field(int), + status = field(str, default = ""), + storage = field(Storage, default = Storage()), + error = field(str, default = ""), +) + +def _format_pct(pct: float) -> str: + tenths = int(pct * 10) + return "{}.{}%".format(tenths // 10, tenths % 10) + +def describe_storage(health: Health) -> str: + """The storage half of `/health` as one line, or "" when the daemon reported + none (a v1 daemon, or an unreachable one).""" + s = health.storage + if not s.reachable: + return "" + parts = [] + if s.max_bytes: + parts.append("{} / {} ({})".format(format_bytes_iec(s.used_bytes), format_bytes_iec(s.max_bytes), _format_pct(s.used_pct))) + elif s.used_bytes: + parts.append(format_bytes_iec(s.used_bytes) + " used") + if s.policy: + parts.append(s.policy) + parts.append("ping {}ms".format(s.ping_ms)) + return " · ".join(parts) + +def _parse_endpoint(endpoint: str) -> tuple: + """`(base_url, socket_path)` for `unix:///path`, `http://host:port` or + `https://host:port`. `socket_path` is None for the plain-HTTP forms.""" if endpoint.startswith("unix://"): return ("http://localhost", endpoint[len("unix://"):]) if endpoint.startswith("http://") or endpoint.startswith("https://"): return (endpoint, None) - msg = "Invalid deliveryd endpoint '{}': must start with unix://, http://, or https://".format(endpoint) - fail(msg) - -def _post(http, endpoint, path, data): - """Make a POST request to deliveryd, handling unix:// endpoints.""" - base_url, socket_path = parse_endpoint(endpoint) - encoded = json.encode(data) - if socket_path: - return http.post(url = base_url + path, headers = {"Content-Type": "application/json"}, data = encoded, unix_socket = socket_path) - return http.post(url = base_url + path, headers = {"Content-Type": "application/json"}, data = encoded) - -def _get(http, endpoint, path): - """Make a GET request to deliveryd, handling unix:// endpoints.""" - base_url, socket_path = parse_endpoint(endpoint) - if socket_path: - return http.get(url = base_url + path, unix_socket = socket_path) - return http.get(url = base_url + path) - -def health(ctx, endpoint): - """ - Check if deliveryd is healthy by calling the /health endpoint. + fail("Invalid deliveryd endpoint '{}': must start with unix://, http://, or https://".format(endpoint)) - Returns: - True if healthy, False otherwise - """ - http = ctx.http() - result = _get(http, endpoint, "/health").map_err(lambda e: str(e)).block() - if type(result) == "string": - return False - return result.status >= 200 and result.status < 300 +def _send(conn: _Conn, method: str, path: str, payload) -> _Response: + """One request, with transport errors flattened into `status = 0` rather + than raised — every caller wants to name the operation in the message.""" + url = conn.base_url + path + data = json.encode(payload) if payload != None else None + if method == "GET": + fut = conn.http.get(url = url, headers = _JSON_HEADERS, unix_socket = conn.socket) + elif method == "QUERY": + fut = conn.http.query(url = url, headers = _JSON_HEADERS, data = data, unix_socket = conn.socket) + elif method == "DELETE": + fut = conn.http.delete(url = url, headers = _JSON_HEADERS, data = data, unix_socket = conn.socket) + else: + fut = conn.http.post(url = url, headers = _JSON_HEADERS, data = data, unix_socket = conn.socket) -def query(ctx, endpoint, ci_host, commit_sha, workspace): - """ - Query deliveryd for delivery state of all targets in a commit. - Returns a dict mapping label -> {output_sha, delivered, delivered_by}. - """ - http = ctx.http() - response = _post(http, endpoint, "/query", { - "ci_host": ci_host, - "commit_sha": commit_sha, - "workspace": workspace, - }).block() - - if response.status < 200 or response.status >= 300: - msg = "deliveryd query failed: " + response.body - fail(msg) - - # try_decode + fail with a clean message — protects `aspect delivery` - # from a corrupted/early-truncated 2xx response from deliveryd - # (rare, but the daemon shares state between many concurrent - # deliveries, so a defensive decode here costs nothing). - data = json.try_decode(response.body, None) + res = fut.map_err(lambda e: str(e)).block() + if type(res) == "string": + return _Response(status = 0, body = res, headers = []) + return _Response(status = res.status, body = res.body, headers = res.headers) + +def _ok(res: _Response) -> bool: + return res.status >= 200 and res.status < 300 + +def _header(res: _Response, name: str) -> str: + for pair in res.headers: + if pair[0].lower() == name: + return pair[1] + return "" + +def _check(conn: _Conn, op: str, res: _Response): + if _ok(res): + return + reason = "transport error: " + res.body if res.status == 0 else "HTTP {}: {}".format(res.status, res.body.strip()) + fail("deliveryd {} failed ({}): {}".format(op, conn.endpoint, reason)) + +def _decode(conn: _Conn, op: str, res: _Response) -> dict: + """Decode a 2xx JSON body. The daemon multiplexes state across concurrent + deliveries, so a truncated body is rare but must not die as a type error.""" + data = json.try_decode(res.body, None) if data == None: - fail("deliveryd /query returned 2xx but body was not valid JSON: " + (response.body or "")[:200].replace("\n", " ")) + fail("deliveryd {} returned 2xx but body was not valid JSON: {}".format(op, res.body[:200].replace("\n", " "))) + return data - targets = data.get("targets", []) or [] +def _negotiate(conn: _Conn) -> tuple: + """`(version, health_response)` from one `/health` probe.""" + res = _send(conn, "GET", "/health", None) + if res.status == 0: + return (_V1, res) + advertised = _header(res, _VERSION_HEADER) + if not advertised.isdigit(): + return (_V1, res) + return (max(_V1, int(advertised)), res) - # Build lookup dict by label - result = {} - for target in targets: - result[target["label"]] = { - "output_sha": target["output_sha"], - "delivered": target["delivered"], - "delivered_by": target.get("delivered_by"), - } - return result +def _to_health(res: _Response, version: int) -> Health: + if res.status == 0: + return Health(ok = False, version = version, error = res.body) + body = json.try_decode(res.body, {}) + if type(body) != "dict": + body = {} + return Health( + ok = _ok(res), + version = version, + status = body.get("status", "") or "", + storage = _to_storage(body.get("storage", {}) or {}), + error = "" if _ok(res) else "HTTP {}: {}".format(res.status, res.body.strip()), + ) -def deliver(ctx, endpoint, ci_host, output_sha, workspace, build_url): - """ - Mark a target as delivered. build_url is stored as the delivery signature - and identifies the CI build that performed the delivery. - """ - http = ctx.http() - response = _post(http, endpoint, "/deliver", { - "ci_host": ci_host, - "output_sha": output_sha, - "workspace": workspace, - "signature": build_url, - }).block() - - if response.status < 200 or response.status >= 300: - msg = "deliveryd deliver failed: " + response.body - fail(msg) - -def record(ctx, endpoint, ci_host, commit_sha, workspace, label, output_sha): - """ - Record a target's output SHA with deliveryd. - This must be called before the target can be queried or delivered. - """ - http = ctx.http() - response = _post(http, endpoint, "/record", { - "ci_host": ci_host, - "commit_sha": commit_sha, - "workspace": workspace, +def _to_storage(s: dict) -> Storage: + # Sizes are coerced: JSON gives an int for a whole percentage and the record + # fields are strictly typed. + if not s.get("reachable", False): + return Storage(reachable = False) + return Storage( + reachable = True, + ping_ms = int(s.get("ping_ms", 0) or 0), + used_bytes = int(s.get("used_memory_bytes", 0) or 0), + max_bytes = int(s.get("maxmemory_bytes", 0) or 0), + used_pct = float(s.get("used_memory_pct", 0) or 0), + policy = s.get("maxmemory_policy", "") or "", + ) + +def _do_health(c: _Client) -> Health: + """Re-probe: reachability, the daemon's view of its storage, and the version + it advertises now.""" + (version, res) = _negotiate(c.conn) + return _to_health(res, version) + +def _do_record(c: _Client, label: str, digest: str) -> None: + """Register `(label, digest)` under the commit so `query`/`deliver` can find + it. v2 keys state by digest alone — the client already holds every digest — + so there is nothing to record and this is a no-op.""" + if c.version >= _V2: + return + res = _send(c.conn, "POST", "/record", { + "ci_host": c.host, + "commit_sha": c.commit, + "workspace": c.prefix, "label": label, - "output_sha": output_sha, - }).map_err(lambda e: e).block() + "output_sha": digest, + }) + _check(c.conn, "record", res) - if type(response) == "string": - msg = "deliveryd record failed: " + response - fail(msg) +def _do_query(c: _Client, digests: dict) -> dict: + """Which of `{label: digest}` are already delivered → `{label: signature}`, + the signature being the build URL that delivered it. Labels absent from the + result are not delivered; labels with an empty digest are never asked about.""" + if c.version >= _V2: + return _query_v2(c, digests) + return _query_v1(c, digests) - if response.status < 200 or response.status >= 300: - msg = "deliveryd record failed: " + response.body - fail(msg) +def _query_v2(c: _Client, digests: dict) -> dict: + # Deduped (two labels can share a digest) in the caller's order. + wanted = {digest: True for digest in digests.values() if digest} + if not wanted: + return {} + res = _send(c.conn, "QUERY", "/delivery", { + "host": c.host, + "prefix": c.prefix, + "digests": list(wanted.keys()), + }) + _check(c.conn, "query", res) + delivered = _decode(c.conn, "query", res).get("delivered", {}) or {} + return { + label: delivered[digest] + for label, digest in digests.items() + if digest and digest in delivered + } -def delete_artifact(ctx, endpoint, ci_host, output_sha, workspace): - """ - Delete artifact metadata (used for cleanup on failed deliveries). +def _query_v1(c: _Client, digests: dict) -> dict: + # v1 answers for the whole commit, keyed by label, and only for targets a + # prior `record` registered; narrowed to the asked-about labels so both + # dialects return the same shape. + res = _send(c.conn, "POST", "/query", { + "ci_host": c.host, + "commit_sha": c.commit, + "workspace": c.prefix, + }) + _check(c.conn, "query", res) + delivered = {} + for target in _decode(c.conn, "query", res).get("targets", []) or []: + label = target.get("label", "") + if label in digests and target.get("delivered"): + delivered[label] = target.get("delivered_by") or "" + return delivered + +def _do_deliver(c: _Client, label: str, digest: str, meta: dict = {}) -> None: + """Record that `label`'s delivery run succeeded. Call after the run — this + persists the outcome, it does not perform the delivery.""" + if c.version >= _V2: + payload = { + "host": c.host, + "prefix": c.prefix, + "digest": digest, + "label": label, + "commit": c.commit, + "signature": c.signature, + } + + # Opaque enrichment the daemon stores verbatim; omitted when empty. + if meta: + payload["meta"] = meta + else: + payload = { + "ci_host": c.host, + "output_sha": digest, + "workspace": c.prefix, + "signature": c.signature, + } + res = _send(c.conn, "POST", "/delivery" if c.version >= _V2 else "/deliver", payload) + _check(c.conn, "deliver", res) + +def _do_delete(c: _Client, digest: str) -> None: + """Drop a digest's state after a failed delivery, so the next run retries it + instead of reading the failure as "already delivered".""" + if c.version >= _V2: + res = _send(c.conn, "DELETE", "/delivery", { + "host": c.host, + "prefix": c.prefix, + "digest": digest, + }) + else: + res = _send(c.conn, "POST", "/artifact/delete", { + "ci_host": c.host, + "output_sha": digest, + "workspace": c.prefix, + }) + _check(c.conn, "delete", res) + +def _new(ctx, host: str, prefix: str, commit: str = "", signature: str = "", endpoint: str = ""): + """Connect to deliveryd and negotiate the API version, once. + + `host` is the CI host label ("gh", "bk", …) and `prefix` the change-detection + namespace; together they key the state. `commit` and `signature` (the build + URL) are the provenance stored with each delivery — `commit` is also the + state key on a v1 daemon. `endpoint` defaults to `ENDPOINT_ENV`. + + The negotiation probe doubles as a reachability check, reported through + `.status` rather than raised — a dead daemon is a startup error the caller + concludes its own surface for, not a mid-run write failure. """ - http = ctx.http() - response = _post(http, endpoint, "/artifact/delete", { - "ci_host": ci_host, - "output_sha": output_sha, - "workspace": workspace, - }).block() - - if response.status < 200 or response.status >= 300: - msg = "deliveryd artifact delete failed: " + response.body - fail(msg) + endpoint = endpoint or ctx.std.env.var(ENDPOINT_ENV) + if not endpoint: + fail("deliveryd endpoint is not set; expected {} in the environment.".format(ENDPOINT_ENV)) + + (base_url, socket) = _parse_endpoint(endpoint) + conn = _Conn(http = ctx.http(), base_url = base_url, socket = socket, endpoint = endpoint) + (version, res) = _negotiate(conn) + + c = _Client( + conn = conn, + version = version, + host = host, + prefix = prefix, + commit = commit, + signature = signature, + ) + return struct( + endpoint = endpoint, + version = version, + status = _to_health(res, version), + health = lambda: _do_health(c), + record = lambda label, digest: _do_record(c, label, digest), + query = lambda digests: _do_query(c, digests), + deliver = lambda label, digest, meta = {}: _do_deliver(c, label, digest, meta), + delete = lambda digest: _do_delete(c, digest), + ) + +deliveryd = namespace( + new = _new, + describe_storage = describe_storage, + ENDPOINT_ENV = ENDPOINT_ENV, + Health = Health, + Storage = Storage, +) + +testonly_parse_endpoint = _parse_endpoint diff --git a/crates/aspect-cli/src/builtins/aspect/private/lib/deliveryd_test.axl b/crates/aspect-cli/src/builtins/aspect/private/lib/deliveryd_test.axl new file mode 100644 index 000000000..c0d941124 --- /dev/null +++ b/crates/aspect-cli/src/builtins/aspect/private/lib/deliveryd_test.axl @@ -0,0 +1,289 @@ +"""Unit tests for lib/deliveryd.axl. + +A struct-based fake `ctx` (like `deployment_flags_test.axl`) serves canned HTTP +responses keyed by `" "` and records every request, so the two +dialects can be asserted on the wire: which endpoint was called, with which +field names, and — for v2 — that `record` sends nothing at all. + +`fail()` cannot be caught in Starlark, so the error paths (non-2xx, corrupt JSON) +are not exercised here; an unreachable daemon is, since that surfaces through +`.status` instead of failing. +""" + +load("./deliveryd.axl", "deliveryd", parse_endpoint = "testonly_parse_endpoint") + +_SOCKET = "unix:///run/deliveryd.sock" +_BASE = "http://localhost" + +_V2_HEALTH = { + "status": "ok", + "version": "2", + "storage": { + "reachable": True, + "ping_ms": 1, + "used_memory_bytes": 3221225472, + "maxmemory_bytes": 55834574848, + "used_memory_pct": 5.77, + "maxmemory_policy": "noeviction", + }, +} + +def _eq(label, got, want): + if got != want: + fail("%s: got %r, want %r" % (label, got, want)) + +def _res(status, body = "", headers = []): + return struct(status = status, body = body if type(body) == "string" else json.encode(body), headers = headers) + +def _future(res): + return struct(map_err = lambda f: struct(block = lambda: res), block = lambda: res) + +def _error_future(message): + return struct(map_err = lambda f: struct(block = lambda: f(message)), block = lambda: message) + +def _health_res(version = "2"): + headers = [("deliveryd-version", version)] if version else [] + return _res(200, _V2_HEALTH, headers) + +def _ctx(responses, calls): + """Fake ctx whose http serves `responses` ({" ": res | None}, + None meaning a transport error) and appends each request to `calls`.""" + + def respond(method, url, data): + path = url[len(_BASE):] + key = method + " " + path + calls.append(struct(method = method, path = path, body = json.try_decode(data, None) if data else None)) + if key not in responses: + fail("unexpected deliveryd request: " + key) + res = responses[key] + if res == None: + return _error_future("connection refused") + return _future(res) + + http = struct( + get = lambda url, headers = {}, unix_socket = None: respond("GET", url, None), + query = lambda url, headers = {}, data = None, unix_socket = None: respond("QUERY", url, data), + post = lambda url, headers = {}, data = None, unix_socket = None: respond("POST", url, data), + delete = lambda url, headers = {}, data = None, unix_socket = None: respond("DELETE", url, data), + ) + return struct( + http = lambda: http, + std = struct(env = struct(var = lambda name: _SOCKET)), + ) + +def _client(responses, calls, **kwargs): + return deliveryd.new( + _ctx(responses, calls), + host = kwargs.pop("host", "gh"), + prefix = kwargs.pop("prefix", "deliver"), + commit = kwargs.pop("commit", "abc123"), + signature = kwargs.pop("signature", "https://ci/build/7"), + **kwargs + ) + +def _test_parse_endpoint(ctx): + (base, sock) = parse_endpoint("unix:///run/deliveryd.sock") + _eq("unix → base", base, "http://localhost") + _eq("unix → socket", sock, "/run/deliveryd.sock") + + (base, sock) = parse_endpoint("http://deliveryd.local:8080") + _eq("http → base passthrough", base, "http://deliveryd.local:8080") + _eq("http → no socket", sock, None) + + (base, sock) = parse_endpoint("https://deliveryd.example.com") + _eq("https → base passthrough", base, "https://deliveryd.example.com") + _eq("https → no socket", sock, None) + +def _test_negotiates_from_health_header(ctx): + calls = [] + dd = _client({"GET /health": _health_res("2")}, calls) + _eq("version header → v2", dd.version, 2) + _eq("negotiation probes /health once", [c.path for c in calls], ["/health"]) + _eq("probe fills .status", dd.status.ok, True) + _eq("probe reads storage", dd.status.storage.reachable, True) + + # A daemon that answers /health without the header predates v2. + dd = _client({"GET /health": _res(200, {"status": "ok"})}, []) + _eq("no version header → v1", dd.version, 1) + + # Endpoint defaults to the env var when not passed explicitly. + _eq("endpoint from env", dd.endpoint, _SOCKET) + +def _test_unreachable_reports_status(ctx): + dd = _client({"GET /health": None}, []) + _eq("unreachable → not ok", dd.status.ok, False) + _eq("unreachable → error carries the transport reason", dd.status.error, "connection refused") + +def _test_v2_record_is_a_noop(ctx): + calls = [] + dd = _client({"GET /health": _health_res()}, calls) + dd.record("//pkg:a", "digest-a") + _eq("v2 record sends nothing", [c.path for c in calls], ["/health"]) + +def _test_v1_record_posts_legacy_fields(ctx): + calls = [] + dd = _client({"GET /health": _res(200, {}), "POST /record": _res(200, "ok")}, calls) + dd.record("//pkg:a", "digest-a") + _eq("v1 record → POST /record", calls[-1].path, "/record") + _eq("v1 record payload", calls[-1].body, { + "ci_host": "gh", + "commit_sha": "abc123", + "workspace": "deliver", + "label": "//pkg:a", + "output_sha": "digest-a", + }) + +def _test_v2_query_batches_and_maps_back(ctx): + calls = [] + dd = _client({ + "GET /health": _health_res(), + "QUERY /delivery": _res(200, {"delivered": {"digest-a": "https://ci/build/1"}}), + }, calls) + + # Two labels share a digest; a third has none and is never asked about. + got = dd.query({"//pkg:a": "digest-a", "//pkg:b": "digest-a", "//pkg:c": "digest-c", "//pkg:d": ""}) + _eq("v2 query → one QUERY /delivery", [c.path for c in calls], ["/health", "/delivery"]) + _eq("v2 query method", calls[-1].method, "QUERY") + _eq("v2 query payload", calls[-1].body, { + "host": "gh", + "prefix": "deliver", + "digests": ["digest-a", "digest-c"], + }) + _eq("delivered digest maps back to every label holding it", got, { + "//pkg:a": "https://ci/build/1", + "//pkg:b": "https://ci/build/1", + }) + +def _test_v2_query_skips_the_round_trip_when_no_digests(ctx): + calls = [] + dd = _client({"GET /health": _health_res()}, calls) + _eq("no digests → empty", dd.query({"//pkg:a": ""}), {}) + _eq("no digests → no request", [c.path for c in calls], ["/health"]) + +def _test_v1_query_narrows_to_delivered_labels(ctx): + calls = [] + dd = _client({ + "GET /health": _res(200, {}), + "POST /query": _res(200, {"targets": [ + {"label": "//pkg:a", "output_sha": "digest-a", "delivered": True, "delivered_by": "https://ci/build/1"}, + {"label": "//pkg:b", "output_sha": "digest-b", "delivered": False}, + {"label": "//pkg:gone", "output_sha": "digest-x", "delivered": True, "delivered_by": "https://ci/build/0"}, + ]}), + }, calls) + + got = dd.query({"//pkg:a": "digest-a", "//pkg:b": "digest-b"}) + _eq("v1 query payload", calls[-1].body, {"ci_host": "gh", "commit_sha": "abc123", "workspace": "deliver"}) + _eq("v1 query → only delivered, only asked-about", got, {"//pkg:a": "https://ci/build/1"}) + +def _test_v2_deliver_and_delete(ctx): + calls = [] + dd = _client({ + "GET /health": _health_res(), + "POST /delivery": _res(200, "ok"), + "DELETE /delivery": _res(200, "ok"), + }, calls) + + dd.deliver("//pkg:a", "digest-a") + _eq("v2 deliver → POST /delivery", (calls[-1].method, calls[-1].path), ("POST", "/delivery")) + _eq("v2 deliver payload", calls[-1].body, { + "host": "gh", + "prefix": "deliver", + "digest": "digest-a", + "label": "//pkg:a", + "commit": "abc123", + "signature": "https://ci/build/7", + }) + + dd.deliver("//pkg:a", "digest-a", meta = {"image": "sha256:beef"}) + _eq("meta rides along when set", calls[-1].body["meta"], {"image": "sha256:beef"}) + + dd.delete("digest-a") + _eq("v2 delete → DELETE /delivery", (calls[-1].method, calls[-1].path), ("DELETE", "/delivery")) + _eq("v2 delete payload", calls[-1].body, {"host": "gh", "prefix": "deliver", "digest": "digest-a"}) + +def _test_v1_deliver_and_delete(ctx): + calls = [] + dd = _client({ + "GET /health": _res(200, {}), + "POST /deliver": _res(200, "ok"), + "POST /artifact/delete": _res(200, "ok"), + }, calls) + + dd.deliver("//pkg:a", "digest-a") + _eq("v1 deliver → POST /deliver", calls[-1].path, "/deliver") + _eq("v1 deliver payload", calls[-1].body, { + "ci_host": "gh", + "output_sha": "digest-a", + "workspace": "deliver", + "signature": "https://ci/build/7", + }) + + dd.delete("digest-a") + _eq("v1 delete → POST /artifact/delete", calls[-1].path, "/artifact/delete") + _eq("v1 delete payload", calls[-1].body, {"ci_host": "gh", "output_sha": "digest-a", "workspace": "deliver"}) + +def _test_health_reprobes(ctx): + calls = [] + dd = _client({"GET /health": _health_res()}, calls) + health = dd.health() + _eq("health re-probes", [c.path for c in calls], ["/health", "/health"]) + _eq("health ok", health.ok, True) + _eq("health version", health.version, 2) + _eq("health status", health.status, "ok") + _eq("health storage reachable", health.storage.reachable, True) + _eq("health storage used", health.storage.used_bytes, 3221225472) + _eq("health storage max", health.storage.max_bytes, 55834574848) + _eq("health storage policy", health.storage.policy, "noeviction") + +def _test_describe_storage(ctx): + dd = _client({"GET /health": _health_res()}, []) + _eq( + "full stats render as one line", + deliveryd.describe_storage(dd.status), + "3.0 GiB / 52.0 GiB (5.7%) · noeviction · ping 1ms", + ) + + # No maxmemory configured: no ratio and no percentage to report. + unbounded = dict(_V2_HEALTH) + unbounded["storage"] = {"reachable": True, "ping_ms": 2, "used_memory_bytes": 1536} + dd = _client({"GET /health": _res(200, unbounded, [("deliveryd-version", "2")])}, []) + _eq("no maxmemory → used only", deliveryd.describe_storage(dd.status), "1.5 KiB used · ping 2ms") + + # A daemon reporting reachability but no memory stats: no size clause at all. + bare = dict(_V2_HEALTH) + bare["storage"] = {"reachable": True, "ping_ms": 3} + dd = _client({"GET /health": _res(200, bare, [("deliveryd-version", "2")])}, []) + _eq("no memory stats → ping only", deliveryd.describe_storage(dd.status), "ping 3ms") + + # A v1 daemon reports no storage at all; the caller drops the line. + dd = _client({"GET /health": _res(200, {})}, []) + _eq("no storage → empty", deliveryd.describe_storage(dd.status), "") + +_UNIT_TESTS = [ + _test_parse_endpoint, + _test_negotiates_from_health_header, + _test_unreachable_reports_status, + _test_v2_record_is_a_noop, + _test_v1_record_posts_legacy_fields, + _test_v2_query_batches_and_maps_back, + _test_v2_query_skips_the_round_trip_when_no_digests, + _test_v1_query_narrows_to_delivered_labels, + _test_v2_deliver_and_delete, + _test_v1_deliver_and_delete, + _test_health_reprobes, + _test_describe_storage, +] + +def _test_impl(ctx): + for t in _UNIT_TESTS: + t(ctx) + print("deliveryd.axl: OK (%d tests)" % len(_UNIT_TESTS)) + return 0 + +deliveryd_tests = task( + summary = "Run the deliveryd AXL unit tests.", + kind = "test-deliveryd", + group = ["dev"], + implementation = _test_impl, + args = {}, +) diff --git a/crates/aspect-cli/src/builtins/aspect/private/lib/text.axl b/crates/aspect-cli/src/builtins/aspect/private/lib/text.axl index a5f657424..a1bab8e0d 100644 --- a/crates/aspect-cli/src/builtins/aspect/private/lib/text.axl +++ b/crates/aspect-cli/src/builtins/aspect/private/lib/text.axl @@ -1,6 +1,7 @@ """Text-formatting utilities shared across the aspect-cli builtins. Currently: + - `format_bytes_iec(n)` — byte counts in IEC units, truncated to tenths. - `format_int_comma(n)` — thousands separators on integers. - `pluralize(count, singular, plural=None)` — `"N noun"` / `"N nouns"` with thousands-separated count. @@ -13,6 +14,27 @@ it's wanted by more than one module and not specific to any one domain (bazel results, lint results, etc.). """ +_IEC_UNITS = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"] + +def format_bytes_iec(n): + """Format bytes in IEC units (`1536` → `1.5 KiB`). + + Values below 1 KiB stay as whole bytes. Larger values are truncated to one + decimal place so status output remains stable without float rounding. + """ + if n < 1024: + return "{} B".format(n) + + # Scaled by 10 throughout so the tenth survives integer division. + scaled = n * 10 + unit = 0 + for _ in range(len(_IEC_UNITS) - 1): + if scaled < 10240: + break + scaled = scaled // 1024 + unit += 1 + return "{}.{} {}".format(scaled // 10, scaled % 10, _IEC_UNITS[unit]) + def format_int_comma(n): """Format an integer with thousands separators: 12345 → "12,345".""" if n < 0: diff --git a/crates/axl-runtime/src/engine/http.rs b/crates/axl-runtime/src/engine/http.rs index 87ce345b4..282138b54 100644 --- a/crates/axl-runtime/src/engine/http.rs +++ b/crates/axl-runtime/src/engine/http.rs @@ -60,6 +60,10 @@ const DEFAULT_HTTP_READ_TIMEOUT: std::time::Duration = std::time::Duration::from /// the read window to expire. const DEFAULT_HTTP_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15); +/// The `QUERY` method (RFC 10008). Neither `http::Method` nor `reqwest::Method` +/// carries a constant for it yet, so it is spelled out here. +const QUERY_METHOD: &str = "QUERY"; + impl Http { pub fn new() -> Self { Self { @@ -303,6 +307,7 @@ pub(crate) fn http_methods(registry: &mut MethodsBuilder) { #[starlark(require = named)] url: values::StringValue, #[starlark(require = named, default = UnpackDictEntries::default())] headers: UnpackDictEntries, + #[starlark(require = named, default = NoneOr::None)] data: NoneOr>, #[starlark(require = named, default = NoneOr::None)] unix_socket: NoneOr, ) -> anyhow::Result> { let url_str = url.as_str().to_string(); @@ -312,6 +317,8 @@ pub(crate) fn http_methods(registry: &mut MethodsBuilder) { .map(|(k, v)| (k.as_str().to_string(), v.as_str().to_string())) .collect(); + let body_kind = build_body_kind(data)?; + match unix_socket.into_option() { Some(socket) => { let fut = async move { @@ -324,12 +331,14 @@ pub(crate) fn http_methods(registry: &mut MethodsBuilder) { }; let uri: hyper::Uri = UnixUri::new(&socket, &path_and_query).into(); + let body_bytes = body_kind_to_bytes(body_kind).await?; + let mut req = hyper::Request::builder().method("DELETE").uri(uri); for (key, value) in &headers_vec { req = req.header(key.as_str(), value.as_str()); } let req = req - .body(Empty::::new()) + .body(Full::new(Bytes::from(body_bytes))) .map_err(|e| anyhow::anyhow!("failed to build request: {}", e))?; let res = client @@ -370,6 +379,7 @@ pub(crate) fn http_methods(registry: &mut MethodsBuilder) { for (key, value) in &headers_vec { req = req.header(key.as_str(), value.as_str()); } + req = apply_body_kind(req, body_kind).await?; let res = req.send().await?; let response = HttpResponse::from_response(res).await?; Ok(response) @@ -639,6 +649,101 @@ pub(crate) fn http_methods(registry: &mut MethodsBuilder) { } } } + + /// `QUERY` (RFC 10008): a safe, idempotent read whose parameters travel in + /// the request body rather than the URI — for reads whose input is too + /// large or too structured for a query string. + fn query<'v>( + this: values::Value<'v>, + #[starlark(require = named)] url: values::StringValue, + #[starlark(require = named, default = UnpackDictEntries::default())] + headers: UnpackDictEntries, + #[starlark(require = named, default = NoneOr::None)] data: NoneOr>, + #[starlark(require = named, default = NoneOr::None)] unix_socket: NoneOr, + ) -> anyhow::Result> { + let url_str = url.as_str().to_string(); + let headers_vec: Vec<(String, String)> = headers + .entries + .into_iter() + .map(|(k, v)| (k.as_str().to_string(), v.as_str().to_string())) + .collect(); + + // reqwest has no `Client::query` builder (its `RequestBuilder::query` + // sets URI parameters), so the method is constructed explicitly. + let method = reqwest::Method::from_str(QUERY_METHOD) + .map_err(|e| anyhow::anyhow!("invalid http method: {}", e))?; + + let body_kind = build_body_kind(data)?; + + match unix_socket.into_option() { + Some(socket) => { + let fut = async move { + let client = HyperClient::unix(); + let parsed = url::Url::parse(&url_str) + .map_err(|e| anyhow::anyhow!("invalid url: {}", e))?; + let path_and_query = match parsed.query() { + Some(q) => format!("{}?{}", parsed.path(), q), + None => parsed.path().to_owned(), + }; + let uri: hyper::Uri = UnixUri::new(&socket, &path_and_query).into(); + + let body_bytes = body_kind_to_bytes(body_kind).await?; + + let mut req = hyper::Request::builder().method(QUERY_METHOD).uri(uri); + for (key, value) in &headers_vec { + req = req.header(key.as_str(), value.as_str()); + } + let req = req + .body(Full::new(Bytes::from(body_bytes))) + .map_err(|e| anyhow::anyhow!("failed to build request: {}", e))?; + + let res = client + .request(req) + .await + .map_err(|e| anyhow::anyhow!("request failed: {}", e))?; + + let status = res.status().as_u16(); + let resp_headers: Vec<(String, String)> = res + .headers() + .iter() + .map(|(n, v)| (n.to_string(), v.to_str().unwrap_or("").to_string())) + .collect(); + let body: Bytes = res + .into_body() + .collect() + .await + .map_err(|e| anyhow::anyhow!("failed to read body: {}", e))? + .to_bytes(); + let body = String::from_utf8_lossy(&body).to_string(); + + Ok(HttpResponse { + status, + headers: resp_headers, + body, + }) + }; + Ok(StarlarkFuture::from_future(fut)) + } + None => { + let client = this + .downcast_ref_err::() + .into_anyhow_result()? + .client + .clone(); + let fut = async move { + let mut req = client.request(method, &url_str); + for (key, value) in &headers_vec { + req = req.header(key.as_str(), value.as_str()); + } + req = apply_body_kind(req, body_kind).await?; + let res = req.send().await?; + let response = HttpResponse::from_response(res).await?; + Ok(response) + }; + Ok(StarlarkFuture::from_future(fut)) + } + } + } } /// Converts a Starlark `data` parameter value into a `BodyKind` synchronously. @@ -810,3 +915,107 @@ impl FutureAlloc for HttpResponse { self.alloc_value(heap) } } + +#[cfg(all(test, unix))] +mod tests { + use std::io::{Read, Write}; + use std::os::unix::net::UnixListener; + use std::thread; + + fn find_header_end(bytes: &[u8]) -> Option { + bytes.windows(4).position(|window| window == b"\r\n\r\n") + } + + fn read_request(mut stream: std::os::unix::net::UnixStream) -> String { + let mut request = Vec::new(); + let mut content_len = None; + + loop { + let mut chunk = [0_u8; 4096]; + let read = stream.read(&mut chunk).expect("read request"); + assert!(read > 0, "client closed before completing the request"); + request.extend_from_slice(&chunk[..read]); + + if let Some(header_end) = find_header_end(&request) { + let body_start = header_end + 4; + let expected = *content_len.get_or_insert_with(|| { + let headers = String::from_utf8_lossy(&request[..header_end]); + headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().expect("content-length")) + }) + .unwrap_or(0) + }); + if request.len() >= body_start + expected { + stream + .write_all( + b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok", + ) + .expect("write response"); + return String::from_utf8(request).expect("HTTP request is UTF-8"); + } + } + } + } + + fn capture_unix_request(call: &str) -> String { + let dir = tempfile::tempdir().expect("tempdir"); + let socket = dir.path().join("http.sock"); + let listener = UnixListener::bind(&socket).expect("bind unix HTTP socket"); + let server = thread::spawn(move || { + let (stream, _) = listener.accept().expect("accept request"); + read_request(stream) + }); + + let code = format!( + r#" +def _impl(ctx): + response = ctx.http().{call}( + url = "http://localhost/delivery", + headers = {{"Content-Type": "application/json"}}, + data = "{{\"host\":\"gh\",\"digest\":\"abc\"}}", + unix_socket = {socket:?}, + ).block() + if response.status != 200 or response.body != "ok": + fail("unexpected response: {{}} {{}}".format(response.status, response.body)) + return 0 + +t = task(implementation = _impl) +"#, + socket = socket.to_string_lossy(), + ); + crate::test::eval(&code) + .run_task(0) + .expect("AXL HTTP request succeeds"); + server.join().expect("server thread") + } + + #[test] + fn query_sends_body_over_unix_socket() { + let request = capture_unix_request("query"); + assert!( + request.starts_with("QUERY /delivery HTTP/1.1\r\n"), + "{request}" + ); + assert!( + request.ends_with("{\"host\":\"gh\",\"digest\":\"abc\"}"), + "{request}" + ); + } + + #[test] + fn delete_sends_body_over_unix_socket() { + let request = capture_unix_request("delete"); + assert!( + request.starts_with("DELETE /delivery HTTP/1.1\r\n"), + "{request}" + ); + assert!( + request.ends_with("{\"host\":\"gh\",\"digest\":\"abc\"}"), + "{request}" + ); + } +}