netget/server/gearman/
actions.rs

1//! Gearman actions: what the model is told, and how its answers become packets.
2//!
3//! The model is the worker. A submitted job raises `gearman_job_submitted`; the model answers
4//! with progress, partial data and exactly one outcome (complete, fail, exception) or an
5//! `ERROR`. Every action is rendered by [`super::wire`], so the model supplies a result, two
6//! numbers or a text, never a packet.
7//!
8//! The job handle is NetGet's. An action may name it (`job_handle`, copied from the event), in
9//! which case the executor renders the whole packet — that is what an action injected from the
10//! dashboard needs. Without it the executor returns the answer structured and the session loop
11//! renders it with the handle of the job being answered.
12
13use super::wire;
14use crate::llm::actions::{
15    protocol_trait::{ActionResult, Protocol, Server},
16    ActionDefinition, Parameter,
17};
18use crate::protocol::log_template::LogTemplate;
19use crate::protocol::EventType;
20use crate::state::app_state::AppState;
21use anyhow::{anyhow, Context, Result};
22use serde_json::{json, Value};
23use std::sync::LazyLock;
24
25pub struct GearmanProtocol;
26
27impl GearmanProtocol {
28    pub fn new() -> Self {
29        Self
30    }
31}
32
33impl Default for GearmanProtocol {
34    fn default() -> Self {
35        Self::new()
36    }
37}
38
39/// The name of the structured result for an answer that did not name its job handle.
40pub const WORK_RESULT: &str = "gearman_work";
41
42impl Protocol for GearmanProtocol {
43    fn get_startup_parameters(&self) -> Vec<crate::llm::actions::ParameterDefinition> {
44        vec![
45            crate::llm::actions::ParameterDefinition {
46                name: "first_byte_timeout_secs".to_string(),
47                type_hint: "number".to_string(),
48                description: "Seconds a new connection may send nothing before the server \
49                              closes it. Default 300, the window a `manual` rule gives a human - \
50                              the peer may be NetGet's own TCP client parked for its operator."
51                    .to_string(),
52                required: false,
53                example: json!(300),
54                default: Some(serde_json::json!(super::FIRST_BYTE_TIMEOUT.as_secs())),
55            },
56            crate::llm::actions::ParameterDefinition {
57                name: "idle_timeout_secs".to_string(),
58                type_hint: "number".to_string(),
59                description: "Seconds the server waits for the next packet or admin line after \
60                              answering one. Default 300. A client waiting for its job's result \
61                              is not idle: the deadline wraps reads, not the model's answer."
62                    .to_string(),
63                required: false,
64                example: json!(300),
65                default: Some(serde_json::json!(super::IDLE_TIMEOUT.as_secs())),
66            },
67        ]
68    }
69    fn get_async_actions(&self, _state: &AppState) -> Vec<ActionDefinition> {
70        Vec::new()
71    }
72    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
73        vec![
74            complete_action(),
75            fail_action(),
76            status_action(),
77            data_action(),
78            exception_action(),
79            error_action(),
80            close_connection_action(),
81        ]
82    }
83    fn protocol_name(&self) -> &'static str {
84        "Gearman"
85    }
86    fn get_event_types(&self) -> Vec<EventType> {
87        vec![GEARMAN_JOB_SUBMITTED_EVENT.clone()]
88    }
89    fn stack_name(&self) -> &'static str {
90        "ETH>IP>TCP>Gearman"
91    }
92    fn keywords(&self) -> Vec<&'static str> {
93        vec!["gearman", "gearmand", "job server", "gearman worker"]
94    }
95    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
96        use crate::protocol::metadata::{
97            DevelopmentState, PrivilegeRequirement, ProtocolMetadataV2,
98        };
99
100        ProtocolMetadataV2::builder()
101            .state(DevelopmentState::Beta)
102            .well_known_port(4730)
103            // 4730 is unprivileged, and so is every port a test picks.
104            .privilege_requirement(PrivilegeRequirement::None)
105            .implementation(
106                "Hand-written Gearman job server over tokio TCP: the binary \\0REQ/\\0RES \
107                 packet protocol (client side of the server) and the text admin protocol, both \
108                 rendered by NetGet; job handles generated by NetGet",
109            )
110            .llm_control(
111                "The worker: for every submitted job, progress (WORK_STATUS), partial output \
112                 (WORK_DATA) and the outcome - WORK_COMPLETE with a result, WORK_FAIL, \
113                 WORK_EXCEPTION, or ERROR",
114            )
115            .e2e_testing(
116                "tests/server/gearman/real_client_test.rs drives the gearmand project's own \
117                 gearman(1) client (libgearman, C++; Homebrew `gearman`, Ubuntu \
118                 `gearman-tools`) through foreground jobs at normal and high priority, a \
119                 background job and --ping, and gearadmin through --status, --workers and \
120                 --server-version, asserting on what they printed and how they exited, with \
121                 Wireshark's gearman dissector reading a recorded exchange. It fails, never \
122                 skips, when the binaries are absent. tests/server/gearman/e2e_test.rs covers \
123                 the mocked-model path on a raw socket.",
124            )
125            .notes(
126                "Implements SUBMIT_JOB and its _BG, _HIGH, _HIGH_BG, _LOW and _LOW_BG variants \
127                 (JOB_CREATED, then WORK_STATUS/WORK_DATA and one of WORK_COMPLETE, WORK_FAIL, \
128                 WORK_EXCEPTION, or ERROR for foreground jobs; nothing after JOB_CREATED for \
129                 background ones), GET_STATUS against the jobs currently in \
130                 flight, ECHO_REQ, OPTION_REQ (only `exceptions`) and SET_CLIENT_ID. The admin \
131                 protocol answers status (jobs in flight per function), workers (none: the \
132                 model is not a worker connection) and version; maxqueue and shutdown are \
133                 refused. Worker packets (CAN_DO, GRAB_JOB, ...) are refused with ERROR and \
134                 the connection closed: the model is the worker. SUBMIT_JOB_SCHED/EPOCH and \
135                 reduce jobs answer ERROR. Packets are capped at 1 MiB, judged from the \
136                 declared size before allocation. On backend failure, no answer, or an answer \
137                 that does not fit the job, a foreground job gets WORK_FAIL (WORK_EXCEPTION \
138                 with a fixed text when the client enabled exceptions).",
139            )
140            .max_inbound_bytes(wire::MAX_PACKET_BYTES)
141            // WORK_FAIL for the job, which every Gearman client reports as a failed job.
142            .answers_on_failure()
143            .build()
144    }
145    fn description(&self) -> &'static str {
146        "Gearman job server - the model is the worker for every function"
147    }
148    fn example_prompt(&self) -> &'static str {
149        "Gearman job server on port 4730 - the reverse function returns its input reversed"
150    }
151    fn group_name(&self) -> &'static str {
152        "Application"
153    }
154    fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
155        use crate::llm::actions::StartupExamples;
156        StartupExamples::new(
157            json!({
158                "type": "open_server",
159                "port": 4730,
160                "base_stack": "gearman",
161                "instruction": "Gearman job server. The function 'reverse' returns its \
162                                workload reversed; the function 'wordcount' returns the number \
163                                of words. Fail any other function."
164            }),
165            json!({
166                "type": "open_server",
167                "port": 4730,
168                "base_stack": "gearman",
169                "event_handlers": [{
170                    "event_pattern": "gearman_job_submitted",
171                    "handler": {
172                        "type": "script",
173                        "language": "python",
174                        "code": "import json, sys\nevent = json.load(sys.stdin)['event']\nif event.get('function') == 'reverse':\n    actions = [{'type': 'complete_gearman_job', 'result': event.get('workload', '')[::-1]}]\nelse:\n    actions = [{'type': 'fail_gearman_job'}]\nprint(json.dumps({'actions': actions}))"
175                    }
176                }]
177            }),
178            json!({
179                "type": "open_server",
180                "port": 4730,
181                "base_stack": "gearman",
182                "event_handlers": [{
183                    "event_pattern": "gearman_job_submitted",
184                    "handler": {
185                        "type": "static",
186                        "actions": [
187                            {"type": "send_gearman_status", "numerator": 1, "denominator": 2},
188                            {"type": "complete_gearman_job", "result": "done"}
189                        ]
190                    }
191                }]
192            }),
193        )
194    }
195}
196
197impl Server for GearmanProtocol {
198    fn spawn(
199        &self,
200        ctx: crate::protocol::SpawnContext,
201    ) -> std::pin::Pin<
202        Box<dyn std::future::Future<Output = anyhow::Result<std::net::SocketAddr>> + Send>,
203    > {
204        Box::pin(async move {
205            let secs = |name: &str| -> anyhow::Result<Option<u64>> {
206                Ok(ctx
207                    .startup_params
208                    .as_ref()
209                    .map(|p| p.get_optional_u64(name))
210                    .transpose()?
211                    .flatten())
212            };
213            let first_byte_timeout_secs = secs("first_byte_timeout_secs")?;
214            let idle_timeout_secs = secs("idle_timeout_secs")?;
215
216            crate::server::gearman::GearmanServer::spawn_with_llm_actions(
217                ctx.legacy_listen_addr(),
218                ctx.llm_client,
219                ctx.state,
220                ctx.status_tx,
221                ctx.server_id,
222                first_byte_timeout_secs,
223                idle_timeout_secs,
224            )
225            .await
226        })
227    }
228
229    fn execute_action(&self, action: Value) -> Result<ActionResult> {
230        let action_type = action
231            .get("type")
232            .and_then(|v| v.as_str())
233            .context("Missing 'type' field in action")?;
234
235        let work = match action_type {
236            "complete_gearman_job" => Work::Complete(text(&action, "result")?.into_bytes()),
237            "fail_gearman_job" => Work::Fail,
238            "send_gearman_status" => {
239                let numerator = number(&action, "numerator")?;
240                let denominator = number(&action, "denominator")?;
241                if numerator > denominator {
242                    return Err(anyhow!(
243                        "numerator ({numerator}) must not exceed denominator ({denominator})"
244                    ));
245                }
246                Work::Status(numerator, denominator)
247            }
248            "send_gearman_data" => Work::Data(text(&action, "data")?.into_bytes()),
249            "send_gearman_exception" => Work::Exception(text(&action, "text")?.into_bytes()),
250            "send_gearman_error" => {
251                let code = action
252                    .get("code")
253                    .and_then(Value::as_str)
254                    .context("send_gearman_error needs 'code'")?;
255                let text = action.get("text").and_then(Value::as_str).unwrap_or("");
256                return Ok(ActionResult::Output(
257                    wire::error(code, text).map_err(|e| anyhow!("send_gearman_error: {e}"))?,
258                ));
259            }
260            "close_connection" => return Ok(ActionResult::CloseConnection),
261            _ => return Err(anyhow!("Unknown Gearman action: {}", action_type)),
262        };
263        match action.get("job_handle").and_then(Value::as_str) {
264            Some(handle) if !handle.is_empty() => {
265                Ok(ActionResult::Output(work.render(handle.as_bytes())))
266            }
267            _ => Ok(ActionResult::Custom {
268                name: WORK_RESULT.to_string(),
269                data: work.to_json(),
270            }),
271        }
272    }
273}
274
275/// One answer to a job, before it has a handle.
276#[derive(Debug, Clone, PartialEq, Eq)]
277pub enum Work {
278    Status(u64, u64),
279    Data(Vec<u8>),
280    Complete(Vec<u8>),
281    Fail,
282    Exception(Vec<u8>),
283}
284
285impl Work {
286    pub fn render(&self, handle: &[u8]) -> Vec<u8> {
287        match self {
288            Work::Status(n, d) => wire::work_status(handle, *n, *d),
289            Work::Data(data) => wire::work_data(handle, data),
290            Work::Complete(result) => wire::work_complete(handle, result),
291            Work::Fail => wire::work_fail(handle),
292            Work::Exception(text) => wire::work_exception(handle, text),
293        }
294    }
295
296    fn to_json(&self) -> Value {
297        let text = |b: &[u8]| String::from_utf8_lossy(b).into_owned();
298        match self {
299            Work::Status(n, d) => json!({"kind": "status", "numerator": n, "denominator": d}),
300            Work::Data(data) => json!({"kind": "data", "data": text(data)}),
301            Work::Complete(result) => json!({"kind": "complete", "result": text(result)}),
302            Work::Fail => json!({"kind": "fail"}),
303            Work::Exception(t) => json!({"kind": "exception", "text": text(t)}),
304        }
305    }
306
307    /// Read back what [`Work::to_json`] produced.
308    pub fn from_json(value: &Value) -> Option<Work> {
309        let s = |k: &str| {
310            value
311                .get(k)
312                .and_then(Value::as_str)
313                .map(|s| s.as_bytes().to_vec())
314        };
315        Some(match value.get("kind")?.as_str()? {
316            "status" => Work::Status(
317                value.get("numerator")?.as_u64()?,
318                value.get("denominator")?.as_u64()?,
319            ),
320            "data" => Work::Data(s("data")?),
321            "complete" => Work::Complete(s("result")?),
322            "fail" => Work::Fail,
323            "exception" => Work::Exception(s("text")?),
324            _ => return None,
325        })
326    }
327}
328
329fn text(action: &Value, name: &str) -> Result<String> {
330    match action.get(name) {
331        Some(Value::String(s)) => Ok(s.clone()),
332        Some(Value::Number(n)) => Ok(n.to_string()),
333        None | Some(Value::Null) => Ok(String::new()),
334        Some(_) => Err(anyhow!("'{name}' must be text")),
335    }
336}
337
338fn number(action: &Value, name: &str) -> Result<u64> {
339    action
340        .get(name)
341        .and_then(|v| {
342            v.as_u64()
343                .or_else(|| v.as_str().and_then(|s| s.trim().parse().ok()))
344        })
345        .with_context(|| format!("'{name}' must be a non-negative integer"))
346}
347
348fn handle_param() -> Parameter {
349    Parameter {
350        name: "job_handle".to_string(),
351        type_hint: "string".to_string(),
352        description: "The job's handle from the event. Optional when answering the event; \
353                      needed when sending to a connection from outside an event."
354            .to_string(),
355        required: false,
356    }
357}
358
359fn complete_action() -> ActionDefinition {
360    ActionDefinition {
361        name: "complete_gearman_job".to_string(),
362        description: "Finish the job successfully with its result (WORK_COMPLETE). The client \
363                      prints the result."
364            .to_string(),
365        parameters: vec![
366            Parameter {
367                name: "result".to_string(),
368                type_hint: "string".to_string(),
369                description: "The job's output as text".to_string(),
370                required: true,
371            },
372            handle_param(),
373        ],
374        example: json!({"type": "complete_gearman_job", "result": "dlrow olleh"}),
375        log_template: Some(
376            LogTemplate::new()
377                .with_info("-> Gearman WORK_COMPLETE")
378                .with_debug("Gearman complete_gearman_job"),
379        ),
380    }
381}
382
383fn fail_action() -> ActionDefinition {
384    ActionDefinition {
385        name: "fail_gearman_job".to_string(),
386        description: "Finish the job as failed (WORK_FAIL), e.g. for a function you do not \
387                      provide or a workload you cannot process."
388            .to_string(),
389        parameters: vec![handle_param()],
390        example: json!({"type": "fail_gearman_job"}),
391        log_template: Some(
392            LogTemplate::new()
393                .with_info("-> Gearman WORK_FAIL")
394                .with_debug("Gearman fail_gearman_job"),
395        ),
396    }
397}
398
399fn status_action() -> ActionDefinition {
400    ActionDefinition {
401        name: "send_gearman_status".to_string(),
402        description: "Report progress (WORK_STATUS numerator/denominator) before finishing; \
403                      follow it with complete_gearman_job or fail_gearman_job in the same \
404                      answer."
405            .to_string(),
406        parameters: vec![
407            Parameter {
408                name: "numerator".to_string(),
409                type_hint: "number".to_string(),
410                description: "How much of the job is done, in any unit you choose".to_string(),
411                required: true,
412            },
413            Parameter {
414                name: "denominator".to_string(),
415                type_hint: "number".to_string(),
416                description: "How much work the whole job is, in the same unit as numerator"
417                    .to_string(),
418                required: true,
419            },
420            handle_param(),
421        ],
422        example: json!({"type": "send_gearman_status", "numerator": 1, "denominator": 2}),
423        log_template: Some(
424            LogTemplate::new()
425                .with_info("-> Gearman WORK_STATUS {numerator}/{denominator}")
426                .with_debug("Gearman send_gearman_status"),
427        ),
428    }
429}
430
431fn data_action() -> ActionDefinition {
432    ActionDefinition {
433        name: "send_gearman_data".to_string(),
434        description: "Send partial output (WORK_DATA) before finishing; the client prints it \
435                      ahead of the final result."
436            .to_string(),
437        parameters: vec![
438            Parameter {
439                name: "data".to_string(),
440                type_hint: "string".to_string(),
441                description: "A piece of the job's output as text".to_string(),
442                required: true,
443            },
444            handle_param(),
445        ],
446        example: json!({"type": "send_gearman_data", "data": "first half, "}),
447        log_template: Some(
448            LogTemplate::new()
449                .with_info("-> Gearman WORK_DATA")
450                .with_debug("Gearman send_gearman_data"),
451        ),
452    }
453}
454
455fn exception_action() -> ActionDefinition {
456    ActionDefinition {
457        name: "send_gearman_exception".to_string(),
458        description: "Finish the job with an exception message (WORK_EXCEPTION). Clients that \
459                      did not ask for exceptions receive WORK_FAIL instead."
460            .to_string(),
461        parameters: vec![
462            Parameter {
463                name: "text".to_string(),
464                type_hint: "string".to_string(),
465                description: "What went wrong".to_string(),
466                required: true,
467            },
468            handle_param(),
469        ],
470        example: json!({"type": "send_gearman_exception", "text": "image is corrupt"}),
471        log_template: Some(
472            LogTemplate::new()
473                .with_info("-> Gearman WORK_EXCEPTION")
474                .with_debug("Gearman send_gearman_exception"),
475        ),
476    }
477}
478
479fn error_action() -> ActionDefinition {
480    ActionDefinition {
481        name: "send_gearman_error".to_string(),
482        description: "Refuse the submission itself with a server ERROR packet instead of \
483                      running it."
484            .to_string(),
485        parameters: vec![
486            Parameter {
487                name: "code".to_string(),
488                type_hint: "string".to_string(),
489                description: "An error identifier such as queue_full or unknown_function"
490                    .to_string(),
491                required: true,
492            },
493            Parameter {
494                name: "text".to_string(),
495                type_hint: "string".to_string(),
496                description: "One line explaining the error".to_string(),
497                required: true,
498            },
499        ],
500        example: json!({"type": "send_gearman_error", "code": "queue_full", "text": "try again later"}),
501        log_template: Some(
502            LogTemplate::new()
503                .with_info("-> Gearman ERROR {code}")
504                .with_debug("Gearman send_gearman_error: code={code}"),
505        ),
506    }
507}
508
509fn close_connection_action() -> ActionDefinition {
510    ActionDefinition {
511        name: "close_connection".to_string(),
512        description: "Close the Gearman connection after any reply".to_string(),
513        parameters: vec![],
514        example: json!({"type": "close_connection"}),
515        log_template: Some(
516            LogTemplate::new()
517                .with_info("Gearman connection closed")
518                .with_debug("Gearman close_connection"),
519        ),
520    }
521}
522
523/// `SUBMIT_JOB` and its variants.
524pub static GEARMAN_JOB_SUBMITTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
525    EventType::new(
526        "gearman_job_submitted",
527        "A client submitted a job. You are the worker: run the function on the workload and \
528         answer with complete_gearman_job (optionally preceded by send_gearman_status or \
529         send_gearman_data), fail_gearman_job, send_gearman_exception, or send_gearman_error. \
530         For a background job the client is not waiting; your answer is only logged.",
531        json!({"type": "complete_gearman_job", "result": "dlrow olleh"}),
532    )
533    .with_parameters(vec![
534        Parameter {
535            name: "function".to_string(),
536            type_hint: "string".to_string(),
537            description: "The function the client asked for".to_string(),
538            required: true,
539        },
540        Parameter {
541            name: "workload".to_string(),
542            type_hint: "string".to_string(),
543            description: "The job's input as text (invalid UTF-8 replaced)".to_string(),
544            required: true,
545        },
546        Parameter {
547            name: "workload_bytes".to_string(),
548            type_hint: "number".to_string(),
549            description: "The input's length in bytes".to_string(),
550            required: true,
551        },
552        Parameter {
553            name: "unique_id".to_string(),
554            type_hint: "string".to_string(),
555            description: "The client's unique id for the job (may be empty)".to_string(),
556            required: true,
557        },
558        Parameter {
559            name: "priority".to_string(),
560            type_hint: "string".to_string(),
561            description: "Queue priority the client asked for".to_string(),
562            required: true,
563        }
564        .with_choices(["normal", "high", "low"]),
565        Parameter {
566            name: "background".to_string(),
567            type_hint: "boolean".to_string(),
568            description: "true when the client is not waiting for the result".to_string(),
569            required: true,
570        },
571        Parameter {
572            name: "job_handle".to_string(),
573            type_hint: "string".to_string(),
574            description: "The handle NetGet gave the job".to_string(),
575            required: true,
576        },
577    ])
578    .with_log_template(
579        LogTemplate::new()
580            .with_info("Gearman job {function} ({workload_bytes} bytes)")
581            .with_debug("Gearman gearman_job_submitted: function={function} handle={job_handle}"),
582    )
583    .with_actions(vec![
584        complete_action(),
585        fail_action(),
586        status_action(),
587        data_action(),
588        exception_action(),
589        error_action(),
590        close_connection_action(),
591    ])
592    .with_alternative_example(json!({"type": "fail_gearman_job"}))
593});
netget/server/gearman/actions.rs

netget/server/gearman/
actions.rs
