twonly/twonly-app
1
0
Fork 0
mirror of https://github.com/twonlyapp/twonly-app.git synced 2026-10-09 02:54:14 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix server ack issue
Some checks are pending
Flutter analyze & test / flutter_analyze_and_test (push) Waiting to run
Details

Browse source
This commit is contained in:
otsmr 2026-09-11 23:53:56 +02:00
parent 512d4d05f8
commit 69281ceed2
2 changed files with 35 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
rust/src/api/runtime/client.rs
View file

@ -293,7 +293,7 @@ impl ApiClient {
/// A replaced socket can emit its final disconnect/shutdown events after
/// the new socket has started. Those events must not overwrite the new
/// connection's state or schedule another reconnect.
async fn is_current_connection(&self, connection: &Arc<WebSocketClient>) -> bool {
pub(super) async fn is_current_connection(&self, connection: &Arc<WebSocketClient>) -> bool {
self.ws_client
.lock()
.await
@ -421,7 +421,7 @@ impl ApiClient {
}
// Tungstenite message
if let stream_tungstenite::tokio_tungstenite::tungstenite::Message::Binary(bytes) = &*msg {
self_clone.handle_incoming(bytes).await;
self_clone.handle_incoming(&connection, bytes).await;
}
}
Err(_) => break,

38
rust/src/api/runtime/request.rs
View file

@ -54,7 +54,11 @@ fn is_dead_transport(error: &SendError) -> bool {
}
impl ApiClient {
pub(crate) async fn handle_incoming(self: &Arc<Self>, bytes: &[u8]) {
pub(crate) async fn handle_incoming(
self: &Arc<Self>,
connection: &Arc<stream_tungstenite::WebSocketClient>,
bytes: &[u8],
) {
let Ok(message) = server_to_client::ServerToClient::decode(bytes) else {
return;
};
@ -68,6 +72,12 @@ impl ApiClient {
}
} else {
let client = self.clone();
// A response belongs to the WebSocket which delivered the
// request. Sequence IDs are scoped to a server session and
// restart when a replacement connection is authenticated,
// so forwarding this response through `client.send()`
// could satisfy an unrelated request on the new session.
let response_connection = connection.clone();
let Some(ctx) = self.context.upgrade() else {
tracing::warn!("context was dropped before handling a server message");
return;
@ -85,10 +95,28 @@ impl ApiClient {
},
)),
};
if let Err(error) =
client.send(acknowledgement.encode_to_vec()).await
{
tracing::warn!("failed to acknowledge server message: {error}");
if !client.is_current_connection(&response_connection).await {
tracing::debug!(
sequence = v0.seq,
"discarding response for a replaced WebSocket session"
);
return;
}
let message = stream_tungstenite::tokio_tungstenite::tungstenite::Message::Binary(
acknowledgement.encode_to_vec().into(),
);
if let Err(error) = response_connection.send_async(message).await {
if is_dead_transport(&error) {
client.schedule_reconnect(
&response_connection,
"server acknowledgement send failed",
);
}
tracing::warn!(
sequence = v0.seq,
"failed to acknowledge server message: {error}"
);
}
}
Err(error) => {