clover's git

ix

1	//! The `Loader` is responsible for resolving, fetching and caching files from
2	//! URL sources. `Loader` is designed to run on it's own thread (the "Loader
3	//! Thread"), and communicate with the compiler core via the `enqueue` function
4	//! and the `completion_cb` callback.
5	//!
6	//! Note that this subsystem is entirely self-contained in the `loader` module,
7	//! and is an optional dependency of the compiler. This allows testing the
8	//! compiler without network determinism, and stress+fuzz testing `Loader`.
9	//!
10	//! The loader's thread-safe allocator owns source file contents.
11
12	/// Shared http client. This also stores the Loader's allocator
13	http: std.http.Client,
14
15	/// This map is accessed by multiple threads and is covered by `entries_mutex`
16	entries: std.StringArrayHashMapUnmanaged(Module),
17	entries_lock: Thread.RwLock,
18
19	/// Pre-allocated up to the max concurrent requests. This is
20	/// allocated at thread start up, and kept alive forever.
21	ongoing: OngoingLoad.List.Slice,
22
23	/// Stores the highest index in `ongoing` that is currently active.
24	/// If this is `0`, then there are no ongoing downloads.
25	last_ongoing: u16,
26
27	/// Thread-safe queue for pending tasks. They are moved
28	/// to `ongoing_downloads` when they initiate.
29	queue: Task.Queue,
30
31	/// External callback when a fetch completes. Called on the Loader thread.
32	completion_ctx: ?*anyopaque,
33	completion_cb: ?*const anyopaque, // cannot use CompletionCallback because dependency loop
34
35	/// This is the parent progress node (or the program root node). This is
36	/// distinct from `progress` because the latter may be stopped and started as
37	/// the thread sleeps (when there are no pending downloads)
38	root_progress: std.Progress.Node,
39	progress: std.Progress.Node,
40
41	cache: ?std.fs.Dir,
42
43	/// To ensure run is not being used on more than one thread.
44	run_lock: std.debug.SafetyLock,
45
46	redirect_behavior: std.http.Client.Request.RedirectBehavior,
47
48	pub const LoaderOptions = struct {
49	allocator: Allocator,
50	max_concurrent_downloads: u16 = 32,
51	cache: ?std.fs.Dir = null,
52	/// Pass the parent progress node
53	progress: std.Progress.Node = .{ .index = .none },
54	completion_cb: ?*const CompletionCallback = null,
55	completion_ctx: ?*anyopaque = null,
56	redirect_behavior: std.http.Client.Request.RedirectBehavior = @enumFromInt(3),
57	};
58
59	pub fn init(options: LoaderOptions) !Loader {
60	var ongoing = OngoingLoad.List{};
61	try ongoing.setCapacity(options.allocator, options.max_concurrent_downloads);
62	ongoing.len = options.max_concurrent_downloads;
63	errdefer ongoing.deinit(options.allocator);
64	const ongoing_slice = ongoing.slice();
65
66	@memset(ongoing_slice.items(.state), .unused);
67	@memset(ongoing_slice.items(.buffer), .{});
68
69	return .{
70	.http = .{ .allocator = options.allocator },
71	.root_progress = options.progress,
72	.ongoing = ongoing_slice,
73	.completion_ctx = options.completion_ctx,
74	.completion_cb = @ptrCast(options.completion_cb),
75	.redirect_behavior = options.redirect_behavior,
76	.queue = .{},
77	.progress = .{ .index = .none },
78	.entries = .{},
79	.entries_lock = .{},
80	.run_lock = .{},
81	.last_ongoing = 0,
82	.cache = options.cache,
83	};
84	}
85
86	pub fn deinit(loader: *Loader) void {
87	loader.entries_lock.lock();
88
89	const entries = loader.entries.entries.slice();
90	for (entries.items(.key), entries.items(.value)) |k, v| {
91	loader.allocator().free(k);
92	v.deinit(loader.allocator());
93	}
94	loader.entries.deinit(loader.allocator());
95
96	for (
97	0..,
98	loader.ongoing.items(.state)[0..loader.last_ongoing],
99	) |i, state| {
100	if (state == .unused) continue;
101	const slot: OngoingLoad.Index = @enumFromInt(i);
102	slot.request(loader).deinit();
103	slot.buffer(loader).deinit(loader.allocator());
104	slot.progress(loader).end();
105	}
106
107	loader.progress.end();
108	loader.ongoing.deinit(loader.allocator());
109
110	// get around assertion in http client
111	// TODO: this might be wrong
112	loader.http.connection_pool.deinit(loader.allocator());
113	loader.http.connection_pool = .{};
114
115	loader.http.deinit();
116
117	loader.* = undefined;
118	}
119
120	pub const Scheme = enum {
121	file,
122	http,
123	https,
124
125	pub fn fromSlice(scheme: []const u8) ?Scheme {
126	return std.meta.stringToEnum(Scheme, scheme);
127	}
128	};
129
130	pub const EnqueueOptions = struct {
131	/// This data is duplicated if stored into `Loader`
132	uri_text: []const u8,
133	};
134
135	/// Retrieve a module from the Loader, returning immediately if it is already
136	/// available. If a module load is pending, an index is returned, which can be
137	/// looked up later when needed.
138	pub fn enqueue(loader: *Loader, opts: EnqueueOptions) Allocator.Error!EnqueueResult {
139	// give a assertion fail on the thread that enqueued it
140	// safe builds will still panic, but instead on the loader's thread
141	if (@import("builtin").mode == .Debug) {
142	_ = Uri.parse(opts.uri_text) catch
143	unreachable; // invalid URI
144	}
145
146	const index = index: {
147	loader.entries_lock.lock();
148	defer loader.entries_lock.unlock();
149
150	// TODO: consider overwriting space in .dead modules
151
152	const gop = try loader.entries.getOrPut(loader.allocator(), opts.uri_text);
153	errdefer gop.value_ptr.* = .dead;
154
155	if (gop.found_existing) {
156	return switch (gop.value_ptr.*) {
157	.available => |data| .{ .available = data },
158	.fetching => .{ .fetching = @enumFromInt(gop.index) },
159	.failed => |fail| .{ .failed = fail },
160	.dead => {
161	gop.value_ptr.* = .fetching;
162	break :index gop.index;
163	},
164	};
165	}
166
167	gop.value_ptr.* = .fetching;
168	gop.key_ptr.* = try loader.allocator().dupe(u8, opts.uri_text);
169
170	break :index gop.index;
171	};
172	const module: Module.Index = @enumFromInt(index);
173
174	errdefer {
175	// Returning an error from enqueue means that the fetch was
176	// not actually enqueued (due to OOM), however since we
177	// already allocated the slot in the hashmap, we'll replace
178	// it with a tombstone.
179	loader.entries_lock.lockShared();
180	defer loader.entries_lock.unlockShared();
181
182	loader.entries.values()[index] = .{ .dead = {} };
183	}
184
185	// Prepare the node
186	const node = try loader.allocator().create(Task.Node);
187	node.* = .{ .data = .{
188	.uri_text = opts.uri_text,
189	.module = module,
190	} };
191	loader.queue.add(node);
192
193	return .{ .fetching = module };
194	}
195
196	/// This function is only used for testing basic behaviors.
197	/// It is useful if you want to test stuff simply.
198	pub fn loadModuleSync(loader: *Loader, opts: EnqueueOptions) !CompletionResult {
199	loader.run_lock.assertUnlocked();
200	const id = switch (try loader.enqueue(opts)) {
201	.available => |available| return .{ .available = available },
202	.failed => |failed| return .{ .failed = failed },
203	.fetching => |id| id,
204	};
205	loader.run(.until_empty);
206	return switch (loader.getModule(id)) {
207	.dead => unreachable,
208	.fetching => unreachable, // .until_empty returned before it was empty
209	.available => |available| return .{ .available = available },
210	.failed => |failed| return .{ .failed = failed },
211	};
212	}
213
214	pub const EnqueueResult = union(enum) {
215	available: Module.Data,
216	failed: Failure,
217	fetching: Module.Index,
218	};
219
220	pub const CompletionResult = union(enum) {
221	available: Module.Data,
222	failed: Failure,
223
224	/// This function is used for quick tests, where
225	/// it is okay to lose potentially relevant error data.
226	pub fn lossyUnwrap(result: CompletionResult) !Module.Loaded {
227	return switch (result) {
228	.available => |data| data,
229	.failed => |fail| fail.err,
230	};
231	}
232	};
233
234	pub const Failure = struct {
235	err: Err,
236	/// Which phase of the fetch had a failure.
237	phase: Phase,
238	/// For some error codes, additional information is helpful.
239	detail: union {
240	none: void,
241	http_status: std.http.Status,
242	} = .{ .none = {} },
243
244	pub const Err = Allocator.Error ||
245	std.http.Client.RequestError ||
246	std.http.Client.Request.SendError ||
247	std.http.Client.Request.FinishError ||
248	std.http.Client.Request.WaitError ||
249	std.http.Client.Request.ReadError ||
250	std.fs.File.OpenError ||
251	std.fs.File.ReadError ||
252	std.posix.PollError ||
253	error{
254	SchemeNotSupported,
255	/// The system `stat` has lied or otherwise file corruption
256	UnexpectedEndOfFile,
257	/// non-200 status codes are errors
258	HttpStatus,
259	};
260
261	pub const Phase = enum {
262	prepare,
263	http_open,
264	http_send,
265	http_finish,
266	http_wait,
267	socket_poll,
268	socket_read,
269	/// an error interacting with cache is a user configuration issue
270	cache_open,
271	cache_read,
272	fs_open,
273	fs_stat,
274	fs_read,
275	};
276
277	pub fn format(fail: Failure, comptime empty: []const u8, _: std.fmt.FormatOptions, writer: std.io.AnyWriter) !void {
278	comptime assert(empty.len == 0);
279
280	switch (fail.err) {
281	error.HttpStatus => {
282	try writer.print("HTTP {d}", .{@intFromEnum(fail.detail.http_status)});
283	if (fail.detail.http_status.phrase()) |phrase| {
284	try writer.print(": {s}", .{phrase});
285	}
286	},
287	inline else => |err| try writer.writeAll(switch (err) {
288	error.HttpStatus => unreachable, // handled
289
290	error.OutOfMemory => "Ran out of memory",
291	error.Unexpected => "Unexpected OS error",
292	error.InputOutput => "I/O Error",
293	error.IsDir => "Unexpected Directory",
294
295	else => comptime camel_case: {
296	// TODO: write a more robust version of this
297	const name = @errorName(err);
298	var buf: [name.len * 2]u8 = undefined;
299	var i = 1;
300	assert(std.ascii.isUpper(name[0]));
301	buf[0] = std.ascii.toLower(name[0]);
302	for (name[1..]) |char| {
303	assert(char < 0x80);
304	if (std.ascii.isUpper(char)) {
305	buf[i] = ' ';
306	buf[i + 1] = std.ascii.toLower(char);
307	i += 2;
308	} else {
309	buf[i] = char;
310	i += 1;
311	}
312	}
313
314	for (.{ "uri", "tls ", "http ", "fd " }) |acronym| {
315	if (std.mem.indexOf(u8, buf[0..i], acronym)) |j| {
316	for (j..j + acronym.len - 1) |k| {
317	buf[k] = std.ascii.toUpper(buf[k]);
318	}
319	}
320	}
321
322	buf[0] = std.ascii.toUpper(name[0]);
323
324	const final = buf[0..i].*;
325	break :camel_case &final;
326	},
327	}),
328	}
329
330	try writer.writeAll(switch (fail.phase) {
331	.prepare,
332	.http_open,
333	.http_send,
334	.http_finish,
335	.http_wait,
336	.socket_poll,
337	.socket_read,
338	=> " while fetching module",
339	.cache_open,
340	.cache_read,
341	=> " while reading from module cache",
342	.fs_open,
343	.fs_stat,
344	.fs_read,
345	=> " while reading file",
346	});
347	}
348	};
349
350	pub const Module = union(enum) {
351	/// The module is not yet ready.
352	fetching,
353	/// The module data is valid.
354	available: Data,
355	/// An error occurred.
356	failed: Failure,
357	/// The module was unloaded.
358	/// Using this module is a use-after-free error.
359	dead,
360
361	pub const Data = struct {
362	contents: [:0]const u8,
363	path_digest: BinDigest,
364	stat: Stat,
365	};
366
367	pub const Meta = struct {
368	path_digest: BinDigest,
369	stat: Stat,
370	};
371
372	pub const Index = enum(u32) {
373	_,
374	};
375
376	pub const Pointer = struct {
377	hash_table_lock: *Thread.RwLock,
378	module: *Module,
379
380	pub fn unlock(ptr: Pointer) void {
381	ptr.hash_table_lock.unlockShared();
382	}
383	};
384
385	fn deinit(module: Module, alloc: Allocator) void {
386	switch (module) {
387	.available => |data| {
388	alloc.free(data.contents);
389	},
390	else => {},
391	}
392	}
393	};
394
395	/// Holds a reading lock to the hash table entries.
396	/// The returned `*Module` may be mutated.
397	pub fn getModulePtr(loader: *Loader, index: Module.Index) Module.Pointer {
398	loader.entries_lock.lockShared();
399	return .{
400	.hash_table_lock = &loader.entries_lock,
401	.module = &loader.entries.values()[@intFromEnum(index)],
402	};
403	}
404
405	pub fn getModule(loader: *Loader, index: Module.Index) Module {
406	loader.entries_lock.lockShared();
407	defer loader.entries_lock.unlockShared();
408	return loader.entries.values()[@intFromEnum(index)];
409	}
410
411	pub fn unloadModule(loader: *Loader, index: Module.Index) void {
412	const locked_module = loader.getModulePtr(index);
413	defer locked_module.unlock();
414	locked_module.module.deinit(loader.allocator());
415	locked_module.module = .dead;
416	}
417
418	const OngoingLoad = struct {
419	/// Initialized to `.unused` at startup
420	state: State,
421	/// Status of the last poll
422	poll_state: PollState,
423	/// The resulting module index
424	module: Module.Index,
425	/// Progress node associated with this fetch
426	progress: std.Progress.Node,
427	/// Stored here to reduce contention
428	meta: Module.Meta,
429	/// The in-progress HTTP request.
430	request: std.http.Client.Request,
431
432	/// This field is used for polling of all ongoing downloads, however the
433	/// indices of these do not correspond to the actual file descriptors used
434	/// for the sockets (stored in request.connection.stream.handle), but
435	/// it is instead known that `loader.ongoing.items(.pollfd)` will be an
436	/// array with the maximum possible of ongoing downloads.
437	pollfd: std.posix.pollfd,
438	pollfd_n: Index,
439
440	/// Static bytes for storing request header information
441	server_header_buffer: ServerHeaderBuffer,
442	/// Initialized to `.{}` at startup
443	buffer: std.ArrayListUnmanaged(u8),
444
445	const ServerHeaderBuffer = [8 * 1024]u8;
446	const PollState = enum { ready, wait };
447
448	/// An index into `loader.ongoing`
449	/// Can be "freed" with `.release(loader)`
450	const Index = enum(u16) {
451	_,
452
453	pub fn release(i: Index, loader: *Loader) void {
454	const states = loader.ongoing.items(.state);
455	const index = @intFromEnum(i);
456
457	states[index] = .unused;
458
459	const node = &loader.ongoing.items(.progress)[index];
460	node.end();
461	node.* = .{ .index = .none };
462
463	// update `last_ongoing` to the latest node
464	if (index + 1 == loader.last_ongoing) {
465	loader.last_ongoing -= 1;
466	while (loader.last_ongoing > 0) {
467	loader.last_ongoing -= 1;
468	if (states[loader.last_ongoing] != .unused) {
469	return;
470	}
471	}
472
473	// Complete the loader-wide progress node.
474	loader.progress.end();
475	loader.progress = .{ .index = .none };
476	}
477	}
478
479	pub fn state(i: Index, loader: *Loader) *State {
480	return &loader.ongoing.items(.state)[@intFromEnum(i)];
481	}
482
483	pub fn module(i: Index, loader: *Loader) *Module.Index {
484	return &loader.ongoing.items(.module)[@intFromEnum(i)];
485	}
486
487	pub fn request(i: Index, loader: *Loader) *std.http.Client.Request {
488	return &loader.ongoing.items(.request)[@intFromEnum(i)];
489	}
490
491	pub fn serverHeaderBuffer(i: Index, loader: *Loader) *ServerHeaderBuffer {
492	return &loader.ongoing.items(.server_header_buffer)[@intFromEnum(i)];
493	}
494
495	pub fn progress(i: Index, loader: *Loader) *std.Progress.Node {
496	return &loader.ongoing.items(.progress)[@intFromEnum(i)];
497	}
498
499	pub fn buffer(i: Index, loader: *Loader) *std.ArrayListUnmanaged(u8) {
500	return &loader.ongoing.items(.buffer)[@intFromEnum(i)];
501	}
502
503	pub fn pollState(i: Index, loader: *Loader) *PollState {
504	return &loader.ongoing.items(.poll_state)[@intFromEnum(i)];
505	}
506
507	pub fn meta(i: Index, loader: *Loader) *Module.Meta {
508	return &loader.ongoing.items(.meta)[@intFromEnum(i)];
509	}
510	};
511
512	const List = std.MultiArrayList(OngoingLoad);
513	const State = enum {
514	/// This slot is not being used
515	unused,
516	wait_for_headers,
517	collecting_body,
518	};
519	};
520
521	const Task = struct {
522	uri_text: []const u8,
523	module: Module.Index,
524
525	const Node = std.DoublyLinkedList(Task).Node;
526
527	/// This first-in-first-out queue is threadsafe in that any thread may call
528	/// `add`, but only the Loader thread may use `takeAll` and `giveBack`.
529	const Queue = struct {
530	head: atomic.Value(?*Node) = .{ .raw = null },
531	tail: atomic.Value(?*Node) = .{ .raw = null },
532	event: Thread.ResetEvent = .{},
533
534	pub fn add(queue: *Queue, node: *Node) void {
535	assert(node.next == null);
536	assert(node.prev == null);
537
538	if (queue.tail.swap(node, .seq_cst)) |last| {
539	last.next = node;
540	node.prev = last;
541	} else {
542	queue.head.store(node, .seq_cst);
543	}
544	queue.event.set();
545	}
546
547	fn takeAll(queue: *Queue) Take {
548	const head = queue.head.swap(null, .monotonic) orelse
549	return .{ .head = null, .tail = undefined };
550	const tail = queue.tail.swap(null, .seq_cst) orelse
551	unreachable; // to get non-null from tail, head must also be null
552	return .{ .head = head, .tail = tail };
553	}
554
555	const Take = struct {
556	head: ?*Node,
557	tail: *Node,
558	};
559
560	fn giveBack(queue: *Queue, take: Take) void {
561	if (take.head == null) return;
562
563	if (queue.tail.cmpxchgStrong(null, take.tail, .seq_cst, .seq_cst) != null) {
564	const old_head = queue.head.load(.seq_cst) orelse head: {
565	// Wait for head.store() to finish
566	queue.event.wait();
567	if (queue.head.load(.seq_cst)) |head|
568	break :head head;
569	// In the unliekly event that two `add`s happen during
570	// `giveBack`, and the first one calls swap, but the second
571	// add hits `queue.event.set`, we have to wait for the first
572	// add to finish assigning `queue.head`.
573	while (true) {
574	std.atomic.spinLoopHint();
575	break :head queue.head.load(.seq_cst) orelse continue;
576	}
577	};
578
579	assert(take.tail.next == null);
580	take.tail.next = old_head;
581	assert(old_head.prev == null);
582	old_head.prev = take.tail;
583	}
584
585	queue.head.store(take.head, .monotonic);
586	}
587	};
588	};
589
590	pub const CompletionCallback = fn (
591	ctx: ?*anyopaque,
592	loader: *Loader,
593	module: Module.Index,
594	result: CompletionResult,
595	) void;
596
597	pub fn run(
598	loader: *Loader,
599	comptime lifetime: enum { until_empty, no_block, forever },
600	) if (lifetime == .forever) noreturn else void {
601	loader.run_lock.lock();
602	defer loader.run_lock.unlock();
603
604	while (true) {
605	loader.queue.event.reset();
606	var enqueued = loader.queue.takeAll();
607	while (enqueued.head) |node| : ({
608	enqueued.head = node.next;
609	loader.allocator().destroy(node);
610	}) {
611	const slot = loader.reserveDownloadSlot() orelse {
612	loader.queue.giveBack(enqueued);
613	break;
614	};
615
616	var phase: Failure.Phase = .prepare;
617	const data = loader.startRequest(slot, node.data.uri_text, &phase) catch |err| {
618	const fail: Failure = .{
619	.err = err,
620	.phase = phase,
621	};
622
623	{
624	const module_lock = loader.getModulePtr(node.data.module);
625	defer module_lock.unlock();
626	module_lock.module.* = .{ .failed = fail };
627	}
628
629	if (loader.completion_cb) |uncast_cb| {
630	const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb));
631	cb(loader.completion_ctx, loader, node.data.module, .{ .failed = fail });
632	}
633
634	slot.release(loader);
635
636	continue;
637	};
638
639	if (data) |result| {
640	// Save the module contents
641	{
642	const locked_module = loader.getModulePtr(node.data.module);
643	defer locked_module.unlock();
644	locked_module.module.* = .{ .available = result };
645	}
646
647	if (loader.completion_cb) |uncast_cb| {
648	const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb));
649	cb(loader.completion_ctx, loader, node.data.module, .{ .available = result });
650	}
651
652	slot.release(loader);
653	} else {
654	slot.module(loader).* = node.data.module;
655	}
656	}
657
658	if (loader.last_ongoing == 0) {
659	switch (lifetime) {
660	.forever => {
661	loader.queue.event.wait();
662	continue;
663	},
664	.until_empty,
665	.no_block,
666	=> return,
667	}
668	}
669
670	const pollfd = loader.ongoing.items(.pollfd);
671	const pollfd_n = loader.ongoing.items(.pollfd_n);
672	var pollfd_count: u16 = 0;
673
674	for (
675	0..,
676	loader.ongoing.items(.state)[0..loader.last_ongoing],
677	) |i, *state| {
678	if (state.* == .unused) continue;
679
680	const slot: OngoingLoad.Index = @enumFromInt(i);
681	var potential_failure: Failure = .{ .err = undefined, .phase = .prepare };
682	const tick_result = loader.tickDownload(slot, state, &potential_failure) catch |err| {
683	potential_failure.err = err;
684
685	// Save the module contents
686	{
687	const locked_module = loader.getModulePtr(slot.module(loader).*);
688	defer locked_module.unlock();
689	locked_module.module.* = .{ .failed = potential_failure };
690	}
691
692	if (loader.completion_cb) |uncast_cb| {
693	const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb));
694	cb(loader.completion_ctx, loader, slot.module(loader).*, .{ .failed = potential_failure });
695	}
696
697	slot.release(loader);
698	continue;
699	};
700	switch (tick_result) {
701	.pending => |fd| {
702	pollfd[pollfd_count] = .{
703	.fd = fd,
704	.events = std.posix.POLL.IN | std.posix.POLL.HUP,
705	.revents = 0,
706	};
707	pollfd_n[pollfd_count] = slot;
708	pollfd_count += 1;
709	},
710	.result => |data| {
711	// Save the module contents
712	{
713	const locked_module = loader.getModulePtr(slot.module(loader).*);
714	defer locked_module.unlock();
715	locked_module.module.* = .{ .available = data };
716	}
717
718	if (loader.completion_cb) |uncast_cb| {
719	const cb: *const CompletionCallback = @alignCast(@ptrCast(uncast_cb));
720	cb(loader.completion_ctx, loader, slot.module(loader).*, .{ .available = data });
721	}
722
723	slot.release(loader);
724	},
725	}
726	}
727
728	if (pollfd_count > 0) {
729	_ = std.posix.poll(pollfd[0..pollfd_count], 500) catch {};
730	for (pollfd_n[0..pollfd_count], pollfd[0..pollfd_count]) |slot, poll| {
731	if (poll.revents != 0)
732	slot.pollState(loader).* = .ready;
733	}
734	}
735
736	if (loader.last_ongoing == 0) {
737	switch (lifetime) {
738	.forever => {
739	loader.queue.event.wait();
740	continue;
741	},
742	.until_empty,
743	.no_block,
744	=> return,
745	}
746	}
747	}
748	}
749
750	fn reserveDownloadSlot(loader: *Loader) ?OngoingLoad.Index {
751	const states = loader.ongoing.items(.state);
752	const index = std.mem.indexOfScalar(OngoingLoad.State, states, .unused) orelse
753	return null;
754	states[index] = .wait_for_headers;
755	loader.last_ongoing = @max(@as(u16, @intCast(index + 1)), loader.last_ongoing);
756	loader.ongoing.items(.progress)[index] = .{ .index = .none };
757	return @enumFromInt(index);
758	}
759
760	fn startProgress(loader: *Loader, name: []const u8, estimated_total_items: usize) std.Progress.Node {
761	if (loader.root_progress.index == .none) {
762	return .{ .index = .none };
763	}
764	if (loader.progress.index == .none) {
765	loader.progress = loader.root_progress.start("Module Loader", 0);
766	}
767	return loader.progress.start(name, estimated_total_items);
768	}
769
770	// returns `null` if a request was initialized and needs to be read from
771	// returns `Module.Data` if a cache entry was retrieved.
772	fn startRequest(
773	loader: *Loader,
774	slot: OngoingLoad.Index,
775	uri_text: []const u8,
776	phase_ptr: *Failure.Phase,
777	) !?Module.Data {
778	const uri = Uri.parse(uri_text) catch
779	unreachable; // pre-validate these in Sema
780
781	const path_digest = hash: {
782	var path_hash: std.Build.Cache.HashHelper = .{};
783	path_hash.addBytes("0.1.6");
784	path_hash.addBytes(uri_text);
785	var bin: BinDigest = undefined;
786	path_hash.hasher.final(&bin);
787	break :hash bin;
788	};
789
790	const scheme = Scheme.fromSlice(uri.scheme) orelse
791	return error.SchemeNotSupported;
792
793	return switch (scheme) {
794	inline else => |s| switch (s) {
795	.file => startRequestFile,
796	.http, .https => startRequestHttp,
797	}(loader, slot, uri_text, phase_ptr, uri, path_digest),
798	};
799	}
800
801	fn startRequestFile(
802	loader: *Loader,
803	_: OngoingLoad.Index,
804	_: []const u8,
805	phase_ptr: *Failure.Phase,
806	uri: Uri,
807	path_digest: BinDigest,
808	) !?Module.Data {
809	var bytes: [std.fs.max_path_bytes]u8 = undefined;
810	var fba = std.heap.FixedBufferAllocator.init(&bytes);
811	const path = uri.path.toRawMaybeAlloc(fba.allocator()) catch |err| switch (err) {
812	error.OutOfMemory => return error.NameTooLong,
813	};
814
815	phase_ptr.* = .fs_open;
816	const f = try std.fs.openFileAbsolute(path, .{});
817	defer f.close();
818
819	phase_ptr.* = .fs_stat;
820	const stat = try f.stat();
821
822	if (stat.size > std.math.maxInt(u32))
823	return error.FileTooBig;
824
825	phase_ptr.* = .fs_read;
826	const contents = try loader.allocator().allocSentinel(u8, @as(usize, @intCast(stat.size)), 0);
827	errdefer loader.allocator().free(contents);
828	const amt = try f.readAll(contents);
829	if (amt != stat.size)
830	return error.UnexpectedEndOfFile;
831
832	return .{
833	.contents = contents,
834	.path_digest = path_digest,
835	.stat = .{
836	.size = stat.size,
837	.inode = stat.inode,
838	.mtime = stat.mtime,
839	},
840	};
841	}
842
843	fn startRequestHttp(
844	loader: *Loader,
845	slot: OngoingLoad.Index,
846	uri_text: []const u8,
847	phase_ptr: *Failure.Phase,
848	uri: Uri,
849	path_digest: BinDigest,
850	) !?Module.Data {
851	if (loader.cache) |cache| try_cache: {
852	const hex_digest = std.fmt.bytesToHex(path_digest, .lower);
853
854	phase_ptr.* = .cache_open;
855	const f = cache.openFile(&hex_digest, .{}) catch |err| switch (err) {
856	else => break :try_cache,
857	error.NetworkNotFound,
858	error.AccessDenied,
859	error.AntivirusInterference,
860	error.Unexpected,
861	error.ProcessFdQuotaExceeded,
862	error.SystemFdQuotaExceeded,
863	=> return err,
864	};
865	defer f.close();
866
867	const stat = try f.stat();
868	const payload_size = stat.size - @sizeOf(u32) - @sizeOf(i128);
869	if (payload_size > std.math.maxInt(u32))
870	return error.FileTooBig;
871
872	phase_ptr.* = .cache_read;
873
874	const r = f.reader();
875
876	const version = r.readInt(u32, .little) catch break :try_cache;
877	if (version != 1) break :try_cache;
878
879	const mtime = r.readInt(i128, .little) catch break :try_cache;
880
881	const contents = try loader.allocator().allocSentinel(u8, @as(usize, @intCast(stat.size)), 0);
882	errdefer loader.allocator().free(contents);
883
884	{
885	const amt = try f.readAll(contents);
886	if (amt != payload_size)
887	return error.UnexpectedEndOfFile;
888	}
889
890	return .{
891	.contents = contents,
892	.path_digest = path_digest,
893	.stat = .{
894	.size = payload_size,
895	.inode = 0,
896	.mtime = mtime,
897	},
898	};
899	}
900
901	phase_ptr.* = .prepare;
902	try slot.buffer(loader).ensureTotalCapacity(loader.allocator(), 4096);
903
904	slot.progress(loader).* = loader.startProgress(uri_text, 0);
905	errdefer slot.progress(loader).end();
906
907	const req = slot.request(loader);
908	phase_ptr.* = .http_open;
909	req.* = try loader.http.open(.GET, uri, .{
910	.server_header_buffer = slot.serverHeaderBuffer(loader),
911	.keep_alive = true,
912	});
913	errdefer req.deinit();
914
915	phase_ptr.* = .http_send;
916	try req.send();
917
918	phase_ptr.* = .http_finish;
919	try req.finish();
920
921	slot.meta(loader).* = .{
922	.path_digest = path_digest,
923	.stat = undefined,
924	};
925
926	return null;
927	}
928
929	const TickResult = union(enum) {
930	result: Module.Data,
931	/// There is more work to wait on, call again later
932	pending: std.posix.fd_t,
933	};
934
935	fn tickDownload(
936	loader: *Loader,
937	slot: OngoingLoad.Index,
938	state: *OngoingLoad.State,
939	failure_ptr: *Failure,
940	) !TickResult {
941	const req = slot.request(loader);
942	switch (state.*) {
943	.unused => unreachable,
944
945	.wait_for_headers => {
946	failure_ptr.phase = .http_wait;
947	try req.wait();
948
949	if (req.response.status.class() != .success) {
950	failure_ptr.detail = .{ .http_status = req.response.status };
951	return error.HttpStatus;
952	}
953
954	state.* = .collecting_body;
955	if (req.response.content_length) |len| {
956	slot.progress(loader).setEstimatedTotalItems(len);
957	// TODO: pre-allocate
958	failure_ptr.phase = .prepare;
959	}
960
961	var it = req.response.iterateHeaders();
962	var mtime = std.time.nanoTimestamp();
963	while (it.next()) |header| {
964	std.debug.print("header {s} = {s}\n", .{ header.name, header.value });
965
966	// TODO: get modified time, get cache control rules.
967	_ = &mtime;
968	}
969	slot.meta(loader).stat.mtime = mtime;
970	},
971
972	else => {},
973	}
974
975	const buffer = slot.buffer(loader);
976	const original_len = buffer.items.len;
977
978	var start_index: usize = original_len;
979	buffer.expandToCapacity();
980	const dest_slice = buffer.items[start_index..];
981
982	const handle = req.connection.?.stream.handle;
983	const poll_state = slot.pollState(loader);
984	if (poll_state.* == .wait) {
985	failure_ptr.phase = .socket_poll;
986	var polls = [_]std.posix.pollfd{.{
987	.fd = handle,
988	.events = std.posix.POLL.IN | std.posix.POLL.HUP,
989	.revents = 0,
990	}};
991	const polled = try std.posix.poll(&polls, 0);
992	if (polled == 0) return .{ .pending = handle };
993	poll_state.* = .ready;
994	}
995
996	failure_ptr.phase = .socket_read;
997
998	// TODO: This is going to block until the dest is full, right?
999	const bytes_read = try req.reader().readAll(dest_slice);
1000	start_index += bytes_read;
1001
1002	slot.progress(loader).setCompletedItems(start_index);
1003
1004	if (bytes_read != 0 and bytes_read == dest_slice.len) {
1005	// This will trigger ArrayList to expand superlinearly at whatever its growth rate is.
1006	try buffer.ensureTotalCapacity(loader.allocator(), start_index + 1);
1007	poll_state.* = .ready;
1008	return .{ .pending = handle };
1009	}
1010
1011	req.deinit();
1012
1013	const meta = slot.meta(loader);
1014
1015	const data: Module.Data = .{
1016	.contents = try buffer.toOwnedSliceSentinel(loader.allocator(), 0),
1017	.path_digest = meta.path_digest,
1018	.stat = .{
1019	.inode = 0,
1020	.size = buffer.items.len,
1021	.mtime = meta.stat.mtime,
1022	},
1023	};
1024
1025	if (loader.cache) |cache|
1026	writeToCache(cache, data) catch {
1027	// TODO: consider a warning for this
1028	};
1029
1030	return .{ .result = data };
1031	}
1032
1033	fn writeToCache(cache: std.fs.Dir, data: Module.Data) !void {
1034	var rng: [16]u8 = undefined;
1035	std.crypto.random.bytes(&rng);
1036
1037	// TODO: i do not trust the atomicy of this on windows
1038	// i think renameat will not use directory handles, when that is totally possible
1039	// this "temp name handle -> rename to final name" should be abstracted later
1040
1041	var temp_name_buf: [32 + ".tmp\x00".len]u8 = undefined;
1042	const temp_name = std.fmt.bufPrintZ(
1043	&temp_name_buf,
1044	"{s}.tmp",
1045	.{&std.fmt.bytesToHex(rng, .lower)},
1046	) catch unreachable; // enough space is allocated
1047
1048	var final_name_buf: [std.Build.Cache.hex_digest_len + "\x00".len]u8 = undefined;
1049	const final_name = std.fmt.bufPrintZ(
1050	&final_name_buf,
1051	"{s}",
1052	.{&std.fmt.bytesToHex(data.path_digest, .lower)},
1053	) catch unreachable; // enough space is allocated
1054
1055	{
1056	const file = try cache.createFileZ(temp_name, .{});
1057
1058	errdefer cache.deleteFile(temp_name) catch {}; // unhandlable
1059
1060	defer file.close();
1061
1062	var bufw = std.io.bufferedWriter(file.writer());
1063	const writer = bufw.writer();
1064
1065	try writer.writeInt(u32, 1, .little);
1066	try writer.writeInt(i128, data.stat.mtime, .little);
1067	try writer.writeAll(data.contents);
1068
1069	try bufw.flush();
1070	}
1071
1072	errdefer cache.deleteFile(temp_name) catch {}; // unhandlable
1073
1074	try cache.renameZ(temp_name, final_name);
1075	}
1076
1077	/// The allocator is stored on the http client struct
1078	/// (There is no cost to the extra dot access)
1079	pub inline fn allocator(loader: *const Loader) Allocator {
1080	return loader.http.allocator;
1081	}
1082
1083	const Loader = @This();
1084
1085	const std = @import("std");
1086	const assert = std.debug.assert;
1087	const atomic = std.atomic;
1088	const Allocator = std.mem.Allocator;
1089	const Uri = std.Uri;
1090	const Thread = std.Thread;
1091
1092	const BinDigest = std.Build.Cache.BinDigest;
1093	const HexDigest = std.Build.Cache.HexDigest;
1094	const Stat = std.Build.Cache.File.Stat;