Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
13 changes: 13 additions & 0 deletions rust/src/engine-core-client/src/protocol/output.rs
Original file line number Diff line number Diff line change
Expand Up @@ -165,6 +165,19 @@ pub struct RequestError {
pub retryable: bool,
}

impl RequestError {
/// `code` of an error caused by the request itself, e.g. a structured-output
/// constraint the engine's backend cannot compile. Python raises
/// `VLLMValidationError` for it.
pub const VALIDATION_ERROR: &str = "validation_error";

/// Whether the request itself is invalid, so frontends report a client
/// error.
pub fn is_validation_error(&self) -> bool {
self.code == Self::VALIDATION_ERROR
}
}

/// Raw per-sequence speculative-decoding accumulator.
///
/// Python models this as a plain `@dataclass`, so it is serialized by msgspec
Expand Down
7 changes: 5 additions & 2 deletions rust/src/server/src/error.rs
Original file line number Diff line number Diff line change
Expand Up @@ -113,10 +113,13 @@ pub fn chat_submit_error(context: &'static str, error: vllm_chat::Error) -> ApiE
server_error!("{}: {}", context, error.to_report_string())
}

/// Classify an engine-reported request failure: a retryable error (e.g. a
/// paused engine's rejection) maps to HTTP 503, anything else to a 500.
/// Classify an engine-reported request failure: a validation error (e.g. a
/// constraint the structured-output backend cannot compile) maps to HTTP 400,
/// a retryable error (e.g. a paused engine's rejection) to 503, anything else
/// to a 500.
pub fn engine_error(error: Option<&RequestError>) -> ApiError {
match error {
Some(error) if error.is_validation_error() => invalid_request!("{}", error.message),
Some(error) if error.retryable => ApiError::ServiceUnavailable {
message: error.message.clone(),
},
Expand Down
18 changes: 17 additions & 1 deletion rust/src/server/src/grpc/convert.rs
Original file line number Diff line number Diff line change
Expand Up @@ -399,6 +399,9 @@ fn to_finish_info(finished: &Finished, token_ids: &[u32]) -> Result<pb::FinishIn
(PbFinishReason::Stop as i32, sr)
}
FinishReason::Length => (PbFinishReason::Length as i32, None),
FinishReason::Error(Some(error)) if error.is_validation_error() => {
return Err(Status::invalid_argument(error.message.clone()));
}
FinishReason::Error(Some(error)) if error.retryable => {
return Err(Status::unavailable(error.message.clone()));
}
Expand Down Expand Up @@ -1042,7 +1045,7 @@ mod tests {
}

#[test]
fn retryable_engine_error_maps_to_unavailable() {
fn engine_errors_map_to_status_codes() {
let error = RequestError {
code: "engine_paused".to_string(),
message: "Generation is paused.".to_string(),
Expand All @@ -1052,6 +1055,19 @@ mod tests {
let status = to_finish_info(&fin, &[]).expect_err("retryable error");
assert_eq!(status.code(), tonic::Code::Unavailable);

let error = RequestError {
code: RequestError::VALIDATION_ERROR.to_string(),
message: "Invalid structured outputs: Grammar error: bad".to_string(),
retryable: false,
};
let fin = finished(FinishReason::Error(Some(Box::new(error))));
let status = to_finish_info(&fin, &[]).expect_err("invalid request");
assert_eq!(status.code(), tonic::Code::InvalidArgument);
assert_eq!(
status.message(),
"Invalid structured outputs: Grammar error: bad"
);

let fin = finished(FinishReason::Error(None));
let status = to_finish_info(&fin, &[]).expect_err("internal error");
assert_eq!(status.code(), tonic::Code::Internal);
Expand Down
6 changes: 3 additions & 3 deletions rust/src/server/src/routes/inference/generate.rs
Original file line number Diff line number Diff line change
Expand Up @@ -298,9 +298,9 @@ fn collect_generate(
}: ResponseOptions,
mm_placeholders: Option<MultiModalPlaceholders>,
) -> Result<GenerateResponse, ApiError> {
if let FinishReason::Error(Some(error)) = &collected.finish_reason
&& error.retryable
{
// Like Python, which raises every engine-typed error, a typed error fails
// the request; an untyped one is reported as finish reason `error`.
if let FinishReason::Error(Some(error)) = &collected.finish_reason {
return Err(engine_error(Some(error)));
}
let logprobs = if include_logprobs {
Expand Down
139 changes: 79 additions & 60 deletions rust/src/server/src/routes/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4146,77 +4146,96 @@ async fn invalid_terminal_finish_reason_is_returned_as_openai_error_sse() {

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn retryable_engine_error_returns_service_unavailable() {
// A retryable engine error (e.g. a paused engine's rejection) is a 503 on
// every route when it arrives before the response starts, an SSE error after.
async fn engine_request_errors_map_to_http_status() {
// A retryable engine error (e.g. a paused engine's rejection) is a 503 and
// a validation error (e.g. a constraint the structured-output backend
// cannot compile) a 400 on every route when it arrives before the response
// starts, an SSE error after.
let model = "Qwen/Qwen1.5-0.5B-Chat";
let messages = json!([{"role": "user", "content": "hello"}]);
let raw = |stream: bool| {
json!({"model": model, "token_ids": [11, 22], "stream": stream,
"sampling_params": {"max_tokens": 2}})
};
let unavailable = StatusCode::SERVICE_UNAVAILABLE;
let cases = [
(
"/v1/completions",
json!({"model": model, "prompt": "hello"}),
unavailable,
),
let paused = RequestError {
code: "engine_paused".to_string(),
message: "Generation is paused.".to_string(),
retryable: true,
};
let invalid = RequestError {
code: RequestError::VALIDATION_ERROR.to_string(),
message: "Invalid structured outputs: Grammar error: bad".to_string(),
retryable: false,
};
for (error, status, error_type) in [
(
"/v1/chat/completions",
json!({"model": model, "messages": messages}),
unavailable,
paused,
StatusCode::SERVICE_UNAVAILABLE,
"ServiceUnavailableError",
),
("/inference/v1/generate", raw(false), unavailable),
("/inference/v1/generate", raw(true), StatusCode::OK),
];
for (uri, request, status) in cases {
let (chat, engine_task) = test_models_with_engine_outputs_and_backend_inner(
b"engine-openai",
|request_id| {
RequestBatchOutputs {
outputs: vec![EngineCoreOutput {
request_id: request_id.to_string(),
finish_reason: Some(EngineCoreFinishReason::Error),
error: Some(Box::new(RequestError {
code: "engine_paused".to_string(),
message: "Generation is paused.".to_string(),
retryable: true,
})),
(invalid, StatusCode::BAD_REQUEST, "BadRequestError"),
] {
let cases = [
(
"/v1/completions",
json!({"model": model, "prompt": "hello"}),
status,
),
(
"/v1/chat/completions",
json!({"model": model, "messages": messages}),
status,
),
("/inference/v1/generate", raw(false), status),
("/inference/v1/generate", raw(true), StatusCode::OK),
];
for (uri, request, expected_status) in cases {
let engine_error = error.clone();
let (chat, engine_task) = test_models_with_engine_outputs_and_backend_inner(
b"engine-openai",
move |request_id| {
RequestBatchOutputs {
outputs: vec![EngineCoreOutput {
request_id: request_id.to_string(),
finish_reason: Some(EngineCoreFinishReason::Error),
error: Some(Box::new(engine_error.clone())),
..Default::default()
}],
..Default::default()
}],
..Default::default()
}
.into()
},
|_| {},
Arc::new(FakeChatBackend::new()),
)
.await;
let mut app = build_router_with_scale_out_endpoints(
Arc::new(AppState::new(vec![model.to_string()], chat)),
true,
);
let response = app
.call(
Request::builder()
.method("POST")
.uri(uri)
.header("content-type", "application/json")
.body(Body::from(request.to_string()))
.expect("build request"),
}
.into()
},
|_| {},
Arc::new(FakeChatBackend::new()),
)
.await
.expect("call app");
.await;
let mut app = build_router_with_scale_out_endpoints(
Arc::new(AppState::new(vec![model.to_string()], chat)),
true,
);
let response = app
.call(
Request::builder()
.method("POST")
.uri(uri)
.header("content-type", "application/json")
.body(Body::from(request.to_string()))
.expect("build request"),
)
.await
.expect("call app");

assert_eq!(response.status(), status, "{uri} {request}");
let body = to_bytes(response.into_body(), usize::MAX).await.expect("read body");
engine_task.await.expect("mock engine task");
let text = String::from_utf8(body.to_vec()).expect("utf8 body");
assert!(
text.contains("\"type\":\"ServiceUnavailableError\"") && text.contains("\"code\":503"),
"{uri} {request}: {text}"
);
assert_eq!(response.status(), expected_status, "{uri} {request}");
let body = to_bytes(response.into_body(), usize::MAX).await.expect("read body");
engine_task.await.expect("mock engine task");
let text = String::from_utf8(body.to_vec()).expect("utf8 body");
assert!(
text.contains(&format!("\"type\":\"{error_type}\""))
&& text.contains(&format!("\"code\":{}", status.as_u16()))
&& text.contains(&error.message),
"{uri} {request}: {text}"
);
}
}
}

Expand Down
2 changes: 1 addition & 1 deletion tests/v1/core/test_async_scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -316,7 +316,7 @@ def test_abort_request_when_structured_output_fsm_cannot_advance():
scheduler.kv_event_publisher = Mock()
scheduler.finished_req_ids = set()
scheduler.finished_req_ids_dict = None
scheduler.grammar_compile_error_reqs = set()
scheduler.grammar_compile_error_reqs = {}
scheduler.encoder_cache_mismatch_reqs = set()
scheduler.vllm_config = Mock()
scheduler.aux_output_connector = None
Expand Down
7 changes: 6 additions & 1 deletion tests/v1/core/test_scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
MultiModalKwargsItem,
PlaceholderRange,
)
from vllm.outputs import RequestError
from vllm.pooling_params import PoolingParams
from vllm.sampling_params import SamplingParams, StructuredOutputsParams
from vllm.utils.hashing import sha256
Expand Down Expand Up @@ -4539,6 +4540,10 @@ def test_grammar_compile_error_finishes_only_request(async_grammar: bool):
assert output.request_id == request.request_id
assert output.finish_reason == FinishReason.ERROR
assert output.stop_reason is None
assert output.error == RequestError(
code="validation_error",
message="Invalid structured outputs: forced FSM compilation error",
)

healthy_request = create_requests(num_requests=1, req_ids=["healthy-request"])[0]
scheduler.add_request(healthy_request)
Expand Down Expand Up @@ -4580,7 +4585,7 @@ def test_abort_request_when_structured_output_fsm_cannot_advance():
scheduler.finished_req_ids = set()
scheduler.finished_req_ids_dict = None
scheduler.aux_output_connector = None
scheduler.grammar_compile_error_reqs = set()
scheduler.grammar_compile_error_reqs = {}
scheduler.encoder_cache_mismatch_reqs = set()
scheduler.vllm_config = Mock()
scheduler.return_sampling_mask = False
Expand Down
19 changes: 19 additions & 0 deletions tests/v1/structured_output/test_backend_guidance.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,25 @@ def mistral_tokenizer():
)


def test_backend_guidance_rejects_grammar_matcher_cannot_build():
# llguidance builds a matcher for a schema with unimplemented keywords
# but leaves it in an error state; compilation must fail the request.
vllm_config = VllmConfig(
structured_outputs_config=StructuredOutputsConfig(backend="guidance")
)
backend = GuidanceBackend(
vllm_config,
tokenizer=AutoTokenizer.from_pretrained(TOKENIZER),
vocab_size=50257,
)

with pytest.raises(ValueError, match="Grammar error: .*uniqueItems"):
backend.compile_grammar(
StructuredOutputOptions.JSON,
json.dumps({"type": "array", "uniqueItems": True}),
)


def test_backend_guidance_rollback_terminated():
# Test that the backend guidance successfully rollbacks from a
# terminated state. This can happen with speculative decoding,
Expand Down
14 changes: 10 additions & 4 deletions vllm/v1/core/sched/scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
from vllm.multimodal import MULTIMODAL_REGISTRY, MultiModalRegistry
from vllm.multimodal.encoder_budget import MultiModalBudget
from vllm.multimodal.utils import get_mm_features_in_window
from vllm.outputs import RequestError
from vllm.utils.math_utils import cdiv
from vllm.v1.core.encoder_cache_manager import (
EncoderCacheManager,
Expand Down Expand Up @@ -249,7 +250,7 @@ def __init__(

# Grammar compilation failures to finish as per-request errors in
# update_from_output.
self.grammar_compile_error_reqs: set[str] = set()
self.grammar_compile_error_reqs: dict[str, RequestError] = {}
self.encoder_cache_mismatch_reqs: set[str] = set()

# Encoder-related.
Expand Down Expand Up @@ -2389,8 +2390,9 @@ def update_from_output(
for request in continuing_reqs:
self._enqueue_waiting_request(request)

error_req_ids = set(self.grammar_compile_error_reqs)
self.grammar_compile_error_reqs.clear()
request_errors = self.grammar_compile_error_reqs
self.grammar_compile_error_reqs = {}
error_req_ids = set(request_errors)
error_req_ids.update(self.encoder_cache_mismatch_reqs)
self.encoder_cache_mismatch_reqs.clear()
if failed_kv_load_req_ids and not self.recompute_kv_load_failures:
Expand All @@ -2412,6 +2414,7 @@ def update_from_output(
finish_reason=request.get_finished_reason(),
events=request.take_events(),
trace_headers=request.trace_headers,
error=request_errors.get(request.request_id),
)
)

Expand Down Expand Up @@ -3285,7 +3288,10 @@ def _handle_blocked_waiting_request(self, request: Request) -> bool:
if not struct_output_req or struct_output_req.grammar is None:
return False
if isinstance(struct_output_req.grammar, Exception):
self.grammar_compile_error_reqs.add(request.request_id)
self.grammar_compile_error_reqs[request.request_id] = RequestError(
code="validation_error",
message=f"Invalid structured outputs: {struct_output_req.grammar}",
)
return False
# Grammar is ready, promote to waiting state.
request.status = RequestStatus.WAITING
Expand Down
8 changes: 4 additions & 4 deletions vllm/v1/structured_output/backend_guidance.py
Original file line number Diff line number Diff line change
Expand Up @@ -196,15 +196,15 @@ def compile_grammar(
log_level=int(os.environ.get("LLGUIDANCE_LOG_LEVEL", "1")),
)

r = GuidanceGrammar(
if err := ll_matcher.get_error():
raise ValueError(f"Grammar error: {err}")

return GuidanceGrammar(
ll_matcher=ll_matcher,
ll_tokenizer=self.ll_tokenizer,
vocab_size=self.vocab_size,
)

r.check_error()
return r

def allocate_token_bitmask(self, max_num_seqs: int):
return llguidance_torch.allocate_token_bitmask(
max_num_seqs, self.ll_tokenizer.vocab_size
Expand Down
Loading