nx_world_telemetry.nx source
↩ module page · 390 lines · 25879 B
1// nx_world_telemetry.nx -- THE WORLD TELEMETRY SINK (gameengine GE33 wt_telemetry_sink, 2026-09-02): the loopback
2// daemon behind nishifamily.com/world/__telemetry that turns a visitor's frame-budget beacon into DURABLE evidence
3// IN THE SOVEREIGN STORE PLANE -- never a file beside it.
4//
5// WHAT ARRIVES: the served world pages (nx_game_page_emit, GE32 HUD) read every frame statistic from an engine door
6// (nx_perf_lib inside the wasm) and beacon the window on pagehide and on a slow beat:
7// POST /world/__telemetry?w=<world> body = JSON {budget_ms,n,total,p50,p95,p99,worst,worst_ever,spikes,over,
8// frames:[ms,ms,...]} (one PF_N-frame window, ~1.5 KB)
9// WHAT IS WRITTEN: ONE ROW appended to the plane knowledge/store/frametrace- (nx_frame_pacing_lib fpc_plane_append ->
10// nx_store_seed_lib's locked append, the same write every board uses; the first beacon bootstraps the plane):
11// id=<world>_<ts> | world | ts | budget(ms:src) | n | total | page p50/p95/p99 | sink p50/p95/p99 | worst |
12// worst_ever | spikes | over | verdict:why | twins | frames_csv
13// The plane is append-only, so it IS the time spine: the latest row for a world is the current window the pacing
14// gate (GE15) judges, and every earlier row is history -- readable by any seat, AI or human, through
15// `nx_store_put knowledge/store/frametrace- load`, and served here at GET /world/__telemetry/spine for a browser.
16// TWO RULERS, ONE ARITHMETIC: the wasm engine and this daemon both compute the percentiles with nx_perf_lib. If the
17// page's numbers and the sink's disagree on the same window, the wasm/native twins have drifted, and every row says
18// so in its `twins` column -- a consistency check nobody has to remember to run.
19// BOUNDS, NAMED: WT_REQ_CAP is the one legitimate cap (a network body's size is unknowable in advance); a body past it
20// is REFUSED with the cap named, never silently cut. The world name is [a-z0-9_]{1,32} because it keys a plane row.
21// HARDENING inherited from nx_http_server_listen by construction: FD_CLOEXEC on the listener, SIGPIPE ignored,
22// SO_REUSEADDR; this daemon binds 127.0.0.1 only (the edge reaches it through the sites proxy), never forks, allocates
23// every buffer once before the accept loop (flat footprint), and answers GET /world/__telemetry/health for the deploy
24// probe. license_tier: ORIGINAL No hw writes (Rule 26).
25import "nx_syscalls.nx"
26import "nx_http_server.nx"
27import "nx_json_lib.nx"
28import "nx_frame_pacing_lib.nx"
29import "nx_world_capture_meta.nx"
30import "nx_capture_append_lib.nx"
31import "nx_gpu_capture_http_lib.nx"
32
33const WT_GPU_ROUTE:*u8="/world/__telemetry/gpu/v1"
34const WT_GPU_PLANE:*u8="knowledge/store/gpu-capture-v1-"
35const WT_PORT_DEFAULT: i64 = 18104
36const WT_BACKLOG: i64 = 16
37const WT_REQ_CAP: i64 = 65536 // NETWORK BODY BOUND -- refused loudly at 400 when exceeded, never trimmed
38const WT_RESP_CAP: i64 = 262144
39const WT_WORLD_CAP: i64 = 64
40// Every frame and every numeric metadata component may need signed-i64 text plus a separator.
41const WT_NUMBER_TEXT_CAP: i64 = 21
42const WT_ROW_CAP: i64 = (PF_N + FPC_COLS * 3) * WT_NUMBER_TEXT_CAP + WT_WORLD_CAP * 2
43const WT_TMO_S: i64 = 5
44const WT_ROUTE: *u8 = "/world/__telemetry"
45const WT_ROUTE_V2: *u8 = "/world/__telemetry/v2"
46const WT_SPINE_V2: *u8 = "/world/__telemetry/v2/spine"
47const WT_PLANE_V2: *u8 = "knowledge/store/frametrace-v2-"
48const WT_HEALTH: *u8 = "/world/__telemetry/health"
49const WT_SPINE_ROUTE: *u8 = "/world/__telemetry/spine"
50const WT_QUERY_W: *u8 = "w="
51const WT_METHOD_GET: i64 = 1
52const WT_METHOD_POST: i64 = 2
53const WT_HTTP_OK: i64 = 200
54const WT_HTTP_NOCONTENT: i64 = 204
55const WT_HTTP_BAD: i64 = 400
56const WT_HTTP_NOTFOUND: i64 = 404
57const WT_HTTP_UNAVAILABLE: i64 = 503
58const WT_CT_TEXT: *u8 = "text/plain; charset=utf-8"
59const WT_CH_QMARK: i64 = 63
60const WT_CH_AMP: i64 = 38
61const WT_CH_TAB: i64 = 9
62const WT_CH_NL: i64 = 10
63const WT_CH_SLASH: i64 = 47
64const WT_CH_COLON: i64 = 58
65const WT_CH_COMMA: i64 = 44
66const WT_CH_UNDERSCORE: i64 = 95
67const WT_PARTIAL_HEAD: i64 = 256
68// counters (slots in one i64 array so /health prints them without a struct)
69const WT_C_BEACONS: i64 = 0
70const WT_C_ROWS: i64 = 1
71const WT_C_REFUSED: i64 = 2
72const WT_C_DISAGREE: i64 = 3
73const WT_C_N: i64 = 4
74
75func wt_slen(s: *u8) -> i64 { var n: i64 = 0; while s[n] != (0 as u8) { n = n + 1 } return n }
76func wt_cat(d: *u8, o: i64, s: *u8) -> i64 { var i: i64 = 0; var p: i64 = o; while s[i] != (0 as u8) { d[p] = s[i]; p = p + 1; i = i + 1 } return p }
77func wt_catn(d: *u8, o: i64, s: *u8, n: i64) -> i64 { var i: i64 = 0; var p: i64 = o; while i < n { d[p] = s[i]; p = p + 1; i = i + 1 } return p }
78func wt_num(d: *u8, o: i64, v: i64) -> i64 {
79 var m: i64 = v
80 var p: i64 = o
81 if m < 0 { d[p] = 45 as u8; p = p + 1; m = 0 - m }
82 if m == 0 { d[p] = 48 as u8; return p + 1 }
83 let t: *u8 = sys_mmap(32)
84 var k: i64 = 0
85 while m > 0 { t[k] = (48 + (m % 10)) as u8; m = m / 10; k = k + 1 }
86 var i: i64 = k - 1
87 while i >= 0 { d[p] = t[i]; p = p + 1; i = i - 1 }
88 sys_munmap(t, 32)
89 return p
90}
91func wt_ch(d: *u8, o: i64, c: i64) -> i64 { d[o] = c as u8; return o + 1 }
92
93// the path bytes before any '?' (the route), as a length
94func wt_route_len(req: *u8, off: i64, len: i64) -> i64 {
95 var i: i64 = 0
96 var f: i64 = len
97 var go: i64 = 1
98 while go == 1 { if i >= len { go = 0 } else { if req[off + i] == (WT_CH_QMARK as u8) { f = i; go = 0 } else { i = i + 1 } } }
99 return f
100}
101func wt_route_is(req: *u8, off: i64, rlen: i64, lit: *u8) -> i64 {
102 let l: i64 = wt_slen(lit)
103 if l != rlen { return 0 }
104 var i: i64 = 0
105 while i < l { if req[off + i] != lit[i] { return 0 } i = i + 1 }
106 return 1
107}
108// `w=<world>` from the query string; returns the copied length (0 when absent)
109func wt_query_world(req: *u8, off: i64, len: i64, out: *u8, cap: i64) -> i64 {
110 let rl: i64 = wt_route_len(req, off, len)
111 if rl >= len { out[0] = 0 as u8; return 0 }
112 var p: i64 = rl + 1
113 let kl: i64 = wt_slen(WT_QUERY_W)
114 var found: i64 = 0 - 1
115 var go: i64 = 1
116 while go == 1 {
117 if p + kl > len { go = 0 } else {
118 var m: i64 = 1
119 var k: i64 = 0
120 while k < kl { if req[off + p + k] != WT_QUERY_W[k] { m = 0; k = kl } else { k = k + 1 } }
121 if m == 1 { found = p + kl; go = 0 } else {
122 var q: i64 = p
123 var go2: i64 = 1
124 while go2 == 1 { if q >= len { go2 = 0 } else { if req[off + q] == (WT_CH_AMP as u8) { go2 = 0 } else { q = q + 1 } } }
125 p = q + 1
126 }
127 }
128 }
129 if found < 0 { out[0] = 0 as u8; return 0 }
130 var o: i64 = 0
131 var j: i64 = found
132 var go3: i64 = 1
133 while go3 == 1 { if j >= len { go3 = 0 } else { if req[off + j] == (WT_CH_AMP as u8) { go3 = 0 } else { if o < cap - 1 { out[o] = req[off + j]; o = o + 1 } j = j + 1 } } }
134 out[o] = 0 as u8
135 return o
136}
137
138func wt_text(cfd: i64, code: i64, msg: *u8) -> i64 {
139 nx_http_write_response(cfd, code, WT_CT_TEXT, wt_slen(WT_CT_TEXT), msg, wt_slen(msg))
140 sys_close(cfd)
141 return 0
142}
143func wt_body(cfd: i64, code: i64, body: *u8, n: i64) -> i64 {
144 nx_http_write_response(cfd, code, WT_CT_TEXT, wt_slen(WT_CT_TEXT), body, n)
145 sys_close(cfd)
146 return 0
147}
148// three percentiles as `a/b/c`
149func wt_triple(d: *u8, o: i64, a: i64, b: i64, c: i64) -> i64 {
150 var p: i64 = wt_num(d, o, a); p = wt_ch(d, p, WT_CH_SLASH); p = wt_num(d, p, b); p = wt_ch(d, p, WT_CH_SLASH); return wt_num(d, p, c)
151}
152
153func main(argc: i64, argv: *i64) -> i64 {
154 var port: i64 = WT_PORT_DEFAULT
155 if argc >= 2 {
156 let ps: *u8 = argv[1] as *u8
157 var pv: i64 = 0
158 var any: i64 = 0
159 var qi: i64 = 0
160 while ps[qi] != (0 as u8) { let c: i64 = ps[qi] as i64; if c >= 48 { if c <= 57 { pv = pv * 10 + (c - 48); any = 1 } } qi = qi + 1 }
161 if any == 1 { port = pv }
162 }
163 let addr: *u8 = sys_mmap(16)
164 nx_http_server_addr_loopback(addr, port)
165 let vp: *i64 = sys_mmap(16) as *i64
166 let lfd: i64 = nx_http_server_listen(addr, WT_BACKLOG, vp)
167 let hello: *u8 = sys_mmap(256)
168 if lfd < 0 {
169 var ho: i64 = wt_cat(hello, 0, "WORLD-TELEMETRY LISTEN-FAIL port=" as *u8); ho = wt_num(hello, ho, port)
170 ho = wt_cat(hello, ho, " verdict=" as *u8); ho = wt_cat(hello, ho, nxs_verdict_name(vp[0]))
171 ho = wt_cat(hello, ho, " -- another process holds the port or the address is refused; nothing was written\n" as *u8)
172 sys_write(1, hello, ho); sys_exit(1); return 1
173 }
174 var ho2: i64 = wt_cat(hello, 0, "WORLD-TELEMETRY-UP 127.0.0.1:" as *u8); ho2 = wt_num(hello, ho2, port)
175 ho2 = wt_cat(hello, ho2, " POST /world/__telemetry?w=<world> | GET /health | GET /spine -> plane " as *u8); ho2 = wt_cat(hello, ho2, FPC_PLANE); ho2 = wt_ch(hello, ho2, WT_CH_NL)
176 sys_write(1, hello, ho2)
177
178 // every buffer once
179 let req: *u8 = sys_mmap(WT_REQ_CAP)
180 let resp: *u8 = sys_mmap(WT_RESP_CAP)
181 let row: *u8 = sys_mmap(WT_ROW_CAP + WCM_OUT_CAP + 4)
182 let meta: *u8 = sys_mmap(WCM_OUT_CAP)
183 let capture_id: *u8 = sys_mmap(65)
184 let capture_fp: *u8 = sys_mmap(65)
185 let capture_hash: *Sha256 = sys_mmap(256) as *Sha256
186 let capture_raw: *u8 = sys_mmap(32)
187 let meta_spans: *i64 = sys_mmap(WCM_FIELDS*16) as *i64
188 let world: *u8 = sys_mmap(WT_WORLD_CAP)
189 let region: *i64 = sys_mmap(PF_WORDS * 8) as *i64
190 let conf: *i64 = sys_mmap(FPC_N * 8) as *i64
191 let ctr: *i64 = sys_mmap(WT_C_N * 8) as *i64
192 let mth: *i64 = sys_mmap(8) as *i64
193 let poff: *i64 = sys_mmap(8) as *i64
194 let plen: *i64 = sys_mmap(8) as *i64
195 let clen: *i64 = sys_mmap(8) as *i64
196 let boff: *i64 = sys_mmap(8) as *i64
197 let rn: *i64 = sys_mmap(8) as *i64
198 let ival: *i64 = sys_mmap(8) as *i64
199 let why: *i64 = sys_mmap(8) as *i64
200 let lp: *i64 = sys_mmap(16) as *i64
201
202 var go: i64 = 1
203 while go == 1 {
204 let cfd: i64 = nx_http_server_accept_one(lfd, vp)
205 if cfd >= 0 {
206 sys_set_socket_timeout(cfd, WT_TMO_S)
207 let rv: i64 = nx_http_server_read_request(cfd, req, WT_REQ_CAP, mth, poff, plen, clen, boff, rn)
208 if rv == NXS_PARSE_ERR {
209 wt_text(cfd, WT_HTTP_BAD, "not an HTTP/1.1 request" as *u8)
210 } else { if rv != NXS_OK { // NXS_SHORT_BODY on the shipping lib: the body overran the cap or stopped short of its Content-Length
211 ctr[WT_C_REFUSED] = ctr[WT_C_REFUSED] + 1
212 wt_text(cfd, WT_HTTP_BAD, "REFUSED: the body exceeds WT_REQ_CAP=65536 bytes or stopped short of its Content-Length -- a beacon is ONE frame window plus its summary, never a history; nothing was recorded" as *u8)
213 } else {
214 let rl: i64 = wt_route_len(req, poff[0], plen[0])
215 if wt_route_is(req,poff[0],rl,WT_GPU_ROUTE)==1 {
216 if mth[0]==WT_METHOD_POST {
217 gch_handle(cfd,WT_GPU_PLANE,((req as i64)+poff[0]) as *u8,plen[0],((req as i64)+boff[0]) as *u8,clen[0],WT_REQ_CAP,WT_REQ_CAP)
218 } else {
219 gch_method(cfd)
220 }
221 } else {
222 if mth[0] == WT_METHOD_GET {
223 if wt_route_is(req, poff[0], rl, WT_HEALTH) == 1 {
224 var o: i64 = wt_cat(resp, 0, "WORLD-TELEMETRY-UP port=" as *u8); o = wt_num(resp, o, port)
225 o = wt_cat(resp, o, " beacons=" as *u8); o = wt_num(resp, o, ctr[WT_C_BEACONS])
226 o = wt_cat(resp, o, " rows_appended=" as *u8); o = wt_num(resp, o, ctr[WT_C_ROWS])
227 o = wt_cat(resp, o, " refused=" as *u8); o = wt_num(resp, o, ctr[WT_C_REFUSED])
228 o = wt_cat(resp, o, " twins_disagree=" as *u8); o = wt_num(resp, o, ctr[WT_C_DISAGREE])
229 o = wt_cat(resp, o, " plane=" as *u8); o = wt_cat(resp, o, FPC_PLANE); o = wt_ch(resp, o, WT_CH_NL)
230 wt_body(cfd, WT_HTTP_OK, resp, o)
231 } else { if wt_route_is(req, poff[0], rl, WT_SPINE_ROUTE) == 1 || wt_route_is(req, poff[0], rl, WT_SPINE_V2) == 1 {
232 // the plane, as a browser can read it: sts_load_fit sizes from the store, so it cannot short-read
233 var read_plane: *u8 = FPC_PLANE
234 if wt_route_is(req, poff[0], rl, WT_SPINE_V2) == 1 { read_plane = WT_PLANE_V2 }
235 let sb: *u8 = sts_load_fit(read_plane, lp)
236 if (sb as i64) == 0 {
237 wt_text(cfd, WT_HTTP_NOTFOUND, "no spine yet: no beacon has been recorded on this host (plane knowledge/store/frametrace- is unseeded)" as *u8)
238 } else {
239 let sn: i64 = lp[0]
240 if sn <= WT_RESP_CAP - WT_PARTIAL_HEAD {
241 wt_body(cfd, WT_HTTP_OK, sb, sn)
242 } else {
243 // the LAST bytes, and the first line SAYS it is a tail -- a silent prefix would read as the whole spine
244 let keep: i64 = WT_RESP_CAP - WT_PARTIAL_HEAD
245 var o2: i64 = wt_cat(resp, 0, "# PARTIAL: last " as *u8); o2 = wt_num(resp, o2, keep)
246 o2 = wt_cat(resp, o2, " of " as *u8); o2 = wt_num(resp, o2, sn)
247 o2 = wt_cat(resp, o2, " bytes -- nx_store_put knowledge/store/frametrace- load reads the whole plane\n" as *u8)
248 o2 = wt_catn(resp, o2, ((sb as i64) + (sn - keep)) as *u8, keep)
249 wt_body(cfd, WT_HTTP_OK, resp, o2)
250 }
251 }
252 } else {
253 wt_text(cfd, WT_HTTP_NOTFOUND, "this sink answers GET /world/__telemetry/health, GET /world/__telemetry/spine and POST /world/__telemetry?w=<world> only" as *u8)
254 } }
255 } else { if mth[0] == WT_METHOD_POST {
256 if wt_route_is(req, poff[0], rl, WT_ROUTE) == 0 && wt_route_is(req, poff[0], rl, WT_ROUTE_V2) == 0 {
257 wt_text(cfd, WT_HTTP_NOTFOUND, "POST is accepted at /world/__telemetry?w=<world> only" as *u8)
258 } else {
259 let is_v2: i64 = wt_route_is(req, poff[0], rl, WT_ROUTE_V2)
260 var meta_n: i64 = 0
261 if is_v2 == 1 { meta_n = wcm_parse((req as i64 + poff[0]) as *u8, plen[0], meta_spans, meta, WCM_OUT_CAP) }
262 let wn: i64 = wt_query_world(req, poff[0], plen[0], world, WT_WORLD_CAP)
263 if fpc_world_safe(world, wn) == 0 || meta_n < 0 {
264 ctr[WT_C_REFUSED] = ctr[WT_C_REFUSED] + 1
265 if is_v2 == 1 {
266 wt_text(cfd, WT_HTTP_BAD, "REFUSED: world must be [a-z0-9_]{1,32}; v2 also requires unique bounded capture, release, wasm, renderer, mode and res tokens; nothing was recorded" as *u8)
267 } else {
268 wt_text(cfd, WT_HTTP_BAD, "REFUSED: ?w=<world> must be [a-z0-9_]{1,32} (it keys the plane row); nothing was recorded" as *u8)
269 }
270 } else {
271 var bend: i64 = boff[0] + clen[0]
272 if bend > rn[0] { bend = rn[0] }
273 let cpres: i64 = fpc_load(FPC_CONF, conf)
274 var budget: i64 = 0
275 var bsrc: *u8 = "page" as *u8
276 if jx_get_int(req, bend, boff[0], "budget_ms" as *u8, ival) > 0 { if ival[0] > 0 { budget = ival[0] } }
277 if budget <= 0 { if cpres == FPC_N { budget = conf[FPC_BUDGET]; bsrc = "conf" as *u8 } else { bsrc = "none" as *u8 } }
278 pf_init(region)
279 let fr: i64 = fpc_frames_parse(req, bend, region, budget)
280 if fr == (0 - 1) {
281 ctr[WT_C_REFUSED] = ctr[WT_C_REFUSED] + 1
282 wt_text(cfd, WT_HTTP_BAD, "REFUSED: the beacon carries no frames:[...] array -- the window is the evidence, a summary alone cannot be re-judged; nothing was recorded" as *u8)
283 } else { if fr == (0 - 2) {
284 ctr[WT_C_REFUSED] = ctr[WT_C_REFUSED] + 1
285 wt_text(cfd, WT_HTTP_BAD, "REFUSED: frames:[...] is malformed (a non-integer inside it); nothing was recorded. Accepted bounds: at most PF_N nonnegative integer milliseconds within FPC_FRAME_MAX_MS; malformed separators and out-of-range values are also refused." as *u8)
286 } else {
287 // the page's own readings (absent -> -1, written as such)
288 var p_n: i64 = 0 - 1; if jx_get_int(req, bend, boff[0], "n" as *u8, ival) > 0 { p_n = ival[0] }
289 var p_total: i64 = 0 - 1; if jx_get_int(req, bend, boff[0], "total" as *u8, ival) > 0 { p_total = ival[0] }
290 var p_p50: i64 = 0 - 1; if jx_get_int(req, bend, boff[0], "p50" as *u8, ival) > 0 { p_p50 = ival[0] }
291 var p_p95: i64 = 0 - 1; if jx_get_int(req, bend, boff[0], "p95" as *u8, ival) > 0 { p_p95 = ival[0] }
292 var p_p99: i64 = 0 - 1; if jx_get_int(req, bend, boff[0], "p99" as *u8, ival) > 0 { p_p99 = ival[0] }
293 var p_worst: i64 = 0 - 1; if jx_get_int(req, bend, boff[0], "worst" as *u8, ival) > 0 { p_worst = ival[0] }
294 var p_wever: i64 = 0 - 1; if jx_get_int(req, bend, boff[0], "worst_ever" as *u8, ival) > 0 { p_wever = ival[0] }
295 var p_spikes: i64 = 0 - 1; if jx_get_int(req, bend, boff[0], "spikes" as *u8, ival) > 0 { p_spikes = ival[0] }
296 var p_over: i64 = 0 - 1; if jx_get_int(req, bend, boff[0], "over" as *u8, ival) > 0 { p_over = ival[0] }
297 // the sink's recomputation over the SAME window through the SAME ruler
298 let s_p50: i64 = pf_pct(region, 500)
299 let s_p95: i64 = pf_pct(region, 950)
300 let s_p99: i64 = pf_pct(region, 990)
301 var twins: *u8 = "AGREE" as *u8
302 var agree: i64 = 0
303 if s_p50 == p_p50 { agree = agree + 1 }
304 if s_p95 == p_p95 { agree = agree + 1 }
305 if s_p99 == p_p99 { agree = agree + 1 }
306 if agree != 3 { twins = "DISAGREE" as *u8; ctr[WT_C_DISAGREE] = ctr[WT_C_DISAGREE] + 1 }
307 // the referee: abstains (never acquits) when the conf is incomplete or the budget unknown
308 why[0] = 0
309 var verdict: i64 = PF_ABSTAIN
310 var whyname: *u8 = "conf-incomplete" as *u8
311 if cpres == FPC_N { if budget > 0 {
312 verdict = fpc_judge(region, conf, budget, why)
313 whyname = fpc_why_name(why[0])
314 if verdict == PF_ABSTAIN { whyname = "below-min-samples" as *u8 }
315 } else { whyname = "budget-unknown" as *u8 } }
316 if is_v2 == 1 {
317 var identity_end: i64 = 0
318 var separators: i64 = 0
319 while identity_end < meta_n { if meta[identity_end] == 9 as u8 { separators = separators+1; if separators == 2 { break } }; identity_end=identity_end+1 }
320 ca_fingerprint(capture_hash,capture_raw,capture_id,"nishi.capture.id.v2\n" as *u8,meta,identity_end,meta,0)
321 ca_fingerprint(capture_hash,capture_raw,capture_fp,"nishi.capture.payload.v2\n" as *u8,meta,meta_n,(req as i64+boff[0]) as *u8,bend-boff[0])
322 }
323 let ts: i64 = sys_now_realtime_sec()
324 // THE ROW (15 columns): id world ts budget n total page sink worst wever spikes over verdict twins frames
325 var o3: i64 = 0
326 if is_v2 == 1 { o3 = wt_cat(row,0,capture_id) } else { o3 = wt_cat(row,0,world); o3=wt_ch(row,o3,WT_CH_UNDERSCORE); o3=wt_num(row,o3,ts) }
327 o3 = wt_ch(row,o3,WT_CH_TAB)
328 o3 = wt_cat(row, o3, world); o3 = wt_ch(row, o3, WT_CH_TAB)
329 o3 = wt_num(row, o3, ts); o3 = wt_ch(row, o3, WT_CH_TAB)
330 o3 = wt_num(row, o3, budget); o3 = wt_ch(row, o3, WT_CH_COLON); o3 = wt_cat(row, o3, bsrc); o3 = wt_ch(row, o3, WT_CH_TAB)
331 o3 = wt_num(row, o3, p_n); o3 = wt_ch(row, o3, WT_CH_TAB)
332 o3 = wt_num(row, o3, p_total); o3 = wt_ch(row, o3, WT_CH_TAB)
333 o3 = wt_triple(row, o3, p_p50, p_p95, p_p99); o3 = wt_ch(row, o3, WT_CH_TAB)
334 o3 = wt_triple(row, o3, s_p50, s_p95, s_p99); o3 = wt_ch(row, o3, WT_CH_TAB)
335 o3 = wt_num(row, o3, p_worst); o3 = wt_ch(row, o3, WT_CH_TAB)
336 o3 = wt_num(row, o3, p_wever); o3 = wt_ch(row, o3, WT_CH_TAB)
337 o3 = wt_num(row, o3, p_spikes); o3 = wt_ch(row, o3, WT_CH_TAB)
338 o3 = wt_num(row, o3, p_over); o3 = wt_ch(row, o3, WT_CH_TAB)
339 o3 = wt_cat(row, o3, fpc_verdict_name(verdict)); o3 = wt_ch(row, o3, WT_CH_COLON); o3 = wt_cat(row, o3, whyname); o3 = wt_ch(row, o3, WT_CH_TAB)
340 o3 = wt_cat(row, o3, twins); o3 = wt_ch(row, o3, WT_CH_TAB)
341 let nf: i64 = pf_n(region)
342 var fi: i64 = 0
343 while fi < nf { if fi > 0 { o3 = wt_ch(row, o3, WT_CH_COMMA) } o3 = wt_num(row, o3, pf_sample(region, fi)); fi = fi + 1 }
344 var write_plane: *u8 = FPC_PLANE
345 if is_v2 == 1 {
346 write_plane = WT_PLANE_V2
347 o3 = wt_cat(row, o3, "\tv2\t" as *u8)
348 o3 = wt_catn(row, o3, meta, meta_n)
349 }
350 var arc: i64 = 0
351 if is_v2 == 1 { arc = ca_append_once(write_plane,capture_id,capture_fp,row,o3) } else { arc = fpc_plane_append(write_plane,row,o3) }
352 ctr[WT_C_BEACONS] = ctr[WT_C_BEACONS] + 1
353 if arc > 0 { ctr[WT_C_ROWS] = ctr[WT_C_ROWS] + 1 }
354 // ANNOUNCE on stdout (the supervisor's log): what landed, where, and the verdict -- a write that
355 // says nothing is indistinguishable from one that never happened
356 var lo: i64 = wt_cat(resp, 0, "BEACON world=" as *u8); lo = wt_cat(resp, lo, world)
357 lo = wt_cat(resp, lo, " frames=" as *u8); lo = wt_num(resp, lo, fr)
358 lo = wt_cat(resp, lo, " budget_ms=" as *u8); lo = wt_num(resp, lo, budget)
359 lo = wt_cat(resp, lo, " sink_p50/p95/p99=" as *u8); lo = wt_triple(resp, lo, s_p50, s_p95, s_p99)
360 lo = wt_cat(resp, lo, " verdict=" as *u8); lo = wt_cat(resp, lo, fpc_verdict_name(verdict))
361 lo = wt_cat(resp, lo, " why=" as *u8); lo = wt_cat(resp, lo, whyname)
362 lo = wt_cat(resp, lo, " twins=" as *u8); lo = wt_cat(resp, lo, twins)
363 lo = wt_cat(resp, lo, " plane_rc=" as *u8); lo = wt_num(resp, lo, arc)
364 lo = wt_cat(resp, lo, " -> " as *u8); lo = wt_cat(resp, lo, write_plane); lo = wt_ch(resp, lo, WT_CH_NL)
365 sys_write(1, resp, lo)
366 if arc >= 0 {
367 wt_body(cfd, WT_HTTP_NOCONTENT, resp, 0)
368 } else {
369 if arc == CA_CONFLICT { wt_text(cfd,409,"CAPTURE-CONFLICT: capture ID already binds different metadata or body; original record retained" as *u8) } else {
370 if is_v2 == 1 {
371 wt_text(cfd, WT_HTTP_UNAVAILABLE, "CAPTURE-STORE-UNAVAILABLE: capture not accepted; retain the exact request for retry; operator must inspect plane health" as *u8)
372 } else {
373 wt_text(cfd, WT_HTTP_BAD, "RECORD-FAILED: the plane append was refused (lock, CAS collision or cap) -- the beacon was judged but NOT stored; retry on the next beat" as *u8)
374 }
375 }
376 }
377 } }
378 }
379 }
380 } else {
381 wt_text(cfd, WT_HTTP_NOTFOUND, "method not served: GET /world/__telemetry/health|spine or POST /world/__telemetry?w=<world>" as *u8)
382 } }
383 }
384 } }
385 sys_close(cfd)
386 }
387 }
388 sys_close(lfd)
389 return 0
390}