Distributed Task Scheduler System Design

A distributed scheduler enables reliable task execution across multiple nodes with automatic failover and load balancing capabilities for large-scale systems.

1. Problem Statement & Requirements

Clarifying questions

1. What scale and throughput requirements are expected?

   • How many scheduled jobs per second can we expect to handle?
   • What is the maximum number of concurrent active jobs?

2. What consistency and reliability guarantees are needed?

   • Should job execution be exactly-once or at-least-once?
   • What is the expected tolerance for job loss or duplication?

3. What are the key entities and their relationships?

   • Are jobs tied to specific user or tenant contexts?
   • Do jobs have dependencies or need to be scheduled in a specific order?

4. What is the read/write ratio and latency sensitivity?

   • Is the system primarily used for scheduling (write-heavy) or querying (read-heavy)?
   • What are the acceptable latencies for job creation, retrieval, and execution?

5. What authentication and authorization model is in place?

   • Will jobs be scheduled by authenticated users or system services?
   • Are there role-based access controls for job management?

6. Are there external integrations or systems that jobs interact with?

   • Do jobs trigger external API calls, database writes, or other services?
   • Is there a requirement for job status updates to external systems?

7. What are the out-of-scope boundaries and constraints?

   • Should the system support job cancellation or modification after creation?
   • Is there a requirement for job history or audit logs?

Functional requirements

• Job Creation: Users can schedule jobs with a specific time or cron expression.
• Job Execution: Scheduled jobs are executed at their designated time or according to a cron schedule.
• Job Retrieval: Users can query and retrieve job details (status, schedule, payload).
• Job Management: Users can cancel or update scheduled jobs.
• Job Status Tracking: Real-time status updates for job execution (pending, running, completed, failed).
• Job Retry Logic: Failed jobs can be retried based on configured policies.
• Job Dependency Management: Jobs may have dependencies or need to be scheduled in a specific order.

Non-functional requirements

• Latency Targets:

  • p50: < 10ms for job creation and retrieval
  • p95: < 50ms for job creation and retrieval
  • p99: < 200ms for job creation and retrieval

• Throughput:

  • Support 10,000 job creations per second
  • Handle 5,000 concurrent job executions

• Availability:

  • 99.9% uptime for job scheduling and execution
  • Support for multi-region deployment with failover capability

Constraints and assumptions

• Assumptions:

  • Jobs are scheduled with a fixed time or cron expression.
  • The system is designed for horizontal scaling and distributed execution.
  • Jobs are idempotent or support idempotent execution patterns.
  • External systems (e.g., job executors) are assumed to be available and responsive.

• Constraints:

  • Jobs must not be lost under any circumstances (strong consistency requirement).
  • The system must support job scheduling across multiple time zones.
  • Job payloads are limited to 1MB in size.

Out of scope

• Job Scheduling Across Multiple Clusters: Cross-cluster scheduling and coordination are out of scope.
• Complex Workflow Orchestration: Full workflow engine features like DAGs or job chaining are not part of this design.
• Real-Time Job Monitoring: Live dashboards or real-time job status updates via webhooks are not included.
• Job Versioning: Version control for job definitions or payloads is not part of this design.
• External Job Dependencies: Integration with external job queues or task runners (e.g., Celery, Kafka) is out of scope.

2. API Design

API Style Selection

gRPC with Protocol Buffers is selected for the distributed scheduler system due to:

• Strong typing and schema evolution support
• Efficient binary serialization with compression
• Built-in support for streaming and bidirectional communication
• Native support for versioning through proto file management
• Better performance characteristics for inter-service communication in distributed systems

Endpoint Definitions

Method	Path	Description	Auth	Rate Limit
CreateSchedule	/scheduler.v1.SchedulerService/CreateSchedule	Creates a new scheduled job	JWT + RBAC	100 req/min
GetSchedule	/scheduler.v1.SchedulerService/GetSchedule	Retrieves schedule by ID	JWT + RBAC	500 req/min
UpdateSchedule	/scheduler.v1.SchedulerService/UpdateSchedule	Updates existing schedule	JWT + RBAC	50 req/min
DeleteSchedule	/scheduler.v1.SchedulerService/DeleteSchedule	Deletes schedule by ID	JWT + RBAC	25 req/min
ListSchedules	/scheduler.v1.SchedulerService/ListSchedules	Lists schedules with pagination	JWT + RBAC	200 req/min
TriggerSchedule	/scheduler.v1.SchedulerService/TriggerSchedule	Manually triggers schedule execution	JWT + RBAC	10 req/min
GetScheduleHistory	/scheduler.v1.SchedulerService/GetScheduleHistory	Retrieves execution history	JWT + RBAC	100 req/min

Request and Response Schemas

CreateScheduleRequest

{
  "name": "string",
  "description": "string",
  "cron_expression": "string",
  "target_service": "string",
  "target_endpoint": "string",
  "payload": "object",
  "enabled": "boolean",
  "timezone": "string",
  "retry_config": {
    "max_attempts": 3,
    "backoff_seconds": 10
  }
}

CreateScheduleResponse

{
  "schedule_id": "uuid",
  "created_at": "timestamp"
}

Schedule

{
  "id": "uuid",
  "name": "string",
  "description": "string",
  "cron_expression": "string",
  "target_service": "string",
  "target_endpoint": "string",
  "payload": "object",
  "enabled": "boolean",
  "timezone": "string",
  "retry_config": {
    "max_attempts": 3,
    "backoff_seconds": 10
  },
  "created_at": "timestamp",
  "updated_at": "timestamp",
  "next_run_time": "timestamp"
}

HTTP Status Codes and Error Response Format

Standard HTTP status codes with custom error responses:

• 200 OK - Successful operation
• 201 Created - Resource created successfully
• 400 Bad Request - Invalid request parameters
• 401 Unauthorized - Authentication required
• 403 Forbidden - Insufficient permissions
• 404 Not Found - Resource not found
• 429 Too Many Requests - Rate limit exceeded
• 500 Internal Server Error - Unexpected server error

Standardized Error Response Format

{
  "error": {
    "code": "SCHEDULE_NOT_FOUND",
    "message": "Schedule with ID abc-123 not found",
    "details": {
      "schedule_id": "abc-123"
    },
    "timestamp": "2023-10-15T14:30:00Z"
  }
}

API Versioning Strategy

Versioned using gRPC package naming convention:

• scheduler.v1.SchedulerService - Current version
• Future: scheduler.v2.SchedulerService for breaking changes

Pagination Design

Cursor-based pagination is used because:

• More efficient than offset-based for large datasets
• Consistent results even when data changes during pagination
• Handles duplicate records better

ListSchedulesRequest

{
  "limit": 50,
  "cursor": "string",
  "filter": {
    "enabled": true,
    "service": "payment-service"
  }
}

ListSchedulesResponse

{
  "schedules": [Schedule],
  "next_cursor": "string",
  "has_more": true
}

Authentication and Authorization

Authentication: JWT tokens with OAuth2 integration for service-to-service authentication.

Authorization: Role-based access control (RBAC) with:

• SCHEDULER_ADMIN - Full CRUD permissions
• SCHEDULER_USER - Read and trigger permissions
• SCHEDULER_VIEWER - Read-only permissions

Rate Limiting Strategy

Per-endpoint rate limits:

• Create/Update/Delete: 100 req/min
• List/Get: 500 req/min
• Trigger: 10 req/min

Implementation: Token bucket algorithm with Redis-based distributed rate limiting.

Idempotency Design

Idempotency Key Strategy:

• Clients provide X-Idempotency-Key header
• Server stores operation result by key for 24 hours
• Duplicate requests with same key return cached result
• Used primarily for CreateSchedule and TriggerSchedule

// Example implementation in service layer
public CreateScheduleResponse createSchedule(CreateScheduleRequest request, String idempotencyKey) {
    if (idempotencyKey != null) {
        CreateScheduleResponse cached = idempotencyCache.get(idempotencyKey);
        if (cached != null) {
            return cached;
        }
    }
    
    // Process request...
    CreateScheduleResponse response = processRequest(request);
    
    if (idempotencyKey != null) {
        idempotencyCache.put(idempotencyKey, response, Duration.ofHours(24));
    }
    
    return response;
}

3. Data Model Design

Entity-Relationship Diagram

erDiagram
    SCHEDULED_JOB {
        string id PK
        string name
        string description
        string cron_expression
        string job_type
        string status
        string owner_id
        string payload
        timestamp scheduled_at
        timestamp executed_at
        timestamp created_at
        timestamp updated_at
        boolean is_active
        integer retry_count
        integer max_retries
        string error_message
        string execution_result
    }

    JOB_EXECUTION {
        string id PK
        string job_id FK
        string status
        timestamp started_at
        timestamp completed_at
        string result
        string error_message
        integer attempt_number
        timestamp created_at
    }

    JOB_DEPENDENCY {
        string id PK
        string job_id FK
        string depends_on_job_id FK
        string dependency_type
        timestamp created_at
    }

    SCHEDULED_JOB ||--o{ JOB_EXECUTION : "has"
    SCHEDULED_JOB ||--o{ JOB_DEPENDENCY : "depends_on"
    JOB_DEPENDENCY ||--o{ SCHEDULED_JOB : "dependent"

Table Schema Definitions

-- Scheduled jobs table
CREATE TABLE scheduled_jobs (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    name VARCHAR(255) NOT NULL,
    description TEXT,
    cron_expression VARCHAR(100) NOT NULL,
    job_type VARCHAR(100) NOT NULL,
    status VARCHAR(50) NOT NULL DEFAULT 'PENDING',
    owner_id VARCHAR(255) NOT NULL,
    payload JSONB,
    scheduled_at TIMESTAMP WITH TIME ZONE NOT NULL,
    executed_at TIMESTAMP WITH TIME ZONE,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    is_active BOOLEAN NOT NULL DEFAULT TRUE,
    retry_count INTEGER NOT NULL DEFAULT 0,
    max_retries INTEGER NOT NULL DEFAULT 3,
    error_message TEXT,
    execution_result JSONB,
    
    -- Indexes
    INDEX idx_scheduled_jobs_owner_id (owner_id),
    INDEX idx_scheduled_jobs_status (status),
    INDEX idx_scheduled_jobs_scheduled_at (scheduled_at),
    INDEX idx_scheduled_jobs_active_status (is_active, status),
    INDEX idx_scheduled_jobs_cron_expression (cron_expression)
);

-- Job execution history table
CREATE TABLE job_executions (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    job_id UUID NOT NULL REFERENCES scheduled_jobs(id) ON DELETE CASCADE,
    status VARCHAR(50) NOT NULL DEFAULT 'PENDING',
    started_at TIMESTAMP WITH TIME ZONE,
    completed_at TIMESTAMP WITH TIME ZONE,
    result JSONB,
    error_message TEXT,
    attempt_number INTEGER NOT NULL DEFAULT 1,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    
    -- Indexes
    INDEX idx_job_executions_job_id (job_id),
    INDEX idx_job_executions_status (status),
    INDEX idx_job_executions_started_at (started_at),
    INDEX idx_job_executions_attempt_number (attempt_number)
);

-- Job dependency relationships
CREATE TABLE job_dependencies (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    job_id UUID NOT NULL REFERENCES scheduled_jobs(id) ON DELETE CASCADE,
    depends_on_job_id UUID NOT NULL REFERENCES scheduled_jobs(id) ON DELETE CASCADE,
    dependency_type VARCHAR(50) NOT NULL DEFAULT 'SEQUENTIAL',
    created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
    
    -- Indexes
    INDEX idx_job_dependencies_job_id (job_id),
    INDEX idx_job_dependencies_depends_on_job_id (depends_on_job_id),
    UNIQUE INDEX idx_unique_dependency (job_id, depends_on_job_id)
);

Primary Key Strategy

UUID as Primary Key - Selected for the following reasons:

1. Distributed System Compatibility: UUIDs are globally unique, eliminating conflicts in distributed deployments
2. No Centralized ID Generation: Avoids single point of failure in ID generation services
3. Sharding Friendliness: UUIDs can be evenly distributed across shards without requiring coordination
4. Audit Trail Integrity: UUIDs provide immutable identifiers that don't change even if job metadata changes

Index Design Justification

Index	Fields	Type	Purpose
idx_scheduled_jobs_owner_id	owner_id	Standard	Fast job lookup by owner
idx_scheduled_jobs_status	status	Standard	Efficient state-based queries
idx_scheduled_jobs_scheduled_at	scheduled_at	Standard	Fast scheduling queue operations
idx_scheduled_jobs_active_status	is_active, status	Composite	Fast job queue filtering
idx_scheduled_jobs_cron_expression	cron_expression	Standard	Efficient cron-based job filtering
idx_job_executions_job_id	job_id	Standard	Fast execution history lookup
idx_job_executions_status	status	Standard	Efficient status-based filtering
idx_job_executions_started_at	started_at	Standard	Fast execution time range queries
idx_job_executions_attempt_number	attempt_number	Standard	Efficient retry tracking
idx_job_dependencies_job_id	job_id	Standard	Fast dependency lookups
idx_job_dependencies_depends_on_job_id	depends_on_job_id	Standard	Fast reverse dependency queries

Data Partitioning Strategy

Time-based Partitioning by scheduled_at:

• Partition Granularity: Daily partitions
• Partitioning Key: DATE(scheduled_at)
• Rationale:
  • Jobs are typically scheduled for future execution, making time-based partitioning ideal
  • Historical job data can be archived to separate storage (e.g., S3) while keeping recent jobs in DB
  • Reduces query scan size for recent job lookups

Soft Delete vs. Hard Delete

Soft Delete Selected:

1. Audit Trail Requirements: Need to maintain job history for compliance and debugging
2. Recovery Capability: Ability to restore accidentally deleted jobs
3. Historical Analysis: Job execution history must persist for analytics and reporting
4. Dependency Management: Deleting jobs would break dependencies that may still be referenced

Audit Trail Design

Audit Fields in scheduled_jobs table:

• created_at: Timestamp when job was created
• updated_at: Timestamp when job was last modified
• owner_id: Identifier of the user or system that created the job

Additional Audit Table (for detailed audit logs):

CREATE TABLE job_audit_log (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    job_id UUID NOT NULL REFERENCES scheduled_jobs(id),
    action VARCHAR(50) NOT NULL, -- CREATE, UPDATE, DELETE, EXECUTE
    actor_id VARCHAR(255) NOT NULL,
    old_values JSONB,
    new_values JSONB,
    timestamp TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW()
);

Data Archival Strategy

Archival Policy:

1. Active Jobs: Retention period of 30 days (jobs that have been scheduled but not yet executed)
2. Completed Jobs: Retention period of 90 days (jobs that have been executed successfully)
3. Failed Jobs: Retention period of 7 days (jobs that have failed and are awaiting manual intervention)
4. Archival Process:
   • Daily batch job to move completed jobs older than 90 days to S3
   • Failed jobs older than 7 days are marked as archived and can be purged after 30 days
   • Archive metadata stored in separate table with reference to S3 location

Retention Rules:

• Jobs with is_active = FALSE and executed_at < NOW() - 90 days are archived
• Jobs with status = 'FAILED' and executed_at < NOW() - 7 days are marked for archival

4. Class Design & Service Layer

UML Class Diagram

classDiagram
    class SchedulerService {
        +scheduleJob(JobRequest)
        +cancelJob(String jobId)
        +getJobStatus(String jobId)
        +rescheduleJob(String jobId, LocalDateTime newScheduleTime)
    }
    
    class JobRepository {
        +save(Job job)
        +findById(String jobId)
        +findByStatusAndScheduledTimeBefore(JobStatus status, LocalDateTime now)
        +updateJobStatus(String jobId, JobStatus status)
        +deleteById(String jobId)
    }
    
    class JobScheduler {
        +scheduleJob(Job job)
        +processScheduledJobs()
        +handleJobExecution(String jobId)
    }
    
    class Job {
        -jobId: String
        -name: String
        -description: String
        -status: JobStatus
        -scheduledTime: LocalDateTime
        -cronExpression: String
        -payload: Map~String, Object~
        -retryCount: int
        -maxRetries: int
        -createdAt: LocalDateTime
        -updatedAt: LocalDateTime
        +validate()
    }
    
    class JobStatus {
        PENDING
        RUNNING
        COMPLETED
        FAILED
        CANCELLED
    }
    
    class JobRequest {
        -name: String
        -description: String
        -scheduledTime: LocalDateTime
        -cronExpression: String
        -payload: Map~String, Object~
        -maxRetries: int
    }
    
    class JobExecutionService {
        +executeJob(String jobId)
        +handleJobFailure(String jobId, Exception error)
        +retryJob(String jobId)
    }
    
    class EventPublisher {
        +publishJobEvent(JobEvent event)
    }
    
    class JobEvent {
        -eventType: String
        -jobId: String
        -status: JobStatus
        -timestamp: LocalDateTime
    }
    
    class JobProcessor {
        +processJob(String jobId)
        +handleExecutionResult(String jobId, ExecutionResult result)
    }
    
    class ExecutionResult {
        -success: boolean
        -output: String
        -error: String
    }
    
    SchedulerService --> JobRepository : uses
    SchedulerService --> JobScheduler : delegates to
    JobScheduler --> JobRepository : persists
    JobScheduler --> JobExecutionService : executes
    JobScheduler --> EventPublisher : publishes events
    JobExecutionService --> JobProcessor : processes
    JobProcessor --> JobRepository : updates status
    
    class JobSchedulerWorker {
        -scheduler: JobScheduler
        -jobProcessor: JobProcessor
        +run()
    }
    
    JobSchedulerWorker --> JobScheduler : polls

State Machine Diagram for Job Entity

stateDiagram-v2
    [*] --> PENDING
    PENDING --> RUNNING : Scheduled
    RUNNING --> COMPLETED : Success
    RUNNING --> FAILED : Failure
    RUNNING --> PENDING : Retry
    FAILED --> PENDING : Retry
    PENDING --> CANCELLED : Cancelled
    RUNNING --> CANCELLED : Cancelled

Service Class Structure and Responsibilities

The service layer is organized around the core domain entities:

1. SchedulerService - Main entry point for job scheduling operations
2. JobScheduler - Core scheduler logic that manages job execution timing and processing
3. JobExecutionService - Handles actual job execution and retry logic
4. JobProcessor - Processes job execution results and updates state

Interface / Abstract Class Design

public interface JobRepository {
    Job save(Job job);
    Optional<Job> findById(String jobId);
    List<Job> findByStatusAndScheduledTimeBefore(JobStatus status, LocalDateTime now);
    void updateJobStatus(String jobId, JobStatus status);
    void deleteById(String jobId);
}

public interface EventPublisher {
    void publishJobEvent(JobEvent event);
}

public enum JobStatus {
    PENDING, RUNNING, COMPLETED, FAILED, CANCELLED
}

public class JobRequest {
    private String name;
    private String description;
    private LocalDateTime scheduledTime;
    private String cronExpression;
    private Map<String, Object> payload;
    private int maxRetries = 3;
    
    // Getters and setters
}

Business Logic Encapsulation Patterns

Strategy Pattern - For handling different job execution strategies:

public interface JobExecutionStrategy {
    ExecutionResult execute(Job job);
}

@Component
public class HttpJobExecutionStrategy implements JobExecutionStrategy {
    private final RestTemplate restTemplate;
    
    @Override
    public ExecutionResult execute(Job job) {
        try {
            // Execute HTTP request based on job payload
            ResponseEntity<String> response = restTemplate.postForEntity(
                (String) job.getPayload().get("url"),
                job.getPayload().get("body"),
                String.class
            );
            
            return new ExecutionResult(true, response.getBody(), null);
        } catch (Exception e) {
            return new ExecutionResult(false, null, e.getMessage());
        }
    }
}

@Component
public class TaskJobExecutionStrategy implements JobExecutionStrategy {
    private final TaskExecutor taskExecutor;
    
    @Override
    public ExecutionResult execute(Job job) {
        try {
            // Execute task in thread pool
            Future<String> future = taskExecutor.submit(() -> {
                // Task logic here
                return "Task completed";
            });
            
            String result = future.get(30, TimeUnit.SECONDS);
            return new ExecutionResult(true, result, null);
        } catch (Exception e) {
            return new ExecutionResult(false, null, e.getMessage());
        }
    }
}

Critical-Path Method Implementation

@Transactional
public Job scheduleJob(JobRequest request) {
    // Validate input
    if (request.getScheduledTime() == null && request.getCronExpression() == null) {
        throw new IllegalArgumentException("Either scheduledTime or cronExpression must be provided");
    }
    
    // Create job entity
    Job job = new Job();
    job.setJobId(UUID.randomUUID().toString());
    job.setName(request.getName());
    job.setDescription(request.getDescription());
    job.setStatus(JobStatus.PENDING);
    job.setScheduledTime(request.getScheduledTime());
    job.setCronExpression(request.getCronExpression());
    job.setPayload(request.getPayload());
    job.setMaxRetries(request.getMaxRetries());
    job.setCreatedAt(LocalDateTime.now());
    job.setUpdatedAt(LocalDateTime.now());
    
    // Validate job
    job.validate();
    
    // Save to database
    Job savedJob = jobRepository.save(job);
    
    // Publish event for monitoring and audit
    eventPublisher.publishJobEvent(new JobEvent(
        "JOB_SCHEDULED",
        savedJob.getJobId(),
        JobStatus.PENDING,
        LocalDateTime.now()
    ));
    
    return savedJob;
}

Transaction Boundary Design

Transaction boundaries are defined per operation:

1. Schedule Job: Single transaction for job creation and persistence
2. Cancel Job: Transaction for status update and cleanup
3. Process Scheduled Jobs: Per-job transaction to ensure atomicity of job execution and state update

Service Interface Contracts

public interface SchedulerService {
    Job scheduleJob(JobRequest request);
    void cancelJob(String jobId);
    JobStatus getJobStatus(String jobId);
    Job rescheduleJob(String jobId, LocalDateTime newScheduleTime);
}

Saga / Orchestration Pattern

For distributed job execution involving external systems, we implement a saga pattern:

public class DistributedJobSaga {
    private final List<SagaStep> steps;
    
    public void execute(Job job) {
        for (int i = 0; i < steps.size(); i++) {
            try {
                steps.get(i).execute(job);
            } catch (Exception e) {
                // Compensate previous steps
                for (int j = i - 1; j >= 0; j--) {
                    steps.get(j).compensate(job);
                }
                throw new JobExecutionException("Saga failed", e);
            }
        }
    }
}

Domain Object / Value Object Design

public class Job {
    private String jobId;
    private String name;
    private String description;
    private JobStatus status;
    private LocalDateTime scheduledTime;
    private String cronExpression;
    private Map<String, Object> payload;
    private int retryCount;
    private int maxRetries;
    private LocalDateTime createdAt;
    private LocalDateTime updatedAt;
    
    public void validate() {
        if (name == null || name.trim().isEmpty()) {
            throw new IllegalArgumentException("Job name is required");
        }
        
        if (scheduledTime == null && cronExpression == null) {
            throw new IllegalArgumentException("Either scheduled time or cron expression must be provided");
        }
        
        if (payload == null) {
            payload = new HashMap<>();
        }
    }
}

public class ExecutionResult {
    private boolean success;
    private String output;
    private String error;
    
    public ExecutionResult(boolean success, String output, String error) {
        this.success = success;
        this.output = output;
        this.error = error;
    }
}

5. Concurrency & Thread Safety

Concurrency Model

The distributed scheduler will use a thread pool model with fixed-size thread pools for concurrent job execution. Each worker thread in the scheduler cluster will handle one job at a time, with a configurable thread pool size per node. This approach provides:

• Predictable resource usage: Fixed thread pool size prevents unbounded resource consumption
• Isolation: Each job executes in its own thread context, preventing interference between jobs
• Scalability: Multiple nodes can process jobs in parallel

Race Condition Prevention Strategy

Race conditions will be prevented through:

1. Database-level atomic operations for job state transitions and updates
2. Optimistic locking with version fields to detect concurrent modifications
3. Distributed locks for critical sections using Redis or database advisory locks
4. Transaction boundaries to ensure atomicity of job state changes

Optimistic vs. Pessimistic Locking Selection

Optimistic locking is selected for the scheduler system because:

• High concurrency: Multiple nodes can process jobs simultaneously without blocking
• Low contention: Job state updates are infrequent and typically don't conflict
• Better performance: Avoids blocking and waiting for locks
• Scalability: Scales better with increasing job volume

Implementation pattern:

// Pessimistic locking would require explicit lock acquisition
// Optimistic locking uses version field in database
@Version
private Long version;

Deadlock Prevention Strategy

Deadlock prevention is achieved through:

1. Consistent lock ordering: All nodes acquire locks in the same order (e.g., by job ID)
2. Timeout-based lock acquisition: Maximum wait time for locks (500ms)
3. Lock hierarchy: Jobs are processed in a consistent order to avoid circular dependencies
4. Transaction timeout: Database transaction timeouts (30 seconds)

Idempotency Implementation for Concurrent Duplicate Requests

Idempotency is implemented using:

1. Job IDempotency Keys: Unique identifiers that prevent duplicate job execution
2. Database uniqueness constraints on job IDs and execution keys
3. State tracking with timestamps: Prevents reprocessing of completed jobs

// Example idempotency check in job execution
public boolean executeJobWithIdempotency(String jobId, String executionKey) {
    // Check if job already completed
    if (jobRepository.isJobCompleted(jobId)) {
        return true;
    }
    
    // Check for duplicate execution
    if (jobRepository.isExecutionDuplicate(executionKey)) {
        return true; // Already processed
    }
    
    try {
        // Acquire lock with timeout
        if (!lockService.acquireLock(jobId, 500)) {
            throw new JobExecutionException("Unable to acquire lock for job: " + jobId);
        }
        
        // Execute job logic
        executeJob(jobId);
        
        // Mark as completed
        jobRepository.markJobCompleted(jobId, executionKey);
        
    } finally {
        lockService.releaseLock(jobId);
    }
    
    return true;
}

Thread Safety Patterns

Thread safety is ensured through:

1. Immutable job configuration objects to prevent modification during execution
2. Thread-local storage for job-specific context (e.g., trace IDs)
3. Synchronized collections where needed for shared state
4. Atomic reference updates for job status counters

Concurrency Control in Job Scheduling Loop

The main scheduling loop uses a thread-safe approach:

// Pseudo-code for scheduler loop
public class JobScheduler {
    private final ScheduledExecutorService executor = 
        Executors.newScheduledThreadPool(10);
    
    @Scheduled(fixedRate = 5000) // Every 5 seconds
    public void processScheduledJobs() {
        List<Job> readyJobs = jobRepository.findReadyJobs();
        
        // Process jobs in parallel with thread pool
        readyJobs.parallelStream().forEach(job -> {
            executor.submit(() -> {
                try {
                    executeJob(job);
                } catch (Exception e) {
                    handleJobFailure(job, e);
                }
            });
        });
    }
}

Memory Consistency Model

The system uses:

1. Happens-before relationships to ensure proper ordering of job state updates
2. Volatile fields for shared job status flags that need to be visible across threads
3. Thread-safe collections (CopyOnWriteArrayList) for job monitoring lists

Lock Granularity Strategy

Locks are applied at the job level to minimize contention:

• Each job gets its own lock key (e.g., lock:job-{jobId})
• Locks are held only for the duration of job execution
• Distributed lock implementation uses Redis with TTL to prevent deadlocks

Thread Pool Configuration

// Thread pool configuration for job execution
@Configuration
public class JobThreadPoolConfig {
    
    @Bean("jobExecutor")
    public ExecutorService jobExecutor() {
        return new ThreadPoolExecutor(
            10,                    // corePoolSize
            50,                    // maximumPoolSize  
            60L,                   // keepAliveTime
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000), // work queue
            new ThreadPoolTaskExecutor.ThreadPoolTaskExecutor()
        );
    }
}

Data Consistency in Concurrent Operations

1. Database transactions for job state updates
2. Read-after-write consistency using database transaction isolation levels (READ_COMMITTED)
3. Eventual consistency model for job status propagation across nodes
4. Optimistic concurrency control with version-based conflict detection

Handling Concurrent Job Updates

When multiple nodes attempt to update the same job:

// Example optimistic locking implementation
public boolean updateJob(Job job) {
    int maxRetries = 3;
    int attempts = 0;
    
    while (attempts < maxRetries) {
        try {
            // Load current job with version
            Job currentJob = jobRepository.findById(job.getId());
            
            // Validate version matches (optimistic locking)
            if (!currentJob.getVersion().equals(job.getVersion())) {
                throw new ConcurrentModificationException(
                    "Job version mismatch: expected " + currentJob.getVersion() + 
                    ", got " + job.getVersion());
            }
            
            // Update job with incremented version
            job.setVersion(currentJob.getVersion() + 1);
            jobRepository.save(job);
            
            return true;
        } catch (ConcurrentModificationException e) {
            attempts++;
            if (attempts >= maxRetries) {
                throw new JobProcessingException("Failed to update job after " + maxRetries + " attempts", e);
            }
            
            // Brief backoff before retry
            Thread.sleep(100 * attempts);
        }
    }
    
    return false;
}

6. Sequence Diagrams for Key Flows

sequenceDiagram
    participant Client
    participant SchedulerService
    participant JobRepository
    participant TaskQueue
    participant WorkerPool
    participant JobExecutor

    Client->>SchedulerService: POST /jobs
    SchedulerService->>JobRepository: Save job metadata
    JobRepository-->>SchedulerService: Return job ID
    SchedulerService->>TaskQueue: Publish job to queue
    TaskQueue-->>SchedulerService: Acknowledge
    SchedulerService-->>Client: Return job ID and status
    TaskQueue->>WorkerPool: Dequeue job
    WorkerPool->>JobExecutor: Execute job
    JobExecutor->>JobRepository: Update job status to RUNNING
    JobExecutor-->>JobRepository: Update job status to COMPLETED/FAILED
    JobExecutor-->>WorkerPool: Acknowledge completion

sequenceDiagram
    participant Client
    participant SchedulerService
    participant JobRepository
    participant TaskQueue
    participant WorkerPool
    participant JobExecutor

    Client->>SchedulerService: POST /jobs with invalid payload
    SchedulerService-->>Client: 400 Bad Request
    Client->>SchedulerService: POST /jobs with valid payload
    SchedulerService->>JobRepository: Validate and save job
    JobRepository-->>SchedulerService: Return job ID
    SchedulerService->>TaskQueue: Publish job to queue with retry logic
    TaskQueue-->>SchedulerService: Acknowledge or retry on failure
    SchedulerService-->>Client: Return job ID and status
    TaskQueue->>WorkerPool: Dequeue job with timeout handling
    WorkerPool->>JobExecutor: Execute job with timeout
    JobExecutor->>JobRepository: Update job status to RUNNING
    JobExecutor-->>JobRepository: Update job status to COMPLETED/FAILED
    JobExecutor-->>WorkerPool: Acknowledge completion or retry on failure

sequenceDiagram
    participant SchedulerService
    participant JobRepository
    participant EventPublisher
    participant EventConsumer

    SchedulerService->>JobRepository: Update job status to COMPLETED
    JobRepository-->>SchedulerService: Confirm update
    SchedulerService->>EventPublisher: Publish job completion event
    EventPublisher-->>SchedulerService: Acknowledge event publish
    EventConsumer->>EventPublisher: Subscribe to job completion events
    EventConsumer-->>SchedulerService: Process event (e.g., trigger downstream jobs)

7. Caching Design

Cache Strategy Selection

For the distributed scheduler, we'll use a cache-aside pattern with LRU eviction policy. This approach provides the best balance between performance and consistency for a scheduler system where:

• Job metadata is frequently accessed but updated infrequently
• Cache misses should be handled gracefully with database fallback
• We need to support distributed cache across multiple scheduler instances

Cache Key Design

Cache keys will follow a structured namespace pattern:

scheduler:{jobId}:{type}
scheduler:jobStatus:{jobId}
scheduler:jobQueue:{queueName}
scheduler:config:{configKey}

TTL Configuration

Data Type	TTL	Reason
Job metadata	5 minutes	Frequent reads, infrequent updates
Job status	1 hour	Status changes are less frequent but need to be available
Queue state	30 seconds	High frequency reads for queue management
Configuration	1 hour	Configuration changes are infrequent but critical

Cache Eviction Policy

We'll implement LRU (Least Recently Used) eviction with:

• Size-based limits (e.g., 100MB max per node)
• Time-based expiration for stale data
• Cache warming during system startup

Cache Stampede Prevention

To prevent cache stampedes:

1. Probabilistic early expiry: Set cache TTL to 80% of configured TTL
2. Distributed locks: Use Redis distributed locks for cache population
3. Cache warming: Pre-populate hot keys during system startup

Distributed Cache Architecture

We'll use Redis Cluster with:

• 3 master nodes for high availability
• 3 replica nodes per master for read scaling
• Automatic failover with Redis Sentinel monitoring
• Cross-zone replication for geographic distribution

Cache Invalidation Strategy

Event-driven invalidation will be the primary strategy:

1. Job status changes trigger cache invalidation for that job
2. Queue updates invalidate queue state caches
3. Configuration changes invalidate config caches
4. TTL-based invalidation as fallback

Cache Implementation Example (Java)

@Component
public class SchedulerCache {
    private final RedisTemplate<String, Object> redisTemplate;
    private final CacheManager cacheManager;
    
    public JobMetadata getCachedJob(String jobId) {
        String key = "scheduler:" + jobId + ":metadata";
        JobMetadata cached = (JobMetadata) redisTemplate.opsForValue().get(key);
        
        if (cached != null) {
            return cached;
        }
        
        // Cache miss - acquire distributed lock
        String lockKey = "lock:" + key;
        if (tryAcquireLock(lockKey, 5000)) {
            try {
                // Double-check pattern
                cached = (JobMetadata) redisTemplate.opsForValue().get(key);
                if (cached != null) {
                    return cached;
                }
                
                // Load from database
                JobMetadata job = jobRepository.findById(jobId);
                if (job != null) {
                    redisTemplate.opsForValue().set(key, job, 300, TimeUnit.SECONDS);
                }
                return job;
            } finally {
                releaseLock(lockKey);
            }
        }
        
        // Fallback to database if lock acquisition fails
        return jobRepository.findById(jobId);
    }
    
    public void invalidateJobCache(String jobId) {
        String[] keys = {
            "scheduler:" + jobId + ":metadata",
            "scheduler:jobStatus:" + jobId,
            "scheduler:jobQueue:*"
        };
        
        for (String key : keys) {
            redisTemplate.delete(key);
        }
    }
}

8. Repository / Data Access Layer

Repository Pattern Implementation

The distributed scheduler follows the repository pattern to encapsulate data access logic and provide a clean abstraction layer between the service layer and database operations.

public interface JobRepository {
    Job save(Job job);
    Optional<Job> findById(UUID id);
    List<Job> findByStatus(JobStatus status, Pageable pageable);
    List<Job> findByScheduledTimeBefore(Instant now, Pageable pageable);
    void updateStatus(UUID id, JobStatus status);
    void updateLastRunTime(UUID id, Instant lastRunTime);
    void delete(UUID id);
    long countByStatus(JobStatus status);
}

public class JobRepositoryImpl implements JobRepository {
    private final JdbcTemplate jdbcTemplate;
    private final ObjectMapper objectMapper;
    
    public JobRepositoryImpl(JdbcTemplate jdbcTemplate, ObjectMapper objectMapper) {
        this.jdbcTemplate = jdbcTemplate;
        this.objectMapper = objectMapper;
    }
    
    @Override
    public Job save(Job job) {
        String sql = """
            INSERT INTO jobs (id, name, description, cron_expression, status, 
                             scheduled_time, last_run_time, next_run_time, 
                             job_data, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """;
            
        jdbcTemplate.update(sql, 
            job.getId(),
            job.getName(),
            job.getDescription(),
            job.getCronExpression(),
            job.getStatus().toString(),
            job.getScheduledTime(),
            job.getLastRunTime(),
            job.getNextRunTime(),
            objectMapper.writeValueAsString(job.getJobData()),
            job.getCreatedAt(),
            job.getUpdatedAt()
        );
        
        return job;
    }
    
    @Override
    public Optional<Job> findById(UUID id) {
        String sql = "SELECT * FROM jobs WHERE id = ?";
        try {
            return Optional.of(jdbcTemplate.queryForObject(sql, new JobRowMapper(), id));
        } catch (EmptyResultDataAccessException e) {
            return Optional.empty();
        }
    }
    
    @Override
    public List<Job> findByStatus(JobStatus status, Pageable pageable) {
        String sql = """
            SELECT * FROM jobs 
            WHERE status = ? 
            ORDER BY scheduled_time ASC 
            LIMIT ? OFFSET ?
            """;
            
        return jdbcTemplate.query(sql, new JobRowMapper(), 
            status.toString(), 
            pageable.getPageSize(), 
            pageable.getOffset());
    }
    
    @Override
    public List<Job> findByScheduledTimeBefore(Instant now, Pageable pageable) {
        String sql = """
            SELECT * FROM jobs 
            WHERE scheduled_time <= ? 
            AND status IN ('PENDING', 'SCHEDULED') 
            ORDER BY scheduled_time ASC 
            LIMIT ? OFFSET ?
            """;
            
        return jdbcTemplate.query(sql, new JobRowMapper(), 
            now, 
            pageable.getPageSize(), 
            pageable.getOffset());
    }
    
    @Override
    public void updateStatus(UUID id, JobStatus status) {
        String sql = "UPDATE jobs SET status = ?, updated_at = ? WHERE id = ?";
        jdbcTemplate.update(sql, status.toString(), Instant.now(), id);
    }
    
    @Override
    public void updateLastRunTime(UUID id, Instant lastRunTime) {
        String sql = "UPDATE jobs SET last_run_time = ?, updated_at = ? WHERE id = ?";
        jdbcTemplate.update(sql, lastRunTime, Instant.now(), id);
    }
    
    @Override
    public void delete(UUID id) {
        String sql = "DELETE FROM jobs WHERE id = ?";
        jdbcTemplate.update(sql, id);
    }
    
    @Override
    public long countByStatus(JobStatus status) {
        String sql = "SELECT COUNT(*) FROM jobs WHERE status = ?";
        return jdbcTemplate.queryForObject(sql, Long.class, status.toString());
    }
}

ORM vs. Raw Query Decisions

For the distributed scheduler, we use raw SQL queries with JdbcTemplate instead of an ORM like Hibernate for the following reasons:

1. Performance: Raw queries provide better performance for high-throughput scenarios where we need to process thousands of jobs per second
2. Control: Direct control over query execution and optimization
3. Simplicity: Avoids the complexity of ORM mapping for simple CRUD operations
4. Scalability: Better support for database-specific optimizations and partitioning strategies

Read vs. Write Separation

The scheduler implements read-write separation using a CQRS-like approach:

public class ReadJobRepository {
    private final JdbcTemplate readJdbcTemplate;
    
    public ReadJobRepository(JdbcTemplate readJdbcTemplate) {
        this.readJdbcTemplate = readJdbcTemplate;
    }
    
    public List<Job> findScheduledJobs(Instant now, int limit) {
        String sql = """
            SELECT * FROM jobs 
            WHERE scheduled_time <= ? 
            AND status IN ('PENDING', 'SCHEDULED') 
            ORDER BY scheduled_time ASC 
            LIMIT ?
            """;
            
        return readJdbcTemplate.query(sql, new JobRowMapper(), now, limit);
    }
    
    public JobStatusCount getJobCounts() {
        String sql = """
            SELECT status, COUNT(*) as count 
            FROM jobs 
            GROUP BY status
            """;
            
        Map<String, Long> counts = new HashMap<>();
        readJdbcTemplate.query(sql, rs -> {
            counts.put(rs.getString("status"), rs.getLong("count"));
        });
        
        return new JobStatusCount(counts);
    }
}

Connection Pool Configuration

@Configuration
public class DatabaseConfig {
    
    @Bean
    public DataSource dataSource() {
        HikariDataSource dataSource = new HikariDataSource();
        dataSource.setJdbcUrl("jdbc:postgresql://localhost:5432/scheduler");
        dataSource.setUsername("scheduler_user");
        dataSource.setPassword("scheduler_password");
        dataSource.setMaximumPoolSize(20);
        dataSource.setMinimumIdle(5);
        dataSource.setConnectionTimeout(30000);
        dataSource.setIdleTimeout(600000);
        dataSource.setMaxLifetime(1800000);
        dataSource.setLeakDetectionThreshold(60000);
        return dataSource;
    }
}

N+1 Query Prevention Strategy

The scheduler uses eager loading and pre-fetching to avoid N+1 query problems:

public class JobWithDependencies {
    private final Job job;
    private final List<JobExecution> executions;
    
    public JobWithDependencies(Job job, List<JobExecution> executions) {
        this.job = job;
        this.executions = executions;
    }
    
    // Getters and methods
}

public class JobWithDependenciesRepository {
    public List<JobWithDependencies> findJobsWithExecutions(List<UUID> jobIds) {
        // Single query to fetch jobs and their executions
        String sql = """
            SELECT j.*, e.* FROM jobs j 
            LEFT JOIN job_executions e ON j.id = e.job_id
            WHERE j.id IN (?)
            """;
            
        // Implementation using custom result mapping to avoid N+1
        return jdbcTemplate.query(sql, new JobWithDependenciesRowMapper(), jobIds);
    }
}

Batch Operation Design

public class BatchJobRepository {
    
    public void batchUpdateStatus(List<UUID> jobIds, JobStatus status) {
        String sql = "UPDATE jobs SET status = ?, updated_at = ? WHERE id = ?";
        
        // Use batch update for better performance
        jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement ps, int i) throws SQLException {
                UUID jobId = jobIds.get(i);
                ps.setString(1, status.toString());
                ps.setTimestamp(2, Timestamp.from(Instant.now()));
                ps.setObject(3, jobId);
            }
            
            @Override
            public int getBatchSize() {
                return jobIds.size();
            }
        });
    }
    
    public void batchSave(List<Job> jobs) {
        String sql = """
            INSERT INTO jobs (id, name, description, cron_expression, status, 
                             scheduled_time, last_run_time, next_run_time, 
                             job_data, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """;
            
        jdbcTemplate.batchUpdate(sql, new BatchPreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement ps, int i) throws SQLException {
                Job job = jobs.get(i);
                ps.setObject(1, job.getId());
                ps.setString(2, job.getName());
                ps.setString(3, job.getDescription());
                ps.setString(4, job.getCronExpression());
                ps.setString(5, job.getStatus().toString());
                ps.setTimestamp(6, Timestamp.from(job.getScheduledTime()));
                ps.setTimestamp(7, job.getLastRunTime() != null ? 
                    Timestamp.from(job.getLastRunTime()) : null);
                ps.setTimestamp(8, Timestamp.from(job.getNextRunTime()));
                ps.setString(9, objectMapper.writeValueAsString(job.getJobData()));
                ps.setTimestamp(10, Timestamp.from(job.getCreatedAt()));
                ps.setTimestamp(11, Timestamp.from(job.getUpdatedAt()));
            }
            
            @Override
            public int getBatchSize() {
                return jobs.size();
            }
        });
    }
}

Transaction Management

@Service
@Transactional
public class JobService {
    private final JobRepository jobRepository;
    private final JobExecutionRepository executionRepository;
    
    public void scheduleJob(Job job) {
        // Transaction boundary includes both job creation and execution tracking
        Job savedJob = jobRepository.save(job);
        JobExecution execution = new JobExecution(savedJob.getId(), 
            JobExecutionStatus.PENDING, Instant.now());
        executionRepository.save(execution);
    }
}

Data Access Layer Design Summary

The repository layer is designed with the following key characteristics:

1. Separation of Concerns: Clear distinction between read and write operations
2. Performance Optimization: Raw SQL queries for maximum efficiency
3. Scalability: Connection pooling and batch operations for high throughput
4. Maintainability: Clean abstraction with well-defined interfaces
5. N+1 Prevention: Eager loading and batch operations to avoid performance bottlenecks
6. Transaction Safety: Proper transaction boundaries for data consistency

This design ensures that the scheduler can handle thousands of concurrent job operations while maintaining optimal database performance and data integrity.

9. Asynchronous Processing & Messaging

Event-driven boundaries in this feature

The distributed scheduler will use a hybrid synchronous/asynchronous approach where:

• Synchronous flows: Job creation, immediate execution, and status updates happen synchronously to ensure consistency and immediate feedback.
• Asynchronous flows: Job scheduling, retry logic, and status notifications are handled asynchronously to decouple components and improve scalability.

The primary async triggers include:

1. Job completion events (success/failure)
2. Scheduled job execution
3. Retry attempts for failed jobs
4. External system notifications (e.g., when a job completes and needs to trigger downstream systems)

Message schema design

{
  "version": "1.0",
  "event_type": "job.completed",
  "timestamp": "2023-10-27T14:30:00Z",
  "job_id": "uuid-12345",
  "status": "success",
  "result": {
    "output": "processed data",
    "duration_ms": 1200
  },
  "error": null,
  "metadata": {
    "source_system": "scheduler-service",
    "correlation_id": "correlation-123"
  }
}

Dead letter queue strategy and poison message handling

Poison messages will be routed to a dedicated dead-letter queue (DLQ) with a separate consumer that:

1. Logs the failed message with full context
2. Triggers an alert to on-call engineers
3. Provides a manual retry mechanism via admin UI or API endpoint
4. Stores the message in a persistent archive for forensic analysis

At-least-once vs. exactly-once delivery

At-least-once delivery is selected for this system because:

• Job execution results are idempotent in most cases
• The system is designed to handle duplicate job executions gracefully
• Simpler implementation and better performance characteristics

Retry policy for consumers

Retry configuration:

• Max attempts: 5
• Backoff strategy: Exponential with jitter (1s, 2s, 4s, 8s, 16s)
• Retryable errors: Network timeouts, service unavailability, database connection issues
• Non-retryable errors: Invalid job configuration, authorization failures

public class RetryPolicy {
    private static final int MAX_RETRIES = 5;
    private static final long BASE_DELAY_MS = 1000;
    
    public boolean shouldRetry(Exception exception) {
        return exception instanceof NetworkException || 
               exception instanceof ServiceUnavailableException ||
               exception instanceof DatabaseConnectionException;
    }
    
    public long getDelay(int attempt) {
        long delay = (long) Math.pow(2, attempt) * BASE_DELAY_MS;
        return delay + ThreadLocalRandom.current().nextLong(0, 500); // Add jitter
    }
}

Outbox pattern for reliable event publishing

The outbox pattern is implemented to ensure eventual consistency between job state updates and event publishing:

public class JobEventOutbox {
    private final EventRepository eventRepository;
    private final EventPublisher eventPublisher;
    
    @Transactional
    public void publishJobEvent(Job job, JobStatus status) {
        // 1. Update job state
        job.setStatus(status);
        jobRepository.save(job);
        
        // 2. Store event in outbox
        JobEvent event = new JobEvent(
            job.getId(),
            status.name(),
            System.currentTimeMillis()
        );
        eventRepository.save(event);
        
        // 3. Publish to message broker
        eventPublisher.publish(event);
    }
}

Sequence diagram for async flows

sequenceDiagram
    participant SchedulerService
    participant JobRepository
    participant EventOutbox
    participant MessageBroker
    participant EventConsumer
    
    SchedulerService->>JobRepository: Update job status to COMPLETED
    JobRepository-->>SchedulerService: Acknowledge update
    SchedulerService->>EventOutbox: Store event in outbox
    EventOutbox-->>SchedulerService: Acknowledge event storage
    SchedulerService->>MessageBroker: Publish job completed event
    MessageBroker-->>SchedulerService: Acknowledge publish
    MessageBroker->>EventConsumer: Deliver event to consumer
    EventConsumer-->>MessageBroker: Acknowledge receipt

10. Error Handling Design

Error Classification Hierarchy

The distributed scheduler implements a multi-level error classification system:

public enum ErrorType {
    VALIDATION_ERROR,        // Input validation failures
    BUSINESS_LOGIC_ERROR,    // Domain-specific business rule violations
    INFRASTRUCTURE_ERROR,    // System-level issues (DB, network, etc.)
    SERVICE_UNAVAILABLE,     // Temporary unavailability of downstream services
    RATE_LIMITED,            // Request throttled by external systems
    AUTHENTICATION_ERROR,    // AuthN/AuthZ failures
    UNKNOWN_ERROR            // Unhandled exceptions
}

public class ErrorHandler {
    public ErrorResult handleException(Exception ex) {
        if (ex instanceof ValidationException) {
            return new ErrorResult(ErrorType.VALIDATION_ERROR, ex.getMessage());
        } else if (ex instanceof BusinessLogicException) {
            return new ErrorResult(ErrorType.BUSINESS_LOGIC_ERROR, ex.getMessage());
        } else if (ex instanceof ServiceUnavailableException) {
            return new ErrorResult(ErrorType.SERVICE_UNAVAILABLE, ex.getMessage());
        } else if (ex instanceof RateLimitExceededException) {
            return new ErrorResult(ErrorType.RATE_LIMITED, ex.getMessage());
        } else {
            return new ErrorResult(ErrorType.UNKNOWN_ERROR, "Internal server error");
        }
    }
}

Error Propagation Strategy

The scheduler follows a layered propagation strategy:

• Validation errors: Caught and converted to structured responses at API boundary
• Business logic errors: Bubble up through service layer to API layer for proper HTTP status codes
• Infrastructure errors: Logged at service level and re-thrown as wrapped exceptions for higher-level handling

Retry Strategy with Exponential Backoff and Jitter

For transient infrastructure issues, the scheduler implements a robust retry mechanism:

public class RetryableOperation {
    private static final int MAX_RETRIES = 3;
    private static final long BASE_DELAY_MS = 1000;
    private static final double JITTER_FACTOR = 0.2;

    public <T> T executeWithRetry(Supplier<T> operation) {
        int attempts = 0;
        Exception lastException = null;

        while (attempts < MAX_RETRIES) {
            try {
                return operation.get();
            } catch (Exception ex) {
                lastException = ex;
                if (!isRetryable(ex) || attempts >= MAX_RETRIES - 1) {
                    throw new RetryFailedException("Operation failed after " + MAX_RETRIES + " attempts", ex);
                }
                
                long delay = calculateBackoffDelay(attempts);
                try {
                    Thread.sleep(delay);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RetryFailedException("Retry interrupted", ie);
                }
                attempts++;
            }
        }
        throw new RetryFailedException("Retry operation failed", lastException);
    }

    private long calculateBackoffDelay(int attempt) {
        long baseDelay = (long) (BASE_DELAY_MS * Math.pow(2, attempt));
        long jitter = (long) (baseDelay * JITTER_FACTOR * (Math.random() - 0.5) * 2);
        return baseDelay + jitter;
    }

    private boolean isRetryable(Exception ex) {
        return ex instanceof ServiceUnavailableException ||
               ex instanceof TimeoutException ||
               ex instanceof NetworkException;
    }
}

Circuit Breaker Design

The scheduler implements a circuit breaker pattern to prevent cascading failures:

public class CircuitBreaker {
    private final int failureThreshold;
    private final long timeoutMs;
    private final int successThreshold;

    private volatile CircuitState state = CircuitState.CLOSED;
    private volatile long lastFailureTime = 0;
    private volatile int failureCount = 0;
    private volatile int successCount = 0;

    public enum CircuitState {
        CLOSED,      // Normal operation - requests allowed
        OPEN,        // Failure threshold exceeded - all requests rejected
        HALF_OPEN    // Partially open - allows test requests
    }

    public <T> T execute(Supplier<T> operation) throws Exception {
        switch (state) {
            case CLOSED:
                return executeClosed(operation);
            case OPEN:
                return executeOpen(operation);
            case HALF_OPEN:
                return executeHalfOpen(operation);
            default:
                throw new IllegalStateException("Unknown circuit state");
        }
    }

    private <T> T executeClosed(Supplier<T> operation) throws Exception {
        try {
            T result = operation.get();
            onSuccess();
            return result;
        } catch (Exception ex) {
            onFailure(ex);
            throw ex;
        }
    }

    private <T> T executeOpen(Supplier<T> operation) throws Exception {
        if (System.currentTimeMillis() - lastFailureTime > timeoutMs) {
            state = CircuitState.HALF_OPEN;
            return executeHalfOpen(operation);
        }
        throw new ServiceUnavailableException("Circuit breaker is OPEN");
    }

    private <T> T executeHalfOpen(Supplier<T> operation) throws Exception {
        try {
            T result = operation.get();
            onSuccess();
            return result;
        } catch (Exception ex) {
            onFailure(ex);
            throw ex;
        }
    }

    private void onSuccess() {
        successCount++;
        if (state == CircuitState.HALF_OPEN && successCount >= successThreshold) {
            state = CircuitState.CLOSED;
            failureCount = 0;
            successCount = 0;
        }
    }

    private void onFailure(Exception ex) {
        failureCount++;
        lastFailureTime = System.currentTimeMillis();
        
        if (state == CircuitState.CLOSED && failureCount >= failureThreshold) {
            state = CircuitState.OPEN;
        } else if (state == CircuitState.HALF_OPEN) {
            state = CircuitState.OPEN;
        }
    }
}

Fallback Behavior Design

For critical operations, the scheduler provides fallback mechanisms:

public class SchedulerFallbacks {
    private final ScheduledTaskRepository taskRepo;
    private final NotificationService notificationService;

    public List<ScheduledTask> getTasksWithFallback(String userId, String status) {
        try {
            return taskRepo.findByUserIdAndStatus(userId, status);
        } catch (Exception ex) {
            // Log the failure for monitoring
            logger.warn("Failed to fetch tasks from DB, using fallback", ex);
            
            // Return cached or default data if available
            return Collections.emptyList(); // Or return from cache layer
        }
    }

    public void sendAlertWithFallback(String taskName, String error) {
        try {
            notificationService.sendAlert(taskName, error);
        } catch (Exception ex) {
            logger.error("Failed to send alert notification: {}", error, ex);
            // Fallback to logging or alternative alerting mechanism
        }
    }
}

Error Logging Format and Alerting Thresholds

Structured logging with consistent format:

public class StructuredErrorLogger {
    private static final Logger logger = LoggerFactory.getLogger(StructuredErrorLogger.class);

    public void logError(String operation, String errorCode, String errorMessage, 
                        Map<String, Object> context) {
        Map<String, Object> logContext = new HashMap<>(context);
        logContext.put("operation", operation);
        logContext.put("errorCode", errorCode);
        logContext.put("errorMessage", errorMessage);
        logContext.put("timestamp", System.currentTimeMillis());
        logContext.put("traceId", MDC.get("trace_id"));
        
        logger.error("Scheduler error occurred: {}", logContext);
    }

    // Alert thresholds
    private static final Map<String, Integer> ALERT_THRESHOLDS = Map.of(
        "DB_CONNECTION_ERROR", 5,     // Alert if >5 DB connection errors in 1 minute
        "TASK_EXECUTION_FAILURE", 3,  // Alert if >3 task execution failures in 5 minutes
        "SERVICE_UNAVAILABLE", 2      // Alert if >2 service unavailability errors in 10 minutes
    );
}

Key Alerting Thresholds

Error Type	Threshold	Alert Level	Escalation
DB_CONNECTION_ERROR	5 occurrences per minute	Warning	On-call engineer
TASK_EXECUTION_FAILURE	3 occurrences per 5 minutes	Critical	On-call engineer + team
SERVICE_UNAVAILABLE	2 occurrences per 10 minutes	Critical	On-call engineer + SRE
RATE_LIMITED	10 occurrences per hour	Warning	Team alert

The error handling design ensures robustness, observability, and graceful degradation while maintaining system stability under various failure conditions.

11. Security Design

Input Validation and Sanitization Strategy

All inputs are validated using a combination of:

• Regex patterns for format validation (e.g., cron expressions, UUIDs)
• Length limits to prevent buffer overflow attacks
• Type checking for all primitive fields
• Custom validators for domain-specific constraints (e.g., valid cron syntax)

public class ScheduleValidator {
    private static final Pattern CRON_PATTERN = Pattern.compile("^[0-9*/,-]{1,100}$");
    private static final Pattern UUID_PATTERN = Pattern.compile("^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$");
    
    public static void validateCronExpression(String cron) {
        if (cron == null || !CRON_PATTERN.matcher(cron).matches()) {
            throw new ValidationException("Invalid cron expression format");
        }
    }
    
    public static void validateScheduleId(String scheduleId) {
        if (scheduleId == null || !UUID_PATTERN.matcher(scheduleId).matches()) {
            throw new ValidationException("Invalid schedule ID format");
        }
    }
}

SQL Injection Prevention

All database queries use parameterized statements with:

• Prepared statements for all dynamic queries
• Whitelisted field names for sorting and filtering operations
• Escape sequences for special characters in string literals

Authentication Middleware Design

Authentication is handled via JWT tokens with:

• OAuth2 Bearer token format for API access
• Token expiration handling with 1-hour TTLs
• Refresh token rotation for enhanced security

Authorization Enforcement

RBAC-based authorization implemented at service layer:

• Role hierarchy: ADMIN > USER > GUEST
• Resource-based permissions: SCHEDULE_READ, SCHEDULE_WRITE, SCHEDULE_EXECUTE
• Context-aware checks for schedule ownership and execution scope

@PreAuthorize("hasPermission(#scheduleId, 'SCHEDULE_WRITE')")
public void updateSchedule(String scheduleId, ScheduleUpdateRequest request) {
    // Implementation
}

Sensitive Data Encryption

• At rest: AES-256 encryption for schedule payloads and configuration data
• In transit: TLS 1.3 with HSTS enforcement for all API endpoints
• Key management: AWS KMS integration for key rotation and lifecycle management

PII Handling and Masking Strategy

• Schedule names and description fields are encrypted at rest
• Execution logs containing sensitive data are masked before logging
• Audit trail excludes PII fields from being stored in plaintext

Secrets Management

• Environment variables for configuration secrets (e.g., database passwords, API keys)
• Vault integration for dynamic credential rotation
• Secrets lifecycle management with automatic rotation every 90 days

Audit Logging Design

• Events logged: SCHEDULE_CREATED, SCHEDULE_UPDATED, SCHEDULE_EXECUTED, SCHEDULE_DELETED
• Log format: Structured JSON with trace_id, user_id, operation_type, timestamp
• Retention policy: 90-day retention for audit logs with archiving to S3

{
  "timestamp": "2023-10-15T14:30:00Z",
  "trace_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "user_id": "user-123",
  "operation_type": "SCHEDULE_CREATED",
  "schedule_id": "uuid-456",
  "details": {
    "name": "encrypted-schedule-name",
    "cron_expression": "encrypted-cron"
  }
}

12. Testing Strategy

Unit Testing Scope and Coverage Targets (%)

• Service Layer: 85% code coverage with focus on business logic, error handling, and boundary conditions
• Repository Layer: 90% coverage with emphasis on query correctness and edge case handling
• Domain Objects: 100% coverage for value objects, immutable entities, and validation logic
• Event Handlers: 80% coverage for async processing logic and error recovery paths
• Core Algorithms: 95% coverage including scheduling algorithm edge cases, time zone handling, and recurrence patterns

Integration Testing Strategy

• Database Integration: Use Testcontainers with PostgreSQL 14+ to test actual query execution, transaction behavior, and connection pooling
• External Service Mocks: WireMock for third-party API integrations (e.g., notification services, external calendar APIs)
• Message Broker Integration: Embedded Kafka or RabbitMQ for testing event-driven workflows and consumer behavior
• Caching Layer: Redis Testcontainers to validate cache hit/miss patterns, TTL behavior, and invalidation flows
• Authentication Integration: Mock OAuth2 server to test token validation, scope checking, and role-based access control

Contract Testing

• API Contracts: Use Pact for consumer-driven contract testing between scheduler service and client applications
• Event Contracts: Define Avro schemas with versioning strategy for inter-service event communication (e.g., job scheduled, job failed)
• Database Contracts: Test query result schemas and data format consistency across different database engines (PostgreSQL, MySQL)

Performance Testing Tool

• Load Testing: k6 for simulating concurrent job scheduling and execution scenarios with realistic user behavior patterns
• Key Scenarios:
  • Concurrent job creation (1000+ simultaneous requests)
  • High-frequency recurring job scheduling (100 jobs/sec)
  • Large batch job processing (5000+ jobs in single batch)
  • Failover and recovery under load conditions
• Metrics Tracked:
  • p50/p95/p99 response times for job creation and execution
  • Throughput (jobs/sec) under different load conditions
  • Memory usage and GC pressure during high-load scenarios

Test Data Management Strategy

• Fixtures: Predefined JSON/YAML files for common job configurations, schedule patterns, and user roles
• Factories: Java-based factory classes for generating test data with realistic constraints (e.g., valid cron expressions, time zones)
• Seeding: Database seeding scripts for initial test environment setup with sample users, jobs, and configurations
• Test Data Isolation: Use separate database schemas or containers per test suite to prevent cross-contamination
• Data Generation: Synthetic data generators for large-scale testing (e.g., generating 100K+ unique job IDs, random time ranges)

// Example unit test for job scheduling logic
@Test
public void testScheduleJobWithValidCronExpression() {
    Job job = new JobBuilder()
        .withName("test-job")
        .withCronExpression("0 0 12 * * ?")
        .withHandler("email-notification-handler")
        .build();
    
    Job scheduledJob = jobScheduler.schedule(job);
    
    assertThat(scheduledJob.getId()).isNotNull();
    assertThat(scheduledJob.getNextRunTime())
        .isAfter(Instant.now());
    assertThat(scheduledJob.getStatus()).isEqualTo(JobStatus.SCHEDULED);
}

13. Observability Design

Structured Logging Schema

All logs follow a consistent JSON schema to enable centralized logging and analysis:

{
  "timestamp": "2023-10-27T14:30:45.123Z",
  "level": "INFO",
  "service": "distributed-scheduler",
  "trace_id": "12345678-1234-1234-1234-123456789012",
  "span_id": "abcdef12-3456-7890-abcd-ef1234567890",
  "user_id": "user-123",
  "job_id": "job-456",
  "operation": "scheduleJob",
  "message": "Job scheduled successfully",
  "duration_ms": 15,
  "error_code": null,
  "error_message": null,
  "stack_trace": null
}

PII masking rules:

• User IDs are masked in logs when not required for debugging
• Job payloads are truncated or redacted in error contexts
• IP addresses and sensitive headers are excluded from logs

Distributed Tracing Instrumentation

Using OpenTelemetry for distributed tracing with custom spans:

Key Operations to Trace:

1. job.scheduling - from job creation to queueing
2. job.execution - from queue consumption to completion
3. job.failure - error handling and retry logic

Span Attributes:

• job.id: Unique job identifier
• job.type: Job type (e.g., "email_notification", "data_sync")
• job.status: Current job status (pending, running, completed, failed)
• job.retry_count: Number of retry attempts
• job.scheduled_at: Original scheduling timestamp
• job.execution_duration_ms: Time taken to execute job

// Example custom span in job execution flow
Span span = tracer.spanBuilder("job.execution")
    .setAttribute("job.id", job.getId())
    .setAttribute("job.type", job.getType())
    .startSpan();

try (Scope scope = span.makeCurrent()) {
    // Job execution logic
    executeJob(job);
} catch (Exception e) {
    span.recordException(e);
    span.setStatus(StatusCode.ERROR);
    throw e;
} finally {
    span.end();
}

Metrics Design

RED Method Metrics (Rate, Errors, Duration)

Metric Name	Type	Labels	Description
job_scheduled_total	Counter	job_type, status	Total jobs scheduled
job_execution_duration_seconds	Histogram	job_type, status	Execution time per job
job_execution_errors_total	Counter	job_type, error_code	Failed job executions
job_queue_size	Gauge	job_type	Current queue size
scheduler_active_jobs	Gauge	job_type	Currently executing jobs
job_retry_count	Histogram	job_type	Number of retries per job

Example Prometheus Metrics:

# Total jobs scheduled by type and status
job_scheduled_total{job_type="email_notification", status="success"} 1250
job_scheduled_total{job_type="email_notification", status="failed"} 15

# Job execution duration histogram
job_execution_duration_seconds_bucket{job_type="data_sync",le="10"} 500
job_execution_duration_seconds_bucket{job_type="data_sync",le="30"} 850
job_execution_duration_seconds_bucket{job_type="data_sync",le="60"} 920
job_execution_duration_seconds_bucket{job_type="data_sync",le="+Inf"} 1000
job_execution_duration_seconds_sum{job_type="data_sync"} 25600.0
job_execution_duration_seconds_count{job_type="data_sync"} 1000

# Active jobs gauge
scheduler_active_jobs{job_type="email_notification"} 42

Health Check Endpoint Design

Endpoints:

1. /health - Application health status
2. /ready - Service readiness (dependencies ready)
3. /live - Process liveness check
4. /startup - Startup completion status

Health Check Response Format:

{
  "status": "healthy",
  "timestamp": "2023-10-27T14:30:45.123Z",
  "components": {
    "database": {
      "status": "healthy",
      "latency_ms": 25
    },
    "message_queue": {
      "status": "healthy",
      "latency_ms": 15
    },
    "cache": {
      "status": "healthy",
      "latency_ms": 5
    }
  },
  "uptime_seconds": 3600,
  "version": "1.2.3"
}

Alerting Thresholds

Critical Alerts (Escalation to On-Call):

Metric	Threshold	Alert Level
job_execution_errors_total (5m)	> 100 errors	Critical
job_scheduled_total (5m)	< 50% of expected rate	Critical
job_execution_duration_seconds (p99)	> 30s	Critical
scheduler_active_jobs	> 500 concurrent jobs	Warning

Warning Alerts:

Metric	Threshold	Alert Level
job_execution_errors_total (5m)	> 20 errors	Warning
job_execution_duration_seconds (p99)	> 15s	Warning
job_queue_size	> 1000 queued jobs	Warning

On-Call Escalation Path:

1. Critical alerts → Slack channel + PagerDuty
2. Warning alerts → Slack channel only
3. Alert severity levels mapped to escalation timeframes (15m for warning, 5m for critical)

14. Open Questions, Risks & Trade-offs

Decisions requiring product / engineering alignment

• Scheduling granularity and precision requirements: Should the system support sub-second scheduling? This impacts the design of the scheduler loop, database indexing strategy, and whether we need to implement a separate high-precision component for microsecond-level scheduling.

• Failure recovery strategy: Should failed jobs be retried automatically or require manual intervention? This affects the design of the job status tracking, retry logic, and whether we need to implement a job dependency graph or just a simple queue.

• Job lifecycle management: How should jobs be handled when their associated entities (e.g., user, project) are deleted or suspended? Do we soft-delete jobs or immediately cancel them?

Known trade-offs and justifications

• Consistency vs. availability: We're leaning toward eventual consistency for job execution status to ensure high availability, but this introduces a risk of temporary inconsistency between the scheduler and job runners. This is acceptable for most use cases where immediate consistency isn't required.

• Single point of failure vs. distributed coordination: Using a single scheduler instance for simplicity introduces SPOF risk, but the design supports horizontal scaling through sharding and leader election. We're starting with a single instance to reduce complexity and can scale out later if needed.

• Eventual consistency vs. strong consistency: Job execution status is eventually consistent to allow for better scalability and availability. This trade-off is acceptable because job execution order is not critical in most use cases.

Risks and mitigation strategies

• Race conditions in job scheduling: Multiple concurrent requests to schedule the same job could result in duplicate jobs. Mitigation: Use database-level constraints and optimistic locking to prevent duplicates.

• Clock drift in distributed systems: Jobs may execute at incorrect times due to clock differences across nodes. Mitigation: Implement a clock synchronization strategy using NTP or use a centralized time source with tolerance windows.

• Resource exhaustion under high load: If job execution is CPU or memory intensive, the system may become unresponsive. Mitigation: Implement job queue prioritization and resource limits per job type.

• Job execution timeout: Jobs that run longer than expected can block the scheduler. Mitigation: Implement job timeout and cancellation logic with proper error handling.

Technical debt acknowledged

• Limited support for complex scheduling patterns: The current design supports basic cron-like scheduling but doesn't handle advanced patterns like "run every 30 minutes on weekdays between 9 AM and 5 PM". This is a known limitation that would require additional complexity in the scheduling engine.

• Hardcoded job execution logic: The current design assumes all jobs execute with similar patterns. In reality, different job types may require different execution strategies (e.g., batch processing vs. real-time). This design will need to be refactored to support job type-specific execution logic.

• No built-in job dependency management: Jobs that depend on each other must be handled at the application level, which introduces additional complexity for job creators. This is a limitation that could be addressed with a more sophisticated scheduling engine.

• Limited monitoring and alerting: The current design doesn't include comprehensive job execution monitoring or alerting for failed jobs. This is a known gap that would require additional integration with monitoring systems.

• No built-in job versioning: Jobs cannot be versioned or rolled back, which is important for production systems where job changes might break existing workflows. This would require additional design work to support job versioning and rollback capabilities.