diff --git a/app/home/api/routers/v1/home.py b/app/home/api/routers/v1/home.py
index 357e0cf..2359088 100644
--- a/app/home/api/routers/v1/home.py
+++ b/app/home/api/routers/v1/home.py
@@ -24,7 +24,7 @@ from app.home.schemas.home_schema import (
     ProcessedInfo,
 )
 from app.utils.upload_blob_as_request import AzureBlobUploader
-from app.utils.chatgpt_prompt import ChatgptService
+from app.utils.chatgpt_prompt import ChatgptService, ChatGPTResponseError
 from app.utils.common import generate_task_id
 from app.utils.logger import get_logger
 from app.utils.nvMapScraper import NvMapScraper, GraphQLException
@@ -64,7 +64,8 @@ KOREAN_CITIES = [
 ]
 # fmt: on
 
-router = APIRouter(tags=["Home"])
+# router = APIRouter(tags=["Home"])
+router = APIRouter()
 
 
 def _extract_region_from_address(road_address: str | None) -> str:
@@ -291,6 +292,15 @@ async def _crawling_logic(url:str):
                 f"[crawling] Step 3 완료 - 마케팅 분석 성공 ({step3_elapsed:.1f}ms)"
             )
 
+        except ChatGPTResponseError as e:
+            step3_elapsed = (time.perf_counter() - step3_start) * 1000
+            logger.error(
+                f"[crawling] Step 3 FAILED - ChatGPT Error: status={e.status}, "
+                f"code={e.error_code}, message={e.error_message} ({step3_elapsed:.1f}ms)"
+            )
+            marketing_analysis = None
+            gpt_status = "failed"
+
         except Exception as e:
             step3_elapsed = (time.perf_counter() - step3_start) * 1000
             logger.error(
@@ -299,6 +309,7 @@ async def _crawling_logic(url:str):
             logger.exception("[crawling] Step 3 상세 오류:")
             # GPT 실패 시에도 크롤링 결과는 반환
             marketing_analysis = None
+            gpt_status = "failed"
     else:
         step2_elapsed = (time.perf_counter() - step2_start) * 1000
         logger.warning(
@@ -318,6 +329,7 @@ async def _crawling_logic(url:str):
                 logger.info(f"[crawling]       - GPT API 호출: {step3_3_elapsed:.1f}ms")
 
     return {
+        "status": gpt_status if 'gpt_status' in locals() else "completed",
         "image_list": scraper.image_link_list,
         "image_count": len(scraper.image_link_list) if scraper.image_link_list else 0,
         "processed_info": processed_info,
diff --git a/app/home/schemas/home_schema.py b/app/home/schemas/home_schema.py
index bd00be9..f308c20 100644
--- a/app/home/schemas/home_schema.py
+++ b/app/home/schemas/home_schema.py
@@ -167,13 +167,37 @@ class MarketingAnalysis(BaseModel):
 class CrawlingResponse(BaseModel):
     """크롤링 응답 스키마"""
 
+    model_config = ConfigDict(
+        json_schema_extra={
+            "example": {
+                "status": "completed",
+                "image_list": ["https://example.com/image1.jpg", "https://example.com/image2.jpg"],
+                "image_count": 2,
+                "processed_info": {
+                    "customer_name": "스테이 머뭄",
+                    "region": "군산",
+                    "detail_region_info": "전북특별자치도 군산시 절골길 18"
+                },
+                "marketing_analysis": {
+                    "report": "마케팅 분석 리포트...",
+                    "tags": ["힐링", "감성숙소"],
+                    "facilities": ["조식", "주차"]
+                }
+            }
+        }
+    )
+
+    status: str = Field(
+        default="completed",
+        description="처리 상태 (completed: 성공, failed: ChatGPT 분석 실패)"
+    )
     image_list: Optional[list[str]] = Field(None, description="이미지 URL 목록")
     image_count: int = Field(..., description="이미지 개수")
     processed_info: Optional[ProcessedInfo] = Field(
         None, description="가공된 장소 정보 (customer_name, region, detail_region_info)"
     )
     marketing_analysis: Optional[MarketingAnalysis] = Field(
-        None, description="마케팅 분석 결과 (report, tags, facilities)"
+        None, description="마케팅 분석 결과 (report, tags, facilities). 실패 시 null"
     )
 
 
diff --git a/app/lyric/schemas/lyric.py b/app/lyric/schemas/lyric.py
index 37cb76a..718f724 100644
--- a/app/lyric/schemas/lyric.py
+++ b/app/lyric/schemas/lyric.py
@@ -108,15 +108,33 @@ class LyricStatusResponse(BaseModel):
     Usage:
         GET /lyric/status/{task_id}
         Returns the current processing status of a lyric generation task.
+
+    Status Values:
+        - processing: 가사 생성 진행 중
+        - completed: 가사 생성 완료
+        - failed: ChatGPT API 오류 또는 생성 실패
     """
 
     model_config = ConfigDict(
         json_schema_extra={
-            "example": {
-                "task_id": "0694b716-dbff-7219-8000-d08cb5fce431",
-                "status": "completed",
-                "message": "가사 생성이 완료되었습니다.",
-            }
+            "examples": [
+                {
+                    "summary": "성공",
+                    "value": {
+                        "task_id": "0694b716-dbff-7219-8000-d08cb5fce431",
+                        "status": "completed",
+                        "message": "가사 생성이 완료되었습니다.",
+                    }
+                },
+                {
+                    "summary": "실패",
+                    "value": {
+                        "task_id": "0694b716-dbff-7219-8000-d08cb5fce431",
+                        "status": "failed",
+                        "message": "가사 생성에 실패했습니다.",
+                    }
+                }
+            ]
         }
     )
 
@@ -131,26 +149,46 @@ class LyricDetailResponse(BaseModel):
     Usage:
         GET /lyric/{task_id}
         Returns the generated lyric content for a specific task.
+
+    Note:
+        - status가 "failed"인 경우 lyric_result에 에러 메시지가 저장됩니다.
+        - 에러 메시지 형식: "ChatGPT Error: {message}" 또는 "Error: {message}"
     """
 
     model_config = ConfigDict(
         json_schema_extra={
-            "example": {
-                "id": 1,
-                "task_id": "0694b716-dbff-7219-8000-d08cb5fce431",
-                "project_id": 1,
-                "status": "completed",
-                "lyric_result": "인스타 감성의 스테이 머뭄\n군산 신흥동 말랭이 마을에서\n여유로운 하루를 보내며\n추억을 만들어가요",
-                "created_at": "2024-01-15T12:00:00",
-            }
+            "examples": [
+                {
+                    "summary": "성공",
+                    "value": {
+                        "id": 1,
+                        "task_id": "0694b716-dbff-7219-8000-d08cb5fce431",
+                        "project_id": 1,
+                        "status": "completed",
+                        "lyric_result": "인스타 감성의 스테이 머뭄\n군산 신흥동 말랭이 마을에서\n여유로운 하루를 보내며\n추억을 만들어가요",
+                        "created_at": "2024-01-15T12:00:00",
+                    }
+                },
+                {
+                    "summary": "실패",
+                    "value": {
+                        "id": 1,
+                        "task_id": "0694b716-dbff-7219-8000-d08cb5fce431",
+                        "project_id": 1,
+                        "status": "failed",
+                        "lyric_result": "ChatGPT Error: Response incomplete: max_output_tokens",
+                        "created_at": "2024-01-15T12:00:00",
+                    }
+                }
+            ]
         }
     )
 
     id: int = Field(..., description="가사 ID")
     task_id: str = Field(..., description="작업 고유 식별자")
     project_id: int = Field(..., description="프로젝트 ID")
-    status: str = Field(..., description="처리 상태")
-    lyric_result: Optional[str] = Field(None, description="생성된 가사")
+    status: str = Field(..., description="처리 상태 (processing, completed, failed)")
+    lyric_result: Optional[str] = Field(None, description="생성된 가사 또는 에러 메시지 (실패 시)")
     created_at: Optional[datetime] = Field(None, description="생성 일시")
 
 
diff --git a/app/lyric/worker/lyric_task.py b/app/lyric/worker/lyric_task.py
index f956164..5be9a95 100644
--- a/app/lyric/worker/lyric_task.py
+++ b/app/lyric/worker/lyric_task.py
@@ -11,7 +11,7 @@ from sqlalchemy.exc import SQLAlchemyError
 
 from app.database.session import BackgroundSessionLocal
 from app.lyric.models import Lyric
-from app.utils.chatgpt_prompt import ChatgptService
+from app.utils.chatgpt_prompt import ChatgptService, ChatGPTResponseError
 from app.utils.prompts.prompts import Prompt
 from app.utils.logger import get_logger
 
@@ -130,6 +130,14 @@ async def generate_lyric_background(
         logger.debug(f"[generate_lyric_background]   - Step 2 (GPT API 호출): {step2_elapsed:.1f}ms")
         logger.debug(f"[generate_lyric_background]   - Step 3 (DB 업데이트): {step3_elapsed:.1f}ms")
 
+    except ChatGPTResponseError as e:
+        elapsed = (time.perf_counter() - task_start) * 1000
+        logger.error(
+            f"[generate_lyric_background] ChatGPT ERROR - task_id: {task_id}, "
+            f"status: {e.status}, code: {e.error_code}, message: {e.error_message} ({elapsed:.1f}ms)"
+        )
+        await _update_lyric_status(task_id, "failed", f"ChatGPT Error: {e.error_message}")
+
     except SQLAlchemyError as e:
         elapsed = (time.perf_counter() - task_start) * 1000
         logger.error(f"[generate_lyric_background] DB ERROR - task_id: {task_id}, error: {e} ({elapsed:.1f}ms)", exc_info=True)
diff --git a/app/user/services/auth.py b/app/user/services/auth.py
index d4cec00..fe83bfb 100644
--- a/app/user/services/auth.py
+++ b/app/user/services/auth.py
@@ -9,6 +9,7 @@ from datetime import datetime, timezone
 from typing import Optional
 
 from sqlalchemy import select, update
+from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from config import prj_settings
@@ -276,10 +277,44 @@ class AuthService:
             thumbnail_image_url=profile.thumbnail_image_url if profile else None,
         )
         session.add(new_user)
-        await session.flush()
-        await session.refresh(new_user)
-        logger.info(f"[AUTH] 신규 사용자 생성 완료 - user_id: {new_user.id}, is_new_user: True")
-        return new_user, True
+
+        try:
+            await session.flush()
+            await session.refresh(new_user)
+            logger.info(f"[AUTH] 신규 사용자 생성 완료 - user_id: {new_user.id}, is_new_user: True")
+            return new_user, True
+        except IntegrityError:
+            # 동시 요청으로 인한 중복 삽입 시도 - 기존 사용자 조회
+            logger.warning(
+                f"[AUTH] IntegrityError 발생 (동시 요청 추정) - kakao_id: {kakao_id}, "
+                "기존 사용자 재조회 시도"
+            )
+            await session.rollback()
+            result = await session.execute(
+                select(User).where(User.kakao_id == kakao_id)
+            )
+            existing_user = result.scalar_one_or_none()
+
+            if existing_user is not None:
+                logger.info(
+                    f"[AUTH] 기존 사용자 재조회 성공 - user_id: {existing_user.id}, "
+                    "is_new_user: False"
+                )
+                # 프로필 정보 업데이트
+                if profile:
+                    existing_user.nickname = profile.nickname
+                    existing_user.profile_image_url = profile.profile_image_url
+                    existing_user.thumbnail_image_url = profile.thumbnail_image_url
+                if kakao_account and kakao_account.email:
+                    existing_user.email = kakao_account.email
+                await session.flush()
+                return existing_user, False
+
+            # 재조회에도 실패한 경우 (매우 드문 경우)
+            logger.error(
+                f"[AUTH] IntegrityError 후 재조회 실패 - kakao_id: {kakao_id}"
+            )
+            raise
 
     async def _save_refresh_token(
         self,
diff --git a/app/utils/chatgpt_prompt.py b/app/utils/chatgpt_prompt.py
index ac88b3b..cbb047b 100644
--- a/app/utils/chatgpt_prompt.py
+++ b/app/utils/chatgpt_prompt.py
@@ -4,45 +4,80 @@ from pydantic import BaseModel
 from openai import AsyncOpenAI
 
 from app.utils.logger import get_logger
-from config import apikey_settings
+from config import apikey_settings, recovery_settings
 from app.utils.prompts.prompts import Prompt
 
 
 # 로거 설정
 logger = get_logger("chatgpt")
 
-# fmt: on
+
+class ChatGPTResponseError(Exception):
+    """ChatGPT API 응답 에러"""
+    def __init__(self, status: str, error_code: str = None, error_message: str = None):
+        self.status = status
+        self.error_code = error_code
+        self.error_message = error_message
+        super().__init__(f"ChatGPT response failed: status={status}, code={error_code}, message={error_message}")
 
 
 class ChatgptService:
     """ChatGPT API 서비스 클래스
-
-    GPT 5.0 모델을 사용하여 마케팅 가사 및 분석을 생성합니다.
     """
 
-    def __init__(self):
-        self.client = AsyncOpenAI(api_key=apikey_settings.CHATGPT_API_KEY)
-        
-    async def _call_structured_output_with_response_gpt_api(self, prompt: str, output_format : dict, model:str) -> dict:
-        content = [{"type": "input_text", "text": prompt}]
-        response = await self.client.responses.create(
-            model=model, 
-            input=[{"role": "user", "content": content}],
-            text = output_format
+    def __init__(self, timeout: float = None):
+        self.timeout = timeout or recovery_settings.CHATGPT_TIMEOUT
+        self.max_retries = recovery_settings.CHATGPT_MAX_RETRIES
+        self.client = AsyncOpenAI(
+            api_key=apikey_settings.CHATGPT_API_KEY,
+            timeout=self.timeout
         )
-        structured_output = json.loads(response.output_text)
-        return structured_output or {}
     
     async def _call_pydantic_output(self, prompt : str, output_format : BaseModel, model : str) -> BaseModel: # 입력 output_format의 경우 Pydantic BaseModel Class를 상속한 Class 자체임에 유의할 것
         content = [{"type": "input_text", "text": prompt}]
-        response = await self.client.responses.parse(
-            model=model, 
-            input=[{"role": "user", "content": content}],
-            text_format=output_format
-        )
-        structured_output = response.output_parsed
-        return structured_output.model_dump() or {}
+        last_error = None
+        for attempt in range(self.max_retries + 1):
+            response = await self.client.responses.parse(
+                model=model, 
+                input=[{"role": "user", "content": content}],
+                text_format=output_format
+            )
+            # Response 디버그 로깅
+            logger.debug(f"[ChatgptService] Response ID: {response.id}")
+            logger.debug(f"[ChatgptService] Response status: {response.status}")
+            logger.debug(f"[ChatgptService] Response model: {response.model}")
 
+            # status 확인: completed, failed, incomplete, cancelled, queued, in_progress
+            if response.status == "completed":
+                logger.debug(f"[ChatgptService] Response output_text: {response.output_text[:200]}..." if len(response.output_text) > 200 else f"[ChatgptService] Response output_text: {response.output_text}")
+                structured_output = response.output_parsed
+                return structured_output.model_dump() or {}
+
+            # 에러 상태 처리
+            if response.status == "failed":
+                error_code = getattr(response.error, 'code', None) if response.error else None
+                error_message = getattr(response.error, 'message', None) if response.error else None
+                logger.warning(f"[ChatgptService] Response failed (attempt {attempt + 1}/{self.max_retries + 1}): code={error_code}, message={error_message}")
+                last_error = ChatGPTResponseError(response.status, error_code, error_message)
+
+            elif response.status == "incomplete":
+                reason = getattr(response.incomplete_details, 'reason', None) if response.incomplete_details else None
+                logger.warning(f"[ChatgptService] Response incomplete (attempt {attempt + 1}/{self.max_retries + 1}): reason={reason}")
+                last_error = ChatGPTResponseError(response.status, reason, f"Response incomplete: {reason}")
+
+            else:
+                # cancelled, queued, in_progress 등 예상치 못한 상태
+                logger.warning(f"[ChatgptService] Unexpected response status (attempt {attempt + 1}/{self.max_retries + 1}): {response.status}")
+                last_error = ChatGPTResponseError(response.status, None, f"Unexpected status: {response.status}")
+
+            # 마지막 시도가 아니면 재시도
+            if attempt < self.max_retries:
+                logger.info(f"[ChatgptService] Retrying request...")
+
+        # 모든 재시도 실패
+        logger.error(f"[ChatgptService] All retries exhausted. Last error: {last_error}")
+        raise last_error
+    
     async def generate_structured_output(
         self,
         prompt : Prompt,
diff --git a/app/utils/creatomate.py b/app/utils/creatomate.py
index f333dfb..a1bfe54 100644
--- a/app/utils/creatomate.py
+++ b/app/utils/creatomate.py
@@ -36,12 +36,29 @@ from typing import Literal
 import httpx
 
 from app.utils.logger import get_logger
-from config import apikey_settings, creatomate_settings
+from config import apikey_settings, creatomate_settings, recovery_settings
 
 # 로거 설정
 logger = get_logger("creatomate")
 
 
+class CreatomateResponseError(Exception):
+    """Creatomate API 응답 오류 시 발생하는 예외
+
+    Creatomate API 렌더링 실패 또는 비정상 응답 시 사용됩니다.
+    재시도 로직에서 이 예외를 catch하여 재시도를 수행합니다.
+
+    Attributes:
+        message: 에러 메시지
+        original_response: 원본 API 응답 (있는 경우)
+    """
+
+    def __init__(self, message: str, original_response: dict | None = None):
+        self.message = message
+        self.original_response = original_response
+        super().__init__(self.message)
+
+
 # Orientation 타입 정의
 OrientationType = Literal["horizontal", "vertical"]
 
@@ -135,7 +152,10 @@ async def get_shared_client() -> httpx.AsyncClient:
     global _shared_client
     if _shared_client is None or _shared_client.is_closed:
         _shared_client = httpx.AsyncClient(
-            timeout=httpx.Timeout(60.0, connect=10.0),
+            timeout=httpx.Timeout(
+                recovery_settings.CREATOMATE_RENDER_TIMEOUT,
+                connect=recovery_settings.CREATOMATE_CONNECT_TIMEOUT,
+            ),
             limits=httpx.Limits(max_keepalive_connections=10, max_connections=20),
         )
     return _shared_client
@@ -217,7 +237,7 @@ class CreatomateService:
         self,
         method: str,
         url: str,
-        timeout: float = 30.0,
+        timeout: float | None = None,
         **kwargs,
     ) -> httpx.Response:
         """HTTP 요청을 수행합니다.
@@ -225,7 +245,7 @@ class CreatomateService:
         Args:
             method: HTTP 메서드 ("GET", "POST", etc.)
             url: 요청 URL
-            timeout: 요청 타임아웃 (초)
+            timeout: 요청 타임아웃 (초). None이면 기본값 사용
             **kwargs: httpx 요청에 전달할 추가 인자
 
         Returns:
@@ -236,15 +256,18 @@ class CreatomateService:
         """
         logger.info(f"[Creatomate] {method} {url}")
 
+        # timeout이 None이면 기본 타임아웃 사용
+        actual_timeout = timeout if timeout is not None else recovery_settings.CREATOMATE_DEFAULT_TIMEOUT
+
         client = await get_shared_client()
 
         if method.upper() == "GET":
             response = await client.get(
-                url, headers=self.headers, timeout=timeout, **kwargs
+                url, headers=self.headers, timeout=actual_timeout, **kwargs
             )
         elif method.upper() == "POST":
             response = await client.post(
-                url, headers=self.headers, timeout=timeout, **kwargs
+                url, headers=self.headers, timeout=actual_timeout, **kwargs
             )
         else:
             raise ValueError(f"Unsupported HTTP method: {method}")
@@ -255,7 +278,7 @@ class CreatomateService:
     async def get_all_templates_data(self) -> dict:
         """모든 템플릿 정보를 조회합니다."""
         url = f"{self.BASE_URL}/v1/templates"
-        response = await self._request("GET", url, timeout=30.0)
+        response = await self._request("GET", url)  # 기본 타임아웃 사용
         response.raise_for_status()
         return response.json()
 
@@ -288,7 +311,7 @@ class CreatomateService:
 
         # API 호출
         url = f"{self.BASE_URL}/v1/templates/{template_id}"
-        response = await self._request("GET", url, timeout=30.0)
+        response = await self._request("GET", url)  # 기본 타임아웃 사용
         response.raise_for_status()
         data = response.json()
 
@@ -433,30 +456,147 @@ class CreatomateService:
     async def make_creatomate_call(
         self, template_id: str, modifications: dict
     ) -> dict:
-        """Creatomate에 렌더링 요청을 보냅니다.
+        """Creatomate에 렌더링 요청을 보냅니다 (재시도 로직 포함).
+
+        Args:
+            template_id: Creatomate 템플릿 ID
+            modifications: 수정사항 딕셔너리
+
+        Returns:
+            Creatomate API 응답 데이터
+
+        Raises:
+            CreatomateResponseError: API 오류 또는 재시도 실패 시
 
         Note:
             response에 요청 정보가 있으니 폴링 필요
         """
         url = f"{self.BASE_URL}/v2/renders"
-        data = {
+        payload = {
             "template_id": template_id,
             "modifications": modifications,
         }
-        response = await self._request("POST", url, timeout=60.0, json=data)
-        response.raise_for_status()
-        return response.json()
+
+        last_error: Exception | None = None
+
+        for attempt in range(recovery_settings.CREATOMATE_MAX_RETRIES + 1):
+            try:
+                response = await self._request(
+                    "POST",
+                    url,
+                    timeout=recovery_settings.CREATOMATE_RENDER_TIMEOUT,
+                    json=payload,
+                )
+
+                if response.status_code == 200 or response.status_code == 201:
+                    return response.json()
+
+                # 재시도 불가능한 오류 (4xx 클라이언트 오류)
+                if 400 <= response.status_code < 500:
+                    raise CreatomateResponseError(
+                        f"Client error: {response.status_code}",
+                        original_response={"status": response.status_code, "text": response.text},
+                    )
+
+                # 재시도 가능한 오류 (5xx 서버 오류)
+                last_error = CreatomateResponseError(
+                    f"Server error: {response.status_code}",
+                    original_response={"status": response.status_code, "text": response.text},
+                )
+
+            except httpx.TimeoutException as e:
+                logger.warning(
+                    f"[Creatomate] Timeout on attempt {attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES + 1}"
+                )
+                last_error = e
+
+            except httpx.HTTPError as e:
+                logger.warning(f"[Creatomate] HTTP error on attempt {attempt + 1}: {e}")
+                last_error = e
+
+            except CreatomateResponseError:
+                raise  # CreatomateResponseError는 재시도하지 않고 즉시 전파
+
+            # 마지막 시도가 아니면 재시도
+            if attempt < recovery_settings.CREATOMATE_MAX_RETRIES:
+                logger.info(
+                    f"[Creatomate] Retrying... ({attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES})"
+                )
+
+        # 모든 재시도 실패
+        raise CreatomateResponseError(
+            f"All {recovery_settings.CREATOMATE_MAX_RETRIES + 1} attempts failed",
+            original_response={"last_error": str(last_error)},
+        )
 
     async def make_creatomate_custom_call(self, source: dict) -> dict:
-        """템플릿 없이 Creatomate에 커스텀 렌더링 요청을 보냅니다.
+        """템플릿 없이 Creatomate에 커스텀 렌더링 요청을 보냅니다 (재시도 로직 포함).
+
+        Args:
+            source: 렌더링 소스 딕셔너리
+
+        Returns:
+            Creatomate API 응답 데이터
+
+        Raises:
+            CreatomateResponseError: API 오류 또는 재시도 실패 시
 
         Note:
             response에 요청 정보가 있으니 폴링 필요
         """
         url = f"{self.BASE_URL}/v2/renders"
-        response = await self._request("POST", url, timeout=60.0, json=source)
-        response.raise_for_status()
-        return response.json()
+
+        last_error: Exception | None = None
+
+        for attempt in range(recovery_settings.CREATOMATE_MAX_RETRIES + 1):
+            try:
+                response = await self._request(
+                    "POST",
+                    url,
+                    timeout=recovery_settings.CREATOMATE_RENDER_TIMEOUT,
+                    json=source,
+                )
+
+                if response.status_code == 200 or response.status_code == 201:
+                    return response.json()
+
+                # 재시도 불가능한 오류 (4xx 클라이언트 오류)
+                if 400 <= response.status_code < 500:
+                    raise CreatomateResponseError(
+                        f"Client error: {response.status_code}",
+                        original_response={"status": response.status_code, "text": response.text},
+                    )
+
+                # 재시도 가능한 오류 (5xx 서버 오류)
+                last_error = CreatomateResponseError(
+                    f"Server error: {response.status_code}",
+                    original_response={"status": response.status_code, "text": response.text},
+                )
+
+            except httpx.TimeoutException as e:
+                logger.warning(
+                    f"[Creatomate] Timeout on attempt {attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES + 1}"
+                )
+                last_error = e
+
+            except httpx.HTTPError as e:
+                logger.warning(f"[Creatomate] HTTP error on attempt {attempt + 1}: {e}")
+                last_error = e
+
+            except CreatomateResponseError:
+                raise  # CreatomateResponseError는 재시도하지 않고 즉시 전파
+
+            # 마지막 시도가 아니면 재시도
+            if attempt < recovery_settings.CREATOMATE_MAX_RETRIES:
+                logger.info(
+                    f"[Creatomate] Retrying... ({attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES})"
+                )
+
+        # 모든 재시도 실패
+        raise CreatomateResponseError(
+            f"All {recovery_settings.CREATOMATE_MAX_RETRIES + 1} attempts failed",
+            original_response={"last_error": str(last_error)},
+        )
 
     # 하위 호환성을 위한 별칭 (deprecated)
     async def make_creatomate_custom_call_async(self, source: dict) -> dict:
@@ -485,7 +625,7 @@ class CreatomateService:
             - failed: 실패
         """
         url = f"{self.BASE_URL}/v1/renders/{render_id}"
-        response = await self._request("GET", url, timeout=30.0)
+        response = await self._request("GET", url)  # 기본 타임아웃 사용
         response.raise_for_status()
         return response.json()
 
diff --git a/app/utils/suno.py b/app/utils/suno.py
index 663b813..87611ef 100644
--- a/app/utils/suno.py
+++ b/app/utils/suno.py
@@ -59,8 +59,28 @@ from typing import Any, List, Optional
 
 import httpx
 
-from config import apikey_settings
 from app.song.schemas.song_schema import PollingSongResponse, SongClipData
+from app.utils.logger import get_logger
+from config import apikey_settings, recovery_settings
+
+logger = get_logger("suno")
+
+
+class SunoResponseError(Exception):
+    """Suno API 응답 오류 시 발생하는 예외
+
+    Suno API 거부 응답 또는 비정상 응답 시 사용됩니다.
+    재시도 로직에서 이 예외를 catch하여 재시도를 수행합니다.
+
+    Attributes:
+        message: 에러 메시지
+        original_response: 원본 API 응답 (있는 경우)
+    """
+
+    def __init__(self, message: str, original_response: dict | None = None):
+        self.message = message
+        self.original_response = original_response
+        super().__init__(self.message)
 
 
 class SunoService:
@@ -122,34 +142,74 @@ class SunoService:
         if genre:
             payload["style"] = genre
 
-        async with httpx.AsyncClient() as client:
-            response = await client.post(
-                f"{self.BASE_URL}/generate",
-                headers=self.headers,
-                json=payload,
-                timeout=30.0,
-            )
-            response.raise_for_status()
-            data = response.json()
+        last_error: Exception | None = None
 
-        # 응답: {"code": 200, "msg": "success", "data": {"taskId": "..."}}
-        # API 응답 검증
-        if data is None:
-            raise ValueError("Suno API returned empty response")
+        for attempt in range(recovery_settings.SUNO_MAX_RETRIES + 1):
+            try:
+                async with httpx.AsyncClient() as client:
+                    response = await client.post(
+                        f"{self.BASE_URL}/generate",
+                        headers=self.headers,
+                        json=payload,
+                        timeout=recovery_settings.SUNO_DEFAULT_TIMEOUT,
+                    )
 
-        if data.get("code") != 200:
-            error_msg = data.get("msg", "Unknown error")
-            raise ValueError(f"Suno API error: {error_msg}")
+                    if response.status_code == 200:
+                        data = response.json()
 
-        response_data = data.get("data")
-        if response_data is None:
-            raise ValueError(f"Suno API response missing 'data' field: {data}")
+                        # API 응답 검증
+                        if data is None:
+                            raise SunoResponseError("Suno API returned empty response")
 
-        task_id = response_data.get("taskId")
-        if task_id is None:
-            raise ValueError(f"Suno API response missing 'taskId': {response_data}")
+                        if data.get("code") != 200:
+                            error_msg = data.get("msg", "Unknown error")
+                            raise SunoResponseError(f"Suno API error: {error_msg}", original_response=data)
 
-        return task_id
+                        response_data = data.get("data")
+                        if response_data is None:
+                            raise SunoResponseError(f"Suno API response missing 'data' field", original_response=data)
+
+                        task_id = response_data.get("taskId")
+                        if task_id is None:
+                            raise SunoResponseError(f"Suno API response missing 'taskId'", original_response=response_data)
+
+                        return task_id
+
+                    # 재시도 불가능한 오류 (4xx 클라이언트 오류)
+                    if 400 <= response.status_code < 500:
+                        raise SunoResponseError(
+                            f"Client error: {response.status_code}",
+                            original_response={"status": response.status_code, "text": response.text},
+                        )
+
+                    # 재시도 가능한 오류 (5xx 서버 오류)
+                    last_error = SunoResponseError(
+                        f"Server error: {response.status_code}",
+                        original_response={"status": response.status_code, "text": response.text},
+                    )
+
+            except httpx.TimeoutException as e:
+                logger.warning(
+                    f"[Suno] Timeout on attempt {attempt + 1}/{recovery_settings.SUNO_MAX_RETRIES + 1}"
+                )
+                last_error = e
+
+            except httpx.HTTPError as e:
+                logger.warning(f"[Suno] HTTP error on attempt {attempt + 1}: {e}")
+                last_error = e
+
+            except SunoResponseError:
+                raise  # SunoResponseError는 재시도하지 않고 즉시 전파
+
+            # 마지막 시도가 아니면 재시도
+            if attempt < recovery_settings.SUNO_MAX_RETRIES:
+                logger.info(f"[Suno] Retrying... ({attempt + 1}/{recovery_settings.SUNO_MAX_RETRIES})")
+
+        # 모든 재시도 실패
+        raise SunoResponseError(
+            f"All {recovery_settings.SUNO_MAX_RETRIES + 1} attempts failed",
+            original_response={"last_error": str(last_error)},
+        )
 
     async def get_task_status(self, task_id: str) -> dict[str, Any]:
         """
@@ -170,7 +230,7 @@ class SunoService:
                 f"{self.BASE_URL}/generate/record-info",
                 headers=self.headers,
                 params={"taskId": task_id},
-                timeout=30.0,
+                timeout=recovery_settings.SUNO_DEFAULT_TIMEOUT,
             )
             response.raise_for_status()
             data = response.json()
@@ -198,7 +258,7 @@ class SunoService:
                 f"{self.BASE_URL}/generate/get-timestamped-lyrics",
                 headers=self.headers,
                 json=payload,
-                timeout=120.0,
+                timeout=recovery_settings.SUNO_LYRIC_TIMEOUT,
             )
             response.raise_for_status()
             data = response.json()
diff --git a/config.py b/config.py
index 448bcd8..970a248 100644
--- a/config.py
+++ b/config.py
@@ -156,6 +156,63 @@ class PromptSettings(BaseSettings):
 
     model_config = _base_config
 
+
+class RecoverySettings(BaseSettings):
+    """외부 API 복구 및 타임아웃 설정
+
+    ChatGPT, Suno, Creatomate API의 타임아웃 및 재시도 설정을 관리합니다.
+    """
+
+    # ============================================================
+    # ChatGPT API 설정
+    # ============================================================
+    CHATGPT_TIMEOUT: float = Field(
+        default=600.0,
+        description="ChatGPT API 타임아웃 (초). OpenAI Python SDK 기본값: 600초 (10분)",
+    )
+    CHATGPT_MAX_RETRIES: int = Field(
+        default=1,
+        description="ChatGPT API 응답 실패 시 최대 재시도 횟수",
+    )
+
+    # ============================================================
+    # Suno API 설정
+    # ============================================================
+    SUNO_DEFAULT_TIMEOUT: float = Field(
+        default=30.0,
+        description="Suno API 기본 요청 타임아웃 (초) - 음악 생성 요청, 상태 조회 등",
+    )
+    SUNO_LYRIC_TIMEOUT: float = Field(
+        default=120.0,
+        description="Suno API 가사 타임스탬프 요청 타임아웃 (초) - 가사 동기화 처리에 더 긴 시간 필요",
+    )
+    SUNO_MAX_RETRIES: int = Field(
+        default=2,
+        description="Suno API 응답 실패 시 최대 재시도 횟수",
+    )
+
+    # ============================================================
+    # Creatomate API 설정
+    # ============================================================
+    CREATOMATE_DEFAULT_TIMEOUT: float = Field(
+        default=30.0,
+        description="Creatomate API 기본 요청 타임아웃 (초) - 템플릿 조회, 상태 조회 등 일반 API 호출",
+    )
+    CREATOMATE_RENDER_TIMEOUT: float = Field(
+        default=60.0,
+        description="Creatomate API 렌더링 요청 타임아웃 (초) - 영상 렌더링 요청 시 더 긴 시간 필요",
+    )
+    CREATOMATE_CONNECT_TIMEOUT: float = Field(
+        default=10.0,
+        description="Creatomate API 연결 타임아웃 (초) - 서버 연결 수립까지의 대기 시간",
+    )
+    CREATOMATE_MAX_RETRIES: int = Field(
+        default=2,
+        description="Creatomate API 응답 실패 시 최대 재시도 횟수",
+    )
+
+    model_config = _base_config
+
 class KakaoSettings(BaseSettings):
     """카카오 OAuth 설정"""
 
@@ -389,3 +446,4 @@ prompt_settings = PromptSettings()
 log_settings = LogSettings()
 kakao_settings = KakaoSettings()
 jwt_settings = JWTSettings()
+recovery_settings = RecoverySettings()
diff --git a/docs/plan/db_lock.md b/docs/plan/db_lock.md
new file mode 100644
index 0000000..0da9ebc
--- /dev/null
+++ b/docs/plan/db_lock.md
@@ -0,0 +1,885 @@
+# DB Lock 및 Upsert 패턴 가이드 (MySQL 전용)
+
+## 목차
+1. [현재 Insert 사용 현황](#1-현재-insert-사용-현황)
+2. [Upsert 패턴 설계](#2-upsert-패턴-설계)
+3. [DB Lock 전략](#3-db-lock-전략)
+4. [데드락 방지 전략](#4-데드락-방지-전략)
+5. [제안 코드](#5-제안-코드)
+6. [사용 예시](#6-사용-예시)
+
+---
+
+## 1. 현재 Insert 사용 현황
+
+### 1.1 테이블별 Insert 분석 및 우선순위
+
+| 테이블 | 파일 위치 | Unique 제약 | 동시 요청 가능성 | Upsert 우선순위 | 키 조합 |
+|--------|-----------|-------------|------------------|-----------------|---------|
+| **SongTimestamp** | song.py:477 | - | **높음** | **1순위** | suno_audio_id + order_idx |
+| **Image** | home.py:503,552,799,821 | - | **중간** | **2순위** | task_id + img_order |
+| **Song** | song.py:188 | - | 중간 | 3순위 | task_id |
+| **Video** | video.py:252 | - | 중간 | 4순위 | task_id |
+| **User** | auth.py:278 | `kakao_id` | **낮음** | ✅ 완료 | kakao_id |
+| **Project** | lyric.py:297 | - | 낮음 | 선택 | task_id |
+| **Lyric** | lyric.py:317 | - | 낮음 | 선택 | task_id |
+| **RefreshToken** | auth.py:315 | `token_hash` | - | 불필요 | - (항상 새로 생성) |
+
+#### 동시 요청 가능성 분석
+
+| 가능성 | 테이블 | 발생 시나리오 |
+|--------|--------|---------------|
+| **높음** | SongTimestamp | 클라이언트가 상태 조회 API를 여러 번 호출 (폴링) → 동일 데이터 중복 삽입 |
+| **중간** | Image | 네트워크 오류로 업로드 재시도, 클라이언트 중복 클릭 |
+| **중간** | Song/Video | 백그라운드 태스크 재실행, 상태 확인 중복 호출 |
+| **낮음** | User | OAuth 인가 코드 일회성으로 동시 요청 거의 불가능 |
+
+> **참고**: User 테이블의 경우 카카오 OAuth 인가 코드(authorization code)가 **일회성**이므로,
+> 동일한 코드로 동시 요청은 불가능합니다. 다만 여러 탭에서 각각 로그인을 시작하는 극히 드문 경우에만 발생 가능합니다.
+
+### 1.2 현재 코드 패턴의 문제점
+
+**실제 문제가 발생하는 케이스: SongTimestamp**
+
+```python
+# song.py:467-479 - 현재 패턴
+# 클라이언트가 /song/status/{song_id}를 여러 번 호출하면 중복 삽입 발생!
+
+for order_idx, timestamped_lyric in enumerate(timestamped_lyrics):
+    song_timestamp = SongTimestamp(
+        suno_audio_id=suno_audio_id,
+        order_idx=order_idx,
+        lyric_line=timestamped_lyric["text"],
+        start_time=timestamped_lyric["start_sec"],
+        end_time=timestamped_lyric["end_sec"],
+    )
+    session.add(song_timestamp)  # 동일 suno_audio_id로 중복 삽입!
+
+await session.commit()
+```
+
+**문제 시나리오:**
+1. 클라이언트가 노래 생성 상태 확인을 위해 폴링
+2. SUCCESS 응답을 받은 후 타임스탬프 저장 로직 실행
+3. 네트워크 지연으로 클라이언트가 재요청
+4. **동일한 데이터가 중복 삽입됨**
+
+---
+
+## 2. Upsert 패턴 설계 (MySQL 전용)
+
+### 2.1 MySQL ON DUPLICATE KEY UPDATE (권장)
+
+MySQL의 `INSERT ... ON DUPLICATE KEY UPDATE` 절을 사용한 원자적 Upsert:
+
+```python
+from sqlalchemy.dialects.mysql import insert as mysql_insert
+
+stmt = mysql_insert(User).values(
+    kakao_id=kakao_id,
+    nickname=nickname,
+    email=email,
+)
+stmt = stmt.on_duplicate_key_update(
+    nickname=stmt.inserted.nickname,  # MySQL은 stmt.inserted 사용
+    email=stmt.inserted.email,
+    updated_at=func.now(),
+)
+await session.execute(stmt)
+
+# MySQL은 RETURNING 미지원 - 별도 조회 필요
+result = await session.execute(select(User).where(User.kakao_id == kakao_id))
+user = result.scalar_one()
+```
+
+**장점:**
+- 원자적 연산 (단일 쿼리)
+- 데드락 위험 최소화
+- 동시 요청에서도 안전
+
+**전제 조건:**
+- Unique 인덱스 필수 (없으면 항상 INSERT만 됨)
+
+### 2.3 비관적 잠금 (Pessimistic Locking)
+
+`SELECT ... FOR UPDATE`를 사용한 행 수준 잠금:
+
+```python
+result = await session.execute(
+    select(User)
+    .where(User.kakao_id == kakao_id)
+    .with_for_update()  # FOR UPDATE 잠금
+)
+user = result.scalar_one_or_none()
+```
+
+---
+
+## 3. DB Lock 전략
+
+### 3.1 잠금 유형
+
+| 잠금 유형 | 사용 시점 | SQLAlchemy 구현 |
+|-----------|-----------|-----------------|
+| **공유 잠금 (Shared)** | 읽기 작업 | `.with_for_update(read=True)` |
+| **배타 잠금 (Exclusive)** | 쓰기 작업 | `.with_for_update()` |
+| **NOWAIT** | 즉시 실패 | `.with_for_update(nowait=True)` |
+| **SKIP LOCKED** | 잠긴 행 건너뛰기 | `.with_for_update(skip_locked=True)` |
+
+### 3.2 잠금 범위 선택
+
+```python
+# 1. 행 수준 잠금 (Row-level) - 권장
+select(User).where(User.id == user_id).with_for_update()
+
+# 2. 키 범위 잠금 (Key-range) - 범위 조회 시
+select(Image).where(Image.task_id == task_id).with_for_update()
+
+# 3. 테이블 잠금 - 피해야 함 (성능 저하)
+```
+
+### 3.3 트랜잭션 격리 수준
+
+```python
+from sqlalchemy import text
+
+# 세션별 격리 수준 설정
+await session.execute(text("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ"))
+```
+
+---
+
+## 4. 데드락 방지 전략
+
+### 4.1 핵심 원칙
+
+1. **일관된 잠금 순서**: 항상 같은 순서로 리소스 접근
+2. **짧은 트랜잭션**: 잠금 유지 시간 최소화
+3. **타임아웃 설정**: 무한 대기 방지
+4. **재시도 로직**: 데드락 발생 시 자동 재시도
+
+### 4.2 잠금 순서 규칙
+
+```python
+# 올바른 순서: 항상 PK 또는 정렬된 순서로 잠금
+async def lock_resources_safely(session, resource_ids: list[int]):
+    """리소스를 ID 순서로 정렬하여 잠금"""
+    sorted_ids = sorted(resource_ids)  # 정렬!
+
+    for resource_id in sorted_ids:
+        await session.execute(
+            select(Resource)
+            .where(Resource.id == resource_id)
+            .with_for_update()
+        )
+```
+
+### 4.3 타임아웃 설정 (MySQL)
+
+```python
+# MySQL 잠금 타임아웃 설정
+await session.execute(text("SET innodb_lock_wait_timeout = 5"))
+```
+
+### 4.4 재시도 로직
+
+```python
+import asyncio
+from sqlalchemy.exc import OperationalError
+
+async def execute_with_retry(
+    session,
+    operation,
+    max_retries: int = 3,
+    base_delay: float = 0.1,
+):
+    """지수 백오프를 사용한 재시도 로직"""
+    for attempt in range(max_retries):
+        try:
+            return await operation(session)
+        except OperationalError as e:
+            if "deadlock" in str(e).lower() and attempt < max_retries - 1:
+                delay = base_delay * (2 ** attempt)
+                await asyncio.sleep(delay)
+                await session.rollback()
+                continue
+            raise
+```
+
+---
+
+## 5. 제안 코드 (MySQL 전용)
+
+### 5.1 공통 Upsert 유틸리티
+
+`app/utils/db_utils.py` 파일 (✅ 이미 생성됨):
+
+```python
+"""
+DB 유틸리티 - Upsert 및 Lock 관리 (MySQL 전용)
+
+MySQL의 INSERT ... ON DUPLICATE KEY UPDATE를 사용한 안전한 Upsert 패턴과
+데드락 방지를 위한 잠금 관리 유틸리티를 제공합니다.
+"""
+
+import asyncio
+import logging
+from typing import Any, Callable, Dict, List, Optional, Type, TypeVar
+
+from sqlalchemy import func, select
+from sqlalchemy.dialects.mysql import insert as mysql_insert
+from sqlalchemy.exc import IntegrityError, OperationalError
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import DeclarativeBase
+
+logger = logging.getLogger(__name__)
+
+T = TypeVar("T", bound=DeclarativeBase)
+
+
+class UpsertResult:
+    """Upsert 결과를 담는 클래스"""
+
+    def __init__(self, entity: Any, created: bool):
+        self.entity = entity
+        self.created = created  # True: INSERT, False: UPDATE
+
+
+async def upsert_by_unique_key(
+    session: AsyncSession,
+    model: Type[T],
+    unique_columns: List[str],
+    values: Dict[str, Any],
+    update_columns: Optional[List[str]] = None,
+    lock_timeout_sec: int = 5,
+) -> UpsertResult:
+    """
+    Unique 키 기반 원자적 Upsert (MySQL ON DUPLICATE KEY UPDATE 사용)
+
+    Args:
+        session: AsyncSession 인스턴스
+        model: SQLAlchemy 모델 클래스
+        unique_columns: Unique 제약 컬럼 목록
+        values: INSERT/UPDATE 값 딕셔너리
+        update_columns: UPDATE 시 변경할 컬럼 목록 (None이면 unique 제외 전체)
+        lock_timeout_sec: 잠금 타임아웃 (초)
+
+    Returns:
+        UpsertResult: 엔티티와 생성 여부
+
+    Example:
+        result = await upsert_by_unique_key(
+            session=session,
+            model=User,
+            unique_columns=['kakao_id'],
+            values={'kakao_id': 12345, 'nickname': '홍길동', 'email': 'test@test.com'},
+            update_columns=['nickname', 'email'],
+        )
+        if result.created:
+            print("새 사용자 생성됨")
+        else:
+            print("기존 사용자 업데이트됨")
+    """
+    from sqlalchemy import text
+
+    # MySQL 잠금 타임아웃 설정
+    await session.execute(text(f"SET innodb_lock_wait_timeout = {lock_timeout_sec}"))
+
+    # UPDATE 컬럼 결정
+    if update_columns is None:
+        update_columns = [k for k in values.keys() if k not in unique_columns]
+
+    # MySQL INSERT ... ON DUPLICATE KEY UPDATE
+    stmt = mysql_insert(model).values(**values)
+
+    update_dict = {col: stmt.inserted[col] for col in update_columns}
+    if hasattr(model, 'updated_at'):
+        update_dict['updated_at'] = func.now()
+
+    stmt = stmt.on_duplicate_key_update(**update_dict)
+
+    await session.execute(stmt)
+
+    # MySQL은 RETURNING 미지원 - 별도 조회 필요
+    filter_conditions = {col: values[col] for col in unique_columns}
+    result = await session.execute(select(model).filter_by(**filter_conditions))
+    entity = result.scalar_one()
+
+    # created 여부 확인 (created_at == updated_at 비교)
+    created = True
+    if hasattr(entity, 'updated_at') and hasattr(entity, 'created_at'):
+        if entity.created_at and entity.updated_at:
+            created = abs((entity.updated_at - entity.created_at).total_seconds()) < 1
+
+    return UpsertResult(entity=entity, created=created)
+
+
+async def get_or_create_with_lock(
+    session: AsyncSession,
+    model: Type[T],
+    filter_by: Dict[str, Any],
+    defaults: Optional[Dict[str, Any]] = None,
+    lock: bool = True,
+    nowait: bool = False,
+) -> UpsertResult:
+    """
+    SELECT FOR UPDATE를 사용한 안전한 Get or Create
+
+    동시 요청에서도 안전하게 작동하며,
+    행이 존재하면 잠금 후 반환, 없으면 생성합니다.
+
+    Args:
+        session: AsyncSession 인스턴스
+        model: SQLAlchemy 모델 클래스
+        filter_by: 조회 조건 딕셔너리
+        defaults: 생성 시 추가할 기본값
+        lock: FOR UPDATE 잠금 사용 여부
+        nowait: 잠금 대기 안함 (즉시 예외 발생)
+
+    Returns:
+        UpsertResult: 엔티티와 생성 여부
+
+    Example:
+        result = await get_or_create_with_lock(
+            session=session,
+            model=User,
+            filter_by={'kakao_id': 12345},
+            defaults={'nickname': '홍길동'},
+        )
+    """
+    # 조회 쿼리 구성
+    query = select(model).filter_by(**filter_by)
+
+    if lock:
+        query = query.with_for_update(nowait=nowait)
+
+    result = await session.execute(query)
+    entity = result.scalar_one_or_none()
+
+    if entity is not None:
+        # 기존 엔티티 반환
+        return UpsertResult(entity=entity, created=False)
+
+    # 새 엔티티 생성
+    create_values = {**filter_by, **(defaults or {})}
+    entity = model(**create_values)
+    session.add(entity)
+
+    try:
+        await session.flush()
+    except IntegrityError:
+        # 동시 INSERT로 인한 충돌 - 다시 조회
+        await session.rollback()
+        result = await session.execute(select(model).filter_by(**filter_by))
+        entity = result.scalar_one()
+        return UpsertResult(entity=entity, created=False)
+
+    return UpsertResult(entity=entity, created=True)
+
+
+async def bulk_upsert(
+    session: AsyncSession,
+    model: Type[T],
+    unique_columns: List[str],
+    records: List[Dict[str, Any]],
+    update_columns: Optional[List[str]] = None,
+) -> int:
+    """
+    대량 Upsert (MySQL ON DUPLICATE KEY UPDATE 사용)
+
+    여러 레코드를 한 번에 Upsert합니다.
+    데드락 방지를 위해 unique 키 기준으로 정렬 후 처리합니다.
+    Unique 인덱스가 반드시 존재해야 합니다.
+
+    Args:
+        session: AsyncSession 인스턴스
+        model: SQLAlchemy 모델 클래스
+        unique_columns: Unique 제약 컬럼 목록
+        records: Upsert할 레코드 딕셔너리 목록
+        update_columns: UPDATE 시 변경할 컬럼 목록
+
+    Returns:
+        int: 처리된 레코드 수
+
+    Example:
+        count = await bulk_upsert(
+            session=session,
+            model=SongTimestamp,
+            unique_columns=['suno_audio_id', 'order_idx'],
+            records=[
+                {'suno_audio_id': 'abc', 'order_idx': 0, 'lyric_line': '가사1'},
+                {'suno_audio_id': 'abc', 'order_idx': 1, 'lyric_line': '가사2'},
+            ],
+        )
+    """
+    if not records:
+        return 0
+
+    # 데드락 방지: unique 키 기준 정렬
+    sorted_records = sorted(
+        records,
+        key=lambda r: tuple(r.get(col, '') for col in unique_columns)
+    )
+
+    # UPDATE 컬럼 결정
+    if update_columns is None:
+        all_columns = set(sorted_records[0].keys())
+        update_columns = list(all_columns - set(unique_columns))
+
+    # MySQL INSERT ... ON DUPLICATE KEY UPDATE
+    stmt = mysql_insert(model).values(sorted_records)
+
+    update_dict = {col: stmt.inserted[col] for col in update_columns}
+    if hasattr(model, 'updated_at'):
+        update_dict['updated_at'] = func.now()
+
+    stmt = stmt.on_duplicate_key_update(**update_dict)
+
+    await session.execute(stmt)
+    return len(sorted_records)
+
+
+async def execute_with_retry(
+    func: Callable,
+    max_retries: int = 3,
+    base_delay: float = 0.1,
+    retry_on: tuple = (OperationalError,),
+) -> Any:
+    """
+    지수 백오프를 사용한 재시도 래퍼
+
+    데드락이나 일시적 오류 발생 시 자동으로 재시도합니다.
+
+    Args:
+        func: 실행할 비동기 함수 (인자 없음)
+        max_retries: 최대 재시도 횟수
+        base_delay: 기본 대기 시간 (초)
+        retry_on: 재시도할 예외 타입들
+
+    Returns:
+        함수 실행 결과
+
+    Example:
+        async def do_work():
+            async with AsyncSessionLocal() as session:
+                await upsert_by_unique_key(...)
+                await session.commit()
+
+        result = await execute_with_retry(do_work)
+    """
+    last_exception = None
+
+    for attempt in range(max_retries):
+        try:
+            return await func()
+        except retry_on as e:
+            last_exception = e
+            error_msg = str(e).lower()
+
+            # 데드락 또는 잠금 타임아웃인 경우만 재시도
+            if "deadlock" in error_msg or "lock" in error_msg:
+                if attempt < max_retries - 1:
+                    delay = base_delay * (2 ** attempt)
+                    logger.warning(
+                        f"DB 작업 실패 (시도 {attempt + 1}/{max_retries}), "
+                        f"{delay:.2f}초 후 재시도: {e}"
+                    )
+                    await asyncio.sleep(delay)
+                    continue
+            raise
+
+    raise last_exception
+
+
+class LockManager:
+    """
+    분산 잠금 관리자 (MySQL 전용)
+
+    여러 리소스에 대한 잠금을 일관된 순서로 획득하여
+    데드락을 방지합니다.
+    """
+
+    def __init__(self, session: AsyncSession, timeout_sec: int = 5):
+        self.session = session
+        self.timeout_sec = timeout_sec
+        self._locked_resources: List[tuple] = []
+
+    async def __aenter__(self):
+        from sqlalchemy import text
+        # MySQL 잠금 타임아웃 설정
+        await self.session.execute(
+            text(f"SET innodb_lock_wait_timeout = {self.timeout_sec}")
+        )
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        # 트랜잭션 종료 시 자동으로 잠금 해제됨
+        self._locked_resources.clear()
+
+    async def lock_rows(
+        self,
+        model: Type[T],
+        ids: List[Any],
+        id_column: str = "id",
+        nowait: bool = False,
+    ) -> List[T]:
+        """
+        여러 행을 ID 순서로 잠금
+
+        Args:
+            model: 모델 클래스
+            ids: 잠금할 ID 목록
+            id_column: ID 컬럼명
+            nowait: 잠금 대기 안함
+
+        Returns:
+            잠긴 엔티티 목록
+        """
+        if not ids:
+            return []
+
+        # 데드락 방지: ID 정렬
+        sorted_ids = sorted(ids)
+
+        column = getattr(model, id_column)
+        query = (
+            select(model)
+            .where(column.in_(sorted_ids))
+            .order_by(column)  # 정렬 순서 유지
+            .with_for_update(nowait=nowait)
+        )
+
+        result = await self.session.execute(query)
+        entities = result.scalars().all()
+
+        self._locked_resources.append((model.__tablename__, sorted_ids))
+        return list(entities)
+
+    async def lock_row(
+        self,
+        model: Type[T],
+        id_value: Any,
+        id_column: str = "id",
+        nowait: bool = False,
+    ) -> Optional[T]:
+        """
+        단일 행 잠금
+
+        Args:
+            model: 모델 클래스
+            id_value: 잠금할 ID
+            id_column: ID 컬럼명
+            nowait: 잠금 대기 안함
+
+        Returns:
+            잠긴 엔티티 또는 None
+        """
+        entities = await self.lock_rows(model, [id_value], id_column, nowait)
+        return entities[0] if entities else None
+```
+
+### 5.2 모델에 Unique Constraint 추가 (권장)
+
+테이블별 Unique 제약 추가가 필요한 경우:
+
+```python
+# app/home/models.py - Image 테이블
+class Image(Base):
+    __tablename__ = "image"
+    __table_args__ = (
+        # task_id + img_order 조합 유니크
+        Index("idx_image_task_order", "task_id", "img_order", unique=True),
+        ...
+    )
+
+# app/song/models.py - SongTimestamp 테이블
+class SongTimestamp(Base):
+    __tablename__ = "song_timestamp"
+    __table_args__ = (
+        # suno_audio_id + order_idx 조합 유니크
+        Index("idx_song_ts_audio_order", "suno_audio_id", "order_idx", unique=True),
+        ...
+    )
+```
+
+---
+
+## 6. 사용 예시
+
+### 6.1 SongTimestamp Bulk Upsert (우선순위 1 - 실제 문제 발생)
+
+**가장 먼저 적용해야 하는 케이스입니다.** 클라이언트의 상태 폴링으로 인해 동일 데이터가 중복 삽입될 수 있습니다.
+
+```python
+# app/song/api/routers/v1/song.py
+
+from app.utils.db_utils import bulk_upsert
+
+# 기존 코드 (문제 있음 - 폴링 시 중복 삽입)
+# for order_idx, timestamped_lyric in enumerate(timestamped_lyrics):
+#     song_timestamp = SongTimestamp(...)
+#     session.add(song_timestamp)
+
+# 개선된 코드 (Upsert로 중복 방지)
+records = [
+    {
+        'suno_audio_id': suno_audio_id,
+        'order_idx': idx,
+        'lyric_line': ts['text'],
+        'start_time': ts['start_sec'],
+        'end_time': ts['end_sec'],
+    }
+    for idx, ts in enumerate(timestamped_lyrics)
+]
+
+await bulk_upsert(
+    session=session,
+    model=SongTimestamp,
+    unique_columns=['suno_audio_id', 'order_idx'],
+    records=records,
+    update_columns=['lyric_line', 'start_time', 'end_time'],
+)
+```
+
+### 6.2 Image 중복 방지 Upsert (우선순위 2)
+
+```python
+# app/home/api/routers/v1/home.py
+
+from app.utils.db_utils import get_or_create_with_lock
+
+async def save_image(session: AsyncSession, task_id: str, img_url: str, img_order: int):
+    """이미지 저장 (중복 방지)"""
+    result = await get_or_create_with_lock(
+        session=session,
+        model=Image,
+        filter_by={'task_id': task_id, 'img_order': img_order},
+        defaults={
+            'img_name': f"image_{img_order}",
+            'img_url': img_url,
+        },
+    )
+
+    if not result.created:
+        # 기존 이미지 URL 업데이트
+        result.entity.img_url = img_url
+
+    return result.entity
+```
+
+### 6.3 User IntegrityError 처리 (✅ 적용 완료)
+
+> **참고**: User 테이블은 OAuth 인가 코드의 일회성 특성상 동시 요청 가능성이 **매우 낮습니다**.
+> `kakao_id` UNIQUE 제약이 있어 중복 시 IntegrityError가 발생하므로,
+> IntegrityError 처리를 추가하여 500 에러 대신 기존 사용자를 재조회합니다.
+
+```python
+# app/user/services/auth.py - 현재 적용된 코드
+
+from sqlalchemy.exc import IntegrityError
+
+# 신규 사용자 생성 부분
+session.add(new_user)
+
+try:
+    await session.flush()
+    await session.refresh(new_user)
+    return new_user, True
+except IntegrityError:
+    # 동시 요청으로 인한 중복 삽입 시도 - 기존 사용자 조회
+    logger.warning(
+        f"[AUTH] IntegrityError 발생 (동시 요청 추정) - kakao_id: {kakao_id}"
+    )
+    await session.rollback()
+    result = await session.execute(
+        select(User).where(User.kakao_id == kakao_id)
+    )
+    existing_user = result.scalar_one_or_none()
+
+    if existing_user is not None:
+        # 프로필 정보 업데이트
+        if profile:
+            existing_user.nickname = profile.nickname
+            existing_user.profile_image_url = profile.profile_image_url
+            existing_user.thumbnail_image_url = profile.thumbnail_image_url
+        if kakao_account and kakao_account.email:
+            existing_user.email = kakao_account.email
+        await session.flush()
+        return existing_user, False
+
+    # 재조회에도 실패한 경우 (매우 드문 경우)
+    raise
+```
+
+**이점:**
+- Upsert 패턴 없이도 안전하게 처리
+- 기존 코드 구조 유지
+- 드문 경우의 500 에러 방지
+
+### 6.4 재시도 로직 사용
+
+```python
+from app.utils.db_utils import execute_with_retry, bulk_upsert
+
+async def safe_timestamp_upsert(suno_audio_id: str, timestamps: list):
+    """안전한 타임스탬프 Upsert (데드락 시 재시도)"""
+
+    async def do_upsert():
+        async with AsyncSessionLocal() as session:
+            records = [
+                {
+                    'suno_audio_id': suno_audio_id,
+                    'order_idx': idx,
+                    'lyric_line': ts['text'],
+                    'start_time': ts['start_sec'],
+                    'end_time': ts['end_sec'],
+                }
+                for idx, ts in enumerate(timestamps)
+            ]
+            count = await bulk_upsert(
+                session=session,
+                model=SongTimestamp,
+                unique_columns=['suno_audio_id', 'order_idx'],
+                records=records,
+            )
+            await session.commit()
+            return count
+
+    return await execute_with_retry(do_upsert, max_retries=3)
+```
+
+### 6.5 LockManager 사용
+
+```python
+from app.utils.db_utils import LockManager
+
+async def update_multiple_resources(session: AsyncSession, project_ids: list[int]):
+    """여러 프로젝트를 안전하게 업데이트"""
+
+    async with LockManager(session, timeout_sec=10) as lock:
+        # 프로젝트들을 ID 순서로 잠금 (데드락 방지)
+        projects = await lock.lock_rows(Project, project_ids)
+
+        for project in projects:
+            project.status = "updated"
+
+        await session.commit()
+```
+
+---
+
+## 7. 마이그레이션 가이드
+
+### 7.1 단계별 적용 순서 (우선순위 기반)
+
+동시 요청 가능성이 높은 테이블부터 적용합니다:
+
+1. **1단계**: `app/utils/db_utils.py` 파일 생성 ✅ (이미 완료)
+2. **2단계**: **SongTimestamp** - Unique 인덱스 추가 + `bulk_upsert` 적용 (우선순위 1)
+   - 동시 요청 가능성 **높음** (폴링으로 인한 중복 삽입)
+3. **3단계**: **Image** - Unique 인덱스 추가 + `get_or_create_with_lock` 적용 (우선순위 2)
+   - 동시 요청 가능성 **중간** (업로드 재시도)
+4. **4단계**: **Song/Video** - 필요시 `get_or_create_with_lock` 적용 (우선순위 3-4)
+   - 동시 요청 가능성 중간 (백그라운드 태스크)
+5. **5단계**: **User** - ✅ IntegrityError 처리 추가 완료
+   - 동시 요청 가능성 **낮음** (OAuth 인가 코드 일회성)
+   - `kakao_id` UNIQUE 제약 + IntegrityError 발생 시 기존 사용자 재조회 처리
+
+> **권장**: 1~3단계까지 우선 적용하고, 나머지는 필요에 따라 적용
+
+### 7.2 Alembic 마이그레이션 예시
+
+```python
+"""Add unique constraints for upsert support
+
+Revision ID: xxxx
+"""
+from alembic import op
+import sqlalchemy as sa
+
+def upgrade():
+    # SongTimestamp: suno_audio_id + order_idx unique
+    op.create_index(
+        'idx_song_ts_audio_order',
+        'song_timestamp',
+        ['suno_audio_id', 'order_idx'],
+        unique=True
+    )
+
+    # Image: task_id + img_order unique (선택적)
+    op.create_index(
+        'idx_image_task_order',
+        'image',
+        ['task_id', 'img_order'],
+        unique=True
+    )
+
+def downgrade():
+    op.drop_index('idx_song_ts_audio_order', table_name='song_timestamp')
+    op.drop_index('idx_image_task_order', table_name='image')
+```
+
+---
+
+## 8. 모니터링 및 디버깅
+
+### 8.1 잠금 모니터링 쿼리 (MySQL)
+
+```sql
+-- 현재 잠금 상태 확인 (InnoDB)
+SELECT
+    r.trx_id AS waiting_trx_id,
+    r.trx_mysql_thread_id AS waiting_thread,
+    r.trx_query AS waiting_query,
+    b.trx_id AS blocking_trx_id,
+    b.trx_mysql_thread_id AS blocking_thread,
+    b.trx_query AS blocking_query
+FROM information_schema.innodb_lock_waits w
+INNER JOIN information_schema.innodb_trx b ON b.trx_id = w.blocking_trx_id
+INNER JOIN information_schema.innodb_trx r ON r.trx_id = w.requesting_trx_id;
+
+-- 현재 실행 중인 트랜잭션 확인
+SELECT * FROM information_schema.innodb_trx;
+
+-- 데드락 로그 확인
+SHOW ENGINE INNODB STATUS;
+
+-- 잠금 대기 중인 쿼리 확인
+SELECT
+    pl.id,
+    pl.user,
+    pl.state,
+    pl.info AS query
+FROM information_schema.processlist pl
+WHERE pl.state LIKE '%lock%';
+```
+
+### 8.2 로깅 설정
+
+```python
+# config.py 또는 logging 설정
+import logging
+
+# SQLAlchemy 엔진 로깅 (디버그 시)
+logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)
+
+# Upsert 유틸리티 로깅
+logging.getLogger('app.utils.db_utils').setLevel(logging.DEBUG)
+```
+
+---
+
+## 9. 요약
+
+| 상황 | 권장 패턴 | 함수 |
+|------|-----------|------|
+| 단일 레코드 Upsert (Unique 키 존재) | ON CONFLICT | `upsert_by_unique_key()` |
+| 단일 레코드 Get or Create | SELECT FOR UPDATE | `get_or_create_with_lock()` |
+| 대량 레코드 Upsert | Bulk ON CONFLICT | `bulk_upsert()` |
+| 데드락 방지 재시도 | 지수 백오프 | `execute_with_retry()` |
+| 다중 행 잠금 | 정렬된 순서 잠금 | `LockManager` |
+
+---
+
+**작성일**: 2026-01-26
+**작성자**: Claude Code (AI Assistant)
diff --git a/docs/plan/fix_plan.md b/docs/plan/fix_plan.md
new file mode 100644
index 0000000..a5f1adc
--- /dev/null
+++ b/docs/plan/fix_plan.md
@@ -0,0 +1,408 @@
+# API 타임아웃 및 재시도 로직 개선 계획
+
+## 개요
+
+외부 API 호출 시 타임아웃 미설정 및 재시도 로직 부재로 인한 안정성 문제를 해결합니다.
+
+---
+
+## 현재 상태
+
+| 모듈 | 외부 API | 타임아웃 | 재시도 |
+|------|----------|----------|--------|
+| Lyric | ChatGPT (OpenAI) | ❌ 미설정 (SDK 기본 ~600초) | ❌ 없음 |
+| Song | Suno API | ✅ 30-120초 | ❌ 없음 |
+| Video | Creatomate API | ✅ 30-60초 | ❌ 없음 |
+
+---
+
+## 수정 계획
+
+### 1. ChatGPT API 타임아웃 설정
+
+**파일:** `app/utils/chatgpt_prompt.py`
+
+**현재 코드:**
+```python
+class ChatgptService:
+    def __init__(self):
+        self.client = AsyncOpenAI(api_key=apikey_settings.CHATGPT_API_KEY)
+```
+
+**수정 코드:**
+```python
+class ChatgptService:
+    # 타임아웃 설정 (초)
+    DEFAULT_TIMEOUT = 60.0  # 전체 타임아웃
+    CONNECT_TIMEOUT = 10.0  # 연결 타임아웃
+
+    def __init__(self):
+        self.client = AsyncOpenAI(
+            api_key=apikey_settings.CHATGPT_API_KEY,
+            timeout=httpx.Timeout(
+                self.DEFAULT_TIMEOUT,
+                connect=self.CONNECT_TIMEOUT,
+            ),
+        )
+```
+
+**필요한 import 추가:**
+```python
+import httpx
+```
+
+---
+
+### 2. 재시도 유틸리티 함수 생성
+
+**파일:** `app/utils/retry.py` (새 파일)
+
+```python
+"""
+API 호출 재시도 유틸리티
+
+지수 백오프(Exponential Backoff)를 사용한 재시도 로직을 제공합니다.
+"""
+
+import asyncio
+import logging
+from functools import wraps
+from typing import Callable, Tuple, Type
+
+logger = logging.getLogger(__name__)
+
+
+class RetryExhaustedError(Exception):
+    """모든 재시도 실패 시 발생하는 예외"""
+    def __init__(self, message: str, last_exception: Exception):
+        super().__init__(message)
+        self.last_exception = last_exception
+
+
+async def retry_async(
+    func: Callable,
+    max_retries: int = 3,
+    base_delay: float = 1.0,
+    max_delay: float = 30.0,
+    exponential_base: float = 2.0,
+    retry_on: Tuple[Type[Exception], ...] = (Exception,),
+    on_retry: Callable[[int, Exception], None] | None = None,
+):
+    """
+    비동기 함수 재시도 실행
+
+    Args:
+        func: 실행할 비동기 함수 (인자 없음)
+        max_retries: 최대 재시도 횟수 (기본: 3)
+        base_delay: 첫 번째 재시도 대기 시간 (초)
+        max_delay: 최대 대기 시간 (초)
+        exponential_base: 지수 백오프 배수 (기본: 2.0)
+        retry_on: 재시도할 예외 타입들
+        on_retry: 재시도 시 호출될 콜백 (attempt, exception)
+
+    Returns:
+        함수 실행 결과
+
+    Raises:
+        RetryExhaustedError: 모든 재시도 실패 시
+
+    Example:
+        result = await retry_async(
+            lambda: api_call(),
+            max_retries=3,
+            retry_on=(httpx.TimeoutException, httpx.HTTPStatusError),
+        )
+    """
+    last_exception = None
+
+    for attempt in range(max_retries + 1):
+        try:
+            return await func()
+        except retry_on as e:
+            last_exception = e
+
+            if attempt == max_retries:
+                break
+
+            # 지수 백오프 계산
+            delay = min(base_delay * (exponential_base ** attempt), max_delay)
+
+            logger.warning(
+                f"[retry_async] 시도 {attempt + 1}/{max_retries + 1} 실패, "
+                f"{delay:.1f}초 후 재시도: {type(e).__name__}: {e}"
+            )
+
+            if on_retry:
+                on_retry(attempt + 1, e)
+
+            await asyncio.sleep(delay)
+
+    raise RetryExhaustedError(
+        f"최대 재시도 횟수({max_retries + 1}회) 초과",
+        last_exception,
+    )
+
+
+def with_retry(
+    max_retries: int = 3,
+    base_delay: float = 1.0,
+    max_delay: float = 30.0,
+    retry_on: Tuple[Type[Exception], ...] = (Exception,),
+):
+    """
+    재시도 데코레이터
+
+    Args:
+        max_retries: 최대 재시도 횟수
+        base_delay: 첫 번째 재시도 대기 시간 (초)
+        max_delay: 최대 대기 시간 (초)
+        retry_on: 재시도할 예외 타입들
+
+    Example:
+        @with_retry(max_retries=3, retry_on=(httpx.TimeoutException,))
+        async def call_api():
+            ...
+    """
+    def decorator(func: Callable):
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            return await retry_async(
+                lambda: func(*args, **kwargs),
+                max_retries=max_retries,
+                base_delay=base_delay,
+                max_delay=max_delay,
+                retry_on=retry_on,
+            )
+        return wrapper
+    return decorator
+```
+
+---
+
+### 3. Suno API 재시도 로직 적용
+
+**파일:** `app/utils/suno.py`
+
+**수정 대상 메서드:**
+- `generate()` - 노래 생성 요청
+- `get_task_status()` - 상태 조회
+- `get_lyric_timestamp()` - 타임스탬프 조회
+
+**수정 예시 (generate 메서드):**
+
+```python
+# 상단 import 추가
+import httpx
+from app.utils.retry import retry_async
+
+# 재시도 대상 예외 정의
+RETRY_EXCEPTIONS = (
+    httpx.TimeoutException,
+    httpx.ConnectError,
+    httpx.ReadError,
+)
+
+async def generate(
+    self,
+    prompt: str,
+    genre: str | None = None,
+    callback_url: str | None = None,
+) -> str:
+    # ... 기존 payload 구성 코드 ...
+
+    async def _call_api():
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                f"{self.BASE_URL}/generate",
+                headers=self.headers,
+                json=payload,
+                timeout=30.0,
+            )
+            response.raise_for_status()
+            return response.json()
+
+    # 재시도 로직 적용
+    data = await retry_async(
+        _call_api,
+        max_retries=3,
+        base_delay=1.0,
+        retry_on=RETRY_EXCEPTIONS,
+    )
+
+    # ... 기존 응답 처리 코드 ...
+```
+
+---
+
+### 4. Creatomate API 재시도 로직 적용
+
+**파일:** `app/utils/creatomate.py`
+
+**수정 대상:**
+- `_request()` 메서드 (모든 API 호출의 기반)
+
+**수정 코드:**
+
+```python
+# 상단 import 추가
+from app.utils.retry import retry_async
+
+# 재시도 대상 예외 정의
+RETRY_EXCEPTIONS = (
+    httpx.TimeoutException,
+    httpx.ConnectError,
+    httpx.ReadError,
+)
+
+async def _request(
+    self,
+    method: str,
+    url: str,
+    timeout: float = 30.0,
+    max_retries: int = 3,
+    **kwargs,
+) -> httpx.Response:
+    """HTTP 요청을 수행합니다 (재시도 로직 포함)."""
+    logger.info(f"[Creatomate] {method} {url}")
+
+    async def _call():
+        client = await get_shared_client()
+        if method.upper() == "GET":
+            response = await client.get(
+                url, headers=self.headers, timeout=timeout, **kwargs
+            )
+        elif method.upper() == "POST":
+            response = await client.post(
+                url, headers=self.headers, timeout=timeout, **kwargs
+            )
+        else:
+            raise ValueError(f"Unsupported HTTP method: {method}")
+        response.raise_for_status()
+        return response
+
+    response = await retry_async(
+        _call,
+        max_retries=max_retries,
+        base_delay=1.0,
+        retry_on=RETRY_EXCEPTIONS,
+    )
+
+    logger.info(f"[Creatomate] Response - Status: {response.status_code}")
+    return response
+```
+
+---
+
+### 5. ChatGPT API 재시도 로직 적용
+
+**파일:** `app/utils/chatgpt_prompt.py`
+
+**수정 코드:**
+
+```python
+# 상단 import 추가
+import httpx
+from openai import APITimeoutError, APIConnectionError, RateLimitError
+from app.utils.retry import retry_async
+
+# 재시도 대상 예외 정의
+RETRY_EXCEPTIONS = (
+    APITimeoutError,
+    APIConnectionError,
+    RateLimitError,
+)
+
+class ChatgptService:
+    DEFAULT_TIMEOUT = 60.0
+    CONNECT_TIMEOUT = 10.0
+    MAX_RETRIES = 3
+
+    def __init__(self):
+        self.client = AsyncOpenAI(
+            api_key=apikey_settings.CHATGPT_API_KEY,
+            timeout=httpx.Timeout(
+                self.DEFAULT_TIMEOUT,
+                connect=self.CONNECT_TIMEOUT,
+            ),
+        )
+
+    async def _call_structured_output_with_response_gpt_api(
+        self, prompt: str, output_format: dict, model: str
+    ) -> dict:
+        content = [{"type": "input_text", "text": prompt}]
+
+        async def _call():
+            response = await self.client.responses.create(
+                model=model,
+                input=[{"role": "user", "content": content}],
+                text=output_format,
+            )
+            return json.loads(response.output_text) or {}
+
+        return await retry_async(
+            _call,
+            max_retries=self.MAX_RETRIES,
+            base_delay=2.0,  # OpenAI Rate Limit 대비 더 긴 대기
+            retry_on=RETRY_EXCEPTIONS,
+        )
+```
+
+---
+
+## 타임아웃 설정 권장값
+
+| API | 용도 | 권장 타임아웃 | 재시도 횟수 | 재시도 간격 |
+|-----|------|---------------|-------------|-------------|
+| ChatGPT | 가사 생성 | 60초 | 3회 | 2초 → 4초 → 8초 |
+| Suno | 노래 생성 요청 | 30초 | 3회 | 1초 → 2초 → 4초 |
+| Suno | 상태 조회 | 30초 | 2회 | 1초 → 2초 |
+| Suno | 타임스탬프 | 120초 | 2회 | 2초 → 4초 |
+| Creatomate | 템플릿 조회 | 30초 | 2회 | 1초 → 2초 |
+| Creatomate | 렌더링 요청 | 60초 | 3회 | 1초 → 2초 → 4초 |
+| Creatomate | 상태 조회 | 30초 | 2회 | 1초 → 2초 |
+
+---
+
+## 구현 순서
+
+1. **1단계: retry.py 유틸리티 생성**
+   - 재사용 가능한 재시도 로직 구현
+   - 단위 테스트 작성
+
+2. **2단계: ChatGPT 타임아웃 설정**
+   - 가장 시급한 문제 (현재 600초 기본값)
+   - 타임아웃 + 재시도 동시 적용
+
+3. **3단계: Suno API 재시도 적용**
+   - generate(), get_task_status(), get_lyric_timestamp()
+
+4. **4단계: Creatomate API 재시도 적용**
+   - _request() 메서드 수정으로 전체 적용
+
+---
+
+## 테스트 체크리스트
+
+각 수정 후 확인 사항:
+
+- [ ] 정상 요청 시 기존과 동일하게 동작
+- [ ] 타임아웃 발생 시 지정된 시간 내 예외 발생
+- [ ] 일시적 오류 시 재시도 후 성공
+- [ ] 모든 재시도 실패 시 적절한 에러 메시지 반환
+- [ ] 로그에 재시도 시도 기록 확인
+
+---
+
+## 롤백 계획
+
+문제 발생 시:
+1. retry.py 사용 코드 제거 (기존 직접 호출로 복구)
+2. ChatGPT 타임아웃 설정 제거 (SDK 기본값으로 복구)
+
+---
+
+## 참고 사항
+
+- OpenAI SDK는 내부적으로 일부 재시도 로직이 있으나, 커스텀 제어가 제한적
+- httpx의 `TimeoutException`은 `ConnectTimeout`, `ReadTimeout`, `WriteTimeout`, `PoolTimeout`을 포함
+- Rate Limit 에러(429)는 재시도 시 더 긴 대기 시간 필요 (Retry-After 헤더 참고)
diff --git a/docs/plan/insert_for_update_plan.md b/docs/plan/insert_for_update_plan.md
new file mode 100644
index 0000000..a75ea51
--- /dev/null
+++ b/docs/plan/insert_for_update_plan.md
@@ -0,0 +1,447 @@
+# INSERT → Upsert 변환 계획서
+
+## 개요
+
+이 문서는 프로젝트 내 INSERT 코드들을 **존재 여부 확인 후 INSERT 또는 DB Lock + UPDATE** 패턴으로 변환하기 위한 계획입니다.
+
+### 적용 범위
+- ✅ 포함: video, song, lyric 모듈의 INSERT 코드
+- ❌ 제외: user 모듈, 이미지 업로드 엔드포인트
+
+### 사용할 유틸리티
+- `app/utils/db_utils.py`의 `get_or_create_with_lock()` 함수
+- `app/utils/db_utils.py`의 `bulk_upsert()` 함수 (대량 INSERT용)
+
+---
+
+## 대상 INSERT 위치 목록
+
+| # | 파일 | 라인 | 테이블 | 키 컬럼 | 우선순위 |
+|---|------|------|--------|---------|----------|
+| 1 | video.py | 252 | Video | task_id | 중 |
+| 2 | song.py | 188 | Song | task_id | 중 |
+| 3 | song.py | 477 | SongTimestamp | suno_audio_id + order_idx | **높음** |
+| 4 | lyric.py | 297 | Project | task_id | 중 |
+| 5 | lyric.py | 317 | Lyric | task_id | 중 |
+
+---
+
+## 1. Video INSERT (video.py:252)
+
+### 현재 코드
+```python
+# app/video/api/routers/v1/video.py:244-254
+
+video = Video(
+    project_id=project_id,
+    lyric_id=lyric_id,
+    song_id=song_id,
+    task_id=task_id,
+    creatomate_render_id=None,
+    status="processing",
+)
+session.add(video)
+await session.commit()
+video_id = video.id
+```
+
+### 문제점
+- `task_id`로 기존 레코드 존재 여부를 확인하지 않음
+- 동일한 `task_id`로 재요청 시 중복 INSERT 발생 가능
+
+### 수정 계획
+
+```python
+# app/video/api/routers/v1/video.py
+
+# 상단 import 추가
+from app.utils.db_utils import get_or_create_with_lock
+
+# 기존 코드 대체 (244-254행)
+result = await get_or_create_with_lock(
+    session=session,
+    model=Video,
+    filter_by={'task_id': task_id},
+    defaults={
+        'project_id': project_id,
+        'lyric_id': lyric_id,
+        'song_id': song_id,
+        'creatomate_render_id': None,
+        'status': 'processing',
+    },
+    lock=True,
+)
+
+video = result.entity
+if not result.created:
+    # 이미 존재하는 경우: 상태 업데이트
+    video.project_id = project_id
+    video.lyric_id = lyric_id
+    video.song_id = song_id
+    video.status = 'processing'
+    video.creatomate_render_id = None
+
+await session.commit()
+video_id = video.id
+```
+
+### 필수 사전 작업
+- Video 테이블에 `task_id` UNIQUE 인덱스 추가 필요
+
+```sql
+ALTER TABLE video ADD UNIQUE INDEX idx_video_task_id (task_id);
+```
+
+---
+
+## 2. Song INSERT (song.py:188)
+
+### 현재 코드
+```python
+# app/song/api/routers/v1/song.py:179-191
+
+song = Song(
+    project_id=project_id,
+    lyric_id=lyric_id,
+    task_id=task_id,
+    suno_task_id=None,
+    status="processing",
+    song_prompt=song_prompt,
+    language=request_body.language,
+)
+session.add(song)
+await session.commit()
+song_id = song.id
+```
+
+### 문제점
+- `task_id`로 기존 레코드 존재 여부를 확인하지 않음
+- 동일한 `task_id`로 재요청 시 중복 INSERT 발생 가능
+
+### 수정 계획
+
+```python
+# app/song/api/routers/v1/song.py
+
+# 상단 import 추가
+from app.utils.db_utils import get_or_create_with_lock
+
+# 기존 코드 대체 (179-191행)
+result = await get_or_create_with_lock(
+    session=session,
+    model=Song,
+    filter_by={'task_id': task_id},
+    defaults={
+        'project_id': project_id,
+        'lyric_id': lyric_id,
+        'suno_task_id': None,
+        'status': 'processing',
+        'song_prompt': song_prompt,
+        'language': request_body.language,
+    },
+    lock=True,
+)
+
+song = result.entity
+if not result.created:
+    # 이미 존재하는 경우: 상태 업데이트
+    song.project_id = project_id
+    song.lyric_id = lyric_id
+    song.status = 'processing'
+    song.song_prompt = song_prompt
+    song.language = request_body.language
+
+await session.commit()
+song_id = song.id
+```
+
+### 필수 사전 작업
+- Song 테이블에 `task_id` UNIQUE 인덱스 추가 필요
+
+```sql
+ALTER TABLE song ADD UNIQUE INDEX idx_song_task_id (task_id);
+```
+
+---
+
+## 3. SongTimestamp INSERT (song.py:477) ⚠️ 높은 우선순위
+
+### 현재 코드
+```python
+# app/song/api/routers/v1/song.py:467-479
+
+for order_idx, timestamped_lyric in enumerate(timestamped_lyrics):
+    song_timestamp = SongTimestamp(
+        suno_audio_id=suno_audio_id,
+        order_idx=order_idx,
+        lyric_line=timestamped_lyric["text"],
+        start_time=timestamped_lyric["start_sec"],
+        end_time=timestamped_lyric["end_sec"],
+    )
+    session.add(song_timestamp)
+
+await session.commit()
+```
+
+### 문제점
+- **폴링 기반 요청으로 인해 중복 INSERT 위험이 높음**
+- `suno_audio_id` + `order_idx` 조합으로 존재 여부 확인 없음
+- 여러 행을 루프에서 개별 INSERT하여 비효율적
+
+### 수정 계획 (bulk_upsert 사용)
+
+```python
+# app/song/api/routers/v1/song.py
+
+# 상단 import 추가
+from app.utils.db_utils import bulk_upsert
+
+# 기존 코드 대체 (467-479행)
+timestamp_records = [
+    {
+        'suno_audio_id': suno_audio_id,
+        'order_idx': order_idx,
+        'lyric_line': timestamped_lyric["text"],
+        'start_time': timestamped_lyric["start_sec"],
+        'end_time': timestamped_lyric["end_sec"],
+    }
+    for order_idx, timestamped_lyric in enumerate(timestamped_lyrics)
+]
+
+await bulk_upsert(
+    session=session,
+    model=SongTimestamp,
+    unique_columns=['suno_audio_id', 'order_idx'],
+    records=timestamp_records,
+    update_columns=['lyric_line', 'start_time', 'end_time'],
+)
+
+await session.commit()
+```
+
+### 필수 사전 작업
+- SongTimestamp 테이블에 복합 UNIQUE 인덱스 추가 필요
+
+```sql
+ALTER TABLE song_timestamp
+ADD UNIQUE INDEX idx_song_timestamp_audio_order (suno_audio_id, order_idx);
+```
+
+### 대안: get_or_create_with_lock 사용 (개별 처리)
+
+만약 `bulk_upsert`를 사용하지 않고 개별 처리가 필요한 경우:
+
+```python
+from app.utils.db_utils import get_or_create_with_lock
+
+for order_idx, timestamped_lyric in enumerate(timestamped_lyrics):
+    result = await get_or_create_with_lock(
+        session=session,
+        model=SongTimestamp,
+        filter_by={
+            'suno_audio_id': suno_audio_id,
+            'order_idx': order_idx,
+        },
+        defaults={
+            'lyric_line': timestamped_lyric["text"],
+            'start_time': timestamped_lyric["start_sec"],
+            'end_time': timestamped_lyric["end_sec"],
+        },
+        lock=True,
+    )
+
+    if not result.created:
+        # 이미 존재하는 경우: 업데이트
+        result.entity.lyric_line = timestamped_lyric["text"]
+        result.entity.start_time = timestamped_lyric["start_sec"]
+        result.entity.end_time = timestamped_lyric["end_sec"]
+
+await session.commit()
+```
+
+**권장사항**: `bulk_upsert` 사용 (성능 및 데드락 방지 측면에서 우수)
+
+---
+
+## 4. Project INSERT (lyric.py:297)
+
+### 현재 코드
+```python
+# app/lyric/api/routers/v1/lyric.py:290-299
+
+project = Project(
+    store_name=request_body.customer_name,
+    region=request_body.region,
+    task_id=task_id,
+    detail_region_info=request_body.detail_region_info,
+    language=request_body.language,
+)
+session.add(project)
+await session.commit()
+await session.refresh(project)
+```
+
+### 문제점
+- `task_id`로 기존 레코드 존재 여부를 확인하지 않음
+- 동일한 `task_id`로 재요청 시 중복 INSERT 발생 가능
+
+### 수정 계획
+
+```python
+# app/lyric/api/routers/v1/lyric.py
+
+# 상단 import 추가
+from app.utils.db_utils import get_or_create_with_lock
+
+# 기존 코드 대체 (290-299행)
+result = await get_or_create_with_lock(
+    session=session,
+    model=Project,
+    filter_by={'task_id': task_id},
+    defaults={
+        'store_name': request_body.customer_name,
+        'region': request_body.region,
+        'detail_region_info': request_body.detail_region_info,
+        'language': request_body.language,
+    },
+    lock=True,
+)
+
+project = result.entity
+if not result.created:
+    # 이미 존재하는 경우: 정보 업데이트
+    project.store_name = request_body.customer_name
+    project.region = request_body.region
+    project.detail_region_info = request_body.detail_region_info
+    project.language = request_body.language
+
+await session.commit()
+await session.refresh(project)
+```
+
+### 필수 사전 작업
+- Project 테이블에 `task_id` UNIQUE 인덱스 추가 필요
+
+```sql
+ALTER TABLE project ADD UNIQUE INDEX idx_project_task_id (task_id);
+```
+
+---
+
+## 5. Lyric INSERT (lyric.py:317)
+
+### 현재 코드
+```python
+# app/lyric/api/routers/v1/lyric.py:308-319
+
+estimated_prompt = lyric_prompt.build_prompt(lyric_input_data)
+lyric = Lyric(
+    project_id=project.id,
+    task_id=task_id,
+    status="processing",
+    lyric_prompt=estimated_prompt,
+    lyric_result=None,
+    language=request_body.language,
+)
+session.add(lyric)
+await session.commit()
+await session.refresh(lyric)
+```
+
+### 문제점
+- `task_id`로 기존 레코드 존재 여부를 확인하지 않음
+- 동일한 `task_id`로 재요청 시 중복 INSERT 발생 가능
+
+### 수정 계획
+
+```python
+# app/lyric/api/routers/v1/lyric.py
+
+# 상단 import (이미 추가되어 있음)
+from app.utils.db_utils import get_or_create_with_lock
+
+# 기존 코드 대체 (308-319행)
+estimated_prompt = lyric_prompt.build_prompt(lyric_input_data)
+
+result = await get_or_create_with_lock(
+    session=session,
+    model=Lyric,
+    filter_by={'task_id': task_id},
+    defaults={
+        'project_id': project.id,
+        'status': 'processing',
+        'lyric_prompt': estimated_prompt,
+        'lyric_result': None,
+        'language': request_body.language,
+    },
+    lock=True,
+)
+
+lyric = result.entity
+if not result.created:
+    # 이미 존재하는 경우: 정보 업데이트
+    lyric.project_id = project.id
+    lyric.status = 'processing'
+    lyric.lyric_prompt = estimated_prompt
+    lyric.lyric_result = None
+    lyric.language = request_body.language
+
+await session.commit()
+await session.refresh(lyric)
+```
+
+### 필수 사전 작업
+- Lyric 테이블에 `task_id` UNIQUE 인덱스 추가 필요
+
+```sql
+ALTER TABLE lyric ADD UNIQUE INDEX idx_lyric_task_id (task_id);
+```
+
+---
+
+## 구현 순서 권장
+
+1. **1단계: DB 마이그레이션** (필수)
+   ```sql
+   -- 모든 UNIQUE 인덱스 추가
+   ALTER TABLE video ADD UNIQUE INDEX idx_video_task_id (task_id);
+   ALTER TABLE song ADD UNIQUE INDEX idx_song_task_id (task_id);
+   ALTER TABLE song_timestamp ADD UNIQUE INDEX idx_song_timestamp_audio_order (suno_audio_id, order_idx);
+   ALTER TABLE project ADD UNIQUE INDEX idx_project_task_id (task_id);
+   ALTER TABLE lyric ADD UNIQUE INDEX idx_lyric_task_id (task_id);
+   ```
+
+2. **2단계: SongTimestamp (높은 우선순위)**
+   - 폴링으로 인한 중복 INSERT 위험이 가장 높음
+   - `bulk_upsert` 사용 권장
+
+3. **3단계: 나머지 테이블 (중간 우선순위)**
+   - Video, Song, Project, Lyric 순으로 적용
+   - 모두 `get_or_create_with_lock` 사용
+
+---
+
+## 롤백 계획
+
+문제 발생 시 원래 코드로 복구:
+
+1. 코드 변경 사항 git revert
+2. UNIQUE 인덱스는 유지 (데이터 무결성에 도움됨)
+
+---
+
+## 테스트 체크리스트
+
+각 수정 후 확인 사항:
+
+- [ ] 새로운 task_id로 요청 시 정상 INSERT
+- [ ] 동일한 task_id로 재요청 시 UPDATE (에러 없이)
+- [ ] 동시 요청 테스트 (2개 이상 동시 요청)
+- [ ] 성능 저하 없는지 확인
+
+---
+
+## 참고 문서
+
+- [db_lock.md](./db_lock.md) - DB Lock 및 Upsert 패턴 가이드
+- [db_utils.py](../../app/utils/db_utils.py) - 유틸리티 함수 구현체
diff --git a/error_plan.md b/error_plan.md
new file mode 100644
index 0000000..043cc1d
--- /dev/null
+++ b/error_plan.md
@@ -0,0 +1,571 @@
+# Suno API & Creatomate API 에러 처리 작업 계획서
+
+## 현재 상태 분석
+
+### ✅ 이미 구현된 항목
+| 항목 | Suno API | Creatomate API |
+|------|----------|----------------|
+| DB failed 상태 저장 | ✅ `song_task.py` | ✅ `video_task.py` |
+| Response failed 상태 반환 | ✅ `song_schema.py` | ✅ `video_schema.py` |
+
+### ❌ 미구현 항목
+| 항목 | Suno API | Creatomate API |
+|------|----------|----------------|
+| 타임아웃 외부화 | ❌ 하드코딩됨 | ❌ 하드코딩됨 |
+| 재시도 로직 | ❌ 없음 | ❌ 없음 |
+| 커스텀 예외 클래스 | ❌ 없음 | ❌ 없음 |
+
+---
+
+## 1. RecoverySettings에 Suno/Creatomate 설정 추가
+
+### 파일: `config.py`
+
+**변경 전:**
+```python
+class RecoverySettings(BaseSettings):
+    """ChatGPT API 복구 및 타임아웃 설정"""
+
+    CHATGPT_TIMEOUT: float = Field(
+        default=600.0,
+        description="ChatGPT API 타임아웃 (초). OpenAI Python SDK 기본값: 600초 (10분)",
+    )
+    CHATGPT_MAX_RETRIES: int = Field(
+        default=1,
+        description="ChatGPT API 응답 실패 시 최대 재시도 횟수",
+    )
+
+    model_config = _base_config
+```
+
+**변경 후:**
+```python
+class RecoverySettings(BaseSettings):
+    """외부 API 복구 및 타임아웃 설정
+
+    ChatGPT, Suno, Creatomate API의 타임아웃 및 재시도 설정을 관리합니다.
+    """
+
+    # ============================================================
+    # ChatGPT API 설정
+    # ============================================================
+    CHATGPT_TIMEOUT: float = Field(
+        default=600.0,
+        description="ChatGPT API 타임아웃 (초). OpenAI Python SDK 기본값: 600초 (10분)",
+    )
+    CHATGPT_MAX_RETRIES: int = Field(
+        default=1,
+        description="ChatGPT API 응답 실패 시 최대 재시도 횟수",
+    )
+
+    # ============================================================
+    # Suno API 설정
+    # ============================================================
+    SUNO_DEFAULT_TIMEOUT: float = Field(
+        default=30.0,
+        description="Suno API 기본 요청 타임아웃 (초)",
+    )
+    SUNO_LYRIC_TIMEOUT: float = Field(
+        default=120.0,
+        description="Suno API 가사 타임스탬프 요청 타임아웃 (초)",
+    )
+    SUNO_MAX_RETRIES: int = Field(
+        default=2,
+        description="Suno API 응답 실패 시 최대 재시도 횟수",
+    )
+
+    # ============================================================
+    # Creatomate API 설정
+    # ============================================================
+    CREATOMATE_DEFAULT_TIMEOUT: float = Field(
+        default=30.0,
+        description="Creatomate API 기본 요청 타임아웃 (초)",
+    )
+    CREATOMATE_RENDER_TIMEOUT: float = Field(
+        default=60.0,
+        description="Creatomate API 렌더링 요청 타임아웃 (초)",
+    )
+    CREATOMATE_CONNECT_TIMEOUT: float = Field(
+        default=10.0,
+        description="Creatomate API 연결 타임아웃 (초)",
+    )
+    CREATOMATE_MAX_RETRIES: int = Field(
+        default=2,
+        description="Creatomate API 응답 실패 시 최대 재시도 횟수",
+    )
+
+    model_config = _base_config
+```
+
+**이유:** 모든 외부 API의 타임아웃/재시도 설정을 `RecoverySettings` 하나에서 통합 관리하여 일관성을 유지합니다.
+
+### 파일: `.env`
+
+**추가할 내용:**
+```env
+# ============================================================
+# 외부 API 타임아웃 및 재시도 설정 (RecoverySettings)
+# ============================================================
+
+# ChatGPT API (기존)
+CHATGPT_TIMEOUT=600.0
+CHATGPT_MAX_RETRIES=1
+
+# Suno API
+SUNO_DEFAULT_TIMEOUT=30.0
+SUNO_LYRIC_TIMEOUT=120.0
+SUNO_MAX_RETRIES=2
+
+# Creatomate API
+CREATOMATE_DEFAULT_TIMEOUT=30.0
+CREATOMATE_RENDER_TIMEOUT=60.0
+CREATOMATE_CONNECT_TIMEOUT=10.0
+CREATOMATE_MAX_RETRIES=2
+```
+
+---
+
+## 2. Suno API 커스텀 예외 클래스 추가
+
+### 파일: `app/utils/suno.py`
+
+**변경 전 (라인 1-20):**
+```python
+import httpx
+import json
+from typing import Optional
+
+from app.utils.logger import get_logger
+
+logger = get_logger("suno")
+```
+
+**변경 후:**
+```python
+import httpx
+import json
+from typing import Optional
+
+from app.utils.logger import get_logger
+from config import recovery_settings
+
+logger = get_logger("suno")
+
+
+class SunoResponseError(Exception):
+    """Suno API 응답 오류 시 발생하는 예외
+
+    Suno API 거부 응답 또는 비정상 응답 시 사용됩니다.
+    재시도 로직에서 이 예외를 catch하여 재시도를 수행합니다.
+
+    Attributes:
+        message: 에러 메시지
+        original_response: 원본 API 응답 (있는 경우)
+    """
+    def __init__(self, message: str, original_response: dict | None = None):
+        self.message = message
+        self.original_response = original_response
+        super().__init__(self.message)
+```
+
+**이유:** ChatGPT API와 동일하게 커스텀 예외 클래스를 추가하여 Suno API 오류를 명확히 구분하고 재시도 로직에서 활용합니다.
+
+---
+
+## 3. Suno API 타임아웃 적용
+
+### 파일: `app/utils/suno.py`
+
+**변경 전 (라인 130):**
+```python
+async with httpx.AsyncClient() as client:
+    response = await client.post(
+        f"{self.base_url}/generate",
+        headers=self.headers,
+        json=payload,
+        timeout=30.0,
+    )
+```
+
+**변경 후:**
+```python
+async with httpx.AsyncClient() as client:
+    response = await client.post(
+        f"{self.base_url}/generate",
+        headers=self.headers,
+        json=payload,
+        timeout=recovery_settings.SUNO_DEFAULT_TIMEOUT,
+    )
+```
+
+**변경 전 (라인 173):**
+```python
+timeout=30.0,
+```
+
+**변경 후:**
+```python
+timeout=recovery_settings.SUNO_DEFAULT_TIMEOUT,
+```
+
+**변경 전 (라인 201):**
+```python
+timeout=120.0,
+```
+
+**변경 후:**
+```python
+timeout=recovery_settings.SUNO_LYRIC_TIMEOUT,
+```
+
+**이유:** 환경변수로 타임아웃을 관리하여 배포 환경별로 유연하게 조정할 수 있습니다.
+
+---
+
+## 4. Suno API 재시도 로직 추가
+
+### 파일: `app/utils/suno.py`
+
+**변경 전 - generate() 메서드:**
+```python
+async def generate(
+    self,
+    lyric: str,
+    style: str,
+    title: str,
+    task_id: str,
+) -> dict:
+    """음악 생성 요청"""
+    payload = {
+        "prompt": lyric,
+        "style": style,
+        "title": title,
+        "customMode": True,
+        "callbackUrl": f"{self.callback_url}?task_id={task_id}",
+    }
+
+    async with httpx.AsyncClient() as client:
+        response = await client.post(
+            f"{self.base_url}/generate",
+            headers=self.headers,
+            json=payload,
+            timeout=30.0,
+        )
+
+        if response.status_code != 200:
+            logger.error(f"Failed to generate music: {response.text}")
+            raise Exception(f"Failed to generate music: {response.status_code}")
+
+        return response.json()
+```
+
+**변경 후:**
+```python
+async def generate(
+    self,
+    lyric: str,
+    style: str,
+    title: str,
+    task_id: str,
+) -> dict:
+    """음악 생성 요청 (재시도 로직 포함)
+
+    Args:
+        lyric: 가사 텍스트
+        style: 음악 스타일
+        title: 곡 제목
+        task_id: 작업 고유 식별자
+
+    Returns:
+        Suno API 응답 데이터
+
+    Raises:
+        SunoResponseError: API 오류 또는 재시도 실패 시
+    """
+    payload = {
+        "prompt": lyric,
+        "style": style,
+        "title": title,
+        "customMode": True,
+        "callbackUrl": f"{self.callback_url}?task_id={task_id}",
+    }
+
+    last_error: Exception | None = None
+
+    for attempt in range(recovery_settings.SUNO_MAX_RETRIES + 1):
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    f"{self.base_url}/generate",
+                    headers=self.headers,
+                    json=payload,
+                    timeout=recovery_settings.SUNO_DEFAULT_TIMEOUT,
+                )
+
+                if response.status_code == 200:
+                    return response.json()
+
+                # 재시도 불가능한 오류 (4xx 클라이언트 오류)
+                if 400 <= response.status_code < 500:
+                    raise SunoResponseError(
+                        f"Client error: {response.status_code}",
+                        original_response={"status": response.status_code, "text": response.text}
+                    )
+
+                # 재시도 가능한 오류 (5xx 서버 오류)
+                last_error = SunoResponseError(
+                    f"Server error: {response.status_code}",
+                    original_response={"status": response.status_code, "text": response.text}
+                )
+
+        except httpx.TimeoutException as e:
+            logger.warning(f"[Suno] Timeout on attempt {attempt + 1}/{recovery_settings.SUNO_MAX_RETRIES + 1}")
+            last_error = e
+
+        except httpx.HTTPError as e:
+            logger.warning(f"[Suno] HTTP error on attempt {attempt + 1}: {e}")
+            last_error = e
+
+        # 마지막 시도가 아니면 재시도
+        if attempt < recovery_settings.SUNO_MAX_RETRIES:
+            logger.info(f"[Suno] Retrying... ({attempt + 1}/{recovery_settings.SUNO_MAX_RETRIES})")
+
+    # 모든 재시도 실패
+    raise SunoResponseError(
+        f"All {recovery_settings.SUNO_MAX_RETRIES + 1} attempts failed",
+        original_response={"last_error": str(last_error)}
+    )
+```
+
+**이유:** 네트워크 오류나 일시적인 서버 오류 시 자동으로 재시도하여 안정성을 높입니다.
+
+---
+
+## 5. Creatomate API 커스텀 예외 클래스 추가
+
+### 파일: `app/utils/creatomate.py`
+
+**변경 전 (라인 1-20):**
+```python
+import asyncio
+import json
+from typing import Any, Optional
+
+import httpx
+
+from app.utils.logger import get_logger
+from config import creatomate_settings
+
+logger = get_logger("creatomate")
+```
+
+**변경 후:**
+```python
+import asyncio
+import json
+from typing import Any, Optional
+
+import httpx
+
+from app.utils.logger import get_logger
+from config import creatomate_settings, recovery_settings
+
+logger = get_logger("creatomate")
+
+
+class CreatomateResponseError(Exception):
+    """Creatomate API 응답 오류 시 발생하는 예외
+
+    Creatomate API 렌더링 실패 또는 비정상 응답 시 사용됩니다.
+    재시도 로직에서 이 예외를 catch하여 재시도를 수행합니다.
+
+    Attributes:
+        message: 에러 메시지
+        original_response: 원본 API 응답 (있는 경우)
+    """
+    def __init__(self, message: str, original_response: dict | None = None):
+        self.message = message
+        self.original_response = original_response
+        super().__init__(self.message)
+```
+
+---
+
+## 6. Creatomate API 타임아웃 적용
+
+### 파일: `app/utils/creatomate.py`
+
+**변경 전 (라인 138):**
+```python
+self._client = httpx.AsyncClient(
+    base_url=self.BASE_URL,
+    headers=self._get_headers(),
+    timeout=httpx.Timeout(60.0, connect=10.0),
+)
+```
+
+**변경 후:**
+```python
+self._client = httpx.AsyncClient(
+    base_url=self.BASE_URL,
+    headers=self._get_headers(),
+    timeout=httpx.Timeout(
+        recovery_settings.CREATOMATE_RENDER_TIMEOUT,
+        connect=recovery_settings.CREATOMATE_CONNECT_TIMEOUT
+    ),
+)
+```
+
+**변경 전 (라인 258, 291):**
+```python
+timeout=30.0
+```
+
+**변경 후:**
+```python
+timeout=recovery_settings.CREATOMATE_DEFAULT_TIMEOUT
+```
+
+**변경 전 (라인 446, 457):**
+```python
+timeout=60.0
+```
+
+**변경 후:**
+```python
+timeout=recovery_settings.CREATOMATE_RENDER_TIMEOUT
+```
+
+---
+
+## 7. Creatomate API 재시도 로직 추가
+
+### 파일: `app/utils/creatomate.py`
+
+**변경 전 - render_with_json() 메서드 (라인 440~):**
+```python
+async def render_with_json(
+    self,
+    template_id: str,
+    modifications: dict[str, Any],
+    task_id: str,
+) -> dict:
+    """JSON 수정사항으로 렌더링 요청"""
+    payload = {
+        "template_id": template_id,
+        "modifications": modifications,
+        "webhook_url": f"{creatomate_settings.CREATOMATE_CALLBACK_URL}?task_id={task_id}",
+    }
+
+    async with httpx.AsyncClient() as client:
+        response = await client.post(
+            f"{self.BASE_URL}/v1/renders",
+            headers=self._get_headers(),
+            json=payload,
+            timeout=60.0,
+        )
+
+        if response.status_code != 200:
+            logger.error(f"Failed to render: {response.text}")
+            raise Exception(f"Failed to render: {response.status_code}")
+
+        return response.json()
+```
+
+**변경 후:**
+```python
+async def render_with_json(
+    self,
+    template_id: str,
+    modifications: dict[str, Any],
+    task_id: str,
+) -> dict:
+    """JSON 수정사항으로 렌더링 요청 (재시도 로직 포함)
+
+    Args:
+        template_id: Creatomate 템플릿 ID
+        modifications: 수정사항 딕셔너리
+        task_id: 작업 고유 식별자
+
+    Returns:
+        Creatomate API 응답 데이터
+
+    Raises:
+        CreatomateResponseError: API 오류 또는 재시도 실패 시
+    """
+    payload = {
+        "template_id": template_id,
+        "modifications": modifications,
+        "webhook_url": f"{creatomate_settings.CREATOMATE_CALLBACK_URL}?task_id={task_id}",
+    }
+
+    last_error: Exception | None = None
+
+    for attempt in range(recovery_settings.CREATOMATE_MAX_RETRIES + 1):
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    f"{self.BASE_URL}/v1/renders",
+                    headers=self._get_headers(),
+                    json=payload,
+                    timeout=recovery_settings.CREATOMATE_RENDER_TIMEOUT,
+                )
+
+                if response.status_code == 200:
+                    return response.json()
+
+                # 재시도 불가능한 오류 (4xx 클라이언트 오류)
+                if 400 <= response.status_code < 500:
+                    raise CreatomateResponseError(
+                        f"Client error: {response.status_code}",
+                        original_response={"status": response.status_code, "text": response.text}
+                    )
+
+                # 재시도 가능한 오류 (5xx 서버 오류)
+                last_error = CreatomateResponseError(
+                    f"Server error: {response.status_code}",
+                    original_response={"status": response.status_code, "text": response.text}
+                )
+
+        except httpx.TimeoutException as e:
+            logger.warning(f"[Creatomate] Timeout on attempt {attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES + 1}")
+            last_error = e
+
+        except httpx.HTTPError as e:
+            logger.warning(f"[Creatomate] HTTP error on attempt {attempt + 1}: {e}")
+            last_error = e
+
+        # 마지막 시도가 아니면 재시도
+        if attempt < recovery_settings.CREATOMATE_MAX_RETRIES:
+            logger.info(f"[Creatomate] Retrying... ({attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES})")
+
+    # 모든 재시도 실패
+    raise CreatomateResponseError(
+        f"All {recovery_settings.CREATOMATE_MAX_RETRIES + 1} attempts failed",
+        original_response={"last_error": str(last_error)}
+    )
+```
+
+---
+
+## 작업 체크리스트
+
+| 순번 | 작업 내용 | 파일 | 상태 |
+|------|----------|------|------|
+| 1 | RecoverySettings에 Suno/Creatomate 설정 추가 | `config.py` | ✅ |
+| 2 | .env에 타임아웃/재시도 환경변수 추가 | `.env` | ✅ |
+| 3 | SunoResponseError 예외 클래스 추가 | `app/utils/suno.py` | ✅ |
+| 4 | Suno 타임아웃 적용 (recovery_settings 사용) | `app/utils/suno.py` | ✅ |
+| 5 | Suno 재시도 로직 추가 | `app/utils/suno.py` | ✅ |
+| 6 | CreatomateResponseError 예외 클래스 추가 | `app/utils/creatomate.py` | ✅ |
+| 7 | Creatomate 타임아웃 적용 (recovery_settings 사용) | `app/utils/creatomate.py` | ✅ |
+| 8 | Creatomate 재시도 로직 추가 | `app/utils/creatomate.py` | ✅ |
+
+---
+
+## 참고사항
+
+- **DB failed 상태 저장**: `song_task.py`와 `video_task.py`에 이미 구현되어 있습니다.
+- **Response failed 상태**: 모든 스키마에 `success`, `status` 필드가 이미 존재합니다.
+- 재시도는 5xx 서버 오류와 타임아웃에만 적용되며, 4xx 클라이언트 오류는 즉시 실패 처리합니다.
+- 모든 타임아웃/재시도 설정은 `RecoverySettings`에서 통합 관리합니다.
diff --git a/main.py b/main.py
index b205341..c3ba268 100644
--- a/main.py
+++ b/main.py
@@ -38,10 +38,10 @@ tags_metadata = [
 - **Refresh Token**: 7일 유효, Access Token 갱신 시 사용
 """,
     },
-    {
-        "name": "Home",
-        "description": "홈 화면 및 프로젝트 관리 API",
-    },
+    # {
+    #     "name": "Home",
+    #     "description": "홈 화면 및 프로젝트 관리 API",
+    # },
     {
         "name": "Crawling",
         "description": "네이버 지도 크롤링 API - 장소 정보 및 이미지 수집",