valitydev · ttt161 · Mar 4, 2026 · Mar 9, 2026
diff --git a/apps/ff_server/src/ff_machine_handler.erl b/apps/ff_server/src/ff_machine_handler.erl
@@ -0,0 +1,44 @@
+-module(ff_machine_handler).
+
+-export([init/2, terminate/3]).
+-export([get_routes/0]).
+
+-spec get_routes() -> _.
+get_routes() ->
+    [
+        {"/traces/internal/source_v1/[:process_id]", ?MODULE, #{namespace => 'ff/source_v1'}},
+        {"/traces/internal/destination_v2/[:process_id]", ?MODULE, #{namespace => 'ff/destination_v2'}},
+        {"/traces/internal/deposit_v1/[:process_id]", ?MODULE, #{namespace => 'ff/deposit_v1'}},
+        {"/traces/internal/withdrawal_v2/[:process_id]", ?MODULE, #{namespace => 'ff/withdrawal_v2'}},
+        {"/traces/internal/withdrawal_session_v2/[:process_id]", ?MODULE, #{namespace => 'ff/withdrawal/session_v2'}}
+    ].
+
+-spec init(cowboy_req:req(), cowboy_http:opts()) ->
+    {ok, cowboy_req:req(), undefined}.
+init(Request, Opts) ->
+    Method = cowboy_req:method(Request),
+    NS = maps:get(namespace, Opts),
+    ProcessID = cowboy_req:binding(process_id, Request),
+    maybe
+        {method_is_valid, true} ?= {method_is_valid, Method =:= <<"GET">>},
+        {process_id_is_valid, true} ?= {process_id_is_valid, is_binary(ProcessID)},
+        {ok, Trace} ?= ff_machine:trace(NS, ProcessID),
+        Body = unicode:characters_to_binary(json:encode(Trace)),
+        Req = cowboy_req:reply(200, #{}, Body, Request),
+        {ok, Req, undefined}
+    else
+        {method_is_valid, false} ->
+            Req1 = cowboy_req:reply(405, #{}, <<"Method Not Allowed">>, Request),
+            {ok, Req1, undefined};
+        {process_id_is_valid, false} ->
+            Req2 = cowboy_req:reply(400, #{}, <<"Invalid ProcessID">>, Request),
+            {ok, Req2, undefined};
+        {error, <<"process not found">>} ->
+            Req3 = cowboy_req:reply(404, #{}, <<"Unknown process">>, Request),
+            {ok, Req3, undefined}
+    end.
+
+-spec terminate(term(), cowboy_req:req(), undefined) ->
+    ok.
+terminate(_Reason, _Req, _State) ->
+    ok.
diff --git a/apps/ff_server/src/ff_server.erl b/apps/ff_server/src/ff_server.erl
@@ -97,7 +97,7 @@ init([]) ->
                 handlers => WoodyHandlers,
                 event_handler => ff_woody_event_handler,
                 additional_routes =>
-                    get_prometheus_routes() ++
+                    get_prometheus_routes() ++ ff_machine_handler:get_routes() ++
                     [erl_health_handle:get_route(enable_health_logging(HealthCheck))]
             }
         )

diff --git a/apps/ff_server/test/ff_withdrawal_handler_SUITE.erl b/apps/ff_server/test/ff_withdrawal_handler_SUITE.erl
@@ -43,6 +43,7 @@
 -export([create_adjustment_already_has_status_error_test/1]).
 -export([create_adjustment_already_has_data_revision_error_test/1]).
 -export([withdrawal_state_content_test/1]).
+-export([trace_withdrawal_test/1]).
 
 -type config() :: ct_helper:config().
 -type test_case_name() :: ct_helper:test_case_name().
@@ -283,6 +284,82 @@ create_withdrawal_ok_test(_C) ->
         FinalWithdrawalState#wthd_WithdrawalState.status
     ).
 
+-spec trace_withdrawal_test(config()) -> test_return().
+trace_withdrawal_test(_C) ->
+    Cash = make_cash({1000, <<"RUB">>}),
+    Ctx = ct_objects:build_default_ctx(),
+    #{
+        party_id := PartyID,
+        wallet_id := WalletID,
+        destination_id := DestinationID
+    } = ct_objects:prepare_standard_environment(Ctx#{body => Cash}),
+    WithdrawalID = genlib:bsuuid(),
+    ExternalID = genlib:bsuuid(),
+    Context = ff_entity_context_codec:marshal(#{<<"NS">> => #{}}),
+    Metadata = ff_entity_context_codec:marshal(#{<<"metadata">> => #{<<"some key">> => <<"some data">>}}),
+    ContactInfo = #fistful_base_ContactInfo{
+        phone_number = <<"1234567890">>,
+        email = <<"test@mail.com">>
+    },
+    Params = #wthd_WithdrawalParams{
+        id = WithdrawalID,
+        party_id = PartyID,
+        wallet_id = WalletID,
+        destination_id = DestinationID,
+        body = Cash,
+        metadata = Metadata,
+        external_id = ExternalID,
+        contact_info = ContactInfo
+    },
+    {ok, _WithdrawalState} = call_withdrawal('Create', {Params, Context}),
+    succeeded = ct_objects:await_final_withdrawal_status(WithdrawalID),
+
+    TraceUrl = <<"http://localhost:8022/traces/internal/withdrawal_v2/", WithdrawalID/binary>>,
+    {ok, 200, _Headers, Ref} = hackney:get(TraceUrl),
+    {ok, Body} = hackney:body(Ref),
+    [
+        #{
+            <<"args">> := [
+                [
+                    #{<<"created">> := _},
+                    #{<<"status_changed">> := <<"pending">>},
+                    #{<<"resource_got">> := #{<<"bank_card">> := _}}
+                ],
+                #{<<"NS">> := #{}}
+            ],
+            <<"error">> := null,
+            <<"events">> := [
+                #{<<"event_id">> := 1, <<"event_timestamp">> := _, <<"event_payload">> := #{<<"created">> := _}},
+                #{<<"event_id">> := 2, <<"event_timestamp">> := _, <<"event_payload">> := #{<<"status_changed">> := _}},
+                #{<<"event_id">> := 3, <<"event_timestamp">> := _, <<"event_payload">> := #{<<"resource_got">> := _}}
+            ],
+            <<"finished">> := _,
+            <<"otel_trace_id">> := _,
+            <<"retry_attempts">> := _,
+            <<"retry_interval">> := _,
+            <<"running">> := _,
+            <<"scheduled">> := _,
+            <<"task_id">> := _,
+            <<"task_metadata">> := #{<<"range">> := #{}},
+            <<"task_status">> := <<"finished">>,
+            <<"task_type">> := <<"init">>
+        },
+        #{<<"task_status">> := <<"finished">>, <<"task_type">> := <<"timeout">>},
+        #{<<"task_status">> := <<"finished">>, <<"task_type">> := <<"timeout">>},
+        #{<<"task_status">> := <<"finished">>, <<"task_type">> := <<"timeout">>},
+        #{<<"task_status">> := <<"finished">>, <<"task_type">> := <<"timeout">>},
+        #{<<"task_status">> := <<"finished">>, <<"task_type">> := <<"timeout">>},
+        #{<<"task_status">> := <<"finished">>, <<"task_type">> := <<"timeout">>},
+        #{
+            <<"args">> := #{<<"notify">> := [<<"session_finished">> | _]},
+            <<"task_status">> := <<"finished">>,
+            <<"task_type">> := <<"call">>
+        },
+        #{<<"task_status">> := <<"finished">>, <<"task_type">> := <<"timeout">>},
+        #{<<"task_status">> := <<"finished">>, <<"task_type">> := <<"timeout">>}
+    ] = json:decode(Body),
+    ok.
+
 -spec create_withdrawal_fail_email_test(config()) -> test_return().
 create_withdrawal_fail_email_test(_C) ->
     Cash = make_cash({1000, <<"RUB">>}),

diff --git a/apps/fistful/src/ff_machine.erl b/apps/fistful/src/ff_machine.erl
@@ -55,6 +55,7 @@
 
 -export([get/3]).
 -export([get/4]).
+-export([trace/2]).
 
 -export([collapse/2]).
 -export([history/4]).
@@ -101,6 +102,8 @@
 
 %%
 
+-define(EPOCH_DIFF, 62167219200).
+
 -spec model(st(Model)) -> Model.
 -spec ctx(st(_)) -> ctx().
 -spec created(st(_)) -> timestamp() | undefined.
@@ -138,6 +141,151 @@ get(Mod, NS, ID, Range) ->
         collapse(Mod, Machine)
     end).
 
+-spec trace(namespace(), id()) -> _.
+trace(NS, ID) ->
+    maybe
+        {ok, MachineTrace} ?= machinery:trace(NS, ID, fistful:backend(NS)),
+        Trace = unmarshal_trace(MachineTrace),
+        {ok, Trace}
+    else
+        {error, _} = Error ->
+            Error
+    end.
+
+unmarshal_trace(MachineTrace) ->
+    lists:map(fun(TraceUnit) -> unmarshal_trace_unit(TraceUnit) end, MachineTrace).
+
+unmarshal_trace_unit(TraceUnit) ->
+    MachineArgs = maps:get(args, TraceUnit, undefined),
+    MachineEvents = maps:get(events, TraceUnit, []),
+    OtelTraceID = extract_trace_id(TraceUnit),
+    Error = extract_error(TraceUnit),
+    maps:merge(
+        maps:without([response, context], TraceUnit),
+        #{
+            args => json_compatible_value(MachineArgs),
+            events => unmarshal_machine_events(MachineEvents),
+            otel_trace_id => OtelTraceID,
+            error => Error
+        }
+    ).
+
+extract_trace_id(#{context := #{<<"otel">> := [OtelTraceID | _]}}) ->
+    OtelTraceID;
+extract_trace_id(_) ->
+    null.
+
+extract_error(#{response := {error, Reason}}) ->
+    json_compatible_value(Reason);
+extract_error(_) ->
+    null.
+
+json_compatible_value([]) ->
+    [];
+json_compatible_value(V) when is_list(V) ->
+    case io_lib:printable_unicode_list(V) of
+        true ->
+            unicode:characters_to_binary(V);
+        false ->
+            [json_compatible_value(E) || E <- V]
+    end;
+json_compatible_value(V) when is_map(V) ->
+    maps:fold(
+        fun(K, Val, Acc) ->
+            Acc#{json_compatible_key(K) => json_compatible_value(Val)}
+        end,
+        #{},
+        V
+    );
+%% tagged tuple - special case or not???
+json_compatible_value({K, V}) when is_atom(K) ->
+    #{K => json_compatible_value(V)};
+json_compatible_value(V) when is_tuple(V) ->
+    %% MAYBE ???
+    %% Elements = [json_compatible_value(E) || E <- tuple_to_list(V)],
+    %% #{<<"__tuple__">> => Elements};
+    [json_compatible_value(E) || E <- tuple_to_list(V)];
+json_compatible_value(true) ->
+    true;
+json_compatible_value(false) ->
+    false;
+json_compatible_value(null) ->
+    null;
+json_compatible_value(undefined) ->
+    null;
+json_compatible_value(V) when is_atom(V) ->
+    erlang:atom_to_binary(V);
+json_compatible_value(V) when is_integer(V) ->
+    V;
+json_compatible_value(V) when is_float(V) ->
+    V;
+json_compatible_value(V) when is_binary(V) ->
+    try unicode:characters_to_binary(V) of
+        Binary when is_binary(Binary) ->
+            Binary;
+        _ ->
+            content(<<"base64">>, base64:encode(V))
+    catch
+        _:_ ->
+            content(<<"base64">>, base64:encode(V))
+    end;
+%% default for other types (pid() | ref() | function() etc)
+json_compatible_value(V) ->
+    CompatVal = unicode:characters_to_binary(io_lib:format("~p", [V])),
+    content(<<"unknown">>, CompatVal).
+
+json_compatible_key(K) when
+    is_atom(K);
+    is_integer(K);
+    is_float(K)
+->
+    K;
+json_compatible_key(K) when is_list(K) ->
+    case io_lib:printable_unicode_list(K) of
+        true ->
+            unicode:characters_to_binary(K);
+        false ->
+            unicode:characters_to_binary(io_lib:format("~p", [K]))
+    end;
+json_compatible_key(K) when is_binary(K) ->
+    try unicode:characters_to_binary(K) of
+        Binary when is_binary(Binary) ->
+            Binary;
+        _ ->
+            base64:encode(K)
+        %% MAYBE ???
+        %% unicode:characters_to_binary(io_lib:format("~p", [K]))
+    catch
+        _:_ ->
+            base64:encode(K)
+        %% MAYBE ???
+        %% unicode:characters_to_binary(io_lib:format("~p", [K]))
+    end;
+json_compatible_key(K) ->
+    unicode:characters_to_binary(io_lib:format("~p", [K])).
+
+content(Type, Payload) ->
+    #{
+        <<"content_type">> => Type,
+        <<"content">> => Payload
+    }.
+
+unmarshal_machine_events(MachineEvents) ->
+    lists:map(
+        fun({EventID, _TsExt, {ev, Ts, Body}}) ->
+            #{
+                event_id => EventID,
+                event_payload => json_compatible_value(Body),
+                event_timestamp => to_unix_microseconds(Ts)
+            }
+        end,
+        MachineEvents
+    ).
+
+to_unix_microseconds({{{_Y, _M, _D}, {_H, _Min, _S}} = DateTime, Microsec}) ->
+    GregorianSeconds = calendar:datetime_to_gregorian_seconds(DateTime),
+    (GregorianSeconds - ?EPOCH_DIFF) * 1000000 + Microsec.
+
 -spec history(module(), namespace(), id(), range()) ->
     {ok, history()}
     | {error, notfound}.

diff --git a/apps/fistful/src/fistful.erl b/apps/fistful/src/fistful.erl
@@ -23,6 +23,7 @@
 -export([repair/5]).
 -export([notify/5]).
 -export([remove/3]).
+-export([trace/3]).
 
 -export([init/4]).
 -export([process_timeout/3]).
@@ -70,6 +71,10 @@ notify(NS, ID, Range, Args, Backend) ->
 remove(NS, ID, Backend) ->
     machinery:remove(NS, ID, set_backend_context(Backend)).
 
+-spec trace(namespace(), id(), machinery:backend(_)) -> _.
+trace(NS, ID, Backend) ->
+    machinery:trace(NS, ID, Backend).
+
 %%
 
 -type handler_opts() :: machinery:handler_opts(#{

diff --git a/apps/machinery_extra/src/machinery_gensrv_backend.erl b/apps/machinery_extra/src/machinery_gensrv_backend.erl
@@ -40,6 +40,7 @@
 -export([get/4]).
 -export([notify/5]).
 -export([remove/3]).
+-export([trace/3]).
 
 %% Gen Server
 
@@ -134,6 +135,10 @@ notify(_NS, _ID, _Range, _Args, _Opts) ->
 remove(_Namespace, _ID, _Opts) ->
     erlang:error({not_implemented, remove}).
 
+-spec trace(namespace(), id(), backend_opts()) -> no_return().
+trace(_Namespace, _ID, _Opts) ->
+    erlang:error({not_implemented, trace}).
+
 %% Gen Server + Supervisor
 
 -spec start_machine_link(logic_handler(_), namespace(), id(), args(_)) -> {ok, pid()}.

diff --git a/rebar.config b/rebar.config
@@ -34,7 +34,8 @@
     {thrift, {git, "https://github.com/valitydev/thrift_erlang.git", {tag, "v1.0.0"}}},
     {woody, {git, "https://github.com/valitydev/woody_erlang", {tag, "v1.1.1"}}},
     {erl_health, {git, "https://github.com/valitydev/erlang-health.git", {branch, "master"}}},
-    {machinery, {git, "https://github.com/valitydev/machinery-erlang.git", {tag, "v1.1.19"}}},
+    %{machinery, {git, "https://github.com/valitydev/machinery-erlang.git", {tag, "v1.1.19"}}},
+    {machinery, {git, "https://github.com/valitydev/machinery-erlang.git", {branch, "epic/process-tracing"}}},
     {damsel, {git, "https://github.com/valitydev/damsel.git", {tag, "v2.2.20"}}},
     {dmt_client, {git, "https://github.com/valitydev/dmt_client.git", {tag, "v2.0.3"}}},
     {fistful_proto, {git, "https://github.com/valitydev/fistful-proto.git", {tag, "v2.0.2"}}},

diff --git a/rebar.lock b/rebar.lock
@@ -71,7 +71,7 @@
   0},
  {<<"machinery">>,
   {git,"https://github.com/valitydev/machinery-erlang.git",
-       {ref,"eac35324e9adc7bfc52e7bb83148335cba01fee8"}},
+       {ref,"052c1d6ec1d2354711d060cbfbeff910410c0907"}},
   0},
  {<<"metrics">>,{pkg,<<"metrics">>,<<"1.0.1">>},2},
  {<<"mg_proto">>,
@@ -95,7 +95,7 @@
   0},
  {<<"progressor">>,
   {git,"https://github.com/valitydev/progressor.git",
-       {ref,"6033631d3e1eb9593acf7841d8a635146ff482e8"}},
+       {ref,"662fee014b1e8bccd72ec3a5959c2fd76a299274"}},
   1},
  {<<"prometheus">>,{pkg,<<"prometheus">>,<<"4.11.0">>},0},
  {<<"prometheus_cowboy">>,{pkg,<<"prometheus_cowboy">>,<<"0.1.9">>},0},