How Shadowbroker Backend Manages Background Jobs and Scheduled Tasks

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 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.

# 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:

_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:

_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, records execution metrics, logs slow tasks, and catches exceptions to prevent individual job failures from crashing the scheduler thread.

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:

_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, 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.
3. Add a job registration inside start_scheduler() using the health wrapper:

_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:

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 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}