bynk/
ready.rs

1//! `bynk dev --ready-json <FILE>` (#2138, end-to-end testing track §3.5,
2//! DECISION F): a machine-readable readiness signal.
3//!
4//! Once every served context has answered an HTTP probe, `bynk dev` writes
5//! `{ "<worker>": "<url>" }` to the file. A test runner (`bynk test --against
6//! dev`) or a wrapper such as `bynk-lang/bynk-dev` waits for the file instead
7//! of looping on a warm-up request. The file is removed when the session starts
8//! (so a stale one is never read as ready) and when it stops.
9//!
10//! Readiness means each Worker answers a request. It does not mean the
11//! dev-registry Service Bindings between them have connected yet: they
12//! converge shortly after start.
13
14use std::collections::{BTreeMap, BTreeSet};
15use std::io::{Read, Write};
16use std::net::{SocketAddr, TcpStream, ToSocketAddrs};
17use std::path::{Path, PathBuf};
18use std::time::{Duration, Instant};
19
20/// How long one probe waits to connect, and then for the response's first
21/// bytes. Short, because the probe runs inside the dev watch loop.
22const PROBE_TIMEOUT: Duration = Duration::from_millis(300);
23
24/// How long without readiness before `bynk dev` notes, once, which contexts
25/// haven't answered.
26pub const SLOW_NOTE_AFTER: Duration = Duration::from_secs(30);
27
28/// Whether something answering HTTP listens at `host:port`. Sends a `HEAD /`
29/// (safe per RFC 9110, so a `GET /` route with side effects is never
30/// tripped) and takes any `HTTP/` status line as an answer, a 404 included:
31/// every emitted Worker has a `fetch`, so any status means it is up.
32pub fn probe(host: &str, port: u16) -> bool {
33    // `localhost` may resolve to `::1` and `127.0.0.1`; the server may listen
34    // on either, so try each address.
35    let Some(mut stream) = resolve(host, port)
36        .into_iter()
37        .find_map(|addr| TcpStream::connect_timeout(&addr, PROBE_TIMEOUT).ok())
38    else {
39        return false;
40    };
41    let _ = stream.set_read_timeout(Some(PROBE_TIMEOUT));
42    let _ = stream.set_write_timeout(Some(PROBE_TIMEOUT));
43    let request = format!("HEAD / HTTP/1.1\r\nHost: {host}:{port}\r\nConnection: close\r\n\r\n");
44    if stream.write_all(request.as_bytes()).is_err() {
45        return false;
46    }
47    let mut head = [0u8; 5];
48    let mut read = 0;
49    while read < head.len() {
50        match stream.read(&mut head[read..]) {
51            Ok(0) | Err(_) => return false,
52            Ok(n) => read += n,
53        }
54    }
55    &head == b"HTTP/"
56}
57
58fn resolve(host: &str, port: u16) -> Vec<SocketAddr> {
59    (host, port)
60        .to_socket_addrs()
61        .map(Iterator::collect)
62        .unwrap_or_default()
63}
64
65/// Why the wrangler passthrough `args` serve somewhere the readiness probe
66/// (plain HTTP on localhost) can't reach: `-- --local-protocol https`, or
67/// `-- --ip` bound to an address that isn't loopback.
68pub fn unprobeable(args: &[String]) -> Option<String> {
69    let value = |flag: &str| {
70        args.iter().enumerate().find_map(|(i, a)| {
71            a.strip_prefix(&format!("{flag}="))
72                .map(str::to_string)
73                .or_else(|| (a == flag).then(|| args.get(i + 1).cloned()).flatten())
74        })
75    };
76    if value("--local-protocol").as_deref() == Some("https") {
77        return Some("`-- --local-protocol https` serves over TLS".to_string());
78    }
79    match value("--ip") {
80        Some(ip)
81            if !matches!(
82                ip.as_str(),
83                "localhost" | "127.0.0.1" | "::1" | "0.0.0.0" | "*"
84            ) =>
85        {
86            Some(format!("`-- --ip {ip}` serves away from localhost"))
87        }
88        _ => None,
89    }
90}
91
92/// The readiness file's contents: each worker's URL, by worker name.
93pub fn ready_json(urls: &BTreeMap<String, String>) -> String {
94    serde_json::to_string_pretty(urls).expect("a string map serialises") + "\n"
95}
96
97/// Write `contents` to `path` atomically: to a sibling temporary file first,
98/// then renamed over `path`, so a reader never sees a half-written file.
99pub fn write_atomically(path: &Path, contents: &str) -> std::io::Result<()> {
100    let mut tmp = path.as_os_str().to_owned();
101    tmp.push(".tmp");
102    let tmp = PathBuf::from(tmp);
103    std::fs::write(&tmp, contents)?;
104    std::fs::rename(&tmp, path)
105}
106
107/// Remove `path` if it is there.
108pub fn remove(path: &Path) {
109    let _ = std::fs::remove_file(path);
110}
111
112/// The pending readiness of one `bynk dev` session: the workers still to
113/// answer, and where to write the file once none remain.
114#[derive(Debug)]
115pub struct Readiness {
116    file: PathBuf,
117    /// Every served worker's port and URL.
118    workers: BTreeMap<String, (u16, String)>,
119    /// The workers that have answered a probe; each is probed until it does.
120    answered: BTreeSet<String>,
121    written: bool,
122    started: Instant,
123    noted_slow: bool,
124}
125
126impl Readiness {
127    /// A session serving `contexts` (worker name, port, URL), writing `file`
128    /// once all of them answer. Removes a stale `file` now.
129    pub fn new(file: PathBuf, contexts: impl IntoIterator<Item = (String, u16, String)>) -> Self {
130        remove(&file);
131        Readiness {
132            file,
133            workers: contexts
134                .into_iter()
135                .map(|(name, port, url)| (name, (port, url)))
136                .collect(),
137            answered: BTreeSet::new(),
138            written: false,
139            started: Instant::now(),
140            noted_slow: false,
141        }
142    }
143
144    /// Probe each worker that hasn't answered yet once (via `probe`, at
145    /// `localhost` and its port) and, the first time every worker has
146    /// answered, write the file. Returns the write's result that one time,
147    /// and `None` otherwise: while a worker is still booting, and on every
148    /// poll after the write. A failed write is reported once and not retried:
149    /// the session keeps serving, without a readiness file.
150    pub fn poll(&mut self, probe: impl Fn(&str, u16) -> bool) -> Option<std::io::Result<()>> {
151        if self.written {
152            return None;
153        }
154        for (name, (port, _)) in &self.workers {
155            if !self.answered.contains(name) && probe("localhost", *port) {
156                self.answered.insert(name.clone());
157            }
158        }
159        if self.answered.len() < self.workers.len() {
160            return None;
161        }
162        self.written = true;
163        let urls: BTreeMap<String, String> = self
164            .workers
165            .iter()
166            .map(|(name, (_, url))| (name.clone(), url.clone()))
167            .collect();
168        Some(write_atomically(&self.file, &ready_json(&urls)))
169    }
170
171    /// Once, when `now` is [`SLOW_NOTE_AFTER`] past the session's start and
172    /// the file still isn't written: a note naming the contexts that haven't
173    /// answered, so a probe that can never succeed (a TLS-only or
174    /// non-loopback server) doesn't hang silently.
175    pub fn slow_note(&mut self, now: Instant) -> Option<String> {
176        if self.written || self.noted_slow || now.duration_since(self.started) < SLOW_NOTE_AFTER {
177            return None;
178        }
179        self.noted_slow = true;
180        let pending: Vec<String> = self
181            .workers
182            .iter()
183            .filter(|(name, _)| !self.answered.contains(*name))
184            .map(|(name, (_, url))| format!("`{name}` ({url})"))
185            .collect();
186        Some(format!(
187            "bynk dev: still waiting for {} to answer before writing the `--ready-json` file; it's written only once every context answers a plain-HTTP probe on localhost",
188            pending.join(", ")
189        ))
190    }
191
192    /// The session is ending: remove the file, so it is never read as ready.
193    pub fn stop(&self) {
194        remove(&self.file);
195    }
196}
197
198#[cfg(test)]
199mod tests {
200    use super::*;
201    use std::net::TcpListener;
202
203    /// A stand-in that answers one request with `reply`, then closes.
204    fn stand_in(reply: &'static [u8]) -> u16 {
205        let listener = TcpListener::bind("127.0.0.1:0").expect("bind a local port");
206        let port = listener.local_addr().unwrap().port();
207        std::thread::spawn(move || {
208            for stream in listener.incoming() {
209                let Ok(mut stream) = stream else { return };
210                let mut buf = [0u8; 512];
211                let _ = stream.read(&mut buf);
212                let _ = stream.write_all(reply);
213            }
214        });
215        port
216    }
217
218    fn free_port() -> u16 {
219        // Bound and dropped: nothing listens on it afterwards.
220        TcpListener::bind("127.0.0.1:0")
221            .unwrap()
222            .local_addr()
223            .unwrap()
224            .port()
225    }
226
227    #[test]
228    fn tls_and_non_loopback_passthroughs_are_flagged() {
229        let args = |a: &[&str]| a.iter().map(|s| s.to_string()).collect::<Vec<_>>();
230        assert!(unprobeable(&args(&["--local-protocol", "https"])).is_some());
231        assert!(unprobeable(&args(&["--local-protocol=https"])).is_some());
232        assert!(unprobeable(&args(&["--ip", "192.168.1.5"])).is_some());
233        assert!(unprobeable(&args(&["--ip", "127.0.0.1"])).is_none());
234        assert!(unprobeable(&args(&["--local-protocol", "http", "--var", "K:V"])).is_none());
235        assert!(unprobeable(&[]).is_none());
236    }
237
238    #[test]
239    fn a_404_is_an_answer() {
240        let port = stand_in(b"HTTP/1.1 404 Not Found\r\ncontent-length: 0\r\n\r\n");
241        assert!(probe("127.0.0.1", port));
242    }
243
244    #[test]
245    fn nothing_listening_is_no_answer() {
246        assert!(!probe("127.0.0.1", free_port()));
247    }
248
249    #[test]
250    fn a_listener_that_never_replies_is_no_answer() {
251        let listener = TcpListener::bind("127.0.0.1:0").expect("bind a local port");
252        let port = listener.local_addr().unwrap().port();
253        // Accepts, then holds the connection open without a byte.
254        std::thread::spawn(move || {
255            let held: Vec<_> = listener.incoming().take(4).collect();
256            std::thread::sleep(Duration::from_secs(5));
257            drop(held);
258        });
259        let started = Instant::now();
260        assert!(!probe("127.0.0.1", port));
261        assert!(
262            started.elapsed() < Duration::from_secs(3),
263            "the probe gives up at its timeout"
264        );
265    }
266
267    #[test]
268    fn only_unanswered_workers_are_probed_and_a_slow_session_is_noted_once() {
269        let dir = std::env::temp_dir().join(format!("bynk-ready-slow-{}", std::process::id()));
270        std::fs::create_dir_all(&dir).unwrap();
271        let mut r = Readiness::new(
272            dir.join("ready.json"),
273            [
274                ("a".to_string(), 1, "http://localhost:1".to_string()),
275                ("b".to_string(), 2, "http://localhost:2".to_string()),
276            ],
277        );
278        let probed = std::cell::RefCell::new(Vec::new());
279        let probe = |_: &str, port: u16| {
280            probed.borrow_mut().push(port);
281            port == 1
282        };
283        assert!(r.poll(probe).is_none());
284        assert!(r.poll(probe).is_none());
285        assert_eq!(
286            *probed.borrow(),
287            [1, 2, 2],
288            "`a` is probed until it answers, not after"
289        );
290        assert!(r.slow_note(Instant::now()).is_none(), "not yet");
291        let later = Instant::now() + SLOW_NOTE_AFTER;
292        let note = r.slow_note(later).expect("noted after the delay");
293        assert!(
294            note.contains("`b` (http://localhost:2)") && !note.contains("`a`"),
295            "{note}"
296        );
297        assert!(r.slow_note(later).is_none(), "noted once");
298        let _ = std::fs::remove_dir_all(&dir);
299    }
300
301    #[test]
302    fn a_non_http_reply_is_no_answer() {
303        let port = stand_in(b"SSH-2.0-nope\r\n");
304        assert!(!probe("127.0.0.1", port));
305    }
306
307    #[test]
308    fn the_file_is_written_once_every_worker_answers_and_removed_on_stop() {
309        let dir = std::env::temp_dir().join(format!("bynk-ready-{}", std::process::id()));
310        std::fs::create_dir_all(&dir).unwrap();
311        let file = dir.join("ready.json");
312        std::fs::write(&file, "stale").unwrap();
313        let mut r = Readiness::new(
314            file.clone(),
315            [
316                ("a".to_string(), 1, "http://localhost:1".to_string()),
317                ("b".to_string(), 2, "http://localhost:2".to_string()),
318            ],
319        );
320        assert!(!file.exists(), "a stale file is removed at start");
321        assert!(r.poll(|_, port| port == 1).is_none());
322        assert!(
323            !file.exists(),
324            "not written while a worker is still booting"
325        );
326        assert!(r.poll(|_, _| true).expect("written now").is_ok());
327        assert!(r.poll(|_, _| true).is_none(), "written once");
328        let written: BTreeMap<String, String> =
329            serde_json::from_str(&std::fs::read_to_string(&file).unwrap()).unwrap();
330        assert_eq!(written["a"], "http://localhost:1");
331        assert_eq!(written["b"], "http://localhost:2");
332        r.stop();
333        assert!(!file.exists(), "removed when the session stops");
334        let _ = std::fs::remove_dir_all(&dir);
335    }
336}
bynk/ready.rs

bynk/
ready.rs
