Documentation
¶
Index ¶
- Constants
- func OverReleaseCount() int64
- func PopulateFromMap(msg hermod.Message, data map[string]any)
- func ReleaseMessage(m hermod.Message)
- func ResetOverReleaseCount()
- func SampleIsCDC(data map[string]any) bool
- func SanitizeMap(m map[string]any) map[string]any
- func SanitizeValue(v any) any
- func TryFixJSON(data []byte) []byte
- type DefaultMessage
- func (m *DefaultMessage) After() []byte
- func (m *DefaultMessage) Before() []byte
- func (m *DefaultMessage) ClearCachedPayload()
- func (m *DefaultMessage) ClearPayloads()
- func (m *DefaultMessage) Clone() hermod.Message
- func (m *DefaultMessage) Data() map[string]any
- func (m *DefaultMessage) DataRef() map[string]any
- func (m *DefaultMessage) ID() string
- func (m *DefaultMessage) MarshalJSON() ([]byte, error)
- func (m *DefaultMessage) Metadata() map[string]string
- func (m *DefaultMessage) MetadataRef() map[string]string
- func (m *DefaultMessage) MetadataValue(key string) (string, bool)
- func (m *DefaultMessage) Operation() hermod.Operation
- func (m *DefaultMessage) Payload() []byte
- func (m *DefaultMessage) PayloadLen() int
- func (m *DefaultMessage) RefCount() int32
- func (m *DefaultMessage) Release()
- func (m *DefaultMessage) Reset()
- func (m *DefaultMessage) Retain()
- func (m *DefaultMessage) Schema() string
- func (m *DefaultMessage) SetAfter(after []byte)
- func (m *DefaultMessage) SetBefore(before []byte)
- func (m *DefaultMessage) SetData(key string, value any)
- func (m *DefaultMessage) SetID(id string)
- func (m *DefaultMessage) SetMetadata(key, value string)
- func (m *DefaultMessage) SetOperation(op hermod.Operation)
- func (m *DefaultMessage) SetPayload(payload []byte)
- func (m *DefaultMessage) SetSchema(schema string)
- func (m *DefaultMessage) SetTable(table string)
- func (m *DefaultMessage) Table() string
- func (m *DefaultMessage) ToMap() map[string]any
Constants ¶
const NonObjectPayloadKey = "payload"
NonObjectPayloadKey is where a payload that is not a JSON object is exposed.
Sources are free to emit a bare string, a scalar, an array, or bytes that are not JSON at all. Such a payload has no field names to merge into the message root, so it is preserved under this single key instead. That keeps it addressable by transformations and, crucially, keeps it in the serialised output: previously only objects survived and everything else was silently dropped.
The name is deliberately "payload": array payloads have always been exposed this way, and it is the template variable the UI advertises for notification sinks ({{.payload}}). Widening the existing key to cover strings and scalars keeps every working template working, where a new name would not.
Variables ¶
This section is empty.
Functions ¶
func OverReleaseCount ¶
func OverReleaseCount() int64
OverReleaseCount reports how many times a message was released after its refcount already reached zero. Any non-zero value means some owner is releasing a reference it does not hold; treat it as a correctness bug, not a tuning signal.
func PopulateFromMap ¶ added in v1.12.0
PopulateFromMap builds a message from a ToMap()-shaped sample -- the shape the workflow editor holds, sends to the preview endpoint and stores as a source sample.
It lives here, rather than beside either of its callers, because it is the single answer to "what would the engine see for this sample". The editor's SQL query builder resolved query variables against the sample map itself while the preview endpoint resolved them against a message built from it, and the two disagree about the CDC envelope: ToMap re-nests the after-image under "after", and this function unwraps it back into the root, which is where the engine's Data() has it. So `{{.after.payload}}` returned rows in the builder and bound NULL in the node that ran the same text. One definition, two readers -- see TestTheBuilderBindsWhatTheEngineBinds.
func ReleaseMessage ¶
ReleaseMessage returns a message to the pool.
func ResetOverReleaseCount ¶
func ResetOverReleaseCount()
ResetOverReleaseCount zeroes the counter. Intended for tests that assert a pipeline runs with balanced reference counting.
func SampleIsCDC ¶ added in v1.12.0
SampleIsCDC reports whether ToMap will serialise this sample as a CDC event, which is what decides whether the data map doubles as the after-image. It mirrors the one condition ToMap uses: a non-empty operation.
func SanitizeMap ¶
SanitizeMap sanitizes all values in a map.
func SanitizeValue ¶
SanitizeValue converts special types (like UUIDs) to JSON-friendly strings.
func TryFixJSON ¶
TryFixJSON attempts to fix common JSON issues like trailing commas to make unmarshaling more lenient.
It works on the bytes rather than on a string copy of them. The first thing this does is decide whether the payload even looks like JSON, and it used to copy the whole body into a string to find out — which for a text, CSV or binary body is the answer "no" at the cost of a full copy, on every message.
Types ¶
type DefaultMessage ¶
type DefaultMessage struct {
// contains filtered or unexported fields
}
DefaultMessage is a concrete implementation of the hermod.Message interface. It uses a sync.Pool to minimize allocations.
func AcquireMessage ¶
func AcquireMessage() *DefaultMessage
AcquireMessage gets a message from the pool.
func (*DefaultMessage) After ¶
func (m *DefaultMessage) After() []byte
func (*DefaultMessage) Before ¶
func (m *DefaultMessage) Before() []byte
func (*DefaultMessage) ClearCachedPayload ¶
func (m *DefaultMessage) ClearCachedPayload()
ClearCachedPayload clears only the marshaled payload bytes.
func (*DefaultMessage) ClearPayloads ¶
func (m *DefaultMessage) ClearPayloads()
ClearPayloads clears the data content of the message but keeps metadata/system fields.
func (*DefaultMessage) Clone ¶
func (m *DefaultMessage) Clone() hermod.Message
func (*DefaultMessage) Data ¶
func (m *DefaultMessage) Data() map[string]any
func (*DefaultMessage) DataRef ¶
func (m *DefaultMessage) DataRef() map[string]any
DataRef returns the live data map. The lock is released before it does, so a caller reading or writing the result holds nothing.
That is safe today only because of who calls it, and the reasoning is not visible from any one call site — so it is written down here. Three rules, and every caller satisfies one:
- Mutate only a private copy. foreach walks and deletes through the DataRef of a Clone, and the PII scan — which recursively ranges the map, the exact operation the runtime makes fatal — is handed a Clone on its own goroutine. Clone deep-copies the data map, which is what makes both sound.
- One consumer per object. The traversal clones when a node has more than one target, so two processNode goroutines never share a message.
- Writes happen only where ownership is exclusive. A message is taken by exactly one worker goroutine, and when it is later fanned out to N sinks — the same object, one goroutine each — every holder is a reader. Sinks mutate only messages they construct themselves, which TestSinksDoNotMutateMessagesTheyAreGiven enforces rather than trusts.
Break any of those and this races SetData, which the runtime reports as `fatal error: concurrent map read and map write` and no recover() can contain. Prefer Data() unless the clone is measurably too expensive; ToMap() if the value outlives the caller's stack. And do not let this comment age the way OrderingKey's ownership argument did — it was true when written and false a caller later.
func (*DefaultMessage) ID ¶
func (m *DefaultMessage) ID() string
func (*DefaultMessage) MarshalJSON ¶
func (m *DefaultMessage) MarshalJSON() ([]byte, error)
func (*DefaultMessage) Metadata ¶
func (m *DefaultMessage) Metadata() map[string]string
func (*DefaultMessage) MetadataRef ¶
func (m *DefaultMessage) MetadataRef() map[string]string
func (*DefaultMessage) MetadataValue ¶ added in v1.8.1
func (m *DefaultMessage) MetadataValue(key string) (string, bool)
MetadataValue reads one metadata entry without copying the map.
Metadata() has to clone — handing out the live map would race with a concurrent SetMetadata — so reading a single key through it copies every other entry to throw them away. The trace propagator does exactly that twice per sink write (traceparent, then tracestate), and it is not alone. This does the lookup under the same read lock, so it is as safe and costs nothing.
func (*DefaultMessage) Operation ¶
func (m *DefaultMessage) Operation() hermod.Operation
func (*DefaultMessage) Payload ¶
func (m *DefaultMessage) Payload() []byte
func (*DefaultMessage) PayloadLen ¶ added in v1.8.1
func (m *DefaultMessage) PayloadLen() int
PayloadLen reports len(Payload()) without copying the payload.
The engine asks for a payload's size far more often than for the payload: batch-by-bytes accounting sums it for every message and the per-write debug log reports it. Going through Payload() for that cloned the bytes and threw the copy away — together 2.8 GB of the 4.9 GB a 150k-message benchmark allocated, to produce two integers.
It materialises the data map exactly as Payload() does, so the two never disagree about a message whose payload has not been marshalled yet; TestPayloadLenMatchesPayload holds them together.
func (*DefaultMessage) RefCount ¶
func (m *DefaultMessage) RefCount() int32
RefCount reports the message's current reference count.
It exists so the ownership contract can actually be asserted rather than reasoned about. Every node executor must return messages the caller owns one reference to; that invariant is invisible without being able to read the count, which is why a violation in the traversal's source branch went unnoticed until it was corrupting data. Use it in tests and diagnostics, not to make control-flow decisions: the value can change under you at any moment.
func (*DefaultMessage) Release ¶
func (m *DefaultMessage) Release()
func (*DefaultMessage) Reset ¶
func (m *DefaultMessage) Reset()
Reset clears the message state so it can be reused.
func (*DefaultMessage) Retain ¶
func (m *DefaultMessage) Retain()
func (*DefaultMessage) Schema ¶
func (m *DefaultMessage) Schema() string
func (*DefaultMessage) SetAfter ¶
func (m *DefaultMessage) SetAfter(after []byte)
func (*DefaultMessage) SetBefore ¶
func (m *DefaultMessage) SetBefore(before []byte)
func (*DefaultMessage) SetData ¶
func (m *DefaultMessage) SetData(key string, value any)
func (*DefaultMessage) SetMetadata ¶
func (m *DefaultMessage) SetMetadata(key, value string)
func (*DefaultMessage) SetOperation ¶
func (m *DefaultMessage) SetOperation(op hermod.Operation)
func (*DefaultMessage) SetPayload ¶
func (m *DefaultMessage) SetPayload(payload []byte)
func (*DefaultMessage) SetSchema ¶
func (m *DefaultMessage) SetSchema(schema string)
func (*DefaultMessage) SetTable ¶
func (m *DefaultMessage) SetTable(table string)
func (*DefaultMessage) Table ¶
func (m *DefaultMessage) Table() string
func (*DefaultMessage) ToMap ¶
func (m *DefaultMessage) ToMap() map[string]any
ToMap returns a snapshot of the message that shares no mutable state with it.
Independence is the contract, not a detail. Every caller either marshals the result or hands it to another goroutine — a trace step recorded against PostgreSQL, a live-viewer frame fanned out to SSE subscribers — while the pipeline keeps transforming the message it came from.
This used to copy the top level only, so each nested map[string]any in the snapshot was the same map the live message holds. SetData with a dotted key walks into that nested map and assigns in place, which put a write inside a map json.Marshal was already iterating on the trace goroutine. Concurrent map access is a runtime fatal error rather than a panic, so the recover() guarding that marshal could not contain it: the engine exited 2 mid-trace.
The deep copy is paid by callers who are about to serialise this anyway, which costs strictly more than copying it. Data() and DataRef() are unchanged — they are the hot read path and they do not cross a goroutine boundary.
Source Files