feat(functions): stream edge function responses

packages/Core/Core/Http/RetryExecutor.cs

@@ -12,13 +12,18 @@ public static class RetryExecutor
     private static readonly ThreadLocal<Random> RandomProvider = new(() => new Random(Guid.NewGuid().GetHashCode()));
 
     /// <summary>Sends via <paramref name="requestFactory"/>, called fresh per attempt since a sent <see cref="HttpRequestMessage"/> can't be reused.</summary>
+    public static Task<HttpResponseMessage> SendAsync(HttpClient client, Func<HttpRequestMessage> requestFactory,
+        RetryOptions options, CancellationToken cancellationToken = default) =>
+        SendAsync(client, requestFactory, options, HttpCompletionOption.ResponseContentRead, cancellationToken);
+
+    /// <summary>Sends a request with the specified <paramref name="completionOption"/>.</summary>
     public static async Task<HttpResponseMessage> SendAsync(HttpClient client, Func<HttpRequestMessage> requestFactory,
-        RetryOptions options, CancellationToken cancellationToken = default)
+        RetryOptions options, HttpCompletionOption completionOption, CancellationToken cancellationToken)
     {
         var attempt = 0;
         while (true)
         {
-            var (response, exception) = await Attempt(client, requestFactory, cancellationToken).ConfigureAwait(false);
+            var (response, exception) = await Attempt(client, requestFactory, completionOption, cancellationToken).ConfigureAwait(false);
             if (!ShouldRetry(response, exception, attempt, options))
                 return response ?? throw exception!;
 
@@ -29,12 +34,12 @@ public static async Task<HttpResponseMessage> SendAsync(HttpClient client, Func<
     }
 
     private static async Task<(HttpResponseMessage? Response, HttpRequestException? Exception)> Attempt(
-        HttpClient client, Func<HttpRequestMessage> requestFactory, CancellationToken cancellationToken)
+        HttpClient client, Func<HttpRequestMessage> requestFactory, HttpCompletionOption completionOption, CancellationToken cancellationToken)
     {
         using var request = requestFactory();
         try
         {
-            return (await client.SendAsync(request, cancellationToken).ConfigureAwait(false), null);
+            return (await client.SendAsync(request, completionOption, cancellationToken).ConfigureAwait(false), null);
         }
         catch (HttpRequestException e)
         {

packages/Core/Core/PublicAPI.Unshipped.txt

@@ -1 +1,2 @@
 #nullable enable
+static Supabase.Core.Http.RetryExecutor.SendAsync(System.Net.Http.HttpClient! client, System.Func<System.Net.Http.HttpRequestMessage!>! requestFactory, Supabase.Core.Http.RetryOptions! options, System.Net.Http.HttpCompletionOption completionOption, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.Task<System.Net.Http.HttpResponseMessage!>!

packages/Functions/Functions.Tests/ClientContractTests.cs

@@ -10,6 +10,7 @@
 using Supabase.Functions;
 using Supabase.Functions.Exceptions;
 using WireMock;
+using WireMock.Models;
 using WireMock.RequestBuilders;
 using WireMock.ResponseBuilders;
 using WireMock.Server;
@@ -28,6 +29,7 @@ namespace Functions.Tests;
 public class ClientContractTests
 {
     private const string FunctionName = "hello";
+    private static readonly TimeSpan Deadline = TimeSpan.FromSeconds(5);
 
     private WireMockServer server = null!;
     private Client client = null!;
@@ -251,10 +253,62 @@ public async Task Invoke_ShouldThrowOperationCanceledException_GivenCallerCancel
             "caller-triggered cancellation must not be mis-reported as an HttpTimeout");
     }
 
+    [TestMethod]
+    public async Task InvokeStream_ShouldReturnBeforeBodyCompletes()
+    {
+        var holdBodyOpen = new TaskCompletionSource();
+        this.RespondWithEvents(async queue =>
+        {
+            queue.Write("data: first\n\n");
+            await holdBodyOpen.Task;
+        });
+
+        using var response = await this.client.InvokeStream(FunctionName).WaitAsync(Deadline);
+        holdBodyOpen.SetResult();
+
+        (await response.Content.ReadAsStringAsync()).Should().Be("data: first\n\n");
+    }
+
+    [TestMethod]
+    public async Task InvokeStream_ShouldBufferTheErrorBody_GivenServerError()
+    {
+        this.RespondWith(500, "internal boom");
+        var act = () => this.client.InvokeStream(FunctionName);
+        var exception = (await act.Should().ThrowAsync<FunctionsException>()).Which;
+        (await exception.Response!.Content.ReadAsStringAsync()).Should().Be("internal boom");
+    }
+
+    [TestMethod]
+    public async Task InvokeStream_ShouldThrowTimeoutException_GivenAStalledErrorBody()
+    {
+        var holdBodyOpen = new TaskCompletionSource();
+        // WireMock forces 200 on an SSE body, so a relay error is what makes this one fail.
+        this.RespondWithEvents(async queue =>
+        {
+            queue.Write("partial");
+            await holdBodyOpen.Task;
+        }, Response.Create().WithHeader("x-relay-error", "true"));
+
+        var act = () => this.client.InvokeStream(FunctionName, options: new InvokeFunctionOptions { HttpTimeout = TimeSpan.FromMilliseconds(100) });
+        var exception = (await act.Should().ThrowAsync<TaskCanceledException>().WaitAsync(Deadline)).Which;
+        holdBodyOpen.SetResult();
+
+        exception.InnerException.Should().BeOfType<TimeoutException>();
+    }
+
     private void RespondWith(int statusCode, string body) =>
         this.server.Given(Request.Create().WithPath($"/functions/v1/{FunctionName}").UsingAnyMethod())
             .RespondWith(Response.Create().WithStatusCode(statusCode).WithHeader("Content-Type", "application/json").WithBody(body));
 
+    private void RespondWithEvents(Func<IBlockingQueue<string?>, Task> body, IResponseBuilder? response = null) =>
+        this.server.Given(Request.Create().WithPath($"/functions/v1/{FunctionName}").UsingAnyMethod())
+            .RespondWith((response ?? Response.Create()).WithHeader("Content-Type", "text/event-stream")
+                .WithSseBody(async (_, queue) =>
+                {
+                    await body(queue);
+                    queue.Close();
+                }));
+
     private IRequestMessage SingleRequest() => this.server.LogEntries.Should().ContainSingle().Which.RequestMessage!;
 
     private string HeaderOf(string name) => this.SingleRequest().Headers![name][0];

packages/Functions/Functions/Client.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.IO;
 using System.Net.Http;
 using System.Runtime.CompilerServices;
 using System.Text;
@@ -82,6 +83,22 @@ public async Task<HttpContent> RawInvoke(
         return (await this.HandleRequest(functionName, url, token, options, cancellationToken)).Content;
     }
 
+    /// <summary>
+    /// Invokes a function and returns a successful response after its headers arrive.
+    /// <see cref="InvokeFunctionOptions.HttpTimeout"/> covers the request and any error body.
+    /// </summary>
+    /// <param name="functionName">Function name, appended to the base URL.</param>
+    /// <param name="token">Bearer token.</param>
+    /// <param name="options">Invocation options.</param>
+    /// <param name="cancellationToken">Cancels the request and error-body reads. Cancel successful body reads separately.</param>
+    /// <returns>The response, which the caller must dispose.</returns>
+    public Task<HttpResponseMessage> InvokeStream(
+        string functionName,
+        string? token = null,
+        InvokeFunctionOptions? options = null,
+        CancellationToken cancellationToken = default
+    ) => this.HandleRequest(functionName, $"{this.baseUrl}/{functionName}", token, options, cancellationToken, HttpCompletionOption.ResponseHeadersRead);
+
     /// <summary>
     /// Invokes a function and returns the Text content of the response.
     /// </summary>
@@ -136,14 +153,16 @@ public async Task<string> Invoke(
     /// <param name="token"></param>
     /// <param name="options"></param>
     /// <param name="cancellationToken"></param>
+    /// <param name="completionOption"></param>
     /// <returns></returns>
     /// <exception cref="FunctionsException"></exception>
     private async Task<HttpResponseMessage> HandleRequest(
         string functionName,
         string url,
         string? token = null,
         InvokeFunctionOptions? options = null,
-        CancellationToken cancellationToken = default
+        CancellationToken cancellationToken = default,
+        HttpCompletionOption completionOption = HttpCompletionOption.ResponseContentRead
     )
     {
         options ??= new InvokeFunctionOptions();
@@ -178,7 +197,7 @@ private async Task<HttpResponseMessage> HandleRequest(
 
         try
         {
-            var response = await RetryExecutor.SendAsync(this.httpClient, () => BuildRequestMessage(options, uri), this.Options.Retry, linkedCts.Token);
+            var response = await RetryExecutor.SendAsync(this.httpClient, () => BuildRequestMessage(options, uri), this.Options.Retry, completionOption, linkedCts.Token);
             statusCode = (int) response.StatusCode;
             var isRelayError = response.Headers.Contains("x-relay-error");
             activity.SetHttpResponseTags(statusCode.Value);
@@ -199,7 +218,9 @@ private async Task<HttpResponseMessage> HandleRequest(
                 errorType = statusCode.Value.ToString();
             }
 
-            var content = await response.Content.ReadAsStringAsync();
+            var content = completionOption == HttpCompletionOption.ResponseContentRead
+                ? await response.Content.ReadAsStringAsync()
+                : await BufferErrorBody(response, linkedCts.Token);
             var exception = new FunctionsException(content)
             {
                 Content = content,
@@ -229,6 +250,31 @@ private async Task<HttpResponseMessage> HandleRequest(
         }
     }
 
+    // Keep the error body readable through FunctionsException.Response.
+    private static async Task<string> BufferErrorBody(HttpResponseMessage response, CancellationToken cancellationToken)
+    {
+        try
+        {
+            using var buffer = new MemoryStream();
+            var stream = await response.Content.ReadAsStreamAsync();
+            await stream.CopyToAsync(buffer, cancellationToken);
+
+            var buffered = new ByteArrayContent(buffer.ToArray());
+            foreach (var header in response.Content.Headers)
+                buffered.Headers.TryAddWithoutValidation(header.Key, header.Value);
+
+            response.Content.Dispose();
+            response.Content = buffered;
+
+            return await buffered.ReadAsStringAsync();
+        }
+        catch
+        {
+            response.Dispose();
+            throw;
+        }
+    }
+
     private static Uri BuildUri(string url)
     {
         var builder = new UriBuilder(url);

packages/Functions/Functions/Interfaces/IFunctionsClient.cs

@@ -40,4 +40,14 @@ public interface IFunctionsClient : IGettableHeaders
     /// <param name="cancellationToken"></param>
     /// <returns></returns>
     Task<HttpContent> RawInvoke(string url, string? token = null, Client.InvokeFunctionOptions? options = null, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Invokes a function and returns a successful response after its headers arrive.
+    /// </summary>
+    /// <param name="url">Function name, appended to the base URL.</param>
+    /// <param name="token">Bearer token.</param>
+    /// <param name="options">Invocation options.</param>
+    /// <param name="cancellationToken">Cancels the request and error-body reads. Cancel successful body reads separately.</param>
+    /// <returns>The response, which the caller must dispose.</returns>
+    Task<HttpResponseMessage> InvokeStream(string url, string? token = null, Client.InvokeFunctionOptions? options = null, CancellationToken cancellationToken = default);
 }

packages/Functions/Functions/PublicAPI.Unshipped.txt

@@ -1 +1,3 @@
 #nullable enable
+Supabase.Functions.Client.InvokeStream(string! functionName, string? token = null, Supabase.Functions.Client.InvokeFunctionOptions? options = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<System.Net.Http.HttpResponseMessage!>!
+Supabase.Functions.Interfaces.IFunctionsClient.InvokeStream(string! url, string? token = null, Supabase.Functions.Client.InvokeFunctionOptions? options = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<System.Net.Http.HttpResponseMessage!>!

packages/Functions/README.md

@@ -58,6 +58,19 @@ var options = new Client.InvokeFunctionOptions
 var result = await client.Invoke<MyResponse>("hello-world", token: SUPABASE_ANON_KEY, options);
 ```
 
+### Streaming a response
+
+`InvokeStream` returns successful responses after the headers arrive. `HttpTimeout` and request cancellation
+cover the request and error-body reads. Cancel successful body reads separately and dispose the response when finished.
+
+```csharp
+using var response = await client.InvokeStream("chat", token: SUPABASE_ANON_KEY);
+using var reader = new StreamReader(await response.Content.ReadAsStreamAsync());
+string? line;
+while ((line = await reader.ReadLineAsync()) != null)
+    Console.WriteLine(line);
+```
+
 ## Observability (OpenTelemetry)
 
 The client emits traces and metrics through `System.Diagnostics`, so you can wire them into

sdk-compliance.yaml

@@ -393,10 +393,10 @@ features:
     symbols:
       - InvokeFunctionOptions.HttpMethod
   functions.invocation.streaming_response:
-    status: partially_implemented
-    note: "RawInvoke returns raw HttpContent, but the request uses the default buffered completion (no incremental text/event-stream streaming)."
+    status: implemented
+    note: "InvokeStream returns response bodies for incremental reading."
     symbols:
-      - Client.RawInvoke
+      - Client.InvokeStream
   functions.invocation.region_selection:
     status: implemented
     note: "Full AWS region list; per-call override or client default; emits x-region."
@@ -628,4 +628,4 @@ features:
     symbols:
       - FileOptions.Metadata
       - FileOptions.CacheControl
-      - FileOptions.ContentType
+      - FileOptions.ContentType