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
10 changes: 6 additions & 4 deletions tests/e2e/batches/COVERAGE.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,9 @@ failures are hard test failures (see `tests/e2e/AGENTS.md`).
| Bedrock | yes (unified only) | yes | yes | yes (unfiltered managed list) | yes (provider-transformed) | S3 (`s3_bucket_name` + `aws_*` + `AWS_BATCH_ROLE_ARN` on model) |
| Bedrock GovCloud (`us-gov-west-1`) | yes (unified only) | yes | no | no | yes (provider-transformed) | S3 (`s3_bucket_name` + `aws_*` on model, resolved from `AWS_GOVCLOUD_ACCESS_KEY_ID` / `AWS_GOVCLOUD_SECRET_ACCESS_KEY` / `AWS_GOVCLOUD_BATCH_S3_BUCKET` / `AWS_GOVCLOUD_BATCH_ROLE_ARN`) |
| Bedrock split S3 identity | no | no | no | no | yes (file upload, content, delete) | S3 signed with `s3_access_key_id` / `s3_secret_access_key` (`AWS_S3_ONLY_ACCESS_KEY_ID` / `AWS_S3_ONLY_SECRET_ACCESS_KEY`, object rights on `AWS_BATCH_S3_BUCKET` only) while `aws_*` is `AWS_BEDROCK_ONLY_ACCESS_KEY_ID` / `AWS_BEDROCK_ONLY_SECRET_ACCESS_KEY`, an identity with no S3 rights on that bucket |
| Bedrock blank S3 env | yes (unified only, on an owned gateway exporting `AWS_S3_ENCRYPTION_KEY_ID` / `AWS_S3_BUCKET_OWNER` as empty strings) | no | no | no | no | S3 (`s3_bucket_name` + `aws_*` + `AWS_BATCH_ROLE_ARN` in the gateway config); blank env vars must be treated as unset, not serialized |

Bedrock cancel maps to `StopModelInvocationJob` and comes back `cancelling`; the
lifecycle asserts it the same way it does for OpenAI (`_CANCEL_ASSERTED_PROVIDERS`).
lifecycle asserts it the same way it does for OpenAI and Azure (`_CANCEL_ASSERTED_PROVIDERS`).
Bedrock has no provider-side list, so list is the proxy's DB-backed managed view: the
unified lifecycle lists with the plain `GET /v1/batches` and the batch must appear
there. Both were gated off until LIT-5730, after LIT-4774 landed cancel support. A batch that completes inside the 2 s pre-cancel window skips the cancel assertion (a documented vacuous pass for the cancel cell, same as OpenAI); the list assertion runs either way.
Expand Down Expand Up @@ -132,8 +131,11 @@ provider when deleted. Model-encoded and managed file IDs route themselves
File deletion and batch cancellation check their responses and retry transient
failures up to three times. Teardown attempts every registered cleanup before
reporting failures as test errors. Already deleted files and batches that are
terminal are safe to clean up again. Managed batch cancellation polls for up to eleven minutes
before input deletion: the ten-minute provider window plus a propagation margin.
terminal are safe to clean up again. Managed batch cancellation polls for up to two minutes
before input deletion. A managed batch still `cancelling` after that is left for the provider to
finish, and its input file is left in place because LiteLLM refuses to delete a file a non-terminal
batch references. Both are reported as `BatchCleanupLeftover` warnings naming their ids rather than
failing the test. Any other status or error still fails
Accepted cancellation may still report validating or in_progress while the provider
updates its state. Raw and model-encoded batches are polled until cancelling or
terminal before input deletion. OpenAI and Azure lifecycle cleanup also deletes
Expand Down
30 changes: 25 additions & 5 deletions tests/e2e/batches/batch_cleanup.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import warnings
from builtins import ExceptionGroup
from collections.abc import Callable
from itertools import count
Expand All @@ -12,8 +13,9 @@
CLEANUP_DELAYS: Final = (1.0, 2.0, 4.0)
BATCH_TERMINAL_STATUSES: Final = frozenset({"completed", "failed", "expired", "cancelled"})
BATCH_PENDING_STATUSES: Final = frozenset({"validating", "in_progress", "finalizing", "cancelling"})
BATCH_CANCEL_TIMEOUT_SECONDS: Final = 660.0
BATCH_CANCEL_TIMEOUT_SECONDS: Final = 120.0
BATCH_CANCEL_POLL_SECONDS: Final = 10.0
FILE_IN_USE_REFUSAL: Final = "batch(es) in non-terminal state"


class BatchCleanupClient(Protocol):
Expand All @@ -26,6 +28,10 @@ def retrieve_batch(self, batch_id: str, *, key: str, provider: str | None = None
def cancel_batch(self, batch_id: str, *, key: str, provider: str | None = None) -> Result[BatchObject]: ...


class BatchCleanupLeftover(UserWarning):
pass


def cleanup_result[R: BaseModel](
action: Callable[[], Result[R]], *, wait: Callable[[float], None] = sleep
) -> Result[R]:
Expand Down Expand Up @@ -59,6 +65,13 @@ def cleanup_file(client: BatchCleanupClient, file_id: str, *, key: str, provider
result: Final = cleanup_result(delete)
if isinstance(result, UnknownApiError) and result.status_code == 404:
return
if isinstance(result, UnknownApiError) and result.status_code == 400 and FILE_IN_USE_REFUSAL in result.body:
warnings.warn(
f"Left file {file_id} in place: LiteLLM refused to delete it while a batch still references it",
BatchCleanupLeftover,
stacklevel=2,
)
return
deleted: Final = _require_cleanup_success(result, f"Delete file {file_id}")
assert deleted.deleted is True or (
deleted.deleted is None and is_managed_id(file_id) and deleted.id == file_id and deleted.object == "file"
Expand Down Expand Up @@ -120,10 +133,17 @@ def cleanup_batch(
)
if current.status == "cancelling" and not needs_terminal_state:
return
assert clock() < deadline, (
f"Batch {batch_id} cancellation did not finish within {BATCH_CANCEL_TIMEOUT_SECONDS}s, "
f"last status {current.status}"
)
if clock() >= deadline:
assert current.status == "cancelling", (
f"Batch {batch_id} cancellation did not finish within {BATCH_CANCEL_TIMEOUT_SECONDS}s, "
f"last status {current.status}"
)
warnings.warn(
f"Left batch {batch_id} cancelling after {BATCH_CANCEL_TIMEOUT_SECONDS}s for the provider to finish",
BatchCleanupLeftover,
stacklevel=2,
)
return
Comment on lines +141 to +146

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Slow cancellations leave resources behind

When a managed batch is still cancelling after two minutes, teardown warns and stops trying to clean it up. Deleting its input file can also return with only a warning. Repeated live Azure runs can therefore leave batches and files behind with no later cleanup attempt. Please add a way to clean up these leftovers after cancellation finishes.

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

wait(BATCH_CANCEL_POLL_SECONDS)


Expand Down
151 changes: 0 additions & 151 deletions tests/e2e/batches/bedrock_env_gateway.py

This file was deleted.

71 changes: 63 additions & 8 deletions tests/e2e/batches/test_batch_cleanup.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,14 @@
from unittest.mock import Mock, call

import pytest
from batch_cleanup import BATCH_CANCEL_TIMEOUT_SECONDS, CLEANUP_DELAYS, cleanup_batch, cleanup_file, cleanup_result
from batch_cleanup import (
BATCH_CANCEL_TIMEOUT_SECONDS,
CLEANUP_DELAYS,
BatchCleanupLeftover,
cleanup_batch,
cleanup_file,
cleanup_result,
)
from batch_client import AZURE_FILE_EXPIRY_SECONDS, BatchObject, FileDeleteResponse, batch_upload_form
from capabilities import CAPABILITIES, Capability
from e2e_http import NetworkError, RateLimitedError, Result, Success, UnknownApiError
Expand All @@ -13,6 +20,10 @@

MANAGED_FILE_ID: Final = "bGl0ZWxsbV9wcm94eTtmaWxlLTE="
MANAGED_BATCH_ID: Final = "bGl0ZWxsbV9wcm94eTtiYXRjaC0x"
IN_USE_REFUSAL: Final = (
f'{{"error":{{"message":"Cannot delete file {MANAGED_FILE_ID}. The file is referenced by 1 batch(es) in '
f'non-terminal state: {MANAGED_BATCH_ID}: cancelling. ","type":"invalid_request_error","code":"400"}}}}'
)


class ExpectedCalls[T]:
Expand Down Expand Up @@ -125,6 +136,29 @@ def test_success_response_must_confirm_deletion(self) -> None:
cleanup_file(client, "file-1", key="test-key")
client.calls.assert_done()

def test_delete_refused_because_a_batch_still_references_the_file_is_left_and_reported(self) -> None:
client: Final = CleanupClient(
calls=ExpectedCalls((f"delete None {MANAGED_FILE_ID}",)),
files=(UnknownApiError(status_code=400, body=IN_USE_REFUSAL),),
)
with pytest.warns(BatchCleanupLeftover, match=MANAGED_FILE_ID):
cleanup_file(client, MANAGED_FILE_ID, key="test-key")
client.calls.assert_done()

@pytest.mark.parametrize(
"failure",
[
UnknownApiError(status_code=400, body="Invalid file id"),
UnknownApiError(status_code=409, body=IN_USE_REFUSAL),
UnknownApiError(status_code=501, body=IN_USE_REFUSAL),
],
)
def test_any_other_delete_failure_still_raises(self, failure: UnknownApiError) -> None:
client: Final = CleanupClient(calls=ExpectedCalls((f"delete None {MANAGED_FILE_ID}",)), files=(failure,))
with pytest.raises(AssertionError, match=f"Delete file {MANAGED_FILE_ID} failed: HTTP {failure.status_code}"):
cleanup_file(client, MANAGED_FILE_ID, key="test-key")
client.calls.assert_done()

def test_cleanup_is_idempotent_when_file_is_already_deleted(self) -> None:
client: Final = CleanupClient(
calls=ExpectedCalls(("delete azure file-1",)),
Expand Down Expand Up @@ -188,29 +222,50 @@ def test_cancelling_batch_is_polled_until_terminal_without_cancelling_again(self
client.calls.assert_done()
delays.assert_done()

def test_cancellation_timeout_is_reported_but_file_and_key_cleanup_still_run(self) -> None:
def test_batch_still_cancelling_at_the_deadline_and_its_input_file_are_left_and_reported(self) -> None:
client: Final = CleanupClient(
calls=ExpectedCalls(
(
f"retrieve None {MANAGED_BATCH_ID}",
f"retrieve None {MANAGED_BATCH_ID}",
"delete None file-1",
f"delete None {MANAGED_FILE_ID}",
"delete key test-key",
)
),
batches=(batch("cancelling"), batch("cancelling")),
files=(deleted_file(),),
files=(UnknownApiError(status_code=400, body=IN_USE_REFUSAL),),
)
times: Final = (0.0, BATCH_CANCEL_TIMEOUT_SECONDS)
ticks: Final[Callable[[], float]] = Mock(side_effect=times)
manager: Final = ResourceManager(client=client, strict_cleanup=True)
key: Final = manager.key()
manager.defer(lambda: cleanup_file(client, "file-1", key=key))
manager.defer(lambda: cleanup_file(client, MANAGED_FILE_ID, key=key))
manager.defer(lambda: cleanup_batch(client, MANAGED_BATCH_ID, key=key, clock=ticks))
with pytest.raises(ExceptionGroup) as caught:
with pytest.warns(BatchCleanupLeftover) as leftovers:
manager.teardown()
assert "cancellation did not finish" in str(caught.value.exceptions[0])
assert "last status cancelling" in str(caught.value.exceptions[0])
client.calls.assert_done()
messages: Final = tuple(str(warning.message) for warning in leftovers)
assert len(messages) == 2
assert MANAGED_BATCH_ID in messages[0] and "cancelling" in messages[0]
assert MANAGED_FILE_ID in messages[1]

@pytest.mark.parametrize(
"last, reported",
[
(batch("in_progress"), f"did not finish within {BATCH_CANCEL_TIMEOUT_SECONDS}s, last status in_progress"),
(UnknownApiError(status_code=403, body="forbidden"), "after cancellation failed: HTTP 403"),
],
)
def test_anything_but_still_cancelling_at_the_deadline_still_fails(
self, last: Result[BatchObject], reported: str
) -> None:
client: Final = CleanupClient(
calls=ExpectedCalls((f"retrieve None {MANAGED_BATCH_ID}",) * 2), batches=(batch("cancelling"), last)
)
times: Final = (0.0, BATCH_CANCEL_TIMEOUT_SECONDS)
ticks: Final[Callable[[], float]] = Mock(side_effect=times)
with pytest.raises(AssertionError, match=reported):
cleanup_batch(client, MANAGED_BATCH_ID, key="test-key", clock=ticks)
client.calls.assert_done()

@pytest.mark.parametrize("status", ["completed", "failed", "expired", "cancelled"])
Expand Down
6 changes: 3 additions & 3 deletions tests/e2e/batches/test_batches_e2e.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,11 +94,11 @@ class _GovCloudBedrockRecord(BaseModel):
model_input: _GovCloudBedrockInput = Field(alias="modelInput")


# Azure / Vertex cancel and the pre-cancel re-retrieve are provider-side flakes
# Vertex cancel and the pre-cancel re-retrieve are provider-side flakes
# (connection refused, brief 500s) and the registry only has one basic cell per
# provider (shared across scenarios). Create + retrieve already prove routing;
# cancel is still deferred for cleanup, just not asserted for these two.
_CANCEL_ASSERTED_PROVIDERS = frozenset({"openai", "bedrock"})
# cancel is still deferred for cleanup, just not asserted for Vertex.
_CANCEL_ASSERTED_PROVIDERS = frozenset({"openai", "azure", "bedrock"})


def _transient_status(status_code: int) -> bool:
Expand Down
Loading
Loading