Skip to content
Documentation
Reference
Go SDK
server
API

server

import "github.com/danmestas/dagnats/server"

server/dry_run.go Dry-run validation: loads config, reports sources, checks prerequisites. Validates environment without starting any components.

server/metrics_auth.go METRICS_AUTH gate around the /metrics exporter. Independent from the console gate: an operator may want the console locked behind basic auth while letting their Prometheus scraper hit /metrics with a service-account token (forward auth). Documented modes:

loopback (default): only 127.0.0.1 / ::1 reaches /metrics.
basic: HTTP Basic Auth from METRICS_BASIC_USER + METRICS_BASIC_PASS.
forward: trust X-Forwarded-User from an upstream proxy.
none: open. Logs a Warn at startup in a non-dev context.

The gate wraps prom.Handler; the rendered output is unchanged.

Index

Constants 

const (

    // DefaultRunsMaxAge is the run-retention window applied when the operator
    // configures nothing (#521). It matches WORKFLOW_HISTORY's 30d so a run's
    // authoritative snapshot never outlives the history it summarizes, and it
    // bounds the otherwise-unbounded workflow_runs KV (the prod disk-growth
    // symptom). The pruner is terminal-only (deletes by CompletedAt), so
    // in-flight runs are never touched and recovery is preserved. An explicit
    // 0/off/disabled still turns pruning off — see parseRetentionDuration.
    DefaultRunsMaxAge = 30 * 24 * time.Hour
)

func LogMetricsAuthStartup 

func LogMetricsAuthStartup(logger *slog.Logger, cfg MetricsAuthConfig, httpAddr string)

LogMetricsAuthStartup emits a single INFO line announcing the resolved auth mode for /metrics, and escalates to WARN when the listener is non-loopback AND the operator picked the open mode. The WARN message is operator-actionable: it tells the reader that the endpoint is open to anyone on the network, not just localhost. Called from the server boot path immediately after LoadMetricsAuthConfigFromEnv resolves.

func PrintDryRun 

func PrintDryRun(w io.Writer, rc ResolvedConfig) bool

PrintDryRun writes the dry-run report to w. Returns true if all validations passed.

type Config

Config holds all server configuration.

type Config struct {
    DataDir         string   `json:"data_dir"`
    HTTPAddr        string   `json:"http_addr"`
    NATSPort        int      `json:"nats_port"`
    LeafRemotes     []string `json:"leaf_remotes"`
    LeafCredentials string   `json:"leaf_credentials"`

    NATSClusterName       string   `json:"nats_cluster_name"`
    NATSClusterRoutes     []string `json:"nats_cluster_routes"`
    NATSClusterAuthToken  string   `json:"nats_cluster_auth_token"`
    NATSJetStreamReplicas int      `json:"nats_jetstream_replicas"`

    MonitorPort   int   `json:"monitor_port"`
    MaxStoreBytes int64 `json:"max_store_bytes"`
    // MaxMemoryBytes caps the JetStream in-memory store
    // (JetStreamMaxMemory) and is applied as the soft Go memory limit at
    // startup (#441). Defaults to defaultMaxMemoryBytes; <= 0 disables the
    // JetStream cap and the Go limit.
    MaxMemoryBytes int64          `json:"max_memory_bytes"`
    Workers        []WorkerConfig `json:"workers"`
    OTLPEndpoint   string         `json:"otlp_endpoint"`

    // RunsMaxAge is the run-retention window for the workflow_runs KV
    // (#453, #521). It DEFAULTS to DefaultRunsMaxAge (30d): an unconfigured
    // serve prunes terminal runs older than 30d, bounding the previously
    // unbounded bucket. When > 0, terminal runs whose CompletedAt is older
    // than this are dropped (delete-only) by the orchestrator's background
    // sweeper; in-flight runs are never touched. Set via DAGNATS_RUNS_MAX_AGE
    // or the runs_max_age config key, which accept a Go duration ("720h") or
    // a d/w suffix ("30d", "2w"). An explicit 0/off/disabled turns pruning
    // off entirely — the escape hatch for operators who want it off.
    RunsMaxAge time.Duration `json:"runs_max_age"`

    // Per-runtime safety bounds (ADR-021 Phase A, #378). These cap a
    // single spawn-tree's resource use so a runaway agent loop cannot
    // fork-bomb the orchestrator. Zero means DEFAULT, not "unlimited" — it
    // resolves to the default consts in internal/api at service
    // construction (additive: an old config file silently inherits the
    // defaults; a safety layer has no "off" switch). #380 owns grant
    // policy; #378 only reads these numeric values.
    //
    // MaxActiveRunsPerRoot caps non-terminal runs sharing a tree-root.
    // MaxDefsPerRoot caps ephemeral defs registered under a tree-root.
    // MaxGenerationDepth caps spawn-chain nesting depth (reuses the
    // existing depth check; default equals engine.MaxNestingDepth).
    // MaxRegistersPerMinutePerRoot rate-limits runtime def registration.
    MaxActiveRunsPerRoot         int `json:"max_active_runs_per_root"`
    MaxDefsPerRoot               int `json:"max_defs_per_root"`
    MaxGenerationDepth           int `json:"max_generation_depth"`
    MaxRegistersPerMinutePerRoot int `json:"max_registers_per_minute_per_root"`

    // NATSWebsocketPort enables an embedded NATS WebSocket
    // listener for browser clients when > 0. 0 (default)
    // disables it — the safe production posture. See ADR-020.
    NATSWebsocketPort int `json:"nats_ws_port"`

    // NATSWebsocketNoTLS turns off TLS for the WebSocket
    // listener. Until top-level NATS TLS is wired this is
    // required when NATSWebsocketPort > 0; the explicit
    // opt-in keeps operators from shipping cleartext to
    // production by accident.
    NATSWebsocketNoTLS bool `json:"nats_ws_no_tls"`

    // FailOnPortConflict makes startup return an error (non-zero exit)
    // instead of auto-falling-back to an ephemeral port when the default
    // NATS port or default HTTP address is already in use. Default false
    // keeps auto-fallback as the documented behavior (#370). Opt-in for
    // operators who want a hard failure when a stale server holds the
    // port.
    FailOnPortConflict bool `json:"fail_on_port_conflict"`

    // Build is the binary's version/revision string, threaded from
    // cli.Version by the serve command (ldflags-stamped). Empty for
    // un-stamped local builds — the console footer degrades empty to
    // the honest "dev" marker (consoleBuildLabel). Not persisted to
    // dagnats.yaml; it is link-time identity, not user config.
    Build string `json:"-"`

    // ConfigFilePath is the absolute path of the dagnats.yaml that
    // was loaded (empty when no file was found). Phase 4 / ADR-018:
    // the server uses it to drive the configfile.Watcher for live
    // reload of workflows and triggers declared in the same file.
    // Not stored in the on-disk file itself — populated by the CLI
    // from the resolved path after the file is loaded.
    ConfigFilePath string `json:"-"`

    // DieWithParent makes a spawned `dagnats serve` self-terminate via
    // the normal graceful shutdown when its parent process dies (#476).
    // Default OFF. Opt-in for sidecar spawners (notify's e2e tests, the
    // eventbus sidecar) whose own cleanup can't run on SIGKILL /
    // `go test -timeout`, so they'd otherwise orphan the server. The
    // CLI's --die-with-parent flag sets it; not persisted to
    // dagnats.yaml — it's an invocation mode, not stored config.
    DieWithParent bool `json:"-"`
}

func ConfigFromEnv 

func ConfigFromEnv() Config

ConfigFromEnv loads config from defaults, config file, then env vars. Config file is dagnats.yaml in CWD. Missing file is not an error. Panics if DataDir is empty or MaxStoreBytes <= 0 after resolution.

func ConfigWithPath 

func ConfigWithPath(configPath string) (Config, string, error)

ConfigWithPath loads config using an explicit path or standard search. Returns the resolved config and the path of the file that was loaded (empty string if no file was found). When configPath is non-empty, the file must exist or an error is returned. Panics if DataDir is empty or MaxStoreBytes <= 0 after resolution.

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig returns platform-appropriate defaults. Panics if dataDir resolves empty.

type ConfigEntry

ConfigEntry holds a resolved config value and its source.

type ConfigEntry struct {
    Key    string
    Value  string
    Source string
}

type MetricsAuthConfig

MetricsAuthConfig captures the gate’s inputs. HTTPAddr is required because the loopback mode rejects non-loopback requests by remote address rather than listener bind.

type MetricsAuthConfig struct {
    Mode      MetricsAuthMode
    BasicUser string
    BasicPass string
}

func LoadMetricsAuthConfigFromEnv 

func LoadMetricsAuthConfigFromEnv(logger *slog.Logger) MetricsAuthConfig

LoadMetricsAuthConfigFromEnv reads METRICS_AUTH + METRICS_BASIC_USER + METRICS_BASIC_PASS from the process environment. Defaults to loopback when METRICS_AUTH is unset. Unknown modes are normalised to loopback with a slog.Warn so a typo fails closed.

type MetricsAuthMode

MetricsAuthMode is the bounded enum the gate accepts. The env var METRICS_AUTH carries the string form; resolveMetricsAuthMode normalises it.

type MetricsAuthMode string

const (
    MetricsAuthLoopback MetricsAuthMode = "loopback"
    MetricsAuthBasic    MetricsAuthMode = "basic"
    MetricsAuthForward  MetricsAuthMode = "forward"
    MetricsAuthNone     MetricsAuthMode = "none"
)

type ResolvedConfig

ResolvedConfig holds config entries with provenance tracking.

type ResolvedConfig struct {
    Config  Config
    Entries []ConfigEntry
}

func ResolveConfig 

func ResolveConfig() ResolvedConfig

ResolveConfig loads config and tracks the source of each value. Returns resolved config with provenance for every key.

type Server

Server is the all-in-one DagNats server lifecycle manager.

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

func New 

func New(cfg Config) *Server

New creates a Server with the given config. Panics if DataDir is empty.

func (*Server) Run 

func (s *Server) Run() error

Run starts all server components, serves HTTP, and blocks until shutdown. Returns nil on clean shutdown, error otherwise.

func (*Server) Stop 

func (s *Server) Stop()

Stop closes the stopCh to trigger shutdown. Safe to call multiple times.

type ValidationResult

ValidationResult holds one check outcome.

type ValidationResult struct {
    Name   string
    Passed bool
    Detail string
}

func DryRunValidate 

func DryRunValidate(cfg Config) ([]ValidationResult, bool)

DryRunValidate checks prerequisites without starting components. Returns validation results and true if all passed.

type WorkerConfig

WorkerConfig defines a config-driven embedded worker handler.

type WorkerConfig struct {
    Task       string
    Exec       string
    HTTP       string
    HTTPMethod string // default: POST
}

type WorkerShim

WorkerShim collects handler registrations before the server starts. Returned by EmbeddedWorker(). The shim is materialized to a real *worker.Worker during startComponents().

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

func EmbeddedWorker 

func EmbeddedWorker(srv *Server) *WorkerShim

EmbeddedWorker creates a WorkerShim bound to srv’s lifecycle. Must be called before Run(). Panics if called after Run(), if srv is nil, or if the max embedded worker limit is exceeded.

func (*WorkerShim) Handle 

func (s *WorkerShim) Handle(taskType string, handler worker.HandlerFunc)

Handle registers a handler for a task type. Panics if called after Run(), if taskType is empty, or if handler is nil.

func (*WorkerShim) HandleLoop 

func (s *WorkerShim) HandleLoop(taskType string, fn func(worker.LoopTask) error)

HandleLoop registers an agent-loop handler. During materialization, the handler is wrapped as HandlerFunc and dispatched via w.Handle.

func (*WorkerShim) HandleSignal 

func (s *WorkerShim) HandleSignal(taskType string, fn func(worker.SignalTask) error)

HandleSignal registers an inter-step signal handler. During materialization, the handler is wrapped as HandlerFunc and dispatched via w.Handle.

func (*WorkerShim) HandleSingleton 

func (s *WorkerShim) HandleSingleton(taskType string, handler worker.HandlerFunc)

HandleSingleton registers a handler that runs as a single- partition consumer. During materialization, translated to worker.HandleSingleton.

func (*WorkerShim) HandleStream 

func (s *WorkerShim) HandleStream(taskType string, fn func(worker.StreamTask) error)

HandleStream registers a streaming-output handler. During materialization, the handler is wrapped as HandlerFunc and dispatched via w.Handle.

func (*WorkerShim) WithGroups 

func (s *WorkerShim) WithGroups(groups ...string)

WithGroups configures this embedded worker for specific worker groups. During materialization, translated to worker.WithGroups(groups…). Panics after Run().

Generated by gomarkdoc