并发上传附件时MS-Graph返回404/412错误的排查求助

核心结论 #

MS Graph支持并发上传同一草稿邮件的多个附件，但你的脚本存在几处关键问题，导致并发时出现404（上传会话失效）和412（范围不匹配）错误。

脚本问题分析 #

1. 同步创建草稿阻塞事件循环
   创建草稿的代码是同步调用：

   draft = self._graph_client.users.by_user_id(email_def.sender_address).messages.post(message)
   

   但整个_send_large_email是异步方法，同步调用会阻塞asyncio事件循环，导致后续并发上传附件时，Graph服务器可能还未完全完成草稿的创建同步，此时创建上传会话或上传分块会出现404错误。

2. 分块上传未校验会话状态
   412错误的核心原因是你上传的Content-Range和Graph服务器记录的已上传范围不匹配。并发上传时，服务器端的会话状态更新可能存在延迟，你的代码直接按本地计算的范围上传，没有先获取会话当前的已上传状态，导致范围冲突。

3. 重复创建aiohttp会话
   upload_chunks方法里每次都新建aiohttp.ClientSession，这会导致TCP连接无法复用，增加请求延迟，还可能触发Graph的连接数限制，间接导致请求失败。

4. 缺少重试机制
   Graph API对并发请求可能出现临时的状态不一致或限流，你的代码遇到错误直接抛出，没有针对404（会话可能需重新获取）、412（范围冲突）等可重试错误做重试处理。

修复后的关键代码修改 #

1. 异步创建草稿 #

# 原同步代码
# draft = self._graph_client.users.by_user_id(email_def.sender_address).messages.post(message)
# 修改为异步调用
draft = await self._graph_client.users.by_user_id(email_def.sender_address).messages.post(message)

2. 复用aiohttp会话 #

在_send_large_email中创建全局会话，传给所有上传任务：

async def _send_large_email(self, email_def: EmailDefinition):
    # ... 创建草稿的代码 ...
    
    # 创建全局aiohttp会话供所有上传任务复用
    async with aiohttp.ClientSession() as session:
        sem = asyncio.Semaphore(4)
        
        async def upload_one(att: EncodedAttachment) -> None:
            async with sem:
                await self.upload_attachment(email_def.sender_address, draft.id, att, session)
        
        tasks = [asyncio.create_task(upload_one(att)) for att in email_def.attachments]
        await asyncio.gather(*tasks)
        
    # ... 发送草稿的代码 ...

3. 分块上传前校验会话状态 #

修改upload_chunks方法，每次上传前获取会话状态，调整上传范围：

@staticmethod
async def upload_chunks(attachment, attachment_size, upload_session, session):
    while True:
        # 获取当前会话状态，确认已上传范围
        async with session.get(upload_session.upload_url) as resp:
            resp.raise_for_status()
            session_status = await resp.json()
            next_expected = session_status.get('nextExpectedRanges', ['0-'])[0]
            start = int(next_expected.split('-')[0])
            if start >= attachment_size:
                break  # 上传完成
        
        chunk = attachment.content[start: start + CHUNK_SIZE]
        end = start + len(chunk) - 1
        total = attachment_size
        
        headers = {
            "Content-Length": str(len(chunk)),
            "Content-Range": f"bytes {start}-{end}/{total}",
            "Content-Type": "application/octet-stream",
        }
        
        await EmailSenderMSG.upload_single_chunk(
            session,
            upload_session.upload_url,
            chunk,
            headers,
            attachment.name,
        )

4. 添加重试机制 #

修改upload_single_chunk方法，针对可重试错误添加重试逻辑：

@staticmethod
async def upload_single_chunk(
        session: aiohttp.ClientSession, url: str, chunk: bytes, headers: dict[str, str], attachment_name: str
) -> None:
    max_retries = 3
    retry_delay = 1
    for attempt in range(max_retries):
        try:
            async with session.put(
                    url,
                    data=chunk,
                    headers=headers,
                    timeout=aiohttp.ClientTimeout(total=CHUNK_UPLOAD_TIMEOUT_SECONDS),
            ) as response:
                if response.status in (412, 404) and attempt < max_retries - 1:
                    await asyncio.sleep(retry_delay * (2 ** attempt))
                    continue
                response.raise_for_status()
                return
        except aiohttp.ClientResponseError as e:
            if e.status in (412, 404) and attempt < max_retries - 1:
                await asyncio.sleep(retry_delay * (2 ** attempt))
                continue
            logger.warning(
                f"Failed to upload chunk after {max_retries} attempts: {e.status}, attachment '{attachment_name}'",
                exc_info=True,
            )
            raise

额外注意事项 #

• 调整信号量大小：建议先从2开始测试，逐步增加到4，避免超过Graph API的并发限制（Graph对同一资源的并发请求有隐性限制）。
• 日志细化：添加上传会话的expirationDateTime日志，避免使用过期的会话上传。

内容的提问来源于stack exchange，提问作者NikosDim