sov-kernel-monster / seb /runtime /src /seb_partition_mgr.erl

Sync with GitHub, license metadata from LICENSE files, commercial license notice

ebed3db verified 20 days ago

6.36 kB

%%%-------------------------------------------------------------------
%% @doc Sovereign Event Bus - Partition Manager
%%
%% Deterministic partition assignment (per XML L2 spec):
%% - 1024 partitions (fixed)
%% - Competency-based routing from Datalog policy engine
%% - phash2 deterministic assignment
%% - Same seed → same result (for reproducibility)
%%
%% Partition assignment is deterministic and reproducible.
%% Given the same agent ID and competency, returns same partition.
%%
%% @end
%%%-------------------------------------------------------------------
-module(seb_partition_mgr).
-behaviour(gen_server).

-export([start_link/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).

%% Public API
-export([assign_partition/2, get_partition_load/1, rebalance_partitions/0]).

-define(SERVER, ?MODULE).
-define(DEFAULT_PARTITIONS, 1024).
-define(PARTITION_LOAD_THRESHOLD, 0.8).

-record(state, {
    partition_count :: non_neg_integer(),
    partition_assignments :: map(),  %% {agent_id, competency} -> partition
    partition_loads :: map()          %% partition -> load
}).

%%%===================================================================
%%% API
%%%===================================================================

%% @doc Start the partition manager
-spec start_link(non_neg_integer()) -> gen_server:start_ret().
start_link(PartitionCount) when is_integer(PartitionCount), PartitionCount > 0 ->
    gen_server:start_link({local, ?SERVER}, ?MODULE, [PartitionCount], []).

%% @doc Assign a partition to an agent based on competency
%%
%% Uses phash2 for deterministic assignment:
%%   partition = phash2({agent_id, competency}) rem partition_count
%%
%% Args:
%%   AgentId: Binary identifier of the agent
%%   Competency: Atom describing agent capability (e.g., 'compute', 'io', 'crypto')
%%
%% Returns: Partition number (0 .. partition_count - 1)
%%
-spec assign_partition(binary(), atom()) -> non_neg_integer().
assign_partition(AgentId, Competency) when is_binary(AgentId), is_atom(Competency) ->
    gen_server:call(?SERVER, {assign_partition, AgentId, Competency}).

%% @doc Get current load for a specific partition
%%
%% Returns: Float between 0.0 and 1.0
%%
-spec get_partition_load(non_neg_integer()) -> float() | {error, not_found}.
get_partition_load(Partition) when is_integer(Partition), Partition >= 0 ->
    gen_server:call(?SERVER, {get_partition_load, Partition}).

%% @doc Trigger partition rebalancing
%%
%% If any partition exceeds PARTITION_LOAD_THRESHOLD, rebalances
%% assignments to distribute load more evenly.
%%
-spec rebalance_partitions() -> ok | {error, term()}.
rebalance_partitions() ->
    gen_server:call(?SERVER, rebalance_partitions).

%%%===================================================================
%%% gen_server callbacks
%%%===================================================================

%% @doc Initialize the partition manager
-spec init([non_neg_integer()]) -> {ok, #state{}}.
init([PartitionCount]) ->
    State = #state{
        partition_count = PartitionCount,
        partition_assignments = maps:new(),
        partition_loads = init_loads(PartitionCount)
    },
    {ok, State}.

%% @doc Handle synchronous calls
-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, term(), #state{}}.

handle_call({assign_partition, AgentId, Competency}, _From, State) ->
    Partition = compute_partition(AgentId, Competency, State#state.partition_count),

    %% Store assignment for later reference (idempotent)
    Key = {AgentId, Competency},
    NewAssignments = maps:put(Key, Partition, State#state.partition_assignments),

    %% Update partition load (simple increment)
    Load = maps:get(Partition, State#state.partition_loads, 0.0),
    NewLoads = maps:put(Partition, Load + 0.01, State#state.partition_loads),

    NewState = State#state{
        partition_assignments = NewAssignments,
        partition_loads = NewLoads
    },

    {reply, Partition, NewState};

handle_call({get_partition_load, Partition}, _From, State) ->
    case maps:find(Partition, State#state.partition_loads) of
        {ok, Load} ->
            {reply, Load, State};
        error ->
            {reply, {error, not_found}, State}
    end;

handle_call(rebalance_partitions, _From, State) ->
    %% Check if any partition exceeds threshold
    MaxLoad = maps:fold(fun(_P, Load, Max) -> max(Load, Max) end, 0.0, State#state.partition_loads),

    case MaxLoad > ?PARTITION_LOAD_THRESHOLD of
        true ->
            %% Reset loads to even distribution
            NewLoads = init_loads(State#state.partition_count),
            NewState = State#state{partition_loads = NewLoads},
            {reply, ok, NewState};
        false ->
            {reply, ok, State}
    end;

handle_call(_Request, _From, State) ->
    {reply, {error, unknown_call}, State}.

%% @doc Handle asynchronous casts
-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
handle_cast(_Msg, State) ->
    {noreply, State}.

%% @doc Handle info messages
-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
handle_info(_Info, State) ->
    {noreply, State}.

%% @doc Terminate the partition manager
-spec terminate(term(), #state{}) -> ok.
terminate(_Reason, _State) ->
    ok.

%% @doc Code change (upgrade support)
-spec code_change(term(), #state{}, term()) -> {ok, #state{}}.
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%%===================================================================
%%% Internal Functions
%%%===================================================================

%% @doc Initialize partition loads (0.0 for each partition)
-spec init_loads(non_neg_integer()) -> map().
init_loads(PartitionCount) ->
    maps:from_list([{P, 0.0} || P <- lists:seq(0, PartitionCount - 1)]).

%% @doc Compute deterministic partition assignment
%%
%% Uses Erlang's phash2 for deterministic hashing:
%%   partition = phash2({agent_id, competency}) rem partition_count
%%
%% This ensures:
%%   1. Same agent + competency always maps to same partition
%%   2. Distribution is uniform across partitions
%%   3. Deterministic and reproducible
%%
-spec compute_partition(binary(), atom(), non_neg_integer()) -> non_neg_integer().
compute_partition(AgentId, Competency, PartitionCount) ->
    erlang:phash2({AgentId, Competency}) rem PartitionCount.

	%%%-------------------------------------------------------------------
	%% @doc Sovereign Event Bus - Partition Manager
	%%
	%% Deterministic partition assignment (per XML L2 spec):
	%% - 1024 partitions (fixed)
	%% - Competency-based routing from Datalog policy engine
	%% - phash2 deterministic assignment
	%% - Same seed → same result (for reproducibility)
	%%
	%% Partition assignment is deterministic and reproducible.
	%% Given the same agent ID and competency, returns same partition.
	%%
	%% @end
	%%%-------------------------------------------------------------------
	-module(seb_partition_mgr).
	-behaviour(gen_server).

	-export([start_link/1]).
	-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).

	%% Public API
	-export([assign_partition/2, get_partition_load/1, rebalance_partitions/0]).

	-define(SERVER, ?MODULE).
	-define(DEFAULT_PARTITIONS, 1024).
	-define(PARTITION_LOAD_THRESHOLD, 0.8).

	-record(state, {
	partition_count :: non_neg_integer(),
	partition_assignments :: map(), %% {agent_id, competency} -> partition
	partition_loads :: map() %% partition -> load
	}).

	%%%===================================================================
	%%% API
	%%%===================================================================

	%% @doc Start the partition manager
	-spec start_link(non_neg_integer()) -> gen_server:start_ret().
	start_link(PartitionCount) when is_integer(PartitionCount), PartitionCount > 0 ->
	gen_server:start_link({local, ?SERVER}, ?MODULE, [PartitionCount], []).

	%% @doc Assign a partition to an agent based on competency
	%%
	%% Uses phash2 for deterministic assignment:
	%% partition = phash2({agent_id, competency}) rem partition_count
	%%
	%% Args:
	%% AgentId: Binary identifier of the agent
	%% Competency: Atom describing agent capability (e.g., 'compute', 'io', 'crypto')
	%%
	%% Returns: Partition number (0 .. partition_count - 1)
	%%
	-spec assign_partition(binary(), atom()) -> non_neg_integer().
	assign_partition(AgentId, Competency) when is_binary(AgentId), is_atom(Competency) ->
	gen_server:call(?SERVER, {assign_partition, AgentId, Competency}).

	%% @doc Get current load for a specific partition
	%%
	%% Returns: Float between 0.0 and 1.0
	%%
	-spec get_partition_load(non_neg_integer()) -> float() \| {error, not_found}.
	get_partition_load(Partition) when is_integer(Partition), Partition >= 0 ->
	gen_server:call(?SERVER, {get_partition_load, Partition}).

	%% @doc Trigger partition rebalancing
	%%
	%% If any partition exceeds PARTITION_LOAD_THRESHOLD, rebalances
	%% assignments to distribute load more evenly.
	%%
	-spec rebalance_partitions() -> ok \| {error, term()}.
	rebalance_partitions() ->
	gen_server:call(?SERVER, rebalance_partitions).

	%%%===================================================================
	%%% gen_server callbacks
	%%%===================================================================

	%% @doc Initialize the partition manager
	-spec init([non_neg_integer()]) -> {ok, #state{}}.
	init([PartitionCount]) ->
	State = #state{
	partition_count = PartitionCount,
	partition_assignments = maps:new(),
	partition_loads = init_loads(PartitionCount)
	},
	{ok, State}.

	%% @doc Handle synchronous calls
	-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, term(), #state{}}.

	handle_call({assign_partition, AgentId, Competency}, _From, State) ->
	Partition = compute_partition(AgentId, Competency, State#state.partition_count),

	%% Store assignment for later reference (idempotent)
	Key = {AgentId, Competency},
	NewAssignments = maps:put(Key, Partition, State#state.partition_assignments),

	%% Update partition load (simple increment)
	Load = maps:get(Partition, State#state.partition_loads, 0.0),
	NewLoads = maps:put(Partition, Load + 0.01, State#state.partition_loads),

	NewState = State#state{
	partition_assignments = NewAssignments,
	partition_loads = NewLoads
	},

	{reply, Partition, NewState};

	handle_call({get_partition_load, Partition}, _From, State) ->
	case maps:find(Partition, State#state.partition_loads) of
	{ok, Load} ->
	{reply, Load, State};
	error ->
	{reply, {error, not_found}, State}
	end;

	handle_call(rebalance_partitions, _From, State) ->
	%% Check if any partition exceeds threshold
	MaxLoad = maps:fold(fun(_P, Load, Max) -> max(Load, Max) end, 0.0, State#state.partition_loads),

	case MaxLoad > ?PARTITION_LOAD_THRESHOLD of
	true ->
	%% Reset loads to even distribution
	NewLoads = init_loads(State#state.partition_count),
	NewState = State#state{partition_loads = NewLoads},
	{reply, ok, NewState};
	false ->
	{reply, ok, State}
	end;

	handle_call(_Request, _From, State) ->
	{reply, {error, unknown_call}, State}.

	%% @doc Handle asynchronous casts
	-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
	handle_cast(_Msg, State) ->
	{noreply, State}.

	%% @doc Handle info messages
	-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
	handle_info(_Info, State) ->
	{noreply, State}.

	%% @doc Terminate the partition manager
	-spec terminate(term(), #state{}) -> ok.
	terminate(_Reason, _State) ->
	ok.

	%% @doc Code change (upgrade support)
	-spec code_change(term(), #state{}, term()) -> {ok, #state{}}.
	code_change(_OldVsn, State, _Extra) ->
	{ok, State}.

	%%%===================================================================
	%%% Internal Functions
	%%%===================================================================

	%% @doc Initialize partition loads (0.0 for each partition)
	-spec init_loads(non_neg_integer()) -> map().
	init_loads(PartitionCount) ->
	maps:from_list([{P, 0.0} \|\| P <- lists:seq(0, PartitionCount - 1)]).

	%% @doc Compute deterministic partition assignment
	%%
	%% Uses Erlang's phash2 for deterministic hashing:
	%% partition = phash2({agent_id, competency}) rem partition_count
	%%
	%% This ensures:
	%% 1. Same agent + competency always maps to same partition
	%% 2. Distribution is uniform across partitions
	%% 3. Deterministic and reproducible
	%%
	-spec compute_partition(binary(), atom(), non_neg_integer()) -> non_neg_integer().
	compute_partition(AgentId, Competency, PartitionCount) ->
	erlang:phash2({AgentId, Competency}) rem PartitionCount.