%%%------------------------------------------------------------------- %%% @doc Macula SDK — Public API for mesh applications. %%% %%% This is the main entry point for applications using the Macula SDK. %%% %%% Apps connect via `connect/2', which returns a `macula_client' %%% pool that internally wraps N peering links to N stations. %%% `publish/4,5', `subscribe/4,5', `unsubscribe/2', `call/5', %%% `advertise/5', `unadvertise/3', `call_stream/5', %%% `advertise_stream/5', and `unadvertise_stream/3' route through %%% the pool with realm-per-call semantics. See `macula_pubsub' for %%% the slice module of the publish/subscribe surface. %%% %%% LOCAL streaming (`call_stream/2,3', `open_stream/3,4', %%% `advertise_stream/2,3', `unadvertise_stream/1') dispatches %%% in-process via `macula_stream_local' — for unit tests and %%% same-BEAM pairs. %%% %%% Erlang distribution over the mesh ships via `join_mesh/1' (V2 %%% pool carrier) or `join_dist_relay/1' (dedicated dist relay). See %%% `macula_dist_pool' / `macula_dist_relay_client'. %%% %%% @end %%%------------------------------------------------------------------- -module(macula). -include_lib("kernel/include/logger.hrl"). %% Connection -export([connect/2, close/1, child_spec/3, status/1, links/1]). %% Reading fields of peer-supplied maps (D26) -export([field/2, field/3, text/1]). %% Pub/Sub — realm-per-call against a V2 pool -export([subscribe/4, subscribe/5, subscribe_callback/4, unsubscribe/2, publish/4, publish/5]). %% RPC — realm-per-call against a V2 pool -export([call/5, call_station/7, call_station/8, advertise/5, unadvertise/3]). %% Signed DHT records — realm-agnostic infrastructure procedures %% (`_dht.put_record', `_dht.find_record', `_dht.find_records_by_type', %% `_dht.records..stored'). The all-zeros realm is the SDK %% convention for protocol-internal traffic. -export([put_record/2, find_record/2, find_record/3, find_records/2, find_records/3, find_records_by_type/2, subscribe_records/3, unsubscribe_records/2]). -export([sign_node_record/2, sign_node_record/3, sign_domain_record/2, withdraw_node_record/3, parse_stations/1]). %% Content-addressed blob storage. `_content.put_block' / %% `_content.get_block' RPCs against the relay's local content %% store. MCID is a 50-byte binary: the hash tag (2, SHA-384), the %% codec (16#55 raw, 16#56 manifest), then the 48-byte SHA-384 hash. The %% relay validates the payload's hash on `put_block' and rejects mismatches. -export([put_content/2, put_content_station/4, put_content_station/5, get_content/2, get_content_station/4, get_content_station/5, find_content_providers/2]). %% Streaming RPC (LOCAL in-process + V2 pool, see PLAN_MACULA_STREAMING.md) -export([ call_stream/2, call_stream/3, call_stream/5, call_stream_station/7, open_stream/3, open_stream/4, advertise_stream/2, advertise_stream/3, advertise_stream/5, advertise_stream/6, unadvertise_stream/1, unadvertise_stream/3, send/2, send/3, recv/1, recv/2, close_stream/1, close_send/1, await_reply/1, await_reply/2, set_reply/2, abort/3 ]). %% Cluster (LAN) -export([ensure_distributed/0, monitor_nodes/0, unmonitor_nodes/0]). %% Mesh Distribution -export([join_mesh/1, join_dist_relay/1, dist_relay_client/0]). -ifdef(TEST). -export([join_pool_args/1]). %% Exports for unit tests — pure helpers that are otherwise private. %% `verify_block_hash/2' moved to `macula_content_transfer' (Phase 1, %% PLAN_PUSH_UPLOAD.md) along with the rest of the content-stream %% transfer internals — see `macula_content_transfer:verify_block_hash/2'. -export([decode_provider/2]). -endif. %% Types -export_type([pool/0, realm/0, topic/0, procedure/0, stream/0, stream_mode/0, stream_handler/0, m_record/0, record_type/0, record_key/0, mcid/0]). -type pool() :: macula_client:pool(). -type realm() :: <<_:256>>. %% 32-byte realm tag. -type topic() :: binary(). -type procedure() :: binary(). -type stream() :: pid(). -type stream_mode() :: server_stream | client_stream | bidi. -type stream_handler() :: fun((stream(), term()) -> any()). -type m_record() :: macula_record:m_record(). -type record_type() :: macula_record:type_tag(). -type record_key() :: <<_:256>>. %% DHT storage key — `macula_record:storage_key/1' output. %%%=================================================================== %%% Connection — V2 (pool, since 3.11.0) %%%=================================================================== %% @doc Connect to the Macula relay mesh and return a pool handle. %% %% `Seeds' is a list of station endpoints: `#{host, port, expected_node_id}' %% maps, or URL binaries or strings. Every seed names the node_id it %% expects, in the seed or in the `expected_node_id' option; otherwise no %% pool starts and `{error, {seeds, expected_node_id_required}}' is %% returned. The pool spawns one peering link per seed and routes ops with %% replication, replay, and event dedup. Returns immediately; link %% handshakes complete asynchronously. %% %% Honored opts (full reference: `macula_client:opts()'): %% %% %% Legacy opts silently dropped (with a one-shot `logger:notice'): %% `relays' (use the `Seeds' positional argument), `realm' (V2 is %% realm-per-call), `site' (no V2 analog), `connections' (one link %% per seed; add more seeds to grow the pool). %% %% See `macula_client' for the canonical pool implementation and %% `macula_pubsub' for the slice module. -spec connect([macula_client:seed()], macula_client:opts()) -> {ok, pool()} | {error, term()}. connect(Seeds, Opts) when is_list(Seeds), is_map(Opts) -> macula_client:connect(Seeds, Opts). %% @doc Stop a V2 pool. Every subscriber receives a final %% `{macula_event_gone, SubRef, pool_closed}' message. -spec close(pool()) -> ok. close(Pool) when is_pid(Pool) -> macula_client:close(Pool). %% @doc OTP child spec to drop a V2 pool into a caller's supervision %% tree. Give `node_identity' as a loader `{Module, Function, Args}' that %% returns `{ok, Key}'. Its `Args' say where the key is and never hold the %% key, because a supervisor logs them when a start fails. See %% `macula_client:child_spec/3'. -spec child_spec(term(), [macula_client:seed()], macula_client:opts()) -> supervisor:child_spec(). child_spec(Id, Seeds, Opts) -> macula_client:child_spec(Id, Seeds, Opts). %% @doc Aggregate health snapshot of a V2 pool. Suitable for %% `/health' or `/status' endpoints; not for hot-loop polling. See %% `macula_client:status/1' for the full shape. -spec status(pool()) -> {ok, macula_client:status()}. status(Pool) when is_pid(Pool) -> macula_client:status(Pool). %% @doc Per-link snapshot of a V2 pool — one entry per spawned link %% with its peer station `node_id' (pubkey), dial `host', `pid', and %% `connected' flag. Use this to resolve a specific station (by pubkey %% or hostname) to its link for targeted, per-station operations. See %% `macula_client:links/1' for the `link_info()' shape. -spec links(pool()) -> {ok, [macula_client:link_info()]}. links(Pool) when is_pid(Pool) -> macula_client:links(Pool). %%%=================================================================== %%% Pub/Sub — realm-per-call against a V2 pool %%%=================================================================== %% @doc Publish to `(Realm, Topic)' on `Pool'. Equivalent to %% `publish/5' with empty opts. -spec publish(pool(), realm(), topic(), term()) -> ok | {error, term()}. publish(Pool, Realm, Topic, Payload) -> macula_pubsub:publish(Pool, Realm, Topic, Payload). %% @doc Publish to `(Realm, Topic)' on `Pool' with options. See %% `macula_pubsub:publish/5' for honored opts. -spec publish(pool(), realm(), topic(), term(), map()) -> ok | {error, term()}. publish(Pool, Realm, Topic, Payload, Opts) -> macula_pubsub:publish(Pool, Realm, Topic, Payload, Opts). %% @doc Subscribe `Subscriber' to `(Realm, Topic)' on `Pool'. %% Equivalent to `subscribe/5' with empty opts. -spec subscribe(pool(), realm(), topic(), pid()) -> {ok, reference()} | {error, {text_too_long | invalid_text, topic}}. subscribe(Pool, Realm, Topic, Subscriber) -> macula_pubsub:subscribe(Pool, Realm, Topic, Subscriber). %% @doc Subscribe `Subscriber' to `(Realm, Topic)' on `Pool' with %% options. The `delivery' option chooses how a single publisher's %% out-of-order arrivals are handled: %% %% Ordering state is kept per publisher, and apart for EVENTs whose %% publisher signature did not verify. %% See `macula_pubsub:subscribe/5'. -spec subscribe(pool(), realm(), topic(), pid(), map()) -> {ok, reference()} | {error, {text_too_long | invalid_text, topic}}. subscribe(Pool, Realm, Topic, Subscriber, Opts) -> macula_pubsub:subscribe(Pool, Realm, Topic, Subscriber, Opts). %% @doc Subscribe with a callback function. The SDK spawns a small %% receiver process internally and invokes the callback once per %% inbound event. See `macula_pubsub:subscribe_callback/4'. -spec subscribe_callback(pool(), realm(), topic(), macula_pubsub:callback()) -> {ok, reference()} | {error, term()}. subscribe_callback(Pool, Realm, Topic, Callback) -> macula_pubsub:subscribe_callback(Pool, Realm, Topic, Callback). %% @doc Drop a pool subscription. Idempotent. -spec unsubscribe(pool(), reference()) -> ok. unsubscribe(Pool, SubRef) when is_pid(Pool), is_reference(SubRef) -> macula_pubsub:unsubscribe(Pool, SubRef). %%%=================================================================== %%% RPC — realm-per-call against a V2 pool %%%=================================================================== %% @doc Call `Procedure' in `Realm' at the provider that serves it: resolve %% the procedure's verified advertisements, authorized against the realm key %% the pool pinned for `Realm', reach the station a candidate names %% directly, and call its provider there. See %% `macula_direct_dial:call/5'. A procedure the pool's linked stations serve %% themselves, such as `_dht.*', goes through %% `macula_client:call_linked_station/5'. -spec call(pool(), realm(), procedure(), term(), 1..600_000) -> {ok, term()} | {error, term()}. call(Pool, Realm, Procedure, Payload, TimeoutMs) -> macula_direct_dial:call(Pool, Realm, Procedure, Payload, TimeoutMs). %% @doc Issue a CALL to `Target', a provider's node_id, at ONE specific %% station, dialing it directly even if it is not in the pool's seed set. %% `Station' is a seed URL (e.g. `<<"quic://[::1]:4433">>'). The pool reuses %% an existing link or dials and monitors a new one, waits for the %% handshake, and calls there. This is the direct-dial data path: resolve a %% provider's serving_station and its endpoint, then reach it in one hop. %% See `macula_client:call_station/7'. -spec call_station(pool(), macula_client:seed(), <<_:256>>, realm(), procedure(), term(), 1..600_000) -> {ok, term()} | {error, term()}. call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs) -> macula_client:call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs). %% @doc As `call_station/7', presenting a capability token to a gated %% provider via `Opts' (`#{ucan_token => Token}'). Empty/absent = none. %% Slice 7b dual-trust. `Opts' also carries the per-call TLS trust %% override for this dial: `verify', `expected_node_id', and %% `pin_tls_cert' (see `macula_client:call_station/9'), and may set %% `dial_timeout_ms', how much of `TimeoutMs' the wait for a fresh link's %% handshake may take (default: all of it). -spec call_station(pool(), macula_client:seed(), <<_:256>>, realm(), procedure(), term(), 1..600_000, map()) -> {ok, term()} | {error, term()}. call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs, Opts) -> Ucan = maps:get(ucan_token, Opts, <<>>), LinkOpts = maps:with([verify, expected_node_id, pin_tls_cert], Opts), DialTimeoutMs = maps:get(dial_timeout_ms, Opts, TimeoutMs), macula_client:call_station(Pool, Station, Target, Realm, Procedure, Payload, TimeoutMs, Ucan, LinkOpts, DialTimeoutMs). %% @doc Register a procedure handler on a V2 pool and advertise it: %% the pool resolves its own D25 provider authorization — the %% realm-signed `org_directory' and the org-signed %% `procedure_delegation' that names the pool's node id, both fetched %% from the DHT — signs the advertisement, and sends it as an %% ADVERTISE frame to every link (replayed on link respawn). The %% procedure must carry an org namespace, and the pool must run a %% provisioned identity whose delegation the org has published, and %% pin the realm's key (`realm_trust' at connect); a missing piece %% fails fast with `{error, {provider_authorization, _}}'. %% %% `Opts' `auth' sets the procedure's policy: `open' (default, serve %% any identified caller), `{ucan_required, Issuer}' (gated to one %% known identity, Slice 7b), or `{realm_member_required, RealmDid, %% RequiredCan}' (gated to realm membership at a specific tier) -- see %% `macula_client:auth_policy()' for the full set. -spec advertise(pool(), realm(), procedure(), macula_client:handler(), map()) -> ok | {error, term()}. advertise(Pool, Realm, Procedure, Handler, Opts) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32 -> Policy = maps:get(auth, Opts, open), advertise_authorized(Pool, Realm, Procedure, fun(EncodedAd) -> macula_client:advertise(Pool, Realm, Procedure, Handler, Policy, EncodedAd) end). %% @doc Stop advertising a procedure on a V2 pool. -spec unadvertise(pool(), realm(), procedure()) -> ok. unadvertise(Pool, Realm, Procedure) -> macula_client:unadvertise(Pool, Realm, Procedure). %%%=================================================================== %%% Signed DHT records (v3.3.0) %%%=================================================================== %%% %%% Records are typed payloads signed in the signed-object format of %%% DESIGN_PQ_SIGNED_FRAMES_AND_RECORDS.md and stored in the relay %%% mesh's distributed hash table. A record travels as its wire form: %%% stations store and forward its bytes as received. A record reaches %%% a caller only after it verifies under the node's crypto profile %%% (`macula_record:verify/2'): `find_record/2' returns the refusal of %%% one that does not, and the list functions and `subscribe_records/3' %%% drop it. %%% %%% See `macula_record' for the record shape, the typed %%% constructors (`node_record/3', `realm_directory/3', %%% `realm_stations/2', `procedure_advertisement/4', %%% `content_announcement/3', `tombstone/2', and the foundation_* %%% constructors), and `storage_key/1' for the DHT addressing rule. %%% %%% Two complementary retrieval paths: %%% %%% - `find_record/2' — fetch one record by its %%% `storage_key/1' output %%% - `find_records_by_type/2' — list every record of a given %%% type tag %%% %%% Plus a live-update channel: %%% %%% - `subscribe_records/3' — receive new records of a type %%% as they are stored %% Procedure + topic shape — hidden from API consumers but exposed %% as documentation. The relay backend (hecate-station and successors) %% MUST advertise these procedures and publish on the per-type %% record-stored topic for the SDK to function. DHT traffic travels %% under the all-zeros realm tag (protocol-internal infrastructure; %% the same convention `macula_dist_pool' uses for tunnel frames). -define(DHT_REALM, <<0:256>>). -define(DHT_PUT_RECORD_PROC, <<"_dht.put_record">>). -define(DHT_FIND_RECORD_PROC, <<"_dht.find_record">>). -define(DHT_FIND_RECORDS_PROC, <<"_dht.find_records">>). -define(DHT_FIND_RECORDS_BY_TYPE_PROC, <<"_dht.find_records_by_type">>). -define(DHT_RECORD_TIMEOUT_MS, 5_000). -define(TYPE_CONTENT_ANNOUNCEMENT, 16#11). %% `_content.*' procedure names, realm, and timeouts moved to %% `macula_content_transfer' (Phase 1, PLAN_PUSH_UPLOAD.md) along with %% the transfer logic that used them. %% @doc Store a signed record in the mesh DHT via a V2 pool. %% %% Build the record via the typed constructors in `macula_record' %% (`node_record/3,4', `content_announcement/3,4', `tombstone/2,3', %% `realm_directory/3,4', `procedure_advertisement/4,5', etc.) and sign %% it. A record this node signs about itself is signed by its pool, %% which holds the node identity key: through %% `macula_client:sign_node_record/2', a domain record (tags 0x20 to %% 0xFF) through `macula_client:sign_domain_record/2', and the %% tombstone of either through `macula_client:withdraw_node_record/3'. %% A realm-, org- or %% foundation-signed record is signed with `macula_record:sign/2' and %% its signer's key. Pass the signed record or its wire form %% (`macula_record:encode/1'). The record travels as its %% wire form; the station verifies it on receipt and stores it under %% `macula_record:storage_key/1', propagating to the K-nearest peers. -spec put_record(pool(), m_record() | binary()) -> ok | {error, term()}. put_record(Pool, #{key := _, tbs := _, signature := _} = Signed) when is_pid(Pool) -> put_record(Pool, macula_record:encode(Signed)); put_record(Pool, Wire) when is_pid(Pool), is_binary(Wire) -> %% Record bytes paced per pool, so a bulk writer stays under the %% station's STORE allowance (D28, 3.5) instead of running into stored 0. %% One bucket per pool is conservative for a pool of several links: its %% calls fan out one link at a time, and the per-link pacer in %% macula_station_link:put_record/3 paces a caller that writes straight %% through a link. ok = macula_store_pacer:await(Pool, byte_size(Wire)), classify_put(macula_client:call_linked_station(Pool, ?DHT_REALM, ?DHT_PUT_RECORD_PROC, Wire, ?DHT_RECORD_TIMEOUT_MS)). classify_put({ok, ok}) -> ok; %% The reply's atom crossed the wire as text and decoded back to the %% codec's {text, _} marker (D26 makes no atom on the wire). classify_put({ok, {text, <<"ok">>}}) -> ok; classify_put({ok, Reply}) -> {error, {unexpected_reply, Reply}}; classify_put({error, _} = E) -> E. %% @doc Fetch a record from the mesh DHT by its %% `macula_record:storage_key/1'. %% %% Returns `{error, not_found}' when no record exists at the key. The %% record is verified under the node's crypto profile before it is %% returned; one that does not verify returns its refusal from %% `macula_record:verify/2', such as `{error, expired}'. -spec find_record(pool(), record_key()) -> {ok, m_record()} | {error, not_found | term()}. find_record(Pool, Key) -> find_record(Pool, Key, ?DHT_RECORD_TIMEOUT_MS). %% @doc As `find_record/2', waiting at most `TimeoutMs' for the reply, for a %% caller that bounds its work by a deadline of its own. -spec find_record(pool(), record_key(), pos_integer()) -> {ok, m_record()} | {error, not_found | term()}. find_record(Pool, Key, TimeoutMs) when is_pid(Pool), is_binary(Key), byte_size(Key) =:= 32, is_integer(TimeoutMs), TimeoutMs > 0 -> classify_find(macula_client:call_linked_station(Pool, ?DHT_REALM, ?DHT_FIND_RECORD_PROC, #{key => Key}, TimeoutMs)). classify_find({ok, Wire}) when is_binary(Wire) -> with_profile(fun(Profile) -> macula_record:verify(Wire, Profile) end); classify_find({ok, not_found}) -> {error, not_found}; %% The reply's atom crossed the wire as text and decoded back to the %% codec's {text, _} marker (D26 makes no atom on the wire). classify_find({ok, {text, <<"not_found">>}}) -> {error, not_found}; classify_find({ok, Reply}) -> {error, {unexpected_reply, Reply}}; classify_find({error, _} = E) -> E. %% @doc Fetch EVERY record stored at `Key' — the full multi-value %% set, e.g. every `procedure_advertisement' under one procedure's %% storage key. Where `find_record/2' returns the first record (or %% `not_found'), this returns the whole list, empty when none. %% %% The relay's local store is a signer-deduped multiset: one record %% per signing key at a storage key, so N providers of one procedure %% return N records. Each record is verified under the node's crypto %% profile, and a record that does not verify is dropped. -spec find_records(pool(), record_key()) -> {ok, [m_record()]} | {error, term()}. find_records(Pool, Key) -> find_records(Pool, Key, ?DHT_RECORD_TIMEOUT_MS). %% @doc As `find_records/2', waiting at most `TimeoutMs' for the reply, for a %% caller that bounds its work by a deadline of its own. -spec find_records(pool(), record_key(), pos_integer()) -> {ok, [m_record()]} | {error, term()}. find_records(Pool, Key, TimeoutMs) when is_pid(Pool), is_binary(Key), byte_size(Key) =:= 32, is_integer(TimeoutMs), TimeoutMs > 0 -> classify_find_list(macula_client:call_linked_station(Pool, ?DHT_REALM, ?DHT_FIND_RECORDS_PROC, #{key => Key}, TimeoutMs)). classify_find_list({ok, Wires}) when is_list(Wires) -> verified_records(Wires); classify_find_list({ok, Reply}) -> {error, {unexpected_reply, Reply}}; classify_find_list({error, _} = E) -> E. %% @doc Return every record of a given type currently visible from %% the pool's connected stations, each verified under the node's %% crypto profile; a record that does not verify is dropped. %% %% Coverage depends on each station's view of the DHT — a single %% station sees its local replicas plus whatever its peers have %% gossiped. Aggregating across the full mesh requires querying %% multiple stations and deduplicating by record key. -spec find_records_by_type(pool(), record_type()) -> {ok, [m_record()]} | {error, term()}. find_records_by_type(Pool, Type) when is_pid(Pool), is_integer(Type), Type >= 0, Type =< 255 -> classify_list(macula_client:call_linked_station(Pool, ?DHT_REALM, ?DHT_FIND_RECORDS_BY_TYPE_PROC, #{type => Type}, ?DHT_RECORD_TIMEOUT_MS)). classify_list({ok, Wires}) when is_list(Wires) -> verified_records(Wires); classify_list({ok, Reply}) -> {error, {unexpected_reply, Reply}}; classify_list({error, _} = E) -> E. %% The records among Wires that verify under the node's crypto profile; %% the rest are dropped. verified_records(Wires) -> with_profile(fun(Profile) -> {ok, [Record || Wire <- Wires, {ok, Record} <- [macula_record:verify(Wire, Profile)]]} end). %% Verify with the node's crypto profile, or return the refusal of a %% node that has none. with_profile(Verify) -> profile_verified(macula_crypto_profile:configured(), Verify). profile_verified({ok, Profile}, Verify) -> Verify(Profile); profile_verified({error, _} = Refusal, _Verify) -> Refusal. %% @doc Subscribe to live record-stored events filtered by type. %% %% The callback receives each newly-stored record of the given type %% that verifies under the node's crypto profile. Returns a %% subscription reference for `unsubscribe_records/2'. %% Topic shape is `_dht.records..stored', rendered with the %% type tag as a decimal integer for log friendliness. -spec subscribe_records(pool(), record_type(), fun((m_record()) -> any())) -> {ok, reference()} | {error, term()}. subscribe_records(Pool, Type, Callback) when is_pid(Pool), is_integer(Type), Type >= 0, Type =< 255, is_function(Callback, 1) -> Topic = record_stored_topic(Type), macula_pubsub:subscribe_callback(Pool, ?DHT_REALM, Topic, wrap_record_callback(Callback)). %% @doc Cancel a `subscribe_records/3' subscription. -spec unsubscribe_records(pool(), reference()) -> ok. unsubscribe_records(Pool, Ref) when is_pid(Pool), is_reference(Ref) -> macula_pubsub:unsubscribe(Pool, Ref). %%%=================================================================== %%% Signing records this node signs about itself — the pool holds the key %%%=================================================================== %% @doc Sign a record this node signs about itself with the pool's node %% identity key, in the pool's own process: a node record, a procedure %% advertisement or a content announcement that names this node. See %% `macula_client:sign_node_record/2' for the record checks and %% refusals. `Opts' `not_after' bounds the signed record's expiry: the %% pool judges it on its own clock, refuses a bound already passed as %% `{error, not_after_passed}', ends the record at the bound when the %% bound comes before the record's lifetime runs out, and keeps the %% built lifetime otherwise. -spec sign_node_record(pool(), m_record()) -> {ok, m_record()} | {error, term()}. sign_node_record(Pool, Record) when is_pid(Pool) -> macula_client:sign_node_record(Pool, Record). -spec sign_node_record(pool(), m_record(), #{not_after := integer()}) -> {ok, m_record()} | {error, term()}. sign_node_record(Pool, Record, Opts) when is_pid(Pool), is_map(Opts) -> macula_client:sign_node_record(Pool, Record, Opts). %% @doc Sign a domain record (tags 0x20 to 0xFF) as this node, with the %% pool's node identity key, in the pool's own process. See %% `macula_client:sign_domain_record/2'. -spec sign_domain_record(pool(), m_record()) -> {ok, m_record()} | {error, term()}. sign_domain_record(Pool, Record) when is_pid(Pool) -> macula_client:sign_domain_record(Pool, Record). %% @doc Withdraw a record this node signed — a node record, a procedure %% advertisement, a content announcement or a domain record — with a %% tombstone signed by the pool's node identity key, in the pool's own %% process. See `macula_client:withdraw_node_record/3'. -spec withdraw_node_record(pool(), m_record() | binary(), macula_record:reason()) -> {ok, m_record()} | {error, term()}. withdraw_node_record(Pool, Withdrawn, Reason) when is_pid(Pool) -> macula_client:withdraw_node_record(Pool, Withdrawn, Reason). %% @doc Parse a `MACULA_STATIONS' seed list: comma-separated %% `@:' entries, the node id as 64 lowercase hex %% characters, the host a DNS name, an IPv4 address or a bracketed IPv6 %% address, the port 1 to 65535. Returns the seeds in order, each pinned %% to its node id. A refusal names the entry's position and what is %% wrong with it, and carries neither the value, nor a node id, nor a %% host. See `macula_stations:parse/1'. -spec parse_stations(binary()) -> {ok, [macula_stations:seed()]} | {error, term()}. parse_stations(Value) when is_binary(Value) -> macula_stations:parse(Value). record_stored_topic(Type) -> iolist_to_binary([<<"_dht.records.">>, integer_to_binary(Type), <<".stored">>]). %% Adapt a 1-arg `(Record) -> any()' user callback to the 3-arg %% `(Topic, Payload, Meta) -> any()' shape `macula_pubsub' delivers. %% %% PubSub delivers the payload as the record's wire form (the %% substrate's `record_fanout' publishes `macula_record:encode/1' %% output on the `_dht.records..stored' topic). The record is %% verified here under the node's crypto profile, so the callback %% receives only verified records. A payload that does not verify is %% dropped silently: surfacing it would force every user to handle %% refusals on a protocol-internal channel. wrap_record_callback(Fun) -> fun(_Topic, Payload, _Meta) -> apply_callback_with_verified(Fun, Payload) end. apply_callback_with_verified(Fun, Payload) -> callback_with(with_profile(fun(Profile) -> macula_record:verify(Payload, Profile) end), Fun). callback_with({ok, Record}, Fun) -> Fun(Record), ok; callback_with({error, _}, _Fun) -> ok. %%%=================================================================== %%% Content-addressed blob storage (v4.2.7+) %%%=================================================================== -type mcid() :: <<_:400>>. %% @doc Store `Bytes' in the mesh's content store and return its MCID %% (Macula Content ID, 50 bytes: tag 2 for SHA-384, codec, then the 48-byte hash). %% Content that fits in one block (`byte_size(Bytes) =< %% macula_manifest:default_chunk_size/0', 256 KiB) is sent as a %% single `_content.put_block', and the MCID is `<<2, 16#55, %% SHA-384(Bytes)>>'. Larger content is split %% into chunks (`macula_manifest:create/1'), each chunk sent %% via its own `_content.put_block', then a `content_manifest' via %% `_content.put_manifest'; the returned MCID is the manifest's %% (`<<2, 16#56, _/binary>>'), Merkle-rooted over every chunk. Either %% way the station verifies each block's hash before accepting it. %% %% The whole transfer — every block call plus the manifest call for %% chunked content — rides one dedicated QUIC stream on one pinned %% pool link (see PLAN_PER_STREAM_QUIC_ISOLATION.md Phase 2), so a %% large blob transfer no longer head-of-line-blocks other RPC/PubSub %% traffic on the same connection. %% %% A thin blocking wrapper over `macula_content_transfer:start_put/2' + %% `await/1' — see that module for the addressable form (a live pid, %% real cancel with a peer-visible abort, pause/resume/multi-stream as %% later phases land). PLAN_PUSH_UPLOAD.md Phase 1. -spec put_content(pool(), binary()) -> {ok, mcid()} | {error, term()}. put_content(Pool, Bytes) when is_pid(Pool), is_binary(Bytes) -> {ok, Pid} = macula_content_transfer:start_put(Pool, Bytes), Result = macula_content_transfer:await(Pid), macula_content_transfer:cancel(Pid), Result. %% @doc As `put_content/2', dialing `Station' directly (reusing a live %% link or dialing + waiting up to `TimeoutMs' for one) instead of %% picking from the pool's existing links — the content-transfer %% counterpart to `call_station/7'. `Station' and `TimeoutMs' mean %% exactly what they do there; the underlying block/manifest transfer %% has its own internal timeouts regardless of `TimeoutMs', which %% bounds only the connect wait. See `macula_direct_dial:put_content/4' %% to resolve a station by identity and put in one call. -spec put_content_station(pool(), macula_client:seed(), binary(), pos_integer()) -> {ok, mcid()} | {error, term()}. put_content_station(Pool, Station, Bytes, TimeoutMs) -> put_content_station(Pool, Station, Bytes, TimeoutMs, #{}). %% @doc As `put_content_station/4', with a per-call TLS trust override %% for this dial — `verify', `expected_node_id', `pin_tls_cert' (see %% `call_station/8'). -spec put_content_station(pool(), macula_client:seed(), binary(), pos_integer(), map()) -> {ok, mcid()} | {error, term()}. put_content_station(Pool, Station, Bytes, TimeoutMs, Opts) -> {ok, Pid} = macula_content_transfer:start_put_station( Pool, Station, Bytes, TimeoutMs, Opts), Result = macula_content_transfer:await(Pid), macula_content_transfer:cancel(Pid), Result. %% @doc Fetch the bytes for a previously-stored MCID. Returns %% `{error, not_found}' if no provider in the pool's reach holds a %% copy (for chunked content, if any single chunk is unreachable). %% Dispatches on the MCID's codec byte: `16#55' (raw/single-block) %% fetches one block, BLAKE3-verified by the station before it leaves %% the store; `16#56' (manifest) fetches the manifest, then every %% chunk in order, reassembles, and verifies the whole against the %% manifest's size and Merkle root before returning. %% %% A thin blocking wrapper over `macula_content_transfer:start_get/2' + %% `await/1' — see the note on `put_content/2'. %% `MCID' must carry one of the two codec bytes `put_content/2' ever %% mints (`16#55' single-block, `16#56' chunked manifest) — anything %% else can't have come from this SDK's own put path (a corrupted %% record, a caller's encoding bug, or hostile input on a path that %% turns user-controlled bytes into an MCID) and is rejected here %% rather than reaching `macula_content_transfer''s internal dispatch, %% whose `is_chunked/2' clauses assume this shape and previously %% crashed the calling process's linked worker on anything else. -spec get_content(pool(), mcid()) -> {ok, binary()} | {error, not_found | invalid_mcid | term()}. get_content(Pool, <<2, Codec, _:48/binary>> = MCID) when is_pid(Pool), (Codec =:= 16#55 orelse Codec =:= 16#56) -> {ok, Pid} = macula_content_transfer:start_get(Pool, MCID), Result = macula_content_transfer:await(Pid), macula_content_transfer:cancel(Pid), Result; get_content(Pool, _MCID) when is_pid(Pool) -> {error, invalid_mcid}. %% @doc As `get_content/2', dialing `Station' directly (reusing a live %% link or dialing + waiting up to `TimeoutMs' for one) instead of %% picking from the pool's existing links — the content-transfer %% counterpart to `call_station/7'. `Station' and `TimeoutMs' mean %% exactly what they do there; the underlying block/manifest transfer %% has its own internal timeouts regardless of `TimeoutMs', which %% bounds only the connect wait. See `find_content_providers/2' to %% resolve a station to dial, or `macula_direct_dial:get_content/3' to %% resolve-and-fetch in one call. -spec get_content_station(pool(), macula_client:seed(), mcid(), pos_integer()) -> {ok, binary()} | {error, not_found | term()}. get_content_station(Pool, Station, MCID, TimeoutMs) -> get_content_station(Pool, Station, MCID, TimeoutMs, #{}). %% @doc As `get_content_station/4', with a per-call TLS trust override %% for this dial — `verify', `expected_node_id', `pin_tls_cert' (see %% `call_station/8'). %% See `get_content/2' on why a malformed `MCID' is rejected here %% rather than reaching `macula_content_transfer'. -spec get_content_station(pool(), macula_client:seed(), mcid(), pos_integer(), map()) -> {ok, binary()} | {error, not_found | invalid_mcid | term()}. get_content_station(Pool, Station, <<2, Codec, _:48/binary>> = MCID, TimeoutMs, Opts) when Codec =:= 16#55 orelse Codec =:= 16#56 -> {ok, Pid} = macula_content_transfer:start_get_station( Pool, Station, MCID, TimeoutMs, Opts), Result = macula_content_transfer:await(Pid), macula_content_transfer:cancel(Pid), Result; get_content_station(_Pool, _Station, _MCID, _TimeoutMs, _Opts) -> {error, invalid_mcid}. %% @doc Resolve every host currently announcing an MCID: hosts that %% stored a chunked put (`_content.put_manifest') and got %% `content_announcement'd automatically by the station on receipt %% (`macula_content_announcer'). `get_content/2' already reaches a %% copy via the connected station's own 1-hop peer relay, so this is %% for a caller that wants to know WHO holds an MCID, or to dial a %% specific one directly with `get_content_station/4,5' — e.g. when the %% connected station's relay hop budget does not reach the host (a %% partial-mesh pair with no mutual peer), or to route around a %% specific host deliberately. %% %% Each entry is verified under the node's crypto profile before its %% `endpoint' is trusted: its signature, and that its signer is the %% `announcer_node' it names, since `macula_record:verify/2' refuses a %% record whose payload names another signer. Unverifiable records and %% records of another type are dropped, not surfaced as errors. %% Single-block content (put via `_content.put_block' alone) is not %% announced: resolving its MCID returns `{ok, []}'. -spec find_content_providers(pool(), mcid()) -> {ok, [map()]} | {error, term()}. find_content_providers(Pool, <<2, _Codec:8, _Hash:48/binary>> = MCID) when is_pid(Pool) -> classify_find_providers( macula_client:call_linked_station(Pool, ?DHT_REALM, ?DHT_FIND_RECORDS_PROC, #{key => macula_record:content_key(MCID)}, ?DHT_RECORD_TIMEOUT_MS)). classify_find_providers({ok, Wires}) when is_list(Wires) -> with_profile(fun(Profile) -> {ok, providers(Wires, Profile)} end); classify_find_providers({ok, Reply}) -> {error, {unexpected_reply, Reply}}; classify_find_providers({error, _} = E) -> E. providers(Wires, Profile) -> lists:filtermap(fun(Wire) -> decode_provider(Wire, Profile) end, Wires). %% A provider from a content announcement that verifies under Profile. %% The verification covers the signer: a record merely stored under the %% right key but signed by a node other than the `announcer_node' it %% names is refused with key_id_mismatch, the same class of gap %% `macula_direct_dial' closes for `station_endpoint'. decode_provider(Signed, Profile) -> provider(macula_record:verify(Signed, Profile)). provider({ok, #{type := ?TYPE_CONTENT_ANNOUNCEMENT} = Record}) -> {true, macula_record:read_content_announcement(Record)}; provider(_RefusedOrAnotherType) -> false. %%%=================================================================== %%% Streaming RPC (v1.5.0+) %%%=================================================================== %%% %%% Streaming RPC ships in two phases (see PLAN_MACULA_STREAMING.md): %%% %%% Phase 1 (this release) — LOCAL dispatch only. Client and server %%% halves both live in the same BEAM and are paired in-process. The %%% public surface below is what cross-node streaming will use; only %%% the transport behind the scenes changes in Phase 2. %%% %%% Phase 2 — wire STREAM_OPEN / STREAM_DATA / STREAM_END / STREAM_ERROR %%% / STREAM_REPLY frames through QUIC, one QUIC stream per call_id. %%% %%% Patterns supported (gRPC taxonomy): %%% server_stream — single Args, streamed reply %%% client_stream — streamed Args, single reply %%% bidi — duplex %%% %%% Example (server-stream): %%% %%% ok = macula:advertise_stream(<<"foo.count">>, server_stream, %%% fun(Stream, #{n := N}) -> %%% [ok = macula:send(Stream, integer_to_binary(I)) %%% || I <- lists:seq(1, N)], %%% macula:close_stream(Stream) %%% end), %%% {ok, S} = macula:call_stream(<<"foo.count">>, #{n => 5}), %%% drain(S). %%% %%% drain(S) -> %%% case macula:recv(S) of %%% {chunk, Bin} -> io:format("~s~n", [Bin]), drain(S); %%% eof -> ok %%% end. %% @doc Open a LOCAL in-process server-stream call. Used for unit %% tests and same-BEAM dispatch via `macula_stream_local'. -spec call_stream(procedure(), term()) -> {ok, stream()} | {error, term()}. call_stream(Procedure, Args) when is_binary(Procedure) -> call_stream(Procedure, Args, #{}). %% @doc Open a LOCAL in-process server-stream call with options. -spec call_stream(procedure(), term(), map()) -> {ok, stream()} | {error, term()}. call_stream(Procedure, Args, Opts) when is_binary(Procedure), is_map(Opts) -> macula_stream_local:call_stream(Procedure, Args, Opts). %% @doc Open a streaming RPC to `Procedure''s provider in `Realm': resolve %% the provider through its `procedure_advertisement' and open the stream %% at its serving station, naming the provider as the target, as %% `call/5' does for a single-reply call. Same as %% `macula_direct_dial:call_stream/5'. The returned stream is bound to %% that station's link (errors with `peer_down' if the link dies; caller %% re-opens). %% `Opts' `ucan_token' presents a UCAN to a streaming procedure %% advertised with an `auth' policy (see `advertise_stream/6'). %% An open whose signed STREAM_OPEN would be longer than %% `max_stream_open_bytes' (1 MiB by default) returns %% `{error, {open_too_large, Limit}}' without sending anything. -spec call_stream(pool(), realm(), procedure(), term(), map()) -> {ok, stream()} | {error, term()}. call_stream(Pool, Realm, Procedure, Args, Opts) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Procedure), is_map(Opts) -> macula_direct_dial:call_stream(Pool, Realm, Procedure, Args, Opts). %% @doc Open a streaming RPC to `Target', a provider's node_id, by DIALING %% a specific station directly (direct-dial): the streaming analogue of %% `call_station/7'. Compose it with DHT resolution %% (`find_records' -> `read_procedure_advertisement' -> `station_endpoint') %% to reach a stream provider in one hop, exactly as a unary caller does. %% `Opts' may set `dial_timeout_ms' (default 10_000) and a `mode'. %% `Opts' also carries the per-call TLS trust override for this dial: %% `verify', `expected_node_id', `pin_tls_cert' (see %% `macula_client:call_station/8'). -spec call_stream_station(pool(), macula_client:seed(), <<_:256>>, realm(), procedure(), term(), map()) -> {ok, stream()} | {error, term()}. call_stream_station(Pool, Station, Target, Realm, Procedure, Args, Opts) when is_pid(Pool), is_binary(Target), byte_size(Target) =:= 32, is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Procedure), is_map(Opts) -> macula_client:call_stream_station(Pool, Station, Target, Realm, Procedure, Args, Opts). %% @doc Open a LOCAL in-process client-stream or bidi call. Used %% for unit tests and same-BEAM dispatch via `macula_stream_local'. -spec open_stream(procedure(), term(), map()) -> {ok, stream()} | {error, term()}. open_stream(Procedure, Args, Opts) when is_binary(Procedure), is_map(Opts) -> macula_stream_local:open_stream(Procedure, Args, Opts). %% @doc Open a LOCAL in-process stream with explicit mode. -spec open_stream(procedure(), term(), map(), stream_mode()) -> {ok, stream()} | {error, term()}. open_stream(Procedure, Args, Opts, Mode) when is_binary(Procedure), is_map(Opts), is_atom(Mode) -> macula_stream_local:open_stream(Procedure, Args, Opts#{mode => Mode}). %% @doc Advertise a LOCAL in-process streaming procedure %% (default: server_stream). -spec advertise_stream(procedure(), stream_handler()) -> ok | {error, term()}. advertise_stream(Procedure, Handler) when is_binary(Procedure), is_function(Handler, 2) -> advertise_stream(Procedure, server_stream, Handler). %% @doc Advertise a LOCAL in-process streaming procedure with mode. -spec advertise_stream(procedure(), stream_mode(), stream_handler()) -> ok | {error, term()}. advertise_stream(Procedure, Mode, Handler) when is_binary(Procedure), is_atom(Mode), is_function(Handler, 2) -> macula_stream_local:advertise(Procedure, Mode, Handler). %% @doc Register a streaming procedure handler on a V2 pool. Fans out to %% every healthy link and stores in pool state for replay on link %% respawn. A caller reaches this provider only through a %% `procedure_advertisement' record that names it; registering the %% handler publishes none. See `macula_client:advertise_stream/5'. -spec advertise_stream(pool(), realm(), procedure(), stream_mode(), stream_handler()) -> ok | {error, term()}. advertise_stream(Pool, Realm, Procedure, Mode, Handler) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Procedure), (Mode =:= server_stream orelse Mode =:= client_stream orelse Mode =:= bidi), is_function(Handler, 2) -> advertise_stream(Pool, Realm, Procedure, Mode, Handler, #{}). %% @doc As `advertise_stream/5', with `Opts'. `auth' sets the streaming %% procedure's policy, the same set `advertise/5' takes: `open' (default), %% `{ucan_required, Issuer}' or `{realm_member_required, RealmDid, %% RequiredCan}'. A consumer presents its token with `call_stream/5''s %% `ucan_token' opt. -spec advertise_stream(pool(), realm(), procedure(), stream_mode(), stream_handler(), map()) -> ok | {error, term()}. advertise_stream(Pool, Realm, Procedure, Mode, Handler, Opts) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Procedure), (Mode =:= server_stream orelse Mode =:= client_stream orelse Mode =:= bidi), is_function(Handler, 2), is_map(Opts) -> Policy = maps:get(auth, Opts, open), advertise_authorized(Pool, Realm, Procedure, fun(EncodedAd) -> macula_client:advertise_stream(Pool, Realm, Procedure, Mode, Handler, Policy, EncodedAd) end). %% The provider-authorization resolution behind `advertise' and %% `advertise_stream': the pool's own D25 chain, fetched from the DHT, %% signed and verified before a single frame goes out. Any missing %% piece fails fast under `provider_authorization'. advertise_authorized(Pool, Realm, Procedure, Fun) -> case macula_record:procedure_org(Procedure) of none -> {error, {provider_authorization, no_org_namespace}}; {error, malformed} -> {error, {provider_authorization, malformed_procedure}}; {org, Org} -> case status(Pool) of {ok, #{self_node_id := NodeId}} -> resolve_org_directory(Pool, Realm, Org, Procedure, NodeId, Fun); {error, _} = E -> E end end. resolve_org_directory(Pool, Realm, Org, Procedure, NodeId, Fun) -> case find_record(Pool, macula_record:org_directory_key(Realm, Org)) of {ok, OrgDir} -> #{org_key := OrgKeyId} = macula_record:read_org_directory(OrgDir), resolve_delegation(Pool, Realm, Procedure, OrgDir, OrgKeyId, NodeId, Fun); {error, not_found} -> {error, {provider_authorization, {org_directory, not_found}}}; {error, _} = E -> {error, {provider_authorization, E}} end. resolve_delegation(Pool, Realm, Procedure, OrgDir, OrgKeyId, NodeId, Fun) -> case find_record(Pool, macula_record:procedure_delegation_key(OrgKeyId, NodeId)) of {ok, Deleg} -> sign_provider_advertisement(Pool, Realm, Procedure, OrgDir, Deleg, NodeId, Fun); {error, not_found} -> {error, {provider_authorization, {procedure_delegation, not_found}}}; {error, _} = E -> {error, {provider_authorization, E}} end. sign_provider_advertisement(Pool, Realm, Procedure, OrgDir, Deleg, NodeId, Fun) -> Authorization = #{org_directory => macula_record:encode(OrgDir), procedure_delegation => macula_record:encode(Deleg)}, Unsigned = macula_record:procedure_advertisement( NodeId, Realm, Procedure, NodeId, #{authorization => Authorization}), case sign_node_record(Pool, Unsigned) of {ok, Signed} -> trusted_provider_advertisement(Pool, Realm, Signed, Fun); {error, _} = E -> {error, {provider_authorization, E}} end. %% The advertisement goes out only after its authorization verifies %% against the realm key this pool pins — a chain the pool cannot %% check never reaches a station. trusted_provider_advertisement(Pool, Realm, Signed, Fun) -> case macula_client:realm_key(Pool, Realm) of {ok, RealmKey} -> {ok, Profile} = macula_crypto_profile:configured(), Trust = #{profile => Profile, realm_key => RealmKey}, case macula_record:verify_authorization( Signed, Trust, erlang:system_time(millisecond)) of ok -> Fun(macula_record:encode(Signed)); {error, _} = E -> {error, {provider_authorization, E}} end; none -> {error, {provider_authorization, no_realm_key}} end. %% @doc Stop advertising a LOCAL streaming procedure. -spec unadvertise_stream(procedure()) -> ok. unadvertise_stream(Procedure) when is_binary(Procedure) -> macula_stream_local:unadvertise(Procedure). %% @doc Stop advertising a streaming procedure on a V2 pool. -spec unadvertise_stream(pool(), realm(), procedure()) -> ok. unadvertise_stream(Pool, Realm, Procedure) when is_pid(Pool), is_binary(Realm), byte_size(Realm) =:= 32, is_binary(Procedure) -> macula_client:unadvertise_stream(Pool, Realm, Procedure). %% @doc Send a binary chunk on the stream. -spec send(stream(), binary()) -> ok | {error, term()}. send(Stream, Bin) when is_pid(Stream), is_binary(Bin) -> macula_stream:send(Stream, Bin). %% @doc Send a chunk with explicit encoding. -spec send(stream(), binary() | term(), raw | msgpack) -> ok | {error, term()}. send(Stream, Body, Encoding) when is_pid(Stream) -> macula_stream:send(Stream, Body, Encoding). %% @doc Receive the next chunk (blocks). -spec recv(stream()) -> {chunk, binary()} | {data, term()} | eof | {error, term()}. recv(Stream) when is_pid(Stream) -> macula_stream:recv(Stream). -spec recv(stream(), timeout()) -> {chunk, binary()} | {data, term()} | eof | {error, term()}. recv(Stream, Timeout) when is_pid(Stream) -> macula_stream:recv(Stream, Timeout). %% @doc Close a V1 stream (both sides). Renamed from `close/1' in %% 3.11.0 because `close/1' now refers to the V2 pool surface. -spec close_stream(stream()) -> ok. close_stream(Stream) when is_pid(Stream) -> macula_stream:close(Stream). %% @doc Half-close the write side; recv still drains. -spec close_send(stream()) -> ok. close_send(Stream) when is_pid(Stream) -> macula_stream:close_send(Stream). %% @doc Wait for the terminal reply (client-stream / bidi). -spec await_reply(stream()) -> {ok, term()} | {error, term()}. await_reply(Stream) when is_pid(Stream) -> macula_stream:await_reply(Stream). -spec await_reply(stream(), timeout()) -> {ok, term()} | {error, term()}. await_reply(Stream, Timeout) when is_pid(Stream) -> macula_stream:await_reply(Stream, Timeout). %% @doc Server-side: emit the terminal reply value. -spec set_reply(stream(), term()) -> ok. set_reply(Stream, Result) when is_pid(Stream) -> macula_stream:set_reply(Stream, Result). %% @doc Abort the stream with an error frame. -spec abort(stream(), binary(), binary()) -> ok | {error, {text_too_long | invalid_text, code}}. abort(Stream, Code, Message) when is_pid(Stream), is_binary(Code), is_binary(Message) -> macula_stream:abort(Stream, Code, Message). %%%=================================================================== %%% Cluster (LAN) %%%=================================================================== %% @doc Ensure this node is running in distributed mode. -spec ensure_distributed() -> ok | {error, term()}. ensure_distributed() -> macula_cluster:ensure_distributed(). %% @doc Subscribe to node up/down events. -spec monitor_nodes() -> ok. monitor_nodes() -> macula_cluster:monitor_nodes(). %% @doc Unsubscribe from node up/down events. -spec unmonitor_nodes() -> ok. unmonitor_nodes() -> macula_cluster:unmonitor_nodes(). %%%=================================================================== %%% Mesh Distribution %%%=================================================================== %% @doc Join the Macula relay mesh with Erlang distribution. %% %% After calling this, standard OTP distribution works across firewalls. %% `Opts' takes: %%
    %%
  • `relays' (required): the V2 pool's seeds, each a map with `host', %% `port' and `expected_node_id', the relay's 32-byte node_id, which %% every dial checks (D16). A relay without one refuses the join with %% `{error, {relays, expected_node_id_required}}' before any pool %% starts.
    • %%
  • `node_identity': the V2 pool's node identity key, %% `macula_node_keys:node_key()'. Default: generated.
    • %%
%% %% Internally builds a V2 `macula_client:pool()' and registers it %% with `macula_dist_pool' as the carrier for `_dist.tunnel.*' %% traffic. Dist tunnel frames travel under the all-zeros realm %% (protocol-internal infrastructure, not bound to any user realm). -spec join_mesh(map()) -> ok | {error, term()}. join_mesh(Opts) -> joined(join_pool_args(Opts)). joined({ok, Relays, PoolOpts}) -> on_pool_for_join(macula_client:connect(Relays, PoolOpts)); joined({error, _} = Refusal) -> Refusal. %% The seeds and options of the pool a join starts: the relays, when every one names the node_id it expects, and the %% node identity key when one is given. join_pool_args(#{relays := Relays} = Opts) -> pinned_relays(lists:all(fun pinned_relay/1, Relays), Relays, maps:with([node_identity], Opts)). pinned_relays(true, Relays, PoolOpts) -> {ok, Relays, PoolOpts}; pinned_relays(false, _Relays, _PoolOpts) -> {error, {relays, expected_node_id_required}}. pinned_relay(#{host := _, port := _, expected_node_id := <<_:256>>}) -> true; pinned_relay(_Relay) -> false. on_pool_for_join({ok, Pool}) -> wait_for_pool(Pool, 30), os:putenv("MACULA_DIST_MODE", "relay"), macula_dist_pool:register_mesh_pool(Pool), macula_dist_pool:advertise_dist_accept(), ?LOG_INFO("[macula] Joined mesh — distribution enabled"), ok; on_pool_for_join({error, Reason}) -> ?LOG_ERROR("[macula] Failed to join mesh: ~p", [Reason]), {error, Reason}. %% @doc Enable Erlang distribution over a dedicated dist relay %% (`macula-io/macula-dist-relay'). %% %% Different from `join_mesh/1': %% - Connects to a dist relay (port 4434, ALPN `macula-dist'), NOT the %% pub/sub station mesh %% - No mesh_client, no pub/sub subscriptions — only dist traffic %% - Uses raw QUIC stream routing with no MessagePack overhead %% %% Options: %% - `url' (required): `<<"quic://relay.example.com:4434">>' %% %% After this returns `ok', standard OTP distribution (`rpc:call/4', %% `gen_server:call/3' across nodes, `pg' groups, etc.) works across %% firewalls via the dist relay. %% %% The relay client runs as a temporary child of the macula application %% supervisor. It does not reconnect: when the relay closes the %% connection the client ends and is not restarted. Monitor the pid from %% `dist_relay_client/0' to learn that, then call this function again. %% Returns `{error, macula_not_started}' when the macula application is %% not running. -spec join_dist_relay(map()) -> ok | {error, term()}. join_dist_relay(Opts) -> Url = maps:get(url, Opts), NodeName = atom_to_binary(node()), start_dist_relay_client(whereis(macula_root), Url, NodeName). start_dist_relay_client(undefined, _Url, _NodeName) -> {error, macula_not_started}; start_dist_relay_client(_Root, Url, NodeName) -> ChildSpec = macula_dist_relay_client:child_spec(Url, NodeName), case supervisor:start_child(macula_root, ChildSpec) of {ok, _Pid} -> os:putenv("MACULA_DIST_MODE", "dist_relay"), ?LOG_INFO("[macula] Joined dist relay ~s, distribution enabled", [Url]), ok; {error, {already_started, _Pid}} -> os:putenv("MACULA_DIST_MODE", "dist_relay"), ?LOG_INFO("[macula] dist_relay_client already running, mode set"), ok; {error, Reason} = Err -> ?LOG_ERROR("[macula] Failed to join dist relay: ~p", [Reason]), Err end. %% @doc The dist relay client that `join_dist_relay/1' started, if it %% is running. %% %% The client exits with `{relay_closed, Reason}' when the relay closes %% the connection and is not restarted. Monitor the returned pid and call %% `join_dist_relay/1' again after it goes down. -spec dist_relay_client() -> {ok, pid()} | {error, not_joined}. dist_relay_client() -> dist_relay_client_result(macula_dist_relay_client:whereis_client()). dist_relay_client_result(undefined) -> {error, not_joined}; dist_relay_client_result(Pid) -> {ok, Pid}. %% @private Wait until the V2 pool has at least one healthy %% station_link (CONNECT/HELLO completed). One-second polling, capped %% at `Retries' iterations. wait_for_pool(_Pool, 0) -> ?LOG_WARNING("[macula] Mesh pool not ready after timeout"); wait_for_pool(Pool, Retries) -> on_pool_status(macula_client:status(Pool), Pool, Retries). on_pool_status({ok, #{healthy_links := N}}, _Pool, _Retries) when N > 0 -> ?LOG_INFO("[macula] Mesh pool connected (~p healthy link(s))", [N]), ok; on_pool_status(_Other, Pool, Retries) -> timer:sleep(1000), wait_for_pool(Pool, Retries - 1). %%%=================================================================== %%% Peer-supplied maps (D26) %%%=================================================================== %% @doc A field of a map a peer supplied, or `undefined' when it is absent. See `field/3'. -spec field(atom() | binary(), map()) -> term(). field(Name, Map) -> field(Name, Map, undefined). %% @doc A field of a map a peer supplied (D26), or `Default' when it is absent. A map from the codec carries its text %% keys as `{text, Bin}'; a map handed over in process may carry atom or binary keys. The lookup tries `{text, Name}', %% then the atom, then the binary, so a handler reads both kinds of map the same way. Looking up a binary name never %% creates an atom. -spec field(atom() | binary(), map(), term()) -> term(). field(Name, Map, Default) when (is_atom(Name) orelse is_binary(Name)), is_map(Map) -> first_found(field_keys(Name), Map, Default). %% @doc The binary of a text value a peer supplied (D26): the binary of `{text, Bin}', a binary unchanged, and %% `badarg' for anything else. -spec text({text, binary()} | binary()) -> binary(). text({text, Bin}) when is_binary(Bin) -> Bin; text(Bin) when is_binary(Bin) -> Bin; text(Other) -> erlang:error(badarg, [Other]). field_keys(Name) when is_atom(Name) -> Bin = atom_to_binary(Name), [{text, Bin}, Name, Bin]; field_keys(Name) when is_binary(Name) -> [{text, Name}] ++ existing_atom(Name) ++ [Name]. existing_atom(Name) -> try [binary_to_existing_atom(Name)] catch error:badarg -> [] end. first_found([], _Map, Default) -> Default; first_found([Key | Keys], Map, Default) -> found(maps:find(Key, Map), Keys, Map, Default). found({ok, Value}, _Keys, _Map, _Default) -> Value; found(error, Keys, Map, Default) -> first_found(Keys, Map, Default).