feat(talis): add dedicated encoder instances for fibre-txsim

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: walldiss

tools/talis/add.go

@@ -41,6 +41,17 @@ func addCmd() *cobra.Command {
 						return fmt.Errorf("unknown provider %q (supported: digitalocean, googlecloud)", provider)
 					}
 				}
+			case "encoder":
+				for i := 0; i < count; i++ {
+					switch provider {
+					case "digitalocean":
+						cfg = cfg.WithDigitalOceanEncoder(region)
+					case "googlecloud":
+						cfg = cfg.WithGoogleCloudEncoder(region)
+					default:
+						return fmt.Errorf("unknown provider %q (supported: digitalocean, googlecloud)", provider)
+					}
+				}
 			case "bridge":
 				log.Println("bridges are not yet supported")
 				return nil
@@ -58,7 +69,7 @@ func addCmd() *cobra.Command {
 	cmd.Flags().StringVarP(&rootDir, "directory", "d", ".", "root directory in which to initialize")
 	cmd.Flags().IntVarP(&count, "count", "c", 0, "Number of nodes to deploy")
 	_ = cmd.MarkFlagRequired("count")
-	cmd.Flags().StringVarP(&nodeType, "type", "t", "", "Type of the node (validator, bridge, light)")
+	cmd.Flags().StringVarP(&nodeType, "type", "t", "", "Type of the node (validator, encoder, bridge, light)")
 	_ = cmd.MarkFlagRequired("type")
 	cmd.Flags().StringVarP(&provider, "provider", "p", "digitalocean", "Provider for the node (digitalocean, googlecloud)")
 	cmd.Flags().StringVarP(&region, "region", "r", "random", "the region to deploy the instance in (random if blank)")

tools/talis/client.go

@@ -77,7 +77,8 @@ func NewDOClient(cfg Config) (*DOClient, error) {
 
 func (c *DOClient) Up(ctx context.Context, workers int) error {
 	insts := make([]Instance, 0)
-	for _, v := range append(c.cfg.Validators, c.cfg.Observability...) {
+	allInstances := append(append(c.cfg.Validators, c.cfg.Observability...), c.cfg.Encoders...)
+	for _, v := range allInstances {
 		if v.Provider != DigitalOcean {
 			log.Println("unexpectedly skipping instance since only DO is supported", v.Name, "in region", v.Region)
 			continue
@@ -149,7 +150,8 @@ func (c *DOClient) countRunningDroplets(ctx context.Context) (int, error) {
 
 func (c *DOClient) Down(ctx context.Context, workers int) error {
 	insts := make([]Instance, 0)
-	for _, v := range append(c.cfg.Validators, c.cfg.Observability...) {
+	allInstances := append(append(c.cfg.Validators, c.cfg.Observability...), c.cfg.Encoders...)
+	for _, v := range allInstances {
 		if v.Provider != DigitalOcean {
 			log.Println("unexpectedly skipping instance since only DO is supported", v.Name, "in region", v.Region)
 			continue

tools/talis/config.go

@@ -21,13 +21,16 @@ const (
 	Light NodeType = "light"
 	// Observability represents a observability monitoring node for Prometheus/Grafana.
 	Observability NodeType = "observability"
+	// Encoder represents a dedicated fibre-txsim encoder node.
+	Encoder NodeType = "encoder"
 )
 
 var (
 	valCount           = atomic.Uint32{}
 	nodeCount          = atomic.Uint32{}
 	lightCount         = atomic.Uint32{}
 	observabilityCount = atomic.Uint32{}
+	encoderCount       = atomic.Uint32{}
 )
 
 // NodeName returns the name of the node based on its type and index. The
@@ -44,6 +47,8 @@ func NodeName(nodeType NodeType) string {
 		index = int(lightCount.Add(1)) - 1
 	case Observability:
 		index = int(observabilityCount.Add(1)) - 1
+	case Encoder:
+		index = int(encoderCount.Add(1)) - 1
 	default:
 		panic(fmt.Sprintf("unknown node type: %s", nodeType))
 	}
@@ -120,7 +125,7 @@ func ExperimentTag(nodeType NodeType, index int, experimentID, chainID string) s
 
 func GetExperimentTag(tags []string) string {
 	for _, tag := range tags {
-		if strings.HasPrefix(tag, "validator-") || strings.HasPrefix(tag, "bridge-") || strings.HasPrefix(tag, "light-") || strings.HasPrefix(tag, "observability-") {
+		if strings.HasPrefix(tag, "validator-") || strings.HasPrefix(tag, "bridge-") || strings.HasPrefix(tag, "light-") || strings.HasPrefix(tag, "observability-") || strings.HasPrefix(tag, "encoder-") {
 			return tag
 		}
 	}
@@ -133,6 +138,7 @@ type Config struct {
 	Bridges       []Instance `json:"bridges,omitempty"`
 	Lights        []Instance `json:"lights,omitempty"`
 	Observability []Instance `json:"observability,omitempty"`
+	Encoders      []Instance `json:"encoders,omitempty"`
 
 	// ChainID is the chain ID of the network. This is used to identify the
 	// network and is also used as the chain ID of the network. It is
@@ -164,6 +170,7 @@ func NewConfig(experiment, chainID string) Config {
 		Bridges:       []Instance{},
 		Lights:        []Instance{},
 		Observability: []Instance{},
+		Encoders:      []Instance{},
 		Experiment:    experiment,
 		ChainID:       TalisChainID(chainID),
 		S3Config: S3Config{
@@ -230,6 +237,18 @@ func (cfg Config) WithGoogleCloudObservability(region string) Config {
 	return cfg
 }
 
+func (cfg Config) WithDigitalOceanEncoder(region string) Config {
+	i := NewDigitalOceanEncoder(region).WithExperiment(cfg.Experiment, cfg.ChainID)
+	cfg.Encoders = append(cfg.Encoders, i)
+	return cfg
+}
+
+func (cfg Config) WithGoogleCloudEncoder(region string) Config {
+	i := NewGoogleCloudEncoder(region).WithExperiment(cfg.Experiment, cfg.ChainID)
+	cfg.Encoders = append(cfg.Encoders, i)
+	return cfg
+}
+
 func (cfg Config) WithChainID(chainID string) Config {
 	cfg.ChainID = TalisChainID(chainID)
 	return cfg
@@ -306,5 +325,12 @@ func (cfg Config) UpdateInstance(name, publicIP, privateIP string) (Config, erro
 			return cfg, nil
 		}
 	}
+	for i := range cfg.Encoders {
+		if cfg.Encoders[i].Name == name {
+			cfg.Encoders[i].PublicIP = publicIP
+			cfg.Encoders[i].PrivateIP = privateIP
+			return cfg, nil
+		}
+	}
 	return cfg, fmt.Errorf("instance %s not found", name)
 }

tools/talis/deployment.go

@@ -127,15 +127,21 @@ func deployCmd() *cobra.Command {
 					}
 					log.Printf("continuing despite validator deployment errors: %v", err)
 				}
-				return deployObservabilityIfConfigured(cmd.Context(), cfg, rootDir, SSHKeyPath, directUpload)
+				if err := deployObservabilityIfConfigured(cmd.Context(), cfg, rootDir, SSHKeyPath, directUpload); err != nil {
+					return err
+				}
+				return deployEncodersIfConfigured(cmd.Context(), cfg, rootDir, SSHKeyPath, workers)
 			}
 			if err := deployPayloadViaS3(cmd.Context(), rootDir, cfg.Validators, tarPath, SSHKeyPath, "/root", "payload/validator_init.sh", 7*time.Minute, cfg.S3Config, workers); err != nil {
 				if !ignoreFailed {
 					return err
 				}
 				log.Printf("continuing despite validator deployment errors: %v", err)
 			}
-			return deployObservabilityIfConfigured(cmd.Context(), cfg, rootDir, SSHKeyPath, directUpload)
+			if err := deployObservabilityIfConfigured(cmd.Context(), cfg, rootDir, SSHKeyPath, directUpload); err != nil {
+				return err
+			}
+			return deployEncodersIfConfigured(cmd.Context(), cfg, rootDir, SSHKeyPath, workers)
 		},
 	}
 
@@ -185,6 +191,35 @@ func deployObservabilityIfConfigured(ctx context.Context, cfg Config, rootDir, s
 	return nil
 }
 
+// deployEncodersIfConfigured creates a lightweight encoder-payload tar and deploys
+// it to all configured encoder instances via S3.
+func deployEncodersIfConfigured(ctx context.Context, cfg Config, rootDir, sshKeyPath string, workers int) error {
+	if len(cfg.Encoders) == 0 {
+		return nil
+	}
+
+	encoderPayloadDir := filepath.Join(rootDir, "encoder-payload")
+	if _, err := os.Stat(encoderPayloadDir); os.IsNotExist(err) {
+		return fmt.Errorf("encoder-payload directory not found — run 'talis genesis' first")
+	}
+
+	encoderTarPath := filepath.Join(rootDir, "encoder-payload.tar.gz")
+	log.Printf("Compressing encoder payload to %s\n", encoderTarPath)
+	tarCmd := exec.Command("tar", "-czf", encoderTarPath, "-C", rootDir, "encoder-payload")
+	tarCmd.Env = append(os.Environ(), "COPYFILE_DISABLE=1")
+	if output, err := tarCmd.CombinedOutput(); err != nil {
+		return fmt.Errorf("failed to compress encoder payload: %w, output: %s", err, string(output))
+	}
+	log.Printf("Sending encoder payload to %d encoder(s)...\n", len(cfg.Encoders))
+
+	if err := deployPayloadViaS3(ctx, rootDir, cfg.Encoders, encoderTarPath, sshKeyPath, "/root", "encoder-payload/encoder_init.sh", 7*time.Minute, cfg.S3Config, workers); err != nil {
+		return fmt.Errorf("encoder deployment: %w", err)
+	}
+
+	log.Printf("Encoder deployment complete\n")
+	return nil
+}
+
 // printGrafanaInfo prints the Grafana URL and credentials after successful observability deployment.
 func printGrafanaInfo(node Instance, rootDir string) {
 	password := readGrafanaPassword(rootDir)

tools/talis/digital_ocean.go

@@ -16,6 +16,7 @@ import (
 
 const (
 	DODefaultValidatorSlug     = "c2-16vcpu-32gb"
+	DODefaultEncoderSlug       = "c2-8vcpu-16gb"
 	DODefaultObservabilitySlug = "s-2vcpu-4gb"
 	DODefaultImage             = "ubuntu-22-04-x64"
 	RandomRegion               = "random"
@@ -50,6 +51,17 @@ func NewDigitalOceanValidator(region string) Instance {
 	return i
 }
 
+func NewDigitalOceanEncoder(region string) Instance {
+	if region == "" || region == RandomRegion {
+		region = RandomDORegion()
+	}
+	i := NewBaseInstance(Encoder)
+	i.Provider = DigitalOcean
+	i.Slug = DODefaultEncoderSlug
+	i.Region = region
+	return i
+}
+
 func NewDigitalOceanObservability(region string) Instance {
 	if region == "" || region == RandomRegion {
 		region = RandomDORegion()
@@ -473,7 +485,7 @@ func checkForRunningDOExperiments(ctx context.Context, client *godo.Client, expe
 
 func hasExperimentTag(tags []string, experimentID, chainID string) bool {
 	for _, tag := range tags {
-		if (strings.HasPrefix(tag, "validator-") || strings.HasPrefix(tag, "bridge-") || strings.HasPrefix(tag, "light-")) &&
+		if (strings.HasPrefix(tag, "validator-") || strings.HasPrefix(tag, "bridge-") || strings.HasPrefix(tag, "light-") || strings.HasPrefix(tag, "encoder-")) &&
 			strings.Contains(tag, experimentID) && strings.Contains(tag, chainID) {
 			return true
 		}

tools/talis/fibre_setup.go

@@ -14,13 +14,14 @@ const SetupFibreSessionName = "setup-fibre"
 
 func setupFibreCmd() *cobra.Command {
 	var (
-		rootDir       string
-		SSHKeyPath    string
-		escrowAmount  string
-		fibrePort     int
-		fees          string
-		workers       int
-		fibreAccounts int
+		rootDir              string
+		SSHKeyPath           string
+		escrowAmount         string
+		fibrePort            int
+		fees                 string
+		workers              int
+		fibreAccounts        int
+		encoderFibreAccounts int
 	)
 
 	cmd := &cobra.Command{
@@ -100,7 +101,47 @@ func setupFibreCmd() *cobra.Command {
 			if err := waitForTmuxSessions(cfg.Validators, resolvedSSHKeyPath, SetupFibreSessionName, 10*time.Minute); err != nil {
 				return fmt.Errorf("waiting for setup-fibre sessions: %w", err)
 			}
-			fmt.Println("Done!")
+			fmt.Println("Validator setup done!")
+
+			// Deposit escrow for encoder accounts.
+			// Each encoder runs deposit-to-escrow from its own machine using its
+			// own keyring, broadcasting via the first validator's RPC endpoint.
+			if len(cfg.Encoders) > 0 && len(cfg.Validators) > 0 {
+				rpcNode := fmt.Sprintf("tcp://%s:26657", cfg.Validators[0].PublicIP)
+				fmt.Printf("Setting up escrow for %d encoder(s) via %s...\n", len(cfg.Encoders), rpcNode)
+
+				for _, enc := range cfg.Encoders {
+					encIndex := extractIndexFromName(enc.Name)
+					keyPrefix := fmt.Sprintf("enc%d", encIndex)
+					nAccounts := encoderFibreAccounts
+
+					var sb strings.Builder
+					for i := 0; i < nAccounts; i++ {
+						keyName := fmt.Sprintf("%s-%d", keyPrefix, i)
+						sb.WriteString(fmt.Sprintf(
+							"celestia-appd tx fibre deposit-to-escrow %s "+
+								"--from %s --keyring-backend=test --home .celestia-app "+
+								"--chain-id %s --fees %s --node %s --yes\n",
+							escrowAmount,
+							keyName,
+							cfg.ChainID, fees, rpcNode,
+						))
+					}
+
+					script := sb.String()
+					fmt.Printf("Running escrow deposits on encoder %s (%s) — %d accounts\n", enc.Name, enc.PublicIP, nAccounts)
+					if err := runScriptInTMux([]Instance{enc}, resolvedSSHKeyPath, script, SetupFibreSessionName, 30*time.Minute); err != nil {
+						return fmt.Errorf("encoder %s escrow setup: %w", enc.Name, err)
+					}
+				}
+
+				fmt.Printf("Waiting for encoder escrow deposits to complete...\n")
+				if err := waitForTmuxSessions(cfg.Encoders, resolvedSSHKeyPath, SetupFibreSessionName, 15*time.Minute); err != nil {
+					return fmt.Errorf("waiting for encoder setup-fibre sessions: %w", err)
+				}
+				fmt.Println("Encoder escrow setup done!")
+			}
+
 			return nil
 		},
 	}
@@ -112,6 +153,7 @@ func setupFibreCmd() *cobra.Command {
 	cmd.Flags().StringVar(&fees, "fees", "5000utia", "transaction fees")
 	cmd.Flags().IntVarP(&workers, "workers", "w", 10, "number of validators to set up in parallel")
 	cmd.Flags().IntVar(&fibreAccounts, "fibre-accounts", 100, "number of fibre worker accounts to deposit escrow for")
+	cmd.Flags().IntVar(&encoderFibreAccounts, "encoder-fibre-accounts", 100, "number of fibre worker accounts per encoder instance")
 
 	return cmd
 }