diff --git a/backend/cpp/audio-cpp/CMakeLists.txt b/backend/cpp/audio-cpp/CMakeLists.txt index 573bcbc8b..18dd5e981 100644 --- a/backend/cpp/audio-cpp/CMakeLists.txt +++ b/backend/cpp/audio-cpp/CMakeLists.txt @@ -105,9 +105,23 @@ target_link_libraries(hw_grpc_proto PUBLIC protobuf::libprotobuf gRPC::grpc++) # # "package" makes sentencepiece use the protobuf found above, which is the one # the generated code was built against. It must be set before add_subdirectory, -# since that is when sentencepiece's own cache entry is created. After it, zero -# google::protobuf:: definitions remain in the executable, and citrinet_asr, -# which parses a SentencePiece ModelProto at load, still tokenizes correctly. +# since that is when sentencepiece's own cache entry is created. +# +# WHAT THAT BUYS IS ONE PROTOBUF RUNTIME, not an executable free of protobuf +# symbols, and the difference matters to whoever checks this next. Measured with +# nm -C --defined-only on the linked grpc-server, 2515 google::protobuf:: +# symbols are still DEFINED in it, and that is what should be there: they are +# generated code, sentencepiece::ModelProto's own _InternalParse and +# CheckTypeAndMergeFrom among them, which name protobuf types in their +# signatures and are compiled into every user of a .proto. Expecting zero would +# send a reader looking for a regression that is not one. +# +# The claim that decides whether the ABI mismatch above is gone is the RUNTIME +# one, and it holds: google::protobuf::internal::ParseContext::ParseMessage is +# UNDEFINED in the executable, so every generated _InternalParse resolves it to +# libprotobuf.so at load instead of to a vendored 3.14 copy. No vendored +# protobuf-lite archive is pulled in at all, and citrinet_asr, which parses a +# SentencePiece ModelProto at load, tokenizes correctly as a result. set(SPM_PROTOBUF_PROVIDER "package" CACHE STRING "Make sentencepiece use the found protobuf, not its vendored 3.14 copy" FORCE) diff --git a/backend/cpp/audio-cpp/audio_units.cpp b/backend/cpp/audio-cpp/audio_units.cpp index 7bc07ab23..99a096cbd 100644 --- a/backend/cpp/audio-cpp/audio_units.cpp +++ b/backend/cpp/audio-cpp/audio_units.cpp @@ -62,11 +62,19 @@ std::int64_t seconds_to_samples(double seconds, int sample_rate) { // // That round trip is exact only below roughly 2^23 samples. Past that the // float samples_to_seconds returns can no longer resolve adjacent indices - // and the trip fails whatever the rounding: measured first failures run - // from 11289602 samples (4.3 min at 44.1 kHz, 2.1 min at 96 kHz) to - // 16384001 (17 min at 16 kHz). That is a property of the float seconds API - // itself, not of the rounding here, and it is why nothing should use these - // to carry a sample-accurate position in a long recording. + // and the trip fails whatever the rounding. Both the first failing INDEX + // and the duration it stands for depend on the rate, so they are listed per + // rate rather than folded into one range; measured: + // + // 16 kHz 16384001 samples 17.1 min + // 44.1 kHz 11289602 samples 4.3 min + // 48 kHz 12288002 samples 4.3 min + // 96 kHz 12288002 samples 2.1 min + // + // The shortest recording this bites is therefore a couple of minutes of + // 96 kHz audio. It is a property of the float seconds API itself, not of + // the rounding here, and it is why nothing should use these to carry a + // sample-accurate position in a long recording. return static_cast(std::llround(scaled)); } diff --git a/backend/cpp/audio-cpp/capability_routing.cpp b/backend/cpp/audio-cpp/capability_routing.cpp index 198a5ff4d..5fc2b63a7 100644 --- a/backend/cpp/audio-cpp/capability_routing.cpp +++ b/backend/cpp/audio-cpp/capability_routing.cpp @@ -166,6 +166,35 @@ const char *const kVoiceEmbedReason = "itself exists upstream, and TitaNet and ECAPA-TDNN exist as internal " "conditioning encoders, but neither is registered as a loadable family"; +// The tasks an RPC is ever willing to route to, independent of request shape. +// +// DERIVED from task_candidates rather than restated, so a task added to an +// RPC's candidate list cannot become inadmissible as a pin by omission. Setting +// every shape flag yields each RPC's widest list: the per-flag branches only +// reorder the same three tasks for Tts, and only ADD Alignment for +// transcription, so the union is what comes back. +std::vector admissible_tasks(Rpc rpc) { + RequestShape widest; + widest.has_voice_reference = true; + widest.has_instructions = true; + widest.has_prompt_text = true; + return task_candidates(rpc, widest); +} + +std::string join_task_names(const std::vector &tasks) { + std::string out; + for (const Task task : tasks) { + if (!out.empty()) { + out += ", "; + } + out += task_name(task); + } + if (out.empty()) { + out = "nothing"; + } + return out; +} + std::string join_attempts(const std::vector &tasks, const std::vector &modes) { std::string out; @@ -324,8 +353,34 @@ Route resolve_route(Rpc rpc, const RequestShape &shape, "align, vad, diar, sep, vdes, spk"; return route; } - // A pinned task is honoured exactly. Silently rerouting would make the - // option meaningless and hide a misconfigured model. + // A pin is honoured exactly, but ONLY on an RPC that could have routed + // to it anyway. It used to replace the candidate list wholesale for + // every RPC, and because the model's `task:` option is copied into the + // shape by all nine handlers, one pin bled across all nine surfaces and + // produced wrong 200s rather than errors: nemotron with task:asr made + // Vad return 200 with zero segments after a full ASR decode, so 14 + // seconds of speech was reported as silence, and silero_vad with + // task:vad made AudioTranscription return 200 with empty text and four + // segments whose spans were VAD segments, which the srt/vtt/lrc writers + // then rendered as a well formed subtitle file of four timed EMPTY + // cues. Refusing is what the docs already promise: "if the family + // cannot serve it, the request is refused rather than rerouted". + // + // Every legitimate pin survives, because a pin only ever names the task + // its own RPC already routes to: svc is in AudioTransform's candidates, + // tts/clon/vdes in TTS's, asr in transcription's, vad and diar in + // theirs. + const std::vector admissible = admissible_tasks(rpc); + if (std::find(admissible.begin(), admissible.end(), pinned) == + admissible.end()) { + route.error = std::string("audio-cpp: this model pins task '") + + task_name(pinned) + "', which the " + rpc_name(rpc) + + " RPC never routes to (it routes to " + + join_task_names(admissible) + + "). Remove the task option to reach this RPC, or call " + "the RPC the pinned task serves"; + return route; + } tasks = {pinned}; } else { tasks = task_candidates(rpc, shape); diff --git a/backend/cpp/audio-cpp/capability_routing_test.cpp b/backend/cpp/audio-cpp/capability_routing_test.cpp index 303d70dd8..b4f44e289 100644 --- a/backend/cpp/audio-cpp/capability_routing_test.cpp +++ b/backend/cpp/audio-cpp/capability_routing_test.cpp @@ -240,6 +240,103 @@ static void test_pinned_task_overrides_routing() { check(!u.ok, "a pinned but unsupported task is refused"); } +// A pin lives on the MODEL, and every one of the nine handlers copies it into +// the shape, so a pin set for one RPC arrives at all of them. It used to +// replace the candidate list wholesale, which turned the other eight into wrong +// 200s rather than errors: nemotron pinned to asr made Vad answer with zero +// segments after a full ASR decode, and silero_vad pinned to vad made +// AudioTranscription answer with empty text and four segments whose spans were +// VAD segments, which the srt/vtt/lrc writers rendered as timed EMPTY cues. +static void test_pin_must_be_admissible_for_the_rpc() { + Capabilities nemotron_asr{"nemotron_asr", + {{Task::Asr, {Mode::Offline, Mode::Streaming}}, + {Task::Vad, {Mode::Offline}}}}; + + // The pin is legitimate on the RPC it was meant for. + RequestShape asr_pin; + asr_pin.pinned_task = "asr"; + const auto transcription = + resolve_route(Rpc::AudioTranscription, asr_pin, nemotron_asr); + check(transcription.ok && transcription.task == Task::Asr, + "an admissible pin is still honoured exactly"); + + // ...and refused on one that never routes to it, EVEN THOUGH the family + // advertises the pinned task. That is the whole point: family support is + // not the question, RPC admissibility is. + const auto vad = resolve_route(Rpc::Vad, asr_pin, nemotron_asr); + check(!vad.ok, "an inadmissible pin is refused rather than served"); + check(vad.error.find("asr") != std::string::npos, + "the refusal names the pinned task"); + check(vad.error.find(rpc_name(Rpc::Vad)) != std::string::npos, + "the refusal names the RPC that cannot serve it"); + + Capabilities silero{"silero_vad", {{Task::Vad, {Mode::Offline}}}}; + RequestShape vad_pin; + vad_pin.pinned_task = "vad"; + const auto vad_ok = resolve_route(Rpc::Vad, vad_pin, silero); + check(vad_ok.ok && vad_ok.task == Task::Vad, "vad is admissible on Vad"); + const auto transcribe_vad = + resolve_route(Rpc::AudioTranscription, vad_pin, silero); + check(!transcribe_vad.ok, + "a vad pin cannot make a transcription request return empty cues"); + + // Every pin a shipped configuration could sensibly set stays reachable on + // the RPC that serves it. This is the list the fix was checked against. + struct AdmissibleCase { + Rpc rpc; + const char *task; + }; + const AdmissibleCase kAdmissible[] = { + {Rpc::AudioTransform, "svc"}, {Rpc::AudioTransform, "sep"}, + {Rpc::AudioTransform, "vc"}, {Rpc::AudioTransform, "s2s"}, + {Rpc::Tts, "tts"}, {Rpc::Tts, "clon"}, + {Rpc::Tts, "vdes"}, {Rpc::TtsStream, "tts"}, + {Rpc::AudioTranscription, "asr"}, + {Rpc::AudioTranscription, "align"}, + {Rpc::AudioTranscriptionStream, "asr"}, + {Rpc::AudioTranscriptionLive, "asr"}, + {Rpc::Vad, "vad"}, {Rpc::Diarize, "diar"}, + {Rpc::SoundGeneration, "gen"}, + }; + for (const auto &entry : kAdmissible) { + Task task = Task::Tts; + check(parse_task_name(entry.task, task), + std::string("known task name: ") + entry.task); + // A family that advertises the pinned task offline and nothing else, so + // the ONLY thing that can refuse the route is the admissibility check. + Capabilities only{"probe", + {{task, {Mode::Offline, Mode::Streaming}}}}; + RequestShape pin; + pin.pinned_task = entry.task; + const auto route = resolve_route(entry.rpc, pin, only); + check(route.ok && route.task == task, + std::string("pin '") + entry.task + "' stays admissible on " + + rpc_name(entry.rpc)); + } + + // And the pins that must NOT cross over, one per RPC pair that was + // observed producing a wrong 200. + const AdmissibleCase kInadmissible[] = { + {Rpc::Vad, "asr"}, {Rpc::Diarize, "asr"}, + {Rpc::AudioTranscription, "vad"}, {Rpc::AudioTranscription, "diar"}, + {Rpc::Tts, "asr"}, {Rpc::Vad, "tts"}, + {Rpc::SoundGeneration, "tts"}, {Rpc::AudioTransform, "asr"}, + }; + for (const auto &entry : kInadmissible) { + Task task = Task::Tts; + check(parse_task_name(entry.task, task), + std::string("known task name: ") + entry.task); + Capabilities only{"probe", + {{task, {Mode::Offline, Mode::Streaming}}}}; + RequestShape pin; + pin.pinned_task = entry.task; + const auto route = resolve_route(entry.rpc, pin, only); + check(!route.ok, + std::string("pin '") + entry.task + "' is refused on " + + rpc_name(entry.rpc)); + } +} + static void test_vad_and_diarize() { Capabilities silero{"silero_vad", {{Task::Vad, {Mode::Offline, Mode::Streaming}}}}; const auto v = resolve_route(Rpc::Vad, RequestShape{}, silero); @@ -484,6 +581,7 @@ int main() { test_prompt_does_not_hijack_asr(); test_audio_transform_prefers_separation(); test_pinned_task_overrides_routing(); + test_pin_must_be_admissible_for_the_rpc(); test_vad_and_diarize(); test_sound_generation(); test_names_round_trip(); diff --git a/backend/cpp/audio-cpp/grpc-server.cpp b/backend/cpp/audio-cpp/grpc-server.cpp index 91e7a3e9f..2b66c1205 100644 --- a/backend/cpp/audio-cpp/grpc-server.cpp +++ b/backend/cpp/audio-cpp/grpc-server.cpp @@ -496,10 +496,17 @@ float audio_duration_seconds(const engine::runtime::AudioBuffer &audio) { // chatterbox advertises clon and no tts at all, so before this every preset-only // or voice-less request to it got that engine-internal message. // -// It cannot misfire on the legitimate case: has_voice_reference is what made -// routing choose VoiceCloning in the first place, so reaching here with a clip -// present is impossible unless the model pinned task:clon, and a clon pin with -// no clip is exactly the misconfiguration worth naming. +// The case it catches is the FALLBACK one, which is worth stating the right way +// round. A clip is what makes routing put VoiceCloning first, but VoiceCloning +// is also the last candidate a clip-less TTS request falls back to +// (task_candidates returns {Tts, VoiceCloning, VoiceDesign} when no clip and no +// instructions are supplied), so a family that advertises clon and no tts at +// all - chatterbox, which is what ships in the gallery - routes EVERY voice-less +// request here. A task:clon pin arrives here the same way. +// +// It cannot misfire on the legitimate case for the opposite reason: a request +// that did supply a clip has voice_is_file set and returns on the line above +// before anything is thrown. // // Deliberately NOT generalised to "every task whose family declares // supports_speaker_reference". That flag lives on the engine's CapabilitySet, is @@ -1827,7 +1834,6 @@ public: std::int64_t consumed_frames = 0; const auto next_frames = [&](std::vector &out) { while (stream->Read(&incoming)) { - idle.touch(); if (incoming.has_config()) { // backend.proto says a second Config resets the decode // session. audio.cpp cannot do that truthfully: a reset @@ -1844,15 +1850,20 @@ public: "' cannot reset a live session mid-stream; close " "this stream and open a new one"); } - if (!incoming.has_audio()) { - // Neither arm of the oneof is set. Nothing to feed, and - // not worth failing a live session over. - continue; - } + // Touched only for a frame the decoder can actually consume, + // and touched AFTER the filters rather than before them. + // Neither arm of the oneof set, or an empty pcm field, is + // nothing to feed and not worth failing a live session over + // - but it is also not the peer proving it is still there, + // and resetting the window for it let one client hold the + // model's only lane indefinitely by writing empty frames + // faster than the window. See live_frame_carries_audio. const auto &pcm = incoming.audio().pcm(); - if (pcm.empty()) { + if (!audiocpp_backend::live_frame_carries_audio( + incoming.has_audio(), pcm.empty())) { continue; } + idle.touch(); out.assign(pcm.begin(), pcm.end()); // Mono, so floats and frames are the same count. Only used // for the duration reported in final_result. diff --git a/backend/cpp/audio-cpp/live_watchdog.cpp b/backend/cpp/audio-cpp/live_watchdog.cpp index d0d7aa001..94263c782 100644 --- a/backend/cpp/audio-cpp/live_watchdog.cpp +++ b/backend/cpp/audio-cpp/live_watchdog.cpp @@ -68,4 +68,8 @@ void IdleWatchdog::run() { } } +bool live_frame_carries_audio(bool has_audio, bool pcm_empty) { + return has_audio && !pcm_empty; +} + } // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/live_watchdog.h b/backend/cpp/audio-cpp/live_watchdog.h index 37e6149bc..7f82ec124 100644 --- a/backend/cpp/audio-cpp/live_watchdog.h +++ b/backend/cpp/audio-cpp/live_watchdog.h @@ -81,4 +81,19 @@ private: std::thread thread_; }; +// Whether one message read off a live stream is a frame the decoder can +// actually consume, which is the ONLY thing that counts as the peer proving it +// is still there. +// +// Split out of the read loop so the distinction is testable, and because +// getting it wrong is silent. The loop used to touch the watchdog on ANY +// message, before it filtered on has_audio and on an empty pcm field, so a peer +// writing unset-oneof or zero-length frames faster than the window held the +// lane forever: no audio was ever fed, no work was ever done, and the timer +// that exists to break exactly that grip was reset by the frames doing it. +// There is one lane per model and one model per process, so that is a single +// client denying the whole backend. The thrown message already said "no audio +// frame arrived"; this is the code agreeing with it. +bool live_frame_carries_audio(bool has_audio, bool pcm_empty); + } // namespace audiocpp_backend diff --git a/backend/cpp/audio-cpp/live_watchdog_test.cpp b/backend/cpp/audio-cpp/live_watchdog_test.cpp index bb00db9dc..184655b2b 100644 --- a/backend/cpp/audio-cpp/live_watchdog_test.cpp +++ b/backend/cpp/audio-cpp/live_watchdog_test.cpp @@ -141,6 +141,70 @@ static void test_a_firing_watchdog_still_joins_cleanly() { check(calls.load() == 1, "the callback ran once, inside the scope"); } +// The predicate the live read loop filters on. +static void test_only_a_frame_with_audio_counts() { + using audiocpp_backend::live_frame_carries_audio; + check(live_frame_carries_audio(true, false), + "a frame with a non-empty pcm field carries audio"); + check(!live_frame_carries_audio(true, true), + "an empty pcm field does not"); + check(!live_frame_carries_audio(false, false), + "an unset audio oneof does not, whatever the pcm field looks like"); + check(!live_frame_carries_audio(false, true), + "and neither does an unset oneof with an empty pcm field"); +} + +// The defect this closes, expressed as behaviour rather than as a call order: +// a peer writing frames the decoder cannot consume, faster than the window, +// used to hold the model's only inference lane forever, because the read loop +// touched the watchdog before it filtered them out. One lane per model and one +// model per process, so that is a single client denying the whole backend, +// which is exactly what the watchdog exists to prevent. +static void test_empty_frames_do_not_hold_the_lane() { + std::atomic cancels{0}; + std::atomic stop{false}; + audiocpp_backend::IdleWatchdog watchdog(80ms, [&cancels] { ++cancels; }); + + // The read loop with the real filter in it: frames arrive continuously, + // none of them carries audio, and only a frame that does may touch. + std::thread peer([&] { + while (!stop.load()) { + if (audiocpp_backend::live_frame_carries_audio(false, true)) { + watchdog.touch(); + } + std::this_thread::sleep_for(5ms); + } + }); + + std::this_thread::sleep_for(600ms); + stop.store(true); + peer.join(); + watchdog.disarm(); + + check(cancels.load() == 1, + "a flood of frames with no audio in them still releases the lane"); + + // The mirror image, so this cannot pass merely because the watchdog always + // fires: a peer that keeps sending audio is left alone, exactly as before. + std::atomic live_cancels{0}; + std::atomic live_stop{false}; + audiocpp_backend::IdleWatchdog live(80ms, [&live_cancels] { ++live_cancels; }); + std::thread speaker([&] { + while (!live_stop.load()) { + if (audiocpp_backend::live_frame_carries_audio(true, false)) { + live.touch(); + } + std::this_thread::sleep_for(5ms); + } + }); + std::this_thread::sleep_for(600ms); + live_stop.store(true); + speaker.join(); + live.disarm(); + check(live_cancels.load() == 0, + "a peer that keeps sending audio is never cancelled"); +} + int main() { test_it_fires_when_nothing_touches_it(); test_touching_defers_it_indefinitely(); @@ -150,6 +214,8 @@ int main() { test_fired_survives_a_later_disarm(); test_the_destructor_joins(); test_a_firing_watchdog_still_joins_cleanly(); + test_only_a_frame_with_audio_counts(); + test_empty_frames_do_not_hold_the_lane(); if (failures) { fprintf(stderr, "%d check(s) failed\n", failures); return 1;