use std::fmt::Write as _;
use super::plan::ConnectionPlan;
use super::scaffold::AwlWorkerScaffold;
use super::text::string_literal;
pub(super) fn emit(request: &AwlWorkerScaffold<'_>, plan: &ConnectionPlan) -> String {
let mut out = String::new();
emit_header(&mut out, request, plan);
emit_imports(&mut out);
emit_connection_table(&mut out, plan);
emit_handler_table(&mut out, plan);
emit_main(&mut out);
emit_build(&mut out);
emit_settings(&mut out);
emit_env_helpers(&mut out);
emit_tests(&mut out);
out
}
fn emit_header(out: &mut String, request: &AwlWorkerScaffold<'_>, plan: &ConnectionPlan) {
let _ = write!(
out,
"//! Generated by `aion awl scaffold` from `{document}` — do not edit; regenerate\n\
//! from the document. The activity bodies live in `handlers.rs`, which is yours.\n\
//!\n\
//! Composition root for the `{queue}` queue's worker: {actions} action(s) across\n\
//! {connections} connection(s).\n\
//!\n\
//! ONE CONNECTION PER NODE. The server routes an activity by (namespace,\n\
//! `task_queue`, node) and by NOTHING else — never by activity type. So a process\n\
//! serving actions pinned to several nodes must dial once per node, each\n\
//! connection registering only the actions a dispatch could reach it with. Two\n\
//! connections on one node would let the server land an activity on the one that\n\
//! holds no handler for it.\n\
//!\n\
//! AN UNPINNED ACTION IS OWED BY EVERY CONNECTION. Its dispatch reaches any worker\n\
//! in the pool, so every connection below registers it — the same reachability rule\n\
//! the server's admission gate applies.\n\
//!\n\
//! A `run`-BODIED ACTION IS SERVER-EXECUTED and appears nowhere here: no worker\n\
//! serves it, and advertising one would promise what this binary does not do.\n\
\n",
document = request.document_name,
queue = plan.task_queue,
actions = plan.servable.len(),
connections = plan.connections.len(),
);
}
fn emit_imports(out: &mut String) {
out.push_str(
"use std::collections::BTreeSet;\n\
use std::time::Duration;\n\
\n\
use aion_worker::{ActivityContext, HandlerFuture, Worker, WorkerConfig};\n\
use serde_json::Value;\n\
\n\
mod declaration;\n\
mod handlers;\n\
\n\
use declaration::{Declaration, TASK_QUEUE};\n\
\n",
);
}
fn emit_connection_table(out: &mut String, plan: &ConnectionPlan) {
out.push_str(
"/// One CONNECTION: the node it registers on, and every action it serves there.\n\
struct Connection {\n\
\x20 /// The locality advertised at registration — the third routing dimension.\n\
\x20 /// `None` registers no locality, which only unpinned dispatches reach.\n\
\x20 node: Option<&'static str>,\n\
\x20 /// The actions served on this connection, in the document's order.\n\
\x20 actions: &'static [&'static str],\n\
}\n\
\n\
/// THE CONNECTION TABLE, generated from the compiled contract: one entry per node,\n\
/// carrying exactly the actions the server's admission gate demands of it.\n\
static CONNECTIONS: &[Connection] = &[\n",
);
for connection in &plan.connections {
let node = connection.node.as_ref().map_or_else(
|| "None".to_owned(),
|node| format!("Some({})", string_literal(node)),
);
let actions = connection
.actions
.iter()
.map(|action| string_literal(action))
.collect::<Vec<_>>()
.join(", ");
let _ = writeln!(
out,
" Connection {{\n node: {node},\n actions: &[{actions}],\n }},"
);
}
out.push_str("];\n\n");
}
fn emit_handler_table(out: &mut String, plan: &ConnectionPlan) {
out.push_str(
"/// An activity handler: untyped JSON in and out, because the types live in the AWL\n\
/// document and not in Rust. The advertisement is exact all the same — it comes\n\
/// from the compiled contract, not from a second Rust rendering of the shapes.\n\
type HandlerFn =\n\
\x20 for<'context> fn(Value, &'context ActivityContext) -> HandlerFuture<'context, Value>;\n\
\n\
/// The handler this binary serves `action` with.\n\
///\n\
/// `None` cannot arise from the generated table — the two are emitted together —\n\
/// but it is reported rather than assumed away, because the alternative is a\n\
/// connection that registers an activity nothing can run.\n\
fn handler(action: &str) -> Option<HandlerFn> {\n\
\x20 match action {\n",
);
let mut names: Vec<&str> = plan
.servable
.iter()
.map(|action| action.name.as_str())
.collect();
names.sort_unstable();
for name in names {
let _ = writeln!(
out,
" {} => Some(handlers::{}),",
string_literal(name),
handler_ident(name),
);
}
out.push_str(" _ => None,\n }\n}\n\n");
}
fn emit_main(out: &mut String) {
out.push_str(
"#[tokio::main]\n\
async fn main() -> anyhow::Result<()> {\n\
\x20 tracing_subscriber::fmt()\n\
\x20 .with_env_filter(\n\
\x20 tracing_subscriber::EnvFilter::try_from_default_env()\n\
\x20 .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new(\"info\")),\n\
\x20 )\n\
\x20 .init();\n\
\n\
\x20 // The declaration is compiled BEFORE anything is served: it is the source of\n\
\x20 // every advertised schema, and both startup guards read it.\n\
\x20 let declaration = Declaration::compile()?;\n\
\x20 let settings = Settings::from_env()?;\n\
\n\
\x20 // GUARD ONE, inside `build`: an activity this binary serves that the document\n\
\x20 // declares no reachable action for is a startup failure, never a handler\n\
\x20 // registered with no advertisement. Every connection is built before any is\n\
\x20 // dialled, so a mis-declared handler fails here and not on the wire.\n\
\x20 let mut workers = Vec::with_capacity(CONNECTIONS.len());\n\
\x20 for connection in CONNECTIONS {\n\
\x20 workers.push(build(connection, &declaration, &settings)?);\n\
\x20 }\n\
\x20 // GUARD TWO: every action the document declares WITHOUT a body requires an\n\
\x20 // out-of-band worker, and this binary is it. One that no connection serves is\n\
\x20 // a startup failure naming the action and its node — never a queue with a\n\
\x20 // permanently parked dispatch. It fires when the document grew an action and\n\
\x20 // the scaffold was not regenerated.\n\
\x20 declaration.require_every_bodyless_action_served(&served())?;\n\
\n\
\x20 tracing::info!(\n\
\x20 task_queue = TASK_QUEUE,\n\
\x20 connections = CONNECTIONS.len(),\n\
\x20 declared_actions = declaration.contract().actions.len(),\n\
\x20 endpoint = %settings.endpoint,\n\
\x20 \"worker starting\"\n\
\x20 );\n\
\n\
\x20 let mut serving = tokio::task::JoinSet::new();\n\
\x20 for worker in workers {\n\
\x20 serving.spawn(worker.run());\n\
\x20 }\n\
\x20 // A connection that ends, ends the process: the queue is only served while\n\
\x20 // every node's connection is up, and a silently missing one would park that\n\
\x20 // node's dispatches forever.\n\
\x20 while let Some(joined) = serving.join_next().await {\n\
\x20 joined??;\n\
\x20 }\n\
\x20 Ok(())\n\
}\n\
\n\
/// Every action the connection table serves, deduplicated.\n\
fn served() -> BTreeSet<String> {\n\
\x20 CONNECTIONS\n\
\x20 .iter()\n\
\x20 .flat_map(|connection| connection.actions.iter().map(|action| (*action).to_owned()))\n\
\x20 .collect()\n\
}\n\
\n",
);
}
fn emit_build(out: &mut String) {
out.push_str(
"/// Builds ONE connection's worker: its own config (carrying the node that separates\n\
/// this process's same-queue connections), and every action it serves registered\n\
/// WITH the wire descriptor the document declares for it.\n\
///\n\
/// The descriptor is not optional. A handler registered without one advertises\n\
/// nothing, and the server's admission gate reports the action as `<missing>` and\n\
/// refuses the whole connection with `WORKER_CONTRACT_MISMATCH`.\n\
fn build(\n\
\x20 connection: &Connection,\n\
\x20 declaration: &Declaration,\n\
\x20 settings: &Settings,\n\
) -> anyhow::Result<Worker> {\n\
\x20 let mut builder = Worker::builder(settings.config(connection.node)?);\n\
\x20 for action in connection.actions {\n\
\x20 let Some(handler) = handler(action) else {\n\
\x20 anyhow::bail!(\n\
\x20 \"the connection table serves activity `{action}`, which this binary \\\n\
\x20 registers no handler for\"\n\
\x20 );\n\
\x20 };\n\
\x20 builder = builder.register_activity_with_descriptor::<Value, Value, _>(\n\
\x20 *action,\n\
\x20 declaration.descriptor(action, connection.node)?,\n\
\x20 handler,\n\
\x20 )?;\n\
\x20 }\n\
\x20 Ok(builder.build()?)\n\
}\n\
\n",
);
}
fn emit_settings(out: &mut String) {
out.push_str(
"/// The connection values this worker is operated with.\n\
///\n\
/// Every one is read from the environment and NONE has an invented default: a\n\
/// concurrency or a reconnect budget the operator did not choose is a policy this\n\
/// crate has no business setting. The queue and the nodes are not here at all —\n\
/// they come from the document.\n\
struct Settings {\n\
\x20 endpoint: String,\n\
\x20 namespace: String,\n\
\x20 identity_prefix: String,\n\
\x20 max_concurrency: usize,\n\
\x20 reconnect_initial_backoff: Duration,\n\
\x20 reconnect_max_backoff: Duration,\n\
\x20 reconnect_max_attempts: usize,\n\
}\n\
\n\
impl Settings {\n\
\x20 /// Reads every connection value, failing loudly and naming the variable.\n\
\x20 fn from_env() -> anyhow::Result<Self> {\n\
\x20 Ok(Self {\n\
\x20 endpoint: require_var(\"AION_WORKER_ENDPOINT\")?,\n\
\x20 namespace: require_var(\"AION_WORKER_NAMESPACE\")?,\n\
\x20 identity_prefix: require_var(\"AION_WORKER_IDENTITY\")?,\n\
\x20 max_concurrency: require_parse(\"AION_WORKER_CONCURRENCY\")?,\n\
\x20 reconnect_initial_backoff: Duration::from_secs_f64(require_parse(\n\
\x20 \"AION_RECONNECT_INITIAL_BACKOFF_SECONDS\",\n\
\x20 )?),\n\
\x20 reconnect_max_backoff: Duration::from_secs_f64(require_parse(\n\
\x20 \"AION_RECONNECT_MAX_BACKOFF_SECONDS\",\n\
\x20 )?),\n\
\x20 reconnect_max_attempts: require_parse(\"AION_RECONNECT_MAX_ATTEMPTS\")?,\n\
\x20 })\n\
\x20 }\n\
\n\
\x20 /// The config for one connection. The node is the routing key that separates\n\
\x20 /// this process's several same-queue connections, and the identity carries it\n\
\x20 /// so an operator can tell them apart in the census.\n\
\x20 fn config(&self, node: Option<&str>) -> anyhow::Result<WorkerConfig> {\n\
\x20 let identity = match node {\n\
\x20 Some(node) => format!(\"{}-{node}\", self.identity_prefix),\n\
\x20 None => self.identity_prefix.clone(),\n\
\x20 };\n\
\x20 let mut builder = WorkerConfig::builder()\n\
\x20 .endpoint(self.endpoint.clone())\n\
\x20 .namespace(self.namespace.clone())\n\
\x20 .task_queue(TASK_QUEUE)\n\
\x20 .identity(identity)\n\
\x20 .max_concurrency(self.max_concurrency)\n\
\x20 .reconnect_initial_backoff(self.reconnect_initial_backoff)\n\
\x20 .reconnect_max_backoff(self.reconnect_max_backoff)\n\
\x20 .reconnect_max_attempts(self.reconnect_max_attempts);\n\
\x20 if let Some(node) = node {\n\
\x20 builder = builder.node(node);\n\
\x20 }\n\
\x20 Ok(builder.build()?)\n\
\x20 }\n\
}\n\
\n",
);
}
fn emit_env_helpers(out: &mut String) {
out.push_str(
"/// Reads a required environment variable, erroring with its name when unset.\n\
fn require_var(name: &str) -> anyhow::Result<String> {\n\
\x20 std::env::var(name)\n\
\x20 .map_err(|_| anyhow::anyhow!(\"required environment variable `{name}` is not set\"))\n\
}\n\
\n\
/// Reads and parses a required environment variable, erroring with its name.\n\
fn require_parse<T>(name: &str) -> anyhow::Result<T>\n\
where\n\
\x20 T: std::str::FromStr,\n\
\x20 T::Err: std::fmt::Display,\n\
{\n\
\x20 let value = require_var(name)?;\n\
\x20 value\n\
\x20 .parse::<T>()\n\
\x20 .map_err(|error| anyhow::anyhow!(\"environment variable `{name}` is invalid: {error}\"))\n\
}\n\
\n",
);
}
fn emit_tests(out: &mut String) {
out.push_str(
"#[cfg(test)]\n\
mod tests {\n\
\x20 use std::collections::BTreeSet;\n\
\n\
\x20 use super::declaration::Declaration;\n\
\x20 use super::{CONNECTIONS, handler, served};\n\
\n\
\x20 type TestResult = Result<(), Box<dyn std::error::Error>>;\n\
\n\
\x20 /// Every action the connection table serves is DECLARED by the embedded\n\
\x20 /// document, is REACHABLE on the node its connection registers, and has a\n\
\x20 /// handler in this binary. Without this the mismatch is discovered on the\n\
\x20 /// dial, where the server refuses the whole connection.\n\
\x20 #[test]\n\
\x20 fn every_served_action_is_declared_reachable_and_handled() -> TestResult {\n\
\x20 let declaration = Declaration::compile()?;\n\
\x20 for connection in CONNECTIONS {\n\
\x20 for action in connection.actions {\n\
\x20 let descriptor = declaration.descriptor(action, connection.node)?;\n\
\x20 assert_eq!(descriptor.name.as_str(), *action);\n\
\x20 assert!(handler(action).is_some(), \"no handler for `{action}`\");\n\
\x20 }\n\
\x20 }\n\
\x20 Ok(())\n\
\x20 }\n\
\n\
\x20 /// Every action the document declares without a body requires an out-of-band\n\
\x20 /// worker, and this binary is it.\n\
\x20 #[test]\n\
\x20 fn every_bodyless_action_is_served() -> TestResult {\n\
\x20 Declaration::compile()?.require_every_bodyless_action_served(&served())?;\n\
\x20 Ok(())\n\
\x20 }\n\
\n\
\x20 /// A `run`-bodied action is executed by the SERVER, so this binary must\n\
\x20 /// neither serve nor advertise one: the admission gate filters those actions\n\
\x20 /// out before deciding what a connection owes, and advertising one would\n\
\x20 /// promise work this binary does not do.\n\
\x20 #[test]\n\
\x20 fn no_server_executed_action_is_served() -> TestResult {\n\
\x20 let declaration = Declaration::compile()?;\n\
\x20 let served: BTreeSet<String> = served();\n\
\x20 for action in &declaration.contract().actions {\n\
\x20 if action.body.is_some() {\n\
\x20 assert!(\n\
\x20 !served.contains(&action.name),\n\
\x20 \"`{}` carries a declarative body; the server executes it\",\n\
\x20 action.name\n\
\x20 );\n\
\x20 }\n\
\x20 }\n\
\x20 Ok(())\n\
\x20 }\n\
\n\
\x20 /// Each node appears at most ONCE in the connection table. Two connections on\n\
\x20 /// one node would let the server land an activity on the one with no handler\n\
\x20 /// for it, because routing never looks at the activity type.\n\
\x20 #[test]\n\
\x20 fn no_node_is_dialled_twice() {\n\
\x20 let mut nodes = BTreeSet::new();\n\
\x20 for connection in CONNECTIONS {\n\
\x20 assert!(\n\
\x20 nodes.insert(connection.node),\n\
\x20 \"node {:?} appears twice in the connection table\",\n\
\x20 connection.node\n\
\x20 );\n\
\x20 }\n\
\x20 }\n\
}\n",
);
}
pub(super) fn handler_ident(action: &str) -> String {
if crate::codegen::activity_worker_rust::is_rust_keyword(action) {
return format!("r#{action}");
}
action.to_owned()
}