File size: 6,361 Bytes
ebed3db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 | %%%-------------------------------------------------------------------
%% @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.
|