Clean up benchmarks, fix PipeReader race, consolidate tests

- Remove benchmarks/throughput-benchmark (replaced by BenchmarkDotNet suite)
- Remove benchmarks/CROSS_SDK_BENCHMARK_SPEC.md and benchmark-runner.sh
- Move integration-test.sh to .github/scripts/
- Fix PipeReader cleanup race: await incomingTask BEFORE reader.Complete()
- Simplify InvocationHandler exception catch to OperationCanceledException
- Fix discovery.proto: add V3/V4 protocol version descriptions
- Fix comment: v4 adds lambda compression, not retry_policy
- Fix ReplayRun benchmark to measure StartAsync replay correctly
- Remove unnecessary Roslyn dependencies from benchmarks project
- Consolidate InterfaceConformanceTests to Theory+InlineData
- Consolidate MockContextFeatureTests to Theory+InlineData
- Remove redundant HandlerAttributeBase tests (covered by property tests)

Author: BeshoyHindy

test/integration-test.sh .github/scripts/integration-test.shtest/integration-test.sh renamed to .github/scripts/integration-test.sh

@@ -21,4 +21,9 @@ enum ServiceDiscoveryProtocolVersion {
   V1 = 1;
   // add custom metadata and documentation for services/handlers
   V2 = 2;
+  // add options for private service, journal retention, idempotency retention,
+  // workflow completion retention, inactivity timeout, abort timeout, enable lazy state
+  V3 = 3;
+  // add lambda compression
+  V4 = 4;
 }

benchmarks/CROSS_SDK_BENCHMARK_SPEC.md

@@ -20,7 +20,7 @@ public static class RestateEndpointRouteBuilderExtensions
             .GetCustomAttribute<AssemblyInformationalVersionAttribute>()?.InformationalVersion ?? "0.0.0"}";
 
     // Supported manifest versions, highest priority first.
-    // v4 adds retry_policy fields we don't support yet, so we cap at v3.
+    // v4 adds lambda compression fields we don't support yet, so we cap at v3.
     private static readonly string[] SupportedContentTypes =
     [
         "application/vnd.restate.endpointmanifest.v3+json",

benchmarks/benchmark-runner.sh

@@ -55,7 +55,7 @@ public async Task HandleAsync(
             var context = CreateContext(sm, service.Type, handler.IsShared, logger, handlerToken);
 
             object? input = null;
-            if (!startInfo.Input.IsEmpty && handler.InputDeserializer is not null)
+            if (handler.InputDeserializer is not null)
                 input = handler.InputDeserializer(new ReadOnlySequence<byte>(startInfo.Input));
 
             var serviceInstance = service.Factory(serviceProvider);
@@ -102,21 +102,21 @@ public async Task HandleAsync(
             // the remaining cleanup from executing AND propagate an exception to Kestrel.
             try { writer.Complete(); } catch { /* already completed or broken */ }
 
+            // Cancel and await the incoming reader task BEFORE completing the reader.
+            // This avoids a "Concurrent reads or writes are not supported" race between
+            // PipeReader.Complete() and ProcessIncomingMessagesAsync's pending ReadAsync().
             try { await incomingCts.CancelAsync().ConfigureAwait(false); } catch { /* ignore */ }
-            try { reader.Complete(); } catch { /* already completed or broken */ }
-
-            // Observe the incoming task to prevent silent exception swallowing.
-            // Expected: OperationCanceledException (cancellation) or InvalidOperationException
-            // (reader.Complete() interrupted a pending read).
             if (incomingTask is not null)
                 try
                 {
                     await incomingTask.ConfigureAwait(false);
                 }
-                catch (Exception ex) when (ex is OperationCanceledException or InvalidOperationException)
+                catch (OperationCanceledException ex)
                 {
                     Log.IncomingReaderStopped(logger, ex, sm.InvocationId);
                 }
+
+            try { reader.Complete(); } catch { /* already completed or broken */ }
         }
     }