Packages
macula
0.46.0
7.1.0
7.0.0
6.0.0
5.2.2
5.2.1
5.2.0
5.1.0
5.0.0
4.8.0
4.7.1
4.7.0
4.6.0
4.5.0
4.4.10
4.4.9
4.4.8
4.4.7
4.4.6
4.4.5
4.4.4
4.4.3
4.4.2
4.4.1
4.4.0
4.3.1
4.3.0
4.2.9
4.2.8
4.2.7
4.2.6
4.2.5
4.2.4
4.2.3
4.2.2
4.2.1
4.2.0
4.1.1
4.1.0
4.0.0
3.16.0
3.15.3
3.15.2
3.15.1
3.14.0
3.13.0
3.12.1
3.12.0
3.11.1
3.11.0
3.10.3
3.10.2
3.10.1
3.9.0
3.8.0
3.7.0
3.5.0
3.4.0
3.3.0
3.2.0
3.1.0
3.0.0
2.1.1
2.1.0
2.0.0
1.5.2
1.5.1
1.4.30
1.4.29
1.4.28
1.4.27
1.4.26
1.4.25
1.4.24
1.4.23
1.4.22
1.4.21
1.4.20
1.4.19
1.4.18
1.4.17
1.4.16
1.4.15
1.4.14
1.4.13
1.4.11
1.4.10
1.4.9
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.1
1.3.0
1.2.0
1.1.0
1.0.10
1.0.9
1.0.8
1.0.7
1.0.6
1.0.5
1.0.4
1.0.3
1.0.2
1.0.1
1.0.0
0.48.6
0.48.5
0.48.4
0.48.3
0.48.2
0.48.1
0.48.0
0.47.1
0.47.0
0.46.3
0.46.1
0.46.0
0.45.3
0.45.2
0.45.1
0.45.0
0.44.2
0.44.1
0.44.0
0.43.3
0.43.2
0.43.1
0.43.0
0.42.9
0.42.8
0.42.7
0.42.6
0.42.5
0.42.4
0.42.3
0.42.2
0.42.1
0.42.0
0.41.1
0.41.0
0.40.1
0.40.0
0.39.9
0.39.8
0.39.7
0.39.6
0.39.5
0.39.4
0.39.3
0.39.2
0.39.1
0.39.0
0.38.8
0.38.7
0.38.6
0.38.5
0.38.4
0.38.3
0.38.2
0.38.1
0.38.0
0.37.7
0.37.6
0.37.5
0.37.4
0.37.3
0.37.2
0.37.1
0.37.0
0.36.6
0.36.5
0.36.4
0.36.3
0.36.2
0.36.1
0.36.0
0.35.4
0.35.3
0.35.2
0.35.1
0.35.0
0.34.1
0.34.0
0.33.1
0.33.0
0.32.5
0.32.4
0.32.3
0.32.2
0.32.1
0.32.0
0.31.9
0.31.8
0.31.7
0.31.6
0.31.5
0.31.4
0.31.3
0.31.2
0.31.1
0.31.0
0.30.10
0.30.9
0.30.8
0.30.7
0.30.6
0.30.5
0.30.4
0.30.3
0.30.2
0.30.1
0.30.0
0.29.0
0.28.3
0.28.2
0.28.1
0.28.0
0.27.1
0.27.0
0.26.1
0.26.0
0.25.6
0.25.5
0.25.4
0.25.3
0.25.2
0.25.1
0.25.0
0.24.6
0.24.5
0.24.4
0.24.3
0.24.2
0.24.1
0.24.0
0.23.3
0.23.2
0.23.1
0.23.0
0.22.12
0.22.11
0.22.10
0.22.9
0.22.8
0.22.7
0.22.6
0.22.5
0.22.4
0.22.3
0.22.2
0.22.1
0.22.0
0.21.7
0.21.6
0.21.5
0.21.4
0.21.2
0.21.1
0.21.0
0.20.25
0.20.24
0.20.23
0.20.22
0.20.21
0.20.20
0.20.19
0.20.18
0.20.17
0.20.16
0.20.15
0.20.14
0.20.13
0.20.12
0.20.11
0.20.10
0.20.9
0.20.8
0.20.7
0.20.6
0.20.5
0.20.3
0.20.2
0.20.1
0.20.0
0.19.2
0.19.1
0.19.0
0.18.1
0.18.0
0.17.4
0.17.3
0.17.2
0.17.1
0.17.0
0.16.6
0.16.5
0.16.4
0.16.3
0.16.2
0.16.1
0.16.0
0.15.1
0.15.0
0.14.3
0.14.2
0.14.1
0.14.0
0.12.6
0.12.5
0.12.3
0.11.3
0.10.2
0.10.1
0.10.0
0.9.2
0.9.1
0.9.0
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.10
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.30
0.7.29
0.7.28
0.7.27
0.7.26
0.7.25
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.0
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.4
0.3.3
0.3.2
0.3.1
Macula HTTP/3 Mesh SDK — connect, subscribe, publish, call, advertise
Current section
Files
Jump to
Current section
Files
src/macula_routing_system/macula_routing_table.erl
%%%-------------------------------------------------------------------
%%% @doc
%%% Routing table for Kademlia DHT.
%%% Manages 256 k-buckets organized by XOR distance.
%%% @end
%%%-------------------------------------------------------------------
-module(macula_routing_table).
%% API
-export([
new/2,
add_node/2,
remove_node/2,
remove_by_endpoint/2,
evict_stale/2,
find_closest/3,
get_bucket/2,
bucket_size/2,
get_all_nodes/1,
update_timestamp/2,
size/1,
local_node_id/1,
k/1
]).
%% Types
-type routing_table() :: #{
local_node_id := binary(),
k := pos_integer(),
buckets := #{0..255 => macula_routing_bucket:bucket()}
}.
-export_type([routing_table/0]).
%%%===================================================================
%%% API Functions
%%%===================================================================
%% @doc Create a new routing table.
-spec new(binary(), pos_integer()) -> routing_table().
new(LocalNodeId, K) ->
#{
local_node_id => LocalNodeId,
k => K,
buckets => #{}
}.
%% @doc Add a node to the routing table.
%% Removes any ghost entries (same endpoint, different node_id) across ALL buckets
%% before adding, since a restarted node may land in a different bucket.
-spec add_node(routing_table(), macula_routing_bucket:node_info()) -> routing_table().
add_node(#{local_node_id := LocalNodeId} = Table, NodeInfo) ->
NodeId = maps:get(node_id, NodeInfo),
Endpoint = macula_routing_bucket:get_endpoint(NodeInfo),
%% Remove ghosts from all buckets (endpoint with different node_id)
CleanTable = remove_ghosts_by_endpoint(Table, Endpoint, NodeId),
BucketIndex = macula_routing_nodeid:bucket_index(LocalNodeId, NodeId),
do_add_node(CleanTable, BucketIndex, NodeInfo).
%% Self node (bucket index 256) - ignore
do_add_node(Table, 256, _NodeInfo) ->
Table;
%% Add to appropriate bucket
do_add_node(#{k := K, buckets := Buckets} = Table, BucketIndex, NodeInfo) ->
Bucket = maps:get(BucketIndex, Buckets, macula_routing_bucket:new(K)),
add_to_bucket(Table, Buckets, BucketIndex, Bucket, NodeInfo).
%% Add node to bucket. If bucket is full, the node is ignored.
%% Standard Kademlia behavior - full buckets indicate well-known nodes.
add_to_bucket(Table, _Buckets, _BucketIndex, Bucket, NodeInfo) ->
case macula_routing_bucket:add_node(Bucket, NodeInfo) of
{error, bucket_full} -> Table;
UpdatedBucket -> update_bucket(Table, _BucketIndex, UpdatedBucket)
end.
update_bucket(#{buckets := Buckets} = Table, BucketIndex, UpdatedBucket) ->
Table#{buckets => Buckets#{BucketIndex => UpdatedBucket}}.
%% @doc Remove a node from the routing table.
-spec remove_node(routing_table(), binary()) -> routing_table().
remove_node(#{local_node_id := LocalNodeId} = Table, NodeId) ->
BucketIndex = macula_routing_nodeid:bucket_index(LocalNodeId, NodeId),
do_remove_node(Table, BucketIndex, NodeId).
do_remove_node(#{buckets := Buckets} = Table, BucketIndex, NodeId) ->
case maps:find(BucketIndex, Buckets) of
{ok, Bucket} ->
UpdatedBucket = macula_routing_bucket:remove_node(Bucket, NodeId),
update_bucket(Table, BucketIndex, UpdatedBucket);
error ->
Table
end.
%% @doc Remove all nodes matching an endpoint from all buckets.
-spec remove_by_endpoint(routing_table(), binary()) -> routing_table().
remove_by_endpoint(#{buckets := Buckets} = Table, Endpoint) ->
NewBuckets = maps:map(fun(_Idx, Bucket) ->
macula_routing_bucket:remove_by_endpoint(Bucket, Endpoint)
end, Buckets),
Table#{buckets => NewBuckets}.
%% @doc Remove nodes not seen since StaleThreshold (millisecond timestamp) from all buckets.
-spec evict_stale(routing_table(), integer()) -> routing_table().
evict_stale(#{buckets := Buckets} = Table, StaleThreshold) ->
NewBuckets = maps:map(fun(_Idx, Bucket) ->
macula_routing_bucket:evict_stale(Bucket, StaleThreshold)
end, Buckets),
Table#{buckets => NewBuckets}.
%% @doc Find k closest nodes to target.
-spec find_closest(routing_table(), binary(), pos_integer()) -> [macula_routing_bucket:node_info()].
find_closest(#{local_node_id := LocalNodeId, buckets := Buckets}, Target, K) ->
TargetBucketIndex = macula_routing_nodeid:bucket_index(LocalNodeId, Target),
AllNodes = collect_nodes_near_bucket(Buckets, TargetBucketIndex),
sort_by_distance_and_take(AllNodes, Target, K).
sort_by_distance_and_take(Nodes, Target, K) ->
WithDistance = [{distance_to(Target, N), N} || N <- Nodes],
Sorted = lists:keysort(1, WithDistance),
[Node || {_Dist, Node} <- lists:sublist(Sorted, K)].
distance_to(Target, #{node_id := NodeId}) ->
macula_routing_nodeid:distance(Target, NodeId).
%% @doc Get bucket by index.
-spec get_bucket(routing_table(), 0..255) -> macula_routing_bucket:bucket().
get_bucket(#{k := K, buckets := Buckets}, BucketIndex) ->
maps:get(BucketIndex, Buckets, macula_routing_bucket:new(K)).
%% @doc Get size of a specific bucket.
-spec bucket_size(routing_table(), 0..255) -> non_neg_integer().
bucket_size(Table, BucketIndex) ->
Bucket = get_bucket(Table, BucketIndex),
macula_routing_bucket:size(Bucket).
%% @doc Get all nodes from all buckets.
-spec get_all_nodes(routing_table()) -> [macula_routing_bucket:node_info()].
get_all_nodes(#{buckets := Buckets}) ->
lists:flatten([
macula_routing_bucket:get_nodes(Bucket)
|| Bucket <- maps:values(Buckets)
]).
%% @doc Update timestamp for a node (moves to tail in its bucket).
-spec update_timestamp(routing_table(), binary()) -> routing_table().
update_timestamp(#{local_node_id := LocalNodeId} = Table, NodeId) ->
BucketIndex = macula_routing_nodeid:bucket_index(LocalNodeId, NodeId),
do_update_timestamp(Table, BucketIndex, NodeId).
do_update_timestamp(#{buckets := Buckets} = Table, BucketIndex, NodeId) ->
case maps:find(BucketIndex, Buckets) of
{ok, Bucket} ->
UpdatedBucket = macula_routing_bucket:update_timestamp(Bucket, NodeId),
update_bucket(Table, BucketIndex, UpdatedBucket);
error ->
Table
end.
%% @doc Get total number of nodes in routing table.
-spec size(routing_table()) -> non_neg_integer().
size(#{buckets := Buckets}) ->
lists:sum([
macula_routing_bucket:size(Bucket)
|| Bucket <- maps:values(Buckets)
]).
%% @doc Get local node ID.
-spec local_node_id(routing_table()) -> binary().
local_node_id(#{local_node_id := NodeId}) ->
NodeId.
%% @doc Get k (bucket capacity).
-spec k(routing_table()) -> pos_integer().
k(#{k := K}) ->
K.
%%%===================================================================
%%% Internal Functions
%%%===================================================================
%% @doc Remove ghost entries: nodes with matching endpoint but different node_id.
%% Scans all buckets since a restarted node may generate a node_id in a different bucket.
-spec remove_ghosts_by_endpoint(routing_table(), binary() | undefined, binary()) -> routing_table().
remove_ghosts_by_endpoint(Table, undefined, _NodeId) ->
Table;
remove_ghosts_by_endpoint(#{buckets := Buckets} = Table, Endpoint, NodeId) ->
NewBuckets = maps:map(fun(_Idx, Bucket) ->
macula_routing_bucket:remove_ghost_by_endpoint(Bucket, Endpoint, NodeId)
end, Buckets),
Table#{buckets => NewBuckets}.
%% @doc Collect nodes from buckets near target bucket (expanding outward).
-spec collect_nodes_near_bucket(#{0..255 => macula_routing_bucket:bucket()}, 0..256) -> [macula_routing_bucket:node_info()].
collect_nodes_near_bucket(Buckets, 256) ->
%% Target is local node - collect from all buckets
lists:flatmap(fun macula_routing_bucket:get_nodes/1, maps:values(Buckets));
collect_nodes_near_bucket(Buckets, StartIndex) ->
Indices = expand_indices(StartIndex, 0, 255),
lists:flatmap(fun(Index) -> get_bucket_nodes(Buckets, Index) end, Indices).
get_bucket_nodes(Buckets, Index) ->
case maps:find(Index, Buckets) of
{ok, Bucket} -> macula_routing_bucket:get_nodes(Bucket);
error -> []
end.
%% @doc Generate list of bucket indices expanding outward from start.
-spec expand_indices(non_neg_integer(), non_neg_integer(), non_neg_integer()) -> [non_neg_integer()].
expand_indices(Start, Min, Max) ->
expand_from_center(Start, 1, Min, Max, [Start]).
expand_from_center(_Start, Offset, Min, Max, Acc) when Offset > Max - Min ->
lists:reverse(Acc);
expand_from_center(Start, Offset, Min, Max, Acc) ->
Acc2 = maybe_add_index(Start + Offset, Max, Acc),
Acc3 = maybe_add_index(Start - Offset, Min, Acc2),
expand_from_center(Start, Offset + 1, Min, Max, Acc3).
maybe_add_index(Index, Max, Acc) when Index =< Max, Index >= 0 ->
[Index | Acc];
maybe_add_index(_Index, _Max, Acc) ->
Acc.