Nearle/kubernetes
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix worker/gateway logic bugs and duplicate CORS headers

Browse Source 
worker.py (both the ConfigMap copy and conf/worker.py):
- Set an explicit ack_wait=60s on the JetStream pull consumer. It was
  previously left at the implicit default (~30s), the same ballpark
  as the outbound HTTP timeout - a slow-but-legitimate external call
  could cause JetStream to redeliver the message to another worker
  while the first was still mid-request, double-processing a
  non-idempotent call (e.g. duplicate order creation).
- Track in-flight tasks and drain them (bounded wait) before closing
  the NATS/HTTP connections on shutdown, instead of cutting them off
  immediately - avoids dropped/duplicated messages on pod restarts.
- Generic exception handler now does nak(delay=5) instead of an
  undelayed nak(), avoiding a tight redelivery loop on a persistent
  bug.
- Missing 'data' field in a message now explicitly drops with a log
  line instead of silently forwarding the entire internal envelope.
- Removed the hardcoded NATS password fallback baked into the source
  (every deployment already supplies it via a Secret at runtime, so
  this was a redundant plaintext copy sitting in a ConfigMap).

app.py (both the ConfigMap copy and conf/app.py):
- Fixed "NATS by connected" typo -> "NATS not connected".
- Same hardcoded-password-fallback removal as worker.py.

CORS:
- conf/nginx-jupiter.conf and the in-cluster jupiter-cors-proxy nginx
  config both add their own CORS headers without stripping any the
  upstream might set, unlike nginx-queue-proxy.conf which does this
  correctly. Added proxy_hide_header for the ACA-* headers in both -
  browsers reject a response with duplicate Access-Control-* values.

docker-compose.yml:
- Added the missing doormile-proxy service (doormile.com -> :8206 ->
  NodePort 30830). nginx-doormile.conf existed but had no service
  wiring it into Traefik, unlike every other app.
This commit is contained in:
Suriya
2026-07-18 16:07:49 +05:30
parent 836c079a05
commit 91dd240431
7 changed files with 101 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
manifests/core/worker-script.yaml
View File

@@ -19,6 +19,7 @@ data:
import aiohttp
import nats
from nats.errors import TimeoutError as NatsTimeoutError
from nats.js.api import ConsumerConfig
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import signal
import sys
@@ -32,17 +33,21 @@ data:
# Configuration
NATS_URL = os.getenv("NATS_URL", "nats://nats-server:4222")
NATS_USER = os.getenv("NATS_USER", "admin")
NATS_PASSWORD = os.getenv("NATS_PASSWORD", "package@321#")
NATS_PASSWORD = os.getenv("NATS_PASSWORD", "")
# Domain Configuration (Env Vars from Deployment)
NATS_STREAM = os.getenv("NATS_STREAM", "ORDERS")
NATS_CONSUMER = os.getenv("NATS_CONSUMER", "orders-worker")
# Supports comma-separated subjects: "api.v1.orders.*,api.v2.orders.*"
FILTER_SUBJECTS = os.getenv("FILTER_SUBJECT", "api.v1.mob.orders.*").split(",")
WORKER_CONCURRENCY = int(os.getenv("WORKER_CONCURRENCY", "10"))
BASE_URL = os.getenv("EXTERNAL_BASE_URL", "https://jupiter.nearle.app")
# Must stay comfortably above the outbound HTTP timeout below (30s), otherwise
# JetStream can redeliver a message to another worker while this one is still
# waiting on a slow-but-legitimate response, causing a duplicate forward.
ACK_WAIT_SECONDS = int(os.getenv("ACK_WAIT_SECONDS", "60"))
# Endpoint mapping is still useful for constructing the target URL
ENDPOINT_MAPPING = {
"/live/api/v1/deliveries/createdeliveries": f"{BASE_URL}/live/api/v1/deliveries/createdeliveries",
@@ -63,6 +68,7 @@ data:
session: aiohttp.ClientSession = None
running = True
semaphore = None # Initialized in main
active_tasks: set = set() # In-flight process_message() tasks, drained on shutdown
def signal_handler(sig, frame):
global running
@@ -87,7 +93,10 @@ data:
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
data_to_forward = payload.get("data", payload)
if "data" not in payload:
print(f"?? Message for {endpoint} missing 'data' field, dropping.")
return "DROP", None
data_to_forward = payload["data"]
# Payload Normalization for logs
if endpoint == "/live/api/v2/deliveries/createdeliverylog":
@@ -97,7 +106,7 @@ data:
data_to_forward = [data_to_forward]
http_method = method.upper() if method else "POST"
try:
# We use the global 'session' here
async with session.request(
@@ -108,7 +117,7 @@ data:
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
body = None
try:
body = await response.json()
@@ -118,7 +127,7 @@ data:
if 200 <= response.status < 300:
return "OK", body
elif response.status >= 500:
print(f"?? Server Error {response.status} from {endpoint}: {body}")
print(f"?? Server Error {response.status} from {endpoint}")
return "RETRY", body
else:
print(f"?? Client Error {response.status} from {endpoint}: {body}")
@@ -136,7 +145,7 @@ data:
messages_in_flight.inc()
start_time = asyncio.get_event_loop().time()
endpoint = "unknown"
try:
payload = json.loads(msg.data.decode())
endpoint = payload.get("endpoint", payload.get("original_path", "unknown"))
@@ -145,14 +154,14 @@ data:
print(f"?? Processing {endpoint}")
api_key = os.getenv("EXTERNAL_ENDPOINT_API_KEY", "")
# Single attempt
status, response_data = await forward_to_external(endpoint, payload, api_key, http_method)
if status == "OK":
await msg.ack()
messages_processed.labels(status="success", endpoint=endpoint).inc()
# Request-Reply Logic
if msg.reply:
reply_payload = json.dumps(response_data) if isinstance(response_data, (dict, list)) else str(response_data)
@@ -162,7 +171,7 @@ data:
# Unrecoverable error or unknown endpoint
await msg.term() # Terminate stops redelivery
messages_processed.labels(status="dropped", endpoint=endpoint).inc()
else: # RETRY
# Let JetStream handle backoff
await msg.nak(delay=2) # Custom delay before redelivery if desired, or just nak()
@@ -173,7 +182,8 @@ data:
await msg.term()
except Exception as e:
print(f"? Critical Worker Error: {e}")
await msg.nak()
# Delay avoids a tight redelivery loop if this is a persistent bug
await msg.nak(delay=5)
finally:
messages_in_flight.dec()
duration = asyncio.get_event_loop().time() - start_time
@@ -181,7 +191,7 @@ data:
async def main():
global nc, js, session, semaphore
print(f"?? Starting Worker for Domain: {NATS_CONSUMER}")
print(f"?? Stream: {NATS_STREAM}, Subjects: {FILTER_SUBJECTS}")
print(f"?? Concurrency: {WORKER_CONCURRENCY}")
@@ -202,12 +212,12 @@ data:
# Create Pull Subscriptions for each filter subject
# All sharing the same Consumer Name ensuring load balancing if multiple pods run this
subs = []
for subject in FILTER_SUBJECTS:
subject = subject.strip()
if not subject: continue
# FIX: On WorkQueue streams, we cannot reuse the same durable name for different filters
# We append a sanitized version of the subject to ensure uniqueness per filter
clean_subject_suffix = subject.replace(".", "_").replace("*", "all").replace(">", "all")
@@ -217,8 +227,9 @@ data:
try:
sub = await js.pull_subscribe(
subject,
durable=durable_name,
stream=NATS_STREAM
durable=durable_name,
stream=NATS_STREAM,
config=ConsumerConfig(ack_wait=ACK_WAIT_SECONDS)
)
subs.append(sub)
except Exception as e:
@@ -238,19 +249,24 @@ data:
# Batch size 10, short timeout to keep loop responsive
msgs = await sub.fetch(10, timeout=0.5)
for m in msgs:
asyncio.create_task(process_message(m))
task = asyncio.create_task(process_message(m))
active_tasks.add(task)
task.add_done_callback(active_tasks.discard)
except NatsTimeoutError:
pass
except Exception as e:
print(f"?? Fetch Error: {e}")
await asyncio.sleep(1)
# Small sleep to prevent tight loop if no messages
# await asyncio.sleep(0.01)
except Exception as e:
print(f"?? Fatal Error: {e}")
finally:
if active_tasks:
print(f"?? Draining {len(active_tasks)} in-flight message(s) before shutdown...")
await asyncio.wait(active_tasks, timeout=25)
if session:
await session.close()
if nc: