Reject leases that expire while waiting for the SQLite writer lock

.github/workflows/codeql.yml

@@ -0,0 +1,28 @@
+name: CodeQL security analysis
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+  schedule:
+    - cron: '18 3 * * 2'
+
+permissions:
+  contents: read
+  actions: read
+  security-events: write
+
+jobs:
+  analyze:
+    name: Python security analysis
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+      - uses: github/codeql-action/init@v4
+        with:
+          languages: python
+          queries: security-extended
+      - uses: github/codeql-action/analyze@v4
+        with:
+          category: /language:python

CHANGELOG.md

@@ -1,3 +1,11 @@
+## 0.1.3 — 2026-10-15
+
+- Sample lease-validation time after acquiring the SQLite write lock. ACK, heartbeat and failure operations now reject a lease that expires during lock contention.
+- Calculate claim intervals, producer delays and operator requeue timestamps after lock acquisition; refresh claim time during expired-job cleanup.
+- Add six real-lock regression scenarios: each fails on the preceding implementation and passes with this fix.
+- Add Python CodeQL analysis alongside the existing correctness, packaging and benchmark CI.
+- No schema migration or change to the single-host, at-least-once contract.
+
 ## 0.1.2 — 2026-10-14
 
 - Fix claim starvation when more than 100 expired terminal leases had priority over runnable work.

README.md

@@ -95,6 +95,8 @@ flowchart LR
     DB -->|max retries| DLQ[Dead-letter jobs]
 ```
 
+Read the [lease expiry during writer contention analysis](docs/notes/lease-expiry-after-lock.md).
+
 Read the [100-expired-leases failure analysis](docs/notes/expired-lease-cleanup-progress.md).
 
 Read [correctness and design decisions](docs/architecture.md), [why SQLite](docs/adr/0001-sqlite.md), and the technical deep dive [why a lease needs fencing tokens](docs/notes/why-fencing-tokens.md). The audit history is persisted in the same transaction as state transitions.
@@ -112,7 +114,7 @@ completes once in that no-fault run. A separate test **abruptly terminates a
 worker process** after a claim and verifies the expired job is reclaimed with a
 new fence. These tests **do not imply exactly-once external side effects**.
 
-GitHub Actions runs the suite (32 tests, including a seeded 640-operation reference model) against Python 3.11, 3.12 and 3.13: [view CI](https://github.com/harshitethic/LeaseLedger/actions/workflows/ci.yml).
+GitHub Actions runs the suite (40 tests, including a seeded 640-operation reference model) against Python 3.11, 3.12 and 3.13: [view CI](https://github.com/harshitethic/LeaseLedger/actions/workflows/ci.yml).
 
 ## Online backup and recovery
 

docs/architecture.md

@@ -16,7 +16,7 @@ stateDiagram-v2
 ### Invariants
 
 1. **Single active ownership in the queue:** `BEGIN IMMEDIATE` serializes claim transactions. Exactly one committed claim can return a given lease version. Older owners have stale fencing numbers.
-2. **Stale acknowledgements do not mutate state:** heartbeat/complete/fail check `job_id`, `owner`, current `fence`, `status='leased'`, and `lease_until > now`.
+2. **Stale acknowledgements do not mutate state:** heartbeat/complete/fail check `job_id`, `owner`, current `fence`, `status='leased'`, and `lease_until > now`, with `now` sampled after acquiring the write lock.
 3. **Crash recovery:** a leased job becomes selectable when `lease_until <= now`. If max attempts were already used, it transitions to `dead` instead.
 4. **Idempotent producer requests:** `dedupe_key` is UNIQUE, and insert+existing ID lookup occurs in one transaction. Key lifetime is the entire retained record, including dead/done.
 5. **Observable transitions:** event insert and state change share a transaction. Interrupted operations roll back, so there is no committed state without its journal entry.
@@ -29,7 +29,7 @@ stateDiagram-v2
 | Worker crashes while processing | Job is reclaimable after lease expiry; new fence |
 | Old worker resumes after reclaim | Old fence rejected by complete/fail/heartbeat |
 | Handler returns success, then worker crashes before ack | Possible duplicate external effect; use downstream idempotency |
-| DB is busy due to another writer | SQLite busy timeout (15 seconds); retry at application boundary |
+| DB is busy due to another writer | SQLite busy timeout (15 seconds); retry at application boundary. A lease that expires during the wait is rejected after acquisition |
 | Job exceeds max attempts | Persistent dead-letter record and event |
 | Producer retries same dedupe key | Existing ID returned; payload is not silently replaced |
 | Worker handler takes longer than lease | Background heartbeat extends the lease; failure is detectable |
@@ -38,7 +38,7 @@ stateDiagram-v2
 
 - SQLite WAL lets readers access persisted data while a writer commits; there is still only **one writer** at a time.
 - `jobs_ready` and `jobs_leases` are indexed for candidate selection. Load-test the actual active/expired mix at scale.
-- The claim transaction may transition up to 100 expired/maxed jobs to dead-letter; the next claim resumes. This avoids an unbounded transaction and still drains poisoned records over repeated calls.
+- Each claim transaction may transition up to 100 expired/maxed jobs to dead-letter. The same call commits that batch and continues selection; readiness time is refreshed for each candidate. This bounds each transaction, not the total claim call.
 - Backoff is `min(max_delay, base_delay * 2 ** (attempts - 1))`, bounded to avoid overflow.
 - Claims are not reservations of CPU time: worker and handler scheduling are OS-controlled.
 - Leases use a stored Unix timestamp, not a monotonic timer, so restarting processes preserves expiration, at the cost of clock-skew sensitivity.

docs/notes/lease-expiry-after-lock.md

@@ -0,0 +1,62 @@
+# Lease expiry must be checked after the writer wait
+
+## Reproduced defect
+
+In v0.1.2, `complete`, `heartbeat` and `fail` sampled the wall clock before
+`BEGIN IMMEDIATE`. Another SQLite writer could hold that lock while the
+lease expired. Once the lock became available, the operation validated the
+lease against its old timestamp and incorrectly mutated the job and journal.
+A heartbeat could revive an expired lease even without a replacement worker.
+
+The same ordering shortened newly granted leases, hid delayed jobs that
+became ready while waiting, and started producer delays before their enqueue
+transaction acquired the lock. The maximum SQLite busy wait is 15 seconds,
+so this matters especially for short leases and contended local databases.
+
+## Fix and contract
+
+Sample the clock inside the acquired write transaction. Claim selection
+refreshes its clock for each cleanup candidate, including after a committed
+100-item cleanup batch. Heartbeat, acknowledgement and failure checks still
+require the current owner, fence, leased status and a strictly future expiry.
+Rejected operations append no event and leave the job unchanged.
+
+The reference point is validation within the write transaction, not the
+instant the caller receives a reply. A process pause or slow commit can still
+consume the remaining lease interval. Wall-clock adjustments remain relevant;
+this is not a hard real-time deadline or exactly-once external-effects guarantee.
+Use appropriately sized leases and downstream idempotency.
+
+## Executable evidence
+
+```bash
+python -m unittest tests.test_lock_wait_expiry -v
+python -m unittest discover -s tests -v
+python -m benchmarks.queue_bench --jobs 80 --workers 4
+```
+
+The regression opens a separate real SQLite connection and acquires its write
+lock. A second thread attempts the queue operation. SQLite's trace callback
+signals the attempted `BEGIN IMMEDIATE`; the test advances an injected clock
+and releases the blocker. No sleep is used to guess whether contention began.
+Only time is controlled: SQLite locking, SQL execution and transaction commits
+are real.
+
+Six scenarios fail on base commit `fd5a99a` and pass after the fix:
+
+- Expired ACK, heartbeat and failure are refused without changing job or events.
+- A delayed job that becomes ready during contention is claimed.
+- A new claim receives its requested interval after lock acquisition.
+- Enqueue timestamps and delays start after lock acquisition.
+
+The rejected-operation cases also reclaim the job with a new fence and
+successfully acknowledge it, proving that rejection preserves recovery.
+
+The complete 40-test suite and 80-job/four-process benchmark smoke passed on
+Kali with Python 3.12.14 and macOS with Python 3.14.7. The smoke checks job accounting; it is not a capacity
+or latency benchmark. The cloud editing environment blocked the existing
+multiprocessing manager's Unix socket, so full local verification used the
+user-authorized device instead of skipping that test or weakening it.
+
+No database schema change or data migration is required. Existing databases
+remain readable. Rolling back the code reintroduces the expiry defect.

leaseledger/__init__.py

@@ -4,4 +4,4 @@
 from .worker import Worker, WorkerStats
 
 __all__ = ["Store", "Lease", "EnqueueResult", "Worker", "WorkerStats"]
-__version__ = "0.1.2"
+__version__ = "0.1.3"

leaseledger/store.py

@@ -154,8 +154,8 @@ def enqueue(
         ):
             raise ValueError("Invalid enqueue options")
         text = self._encode(payload)
-        now = self._now()
         with self._write() as db:
+            now = self._now()
             cursor = db.execute(
                 """INSERT INTO jobs
                    (topic,payload,status,priority,available_at,max_attempts,dedupe_key,created_at,updated_at)
@@ -177,12 +177,15 @@ def claim(self, owner: str, *, lease_seconds: float = 30, topic: str | None = No
             raise ValueError("Invalid lease parameters")
         if topic is not None and not 1 <= len(topic) <= 100:
             raise ValueError("Invalid topic")
-        now = self._now()
         while True:
             with self._write() as db:
                 # Process at most 100 expired final-attempt leases per transaction.
                 # Commit each batch; do not report queue empty after 100 tombstones.
                 for _ in range(100):
+                    # Lock acquisition can wait up to busy_timeout. Sample
+                    # readiness/expiry after that wait, and again after each
+                    # cleanup item, rather than granting an already-old lease.
+                    now = self._now()
                     predicate = "" if topic is None else " AND topic = :topic"
                     row = db.execute(
                         f"""SELECT * FROM jobs WHERE
@@ -238,8 +241,9 @@ def _valid_lease_update(
     def heartbeat(self, lease: Lease, *, extend_seconds: float = 30) -> bool:
         if not math.isfinite(extend_seconds) or not 1 <= extend_seconds <= 86400:
             raise ValueError("Invalid heartbeat interval")
-        now = self._now()
         with self._write() as db:
+            # Validate expiry only after acquiring the SQLite writer lock.
+            now = self._now()
             ok = self._valid_lease_update(
                 db,
                 lease,
@@ -252,8 +256,8 @@ def heartbeat(self, lease: Lease, *, extend_seconds: float = 30) -> bool:
             return ok
 
     def complete(self, lease: Lease) -> bool:
-        now = self._now()
         with self._write() as db:
+            now = self._now()
             ok = self._valid_lease_update(
                 db,
                 lease,
@@ -273,9 +277,9 @@ def fail(self, lease: Lease, reason: str, *, base_delay: float = 1.0, max_delay:
             or max_delay < base_delay
         ):
             raise ValueError("Invalid backoff settings")
-        now = self._now()
         reason = reason[:500]
         with self._write() as db:
+            now = self._now()
             row = db.execute(
                 """SELECT attempts,max_attempts FROM jobs WHERE id=? AND status='leased'
                    AND lease_owner=? AND fence=? AND lease_until>?""",
@@ -294,8 +298,8 @@ def fail(self, lease: Lease, reason: str, *, base_delay: float = 1.0, max_delay:
             return True
 
     def requeue_dead(self, job_id: int) -> bool:
-        now = self._now()
         with self._write() as db:
+            now = self._now()
             changed = (
                 db.execute(
                     """UPDATE jobs SET status='queued',attempts=0,available_at=?,

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "leaseledger"
-version = "0.1.2"
+version = "0.1.3"
 description = "Durable SQLite task queue with fenced leases, retries, and audit events"
 readme = "README.md"
 requires-python = ">=3.11"

tests/test_lock_wait_expiry.py

@@ -0,0 +1,92 @@
+"""A real SQLite writer wait must not freeze lease-validation time."""
+
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import closing, contextmanager
+from pathlib import Path
+import sqlite3
+import tempfile
+import threading
+import unittest
+
+from leaseledger import Store
+from tests.test_store import MutableClock
+
+
+class ObservedStore(Store):
+    @contextmanager
+    def _connection(self):
+        with super()._connection() as connection:
+            # SQLite calls this immediately before executing BEGIN IMMEDIATE.
+            # The other connection already holds the write lock at that point.
+            connection.set_trace_callback(
+                lambda sql: self.write_started.set() if sql == "BEGIN IMMEDIATE" else None
+            )
+            yield connection
+
+
+class LockWaitExpiryTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.path = Path(self.temp.name) / "jobs.db"
+        self.clock = MutableClock()
+        self.queue = ObservedStore(self.path, clock=self.clock)
+        self.queue.write_started = threading.Event()
+
+    def after_lock_wait(self, operation, *, elapsed=10):
+        self.queue.write_started.clear()
+        with closing(sqlite3.connect(self.path, isolation_level=None)) as blocker:
+            blocker.execute("BEGIN IMMEDIATE")
+            with ThreadPoolExecutor(max_workers=1) as executor:
+                future = executor.submit(operation)
+                try:
+                    self.assertTrue(self.queue.write_started.wait(timeout=3), "writer never attempted BEGIN")
+                    self.assertFalse(future.done(), "operation bypassed the held SQLite write lock")
+                    self.clock.advance(elapsed)
+                finally:
+                    blocker.rollback()
+                return future.result(timeout=3)
+
+    def assert_expired_operation_refused(self, action):
+        job = self.queue.enqueue("work", {})
+        lease = self.queue.claim("worker", lease_seconds=10)
+        before = self.queue.get(job.job_id)
+        events = self.queue.events(job.job_id)
+        self.assertFalse(self.after_lock_wait(lambda: action(lease)))
+        self.assertEqual(self.queue.get(job.job_id), before)
+        self.assertEqual(self.queue.events(job.job_id), events)
+        replacement = self.queue.claim("replacement")
+        self.assertEqual(replacement.fence, lease.fence + 1)
+        self.assertTrue(self.queue.complete(replacement))
+
+    def test_ack_refuses_lease_expired_while_waiting_for_writer(self):
+        self.assert_expired_operation_refused(self.queue.complete)
+
+    def test_heartbeat_cannot_revive_lease_expired_during_writer_wait(self):
+        self.assert_expired_operation_refused(self.queue.heartbeat)
+
+    def test_failure_cannot_requeue_lease_expired_during_writer_wait(self):
+        self.assert_expired_operation_refused(lambda lease: self.queue.fail(lease, "temporary"))
+
+    def test_claim_sees_job_becoming_ready_during_writer_wait(self):
+        job = self.queue.enqueue("delayed", {}, delay=5)
+        lease = self.after_lock_wait(lambda: self.queue.claim("worker", lease_seconds=30))
+        self.assertIsNotNone(lease)
+        self.assertEqual(lease.job_id, job.job_id)
+        self.assertEqual(lease.expires_at, self.clock.value + 30)
+        self.assertEqual(self.queue.events(job.job_id)[-1]["occurred_at"], self.clock.value)
+
+    def test_claim_grants_full_interval_after_lock_acquisition(self):
+        self.queue.enqueue("work", {})
+        lease = self.after_lock_wait(lambda: self.queue.claim("worker", lease_seconds=5))
+        self.assertEqual(lease.expires_at, self.clock.value + 5)
+        self.assertTrue(self.queue.complete(lease))
+
+    def test_enqueue_delay_starts_after_lock_acquisition(self):
+        job = self.after_lock_wait(lambda: self.queue.enqueue("work", {}, delay=5))
+        row = self.queue.get(job.job_id)
+        self.assertEqual(row["created_at"], self.clock.value)
+        self.assertEqual(row["available_at"], self.clock.value + 5)
+        self.assertIsNone(self.queue.claim("worker"))
+        self.clock.advance(5)
+        self.assertEqual(self.queue.claim("worker").job_id, job.job_id)