Refactor GetUnspentOutput message

Add new chain_sync_date field for conflict resolver and update offset

lib/archethic.ex

@@ -413,10 +413,10 @@ defmodule Archethic do
   """
   @spec get_unspent_outputs(
           address :: Crypto.prepended_hash(),
-          paging_offset :: non_neg_integer(),
+          paging_offset :: Crypto.sha256() | nil,
           limit :: non_neg_integer()
         ) :: list(UnspentOutput.t())
-  def get_unspent_outputs(address, paging_offset \\ 0, limit \\ 0) do
+  def get_unspent_outputs(address, paging_offset \\ nil, limit \\ 0) do
     previous_summary_time = BeaconChain.previous_summary_time(DateTime.utc_now())
 
     nodes =

lib/archethic/crypto.ex

@@ -65,6 +65,8 @@ defmodule Archethic.Crypto do
   """
   @type prepended_hash :: <<_::16, _::_*8>>
 
+  @type sha256 :: <<_::32>>
+
   @typedoc """
   Binary representing a key prepend by two bytes:
   - to identify the elliptic curve for a key

lib/archethic/p2p/message/get_unspent_outputs.ex

@@ -3,11 +3,12 @@ defmodule Archethic.P2P.Message.GetUnspentOutputs do
   Represents a message to request the list of unspent outputs from a transaction
   """
   @enforce_keys [:address]
-  defstruct [:address, offset: 0, limit: 0]
+  defstruct [:address, offset: nil, limit: 0]
 
   alias Archethic.Crypto
   alias Archethic.P2P.Message.UnspentOutputList
 
+  alias Archethic.TransactionChain
   alias Archethic.TransactionChain.Transaction.ValidationStamp.LedgerOperations.UnspentOutput
 
   alias Archethic.TransactionChain.Transaction.ValidationStamp.LedgerOperations.VersionedUnspentOutput
@@ -23,43 +24,78 @@ defmodule Archethic.P2P.Message.GetUnspentOutputs do
              )
 
   @type t :: %__MODULE__{
-          address: Crypto.versioned_hash(),
-          offset: non_neg_integer(),
+          address: Crypto.prepended_hash(),
+          offset: Crypto.sha256() | nil,
           limit: non_neg_integer()
         }
 
   @spec process(__MODULE__.t(), Crypto.key()) :: UnspentOutputList.t()
   def process(%__MODULE__{address: genesis_address, offset: offset, limit: limit}, _) do
-    {utxos, more?, offset} =
+    sorted_utxos =
       genesis_address
       |> UTXO.stream_unspent_outputs()
       |> Enum.sort_by(fn %VersionedUnspentOutput{
                            unspent_output: %UnspentOutput{timestamp: timestamp}
                          } ->
         if is_nil(timestamp), do: DateTime.from_unix!(0), else: timestamp
       end)
-      |> Utils.limit_list(limit, offset, @threshold, fn utxo ->
-        utxo
-        |> VersionedUnspentOutput.serialize()
-        |> byte_size
-      end)
 
-    %UnspentOutputList{
-      unspent_outputs: utxos,
-      offset: offset,
-      more?: more?
-    }
+    case get_numerical_offset(sorted_utxos, offset) do
+      nil ->
+        %UnspentOutputList{
+          unspent_outputs: [],
+          offset: nil,
+          more?: false,
+          last_chain_sync_date: DateTime.from_unix!(0, :millisecond)
+        }
+
+      offset ->
+        {utxos, more?, _offset} =
+          Utils.limit_list(sorted_utxos, limit, offset, @threshold, fn utxo ->
+            utxo |> VersionedUnspentOutput.serialize() |> byte_size
+          end)
+
+        offset =
+          if Enum.empty?(utxos),
+            do: nil,
+            else: utxos |> List.last() |> VersionedUnspentOutput.hash()
+
+        {_, last_chain_sync_date} = TransactionChain.get_last_address(genesis_address)
+
+        %UnspentOutputList{
+          unspent_outputs: utxos,
+          offset: offset,
+          more?: more?,
+          last_chain_sync_date: last_chain_sync_date
+        }
+    end
+  end
+
+  defp get_numerical_offset(_utxos, nil), do: 0
+
+  defp get_numerical_offset(utxos, offset) do
+    case Enum.find_index(utxos, &(VersionedUnspentOutput.hash(&1) == offset)) do
+      nil -> nil
+      index -> index + 1
+    end
   end
 
   @spec serialize(t()) :: bitstring()
   def serialize(%__MODULE__{address: tx_address, offset: offset, limit: limit}) do
-    <<tx_address::binary, VarInt.from_value(offset)::binary, VarInt.from_value(limit)::binary>>
+    offset_bin = if is_nil(offset), do: <<0::1>>, else: <<1::1, offset::binary>>
+    <<tx_address::binary, offset_bin::bitstring, VarInt.from_value(limit)::binary>>
   end
 
   @spec deserialize(bitstring()) :: {t(), bitstring()}
   def deserialize(<<rest::bitstring>>) do
     {address, rest} = Utils.deserialize_address(rest)
-    {offset, rest} = VarInt.get_value(rest)
+
+    {offset, rest} =
+      case rest do
+        <<0::1, rest::bitstring>> -> {nil, rest}
+        <<1::1, offset::binary-size(32), rest::bitstring>> -> {offset, rest}
+      end
+
     {limit, rest} = VarInt.get_value(rest)
 
     {%__MODULE__{address: address, offset: offset, limit: limit}, rest}

lib/archethic/p2p/message/unspent_output_list.ex

@@ -2,19 +2,28 @@ defmodule Archethic.P2P.Message.UnspentOutputList do
   @moduledoc """
   Represents a message with a list of unspent outputs
   """
-  defstruct unspent_outputs: [], more?: false, offset: 0
+  defstruct [:last_chain_sync_date, unspent_outputs: [], more?: false, offset: nil]
+
+  alias Archethic.Crypto
 
   alias Archethic.TransactionChain.Transaction.ValidationStamp.LedgerOperations.VersionedUnspentOutput
+
   alias Archethic.Utils.VarInt
 
   @type t :: %__MODULE__{
           unspent_outputs: list(VersionedUnspentOutput.t()),
           more?: boolean(),
-          offset: non_neg_integer()
+          offset: Crypto.sha256() | nil,
+          last_chain_sync_date: DateTime.t()
         }
 
   @spec serialize(t()) :: bitstring()
-  def serialize(%__MODULE__{unspent_outputs: unspent_outputs, more?: more?, offset: offset}) do
+  def serialize(%__MODULE__{
+        unspent_outputs: unspent_outputs,
+        more?: more?,
+        offset: offset,
+        last_chain_sync_date: last_chain_sync_date
+      }) do
     unspent_outputs_bin =
       unspent_outputs
       |> Stream.map(&VersionedUnspentOutput.serialize/1)
@@ -28,8 +37,10 @@ defmodule Archethic.P2P.Message.UnspentOutputList do
 
     more_bit = if more?, do: 1, else: 0
 
+    offset_bin = if is_nil(offset), do: <<0::1>>, else: <<1::1, offset::binary>>
+
     <<encoded_unspent_outputs_length::binary, unspent_outputs_bin::bitstring, more_bit::1,
-      VarInt.from_value(offset)::binary>>
+      offset_bin::bitstring, DateTime.to_unix(last_chain_sync_date, :millisecond)::64>>
   end
 
   @spec deserialize(bitstring()) :: {t(), bitstring}
@@ -41,9 +52,18 @@ defmodule Archethic.P2P.Message.UnspentOutputList do
 
     more? = more_bit == 1
 
-    {offset, rest} = VarInt.get_value(rest)
+    {offset, <<last_chain_sync_date::64, rest::bitstring>>} =
+      case rest do
+        <<0::1, rest::bitstring>> -> {nil, rest}
+        <<1::1, offset::binary-size(32), rest::bitstring>> -> {offset, rest}
+      end
 
-    {%__MODULE__{unspent_outputs: unspent_outputs, more?: more?, offset: offset}, rest}
+    {%__MODULE__{
+       unspent_outputs: unspent_outputs,
+       more?: more?,
+       offset: offset,
+       last_chain_sync_date: DateTime.from_unix!(last_chain_sync_date, :millisecond)
+     }, rest}
   end
 
   defp deserialize_versioned_unspent_output_list(rest, 0, _acc), do: {[], rest}

lib/archethic/transaction_chain.ex

@@ -697,7 +697,7 @@ defmodule Archethic.TransactionChain do
 
   def fetch_unspent_outputs(address, nodes, opts)
       when is_binary(address) and is_list(nodes) and is_list(opts) do
-    offset = Keyword.get(opts, :paging_offset, 0)
+    offset = Keyword.get(opts, :paging_offset, nil)
     limit = Keyword.get(opts, :limit, 0)
 
     Stream.resource(
@@ -726,9 +726,24 @@ defmodule Archethic.TransactionChain do
 
   defp do_fetch_unspent_outputs(address, nodes, offset, limit) do
     conflict_resolver = fn results ->
-      results
-      |> Enum.sort_by(&length(&1.unspent_outputs), :desc)
-      |> List.first()
+      %UnspentOutputList{last_chain_sync_date: highest_date} =
+        Enum.max_by(results, & &1.last_chain_sync_date, DateTime)
+
+      synced_results = Enum.filter(results, &(&1.last_chain_sync_date == highest_date))
+
+      merged_utxos = synced_results |> Enum.flat_map(& &1.unspent_outputs) |> Enum.uniq()
+
+      offset =
+        if Enum.empty?(merged_utxos),
+          do: nil,
+          else: merged_utxos |> List.last() |> VersionedUnspentOutput.hash()
+
+      %UnspentOutputList{
+        unspent_outputs: merged_utxos,
+        more?: Enum.any?(synced_results, & &1.more?),
+        last_chain_sync_date: highest_date,
+        offset: offset
+      }
     end
 
     case P2P.quorum_read(

lib/archethic_web/api/graphql/schema.ex

@@ -227,11 +227,11 @@ defmodule ArchethicWeb.API.GraphQL.Schema do
     """
     field :chain_unspent_outputs, list_of(:unspent_output) do
       arg(:address, non_null(:address))
-      arg(:paging_offset, :non_neg_integer)
+      arg(:paging_offset, :sha256_hash)
       arg(:limit, :pos_integer)
 
       resolve(fn args = %{address: address}, _ ->
-        paging_offset = Map.get(args, :paging_offset, 0)
+        paging_offset = Map.get(args, :paging_offset, nil)
         limit = Map.get(args, :limit, 0)
         Resolver.get_genesis_unspent_outputs(address, paging_offset, limit)
       end)

lib/archethic_web/api/graphql/schema/beacon_chain_summary_type.ex

@@ -50,7 +50,7 @@ defmodule ArchethicWeb.API.GraphQL.Schema.BeaconChainSummary do
     field(:movements_addresses, list_of(:address))
     field(:type, :string)
     field(:fee, :integer)
-    field(:validation_stamp_checksum, :hash)
+    field(:validation_stamp_checksum, :versioned_hash)
   end
 
   scalar :p2p_availabilities do

lib/archethic_web/api/graphql/schema/hash_type.ex

@@ -10,21 +10,29 @@ defmodule ArchethicWeb.API.GraphQL.Schema.HashType do
   The Hash appears in a JSON response as Base16 formatted string. The parsed hash will
   be converted to a binary and any invalid hash with an invalid algorithm or invalid size will be rejected
   """
-  scalar :hash do
+  scalar :versioned_hash do
     serialize(&Base.encode16/1)
-    parse(&parse_hash/1)
+    parse(&parse_hash(&1, :versioned))
   end
 
-  @spec parse_hash(Absinthe.Blueprint.Input.String.t()) :: {:ok, binary()} | :error
-  defp parse_hash(%Absinthe.Blueprint.Input.String{value: hash}) do
+  scalar :sha256_hash do
+    serialize(&Base.encode16/1)
+    parse(&parse_hash(&1, :sha256))
+  end
+
+  @spec parse_hash(Absinthe.Blueprint.Input.String.t(), hash_type :: :versioned | :sha256) ::
+          {:ok, binary()} | :error
+  defp parse_hash(%Absinthe.Blueprint.Input.String{value: hash}, hash_type) do
     with {:ok, hash} <- Base.decode16(hash, case: :mixed),
-         true <- Crypto.valid_hash?(hash) do
+         true <- valid_hash?(hash, hash_type) do
       {:ok, hash}
     else
-      _ ->
-        :error
+      _ -> :error
     end
   end
 
-  defp parse_hash(_), do: :error
+  defp parse_hash(_, _), do: :error
+
+  defp valid_hash?(hash, :versioned), do: Crypto.valid_hash?(hash)
+  defp valid_hash?(hash, :sha256), do: match?(<<_::32>>, hash)
 end

lib/archethic_web/api/graphql/schema/resolver.ex

@@ -348,7 +348,7 @@ defmodule ArchethicWeb.API.GraphQL.Schema.Resolver do
     |> paginate_transactions(page)
   end
 
-  def get_genesis_unspent_outputs(address, paging_offset \\ 0, limit \\ 0) do
+  def get_genesis_unspent_outputs(address, paging_offset \\ nil, limit \\ 0) do
     {:ok, Archethic.get_unspent_outputs(address, paging_offset, limit)}
   end
 end

lib/archethic_web/api/graphql/schema/transaction_type.ex

@@ -141,7 +141,7 @@ defmodule ArchethicWeb.API.GraphQL.Schema.TransactionType do
   object :validation_stamp do
     field(:timestamp, :timestamp)
     field(:proof_of_work, :public_key)
-    field(:proof_of_integrity, :hash)
+    field(:proof_of_integrity, :versioned_hash)
     field(:ledger_operations, :ledger_operations)
     field(:signature, :hex)
     field(:protocol_version, :integer)