From 9af16805775d65455b0d6d0f44e433fc2b311cf7 Mon Sep 17 00:00:00 2001 From: Claudio Maradonna Date: Tue, 1 Sep 2026 15:36:03 +0200 Subject: [PATCH] fix(ds4): cancel abandoned inference Propagate gRPC cancellation into DS4 prompt synchronization and poll it at decode boundaries. Stop on failed stream writes and skip parser finalization and KV persistence for abandoned partial requests. Assisted-by: Codex:gpt-5.6-sol Signed-off-by: Claudio Maradonna --- backend/cpp/ds4/grpc-server.cpp | 206 +++++++--- backend/cpp/ds4/request_lifecycle.h | 111 ++++++ backend/cpp/ds4/request_lifecycle_test.cpp | 414 +++++++++++++++++++++ docs/content/features/backends.md | 9 + 4 files changed, 692 insertions(+), 48 deletions(-) create mode 100644 backend/cpp/ds4/request_lifecycle.h create mode 100644 backend/cpp/ds4/request_lifecycle_test.cpp diff --git a/backend/cpp/ds4/grpc-server.cpp b/backend/cpp/ds4/grpc-server.cpp index 68ebdd3e3551..d00f6ba3f61e 100644 --- a/backend/cpp/ds4/grpc-server.cpp +++ b/backend/cpp/ds4/grpc-server.cpp @@ -12,6 +12,7 @@ #include "dsml_renderer.h" // populated in Task 16 #include "generation_limits.h" #include "kv_cache.h" // populated in Task 17 +#include "request_lifecycle.h" extern "C" { #include "ds4.h" @@ -36,6 +37,7 @@ extern "C" { #include #include #include +#include #include using grpc::Server; @@ -70,6 +72,21 @@ int g_route_timeout_sec = 60; std::atomic g_server{nullptr}; +static bool server_context_cancelled(void *ud) { + return static_cast(ud)->IsCancelled(); +} + +static void set_session_cancel(void *target, ds4cpp::CancelCallback callback, + void *userdata) noexcept { + ds4_session_set_cancel(static_cast(target), callback, userdata); +} + +static bool request_should_continue(ds4cpp::RequestLifecycle *request, + ServerContext *context) { + request->ObserveContextCancellation(context->IsCancelled()); + return request->ShouldContinue(); +} + // Parse a "key:value" option string. Returns empty when no colon. static std::pair split_option(const std::string &opt) { auto colon = opt.find(':'); @@ -239,37 +256,58 @@ static bool apply_engine_option(ds4_engine_options *opt, const std::string &key, // When acting as a distributed coordinator, block until the worker route // covers all layers (ds4_session_distributed_route_ready == 1) or the timeout -// elapses. Returns an empty string on success, or an error message to return -// to the client. No-op when not distributed. +// elapses. No-op when not distributed. // // Takes the g_engine_mu lock by reference and RELEASES it during each poll // sleep. The wait can span up to g_route_timeout_sec seconds while workers // connect; holding g_engine_mu the whole time would block the Status/Health // readiness probes (they also lock g_engine_mu), making LocalAI's loader treat // a still-starting worker as hung. -static std::string wait_route_ready(std::unique_lock &lock) { - if (!g_distributed) return ""; +struct RouteWaitResult { + ds4cpp::RouteWaitDecision decision; + std::string error; +}; + +static RouteWaitResult wait_route_ready(std::unique_lock &lock, + ServerContext *context) { + if (!g_distributed) return {ds4cpp::RouteWaitDecision::Ready, ""}; char err[256] = {0}; const int deadline_polls = g_route_timeout_sec * 10; // 100ms per poll for (int i = 0; i <= deadline_polls; ++i) { int ready = ds4_session_distributed_route_ready(g_session, err, sizeof(err)); - if (ready == 1) return ""; - if (ready < 0) { - return std::string("ds4 distributed route error: ") + - (err[0] ? err : "unknown"); + switch (ds4cpp::DecideRouteWait(ready, context->IsCancelled())) { + case ds4cpp::RouteWaitDecision::Ready: + return {ds4cpp::RouteWaitDecision::Ready, ""}; + case ds4cpp::RouteWaitDecision::Error: + return {ds4cpp::RouteWaitDecision::Error, + std::string("ds4 distributed route error: ") + + (err[0] ? err : "unknown")}; + case ds4cpp::RouteWaitDecision::Cancelled: + return {ds4cpp::RouteWaitDecision::Cancelled, ""}; + case ds4cpp::RouteWaitDecision::Pending: + break; } + if (i == deadline_polls) break; // Release the lock while sleeping so Status/Health and other RPCs can // interleave during worker startup. lock.unlock(); struct timespec ts = {0, 100L * 1000L * 1000L}; // 100ms nanosleep(&ts, nullptr); lock.lock(); + if (context->IsCancelled()) { + return {ds4cpp::RouteWaitDecision::Cancelled, ""}; + } // A concurrent Free() may have torn down the engine while we slept. if (!g_engine || !g_session) { - return "ds4: model unloaded while waiting for distributed route"; + return {ds4cpp::RouteWaitDecision::Error, + "ds4: model unloaded while waiting for distributed route"}; } } - return "ds4 distributed route incomplete: workers not connected (layers uncovered)"; + if (context->IsCancelled()) { + return {ds4cpp::RouteWaitDecision::Cancelled, ""}; + } + return {ds4cpp::RouteWaitDecision::Error, + "ds4 distributed route incomplete: workers not connected (layers uncovered)"}; } static void append_token_text(ds4_engine *engine, int token, std::string &out) { @@ -342,9 +380,9 @@ static void collect_done(void *) {} struct StreamCtx { ds4_engine *engine; ServerWriter *writer; + ds4cpp::RequestLifecycle *request; ds4cpp::DsmlParser parser; int tokens; - bool aborted; // Track which tool indices we've seen TOOL_START for, so subsequent // ARGS deltas can elide the redundant id/name fields. std::vector tool_started; @@ -352,7 +390,7 @@ struct StreamCtx { static void stream_emit(void *ud, int token) { auto *s = static_cast(ud); - if (s->aborted) return; + if (!s->request->ShouldContinue()) return; if (token == ds4_token_eos(s->engine)) return; size_t len = 0; const char *text = ds4_token_text(s->engine, token, &len); @@ -402,7 +440,7 @@ static void stream_emit(void *ud, int token) { reply.set_message(chunk); reply.set_tokens(1); if (any_field) { - if (!s->writer->Write(reply)) s->aborted = true; + s->request->ObserveStreamWrite(s->writer->Write(reply)); } s->tokens++; } @@ -758,15 +796,19 @@ class DS4Backend final : public backend::Backend::Service { return GStatus::OK; } - GStatus Predict(ServerContext *, const backend::PredictOptions *request, + GStatus Predict(ServerContext *context, const backend::PredictOptions *request, backend::Reply *reply) override { std::unique_lock lock(g_engine_mu); if (!g_engine || !g_session) { return GStatus(StatusCode::FAILED_PRECONDITION, "ds4: model not loaded"); } if (GStatus id = check_model_identity(request); !id.ok()) return id; - if (std::string route_err = wait_route_ready(lock); !route_err.empty()) { - return GStatus(StatusCode::UNAVAILABLE, route_err); + RouteWaitResult route = wait_route_ready(lock, context); + if (route.decision == ds4cpp::RouteWaitDecision::Cancelled) { + return GStatus(StatusCode::CANCELLED, "ds4 request cancelled"); + } + if (route.decision == ds4cpp::RouteWaitDecision::Error) { + return GStatus(StatusCode::UNAVAILABLE, route.error); } ds4_tokens prompt = {}; build_prompt(g_engine, request, &prompt); @@ -777,6 +819,7 @@ class DS4Backend final : public backend::Backend::Service { CollectCtx collect = { g_engine, "", ds4cpp::DsmlParser(starts_in_thinking), reply, 0, {}, "", ""}; + ds4cpp::RequestLifecycle lifecycle; std::string cache_key = render_prompt_text(request); size_t cache_hit = maybe_load_cache(cache_key); (void)cache_hit; // future: skip prompt prefix if hit covers full prompt @@ -788,10 +831,19 @@ class DS4Backend final : public backend::Backend::Service { // Either way g_session advances so the disk KV cache picks up a // real checkpoint after the call (see maybe_save_cache below). char err[256] = {0}; - int rc = ds4_session_sync(g_session, &prompt, err, sizeof(err)); + int rc; + { + ds4cpp::CancelCallbackScope cancel_scope( + g_session, set_session_cancel, server_context_cancelled, context); + rc = ds4_session_sync(g_session, &prompt, err, sizeof(err)); + } int prompt_len = prompt.len; ds4_tokens_free(&prompt); - if (rc == 0) { + if (rc == DS4_SESSION_SYNC_INTERRUPTED) { + lifecycle.ObserveContextCancellation(true); + } + const bool generation_started = rc == 0; + if (generation_started) { const int n_predict = ds4cpp::EffectiveGenerationLimit( request->tokens(), ds4_session_ctx(g_session), ds4_session_pos(g_session)); @@ -799,6 +851,7 @@ class DS4Backend final : public backend::Backend::Service { const int draft_max = ds4_engine_mtp_draft_tokens(g_engine); int produced = 0; while (produced < n_predict) { + if (!request_should_continue(&lifecycle, context)) break; SampleParams sp = compute_sample_params(request, collect.parser, think_enabled); int first; if (sp.temperature <= 0.0f) { @@ -823,6 +876,10 @@ class DS4Backend final : public backend::Backend::Service { if (n < 0) { rc = -1; break; } bool stop = false; for (int j = 0; j < n; ++j) { + if (!request_should_continue(&lifecycle, context)) { + stop = true; + break; + } if (accepted[j] == eos) { stop = true; break; } collect_emit(&collect, accepted[j]); if (++produced >= n_predict) { stop = true; break; } @@ -831,12 +888,26 @@ class DS4Backend final : public backend::Backend::Service { } else { collect_emit(&collect, first); if (++produced >= n_predict) break; + if (!request_should_continue(&lifecycle, context)) break; rc = ds4_session_eval(g_session, first, err, sizeof(err)); if (rc != 0) break; } } - collect_done(&collect); } + + request_should_continue(&lifecycle, context); + ds4cpp::TerminalDecision terminal = ds4cpp::ResolveTerminalDecision( + rc == DS4_SESSION_SYNC_INTERRUPTED, rc != 0, + !lifecycle.ShouldFinalize()); + if (!terminal.should_finalize) { + if (terminal.cause == ds4cpp::TerminalCause::EngineError) { + return GStatus(StatusCode::INTERNAL, + std::string("ds4 generation failed: ") + err); + } + return GStatus(StatusCode::CANCELLED, + "ds4 request cancelled"); + } + if (generation_started) collect_done(&collect); maybe_save_cache(cache_key); // Flush any buffered parser state. @@ -844,7 +915,7 @@ class DS4Backend final : public backend::Backend::Service { collect.parser.Flush(events); apply_events(&collect, events); - if (rc != 0) { + if (terminal.cause == ds4cpp::TerminalCause::EngineError) { return GStatus(StatusCode::INTERNAL, std::string("ds4 generation failed: ") + err); } @@ -867,15 +938,19 @@ class DS4Backend final : public backend::Backend::Service { return GStatus::OK; } - GStatus PredictStream(ServerContext *, const backend::PredictOptions *request, + GStatus PredictStream(ServerContext *context, const backend::PredictOptions *request, ServerWriter *writer) override { std::unique_lock lock(g_engine_mu); if (!g_engine || !g_session) { return GStatus(StatusCode::FAILED_PRECONDITION, "ds4: model not loaded"); } if (GStatus id = check_model_identity(request); !id.ok()) return id; - if (std::string route_err = wait_route_ready(lock); !route_err.empty()) { - return GStatus(StatusCode::UNAVAILABLE, route_err); + RouteWaitResult route = wait_route_ready(lock, context); + if (route.decision == ds4cpp::RouteWaitDecision::Cancelled) { + return GStatus(StatusCode::CANCELLED, "ds4 request cancelled"); + } + if (route.decision == ds4cpp::RouteWaitDecision::Error) { + return GStatus(StatusCode::UNAVAILABLE, route.error); } ds4_tokens prompt = {}; build_prompt(g_engine, request, &prompt); @@ -883,9 +958,10 @@ class DS4Backend final : public backend::Backend::Service { const bool think_enabled = ds4_think_mode_enabled(parse_think_mode(request)); const bool starts_in_thinking = think_enabled && request->usetokenizertemplate() && request->messages_size() > 0; + ds4cpp::RequestLifecycle lifecycle; StreamCtx s = { - g_engine, writer, ds4cpp::DsmlParser(starts_in_thinking), - 0, false, {}}; + g_engine, writer, &lifecycle, + ds4cpp::DsmlParser(starts_in_thinking), 0, {}}; std::string cache_key = render_prompt_text(request); size_t cache_hit = maybe_load_cache(cache_key); (void)cache_hit; @@ -893,16 +969,26 @@ class DS4Backend final : public backend::Backend::Service { // Manual loop on g_session - see Predict() above for the rationale. // MTP speculative path used when ds4_engine_mtp_draft_tokens > 0. char err[256] = {0}; - int rc = ds4_session_sync(g_session, &prompt, err, sizeof(err)); + int rc; + { + ds4cpp::CancelCallbackScope cancel_scope( + g_session, set_session_cancel, server_context_cancelled, context); + rc = ds4_session_sync(g_session, &prompt, err, sizeof(err)); + } ds4_tokens_free(&prompt); - if (rc == 0) { + if (rc == DS4_SESSION_SYNC_INTERRUPTED) { + lifecycle.ObserveContextCancellation(true); + } + const bool generation_started = rc == 0; + if (generation_started) { const int n_predict = ds4cpp::EffectiveGenerationLimit( request->tokens(), ds4_session_ctx(g_session), ds4_session_pos(g_session)); const int eos = ds4_token_eos(g_engine); const int draft_max = ds4_engine_mtp_draft_tokens(g_engine); int produced = 0; - while (produced < n_predict && !s.aborted) { + while (produced < n_predict) { + if (!request_should_continue(&lifecycle, context)) break; SampleParams sp = compute_sample_params(request, s.parser, think_enabled); int first; if (sp.temperature <= 0.0f) { @@ -926,43 +1012,67 @@ class DS4Backend final : public backend::Backend::Service { if (n < 0) { rc = -1; break; } bool stop = false; for (int j = 0; j < n; ++j) { + if (!request_should_continue(&lifecycle, context)) { + stop = true; + break; + } if (accepted[j] == eos) { stop = true; break; } stream_emit(&s, accepted[j]); - if (s.aborted) { stop = true; break; } + if (!lifecycle.ShouldContinue()) { stop = true; break; } if (++produced >= n_predict) { stop = true; break; } } if (stop) break; } else { stream_emit(&s, first); - if (s.aborted || ++produced >= n_predict) break; + if (!lifecycle.ShouldContinue() || ++produced >= n_predict) break; + if (!request_should_continue(&lifecycle, context)) break; rc = ds4_session_eval(g_session, first, err, sizeof(err)); if (rc != 0) break; } } - stream_done(&s); } - maybe_save_cache(cache_key); - // Flush parser state. - std::vector events; - s.parser.Flush(events); - if (!events.empty() && !s.aborted) { - backend::Reply reply; - auto *delta = reply.add_chat_deltas(); - for (const auto &e : events) { - if (e.type == ds4cpp::ParserEvent::CONTENT) { - delta->set_content(delta->content() + e.text); - } else if (e.type == ds4cpp::ParserEvent::REASONING) { - delta->set_reasoning_content(delta->reasoning_content() + e.text); + request_should_continue(&lifecycle, context); + ds4cpp::TerminalDecision terminal = ds4cpp::ResolveTerminalDecision( + rc == DS4_SESSION_SYNC_INTERRUPTED, rc != 0, + !lifecycle.ShouldFinalize()); + terminal = ds4cpp::RunPostlude( + terminal, + [&]() { + ds4cpp::DsmlParser staged_parser = s.parser; + std::vector events; + staged_parser.Flush(events); + bool write_succeeded = true; + if (!events.empty()) { + backend::Reply reply; + auto *delta = reply.add_chat_deltas(); + for (const auto &e : events) { + if (e.type == ds4cpp::ParserEvent::CONTENT) { + delta->set_content(delta->content() + e.text); + } else if (e.type == ds4cpp::ParserEvent::REASONING) { + delta->set_reasoning_content( + delta->reasoning_content() + e.text); + } + } + write_succeeded = s.writer->Write(reply); } - } - s.writer->Write(reply); - } - - if (rc != 0 && !s.aborted) { + lifecycle.ObserveStreamWrite(write_succeeded); + request_should_continue(&lifecycle, context); + if (!lifecycle.ShouldFinalize()) return false; + s.parser = std::move(staged_parser); + if (generation_started) stream_done(&s); + return true; + }, + [&]() { maybe_save_cache(cache_key); }); + + if (terminal.cause == ds4cpp::TerminalCause::EngineError) { return GStatus(StatusCode::INTERNAL, std::string("ds4 generation failed: ") + err); } + if (terminal.cause == ds4cpp::TerminalCause::Cancelled) { + return GStatus(StatusCode::CANCELLED, + "ds4 request cancelled"); + } return GStatus::OK; } diff --git a/backend/cpp/ds4/request_lifecycle.h b/backend/cpp/ds4/request_lifecycle.h new file mode 100644 index 000000000000..c3bc88f3c87c --- /dev/null +++ b/backend/cpp/ds4/request_lifecycle.h @@ -0,0 +1,111 @@ +// SPDX-License-Identifier: MIT +#pragma once + +namespace ds4cpp { + +using CancelCallback = bool (*)(void *); +using CancelSetter = void (*)(void *, CancelCallback, void *) noexcept; + +class CancelCallbackScope { +public: + CancelCallbackScope(void *target, CancelSetter setter, + CancelCallback callback, void *userdata) noexcept + : target_(target), setter_(setter) { + setter_(target_, callback, userdata); + } + + ~CancelCallbackScope() noexcept { + setter_(target_, nullptr, nullptr); + } + + CancelCallbackScope(const CancelCallbackScope &) = delete; + CancelCallbackScope &operator=(const CancelCallbackScope &) = delete; + +private: + void *target_; + CancelSetter setter_; +}; + +enum class RouteWaitDecision { + Pending, + Ready, + Error, + Cancelled, +}; + +inline RouteWaitDecision DecideRouteWait(int route_status, bool cancelled) { + if (cancelled) return RouteWaitDecision::Cancelled; + if (route_status > 0) return RouteWaitDecision::Ready; + if (route_status < 0) return RouteWaitDecision::Error; + return RouteWaitDecision::Pending; +} + +enum class TerminalCause { + Success, + Cancelled, + EngineError, +}; + +inline TerminalCause DecideTerminalCause(bool sync_interrupted, + bool engine_error, + bool abandoned) { + if (sync_interrupted) return TerminalCause::Cancelled; + if (engine_error) return TerminalCause::EngineError; + if (abandoned) return TerminalCause::Cancelled; + return TerminalCause::Success; +} + +struct TerminalDecision { + TerminalCause cause; + bool should_finalize; +}; + +inline TerminalDecision ResolveTerminalDecision(bool sync_interrupted, + bool engine_error, + bool abandoned) { + return { + DecideTerminalCause(sync_interrupted, engine_error, abandoned), + !sync_interrupted && !abandoned, + }; +} + +template +TerminalDecision RunPostlude(TerminalDecision terminal, + Finalize transactional_finalize, + Persist persist) { + if (!terminal.should_finalize) return terminal; + if (!transactional_finalize()) { + terminal.should_finalize = false; + if (terminal.cause != TerminalCause::EngineError) { + terminal.cause = TerminalCause::Cancelled; + } + return terminal; + } + persist(); + return terminal; +} + +class RequestLifecycle { +public: + void ObserveContextCancellation(bool cancelled) { + context_cancelled_ = context_cancelled_ || cancelled; + } + + void ObserveStreamWrite(bool succeeded) { + stream_write_aborted_ = stream_write_aborted_ || !succeeded; + } + + bool ShouldContinue() const { + return !context_cancelled_ && !stream_write_aborted_; + } + + bool ShouldFinalize() const { + return ShouldContinue(); + } + +private: + bool context_cancelled_ = false; + bool stream_write_aborted_ = false; +}; + +} // namespace ds4cpp diff --git a/backend/cpp/ds4/request_lifecycle_test.cpp b/backend/cpp/ds4/request_lifecycle_test.cpp new file mode 100644 index 000000000000..d6b968604d04 --- /dev/null +++ b/backend/cpp/ds4/request_lifecycle_test.cpp @@ -0,0 +1,414 @@ +// SPDX-License-Identifier: MIT +// Standalone regression tests for DS4 request cancellation policy. + +#include "request_lifecycle.h" + +#include + +namespace { + +int failures = 0; + +struct FakeCancelTarget { + ds4cpp::CancelCallback callback = nullptr; + void *userdata = nullptr; + int installs = 0; + int clears = 0; +}; + +struct PostludeCounts { + int finalize_attempts = 0; + int finalize_commits = 0; + int cache_persists = 0; + bool cache_followed_commit = true; +}; + +ds4cpp::TerminalDecision run_fake_postlude( + ds4cpp::TerminalDecision terminal, bool finalize_succeeds, + PostludeCounts *counts) { + return ds4cpp::RunPostlude( + terminal, + [=]() { + counts->finalize_attempts++; + if (!finalize_succeeds) return false; + counts->finalize_commits++; + return true; + }, + [=]() { + counts->cache_followed_commit = counts->finalize_commits == 1; + counts->cache_persists++; + }); +} + +bool fake_cancel(void *) { + return false; +} + +void fake_set_cancel(void *target, ds4cpp::CancelCallback callback, + void *userdata) noexcept { + auto *fake = static_cast(target); + fake->callback = callback; + fake->userdata = userdata; + if (callback) { + fake->installs++; + } else { + fake->clears++; + } +} + +void check(bool condition, const char *name) { + if (condition) return; + std::fprintf(stderr, "FAIL %s\n", name); + failures++; +} + +// Production mutation caught: treating an active request as abandoned would +// skip its parser finalization and cache save. +void test_active_request_continues_and_finalizes() { + ds4cpp::RequestLifecycle request; + + check(request.ShouldContinue(), "active:continue"); + check(request.ShouldFinalize(), "active:finalize"); +} + +// Production mutation caught: omitting the ServerContext cancellation branch +// would continue decoding and finalize a partial response. +void test_context_cancellation_stops_without_finalizing() { + ds4cpp::RequestLifecycle request; + + request.ObserveContextCancellation(true); + + check(!request.ShouldContinue(), "context_cancelled:stop"); + check(!request.ShouldFinalize(), "context_cancelled:no_finalize"); +} + +// Production mutation caught: ignoring ServerWriter::Write failure would keep +// streaming and finalize a response whose client has gone away. +void test_stream_write_abort_stops_without_finalizing() { + ds4cpp::RequestLifecycle request; + + request.ObserveStreamWrite(false); + + check(!request.ShouldContinue(), "write_abort:stop"); + check(!request.ShouldFinalize(), "write_abort:no_finalize"); +} + +// Production mutation caught: combining cancellation and write failure with +// AND would fail to stop when either signal occurs on its own. +void test_cancellation_and_write_abort_are_independent_or_conditions() { + ds4cpp::RequestLifecycle cancelled; + cancelled.ObserveContextCancellation(true); + cancelled.ObserveStreamWrite(true); + + ds4cpp::RequestLifecycle write_aborted; + write_aborted.ObserveContextCancellation(false); + write_aborted.ObserveStreamWrite(false); + + check(!cancelled.ShouldContinue(), "or:context_only"); + check(!write_aborted.ShouldContinue(), "or:write_only"); +} + +// Production mutation caught: treating an incomplete distributed route as an +// error would return before workers have time to connect. +void test_route_wait_pending() { + check(ds4cpp::DecideRouteWait(0, false) == + ds4cpp::RouteWaitDecision::Pending, + "route_wait:pending"); +} + +// Production mutation caught: failing to recognize a complete route would +// keep a ready inference request in the polling loop. +void test_route_wait_ready() { + check(ds4cpp::DecideRouteWait(1, false) == + ds4cpp::RouteWaitDecision::Ready, + "route_wait:ready"); +} + +// Production mutation caught: ignoring a route probe error would poll until a +// misleading timeout instead of returning UNAVAILABLE promptly. +void test_route_wait_error() { + check(ds4cpp::DecideRouteWait(-1, false) == + ds4cpp::RouteWaitDecision::Error, + "route_wait:error"); +} + +// Production mutation caught: omitting cancellation from route waiting would +// leave an abandoned request blocked until the distributed timeout. +void test_route_wait_cancellation() { + check(ds4cpp::DecideRouteWait(0, true) == + ds4cpp::RouteWaitDecision::Cancelled, + "route_wait:cancelled"); +} + +// Production mutation caught: checking route errors before cancellation would +// report UNAVAILABLE for a request the client already abandoned. +void test_route_wait_cancellation_precedes_error() { + check(ds4cpp::DecideRouteWait(-1, true) == + ds4cpp::RouteWaitDecision::Cancelled, + "route_wait:cancellation_precedence"); +} + +// Production mutation caught: classifying a successful active request as a +// terminal failure would suppress its normal response finalization. +void test_terminal_success() { + check(ds4cpp::DecideTerminalCause(false, false, false) == + ds4cpp::TerminalCause::Success, + "terminal:success"); +} + +// Production mutation caught: treating DS4's cooperative sync interruption +// as an ordinary engine error would return INTERNAL instead of CANCELLED. +void test_terminal_sync_interruption_is_cancelled() { + check(ds4cpp::DecideTerminalCause(true, true, true) == + ds4cpp::TerminalCause::Cancelled, + "terminal:sync_interrupted"); +} + +// Production mutation caught: treating every nonzero engine result as client +// abandonment would hide genuine DS4 failures behind CANCELLED. +void test_terminal_engine_error() { + check(ds4cpp::DecideTerminalCause(false, true, false) == + ds4cpp::TerminalCause::EngineError, + "terminal:engine_error"); +} + +// Production mutation caught: ignoring an rc==0 context cancellation would +// finalize and cache an abandoned request. +void test_terminal_context_abandonment() { + ds4cpp::RequestLifecycle request; + request.ObserveContextCancellation(true); + + check(ds4cpp::DecideTerminalCause( + false, false, !request.ShouldFinalize()) == + ds4cpp::TerminalCause::Cancelled, + "terminal:context_abandonment"); +} + +// Production mutation caught: ignoring an rc==0 stream write failure would +// finalize and cache an abandoned streaming request. +void test_terminal_write_abandonment() { + ds4cpp::RequestLifecycle request; + request.ObserveStreamWrite(false); + + check(ds4cpp::DecideTerminalCause( + false, false, !request.ShouldFinalize()) == + ds4cpp::TerminalCause::Cancelled, + "terminal:write_abandonment"); +} + +// Production mutation caught: checking late cancellation or write failure +// before a determined ordinary DS4 error would replace INTERNAL with CANCELLED. +void test_terminal_engine_error_precedes_late_abandonment() { + ds4cpp::RequestLifecycle cancelled; + cancelled.ObserveContextCancellation(true); + ds4cpp::RequestLifecycle write_aborted; + write_aborted.ObserveStreamWrite(false); + + check(ds4cpp::DecideTerminalCause( + false, true, !cancelled.ShouldFinalize()) == + ds4cpp::TerminalCause::EngineError, + "terminal:engine_error_precedes_cancellation"); + check(ds4cpp::DecideTerminalCause( + false, true, !write_aborted.ShouldFinalize()) == + ds4cpp::TerminalCause::EngineError, + "terminal:engine_error_precedes_write_abort"); +} + +// Production mutation caught: using status precedence alone to gate side +// effects would finalize and persist an engine-error request abandoned later. +void test_abandoned_engine_error_keeps_internal_without_finalizing() { + ds4cpp::RequestLifecycle request; + request.ObserveContextCancellation(true); + + ds4cpp::TerminalDecision terminal = ds4cpp::ResolveTerminalDecision( + false, true, !request.ShouldFinalize()); + + check(terminal.cause == ds4cpp::TerminalCause::EngineError, + "terminal_decision:abandoned_engine_error_status"); + check(!terminal.should_finalize, + "terminal_decision:abandoned_engine_error_no_finalize"); +} + +// Production mutation caught: suppressing side effects for every engine error +// would change the existing finalization and cache behavior of active failures. +void test_active_engine_error_still_finalizes() { + ds4cpp::RequestLifecycle request; + + ds4cpp::TerminalDecision terminal = ds4cpp::ResolveTerminalDecision( + false, true, !request.ShouldFinalize()); + + check(terminal.cause == ds4cpp::TerminalCause::EngineError, + "terminal_decision:active_engine_error_status"); + check(terminal.should_finalize, + "terminal_decision:active_engine_error_finalize"); +} + +// Production mutation caught: persisting before committed finalization would +// cache a state whose final buffered stream reply was never completed. +void test_postlude_active_success_commits_then_persists() { + PostludeCounts counts; + + ds4cpp::TerminalDecision terminal = run_fake_postlude( + {ds4cpp::TerminalCause::Success, true}, true, &counts); + + check(terminal.cause == ds4cpp::TerminalCause::Success, + "postlude:success_outcome"); + check(terminal.should_finalize, "postlude:success_committed"); + check(counts.finalize_attempts == 1, "postlude:success_attempts"); + check(counts.finalize_commits == 1, "postlude:success_commits"); + check(counts.cache_persists == 1, "postlude:success_cache"); + check(counts.cache_followed_commit, "postlude:success_cache_order"); +} + +// Production mutation caught: starting the postlude for an already-cancelled +// request would flush buffered parser state or persist an abandoned session. +void test_postlude_cancellation_skips_all_side_effects() { + PostludeCounts counts; + + ds4cpp::TerminalDecision terminal = run_fake_postlude( + {ds4cpp::TerminalCause::Cancelled, false}, true, &counts); + + check(terminal.cause == ds4cpp::TerminalCause::Cancelled, + "postlude:cancelled_outcome"); + check(counts.finalize_attempts == 0, "postlude:cancelled_attempts"); + check(counts.finalize_commits == 0, "postlude:cancelled_commits"); + check(counts.cache_persists == 0, "postlude:cancelled_cache"); +} + +// Production mutation caught: committing the live parser or cache after a +// failed final Write would publish an abandoned streaming postlude. +void test_postlude_finalize_failure_cancels_without_commit_or_cache() { + PostludeCounts counts; + + ds4cpp::TerminalDecision terminal = run_fake_postlude( + {ds4cpp::TerminalCause::Success, true}, false, &counts); + + check(terminal.cause == ds4cpp::TerminalCause::Cancelled, + "postlude:write_failure_outcome"); + check(!terminal.should_finalize, "postlude:write_failure_not_committed"); + check(counts.finalize_attempts == 1, "postlude:write_failure_attempts"); + check(counts.finalize_commits == 0, "postlude:write_failure_commits"); + check(counts.cache_persists == 0, "postlude:write_failure_cache"); +} + +// Production mutation caught: skipping the postlude for every engine error +// would change active internal-error finalization and cache behavior. +void test_postlude_active_engine_error_finalizes_and_persists() { + PostludeCounts counts; + + ds4cpp::TerminalDecision terminal = run_fake_postlude( + {ds4cpp::TerminalCause::EngineError, true}, true, &counts); + + check(terminal.cause == ds4cpp::TerminalCause::EngineError, + "postlude:engine_error_outcome"); + check(counts.finalize_attempts == 1, "postlude:engine_error_attempts"); + check(counts.finalize_commits == 1, "postlude:engine_error_commits"); + check(counts.cache_persists == 1, "postlude:engine_error_cache"); + check(counts.cache_followed_commit, "postlude:engine_error_cache_order"); +} + +// Production mutation caught: replacing every failed transactional finalize +// with cancellation would hide an already-determined engine error. +void test_postlude_engine_error_finalize_failure_preserves_internal() { + PostludeCounts counts; + + ds4cpp::TerminalDecision terminal = run_fake_postlude( + {ds4cpp::TerminalCause::EngineError, true}, false, &counts); + + check(terminal.cause == ds4cpp::TerminalCause::EngineError, + "postlude:engine_error_write_failure_outcome"); + check(!terminal.should_finalize, + "postlude:engine_error_write_failure_not_committed"); + check(counts.finalize_attempts == 1, + "postlude:engine_error_write_failure_attempts"); + check(counts.finalize_commits == 0, + "postlude:engine_error_write_failure_commits"); + check(counts.cache_persists == 0, + "postlude:engine_error_write_failure_cache"); +} + +// Production mutation caught: status precedence must not grant side-effect +// permission to an engine-error request that was also abandoned. +void test_postlude_abandoned_engine_error_skips_all_side_effects() { + PostludeCounts counts; + + ds4cpp::TerminalDecision terminal = run_fake_postlude( + {ds4cpp::TerminalCause::EngineError, false}, true, &counts); + + check(terminal.cause == ds4cpp::TerminalCause::EngineError, + "postlude:abandoned_engine_error_outcome"); + check(counts.finalize_attempts == 0, + "postlude:abandoned_engine_error_attempts"); + check(counts.finalize_commits == 0, + "postlude:abandoned_engine_error_commits"); + check(counts.cache_persists == 0, + "postlude:abandoned_engine_error_cache"); +} + +// Production mutation caught: failing to install the request callback would +// make DS4 prompt synchronization unable to observe client cancellation. +void test_cancel_callback_scope_installs_callback() { + FakeCancelTarget target; + int request_context = 42; + + { + ds4cpp::CancelCallbackScope scope( + &target, fake_set_cancel, fake_cancel, &request_context); + check(target.callback == fake_cancel, "cancel_scope:callback_installed"); + check(target.userdata == &request_context, "cancel_scope:userdata_installed"); + check(target.installs == 1, "cancel_scope:installed_once"); + } +} + +// Production mutation caught: failing to clear the callback at every scope +// exit would leave DS4 pointing at a destroyed stack-owned ServerContext. +void test_cancel_callback_scope_clears_callback() { + FakeCancelTarget target; + int request_context = 42; + + { + ds4cpp::CancelCallbackScope scope( + &target, fake_set_cancel, fake_cancel, &request_context); + } + + check(target.callback == nullptr, "cancel_scope:callback_cleared"); + check(target.userdata == nullptr, "cancel_scope:userdata_cleared"); + check(target.clears == 1, "cancel_scope:cleared_once"); +} + +} // namespace + +int main() { + test_active_request_continues_and_finalizes(); + test_context_cancellation_stops_without_finalizing(); + test_stream_write_abort_stops_without_finalizing(); + test_cancellation_and_write_abort_are_independent_or_conditions(); + test_route_wait_pending(); + test_route_wait_ready(); + test_route_wait_error(); + test_route_wait_cancellation(); + test_route_wait_cancellation_precedes_error(); + test_terminal_success(); + test_terminal_sync_interruption_is_cancelled(); + test_terminal_engine_error(); + test_terminal_context_abandonment(); + test_terminal_write_abandonment(); + test_terminal_engine_error_precedes_late_abandonment(); + test_abandoned_engine_error_keeps_internal_without_finalizing(); + test_active_engine_error_still_finalizes(); + test_postlude_active_success_commits_then_persists(); + test_postlude_cancellation_skips_all_side_effects(); + test_postlude_finalize_failure_cancels_without_commit_or_cache(); + test_postlude_active_engine_error_finalizes_and_persists(); + test_postlude_engine_error_finalize_failure_preserves_internal(); + test_postlude_abandoned_engine_error_skips_all_side_effects(); + test_cancel_callback_scope_installs_callback(); + test_cancel_callback_scope_clears_callback(); + + if (failures == 0) { + std::fprintf(stderr, "all request_lifecycle checks passed\n"); + return 0; + } + std::fprintf(stderr, "%d check(s) failed\n", failures); + return 1; +} diff --git a/docs/content/features/backends.md b/docs/content/features/backends.md index 7d2ec66a4f4a..2d04b987c592 100644 --- a/docs/content/features/backends.md +++ b/docs/content/features/backends.md @@ -186,3 +186,12 @@ LocalAI supports various types of backends: - **Utility Backends**: For reranking, PII/NER token classification, fine-tuning, quantization, and vector storage (e.g., rerankers, privacy-filter.cpp, TRL, local-store, valkey-store) See the [Backend & Model Compatibility Table]({{%relref "reference/compatibility-table" %}}) for the full catalog. + +### DS4 request cancellation + +The DS4 backend stops inference when a client cancels or disconnects, including +when a streaming response can no longer be written. Already-streamed chunks +cannot be retracted; DS4 does not flush incomplete buffered parser state or +persist an abandoned request to the disk KV cache. Cancellation is cooperative: +DS4 checks it at safe prompt-prefill and decode-loop boundaries, so a GPU kernel +already in flight may finish before the request stops.