Fix RPC recipes errors being swallowed when in batch (#7226)

Author: greg-at-moderne

rewrite-core/src/main/java/org/openrewrite/scheduling/RecipeRunCycle.java

@@ -406,7 +406,14 @@ void clear() {
             response = rpc.batchVisit(originalBefore, ctx, rootCursor, batch.items);
         } catch (Throwable t) {
             if (!batch.recipeStacks.isEmpty()) {
-                handleError(batch.recipeStacks.get(0).peek(), originalBefore, originalBefore, t);
+                SourceFile beforeError = source;
+                if (!(t instanceof RecipeRunException)) {
+                    source = Markup.error(source, t);
+                }
+                source = handleError(batch.recipeStacks.get(0).peek(), originalBefore, source, t);
+                if (source != null && source != beforeError) {
+                    source = addRecipesThatMadeChanges(batch.recipeStacks.get(0), source);
+                }
             }
             batch.clear();
             return source;

rewrite-core/src/test/java/org/openrewrite/rpc/RewriteRpcTest.java

@@ -25,6 +25,8 @@
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.openrewrite.*;
+import org.openrewrite.internal.InMemoryLargeSourceSet;
+import org.openrewrite.marker.Markup;
 import org.openrewrite.marker.Markers;
 import org.openrewrite.config.CompositeRecipe;
 import org.openrewrite.config.Environment;
@@ -413,6 +415,37 @@ void singleRpcRecipeNoBatch() {
         );
     }
 
+    /**
+     * When an RPC batch visit throws an exception (e.g. a recipe visitor fails on the
+     * remote side), the error must be visible: the source file should carry a Markup.Error
+     * marker and the error should be recorded in the SourcesFileErrors data table.
+     */
+    @Test
+    void batchVisitExceptionProducesErrorMarker() {
+        // given
+        // Two consecutive same-RPC recipes are required for the batch path to kick in
+        Recipe r1 = client.prepareRecipe("org.openrewrite.rpc.RewriteRpcTest$ThrowingRpcRecipe", Map.of());
+        Recipe r2 = client.prepareRecipe("org.openrewrite.rpc.RewriteRpcTest$ThrowingRpcRecipe", Map.of());
+
+        var errors = new java.util.ArrayList<Throwable>();
+        var ctx = new InMemoryExecutionContext(errors::add);
+        var source = PlainText.builder().text("hello").sourcePath(Path.of("test.txt")).build();
+
+        // when
+        RecipeRun run = new RecipeScheduler().scheduleRun(
+          new CompositeRecipe(List.of(r1, r2)),
+          new InMemoryLargeSourceSet(List.of(source)), ctx, 1, 1);
+
+        // then
+        assertThat(errors).isNotEmpty();
+
+        List<Result> results = run.getChangeset().getAllResults();
+        assertThat(results).isNotEmpty();
+        assertThat(results.getFirst().getAfter().getMarkers().findFirst(Markup.Error.class))
+          .describedAs("Source should carry Markup.Error so the failure is visible")
+          .isPresent();
+    }
+
     @Test
     void getCursor() {
         var parent = new Cursor(null, Cursor.ROOT_VALUE);
@@ -432,6 +465,29 @@ public PlainText visitText(PlainText text, Integer p) {
         }
     }
 
+    @SuppressWarnings("unused")
+    public static class ThrowingRpcRecipe extends Recipe {
+        @Override
+        public String getDisplayName() {
+            return "Throwing RPC recipe";
+        }
+
+        @Override
+        public String getDescription() {
+            return "Test recipe that throws during visit.";
+        }
+
+        @Override
+        public TreeVisitor<?, ExecutionContext> getVisitor() {
+            return new PlainTextVisitor<>() {
+                @Override
+                public PlainText visitText(PlainText text, ExecutionContext ctx) {
+                    throw new RuntimeException("boom from RPC");
+                }
+            };
+        }
+    }
+
     @SuppressWarnings("unused")
     static class RecipeWithRecipeList extends Recipe {
         @Override