async CCR-Update enabler (#22 Increment 1)

apps/smf_aaa/src/smf_aaa_charging.erl

@@ -6,6 +6,7 @@
 
 -export([new/2,
 	 gy_ccr_initial/4, gy_ccr_update/4, gy_ccr_terminate/4,
+	 gy_ccr_initial_issue/4,
 	 rf_initial/4, rf_update/4, rf_terminate/4,
 	 terminate/3,
 	 handle_reply/4]).
@@ -33,6 +34,13 @@ gy_ccr_update(Ctx, Session, SOpts, Opts) ->
 gy_ccr_terminate(Ctx, Session, SOpts, Opts) ->
     invoke(Ctx, Session, SOpts, {gy, 'CCR-Terminate'}, Opts).
 
+%% gy_ccr_initial_issue/4 — the async_m counterpart of gy_ccr_initial/4: runs
+%% the CCR-Initial pre-send half on the Gy handler and returns the Promise
+%% instead of blocking for the CCA. Session_at_send/State_at_send ride the
+%% async_m do-block across the await and get folded by smf_aaa_ro:fold_cca/5.
+gy_ccr_initial_issue(Ctx, Session, SOpts, Opts) ->
+    issue(Ctx, Session, SOpts, {gy, 'CCR-Initial'}, Opts).
+
 rf_initial(Ctx, Session, SOpts, Opts) ->
     invoke(Ctx, Session, SOpts, {rf, 'Initial'}, Opts).
 
@@ -71,6 +79,20 @@ invoke(#charging_ctx{app_id = AppId, handlers = Handlers0} = Ctx,
 	    {Result, Ctx#charging_ctx{handlers = Handlers1}, Session2, Events}
     end.
 
+%% issue/5 — single-handler counterpart of invoke/5 for the async_m
+%% pre-send path: no pipeline fan-out, no Events accumulator — just the one
+%% handler's CCR-Initial pre-send + send_request, returning its Promise.
+issue(#charging_ctx{app_id = AppId, handlers = Handlers0}, Session0, SOpts, Procedure, Opts0) ->
+    Opts = Opts0#{now => maps:get(now, Opts0, erlang:monotonic_time())},
+    Session1 = smf_aaa_session:session_merge(Session0, SOpts),
+    #{procedures := Procedures} = smf_aaa:get_application(AppId),
+    [SvcOpts = #{service := Service} | _] = smf_aaa_session:get_services(Procedure, Procedures),
+    Svc = smf_aaa:get_service(Service),
+    StepOpts = maps:merge(Opts, maps:merge(Svc, SvcOpts)),
+    Handler = maps:get(handler, Svc),
+    HState0 = maps:get(Handler, Handlers0, undefined),
+    Handler:ccr_initial_issue(Session1, StepOpts, HState0).
+
 run_pipeline(_Procedure, [], Session, Events, _Opts, Handlers) ->
     {ok, Session, Events, Handlers};
 run_pipeline(Procedure, [#{service := Service} = SvcOpts | Tail], Session0, Events0, Opts, Handlers0) ->

apps/smf_aaa/src/smf_aaa_gx.erl

@@ -24,6 +24,9 @@
 	 initialize_handler/1, initialize_service/2, invoke/6, handle_response/6]).
 -export([to_session/3, from_session/2]).
 
+%% async diameter bridge (see smf_aaa_pcf.erl / async_m)
+-export([ccr_initial_issue/3, ccr_update_issue/3, fold_cca/5]).
+
 -export([get_state_atom/1]).
 -ignore_xref([from_session/2, get_state_atom/1]).
 
@@ -105,18 +108,12 @@ invoke(_Service, init, Session, Events, _Opts, _State) ->
 
 invoke(_Service, {_, 'CCR-Initial'}, Session0, Events, Opts,
        #state{state = stopped} = State0) ->
-    State = inc_request_number(State0#state{state = started}),
-    Keys = ['InPackets', 'OutPackets', 'InOctets', 'OutOctets', 'Acct-Session-Time'],
-    Session = maps:without(Keys, Session0),
-    RecType = ?'DIAMETER_GX_CC-REQUEST-TYPE_INITIAL_REQUEST',
-    Request = make_CCR(RecType, Session, Opts, State),
+    {Request, Session, State} = ccr_initial_pre_send(Session0, Opts, State0),
     await_response(send_request(Request, Opts), Session, Events, State, Opts);
 
-invoke(_Service, {_, 'CCR-Update'}, Session, Events, Opts,
+invoke(_Service, {_, 'CCR-Update'}, Session0, Events, Opts,
        #state{state = started} = State0) ->
-    State = inc_request_number(State0),
-    RecType = ?'DIAMETER_GX_CC-REQUEST-TYPE_UPDATE_REQUEST',
-    Request = make_CCR(RecType, Session, Opts, State),
+    {Request, Session, State} = ccr_update_pre_send(Session0, Opts, State0),
     await_response(send_request(Request, Opts), Session, Events, State, Opts);
 
 invoke(_Service, {_, 'CCR-Terminate'}, Session, Events, Opts,
@@ -165,6 +162,59 @@ handle_response(Promise, Msg, Session, Events, Opts, #state{pending = Promise} =
 handle_response(_Promise, _Msg, Session, Events, _Opts, State) ->
     {ok, Session, Events, State}.
 
+%%%===================================================================
+%%% async_m diameter bridge
+%%%===================================================================
+
+%% pre-send half of CCR-Initial: #state{} transition + volume-key strip +
+%% make_CCR. Shared by invoke/6 (blocking path) and ccr_initial_issue/3
+%% (async_m path) so the two can't diverge.
+ccr_initial_pre_send(Session0, Opts, State0) ->
+    State = inc_request_number(State0#state{state = started}),
+    Keys = ['InPackets', 'OutPackets', 'InOctets', 'OutOctets', 'Acct-Session-Time'],
+    Session = maps:without(Keys, Session0),
+    RecType = ?'DIAMETER_GX_CC-REQUEST-TYPE_INITIAL_REQUEST',
+    Request = make_CCR(RecType, Session, Opts, State),
+    {Request, Session, State}.
+
+%% ccr_initial_issue/3 — the async_m counterpart of invoke/6's CCR-Initial
+%% clause: runs the same pre-send steps, but hands back the Promise instead
+%% of blocking on await_response/parking it in #state.pending.
+-spec ccr_initial_issue(map(), map(), #state{}) -> {reference(), map(), #state{}}.
+ccr_initial_issue(Session0, Opts, State0) ->
+    {Request, Session, State} = ccr_initial_pre_send(Session0, Opts, State0),
+    Promise = send_request(Request, Opts),
+    {Promise, Session, State}.
+
+%% pre-send half of CCR-Update: #state{} transition + make_CCR. Shared by
+%% invoke/6 (blocking path) and ccr_update_issue/3 (async_m path) so the
+%% two can't diverge.
+ccr_update_pre_send(Session, Opts, State0) ->
+    State = inc_request_number(State0),
+    RecType = ?'DIAMETER_GX_CC-REQUEST-TYPE_UPDATE_REQUEST',
+    Request = make_CCR(RecType, Session, Opts, State),
+    {Request, Session, State}.
+
+%% ccr_update_issue/3 — the async_m counterpart of invoke/6's CCR-Update
+%% clause: runs the same pre-send steps, but hands back the Promise instead
+%% of blocking on await_response/parking it in #state.pending.
+-spec ccr_update_issue(map(), map(), #state{}) -> {reference(), map(), #state{}}.
+ccr_update_issue(Session0, Opts, State0) ->
+    {Request, Session, State} = ccr_update_pre_send(Session0, Opts, State0),
+    Promise = send_request(Request, Opts),
+    {Promise, Session, State}.
+
+%% fold_cca/5 — the async_m counterpart of await_response's blocking
+%% handle_cca call: fold a raw CCA (or diameter error) into the session and
+%% events, re-wrapping the resulting #state{} into a #pcf_ctx{} exactly as
+%% the fire-and-forget handle_reply/4 path (smf_aaa_pcf.erl) does.
+fold_cca(Msg, Session, Events, Opts, State) ->
+    {Result, Session1, Events1, State1} = handle_cca(Msg, Session, Events, Opts, State),
+    {Result, Session1, Events1, wrap_ctx(State1)}.
+
+wrap_ctx(State) ->
+    #pcf_ctx{handlers = #{?MODULE => State}}.
+
 %%===================================================================
 %% DIAMETER handler callbacks
 %%===================================================================

apps/smf_aaa/src/smf_aaa_pcf.erl

@@ -6,6 +6,7 @@
 
 -export([new/2,
 	 ccr_initial/4, ccr_update/4, ccr_terminate/4,
+	 ccr_initial_issue/4, ccr_update_issue/4,
 	 terminate/3,
 	 handle_reply/4]).
 -ignore_xref([terminate/3]).
@@ -32,6 +33,20 @@ ccr_update(Ctx, Session, SOpts, Opts) ->
 ccr_terminate(Ctx, Session, SOpts, Opts) ->
     invoke(Ctx, Session, SOpts, {gx, 'CCR-Terminate'}, Opts).
 
+%% ccr_initial_issue/4 — the async_m counterpart of ccr_initial/4: runs the
+%% CCR-Initial pre-send half on the Gx handler and returns the Promise
+%% instead of blocking for the CCA. Session_at_send/State_at_send ride the
+%% async_m do-block across the await and get folded by smf_aaa_gx:fold_cca/5.
+ccr_initial_issue(Ctx, Session, SOpts, Opts) ->
+    issue(Ctx, Session, SOpts, {gx, 'CCR-Initial'}, Opts).
+
+%% ccr_update_issue/4 — the async_m counterpart of ccr_update/4: runs the
+%% CCR-Update pre-send half on the Gx handler and returns the Promise
+%% instead of blocking for the CCA. Session_at_send/State_at_send ride the
+%% async_m do-block across the await and get folded by smf_aaa_gx:fold_cca/5.
+ccr_update_issue(Ctx, Session, SOpts, Opts) ->
+    issue(Ctx, Session, SOpts, {gx, 'CCR-Update'}, Opts).
+
 terminate(Ctx, Session, Opts) ->
     invoke(Ctx, Session, #{'Termination-Cause' => error}, terminate, Opts).
 
@@ -60,6 +75,25 @@ invoke(#pcf_ctx{app_id = AppId, handlers = Handlers0} = Ctx,
 	    {Result, Ctx#pcf_ctx{handlers = Handlers1}, Session2, Events}
     end.
 
+%% issue/5 — single-handler counterpart of invoke/5 for the async_m
+%% pre-send path: no pipeline fan-out, no Events accumulator — just the one
+%% handler's CCR-Initial/CCR-Update pre-send + send_request, returning its
+%% Promise.
+issue(#pcf_ctx{app_id = AppId, handlers = Handlers0}, Session0, SOpts, Procedure, Opts0) ->
+    Opts = Opts0#{now => maps:get(now, Opts0, erlang:monotonic_time())},
+    Session1 = smf_aaa_session:session_merge(Session0, SOpts),
+    #{procedures := Procedures} = smf_aaa:get_application(AppId),
+    [SvcOpts = #{service := Service} | _] = smf_aaa_session:get_services(Procedure, Procedures),
+    Svc = smf_aaa:get_service(Service),
+    StepOpts = maps:merge(Opts, maps:merge(Svc, SvcOpts)),
+    Handler = maps:get(handler, Svc),
+    HState0 = maps:get(Handler, Handlers0, undefined),
+    IssueFun = issue_fun(Procedure),
+    Handler:IssueFun(Session1, StepOpts, HState0).
+
+issue_fun({_, 'CCR-Initial'}) -> ccr_initial_issue;
+issue_fun({_, 'CCR-Update'}) -> ccr_update_issue.
+
 run_pipeline(_Procedure, [], Session, Events, _Opts, Handlers) ->
     {ok, Session, Events, Handlers};
 run_pipeline(Procedure, [#{service := Service} = SvcOpts | Tail], Session0, Events0, Opts, Handlers0) ->

apps/smf_aaa/src/smf_aaa_ro.erl

@@ -23,6 +23,9 @@
 -export([validate_handler/1, validate_service/3, validate_procedure/5,
 	 initialize_handler/1, initialize_service/2, invoke/6, handle_response/6]).
 -export([to_session/3, from_session/2]).
+
+%% async diameter bridge (see smf_aaa_charging.erl / async_m)
+-export([ccr_initial_issue/3, fold_cca/5]).
 -export([get_state_atom/1]).
 -ignore_xref([from_session/2, get_state_atom/1]).
 
@@ -110,11 +113,7 @@ invoke(_Service, init, Session, Events, _Opts, _State) ->
 
 invoke(_Service, {_, 'CCR-Initial'}, Session0, Events, Opts,
        #state{state = stopped} = State0) ->
-    State = inc_request_number(State0#state{state = started}),
-    Keys = ['InPackets', 'OutPackets', 'InOctets', 'OutOctets', 'Acct-Session-Time'],
-    Session = maps:without(Keys, Session0),
-    RecType = ?'DIAMETER_RO_CC-REQUEST-TYPE_INITIAL_REQUEST',
-    Request = make_CCR(RecType, Session, Opts, State),
+    {Request, Session, State} = ccr_initial_pre_send(Session0, Opts, State0),
     await_response(send_request(Request, Opts), Session, Events, State, Opts);
 
 invoke(_Service, {_, 'CCR-Update'}, Session, Events, Opts,
@@ -191,6 +190,42 @@ handle_response(Promise, Msg, Session, Events, Opts, #state{pending = Promise} =
 handle_response(_Promise, _Msg, Session, Events, _Opts, State) ->
     {ok, Session, Events, State}.
 
+%%%===================================================================
+%%% async_m diameter bridge
+%%%===================================================================
+
+%% pre-send half of CCR-Initial: #state{} transition + volume-key strip +
+%% make_CCR (which itself builds the initial credit request). Shared by
+%% invoke/6 (blocking path) and ccr_initial_issue/3 (async_m path) so the
+%% two can't diverge.
+ccr_initial_pre_send(Session0, Opts, State0) ->
+    State = inc_request_number(State0#state{state = started}),
+    Keys = ['InPackets', 'OutPackets', 'InOctets', 'OutOctets', 'Acct-Session-Time'],
+    Session = maps:without(Keys, Session0),
+    RecType = ?'DIAMETER_RO_CC-REQUEST-TYPE_INITIAL_REQUEST',
+    Request = make_CCR(RecType, Session, Opts, State),
+    {Request, Session, State}.
+
+%% ccr_initial_issue/3 — the async_m counterpart of invoke/6's CCR-Initial
+%% clause: runs the same pre-send steps, but hands back the Promise instead
+%% of blocking on await_response/parking it in #state.pending.
+-spec ccr_initial_issue(map(), map(), #state{}) -> {reference(), map(), #state{}}.
+ccr_initial_issue(Session0, Opts, State0) ->
+    {Request, Session, State} = ccr_initial_pre_send(Session0, Opts, State0),
+    Promise = send_request(Request, Opts),
+    {Promise, Session, State}.
+
+%% fold_cca/5 — the async_m counterpart of await_response's blocking
+%% handle_cca call: fold a raw CCA (or diameter error) into the session and
+%% events, re-wrapping the resulting #state{} into a #charging_ctx{} exactly
+%% as the fire-and-forget handle_reply/4 path (smf_aaa_charging.erl) does.
+fold_cca(Msg, Session, Events, Opts, State) ->
+    {Result, Session1, Events1, State1} = handle_cca(Msg, Session, Events, Opts, State),
+    {Result, Session1, Events1, wrap_ctx(State1)}.
+
+wrap_ctx(State) ->
+    #charging_ctx{handlers = #{?MODULE => State}}.
+
 %%===================================================================
 %% DIAMETER handler callbacks
 %%===================================================================

apps/smf_aaa/src/smf_aaa_static.erl

@@ -13,6 +13,9 @@
 -export([validate_handler/1, validate_service/3, validate_procedure/5,
 	 initialize_handler/1, initialize_service/2, invoke/6, handle_response/6]).
 
+%% async_m mock API
+-export([ccr_initial_issue/3, ccr_update_issue/3]).
+
 -export([get_state_atom/1]).
 -ignore_xref([get_state_atom/1]).
 
@@ -58,6 +61,44 @@ invoke(_Service, _Procedure, Session, Events, Opts, State) ->
 handle_response(_Promise, _Msg, Session, Events, _Opts, State) ->
     {ok, Session, Events, State}.
 
+%% ccr_initial_issue/3 — async_m mock counterpart of invoke/6: resolves the
+%% canned answer exactly as invoke/6 does, but instead of processing it
+%% inline, hands back a Promise and delivers the raw CCA to Owner
+%% (self() at call time), the same way a real Gx handler's issue/3 would.
+-spec ccr_initial_issue(map(), map(), term()) -> {reference(), map(), term()}.
+ccr_initial_issue(Session, Opts, State) ->
+    issue(Session, Opts, State).
+
+%% ccr_update_issue/3 — see ccr_initial_issue/3.
+-spec ccr_update_issue(map(), map(), term()) -> {reference(), map(), term()}.
+ccr_update_issue(Session, Opts, State) ->
+    issue(Session, Opts, State).
+
+%% issue/3 — shared by ccr_initial_issue/3 and ccr_update_issue/3: resolve
+%% the canned answer's AVP map exactly as invoke/6 does (#{answers :=
+%% Answers, answer := Answer}), shape it as a raw ['CCA' | Avps] matching
+%% smf_aaa_gx:handle_cca/5's pattern, and deliver it to Owner.
+issue(Session, #{answers := Answers, answer := Answer} = Opts, State) ->
+    AVPs =
+	case Answers of
+	    #{Answer := #{avps := A}} -> A;
+	    _ -> #{}
+	end,
+    Promise = make_ref(),
+    Owner = self(),
+    Owner ! {'$reply', Promise, smf_aaa_gx, ['CCA' | to_raw_cca_avps(AVPs)], Opts},
+    {Promise, Session, State}.
+
+%% to_raw_cca_avps/1 — shape a canned answer's AVP map as the raw CCA AVPs
+%% smf_aaa_gx:handle_cca/5 pattern-matches: 'Result-Code' list-wrapped
+%% ([RC]); everything else (including repeated AVPs like
+%% Charging-Rule-Install/-Remove) passes through unchanged, since the
+%% static config already list-wraps those.
+to_raw_cca_avps(#{'Result-Code' := RC} = AVPs) when is_integer(RC) ->
+    AVPs#{'Result-Code' => [RC]};
+to_raw_cca_avps(AVPs) ->
+    AVPs.
+
 %%%===================================================================
 %%% Options Validation
 %%%===================================================================

apps/smf_aaa/test/smf_aaa_fold_SUITE.erl

@@ -0,0 +1,57 @@
+%% Copyright 2026, Next-NF
+
+%% Tests for smf_aaa_gx:fold_cca/5 and smf_aaa_ro:fold_cca/5 — the async_m
+%% counterpart of the blocking handle_cca fold done by await_response/5.
+
+-module(smf_aaa_fold_SUITE).
+
+-compile([export_all, nowarn_export_all]).
+
+-include_lib("common_test/include/ct.hrl").
+-include("../include/smf_aaa_session.hrl").
+
+all() ->
+    [gx_install, gx_timeout, gy_update_credits, gy_timeout].
+
+%%%===================================================================
+%%% Gx (smf_aaa_gx)
+%%%===================================================================
+
+gx_install(_Config) ->
+    {ok, Session0, [], State0} = smf_aaa_gx:invoke(gx, init, #{}, [], #{}, undefined),
+    CCA = ['CCA' | #{'Result-Code' => [2001],
+		      'Charging-Rule-Install' =>
+			  [#{'Charging-Rule-Base-Name' => [<<"m2m0001">>]}]}],
+    {ok, _Session1, Events, #pcf_ctx{}} =
+	smf_aaa_gx:fold_cca(CCA, Session0, [], #{}, State0),
+    {pcc, install, _} = lists:keyfind(pcc, 1, Events),
+    ok.
+
+gx_timeout(_Config) ->
+    {ok, Session0, [], State0} = smf_aaa_gx:invoke(gx, init, #{}, [], #{}, undefined),
+    {{error, timeout}, _Session1, _Events, #pcf_ctx{}} =
+	smf_aaa_gx:fold_cca({error, timeout}, Session0, [], #{}, State0),
+    ok.
+
+%%%===================================================================
+%%% Gy (smf_aaa_ro)
+%%%===================================================================
+
+gy_update_credits(_Config) ->
+    {ok, Session0a, [], State0} = smf_aaa_ro:invoke(gy, init, #{}, [], #{}, undefined),
+    Session0 = Session0a#{credits => #{1000 => empty}},
+    CCA = ['CCA' | #{'Result-Code' => 2001,
+		      'Multiple-Services-Credit-Control' =>
+			  [#{'Rating-Group' => [1000],
+			     'Granted-Service-Unit' => [#{'CC-Time' => [3600]}],
+			     'Result-Code' => [2001]}]}],
+    {ok, _Session1, Events, #charging_ctx{}} =
+	smf_aaa_ro:fold_cca(CCA, Session0, [], #{}, State0),
+    {update_credits, _} = lists:keyfind(update_credits, 1, Events),
+    ok.
+
+gy_timeout(_Config) ->
+    {ok, Session0, [], State0} = smf_aaa_ro:invoke(gy, init, #{}, [], #{}, undefined),
+    {{error, timeout}, _Session1, _Events, #charging_ctx{}} =
+	smf_aaa_ro:fold_cca({error, timeout}, Session0, [], #{}, State0),
+    ok.