Skip to main content

max / makenotwork

8.9 KB · 253 lines History Blame Raw
1 //! End-to-end regression test for the SSH backpressure deadlock.
2 //!
3 //! This is the test that was missing when the push hang shipped. The bug could
4 //! not be reached from a unit test: it lives in russh's session loop, and
5 //! `ChannelId` and `Channel<Msg>` cannot be constructed outside that crate. So
6 //! this drives the real thing — a real `git push`, over the real `ssh` client,
7 //! against [`MnwServer`] and [`MnwHandler`], with only the MNW API stubbed.
8 //!
9 //! WHAT IT PINS. russh hands every inbound `CHANNEL_DATA` first to the
10 //! `Channel`'s bounded queue (`channel_buffer_size`, 100 messages) with a
11 //! blocking `send().await`, and only then to `Handler::data`. A `Channel` the
12 //! handler parks and never reads therefore wedges the whole session loop once
13 //! 100 packets have arrived, and the client hangs forever mid-pack. The push
14 //! below is deliberately larger than that buffer: incompressible random bytes,
15 //! so the pack cannot shrink under it. Against the pre-fix handler this test
16 //! hangs until its timeout; against the fixed one it finishes in about a
17 //! second.
18 //!
19 //! Requires `git` and `ssh` on PATH.
20
21 use std::path::Path;
22 use std::process::Stdio;
23 use std::sync::Arc;
24 use std::time::Duration;
25
26 use russh::server::Server as _;
27 use tokio::io::{AsyncReadExt, AsyncWriteExt};
28 use tokio::net::TcpListener;
29 use tokio::process::Command;
30
31 use crate::api::MnwApiClient;
32 use crate::rate_limit::AuthRateLimiter;
33 use crate::ssh::MnwServer;
34
35 /// Comfortably past russh's 100-message channel buffer. 6 MiB of random bytes
36 /// packs to roughly 6 MiB, which is ~190 packets at the 32 KiB maximum packet
37 /// size. `shop`, the repo that first hit this, was 4.5 MiB.
38 const PAYLOAD_BYTES: usize = 6 * 1024 * 1024;
39
40 /// Generous enough that a slow machine does not produce a false red, short
41 /// enough that a real regression fails the suite rather than hanging CI. The
42 /// fixed path takes about a second.
43 const PUSH_TIMEOUT: Duration = Duration::from_secs(90);
44
45 async fn run(cmd: &str, args: &[&str], cwd: &Path) -> anyhow::Result<()> {
46 let out = Command::new(cmd)
47 .args(args)
48 .current_dir(cwd)
49 .output()
50 .await?;
51 anyhow::ensure!(
52 out.status.success(),
53 "{cmd} {args:?} failed: {}",
54 String::from_utf8_lossy(&out.stderr)
55 );
56 Ok(())
57 }
58
59 /// Minimal stand-in for the MNW server's two internal endpoints. Accepts every
60 /// key and authorizes every operation: this test is about transport
61 /// backpressure, and the authorization paths have their own coverage.
62 async fn spawn_stub_api(repo_path: String) -> anyhow::Result<String> {
63 let listener = TcpListener::bind("127.0.0.1:0").await?;
64 let addr = listener.local_addr()?;
65
66 tokio::spawn(async move {
67 loop {
68 let Ok((mut sock, _)) = listener.accept().await else {
69 break;
70 };
71 let repo_path = repo_path.clone();
72 tokio::spawn(async move {
73 // Read up to the end of the headers; that carries the request
74 // line, which is all this stub dispatches on. Any body after it
75 // is ignored, and the connection is closed per response, so
76 // there is nothing to keep parsing.
77 let mut buf = vec![0u8; 8192];
78 let Ok(n) = sock.read(&mut buf).await else {
79 return;
80 };
81 let req = String::from_utf8_lossy(&buf[..n]);
82
83 let body = if req.contains("/api/internal/git/authorize") {
84 format!(r#"{{"repo_path":"{repo_path}"}}"#)
85 } else {
86 r#"{"user_id":"test-user","username":"max","display_name":null,
87 "creator_tier":null,"can_create_projects":true,"suspended":false,
88 "actor_token":"stub-actor-token"}"#
89 .replace(['\n', ' '], "")
90 };
91
92 let resp = format!(
93 "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
94 body.len()
95 );
96 let _ = sock.write_all(resp.as_bytes()).await;
97 let _ = sock.flush().await;
98 });
99 }
100 });
101
102 Ok(format!("http://{addr}"))
103 }
104
105 /// Start the real SSH server on an ephemeral port. Returns the port.
106 ///
107 /// `git_user` is empty on purpose: it puts `spawn_git_process` on its direct
108 /// path, so the test needs no sudo grant. Everything else is production code.
109 async fn spawn_ssh_server(api_url: String) -> anyhow::Result<u16> {
110 let listener = TcpListener::bind("127.0.0.1:0").await?;
111 let port = listener.local_addr()?.port();
112
113 let host_key =
114 russh::keys::PrivateKey::random(&mut rand::rng(), russh::keys::Algorithm::Ed25519)?;
115
116 let mut methods = russh::MethodSet::empty();
117 methods.push(russh::MethodKind::PublicKey);
118
119 let config = Arc::new(russh::server::Config {
120 methods,
121 keys: vec![host_key],
122 auth_rejection_time: Duration::from_millis(0),
123 auth_rejection_time_initial: Some(Duration::from_millis(0)),
124 ..Default::default()
125 });
126
127 let api = MnwApiClient::new(api_url, "stub-service-token".to_string());
128 let staging = Arc::new(std::env::temp_dir().join("mnw-cli-backpressure-staging"));
129 let mut server = MnwServer::new(
130 api,
131 staging,
132 String::new(),
133 Arc::new(AuthRateLimiter::new()),
134 );
135
136 tokio::spawn(async move {
137 let _ = server.run_on_socket(config, &listener).await;
138 });
139
140 Ok(port)
141 }
142
143 /// A push larger than russh's channel buffer completes instead of hanging.
144 ///
145 /// The whole point is the size: shrink the payload under ~100 packets and this
146 /// test passes against the deadlocked handler too, which is exactly why the bug
147 /// survived `makeover-touch` pushing fine the day before `shop` stalled.
148 #[tokio::test]
149 async fn a_push_larger_than_the_channel_buffer_completes() {
150 let tmp = tempfile::tempdir().unwrap();
151 let root = tmp.path();
152
153 // Client identity. The stub API accepts whatever fingerprint it is asked
154 // about, so the key only has to exist and be offered.
155 let key_path = root.join("id_ed25519");
156 run(
157 "ssh-keygen",
158 &[
159 "-t",
160 "ed25519",
161 "-N",
162 "",
163 "-C",
164 "backpressure-test",
165 "-f",
166 key_path.to_str().unwrap(),
167 ],
168 root,
169 )
170 .await
171 .expect("ssh-keygen; is openssh installed?");
172
173 // The bare repo the push lands in, standing in for what the server would
174 // have auto-created.
175 let bare = root.join("max").join("backpressure.git");
176 std::fs::create_dir_all(bare.parent().unwrap()).unwrap();
177 run(
178 "git",
179 &["init", "--bare", "-b", "main", bare.to_str().unwrap()],
180 root,
181 )
182 .await
183 .unwrap();
184
185 // Source repo carrying incompressible data, so the pack stays big.
186 let src = root.join("src");
187 std::fs::create_dir_all(&src).unwrap();
188 run("git", &["init", "-b", "main"], &src).await.unwrap();
189 run("git", &["config", "user.email", "t@example.com"], &src)
190 .await
191 .unwrap();
192 run("git", &["config", "user.name", "test"], &src)
193 .await
194 .unwrap();
195
196 let mut blob = vec![0u8; PAYLOAD_BYTES];
197 rand::fill(&mut blob[..]);
198 std::fs::write(src.join("payload.bin"), &blob).unwrap();
199 run("git", &["add", "."], &src).await.unwrap();
200 run("git", &["commit", "-m", "payload"], &src)
201 .await
202 .unwrap();
203
204 let api_url = spawn_stub_api(bare.to_str().unwrap().to_string())
205 .await
206 .unwrap();
207 let port = spawn_ssh_server(api_url).await.unwrap();
208
209 let ssh_command = format!(
210 "ssh -p {port} -i {} -o IdentitiesOnly=yes -o StrictHostKeyChecking=no \
211 -o UserKnownHostsFile=/dev/null -o LogLevel=ERROR",
212 key_path.display()
213 );
214
215 let push = Command::new("git")
216 .args([
217 "push",
218 &format!("ssh://git@127.0.0.1:{port}/max/backpressure.git"),
219 "main",
220 ])
221 .current_dir(&src)
222 .env("GIT_SSH_COMMAND", &ssh_command)
223 .stdout(Stdio::piped())
224 .stderr(Stdio::piped())
225 .output();
226
227 let out = tokio::time::timeout(PUSH_TIMEOUT, push)
228 .await
229 .expect(
230 "push hung: the session loop stopped reading mid-pack, which is the \
231 deadlock this test exists to catch",
232 )
233 .unwrap();
234
235 assert!(
236 out.status.success(),
237 "push failed: {}",
238 String::from_utf8_lossy(&out.stderr)
239 );
240
241 // The bytes actually landed, not merely a session that failed politely.
242 let refs = Command::new("git")
243 .args(["--git-dir", bare.to_str().unwrap(), "rev-parse", "main"])
244 .output()
245 .await
246 .unwrap();
247 assert!(
248 refs.status.success(),
249 "pushed ref missing from the bare repo: {}",
250 String::from_utf8_lossy(&refs.stderr)
251 );
252 }
253