Stateful Component Example
A complete example of a component whose state survives restarts and propagates across replicas.
Overview
This counter component maintains state that must be shared across multiple replicas of a module. It demonstrates:
- The
module.Statestorage primitive (injected via theStatefulcapability) module.Basefor lifecycle boilerplate- Leader-gated mutations
- A dashboard control with a Reset button
- The reconcile-port metadata mechanism that backs it all
How State Works in the SDK
Components do not write Kubernetes objects directly. The SDK injects a module.State backend (via OnState, which module.Base implements for you):
type State interface {
Get(ctx context.Context, key string) (value []byte, ok bool, err error)
Set(ctx context.Context, key string, value []byte) error
Delete(ctx context.Context, key string) error
List(ctx context.Context, prefix string) ([]string, error)
Scoped(scope, id string) State
}
The default backend reads through the controller-runtime cache (kept fresh by the TinyNode watch) and writes through the reconcile-port debouncer into status.metadata of the node. Every replica observes the same state via the watch, so replicas converge without hand-rolled flag bookkeeping. Reads are eventually consistent across replicas; read-your-writes holds within the writing instance.
Complete Implementation
package counter
import (
"context"
"fmt"
"strconv"
"sync"
"github.com/tiny-systems/module/api/v1alpha1"
"github.com/tiny-systems/module/module"
"github.com/tiny-systems/module/pkg/utils"
"github.com/tiny-systems/module/registry"
)
const (
ComponentName = "stateful_counter"
IncrementPort = "increment"
ValuePort = "value"
stateKeyCount = "count"
)
type Context any
// Settings configuration
type Settings struct {
InitialValue int64 `json:"initialValue" title:"Initial Value" default:"0" description:"Starting count value"`
}
// IncrementInput for changing the counter (negative amount decrements)
type IncrementInput struct {
Context Context `json:"context,omitempty" configurable:"true" title:"Context"`
Amount int64 `json:"amount" title:"Amount" default:"1" description:"Amount to add (negative to subtract)"`
}
// CountOutput emitted after each operation
type CountOutput struct {
Context Context `json:"context,omitempty" title:"Context"`
Value int64 `json:"value" title:"Current Value"`
Previous int64 `json:"previous" title:"Previous Value"`
}
// Control is the _control port schema: current value + Reset button
type Control struct {
Value int64 `json:"value" readonly:"true" title:"Current Value"`
Reset bool `json:"reset" format:"button" title:"Reset Counter" required:"true"`
}
type Component struct {
module.Base // satisfies Stateful, IdentityAware, ClientAware, EmitterAware
mu sync.Mutex
settings Settings
}
func (c *Component) Instance() module.Component {
return &Component{}
}
func (c *Component) GetInfo() module.ComponentInfo {
return module.ComponentInfo{
Name: ComponentName,
Description: "Stateful Counter",
Info: "Counter whose value survives pod restarts and is shared across module replicas. Increment via the increment port; reset from the dashboard.",
Tags: []string{"counter", "state", "stateful"},
}
}
// OnSettings receives Settings from the SettingsPort.
func (c *Component) OnSettings(_ context.Context, msg any) error {
in, ok := msg.(Settings)
if !ok {
return fmt.Errorf("invalid settings")
}
c.mu.Lock()
defer c.mu.Unlock()
c.settings = in
return nil
}
// readCount loads the persisted value, falling back to the configured initial value.
func (c *Component) readCount(ctx context.Context) (int64, error) {
st := c.State()
if st == nil {
return 0, fmt.Errorf("state backend not injected yet")
}
raw, ok, err := st.Get(ctx, stateKeyCount)
if err != nil {
return 0, err
}
if !ok {
return c.settings.InitialValue, nil
}
return strconv.ParseInt(string(raw), 10, 64)
}
func (c *Component) writeCount(ctx context.Context, v int64) error {
return c.State().Set(ctx, stateKeyCount, []byte(strconv.FormatInt(v, 10)))
}
// Handle processes increment messages and returns a module.Result.
func (c *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
if port != IncrementPort {
return module.Fail(fmt.Errorf("unknown port: %s", port))
}
in, ok := msg.(IncrementInput)
if !ok {
return module.Fail(fmt.Errorf("invalid message: %T", msg))
}
// Serialize mutations within this instance; the leader gate below
// serializes across replicas.
c.mu.Lock()
defer c.mu.Unlock()
previous, err := c.readCount(ctx)
if err != nil {
return module.Fail(err)
}
amount := in.Amount
if amount == 0 {
amount = 1
}
value := previous + amount
if err := c.writeCount(ctx, value); err != nil {
return module.Fail(err)
}
// Refresh the dashboard widget in the background loop-safe way
c.Emit(ctx, v1alpha1.ControlPort, Control{Value: value, Reset: true})
// Emit the new value downstream and return the handler's Result
return output(ctx, ValuePort, CountOutput{
Context: in.Context,
Value: value,
Previous: previous,
})
}
// OnControl handles the Reset button (ControlHandler capability).
func (c *Component) OnControl(ctx context.Context, msg any) error {
ctrl, ok := msg.(Control)
if !ok || !ctrl.Reset {
return nil
}
// Only the leader mutates shared state from control clicks
if !utils.IsLeader(ctx) {
return nil
}
c.mu.Lock()
defer c.mu.Unlock()
if err := c.writeCount(ctx, c.settings.InitialValue); err != nil {
return err
}
c.Emit(ctx, v1alpha1.ControlPort, Control{Value: c.settings.InitialValue, Reset: true})
return nil
}
func (c *Component) Ports() []module.Port {
return []module.Port{
{Name: v1alpha1.ReconcilePort}, // state writes flow through reconcile
{
Name: v1alpha1.SettingsPort,
Label: "Settings",
Configuration: c.settings,
},
{
Name: IncrementPort,
Label: "Increment",
Position: module.Left,
Configuration: IncrementInput{},
},
{
Name: ValuePort,
Label: "Value",
Position: module.Right,
Source: true,
Configuration: new(CountOutput),
},
{
Name: v1alpha1.ControlPort,
Label: "Control",
Source: true,
Configuration: Control{Reset: true},
},
}
}
var (
_ module.Component = (*Component)(nil)
_ module.SettingsHandler = (*Component)(nil)
_ module.ControlHandler = (*Component)(nil)
_ module.Stateful = (*Component)(nil) // via embedded module.Base
)
func init() {
registry.Register((&Component{}).Instance())
}
State Propagation Flow
+-------------------------------------------------------------------------+
| Kubernetes Cluster |
| |
| +------------------------------------------------------------------+ |
| | TinyNode CR | |
| | status: | |
| | metadata: | |
| | <state prefix>count: "42" | |
| +---------------------------+--------------------------------------+ |
| | watch (cache updates) |
| +---------------+---------------+ |
| v v v |
| +---------------+ +---------------+ +---------------+ |
| | Pod 1 | | Pod 2 | | Pod 3 | |
| | | | | | | |
| | State.Set -->| | State.Get | | State.Get | |
| | (debounced | | (reads the | | (reads the | |
| | reconcile | | watch cache)| | watch cache)| |
| | patch) | | | | | |
| +---------------+ +---------------+ +---------------+ |
+-------------------------------------------------------------------------+
Key Patterns Demonstrated
1. State via module.State, Not Hand-Rolled CR Writes
raw, ok, err := c.State().Get(ctx, stateKeyCount)
err = c.State().Set(ctx, stateKeyCount, []byte("42"))
Values are opaque []byte — serialize with strconv/encoding/json as needed. An absent key returns (nil, false, nil), not an error.
2. module.Base Removes Lifecycle Boilerplate
Embedding Base satisfies Stateful, IdentityAware, ClientAware, and EmitterAware. Accessors: c.State(), c.Identity(), c.Client(), c.Emit(...).
3. Leader-Gated Control Actions
if !utils.IsLeader(ctx) {
return nil
}
Dashboard clicks fan out to all replicas; only the leader acts.
4. Direct Metadata Access When You Need It
Underneath, State patches status.metadata through the reconcile port. Components with custom needs can use that mechanism directly — the reconcile-port payload is an updater function:
c.Emit(context.Background(), v1alpha1.ReconcilePort, func(n *v1alpha1.TinyNode) error {
if n.Status.Metadata == nil {
n.Status.Metadata = make(map[string]string)
}
n.Status.Metadata["counter-value"] = "42"
return nil
})
Restore in OnReconcile(ctx context.Context, node v1alpha1.TinyNode) error — see the ticker and cron components in common-module for the full restore-after-restart pattern.
5. Durable Run-Scoped State
For state that belongs to a run rather than a node (checkpoints of a durable execution), scope it:
runState := c.State().Scoped(module.ScopeExecution, runID)
Backed by JetStream KV when a broker is configured; degrades safely to node scope otherwise.
Usage Example
Settings
_settings port configuration:
{ "initialValue": 0 }
Increment Edge
Edge mapping into increment:
{ "amount": 1, "context": { "source": "api_request" } }
State Lifecycle
- Injection: framework calls
OnState(viaBase) beforeOnReconcile/OnSettings - Read:
State.Getserves from the watch-updated cache - Mutate: leader processes the message,
State.Setdebounces a patch through the reconcile port - Propagate: other replicas see the new value after the patch commits and the watch event lands
- Recover: after a pod restart the first
Getreads the persisted value — nothing else to do
Best Practices
- Keep state small — it lives in the node's status; store counters, cursors, and flags, not datasets (use the database-module for data)
- Expect eventual consistency across replicas — design operations to be idempotent where possible
- Gate mutations on the leader for control-driven and reconcile-driven writes
- Never block on state in tight loops —
Setis debounced; batch your updates
Extension Ideas
- Operation IDs: dedupe increments by ID for exactly-once semantics
- History: keep a bounded event log under a
history:key prefix (seeState.List) - Execution scope: per-run counters via
Scoped(module.ScopeExecution, runID)