pizza

What is Pizza?

Pizza is a distributed task engine and actor runtime built on top of Kubernetes without redis/postgres. It provides a simple and scalable way to run tasks and actors in a distributed environment, similar to asynq but designed to run natively on Kubernetes StatefulSets — no Redis or external broker required.

Key Features

• Kubernetes-native: Designed to run as a Kubernetes StatefulSet, leveraging stable pod identities for reliable task ownership and heartbeat tracking.
• No external dependencies: Pizza uses Kubernetes itself as the coordination layer — no Redis, no Postgres.
• Distributed task processing: Enqueue and process tasks across multiple workers with at-least-once delivery and heartbeat-based failure recovery.
• Actor runtime: Supports long-running actors alongside one-shot tasks.

Example

package main

import (
	"context"
	"encoding/json"
	"log/slog"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"

	"github.com/abit2/pizza"
	"github.com/abit2/pizza/db"
	"github.com/abit2/pizza/log"
	"github.com/abit2/pizza/task/task/generated"
	"github.com/abit2/pizza/utils"
	badger "github.com/dgraph-io/badger/v4"
)

// EmailHandler processes "send_email" tasks.
type EmailHandler struct{}

func (h *EmailHandler) Process(ctx context.Context, task *generated.Task) error {
	// task.GetPayload() contains the raw bytes you enqueued
	slog.Info("sending email", "payload", string(task.GetPayload()))
	return nil
}

func main() {
	// 1. Open an embedded Badger database.
	//    On a Kubernetes StatefulSet mount this path on a PersistentVolume so
	//    each pod keeps its own shard of the queue across restarts.
	bdb, err := badger.Open(badger.DefaultOptions("./data"))
	if err != nil {
		panic(err)
	}
	defer bdb.Close()

	// 2. Wrap Badger in Pizza's DB layer.
	clock := utils.NewRealClock()
	lvl := new(slog.LevelVar)
	lvl.Set(slog.LevelInfo)
	logger := log.NewLogger(slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: lvl})))

	dbWrap, err := db.New(bdb, logger, nil, clock)
	if err != nil {
		panic(err)
	}

	// 3. Enqueue a task.
	headers, _ := json.Marshal(map[string]string{
		pizza.TaskType: "send_email", // routes to the matching Handler
	})
	_, err = dbWrap.Enqueue([]byte("notifications"), []byte(`{"to":"user@example.com"}`), headers, 3)
	if err != nil {
		panic(err)
	}

	// 4. Start the server.
	srv := pizza.NewServer(logger, dbWrap, &pizza.ServerConfig{
		Queues:                  []string{"notifications"},
		Concurrency:             10,
		PromiseInterval:         5 * time.Second,
		RecoveryInterval:        10 * time.Second,
		GracefulShutdownTimeout: 5 * time.Second,
	}, clock)

	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	var wg sync.WaitGroup
	srv.Run(ctx, &wg)
	// NOTE: run the srv before registering the Handlers
	srv.SetupHandler(map[string]pizza.Handler{
		"send_email": &EmailHandler{},
	})

	<-ctx.Done()
	srv.Stop()
	wg.Wait()
}

How it maps to a Kubernetes StatefulSet

Each StatefulSet replica opens its own Badger volume (via a volumeClaimTemplate). Tasks are enqueued into a shared queue name; any worker (running concurrently) can dequeue and process them. Heartbeat-based lease recovery ensures that if a pod crashes mid-task, the work is automatically re-queued and processed by the worker.