# How Shadowbroker Backend Manages Background Jobs and Scheduled Tasks

> Discover how Shadowbroker backend excels at managing background jobs and scheduled tasks using APScheduler and interval/cron triggers for reliable, isolated task execution.

- Repository: [Shadowbroker/Shadowbroker](https://github.com/BigBodyCobain/Shadowbroker)
- Tags: internals
- Published: 2026-05-07

---

**The Shadowbroker backend uses APScheduler to run a daemon-mode BackgroundScheduler that orchestrates periodic and on-demand background jobs through interval and cron triggers, with each task wrapped in a health-monitoring decorator to isolate failures.**

The BigBodyCobain/Shadowbroker repository implements a robust job scheduling system to continuously ingest data from external APIs, satellites, and sensor networks. The architecture centralizes how the backend manages background jobs and scheduled tasks within a single service module, leveraging Python’s `concurrent.futures` alongside APScheduler to balance concurrency with resource constraints.

## Scheduler Initialization and Daemon Architecture

The scheduling lifecycle begins in [`backend/services/data_fetcher.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/services/data_fetcher.py) via the `start_scheduler()` function. This method instantiates a **daemon-mode `BackgroundScheduler`**, which runs in a background thread and terminates automatically when the main process exits.

```python

# backend/services/data_fetcher.py – start_scheduler()

_scheduler = BackgroundScheduler(daemon=True)

```

Setting `daemon=True` ensures that the scheduler does not block the main application thread during shutdown. After instantiation, the function registers all job definitions and invokes `_scheduler.start()` to begin execution.

## Job Registration and Trigger Types

Jobs are registered using the `add_job()` method, which supports both **interval** (time-delta based) and **cron** (calendar-based) triggers. Each registration includes explicit concurrency controls via `max_instances=1` and `misfire_grace_time` parameters to prevent overlapping runs and handle missed executions gracefully.

### Interval-Based Execution

High-frequency data fetchers use the `"interval"` trigger. The fast-tier data refresh job executes every 60 seconds to update flight, ship, and satellite positions:

```python
_scheduler.add_job(
    lambda: _run_task_with_health(update_fast_data, "update_fast_data"),
    "interval",
    seconds=60,
    id="fast_tier",
    max_instances=1,
    misfire_grace_time=30,
)

```

Similarly, the Oracle resolution sweep runs hourly with a larger grace period for recovery:

```python
_scheduler.add_job(
    lambda: _run_task_with_health(_oracle_resolution_sweep, "oracle_sweep"),
    "interval",
    hours=1,
    id="oracle_sweep",
    max_instances=1,
    misfire_grace_time=300,
)

```

### Cron-Based Scheduling

For fixed-time operations, the backend employs cron expressions. Daily jobs—such as UAP sightings aggregation and wastewater sampling—execute at 12:00 UTC using cron triggers with optional jitter to prevent thundering herd issues against external APIs.

## Task Health Monitoring and Error Isolation

Rather than invoking fetcher functions directly, the scheduler wraps every job in `_run_task_with_health()`. This wrapper, defined in [`backend/services/data_fetcher.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/services/data_fetcher.py), records execution metrics, logs slow tasks, and catches exceptions to prevent individual job failures from crashing the scheduler thread.

```python
def _run_task_with_health(func, name: str | None = None):
    task_name = name or getattr(func, "__name__", "task")
    start = time.perf_counter()
    try:
        func()
        # record_success / log slow tasks …

    except Exception as e:
        # record_failure …

```

This pattern ensures that the BackgroundScheduler remains stable even when external data sources timeout or return malformed payloads.

## Concurrency Control and Resource Management

To avoid thread explosion while maintaining parallel throughput, the backend utilizes a shared **ThreadPoolExecutor**:

```python
_SHARED_EXECUTOR = concurrent.futures.ThreadPoolExecutor(
    max_workers=max(2, _FETCH_WORKERS), thread_name_prefix="fetch"
)

```

While the executor handles concurrent task dispatch, individual jobs enforce their own concurrency limits. For example, slow-tier enrichment jobs (news, weather) respect `_SLOW_FETCH_CONCURRENCY` limits to avoid overwhelming third-party rate limits.

## Tiered Job Categories

The Shadowbroker backend organizes background jobs into distinct tiers based on data velocity and criticality:

- **Fast-tier** – Executes every 60 seconds for high-velocity data (flights, ships, satellites).
- **Slow-tier** – Runs every 5 minutes for enrichment data (news, weather, CCTV metadata).
- **Time-critical** – Ukraine air-raid alerts every 2 minutes; weather alerts every 5 minutes.
- **Daily/periodic** – UAP sightings, wastewater sampling, and mesh map refreshes scheduled via cron.

## Graceful Startup and Cache Seeding

In [`backend/main.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/main.py), the application calls `seed_startup_caches()` before `start_scheduler()` to hydrate in-memory caches from disk. This ensures the API can serve stale-but-consistent data immediately while background jobs populate fresh data. Certain resource-intensive jobs—such as CCTV ingest—are delayed by `_STARTUP_CCTV_INGEST_DELAY_S` seconds to prioritize application readiness.

## Extending the Scheduler: Adding New Jobs

To register a new periodic task in the Shadowbroker backend:

1. Implement the fetcher function in `backend/services/fetchers/`.
2. Import the function into [`backend/services/data_fetcher.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/services/data_fetcher.py).
3. Add a job registration inside `start_scheduler()` using the health wrapper:

```python
_scheduler.add_job(
    lambda: _run_task_with_health(fetch_my_feature, "fetch_my_feature"),
    "interval",
    minutes=10,
    id="my_feature",
    max_instances=1,
    misfire_grace_time=120,
)

```

For one-off executions outside the scheduler loop, developers can invoke the task directly:

```python
from services.data_fetcher import _run_task_with_health, fetch_my_feature
_run_task_with_health(fetch_my_feature, "fetch_my_feature")

```

## Summary

- **APScheduler Backend** – The Shadowbroker backend relies on a daemon-mode `BackgroundScheduler` from APScheduler to manage all asynchronous work.
- **Dual Trigger Support** – Jobs use either `"interval"` triggers for recurring deltas or `"cron"` triggers for fixed-time execution.
- **Defensive Wrappers** – Every scheduled task runs through `_run_task_with_health()`, which isolates exceptions and records performance metrics.
- **Controlled Concurrency** – A shared `ThreadPoolExecutor` limits total worker threads, while `max_instances=1` prevents individual jobs from overlapping.
- **Tiered Architecture** – The system separates jobs into fast-tier (60s), slow-tier (5min), and daily cron categories to optimize resource allocation.

## Frequently Asked Questions

### What Python library handles background jobs in Shadowbroker?

The backend uses **APScheduler** (Advanced Python Scheduler) to manage background jobs and scheduled tasks. Specifically, it instantiates a `BackgroundScheduler` in daemon mode within [`backend/services/data_fetcher.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/services/data_fetcher.py) to ensure jobs run in background threads without blocking the main application lifecycle.

### How does Shadowbroker prevent scheduled tasks from overlapping?

Each job registration includes `max_instances=1`, ensuring only one instance of a specific job can run at any given time. Additionally, the `misfire_grace_time` parameter defines how many seconds a job may be late before it is skipped entirely, preventing queue buildup during outages.

### How can I add a new periodic data fetcher to the scheduler?

Create your fetcher function in `backend/services/fetchers/`, then register it in `start_scheduler()` within [`backend/services/data_fetcher.py`](https://github.com/BigBodyCobain/Shadowbroker/blob/main/backend/services/data_fetcher.py) using `_scheduler.add_job()`. Wrap the function call with `_run_task_with_health()` to automatically gain error handling and execution logging without additional boilerplate.

### What happens if a background job fails or raises an exception?

The `_run_task_with_health()` wrapper catches all exceptions, logs the failure via the internal health monitoring system, and allows the scheduler to continue running. This design prevents transient API failures from crashing the entire BackgroundScheduler thread or affecting other concurrent jobs.