RPC: explicitly fail if RPC receives a message it has no codec for (#7342)

* Defensive mechanism for missing codecs

* Defensive mechanism for missing codecs

Author: greg-at-moderne

rewrite-core/src/main/java/org/openrewrite/rpc/RpcReceiveQueue.java

@@ -138,6 +138,11 @@ public <T> T receive(@Nullable T before, @Nullable UnaryOperator<T> onChange) {
                     after = codec.rpcReceive(before, this);
                 } else if (message.getValueType() == null) {
                     after = message.getValue();
+                } else if (message.getState() == RpcObjectData.State.ADD && message.getValue() == null) {
+                    throw new IllegalStateException(
+                            "No RPC codec registered on the Java side for '" + message.getValueType() + "'. " +
+                                    "The remote side has a codec and sent property messages that will not be consumed, " +
+                                    "causing RPC queue desynchronization.");
                 } else {
                     after = before;
                 }

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

@@ -26,6 +26,7 @@
 import java.util.*;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 public class RpcReceiveQueueTest {
 
@@ -106,6 +107,19 @@ void changePropertyType() {
         assertThat(((Checksum) received).getAlgorithm()).isEqualTo("SHA-256");
     }
 
+    @Test
+    void detectsMissingCodecOnReceiverSide() {
+        // given
+        batches.addLast(List.of(
+            new RpcObjectData(RpcObjectData.State.ADD, "java.lang.StringBuilder", null, null, false)
+        ));
+
+        // when / then
+        assertThatThrownBy(() -> rq.receive(null))
+            .isInstanceOf(IllegalStateException.class)
+            .hasMessageContaining("No RPC codec registered on the Java side for 'java.lang.StringBuilder'");
+    }
+
     private List<RpcObjectData> encode(List<RpcObjectData> batch) {
         List<RpcObjectData> encoded = new ArrayList<>();
         for (RpcObjectData data : batch) {

rewrite-csharp/csharp/OpenRewrite/Core/Rpc/RpcReceiveQueue.cs

@@ -151,10 +151,12 @@ public RpcObjectData Take()
                     // Simple value types (enums, primitives) sent with both valueType and value
                     after = ExtractValue<T>(message.Value);
                 }
-                else if (message.ValueType != null)
+                else if (message.State == ADD && message.ValueType != null)
                 {
-                    // ValueType set but no value and no codec - keep before
-                    after = before;
+                    throw new InvalidOperationException(
+                        $"No RPC codec registered on the C# side for '{message.ValueType}'. " +
+                        "The remote side has a codec and sent property messages that will not be consumed, " +
+                        "causing RPC queue desynchronization.");
                 }
                 else
                 {

rewrite-javascript/rewrite/src/rpc/queue.ts

@@ -348,6 +348,12 @@ export class RpcReceiveQueue {
                         after = await codec.rpcReceive(before, this);
                     } else if (message.value !== undefined) {
                         after = message.valueType ? {kind: message.valueType, ...message.value} : message.value;
+                    } else if (message.state === RpcObjectState.ADD && message.valueType) {
+                        throw new Error(
+                            `No RPC codec registered on the TypeScript side for '${message.valueType}'. ` +
+                            `The Java side has a codec and sent property messages that will not be consumed, ` +
+                            `causing RPC queue desynchronization.`
+                        );
                     } else {
                         after = before;
                     }

rewrite-javascript/rewrite/test/rpc/queue.test.ts

@@ -1,5 +1,6 @@
 import {Json} from "../../src/json";
-import {asRef, ReferenceMap, RpcReceiveQueue, RpcSendQueue} from "../../src/rpc";
+import {asRef, ReferenceMap, RpcReceiveQueue, RpcSendQueue, RpcObjectState} from "../../src/rpc";
+import type {RpcObjectData} from "../../src/rpc";
 
 describe("RPC queues", () => {
 
@@ -50,4 +51,17 @@ describe("RPC queues", () => {
         expect(received.value.kind).toBe(Json.Kind.Identifier);
         expect(received.value.kind).not.toBe(beforeWrapper.value.kind);
     });
+
+    test("detects missing codec on receiver side", async () => {
+        // given
+        const batch: RpcObjectData[] = [
+            {state: RpcObjectState.ADD, valueType: "com.example.UnknownMarker", value: undefined},
+        ];
+        const rq = new RpcReceiveQueue(new Map(), undefined, async () => batch, undefined, false);
+
+        // when / then
+        await expect(rq.receive(undefined)).rejects.toThrow(
+            "No RPC codec registered on the TypeScript side for 'com.example.UnknownMarker'"
+        );
+    });
 });

rewrite-python/rewrite/src/rewrite/rpc/receive_queue.py

@@ -192,6 +192,12 @@ def _do_change(
                     after = {'kind': message.value_type, **message.value} if isinstance(message.value, dict) else message.value
                 else:
                     after = message.value
+            elif message.state == RpcObjectState.ADD and message.value_type:
+                raise RuntimeError(
+                    f"No RPC codec registered on the Python side for '{message.value_type}'. "
+                    "The remote side has a codec and sent property messages that will not be consumed, "
+                    "causing RPC queue desynchronization."
+                )
             else:
                 after = before