Simpler hello world API

[pickfire]

The hello world in readme seemed a bit long compared to axum few lines.

use axum::{
    routing::get,
    Router,
};

#[tokio::main]
async fn main() {
    // build our application with a single route
    let app = Router::new().route("/", get(|| async { "Hello, World!" }));

    // run our app with hyper, listening globally on port 3000
    let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
    axum::serve(listener, app).await.unwrap();
}

With apalis:

use apalis::prelude::*;
use serde::{Deserialize, Serialize};
use apalis_redis::{RedisStorage, Config};

#[derive(Debug, Deserialize, Serialize)]
struct Email {
    to: String,
}

async fn send_email(job: Email, data: Data<usize>) -> Result<(), Error> {
    Ok(())
}

#[tokio::main]
async fn main() {
    let redis = std::env::var("REDIS_URL").expect("Missing REDIS_URL env variable");
    let conn = apalis_redis::connect(redis).await.unwrap();
    let storage = RedisStorage::new(conn);
    Monitor::new()
        .register({
            WorkerBuilder::new(&format!("quick-sand"))
                .concurrency(2)
                .data(0usize)
                .backend(storage.clone())
                .build_fn(send_email)
        })
        .run()
        .await
        .unwrap();
}

Can we simplify it further?

use apalis_redis::RedisStorage;
use apalis::{Monitor, Router, task};

#[tokio::main]
async fn main() {
    let monitor = Monitor::new().register(task(|| async { "Hello, world!" }));

    let storage = RedisStorage::from_url("redis://").await.unwrap();
    apalis::start(storage, monitor).await.unwrap();
}

I don't really have that much experience with apalis and I did notice that apalis supports multiple backends. Maybe something simpler would make it less overwhelming for new users?

[geofmureithi]

While apalis is influenced by axum ergonomics I dont think that is possible. Here is a minimum example:

#[tokio::main]
async fn main() {
    let storage = RedisStorage::from_url("redis://").await.unwrap();
    WorkerBuilder::new("worker_name")
        .backend(storage)
        .build(|task_type: Email | async { "Hello, world!" })
        .run()
        .await
        .unwrap();
}

This is the best we can do without having problems with typing. Remember axum just handles req/res and we handle req/res and streams from backends.

[geofmureithi]

If you need to solve multiple tasks types with the same worker, please use enum.
If you need to run multiple workers, there are two ways:

1. Use monitor, which offers a single point for event handling and restarts
2. Worker API allows different ways that return either a Future or a Stream which should allow you to combine workers as you wish.

This information will be on the website documentation which is currently being worked on.

[pickfire]

Yes, enum does work. But then we basically have to use enum to do another layer of routing if we need many functions (which I believe most task servers do), and at the same time apalis board won't have the information of the task name.

I imagined an API like this:

use apalis_redis::RedisStorage;
use apalis::{Monitor, Router, task};

async fn task_b() -> &'static str {
    "Hello b"
}

#[tokio::main]
async fn main() {
    let monitor = Monitor::new()
        .register(task_fn("task_a", || async { "Hello, world!" }))
        .register(task(task_b));

    let storage = RedisStorage::from_url("redis://").await.unwrap();
    apalis::start(storage, monitor).await.unwrap();
}

And when submitting the task:

use apalis_redis::RedisStorage;
use apalis::{Router, task};

async fn task_b() -> &'static str {
    "Hello b"
}

#[tokio::main]
async fn main() {
    let storage = RedisStorage::from_url("redis://").await.unwrap();
    storage.submit(task_b, ()).await.unwrap();;
}

Otherwise users would have to do this, since I doubt the user would have the same request and response, which is a lot of boilerplates:

use apalis_redis::RedisStorage;
use apalis::prelude::WorkerBuilder;

enum TaskPayload {
    TaskA,
    TaskB(usize),
}

enum TaskResult {
    TaskA(&'static str),
    TaskB(usize),
}

async fn router(payload: TaskPayload) -> Result<TaskResult, ()> {
    Ok(match payload {
        TaskA => TaskResult::TaskA(task_a()),
        TaskB(n) => TaskResult::TaskB(task_b(n)),
    })
}

async fn task_a() -> &'static str {
    "Hello a"
}

async fn task_b(n: usize) -> usize {
    1 + n
}

#[tokio::main]
async fn main() {
    let storage = RedisStorage::from_url("redis://").await.unwrap();
    let worker = WorkerBuilder::new("worker_name").backend(storage).build(router);
    worker.run().await.unwrap();
}

async fn _submit_task() {
    let storage = RedisStorage::from_url("redis://").await.unwrap();
    storage.push(TaskPayload::TaskB(1)).await.unwrap();
}

Was trying to do an MVP to check if it is possible to do it type-safe, the API I based it from dramatiq. I got it working from claude: https://play.rust-lang.org/?version=stable&mode=debug&edition=2024&gist=283d811dbc7e6aae0028eb239b6f4879

Or this, based on tower + type name + handling no arguments instead, without having to manually write type erasure https://play.rust-lang.org/?version=stable&mode=debug&edition=2024&gist=2e059cd22ade936a675dfef32a4f967e

[geofmureithi]

• Tower already provides a routing middleware.
• Interms of the api suggested here, I think the only thing you would need to change is instead of calling it Broker, you can call it Router or something similar then implement Service then you can do:

let router = Router::new();
        let router = router
            .register(add_2)
            .register(concat_a)
            .register(get_current_time); 
let worker = WorkerBuilder::new("worker_name").backend(storage).build(router);
worker.run().await.unwrap();

You should be able to pass in anything that implements Service to the build function. This is the flexibility that apalis offers. You can easily adapt it to your usecase.

Another thing to note is that what you are trying to attempt is against apalis's philosophy. Here are some main thoughts to consider:

1. Apalis is modeled behind a 1 worker to 1 job type philosophy (with each type being strongly typed). You are adding multiple jobs under 1 config, meaning the board will not be filterable among other functionality. Also remember that apalis supports multiple backends and with different capabilities and functionality.
2. The API .register(task_fn("task_a", || async { "Hello, world!" })) only makes sense on paper but in reality there is a difference between workers and queues. A single queue may have many workers, and each worker needs to send a unique heartbeat else we can never find stalled jobs. The api suggested does not also indicate where the jobs will be fetched from The api suggested only works for one backend yet one my want to use more than one backend eg cron + postgres.
3. Remember that there is the Backend trait that you haven't discussed. Wiring this together requires type constraints that make the API more verbose.
4. I don't think this is possible storage.submit(task_b, ()).await.unwrap(); (without macros). task_b would be a functional pointer and again might not be the name of the queue. I do have another similar API but it would target Task instead of Backend.
5. Apalis doesnt have any global variables, you compose everything as required.
6. Your example works with json because json is schema-less but this may not work as expected with other codecs leading to hacks like this.
7. Apalis uses dependency injection to provide relevant context for task execution and this also requires some extra constraints connecting Task to Backend, Worker and TaskFn.

I suspect there are improvements that can be made to the ergonomics but not to the level you are suggesting. I also Rust is not as flexible as python. I would want to keep the API macro free and without any globals: explicit is better than implicit

That said I am open to improvements via extension traits targeting Worker, WorkerBuilder or Monitor and we can discuss some of the constraints/implications.

[pickfire]

Hmm, I guess we are targeting different segments, on your end mainly around large tasks. And on my side currently I am dealing with lots of small tasks (I would have to do 15 different queues on 4 environments) and a couple large tasks. Yes, definitely if the workload is balanced it is much easier to reason about and won't get k8s silent oom kills (currently celery all on one queue, probably splitting it out).

So I guess most likely there will be 2 splits, io heavy or cpu heavy, mine is on io/network heavy, mostly waiting for rate limits. And don't really want to spend so much resource to micro-manage each queues. Hmm, I never thought about a usecase where I have 1 task using 2 storage at the same time, more like I need N tasks using 1 queue. If I were to split it out, I just put M cpu heavy tasks on 2nd queue.

I would be more than happy to contribute to improve the ergonomics to what I mentioned, we can brainstorm it out, although I won't be using it soon myself yet, I will be more than happy to contribute. But I am just worried about the timeline given that it's already 1.0.0-alpha.7 now to do any breaking changes.

Note, I won't be seeing myself dogfooding anytime soon (due to having a hard time convincing my boss).

1. Different goals here, I need 2 redis queues (io heavy and cpu/ram heavy) with ease of managing many small tasks and few heavy tasks, most time spent waiting for rate limiting.
2. At least on our use case, we need to spread out the workers to do different tasks, instead of doing the same tasks due to rate limiting.
3. Backend trait from the playground I sent, tasks are separated as another layer.
4. Can take a look at the bottom part of the playground, https://play.rust-lang.org/?version=stable&mode=debug&edition=2024&gist=2e059cd22ade936a675dfef32a4f967e. How it works is that the registry is separated from task submission, where task submission only need into task while keeping it type-safe. But here it requires the original function type.
5. Yup, similar to what I shared, except the routing layer have might need a map from task name to task function. I am thinking to sideload the metadata with the original data like {"_meta": {...}} where there be task_name for routing as well as be middleware layers (at most one task at a time layer, opentelemetry tracing layer, or add extra task metadata).
6. Yes, but we could always keep the other layer schema-less and the inner layer specified by the users.
7. Yes, but looks like because apalis put multiple generics at one layer, it looks harder to read campared to axum.

I suspect there are improvements that can be made to the ergonomics but not to the level you are suggesting. I also Rust is not as flexible as python. I would want to keep the API macro free and without any globals: explicit is better than implicit

I am definitely aligned with you here, and if we could make the routing layer optionally opt-in, it would be a plus. I am thinking to do some experimenting on a separate repo to do a PoC on more ergonomic API (maybe with only redis) with the API based mostly on dramatiq, not sure how that would help you.

Yes, for our current stuff even though there are less maintenance cost and we can save some money with basic bin packing, there are some downsides to mixing tasks, here are some pitfalls:

• silent k8s OOM on single thread due to running one single celery pod running multiple processes, we could run one process but that defeats the purpose, two ways to solve this, one could separate heavy cpu/mem tasks into a separate queue with a lower concurrency per worker, or just use only one worker for one (smaller) pod
• which leads to observerability issue, mainly due to not having tools being able to observe memory usage on concurrency level, not knowing the memory usage of each task (even within one task, based on input the memory usage may vary) makes it hard to debug issues, I wish tokio console have the ability to check cpu or mem usage per task?

[geofmureithi]

Backend trait from the playground I sent, tasks are separated as another layer.

The statement above tells me we are not talking about the same things (The Backend trait is the core apalis-core trait). All the features you are describing including small tasks are well catered for. I would encourage you check the different components in the latest docs: https://docs.rs/apalis-core/1.0.0-alpha.7/apalis_core/index.html

The specific ergonomics you are requesting are pretty much possible currently, eg in apalis-workflow. I just havent had such a request and the API wasnt stable yet.

Regarding:

Yup, similar to what I shared, except the routing layer have might need a map from task name to task function. I am thinking to sideload the metadata with the original data like {"_meta": {...}} where there be task_name for routing as well as be middleware layers (at most one task at a time layer, opentelemetry tracing layer, or add extra task metadata).

This is solved here: https://docs.rs/apalis-core/1.0.0-alpha.7/apalis_core/task/metadata/trait.MetadataExt.html

Yes, but we could always keep the other layer schema-less and the inner layer specified by the users:

Not sure what you mean here, but the Backend decides the Codec and whether to change it. Json works for you simply because the Vec<u8> representation of a JSON blob is compatible with type T: Serialize but this would not work with bincode because Vec<u8> and T would be aligned differently. That is the problem with your implementation.

Yes, but looks like because apalis put multiple generics at one layer, it looks harder to read compared to axum.

What do you specifically mean? what Generics?

which leads to observerability issue, mainly due to not having tools being able to observe memory usage on concurrency level, not knowing the memory usage of each task (even within one task, based on input the memory usage may vary) makes it hard to debug issues, I wish tokio console have the ability to check cpu or mem usage per task?

Maybe: https://docs.rs/tokio-metrics/latest/tokio_metrics/

Conclusion

• I will include a single worker to multiple jobs impl using the MakeShared trait.
• Beyond that, the other changes are well doable with functional rust.
• Feedback from other users is that ergonomics is not the biggest headache. As opposed to axum, most apalis users setup their handlers only once.

[geofmureithi]

And FYI your example is outdated, please check the latest README.md

[geofmureithi]

Closed in favor of #615