tiny

Client Pool

The AddressPool (internal/client) manages gRPC connections to remote modules on the fallback wire — it is in play only when the module runs without TINY_NATS_URL. It keeps a module-name → address table, creates clients lazily, pre-warms connections, and shares one connection per remote address.

Overview

+-----------------------------------------------------------------------+
|                       ADDRESS POOL ARCHITECTURE                       |
+-----------------------------------------------------------------------+

                  +---------------------------------------+
                  |             ADDRESS POOL              |
                  |                                       |
                  |  addressTable (module -> addr):       |
                  |  +----------------------------------+ |
                  |  | common-module -> 10.0.4.12:41231 | |
                  |  | http-module   -> 10.0.5.3:38795  | |
                  |  +----------------------------------+ |
                  |                                       |
                  |  clients / conns (addr -> client):    |
                  |  +----------------------------------+ |
                  |  | 10.0.4.12:41231 -> *grpc.Conn    | |
                  |  | 10.0.5.3:38795  -> *grpc.Conn    | |
                  |  +----------------------------------+ |
                  +------------------+--------------------+
                                     |
                       lazy, shared, pre-warmed gRPC
                                connections

The Pool Interface

The controller-facing surface is deliberately tiny:

type Pool interface {
    Register(moduleName, addr string)
    Deregister(moduleName string)
}

AddressPool implements it, plus the send path (Handler) and a lifecycle hook (Start):

type AddressPool struct {
    addressTable cmap.ConcurrentMap[string, string]                     // module -> addr
    clients      cmap.ConcurrentMap[string, module.ModuleServiceClient] // addr -> client
    conns        cmap.ConcurrentMap[string, *grpc.ClientConn]           // addr -> conn
    // ...
}

pool := client.NewPool().SetLogger(l)

Core Operations

Register

Called when a remote module is discovered. Stores the address and pre-warms the connection in the background so the first message doesn't pay the dial cost:

func (p *AddressPool) Register(moduleName, addr string) {
    p.addressTable.Set(moduleName, addr)

    // Pre-warm connection in background to avoid cold start latency
    go func() {
        if _, err := p.getClient(ctx, addr); err != nil {
            // logged; the next send retries
        }
    }()
}

Deregister

Called when a module goes away. Removes the address; the connection is closed only if no other module shares it:

func (p *AddressPool) Deregister(moduleName string) {
    addr, ok := p.addressTable.Get(moduleName)
    p.addressTable.Remove(moduleName)
    if !ok {
        return
    }
    // close conn only when no other module uses the same address
    // (checked by iterating addressTable)
}

Handler — the Send Path

Handler matches the transport contract the scheduler binds to (func(ctx, *runner.Msg) ([]byte, error)):

func (p *AddressPool) Handler(ctx context.Context, msg *runner.Msg) ([]byte, error) {
    moduleName, _, err := module.ParseFullName(msg.To)
    if err != nil {
        return nil, err
    }

    addr, ok := p.addressTable.Get(moduleName)
    if !ok {
        return nil, fmt.Errorf("%s module address is unknown", moduleName)
    }

    client, err := p.getClient(ctx, addr) // lazy: created on first use
    if err != nil {
        return nil, err
    }

    if msg.Depth > 0 {
        md := metadata.Pairs("x-message-depth", strconv.Itoa(msg.Depth))
        ctx = metadata.NewOutgoingContext(ctx, md)
    }

    resp, err := client.Message(ctx, &module.MessageRequest{
        From:    msg.From,
        Payload: msg.Data,
        EdgeID:  msg.EdgeID,
        To:      msg.To,
    })
    if err != nil {
        return nil, err
    }
    return resp.Data, nil
}

Lazy Clients

getClient returns a cached client per address or dials a new one:

conn, err := grpc.NewClient(addr,
    grpc.WithTransportCredentials(insecure.NewCredentials()),
    grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
    grpc.WithKeepaliveParams(keepalive.ClientParameters{
        Time:                10 * time.Second, // ping every 10s if idle
        Timeout:             3 * time.Second,  // wait 3s for ping ack
        PermitWithoutStream: true,             // ping even without active RPCs
    }),
)

// Trigger immediate connection instead of waiting for first RPC
conn.Connect()

Connections are tied to the pool's run context: when Start(ctx)'s context is cancelled, every connection is closed.

Integration with Discovery

The TinyModuleReconciler drives the pool from TinyModule CRs:

func (r *TinyModuleReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
    instance := &operatorv1alpha1.TinyModule{}
    if err := r.Get(ctx, req.NamespacedName, instance); err != nil {
        if errors.IsNotFound(err) {
            r.ClientPool.Deregister(req.Name) // module removed
            return reconcile.Result{}, nil
        }
        return reconcile.Result{}, err
    }

    // Remote module — all pods register so they can send messages to it
    if req.Name != r.Module.GetNameSanitised() {
        if instance.Status.Addr != "" {
            r.ClientPool.Register(req.Name, instance.Status.Addr)
        }
        return ctrl.Result{}, nil
    }

    // Own module — the leader publishes status (addr, version, components)
    // ...
}

Each module advertises its own listen address in its TinyModule status.addr; every other module's reconciler picks it up and registers it.

Connection Lifecycle

1. TinyModule CR created / updated with status.addr
   |
   v
2. Reconciler calls Pool.Register(module, addr)
   |
   +-> address stored; connection pre-warmed in background
   |
   v
3. Messages flow through Handler
   |
   +-> keepalive (10s/3s) maintains health; gRPC auto-reconnects
   |
   v
4. TinyModule CR deleted
   |
   v
5. Reconciler calls Pool.Deregister(module)
   |
   +-> conn closed unless another module shares the address
   |
   v
6. Pool run context cancelled (shutdown) -> all connections closed

Reconnection

Transient failures are handled by gRPC itself — the connection state machine retries with exponential backoff, and keepalive pings (every 10s, 3s timeout, even without active RPCs) detect dead peers. The pool does not run its own health-check loop.

Error Handling

resp, err := client.Message(ctx, req)
if err != nil {
    st, ok := status.FromError(err)
    if ok {
        switch st.Code() {
        case codes.Unavailable:
            // module is down / address stale
        case codes.DeadlineExceeded:
            // timeout — the blocking hop outlived ctx
        }
    }
    return nil, err
}

An unknown module (module address is unknown) means the target's TinyModule CR hasn't been reconciled yet or its status.addr is empty.

Debugging

Log Lines

"address pool: registering module"     module=http-module addr=...
"address pool: connection pre-warmed"  module=http-module addr=...
"address pool: deregistering module"   module=http-module
"address pool: closing connection"     module=http-module addr=...
"grpc client: connection failed"       addr=...

gRPC Client Logging

export GRPC_GO_LOG_VERBOSITY_LEVEL=99
export GRPC_GO_LOG_SEVERITY_LEVEL=info

Best Practices

1. Let the Pool Own Connections

One connection per remote address serves all RPCs to that module — never dial per message.

2. Don't Block Startup on Connectivity

Registration pre-warms in the background and getClient is lazy; a module that starts before its peers simply connects when they appear.

3. Remember It's the Fallback

With TINY_NATS_URL set, cross-module traffic bypasses the pool entirely — debugging connection issues starts with knowing which wire you're on (the startup log prints transport: core req/reply, transport: jetstream-backed durable wire, or falls back to gRPC).

Next Steps