Files
browser/src/main.zig
T
Karl Seguin 87320a506d chore: Move ownership of the Inbox from HttpClient to WebSocker Driver
Currently, the HttpClient owns the inbox and its borrowed by the Link. This is
a bit backwards, but it also means that we can't eagerly create a Link: the
Link needs the inbox, so it needs the HttpClient, which is created by the
Browser (which creates an Isolate).

Remember, the Inbox is one of the few things shared between the main thread
and the worker, so either end can own it and the other can borrow it.

This switches the ownership so that the HttpClient now borrows the Inbox from
the Server's side of the Link (the WebSocket).

The main goal of this change is to prepare for more advanced HTTP WebDriver
flows. The more we can create _without_ a Browser, the fewer edge cases we have
to deal with (Browser because it's expensive and has to be created on the
Worker thread due to how V8::Isolate works).
2026-09-03 16:51:12 +08:00

415 lines
16 KiB
Zig

// Copyright (C) 2023-2026 Lightpanda (Selecy SAS)
//
// Francis Bouvier <francis@lightpanda.io>
// Pierre Tachoire <pierre@lightpanda.io>
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.
const std = @import("std");
const lp = @import("lightpanda");
const Allocator = std.mem.Allocator;
const log = lp.log;
const App = lp.App;
const Config = lp.Config;
const SigHandler = @import("Sighandler.zig");
pub const panic = lp.crash_handler.panic;
pub const std_options: std.Options = .{
.signal_stack_size = null,
};
pub fn main(init: std.process.Init) !void {
// allocator
// - in Debug mode we use the General Purpose Allocator to detect memory leaks
// - in Release mode we use the c allocator
var gpa_instance: std.heap.DebugAllocator(.{ .stack_trace_frames = 10 }) = .init;
const gpa = if (lp.IS_DEBUG) gpa_instance.allocator() else std.heap.c_allocator;
defer if (lp.IS_DEBUG) {
if (gpa_instance.detectLeaks() != 0) std.process.exit(1);
};
// arena for main-specific allocations
var main_arena_instance = std.heap.ArenaAllocator.init(gpa);
const main_arena = main_arena_instance.allocator();
defer main_arena_instance.deinit();
run(gpa, main_arena, init.minimal.args) catch |err| {
if (err == error.UserCancelled) std.process.exit(130);
// error.AgentFailed: the agent thread reported the failure in-context.
// error.PageFailed: fetch already logged every failing url.
// lp.Agent.UserError: a user-facing message was already printed.
if (err == error.AgentFailed or err == error.PageFailed or lp.Agent.isUserError(err)) std.process.exit(1);
// curl's code for --fail on an HTTP error, also already reported per url.
if (err == error.HttpError) std.process.exit(22);
log.fatal(.app, "exit", .{ .err = err });
std.process.exit(1);
};
}
fn run(allocator: Allocator, main_arena: Allocator, proc_args: std.process.Args) !void {
lp.core_dump.disableIfRequested();
const args = try Config.parseArgs(main_arena, proc_args);
defer args.deinit(main_arena);
switch (args.mode) {
.help => |tag| return args.printUsageAndExit(main_arena, tag, true),
.version => |opts| {
if (opts.check) {
try lp.checkVersion(allocator, &args);
} else {
var stdout = std.Io.File.stdout().writerStreaming(lp.io, &.{});
try stdout.interface.print("{s}\n", .{lp.build_config.version});
}
return std.process.cleanExit(lp.io);
},
.agent => |opts| if (opts.list_models) {
try lp.Agent.listModels(allocator, opts);
return std.process.cleanExit(lp.io);
},
else => {},
}
// must be installed before any other threads
const sighandler = try main_arena.create(SigHandler);
sighandler.* = .{ .arena = main_arena };
try sighandler.install();
// _app is global to handle graceful shutdown.
var app = try App.init(allocator, &args);
defer app.deinit();
app.telemetry.record(.{ .run = {} });
logConfigTips(app.config);
defer if (app.config.dumpMetricsOnExit()) {
var writer = std.Io.File.stdout().writerStreaming(lp.io, &.{});
lp.metrics.write(&writer.interface);
};
switch (args.mode) {
.serve => |opts| {
log.debug(.app, "startup", .{ .mode = "serve", .snapshot = app.snapshot.fromEmbedded() });
const address = std.Io.net.IpAddress.parse(opts.host, opts.port) catch |err| {
log.fatal(.app, "invalid server address", .{ .err = err, .host = opts.host, .port = opts.port });
return args.printUsageAndExit(main_arena, .serve, false);
};
var server = lp.Server.init(app, address) catch |err| {
if (err == error.AddressInUse) {
log.fatal(.app, "address already in use", .{
.host = opts.host,
.port = opts.port,
.hint = "Another process is already listening on this address. " ++
"Stop the other process or use --port to choose a different port.",
});
} else {
log.fatal(.app, "server run error", .{ .err = err });
}
return err;
};
defer server.deinit();
try sighandler.on(lp.Server.shutdown, .{server});
server.run();
},
.fetch => |opts| {
const urls = opts.url.items;
if (urls.len == 0) {
log.fatal(.app, "missing URL", .{});
return error.MissingArgument;
}
// Plain (non-JSON) dump writes one document to stdout with no
// framing, so it can't disambiguate more than one page.
if (urls.len == 1) {
log.debug(.app, "startup", .{
.mode = "fetch",
.dump_mode = opts.dump,
.url = urls[0],
.snapshot = app.snapshot.fromEmbedded(),
});
} else {
if (opts.json == false) {
log.fatal(.app, "multiple URLs require --json", .{});
return error.InvalidArgument;
}
log.debug(.app, "startup", .{
.mode = "fetch",
.dump_mode = opts.dump,
.url_count = urls.len,
.snapshot = app.snapshot.fromEmbedded(),
});
}
if (opts.dump == null and opts.dump_selector != null) {
log.fatal(.app, "--dump-selector needs --dump", .{});
return error.InvalidArgument;
}
if (opts.dump == null and opts.dump_max_bytes != null) {
log.fatal(.app, "--dump-max-bytes needs --dump", .{});
return error.InvalidArgument;
}
if (opts.dump_max_bytes != null and opts.dump != .html and opts.dump != .markdown) {
log.fatal(.app, "--dump-max-bytes needs text", .{ .dump = opts.dump, .allowed = "html, markdown" });
return error.InvalidArgument;
}
var fetch_opts = lp.FetchOpts{
.wait_ms = opts.wait_ms,
.wait_until = opts.wait_until,
.wait_script = opts.wait_script,
.inject_script = opts.inject_script,
.wait_selector = opts.wait_selector,
.dump_mode = opts.dump,
.selector = opts.dump_selector,
.fail_on_http_error = opts.fail_on_http_error,
.dump = .{
.strip = opts.strip_mode,
.with_base = opts.with_base,
.with_frames = opts.with_frames,
.max_bytes = opts.dump_max_bytes,
},
.json = opts.json,
};
var writer = std.Io.File.stdout().writerStreaming(lp.io, &.{});
if (opts.dump != null or opts.json) {
fetch_opts.writer = &writer.interface;
}
// Browser owns a V8 isolate, which has thread affinity — it must
// be init/used/deinit on the same thread (fetchThread, below). So
// we can't treat Browser like the above serve path treats Server.
// We need Browser to be createdin fetchThread and to get a reference
// to it here.
var ft: FetchTerminator = .{};
try sighandler.on(FetchTerminator.terminate, .{&ft});
if (opts.terminate_ms) |ms| {
try sighandler.deadline(ms);
}
var fetch_err: ?anyerror = null;
var worker_thread = try std.Thread.spawn(.{}, fetchThread, .{ app, &ft, urls, fetch_opts, &fetch_err });
worker_thread.join();
if (fetch_err) |err| return err;
},
.mcp => |opts| {
log.info(.mcp, "starting server", .{});
// --port serves MCP over HTTP instead of stdio. It and --cdp-port
// each run their accept loop on this thread, so they can't combine.
if (opts.port) |port| {
if (opts.cdp_port != null) {
log.fatal(.mcp, "port conflicts with cdp-port", .{ .hint = "both need the main thread for their accept loop" });
return error.InvalidArgument;
}
const address = std.Io.net.IpAddress.parse(opts.host, port) catch |err| {
log.fatal(.mcp, "invalid address", .{ .err = err, .host = opts.host, .port = port });
return err;
};
const http_server = try lp.mcp.HttpServer.init(allocator, app);
defer http_server.deinit();
// A signal stops the accept loop, run() returns, deinit joins.
try sighandler.on(lp.mcp.HttpServer.stop, .{http_server});
http_server.run(address) catch |err| {
log.fatal(.mcp, "mcp http error", .{ .err = err });
return err;
};
return;
}
var cdp_server: ?*lp.Server = null;
if (opts.cdp_port) |port| {
const address = std.Io.net.IpAddress.parse("127.0.0.1", port) catch |err| {
log.fatal(.mcp, "invalid cdp address", .{ .err = err, .port = port });
return;
};
cdp_server = try lp.Server.init(app, address);
try sighandler.on(lp.Server.shutdown, .{cdp_server.?});
}
defer if (cdp_server) |s| s.deinit();
var mcp_err: ?anyerror = null;
{
var worker_thread = try std.Thread.spawn(.{}, mcpThread, .{ allocator, app, cdp_server, &mcp_err });
defer worker_thread.join();
// mcp talks over stdio on mcpThread. Only run the CDP accept/read
// loop when an optional CDP server was started; otherwise the main
// thread just waits for the worker.
if (cdp_server) |s| {
s.run();
}
}
if (mcp_err) |err| return err;
},
.agent => |opts| {
log.info(.app, "starting agent", .{});
// Ctrl-C cancels the current turn; signals never kill the
// process. `/quit` (or Ctrl-D on an empty prompt) exits.
sighandler.no_hard_exit = true;
var sig_bridge: lp.Agent.SigBridge = .{};
try sighandler.on(lp.Agent.SigBridge.onSignal, .{&sig_bridge});
var failed: bool = false;
var cancelled: bool = false;
{
var worker_thread = try std.Thread.spawn(.{}, agentThread, .{
allocator,
app,
opts,
&failed,
&cancelled,
&sig_bridge,
});
worker_thread.join();
}
if (cancelled) return error.UserCancelled;
if (failed) return error.AgentFailed;
},
else => unreachable,
}
}
fn agentThread(
allocator: std.mem.Allocator,
app: *App,
opts: Config.Agent,
failed: *bool,
cancelled: *bool,
sig_bridge: *lp.Agent.SigBridge,
) void {
var agent_instance = lp.Agent.init(allocator, app, opts) catch |err| {
if (err == error.UserCancelled) {
cancelled.* = true;
} else {
// UserError: message already printed inside Agent.init.
if (!lp.Agent.isUserError(err)) log.fatal(.app, "agent init error", .{ .err = err });
failed.* = true;
}
return;
};
sig_bridge.attach(agent_instance);
defer agent_instance.deinit();
defer sig_bridge.detach();
if (agent_instance.ai_client) |cli| {
app.telemetry.record(.{
.llm = app.telemetry.llm_init(@tagName(cli), agent_instance.model),
});
} else {
app.telemetry.record(.{
.llm = app.telemetry.llm_init("nollm", null),
});
}
if (!agent_instance.run()) {
failed.* = true;
}
}
const FetchTerminator = struct {
mutex: std.Io.Mutex = .init,
browser: ?*lp.Browser = null,
fn storeBrowser(self: *FetchTerminator, browser: *lp.Browser) void {
self.mutex.lockUncancelable(lp.io);
defer self.mutex.unlock(lp.io);
self.browser = browser;
}
fn releaseBrowser(self: *FetchTerminator) void {
self.mutex.lockUncancelable(lp.io);
defer self.mutex.unlock(lp.io);
const b = self.browser orelse return;
b.env.cancelTerminate();
self.browser = null;
}
fn terminate(self: *FetchTerminator) void {
self.mutex.lockUncancelable(lp.io);
defer self.mutex.unlock(lp.io);
const b = self.browser orelse return;
b.env.terminate();
self.browser = null;
}
};
fn fetchThread(app: *App, ft: *FetchTerminator, urls: []const [:0]const u8, fetch_opts: lp.FetchOpts, err_out: *?anyerror) void {
var browser: lp.Browser = undefined;
browser.init(app, .{}) catch |err| {
err_out.* = err;
log.fatal(.app, "browser init error", .{ .err = err });
return;
};
defer browser.deinit();
ft.storeBrowser(&browser);
// if this exits normally, we want to disarm the FetchTerminator so that
// any subsequent sighandlers don't try to shutdown an already (or in-the-
// process-of) shutting down browser/env
defer ft.releaseBrowser();
lp.fetch(app, &browser, urls, fetch_opts) catch |err| {
err_out.* = err;
// Both are already reported per url by fetch itself.
if (err == error.PageFailed or err == error.HttpError) return;
log.fatal(.app, "fetch error", .{ .err = err, .url_count = urls.len });
};
}
fn mcpThread(allocator: std.mem.Allocator, app: *App, cdp_server: ?*lp.Server, err_out: *?anyerror) void {
defer if (cdp_server) |s| {
s.shutdown();
};
var stdout = std.Io.File.stdout().writerStreaming(lp.io, &.{});
var mcp_server: *lp.mcp.Server = lp.mcp.Server.init(allocator, app, &stdout.interface) catch |err| {
err_out.* = err;
log.fatal(.mcp, "mcp init error", .{ .err = err });
return;
};
defer mcp_server.deinit();
var stdin_buf: [64 * 1024]u8 = undefined;
var stdin = std.Io.File.stdin().readerStreaming(lp.io, &stdin_buf);
lp.mcp.router.processRequests(mcp_server, &stdin.interface, std.Io.File.stdin()) catch |err| {
err_out.* = err;
log.fatal(.mcp, "mcp error", .{ .err = err });
};
}
fn logConfigTips(config: *const Config) void {
var count: usize = 0;
var tips: [2]log.KV = undefined;
if (config.obeyRobots() == false) {
tips[count] = .init("robots", "use '--obey-robots' to use a sites robots.txt");
count += 1;
}
if (count > 0) {
tips[count] = .init("meta", "use '--log-filter note' to silence this message");
log.logKVs(.note, .note, "config tips", tips[0 .. count + 1]);
}
}