BUILD114 — исправление Agent / Mapping / Settings
This commit is contained in:
@@ -754,14 +754,20 @@ class VmixAgentHub:
|
||||
entries: list[dict[str, Any]],
|
||||
use_batch: bool,
|
||||
) -> dict[str, Any]:
|
||||
"""Deliver Mapping commands reliably, with chunking and adaptive fallback.
|
||||
"""BUILD113: send Mapping as simple ordered vmix.command frames, no ACK waits.
|
||||
|
||||
Large mappings are grouped by Input and sent in bounded batches. A timed
|
||||
out/failed batch is recursively split; a single failed command falls back
|
||||
to the legacy one-command transport. Mapping commands are idempotent
|
||||
(SetText/SetImage/SetColor/visibility), so this retry strategy is safe.
|
||||
Mapping used to use vmix.batch + ACK + recursive retries. In the field this
|
||||
could stall for a long time when ACK delivery/state drifted, even though
|
||||
ordinary interactive commands were healthy. Before every Mapping push we
|
||||
re-send the authoritative match.assign for this device, then deliver each
|
||||
concrete vMix command in strict WebSocket order. Success here means the
|
||||
frame reached the live Agent socket; Mapping never blocks on command.ack.
|
||||
"""
|
||||
outcomes: list[dict[str, Any] | None] = [None] * len(entries)
|
||||
input_groups = len({
|
||||
str(((entry.get("command") or {}).get("Input") or "")).strip()
|
||||
for entry in entries
|
||||
})
|
||||
stats = {
|
||||
"chunks_total": 0,
|
||||
"chunks_ok": 0,
|
||||
@@ -769,104 +775,97 @@ class VmixAgentHub:
|
||||
"packets_sent": 0,
|
||||
"retries": 0,
|
||||
"fallback_commands": 0,
|
||||
"input_groups": len({
|
||||
str(((entry.get("command") or {}).get("Input") or "")).strip()
|
||||
for entry in entries
|
||||
}),
|
||||
"commands_sent": 0,
|
||||
"input_groups": input_groups,
|
||||
"ack_waited": False,
|
||||
"transport": "sequential_no_ack",
|
||||
}
|
||||
|
||||
async def send_single(
|
||||
index: int, entry: dict[str, Any], previous_reason: str = "", *, count_fallback: bool = True
|
||||
) -> None:
|
||||
if count_fallback:
|
||||
stats["fallback_commands"] += 1
|
||||
try:
|
||||
ack = await self.send_vmix_command(
|
||||
device_id, assignment_id=assignment_id, match_id=match_id,
|
||||
command=entry.get("command") or {}, timeout=5.0,
|
||||
# Re-assert the current assignment immediately before Mapping. This heals
|
||||
# stale Agent-side match/assignment state and prevents match_mismatch /
|
||||
# assignment_mismatch from an earlier browser/session switch.
|
||||
assignment_payload: dict[str, Any] | None = None
|
||||
with self.database.session() as session:
|
||||
device = session.scalar(select(VmixDevice).where(VmixDevice.device_uuid == device_id))
|
||||
if device is not None:
|
||||
assignment = session.scalar(
|
||||
select(VmixAssignment).where(
|
||||
and_(
|
||||
VmixAssignment.device_id == device.id,
|
||||
VmixAssignment.assignment_key == assignment_id,
|
||||
VmixAssignment.active.is_(True),
|
||||
)
|
||||
)
|
||||
)
|
||||
if bool(ack.get("ok")):
|
||||
outcomes[index] = dict(ack)
|
||||
else:
|
||||
outcomes[index] = {
|
||||
"ok": False,
|
||||
"reason": str(ack.get("reason") or ack.get("error") or previous_reason or "vmix_error"),
|
||||
}
|
||||
except Exception as error:
|
||||
outcomes[index] = {
|
||||
"ok": False,
|
||||
"reason": str(getattr(error, "detail", error) or previous_reason or "vmix_error")[:300],
|
||||
if assignment is not None and str(assignment.game_external_id or "") == str(match_id or ""):
|
||||
assignment_payload = self._assignment_payload(assignment, device_id=device_id)
|
||||
|
||||
if assignment_payload is not None:
|
||||
assignment_delivered = await self.send(
|
||||
device_id,
|
||||
{"type": "match.assign", "protocol": AGENT_PROTOCOL_VERSION, **assignment_payload},
|
||||
)
|
||||
if not assignment_delivered:
|
||||
return {
|
||||
"results": [{"ok": False, "reason": "Agent сейчас offline"} for _ in entries],
|
||||
**stats,
|
||||
"assignment_refreshed": False,
|
||||
}
|
||||
|
||||
async def send_chunk(chunk: list[tuple[int, dict[str, Any]]]) -> None:
|
||||
if not chunk:
|
||||
return
|
||||
commands = [entry.get("command") or {} for _, entry in chunk]
|
||||
stats["packets_sent"] += 1
|
||||
try:
|
||||
ack = await self.send_vmix_batch(
|
||||
device_id, assignment_id=assignment_id, match_id=match_id, commands=commands,
|
||||
timeout=max(6.0, min(10.0, 4.0 + len(commands) * 0.06)),
|
||||
)
|
||||
ack_results = ack.get("results") if isinstance(ack.get("results"), list) else None
|
||||
if ack_results is None:
|
||||
if bool(ack.get("ok")):
|
||||
for index, _entry in chunk:
|
||||
outcomes[index] = {"ok": True}
|
||||
return
|
||||
raise RuntimeError(str(ack.get("reason") or ack.get("error") or "vmix_batch_error"))
|
||||
|
||||
failed: list[tuple[int, dict[str, Any], str]] = []
|
||||
for position, (index, entry) in enumerate(chunk):
|
||||
item_ack = ack_results[position] if position < len(ack_results) and isinstance(ack_results[position], dict) else {}
|
||||
if bool(item_ack.get("ok")):
|
||||
outcomes[index] = dict(item_ack)
|
||||
else:
|
||||
failed.append((
|
||||
index, entry,
|
||||
str(item_ack.get("reason") or item_ack.get("error") or "vmix_batch_error"),
|
||||
))
|
||||
# A batch ACK can contain isolated command failures. Retry only
|
||||
# those commands once through the single-command transport.
|
||||
for index, entry, reason in failed:
|
||||
stats["retries"] += 1
|
||||
await send_single(index, entry, reason)
|
||||
return
|
||||
except Exception as error:
|
||||
reason = str(getattr(error, "detail", error))[:300]
|
||||
if len(chunk) > 1:
|
||||
stats["retries"] += 1
|
||||
middle = max(1, len(chunk) // 2)
|
||||
await send_chunk(chunk[:middle])
|
||||
await send_chunk(chunk[middle:])
|
||||
return
|
||||
index, entry = chunk[0]
|
||||
stats["retries"] += 1
|
||||
await send_single(index, entry, reason)
|
||||
# Yield once so the Agent read loop can process match.assign before the
|
||||
# first vMix command while preserving WebSocket frame order.
|
||||
await asyncio.sleep(0)
|
||||
|
||||
async with self._device_mapping_apply_lock(device_id):
|
||||
if use_batch:
|
||||
chunks = self._mapping_batch_chunks(entries)
|
||||
stats["chunks_total"] = len(chunks)
|
||||
for chunk in chunks:
|
||||
await send_chunk(chunk)
|
||||
failed_now = sum(1 for index, _entry in chunk if outcomes[index] is None or not bool(outcomes[index].get("ok")))
|
||||
if failed_now == 0:
|
||||
stats["chunks_ok"] += 1
|
||||
else:
|
||||
stats["chunks_failed"] += 1
|
||||
else:
|
||||
stats["chunks_total"] = len(entries)
|
||||
for index, entry in enumerate(entries):
|
||||
await send_single(index, entry, count_fallback=False)
|
||||
if outcomes[index] is not None and bool(outcomes[index].get("ok")):
|
||||
stats["chunks_ok"] += 1
|
||||
else:
|
||||
stats["chunks_failed"] += 1
|
||||
for index, entry in enumerate(entries):
|
||||
command = entry.get("command") if isinstance(entry, dict) else None
|
||||
if not isinstance(command, dict) or not str(command.get("Function") or "").strip():
|
||||
outcomes[index] = {"ok": False, "reason": "invalid_mapping_command"}
|
||||
continue
|
||||
request_id = secrets.token_urlsafe(12)
|
||||
delivered = await self.send(
|
||||
device_id,
|
||||
{
|
||||
"type": "vmix.command",
|
||||
"protocol": AGENT_PROTOCOL_VERSION,
|
||||
"request_id": request_id,
|
||||
"device_id": device_id,
|
||||
"assignment_id": assignment_id,
|
||||
"match_id": match_id,
|
||||
"command": command,
|
||||
},
|
||||
)
|
||||
if not delivered:
|
||||
outcomes[index] = {
|
||||
"ok": False,
|
||||
"reason": "Agent сейчас offline",
|
||||
"delivery": "agent-websocket",
|
||||
"ack_waited": False,
|
||||
}
|
||||
# A failed WebSocket send means the connection is gone; there
|
||||
# is no value in attempting hundreds more fields.
|
||||
for tail in range(index + 1, len(entries)):
|
||||
outcomes[tail] = {
|
||||
"ok": False,
|
||||
"reason": "Agent сейчас offline",
|
||||
"delivery": "agent-websocket",
|
||||
"ack_waited": False,
|
||||
}
|
||||
break
|
||||
stats["commands_sent"] += 1
|
||||
outcomes[index] = {
|
||||
"ok": True,
|
||||
"delivery": "agent-websocket",
|
||||
"ack_waited": False,
|
||||
"request_id": request_id,
|
||||
}
|
||||
|
||||
return {
|
||||
"results": [item if isinstance(item, dict) else {"ok": False, "reason": "mapping_transport_no_ack"} for item in outcomes],
|
||||
"results": [
|
||||
item if isinstance(item, dict) else {"ok": False, "reason": "mapping_transport_not_sent"}
|
||||
for item in outcomes
|
||||
],
|
||||
**stats,
|
||||
"assignment_refreshed": assignment_payload is not None,
|
||||
}
|
||||
|
||||
async def receive_command_ack(self, device_id: str, message: dict[str, Any]) -> None:
|
||||
@@ -1242,7 +1241,7 @@ class VmixAgentHub:
|
||||
return result
|
||||
|
||||
use_batch = self._agent_supports_batch(agent_version)
|
||||
result["transport"] = "chunked_batch" if use_batch else "legacy"
|
||||
result["transport"] = "sequential_no_ack"
|
||||
transport = await self._send_mapping_entries_resilient(
|
||||
device_id,
|
||||
assignment_id=assignment_id,
|
||||
@@ -1256,6 +1255,9 @@ class VmixAgentHub:
|
||||
result["batch_packets_sent"] = int(transport.get("packets_sent") or 0)
|
||||
result["batch_retries"] = int(transport.get("retries") or 0)
|
||||
result["fallback_commands"] = int(transport.get("fallback_commands") or 0)
|
||||
result["mapping_commands_sent"] = int(transport.get("commands_sent") or 0)
|
||||
result["mapping_ack_waited"] = bool(transport.get("ack_waited", False))
|
||||
result["assignment_refreshed"] = bool(transport.get("assignment_refreshed", False))
|
||||
result["input_groups"] = int(transport.get("input_groups") or 0)
|
||||
result["batch_max_commands"] = MAPPING_BATCH_MAX_COMMANDS
|
||||
|
||||
@@ -1818,7 +1820,7 @@ class VmixAgentHub:
|
||||
|
||||
if prepared:
|
||||
use_batch = self._agent_supports_batch(agent_version)
|
||||
result["transport"] = "chunked_batch" if use_batch else "legacy"
|
||||
result["transport"] = "sequential_no_ack"
|
||||
transport = await self._send_mapping_entries_resilient(
|
||||
device_id, assignment_id=assignment_id, match_id=match_id,
|
||||
entries=prepared, use_batch=use_batch,
|
||||
@@ -1829,6 +1831,9 @@ class VmixAgentHub:
|
||||
result["batch_packets_sent"] = int(transport.get("packets_sent") or 0)
|
||||
result["batch_retries"] = int(transport.get("retries") or 0)
|
||||
result["fallback_commands"] = int(transport.get("fallback_commands") or 0)
|
||||
result["mapping_commands_sent"] = int(transport.get("commands_sent") or 0)
|
||||
result["mapping_ack_waited"] = bool(transport.get("ack_waited", False))
|
||||
result["assignment_refreshed"] = bool(transport.get("assignment_refreshed", False))
|
||||
result["input_groups"] = int(transport.get("input_groups") or 0)
|
||||
ack_results = transport.get("results") if isinstance(transport.get("results"), list) else []
|
||||
for pos, item in enumerate(prepared):
|
||||
|
||||
Reference in New Issue
Block a user