Deny ephemeral values in count (#3924)

Signed-off-by: Andrei Ciobanu <andrei.ciobanu@opentofu.org>

CHANGELOG.md

@@ -64,6 +64,7 @@ value was declared inside the module. ([#3067](https://github.com/opentofu/opent
 - The `gcs` backend now supports a `universe_domain` option to support sovereign GCP services. ([#3758](https://github.com/opentofu/opentofu/issues/3758))
 - OpenTofu now consistently sends "null" to `key_provider "external"` programs when only encryption the key is requested. ([#3672](https://github.com/opentofu/opentofu/pull/3672))
 - Ephemeral resources are not stored in the plan anymore. ([#3897](https://github.com/opentofu/opentofu/pull/3897))
+- `count` cannot use ephemeral values ([#3924](https://github.com/opentofu/opentofu/pull/3924))
 
 ## Previous Releases
 

internal/command/e2etest/primary_test.go

@@ -253,6 +253,9 @@ func TestPrimaryChdirOption(t *testing.T) {
 //     the status update of their execution.
 func TestEphemeralWorkflowAndOutput(t *testing.T) {
 	t.Parallel()
+	if runtime.GOOS == "linux" && runtime.GOARCH == "arm" {
+		t.Skip("Test skipped due to inability to run on linux_arm. Might need further investigation to make it runnable")
+	}
 
 	pluginVersionRunner := func(t *testing.T, testdataPath string, providerBuilderFunc func(*testing.T, string)) {
 		tf := e2e.NewBinary(t, tofuBin, testdataPath)
@@ -303,27 +306,26 @@ Plan: 2 to add, 0 to change, 0 to destroy.
 Changes to Outputs:
   + final_output = "just a simple resource to ensure that the second provider it's working fine"`
 
-			entriesChecker := &outputEntriesChecker{phase: "plan"}
-			entriesChecker.addChecks(
-				outputEntry{[]string{"data.simple_resource.test_data1: Reading..."}, true},
-				outputEntry{[]string{"data.simple_resource.test_data1: Read complete after"}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Opening..."}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Open complete after"}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Opening..."}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Open complete after"}, true},
-				outputEntry{[]string{"module.call.ephemeral.simple_resource.deferred_ephemeral: Deferred due to unknown configuration"}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Closing..."}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Close complete after"}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Closing..."}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Close complete after"}, true},
-			)
+			checker := outputEntriesChecker{
+				outputCheckContains{[]string{"data.simple_resource.test_data1: Reading..."}, true},
+				outputCheckContains{[]string{"data.simple_resource.test_data1: Read complete after"}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Opening..."}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Open complete after"}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Opening..."}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Open complete after"}, true},
+				outputCheckContains{[]string{"module.call.ephemeral.simple_resource.deferred_ephemeral: Deferred due to unknown configuration"}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Closing..."}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Close complete after"}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Closing..."}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Close complete after"}, true},
+			}
 			out := stripAnsi(stdout)
 
 			if !strings.Contains(out, expectedChangesOutput) {
 				t.Errorf("wrong plan output:\nstdout:%s\nstderr:%s", stdout, stderr)
 				t.Log(cmp.Diff(out, expectedChangesOutput))
 			}
-			entriesChecker.check(t, out)
+			checker.check(t, "plan", out)
 
 			// assert plan file content
 			plan, err := tf.Plan("tfplan")
@@ -430,45 +432,44 @@ Changes to Outputs:
 			expectedChangesOutput := `Apply complete! Resources: 2 added, 0 changed, 0 destroyed.`
 			// NOTE: the non-required ones are dependent on the performance of the platform that this test is running on.
 			// In CI, if we would make this as required, this test might be flaky.
-			entriesChecker := outputEntriesChecker{phase: "apply"}
-			entriesChecker.addChecks(
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Opening..."}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Open complete after"}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Opening..."}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Open complete after"}, true},
-				outputEntry{[]string{"module.call.ephemeral.simple_resource.deferred_ephemeral: Opening..."}, true},
-				outputEntry{[]string{"module.call.ephemeral.simple_resource.deferred_ephemeral: Open complete after"}, true},
-				outputEntry{[]string{"module.call.data.simple_resource.deferred_data: Reading..."}, true},
-				outputEntry{[]string{"module.call.data.simple_resource.deferred_data: Read complete after"}, true},
-				outputEntry{[]string{"simple_resource.test_res: Creating..."}, true},
-				outputEntry{[]string{"simple_resource.test_res_second_provider: Creating..."}, true},
-				outputEntry{[]string{"simple_resource.test_res_second_provider: Creation complete after"}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Renewing..."}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Renew complete after"}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Renewing..."}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Renew complete after"}, true},
-				outputEntry{[]string{"simple_resource.test_res: Creation complete after"}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Closing..."}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Close complete after"}, true},
-				outputEntry{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Closing..."}, true},
-				outputEntry{[]string{"simple_resource.test_res: Provisioning with 'local-exec'..."}, true},
-				outputEntry{[]string{
+			checker := outputEntriesChecker{
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Opening..."}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Open complete after"}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Opening..."}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Open complete after"}, true},
+				outputCheckContains{[]string{"module.call.ephemeral.simple_resource.deferred_ephemeral: Opening..."}, true},
+				outputCheckContains{[]string{"module.call.ephemeral.simple_resource.deferred_ephemeral: Open complete after"}, true},
+				outputCheckContains{[]string{"module.call.data.simple_resource.deferred_data: Reading..."}, true},
+				outputCheckContains{[]string{"module.call.data.simple_resource.deferred_data: Read complete after"}, true},
+				outputCheckContains{[]string{"simple_resource.test_res: Creating..."}, true},
+				outputCheckContains{[]string{"simple_resource.test_res_second_provider: Creating..."}, true},
+				outputCheckContains{[]string{"simple_resource.test_res_second_provider: Creation complete after"}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Renewing..."}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Renew complete after"}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Renewing..."}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Renew complete after"}, true},
+				outputCheckContains{[]string{"simple_resource.test_res: Creation complete after"}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Closing..."}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[0]: Close complete after"}, true},
+				outputCheckContains{[]string{"ephemeral.simple_resource.test_ephemeral[1]: Closing..."}, true},
+				outputCheckContains{[]string{"simple_resource.test_res: Provisioning with 'local-exec'..."}, true},
+				outputCheckContains{[]string{
 					`simple_resource.test_res (local-exec): Executing: ["/bin/sh" "-c" "echo \"visible test value\""]`,
 					`simple_resource.test_res (local-exec): Executing: ["cmd" "/C" "echo \"visible test value\""]`,
 				}, true},
-				outputEntry{[]string{
+				outputCheckContains{[]string{
 					`simple_resource.test_res (local-exec): visible test value`,
 					`simple_resource.test_res (local-exec): \"visible test value\"`,
 				}, true},
-				outputEntry{[]string{"simple_resource.test_res (local-exec): (output suppressed due to ephemeral value in config)"}, true},
-			)
+				outputCheckContains{[]string{"simple_resource.test_res (local-exec): (output suppressed due to ephemeral value in config)"}, true},
+			}
 			out := stripAnsi(stdout)
 
 			if !strings.Contains(out, expectedChangesOutput) {
 				t.Errorf("wrong apply output:\nstdout:%s\nstderr%s", stdout, stderr)
 				t.Log(cmp.Diff(out, expectedChangesOutput))
 			}
-			entriesChecker.check(t, out)
+			checker.check(t, "apply", out)
 		}
 		{ // DESTROY
 			stdout, stderr, err := tf.Run("destroy", `-var=simple_input=plan_val`, `-var=ephemeral_input=ephemeral_val`, "-auto-approve")
@@ -513,6 +514,158 @@ Changes to Outputs:
 	}
 }
 
+func TestEphemeralRepetitionData(t *testing.T) {
+	t.Parallel()
+
+	tf := e2e.NewBinary(t, tofuBin, "testdata/ephemeral-repetition")
+	buildSimpleProvider(t, "6", tf.WorkDir(), "simple")
+	exec := func(
+		t *testing.T,
+		chdir string,
+		expectDestroyError bool,
+		expectedPlanOutput []outputCheck,
+		expectedApplyOutput []outputCheck,
+		expectedDestroyOutput []outputCheck,
+	) {
+		{ // INIT
+			_, stderr, err := tf.Run("-chdir="+chdir, "init", "-plugin-dir=../cache")
+			if err != nil {
+				t.Fatalf("unexpected init error: %s\nstderr:\n%s", err, stderr)
+			}
+		}
+
+		{ // PLAN
+			stdout, stderr, err := tf.Run("-chdir="+chdir, "plan")
+			combined := fmt.Sprintf("%s\n\n%s", stripAnsi(stdout), stripAnsi(stderr))
+			if err == nil {
+				t.Errorf("expected to have an error during plan but got nothing. output:\n%s", combined)
+			}
+			entriesChecker := outputEntriesChecker(expectedPlanOutput)
+			entriesChecker.check(t, "plan", combined)
+		}
+		{ // APPLY
+			stdout, stderr, err := tf.Run("-chdir="+chdir, "apply", "-auto-approve")
+			combined := fmt.Sprintf("%s\n\n%s", stripAnsi(stdout), stripAnsi(stderr))
+			if err == nil {
+				t.Errorf("expected to have an error during apply but got nothing. output:\n%s", combined)
+			}
+			entriesChecker := outputEntriesChecker(expectedApplyOutput)
+			entriesChecker.check(t, "apply", combined)
+		}
+		{ // DESTROY
+			stdout, stderr, err := tf.Run("-chdir="+chdir, "destroy", "-auto-approve")
+			combined := fmt.Sprintf("%s\n\n%s", stripAnsi(stdout), stripAnsi(stderr))
+			if !expectDestroyError && err != nil {
+				t.Errorf("expected to have no error during destroy. got %q instead. output:\n%s", err, combined)
+			} else if expectDestroyError && err == nil {
+				t.Errorf("expected to have an error during destroy. got null instead. output:\n%s", combined)
+			}
+			entriesChecker := outputEntriesChecker(expectedDestroyOutput)
+			entriesChecker.check(t, "destroy", combined)
+		}
+	}
+	cases := map[string]struct {
+		chdir                 string
+		expectDestroyError    bool
+		expectedPlanOutput    []outputCheck
+		expectedApplyOutput   []outputCheck
+		expectedDestroyOutput []outputCheck
+	}{
+		"lifecycle.enabled": {
+			chdir:              "enabled",
+			expectDestroyError: false,
+			expectedPlanOutput: []outputCheck{
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Opening...`, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Open complete after`, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Closing...`, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Close complete after`, true},
+				outputCheckContains{[]string{"Error: Invalid enabled argument"}, true},
+				outputCheckContains{[]string{`on main.tf line 20, in data "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 27, in resource "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 34, in ephemeral "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`ephemeral "simple_resource" "res"`}, true},
+			},
+			expectedApplyOutput: []outputCheck{
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Opening...`, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Open complete after`, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Closing...`, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Close complete after`, true},
+				outputCheckContains{[]string{"Error: Invalid enabled argument"}, true},
+				outputCheckContains{[]string{`on main.tf line 20, in data "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 27, in resource "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 34, in ephemeral "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`ephemeral "simple_resource" "res"`}, true},
+			},
+			expectedDestroyOutput: []outputCheck{
+				outputCheckContains{[]string{`No changes. No objects need to be destroyed.`}, true},
+			},
+		},
+		"count": {
+			chdir:              "count",
+			expectDestroyError: true,
+			expectedPlanOutput: []outputCheck{
+				outputCheckContains{[]string{"Error: Invalid count argument"}, true},
+				outputCheckContains{[]string{`on main.tf line 18, in data "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 23, in resource "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 28, in ephemeral "simple_resource" "res"`}, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Opening...`, true},
+			},
+			expectedApplyOutput: []outputCheck{
+				outputCheckContains{[]string{"Error: Invalid count argument"}, true},
+				outputCheckContains{[]string{`on main.tf line 18, in data "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 23, in resource "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 28, in ephemeral "simple_resource" "res"`}, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Opening...`, true},
+			},
+			expectedDestroyOutput: []outputCheck{
+				outputCheckContains{[]string{"Error: Invalid count argument"}, true},
+				outputCheckContains{[]string{`on main.tf line 18, in data "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 23, in resource "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 28, in ephemeral "simple_resource" "res"`}, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Opening...`, true},
+			},
+		},
+		"for_each": {
+			chdir:              "for_each",
+			expectDestroyError: true,
+			expectedPlanOutput: []outputCheck{
+				outputCheckContains{[]string{"Error: Invalid for_each argument"}, true},
+				outputCheckContains{[]string{`on main.tf line 19, in data "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 24, in resource "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 29, in ephemeral "simple_resource" "res"`}, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Opening...`, true},
+			},
+			expectedApplyOutput: []outputCheck{
+				outputCheckContains{[]string{"Error: Invalid for_each argument"}, true},
+				outputCheckContains{[]string{`on main.tf line 19, in data "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 24, in resource "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 29, in ephemeral "simple_resource" "res"`}, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Opening...`, true},
+			},
+			expectedDestroyOutput: []outputCheck{
+				outputCheckContains{[]string{"Error: Invalid for_each argument"}, true},
+				outputCheckContains{[]string{`on main.tf line 19, in data "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 24, in resource "simple_resource" "res"`}, true},
+				outputCheckContains{[]string{`on main.tf line 29, in ephemeral "simple_resource" "res"`}, true},
+				outputCheckDoesNotContain{`ephemeral.simple_resource.res: Opening...`, true},
+			},
+		},
+	}
+	for name, tt := range cases {
+		t.Run(name, func(t *testing.T) {
+			exec(
+				t,
+				tt.chdir,
+				tt.expectDestroyError,
+				tt.expectedPlanOutput,
+				tt.expectedApplyOutput,
+				tt.expectedDestroyOutput,
+			)
+		})
+	}
+
+}
+
 // This function builds and moves to a directory called "cache" inside the workdir,
 // the version of the provider passed as argument.
 // Instead of using this function directly, the pre-configured functions buildV5TestProvider and
@@ -566,45 +719,55 @@ func buildSimpleProvider(t *testing.T, version string, workdir string, buildOutN
 	}
 }
 
-type outputEntry struct {
+type outputCheckContains struct {
 	variants []string
-	required bool
+	strict   bool
 }
 
-func (oe outputEntry) in(out string) bool {
+func (oe outputCheckContains) check(t *testing.T, hint, in string) {
 	for _, v := range oe.variants {
-		if strings.Contains(out, v) {
-			return true
+		if strings.Contains(in, v) {
+			return
 		}
 	}
-	return false
+	if oe.strict {
+		t.Errorf("[%s] output does not contain required content %s\nout:%s", hint, oe.String(), in)
+	} else {
+		// We don't want to fail the test for outputs that are performance and time dependent
+		// as the renew status updates
+		t.Logf("[%s] output does not contain %s\nout:%s", hint, oe.String(), in)
+	}
 }
 
-func (oe outputEntry) String() string {
+func (oe outputCheckContains) String() string {
 	return `"` + strings.Join(oe.variants, `" OR "`) + `"`
 }
 
-type outputEntriesChecker struct {
-	entries []outputEntry
-	phase   string
+type outputCheckDoesNotContain struct {
+	token  string
+	strict bool
 }
 
-func (oec *outputEntriesChecker) addChecks(entries ...outputEntry) {
-	oec.entries = append(oec.entries, entries...)
+func (oe outputCheckDoesNotContain) check(t *testing.T, hint, in string) {
+	if !strings.Contains(in, oe.token) {
+		return
+	}
+	if oe.strict {
+		t.Errorf("[%s] output contains %q but it shouldn't\nout:%s", hint, oe.token, in)
+	} else {
+		t.Logf("[%s] output contains %q but it shouldn't\nout:%s", hint, oe.token, in)
+	}
 }
 
-func (oec *outputEntriesChecker) check(t *testing.T, contentToCheckIn string) {
-	for _, entry := range oec.entries {
-		if entry.in(contentToCheckIn) {
-			continue
-		}
-		if entry.required {
-			t.Errorf("%s output does not contain required content %s\nout:%s", oec.phase, entry.String(), contentToCheckIn)
-		} else {
-			// We don't want to fail the test for outputs that are performance and time dependent
-			// as the renew status updates
-			t.Logf("%s output does not contain %s\nout:%s", oec.phase, entry.String(), contentToCheckIn)
-		}
+type outputCheck interface {
+	check(t *testing.T, hint, target string)
+}
+
+type outputEntriesChecker []outputCheck
+
+func (oec outputEntriesChecker) check(t *testing.T, phase string, contentToCheckIn string) {
+	for _, entry := range oec {
+		entry.check(t, phase, contentToCheckIn)
 	}
 }
 

internal/command/e2etest/testdata/ephemeral-repetition/count/main.tf

@@ -0,0 +1,31 @@
+// the provider-plugin tests uses the -plugin-cache flag so terraform pulls the
+// test binaries instead of reaching out to the registry.
+terraform {
+  required_providers {
+    simple = {
+      source = "registry.opentofu.org/hashicorp/simple"
+    }
+  }
+}
+
+variable "in" {
+  type      = number
+  default   = 1
+  ephemeral = true
+}
+
+data "simple_resource" "res" {
+  count = var.in
+  value = "test value"
+}
+
+resource "simple_resource" "res" {
+  count = var.in
+  value = "test value"
+}
+
+ephemeral "simple_resource" "res" {
+  count = var.in
+  value = "test value"
+}
+

internal/command/e2etest/testdata/ephemeral-repetition/enabled/main.tf

@@ -0,0 +1,38 @@
+// the provider-plugin tests uses the -plugin-cache flag so terraform pulls the
+// test binaries instead of reaching out to the registry.
+terraform {
+  required_providers {
+    simple = {
+      source = "registry.opentofu.org/hashicorp/simple"
+    }
+  }
+}
+
+
+variable "in" {
+  type      = bool
+  default   = true
+  ephemeral = true
+}
+
+data "simple_resource" "res" {
+  lifecycle {
+    enabled = var.in
+  }
+  value = "test value"
+}
+
+resource "simple_resource" "res" {
+  lifecycle {
+    enabled = var.in
+  }
+  value = "test value"
+}
+
+ephemeral "simple_resource" "res" {
+  lifecycle {
+    enabled = var.in
+  }
+  value = "test value"
+}
+

internal/command/e2etest/testdata/ephemeral-repetition/for_each/main.tf

@@ -0,0 +1,32 @@
+// the provider-plugin tests uses the -plugin-cache flag so terraform pulls the
+// test binaries instead of reaching out to the registry.
+terraform {
+  required_providers {
+    simple = {
+      source = "registry.opentofu.org/hashicorp/simple"
+    }
+  }
+}
+
+
+variable "in" {
+  type      = set(string)
+  default   = ["val"]
+  ephemeral = true
+}
+
+data "simple_resource" "res" {
+  for_each = var.in
+  value = "test value"
+}
+
+resource "simple_resource" "res" {
+  for_each = var.in
+  value = "test value"
+}
+
+ephemeral "simple_resource" "res" {
+  for_each = var.in
+  value = "test value"
+}
+

internal/lang/evalchecks/eval_count.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/hashicorp/hcl/v2"
 	"github.com/opentofu/opentofu/internal/addrs"
+	"github.com/opentofu/opentofu/internal/lang/marks"
 	"github.com/opentofu/opentofu/internal/tfdiags"
 	"github.com/zclconf/go-cty/cty"
 	"github.com/zclconf/go-cty/cty/gocty"
@@ -80,7 +81,20 @@ func EvaluateCountExpressionValue(expr hcl.Expression, ctx EvaluateFunc) (cty.Va
 
 	// Unmark the count value, sensitive values are allowed in count but not for_each,
 	// as using it here will not disclose the sensitive value
-	countVal, _ = countVal.Unmark()
+	countVal, valMarks := countVal.Unmark()
+
+	// We do not allow ephemeral values in the count value
+	if _, ok := valMarks[marks.Ephemeral]; ok {
+		diags = diags.Append(&hcl.Diagnostic{
+			Severity:   hcl.DiagError,
+			Summary:    "Invalid count argument",
+			Detail:     "Ephemeral values, or values derived from ephemeral values, cannot be used as count arguments. If used, the ephemeral value could be exposed as a resource instance key.",
+			Subject:    expr.Range().Ptr(),
+			Expression: expr,
+			Extra:      DiagnosticCausedByConfidentialValues(true),
+		})
+		return nullCount, diags
+	}
 
 	switch {
 	case countVal.IsNull():

internal/lang/evalchecks/eval_count_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/hashicorp/hcl/v2"
 	"github.com/hashicorp/hcl/v2/hcltest"
 	"github.com/opentofu/opentofu/internal/addrs"
+	"github.com/opentofu/opentofu/internal/lang/marks"
 	"github.com/opentofu/opentofu/internal/tfdiags"
 	"github.com/zclconf/go-cty/cty"
 )
@@ -114,6 +115,17 @@ func TestEvaluateCountExpression_errors(t *testing.T) {
 			"The \"count\" value depends on resource attributes that cannot be determined until apply, so OpenTofu cannot predict how many instances will be created.\n\nTo work around this, use the planning option -exclude=module.a.bar.foo to first apply without this object, and then apply normally to converge.",
 			true,
 		},
+		"ephemeral value": {
+			cty.NumberIntVal(1).Mark(marks.Ephemeral),
+			addrs.Resource{
+				Mode: addrs.ManagedResourceMode,
+				Name: "foo",
+				Type: "bar",
+			}.Absolute(addrs.RootModuleInstance.Child("a", addrs.NoKey)),
+			"Invalid count argument",
+			"Ephemeral values, or values derived from ephemeral values, cannot be used as count arguments. If used, the ephemeral value could be exposed as a resource instance key.",
+			false,
+		},
 	}
 
 	for name, test := range tests {

website/docs/language/meta-arguments/count.mdx

@@ -127,3 +127,12 @@ _index_ instead of the string values in the list. If an element was removed from
 the middle of the list, every instance _after_ that element would see its
 `subnet_id` value change, resulting in more remote object changes than intended.
 Using `for_each` gives the same flexibility without the extra churn.
+
+## Usage of ephemeral values
+
+As `count` allows usage of expressions, if such an expression contains a reference
+to an ephemeral value, the whole resulted value to be used for the `count` meta-argument
+will be ephemeral too.
+Usage of such values in `resource`, `data` and `ephemeral` blocks will raise an error.
+
+This is needed to be sure that OpenTofu stores no ephemeral information in the plan or the state.