MLambda.Actors.Core 1.2.0

MLambda Actors

MLambda is a Reactive Actor Model framework for .NET with built-in clustering, gossip protocol, and mTLS security. It provides a lightweight actor system with guardian hierarchy supervision, reactive message passing via System.Reactive, and a clean API built on C# pattern matching.

Install

dotnet add package MLambda.Actors

Features

• Guardian Hierarchy -- Root/System/User/Temp actor tree with parent-child supervision
• Supervision Strategies -- OneForOne (only failed child affected) and AllForOne (all siblings affected)
• Become/Unbecome -- Dynamic behavior switching at runtime
• Message Stashing -- Temporarily buffer messages and replay them later
• DeathWatch -- Watch/Unwatch/Terminated pattern for monitoring actor lifecycle
• Lifecycle Hooks -- PreStart, PostStop, PreRestart, PostRestart callbacks
• Reactive Messaging -- All responses are IObservable<T> via System.Reactive
• DI Integration -- Register actors with Microsoft.Extensions.DependencyInjection

Quick Start

1. Install

Add the MLambda.Actors packages to your project.

2. Register Services

var services = new ServiceCollection();
services.AddActor();                    // Core actor system
services.AddActor<MyActor>();           // Register your actors

3. Define an Actor

Inherit from Actor and override Receive using C# pattern matching:

[Route("/counter")]
public class CounterActor : Actor
{
    private int count;

    protected override Behavior Receive(object data) =>
        data switch
        {
            Increment _ => Actor.Behavior<int>(this.HandleIncrement),
            GetCount _  => Actor.Behavior<int>(() => Observable.Return(this.count)),
            _           => Actor.Ignore,
        };

    private IObservable<int> HandleIncrement()
    {
        this.count++;
        return Observable.Return(this.count);
    }
}

4. Spawn and Send Messages

// Inject IUserContext via DI
var address = await userContext.Spawn<CounterActor>();

// Synchronous send (request-response)
int count = await address.Send<Increment, int>(new Increment());

// Fire-and-forget
address.Send(new Increment()).Subscribe();

Concepts

Behavior and Pattern Matching

The Receive method returns a Behavior delegate (IObservable<object> Behavior(IContext context)). Use Actor.Behavior(...) factory methods to create behaviors from your handler methods:

protected override Behavior Receive(object data) =>
    data switch
    {
        string message => Actor.Behavior(this.Print, message),   // with parameter
        int value      => Actor.Behavior(this.Process, value),   // typed parameter
        _              => Actor.Ignore,                          // unhandled
    };

Handler methods can optionally accept IContext as the first parameter to access the actor context (Self, Spawn, Watch, etc.).

Become / Unbecome

Actors can switch their message handling behavior at runtime:

public class BecomeActor : Actor
{
    protected override Behavior Receive(object data) =>
        data switch
        {
            SetMood m when m.Mood == "happy" => Actor.Behavior(this.SwitchToHappy),
            AskMood _ => Actor.Behavior<string>(() => Observable.Return("normal")),
            _ => Actor.Ignore,
        };

    private IObservable<string> SwitchToHappy(IContext ctx)
    {
        this.Become(this.HappyBehavior);  // Switch behavior
        return Observable.Return("now happy");
    }

    private Behavior HappyBehavior(object data) =>
        data switch
        {
            AskMood _ => Actor.Behavior<string>(() => Observable.Return("happy")),
            SetMood m when m.Mood == "normal" => Actor.Behavior(this.RevertToNormal),
            _ => Actor.Ignore,
        };

    private IObservable<string> RevertToNormal(IContext ctx)
    {
        this.Unbecome();  // Revert to default Receive
        return Observable.Return("back to normal");
    }
}

Message Stashing

Actors can stash messages they cannot yet handle and replay them later. This is commonly combined with Become for initialization patterns:

public class StashActor : Actor
{
    private readonly List<string> processed = new();

    protected override Behavior Receive(object data) =>
        data switch
        {
            Initialize _ => Actor.Behavior(this.HandleInit),
            string _     => Actor.Behavior(this.StashIt),    // Not ready yet
            _ => Actor.Ignore,
        };

    private IObservable<string> HandleInit(IContext ctx)
    {
        this.Become(this.ReadyBehavior);
        this.UnstashAll();  // Replay all stashed messages
        return Observable.Return("initialized");
    }

    private IObservable<string> StashIt(IContext ctx)
    {
        this.Stash?.Stash();  // Buffer current message
        return Observable.Return("stashed");
    }

    private Behavior ReadyBehavior(object data) =>
        data switch
        {
            string msg => Actor.Behavior(this.Process, msg),
            _ => Actor.Ignore,
        };

    private IObservable<string> Process(string msg)
    {
        this.processed.Add(msg);
        return Observable.Return($"processed: {msg}");
    }
}

Supervision Strategies

Define how parent actors handle child failures:

OneForOne -- Only the failing child is affected:

public class MyActor : Actor
{
    public override ISupervisor Supervisor => Strategy.OneForOne(
        decider => decider
            .When<InvalidOperationException>(Directive.Resume)
            .When<InvalidCastException>(Directive.Restart)
            .Default(Directive.Escalate));
}

AllForOne -- All sibling children are affected when one fails:

public class ParentActor : Actor
{
    private readonly IBucket bucket;

    public ParentActor(IBucket bucket) => this.bucket = bucket;

    public override ISupervisor Supervisor => Strategy.AllForOne(
        decider => decider
            .When<InvalidOperationException>(Directive.Resume)
            .When<InvalidCastException>(Directive.Restart)
            .Default(Directive.Escalate),
        this.bucket);
}

Directives: Resume (ignore error), Restart (recreate actor), Stop (terminate), Escalate (pass to parent).

DeathWatch

Monitor another actor's lifecycle:

protected override Behavior Receive(object data) =>
    data switch
    {
        WatchTarget t  => Actor.Behavior(this.StartWatching, t),
        Terminated t   => Actor.Behavior(this.OnTerminated, t),
        _ => Actor.Ignore,
    };

private IObservable<string> StartWatching(IContext ctx, WatchTarget target)
{
    ctx.Watch(target.Address);  // Register for Terminated notifications
    return Observable.Return("watching");
}

private IObservable<string> OnTerminated(Terminated t)
{
    // The watched actor has stopped
    return Observable.Return("target terminated");
}

Lifecycle Hooks

Override lifecycle methods to execute code at specific points:

public class MyActor : Actor
{
    public override void PreStart()    { /* Before first message */ }
    public override void PostStop()    { /* After actor stops */ }
    public override void PreRestart(Exception reason)  { /* Before restart */ }
    public override void PostRestart(Exception reason) { /* After restart */ }

    protected override Behavior Receive(object data) => Actor.Ignore;
}

Packages

Examples

• Standalone Demo -- Single-process actor system without external dependencies.
• HelloWorld -- Minimalist "Hello World" actor example.
• Player Demo -- Advanced distributed gaming demo with clustering and mTLS.

Architecture

Asteroid ──TCP──► Cluster ◄──Gossip──► Cluster
                     │                     │
                 Satellite             Satellite
                 (Actors)              (Actors)

• Cluster: Gossip mesh + CRDT routing + service discovery
• Satellite: Hosts user actors, connects to cluster
• Asteroid: Lightweight gateway that routes messages to the cluster
• Fortress: Optional mTLS layer with PSK bootstrap and auto-rotating X.509 certificates

Project Structure

src/
  MLambda.Actors.Abstraction/   # Interfaces and base classes
  MLambda.Actors/               # Core implementation
  MLambda.Actors.Core/          # DI registration
  MLambda.Actors.Network/       # TCP transport
  MLambda.Actors.Gossip/        # Gossip protocol
  MLambda.Actors.Cluster/       # Cluster system actors
  MLambda.Actors.Satellite/     # Satellite node architecture
  MLambda.Actors.Asteroids/     # Lightweight gateway
  MLambda.Actors.Fortress/      # mTLS security layer
  MLambda.Actors.Monitoring/    # OpenTelemetry integration
test/
  MLambda.Actors.Test/          # SpecFlow BDD tests

Testing

Tests use SpecFlow (BDD) with Gherkin feature files:

dotnet test

Example scenario:

Feature: Message stashing with Become

  Scenario: Messages sent before initialization are stashed and processed after
    Given a stash actor
    When the messages "alpha", "beta", "gamma" are sent
    And the actor is initialized
    And the processed messages are queried
    Then the processed messages should be "alpha", "beta", "gamma"

Kubernetes Development Workflow

Kind Cluster Setup

Create a local Kind cluster for development:

kind create cluster --name mlambda

Building and Loading Images

Build the Docker image and load it into Kind. Always use unique tags — Kind caches :latest aggressively and may not pick up new builds:

# Build with a unique tag
docker build -f src/MLambda.Actors.Cluster.Service/Dockerfile -t mlambda/cluster-console:v1 .

# Load into Kind (required — Kind can't pull from local Docker)
kind load docker-image mlambda/cluster-console:v1 --name mlambda

# Update the running deployment
kubectl set image statefulset/mlambda-cluster mlambda-cluster=mlambda/cluster-console:v1 -n mlambda-system
kubectl rollout status statefulset/mlambda-cluster -n mlambda-system

For subsequent builds, increment the tag (v2, v3, etc.) to force Kind to use the new image.

Port Forwarding

To connect a local Asteroid client to cluster pods inside Kind:

# Forward two cluster pods (run each in a separate terminal or background)
kubectl port-forward -n mlambda-system mlambda-cluster-0 15001:5000
kubectl port-forward -n mlambda-system mlambda-cluster-1 15002:5000

Then set CLUSTER_NODES for the demo client:

CLUSTER_NODES="localhost,15001;localhost,15002" dotnet run --project demo/MLambda.Actors.Demo.Player

Note: The CLUSTER_NODES format is host,port pairs separated by ; (not host:port).

Windows caveat: Background port-forward processes can die silently. If connections fail, kill stale port-forward processes and restart them.

Scaling Satellites

# Scale up
kubectl scale deployment mlambda-satellite -n mlambda-system --replicas=3

# Scale down
kubectl scale deployment mlambda-satellite -n mlambda-system --replicas=1

After scaling, wait ~10 seconds for heartbeats to propagate. The cluster automatically purges stale satellite entries and re-routes actors to live satellites.

Viewing Cluster Logs

# All cluster pod logs
kubectl logs -n mlambda-system -l app=mlambda-cluster --tail=50

# Follow a specific pod
kubectl logs -n mlambda-system mlambda-cluster-0 -f

# Satellite logs
kubectl logs -n mlambda-system -l app=mlambda-satellite --tail=50

Troubleshooting Distributed Clusters

Resilience Architecture

The cluster uses layered supervision to handle failures in distributed actor creation and message delivery:

1. Transport-level timeouts (TcpTransport.Send) — Every TCP send has a 5-second timeout via CancellationTokenSource. If a node is unreachable, the send fails fast instead of hanging indefinitely.

2. Actor creation supervision (CreatorActor) — When a satellite is asked to create an actor, a 10-second timer starts. If the satellite doesn't respond in time, CreatorActor sends a failed CreateActorResponse back to RouteActor, which resets the route to Dead for retry on another satellite.

3. Stale satellite purging (RouteActor.PurgeStaleSatellites) — Satellites have a LastSeen timestamp updated on every heartbeat. Entries older than 15 seconds are removed, and their routes are marked Dead so actors get recreated on live satellites.

4. Route validation on dispatch (RouteActor.HandleDispatchWork) — Before dispatching to a satellite, the route actor verifies the target satellite is still in the live satellites list. If the satellite is gone (e.g., after scale-down), the route is marked Dead and the actor is recreated elsewhere.

5. Periodic re-registration (NodeLifecycleService.SendHeartbeats) — Satellites re-send their registration message alongside every heartbeat (every 5 seconds). This ensures all cluster nodes eventually learn about all satellites, even if the initial registration was lost.

Common Issues

Debugging Strategy

1. Add diagnostic logging — Temporarily add Console.WriteLine at key points: RouteActor.HandleDispatchWork, CreatorActor.HandleCreateActorRequest, NodeLifecycleService.SendHeartbeats.
2. Check cluster logs — Look for satellite registration, route table updates, and timeout errors across all cluster pods.
3. Verify satellite liveness — Check that satellites appear in heartbeat logs on all cluster nodes (not just one).
4. Inspect route table state — Log route status (Running, Waiting, Dead) transitions to identify stuck or stale entries.
5. Remove diagnostics — Clean up Console.WriteLine calls after fixing.

License

This project is licensed under the MIT License. See LICENSE for details.

<p align="center"> <a href="https://www.buymeacoffee.com/yordivad" target="_blank"> <img src="https://cdn.buymeacoffee.com/buttons/default-orange.png" alt="Buy Me A Coffee" width="217px" height="51px"> </a> </p>