fix: bind persistent native publisher admission to Pod UID

.github/workflows/native-supervisor-test.yml

@@ -0,0 +1,24 @@
+name: Native Supervisor Tests
+on:
+  push:
+    branches: [main]
+    paths: [deploy/catchup/**, .github/workflows/native-supervisor-test.yml]
+  pull_request:
+    paths: [deploy/catchup/**, .github/workflows/native-supervisor-test.yml]
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  admission:
+    runs-on: ubuntu-24.04
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@v4
+      - name: Install disposable etcd fixture dependency
+        run: sudo apt-get update && sudo apt-get install -y etcd-server
+      - name: Verify supervisor identity, CAS, backoff and join behavior
+        env:
+          ETCD_BIN: /usr/bin/etcd
+        run: python3 -m unittest discover -s deploy/catchup -p 'test_native_supervisor.py' -v

deploy/catchup/native-publisher.example.json

@@ -0,0 +1,28 @@
+{
+  "apiVersion": "v1",
+  "kind": "Pod",
+  "metadata": {"name": "native-publisher-example"},
+  "spec": {
+    "restartPolicy": "Never",
+    "automountServiceAccountToken": false,
+    "terminationGracePeriodSeconds": 1800,
+    "containers": [{
+      "name": "publisher",
+      "image": "YOUR_REGISTRY/lance-context@sha256:REPLACE_WITH_TESTED_IMAGE_DIGEST",
+      "command": ["python3", "-u", "/supervisor/native_supervisor.py"],
+      "envFrom": [{"secretRef": {"name": "lance-native-publisher-config"}}],
+      "env": [
+        {"name": "POD_UID", "valueFrom": {"fieldRef": {"apiVersion": "v1", "fieldPath": "metadata.uid"}}},
+        {"name": "HOSTNAME", "valueFrom": {"fieldRef": {"apiVersion": "v1", "fieldPath": "metadata.name"}}}
+      ],
+      "resources": {
+        "requests": {"cpu": "2", "memory": "16Gi"},
+        "limits": {"cpu": "2", "memory": "16Gi"}
+      },
+      "readinessProbe": {"exec": {"command": ["test", "-f", "/tmp/native-supervisor-ready"]}},
+      "volumeMounts": [{"name": "supervisor", "mountPath": "/supervisor", "readOnly": true}],
+      "securityContext": {"allowPrivilegeEscalation": false}
+    }],
+    "volumes": [{"name": "supervisor", "configMap": {"name": "native-supervisor-REPLACE_WITH_CONTENT_HASH"}}]
+  }
+}

deploy/catchup/native-publisher.md

@@ -0,0 +1,76 @@
+# Persistent native publisher admission
+
+`native_supervisor.py` tracks the existing external persistent-publisher
+supervisor alongside the native catch-up deployment assets. It adds Pod UID
+admission fencing. It does not replace or enable the master's native Kubernetes
+Job controller described in `docs/master-catchup.md`, and does not create, delete,
+restart, or take ownership of any Pod or table.
+
+A Kubernetes Pod name can be reused. Previously a replacement under the same
+name could accept an operator record intended for its predecessor. Before every
+attempt, this supervisor now requires all of:
+
+- operator `phase` is `native_parallel_active`;
+- `native_job` and the exact persistent `catchup-active/<target hex>` owner match
+  `CATCHUP_JOB_NAME`;
+- `native_pod` matches the Pod name;
+- `native_pod_uid` matches `POD_UID`, supplied by the downward API.
+
+A missing local UID fails startup before readiness. A missing or mismatched
+authorized UID leaves the supervisor in standby without touching the attempt or
+failure ledger. The attempt CAS still compares the **entire raw operator value**,
+the owner, and the old attempt. A concurrent lifecycle change after the UID check
+therefore rejects admission. Attempts include the admitted `pod_uid` for diagnosis.
+
+## Deployment contract
+
+Use `native-publisher.example.json` as the standalone Pod shape. Mount an immutable,
+content-addressed ConfigMap containing this script, and pin a tested image with
+Python 3 and `/usr/local/bin/lance-context-master`. Do not feed this standalone
+Pod manifest into `CATCHUP_POD_TEMPLATE`, which accepts the native Job controller's
+different PodSpec contract.
+
+**Require one container and `restartPolicy: Never`, and verify both on the live
+Pod before activation.** The Pod UID then identifies one supervisor incarnation.
+An `Always` or `OnFailure` Pod can restart its container under the same UID and is
+not supported by this protocol. The example uses the downward API for both UID
+and name; do not supply a literal UID or reuse another Pod's value. The supervisor
+has no Kubernetes credential and cannot independently inspect its restart policy.
+
+The reviewed environment/Secret must configure the target, persistent job,
+storage, etcd prefix/endpoints, worker endpoints, historical shard identities,
+owned target, append settings and byte budgets for the native binary. In
+particular `MERGE_OWNED_TARGETS` must include `CATCHUP_TARGET`. Size memory limits
+above the supervisor's existing 12 GiB admission watermark; that watermark is not
+a bound on one admitted operation. Keep native byte reservations and Lance memory
+limits in force. Configure `CATCHUP_SLICE_SECS` as a soft admission slice and allow
+already admitted native work to join during graceful termination.
+
+Create each new Pod in standby, read its actual Ready UID/container identity, then
+CAS the operator to authorize that UID while preserving the persistent owner and
+all attempt/failure records. A same-name foreign Pod created between observation
+and activation cannot use the stale UID, even if the activation CAS succeeds.
+Do not activate a replacement until the predecessor has qualified terminal/join
+evidence and native execution recovery has completed. Pod disappearance, an API
+timeout, or an expired lease alone is not that evidence.
+
+The wrapper retains the existing backoff and child-join behavior: SIGTERM stops
+new admissions and waits for the native child; unknown outcomes do not clear
+ownership or bypass the native execution fence. The native executable remains
+responsible for claims, cancellation, memory reservations, storage barriers and
+watermarks. This script has no general automatic recreation loop. Migration to
+the native Job controller remains separate work.
+
+## Tests
+
+```sh
+ETCD_BIN=/path/to/etcd python3 -m unittest discover \
+  -s deploy/catchup -p 'test_native_supervisor.py' -v
+```
+
+Tests run real supervisor processes and a disposable loopback-only etcd. The
+native child is a recording stub: they prove stale UID rejection, full-operator
+CAS races, subsequent admission checks, preserved backoff and joined shutdown,
+not Lance payload correctness or Kubernetes lifecycle behavior. Validate the
+changed supervisor separately with real Kubernetes downward API identities and
+the native binary against isolated storage before production migration.

deploy/catchup/native_supervisor.py

@@ -0,0 +1,246 @@
+"""Admission supervisor for externally managed persistent native publishers.
+
+Every admission is bound to the actual Pod UID, not just its reusable name.
+Data work and execution fencing remain in the native binary. This supervisor
+never clears ownership, replays an uncertain attempt, or kills a native child.
+Deploy only in a one-container Pod with restartPolicy Never; see native-publisher.md.
+"""
+
+import base64
+import json
+import os
+import pathlib
+import re
+import signal
+import subprocess
+import threading
+import time
+import urllib.request
+import uuid
+
+NATIVE_BINARY = "/usr/local/bin/lance-context-master"
+MEMORY_CURRENT = pathlib.Path("/sys/fs/cgroup/memory.current")
+READY_FILE = pathlib.Path("/tmp/native-supervisor-ready")
+
+
+def authorized(operator, job, pod_name, pod_uid):
+    """Reject missing identity and stale authorization for a reused Pod name."""
+    return bool(
+        pod_uid
+        and operator.get("phase") == "native_parallel_active"
+        and operator.get("native_job") == job
+        and operator.get("native_pod") == pod_name
+        and operator.get("native_pod_uid") == pod_uid
+    )
+
+
+def main():
+    UID = os.environ.get("POD_UID", "").strip()
+    if not UID:
+        raise SystemExit(
+            "POD_UID from metadata.uid is required; refusing native admission"
+        )
+    T = os.environ["CATCHUP_TARGET"]
+    J = os.environ["CATCHUP_JOB_NAME"]
+    P = os.environ["ETCD_PREFIX"].rstrip("/")
+    H = T.encode().hex()
+    E = os.environ["ETCD_ENDPOINTS"].rstrip("/") + "/v3/"
+    stop = threading.Event()
+    proc = None
+    enc = lambda s: base64.b64encode(s.encode()).decode()
+
+    def log(event, **kw):
+        print(
+            json.dumps(
+                {"at": time.time(), "event": event, "target": T, "job": J, **kw}
+            ),
+            flush=True,
+        )
+
+    def read(suffix):
+        r = urllib.request.Request(
+            E + "kv/range",
+            data=json.dumps({"key": enc(P + "/" + suffix)}).encode(),
+            headers={"Content-Type": "application/json"},
+        )
+        with urllib.request.urlopen(r, timeout=5) as f:
+            k = json.load(f).get("kvs", [])
+        return base64.b64decode(k[0]["value"]).decode() if k else None
+
+    def cas_attempt(old, new, operator):
+        if not authorized(json.loads(operator), J, os.environ["HOSTNAME"], UID):
+            return None
+        key = P + "/merge-rollout-attempts/" + H
+        compares = [
+            {
+                "key": enc(P + "/catchup-active/" + H),
+                "target": "VALUE",
+                "result": "EQUAL",
+                "value": enc(J),
+            },
+            {
+                "key": enc(P + "/merge-rollout-operator/" + H),
+                "target": "VALUE",
+                "result": "EQUAL",
+                "value": enc(operator),
+            },
+        ]
+        compares.append(
+            {"key": enc(key), "target": "VERSION", "result": "EQUAL", "version": "0"}
+            if old is None
+            else {
+                "key": enc(key),
+                "target": "VALUE",
+                "result": "EQUAL",
+                "value": enc(old),
+            }
+        )
+        value = json.dumps(new, separators=(",", ":"))
+        r = urllib.request.Request(
+            E + "kv/txn",
+            data=json.dumps(
+                {
+                    "compare": compares,
+                    "success": [
+                        {"request_put": {"key": enc(key), "value": enc(value)}}
+                    ],
+                }
+            ).encode(),
+            headers={"Content-Type": "application/json"},
+        )
+        with urllib.request.urlopen(r, timeout=5) as f:
+            out = json.load(f)
+        return value if out.get("succeeded") else None
+
+    def halt(*args):
+        stop.set()
+        log("admission_stopping_waiting_for_native_exit")
+
+    signal.signal(signal.SIGTERM, halt)
+    signal.signal(signal.SIGINT, halt)
+    READY_FILE.touch()
+    log("standby")
+    while not stop.is_set():
+        try:
+            operator = read("merge-rollout-operator/" + H)
+            op = json.loads(operator or "{}")
+            if (
+                not authorized(op, J, os.environ["HOSTNAME"], UID)
+                or read("catchup-active/" + H) != J
+            ):
+                stop.wait(10)
+                continue
+            # The admission supervisor does not bypass native persistent failure backoff.
+            failure = json.loads(
+                read("merge-failures/" + H + "/" + b"master:catchup".hex()) or "{}"
+            )
+            old_attempt = read("merge-rollout-attempts/" + H)
+            previous = json.loads(old_attempt or "{}")
+            if previous and previous.get("job") != J:
+                raise RuntimeError("attempt ledger belongs to a different job")
+            delay = max(
+                0,
+                max(failure.get("next_retry_ms", 0), previous.get("next_retry_ms", 0))
+                / 1000
+                - time.time(),
+            )
+            if delay > 0:
+                log("persistent_backoff", retry_after_seconds=round(delay))
+                stop.wait(min(delay, 60))
+                continue
+            if int(MEMORY_CURRENT.read_text()) > 12 * 2**30:
+                log("memory_admission_delayed")
+                stop.wait(15)
+                continue
+            # The native executable claims only J's target, reconciles old execution,
+            # enforces shared byte reservations and idle progress, then commits watermarks.
+            attempt = {
+                "job": J,
+                "pod_uid": UID,
+                "id": str(uuid.uuid4()),
+                "started_at": time.time(),
+                "consecutive_failures": previous.get("consecutive_failures", 0) + 1,
+                "state": "admitted",
+            }
+            attempt["next_retry_ms"] = int(
+                (
+                    time.time()
+                    + min(3600, 60 * 2 ** min(attempt["consecutive_failures"], 6))
+                )
+                * 1000
+            )
+            attempt["needs_attention"] = attempt["consecutive_failures"] >= 3
+            admitted = cas_attempt(old_attempt, attempt, operator)
+            if admitted is None:
+                stop.wait(10)
+                continue
+            started = time.monotonic()
+            reclaimed = 0
+            pass_errors = 0
+            proc = subprocess.Popen(
+                [NATIVE_BINARY],
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+                text=True,
+                bufsize=1,
+            )
+            log("native_pass_started", pid=proc.pid)
+            for line in proc.stdout:
+                print(line.rstrip(), flush=True)
+                clean = re.sub(r"\x1b\[[0-9;]*m", "", line)
+                if "parallel rollout append completed" in clean:
+                    n = re.search(r"\breclaimed=(\d+)", clean)
+                    if n:
+                        reclaimed += int(n.group(1))
+                    n = re.search(r"\bfailed=(\d+)", clean)
+                    if n:
+                        pass_errors += int(n.group(1))
+            code = proc.wait()
+            seconds = time.monotonic() - started
+            proc = None
+            terminal = dict(
+                attempt,
+                state="succeeded" if code == 0 else "failed",
+                exit_code=code,
+                finished_at=time.time(),
+                reclaimed=reclaimed,
+                seconds=seconds,
+            )
+            if code == 0:
+                terminal.update(
+                    consecutive_failures=0,
+                    needs_attention=False,
+                    next_retry_ms=int((time.time() + (2 if reclaimed else 30)) * 1000),
+                )
+            else:
+                terminal["next_retry_ms"] = int(
+                    (
+                        time.time()
+                        + min(3600, 60 * 2 ** min(attempt["consecutive_failures"], 6))
+                    )
+                    * 1000
+                )
+            if cas_attempt(admitted, terminal, operator) is None:
+                log("attempt_record_changed_ownership_untouched")
+            log(
+                "native_pass_finished",
+                exit_code=code,
+                reclaimed=reclaimed,
+                seconds=seconds,
+                generations_per_second=reclaimed / seconds,
+                failed_shards=pass_errors,
+            )
+            # Let ordinary flushes proceed; zero-work passes must not hammer etcd/catalogs.
+            stop.wait(2 if code == 0 and reclaimed > 0 else 30)
+        except Exception as e:  # noqa: BLE001 - preserve native child join on any supervisor error
+            log("supervisor_error_ownership_untouched", error=repr(e))
+            if proc is not None:
+                # Do not launch a second child after a controller error. Wait for native exit.
+                proc.wait()
+                proc = None
+            stop.wait(30)
+    log("supervisor_stopped")
+
+
+if __name__ == "__main__":
+    main()