feat(metering): meter request and response bytes

src/preloaded/payment/dev_metering.erl

@@ -10,16 +10,19 @@
 %%%
 %%% Calls to `consume/3' outside an active session are no-ops, so callers do
 %%% not need to check whether metering is enabled. Resource names are normalized
-%%% keys, such as `arweave-bytes' and `beam-reductions'. The operator sets
-%%% `metering-rates' in the node message as a map of resource name to AO token
-%%% units per resource unit.
+%%% keys, such as `arweave-bytes' and `beam-reductions'. The device also meters
+%%% `request-bytes' and `response-bytes' from the bodies P4 supplies. The
+%%% operator sets `metering-rates' in the node message as a map of resource name
+%%% to AO token units per resource unit.
 -module(dev_metering).
 -export([info/1, estimate/3, price/3, is_active/0, consume/3]).
 
 -include_lib("eunit/include/eunit.hrl").
 
 -define(METERING_KEY, {dev_metering, state}).
 -define(BEAM_REDUCTIONS, <<"beam-reductions">>).
+-define(REQUEST_BYTES, <<"request-bytes">>).
+-define(RESPONSE_BYTES, <<"response-bytes">>).
 
 %% @doc Device API information.
 info(_) ->
@@ -32,7 +35,7 @@ info(_) ->
     }.
 
 %% @doc Start a metering session for the request.
-estimate(_Base, _EstimateReq, _Opts) ->
+estimate(_Base, EstimateReq, Opts) ->
     {reductions, Reductions} = erlang:process_info(self(), reductions),
     erlang:put(
         ?METERING_KEY,
@@ -41,10 +44,12 @@ estimate(_Base, _EstimateReq, _Opts) ->
             meters => #{}
         }
     ),
+    consume(?REQUEST_BYTES, body_size(EstimateReq, Opts), Opts),
     {ok, 0}.
 
 %% @doc Close the metering session and calculate the final AO token price.
-price(_Base, _PriceReq, Opts) ->
+price(_Base, PriceReq, Opts) ->
+    consume(?RESPONSE_BYTES, body_size(PriceReq, Opts), Opts),
     Rates = hb_opts:get(<<"metering-rates">>, #{}, Opts),
     Price =
         maps:fold(
@@ -91,6 +96,41 @@ consume(Resource, Amount, _Opts) ->
             end
     end.
 
+%% @doc Size the body P4 hands to the pricing device.
+%%
+%% `estimate' receives the inbound request and `price' the result, so a body
+%% measured here covers whatever the node carried, including payload a device
+%% relayed on another node's behalf. A `bundle: true' commitment includes linked
+%% content in the body, so those links are loaded before sizing. Otherwise links
+%% are sized as links.
+%%
+%% The size is of the ETF encoding, not the wire form. It is stable and
+%% monotone in payload size, but a payer cannot reproduce it from the bytes it
+%% sent, and the ratio to wire size varies by request shape.
+body_size(Req, Opts) when is_map(Req) ->
+    case hb_maps:get(<<"body">>, Req, not_found, Opts) of
+        not_found -> 0;
+        Body -> term_size(maybe_load_body(Body, Opts))
+    end;
+body_size(_Req, _Opts) ->
+    0.
+
+%% @doc Load linked content included by any bundle commitment.
+maybe_load_body(Body, Opts) ->
+    Commitments =
+        hb_message:commitments(#{ <<"bundle">> => <<"true">> }, Body, Opts),
+    case map_size(Commitments) of
+        0 -> Body;
+        _ -> hb_cache:ensure_all_loaded(Body, Opts)
+    end.
+
+%% @doc Return the encoded size of a body term.
+term_size(Bin) when is_binary(Bin) -> byte_size(Bin);
+term_size(Term) ->
+    try erlang:external_size(Term)
+    catch _:_ -> 0
+    end.
+
 %% @doc Add the process reductions delta to the active metering state.
 meter_reductions(undefined) ->
     #{ meters => #{} };
@@ -252,3 +292,124 @@ p4_response_charge_test() ->
     after
         hb_mock_server:stop(ServerHandle)
     end.
+
+%% @doc Request and response body meters contribute to the same P4 session.
+request_and_response_bytes_test() ->
+    Opts = #{
+        <<"store">> => hb_test_utils:test_store(),
+        <<"metering-rates">> => #{
+            ?REQUEST_BYTES => 1,
+            ?RESPONSE_BYTES => 1,
+            ?BEAM_REDUCTIONS => 0
+        }
+    },
+    Metering = #{ <<"device">> => <<"metering@1.0">> },
+    Request = #{ <<"body">> => binary:copy(<<"q">>, 1000) },
+    Response = #{ <<"body">> => binary:copy(<<"r">>, 50000) },
+    {ok, 0} =
+        hb_ao:resolve(
+            Metering,
+            #{ <<"path">> => <<"estimate">>, <<"body">> => Request },
+            Opts
+        ),
+    {ok, Price} =
+        hb_ao:resolve(
+            Metering,
+            #{ <<"path">> => <<"price">>, <<"body">> => Response },
+            Opts
+        ),
+    % Both bodies are charged, so the total tracks the larger one and cannot be
+    % explained by the request alone.
+    ?assert(Price > 50000),
+    ?assert(Price < 60000).
+
+%% @doc Response byte charges grow with the loaded payload.
+response_bytes_scale_with_payload_test() ->
+    Opts = #{
+        <<"store">> => hb_test_utils:test_store(),
+        <<"metering-rates">> => #{ ?RESPONSE_BYTES => 1 }
+    },
+    Metering = #{ <<"device">> => <<"metering@1.0">> },
+    Price =
+        fun(Bytes) ->
+            {ok, 0} =
+                hb_ao:resolve(Metering, #{ <<"path">> => <<"estimate">> }, Opts),
+            {ok, P} =
+                hb_ao:resolve(
+                    Metering,
+                    #{
+                        <<"path">> => <<"price">>,
+                        <<"body">> => #{ <<"body">> => binary:copy(<<"x">>, Bytes) }
+                    },
+                    Opts
+                ),
+            P
+        end,
+    Small = Price(1000),
+    Large = Price(1000000),
+    % A flat charge means the body is not reaching the meter.
+    ?assert(Large > Small * 100).
+
+%% @doc Linked bodies are loaded only when a commitment includes the bundle.
+linked_body_metering_respects_bundle_commitment_test() ->
+    Opts = #{
+        <<"store">> => hb_test_utils:test_store(),
+        <<"priv-wallet">> => ar_wallet:new(),
+        <<"metering-rates">> => #{ ?RESPONSE_BYTES => 1 }
+    },
+    Metering = #{ <<"device">> => <<"metering@1.0">> },
+    Payload = binary:copy(<<"x">>, 1000000),
+    Price =
+        fun(Bundle) ->
+            Response =
+                hb_message:commit(
+                    #{ <<"body">> => #{ <<"payload">> => Payload } },
+                    Opts,
+                    #{
+                        <<"commitment-device">> => <<"httpsig@1.0">>,
+                        <<"bundle">> => Bundle
+                    }
+                ),
+            {ok, _} = hb_cache:write(Response, Opts),
+            {ok, LinkedResponse} =
+                hb_cache:read(hb_message:id(Response, all, Opts), Opts),
+            {ok, 0} =
+                hb_ao:resolve(Metering, #{ <<"path">> => <<"estimate">> }, Opts),
+            {ok, Result} =
+                hb_ao:resolve(
+                    Metering,
+                    #{ <<"path">> => <<"price">>, <<"body">> => LinkedResponse },
+                    Opts
+                ),
+            Result
+        end,
+    LinkedPrice = Price(false),
+    BundledPrice = Price(true),
+    ?assert(LinkedPrice < 10000),
+    ?assert(BundledPrice > LinkedPrice * 100).
+
+%% @doc Byte meters remain inert when the operator configures no rate.
+unpriced_resources_cost_nothing_test() ->
+    Opts = #{
+        <<"store">> => hb_test_utils:test_store(),
+        <<"metering-rates">> => #{}
+    },
+    Metering = #{ <<"device">> => <<"metering@1.0">> },
+    {ok, 0} =
+        hb_ao:resolve(
+            Metering,
+            #{
+                <<"path">> => <<"estimate">>,
+                <<"body">> => #{ <<"body">> => binary:copy(<<"x">>, 100000) }
+            },
+            Opts
+        ),
+    {ok, 0} =
+        hb_ao:resolve(
+            Metering,
+            #{
+                <<"path">> => <<"price">>,
+                <<"body">> => #{ <<"body">> => binary:copy(<<"x">>, 100000) }
+            },
+            Opts
+        ).