netget/server/beanstalkd/
wire.rs

1//! The beanstalkd wire format: command parsing, reply rendering, and the check that a reply
2//! answers the command it is sent for.
3//!
4//! Everything here is a pure function of its arguments, shared by the session loop, the action
5//! executor and the tests. **The model never writes a byte of framing**: it supplies job ids,
6//! job bodies, a status word from a fixed list and the key/value pairs of a stats report, and
7//! this module decides the reply line, the byte count and the YAML. A job body that happens to
8//! contain `\r\nINSERTED 9\r\n` is counted into `<bytes>` and read by the client as body, not as
9//! a second reply.
10//!
11//! The protocol is upstream's `doc/protocol.txt` (beanstalkd 1.13): lines end in CRLF,
12//! commands are lower-case and case-sensitive, tube names are at most 200 bytes of
13//! `[A-Za-z0-9+/;.$_()-]` and do not begin with `-`, and a job body is `<bytes>` octets followed
14//! by CRLF.
15
16/// The longest command line a peer may send, **including** the trailing CRLF.
17///
18/// Upstream's `LINE_BUF_SIZE` is 224 (a 200-byte tube name plus the longest verb and its
19/// numbers). Upstream answers an over-long line with `BAD_FORMAT`; NetGet does too, and then
20/// closes, because there is no way to find the start of the next command in a stream that
21/// ignored the limit.
22pub const MAX_LINE_BYTES: usize = 224;
23
24/// The largest job body `put` may declare: upstream's default `max-job-size` (`-z`, 65535).
25///
26/// Checked against the **declared** `<bytes>` before a single body byte is read, so a `put`
27/// declaring four gigabytes is answered `JOB_TOO_BIG` at once and nothing is buffered for it.
28/// The same limit applies to the bodies the model hands out in `RESERVED`/`FOUND`, because a
29/// real beanstalkd never holds a job larger than it would have accepted.
30pub const MAX_JOB_BYTES: usize = 65_535;
31
32/// The longest tube name (upstream's `MAX_TUBE_NAME_LEN` - 1).
33pub const MAX_TUBE_NAME: usize = 200;
34
35/// How many bytes of an oversize body NetGet reads and discards after `JOB_TOO_BIG` to keep the
36/// connection in step, as upstream does. A `put` declaring more than this is answered and then
37/// closed instead: skipping a gigabyte to stay in sync is not worth holding the connection for.
38pub const MAX_DISCARD_BYTES: u64 = 1024 * 1024;
39
40/// The most entries a stats report or tube list the model writes may carry.
41pub const MAX_YAML_ENTRIES: usize = 512;
42
43/// The longest scalar value in a stats report.
44pub const MAX_YAML_VALUE: usize = 200;
45
46/// Is `name` a tube name upstream would accept?
47pub fn valid_tube_name(name: &str) -> bool {
48    !name.is_empty()
49        && name.len() <= MAX_TUBE_NAME
50        && !name.starts_with('-')
51        && name
52            .bytes()
53            .all(|b| b.is_ascii_alphanumeric() || b"-+/;.$_()".contains(&b))
54}
55
56/// One parsed command line.
57#[derive(Debug, Clone, PartialEq, Eq)]
58pub enum Command {
59    Put {
60        priority: u32,
61        delay: u32,
62        ttr: u32,
63        /// The declared body length. `None` when the number does not fit a `u64`, which is
64        /// larger than any bound and is refused the same way.
65        bytes: Option<u64>,
66    },
67    Use(String),
68    Reserve,
69    ReserveWithTimeout(u32),
70    ReserveJob(u64),
71    Delete(u64),
72    Release {
73        id: u64,
74        priority: u32,
75        delay: u32,
76    },
77    Bury {
78        id: u64,
79        priority: u32,
80    },
81    Touch(u64),
82    Watch(String),
83    Ignore(String),
84    Peek(u64),
85    PeekReady,
86    PeekDelayed,
87    PeekBuried,
88    Kick(u64),
89    KickJob(u64),
90    StatsJob(u64),
91    StatsTube(String),
92    Stats,
93    ListTubes,
94    ListTubeUsed,
95    ListTubesWatched,
96    PauseTube {
97        tube: String,
98        delay: u32,
99    },
100    Quit,
101    /// A known command with missing, extra or malformed arguments.
102    BadFormat,
103    /// Not a beanstalkd command.
104    Unknown,
105}
106
107fn num<T: std::str::FromStr>(s: &str) -> Option<T> {
108    // Digits only: upstream's strtoul would also take a sign or leading space, but no client
109    // sends either and accepting `-1` as a huge unsigned number is not a courtesy.
110    if s.is_empty() || !s.bytes().all(|b| b.is_ascii_digit()) {
111        return None;
112    }
113    s.parse().ok()
114}
115
116/// Parse one command line, without its CRLF. Commands are case-sensitive, as upstream's are.
117pub fn parse_command(line: &str) -> Command {
118    // Upstream separates arguments with single spaces; splitting on runs of spaces accepts the
119    // same lines and a few more, and never splits a tube name (which cannot contain a space).
120    let parts: Vec<&str> = line.split(' ').filter(|p| !p.is_empty()).collect();
121    let Some((verb, args)) = parts.split_first() else {
122        return Command::Unknown;
123    };
124    let tube = |s: &str| -> Option<String> { valid_tube_name(s).then(|| s.to_string()) };
125    let parsed = match (*verb, args) {
126        ("put", [p, d, t, b]) => (|| {
127            Some(Command::Put {
128                priority: num(p)?,
129                delay: num(d)?,
130                ttr: num(t)?,
131                bytes: if b.is_empty() || !b.bytes().all(|c| c.is_ascii_digit()) {
132                    return None;
133                } else {
134                    b.parse().ok()
135                },
136            })
137        })(),
138        ("use", [t]) => tube(t).map(Command::Use),
139        ("reserve", []) => Some(Command::Reserve),
140        ("reserve-with-timeout", [s]) => num(s).map(Command::ReserveWithTimeout),
141        ("reserve-job", [id]) => num(id).map(Command::ReserveJob),
142        ("delete", [id]) => num(id).map(Command::Delete),
143        ("release", [id, p, d]) => (|| {
144            Some(Command::Release {
145                id: num(id)?,
146                priority: num(p)?,
147                delay: num(d)?,
148            })
149        })(),
150        ("bury", [id, p]) => (|| {
151            Some(Command::Bury {
152                id: num(id)?,
153                priority: num(p)?,
154            })
155        })(),
156        ("touch", [id]) => num(id).map(Command::Touch),
157        ("watch", [t]) => tube(t).map(Command::Watch),
158        ("ignore", [t]) => tube(t).map(Command::Ignore),
159        ("peek", [id]) => num(id).map(Command::Peek),
160        ("peek-ready", []) => Some(Command::PeekReady),
161        ("peek-delayed", []) => Some(Command::PeekDelayed),
162        ("peek-buried", []) => Some(Command::PeekBuried),
163        ("kick", [n]) => num(n).map(Command::Kick),
164        ("kick-job", [id]) => num(id).map(Command::KickJob),
165        ("stats-job", [id]) => num(id).map(Command::StatsJob),
166        ("stats-tube", [t]) => tube(t).map(Command::StatsTube),
167        ("stats", []) => Some(Command::Stats),
168        ("list-tubes", []) => Some(Command::ListTubes),
169        ("list-tube-used", []) => Some(Command::ListTubeUsed),
170        ("list-tubes-watched", []) => Some(Command::ListTubesWatched),
171        ("pause-tube", [t, d]) => (|| {
172            Some(Command::PauseTube {
173                tube: tube(t)?,
174                delay: num(d)?,
175            })
176        })(),
177        ("quit", []) => Some(Command::Quit),
178        (
179            "put"
180            | "use"
181            | "reserve"
182            | "reserve-with-timeout"
183            | "reserve-job"
184            | "delete"
185            | "release"
186            | "bury"
187            | "touch"
188            | "watch"
189            | "ignore"
190            | "peek"
191            | "peek-ready"
192            | "peek-delayed"
193            | "peek-buried"
194            | "kick"
195            | "kick-job"
196            | "stats-job"
197            | "stats-tube"
198            | "stats"
199            | "list-tubes"
200            | "list-tube-used"
201            | "list-tubes-watched"
202            | "pause-tube"
203            | "quit",
204            _,
205        ) => None,
206        _ => return Command::Unknown,
207    };
208    parsed.unwrap_or(Command::BadFormat)
209}
210
211/// The status words `send_beanstalkd_status` may send. Everything else on the wire — `USING`,
212/// `WATCHING`, `NOT_IGNORED`, `BAD_FORMAT`, `UNKNOWN_COMMAND`, `JOB_TOO_BIG`, `EXPECTED_CRLF`
213/// — is NetGet's own answer about the connection, and `INSERTED`/`RESERVED`/`FOUND`/`OK` have
214/// actions of their own because they carry data.
215pub const STATUS_WORDS: &[&str] = &[
216    "DELETED",
217    "RELEASED",
218    "BURIED",
219    "TOUCHED",
220    "KICKED",
221    "PAUSED",
222    "NOT_FOUND",
223    "TIMED_OUT",
224    "DEADLINE_SOON",
225    "DRAINING",
226    "OUT_OF_MEMORY",
227    "INTERNAL_ERROR",
228];
229
230/// `STATUS` or `KICKED <count>`.
231pub fn render_status(status: &str, count: Option<u64>) -> Result<String, String> {
232    let word = status.trim().to_ascii_uppercase();
233    if !STATUS_WORDS.contains(&word.as_str()) {
234        return Err(format!(
235            "status must be one of {}, got {status:?}",
236            STATUS_WORDS.join(", ")
237        ));
238    }
239    match (word.as_str(), count) {
240        ("KICKED", Some(n)) => Ok(format!("KICKED {n}\r\n")),
241        (_, Some(_)) => Err(format!(
242            "'count' is only meaningful with KICKED, not {word}"
243        )),
244        (_, None) => Ok(format!("{word}\r\n")),
245    }
246}
247
248/// `INSERTED <id>`, or `BURIED <id>` when the job went straight to the buried list.
249pub fn render_inserted(id: u64, buried: bool) -> Result<String, String> {
250    check_id(id)?;
251    Ok(format!(
252        "{} {id}\r\n",
253        if buried { "BURIED" } else { "INSERTED" }
254    ))
255}
256
257/// Which job-carrying reply to write.
258#[derive(Debug, Clone, Copy, PartialEq, Eq)]
259pub enum JobReply {
260    Reserved,
261    Found,
262}
263
264/// `RESERVED <id> <bytes>\r\n<body>\r\n` or `FOUND …`. The count is the body's byte length,
265/// computed here, so whatever the body contains is read as body.
266pub fn render_job(kind: JobReply, id: u64, body: &str) -> Result<String, String> {
267    check_id(id)?;
268    if body.len() > MAX_JOB_BYTES {
269        return Err(format!(
270            "job body is {} bytes; beanstalkd's max-job-size is {MAX_JOB_BYTES}",
271            body.len()
272        ));
273    }
274    let word = match kind {
275        JobReply::Reserved => "RESERVED",
276        JobReply::Found => "FOUND",
277    };
278    Ok(format!("{word} {id} {}\r\n{body}\r\n", body.len()))
279}
280
281fn check_id(id: u64) -> Result<(), String> {
282    if id == 0 {
283        return Err("job ids start at 1".to_string());
284    }
285    Ok(())
286}
287
288/// A scalar in a stats report, rendered the way beanstalkd renders it.
289fn yaml_scalar(key: &str, value: &serde_json::Value) -> Result<String, String> {
290    use serde_json::Value;
291    let text = match value {
292        Value::Number(n) => n.to_string(),
293        Value::Bool(b) => b.to_string(),
294        Value::String(s) => s.clone(),
295        Value::Null => return Err(format!("stat '{key}' is null; give a number or a string")),
296        _ => {
297            return Err(format!(
298                "stat '{key}' is a list or an object; a beanstalkd stats report is flat \
299                 key: value lines"
300            ))
301        }
302    };
303    // Clients decode the report as ASCII (greenstalk calls `.decode("ascii")`), and a newline
304    // would start a line of its own. Refused rather than rewritten: the model wrote the value
305    // and can be told, and a rewritten value would disagree with the decision the log records.
306    if text.is_empty()
307        || text.len() > MAX_YAML_VALUE
308        || !text.bytes().all(|b| (0x20..0x7f).contains(&b))
309    {
310        return Err(format!(
311            "stat '{key}' must be 1..={MAX_YAML_VALUE} printable ASCII characters, got {text:?}"
312        ));
313    }
314    Ok(text)
315}
316
317fn valid_stat_key(key: &str) -> bool {
318    !key.is_empty()
319        && key.len() <= 64
320        && key
321            .bytes()
322            .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_')
323}
324
325/// `OK <bytes>\r\n---\nkey: value\n…\r\n` — the YAML dictionary `stats`, `stats-tube` and
326/// `stats-job` answer with.
327pub fn render_stats(stats: &serde_json::Map<String, serde_json::Value>) -> Result<String, String> {
328    if stats.len() > MAX_YAML_ENTRIES {
329        return Err(format!("at most {MAX_YAML_ENTRIES} stats"));
330    }
331    let mut yaml = String::from("---\n");
332    for (key, value) in stats {
333        if !valid_stat_key(key) {
334            return Err(format!(
335                "stat name {key:?} must be letters, digits, '-' or '_' (like current-jobs-ready)"
336            ));
337        }
338        yaml.push_str(&format!("{key}: {}\n", yaml_scalar(key, value)?));
339    }
340    Ok(ok_block(&yaml))
341}
342
343/// `OK <bytes>\r\n---\n- name\n…\r\n` — the YAML list `list-tubes` and `list-tubes-watched`
344/// answer with. Every entry must be a tube name, so no entry can break the list.
345pub fn render_tube_list<S: AsRef<str>>(tubes: &[S]) -> Result<String, String> {
346    if tubes.len() > MAX_YAML_ENTRIES {
347        return Err(format!("at most {MAX_YAML_ENTRIES} tubes"));
348    }
349    let mut yaml = String::from("---\n");
350    for tube in tubes {
351        let tube = tube.as_ref();
352        if !valid_tube_name(tube) {
353            return Err(format!(
354                "{tube:?} is not a tube name: 1-{MAX_TUBE_NAME} of letters, digits and \
355                 -+/;.$_() not starting with '-'"
356            ));
357        }
358        yaml.push_str(&format!("- {tube}\n"));
359    }
360    Ok(ok_block(&yaml))
361}
362
363fn ok_block(yaml: &str) -> String {
364    format!("OK {}\r\n{yaml}\r\n", yaml.len())
365}
366
367/// The first line of a rendered reply, split into its word and arguments.
368pub fn reply_head(reply: &[u8]) -> (String, Vec<String>) {
369    let end = reply
370        .windows(2)
371        .position(|w| w == b"\r\n")
372        .unwrap_or(reply.len());
373    let line = String::from_utf8_lossy(&reply[..end]);
374    let mut parts = line.split(' ').map(str::to_string);
375    let word = parts.next().unwrap_or_default();
376    (word, parts.collect())
377}
378
379/// Does `reply` — something the executor rendered — answer `command`?
380///
381/// The executor is stateless and cannot know which command it is answering, so a model can
382/// produce a well-formed reply to the wrong command: `RESERVED` to a `delete`, a stats
383/// dictionary to `list-tubes`, `KICKED` without a count to `kick`. Each would be read by the
384/// client as something else. `OUT_OF_MEMORY` and `INTERNAL_ERROR` answer anything.
385pub fn reply_fits(command: &Command, reply: &[u8]) -> bool {
386    let (word, args) = reply_head(reply);
387    let n = args.len();
388    if n == 0 && (word == "OUT_OF_MEMORY" || word == "INTERNAL_ERROR") {
389        return true;
390    }
391    // For `OK <bytes>`, the YAML payload: a list has only `- ` lines, a dictionary none.
392    let yaml_lines = || -> Option<Vec<String>> {
393        let size: usize = args.first()?.parse().ok()?;
394        let start = reply.windows(2).position(|w| w == b"\r\n")? + 2;
395        let payload = reply.get(start..start.checked_add(size)?)?;
396        let text = std::str::from_utf8(payload).ok()?;
397        let body = text.strip_prefix("---\n")?;
398        Some(body.lines().map(str::to_string).collect())
399    };
400    let is_list = || yaml_lines().is_some_and(|l| l.iter().all(|l| l.starts_with("- ")));
401    let is_dict = || yaml_lines().is_some_and(|l| l.iter().all(|l| !l.starts_with("- ")));
402    use Command::*;
403    match command {
404        Put { .. } => matches!(
405            (word.as_str(), n),
406            ("INSERTED", 1) | ("BURIED", 1) | ("DRAINING", 0)
407        ),
408        Reserve => matches!((word.as_str(), n), ("RESERVED", 2) | ("DEADLINE_SOON", 0)),
409        ReserveWithTimeout(_) => matches!(
410            (word.as_str(), n),
411            ("RESERVED", 2) | ("DEADLINE_SOON", 0) | ("TIMED_OUT", 0)
412        ),
413        ReserveJob(_) => matches!((word.as_str(), n), ("RESERVED", 2) | ("NOT_FOUND", 0)),
414        Delete(_) => matches!((word.as_str(), n), ("DELETED", 0) | ("NOT_FOUND", 0)),
415        Release { .. } => matches!(
416            (word.as_str(), n),
417            ("RELEASED", 0) | ("BURIED", 0) | ("NOT_FOUND", 0)
418        ),
419        Bury { .. } => matches!((word.as_str(), n), ("BURIED", 0) | ("NOT_FOUND", 0)),
420        Touch(_) => matches!((word.as_str(), n), ("TOUCHED", 0) | ("NOT_FOUND", 0)),
421        Peek(_) | PeekReady | PeekDelayed | PeekBuried => {
422            matches!((word.as_str(), n), ("FOUND", 2) | ("NOT_FOUND", 0))
423        }
424        Kick(_) => word == "KICKED" && n == 1,
425        KickJob(_) => matches!((word.as_str(), n), ("KICKED", 0) | ("NOT_FOUND", 0)),
426        Stats => word == "OK" && n == 1 && is_dict(),
427        StatsJob(_) | StatsTube(_) => {
428            (word == "OK" && n == 1 && is_dict()) || (word == "NOT_FOUND" && n == 0)
429        }
430        ListTubes => word == "OK" && n == 1 && is_list(),
431        PauseTube { .. } => matches!((word.as_str(), n), ("PAUSED", 0) | ("NOT_FOUND", 0)),
432        _ => false,
433    }
434}
435
436/// Is this reply a refusal (the model saying no) rather than an answer?
437pub fn is_refusal(reply: &[u8]) -> bool {
438    let (word, _) = reply_head(reply);
439    matches!(
440        word.as_str(),
441        "NOT_FOUND"
442            | "TIMED_OUT"
443            | "DEADLINE_SOON"
444            | "DRAINING"
445            | "OUT_OF_MEMORY"
446            | "INTERNAL_ERROR"
447    )
448}
netget/server/beanstalkd/wire.rs

netget/server/beanstalkd/
wire.rs
