Packages
macula
9.13.8
9.13.8
9.8.2
9.8.1
9.8.0
9.5.0
9.4.0
9.3.1
9.3.0
9.2.0
9.1.1
9.1.0
9.0.0
8.7.0
8.6.0
8.5.0
8.4.1
8.4.0
8.3.0
8.2.0
8.1.0
8.0.2
8.0.1
8.0.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_download.erl
%%%-------------------------------------------------------------------
%%% @doc Behaviour for supervised content downloads (the get/fetch side).
%%%
%%% `start_link/4,5' returns immediately with a pid, delivers the
%%% outcome to `Module:handle_downloaded/2', and publishes
%%% `sharing.get_started_v1' / `sharing.get_completed_v1' mesh facts
%%% around the transfer — including `outcome => cancelled' if the
%%% download is cancelled before the get resolves.
%%%
%%% This is content sharing, not general-purpose RPC streaming — see
%%% `macula_streamer' / `macula_stream_sink' for that (`streaming.*'
%%% facts belong to that pair).
%%%
%%% == Real cancel, real underneath ==
%%%
%%% Internally this drives `macula_content_transfer' (PLAN_PUSH_UPLOAD.md
%%% Phase 4) rather than a blocking `macula:get_content/2' call run in
%%% a linked worker — see `macula_feeder''s module doc for the full
%%% reasoning (the same gap, the same fix, mirrored here): a blocking
%%% call gives `cancel/1' no addressable handle to the actual transfer,
%%% so it could only ever kill the local worker waiting on it, leaving
%%% the underlying `macula_content_transfer' orphaned — running to
%%% completion or sitting resolved-but-never-reaped forever, since
%%% nothing links a `gen_server:call' caller's death to the callee.
%%% This module now holds the `macula_content_transfer' pid directly (a
%%% `content_transfer' state field, alongside the lightweight resolve +
%%% await proxy `worker' that reports it back) so `cancel/1' reaches it
%%% for a real, peer-visible QUIC RESET_STREAM abort. The share_id this
%%% module already minted for its own `sharing.*' mesh facts is
%%% threaded through as `macula_content_transfer''s own `share_id' too.
%%%
%%% == Direct-dial ==
%%%
%%% `start_link/4,5' fetches through the pool's own connected link
%%% (whichever `pick_connected_link/1' picks), reaching a copy via that
%%% station's 1-hop peer relay. `start_link_direct/4,5' is the
%%% direct-dial counterpart: it resolves `Mcid''s provider from its
%%% signed `content_announcement' (published automatically by the
%%% provider's station on receipt — nothing to advertise explicitly, no
%%% direct-dial counterpart needed on the `macula_feeder' side, a fast,
%%% non-addressable DHT lookup that stays a plain blocking call inside
%%% the resolve+await proxy — nothing has ever needed to cancel
%%% mid-resolve) and dials that station directly, in one hop, via
%%% `macula_content_transfer:start_get_station/5', instead of depending
%%% on the caller's own station being able to reach it via relay. Only
%%% chunked content is discoverable this way — see
%%% `macula:find_content_providers/2'. See `macula_direct_dial''s module
%%% doc, "Content" section, for the trust model (deliberately lighter
%%% than RPC's — content is self-verifying by hash).
%%%
%%% == Example ==
%%%
%%% ```
%%% -module(doc_download).
%%% -behaviour(macula_download).
%%% -export([init/1, handle_downloaded/2]).
%%%
%%% init(Parent) -> {ok, Parent}.
%%%
%%% handle_downloaded(Result, Parent) ->
%%% Parent ! {downloaded, Result},
%%% {stop, normal, Parent}.
%%% '''
%%%
%%% ```
%%% {ok, Pid} = macula_download:start_link(doc_download, Pool, Realm,
%%% Mcid, self()).
%%% '''
%%% @end
%%%-------------------------------------------------------------------
-module(macula_download).
-behaviour(gen_server).
-export([start_link/4, start_link/5]).
-export([start_link_direct/4, start_link_direct/5]).
-export([cancel/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]).
-callback init(Args :: term()) ->
{ok, State :: term()} | {stop, Reason :: term()}.
-callback handle_downloaded(Result :: {ok, binary()} | {error, term()}, State :: term()) ->
{noreply, NewState :: term()} | {stop, Reason :: term(), NewState :: term()}.
-define(GET_STARTED, <<"sharing.get_started_v1">>).
-define(GET_COMPLETED, <<"sharing.get_completed_v1">>).
%% Bounds only the QUIC handshake wait when `start_link_direct/4,5'
%% must dial a fresh link — matches `macula_client:connect/2''s own
%% `connect_timeout_ms' default. The block/manifest transfer that
%% follows has its own separate, internal timeouts regardless.
-define(DIRECT_DIAL_CONNECT_TIMEOUT_MS, 30_000).
-record(dstate, {
module :: module(),
pool :: macula:pool(),
realm :: macula:realm(),
announce :: boolean(),
share_id :: binary(),
worker :: pid(),
content_transfer :: pid() | undefined,
completed :: boolean(),
user :: term()
}).
%% @doc Start a download. Fetches `Mcid' via `Pool'.
-spec start_link(module(), macula:pool(), macula:realm(), macula:mcid()) ->
{ok, pid()} | {error, term()}.
start_link(Module, Pool, Realm, Mcid) ->
start_link(Module, Pool, Realm, Mcid, undefined).
%% @doc As `start_link/4', with `Args' passed to `Module:init/1'.
-spec start_link(module(), macula:pool(), macula:realm(), macula:mcid(), term()) ->
{ok, pid()} | {error, term()}.
start_link(Module, Pool, Realm, Mcid, Args) ->
gen_server:start_link(?MODULE,
{pooled, Module, Pool, Realm, Mcid, true, Args}, []).
%% @doc As `start_link/4', but resolves and dials the MCID's provider
%% directly instead of fetching through the pool's existing links. See
%% the "Direct-dial" section above.
-spec start_link_direct(module(), macula:pool(), macula:realm(),
macula:mcid()) -> {ok, pid()} | {error, term()}.
start_link_direct(Module, Pool, Realm, Mcid) ->
start_link_direct(Module, Pool, Realm, Mcid, undefined).
%% @doc As `start_link_direct/4', with `Args' passed to `Module:init/1'.
-spec start_link_direct(module(), macula:pool(), macula:realm(),
macula:mcid(), term()) -> {ok, pid()} | {error, term()}.
start_link_direct(Module, Pool, Realm, Mcid, Args) ->
gen_server:start_link(?MODULE,
{direct, Module, Pool, Realm, Mcid, true, Args}, []).
%% @doc Cancel an in-flight download. Publishes `sharing.get_completed_v1'
%% with `outcome => cancelled' if the get had not resolved yet.
-spec cancel(pid()) -> ok.
cancel(Pid) -> gen_server:stop(Pid).
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
%% @private
init({DialMode, Module, Pool, Realm, Mcid, Announce, InitArgs}) ->
process_flag(trap_exit, true),
case Module:init(InitArgs) of
{ok, UserState} ->
ShareId = crypto:strong_rand_bytes(16),
publish(Announce, Pool, Realm, ?GET_STARTED,
#{share_id => ShareId, mcid => Mcid,
chunked => is_chunked_mcid(Mcid)}),
Worker = spawn_worker(DialMode, Pool, Mcid, ShareId),
{ok, #dstate{module = Module, pool = Pool, realm = Realm,
announce = Announce, share_id = ShareId,
worker = Worker, content_transfer = undefined,
completed = false, user = UserState}};
{stop, Reason} ->
{stop, Reason}
end.
%% The lightweight proxy: start the addressable transfer, report its
%% pid back immediately (so `terminate/2' can reach it even if this
%% proxy itself gets killed mid-flight), block for the outcome, reap
%% the transfer (a no-op if it's already being cancelled from outside
%% — see `reap_content_transfer/1'), report the outcome.
spawn_worker(pooled, Pool, Mcid, ShareId) ->
Parent = self(),
spawn_link(fun() ->
{ok, CTPid} = macula_content_transfer:start_get(Pool, Mcid, #{share_id => ShareId}),
Parent ! {content_transfer, CTPid},
Result = macula_content_transfer:await(CTPid),
catch macula_content_transfer:cancel(CTPid),
Parent ! {download_result, Result}
end);
%% Resolving `Mcid''s provider stays a plain blocking DHT lookup here
%% (matches what `macula_direct_dial:get_content/3' already did) —
%% only the transfer itself becomes addressable.
spawn_worker(direct, Pool, Mcid, ShareId) ->
Parent = self(),
spawn_link(fun() ->
case macula_direct_dial:resolve_content_provider(Pool, Mcid) of
{ok, #{announcer_node := Node, endpoint := Endpoint}} ->
Opts = #{share_id => ShareId, expected_node_id => Node,
pin_tls_cert => false, verify => none},
{ok, CTPid} = macula_content_transfer:start_get_station(
Pool, Endpoint, Mcid, ?DIRECT_DIAL_CONNECT_TIMEOUT_MS, Opts),
Parent ! {content_transfer, CTPid},
Result = macula_content_transfer:await(CTPid),
catch macula_content_transfer:cancel(CTPid),
Parent ! {download_result, Result};
{error, Reason} ->
Parent ! {download_result, {error, {unresolved, Reason}}}
end
end).
%% @private
handle_call(_Request, _From, State) ->
{reply, {error, unsupported}, State}.
%% @private
handle_cast(_Msg, State) -> {noreply, State}.
%% @private
handle_info({content_transfer, CTPid}, State) ->
{noreply, State#dstate{content_transfer = CTPid}};
handle_info({download_result, Result}, State) ->
NewState = announce_completed(State, Result),
#dstate{module = Module, user = User} = NewState,
deliver(Module:handle_downloaded(Result, User), NewState#dstate{content_transfer = undefined});
handle_info({'EXIT', Worker, Reason}, #dstate{worker = Worker} = State)
when Reason =/= normal ->
{stop, {worker_crashed, Reason}, State};
handle_info(_Msg, State) ->
{noreply, State}.
deliver({noreply, NewUser}, State) -> {noreply, State#dstate{user = NewUser}};
deliver({stop, Reason, NewUser}, State) -> {stop, Reason, State#dstate{user = NewUser}}.
%% @private
terminate(_Reason, #dstate{worker = Worker, completed = true}) ->
unlink(Worker),
exit(Worker, kill),
ok;
terminate(_Reason, #dstate{content_transfer = CTPid} = State) ->
unlink(State#dstate.worker),
exit(State#dstate.worker, kill),
reap_content_transfer(CTPid),
_ = announce_completed(State, {error, cancelled}),
ok.
%% Killing the proxy `worker' does NOT cascade into stopping the
%% `macula_content_transfer' it started — see `macula_feeder''s
%% identical helper for the full reasoning. `undefined' covers the
%% window before `{content_transfer, CTPid}' has arrived yet (still
%% resolving, for direct-dial). `catch' covers the benign race between
%% the proxy's own natural reap and an external `cancel/1' landing at
%% the same time.
reap_content_transfer(undefined) -> ok;
reap_content_transfer(CTPid) -> catch macula_content_transfer:cancel(CTPid), ok.
announce_completed(#dstate{completed = true} = State, _Result) ->
State;
announce_completed(#dstate{pool = Pool, realm = Realm, announce = Announce,
share_id = ShareId} = State, Result) ->
publish(Announce, Pool, Realm, ?GET_COMPLETED,
outcome_fields(#{share_id => ShareId}, Result)),
State#dstate{completed = true}.
outcome_fields(Base, {ok, Bytes}) ->
Base#{outcome => completed, size => byte_size(Bytes)};
outcome_fields(Base, {error, cancelled}) ->
Base#{outcome => cancelled};
outcome_fields(Base, {error, Reason}) ->
Base#{outcome => failed, reason => Reason}.
is_chunked_mcid(<<1, 16#56, _/binary>>) -> true;
is_chunked_mcid(_) -> false.
publish(false, _, _, _, _) -> ok;
publish(true, Pool, Realm, Topic, Payload) ->
_ = macula:publish(Pool, Realm, Topic, Payload), ok.