slave(M3): boot a slave's shards and run until interrupted

cmd/slave/README.md

@@ -14,6 +14,50 @@ make slave
 This installs the binary to `./build/bin/slave` (the same convention as `geth`).
 The commands below are run from the repo root so the relative config paths resolve.
 
+## Running a slave
+
+The default action boots every shard assigned to `--node_id` and runs until
+interrupted — a drop-in for how pyquarkchain's `cluster.py` starts a slave:
+
+```
+./build/bin/slave --cluster_config ./qkc/config/singularity/devnet.json --node_id S0
+```
+
+```
+INFO [..] slave booting               node_id=S0
+INFO [..] genesis committed           shard=0x00000001 genesis=661b12..792667
+INFO [..] shard started               shard=0x00000001 genesis=661b12..792667 head=0
+INFO [..] genesis committed           shard=0x00040001 genesis=c80ea9..5496d2
+INFO [..] shard started               shard=0x00040001 genesis=c80ea9..5496d2 head=0
+INFO [..] slave running               node_id=S0 shards=2
+```
+
+Each owned shard gets an isolated chaindb under
+`{DB_PATH_ROOT}/shard-0x{full_shard_id}/` (relative to the working directory),
+stands on its QuarkChain minor genesis block, and reports that block's hash and
+its head height. An empty `DB_PATH_ROOT` is pyquarkchain's mem-db mode
+(`use_mem_db`): every shard runs on an ephemeral in-memory database and nothing
+is written to disk.
+
+`^C` (or SIGTERM) shuts every shard down cleanly and exits 0; a second signal
+force-quits. The handler is installed before any resource opens, so a signal that
+lands mid-boot is honored as soon as boot settles.
+
+Rerunning against the same datadir revalidates each shard's stored genesis block
+against the config and logs `existing genesis validated` instead of
+`genesis committed`. A datadir initialized from a different config is refused
+rather than reused, and the process exits non-zero:
+
+```
+slave S0: shard 0x00000001: stored genesis 0x661b12…792667 does not match config
+genesis 0x04493a…8a7a0f (db ./qkc-data/devnet/shard-0x00000001) — cluster config
+changed since initialization
+```
+
+Only geth's logging and file-based profiling debug flags are exposed. The debug
+flags that would open a socket — the `--pprof` HTTP server and `--pyroscope.*`
+push — are deliberately not registered, keeping the process free of network I/O.
+
 ## Subcommands
 
 ### `slave config`

cmd/slave/main.go

@@ -81,6 +81,7 @@ func newApp() *cli.App {
 		debug.Exit()
 		return nil
 	}
+	app.Action = runSlave
 	app.Commands = []*cli.Command{
 		configCommand,
 		genesisCommand,

cmd/slave/run.go

@@ -0,0 +1,70 @@
+// Copyright 2026-2027, QuarkChain.
+
+package main
+
+import (
+	"context"
+	"fmt"
+	"os/signal"
+	"syscall"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/qkc"
+	"github.com/ethereum/go-ethereum/qkc/config"
+	"github.com/ethereum/go-ethereum/qkc/shard"
+	"github.com/ethereum/go-ethereum/qkc/slave"
+	"github.com/urfave/cli/v2"
+)
+
+// runSlave is the default action: boot every shard owned by --node_id and run
+// until interrupted, then shut down cleanly.
+func runSlave(ctx *cli.Context) error {
+	// Catch SIGINT/SIGTERM before any resource opens, so a signal that lands
+	// during config load or shard boot still funnels into the blocking Stop()
+	// below instead of the OS default termination. The watcher restores default
+	// signal handling the moment the first signal lands, so a second signal
+	// force-quits the process even while boot or the shutdown drain is still
+	// in flight.
+	sigCtx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+	go func() {
+		<-sigCtx.Done()
+		stop()
+	}()
+
+	log.Info("slave booting", "node_id", ctx.String(nodeIDFlag.Name))
+	cfg, err := loadClusterConfig(ctx)
+	if err != nil {
+		return err
+	}
+	backend, err := bootSlave(cfg, ctx.String(nodeIDFlag.Name))
+	if err != nil {
+		return err
+	}
+
+	if sigCtx.Err() == nil {
+		log.Info("slave running", "node_id", backend.ID, "shards", len(backend.Shards()))
+		<-sigCtx.Done()
+	}
+	log.Info("slave shutting down", "node_id", backend.ID)
+	return backend.Stop()
+}
+
+// bootSlave narrows the cluster config to the slave identified by nodeID, derives
+// the root genesis, and boots every owned shard.
+func bootSlave(cfg *config.ClusterConfig, nodeID string) (*slave.SlaveBackend, error) {
+	if nodeID == "" {
+		return nil, fmt.Errorf("--%s is required (e.g. S0)", nodeIDFlag.Name)
+	}
+	slaveCtx, err := cfg.ResolveSlave(nodeID)
+	if err != nil {
+		return nil, err
+	}
+	root, err := qkc.CreateRootBlock(cfg.Quarkchain)
+	if err != nil {
+		return nil, err
+	}
+	// TODO: inject the real chain service (the geth-core shard chain) here once it
+	// exists; Options{} uses the stub.
+	return slave.New(slaveCtx, root, shard.Options{})
+}

cmd/slave/run_test.go

@@ -0,0 +1,173 @@
+// Copyright 2026-2027, QuarkChain.
+
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/internal/reexec"
+	"github.com/ethereum/go-ethereum/qkc/config"
+	"go.uber.org/goleak"
+)
+
+func TestMain(m *testing.M) {
+	reexec.Register("slave-test", main)
+	if reexec.Init() {
+		return
+	}
+	// No ignore list: with metrics disabled (no --metrics flag registered at
+	// all) neither geth nor pebble leaves a background goroutine behind after
+	// Stop(). When bootSlave wires the real chain, fix its Stop path instead of
+	// allowlisting slave-owned goroutines here.
+	goleak.VerifyTestMain(m)
+}
+
+// writeFixtureWithDBRoot copies a fixture with its DB_PATH_ROOT redirected to
+// dbRoot and returns the rewritten config's path.
+func writeFixtureWithDBRoot(t *testing.T, path, dbRoot string) string {
+	t.Helper()
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read fixture: %v", err)
+	}
+	var doc map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &doc); err != nil {
+		t.Fatalf("unmarshal fixture: %v", err)
+	}
+	doc["DB_PATH_ROOT"], _ = json.Marshal(dbRoot)
+	rewritten, err := json.Marshal(doc)
+	if err != nil {
+		t.Fatalf("marshal fixture: %v", err)
+	}
+	tmpPath := filepath.Join(t.TempDir(), "cluster_config.json")
+	if err := os.WriteFile(tmpPath, rewritten, 0o644); err != nil {
+		t.Fatalf("write fixture: %v", err)
+	}
+	return tmpPath
+}
+
+// writeFixtureWithTempDBRoot is writeFixtureWithDBRoot into a fresh t.TempDir(),
+// so booting from it never writes into the repo tree.
+func writeFixtureWithTempDBRoot(t *testing.T, path string) string {
+	t.Helper()
+	return writeFixtureWithDBRoot(t, path, t.TempDir())
+}
+
+func loadFixtureWithTempDBRoot(t *testing.T, path string) *config.ClusterConfig {
+	t.Helper()
+	cfg, err := config.LoadClusterConfig(writeFixtureWithTempDBRoot(t, path))
+	if err != nil {
+		t.Fatalf("load rewritten fixture: %v", err)
+	}
+	return cfg
+}
+
+// TestBootSlave drives the default action's boot pipeline end to end from each
+// real network config: resolve S0, derive the root genesis, start both owned
+// shards, stop, and boot again from the same datadir.
+func TestBootSlave(t *testing.T) {
+	for _, f := range fixtures {
+		t.Run(f.name, func(t *testing.T) {
+			cfg := loadFixtureWithTempDBRoot(t, f.path)
+
+			backend, err := bootSlave(cfg, "S0")
+			if err != nil {
+				t.Fatalf("bootSlave: %v", err)
+			}
+			if got := len(backend.Shards()); got != 2 {
+				t.Errorf("booted %d shards, want 2", got)
+			}
+			if err := backend.Stop(); err != nil {
+				t.Fatalf("Stop: %v", err)
+			}
+
+			backend, err = bootSlave(cfg, "S0")
+			if err != nil {
+				t.Fatalf("bootSlave(reopen): %v", err)
+			}
+			if err := backend.Stop(); err != nil {
+				t.Fatalf("Stop(reopen): %v", err)
+			}
+		})
+	}
+}
+
+func TestBootSlaveRejectsBadNodeID(t *testing.T) {
+	cfg := loadFixtureWithTempDBRoot(t, fixtures[0].path)
+	if _, err := bootSlave(cfg, ""); err == nil || !strings.Contains(err.Error(), "--node_id is required") {
+		t.Errorf("bootSlave(\"\") err = %v, want required-flag error", err)
+	}
+	if _, err := bootSlave(cfg, "S9"); err == nil || !strings.Contains(err.Error(), `unknown node id "S9"`) {
+		t.Errorf("bootSlave(S9) err = %v, want unknown node id", err)
+	}
+}
+
+// TestRunHonorsSignalDuringStartup sends SIGTERM as soon as the run action
+// reports its signal handler installed ("slave booting"), while shard boot is
+// typically still in flight. Whichever window the signal actually lands in,
+// the process must exit 0 through the clean shutdown path and leave a datadir
+// that reopens without complaint.
+func TestRunHonorsSignalDuringStartup(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("POSIX signals are not available on windows")
+	}
+	cfgPath := writeFixtureWithTempDBRoot(t, fixtures[0].path)
+
+	cmd := exec.Command(reexec.Self(), "--cluster_config", cfgPath, "--node_id", "S0")
+	cmd.Args[0] = "slave-test"
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		t.Fatalf("stderr pipe: %v", err)
+	}
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	watchdog := time.AfterFunc(30*time.Second, func() { cmd.Process.Kill() })
+	defer watchdog.Stop()
+
+	var out strings.Builder
+	signalled := false
+	scanner := bufio.NewScanner(stderr)
+	for scanner.Scan() {
+		line := scanner.Text()
+		out.WriteString(line)
+		out.WriteByte('\n')
+		if !signalled && strings.Contains(line, "slave booting") {
+			if err := cmd.Process.Signal(syscall.SIGTERM); err != nil {
+				t.Fatalf("SIGTERM: %v", err)
+			}
+			signalled = true
+		}
+	}
+	err = cmd.Wait()
+	if !signalled {
+		t.Fatalf("never saw %q in output:\n%s", "slave booting", out.String())
+	}
+	if err != nil {
+		t.Fatalf("slave exited with %v, want exit 0; output:\n%s", err, out.String())
+	}
+	if !strings.Contains(out.String(), "slave shutting down") {
+		t.Errorf("clean shutdown path not taken; output:\n%s", out.String())
+	}
+
+	cfg, err := config.LoadClusterConfig(cfgPath)
+	if err != nil {
+		t.Fatalf("reload config: %v", err)
+	}
+	backend, err := bootSlave(cfg, "S0")
+	if err != nil {
+		t.Fatalf("reopen datadir after SIGTERM: %v", err)
+	}
+	if err := backend.Stop(); err != nil {
+		t.Fatalf("Stop: %v", err)
+	}
+}

qkc/slave/backend.go

@@ -0,0 +1,90 @@
+// Copyright 2026-2027, QuarkChain.
+
+// Package slave runs one slave identity's shards inside a single process,
+// mirroring pyquarkchain's SlaveServer. The slave process itself owns no
+// database: all persistent state lives in the per-shard chaindbs, and the shard
+// registry is rebuilt from config on every boot. It performs no network I/O.
+package slave
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/qkc/account"
+	"github.com/ethereum/go-ethereum/qkc/config"
+	"github.com/ethereum/go-ethereum/qkc/shard"
+	"github.com/ethereum/go-ethereum/qkc/types"
+)
+
+// SlaveBackend hosts the shards owned by one slave identity, keyed by branch as
+// in pyquarkchain's SlaveServer.
+type SlaveBackend struct {
+	ID     string
+	shards map[account.Branch]*shard.Shard
+	order  []account.Branch // boot (config) order, for deterministic iteration and shutdown
+
+	stopOnce sync.Once
+	stopErr  error
+}
+
+// New boots every shard the context's slave owns, eagerly and in config order.
+// Eager construction is interim scaffolding: with no master in the cluster yet it
+// is the only way to bring the shards up.
+//
+// TODO: move shard creation behind the master's PING(root_tip) trigger, as
+// pyquarkchain does (slave.py:927), preserving rollback and blocking shutdown for
+// shards created after boot.
+//
+// On any shard failure the shards already started are stopped and their databases
+// closed before the error returns, so the datadir stays reopenable.
+func New(ctx *config.SlaveContext, rootGenesis *types.RootBlockHeader, opts shard.Options) (*SlaveBackend, error) {
+	b := &SlaveBackend{
+		ID:     ctx.ID,
+		shards: make(map[account.Branch]*shard.Shard, len(ctx.FullShardIDs())),
+	}
+	for _, id := range ctx.FullShardIDs() {
+		branch := account.NewBranch(id)
+		s, err := shard.New(ctx, branch, rootGenesis, ctx.DBPathRoot, opts)
+		if err != nil {
+			return nil, errors.Join(fmt.Errorf("slave %s: %w", b.ID, err), b.Stop())
+		}
+		b.shards[branch] = s
+		b.order = append(b.order, branch)
+
+		height, _ := s.Chain().Head()
+		log.Info("shard started", "shard", fmt.Sprintf("0x%08x", id), "genesis", s.Chain().GenesisHash(), "head", height)
+	}
+	return b, nil
+}
+
+// Shard returns the shard hosting branch, or nil if this slave does not own it.
+func (b *SlaveBackend) Shard(branch account.Branch) *shard.Shard {
+	return b.shards[branch]
+}
+
+// Shards returns the hosted shards in boot (config) order.
+func (b *SlaveBackend) Shards() []*shard.Shard {
+	out := make([]*shard.Shard, 0, len(b.order))
+	for _, branch := range b.order {
+		out = append(out, b.shards[branch])
+	}
+	return out
+}
+
+// Stop shuts every shard down — chain first, then database — and blocks until all
+// are stopped and their databases closed. It is idempotent; a per-shard failure is
+// collected rather than aborting the remaining shutdowns.
+func (b *SlaveBackend) Stop() error {
+	b.stopOnce.Do(func() {
+		var errs []error
+		for _, branch := range b.order {
+			if err := b.shards[branch].Stop(); err != nil {
+				errs = append(errs, fmt.Errorf("stop shard 0x%08x: %w", branch.GetFullShardID(), err))
+			}
+		}
+		b.stopErr = errors.Join(errs...)
+	})
+	return b.stopErr
+}