WIP. Peer loader implemented and used to restore blockchain history.

Get rid of internal package to simplify testing with mockery.

Author: alexeykiselev

.mockery.yaml

@@ -0,0 +1,18 @@
+quiet: False
+disable-version-string: True
+with-expecter: True
+mockname: "Mock{{.InterfaceName}}"
+filename: "{{.InterfaceName | snakecase}}.go"
+
+inpackage: True
+
+packages:
+  github.com/alexeykiselev/waves-fork-detector/peers:
+    interfaces:
+      HistoryRequester:
+  github.com/alexeykiselev/waves-fork-detector/chains:
+    interfaces:
+      HistoryProvider:
+  github.com/alexeykiselev/waves-fork-detector/loading:
+    interfaces:
+      Reporter:

Makefile

@@ -70,3 +70,6 @@ dist: clean release
 	@cd ./build/bin/darwin-amd64/; tar pzcvf ../../dist/forkdetector_$(VERSION)_macOS-amd64.tar.gz ./forkdetector*
 	@cd ./build/bin/darwin-arm64/; tar pzcvf ../../dist/forkdetector_$(VERSION)_macOS-arm64.tar.gz ./forkdetector*
 
+
+mock:
+	@mockery

internal/api.go renamed to api.go

@@ -1,11 +1,12 @@
-package internal
+package main
 
 import (
 	"compress/flate"
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"math/big"
 	"net/http"
 	"net/netip"
 	"runtime"
@@ -16,8 +17,8 @@ import (
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 
-	"github.com/alexeykiselev/waves-fork-detector/internal/chains"
-	"github.com/alexeykiselev/waves-fork-detector/internal/peers"
+	"github.com/alexeykiselev/waves-fork-detector/chains"
+	"github.com/alexeykiselev/waves-fork-detector/peers"
 )
 
 const defaultTimeout = 30 * time.Second
@@ -58,6 +59,14 @@ type status struct {
 	GoroutinesCount     int `json:"goroutines_count"`
 }
 
+type HeadInfo struct {
+	Number    uint64    `json:"number"`
+	ID        string    `json:"id"`
+	Height    uint32    `json:"height"`
+	Score     *big.Int  `json:"score"`
+	Timestamp time.Time `json:"timestamp"`
+}
+
 type API struct {
 	ctx      context.Context
 	wait     func() error
@@ -120,13 +129,15 @@ func (a *API) runServer() error {
 
 func (a *API) routes() chi.Router {
 	r := chi.NewRouter()
-	r.Get("/status", a.status)           // Status information
 	r.Get("/peers/all", a.peers)         // Returns the list of all known peers
 	r.Get("/peers/friendly", a.friendly) // Returns the list of peers that have been successfully connected at least once
 	r.Get("/connections", a.connections) // Returns the list of active connections
-	r.Get("/forks", a.forks)             // Returns the combined info about forks for all connected peers
-	r.Get("/all-forks", a.allForks)      // Returns the combined info about all registered forks
-	r.Get("/fork/{address}", a.fork)     // Returns the info about fork of the given peer
+	r.Get("/heads", a.heads)             // Returns the combined info about heads for all connected peers
+
+	r.Get("/status", a.status)       // Status information
+	r.Get("/forks", a.forks)         // Returns the combined info about forks for all connected peers
+	r.Get("/all-forks", a.allForks)  // Returns the combined info about all registered forks
+	r.Get("/fork/{address}", a.fork) // Returns the info about fork of the given peer
 	return r
 }
 
@@ -203,6 +214,34 @@ func (a *API) connections(w http.ResponseWriter, _ *http.Request) {
 	}
 }
 
+func (a *API) heads(w http.ResponseWriter, _ *http.Request) {
+	heads, err := a.linkage.Heads()
+	if err != nil {
+		http.Error(w, fmt.Sprintf("Failed to complete request: %v", err), http.StatusInternalServerError)
+		return
+	}
+	infos := make([]HeadInfo, len(heads))
+	for i, h := range heads {
+		b, blErr := a.linkage.Block(h.BlockID)
+		if blErr != nil {
+			http.Error(w, fmt.Sprintf("Failed to complete request: %v", blErr), http.StatusInternalServerError)
+			return
+		}
+		infos[i] = HeadInfo{
+			Number:    h.ID,
+			ID:        h.BlockID.String(),
+			Height:    b.Height,
+			Score:     b.Score,
+			Timestamp: time.UnixMilli(int64(b.Timestamp)),
+		}
+	}
+	err = json.NewEncoder(w).Encode(infos)
+	if err != nil {
+		http.Error(w, fmt.Sprintf("Failed to marshal heads to JSON: %v", err), http.StatusInternalServerError)
+		return
+	}
+}
+
 func (a *API) forks(w http.ResponseWriter, _ *http.Request) {
 	nodes, err := a.registry.Connections()
 	if err != nil {

internal/chains/entities.go renamed to chains/entities.go

@@ -11,11 +11,22 @@ import (
 )
 
 var (
-	ErrBlockNotFound  = fmt.Errorf("block not found")
-	ErrUnleashedPeer  = fmt.Errorf("peer not leashed")
 	ErrParentNotFound = fmt.Errorf("parent not found")
+	ErrRefNotFound    = fmt.Errorf("reference block not found")
 )
 
+type Leasher interface {
+	Leash(addr netip.Addr) (proto.BlockID, error)
+	MoveLeash(id proto.BlockID, addr netip.Addr) error
+}
+
+type HistoryProvider interface {
+	Leasher
+	HasBlock(id proto.BlockID) (bool, error)
+	LastIDs(id proto.BlockID, count int) ([]proto.BlockID, error)
+	PutBlock(block *proto.Block, addr netip.Addr) error
+}
+
 type Linkage struct {
 	scheme  proto.Scheme
 	genesis proto.BlockID
@@ -72,11 +83,39 @@ func (l *Linkage) PutBlock(block *proto.Block, addr netip.Addr) error {
 	return nil
 }
 
+func (l *Linkage) PutMicroBlock(inv *proto.MicroBlockInv, addr netip.Addr) error {
+	ok, err := l.st.hasBlock(inv.TotalBlockID)
+	if err != nil {
+		return err
+	}
+	if ok {
+		if ulErr := l.st.updateLeash(addr, inv.TotalBlockID); ulErr != nil {
+			return ulErr
+		}
+		return nil
+	}
+	hasRef, err := l.st.hasBlock(inv.Reference)
+	if err != nil {
+		return err
+	}
+	if !hasRef {
+		return ErrRefNotFound
+	}
+	if insErr := l.st.putMicroBlock(inv); insErr != nil {
+		return insErr
+	}
+	if ulErr := l.st.updateLeash(addr, inv.TotalBlockID); ulErr != nil {
+		return ulErr
+	}
+	return nil
+}
+
+// Leash return the block ID the peer is leashed to. For an unleashed peer the genesis block ID is returned.
 func (l *Linkage) Leash(addr netip.Addr) (proto.BlockID, error) {
 	lsh, err := l.st.leash(addr)
 	if err != nil {
 		if errors.Is(err, leveldb.ErrNotFound) {
-			return proto.BlockID{}, ErrUnleashedPeer
+			return l.genesis, nil // Return genesis block ID in case of no leash for the peer.
 		}
 		return proto.BlockID{}, err
 	}

internal/chains/linkage.go renamed to chains/linkage.go

@@ -33,16 +33,17 @@ func TestLeashUpdate(t *testing.T) {
 	bl3 := createNthBlock(t, bl2.BlockID(), 3)
 
 	a := netip.MustParseAddr("8.8.8.8")
-	_, err := lk.Leash(a)
-	assert.ErrorIs(t, err, ErrUnleashedPeer)
+	id, err := lk.Leash(a)
+	assert.NoError(t, err)
+	assert.Equal(t, id1, id)
 
 	err = lk.PutBlock(bl2, a)
 	require.NoError(t, err)
 
 	ok, err := lk.hasBlock(bl2.BlockID())
 	require.NoError(t, err)
 	assert.True(t, ok)
-	id, err := lk.Leash(a)
+	id, err = lk.Leash(a)
 	require.NoError(t, err)
 	assert.Equal(t, bl2.BlockID(), id)
 
@@ -66,15 +67,17 @@ func TestMultipleLeashes(t *testing.T) {
 	a := netip.MustParseAddr("8.8.8.8")
 	b := netip.MustParseAddr("9.9.9.9")
 
-	_, err := lk.Leash(a)
-	assert.ErrorIs(t, err, ErrUnleashedPeer)
-	_, err = lk.Leash(b)
-	assert.ErrorIs(t, err, ErrUnleashedPeer)
+	id, err := lk.Leash(a)
+	assert.NoError(t, err)
+	assert.Equal(t, id1, id)
+	id, err = lk.Leash(b)
+	assert.NoError(t, err)
+	assert.Equal(t, id1, id)
 
 	err = lk.PutBlock(bl2, a)
 	require.NoError(t, err)
 
-	id, err := lk.Leash(a)
+	id, err = lk.Leash(a)
 	require.NoError(t, err)
 	assert.Equal(t, bl2.BlockID(), id)
 

internal/chains/linkage_internal_test.go renamed to chains/linkage_internal_test.go

@@ -176,6 +176,63 @@ func (s *storage) putProtoBlock(block *proto.Block) error {
 	return s.db.Write(batch, nil)
 }
 
+func (s *storage) putMicroBlock(inv *proto.MicroBlockInv) error {
+	sn, err := s.db.GetSnapshot()
+	if err != nil {
+		return fmt.Errorf("failed to put micro-block: %w", err)
+	}
+	defer sn.Release()
+	n, err := s.getCounter(sn, blocksCounter)
+	if err != nil {
+		return fmt.Errorf("failed to put micro-block: %w", err)
+	}
+	// Get reference block and its number.
+	refNum, err := s.getNumOfBlockID(sn, inv.Reference)
+	if err != nil {
+		return fmt.Errorf("failed to put micro-block: %w", err)
+	}
+
+	ref, err := s.getBlock(sn, refNum)
+	if err != nil {
+		return fmt.Errorf("failed to put micro-block: %w", err)
+	}
+
+	parentID, err := proto.NewBlockIDFromBytes(ref.Parent)
+	if err != nil {
+		return fmt.Errorf("failed to put micro-block: %w", err)
+	}
+
+	batch := new(leveldb.Batch)
+
+	// Put next block number.
+	n++
+	s.putCounter(batch, blocksCounter, n)
+
+	b := block{
+		Height:    ref.Height,
+		Timestamp: ref.Timestamp,
+		Score:     ref.Score,
+		ID:        inv.TotalBlockID.Bytes(),
+		Parent:    ref.Parent,
+		Generator: ref.Generator,
+	}
+
+	s.putID(batch, blockPrefix, inv.TotalBlockID, n) // Save reference ID -> num.
+	// Put information about block by its number.
+	if putErr := s.putBlock(batch, n, b); putErr != nil {
+		return fmt.Errorf("failed to put micro-block: %w", putErr)
+	}
+
+	// Put block into the chain.
+	if lnkErr := s.putNewLinkForBlock(sn, batch, n, inv.TotalBlockID, b.Height, parentID); lnkErr != nil {
+		return fmt.Errorf("failed to put micro-block: %w", lnkErr)
+	}
+	if phErr := s.putHead(sn, batch, n, refNum); phErr != nil {
+		return fmt.Errorf("failed to put micro-block: %w", phErr)
+	}
+	return s.db.Write(batch, nil)
+}
+
 func (s *storage) getAncestors(id proto.BlockID, count int) ([]proto.BlockID, error) {
 	sn, err := s.db.GetSnapshot()
 	if err != nil {
@@ -397,7 +454,7 @@ func (s *storage) putNewLinkForBlock(
 		parents = append(parents, pn)
 		pl, glErr := s.getLink(sn, pn)
 		if glErr != nil {
-			return err
+			return glErr
 		}
 		if len(pl.Parents) > l {
 			pn = pl.Parents[l]

internal/chains/storage.go renamed to chains/storage.go

@@ -1,4 +1,4 @@
-package internal
+package main
 
 import (
 	"context"
@@ -13,7 +13,7 @@ import (
 	"github.com/wavesplatform/gowaves/pkg/p2p/peer"
 	"github.com/wavesplatform/gowaves/pkg/proto"
 
-	"github.com/alexeykiselev/waves-fork-detector/internal/peers"
+	"github.com/alexeykiselev/waves-fork-detector/peers"
 )
 
 const (