fix: resolve infinite retry and ffmpeg blocking issues

- Delete SQS message IMMEDIATELY upon receipt (at-most-once delivery)
- Redirect ffmpeg stderr to DEVNULL to prevent pipe buffer blocking
- Implement application-level retry using send_message with customRetryCount
- This prevents SQS visibility timeout issues and zombie processes

Author: hamlsy

echoshot_ai_server/core/sqs_client.py

@@ -133,6 +133,20 @@ def __init__(self):
         self.queue_url = settings.SQS_QUEUE_URL
         self.sqs_client = boto3.client('sqs', region_name=settings.AWS_REGION)
 
+    def send_message(self, message_body: Dict[str, Any], delay_seconds: int = 0) -> bool:
+        """메시지 전송 (재시도용)"""
+        try:
+            self.sqs_client.send_message(
+                QueueUrl=self.queue_url,
+                MessageBody=json.dumps(message_body),
+                DelaySeconds=delay_seconds
+            )
+            logger.info(f"Message sent to SQS queue (retry): job_id={message_body.get('job_id') or message_body.get('jobId')}")
+            return True
+        except ClientError as e:
+            logger.error(f"Failed to send message to SQS: {e}")
+            return False
+
     def receive_messages(self, max_messages: int = 1,
                          visibility_timeout: int = 300) -> List[Job]:
         """SQS 메시지 수신 및 Job 객체로 변환"""
@@ -176,14 +190,20 @@ def receive_messages(self, max_messages: int = 1,
                     parsed = parse_sqs_message_body(body)
 
                     # Job 객체 생성
+                    # 재시도 횟수: SQS 속성 또는 메시지 본문의 customRetryCount 사용
+                    sqs_retry_count = receive_count - 1 # receive_count는 1부터 시작
+                    custom_retry_count = body.get('customRetryCount', 0)
+                    final_retry_count = max(sqs_retry_count, custom_retry_count)
+
                     job = Job(
                         job_id=parsed['job_id'],
                         user_id=parsed['user_id'],
                         task_type=parsed['task_type'],
                         source_s3_key=parsed['source_s3_key'],
                         parameters=parsed['parameters'],
                         receipt_handle=msg['ReceiptHandle'],
-                        metadata=parsed.get('metadata')
+                        metadata=parsed.get('metadata'),
+                        retry_count=final_retry_count
                     )
                     jobs.append(job)
                     logger.debug(f"Successfully converted message to Job: {job.job_id} (task_type={job.task_type}, user_id={job.user_id}, s3_key={parsed['source_s3_key']})")

echoshot_ai_server/services/worker_pool.py

@@ -89,17 +89,28 @@ def _worker_loop(self, worker_id: int) -> None:
 
                 logger.info(f"Worker {worker_id} processing job {job.job_id}")
 
-                # Job 처리
+                # 1. 즉시 삭제 (At-most-once delivery 보장, 중복 실행 방지)
+                self.sqs_client.delete_message(job.receipt_handle)
+                logger.info(f"Job {job.job_id} message deleted from SQS immediately to prevent duplicate processing")
+
+                # 2. Job 처리
                 result = self.job_processor.process_job(job)
 
-                # SQS 메시지 삭제 (성공/실패 관계없이 항상 삭제)
-                # 재시도는 SQS의 ApproximateReceiveCount로 관리됨
-                deleted = self.sqs_client.delete_message(job.receipt_handle)
+                # 3. 실패 시 재시도 처리 (Application-level retry)
+                if result.status != JobStatus.COMPLETED:
+                    if job.retry_count < self.job_processor.max_retries:
+                        # 재시도 횟수 증가하여 새 메시지 발행
+                        retry_payload = job.to_dict()
+                        retry_payload['customRetryCount'] = job.retry_count + 1
+                        
+                        logger.warning(f"Job {job.job_id} failed. Re-queueing for retry ({retry_payload['customRetryCount']}/{self.job_processor.max_retries})")
+                        self.sqs_client.send_message(retry_payload, delay_seconds=60) # 60초 딜레이
+                    else:
+                        logger.error(f"Job {job.job_id} exceeded max retries ({job.retry_count}). Dropping message.")
 
+                # 성공 시에는 이미 삭제했으므로 추가 동작 없음
                 if result.status == JobStatus.COMPLETED:
-                    logger.info(f"Job {job.job_id} completed successfully, SQS message deleted={deleted}")
-                else:
-                    logger.warning(f"Job {job.job_id} failed, SQS message deleted={deleted}")
+                    logger.info(f"Job {job.job_id} completed successfully")
 
             except Empty:
                 continue

echoshot_ai_server/tasks/upscale_task.py

@@ -251,7 +251,7 @@ def _process_video(
             ffmpeg_cmd,
             stdin=subprocess.PIPE,
             stdout=subprocess.DEVNULL,
-            stderr=subprocess.PIPE
+            stderr=subprocess.DEVNULL  # 블로킹 방지를 위해 DEVNULL 사용
         )
 
         # 프레임 처리
@@ -289,8 +289,8 @@ def _process_video(
             ffmpeg_proc.wait()
 
             if ffmpeg_proc.returncode != 0:
-                stderr = ffmpeg_proc.stderr.read().decode()
-                raise RuntimeError(f"ffmpeg 오류: {stderr}")
+                # stderr를 DEVNULL로 보냈으므로 상세 에러 메시지는 확인 불가
+                raise RuntimeError(f"ffmpeg 오류 발생 (return code: {ffmpeg_proc.returncode})")
 
     def _build_ffmpeg_cmd(
             self,