Scheduler Service

Relevant source files

The following files were used as context for generating this wiki page:

• monitor/scheduler.py
• monitor/main.py
• monitor/config.py
• README.md
• test_feeds.py
• WARP.md

The Scheduler Service is the core component responsible for orchestrating the periodic monitoring of technical blog feeds. It is built upon the APScheduler library and integrated into the application's asynchronous lifecycle. Its primary function is to trigger feed processing jobs at configurable intervals for each specified blog source, acting as the "heartbeat" of the entire monitoring daemon.

The service is initialized within the main AppContext and handles scheduling, executing, and managing the lifecycle of all feed monitoring tasks. It ensures that feeds are checked regularly for new content, which is then passed down the processing pipeline.

Sources: monitor/main.py:151-177, README.md:5-6, WARP.md:52-53

Architecture and Initialization

The scheduler is an instance of apscheduler.schedulers.asyncio.AsyncIOScheduler, which is designed to work within an asyncio event loop. It is initialized and started as part of the application's startup sequence, managed by the AppContext class.

Sources: monitor/main.py:33, 151

Initialization Flow

The AppContext._init_scheduler method handles the creation and configuration of the scheduler instance. It configures job stores, sets default job parameters, and starts the scheduler. Once initialized, the _schedule_jobs method is called to populate it with the feed monitoring tasks.

Sources: monitor/main.py:151-177

The following diagram illustrates the scheduler initialization process within the application context.

Upon application shutdown, the AppContext.shutdown method ensures the scheduler is stopped gracefully.

Sources: monitor/main.py:201-203

Configuration

The scheduler's behavior is configured through the SchedulerConfig and FeedConfig models defined in monitor/config.py. These settings are loaded from environment variables or a .env file.

Sources: monitor/config.py:202-209, 39-44

SchedulerConfig

This class defines the core settings for the APScheduler instance.

Parameter	Type	Default	Description
job_store_type	str	"memory"	The type of job store to use (e.g., memory, redis).
job_store_url	Optional[str]	None	The URL for the job store (e.g., Redis connection string).
max_instances	int	1	The maximum number of concurrent instances for each job.
timezone	str	"UTC"	The timezone for the scheduler.
coalesce	bool	True	Whether to run a job only once if it was missed multiple times.
misfire_grace_time	int	60	Seconds a job can be late before it is considered "misfired".

Sources: monitor/config.py:202-209

FeedConfig

The check_interval_minutes from FeedConfig is used to determine how often each feed monitoring job should run.

Parameter	Type	Default	Description
check_interval_minutes	int	60	The interval in minutes for polling the feed.

Sources: monitor/config.py:41

Job Scheduling

Jobs are dynamically scheduled based on the list of feeds provided in the application settings. The monitor.scheduler.schedule_feed_jobs function is responsible for this process.

Sources: monitor/main.py:173-175, monitor/scheduler.py

Scheduling Logic

The schedule_feed_jobs function iterates through each FeedConfig. If a feed is enabled, it adds a new job to the scheduler. The job is configured to run at a fixed interval specified by check_interval_minutes. The target of the job is the process_feed function in monitor/main.py, with the feed's name passed as an argument.

The flow is as follows:

Job Execution

When a scheduled job is triggered, it executes the process_feed async function located in monitor/main.py. This function orchestrates the entire pipeline for a single feed, from fetching new posts to storing their embeddings.

Sources: monitor/main.py:233-238, monitor/scheduler.py:25

Execution Sequence

The following sequence diagram shows the interactions when a process_feed job runs.

The key steps within process_feed are:

1. Retrieve the FeedConfig for the given feed name.
2. Invoke process_feed_posts to discover new articles by checking against the cache.
3. If full_content_capture is enabled, concurrently process each new article using process_individual_article to render the page and extract full content.
4. For each new post, create a concurrent process_post task.
5. Each process_post task handles rendering, content extraction, embedding generation, and finally upserting the EmbeddingRecord into the vector database.
6. Concurrency is managed using an asyncio.Semaphore to limit the number of simultaneous post-processing tasks.

Sources: monitor/main.py:255-310

Summary

The Scheduler Service is the automated trigger for the entire data ingestion pipeline. It leverages APScheduler to run jobs at user-defined intervals, ensuring that the system stays up-to-date with the latest posts from monitored technical blogs. Its configuration is flexible, allowing for different job stores and fine-grained control over timing and execution, while its integration with the application's async context ensures robust and graceful operation.

Sources: monitor/main.py:151-177, 447-458, monitor/config.py:202-209