Merge branch 'main' into zaid/double-instrinsics

pulumi · Jan 9, 2025 · 380d97d · 380d97d
2 parents bcc2499 + cd7a05c
commit 380d97d
Show file tree

Hide file tree

Showing 3 changed files with 80 additions and 27 deletions.
diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md
@@ -2,5 +2,6 @@
 
  - Fix emitted functions for `assetArchive` and `remoteAsset` in generated programs
  - Fix generation of `double` literals in generated programs
+ - Avoid calling invokes with dependencies on unknown resources
 
-### Bug Fixes
+### Bug Fixes
diff --git a/sdk/java/pulumi/src/main/java/com/pulumi/deployment/internal/DeploymentImpl.java b/sdk/java/pulumi/src/main/java/com/pulumi/deployment/internal/DeploymentImpl.java
@@ -91,6 +91,7 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
@@ -541,23 +542,36 @@ public <T> Output<T> invoke(String token, TypeShape<T> targetType, InvokeArgs ar
                 ? CompletableFuture.completedFuture(null)
                 : packageRef;
 
-            // Find all the resource dependencies from dependsOn. We need to wait for these futures to complete
-            // before calling the invoke.
-            var depsFuture = this.prepare.getAllTransitivelyReferencedResourceUrnsAsync(ImmutableSet.copyOf(options.getDependsOn()));
+            // The expanded set of dependencies, including children of components.
+            var transitiveDeps = this.prepare.getAllTransitivelyReferencedResources(ImmutableSet.copyOf(options.getDependsOn()));
+            // If we depend on any CustomResources, we need to ensure that their
+            // ID is known before proceeding. If it is not known, we will return
+            // an unknown result.
+            var hasUnknownIDs = CompletableFutures.allOf(transitiveDeps
+                .filter(r -> r instanceof CustomResource)
+                .map(r -> Internal.of(((CustomResource) r).id()).isKnown())
+                .collect(ImmutableSet.toImmutableSet())
+            ).thenApply(s -> s.stream().anyMatch(known -> !known));
 
             // Wait for all values from args to be available, and then perform the RPC.
-            return new OutputInternal<>(this.featureSupport.monitorSupportsResourceReferences()
-                    .thenCompose(keepResources -> this.serializeInvokeArgs(token, args, keepResources))
-                    .thenCompose(serializedArgs -> {
-                        if (!serializedArgs.containsUnknowns) {
-                            return CompletableFuture.allOf(depsFuture, packageRefFuture)
-                                    .thenCompose(v -> this.invokeRawAsync(token, serializedArgs, options, packageRefFuture.join()))
-                                    .thenApply(result -> parseInvokeResponse(token, targetType, result))
-                                    .thenApply(output -> output.withDependencies(options.getDependsOn()));
-                        } else {
-                            return CompletableFuture.completedFuture(OutputData.unknown());
-                        }
-                    }));
+            return new OutputInternal<>(CompletableFuture.allOf(
+                    this.featureSupport.monitorSupportsResourceReferences(),
+                    hasUnknownIDs)
+                .thenCompose(ignored -> {
+                    boolean keepResources = this.featureSupport.monitorSupportsResourceReferences().join();
+                    boolean hasUnknown = hasUnknownIDs.join();
+
+                    return this.serializeInvokeArgs(token, args, keepResources)
+                        .thenCompose(serializedArgs -> {
+                            if (serializedArgs.containsUnknowns || hasUnknown) {
+                                return CompletableFuture.completedFuture(OutputData.unknown());
+                            } else {
+                                return packageRefFuture
+                                    .thenCompose(packageRefString -> this.invokeRawAsync(token, serializedArgs, options, packageRefString))
+                                    .thenApply(result -> parseInvokeResponse(token, targetType, result));
+                            }
+                        });
+                }));
         }
 
         private <T> OutputData<T> parseInvokeResponse(
@@ -1073,7 +1087,7 @@ private CompletableFuture<List<Resource>> gatherExplicitDependenciesAsync(Output
             return Internal.of(resources).getValueOrDefault(List.of());
         }
 
-        private CompletableFuture<ImmutableSet<String>> getAllTransitivelyReferencedResourceUrnsAsync(
+        private Stream<Resource> getAllTransitivelyReferencedResources(
                 ImmutableSet<Resource> resources
         ) {
             // Go through 'resources', but transitively walk through **Component** resources, collecting any
@@ -1112,14 +1126,18 @@ private CompletableFuture<ImmutableSet<String>> getAllTransitivelyReferencedReso
                             return Internal.from(resource).getRemote();
                         }
                         return false; // Unreachable
-                    })
-                    .map(resource -> Internal.of(resource.urn()).getValueOrDefault(""))
-                    .collect(toImmutableSet());
-            return CompletableFutures.allOf(transitivelyReachableCustomResources)
-                    .thenApply(ts -> ts.stream()
-                            .filter(Strings::isNonEmptyOrNull)
-                            .collect(toImmutableSet())
-                    );
+                    });
+            return transitivelyReachableCustomResources;
+        }
+
+        private CompletableFuture<ImmutableSet<String>> getAllTransitivelyReferencedResourceUrnsAsync(
+                ImmutableSet<Resource> resources) {
+            var urns = getAllTransitivelyReferencedResources(resources)
+                .map(resource -> Internal.of(resource.urn()).getValueOrDefault(""));
+            return CompletableFutures.allOf(urns.collect(ImmutableSet.toImmutableSet()))
+                .thenApply(strings -> strings.stream().filter(Strings::isNonEmptyOrNull))
+                .thenApply(urn ->  urn.collect(ImmutableSet.toImmutableSet())
+            );
         }
 
         /**

diff --git a/sdk/java/pulumi/src/test/java/com/pulumi/deployment/DeploymentInvokeDependsOnTest.java b/sdk/java/pulumi/src/test/java/com/pulumi/deployment/DeploymentInvokeDependsOnTest.java
@@ -8,7 +8,6 @@
 import com.pulumi.core.annotations.CustomType.Setter;
 import com.pulumi.core.annotations.Import;
 import com.pulumi.core.annotations.ResourceType;
-import com.pulumi.core.internal.Internal;
 import com.pulumi.deployment.internal.Runner;
 import com.pulumi.resources.Resource;
 import com.pulumi.resources.InvokeArgs;
@@ -30,6 +29,7 @@
 import java.lang.Thread;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 
 public class DeploymentInvokeDependsOnTest {
 
@@ -43,7 +43,7 @@ void testInvokesDependsOn() {
         var marker = new ResolveMarker();
 
         var test = PulumiTestInternal.builder()
-                .options(TestOptions.builder().preview(true).build())
+                .options(TestOptions.builder().build())
                 .mocks(new Mocks() {
                     @Override
                     public CompletableFuture<ResourceResult> newResourceAsync(ResourceArgs args) {
@@ -92,6 +92,40 @@ public CompletableFuture<Map<String, Object>> callAsync(CallArgs args) {
         assertThat(result.exitCode()).isEqualTo(Runner.ProcessExitedSuccessfully);
     }
 
+    @Test
+    void testInvokesDependsOnUnknown() {
+        var test = PulumiTestInternal.builder()
+                .options(TestOptions.builder().preview(true).build())
+                .mocks(new Mocks() {
+                    @Override
+                    public CompletableFuture<ResourceResult> newResourceAsync(ResourceArgs args) {
+                        return CompletableFuture
+                                .supplyAsync(() -> ResourceResult.of(Optional.empty(), ImmutableMap.of()));
+                    }
+
+                    @Override
+                    public CompletableFuture<Map<String, Object>> callAsync(CallArgs args) {
+                        return CompletableFuture.completedFuture(ImmutableMap.of());
+                    }
+                })
+                .build();
+
+        var result = test.runTest(ctx -> {
+            var res = new MyCustomResource("r1", null, CustomResourceOptions.builder().build());
+            var deps = new ArrayList<Resource>();
+            deps.add(res);
+
+            var opts = new InvokeOutputOptions(null, null, null, deps);
+            CustomInvokes.doStuff(CustomArgs.Empty, opts).applyValue(r -> {
+                assertFalse(true, "invoke should not be called!");
+                return r;
+            });
+        });
+
+        assertThat(result.exceptions()).hasSize(0);
+        assertThat(result.exitCode()).isEqualTo(Runner.ProcessExitedSuccessfully);
+    }
+
     public static final class MyArgs extends ResourceArgs {
     }