Multi-Replica Coordination

This page provides complete examples of coordinating multiple pod replicas in TinySystems modules.

HTTP Server: Complete Example

The http-module demonstrates multi-replica coordination for HTTP servers.

The Challenge

Multiple pods need to:

1. Listen on the same port
2. Share traffic via load balancing
3. Coordinate port assignment
4. Handle leader failover

Solution Architecture

+-----------------------------------------------------------------------------+
|                    HTTP SERVER MULTI-REPLICA ARCHITECTURE                    |
+-----------------------------------------------------------------------------+

                    +---------------------------------------+
                    |             TinyNode CR               |
                    |                                       |
                    |  status:                              |
                    |    metadata:                          |
                    |      http-server-port: "8080"         |
                    +---------------------------------------+
                           ^                    |
                           | Patch              | Watch
                           |                    v
         +-----------------+-------------------------------------+
         |                                                        |
    +----+----+         +---------+         +---------+         |
    | LEADER  |         | READER  |         | READER  |         |
    |         |         |         |         |         |         |
    | 1. Start|         | Wait for|         | Wait for|         |
    |    :8080|         | port... |         | port... |         |
    |         |         |         |         |         |         |
    | 2. Write|         |         |         |         |         |
    |   "8080"|---------|---------|---------|         |         |
    |         |         |         |         |         |         |
    |         |         | 3. Read |         | 3. Read |         |
    |         |         |   "8080"|         |   "8080"|         |
    |         |         |         |         |         |         |
    | :8080   |         | :8080   |         | :8080   |         |
    +----+----+         +----+----+         +----+----+         |
         |                   |                   |               |
         +-------------------+-------------------+               |
                             |                                   |
                    Kubernetes Service                           |
                    (load balancing)                             |
                             |                                   |
                    External Traffic                             |
                                                                 |
+----------------------------------------------------------------+

Implementation

// server.go
package server

import (
    "context"
    "fmt"
    "net"
    "net/http"
    "strconv"
    "sync"

    "github.com/tiny-systems/module/api/v1alpha1"
    "github.com/tiny-systems/module/module"
    "github.com/tiny-systems/module/pkg/utils"
)

const (
    PortMetadata = "http-server-port"
)

type Server struct {
    settings    Settings
    client      module.Client  // ExposePort / DisclosePort, set in OnClient
    emit        module.Handler // long-lived emitter, set in OnEmitter
    nodeName    string
    currentPort int
    listener    net.Listener
    httpServer  *http.Server
    startStopMu sync.Mutex
}

// Capability interfaces — the framework wires these during node update.

func (s *Server) OnClient(k module.K8sClient) {
    if pc, ok := k.(module.Client); ok {
        s.client = pc
    }
}

func (s *Server) OnEmitter(emit module.Handler) { s.emit = emit }

func (s *Server) OnSettings(ctx context.Context, settings any) error {
    conf, ok := settings.(Settings)
    if !ok {
        return fmt.Errorf("unexpected settings type %T", settings)
    }
    s.settings = conf
    return nil
}

func (s *Server) OnReconcile(ctx context.Context, node v1alpha1.TinyNode) error {
    s.nodeName = node.Name

    // Read configured port from metadata
    configuredPort := 0
    if portStr, ok := node.Status.Metadata[PortMetadata]; ok {
        configuredPort, _ = strconv.Atoi(portStr)
    }

    s.startStopMu.Lock()
    defer s.startStopMu.Unlock()

    // Already running on correct port?
    if configuredPort == s.currentPort && s.currentPort > 0 {
        return nil
    }

    // No port assigned yet?
    if configuredPort == 0 {
        if utils.IsLeader(ctx) {
            // Leader: start and publish port
            return s.startAndPublish(ctx)
        }
        // Reader: wait for port assignment
        return nil
    }

    // Port assigned - start on that port
    return s.startOnPort(ctx, configuredPort)
}

// Handle receives only business ports (e.g. the blocking request port).
func (s *Server) Handle(ctx context.Context, output module.Handler, port string, msg any) module.Result {
    if port == RequestPort {
        return s.handleRequest(ctx, output, msg)
    }
    return module.Ok(nil)
}

func (s *Server) startAndPublish(ctx context.Context) error {
    // Start on random available port
    listener, err := net.Listen("tcp", ":0")
    if err != nil {
        return err
    }

    actualPort := listener.Addr().(*net.TCPAddr).Port
    s.listener = listener
    s.currentPort = actualPort

    // Start HTTP server
    s.httpServer = &http.Server{Handler: s.handler()}
    go s.httpServer.Serve(listener)

    // Publish port to metadata (updater returns error)
    s.emit(ctx, v1alpha1.ReconcilePort, func(node *v1alpha1.TinyNode) error {
        if node.Status.Metadata == nil {
            node.Status.Metadata = make(map[string]string)
        }
        node.Status.Metadata[PortMetadata] = strconv.Itoa(actualPort)
        return nil
    })

    // Expose via Ingress (mutates the module release Service/Ingress)
    if _, err := s.client.ExposePort(ctx, s.nodeName, s.settings.Hostnames, actualPort); err != nil {
        return err
    }

    return nil
}

func (s *Server) startOnPort(ctx context.Context, port int) error {
    // Stop existing server if any
    s.stop()

    // Start on specified port
    listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
    if err != nil {
        return err
    }

    s.listener = listener
    s.currentPort = port

    s.httpServer = &http.Server{Handler: s.handler()}
    go s.httpServer.Serve(listener)

    return nil
}

func (s *Server) stop() {
    if s.httpServer != nil {
        s.httpServer.Close()
        s.httpServer = nil
    }
    if s.listener != nil {
        s.listener.Close()
        s.listener = nil
    }
    s.currentPort = 0
}

Failover Scenario

Timeline of leader failover:

T=0:   Leader (Pod A) running on :8080, published to metadata
       Readers (Pod B, C) running on :8080

T=5:   Pod A crashes

T=5-20: Lease expires, Pod B acquires lease

T=20:  Pod B becomes leader
       Metadata still has "8080" - no change needed
       All pods continue serving on :8080

T=20+: Traffic continues flowing to Pod B and C
       No downtime (except Pod A's share of traffic)

Ticker: Periodic Emission

The ticker component demonstrates leader-only periodic operations.

// ticker.go
type Ticker struct {
    settings   Settings
    emit       module.Handler // set in OnEmitter
    cancelFunc context.CancelFunc
    mu         sync.Mutex
    isRunning  bool
}

func (t *Ticker) OnEmitter(emit module.Handler) { t.emit = emit }

// Control messages fan out to every pod; only the leader acts.
func (t *Ticker) OnControl(ctx context.Context, msg any) error {
    if !utils.IsLeader(ctx) {
        return nil
    }

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

    t.mu.Lock()
    defer t.mu.Unlock()

    if control.Start && !t.isRunning {
        var runCtx context.Context
        runCtx, t.cancelFunc = context.WithCancel(context.Background())
        t.isRunning = true

        // Update UI (control-port emit refreshes the published status)
        t.emit(context.Background(), v1alpha1.ControlPort, nil)

        // Start emitting in the background
        go t.run(runCtx)
        return nil
    }

    if control.Stop && t.isRunning {
        if t.cancelFunc != nil {
            t.cancelFunc()
        }
        t.isRunning = false
        t.emit(context.Background(), v1alpha1.ControlPort, nil)
    }

    return nil
}

func (t *Ticker) run(ctx context.Context) {
    timer := time.NewTimer(time.Duration(t.settings.Delay) * time.Millisecond)
    defer timer.Stop()

    for {
        select {
        case <-timer.C:
            // Emit message (blocks until downstream completes)
            t.emit(ctx, OutPort, t.settings.Context)
            timer.Reset(time.Duration(t.settings.Delay) * time.Millisecond)

        case <-ctx.Done():
            t.mu.Lock()
            t.isRunning = false
            t.mu.Unlock()
            return
        }
    }
}

func (t *Ticker) getControl() Control {
    t.mu.Lock()
    defer t.mu.Unlock()

    return Control{
        IsRunning: t.isRunning,
        // Show Start button if not running, Stop button if running
    }
}

Signal: Manual Trigger

The signal component shows button-based leader-only actions.

// signal.go
type Signal struct {
    settings   Settings
    emit       module.Handler // set in OnEmitter
    cancelFunc context.CancelFunc
    mu         sync.Mutex
}

func (s *Signal) OnEmitter(emit module.Handler) { s.emit = emit }

func (s *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
    }

    s.mu.Lock()
    // Cancel any previous flow
    if s.cancelFunc != nil {
        s.cancelFunc()
        s.cancelFunc = nil
    }

    if control.Reset {
        s.mu.Unlock()
        s.emit(context.Background(), v1alpha1.ControlPort, nil) // refresh UI
        return nil
    }

    if control.Send {
        var runCtx context.Context
        runCtx, s.cancelFunc = context.WithCancel(context.Background())
        s.mu.Unlock()

        // Update UI
        s.emit(context.Background(), v1alpha1.ControlPort, nil)

        // Send to output (blocks until downstream completes)
        res := s.emit(runCtx, OutPort, control.Context)

        // Flow complete
        s.mu.Lock()
        s.cancelFunc = nil
        s.mu.Unlock()

        s.emit(context.Background(), v1alpha1.ControlPort, nil)
        return res.Err()
    }
    s.mu.Unlock()
    return nil
}

Coordination Patterns Summary

Pattern 1: Port Assignment

if utils.IsLeader(ctx) && portNotAssigned {
    port := startOnRandomPort()
    publishPortToMetadata(port)
} else if portAssigned {
    startOnAssignedPort()
}

Pattern 2: Singleton Operation

if utils.IsLeader(ctx) {
    // Only one pod does this
    doExpensiveOperation()
}

Pattern 3: State Machine

if utils.IsLeader(ctx) {
    switch currentState {
    case StateIdle:
        transitionTo(StateRunning)
    case StateRunning:
        doWork()
    }
    publishStateToMetadata()
} else {
    state := readStateFromMetadata()
    actOnState(state)
}

Pattern 4: Resource Ownership

if utils.IsLeader(ctx) {
    createIngressRule()
    createServicePort()
}
// All pods use the created resources

Best Practices

1. Use Mutex for State

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

2. Update UI After State Changes

Emitting to the control port invalidates the runner's port cache and republishes status (debounced), so dashboards see the new control state (e.g. Start ↔ Stop buttons):

emit(context.Background(), v1alpha1.ControlPort, nil)

3. Handle Context Cancellation

select {
case <-ctx.Done():
    return ctx.Err()
case result := <-workDone:
    return result
}

4. Idempotent Operations

if s.currentPort == configuredPort {
    return nil  // Already in desired state
}

Next Steps

• Horizontal Scaling - Scaling best practices
• Leader Election - Election mechanics
• CR-Based State Propagation - State sharing