nx_torrent_get.nx source
↩ module page · 1047 lines · 85726 B
1// nx_torrent_get.nx -- the DOWNLOAD WORKER (X-TORRENT-LIVE-001 L5): magnet -> a real file on disk +
2// a live STATUS file the daemon/UI polls. Composes the proven engine (mg_parse + cl_announce + BEP-3
3// handshake + BEP-9 metadata + BEP-3 data pieces + sha1 verify). FORK-BOUNDED per peer (parent
4// wait4(WNOHANG)+sys_sleep_ms+nx_kill so an unreachable peer can't hang ~127s). Forked children share
5// progress via on-disk files: outfile (non-trunc, write each verified piece at its offset), donefile
6// (bitmap, set [i]=1 per piece), npcfile (piece count). argv[1]=magnet argv[2]=outdir.
7// nx_sov_build_run drops args -> run the built elf directly. license_tier: ORIGINAL
8//
9// module: nishi-core.torrent.get
10// depends: nishi-core.torrent.peer_piece_live, nishi-core.torrent.magnet
11import "nx_connect.nx" // bounded connect -- SO_RCVTIMEO does NOT bound connect()
12import "nx_peer_piece_live.nx"
13import "nx_magnet.nx"
14import "nx_http_tracker.nx"
15import "nx_dht_get_peers.nx"
16import "nx_endgame.nx" // rarest-first availability map + endgame last-piece racing (Phase 2; re-exports nx_avail_map's am_*)
17import "nx_pex.nx" // BEP-11 peer-exchange: harvest MORE peers (incl. seeders) from connected peers -> finish trackerless/stuck swarms
18import "nx_tracker_list.nx" // DATA-DRIVEN HTTP/HTTPS tracker list (trackers.txt) -> tl_next_line -> wider peer discovery
19import "nx_mse_wire.nx" // MSE/PE anti-throttle: try an encrypted handshake per peer, fall back to plaintext (DPI can't fingerprint)
20import "nx_ipfilter.nx" // ip-filter: drop known-bad / anti-P2P-monitor peer IPs from the peer list (privacy while WAN-seeding)
21import "nx_poolgov.nx" // seq1402/1410/1450: pool width DERIVED from this machine + the live shared budget, never a literal
22const TG_MAGIC_1410: i64 = 1410
23const TG_MAGIC_1450: i64 = 1450
24const TG_MAGIC_1024: i64 = 1024
25const TG_MAGIC_15728640: i64 = 15728640
26const TG_MAGIC_65536: i64 = 65536
27const TG_MAGIC_262144: i64 = 262144
28const TG_MAGIC_40000: i64 = 40000
29// seq1317: ticks a peer-pool slot must rest after being FILLED before it may be refilled. The loop
30// sleeps 2s per tick, so 2 = a ~4s floor between forks into the SAME slot -- invisible to a healthy
31// peer (its slot stays occupied) and decisive against a dead one (which used to re-fork every tick).
32const TG_SLOT_COOLDOWN: i64 = 2
33// Peer connect budget. The 6s sys_set_socket_timeout beside each connect() only ever governed
34// READS -- connect() itself ran to the kernel's ~127s SYN ceiling, so one dead peer held a slot
35// for two minutes. Same 6s intent, now actually enforced on the connect.
36const TG_CONNECT_MS: i64 = 6000
37// Watchdog for the BACKGROUND ANNOUNCE (tg_announce_bg). Ticks are 2s. A tracker round that has
38// not finished in 90s is against a black-holed host -- the kernel's own SYN timeout is ~127s, so
39// this reclaims the slot BEFORE the kernel would, and well inside the 30s announce cadence's
40// tolerance for a skipped cycle. Derived from those two measured numbers, not chosen by taste.
41const TG_ANN_MAX_TICKS: i64 = 45
42const TG_MAGIC_2000000: i64 = 2000000
43const TG_MAGIC_2000000: i64 = 2000000
44const TG_MAGIC_16383: i64 = 16383
45const TG_MAGIC_16384: i64 = 16384
46const TG_MAGIC_1000000000: i64 = 1000000000
47const TG_MAGIC_65535: i64 = 65535
48const TG_MAGIC_1337: i64 = 1337
49const TG_MAGIC_262143: i64 = 262143
50const TG_MAGIC_80000: i64 = 80000
51const TG_MAGIC_2000: i64 = 2000
52
53// Endgame entry: once this many or fewer OBTAINABLE pieces remain (all already in-flight), idle
54// children race them instead of exiting, to beat the one slow holder (the dominant tail-latency
55// source). ~half the 32-child pool: above it there's fresh non-duplicated work; below it, racing
56// the tail beats leaving children idle. A heuristic lever, not a correctness value (SHA1 still guards).
57const TG_ENDGAME_THRESHOLD: i64 = 16
58// BULK unchoke-patience window (seconds). A pure-leecher must OUTLAST a seeder's optimistic-unchoke rotation
59// (~30s in mainline) to ever get served -- bailing at the 6s connect timeout or on the first CHOKE is THE
60// "holders exist but we never pulled" bug (seedeval "IGNORING", 2464 missing pieces w/ holders). We keep the
61// connection alive (keep-alives) waiting for the unchoke up to this window, then free the slot. Bounds slot
62// hold-time so the pool still turns over if every holder keeps choking us.
63const TG_UNCHOKE_WAIT: i64 = 90
64// DATA-PHASE read timeout (seconds). The 6s CONNECT timeout is far too short ONCE a seeder unchokes us:
65// a busy public seeder queues our request behind many peers, so its FIRST piece block -- and inter-block
66// gaps -- routinely exceed 6s. tg_read_msg then returns a spurious timeout and the child gives up
67// mid-request, so we pull NOTHING despite being unchoked. THIS is the residual "IGNORING" bug the pre-unchoke
68// patience window never covered: seedeval shows 23 seeders unchoke us yet have=0. Extending the read timeout
69// to a full optimistic-unchoke interval lets a slow-but-willing seeder actually feed us. Bounded: a seeder
70// truly silent this long AFTER unchoking is genuinely dead -> the read times out once, we free the slot.
71const TG_DATA_TIMEOUT: i64 = 30
72
73// per-connection DH private key for MSE (varies by wall-clock; nonzero ~128-bit). Hardening TODO: CSPRNG.
74func tg_dhpriv(x: *i64) -> i64 { return mse_gen_priv(x) } // CSPRNG per-connection key (nx_mse_wire)
75// MSE is OPT-IN (marker file) so it can't silently break real downloads before it's field-verified vs real
76// clients. Default (no marker) = plaintext-first = the known-good behavior. `touch /volume1/ai/torrent/mse.on` to enable.
77func tg_mse_on() -> i64 { let f: i64 = sys_openat_rd("/volume1/ai/torrent/mse.on" as *u8); if f>=0 { sys_close(f); return 1 } return 0 }
78// MULTI-FILE torrents have NO top-level "length" -- the REAL total is the SUM of the info-dict "files" list's
79// per-file "length". Using npc*plen (the old fallback) OVER-sizes the last piece (treats a partial last piece
80// as full plen) so it NEVER completes -> stuck at 99%/missing=1. This sums files[].length. Falls back to
81// npc*plen only if the list can't be parsed (fail-safe, same as before).
82func tg_multifile_total(meta: *u8, msize: i64, npc: i64, plen: i64) -> i64 {
83 // seq1786: the FOUR bail points below all returned npc*plen SILENTLY and were therefore
84 // INDISTINGUISHABLE -- a parse failure could not be told from a legitimate total, which is how
85 // this survived undetected. Each now returns a distinct NEGATIVE sentinel; the CALLER substitutes
86 // npc*plen so BEHAVIOUR IS UNCHANGED, and records WHICH bail fired.
87 // -1 = no "files" key -2 = "files" is not a list -3 = nx_bc_skip made no progress
88 // -4 = walked the list and summed nothing
89 // A caller MUST treat any negative return as "total unknown", never as a size.
90 let fo: i64 = nx_bc_dict_get(meta, 0, msize, "files" as *u8, 5)
91 if fo < 0 { return 0-1 }
92 if meta[fo] != (108 as u8) { return 0-2 } // 'l' = list of file dicts
93 var sum: i64 = 0; var o: i64 = fo + 1
94 while o < msize {
95 if meta[o] == (101 as u8) { if sum > 0 { return sum } return 0-4 } // 'e' ends the list
96 let lo: i64 = nx_bc_dict_get(meta, o, msize, "length" as *u8, 6)
97 if lo >= 0 { let iv: *i64 = sys_mmap(16) as *i64; if nx_bc_int(meta, lo, msize, iv) >= 0 { sum = sum + iv[0] } }
98 let nxt: i64 = nx_bc_skip(meta, o, msize) // advance past this file-dict
99 if nxt <= o { if sum > 0 { return sum } return 0-3 } // no progress -> bail
100 o = nxt
101 }
102 if sum > 0 { return sum } return 0-4
103}
104// seq1804: the grinding-tail status line used to pass npc*plen as `total` -- a FABRICATED figure,
105// indistinguishable from a measured one, that over-states the torrent by up to a full piece
106// (BANK-155: 4676648960 reported vs 4674889513 actual). It misled THREE of my own investigations
107// into believing the bencode walker had failed when it never had. The parent loop reads npc/plen from
108// download.npc and never parses metadata -- but the metadata is RIGHT THERE on disk, so read it and
109// report the truth. Falls back to npc*plen ONLY on a genuine read/parse failure, which is now a real
110// event rather than the default.
111// LAW: a derived field and a measured field must not share a name.
112func tg_meta_total_of(metapath: *u8, npc: i64, plen: i64) -> i64 {
113 let fd: i64 = sys_openat_rd(metapath); if fd < 0 { return npc*plen }
114 let mb2: *u8 = sys_mmap(TG_MAGIC_262144); let n: i64 = sys_read(fd, mb2, TG_MAGIC_262144 - 1); sys_close(fd)
115 if n <= 0 { return npc*plen }
116 let lno: i64 = nx_bc_dict_get(mb2, 0, n, "length" as *u8, 6)
117 if lno >= 0 { let iv: *i64 = sys_mmap(16) as *i64; if nx_bc_int(mb2, lno, n, iv) >= 0 { return iv[0] } }
118 let t: i64 = tg_multifile_total(mb2, n, npc, plen)
119 if t > 0 { return t }
120 return npc*plen
121}
122// SELECTIVE DOWNLOAD (operator: "if the file isnt the video and it stops the torrent, trim it"). Build a
123// per-piece WANTED mask: a piece is wanted iff it overlaps a file whose size >= min_bytes (the real content;
124// small cover/nfo/sample/txt junk is skipped). Single-file torrent (top-level "length") -> all wanted. If NO
125// file qualifies (e.g. an all-small-files torrent) -> ALL wanted (never accidentally want nothing = safe).
126// Boundary pieces that overlap ANY wanted file are wanted (needed for that file's bytes). Returns wanted_count.
127func tg_build_wanted(meta: *u8, msize: i64, npc: i64, plen: i64, wanted: *u8, min_bytes: i64) -> i64 {
128 var i: i64=0; while i<npc { wanted[i]=0 as u8; i=i+1 }
129 let lno: i64 = nx_bc_dict_get(meta, 0, msize, "length" as *u8, 6)
130 if lno >= 0 { i=0; while i<npc { wanted[i]=1 as u8; i=i+1 } return npc }
131 let fo: i64 = nx_bc_dict_get(meta, 0, msize, "files" as *u8, 5)
132 if fo < 0 { i=0; while i<npc { wanted[i]=1 as u8; i=i+1 } return npc }
133 if meta[fo] != (108 as u8) { i=0; while i<npc { wanted[i]=1 as u8; i=i+1 } return npc }
134 var off: i64=0; var o: i64=fo+1; var go2: i64=1
135 while go2==1 {
136 if o>=msize { go2=0 } else { if meta[o]==(101 as u8) { go2=0 } else {
137 let flo: i64 = nx_bc_dict_get(meta, o, msize, "length" as *u8, 6)
138 var flen: i64=0; if flo>=0 { let iv: *i64=sys_mmap(16) as *i64; if nx_bc_int(meta,flo,msize,iv)>=0 { flen=iv[0] } }
139 if flen >= min_bytes { var b: i64 = off/plen; let bend: i64 = (off+flen-1)/plen
140 while b<=bend { if b>=0 { if b<npc { wanted[b]=1 as u8 } } b=b+1 } }
141 off = off + flen
142 let nxt: i64 = nx_bc_skip(meta, o, msize); if nxt<=o { go2=0 } else { o=nxt }
143 } }
144 }
145 var wc: i64=0; i=0; while i<npc { if (wanted[i] as i64)==1 { wc=wc+1 } i=i+1 }
146 if wc==0 { i=0; while i<npc { wanted[i]=1 as u8; i=i+1 } return npc } // safety: nothing qualified -> want all
147 return wc
148}
149// the min-file-size (bytes) that marks a file as WANTED content. Data-driven: <selfdir>/torrent_wanted.conf
150// (a single integer) overrides; default 15 MB (video >> cover/nfo/sample). Config, not a magic number (rule 11).
151func tg_wanted_min() -> i64 {
152 let p: *u8 = sys_mmap(TG_MAGIC_1024); var o: i64 = pm_bstr(p, 0, tg_self_dir()); o = pm_bstr(p, o, "/torrent_wanted.conf" as *u8); p[o]=0 as u8
153 let fd: i64 = sys_openat_rd(p); if fd < 0 { return TG_MAGIC_15728640 }
154 let b: *u8 = sys_mmap(64); let n: i64 = sys_read(fd, b, 63); sys_close(fd)
155 var v: i64=0; var i: i64=0; while i<n { if b[i]>=(48 as u8) { if b[i]<=(57 as u8) { v=v*10+((b[i] as i64)-48) } } i=i+1 }
156 if v<=0 { return TG_MAGIC_15728640 }
157 return v
158}
159// pm_read_msg over an MseCtx (transparent decrypt; plaintext ctx == raw). >0 body len, 0 keep-alive, <0 error.
160func tg_read_msg(ctx: *MseCtx, fd: i64, mb: *u8, cap: i64) -> i64 {
161 if mse_read(ctx, fd, mb, 4) != 4 { return 0-1 }
162 let blen: i64 = _pw_get_u32(mb, 0)
163 if blen == 0 { return 0 }
164 if blen > cap-4 { return 0-1 }
165 if mse_read(ctx, fd, mb+4, blen) != blen { return 0-1 }
166 return blen
167}
168
169// raw open O_RDWR|O_CREAT (0x42) -- NO O_TRUNC, so children write at offsets without wiping.
170func tg_open_rw(path: *u8) -> i64 { return __syscall(SYS_OPENAT, AT_FDCWD, path as i64, 0x42, 0x1a4, 0, 0) }
171
172// PEX harvest (BEP-11): append gossiped compact peers (6B each: ip4 BE + port BE) to the shared
173// download.pex log under flock. Append-only + best-effort DISCOVERY -- never a data invariant (SHA1
174// still guards every piece), so a dropped append is harmless. The parent merges these into the peer
175// set on each re-announce. Reuses the avail-map flock idiom (sys_flock LOCK_EX) + the proven
176// O_WRONLY|O_CREAT|O_APPEND (0x441) atomic-append open.
177func tg_pex_append(pexpath: *u8, ips: *i64, ports: *i64, cnt: i64) -> i64 {
178 if cnt <= 0 { return 0 }
179 let fd: i64 = __syscall(SYS_OPENAT, AT_FDCWD, pexpath as i64, 0x441, 0x1a4, 0, 0)
180 if fd < 0 { return 0 }
181 sys_flock(fd, SYS_LOCK_EX)
182 let buf: *u8 = sys_mmap(cnt * 6 + 16)
183 var o: i64 = 0; var i: i64 = 0
184 while i < cnt {
185 let ip: i64 = ips[i]; let pt: i64 = ports[i]
186 buf[o]=((ip>>24)&0xff) as u8; buf[o+1]=((ip>>16)&0xff) as u8; buf[o+2]=((ip>>8)&0xff) as u8; buf[o+3]=(ip&0xff) as u8
187 buf[o+4]=((pt>>8)&0xff) as u8; buf[o+5]=(pt&0xff) as u8
188 o = o + 6; i = i + 1
189 }
190 cl_write_n(fd, buf, o)
191 sys_flock(fd, SYS_LOCK_UN)
192 sys_close(fd)
193 return cnt
194}
195
196// parent: merge harvested PEX peers (download.pex, 6B records) into the live peer set, deduped
197// against the current set. Returns the new count (bounded by max). Lock-free read -- a torn tail
198// record is ignored (i+6<=rn) and re-read next cycle.
199func tg_pex_merge(pexpath: *u8, ips: *i64, ports: *i64, have: i64, max: i64) -> i64 {
200 let fd: i64 = sys_openat_rd(pexpath)
201 if fd < 0 { return have }
202 let buf: *u8 = sys_mmap(6 * TG_MAGIC_65536 + 16)
203 let rn: i64 = sys_read(fd, buf, 6 * TG_MAGIC_65536); sys_close(fd)
204 if rn < 6 { return have }
205 var h: i64 = have
206 var i: i64 = 0
207 while i + 6 <= rn {
208 let ip: i64 = ((buf[i] as i64)<<24)|((buf[i+1] as i64)<<16)|((buf[i+2] as i64)<<8)|(buf[i+3] as i64)
209 let pt: i64 = ((buf[i+4] as i64)<<8)|(buf[i+5] as i64)
210 var dup: i64 = 0; var j: i64 = 0
211 while j < h { if ips[j]==ip { if ports[j]==pt { dup=1; j=h } } j=j+1 }
212 if dup == 0 { if h < max { ips[h]=ip; ports[h]=pt; h=h+1 } }
213 i = i + 6
214 }
215 return h
216}
217
218// ip-filter: compact the peer list in place, dropping any IP in a blocked range. Runs after each announce/PEX
219// merge in the PARENT -> children inherit the clean list via fork COW (no per-connect cost, no tg_from_peer
220// arg-cap change). cnt<=0 (no blocklist) -> pass-through no-op.
221func tg_ipf_filter(ips: *i64, ports: *i64, np: i64, arr: *i64, cnt: i64) -> i64 {
222 if cnt <= 0 { return np }
223 var w: i64 = 0; var i: i64 = 0
224 while i < np { if ipf_blocked(arr, cnt, ips[i]) == 0 { ips[w]=ips[i]; ports[w]=ports[i]; w=w+1 } i=i+1 }
225 return w
226}
227
228// read outdir/download.control -> 1 if the operator asked to stop (paused/removed), else 0.
229// pure file-based control (the daemon writes the file); no kill/getpid/setpgid needed.
230func tg_stop(ctlpath: *u8) -> i64 {
231 let fd: i64 = sys_openat_rd(ctlpath)
232 if fd < 0 { return 0 }
233 let b: *u8 = sys_mmap(16); let r: i64 = sys_read(fd, b, 15); sys_close(fd)
234 if r <= 0 { return 0 }
235 let c: i64 = b[0] as i64
236 if c == 112 { return 1 } // 'p' = paused
237 if c == 114 { return 1 } // 'r' = removed
238 return 0
239}
240
241func tg_status(path: *u8, name: *u8, total: i64, done: i64, have: i64, npc: i64, peers: i64, state: *u8) -> i64 {
242 let fd: i64 = sys_openat_wr(path, 0x1a4)
243 if fd < 0 { return 0 }
244 // newline-delimited key=value (value = rest of line) so a name with spaces parses cleanly
245 cl_w(fd, "total=" as *u8); cl_wn(fd, total); cl_w(fd, "\n" as *u8)
246 cl_w(fd, "done=" as *u8); cl_wn(fd, done); cl_w(fd, "\n" as *u8)
247 cl_w(fd, "have=" as *u8); cl_wn(fd, have); cl_w(fd, "\n" as *u8)
248 cl_w(fd, "pieces=" as *u8); cl_wn(fd, npc); cl_w(fd, "\n" as *u8)
249 cl_w(fd, "peers=" as *u8); cl_wn(fd, peers); cl_w(fd, "\n" as *u8)
250 cl_w(fd, "state=" as *u8); cl_w(fd, state); cl_w(fd, "\n" as *u8)
251 cl_w(fd, "name=" as *u8); cl_w(fd, name); cl_w(fd, "\n" as *u8)
252 sys_close(fd)
253 return 0
254}
255
256// CHILD: download as many NEEDED pieces as this peer serves; write each to outfile@offset + set
257// donefile bit. Returns pieces completed by this child.
258// STREAM MODE: present (the daemon writes it when a /stream request arrives) => download SEQUENTIALLY so
259// the contiguous playable prefix grows front-to-back for a <video> (rarest-first leaves holes that stall
260// playback). Absent => normal rarest-first+endgame (best for swarm health). 1 if the sidecar exists.
261func tg_stream_on(path: *u8) -> i64 { let fd: i64 = sys_openat_rd(path); if fd < 0 { return 0 } sys_close(fd); return 1 }
262
263// sequential pick: the LOWEST-index NEEDED piece this peer HAS. Pass 1 prefers a piece no other child is
264// downloading (in-flight==0) so the pool spreads; pass 2 allows an in-flight piece (race) so the prefix
265// always advances. -1 if the peer has nothing needed. (Relates to nx_sp_sequential, peer+in-flight aware.)
266func tg_seq_pick(donebits: *u8, peerbits: *u8, byte_len: i64, havebits: i64, inf: *i64, npc: i64) -> i64 {
267 var p: i64 = 0
268 while p < npc {
269 if (donebits[p] as i64) == 0 { var has: i64 = 1; if havebits == 1 { if nx_pw_bitfield_has(peerbits, byte_len, p) == 0 { has = 0 } } if has == 1 { if inf[p] == 0 { return p } } }
270 p = p + 1
271 }
272 p = 0
273 while p < npc {
274 if (donebits[p] as i64) == 0 { var h2: i64 = 1; if havebits == 1 { if nx_pw_bitfield_has(peerbits, byte_len, p) == 0 { h2 = 0 } } if h2 == 1 { return p } }
275 p = p + 1
276 }
277 return 0 - 1
278}
279
280// SEED what we already have (operator OK'd uploading -> s-class exceed): send our BITFIELD (peers learn
281// our pieces) + UNCHOKE (they may request). Earns RECIPROCITY -- a LEECHER that holds a piece we still
282// need (e.g. the last piece) but needs one WE have will TRADE it to us, instead of us begging a non-
283// ---- PARTIAL-PIECE PERSISTENCE (reciprocity-starvation fix) ----
284// Saturated public seeders give a pure-leecher only BRIEF optimistic-unchoke windows -- too short to move a
285// whole 2-4MB piece. Without this a child accumulates a piece in pbuf and DISCARDS it on CHOKE -> zero progress
286// on stingy swarms (seedeval "IGNORING": 23 seeders unchoke us yet have=0). Persist the received CONTIGUOUS
287// PREFIX of the in-flight piece to a staging file (download.pstage) + its byte-count (download.ppart) so the
288// NEXT window (this or another child) RESUMES from there -> pieces complete across many brief windows.
289// download.part stays VERIFIED-ONLY (verify/extract UNCHANGED); SHA1 still gates completion, so a torn/stale
290// prefix just fails verify -> redownload = SAFE BY CONSTRUCTION. cprefix (contiguous), NOT pbytes (total), is
291// persisted so a resumed prefix can never be garbage from an out-of-order gap. The staged prefix is bounded by
292// PCAP (block receipt drops beyond PCAP) so a load can never overflow pbuf.
293func tg_ppart_get(ppath: *u8, piece: i64) -> i64 {
294 let fd: i64 = sys_openat_rd(ppath); if fd < 0 { return 0 }
295 let b: *u8 = sys_mmap(16); sys_lseek(fd, piece*8, 0); let r: i64 = sys_read(fd, b, 8); sys_close(fd)
296 if r < 8 { return 0 }
297 var v: i64 = 0; var i: i64 = 0; while i < 8 { v = v | ((b[i] as i64) << (i*8)); i = i + 1 }
298 if v < 0 { return 0 }
299 return v
300}
301func tg_ppart_set(ppath: *u8, piece: i64, val: i64) -> i64 {
302 let fd: i64 = tg_open_rw(ppath); if fd < 0 { return 0 } // O_RDWR|O_CREAT, NO truncate -> per-piece 8-byte slot, sparse
303 let b: *u8 = sys_mmap(16); var i: i64 = 0; while i < 8 { b[i] = ((val >> (i*8)) & 0xff) as u8; i = i + 1 }
304 sys_lseek(fd, piece*8, 0); cl_write_n(fd, b, 8); sys_close(fd)
305 return 0
306}
307// flush pbuf[0..cprefix] of `piece` to the staging file + record cprefix. Called on CHOKE / child exit.
308func tg_ppart_flush(spath: *u8, ppath: *u8, piece: i64, plen: i64, pbuf: *u8, cprefix: i64) -> i64 {
309 if cprefix <= 0 { return 0 }
310 let fd: i64 = tg_open_rw(spath); if fd < 0 { return 0 }
311 sys_lseek(fd, piece*plen, 0); cl_write_n(fd, pbuf, cprefix); sys_close(fd)
312 tg_ppart_set(ppath, piece, cprefix)
313 return 0
314}
315// load `piece`'s staged prefix (cprefix bytes) into pbuf. Returns bytes loaded (0 => start fresh, safe).
316func tg_ppart_load(spath: *u8, piece: i64, plen: i64, pbuf: *u8, cprefix: i64) -> i64 {
317 if cprefix <= 0 { return 0 }
318 let fd: i64 = sys_openat_rd(spath); if fd < 0 { return 0 }
319 sys_lseek(fd, piece*plen, 0); let r: i64 = sys_read(fd, pbuf, cprefix); sys_close(fd)
320 if r < cprefix { return 0 }
321 return cprefix
322}
323// reciprocating seeder for a free optimistic-unchoke. THE "IGNORING" completion fix.
324func tg_seed_hello(ctx: *MseCtx, fd: i64, lbits: *u8, npc: i64) -> i64 {
325 let nbytes: i64 = (npc + 7) / 8
326 let bf: *u8 = sys_mmap(nbytes + 16); var b: i64 = 0; while b < nbytes { bf[b] = 0 as u8; b = b + 1 }
327 var p: i64 = 0
328 while p < npc { if (lbits[p] as i64) == 1 { let by: i64 = p / 8; let bit: i64 = 7 - (p - by*8); bf[by] = ((bf[by] as i64) | (1 << bit)) as u8 } p = p + 1 }
329 let m1: *u8 = sys_mmap(nbytes + 16); mse_write(ctx, fd, m1, nx_pw_build_msg(NX_PW_BITFIELD, bf, nbytes, m1))
330 let m2: *u8 = sys_mmap(16); mse_write(ctx, fd, m2, nx_pw_build_msg(NX_PW_UNCHOKE, 0 as *u8, 0, m2))
331 return 0
332}
333// PER-CALL 256 KiB BUFFERS -> LAZY STATICS + EXPLICIT ZERO-FILL (2026-08-14).
334// tg_from_peer allocated FOUR TG_MAGIC_262144 buffers (lbits/wbits/pickbits/meta) at entry and freed
335// none, and it has several early `return 0` paths that shed everything allocated so far. That is ~1 MiB
336// leaked PER PEER CONNECTION. MEASURED by nx_leak_check on five live nx_torrent_get processes:
337// mem LEAK 9,921-16,605 kB/min and rss LEAK 3,341-10,277 kB/min each -- ~16 calls/min at ~1 MiB.
338// Same class as ts_wn here and ss_hget in nx_seg_store: a per-call allocation in a function the caller
339// runs in a loop is a per-CALLER-LOOP leak.
340//
341// WHY THE ZERO-FILL IS LOAD-BEARING AND NOT DEFENSIVE PADDING: a fresh sys_mmap is ZERO-FILLED by the
342// kernel; a reused static is not. lbits is filled by `sys_read(donefd, lbits, TG_MAGIC_262144)`, and
343// when download.done is SHORTER than the buffer the remainder previously read as zeros = "piece NOT
344// done". A dirty static would carry a previous peer's DONE bits into the tail and the picker would skip
345// pieces it never fetched -- silent data corruption, not a memory cost. Zeroing restores the exact
346// precondition, so the static is byte-for-byte equivalent in effect to the fresh mmap it replaces.
347// That equivalence IS the correctness proof; it is why this needed no behavioural change anywhere else.
348// wbits/pickbits/meta are rebuilt per call (wbuilt/msize/mdone all reset), but they are zeroed on the
349// same rule rather than on a case-by-case argument -- one invariant beats four separate ones.
350// COST: 1 MiB of stores per call replaces 1 MiB of mmap + page faults, so this is cheaper, not dearer.
351static tg_scr_lbits: *u8
352static tg_scr_wbits: *u8
353static tg_scr_pick: *u8
354static tg_scr_meta: *u8
355func tg_zero(p: *u8, n: i64) -> i64 { var i: i64 = 0; while i < n { p[i] = 0 as u8; i = i + 1 } return 0 }
356func tg_from_peer(ipv: i64, port: i64, ih: *u8, mypid: *u8, outpath: *u8, donepath: *u8, npcpath: *u8, metapath: *u8, statpath: *u8, ctlpath: *u8, availpath: *u8, inflightpath: *u8, streampath: *u8, pexpath: *u8, peers: i64) -> i64 {
357 let dest: *u8 = sys_mmap(16); cl_sockaddr(dest, port, (ipv>>24)&0xff, (ipv>>16)&0xff, (ipv>>8)&0xff, ipv&0xff)
358 var fd: i64 = sys_socket(AF_INET, SOCK_STREAM, 0)
359 if fd < 0 { return 0 }
360 sys_set_socket_timeout(fd, 6)
361 if nx_connect_bounded(fd, dest, 16, TG_CONNECT_MS) != 0 { sys_close(fd); return 0 }
362 // MSE/PE ANTI-THROTTLE (OPT-IN via /volume1/ai/torrent/mse.on): try an encrypted handshake first (DPI can't
363 // fingerprint it); on reject the socket is dead -> reconnect + plaintext. DEFAULT (no marker) = plaintext-first
364 // = the known-good download path (MSE stays off until field-verified vs real clients, so it can't stall real DLs).
365 let ctx: *MseCtx = sys_mmap(MSE_CTX_BYTES) as *MseCtx
366 if tg_mse_on() == 1 {
367 let xpriv: *i64 = sys_mmap((MSE_N+2)*8) as *i64; tg_dhpriv(xpriv)
368 if mse_connect(fd, ih, xpriv, ctx) != 1 {
369 sys_close(fd)
370 fd = sys_socket(AF_INET, SOCK_STREAM, 0); if fd < 0 { return 0 }
371 sys_set_socket_timeout(fd, 6)
372 if nx_connect_bounded(fd, dest, 16, TG_CONNECT_MS) != 0 { sys_close(fd); return 0 }
373 mse_ctx_plain(ctx)
374 }
375 } else { mse_ctx_plain(ctx) } // DEFAULT plaintext (byte-identical to pre-MSE behavior)
376 let ch: *u8 = sys_mmap(128); nx_pw_build_handshake(ih, mypid, ch); ch[25] = (ch[25] as i64 | 0x10) as u8
377 if mse_write(ctx, fd, ch, 68) != 68 { sys_close(fd); return 0 }
378 let ph: *u8 = sys_mmap(128); if mse_read(ctx, fd, ph, 68) != 68 { sys_close(fd); return 0 }
379 let gih: *u8 = sys_mmap(20); let gpid: *u8 = sys_mmap(20); if nx_pw_parse_handshake(ph, 68, gih, gpid) != 1 { sys_close(fd); return 0 }
380 let ceb: *u8 = sys_mmap(64); let cm: *u8 = sys_mmap(128); mse_write(ctx, fd, cm, pm_build_ext(0, ceb, pm_bstr(ceb, 0, "d1:md11:ut_metadatai1e6:ut_pexi3eee" as *u8), cm)) // advertise ut_metadata=1 + ut_pex=3 (PEX_OUR_ID)
381
382 let outfd: i64 = tg_open_rw(outpath); let donefd: i64 = tg_open_rw(donepath)
383 if (tg_scr_lbits as i64)==0 { tg_scr_lbits = sys_mmap(TG_MAGIC_262144) }
384 let lbits: *u8 = tg_scr_lbits
385 tg_zero(lbits, TG_MAGIC_262144) // MUST precede the read: restores the fresh-mmap zero precondition
386 sys_lseek(donefd, 0, 0); sys_read(donefd, lbits, TG_MAGIC_262144) // load current progress
387 // SELECTIVE DOWNLOAD: download.wanted (sibling of download.done) = per-piece wanted mask; pickbits = lbits | ~wanted
388 // so the picker treats unwanted (junk-file) pieces as done -> never fetches them. Built once after metadata resolves.
389 let wantedpath: *u8 = sys_mmap(640); var wpo: i64=0; var lastsl: i64=0; var wci: i64=0
390 while donepath[wci]!=(0 as u8) { if donepath[wci]==(47 as u8) { lastsl=wci } wci=wci+1 }
391 wci=0; while wci<=lastsl { wantedpath[wpo]=donepath[wci]; wpo=wpo+1; wci=wci+1 } wpo=pm_bstr(wantedpath, wpo, "download.wanted" as *u8); wantedpath[wpo]=0 as u8
392 if (tg_scr_wbits as i64)==0 { tg_scr_wbits = sys_mmap(TG_MAGIC_262144) }
393 if (tg_scr_pick as i64)==0 { tg_scr_pick = sys_mmap(TG_MAGIC_262144) }
394 let wbits: *u8 = tg_scr_wbits; let pickbits: *u8 = tg_scr_pick; var wbuilt: i64=0
395 tg_zero(wbits, TG_MAGIC_262144); tg_zero(pickbits, TG_MAGIC_262144)
396 // partial-prefix persistence staging (siblings of download.part; same dir-prefix idiom as download.wanted)
397 let spath: *u8 = sys_mmap(640); var spo: i64=0; var si2: i64=0; while si2<=lastsl { spath[spo]=donepath[si2]; spo=spo+1; si2=si2+1 } spo=pm_bstr(spath, spo, "download.pstage" as *u8); spath[spo]=0 as u8
398 let ppath: *u8 = sys_mmap(640); var ppo2: i64=0; var pi3: i64=0; while pi3<=lastsl { ppath[ppo2]=donepath[pi3]; ppo2=ppo2+1; pi3=pi3+1 } ppo2=pm_bstr(ppath, ppo2, "download.ppart" as *u8); ppath[ppo2]=0 as u8
399
400 if (tg_scr_meta as i64)==0 { tg_scr_meta = sys_mmap(TG_MAGIC_262144) }
401 let meta: *u8 = tg_scr_meta
402 tg_zero(meta, TG_MAGIC_262144)
403 var msize: i64=0; var putm: i64=0; var gmeta: i64=0; var nmeta: i64=0; var rmeta: i64=0; var mdone: i64=0
404 var plen: i64=0; var poff: i64=0; var total: i64=0; var npc: i64=0; var unchoked: i64=0
405 let name: *u8 = sys_mmap(256); name[0]=63 as u8; name[1]=0 as u8
406 let mb: *u8 = sys_mmap(TG_MAGIC_40000); let pbuf: *u8 = sys_mmap(PCAP)
407 // seq1734: counts block receipts refused for landing beyond PCAP. Must never be silent -- a
408 // truncated piece can never verify, so a non-zero value here is the difference between
409 // "this torrent is slow" and "this torrent is structurally impossible with this buffer".
410 var dropped_blocks: i64 = 0
411 let peerbits: *u8 = sys_mmap(TG_MAGIC_40000) // THIS peer's advertised pieces (from BITFIELD + HAVE), MSB-first per BEP-3
412 var have: i64=0; var cur: i64=0-1; var pbytes: i64=0; var loops: i64=0; var go: i64=1; var chmsg: i64=0; var havebits: i64=0; var patient: i64=0 // tail-patience: stay glued to a holder waiting for its unchoke (the seedeval "IGNORING" fix)
413 var cprefix: i64=0 // contiguous received prefix of the in-flight piece (persisted on choke so brief unchoke windows accumulate)
414 let t_start: i64 = sys_now_realtime_sec() // connection start -> bounds the bulk unchoke-patience window (TG_UNCHOKE_WAIT)
415 var avail_added: i64=0; var avbuf: *i64 = 0 as *i64; var effbuf: *i64 = 0 as *i64; var infbuf: *i64 = 0 as *i64 // rarest-first + endgame scratch (lazy-alloc once npc known)
416 // METADATA CACHE LOAD: if a prior peer already resolved + SHA1-verified the metadata, load it from
417 // disk and skip the ut_metadata round-trip entirely (kills the per-child refetch tax). The SHA1
418 // re-verify guards a partial/corrupt cache -> falls back to the normal ut_metadata fetch.
419 let mcfd: i64 = sys_openat_rd(metapath)
420 if mcfd >= 0 {
421 let mr: i64 = sys_read(mcfd, meta, TG_MAGIC_262144); sys_close(mcfd)
422 if mr > 0 {
423 let gc: *u8 = sys_mmap(20); sha1(meta, mr, gc); var ec: i64=1; var xc: i64=0; while xc<20 { if gc[xc]!=ih[xc] { ec=0; xc=20 } else { xc=xc+1 } }
424 if ec==1 {
425 msize=mr; mdone=1; rmeta=1
426 let plo: i64=nx_bc_dict_get(meta,0,msize,"piece length" as *u8,12); let iv5: *i64=sys_mmap(16) as *i64; if plo>=0 { nx_bc_int(meta,plo,msize,iv5); plen=iv5[0] }
427 let pco: i64=nx_bc_dict_get(meta,0,msize,"pieces" as *u8,6); let so5: *i64=sys_mmap(16) as *i64; let sl5: *i64=sys_mmap(16) as *i64; if pco>=0 { nx_bc_str(meta,pco,msize,so5,sl5); poff=so5[0]; npc=sl5[0]/20 }
428 let lno: i64=nx_bc_dict_get(meta,0,msize,"length" as *u8,6); if lno>=0 { let iv6: *i64=sys_mmap(16) as *i64; nx_bc_int(meta,lno,msize,iv6); total=iv6[0] } else { total=tg_multifile_total(meta, msize, npc, plen) }
429 // seq1786: substitute the historical npc*plen fallback so BEHAVIOUR IS UNCHANGED, but
430 // publish WHICH bail fired -- a stuck torrent now names its own parse failure in
431 // download.status instead of silently reporting an invented total. A negative total
432 // must NEVER reach the piece math: pl_i = total - cur*plen would go wildly wrong.
433 if total < 0 {
434 if total == 0-1 { tg_status(statpath,name,0,0,0,npc,0,"meta-bail-nofiles" as *u8) }
435 if total == 0-2 { tg_status(statpath,name,0,0,0,npc,0,"meta-bail-notlist" as *u8) }
436 if total == 0-3 { tg_status(statpath,name,0,0,0,npc,0,"meta-bail-noprogress" as *u8) }
437 if total == 0-4 { tg_status(statpath,name,0,0,0,npc,0,"meta-bail-empty" as *u8) }
438 total = npc*plen
439 }
440 let nmo: i64=nx_bc_dict_get(meta,0,msize,"name" as *u8,4); if nmo>=0 { let so6: *i64=sys_mmap(16) as *i64; let sl6: *i64=sys_mmap(16) as *i64; if nx_bc_str(meta,nmo,msize,so6,sl6)>=0 { var k: i64=0; var nn: i64=sl6[0]; if nn>200 {nn=200} while k<nn { name[k]=meta[so6[0]+k]; k=k+1 } name[nn]=0 as u8 } }
441 let im: *u8=sys_mmap(16); mse_write(ctx,fd,im,nx_pw_build_msg(NX_PW_INTERESTED,0 as *u8,0,im))
442 // TAIL PATIENCE (the seedeval "IGNORING" fix): in the endgame (<= threshold pieces left) a
443 // holder is almost always a SEEDER that needs ~10-30s to optimistically-unchoke our pure-
444 // leecher. Extend this peer's read timeout so we WAIT for that unchoke instead of cycling
445 // off at the 6s connect timeout -- THE reason "76 peers HAVE the last piece but we never
446 // pull it". Only at the tail, so the bulk download keeps cycling fast off real chokers.
447 var done_n: i64=0; var dq: i64=0; while dq<npc { if (lbits[dq] as i64)==1 { done_n=done_n+1 } dq=dq+1 }
448 if npc - done_n <= TG_ENDGAME_THRESHOLD { sys_set_socket_timeout(fd, 45); patient=1 }
449 tg_seed_hello(ctx, fd, lbits, npc) // advertise what we hold + unchoke -> a leecher can TRADE us the tail piece
450 }
451 }
452 }
453 // LEAK ROOT-FIX 2026-07-30: the SEED/SERVE path (NX_PW_REQUEST, below) used to sys_mmap(rln+32)
454 // TWICE PER BLOCK SERVED and NEVER munmap -- ~40KB of leaked pages per served block, which is why
455 // long-lived children reached 1.5GB RSS (measured: 3 pids at 0.8-1.6GB, nx_leak_check CRITICAL).
456 // rln is BOUNDED (rln<=TG_MAGIC_16384 is checked at the serve guard), so these two buffers are
457 // LOOP-INVARIANT: map them ONCE here and reuse them every request. DO NOT move them back inside
458 // the loop -- that reintroduces the leak. This loop runs up to TG_MAGIC_2000000 iterations.
459 let sv_pbuf: *u8 = sys_mmap(TG_MAGIC_16384 + 64)
460 let sv_mbuf: *u8 = sys_mmap(TG_MAGIC_16384 + 64)
461 while go == 1 {
462 loops = loops + 1
463 if (loops % 512) == 0 { if tg_stop(ctlpath) == 1 { go = 0 } }
464 if loops > TG_MAGIC_2000000 { go = 0 }
465 if go == 1 {
466 let t_r0: i64 = sys_now_realtime_sec()
467 let blen: i64 = tg_read_msg(ctx, fd, mb, TG_MAGIC_40000)
468 if blen < 0 {
469 // Distinguish a recv TIMEOUT (blocked ~the socket timeout -> connection alive, peer just
470 // silent) from a real CLOSE/error (returns fast). While still waiting for a FIRST unchoke and
471 // inside the patience window, a timeout is NOT death: keep-alive and STAY so the seeder can
472 // rotate its optimistic-unchoke onto us. A fast return, or an active/expired peer -> give up.
473 var giveup: i64 = 1
474 if (sys_now_realtime_sec() - t_r0) >= 4 { if mdone==1 { if unchoked==0 { if cur<0 {
475 if (sys_now_realtime_sec() - t_start) < TG_UNCHOKE_WAIT {
476 let ka: *u8=sys_mmap(8); ka[0]=0 as u8; ka[1]=0 as u8; ka[2]=0 as u8; ka[3]=0 as u8; mse_write(ctx, fd, ka, 4); giveup=0
477 }
478 } } } }
479 if giveup == 1 { go = 0 }
480 }
481 if blen > 0 {
482 let id: i64 = nx_pw_msg_id(mb); let nt: i64 = 4 + blen
483 if id == PW_EXT {
484 let xi: i64 = mb[5] as i64
485 if xi == 0 {
486 let mo: i64 = nx_bc_dict_get(mb,6,nt,"m" as *u8,1); if mo>=0 { let uo: i64=nx_bc_dict_get(mb,mo,nt,"ut_metadata" as *u8,11); if uo>=0 { let iv: *i64=sys_mmap(16) as *i64; if nx_bc_int(mb,uo,nt,iv)>=0 { putm=iv[0] } } }
487 let so: i64 = nx_bc_dict_get(mb,6,nt,"metadata_size" as *u8,13); if so>=0 { let iv2: *i64=sys_mmap(16) as *i64; if nx_bc_int(mb,so,nt,iv2)>=0 { msize=iv2[0] } }
488 if putm>0 { if msize>0 { if rmeta==0 { nmeta=(msize+TG_MAGIC_16383)/TG_MAGIC_16384; var pi: i64=0; while pi<nmeta { let rb: *u8=sys_mmap(64); var ro: i64=pm_bstr(rb,0,"d8:msg_typei0e5:piecei" as *u8); ro=pm_bint(rb,ro,pi); ro=pm_bstr(rb,ro,"ee" as *u8); let rm: *u8=sys_mmap(128); mse_write(ctx,fd,rm,pm_build_ext(putm,rb,ro,rm)); pi=pi+1 } rmeta=1 } } }
489 }
490 if xi == 1 { if mdone==0 {
491 let pio: i64=nx_bc_dict_get(mb,6,nt,"piece" as *u8,5); var px: i64=0; if pio>=0 { let iv3: *i64=sys_mmap(16) as *i64; if nx_bc_int(mb,pio,nt,iv3)>=0 { px=iv3[0] } }
492 let raw: i64=nx_bc_skip(mb,6,nt); if raw>0 { let rl: i64=nt-raw; var c: i64=0; while c<rl { if px*TG_MAGIC_16384+c<TG_MAGIC_262144 { meta[px*TG_MAGIC_16384+c]=mb[raw+c] } c=c+1 } gmeta=gmeta+1 }
493 if gmeta>=nmeta { if nmeta>0 {
494 let gh: *u8=sys_mmap(20); sha1(meta,msize,gh); var eq: i64=1; var x: i64=0; while x<20 { if gh[x]!=ih[x] { eq=0; x=20 } else { x=x+1 } }
495 if eq==1 { mdone=1
496 let plo: i64=nx_bc_dict_get(meta,0,msize,"piece length" as *u8,12); let iv5: *i64=sys_mmap(16) as *i64; if plo>=0 { nx_bc_int(meta,plo,msize,iv5); plen=iv5[0] }
497 let pco: i64=nx_bc_dict_get(meta,0,msize,"pieces" as *u8,6); let so5: *i64=sys_mmap(16) as *i64; let sl5: *i64=sys_mmap(16) as *i64; if pco>=0 { nx_bc_str(meta,pco,msize,so5,sl5); poff=so5[0]; npc=sl5[0]/20 }
498 let lno: i64=nx_bc_dict_get(meta,0,msize,"length" as *u8,6); if lno>=0 { let iv6: *i64=sys_mmap(16) as *i64; nx_bc_int(meta,lno,msize,iv6); total=iv6[0] } else { total=tg_multifile_total(meta, msize, npc, plen) }
499 // seq1786: substitute the historical npc*plen fallback so BEHAVIOUR IS UNCHANGED, but
500 // publish WHICH bail fired -- a stuck torrent now names its own parse failure in
501 // download.status instead of silently reporting an invented total. A negative total
502 // must NEVER reach the piece math: pl_i = total - cur*plen would go wildly wrong.
503 if total < 0 {
504 if total == 0-1 { tg_status(statpath,name,0,0,0,npc,0,"meta-bail-nofiles" as *u8) }
505 if total == 0-2 { tg_status(statpath,name,0,0,0,npc,0,"meta-bail-notlist" as *u8) }
506 if total == 0-3 { tg_status(statpath,name,0,0,0,npc,0,"meta-bail-noprogress" as *u8) }
507 if total == 0-4 { tg_status(statpath,name,0,0,0,npc,0,"meta-bail-empty" as *u8) }
508 total = npc*plen
509 }
510 let nmo: i64=nx_bc_dict_get(meta,0,msize,"name" as *u8,4); if nmo>=0 { let so6: *i64=sys_mmap(16) as *i64; let sl6: *i64=sys_mmap(16) as *i64; if nx_bc_str(meta,nmo,msize,so6,sl6)>=0 { var k: i64=0; var nn: i64=sl6[0]; if nn>200 {nn=200} while k<nn { name[k]=meta[so6[0]+k]; k=k+1 } name[nn]=0 as u8 } }
511 let nf: i64=sys_openat_wr(npcpath,0x1a4); if nf>=0 { let nb: *u8=sys_mmap(16); nb[0]=(npc>>24)&0xff; nb[1]=(npc>>16)&0xff; nb[2]=(npc>>8)&0xff; nb[3]=npc&0xff; nb[4]=(plen>>24)&0xff; nb[5]=(plen>>16)&0xff; nb[6]=(plen>>8)&0xff; nb[7]=plen&0xff; cl_write_n(nf,nb,8); sys_close(nf) }
512 let mwf: i64=sys_openat_wr(metapath,0x1a4); if mwf>=0 { cl_write_n(mwf, meta, msize); sys_close(mwf) } // CACHE the verified metadata so sibling/recycled children skip the ut_metadata refetch
513 let im: *u8=sys_mmap(16); mse_write(ctx,fd,im,nx_pw_build_msg(NX_PW_INTERESTED,0 as *u8,0,im))
514 // reflect pieces ALREADY on disk (resume) instead of hardcoding 0 -- lbits
515 // was loaded from the donefile, so this shows preserved progress immediately.
516 var dn82: i64=0; var q82: i64=0; while q82<npc { if (lbits[q82] as i64)==1 { dn82=dn82+1 } q82=q82+1 }
517 tg_status(statpath,name,total,dn82*plen,dn82,npc,peers,"downloading" as *u8)
518 tg_seed_hello(ctx, fd, lbits, npc) // advertise + unchoke from first metadata so reciprocity bootstraps the bulk + 0%-non-starters
519 }
520 } }
521 } }
522 if xi == PEX_OUR_ID {
523 let pip: *i64 = sys_mmap(8*64) as *i64; let ppo: *i64 = sys_mmap(8*64) as *i64
524 let pc: i64 = pex_parse_added(mb, 6, nt, pip, ppo, 64)
525 if pc > 0 { tg_pex_append(pexpath, pip, ppo, pc) }
526 }
527 }
528 if id == NX_PW_UNCHOKE { unchoked=1; chmsg=0; sys_set_socket_timeout(fd, TG_DATA_TIMEOUT) } // DATA PHASE: a busy seeder's first block + inter-block gaps blow past the 6s connect timeout -> extend so we actually RECEIVE the pieces we just requested instead of bailing mid-request (the seedeval "23 seeders unchoke us, have=0 = IGNORING" fix)
529 if id == NX_PW_CHOKE { unchoked=0; if cur>=0 { tg_ppart_flush(spath, ppath, cur, plen, pbuf, cprefix); eg_clear(inflightpath, cur); cur=0-1; pbytes=0; cprefix=0 } } // choked: PERSIST the contiguous prefix (so the next window resumes it) + release the in-flight piece + return to waiting. DON'T give up here -- TG_UNCHOKE_WAIT bounds the wait so we outlast the optimistic-unchoke rotation (the "IGNORING" fix). Persisting the prefix is the reciprocity-starvation completion fix.
530 if id == NX_PW_BITFIELD { let bl2: i64=blen-1; var c: i64=0; while c<bl2 { if c<TG_MAGIC_40000 { peerbits[c]=mb[5+c] } c=c+1 } havebits=1 } // record which pieces THIS peer has
531 if id == NX_PW_HAVE { let hp: i64=_pw_get_u32(mb,5); let by: i64=hp/8; if by<TG_MAGIC_40000 { peerbits[by]=((peerbits[by] as i64) | (1 << (7 - (hp - by*8)))) as u8 } havebits=1; if avail_added==1 { if mdone==1 { am_add_have(availpath, hp) } } }
532 if id == NX_PW_PIECE {
533 let pidx: i64=_pw_get_u32(mb,5); let beg: i64=_pw_get_u32(mb,9); let bl: i64=blen-9
534 if pidx==cur { if beg+bl<=PCAP { var c: i64=0; while c<bl { pbuf[beg+c]=mb[13+c]; c=c+1 } pbytes=pbytes+bl; if beg==cprefix { cprefix=cprefix+bl } } }
535 // seq1734: a block landing beyond PCAP used to vanish here with NO else branch --
536 // silent data loss that made any plen>PCAP torrent retry forever at have=0. The cap
537 // is now 32 MiB, but the DROP MUST STAY VISIBLE: count it so a future oversized
538 // piece surfaces as a number instead of an eternal 0%.
539 if pidx==cur { if beg+bl>PCAP { dropped_blocks=dropped_blocks+1 } } // cprefix = length of the in-order prefix (what we can safely persist across a choke)
540 var pl_i: i64=plen; if cur==npc-1 { pl_i=total-cur*plen }
541 if pbytes>=pl_i { if pl_i>0 {
542 let g2: *u8=sys_mmap(20); sha1(pbuf,pl_i,g2); var ev: i64=1; var y: i64=0; while y<20 { if g2[y]!=meta[poff+cur*20+y] { ev=0; y=20 } else { y=y+1 } }
543 if ev==1 {
544 sys_lseek(outfd, cur*plen, 0); cl_write_n(outfd, pbuf, pl_i)
545 lbits[cur]=1 as u8; sys_lseek(donefd, cur, 0); let one: *u8=sys_mmap(4); one[0]=1 as u8; cl_write_n(donefd, one, 1)
546 have=have+1
547 var dn: i64=0; var q: i64=0; while q<npc { if (lbits[q] as i64)==1 { dn=dn+1 } q=q+1 }
548 tg_status(statpath,name,total,dn*plen,dn,npc,peers,"downloading" as *u8)
549 let hp2: *u8=sys_mmap(8); hp2[0]=((cur>>24)&0xff) as u8; hp2[1]=((cur>>16)&0xff) as u8; hp2[2]=((cur>>8)&0xff) as u8; hp2[3]=(cur&0xff) as u8; let hm: *u8=sys_mmap(16); mse_write(ctx,fd,hm,nx_pw_build_msg(NX_PW_HAVE,hp2,4,hm)) // advertise our NEW piece -> a peer needing it requests -> reciprocity
550 }
551 // seq1745: a SHA-1 verify failure used to be indistinguishable from success --
552 // same cleanup, no counter, no status. Four hours of failing every piece looked
553 // exactly like four hours of idle, which is why BANK-155 (done frozen at 30/2230
554 // since 13:22 while avail+wanted kept updating at 17:31) could not be diagnosed.
555 // Surfacing it costs one status write and DECIDES between the two live
556 // hypotheses: verify-failure vs reciprocity starvation. Same class as the PCAP
557 // silent drop -- a discard with no counter is invisible data loss.
558 if ev==0 { var dnf: i64=0; var qf: i64=0; while qf<npc { if (lbits[qf] as i64)==1 { dnf=dnf+1 } qf=qf+1 } tg_status(statpath,name,total,dnf*plen,dnf,npc,peers,"verify-fail" as *u8) }
559 tg_ppart_set(ppath, cur, 0); eg_clear(inflightpath, cur); cur=0-1; pbytes=0; cprefix=0 // piece done (or verify-failed) -> clear staged partial + no longer in-flight by this child
560 } }
561 }
562 if id == NX_PW_REQUEST { let ri: i64=_pw_get_u32(mb,5); let rqb: i64=_pw_get_u32(mb,9); let rln: i64=_pw_get_u32(mb,13)
563 if rln>0 { if rln<=TG_MAGIC_16384 { if ri>=0 { if ri<npc { if (lbits[ri] as i64)==1 { // we HAVE the requested piece -> SERVE it (seed back = reciprocity)
564 let sp: *u8=sv_pbuf
565 sp[0]=((ri>>24)&0xff) as u8; sp[1]=((ri>>16)&0xff) as u8; sp[2]=((ri>>8)&0xff) as u8; sp[3]=(ri&0xff) as u8
566 sp[4]=((rqb>>24)&0xff) as u8; sp[5]=((rqb>>16)&0xff) as u8; sp[6]=((rqb>>8)&0xff) as u8; sp[7]=(rqb&0xff) as u8
567 sys_lseek(outfd, ri*plen+rqb, 0); let got: i64=sys_read(outfd, (sp as i64 + 8) as *u8, rln)
568 if got==rln { let sm: *u8=sv_mbuf; mse_write(ctx, fd, sm, nx_pw_build_msg(NX_PW_PIECE, sp, 8+rln, sm)) }
569 } } } } } }
570 if mdone==1 { if unchoked==1 { if cur<0 {
571 sys_lseek(donefd, 0, 0); sys_read(donefd, lbits, npc) // refresh: skip pieces other parallel peers already finished
572 if wbuilt==0 { tg_build_wanted(meta, msize, npc, plen, wbits, tg_wanted_min()); let wf3: i64=tg_open_rw(wantedpath); if wf3>=0 { sys_lseek(wf3,0,0); cl_write_n(wf3, wbits, npc); sys_close(wf3) } wbuilt=1 }
573 var wpk: i64=0; while wpk<npc { if (wbits[wpk] as i64)==0 { pickbits[wpk]=1 as u8 } else { pickbits[wpk]=lbits[wpk] } wpk=wpk+1 } // pickbits: unwanted treated as done
574 // SMART LAYER (Phase 2): contribute THIS peer's availability to the shared cross-peer
575 // map ONCE -- deferred to here (not BITFIELD-receipt time) because the bitfield usually
576 // arrives before we know npc, and am_add_bitfield needs npc to know how far to scan.
577 if havebits==1 { if avail_added==0 { am_add_bitfield(availpath, peerbits, TG_MAGIC_40000, npc); avail_added=1 } }
578 if avbuf == (0 as *i64) { avbuf = sys_mmap(npc*8+64) as *i64; effbuf = sys_mmap(npc*8+64) as *i64; infbuf = sys_mmap(npc*8+64) as *i64 }
579 // SMART PICK (Phase 2): phase-1 = rarest NEEDED piece this peer HAS that no other child
580 // is already downloading (clean work-split -- kills the duplicate-download waste the old
581 // done-only scan allowed); phase-2 = in the tail (<= TG_ENDGAME_THRESHOLD obtainable
582 // pieces left, all in-flight) RACE the rarest to beat the one slow holder. eg_pick is
583 // self-sufficient (handles no-bitfield seeders + the avail map), so -1 means genuinely
584 // nothing for this peer right now -> free the slot (the pool refills with another peer).
585 var np: i64 = 0 - 1
586 if tg_stream_on(streampath) == 1 { am_load(inflightpath, infbuf, npc); np = tg_seq_pick(pickbits, peerbits, TG_MAGIC_40000, havebits, infbuf, npc) }
587 else { np = eg_pick(availpath, inflightpath, pickbits, peerbits, TG_MAGIC_40000, havebits, npc, TG_ENDGAME_THRESHOLD, avbuf, effbuf, infbuf) }
588 if np<0 { go=0 } else { cur=np; eg_mark(inflightpath, cur)
589 var pl_i: i64=plen; if cur==npc-1 { pl_i=total-cur*plen }
590 var pp: i64 = tg_ppart_get(ppath, cur); if pp >= pl_i { pp = 0 } // RESUME a staged prefix (defensive: stale/full -> fresh)
591 if pp > 0 { pbytes = tg_ppart_load(spath, cur, plen, pbuf, pp); cprefix = pbytes } else { pbytes = 0; cprefix = 0 }
592 let rq: *u8=sys_mmap(32); var b: i64=cprefix; while b<pl_i { var bk: i64=TG_MAGIC_16384; if pl_i-b<bk { bk=pl_i-b } mse_write(ctx,fd,rq,nx_pw_build_request(cur,b,bk,rq)); b=b+TG_MAGIC_16384 } // request only the MISSING blocks (skip the resumed prefix); one req buffer reused
593 }
594 } } }
595 if mdone==1 { if unchoked==0 { if cur<0 { if (sys_now_realtime_sec() - t_start) >= TG_UNCHOKE_WAIT { go=0 } } } } // give up ONLY after the unchoke-patience window elapses (covers choke-spammers that keep us choked-and-idle while sending messages, so the read-timeout path never fires). Bounds slot hold-time; productive/unchoked peers never reach here.
596 }
597 }
598 }
599 if cur>=0 { tg_ppart_flush(spath, ppath, cur, plen, pbuf, cprefix); eg_clear(inflightpath, cur) } // abandoned mid-piece (choke/timeout) -> PERSIST the contiguous prefix (resumed by the next child) + release the in-flight claim
600 sys_close(outfd); sys_close(donefd); sys_close(fd)
601 return have
602}
603
604// one HTTP/TCP tracker announce, appended to the peer set (dedup later). Bypasses ISP UDP blocking.
605func tg_http_one(host: *u8, port: i64, path: *u8, ih: *u8, pid: *u8, ips: *i64, ports: *i64, have: i64, max: i64) -> i64 {
606 let ip4: *u8 = sys_mmap(4)
607 if nx_dns_a_resolve(host, ip4) != 1 { return 0 }
608 let tip: *i64 = sys_mmap(8*80) as *i64; let tpo: *i64 = sys_mmap(8*80) as *i64
609 let c: i64 = ht_announce(host, ip4, port, path, ih, pid, TG_MAGIC_1000000000, tip, tpo, 80)
610 if c <= 0 { return 0 }
611 var added: i64 = 0; var k: i64 = 0
612 while k < c { if have + added < max { ips[have+added] = tip[k]; ports[have+added] = tpo[k]; added = added + 1 } k = k + 1 }
613 return added
614}
615
616// ---- DATA-DRIVEN WIDE tracker announce (Cardinal 11) ----------------------------------------------
617// Announce to EVERY HTTP/HTTPS tracker in <selfdir>/trackers.txt (one URL/line; '#' comments + blanks
618// skipped via tl_next_line) and append discovered peers to download.pex -- the SAME shared file BEP-11
619// PEX uses, so the parent's existing tg_pex_merge folds them into the live peer set. Widening the
620// tracker set = better odds of reaching the one seed holding a rare tail piece (the "stuck at 99%" fix
621// for poorly-seeded swarms). Absent file (e.g. dev box without trackers.txt) -> no-op; the synchronous
622// UDP + baked-HTTP + DHT discovery still runs. Returns peers appended.
623func tg_wide_to_pex(selfdir: *u8, outdir: *u8, ih: *u8, pid: *u8) -> i64 {
624 let path: *u8 = sys_mmap(640)
625 var pp: i64 = pm_bstr(path, 0, selfdir); pp = pm_bstr(path, pp, "/trackers.txt" as *u8); path[pp] = 0 as u8
626 let fd: i64 = sys_openat_rd(path)
627 if fd < 0 { return 0 }
628 let buf: *u8 = sys_mmap(TG_MAGIC_65536)
629 let n: i64 = sys_read(fd, buf, TG_MAGIC_65535); sys_close(fd)
630 if n <= 0 { return 0 }
631 let pexpath: *u8 = sys_mmap(640); var xp: i64 = pm_bstr(pexpath, 0, outdir); xp = pm_bstr(pexpath, xp, "/download.pex" as *u8); pexpath[xp] = 0 as u8
632 let tip: *i64 = sys_mmap(8*80) as *i64; let tpo: *i64 = sys_mmap(8*80) as *i64
633 let line: *u8 = sys_mmap(512)
634 var added: i64 = 0
635 var p2: i64 = tl_next_line(buf, n, 0, line)
636 while p2 >= 0 {
637 let cnt: i64 = ht_announce_url(line, ih, pid, TG_MAGIC_1000000000, tip, tpo, 80)
638 if cnt > 0 { tg_pex_append(pexpath, tip, tpo, cnt); added = added + cnt }
639 p2 = tl_next_line(buf, n, p2, line)
640 }
641 return added
642}
643// Run the wide announce in a DETACHED grandchild (double-fork -> reparents to init, no zombie) so a
644// firewalled tracker's ~127s connect() hang can NEVER stall the download (nx_http_client has no connect
645// timeout -- the very reason peers are fork-bounded). One-shot per worker run; a re-add starts a fresh
646// worker -> a fresh wide scan. The immediate child is reaped here; the grandchild is reaped by init.
647func tg_spawn_wide_announce(selfdir: *u8, outdir: *u8, ih: *u8, pid: *u8) -> i64 {
648 let p1: i64 = sys_fork()
649 if p1 == 0 {
650 let p2: i64 = sys_fork()
651 if p2 == 0 {
652 var fdx: i64 = 3; while fdx < 64 { sys_close(fdx); fdx = fdx + 1 }
653 tg_wide_to_pex(selfdir, outdir, ih, pid)
654 sys_exit(0)
655 }
656 sys_exit(0)
657 }
658 let st1: *i64 = sys_mmap(16) as *i64
659 sys_wait4(p1, st1, 0)
660 return 0
661}
662
663// MULTI-CHANNEL peer discovery -- the TEAM picks the set (UDP + HTTP/TCP); nothing external decides.
664// HTTP/TCP survives ISP UDP-tracker blocking AND a single tracker's announce rate-limit.
665func tg_announce_all(ih: *u8, pid: *u8, ips: *i64, ports: *i64, max: i64) -> i64 {
666 var have: i64 = cl_announce(ih, pid, ips, ports, max) // UDP open.stealth.si:80 (existing)
667 if have < 0 { have = 0 }
668 have = have + tg_http_one("open.stealth.si" as *u8, 80, "/announce" as *u8, ih, pid, ips, ports, have, max) // same tracker over TCP
669 have = have + tg_http_one("tracker.opentrackr.org" as *u8, TG_MAGIC_1337, "/announce" as *u8, ih, pid, ips, ports, have, max)
670 // DHT (BEP-5): trackerless peer discovery -- the ONLY source for tracker-less magnets (e.g. the
671 // operator's javdb magnets have no tr=). Append DHT peers to the set; reuses pid as the node id.
672 let dip: *i64 = sys_mmap(8*256) as *i64; let dpo: *i64 = sys_mmap(8*256) as *i64
673 let dc: i64 = dgp_find(ih, pid, dip, dpo, max)
674 var k: i64 = 0
675 while k < dc { if have < max { ips[have] = dip[k]; ports[have] = dpo[k]; have = have + 1 } k = k + 1 }
676 return have
677}
678
679// ---- BACKGROUND ANNOUNCE: THE PARENT MUST NEVER BLOCK ON THE NETWORK ------------------------------
680// tg_announce_all does a UDP tracker round (cl_announce), two DNS lookups + TCP tracker connects
681// (tg_http_one -> ht_announce) and a DHT get_peers (dgp_find). EVERY one of those can park the caller for
682// the kernel's full SYN / datagram timeout (~127s) against a black-holed tracker, and SO_RCVTIMEO does
683// NOT bound connect() -- so the 6s sys_set_socket_timeout calls on the PEER path buy nothing here.
684//
685// MEASURED 2026-07-31: worker parents sat in __inet_stream_connect and __skb_recv_datagram holding
686// EXACTLY 32 zombie children -- 32 == MAXP, the whole peer pool. While the parent is parked, the slot
687// reaper and the refill loop below it do not run, so every peer child that exits during the stall becomes
688// a zombie AND its slot is never refilled: the pool goes fully idle for the length of one stalled
689// announce. The zombies were the SYMPTOM that made it visible; the blocked parent was the defect.
690//
691// FIX: fork the announce. The child appends its peers to download.pex -- the same file BEP-11 PEX already
692// uses, that tg_wide_to_pex already publishes to, and that the parent already merges from on this exact
693// code path -- so discovery moves off the critical path with no new plumbing. The parent's loop stays hot
694// and its existing general wait4 sweep reaps the announce child. Peers land one cycle later; that costs
695// 30s of freshness and buys back a 2-minute total pool stall per bad tracker.
696//
697// One announce in flight at a time (slot[0]); a child still alive after TG_ANN_MAX_TICKS is SIGKILLed so
698// one wedged tracker cannot permanently silence discovery. Returns 1 if a fresh announce was started.
699func tg_announce_bg(ihb: *u8, mypid: *u8, pexpath: *u8, slot: *i64, stbuf: *i64, ticks: i64) -> i64 {
700 if slot[0] > 0 {
701 let r: i64 = sys_wait4(slot[0], stbuf, WNOHANG)
702 if r == 0 {
703 // Still running. Reclaim it if it outlived the watchdog, but NEVER block waiting on it --
704 // blocking here is the exact defect this function exists to remove.
705 if ticks - slot[1] >= TG_ANN_MAX_TICKS { nx_kill(slot[0], 9); slot[0] = 0 }
706 return 0
707 }
708 // r>0 = reaped right here; r<0 = the loop's general sweep already reaped it. Both mean FREE.
709 // Treating r<0 as "still running" would strand the slot on a dead pid and stop announcing forever.
710 slot[0] = 0
711 }
712 let p: i64 = sys_fork()
713 if p == 0 {
714 let aip: *i64 = sys_mmap(8 * 80) as *i64
715 let apo: *i64 = sys_mmap(8 * 80) as *i64
716 let c: i64 = tg_announce_all(ihb, mypid, aip, apo, 80)
717 if c > 0 { tg_pex_append(pexpath, aip, apo, c) }
718 sys_exit(0)
719 }
720 if p > 0 { slot[0] = p; slot[1] = ticks }
721 return 1
722}
723
724// on finish, split download.part into the real media files (multi-file torrents) so the operator sees a
725// playable REBDB-1026...mkv, not download.part. Best-effort: if the extractor elf is absent, exec fails
726// harmlessly (the manual extractor + .part remain). Streaming split lives in nx_torrent_extract.
727// The directory THIS worker binary lives in, read from /proc/self/cmdline argv[0] (mirrors the
728// daemon's d_self_dir). The extract + gallery-index elfs are deployed as the worker's SIBLINGS, so
729// they resolve on BOTH the dev box (/tmp) AND the NAS hub (/volume1/ai/torrent). Fallback "/tmp".
730func tg_self_dir() -> *u8 {
731 let buf: *u8 = sys_mmap(TG_MAGIC_1024)
732 let fd: i64 = sys_openat_rd("/proc/self/cmdline" as *u8)
733 var n: i64 = 0
734 if fd >= 0 { n = sys_read(fd, buf, 1023); sys_close(fd) }
735 if n <= 0 { let fb: *u8 = sys_mmap(8); fb[0]=47 as u8; fb[1]=116 as u8; fb[2]=109 as u8; fb[3]=112 as u8; fb[4]=0 as u8; return fb }
736 var z: i64 = 0; var dz: i64 = 0
737 while dz == 0 { if z >= n { dz = 1 } else { if buf[z] == (0 as u8) { dz = 1 } else { z = z + 1 } } }
738 var i: i64 = z - 1; var df: i64 = 0
739 while df == 0 { if i <= 0 { df = 1 } else { if buf[i] == (47 as u8) { buf[i] = 0 as u8; df = 1 } else { i = i - 1 } } }
740 if buf[0] == (0 as u8) { let fb: *u8 = sys_mmap(8); fb[0]=47 as u8; fb[1]=116 as u8; fb[2]=109 as u8; fb[3]=112 as u8; fb[4]=0 as u8; return fb }
741 return buf
742}
743// AUTO-SEED (the "sharing" half): on FULL completion, register this torrent in the seeder's registry so we
744// SHARE it back (download->seed reciprocity). srcpath = download.part = the EXACT concatenated payload the
745// piece hashes are computed over, so the seeder serves correct pieces for single AND multi file. total = the
746// .part file SIZE (== the exact byte total, no metadata re-parse). Idempotent (skips if the info_hash hex is
747// already registered). ONLY called on done_total>=npc -> never registers a partial (which would serve corrupt
748// zero-filled pieces to peers). regpath is a PARAM (config, not hardcoded) so it is gate-testable.
749func tg_autoseed(outdir: *u8, ih: *u8, plen: i64, name: *u8, regpath: *u8) -> i64 {
750 if plen <= 0 { return 0 }
751 let part: *u8 = sys_mmap(640); var po: i64=pm_bstr(part,0,outdir); po=pm_bstr(part,po,"/download.part" as *u8); part[po]=0 as u8
752 let pf: i64 = sys_openat_rd(part); if pf<0 { return 0 }
753 let total: i64 = sys_lseek(pf, 0, 2); sys_close(pf) // SEEK_END -> exact byte total
754 if total <= 0 { return 0 }
755 // ONLY seed a FULLY-complete torrent: a selective/complete-on-wanted download lacks the junk pieces, but the
756 // seeder advertises a FULL bitfield -> seeding a partial would serve corrupt (zero) junk pieces + get us banned.
757 let npcp: *u8=sys_mmap(640); var npo: i64=pm_bstr(npcp,0,outdir); npo=pm_bstr(npcp,npo,"/download.npc" as *u8); npcp[npo]=0 as u8
758 let nfd: i64=sys_openat_rd(npcp); if nfd<0 { return 0 } let nb: *u8=sys_mmap(16); let nn: i64=sys_read(nfd,nb,16); sys_close(nfd)
759 if nn<4 { return 0 } let npc_a: i64=((nb[0] as i64)<<24)|((nb[1] as i64)<<16)|((nb[2] as i64)<<8)|(nb[3] as i64)
760 let donp: *u8=sys_mmap(640); var dpo: i64=pm_bstr(donp,0,outdir); dpo=pm_bstr(donp,dpo,"/download.done" as *u8); donp[dpo]=0 as u8
761 let dfd: i64=sys_openat_rd(donp); if dfd<0 { return 0 } let dbf: *u8=sys_mmap(TG_MAGIC_262144); let dnn: i64=sys_read(dfd,dbf,TG_MAGIC_262144); sys_close(dfd)
762 var dc: i64=0; var di: i64=0; while di<dnn { if (dbf[di] as i64)==1 { dc=dc+1 } di=di+1 }
763 if dc <= 0 { return 0 } // nothing downloaded -> nothing to share (else OK: the seeder advertises+serves ONLY download.done pieces, so a selective/complete-on-wanted torrent seeds honestly)
764 let hd: *u8 = "0123456789abcdef" as *u8; let hx: *u8 = sys_mmap(48)
765 var i: i64=0; while i<20 { hx[i*2]=hd[((ih[i] as i64)>>4)&15]; hx[i*2+1]=hd[(ih[i] as i64)&15]; i=i+1 } hx[40]=0 as u8
766 let rf: i64 = sys_openat_rd(regpath)
767 if rf>=0 { let rb: *u8=sys_mmap(TG_MAGIC_262144); let rn: i64=sys_read(rf,rb,TG_MAGIC_262143); sys_close(rf)
768 var s: i64=0; while s+40<=rn { var m: i64=1; var k: i64=0; while k<40 { if rb[s+k]!=hx[k] { m=0; k=40 } else { k=k+1 } } if m==1 { return 0 } s=s+1 } } // already registered -> idempotent
769 let row: *u8=sys_mmap(TG_MAGIC_1024); var o: i64=0
770 o=pm_bstr(row,o,name); row[o]=9 as u8; o=o+1 // name \t
771 var j: i64=0; while j<40 { row[o]=hx[j]; o=o+1; j=j+1 } row[o]=9 as u8; o=o+1 // ih_hex40 \t
772 o=pm_bint(row,o,plen); row[o]=9 as u8; o=o+1 // piece_length \t
773 o=pm_bint(row,o,total); row[o]=9 as u8; o=o+1 // total \t
774 o=pm_bstr(row,o,part); row[o]=10 as u8; o=o+1 // srcpath \n
775 let wf: i64 = __syscall(SYS_OPENAT, AT_FDCWD, regpath as i64, 0x441, 0x1a4, 0, 0) // O_WRONLY|O_CREAT|O_APPEND
776 if wf>=0 { cl_write_n(wf, row, o); sys_close(wf) }
777 return 0
778}
779func tg_run_extract(outdir: *u8) -> i64 {
780 // CRITICAL (NAS): /tmp is mounted NOEXEC on the hub -- the old hardcoded "/tmp/...elf" execve here
781 // FAILED silently, so completed downloads stayed as download.part and never became playable media
782 // (and the gallery never refreshed). Resolve the SELF-DIR sibling FIRST (the deployed bundle path,
783 // /volume1/ai/torrent), then fall back to /tmp (dev box). In the child, the first execve replaces
784 // the process on success; it only returns on failure, so the second execve is the fallback. This is
785 // strictly safer than before on BOTH hosts. The extract elf being absent is still harmless (-> 127).
786 let sd: *u8 = tg_self_dir()
787 let exself: *u8 = sys_mmap(TG_MAGIC_1024); var eo: i64 = pm_bstr(exself,0,sd); eo=pm_bstr(exself,eo,"/nx_torrent_extract.sov.elf" as *u8); exself[eo]=0 as u8
788 let extmp: *u8 = "/tmp/nx_torrent_extract.sov.elf" as *u8
789 let exargv: *i64 = sys_mmap(32) as *i64; exargv[0]=exself as i64; exargv[1]=outdir as i64; exargv[2]=0
790 let exenvp: *i64 = sys_mmap(16) as *i64; exenvp[0]=0
791 let pid: i64 = sys_fork()
792 if pid==0 { sys_execve(exself, exargv, exenvp); sys_execve(extmp, exargv, exenvp); sys_exit(127) }
793 let stx: *i64 = sys_mmap(16) as *i64; sys_wait4(pid, stx, 0)
794 // GALLERY REFRESH: re-index the recordings/torrent video roots so the just-extracted media
795 // shows up + streams in the Nishi gallery. nx_galx_vidindex writes an ABSOLUTE index path so
796 // this is CWD-independent; exec failing (elf absent) is harmless (gallery just isn't refreshed).
797 let gself: *u8 = sys_mmap(TG_MAGIC_1024); var go: i64 = pm_bstr(gself,0,sd); go=pm_bstr(gself,go,"/nx_galx_vidindex.sov.elf" as *u8); gself[go]=0 as u8
798 let gtmp: *u8 = "/tmp/nx_galx_vidindex.sov.elf" as *u8
799 let gargv: *i64 = sys_mmap(16) as *i64; gargv[0]=gself as i64; gargv[1]=0
800 let genvp: *i64 = sys_mmap(16) as *i64; genvp[0]=0
801 let gpid: i64 = sys_fork()
802 if gpid==0 { sys_execve(gself, gargv, genvp); sys_execve(gtmp, gargv, genvp); sys_exit(127) }
803 let gstx: *i64 = sys_mmap(16) as *i64; sys_wait4(gpid, gstx, 0)
804 // ANALYSIS-ON-INGEST 2026-07-14: after re-indexing, trigger the derivation pipeline (thumbnail prewarm + NXVI
805 // duration/keyframe/markers) for the just-extracted media via the GUARDED galxpipeline hostctl sub
806 // (single-instance -> no stacking across concurrent torrents; it spawns the pipeline detached). Harmless if
807 // nx_hostctl is absent on this host (-> 127), matching the vidindex fork above.
808 let hcself: *u8 = "/volume1/homes/elderwesto/nishihost/nx_hostctl" as *u8
809 let pargv: *i64 = sys_mmap(16) as *i64; pargv[0]=hcself as i64; pargv[1]="galxpipeline" as *u8 as i64; pargv[2]=0
810 let penvp: *i64 = sys_mmap(16) as *i64; penvp[0]="PATH=/usr/bin:/bin" as *u8 as i64; penvp[1]=0
811 let ppid: i64 = sys_fork()
812 if ppid==0 { sys_execve(hcself, pargv, penvp); sys_exit(127) }
813 let pstx: *i64 = sys_mmap(16) as *i64; sys_wait4(ppid, pstx, 0)
814 return 0
815}
816
817func main(argc: i64, argv: *i64) -> i64 {
818 var magnet: *u8 = "magnet:?xt=urn:btih:dd8255ecdc7ca55fb0bbf81323d87062db1f6d1c&dn=Big+Buck+Bunny" as *u8
819 if argc >= 2 { magnet = argv[1] as *u8 }
820 var outdir: *u8 = "/mnt/c/Users/elder/Downloads/nishi-torrents" as *u8
821 if argc >= 3 { outdir = argv[2] as *u8 }
822 sys_mkdir(outdir, 0x1ff)
823 // SINGLE WORKER PER TORRENT (s-class robustness, the hardware rung): take a NON-BLOCKING exclusive
824 // flock on download.lock. If another worker for THIS torrent already holds it (a duplicate from a
825 // re-add / startup autoresume / the retry-sweep / a racing daemon restart), exit NOW instead of
826 // thrashing the swarm with redundant peer connections + DHT bootstraps -- the "lots of stalls /
827 // non-starters under load" failure mode. The kernel releases the lock when this process dies, so a
828 // dead worker's torrent is instantly re-spawnable. lockfd is intentionally kept open for our lifetime.
829 let lockp: *u8 = sys_mmap(640); var lkp: i64 = pm_bstr(lockp, 0, outdir); lkp = pm_bstr(lockp, lkp, "/download.lock" as *u8); lockp[lkp] = 0 as u8
830 let lockfd: i64 = sys_openat_wr(lockp, 0x1a4)
831 if lockfd >= 0 { if sys_flock(lockfd, SYS_LOCK_EX | SYS_LOCK_NB) != 0 { cl_w(1, "TGET already-running (download.lock held) -> exit clean\n" as *u8); sys_exit(0); return 0 } }
832
833 let ihb: *u8 = sys_mmap(20); let trk: *i64 = sys_mmap(8*32) as *i64; let ntrk: *i64 = sys_mmap(16) as *i64; let nm: *u8 = sys_mmap(512)
834 if mg_parse(magnet, ihb, trk, 32, ntrk, nm) != 1 { cl_w(1, "TGET verdict=RED reason=bad-magnet\n" as *u8); sys_exit(1); return 1 }
835 let mypid: *u8 = sys_mmap(20); let ps: *u8 = "-NX0001-000000000000" as *u8; var z: i64=0; while z<20 { mypid[z]=ps[z]; z=z+1 }
836
837 // emit a status BEFORE the (slow ~15-20s) DHT/tracker discovery so the UI shows "discovering peers"
838 // instead of looking dead/failed. Fresh add only -- a resume keeps its preserved have-count status.
839 if argc < 4 {
840 let estat: *u8 = sys_mmap(512); var es: i64=pm_bstr(estat,0,outdir); es=pm_bstr(estat,es,"/download.status" as *u8); estat[es]=0 as u8
841 tg_status(estat, nm, 0, 0, 0, 0, 0, "discovering peers" as *u8)
842 }
843 let ips: *i64 = sys_mmap(8*80) as *i64; let ports: *i64 = sys_mmap(8*80) as *i64
844 var np: i64 = tg_announce_all(ihb, mypid, ips, ports, 80)
845 cl_w(1, "TGET peers=" as *u8); cl_wn(1, np); cl_w(1, "\n" as *u8)
846 // ip-filter: load the blocklist ONCE (children inherit via fork COW); absent file -> ipf_n=0 -> no-op.
847 let ipf_arr: *i64 = sys_mmap(TG_MAGIC_80000*16 + 64) as *i64
848 let ipf_n: i64 = ipf_load("/volume1/ai/torrent/data/ipfilter.bin" as *u8, ipf_arr, TG_MAGIC_80000)
849 if ipf_n > 0 { let ipfb: i64 = np; np = tg_ipf_filter(ips, ports, np, ipf_arr, ipf_n); cl_w(1, "TGET ipfilter ranges=" as *u8); cl_wn(1, ipf_n); cl_w(1, " peers " as *u8); cl_wn(1, ipfb); cl_w(1, "->" as *u8); cl_wn(1, np); cl_w(1, "\n" as *u8) }
850 if np < 1 {
851 let sp2: *u8 = sys_mmap(512); var s2: i64=pm_bstr(sp2,0,outdir); s2=pm_bstr(sp2,s2,"/download.status" as *u8); sp2[s2]=0 as u8
852 tg_status(sp2, "no seeders found" as *u8, 0, 0, 0, 0, 0, "stalled" as *u8)
853 cl_w(1, "TGET verdict=RED reason=no-peers (tracker rate-limit? re-run)\n" as *u8); sys_exit(1); return 1
854 }
855
856 let outpath: *u8 = sys_mmap(512); var o: i64=pm_bstr(outpath,0,outdir); o=pm_bstr(outpath,o,"/download.part" as *u8); outpath[o]=0 as u8
857 let donepath: *u8 = sys_mmap(512); var d: i64=pm_bstr(donepath,0,outdir); d=pm_bstr(donepath,d,"/download.done" as *u8); donepath[d]=0 as u8
858 let npcpath: *u8 = sys_mmap(512); var nn2: i64=pm_bstr(npcpath,0,outdir); nn2=pm_bstr(npcpath,nn2,"/download.npc" as *u8); npcpath[nn2]=0 as u8
859 let metapath: *u8 = sys_mmap(512); var mpp: i64=pm_bstr(metapath,0,outdir); mpp=pm_bstr(metapath,mpp,"/download.meta" as *u8); metapath[mpp]=0 as u8 // shared metadata cache (fetch once, not per-child)
860 let statpath: *u8 = sys_mmap(512); var s: i64=pm_bstr(statpath,0,outdir); s=pm_bstr(statpath,s,"/download.status" as *u8); statpath[s]=0 as u8
861 let ctlpath: *u8 = sys_mmap(512); var cc: i64=pm_bstr(ctlpath,0,outdir); cc=pm_bstr(ctlpath,cc,"/download.control" as *u8); ctlpath[cc]=0 as u8
862 let availpath: *u8 = sys_mmap(512); var ap: i64=pm_bstr(availpath,0,outdir); ap=pm_bstr(availpath,ap,"/download.avail" as *u8); availpath[ap]=0 as u8 // shared cross-peer rarity map
863 let inflightpath: *u8 = sys_mmap(512); var ip2: i64=pm_bstr(inflightpath,0,outdir); ip2=pm_bstr(inflightpath,ip2,"/download.inflight" as *u8); inflightpath[ip2]=0 as u8 // shared in-flight refcount (endgame)
864 let streampath: *u8 = sys_mmap(512); var sp3: i64=pm_bstr(streampath,0,outdir); sp3=pm_bstr(streampath,sp3,"/download.stream" as *u8); streampath[sp3]=0 as u8 // stream-mode flag (sequential) when present
865 let pexpath: *u8 = sys_mmap(512); var pxp: i64=pm_bstr(pexpath,0,outdir); pxp=pm_bstr(pexpath,pxp,"/download.pex" as *u8); pexpath[pxp]=0 as u8 // shared PEX-harvested peer log (BEP-11)
866 // fresh add (argc<4) truncates done/npc/part for a clean run; RESUME (argv[3] present) keeps the
867 // partial so the worker continues from the donefile bitmap instead of re-downloading from zero.
868 if argc < 4 {
869 let z1: i64=sys_openat_wr(donepath,0x1a4); if z1>=0 { sys_close(z1) }
870 let z2: i64=sys_openat_wr(npcpath,0x1a4); if z2>=0 { sys_close(z2) }
871 let z3: i64=sys_openat_wr(outpath,0x1a4); if z3>=0 { sys_close(z3) }
872 let z5: i64=sys_openat_wr(pexpath,0x1a4); if z5>=0 { sys_close(z5) } // fresh add: drop stale harvested peers
873 }
874 // DATA-DRIVEN WIDE tracker scan (one-shot, detached -> download.pex; off the parent's critical path
875 // so a hung tracker can't stall the grind). Fires for fresh AND resume, so a re-added stuck torrent
876 // re-scans the wider tracker set for the seed holding its missing tail piece. Runs AFTER the pex
877 // truncate above so a fresh add never wipes the peers it appends.
878 tg_spawn_wide_announce(tg_self_dir(), outdir, ihb, mypid)
879 // fresh add: show "resolving/started". RESUME: leave the pre-pause status intact (preserved
880 // have-count stays visible) so the UI doesn't flash back to 0% while metadata re-resolves.
881 if argc < 4 { tg_status(statpath,"resolving" as *u8,0,0,0,0,np,"started" as *u8) }
882
883 // SMART LAYER: zero the shared availability + in-flight maps ONCE before any child runs (both fresh
884 // + resume -- both are per-session, rebuilt from THIS run's live swarm). 262144 = the donefile ceiling.
885 am_init(availpath, TG_MAGIC_262144)
886 eg_init(inflightpath, TG_MAGIC_262144)
887
888 let st: *i64 = sys_mmap(16) as *i64
889 var done_total: i64 = 0; var npc: i64 = 0
890 var fin: i64 = 0
891 // S-CLASS THROUGHPUT: a self-refilling pool of MAXP download children. Each child connects to ONE
892 // peer (ring round-robin over the discovered set), downloads the pieces that peer serves, and EXITS
893 // the instant it is choked/idle (see tg_from_peer) -- so a choking peer can never freeze a slot.
894 // Every 2s the parent reaps dead children and IMMEDIATELY refills every free slot from the peer ring
895 // (no gate). The pool stays full and constantly cycles connections => continuous optimistic-unchokes
896 // => sustained throughput instead of "fast burst then freeze". Re-announce every ~30s for fresh peers.
897 // seq1402/seq1410 -- POOL WIDTH IS DERIVED, NOT DECLARED. This was `let MAXP: i64 = 32`, a hardcoded
898 // literal (rule-11) that made every machine behave identically: the NAS (4c/8t), the west server, the
899 // laptop and any future box all ran 32 slots, sized for hardware nobody measured. Worse, 32 was
900 // PER-INSTANCE, and nx_procchurn measured 143 forks/sec = ~9 concurrent instances = ~288 concurrent
901 // slots against a box with 8 threads. NOBODY CHOSE 143; it was 32 x N.
902 // LAW: A PER-INSTANCE LIMIT IS NOT A LIMIT -- N instances multiply it and nobody ever picks N.
903 // What is NOT changed: the aggressive refill/cycling below. Constant connection cycling IS the
904 // BitTorrent throughput strategy (continuous optimistic-unchokes); throttling it would attack the
905 // feature to flatten a graph (rule 25). Only the WIDTH is governed; inside the budget the pool stays
906 // exactly as aggressive as it ever was.
907 // slots_per_cpu=4 reproduces the historical 32 on this 8-thread NAS => BEHAVIOUR-PRESERVING HERE,
908 // and correctly scales up on a bigger box / down inside a cgroup. Governor gate: 18/18 GREEN.
909 // ⚠LANDING BLOCKED BY LAYERING, NOT BY DESIGN: nx_poolwidth_lib lives in _hdl_build/ and a runtime/
910 // organ CANNOT import from there (`nx_compile_x86: expand_imports failed`). The governor is built and
911 // gate-proven 18/18; it must MOVE to runtime/ (the primitive layer, beside nx_hw.nx) before this line
912 // can become `pw_width(pw_budget_from_hw(4,8), pw_count_procs("nx_torrent_get"), pw_load_centi(), 800, 2, 32)`.
913 // Moving it needs care: a second copy under the same basename would SHADOW (seq207/284 dual-copy).
914 // REVERTED 2026-07-30 (seq1666/seq1681). The pg_* governor that stood here was on the WRONG AXIS
915 // and cost real downloads. A peer slot is a socket plus a piece buffer on a NETWORK-BLOCKED child:
916 // it consumes almost no CPU, so budgeting slots as cpus x 4 spent capability to relieve a resource
917 // peer slots never consumed. MEASURED HARM: nx_torstat returned VERDICT=IGNORING on BANK-155
918 // (2200/2200 missing pieces that DO have holders among connected peers) and ajvr-166 (468/1069)
919 // = 2668 fetchable pieces never pulled -- the exact signature of too few slots, because a pure
920 // leecher never wins optimistic-unchoke before its child times out.
921 // The block above claimed slots_per_cpu=4 was BEHAVIOUR-PRESERVING (4 x 8 = the historical 32).
922 // IT WAS NOT: 32 became the AGGREGATE budget SPLIT across instances. nx_poolgov_gate measures it
923 // directly -- T1 = 3 slots each across 9 instances, T6 = 2 under real load. 288 -> 27 aggregate,
924 // shipped believing it was a no-op, and the gate recorded the squeeze as a PASS.
925 // LAW: A GATE PROVES THE MECHANISM IS ARITHMETICALLY SOUND, NEVER THAT THE POLICY IS RIGHT.
926 // The per-instance-multiplication insight above IS STILL CORRECT and must not be lost: 32 x N
927 // instances is a width nobody chose. But the aggregate cap has to be derived from the resource
928 // slots ACTUALLY consume -- fd/socket budget (RLIMIT_NOFILE), or measured network saturation --
929 // never from CPU count, and never squeezed by a loadavg that on this host is dominated by
930 // D-state disk wait (measured: load 16.58 with only 4 runnable on 8 cores).
931 // Restoring the proven constant until that governor exists (rule 3: stop patching, rewrite).
932 let MAXP: i64 = 32
933 let cpids: *i64 = sys_mmap(8 * 64) as *i64 // child pid per slot (0/neg = free)
934 // seq1317 PER-SLOT COOLDOWN. This pool refilled EVERY free slot on EVERY tick with a fresh fork and
935 // no backoff whatsoever, so a peer that fails instantly frees its slot and is re-forked ~2s later,
936 // forever -- a self-inflicted fork treadmill (the perf lane measured bursts of 179 forks/s against an
937 // 85/s since-boot average). A slot is now ineligible for TG_SLOT_COOLDOWN ticks after it is FILLED,
938 // which bounds the refill rate structurally without needing to classify why a child died: a healthy
939 // long-lived peer is unaffected (its slot is occupied anyway), while a dead-peer slot stops spinning.
940 let cnext: *i64 = sys_mmap(8 * 64) as *i64 // earliest tick this slot may be refilled
941 var ci: i64 = 0
942 while ci < 64 { cnext[ci] = 0; ci = ci + 1 }
943 // Background-announce state: [0] = announce child pid (0 = idle), [1] = the tick it was forked.
944 // Allocated ONCE rather than per call so a multi-hour download does not leak a page per announce.
945 let annslot: *i64 = sys_mmap(16) as *i64
946 annslot[0] = 0; annslot[1] = 0
947 let annst: *i64 = sys_mmap(8) as *i64
948 var ring: i64 = 0 // round-robin index over the peer set
949 var s0: i64 = 0
950 while s0 < MAXP {
951 let pid: i64 = sys_fork()
952 if pid == 0 { tg_from_peer(ips[ring % np], ports[ring % np], ihb, mypid, outpath, donepath, npcpath, metapath, statpath, ctlpath, availpath, inflightpath, streampath, pexpath, np); sys_exit(0) }
953 cpids[s0] = pid; ring = ring + 1; s0 = s0 + 1
954 }
955 let nslots: i64 = MAXP
956 var last_done: i64 = 0 - 1; var stall_ticks: i64 = 0; var ticks: i64 = 0; var plen_p: i64 = 0
957 let lbp: *u8 = sys_mmap(TG_MAGIC_262144); let avp: *i64 = sys_mmap(TG_MAGIC_262144 * 8) as *i64; var done_rn: i64 = 0; var finalized: i64 = 0 // auto-finalize tail (reuse donefile + avail map)
958 // SELECTIVE DOWNLOAD completion: DONE when the WANTED pieces (not all npc) are complete -> a junk/sample file
959 // whose pieces are unavailable no longer blocks the torrent. wbp=download.wanted mask (sibling of donepath);
960 // effdone = have|unwanted (so the tail/finalize logic ignores skipped junk). No download.wanted -> all-npc.
961 let wbp: *u8 = sys_mmap(TG_MAGIC_262144); let effdone: *u8 = sys_mmap(TG_MAGIC_262144)
962 let wantedpath_p: *u8 = sys_mmap(640); var wpp: i64=0; var lsl: i64=0; var wpi: i64=0
963 while donepath[wpi]!=(0 as u8) { if donepath[wpi]==(47 as u8) { lsl=wpi } wpi=wpi+1 }
964 wpi=0; while wpi<=lsl { wantedpath_p[wpp]=donepath[wpi]; wpp=wpp+1; wpi=wpi+1 } wpp=pm_bstr(wantedpath_p, wpp, "download.wanted" as *u8); wantedpath_p[wpp]=0 as u8
965 var wanted_count: i64=0; var wanted_done: i64=0
966 while fin == 0 {
967 sys_sleep_ms(TG_MAGIC_2000); ticks = ticks + 1
968 if tg_stop(ctlpath) == 1 { fin = 1 } else {
969 var s: i64 = 0
970 while s < nslots { if cpids[s] > 0 { let r: i64 = sys_wait4(cpids[s], st, WNOHANG); if r > 0 { cpids[s] = 0 } } s = s + 1 }
971 while sys_wait4(0 - 1, st, WNOHANG) > 0 { } // reap ANY other exited child (untracked bounded-connect/announce forks) -- nx_zombie_audit flagged these leaking
972 let df: i64=sys_openat_rd(donepath); if df>=0 { let rn: i64=sys_read(df, lbp, TG_MAGIC_262144); sys_close(df); done_rn=rn; var dn: i64=0; var q: i64=0; while q<rn { if (lbp[q] as i64)==1 { dn=dn+1 } q=q+1 } done_total=dn }
973 let nf: i64=sys_openat_rd(npcpath); if nf>=0 { let nb: *u8=sys_mmap(16); let rn2: i64=sys_read(nf,nb,16); sys_close(nf); if rn2>=4 { npc=((nb[0] as i64)<<24)|((nb[1] as i64)<<16)|((nb[2] as i64)<<8)|(nb[3] as i64) } if rn2>=8 { plen_p=((nb[4] as i64)<<24)|((nb[5] as i64)<<16)|((nb[6] as i64)<<8)|(nb[7] as i64) } }
974 // SELECTIVE: recompute wanted_count/wanted_done + effdone(have|unwanted) from fresh lbp + download.wanted
975 wanted_count=0; wanted_done=0; var wloaded: i64=0
976 if npc > 0 { let wf: i64=sys_openat_rd(wantedpath_p); if wf>=0 { let wn: i64=sys_read(wf, wbp, TG_MAGIC_262144); sys_close(wf); if wn>=npc { wloaded=1 } }
977 var eq: i64=0
978 while eq<npc { var hv: i64=0; if eq<done_rn { if (lbp[eq] as i64)==1 { hv=1 } }
979 if wloaded==1 { if (wbp[eq] as i64)==1 { wanted_count=wanted_count+1; if hv==1 { wanted_done=wanted_done+1; effdone[eq]=1 as u8 } else { effdone[eq]=0 as u8 } } else { effdone[eq]=1 as u8 } }
980 else { effdone[eq]=lbp[eq]; if hv==1 { wanted_done=wanted_done+1 } }
981 eq=eq+1 }
982 if wloaded==0 { wanted_count=npc } }
983 if wanted_count > 0 { if wanted_done >= wanted_count { fin = 1 } } // WANTED-complete (junk pieces ignored)
984 if fin == 0 { if (ticks % 15) == 0 { tg_announce_bg(ihb, mypid, pexpath, annslot, annst, ticks); np = tg_pex_merge(pexpath, ips, ports, np, 80); np = tg_ipf_filter(ips, ports, np, ipf_arr, ipf_n) } } // fresh peers every ~30s (tracker+DHT+harvested PEX); ip-filter the refreshed list
985 if fin == 0 {
986 var s2: i64 = 0
987 while s2 < MAXP { if cpids[s2] <= 0 { if ticks >= cnext[s2] {
988 let pid2: i64 = sys_fork()
989 if pid2 == 0 { tg_from_peer(ips[ring % np], ports[ring % np], ihb, mypid, outpath, donepath, npcpath, metapath, statpath, ctlpath, availpath, inflightpath, streampath, pexpath, np); sys_exit(0) }
990 cpids[s2] = pid2; cnext[s2] = ticks + TG_SLOT_COOLDOWN; ring = ring + 1
991 } } s2 = s2 + 1 }
992 }
993 if done_total == last_done { stall_ticks = stall_ticks + 1 } else { stall_ticks = 0 }
994 last_done = done_total
995 // KEEP GRINDING THE TAIL: never abandon a partial download while peers exist. On a plateau
996 // (no new piece for ~16s) re-announce HARD every ~16s for fresh peers/seeders + keep cycling
997 // the pool, and publish an HONEST status that PRESERVES the have-count (name=nm, real bytes)
998 // instead of flashing "done=0 stalled". Real S-class clients grind the tail, they don't quit.
999 if fin == 0 { if stall_ticks >= 8 {
1000 if (stall_ticks % 8) == 0 { tg_announce_bg(ihb, mypid, pexpath, annslot, annst, ticks); np = tg_pex_merge(pexpath, ips, ports, np, 80); np = tg_ipf_filter(ips, ports, np, ipf_arr, ipf_n) }
1001 tg_status(statpath, nm, tg_meta_total_of(metapath, npc, plen_p), done_total*plen_p, done_total, npc, np, "downloading (grinding tail)" as *u8)
1002 } }
1003 if npc == 0 { if ticks > 90 { fin = 1 } } // never resolved metadata in ~180s -> give up
1004 // AUTO-FINALIZE the tail: don't grind a hopeless last piece for 20 min or leave a 99%-done
1005 // .part the operator can't open. On a sustained plateau, EXTRACT what we have NOW (a playable
1006 // file, "most of a film") and stop -- (a) >=90% held + 60s plateau, or (b) the pieces we still
1007 // need are provably UNOBTAINABLE here (no connected peer has them -> avail 0) after a 2-min
1008 // plateau. This is the no-seeder-last-piece recovery (the 300MIUM-class fix).
1009 // TAIL POLICY: at >=90% + plateau, EXTRACT a playable file ONCE (operator watches the "most of a
1010 // film" now) but KEEP GRINDING while the missing pieces are OBTAINABLE (a connected peer has them).
1011 // Only STOP when the tail is genuinely UNOBTAINABLE (avail 0) or the 20-min ultimate giveup below.
1012 // BUG FIXED 2026-07-04: the old code STOPPED (fin=1) at 90%+60s UNCONDITIONALLY -> abandoned an
1013 // obtainable last piece forever whenever the holders took >60s to optimistically-unchoke our
1014 // leecher ("stuck at 99% with 177 holders"). Now an obtainable tail is ground until it lands.
1015 if fin == 0 { if npc > 0 { if done_total > 0 { if done_rn >= npc {
1016 am_load(availpath, avp, npc); let obt: i64 = eg_obtainable_remaining(effdone, avp, npc) // effdone ignores unwanted junk
1017 if finalized == 0 { if stall_ticks >= 30 { if wanted_done * 100 >= wanted_count * 90 {
1018 tg_run_extract(outdir); finalized = 1
1019 tg_status(statpath, nm, npc*plen_p, done_total*plen_p, done_total, npc, np, "playable extracted - still grinding the last piece(s)" as *u8)
1020 } } }
1021 if stall_ticks >= 60 { if obt == 0 {
1022 tg_run_extract(outdir)
1023 tg_status(statpath, nm, npc*plen_p, done_total*plen_p, done_total, npc, np, "finalized - swarm has no more (tail unobtainable)" as *u8)
1024 fin = 1
1025 } }
1026 } } } }
1027 if npc > 0 { if stall_ticks > 600 { fin = 1 } } // ultimate giveup: ~20 min of zero new pieces -> exit (re-add resumes)
1028 }
1029 }
1030 var k: i64 = 0; while k < nslots { if cpids[k] > 0 { nx_kill(cpids[k], 9); sys_wait4(cpids[k], st, 0) } k = k + 1 }
1031 cl_w(1, "TGET done_pieces=" as *u8); cl_wn(1, done_total); cl_w(1, "/" as *u8); cl_wn(1, npc); cl_w(1, " ticks=" as *u8); cl_wn(1, ticks); cl_w(1, " stall=" as *u8); cl_wn(1, stall_ticks); cl_w(1, " np=" as *u8); cl_wn(1, np); cl_w(1, " stop=" as *u8); cl_wn(1, tg_stop(ctlpath)); cl_w(1, " -> " as *u8); cl_w(1, outpath); cl_w(1, "\n" as *u8)
1032 if wanted_count > 0 { if wanted_done >= wanted_count { cl_w(1, "TGET verdict=GREEN COMPLETE (wanted="); cl_wn(1, wanted_done); cl_w(1, "/"); cl_wn(1, wanted_count); cl_w(1, ")\n" as *u8)
1033 let asreg: *u8=sys_mmap(TG_MAGIC_1024); var aso: i64=pm_bstr(asreg,0,tg_self_dir()); aso=pm_bstr(asreg,aso,"/seed_index.conf" as *u8); asreg[aso]=0 as u8
1034 tg_autoseed(outdir, ihb, plen_p, nm, asreg) // SHARE it back (skips partials internally: only FULL npc seeds)
1035 tg_run_extract(outdir); sys_exit(0); return 0 } }
1036 if done_total > 0 {
1037 if finalized == 0 { // not already extracted by the in-loop auto-finalize
1038 tg_status(statpath, nm, npc*plen_p, done_total*plen_p, done_total, npc, np, "partial - re-add to resume" as *u8)
1039 if npc > 0 { if done_total * 100 >= npc * 90 { tg_run_extract(outdir) } } // near-complete (>=90%) -> auto-extract playable media
1040 }
1041 cl_w(1, "TGET verdict=PARTIAL\n" as *u8); sys_exit(0); return 0
1042 }
1043 cl_w(1, "TGET verdict=RED reason=no-pieces\n" as *u8)
1044 tg_status(statpath, nm, 0, 0, done_total, npc, np, "stalled - no seeder served data" as *u8)
1045 sys_exit(1)
1046 return 1
1047}