alpenthrilmonic

cmd/mithril/node/harmonic.go

@@ -0,0 +1,143 @@
+package node
+
+import (
+	"context"
+	"crypto/ed25519"
+	"fmt"
+
+	"github.com/Overclock-Validator/mithril/pkg/blockprod"
+	"github.com/Overclock-Validator/mithril/pkg/gossip"
+	"github.com/Overclock-Validator/mithril/pkg/harmonic"
+	blockenginepb "github.com/Overclock-Validator/mithril/pkg/harmonic/proto/blockengine"
+	"github.com/Overclock-Validator/mithril/pkg/mlog"
+	"github.com/Overclock-Validator/mithril/pkg/tpu"
+	solana "github.com/gagliardetto/solana-go"
+	"github.com/spf13/pflag"
+)
+
+// harmonicSettings is the validated [harmonic] section.
+type harmonicSettings struct {
+	enabled            bool
+	blockEngine        harmonic.Endpoint
+	relayer            harmonic.Endpoint
+	insecureSkipVerify bool
+	strategy           blockenginepb.SchedulingStrategy
+	workers            int
+	// tips lacks the vote account and identity, which come from [validator].
+	tips harmonic.TipConfig
+}
+
+// registerHarmonicFlags adds a flag for each [harmonic] key.
+func registerHarmonicFlags(flags *pflag.FlagSet) {
+	flags.Bool("harmonic-enabled", false, "Run leader slots from a Harmonic block engine (validator mode)")
+	flags.String("harmonic-block-engine-url", "", "Harmonic block engine URL (http:// or https://)")
+	flags.String("harmonic-block-engine-tls-domain", "", "TLS server name override for the block engine")
+	flags.String("harmonic-relayer-url", "", "Optional Harmonic relayer URL")
+	flags.String("harmonic-relayer-tls-domain", "", "TLS server name override for the relayer")
+	flags.Bool("harmonic-tls-cert-verify", true, "Verify Harmonic TLS certificates (false for development only)")
+	flags.String("harmonic-strategy", "fba", "Block engine scheduling strategy: fba, fifo or mrev")
+	flags.String("harmonic-tip-payment-program", "", "Tip payment program")
+	flags.String("harmonic-tip-distribution-program", "", "Tip distribution program")
+	flags.String("harmonic-merkle-root-upload-authority", "", "Merkle root upload authority for tip distribution accounts")
+	flags.Int64("harmonic-commission-bps", 0, "Validator commission on tips, in basis points (0-10000)")
+	flags.Int64("harmonic-num-workers", 8, "Threads executing block bundles that do not conflict in parallel")
+}
+
+// optionGetters resolve an option from its CLI flag when set, else from the
+// config file.
+type optionGetters struct {
+	getString func(cliKey, tomlKey string) string
+	getInt64  func(cliKey, tomlKey string) int64
+	getBool   func(cliKey, tomlKey string) bool
+}
+
+// loadHarmonicSettings reads [harmonic]. Harmonic is off unless enabled, and
+// then every key it needs must be valid.
+func loadHarmonicSettings(get optionGetters) (harmonicSettings, error) {
+	s := harmonicSettings{enabled: get.getBool("harmonic-enabled", "harmonic.enabled")}
+	if !s.enabled {
+		return s, nil
+	}
+	var err error
+	if s.blockEngine, err = harmonic.ParseEndpoint(get.getString("harmonic-block-engine-url", "harmonic.block_engine_url"),
+		get.getString("harmonic-block-engine-tls-domain", "harmonic.block_engine_tls_domain")); err != nil {
+		return s, fmt.Errorf("harmonic.block_engine_url: %w", err)
+	}
+	if url := get.getString("harmonic-relayer-url", "harmonic.relayer_url"); url != "" {
+		if s.relayer, err = harmonic.ParseEndpoint(url, get.getString("harmonic-relayer-tls-domain", "harmonic.relayer_tls_domain")); err != nil {
+			return s, fmt.Errorf("harmonic.relayer_url: %w", err)
+		}
+	}
+	s.insecureSkipVerify = !get.getBool("harmonic-tls-cert-verify", "harmonic.tls_cert_verify")
+	if s.strategy, err = harmonic.ParseStrategy(get.getString("harmonic-strategy", "harmonic.strategy")); err != nil {
+		return s, fmt.Errorf("harmonic.strategy: %w", err)
+	}
+	for _, key := range []struct {
+		flag, name string
+		dst        *solana.PublicKey
+	}{
+		{"harmonic-tip-payment-program", "harmonic.tip_payment_program", &s.tips.TipPaymentProgram},
+		{"harmonic-tip-distribution-program", "harmonic.tip_distribution_program", &s.tips.TipDistributionProgram},
+		{"harmonic-merkle-root-upload-authority", "harmonic.merkle_root_upload_authority", &s.tips.MerkleRootUploadAuthority},
+	} {
+		value := get.getString(key.flag, key.name)
+		if value == "" {
+			return s, fmt.Errorf("%s is required when harmonic.enabled=true", key.name)
+		}
+		if *key.dst, err = solana.PublicKeyFromBase58(value); err != nil {
+			return s, fmt.Errorf("%s: %w", key.name, err)
+		}
+	}
+	bps := get.getInt64("harmonic-commission-bps", "harmonic.commission_bps")
+	if bps < 0 || bps > 10_000 {
+		return s, fmt.Errorf("harmonic.commission_bps must be between 0 and 10000, got %d", bps)
+	}
+	s.tips.CommissionBps = uint16(bps)
+	workers := get.getInt64("harmonic-num-workers", "harmonic.num_workers")
+	if workers < 1 || workers > 64 {
+		return s, fmt.Errorf("harmonic.num_workers must be between 1 and 64, got %d", workers)
+	}
+	s.workers = int(workers)
+	return s, nil
+}
+
+// startHarmonic runs the Harmonic service until the returned stop function,
+// which waits for it to finish. The service's hooks must be given to the
+// leader loop with GateTPU set.
+func startHarmonic(ctx context.Context, s harmonicSettings, identity ed25519.PrivateKey, voteAccount solana.PublicKey,
+	tpuService *tpu.TPU, gossipClient *gossip.Client, localTPU gossip.TPUSockets, txV1Enabled func() bool,
+) (blockprod.LeaderHooks, func(), error) {
+	tips := s.tips
+	tips.VoteAccount = voteAccount
+	tips.Identity = solana.PrivateKey(identity)
+	service, err := harmonic.NewService(harmonic.ServiceConfig{
+		BlockEngine:        s.blockEngine,
+		Relayer:            s.relayer,
+		InsecureSkipVerify: s.insecureSkipVerify,
+		Strategy:           s.strategy,
+		Tips:               tips,
+		TPU:                tpuService,
+		Gossip:             gossipClient,
+		LocalTPU:           localTPU,
+		TxV1Enabled:        txV1Enabled,
+		Workers:            s.workers,
+	})
+	if err != nil {
+		return nil, nil, err
+	}
+	ctx, cancel := context.WithCancel(ctx)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		service.Run(ctx)
+	}()
+	relayer := "none"
+	if s.relayer.Target != "" {
+		relayer = s.relayer.Target
+	}
+	mlog.Log.Infof("Harmonic enabled: block engine %s, relayer %s, strategy %s, workers %d", s.blockEngine.Target, relayer, s.strategy, s.workers)
+	return service.Hooks(), func() {
+		cancel()
+		<-done
+	}, nil
+}

cmd/mithril/node/harmonic_test.go

@@ -0,0 +1,181 @@
+package node
+
+import (
+	"fmt"
+	"maps"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/Overclock-Validator/mithril/pkg/config"
+	blockenginepb "github.com/Overclock-Validator/mithril/pkg/harmonic/proto/blockengine"
+	"github.com/spf13/pflag"
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// flagsThenConfig resolves options as runLive does: a set flag wins over the
+// config file.
+func flagsThenConfig(flags *pflag.FlagSet) optionGetters {
+	changed := func(cliKey string) *pflag.Flag {
+		if f := flags.Lookup(cliKey); f != nil && f.Changed {
+			return f
+		}
+		return nil
+	}
+	return optionGetters{
+		getString: func(cliKey, tomlKey string) string {
+			if f := changed(cliKey); f != nil {
+				return f.Value.String()
+			}
+			return config.GetString(tomlKey)
+		},
+		getInt64: func(cliKey, tomlKey string) int64 {
+			if f := changed(cliKey); f != nil {
+				v, _ := strconv.ParseInt(f.Value.String(), 10, 64)
+				return v
+			}
+			return config.GetInt64(tomlKey)
+		},
+		getBool: func(cliKey, tomlKey string) bool {
+			return resolveBoolOption(flags.Lookup(cliKey), config.IsSet(tomlKey), config.GetBool(tomlKey))
+		},
+	}
+}
+
+// loadHarmonic loads a [harmonic] section of the given keys, whose values are
+// TOML literals, under the given flag values.
+func loadHarmonic(t *testing.T, values map[string]string, flagValues map[string]string) (harmonicSettings, error) {
+	t.Helper()
+	var toml strings.Builder
+	toml.WriteString("[harmonic]\n")
+	for key, value := range values {
+		fmt.Fprintf(&toml, "%s = %s\n", key, value)
+	}
+	viper.Reset()
+	t.Cleanup(viper.Reset)
+	config.ApplyDefaults(viper.GetViper())
+	viper.SetConfigType("toml")
+	require.NoError(t, viper.ReadConfig(strings.NewReader(toml.String())))
+	flags := pflag.NewFlagSet("run", pflag.ContinueOnError)
+	registerHarmonicFlags(flags)
+	for name, value := range flagValues {
+		require.NoError(t, flags.Set(name, value))
+	}
+	return loadHarmonicSettings(flagsThenConfig(flags))
+}
+
+func loadHarmonicTOML(t *testing.T, values map[string]string) (harmonicSettings, error) {
+	t.Helper()
+	return loadHarmonic(t, values, nil)
+}
+
+// enabledHarmonic is a minimal enabled [harmonic] section with overrides.
+func enabledHarmonic(overrides map[string]string) map[string]string {
+	values := map[string]string{
+		"enabled":                      "true",
+		"block_engine_url":             `"https://be.example.com"`,
+		"tip_payment_program":          `"T1pyyaTNZsKv2WcRAB8oVnk93mLJw2XzjtVYqCsaHqt"`,
+		"tip_distribution_program":     `"4R3gSG8BpU4t19KYj8CfnbtRpnT8gtk4dvTHxVRwc2r7"`,
+		"merkle_root_upload_authority": `"8F4jGUmxF36vQ6yabnsxX6AQVXdKBhs8kGSUuRKSg8Xt"`,
+	}
+	maps.Copy(values, overrides)
+	return values
+}
+
+func TestHarmonicIsOffByDefault(t *testing.T) {
+	s, err := loadHarmonicTOML(t, nil)
+	require.NoError(t, err)
+	assert.False(t, s.enabled)
+
+	// Nothing else is read while disabled.
+	s, err = loadHarmonicTOML(t, map[string]string{"block_engine_url": `"not a url"`})
+	require.NoError(t, err)
+	assert.False(t, s.enabled)
+}
+
+func TestHarmonicSettings(t *testing.T) {
+	s, err := loadHarmonicTOML(t, enabledHarmonic(nil))
+	require.NoError(t, err)
+	assert.True(t, s.enabled)
+	assert.Equal(t, "be.example.com:443", s.blockEngine.Target)
+	assert.True(t, s.blockEngine.TLS)
+	assert.Empty(t, s.relayer.Target, "the relayer is optional")
+	assert.False(t, s.insecureSkipVerify, "certificates are verified by default")
+	assert.Equal(t, blockenginepb.SchedulingStrategy_SCHEDULING_STRATEGY_FBA, s.strategy)
+	assert.Equal(t, "8F4jGUmxF36vQ6yabnsxX6AQVXdKBhs8kGSUuRKSg8Xt", s.tips.MerkleRootUploadAuthority.String())
+	assert.Zero(t, s.tips.CommissionBps)
+	assert.Equal(t, 8, s.workers, "eight workers by default")
+
+	s, err = loadHarmonicTOML(t, enabledHarmonic(map[string]string{
+		"relayer_url":     `"http://relayer.example.com:11226"`,
+		"tls_cert_verify": "false",
+		"strategy":        `"MREV"`,
+		"commission_bps":  "800",
+		"num_workers":     "4",
+	}))
+	require.NoError(t, err)
+	assert.Equal(t, "relayer.example.com:11226", s.relayer.Target)
+	assert.True(t, s.insecureSkipVerify)
+	assert.Equal(t, blockenginepb.SchedulingStrategy_SCHEDULING_STRATEGY_MREV, s.strategy)
+	assert.Equal(t, uint16(800), s.tips.CommissionBps)
+	assert.Equal(t, 4, s.workers)
+
+	for name, bad := range map[string]map[string]string{
+		"no block engine url":  {"block_engine_url": `""`},
+		"bad block engine url": {"block_engine_url": `"be.example.com"`},
+		"bad relayer url":      {"relayer_url": `"ftp://relayer.example.com"`},
+		"bad strategy":         {"strategy": `"fastest"`},
+		"no tip program":       {"tip_payment_program": `""`},
+		"bad authority":        {"merkle_root_upload_authority": `"not base58"`},
+		"commission over 100%": {"commission_bps": "10001"},
+		"negative commission":  {"commission_bps": "-1"},
+		"no workers":           {"num_workers": "0"},
+		"too many workers":     {"num_workers": "65"},
+	} {
+		_, err := loadHarmonicTOML(t, enabledHarmonic(bad))
+		assert.Errorf(t, err, "%s", name)
+	}
+}
+
+func TestHarmonicFlagsOverrideTheConfigFile(t *testing.T) {
+	require.NotNil(t, Run.Flags().Lookup("harmonic-enabled"))
+
+	// Every key from flags alone.
+	s, err := loadHarmonic(t, nil, map[string]string{
+		"harmonic-enabled":                      "true",
+		"harmonic-block-engine-url":             "https://be.example.com",
+		"harmonic-block-engine-tls-domain":      "be.internal",
+		"harmonic-relayer-url":                  "http://relayer.example.com:11226",
+		"harmonic-relayer-tls-domain":           "relayer.internal",
+		"harmonic-tls-cert-verify":              "false",
+		"harmonic-strategy":                     "fifo",
+		"harmonic-tip-payment-program":          "T1pyyaTNZsKv2WcRAB8oVnk93mLJw2XzjtVYqCsaHqt",
+		"harmonic-tip-distribution-program":     "4R3gSG8BpU4t19KYj8CfnbtRpnT8gtk4dvTHxVRwc2r7",
+		"harmonic-merkle-root-upload-authority": "8F4jGUmxF36vQ6yabnsxX6AQVXdKBhs8kGSUuRKSg8Xt",
+		"harmonic-commission-bps":               "500",
+		"harmonic-num-workers":                  "2",
+	})
+	require.NoError(t, err)
+	assert.True(t, s.enabled)
+	assert.Equal(t, "be.example.com:443", s.blockEngine.Target)
+	assert.Equal(t, "be.internal", s.blockEngine.ServerName)
+	assert.Equal(t, "relayer.example.com:11226", s.relayer.Target)
+	assert.Equal(t, "relayer.internal", s.relayer.ServerName)
+	assert.True(t, s.insecureSkipVerify)
+	assert.Equal(t, blockenginepb.SchedulingStrategy_SCHEDULING_STRATEGY_FIFO, s.strategy)
+	assert.Equal(t, "T1pyyaTNZsKv2WcRAB8oVnk93mLJw2XzjtVYqCsaHqt", s.tips.TipPaymentProgram.String())
+	assert.Equal(t, "4R3gSG8BpU4t19KYj8CfnbtRpnT8gtk4dvTHxVRwc2r7", s.tips.TipDistributionProgram.String())
+	assert.Equal(t, "8F4jGUmxF36vQ6yabnsxX6AQVXdKBhs8kGSUuRKSg8Xt", s.tips.MerkleRootUploadAuthority.String())
+	assert.Equal(t, uint16(500), s.tips.CommissionBps)
+	assert.Equal(t, 2, s.workers)
+
+	// A flag beats the config file, including an explicit false.
+	s, err = loadHarmonic(t, enabledHarmonic(map[string]string{"strategy": `"mrev"`}), map[string]string{"harmonic-strategy": "fba"})
+	require.NoError(t, err)
+	assert.Equal(t, blockenginepb.SchedulingStrategy_SCHEDULING_STRATEGY_FBA, s.strategy)
+	s, err = loadHarmonic(t, enabledHarmonic(nil), map[string]string{"harmonic-enabled": "false"})
+	require.NoError(t, err)
+	assert.False(t, s.enabled)
+}