Skip to content

Commit 53cfadc

Browse files
authored
test(server): keep relays queued during session setup behind SessionAccepted (#4689)
The supervisor drops a session whose first frame is not SessionAccepted, and the gateway registers a session before it acknowledges it. #4320 fixed the ordering by routing registry traffic through a private channel whose forwarder starts only after the ack is queued, but nothing tested it. The new tests pause setup in its endpoint-status reset, open a relay on the registered but unacknowledged session, and check that the stream yields SessionAccepted and then the queued RelayOpen, in both poll and push mode. With the test harness ported, the poll test fails on the code before #4320. Closes #4687 Signed-off-by: Emilien Macchi <emacchi@redhat.com>
1 parent ba302a6 commit 53cfadc

1 file changed

Lines changed: 84 additions & 0 deletions

File tree

‎crates/openshell-server/src/supervisor_session.rs‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3173,6 +3173,90 @@ mod tests {
31733173
);
31743174
}
31753175

3176+
#[tokio::test]
3177+
async fn relay_queued_during_setup_follows_session_accepted() {
3178+
assert_relay_queued_during_setup_follows_session_accepted(StreamFeatures::Legacy).await;
3179+
}
3180+
3181+
#[tokio::test]
3182+
async fn relay_queued_during_push_setup_follows_session_accepted() {
3183+
assert_relay_queued_during_setup_follows_session_accepted(StreamFeatures::Apply).await;
3184+
}
3185+
3186+
/// The supervisor drops a session whose first frame is not
3187+
/// `SessionAccepted`, so a relay routed to a session that is registered
3188+
/// but not yet acknowledged must wait behind the ack.
3189+
async fn assert_relay_queued_during_setup_follows_session_accepted(features: StreamFeatures) {
3190+
let mut state = crate::grpc::test_support::test_server_state().await;
3191+
if features == StreamFeatures::Apply {
3192+
crate::grpc::test_support::enable_push_delivery(&mut state);
3193+
}
3194+
let mut sandbox = sandbox_record("sb-setup", "sb-setup");
3195+
// A push bootstrap needs a spec.
3196+
sandbox.spec = Some(openshell_core::proto::SandboxSpec::default());
3197+
state.store.put_message(&sandbox).await.unwrap();
3198+
// Setup takes this guard for its endpoint-status reset, which runs
3199+
// after registration and before the ack.
3200+
let guard = state
3201+
.compute
3202+
.sandbox_mutation_guard_by_id("sb-setup")
3203+
.await
3204+
.unwrap()
3205+
.expect("sandbox exists");
3206+
let connect = tokio::spawn({
3207+
let state = Arc::clone(&state);
3208+
async move {
3209+
crate::grpc::test_support::connect_supervisor_stream(&state, "sb-setup", features)
3210+
.await
3211+
}
3212+
});
3213+
let (session_id, _) = state
3214+
.supervisor_sessions
3215+
.wait_for_session(
3216+
"sb-setup",
3217+
tokio::time::Instant::now() + Duration::from_secs(5),
3218+
true,
3219+
)
3220+
.await
3221+
.expect("session registers before its setup finishes");
3222+
let (channel_id, _relay_rx) = state
3223+
.supervisor_sessions
3224+
.open_relay("sb-setup", Duration::from_secs(5))
3225+
.await
3226+
.expect("relay is queued on the registered session");
3227+
// Endpoint-status authority is set before the ack, so the relay
3228+
// went out while the session was registered but not acknowledged.
3229+
assert!(
3230+
!state
3231+
.supervisor_sessions
3232+
.is_endpoint_status_authority("sb-setup", &session_id)
3233+
);
3234+
drop(guard);
3235+
3236+
let mut harness = connect.await.unwrap().expect("supervisor connects");
3237+
let Some(gateway_message::Payload::SessionAccepted(accepted)) =
3238+
first_gateway_message(&mut harness).await.payload
3239+
else {
3240+
panic!("expected SessionAccepted ahead of the queued relay");
3241+
};
3242+
assert_eq!(accepted.session_id, session_id);
3243+
assert_eq!(
3244+
accepted.config_push_enabled,
3245+
features == StreamFeatures::Apply
3246+
);
3247+
match first_gateway_message(&mut harness).await.payload {
3248+
Some(gateway_message::Payload::RelayOpen(open)) => {
3249+
assert_eq!(open.channel_id, channel_id);
3250+
}
3251+
other => panic!("expected the queued RelayOpen, got {other:?}"),
3252+
}
3253+
assert!(
3254+
state
3255+
.supervisor_sessions
3256+
.is_current_session("sb-setup", &accepted.session_id)
3257+
);
3258+
}
3259+
31763260
use tokio::io::{AsyncReadExt, AsyncWriteExt};
31773261

31783262
async fn test_store() -> Arc<Store> {

0 commit comments

Comments
 (0)