go-blockchain/sync_loop.go

// Copyright (c) 2017-2026 Lethean (https://lt.hn)
//
// Licensed under the European Union Public Licence (EUPL) version 1.2.
// SPDX-License-Identifier: EUPL-1.2

package blockchain

import (
	"context"
	"crypto/rand"
	"encoding/binary"
	"fmt"
	"log"
	"net"
	"time"

	coreerr "dappco.re/go/core/log"

	"dappco.re/go/core/blockchain/chain"
	"dappco.re/go/core/blockchain/config"
	"dappco.re/go/core/blockchain/p2p"
	levin "dappco.re/go/core/p2p/node/levin"
)

var readPeerID = rand.Read
var chainHeight = func(blockchain *chain.Chain) (uint64, error) {
	return blockchain.Height()
}

func runChainSyncLoop(ctx context.Context, blockchain *chain.Chain, chainConfig *config.ChainConfig, hardForks []config.HardFork, seed string) {
	opts := chain.SyncOptions{
		VerifySignatures: false,
		Forks:            hardForks,
	}

	for {
		select {
		case <-ctx.Done():
			return
		default:
		}

		if err := runChainSyncOnce(ctx, blockchain, chainConfig, opts, seed); err != nil {
			log.Printf("%s: %v (retrying in 10s)", chainSyncDisplayName, err)
			select {
			case <-ctx.Done():
				return
			case <-time.After(10 * time.Second):
			}
			continue
		}

		select {
		case <-ctx.Done():
			return
		case <-time.After(30 * time.Second):
		}
	}
}

func runChainSyncOnce(ctx context.Context, blockchain *chain.Chain, chainConfig *config.ChainConfig, opts chain.SyncOptions, seed string) error {
	dialer := net.Dialer{Timeout: 10 * time.Second}
	tcpConn, err := dialer.DialContext(ctx, "tcp", seed)
	if err != nil {
		return coreerr.E("runChainSyncOnce", fmt.Sprintf("dial %s", seed), err)
	}
	defer tcpConn.Close()

	// Close the socket if the caller cancels so blocked handshake reads unwind promptly.
	done := make(chan struct{})
	go func() {
		select {
		case <-ctx.Done():
			_ = tcpConn.Close()
		case <-done:
		}
	}()
	defer close(done)

	levinConnection := levin.NewConnection(tcpConn)

	var peerIDBuf [8]byte
	if _, err := readPeerID(peerIDBuf[:]); err != nil {
		return coreerr.E("runChainSyncOnce", "read peer ID", err)
	}
	peerID := binary.LittleEndian.Uint64(peerIDBuf[:])

	localHeight, err := chainHeight(blockchain)
	if err != nil {
		return coreerr.E("runChainSyncOnce", "read local height", err)
	}

	handshakeReq := p2p.HandshakeRequest{
		NodeData: p2p.NodeData{
			NetworkID: chainConfig.NetworkID,
			PeerID:    peerID,
			LocalTime: time.Now().Unix(),
			MyPort:    0,
		},
		PayloadData: p2p.CoreSyncData{
			CurrentHeight:  localHeight,
			ClientVersion:  config.ClientVersion,
			NonPruningMode: true,
		},
	}
	payload, err := p2p.EncodeHandshakeRequest(&handshakeReq)
	if err != nil {
		return coreerr.E("runChainSyncOnce", "encode handshake", err)
	}
	if err := tcpConn.SetDeadline(time.Now().Add(10 * time.Second)); err != nil {
		return coreerr.E("runChainSyncOnce", "set handshake deadline", err)
	}
	if err := levinConnection.WritePacket(p2p.CommandHandshake, payload, true); err != nil {
		return coreerr.E("runChainSyncOnce", "write handshake", err)
	}

	hdr, data, err := levinConnection.ReadPacket()
	if err != nil {
		return coreerr.E("runChainSyncOnce", "read handshake", err)
	}
	if err := validateHandshakeHeader(hdr); err != nil {
		return coreerr.E("runChainSyncOnce", "validate handshake header", err)
	}

	var handshakeResp p2p.HandshakeResponse
	if err := handshakeResp.Decode(data); err != nil {
		return coreerr.E("runChainSyncOnce", "decode handshake", err)
	}
	if err := validateHandshakePeer(chainConfig, &handshakeResp); err != nil {
		return coreerr.E("runChainSyncOnce", "validate handshake", err)
	}
	if err := tcpConn.SetDeadline(time.Time{}); err != nil {
		return coreerr.E("runChainSyncOnce", "clear handshake deadline", err)
	}

	localSync := p2p.CoreSyncData{
		CurrentHeight:  localHeight,
		ClientVersion:  config.ClientVersion,
		NonPruningMode: true,
	}
	p2pConn := chain.NewLevinP2PConn(levinConnection, handshakeResp.PayloadData.CurrentHeight, localSync)

	return blockchain.P2PSync(ctx, p2pConn, opts)
}

func validateHandshakeHeader(hdr levin.Header) error {
	if hdr.Command != uint32(p2p.CommandHandshake) {
		return fmt.Errorf("unexpected command %d", hdr.Command)
	}
	if hdr.ReturnCode < 0 {
		return fmt.Errorf("handshake rejected with return code %d", hdr.ReturnCode)
	}
	return nil
}

func validateHandshakePeer(chainConfig *config.ChainConfig, handshakeResp *p2p.HandshakeResponse) error {
	if handshakeResp.NodeData.NetworkID != chainConfig.NetworkID {
		return fmt.Errorf("peer network ID %x does not match expected %x", handshakeResp.NodeData.NetworkID, chainConfig.NetworkID)
	}

	if err := p2p.ValidatePeerClientVersion(chainConfig.NetworkID, handshakeResp.PayloadData.ClientVersion); err != nil {
		return err
	}

	return nil
}