Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion swanlab/sdk/internal/core_python/transport/dispatch.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,8 @@ def _upload_chunk(self, record_type: str, chunk: Sequence[Record]) -> bool:
"""单 chunk 上传。返回 True 表示成功,False 表示失败。"""
if self._sender is None:
raise RuntimeError("sender not set")
with safe.block(message=f"record chunk upload failed, record_type={record_type!r}"):
with safe.block(message=f"record chunk upload failed, record_type={record_type!r}", write_to_tty=False):
# ⚠️:在持续断网的状态下,上传失败可能会重复打印,破坏掉 transport 层的 UploadWarningThrottle
self._sender.upload(record_type, chunk)
return True
return False
Expand Down
5 changes: 4 additions & 1 deletion swanlab/sdk/internal/core_python/transport/sender.py
Original file line number Diff line number Diff line change
Expand Up @@ -138,9 +138,12 @@ def upload(self, record_type: str, records: Sequence[Record]) -> None:
handler(records)
except ApiError as e:
if e.response.status_code >= 500:
# 基础设施错误,不按业务逻辑处理,此时直接抛出异常,交给上游重试
# 基础设施错误(断网/5xx),向上抛出交由 transport 重试;raise 路径不会执行 advance
raise
Comment thread
Nexisato marked this conversation as resolved.
# 4xx:后端业务拒绝,重试也不会成功;不计入 uploaded,进度条原地不动以如实反映失败
console.warning(f"Failed to upload {record_type} records, skipping; error: {e}")
return
# 仅当本批真正上传成功后才递进 uploaded(网络失败会 raise,走不到这里)
if self._tracker is not None:
self._tracker.advance_records(len(records))

Expand Down
17 changes: 15 additions & 2 deletions tests/unit/sdk/internal/core_python/transport/test_sender.py
Original file line number Diff line number Diff line change
Expand Up @@ -339,7 +339,20 @@ def _fake_upload_resource(_, __, *, paths, buffers, content_types=None, tracker=
assert list(snapshots[1].files) == []


def test_upload_advances_records_for_non_retryable_api_error(tmp_path: Path):
def test_upload_advances_records_only_on_success(tmp_path: Path):
"""成功上传才递进 uploaded,进度条仅保留上传成功的进度。"""
tracker = UploadTracker()
sender = _make_sender(tmp_path)
sender.set_tracker(tracker)
sender._upload_handlers["scalar"] = MagicMock()

sender.upload("scalar", [Record(), Record()])

assert tracker.snapshot().uploaded_records == 2


def test_upload_does_not_advance_records_for_non_retryable_api_error(tmp_path: Path):
"""4xx 业务拒绝不计入 uploaded:失败原地不动,进度条不递进(也不抛异常,避免上层无意义重试)。"""
tracker = UploadTracker()
sender = _make_sender(tmp_path)
sender.set_tracker(tracker)
Expand All @@ -355,7 +368,7 @@ def test_upload_advances_records_for_non_retryable_api_error(tmp_path: Path):

sender.upload("scalar", [Record()])

assert tracker.snapshot().uploaded_records == 1
assert tracker.snapshot().uploaded_records == 0


def test_upload_does_not_advance_records_for_retryable_api_error(tmp_path: Path):
Expand Down
Loading