%% @doc The attachment phase of a replication run. Runs AFTER the doc %% loop (independent lifecycles, independent checkpoint): fold the %% source's attachment feed since the attachment checkpoint, digest-diff %% each batch against the target in one round trip, stream the missing %% blobs (digest-verified at the target's commit point), apply %% LWW-guarded deletes, checkpoint per batch. %% %% Degrades to `skipped' when either transport (or the underlying %% attachment backend) has no feed support. Redelivery is safe %% end-to-end: digest diff answers `have', and the origin-HLC guard %% turns replays into `{ok, ignored}'. -module(barrel_rep_att). -include("barrel_docdb.hrl"). -export([sync/5, sync_from/7]). -define(DEFAULT_BATCH, 100). -spec sync(term(), term(), module(), module(), map()) -> {ok, map()} | skipped | {error, term()}. sync(Source, Target, SourceTransport, TargetTransport, Opts) -> case supports(SourceTransport) andalso supports(TargetTransport) of false -> skipped; true -> do_sync(Source, Target, SourceTransport, TargetTransport, Opts) end. supports(Transport) -> _ = code:ensure_loaded(Transport), erlang:function_exported(Transport, att_changes, 3). do_sync(Source, Target, SourceTransport, TargetTransport, Opts) -> RepId = maps:get(rep_id, Opts), Checkpoint = barrel_rep_checkpoint:new(#{ id => <<"att-", RepId/binary>>, source => Source, target => Target, source_transport => SourceTransport, target_transport => TargetTransport, options => Opts }), StartSeq0 = barrel_rep_checkpoint:get_start_seq(Checkpoint), StartSeq = maybe_reset_on_att_floor(StartSeq0, Source, SourceTransport, RepId), %% bridge the checkpoint object into the fun-based loop: each %% call persists the given seq under the run's session CheckpointFun = fun(Seq) -> barrel_rep_checkpoint:write_checkpoint( barrel_rep_checkpoint:set_last_seq(Seq, Checkpoint)) end, run(Source, Target, SourceTransport, TargetTransport, StartSeq, CheckpointFun, Opts). %% @doc Run the attachment phase from an explicit since with a %% caller-supplied checkpoint fun (invoked with the last visited feed %% HLC after each batch, and once more on completion). This is the %% timeline merge entry point; sync/5 wraps it with the %% replication-checkpoint machinery. -spec sync_from(term(), term(), module(), module(), barrel_hlc:timestamp() | first, fun((barrel_hlc:timestamp() | first) -> ok), map()) -> {ok, map()} | skipped | {error, term()}. sync_from(Source, Target, SourceTransport, TargetTransport, Since, CheckpointFun, Opts) -> case supports(SourceTransport) andalso supports(TargetTransport) of false -> skipped; true -> run(Source, Target, SourceTransport, TargetTransport, Since, CheckpointFun, Opts) end. run(Source, Target, SourceTransport, TargetTransport, Since, CheckpointFun, Opts) -> BatchSize = maps:get(att_batch_size, Opts, ?DEFAULT_BATCH), Stats0 = #{atts_written => 0, atts_skipped => 0, atts_ignored => 0, atts_deleted => 0, att_read_failures => 0, att_write_failures => 0, att_bytes_written => 0}, loop(Source, Target, SourceTransport, TargetTransport, Since, BatchSize, CheckpointFun, Stats0). loop(Source, Target, SourceTransport, TargetTransport, Since, BatchSize, CheckpointFun, Stats) -> case SourceTransport:att_changes(Source, Since, #{limit => BatchSize}) of {error, att_sync_unsupported} -> %% the transport speaks the protocol but the backend has %% no feed skipped; {error, _} = Error -> Error; {ok, [], _LastSeq} -> ok = CheckpointFun(Since), {ok, Stats}; {ok, Entries, LastSeq} -> case process_batch(Source, Target, SourceTransport, TargetTransport, Entries, Stats) of {ok, Stats2} -> ok = CheckpointFun(LastSeq), loop(Source, Target, SourceTransport, TargetTransport, LastSeq, BatchSize, CheckpointFun, Stats2); {error, _} = Error -> Error end end. process_batch(Source, Target, SourceTransport, TargetTransport, Entries, Stats0) -> {Puts, Deletes} = lists:partition( fun(#{op := Op}) -> Op =:= put end, Entries), DiffInput = [maps:with([id, name, digest], E) || E <- Puts], case diff(TargetTransport, Target, DiffInput) of {ok, Diff} -> Missing = sets:from_list( [{Id, Name} || #{id := Id, name := Name, status := missing} <- Diff]), Stats1 = lists:foldl( fun(Entry, Acc) -> #{id := Id, name := Name} = Entry, case sets:is_element({Id, Name}, Missing) of true -> transfer(Source, Target, SourceTransport, TargetTransport, Entry, Acc); false -> bump(atts_skipped, Acc) end end, Stats0, Puts), Stats2 = lists:foldl( fun(#{id := Id, name := Name, origin := Origin}, Acc) -> case TargetTransport:delete_attachment( Target, Id, Name, #{origin_hlc => Origin}) of ok -> bump(atts_deleted, Acc); {error, Reason} -> logger:warning( "attachment delete ~s/~s failed: ~p", [Id, Name, Reason]), bump(att_write_failures, Acc) end end, Stats1, Deletes), {ok, Stats2}; {error, _} = Error -> Error end. diff(_TargetTransport, _Target, []) -> {ok, []}; diff(TargetTransport, Target, DiffInput) -> TargetTransport:diff_attachments(Target, DiffInput). transfer(Source, Target, SourceTransport, TargetTransport, Entry, Stats) -> #{id := Id, name := Name, digest := Digest, length := Length, content_type := ContentType, origin := Origin} = Entry, case SourceTransport:get_attachment_stream(Source, Id, Name) of {ok, _Info, ReadFun} -> Meta = #{content_type => ContentType, digest => Digest, length => Length, origin_hlc => Origin}, case TargetTransport:put_attachment(Target, Id, Name, Meta, ReadFun) of {ok, ignored} -> bump(atts_ignored, Stats); {ok, _} -> bump(att_bytes_written, Length, bump(atts_written, Stats)); {error, Reason} -> logger:warning("attachment put ~s/~s failed: ~p", [Id, Name, Reason]), bump(att_write_failures, Stats) end; {error, not_found} -> %% deleted at the source since the feed read; its delete %% entry sits later in the feed bump(att_read_failures, Stats); {error, Reason} -> logger:warning("attachment read ~s/~s failed: ~p", [Id, Name, Reason]), bump(att_read_failures, Stats) end. %% A checkpoint below the source's attachment floor may predate %% forgotten tombstones: restart from the beginning (put rows are never %% swept, so a full feed walk re-enumerates everything; digest diff %% keeps the re-transfer cost at metadata only). maybe_reset_on_att_floor(first, _Source, _SourceTransport, _RepId) -> first; maybe_reset_on_att_floor(StartSeq, Source, SourceTransport, RepId) -> Floor = try SourceTransport:db_info(Source) of {ok, #{att_floor := F}} -> F; _ -> undefined catch _:_ -> undefined end, case Floor of undefined -> StartSeq; _ -> case barrel_hlc:less(StartSeq, Floor) of true -> logger:warning( "attachment checkpoint for ~s is below the " "source floor; forcing full resync", [RepId]), first; false -> StartSeq end end. bump(Key, Stats) -> bump(Key, 1, Stats). bump(Key, N, Stats) -> maps:update_with(Key, fun(V) -> V + N end, N, Stats).