InEngine

InEngine is a plugin-based application that allows commands to be queued, scheduled, and run directly.

How It Works

InEngine uses a WASM plugin system to load and run commands. It also has a built-in command to run external programs.

Run the echo command:

inengine echo --text "Hello, world"

Run an external program with the exec command. The program must be listed in the [exec_whitelist] table of inengine.toml:

[exec_whitelist]
ls = "/bin/ls"

inengine exec -e ls -- -lhp

See Exec for the timeout and argument options.

Quick Start: Queuing

Publish a command to the queue:

inengine queue publish --plugin inengine-core --command echo --args "text=Hello, world"

Consume the next message from the queue:

inengine queue consume

Quick Start: Server Mode

Start the server to run scheduled commands and consume queued messages:

inengine -s

Quick Start: Plugins

Add the plugin directory to the [plugins] table of inengine.toml:

[plugins]
sample = "target/plugins/sample"

Run a command from the plugin:

inengine plugin --name sample --command greet --args name=Ethan

Commands

Commands can be WASM plugin modules, Rust types, or external programs. They can be queued, scheduled, or run from the command line.

Echo

Print text to the terminal:

inengine echo --text "Hello, world"

Exec

Run an external program. The program must be listed in the exec_whitelist section of inengine.toml.

Add a program to the whitelist:

[exec_whitelist]
hello = "/path/to/hello.py"

Run the program:

inengine exec -e hello

Pass arguments after --:

inengine exec -e hello -- --verbose --count 5

Set a timeout in seconds. The default is 900 seconds. Use 0 for no limit:

inengine exec -e hello -t 60

The program also stops at the timeout of a scheduled job and when the server stops. Exec kills only the program it started, so a process that the program starts in the background keeps running.

Exec prints the standard output of the program, then its standard error. When a stream is larger than 1 MiB, Exec keeps its first 512 KiB and its last 512 KiB, and prints a line between them that says how many bytes it did not keep:

[1048584 bytes not kept]

Chain

Multiple commands can run in sequence as a chain. If one command in the chain fails, the rest do not run. A chain is built in Rust code. See Scheduling a Chain for an example.

Output

Commands can write colored text to the terminal. The available output methods are:

Write line breaks with newlines, which takes the number of newline characters to write.

Progress Bar

Commands can show a progress bar in the terminal:

1. Call set_progress_max with the total number of steps.
2. Call update_progress with the current step number.

View Commands

Run inengine with no arguments to see the list of commands:

inengine

View Help

Run a command with --help to see its options:

inengine echo --help
inengine queue publish --help

Configuration

Overview

Configuration is stored in an inengine.toml file. Use the -c flag to specify a different file:

inengine -c /etc/inengine/production.toml echo --text "hello"

Copy inengine.example.toml to get started. Every setting has a default value. Keep only the lines you change.

Full Example

[plugins]
sample = "plugins/sample"

[exec_whitelist]
hello = "/usr/bin/hello"

[mail]
host = "localhost"
port = 25
from = "no-reply@example.com"
username = ""
password = ""
encryption = "none"

[queue]
use_compression = false
primary_queue_consumers = 8
secondary_queue_consumers = 8
queue_driver = "file"
queue_name = "InEngineQueue"
max_retries = 0
recover_on_start = false

[queue.redis]
host = "127.0.0.1"
port = 6379
database = 0
password = ""

[queue.file]
base_path = "../"

[plugin_limits]
timeout_seconds = 900
memory_mib = 256

Plugins

Exec Whitelist

Mail

none sends mail in plain text. starttls, usually on port 587, upgrades the connection to TLS and refuses a server that cannot. tls, usually on port 465, uses TLS from the first byte. A user name with none is an error unless the host is localhost or a loopback address, so a password never crosses the network in plain text. A port that does not match the encryption is a warning.

Queue

General

Warning: Enable recover_on_start only when this server is the only consumer of the queue. Recovery cannot tell a message left by a crash from one that another consumer is still processing.

Redis

Used when queue_driver is "redis". Requires Redis 6.2 or later.

File

Used when queue_driver is "file".

Plugin Limits

Logging

InEngine uses the INENGINE_LOG environment variable to control log output. The variable accepts any valid tracing filter directive.

INENGINE_LOG=info inengine -s

INENGINE_LOG=inengine_core=debug inengine -s

The default level is warn.

Queuing

InEngine’s queue allows commands to run in the background. A command should be queued when it needs to run once, as soon as possible, outside the current process.

Queue Drivers

Set the queue_driver field in inengine.toml to select a driver.

File

The file driver writes queued messages to the file system. It is useful for development and testing.

[queue]
queue_driver = "file"

[queue.file]
base_path = "../"

Redis

Redis is suitable for production use. InEngine uses Redis LMOVE for durable queue operations. Messages are not lost if the process stops.

Redis 6.2 or later is required.

[queue]
queue_driver = "redis"

[queue.redis]
host = "127.0.0.1"
port = 6379
database = 0
password = ""

Publishing Commands

Publish a command to the queue by its plugin name and command name. The built-in commands use the plugin name inengine-core:

inengine queue publish --plugin inengine-core --command echo --args "text=Hello"

Publish a command from a plugin by the plugin name in inengine.toml:

inengine queue publish --plugin sample --command greet --args name=Ethan

The publisher does not check that the command exists. A consumer that cannot find the command moves the message to the failed stage.

Publish with a JSON payload:

inengine queue publish --plugin inengine-core --command echo --json '{"text":"Hello"}'

Publish to the secondary queue:

inengine queue publish --plugin inengine-core --command echo --args "text=Hello" --secondary

Set the maximum number of retries. The value must be from 0 to 100:

inengine queue publish --plugin inengine-core --command echo --args "text=Hello" --max-retries 3

Consuming Commands

Consume one message:

inengine queue consume

Consume a specific number of messages:

inengine queue consume --count 10

Consume all messages:

inengine queue consume --all

Consume from the secondary queue:

inengine queue consume --secondary

Viewing Queue Lengths

The queue length command shows a count of pending, in-progress, and failed messages for each queue:

inengine queue length

The queue status command prints the same counts.

Peeking at Queued Commands

View pending messages:

inengine queue peek --pending

View failed messages:

inengine queue peek --failed

View in-progress messages:

inengine queue peek --in-progress

View messages as JSON:

inengine queue peek --pending --json

View a specific range (default is 0 to 9):

inengine queue peek --pending --from 0 --to 99

Peek at the secondary queue:

inengine queue peek --pending --secondary

Republishing Failed Commands

Commands that fail are moved to a failed stage. Republish them back to the pending queue:

inengine queue republish

Set a limit (default is 100):

inengine queue republish --limit 1000

Republish from the secondary queue:

inengine queue republish --secondary

Recovering In-Progress Messages

Warning: Stop every other consumer of the queue first. Recovery cannot tell a message left by a crash from one that another consumer is still processing. That message would run twice.

Move in-progress messages back to the pending queue:

inengine queue recover

Recover the secondary queue:

inengine queue recover --secondary

Flushing the Queue

Clear pending messages:

inengine queue flush --pending

Clear failed messages:

inengine queue flush --failed

Clear in-progress messages:

inengine queue flush --in-progress

Clear messages from the secondary queue:

inengine queue flush --pending --secondary

Primary and Secondary Queue

There is no difference between the primary and secondary queues except that the primary queue is the default. Use the secondary queue for a different class of work. For example, send long-running jobs to the secondary queue and short jobs to the primary queue.

For more than two queues, run separate InEngine instances with different configuration files:

inengine -c queue-a.toml -s
inengine -c queue-b.toml -s

Message Compression

Enable use_compression to compress queued messages with gzip. This reduces memory use when commands carry a lot of internal state. In a high-throughput scenario with limited bandwidth, compression also reduces network use.

Do not enable compression when queued commands are small. The CPU cost of compression can reduce throughput without a benefit.

[queue]
use_compression = true

Scheduling

Commands can run on recurring cron schedules.

A scheduled command is different from a queued command. Queue a command to run it once, as soon as possible. Schedule a command to run it many times, on a fixed interval.

InEngine defines schedules in code, not in a data store. This lets the schedule live alongside the command code in version control.

Scheduling a Command

Use the Schedule struct to register commands with cron expressions.

#![allow(unused)]
fn main() {
use inengine_core::scheduling::schedule::Schedule;

let mut schedule = Schedule::new();

let job = schedule
    .command(MyCommand { /* ... */ })
    .every_five_minutes()?;
schedule.add(job);
}

Use a custom cron expression:

#![allow(unused)]
fn main() {
let job = schedule
    .command(MyCommand { /* ... */ })
    .cron("15 * * * * *")?;
schedule.add(job);
}

Convenience Methods

Scheduling a Chain

Schedule a group of commands to run as a chain. The chain stops if one command fails.

#![allow(unused)]
fn main() {
let cmds: Vec<Box<dyn Command>> = vec![
    Box::new(TransferFile { path: path1 }),
    Box::new(TransferFile { path: path2 }),
    Box::new(TransferFile { path: path3 }),
];

let job = schedule.commands(cmds).daily()?;
schedule.add(job);
}

Lifecycle Methods

Scheduled jobs support lifecycle hooks.

Ping a URL before or after the command runs:

#![allow(unused)]
fn main() {
let job = schedule
    .command(MyCommand { /* ... */ })
    .every_five_minutes()?
    .ping_before("https://example.com/start")
    .ping_after("https://example.com/done");
schedule.add(job);
}

Write command output to a file:

#![allow(unused)]
fn main() {
let job = schedule
    .command(MyCommand { /* ... */ })
    .hourly()?
    .write_output_to("/var/log/my-command.log");
schedule.add(job);
}

Append command output to a file:

#![allow(unused)]
fn main() {
let job = schedule
    .command(MyCommand { /* ... */ })
    .hourly()?
    .append_output_to("/var/log/my-command.log");
schedule.add(job);
}

Email command output:

#![allow(unused)]
fn main() {
let job = schedule
    .command(MyCommand { /* ... */ })
    .daily()?
    .email_output_to("ops@example.com");
schedule.add(job);
}

The output is written, appended, and mailed when the command succeeds and when it fails. The mail subject is Command completed: <name> or Command failed: <name>. A failure mail ends with the error and each of its causes. Nothing is written or mailed for a command that the server stopped, a command that ignored its deadline, or a command that panicked. Nothing is written or mailed for a successful run whose ping_after request fails, because that run fails.

A ping fails when the server answers with an error status, and the job fails with it.

Set a timeout:

#![allow(unused)]
fn main() {
use std::time::Duration;

let job = schedule
    .command(MyCommand { /* ... */ })
    .hourly()?
    .timeout(Duration::from_secs(300));
schedule.add(job);
}

Run custom logic before or after. The hooks run on the server’s async threads, so they must not block:

#![allow(unused)]
fn main() {
let job = schedule
    .command(MyCommand { /* ... */ })
    .every_five_minutes()?
    .before_run(|| println!("starting"))
    .after_run(|| println!("finished"));
schedule.add(job);
}

Running a Schedule

Give the schedule to the server:

#![allow(unused)]
fn main() {
use std::sync::Arc;

use inengine_core::queue::CommandRegistry;
use inengine_core::server::host::Server;

let server = Server::new(settings, Arc::new(CommandRegistry::new()), schedule);
server.run(cancel).await;
}

A plugin can also add schedules. See Plugins.

Viewing Scheduled Commands

List the plugin schedules and their next fire time:

inengine schedule list

Running the Scheduler

The server runs the plugin schedules:

inengine -s

A schedule built in Rust code runs only in a server that your program starts with Server::new.

Server

When run as a server, InEngine runs scheduled commands and consumes queued messages.

Running the Server

Start the server in the foreground:

inengine -s

Use a custom configuration file:

inengine -c /etc/inengine/production.toml -s

Stopping the Server

The server stops on SIGINT (Ctrl+C) or SIGTERM. It stops each running plugin call and each program that the exec command started within about 10 milliseconds. A queued message whose command was stopped returns to pending, and the attempt does not count. The server waits for every other running command to finish. A consumer that waits for a message finishes that wait first, so the server can take up to 5 seconds to stop. A message that such a consumer claims after the stop signal returns to pending without a run. A scheduled command that ran past its timeout is not waited for, and the process exits at most 10 seconds after the server stops.

Running with systemd

Create a unit file at /etc/systemd/system/inengine.service:

[Unit]
Description=InEngine
After=network.target

[Service]
Type=simple
User=inengine
WorkingDirectory=/opt/inengine
ExecStart=/opt/inengine/inengine -s
Restart=on-failure

[Install]
WantedBy=multi-user.target

Enable and start the service:

sudo systemctl enable inengine
sudo systemctl start inengine

Running with Supervisor

Create a configuration file at /etc/supervisor/conf.d/inengine.conf:

[program:inengine]
process_name=%(program_name)s_%(process_num)02d
directory=/opt/inengine
command=/opt/inengine/inengine -s
autostart=true
autorestart=true
user=inengine
numprocs=1
redirect_stderr=true
stdout_logfile=/var/log/inengine.log

Reload and start:

sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start inengine:*

Logging

Set the INENGINE_LOG environment variable to control log output:

INENGINE_LOG=info inengine -s

Plugins

InEngine loads plugins as WASM components built with the WebAssembly Component Model. A plugin provides commands and schedules. A plugin command can run from the command line, from the queue, or on a schedule.

Running a Plugin Command

inengine plugin --name sample --command greet --args name=Ethan

Deploying a Plugin

1. Build the plugin as a .wasm component.
2. Copy the .wasm file to a directory.
3. Add the directory to inengine.toml:

[plugins]
sample = "/path/to/plugin/directory"

The key is the plugin name. InEngine loads every .wasm file in the directory. A relative path resolves against the working directory. The name inengine-core is reserved for the built-in commands.

Queuing a Plugin Command

Publish a plugin command with the plugin name and the command name:

inengine queue publish --plugin sample --command greet --args name=Ethan

The server and inengine queue consume both load the plugins in inengine.toml, so either one can run the message.

WIT Interface

Plugins implement the inengine:plugin WIT world. The interface has two parts: commands and scheduling.

package inengine:plugin;

interface commands {
    record command-descriptor {
        name: string,
        help-text: string,
    }

    list-commands: func() -> list<command-descriptor>;
    run-command: func(name: string, args-json: string) -> result<string, string>;
}

interface scheduling {
    record schedule-entry {
        command-name: string,
        cron-expression: string,
    }

    list-schedules: func() -> list<schedule-entry>;
}

world plugin {
    export commands;
    export scheduling;
}

Scheduling from a Plugin

Return ScheduleEntry records from list_schedules to schedule a plugin command:

#![allow(unused)]
fn main() {
fn list_schedules() -> Vec<exports::inengine::plugin::scheduling::ScheduleEntry> {
    vec![
        exports::inengine::plugin::scheduling::ScheduleEntry {
            command_name: "greet".to_string(),
            cron_expression: "0 */5 * * * *".to_string(),
        },
    ]
}
}

The server runs each entry on its cron schedule, with the arguments {}. inengine schedule list shows each entry as a job named <plugin>::<command>. An entry for a command that the plugin does not list, or with a cron expression that does not parse, is skipped with a warning.

Limits

Each plugin call runs in a sandbox with two limits from the [plugin_limits] table of inengine.toml:

[plugin_limits]
timeout_seconds = 900
memory_mib = 256

A call that runs longer than timeout_seconds is stopped and fails. A scheduled job with a shorter timeout stops the call at that timeout. A call that needs more memory than memory_mib fails. Loading a plugin has a fixed limit of 10 seconds.

Writing a Plugin

Prerequisites

Install cargo-component and the wasm32-unknown-unknown target:

cargo install cargo-component --locked
rustup target add wasm32-unknown-unknown

From the rust directory, just install-tools installs both.

Create the Crate

Create a new Rust library crate with crate-type = ["cdylib"]. Add wit-bindgen and serde_json as dependencies. Copy the rust/inengine-core/wit/plugin.wit file from the InEngine repository into a wit folder in your crate. cargo-component then writes src/bindings.rs on each build. Your crate does not use that file, because wit_bindgen::generate! generates the bindings, so add it to .gitignore.

Implement the Interfaces

Use wit_bindgen::generate! to generate guest bindings, then implement the traits:

#![allow(unused)]
fn main() {
wit_bindgen::generate!({
    path: "wit",
    world: "plugin",
});

struct MyPlugin;

export!(MyPlugin);

impl exports::inengine::plugin::commands::Guest for MyPlugin {
    fn list_commands() -> Vec<exports::inengine::plugin::commands::CommandDescriptor> {
        vec![
            exports::inengine::plugin::commands::CommandDescriptor {
                name: "greet".to_string(),
                help_text: "Print a greeting".to_string(),
            },
        ]
    }

    fn run_command(name: String, args_json: String) -> Result<String, String> {
        match name.as_str() {
            "greet" => {
                let who = serde_json::from_str::<serde_json::Value>(&args_json)
                    .ok()
                    .and_then(|v| v.get("name").and_then(|n| n.as_str()).map(String::from))
                    .unwrap_or_else(|| "world".to_string());
                Ok(format!("Hello, {who}!"))
            }
            _ => Err(format!("unknown command: {name}")),
        }
    }
}

impl exports::inengine::plugin::scheduling::Guest for MyPlugin {
    fn list_schedules() -> Vec<exports::inengine::plugin::scheduling::ScheduleEntry> {
        vec![]
    }
}
}

Build

From the plugin directory:

cargo component build --release --target wasm32-unknown-unknown

The compiled component is written to target/wasm32-unknown-unknown/release/<crate_name>.wasm.

Build for wasm32-unknown-unknown rather than a WASI target. The host provides no WASI imports, so a component built for wasm32-wasip1 or wasm32-wasip2 fails to load with an error about a missing wasi:cli implementation.

From the rust directory, just build-sample-plugin builds the sample plugin and copies it to target/plugins/sample/.

Command Arguments

The args_json parameter is a JSON object. When a user passes --args key=value pairs, they arrive as {"key": "value"}.

License

InEngine is licensed under the Apache License, Version 2.0. The full text is in the LICENSE file at the root of the repository.