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