nx_ingest_runner.nx source
↩ module page · 148 lines · 4904 B
1// nx_ingest_runner.nx -- composition primitive for autonomous ingest.
2//
3// Wraps disk_budget / bloom / shard_writer / checkpoint / progress_watchdog
4// behind one .offer(key, payload) call. Source-specific parsers feed
5// records in; the runner handles budgeting, dedup, rotation, resume,
6// stall detection.
7//
8// genealogy_id: lambda_architecture_kreps_2014 + map_reduce_dean_2004
9// lineage_id: composition
10
11// nx_safety_envelope:
12// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
13// sil_target: SIL1
14// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
15// verdict: NOT_YET_EVALUATED
16
17import "nx_syscalls.nx"
18import "nx_tier.nx"
19import "nx_disk_budget.nx"
20import "nx_bloom.nx"
21import "nx_bloom_capacity.nx"
22import "nx_shard_writer.nx"
23import "nx_checkpoint.nx"
24import "nx_progress_watchdog.nx"
25
26const NX_OFFER_EMITTED: nx_int = 0
27const NX_OFFER_DUPLICATE: nx_int = 1
28const NX_OFFER_BUDGET_HALT: nx_int = -1
29const NX_OFFER_WRITE_FAIL: nx_int = -2
30
31const NX_INGEST_CKPT_EVERY: nx_int = 1000
32
33struct NxIngestRun {
34 disk_budget: *NxDiskBudget,
35 shard_writer: *NxShardWriter,
36 dedup_bloom: *Bloom,
37 checkpoint: *NxCheckpoint,
38 watchdog: *NxWatchdog,
39
40 n_offered: nx_int,
41 n_emitted: nx_int,
42 n_duplicate: nx_int,
43 n_budget_halt: nx_int,
44 n_write_fail: nx_int,
45 n_stalls: nx_int,
46}
47
48const NX_INGEST_RUN_BYTES: nx_size = 96
49
50func nx_ingest_run_new(
51 output_prefix: *u8,
52 disk_budget_bytes: nx_size,
53 shard_rotate_bytes: nx_size,
54 expected_n_items: nx_int,
55 bloom_fpr_profile: nx_int,
56 checkpoint_path: *u8,
57 watchdog_timeout_ms: nx_int
58) -> *NxIngestRun {
59 let raw: *u8 = sys_mmap(NX_INGEST_RUN_BYTES)
60 let r: *NxIngestRun = raw as *NxIngestRun
61
62 r.disk_budget = nx_disk_budget_new(disk_budget_bytes)
63 r.shard_writer = nx_shard_writer_open(output_prefix, shard_rotate_bytes)
64 if r.shard_writer == (0 as *NxShardWriter) { return 0 as *NxIngestRun }
65 r.dedup_bloom = nx_bloom_new_for_capacity(expected_n_items, bloom_fpr_profile)
66 if r.dedup_bloom == (0 as *Bloom) { return 0 as *NxIngestRun }
67 r.checkpoint = nx_checkpoint_open(checkpoint_path)
68 r.watchdog = nx_progress_watchdog_new(watchdog_timeout_ms)
69
70 r.n_offered = 0
71 r.n_emitted = 0
72 r.n_duplicate = 0
73 r.n_budget_halt = 0
74 r.n_write_fail = 0
75 r.n_stalls = 0
76 return r
77}
78
79func nx_ingest_run_offer(
80 r: *NxIngestRun,
81 key: *u8,
82 key_len: nx_size,
83 payload: *u8,
84 payload_len: nx_size
85) -> nx_int {
86 r.n_offered = r.n_offered + 1
87 nx_progress_watchdog_heartbeat(r.watchdog)
88
89 if bloom_contains(r.dedup_bloom, key, key_len) == 1 {
90 r.n_duplicate = r.n_duplicate + 1
91 return NX_OFFER_DUPLICATE
92 }
93
94 let need: nx_size = payload_len + 1
95 if nx_disk_budget_can_write(r.disk_budget, need) == 0 {
96 r.n_budget_halt = r.n_budget_halt + 1
97 return NX_OFFER_BUDGET_HALT
98 }
99
100 bloom_insert(r.dedup_bloom, key, key_len)
101 let w: nx_int = nx_shard_writer_emit(r.shard_writer, payload, payload_len)
102 if w < 0 {
103 r.n_write_fail = r.n_write_fail + 1
104 return NX_OFFER_WRITE_FAIL
105 }
106 nx_disk_budget_account(r.disk_budget, w as nx_size)
107
108 r.n_emitted = r.n_emitted + 1
109
110 if (r.n_emitted % NX_INGEST_CKPT_EVERY) == 0 {
111 nx_checkpoint_set(r.checkpoint, r.n_emitted)
112 }
113
114 return NX_OFFER_EMITTED
115}
116
117func nx_ingest_run_is_stalled(r: *NxIngestRun) -> nx_int {
118 let v: nx_int = nx_progress_watchdog_check(r.watchdog)
119 if v == NX_WATCHDOG_STALLED { r.n_stalls = r.n_stalls + 1; return 1 }
120 return 0
121}
122
123func nx_ingest_run_close(r: *NxIngestRun) -> nx_int {
124 nx_checkpoint_set(r.checkpoint, r.n_emitted)
125 nx_checkpoint_close(r.checkpoint)
126 nx_shard_writer_close(r.shard_writer)
127 return 0
128}
129
130func nx_ingest_run_n_offered(r: *NxIngestRun) -> nx_int { return r.n_offered }
131func nx_ingest_run_n_emitted(r: *NxIngestRun) -> nx_int { return r.n_emitted }
132func nx_ingest_run_n_duplicate(r: *NxIngestRun) -> nx_int { return r.n_duplicate }
133func nx_ingest_run_n_budget_halt(r: *NxIngestRun) -> nx_int { return r.n_budget_halt }
134func nx_ingest_run_n_write_fail(r: *NxIngestRun) -> nx_int { return r.n_write_fail }
135func nx_ingest_run_n_stalls(r: *NxIngestRun) -> nx_int { return r.n_stalls }
136func nx_ingest_run_n_shards(r: *NxIngestRun) -> nx_int { return nx_shard_writer_n_shards(r.shard_writer) }
137
138func nx_ingest_run_disk_used_bytes(r: *NxIngestRun) -> nx_size {
139 return nx_disk_budget_used(r.disk_budget)
140}
141
142func nx_ingest_run_disk_pct_used(r: *NxIngestRun) -> nx_int {
143 return nx_disk_budget_pct_used(r.disk_budget)
144}
145
146func nx_ingest_run_resume_from(r: *NxIngestRun) -> nx_int {
147 return nx_checkpoint_get(r.checkpoint)
148}