nx_shard_writer.nx source
↩ module page · 172 lines · 5454 B
1// nx_shard_writer.nx -- size-rotating sharded output writer.
2//
3// When the current shard would exceed `rotate_bytes`, closes it and
4// opens the next. Shards named <prefix>-shard-NNNN.jsonl (4-digit
5// zero-padded for sortability).
6//
7// genealogy_id: log4j_size_rolling_policy
8// lineage_id: output_rotation
9
10// nx_safety_envelope:
11// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
12// sil_target: SIL1
13// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
14// verdict: NOT_YET_EVALUATED
15
16import "nx_syscalls.nx"
17import "nx_tier.nx"
18import "nx_str.nx"
19import "nx_strconv.nx"
20
21const NX_SHARDW_OK: nx_int = 0
22const NX_SHARDW_OPEN_FAIL: nx_int = -1
23const NX_SHARDW_WRITE_FAIL: nx_int = -2
24const NX_SHARDW_CLOSED: nx_int = -3
25
26const NX_SHARDW_PATH_BUF: nx_size = 512
27const NX_SHARDW_NEWLINE: nx_int = 10
28
29struct NxShardWriter {
30 prefix_buf: *u8,
31 rotate_bytes: nx_size,
32 current_fd: nx_fd,
33 current_bytes: nx_size,
34 shard_index: nx_int,
35 total_bytes: nx_size,
36 total_records: nx_int,
37 closed: nx_int,
38 add_newline: nx_int,
39}
40
41const NX_SHARDW_STRUCT_BYTES: nx_size = 80
42
43func nx_shardw_format_4d(n: nx_int, buf: *u8) -> nx_int {
44 var v: nx_int = n
45 if v < 0 { v = 0 }
46 if v > 9999 { v = 9999 }
47 buf[0] = (0x30 + (v / 1000)) as u8
48 buf[1] = (0x30 + ((v / 100) % 10)) as u8
49 buf[2] = (0x30 + ((v / 10) % 10)) as u8
50 buf[3] = (0x30 + (v % 10)) as u8
51 return 0
52}
53
54func nx_shardw_path(w: *NxShardWriter, idx: nx_int, out: *u8) -> nx_size {
55 nx_str_cpy(out, w.prefix_buf)
56 let plen: nx_int = nx_str_len(w.prefix_buf)
57 let suffix: *u8 = "-shard-" as *u8
58 var i: nx_int = 0
59 while suffix[i] != 0 {
60 out[plen + i] = suffix[i]
61 i = i + 1
62 }
63 let after_dash: nx_int = plen + i
64 let nbuf: *u8 = ((out as nx_size) + after_dash) as *u8
65 nx_shardw_format_4d(idx, nbuf)
66 let ext: *u8 = ".jsonl" as *u8
67 var j: nx_int = 0
68 while ext[j] != 0 {
69 out[after_dash + 4 + j] = ext[j]
70 j = j + 1
71 }
72 out[after_dash + 4 + j] = 0
73 return (after_dash + 4 + j) as nx_size
74}
75
76func nx_shardw_open_shard(w: *NxShardWriter, idx: nx_int) -> nx_fd {
77 let path_buf: *u8 = sys_mmap(NX_SHARDW_PATH_BUF)
78 nx_shardw_path(w, idx, path_buf)
79 let fd: nx_fd = __syscall(SYS_OPENAT, AT_FDCWD, path_buf as i64, 0x241, 0x1A4, 0, 0)
80 return fd
81}
82
83func nx_shard_writer_open(prefix: *u8, rotate_bytes: nx_size) -> *NxShardWriter {
84 let raw: *u8 = sys_mmap(NX_SHARDW_STRUCT_BYTES)
85 let w: *NxShardWriter = raw as *NxShardWriter
86
87 let plen: nx_int = nx_str_len(prefix)
88 let pbuf: *u8 = sys_mmap(plen + 1)
89 nx_str_cpy(pbuf, prefix)
90 w.prefix_buf = pbuf
91 w.rotate_bytes = rotate_bytes
92 w.shard_index = 0
93 w.current_bytes = 0
94 w.total_bytes = 0
95 w.total_records = 0
96 w.closed = 0
97 w.add_newline = 1
98
99 let fd: nx_fd = nx_shardw_open_shard(w, 0)
100 if fd < 0 { return 0 as *NxShardWriter }
101 w.current_fd = fd
102 return w
103}
104
105func nx_shard_writer_set_raw(w: *NxShardWriter) -> nx_int {
106 w.add_newline = 0
107 return 0
108}
109
110func nx_shard_writer_close(w: *NxShardWriter) -> nx_int {
111 if w.closed == 1 { return NX_SHARDW_OK }
112 if w.current_fd >= 0 { sys_close(w.current_fd) }
113 w.current_fd = -1
114 w.closed = 1
115 return NX_SHARDW_OK
116}
117
118func nx_shardw_rotate(w: *NxShardWriter) -> nx_int {
119 if w.current_fd >= 0 { sys_close(w.current_fd) }
120 w.shard_index = w.shard_index + 1
121 let fd: nx_fd = nx_shardw_open_shard(w, w.shard_index)
122 if fd < 0 { w.current_fd = -1; return NX_SHARDW_OPEN_FAIL }
123 w.current_fd = fd
124 w.current_bytes = 0
125 return NX_SHARDW_OK
126}
127
128func nx_shardw_write_all(fd: nx_fd, buf: *u8, len: nx_size) -> nx_size {
129 var w: nx_size = 0
130 while w < len {
131 let p: *u8 = ((buf as nx_size) + w) as *u8
132 let r: nx_int = sys_write(fd, p, len - w)
133 if r <= 0 { return w }
134 w = w + r
135 }
136 return w
137}
138
139func nx_shard_writer_emit(w: *NxShardWriter, rec: *u8, len: nx_size) -> nx_int {
140 if w.closed == 1 { return NX_SHARDW_CLOSED }
141
142 var extra: nx_size = 0
143 if w.add_newline == 1 { extra = 1 }
144 let need: nx_size = len + extra
145
146 if w.current_bytes > 0 {
147 if w.current_bytes + need > w.rotate_bytes {
148 let rr: nx_int = nx_shardw_rotate(w)
149 if rr != NX_SHARDW_OK { return rr }
150 }
151 }
152
153 let n1: nx_size = nx_shardw_write_all(w.current_fd, rec, len)
154 if n1 != len { return NX_SHARDW_WRITE_FAIL }
155 if w.add_newline == 1 {
156 let nl_buf: *u8 = sys_mmap(NX_BUF_TINY)
157 nl_buf[0] = NX_SHARDW_NEWLINE as u8
158 let n2: nx_size = nx_shardw_write_all(w.current_fd, nl_buf, 1)
159 if n2 != 1 { return NX_SHARDW_WRITE_FAIL }
160 }
161
162 w.current_bytes = w.current_bytes + need
163 w.total_bytes = w.total_bytes + need
164 w.total_records = w.total_records + 1
165 return need as nx_int
166}
167
168func nx_shard_writer_shard_index(w: *NxShardWriter) -> nx_int { return w.shard_index }
169func nx_shard_writer_total_records(w: *NxShardWriter) -> nx_int { return w.total_records }
170func nx_shard_writer_total_bytes(w: *NxShardWriter) -> nx_size { return w.total_bytes }
171func nx_shard_writer_current_bytes(w: *NxShardWriter) -> nx_size { return w.current_bytes }
172func nx_shard_writer_n_shards(w: *NxShardWriter) -> nx_int { return w.shard_index + 1 }