Page MenuHomePhabricator
Differential D13607

[tunnelbroker] Trigger AMQP reconnect when closing session
ClosedPublic
Actions

Authored by bartek on Oct 3 2024, 1:37 PM.
Tags
None
Referenced Files
Unknown Object (File)
Wed, Mar 26, 6:21 AM
Unknown Object (File)
Mar 11 2025, 3:57 PM
Unknown Object (File)
Mar 11 2025, 3:52 PM
Unknown Object (File)
Mar 11 2025, 3:40 PM
Unknown Object (File)
Mar 11 2025, 3:26 PM
Unknown Object (File)
Mar 11 2025, 3:26 PM
Unknown Object (File)
Mar 3 2025, 6:18 AM
Unknown Object (File)
Feb 28 2025, 4:28 AM
View All 98 Files
Subscribers
ashoat
tomek

Details

Reviewers
kamil
varun
will
Commits
rCOMM5bf4d6b9a469: [tunnelbroker] Trigger AMQP reconnect when closing session
Summary

Most of errors were happening here. WS session was closed so consumer cancelation and queue removal calls weren't able to succeed.

Depends on D13606

Test Plan

Observing logs on staging. Possible but hard to reproduce with commtest
Can also be reproduced by:

  • Connecting a device to healthy Tunnelbroker
  • Don't make any traffic
  • Disconnect RabbitMQ
  • Then disconnect device

Diff Detail

Repository
rCOMM Comm
Lint
Lint Not Applicable
Unit
Tests Not Applicable

Event Timeline

bartek created this revision.Oct 3 2024, 1:37 PM
bartek held this revision as a draft.
Herald added subscribers: tomek, ashoat. · View Herald TranscriptOct 3 2024, 1:37 PM
bartek added a child revision: D13608: [tunnelbroker] Ignore connection error messages handled elsewhere.Oct 3 2024, 1:38 PM
Harbormaster completed remote builds in B32058: Diff 44887.Oct 3 2024, 2:17 PM
bartek updated this revision to Diff 44895.Oct 4 2024, 3:32 AM

Rebase

bartek mentioned this in D13611: [tunnelbroker] Recover AMQP before publishing messages.Oct 4 2024, 3:38 AM
bartek edited the summary of this revision. (Show Details)Oct 4 2024, 4:04 AM
bartek edited the test plan for this revision. (Show Details)
Harbormaster failed remote builds in B32064: Diff 44895!Oct 4 2024, 4:07 AM
bartek updated this revision to Diff 44904.Oct 4 2024, 8:18 AM

Remove connection pool

bartek edited the summary of this revision. (Show Details)Oct 4 2024, 8:40 AM
bartek published this revision for review.Oct 4 2024, 8:44 AM
bartek added inline comments.
services/tunnelbroker/src/websockets/session.rs
743–747 ↗(On Diff #44904)

According to RabbitMQ docs, if the channel is closed, all its consumers are also closed, so instead of recovering here, we can simply ensure connection is okay and exit early

757–762 ↗(On Diff #44904)

This is to avoid spamming error alerts while connection errors are handled by our code elsewhere

766–771 ↗(On Diff #44904)

We can probably consider having a separate task for cleaning up queues in case they're not able to be properly deleted here

Harbormaster completed remote builds in B32072: Diff 44904.Oct 4 2024, 8:55 AM
kamil accepted this revision.Oct 7 2024, 2:22 AM
kamil added inline comments.
services/tunnelbroker/src/websockets/session.rs
743–747 ↗(On Diff #44904)

good idea to ensure connection is okay

766–771 ↗(On Diff #44904)

This seems reasonable

This revision is now accepted and ready to land.Oct 7 2024, 2:22 AM
bartek updated this revision to Diff 44925.Oct 7 2024, 7:07 AM

Rebase

Harbormaster completed remote builds in B32089: Diff 44925.Oct 7 2024, 7:25 AM
Closed by commit rCOMM5bf4d6b9a469: [tunnelbroker] Trigger AMQP reconnect when closing session (authored by bartek). · Explain WhyOct 7 2024, 7:28 AM
This revision was automatically updated to reflect the committed changes.
bartek added a commit: rCOMM5bf4d6b9a469: [tunnelbroker] Trigger AMQP reconnect when closing session.
bartek mentioned this in rCOMM51a32d2cff2a: [tunnelbroker] Recover AMQP before publishing messages.

Revision Contents
Changeset List

PathSize
services/
tunnelbroker/
src/
13 lines
websockets/
34 lines

Diff 44935

View Options

services/tunnelbroker/src/amqp.rs

Show First 20 Lines • Show All 90 Lines • ▼ Show 20 Lines Ok(Self {
.try_into() .try_into()
.expect("Channels vec size doesn't match array size"), .expect("Channels vec size doesn't match array size"),
}) })
} }
fn get_channel( fn get_channel(
&self, &self,
id_hash: impl std::hash::Hash, id_hash: impl std::hash::Hash,
) -> Result<lapin::Channel, lapin::Error> { ) -> Option<lapin::Channel> {
let channel_idx = Self::channel_idx_for_hash(id_hash); let channel_idx = Self::channel_idx_for_hash(id_hash);
let channel = self.channels[channel_idx].clone(); let channel = self.channels[channel_idx].clone();
let channel_id = channel.id(); let channel_id = channel.id();
tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel"); tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel");
Ok(channel)
if channel.status().connected() {
return Some(channel);
}
warn!(channel_id, channel_idx, "Channel is dead!");
None
}
async fn reset_channel(
&mut self,
id_hash: impl std::hash::Hash,
) -> Result<lapin::Channel, lapin::Error> {
let channel_idx = Self::channel_idx_for_hash(id_hash);
let existing_channel = &self.channels[channel_idx];
if existing_channel.status().connected() {
return Ok(existing_channel.clone());
}
let new_channel = self.conn.create_channel().await?;
debug!(
"Channel for idx={channel_idx} was recreated, new Id={}",
new_channel.id()
);
self.channels[channel_idx] = new_channel.clone();
Ok(new_channel)
} }
fn is_connected(&self) -> bool { fn is_connected(&self) -> bool {
self.conn.status().connected() self.conn.status().connected()
} }
fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize { fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize {
// We have channel pool and want to distribute them between connected // We have channel pool and want to distribute them between connected
Show All 22 Lines pub async fn connect() -> Result<Self, lapin::Error> {
let inner = Arc::new(RwLock::new(conn)); let inner = Arc::new(RwLock::new(conn));
info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri); info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri);
Ok(Self { inner }) Ok(Self { inner })
} }
pub async fn channel( pub async fn channel(
&self, &self,
id_hash: impl std::hash::Hash, id_hash: impl std::hash::Hash + Clone,
) -> Result<lapin::Channel, lapin::Error> { ) -> Result<lapin::Channel, lapin::Error> {
if !self.is_connected().await { if !self.is_connected().await {
warn!("AMQP disconnected while retrieving channel"); warn!("AMQP disconnected while retrieving channel");
self.reset_conn().await?; self.reset_conn().await?;
} }
self.inner.read().await.get_channel(id_hash)
{
let inner = self.inner.read().await;
if let Some(channel) = inner.get_channel(id_hash.clone()) {
return Ok(channel);
}
}
self.inner.write().await.reset_channel(id_hash).await
} }
async fn reset_conn(&self) -> Result<(), lapin::Error> { async fn reset_conn(&self) -> Result<(), lapin::Error> {
debug!("Resetting connection..."); debug!("Resetting connection...");
let mut inner = self.inner.write().await; let mut inner = self.inner.write().await;
if !inner.is_connected() { if !inner.is_connected() {
let new_conn = ConnectionInner::new().await?; let new_conn = ConnectionInner::new().await?;
*inner = new_conn; *inner = new_conn;
▲ Show 20 Lines • Show All 41 Lines • Show Last 20 Lines
View Options

services/tunnelbroker/src/websockets/session.rs

Show First 20 Lines • Show All 90 Lines • ▼ Show 20 Lines Ok(Self {
.try_into() .try_into()
.expect("Channels vec size doesn't match array size"), .expect("Channels vec size doesn't match array size"),
}) })
} }
fn get_channel( fn get_channel(
&self, &self,
id_hash: impl std::hash::Hash, id_hash: impl std::hash::Hash,
) -> Result<lapin::Channel, lapin::Error> { ) -> Option<lapin::Channel> {
let channel_idx = Self::channel_idx_for_hash(id_hash); let channel_idx = Self::channel_idx_for_hash(id_hash);
let channel = self.channels[channel_idx].clone(); let channel = self.channels[channel_idx].clone();
let channel_id = channel.id(); let channel_id = channel.id();
tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel"); tracing::trace!(channel_id, channel_idx, "Retrieving AMQP Channel");
Ok(channel)
if channel.status().connected() {
return Some(channel);
}
warn!(channel_id, channel_idx, "Channel is dead!");
None
}
async fn reset_channel(
&mut self,
id_hash: impl std::hash::Hash,
) -> Result<lapin::Channel, lapin::Error> {
let channel_idx = Self::channel_idx_for_hash(id_hash);
let existing_channel = &self.channels[channel_idx];
if existing_channel.status().connected() {
return Ok(existing_channel.clone());
}
let new_channel = self.conn.create_channel().await?;
debug!(
"Channel for idx={channel_idx} was recreated, new Id={}",
new_channel.id()
);
self.channels[channel_idx] = new_channel.clone();
Ok(new_channel)
} }
fn is_connected(&self) -> bool { fn is_connected(&self) -> bool {
self.conn.status().connected() self.conn.status().connected()
} }
fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize { fn channel_idx_for_hash(id_hash: impl std::hash::Hash) -> usize {
// We have channel pool and want to distribute them between connected // We have channel pool and want to distribute them between connected
Show All 22 Lines pub async fn connect() -> Result<Self, lapin::Error> {
let inner = Arc::new(RwLock::new(conn)); let inner = Arc::new(RwLock::new(conn));
info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri); info!("Connected to AMQP endpoint: {}", &CONFIG.amqp_uri);
Ok(Self { inner }) Ok(Self { inner })
} }
pub async fn channel( pub async fn channel(
&self, &self,
id_hash: impl std::hash::Hash, id_hash: impl std::hash::Hash + Clone,
) -> Result<lapin::Channel, lapin::Error> { ) -> Result<lapin::Channel, lapin::Error> {
if !self.is_connected().await { if !self.is_connected().await {
warn!("AMQP disconnected while retrieving channel"); warn!("AMQP disconnected while retrieving channel");
self.reset_conn().await?; self.reset_conn().await?;
} }
self.inner.read().await.get_channel(id_hash)
{
let inner = self.inner.read().await;
if let Some(channel) = inner.get_channel(id_hash.clone()) {
return Ok(channel);
}
}
self.inner.write().await.reset_channel(id_hash).await
} }
async fn reset_conn(&self) -> Result<(), lapin::Error> { async fn reset_conn(&self) -> Result<(), lapin::Error> {
debug!("Resetting connection..."); debug!("Resetting connection...");
let mut inner = self.inner.write().await; let mut inner = self.inner.write().await;
if !inner.is_connected() { if !inner.is_connected() {
let new_conn = ConnectionInner::new().await?; let new_conn = ConnectionInner::new().await?;
*inner = new_conn; *inner = new_conn;
▲ Show 20 Lines • Show All 41 Lines • Show Last 20 Lines