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.