forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
10.9 KiB
1//! Full boot sequence: config, store, git checks, gitconfig and hooks, routers, SSH task.
2//!
3//! Three servers run until one of them stops or a shutdown signal arrives:
4//! - public HTTP on `config.listen` (UI and smart-HTTP),
5//! - raw user-content origin on `config.user_content_listen` (raw.rs, explicit routes only),
6//! - internal hook API on the unix socket `data_dir/internal.sock`,
7//! - SSH on `config.ssh_listen` (forge_ssh, which has no shutdown hook, so it is aborted).
8
9use std::os::unix::fs::PermissionsExt;
10use std::path::PathBuf;
11use std::sync::Arc;
12use std::time::Duration;
13
14use forge_core::{Config, Store, secret};
15use forge_git::GitEnv;
16use forge_ssh::SshConfig;
17use tokio::net::{TcpListener, UnixListener};
18use tokio::signal::unix::{SignalKind, signal};
19use tokio::sync::Notify;
20use tokio::sync::watch;
21use topcoat::cookie::RouterBuilderCookieExt;
22use topcoat::router::tower::TowerLayer;
23use topcoat::router::{Compression, Router, RouterBuilderDiscoverExt, RouterService};
24
25use crate::internal;
26use crate::raw;
27use crate::ssh_backend::SshBackend;
28use crate::state::AppState;
29
30/// Minimum git version forge relies on (receive-pack and stateless-rpc behaviour).
31pub(crate) const MIN_GIT: (u32, u32, u32) = (2, 38, 0);
32
33/// Boots every component and serves until shutdown.
34pub async fn serve(config: Config) -> forge_core::Result<()> {
35 let listen = config.listen;
36 let user_content_listen = config.user_content_listen;
37 let ssh_listen = config.ssh_listen;
38
39 std::fs::create_dir_all(&config.data_dir)?;
40 // The data dir holds the instance key and the internal socket. Keep it private to this user.
41 std::fs::set_permissions(&config.data_dir, std::fs::Permissions::from_mode(0o700))?;
42 let data_dir = config.data_dir.canonicalize()?;
43
44 let key = Arc::new(secret::load_or_create_key(&data_dir.join("forge.key"))?);
45 let store = Store::open(&config.database).await?;
46 store.set_secret_key(*key);
47
48 let git = Arc::new(GitEnv {
49 git_bin: find_git(),
50 data_dir: data_dir.clone(),
51 });
52 forge_git::version::require_min_version(&git, MIN_GIT).await?;
53 let version = forge_git::version::git_version(&git).await?;
54 forge_git::setup::write_managed_gitconfig(&data_dir)?;
55 let forge_bin = std::env::current_exe()?.canonicalize()?;
56 forge_git::setup::write_hook_shims(&data_dir, &forge_bin)?;
57
58 // Hash the running binary once, before any runner can ask for it. Hashing reads the whole
59 // executable, which takes tens of seconds for a debug build, so it runs off the boot path and
60 // the listeners bind first. Any caller that needs the hash waits on the same `OnceLock`.
61 tokio::task::spawn_blocking(|| {
62 crate::self_update::running();
63 });
64
65 let internal_token: Arc<str> = Arc::from(random_hex());
66 let admin_token: Arc<str> = Arc::from(random_hex());
67 crate::admin_client::write_admin_token(&data_dir, &admin_token)?;
68 let internal_socket = data_dir.join("internal.sock");
69 // sun_path holds at most 108 bytes including the NUL terminator.
70 if internal_socket.as_os_str().len() >= 108 {
71 return Err(forge_core::Error::Config(format!(
72 "internal socket path is too long for a unix socket (max 107 bytes): {}; choose a shorter data_dir",
73 internal_socket.display()
74 )));
75 }
76 match std::fs::remove_file(&internal_socket) {
77 Ok(()) => {}
78 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
79 Err(e) => return Err(e.into()),
80 }
81 let unix = UnixListener::bind(&internal_socket)?;
82 std::fs::set_permissions(&internal_socket, std::fs::Permissions::from_mode(0o600))?;
83
84 let mailer: Arc<dyn crate::mail::Mailer> = match config.mail.transport {
85 forge_core::config::MailTransport::Smtp => {
86 Arc::new(crate::mail::SmtpMailer::new(store.clone()))
87 }
88 forge_core::config::MailTransport::Log => {
89 Arc::new(crate::mail::LogMailer::new(data_dir.join("mail-log")))
90 }
91 };
92 let state = AppState {
93 config: Arc::new(config),
94 store,
95 git,
96 key,
97 internal_token,
98 admin_token,
99 jobs: Arc::new(crate::backup::Jobs::default()),
100 internal_socket: internal_socket.clone(),
101 throttle: Arc::new(crate::throttle::Throttle::default()),
102 webhook_waker: Arc::new(Notify::new()),
103 mirror_waker: Arc::new(Notify::new()),
104 blobs: crate::blobs::BlobStore::open(&data_dir.join("blobs"))?,
105 replication: Arc::new(crate::replication::ReplicationQueue::default()),
106 ci_waker: Arc::new(Notify::new()),
107 mailer,
108 mail_waker: Arc::new(Notify::new()),
109 notify_hub: Arc::new(crate::notify_hub::NotifyHub::new()),
110 ci_live: Arc::new(crate::ci::CiLive::default()),
111 };
112 if let Err(e) = crate::signatures::rebuild_allowed_signers(&state).await {
113 tracing::error!(error = %e, "building the SSH allowed-signers file failed");
114 }
115 if let Err(e) = crate::signatures::rebuild_gpg_keyring(&state).await {
116 tracing::error!(error = %e, "rebuilding the GPG keyring failed");
117 }
118 crate::webhooks::worker::spawn(state.clone());
119 crate::mirrors::worker::spawn(state.clone());
120 crate::advisories::worker::spawn(state.clone());
121 crate::mail::worker::spawn(state.clone());
122 if state.config.backup.enabled() {
123 crate::backup::scheduler::spawn(state.clone());
124 }
125 if state.config.replication.enabled() {
126 crate::replication::worker::spawn(state.clone());
127 }
128 crate::ci::reaper::spawn(state.clone());
129
130 // Topcoat's builder-wide compression is off because it cannot filter by content type or path.
131 // `ResponseCompression` compresses text responses only and skips git, LFS, raw and archive paths.
132 let public: Router = Router::builder()
133 .discover()
134 .app_context(state.clone())
135 .cookies()
136 .layer(crate::pages::settings_profile::avatar_body_limit())
137 .layer(crate::pages::repo_ctx::RepoMovedLayer)
138 .layer(TowerLayer::new(crate::compression::ResponseCompression))
139 .compression(Compression::off())
140 .build();
141 let internal_router = internal::router(state.clone());
142 let raw_router = raw::router(state.clone());
143
144 let tcp = TcpListener::bind(listen).await?;
145 let raw_tcp = TcpListener::bind(user_content_listen).await.map_err(|e| {
146 forge_core::Error::Config(format!(
147 "cannot bind the user-content listener {user_content_listen}: {e}"
148 ))
149 })?;
150
151 tracing::info!(
152 listen = %listen,
153 user_content = %user_content_listen,
154 ssh = %ssh_listen,
155 data_dir = %data_dir.display(),
156 git = %format_version(version),
157 "forge starting"
158 );
159
160 let (stop_tx, stop_rx) = watch::channel(false);
161
162 // Open notification streams never end on their own, so they are closed with the rest.
163 let hub = Arc::clone(&state.notify_hub);
164 let hub_stop = stop_rx.clone();
165 tokio::spawn(async move {
166 wait_stop(hub_stop).await;
167 hub.close();
168 });
169
170 let ssh_config = SshConfig {
171 listen: ssh_listen,
172 host_key_path: data_dir.join("ssh_host_ed25519_key"),
173 auth_timeout: Duration::from_secs(30),
174 inactivity_timeout: Duration::from_secs(10 * 60),
175 };
176 let ssh_backend = Arc::new(SshBackend {
177 state: state.clone(),
178 });
179 let ssh_stop = stop_tx.clone();
180 let mut ssh_task = tokio::spawn(async move {
181 let result = forge_ssh::run(ssh_config, ssh_backend).await;
182 // An SSH failure brings the whole process down.
183 ssh_stop.send_replace(true);
184 result
185 });
186
187 let signal_stop = stop_tx.clone();
188 tokio::spawn(async move {
189 tokio::select! {
190 () = shutdown_signal() => tracing::info!("shutdown signal received"),
191 () = crate::self_update::restart_requested() => tracing::info!("self-update: restarting"),
192 }
193 signal_stop.send_replace(true);
194 });
195
196 let http_stop = stop_tx.clone();
197 let http_rx = stop_rx.clone();
198 let http = async move {
199 let result =
200 topcoat::serve_until(tcp, RouterService::new(public), wait_stop(http_rx)).await;
201 http_stop.send_replace(true);
202 result
203 };
204
205 let internal_stop = stop_tx.clone();
206 let internal_rx = stop_rx.clone();
207 let internal_server = async move {
208 let result = topcoat::serve_until(
209 unix,
210 RouterService::new(internal_router),
211 wait_stop(internal_rx),
212 )
213 .await;
214 internal_stop.send_replace(true);
215 result
216 };
217
218 let raw_stop = stop_tx.clone();
219 let raw_rx = stop_rx.clone();
220 let raw_server = async move {
221 let result =
222 topcoat::serve_until(raw_tcp, RouterService::new(raw_router), wait_stop(raw_rx)).await;
223 raw_stop.send_replace(true);
224 result
225 };
226
227 let (http_result, internal_result, raw_result) =
228 tokio::join!(http, internal_server, raw_server);
229
230 // forge_ssh::run has no shutdown hook. If it is still running it is cancelled here.
231 ssh_task.abort();
232 let ssh_result: std::io::Result<()> = match (&mut ssh_task).await {
233 Ok(result) => result.map_err(|e| std::io::Error::other(e.to_string())),
234 Err(e) if e.is_cancelled() => Ok(()),
235 Err(e) => Err(std::io::Error::other(e.to_string())),
236 };
237
238 let _ = std::fs::remove_file(&internal_socket);
239 tracing::info!("forge stopped");
240
241 http_result?;
242 internal_result?;
243 raw_result?;
244 ssh_result?;
245 Ok(())
246}
247
248/// Resolves when the shared stop flag is set.
249async fn wait_stop(mut rx: watch::Receiver<bool>) {
250 let _ = rx.wait_for(|stopped| *stopped).await;
251}
252
253/// Resolves on ctrl_c or SIGTERM.
254async fn shutdown_signal() {
255 let ctrl_c = async {
256 if let Err(e) = tokio::signal::ctrl_c().await {
257 tracing::warn!(error = %e, "ctrl_c handler unavailable");
258 std::future::pending::<()>().await;
259 }
260 };
261 let terminate = async {
262 match signal(SignalKind::terminate()) {
263 Ok(mut term) => {
264 term.recv().await;
265 }
266 Err(e) => {
267 tracing::warn!(error = %e, "SIGTERM handler unavailable");
268 std::future::pending::<()>().await;
269 }
270 }
271 };
272 tokio::select! {
273 () = ctrl_c => {}
274 () = terminate => {}
275 }
276}
277
278/// Finds the git binary on PATH, falling back to /usr/bin/git.
279pub(crate) fn find_git() -> PathBuf {
280 std::env::var_os("PATH")
281 .and_then(|paths| {
282 std::env::split_paths(&paths)
283 .map(|dir| dir.join("git"))
284 .find(|candidate| candidate.is_file())
285 })
286 .and_then(|candidate| candidate.canonicalize().ok())
287 .unwrap_or_else(|| PathBuf::from("/usr/bin/git"))
288}
289
290/// 32 random bytes as lowercase hex. Used as the per-boot internal API token.
291fn random_hex() -> String {
292 let mut buf = [0u8; 32];
293 rand::fill(&mut buf);
294 hex::encode(buf)
295}
296
297fn format_version((major, minor, patch): (u32, u32, u32)) -> String {
298 format!("{major}.{minor}.{patch}")
299}