Blockchain updates for L2 extension

Makefile

@@ -226,6 +226,8 @@ proto:
 	@protoc --proto_path=pkg/grpc/protobuf-schemas/proto/ --go_out=./ --go_opt=module=$(MODULE) --go-vtproto_out=./ --go-vtproto_opt=features=marshal_strict+unmarshal+size --go-vtproto_opt=module=$(MODULE) pkg/grpc/protobuf-schemas/proto/waves/lang/*.proto
 	@protoc --proto_path=pkg/grpc/protobuf-schemas/proto/ --go_out=./ --go_opt=module=$(MODULE) --go-vtproto_out=./ --go-vtproto_opt=features=marshal_strict+unmarshal+size --go-vtproto_opt=module=$(MODULE) pkg/grpc/protobuf-schemas/proto/waves/events/*.proto
 	@protoc --proto_path=pkg/grpc/protobuf-schemas/proto/ --go_out=./ --go_opt=module=$(MODULE) --go-grpc_out=./ --go-grpc_opt=require_unimplemented_servers=false --go-grpc_opt=module=$(MODULE) pkg/grpc/protobuf-schemas/proto/waves/events/grpc/*.proto
+proto-l2:
+	@protoc --proto_path=pkg/grpc/protobuf-schemas/proto/ --proto_path=pkg/grpc/l2/blockchain_info/ --go_out=./ --go_opt=module=$(MODULE) --go-vtproto_out=./ --go-vtproto_opt=features=marshal_strict+unmarshal+size --go-vtproto_opt=module=$(MODULE) pkg/grpc/l2/blockchain_info/*.proto
 
 build-node-mainnet-amd64-deb-package: release-node
 	@mkdir -p build/dist

cmd/blockchaininfo/nats_subscriber.go

@@ -0,0 +1,183 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"log"
+	"os"
+	"os/signal"
+	"strconv"
+	"strings"
+	"syscall"
+
+	"go.uber.org/zap"
+
+	"github.com/nats-io/nats.go"
+	"github.com/pkg/errors"
+	"github.com/wavesplatform/gowaves/pkg/blockchaininfo"
+	g "github.com/wavesplatform/gowaves/pkg/grpc/l2/blockchain_info"
+	"github.com/wavesplatform/gowaves/pkg/proto"
+)
+
+func printBlockInfo(blockInfoProto *g.BlockInfo) error {
+	blockInfo, err := blockchaininfo.BUpdatesInfoFromProto(blockInfoProto)
+	if err != nil {
+		return err
+	}
+	blockInfoJSON, err := json.Marshal(blockInfo)
+	if err != nil {
+		return err
+	}
+	log.Println(string(blockInfoJSON))
+	return nil
+}
+
+func printContractInfo(contractInfoProto *g.L2ContractDataEntries, scheme proto.Scheme, path string) error {
+	contractInfo, err := blockchaininfo.L2ContractDataEntriesFromProto(contractInfoProto, scheme)
+	if err != nil {
+		return err
+	}
+	// Delete data entries are not going to have "type"
+	prettyJSON, err := json.MarshalIndent(contractInfo, "", "    ")
+	if err != nil {
+		log.Println("Error converting to pretty JSON:", err)
+		return err
+	}
+	heightStr := strconv.FormatUint(contractInfoProto.Height, 10)
+	// Write the pretty JSON to a file
+	err = os.WriteFile(path+heightStr+".json", prettyJSON, 0600)
+	if err != nil {
+		log.Println("Error writing to file:", err)
+		return err
+	}
+
+	return nil
+}
+
+func receiveBlockUpdates(msg *nats.Msg) {
+	blockUpdatesInfo := new(g.BlockInfo)
+	unmrshlErr := blockUpdatesInfo.UnmarshalVT(msg.Data)
+	if unmrshlErr != nil {
+		log.Printf("failed to unmarshal block updates, %v", unmrshlErr)
+		return
+	}
+
+	err := printBlockInfo(blockUpdatesInfo)
+	if err != nil {
+		return
+	}
+	log.Printf("Received on %s:\n", msg.Subject)
+}
+
+func receiveContractUpdates(msg *nats.Msg, contractMsg []byte, scheme proto.Scheme, path string) []byte {
+	zap.S().Infof("Received on %s:\n", msg.Subject)
+
+	switch msg.Data[0] {
+	case blockchaininfo.NoPaging:
+		contractMsg = msg.Data[1:]
+		contractUpdatesInfo := new(g.L2ContractDataEntries)
+		if err := contractUpdatesInfo.UnmarshalVT(contractMsg); err != nil {
+			log.Printf("Failed to unmarshal contract updates: %v", err)
+			return contractMsg
+		}
+		if err := printContractInfo(contractUpdatesInfo, scheme, path); err != nil {
+			log.Printf("Failed to print contract info: %v", err)
+			return contractMsg
+		}
+		contractMsg = nil
+
+	case blockchaininfo.StartPaging:
+		contractMsg = append(contractMsg, msg.Data[1:]...)
+
+	case blockchaininfo.EndPaging:
+		if contractMsg != nil {
+			contractMsg = append(contractMsg, msg.Data[1:]...)
+			contractUpdatesInfo := new(g.L2ContractDataEntries)
+			if err := contractUpdatesInfo.UnmarshalVT(contractMsg); err != nil {
+				log.Printf("Failed to unmarshal contract updates: %v", err)
+				return contractMsg
+			}
+
+			go func() {
+				if err := printContractInfo(contractUpdatesInfo, scheme, path); err != nil {
+					log.Printf("Failed to print contract info updates: %v", err)
+				}
+			}()
+			contractMsg = nil
+		}
+	}
+
+	return contractMsg
+}
+
+func main() {
+	var (
+		blockchainType string
+		updatesPath    string
+		natsURL        string
+	)
+	// Initialize the zap logger
+	l, err := zap.NewProduction()
+	if err != nil {
+		log.Fatalf("failed to initialize zap logger: %v", err)
+	}
+	defer func(l *zap.Logger) {
+		syncErr := l.Sync()
+		if syncErr != nil {
+			log.Fatalf("failed to sync zap logger %v", syncErr)
+		}
+	}(l)
+
+	flag.StringVar(&blockchainType, "blockchain-type", "testnet", "Blockchain scheme (e.g., stagenet, testnet, mainnet)")
+	flag.StringVar(&updatesPath, "updates-path", "", "File path to store contract updates")
+	flag.StringVar(&natsURL, "nats-url", nats.DefaultURL, "URL for the NATS server")
+
+	flag.Parse()
+
+	scheme, err := schemeFromString(blockchainType)
+	if err != nil {
+		zap.S().Fatalf("Failed to parse the blockchain type: %v", err)
+	}
+	ctx, done := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer done()
+	// Connect to a NATS server
+	nc, err := nats.Connect(natsURL)
+	if err != nil {
+		zap.S().Fatalf("Failed to connect to nats server: %v", err)
+		return
+	}
+	defer nc.Close()
+
+	_, err = nc.Subscribe(blockchaininfo.BlockUpdates, func(msg *nats.Msg) {
+		receiveBlockUpdates(msg)
+	})
+	if err != nil {
+		zap.S().Fatalf("Failed to subscribe to block updates: %v", err)
+		return
+	}
+
+	var contractMsg []byte
+	_, err = nc.Subscribe(blockchaininfo.ContractUpdates, func(msg *nats.Msg) {
+		contractMsg = receiveContractUpdates(msg, contractMsg, scheme, updatesPath)
+	})
+	if err != nil {
+		zap.S().Fatalf("Failed to subscribe to contract updates: %v", err)
+		return
+	}
+	<-ctx.Done()
+	zap.S().Info("NATS subscriber finished...")
+}
+
+func schemeFromString(networkType string) (proto.Scheme, error) {
+	switch strings.ToLower(networkType) {
+	case "mainnet":
+		return proto.MainNetScheme, nil
+	case "testnet":
+		return proto.TestNetScheme, nil
+	case "stagenet":
+		return proto.StageNetScheme, nil
+	default:
+		return 0, errors.New("invalid blockchain type string")
+	}
+}

cmd/importer/importer.go

@@ -192,7 +192,7 @@ func runImporter(c *cfg) error {
 		return err
 	}
 
-	st, err := state.NewState(c.dataDirPath, false, c.params(fds), ss, false)
+	st, err := state.NewState(c.dataDirPath, false, c.params(fds), ss, false, nil)
 	if err != nil {
 		return fmt.Errorf("failed to create state: %w", err)
 	}

cmd/node/node.go

@@ -20,11 +20,13 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/wavesplatform/gowaves/pkg/api"
+	"github.com/wavesplatform/gowaves/pkg/blockchaininfo"
 	"github.com/wavesplatform/gowaves/pkg/crypto"
 	"github.com/wavesplatform/gowaves/pkg/grpc/server"
 	"github.com/wavesplatform/gowaves/pkg/libs/microblock_cache"
@@ -72,51 +74,52 @@ var defaultPeers = map[string]string{
 type config struct {
 	isParsed bool
 
-	logLevel                   zapcore.Level
-	logDevelopment             bool
-	logNetwork                 bool
-	logNetworkData             bool
-	logFSM                     bool
-	statePath                  string
-	blockchainType             string
-	peerAddresses              string
-	declAddr                   string
-	nodeName                   string
-	cfgPath                    string
-	apiAddr                    string
-	apiKey                     string
-	apiMaxConnections          int
-	rateLimiterOptions         string
-	grpcAddr                   string
-	grpcAPIMaxConnections      int
-	enableMetaMaskAPI          bool
-	enableMetaMaskAPILog       bool
-	enableGrpcAPI              bool
-	blackListResidenceTime     time.Duration
-	buildExtendedAPI           bool
-	serveExtendedAPI           bool
-	buildStateHashes           bool
-	bindAddress                string
-	disableOutgoingConnections bool
-	minerVoteFeatures          string
-	disableBloomFilter         bool
-	reward                     int64
-	obsolescencePeriod         time.Duration
-	walletPath                 string
-	walletPassword             string
-	limitAllConnections        uint
-	minPeersMining             int
-	disableMiner               bool
-	profiler                   bool
-	prometheus                 string
-	metricsID                  int
-	metricsURL                 string
-	dropPeers                  bool
-	dbFileDescriptors          uint
-	newConnectionsLimit        int
-	disableNTP                 bool
-	microblockInterval         time.Duration
-	enableLightMode            bool
+	logLevel                      zapcore.Level
+	logDevelopment                bool
+	logNetwork                    bool
+	logNetworkData                bool
+	logFSM                        bool
+	statePath                     string
+	blockchainType                string
+	peerAddresses                 string
+	declAddr                      string
+	nodeName                      string
+	cfgPath                       string
+	apiAddr                       string
+	apiKey                        string
+	apiMaxConnections             int
+	rateLimiterOptions            string
+	grpcAddr                      string
+	grpcAPIMaxConnections         int
+	enableMetaMaskAPI             bool
+	enableMetaMaskAPILog          bool
+	enableGrpcAPI                 bool
+	blackListResidenceTime        time.Duration
+	buildExtendedAPI              bool
+	serveExtendedAPI              bool
+	buildStateHashes              bool
+	bindAddress                   string
+	disableOutgoingConnections    bool
+	minerVoteFeatures             string
+	disableBloomFilter            bool
+	reward                        int64
+	obsolescencePeriod            time.Duration
+	walletPath                    string
+	walletPassword                string
+	limitAllConnections           uint
+	minPeersMining                int
+	disableMiner                  bool
+	profiler                      bool
+	prometheus                    string
+	metricsID                     int
+	metricsURL                    string
+	dropPeers                     bool
+	dbFileDescriptors             uint
+	newConnectionsLimit           int
+	disableNTP                    bool
+	microblockInterval            time.Duration
+	enableLightMode               bool
+	enableBlockchainUpdatesPlugin bool
 }
 
 var errConfigNotParsed = stderrs.New("config is not parsed")
@@ -169,6 +172,7 @@ func (c *config) logParameters() {
 	zap.S().Debugf("disable-ntp: %t", c.disableNTP)
 	zap.S().Debugf("microblock-interval: %s", c.microblockInterval)
 	zap.S().Debugf("enable-light-mode: %t", c.enableLightMode)
+	zap.S().Debugf("enable-blockchain-updates-plugin: %t", c.enableBlockchainUpdatesPlugin)
 }
 
 func (c *config) parse() {
@@ -266,6 +270,9 @@ func (c *config) parse() {
 		"Interval between microblocks.")
 	flag.BoolVar(&c.enableLightMode, "enable-light-mode", false,
 		"Start node in light mode")
+
+	flag.BoolVar(&c.enableBlockchainUpdatesPlugin, "enable-blockchain-info", false,
+		"Turn on blockchain updates plugin")
 	flag.Parse()
 	c.logLevel = *l
 }
@@ -393,7 +400,17 @@ func runNode(ctx context.Context, nc *config) (_ io.Closer, retErr error) {
 		return nil, errors.Wrap(err, "failed to create state parameters")
 	}
 
-	st, err := state.NewState(path, true, params, cfg, nc.enableLightMode)
+	var bUpdatesExtension *blockchaininfo.BlockchainUpdatesExtension
+	if nc.enableBlockchainUpdatesPlugin {
+		var bUErr error
+		bUpdatesExtension, bUErr = runBlockchainUpdatesPlugin(ctx, cfg)
+		if bUErr != nil {
+			return nil, errors.Wrap(bUErr, "failed to run blockchain updates plugin")
+		}
+	}
+
+	// Send updatesChannel into BlockchainSettings. Write updates into this channel
+	st, err := state.NewState(path, true, params, cfg, nc.enableLightMode, bUpdatesExtension)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to initialize node's state")
 	}
@@ -795,6 +812,33 @@ func runAPIs(
 	return nil
 }
 
+func runBlockchainUpdatesPlugin(
+	ctx context.Context,
+	cfg *settings.BlockchainSettings,
+) (*blockchaininfo.BlockchainUpdatesExtension, error) {
+	const l2ContractAddr = "3Msx4Aq69zWUKy4d1wyKnQ4ofzEDAfv5Ngf"
+
+	l2address, cnvrtErr := proto.NewAddressFromString(l2ContractAddr)
+	if cnvrtErr != nil {
+		return nil, errors.Wrapf(cnvrtErr, "failed to convert L2 contract address %q", l2ContractAddr)
+	}
+
+	bUpdatesExtensionState := blockchaininfo.NewBUpdatesExtensionState(
+		blockchaininfo.StoreBlocksLimit,
+		cfg.AddressSchemeCharacter,
+	)
+
+	updatesChannel := make(chan blockchaininfo.BUpdatesInfo)
+	go func() {
+		err := bUpdatesExtensionState.RunBlockchainUpdatesPublisher(ctx, updatesChannel, cfg.AddressSchemeCharacter)
+		if err != nil {
+			zap.S().Fatalf("Failed to run blockchain updates publisher: %v", err)
+		}
+	}()
+
+	return blockchaininfo.NewBlockchainUpdatesExtension(ctx, l2address, updatesChannel), nil
+}
+
 func FromArgs(scheme proto.Scheme, c *config) func(s *settings.NodeSettings) error {
 	return func(s *settings.NodeSettings) error {
 		s.DeclaredAddr = c.declAddr

cmd/rollback/main.go

@@ -77,7 +77,7 @@ func main() {
 	params.BuildStateHashes = *buildStateHashes
 	params.StoreExtendedApiData = *buildExtendedAPI
 
-	s, err := state.NewState(*statePath, true, params, cfg, false)
+	s, err := state.NewState(*statePath, true, params, cfg, false, nil)
 	if err != nil {
 		zap.S().Error(err)
 		return