queue_relay.cpp
Rama main · réplica de lectura del repositorio original.
// SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0
#include "queue_room_internal.h"
namespace bridge {
void QueueRoom::Impl::fastPoll() {
fastUntil = std::chrono::steady_clock::now() + std::chrono::seconds(3);
nextPoll = {};
}
upnp::QueueOwnershipIO QueueRoom::Impl::ownershipIO() {
upnp::QueueOwnershipIO io;
io.queueTransport = "x-rincon-queue:" + room.uuid + "#0";
io.browse = [this] { return coordinator.browse(); };
io.owned = [this](const auto &uri) { return owned.count(uri) != 0; };
io.playlists = [this] { return coordinator.playlists(); };
io.save = [this](const auto &title) { return actions.save(title).response.value("AssignedObjectID"); };
io.deletePlaylist = [this](const auto &id) { coordinator.destroyPlaylist(id); };
io.transport = [this] {
return coordinator.transport("GetMediaInfo", {{"InstanceID", "0"}}).response.value("CurrentURI");
};
io.prepareTransport = [this] {
const auto ownQueue = "x-rincon-queue:" + room.uuid + "#0";
const auto uri = coordinator.transport("GetMediaInfo", {{"InstanceID", "0"}}).response.value("CurrentURI");
if (uri == ownQueue) return;
actions.selectQueue(room.uuid);
if (coordinator.transport("GetMediaInfo", {{"InstanceID", "0"}}).response.value("CurrentURI") != ownQueue)
throw std::runtime_error("queue-transport-switch-unconfirmed");
queueSelectedForAdmission = true;
logQueue("admission transport switched and confirmed before transport settings");
};
io.crossfade = [this] { return actions.crossfade().response.value("CrossfadeMode") == "1"; };
io.setCrossfade = [this](bool value) {
try {
actions.crossfade(value);
} catch (const std::exception &e) {
const std::string error = e.what();
if (releasing || queueSelectedForAdmission || error.find("fault=7") == std::string::npos)
throw;
// A setting was refused before our first switch, possibly after a source race.
actions.selectQueue(room.uuid);
if (coordinator.transport("GetMediaInfo", {{"InstanceID", "0"}}).response.value("CurrentURI") !=
"x-rincon-queue:" + room.uuid + "#0")
throw std::runtime_error("queue-transport-switch-unconfirmed");
queueSelectedForAdmission = true;
logQueue("admission retry after confirmed queue switch: " + error);
actions.crossfade(value); // Exactly one retry; a post-switch failure faults the room.
}
if (actions.crossfade().response.value("CrossfadeMode") != (value ? "1" : "0"))
throw std::runtime_error("queue-crossfade-unconfirmed");
};
io.remove = [this](auto update, auto first, auto count) { actions.remove(update, first, count); };
io.stop = [this] { actions.stop(); };
io.log = logQueue;
return io;
}
void QueueRoom::Impl::releaseWindowParent(uint64_t token, uint64_t before) {
if (before == UINT64_MAX) speakerNavFirst.erase(token);
std::lock_guard<std::mutex> lock(sourceMutex);
for (const auto &source : sourceTokens)
if (source.second == token && source.first < before && source.first < UINT64_MAX / 2)
queue.releaseParent(source.first);
}
void QueueRoom::Impl::release(bool stop) {
refreshUpcoming = false;
lmsNavigation = false;
navigationDeadline = {};
navigationItem.reset();
publishedWindowIndex.reset();
auto cleanupOwned = std::move(owned);
owned.clear();
std::set<uint64_t> parents;
for (const auto &item : items) {
parents.insert(item.owner ? item.owner : item.stream);
if (item.owner && !item.segmentOffset) parents.insert(item.stream);
}
for (auto parent : parents) queue.releaseParent(parent);
std::set<uint64_t> tokens;
{
std::lock_guard<std::mutex> lock(sourceMutex);
for (auto parent : parents) {
const auto token = sourceTokens.find(parent);
if (token != sourceTokens.end()) tokens.insert(token->second);
}
}
for (auto token : tokens) releaseWindowParent(token);
for (const auto &lazy : lazyItems) lazy.second->cancel();
lazyItems.clear();
selections.clear();
selectionOrder.clear();
speakerNavCommands.clear();
speakerNavFirst.clear();
replacementReasons.clear();
items.clear();
releasing = true;
ownership.release(!sourceAway,
[cleanupOwned = std::move(cleanupOwned)](const auto &uri) { return cleanupOwned.count(uri) != 0; }, stop);
releasing = false;
started = false;
waitingForPlay = false;
current = 0;
lastElapsed = 0;
lastPositionAt = {};
state = "STOPPED";
publish();
}
void QueueRoom::Impl::command(char c, uint64_t next) {
const bool navigationTimeout = c == 'Z';
if (navigationTimeout) c = 'q';
if (c == 'H') {
const auto snapshot = reader.read(host, mac);
const auto audible = std::find_if(items.begin(), items.end(), [&](const auto &i) { return i.stream == current; });
if (snapshot.mode == "pause" && audible != items.end() &&
snapshot.index < snapshot.entries.size() && snapshot.entries[snapshot.index].url == audible->sourceUrl) {
pausedFlushRetained = true;
logQueue("pause flush retained: LMS mode=pause same source, queue and spool kept");
return;
}
finitePause = false;
queue.flush();
command('q', next);
return;
}
if (c == 'M' || c == 'R') {
if (!roomStopped) {
if (c == 'M') {
const auto live = speaker->speaker();
std::printf("yeney: room is group member room=%s coordinator=%s; not playing\n",
room.name.c_str(), live.coordinator.c_str());
}
roomFault(c == 'M' ? "group-member" : "topology-unavailable", c == 'M');
}
return;
}
if (c == 'f') {
ownUntil = std::chrono::steady_clock::now() + std::chrono::seconds(3);
return;
}
if (c == 'F') {
deferredNext = next;
deferredStop = std::chrono::steady_clock::now() + std::chrono::milliseconds(250);
return;
}
if (c == 'C') {
deferredStop = {};
queue_audio::PlaylistSnapshot snapshot;
bool known = false;
try {
snapshot = reader.read(host, mac);
known = true;
} catch (const std::exception &e) {
logQueue("resume echo verification unavailable: " + std::string(e.what()));
}
const auto audible =
std::find_if(items.begin(), items.end(), [&](const auto &i) { return i.stream == current; });
const auto source = audible != items.end() && audible->owner ? audible->owner : current;
const auto oldRunElapsed = queue.audibleElapsedMs(source).value_or(0);
const bool sameClock = std::abs(int64_t(snapshot.timeMs) - int64_t(resumeSnapshot.timeMs)) <= 500;
// Native LMS can expose the new seek offset plus the preceding run's
// STAT clock until our cached STMs reanchors that same source. Verify
// that exact clock equation, rather than widening the seek tolerance.
const bool pendingNativeClock = pausedFlushRetained && pathReason == "remote-finite" &&
std::abs(int64_t(snapshot.timeMs) - int64_t(oldRunElapsed) - int64_t(resumeSnapshot.timeMs)) <= 500;
const bool same =
known && audible != items.end() && snapshot.index == resumeSnapshot.index &&
snapshot.index < snapshot.entries.size() && snapshot.index < resumeSnapshot.entries.size() &&
snapshot.entries[snapshot.index].url == audible->sourceUrl &&
snapshot.entries[snapshot.index].url == resumeSnapshot.entries[snapshot.index].url &&
(sameClock || pendingNativeClock);
logQueue("resume echo decision=" + std::string(same ? "adopt" : "replace") +
" lms_time_ms=" + std::to_string(snapshot.timeMs) +
" saved_time_ms=" + std::to_string(resumeSnapshot.timeMs) +
" old_run_elapsed_ms=" + std::to_string(oldRunElapsed) +
" pending_native_clock=" + std::to_string(pendingNativeClock));
pausedFlushRetained = false;
if (same) {
queue.restartCached();
queue.confirmCachedStart();
fastPoll();
logQueue("resume echo adopted: same LMS item/time, retained audio, no renderer restart");
} else {
queue.flush();
command('Z', next); // Existing cached-resume mismatch replacement.
queue.resumeReceivedStart();
logQueue("resume echo changed item/time: honour LMS selection");
}
return;
}
if (c == 'q') {
if (!navigationTimeout && started && !externalReleased && !sourceAway) {
// A controller navigation starts a fresh producer (q+s), unlike
// decode-ahead's natural s. Wait for the paired start; its status
// read verifies the target before any speaker edit.
{
std::lock_guard<std::mutex> lock(sourceMutex);
navigationFrom = publishedWindowIndex.value_or(playlist.index);
for (const auto &entry : items) if (entry.stream == current) {
const auto owner = entry.owner ? entry.owner : entry.stream;
const auto index = sourceIndices.find(owner);
if (index != sourceIndices.end()) {
const auto selection = selections.find(index->second);
if (selection == selections.end() || selection->second.phase != Selection::Producing)
navigationFrom = index->second;
}
}
navigationEdits = queueEdits;
lmsNavigation = true;
remoteNaturalSuccessor = false;
finitePause = false;
pausedFlushRetained = false;
handoffAt = {};
navigationFirst = next;
navigationItem.reset();
navigationDeadline = std::chrono::steady_clock::now() +
std::chrono::milliseconds(queue_audio::QueueWindow::settleMs);
stoppedPlay = true; // The next s describes snapshot.index itself.
ownUntil = std::chrono::steady_clock::now() + std::chrono::seconds(3);
return;
}
}
remoteNaturalSuccessor = false;
finitePause = false;
pausedFlushRetained = false;
handoffAt = {};
if (externalReleased)
stopAcknowledged = true;
expectedState = "STOPPED";
selecting = selectionKnown = false;
stoppedPlay = true;
firstStream = next;
if (!roomStopped) release(ownership.active() && !externalReleased);
faulted = false;
fault.clear();
playlist = {};
publish();
} else if (c == 's' && externalReleased) {
externalReleased = false;
sourceAway = false;
faulted = false;
fault.clear();
firstStream = next;
playlist = {};
} else if (c == 's' && state == "STOPPED" && (started || faulted)) {
if (started || faulted)
release();
firstStream = next;
playlist = {};
faulted = false;
fault.clear();
publish();
} else if (c == 'p' && started) {
expectedState = "PAUSED";
if (state != "PAUSED")
actions.pause();
} else if (c == 'u' && started) {
expectedState = "PLAYING";
if (state == "PAUSED" && !sourceAway) {
actions.play();
playAt = std::chrono::steady_clock::now();
waitingForPlay = true;
fastPoll();
}
}
}
void QueueRoom::Impl::append() {
if (!paths.isRelay() || faulted || externalReleased || (started && sourceAway))
return;
for (const auto &record : queue.snapshot()) {
const auto owner = record.parent ? record.parent : record.source.occurrence;
if (owner < firstStream || (record.source.occurrence > UINT64_MAX / 2 && !record.parent))
continue;
if (record.item && !record.incomplete.empty())
continue; // LMS cancelled this render, not a speaker rejection.
if (!record.fault.empty())
{ if (record.storageFault) throw std::runtime_error(record.fault); throw TrackFault(record.fault); }
if (!record.item || !record.incomplete.empty())
continue;
if (!record.parent) {
std::lock_guard<std::mutex> lock(sourceMutex);
const auto token = sourceTokens.find(owner);
if (token != sourceTokens.end()) {
const auto first = speakerNavFirst.find(token->second);
if (first != speakerNavFirst.end() && owner < first->second)
continue; // Our selection cancels decode-ahead; only its paired producer may bind.
}
if (token != sourceTokens.end() && !window.find(token->second)) {
queue.releaseParent(owner);
continue; // An LMS edit removed this producer before publication.
}
}
if (std::any_of(items.begin(), items.end(),
[&](const auto &i) { return i.stream == record.source.occurrence || i.owner == record.source.occurrence; }))
continue;
if (lmsNavigation && owner < navigationFirst)
continue; // A superseded producer cannot enlarge the controller's navigation window.
if (record.source.rate > queue.maxSampleRate())
throw TrackFault("queue-rate-unsupported rate=" + std::to_string(record.source.rate) +
" cap=" + std::to_string(queue.maxSampleRate()));
std::optional<uint64_t> length;
try {
length = record.item->contentLength();
} catch (const queue_audio::PublicationSuperseded &) {
continue; // Storage published this pending item after snapshot(); retry the new immutable item.
} catch (const queue_audio::ItemCancelled &) {
continue;
}
if (!length)
continue;
if (!record.fault.empty())
{ if (record.storageFault) throw std::runtime_error(record.fault); throw TrackFault(record.fault); }
if (!record.parent) {
std::lock_guard<std::mutex> lock(sourceMutex);
const auto token = sourceTokens.find(owner);
if (token != sourceTokens.end()) {
const auto lazy = lazyItems.find(token->second);
const auto selection = selections.find(sourceIndices.at(owner));
const auto speakerSelection = selection != selections.end() && selection->second.token == token->second &&
selection->second.phase == Selection::Producing && speakerNavCommands[token->second];
if (!lmsNavigation || owner < navigationFirst || token->second != navigationToken || speakerSelection) {
auto cached = std::find_if(items.begin(), items.end(), [&](const auto &i) {
const auto identity = sourceTokens.find(i.owner ? i.owner : i.stream);
return i.item && !i.segmentOffset && identity != sourceTokens.end() &&
identity->second == token->second;
});
if (cached != items.end() && record.source.occurrence < cached->stream)
continue; // A newer producer already owns this retained URI's clock.
if (cached != items.end() && cached->item == record.item && !cached->originMs) {
// After LMS Previous, natural decode-ahead may revisit
// a retained upcoming parent. Bind its new producer
// clock to the existing URI instead of appending it.
if (lmsNavigation && token->second == navigationToken && speakerSelection) {
lmsNavigation = navigationSelecting = false;
navigationDeadline = {};
navigationItem.reset();
logQueue("window pairing lms_index=" + std::to_string(sourceIndices.at(owner)) +
" decision=paired lms_commands=1");
}
if (!cached->owner) cached->owner = cached->stream;
if (current == cached->stream) current = record.source.occurrence;
cached->stream = record.source.occurrence;
selectionPublished(sourceIndices.at(owner));
windowDirty = true;
continue;
}
}
auto entry = std::find_if(items.begin(), items.end(), [&](const auto &i) {
return !i.item && sourceTokens[i.stream] == token->second;
});
if (!lmsNavigation && lazy != lazyItems.end() && entry != items.end()) {
if (!lazy->second->publish(record.item)) {
queue.releaseParent(owner);
selections.erase(sourceIndices.at(owner));
continue;
}
entry->stream = record.source.occurrence;
entry->item = record.item;
entry->frames = (*length - 44) / 6;
entry->originMs = record.originMs;
if (speakerSelection) {
const auto reason = replacementReasons.find(token->second);
if (reason != replacementReasons.end()) {
logQueue("window target re-produced lms_index=" + std::to_string(sourceIndices.at(owner)) +
" origin_ms=" + std::to_string(record.originMs) + " lms_commands=1 reason=" + reason->second);
logQueue("window pairing lms_index=" + std::to_string(sourceIndices.at(owner)) +
" decision=re-produced reason=" + reason->second + " lms_commands=1");
replacementReasons.erase(reason);
} else logQueue("window pairing lms_index=" + std::to_string(sourceIndices.at(owner)) +
" decision=paired lms_commands=1");
}
logQueue("lazy bound lms_index=" + std::to_string(sourceIndices[owner]) +
" lms_commands=" + std::to_string(speakerNavCommands[token->second]));
selectionPublished(sourceIndices.at(owner));
windowDirty = true;
continue;
}
}
}
queue_audio::PlaylistEntry track;
{
std::lock_guard<std::mutex> lock(sourceMutex);
auto found = sourceTracks.find(record.parent ? record.parent : record.source.occurrence);
if (found == sourceTracks.end())
throw std::runtime_error("queue-stream-metadata-missing");
track = found->second;
remote[record.source.occurrence] = false;
}
if (!ownership.active()) {
auto now = std::time(nullptr);
std::tm tm{};
localtime_r(&now, &tm);
char title[64];
std::strftime(title, sizeof title, "Before YeneY %Y-%m-%d %H:%M", &tm);
queueSelectedForAdmission = false;
ownership.acquire(title);
logQueue("ownership acquired");
publish();
}
const auto before = coordinator.browse();
std::vector<std::string> expected;
for (const auto &i : items)
expected.push_back(i.uri);
if (before.uris != expected)
throw std::runtime_error("queue-changed-while-owned");
upnp::QueueMetadata metadata;
// Fixed legacy route namespace, not a playback epoch. Stream
// occurrence IDs are unique for the lifetime of this connection.
try {
metadata.url = speaker->controllerUri() + route.publish(room.name, 1, 1, record.item);
} catch (const queue_audio::ItemCancelled &) {
continue;
}
metadata.session = session;
metadata.title = track.title;
metadata.artist = track.artist;
metadata.album = track.album;
metadata.artwork = track.artwork(host);
metadata.size = *length;
metadata.frames = (*length - 44) / 6;
metadata.rate = record.source.rate;
metadata.preciseDuration = record.estimated;
const auto changes = upnp::sanitizeXmlFields({{"title", &metadata.title},
{"artist", &metadata.artist},
{"album", &metadata.album},
{"artwork", &metadata.artwork}});
if (!changes.empty())
logQueue("metadata sanitized mode=queue occurrence=" + std::to_string(record.source.occurrence) +
" " + changes);
owned.insert(metadata.url); // include an enqueue whose reply is lost
size_t insertion = items.size();
if (record.parent) {
auto previous = std::find_if(items.rbegin(), items.rend(), [&](const auto &i) {
return i.stream == record.parent || i.owner == record.parent;
});
if (previous == items.rend()) throw std::runtime_error("continuation predecessor missing");
insertion = size_t(std::distance(items.begin(), previous.base()));
}
if (record.parent) logQueue("continuation insertion attempted parent=" + std::to_string(record.parent) +
" occurrence=" + std::to_string(record.source.occurrence));
upnp::SoapResult result;
const auto didl = upnp::queueDidl(metadata);
if (lmsNavigation && owner >= navigationFirst && !record.parent) {
std::lock_guard<std::mutex> lock(sourceMutex);
const auto identity = sourceTokens.find(owner);
if (identity != sourceTokens.end() && identity->second == navigationToken) {
// Admission can race the CLI selection. The generation is a
// lower bound; the occurrence token identifies its producer.
navigationFirst = owner;
navigationItem = Entry{record.source.occurrence, metadata.url, record.item, track.url,
metadata.frames, metadata.title, 0, 0, didl, record.originMs};
windowDirty = true;
nextWindowPoll = {};
return;
}
}
{
std::lock_guard<std::mutex> lock(mutex);
if (!commands.empty()) return; // A controller move supersedes this prepared append.
}
try {
result = actions.add(metadata.url, didl,
insertion == items.size() ? 0 : unsigned(insertion + 1));
} catch (const std::exception &e) {
if (record.parent) logQueue("continuation insertion failed parent=" + std::to_string(record.parent) +
" reason=" + e.what());
throw;
}
if (record.parent) {
double remaining = double(record.segmentOffset) / metadata.rate;
const auto playing = std::find_if(items.begin(), items.end(), [&](const auto &i) { return i.stream == current; });
if (playing != items.end() && (playing->stream == record.parent || playing->owner == record.parent))
remaining -= double(playing->segmentOffset) / metadata.rate + lastElapsed;
const auto margin = remaining - 10;
logQueue(std::string(margin < 0 ? "WARNING " : "") + "continuation inserted parent=" +
std::to_string(record.parent) + " decision=" + record.continuationDecision + " prefetch_margin_s=" + std::to_string(margin));
}
logQueue("item occurrence=" + std::to_string(record.source.occurrence) +
" title=" + RoomStatus::quote(metadata.title));
expected.insert(expected.begin() + insertion, metadata.url);
const auto after = coordinator.browse();
if (result.response.value("NumTracksAdded") != "1" || after.uris != expected ||
after.update == before.update)
{
if (record.parent) logQueue("continuation insertion failed reason=queue-enqueue-unconfirmed parent=" + std::to_string(record.parent));
throw std::runtime_error("queue-enqueue-unconfirmed");
}
if (record.parent) logQueue("continuation insertion confirmed parent=" + std::to_string(record.parent) +
" occurrence=" + std::to_string(record.source.occurrence));
items.insert(items.begin() + insertion, {record.source.occurrence, metadata.url, record.item, track.url,
metadata.frames, metadata.title, record.parent, record.segmentOffset, didl});
windowDirty = true;
publish();
if (!started) {
if (!queueSelectedForAdmission) actions.selectQueue(room.uuid);
actions.seekTrack(1);
expectedState = "PLAYING";
actions.play();
playAt = std::chrono::steady_clock::now();
waitingForPlay = true;
fastPoll();
started = true;
}
}
}
void QueueRoom::Impl::roomFault(const std::string &reason, bool member) {
if (roomStopped) return;
roomStopped = true;
logQueue("end decision=fault occurrence=" + std::to_string(current) + " reason=" + reason);
fault = reason;
faulted = true;
selecting = selectionKnown = false;
paths.cancelSelection();
if (!member) {
try { release(!externalReleased); }
catch (const std::exception &e) { logQueue("release failed=" + std::string(e.what())); }
}
if (member) {
externalReleased = true;
sourceAway = true;
release();
}
// A member must never write to either its old queue or the new coordinator.
expectedState = state = "STOPPED";
started = waitingForPlay = false;
externalReleased = true;
stopAcknowledged = false;
publish();
speakerCommand("stop", member ? "group-member" : reason == "track-fault-limit" ? reason : "room-fault");
}
void QueueRoom::Impl::trimHistory() {
if (!started || faulted || externalReleased || sourceAway || !current || lmsNavigation) return;
using Edits = queue_audio::WindowEdits;
std::vector<Edits::Parent> parents;
uint64_t audible = 0;
for (const auto &item : items) {
const auto parent = item.owner ? item.owner : item.stream;
if (parents.empty() || parents.back().token != parent) parents.push_back({parent, {}});
parents.back().uris.push_back(item.uri);
if (item.stream == current) audible = parent;
}
const auto playing = std::find_if(parents.begin(), parents.end(),
[audible](const auto &p) { return p.token == audible; });
if (playing == parents.end()) return;
const auto history = size_t(playing - parents.begin());
if (history <= queue_audio::QueueWindow::historyCount) return;
const auto count = history - queue_audio::QueueWindow::historyCount;
std::vector<Edits::Parent> wanted(parents.begin() + count, parents.end());
Edits::IO io;
io.browse = [this] {
const auto snapshot = coordinator.browse();
return Edits::Snapshot{snapshot.update, snapshot.uris};
};
io.remove = [this](auto update, auto first, auto n) { actions.remove(update, first, n); };
io.insert = [](auto, auto, auto) { throw std::logic_error("history trim attempted insertion"); };
io.log = [this, &parents](const auto &message) {
std::string indices;
std::lock_guard<std::mutex> lock(sourceMutex);
for (const auto &parent : parents) {
const auto found = sourceIndices.find(parent.token);
indices += (indices.empty() ? "" : ",") +
(found == sourceIndices.end() ? std::string("unknown") : std::to_string(found->second));
}
logQueue(message + " lms_indices=" + indices + " reason=history-limit");
};
Edits(std::move(io)).apply(parents, wanted, audible);
for (size_t n = 0; n < count; ++n) {
const auto token = parents[n].token;
uint64_t identity = 0;
{
std::lock_guard<std::mutex> lock(sourceMutex);
const auto found = sourceTokens.find(token);
if (found != sourceTokens.end()) identity = found->second;
}
if (identity) releaseWindowParent(identity);
for (const auto &item : items)
if (item.owner == token && !item.segmentOffset) queue.releaseParent(item.stream);
for (const auto &uri : parents[n].uris) owned.erase(uri);
items.erase(std::remove_if(items.begin(), items.end(), [token](const auto &item) {
return (item.owner ? item.owner : item.stream) == token;
}), items.end());
queue.releaseParent(token);
}
publish();
}
void QueueRoom::Impl::run() {
for (;;) {
std::deque<std::pair<char, uint64_t>> pending;
{
std::unique_lock<std::mutex> lock(mutex);
changed.wait_for(lock, std::chrono::milliseconds(100),
[&] { return closing || !commands.empty(); });
if (closing)
break;
pending.swap(commands);
}
try {
for (const auto &c : pending) {
if (c.first == 'L') activateLazy(c.second);
else if (c.first == 'I')
beginSelection(c.second);
else if (c.first == 'Q') {
if (externalReleased) stopAcknowledged = true;
selecting = selectionKnown = false;
handoffAt = {};
stoppedPlay = true;
state = "STOPPED";
publish();
} else
command(c.first, c.second);
}
selectPath();
if (navigationDeadline != std::chrono::steady_clock::time_point{} &&
std::chrono::steady_clock::now() >= navigationDeadline) {
navigationDeadline = {};
if (selecting || selectionDeferred.load() || std::any_of(selections.begin(), selections.end(),
[](const auto &selection) { return selection.second.phase != Selection::Published; })) {
navigationDeadline = std::chrono::steady_clock::now() +
std::chrono::milliseconds(queue_audio::QueueWindow::settleMs);
try {
const auto snapshot = reader.read(host, mac);
if (snapshot.mode == "stop") {
const auto fresh = reader.read(host, mac);
if (fresh.mode == "stop") {
command('Z', navigationFirst); // Confirmed user stop cancels owned requests.
if (snapshot.entries.empty() && fresh.entries.empty())
logQueue("window rebuild reason=clear lms_commands=0");
}
}
} catch (const std::exception &) { logQueue("window snapshot unavailable reason=cli-read"); }
} else {
// A standalone q is explicit LMS stop authority. Status can
// lag it, but an empty status alone is never clear authority.
try {
const auto snapshot = reader.read(host, mac);
if (snapshot.entries.empty()) {
const auto fresh = reader.read(host, mac);
if (fresh.entries.empty()) {
lmsNavigation = navigationSelecting = false;
logQueue("window rebuild reason=clear lms_commands=0");
logQueue("window lms-nav from_index=" + std::to_string(navigationFrom) +
" to_index=0 action=rebuild history=0 lms_commands=0 kind=clear");
} else logQueue("window snapshot rejected reason=unconfirmed-clear");
}
} catch (const std::exception &) { logQueue("window snapshot unavailable reason=cli-read"); }
command('Z', navigationFirst);
}
}
if (deferredStop != std::chrono::steady_clock::time_point{} &&
std::chrono::steady_clock::now() >= deferredStop) {
deferredStop = {};
resumeEchoUntil = 0;
resumeFlush = false;
queue.flush();
command('Z', deferredNext); // Resume echo expired without a start.
}
observe();
append();
advanceSelections();
trimHistory();
reconcileWindow();
} catch (const queue_audio::ItemCancelled &) {
// Producer cancellation raced a snapshot; observe the next run.
} catch (const std::exception &e) {
if (sourceAway && started && !externalReleased) {
externalStop("speaker-source-change");
continue;
}
if (roomStopped) continue;
fault = e.what();
logQueue("relay fault=" + fault);
const bool trackFault = dynamic_cast<const TrackFault *>(&e) != nullptr;
if (!trackFault) {
roomFault(fault, fault.compare(0, 17, "room-group-member") == 0);
continue;
}
const auto now = std::chrono::steady_clock::now();
while (!trackFaults.empty() && now - trackFaults.front() > std::chrono::seconds(10))
trackFaults.pop_front();
trackFaults.push_back(now);
if (trackFaults.size() >= 2) {
logQueue("track-fault-limit: two track faults within 10 s; stopping room");
roomFault("track-fault-limit");
continue;
}
faulted = true;
try { release(!externalReleased); }
catch (const std::exception &cleanup) { logQueue(cleanup.what()); }
queue.failQueueTrack();
publish();
}
}
if (roomStopped) return;
try {
release(ownership.active() && !externalReleased);
} catch (const std::exception &e) {
logQueue("release failed=" + std::string(e.what()));
}
}
} // namespace bridge