- encapsulate the logging and ar_mining_stats for the solution lifecycle

- get rid of the solution evnts and use direct casts
ArweaveTeam · Jul 22, 2024 · 1c7ad6a · 1c7ad6a
1 parent e7360eb
commit 1c7ad6a
Show file tree

Hide file tree

Showing 12 changed files with 297 additions and 277 deletions.
diff --git a/apps/arweave/include/ar_mining.hrl b/apps/arweave/include/ar_mining.hrl
@@ -54,4 +54,10 @@
 	steps = []
 }).
 
+%% @doc Solution validation response.
+-record(solution_response, {
+	indep_hash = <<>>,
+	status = <<>>
+}).
+
 -endif.
diff --git a/apps/arweave/include/ar_pool.hrl b/apps/arweave/include/ar_pool.hrl
@@ -34,12 +34,6 @@
 	partition_upper_bound = 0
 }).
 
-%% @doc Partial solution validation response.
--record(partial_solution_response, {
-	indep_hash = <<>>,
-	status = <<>>
-}).
-
 %% @doc A set of coordinated mining jobs provided by the pool.
 %%
 %% Miners fetch and submit pool CM jobs via the same POST /pool_cm_jobs endpoint.

diff --git a/apps/arweave/src/ar_events_sup.erl b/apps/arweave/src/ar_events_sup.erl
@@ -55,8 +55,6 @@ init([]) ->
 		?CHILD(ar_events, chunk_storage, worker),
 		%% Events: add_range, remove_range, global_remove_range, cut, global_cut.
 		?CHILD(ar_events, sync_record, worker),
-		%% Events: rejected, stale, partial, accepted.
-		?CHILD(ar_events, solution, worker),
 		%% Used for the testing purposes.
 		?CHILD(ar_events, testing, worker)
 	]}}.
diff --git a/apps/arweave/src/ar_http_iface_client.erl b/apps/arweave/src/ar_http_iface_client.erl
@@ -619,7 +619,7 @@ post_partial_solution(Peer, Solution) ->
 				ar_serialize:jsonify(ar_serialize:solution_to_json_struct(Solution))
 		end,
 	Req = build_cm_or_pool_request(post, Peer, "/partial_solution", Payload),
-	handle_post_partial_solution_response(ar_http:req(Req#{
+	handle_post_solution_response(ar_http:req(Req#{
 		timeout => 20 * 1000,
 		connect_timeout => 5 * 1000
 	})).
@@ -704,14 +704,14 @@ handle_post_pool_cm_jobs_response({ok, {{<<"200">>, _}, _, _, _, _}}) ->
 handle_post_pool_cm_jobs_response(Reply) ->
 	{error, Reply}.
 
-handle_post_partial_solution_response({ok, {{<<"200">>, _}, _, Body, _, _}}) ->
+handle_post_solution_response({ok, {{<<"200">>, _}, _, Body, _, _}}) ->
 	case catch jiffy:decode(Body, [return_maps]) of
 		{'EXIT', _} ->
 			{error, invalid_json};
 		Response ->
 			{ok, Response}
 	end;
-handle_post_partial_solution_response(Reply) ->
+handle_post_solution_response(Reply) ->
 	{error, Reply}.
 
 handle_get_jobs_response({ok, {{<<"200">>, _}, _, Body, _, _}}) ->

diff --git a/apps/arweave/src/ar_http_iface_middleware.erl b/apps/arweave/src/ar_http_iface_middleware.erl
@@ -2558,8 +2558,9 @@ handle_post_partial_solution_pool_server(Req, Pid) ->
 				{'EXIT', _} ->
 					{400, #{}, jiffy:encode(#{ error => invalid_json }), Req2};
 				Solution ->
-					Response = ar_pool:process_partial_solution(Solution),
-					JSON = ar_serialize:partial_solution_response_to_json_struct(Response),
+					ar_mining_router:received_solution(Solution, []),
+					Response = ar_mining_router:route_solution(Solution),
+					JSON = ar_serialize:solution_response_to_json_struct(Response),
 					{200, #{}, ar_serialize:jsonify(JSON), Req2}
 			end;
 		{error, body_size_too_large} ->
@@ -2571,8 +2572,15 @@ handle_post_partial_solution_pool_server(Req, Pid) ->
 handle_post_partial_solution_cm_exit_peer_pool_client(Req, Pid) ->
 	case read_complete_body(Req, Pid) of
 		{ok, Body, Req2} ->
-			ar_pool:post_partial_solution(Body),
-			{200, #{}, jiffy:encode(#{}), Req2};
+			case catch ar_serialize:json_map_to_solution(
+				jiffy:decode(Body, [return_maps])) of
+				{'EXIT', _} ->
+					{400, #{}, jiffy:encode(#{ error => invalid_json }), Req2};
+				Solution ->
+					ar_mining_router:received_solution(Solution, []),
+					ar_pool:post_partial_solution(Solution),
+					{200, #{}, jiffy:encode(#{}), Req2}
+			end;
 		{error, body_size_too_large} ->
 			{413, #{}, <<"Payload too large">>, Req};
 		{error, timeout} ->
@@ -3261,6 +3269,7 @@ handle_mining_h2(Req, Pid) ->
 									{200, #{}, <<>>, Req2};
 								_ ->
 									ar_mining_stats:h2_received_from_peer(Peer),
+									ar_mining_router:received_solution(Candidate, h2),
 									ar_mining_router:prepare_solution(Candidate),
 									{200, #{}, <<>>, Req}
 							end
@@ -3276,24 +3285,19 @@ handle_mining_cm_publish(Req, Pid) ->
 	Peer = ar_http_util:arweave_peer(Req),
 	case read_complete_body(Req, Pid) of
 		{ok, Body, Req2} ->
-			case ar_serialize:json_decode(Body, [return_maps]) of
-				{ok, JSON} ->
-					case catch ar_serialize:json_map_to_solution(JSON) of
-						{'EXIT', _} ->
-							{400, #{}, jiffy:encode(#{ error => invalid_json }), Req2};
-						Solution ->
-							ar:console("Block candidate ~p from ~p ~n", [
-								ar_util:encode(Solution#mining_solution.solution_hash),
-								ar_util:format_peer(Peer)]),
-							?LOG_INFO("Block candidate ~p from ~p ~n", [
-								ar_util:encode(Solution#mining_solution.solution_hash),
-								ar_util:format_peer(Peer)]),
-							ar_mining_router:route_solution(Solution),
-							{200, #{}, <<>>, Req}
-					end;
-				{error, _} ->
-					{400, #{}, jiffy:encode(#{ error => invalid_json }), Req2}
+			case catch ar_serialize:json_map_to_solution(
+					jiffy:decode(Body, [return_maps])) of
+				{'EXIT', _} ->
+					{400, #{}, jiffy:encode(#{ error => invalid_json }), Req2};
+				Solution ->
+					ar_mining_router:received_solution(Solution,
+						[{peer, ar_util:format_peer(Peer)}]),
+					Response = ar_mining_router:route_solution(Solution),
+					JSON = ar_serialize:solution_response_to_json_struct(Response),
+					{200, #{}, ar_serialize:jsonify(JSON), Req2}
 			end;
 		{error, body_size_too_large} ->
-			{413, #{}, <<"Payload too large">>, Req}
+			{413, #{}, <<"Payload too large">>, Req};
+		{error, timeout} ->
+			{500, #{}, <<"Handler timeout">>, Req}
 	end.
diff --git a/apps/arweave/src/ar_mining_router.erl b/apps/arweave/src/ar_mining_router.erl
@@ -2,7 +2,12 @@
 
 -behaviour(gen_server).
 
--export([start_link/0, prepare_solution/1, route_solution/1, route_h1/2, route_h2/1]).
+-export([start_link/0,
+	prepare_solution/1, route_solution/1,
+	found_solution/3, received_solution/2, received_solution/3,
+	reject_solution/3, reject_solution/4,
+	accept_solution/1, accept_solution/2, accept_block_solution/2, accept_block_solution/3,
+	route_h1/2, route_h2/1]).
 
 -export([init/1, handle_cast/2, handle_call/3, handle_info/2, terminate/2]).
 
@@ -13,6 +18,7 @@
 -include_lib("stdlib/include/ms_transform.hrl").
 
 -record(state, {
+	request_pid_by_ref = maps:new()
 }).
 
 %%%===================================================================
@@ -30,7 +36,116 @@ prepare_solution(Candidate) ->
 
 route_solution(Solution) ->
 	{ok, Config} = application:get_env(arweave, config),
-	route_solution(Config#config.cm_exit_peer, Config#config.is_pool_client, Solution).
+	gen_server:call(?MODULE, {route_solution, Config, Solution}).
+
+found_solution(#mining_candidate{} = Candidate, WhichHash, ExtraLogs) ->
+	#mining_candidate{
+		mining_address = MiningAddress,
+		nonce_limiter_output = NonceLimiterOutput,
+		seed = Seed, next_seed = NextSeed, 
+		start_interval_number = StartIntervalNumber, step_number = StepNumber } = Candidate,
+
+	{Hash, PartitionNumber} = case WhichHash of 
+		h1 ->
+			{Candidate#mining_candidate.h1, Candidate#mining_candidate.partition_number};
+		h2 ->
+			{Candidate#mining_candidate.h2, Candidate#mining_candidate.partition_number2}
+	end,
+	?LOG_INFO([
+		{event, solution_lifecycle},
+		{status, found},
+		{hash, ar_util:safe_encode(Hash)},
+		{mining_address, ar_util:safe_encode(MiningAddress)},
+		{partition, PartitionNumber},
+		{seed, Seed},
+		{next_seed, NextSeed},
+		{start_interval_number, StartIntervalNumber},
+		{step_number, StepNumber},
+		{nonce_limiter_output, ar_util:safe_encode(NonceLimiterOutput)}] ++
+		ExtraLogs),		
+	ar_mining_stats:solution(found).
+
+received_solution(#mining_candidate{} = Candidate, WhichHash, ExtraLogs) ->
+	#mining_candidate{
+		mining_address = MiningAddress,
+		nonce_limiter_output = NonceLimiterOutput,
+		seed = Seed, next_seed = NextSeed, 
+		start_interval_number = StartIntervalNumber, step_number = StepNumber } = Candidate,
+
+	{Hash, PartitionNumber} = case WhichHash of 
+		h1 ->
+			{Candidate#mining_candidate.h1, Candidate#mining_candidate.partition_number};
+		h2 ->
+			{Candidate#mining_candidate.h2, Candidate#mining_candidate.partition_number2}
+	end,
+	received_solution(Hash, MiningAddress, PartitionNumber, Seed, NextSeed,
+		StartIntervalNumber, StepNumber, NonceLimiterOutput, ExtraLogs).
+
+received_solution(Solution, ExtraLogs) ->
+	#mining_solution{
+		mining_address = MiningAddress,
+		nonce_limiter_output = NonceLimiterOutput, partition_number = PartitionNumber,
+		solution_hash = H, seed = Seed, next_seed = NextSeed, 
+		start_interval_number = StartIntervalNumber, step_number = StepNumber } = Solution,
+	received_solution(H, MiningAddress, PartitionNumber, Seed, NextSeed, StartIntervalNumber,
+		StepNumber, NonceLimiterOutput, ExtraLogs).
+
+reject_solution(Solution, Reason, ExtraLogs) ->
+	#mining_solution{
+		mining_address = MiningAddress,
+		nonce_limiter_output = NonceLimiterOutput, partition_number = PartitionNumber,
+		recall_byte1 = RecallByte1, recall_byte2 = RecallByte2,
+		solution_hash = H, seed = Seed, next_seed = NextSeed, 
+		start_interval_number = StartIntervalNumber, step_number = StepNumber } = Solution,
+	ar:console("WARNING: solution was rejected. Check logs for more details~n"),
+	?LOG_WARNING([
+		{event, solution_lifecycle},
+		{status, rejected},
+		{reason, Reason},
+		{hash, ar_util:safe_encode(H)},
+		{mining_address, ar_util:safe_encode(MiningAddress)},
+		{partition, PartitionNumber},
+		{recall_byte1, RecallByte1},
+		{recall_byte2, RecallByte2},
+		{seed, Seed},
+		{next_seed, NextSeed},
+		{start_interval_number, StartIntervalNumber},
+		{step_number, StepNumber},
+		{nonce_limiter_output, ar_util:safe_encode(NonceLimiterOutput)}] ++
+		ExtraLogs),		
+	ar_mining_stats:solution(rejected).
+
+reject_solution(Solution, Reason, ExtraLogs, Ref) ->
+	reject_solution(Solution, Reason, ExtraLogs),
+	Status = iolist_to_binary([<<"rejected_">>, atom_to_binary(Reason)]),
+	gen_server:cast(?MODULE, {solution_response, Status, <<>>, Ref}).
+
+accept_solution(Solution) ->
+	#mining_solution{ mining_address = MiningAddress, solution_hash = H } = Solution,
+	?LOG_WARNING([
+		{event, solution_lifecycle},
+		{status, accepted},
+		{hash, ar_util:safe_encode(H)},
+		{mining_address, ar_util:safe_encode(MiningAddress)}]),		
+	ar_mining_stats:solution(accepted).
+
+accept_solution(Solution, Ref) ->
+	accept_solution(Solution),
+	gen_server:cast(?MODULE, {solution_response, <<"accepted">>, <<>>, Ref}).
+
+accept_block_solution(Solution, BlockH) ->
+	#mining_solution{ mining_address = MiningAddress, solution_hash = H } = Solution,
+	?LOG_WARNING([
+		{event, solution_lifecycle},
+		{status, accepted},
+		{hash, ar_util:safe_encode(H)},
+		{mining_address, ar_util:safe_encode(MiningAddress)},
+		{block_hash, ar_util:safe_encode(BlockH)}]),		
+	ar_mining_stats:solution(accepted).
+
+accept_block_solution(Solution, BlockH, Ref) ->
+	accept_block_solution(Solution, BlockH),
+	gen_server:cast(?MODULE, {solution_response, <<"accepted_block">>, BlockH, Ref}).
 
 route_h1(Candidate, DiffPair) ->
 	{ok, Config} = application:get_env(arweave, config),
@@ -53,10 +168,27 @@ route_h2(Candidate) ->
 init([]) ->
 	{ok, #state{}}.
 
+handle_call({route_solution, Config, Solution}, From, State) ->
+	#state{ request_pid_by_ref = Map } = State,
+	Ref = make_ref(),
+	case route_solution(Config, Solution, Ref) of
+		noreply ->
+			{noreply, State#state{ request_pid_by_ref = maps:put(Ref, From, Map) }};
+		Reply ->
+			{reply, Reply, State}
+	end;
+
 handle_call(Request, _From, State) ->
 	?LOG_WARNING([{event, unhandled_call}, {module, ?MODULE}, {request, Request}]),
 	{reply, ok, State}.
 
+handle_cast({solution_response, Status, BlockH, {pool, Ref}}, State) ->
+	#state{ request_pid_by_ref = Map } = State,
+	PID = maps:get(Ref, Map),
+	gen_server:reply(PID,
+			#solution_response{ indep_hash = BlockH, status = Status }),
+	{noreply, State#state{ request_pid_by_ref = maps:remove(Ref, Map) }};
+
 handle_cast(Cast, State) ->
 	?LOG_WARNING([{event, unhandled_cast}, {module, ?MODULE}, {cast, Cast}]),
 	{noreply, State}.
@@ -72,13 +204,31 @@ terminate(_Reason, _State) ->
 %%% Private functions.
 %%%===================================================================
 
-route_solution(not_set, true, Solution) ->
+received_solution(Hash, MiningAddress, PartitionNumber, Seed, NextSeed,
+		StartIntervalNumber, StepNumber, NonceLimiterOutput, ExtraLogs) ->
+	?LOG_INFO([
+		{event, solution_lifecycle},
+		{status, received},
+		{hash, ar_util:safe_encode(Hash)},
+		{mining_address, ar_util:safe_encode(MiningAddress)},
+		{partition, PartitionNumber},
+		{seed, Seed},
+		{next_seed, NextSeed},
+		{start_interval_number, StartIntervalNumber},
+		{step_number, StepNumber},
+		{nonce_limiter_output, ar_util:safe_encode(NonceLimiterOutput)}] ++
+		ExtraLogs),
+	ar_mining_stats:solution(received).
+
+route_solution(#config{ is_pool_server = true }, Solution, Ref) ->
+	ar_pool:process_partial_solution(Solution, Ref);
+route_solution(#config{ cm_exit_peer = not_set, is_pool_client = true }, Solution, Ref) ->
 	%% When posting a partial solution the pool client will skip many of the validation steps
 	%% that are normally performed before sharing a solution.
 	ar_pool:post_partial_solution(Solution);
-route_solution(not_set, _IsPoolClient, Solution) ->
+route_solution(#config{ cm_exit_peer = not_set, is_pool_client = false }, Solution, Ref) ->
 	ar_mining_server:validate_solution(Solution);
-route_solution(ExitPeer, true, Solution) ->
+route_solution(#config{ cm_exit_peer = ExitPeer, is_pool_client = true }, Solution, Ref) ->
 	case ar_http_iface_client:post_partial_solution(ExitPeer, Solution) of
 		{ok, _} ->
 			ok;
@@ -88,7 +238,7 @@ route_solution(ExitPeer, true, Solution) ->
 			ar:console("We found a partial solution but failed to reach the exit node, "
 					"error: ~p.", [io_lib:format("~p", [Reason])])
 	end;
-route_solution(ExitPeer, _IsPoolClient, Solution) ->
+route_solution(#config{ cm_exit_peer = ExitPeer, is_pool_client = false }, Solution, Ref) ->
 	case ar_http_iface_client:cm_publish_send(ExitPeer, Solution) of
 		{ok, _} ->
 			ok;
@@ -99,4 +249,4 @@ route_solution(ExitPeer, _IsPoolClient, Solution) ->
 			ar:console("We found a solution but failed to reach the exit node, "
 					"error: ~p.", [io_lib:format("~p", [Reason])]),
 			ar_mining_stats:solution(rejected)
-	end.
+	end.