[Merged by Bors] - Add node merge tool

.github/workflows/dockerhub.yml

@@ -26,6 +26,7 @@ jobs:
       uses: actions/checkout@v4
     - name: Build docker images
       run: |
+        export VERSION="${{ github.ref_name }}"
         make dockerbuild-go
         make dockerbuild-bs
 

.github/workflows/release.yml

@@ -58,6 +58,10 @@ jobs:
         run: |
           make install
           make build VERSION=${{ github.ref_name }}
+      - name: Build node-merge
+        shell: bash
+        run: |
+          make node-merge VERSION=${{ github.ref_name }}
 
       - name: Create release archive
         shell: bash

Dockerfile

@@ -3,7 +3,7 @@ FROM ubuntu:22.04 AS linux
 ENV DEBIAN_FRONTEND noninteractive
 ENV SHELL /bin/bash
 ARG TZ=Etc/UTC
-ENV TZ $TZ
+ENV TZ=${TZ}
 USER root
 RUN set -ex \
    && apt-get update --fix-missing \
@@ -27,6 +27,8 @@ ENV LANGUAGE en_US.UTF-8
 ENV LC_ALL en_US.UTF-8
 
 FROM golang:1.21 as builder
+ARG VERSION=""
+ENV VERSION=${VERSION}
 RUN set -ex \
    && apt-get update --fix-missing \
    && apt-get install -qy --no-install-recommends \
@@ -48,8 +50,9 @@ RUN go mod download
 COPY . .
 
 # And compile the project
-RUN --mount=type=cache,id=build,target=/root/.cache/go-build make build
+RUN --mount=type=cache,id=build,target=/root/.cache/go-build make build VERSION=${VERSION}
 RUN --mount=type=cache,id=build,target=/root/.cache/go-build make gen-p2p-identity
+RUN --mount=type=cache,id=build,target=/root/.cache/go-build make merge-nodes VERSION=${VERSION}
 
 # In this last stage, we start from a fresh image, to reduce the image size and not ship the Go compiler in our production artifacts.
 FROM linux AS spacemesh
@@ -59,6 +62,7 @@ COPY --from=builder /src/build/go-spacemesh /bin/
 COPY --from=builder /src/build/service /bin/
 COPY --from=builder /src/build/libpost.so /bin/
 COPY --from=builder /src/build/gen-p2p-identity /bin/
+COPY --from=builder /src/build/merge-nodes /bin/
 
 ENTRYPOINT ["/bin/go-spacemesh"]
 EXPOSE 7513

Makefile

@@ -70,6 +70,10 @@ get-libs: get-postrs-lib get-postrs-service
 
 get-profiler: get-postrs-profiler
 
+merge-nodes:
+	cd cmd/merge-nodes ; go build -o $(BIN_DIR)$@$(EXE) -ldflags "-X main.version=${VERSION}" .
+.PHONY: merge-nodes
+
 gen-p2p-identity:
 	cd cmd/gen-p2p-identity ; go build -o $(BIN_DIR)$@$(EXE) .
 .PHONY: gen-p2p-identity
@@ -149,7 +153,10 @@ list-versions:
 .PHONY: list-versions
 
 dockerbuild-go:
-	DOCKER_BUILDKIT=1 docker build -t go-spacemesh:$(SHA) -t $(DOCKER_HUB)/$(DOCKER_IMAGE_REPO):$(DOCKER_IMAGE_VERSION) .
+	DOCKER_BUILDKIT=1 docker build \
+		--build-arg VERSION=${VERSION} \
+		-t go-spacemesh:$(SHA) \
+		-t $(DOCKER_HUB)/$(DOCKER_IMAGE_REPO):$(DOCKER_IMAGE_VERSION) .
 .PHONY: dockerbuild-go
 
 dockerpush: dockerbuild-go dockerpush-only

api/grpcserver/grpcserver_test.go

@@ -535,7 +535,7 @@ func TestNewLocalServer(t *testing.T) {
 			svc := NewNodeService(peerCounter, meshApi, genTime, syncer, "v0.0.0", "cafebabe")
 			grpcService, err := NewWithServices(cfg.PostListener, logger, cfg, []ServiceAPI{svc})
 			if tc.warn {
-				require.Equal(t, observedLogs.Len(), 1, "Expected a warning log")
+				require.Equal(t, 1, observedLogs.Len(), "Expected a warning log")
 				require.Equal(t, observedLogs.All()[0].Message, "unsecured grpc server is listening on a public IP address")
 				require.Equal(t, observedLogs.All()[0].ContextMap()["address"], tc.listener)
 				return

cmd/merge-nodes/internal/errors.go

@@ -0,0 +1,10 @@
+package internal
+
+import (
+	"errors"
+)
+
+var (
+	ErrSupervisedNode = errors.New("merging of supervised smeshing nodes is not supported")
+	ErrInvalidSchema  = errors.New("database has an invalid schema version")
+)

cmd/merge-nodes/internal/merge_action.go

@@ -0,0 +1,204 @@
+package internal
+
+import (
+	"context"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"slices"
+
+	"go.uber.org/zap"
+
+	"github.com/spacemeshos/go-spacemesh/signing"
+	"github.com/spacemeshos/go-spacemesh/sql"
+	"github.com/spacemeshos/go-spacemesh/sql/localsql"
+)
+
+const (
+	localDbFile = "local.sql"
+
+	keyDir                  = "identities"
+	supervisedIDKeyFileName = "local.key"
+)
+
+func MergeDBs(ctx context.Context, dbLog *zap.Logger, from, to string) error {
+	// Open the target database
+	var dstDB *localsql.Database
+	var err error
+	dstDB, err = openDB(dbLog, to)
+	switch {
+	case errors.Is(err, fs.ErrNotExist):
+		// target database does not exist, create it
+		dbLog.Info("target database does not exist, creating it", zap.String("path", to))
+		if err := os.MkdirAll(to, 0o700); err != nil {
+			return fmt.Errorf("create target directory: %w", err)
+		}
+		if err := os.MkdirAll(filepath.Join(to, keyDir), 0o700); err != nil {
+			return fmt.Errorf("create target key directory: %w", err)
+		}
+
+		dstDB, err = localsql.Open("file:"+filepath.Join(to, localDbFile),
+			sql.WithLogger(dbLog),
+		)
+		if err != nil {
+			return err
+		}
+		defer dstDB.Close()
+	case err != nil:
+		return fmt.Errorf("open target database: %w", err)
+	default:
+		defer dstDB.Close()
+		// target database exists, check if there is at least one key in the target key directory
+		// not named supervisedIDKeyFileName
+		if err := checkIdentities(dbLog, to); err != nil {
+			dbLog.Error("target appears to be a supervised node - only merging of remote smeshers is supported")
+			return err
+		}
+	}
+
+	// Open the source database
+	srcDB, err := openDB(dbLog, from)
+	if err != nil {
+		return fmt.Errorf("open source database: %w", err)
+	}
+	if err := srcDB.Close(); err != nil {
+		return fmt.Errorf("close source database: %w", err)
+	}
+
+	if err := checkIdentities(dbLog, from); err != nil {
+		dbLog.Error("source appears to be a supervised node - only merging of remote smeshers is supported")
+		return err
+	}
+
+	// copy files from `from` to `to`
+	dir := filepath.Join(from, keyDir)
+	err = filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return fmt.Errorf("failed to walk directory at %s: %w", path, err)
+		}
+
+		// skip subdirectories and files in them
+		if d.IsDir() && path != dir {
+			return fs.SkipDir
+		}
+
+		// skip files that are not identity files
+		if filepath.Ext(path) != ".key" {
+			return nil
+		}
+
+		signer, err := signing.NewEdSigner(
+			signing.FromFile(path),
+		)
+		if err != nil {
+			return fmt.Errorf("not a valid key file %s: %w", d.Name(), err)
+		}
+
+		dstPath := filepath.Join(to, keyDir, d.Name())
+		if _, err := os.Stat(dstPath); err == nil {
+			return fmt.Errorf("identity file %s already exists: %w", d.Name(), fs.ErrExist)
+		}
+
+		dst := make([]byte, hex.EncodedLen(len(signer.PrivateKey())))
+		hex.Encode(dst, signer.PrivateKey())
+		err = os.WriteFile(dstPath, dst, 0o600)
+		if err != nil {
+			return fmt.Errorf("failed to write identity file: %w", err)
+		}
+
+		dbLog.Info("copied identity",
+			zap.String("name", d.Name()),
+		)
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	dbLog.Info("merging databases", zap.String("from", from), zap.String("to", to))
+	err = dstDB.WithTx(ctx, func(tx *sql.Tx) error {
+		enc := func(stmt *sql.Statement) {
+			stmt.BindText(1, filepath.Join(from, localDbFile))
+		}
+		if _, err := tx.Exec("ATTACH DATABASE ?1 AS srcDB;", enc, nil); err != nil {
+			return fmt.Errorf("attach source database: %w", err)
+		}
+		if _, err := tx.Exec("INSERT INTO main.initial_post SELECT * FROM srcDB.initial_post;", nil, nil); err != nil {
+			return fmt.Errorf("merge initial_post: %w", err)
+		}
+		if _, err := tx.Exec("INSERT INTO main.challenge SELECT * FROM srcDB.challenge;", nil, nil); err != nil {
+			return fmt.Errorf("merge challenge: %w", err)
+		}
+		if _, err := tx.Exec(
+			"INSERT INTO main.poet_registration SELECT * FROM srcDB.poet_registration;", nil, nil,
+		); err != nil {
+			return fmt.Errorf("merge poet_registration: %w", err)
+		}
+		if _, err := tx.Exec("INSERT INTO main.nipost SELECT * FROM srcDB.nipost;", nil, nil); err != nil {
+			return fmt.Errorf("merge nipost: %w", err)
+		}
+		return nil
+	})
+	if err != nil {
+		return fmt.Errorf("start transaction: %w", err)
+	}
+
+	if err := dstDB.Close(); err != nil {
+		return fmt.Errorf("close target database: %w", err)
+	}
+	return nil
+}
+
+func openDB(dbLog *zap.Logger, path string) (*localsql.Database, error) {
+	dbPath := filepath.Join(path, localDbFile)
+	if _, err := os.Stat(dbPath); err != nil {
+		return nil, fmt.Errorf("open database %s: %w", dbPath, err)
+	}
+
+	migrations, err := sql.LocalMigrations()
+	if err != nil {
+		return nil, fmt.Errorf("get local migrations: %w", err)
+	}
+
+	db, err := localsql.Open("file:"+dbPath,
+		sql.WithLogger(dbLog),
+		sql.WithMigrations([]sql.Migration{}), // do not migrate database when opening
+	)
+	if err != nil {
+		return nil, fmt.Errorf("open source database %s: %w", dbPath, err)
+	}
+
+	// check if the source database has the right schema
+	var version int
+	_, err = db.Exec("PRAGMA user_version;", nil, func(stmt *sql.Statement) bool {
+		version = stmt.ColumnInt(0)
+		return true
+	})
+	if err != nil {
+		return nil, fmt.Errorf("get source database schema for %s: %w", dbPath, err)
+	}
+	if version != len(migrations) {
+		db.Close()
+		return nil, ErrInvalidSchema
+	}
+	return db, nil
+}
+
+func checkIdentities(dbLog *zap.Logger, path string) error {
+	dir := filepath.Join(path, keyDir)
+	if err := os.MkdirAll(dir, 0o700); err != nil {
+		return err
+	}
+
+	files, err := os.ReadDir(dir)
+	if err != nil {
+		return fmt.Errorf("read target key directory: %w", err)
+	}
+	if slices.ContainsFunc(files, func(e fs.DirEntry) bool { return e.Name() == supervisedIDKeyFileName }) {
+		return ErrSupervisedNode
+	}
+	return nil
+}