lab/distribution
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

P4.5: remote dist ci publish --server --token

Browse Source 
The PLAN.md CI contract in remote mode. client/http_client.sx is the
temporary in-repo HTTP client shim over std.socket (connect(2) via libc
FFI), targeting http://<ipv4-or-localhost>:<port> — DNS and TLS are
loud v0 boundaries (https:// is refused; the deployment story
terminates TLS at a reverse proxy).

publish/remote.sx drives the same manifest front half as local publish,
uploads each artifact through POST /api/upload (verifying a
manifest-declared sha256 against the server-computed digest), then
publishes via POST /api/apps/<slug>/releases. The 200 response is
parsed back into a PublishOutcome so --json/human rendering reuse the
local paths; non-2xx responses pass the server's JSON error code
through report_failure. ci publish takes exactly one of --local-store
or --server (+--token); anything else is a usage error.

make test 18/18 (new: remote_publish.sx pinned acceptance against a
live distd).
This commit is contained in:
agra
2026-06-12 11:28:34 +03:00
parent e2a5150542
commit 0a6fa65c58
4 changed files with 658 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

207
src/client/http_client.sx Normal file
View File

@@ -0,0 +1,207 @@
// =====================================================================
// http_client.sx — minimal HTTP/1.1 client over `std.socket` (subplan 03,
// Slice 3): the temporary in-repo twin of the server shim in
// src/server/http.sx, liftable into the sx stdlib alongside it.
//
// Exactly what remote publish needs: POST one body to one path on one
// server, with a bearer credential, and get the status + JSON body back.
//
// `--server` URL boundary (v0, loud): `http://<ipv4-or-localhost>:<port>`.
// No DNS (getaddrinfo) and no TLS — `https://` is refused with a message
// pointing at the reverse-proxy deployment story; LAN/NAS targets are
// dotted quads or localhost.
//
// The response is read to EOF — distd always answers `Connection: close`
// — into a fixed RESP_CAP buffer (responses are JSON, never artifacts),
// then split at the blank line into status code + body.
// =====================================================================
#import "modules/std.sx";
sock :: #import "modules/std/socket.sx";
hs :: #import "../server/http.sx";
netc :: #library "c";
c_connect :: (fd: i32, addr: *sock.SockAddr, addrlen: u32) -> i32 #foreign netc "connect";
RESP_CAP :: 262144;
ClientErr :: error {
BadUrl, // not http://<ipv4-or-localhost>:<port>
Tls, // https:// — unsupported by design (proxy terminates)
Connect, // socket/connect failed
Send, // write failed mid-request
BadResponse, // EOF before a parseable status line + blank line
}
// A parsed `--server` target.
ServerUrl :: struct {
host: string = ""; // as written (for the Host: header)
addr: u32 = 0; // IPv4, network byte order
port: i64 = 80;
}
// One response, body as a view into client-allocated memory.
HttpResponse :: struct {
status: i64 = 0;
body: string = "";
}
// Dotted-quad IPv4 (or "localhost") -> sin_addr in network byte order.
ipv4_addr :: (host: string) -> ?u32 {
if host == "localhost" { return 0x0100007F; } // 127.0.0.1
octets : [4]i64 = .[ -1, -1, -1, -1 ];
oi := 0;
v : i64 = -1;
i := 0;
while i < host.len {
c := host[i];
if c == 46 { // '.'
if v < 0 or oi >= 3 { return null; }
octets[oi] = v;
oi += 1;
v = -1;
} else {
if c < 48 or c > 57 { return null; }
if v < 0 { v = 0; }
v = v * 10 + (c - 48);
if v > 255 { return null; }
}
i += 1;
}
if v < 0 or oi != 3 { return null; }
octets[3] = v;
// network byte order = a,b,c,d in memory = little-endian a|b<<8|c<<16|d<<24
return cast(u32) (octets[0] | (octets[1] << 8) | (octets[2] << 16) | (octets[3] << 24));
}
// Parse `http://<host>:<port>` (port required; host ipv4 or localhost).
parse_server_url :: (url: string) -> (ServerUrl, !ClientErr) {
HTTPS :: "https://";
HTTP :: "http://";
if url.len >= 8 {
h := string.{ ptr = url.ptr, len = 8 };
if h == HTTPS { raise error.Tls; }
}
if url.len <= 7 { raise error.BadUrl; }
p := string.{ ptr = url.ptr, len = 7 };
if p != HTTP { raise error.BadUrl; }
rest := string.{ ptr = @url[7], len = url.len - 7 };
// split host:port (no path component allowed)
ci := -1;
i := 0;
while i < rest.len {
if rest[i] == 47 { raise error.BadUrl; } // '/' — no path in --server
if rest[i] == 58 { ci = i; } // ':'
i += 1;
}
if ci <= 0 or ci >= rest.len - 1 { raise error.BadUrl; }
host := string.{ ptr = rest.ptr, len = ci };
pstr := string.{ ptr = @rest[ci + 1], len = rest.len - ci - 1 };
port : i64 = 0;
j := 0;
while j < pstr.len {
c := pstr[j];
if c < 48 or c > 57 { raise error.BadUrl; }
port = port * 10 + (c - 48);
j += 1;
}
if port < 1 or port > 65535 { raise error.BadUrl; }
aq := ipv4_addr(host);
if aq == null { raise error.BadUrl; }
return ServerUrl.{ host = host, addr = aq!, port = port };
}
// Write all of `data` to `fd`.
send_all :: (fd: i32, data: string) -> !ClientErr {
off : i64 = 0;
while off < data.len {
n := sock.write(fd, @data[off], xx (data.len - off));
if n <= 0 { raise error.Send; }
n2 : i64 = xx n;
off += n2;
}
return;
}
// POST `body` to `path` on `srv`. `bearer` "" sends no Authorization
// header. The response buffer comes from `context.allocator`.
http_post :: (srv: ServerUrl, path: string, bearer: string, content_type: string, body: string) -> (HttpResponse, !ClientErr) {
fd := sock.socket(sock.AF_INET, sock.SOCK_STREAM, 0);
if fd < 0 { raise error.Connect; }
addr : sock.SockAddr = .{
sin_len = 16, sin_family = 2,
sin_port = sock.htons(srv.port), sin_addr = srv.addr,
};
if c_connect(fd, @addr, 16) < 0 { sock.close(fd); raise error.Connect; }
hs.set_read_timeout(fd, 10);
h := concat("POST ", concat(path, " HTTP/1.1\r\n"));
h = concat(h, concat("Host: ", concat(srv.host, "\r\n")));
if bearer.len > 0 {
h = concat(h, concat("Authorization: Bearer ", concat(bearer, "\r\n")));
}
h = concat(h, concat("Content-Type: ", concat(content_type, "\r\n")));
h = concat(h, concat("Content-Length: ", concat(int_to_string(body.len), "\r\n")));
h = concat(h, "Connection: close\r\n\r\n");
serr := false;
send_all(fd, h) catch { serr = true; };
if !serr { send_all(fd, body) catch { serr = true; }; }
if serr { sock.close(fd); raise error.Send; }
// Read to EOF (the server closes after one response).
buf : [*]u8 = xx context.allocator.alloc_bytes(RESP_CAP);
filled : i64 = 0;
while filled < RESP_CAP {
n := sock.read(fd, @buf[filled], xx (RESP_CAP - filled));
if n <= 0 { break; }
n2 : i64 = xx n;
filled += n2;
}
sock.close(fd);
raw := string.{ ptr = buf, len = filled };
return parse_response(raw);
}
// Split a raw response into status code + body.
parse_response :: (raw: string) -> (HttpResponse, !ClientErr) {
// status: the integer after the first space ("HTTP/1.1 200 OK")
sp := -1;
i := 0;
while i < raw.len {
if raw[i] == 32 { sp = i; break; }
i += 1;
}
if sp < 0 or sp + 4 > raw.len { raise error.BadResponse; }
status : i64 = 0;
j := sp + 1;
digits := 0;
while j < raw.len {
c := raw[j];
if c < 48 or c > 57 { break; }
status = status * 10 + (c - 48);
digits += 1;
j += 1;
}
if digits != 3 { raise error.BadResponse; }
// body: after the first blank line
k := 0;
body := "";
found := false;
while k + 3 < raw.len {
if raw[k] == 13 and raw[k+1] == 10 and raw[k+2] == 13 and raw[k+3] == 10 {
start := k + 4;
body = string.{ ptr = @raw[start], len = raw.len - start };
found = true;
break;
}
k += 1;
}
if !found { raise error.BadResponse; }
return HttpResponse.{ status = status, body = body };
}

44
src/dist.sx
View File

@@ -33,6 +33,7 @@
#import "modules/std/cli.sx";
jout :: #import "json_out.sx";
pl :: #import "publish/publish.sx";
rem :: #import "publish/remote.sx";
ops :: #import "release/ops.sx";
tops :: #import "token/ops.sx";
srv :: #import "server/distd.sx";
@@ -50,7 +51,7 @@ emit_human :: (s: string, json_mode: bool) {
if json_mode { eputs(s); } else { out(s); }
}
HELP :: "dist — application distribution CLI\n\nUsage:\n dist <group> <command> [flags]\n\nGroups & commands:\n ci\n ci publish publish a release from CI\n --manifest <path> publish manifest (dist.json) to read\n --local-store <dir> local artifact store + db.json directory\n release\n release promote point a channel at a release\n --app <slug> app the channel belongs to\n --channel <name> channel to move\n --release <id> release id to promote\n --local-store <dir> local artifact store + db.json directory\n release rollback move a channel back to its previous release\n --app <slug> app the channel belongs to\n --channel <name> channel to roll back\n --local-store <dir> local artifact store + db.json directory\n server\n server run serve the store over HTTP (0.0.0.0)\n --local-store <dir> local artifact store + db.json directory\n --port <n> TCP port (default 8787)\n GET (public): / (HTML index), /healthz, /api/apps, /api/apps/<slug>, /download/<sha256>\n POST (Bearer token, publish scope): /api/upload, /api/apps/<slug>/releases,\n /api/apps/<slug>/channels/<name>/promote, /api/apps/<slug>/channels/<name>/rollback\n token\n token create mint a scoped automation token (secret shown ONCE)\n --name <name> token name, [a-z0-9._-]\n --local-store <dir> local artifact store + db.json directory\n --scope <words> space-separated scopes: publish read (default: publish)\n --app <slug> restrict to one app (default: any)\n --channel <name> restrict to one channel (default: any)\n --expires-in <secs> lifetime in seconds (default: never expires)\n token list tokens with lifecycle status (never the secret)\n --local-store <dir> local artifact store + db.json directory\n token revoke revoke a token by id\n --id <token-id> token to revoke\n --local-store <dir> local artifact store + db.json directory\n\nGlobal flags:\n --json emit machine-readable JSON on stdout; human text to stderr\n -h, --help show this help and exit\n\nExit codes:\n 0 success\n 1 command failed (publish/promote/rollback/token op aborted or server could not bind)\n 64 usage error (no command, or an unknown/missing command or flag)\n";
HELP :: "dist — application distribution CLI\n\nUsage:\n dist <group> <command> [flags]\n\nGroups & commands:\n ci\n ci publish publish a release from CI\n --manifest <path> publish manifest (dist.json) to read\n --local-store <dir> publish into a local store directory, OR:\n --server <url> publish against a running distd (http://<ipv4-or-localhost>:<port>)\n --token <secret> bearer token for --server (mint with: dist token create)\n release\n release promote point a channel at a release\n --app <slug> app the channel belongs to\n --channel <name> channel to move\n --release <id> release id to promote\n --local-store <dir> local artifact store + db.json directory\n release rollback move a channel back to its previous release\n --app <slug> app the channel belongs to\n --channel <name> channel to roll back\n --local-store <dir> local artifact store + db.json directory\n server\n server run serve the store over HTTP (0.0.0.0)\n --local-store <dir> local artifact store + db.json directory\n --port <n> TCP port (default 8787)\n GET (public): / (HTML index), /healthz, /api/apps, /api/apps/<slug>, /download/<sha256>\n POST (Bearer token, publish scope): /api/upload, /api/apps/<slug>/releases,\n /api/apps/<slug>/channels/<name>/promote, /api/apps/<slug>/channels/<name>/rollback\n token\n token create mint a scoped automation token (secret shown ONCE)\n --name <name> token name, [a-z0-9._-]\n --local-store <dir> local artifact store + db.json directory\n --scope <words> space-separated scopes: publish read (default: publish)\n --app <slug> restrict to one app (default: any)\n --channel <name> restrict to one channel (default: any)\n --expires-in <secs> lifetime in seconds (default: never expires)\n token list tokens with lifecycle status (never the secret)\n --local-store <dir> local artifact store + db.json directory\n token revoke revoke a token by id\n --id <token-id> token to revoke\n --local-store <dir> local artifact store + db.json directory\n\nGlobal flags:\n --json emit machine-readable JSON on stdout; human text to stderr\n -h, --help show this help and exit\n\nExit codes:\n 0 success\n 1 command failed (publish/promote/rollback/token op aborted or server could not bind)\n 64 usage error (no command, or an unknown/missing command or flag)\n";
// True if `name` appears as a token in `args`.
has_flag :: (args: []string, name: string) -> bool {
@@ -96,18 +97,39 @@ report_failure :: (cmd: string, fail: jout.CliFailure, json_mode: bool) -> noret
exit_command_failed();
}
// `dist ci publish` — the real local publish pipeline (P3.4a). Runs the
// publish, then renders the outcome: in json mode a single stable JSON
// object on stdout with the human progress note on stderr (the `--json`
// purity contract); otherwise a readable summary on stdout. A failed
// publish reports through `report_failure` (stderr sentence, JSON error
// object under --json, exit 1).
// `dist ci publish` — the publish pipeline (P3.4a local / P4.5 remote).
// `--local-store <dir>` publishes into a local store; `--server <url>
// --token <secret>` publishes against a running distd. Both produce the
// same outcome shape, rendered identically: in json mode a single stable
// JSON object on stdout with the human progress note on stderr (the
// `--json` purity contract); otherwise a readable summary on stdout. A
// failed publish reports through `report_failure` (stderr sentence, JSON
// error object under --json, exit 1).
handle_ci_publish :: (p: *Parsed, json_mode: bool) {
manifest_path := p.value_of("manifest");
store_dir := p.value_of("local-store");
has_local := p.is_set("local-store");
has_server := p.is_set("server");
if has_local == has_server {
eputs("dist: ci publish needs exactly one of --local-store <dir> or --server <url>\n");
exit_usage();
}
if has_server and !p.is_set("token") {
eputs("dist: --server requires --token <secret>\n");
exit_usage();
}
fail : jout.CliFailure = .{};
o, e := pl.run_publish(manifest_path, store_dir, @fail);
o : pl.PublishOutcome = .{};
e := false;
if has_local {
lo, le := pl.run_publish(manifest_path, p.value_of("local-store"), @fail);
if le { e = true; }
if !le { o = lo; }
} else {
ro, rerr := rem.run_remote_publish(manifest_path, p.value_of("server"), p.value_of("token"), @fail);
if rerr { e = true; }
if !rerr { o = ro; }
}
if e {
report_failure("ci publish", fail, json_mode);
}
@@ -380,7 +402,9 @@ main :: () -> ! {
// being declared here.
publish_flags : []FlagSpec = .[
FlagSpec.{ name = "manifest", takes_value = true, required = true },
FlagSpec.{ name = "local-store", takes_value = true, required = true },
FlagSpec.{ name = "local-store", takes_value = true, required = false },
FlagSpec.{ name = "server", takes_value = true, required = false },
FlagSpec.{ name = "token", takes_value = true, required = false },
];
promote_flags : []FlagSpec = .[
FlagSpec.{ name = "app", takes_value = true, required = true },

257
src/publish/remote.sx Normal file
View File

@@ -0,0 +1,257 @@
// =====================================================================
// remote.sx — `dist ci publish --server --token` (subplan 03, Slice 3):
// the PLAN.md CI contract in remote mode, against a running distd.
//
// Same manifest front half as local publish (parse, required fields,
// on-disk artifact existence), then over HTTP:
//
// per artifact: read bytes -> POST /api/upload -> the server answers
// the content digest; a manifest-DECLARED sha256 is
// checked against it (corruption in transit is loud);
// then: POST /api/apps/<slug>/releases naming the uploaded
// digests -> the server runs the same commit pipeline
// as a local publish.
//
// The server's 200 response IS the publish outcome: it is parsed back
// into a `PublishOutcome`, so `--json` and human rendering reuse the
// P3.4a paths unchanged. Any non-2xx is surfaced by lifting the server's
// JSON error object (code + message) into the CLI failure — the remote
// and local failure surfaces are the same contract.
// =====================================================================
#import "modules/std.sx";
#import "modules/std/json.sx";
#import "modules/std/fs.sx";
#import "../domain/platform.sx";
#import "../manifest/manifest.sx";
#import "publish.sx";
mani :: #import "../manifest/manifest.sx";
pl :: #import "publish.sx";
jout :: #import "../json_out.sx";
cl :: #import "../client/http_client.sx";
dbr :: #import "../repo/db.sx";
jrem :: #import "modules/std/json.sx";
RemoteErr :: error {
Manifest, // manifest failed to load / parse / validate
Url, // --server is not http://<ipv4-or-localhost>:<port>
Transport, // connect / send / unparseable response
Server, // the server answered a JSON error (code passed through)
}
// ── response helpers ──────────────────────────────────────────────────
rm_find :: (o: Object, key: string) -> ?Value {
i := 0;
while i < o.len {
if o.items[i].key == key { return o.items[i].val; }
i += 1;
}
return null;
}
rm_str :: (o: Object, key: string) -> string {
vq := rm_find(o, key);
if vq == null { return ""; }
v := vq!;
if v != .str { return ""; }
return v.str;
}
rm_int :: (o: Object, key: string) -> i64 {
vq := rm_find(o, key);
if vq == null { return 0; }
v := vq!;
if v != .int_ { return 0; }
return v.int_;
}
// Lift a server JSON error body into the CLI failure (falling back to the
// raw status when the body isn't the error shape).
server_failure :: (resp: cl.HttpResponse, fail_out: *jout.CliFailure) {
fail_out.code = "server.error";
fail_out.message = concat("server answered HTTP ", int_to_string(resp.status));
v, pe := jrem.parse(resp.body, context.allocator);
if pe { return; }
if v != .object { return; }
eq := rm_find(v.object, "error");
if eq == null { return; }
ev := eq!;
if ev != .object { return; }
code := rm_str(ev.object, "code");
msg := rm_str(ev.object, "message");
if code.len > 0 { fail_out.code = code; }
if msg.len > 0 { fail_out.message = msg; }
}
// ── pipeline ──────────────────────────────────────────────────────────
run_remote_publish :: (manifest_path: string, server: string, token: string, fail_out: *jout.CliFailure) -> (PublishOutcome, !RemoteErr) {
alloc := context.allocator;
m, me := mani.load_manifest(manifest_path, alloc);
if me {
fail_out.code = pl.manifest_code(me);
fail_out.message = pl.manifest_message(me, manifest_path);
raise error.Manifest;
}
base_dir := dirname(manifest_path);
srv, ue := cl.parse_server_url(server);
if ue {
if ue == error.Tls {
fail_out.code = "server.tls_unsupported";
fail_out.message = "https:// is not supported by the client shim; terminate TLS at a reverse proxy and target http://<host>:<port>";
} else {
fail_out.code = "server.bad_url";
fail_out.message = concat("--server must be http://<ipv4-or-localhost>:<port>, got: ", server);
}
raise error.Url;
}
// 1. Upload every artifact; collect the server-confirmed digests.
keys : List(string) = .{};
sizes : List(i64) = .{};
i := 0;
while i < m.artifacts.len {
ma := m.artifacts.items[i];
src := path_join(base_dir, ma.path);
sb := read_file(src);
if sb == null {
fail_out.code = "store.read";
fail_out.message = concat("artifact file could not be read: ", src);
raise error.Transport;
}
bytes := sb!;
resp, te := cl.http_post(srv, "/api/upload", token, "application/octet-stream", bytes);
if te {
fail_out.code = "server.unreachable";
fail_out.message = concat("upload could not reach the server: ", server);
raise error.Transport;
}
if resp.status != 200 {
server_failure(resp, fail_out);
raise error.Server;
}
uv, upe := jrem.parse(resp.body, alloc);
key := "";
if !upe {
if uv == .object { key = rm_str(uv.object, "sha256"); }
}
if key.len != 64 {
fail_out.code = "server.bad_response";
fail_out.message = "upload response did not carry a sha256 key";
raise error.Transport;
}
// Trust-but-verify a declared digest: the server hashed what
// ARRIVED; a mismatch means corruption (or the wrong file) and
// must abort before any release names the object.
if ma.sha256.len > 0 and ma.sha256 != key {
fail_out.code = "remote.digest_mismatch";
fail_out.message = concat("uploaded bytes hash differently than the manifest declares: ", src);
raise error.Server;
}
keys.append(key, alloc);
sizes.append(bytes.len, alloc);
i += 1;
}
// 2. Publish the release over the uploaded objects.
body := try release_request_body(@m, keys, sizes, base_dir);
rresp, re := cl.http_post(srv, concat(concat("/api/apps/", m.app), "/releases"), token, "application/json", body);
if re {
fail_out.code = "server.unreachable";
fail_out.message = concat("release publish could not reach the server: ", server);
raise error.Transport;
}
if rresp.status != 200 {
server_failure(rresp, fail_out);
raise error.Server;
}
o, oe := outcome_from_response(rresp.body);
if oe {
fail_out.code = "server.bad_response";
fail_out.message = "publish response was not the expected JSON shape";
raise error.Transport;
}
return o;
}
// The JSON body for POST /api/apps/<slug>/releases, naming uploaded
// objects by digest plus the manifest's per-artifact metadata.
release_request_body :: (m: *Manifest, keys: List(string), sizes: List(i64), base_dir: string) -> (string, !RemoteErr) {
gpa := GPA.init();
root : Object = .{};
root.put("version", .str(m.version), xx gpa);
root.put("channel", .str(m.channel), xx gpa);
arr : Array = .{};
i := 0;
while i < m.artifacts.len {
ma := m.artifacts.items[i];
ao : Object = .{};
ao.put("platform", .str(dbr.platform_str(ma.platform)), xx gpa);
ao.put("sha256", .str(keys.items[i]), xx gpa);
ao.put("size_bytes", .int_(sizes.items[i]), xx gpa);
fname := if ma.filename.len > 0 then ma.filename else basename(path_join(base_dir, ma.path));
ao.put("filename", .str(fname), xx gpa);
if ma.content_type.len > 0 { ao.put("content_type", .str(ma.content_type), xx gpa); }
if ma.metadata.len > 0 { ao.put("metadata", .str(ma.metadata), xx gpa); }
arr.add(.object(ao), xx gpa);
i += 1;
}
root.put("artifacts", .array(arr), xx gpa);
buf : [*]u8 = xx context.allocator.alloc_bytes(65536);
rootv : Value = .object(root);
werr := false;
n := write_to_buffer(rootv, string.{ ptr = buf, len = 65536 }) catch { werr = true; 0 };
if werr { raise error.Transport; }
return string.{ ptr = buf, len = n };
}
// Parse the server's publish response back into the CLI's outcome shape
// so both render paths (json / human) are shared with local publish.
outcome_from_response :: (body: string) -> (PublishOutcome, !RemoteErr) {
alloc := context.allocator;
v, pe := jrem.parse(body, alloc);
if pe { raise error.Transport; }
if v != .object { raise error.Transport; }
ro := v.object;
if rm_str(ro, "status") != "published" { raise error.Transport; }
relq := rm_find(ro, "release");
if relq == null { raise error.Transport; }
relv := relq!;
if relv != .object { raise error.Transport; }
rel := relv.object;
o : PublishOutcome = .{
release_id = rm_str(rel, "id"),
app_id = rm_str(rel, "app_id"),
version = rm_str(rel, "version"),
channel = rm_str(rel, "channel"),
};
aq := rm_find(ro, "artifacts");
if aq == null { raise error.Transport; }
av := aq!;
if av != .array { raise error.Transport; }
arr := av.array;
i := 0;
while i < arr.len {
if arr.items[i] != .object { raise error.Transport; }
ao := arr.items[i].object;
o.artifacts.append(PublishedArtifact.{
id = rm_str(ao, "id"),
platform_name = rm_str(ao, "platform"),
size_bytes = rm_int(ao, "size_bytes"),
sha256 = rm_str(ao, "sha256"),
url = rm_str(ao, "url"),
}, alloc);
i += 1;
}
return o;
}