Merge branch 'feat/cache-reset' of github.com:iotaledger/iota-core in…

…to feat/reactive-chainmanager

.github/workflows/golangci-lint.yml

@@ -6,7 +6,6 @@ on:
       - 'documentation/**'
       - 'scripts/**'
       - 'tools/**'
-
 jobs:
   golangci-lint:
     name: GolangCI-Lint
@@ -15,12 +14,26 @@ jobs:
       - name: Check out code into the Go module directory
         uses: actions/checkout@v3
 
+      - name: Checkout custom linter
+        uses: actions/checkout@v3
+        with:
+          repository: karimodm/typegroupingcheck
+          path: typegroupingcheck 
+
+      - name: Setup go
+        uses: actions/setup-go@v4
+        with:
+          go-version-file: './typegroupingcheck/go.mod'
+
+      - name: Build custom linter
+        working-directory: ./typegroupingcheck
+        run: |
+          go build -buildmode=plugin -o typegroupingcheck.so
+
       - name: golangci-lint
-        uses: reviewdog/action-golangci-lint@v2
+        uses: golangci/golangci-lint-action@v3
         with:
-          version: v1.53.3
-          github_token: ${{ secrets.GITHUB_TOKEN }}
-          golangci_lint_flags: "--timeout=10m"
-          reporter: github-pr-check
-          filter_mode: nofilter
-          fail_on_error: true
+          skip-cache: true
+          version: latest
+          install-mode: goinstall
+          args: --timeout=10m --enable typegroupingcheck

.golangci.yml

@@ -33,12 +33,17 @@ linters-settings:
                       desc: Should be replaced with "github.com/iotaledger/hive.go/ierrors" package
                     - pkg: "github.com/pkg/errors"
                       desc: Should be replaced with "github.com/iotaledger/hive.go/ierrors" package
+    custom:
+        typegroupingcheck:
+            path: ./typegroupingcheck/typegroupingcheck.so
+            description: Check for grouped types in functions' parameters
 
 linters:
     # Disable all linters.
     disable-all: true
     # Enable specific linter
     enable:
+        #- typegroupingcheck
         - errcheck
         - gosimple
         - govet

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/debug_models.go

@@ -37,16 +37,16 @@ type (
 	}
 
 	Validator struct {
-		AccountID      iotago.AccountID `serix:"0,mapKey=accountId"`
-		SeatIndex      uint8            `serix:"1,mapKey=seatIndex"`
-		PoolStake      iotago.BaseToken `serix:"2,mapKey=poolStake"`
-		ValidatorStake iotago.BaseToken `serix:"3,mapKey=validatorStake"`
-		FixedCost      iotago.Mana      `serix:"4,mapKey=fixedCost"`
+		AccountID      iotago.AccountID `serix:""`
+		SeatIndex      uint8            `serix:""`
+		PoolStake      iotago.BaseToken `serix:""`
+		ValidatorStake iotago.BaseToken `serix:""`
+		FixedCost      iotago.Mana      `serix:""`
 	}
 
 	ValidatorsSummaryResponse struct {
-		ValidatorSeats []*Validator `serix:"0,lengthPrefixType=uint8,mapKey=validatorSeats"`
-		ActiveSeats    []uint32     `serix:"1,lengthPrefixType=uint8,mapKey=activeSeats"`
+		ValidatorSeats []*Validator `serix:"lenPrefix=uint8"`
+		ActiveSeats    []uint32     `serix:"lenPrefix=uint8"`
 	}
 
 	BlockChangesResponse struct {

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.Engines.Main.Get().Notarization.ForceCommitUntil(slot.Unwrap())
 	if err != nil {

components/metrics/collector/collector.go

@@ -44,7 +44,7 @@ func (c *Collector) Collect() {
 // Update updates the value of the existing metric defined by the subsystem and metricName.
 // Note that the label values must be passed in the same order as they were defined in the metric, and must match the
 // number of labels defined in the metric.
-func (c *Collector) Update(subsystem, metricName string, metricValue float64, labelValues ...string) {
+func (c *Collector) Update(subsystem string, metricName string, metricValue float64, labelValues ...string) {
 	m := c.getMetric(subsystem, metricName)
 	if m != nil {
 		m.update(metricValue, labelValues...)
@@ -54,15 +54,15 @@ func (c *Collector) Update(subsystem, metricName string, metricValue float64, la
 // Increment increments the value of the existing metric defined by the subsystem and metricName.
 // Note that the label values must be passed in the same order as they were defined in the metric, and must match the
 // number of labels defined in the metric.
-func (c *Collector) Increment(subsystem, metricName string, labels ...string) {
+func (c *Collector) Increment(subsystem string, metricName string, labels ...string) {
 	m := c.getMetric(subsystem, metricName)
 	if m != nil {
 		m.increment(labels...)
 	}
 }
 
 // DeleteLabels deletes the metric with the given labels values.
-func (c *Collector) DeleteLabels(subsystem, metricName string, labelValues map[string]string) {
+func (c *Collector) DeleteLabels(subsystem string, metricName string, labelValues map[string]string) {
 	m := c.getMetric(subsystem, metricName)
 	if m != nil {
 		m.deleteLabels(labelValues)
@@ -85,7 +85,7 @@ func (c *Collector) Shutdown() {
 	}
 }
 
-func (c *Collector) getMetric(subsystem, metricName string) *Metric {
+func (c *Collector) getMetric(subsystem string, metricName string) *Metric {
 	col := c.getCollection(subsystem)
 	if col != nil {
 		return col.GetMetric(metricName)

components/protocol/component.go

@@ -61,7 +61,7 @@ type dependencies struct {
 }
 
 type jsonProtocolParameters struct {
-	ProtocolParameters []iotago.ProtocolParameters `serix:"0,mapKey=protocolParameters"`
+	ProtocolParameters []iotago.ProtocolParameters `serix:""`
 }
 
 func readProtocolParameters() []iotago.ProtocolParameters {