Message Flow
Understanding how messages flow through the system is essential for building reliable components. This page explains the complete execution path from trigger to completion.
Execution Overview
+-----------------------------------------------------------------------------+
| MESSAGE EXECUTION FLOW |
+-----------------------------------------------------------------------------+
TRIGGER TRANSPORT COMPONENT
------- --------- ---------
+---------------+ +-----------------+
| HTTP Request | | NATS publish |
| Timer/Cron |-------->| (wire.Publish, |
| Manual Button | | From="signal") |
| Edge Output | +--------+--------+
+---------------+ |
v
+-----------------+
| Module receiver |
| (queue group) |
+--------+--------+
|
v
+-----------------+
| Scheduler |
| .Handle() |
+--------+--------+
|
v
+-----------------+
| Runner |
| msg handler |
+--------+--------+
|
v
+-----------------+
| Component |
| .Handle() |
+--------+--------+
|
v
+-----------------+
| output(ctx, |
| port, data) |
+--------+--------+
|
+------------------------+------------------------+
| | |
v v v
+-----------------+ +-----------------+ +------------------------+
| No edge | | Same module | | Different module |
| (terminal) | | (in-process) | | (NATS request/reply, |
+-----------------+ +--------+--------+ | gRPC fallback) |
| +-----------+------------+
+--------------+-----------+
|
v
+-----------------+
| Edge Expression |
| Evaluation |
+--------+--------+
|
v
+-----------------+
| Next Component |
| .Handle() |
+-----------------+
Detailed Execution Steps
Step 1: Trigger Creation
Messages can be triggered by:
- HTTP Request: HTTP Server component receives a request
- Timer: Ticker component fires at intervals
- Manual: User clicks a button in the UI
- Edge: Previous node sends output
Step 2: External Signal Publish
External triggers (dashboard Send, MCP send_signal, debug tools) are plain NATS publishes via pkg/wire.Publish — there is no CRD and no controller round-trip:
import "github.com/tiny-systems/module/pkg/wire"
// targetNode is the node's full name (flowID.module.nodename)
reply, err := wire.Publish(ctx, nc, targetNode, "input", payloadJSON, wire.Options{
From: wire.FromSignal, // "signal" — marks this as an external trigger
WaitForReply: true, // block for the synchronous response
})
The subject encodes the target module:
tinymodule.<module>.msg— business ports; the module's pods share a queue group, so replicas load-balancetinymodule.<module>.sysmsg— system ports (any port starting with_); fans out to every pod, and an in-handler leader check gates the action
Routing metadata travels in NATS headers (x-to, x-from, x-edge-id, W3C traceparent).
Step 3: Module Receiver Dispatch
Each module pod subscribes at startup. The receiver reads the x-to header (<node-full-name>:<port>) and hands the message to the scheduler. Because From is "signal", the runner unmarshals the raw payload directly into the port's Configuration struct — edge-config evaluation is skipped for external signals.
Step 4: Scheduler Routing
The scheduler finds the runner for the target node:
// Conceptually:
func (s *Schedule) Handle(ctx context.Context, msg *runner.Msg) (any, error) {
nodeName, portName := utils.ParseFullPortName(msg.To)
instance, ok := s.instancesMap.Get(nodeName)
if !ok {
// Node belongs to a different module — route over the wire
return s.routeToRemoteModule(ctx, msg)
}
// Local dispatch — direct in-process call
return instance.Handle(ctx, msg)
}
Step 5: Runner Processing
The Runner wraps the component and handles:
- Edge-config expression evaluation and message deserialization
- Settings deduplication
- OpenTelemetry tracing
- Error handling and per-edge retry
// Conceptually:
func (r *Runner) msgHandler(ctx context.Context, msg *runner.Msg) (any, error) {
ctx, span := r.tracer.Start(ctx, ...)
defer span.End()
// Evaluate edge expressions / deserialize into the port's typed struct
data := r.evaluateAndDeserialize(msg, port)
// System ports dispatch to capability interfaces (OnSettings, OnControl, ...)
// Business ports reach Component.Handle, which returns module.Result
res := r.component.Handle(ctx, r.outputHandler, port, data)
return res.Value(), res.Err()
}
Step 6: Component Handle
Your component receives the message:
func (c *MyComponent) Handle(
ctx context.Context,
output module.Handler,
port string,
msg any,
) module.Result {
input := msg.(InputType)
// Process...
result := process(input)
// Send to output port — chain the Result up the call stack
return output(ctx, "output", result)
}
Step 7: Output Handler
The injected output handler routes the emit along the node's edges. For each edge from the port it serializes the data, dispatches it (honoring the edge's RetryPolicy — see Error Handling), and returns a module.Result:
- No edge: the emit is a terminal no-op.
- Same module: direct in-process dispatch through the scheduler.
- Different module: over the wire — NATS core request/reply when
TINY_NATS_URLis set, gRPC otherwise. - Durable flow: the hop is stamped with a run id and idempotency key and published fire-and-forget to the JetStream work queue.
Step 8: Edge Expression Evaluation
Data is transformed using expressions:
// Edge configuration
{
"userId": "{{$.user.id}}",
"fullName": "{{'Hello ' + $.user.firstName + ' ' + $.user.lastName}}",
"count": "{{$.items.length}}"
}
// Source data
{
"user": {"id": "123", "firstName": "John", "lastName": "Doe"},
"items": [1, 2, 3]
}
// Result after evaluation
{
"userId": "123",
"fullName": "Hello John Doe",
"count": 3
}
Step 9: Next Node
The process repeats for the next node in the flow.
Same Module vs Cross-Module
Same Module Communication
+-------------------------------------------------------------+
| SAME MODULE |
| |
| Component A In-process call Component B |
| +-----------+ -----------------> +-----------+ |
| | output() | | Handle() | |
| +-----------+ Same process, +-----------+ |
| shared scheduler |
+-------------------------------------------------------------+
- Fast: Direct function call through the local scheduler
- Same process: No network hop
Cross-Module Communication
+----------------------+ NATS request/reply +----------------------+
| MODULE A | <-------------------------> | MODULE B |
| | tinymodule.<module-b>.msg | |
| Component A | | Component C |
| +-----------+ | 1. Serialize (JSON) | +-----------+ |
| | output() |-------|-----------------------------|--| Handle() | |
| +-----------+ | 2. Publish, block on | +-----------+ |
| | reply inbox | |
+----------------------+ 3. Deserialize +----------------------+
- Primary transport: NATS core request/reply (when
TINY_NATS_URLis set). The sender blocks on the reply inbox, preserving the blocking I/O model; the target module's pods load-balance via their queue group. - Fallback transport: gRPC over TCP when the runtime starts without
TINY_NATS_URL. - Durable flows: hops are published to a JetStream work queue instead (fire-and-forget with idempotency keys) — see Blocking vs Async.
- Serialization: JSON marshaling/unmarshaling in both cases.
Blocking Behavior
Critical: The classic execution path is blocking.
When you call output(ctx, "output", data):
- The message is sent to the next node
- The next node's
Handle()is called - All downstream processing completes
- Only then does
output()return
func (c *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
fmt.Println("Before output")
// This BLOCKS until downstream chain completes
res := output(ctx, "output", msg)
fmt.Println("After output") // Runs after ALL downstream is done
return res
}
Nodes in a durable flow behave differently: emits return as soon as the hop is persisted to the work queue. See Blocking vs Async for details.
Error Propagation
On the classic path, failures propagate back through the call chain as module.Result:
Node A Node B Node C
| | |
| output() ---------------|-------------------------|--> returns Fail(err)
| | |
| <-----------------------|<------------------------|--- failure propagates
| | |
| Handle() returns res | |
| | |
func (c *ComponentB) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
result := process(msg)
if result.Error != nil {
return module.Fail(result.Error) // Failure goes back to Node A
}
return output(ctx, "output", result)
}
Next Steps
- Blocking vs Async - Understand execution modes
- Error Handling - Handle errors properly
- Internal Routing - Deep dive into routing