Replace instance_idx with leader leasing

config/config.default.yaml

@@ -5,7 +5,6 @@
 # Edit the appropriate config file instead.
 debug: false # Run server with increased compatibility for breakpoint debugging.
 instance_env: dev # Instance environment. Used in Sentry, userAgent of subqueries, instance-appropriate behavior, etc.
-instance_idx: 0 # Instance index. Use when multiple Retriever instances are run, so a leader can be determined.
 max_request_size: 2147483648 # Maximum request body size in bytes, post-decompression. Larger bodies are rejected with 413.
 log_level: DEBUG # Level of application logs to print/keep.
 host: 0.0.0.0 # Uvicorn listen host.
@@ -55,6 +54,7 @@ redis:
   shutdown_timeout: 3 # Time in seconds to wait for batched tasks to finish before force-quitting.
   heartbeat_interval_seconds: 60 # Cadence at which workers / background / main re-register their process entry.
   process_ttl_seconds: 300 # TTL on each registered process entry; set to several heartbeat intervals so a single missed refresh doesn't drop the entry.
+  leader_lease_ttl_seconds: 180 # TTL on the build-leader lease, renewed each heartbeat interval; set to several intervals so a transient renewal failure doesn't trigger a leadership handoff.
 mongo:
   host: localhost
   port: 27017

src/retriever/background.py

@@ -16,6 +16,7 @@
 from retriever.lookup.subclass import SubclassMapping
 from retriever.metadata.optable import OpTableManager
 from retriever.utils.general import tolerate_init
+from retriever.utils.leader import LEADER_ELECTION
 from retriever.utils.logs import add_mongo_sink
 from retriever.utils.mongo import MongoClient, MongoQueue
 from retriever.utils.orphan_detection import periodically_mark_orphans
@@ -50,15 +51,18 @@ async def _background_async() -> None:
         role_label="Background",
     )
 
-    # The leader doesn't see query traffic, so opt into periodic backend pings.
+    # The builder doesn't see query traffic, so opt into periodic backend pings.
     tier_manager.enable_periodic_healthchecks()
     await tier_manager.initialize_drivers()
     metakg_manager = OpTableManager()
-    metakg_manager.promote_to_leader()
+    metakg_manager.promote_to_builder()
     await tolerate_init("OpTable build", metakg_manager.initialize())
     subclass_manager = SubclassMapping()
-    subclass_manager.promote_to_leader()
+    subclass_manager.promote_to_builder()
     await tolerate_init("Subclass map build", subclass_manager.initialize())
+    # Managers have registered their on_acquire hooks; contend for the build lease
+    # so exactly one instance drives the builds across the shared Redis.
+    await tolerate_init("Leader election", LEADER_ELECTION.start())
     orphan_task = asyncio.create_task(
         periodically_mark_orphans(), name="orphan-detection"
     )
@@ -79,6 +83,9 @@ async def _background_async() -> None:
     with contextlib.suppress(asyncio.CancelledError):
         await orphan_task
     # Heartbeat task lives in RedisClient().tasks; cancelled in its wrapup.
+    # Relinquish the build lease first so a peer can take over promptly, and
+    # while Redis is still up to release it (rather than waiting on TTL).
+    await LEADER_ELECTION.stop()
     await SubclassMapping().wrapup()
     await metakg_manager.wrapup()
     await RedisClient().wrapup()

src/retriever/config/general.py

@@ -74,6 +74,12 @@ class RedisSettings(BaseModel):
             description="TTL on each registered process entry; set to several heartbeat intervals so a single missed refresh doesn't drop the entry."
         ),
     ] = 300
+    leader_lease_ttl_seconds: Annotated[
+        int,
+        Field(
+            description="TTL on the build-leader lease, renewed each heartbeat interval. Set to several intervals to avoid touchy handoffs."
+        ),
+    ] = 180
 
 
 class MongoSettings(BaseModel):
@@ -346,12 +352,6 @@ class GeneralConfig(CommentedSettings):
             description="Instance environment. Used in Sentry, userAgent of subqueries, instance-appropriate behavior, etc."
         ),
     ] = "dev"
-    instance_idx: Annotated[
-        int,
-        Field(
-            description="Instance index. Use when multiple Retriever instances are run, so a leader can be determined."
-        ),
-    ] = 0
     max_request_size: Annotated[
         int,
         Field(

src/retriever/lookup/subclass.py

@@ -12,13 +12,14 @@
     CURIE,
 )
 from retriever.utils.general import BatchedAction
+from retriever.utils.leader import LEADER_ELECTION
 from retriever.utils.redis import SUBCLASS_META_KEY, TIER_RECOVERED_CHANNEL, RedisClient
 
 REDIS_CLIENT = RedisClient()
 
 MAPPING_ID = "SubclassHashMap"
 MAPPING_BUILD_ID = f"{MAPPING_ID}:next"
-"""Temp key the leader builds into, then atomically renames onto `MAPPING_ID`."""
+"""Temp key the builder builds into, then atomically renames onto `MAPPING_ID`."""
 
 
 class SubclassMapping(BatchedAction):
@@ -30,23 +31,23 @@ class SubclassMapping(BatchedAction):
     flush_time: float = 0
     multibatch: bool = True
 
-    is_leader: bool = False
+    is_builder: bool = False
     subscriptions: dict[CURIE, list[Callable[[list[CURIE] | None], None]]]
     _refresh_lock: asyncio.Lock
     _pending_refresh: bool = False
 
     redis_setup_batch_size: int = 5000
 
     def __init__(self) -> None:
-        """Initialize without leader role; call `promote_to_leader()` to flip the flag."""
+        """Initialize without builder role; call `promote_to_builder()` to flip the flag."""
         self.subscriptions = {}
         self._refresh_lock = asyncio.Lock()
         self._pending_refresh = False
         super().__init__()
 
-    def promote_to_leader(self) -> None:
-        """Flip this instance to leader mode. Must be called before `initialize()`."""
-        self.is_leader = True
+    def promote_to_builder(self) -> None:
+        """Flip this process to builder mode. Must be called before `initialize()`."""
+        self.is_builder = True
 
     @override
     async def initialize(self) -> None:
@@ -55,12 +56,15 @@ async def initialize(self) -> None:
             logger.info("Implicit subclassing disabled, skipping initialization.")
             return await super().initialize()
 
-        if not self.is_leader:  # Only need leader to update the redis setup
+        if not self.is_builder:  # Only the builder updates the redis setup
             return await super().initialize()
 
         if self.initialized:
             return  # rebuild loop already running
 
+        # Rebuild as soon as this instance wins the build lease.
+        LEADER_ELECTION.on_acquire(self.refresh)
+
         try:
             await self.refresh()
             self.tasks.append(asyncio.create_task(self.rebuild()))
@@ -72,7 +76,7 @@ async def initialize(self) -> None:
         REDIS_CLIENT.on_recover(self.refresh)
         tier_manager.get_driver(1).on_recover(self.refresh)
         # Also listen for worker-detected tier 1 recovery via Redis so
-        # the rebuild fires faster than the leader's own periodic ping.
+        # the rebuild fires faster than the builder's own periodic ping.
         with contextlib.suppress(Exception):
             await REDIS_CLIENT.subscribe(
                 TIER_RECOVERED_CHANNEL, self._on_remote_tier_recover
@@ -82,8 +86,8 @@ async def initialize(self) -> None:
 
     @override
     async def wrapup(self) -> None:
-        """Unsubscribe the leader's tier-recovery listener, then cancel the rebuild loop."""
-        if self.is_leader:
+        """Unsubscribe the builder's tier-recovery listener, then cancel the rebuild loop."""
+        if self.is_builder:
             with contextlib.suppress(Exception):
                 await REDIS_CLIENT.unsubscribe(
                     TIER_RECOVERED_CHANNEL, self._on_remote_tier_recover
@@ -98,6 +102,10 @@ async def _on_remote_tier_recover(self, message: str) -> None:
 
     async def refresh(self) -> None:
         """Rebuild and publish the subclass mapping; concurrent calls collapse to one trailing rebuild."""
+        # Only the leader writes the shared mapping, so instances don't race on
+        # the temp build key.
+        if not LEADER_ELECTION.is_leader:
+            return
         self._pending_refresh = True
         if self._refresh_lock.locked():
             logger.debug(

src/retriever/metadata/optable.py

@@ -37,6 +37,7 @@
 from retriever.utils import biolink
 from retriever.utils.biolink import expand
 from retriever.utils.general import AsyncDaemon
+from retriever.utils.leader import LEADER_ELECTION
 from retriever.utils.redis import (
     OP_TABLE_KEY,
     OP_TABLE_META_KEY,
@@ -97,30 +98,30 @@ class OpTableManager(AsyncDaemon):
     update_lock: asyncio.Lock
     _refresh_lock: asyncio.Lock
     _pending_refresh: bool = False
-    is_leader: bool = False
+    is_builder: bool = False
 
     def __init__(self) -> None:
-        """Initialize without leader role; call `promote_to_leader()` to flip the flag."""
+        """Initialize without builder role; call `promote_to_builder()` to flip the flag."""
         self.update_lock = asyncio.Lock()
         self._refresh_lock = asyncio.Lock()
         self._pending_refresh = False
         super().__init__()
 
-    def promote_to_leader(self) -> None:
-        """Flip this instance to leader mode. Must be called before `initialize()`."""
-        self.is_leader = True
+    def promote_to_builder(self) -> None:
+        """Flip this process to builder mode. Must be called before `initialize()`."""
+        self.is_builder = True
 
     @override
     def get_task_funcs(self) -> list[Callable[[], Coroutine[None, None, None]]]:
         tasks = list[Callable[[], Coroutine[None, None, None]]]()
-        if self.is_leader and CONFIG.job.metakg.build_time > -1:
+        if self.is_builder and CONFIG.job.metakg.build_time > -1:
             tasks.append(self.periodic_build_op_table)
         return tasks
 
     @override
     async def initialize(self) -> None:
         """Start the appropriate tasks for a given process."""
-        if self.is_leader:
+        if self.is_builder:
             # Register hooks before the initial refresh so a startup
             # against a down dependency still recovers later.
             REDIS_CLIENT.on_recover(self.refresh)
@@ -130,6 +131,8 @@ async def initialize(self) -> None:
                 await REDIS_CLIENT.subscribe(
                     TIER_RECOVERED_CHANNEL, self._on_remote_tier_recover
                 )
+            # Rebuild as soon as this instance wins the build lease.
+            LEADER_ELECTION.on_acquire(self.refresh)
             try:
                 await self.refresh()
             except Exception:
@@ -147,7 +150,7 @@ async def initialize(self) -> None:
             for tier_idx in range(0, 2):
                 driver = tier_manager.get_driver(tier_idx)
                 driver.on_recover(self._on_tier_recover)
-                # Tell the leader so it rebuilds without waiting on its periodic ping.
+                # Tell the builder so it rebuilds without waiting on its periodic ping.
                 driver.on_recover(self._make_remote_publisher(tier_idx))
         return await super().initialize()
 
@@ -165,7 +168,7 @@ async def _publish() -> None:
         return _publish
 
     async def _on_remote_tier_recover(self, _message: str) -> None:
-        """Leader-side subscriber callback for cross-process tier recovery."""
+        """Builder-side subscriber callback for cross-process tier recovery."""
         await self.refresh()
 
     async def refresh(self) -> None:
@@ -217,7 +220,7 @@ async def degraded_local_build(self) -> None:
     @override
     async def wrapup(self) -> None:
         """Cancel running tasks so connections can close."""
-        if self.is_leader:
+        if self.is_builder:
             with contextlib.suppress(Exception):
                 await REDIS_CLIENT.unsubscribe(
                     TIER_RECOVERED_CHANNEL, self._on_remote_tier_recover
@@ -374,7 +377,9 @@ async def _collect_tier_ops(self, *, bypass_cache: bool = False) -> OperationTab
 
     async def build_operation_table(self) -> None:
         """Build Retriever's internal OperationTable and store it to Redis."""
-        if CONFIG.instance_idx != 0:
+        # Build+publish only when this builder's instance is the elected leader;
+        # workers (is_builder False) may still build on demand via get_op_table.
+        if self.is_builder and not LEADER_ELECTION.is_leader:
             return
 
         logger.info("Building Operation Table...")
@@ -386,9 +391,9 @@ async def build_operation_table(self) -> None:
         logger.success(
             f"Built Operation Table containing {len(op_table.operations_flat)} operations / {len(op_table.nodes)} nodes."
         )
-        # The leader never reads _operation_table back - it only exists to push to
+        # The builder never reads _operation_table back - it only exists to push to
         # Redis. Drop the reference so the snapshot doesn't sit in process memory.
-        if self.is_leader:
+        if self.is_builder:
             async with self.update_lock:
                 self._operation_table = None
 
@@ -428,7 +433,7 @@ async def get_op_table(self) -> OperationTable:
                     op_table = self._operation_table
                 if op_table is not None:
                     return op_table
-                if not REDIS_CLIENT.up and not self.is_leader:
+                if not REDIS_CLIENT.up and not self.is_builder:
                     # Worker can't pull the published copy; build from
                     # available tiers and re-check.
                     await self.degraded_local_build()

src/retriever/status.py

@@ -350,17 +350,26 @@ async def status_root() -> StatusSnapshot:
     redis_mem_r: object = None
     metakg_r: object = None
     subclass_r: object = None
+    leader_r: object = None
+    leader_elected_r: object = None
     if redis_client.up:
+        # gather()'s typed tuple overloads stop at 6 awaitables; past that it
+        # returns list[...], so route through object before the tuple cast.
         redis_results = cast(
-            tuple[object, object, object, object, object, object],
-            await asyncio.gather(
-                redis_client.list_main(),
-                redis_client.list_background(),
-                redis_client.list_workers(),
-                redis_client.used_memory_bytes(),
-                redis_client.metakg_freshness(),
-                redis_client.subclass_freshness(),
-                return_exceptions=True,
+            tuple[object, object, object, object, object, object, object, object],
+            cast(
+                object,
+                await asyncio.gather(
+                    redis_client.list_main(),
+                    redis_client.list_background(),
+                    redis_client.list_workers(),
+                    redis_client.used_memory_bytes(),
+                    redis_client.metakg_freshness(),
+                    redis_client.subclass_freshness(),
+                    redis_client.get_leader(),
+                    redis_client.get_leader_elected_at(),
+                    return_exceptions=True,
+                ),
             ),
         )
         (
@@ -370,6 +379,8 @@ async def status_root() -> StatusSnapshot:
             redis_mem_r,
             metakg_r,
             subclass_r,
+            leader_r,
+            leader_elected_r,
         ) = redis_results
         if any(isinstance(r, BaseException) for r in redis_results):
             redis_client.request_health_check()
@@ -400,6 +411,8 @@ async def status_root() -> StatusSnapshot:
     stuck_job_count = _unwrap(stuck_r)
     metakg_record = cast(FreshnessRecord | None, _unwrap(metakg_r))
     subclass_record = cast(FreshnessRecord | None, _unwrap(subclass_r))
+    leader = cast(str | None, _unwrap(leader_r))
+    leader_elected_at = cast(datetime | None, _unwrap(leader_elected_r))
 
     # `registry_available` flips False when *any* of the three registry
     # reads failed (they all hit Redis; one failing means we can't trust
@@ -498,6 +511,8 @@ async def status_root() -> StatusSnapshot:
         # True on their own snapshot - not currently wired.
         metakg=_metakg_row(metakg_record, self_reported=False, now=now),
         subclass_map=_subclass_map_row(subclass_record),
+        leader=leader,
+        leader_elected_at=leader_elected_at,
     )
 
 

src/retriever/types/status.py

@@ -309,3 +309,7 @@ class StatusSnapshot(TypedDict):
     tiers: list[StatusTier]
     metakg: StatusMetaKG
     subclass_map: StatusSubclassMap
+    leader: str | None
+    """Builder token of the current leader; None when the cluster is leaderless."""
+    leader_elected_at: datetime | None
+    """Best-effort time the current leader won the lease; None if unknown/leaderless."""