Add Redis-based lock for distributed scheduler coordination

[peaktwilight]

Summary

• In-memory Scheduler had no cross-process coordination, so running multiple API or worker replicas caused every scheduled tick to fire once per instance.
• Added DistributedLock (Redis SET NX EX) and wired it into Scheduler.tick() keyed by (job_name, tick_bucket) so only the first instance to acquire the key runs the callback; the rest skip that tick.
• Losing instances advance their local last_run so they don't retry in a tight loop within the same window. TTL defaults to 60s — longer than typical playbook runs, short enough that a crashed holder doesn't stall the next interval.
• No Celery Beat migration: keeps the existing loop, adds coordination on top.

Closes #111

Test plan

• pytest tests/test_scheduler.py — 14 passing, including a race test that runs two schedulers concurrently against a shared fake Redis and asserts exactly one executes.
• ruff check src/ tests/ clean.
• CI green across all jobs.

Summary by CodeRabbit

• New Features

  • Scheduler now supports distributed multi-instance deployments with automatic lock-based execution coordination, preventing duplicate scheduled job runs across processes.

• Tests

  • Added comprehensive test coverage for distributed scheduler locking, concurrent instance behavior, and lock expiration scenarios.

[coderabbitai]

📝 Walkthrough

Walkthrough

The changes add Redis-backed distributed locking to the scheduler to prevent duplicate job executions across multiple process instances. The scheduler now attempts to acquire per-tick locks before executing callbacks, skipping execution if lock acquisition fails while still advancing job state.

Changes

Cohort / File(s)	Summary
Distributed Locking Infrastructure src/opensoar/core/scheduler.py	Introduced RedisLikeClient protocol, DistributedLock class for Redis-based lock coordination, and DEFAULT_LOCK_TTL_SECONDS constant. Added _tick_bucket() and _lock_key() helpers to compute tick windows and derive lock identifiers from wall-clock time and job intervals.
Scheduler Lock Integration src/opensoar/core/scheduler.py	Extended Scheduler.__init__ to accept optional lock, instance_id, and lock_ttl_seconds parameters. Modified tick() control flow to attempt lock acquisition when a job is due; skips callback execution on lock contention but still advances last_run and records last_tick_id. Added job registration tracking via last_tick_id field.
Distributed Lock Testing tests/test_scheduler.py	Added FakeRedisBackend emulator supporting Redis set(nx, ex), get, and delete operations. Introduced unit tests for DistributedLock.acquire() and multi-instance scheduler scenarios validating single execution across concurrent ticks, lock expiry behavior, and state advancement on skipped executions.

Sequence Diagram

sequenceDiagram
    participant S1 as Scheduler Instance 1
    participant S2 as Scheduler Instance 2
    participant DL as DistributedLock
    participant Redis as Redis

    rect rgba(100, 150, 200, 0.5)
        Note over S1,S2: Concurrent Tick (same interval window)
    end
    
    par S1 → Redis
        S1->>DL: acquire(lock_key, ttl=60)
        DL->>Redis: SET lock_key "1" NX EX 60
        Redis-->>DL: OK (success)
        DL-->>S1: true
        S1->>S1: execute callback()
        S1->>S1: update last_run, last_tick_id
    and S2 → Redis
        S2->>DL: acquire(lock_key, ttl=60)
        DL->>Redis: SET lock_key "1" NX EX 60
        Redis-->>DL: nil (exists)
        DL-->>S2: false
        S2->>S2: skip callback()
        S2->>S2: advance last_run, record last_tick_id
    end

    rect rgba(100, 150, 200, 0.5)
        Note over S1,S2: Lock expires after ~60 seconds
    end
    
    S1->>DL: acquire(lock_key, ttl=60)
    DL->>Redis: SET lock_key "1" NX EX 60
    Redis-->>DL: OK (lock expired)
    DL-->>S1: true
    S1->>S1: execute callback() again

Loading

Estimated Code Review Effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

Poem

🐰 The scheduler hops with locks in hand,
Across the Redis lands so grand!
No duplicate runs will ever be,
One job per tick, for all to see. ✨

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 45.45% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately and specifically describes the main change: adding Redis-based distributed locking to the scheduler for multi-instance coordination.
Linked Issues check	✅ Passed	The PR directly addresses issue #111 by implementing Redis-based distributed locking to prevent duplicate scheduled job executions across multiple instances.
Out of Scope Changes check	✅ Passed	All changes are scoped to implementing distributed scheduler locking via Redis; no out-of-scope modifications to unrelated functionality were introduced.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch fix/scheduler-distributed-lock

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

🧹 Nitpick comments (2)

src/opensoar/core/scheduler.py (2)

134-149: Wall-clock tick bucket relies on reasonable clock synchronization.

The design correctly uses wall-clock time for distributed coordination (vs. monotonic for local elapsed checks). Worth noting: if server clocks drift by more than a few seconds, concurrent instances could compute different buckets and both execute. This is acceptable for most deployments with NTP, but consider adding a brief note about the clock sync assumption in the module docstring if this scheduler will be used in environments with unreliable time sync.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/opensoar/core/scheduler.py` around lines 134 - 149, Add a short note to
the module docstring explaining that _tick_bucket and _lock_key rely on
wall-clock time and therefore assume reasonable clock synchronization (e.g.,
NTP) across hosts; state that clock drift beyond the tick interval can cause
multiple instances to compute different buckets and potentially run the same job
concurrently, and recommend ensuring time sync or using an alternative
coordination mechanism for unreliable-time environments.

---

185-191: Callback failure leaves job in retriable state.

When the callback raises an exception, last_run is not updated. Combined with the held lock, the effective behavior is:

• While lock TTL remains: subsequent tick() calls skip (lock denied) and update last_run
• After TTL expires: job becomes eligible for re-execution

This is reasonable for retry semantics, but be aware that a consistently failing job could retry indefinitely. Consider whether you want to update last_run even on failure to enforce the interval between attempts, or add retry limits/backoff for persistent failures in a follow-up.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@src/opensoar/core/scheduler.py` around lines 185 - 191, The issue: when
job["callback"] raises, job["last_run"] is left unchanged so the job can be
retried immediately after lock TTL; fix by ensuring last_run is updated on
failure (and optionally track failures for backoff/retries). Concretely, inside
the tick() job execution block where job["callback"] is awaited and in the
except Exception: handler (the block that calls logger.exception(f"Scheduler:
job '{name}' failed")), set job["last_run"] = time.monotonic() and
job["last_tick_id"] = tick_id so the interval is enforced; additionally consider
adding/initializing job["failure_count"] and incrementing it in the except
handler (and use it elsewhere to implement max_retries/backoff) instead of
leaving retry behavior unbounded.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Nitpick comments:
In `@src/opensoar/core/scheduler.py`:
- Around line 134-149: Add a short note to the module docstring explaining that
_tick_bucket and _lock_key rely on wall-clock time and therefore assume
reasonable clock synchronization (e.g., NTP) across hosts; state that clock
drift beyond the tick interval can cause multiple instances to compute different
buckets and potentially run the same job concurrently, and recommend ensuring
time sync or using an alternative coordination mechanism for unreliable-time
environments.
- Around line 185-191: The issue: when job["callback"] raises, job["last_run"]
is left unchanged so the job can be retried immediately after lock TTL; fix by
ensuring last_run is updated on failure (and optionally track failures for
backoff/retries). Concretely, inside the tick() job execution block where
job["callback"] is awaited and in the except Exception: handler (the block that
calls logger.exception(f"Scheduler: job '{name}' failed")), set job["last_run"]
= time.monotonic() and job["last_tick_id"] = tick_id so the interval is
enforced; additionally consider adding/initializing job["failure_count"] and
incrementing it in the except handler (and use it elsewhere to implement
max_retries/backoff) instead of leaving retry behavior unbounded.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 1172a568-3a51-4e87-a744-a9963d1438c1

📥 Commits

Reviewing files that changed from the base of the PR and between 1dbc05a and 94594e3.

📒 Files selected for processing (2)

• src/opensoar/core/scheduler.py
• tests/test_scheduler.py