확인·기한·회수의 주인을 정한다

한 줄 요약

전송 버퍼가 비었다는 사실, 업무가 끝났다는 확인, 연결을 회수했다는 사실은 서로 다른 증거입니다.

왜 이게 필요했나

서버가 write를 마친 뒤 성공 로그를 남겼는데 사용자의 화면은 바뀌지 않습니다. 바이트가 운영체제에 맡겨진 것과 상대 프로그램이 업무를 마친 것은 다르기 때문입니다. 반대로 상대는 처리했지만 ACK가 오기 전에 연결이 끊겼을 수도 있습니다. 실시간 프로그램은 이 불확실성을 숨기지 않고 메시지 계약·기한·복구 위치로 표현해야 합니다.

이번 계약은 아주 작습니다. 서버는 ASCII로 EVENT 7 뒤에 개행을 보내고, 클라이언트는 해당 업무를 끝냈다고 가정한 다음 ACK 7 뒤에 개행을 돌려줍니다. 순번은 0부터 2147483647까지의 정수입니다. 인증된 업무 프로토콜이나 영속 메시지 브로커가 아니라, 확인과 수명을 배울 수 있는 교육용 프로토콜입니다.

어떻게 동작하나

StreamWriter.write는 데이터를 쓰기 버퍼에 넣습니다. await drain은 전송 흐름 제어가 허용할 때까지 기다립니다. drain의 완료가 상대 애플리케이션의 완료는 아니므로, 이어서 reader.readline으로 같은 순번의 ACK를 기다립니다. EOF, 다른 순번, 잘못된 형식은 ConnectionError로 처리합니다. 전송 함수는 여러 이벤트가 쓰는 지속 연결을 임의로 닫지 않습니다.

시간 예산은 이벤트 하나에 한 번 부여합니다. 전체 1초인데 drain에서 0.8초를 썼다면 ACK에는 약 0.2초만 남습니다. 각 await에 새로 1초를 주면 최대 대기 시간이 작업 수에 따라 늘어납니다. 시작 때 monotonic 시계로 마감 시각을 정하고 매 대기 전에 남은 시간을 계산하세요. 마지막 결과가 돌아온 뒤에도 기한을 넘지 않았는지 확인합니다. 벽시계의 시각 보정은 경과 시간 판정에 쓰지 않습니다.

큐 작업자는 get으로 항목 하나를 인수한 뒤 send_one을 실행합니다. 성공이든 예외든 인수한 항목의 정리는 finally에서 task_done으로 끝냅니다. ACK 전에 task_done을 호출하면 발행 측의 join이 실제 확인보다 먼저 풀립니다. 반대로 예외 때 task_done을 빠뜨리면 작업은 죽었는데 join만 계속 기다립니다. 정리와 업무 성공은 다른 기록이라는 점을 여기서 다시 적용합니다.

작업자 함수가 연결의 소유권을 인수했다면 빈 큐에서 취소되거나 ACK 대기 중 실패해도 writer.close를 호출해야 합니다. wait_closed에도 유한한 기한을 두고, 종료가 끝나지 않으면 transport.abort로 강제 회수합니다. 큐에 아직 남아 있는 항목의 폐기·재시도는 구독자 관리자의 책임입니다. 작업자 하나가 자기 처리 중 항목을 정리하는 것과 전체 큐를 비우는 것을 섞지 않습니다.

재접속에는 마지막으로 확실히 처리한 순번이 필요합니다. Cursor(last=7)에 8이 오면 연속 진행이고 7이 다시 오면 중복입니다. 10이 먼저 오면 8과 9가 비어 있으므로 Gap을 내고 last를 움직이지 않습니다. 로그 재생이나 신뢰할 수 있는 스냅샷으로 누락을 복구한 뒤 진행해야 합니다. 그냥 큰 순번을 채택하면 누락이 영구적으로 숨겨집니다.

여기서 만드는 Cursor는 메모리 안에서 연속성을 판단하는 작은 부품입니다. accept가 True이면 last를 즉시 전진시키므로, 실패할 수 있는 실제 결제 함수를 호출하기 전에 그대로 사용하면 안 됩니다. 업무 적용과 체크포인트를 원자적으로 저장하는 기능은 이 코스에 없습니다. 프로세스 재시작 시 메모리도 사라집니다. 중복 판단 시험이 통과했다고 정확히 한 번 처리를 보장하지 않습니다.

현장에서 만나는 모습

운영에서는 구독자 연결 해제, 발행자 취소, 프로세스 종료가 서로 다른 시점에 겹칩니다. 정상 ACK 시험만 성공하면 이런 종료 경로의 누수를 놓칩니다. 반복해서 연결을 끊어도 소켓과 태스크가 남지 않는지 확인해야 합니다. 리스너를 먼저 닫아 새 연결을 막고, 기존 태스크와 연결을 정리한 뒤 서버 종료를 기다리는 순서도 중요합니다. Python 3.12의 Server.wait_closed는 활성 연결 종료를 기다립니다.

마지막 실제 TCP 시험은 두 구독자의 부분 장애와 회수를 관측합니다. 앞 단계는 가짜 시계로 전체 기한을 결정적으로 검사하고, ACK가 영원히 오지 않는 경우도 실제 비동기 대기로 검사합니다. 두 방식은 대체 관계가 아닙니다. 가짜 시계는 경계 계산을, 실제 연결은 배선과 수명을 확인합니다.

다음 실습에서 할 것

기한 시험에서는 1초를 실제로 여러 번 기다리는 대신 주입한 시계를 움직여 경계를 검사할 수 있습니다. 그러나 가짜 reader가 즉시 반환하면 무한 대기를 끊는 코드가 없어도 그 예만 통과할 수 있습니다. 그래서 영원히 반환하지 않는 reader도 별도로 사용합니다. 실패 시험이 오래 걸린다고 제거하면 바로 그 결함이 운영에서 무한 대기가 됩니다. 계산 경계와 실제 취소를 모두 남기는 이유입니다.

8단계에서 큐 정책·커서·ACK 전체 기한·작업자·broadcast를 완성합니다. 모든 코드는 /root/realtime/delivery.py에 두고, 서버는 검사기가 임시 포트로 준비합니다. 정답을 보기 전에 실패 메시지가 어느 계층의 증거를 요구하는지 먼저 읽어 보세요.