nx_ingest_runner_test.nx source
↩ module page · 87 lines · 2918 B
1// nx_ingest_runner_test.nx -- end-to-end stress of the 5-defense runner.
2
3import "nx_syscalls.nx"
4import "nx_tier.nx"
5import "nx_strconv.nx"
6import "nx_disk_budget.nx"
7import "nx_bloom_capacity.nx"
8import "nx_checkpoint.nx"
9import "nx_progress_watchdog.nx"
10import "nx_shard_writer.nx"
11import "nx_ingest_runner.nx"
12
13func make_key(idx: nx_int, buf: *u8) -> nx_int {
14 buf[0] = 0x6B
15 buf[1] = (0x30 + (idx / 1000)) as u8
16 buf[2] = (0x30 + ((idx / 100) % 10)) as u8
17 buf[3] = (0x30 + ((idx / 10) % 10)) as u8
18 buf[4] = (0x30 + (idx % 10)) as u8
19 return 5
20}
21
22func main() -> nx_exit {
23 let prefix: *u8 = "/tmp/nx_irun_t1" as *u8
24 let ckpt: *u8 = "/tmp/nx_irun_t1.checkpoint" as *u8
25
26 let r: *NxIngestRun = nx_ingest_run_new(
27 prefix,
28 510 as nx_size,
29 100 as nx_size,
30 200,
31 NX_BLOOM_PROFILE_10PCT,
32 ckpt,
33 60000
34 )
35 if r == (0 as *NxIngestRun) { return 11 }
36 if nx_ingest_run_n_offered(r) != 0 { return 12 }
37 if nx_ingest_run_n_emitted(r) != 0 { return 13 }
38 if nx_ingest_run_n_duplicate(r) != 0 { return 14 }
39 if nx_ingest_run_n_budget_halt(r) != 0 { return 15 }
40
41 let k0: *u8 = sys_mmap(NX_BUF_TINY)
42 make_key(1, k0)
43 let v0: nx_int = nx_ingest_run_offer(r, k0, 5 as nx_size, k0, 5 as nx_size)
44 if v0 != NX_OFFER_EMITTED { return 21 }
45 if nx_ingest_run_n_emitted(r) != 1 { return 22 }
46 if nx_ingest_run_n_offered(r) != 1 { return 23 }
47
48 let v1: nx_int = nx_ingest_run_offer(r, k0, 5 as nx_size, k0, 5 as nx_size)
49 if v1 != NX_OFFER_DUPLICATE { return 31 }
50 if nx_ingest_run_n_emitted(r) != 1 { return 32 }
51 if nx_ingest_run_n_duplicate(r) != 1 { return 33 }
52 if nx_ingest_run_n_offered(r) != 2 { return 34 }
53
54 let kbuf: *u8 = sys_mmap(NX_BUF_TINY)
55 var i: nx_int = 2
56 var first_halt: nx_int = -1
57 while i <= 150 {
58 make_key(i, kbuf)
59 let v: nx_int = nx_ingest_run_offer(r, kbuf, 5 as nx_size, kbuf, 5 as nx_size)
60 if v == NX_OFFER_BUDGET_HALT {
61 if first_halt < 0 { first_halt = i }
62 }
63 i = i + 1
64 }
65
66 if first_halt < 0 { return 51 }
67 if nx_ingest_run_n_budget_halt(r) < 1 { return 52 }
68 let used: nx_size = nx_ingest_run_disk_used_bytes(r)
69 if used > 510 { return 53 }
70 if nx_ingest_run_disk_pct_used(r) < 80 { return 54 }
71 if nx_ingest_run_n_shards(r) < 2 { return 55 }
72
73 let lhs: nx_int = nx_ingest_run_n_emitted(r) + nx_ingest_run_n_duplicate(r) + nx_ingest_run_n_budget_halt(r)
74 if lhs != nx_ingest_run_n_offered(r) { return 56 }
75
76 if nx_ingest_run_is_stalled(r) != 0 { return 71 }
77 if nx_ingest_run_n_stalls(r) != 0 { return 72 }
78
79 let final_emitted: nx_int = nx_ingest_run_n_emitted(r)
80 nx_ingest_run_close(r)
81
82 let cp: *NxCheckpoint = nx_checkpoint_open(ckpt)
83 if cp == (0 as *NxCheckpoint) { return 61 }
84 if nx_checkpoint_get(cp) != final_emitted { return 62 }
85
86 return 0
87}