Page MenuHomePhabricator
Differential D13606

[tunnelbroker] Recover AMQP consumers
ClosedPublic
Actions

Authored by bartek on Oct 3 2024, 1:37 PM.
Tags
None
Referenced Files
Unknown Object (File)
Tue, Apr 1, 2:44 AM
Unknown Object (File)
Mon, Mar 31, 7:30 AM
Unknown Object (File)
Mon, Mar 24, 5:11 AM
Unknown Object (File)
Mar 12 2025, 3:18 AM
Unknown Object (File)
Mar 11 2025, 4:16 PM
Unknown Object (File)
Mar 11 2025, 4:11 PM
Unknown Object (File)
Mar 11 2025, 4:11 PM
Unknown Object (File)
Mar 11 2025, 3:48 PM
View All 89 Files
Subscribers
ashoat
tomek

Details

Reviewers
kamil
varun
will
Commits
rCOMM7649ca4931b2: [tunnelbroker] Recover AMQP consumers
Summary

When a consumer returns a confirmation error, we can try restoring it without closing the WebSocket session

Depends on D13605

Test Plan

Test plan from D13595 (commtest) sometimes failed on this confirmation. Now it's no longer happening

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: D13607: [tunnelbroker] Trigger AMQP reconnect when closing session.Oct 3 2024, 1:37 PM
Harbormaster completed remote builds in B32057: Diff 44886.Oct 3 2024, 2:11 PM
bartek mentioned this in D13611: [tunnelbroker] Recover AMQP before publishing messages.Oct 4 2024, 3:38 AM
bartek published this revision for review.Oct 4 2024, 3:58 AM
bartek edited the summary of this revision. (Show Details)
bartek edited the test plan for this revision. (Show Details)
bartek added inline comments.
services/tunnelbroker/src/websockets/mod.rs
254 ↗(On Diff #44886)

Should it be warn or error?

bartek updated this revision to Diff 44903.Oct 4 2024, 8:18 AM

Remove connection pool

Harbormaster completed remote builds in B32071: Diff 44903.Oct 4 2024, 8:46 AM
kamil added inline comments.Oct 7 2024, 2:15 AM
services/tunnelbroker/src/amqp.rs
137–143 ↗(On Diff #44903)

is it needed? should we use it here?

services/tunnelbroker/src/websockets/mod.rs
248 ↗(On Diff #44903)

do we always want to reset or only on connection errors?

254 ↗(On Diff #44886)

maybe error to first have a better view of when and how often this is happening, and later we can change to warn?

kamil accepted this revision.Oct 7 2024, 2:37 AM
This revision is now accepted and ready to land.Oct 7 2024, 2:37 AM
bartek added inline comments.Oct 7 2024, 4:38 AM
services/tunnelbroker/src/amqp.rs
137–143 ↗(On Diff #44903)

Should be in the next diff

services/tunnelbroker/src/websockets/mod.rs
248 ↗(On Diff #44903)

On all channel errors. Generally all errors here will need Consumer reset, but another condition inside reset_failed_amqp is not a bad idea

bartek updated this revision to Diff 44924.Oct 7 2024, 7:06 AM

Address feedback

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

Revision Contents
Changeset List

PathSize
services/
tunnelbroker/
src/
websockets/
32 lines
22 lines

Diff 44936

View Options

services/tunnelbroker/src/websockets/mod.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