Documentation
¶
Overview ¶
Example ¶
The README shows this under "As a Library". It lived only in the README and named three packages that do not exist -- pkg/sink/stdout and pkg/buffer are really pkg/comm/sink/stdout and pkg/comm/buffer -- and passed no formatter to a constructor that requires one. Nothing caught it because the module path in go.mod matched no repository, so the example could not have been compiled by anyone, including whoever wrote it.
It is a compiled example now, in an external test package so the imports read the way a caller's would. There is no "Output:" comment on purpose: `go test` compiles an example without one but does not run it, which is what this needs -- Start blocks on a real source, and the thing worth checking is that the code a new user copies still builds.
package main
import (
"context"
"io"
"time"
"github.com/gsoultan/hermod"
"github.com/gsoultan/hermod/pkg/comm/buffer"
"github.com/gsoultan/hermod/pkg/comm/formatter/json"
"github.com/gsoultan/hermod/pkg/comm/sink/stdout"
"github.com/gsoultan/hermod/pkg/engine"
"github.com/gsoultan/hermod/pkg/engine/config"
)
// The README shows this under "As a Library". It lived only in the README and
// named three packages that do not exist -- pkg/sink/stdout and pkg/buffer are
// really pkg/comm/sink/stdout and pkg/comm/buffer -- and passed no formatter to
// a constructor that requires one. Nothing caught it because the module path in
// go.mod matched no repository, so the example could not have been compiled by
// anyone, including whoever wrote it.
//
// It is a compiled example now, in an external test package so the imports read
// the way a caller's would. There is no "Output:" comment on purpose: `go test`
// compiles an example without one but does not run it, which is what this
// needs -- Start blocks on a real source, and the thing worth checking is that
// the code a new user copies still builds.
func main() {
source := exampleSource{}
sinks := []hermod.Sink{stdout.NewStdoutSink(json.NewJSONFormatter())}
buf := buffer.NewRingBuffer(1024)
eng := engine.NewEngine(source, sinks, buf)
eng.SetConfig(config.Config{
MaxRetries: 5,
RetryInterval: 200 * time.Millisecond,
})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
_ = eng.Start(ctx)
}
// exampleSource stands in for a real source. Read returning io.EOF is what a
// finite source does once it is drained.
type exampleSource struct{}
func (exampleSource) Read(context.Context) (hermod.Message, error) { return nil, io.EOF }
func (exampleSource) Ack(context.Context, hermod.Message) error { return nil }
func (exampleSource) Ping(context.Context) error { return nil }
func (exampleSource) Close() error { return nil }
Output:
Index ¶
- Constants
- func NewDefaultLogger() hermod.Logger
- func PendingOverReleaseCount() int64
- func ResetTraceStepsDroppedCount()
- func TraceStepsDroppedCount() int64
- type BackpressureStrategy
- type CheckpointManager
- type Config
- type Engine
- func (e *Engine) AcquireNode(ctx context.Context, nodeID string) error
- func (e *Engine) Checkpoint(ctx context.Context) error
- func (e *Engine) DeadLetterNodeFailure(ctx context.Context, nodeID string, msg hermod.Message, cause error) bool
- func (e *Engine) DeadLetterOrphanedMessage(ctx context.Context, nodeID string, msg hermod.Message, cause error) bool
- func (e *Engine) DetectAnomaly(duration time.Duration) bool
- func (e *Engine) DrainDLQ(ctx context.Context) error
- func (e *Engine) GetConcurrency() int
- func (e *Engine) GetSinkConfigs() []config.SinkConfig
- func (e *Engine) GetSinks() []hermod.Sink
- func (e *Engine) GetSource() hermod.Source
- func (e *Engine) GetStatus() telemetry.StatusUpdate
- func (e *Engine) HardStop()
- func (e *Engine) IsDryRun() bool
- func (e *Engine) IsSafeMode() bool
- func (e *Engine) LastMsgTime() time.Time
- func (e *Engine) RecordTraceStep(ctx context.Context, msg hermod.Message, nodeID string, start time.Time, ...)
- func (e *Engine) RecordTraceStepSnapshot(ctx context.Context, msg hermod.Message, nodeID string, start time.Time, ...)
- func (e *Engine) ReleaseNode(nodeID string)
- func (e *Engine) SetCheckpointHandler(fn func(context.Context, map[string]string) error)
- func (e *Engine) SetConfig(cfg config.Config)
- func (e *Engine) SetDeadLetterSink(snk hermod.Sink)
- func (e *Engine) SetIDs(workflowID string, sourceID string, sinkIDs []string)
- func (e *Engine) SetLogger(l hermod.Logger)
- func (e *Engine) SetOnStall(fn func(reason string))
- func (e *Engine) SetOnStatusChange(fn func(telemetry.StatusUpdate))
- func (e *Engine) SetOutboxStorage(outbox hermod.OutboxStorage)
- func (e *Engine) SetRouter(r RouterFunc)
- func (e *Engine) SetSafeMode(enabled bool)
- func (e *Engine) SetSinkConfigs(configs []config.SinkConfig)
- func (e *Engine) SetSinkIDs(ids []string)
- func (e *Engine) SetSinkTypes(types []string)
- func (e *Engine) SetSourceConfig(cfg config.SourceConfig)
- func (e *Engine) SetSourceID(id string)
- func (e *Engine) SetTraceRecorder(tr hermod.TraceRecorder)
- func (e *Engine) SetValidator(v schema.Validator)
- func (e *Engine) SetWorkflowID(id string)
- func (e *Engine) SimulateFailure(duration time.Duration)
- func (e *Engine) Start(ctx context.Context) error
- func (e *Engine) UpdateConcurrency(n int)
- func (e *Engine) UpdateEdgeMetric(sourceNodeID string, targetNodeID string, count uint64)
- func (e *Engine) UpdateNodeErrorMetric(nodeID string, count uint64)
- func (e *Engine) UpdateNodeMetric(nodeID string, count uint64)
- func (e *Engine) UpdateNodeSample(nodeID string, data map[string]any)
- func (e *Engine) UpdateSinkConfig(sinkID string, update func(*config.SinkConfig))
- func (e *Engine) WillTrace(msg hermod.Message) bool
- type RoutedMessage
- type RouterFunc
- type Runner
- type SinkConfig
- type SourceConfig
- type StatusUpdate
Examples ¶
Constants ¶
const DefaultLagWarnBytes uint64 = 256 << 20
DefaultLagWarnBytes is how much un-acknowledged WAL a source may retain before it is reported. A replication slot that stops advancing pins WAL on the *source* database, so an unnoticed stall does not degrade Hermod — it fills the customer's primary and takes it down. 256 MiB is small enough to be a warning rather than an incident and large enough not to fire on a normal backlog.
const DefaultStallThreshold = 60 * time.Second
DefaultStallThreshold is how long a pipeline may hold outstanding work without completing any of it before it is treated as wedged rather than busy. Slow sinks and long retry backoffs are normal, so this is deliberately generous; the failure it exists to catch lasts indefinitely, not seconds.
const DefaultStreamSilenceInterval = 10 * time.Second
DefaultStreamSilenceInterval is how often a source's push stream is sampled for silence when the engine config does not say.
const MetaDeadLettered = "_hermod_dead_lettered"
MetaDeadLettered marks a message the workflow already parked in the dead-letter sink.
A node that fails is dead-lettered where it failed, and the traversal then resolves no sink for that message. It reaches the engine as an empty target list — the same shape as a message nothing ever handled — so the engine parked it a second time. One failed node produced two identical rows in the queue, differing only in which metadata the second pass overwrote, and whoever drained the queue had to work out they were the same event.
The marker says the message is already preserved: acknowledge the source and do not park it again. It is set only after a park that succeeded, so a refused park still takes the un-acknowledged path and is redelivered.
const MetaDeliveredInline = "_hermod_delivered_inline"
MetaDeliveredInline marks a message that a sink node already wrote itself.
A sink node with `sequential: true` writes inline and deliberately routes nothing, so the router hands the engine an empty target list. That is indistinguishable, at the engine, from a workflow that resolved none of its sinks — which is data loss and is handled by refusing to acknowledge. Applied to a message that was in fact delivered, that refusal pins the source: a PostgreSQL replication slot stops advancing and retains WAL for the life of the workflow, while every message is re-read on the next start.
The marker is what tells those two cases apart. It is set only after the inline write succeeded, so a failed write still takes the un-acknowledged path and is redelivered.
Variables ¶
This section is empty.
Functions ¶
func NewDefaultLogger ¶
NewDefaultLogger creates a DefaultLogger with stderr output and timestamps.
func PendingOverReleaseCount ¶
func PendingOverReleaseCount() int64
PendingOverReleaseCount reports how many times a pendingMessage was released more often than it was referenced. Any non-zero value indicates unbalanced release bookkeeping somewhere in the writer paths.
func ResetTraceStepsDroppedCount ¶ added in v1.8.1
func ResetTraceStepsDroppedCount()
ResetTraceStepsDroppedCount zeroes the counter, for tests.
func TraceStepsDroppedCount ¶ added in v1.8.1
func TraceStepsDroppedCount() int64
TraceStepsDroppedCount reports how many trace steps were discarded because the recorder could not keep up.
Types ¶
type BackpressureStrategy ¶
type BackpressureStrategy = config.BackpressureStrategy
type CheckpointManager ¶
type CheckpointManager struct {
// contains filtered or unexported fields
}
func NewCheckpointManager ¶
func (*CheckpointManager) Checkpoint ¶
func (m *CheckpointManager) Checkpoint(ctx context.Context) error
func (*CheckpointManager) IsInCheckpoint ¶
func (m *CheckpointManager) IsInCheckpoint() bool
type Config ¶
Re-export common types from sub-packages for backward compatibility
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns the default configuration for the Engine.
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine orchestrates the data flow from Source to Sinks. It acts as a Facade, delegating complex tasks to internal components.
func (*Engine) AcquireNode ¶
AcquireNode attempts to acquire a concurrency slot for a specific node. This implements node-level backpressure.
func (*Engine) DeadLetterNodeFailure ¶
func (e *Engine) DeadLetterNodeFailure(ctx context.Context, nodeID string, msg hermod.Message, cause error) bool
DeadLetterNodeFailure sends a message that a workflow node could not process to the dead-letter sink, and reports whether it went anywhere.
Node failures did not reach the dead-letter sink at all. The traversal logged "Node %s failed" and released the message, so a workflow with a dead-letter sink configured still lost every message a transformation, condition or any other node rejected — the DLQ covered validation and sink failures only, and the gap was invisible because the log line looked like handling.
Returning false means nothing was configured to catch it, which is the caller's cue to say so rather than imply the message was kept.
func (*Engine) DeadLetterOrphanedMessage ¶ added in v1.8.1
func (e *Engine) DeadLetterOrphanedMessage(ctx context.Context, nodeID string, msg hermod.Message, cause error) bool
DeadLetterOrphanedMessage parks a failed message that has no source left holding it, and reports whether that worked.
It ignores dry-run, which every other write path honours. The reason the others can decline is that declining preserves the message: nothing is acknowledged, so it stays on the source and the next run sees it again. A resumed message has no source — the suspended row is deleted as soon as the resume returns — so declining here would not preserve it, it would destroy it. Losing data is the one outcome dry-run exists to prevent, so this is the one place a dry run writes.
func (*Engine) DrainDLQ ¶
DrainDLQ attempts to wrap the current source with a PrioritySource to drain DLQ messages.
func (*Engine) GetConcurrency ¶
func (*Engine) GetSinkConfigs ¶
func (e *Engine) GetSinkConfigs() []config.SinkConfig
func (*Engine) GetStatus ¶
func (e *Engine) GetStatus() telemetry.StatusUpdate
GetStatus returns the current status of the engine.
func (*Engine) HardStop ¶
func (e *Engine) HardStop()
HardStop attempts to forcefully and quickly stop the engine.
func (*Engine) IsDryRun ¶ added in v1.8.1
IsDryRun reports whether this workflow is running as a preview: reading and processing normally, but writing to no sink and acknowledging nothing.
func (*Engine) IsSafeMode ¶
func (*Engine) LastMsgTime ¶
LastMsgTime returns the time of the last message received from the source.
func (*Engine) RecordTraceStep ¶
func (*Engine) RecordTraceStepSnapshot ¶ added in v1.8.1
func (e *Engine) RecordTraceStepSnapshot(ctx context.Context, msg hermod.Message, nodeID string, start time.Time, before, after map[string]any, err error)
RecordTraceStepSnapshot records a step whose payload was captured earlier, for a caller whose work mutates the message it is tracing.
The router is why this exists. For a node-graph workflow the engine's router *is* the traversal — setupWorkflowRouter walks the whole DAG inside it — so a snapshot taken when it returns is the pipeline's output, while the step's timestamp is when routing began. The viewer orders steps by timestamp and rebuilds each "before" from the previous "after", so the two together put the final payload between the message arriving and the first node running.
Pass the snapshot taken before the work started and the halves agree. Guard the capture with WillTrace so an untraced workflow pays nothing.
func (*Engine) ReleaseNode ¶
ReleaseNode releases a concurrency slot for a specific node.
func (*Engine) SetCheckpointHandler ¶
func (*Engine) SetDeadLetterSink ¶
func (*Engine) SetOnStall ¶
SetOnStall registers a supervisor for this engine. The watchdog calls it once per stall episode, on its own goroutine, when the pipeline is holding work it has stopped completing.
This exists because a data pipeline has many ways to wedge and only one reliable cure. Three separate wedge causes were found and fixed by hand (a double-pooled pendingMessage, a circuit-breaker self-deadlock, a source that retired itself silently) and the symptom still recurred. Restarting the workflow recovered it every time, losing nothing, because the replication slot holds un-acknowledged WAL until the new engine re-reads it. So rather than assume the last bug was the last bug, supervise: detect the stall and rebuild the engine, the way a process supervisor or a liveness probe would.
func (*Engine) SetOnStatusChange ¶
func (e *Engine) SetOnStatusChange(fn func(telemetry.StatusUpdate))
func (*Engine) SetOutboxStorage ¶
func (e *Engine) SetOutboxStorage(outbox hermod.OutboxStorage)
func (*Engine) SetRouter ¶
func (e *Engine) SetRouter(r RouterFunc)
func (*Engine) SetSafeMode ¶
func (*Engine) SetSinkConfigs ¶
func (e *Engine) SetSinkConfigs(configs []config.SinkConfig)
func (*Engine) SetSinkIDs ¶
func (*Engine) SetSinkTypes ¶
func (*Engine) SetSourceConfig ¶
func (e *Engine) SetSourceConfig(cfg config.SourceConfig)
func (*Engine) SetSourceID ¶
func (*Engine) SetTraceRecorder ¶
func (e *Engine) SetTraceRecorder(tr hermod.TraceRecorder)
func (*Engine) SetValidator ¶
func (*Engine) SetWorkflowID ¶
func (*Engine) SimulateFailure ¶
func (*Engine) UpdateConcurrency ¶
func (*Engine) UpdateEdgeMetric ¶
func (*Engine) UpdateNodeErrorMetric ¶
func (*Engine) UpdateNodeMetric ¶
func (*Engine) UpdateNodeSample ¶
UpdateNodeSample stores the latest payload sample for a node. Callers must pass an independent map (e.g. produced by Registry.getConsistentData) that is not mutated afterwards; the value is stored as-is to avoid an extra full-payload JSON round-trip on every message.
func (*Engine) UpdateSinkConfig ¶
func (e *Engine) UpdateSinkConfig(sinkID string, update func(*config.SinkConfig))
func (*Engine) WillTrace ¶ added in v1.8.1
WillTrace reports whether a step recorded for this message would be kept.
Callers that have to build a payload snapshot *before* the work they are tracing need this: without it they pay a ToMap() per message on a workflow whose tracing is off, which is the common case.
type RoutedMessage ¶
RoutedMessage represents a message and its target sink index.
type RouterFunc ¶
RouterFunc is a function that routes a message to one or more sinks.
type SinkConfig ¶
type SinkConfig = config.SinkConfig
type SourceConfig ¶
type SourceConfig = config.SourceConfig
type StatusUpdate ¶
type StatusUpdate = telemetry.StatusUpdate
Source Files
Directories