use super::wire;
use crate::llm::actions::{
protocol_trait::{ActionResult, Protocol, Server},
ActionDefinition, Parameter,
};
use crate::protocol::log_template::LogTemplate;
use crate::protocol::EventType;
use crate::state::app_state::AppState;
use anyhow::{anyhow, Context, Result};
use serde_json::{json, Value};
use std::sync::LazyLock;
pub struct GearmanProtocol;
impl GearmanProtocol {
pub fn new() -> Self {
Self
}
}
impl Default for GearmanProtocol {
fn default() -> Self {
Self::new()
}
}
pub const WORK_RESULT: &str = "gearman_work";
impl Protocol for GearmanProtocol {
fn get_startup_parameters(&self) -> Vec<crate::llm::actions::ParameterDefinition> {
vec![
crate::llm::actions::ParameterDefinition {
name: "first_byte_timeout_secs".to_string(),
type_hint: "number".to_string(),
description: "Seconds a new connection may send nothing before the server \
closes it. Default 300, the window a `manual` rule gives a human - \
the peer may be NetGet's own TCP client parked for its operator."
.to_string(),
required: false,
example: json!(300),
default: Some(serde_json::json!(super::FIRST_BYTE_TIMEOUT.as_secs())),
},
crate::llm::actions::ParameterDefinition {
name: "idle_timeout_secs".to_string(),
type_hint: "number".to_string(),
description: "Seconds the server waits for the next packet or admin line after \
answering one. Default 300. A client waiting for its job's result \
is not idle: the deadline wraps reads, not the model's answer."
.to_string(),
required: false,
example: json!(300),
default: Some(serde_json::json!(super::IDLE_TIMEOUT.as_secs())),
},
]
}
fn get_async_actions(&self, _state: &AppState) -> Vec<ActionDefinition> {
Vec::new()
}
fn get_sync_actions(&self) -> Vec<ActionDefinition> {
vec![
complete_action(),
fail_action(),
status_action(),
data_action(),
exception_action(),
error_action(),
close_connection_action(),
]
}
fn protocol_name(&self) -> &'static str {
"Gearman"
}
fn get_event_types(&self) -> Vec<EventType> {
vec![GEARMAN_JOB_SUBMITTED_EVENT.clone()]
}
fn stack_name(&self) -> &'static str {
"ETH>IP>TCP>Gearman"
}
fn keywords(&self) -> Vec<&'static str> {
vec!["gearman", "gearmand", "job server", "gearman worker"]
}
fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
use crate::protocol::metadata::{
DevelopmentState, PrivilegeRequirement, ProtocolMetadataV2,
};
ProtocolMetadataV2::builder()
.state(DevelopmentState::Beta)
.well_known_port(4730)
.privilege_requirement(PrivilegeRequirement::None)
.implementation(
"Hand-written Gearman job server over tokio TCP: the binary \\0REQ/\\0RES \
packet protocol (client side of the server) and the text admin protocol, both \
rendered by NetGet; job handles generated by NetGet",
)
.llm_control(
"The worker: for every submitted job, progress (WORK_STATUS), partial output \
(WORK_DATA) and the outcome - WORK_COMPLETE with a result, WORK_FAIL, \
WORK_EXCEPTION, or ERROR",
)
.e2e_testing(
"tests/server/gearman/real_client_test.rs drives the gearmand project's own \
gearman(1) client (libgearman, C++; Homebrew `gearman`, Ubuntu \
`gearman-tools`) through foreground jobs at normal and high priority, a \
background job and --ping, and gearadmin through --status, --workers and \
--server-version, asserting on what they printed and how they exited, with \
Wireshark's gearman dissector reading a recorded exchange. It fails, never \
skips, when the binaries are absent. tests/server/gearman/e2e_test.rs covers \
the mocked-model path on a raw socket.",
)
.notes(
"Implements SUBMIT_JOB and its _BG, _HIGH, _HIGH_BG, _LOW and _LOW_BG variants \
(JOB_CREATED, then WORK_STATUS/WORK_DATA and one of WORK_COMPLETE, WORK_FAIL, \
WORK_EXCEPTION, or ERROR for foreground jobs; nothing after JOB_CREATED for \
background ones), GET_STATUS against the jobs currently in \
flight, ECHO_REQ, OPTION_REQ (only `exceptions`) and SET_CLIENT_ID. The admin \
protocol answers status (jobs in flight per function), workers (none: the \
model is not a worker connection) and version; maxqueue and shutdown are \
refused. Worker packets (CAN_DO, GRAB_JOB, ...) are refused with ERROR and \
the connection closed: the model is the worker. SUBMIT_JOB_SCHED/EPOCH and \
reduce jobs answer ERROR. Packets are capped at 1 MiB, judged from the \
declared size before allocation. On backend failure, no answer, or an answer \
that does not fit the job, a foreground job gets WORK_FAIL (WORK_EXCEPTION \
with a fixed text when the client enabled exceptions).",
)
.max_inbound_bytes(wire::MAX_PACKET_BYTES)
.answers_on_failure()
.build()
}
fn description(&self) -> &'static str {
"Gearman job server - the model is the worker for every function"
}
fn example_prompt(&self) -> &'static str {
"Gearman job server on port 4730 - the reverse function returns its input reversed"
}
fn group_name(&self) -> &'static str {
"Application"
}
fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
use crate::llm::actions::StartupExamples;
StartupExamples::new(
json!({
"type": "open_server",
"port": 4730,
"base_stack": "gearman",
"instruction": "Gearman job server. The function 'reverse' returns its \
workload reversed; the function 'wordcount' returns the number \
of words. Fail any other function."
}),
json!({
"type": "open_server",
"port": 4730,
"base_stack": "gearman",
"event_handlers": [{
"event_pattern": "gearman_job_submitted",
"handler": {
"type": "script",
"language": "python",
"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}))"
}
}]
}),
json!({
"type": "open_server",
"port": 4730,
"base_stack": "gearman",
"event_handlers": [{
"event_pattern": "gearman_job_submitted",
"handler": {
"type": "static",
"actions": [
{"type": "send_gearman_status", "numerator": 1, "denominator": 2},
{"type": "complete_gearman_job", "result": "done"}
]
}
}]
}),
)
}
}
impl Server for GearmanProtocol {
fn spawn(
&self,
ctx: crate::protocol::SpawnContext,
) -> std::pin::Pin<
Box<dyn std::future::Future<Output = anyhow::Result<std::net::SocketAddr>> + Send>,
> {
Box::pin(async move {
let secs = |name: &str| -> anyhow::Result<Option<u64>> {
Ok(ctx
.startup_params
.as_ref()
.map(|p| p.get_optional_u64(name))
.transpose()?
.flatten())
};
let first_byte_timeout_secs = secs("first_byte_timeout_secs")?;
let idle_timeout_secs = secs("idle_timeout_secs")?;
crate::server::gearman::GearmanServer::spawn_with_llm_actions(
ctx.legacy_listen_addr(),
ctx.llm_client,
ctx.state,
ctx.status_tx,
ctx.server_id,
first_byte_timeout_secs,
idle_timeout_secs,
)
.await
})
}
fn execute_action(&self, action: Value) -> Result<ActionResult> {
let action_type = action
.get("type")
.and_then(|v| v.as_str())
.context("Missing 'type' field in action")?;
let work = match action_type {
"complete_gearman_job" => Work::Complete(text(&action, "result")?.into_bytes()),
"fail_gearman_job" => Work::Fail,
"send_gearman_status" => {
let numerator = number(&action, "numerator")?;
let denominator = number(&action, "denominator")?;
if numerator > denominator {
return Err(anyhow!(
"numerator ({numerator}) must not exceed denominator ({denominator})"
));
}
Work::Status(numerator, denominator)
}
"send_gearman_data" => Work::Data(text(&action, "data")?.into_bytes()),
"send_gearman_exception" => Work::Exception(text(&action, "text")?.into_bytes()),
"send_gearman_error" => {
let code = action
.get("code")
.and_then(Value::as_str)
.context("send_gearman_error needs 'code'")?;
let text = action.get("text").and_then(Value::as_str).unwrap_or("");
return Ok(ActionResult::Output(
wire::error(code, text).map_err(|e| anyhow!("send_gearman_error: {e}"))?,
));
}
"close_connection" => return Ok(ActionResult::CloseConnection),
_ => return Err(anyhow!("Unknown Gearman action: {}", action_type)),
};
match action.get("job_handle").and_then(Value::as_str) {
Some(handle) if !handle.is_empty() => {
Ok(ActionResult::Output(work.render(handle.as_bytes())))
}
_ => Ok(ActionResult::Custom {
name: WORK_RESULT.to_string(),
data: work.to_json(),
}),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Work {
Status(u64, u64),
Data(Vec<u8>),
Complete(Vec<u8>),
Fail,
Exception(Vec<u8>),
}
impl Work {
pub fn render(&self, handle: &[u8]) -> Vec<u8> {
match self {
Work::Status(n, d) => wire::work_status(handle, *n, *d),
Work::Data(data) => wire::work_data(handle, data),
Work::Complete(result) => wire::work_complete(handle, result),
Work::Fail => wire::work_fail(handle),
Work::Exception(text) => wire::work_exception(handle, text),
}
}
fn to_json(&self) -> Value {
let text = |b: &[u8]| String::from_utf8_lossy(b).into_owned();
match self {
Work::Status(n, d) => json!({"kind": "status", "numerator": n, "denominator": d}),
Work::Data(data) => json!({"kind": "data", "data": text(data)}),
Work::Complete(result) => json!({"kind": "complete", "result": text(result)}),
Work::Fail => json!({"kind": "fail"}),
Work::Exception(t) => json!({"kind": "exception", "text": text(t)}),
}
}
pub fn from_json(value: &Value) -> Option<Work> {
let s = |k: &str| {
value
.get(k)
.and_then(Value::as_str)
.map(|s| s.as_bytes().to_vec())
};
Some(match value.get("kind")?.as_str()? {
"status" => Work::Status(
value.get("numerator")?.as_u64()?,
value.get("denominator")?.as_u64()?,
),
"data" => Work::Data(s("data")?),
"complete" => Work::Complete(s("result")?),
"fail" => Work::Fail,
"exception" => Work::Exception(s("text")?),
_ => return None,
})
}
}
fn text(action: &Value, name: &str) -> Result<String> {
match action.get(name) {
Some(Value::String(s)) => Ok(s.clone()),
Some(Value::Number(n)) => Ok(n.to_string()),
None | Some(Value::Null) => Ok(String::new()),
Some(_) => Err(anyhow!("'{name}' must be text")),
}
}
fn number(action: &Value, name: &str) -> Result<u64> {
action
.get(name)
.and_then(|v| {
v.as_u64()
.or_else(|| v.as_str().and_then(|s| s.trim().parse().ok()))
})
.with_context(|| format!("'{name}' must be a non-negative integer"))
}
fn handle_param() -> Parameter {
Parameter {
name: "job_handle".to_string(),
type_hint: "string".to_string(),
description: "The job's handle from the event. Optional when answering the event; \
needed when sending to a connection from outside an event."
.to_string(),
required: false,
}
}
fn complete_action() -> ActionDefinition {
ActionDefinition {
name: "complete_gearman_job".to_string(),
description: "Finish the job successfully with its result (WORK_COMPLETE). The client \
prints the result."
.to_string(),
parameters: vec![
Parameter {
name: "result".to_string(),
type_hint: "string".to_string(),
description: "The job's output as text".to_string(),
required: true,
},
handle_param(),
],
example: json!({"type": "complete_gearman_job", "result": "dlrow olleh"}),
log_template: Some(
LogTemplate::new()
.with_info("-> Gearman WORK_COMPLETE")
.with_debug("Gearman complete_gearman_job"),
),
}
}
fn fail_action() -> ActionDefinition {
ActionDefinition {
name: "fail_gearman_job".to_string(),
description: "Finish the job as failed (WORK_FAIL), e.g. for a function you do not \
provide or a workload you cannot process."
.to_string(),
parameters: vec![handle_param()],
example: json!({"type": "fail_gearman_job"}),
log_template: Some(
LogTemplate::new()
.with_info("-> Gearman WORK_FAIL")
.with_debug("Gearman fail_gearman_job"),
),
}
}
fn status_action() -> ActionDefinition {
ActionDefinition {
name: "send_gearman_status".to_string(),
description: "Report progress (WORK_STATUS numerator/denominator) before finishing; \
follow it with complete_gearman_job or fail_gearman_job in the same \
answer."
.to_string(),
parameters: vec![
Parameter {
name: "numerator".to_string(),
type_hint: "number".to_string(),
description: "How much of the job is done, in any unit you choose".to_string(),
required: true,
},
Parameter {
name: "denominator".to_string(),
type_hint: "number".to_string(),
description: "How much work the whole job is, in the same unit as numerator"
.to_string(),
required: true,
},
handle_param(),
],
example: json!({"type": "send_gearman_status", "numerator": 1, "denominator": 2}),
log_template: Some(
LogTemplate::new()
.with_info("-> Gearman WORK_STATUS {numerator}/{denominator}")
.with_debug("Gearman send_gearman_status"),
),
}
}
fn data_action() -> ActionDefinition {
ActionDefinition {
name: "send_gearman_data".to_string(),
description: "Send partial output (WORK_DATA) before finishing; the client prints it \
ahead of the final result."
.to_string(),
parameters: vec![
Parameter {
name: "data".to_string(),
type_hint: "string".to_string(),
description: "A piece of the job's output as text".to_string(),
required: true,
},
handle_param(),
],
example: json!({"type": "send_gearman_data", "data": "first half, "}),
log_template: Some(
LogTemplate::new()
.with_info("-> Gearman WORK_DATA")
.with_debug("Gearman send_gearman_data"),
),
}
}
fn exception_action() -> ActionDefinition {
ActionDefinition {
name: "send_gearman_exception".to_string(),
description: "Finish the job with an exception message (WORK_EXCEPTION). Clients that \
did not ask for exceptions receive WORK_FAIL instead."
.to_string(),
parameters: vec![
Parameter {
name: "text".to_string(),
type_hint: "string".to_string(),
description: "What went wrong".to_string(),
required: true,
},
handle_param(),
],
example: json!({"type": "send_gearman_exception", "text": "image is corrupt"}),
log_template: Some(
LogTemplate::new()
.with_info("-> Gearman WORK_EXCEPTION")
.with_debug("Gearman send_gearman_exception"),
),
}
}
fn error_action() -> ActionDefinition {
ActionDefinition {
name: "send_gearman_error".to_string(),
description: "Refuse the submission itself with a server ERROR packet instead of \
running it."
.to_string(),
parameters: vec![
Parameter {
name: "code".to_string(),
type_hint: "string".to_string(),
description: "An error identifier such as queue_full or unknown_function"
.to_string(),
required: true,
},
Parameter {
name: "text".to_string(),
type_hint: "string".to_string(),
description: "One line explaining the error".to_string(),
required: true,
},
],
example: json!({"type": "send_gearman_error", "code": "queue_full", "text": "try again later"}),
log_template: Some(
LogTemplate::new()
.with_info("-> Gearman ERROR {code}")
.with_debug("Gearman send_gearman_error: code={code}"),
),
}
}
fn close_connection_action() -> ActionDefinition {
ActionDefinition {
name: "close_connection".to_string(),
description: "Close the Gearman connection after any reply".to_string(),
parameters: vec![],
example: json!({"type": "close_connection"}),
log_template: Some(
LogTemplate::new()
.with_info("Gearman connection closed")
.with_debug("Gearman close_connection"),
),
}
}
pub static GEARMAN_JOB_SUBMITTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
EventType::new(
"gearman_job_submitted",
"A client submitted a job. You are the worker: run the function on the workload and \
answer with complete_gearman_job (optionally preceded by send_gearman_status or \
send_gearman_data), fail_gearman_job, send_gearman_exception, or send_gearman_error. \
For a background job the client is not waiting; your answer is only logged.",
json!({"type": "complete_gearman_job", "result": "dlrow olleh"}),
)
.with_parameters(vec![
Parameter {
name: "function".to_string(),
type_hint: "string".to_string(),
description: "The function the client asked for".to_string(),
required: true,
},
Parameter {
name: "workload".to_string(),
type_hint: "string".to_string(),
description: "The job's input as text (invalid UTF-8 replaced)".to_string(),
required: true,
},
Parameter {
name: "workload_bytes".to_string(),
type_hint: "number".to_string(),
description: "The input's length in bytes".to_string(),
required: true,
},
Parameter {
name: "unique_id".to_string(),
type_hint: "string".to_string(),
description: "The client's unique id for the job (may be empty)".to_string(),
required: true,
},
Parameter {
name: "priority".to_string(),
type_hint: "string".to_string(),
description: "Queue priority the client asked for".to_string(),
required: true,
}
.with_choices(["normal", "high", "low"]),
Parameter {
name: "background".to_string(),
type_hint: "boolean".to_string(),
description: "true when the client is not waiting for the result".to_string(),
required: true,
},
Parameter {
name: "job_handle".to_string(),
type_hint: "string".to_string(),
description: "The handle NetGet gave the job".to_string(),
required: true,
},
])
.with_log_template(
LogTemplate::new()
.with_info("Gearman job {function} ({workload_bytes} bytes)")
.with_debug("Gearman gearman_job_submitted: function={function} handle={job_handle}"),
)
.with_actions(vec![
complete_action(),
fail_action(),
status_action(),
data_action(),
exception_action(),
error_action(),
close_connection_action(),
])
.with_alternative_example(json!({"type": "fail_gearman_job"}))
});