cmd: fetch and broadcast all when not all are activated (#3489)

In the scenario where not all keys are activated, fetching and broadcasting from the obol API wasn't possible. This PR fixes this and only throws an error if a key hasn't exited.

category: bug
ticket: none

app/eth2wrap/eth2wrap.go

@@ -167,6 +167,7 @@ func provide[O any](ctx context.Context, clients []Client, fallbacks []Client,
 				if bestSelector != nil {
 					bestSelector.Increment(res.Input.client.Address())
 				}
+
 				return res.Output, nil
 			}
 

app/eth2wrap/eth2wrap_test.go

@@ -179,7 +179,6 @@ func TestFallback(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-
 			var calledMu sync.Mutex
 			primaryCalled := make([]bool, len(tt.primaryErrs))
 			fallbackCalled := make([]bool, len(tt.fallbackErrs))
@@ -199,6 +198,7 @@ func TestFallback(t *testing.T) {
 					calledMu.Lock()
 					primaryCalled[i] = true
 					calledMu.Unlock()
+
 					return returnValue, primaryErr
 				}
 				primaryClients[i] = cl
@@ -214,6 +214,7 @@ func TestFallback(t *testing.T) {
 					calledMu.Lock()
 					fallbackCalled[i] = true
 					calledMu.Unlock()
+
 					return returnValue, fallbackErr
 				}
 				fallbackClients[i] = cl

app/obolapi/api.go

@@ -150,6 +150,9 @@ func httpGet(ctx context.Context, url *url.URL, headers map[string]string) (io.R
 	}
 
 	if res.StatusCode/100 != 2 {
+		if res.StatusCode == http.StatusNotFound {
+			return nil, ErrNoExit
+		}
 		data, err := io.ReadAll(res.Body)
 		if err != nil {
 			return nil, errors.Wrap(err, "read POST response", z.Int("status", res.StatusCode))

cmd/exit_broadcast.go

@@ -169,6 +169,11 @@ func runBcastFullExit(ctx context.Context, config exitConfig) error {
 				valCtx := log.WithCtx(ctx, z.Str("validator_public_key", validatorPubKeyHex))
 				exit, err := fetchFullExit(valCtx, "", config, cl, identityKey, validatorPubKeyHex)
 				if err != nil {
+					if errors.Is(err, obolapi.ErrNoExit) {
+						log.Warn(ctx, fmt.Sprintf("full exit data from Obol API for validator %v not available (validator may not be activated)", validatorPubKeyHex), nil)
+						continue
+					}
+
 					return errors.Wrap(err, "fetch full exit for all validators from public key")
 				}
 				validatorPubKey, err := core.PubKeyFromBytes(validator.GetPublicKey())

cmd/exit_broadcast_internal_test.go

@@ -45,6 +45,16 @@ func Test_runBcastFullExitCmd(t *testing.T) {
 		t.Parallel()
 		testRunBcastFullExitCmdFlow(t, true, true)
 	})
+	t.Run("main flow from api for all with already exited validator", func(t *testing.T) {
+		t.Parallel()
+		testRunBcastFullExitCmdFlow(t, false, false)
+		testRunBcastFullExitCmdFlow(t, false, true)
+	})
+	t.Run("main flow from file for all with already exited validator", func(t *testing.T) {
+		t.Parallel()
+		testRunBcastFullExitCmdFlow(t, true, false)
+		testRunBcastFullExitCmdFlow(t, true, true)
+	})
 	t.Run("config", Test_runBcastFullExitCmd_Config)
 }
 
@@ -426,3 +436,107 @@ func TestExitBroadcastCLI(t *testing.T) {
 		})
 	}
 }
+
+func TestExitBcastFullExitNotActivated(t *testing.T) {
+	ctx := context.Background()
+
+	valAmt := 10
+	operatorAmt := 4
+
+	random := rand.New(rand.NewSource(int64(0)))
+
+	lock, enrs, keyShares := cluster.NewForT(
+		t,
+		valAmt,
+		operatorAmt,
+		operatorAmt,
+		0,
+		random,
+	)
+
+	root := t.TempDir()
+
+	operatorShares := make([][]tbls.PrivateKey, operatorAmt)
+
+	for opIdx := range operatorAmt {
+		for _, share := range keyShares {
+			operatorShares[opIdx] = append(operatorShares[opIdx], share[opIdx])
+		}
+	}
+
+	mBytes, err := json.Marshal(lock)
+	require.NoError(t, err)
+
+	validatorSet := beaconmock.ValidatorSet{}
+
+	for idx, v := range lock.Validators {
+		validatorSet[eth2p0.ValidatorIndex(idx)] = &eth2v1.Validator{
+			Index:   eth2p0.ValidatorIndex(idx),
+			Balance: 42,
+			Status:  eth2v1.ValidatorStateActiveOngoing,
+			Validator: &eth2p0.Validator{
+				PublicKey:             eth2p0.BLSPubKey(v.PubKey),
+				WithdrawalCredentials: testutil.RandomBytes32(),
+			},
+		}
+	}
+
+	beaconMock, err := beaconmock.New(
+		beaconmock.WithValidatorSet(validatorSet),
+		beaconmock.WithEndpoint("/eth/v1/beacon/pool/voluntary_exits", ""),
+	)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, beaconMock.Close())
+	}()
+
+	eth2Cl, err := eth2Client(ctx, []string{}, map[string]string{}, []string{beaconMock.Address()}, 10*time.Second, [4]byte(lock.ForkVersion))
+	require.NoError(t, err)
+
+	handler, addLockFiles := obolapimock.MockServer(false, eth2Cl)
+	srv := httptest.NewServer(handler)
+	addLockFiles(lock)
+	defer srv.Close()
+
+	writeAllLockData(t, root, operatorAmt, enrs, operatorShares, mBytes)
+
+	for idxOp := range operatorAmt {
+		// submit partial exits only for a subset
+		for idxVal := range valAmt / 2 {
+			baseDir := filepath.Join(root, fmt.Sprintf("op%d", idxOp))
+
+			config := exitConfig{
+				BeaconNodeEndpoints: []string{beaconMock.Address()},
+				PrivateKeyPath:      filepath.Join(baseDir, "charon-enr-private-key"),
+				ValidatorKeysDir:    filepath.Join(baseDir, "validator_keys"),
+				LockFilePath:        filepath.Join(baseDir, "cluster-lock.json"),
+				PublishAddress:      srv.URL,
+				ExitEpoch:           194048,
+				BeaconNodeTimeout:   30 * time.Second,
+				PublishTimeout:      10 * time.Second,
+			}
+
+			config.ValidatorPubkey = lock.Validators[idxVal].PublicKeyHex()
+
+			require.NoError(t, runSignPartialExit(ctx, config), "operator index: %v", idxOp)
+		}
+	}
+
+	baseDir := filepath.Join(root, fmt.Sprintf("op%d", 0))
+
+	config := exitConfig{
+		BeaconNodeEndpoints: []string{beaconMock.Address()},
+		PrivateKeyPath:      filepath.Join(baseDir, "charon-enr-private-key"),
+		ValidatorKeysDir:    filepath.Join(baseDir, "validator_keys"),
+		LockFilePath:        filepath.Join(baseDir, "cluster-lock.json"),
+		PublishAddress:      srv.URL,
+		ExitEpoch:           194048,
+		BeaconNodeTimeout:   30 * time.Second,
+		PublishTimeout:      10 * time.Second,
+	}
+
+	// exit all and do not fail on non-existing keys
+	config.All = true
+
+	require.NoError(t, runBcastFullExit(ctx, config))
+}

cmd/exit_fetch.go

@@ -128,7 +128,12 @@ func runFetchExit(ctx context.Context, config exitConfig) error {
 
 			fullExit, err := oAPI.GetFullExit(valCtx, validatorPubKeyHex, cl.GetInitialMutationHash(), shareIdx, identityKey)
 			if err != nil {
-				return errors.Wrap(err, "load full exit data from Obol API", z.Str("validator_public_key", validatorPubKeyHex))
+				if errors.Is(err, obolapi.ErrNoExit) {
+					log.Warn(ctx, fmt.Sprintf("full exit data from Obol API for validator %v not available (validator may not be activated)", validatorPubKeyHex), nil)
+					continue
+				}
+
+				return errors.Wrap(err, "broadcast full exit for all validators from public key")
 			}
 
 			err = writeExitToFile(valCtx, validatorPubKeyHex, config.FetchedExitPath, fullExit)

cmd/exit_fetch_internal_test.go

@@ -246,3 +246,116 @@ func TestExitFetchCLI(t *testing.T) {
 		})
 	}
 }
+
+func TestFetchExitFullFlowNotActivated(t *testing.T) {
+	ctx := context.Background()
+
+	valAmt := 10
+	operatorAmt := 4
+
+	random := rand.New(rand.NewSource(int64(0)))
+
+	lock, enrs, keyShares := cluster.NewForT(
+		t,
+		valAmt,
+		operatorAmt,
+		operatorAmt,
+		0,
+		random,
+	)
+
+	root := t.TempDir()
+
+	operatorShares := make([][]tbls.PrivateKey, operatorAmt)
+
+	for opIdx := range operatorAmt {
+		for _, share := range keyShares {
+			operatorShares[opIdx] = append(operatorShares[opIdx], share[opIdx])
+		}
+	}
+
+	mBytes, err := json.Marshal(lock)
+	require.NoError(t, err)
+
+	validatorSet := beaconmock.ValidatorSet{}
+
+	for idx, v := range lock.Validators {
+		validatorSet[eth2p0.ValidatorIndex(idx)] = &eth2v1.Validator{
+			Index:   eth2p0.ValidatorIndex(idx),
+			Balance: 42,
+			Status:  eth2v1.ValidatorStateActiveOngoing,
+			Validator: &eth2p0.Validator{
+				PublicKey:             eth2p0.BLSPubKey(v.PubKey),
+				WithdrawalCredentials: testutil.RandomBytes32(),
+			},
+		}
+	}
+
+	beaconMock, err := beaconmock.New(
+		beaconmock.WithValidatorSet(validatorSet),
+	)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, beaconMock.Close())
+	}()
+
+	eth2Cl, err := eth2Client(ctx, []string{}, map[string]string{}, []string{beaconMock.Address()}, 10*time.Second, [4]byte(lock.ForkVersion))
+	require.NoError(t, err)
+
+	handler, addLockFiles := obolapimock.MockServer(false, eth2Cl)
+	srv := httptest.NewServer(handler)
+	addLockFiles(lock)
+	defer srv.Close()
+
+	writeAllLockData(t, root, operatorAmt, enrs, operatorShares, mBytes)
+
+	for idxOp := range operatorAmt {
+		// submit partial exits only for a subset
+		for idxVal := range valAmt / 2 {
+			baseDir := filepath.Join(root, fmt.Sprintf("op%d", idxOp))
+
+			config := exitConfig{
+				BeaconNodeEndpoints: []string{beaconMock.Address()},
+				ValidatorPubkey:     lock.Validators[0].PublicKeyHex(),
+				PrivateKeyPath:      filepath.Join(baseDir, "charon-enr-private-key"),
+				ValidatorKeysDir:    filepath.Join(baseDir, "validator_keys"),
+				LockFilePath:        filepath.Join(baseDir, "cluster-lock.json"),
+				PublishAddress:      srv.URL,
+				ExitEpoch:           194048,
+				BeaconNodeTimeout:   30 * time.Second,
+				PublishTimeout:      10 * time.Second,
+			}
+			config.ValidatorPubkey = lock.Validators[idxVal].PublicKeyHex()
+
+			require.NoError(t, runSignPartialExit(ctx, config), "operator index: %v", idxOp)
+		}
+	}
+
+	baseDir := filepath.Join(root, fmt.Sprintf("op%d", 0))
+
+	config := exitConfig{
+		ValidatorPubkey: lock.Validators[0].PublicKeyHex(),
+		PrivateKeyPath:  filepath.Join(baseDir, "charon-enr-private-key"),
+		LockFilePath:    filepath.Join(baseDir, "cluster-lock.json"),
+		PublishAddress:  srv.URL,
+		FetchedExitPath: root,
+		PublishTimeout:  10 * time.Second,
+		All:             true,
+	}
+
+	require.NoError(t, runFetchExit(ctx, config))
+
+	for idxVal := range valAmt / 2 {
+		exitFilePath := filepath.Join(root, fmt.Sprintf("exit-%s.json", lock.Validators[idxVal].PublicKeyHex()))
+
+		require.FileExists(t, exitFilePath)
+
+		f, err := os.Open(exitFilePath)
+		require.NoError(t, err)
+
+		var finalExit eth2p0.SignedVoluntaryExit
+		require.NoError(t, json.NewDecoder(f).Decode(&finalExit))
+
+		require.NotEmpty(t, finalExit)
+	}
+}