Skip to content

Drop changes if channel for updates is full #491

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
somtochiama wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Drop changes if channel for updates is full #491

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7ccccb1
drop changes if updates channel is full
somtochiama Jun 23, 2026
1deeb62
correct check
somtochiama Jun 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 6 additions & 2 deletions crates/corro-types/src/pubsub.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,8 +69,12 @@ pub struct MatcherCreated {
const SUB_EVENT_CHANNEL_CAP: usize = 512;

impl Manager<MatcherHandle> for SubsManager {
fn trait_type(&self) -> String {
"subs".to_string()
fn trait_type(&self) -> &'static str {
"subs"
}

fn fallible(&self) -> bool {
false
}

fn get(&self, id: &Uuid) -> Option<MatcherHandle> {
Expand Down
33 changes: 21 additions & 12 deletions crates/corro-types/src/updates.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,8 @@ use tripwire::Tripwire;
use uuid::Uuid;

pub trait Manager<H> {
fn trait_type(&self) -> String;
fn trait_type(&self) -> &'static str;
fn fallible(&self) -> bool;
fn get(&self, id: &Uuid) -> Option<H>;
fn remove(&self, id: &Uuid) -> Option<H>;
fn get_handles(&self) -> BTreeMap<Uuid, H>;
Expand Down Expand Up @@ -61,8 +62,12 @@ impl std::fmt::Debug for HandleMetrics {
pub struct UpdatesManager(Arc<RwLock<InnerUpdatesManager>>);

impl Manager<UpdateHandle> for UpdatesManager {
fn trait_type(&self) -> String {
"updates".to_string()
fn trait_type(&self) -> &'static str {
"updates"
}

fn fallible(&self) -> bool {
true
}

fn get(&self, id: &Uuid) -> Option<UpdateHandle> {
Expand Down Expand Up @@ -461,15 +466,19 @@ where
error!(sub_id = %id, "could not send change candidates to {trait_type} handler: {e}");
match e {
mpsc::error::TrySendError::Full(item) => {
warn!("channel is full, falling back to async send");

let changes_tx = handle.changes_tx();
let id = *id;
tokio::spawn(async move {
if let Err(e) = changes_tx.send(item).await {
error!(sub_id = %id, "could not send match_change candidates to handler: {e}");
}
});
if !manager.fallible() {
warn!("channel is full, falling back to async send");
let changes_tx = handle.changes_tx();
let id = *id;
tokio::spawn(async move {
if let Err(e) = changes_tx.send(item).await {
error!(sub_id = %id, "could not send match_change candidates to handler: {e}");
}
});
} else {
counter!("corro.matcher.changes.dropped", "type" => trait_type)
.increment(1);
}
}
mpsc::error::TrySendError::Closed(_) => {
if let Some(handle) = manager.remove(id) {
Expand Down
Loading