diff --git a/.github/workflows/pr-checks.yml b/.github/workflows/pr-checks.yml index 46cf8bb..e20fe0b 100644 --- a/.github/workflows/pr-checks.yml +++ b/.github/workflows/pr-checks.yml @@ -108,29 +108,7 @@ jobs: - Do not invent labels - If "risk-high" is present, do not include "safe-small-change" - If tests already sufficiently cover the change, do not include "needs-test" - claude_args: | - --json-schema { - "type": "object", - "properties": { - "labels": { - "type": "array", - "items": { - "type": "string", - "enum": [ - "needs-test", - "risk-high", - "needs-human-review", - "safe-small-change" - ] - }, - "uniqueItems": true - }, - "summary": { - "type": "string" - } - }, - "required": ["labels"] - } + claude_args: '--json-schema {"type":"object","properties":{"labels":{"type":"array","items":{"type":"string","enum":["needs-test","risk-high","needs-human-review","safe-small-change"]},"uniqueItems":true},"summary":{"type":"string"}},"required":["labels"]}' - name: Debug structured output env: diff --git a/chronis/__init__.py b/chronis/__init__.py index 7528dd3..53119f6 100644 --- a/chronis/__init__.py +++ b/chronis/__init__.py @@ -49,8 +49,6 @@ scheduler.resume_job("daily-report") scheduler.cancel_job("daily-report") """ - -# Core classes from chronis.adapters.lock import InMemoryLock from chronis.adapters.storage import InMemoryStorage from chronis.core import ( @@ -63,16 +61,12 @@ ) __all__ = [ - # Core "PollingScheduler", "JobInfo", "JobStatus", - # Exceptions "SchedulerError", "JobAlreadyExistsError", "JobNotFoundError", - # Storage Adapters "InMemoryStorage", - # Lock Adapters "InMemoryLock", ] diff --git a/chronis/adapters/lock/memory.py b/chronis/adapters/lock/memory.py index 8013719..f18b213 100644 --- a/chronis/adapters/lock/memory.py +++ b/chronis/adapters/lock/memory.py @@ -55,14 +55,11 @@ def __init__(self) -> None: Logs a warning to remind developers this is not for production use. """ - # Store locks as: lock_key -> (owner_id, expiry_time) self._locks: dict[str, tuple[str, float]] = {} - self._mutex = threading.Lock() # Global lock for _locks dict - # Store conditions for blocking: lock_key -> Condition + self._mutex = threading.Lock() self._conditions: dict[str, threading.Condition] = {} - self.instance_token = str(uuid.uuid4()) # Unique instance identifier + self.instance_token = str(uuid.uuid4()) - # Warn about production usage logger.warning( "InMemoryLock is for testing/development only. " "Use RedisLock or DynamoDBLockAdapter in production." @@ -97,19 +94,15 @@ def acquire( token = owner_id or self.instance_token expiry_time = time.time() + ttl_seconds - # Ensure condition exists with self._mutex: if lock_key not in self._conditions: self._conditions[lock_key] = threading.Condition(self._mutex) condition = self._conditions[lock_key] - # Acquire lock with condition with condition: - # Clean up expired lock self._cleanup_expired_lock(lock_key) - # Try to acquire if lock_key not in self._locks: self._locks[lock_key] = (token, expiry_time) return True @@ -117,25 +110,19 @@ def acquire( if not blocking: return False - # Blocking mode: wait with timeout start_time = time.time() remaining_timeout = timeout while lock_key in self._locks: - # Check timeout if timeout is not None: elapsed = time.time() - start_time remaining_timeout = timeout - elapsed if remaining_timeout <= 0: return False - # Wait for signal (released by release()) condition.wait(timeout=remaining_timeout) - - # Clean up expired lock self._cleanup_expired_lock(lock_key) - # Retry acquisition if lock_key not in self._locks: self._locks[lock_key] = (token, expiry_time) return True @@ -170,15 +157,10 @@ def release(self, lock_key: str, owner_id: str | None = None) -> bool: return False stored_owner, _ = self._locks[lock_key] - - # Verify ownership if stored_owner != token: return False - # Release lock del self._locks[lock_key] - - # Signal one waiting thread condition.notify() return True @@ -208,19 +190,15 @@ def extend( new_expiry_time = time.time() + ttl_seconds with self._mutex: - # Clean up expired lock before extending self._cleanup_expired_lock(lock_key) if lock_key not in self._locks: return False stored_owner, _ = self._locks[lock_key] - - # Verify ownership if stored_owner != token: return False - # Extend TTL self._locks[lock_key] = (stored_owner, new_expiry_time) return True @@ -244,7 +222,6 @@ def reset(self, lock_key: str) -> bool: with condition: if lock_key in self._locks: del self._locks[lock_key] - # Signal all waiting threads condition.notify_all() return True return False diff --git a/chronis/adapters/storage/memory.py b/chronis/adapters/storage/memory.py index 9b6cbd2..e805ac9 100644 --- a/chronis/adapters/storage/memory.py +++ b/chronis/adapters/storage/memory.py @@ -83,14 +83,11 @@ def compare_and_swap_job( current_job = self._jobs[job_id] - # Compare: Check if all expected values match current values for field, expected_value in expected_values.items(): current_value = current_job.get(field) if current_value != expected_value: - # Mismatch - return failure without updating return (False, None) - # Swap: All expectations matched, apply updates self._jobs[job_id].update(updates) # type: ignore[typeddict-item] self._jobs[job_id]["updated_at"] = utc_now().isoformat() # type: ignore[typeddict-item] return (True, self._jobs[job_id].copy()) # type: ignore[return-value] diff --git a/chronis/contrib/adapters/lock/redis.py b/chronis/contrib/adapters/lock/redis.py index 67f6206..39e1e18 100644 --- a/chronis/contrib/adapters/lock/redis.py +++ b/chronis/contrib/adapters/lock/redis.py @@ -112,10 +112,9 @@ def acquire( signal_key = f"{full_key}:signal" token = owner_id or self.instance_token - # Try to acquire lock result = self.redis.set( name=full_key, - value=token, # Store owner token for verification + value=token, nx=True, ex=ttl_seconds, ) @@ -123,15 +122,12 @@ def acquire( if result or not blocking: return bool(result) - # Blocking mode: wait for signal start_time = time.time() remaining_timeout = timeout while True: - # Wait for signal using BLPOP (efficient, no spinloop) result = self.redis.blpop(signal_key, timeout=remaining_timeout or 0) - # Retry acquisition acquired = self.redis.set( name=full_key, value=token, @@ -142,7 +138,6 @@ def acquire( if acquired: return True - # Check timeout if timeout is not None: elapsed = time.time() - start_time remaining_timeout = timeout - elapsed @@ -169,13 +164,12 @@ def release(self, lock_key: str, owner_id: str | None = None) -> bool: signal_key = f"{full_key}:signal" token = owner_id or self.instance_token - # Atomic release with ownership verification using Lua result = self.redis.eval( LUA_RELEASE_SCRIPT, - 2, # num keys - full_key, # KEYS[1] - signal_key, # KEYS[2] - token, # ARGV[1] + 2, + full_key, + signal_key, + token, ) return result == 1 @@ -204,13 +198,12 @@ def extend( full_key = f"{self.key_prefix}{lock_key}" token = owner_id or self.instance_token - # Atomic extend with ownership verification using Lua result = self.redis.eval( LUA_EXTEND_SCRIPT, - 1, # num keys - full_key, # KEYS[1] - token, # ARGV[1] - ttl_seconds, # ARGV[2] + 1, + full_key, + token, + ttl_seconds, ) return result == 1 diff --git a/chronis/contrib/adapters/storage/postgres/adapter.py b/chronis/contrib/adapters/storage/postgres/adapter.py index 21f25e0..8d5e532 100644 --- a/chronis/contrib/adapters/storage/postgres/adapter.py +++ b/chronis/contrib/adapters/storage/postgres/adapter.py @@ -191,16 +191,13 @@ def update_job(self, job_id: str, updates: JobUpdateData) -> JobStorageData: Raises: ValueError: If job not found """ - # Get current data job_data = self.get_job(job_id) if job_data is None: raise ValueError(f"Job {job_id} not found") - # Merge updates job_data.update(updates) # type: ignore[typeddict-item] job_data["updated_at"] = utc_now().isoformat() # type: ignore[typeddict-item] - # Update in database with self.conn.cursor() as cursor: cursor.execute( sql.SQL(""" @@ -248,12 +245,10 @@ def compare_and_swap_job( Raises: ValueError: If job_id not found """ - # Get current data job_data = self.get_job(job_id) if job_data is None: raise ValueError(f"Job {job_id} not found") - # Build WHERE conditions dynamically where_conditions = ["job_id = %s"] where_params: list[Any] = [job_id] @@ -265,17 +260,14 @@ def compare_and_swap_job( where_conditions.append("next_run_time = %s") where_params.append(expected_value) else: - # For other fields, check JSONB data where_conditions.append("data->>%s = %s") where_params.append(field) where_params.append(str(expected_value)) - # Merge updates updated_job_data = job_data.copy() updated_job_data.update(updates) # type: ignore[typeddict-item] updated_job_data["updated_at"] = utc_now().isoformat() # type: ignore[typeddict-item] - # Atomic UPDATE with WHERE conditions with self.conn.cursor() as cursor: where_clause = " AND ".join(where_conditions) query = sql.SQL(""" @@ -307,7 +299,6 @@ def compare_and_swap_job( self.conn.commit() if rows_affected == 0: - # No rows updated - expectations didn't match return (False, None) return (True, updated_job_data) @@ -344,12 +335,10 @@ def query_jobs( params: list[Any] = [] if filters: - # Status filter if "status" in filters: conditions.append("status = %s") params.append(filters["status"]) - # Time filter if "next_run_time_lte" in filters: conditions.append("next_run_time <= %s") params.append(filters["next_run_time_lte"]) @@ -358,22 +347,18 @@ def query_jobs( conditions.append("updated_at <= %s") params.append(filters["updated_at_lte"]) - # Metadata filters (JSONB containment) for key, value in filters.items(): if key.startswith("metadata."): metadata_key = key.replace("metadata.", "") - # Use JSONB @> operator for containment conditions.append("metadata @> %s::jsonb") params.append(json.dumps({metadata_key: value})) - # Build WHERE clause safely where_clause = ( sql.SQL("WHERE {}").format(sql.SQL(" AND ").join(sql.SQL(c) for c in conditions)) if conditions else sql.SQL("") ) - # Build main query with sql.Identifier for table name query = sql.SQL(""" SELECT data FROM {} @@ -384,7 +369,6 @@ def query_jobs( where_clause, ) - # Add LIMIT and OFFSET as parameterized values if limit: query += sql.SQL(" LIMIT %s") params.append(limit) @@ -392,7 +376,6 @@ def query_jobs( query += sql.SQL(" OFFSET %s") params.append(offset) - # Execute query with self.conn.cursor() as cursor: cursor.execute(query, params) rows = cursor.fetchall() diff --git a/chronis/contrib/adapters/storage/postgres/migration.py b/chronis/contrib/adapters/storage/postgres/migration.py index 2ac9243..f2c6c43 100644 --- a/chronis/contrib/adapters/storage/postgres/migration.py +++ b/chronis/contrib/adapters/storage/postgres/migration.py @@ -59,7 +59,6 @@ class MigrationRunner: """Executes database migrations with version tracking.""" HISTORY_TABLE = "chronis_migration_history" - # Advisory lock key for migration synchronization (arbitrary unique number) MIGRATION_LOCK_KEY = 7283946501 def __init__(self, connection: Any, migrations_dir: Path | str) -> None: @@ -144,21 +143,13 @@ def _execute_migration(self, migration: Migration) -> None: import hashlib import time - # Read migration SQL sql_content = migration.filepath.read_text(encoding="utf-8") - - # Calculate checksum for verification checksum = hashlib.sha256(sql_content.encode()).hexdigest() - - # Execute migration start_time = time.time() with self.conn.cursor() as cursor: try: - # Execute the migration SQL cursor.execute(sql_content) - - # Record in history execution_time_ms = int((time.time() - start_time) * 1000) cursor.execute( @@ -220,11 +211,9 @@ def migrate(self, target_version: int | None = None) -> int: try: self._ensure_history_table() - # Get applied and pending migrations applied_versions = self._get_applied_versions() all_migrations = self._discover_migrations() - # Filter pending migrations pending = [ m for m in all_migrations @@ -236,7 +225,6 @@ def migrate(self, target_version: int | None = None) -> int: print("✓ No pending migrations") return 0 - # Execute pending migrations in order print(f"Found {len(pending)} pending migration(s)") for migration in pending: diff --git a/chronis/contrib/adapters/storage/redis.py b/chronis/contrib/adapters/storage/redis.py index d0c0acf..f237eff 100644 --- a/chronis/contrib/adapters/storage/redis.py +++ b/chronis/contrib/adapters/storage/redis.py @@ -121,20 +121,13 @@ def _update_indexes(self, job_data: JobStorageData, pipeline: Any = None) -> Non pipe = pipeline or self.redis - # Add to all jobs set pipe.sadd(self._make_all_jobs_key(), job_id) - # Add to time index (sorted set) score = self._timestamp_to_score(next_run_time) pipe.zadd(self._make_time_index_key(), {job_id: score}) - # Add to status index pipe.sadd(self._make_status_index_key(status), job_id) - if pipeline is None: - # No pipeline provided, changes already executed - pass - def _remove_from_indexes(self, job_id: str, old_status: str | None = None) -> None: """ Remove job from all indexes. @@ -145,17 +138,12 @@ def _remove_from_indexes(self, job_id: str, old_status: str | None = None) -> No """ pipe = self.redis.pipeline() - # Remove from all jobs set pipe.srem(self._make_all_jobs_key(), job_id) - - # Remove from time index pipe.zrem(self._make_time_index_key(), job_id) - # Remove from all status indexes (if old_status unknown, try common ones) if old_status: pipe.srem(self._make_status_index_key(old_status), job_id) else: - # Remove from all possible status indexes for status in ["pending", "scheduled", "running", "paused", "failed"]: pipe.srem(self._make_status_index_key(status), job_id) @@ -179,12 +167,10 @@ def create_job(self, job_data: JobStorageData) -> JobStorageData: job_id = job_data["job_id"] key = self._make_job_key(job_id) - # Atomic check-and-set using HSETNX (returns 1 if created, 0 if exists) created = self.redis.hsetnx(key, "data", self._serialize(dict(job_data))) if not created: raise ValueError(f"Job {job_id} already exists") - # Update indexes (job data is already stored) pipe = self.redis.pipeline() self._update_indexes(job_data, pipeline=pipe) pipe.execute() @@ -256,26 +242,20 @@ def update_job(self, job_id: str, updates: JobUpdateData) -> JobStorageData: if job_data is None: raise ValueError(f"Job {job_id} not found") - # Remember old values for index updates old_status = job_data.get("status") - # Merge updates job_data.update(updates) # type: ignore[typeddict-item] job_data["updated_at"] = utc_now().isoformat() # type: ignore[typeddict-item] - # Use pipeline for atomic updates pipe = self.redis.pipeline() - # Update job data key = self._make_job_key(job_id) pipe.hset(key, "data", self._serialize(dict(job_data))) - # Remove from old status index if status changed new_status = job_data.get("status") if old_status and old_status != new_status: pipe.srem(self._make_status_index_key(old_status), job_id) - # Update indexes self._update_indexes(job_data, pipeline=pipe) pipe.execute() @@ -306,55 +286,42 @@ def compare_and_swap_job( """ key = self._make_job_key(job_id) - # Check if job exists first (before WATCH) job_data = self.get_job(job_id) if job_data is None: raise ValueError(f"Job {job_id} not found") - # Use Redis WATCH for optimistic locking with self.redis.pipeline() as pipe: try: - # Watch the key for changes pipe.watch(key) - # Re-read under WATCH for consistency raw_data = pipe.hget(key, "data") if raw_data is None: pipe.unwatch() raise ValueError(f"Job {job_id} not found") job_data = self._deserialize(raw_data) # type: ignore[assignment] - # Check if expected values match for field, expected_value in expected_values.items(): current_value = job_data.get(field) if current_value != expected_value: - # Mismatch - return failure pipe.unwatch() return (False, None) - # Remember old values for index updates old_status = job_data.get("status") - # Merge updates updated_job_data = job_data.copy() updated_job_data.update(updates) # type: ignore[typeddict-item] updated_job_data["updated_at"] = utc_now().isoformat() # type: ignore[typeddict-item] - # Start transaction pipe.multi() - # Update job data pipe.hset(key, "data", self._serialize(dict(updated_job_data))) - # Remove from old status index if status changed new_status = updated_job_data.get("status") if old_status and old_status != new_status: pipe.srem(self._make_status_index_key(old_status), job_id) - # Update indexes self._update_indexes(updated_job_data, pipeline=pipe) - # Execute transaction pipe.execute() return (True, updated_job_data) @@ -362,7 +329,6 @@ def compare_and_swap_job( except ValueError: raise except Exception: - # Transaction failed (key was modified by another instance) return (False, None) def delete_job(self, job_id: str) -> bool: @@ -375,16 +341,13 @@ def delete_job(self, job_id: str) -> bool: 2. DEL job data 3. Remove from all indexes """ - # Get job to know its status job_data = self.get_job(job_id) old_status = job_data.get("status") if job_data else None - # Delete job data key = self._make_job_key(job_id) result = self.redis.delete(key) if result > 0: - # Remove from indexes self._remove_from_indexes(job_id, old_status) return True @@ -417,19 +380,15 @@ def query_jobs( job_ids: set[str] | None = None if filters: - # Use time index for time queries if "next_run_time_lte" in filters: max_time = filters["next_run_time_lte"] max_score = self._timestamp_to_score(max_time) - # Get all matching job IDs from time index - # (offset/limit applied later after intersection and sorting) time_filtered_ids = self.redis.zrangebyscore( self._make_time_index_key(), min=0, max=max_score ) job_ids = set(time_filtered_ids) if time_filtered_ids else set() - # Use status index for status queries if "status" in filters: status = filters["status"] status_filtered_ids = self.redis.smembers(self._make_status_index_key(status)) @@ -437,19 +396,15 @@ def query_jobs( if job_ids is None: job_ids = set(status_filtered_ids) else: - # Intersection with time filter job_ids &= set(status_filtered_ids) - # If no indexed filters used, get all job IDs if job_ids is None: job_ids = set(self.redis.smembers(self._make_all_jobs_key())) - # Fetch job data for filtered IDs using batch operation job_ids_list = list(job_ids) jobs_dict = self.get_jobs_batch(job_ids_list) jobs = list(jobs_dict.values()) - # Apply client-side filters on fetched results if filters: if "updated_at_lte" in filters: cutoff = filters["updated_at_lte"] @@ -464,10 +419,8 @@ def query_jobs( metadata_key = key.replace("metadata.", "") jobs = [j for j in jobs if j.get("metadata", {}).get(metadata_key) == value] - # Sort by next_run_time jobs.sort(key=lambda j: j.get("next_run_time") or "") - # Apply offset and limit (single place, after all filtering and sorting) if offset: jobs = jobs[offset:] if limit: diff --git a/chronis/core/__init__.py b/chronis/core/__init__.py index ce54bb0..b5b3927 100644 --- a/chronis/core/__init__.py +++ b/chronis/core/__init__.py @@ -11,13 +11,10 @@ __all__ = [ "JobStatus", - # Exceptions "SchedulerError", "JobAlreadyExistsError", "JobNotFoundError", - # Job Management "JobDefinition", "JobInfo", - # Scheduler "PollingScheduler", ] diff --git a/chronis/core/base/scheduler.py b/chronis/core/base/scheduler.py index 99cb9f4..ca1fc04 100644 --- a/chronis/core/base/scheduler.py +++ b/chronis/core/base/scheduler.py @@ -70,29 +70,24 @@ def __init__( self.on_failure = on_failure self.on_success = on_success - # Running state self._running = False - # Job function and callback registries self._job_registry: dict[str, Callable] = {} self._failure_handler_registry: dict[str, OnFailureCallback] = {} self._success_handler_registry: dict[str, OnSuccessCallback] = {} self._registry_lock = threading.RLock() - # Initialize structured logger base_logger = ( logger.logger if isinstance(logger, ContextLogger) else (logger or _default_logger) ) self.logger = ContextLogger(base_logger, {"component": self.__class__.__name__}) - # Initialize execution coordinator (shared across all scheduler types) self._init_execution_coordinator() def _init_execution_coordinator(self) -> None: """Initialize execution coordinator with thread pool.""" from concurrent.futures import ThreadPoolExecutor - # Initialize ThreadPoolExecutor for job execution self._executor = ThreadPoolExecutor( max_workers=self.max_workers, thread_name_prefix="chronis-worker-" ) @@ -147,10 +142,6 @@ def register_job_function(self, name: str, func: Callable) -> None: with self._registry_lock: self._job_registry[name] = func - # ======================================== - # Job CRUD Operations (Common) - # ======================================== - def create_job(self, job: JobDefinition) -> JobInfo: """ Create new job. @@ -164,17 +155,14 @@ def create_job(self, job: JobDefinition) -> JobInfo: Raises: JobAlreadyExistsError: Job already exists """ - # Register on_failure handler if provided if job.on_failure: with self._registry_lock: self._failure_handler_registry[job.job_id] = job.on_failure - # Register on_success handler if provided if job.on_success: with self._registry_lock: self._success_handler_registry[job.job_id] = job.on_success - # Create job in storage job_data = job.to_dict() try: result = self.storage.create_job(job_data) # type: ignore[arg-type] @@ -237,7 +225,6 @@ def update_job( Raises: JobNotFoundError: Job not found """ - # Build updates dictionary updates_dict: dict[str, Any] = { k: v for k, v in { @@ -326,7 +313,6 @@ def _transition_job_state( action: str, ) -> bool: """Validate and apply a job state transition using CAS for atomicity.""" - # Check job exists first job = self.get_job(job_id) if not job: raise JobNotFoundError( @@ -334,7 +320,6 @@ def _transition_job_state( "Use scheduler.query_jobs() to see available jobs." ) - # Try CAS for each allowed status now_str = utc_now().isoformat() for expected_status in allowed: try: @@ -353,7 +338,6 @@ def _transition_job_state( "Use scheduler.query_jobs() to see available jobs." ) from None - # CAS failed for all allowed statuses - re-read for error message current_job = self.get_job(job_id) current_status = current_job.status.value if current_job else "unknown" allowed_str = " or ".join(s.value.upper() for s in allowed) diff --git a/chronis/core/common/exceptions.py b/chronis/core/common/exceptions.py index aa80f1e..c41982e 100644 --- a/chronis/core/common/exceptions.py +++ b/chronis/core/common/exceptions.py @@ -1,11 +1,6 @@ """Custom exceptions for Chronis.""" -# ============================================================================ -# Base Exception Classes -# ============================================================================ - - class SchedulerError(Exception): """Base exception for all scheduler errors.""" @@ -46,11 +41,6 @@ class SchedulerStateError(SchedulerError): pass -# ============================================================================ -# Configuration Errors (Development-time) -# ============================================================================ - - class FunctionNotRegisteredError(SchedulerConfigurationError): """ Function not registered with scheduler. @@ -61,11 +51,6 @@ class FunctionNotRegisteredError(SchedulerConfigurationError): pass -# ============================================================================ -# Lookup Errors (Runtime) -# ============================================================================ - - class JobNotFoundError(SchedulerLookupError): """ Job not found. @@ -94,11 +79,6 @@ class JobAlreadyExistsError(SchedulerLookupError): pass -# ============================================================================ -# State Errors (Runtime) -# ============================================================================ - - class InvalidJobStateError(SchedulerStateError): """ Invalid job state transition. diff --git a/chronis/core/execution/coordinator.py b/chronis/core/execution/coordinator.py index b73e464..3d5b71b 100644 --- a/chronis/core/execution/coordinator.py +++ b/chronis/core/execution/coordinator.py @@ -77,8 +77,6 @@ def try_execute(self, job_data: dict[str, Any], on_complete: Callable[[str], Non job_logger = self.logger.with_context(job_id=job_id, job_name=job_name) - # CAS is the authoritative check; no pre-check needed (avoids TOCTOU) - with self._acquire_lock_context(lock_key) as lock_acquired: if not lock_acquired: return False @@ -133,8 +131,6 @@ def _trigger_execution( ) future.add_done_callback(lambda f: on_complete(job_id)) else: - # For sync jobs with timeout, use threading.Event to coordinate - # between the worker thread and the Timer thread. timed_out = threading.Event() if timeout_seconds else None future = self.executor.submit( @@ -178,10 +174,6 @@ def _on_timeout() -> None: ) raise submit_error - # ------------------------------------------------------------------ - # Execution dispatch (delegates to JobExecutor) - # ------------------------------------------------------------------ - def _execute_in_background( self, job_data: dict[str, Any], @@ -218,10 +210,6 @@ def shutdown_async(self, wait: bool = True) -> None: """Shut down the shared async event loop.""" self._job_executor.shutdown_async(wait) - # ------------------------------------------------------------------ - # Success/failure handlers - # ------------------------------------------------------------------ - def _handle_job_success(self, job_data: dict[str, Any], job_logger: ContextLogger) -> None: """Handle successful job execution (shared by sync and async paths).""" job_id = job_data["job_id"] @@ -232,10 +220,8 @@ def _handle_job_success(self, job_data: dict[str, Any], job_logger: ContextLogge ) if next_status is None: - # DATE job: delete directly self.storage.delete_job(job_id) else: - # Recurring job: combine ALL updates into single storage call updates: JobUpdateData = { "status": next_status.value, "updated_at": now.isoformat(), @@ -287,10 +273,6 @@ def _handle_job_failure( exc_info=True, ) - # ------------------------------------------------------------------ - # CAS and state management - # ------------------------------------------------------------------ - def _try_claim_job_with_cas( self, job_id: str, job_data: dict[str, Any] ) -> tuple[bool, dict[str, Any] | None]: @@ -311,31 +293,25 @@ def _try_claim_job_with_cas( trigger_type = job_data["trigger_type"] current_time_str = utc_now().isoformat() - # Build expected values - job must match these to be claimed expected_values = { "status": JobStatus.SCHEDULED.value, } - # Also verify next_run_time hasn't changed (prevents stale queue entries) queue_next_run = job_data.get("next_run_time") if queue_next_run: - # Only claim if next_run_time is still in the past if queue_next_run > current_time_str: return (False, None) expected_values["next_run_time"] = queue_next_run - # Build updates - what to change if expectations match updates: JobUpdateData = { "status": JobStatus.RUNNING.value, "updated_at": current_time_str, } - # For recurring jobs, optimistically update next_run_time if trigger_type != TriggerType.DATE.value: trigger_args = job_data["trigger_args"] timezone = job_data.get("timezone", "UTC") - # Use scheduled time as base to prevent drift accumulation scheduled_time = job_data.get("next_run_time") base_time = None if scheduled_time: @@ -345,8 +321,7 @@ def _try_claim_job_with_cas( trigger_type, trigger_args, timezone, current_time=base_time ) - # Handle misfire: if next_run_time is in the past, recalculate from now - # Exception: run_all policy keeps incremental time for catch-up execution + # run_all policy keeps incremental time for catch-up execution if utc_time and utc_time <= utc_now() and job_data.get("if_missed") != "run_all": utc_time, local_time = NextRunTimeCalculator.calculate_with_local_time( trigger_type, trigger_args, timezone @@ -357,7 +332,6 @@ def _try_claim_job_with_cas( if local_time: updates["next_run_time_local"] = local_time.isoformat() - # Atomic compare-and-swap operation try: success, updated_job = self.storage.compare_and_swap_job( job_id, expected_values, updates @@ -367,11 +341,7 @@ def _try_claim_job_with_cas( result["_original_scheduled_time"] = queue_next_run return (True, result) return (False, None) - except ValueError: - # Job not found (deleted) - return (False, None) except Exception: - # Other errors - don't claim job (expected in distributed environments) return (False, None) def _update_job_status(self, job_id: str, status: JobStatus) -> None: @@ -384,10 +354,6 @@ def _update_job_status(self, job_id: str, status: JobStatus) -> None: }, ) - # ------------------------------------------------------------------ - # Callbacks - # ------------------------------------------------------------------ - def _invoke_success_callback(self, job_id: str, job_data: dict[str, Any]) -> None: """Invoke job-specific and global success handlers.""" handler = self.success_handler_registry.get(job_id) @@ -424,10 +390,6 @@ def _safe_invoke(self, handler: Any, label: str, job_id: str, *args: Any) -> Non exc_info=True, ) - # ------------------------------------------------------------------ - # Retry and lock management - # ------------------------------------------------------------------ - def _schedule_retry( self, job_data: dict[str, Any], next_retry_count: int, job_logger: ContextLogger ) -> None: diff --git a/chronis/core/execution/job_executor.py b/chronis/core/execution/job_executor.py index 5f0992e..00552fd 100644 --- a/chronis/core/execution/job_executor.py +++ b/chronis/core/execution/job_executor.py @@ -26,7 +26,6 @@ def __init__( self.function_registry = function_registry self.logger = logger - # Shared event loop for async job execution (lazy-initialized) self._async_loop: asyncio.AbstractEventLoop | None = None self._async_thread: threading.Thread | None = None self._async_lock = threading.Lock() @@ -89,7 +88,6 @@ def ensure_async_loop(self) -> asyncio.AbstractEventLoop: """Get or create the shared event loop for async jobs (thread-safe).""" if self._async_loop is None or self._async_loop.is_closed(): with self._async_lock: - # Double-checked locking to avoid redundant loop creation if self._async_loop is None or self._async_loop.is_closed(): self._async_loop = asyncio.new_event_loop() self._async_thread = threading.Thread( diff --git a/chronis/core/execution/job_queue.py b/chronis/core/execution/job_queue.py index 66e3e2d..7ce7700 100644 --- a/chronis/core/execution/job_queue.py +++ b/chronis/core/execution/job_queue.py @@ -39,19 +39,11 @@ def __init__(self, max_queue_size: int = 100) -> None: """ self.max_queue_size = max_queue_size - # Pending queue: job IDs waiting to be executed (Priority Queue) - # Items are tuples: (negative_priority, sequence, job_id) self._pending_queue: PriorityQueue[tuple[int, int, str]] = PriorityQueue( maxsize=max_queue_size ) - - # Sequence counter for FIFO ordering within same priority self._sequence_counter = 0 - - # In-flight jobs: job IDs dequeued and sent for execution (thread-safe set) self._in_flight_jobs: set[str] = set() - - # All tracked job IDs (pending + in-flight) for duplicate prevention self._known_job_ids: set[str] = set() self._lock = threading.RLock() @@ -112,10 +104,7 @@ def get_next_job(self) -> str | None: Job ID or None if queue is empty """ try: - # Dequeue tuple: (negative_priority, sequence, job_id) _priority, _sequence, job_id = self._pending_queue.get_nowait() - - # Mark as in-flight (keep in _known_job_ids) with self._lock: self._in_flight_jobs.add(job_id) diff --git a/chronis/core/jobs/definition.py b/chronis/core/jobs/definition.py index f87715e..669681f 100644 --- a/chronis/core/jobs/definition.py +++ b/chronis/core/jobs/definition.py @@ -25,14 +25,12 @@ class JobDefinition(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) - # Required fields job_id: str name: str trigger_type: TriggerType trigger_args: dict[str, Any] func: Callable | str - # Scheduling options timezone: str = "UTC" args: tuple | None = None kwargs: dict[str, Any] | None = None @@ -40,7 +38,6 @@ class JobDefinition(BaseModel): next_run_time: datetime | None = None metadata: dict[str, Any] | None = None - # Execution options on_failure: "OnFailureCallback | None" = None on_success: "OnSuccessCallback | None" = None max_retries: int = 0 @@ -158,18 +155,13 @@ class JobInfo: metadata: dict[str, Any] created_at: datetime updated_at: datetime - # Retry information max_retries: int = 0 retry_delay_seconds: int = 60 retry_count: int = 0 - # Timeout information timeout_seconds: int | None = None - # Priority information priority: int = 5 - # Misfire information if_missed: str | None = None misfire_threshold_seconds: int = 60 - # Execution history last_run_time: datetime | None = None last_scheduled_time: datetime | None = None diff --git a/chronis/core/schedulers/fluent_builders.py b/chronis/core/schedulers/fluent_builders.py index 5c83178..90ff918 100644 --- a/chronis/core/schedulers/fluent_builders.py +++ b/chronis/core/schedulers/fluent_builders.py @@ -66,10 +66,6 @@ def __init__(self, scheduler: PollingScheduler) -> None: self._if_missed: Literal["skip", "run_once", "run_all"] | None = None self._misfire_threshold_seconds: int = 60 - # ======================================== - # Trigger Methods (3 APIs) - # ======================================== - def every( self, seconds: int | None = None, @@ -191,10 +187,6 @@ def once(self, when: str | datetime) -> FluentJobBuilder: self._trigger_params = {"run_date": when} return self - # ======================================== - # Config Method - # ======================================== - def config( self, *, @@ -262,10 +254,6 @@ def config( return self - # ======================================== - # Run Method - # ======================================== - def run( self, func: Callable | str, diff --git a/chronis/core/schedulers/polling_scheduler.py b/chronis/core/schedulers/polling_scheduler.py index 9272689..3374902 100644 --- a/chronis/core/schedulers/polling_scheduler.py +++ b/chronis/core/schedulers/polling_scheduler.py @@ -131,7 +131,6 @@ def __init__( Raises: ValueError: If parameters are invalid """ - # Validate polling-specific parameters if polling_interval_seconds < self.MIN_POLLING_INTERVAL: raise ValueError(f"polling_interval_seconds must be >= {self.MIN_POLLING_INTERVAL}") if polling_interval_seconds > self.MAX_POLLING_INTERVAL: @@ -144,7 +143,6 @@ def __init__( "to prevent premature lock expiration" ) - # Initialize base scheduler super().__init__( storage_adapter=storage_adapter, lock_adapter=lock_adapter, @@ -156,21 +154,14 @@ def __init__( on_success=on_success, ) - # Polling-specific configuration self.polling_interval_seconds = polling_interval_seconds self.max_queue_size = max_queue_size or (self.max_workers * 5) self.executor_interval_seconds = executor_interval_seconds or max( 1, polling_interval_seconds / 2 ) - # Initialize job queue for backpressure control self._job_queue = JobQueue(max_queue_size=self.max_queue_size) - - # Random node ID for distributed job ordering - # Each scheduler instance gets a unique ID to reduce lock contention self._node_id = str(uuid.uuid4()) - - # Initialize periodic runner for polling and execution tasks self._runner = _PeriodicRunner() def start(self) -> None: @@ -192,12 +183,9 @@ def start(self) -> None: "Scheduler is already running. Call scheduler.stop() first to restart." ) - # Register periodic tasks self._runner = _PeriodicRunner() self._runner.add_task(self._execute_queued_jobs, self.executor_interval_seconds, "executor") self._runner.add_task(self._enqueue_jobs, self.polling_interval_seconds, "polling") - - # Start periodic runner (non-blocking, daemon threads) self._runner.start() self._running = True @@ -226,13 +214,8 @@ def stop(self) -> dict[str, Any]: "was_running": False, } - # Shutdown periodic runner (stops polling for new jobs) self._runner.shutdown(wait=True) - - # Wait for async jobs on shared event loop self._execution_coordinator.shutdown_async(wait=True) - - # Shutdown thread pool executor (waits for sync jobs) self._executor.shutdown(wait=True, cancel_futures=False) self._running = False @@ -258,10 +241,6 @@ def get_queue_status(self) -> dict[str, Any]: """ return self._job_queue.get_status() - # ------------------------------------------------------------------------ - # Internal Methods (Polling Logic) - # ------------------------------------------------------------------------ - def _recover_stuck_jobs(self) -> None: """Detect and recover jobs stuck in RUNNING state after process crash.""" try: @@ -338,14 +317,11 @@ def _get_ready_jobs(self, limit: int | None = None) -> list[Any]: """ current_time = utc_now() - # Query jobs that are ready (next_run_time <= current_time) - # Over-fetch by 1.5x to account for node-specific hash reordering - # which may push some jobs beyond the limit boundary after re-sorting + # Over-fetch to account for node-specific hash reordering after re-sorting query_limit = int(limit * 1.5 + 1) if limit else None filters = {"status": "scheduled", "next_run_time_lte": current_time.isoformat()} jobs = self.storage.query_jobs(filters=filters, limit=query_limit) - # Classify into normal and misfired jobs normal_jobs, misfired_jobs = MisfireDetector.classify_due_jobs( jobs, current_time.isoformat() ) @@ -357,28 +333,23 @@ def _get_ready_jobs(self, limit: int | None = None) -> list[Any]: job_ids=[j.get("job_id") for j in misfired_jobs], ) - # Apply misfire policy ready_misfired = [] for job in misfired_jobs: policy = job.get("if_missed", "run_once") if policy == "skip": self._skip_misfired_job(job) else: - # run_once / run_all: include in ready jobs ready_misfired.append(job) - # Combine all jobs all_jobs = normal_jobs + ready_misfired - # Sort by priority (descending), then by node-specific hash, then by next_run_time - # The hash ensures each scheduler instance processes jobs in a different order, + # Hash ensures each scheduler instance processes jobs in different order, # reducing lock contention in distributed environments def sort_key(job: Any) -> tuple[int, int, str]: priority = job.get("priority", 5) job_id = job.get("job_id", "") next_run = job.get("next_run_time", "") - # Create node-specific hash to distribute jobs across schedulers combined = f"{self._node_id}:{job_id}" hash_val = int(hashlib.md5(combined.encode()).hexdigest()[:8], 16) @@ -386,7 +357,6 @@ def sort_key(job: Any) -> tuple[int, int, str]: all_jobs.sort(key=sort_key) # type: ignore[arg-type] - # Apply limit if specified if limit is not None: all_jobs = all_jobs[:limit] @@ -440,11 +410,9 @@ def _execute_queued_jobs(self) -> None: for concurrent execution, improving throughput in distributed environments. """ try: - # Process jobs in batches for better concurrency - batch_size = 100 # Process up to 100 jobs per iteration (increased for throughput) + batch_size = 100 while not self._job_queue.is_empty(): - # Get batch of job IDs from queue job_ids = [] for _ in range(batch_size): if self._job_queue.is_empty(): @@ -456,34 +424,24 @@ def _execute_queued_jobs(self) -> None: if not job_ids: break - # Fetch all job data in single batch operation (optimized!) jobs_dict = self.storage.get_jobs_batch(job_ids) - # Process batch: submit for execution - # Each try_execute() is non-blocking and submits to thread pool for job_id in job_ids: job_data = jobs_dict.get(job_id) if job_data is None: - # Job was deleted - mark as completed and continue self._job_queue.mark_completed(job_id) continue - # Try to execute with distributed lock (non-blocking) executed = self._execution_coordinator.try_execute( dict(job_data), on_complete=self._job_queue.mark_completed ) - # If not executed (lock failed or wrong status), mark as completed in queue if not executed: self._job_queue.mark_completed(job_id) except Exception as e: self.logger.error("Executor error", error=str(e)) - # ======================================== - # Simplified Public API (TriggerType hidden) - # ======================================== - def create_interval_job( self, func: Callable | str, @@ -655,10 +613,6 @@ def create_date_job( ) ) - # ======================================== - # Fluent Builder API (Simplified Interface) - # ======================================== - def every( self, seconds: int | None = None, diff --git a/chronis/core/triggers/cron_calc.py b/chronis/core/triggers/cron_calc.py index d52c1c0..74d5380 100644 --- a/chronis/core/triggers/cron_calc.py +++ b/chronis/core/triggers/cron_calc.py @@ -12,10 +12,6 @@ from calendar import monthrange from datetime import datetime, timedelta, tzinfo -# --------------------------------------------------------------------------- -# Constants -# --------------------------------------------------------------------------- - WEEKDAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"] MONTHS = [ "jan", @@ -77,10 +73,6 @@ def _datetime_ceil(dateval: datetime) -> datetime: return dateval.replace(second=0, microsecond=0) -# --------------------------------------------------------------------------- -# Expressions -# --------------------------------------------------------------------------- - class AllExpression: value_re = re.compile(r"\*(?:/(?P\d+))?$") @@ -215,10 +207,6 @@ def get_next_value(self, date: datetime, field: BaseField) -> int | None: return monthrange(date.year, date.month)[1] -# --------------------------------------------------------------------------- -# Fields -# --------------------------------------------------------------------------- - class BaseField: REAL = True @@ -298,11 +286,6 @@ class MonthField(BaseField): } -# --------------------------------------------------------------------------- -# Main calculation -# --------------------------------------------------------------------------- - - def _build_fields( *, year: str | int | None = None, @@ -437,7 +420,6 @@ def calculate_next_cron_time( ) tz = current.tzinfo - # Handle folded datetimes (DST fall-back) if current.fold == 1: current = datetime.fromisoformat(current.isoformat()) + timedelta(microseconds=1) diff --git a/chronis/type_defs.py b/chronis/type_defs.py index f5cb167..98205cb 100644 --- a/chronis/type_defs.py +++ b/chronis/type_defs.py @@ -2,13 +2,11 @@ from typing import Any, Literal, NotRequired, TypedDict -# Type aliases for string literal types type TriggerTypeStr = Literal["interval", "cron", "date"] type JobStatusStr = Literal["pending", "scheduled", "running", "paused", "failed"] type MisfirePolicyStr = Literal["skip", "run_once", "run_all"] -# Storage adapter data format class JobStorageData(TypedDict): """Type definition for job data stored in storage adapters.""" @@ -26,22 +24,17 @@ class JobStorageData(TypedDict): metadata: dict[str, Any] created_at: str updated_at: str - # Retry configuration max_retries: int retry_delay_seconds: int retry_count: int - # Timeout configuration timeout_seconds: int | None - # Priority configuration priority: int - # Misfire configuration if_missed: MisfirePolicyStr misfire_threshold_seconds: int last_run_time: str | None last_scheduled_time: str | None -# Trigger args by type class IntervalTriggerArgs(TypedDict, total=False): """Type definition for interval trigger arguments.""" @@ -70,7 +63,6 @@ class DateTriggerArgs(TypedDict): run_date: str -# Query filter dictionaries class JobQueryFilter(TypedDict, total=False): """Type definition for job query filters.""" @@ -78,8 +70,6 @@ class JobQueryFilter(TypedDict, total=False): trigger_type: TriggerTypeStr next_run_time_lte: str next_run_time_gte: str - # Metadata filters are dynamic: "metadata.{key}": value - # So we can't type them strictly here class JobUpdateData(TypedDict, total=False): diff --git a/chronis/utils/logging.py b/chronis/utils/logging.py index 2a2f651..2a33a94 100644 --- a/chronis/utils/logging.py +++ b/chronis/utils/logging.py @@ -17,17 +17,14 @@ def setup_logger(name: str = "scheduler", level: int = logging.INFO) -> logging. """ logger = logging.getLogger(name) logger.setLevel(level) - logger.propagate = False # Prevent duplicate logs to root logger + logger.propagate = False - # Prevent duplicate handlers if logger.handlers: return logger - # Console handler handler = logging.StreamHandler() handler.setLevel(level) - # Structured format (time, level, context, message) formatter = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - [%(context)s] - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", @@ -75,5 +72,4 @@ def with_context(self, **context: Any) -> "ContextLogger": return ContextLogger(self.logger, {**self.context, **context}) -# Default logger _default_logger = setup_logger()