edison/chanora
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: address post-event-driven issues and client info parity (#16)

Browse Source 
* fix(ui): restore speaking status indicators

Speaking state (isSpeaking) is computed from voice activity timestamps
in the protocol layer and cannot be represented as a discrete delta.
The event-driven refactor removed periodic snapshot refreshes, causing
speaking indicators to go stale.

Adds a 750ms periodic snapshot refresh (matching SPEAKING_ACTIVITY_WINDOW)
while the audio stats timer is active (in-channel only). Structural
changes (moves, joins, leaves) are still handled by instant deltas.

* fix(ui): hide server query clients from delta joins

When a ServerQuery client sends a message, a ClientJoined delta fires.
Before PR#15 the periodic snapshot rebuild would include the SQ client
but the snapshot_view filter hid it. With deltas, the client persisted
in the local snapshot. Now ClientJoined deltas skip SQ clients entirely.

* fix(proto): log getconnectioninfo errors instead of silently discarding

Ping and packet loss showing 'Unknown' in the client info sheet is
caused by getconnectioninfo failures being silently swallowed. Now
logs the error with the client_id so the root cause can be diagnosed
(e.g. missing b_client_connectioninfo_view permission on the server).

Also logs clientgetvariables failures.

* fix(proto): refresh non-self client profiles before mapping

* feat(protocol): add ping deviation to client profiles

* chore(ui): regenerate Flutter bridge bindings for ping deviation

* fix(l10n): add ping deviation labels to client info

* feat(ui): show ping deviation in client info sheet
This commit is contained in:
Edison Jwa
2026-06-03 18:26:29 +09:00
committed by GitHub
parent 808324f374
commit 29afbb5e97
14 changed files with 558 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/chanora_flutter/lib/l10n/app_en.arb
+1
View File
@@ -227,6 +227,7 @@
"clientInfoOnline": "Online",
"clientInfoIdle": "Idle",
"clientInfoPing": "Ping",
"clientInfoPingDeviation": "Ping deviation",
"clientInfoAddress": "Address",
"clientInfoPacketLossClientToServer": "Packet loss C->S",
"clientInfoPacketLossServerToClient": "Packet loss S->C",
apps/chanora_flutter/lib/l10n/app_zh.arb
+1
View File
@@ -176,6 +176,7 @@
"clientInfoOnline": "在线时长",
"clientInfoIdle": "空闲",
"clientInfoPing": "延迟",
"clientInfoPingDeviation": "延迟偏差",
"clientInfoAddress": "地址",
"clientInfoPacketLossClientToServer": "丢包 C->S",
"clientInfoPacketLossServerToClient": "丢包 S->C",
apps/chanora_flutter/lib/l10n/generated/app_localizations.dart
+6
View File
@@ -1069,6 +1069,12 @@ abstract class AppL10n {
/// **'Ping'**
String get clientInfoPing;
/// No description provided for @clientInfoPingDeviation.
///
/// In en, this message translates to:
/// **'Ping deviation'**
String get clientInfoPingDeviation;
/// No description provided for @clientInfoAddress.
///
/// In en, this message translates to:
apps/chanora_flutter/lib/l10n/generated/app_localizations_en.dart
+3
View File
@@ -543,6 +543,9 @@ class AppL10nEn extends AppL10n {
@override
String get clientInfoPing => 'Ping';
@override
String get clientInfoPingDeviation => 'Ping deviation';
@override
String get clientInfoAddress => 'Address';
apps/chanora_flutter/lib/l10n/generated/app_localizations_zh.dart
+3
View File
@@ -531,6 +531,9 @@ class AppL10nZh extends AppL10n {
@override
String get clientInfoPing => '延迟';
@override
String get clientInfoPingDeviation => '延迟偏差';
@override
String get clientInfoAddress => '地址';
apps/chanora_flutter/lib/main.dart
+13
View File
@@ -809,6 +809,7 @@ class _BetaHomeState extends State<_BetaHome> with WidgetsBindingObserver {
return updated;
});
case rust.BridgeEvent_ClientJoined(:final clientId, :final channelId, :final name, :final inputMuted, :final outputMuted, :final isServerQuery, :final talkPower, :final talkPowerGranted):
if (!isServerQuery) {
_applyClientAdd(rust.BridgeClient(
id: clientId,
channel: channelId,
@@ -820,6 +821,7 @@ class _BetaHomeState extends State<_BetaHome> with WidgetsBindingObserver {
talkPower: talkPower,
talkPowerGranted: talkPowerGranted,
));
}
case rust.BridgeEvent_ClientLeft(:final clientId):
_applyClientRemove(clientId);
case rust.BridgeEvent_ClientUpdated(:final clientId, :final inputMuted, :final outputMuted, :final isServerQuery, :final talkPower, :final talkPowerGranted):
@@ -871,14 +873,25 @@ class _BetaHomeState extends State<_BetaHome> with WidgetsBindingObserver {
if (_serverReachable) _ensureStatsTimer();
}
DateTime? _lastSpeakingRefresh;
static const _speakingRefreshInterval = Duration(milliseconds: 750);
void _ensureStatsTimer() {
if (_statsTimer != null) return;
_lastSpeakingRefresh = null;
_statsTimer = Timer.periodic(const Duration(milliseconds: 250), (_) async {
try {
final s = await rust.audioStats();
if (!mounted) return;
setState(() => _audioStats = s);
} catch (_) {}
final now = DateTime.now();
if (_inChannel &&
(_lastSpeakingRefresh == null ||
now.difference(_lastSpeakingRefresh!) >= _speakingRefreshInterval)) {
_lastSpeakingRefresh = now;
unawaited(_refreshSnapshot(recordActivity: false, reportErrors: false));
}
});
}
apps/chanora_flutter/lib/src/rust/api.dart
+6
View File
@@ -917,6 +917,9 @@ class BridgeClientProfile {
/// Current ping in milliseconds.
final PlatformInt64? pingMilliseconds;
/// Current ping deviation in milliseconds.
final PlatformInt64? pingDeviationMilliseconds;
/// Client address. Empty when permission-gated.
final String clientAddress;
@@ -963,6 +966,7 @@ class BridgeClientProfile {
this.onlineSeconds,
this.idleMilliseconds,
this.pingMilliseconds,
this.pingDeviationMilliseconds,
required this.clientAddress,
required this.serverGroups,
required this.channelGroup,
@@ -992,6 +996,7 @@ class BridgeClientProfile {
onlineSeconds.hashCode ^
idleMilliseconds.hashCode ^
pingMilliseconds.hashCode ^
pingDeviationMilliseconds.hashCode ^
clientAddress.hashCode ^
serverGroups.hashCode ^
channelGroup.hashCode ^
@@ -1023,6 +1028,7 @@ class BridgeClientProfile {
onlineSeconds == other.onlineSeconds &&
idleMilliseconds == other.idleMilliseconds &&
pingMilliseconds == other.pingMilliseconds &&
pingDeviationMilliseconds == other.pingDeviationMilliseconds &&
clientAddress == other.clientAddress &&
serverGroups == other.serverGroups &&
channelGroup == other.channelGroup &&
apps/chanora_flutter/lib/src/rust/frb_generated.dart
+18 -12
View File
@@ -1844,8 +1844,8 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi {
BridgeClientProfile dco_decode_bridge_client_profile(dynamic raw) {
// Codec=Dco (DartCObject based), see doc to use other codecs
final arr = raw as List<dynamic>;
if (arr.length != 25)
throw Exception('unexpected arr length: expect 25 but see ${arr.length}');
if (arr.length != 26)
throw Exception('unexpected arr length: expect 26 but see ${arr.length}');
return BridgeClientProfile(
id: dco_decode_u_64(arr[0]),
channel: dco_decode_u_64(arr[1]),
@@ -1862,16 +1862,17 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi {
onlineSeconds: dco_decode_opt_box_autoadd_i_64(arr[12]),
idleMilliseconds: dco_decode_opt_box_autoadd_i_64(arr[13]),
pingMilliseconds: dco_decode_opt_box_autoadd_i_64(arr[14]),
clientAddress: dco_decode_String(arr[15]),
serverGroups: dco_decode_list_String(arr[16]),
channelGroup: dco_decode_String(arr[17]),
avatarPath: dco_decode_String(arr[18]),
bytesDownloadedMonth: dco_decode_opt_box_autoadd_u_64(arr[19]),
bytesUploadedMonth: dco_decode_opt_box_autoadd_u_64(arr[20]),
bytesDownloadedTotal: dco_decode_opt_box_autoadd_u_64(arr[21]),
bytesUploadedTotal: dco_decode_opt_box_autoadd_u_64(arr[22]),
packetLossClientToServerTotal: dco_decode_opt_box_autoadd_f_32(arr[23]),
packetLossServerToClientTotal: dco_decode_opt_box_autoadd_f_32(arr[24]),
pingDeviationMilliseconds: dco_decode_opt_box_autoadd_i_64(arr[15]),
clientAddress: dco_decode_String(arr[16]),
serverGroups: dco_decode_list_String(arr[17]),
channelGroup: dco_decode_String(arr[18]),
avatarPath: dco_decode_String(arr[19]),
bytesDownloadedMonth: dco_decode_opt_box_autoadd_u_64(arr[20]),
bytesUploadedMonth: dco_decode_opt_box_autoadd_u_64(arr[21]),
bytesDownloadedTotal: dco_decode_opt_box_autoadd_u_64(arr[22]),
bytesUploadedTotal: dco_decode_opt_box_autoadd_u_64(arr[23]),
packetLossClientToServerTotal: dco_decode_opt_box_autoadd_f_32(arr[24]),
packetLossServerToClientTotal: dco_decode_opt_box_autoadd_f_32(arr[25]),
);
}
@@ -2577,6 +2578,9 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi {
var var_onlineSeconds = sse_decode_opt_box_autoadd_i_64(deserializer);
var var_idleMilliseconds = sse_decode_opt_box_autoadd_i_64(deserializer);
var var_pingMilliseconds = sse_decode_opt_box_autoadd_i_64(deserializer);
var var_pingDeviationMilliseconds = sse_decode_opt_box_autoadd_i_64(
deserializer,
);
var var_clientAddress = sse_decode_String(deserializer);
var var_serverGroups = sse_decode_list_String(deserializer);
var var_channelGroup = sse_decode_String(deserializer);
@@ -2611,6 +2615,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi {
onlineSeconds: var_onlineSeconds,
idleMilliseconds: var_idleMilliseconds,
pingMilliseconds: var_pingMilliseconds,
pingDeviationMilliseconds: var_pingDeviationMilliseconds,
clientAddress: var_clientAddress,
serverGroups: var_serverGroups,
channelGroup: var_channelGroup,
@@ -3436,6 +3441,7 @@ class RustLibApiImpl extends RustLibApiImplPlatform implements RustLibApi {
sse_encode_opt_box_autoadd_i_64(self.onlineSeconds, serializer);
sse_encode_opt_box_autoadd_i_64(self.idleMilliseconds, serializer);
sse_encode_opt_box_autoadd_i_64(self.pingMilliseconds, serializer);
sse_encode_opt_box_autoadd_i_64(self.pingDeviationMilliseconds, serializer);
sse_encode_String(self.clientAddress, serializer);
sse_encode_list_String(self.serverGroups, serializer);
sse_encode_String(self.channelGroup, serializer);
apps/chanora_flutter/lib/widgets/client_info_sheet.dart
+5
View File
@@ -214,6 +214,11 @@ class _ClientInfoContent extends StatelessWidget {
l10n.clientInfoPing,
_formatMilliseconds(profile.pingMilliseconds, l10n),
),
if (profile.pingDeviationMilliseconds != null)
_InfoRowData(
l10n.clientInfoPingDeviation,
_formatMilliseconds(profile.pingDeviationMilliseconds, l10n),
),
_InfoRowData(
l10n.clientInfoAddress,
_emptyAsHidden(profile.clientAddress, l10n),
apps/chanora_flutter/test/widgets/client_info_sheet_test.dart
+48
View File
@@ -20,6 +20,7 @@ void main() {
onlineSeconds: 3661,
idleMilliseconds: 42000,
pingMilliseconds: 38,
pingDeviationMilliseconds: 7,
clientAddress: '203.0.113.24',
serverGroups: const ['Admin', 'Talk Power'],
channelGroup: 'Guest',
@@ -77,9 +78,56 @@ void main() {
expect(find.text('Connection'), findsOneWidget);
expect(find.text('1h 1m 1s'), findsOneWidget);
expect(find.text('42.00 s'), findsOneWidget);
expect(find.text('7 ms'), findsOneWidget);
expect(find.text('203.0.113.24'), findsOneWidget);
});
testWidgets('hides ping deviation row when the value is absent', (tester) async {
await tester.pumpWidget(
MaterialApp(
localizationsDelegates: AppL10n.localizationsDelegates,
supportedLocales: AppL10n.supportedLocales,
home: Scaffold(
body: SizedBox(
height: 500,
child: ClientInfoSheet(
clientName: 'Bob',
loadProfile: () async => rust.BridgeClientProfile(
id: BigInt.from(101),
channel: BigInt.from(7),
name: 'Bob',
uniqueId: 'client-unique-id',
databaseId: BigInt.from(55),
countryCode: 'US',
description: 'Operator',
version: '3.6.2',
platform: 'Windows',
onlineSeconds: 3661,
idleMilliseconds: 42000,
pingMilliseconds: 38,
clientAddress: '203.0.113.24',
serverGroups: const ['Admin'],
channelGroup: 'Guest',
avatarPath: '/avatar_aabbcc',
bytesDownloadedMonth: BigInt.from(2048),
bytesUploadedMonth: BigInt.from(4096),
bytesDownloadedTotal: BigInt.from(1048576),
bytesUploadedTotal: BigInt.from(2097152),
packetLossClientToServerTotal: 0.0123,
packetLossServerToClientTotal: 0.0456,
),
),
),
),
),
);
await tester.pumpAndSettle();
expect(find.text('Ping deviation'), findsNothing);
expect(find.text('7 ms'), findsNothing);
});
testWidgets('loading and error states use localized copy', (tester) async {
await tester.pumpWidget(
MaterialApp(
crates/chanora_bridge/src/api.rs
+3
View File
@@ -473,6 +473,8 @@ pub struct BridgeClientProfile {
pub idle_milliseconds: Option<i64>,
/// Current ping in milliseconds.
pub ping_milliseconds: Option<i64>,
/// Current ping deviation in milliseconds.
pub ping_deviation_milliseconds: Option<i64>,
/// Client address. Empty when permission-gated.
pub client_address: String,
/// Resolved server group names.
@@ -534,6 +536,7 @@ impl From<chanora_core::ClientProfile> for BridgeClientProfile {
online_seconds: profile.online_seconds,
idle_milliseconds: profile.idle_milliseconds,
ping_milliseconds: profile.ping_milliseconds,
ping_deviation_milliseconds: profile.ping_deviation_milliseconds,
client_address: profile.client_address,
server_groups: profile.server_groups,
channel_group: profile.channel_group,
crates/chanora_bridge/src/frb_generated.rs
+6
View File
@@ -2052,6 +2052,7 @@ impl SseDecode for crate::api::BridgeClientProfile {
let mut var_onlineSeconds = <Option<i64>>::sse_decode(deserializer);
let mut var_idleMilliseconds = <Option<i64>>::sse_decode(deserializer);
let mut var_pingMilliseconds = <Option<i64>>::sse_decode(deserializer);
let mut var_pingDeviationMilliseconds = <Option<i64>>::sse_decode(deserializer);
let mut var_clientAddress = <String>::sse_decode(deserializer);
let mut var_serverGroups = <Vec<String>>::sse_decode(deserializer);
let mut var_channelGroup = <String>::sse_decode(deserializer);
@@ -2078,6 +2079,7 @@ impl SseDecode for crate::api::BridgeClientProfile {
online_seconds: var_onlineSeconds,
idle_milliseconds: var_idleMilliseconds,
ping_milliseconds: var_pingMilliseconds,
ping_deviation_milliseconds: var_pingDeviationMilliseconds,
client_address: var_clientAddress,
server_groups: var_serverGroups,
channel_group: var_channelGroup,
@@ -3077,6 +3079,9 @@ impl flutter_rust_bridge::IntoDart for crate::api::BridgeClientProfile {
self.online_seconds.into_into_dart().into_dart(),
self.idle_milliseconds.into_into_dart().into_dart(),
self.ping_milliseconds.into_into_dart().into_dart(),
self.ping_deviation_milliseconds
.into_into_dart()
.into_dart(),
self.client_address.into_into_dart().into_dart(),
self.server_groups.into_into_dart().into_dart(),
self.channel_group.into_into_dart().into_dart(),
@@ -3835,6 +3840,7 @@ impl SseEncode for crate::api::BridgeClientProfile {
<Option<i64>>::sse_encode(self.online_seconds, serializer);
<Option<i64>>::sse_encode(self.idle_milliseconds, serializer);
<Option<i64>>::sse_encode(self.ping_milliseconds, serializer);
<Option<i64>>::sse_encode(self.ping_deviation_milliseconds, serializer);
<String>::sse_encode(self.client_address, serializer);
<Vec<String>>::sse_encode(self.server_groups, serializer);
<String>::sse_encode(self.channel_group, serializer);
crates/chanora_protocol/src/adapter.rs
+439 -211
View File
@@ -31,8 +31,8 @@ use tsclientlib::data::{self, Channel, Client};
use tsclientlib::messages::s2c::{InClientDbInfoPart, InMessage};
use tsclientlib::prelude::*;
use tsclientlib::{
ChannelId as TsChannelId, ClientId as TsClientId, Connection, DisconnectOptions, Identity,
MessageHandle, OutCommandExt, StreamItem, Version,
ChannelId as TsChannelId, ClientId as TsClientId, Connection, ConnectionStats,
DisconnectOptions, Identity, MessageHandle, OutCommandExt, StreamItem, Version,
};
use tsproto_packets::packets::{Direction, Flags, InAudioBuf, OutCommand, OutPacket, PacketType};
use tsproto_types::ClientType;
@@ -45,6 +45,16 @@ use crate::ProtocolError;
const SPEAKING_ACTIVITY_WINDOW: Duration = Duration::from_millis(750);
const INBOUND_VOICE_SEND_TIMEOUT: Duration = Duration::from_millis(40);
const PROFILE_REFRESH_RESULT_TIMEOUT: Duration = Duration::from_secs(3);
type PendingMoves = HashMap<
MessageHandle,
(
u64,
Option<oneshot::Sender<Result<(), ProtocolError>>>,
std::time::Instant,
),
>;
#[derive(Debug, PartialEq, Eq)]
enum SendTimeoutError<T> {
@@ -652,14 +662,7 @@ async fn connection_task(
// resolve the oneshot back to the caller. Entries also carry a
// deadline so a server that never replies doesn't leak the
// reply channel — at most 3 s of pending state per move.
let mut pending_moves: HashMap<
MessageHandle,
(
u64,
Option<oneshot::Sender<Result<(), ProtocolError>>>,
std::time::Instant,
),
> = HashMap::new();
let mut pending_moves: PendingMoves = HashMap::new();
let mut voice_activity: HashMap<u64, Instant> = HashMap::new();
// Main loop: pump events, service requests, forward voice.
@@ -680,157 +683,16 @@ async fn connection_task(
Ok(Some(Ok(item))) => {
match item {
StreamItem::Audio(buf) => {
let from = packet_sender_id(&buf);
if let Some(from) = from {
voice_activity.insert(from, Instant::now());
let inbound = InboundVoice {
from_client: from,
packet: buf,
};
match send_with_timeout(
&voice_in_tx,
inbound,
INBOUND_VOICE_SEND_TIMEOUT,
)
.await
{
Ok(()) => {}
Err(SendTimeoutError::Timeout(_)) => {
warn!(
target: "chanora_protocol",
from_client = from,
timeout_ms = INBOUND_VOICE_SEND_TIMEOUT.as_millis() as u64,
"inbound voice queue stayed full; dropping packet"
);
handle_audio_stream_item(&voice_in_tx, &mut voice_activity, buf).await;
}
Err(SendTimeoutError::Closed(_)) => {
warn!(
target: "chanora_protocol",
from_client = from,
"inbound voice consumer closed; dropping packet"
);
}
}
}
}
StreamItem::BookEvents(events) => {
for ev in events {
if let tsclientlib::events::Event::PropertyChanged {
id: ts_bookkeeping::events::PropertyId::ClientChannel(client_id),
..
} = &ev
{
let own_client = con.get_state().ok().map(|state| state.own_client);
if let Some(state) = con.get_state().ok() {
if let Some(client) = state.clients.get(client_id) {
let _ = delta_tx.try_send(ProtocolDelta::ClientMoved {
client_id: client_id.0 as u64,
new_channel_id: client.channel.0,
});
}
}
if own_client == Some(*client_id) {
let current_channel = con.get_state().ok().and_then(|state| {
state.clients.get(client_id).map(|client| client.channel.0)
});
if let Some(current_channel) = current_channel {
let matched: Vec<MessageHandle> = pending_moves
.iter()
.filter_map(|(handle, (target_channel, _, _))| {
if *target_channel == current_channel {
Some(*handle)
} else {
None
}
})
.collect();
for handle in matched {
if let Some((_, reply, _)) =
pending_moves.remove(&handle)
{
info!(
target: "chanora_protocol",
channel_id = current_channel,
"client_move resolved by authoritative self channel change"
);
if let Some(reply) = reply {
let _ = reply.send(Ok(()));
}
}
}
}
}
}
if let Some(activity) = format_server_activity(&con, &ev) {
let _ = activity_tx.try_send(ServerActivity { message: activity });
}
forward_delta(&con, &ev, &delta_tx);
if let tsclientlib::events::Event::Message {
target,
invoker,
message,
} = ev
{
let mapped = match target {
tsclientlib::MessageTarget::Server => MessageTarget::Server,
tsclientlib::MessageTarget::Channel => MessageTarget::Channel,
tsclientlib::MessageTarget::Client(id) => {
MessageTarget::Client(id.0 as u64)
}
tsclientlib::MessageTarget::Poke(id) => {
MessageTarget::Poke(id.0 as u64)
}
};
let _ = chat_tx.try_send(ChatMessage {
sender_id: ClientId(invoker.id.0 as u64),
sender_name: sanitize(&invoker.name),
message: sanitize(&message),
target: mapped,
});
}
}
}
StreamItem::MessageResult(handle, result) => {
if let Some((_target_channel, reply, _deadline)) =
pending_moves.remove(&handle)
{
let mapped = match result {
Ok(()) => Ok(()),
Err(cmd_err) => {
// tsclientlib's CommandError carries a
// typed `TsError` (the canonical TS3
// error code) plus an optional missing
// permission. We convert to our typed
// ProtocolError::ServerRejected so the
// upper layers can render a localised
// explanation by code instead of a
// generic backend string.
let code = cmd_err.error as u32;
let message = cmd_err.error.to_string();
info!(
target: "chanora_protocol",
code,
message = %message,
"server rejected client_move"
);
Err(ProtocolError::ServerRejected { code, message })
}
};
if let Some(reply) = reply {
let _ = reply.send(mapped);
} else if let Err(err) = mapped {
info!(
target: "chanora_protocol",
error = %err,
"client_move completed in background with error"
);
}
}
}
_ => { /* book / message / other events: ignore */ }
other => handle_non_audio_stream_item(
&con,
other,
&chat_tx,
&activity_tx,
&delta_tx,
&mut pending_moves,
),
}
}
Ok(Some(Err(e))) => {
@@ -926,7 +788,17 @@ async fn connection_task(
let _ = reply.send(r);
}
Ok(Request::FetchClientProfile { client_id, reply }) => {
let r = fetch_client_profile(&mut con, client_id).await;
let r = fetch_client_profile(
&mut con,
client_id,
&voice_in_tx,
&chat_tx,
&activity_tx,
&delta_tx,
&mut pending_moves,
&mut voice_activity,
)
.await;
let _ = reply.send(r);
}
Ok(Request::Disconnect(reply)) => {
@@ -947,6 +819,154 @@ async fn connection_task(
}
}
async fn handle_audio_stream_item(
voice_in_tx: &mpsc::Sender<InboundVoice>,
voice_activity: &mut HashMap<u64, Instant>,
buf: InAudioBuf,
) {
let from = packet_sender_id(&buf);
if let Some(from) = from {
voice_activity.insert(from, Instant::now());
let inbound = InboundVoice {
from_client: from,
packet: buf,
};
match send_with_timeout(voice_in_tx, inbound, INBOUND_VOICE_SEND_TIMEOUT).await {
Ok(()) => {}
Err(SendTimeoutError::Timeout(_)) => {
warn!(
target: "chanora_protocol",
from_client = from,
timeout_ms = INBOUND_VOICE_SEND_TIMEOUT.as_millis() as u64,
"inbound voice queue stayed full; dropping packet"
);
}
Err(SendTimeoutError::Closed(_)) => {
warn!(
target: "chanora_protocol",
from_client = from,
"inbound voice consumer closed; dropping packet"
);
}
}
}
}
fn handle_non_audio_stream_item(
con: &Connection,
item: StreamItem,
chat_tx: &mpsc::Sender<ChatMessage>,
activity_tx: &mpsc::Sender<ServerActivity>,
delta_tx: &mpsc::Sender<ProtocolDelta>,
pending_moves: &mut PendingMoves,
) {
match item {
StreamItem::BookEvents(events) => {
for ev in events {
if let tsclientlib::events::Event::PropertyChanged {
id: ts_bookkeeping::events::PropertyId::ClientChannel(client_id),
..
} = &ev
{
let own_client = con.get_state().ok().map(|state| state.own_client);
if let Some(state) = con.get_state().ok() {
if let Some(client) = state.clients.get(client_id) {
let _ = delta_tx.try_send(ProtocolDelta::ClientMoved {
client_id: client_id.0 as u64,
new_channel_id: client.channel.0,
});
}
}
if own_client == Some(*client_id) {
let current_channel = con.get_state().ok().and_then(|state| {
state.clients.get(client_id).map(|client| client.channel.0)
});
if let Some(current_channel) = current_channel {
let matched: Vec<MessageHandle> = pending_moves
.iter()
.filter_map(|(handle, (target_channel, _, _))| {
if *target_channel == current_channel {
Some(*handle)
} else {
None
}
})
.collect();
for handle in matched {
if let Some((_, reply, _)) = pending_moves.remove(&handle) {
info!(
target: "chanora_protocol",
channel_id = current_channel,
"client_move resolved by authoritative self channel change"
);
if let Some(reply) = reply {
let _ = reply.send(Ok(()));
}
}
}
}
}
}
if let Some(activity) = format_server_activity(con, &ev) {
let _ = activity_tx.try_send(ServerActivity { message: activity });
}
forward_delta(con, &ev, delta_tx);
if let tsclientlib::events::Event::Message {
target,
invoker,
message,
} = ev
{
let mapped = match target {
tsclientlib::MessageTarget::Server => MessageTarget::Server,
tsclientlib::MessageTarget::Channel => MessageTarget::Channel,
tsclientlib::MessageTarget::Client(id) => MessageTarget::Client(id.0 as u64),
tsclientlib::MessageTarget::Poke(id) => MessageTarget::Poke(id.0 as u64),
};
let _ = chat_tx.try_send(ChatMessage {
sender_id: ClientId(invoker.id.0 as u64),
sender_name: sanitize(&invoker.name),
message: sanitize(&message),
target: mapped,
});
}
}
}
StreamItem::MessageResult(handle, result) => {
if let Some((_target_channel, reply, _deadline)) = pending_moves.remove(&handle) {
let mapped = match result {
Ok(()) => Ok(()),
Err(cmd_err) => {
let code = cmd_err.error as u32;
let message = cmd_err.error.to_string();
info!(
target: "chanora_protocol",
code,
message = %message,
"server rejected client_move"
);
Err(ProtocolError::ServerRejected { code, message })
}
};
if let Some(reply) = reply {
let _ = reply.send(mapped);
} else if let Err(err) = mapped {
info!(
target: "chanora_protocol",
error = %err,
"client_move completed in background with error"
);
}
}
}
StreamItem::Audio(_) => unreachable!("audio handled separately"),
_ => {}
}
}
async fn resolve_server_socket(address: &str) -> Result<SocketAddr, ProtocolError> {
let resolver = ChanoraResolver::new().map_err(|err| ProtocolError::DnsFailed {
host: address.to_string(),
@@ -1095,37 +1115,16 @@ fn send_text_to_mode(
async fn fetch_client_profile(
con: &mut Connection,
client_id: u64,
voice_in_tx: &mpsc::Sender<InboundVoice>,
chat_tx: &mpsc::Sender<ChatMessage>,
activity_tx: &mpsc::Sender<ServerActivity>,
delta_tx: &mpsc::Sender<ProtocolDelta>,
pending_moves: &mut PendingMoves,
voice_activity: &mut HashMap<u64, Instant>,
) -> Result<ClientProfile, ProtocolError> {
let target_id = TsClientId(client_id as u16);
let (database_id, uid_b64) = {
let state = con
.get_state()
.map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;
let client = find_client_by_id(state.clients.values(), client_id)?;
(
client.database_id,
client.uid.as_ref().map(|uid| uid_to_b64(uid.as_ref())),
)
};
let _ = request_messages(con, build_command("servergrouplist", &[], &[])).await;
let _ = request_messages(con, build_command("channelgrouplist", &[], &[])).await;
let _ = request_messages(
con,
build_command(
"clientgetvariables",
&[("clid", client_id.to_string())],
&[],
),
)
.await;
let _ = request_messages(
con,
build_command("getconnectioninfo", &[("clid", client_id.to_string())], &[]),
)
.await;
let db_info = request_client_db_info(con, database_id).await.ok();
let (database_id, uid_b64, has_optional, has_connection, is_own, needs_server_groups, needs_channel_groups) = {
let state = con
.get_state()
.map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;
@@ -1133,8 +1132,130 @@ async fn fetch_client_profile(
.clients
.get(&target_id)
.ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found")))?;
(
client.database_id,
client.uid.as_ref().map(|uid| uid_to_b64(uid.as_ref())),
client.optional_data.is_some(),
client.connection_data.is_some(),
state.own_client == target_id,
state.server_groups.is_empty(),
state.channel_groups.is_empty(),
)
};
let refresh_plan = client_profile_refresh_plan(
is_own,
has_optional,
has_connection,
needs_server_groups,
needs_channel_groups,
);
if refresh_plan.needs_server_groups {
let _ = request_messages(
con,
build_command("servergrouplist", &[], &[]),
voice_in_tx,
chat_tx,
activity_tx,
delta_tx,
pending_moves,
voice_activity,
)
.await;
}
if refresh_plan.needs_channel_groups {
let _ = request_messages(
con,
build_command("channelgrouplist", &[], &[]),
voice_in_tx,
chat_tx,
activity_tx,
delta_tx,
pending_moves,
voice_activity,
)
.await;
}
if refresh_plan.needs_client_variables {
if let Err(e) = request_messages(
con,
build_command(
"clientgetvariables",
&[("clid", client_id.to_string())],
&[],
),
voice_in_tx,
chat_tx,
activity_tx,
delta_tx,
pending_moves,
voice_activity,
)
.await
{
warn!(
target: "chanora_protocol",
client_id,
error = %e,
"clientgetvariables failed"
);
}
}
if refresh_plan.needs_connection_info {
if let Err(e) = request_messages(
con,
build_command("getconnectioninfo", &[("clid", client_id.to_string())], &[]),
voice_in_tx,
chat_tx,
activity_tx,
delta_tx,
pending_moves,
voice_activity,
)
.await
{
warn!(
target: "chanora_protocol",
client_id,
error = %e,
"getconnectioninfo failed (ping/packet_loss will be unavailable)"
);
}
}
let db_info = if refresh_plan.needs_client_db_info {
request_client_db_info(
con,
database_id,
voice_in_tx,
chat_tx,
activity_tx,
delta_tx,
pending_moves,
voice_activity,
)
.await
.ok()
} else {
None
};
let state = con
.get_state()
.map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;
let client = state
.clients
.get(&target_id)
.ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found")))?;
let optional = client.optional_data.as_ref();
let connection = client.connection_data.as_ref();
let net_stats: Option<&ConnectionStats> = if is_own {
con.get_network_stats().ok()
} else {
None
};
let server_group_names: HashMap<_, _> = state
.server_groups
.iter()
@@ -1182,22 +1303,19 @@ async fn fetch_client_profile(
.or_else(|| db_info.as_ref().map(|info| info.created.unix_timestamp())),
last_connected_unix_seconds: optional
.map(|info| info.last_connected.unix_timestamp())
.or_else(|| {
db_info
.as_ref()
.map(|info| info.last_connected.unix_timestamp())
}),
.or_else(|| db_info.as_ref().map(|info| info.last_connected.unix_timestamp())),
connections_total: optional
.map(|info| u64::from(info.connections_total))
.or_else(|| {
db_info
.as_ref()
.map(|info| u64::from(info.connections_total))
}),
.or_else(|| db_info.as_ref().map(|info| u64::from(info.connections_total))),
online_seconds: connection
.and_then(|info| info.connected_time.map(|duration| duration.whole_seconds())),
idle_milliseconds: connection.map(|info| duration_millis(info.idle_time)),
ping_milliseconds: connection.and_then(|info| info.ping.map(duration_millis)),
ping_milliseconds: net_stats
.and_then(|s| std_duration_millis(s.rtt))
.or_else(|| connection.and_then(|info| info.ping.map(duration_millis))),
ping_deviation_milliseconds: net_stats
.and_then(|s| std_duration_millis(s.rtt_dev))
.or_else(|| connection.and_then(|info| info.ping_deviation.map(duration_millis))),
client_address: connection
.and_then(|info| info.client_address.map(|address| address.to_string()))
.unwrap_or_default(),
@@ -1219,10 +1337,16 @@ async fn fetch_client_profile(
bytes_uploaded_total: optional
.map(|info| info.bytes_uploaded_total)
.or_else(|| db_info.as_ref().map(|info| info.bytes_uploaded_total)),
packet_loss_client_to_server_total: connection
.map(|info| info.client_to_server_packetloss_total),
packet_loss_server_to_client_total: connection
.and_then(|info| info.server_to_client_packetloss_total),
packet_loss_client_to_server_total: net_stats
.map(|s| s.get_packetloss())
.or_else(|| {
connection.map(|info| info.client_to_server_packetloss_total)
}),
packet_loss_server_to_client_total: net_stats
.map(|s| s.get_packetloss_s2c_total())
.or_else(|| {
connection.and_then(|info| info.server_to_client_packetloss_total)
}),
})
}
@@ -1237,21 +1361,63 @@ fn build_command(name: &str, args: &[(&str, String)], flags: &[&str]) -> OutComm
command
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct ClientProfileRefreshPlan {
needs_server_groups: bool,
needs_channel_groups: bool,
needs_client_variables: bool,
needs_connection_info: bool,
needs_client_db_info: bool,
}
fn client_profile_refresh_plan(
is_own: bool,
has_optional: bool,
has_connection: bool,
needs_server_groups: bool,
needs_channel_groups: bool,
) -> ClientProfileRefreshPlan {
ClientProfileRefreshPlan {
needs_server_groups,
needs_channel_groups,
needs_client_variables: !has_optional || !is_own,
needs_connection_info: !has_connection || !is_own,
needs_client_db_info: !has_optional || !is_own,
}
}
async fn request_messages(
con: &mut Connection,
command: OutCommand,
voice_in_tx: &mpsc::Sender<InboundVoice>,
chat_tx: &mpsc::Sender<ChatMessage>,
activity_tx: &mpsc::Sender<ServerActivity>,
delta_tx: &mpsc::Sender<ProtocolDelta>,
pending_moves: &mut PendingMoves,
voice_activity: &mut HashMap<u64, Instant>,
) -> Result<Vec<InMessage>, ProtocolError> {
let handle = command
.send_with_result(con)
.map_err(|e| ProtocolError::Backend(format!("send command: {e}")))?;
let mut messages = Vec::new();
let deadline = Instant::now() + PROFILE_REFRESH_RESULT_TIMEOUT;
loop {
let item = con
.events()
.next()
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
return Err(ProtocolError::Backend(
"profile refresh command timed out".to_string(),
));
}
let item = {
let mut stream = con.events();
tokio::time::timeout(remaining, stream.next())
.await
.map_err(|_| {
ProtocolError::Backend("profile refresh command timed out".to_string())
})?
.ok_or_else(|| ProtocolError::Lost("event stream ended".to_string()))?
.map_err(|e| ProtocolError::Lost(e.to_string()))?;
.map_err(|e| ProtocolError::Lost(e.to_string()))?
};
match item {
StreamItem::MessageEvent(message) => messages.push(message),
StreamItem::MessageResult(reply, status) if reply == handle => {
@@ -1261,7 +1427,17 @@ async fn request_messages(
})?;
return Ok(messages);
}
_ => {}
StreamItem::Audio(buf) => {
handle_audio_stream_item(voice_in_tx, voice_activity, buf).await;
}
other => handle_non_audio_stream_item(
con,
other,
chat_tx,
activity_tx,
delta_tx,
pending_moves,
),
}
}
}
@@ -1269,10 +1445,22 @@ async fn request_messages(
async fn request_client_db_info(
con: &mut Connection,
dbid: tsclientlib::ClientDbId,
voice_in_tx: &mpsc::Sender<InboundVoice>,
chat_tx: &mpsc::Sender<ChatMessage>,
activity_tx: &mpsc::Sender<ServerActivity>,
delta_tx: &mpsc::Sender<ProtocolDelta>,
pending_moves: &mut PendingMoves,
voice_activity: &mut HashMap<u64, Instant>,
) -> Result<InClientDbInfoPart, ProtocolError> {
let messages = request_messages(
con,
build_command("clientdbinfo", &[("cldbid", dbid.0.to_string())], &[]),
voice_in_tx,
chat_tx,
activity_tx,
delta_tx,
pending_moves,
voice_activity,
)
.await?;
for message in messages {
@@ -1310,6 +1498,10 @@ fn duration_millis(duration: time::Duration) -> i64 {
.clamp(i64::MIN as i128, i64::MAX as i128) as i64
}
fn std_duration_millis(duration: std::time::Duration) -> Option<i64> {
duration.as_millis().try_into().ok()
}
fn uid_to_b64(uid: &tsclientlib::Uid) -> String {
BASE64_STANDARD.encode(&uid.0)
}
@@ -1709,8 +1901,9 @@ const _: () = {
#[cfg(test)]
mod tests {
use super::{
is_server_query_client_type, send_with_timeout, server_socket_from_config,
sort_channels_tree_by, ConnectConfig, SendTimeoutError,
client_profile_refresh_plan, is_server_query_client_type, send_with_timeout,
server_socket_from_config, sort_channels_tree_by, std_duration_millis,
ConnectConfig, SendTimeoutError,
};
use std::time::Duration;
use tokio::sync::mpsc;
@@ -1772,6 +1965,41 @@ mod tests {
assert_eq!(server_socket_from_config(&cfg), Some(addr));
}
#[test]
fn std_duration_millis_returns_none_when_duration_exceeds_i64() {
let overflow = Duration::from_millis(i64::MAX as u64) + Duration::from_millis(1);
assert_eq!(std_duration_millis(overflow), None);
}
#[test]
fn non_self_profile_refresh_requests_variables_connection_and_db_info() {
let plan = client_profile_refresh_plan(false, true, true, false, false);
assert!(plan.needs_client_variables);
assert!(plan.needs_connection_info);
assert!(plan.needs_client_db_info);
}
#[test]
fn own_profile_refresh_skips_complete_optional_and_connection_requests() {
let plan = client_profile_refresh_plan(true, true, true, false, false);
assert!(!plan.needs_server_groups);
assert!(!plan.needs_channel_groups);
assert!(!plan.needs_client_variables);
assert!(!plan.needs_connection_info);
assert!(!plan.needs_client_db_info);
}
#[test]
fn own_profile_refresh_requests_missing_group_lists() {
let plan = client_profile_refresh_plan(true, true, true, true, true);
assert!(plan.needs_server_groups);
assert!(plan.needs_channel_groups);
}
#[test]
fn channel_sort_linked_list_under_one_parent() {
// Server emits four root-level channels in arbitrary HashMap
crates/chanora_protocol/src/dto.rs
+2
View File
@@ -120,6 +120,8 @@ pub struct ClientProfile {
pub idle_milliseconds: Option<i64>,
/// Current ping in milliseconds, when visible.
pub ping_milliseconds: Option<i64>,
/// Current ping deviation in milliseconds, when visible.
pub ping_deviation_milliseconds: Option<i64>,
/// Client address. Empty when permission-gated.
pub client_address: String,
/// Resolved server group names for the online client.