Documentation
¶
Overview ¶
Package config provides configuration management for StreamKit applications.
This package handles loading, validation, and dynamic updates of application configuration from JSON files, environment variables, and NATS KV store.
Core Components ¶
Config: Main configuration structure containing platform settings, NATS connection details, service configurations, and component definitions.
SafeConfig: Thread-safe wrapper using RWMutex and deep cloning to prevent concurrent access issues and accidental mutations.
Manager: Manages the complete lifecycle of configuration including initialization, NATS KV watching, change notifications via channels, and graceful shutdown with timeout handling.
Loader: Loads configuration with layer merging (base + overrides) and environment variable substitution for flexible deployment scenarios.
Basic Usage ¶
Loading configuration from files with layer merging:
loader := config.NewLoader()
loader.AddLayer("config/base.json")
loader.AddLayer("config/production.json") // Overrides base
loader.EnableValidation(true)
cfg, err := loader.Load()
if err != nil {
log.Fatal(err)
}
Dynamic Configuration ¶
Using Manager for runtime updates via NATS KV:
cm, err := config.NewConfigManager(cfg, natsClient, logger)
if err != nil {
log.Fatal(err)
}
// Start watching for config changes
if err := cm.Start(ctx); err != nil {
log.Fatal(err)
}
defer cm.Stop(5 * time.Second)
// Subscribe to specific config changes
updates := cm.OnChange("services.*")
for update := range updates {
log.Printf("Service config changed: %s", update.Key)
}
Thread-Safe Access ¶
SafeConfig ensures thread-safe access to configuration:
safeConfig := cm.GetConfig()
// Read config (deep copy returned, safe to use)
cfg := safeConfig.Get()
// Read-modify-write atomically: Mutate holds the write lock across the whole
// clone → mutate → swap so concurrent mutations cannot lose one another's
// change (gh#515). Do NOT do Get() → mutate → Update() — the lock is released
// between the read and the swap, so a concurrent writer can clobber you.
safeConfig.Mutate(func(cfg *Config) error {
c := cfg.Components["my-component"]
c.Enabled = true
cfg.Components["my-component"] = c
return nil
})
// Push updates to NATS KV
cm.PushToKV(ctx)
Environment Variable Overrides ¶
Configuration values can be overridden using environment variables:
# Override platform ID export STREAMKIT_PLATFORM_ID="prod-cluster-01" # Override NATS URLs (comma-separated) export STREAMKIT_NATS_URLS="nats://server1:4222,nats://server2:4222"
Layer Merging ¶
Configuration layers are merged with last-wins semantics:
base.json:
{"platform": {"id": "dev", "log_level": "debug"}}
production.json:
{"platform": {"id": "prod"}}
Result:
{"platform": {"id": "prod", "log_level": "debug"}}
Security ¶
The package includes security validation:
- File size limits (10MB max) to prevent memory exhaustion
- JSON depth validation (100 levels max) to prevent DoS attacks
- Path validation to prevent directory traversal
- Regular file checks (no symlinks or device files)
Configuration Structure ¶
The main Config struct contains:
type Config struct {
Platform PlatformConfig // Platform metadata
NATS NATSConfig // Message bus connection
Services map[string]any // Service configurations
Components map[string]ComponentConfig // Component definitions
}
See the README.md file for detailed examples and configuration patterns.
Package config provides configuration management for SemStreams.
Example (ComponentAccess) ¶
Example_componentAccess demonstrates type-safe component configuration access.
package main
import (
"fmt"
)
func main() {
// Assume we have a loaded configuration
// cfg := loadConfig()
// Get component configuration with type checking
// comp, exists := cfg.Components["udp-input"]
// if !exists {
// log.Fatal("Component not found")
// }
// Access component properties
// componentType := comp.Type
// enabled := comp.Enabled
// config := comp.Config
// Type-safe access to nested config using helpers
// bindAddr := cfg.GetString("components.udp-input.config.bind_address")
// port := cfg.GetInt("components.udp-input.config.port")
fmt.Println("Type-safe component access")
}
Output: Type-safe component access
Index ¶
- Constants
- Variables
- func CompareVersions(v1, v2 string) (int, error)
- func DeriveStreamName(subject string) string
- func DeriveStreamSubjects(subject string) []string
- func ExpandEnvWithDefaults(s string) string
- func FrameworkStreamAutoCreate(cfg *Config, name string) (*natsclient.StreamAutoCreateConfig, bool)
- func GetBool(cfg map[string]any, key string, defaultVal bool) bool
- func GetComponentConfig(cfg map[string]any, name string) (map[string]any, error)
- func GetFloat64(cfg map[string]any, key string, defaultVal float64) float64
- func GetInt(cfg map[string]any, key string, defaultVal int) int
- func GetNestedBool(cfg map[string]any, keys []string, defaultVal bool) bool
- func GetNestedInt(cfg map[string]any, keys []string, defaultVal int) int
- func GetNestedString(cfg map[string]any, keys []string, defaultVal string) string
- func GetString(cfg map[string]any, key string, defaultVal string) string
- func GetStringSlice(cfg map[string]any, key string, defaultVal []string) []string
- func HasKey(cfg map[string]any, key string) bool
- func HasNestedKey(cfg map[string]any, keys []string) bool
- type ArchivalStream
- type ArchivalStreamStatus
- type ArchivalStreams
- type BucketConfig
- type ComponentConfigs
- type Config
- type CoreServicesConfig
- type JetStreamConfig
- type Loader
- type Manager
- func (cm *Manager) DeleteComponentFromKV(ctx context.Context, name string) error
- func (cm *Manager) GetConfig() *SafeConfig
- func (cm *Manager) OnChange(pattern string) <-chan Update
- func (cm *Manager) PushToKV(ctx context.Context) error
- func (cm *Manager) PutComponentToKV(ctx context.Context, name string, compConfig types.ComponentConfig) error
- func (cm *Manager) Start(ctx context.Context) error
- func (cm *Manager) Stop(timeout time.Duration) error
- func (cm *Manager) WatchModelRegistry() <-chan *model.Registry
- type MinimalConfig
- type NATSConfig
- type NATSTLSConfig
- type PlatformConfig
- type PortDefinition
- type PortsConfig
- type SafeConfig
- type StreamConfig
- type StreamConfigs
- type StreamExceptionReport
- type StreamMigrationOverride
- type StreamMigrationOverrideStatus
- type StreamMigrationOverrides
- type StreamsManager
- type Update
Examples ¶
Constants ¶
const ( StorageModeMemory = "memory" // In-memory only (original implementation) StorageModeKV = "kv" // NATS KV only (no local cache) StorageModeHybrid = "hybrid" // KV + local cache (recommended for production) )
Storage mode constants
const ( // StreamDiscardOld evicts the oldest messages once the stream reaches a // limit. The write always succeeds; the data loss is silent and at the tail. StreamDiscardOld = "old" // StreamDiscardNew refuses the write once the stream reaches a limit. Nothing // is evicted; producers see the rejection instead. See discardPolicyGuidance. StreamDiscardNew = "new" )
Discard-policy declaration values for StreamConfig.Discard. These are the operator-facing spellings of jetstream.DiscardOld / jetstream.DiscardNew; the mapping happens once, in buildStreamConfig.
Variables ¶
var ( // ErrStreamBoundsUndeclared is returned when an ordinary stream carries no // explicitly declared finite MaxAge, finite MaxBytes, or discard policy and // is not admitted by an archival declaration or an active migration override. // // It is natsclient's value rather than a second one of the same name. The same // requirement is enforced at two seams — this declarative path and // natsclient.Client.EnsureStream — and a caller testing for "bounds are not // declared" should not have to know which door refused it. ErrStreamBoundsUndeclared = natsclient.ErrStreamBoundsUndeclared // ErrStreamMigrationOverrideInvalid is returned when a migration override is // not a valid time-limited bridge — no owner, or no expiry. An open-ended // override is rejected at validation so a bridge cannot become permanent. ErrStreamMigrationOverrideInvalid = errors.New("stream migration override is not a valid time-limited bridge") // ErrStreamMigrationOverrideExpired is returned when an override's expiry has // passed. Readiness fails rather than degrading quietly: the bridge was // declared with an end date and the end date is the whole point. ErrStreamMigrationOverrideExpired = errors.New("stream migration override has expired") // ErrArchivalStreamInvalid is returned when an archival declaration omits its // owner or its reason, so archival cannot become a silent way to opt out of // bounds. ErrArchivalStreamInvalid = errors.New("archival stream declaration is incomplete") )
Bounds-contract sentinels. Classifiable so a boot path, an operator tool, or a test can tell an undeclared bound from an expired bridge from a malformed archival declaration, all of which fail readiness for different reasons and have different fixes.
var ErrBackingStreamNotProvisionable = natsclient.ErrBackingStreamNotProvisionable
ErrBackingStreamNotProvisionable is the refusal sentinel, re-exported from natsclient so config callers and tests need not reach across for it. It is the SAME error value, so errors.Is matches a refusal raised at any provisioning seam — this provisioner, Client.EnsureStream, or Client.CreateStream.
var ErrStreamNonEditableDrift = errors.New("existing stream diverges in a field that cannot be reconciled in place")
ErrStreamNonEditableDrift is returned when an existing ordinary stream's live configuration diverges from its declaration in a field the provisioner will not change on a live stream. Readiness fails rather than accepting the divergence in silence, because a stream that reports one storage tier or retention policy while its declaration states another is a configuration two people can read opposite answers from.
Functions ¶
func CompareVersions ¶
CompareVersions compares two semver version strings Returns:
-1 if v1 < v2 0 if v1 == v2 1 if v1 > v2 error if either version is invalid
func DeriveStreamName ¶
DeriveStreamName extracts stream name from subject convention. Convention: subject "component.action.type" → stream "COMPONENT" Examples:
"objectstore.stored.entity" → "OBJECTSTORE" "sensor.processed.entity" → "SENSOR" "rule.triggered.alert" → "RULE"
func DeriveStreamSubjects ¶
DeriveStreamSubjects creates wildcard pattern for stream capture. Convention: subject "component.action.type" → ["component.>"] Examples:
"objectstore.stored.entity" → ["objectstore.>"] "sensor.processed.entity" → ["sensor.>"]
func ExpandEnvWithDefaults ¶
ExpandEnvWithDefaults expands environment variables in a string, supporting ${VAR:-default} syntax for default values.
Patterns:
- ${VAR} - expands to value of VAR, or empty if unset
- ${VAR:-default} - expands to value of VAR, or "default" if unset
- $VAR - expands to value of VAR (uppercase identifiers only; lowercase prefixes belong to the rule engine's substitution namespaces and pass through unchanged)
func FrameworkStreamAutoCreate ¶
func FrameworkStreamAutoCreate(cfg *Config, name string) (*natsclient.StreamAutoCreateConfig, bool)
FrameworkStreamAutoCreate returns the EFFECTIVE declaration for a framework-guaranteed stream, in the form a consumer's auto-create path takes. ok is false for a name this configuration does not declare, or for one whose declaration does not resolve.
It exists so a consumer that auto-creates one of these streams recreates it with the bounds that are actually in force rather than inventing its own — or, as was the case, inventing none. The framework's HEALTH, METRICS and FLOWS streams are memory-backed: a NATS restart destroys them, and the reconnect that recreates them must not be the moment their declaration is quietly replaced.
It resolves through planStreams — the SAME path boot provisioning uses — rather than reading the framework constants directly. That distinction is the whole point of taking a *Config: `cfg.streams` is the highest-priority declaration and an operator may override any framework stream's storage, bounds or discard policy there. Reading the constants would recreate the stream with the built-in values and silently discard the operator's, which is the same class of quiet replacement this accessor was introduced to stop.
A consumer would ideally not provision at all (a reader binds by name — see the ownership contract in natsclient's package doc). Auto-create stays because it is what recovers these streams mid-process after the server restarts, and recovering them WITH the declaration in force is the point.
func GetComponentConfig ¶
GetComponentConfig safely extracts a component configuration section
func GetFloat64 ¶
GetFloat64 safely extracts a float64 value from a config map
func GetNestedBool ¶
GetNestedBool safely extracts a nested boolean value from a config map
func GetNestedInt ¶
GetNestedInt safely extracts a nested integer value from a config map
func GetNestedString ¶
GetNestedString safely extracts a nested string value from a config map
func GetStringSlice ¶
GetStringSlice safely extracts a string slice from a config map
Types ¶
type ArchivalStream ¶
type ArchivalStream struct {
// Owner is the team or component accountable for this stream's growth.
// Required — an archival stream's only remaining lever is capacity, and
// capacity questions need an addressee.
Owner string `json:"owner"`
// Reason states why permanence is the contract. Required — without it,
// "archival" is just an unbounded stream with better vocabulary.
Reason string `json:"reason"`
}
ArchivalStream declares ONE ordinary stream whose contract is permanence: nothing may ever be evicted from it. It is exempt from the finite-bounds requirement by declaration, and readiness reports it as a named PERMANENT exception structurally distinct from a time-limited migration override.
Both fields are required. An archive expressible only as a renewed override would train an operator to renew without reading, which is exactly what makes the genuinely time-limited overrides invisible.
type ArchivalStreamStatus ¶
type ArchivalStreamStatus struct {
Stream string // the permanently unbounded ordinary stream
Owner string // who is accountable for its growth
Reason string // why permanence is the contract
}
ArchivalStreamStatus is one permanent exception as readiness reports it.
This is a DIFFERENT TYPE from StreamMigrationOverrideStatus rather than the same struct with a flag, and the difference is load-bearing: it has no expiry field and no remaining-time field, so no surface can render a permanent exception as though it were counting down, and no operator can renew it.
type ArchivalStreams ¶
type ArchivalStreams map[string]ArchivalStream
ArchivalStreams maps ordinary stream name to its archival declaration.
type BucketConfig ¶
type BucketConfig struct {
Name string `json:"name,omitempty"` // Override default name if needed
TTL time.Duration `json:"ttl"` // 0 = no expiration
History int `json:"history"` // Number of versions to keep
MaxBytes int64 `json:"max_bytes,omitempty"` // Size limit (0 = unlimited)
Replicas int `json:"replicas,omitempty"` // Replication factor
}
BucketConfig defines configuration for a single KV bucket
type ComponentConfigs ¶
type ComponentConfigs map[string]types.ComponentConfig
ComponentConfigs holds component instance configurations. The map key is the instance name (e.g., "udp-sensor-main"). Components are only created if both: 1. Their factory has been registered via init() 2. They have an entry in this config map with enabled=true
type Config ¶
type Config struct {
Version string `json:"version"` // Semantic version (e.g., "1.0.0") for KV sync control
Platform PlatformConfig `json:"platform"`
Security security.Config `json:"security,omitempty"` // Platform-wide security configuration
NATS NATSConfig `json:"nats"`
Services types.ServiceConfigs `json:"services"` // Map of service configs
Components ComponentConfigs `json:"components"` // Map of component instance configs
Streams StreamConfigs `json:"streams,omitempty"` // Explicit JetStream stream definitions
ModelRegistry *model.Registry `json:"model_registry,omitempty"` // Unified model endpoint registry
// StreamMigrationOverrides admits named ordinary streams that predate the
// bounds contract, each for a declared period. Readiness reports every
// active override and FAILS once one expires; an override with no expiry is
// rejected at validation, so a bridge cannot become permanent.
StreamMigrationOverrides StreamMigrationOverrides `json:"stream_migration_overrides,omitempty"`
// ArchivalStreams declares ordinary streams whose contract is permanence.
// Structurally separate from StreamMigrationOverrides: an archive that could
// only be expressed as a renewed override would train operators to renew
// without reading, which is what makes genuinely time-limited overrides
// invisible.
ArchivalStreams ArchivalStreams `json:"archival_streams,omitempty"`
}
Config represents the complete application configuration Simplified to 6 fields: Version (semver), Platform (identity), Security (TLS), NATS (connection), Services, Components
func (*Config) GetPlatform ¶
GetPlatform returns the platform identifier (prefer instance_id over id)
func (*Config) SaveToFile ¶
SaveToFile saves the configuration to a JSON file
func (*Config) UnmarshalJSON ¶
UnmarshalJSON implements custom JSON unmarshaling for Config
type CoreServicesConfig ¶
type CoreServicesConfig struct {
MessageLogger bool `json:"message_logger"` // Debug tool
Discovery bool `json:"discovery"` // Component discovery
}
CoreServicesConfig defines which core services to enable
type JetStreamConfig ¶
type JetStreamConfig struct {
Enabled bool `json:"enabled"`
Domain string `json:"domain,omitempty"`
// MaxMemory is the operator's expected server-side
// max_memory_store limit. VerifyJetStreamLimits Warns at boot if
// this exceeds the server's actual AccountInfo.Limits.MaxMemory.
// 0 disables the check (the default for operators who manage
// JetStream sizing entirely via nats.conf without surfacing intent
// in the framework config).
MaxMemory int64 `json:"max_memory,omitempty"`
// MaxFileStore is the operator's expected server-side
// max_file_store limit. Same Warn-on-gap semantics as MaxMemory.
MaxFileStore int64 `json:"max_file_store,omitempty"`
// RetentionPolicy and ReplicationFactor are reserved for future
// per-stream-default plumbing — they're per-stream concepts in
// JetStream, not account-level, so the framework would have to
// route them through StreamConfig defaults rather than at this
// level. Not currently honored anywhere.
RetentionPolicy string `json:"retention_policy,omitempty"`
ReplicationFactor int `json:"replication_factor,omitempty"`
}
JetStreamConfig for JetStream settings. Per #101, fields below are either verification hints (MaxMemory/MaxFileStore) or reserved for future plumbing (Domain/RetentionPolicy/ReplicationFactor); JetStream account limits are exclusively server-side configuration via nats.conf — the nats.go SDK exposes AccountInfo as read-only and offers no client-side mutation surface, so the framework cannot push operator intent. What it CAN do is compare configured-vs-server at boot and Warn on gap (see StreamsManager.VerifyJetStreamLimits) so an operator who set max_file_store: 10GB in the framework config but didn't update nats.conf isn't left wondering why stream-create eventually fails with "insufficient storage resources".
type Loader ¶
type Loader struct {
// contains filtered or unexported fields
}
Loader handles configuration loading with layers and overrides
func (*Loader) EnableValidation ¶
EnableValidation enables or disables configuration validation
func (*Loader) Load ¶
Load loads and merges all configuration layers
Example ¶
ExampleLoader_Load demonstrates loading configuration from multiple layers with environment variable overrides and validation.
package main
import (
"fmt"
"log"
"github.com/c360studio/semstreams/config"
)
func main() {
loader := config.NewLoader()
// Add base configuration layer
loader.AddLayer("testdata/base.json")
// Add environment-specific overrides
loader.AddLayer("testdata/production.json")
// Enable validation to catch errors early
loader.EnableValidation(true)
// Load merged configuration
cfg, err := loader.Load()
if err != nil {
log.Fatal(err)
}
fmt.Println(cfg.Platform.ID)
}
Output: test-platform
Example (EnvironmentOverrides) ¶
ExampleLoader_Load_environmentOverrides demonstrates using environment variables to override configuration values at runtime.
package main
import (
"fmt"
"log"
"github.com/c360studio/semstreams/config"
)
func main() {
// Set environment variables (in real usage, these would be set externally)
// export STREAMKIT_PLATFORM_ID="prod-cluster-01"
// export STREAMKIT_NATS_URLS="nats://server1:4222,nats://server2:4222"
loader := config.NewLoader()
loader.AddLayer("testdata/base.json")
cfg, err := loader.Load()
if err != nil {
log.Fatal(err)
}
// Platform ID and NATS URLs can be overridden via environment
fmt.Printf("Platform: %s\n", cfg.Platform.ID)
fmt.Printf("NATS URLs: %v\n", cfg.NATS.URLs)
}
Output:
func (*Loader) LoadFromBytes ¶
LoadFromBytes loads configuration from JSON bytes. This is useful when you need to pre-process the configuration (e.g., environment variable expansion) before loading.
The data is validated and merged with defaults, just like LoadFile.
type Manager ¶
type Manager struct {
// contains filtered or unexported fields
}
Manager provides centralized configuration management with channel-based updates
Example ¶
ExampleManager demonstrates the complete lifecycle of dynamic configuration management with NATS KV watching.
package main
import (
"fmt"
)
func main() {
// This example shows the complete pattern, but cannot run without NATS
// In real usage:
// 1. Load initial configuration
// loader := config.NewLoader()
// loader.AddLayer("config/base.json")
// cfg, err := loader.Load()
// 2. Create Manager with NATS client
// cm, err := config.NewConfigManager(cfg, natsClient, logger)
// if err != nil {
// log.Fatal(err)
// }
// 3. Start watching for changes
// ctx := context.Background()
// if err := cm.Start(ctx); err != nil {
// log.Fatal(err)
// }
// defer cm.Stop(5 * time.Second)
// 4. Subscribe to configuration changes
// updates := cm.OnChange("components.*")
// go func() {
// for update := range updates {
// log.Printf("Component config changed: %s = %v",
// update.Key, update.Value)
// }
// }()
// 5. Push local changes to NATS KV
// safeConfig := cm.GetConfig()
// safeConfig.Update(func(cfg *config.Config) {
// cfg.Components["new-component"] = config.ComponentConfig{
// Type: "processor/json_map",
// Enabled: true,
// }
// })
// cm.PushToKV(ctx)
fmt.Println("Dynamic configuration management")
}
Output: Dynamic configuration management
func NewConfigManager ¶
func NewConfigManager(cfg *Config, natsClient *natsclient.Client, logger *slog.Logger) (*Manager, error)
NewConfigManager creates a new configuration manager
func (*Manager) DeleteComponentFromKV ¶
DeleteComponentFromKV deletes a component's configuration from NATS KV. This should be called when a component is removed (e.g., during undeploy). PushToKV only puts keys that exist in memory - it doesn't delete removed keys.
This method applies the removal to the in-memory config synchronously (the engine pattern), so a runtime caller invokes only this method to remove a component and the ComponentManager reconciles (tears down) it. The Delete generates a watcher event at a fresh revision the underlying API doesn't expose, so no watermark bump: the delete event arrives and is handled by handleUpdate — either external (applies the already-idempotent delete) or, when a later engine write has raised the high-water above it, engine-owned (skips the redundant re-apply but STILL notifies, gh#388). Either path reconciles the teardown.
func (*Manager) GetConfig ¶
func (cm *Manager) GetConfig() *SafeConfig
GetConfig returns the current configuration
func (*Manager) OnChange ¶
OnChange subscribes to configuration changes matching the pattern Returns a channel that receives updates when configuration changes Pattern examples:
- "services.metrics" - exact match
- "services.*" - all services
- "components.*" - all components
- "components.udp-*" - components starting with udp-
Example ¶
ExampleManager_OnChange demonstrates subscribing to specific configuration change patterns.
package main
import (
"fmt"
)
func main() {
// Assume we have a running Manager
// cm := getConfigManager()
// Subscribe to all service configuration changes
// serviceUpdates := cm.OnChange("services.*")
// Subscribe to specific component changes
// componentUpdates := cm.OnChange("components.my-component")
// Subscribe to platform configuration
// platformUpdates := cm.OnChange("platform")
// Process updates
// go func() {
// for update := range serviceUpdates {
// log.Printf("Service updated: %s", update.Key)
// // React to configuration change
// handleServiceUpdate(update)
// }
// }()
fmt.Println("Subscribed to configuration changes")
}
Output: Subscribed to configuration changes
func (*Manager) PushToKV ¶
PushToKV pushes the current configuration to NATS KV This is useful for initial setup or config synchronization
Example ¶
ExampleManager_PushToKV demonstrates pushing local configuration changes to NATS KV for distribution to other instances.
package main
import (
"fmt"
)
func main() {
// This demonstrates the pattern for pushing config updates
// Get the safe config wrapper
// safeConfig := cm.GetConfig()
// Make local changes
// safeConfig.Update(func(cfg *config.Config) {
// cfg.Platform.LogLevel = "debug"
// cfg.Components["processor-1"].Enabled = false
// })
// Push changes to NATS KV
// ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
// defer cancel()
//
// if err := cm.PushToKV(ctx); err != nil {
// log.Printf("Failed to push config: %v", err)
// }
// Other instances watching the KV will receive the updates
fmt.Println("Configuration pushed to NATS KV")
}
Output: Configuration pushed to NATS KV
func (*Manager) PutComponentToKV ¶
func (cm *Manager) PutComponentToKV(ctx context.Context, name string, compConfig types.ComponentConfig) error
PutComponentToKV writes a single component's configuration to NATS KV. This is more efficient than PushToKV when only one component has changed, and avoids race conditions with KV watchers when multiple operations are in flight.
This method makes the engine pattern (write KV → apply in-memory → bump watermark) self-contained: it applies the component to the in-memory config synchronously, so a runtime caller invokes only this method to add a component and the ComponentManager reconciles (spawns) it. The revision returned by KV.Put is captured into engineHighWaterRev so the watcher's handleUpdate skips the redundant re-apply of the resulting event — but still notifies subscribers (gh#388). KV-write is first so a failed Put leaves in-memory state untouched.
func (*Manager) Stop ¶
Stop stops watching for configuration changes
Example ¶
ExampleManager_Stop demonstrates graceful shutdown of Manager.
package main
import (
"fmt"
)
func main() {
// Assume we have a running Manager
// cm := getConfigManager()
// Graceful shutdown with timeout
// timeout := 5 * time.Second
// if err := cm.Stop(timeout); err != nil {
// log.Printf("Manager shutdown error: %v", err)
// }
// Stop is idempotent - safe to call multiple times
// cm.Stop(timeout) // No error
fmt.Println("Manager stopped gracefully")
}
Output: Manager stopped gracefully
func (*Manager) WatchModelRegistry ¶
WatchModelRegistry returns a channel that emits the latest *model.Registry whenever the model_registry KV key changes. The channel is buffered (cap 1); slow consumers see the most recent registry on their next read — intermediate updates coalesce.
Use this for external library consumers that hold their own *model.Registry alongside the semstreams runtime and need to refresh it on KV change. semstreams components do NOT need this — they're restarted by ComponentManager when their factory declares component.DepModelRegistry.
See model.Watch for a one-line consumer pattern.
The channel closes when the manager Stop()s.
type MinimalConfig ¶
type MinimalConfig struct {
Platform PlatformConfig `json:"platform"`
NATS NATSConfig `json:"nats"`
Services CoreServicesConfig `json:"services"`
}
MinimalConfig represents the core application configuration This is a simplified version focusing only on essential services
Example ¶
ExampleMinimalConfig demonstrates using the simplified MinimalConfig for basic StreamKit applications.
package main
import (
"fmt"
)
func main() {
// MinimalConfig provides a simplified configuration structure
// for applications that don't need the full Config complexity
// Load minimal configuration
// cfg, err := config.LoadMinimalConfig("config/minimal.json")
// if err != nil {
// log.Fatal(err)
// }
// Access core settings
// platformID := cfg.Platform.ID
// natsURLs := cfg.NATS.URLs
// messageLoggerEnabled := cfg.Services.MessageLogger
// MinimalConfig includes:
// - Platform configuration (ID, environment, logging)
// - NATS connection settings
// - Core service toggles (message logger, discovery)
fmt.Println("Minimal configuration for simple applications")
}
Output: Minimal configuration for simple applications
func LoadMinimalConfig ¶
func LoadMinimalConfig(path string) (*MinimalConfig, error)
LoadMinimalConfig loads configuration from a file
func (*MinimalConfig) ToJSON ¶
func (c *MinimalConfig) ToJSON() (string, error)
ToJSON converts config to JSON string for debugging
func (*MinimalConfig) Validate ¶
func (c *MinimalConfig) Validate() error
Validate checks if the minimal config is valid
type NATSConfig ¶
type NATSConfig struct {
URLs []string `json:"urls,omitempty"`
MaxReconnects int `json:"max_reconnects,omitempty"`
ReconnectWait time.Duration `json:"reconnect_wait,omitempty"`
Username string `json:"username,omitempty"`
Password string `json:"password,omitempty"`
Token string `json:"token,omitempty"`
TLS NATSTLSConfig `json:"tls,omitempty"`
JetStream JetStreamConfig `json:"jetstream,omitempty"`
}
NATSConfig defines NATS connection settings
type NATSTLSConfig ¶
type NATSTLSConfig struct {
Enabled bool `json:"enabled"`
CertFile string `json:"cert_file,omitempty"`
KeyFile string `json:"key_file,omitempty"`
CAFile string `json:"ca_file,omitempty"`
}
NATSTLSConfig for secure NATS connections
type PlatformConfig ¶
PlatformConfig is an alias for platform.Config.
PlatformConfig was promoted to its own leaf package so message/ (and any future consumer of platform identity) can reference it without importing all of config. Keeping the alias preserves config.PlatformConfig usage at every call site that already had it. New code should prefer platform.Config directly. Removing the alias would be a mechanical rename across ~10 files; defer until there's a reason.
type PortDefinition ¶
type PortDefinition = component.PortDefinition
PortDefinition is the canonical component port definition.
type PortsConfig ¶
type PortsConfig = component.PortConfig
PortsConfig is the canonical component port configuration.
type SafeConfig ¶
type SafeConfig struct {
// contains filtered or unexported fields
}
SafeConfig provides thread-safe access to configuration
func NewSafeConfig ¶
func NewSafeConfig(cfg *Config) *SafeConfig
NewSafeConfig creates a new thread-safe config wrapper
func (*SafeConfig) Get ¶
func (sc *SafeConfig) Get() *Config
Get returns a deep copy of the current configuration
Example ¶
ExampleSafeConfig_Get demonstrates thread-safe configuration access. The Get method returns a deep copy, preventing accidental mutations.
package main
import (
"fmt"
)
func main() {
// Assume we have a Manager instance
// safeConfig := configManager.GetConfig()
// Get returns a deep copy - safe to use without locks
// cfg := safeConfig.Get()
// Read configuration values
// platformID := cfg.Platform.ID
// natsURLs := cfg.NATS.URLs
// The returned config is a copy, so modifications don't affect
// the shared state
// cfg.Platform.ID = "modified" // Only affects this copy
fmt.Println("Thread-safe configuration access")
}
Output: Thread-safe configuration access
func (*SafeConfig) Mutate ¶
func (sc *SafeConfig) Mutate(fn func(*Config) error) error
Mutate atomically applies a read-modify-write to the configuration. It holds the write lock across the WHOLE clone → mutate → validate → swap sequence, so two concurrent mutations cannot lose one another's change (gh#515). Without this, a caller doing Get() (a clone) → mutate → Update() races another doing the same: each starts from the same base and the second swap clobbers the first. -race does not flag it — each Get/Update is individually locked; the atomicity violation is at the compound level.
fn receives a private deep copy it may mutate freely; returning a non-nil error aborts the mutation with no change. The result is validated before the swap.
Re-entrancy contract: fn MUST NOT call any SafeConfig method (Get/Update/Mutate) — it operates only on the draft it is handed, or it self-deadlocks on sc.mu. Do post-mutation side effects (KV writes, subscriber notification) AFTER Mutate returns; capture any value they need into a variable inside fn.
Unlike Update, validation runs INSIDE the write lock (it must, to keep the whole RMW atomic). Validate does filesystem I/O (os.Stat) when TLS is configured, so a mutation briefly blocks concurrent Get() readers — acceptable because config mutations are rare and off the hot path.
func (*SafeConfig) Update ¶
func (sc *SafeConfig) Update(cfg *Config) error
Update atomically updates the configuration after validation
Example ¶
ExampleSafeConfig_Update demonstrates atomic configuration updates.
package main
import (
"fmt"
)
func main() {
// Assume we have a Manager instance
// safeConfig := configManager.GetConfig()
// Update configuration atomically
// safeConfig.Update(func(cfg *config.Config) {
// // Enable a component
// if comp, exists := cfg.Components["my-component"]; exists {
// comp.Enabled = true
// cfg.Components["my-component"] = comp
// }
// })
fmt.Println("Configuration updated atomically")
}
Output: Configuration updated atomically
type StreamConfig ¶
type StreamConfig struct {
Subjects []string `json:"subjects"` // Subjects captured by this stream
Storage string `json:"storage,omitempty"` // "file" or "memory" (default: file)
MaxAge string `json:"max_age,omitempty"` // Required: message TTL (e.g., "168h", "7d")
MaxBytes int64 `json:"max_bytes,omitempty"` // Required: max storage in bytes; must be > 0
Retention string `json:"retention,omitempty"` // "limits", "interest", "workqueue" (default: limits)
Replicas int `json:"replicas,omitempty"` // Replication factor (default: 1)
// Discard is the operator's choice of what happens when the stream reaches
// MaxBytes or MaxAge: StreamDiscardOld evicts the oldest messages,
// StreamDiscardNew refuses the write (producers see NATS 503
// err_code=10077). Required for an ordinary stream — it was hardcoded to
// DiscardOld before, so the policy was never the operator's choice.
Discard string `json:"discard,omitempty"`
// Duplicates is the server-side duplicate-detection window for the
// Nats-Msg-Id header (e.g. "2m", "30m", "1h"). Producers using
// Client.PublishToStreamWithMsgID rely on this window to collapse
// redeliveries of the same logical event (ADR-055 §5, "T1"). Empty
// leaves the window unset, so the NATS server applies its default of
// 2 minutes — adequate for steady-state redelivery but shorter than a
// restart/recovery replay horizon. Streams whose producers need
// dedup across that horizon set this explicitly. Must be <= MaxAge:
// the NATS server REJECTS (does not clamp) an explicit window larger
// than MaxAge, so createStream clamps it down to MaxAge with a warning
// rather than letting EnsureStreams abort boot.
Duplicates string `json:"duplicates,omitempty"`
}
StreamConfig defines configuration for a JetStream stream.
MaxAge, MaxBytes, and Discard are REQUIRED for an ordinary stream and are never supplied by a framework default. A bound the operator never chose is indistinguishable in the operator surface from one they did, which is the condition the bounds contract exists to end. The two ways out are both explicit declarations of their own: `archival_streams` for a stream whose contract is permanence, and `stream_migration_overrides` for a time-limited bridge. See ValidateStreamDeclarations.
type StreamConfigs ¶
type StreamConfigs map[string]StreamConfig
StreamConfigs is a map of stream name to configuration.
type StreamExceptionReport ¶
type StreamExceptionReport struct {
// MigrationOverrides are time-limited bridges. Every entry here is a
// scheduled future readiness failure. Sorted by stream name.
MigrationOverrides []StreamMigrationOverrideStatus
// Archival are permanent, declared exceptions. Sorted by stream name.
Archival []ArchivalStreamStatus
}
StreamExceptionReport is readiness's account of every ordinary stream admitted WITHOUT finite bounds, split by the kind of exception that admitted it.
The split is structural, not cosmetic. A single list with a kind flag would let an operator surface render "permanent" and "expires in March" through the same widget, and the whole reason archival exists as its own classification is that blurring those two is what trains people to renew bridges without reading them.
func ValidateStreamDeclarations ¶
func ValidateStreamDeclarations(cfg *Config) (StreamExceptionReport, error)
ValidateStreamDeclarations resolves every ordinary stream this configuration would provision and enforces the bounds contract on it, returning readiness's account of every stream admitted without finite bounds.
It is pure and I/O-free, which is what lets Config.Validate run it: without that, `semstreams --validate` would print "✓ Configuration is valid" for a configuration that hard-fails at boot.
type StreamMigrationOverride ¶
type StreamMigrationOverride struct {
// Owner is the team, component, or person accountable for completing the
// migration. Required — an exception nobody owns never ends.
Owner string `json:"owner"`
// Expires is the date the bridge ends, as "2006-01-02" or a full RFC3339
// timestamp. A date-only value is INCLUSIVE of that day: "2026-09-30" expires
// at the end of 2026-09-30 UTC, which is how an operator reads it. Required.
Expires string `json:"expires"`
// Reason is optional prose for the next reader. Unlike ArchivalStream.Reason
// it is not required, because an override's justification is its expiry.
Reason string `json:"reason,omitempty"`
}
StreamMigrationOverride admits ONE existing ordinary stream that predates the bounds contract, as a named, time-limited exception. It is a migration bridge and nothing else: a stream whose contract is permanence is ARCHIVAL and must use that classification, so the two never share an instrument.
Owner and Expires are both required. An override with no expiry is rejected at validation, because an override's value comes from being rare and alarming and an open-ended one is neither.
type StreamMigrationOverrideStatus ¶
type StreamMigrationOverrideStatus struct {
Stream string // the admitted ordinary stream
Owner string // who is accountable for finishing the migration
Reason string // optional prose, empty when not declared
Expires time.Time // when readiness starts failing
Remaining time.Duration // time left, at the moment readiness was evaluated
}
StreamMigrationOverrideStatus is one active, time-limited exception as readiness reports it.
func ExpiredMigrationOverrides ¶
func ExpiredMigrationOverrides(cfg *Config, now time.Time) []StreamMigrationOverrideStatus
ExpiredMigrationOverrides returns every migration override whose expiry has passed as of now, so a RUNNING instance can report a bridge that ended while it was up.
It exists because expiry is otherwise evaluated only at configuration validation and at provisioning — both boot-time. An instance that started before the deadline would otherwise run indefinitely past it with nothing saying so, and the whole value of a bridge is that it ends.
Enforcement stays at boot. This function REPORTS, and that split is deliberate: the stream a lapsed override admits is still working, and taking a healthy fleet out of service simultaneously because a calendar date passed would be a self-inflicted outage over a hygiene failure. The refusal lands at the next boot, which is when an operator can act on it anyway.
A malformed override (no expiry, unparseable date) is NOT returned here: it is rejected at validation and never reaches a running instance.
type StreamMigrationOverrides ¶
type StreamMigrationOverrides map[string]StreamMigrationOverride
StreamMigrationOverrides maps ordinary stream name to its migration override.
type StreamsManager ¶
type StreamsManager struct {
// contains filtered or unexported fields
}
StreamsManager handles JetStream stream creation and management.
func NewStreamsManager ¶
func NewStreamsManager(natsClient *natsclient.Client, logger *slog.Logger) *StreamsManager
NewStreamsManager creates a new StreamsManager.
func (*StreamsManager) EnsureStreams ¶
func (sm *StreamsManager) EnsureStreams(ctx context.Context, cfg *Config) error
EnsureStreams creates all required JetStream streams based on: 1. System streams (LOGS for out-of-band logging) 2. Explicit streams defined in config.Streams (highest priority) 3. Streams derived from component JetStream output ports
Every resolved ordinary stream must carry an explicitly declared finite MaxAge, a finite MaxBytes, and a discard policy, unless an archival declaration or an active migration override admits it. That check runs BEFORE any stream is created, so a configuration missing a bound fails closed with a complete diagnostic rather than provisioning half its streams and then stopping — a partially-provisioned account is harder to reason about than one that never started.
func (*StreamsManager) VerifyJetStreamLimits ¶
func (sm *StreamsManager) VerifyJetStreamLimits(ctx context.Context, cfg *Config) error
VerifyJetStreamLimits reads the operator's MaxMemory / MaxFileStore hints from cfg.NATS.JetStream and logs a Warn for each value that exceeds the server's actual account limit. JetStream account limits are server-side configuration (nats.conf or jetstream-domain configuration); the nats.go SDK exposes AccountInfo as read-only, so the framework cannot push the operator's intent — but it CAN surface the gap loudly so an operator who set max_file_store: 10GB in the framework config but didn't update nats.conf isn't left wondering why stream-create fails with "insufficient storage resources" at runtime (#101). Zero or unset values skip the check.
Best-effort: a failure to fetch AccountInfo (server down, JetStream disabled, AccountInfo unsupported) logs at Debug and returns nil — the check is diagnostic, not gating. EnsureStreams runs anyway and any actual capacity miss surfaces as a CreateStream error there.
type Update ¶
type Update struct {
Path string // Changed path (e.g., "services.metrics")
Config *SafeConfig // Full latest configuration
}
Update represents a configuration change notification