Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type BackpressureStrategy ¶
type BackpressureStrategy string
BackpressureStrategy defines how to handle overflow when a sink is slow.
const ( // BPBlock blocks the source until there is room in the buffer. BPBlock BackpressureStrategy = "block" // BPDropOldest drops the oldest message in the buffer to make room. BPDropOldest BackpressureStrategy = "drop_oldest" // BPDropNewest drops the incoming message if the buffer is full. BPDropNewest BackpressureStrategy = "drop_newest" // BPSampling drops the incoming message with a configured probability. BPSampling BackpressureStrategy = "sampling" // BPSpillToDisk writes the incoming message to disk if the buffer is full. BPSpillToDisk BackpressureStrategy = "spill_to_disk" )
type Config ¶
type Config struct {
MaxRetries int `json:"max_retries"`
RetryInterval time.Duration `json:"retry_interval"`
ReconnectInterval time.Duration `json:"reconnect_interval"`
StatusInterval time.Duration `json:"status_interval"`
PrioritizeDLQ bool `json:"prioritize_dlq"`
DryRun bool `json:"dry_run"`
CheckpointInterval time.Duration `json:"checkpoint_interval"`
TraceSampleRate float64 `json:"trace_sample_rate"` // 0.0 to 1.0
AdaptiveThroughput bool `json:"adaptive_throughput"`
MaxMemoryMB uint64 `json:"max_memory_mb"`
OutboxRelayInterval time.Duration `json:"outbox_relay_interval"`
// MaxInflight bounds the number of messages processed concurrently across the pipeline.
// Keep this conservative to limit memory usage. Defaults to 128.
MaxInflight int `json:"max_inflight"`
// DrainTimeout controls how long to wait for sink writers to drain on shutdown before logging a warning.
// Does not forcibly terminate writers; set to 0 to wait indefinitely.
DrainTimeout time.Duration `json:"drain_timeout"`
// StallThreshold is how long the pipeline may hold outstanding work without
// completing any of it before it is reported as stalled. A wedged pipeline
// is otherwise indistinguishable from an idle one: it keeps reporting
// "running" while delivering nothing. Set to 0 to use the default.
StallThreshold time.Duration `json:"stall_threshold"`
// LagWarnBytes is how much un-acknowledged WAL a source may retain before it
// is reported. Retention accumulates on the source database, so this guards
// someone else's disk, not Hermod's. Set to 0 to use the default.
LagWarnBytes uint64 `json:"lag_warn_bytes"`
// StreamSilenceInterval is how often a source's push stream is checked for
// silence. The threshold itself belongs to the source, which derives it from
// the server's keepalive cadence; this only controls how promptly the
// silence is noticed. Set to 0 to use the default.
StreamSilenceInterval time.Duration `json:"stream_silence_interval"`
}
Config holds configuration for the Engine.
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns the default configuration for the Engine.
The recovery thresholds are operational levers — an incident may call for detecting a stall sooner, and a workflow with a legitimately slow sink for detecting it later — so they are readable from the environment rather than fixed at build time.
type ShutdownBudget ¶
type ShutdownBudget struct {
// Total bounds the whole stop. Keep it under the orchestrator's grace
// period, or the process is killed rather than allowed to finish.
Total time.Duration
// PerEngine bounds stopping one workflow, and the parallel StopAll over all
// of them.
PerEngine time.Duration
// Drain bounds sink writes once shutdown has begun.
Drain time.Duration
// Grace is the extra time writers get to unwind after Drain expires. Their
// write contexts are already cancelled by then, so this only covers
// returning, not working.
Grace time.Duration
}
ShutdownBudget is how long each stage of a graceful stop may take.
These used to be independent magic numbers scattered across the worker, the registry and the engine — 60s for worker cleanup, 45s for StopAll, 35s for a single workflow stop, 10s to drain plus 10s of grace. Nothing related them, and the outermost was double the innermost, so the layers did not nest: a stage could be cut off by a *parent* deadline it had never heard of, halfway through the drain that exists to avoid losing data.
Worse, the total exceeded the orchestrator's patience. Kubernetes sends SIGKILL after terminationGracePeriodSeconds — 30 seconds by default — so a 60-second cleanup was killed mid-drain, discarding exactly the messages the drain protects. A pipeline that stops cleanly on a laptop lost data on every rolling deploy.
One total now governs everything and the stages are derived from it, so the nesting is arithmetic rather than a promise. The default sits below the default grace period with margin to spare.
func Shutdown ¶
func Shutdown() ShutdownBudget
Shutdown returns the budget, honouring HERMOD_SHUTDOWN_TIMEOUT.
Raising it is correct when the orchestrator's grace period has been raised to match; lowering it trades drain completeness for a faster stop. A value that cannot be parsed falls back to the default rather than failing startup — a typo in a tuning knob should not stop a worker from booting.
func (ShutdownBudget) ClampDrain ¶
func (b ShutdownBudget) ClampDrain(configured time.Duration) time.Duration
ClampDrain fits an operator-configured DrainTimeout inside the budget.
SinkConfig.DrainTimeout is user-facing and predates this budget, so it can be set larger than the whole shutdown. Honour it when it fits and clamp when it does not, rather than letting a per-sink setting silently overrun the process-wide deadline.
type SinkConfig ¶
type SinkConfig struct {
MaxRetries int `json:"max_retries"`
RetryInterval time.Duration `json:"retry_interval"`
RetryIntervals []time.Duration `json:"retry_intervals"`
BatchSize int `json:"batch_size"`
BatchTimeout time.Duration `json:"batch_timeout"`
// BatchBytes triggers a flush when the accumulated payload size reaches this threshold.
// Set to 0 to disable byte-based flushing.
BatchBytes int `json:"batch_bytes"`
AdaptiveBatching bool `json:"adaptive_batching"`
Concurrency int `json:"concurrency"`
// Per-key sharding for ordered concurrency
ShardCount int `json:"shard_count"`
ShardKeyMeta string `json:"shard_key_meta"` // when empty, use Message.ID()
// Circuit Breaker settings
CircuitBreakerThreshold int `json:"cb_threshold"`
CircuitBreakerInterval time.Duration `json:"cb_interval"`
CircuitBreakerCoolDown time.Duration `json:"cb_cool_off"`
// Backpressure settings
BackpressureStrategy BackpressureStrategy `json:"backpressure_strategy"`
BackpressureBuffer int `json:"backpressure_buffer"`
SamplingRate float64 `json:"sampling_rate"` // 0.0 to 1.0
// Spill to Disk settings
SpillPath string `json:"spill_path"`
SpillMaxSize int `json:"spill_max_size"`
}
SinkConfig holds configuration for a specific sink.
type SourceConfig ¶
SourceConfig holds configuration for a specific source.
Source Files