registry

package
v1.16.2 Latest Latest
Warning

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

 Go to latest Published: Sep 29, 2026 License: MIT Imports: 61 Imported by: 0

Documentation

Index

Constants

View Source 
const MaxLookupCacheSize = 10000

maxLookupCacheSize bounds the number of entries kept in the in-memory lookup cache so it cannot grow without limit and leak memory.

Variables

This section is empty.

Functions

This section is empty.

Types

type DatabaseLogger 

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

func NewDatabaseLogger 

func NewDatabaseLogger(parentCtx context.Context, s LogCreator, workflowID string, fallback hermod.Logger) *DatabaseLogger

func (*DatabaseLogger) Close 

func (l *DatabaseLogger) Close()

Close stops the background flusher and drains what is buffered. It is safe to call more than once.

func (*DatabaseLogger) Debug 

func (l *DatabaseLogger) Debug(msg string, keysAndValues ...any)

func (*DatabaseLogger) DebugEnabled added in v1.8.1 

func (l *DatabaseLogger) DebugEnabled() bool

DebugEnabled reports whether Debug output is kept.

It exists so a caller can skip building an argument that is about to be discarded. The engine's per-write line measures the message's payload, which for a data-map message means marshalling it to JSON; a logger that cannot answer this is assumed to want the line, so without it that marshal ran for every message written.

func (*DatabaseLogger) Error 

func (l *DatabaseLogger) Error(msg string, keysAndValues ...any)

func (*DatabaseLogger) Flush 

func (l *DatabaseLogger) Flush()

func (*DatabaseLogger) Info 

func (l *DatabaseLogger) Info(msg string, keysAndValues ...any)

func (*DatabaseLogger) UpdateStorage 

func (l *DatabaseLogger) UpdateStorage(s LogCreator)

func (*DatabaseLogger) Warn 

func (l *DatabaseLogger) Warn(msg string, keysAndValues ...any)

type DebuggerEvent 

type DebuggerEvent struct {
	WorkflowID string         `json:"workflow_id"`
	NodeID     string         `json:"node_id"`
	MsgID      string         `json:"msg_id"`
	Data       map[string]any `json:"data"`
	State      string         `json:"state"` // "paused", "resumed", "aborted"
}

type LiveMessage 

type LiveMessage struct {
	WorkflowID string         `json:"workflow_id"`
	NodeID     string         `json:"node_id"`
	Timestamp  time.Time      `json:"timestamp"`
	Data       map[string]any `json:"data"`
	IsError    bool           `json:"is_error"`
	Error      string         `json:"error,omitempty"`
}

type LogCreator 

type LogCreator interface {
	CreateLog(ctx context.Context, log storage.Log) error
	CreateLogs(ctx context.Context, logs []storage.Log) error
}

type PIIStats 

type PIIStats struct {
	Discoveries map[string]uint64 `json:"discoveries"`
	LastUpdated time.Time         `json:"last_updated"`
}

type Registry 

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

func NewRegistry 

func NewRegistry(s storage.Storage, ls ...storage.Storage) *Registry

func (*Registry) ApplyTransformation 

func (r *Registry) ApplyTransformation(ctx context.Context, msg hermod.Message, transType string, config map[string]any) (hermod.Message, error)

func (*Registry) BroadcastLiveMessage 

func (r *Registry) BroadcastLiveMessage(workflowID, nodeID string, msg hermod.Message, isError bool, errMsg string)

func (*Registry) BroadcastLog 

func (r *Registry) BroadcastLog(engineID, level, msg, data string)

func (*Registry) BroadcastStatus 

func (r *Registry) BroadcastStatus(update telemetry.StatusUpdate)

func (*Registry) BrowseSinkTable 

func (r *Registry) BrowseSinkTable(ctx context.Context, cfg factory.SinkConfig, table string, limit int) ([]hermod.Message, error)

func (*Registry) CanTransform added in v1.8.1 

func (r *Registry) CanTransform(transType string) bool

CanTransform reports whether transType names a registered transformer.

Some names are both: "validator" is a transformer and a node executor, and the transformer is the one a preview wants. Callers deciding how to preview a node need to ask this first, or a node that transforms perfectly well gets turned away for being a node.

func (*Registry) Close 

func (r *Registry) Close()

func (*Registry) ContextWithPipelineSnapshot 

func (r *Registry) ContextWithPipelineSnapshot(ctx context.Context) context.Context

ContextWithPipelineSnapshot returns a new context with a shared snapshot pointer for optimizing transformation pipelines by reducing redundant ToMap() calls.

func (*Registry) CreateLog 

func (r *Registry) CreateLog(ctx context.Context, l storage.Log) error

func (*Registry) CreateLogs 

func (r *Registry) CreateLogs(ctx context.Context, logs []storage.Log) error

func (*Registry) CreateSink 

func (r *Registry) CreateSink(ctx context.Context, cfg factory.SinkConfig) (hermod.Sink, error)

func (*Registry) CreateSource 

func (r *Registry) CreateSource(ctx context.Context, cfg factory.SourceConfig) (hermod.Source, error)

func (*Registry) DebuggerCommand 

func (r *Registry) DebuggerCommand(workflowID, msgID, action string)

func (*Registry) DeleteLogs 

func (r *Registry) DeleteLogs(ctx context.Context, filter storage.LogFilter) error

func (*Registry) DeleteWorkflowTraces added in v1.10.0 

func (r *Registry) DeleteWorkflowTraces(ctx context.Context, workflowID string)

DeleteWorkflowTraces removes the traces of a workflow that has been deleted.

Here rather than in storage.DeleteWorkflow because traces live in the log store, and SetLogStorage may point that at a different database from the one holding workflows — a cascade inside the workflow store would delete from the wrong place and silently leave the real rows behind. The Registry is the object that knows both.

Best effort by design: the hourly sweep reclaims anything this misses, so a failure here costs space until the next tick rather than failing the delete the operator actually asked for.

func (*Registry) DiscoverDatabases 

func (r *Registry) DiscoverDatabases(ctx context.Context, cfg factory.SourceConfig) ([]string, error)

func (*Registry) DiscoverPublications 

func (r *Registry) DiscoverPublications(ctx context.Context, cfg factory.SourceConfig) ([]hermod.PublicationInfo, error)

func (*Registry) DiscoverReplicationSlots 

func (r *Registry) DiscoverReplicationSlots(ctx context.Context, cfg factory.SourceConfig) ([]hermod.ReplicationSlotInfo, error)

func (*Registry) DiscoverSinkColumns 

func (r *Registry) DiscoverSinkColumns(ctx context.Context, cfg factory.SinkConfig, table string) ([]hermod.ColumnInfo, error)

func (*Registry) DiscoverSinkDatabases 

func (r *Registry) DiscoverSinkDatabases(ctx context.Context, cfg factory.SinkConfig) ([]string, error)

func (*Registry) DiscoverSinkTables 

func (r *Registry) DiscoverSinkTables(ctx context.Context, cfg factory.SinkConfig) ([]string, error)

func (*Registry) DiscoverSourceColumns 

func (r *Registry) DiscoverSourceColumns(ctx context.Context, cfg factory.SourceConfig, table string) ([]hermod.ColumnInfo, error)

func (*Registry) DiscoverTables 

func (r *Registry) DiscoverTables(ctx context.Context, cfg factory.SourceConfig) ([]string, error)

func (*Registry) DrainWorkflowDLQ 

func (r *Registry) DrainWorkflowDLQ(ctx context.Context, id string) error

func (*Registry) EvaluateConditions 

func (r *Registry) EvaluateConditions(msg hermod.Message, conditions []map[string]any) bool

func (*Registry) ExecSinkStatement 

func (r *Registry) ExecSinkStatement(ctx context.Context, cfg factory.SinkConfig, stmt string) error

func (*Registry) ExecuteSQL 

func (r *Registry) ExecuteSQL(ctx context.Context, cfg factory.SourceConfig, query string, userSample map[string]any) ([]map[string]any, error)

func (*Registry) ExecuteSinkSQL 

func (r *Registry) ExecuteSinkSQL(ctx context.Context, cfg factory.SinkConfig, query string, userSample map[string]any) ([]map[string]any, error)

func (*Registry) GetAllStatuses 

func (r *Registry) GetAllStatuses() []telemetry.StatusUpdate

func (*Registry) GetDB 

func (r *Registry) GetDB(ctx context.Context, typeName string, config map[string]string) (*sql.DB, error)

func (*Registry) GetDQScorer 

func (r *Registry) GetDQScorer() *governance.Scorer

func (*Registry) GetDashboardHistory added in v1.8.1 

func (r *Registry) GetDashboardHistory(ctx context.Context, vhost string, since time.Time, limit int) ([]storage.DashboardSample, error)

GetDashboardHistory reads back the persisted series for a vhost.

func (*Registry) GetDashboardStats 

func (r *Registry) GetDashboardStats(ctx context.Context, vhost string) (storage.DashboardStats, error)

func (*Registry) GetEngine 

func (r *Registry) GetEngine(id string) (*pkgengine.Engine, bool)

func (*Registry) GetLogStorage 

func (r *Registry) GetLogStorage() storage.Storage

func (*Registry) GetLogger 

func (r *Registry) GetLogger() hermod.Logger

func (*Registry) GetLookupCache 

func (r *Registry) GetLookupCache(key string) (any, bool)

func (*Registry) GetLookupCacheSize 

func (r *Registry) GetLookupCacheSize() (int, int)

func (*Registry) GetMeshManager 

func (r *Registry) GetMeshManager() *mesh.Manager

func (*Registry) GetNodeState 

func (r *Registry) GetNodeState(key string) (any, bool)

func (*Registry) GetNodeStates 

func (r *Registry) GetNodeStates(ctx context.Context, workflowID string) (map[string]any, error)

func (*Registry) GetOrOpenDB 

func (r *Registry) GetOrOpenDB(src storage.Source) (*sql.DB, error)

func (*Registry) GetOrOpenDBByID 

func (r *Registry) GetOrOpenDBByID(ctx context.Context, id string) (*sql.DB, string, error)

func (*Registry) GetPIIStats 

func (r *Registry) GetPIIStats() map[string]*PIIStats

func (*Registry) GetSink 

func (r *Registry) GetSink(workflowID, nodeID string) (hermod.Sink, bool)

func (*Registry) GetSinkConfig 

func (r *Registry) GetSinkConfig(ctx context.Context, id string) (storage.Sink, error)

func (*Registry) GetSinkConfigs 

func (r *Registry) GetSinkConfigs(id string) ([]factory.SinkConfig, bool)

func (*Registry) GetSource 

func (r *Registry) GetSource(ctx context.Context, id string) (storage.Source, error)

func (*Registry) GetSourceConfig 

func (r *Registry) GetSourceConfig(ctx context.Context, id string) (storage.Source, error)

func (*Registry) GetSourceConfigs 

func (r *Registry) GetSourceConfigs(id string) ([]factory.SourceConfig, bool)

func (*Registry) GetSourceFactoryConfig 

func (r *Registry) GetSourceFactoryConfig(ctx context.Context, id string) (factory.SourceConfig, error)

func (*Registry) GetSourceFormSamples 

func (r *Registry) GetSourceFormSamples(ctx context.Context, path string, limit int) ([]hermod.Message, error)

func (*Registry) GetStorage 

func (r *Registry) GetStorage() storage.Storage

func (*Registry) GetWorkflowConfig 

func (r *Registry) GetWorkflowConfig(id string) (storage.Workflow, bool)

func (*Registry) GetWorkflowHealth 

func (r *Registry) GetWorkflowHealth(id string) (storage.WorkflowHealth, bool)

GetWorkflowHealth returns a real-time health summary for a running workflow.

func (*Registry) GetWorkflowStatus 

func (r *Registry) GetWorkflowStatus(id string) (telemetry.StatusUpdate, bool)

func (*Registry) IsBranchPreviewable added in v1.8.1 

func (r *Registry) IsBranchPreviewable(nodeType string) bool

IsBranchPreviewable reports whether nodeType names a routing node the editor can preview, so a caller can pick the right path before doing any work rather than reacting to an error afterwards.

func (*Registry) IsDebuggerAttached 

func (r *Registry) IsDebuggerAttached(workflowID string) bool

func (*Registry) IsEngineRunning 

func (r *Registry) IsEngineRunning(id string) bool

func (*Registry) IsResourceInUse 

func (r *Registry) IsResourceInUse(ctx context.Context, resourceID string, excludeID string, isSource bool) bool

func (*Registry) IsWorkflowNode added in v1.8.1 

func (r *Registry) IsWorkflowNode(nodeType string) bool

IsWorkflowNode reports whether nodeType names a registered node executor.

Used to tell two failures apart. "stateful" is a real node that a preview cannot run, because running it would mutate state a live workflow depends on; "statefull" is a typo. Both used to produce 

unknown transformation type: no transformer is registered under that name

which describes the typo and misdescribes the node.

func (*Registry) Logger 

func (r *Registry) Logger() hermod.Logger

func (*Registry) NotifyWorkerShutdown added in v1.11.0 

func (r *Registry) NotifyWorkerShutdown(ctx context.Context, activeWorkflows int)

NotifyWorkerShutdown reports that this worker is going down and taking its running workflows with it.

One alert for the worker rather than one per workflow: a worker running forty workflows would otherwise send forty messages at the moment its operator is least able to read them.

func (*Registry) PauseForDebugger 

func (r *Registry) PauseForDebugger(workflowID, nodeID string, msg hermod.Message)

func (*Registry) PreviewBranch added in v1.8.1 

func (r *Registry) PreviewBranch(ctx context.Context, nodeType string, config map[string]any, msg hermod.Message) ([]hermod.Message, string, error)

PreviewBranch runs a routing node against a sample message and reports the branch it would take.

The editor's "Test" button used to have one path: the transformer registry. That covers every node whose job is to change a message, and none of the nodes whose job is to choose where it goes. switch, condition and router are registered as node executors and dispatched at runtime by executeNodeInternal, so a workflow containing them ran correctly while the preview of the same node answered 

unknown transformation type "switch": no transformer is registered under that name

which reads as a broken configuration rather than a preview that never looked in the right registry.

Only executors that declare themselves interfaces.PreviewSafe run here. A node that exists but has not declared it is refused by name, because a preview is not a reason to write to a sink or record an approval.

func (*Registry) PurgeLogs 

func (r *Registry) PurgeLogs(ctx context.Context, before time.Time) error

func (*Registry) RebuildWorkflow 

func (r *Registry) RebuildWorkflow(ctx context.Context, workflowID string, fromOffset int64) error

func (*Registry) RecordCircuitBreakerFailure 

func (r *Registry) RecordCircuitBreakerFailure(workflowID, breakerNodeID string)

RecordCircuitBreakerFailure counts a downstream failure against a circuit breaker node.

The breaker reads this count and opens once it passes its threshold. Nothing incremented it before, so a node offered in the editor as "Stop flow on failure threshold" always reported success however broken the downstream was — a control that cannot fire, which is worse than no control because someone believes it is there.

The registry is the natural place for this: it already holds node state and is what the traversal has a handle on.

func (*Registry) RecordStep 

func (r *Registry) RecordStep(ctx context.Context, workflowID, messageID string, step hermod.TraceStep)

func (*Registry) ResumeApproval 

func (r *Registry) ResumeApproval(ctx context.Context, app storage.Approval, branch string) error

ResumeApproval resumes a halted workflow at an approval node with the specified decision branch ("approved" or "rejected").

func (*Registry) RunWorkflowNode 

func (r *Registry) RunWorkflowNode(workflowID string, node *storage.WorkflowNode, msg hermod.Message) ([]hermod.Message, string, error)

func (*Registry) SampleSinkTable 

func (r *Registry) SampleSinkTable(ctx context.Context, cfg factory.SinkConfig, table string) (hermod.Message, error)

func (*Registry) SampleTable 

func (r *Registry) SampleTable(ctx context.Context, cfg factory.SourceConfig, table string) (hermod.Message, error)

func (*Registry) SetConfig 

func (r *Registry) SetConfig(cfg config.Config)

func (*Registry) SetFactories 

func (r *Registry) SetFactories(sourceFactory SourceFactory, sinkFactory SinkFactory)

func (*Registry) SetLogStorage 

func (r *Registry) SetLogStorage(s storage.Storage)

func (*Registry) SetLogger 

func (r *Registry) SetLogger(logger hermod.Logger)

func (*Registry) SetLookupCache 

func (r *Registry) SetLookupCache(key string, value any, ttl time.Duration)

SetLookupCache stores a value with an optional TTL. Expired entries are reclaimed lazily on read and during a bounded sweep on write, so no per-key goroutine is spawned (which previously leaked goroutines and memory under high lookup throughput).

func (*Registry) SetNodeState 

func (r *Registry) SetNodeState(key string, val any)

func (*Registry) SetSecretManager 

func (r *Registry) SetSecretManager(mgr secrets.Manager)

func (*Registry) SetSinkFactory 

func (r *Registry) SetSinkFactory(f SinkFactory)

func (*Registry) SetSourceFactory 

func (r *Registry) SetSourceFactory(f SourceFactory)

func (*Registry) SetStateStore 

func (r *Registry) SetStateStore(ss hermod.StateStore)

func (*Registry) SetStorage 

func (r *Registry) SetStorage(s storage.Storage)

func (*Registry) SetWorkerID added in v1.11.0 

func (r *Registry) SetWorkerID(id string)

SetWorkerID records which worker this registry belongs to, so a worker-level alert can say which one went down. A deployment runs several and "a worker is shutting down" is not actionable without the name.

func (*Registry) SimulateWorkflow added in v1.13.0 

func (r *Registry) SimulateWorkflow(ctx context.Context, wf storage.Workflow, in SimulationInput) ([]WorkflowStepResult, error)

SimulateWorkflow runs sample messages through a workflow without starting an engine, and reports what every node emitted.

Partial previews a workflow that is still being built. It keeps the checks that make the workflow a graph — a source, no cycle, no edge to nowhere — and drops the ones about whether it could run: a reachable sink, configured references, the dead-letter settings. None of those change what a node emits, and a workflow is missing its sink exactly while its nodes are being set up, which is when the editor needs the preview.

func (*Registry) StartWorkflow 

func (r *Registry) StartWorkflow(id string, wf storage.Workflow) error

StartWorkflow creates and starts a workflow engine for the given workflow configuration.

func (*Registry) StateStore 

func (r *Registry) StateStore() hermod.StateStore

func (*Registry) StatusSubscriberCount 

func (r *Registry) StatusSubscriberCount() int

StatusSubscriberCount returns the number of active status subscribers. It is primarily useful for observability and tests that assert subscribers are released when a client disconnects.

func (*Registry) StopAll 

func (r *Registry) StopAll()

func (*Registry) StopEngine 

func (r *Registry) StopEngine(ctx context.Context, id string) error

StopEngine stops a workflow on an operator's instruction. Unlike StopEngineWithoutUpdate — which the supervisor and the worker's own reconciliation use — this marks the end of a stall episode, so the workflow's automatic-restart budget starts fresh the next time it runs.

func (*Registry) StopEngineWithoutUpdate 

func (r *Registry) StopEngineWithoutUpdate(ctx context.Context, id string) error

func (*Registry) Storage 

func (r *Registry) Storage() interfaces.RegistryStorage

func (*Registry) SubscribeDashboardStats 

func (r *Registry) SubscribeDashboardStats(vhost string) chan storage.DashboardStats

func (*Registry) SubscribeDebugger 

func (r *Registry) SubscribeDebugger(workflowID string) chan DebuggerEvent

func (*Registry) SubscribeLiveMessages 

func (r *Registry) SubscribeLiveMessages() chan LiveMessage

func (*Registry) SubscribeLogs 

func (r *Registry) SubscribeLogs() chan storage.Log

func (*Registry) SubscribeStatus 

func (r *Registry) SubscribeStatus() chan telemetry.StatusUpdate

func (*Registry) SubscribeWorkflowLiveMessages 

func (r *Registry) SubscribeWorkflowLiveMessages(workflowID string) chan LiveMessage

func (*Registry) SubscribeWorkflowLogs 

func (r *Registry) SubscribeWorkflowLogs(workflowID string) chan storage.Log

func (*Registry) SubscribeWorkflowStatus 

func (r *Registry) SubscribeWorkflowStatus(workflowID string) chan telemetry.StatusUpdate

func (*Registry) TestSink 

func (r *Registry) TestSink(ctx context.Context, cfg factory.SinkConfig) error

func (*Registry) TestSource 

func (r *Registry) TestSource(ctx context.Context, cfg factory.SourceConfig) error

func (*Registry) TestTransformationPipeline 

func (r *Registry) TestTransformationPipeline(ctx context.Context, transformations []storage.Transformation, msg hermod.Message) ([]hermod.Message, error)

func (*Registry) TestWorkflow 

func (r *Registry) TestWorkflow(ctx context.Context, wf storage.Workflow, msg hermod.Message) ([]WorkflowStepResult, error)

TestWorkflow runs msg through a workflow the engine could start, seeding every source node with it. It is what the editor's Test button calls.

func (*Registry) TriggerSnapshot 

func (r *Registry) TriggerSnapshot(ctx context.Context, sourceID string, tables ...string) error

func (*Registry) UnsubscribeDashboardStats 

func (r *Registry) UnsubscribeDashboardStats(ch chan storage.DashboardStats)

func (*Registry) UnsubscribeDebugger 

func (r *Registry) UnsubscribeDebugger(workflowID string, ch chan DebuggerEvent)

func (*Registry) UnsubscribeLiveMessages 

func (r *Registry) UnsubscribeLiveMessages(ch chan LiveMessage)

func (*Registry) UnsubscribeLogs 

func (r *Registry) UnsubscribeLogs(ch chan storage.Log)

func (*Registry) UnsubscribeStatus 

func (r *Registry) UnsubscribeStatus(ch chan telemetry.StatusUpdate)

func (*Registry) UpdateNodeState 

func (r *Registry) UpdateNodeState(ctx context.Context, workflowID, nodeID string, state any) error

func (*Registry) UpdateSink 

func (r *Registry) UpdateSink(ctx context.Context, snk storage.Sink) error

func (*Registry) UpdateSinkStatus 

func (r *Registry) UpdateSinkStatus(ctx context.Context, id string, status string) error

func (*Registry) UpdateSource 

func (r *Registry) UpdateSource(ctx context.Context, src storage.Source) error

func (*Registry) UpdateSourceStatus 

func (r *Registry) UpdateSourceStatus(ctx context.Context, id string, status string) error

func (*Registry) ValidateWorkflow 

func (r *Registry) ValidateWorkflow(ctx context.Context, wf storage.Workflow) error

type SimulationInput added in v1.13.0 

type SimulationInput struct {
	// Message seeds every source node PerSource does not name. Nil leaves such
	// a source unseeded, and everything downstream of it unreached.
	Message hermod.Message
	// PerSource seeds source nodes, by node ID, with their own sample.
	PerSource map[string]hermod.Message
	// Partial previews a workflow that is still being built; see
	// SimulateWorkflow.
	Partial bool
}

SimulationInput is what a simulation feeds its source nodes.

type SinkFactory 

type SinkFactory func(factory.SinkConfig) (hermod.Sink, error)

type SourceFactory 

type SourceFactory func(factory.SourceConfig) (hermod.Source, error)

type WorkflowStepResult 

type WorkflowStepResult struct {
	NodeID   string            `json:"node_id"`
	NodeType string            `json:"node_type"`
	Payload  map[string]any    `json:"payload,omitempty"`
	Metadata map[string]string `json:"metadata,omitempty"`
	Error    string            `json:"error,omitempty"`
	Filtered bool              `json:"filtered,omitempty"`
	// Skipped marks a node no message reached: every edge into it was on a
	// branch that was not taken. Filtered is set as well, which is how such a
	// node was reported before Skipped existed; Skipped is what tells it apart
	// from a node that was reached and emitted nothing.
	Skipped bool   `json:"skipped,omitempty"`
	Branch  string `json:"branch,omitempty"`
	// TakenEdges are the IDs of the edges this node's output travelled along:
	// the path the editor highlights. Empty for a node that emitted nothing.
	TakenEdges []string `json:"taken_edges,omitempty"`
}

Source Files

View all Source files

Directories

Path Synopsis
Package nodes is the one import that makes every workflow node type available.
 Package nodes is the one import that makes every workflow node type available.
control
core
flow
reliability
util

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.