메신저 만들기

CREATE TABLE room_seq (
  room_id BIGINT PRIMARY KEY,
  seq     BIGINT NOT NULL DEFAULT 0
);
 
CREATE TABLE messages (
  room_id       BIGINT NOT NULL,
  seq           BIGINT NOT NULL,
  client_msg_id TEXT   NOT NULL,
  sender_id     BIGINT NOT NULL,
  body          TEXT   NOT NULL,
  created_at    TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (room_id, seq),
  UNIQUE (room_id, client_msg_id)   -- 재전송 멱등성의 실제 강제 지점
);

room_locks: defaultdict[int, asyncio.Lock] = defaultdict(asyncio.Lock)
 
async def join_room(ws, room_id: int):
    async with room_locks[room_id]:
        first_local = not local_rooms[room_id]
        local_rooms[room_id].add(ws)
        if first_local:
            # 이 프로세스에 이 방 사람이 처음 생겼을 때만 Redis 구독
            await pubsub.subscribe(f"room:{room_id}")
 
async def leave_room(ws, room_id: int):
    async with room_locks[room_id]:
        local_rooms[room_id].discard(ws)
        if not local_rooms[room_id]:
            await pubsub.unsubscribe(f"room:{room_id}")
            if not local_rooms[room_id]:      # await 사이에 재입장했는지 확인
                del local_rooms[room_id]
            else:
                await pubsub.subscribe(f"room:{room_id}")

async def handle_send(room_id: int, sender_id: int, client_msg_id: str, body: str):
    # 0) 재전송 확인 — 채번 전에 본다. 여기서 걸러야 seq가 낭비되지 않는다
    seq = await db.fetchval(
        "SELECT seq FROM messages WHERE room_id = $1 AND client_msg_id = $2",
        room_id, client_msg_id,
    )
 
    if seq is None:
        # 1) 채번 + 저장을 한 트랜잭션으로. 여기까지 커밋되면 메시지는 살아남는다
        async with db.transaction():
            seq = await db.fetchval("""
                INSERT INTO room_seq (room_id, seq) VALUES ($1, 1)
                ON CONFLICT (room_id) DO UPDATE SET seq = room_seq.seq + 1
                RETURNING seq
            """, room_id)
            await db.execute("""
                INSERT INTO messages (room_id, seq, client_msg_id, sender_id, body)
                VALUES ($1, $2, $3, $4, $5)
                ON CONFLICT (room_id, client_msg_id) DO NOTHING
            """, room_id, seq, client_msg_id, sender_id, body)
 
    payload = {"roomId": room_id, "seq": seq, "clientMsgId": client_msg_id,
               "senderId": sender_id, "body": body}
 
    # 2) 부가 처리 팬아웃. 실패해도 메시지는 DB에 있으므로 복구 경로가 메꾼다
    await producer.send_and_wait("chat.messages", key=room_id, value=payload)
 
    # 3) 실시간 팬아웃
    await redis.publish(f"room:{room_id}", payload)
    return payload

acks="all"                 # ISR 전체 반영 확인
enable_idempotence=True    # 재시도 시 파티션 내 중복/재정렬 방지
max_in_flight_requests_per_connection=5   # 멱등 활성 시 5까지 순서 안전

def spawn_pubsub_reader():
    task = asyncio.create_task(pubsub_reader())
    task.add_done_callback(on_reader_dead)
    return task
 
def on_reader_dead(task: asyncio.Task):
    if shutting_down.is_set() or task.cancelled():
        return
    log.error("pubsub_reader died", exc_info=task.exception())
    asyncio.get_running_loop().call_later(1.0, resubscribe_and_respawn)
 
async def resubscribe_and_respawn():
    await pubsub.reset()
    for room_id in list(local_rooms.keys()):     # 이 파드가 들고 있던 방 전부 재구독
        await pubsub.subscribe(f"room:{room_id}")
    spawn_pubsub_reader()

terminationGracePeriodSeconds: 300
readinessProbe:
  httpGet: { path: /readyz, port: 8080 } # `/readyz`는 `shutting_down.is_set()`이면 503을 반환
  periodSeconds: 2
  failureThreshold: 1                    # 즉시 NotReady. 기본값 3이면 6초를 더 기다린다
strategy:
  rollingUpdate:
    maxSurge: 2
    maxUnavailable: 0 # 새 파드가 Ready된 다음에 구 파드를 빼야 재접속을 받아줌

import signal, asyncio, random
 
shutting_down = asyncio.Event()
 
async def graceful_shutdown():
    if shutting_down.is_set():   # SIGTERM이 두 번 와도 태스크는 하나만
        return
    shutting_down.set()          # readinessProbe가 이제 503 반환
 
    # 1) LB 타겟 등록 해제 전파 대기
    #    Endpoints 갱신 + kube-proxy/LB 반영까지. ALB면 deregistration_delay도 별도로 있다
    #    이 동안에도 기존 커넥션은 정상 동작한다
    await asyncio.sleep(25)
 
    # 2) 커넥션을 천천히 빼기 (초당 약 1000소켓)
    all_ws = [ws for socks in local_rooms.values() for ws in socks]
    random.shuffle(all_ws)
 
    for i, ws in enumerate(all_ws):
        await ws.send_json({"type": "reconnect", "after_ms": random.randint(0, 60_000)})
        await ws.close(code=1001)
        if i % 500 == 0:
            await asyncio.sleep(0.5)
 
    # 3) 정리
    await pubsub.close()
    await producer.stop()
 
loop.add_signal_handler(signal.SIGTERM, lambda: asyncio.create_task(graceful_shutdown()))

members = await db.fetch("SELECT user_id FROM room_members WHERE room_id = $1", room_id)