gorch

package
v0.3.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 27, 2026 License: MIT Imports: 15 Imported by: 0

Documentation

Overview

Package gorch is a composable Go orchestrator for managing goroutine lifecycles. It handles start, stop, cron scheduling, and inter-service pub-sub messaging with self-healing restart. Orchestrators can be nested — a service may create and manage its own gorch instance for sub-services.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrAlreadyStarted  = errors.New("gorch: orchestrator already started")
	ErrInvalidCron     = errors.New("gorch: invalid cron expression")
	ErrStopTimeout     = errors.New("gorch: stop timed out waiting for services")
	ErrDuplicateName   = errors.New("gorch: duplicate service name")
	ErrDependencyCycle = errors.New("gorch: dependency cycle detected")
	ErrStartAborted    = errors.New("gorch: start aborted due to dependency failure")
)

Sentinel errors

Functions

func RegisterType added in v0.2.0 

func RegisterType[T any](m *Messenger) error

RegisterType registers T with encoding/gob so it can be used with TypedPublish and TypedSubscribe. Must be called before any typed operations for the type. Recovers from gob panics and returns an error if the type is not gob-compatible. Thread-safe. ponytail: standalone func (not method) because Go does not support generic methods on non-generic types.

func TypedPublish added in v0.2.0 

func TypedPublish[T any](m *Messenger, msg T, topics ...string)

TypedPublish gob-encodes msg and publishes it as a Message to the given topics. Silently drops the message if the type has not been registered via RegisterType. Thread-safe.

func TypedRequest added in v0.3.0 

func TypedRequest[TReq, TResp any](m *Messenger, ctx context.Context, req TReq, topic string) (TResp, error)

TypedRequest sends a typed request and waits for a typed response. It gob-encodes the request, publishes it via Request, and gob-decodes the response. Returns the decoded response or an error.

func TypedSubscribe added in v0.2.0 

func TypedSubscribe[T any](m *Messenger, topic string) (<-chan T, func())

TypedSubscribe subscribes to topic and returns a typed receive-only channel and an unsubscribe function. Messages published via TypedPublish are gob-decoded into T before delivery. Non-Message values and unrecognized types are silently dropped. Thread-safe.

Types

type Backoff added in v0.2.0 

type Backoff interface {
	Next(retry int) time.Duration
}

Backoff computes the delay before the next retry attempt. retry is 1-based (first retry = 1).

type Config 

type Config struct {
	LogLevel LogLevel // defaults to LogLevelInfo if zero

	// DefaultStartTimeout is the default per-service start deadline.
	// 0 means no timeout (use WithStartTimeout per-service).
	DefaultStartTimeout time.Duration

	// Health check configuration.
	// HealthInterval: how often to probe. Default: 30s.
	// HealthTimeout: per-probe deadline. Default: 5s.
	// HealthThreshold: consecutive failures before restart. Default: 3.
	// A zero HealthInterval disables health checks entirely.
	HealthInterval  time.Duration
	HealthTimeout   time.Duration
	HealthThreshold int

	// Global lifecycle hooks (called for every service unless overridden).
	OnBeforeStart func(name string) error
	OnAfterStart  func(name string, err error)
	OnBeforeStop  func(name string) error
	OnAfterStop   func(name string, err error)

	// State-change callbacks.
	OnStateChange func(name string, from, to ServiceStatus)
	OnCrash       func(name string, err error)

	// Health check hooks.
	BeforeHealthCheck func(name string) error
	AfterHealthCheck  func(name string, err error)
}

Config holds orchestrator configuration.

type ConstantBackoff added in v0.2.0 

type ConstantBackoff struct {
	Delay time.Duration
}

ConstantBackoff always returns the same delay regardless of retry count.

func (ConstantBackoff) Next added in v0.2.0 

func (b ConstantBackoff) Next(retry int) time.Duration

Next returns Delay (ignores retry count).

type CronMode 

type CronMode int

CronMode — concurrency policy when cron fires while previous invocation still runs. 

const (
	CronParallel CronMode = iota // fire in new goroutine regardless
	CronQueue                    // serialize: wait for previous to finish
	CronSkip                     // drop this tick entirely
)

type ExponentialBackoff added in v0.2.0 

type ExponentialBackoff struct {
	Initial time.Duration
	Max     time.Duration
	Factor  float64
}

ExponentialBackoff produces delays: initial * factor^(retry-1), capped at max. ponytail: no jitter; add WithJitter(bool) if thundering-herd becomes a problem.

func (ExponentialBackoff) Next added in v0.2.0 

func (b ExponentialBackoff) Next(retry int) time.Duration

Next returns initial * factor^(retry-1), capped at Max.

type HealthChecker added in v0.2.0 

type HealthChecker interface {
	Health(ctx context.Context) error
}

HealthChecker is implemented by services that can report their own health. Health is called periodically by the orchestrator. A non-nil error means the service is unhealthy.

type LogLevel 

type LogLevel int

LogLevel controls minimum log severity output by the log-pump. 

const (
	LogLevelDebug LogLevel = iota
	LogLevelInfo
	LogLevelWarn
	LogLevelError
)

func (LogLevel) String 

func (l LogLevel) String() string

type Message added in v0.2.0 

type Message struct {
	Payload    []byte
	Topic      string
	ReplyTopic string
	TypeName   string
}

Message is the envelope for typed pub-sub and request-reply messaging. Publishers encode their payload into Payload; subscribers decode it. ReplyTopic is set automatically by Request/RequestAsync so responders know where to send the reply.

type Messenger 

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

Messenger — pub-sub with topics (Socket.IO rooms style). A nil or empty topics slice in Publish broadcasts to ALL subscribers.

func (*Messenger) Drain added in v0.3.0 

func (m *Messenger) Drain()

Drain closes all subscriber channels and clears all subscriptions. After Drain, the Messenger is empty and no new publishes will be received by prior subscribers. Thread-safe.

func (*Messenger) Publish 

func (m *Messenger) Publish(msg any, topics ...string)

Publish sends msg to subscribers. Non-blocking: if a subscriber's channel is full, the message is dropped for that subscriber. Thread-safe. If topics is empty or nil, broadcasts to ALL subscribers on every topic.

func (*Messenger) Request added in v0.2.0 

func (m *Messenger) Request(ctx context.Context, msg any, topic string) (any, error)

Request publishes a request message and waits for a single reply. It creates a temporary reply topic, subscribes to it, publishes the request, and returns the first response (or an error if ctx expires). The responding service receives a Message on its channel; it should Publish the response on msg.ReplyTopic. Thread-safe.

func (*Messenger) RequestAsync added in v0.2.0 

func (m *Messenger) RequestAsync(ctx context.Context, msg any, topic string) (<-chan any, error)

RequestAsync is like Request but returns immediately with a response channel. The caller must select on the channel and ctx.Done(). Thread-safe.

func (*Messenger) Subscribe 

func (m *Messenger) Subscribe(topic string) (<-chan any, func())

Subscribe registers interest in a topic. Returns a receive-only channel and an unsubscribe function. The channel is buffered (cap 16). Thread-safe.

func (*Messenger) SubscribeWithBuffer added in v0.3.0 

func (m *Messenger) SubscribeWithBuffer(topic string, bufSize int) (<-chan any, func())

SubscribeWithBuffer registers interest in a topic with a caller-specified buffer size. Returns a receive-only channel and an unsubscribe function. Thread-safe.

type Metrics added in v0.3.0 

type Metrics struct {
	Starts      int64
	Stops       int64
	Crashes     int64
	Restarts    int64
	HealthFails int64
}

Metrics holds counter snapshots for orchestrator-level events.

type Orchestrator 

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

Orchestrator manages service lifecycles.

func New 

func New(cfg Config) *Orchestrator

New creates a new Orchestrator. Each call returns a fresh, independent instance. Orchestrators can be nested: a service may create its own gorch to manage sub-services.

func (*Orchestrator) Count added in v0.2.0 

func (o *Orchestrator) Count() int

Count returns the total number of registered services. Thread-safe.

func (*Orchestrator) Done added in v0.3.0 

func (o *Orchestrator) Done() <-chan struct{}

Done returns a channel that closes when all managed goroutines (services, log-pump, health-check loop) have exited. The orchestrator must be stopped (via Stop or Run returning) before the channel closes.

func (*Orchestrator) Health added in v0.2.0 

func (o *Orchestrator) Health() map[string]error

Health probes all registered services that implement HealthChecker. Returns a map of service name to error (nil = healthy). Services that don't implement HealthChecker are reported as nil. Thread-safe.

func (*Orchestrator) IsReady added in v0.3.0 

func (o *Orchestrator) IsReady(name string) bool

IsReady reports whether a named service is running and ready to serve.

func (*Orchestrator) Metrics added in v0.3.0 

func (o *Orchestrator) Metrics() Metrics

Metrics returns a snapshot of orchestrator-level event counters.

func (*Orchestrator) Names added in v0.2.0 

func (o *Orchestrator) Names() []string

Names returns the names of all registered services in registration order. Thread-safe.

func (*Orchestrator) Register 

func (o *Orchestrator) Register(svc Service, opts ...RegisterOption) error

Register adds a service to the orchestrator. Must be called before Start(). Returns ErrAlreadyStarted if the orchestrator has already been started. Returns ErrDuplicateName if WithName conflicts with another service. Returns ErrDependencyCycle if DependsOn introduces a cycle. Thread-safe.

func (*Orchestrator) RegisterFunc added in v0.3.0 

func (o *Orchestrator) RegisterFunc(name string, startFn func(ctx ServiceContext) error, stopFn func() error, opts ...RegisterOption) error

RegisterFunc registers a closure-based service.

func (*Orchestrator) Run added in v0.2.0 

func (o *Orchestrator) Run(stopTimeout time.Duration, signals ...os.Signal) error

Run starts the orchestrator, blocks on SIGINT/SIGTERM, then stops. Returns any error from Start or aggregated errors from Stop. Optional signals override the default signal set.

func (*Orchestrator) Start 

func (o *Orchestrator) Start() error

Start begins the orchestrator lifecycle. Returns ErrAlreadyStarted if already started. Idempotent: calling Start multiple times returns nil after the first. Thread-safe.

func (*Orchestrator) StartGroup added in v0.3.0 

func (o *Orchestrator) StartGroup(group string) error

StartGroup starts all services in the named group in topological order.

func (*Orchestrator) Status added in v0.2.0 

func (o *Orchestrator) Status(name string) (ServiceStatus, bool)

Status returns the current lifecycle status of a named service. ok is false if no service with that name is registered. Thread-safe.

func (*Orchestrator) Statuses added in v0.2.0 

func (o *Orchestrator) Statuses() map[string]ServiceStatus

Statuses returns a map of service name to status for all registered services. Thread-safe.

func (*Orchestrator) StatusesByGroup added in v0.3.0 

func (o *Orchestrator) StatusesByGroup(group string) map[string]ServiceStatus

StatusesByGroup returns a map of service name to status for all services in the named group. Thread-safe.

func (*Orchestrator) StatusesByLabel added in v0.3.0 

func (o *Orchestrator) StatusesByLabel(key, value string) map[string]ServiceStatus

StatusesByLabel returns a map of service name to status for all services matching the given label key-value pair. Thread-safe.

func (*Orchestrator) Stop 

func (o *Orchestrator) Stop(timeout time.Duration) error

Stop gracefully shuts down the orchestrator. Waits up to timeout for services to finish. Returns aggregated errors from all Stop failures, or ErrStopTimeout if services don't all stop within the timeout. Thread-safe. Safe to call on an orchestrator that was never started.

func (*Orchestrator) StopGroup added in v0.3.0 

func (o *Orchestrator) StopGroup(group string, timeout time.Duration) error

StopGroup stops all non-cron, non-runOnce services in the named group in reverse topological order. Errors are aggregated via errors.Join.

func (*Orchestrator) WaitFor added in v0.3.0 

func (o *Orchestrator) WaitFor(name string, target ServiceStatus, timeout time.Duration) error

WaitFor blocks until the named service reaches target status or timeout expires. Polls at 50ms intervals. Returns an error on timeout or if the service is not found.

type ReadinessChecker added in v0.3.0 

type ReadinessChecker interface {
	Ready(ctx context.Context) error
}

ReadinessChecker is implemented by services that distinguish "running" from "ready to serve". Ready returns nil when the service can accept traffic.

type RegisterOption 

type RegisterOption func(*registerConfig)

RegisterOption — functional options for Register.

func DependsOn added in v0.2.0 

func DependsOn(names ...string) RegisterOption

DependsOn declares that this service must start after the named services and stop before them. Cycles are detected at registration time.

func DependsOnSoft added in v0.3.0 

func DependsOnSoft(names ...string) RegisterOption

DependsOnSoft declares soft dependencies: start after the named services if they are present, but ignore any that are not registered.

func WithBackoff added in v0.2.0 

func WithBackoff(b Backoff) RegisterOption

WithBackoff sets the backoff strategy for self-heal restarts. If nil or not set, the default is 1s constant backoff.

func WithCron 

func WithCron(spec string, mode CronMode) RegisterOption

WithCron registers the service to run on a 6-field cron schedule (seconds included).

func WithGroup added in v0.3.0 

func WithGroup(name string) RegisterOption

WithGroup assigns the service to a named group for filtering.

func WithLabel added in v0.3.0 

func WithLabel(key, value string) RegisterOption

WithLabel attaches a key-value label to the service for filtering.

func WithMaxRetries added in v0.2.0 

func WithMaxRetries(max int) RegisterOption

WithMaxRetries sets the maximum number of self-heal restarts. 0 means unlimited (up to context cancellation). After the limit is reached, the service transitions to StatusStopped.

func WithName added in v0.2.0 

func WithName(name string) RegisterOption

WithName assigns a human-readable name used for dependency ordering, status queries, and lifecycle hooks. Names must be unique across all registered services.

func WithOnAfterStart added in v0.2.0 

func WithOnAfterStart(fn func(name string, err error)) RegisterOption

WithOnAfterStart sets a per-service hook called after Start() returns.

func WithOnAfterStop added in v0.2.0 

func WithOnAfterStop(fn func(name string, err error)) RegisterOption

WithOnAfterStop sets a per-service hook called after Stop() returns.

func WithOnBeforeStart added in v0.2.0 

func WithOnBeforeStart(fn func(name string) error) RegisterOption

WithOnBeforeStart sets a per-service hook called just before Start(). If the hook returns an error, Start() is aborted for this service.

func WithOnBeforeStop added in v0.2.0 

func WithOnBeforeStop(fn func(name string) error) RegisterOption

WithOnBeforeStop sets a per-service hook called just before Stop(). If the hook returns an error, Stop() is still called.

func WithResetAfter added in v0.2.0 

func WithResetAfter(d time.Duration) RegisterOption

WithResetAfter sets a stability window. If the service runs continuously for this duration without crashing, the retry counter resets to zero.

func WithRunOnce added in v0.2.0 

func WithRunOnce() RegisterOption

WithRunOnce marks a service as a one-shot init task. It runs before persistent services, never receives Stop(), and transitions to StatusStopped when Start returns. If Start returns an error, startup aborts.

func WithSelfHeal 

func WithSelfHeal(factory func() Service) RegisterOption

WithSelfHeal enables auto-restart: when the service crashes (returns error or panics), the orchestrator calls factory() for a fresh instance and restarts it.

func WithStartCondition added in v0.3.0 

func WithStartCondition(fn func() bool) RegisterOption

WithStartCondition sets a function called at startup. If it returns false, the service is skipped (not started). nil or not set means always start.

func WithStartTimeout added in v0.2.0 

func WithStartTimeout(d time.Duration) RegisterOption

WithStartTimeout sets the maximum time to wait for this service's Start to return. Overrides Config.DefaultStartTimeout. A zero duration means no timeout (use with caution).

func WithStopTimeout added in v0.3.0 

func WithStopTimeout(d time.Duration) RegisterOption

WithStopTimeout sets a per-service timeout on Stop(). If Stop() does not return within this duration, the orchestrator proceeds with shutdown.

type Service 

type Service interface {
	Start(ctx context.Context) error // blocks; for cron: runs per-tick; for non-cron: runs until ctx cancelled
	Stop() error                     // cleanup signal beyond context cancellation
}

Service — every managed goroutine implements this.

type ServiceContext 

type ServiceContext struct {
	context.Context
	Logger    *ServiceLogger
	Messenger *Messenger
}

ServiceContext — what the orchestrator hands each service. Embeds context.Context so it satisfies the context.Context interface and can be passed directly to Service.Start. Carries the orchestrator's cancellation context.

type ServiceLogger 

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

ServiceLogger — a logger that doesn't log; it sends entries to gorch's log channel. gorch consumes the channel and does the actual output (formatting, writing to stderr).

func (*ServiceLogger) Debug 

func (l *ServiceLogger) Debug(msg string, args ...any)

func (*ServiceLogger) Error 

func (l *ServiceLogger) Error(msg string, args ...any)

func (*ServiceLogger) Info 

func (l *ServiceLogger) Info(msg string, args ...any)

func (*ServiceLogger) Warn 

func (l *ServiceLogger) Warn(msg string, args ...any)

type ServiceStatus added in v0.2.0 

type ServiceStatus int

ServiceStatus represents the lifecycle state of a registered service. 

const (
	StatusRegistered ServiceStatus = iota
	StatusStarting
	StatusRunning
	StatusStopping
	StatusStopped
	StatusCrashed
)

func (ServiceStatus) String added in v0.2.0 

func (s ServiceStatus) String() string

String returns a human-readable name for the status.

type Validator added in v0.3.0 

type Validator interface {
	Validate() error
}

Validator is implemented by services that validate their configuration at Register time. Validate is called immediately during Register; a non-nil error causes Register to return that error.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.