Leader-Reader Pattern

The leader-reader pattern is how TinySystems modules coordinate across multiple replicas. Understanding this pattern is essential for building scalable components.

Overview

+-----------------------------------------------------------------------+
|                        LEADER-READER PATTERN                          |
+-----------------------------------------------------------------------+

+-----------------------------------------------------------------------+
|                          KUBERNETES LEASE                             |
|                                                                       |
|   Only one pod holds the lease at a time                              |
|   Lease holder = LEADER                                               |
|   Other pods = READERS                                                |
+-----------------------------------------------------------------------+
                                   |
         +-------------------------+-------------------------+
         |                         |                         |
         v                         v                         v
  +-------------+           +-------------+           +-------------+
  |   LEADER    |           |   READER    |           |   READER    |
  |             |           |             |           |             |
  | +---------+ |           | +---------+ |           | +---------+ |
  | |  Write  | |           | |  Watch  | |           | |  Watch  | |
  | |   CRs   | |           | |  Only   | |           | |  Only   | |
  | +---------+ |           | +---------+ |           | +---------+ |
  |             |           |             |           |             |
  | +---------+ |           | +---------+ |           | +---------+ |
  | | Act on  | |           | | Handle  | |           | | Handle  | |
  | | Control | |           | |Messages | |           | |Messages | |
  | +---------+ |           | +---------+ |           | +---------+ |
  |             |           |             |           |             |
  | +---------+ |           | +---------+ |           | +---------+ |
  | | Update  | |           | |  Local  | |           | |  Local  | |
  | | Status  | |           | |Reconcile| |           | |Reconcile| |
  | +---------+ |           | +---------+ |           | +---------+ |
  +-------------+           +-------------+           +-------------+

Responsibilities

Leader Pod

Reader Pods

Checking Leadership

Leadership is stamped onto the context of lifecycle callbacks (OnControl, OnReconcile):

import "github.com/tiny-systems/module/pkg/utils"

func (c *Component) OnControl(ctx context.Context, msg any) error {
    if utils.IsLeader(ctx) {
        // Leader-only code
    } else {
        // Reader code (or skip)
    }
    return nil
}

Pattern Implementation

Basic Pattern

System ports never reach Handle — each concern lands on its capability interface:

// ALL pods: store settings
func (c *Component) OnSettings(ctx context.Context, settings any) error {
    conf, ok := settings.(Settings)
    if !ok {
        return fmt.Errorf("unexpected settings type %T", settings)
    }
    c.settings = conf
    return nil
}

// LEADER only: control writes fan out to every pod; the check gates action
func (c *Component) OnControl(ctx context.Context, msg any) error {
    if !utils.IsLeader(ctx) {
        return nil
    }
    return c.handleControl(ctx, msg)
}

// ALL pods: read shared state; LEADER also writes it
func (c *Component) OnReconcile(ctx context.Context, node v1alpha1.TinyNode) error {
    return c.handleReconcile(ctx, node)
}

// ALL pods: handle incoming business messages
func (c *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    if port == "input" {
        return c.handleInput(ctx, output, msg)
    }
    return module.Ok(nil)
}

Signal Component Example

// signal.go - Manual flow trigger
func (t *Signal) OnEmitter(emit module.Handler) { t.emit = emit }

func (t *Signal) OnControl(ctx context.Context, msg any) error {
    // Only leader processes button clicks
    if !utils.IsLeader(ctx) {
        return nil
    }

    control, ok := msg.(Control)
    if !ok {
        return nil
    }

    if control.Send {
        // Start the flow
        t.mu.Lock()
        if t.cancelFunc != nil {
            t.cancelFunc()
        }
        var runCtx context.Context
        runCtx, t.cancelFunc = context.WithCancel(context.Background())
        t.mu.Unlock()

        // Send to output via the injected emitter
        return t.emit(runCtx, OutPort, control.Context).Err()
    }

    if control.Reset {
        t.mu.Lock()
        if t.cancelFunc != nil {
            t.cancelFunc()
            t.cancelFunc = nil
        }
        t.mu.Unlock()
    }
    return nil
}

Ticker Component Example

// ticker.go - Periodic emission
func (t *Ticker) OnEmitter(emit module.Handler) { t.emit = emit }

func (t *Ticker) OnControl(ctx context.Context, msg any) error {
    // Only leader starts/stops the ticker
    if !utils.IsLeader(ctx) {
        return nil
    }

    control, ok := msg.(Control)
    if !ok {
        return nil
    }
    if control.Start {
        go t.startEmitting()
    }
    if control.Stop {
        t.stopEmitting()
    }
    return nil
}

func (t *Ticker) startEmitting() {
    t.mu.Lock()
    runCtx, cancel := context.WithCancel(context.Background())
    t.cancelFunc = cancel
    t.mu.Unlock()

    // Emit loop — the emitter from OnEmitter stays valid for the
    // runner's lifetime and is safe to call from goroutines
    timer := time.NewTimer(time.Duration(t.settings.Delay) * time.Millisecond)
    for {
        select {
        case <-timer.C:
            t.emit(runCtx, OutPort, t.settings.Context)
            timer.Reset(time.Duration(t.settings.Delay) * time.Millisecond)
        case <-runCtx.Done():
            return
        }
    }
}

Controller-Level Pattern

Controllers also implement the pattern:

// tinynode_controller.go (simplified)
func (r *TinyNodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    node := &v1alpha1.TinyNode{}
    r.Get(ctx, req.NamespacedName, node)

    // ALL pods: update local scheduler (leadership rides the ctx)
    r.Scheduler.Update(utils.WithLeader(ctx, r.IsLeader.Load()), node)

    // LEADER only: patch CR status
    if !r.IsLeader.Load() {
        // Requeue to stay in sync (and to publish once leadership is won)
        return ctrl.Result{RequeueAfter: time.Minute}, nil
    }

    // Leader: patch status built during Scheduler.Update
    r.Status().Patch(ctx, node, client.MergeFrom(originNode))

    return ctrl.Result{RequeueAfter: 5 * time.Minute}, nil
}

External signals follow the same split at the transport level: system-port writes are published to the fan-out subject tinymodule.<module>.sysmsg, every pod receives them, and each component's in-handler IsLeader check decides whether to act. Business-port messages go to the queue-group subject and are load-balanced across pods. (The former TinySignal controller is gone along with the CRD.)

Failover Handling

When the leader fails:

Time -------------------------------------------------------------->

    Leader Pod A              Reader Pod B              Reader Pod C
         |                         |                         |
         | <-- Lease held          |                         |
         |                         |                         |
         X (Pod A dies)            |                         |
                                   |                         |
         | <-- Lease expires       |                         |
              (15 seconds)         |                         |
                                   |                         |
                                   | Acquires lease          |
                                   | -------------->         |
                                   |                         |
                                   | isLeader = true         |
                                   |                         |
                                   | Continues operations    |
                                   |                         |
                                   v                         v

Code for Failover

type Component struct {
    cancelFunc context.CancelFunc
    mu         sync.Mutex
    wasLeader  bool
}

func (c *Component) OnReconcile(ctx context.Context, node v1alpha1.TinyNode) error {
    isLeader := utils.IsLeader(ctx)

    c.mu.Lock()
    defer c.mu.Unlock()

    if isLeader && !c.wasLeader {
        // Just became leader - take over
        c.onBecameLeader(ctx)
        c.wasLeader = true
    } else if !isLeader && c.wasLeader {
        // Lost leadership - stop leader-only work
        c.onLostLeadership()
        c.wasLeader = false
    }
    return nil
}

func (c *Component) onBecameLeader(ctx context.Context) {
    // Initialize leader-only resources
    log.Info("became leader, taking over")

    // Resume any paused operations
    c.startLeaderOnlyWork(ctx)
}

func (c *Component) onLostLeadership() {
    // Stop leader-only work
    log.Info("lost leadership, stopping leader-only work")

    if c.cancelFunc != nil {
        c.cancelFunc()
        c.cancelFunc = nil
    }
}

Best Practices

1. Don't Assume Continuous Leadership

utils.IsLeader(ctx) reads a value stamped on the context at dispatch time — it does NOT update inside a long-running goroutine. Track leadership across OnReconcile calls and cancel background work when it changes:

// Bad: leadership snapshot from one callback runs work forever
func (c *Component) OnControl(ctx context.Context, msg any) error {
    if utils.IsLeader(ctx) {
        go func() {
            for {
                c.doWork() // Might not be leader anymore!
            }
        }()
    }
    return nil
}

// Good: cancellable work, torn down when a later callback sees
// leadership lost (see "Code for Failover" above)
func (c *Component) OnControl(ctx context.Context, msg any) error {
    if !utils.IsLeader(ctx) {
        return nil
    }
    c.mu.Lock()
    workCtx, cancel := context.WithCancel(context.Background())
    c.cancelFunc = cancel // cancelled in onLostLeadership
    c.mu.Unlock()

    go func() {
        for {
            select {
            case <-workCtx.Done():
                return
            default:
                c.doWork()
            }
        }
    }()
    return nil
}

2. Idempotent State Updates

// Good: Check before updating (updater returns error)
output(ctx, v1alpha1.ReconcilePort, func(node *v1alpha1.TinyNode) error {
    if node.Status.Metadata["initialized"] != "true" {
        // Only initialize once
        node.Status.Metadata["initialized"] = "true"
        c.performInitialization()
    }
    return nil
})

3. Graceful Degradation

// All pods handle business messages
func (c *Component) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    if port == "input" {
        return c.processInput(ctx, output, msg)
    }
    return module.Ok(nil)
}

// But only the leader writes shared state
func (c *Component) OnReconcile(ctx context.Context, node v1alpha1.TinyNode) error {
    if utils.IsLeader(ctx) {
        c.updateStatus(ctx)
    }
    return nil
}

Next Steps

• Multi-Replica Coordination - Complete examples
• Horizontal Scaling - Scaling best practices
• Leader Election - Election mechanics