sup

README

sup

sup is a small actor supervision toolkit for Go, with an optional reactive signal package.

It provides typed inboxes for actor communication, OTP-style supervision with restart policies, reactive values that can be composed and observed, and an optional HTTP hub for inspecting actors, signals, supervision, and events.

Features

• Idiomatic actors — An actor is any value that implements ID() and Run(context.Context) error.
• Supervisor trees — Supervisors are actors too, so they can supervise actors or other supervisors.
• Restart policies — Permanent, Transient, and Temporary policies control when actors restart.
• Panic recovery — Panics are recovered, wrapped with a stack trace, reported, and handled by the restart policy.
• Typed inboxes — CastInbox[T] and CallInbox[T, R] provide type-safe asynchronous and request/reply messaging.
• Reactive signals — The rx package provides passive signals and supervised derived state.
• Channel helpers — Map, Filter, Distinct, Debounce, and explicit throttle helpers compose ordinary channels.
• Runtime inspection — Optional Inspect, runtime events, and hub expose useful metadata for debugging and dashboards.

Installation

go get github.com/webermarci/sup

Quick start

This example defines a counter actor with a fire-and-forget increment inbox and a request/reply get inbox.

package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/webermarci/sup"
)

type GetMessage struct{}

type IncrementMessage struct {
	Amount int
}

type Counter struct {
	id             string
	GetInbox       *sup.CallInbox[GetMessage, int]
	IncrementInbox *sup.CastInbox[IncrementMessage]
	State          int
}

func NewCounter(id string) *Counter {
	return &Counter{
		id:             id,
		GetInbox:       sup.NewCallInbox[GetMessage, int](8),
		IncrementInbox: sup.NewCastInbox[IncrementMessage](8),
	}
}

func (c *Counter) ID() string {
	return c.id
}

func (c *Counter) Get(ctx context.Context) (int, error) {
	return c.GetInbox.Call(ctx, GetMessage{})
}

func (c *Counter) Increment(ctx context.Context, amount int) error {
	return c.IncrementInbox.Cast(ctx, IncrementMessage{Amount: amount})
}

func (c *Counter) Run(ctx context.Context) error {
	for {
		select {
		case <-ctx.Done():
			return nil

		case req := <-c.GetInbox.Receive():
			req.Reply(c.State, nil)

		case msg := <-c.IncrementInbox.Receive():
			c.State += msg.Amount
		}
	}
}

func main() {
	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer cancel()

	counter := NewCounter("counter")

	supervisor := sup.NewSupervisor("root",
		sup.WithPolicy(sup.Permanent),
		sup.WithRestartDelay(time.Second),
		sup.WithRestartLimit(5, 10*time.Second),
		sup.WithEventSink(sup.EventSinkFunc(func(event sup.Event) {
			if event.Type == sup.EventActorStopped && event.Err != nil {
				fmt.Printf("actor %s failed: %v\n", event.Actor.ID(), event.Err)
			}
		})),
		sup.WithActors(counter),
	)

	done := make(chan error, 1)
	go func() { done <- supervisor.Run(ctx) }()

	_ = counter.Increment(ctx, 10)
	_ = counter.Increment(ctx, 32)

	count, err := counter.Get(ctx)
	if err != nil {
		panic(err)
	}

	fmt.Println("count:", count)

	cancel()
	if err := <-done; err != nil {
		fmt.Println("supervisor stopped:", err)
	}
}

Actors

An actor implements the Actor interface:

type Actor interface {
	ID() string
	Run(context.Context) error
}

Inspection is an optional capability:

type Inspectable interface {
	Inspect() sup.Spec
}

Actor identity is ordinary application state:

type Worker struct {
	id string
}

func NewWorker(id string) *Worker {
	return &Worker{id: id}
}

func (w *Worker) ID() string {
	return w.id
}

For stateless actors, use ActorFunc:

worker := sup.ActorFunc("health", func(ctx context.Context) error {
	<-ctx.Done()
	return nil
})

Actor functions are inspectable by default with kind "actor". Use WithSpec to declare a more specific description:

poller := sup.ActorFunc(
	"temperature.poller",
	runPoller,
	sup.WithSpec(sup.Spec{
		Kind:         "poller",
		Dependencies: []string{"temperature"},
		Metadata: map[string]any{
			"interval": time.Second.String(),
		},
	}),
)

Lifecycle contract

Every actor follows the same Run contract:

• Context cancellation is a clean shutdown and returns nil.
• Intentional completion returns nil.
• A failure that may require a restart returns a non-nil error.
• A panic is recovered by the supervisor and treated as a failure.

Supervisor.Run follows the same contract, so supervisors can be nested as ordinary actors. It returns an error when a restart limit is exceeded or when the same supervisor is run concurrently.

Supervisors

Supervisors run actors and restart them according to a restart policy.

supervisor := sup.NewSupervisor("root",
	sup.WithPolicy(sup.Transient),
	sup.WithRestartDelay(500*time.Millisecond),
	sup.WithRestartLimit(3, time.Minute),
	sup.WithActors(actorA, actorB),
)

if err := supervisor.Run(ctx); err != nil {
	// A restart limit or concurrent Run stopped the supervisor.
}

Restart policies

Policy	Clean exit (nil)	Error or panic
Permanent	restart	restart
Transient	stop	restart
Temporary	stop	stop

Runtime events

An EventSink receives lifecycle events from a supervisor and all of its descendant supervisors. Different actors may call event sinks concurrently, so event sinks should return promptly and synchronize their own state. Sink panics are recovered and do not interrupt supervision.

events := sup.EventSinkFunc(func(event sup.Event) {
	fmt.Printf("%s %s\n", event.Type, event.Actor.ID())

	if event.Err != nil {
		fmt.Println("error:", event.Err)
	}
})

root := sup.NewSupervisor("root",
	sup.WithEventSink(events),
	sup.WithActors(worker),
)

Each event carries its occurrence time, actor, supervisor, error, and restart count as native Go values. Presentation layers such as hub decide how to serialize them. Multiple event sinks can independently record, log, aggregate, or forward the same authoritative runtime event.

Typed inboxes

CastInbox[T] is for asynchronous messages. CallInbox[T, R] is for request/reply interactions.

Cast inbox

inbox := sup.NewCastInbox[IncrementMessage](8)

err := inbox.Cast(ctx, IncrementMessage{Amount: 1})    // blocks until queued or ctx is done
err = inbox.TryCast(ctx, IncrementMessage{Amount: 1}) // returns ErrInboxFull if full

for {
	select {
	case <-ctx.Done():
		return
	case msg := <-inbox.Receive():
		// process msg
	}
}

Call inbox

inbox := sup.NewCallInbox[GetMessage, int](8)

value, err := inbox.Call(ctx, GetMessage{})

for {
	select {
	case <-ctx.Done():
		return
	case req := <-inbox.Receive():
		if !req.Reply(42, nil) {
			// The caller canceled or this request was already answered.
		}
	}
}

CallRequest.Context exposes the caller's context so work can stop after the caller leaves. Reply never blocks and returns whether the reply was accepted.

Their lifetime is controlled by the actor's context; inboxes are not closed independently.

rx

The github.com/webermarci/sup/rx package provides shared reactive state and composable channel helpers. It is a signal library, not an implementation of ReactiveX.

Signal

A Signal[V] is passive, concurrency-safe state with independent subscribers:

count := rx.NewSignal("count", 0)

values := count.Subscribe(ctx)

go func() {
	for value := range values {
		fmt.Println(value)
	}
}()

count.Set(1)
fmt.Println(count.Value())

Every subscription starts with the current value and has capacity one. Pending updates are coalesced to the latest value, so slow subscribers never block Set. Each call to Subscribe creates an independent stream, and its channel closes when the subscription context is canceled.

Every Set is a publication, even when it repeats the current value. Signals do not run under a supervisor; actors own behavior and set signals:

poller := sup.ActorFunc("random.poller", func(ctx context.Context) error {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			random.Set(rand.IntN(100))
		case <-ctx.Done():
			return nil
		}
	}
})

Channel helpers

Helpers accept and return ordinary channels, so they can be nested. They do not need contexts: cancellation closes the subscription channel, and that closure propagates through the pipeline.

processed := rx.Map(
	rx.Filter(
		rx.Debounce(count.Subscribe(ctx), 100*time.Millisecond),
		func(value int) bool { return value >= 10 },
	),
	func(value int) string { return strconv.Itoa(value) },
)

Available helpers:

• Map transforms values and may change their type.
• Filter drops values that do not match.
• Distinct drops consecutive equal comparable values.
• DistinctFunc uses a custom equality function.
• Debounce emits after a quiet period.
• ThrottleFirst immediately emits the first value in each interval.
• ThrottleLatest emits the latest value at the end of each interval.

Helper outputs have capacity one and coalesce pending values to the latest. They close when their input closes. Debounce and throttle durations must be positive.

Derived

Derived[V] is shared, read-only reactive state computed from one or more dependencies. Unlike Signal, it is an actor because it actively maintains its computed value:

count := rx.NewSignal("count", 0)
doubled := rx.NewDerived("doubled", func() int {
	return count.Value() * 2
}, count)

root := sup.NewSupervisor("root", sup.WithActors(doubled))
go root.Run(ctx)

Signals and derived values both provide ID, Value, Subscribe, and Watch. Only signals provide Set.

Dependency notifications already pending are coalesced. Updates to independent dependencies are not transactional, so a derived signal may publish intermediate results before converging on the latest state. Values that must change atomically should be stored together in one signal.

Construction computes the initial snapshot. When Run starts, it registers every dependency watch and publishes exactly one refreshed startup snapshot before processing later dependency notifications. Derived dependency graphs must be acyclic.

Hub

The github.com/webermarci/sup/hub package exposes explicitly registered actors and signals over HTTP. It reads signal values directly from their sources, streams changes and lifecycle events, and builds a pruned supervision graph from runtime registration events. Its HTML overview takes a fresh snapshot through the same HTTP API on each page load, then applies live signal and event updates from the SSE stream. It has no frontend dependencies.

enabled := rx.NewSignal("counter_enabled", true)

dashboard := hub.New("dashboard",
	hub.WithActor(counter),
	hub.WithSignal(counter.StateSignal),
	hub.WithWritableSignal(enabled),
	hub.WithEventHistoryLimit(128),
)

root := sup.NewSupervisor("root",
	sup.WithEventSink(dashboard),
	sup.WithActors(counter, dashboard),
)

go root.Run(ctx)
go http.ListenAndServe(":8080", dashboard.Handler())

The overview is served at /. The handler is prefix-agnostic, so applications can also mount it below a path with http.StripPrefix.

WithSignal is always read-only over HTTP. WithWritableSignal explicitly allows outside clients to change the signal with PATCH. Actors can subscribe to that signal using the ordinary rx API; the hub does not define a separate command or control model. Writable signals can also be changed with the forms in the HTML overview, which call the same PATCH endpoint without reloading the page.

PATCH /signals/counter_enabled
Content-Type: application/json

{"value": false}

Only explicitly registered actors have their lifecycle events exposed. The graph includes the supervisors needed to connect those actors to the root and omits unrelated internal branches. Recent event history is globally bounded; live event streaming is unaffected by the history limit.

Endpoints include:

• GET /
• GET /actors
• GET /actors/{actorID}
• GET /signals
• GET /signals/{signalID}
• PATCH /signals/{signalID}
• GET /graph
• GET /events
• GET /events/stream

Run the traffic-light example for a compact tour of the whole stack:

go run ./examples/dashboard

It combines a typed inbox, nested supervision, writable controls, derived state, and live SSE updates. Change cycle_length_seconds to speed up or slow down the intersection. Set paused to true to put the traffic light into a stationary yellow state and turn the pedestrian signal off.

Packages

• sup — Core actors, supervisors, typed inboxes, and runtime events.
• rx — Reactive signals, derived state, and channel helpers.
• sup/hub — HTTP API for actors, signals, supervision, and events.
• sup/process — Actor adapter for os/exec commands.
• sup/mesh — NATS-backed actor for subscriptions.
• sup/modbus — Modbus actor for TCP/RTU/ASCII clients.
• sup/mqtt — MQTT actor for publish/subscribe clients.
• sup/sse — Server-Sent Events client actor.
• sup/ws — WebSocket client actor.

Documentation

Overview

Package sup provides actors, supervision, runtime events, and typed message inboxes for building concurrent systems.

Index

• Variables
• type Actor
  • func ActorFunc(id string, fn func(ctx context.Context) error, options ...ActorFuncOption) Actor
• type ActorFuncOption
  • func WithSpec(spec Spec) ActorFuncOption
• type CallInbox
  • func NewCallInbox[T any, R any](size int) *CallInbox[T, R]
  • func (i *CallInbox[T, R]) Call(ctx context.Context, message T) (R, error)
  • func (i CallInbox) Receive() <-chan T
  • func (i *CallInbox[T, R]) TryCall(ctx context.Context, message T) (R, error)
• type CallRequest
  • func (r CallRequest[T, R]) Context() context.Context
  • func (r CallRequest[T, R]) Payload() T
  • func (r CallRequest[T, R]) Reply(value R, err error) bool
• type CastInbox
  • func NewCastInbox[T any](size int) *CastInbox[T]
  • func (i *CastInbox[T]) Cast(ctx context.Context, message T) error
  • func (i CastInbox) Receive() <-chan T
  • func (i *CastInbox[T]) TryCast(ctx context.Context, message T) error
• type Event
• type EventSink
• type EventSinkFunc
  • func (f EventSinkFunc) HandleEvent(event Event)
• type EventType
• type Inspectable
• type RestartPolicy
• type Spec
• type Supervisor
  • func NewSupervisor(id string, options ...SupervisorOption) *Supervisor
  • func (s *Supervisor) ID() string
  • func (s *Supervisor) Inspect() Spec
  • func (s *Supervisor) Run(ctx context.Context) error
• type SupervisorOption
  • func WithActors(actors ...Actor) SupervisorOption
  • func WithEventSink(sinks ...EventSink) SupervisorOption
  • func WithPolicy(policy RestartPolicy) SupervisorOption
  • func WithRestartDelay(delay time.Duration) SupervisorOption
  • func WithRestartLimit(maxRestarts int, window time.Duration) SupervisorOption

Variables

var ErrInboxFull = errors.New("sup: inbox is full")

ErrInboxFull is returned when a non-blocking send cannot be queued.

var ErrSupervisorRunning = errors.New("sup: supervisor is already running")

ErrSupervisorRunning is returned when Run is called while the supervisor is already running.

Types

type Actor

type Actor interface {
	// ID returns the actor id.
	ID() string

	// Run starts the actor and blocks until it stops or fails. Context
	// cancellation is a clean shutdown and must return nil.
	Run(context.Context) error
}

Actor represents a concurrent entity that can be supervised.

Run must return nil when its context is canceled or the actor completes intentionally. It should return a non-nil error only for failures that may require a restart. Panics are recovered by Supervisor and treated as failures.

func ActorFunc

func ActorFunc(
	id string,
	fn func(ctx context.Context) error,
	options ...ActorFuncOption,
) Actor

ActorFunc creates a simple stateless actor from a function.

type ActorFuncOption

type ActorFuncOption func(*actorFunc)

ActorFuncOption configures an actor created by ActorFunc.

func WithSpec

func WithSpec(spec Spec) ActorFuncOption

WithSpec configures an actor function's structured inspection spec.

type CallInbox

type CallInbox[T any, R any] struct {
	// contains filtered or unexported fields
}

CallInbox queues synchronous requests and delivers replies to callers.

func NewCallInbox

func NewCallInbox[T any, R any](size int) *CallInbox[T, R]

NewCallInbox creates a call inbox with the given buffer size.

func (*CallInbox[T, R]) Call

func (i *CallInbox[T, R]) Call(ctx context.Context, message T) (R, error)

Call sends a request and waits for a reply or context cancellation.

func (CallInbox) Receive

func (i CallInbox) Receive() <-chan T

Receive returns the read-only queue channel.

func (*CallInbox[T, R]) TryCall

func (i *CallInbox[T, R]) TryCall(ctx context.Context, message T) (R, error)

TryCall attempts to queue a request without blocking on inbox capacity. If the request is queued, it still waits for a reply or context cancellation.

type CallRequest

type CallRequest[T any, R any] struct {
	// contains filtered or unexported fields
}

CallRequest wraps a payload with a reply channel for synchronous calls.

func (CallRequest[T, R]) Context

func (r CallRequest[T, R]) Context() context.Context

Context returns the caller's context.

func (CallRequest[T, R]) Payload

func (r CallRequest[T, R]) Payload() T

Payload returns the request's payload.

func (CallRequest[T, R]) Reply

func (r CallRequest[T, R]) Reply(value R, err error) bool

Reply sends the response back to the caller. It returns false when the caller has canceled, the request is invalid, or a reply was already sent.

type CastInbox

type CastInbox[T any] struct {
	// contains filtered or unexported fields
}

CastInbox queues asynchronous messages for an actor.

func NewCastInbox

func NewCastInbox[T any](size int) *CastInbox[T]

NewCastInbox creates a cast inbox with the given buffer size.

func (*CastInbox[T]) Cast

func (i *CastInbox[T]) Cast(ctx context.Context, message T) error

Cast sends a message, blocking until it is queued or the context is canceled.

func (CastInbox) Receive

func (i CastInbox) Receive() <-chan T

Receive returns the read-only queue channel.

func (*CastInbox[T]) TryCast

func (i *CastInbox[T]) TryCast(ctx context.Context, message T) error

TryCast sends a message without blocking when the inbox is full.

type Event

type Event struct {
	Type         EventType
	Time         time.Time
	Actor        Actor
	Supervisor   *Supervisor
	Err          error
	RestartCount int
}

Event describes an actor lifecycle transition emitted by a supervisor.

type EventSink

type EventSink interface {
	HandleEvent(Event)
}

EventSink consumes runtime events emitted by a supervisor tree. HandleEvent may be called concurrently by different actors and must return promptly. A panic is recovered so that event handling cannot interrupt supervision.

type EventSinkFunc

type EventSinkFunc func(Event)

EventSinkFunc adapts a function into an EventSink.

func (EventSinkFunc) HandleEvent

func (f EventSinkFunc) HandleEvent(event Event)

HandleEvent calls f with event.

type EventType

type EventType string

EventType identifies the kind of event that occurred.

const (
	// EventActorRegistered is emitted when an actor is registered with a supervisor.
	EventActorRegistered EventType = "actor:registered"

	// EventActorStarted is emitted when an actor starts running.
	EventActorStarted EventType = "actor:started"

	// EventActorStopped is emitted when an actor stops running.
	EventActorStopped EventType = "actor:stopped"

	// EventActorRestarting is emitted before a supervisor restarts an actor.
	EventActorRestarting EventType = "actor:restarting"
)

type Inspectable

type Inspectable interface {
	Inspect() Spec
}

Inspectable is implemented by actors that expose structured metadata. Inspection is optional and does not affect actor execution or supervision.

type RestartPolicy

type RestartPolicy uint8

RestartPolicy controls when a supervisor restarts a stopped actor.

const (
	// Permanent always restarts actors, even after clean exits.
	Permanent RestartPolicy = iota

	// Transient restarts actors after errors or panics, but not after clean exits.
	Transient

	// Temporary never restarts actors.
	Temporary
)

type Spec

type Spec struct {
	Kind         string         `json:"kind"`
	Dependencies []string       `json:"dependencies"`
	Metadata     map[string]any `json:"metadata"`
}

Spec represents a structured description of an inspectable component. It captures the component's kind, its dependency names, and arbitrary configuration metadata for visualization and debugging purposes.

type Supervisor

type Supervisor struct {
	// contains filtered or unexported fields
}

Supervisor runs a fixed set of actors and restarts them according to its restart policy. A supervisor is itself an Actor and may be nested.

func NewSupervisor

func NewSupervisor(id string, options ...SupervisorOption) *Supervisor

NewSupervisor creates a supervisor with the given id and options.

func (*Supervisor) ID

func (s *Supervisor) ID() string

ID returns the supervisor id.

func (*Supervisor) Inspect

func (s *Supervisor) Inspect() Spec

Inspect returns the supervisor spec.

func (*Supervisor) Run

func (s *Supervisor) Run(ctx context.Context) error

Run starts all configured actors and blocks until they have all stopped, a restart limit is exceeded, or ctx is canceled. A supervisor may be run again after Run returns. Concurrent calls return ErrSupervisorRunning.

type SupervisorOption

type SupervisorOption func(*Supervisor)

SupervisorOption configures a Supervisor during construction.

func WithActors

func WithActors(actors ...Actor) SupervisorOption

WithActors adds actors that start with the supervisor.

func WithEventSink

func WithEventSink(sinks ...EventSink) SupervisorOption

WithEventSink adds an event sink for runtime events from this supervisor and its descendant supervisors.

func WithPolicy

func WithPolicy(policy RestartPolicy) SupervisorOption

WithPolicy configures the supervisor restart policy.

func WithRestartDelay

func WithRestartDelay(delay time.Duration) SupervisorOption

WithRestartDelay configures the exact delay between actor restart attempts.

func WithRestartLimit

func WithRestartLimit(maxRestarts int, window time.Duration) SupervisorOption

WithRestartLimit limits restarts within a time window.