netget/server/gearman/
wire.rs

1//! The Gearman wire formats: the binary packet protocol and the text admin protocol.
2//!
3//! Everything here is a pure function of its arguments, shared by the session loop, the action
4//! executor, the tests and the `gearman_packet` fuzz target. **The model never writes a
5//! packet**: it supplies a result, a pair of progress numbers or an error text, and this module
6//! decides the magic, the type, the length and the NUL separators.
7//!
8//! Binary packet (protocol.txt, gearmand 1.1):
9//!
10//! ```text
11//! "\0REQ" | "\0RES"   type (u32 BE)   size (u32 BE)   args, NUL-separated
12//! ```
13//!
14//! Every packet type has a fixed number of arguments; the last one runs to the end of the data
15//! and may itself contain NULs (a workload or a result is binary). [`split_args`] splits on the
16//! first `n - 1` NULs only.
17//!
18//! The admin protocol is text: a line (`status`, `workers`, `version`, …) answered by lines,
19//! a multi-line answer ending in `.`. A connection may interleave the two; a leading `\0` byte
20//! says which one comes next.
21
22/// Packet magics.
23pub const MAGIC_REQ: &[u8; 4] = b"\0REQ";
24pub const MAGIC_RES: &[u8; 4] = b"\0RES";
25pub const HEADER_LEN: usize = 12;
26
27/// The largest packet body NetGet reads: 1 MiB, judged from the declared size before anything
28/// is allocated. gearmand itself has no small limit, because a real server queues workloads for
29/// workers; here the workload goes into a model prompt, and a megabyte is already far past
30/// anything a prompt should carry.
31pub const MAX_PACKET_BYTES: usize = 1024 * 1024;
32
33/// The longest admin-protocol line, CRLF included.
34pub const MAX_ADMIN_LINE: usize = 1024;
35
36/// libgearman's own limits: `GEARMAN_FUNCTION_MAX_SIZE` and `GEARMAN_MAX_UNIQUE_SIZE`.
37pub const MAX_FUNCTION_NAME: usize = 512;
38pub const MAX_UNIQUE: usize = 64;
39
40// Packet types this server reads or writes. The numbers are the protocol's.
41pub const CAN_DO: u32 = 1;
42pub const CANT_DO: u32 = 2;
43pub const RESET_ABILITIES: u32 = 3;
44pub const PRE_SLEEP: u32 = 4;
45pub const SUBMIT_JOB: u32 = 7;
46pub const JOB_CREATED: u32 = 8;
47pub const GRAB_JOB: u32 = 9;
48pub const WORK_STATUS: u32 = 12;
49pub const WORK_COMPLETE: u32 = 13;
50pub const WORK_FAIL: u32 = 14;
51pub const GET_STATUS: u32 = 15;
52pub const ECHO_REQ: u32 = 16;
53pub const ECHO_RES: u32 = 17;
54pub const SUBMIT_JOB_BG: u32 = 18;
55pub const ERROR: u32 = 19;
56pub const STATUS_RES: u32 = 20;
57pub const SUBMIT_JOB_HIGH: u32 = 21;
58pub const SET_CLIENT_ID: u32 = 22;
59pub const CAN_DO_TIMEOUT: u32 = 23;
60pub const ALL_YOURS: u32 = 24;
61pub const WORK_EXCEPTION: u32 = 25;
62pub const OPTION_REQ: u32 = 26;
63pub const OPTION_RES: u32 = 27;
64pub const WORK_DATA: u32 = 28;
65pub const WORK_WARNING: u32 = 29;
66pub const GRAB_JOB_UNIQ: u32 = 30;
67pub const SUBMIT_JOB_HIGH_BG: u32 = 32;
68pub const SUBMIT_JOB_LOW: u32 = 33;
69pub const SUBMIT_JOB_LOW_BG: u32 = 34;
70pub const GRAB_JOB_ALL: u32 = 39;
71
72/// A parsed packet header.
73#[derive(Debug, Clone, Copy, PartialEq, Eq)]
74pub struct Header {
75    /// `\0REQ` (true) or `\0RES`.
76    pub request: bool,
77    pub packet_type: u32,
78    pub size: u32,
79}
80
81#[derive(Debug, Clone, Copy, PartialEq, Eq)]
82pub enum HeaderError {
83    BadMagic,
84    /// The declared size is past [`MAX_PACKET_BYTES`].
85    TooLarge(u32),
86}
87
88/// Parse a 12-byte header. The size is judged here, before the caller allocates for the body.
89pub fn parse_header(bytes: &[u8]) -> Result<Header, HeaderError> {
90    if bytes.len() < HEADER_LEN {
91        return Err(HeaderError::BadMagic);
92    }
93    let request = match &bytes[..4] {
94        m if m == MAGIC_REQ => true,
95        m if m == MAGIC_RES => false,
96        _ => return Err(HeaderError::BadMagic),
97    };
98    let packet_type = u32::from_be_bytes(bytes[4..8].try_into().unwrap());
99    let size = u32::from_be_bytes(bytes[8..12].try_into().unwrap());
100    if size as usize > MAX_PACKET_BYTES {
101        return Err(HeaderError::TooLarge(size));
102    }
103    Ok(Header {
104        request,
105        packet_type,
106        size,
107    })
108}
109
110/// A complete packet.
111pub fn encode(request: bool, packet_type: u32, args: &[&[u8]]) -> Vec<u8> {
112    let size: usize = args.iter().map(|a| a.len()).sum::<usize>() + args.len().saturating_sub(1);
113    let mut out = Vec::with_capacity(HEADER_LEN + size);
114    out.extend_from_slice(if request { MAGIC_REQ } else { MAGIC_RES });
115    out.extend_from_slice(&packet_type.to_be_bytes());
116    out.extend_from_slice(&(size as u32).to_be_bytes());
117    for (i, arg) in args.iter().enumerate() {
118        if i > 0 {
119            out.push(0);
120        }
121        out.extend_from_slice(arg);
122    }
123    out
124}
125
126/// A response packet (`\0RES`).
127pub fn response(packet_type: u32, args: &[&[u8]]) -> Vec<u8> {
128    encode(false, packet_type, args)
129}
130
131/// Split `data` into exactly `n` arguments on its first `n - 1` NULs; the last argument keeps
132/// any NULs of its own. `None` when there are too few separators.
133pub fn split_args(data: &[u8], n: usize) -> Option<Vec<&[u8]>> {
134    if n == 0 {
135        return data.is_empty().then(Vec::new);
136    }
137    let mut out = Vec::with_capacity(n);
138    let mut rest = data;
139    for _ in 0..n - 1 {
140        let at = rest.iter().position(|b| *b == 0)?;
141        out.push(&rest[..at]);
142        rest = &rest[at + 1..];
143    }
144    out.push(rest);
145    Some(out)
146}
147
148/// How many arguments a request of this type carries, for the types this server reads.
149pub fn request_arg_count(packet_type: u32) -> Option<usize> {
150    Some(match packet_type {
151        SUBMIT_JOB | SUBMIT_JOB_BG | SUBMIT_JOB_HIGH | SUBMIT_JOB_HIGH_BG | SUBMIT_JOB_LOW
152        | SUBMIT_JOB_LOW_BG => 3,
153        GET_STATUS | ECHO_REQ | OPTION_REQ | SET_CLIENT_ID => 1,
154        _ => return None,
155    })
156}
157
158/// Worker-side request types. NetGet runs jobs itself — the model is the worker — so these are
159/// refused rather than half-served.
160pub fn is_worker_packet(packet_type: u32) -> bool {
161    matches!(
162        packet_type,
163        CAN_DO
164            | CANT_DO
165            | RESET_ABILITIES
166            | PRE_SLEEP
167            | GRAB_JOB
168            | WORK_STATUS
169            | WORK_COMPLETE
170            | WORK_FAIL
171            | CAN_DO_TIMEOUT
172            | ALL_YOURS
173            | WORK_EXCEPTION
174            | WORK_DATA
175            | WORK_WARNING
176            | GRAB_JOB_UNIQ
177            | GRAB_JOB_ALL
178    )
179}
180
181#[derive(Debug, Clone, Copy, PartialEq, Eq)]
182pub enum Priority {
183    Normal,
184    High,
185    Low,
186}
187
188impl Priority {
189    pub fn name(self) -> &'static str {
190        match self {
191            Priority::Normal => "normal",
192            Priority::High => "high",
193            Priority::Low => "low",
194        }
195    }
196}
197
198/// One client request.
199#[derive(Debug, Clone, PartialEq, Eq)]
200pub enum Request {
201    Submit {
202        function: String,
203        unique: String,
204        workload: Vec<u8>,
205        priority: Priority,
206        background: bool,
207    },
208    GetStatus(Vec<u8>),
209    Echo(Vec<u8>),
210    Option(Vec<u8>),
211    SetClientId,
212    /// A worker packet (`CAN_DO`, `GRAB_JOB`, …).
213    Worker(u32),
214    /// A request type this server does not implement (`SUBMIT_JOB_SCHED`, `SUBMIT_REDUCE_JOB`,
215    /// an unknown number, a `\0RES` sent as a request).
216    Unsupported(u32),
217}
218
219/// Why a request's arguments were refused. The codes go to the peer in an `ERROR` packet.
220#[derive(Debug, Clone, Copy, PartialEq, Eq)]
221pub enum ArgError {
222    /// Fewer NUL separators than the type's argument count.
223    WrongArgCount,
224    /// Function name empty or past [`MAX_FUNCTION_NAME`], or unique id past [`MAX_UNIQUE`].
225    BadName,
226}
227
228/// Parse a request packet's body.
229pub fn parse_request(header: &Header, data: &[u8]) -> Result<Request, ArgError> {
230    let t = header.packet_type;
231    if !header.request {
232        return Ok(Request::Unsupported(t));
233    }
234    if is_worker_packet(t) {
235        return Ok(Request::Worker(t));
236    }
237    let Some(n) = request_arg_count(t) else {
238        return Ok(Request::Unsupported(t));
239    };
240    let args = split_args(data, n).ok_or(ArgError::WrongArgCount)?;
241    Ok(match t {
242        SUBMIT_JOB | SUBMIT_JOB_BG | SUBMIT_JOB_HIGH | SUBMIT_JOB_HIGH_BG | SUBMIT_JOB_LOW
243        | SUBMIT_JOB_LOW_BG => {
244            let (function, unique) = (args[0], args[1]);
245            if function.is_empty()
246                || function.len() > MAX_FUNCTION_NAME
247                || unique.len() > MAX_UNIQUE
248            {
249                return Err(ArgError::BadName);
250            }
251            Request::Submit {
252                function: String::from_utf8_lossy(function).into_owned(),
253                unique: String::from_utf8_lossy(unique).into_owned(),
254                workload: args[2].to_vec(),
255                priority: match t {
256                    SUBMIT_JOB_HIGH | SUBMIT_JOB_HIGH_BG => Priority::High,
257                    SUBMIT_JOB_LOW | SUBMIT_JOB_LOW_BG => Priority::Low,
258                    _ => Priority::Normal,
259                },
260                background: matches!(t, SUBMIT_JOB_BG | SUBMIT_JOB_HIGH_BG | SUBMIT_JOB_LOW_BG),
261            }
262        }
263        GET_STATUS => Request::GetStatus(args[0].to_vec()),
264        ECHO_REQ => Request::Echo(args[0].to_vec()),
265        OPTION_REQ => Request::Option(args[0].to_vec()),
266        _ => Request::SetClientId,
267    })
268}
269
270/// A number the model gave for `WORK_STATUS`, as the decimal digits the protocol carries.
271fn digits(n: u64) -> Vec<u8> {
272    n.to_string().into_bytes()
273}
274
275pub fn job_created(handle: &[u8]) -> Vec<u8> {
276    response(JOB_CREATED, &[handle])
277}
278pub fn work_status(handle: &[u8], numerator: u64, denominator: u64) -> Vec<u8> {
279    response(
280        WORK_STATUS,
281        &[handle, &digits(numerator), &digits(denominator)],
282    )
283}
284pub fn work_data(handle: &[u8], data: &[u8]) -> Vec<u8> {
285    response(WORK_DATA, &[handle, data])
286}
287pub fn work_complete(handle: &[u8], result: &[u8]) -> Vec<u8> {
288    response(WORK_COMPLETE, &[handle, result])
289}
290pub fn work_fail(handle: &[u8]) -> Vec<u8> {
291    response(WORK_FAIL, &[handle])
292}
293pub fn work_exception(handle: &[u8], text: &[u8]) -> Vec<u8> {
294    response(WORK_EXCEPTION, &[handle, text])
295}
296pub fn status_res(handle: &[u8], known: bool, running: bool, num: u64, den: u64) -> Vec<u8> {
297    response(
298        STATUS_RES,
299        &[
300            handle,
301            if known { b"1" } else { b"0" },
302            if running { b"1" } else { b"0" },
303            &digits(num),
304            &digits(den),
305        ],
306    )
307}
308
309/// An `ERROR` packet. The code is an identifier (letters, digits, `_`) and the text one line
310/// without NULs, so neither can add an argument or a packet.
311pub fn error(code: &str, text: &str) -> Result<Vec<u8>, String> {
312    if code.is_empty()
313        || code.len() > 64
314        || !code.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_')
315    {
316        return Err(format!(
317            "error code must be 1-64 letters, digits or '_', got {code:?}"
318        ));
319    }
320    if text.len() > 1024 || text.contains('\0') {
321        return Err("error text must be at most 1024 bytes and contain no NUL".to_string());
322    }
323    Ok(response(ERROR, &[code.as_bytes(), text.as_bytes()]))
324}
325
326/// Read a response packet back: `(type, args)` split by the type's argument count, or `None`
327/// for anything this module would not have written. The session loop uses it to check a
328/// rendered reply against the job it answers.
329pub fn read_response(packet: &[u8]) -> Option<(u32, Vec<Vec<u8>>)> {
330    let header = parse_header(packet).ok()?;
331    if header.request || packet.len() != HEADER_LEN + header.size as usize {
332        return None;
333    }
334    let n = match header.packet_type {
335        JOB_CREATED | WORK_FAIL | ECHO_RES | OPTION_RES => 1,
336        WORK_COMPLETE | WORK_DATA | WORK_WARNING | WORK_EXCEPTION | ERROR => 2,
337        WORK_STATUS => 3,
338        STATUS_RES => 5,
339        _ => return None,
340    };
341    let args = split_args(&packet[HEADER_LEN..], n)?;
342    Some((
343        header.packet_type,
344        args.into_iter().map(<[u8]>::to_vec).collect(),
345    ))
346}
347
348/// One admin-protocol command.
349#[derive(Debug, Clone, PartialEq, Eq)]
350pub enum AdminCommand {
351    Status,
352    Workers,
353    Version,
354    /// `maxqueue` changes server state; refused.
355    MaxQueue,
356    /// `shutdown`; refused.
357    Shutdown,
358    Unknown,
359}
360
361pub fn parse_admin(line: &str) -> AdminCommand {
362    match line.split_whitespace().next().unwrap_or("") {
363        "status" => AdminCommand::Status,
364        "workers" => AdminCommand::Workers,
365        "version" => AdminCommand::Version,
366        "maxqueue" => AdminCommand::MaxQueue,
367        "shutdown" => AdminCommand::Shutdown,
368        _ => AdminCommand::Unknown,
369    }
370}
371
372/// gearmand's admin error line: `ERR <CODE> <text with spaces as +>`.
373pub fn admin_error(code: &str, text: &str) -> String {
374    format!("ERR {code} {}\n", text.replace(' ', "+"))
375}
376
377/// The `status` answer: `FUNCTION\tTOTAL\tRUNNING\tAVAILABLE_WORKERS` per line, then `.`.
378pub fn admin_status(rows: &[(String, u64, u64, u64)]) -> String {
379    let mut out = String::new();
380    for (function, total, running, workers) in rows {
381        let name = crate::utils::sanitize::line_field(function);
382        out.push_str(&format!("{name}\t{total}\t{running}\t{workers}\n"));
383    }
384    out.push_str(".\n");
385    out
386}
netget/server/gearman/wire.rs

netget/server/gearman/
wire.rs
