From 26404240c1a03ed3d7f7148aeda0f2e8951debf6 Mon Sep 17 00:00:00 2001 From: Paul Fidika Date: Thu, 23 Jul 2026 21:43:00 -0600 Subject: [PATCH 1/2] =?UTF-8?q?pgw#628:=20residency=20reporting=20v2=20?= =?UTF-8?q?=E2=80=94=20content-addressed=20idempotent=20observations=20(0.?= =?UTF-8?q?52.0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Worker half of th#1070: every applied HelloAck opens a republish epoch; the reconcile pass re-announces verified cached identities (exact ref+digest) once per epoch even when unchanged — a re-received plan (redrive/overdue resend/reconnect) is the hub asking for a resync, and success observations are safe to emit late, twice, or across plan revisions under the v2 hub's digest acceptance. Job-path cache hits within an epoch stay deduped. gw#614 same-set no-cancel unchanged. Revert-turns-red test; mypy+ruff clean; full suite 649 passed/5 skipped. No forced fleet rebuild — v2 hubs accept 0.44-0.51 reports fine. --- CHANGELOG.md | 17 ++++ pyproject.toml | 2 +- src/gen_worker/executor.py | 29 ++++++- tests/test_residency_republish_pgw628.py | 100 +++++++++++++++++++++++ 4 files changed, 144 insertions(+), 4 deletions(-) create mode 100644 tests/test_residency_republish_pgw628.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d3b50526..f10c6c26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,22 @@ # Changelog +## 0.52.0 (2026-07-24) + +- **pgw#628: residency reporting v2 — content-addressed idempotent + observations (worker half of th#1070).** Every applied HelloAck opens a + republish epoch: the reconcile pass re-announces verified cached + identities (ON_DISK/IN_RAM/IN_VRAM with exact ref+digest) once per epoch + even when unchanged — a re-received plan (hub redrive, overdue resend, + reconnect) is the hub asking for a resync, and success observations are + now safe to emit late, twice, or across plan revisions (the v2 hub + accepts them by digest, generations survive only for cancel/evict/ + failure attribution). Job-path cache hits within one epoch stay deduped + (no event spam). gw#614's no-cancel-on-same-set reconcile behavior is + unchanged — under the v2 hub it is simply correct instead of a trap. + Version floor note: endpoint images need no forced rebuild — v2 hubs + accept 0.44–0.51 success reports fine; images pick up >=0.52.0 on their + next routine rebuild for the lost-observation resync hardening. + ## 0.51.0 (2026-07-24) - **gw#627: Conv2d additive-branch support in the runtime LoRA overlay.** diff --git a/pyproject.toml b/pyproject.toml index 05813dcd..c904876d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "gen-worker" -version = "0.51.0" +version = "0.52.0" description = "A library used to build custom functions in Cozy Creator's serverless function platform." readme = "README.md" license = "MIT" diff --git a/src/gen_worker/executor.py b/src/gen_worker/executor.py index fa3f5358..bbd20371 100644 --- a/src/gen_worker/executor.py +++ b/src/gen_worker/executor.py @@ -696,6 +696,15 @@ def __init__( # pipeline is still in RAM/VRAM. Keep the disk identity separately # until record teardown makes it the highest residency tier. self._disk_identities: Dict[str, _ResidencyIdentity] = {} + # pgw#628 (th#1070 residency v2): every applied HelloAck opens a new + # republish epoch. The reconcile pass re-announces verified cached + # identities the hub re-asked about even when unchanged — observations + # are content-addressed and idempotent hub-side, and a force-resent + # plan is exactly the hub saying "tell me again" (redrive/overdue + # resends could otherwise never heal a lost success observation). + # Job-path ensure_local calls within the same epoch stay deduped. + self._residency_republish_epoch = 0 + self._identity_publish_epochs: Dict[str, int] = {} self._identity_lock = threading.RLock() # Cold-ref waiters (th#763): ensure_local blocks here until the # hub's re-minted DOWNLOAD banks a snapshot for the ref. @@ -781,6 +790,7 @@ def _on_residency_event( with self._identity_lock: self._resident_identities.pop(ref, None) self._disk_identities.pop(ref, None) + self._identity_publish_epochs.pop(ref, None) try: loop = asyncio.get_running_loop() except RuntimeError: @@ -979,6 +989,7 @@ def replace_desired_snapshots( for ref, snapshot in stored.items() } with self._identity_lock: + self._residency_republish_epoch += 1 self._desired_snapshot_identities = desired # Generations belong only to the current desired identity. Leave # actual resident identity untouched: those bytes may still be in @@ -1044,8 +1055,15 @@ def activate_disk_identity(self, ref: str) -> _ResidencyIdentity: async def _confirm_cached_identity( self, ref: str, identity: _ResidencyIdentity, ) -> None: - """Publish exact identity when verified cached bytes satisfy a newer - desired generation without requiring a redundant download.""" + """Publish exact identity when verified cached bytes satisfy the + desired state without requiring a redundant download. + + pgw#628 (th#1070 residency v2): the emission is content-addressed and + idempotent hub-side, so it is re-sent once per applied-HelloAck epoch + even when the identity is unchanged — a re-received plan (redrive, + overdue resend, reconnect) is the hub asking for a resync, and a + worker that never re-announces can strand a lost success observation + forever. Job-path calls within the same epoch remain deduped.""" tier = self.residency.tier(ref) digest, _ = identity if not digest: @@ -1060,7 +1078,12 @@ async def _confirm_cached_identity( # serving the new snapshot. if tier in (residency_mod.Tier.RAM, residency_mod.Tier.VRAM) and current[0] != digest: return - if not self._set_resident_identity(ref, identity): + changed = self._set_resident_identity(ref, identity) + with self._identity_lock: + epoch = self._residency_republish_epoch + republish = self._identity_publish_epochs.get(ref) != epoch + self._identity_publish_epochs[ref] = epoch + if not changed and not republish: return state = { residency_mod.Tier.DISK: pb.MODEL_STATE_ON_DISK, diff --git a/tests/test_residency_republish_pgw628.py b/tests/test_residency_republish_pgw628.py new file mode 100644 index 00000000..b0456aad --- /dev/null +++ b/tests/test_residency_republish_pgw628.py @@ -0,0 +1,100 @@ +"""pgw#628 (th#1070 residency protocol v2, worker half): success observations +are (ref, digest, state) — content-addressed, idempotent, and safe to emit +twice. A re-received desired plan (hub redrive, overdue resend, reconnect) is +the hub asking for a resync, so the reconcile pass re-announces verified +cached identities once per applied-HelloAck epoch even when nothing changed. +Within one epoch the identity dedupe still holds (no event spam). The gw#614 +no-cancel-on-same-set behavior is untouched — under v2 it is simply correct +instead of a trap. +""" + +from __future__ import annotations + +import time + +from gen_worker.pb import worker_scheduler_pb2 as pb + +from harness.blob_host import BlobHost +from harness.hub_double import Conn, hub_double, is_model_event, is_ready + +_MODEL_REF = "harness/residency-tiny" + + +def _disk_only_ack(snapshot: pb.Snapshot, generation: int) -> pb.HelloAck: + return pb.HelloAck( + protocol_version=pb.PROTOCOL_VERSION_CURRENT, + desired_residency=pb.DesiredResidency( + generation=generation, + disk_refs=[_MODEL_REF], + snapshots={_MODEL_REF: snapshot}, + ), + ) + + +def _count_on_disk(conn: Conn) -> int: + with conn._recv_cond: + return sum( + 1 + for m in conn.received + if m.WhichOneof("msg") == "model_event" + and m.model_event.ref == _MODEL_REF + and m.model_event.state == pb.MODEL_STATE_ON_DISK + ) + + +def _wait_on_disk_count(conn: Conn, want: int, timeout: float = 15.0) -> None: + deadline = time.monotonic() + timeout + while _count_on_disk(conn) < want: + if time.monotonic() > deadline: + raise TimeoutError( + f"expected {want} ON_DISK re-reports, saw {_count_on_disk(conn)}" + ) + time.sleep(0.05) + + +def test_reissued_plan_republishes_held_identity(tmp_path) -> None: + """Each applied plan re-send yields exactly ONE fresh ON_DISK re-report + carrying the exact (ref, digest): the idempotent resync a v2 hub can + always absorb and that heals a lost success observation. Within an + epoch the dedupe holds — re-announce is per applied ack, never spam.""" + blobs = BlobHost(tmp_path) + try: + snapshot = blobs.one_file_snapshot("snap-1", "blob", b"tiny-weights") + with hub_double() as (scheduler, _harness): + conn = scheduler.wait_connection(0) + conn.wait_for(is_ready) + + conn.send(hello_ack=_disk_only_ack(snapshot, generation=1)) + first = conn.wait_for( + is_model_event(_MODEL_REF, pb.MODEL_STATE_ON_DISK) + ).model_event + assert first.snapshot_digest == "snap-1" + assert first.residency_generation == 1 + + # The hub re-sends the SAME plan (redrive / overdue resend): the + # worker must re-announce the held bytes, not stay silent behind + # its identity dedupe. + baseline = _count_on_disk(conn) + conn.send(hello_ack=_disk_only_ack(snapshot, generation=1)) + _wait_on_disk_count(conn, baseline + 1) + events = [ + m.model_event + for m in conn.received + if m.WhichOneof("msg") == "model_event" + and m.model_event.ref == _MODEL_REF + and m.model_event.state == pb.MODEL_STATE_ON_DISK + ] + assert events[-1].snapshot_digest == "snap-1" + assert events[-1].residency_generation == 1 + + # And exactly one per applied plan: no runaway re-announce loop. + time.sleep(0.5) + assert _count_on_disk(conn) == baseline + 1 + + # A third re-send opens a third epoch: one more re-report. + conn.send(hello_ack=_disk_only_ack(snapshot, generation=1)) + _wait_on_disk_count(conn, baseline + 2) + time.sleep(0.5) + assert _count_on_disk(conn) == baseline + 2 + finally: + blobs.shutdown() From 5d1371a8a970322d547f7bf46ad908b4aabba81e Mon Sep 17 00:00:00 2001 From: Paul Fidika Date: Thu, 23 Jul 2026 21:47:03 -0600 Subject: [PATCH 2/2] pgw#628: uv.lock version re-resolve (0.52.0) --- uv.lock | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/uv.lock b/uv.lock index a906bedf..9741d735 100644 --- a/uv.lock +++ b/uv.lock @@ -583,7 +583,7 @@ wheels = [ [[package]] name = "gen-worker" -version = "0.51.0" +version = "0.52.0" source = { editable = "." } dependencies = [ { name = "blake3" },