alice.pds.demo.boltless.dev
This repository has no description
1use std::net::SocketAddr;
2use std::num::NonZeroUsize;
3use std::path::PathBuf;
4use std::process::ExitCode;
5use std::sync::Arc;
6use std::time::Duration;
7
8use anyhow::{Context, anyhow};
9use bobbin_edge_index::{CoverageWatch, EdgeStore, HydrantCursor, StateIndex};
10use bobbin_ingest::{
11 IngestConfig, IngestRuntime, RepoIdResolver, WarmingBuffer, run as run_ingest,
12};
13use bobbin_knot_ingest::{CapabilityGate, KnotClient, KnotRegistry, Orchestrator};
14use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig, classify_ip};
15use bobbin_record_lru::{CacheCapacity, LruRecordStore, RecordStore};
16use bobbin_runtime::{
17 Clock, GuardedWs, MemoryBudget, NetworkError, OsEntropy, RuntimeHasher, SystemClock,
18 TungsteniteWs, WsTransport,
19};
20use bobbin_search::{SearchIndex, SearchReader};
21use bobbin_slingshot_client::SlingshotClient;
22use bobbin_xrpc::{
23 AppState, HeavyLimiter, MaxInFlight, PerRequestAnonBytes, ReservedFloor, router,
24};
25use clap::{Parser, Subcommand};
26use tokio::signal::unix::{SignalKind, signal};
27use tokio::task::JoinHandle;
28use tokio_util::sync::CancellationToken;
29use tracing::level_filters::LevelFilter;
30use tracing_subscriber::EnvFilter;
31
32mod config;
33mod mem;
34
35use config::{BobbinConfig, LogFormat};
36
37const BASE_RESIDENT_BYTES: u64 = 48 * 1024 * 1024;
38
39#[global_allocator]
40static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
41
42#[used]
43#[unsafe(no_mangle)]
44pub static malloc_conf: Option<&'static core::ffi::c_char> = Some(unsafe {
45 &*c"narenas:8,dirty_decay_ms:3000,muzzy_decay_ms:3000,background_thread:true".as_ptr()
46});
47
48#[derive(Parser)]
49#[command(name = "bobbin", about = "Read-only AppView for Tangled records")]
50struct Cli {
51 /// Path to a TOML config file. Environment variables override file values
52 /// for any `BOBBIN_*` setting; `/etc/bobbin/config.toml` is consulted as a
53 /// final fallback so distro packaging can drop a default in place.
54 #[arg(short, long, value_name = "FILE", env = "BOBBIN_CONFIG")]
55 config: Option<PathBuf>,
56
57 #[command(subcommand)]
58 command: Option<Command>,
59}
60
61#[derive(Subcommand)]
62enum Command {
63 /// Print a fully-commented TOML template to stdout. Use this to seed
64 /// `config.toml` for a fresh deploy.
65 ConfigTemplate,
66 /// Load and validate the configuration without starting the server.
67 Validate,
68}
69
70#[tokio::main]
71async fn main() -> ExitCode {
72 let cli = Cli::parse();
73
74 if let Some(Command::ConfigTemplate) = cli.command {
75 print!("{}", config::template());
76 return ExitCode::SUCCESS;
77 }
78
79 let cfg = match config::load(cli.config.as_ref()) {
80 Ok(c) => c,
81 Err(e) => {
82 eprintln!("failed to load configuration: {e:#}");
83 return ExitCode::FAILURE;
84 }
85 };
86
87 if matches!(cli.command, Some(Command::Validate)) {
88 if let Err(e) = init_tracing(&cfg, Arc::new(SystemClock::new())) {
89 eprintln!("failed to install tracing subscriber: {e}");
90 return ExitCode::FAILURE;
91 }
92 println!("configuration is valid");
93 return ExitCode::SUCCESS;
94 }
95
96 match run(cfg).await {
97 Ok(()) => ExitCode::SUCCESS,
98 Err(e) => {
99 tracing::error!(error = ?e, "fatal");
100 ExitCode::FAILURE
101 }
102 }
103}
104
105struct ClockTimer(Arc<dyn Clock>);
106
107impl tracing_subscriber::fmt::time::FormatTime for ClockTimer {
108 fn format_time(&self, w: &mut tracing_subscriber::fmt::format::Writer<'_>) -> std::fmt::Result {
109 write!(w, "{}", self.0.now_unix_micros().raw())
110 }
111}
112
113fn init_tracing(cfg: &BobbinConfig, clock: Arc<dyn Clock>) -> Result<(), String> {
114 let combined = format!("{},{}", LevelFilter::INFO, cfg.log.filter);
115 let format: LogFormat = cfg.log.format.parse()?;
116 let timer = ClockTimer(clock);
117 match format {
118 LogFormat::Text => {
119 let filter = EnvFilter::try_new(&combined)
120 .map_err(|e| format!("invalid log filter `{}`: {e}", cfg.log.filter))?;
121 tracing_subscriber::fmt()
122 .with_env_filter(filter)
123 .with_timer(timer)
124 .try_init()
125 .map_err(|e| e.to_string())
126 }
127 LogFormat::Json => {
128 let filter = EnvFilter::try_new(&combined)
129 .map_err(|e| format!("invalid log filter `{}`: {e}", cfg.log.filter))?;
130 tracing_subscriber::fmt()
131 .with_env_filter(filter)
132 .json()
133 .with_timer(timer)
134 .try_init()
135 .map_err(|e| e.to_string())
136 }
137 }
138}
139
140async fn run(cfg: BobbinConfig) -> anyhow::Result<()> {
141 let clock: Arc<dyn Clock> = Arc::new(SystemClock::new());
142 init_tracing(&cfg, clock.clone())
143 .map_err(|e| anyhow!("failed to install tracing subscriber: {e}"))?;
144
145 let (budget, budget_source) = mem::detect_budget();
146 tracing::info!(
147 source = ?budget_source,
148 budget_bytes = budget.map(MemoryBudget::bytes),
149 "memory budget detected"
150 );
151 if let Some(b) = budget {
152 mem::try_set_high(b);
153 }
154 let lru_cap = mem::lru_bytes(budget, cfg.record_cache.lru_bytes);
155 let search_heap_cap = mem::search_heap_bytes(budget, cfg.search.heap_bytes);
156 if budget.is_some() {
157 tracing::info!(
158 lru_bytes = lru_cap,
159 search_heap_bytes = search_heap_cap,
160 "constrained cache sizing"
161 );
162 }
163 let limiter = budget.map(|b| {
164 let reserved = ReservedFloor::new(
165 BASE_RESIDENT_BYTES
166 .saturating_add(cfg.backpressure.reserved_index_bytes)
167 .saturating_add(search_heap_cap)
168 .saturating_add(lru_cap),
169 );
170 let per_request = PerRequestAnonBytes::new(cfg.backpressure.per_request_anon_bytes);
171 let max = MaxInFlight::from_budget(b, reserved, per_request);
172 tracing::info!(max_in_flight = max.get(), "heavy-request concurrency cap");
173 Arc::new(HeavyLimiter::new(max))
174 });
175
176 let entropy = Arc::new(OsEntropy);
177 let hasher = RuntimeHasher::from_entropy(&*entropy);
178 let ws = TungsteniteWs::shared();
179
180 let records: Arc<dyn RecordStore> =
181 Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(lru_cap)));
182 let slingshot = SlingshotClient::with_default_http(cfg.slingshot.url.clone())?;
183 let resolver = Arc::new(RepoIdResolver::with_slingshot(
184 slingshot.clone(),
185 clock.clone(),
186 hasher.clone(),
187 ));
188 let edges = Arc::new(EdgeStore::new(hasher.clone()));
189 let issue_states = Arc::new(StateIndex::new(hasher.clone()));
190 let pull_statuses = Arc::new(StateIndex::new(hasher.clone()));
191 let coverage = Arc::new(CoverageWatch::new());
192 let warming_buffer = Arc::new(WarmingBuffer::new(hasher.clone()));
193 let knot_registry = Arc::new(KnotRegistry::new());
194 let knots = Arc::new(KnotProxy::new(
195 KnotProxyConfig {
196 allow_private_hosts: cfg.knot.allow_private,
197 require_https: cfg.knot.require_https,
198 ..KnotProxyConfig::default()
199 },
200 KnotHttpConfig::default(),
201 clock.clone(),
202 hasher,
203 )?);
204 let search_heap = usize::try_from(search_heap_cap)
205 .with_context(|| format!("search heap {search_heap_cap} exceeds usize"))?;
206 let search = Arc::new(SearchIndex::new(search_heap, clock.clone())?);
207
208 let configured_parallelism = NonZeroUsize::new(cfg.ingest.parallelism)
209 .ok_or_else(|| anyhow!("ingest.parallelism must be at least 1"))?;
210 let parallelism = mem::ingest_parallelism(budget, configured_parallelism);
211 tracing::info!(
212 configured = configured_parallelism.get(),
213 effective = parallelism.get(),
214 "ingest parallelism"
215 );
216 let ingest_cfg = IngestConfig {
217 hydrant_base: cfg.hydrant.url.clone(),
218 start_cursor: HydrantCursor::new(cfg.hydrant.start_cursor),
219 parallelism,
220 };
221 let cancel = CancellationToken::new();
222 let ingest_coverage = coverage.clone();
223
224 let knot_acl_dev = !cfg.knot.require_https;
225 let knot_allow_private = cfg.knot.allow_private;
226 let knot_client = KnotClient::with_default_http(knot_allow_private)?;
227 let knot_gate = Arc::new(CapabilityGate::new(
228 knot_client.clone(),
229 clock.clone(),
230 knot_acl_dev,
231 knot_allow_private,
232 ));
233 let knot_ws: Arc<dyn WsTransport> = if knot_allow_private {
234 ws.clone()
235 } else {
236 GuardedWs::shared(Arc::new(|addrs: &[SocketAddr]| {
237 match addrs.iter().find_map(|sa| classify_ip(&sa.ip())) {
238 Some(reason) => Err(NetworkError::Connect(format!(
239 "knot eventstream resolves to {reason} address space"
240 ))),
241 None => Ok(()),
242 }
243 }))
244 };
245
246 let ingest_runtime = IngestRuntime {
247 store: edges.clone(),
248 issue_states: issue_states.clone(),
249 pull_statuses: pull_statuses.clone(),
250 coverage: coverage.clone(),
251 search: search.clone(),
252 records: records.clone(),
253 resolver: resolver.clone(),
254 clock: clock.clone(),
255 entropy,
256 ws: ws.clone(),
257 cancel: cancel.clone(),
258 disconnects: None,
259 warming_shadow: None,
260 warming_buffer: Some(warming_buffer),
261 knot_registry: Some(knot_registry.clone()),
262 knot_gate: Some(knot_gate.clone()),
263 };
264 let mut ingest_handle = tokio::spawn(run_ingest(ingest_cfg, ingest_runtime));
265
266 let knot_orchestrator = Orchestrator {
267 client: Arc::new(knot_client),
268 gate: knot_gate,
269 registry: knot_registry,
270 store: edges.clone(),
271 ws: knot_ws,
272 clock: clock.clone(),
273 dev: knot_acl_dev,
274 allow_private: knot_allow_private,
275 cancel: cancel.clone(),
276 };
277 let _knot_acl_handle = tokio::spawn(knot_orchestrator.run());
278
279 let _adaptive_watcher = budget.zip(limiter.as_ref()).map(|(b, l)| {
280 mem::spawn_adaptive_watcher(
281 l.clone(),
282 clock.clone(),
283 b,
284 mem::AdaptiveThresholds {
285 interval: Duration::from_millis(cfg.backpressure.adjust_interval_ms),
286 relieve_below_ratio: cfg.backpressure.relieve_below_ratio,
287 tighten_above_ratio: cfg.backpressure.tighten_above_ratio,
288 },
289 cancel.clone(),
290 )
291 });
292
293 let debug_bind: Option<SocketAddr> =
294 if cfg.server.debug_bind.is_empty() {
295 None
296 } else {
297 Some(cfg.server.debug_bind.parse().with_context(|| {
298 format!("invalid server.debug_bind `{}`", cfg.server.debug_bind)
299 })?)
300 };
301 let mem_probe = debug_bind.is_some().then(|| mem::MemProbe {
302 edges: edges.clone(),
303 search: search.clone(),
304 records: records.clone(),
305 issue_states: issue_states.clone(),
306 pull_statuses: pull_statuses.clone(),
307 });
308 let state = AppState::new(
309 records,
310 slingshot,
311 edges,
312 issue_states,
313 pull_statuses,
314 coverage,
315 knots,
316 search as Arc<dyn SearchReader>,
317 resolver,
318 )
319 .with_limiter(limiter);
320 let app = router(state);
321
322 let _debug_server = match (debug_bind, mem_probe) {
323 (Some(addr), Some(probe)) => {
324 let debug_app = mem::debug_router(probe);
325 let debug_cancel = cancel.clone();
326 Some(tokio::spawn(async move {
327 match bind_listener(addr) {
328 Ok(listener) => {
329 tracing::info!(%addr, "debug endpoints bound, keep this loopback-only");
330 let shutdown = async move { debug_cancel.cancelled().await };
331 if let Err(e) = axum::serve(listener, debug_app)
332 .with_graceful_shutdown(shutdown)
333 .await
334 {
335 tracing::warn!(error = %e, "debug endpoint server exited with error");
336 }
337 }
338 Err(e) => {
339 tracing::warn!(%addr, error = %e, "could not bind debug endpoints");
340 }
341 }
342 }))
343 }
344 _ => None,
345 };
346
347 let binds = cfg.server.binds.clone();
348 let hydrant_url = cfg.hydrant.url.as_str().to_owned();
349 let slingshot_url = cfg.slingshot.url.as_str().to_owned();
350 let grace = Duration::from_secs(cfg.server.shutdown_grace_secs);
351 let bind_display = binds
352 .iter()
353 .map(SocketAddr::to_string)
354 .collect::<Vec<_>>()
355 .join(",");
356 tracing::info!(binds = %bind_display, %hydrant_url, %slingshot_url, "bobbin listening");
357
358 let signal_cancel = cancel.clone();
359 let mut server_handle = tokio::spawn(serve_all(binds, app, signal_cancel));
360
361 tokio::select! {
362 res = &mut server_handle => {
363 cancel.cancel();
364 let cursor = ingest_coverage.snapshot().last_cursor().raw();
365 tracing::info!(grace_secs = grace.as_secs(), cursor, "draining ingest");
366 drain_with_grace("ingest", grace, &mut ingest_handle, clock.as_ref()).await;
367 match res {
368 Ok(Ok(())) => Ok(()),
369 Ok(Err(e)) => Err(anyhow::Error::from(e)).context("axum server failed"),
370 Err(join) => Err(anyhow!("server task panicked: {join}")),
371 }
372 }
373 res = &mut ingest_handle => {
374 cancel.cancel();
375 let cursor = ingest_coverage.snapshot().last_cursor().raw();
376 tracing::info!(grace_secs = grace.as_secs(), cursor, "draining server");
377 drain_with_grace("server", grace, &mut server_handle, clock.as_ref()).await;
378 match res {
379 Ok(Ok(())) => Err(anyhow!("ingest run loop exited; loop is supposed to be infinite")),
380 Ok(Err(e)) => Err(anyhow::Error::from(e)).context("ingest exited"),
381 Err(join) => Err(anyhow!("ingest task panicked: {join}")),
382 }
383 }
384 }
385}
386
387async fn drain_with_grace<T>(
388 label: &'static str,
389 grace: Duration,
390 handle: &mut JoinHandle<T>,
391 clock: &dyn Clock,
392) {
393 tokio::select! {
394 _ = &mut *handle => {}
395 _ = clock.sleep(grace) => {
396 tracing::warn!(
397 grace_secs = grace.as_secs(),
398 label,
399 "task did not stop within grace, aborting"
400 );
401 handle.abort();
402 }
403 }
404}
405
406async fn serve_all(
407 binds: Vec<SocketAddr>,
408 app: axum::Router,
409 cancel: CancellationToken,
410) -> std::io::Result<()> {
411 let listeners = futures::future::try_join_all(binds.into_iter().map(|addr| async move {
412 let listener = bind_listener(addr)?;
413 tracing::info!(%addr, "bobbin listener bound");
414 Ok::<_, std::io::Error>(listener)
415 }))
416 .await?;
417
418 let trigger = cancel.clone();
419 let signal_task = tokio::spawn(async move {
420 wait_for_shutdown().await;
421 tracing::info!("shutdown signal received, draining server");
422 trigger.cancel();
423 });
424
425 let services = listeners.into_iter().map(|listener| {
426 let app = app.clone();
427 let cancel = cancel.clone();
428 async move {
429 axum::serve(listener, app)
430 .with_graceful_shutdown(async move { cancel.cancelled().await })
431 .await
432 }
433 });
434
435 let result = futures::future::try_join_all(services).await.map(|_| ());
436 signal_task.abort();
437 result
438}
439
440fn bind_listener(addr: SocketAddr) -> std::io::Result<tokio::net::TcpListener> {
441 let domain = match addr {
442 SocketAddr::V4(_) => socket2::Domain::IPV4,
443 SocketAddr::V6(_) => socket2::Domain::IPV6,
444 };
445 let socket = socket2::Socket::new(domain, socket2::Type::STREAM, Some(socket2::Protocol::TCP))?;
446 if matches!(addr, SocketAddr::V6(_)) {
447 socket.set_only_v6(true)?;
448 }
449 socket.set_reuse_address(true)?;
450 socket.set_nonblocking(true)?;
451 socket.bind(&addr.into())?;
452 socket.listen(1024)?;
453 let std_listener: std::net::TcpListener = socket.into();
454 tokio::net::TcpListener::from_std(std_listener)
455}
456
457async fn wait_for_shutdown() {
458 let ctrl_c = tokio::signal::ctrl_c();
459 let mut sigterm = match signal(SignalKind::terminate()) {
460 Ok(s) => s,
461 Err(e) => {
462 tracing::warn!(
463 ?e,
464 "could not install SIGTERM handler, shutdown will only honor ctrl-c"
465 );
466 ctrl_c.await.ok();
467 return;
468 }
469 };
470 tokio::select! {
471 _ = ctrl_c => {}
472 _ = sigterm.recv() => {}
473 }
474}
475
476#[cfg(test)]
477mod tests {
478 use super::*;
479
480 #[tokio::test(start_paused = true)]
481 async fn drain_returns_immediately_when_task_already_done() {
482 let mut handle = tokio::spawn(async { 7u32 });
483 tokio::time::advance(Duration::from_millis(1)).await;
484 let start = tokio::time::Instant::now();
485 drain_with_grace(
486 "test",
487 Duration::from_secs(60),
488 &mut handle,
489 &SystemClock::new(),
490 )
491 .await;
492 assert!(start.elapsed() < Duration::from_millis(10));
493 }
494
495 #[tokio::test(start_paused = true)]
496 async fn drain_aborts_runaway_task_after_grace() {
497 let mut handle = tokio::spawn(async {
498 std::future::pending::<()>().await;
499 });
500 let grace = Duration::from_secs(5);
501 let start = tokio::time::Instant::now();
502 drain_with_grace("test", grace, &mut handle, &SystemClock::new()).await;
503 assert!(start.elapsed() >= grace);
504 let outcome = handle.await;
505 assert!(outcome.is_err() && outcome.unwrap_err().is_cancelled());
506 }
507
508 #[test]
509 fn typo_filter_keeps_info_default_for_other_targets() {
510 let filter = format!("{},{}", LevelFilter::INFO, "blah_invalid");
511 let parsed = EnvFilter::try_new(&filter).expect("filter parses");
512 let rendered = parsed.to_string();
513 assert!(
514 rendered.contains("info"),
515 "expected info default, got {rendered}"
516 );
517 assert!(
518 rendered.contains("blah_invalid"),
519 "expected user override, got {rendered}",
520 );
521 }
522
523 #[test]
524 fn explicit_user_level_overrides_info_default() {
525 let filter = format!("{},{}", LevelFilter::INFO, "warn");
526 let parsed = EnvFilter::try_new(&filter).expect("filter parses");
527 assert_eq!(parsed.to_string(), "warn");
528 }
529}