[iris] Reject stale clients on root LaunchJob submissions (#5108)

Add client_revision_date to LaunchJobRequest, stamped at wheel build
time (via scripts/python_libs_package.py) and from git log on editable
installs. Controller rejects root submissions older than
MIN_CLIENT_REVISION_DATE; nested submissions are exempt so in-flight
jobs aren't killed by floor bumps. Empty string is treated as the
feature introduction date so already-deployed clients get a 2-week grace
window. Also deletes lib/iris/hatch_build.py (vestigial now that protos
are checked in).

Fixes #4840

Author: rjpower

lib/iris/Dockerfile

@@ -55,7 +55,7 @@ ENV PATH="/app/.venv/bin:$PATH"
 # Rigging is a workspace-local dep not on PyPI with a compatible API.
 # Copy only metadata first so source changes don't bust the dep cache.
 COPY lib/rigging/pyproject.toml ./lib/rigging/pyproject.toml
-COPY lib/iris/pyproject.toml lib/iris/hatch_build.py ./lib/iris/
+COPY lib/iris/pyproject.toml ./lib/iris/
 RUN --mount=type=cache,target=/root/.cache/uv \
     printf '[tool.uv.workspace]\nmembers = ["lib/iris", "lib/rigging"]\n\n[tool.uv.sources]\nmarin-rigging = { workspace = true }\n' > pyproject.toml \
     && uv sync --package marin-iris --no-install-project

lib/iris/Dockerfile.dockerignore

@@ -3,7 +3,6 @@
 *
 
 !lib/iris/pyproject.toml
-!lib/iris/hatch_build.py
 !lib/iris/src/
 !lib/iris/examples/
 !lib/iris/config/

lib/iris/hatch_build.py

@@ -66,8 +66,6 @@ dev = [
     "pytest>=8.4",
 ]
 
-[tool.hatch.build.hooks.custom]
-
 [tool.hatch.build.targets.wheel]
 packages = ["src/iris"]
 

lib/iris/pyproject.toml

@@ -0,0 +1,8 @@
+# Copyright The Marin Authors
+# SPDX-License-Identifier: Apache-2.0
+
+# Auto-generated. Overwritten by scripts/python_libs_package.py during wheel
+# builds. Editable installs leave BUILD_DATE empty; iris.version falls back to
+# `git log` on the iris source tree.
+
+BUILD_DATE = ""

lib/iris/src/iris/_build_info.py

@@ -25,6 +25,7 @@
 from iris.rpc.logging_connect import LogServiceClientSync
 from iris.rpc.errors import call_with_retry, format_connect_error, poll_with_retries
 from iris.time_proto import duration_to_proto
+from iris.version import client_revision_date
 from rigging.timing import Deadline, Duration, ExponentialBackoff
 
 logger = logging.getLogger(__name__)
@@ -132,6 +133,7 @@ def submit_job(
             task_image=task_image or "",
             priority_band=priority_band,
             submit_argv=submit_argv or [],
+            client_revision_date=client_revision_date(),
         )
         if self._bundle_id:
             request.bundle_id = self._bundle_id

lib/iris/src/iris/cluster/client/remote_client.py

@@ -17,6 +17,7 @@
 import uuid
 import dataclasses
 from dataclasses import dataclass
+from datetime import date, timedelta
 from typing import Any, Protocol
 
 from connectrpc.code import Code
@@ -124,6 +125,44 @@
 # Maximum bundle size in bytes (25 MB) - matches client-side limit
 MAX_BUNDLE_SIZE_BYTES = 25 * 1024 * 1024
 
+# A root LaunchJob submission is rejected if its client_revision_date is more
+# than FRESHNESS_WINDOW older than today. Clients get exactly this long to
+# upgrade after a new marin-iris release is cut.
+FRESHNESS_WINDOW = timedelta(days=14)
+
+# Date this freshness check shipped. An empty client_revision_date is
+# interpreted as this date — already-deployed clients that don't set the field
+# start being rejected FRESHNESS_WINDOW after rollout.
+FEATURE_INTRODUCTION_DATE = date(2026, 4, 22)
+
+
+def _check_client_freshness(client_date_str: str, now: date) -> None:
+    """Reject root LaunchJob submissions whose client is older than FRESHNESS_WINDOW.
+
+    Empty string is treated as FEATURE_INTRODUCTION_DATE so old clients (which
+    don't set the field at all) behave as if they shipped the day this check
+    rolled out.
+    """
+    if not client_date_str:
+        client_date = FEATURE_INTRODUCTION_DATE
+    else:
+        try:
+            client_date = date.fromisoformat(client_date_str)
+        except ValueError as err:
+            raise ConnectError(
+                Code.INVALID_ARGUMENT,
+                f"client_revision_date must be ISO YYYY-MM-DD, got {client_date_str!r}",
+            ) from err
+    floor = now - FRESHNESS_WINDOW
+    if client_date < floor:
+        raise ConnectError(
+            Code.FAILED_PRECONDITION,
+            f"marin-iris client is too old (build {client_date.isoformat()}; "
+            f"minimum {floor.isoformat()}). Run `uv sync` or upgrade "
+            f"marin-iris and retry.",
+        )
+
+
 USER_TASK_STATES = (
     job_pb2.TASK_STATE_PENDING,
     job_pb2.TASK_STATE_ASSIGNED,
@@ -1040,6 +1079,12 @@ def launch_job(
 
         job_id = JobName.from_wire(request.name)
 
+        # Reject root submissions from stale clients. Nested submissions (from
+        # a job already running in the cluster) are exempt — the workload would
+        # otherwise crash mid-flight as the freshness window slides forward.
+        if job_id.is_root:
+            _check_client_freshness(request.client_revision_date, date.today())
+
         # When an auth provider is configured, override the user segment with
         # the verified identity to prevent impersonation. Only override for
         # root-level submissions; child jobs inherit the parent's user.

lib/iris/src/iris/cluster/controller/service.py

@@ -106,6 +106,14 @@ message Controller {
     // Empty when submitted programmatically. Bookkeeping only; not
     // interpreted by the controller.
     repeated string submit_argv = 35;
+
+    // ISO date (YYYY-MM-DD) of the client's marin-iris build. For wheels this
+    // is stamped at build time by scripts/python_libs_package.py; for editable
+    // installs it's derived from `git log` on the iris source tree. Root
+    // submissions older than FRESHNESS_WINDOW (today - 14d) are rejected.
+    // Nested submissions (is_root == false) are exempt. Empty string is
+    // interpreted as FEATURE_INTRODUCTION_DATE — see service.py.
+    string client_revision_date = 36;
   }
 
   message LaunchJobResponse {

lib/iris/src/iris/rpc/controller.proto

@@ -47,7 +47,7 @@ class Controller(_message.Message):
     JOB_QUERY_SCOPE_ROOTS: Controller.JobQueryScope
     JOB_QUERY_SCOPE_CHILDREN: Controller.JobQueryScope
     class LaunchJobRequest(_message.Message):
-        __slots__ = ("name", "entrypoint", "resources", "environment", "bundle_id", "bundle_blob", "scheduling_timeout", "ports", "max_task_failures", "max_retries_failure", "max_retries_preemption", "constraints", "coscheduling", "replicas", "timeout", "fail_if_exists", "reservation", "preemption_policy", "existing_job_policy", "priority_band", "task_image", "submit_argv")
+        __slots__ = ("name", "entrypoint", "resources", "environment", "bundle_id", "bundle_blob", "scheduling_timeout", "ports", "max_task_failures", "max_retries_failure", "max_retries_preemption", "constraints", "coscheduling", "replicas", "timeout", "fail_if_exists", "reservation", "preemption_policy", "existing_job_policy", "priority_band", "task_image", "submit_argv", "client_revision_date")
         NAME_FIELD_NUMBER: _ClassVar[int]
         ENTRYPOINT_FIELD_NUMBER: _ClassVar[int]
         RESOURCES_FIELD_NUMBER: _ClassVar[int]
@@ -70,6 +70,7 @@ class Controller(_message.Message):
         PRIORITY_BAND_FIELD_NUMBER: _ClassVar[int]
         TASK_IMAGE_FIELD_NUMBER: _ClassVar[int]
         SUBMIT_ARGV_FIELD_NUMBER: _ClassVar[int]
+        CLIENT_REVISION_DATE_FIELD_NUMBER: _ClassVar[int]
         name: str
         entrypoint: _job_pb2.RuntimeEntrypoint
         resources: _job_pb2.ResourceSpecProto
@@ -92,7 +93,8 @@ class Controller(_message.Message):
         priority_band: _job_pb2.PriorityBand
         task_image: str
         submit_argv: _containers.RepeatedScalarFieldContainer[str]
-        def __init__(self, name: _Optional[str] = ..., entrypoint: _Optional[_Union[_job_pb2.RuntimeEntrypoint, _Mapping]] = ..., resources: _Optional[_Union[_job_pb2.ResourceSpecProto, _Mapping]] = ..., environment: _Optional[_Union[_job_pb2.EnvironmentConfig, _Mapping]] = ..., bundle_id: _Optional[str] = ..., bundle_blob: _Optional[bytes] = ..., scheduling_timeout: _Optional[_Union[_time_pb2.Duration, _Mapping]] = ..., ports: _Optional[_Iterable[str]] = ..., max_task_failures: _Optional[int] = ..., max_retries_failure: _Optional[int] = ..., max_retries_preemption: _Optional[int] = ..., constraints: _Optional[_Iterable[_Union[_job_pb2.Constraint, _Mapping]]] = ..., coscheduling: _Optional[_Union[_job_pb2.CoschedulingConfig, _Mapping]] = ..., replicas: _Optional[int] = ..., timeout: _Optional[_Union[_time_pb2.Duration, _Mapping]] = ..., fail_if_exists: _Optional[bool] = ..., reservation: _Optional[_Union[_job_pb2.ReservationConfig, _Mapping]] = ..., preemption_policy: _Optional[_Union[_job_pb2.JobPreemptionPolicy, str]] = ..., existing_job_policy: _Optional[_Union[_job_pb2.ExistingJobPolicy, str]] = ..., priority_band: _Optional[_Union[_job_pb2.PriorityBand, str]] = ..., task_image: _Optional[str] = ..., submit_argv: _Optional[_Iterable[str]] = ...) -> None: ...
+        client_revision_date: str
+        def __init__(self, name: _Optional[str] = ..., entrypoint: _Optional[_Union[_job_pb2.RuntimeEntrypoint, _Mapping]] = ..., resources: _Optional[_Union[_job_pb2.ResourceSpecProto, _Mapping]] = ..., environment: _Optional[_Union[_job_pb2.EnvironmentConfig, _Mapping]] = ..., bundle_id: _Optional[str] = ..., bundle_blob: _Optional[bytes] = ..., scheduling_timeout: _Optional[_Union[_time_pb2.Duration, _Mapping]] = ..., ports: _Optional[_Iterable[str]] = ..., max_task_failures: _Optional[int] = ..., max_retries_failure: _Optional[int] = ..., max_retries_preemption: _Optional[int] = ..., constraints: _Optional[_Iterable[_Union[_job_pb2.Constraint, _Mapping]]] = ..., coscheduling: _Optional[_Union[_job_pb2.CoschedulingConfig, _Mapping]] = ..., replicas: _Optional[int] = ..., timeout: _Optional[_Union[_time_pb2.Duration, _Mapping]] = ..., fail_if_exists: _Optional[bool] = ..., reservation: _Optional[_Union[_job_pb2.ReservationConfig, _Mapping]] = ..., preemption_policy: _Optional[_Union[_job_pb2.JobPreemptionPolicy, str]] = ..., existing_job_policy: _Optional[_Union[_job_pb2.ExistingJobPolicy, str]] = ..., priority_band: _Optional[_Union[_job_pb2.PriorityBand, str]] = ..., task_image: _Optional[str] = ..., submit_argv: _Optional[_Iterable[str]] = ..., client_revision_date: _Optional[str] = ...) -> None: ...
     class LaunchJobResponse(_message.Message):
         __slots__ = ("job_id",)
         JOB_ID_FIELD_NUMBER: _ClassVar[int]