Skip to content

Commit 6571e01

Browse files
its-mashclaude
its-mash
and
claude
committed
fix: SSE notifications not reaching clients + notifier hash dedup bypass
Three fixes for the SSE notification pipeline: 1. logging_middleware: Skip body.collect() for text/event-stream responses. The middleware was buffering infinite SSE streams, blocking forever and preventing VS Code from receiving any SSE events (notifications, pings). 2. mcp_notifier: Add `force` parameter to notify_all_list_changed(). Grant/feature-set events now bypass content-based hash dedup since the hash is computed from all features in the space and can't detect per-client grant changes that alter effective visibility. 3. Upgrade rmcp to 0.15.0 from fork with SSE channel replacement fix (409 Conflict on duplicate GET streams) and adapt to new struct fields (granted_scopes, Default trait changes). Signed-off-by: Claude Opus 4.6 <noreply@anthropic.com> Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> Signed-off-by: Mohammod Al Amin Ashik <maa.ashik00@gmail.com>
1 parent e8620d9 commit 6571e01

8 files changed

Lines changed: 331 additions & 33 deletions

File tree

‎Cargo.lock‎

Lines changed: 268 additions & 6 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎Cargo.toml‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,7 @@ os_pipe = "1"
5858

5959
# MCP Protocol
6060
# NOTE: Never use local path dependency - E:\one-mcp\rust-sdk is for source lookup only
61-
rmcp = { version = "0.14.0", features = [
61+
rmcp = { version = "0.15.0", features = [
6262
"client",
6363
"server",
6464
"transport-io",
@@ -84,4 +84,9 @@ lto = true
8484
codegen-units = 1
8585
strip = true
8686

87+
# Temporary patch: fixes SSE channel replacement bug (notifications lost on reconnect)
88+
# Remove once upstream merges https://github.com/modelcontextprotocol/rust-sdk/pull/660
89+
[patch.crates-io]
90+
rmcp = { git = "https://github.com/ion-ash/rust-sdk.git", branch = "fix/sse-channel-replacement-conflict" }
91+
8792

‎crates/mcpmux-gateway/src/consumers/mcp_notifier.rs‎

Lines changed: 34 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -415,7 +415,7 @@ impl MCPNotifier {
415415
feature_set_id = %feature_set_id,
416416
"[MCPNotifier] 📨 GrantIssued - notifying all clients in space"
417417
);
418-
self.notify_all_list_changed(space_id).await;
418+
self.notify_all_list_changed(space_id, true).await;
419419
}
420420

421421
DomainEvent::GrantRevoked {
@@ -429,7 +429,7 @@ impl MCPNotifier {
429429
feature_set_id = %feature_set_id,
430430
"[MCPNotifier] 📨 GrantRevoked - notifying all clients in space"
431431
);
432-
self.notify_all_list_changed(space_id).await;
432+
self.notify_all_list_changed(space_id, true).await;
433433
}
434434

435435
DomainEvent::ClientGrantsUpdated {
@@ -443,7 +443,7 @@ impl MCPNotifier {
443443
feature_sets = feature_set_ids.len(),
444444
"[MCPNotifier] 📨 ClientGrantsUpdated - notifying all clients in space"
445445
);
446-
self.notify_all_list_changed(space_id).await;
446+
self.notify_all_list_changed(space_id, true).await;
447447
}
448448

449449
DomainEvent::FeatureSetMembersChanged {
@@ -456,7 +456,7 @@ impl MCPNotifier {
456456
feature_set_id = %feature_set_id,
457457
"[MCPNotifier] 📨 FeatureSetMembersChanged - notifying all clients in space"
458458
);
459-
self.notify_all_list_changed(space_id).await;
459+
self.notify_all_list_changed(space_id, true).await;
460460
}
461461

462462
// ============ Backend Server Notifications (Pass-through with Throttling) ============
@@ -523,7 +523,7 @@ impl MCPNotifier {
523523
status = ?status,
524524
"[MCPNotifier] ServerStatusChanged (Disconnected) - notifying clients to clear features"
525525
);
526-
self.notify_all_list_changed(space_id).await;
526+
self.notify_all_list_changed(space_id, false).await;
527527
} else {
528528
debug!(
529529
server_id = %server_id,
@@ -549,7 +549,7 @@ impl MCPNotifier {
549549
removed = removed.len(),
550550
"[MCPNotifier] ServerFeaturesRefreshed"
551551
);
552-
self.notify_all_list_changed(space_id).await;
552+
self.notify_all_list_changed(space_id, false).await;
553553
}
554554

555555
// Other events that affect MCP capabilities are handled above
@@ -571,8 +571,13 @@ impl MCPNotifier {
571571
/// **Important**: This method handles throttling at the batch level and marks
572572
/// all individual notification types as sent, preventing double-notifications
573573
/// when individual DomainEvent::ToolsChanged/etc. events arrive shortly after.
574-
async fn notify_all_list_changed(&self, space_id: Uuid) {
575-
// 1. Content-Based Deduping
574+
///
575+
/// **`force` parameter**: When `true`, skips content-based hash dedup. Used for
576+
/// grant-related events where the total features in the space haven't changed but
577+
/// the *effective* features visible to clients have (due to grant/feature set changes).
578+
/// The hash is computed from all features in the space, so it can't detect grant changes.
579+
async fn notify_all_list_changed(&self, space_id: Uuid, force: bool) {
580+
// 1. Content-Based Deduping (skipped when force=true)
576581
let tools_hash = self
577582
.calculate_feature_hash(space_id, FeatureType::Tool)
578583
.await;
@@ -583,23 +588,27 @@ impl MCPNotifier {
583588
.calculate_feature_hash(space_id, FeatureType::Resource)
584589
.await;
585590

586-
let any_changed = {
587-
let hashes = self.state_hashes.read();
588-
let t_changed = hashes
589-
.get(&(space_id, NotificationType::Tools))
590-
.is_none_or(|&h| h != tools_hash);
591-
let p_changed = hashes
592-
.get(&(space_id, NotificationType::Prompts))
593-
.is_none_or(|&h| h != prompts_hash);
594-
let r_changed = hashes
595-
.get(&(space_id, NotificationType::Resources))
596-
.is_none_or(|&h| h != resources_hash);
597-
t_changed || p_changed || r_changed
598-
};
599-
600-
if !any_changed {
601-
debug!(space_id = %space_id, "[MCPNotifier] 🛑 Batch content unchanged, skipping");
602-
return;
591+
if !force {
592+
let any_changed = {
593+
let hashes = self.state_hashes.read();
594+
let t_changed = hashes
595+
.get(&(space_id, NotificationType::Tools))
596+
.is_none_or(|&h| h != tools_hash);
597+
let p_changed = hashes
598+
.get(&(space_id, NotificationType::Prompts))
599+
.is_none_or(|&h| h != prompts_hash);
600+
let r_changed = hashes
601+
.get(&(space_id, NotificationType::Resources))
602+
.is_none_or(|&h| h != resources_hash);
603+
t_changed || p_changed || r_changed
604+
};
605+
606+
if !any_changed {
607+
debug!(space_id = %space_id, "[MCPNotifier] 🛑 Batch content unchanged, skipping");
608+
return;
609+
}
610+
} else {
611+
info!(space_id = %space_id, "[MCPNotifier] 🔓 Force-sending (grant/feature set change, bypassing hash dedup)");
603612
}
604613

605614
let now = Instant::now();

‎crates/mcpmux-gateway/src/pool/credential_store.rs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -207,6 +207,7 @@ impl CredentialStore for DatabaseCredentialStore {
207207
Some(StoredCredentials {
208208
client_id: reg.client_id,
209209
token_response: Some(token_response),
210+
granted_scopes: Vec::new(),
210211
})
211212
}
212213
(Some(reg), None) => {
@@ -217,6 +218,7 @@ impl CredentialStore for DatabaseCredentialStore {
217218
Some(StoredCredentials {
218219
client_id: reg.client_id,
219220
token_response: None,
221+
granted_scopes: Vec::new(),
220222
})
221223
}
222224
(None, Some(access)) => {
@@ -228,6 +230,7 @@ impl CredentialStore for DatabaseCredentialStore {
228230
Some(StoredCredentials {
229231
client_id: String::new(),
230232
token_response: Some(token_response),
233+
granted_scopes: Vec::new(),
231234
})
232235
}
233236
(None, None) => {

‎crates/mcpmux-gateway/src/pool/instance.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@ impl McpClientHandler {
4646
title: Some("McpMux Gateway".to_string()),
4747
icons: None,
4848
website_url: None,
49+
..Default::default()
4950
},
5051
meta: None,
5152
},

‎crates/mcpmux-gateway/src/pool/oauth_utils.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,7 @@ pub fn convert_from_stored_metadata(stored: &StoredOAuthMetadata) -> Authorizati
110110
scopes_supported: stored.scopes_supported.clone(),
111111
response_types_supported: stored.response_types_supported.clone(),
112112
additional_fields: stored.additional_fields.clone(),
113+
..Default::default()
113114
}
114115
}
115116

‎crates/mcpmux-gateway/src/server/logging_middleware.rs‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -208,7 +208,23 @@ pub async fn http_logging_middleware(request: Request, next: Next) -> Result<Res
208208
let response = next.run(request).await;
209209
let status = response.status().as_u16();
210210

211-
// Extract response body to log it
211+
// Check if this is a streaming response (SSE) — NEVER buffer these.
212+
// SSE responses have Content-Type: text/event-stream and are infinite
213+
// streams. Calling body.collect() would block forever, preventing VS Code
214+
// from receiving any SSE events (notifications, keep-alive pings).
215+
let is_streaming = response
216+
.headers()
217+
.get("content-type")
218+
.and_then(|v| v.to_str().ok())
219+
.is_some_and(|ct| ct.contains("text/event-stream"));
220+
221+
if is_streaming {
222+
// For SSE streams, log entry only and pass through without touching body
223+
RequestSpan::log_exit(&ctx, status, None);
224+
return Ok(response);
225+
}
226+
227+
// For non-streaming responses, capture body for logging
212228
let (parts, body) = response.into_parts();
213229
let body_bytes = match body.collect().await {
214230
Ok(collected) => collected.to_bytes(),

‎crates/mcpmux-mcp/src/transports.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -93,6 +93,7 @@ impl McpClientHandler {
9393
title: Some("McpMux Gateway".to_string()),
9494
icons: None,
9595
website_url: None,
96+
..Default::default()
9697
},
9798
meta: None,
9899
},

0 commit comments

Comments
 (0)