HISTORY: P0.1 таймауты + P0.2 рестарт PASS + найден дефект воскрешения удалённых очередей

HISTORY/2026-08-14-session-log.md

@@ -771,3 +771,23 @@ MSS clamp 1448, PMTUD мёртв). Кластером (kubectl) НЕ чинит
 прокси на ВМ с MTU 1400 на ens192 (вход на ВМ чист — факт drhider; выход ВМ→кластер
 починится уменьшением MTU ВМ); 3) временным лимитом размера в сервисе.
 
+## План Соннета — выполнение (15.08.2026)
+
+**Backup**: ветка `backup-v0.1.35-2026-08-15` (запушена в Gitea), работа в `main`.
+
+**P0.1 Серверные таймауты — СДЕЛАНО** (коммит `af0325e`, digest `sha256:4385597f…`):
+`http.Server{ReadHeaderTimeout:10s, WriteTimeout:35s, IdleTimeout:90s, MaxHeaderBytes:8192}`,
+`ReadTimeout` убран (не рвать большие/медленные тела и long-poll).
+
+**P0.2 Рестарт с in-flight — СДЕЛАНО, PASS** (`tests/reboot_probe.py`):
+10 очередей × 20 сообщений + 100 сообщений (50 in-flight) → `kubectl delete pod
+--grace-period=0 --force` → после подъёма: /health ok; restore 10×20 OK;
+main: получено 100, уникальных 100, пропущено 0, дублей 0.
+
+**НАЙДЕН ДЕФЕКТ (критичный)**: после деплоя таймаутов под рестартовал — и у
+тенанта «воскресли» 50 удалённых ранее очередей (лимит 50 → LimitExceeded).
+Причина: `DeleteQueue` пишет удаление в Redis АСИНХРОННО (asyncWrite); при
+SIGKILL/рестарте пода запись не успевает → очередь остаётся в Redis →
+`LoadAllQueues` восстанавливает её. Удаления должны быть синхронными
+(хотя бы DEL очереди) — НЕ ИСПРАВЛЕНО, задокументировано.
+

tests/reboot_probe.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+"""tests/reboot_probe.py — тест рестарта пода с in-flight сообщениями (план Соннета P0).
+
+Фаза 1 (phase1): 10 очередей x 20 сообщений (проверка восстановления из Redis)
++ основная очередь со 100 сообщениями, из которых 50 выдаются (in-flight).
+Фаза 2 (phase2, ПОСЛЕ kubectl delete pod --grace-period=0):
+- /health поднимается;
+- во всех 10 очередях должны быть ровно по 20 сообщений;
+- из основной должны выдаться все 100 уникальных тел (50 in-flight + 50 видимых).
+"""
+import json
+import os
+import sys
+import time
+import uuid
+
+import boto3
+import urllib.request
+from botocore.config import Config
+
+ENDPOINT = os.environ.get("ENDPOINT_URL", "https://sqs.containerk8s.dev.nubes.ru")
+REGION = os.environ.get("REGION", "us-east-1")
+STATE = "/tmp/reboot_state.json"
+
+sqs = boto3.client("sqs", endpoint_url=ENDPOINT, region_name=REGION,
+                   config=Config(connect_timeout=10, read_timeout=30, retries={"max_attempts": 3}))
+
+
+def wait_health(timeout=180):
+    t0 = time.time()
+    while time.time() - t0 < timeout:
+        try:
+            with urllib.request.urlopen(ENDPOINT + "/health", timeout=10) as r:
+                if r.status == 200:
+                    return True
+        except Exception:
+            pass
+        time.sleep(3)
+    return False
+
+
+def drain(url):
+    got = []
+    while True:
+        msgs = sqs.receive_message(QueueUrl=url, MaxNumberOfMessages=10).get("Messages", [])
+        if not msgs:
+            break
+        got.extend(msgs)
+    return got
+
+
+def phase1():
+    uid = str(uuid.uuid4())[:8]
+    state = {"uid": uid, "restore_queues": [], "main_queue": None, "bodies": []}
+
+    # 10 очередей по 20 сообщений — проверка восстановления из Redis
+    for i in range(10):
+        u = sqs.create_queue(QueueName="reboot-restore-%d-%s" % (i, uid))["QueueUrl"]
+        for j in range(20):
+            sqs.send_message(QueueUrl=u, MessageBody="r-%d-%d-%s" % (i, j, uid))
+        state["restore_queues"].append(u)
+        print("restore queue %d ready" % i, flush=True)
+
+    # основная: 100 сообщений, 50 выдаём (in-flight)
+    u = sqs.create_queue(QueueName="reboot-main-%s" % uid)["QueueUrl"]
+    bodies = ["m-%03d-%s" % (i, uid) for i in range(100)]
+    for b in bodies:
+        sqs.send_message(QueueUrl=u, MessageBody=b)
+    first = sqs.receive_message(QueueUrl=u, MaxNumberOfMessages=10, WaitTimeSeconds=1).get("Messages", [])
+    # добираем до 50 in-flight (Max=10 за вызов)
+    while len(first) < 50:
+        batch = sqs.receive_message(QueueUrl=u, MaxNumberOfMessages=10, WaitTimeSeconds=1).get("Messages", [])
+        if not batch:
+            break
+        first.extend(batch)
+    print("in-flight на момент рестарта: %d" % len(first), flush=True)
+    state["main_queue"] = u
+    state["bodies"] = bodies
+    json.dump(state, open(STATE, "w"))
+    print("PHASE1_DONE", flush=True)
+
+
+def phase2():
+    st = json.load(open(STATE))
+    print("ждём /health после рестарта...", flush=True)
+    if not wait_health():
+        print("HEALTH_FAIL: /health не поднялся", flush=True)
+        sys.exit(1)
+    print("health ok", flush=True)
+
+    # восстановление 10 очередей
+    restore_ok = True
+    for u in st["restore_queues"]:
+        msgs = drain(u)
+        if len(msgs) != 20:
+            restore_ok = False
+            print("RESTORE FAIL: %s -> %d сообщений" % (u.split("/")[-1], len(msgs)), flush=True)
+    print("restore: %s" % ("OK (10x20)" if restore_ok else "FAIL"), flush=True)
+
+    # основная: все 100 уникальных
+    got = drain(st["main_queue"])
+    got_bodies = {m["Body"] for m in got}
+    want = set(st["bodies"])
+    missing = want - got_bodies
+    dups = len(got) - len(got_bodies)
+    print("main: получено %d, уникальных %d, пропущено %d, дублей %d"
+          % (len(got), len(got_bodies), len(missing), dups), flush=True)
+    print("MAIN_RESULT: %s" % ("PASS" if not missing else "FAIL (потеря!)"), flush=True)
+
+    # cleanup
+    for u in st["restore_queues"] + [st["main_queue"]]:
+        try:
+            sqs.delete_queue(QueueUrl=u)
+        except Exception:
+            pass
+    print("PHASE2_DONE", flush=True)
+
+
+if __name__ == "__main__":
+    if len(sys.argv) != 2 or sys.argv[1] not in ("phase1", "phase2"):
+        print("usage: reboot_probe.py phase1|phase2")
+        sys.exit(2)
+    if sys.argv[1] == "phase1":
+        phase1()
+    else:
+        phase2()