Cluster of bots

[Warrolen]

I'm building a grid trading bot for my personal portfolio that should be capable of handling millions of grid bots simultaneously. Each bot would subscribe to several market data streams through the websocket protocol.

To get started with Akka.NET, I set up the dotnet new akkawebapi template.

Here is what I did so far:

[ApiController]
[Route("[controller]")]
public class BotController(IRequiredActor<BotActor> botActor) : ControllerBase
{
    private readonly IActorRef _botActor = botActor.ActorRef;

    [HttpPost("start/{botId}")]
    public async Task<IActionResult> Start(string botId)
    {
        var response = await _botActor.Ask<BotOperationResponse>(new StartBotCommand(botId), TimeSpan.FromSeconds(5));
        return response.Success ? Ok(response) : BadRequest(response.Message);
    }

    [HttpPost("stop/{botId}")]
    public async Task<IActionResult> Stop(string botId)
    {
        var response = await _botActor.Ask<BotOperationResponse>(new StopBotCommand(botId), TimeSpan.FromSeconds(5));
        return response.Success ? Ok(response) : BadRequest(response.Message);
    }

    // TODO: How do I fetch a specific bot by ID? How do I list all bot instances?
}

// AkkaConfiguration.cs
public static AkkaConfigurationBuilder ConfigureBotActors(this AkkaConfigurationBuilder builder,
    IServiceProvider serviceProvider)
{
    var settings = serviceProvider.GetRequiredService<AkkaSettings>();
    var botMessageExtractor = CreateBotMessageRouter();

    if (settings.UseClustering)
    {
        return builder.WithShardRegion<BotActor>("bot",
            (system, registry, resolver) => entityId => Props.Create(() => new BotActor(entityId)),
            botMessageExtractor, settings.ShardOptions);
    }
    else
    {
        return builder.WithActors((system, registry, resolver) =>
        {
            var botActor = system.ActorOf(Props.Create<BotActor>(), "botActor");
            registry.Register<BotActor>(botActor);
        });
    }
}

public static HashCodeMessageExtractor CreateBotMessageRouter() =>
    HashCodeMessageExtractor.Create(
        maxNumberOfShards: 30,
        entityIdExtractor: msg =>
        {
            return msg switch
            {
                IWithBotId botId => botId.BotId,
                _ => null
            };
        },
        messageExtractor: msg => msg);

// BotActor.cs
public class BotActor : ReceivePersistentActor
{
    private readonly HashSet<IActorRef> _subscribers = [];
    private BotState _botState;
    private readonly ILoggingAdapter _log = Context.GetLogger();

    public BotActor(string botId)
    {
        PersistenceId = $"Bot_{botId}";
        _botState = new BotState(botId, false);

        Recover<BotStarted>(@event =>
        {
            _botState = _botState with { IsRunning = true };
        });

        Recover<BotStopped>(@event =>
        {
            _botState = _botState with { IsRunning = false };
        });

        Command<StartBotCommand>(cmd =>
        {
            if (!_botState.IsRunning)
            {
                var @event = new BotStarted(cmd.BotId);
                Persist(@event, e =>
                {
                    _botState = _botState with { IsRunning = true };
                    _log.Info("Bot started: {0}", cmd.BotId);
                    NotifySubscribers(@event);
                    Sender.Tell(new BotOperationResponse(cmd.BotId, true, "Bot started successfully"));
                });
            }
            else
            {
                Sender.Tell(new BotOperationResponse(cmd.BotId, false, "Bot is already running"));
            }
        });

        Command<StopBotCommand>(cmd =>
        {
            if (_botState.IsRunning)
            {
                var @event = new BotStopped(cmd.BotId);
                Persist(@event, e =>
                {
                    _botState = _botState with { IsRunning = false };
                    _log.Info("Bot stopped: {0}", cmd.BotId);
                    NotifySubscribers(@event);
                    Sender.Tell(new BotOperationResponse(cmd.BotId, true, "Bot stopped successfully"));
                });
            }
            else
            {
                Sender.Tell(new BotOperationResponse(cmd.BotId, false, "Bot is not running"));
            }
        });
    }

    private void NotifySubscribers(IBotEvent @event)
    {
        foreach (var subscriber in _subscribers)
        {
            subscriber.Tell(@event);
        }
    }

    public override string PersistenceId { get; }
}

public record BotState(string BotId, bool IsRunning);

public record BotOperationResponse(string BotId, bool Success, string Message);

// BotCommands.cs
public interface IBotCommand : IWithBotId;

public sealed record StartBotCommand(string BotId) : IBotCommand;
public sealed record StopBotCommand(string BotId) : IBotCommand;

// BotEvents.cs
public interface IBotEvent : IWithBotId;

public sealed record BotStarted(string BotId) : IBotEvent;
public sealed record BotStopped(string BotId) : IBotEvent;

// BotQueries.cs
public interface IBotQuery : IWithBotId;

public sealed record FetchBot(string BotId) : IBotQuery;

// IWithBotId.cs
public interface IWithBotId
{
    string BotId { get; }
}

The question is how do I add:

1. an endpoint that fetches a specific bot by ID
2. an endpoint that fetches all bot instances and their states

I'm not sure how to design it properly. I know that when I call the API endpoint to start a bot, it does not necessarily spawn a new BotActor instance. These actors are basically created, managed, and interacted with as if they are local, regardless of their actual location in the cluster.

https://github.com/Warrolen/grid