gorch

package module
v0.6.0 Latest Latest
Warning

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

 Go to latest Published: Sep 11, 2026 License: MIT Imports: 16 Imported by: 0

README

gorch — Go Orchestrator Library

Manage goroutine lifecycles — start, stop, cron scheduling, pub-sub messaging, dependency ordering, health checks, and self-healing — with a small, composable API.

import "github.com/lorenzo-vecchio/gorch"

Install

go get github.com/lorenzo-vecchio/gorch@latest

Requires Go 1.25+.

Features

  • Service lifecycle — Start/Stop with context cancellation and graceful shutdown.
  • Run() convenience — single call starts, blocks on OS signals, then stops.
  • Dependency ordering — declare dependencies with DependsOn, cycle detection at registration, topological start and reverse-topological stop.
  • Start timeout — per-service start deadline via WithStartTimeout, with a DefaultStartTimeout config default.
  • Cron scheduling — 6-field cron (seconds included) with three concurrency modes: Parallel, Queue, Skip.
  • Pub-sub Messenger — topic-based messaging between services (Socket.IO rooms style), non-blocking sends, request-reply, and typed messages.
  • Self-healing — auto-restart crashed services with a factory-provided fresh instance and configurable backoff/retry.
  • Health checks — HealthChecker interface; orchestrator probes services on configurable intervals, auto-restarts unhealthy services.
  • Backoff & retry — ExponentialBackoff and ConstantBackoff strategies, max retries, stability-window retry reset.
  • One-shot services — init/gate tasks that run once before persistent services; Stop() is called at shutdown.
  • Lifecycle hooks — OnBeforeStart, OnAfterStart, OnBeforeStop, OnAfterStop (global or per-service overrides).
  • Status introspection — Status, Statuses, Names, Count for runtime observability.
  • Error aggregation — errors.Join in Start/Stop so all failures are reported, not just the first.
  • Nestable orchestrators — a service can create its own gorch for sub-services.
  • Structured logging — channel-based log-pump writes to stderr; services call Info/Error/Debug/Warn on a ServiceLogger.
  • Custom logger — inject any logger satisfying the Logger interface (e.g. *slog.Logger); service name is prepended as a key-value pair to every call.
  • RegisterFunc — closure-based services for simple cases; no boilerplate struct needed.
  • Service groups — WithGroup, StartGroup, StopGroup, StatusesByGroup for operating on subsets.
  • Labels — WithLabel + StatusesByLabel for metadata filtering.
  • Soft dependencies — DependsOnSoft for optional service ordering; start after if present, no error if missing.
  • Readiness checks — ReadinessChecker interface + IsReady(); separate "alive" from "ready to serve."
  • State-change hooks — OnStateChange + OnCrash callbacks for external observability without polling.
  • WaitFor — Block until a service reaches a target status.
  • TypedRequest — Typed request-reply without losing type safety: TypedRequest[TReq, TResp](messenger, ctx, req, topic).
  • Metrics — atomic int64 counters (Starts, Stops, Crashes, Restarts, HealthFails), exposed via Metrics() snapshot.
  • Validator interface — Validate() error called at Register for early config checks.
  • WithStartCondition — Skip a service at runtime via a func() bool.
  • Per-service stop timeout — WithStopTimeout controls how long to wait for Stop().
  • Configurable channel buffer — SubscribeWithBuffer for the Messenger.
  • Health check hooks — BeforeHealthCheck / AfterHealthCheck for instrumenting probes.
  • Messenger.Drain — Gracefully close all subscriber channels and clear subscriptions.
  • Done() channel — Non-blocking shutdown notification; closes when all goroutines finish.

Concurrency

All methods are safe to call from multiple goroutines unless noted otherwise. The table below summarizes what may run concurrently with a live Start/Stop.

Method group Concurrent with Start/Stop
Register, RegisterFunc No — call before Start. During or after the lifecycle they return ErrAlreadyStarted.
Start, Stop Yes — against each other. Guarded by sync.Once; the orchestrator is single-shot, so after a successful Stop neither can run again.
Status, Statuses, Names, Count Yes — safe to read while services run and during shutdown.
Health, IsReady, WaitFor Yes — each probe/tick takes its own read lock; IsReady honors the caller's ctx.
Metrics, Done Yes — atomic counters and a lazily cached channel.
StartGroup, StopGroup Not synchronized with Start/Stop; drive one lifecycle per orchestrator.
Messenger (Subscribe, Publish, Request, RequestAsync, Drain) and the typed helpers Yes — all Messenger methods are safe for concurrent use.

Service implementations are responsible for their own internal concurrency: Start runs in its own goroutine and Stop may be called from another after context cancellation.

Contract

These guarantees are part of the public API and are relied upon by callers.

  • Wire format is gob. encoding/gob is the serialization format for typed messages and request-reply payloads. It is public contract: types passed through TypedPublish/TypedSubscribe/TypedRequest/TypedRespond must be gob-compatible, and changing a field layout is a breaking change.
  • Publish is drop-only. Delivery is best-effort. When a subscriber's channel is full the message is dropped for that subscriber; gorch never blocks the publisher and never replays dropped messages.
  • Lifecycle is single-shot. After a successful Stop, neither Start nor Register can be used again; both return ErrAlreadyStarted. A failed Start does not consume the lifecycle and may be retried.
  • Errors are aggregated. Start and Stop join every failure with errors.Join, so a single call reports all causes, not just the first. Use errors.Is/errors.As to inspect them.

Sentinel errors returned by the orchestrator:

Error Returned by Meaning
ErrAlreadyStarted Register, Start Called after the orchestrator already started (or was stopped).
ErrDuplicateName Register Two services share a WithName.
ErrDependencyCycle Register, Start A hard or soft dependency chain loops.
ErrStartAborted Start A hard/soft dependency failed or was skipped.
ErrInvalidCron Start A WithCron spec is invalid.
ErrUnsupportedOption Register WithSelfHeal combined with WithCron/WithRunOnce.
ErrStopTimeout Stop Services did not stop within the caller's timeout.

Quick start

package main

import (
    "context"
    "time"

    "github.com/lorenzo-vecchio/gorch"
)

type MyService struct{}

func (s *MyService) Start(ctx gorch.ServiceContext) error {
    <-ctx.Done()
    return nil
}

func (s *MyService) Stop() error { return nil }

func main() {
    orch := gorch.New(gorch.WithLogLevel(gorch.LogLevelInfo))
    orch.Register(&MyService{})

    // Blocks until SIGINT/SIGTERM, then stops gracefully.
    if err := orch.Run(10 * time.Second); err != nil {
        panic(err)
    }
}

API

Service interface
type Service interface {
    Start(ctx gorch.ServiceContext) error
    Stop() error
}

ServiceContext (the ctx passed to Start) embeds context.Context and carries a *ServiceLogger and *Messenger. Existing <-ctx.Done() / ctx.Err() bodies keep working unchanged.

Orchestrator
orch := gorch.New(
    gorch.WithLogLevel(gorch.LogLevelInfo),
    gorch.WithDefaultStartTimeout(5 * time.Second),
)
orch.Register(svc, gorch.WithCron("@every 5s", gorch.CronSkip))
orch.Register(svc, gorch.WithSelfHeal(func() gorch.Service { return &MyService{} }))
orch.Start()
orch.Stop(10 * time.Second)

Configuration uses functional options. New() with no options uses the defaults (Info log level, health checks every 30s with a 5s probe timeout).

Run() convenience

Run starts the orchestrator, blocks until a signal is received (SIGINT and SIGTERM by default, configurable via variadic signals), then stops.

// Default: waits for SIGINT or SIGTERM.
orch.Run(10 * time.Second)

// Custom signals.
orch.Run(10 * time.Second, syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)
Dependency ordering

Services declare names and dependencies via WithName and DependsOn. Cycles are detected at Register time. Services start in topological order (independent services in parallel within each level) and stop in reverse topological order.

orch.Register(dbSvc,   gorch.WithName("db"))
orch.Register(cacheSvc, gorch.WithName("cache"))
orch.Register(apiSvc,  gorch.WithName("api"), gorch.DependsOn("db", "cache"))
// Start: (db, cache) in parallel → api. Stop: api → (cache, db).
Start timeout

Per-service start deadline, with a config-level default.

orch := gorch.New(gorch.WithDefaultStartTimeout(5 * time.Second))
orch.Register(svc, gorch.WithStartTimeout(30 * time.Second)) // per-service override
Cron modes
Mode Behavior
CronParallel Fire every tick, overlapping runs allowed.
CronQueue Serialize — wait for the previous run to finish.
CronSkip Drop ticks that would overlap.
Status introspection
status, ok := orch.Status("db")            // ServiceStatus, bool
all := orch.Statuses()                     // map[string]ServiceStatus
names := orch.Names()                      // []string in registration order
count := orch.Count()                      // total registered services

ServiceStatus values: StatusRegistered, StatusStarting, StatusRunning, StatusStopping, StatusStopped, StatusCrashed, StatusSucceeded. Each has a String() method. StatusSucceeded marks a one-shot service whose Start completed without error (a successful gate); dependents are not aborted by it.

One-shot / init services

WithRunOnce marks a service as a one-shot init task. It runs before persistent services and transitions to StatusSucceeded when Start returns. Stop() is called at orchestrator shutdown (make it idempotent). If Start returns an error, startup aborts.

orch.Register(migrator, gorch.WithRunOnce())
Lifecycle hooks

Global hooks are set via functional options, or per-service via RegisterOption.

orch := gorch.New(
    gorch.WithGlobalOnBeforeStart(func(name string) error {
        log.Printf("starting %s", name)
        return nil
    }),
    gorch.WithGlobalOnAfterStop(func(name string, err error) {
        log.Printf("stopped %s, err=%v", name, err)
    }),
)

// Per-service override:
orch.Register(svc, gorch.WithOnBeforeStart(func(name string) error {
    return checkPrerequisites()
}))
Self-healing with backoff & retry

Self-heal restarts crashed services with backoff, retry limits, and a stability window.

orch.Register(svc,
    gorch.WithSelfHeal(func() gorch.Service { return &MyService{} }),
    gorch.WithBackoff(gorch.ExponentialBackoff{
        Initial: 1 * time.Second,
        Max:     30 * time.Second,
        Factor:  2.0,
    }),
    gorch.WithMaxRetries(5),              // give up after 5 retries (0 = unlimited)
    gorch.WithResetAfter(2 * time.Minute),  // reset retry count if service runs this long
)

ConstantBackoff returns the same delay every time.

gorch.WithBackoff(gorch.ConstantBackoff{Delay: 3 * time.Second})
Health checks

Services implement HealthChecker to report their health. The orchestrator probes on a configurable interval. After HealthThreshold consecutive failures, a self-healing service is restarted.

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

orch := gorch.New(
    gorch.WithHealthChecks(30*time.Second,
        gorch.WithProbeTimeout(5*time.Second),
        gorch.WithFailureThreshold(3)),
    // interval, per-probe timeout, consecutive failures before restart
)

// Or disable the health-check loop entirely:
orch := gorch.New(gorch.WithHealthChecksDisabled())

Note: automatic restart on health failure requires WithSelfHeal (a factory to create a fresh instance). Without a factory, health failures are only logged and counted in Metrics().HealthFails.

Manual health check:

results := orch.Health() // map[string]error, nil = healthy
Messenger
ch, unsub := messenger.Subscribe("topic")
messenger.Publish(msg, "topic")  // send to topic subscribers
messenger.Publish(msg)           // broadcast to ALL subscribers
Request-reply

Request publishes a message and blocks until a response arrives (or ctx expires). The responder receives a Message with a ReplyTopic field and publishes its reply there. The payload is gob-encoded, so decoding it by hand is error-prone: implement the responder with the typed helpers (TypedSubscribeRequest/TypedRespond) instead.

// Requestor:
resp, err := messenger.Request(ctx, payload, "orders.create")

// Responder (inside a service goroutine): messages arrive already decoded,
// and the reply is encoded by TypedRespond — see "Typed Request-Reply".
reqCh, _ := gorch.TypedSubscribeRequest[CreateOrderReq](messenger, "orders.create")
for env := range reqCh {
    gorch.TypedRespond(messenger, processOrder(env.Value), env.ReplyTopic)
}

For a fully type-safe round trip (typed requestor included), use TypedRequest — described in "Typed Request-Reply" below. RequestAsync returns a response channel immediately without blocking.

Typed messages

RegisterType, TypedPublish, and TypedSubscribe provide gob-encoded type-safe messaging.

type OrderEvent struct {
    OrderID string
    Status  string
}

gorch.RegisterType[OrderEvent](messenger)

// Publisher:
gorch.TypedPublish(messenger, OrderEvent{OrderID: "42", Status: "shipped"}, "orders")

// Subscriber:
ch, unsub := gorch.TypedSubscribe[OrderEvent](messenger, "orders")
for evt := range ch {
    fmt.Println(evt.OrderID) // typed, no cast needed
}
Logging

Services log via ServiceLogger:

sc.Logger.Info("request completed", "status", 200, "latency", 12*time.Millisecond)
// 2026-07-27 14:30:05.123 INFO  my-service --- request completed status=200 latency=12ms
// (the prefix is the service name: WithName, or the auto-assigned $N)

The built-in log-pump writes to os.Stderr. Log level filters entries: Debug < Info < Warn < Error.

Custom logger

Inject any logger that satisfies the Logger interface via WithLogger. *slog.Logger from the standard library satisfies this interface directly.

import "log/slog"

orch := gorch.New(gorch.WithLogger(slog.Default()))

When a custom logger is set, the built-in log-pump is disabled entirely. The service name is prepended as "service"=<name> to every log call so the custom logger can include or exclude it as needed. WithLogLevel is ignored — the custom logger manages its own level filtering.

// With slog, the service name appears as a structured key-value pair:
// level=INFO msg="request completed" service=api status=200 latency=12ms
RegisterFunc

For simple services where a struct is boilerplate, RegisterFunc accepts closures directly.

orch.RegisterFunc("health-server", func(ctx gorch.ServiceContext) error {
    srv := &http.Server{Addr: ":8080"}
    go func() { <-ctx.Done(); srv.Shutdown(context.Background()) }()
    return srv.ListenAndServe()
}, nil) // nil Stop func — stops purely via context cancellation

A Stop func can be nil if the service cleans up via context cancellation alone.

Groups

Assign services to named groups with WithGroup, then operate on subsets.

orch.Register(dbSvc, gorch.WithName("db"), gorch.WithGroup("infra"))
orch.Register(cacheSvc, gorch.WithName("cache"), gorch.WithGroup("infra"))
orch.Register(apiSvc, gorch.WithName("api"), gorch.WithGroup("app"), gorch.DependsOn("db", "cache"))

// Start or stop only a group.
err := orch.StartGroup("infra")
err = orch.StopGroup("app", 5*time.Second)

// Filter statuses by group.
infra := orch.StatusesByGroup("infra") // map[string]ServiceStatus
Labels

Attach arbitrary key-value tags for filtering and introspection.

orch.Register(svc, gorch.WithLabel("tier", "critical"))
orch.Register(svc, gorch.WithLabel("team", "payments"))

critical := orch.StatusesByLabel("tier", "critical")
Soft dependencies

DependsOnSoft orders a service after its soft dependencies if they are registered, but does not fail if they are missing.

orch.Register(apiSvc,
    gorch.WithName("api"),
    gorch.DependsOn("db"),          // hard: must exist
    gorch.DependsOnSoft("metrics"), // soft: start after if present, ignore if missing
)
Readiness

ReadinessChecker separates "running" from "ready to serve." Use IsReady() to gate traffic routing without killing the service.

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

// On the orchestrator:
if orch.IsReady(ctx, "api") {
    // route traffic
}
State-change hooks

OnStateChange fires on every status transition. OnCrash fires specifically on Running -> Crashed. Wire these to Prometheus counters, Slack webhooks, or a status page instead of polling.

orch := gorch.New(
    gorch.WithOnStateChange(func(name string, from, to gorch.ServiceStatus) {
        log.Printf("%s: %s -> %s", name, from, to)
    }),
    gorch.WithOnCrash(func(name string, err error) {
        notifications.Send(name + " crashed")
    }),
)
WaitFor

Block until a service reaches a target status (or times out). Useful for tests and services that need external coordination.

err := orch.WaitFor("db", gorch.StatusRunning, 10*time.Second)
Typed Request-Reply

TypedRequest provides type-safe request-reply without falling back to the untyped Message API. Pair it with TypedSubscribeRequest (decodes requests into a TypedEnvelope carrying the value and ReplyTopic) and TypedRespond (encodes and publishes the reply).

type CreateOrderReq struct {
    ItemID string
    Qty    int
}
type CreateOrderResp struct {
    OrderID string
    Status  string
}

// Responder (inside a service goroutine):
reqCh, unsub := gorch.TypedSubscribeRequest[CreateOrderReq](messenger, "orders.create")
defer unsub()
go func() {
    for env := range reqCh {
        result := processOrder(env.Value)
        gorch.TypedRespond(messenger, result, env.ReplyTopic)
    }
}()

// Requestor:
resp, err := gorch.TypedRequest[CreateOrderReq, CreateOrderResp](
    messenger, ctx, req, "orders.create",
)

No manual gob encoding is required anywhere in user code.

Metrics

Metrics() returns a snapshot of atomic counters for orchestrator-level events. The user wires these into their own monitoring system — no metrics library dependency.

stats := orch.Metrics()
fmt.Printf("starts=%d stops=%d crashes=%d restarts=%d healthFails=%d\n",
    stats.Starts, stats.Stops, stats.Crashes, stats.Restarts, stats.HealthFails)
Validator

Implement the Validator interface to catch config errors at Register time (before Start).

type Validator interface {
    Validate() error
}

func (s *MyService) Validate() error {
    if s.Port == 0 {
        return fmt.Errorf("port must be set")
    }
    return nil
}

// Register returns the validation error immediately:
err := orch.Register(svc)
WithStartCondition

Skip a service at runtime without removing its registration. The condition function is evaluated just before startup.

orch.Register(svc, gorch.WithStartCondition(func() bool {
    return os.Getenv("FEATURE_ENABLED") == "true"
}))
Per-service StopTimeout

WithStopTimeout sets a per-service deadline on Stop(). The orchestrator proceeds with shutdown even if this service takes longer.

orch.Register(svc, gorch.WithStopTimeout(3 * time.Second))
Messenger buffer size

SubscribeWithBuffer lets callers set the buffer capacity to prevent slow consumers from blocking publishers.

ch, unsub := messenger.SubscribeWithBuffer("high-throughput", 256)
Health check hooks

BeforeHealthCheck and AfterHealthCheck provide instrumentation points around every health probe without wrapping every HealthChecker.

orch := gorch.New(
    gorch.WithHealthChecks(30*time.Second,
        gorch.WithProbeTimeout(5*time.Second),
        gorch.WithFailureThreshold(3)),
    gorch.WithBeforeHealthCheck(func(name string) error {
        metrics.Inc("health_checks_total")
        return nil
    }),
    gorch.WithAfterHealthCheck(func(name string, err error) {
        if err != nil {
            metrics.Inc("health_checks_failed")
        }
    }),
)
Drain and Done

Drain() closes all subscriber channels and clears subscriptions. Done() returns a channel that closes when all goroutines (services, log-pump, health-check loop) have exited — useful for non-blocking shutdown.

// Gracefully flush pending messages before shutdown.
messenger.Drain()

// Non-blocking wait for full shutdown.
select {
case <-orch.Done():
case <-time.After(10 * time.Second):
}

Compatibility & versioning

gorch follows Semantic Versioning. Until 1.0, minor releases may contain breaking changes; every one is marked Breaking in the CHANGELOG and covered by a migration note.

  • Stable — exported identifiers (types, functions, methods), the Service, HealthChecker, ReadinessChecker, Validator, and Logger interfaces, the ServiceStatus values, the sentinel errors above, and the gob wire format.
  • Not stable — the built-in logger's exact output format and key ordering, internal goroutine counts, and anything unexported. Do not parse log lines.
  • Deprecations — a deprecated exported identifier keeps working for at least one minor release and is listed as Deprecated: in its doc comment and in the CHANGELOG before removal.

See MIGRATION.md for upgrade steps between releases.

Examples

Development

go test . -coverprofile=coverage.out
go tool cover -func=coverage.out | grep total  # must be 100.0%
go test . -bench . -benchmem
go vet ./...
gofmt -w .

License

MIT

Documentation

Overview

Package gorch orchestrates the lifecycle of long-running goroutines.

It is a small, dependency-light runtime supervisor for a Go program that starts several cooperating services and must start, stop, and supervise them in a defined order.

What it does

  • Starts services in dependency order and stops them in reverse order.
  • Runs cron-scheduled ticks and one-shot init (runOnce) gates.
  • Self-heals services that crash, using a factory plus backoff and retry.
  • Probes health and readiness and can restart unhealthy services.
  • Carries a topic pub-sub Messenger between services.

Use case

gorch is for the composition root of a single process: the place where you wire a handful of long-lived components (an HTTP server, a worker pool, a cache refresher, migrations) and want deterministic startup, supervision, and graceful shutdown without adopting a framework.

What it is not

gorch is not a scheduler for distributed work, a durable queue, a service mesh, or a replacement for context. It supervises goroutines inside one process and nothing more. It does not persist state across restarts, retry with deduplication, or guarantee delivery of messages.

Contract

  • Lifecycle is single-shot: after a successful Stop the orchestrator cannot be restarted.
  • The wire format is encoding/gob and is part of the public contract; types passed through the typed Messenger helpers must be gob-compatible.
  • Publish is drop-only: when a subscriber's buffer is full the message is dropped for that subscriber.

See the README for the concurrency table and the per-method guarantees.

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")
	ErrUnsupportedOption = errors.New("gorch: unsupported option combination")
)

Sentinel errors

Functions

func RegisterType 

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 

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 

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 requestMessage, and gob-decodes the response. Returns the decoded response or an error.

func TypedRespond 

func TypedRespond[TResp any](m *Messenger, resp TResp, replyTopic string)

TypedRespond gob-encodes resp and publishes it to replyTopic. Pair with TypedSubscribeRequest to implement a typed request responder. Silently drops the reply if the type is not gob-encodable. Thread-safe.

func TypedSubscribe 

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.

func TypedSubscribeRequest 

func TypedSubscribeRequest[TReq any](m *Messenger, topic string) (<-chan TypedEnvelope[TReq], func())

TypedSubscribeRequest subscribes to topic and returns a channel of decoded request envelopes. Pair with TypedRespond to reply. Non-Message values and decode failures are silently dropped. Thread-safe.

Types

type Backoff 

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 ConstantBackoff 

type ConstantBackoff struct {
	Delay time.Duration
}

ConstantBackoff always returns the same delay regardless of retry count.

func (ConstantBackoff) Next 

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

Next returns Delay (ignores retry count).

type CronMode 

type CronMode int
const (
	CronParallel CronMode = iota // fire in new goroutine regardless
	// CronQueue serializes ticks on a per-entry mutex: an overlapping tick
	// blocks until the previous one finishes. robfig/cron spawns a goroutine per
	// tick, so a long-running tick makes later ones pile up as blocked goroutines.
	CronQueue
	CronSkip // drop this tick entirely
)

type ExponentialBackoff 

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 

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

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

type HealthCheckOption 

type HealthCheckOption func(*config)

HealthCheckOption refines the periodic health-check loop configured by WithHealthChecks. It exists so the probe timeout and failure threshold cannot be swapped by position at the call site.

func WithFailureThreshold 

func WithFailureThreshold(n int) HealthCheckOption

WithFailureThreshold sets how many consecutive probe failures are tolerated before a self-healing service is restarted. Zero falls back to the default (3).

func WithProbeTimeout 

func WithProbeTimeout(d time.Duration) HealthCheckOption

WithProbeTimeout sets the per-probe deadline for each health check. Zero falls back to the default (5s).

type HealthChecker 

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
const (
	LogLevelDebug LogLevel = iota
	LogLevelInfo
	LogLevelWarn
	LogLevelError
)

func (LogLevel) String 

func (l LogLevel) String() string

type Logger 

type Logger interface {
	Info(msg string, args ...any)
	Error(msg string, args ...any)
	Debug(msg string, args ...any)
	Warn(msg string, args ...any)
}

type Message 

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
}

func (*Messenger) Drain 

func (m *Messenger) Drain()

Drain closes all subscriber channels and clears all subscriptions. Buffered messages are delivered to receivers before they observe the close. After Drain, the Messenger is empty, Publish is a no-op, and a subsequent Subscribe re-initializes the subscription map. 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 

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 

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(). The returned channel is delivered to exactly once on reply, and the forwarding goroutine exits on either a reply or context cancellation. 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 

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. Safe to call after Drain (subscriptions are lazily re-initialized). Thread-safe.

type Metrics 

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

type Option 

type Option func(*config)

Option configures an Orchestrator at construction via New.

func WithAfterHealthCheck 

func WithAfterHealthCheck(fn func(name string, err error)) Option

WithAfterHealthCheck sets a hook fired after each health probe.

func WithBeforeHealthCheck 

func WithBeforeHealthCheck(fn func(name string) error) Option

WithBeforeHealthCheck sets a hook fired before each health probe.

func WithDefaultStartTimeout 

func WithDefaultStartTimeout(d time.Duration) Option

WithDefaultStartTimeout sets the default per-service start deadline. 0 means no timeout (use WithStartTimeout per-service).

func WithGlobalOnAfterStart 

func WithGlobalOnAfterStart(fn func(name string, err error)) Option

WithGlobalOnAfterStart sets a global hook called after each service's Start returns.

func WithGlobalOnAfterStop 

func WithGlobalOnAfterStop(fn func(name string, err error)) Option

WithGlobalOnAfterStop sets a global hook called after each service's Stop returns.

func WithGlobalOnBeforeStart 

func WithGlobalOnBeforeStart(fn func(name string) error) Option

WithGlobalOnBeforeStart sets a global hook called just before each service's Start.

func WithGlobalOnBeforeStop 

func WithGlobalOnBeforeStop(fn func(name string) error) Option

WithGlobalOnBeforeStop sets a global hook called just before each service's Stop.

func WithHealthChecks 

func WithHealthChecks(interval time.Duration, opts ...HealthCheckOption) Option

WithHealthChecks enables periodic health checks at the given interval. The probe timeout and failure threshold default to 5s and 3; override them with WithProbeTimeout and WithFailureThreshold. An interval of zero enables the loop at the default 30s. Use WithHealthChecksDisabled to turn it off.

func WithHealthChecksDisabled 

func WithHealthChecksDisabled() Option

WithHealthChecksDisabled disables the periodic health-check loop entirely.

func WithLogLevel 

func WithLogLevel(lvl LogLevel) Option

WithLogLevel sets the minimum log level. Defaults to LogLevelInfo when absent. Ignored when a custom Logger is set.

func WithLogger 

func WithLogger(l Logger) Option

WithLogger sets a custom logger. When set, gorch sends all log output through it instead of the built-in stderr logger.

func WithOnCrash 

func WithOnCrash(fn func(name string, err error)) Option

WithOnCrash sets a callback fired when a service reaches StatusCrashed.

func WithOnStateChange 

func WithOnStateChange(fn func(name string, from, to ServiceStatus)) Option

WithOnStateChange sets a callback fired on every status transition.

type Orchestrator 

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

Orchestrator manages service lifecycles.

func New 

func New(opts ...Option) *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. Configure via Option functions; the zero-option call uses the defaults (LogLevelInfo, health checks every 30s with a 5s probe timeout).

func (*Orchestrator) Count 

func (o *Orchestrator) Count() int

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

func (*Orchestrator) Done 

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. The channel is created lazily and cached: repeated calls return the same channel.

func (*Orchestrator) Health 

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

func (*Orchestrator) IsReady 

func (o *Orchestrator) IsReady(ctx context.Context, name string) bool

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. IsReady reports whether a named service is running and ready to serve. The ReadinessChecker probe (if any) runs with the given ctx, so callers can bound how long they wait (e.g. IsReady(ctx, name) with a deadline context). Thread-safe.

func (*Orchestrator) Metrics 

func (o *Orchestrator) Metrics() Metrics

Metrics returns a snapshot of orchestrator-level event counters.

func (*Orchestrator) Names 

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 

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

RegisterFunc registers a closure-based service under the given name. Thread-safe.

func (*Orchestrator) Run 

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 (SIGINT, SIGTERM).

func (*Orchestrator) Start 

func (o *Orchestrator) Start() error

Start begins the orchestrator lifecycle. Returns ErrAlreadyStarted if already started. If Start fails, the orchestrator is reset and may be started again (e.g. to retry after a transient dependency failure). A persistent service that returns an error synchronously aborts Start only when a start timeout is set; without one its launch is fire-and-forget by construction. An orchestrator is single-shot: after a successful Stop it cannot be restarted; a subsequent Start (or Register) returns ErrAlreadyStarted. Thread-safe.

func (*Orchestrator) StartGroup 

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

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

func (*Orchestrator) Status 

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 

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 

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 

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 shuts down the orchestrator, waiting 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 (no-op). An orchestrator is single-shot: after a successful Stop it cannot be restarted; a subsequent Start (or Register) returns ErrAlreadyStarted.

func (*Orchestrator) StopGroup 

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 

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 

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 

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 

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 

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 

func WithGroup(name string) RegisterOption

WithGroup assigns the service to a named group for filtering.

func WithLabel 

func WithLabel(key, value string) RegisterOption

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

func WithMaxRetries 

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 

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 

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

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

func WithOnAfterStop 

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

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

func WithOnBeforeStart 

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 

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 

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 

func WithRunOnce() RegisterOption

WithRunOnce marks a service as a one-shot init task. It runs before persistent services and transitions to StatusSucceeded when Start returns. Stop() is called at orchestrator shutdown — make Stop idempotent. 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 

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 

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). With a timeout, a synchronous error from a persistent service's Start aborts the whole orchestrator Start (deterministic failure); without one the launch is fire-and-forget. Self-heal services are never aborted this way: an exit is handled by their restart policy.

func WithStopTimeout 

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 ServiceContext) error // blocks; for cron: runs per-tick; for non-cron: runs until ctx cancelled
	Stop() error                    // cleanup signal beyond context cancellation
}

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). When a custom Logger is set via Config.Logger, ServiceLogger delegates to it instead of the channel, prepending "service"=<name> to the key-value pairs.

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 

type ServiceStatus int

ServiceStatus represents the lifecycle state of a registered service. 

const (
	StatusRegistered ServiceStatus = iota
	StatusStarting
	StatusRunning
	StatusStopping
	StatusStopped
	StatusCrashed
	// StatusSucceeded marks a runOnce service whose Start completed without
	// error: it is a successful gate, distinct from StatusStopped so dependents
	// are not aborted by a gate that did its job.
	StatusSucceeded
)

func (ServiceStatus) String 

func (s ServiceStatus) String() string

String returns a human-readable name for the status.

type TypedEnvelope 

type TypedEnvelope[T any] struct {
	Value      T
	ReplyTopic string
}

TypedEnvelope carries a decoded typed request value together with the reply topic the responder should publish to. Produced by TypedSubscribeRequest.

type Validator 

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

Directories

Path Synopsis
examples
advanced command
Advanced example: groups, labels, soft dependencies, RegisterFunc, Validator, ReadinessChecker, HealthChecker, state-change hooks, health-check hooks, WithStartCondition, WaitFor, Metrics, and Done().
 Advanced example: groups, labels, soft dependencies, RegisterFunc, Validator, ReadinessChecker, HealthChecker, state-change hooks, health-check hooks, WithStartCondition, WaitFor, Metrics, and Done().
basic command
Basic example: service lifecycle, cron scheduling, and graceful shutdown.
 Basic example: service lifecycle, cron scheduling, and graceful shutdown.
pubsub command
Pub-sub example: services communicating via topics through the Messenger.
 Pub-sub example: services communicating via topics through the Messenger.
typedreq command
Typed request-reply example: a service handles typed requests via TypedSubscribeRequest + TypedRespond, and a requester issues a TypedRequest.
 Typed request-reply example: a service handles typed requests via TypedSubscribeRequest + TypedRespond, and a requester issues a TypedRequest.

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.