Packages
macula
11.2.0
15.0.0-rc.2
15.0.0-rc.1
14.10.0
14.9.0
14.8.2
14.8.1
14.8.0
14.7.0
14.6.0
14.5.0
14.4.0
14.3.0
14.2.1
14.2.0
14.1.0
14.0.0
13.6.0
13.5.0
13.4.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_upload.erl
%%%-------------------------------------------------------------------
%%% @doc Behaviour for supervised content uploads (the receiver side of
%%% a push-initiated upload — PLAN_PUSH_UPLOAD.md Phase 6, the
%%% recipient `macula_pusher' pushes at).
%%%
%%% `advertise/5,6' registers an upload procedure. Each inbound push
%%% starts one supervised, ephemeral receiver: it reads the manifest
%%% `macula_pusher' passed as the stream's open-time `Args' (decoded
%%% via `macula_manifest:from_wire/1', the same decode content
%%% sharing's own `_content.get_manifest' path already relies on),
%%% uses it only when its MCID, recomputed from its canonical fields,
%%% is the MCID it names (`macula_manifest:verify_mcid/2'),
%%% accumulates pushed chunks, and once the sender half-closes,
%%% reassembles and verifies them against the manifest with
%%% `macula_manifest:verify/2' — receiver-side verification, never
%%% sender-trusted, matching content-sharing's existing "content is
%%% self-verifying by hash" model exactly; the sender's claimed
%%% manifest proves nothing on its own. Delivers
%%% `{ok, Mcid, Bytes} | {error, _}' to `Module:handle_uploaded/2', and
%%% publishes `sharing.upload_started_v1' / `sharing.upload_completed_v1'
%%% mesh facts around each push's lifetime.
%%%
%%% == A correction from the plan's literal wording ==
%%%
%%% The plan that named this module described it as "mirroring
%%% `macula_download''s shape" while, in the same sentence, also
%%% saying its public API is `advertise'/`advertise_direct' —
%%% `macula_download' doesn't have those; it has `start_link/4,5' (a
%%% one-shot, caller-initiated fetch). What this module actually is —
%%% a long-lived, ADVERTISED provider spawning one ephemeral child per
%%% inbound push — mirrors `macula_streamer''s shape, not
%%% `macula_download''s. Built directly on top of it: this module IS a
%%% `macula_streamer' callback module internally (`?MODULE' is passed
%%% as `macula_streamer:advertise/6''s own `Module' argument, closing
%%% over the caller's `Module'/`Args' in its own state), reusing
%%% Phase 5's supervision, `client_stream' receive loop
%%% (`handle_chunk/2'), and abort-wired cancel for free — "inherits...
%%% cancel from day one, nothing to retrofit later," per the plan's own
%%% intro to this phase.
%%%
%%% == The terminal reply, and why `handle_uploaded/2' return values
%%% are ignored ==
%%%
%%% Once the sender half-closes, `macula_streamer' calls this module's
%%% `handle_eof/1' (Phase 6's new optional callback on that module) —
%%% the one place with access to the raw stream needed to set a
%%% terminal reply. Verification happens right there: `{reply, {ok,
%%% Mcid}, NewState}' on success or `{reply, {error, Reason}, NewState}'
%%% on failure, which `macula_streamer' turns into
%%% `macula_stream:set_reply/2' / `set_error/2' — exactly the channel
%%% `macula_pusher''s own `macula:await_reply/1' blocks on. The Mcid
%%% itself needs no separate round trip: it's a field already present
%%% in the manifest both sides hold (`macula_manifest' mirrors
%%% macula-station's own algorithm BYTE-FOR-BYTE). It is trustworthy to
%%% echo back because `handle_open/2' recomputed it from the manifest's
%%% canonical fields, and the actually-received bytes then verified
%%% against that manifest's size and root hash.
%%%
%%% `Module:handle_uploaded/2' — the LOCAL delivery, to whatever
%%% application registered this upload handler — fires separately, from
%%% this module's own `terminate/2' (after the wire-level reply is
%%% already set, so a slow local callback can never block it). Because
%%% `terminate/2' can no longer act on a `{stop, _, _}' return the way a
%%% live `handle_info' clause could, `handle_uploaded/2''s return value
%%% is ignored — `any()', not `{noreply,_} | {stop,_,_}' — the same
%%% shape `macula_stream_sink''s own `handle_close/2' already uses for
%%% the identical reason (it too only ever fires from `terminate/2').
%%% This is a deliberate divergence from `macula_feeder'/`macula_download''s
%%% `handle_fed'/`handle_downloaded' callbacks, which DO get a
%%% meaningful `{stop,_,_}' because they fire from a live `handle_info'
%%% clause, not `terminate/2' — "mirrors macula_feeder's shape exactly"
%%% does not hold for this specific callback's return contract, traced
%%% to WHERE each one is actually invoked from.
%%%
%%% A malicious or buggy sender that never half-closes, or that pushes
%%% more chunks than its own manifest declared, is bounded: exceeding
%%% the declared `chunk_count' stops the receiver (a genuine
%%% `{error, too_many_chunks}', abort-wired same as any other non-normal
%%% stop) rather than accumulating chunks without limit — the manifest
%%% is a system-boundary input from an untrusted remote peer, worth
%%% guarding explicitly, unlike `macula_manifest:verify/2''s own
%%% eventual `size_mismatch', which would only catch this after the
%%% fact.
%%%
%%% == Direct-dial ==
%%%
%%% `advertise_direct/6,7' is `macula_streamer:advertise_direct/6,7'
%%% verbatim, `mode => client_stream' folded into `Opts' before
%%% forwarding — see that module's own "Direct-dial" section. (Building
%%% this surfaced a real, separate bug in `macula_streamer:advertise_direct/7'
%%% itself: it called the arity-5 `advertise/5', silently discarding
%%% whatever `mode' `Opts' carried, so ANY direct-dial-advertised
%%% `client_stream' procedure — not just this module's — would have been
%%% served as `server_stream' instead, with no error anywhere to say
%%% so. Fixed at the source, in `macula_streamer.erl' itself, per this
%%% project's "fix bugs in owned libraries immediately" rule — not
%%% specific to Phase 6, but found while building it.)
%%%
%%% == Stream I/O ==
%%%
%%% `advertise/6' and `advertise_direct/7' pass their options on to
%%% `macula_streamer', `stream_io' and `advertise_stream' among them, and
%%% `advertise_direct/7' also `publish_advertisement' (see its "Stream
%%% I/O" section). Each upload announces its `sharing.upload_*' facts
%%% with `fact_publish', `macula:publish/4' by default.
%%%
%%% == Example ==
%%%
%%% ```
%%% -module(doc_upload).
%%% -behaviour(macula_upload).
%%% -export([init/1, handle_uploaded/2]).
%%%
%%% init(Parent) -> {ok, Parent}.
%%%
%%% handle_uploaded(Result, Parent) ->
%%% Parent ! {uploaded, Result},
%%% ok.
%%% '''
%%%
%%% ```
%%% {ok, _Sup} = macula_upload:advertise(Pool, Realm,
%%% <<"bulk.ingest">>, doc_upload, self()).
%%% '''
%%% @end
%%%-------------------------------------------------------------------
-module(macula_upload).
-behaviour(macula_streamer).
-export([advertise/5, advertise/6, advertise_direct/6, advertise_direct/7,
unadvertise/3]).
-export([init/1, handle_open/2, handle_chunk/2, handle_eof/1, terminate/2]).
-callback init(Args :: term()) ->
{ok, State :: term()} | {stop, Reason :: term()}.
-callback handle_uploaded(
Result :: {ok, macula:mcid(), binary()} | {error, term()},
State :: term()) -> any().
-define(UPLOAD_STARTED, <<"sharing.upload_started_v1">>).
-define(UPLOAD_COMPLETED, <<"sharing.upload_completed_v1">>).
-record(ustate, {
module :: module(),
pool :: macula:pool(),
realm :: macula:realm(),
announce :: boolean(),
fact_publish :: macula_lifetime_announcer:publish(),
share_id :: binary() | undefined,
manifest :: map() | {error, term()} | undefined,
acc = [] :: [binary()],
result :: {ok, macula:mcid(), binary()} | {error, term()} | undefined,
user :: term()
}).
%% @doc Advertise `Procedure' on `Pool'/`Realm' as an upload target.
%% Each push sent at it starts one supervised, ephemeral receiver,
%% threading state through `Module:init/1' and delivering the final
%% outcome to `Module:handle_uploaded/2'.
-spec advertise(macula:pool(), macula:realm(), macula:procedure(),
module(), term()) -> {ok, pid()} | {error, term()}.
advertise(Pool, Realm, Procedure, Module, Args) ->
advertise(Pool, Realm, Procedure, Module, Args, #{}).
%% @doc As `advertise/5'. `Opts' may include `announce' (default
%% `true') for this module's OWN `sharing.upload_*' facts and
%% `fact_publish', the function it announces them with. The other options,
%% such as `auth', `reuse_sup', `stream_io' and `advertise_stream', go on
%% to `macula_streamer:advertise/6', with `mode' `client_stream' (see
%% "Stream I/O" above).
-spec advertise(macula:pool(), macula:realm(), macula:procedure(),
module(), term(), map()) -> {ok, pid()} | {error, term()}.
advertise(Pool, Realm, Procedure, Module, Args, Opts) ->
Announce = maps:get(announce, Opts, true),
FactPublish = arity_4(maps:get(fact_publish, Opts, fun macula:publish/4)),
StreamerOpts = maps:remove(fact_publish, Opts),
macula_streamer:advertise(Pool, Realm, Procedure, ?MODULE,
{Module, Pool, Realm, Announce, FactPublish, Args},
StreamerOpts#{mode => client_stream, announce => false}).
%% @doc As `advertise/5', and additionally publishes a signed
%% `procedure_advertisement' DHT record naming this pool's connected
%% station as the server, so `macula_pusher:start_link_direct/5,6' can
%% resolve and dial here directly. See `macula_streamer:advertise_direct/6,7'.
-spec advertise_direct(macula:pool(), macula:realm(), macula:procedure(),
module(), term(), macula_node_keys:node_key()) ->
{ok, pid()} | {error, term()}.
advertise_direct(Pool, Realm, Procedure, Module, Args, NodeIdentity) ->
advertise_direct(Pool, Realm, Procedure, Module, Args, NodeIdentity, #{}).
%% @doc As `advertise_direct/6', with `Opts' forwarded to
%% `macula_streamer:advertise_direct/7', and so to the advertisement
%% publish, e.g. `authorization', the provider authorization an org
%% namespaced procedure needs, except `fact_publish', the function this
%% module announces its own facts with.
-spec advertise_direct(macula:pool(), macula:realm(), macula:procedure(),
module(), term(), macula_node_keys:node_key(), map()) ->
{ok, pid()} | {error, term()}.
advertise_direct(Pool, Realm, Procedure, Module, Args, NodeIdentity, Opts) ->
Announce = maps:get(announce, Opts, true),
FactPublish = arity_4(maps:get(fact_publish, Opts, fun macula:publish/4)),
StreamerOpts = maps:remove(fact_publish, Opts),
macula_streamer:advertise_direct(Pool, Realm, Procedure, ?MODULE,
{Module, Pool, Realm, Announce, FactPublish, Args},
NodeIdentity,
StreamerOpts#{mode => client_stream, announce => false}).
%% @doc Stop advertising `Procedure'.
-spec unadvertise(macula:pool(), macula:realm(), macula:procedure()) -> ok.
unadvertise(Pool, Realm, Procedure) ->
macula_streamer:unadvertise(Pool, Realm, Procedure).
%%%===================================================================
%%% macula_streamer callbacks (internal — this module IS the streamer
%%% callback module; the caller's OWN Module/Args are closed over
%%% above and delegated to below)
%%%===================================================================
%% @private
init({Module, Pool, Realm, Announce, FactPublish, InitArgs}) ->
case Module:init(InitArgs) of
{ok, UserState} ->
{ok, #ustate{module = Module, pool = Pool, realm = Realm,
announce = Announce, fact_publish = FactPublish,
user = UserState}};
{stop, Reason} ->
{stop, Reason}
end.
%% @private A manifest that fails to decode, or whose MCID does not
%% recompute from its canonical fields, is NOT rejected via
%% `{stop, Reason, State}' here — a `handle_open/2' stop makes
%% `macula_streamer:init/1' itself return `{stop, Reason}', a genuine
%% gen_server init failure, and OTP never calls `terminate/2' for a
%% process that failed to start. That would silently drop this push:
%% `Module:handle_uploaded/2' never fires, no
%% `sharing.upload_completed_v1' fact publishes, AND (traced, not just
%% theorized) the sender's own `macula:await_reply/1' would hang or
%% crash rather than seeing a clean `{error,_}' — nothing settles its
%% wire-level reply either, since the stream never reaches
%% `handle_eof/1'. Accepting the stream and stashing the decode error
%% in `manifest' instead means `handle_eof/1' below — the ONE place
%% already wired to set a terminal reply — reports it correctly on
%% both sides once the sender closes, exactly like any other failure.
handle_open(StreamArgs, State) ->
case bind_manifest(macula_manifest:from_wire(StreamArgs)) of
{ok, Manifest} ->
{ok, announce_started(State#ustate{manifest = Manifest})};
{error, Reason} ->
{ok, State#ustate{manifest = {error, {invalid_manifest, Reason}}}}
end.
%% The manifest names its content by MCID, and is used only when that MCID,
%% recomputed from its canonical fields, is the one it names.
bind_manifest({ok, Manifest}) ->
bound_manifest(macula_manifest:verify_mcid(Manifest, maps:get(mcid, Manifest)),
Manifest);
bind_manifest({error, _} = E) ->
E.
bound_manifest(ok, Manifest) -> {ok, Manifest};
bound_manifest({error, _} = E, _Manifest) -> E.
announce_started(#ustate{pool = Pool, realm = Realm, announce = Announce,
fact_publish = FactPublish, manifest = Manifest} = State) ->
ShareId = crypto:strong_rand_bytes(16),
publish(Announce, FactPublish, Pool, Realm, ?UPLOAD_STARTED,
#{share_id => ShareId, mcid => maps:get(mcid, Manifest),
size => maps:get(size, Manifest)}),
State#ustate{share_id = ShareId}.
%% @private Chunks pushed against a manifest that never decoded are
%% dropped, not accumulated — there's no `chunk_count' to bound
%% against, and no point: `handle_eof/1' will report the real error
%% once the sender closes, same as if no chunks had arrived at all.
%% Otherwise bounded: a sender exceeding its own manifest's declared
%% `chunk_count' is stopped rather than accumulated without limit —
%% see the module doc.
handle_chunk(_Chunk, #ustate{manifest = {error, _}} = State) ->
{noreply, State};
handle_chunk(Chunk, #ustate{manifest = #{chunk_count := ChunkCount},
acc = Acc} = State)
when length(Acc) >= ChunkCount ->
{stop, too_many_chunks, State#ustate{acc = [Chunk | Acc]}};
handle_chunk(Chunk, #ustate{acc = Acc} = State) ->
{noreply, State#ustate{acc = [Chunk | Acc]}}.
%% @private The sender is done. Reassemble, verify, and hand the
%% terminal reply straight to `macula_streamer' — see the module doc's
%% "terminal reply" section for why this can't happen anywhere else.
handle_eof(#ustate{manifest = {error, Reason}} = State) ->
{reply, {error, Reason}, State#ustate{result = {error, Reason}}};
handle_eof(#ustate{manifest = Manifest, acc = Acc} = State) ->
Data = iolist_to_binary(lists:reverse(Acc)),
reply_for(macula_manifest:verify(Manifest, Data), Manifest, Data, State).
reply_for(ok, Manifest, Data, State) ->
Mcid = maps:get(mcid, Manifest),
{reply, {ok, Mcid}, State#ustate{result = {ok, Mcid, Data}}};
reply_for({error, Reason}, _Manifest, _Data, State) ->
{reply, {error, Reason}, State#ustate{result = {error, Reason}}}.
%% @private
terminate(Reason, #ustate{module = Module, pool = Pool, realm = Realm,
announce = Announce, fact_publish = FactPublish,
share_id = ShareId, result = Result, user = UserState}) ->
announce_completed(ShareId, Announce, FactPublish, Pool, Realm, Reason, Result),
_ = Module:handle_uploaded(final_result(Reason, Result), UserState),
ok.
%% No `sharing.upload_started_v1' ever fired (a bad manifest failed
%% `handle_open/2' before `share_id' was ever minted) — nothing to
%% close out.
announce_completed(undefined, _Announce, _FactPublish, _Pool, _Realm, _Reason, _Result) ->
ok;
announce_completed(ShareId, Announce, FactPublish, Pool, Realm, Reason, Result) ->
publish(Announce, FactPublish, Pool, Realm, ?UPLOAD_COMPLETED,
outcome_fields(#{share_id => ShareId}, final_result(Reason, Result))).
final_result(_Reason, {ok, _Mcid, _Data} = R) -> R;
final_result(_Reason, {error, _} = R) -> R;
final_result(Reason, undefined) -> {error, Reason}.
outcome_fields(Base, {ok, Mcid, Data}) ->
Base#{outcome => completed, mcid => Mcid, size => byte_size(Data)};
outcome_fields(Base, {error, Reason}) ->
Base#{outcome => failed, reason => Reason}.
publish(false, _FactPublish, _, _, _, _) -> ok;
publish(true, FactPublish, Pool, Realm, Topic, Payload) ->
_ = FactPublish(Pool, Realm, Topic, Payload), ok.
arity_4(Fun) when is_function(Fun, 4) -> Fun.