backend(audio-cpp): serialize runs with a wedge-aware guard

audio.cpp sessions are not reentrant and a wedged CUDA call cannot be
cancelled, so a plain mutex would pile every worker thread behind a stuck GPU.
Callers waiting past the configured bound, or arriving while the holder has
already overrun it, fail fast instead.

A caller that queues behind a healthy run deliberately does not stamp the
clock: only the thread that takes the lock does. Stamping on arrival would
restart the wedge clock on every request and hide a stuck run from everyone
behind it, which is the pile-up this guard exists to prevent.

Assisted-by: Claude:claude-opus-5 [Claude Code]
Signed-off-by: Ettore Di Giacinto <mudler@localai.io>

backend/cpp/audio-cpp/CMakeLists.txt

@@ -90,6 +90,7 @@ add_executable(${TARGET}
     capability_routing.cpp
     audio_units.cpp
     transcript_assembly.cpp
+    run_guard.cpp
 )
 
 target_include_directories(${TARGET} PRIVATE

backend/cpp/audio-cpp/run_guard.cpp

@@ -0,0 +1,70 @@
+#include "run_guard.h"
+
+#include <chrono>
+
+namespace audiocpp_backend {
+
+std::int64_t steady_now_ms() {
+    return std::chrono::duration_cast<std::chrono::milliseconds>(
+               std::chrono::steady_clock::now().time_since_epoch())
+        .count();
+}
+
+bool run_has_overrun(std::int64_t busy_since_ms, std::int64_t now_ms,
+                     int timeout_ms) {
+    if (timeout_ms <= 0 || busy_since_ms == 0) {
+        return false;
+    }
+    return (now_ms - busy_since_ms) > timeout_ms;
+}
+
+int resolve_busy_timeout_ms(int ceiling, int requested) {
+    if (requested <= 0) {
+        return ceiling;
+    }
+    if (ceiling <= 0) {
+        return requested;
+    }
+    return requested < ceiling ? requested : ceiling;
+}
+
+RunGuard::Lock::Lock(RunGuard &guard, std::unique_lock<std::timed_mutex> lock)
+    : guard_(&guard), lock_(std::move(lock)) {}
+
+RunGuard::Lock::~Lock() {
+    if (guard_ != nullptr && lock_.owns_lock()) {
+        guard_->busy_since_ms_.store(0, std::memory_order_release);
+    }
+}
+
+RunGuard::Lock RunGuard::acquire(int timeout_ms, const std::string &label) {
+    std::unique_lock<std::timed_mutex> lock(mutex_, std::defer_lock);
+    if (timeout_ms <= 0) {
+        lock.lock(); // guard disabled: unbounded wait
+    } else {
+        // If the current holder has already run past the bound, treat it as
+        // wedged and fail fast rather than queue behind it. This is what stops
+        // requests piling up on a stuck GPU. Nothing is stored here: only the
+        // thread that actually takes the lock stamps the clock, so a rejected
+        // caller cannot hide the wedge from the callers after it.
+        const auto since = busy_since_ms_.load(std::memory_order_acquire);
+        const auto now_ms = steady_now_ms();
+        if (run_has_overrun(since, now_ms, timeout_ms)) {
+            throw BusyError("audio-cpp: model '" + label +
+                            "' is busy: the current inference has run " +
+                            std::to_string(now_ms - since) +
+                            " ms (busy_timeout_ms=" + std::to_string(timeout_ms) +
+                            "); it has likely wedged and cannot be cancelled");
+        }
+        if (!lock.try_lock_for(std::chrono::milliseconds(timeout_ms))) {
+            throw BusyError("audio-cpp: model '" + label +
+                            "' is busy: timed out after " +
+                            std::to_string(timeout_ms) +
+                            " ms waiting for the inference lock");
+        }
+    }
+    busy_since_ms_.store(steady_now_ms(), std::memory_order_release);
+    return Lock(*this, std::move(lock));
+}
+
+} // namespace audiocpp_backend

backend/cpp/audio-cpp/run_guard.h

@@ -0,0 +1,76 @@
+#pragma once
+
+// Serializes runs on one loaded model and bounds how long a caller waits.
+//
+// A single audio.cpp model runs one request at a time; its sessions are not
+// reentrant. If an inference wedges the GPU, a CUDA call that never returns
+// cannot be cancelled from userspace, so later requests would block forever
+// and every gRPC worker thread would pile up behind it. With a positive
+// timeout a caller instead gives up and throws BusyError, which grpc-server.cpp
+// maps to UNAVAILABLE so the client can retry or fail over. A timeout of 0
+// restores unbounded std::mutex behaviour.
+//
+// This mirrors the behaviour upstream documents in app/server/busy_guard.h.
+// That header is deliberately not copied: it is outside the public include
+// path, and it is Apache-2.0 while this repository is MIT.
+
+#include <atomic>
+#include <cstdint>
+#include <mutex>
+#include <stdexcept>
+#include <string>
+
+namespace audiocpp_backend {
+
+class BusyError : public std::runtime_error {
+public:
+    using std::runtime_error::runtime_error;
+};
+
+std::int64_t steady_now_ms();
+
+// True when a caller should fail fast instead of queueing: the current holder
+// acquired the guard at busy_since_ms and has already run longer than
+// timeout_ms. busy_since_ms == 0 means idle; timeout_ms <= 0 disables the guard.
+bool run_has_overrun(std::int64_t busy_since_ms, std::int64_t now_ms,
+                     int timeout_ms);
+
+// Resolves the bound for one run. `ceiling` is the model's configured policy;
+// `requested` is a per-request override where a value <= 0 means unset. A
+// request may ask for a shorter bound than policy but never a longer one, so a
+// client cannot weaken the guard. A ceiling of 0 means unbounded and therefore
+// compares as infinity.
+int resolve_busy_timeout_ms(int ceiling, int requested);
+
+class RunGuard {
+public:
+    // Holds the guard for the duration of a run and marks it idle again on
+    // release, including when the run throws, so the fail-fast check never
+    // sees a stale timestamp. Move only.
+    class Lock {
+    public:
+        Lock() = default;
+        Lock(RunGuard &guard, std::unique_lock<std::timed_mutex> lock);
+        Lock(Lock &&) = default;
+        Lock &operator=(Lock &&) = default;
+        Lock(const Lock &) = delete;
+        Lock &operator=(const Lock &) = delete;
+        ~Lock();
+
+    private:
+        RunGuard *guard_ = nullptr;
+        std::unique_lock<std::timed_mutex> lock_;
+    };
+
+    // `label` identifies the model in the error message. Throws BusyError when
+    // timeout_ms is positive and either the current holder has already overrun
+    // (fail fast, no wait) or the wait itself times out.
+    Lock acquire(int timeout_ms, const std::string &label);
+
+private:
+    friend class Lock;
+    std::timed_mutex mutex_;
+    std::atomic<std::int64_t> busy_since_ms_{0};
+};
+
+} // namespace audiocpp_backend

backend/cpp/audio-cpp/run_guard_test.cpp

@@ -0,0 +1,364 @@
+// Unit tests for run_guard. Standard library only. The harness compiles this
+// as a single translation unit, so the implementation is included directly.
+//
+// Every test that can block waits through AsyncAcquire, which bounds the wait
+// and reports a named failure. A guard that never returns must fail this suite,
+// not hang it: a hung test wedges CI far longer than a red one.
+
+#include "run_guard.cpp"
+
+#include <atomic>
+#include <chrono>
+#include <cstdio>
+#include <future>
+#include <string>
+#include <thread>
+#include <vector>
+
+static int failures = 0;
+
+static void check(bool ok, const std::string &name) {
+    if (!ok) {
+        failures++;
+        fprintf(stderr, "FAIL: %s\n", name.c_str());
+    } else {
+        fprintf(stderr, "ok:   %s\n", name.c_str());
+    }
+}
+
+using namespace audiocpp_backend;
+
+namespace {
+
+struct Outcome {
+    bool threw = false;
+    std::string message;
+};
+
+// Runs one acquire on a worker thread so that a guard which waits forever fails
+// a named check instead of hanging the suite. The worker is joined by the
+// future's destructor, so every test must first unblock whatever holds the
+// guard before this object goes out of scope.
+class AsyncAcquire {
+public:
+    AsyncAcquire(RunGuard &guard, int timeout_ms, const std::string &label)
+        : future_(std::async(std::launch::async, [&guard, timeout_ms, label] {
+              Outcome out;
+              try {
+                  auto lock = guard.acquire(timeout_ms, label);
+                  (void)lock;
+              } catch (const BusyError &err) {
+                  out.threw = true;
+                  out.message = err.what();
+              }
+              return out;
+          })) {}
+
+    // True when the acquire returned within watchdog_ms.
+    bool returned_within(int watchdog_ms) {
+        return future_.wait_for(std::chrono::milliseconds(watchdog_ms)) ==
+               std::future_status::ready;
+    }
+
+    // Only valid once returned_within() reported true, or after the holder has
+    // been released; otherwise it blocks.
+    Outcome outcome() { return future_.get(); }
+
+private:
+    std::future<Outcome> future_;
+};
+
+// A run held open on its own thread until release() is called, so the main
+// thread can observe a guard that is genuinely busy.
+class HeldRun {
+public:
+    explicit HeldRun(RunGuard &guard) {
+        thread_ = std::thread([this, &guard] {
+            auto lock = guard.acquire(0, "holder");
+            held_.store(true);
+            while (!release_.load()) {
+                std::this_thread::sleep_for(std::chrono::milliseconds(1));
+            }
+        });
+        while (!held_.load()) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        }
+    }
+
+    void release() {
+        release_.store(true);
+        if (thread_.joinable()) {
+            thread_.join();
+        }
+    }
+
+    ~HeldRun() { release(); }
+
+private:
+    std::thread thread_;
+    std::atomic<bool> held_{false};
+    std::atomic<bool> release_{false};
+};
+
+const int kWatchdogMs = 5000;
+
+} // namespace
+
+static void test_steady_clock_advances() {
+    const std::int64_t before = steady_now_ms();
+    std::this_thread::sleep_for(std::chrono::milliseconds(30));
+    const std::int64_t after = steady_now_ms();
+    check(after >= before + 10, "steady_now_ms advances with real time");
+}
+
+static void test_overrun_predicate() {
+    // busy_since_ms == 0 means idle: never an overrun.
+    check(!run_has_overrun(0, 100000, 1000), "idle is never an overrun");
+    // timeout_ms <= 0 disables the guard.
+    check(!run_has_overrun(1000, 100000, 0), "zero timeout disables the check");
+    check(!run_has_overrun(1000, 100000, -5), "negative timeout disables the check");
+    // Held for 500ms under a 1000ms bound: fine.
+    check(!run_has_overrun(1000, 1500, 1000), "within the bound is not an overrun");
+    // Exactly at the bound is not yet an overrun.
+    check(!run_has_overrun(1000, 2000, 1000), "exactly at the bound is not an overrun");
+    // One millisecond past the bound is.
+    check(run_has_overrun(1000, 2001, 1000), "past the bound is an overrun");
+    check(run_has_overrun(1000, 100000, 1000), "long past the bound is an overrun");
+}
+
+static void test_timeout_resolution() {
+    // No per-request value: policy applies.
+    check(resolve_busy_timeout_ms(5000, 0) == 5000, "unset request uses the ceiling");
+    check(resolve_busy_timeout_ms(5000, -1) == 5000, "negative request uses the ceiling");
+    // A request may ask for a shorter bound than policy.
+    check(resolve_busy_timeout_ms(5000, 1000) == 1000, "shorter request is honoured");
+    // A request may never ask for a longer bound than policy.
+    check(resolve_busy_timeout_ms(1000, 5000) == 1000, "longer request is capped");
+    // Equal on both sides is that value, not something else.
+    check(resolve_busy_timeout_ms(1000, 1000) == 1000, "an equal request is that value");
+    // Unbounded policy honours whatever the request asks for.
+    check(resolve_busy_timeout_ms(0, 2000) == 2000, "unbounded policy honours the request");
+    check(resolve_busy_timeout_ms(0, 0) == 0, "unbounded on both sides stays unbounded");
+}
+
+static void test_uncontended_acquire() {
+    RunGuard guard;
+    bool threw = false;
+    try {
+        auto lock = guard.acquire(1000, "model");
+        (void)lock;
+    } catch (const BusyError &) {
+        threw = true;
+    }
+    check(!threw, "an uncontended acquire succeeds");
+}
+
+static void test_release_clears_the_busy_timestamp() {
+    RunGuard guard;
+    {
+        auto lock = guard.acquire(1000, "model");
+        (void)lock;
+    }
+    // Sleep well past the bound the next caller will use. Releasing must mark
+    // the guard idle again: a stale timestamp would make this caller fail fast
+    // against a model that has not been running anything for 200ms.
+    std::this_thread::sleep_for(std::chrono::milliseconds(200));
+
+    bool threw = false;
+    std::string message;
+    try {
+        auto lock = guard.acquire(20, "model");
+        (void)lock;
+    } catch (const BusyError &err) {
+        threw = true;
+        message = err.what();
+    }
+    check(!threw, "release clears the busy timestamp");
+    if (threw) {
+        fprintf(stderr, "      unexpected BusyError: %s\n", message.c_str());
+    }
+}
+
+static void test_exception_in_run_still_releases() {
+    RunGuard guard;
+    try {
+        auto lock = guard.acquire(1000, "model");
+        (void)lock;
+        throw std::runtime_error("inference blew up");
+    } catch (const std::runtime_error &) {
+        // swallowed
+    }
+    // Same reasoning as above: the unwind must clear the timestamp, not just
+    // drop the mutex, or the next caller fails fast against an idle model.
+    std::this_thread::sleep_for(std::chrono::milliseconds(200));
+
+    bool threw = false;
+    try {
+        auto lock = guard.acquire(20, "model");
+        (void)lock;
+    } catch (const BusyError &) {
+        threw = true;
+    }
+    check(!threw, "a throwing run still releases the guard");
+}
+
+static void test_contended_acquire_times_out() {
+    RunGuard guard;
+    HeldRun holder(guard);
+
+    // 300ms is generous on purpose: the waiter must still be inside its bound
+    // when it looks, so that this exercises the wait path and not the fail-fast
+    // path. Only a 300ms stall between the holder acquiring and the waiter
+    // checking would confuse the two.
+    AsyncAcquire waiter(guard, 300, "my-model");
+    const bool returned = waiter.returned_within(kWatchdogMs);
+    check(returned, "a contended acquire returns rather than waiting forever");
+
+    if (returned) {
+        const Outcome out = waiter.outcome();
+        check(out.threw, "a contended acquire past its timeout throws BusyError");
+        check(out.message.find("my-model") != std::string::npos,
+              "the error names the model");
+        check(out.message.find("timed out after") != std::string::npos,
+              "waiting out the bound reports the wait, not a wedge");
+    }
+
+    holder.release();
+
+    // After the holder releases, the guard is usable again.
+    bool after_threw = false;
+    try {
+        auto lock = guard.acquire(1000, "my-model");
+        (void)lock;
+    } catch (const BusyError &) {
+        after_threw = true;
+    }
+    check(!after_threw, "the guard is reusable once released");
+}
+
+static void test_fails_fast_when_the_holder_has_overrun() {
+    RunGuard guard;
+    HeldRun holder(guard);
+
+    // Let the holder run past every bound the callers below use, so each of
+    // them must fail fast on the holder's own start time instead of queueing.
+    std::this_thread::sleep_for(std::chrono::milliseconds(250));
+
+    AsyncAcquire first(guard, 40, "wedged-model");
+    const bool first_returned = first.returned_within(kWatchdogMs);
+    check(first_returned, "a caller meeting a wedged run returns");
+    if (first_returned) {
+        const Outcome out = first.outcome();
+        check(out.threw, "a caller meeting a wedged run throws BusyError");
+        check(out.message.find("has run") != std::string::npos,
+              "the error reports the wedged run, not a plain wait timeout");
+        check(out.message.find("wedged-model") != std::string::npos,
+              "the fail-fast error names the model");
+    }
+
+    // A second caller must see the same thing. A waiter that stamps its own
+    // arrival time over the holder's would hide the wedge from everyone after
+    // it, which is exactly the pile-up this guard exists to prevent.
+    AsyncAcquire second(guard, 40, "wedged-model");
+    const bool second_returned = second.returned_within(kWatchdogMs);
+    check(second_returned, "a second caller meeting a wedged run returns");
+    if (second_returned) {
+        const Outcome out = second.outcome();
+        check(out.threw, "a second caller meeting a wedged run throws BusyError");
+        check(out.message.find("has run") != std::string::npos,
+              "a rejected waiter does not reset the wedge clock");
+    }
+
+    holder.release();
+}
+
+// A caller that queues behind a healthy run must not stamp its own arrival over
+// the holder's start time. If it did, the wedge clock would restart on every
+// arrival and a run that never finishes would look young forever to everyone
+// who arrives after: the pile-up this guard exists to prevent, back again.
+static void test_a_queued_waiter_does_not_reset_the_wedge_clock() {
+    RunGuard guard;
+    HeldRun holder(guard);
+
+    // 300ms in, a caller with a 5s bound is nowhere near its own limit, so it
+    // queues rather than failing fast. That is the caller that could clobber
+    // the clock.
+    std::this_thread::sleep_for(std::chrono::milliseconds(300));
+    AsyncAcquire queued(guard, 5000, "patient-model");
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+
+    // 400ms in, this caller's own 250ms bound is already exceeded by the
+    // holder, so it must fail fast. It only can if the clock still reads the
+    // holder's start: 400ms of holding, not the 100ms since the queued caller
+    // arrived.
+    AsyncAcquire impatient(guard, 250, "wedged-model");
+    const bool returned = impatient.returned_within(kWatchdogMs);
+    check(returned, "a caller arriving behind a queued waiter returns");
+    if (returned) {
+        const Outcome out = impatient.outcome();
+        check(out.threw, "a caller arriving behind a queued waiter throws BusyError");
+        check(out.message.find("has run") != std::string::npos,
+              "a queued waiter does not reset the wedge clock");
+    }
+
+    holder.release();
+
+    const bool queued_returned = queued.returned_within(kWatchdogMs);
+    check(queued_returned, "the queued waiter returns once the holder releases");
+    if (queued_returned) {
+        check(!queued.outcome().threw, "the queued waiter gets its turn");
+    }
+}
+
+static void test_runs_are_serialized() {
+    RunGuard guard;
+    std::atomic<int> concurrent{0};
+    std::atomic<int> peak{0};
+    std::atomic<bool> unexpected_busy{false};
+
+    std::vector<std::thread> threads;
+    for (int t = 0; t < 4; ++t) {
+        threads.emplace_back([&] {
+            for (int i = 0; i < 40; ++i) {
+                try {
+                    auto lock = guard.acquire(0, "model");
+                    const int inside = ++concurrent;
+                    int seen = peak.load();
+                    while (inside > seen &&
+                           !peak.compare_exchange_weak(seen, inside)) {
+                    }
+                    std::this_thread::sleep_for(std::chrono::microseconds(20));
+                    --concurrent;
+                } catch (const BusyError &) {
+                    unexpected_busy.store(true);
+                    return;
+                }
+            }
+        });
+    }
+    for (auto &thread : threads) {
+        thread.join();
+    }
+
+    check(!unexpected_busy.load(), "an unbounded acquire never gives up");
+    check(peak.load() == 1, "only one run holds the guard at a time");
+}
+
+int main() {
+    test_steady_clock_advances();
+    test_overrun_predicate();
+    test_timeout_resolution();
+    test_uncontended_acquire();
+    test_release_clears_the_busy_timestamp();
+    test_exception_in_run_still_releases();
+    test_contended_acquire_times_out();
+    test_fails_fast_when_the_holder_has_overrun();
+    test_a_queued_waiter_does_not_reset_the_wedge_clock();
+    test_runs_are_serialized();
+    if (failures) {
+        fprintf(stderr, "%d check(s) failed\n", failures);
+        return 1;
+    }
+    fprintf(stderr, "all run_guard checks passed\n");
+    return 0;
+}