Packages
macula
12.1.0
13.3.0
13.2.2
13.2.1
13.2.0
13.1.0
13.0.1
13.0.0
12.12.0
12.11.1
12.11.0
12.10.0
12.9.1
12.9.0
12.8.0
12.7.0
12.6.0
12.5.1
12.5.0
12.4.0
12.3.0
12.2.1
12.2.0
12.1.0
12.0.0
11.5.0
11.4.0
11.3.1
11.3.0
11.2.0
11.1.0
11.0.0
10.25.0
10.24.0
10.23.0
10.22.0
10.21.0
10.20.3
10.20.2
10.20.0
10.19.2
10.19.1
10.19.0
10.18.0
10.17.0
10.16.0
10.15.0
10.14.5
10.14.4
10.14.2
10.14.1
10.14.0
10.13.2
10.13.1
10.11.0
10.10.2
10.10.1
10.10.0
10.9.1
10.9.0
10.8.0
10.7.0
10.5.8
10.5.7
10.5.6
10.5.5
10.5.4
10.5.3
10.5.2
10.5.1
10.5.0
10.4.0
10.2.0
10.1.1
10.1.0
10.0.2
10.0.1
10.0.0
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', which asks this process to start the transfer)
%%% 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. Every
%%% provider that announced `Mcid' is a candidate: one whose fetch fails,
%%% or whose bytes don't verify, is passed over for the next, within
%%% `?DIRECT_DIAL_TIMEOUT_MS' (`macula_direct_dial:fetch_content/4'). 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).
%%%
%%% == Transfer I/O ==
%%%
%%% A download starts, awaits and cancels its transfer with `start_get/3',
%%% `start_get_station/5', `await/1' and `cancel/1', the
%%% `macula_content_transfer' ones by default; chooses its provider for
%%% `start_link_direct' with `fetch_content/4', `macula_direct_dial''s by
%%% default; and announces its facts with `fact_publish',
%%% `macula:publish/4' by default. `start_link/6' and
%%% `start_link_direct/6' take them in their start options, the four
%%% transfer functions as `transfer_io', checked by
%%% `macula_content_transfer:transfer_io/2', and pass a `link_io' option on
%%% to the transfer they start. A function of another shape is refused
%%% with `function_clause', in the caller.
%%%
%%% == 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, start_link/6]).
-export([start_link_direct/4, start_link_direct/5, start_link_direct/6]).
-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 how long `start_link_direct/4,5' looks for a provider whose fetch
%% succeeds: the DHT lookups, each provider's connect wait, and moving on
%% to the next provider after one fails. A transfer that has started runs
%% to its own end, bounded by its internal timeouts and by `cancel/1'.
%% Matches `macula_client:connect/2''s own `connect_timeout_ms' default.
-define(DIRECT_DIAL_TIMEOUT_MS, 30_000).
-export_type([start_opts/0]).
-type start_opts() :: #{transfer_io => macula_content_transfer:transfer_io(),
fetch_content => fun((macula:pool(), macula:mcid(), pos_integer(),
fun()) -> {ok, binary()} | {error, term()}),
fact_publish => macula_lifetime_announcer:publish(),
link_io => macula_content_transfer:link_io()}.
-record(dstate, {
module :: module(),
pool :: macula:pool(),
realm :: macula:realm(),
announce :: boolean(),
transfer_io :: macula_content_transfer:transfer_io(),
fact_publish :: macula_lifetime_announcer:publish(),
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) ->
start_link(Module, Pool, Realm, Mcid, Args, #{}).
%% @doc As `start_link/5', with start options (see "Transfer I/O" above).
-spec start_link(module(), macula:pool(), macula:realm(), macula:mcid(), term(),
start_opts()) -> {ok, pid()} | {error, term()}.
start_link(Module, Pool, Realm, Mcid, Args, Opts) when is_map(Opts) ->
gen_server:start_link(?MODULE,
{pooled, Module, Pool, Realm, Mcid, true, Args, functions(Opts)}, []).
%% @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) ->
start_link_direct(Module, Pool, Realm, Mcid, Args, #{}).
%% @doc As `start_link_direct/5', with start options (see "Transfer I/O"
%% above).
-spec start_link_direct(module(), macula:pool(), macula:realm(),
macula:mcid(), term(), start_opts()) -> {ok, pid()} | {error, term()}.
start_link_direct(Module, Pool, Realm, Mcid, Args, Opts) when is_map(Opts) ->
gen_server:start_link(?MODULE,
{direct, Module, Pool, Realm, Mcid, true, Args, functions(Opts)}, []).
%% @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).
%% The functions a download runs on, from its start options or else the
%% defaults; one of another shape is refused with function_clause, in the
%% caller. `link_io' goes on to the transfer the download starts.
functions(Opts) ->
TransferIo = macula_content_transfer:transfer_io(default_transfer_io(),
maps:get(transfer_io, Opts, undefined)),
#{transfer_io => TransferIo,
fetch_content =>
arity_4(maps:get(fetch_content, Opts, fun macula_direct_dial:fetch_content/4)),
fact_publish => arity_4(maps:get(fact_publish, Opts, fun macula:publish/4)),
transfer_opts => maps:with([link_io], Opts)}.
default_transfer_io() ->
#{start_get => fun macula_content_transfer:start_get/3,
start_get_station => fun macula_content_transfer:start_get_station/5,
await => fun macula_content_transfer:await/1,
cancel => fun macula_content_transfer:cancel/1}.
arity_4(Fun) when is_function(Fun, 4) -> Fun.
%%%===================================================================
%%% gen_server callbacks
%%%===================================================================
%% @private `Mcid' is rejected here, before `Module:init/1' runs or
%% anything is spawned/announced, when it doesn't carry one of the two
%% codec bytes `macula:put_content/2' ever mints — see `macula:get_content/2'
%% for the full reasoning. This is a share/download entry point, so
%% `Mcid' plausibly comes from outside this SDK (a share link, a
%% caller's own storage) rather than always being freshly minted.
init({DialMode, Module, Pool, Realm, <<2, Codec, _:48/binary>> = Mcid, Announce, InitArgs,
#{transfer_io := TransferIo, fact_publish := FactPublish} = Functions})
when Codec =:= 16#55 orelse Codec =:= 16#56 ->
process_flag(trap_exit, true),
case Module:init(InitArgs) of
{ok, UserState} ->
ShareId = crypto:strong_rand_bytes(16),
publish(Announce, FactPublish, Pool, Realm, ?GET_STARTED,
#{share_id => ShareId, mcid => Mcid,
chunked => is_chunked_mcid(Mcid)}),
Worker = spawn_worker(DialMode, Functions, Pool, Mcid, ShareId),
{ok, #dstate{module = Module, pool = Pool, realm = Realm,
announce = Announce, transfer_io = TransferIo,
fact_publish = FactPublish, share_id = ShareId,
worker = Worker, content_transfer = undefined,
completed = false, user = UserState}};
{stop, Reason} ->
{stop, Reason}
end;
init({_DialMode, _Module, _Pool, _Realm, _Mcid, _Announce, _InitArgs, _Functions}) ->
{stop, invalid_mcid}.
%% The lightweight proxy: have this download start the addressable
%% transfer (see `run_transfer/2'), 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, #{transfer_io := TransferIo, transfer_opts := TransferOpts}, Pool, Mcid,
ShareId) ->
Parent = self(),
Start = pooled_get(TransferIo, Pool, Mcid, TransferOpts#{share_id => ShareId}),
spawn_link(fun() -> run_transfer(Parent, TransferIo, Start) end);
%% Choosing among `Mcid''s providers runs in the worker through the
%% download's fetch function, `macula_direct_dial:fetch_content/4' by
%% default, and each provider's transfer is started by this download
%% (`transfer/3'), so a cancel reaches whichever transfer is running.
spawn_worker(direct, Functions, Pool, Mcid, ShareId) ->
Parent = self(),
spawn_link(fun() -> direct_worker_run(Functions, Pool, Mcid, ShareId, Parent) end).
direct_worker_run(#{transfer_io := TransferIo, transfer_opts := TransferOpts,
fetch_content := FetchContent}, Pool, Mcid, ShareId, Parent) ->
Fetch = fun(Endpoint, Pinned, ConnectMs, _RemainingMs) ->
Opts = maps:merge(TransferOpts, Pinned#{share_id => ShareId}),
transfer(Parent, TransferIo,
station_get(TransferIo, Pool, Endpoint, Mcid, ConnectMs, Opts))
end,
Parent ! {download_result, FetchContent(Pool, Mcid, ?DIRECT_DIAL_TIMEOUT_MS, Fetch)}.
pooled_get(#{start_get := StartGet}, Pool, Mcid, Opts) ->
fun() -> StartGet(Pool, Mcid, Opts) end.
station_get(#{start_get_station := StartGetStation}, Pool, Endpoint, Mcid, ConnectMs, Opts) ->
fun() -> StartGetStation(Pool, Endpoint, Mcid, ConnectMs, Opts) end.
%% This download starts the transfer itself, in a call from the worker,
%% so the transfer's pid is in the state before any `cancel/1' is
%% handled: a cancel handled first finds nothing started, and a cancel
%% handled after finds the pid.
transfer(Parent, #{await := Await} = TransferIo, Start) ->
{ok, CTPid} = gen_server:call(Parent, {start_transfer, Start}, infinity),
Result = Await(CTPid),
reap_content_transfer(TransferIo, CTPid),
Result.
run_transfer(Parent, TransferIo, Start) ->
Parent ! {download_result, transfer(Parent, TransferIo, Start)}.
%% @private
handle_call({start_transfer, Start}, _From, State) ->
Started = Start(),
{reply, Started, record_transfer(Started, State)};
handle_call(_Request, _From, State) ->
{reply, {error, unsupported}, State}.
record_transfer({ok, CTPid}, State) -> State#dstate{content_transfer = CTPid};
record_transfer(_NotStarted, State) -> State.
%% @private
handle_cast(_Msg, State) -> {noreply, State}.
%% @private
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{transfer_io = TransferIo, content_transfer = CTPid} = State) ->
unlink(State#dstate.worker),
exit(State#dstate.worker, kill),
reap_content_transfer(TransferIo, CTPid),
_ = announce_completed(State, {error, cancelled}),
ok.
%% Killing the proxy `worker' does NOT cascade into stopping the
%% `macula_content_transfer' it waits on — see `macula_feeder''s
%% identical helper for the full reasoning. `undefined' covers a cancel
%% handled before the worker asked for the transfer to start (still
%% resolving, for direct-dial): nothing was started. `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(_TransferIo, undefined) -> ok;
reap_content_transfer(#{cancel := Cancel}, CTPid) ->
try Cancel(CTPid) catch _:_ -> ok end,
ok.
announce_completed(#dstate{completed = true} = State, _Result) ->
State;
announce_completed(#dstate{pool = Pool, realm = Realm, announce = Announce,
fact_publish = FactPublish, share_id = ShareId} = State,
Result) ->
publish(Announce, FactPublish, 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(<<2, 16#56, _/binary>>) -> true;
is_chunked_mcid(_) -> false.
publish(false, _FactPublish, _, _, _, _) -> ok;
publish(true, FactPublish, Pool, Realm, Topic, Payload) ->
_ = FactPublish(Pool, Realm, Topic, Payload), ok.