alice.pds.demo.boltless.dev
This repository has no description
28 kB
704 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1mod allocator;
2
3#[global_allocator]
4static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
5
6#[allow(non_upper_case_globals)]
7#[unsafe(export_name = "_rjem_malloc_conf")]
8pub static malloc_conf: &[u8] =
9 b"background_thread:true,retain:false,dirty_decay_ms:0,muzzy_decay_ms:0\0";
10
11use std::collections::BTreeSet;
12use std::num::{NonZeroU32, NonZeroU64};
13use std::path::PathBuf;
14use std::sync::Arc;
15use std::time::Duration;
16
17use tokio_util::sync::CancellationToken;
18
19use anyhow::Context;
20use axum::Json;
21use axum::response::Html;
22use axum::routing::get;
23use base64::Engine;
24use knot_atproto::Atproto;
25use knot_config::HomepageSource;
26use knot_index::Index;
27use knot_runtime::{Clock, HttpTransport, OsEntropy, ReqwestHttp, SystemClock};
28use knot_secrets::{MasterKey, SealedStore};
29use knot_types::{ActorId, AuthorName, BranchName, CiLogsAddr, Email, KnotHostname, ObjectCount};
30use knot_xrpc::XrpcState;
31use tower_http::services::ServeFile;
32
33const MAINTENANCE_SHUTDOWN_DRAIN: Duration = Duration::from_secs(30);
34const EDGE_SHUTDOWN_DRAIN: Duration = Duration::from_secs(40);
35
36const DEFAULT_HOMEPAGE: &str = include_str!("homepage.html");
37
38struct IndexRepos(Arc<Index>);
39
40impl knot_maintenance::RepoSource for IndexRepos {
41 fn repos(&self) -> Vec<knot_types::RepoDid> {
42 self.0.hosted_repos()
43 }
44
45 fn ready_repos(&self) -> Option<Vec<knot_types::RepoDid>> {
46 match self.0.coverage().registry {
47 knot_index::Coverage::Ready => Some(self.0.hosted_repos()),
48 knot_index::Coverage::Warming => None,
49 }
50 }
51}
52
53struct AtprotoHandleResolver<H, C> {
54 atproto: Arc<Atproto<H, C>>,
55}
56
57impl<H: HttpTransport, C: Clock> knot_pack::HandleResolver for AtprotoHandleResolver<H, C> {
58 fn resolve(
59 &self,
60 handle: knot_types::Handle,
61 ) -> std::pin::Pin<
62 Box<dyn std::future::Future<Output = Option<knot_types::AccountDid>> + Send + '_>,
63 > {
64 Box::pin(async move { self.atproto.resolve_handle_to_did(&handle).await.ok() })
65 }
66}
67
68fn init_tracing() {
69 let filter = tracing_subscriber::EnvFilter::try_from_default_env()
70 .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));
71 tracing_subscriber::fmt()
72 .with_env_filter(filter)
73 .with_writer(std::io::stderr)
74 .init();
75}
76
77#[tokio::main]
78async fn main() -> anyhow::Result<()> {
79 #[cfg(target_os = "linux")]
80 rustix::process::set_dumpable_behavior(rustix::process::DumpableBehavior::NotDumpable)
81 .context("disable core dumps and ptrace attachment")?;
82
83 if std::env::args().nth(1).as_deref() == Some("config-template") {
84 print!("{}", knot_config::template());
85 return Ok(());
86 }
87
88 init_tracing();
89
90 tracing::info!("!");
91 tracing::info!("!");
92 tracing::info!("!");
93 tracing::info!("> If knot1 was so good then why isn't there a... ( ˶°ㅁ°)");
94 tracing::info!("...");
95 tracing::info!("Welcome to knot2!");
96 tracing::info!("This code was made with love.");
97 tracing::info!("Hachapuri is sho tasty, definitely worth a try. Better than pizza tbh.");
98 tracing::info!("!");
99 tracing::info!("!");
100 tracing::info!("!");
101
102 let config_path = std::env::args().nth(1).map(PathBuf::from);
103 let config = knot_config::load(config_path.as_deref()).context("load configuration")?;
104 config
105 .verify_environment()
106 .context("verify runtime environment")?;
107
108 let resources = knot_resource::init(knot_resource::Ceilings {
109 max_threads: match config.resources.max_threads {
110 0 => None,
111 n => Some(knot_resource::ThreadCount::new(n as usize)),
112 },
113 max_memory: match config.resources.max_memory_bytes {
114 0 => None,
115 n => Some(knot_resource::MemoryBudget::new(n)),
116 },
117 });
118 tracing::info!(
119 threads = resources.threads.get(),
120 memory_bytes = resources.memory.map(knot_resource::MemoryBudget::get),
121 memory_source = ?resources.memory_source,
122 memory_high_bytes = resources
123 .memory_high_bytes
124 .map(knot_resource::MemoryHighBytes::get),
125 "resource governor initialized"
126 );
127 allocator::install();
128
129 // I made the hostname stay aa string in the config,
130 // so that confique can layer env-over-file.
131 // Here's where it becomes a real type,
132 // and the DID + the service url stack on this.
133 let hostname = KnotHostname::new(config.server.hostname.clone())
134 .context("server.hostname isn't a valid knot hostname")?;
135 let knot_did = hostname.knot_did();
136 let object_format = config.object_format().context("parse git.object_format")?;
137 let default_branch =
138 BranchName::new(config.repo.default_branch.as_str()).context("parse default branch")?;
139 let layout = knot_git::Layout::new(&config.repo.scan_path)
140 .with_default_branch(default_branch)
141 .with_object_format(object_format)
142 .reserving_meta(&knot_did)
143 .context("reserve meta-repo path")?;
144
145 let swept = knot_pack::sweep_incoming(&config.repo.scan_path);
146 if swept > 0 {
147 tracing::info!(swept, "swept abandoned receive staging directories");
148 }
149
150 layout
151 .bootstrap_meta(&knot_did)
152 .context("bootstrap meta-repo")?;
153 let meta_path = layout
154 .meta_path(&knot_did)
155 .context("resolve meta-repo path")?;
156
157 let index = Arc::new(Index::new(meta_path.clone(), layout.clone()));
158 index.rebuild().context("rebuild index from meta-repo")?;
159 tracing::info!(coverage = ?index.coverage(), "index ready");
160
161 let warm = Arc::clone(&index);
162 tokio::task::spawn_blocking(move || warm.warm_collaborators());
163
164 let http = ReqwestHttp::new(config.http_limits()).context("build outbound HTTP client")?;
165 let git_http: Arc<dyn knot_runtime::HttpTransport> = Arc::new(
166 ReqwestHttp::new(config.fork_http_limits()).context("build outbound git fetch client")?,
167 );
168 let atproto = Arc::new(Atproto::new(
169 http,
170 SystemClock,
171 knot_did.clone(),
172 knot_atproto::PlcDirectory::new(config.atproto.plc_directory.clone())
173 .context("atproto.plc_directory isn't a valid PLC base URL")?,
174 ));
175 let admins: BTreeSet<_> = config.server.admins.iter().cloned().collect();
176 let admission = config.acl.admission;
177 let service_owner = config
178 .server
179 .admins
180 .first()
181 .cloned()
182 .context("at least one admin is configured")?;
183
184 let master_key = MasterKey::new(
185 base64::engine::general_purpose::STANDARD
186 .decode(
187 std::env::var(&config.secrets.master_key_env)
188 .context("read master key from environment")?
189 .trim(),
190 )
191 .context("decode master key as base64")?,
192 )
193 .context("master key from environment")?;
194 let secrets = Arc::new(
195 SealedStore::open(
196 &config.secrets.sealed_key_file,
197 &master_key,
198 Box::new(OsEntropy),
199 )
200 .context("open sealed key store")?,
201 );
202 let knot_signing_key = secrets
203 .ensure(&knot_did)
204 .context("seal knot's own signing key")?;
205 let knot_actor = ActorId::from_secp256k1(knot_signing_key.as_bytes());
206 let appview_endpoint = config.server.appview_endpoint.clone();
207 let knot_service_url = knot_types::KnotServiceUrl::new(format!("https://{hostname}"))
208 .context("server.hostname doesn't form a valid knot service URL")?;
209 let did_document =
210 knot_atproto::knot_did_document(&knot_did, &knot_signing_key, &knot_service_url);
211
212 let http_addr = config.server.listen_addr;
213 let listen_limits = knot_edge::ListenLimits::new(
214 knot_edge::HeaderTimeout::from_millis(
215 NonZeroU64::new(config.server.listen_header_timeout_ms)
216 .context("server.listen_header_timeout_ms must be greater than zero")?,
217 ),
218 knot_edge::IdleTimeout::from_millis(
219 NonZeroU64::new(config.server.listen_idle_timeout_ms)
220 .context("server.listen_idle_timeout_ms must be greater than zero")?,
221 ),
222 NonZeroU32::new(config.server.listen_max_connections)
223 .context("server.listen_max_connections must be greater than zero")?,
224 );
225 // A header name that doesn't parse will never match,
226 // `effective_peer` falls back to socket,
227 // and every request in the world shares
228 // the proxy's address + its one ratelimit bucket.
229 // So... better to refuse to start.
230 let trusted_proxy_header = config
231 .xrpc
232 .trusted_proxy_header
233 .as_deref()
234 .map(|header| axum::http::HeaderName::from_bytes(header.as_bytes()))
235 .transpose()
236 .context("xrpc.trusted_proxy_header isn't a valid HTTP header name")?;
237 let edge_guards = knot_edge::EdgeGuards::new(
238 knot_edge::RequestsPerSecond::new(
239 NonZeroU32::new(config.server.listen_rate_limit_per_second)
240 .context("server.listen_rate_limit_per_second must be greater than zero")?,
241 ),
242 knot_edge::BurstSize::new(
243 NonZeroU32::new(config.server.listen_rate_limit_burst)
244 .context("server.listen_rate_limit_burst must be greater than zero")?,
245 ),
246 knot_edge::MaxInflightRequests::new(
247 NonZeroU32::new(config.server.listen_max_inflight_requests)
248 .context("server.listen_max_inflight_requests must be greater than zero")?,
249 ),
250 knot_edge::RequestTimeout::from_millis(
251 NonZeroU64::new(config.server.listen_request_timeout_ms)
252 .context("server.listen_request_timeout_ms must be greater than zero")?,
253 ),
254 knot_edge::BodyInactivityTimeout::from_millis(
255 NonZeroU64::new(config.server.listen_body_timeout_ms)
256 .context("server.listen_body_timeout_ms must be greater than zero")?,
257 ),
258 knot_edge::WriteRequestTimeout::from_millis(
259 NonZeroU64::new(config.server.listen_write_request_timeout_ms)
260 .context("server.listen_write_request_timeout_ms must be greater than zero")?,
261 ),
262 trusted_proxy_header.clone(),
263 );
264 let tls_setup = build_tls_setup(&config, &hostname).context("assemble TLS configuration")?;
265 if config.tls.http3 && tls_setup.is_none() {
266 tracing::warn!(
267 "tls.http3 is set without any TLS certificate, so HTTP/3 won't start. Configure a static cert or ACME to serve h3."
268 );
269 }
270 if tls_setup.is_none() && config.xrpc.trusted_proxy_header.is_none() {
271 tracing::warn!(
272 "running plaintext behind a reverse proxy without xrpc.trusted_proxy_header. Per-IP rate limiting will key on the proxy socket address, throttling all clients as one. Set xrpc.trusted_proxy_header to the header your proxy appends."
273 );
274 }
275 if config.tls.acme_enabled && http_addr.port() != 443 {
276 tracing::warn!(
277 listen_port = http_addr.port(),
278 "ACME validation over TLS-ALPN-01 needs the certificate authority to reach this host on TCP 443. Map 443 to the listen port if it differs."
279 );
280 }
281 if config.tls.acme_enabled && config.tls.acme_staging {
282 tracing::warn!(
283 "ACME is using the Let's Encrypt staging directory. Its certificates aren't browser-trusted. Unset tls.acme_staging for real certificates."
284 );
285 }
286 let ssh_addr = config.server.ssh_listen_addr;
287 let ssh_max_pack_bytes = config.server.ssh_max_pack_bytes as usize;
288 let pack_limits = knot_pack::PackLimits {
289 max_objects: ObjectCount::from(config.pack.max_objects),
290 max_total_bytes: knot_pack::MaxTotalBytes::new(config.pack.max_total_bytes),
291 ..knot_pack::PackLimits::default()
292 };
293 knot_pack::init_selection_limits(knot_pack::SelectionLimits {
294 max_objects: ObjectCount::from(config.pack.selection_max_objects),
295 time_budget: Duration::from_secs(config.pack.selection_time_budget_secs),
296 });
297 let host_key = knot_ssh::load_or_create_host_key(&config.server.ssh_host_key_file)
298 .context("load or create SSH host key")?;
299
300 let xrpc_limits = knot_xrpc::LimitConfig {
301 rate: Some(knot_xrpc::RateLimit {
302 burst: knot_xrpc::Burst::new(config.xrpc.preauth_burst),
303 refill: knot_xrpc::RefillMicros::new(
304 config.xrpc.preauth_refill_ms.saturating_mul(1_000),
305 ),
306 }),
307 per_peer_inflight: Some(knot_xrpc::PerPeerInflight::new(
308 config.xrpc.per_peer_inflight as usize,
309 )),
310 global_inflight: Some(knot_xrpc::GlobalInflight::new(
311 config.xrpc.global_inflight as usize,
312 )),
313 };
314 let byte_limits = knot_xrpc::ByteLimits {
315 body: knot_xrpc::BodyLimit::new(config.xrpc.max_body_bytes as usize),
316 patch: knot_xrpc::PatchLimit::new(config.xrpc.max_patch_bytes as usize),
317 patch_decompressed: knot_xrpc::PatchDecompressedLimit::new(
318 config.xrpc.max_patch_decompressed_bytes,
319 ),
320 response: knot_xrpc::ResponseLimit::new(config.xrpc.max_response_bytes as usize),
321 archive: knot_xrpc::ArchiveLimit::new(config.xrpc.max_archive_bytes),
322 fork_pack: knot_xrpc::ForkPackLimit::new(config.xrpc.fork_max_pack_bytes),
323 pack: knot_xrpc::MaxWireBytes::new(ssh_max_pack_bytes),
324 };
325 let budgets = knot_xrpc::Budgets {
326 tree_last_commit: knot_xrpc::TreeReadBudget::new(knot_xrpc::ReadBudget::Within(
327 Duration::from_millis(config.xrpc.tree_last_commit_budget_ms),
328 )),
329 blob_last_commit: knot_xrpc::BlobReadBudget::new(knot_xrpc::ReadBudget::Within(
330 Duration::from_millis(config.xrpc.blob_last_commit_budget_ms),
331 )),
332 languages: knot_xrpc::LanguagesReadBudget::new(knot_xrpc::ReadBudget::Within(
333 Duration::from_millis(config.xrpc.languages_budget_ms),
334 )),
335 languages_push: knot_xrpc::LanguagesPushBudget::new(Duration::from_millis(
336 config.xrpc.languages_push_budget_ms,
337 )),
338 };
339 let committer = knot_xrpc::Committer {
340 name: AuthorName::new(config.git.user_name.clone()),
341 email: Email::new(config.git.user_email.clone()),
342 };
343 let reservations = Arc::new(knot_xrpc::Reservations::new(
344 knot_xrpc::ReservationTtl::new(config.xrpc.reservation_ttl_secs as i64),
345 knot_xrpc::PerActorQuota::new(config.xrpc.per_actor_reservations as usize),
346 knot_xrpc::GlobalQuota::new(config.xrpc.max_pending_reservations as usize),
347 ));
348 let replay_bounds = knot_events::ReplayBounds::new(
349 knot_events::ReplayEvents::new(config.xrpc.events_replay_buffer as usize)
350 .context("xrpc.events_replay_buffer must be greater than zero")?,
351 knot_events::ReplayBytes::new(config.xrpc.events_replay_bytes as usize)
352 .context("xrpc.events_replay_bytes must be greater than zero")?,
353 );
354 let events = Arc::new(knot_events::EventLog::new(SystemClock, replay_bounds));
355 let subscriber_gate = Arc::new(knot_events::SubscriberGate::new(
356 knot_events::GlobalSubscriberLimit::new(config.xrpc.events_max_subscribers as usize),
357 knot_events::PerPeerSubscriberLimit::new(config.xrpc.events_max_per_peer as usize),
358 ));
359
360 let maintenance_enabled = config.maintenance.enabled;
361 let maintenance_options = knot_maintenance::Options::from_config(&config.maintenance);
362 let maintenance_interval = Duration::from_secs(config.maintenance.interval_secs);
363 let maintenance_large_push =
364 knot_maintenance::PushBytes::new(config.maintenance.large_push_bytes);
365
366 let lfs_handle = config
367 .lfs
368 .store_path
369 .as_ref()
370 .map(|path| {
371 knot_lfs::LfsHandle::open(
372 knot_lfs::LfsStorePath::new(path),
373 knot_lfs::LfsSize::new(config.lfs.max_object_bytes),
374 knot_lfs::FreeSpaceFloor::new(config.lfs.free_space_floor_bytes),
375 )
376 .inspect(|_| {
377 tracing::info!(store = %path.display(), "git-lfs capability enabled");
378 })
379 })
380 .transpose()
381 .context("open LFS object store")?;
382 let lfs_max_ssh_transfers = config.lfs.max_ssh_transfers as usize;
383 let lfs_max_http_downloads = config.lfs.max_http_downloads as usize;
384 let lfs_gc_grace = knot_maintenance::lfs_grace(
385 knot_maintenance::GcGrace::from_secs(config.lfs.gc_grace_secs),
386 knot_maintenance::ReflogRetention::from_secs(config.maintenance.reflog_expire_secs),
387 );
388 let lfs_gc_interval =
389 knot_maintenance::SweepInterval::new(Duration::from_secs(config.lfs.gc_interval_secs));
390 if lfs_handle.is_some() && !maintenance_enabled {
391 tracing::warn!(
392 "LFS store configured but maintenance is disabled. Unreferenced LFS objects will accumulate with no garbage collection or orphan sweep."
393 );
394 }
395
396 let pack_cache_config = knot_pack::CacheConfig {
397 enabled: config.pack_cache.enabled,
398 ttl: Duration::from_secs(config.pack_cache.ttl_secs),
399 max_entry_bytes: knot_pack::MaxEntryBytes::new(config.pack_cache.max_entry_bytes as usize),
400 max_total_bytes: knot_pack::MaxCacheBytes::new(knot_resource::pack_cache_bytes(
401 config.pack_cache.max_total_bytes,
402 ) as usize),
403 };
404
405 let ci_logs = config
406 .ci
407 .logs_addr
408 .as_deref()
409 .map(CiLogsAddr::new)
410 .transpose()
411 .context("ci.logs_addr must be host:port")?;
412
413 let homepage = config.homepage.source();
414 let catalog = Arc::new(
415 knot_messages::Catalog::parse(&config.messages).context("parse message templates")?,
416 );
417
418 knot_config::init(config);
419
420 let (maintenance_handle, maintenance_shutdown, maintenance_task) = if maintenance_enabled {
421 let (scheduler, handle) = knot_maintenance::Scheduler::new(
422 layout.clone(),
423 Arc::new(IndexRepos(Arc::clone(&index))),
424 SystemClock,
425 maintenance_options,
426 maintenance_interval,
427 maintenance_large_push,
428 );
429 let scheduler = match &lfs_handle {
430 Some(lfs) => {
431 scheduler.with_lfs_gc(Arc::clone(&lfs.store), lfs_gc_grace, lfs_gc_interval)
432 }
433 None => scheduler,
434 };
435 let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
436 let task = tokio::spawn(scheduler.run(shutdown_rx));
437 tracing::info!(
438 interval_secs = maintenance_interval.as_secs(),
439 "maintenance scheduler running"
440 );
441 (handle, Some(shutdown_tx), Some(task))
442 } else {
443 (knot_maintenance::MaintenanceHandle::disabled(), None, None)
444 };
445
446 let slots = knot_resource::Slots::for_machine();
447
448 let ssh_base = knot_ssh::SshState::new(
449 layout.clone(),
450 Arc::clone(&index),
451 Arc::clone(&atproto),
452 knot_actor,
453 Arc::clone(&events),
454 hostname.clone(),
455 appview_endpoint.clone(),
456 admins.clone(),
457 admission,
458 byte_limits.pack,
459 budgets.languages_push,
460 ci_logs.clone(),
461 )
462 .with_maintenance(maintenance_handle.clone())
463 .with_limits(pack_limits)
464 .with_slots(slots.clone())
465 .with_catalog(Arc::clone(&catalog));
466 let ssh_state = Arc::new(match &lfs_handle {
467 Some(handle) => ssh_base.with_lfs(handle.clone(), lfs_max_ssh_transfers),
468 None => ssh_base,
469 });
470
471 let xrpc_state = Arc::new(XrpcState {
472 layout: layout.clone(),
473 index: Arc::clone(&index),
474 atproto: Arc::clone(&atproto),
475 secrets,
476 entropy: Arc::new(OsEntropy),
477 ci_logs,
478 admins,
479 admission,
480 knot_did,
481 knot_hostname: hostname,
482 meta_path,
483 knot_service_url,
484 limiter: Arc::new(knot_xrpc::PreAuthLimiter::with_config(xrpc_limits)),
485 cob_locks: Arc::new(knot_xrpc::CobLocks::default()),
486 reservations,
487 trusted_proxy_header,
488 committer,
489 byte_limits,
490 budgets,
491 git_http,
492 pack_limits,
493 service_owner,
494 subscriber_gate,
495 maintenance: maintenance_handle,
496 appview: appview_endpoint,
497 slots: slots.clone(),
498 events,
499 lfs: lfs_handle.map(|handle| knot_xrpc::LfsWeb::new(handle, lfs_max_http_downloads)),
500 catalog: Arc::clone(&catalog),
501 });
502
503 let resolver: Arc<dyn knot_pack::RepoResolver> = {
504 let index = Arc::clone(&index);
505 Arc::new(move |target: &knot_pack::RepoTarget| match target {
506 knot_pack::RepoTarget::Did(did) => {
507 knot_pack::RepoLookup::from_resolved(index.owner_of(did), |_| did.clone())
508 }
509 knot_pack::RepoTarget::OwnerRkey(owner, rkey) => {
510 knot_pack::RepoLookup::from_resolved(index.resolve_repo(owner, rkey), |found| found)
511 }
512 })
513 };
514 let receive_advertiser = knot_xrpc::receive_advertiser(Arc::clone(&xrpc_state));
515 let handle_resolver: Arc<dyn knot_pack::HandleResolver> = Arc::new(AtprotoHandleResolver {
516 atproto: Arc::clone(&atproto),
517 });
518 let (write_routes, early_data_safe) = knot_pack::edge_routes(
519 layout,
520 resolver,
521 Some(receive_advertiser),
522 Some(handle_resolver),
523 slots.pack.clone(),
524 pack_cache_config,
525 Arc::clone(&catalog),
526 xrpc_state.knot_hostname.clone(),
527 Arc::new(SystemClock),
528 );
529 let base_router = write_routes.merge(knot_xrpc::router(xrpc_state)).route(
530 "/.well-known/did.json",
531 get(move || {
532 let document = did_document.clone();
533 async move { Json(document) }
534 }),
535 );
536 let base_router = match homepage {
537 HomepageSource::Disabled => base_router,
538 HomepageSource::Default => base_router.route("/", get(|| async { Html(DEFAULT_HOMEPAGE) })),
539 HomepageSource::File(path) => base_router.route_service("/", ServeFile::new(path)),
540 };
541 let app = knot_edge::RequiresFullHandshake::new(base_router);
542 let scheme = if tls_setup.is_some() { "https" } else { "http" };
543 let edge_config = knot_edge::EdgeConfig {
544 http_addr: knot_edge::PublicBind::new(http_addr),
545 limits: listen_limits,
546 guards: edge_guards,
547 tls: tls_setup,
548 };
549 tracing::info!("listening on {scheme}://{http_addr} and ssh://{ssh_addr}");
550
551 let shutdown = CancellationToken::new();
552 tokio::spawn(allocator::govern_decay(shutdown.clone()));
553 let mut edge_task = tokio::spawn(knot_edge::serve(
554 edge_config,
555 app,
556 early_data_safe,
557 shutdown.clone(),
558 ));
559 let mut ssh_task = {
560 let shutdown = shutdown.clone();
561 tokio::spawn(async move { knot_ssh::serve(ssh_addr, host_key, ssh_state, shutdown).await })
562 };
563
564 let exit = tokio::select! {
565 result = &mut edge_task => FirstExit::Edge(result),
566 result = &mut ssh_task => FirstExit::Ssh(result),
567 () = shutdown_signal() => {
568 tracing::info!("shutdown signal received");
569 FirstExit::Signal
570 }
571 };
572 shutdown.cancel();
573 let drain = async {
574 match &exit {
575 FirstExit::Edge(_) => {
576 let _ = (&mut ssh_task).await;
577 }
578 FirstExit::Ssh(_) => {
579 let _ = (&mut edge_task).await;
580 }
581 FirstExit::Signal => {
582 let _ = (&mut edge_task).await;
583 let _ = (&mut ssh_task).await;
584 }
585 }
586 };
587 if tokio::time::timeout(EDGE_SHUTDOWN_DRAIN, drain)
588 .await
589 .is_err()
590 {
591 tracing::warn!(
592 timeout_secs = EDGE_SHUTDOWN_DRAIN.as_secs(),
593 "aborting edge drain after timeout"
594 );
595 }
596 if let Some(shutdown) = maintenance_shutdown {
597 let _ = shutdown.send(true);
598 }
599 if let Some(task) = maintenance_task
600 && tokio::time::timeout(MAINTENANCE_SHUTDOWN_DRAIN, task)
601 .await
602 .is_err()
603 {
604 tracing::warn!(
605 timeout_secs = MAINTENANCE_SHUTDOWN_DRAIN.as_secs(),
606 "aborting maintenance drain after timeout :3"
607 );
608 }
609 match exit {
610 FirstExit::Edge(result) => result
611 .context("edge server task panicked")?
612 .context("serve edge")?,
613 FirstExit::Ssh(result) => result
614 .context("ssh server task panicked")?
615 .context("serve ssh")?,
616 FirstExit::Signal => {}
617 }
618 Ok(())
619}
620
621enum FirstExit {
622 Edge(Result<Result<(), knot_edge::EdgeError>, tokio::task::JoinError>),
623 Ssh(Result<Result<(), knot_ssh::SshError>, tokio::task::JoinError>),
624 Signal,
625}
626
627fn build_tls_setup(
628 config: &knot_config::KnotConfig,
629 hostname: &knot_types::KnotHostname,
630) -> anyhow::Result<Option<knot_edge::TlsSetup>> {
631 let tls = &config.tls;
632 let source = if tls.acme_enabled {
633 knot_edge::CertSource::Acme(knot_edge::AcmeParams {
634 domains: vec![hostname.clone()],
635 contact: knot_edge::AcmeContact::new(
636 tls.acme_contact
637 .clone()
638 .context("tls.acme_contact is required when ACME is enabled")?,
639 )?,
640 cache_dir: knot_edge::AcmeCacheDir::new(
641 tls.acme_cache_dir
642 .clone()
643 .context("tls.acme_cache_dir is required when ACME is enabled")?,
644 ),
645 production: !tls.acme_staging,
646 })
647 } else {
648 match (&tls.cert_path, &tls.key_path) {
649 (Some(cert_path), Some(key_path)) => {
650 knot_edge::CertSource::Static(knot_edge::StaticCertPaths {
651 cert_path: knot_edge::CertChainPath::new(cert_path.clone()),
652 key_path: knot_edge::PrivateKeyPath::new(key_path.clone()),
653 })
654 }
655 _ => return Ok(None),
656 }
657 };
658
659 let internal = match tls.mtls_enabled {
660 true => Some(knot_edge::InternalTls {
661 addr: knot_edge::InternalBind::new(config.server.internal_listen_addr),
662 client_ca_path: knot_edge::ClientCaPath::new(
663 tls.mtls_client_ca_path
664 .clone()
665 .context("tls.mtls_client_ca_path is required when mTLS is enabled")?,
666 ),
667 spki_pin: knot_edge::SpkiPin::from_base64(
668 tls.mtls_admin_spki_pin
669 .as_deref()
670 .context("tls.mtls_admin_spki_pin is required when mTLS is enabled")?,
671 )
672 .context("parse tls.mtls_admin_spki_pin")?,
673 }),
674 false => None,
675 };
676
677 Ok(Some(knot_edge::TlsSetup {
678 source,
679 http3: tls.http3,
680 internal,
681 }))
682}
683
684async fn shutdown_signal() {
685 let interrupt = async {
686 let _ = tokio::signal::ctrl_c().await;
687 };
688 #[cfg(unix)]
689 let terminate = async {
690 match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {
691 Ok(mut stream) => {
692 stream.recv().await;
693 }
694 Err(_) => std::future::pending::<()>().await,
695 }
696 };
697 #[cfg(not(unix))]
698 let terminate = std::future::pending::<()>();
699
700 tokio::select! {
701 () = interrupt => {}
702 () = terminate => {}
703 }
704}