fed-sx-m2: Step 9c — auto-Accept backfill drain + 6 tests
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 50s
Some checks failed
Test, Build, and Deploy / test-build-deploy (push) Failing after 50s
maybe_auto_accept/3 in http_server.erl now calls maybe_backfill/3
after the Accept publish. Flow:
inbound Follow{actor: bob, object: alice, backfill: SPEC} lands
-> pipeline ok -> append_inbox + broadcast (Step 6b)
-> maybe_auto_accept fires (Step 6c)
-> publish Accept{actor: alice, object: Follow} (Step 6c)
-> maybe_backfill (Step 9c)
-> backfill_enabled cfg gate
-> :backfill present on Follow
-> backfill:parse_mode -> Mode
-> nx_kernel:log_state_for(alice) -> LogState
-> backfill:slice(Mode, LogState, true) -> [Wrapped]
-> deliver_backfill(bob, Slice):
whereis(bob) cfg gate (peer worker registered)
-> delivery_worker:enqueue(bob, A) for each
Cfg surface:
{backfill_enabled, true} gate the drain (default off)
{auto_accept_follows, true} Step 6c gate (required)
Each backfilled entry carries {backfilled, true} (per design §13.3,
:id preserved so the receiver's replay defence still catches the
forward-going copy).
6/6 in next/tests/backfill_drain.sh:
- Follow with {backfill, {last_n, 2}} + 3 pre-published notes
-> bob's delivery_worker has exactly 2 pending entries
- Each entry carries {backfilled, true}
- :backfill_enabled absent -> no drain (back-compat)
- Follow without :backfill field -> no drain
- Missing peer worker (no whereis) -> silently skipped + 202
Step 9 fully closed (9a slicing + 9b ?since route + 9c
Accept-drain). The live HTTP dispatch of the queued entries
still gates on Blockers #2 (httpc).
This commit is contained in:
@@ -1178,12 +1178,59 @@ maybe_auto_accept(TargetAtom, Activity, Cfg) ->
|
||||
case envelope:get_field(type, Activity) of
|
||||
{ok, follow} ->
|
||||
AcceptRequest = [{type, accept}, {object, Activity}],
|
||||
nx_kernel:publish_to(TargetAtom, AcceptRequest);
|
||||
nx_kernel:publish_to(TargetAtom, AcceptRequest),
|
||||
maybe_backfill(TargetAtom, Activity, Cfg);
|
||||
_ -> ok
|
||||
end;
|
||||
_ -> ok
|
||||
end.
|
||||
|
||||
%% maybe_backfill/3 — Step 9c. If Cfg carries
|
||||
%% `{backfill_enabled, true}` AND the Follow activity carries a
|
||||
%% `:backfill` field, parse the mode, slice the receiving actor's
|
||||
%% outbox per `backfill:slice/3` (Wrap=true so each entry carries
|
||||
%% `{backfilled, true}`), and enqueue each onto the new follower's
|
||||
%% delivery_worker (registered under the follower's actor-id atom).
|
||||
%%
|
||||
%% Missing delivery_worker for the peer is silently skipped — the
|
||||
%% kernel manager lazily creates workers (or won't, in single-kernel
|
||||
%% in-process tests where the peer-worker is set up explicitly).
|
||||
|
||||
maybe_backfill(TargetAtom, FollowActivity, Cfg) ->
|
||||
case field(backfill_enabled, Cfg) of
|
||||
true ->
|
||||
case envelope:get_field(backfill, FollowActivity) of
|
||||
{ok, Spec} ->
|
||||
Mode = backfill:parse_mode(Spec),
|
||||
drain_backfill(TargetAtom, FollowActivity, Mode);
|
||||
_ -> ok
|
||||
end;
|
||||
_ -> ok
|
||||
end.
|
||||
|
||||
drain_backfill(TargetAtom, FollowActivity, Mode) ->
|
||||
case nx_kernel:log_state_for(TargetAtom) of
|
||||
{ok, LogState} ->
|
||||
Slice = backfill:slice(Mode, LogState, true),
|
||||
case envelope:get_field(actor, FollowActivity) of
|
||||
{ok, PeerId} when is_atom(PeerId) ->
|
||||
deliver_backfill(PeerId, Slice);
|
||||
_ -> ok
|
||||
end;
|
||||
_ -> ok
|
||||
end.
|
||||
|
||||
deliver_backfill(PeerId, Activities) ->
|
||||
case erlang:whereis(PeerId) of
|
||||
undefined -> ok;
|
||||
_ -> enqueue_backfill_each(PeerId, Activities)
|
||||
end.
|
||||
|
||||
enqueue_backfill_each(_, []) -> ok;
|
||||
enqueue_backfill_each(PeerId, [A | Rest]) ->
|
||||
delivery_worker:enqueue(PeerId, A),
|
||||
enqueue_backfill_each(PeerId, Rest).
|
||||
|
||||
%% broadcast_to_inbox_projections/2 — Step 6b. Cfg may carry
|
||||
%% `{inbox_projections, [Name, ...]}` listing projection gen_servers
|
||||
%% that should see every successfully-ingested inbound activity.
|
||||
|
||||
Reference in New Issue
Block a user