Merge remote-tracking branch 'origin/develop' into feat/upgrade.hive.go

components/debugapi/blocks.go

@@ -16,7 +16,13 @@ func getSlotBlockIDs(index iotago.SlotIndex) (*BlockChangesResponse, error) {
 	}
 
 	includedBlocks := make([]string, 0)
-	tangleTree := ads.NewSet[iotago.Identifier](mapdb.NewMapDB(), iotago.BlockID.Bytes, iotago.BlockIDFromBytes)
+	tangleTree := ads.NewSet[iotago.Identifier](
+		mapdb.NewMapDB(),
+		iotago.Identifier.Bytes,
+		iotago.IdentifierFromBytes,
+		iotago.BlockID.Bytes,
+		iotago.BlockIDFromBytes,
+	)
 
 	_ = blocksForSlot.StreamKeys(func(blockID iotago.BlockID) error {
 		includedBlocks = append(includedBlocks, blockID.String())

components/debugapi/transactions.go

@@ -21,7 +21,13 @@ func storeTransactionsPerSlot(scd *notarization.SlotCommittedDetails) error {
 	if err != nil {
 		return ierrors.Wrapf(err, "failed to retrieve state diff for slot %d", slot)
 	}
-	mutationsTree := ads.NewSet[iotago.Identifier](mapdb.NewMapDB(), iotago.TransactionID.Bytes, iotago.TransactionIDFromBytes)
+	mutationsTree := ads.NewSet[iotago.Identifier](
+		mapdb.NewMapDB(),
+		iotago.Identifier.Bytes,
+		iotago.IdentifierFromBytes,
+		iotago.TransactionID.Bytes,
+		iotago.TransactionIDFromBytes,
+	)
 	tcs := &TransactionsChangesResponse{
 		Index:                slot,
 		IncludedTransactions: make([]string, 0),

components/inx/server_blocks.go

@@ -12,6 +12,7 @@ import (
 	"github.com/iotaledger/hive.go/runtime/workerpool"
 	inx "github.com/iotaledger/inx/go"
 	"github.com/iotaledger/iota-core/pkg/blockhandler"
+	"github.com/iotaledger/iota-core/pkg/model"
 	"github.com/iotaledger/iota-core/pkg/protocol/engine/blocks"
 	iotago "github.com/iotaledger/iota.go/v4"
 )
@@ -123,6 +124,33 @@ func (s *Server) ListenToConfirmedBlocks(_ *inx.NoParams, srv inx.INX_ListenToCo
 	return ctx.Err()
 }
 
+func (s *Server) ReadAcceptedBlocks(slot *inx.SlotIndex, srv inx.INX_ReadAcceptedBlocksServer) error {
+	blocksStore, err := deps.Protocol.MainEngineInstance().Storage.Blocks(slot.Unwrap())
+	if err != nil {
+		return status.Errorf(codes.InvalidArgument, "failed to get blocks: %s", err.Error())
+	}
+
+	if err := blocksStore.ForEachBlockInSlot(func(block *model.Block) error {
+		metadata, err := getINXBlockMetadata(block.ID())
+		if err != nil {
+			return err
+		}
+
+		payload := &inx.BlockWithMetadata{
+			Metadata: metadata,
+			Block: &inx.RawBlock{
+				Data: block.Data(),
+			},
+		}
+
+		return srv.Send(payload)
+	}); err != nil {
+		return status.Errorf(codes.Internal, "failed to iterate blocks: %s", err.Error())
+	}
+
+	return nil
+}
+
 func (s *Server) SubmitBlock(ctx context.Context, rawBlock *inx.RawBlock) (*inx.BlockId, error) {
 	block, err := rawBlock.UnwrapBlock(deps.Protocol)
 	if err != nil {

components/inx/server_commitments.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/iotaledger/hive.go/ierrors"
 	"github.com/iotaledger/hive.go/kvstore"
+	"github.com/iotaledger/hive.go/runtime/workerpool"
 	inx "github.com/iotaledger/inx/go"
 	"github.com/iotaledger/iota-core/pkg/model"
 	iotago "github.com/iotaledger/iota.go/v4"
@@ -22,6 +23,135 @@ func inxCommitment(commitment *model.Commitment) *inx.Commitment {
 	}
 }
 
+func (s *Server) ListenToCommitments(req *inx.SlotRangeRequest, srv inx.INX_ListenToCommitmentsServer) error {
+	createCommitmentPayloadForSlotAndSend := func(slot iotago.SlotIndex) error {
+		commitment, err := deps.Protocol.MainEngineInstance().Storage.Commitments().Load(slot)
+		if err != nil {
+			if ierrors.Is(err, kvstore.ErrKeyNotFound) {
+				return status.Errorf(codes.NotFound, "commitment slot %d not found", slot)
+			}
+
+			return err
+		}
+
+		if err := srv.Send(inxCommitment(commitment)); err != nil {
+			return ierrors.Errorf("send error: %w", err)
+		}
+
+		return nil
+	}
+
+	sendSlotsRange := func(startSlot iotago.SlotIndex, endSlot iotago.SlotIndex) error {
+		for currentSlot := startSlot; currentSlot <= endSlot; currentSlot++ {
+			if err := createCommitmentPayloadForSlotAndSend(currentSlot); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	}
+
+	// if a startSlot is given, we send all available commitments including the start slot.
+	// if an endSlot is given, we send all available commitments up to and including min(latestCommitmentSlot, endSlot).
+	// if no startSlot is given, but an endSlot, we don't send previous commitments.
+	sendPreviousSlots := func(startSlot iotago.SlotIndex, endSlot iotago.SlotIndex) (iotago.SlotIndex, error) {
+		if startSlot == 0 {
+			// no need to send previous commitments
+			return 0, nil
+		}
+
+		latestCommitment := deps.Protocol.MainEngineInstance().SyncManager.LatestCommitment()
+
+		if startSlot > latestCommitment.Slot() {
+			// no need to send previous commitments
+			return 0, nil
+		}
+
+		// Stream all available commitments first
+		prunedEpoch, hasPruned := deps.Protocol.MainEngineInstance().SyncManager.LastPrunedEpoch()
+		if hasPruned && startSlot <= deps.Protocol.CommittedAPI().TimeProvider().EpochEnd(prunedEpoch) {
+			return 0, status.Errorf(codes.InvalidArgument, "given startSlot %d is older than the current pruningSlot %d", startSlot, deps.Protocol.CommittedAPI().TimeProvider().EpochEnd(prunedEpoch))
+		}
+
+		if endSlot == 0 || endSlot > latestCommitment.Slot() {
+			endSlot = latestCommitment.Slot()
+		}
+
+		if err := sendSlotsRange(startSlot, endSlot); err != nil {
+			return 0, err
+		}
+
+		return endSlot, nil
+	}
+
+	stream := &streamRange{
+		start: iotago.SlotIndex(req.GetStartSlot()),
+		end:   iotago.SlotIndex(req.GetEndSlot()),
+	}
+
+	var err error
+	stream.lastSent, err = sendPreviousSlots(stream.start, stream.end)
+	if err != nil {
+		return err
+	}
+
+	if stream.isBounded() && stream.lastSent >= stream.end {
+		// We are done sending, so close the stream
+		return nil
+	}
+
+	catchUpFunc := func(start iotago.SlotIndex, end iotago.SlotIndex) error {
+		err := sendSlotsRange(start, end)
+		if err != nil {
+			err := ierrors.Errorf("sendSlotsRange error: %w", err)
+			Component.LogError(err.Error())
+
+			return err
+		}
+
+		return nil
+	}
+
+	sendFunc := func(_ iotago.SlotIndex, payload *inx.Commitment) error {
+		if err := srv.Send(payload); err != nil {
+			err := ierrors.Errorf("send error: %w", err)
+			Component.LogError(err.Error())
+
+			return err
+		}
+
+		return nil
+	}
+
+	var innerErr error
+	ctx, cancel := context.WithCancel(Component.Daemon().ContextStopped())
+
+	wp := workerpool.New("ListenToCommitments", workerpool.WithWorkerCount(workerCount)).Start()
+
+	unhook := deps.Protocol.Events.Engine.Notarization.LatestCommitmentUpdated.Hook(func(commitment *model.Commitment) {
+		done, err := handleRangedSend1(commitment.Slot(), inxCommitment(commitment), stream, catchUpFunc, sendFunc)
+		switch {
+		case err != nil:
+			innerErr = err
+			cancel()
+
+		case done:
+			cancel()
+		}
+	}).Unhook
+
+	<-ctx.Done()
+	unhook()
+
+	// We need to wait until all tasks are done, otherwise we might call
+	// "SendMsg" and "CloseSend" in parallel on the grpc stream, which is
+	// not safe according to the grpc docs.
+	wp.Shutdown()
+	wp.ShutdownComplete.Wait()
+
+	return innerErr
+}
+
 func (s *Server) ForceCommitUntil(_ context.Context, slot *inx.SlotIndex) (*inx.NoParams, error) {
 	err := deps.Protocol.MainEngineInstance().Notarization.ForceCommitUntil(slot.Unwrap())
 	if err != nil {

go.mod

@@ -10,30 +10,29 @@ require (
 	github.com/google/uuid v1.4.0
 	github.com/gorilla/websocket v1.5.0
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
-	github.com/iotaledger/hive.go/ads v0.0.0-20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/app v0.0.0-20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/constraints v0.0.0-20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/core v1.0.0-rc.3.0.20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/crypto v0.0.0-20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/ds v0.0.0-20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/ierrors v0.0.0-20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/kvstore v0.0.0-20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/lo v0.0.0-20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/logger v0.0.0-20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/runtime v0.0.0-20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/serializer/v2 v2.0.0-rc.1.0.20231107225803-f89acd088c10
-	github.com/iotaledger/hive.go/stringify v0.0.0-20231107225803-f89acd088c10
-	github.com/iotaledger/inx-app v1.0.0-rc.3.0.20231031135002-4c79ea5193f5
-	github.com/iotaledger/inx/go v1.0.0-rc.2.0.20231031134131-b6ad918dc1ac
-	github.com/iotaledger/iota.go/v4 v4.0.0-20231102113728-20b8d01e826e
+	github.com/iotaledger/hive.go/ads v0.0.0-20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/app v0.0.0-20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/constraints v0.0.0-20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/core v1.0.0-rc.3.0.20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/crypto v0.0.0-20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/ds v0.0.0-20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/ierrors v0.0.0-20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/kvstore v0.0.0-20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/lo v0.0.0-20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/logger v0.0.0-20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/runtime v0.0.0-20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/serializer/v2 v2.0.0-rc.1.0.20231108050255-98e0fa35e936
+	github.com/iotaledger/hive.go/stringify v0.0.0-20231108050255-98e0fa35e936
+	github.com/iotaledger/inx-app v1.0.0-rc.3.0.20231108104504-1445f545de82
+	github.com/iotaledger/inx/go v1.0.0-rc.2.0.20231108104322-f301c3573998
+	github.com/iotaledger/iota.go/v4 v4.0.0-20231108103955-bf75d703d8aa
 	github.com/labstack/echo/v4 v4.11.2
 	github.com/labstack/gommon v0.4.0
 	github.com/libp2p/go-libp2p v0.32.0
 	github.com/libp2p/go-libp2p-kad-dht v0.25.1
 	github.com/mr-tron/base58 v1.2.0
 	github.com/multiformats/go-multiaddr v0.12.0
 	github.com/multiformats/go-varint v0.0.7
-	github.com/orcaman/writerseeker v0.0.0-20200621085525-1d3f536ff85e
 	github.com/otiai10/copy v1.14.0
 	github.com/prometheus/client_golang v1.17.0
 	github.com/spf13/pflag v1.0.5
@@ -43,6 +42,7 @@ require (
 	go.uber.org/atomic v1.11.0
 	go.uber.org/dig v1.17.1
 	golang.org/x/crypto v0.14.0
+	golang.org/x/exp v0.0.0-20231006140011-7918f672742d
 	google.golang.org/grpc v1.59.0
 	google.golang.org/protobuf v1.31.0
 )
@@ -89,7 +89,7 @@ require (
 	github.com/huin/goupnp v1.3.0 // indirect
 	github.com/iancoleman/orderedmap v0.3.0 // indirect
 	github.com/iotaledger/grocksdb v1.7.5-0.20230220105546-5162e18885c7 // indirect
-	github.com/iotaledger/hive.go/log v0.0.0-20231107225803-f89acd088c10 // indirect
+	github.com/iotaledger/hive.go/log v0.0.0-20231108050255-98e0fa35e936 // indirect
 	github.com/ipfs/boxo v0.13.1 // indirect
 	github.com/ipfs/go-cid v0.4.1 // indirect
 	github.com/ipfs/go-datastore v0.6.0 // indirect
@@ -169,7 +169,6 @@ require (
 	go.uber.org/mock v0.3.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.26.0 // indirect
-	golang.org/x/exp v0.0.0-20231006140011-7918f672742d // indirect
 	golang.org/x/image v0.13.0 // indirect
 	golang.org/x/mod v0.13.0 // indirect
 	golang.org/x/net v0.17.0 // indirect