Distributed Cron & Job Scheduling.

# /etc/crontab
0 9 * * * /scripts/send_daily_report.sh
*/5 * * * * /scripts/cleanup_old_sessions.sh
0 0 * * * /scripts/generate_invoices.sh

def run_scheduled_job(job_id):
    # Try to acquire lock
    if db.execute("UPDATE jobs SET locked = true WHERE id = ? AND locked = false", job_id):
        try:
            execute_job(job_id)
        finally:
            db.execute("UPDATE jobs SET locked = false WHERE id = ?", job_id)

def run_scheduled_job(job_id):
    lock_until = datetime.now() + timedelta(minutes=5)
    
    if db.execute("""
        UPDATE jobs 
        SET locked_until = ? 
        WHERE id = ? AND (locked_until IS NULL OR locked_until < NOW())
    """, lock_until, job_id):
        try:
            execute_job(job_id)
        finally:
            db.execute("UPDATE jobs SET locked_until = NULL WHERE id = ?", job_id)

def run_scheduled_job(job_id):
    lock = redis.set(f"lock:{job_id}", server_id, nx=True, ex=300)
    
    if lock:
        try:
            execute_job(job_id)
        finally:
            # Only release if we still own the lock
            if redis.get(f"lock:{job_id}") == server_id:
                redis.delete(f"lock:{job_id}")

class FencedJobExecutor:
    def __init__(self, db):
        self.db = db
    
    def execute_job(self, job_id: str, fencing_token: int, job_func):
        """Execute job only if fencing token is current."""
        
        # Atomically check and update token
        result = self.db.execute("""
            UPDATE jobs 
            SET last_execution_token = ?,
                last_execution_started = NOW()
            WHERE id = ? 
            AND (last_execution_token IS NULL OR last_execution_token <= ?)  -- same leader may rerun; older tokens rejected
            RETURNING id
        """, fencing_token, job_id, fencing_token)
        
        if not result:
            raise StaleLeaderError(f"Token {fencing_token} is stale")
        
        try:
            job_func()
            
            # Mark completion with token
            self.db.execute("""
                UPDATE jobs 
                SET last_execution_completed = NOW()
                WHERE id = ? AND last_execution_token = ?
            """, job_id, fencing_token)
            
        except Exception as e:
            # Mark failure with token
            self.db.execute("""
                UPDATE jobs 
                SET last_execution_error = ?
                WHERE id = ? AND last_execution_token = ?
            """, str(e), job_id, fencing_token)
            raise

import etcd3
from threading import Thread, Event
import time

class EtcdLeaderElection:
    """Leader election using etcd leases."""
    
    def __init__(self, etcd_client, election_name: str, ttl: int = 10):
        self.client = etcd_client
        self.election_name = election_name
        self.ttl = ttl
        self.lease = None
        self.is_leader = False
        self.leader_key = f"/elections/{election_name}/leader"
        self.stop_event = Event()
    
    def campaign(self, candidate_id: str, on_elected, on_demoted):
        """Campaign to become leader."""
        
        while not self.stop_event.is_set():
            try:
                # Create a lease that must be renewed
                self.lease = self.client.lease(self.ttl)
                
                # Try to become leader
                success, _ = self.client.transaction(
                    compare=[
                        self.client.transactions.create(self.leader_key) == 0
                    ],
                    success=[
                        self.client.transactions.put(
                            self.leader_key, 
                            candidate_id,
                            lease=self.lease
                        )
                    ],
                    failure=[]
                )
                
                if success:
                    self.is_leader = True
                    on_elected()
                    self._maintain_leadership(on_demoted)
                else:
                    # Someone else is leader, watch for changes
                    self._wait_for_leader_change()
            
            except Exception as e:
                if self.is_leader:
                    self.is_leader = False
                    on_demoted()
                time.sleep(1)
    
    def _maintain_leadership(self, on_demoted):
        """Keep refreshing lease while leader."""
        
        while self.is_leader and not self.stop_event.is_set():
            try:
                self.lease.refresh()
                time.sleep(self.ttl / 3)
            except:
                self.is_leader = False
                on_demoted()
                break
    
    def _wait_for_leader_change(self):
        """Watch for leader key deletion."""
        
        events, cancel = self.client.watch(self.leader_key)
        
        for event in events:
            if isinstance(event, etcd3.events.DeleteEvent):
                cancel()
                return  # Leader gone, try to become leader
            
            if self.stop_event.is_set():
                cancel()
                return
    
    def resign(self):
        """Voluntarily give up leadership."""
        
        if self.is_leader and self.lease:
            self.lease.revoke()
            self.is_leader = False
    
    def stop(self):
        """Stop the election process."""
        self.stop_event.set()
        self.resign()

from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
from typing import Callable, Optional
import uuid

class JobStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    DEAD = "dead"  # Exceeded all retries

@dataclass
class ScheduledJob:
    id: str
    name: str
    cron_expression: str
    handler: str  # e.g., "app.jobs.send_daily_report"
    
    # Execution tracking
    status: JobStatus = JobStatus.PENDING
    last_run_at: Optional[datetime] = None
    next_run_at: Optional[datetime] = None
    
    # Current execution
    execution_id: Optional[str] = None
    execution_started_at: Optional[datetime] = None
    execution_token: Optional[int] = None
    
    # Retry configuration
    max_retries: int = 3
    retry_count: int = 0
    timeout_seconds: int = 3600  # 1 hour default
    
    # Failure tracking
    last_error: Optional[str] = None
    consecutive_failures: int = 0

class AtLeastOnceJobExecutor:
    """
    Job executor with at-least-once guarantee.
    
    Guarantees:
    - Jobs will run at least once (may run multiple times on failures)
    - Failed jobs will be retried
    - Stuck jobs will be detected and re-run
    - Jobs are idempotent (caller's responsibility)
    """
    
    def __init__(self, db, job_registry, fencing_token: int):
        self.db = db
        self.job_registry = job_registry
        self.fencing_token = fencing_token
    
    def execute_job(self, job: ScheduledJob) -> bool:
        """
        Execute a scheduled job with at-least-once semantics.
        Returns True if job completed successfully.
        """
        
        execution_id = str(uuid.uuid4())
        
        # Try to claim the job
        claimed = self._claim_job(job.id, execution_id)
        if not claimed:
            return False  # Another instance is running it
        
        try:
            # Get the handler function
            handler = self.job_registry.get_handler(job.handler)
            if not handler:
                raise ValueError(f"Unknown job handler: {job.handler}")
            
            # Execute with timeout
            self._execute_with_timeout(handler, job.timeout_seconds)
            
            # Mark success
            self._mark_completed(job.id, execution_id)
            return True
        
        except TimeoutError:
            self._mark_failed(job.id, execution_id, "Job timed out")
            return False
        
        except Exception as e:
            self._mark_failed(job.id, execution_id, str(e))
            return False
    
    def _claim_job(self, job_id: str, execution_id: str) -> bool:
        """
        Atomically claim a job for execution.
        Uses fencing token to prevent stale leaders from claiming.
        """
        
        result = self.db.execute("""
            UPDATE scheduled_jobs
            SET status = 'running',
                execution_id = ?,
                execution_started_at = NOW(),
                execution_token = ?
            WHERE id = ?
            AND status IN ('pending', 'failed')
            AND (execution_token IS NULL OR execution_token <= ?)  -- reject older leaders only
            RETURNING id
        """, execution_id, self.fencing_token, job_id, self.fencing_token)
        
        return result is not None
    
    def _execute_with_timeout(self, handler: Callable, timeout: int):
        """Execute handler with timeout."""
        
        import signal
        
        def timeout_handler(signum, frame):
            raise TimeoutError("Job execution timed out")
        
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(timeout)
        
        try:
            handler()
        finally:
            signal.alarm(0)
    
    def _mark_completed(self, job_id: str, execution_id: str):
        """Mark job as completed."""
        
        self.db.execute("""
            UPDATE scheduled_jobs
            SET status = 'completed',
                last_run_at = NOW(),
                next_run_at = ?,
                retry_count = 0,
                consecutive_failures = 0,
                last_error = NULL
            WHERE id = ? AND execution_id = ?
        """, self._calculate_next_run(job_id), job_id, execution_id)
    
    def _mark_failed(self, job_id: str, execution_id: str, error: str):
        """Mark job as failed, potentially scheduling retry."""
        
        job = self.db.get_job(job_id)
        
        if job.retry_count < job.max_retries:
            # Schedule retry
            retry_delay = self._calculate_retry_delay(job.retry_count)
            
            self.db.execute("""
                UPDATE scheduled_jobs
                SET status = 'pending',
                    next_run_at = NOW() + INTERVAL ? SECOND,
                    retry_count = retry_count + 1,
                    consecutive_failures = consecutive_failures + 1,
                    last_error = ?
                WHERE id = ? AND execution_id = ?
            """, retry_delay, error, job_id, execution_id)
        else:
            # Move to dead status
            self.db.execute("""
                UPDATE scheduled_jobs
                SET status = 'dead',
                    last_error = ?,
                    consecutive_failures = consecutive_failures + 1
                WHERE id = ? AND execution_id = ?
            """, error, job_id, execution_id)
            
            # Alert operations
            self._alert_dead_job(job, error)
    
    def _calculate_retry_delay(self, retry_count: int) -> int:
        """Exponential backoff for retries."""
        base_delay = 60  # 1 minute
        max_delay = 3600  # 1 hour
        delay = base_delay * (2 ** retry_count)
        return min(delay, max_delay)

class StuckJobDetector:
    """
    Detect and recover jobs that are stuck in 'running' state.
    
    A job is considered stuck if:
    - Status is 'running'
    - Started more than timeout_seconds ago
    - No heartbeat received recently
    """
    
    def __init__(self, db, alert_service):
        self.db = db
        self.alert_service = alert_service
    
    def find_stuck_jobs(self) -> list:
        """Find jobs that appear to be stuck."""
        
        return self.db.query("""
            SELECT * FROM scheduled_jobs
            WHERE status = 'running'
            AND execution_started_at < NOW() - INTERVAL timeout_seconds SECOND
            AND (last_heartbeat_at IS NULL 
                 OR last_heartbeat_at < NOW() - INTERVAL 60 SECOND)
        """)
    
    def recover_stuck_jobs(self):
        """Reset stuck jobs to pending for re-execution."""
        
        stuck_jobs = self.find_stuck_jobs()
        
        for job in stuck_jobs:
            # Log the stuck job
            logger.warning(
                "Recovering stuck job",
                job_id=job.id,
                execution_id=job.execution_id,
                started_at=job.execution_started_at
            )
            
            # Check if it's a chronic problem
            if job.consecutive_failures >= 3:
                self.alert_service.alert(
                    severity="high",
                    message=f"Job {job.name} has failed {job.consecutive_failures} times",
                    job_id=job.id
                )
            
            # Reset to pending (will be picked up on next schedule)
            self.db.execute("""
                UPDATE scheduled_jobs
                SET status = 'failed',
                    last_error = 'Job stuck - server may have crashed',
                    consecutive_failures = consecutive_failures + 1
                WHERE id = ? AND execution_id = ?
            """, job.id, job.execution_id)
    
    def run_detector(self, interval: int = 60):
        """Run stuck job detection periodically."""
        
        while True:
            try:
                self.recover_stuck_jobs()
            except Exception as e:
                logger.error("Stuck job detector error", error=str(e))
            
            time.sleep(interval)

class HeartbeatJobExecutor:
    """
    Job executor with heartbeat support for long-running jobs.
    """
    
    def __init__(self, db, heartbeat_interval: int = 30):
        self.db = db
        self.heartbeat_interval = heartbeat_interval
        self.heartbeat_thread = None
        self.stop_heartbeat = False
    
    def execute_with_heartbeat(self, job: ScheduledJob, handler: Callable):
        """Execute job while sending heartbeats."""
        
        self.stop_heartbeat = False
        
        # Start heartbeat thread
        self.heartbeat_thread = Thread(
            target=self._heartbeat_loop,
            args=(job.id, job.execution_id)
        )
        self.heartbeat_thread.start()
        
        try:
            handler()
        finally:
            self.stop_heartbeat = True
            self.heartbeat_thread.join()
    
    def _heartbeat_loop(self, job_id: str, execution_id: str):
        """Send periodic heartbeats."""
        
        while not self.stop_heartbeat:
            try:
                self.db.execute("""
                    UPDATE scheduled_jobs
                    SET last_heartbeat_at = NOW()
                    WHERE id = ? AND execution_id = ?
                """, job_id, execution_id)
            except:
                pass  # Best effort
            
            time.sleep(self.heartbeat_interval)

"""
Distributed Cron System
Combines: Leader Election + Job Scheduling + At-Least-Once Execution
"""

import asyncio
import time
import uuid
from datetime import datetime, timedelta, timezone
from typing import Dict, List, Callable, Optional
from dataclasses import dataclass, field
from croniter import croniter
import structlog

logger = structlog.get_logger()

# =============================================================================
# Configuration
# =============================================================================

@dataclass
class CronConfig:
    # Leader election
    etcd_endpoints: List[str] = field(default_factory=lambda: ["localhost:2379"])
    election_ttl: int = 10  # seconds
    
    # Scheduling
    tick_interval: int = 1  # seconds
    lookahead_seconds: int = 60  # How far ahead to schedule
    
    # Execution
    default_timeout: int = 3600  # 1 hour
    max_retries: int = 3
    stuck_job_threshold: int = 300  # 5 minutes without heartbeat
    
    # Workers
    num_workers: int = 4
    heartbeat_interval: int = 30

# =============================================================================
# Job Definitions
# =============================================================================

@dataclass
class JobDefinition:
    name: str
    cron: str
    handler: str
    timeout: int = 3600
    max_retries: int = 3
    enabled: bool = True
    metadata: Dict = field(default_factory=dict)

@dataclass
class JobExecution:
    id: str
    job_name: str
    scheduled_at: datetime
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    status: str = "pending"  # pending, running, completed, failed, dead
    fencing_token: Optional[int] = None
    worker_id: Optional[str] = None
    error: Optional[str] = None
    retry_count: int = 0

# =============================================================================
# Distributed Cron Scheduler
# =============================================================================

class DistributedCronScheduler:
    """
    Distributed cron scheduler with leader election.
    
    Only the leader schedules jobs. Workers execute them.
    """
    
    def __init__(
        self,
        config: CronConfig,
        etcd_client,
        job_store,
        job_queue,
        job_registry: Dict[str, Callable]
    ):
        self.config = config
        self.etcd = etcd_client
        self.job_store = job_store
        self.job_queue = job_queue
        self.job_registry = job_registry
        
        self.is_leader = False
        self.fencing_token = 0
        self.node_id = str(uuid.uuid4())[:8]
        
        self.leader_election = EtcdLeaderElection(
            etcd_client,
            "cron-scheduler",
            ttl=config.election_ttl
        )
    
    async def start(self):
        """Start the distributed cron system."""
        
        logger.info("Starting distributed cron", node_id=self.node_id)
        self.loop = asyncio.get_running_loop()
        
        # Start leader election in background
        asyncio.create_task(self._run_election())
        
        # Start workers
        for i in range(self.config.num_workers):
            asyncio.create_task(self._run_worker(f"{self.node_id}-worker-{i}"))
        
        # Start stuck job detector
        asyncio.create_task(self._run_stuck_job_detector())
        
        # Keep running
        while True:
            await asyncio.sleep(1)
    
    async def _run_election(self):
        """Run leader election."""
        
        # campaign() is a blocking loop (python-etcd3 is synchronous): run it in a thread
        await asyncio.to_thread(
            self.leader_election.campaign,
            self.node_id,
            self._on_elected,
            self._on_demoted
        )
    
    def _on_elected(self):
        """Called when this node becomes leader."""
        
        self.is_leader = True
        # Fencing token must strictly increase across elections: use the etcd
        # revision at which our leader key was created. Never wall-clock time—
        # a new leader with a lagging clock would get a *smaller* token.
        _, meta = self.etcd.get(self.leader_election.leader_key)
        self.fencing_token = meta.create_revision
        
        logger.info(
            "Became leader",
            node_id=self.node_id,
            fencing_token=self.fencing_token
        )
        
        # Start scheduler loop (we're on the election thread, so hand it to the event loop)
        self.loop.call_soon_threadsafe(
            lambda: asyncio.create_task(self._scheduler_loop())
        )
    
    def _on_demoted(self):
        """Called when this node loses leadership."""
        
        self.is_leader = False
        
        logger.info("Lost leadership", node_id=self.node_id)
    
    async def _scheduler_loop(self):
        """Main scheduler loop (runs only on leader)."""
        
        logger.info("Starting scheduler loop", node_id=self.node_id)
        
        while self.is_leader:
            try:
                await self._schedule_due_jobs()
            except Exception as e:
                logger.error("Scheduler error", error=str(e))
            
            await asyncio.sleep(self.config.tick_interval)
        
        logger.info("Scheduler loop stopped", node_id=self.node_id)
    
    async def _schedule_due_jobs(self):
        """Find and queue jobs that are due to run."""
        
        now = datetime.now(timezone.utc)
        
        # Get all job definitions
        jobs = await self.job_store.get_all_jobs()
        
        for job in jobs:
            if not job.enabled:
                continue
            
            # Check if job is due
            cron = croniter(job.cron, now - timedelta(seconds=self.config.lookahead_seconds))
            next_run = cron.get_next(datetime)
            
            if next_run <= now:
                # Check if already scheduled
                existing = await self.job_store.get_pending_execution(
                    job.name, 
                    next_run
                )
                
                if not existing:
                    # Create execution record
                    execution = JobExecution(
                        id=str(uuid.uuid4()),
                        job_name=job.name,
                        scheduled_at=next_run,
                        fencing_token=self.fencing_token
                    )
                    
                    await self.job_store.create_execution(execution)
                    
                    # Queue for workers
                    await self.job_queue.enqueue({
                        'execution_id': execution.id,
                        'job_name': job.name,
                        'handler': job.handler,
                        'timeout': job.timeout,
                        'fencing_token': self.fencing_token
                    })
                    
                    logger.info(
                        "Scheduled job",
                        job_name=job.name,
                        execution_id=execution.id,
                        scheduled_at=next_run
                    )
    
    async def _run_worker(self, worker_id: str):
        """Worker loop that executes jobs."""
        
        logger.info("Starting worker", worker_id=worker_id)
        
        while True:
            try:
                # Get job from queue
                task = await self.job_queue.dequeue(timeout=5)
                
                if task:
                    await self._execute_job(worker_id, task)
            
            except Exception as e:
                logger.error("Worker error", worker_id=worker_id, error=str(e))
                await asyncio.sleep(1)
    
    async def _execute_job(self, worker_id: str, task: dict):
        """Execute a job task."""
        
        execution_id = task['execution_id']
        job_name = task['job_name']
        handler_name = task['handler']
        timeout = task['timeout']
        fencing_token = task['fencing_token']
        
        logger.info(
            "Executing job",
            worker_id=worker_id,
            execution_id=execution_id,
            job_name=job_name
        )
        
        # Claim the execution
        claimed = await self.job_store.claim_execution(
            execution_id,
            worker_id,
            fencing_token
        )
        
        if not claimed:
            logger.warning(
                "Could not claim execution",
                execution_id=execution_id,
                reason="Already claimed or stale token"
            )
            return
        
        # Get handler
        handler = self.job_registry.get(handler_name)
        if not handler:
            await self.job_store.fail_execution(
                execution_id,
                f"Unknown handler: {handler_name}"
            )
            return
        
        # Start heartbeat
        heartbeat_task = asyncio.create_task(
            self._heartbeat_loop(execution_id, worker_id)
        )
        
        try:
            # Execute with timeout
            await asyncio.wait_for(
                handler(),
                timeout=timeout
            )
            
            # Mark success
            await self.job_store.complete_execution(execution_id)
            
            logger.info(
                "Job completed",
                execution_id=execution_id,
                job_name=job_name
            )
        
        except asyncio.TimeoutError:
            await self.job_store.fail_execution(
                execution_id,
                "Job timed out"
            )
            
            logger.error(
                "Job timed out",
                execution_id=execution_id,
                job_name=job_name,
                timeout=timeout
            )
        
        except Exception as e:
            await self.job_store.fail_execution(
                execution_id,
                str(e)
            )
            
            logger.error(
                "Job failed",
                execution_id=execution_id,
                job_name=job_name,
                error=str(e)
            )
        
        finally:
            heartbeat_task.cancel()
    
    async def _heartbeat_loop(self, execution_id: str, worker_id: str):
        """Send heartbeats while job is running."""
        
        while True:
            try:
                await self.job_store.update_heartbeat(execution_id, worker_id)
                await asyncio.sleep(self.config.heartbeat_interval)
            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.error("Heartbeat error", error=str(e))
    
    async def _run_stuck_job_detector(self):
        """Detect and recover stuck jobs."""
        
        while True:
            try:
                stuck_jobs = await self.job_store.find_stuck_executions(
                    threshold_seconds=self.config.stuck_job_threshold
                )
                
                for execution in stuck_jobs:
                    logger.warning(
                        "Found stuck job",
                        execution_id=execution.id,
                        job_name=execution.job_name,
                        started_at=execution.started_at
                    )
                    
                    # Reset to failed for retry
                    await self.job_store.fail_execution(
                        execution.id,
                        "Job stuck - worker may have crashed"
                    )
                    
                    # Re-queue if retries remaining
                    job = await self.job_store.get_job(execution.job_name)
                    if execution.retry_count < job.max_retries:
                        await self.job_queue.enqueue({
                            'execution_id': execution.id,
                            'job_name': execution.job_name,
                            'handler': job.handler,
                            'timeout': job.timeout,
                            'fencing_token': self.fencing_token,
                            'retry_count': execution.retry_count + 1
                        })
            
            except Exception as e:
                logger.error("Stuck job detector error", error=str(e))
            
            await asyncio.sleep(60)

# Before scheduling, check if already running
existing = await self.job_store.get_running_execution(job_name)
if existing:
    if existing.is_stale():  # No heartbeat for 5+ minutes
        # Assume dead, reschedule
        pass
    else:
        # Still running, skip
        return

class IdempotentEmailJob:
    """Send daily email report - idempotent implementation."""
    
    async def run(self, execution_id: str):
        # Get all users who should receive email
        users = await self.db.get_email_recipients()
        
        for user in users:
            # Check if already sent (idempotency)
            already_sent = await self.db.check_email_sent(
                execution_id=execution_id,
                user_id=user.id
            )
            
            if already_sent:
                continue
            
            # Send email
            await self.email_service.send(user.email, self.report)
            
            # Mark as sent (for idempotency)
            await self.db.mark_email_sent(
                execution_id=execution_id,
                user_id=user.id
            )

# Worker checks fencing token before executing
async def execute_job(self, task):
    current_token = await self.get_current_fencing_token()
    
    if task['fencing_token'] < current_token:
        # Stale task from old leader
        logger.warning("Rejecting stale task", 
            task_token=task['fencing_token'],
            current_token=current_token)
        return
    
    # Safe to execute
    await self._do_execute(task)

class ResilientJobExecution:
    """
    Job execution that handles leader failure gracefully.
    """
    
    def __init__(self, job_store, fencing_store):
        self.job_store = job_store
        self.fencing_store = fencing_store
    
    async def execute_with_resilience(
        self,
        job: JobDefinition,
        execution: JobExecution,
        handler: Callable
    ):
        """
        Execute job with full resilience:
        - Fencing token validation
        - Idempotency support
        - Progress tracking
        - Graceful failure handling
        """
        
        # Step 1: Validate fencing token
        current_token = await self.fencing_store.get_current_token()
        if execution.fencing_token < current_token:
            raise StaleExecutionError(
                f"Execution {execution.id} has stale token"
            )
        
        # Step 2: Create execution context for idempotency
        context = ExecutionContext(
            execution_id=execution.id,
            job_name=job.name,
            checkpoint_store=self.job_store
        )
        
        try:
            # Step 3: Execute with progress tracking
            await handler(context)
            
            # Step 4: Mark complete
            await self.job_store.complete_execution(
                execution.id,
                fencing_token=execution.fencing_token
            )
            
        except Exception as e:
            # Step 5: Handle failure
            await self._handle_failure(execution, e)
            raise
    
    async def _handle_failure(self, execution: JobExecution, error: Exception):
        """Handle job failure."""
        
        # Update execution record
        await self.job_store.fail_execution(
            execution.id,
            str(error),
            fencing_token=execution.fencing_token
        )
        
        # Check if retryable
        job = await self.job_store.get_job(execution.job_name)
        
        if execution.retry_count < job.max_retries:
            # Schedule retry
            await self.job_store.schedule_retry(
                execution.id,
                delay_seconds=self._calculate_backoff(execution.retry_count)
            )
        else:
            # Move to dead letter
            await self.job_store.mark_dead(execution.id)
            await self._alert_dead_job(execution, error)


class ExecutionContext:
    """
    Context for idempotent job execution.
    Provides checkpoint/resume capabilities.
    """
    
    def __init__(self, execution_id: str, job_name: str, checkpoint_store):
        self.execution_id = execution_id
        self.job_name = job_name
        self.checkpoint_store = checkpoint_store
    
    async def checkpoint(self, key: str, data: dict):
        """Save progress checkpoint."""
        await self.checkpoint_store.save_checkpoint(
            self.execution_id,
            key,
            data
        )
    
    async def get_checkpoint(self, key: str) -> Optional[dict]:
        """Get saved checkpoint."""
        return await self.checkpoint_store.get_checkpoint(
            self.execution_id,
            key
        )
    
    async def mark_step_complete(self, step: str):
        """Mark a step as complete (for idempotency)."""
        await self.checkpoint_store.mark_step_complete(
            self.execution_id,
            step
        )
    
    async def is_step_complete(self, step: str) -> bool:
        """Check if step already completed."""
        return await self.checkpoint_store.is_step_complete(
            self.execution_id,
            step
        )

async def generate_monthly_invoices(ctx: ExecutionContext):
    """
    Generate invoices for all customers.
    Resumable and idempotent.
    """
    
    # Get last checkpoint (if resuming)
    checkpoint = await ctx.get_checkpoint("progress")
    last_customer_id = checkpoint.get("last_customer_id") if checkpoint else None
    
    # Get customers to process
    customers = await db.get_customers_for_invoicing(after_id=last_customer_id)
    
    for customer in customers:
        # Check if already processed (idempotency)
        step_key = f"customer_{customer.id}"
        if await ctx.is_step_complete(step_key):
            continue
        
        # Generate invoice
        invoice = await generate_invoice(customer)
        
        # Save invoice
        await db.save_invoice(invoice)
        
        # Send email
        await send_invoice_email(customer, invoice)
        
        # Mark step complete
        await ctx.mark_step_complete(step_key)
        
        # Checkpoint progress
        await ctx.checkpoint("progress", {"last_customer_id": customer.id})
    
    logger.info("Invoice generation complete", 
        execution_id=ctx.execution_id,
        customers_processed=len(customers))

# celeryconfig.py
from celery.schedules import crontab

beat_schedule = {
    'send-daily-report': {
        'task': 'tasks.send_daily_report',
        'schedule': crontab(hour=9, minute=0),
    },
}

# Use Redis for distributed locking
from redbeat import RedBeatSchedulerEntry

# Only one beat will schedule tasks

apiVersion: batch/v1
kind: CronJob
metadata:
  name: daily-report
spec:
  schedule: "0 9 * * *"
  concurrencyPolicy: Forbid  # Don't run if previous still running
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: report
            image: my-report-generator
          restartPolicy: OnFailure

from temporalio import workflow
from datetime import timedelta

@workflow.defn
class DailyReportWorkflow:
    @workflow.run
    async def run(self):
        # This is durable - survives crashes
        await workflow.execute_activity(
            generate_report,
            start_to_close_timeout=timedelta(hours=1)
        )
        await workflow.execute_activity(
            send_emails,
            start_to_close_timeout=timedelta(minutes=30)
        )

# Schedule
schedule = Schedule(
    action=ScheduleActionStartWorkflow(
        DailyReportWorkflow.run,
        id="daily-report",
        task_queue="reports",
    ),
    spec=ScheduleSpec(cron_expressions=["0 9 * * *"]),
)

class OrderProcessingJob:
    """
    Daily job to process pending orders.
    Demonstrates all Week 2 patterns.
    """
    
    def __init__(
        self,
        db,
        payment_client,
        inventory_client,
        webhook_service,
        circuit_breaker,
        idempotency_store
    ):
        self.db = db
        self.payment = payment_client
        self.inventory = inventory_client
        self.webhooks = webhook_service
        self.circuit = circuit_breaker
        self.idempotency = idempotency_store
    
    async def run(self, ctx: ExecutionContext):
        """
        Process all pending orders.
        
        Patterns used:
        - Idempotency: Skip already-processed orders
        - Timeouts: Limit time per order
        - Circuit breaker: Protect external calls
        - Webhooks: Notify external systems
        - Checkpointing: Resume after crash
        """
        
        # Get checkpoint (for resume after crash) [Day 5]
        checkpoint = await ctx.get_checkpoint("progress")
        last_order_id = checkpoint.get("last_order_id") if checkpoint else None
        
        # Get pending orders
        orders = await self.db.get_pending_orders(after_id=last_order_id)
        
        for order in orders:
            # Idempotency check [Day 2]
            idem_key = f"process_order_{order.id}"
            if await self.idempotency.exists(idem_key):
                continue
            
            try:
                # Process with timeout [Day 1]
                await asyncio.wait_for(
                    self._process_order(order, ctx),
                    timeout=30
                )
                
                # Mark idempotent [Day 2]
                await self.idempotency.set(idem_key, ttl=86400)
                
            except asyncio.TimeoutError:
                logger.error("Order processing timeout", order_id=order.id)
                # Don't checkpoint this order: the checkpoint still points at the
                # last *completed* one, so this order is retried on resume
                raise
            
            except CircuitOpenError:
                # Circuit open, stop processing [Day 3]
                logger.warning("Circuit open, pausing job")
                # Checkpoint already points at the last completed order
                raise
            
            # Checkpoint progress [Day 5]
            await ctx.checkpoint("progress", {"last_order_id": order.id})
    
    async def _process_order(self, order, ctx: ExecutionContext):
        """Process single order with all patterns."""
        
        # Step 1: Check inventory (with circuit breaker) [Day 3]
        try:
            available = await self.circuit.call(
                lambda: self.inventory.check(order.items)
            )
        except CircuitOpenError:
            raise  # Propagate to caller
        
        if not available:
            await self.db.mark_order_backordered(order.id)
            return
        
        # Step 2: Charge payment (with timeout) [Day 1]
        try:
            payment_result = await asyncio.wait_for(
                self.payment.charge(
                    order.user_id,
                    order.total,
                    idempotency_key=f"order_{order.id}"  # Day 2
                ),
                timeout=10
            )
        except asyncio.TimeoutError:
            logger.error("Payment timeout", order_id=order.id)
            raise
        
        # Step 3: Reserve inventory
        await self.inventory.reserve(order.items, order.id)
        
        # Step 4: Update order status
        await self.db.mark_order_processing(order.id, payment_result.id)
        
        # Step 5: Send webhook (async) [Day 4]
        await self.webhooks.emit(
            event_type="order.processing",
            data={
                "order_id": order.id,
                "status": "processing",
                "payment_id": payment_result.id
            }
        )