fix(worker): fence leases and align grpc runtimes

Author: ysalitrynskyi

CHANGELOG.md

@@ -2,6 +2,23 @@
 
 All notable changes to the Spooled Python SDK are documented here.
 
+## [1.0.21] - 2026-07-12
+
+### Fixed
+
+- Aligned the optional gRPC runtime and generator floors with the checked-in
+  generated stubs (`grpcio`/`grpcio-tools` 1.80.0+, protobuf 6.33.5+) and
+  regenerated the stubs and lockfile.
+- Worker and gRPC registration defaults now use the package's canonical version
+  instead of reporting `1.0.0`; source-only imports retain a synchronized
+  `1.0.21` fallback when installed package metadata is unavailable.
+- Sync and async workers now bind heartbeat, settlement, and cleanup to the exact
+  active execution, so a stale lease cannot borrow, settle, remove, or cancel a
+  replacement execution for the same job ID.
+- Added regression coverage for bidirectional stream lease fencing, async
+  heartbeat lease serialization, unary complete/fail/renew wire fields, and
+  generated-stub imports at the declared minimum dependency versions.
+
 ## [1.0.20] - 2026-07-11
 
 ### Added

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "spooled"
-version = "1.0.20"
+version = "1.0.21"
 description = "Official Python SDK for Spooled Cloud job queue service"
 readme = "README.md"
 license = "Apache-2.0"
@@ -41,12 +41,12 @@ dependencies = [
 
 [project.optional-dependencies]
 grpc = [
-    "grpcio>=1.60.0",
-    "grpcio-tools>=1.60.0",
-    # Security floor: CVE-2026-0994 (JSON recursion depth bypass) is fixed in
-    # protobuf 5.29.6 (and 6.33.5 on the 6.x line); older floors permitted
-    # vulnerable releases.
-    "protobuf>=5.29.6",
+    # Keep runtime floors aligned with the checked-in generated stubs.
+    "grpcio>=1.80.0",
+    "grpcio-tools>=1.80.0",
+    # Generated protobuf code requires 6.31.1+. Use the patched 6.x security
+    # floor for CVE-2026-0994 rather than permitting vulnerable 6.x releases.
+    "protobuf>=6.33.5",
 ]
 realtime = [
     "websockets>=12.0",
@@ -73,6 +73,9 @@ Documentation = "https://spooled.cloud/docs/"
 Repository = "https://github.com/spooled-cloud/spooled-sdk-python"
 Changelog = "https://github.com/spooled-cloud/spooled-sdk-python/blob/main/CHANGELOG.md"
 
+[tool.hatch.build]
+artifacts = ["src/spooled/grpc/stubs"]
+
 [tool.hatch.build.targets.wheel]
 packages = ["src/spooled"]
 

scripts/generate_grpc.sh

@@ -19,11 +19,19 @@ python -m grpc_tools.protoc \
     --pyi_out="$OUTPUT_DIR" \
     "$PROTO_DIR/spooled.proto"
 
-# Create __init__.py
-cat > "$OUTPUT_DIR/__init__.py" << 'EOF'
-"""Generated gRPC stubs."""
-from spooled.grpc.stubs.spooled_pb2 import *
-from spooled.grpc.stubs.spooled_pb2_grpc import *
-EOF
+# grpcio-tools generates a top-level sibling import; rewrite it for this package.
+python - "$OUTPUT_DIR/spooled_pb2_grpc.py" << 'PY'
+from pathlib import Path
+import sys
+
+path = Path(sys.argv[1])
+content = path.read_text()
+content = content.replace(
+    "import spooled_pb2 as spooled__pb2",
+    "from spooled.grpc.stubs import spooled_pb2 as spooled__pb2",
+)
+path.write_text(content)
+PY
+
 
 echo "gRPC stubs generated successfully in $OUTPUT_DIR"

src/spooled/__init__.py

@@ -13,6 +13,7 @@
     >>> print(f"Created job: {result.id}")
 """
 
+from spooled._version import __version__
 from spooled.config import (
     CircuitBreakerConfig,
     RetryConfig,
@@ -35,15 +36,6 @@
     is_spooled_error,
 )
 
-# Version — resolved from the installed package metadata so it never drifts from
-# the released version (the previous hardcoded literal lagged the real release).
-try:
-    from importlib.metadata import version as _pkg_version
-
-    __version__ = _pkg_version("spooled")
-except Exception:  # pragma: no cover - source/editable checkout without metadata
-    __version__ = "1.0.20"
-
 # These will be imported when the client modules are created
 __all__ = [
     # Version

src/spooled/_version.py

@@ -0,0 +1,8 @@
+"""Canonical access to the installed Spooled SDK package version."""
+
+from importlib.metadata import PackageNotFoundError, version
+
+try:
+    __version__ = version("spooled")
+except PackageNotFoundError:  # pragma: no cover - source checkout without installed metadata
+    __version__ = "1.0.21"

src/spooled/grpc/client.py

@@ -18,6 +18,8 @@
 from google.protobuf import struct_pb2, timestamp_pb2
 from pydantic import BaseModel, Field
 
+from spooled._version import __version__
+
 # Optional gRPC imports
 try:
     import grpc
@@ -179,7 +181,7 @@ class GrpcRegisterWorkerRequest(BaseModel):
     hostname: str
     worker_type: str = "python"
     max_concurrency: int = Field(default=5, ge=1, le=100)
-    version: str = "1.0.0"
+    version: str = __version__
     metadata: dict[str, str] | None = None
 
     model_config = {"extra": "forbid"}
@@ -899,7 +901,7 @@ def register(
         *,
         worker_type: str = "python",
         max_concurrency: int = 5,
-        version: str = "1.0.0",
+        version: str = __version__,
         metadata: dict[str, str] | None = None,
     ) -> GrpcRegisterWorkerResponse:
         """

src/spooled/worker/async_worker.py

@@ -13,6 +13,7 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
+from spooled._version import __version__
 from spooled.types.jobs import ClaimedJob
 from spooled.worker.types import (
     AsyncJobContext,
@@ -75,7 +76,7 @@ def __init__(
         shutdown_timeout: float = 30.0,
         hostname: str | None = None,
         worker_type: str = "python",
-        version: str = "1.0.0",
+        version: str = __version__,
         metadata: dict[str, Any] | None = None,
     ) -> None:
         """
@@ -116,6 +117,7 @@ def __init__(
         self._worker_heartbeat_task: asyncio.Task[None] | None = None
         self._shutdown_event: asyncio.Event | None = None
         self._semaphore: asyncio.Semaphore | None = None
+        self._active_jobs_lock = asyncio.Lock()
 
         # Event handlers
         self._event_handlers: dict[WorkerEvent, list[Callable[..., Any]]] = {}
@@ -289,7 +291,7 @@ async def stop(self) -> None:
             for job_id, active in list(self._active_jobs.items()):
                 if self._debug:
                     self._debug(f"Force-failing job {job_id} due to shutdown timeout", None)
-                await self._fail_job(active.job, "Worker shutdown timeout")
+                await self._fail_job(active, "Worker shutdown timeout")
 
         # Deregister worker
         if self._worker_id:
@@ -352,14 +354,15 @@ async def _process_job(self, job: ClaimedJob) -> None:
                 abort_event=abort_event,
             )
 
-            self._active_jobs[job.id] = active
+            async with self._active_jobs_lock:
+                self._active_jobs[job.id] = active
 
             # Start per-job heartbeat
             heartbeat_interval = (
                 self._options.lease_duration * self._options.heartbeat_fraction
             )
             active.heartbeat_task = asyncio.create_task(
-                self._job_heartbeat_loop(job.id, heartbeat_interval)
+                self._job_heartbeat_loop(active, heartbeat_interval)
             )
 
             # Execute handler
@@ -394,7 +397,7 @@ async def _execute_handler(self, active: ActiveJob) -> None:
                 return
 
             # Complete the job
-            await self._complete_job(job, result)
+            await self._complete_job(active, result)
 
         except Exception as e:
             # Check if aborted
@@ -404,25 +407,30 @@ async def _execute_handler(self, active: ActiveJob) -> None:
                 return
 
             error_message = str(e)
-            await self._fail_job(job, error_message)
+            await self._fail_job(active, error_message)
 
         finally:
-            await self._cleanup_job(job.id)
-
-    async def _complete_job(self, job: ClaimedJob, result: dict[str, Any] | None) -> None:
-        """Complete a job."""
-        if not self._worker_id:
-            return
+            await self._cleanup_job(active)
 
+    async def _complete_job(self, active: ActiveJob, result: dict[str, Any] | None) -> None:
+        """Complete an active execution only while it still owns the job slot."""
+        job = active.job
         try:
+            async with self._active_jobs_lock:
+                worker_id = self._worker_id
+                if self._active_jobs.get(job.id) is not active or not worker_id:
+                    return
             params: dict[str, Any] = {
-                "worker_id": self._worker_id,
+                "worker_id": worker_id,
                 "result": result,
             }
             if job.lease_id is not None:
                 params["lease_id"] = job.lease_id
             await self._client.jobs.complete(job.id, params)
 
+            async with self._active_jobs_lock:
+                if self._active_jobs.get(job.id) is not active:
+                    return
             self._emit("job:completed", JobCompletedEventData(
                 job_id=job.id,
                 queue_name=job.queue_name,
@@ -433,22 +441,27 @@ async def _complete_job(self, job: ClaimedJob, result: dict[str, Any] | None) ->
             if self._debug:
                 self._debug(f"Failed to complete job {job.id}: {e}", None)
 
-    async def _fail_job(self, job: ClaimedJob, error_message: str) -> None:
-        """Fail a job."""
-        if not self._worker_id:
-            return
-
+    async def _fail_job(self, active: ActiveJob, error_message: str) -> None:
+        """Fail an active execution only while it still owns the job slot."""
+        job = active.job
         will_retry = job.retry_count < job.max_retries
 
         try:
+            async with self._active_jobs_lock:
+                worker_id = self._worker_id
+                if self._active_jobs.get(job.id) is not active or not worker_id:
+                    return
             params: dict[str, Any] = {
-                "worker_id": self._worker_id,
+                "worker_id": worker_id,
                 "error": error_message,
             }
             if job.lease_id is not None:
                 params["lease_id"] = job.lease_id
             await self._client.jobs.fail(job.id, params)
 
+            async with self._active_jobs_lock:
+                if self._active_jobs.get(job.id) is not active:
+                    return
             self._emit("job:failed", JobFailedEventData(
                 job_id=job.id,
                 queue_name=job.queue_name,
@@ -460,36 +473,42 @@ async def _fail_job(self, job: ClaimedJob, error_message: str) -> None:
             if self._debug:
                 self._debug(f"Failed to fail job {job.id}: {e}", None)
 
-    async def _cleanup_job(self, job_id: str) -> None:
-        """Clean up after job completion."""
-        active = self._active_jobs.pop(job_id, None)
-        if active and active.heartbeat_task:
+    async def _cleanup_job(self, active: ActiveJob) -> None:
+        """Clean up only the exact execution that finished."""
+        async with self._active_jobs_lock:
+            if self._active_jobs.get(active.job.id) is active:
+                del self._active_jobs[active.job.id]
+        if active.heartbeat_task and active.heartbeat_task is not asyncio.current_task():
             active.heartbeat_task.cancel()
             try:
                 await active.heartbeat_task
             except asyncio.CancelledError:
                 pass
 
-    async def _job_heartbeat_loop(self, job_id: str, interval: float) -> None:
-        """Job heartbeat loop."""
-        while job_id in self._active_jobs and self._worker_id:
+    async def _job_heartbeat_loop(self, active: ActiveJob, interval: float) -> None:
+        """Send heartbeats bound to one immutable execution."""
+        job = active.job
+        while True:
             await asyncio.sleep(interval)
 
-            active = self._active_jobs.get(job_id)
-            if active is None or not self._worker_id:
-                break
-
             try:
+                async with self._active_jobs_lock:
+                    worker_id = self._worker_id
+                    if self._active_jobs.get(job.id) is not active or not worker_id:
+                        break
                 params: dict[str, Any] = {
-                    "worker_id": self._worker_id,
+                    "worker_id": worker_id,
                     "lease_duration_secs": self._options.lease_duration,
                 }
-                if active.job.lease_id is not None:
-                    params["lease_id"] = active.job.lease_id
-                await self._client.jobs.heartbeat(job_id, params)
+                if job.lease_id is not None:
+                    params["lease_id"] = job.lease_id
+                await self._client.jobs.heartbeat(job.id, params)
+                async with self._active_jobs_lock:
+                    if self._active_jobs.get(job.id) is not active:
+                        break
             except Exception as e:
                 if self._debug:
-                    self._debug(f"Job heartbeat failed for {job_id}: {e}", None)
+                    self._debug(f"Job heartbeat failed for {job.id}: {e}", None)
 
     async def _worker_heartbeat_loop(self, interval: float) -> None:
         """Worker heartbeat loop."""