nx_dist_serve.nx source
↩ module page · 124 lines · 8316 B
1// nx_dist_serve.nx -- standalone loopback HTTP server for the hub DOWNLOAD transport: GET /dist/<cid>/<name>
2// resolves <cid> in /volume1/ai/dist/dist_index.conf (Napster registry) -> the internal source path -> STREAMS the
3// real bytes (chunked, multi-GB safe) with the correct Content-Length. Content-addressed: the url carries the cid,
4// the daemon maps cid -> file. ADDITIVE + ISOLATED (new port, touches nothing) -> never-brick by construction;
5// mirrors nx_fin_serve/nx_tools_api_serve. sites.elf proxies nishifamily.com/dist -> 127.0.0.1:<port>.
6// nx_dist_serve serve [port] -- accept-loop, serve content by cid
7// nx_dist_serve probe <port> <path> -- sovereign plain-HTTP self-test of the live round-trip
8// license_tier: ORIGINAL
9import "nx_http_server.nx"
10import "nx_connect.nx" // bounded connect: a raw sys_connect hangs ~127s on a black-holed host
11const DS_MAGIC_1048576: i64 = 1048576
12const DS_MAGIC_8192: i64 = 8192
13const DS_MAGIC_1024: i64 = 1024
14const DS_MAGIC_4096: i64 = 4096
15const DS_MAGIC_65536: i64 = 65536
16const DS_MAGIC_2048: i64 = 2048
17const DS_MAGIC_262144: i64 = 262144
18
19const DS_PORT_DEFAULT: i64 = 18099
20const DS_IDX: *u8 = "/volume1/ai/dist/dist_index.conf" as *u8
21
22func ds_slen(s: *u8) -> i64 { var n: i64=0; while s[n]!=(0 as u8){n=n+1} return n }
23func ds_puts(s: *u8) -> i64 { sys_write(1, s, ds_slen(s)); return 0 }
24func ds_cat(dst: *u8, off: i64, s: *u8) -> i64 { var o: i64=off; var i: i64=0; while s[i]!=(0 as u8){dst[o]=s[i];o=o+1;i=i+1} return o }
25func ds_catn(dst: *u8, off: i64, v: i64) -> i64 { if v==0 { dst[off]=48 as u8; return off+1 } let t:*u8=sys_mmap(24); var m:i64=v; var k:i64=0; while m>0 { t[k]=(48+(m%10)) as u8; m=m/10; k=k+1 } var o:i64=off; var i:i64=k-1; while i>=0 { dst[o]=t[i]; o=o+1; i=i-1 } return o }
26func ds_atoi(s: *u8) -> i64 { var v: i64=0; var i: i64=0; while s[i]!=(0 as u8){ let c: i64=s[i] as i64; if c<48 { return v } if c>57 { return v } v=v*10+(c-48); i=i+1 } return v }
27func ds_eol(b: *u8, from: i64, blen: i64) -> i64 { var e: i64=from; while e<blen { if b[e]==10 as u8 { return e } e=e+1 } return blen }
28func ds_tab(b: *u8, from: i64, limit: i64) -> i64 { var e: i64=from; while e<limit { if b[e]==9 as u8 { return e } e=e+1 } return limit }
29func ds_memeq(a: *u8, b: *u8, n: i64) -> i64 { var i: i64=0; while i<n { if a[i]!=b[i] { return 0 } i=i+1 } return 1 }
30func ds_field(row: *u8, rlen: i64, idx: i64, out: *u8) -> i64 { var fs: i64=0; var f: i64=0; while f<idx { let t: i64=ds_tab(row,fs,rlen); fs=t+1; f=f+1 } let fe: i64=ds_tab(row,fs,rlen); var o: i64=0; var k: i64=fs; while k<fe { out[o]=row[k]; o=o+1; k=k+1 } out[o]=0 as u8; return o }
31
32// pull the cid out of a request line: locate "/dist/", copy chars until '/' or ' '.
33func ds_req_cid(req: *u8, reqlen: i64, out: *u8) -> i64 {
34 let pat: *u8 = "/dist/" as *u8; let pl: i64 = 6
35 var start: i64 = 0 - 1; var i: i64 = 0
36 while i+pl<=reqlen { var ok: i64=1; var j: i64=0; while j<pl { if req[i+j]!=pat[j] { ok=0 } j=j+1 } if ok==1 { start=i+pl; i=reqlen } else { i=i+1 } }
37 if start<0 { out[0]=0 as u8; return 0 }
38 var o: i64=0; var k: i64=start; var run: i64=1
39 while run==1 { if k>=reqlen { run=0 } else { let c: i64=req[k] as i64; if c==47 { run=0 } else { if c==32 { run=0 } else { out[o]=req[k]; o=o+1; k=k+1 } } } }
40 out[o]=0 as u8; return o
41}
42// resolve cid -> source path (out_src) + return size, or -1 if not in the registry.
43func ds_lookup(cid: *u8, out_src: *u8) -> i64 {
44 let buf: *u8 = sys_mmap(DS_MAGIC_1048576)
45 let fd: i64 = sys_openat_rd(DS_IDX); if fd<0 { return 0-1 }
46 let blen: i64 = sys_read(fd, buf, DS_MAGIC_1048576); sys_close(fd); if blen<=0 { return 0-1 }
47 let cl: i64 = ds_slen(cid)
48 var ls: i64=0
49 while ls<blen {
50 let le: i64 = ds_eol(buf, ls, blen)
51 let f0: i64 = ds_tab(buf, ls, le); let cs: i64 = f0+1; let ce: i64 = ds_tab(buf, cs, le)
52 if ce-cs==cl { if ds_memeq((buf as i64 + cs) as *u8, cid, cl)==1 {
53 let szs: *u8 = sys_mmap(32); ds_field((buf as i64 + ls) as *u8, le-ls, 2, szs)
54 ds_field((buf as i64 + ls) as *u8, le-ls, 5, out_src)
55 return ds_atoi(szs)
56 } }
57 ls = le+1
58 }
59 return 0 - 1
60}
61
62func ds_serve(port: i64) -> i64 {
63 let addr: *u8 = sys_mmap(16); nx_http_server_addr_loopback(addr, port)
64 let v: *i64 = sys_mmap(16) as *i64
65 let lfd: i64 = nx_http_server_listen(addr, 16, v)
66 if lfd < 0 { ds_puts("nx_dist_serve LISTEN-FAIL\n" as *u8); sys_exit(1); return 1 }
67 let bnr: *u8 = sys_mmap(64); var bo: i64 = ds_cat(bnr, 0, "nx_dist_serve LIVE on 127.0.0.1:" as *u8); bo = ds_catn(bnr, bo, port); bnr[bo]=10 as u8; sys_write(1, bnr, bo+1)
68 let req: *u8 = sys_mmap(DS_MAGIC_8192); let cid: *u8 = sys_mmap(256); let src: *u8 = sys_mmap(DS_MAGIC_1024)
69 let hdr: *u8 = sys_mmap(DS_MAGIC_4096); let chunk: *u8 = sys_mmap(DS_MAGIC_65536)
70 var run: i64 = 1
71 while run == 1 {
72 let cfd: i64 = nx_http_server_accept_one(lfd, v)
73 if cfd >= 0 {
74 let n: i64 = sys_read(cfd, req, DS_MAGIC_8192)
75 if n > 0 {
76 ds_req_cid(req, n, cid)
77 let size: i64 = ds_lookup(cid, src)
78 if size >= 0 {
79 var ho: i64 = ds_cat(hdr, 0, "HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\nContent-Length: " as *u8)
80 ho = ds_catn(hdr, ho, size); ho = ds_cat(hdr, ho, "\r\nConnection: close\r\n\r\n" as *u8)
81 sys_write(cfd, hdr, ho)
82 let ffd: i64 = sys_openat_rd(src)
83 if ffd >= 0 { var r: i64 = sys_read(ffd, chunk, DS_MAGIC_65536); while r>0 { sys_write(cfd, chunk, r); r = sys_read(ffd, chunk, DS_MAGIC_65536) } sys_close(ffd) }
84 } else {
85 // PROTOCOL LAW (proven twice: /library 10s stall + bare-/dist public hang): every response
86 // MUST carry an exact Content-Length -- a CL-less response never completes for public
87 // clients through the relay chain. Body built separately so the length is computed.
88 let nb: *u8 = "cid not in hub index\n" as *u8
89 let nbl: i64 = ds_slen(nb)
90 var eo: i64 = ds_cat(hdr, 0, "HTTP/1.1 404 Not Found\r\nContent-Type: text/plain\r\nContent-Length: " as *u8)
91 eo = ds_catn(hdr, eo, nbl)
92 eo = ds_cat(hdr, eo, "\r\nConnection: close\r\n\r\n" as *u8)
93 eo = ds_cat(hdr, eo, nb)
94 sys_write(cfd, hdr, eo)
95 }
96 }
97 sys_close(cfd)
98 }
99 }
100 return 0
101}
102
103func ds_probe(port: i64, path: *u8) -> i64 {
104 let addr: *u8 = sys_mmap(16); nx_http_server_addr_loopback(addr, port)
105 let fd: i64 = sys_socket(2, 1, 0)
106 if fd < 0 { ds_puts("PROBE socket-fail\n" as *u8); sys_exit(1); return 1 }
107 if nx_connect_bounded(fd, addr, 16, NX_CONN_DEFAULT_MS) < 0 { ds_puts("PROBE connect-fail\n" as *u8); sys_close(fd); sys_exit(1); return 1 }
108 let req: *u8 = sys_mmap(DS_MAGIC_2048); var o: i64 = ds_cat(req, 0, "GET " as *u8); o = ds_cat(req, o, path); o = ds_cat(req, o, " HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n" as *u8)
109 sys_write(fd, req, o)
110 let buf: *u8 = sys_mmap(DS_MAGIC_262144); var total: i64=0; var run: i64=1
111 while run==1 { let r: i64 = sys_read(fd, ((buf as i64)+total) as *u8, DS_MAGIC_262144-total); if r<=0 { run=0 } else { total=total+r } }
112 sys_close(fd)
113 var show: i64 = total; if show > 400 { show = 400 }
114 sys_write(1, buf, show); ds_puts("\n... [" as *u8); let mb: *u8=sys_mmap(32); let mo: i64=ds_catn(mb,0,total); sys_write(1,mb,mo); ds_puts(" bytes total received]\n" as *u8)
115 return 0
116}
117
118func main(argc: i64, argv: *i64) -> i64 {
119 if argc < 2 { ds_puts("usage: nx_dist_serve serve [port] | probe <port> <path>\n" as *u8); sys_exit(2); return 2 }
120 let mode: *u8 = argv[1] as *u8
121 if mode[0] == (115 as u8) { var port: i64 = DS_PORT_DEFAULT; if argc >= 3 { port = ds_atoi(argv[2] as *u8) } return ds_serve(port) }
122 if mode[0] == (112 as u8) { if argc < 4 { ds_puts("usage: nx_dist_serve probe <port> <path>\n" as *u8); sys_exit(2); return 2 } return ds_probe(ds_atoi(argv[2] as *u8), argv[3] as *u8) }
123 ds_puts("unknown mode (serve|probe)\n" as *u8); sys_exit(2); return 2
124}