Stop shattering agent Uris that contain a comma (GH-3733)

src/Testing/CoreTests/Runtime/Agents/agent_command_serialization.cs

@@ -86,4 +86,65 @@ public void AgentsStopped_with_no_agents()
 
         other.AgentUris.ShouldBeEmpty();
     }
+
+    // GH-3733: a comma is a legal URI sub-delimiter (RFC 3986), permitted unescaped in a path segment, and
+    // agent URIs embed operator- and tenant-supplied strings -- an event-subscription URI carries the
+    // projection name and the tenant id. Joining on a comma shattered any such agent into fragments, and
+    // because the read side built the array in one LINQ projection the resulting UriFormatException took out
+    // the confirmation for the ENTIRE batch. The third case below is the one that mattered in production:
+    // two agents, one with a comma, and neither confirmed.
+    public static IEnumerable<object[]> CommaBearingUris()
+    {
+        yield return [new[] { new Uri("event-subscriptions://marten/main/db/proj/all/v1/a,b") }];
+        yield return [new[] { new Uri("event-subscriptions://marten/main/db/proj/all/v1/a, b") }];
+        yield return
+        [
+            new[]
+            {
+                new Uri("event-subscriptions://marten/main/db/proj/all/v1/plain"),
+                new Uri("event-subscriptions://marten/main/db/proj/all/v1/a,b")
+            }
+        ];
+    }
+
+    [Theory]
+    [MemberData(nameof(CommaBearingUris))]
+    public void round_trip_agent_uris_containing_a_comma(Uri[] uris)
+    {
+        roundTrip(new StartAgents(uris)).AgentUris.ShouldBe(uris);
+        roundTrip(new AgentsStarted(uris)).AgentUris.ShouldBe(uris);
+        roundTrip(new StopAgents(uris)).AgentUris.ShouldBe(uris);
+        roundTrip(new AgentsStopped(uris)).AgentUris.ShouldBe(uris);
+    }
+
+    // The comma stays the default on the wire so a rolling upgrade keeps working in both directions: a
+    // 6.24.0 node has to be able to read what a 6.24.1 node writes, and vice versa. Only a payload that
+    // actually contains a comma -- the case that was already broken on both -- switches format.
+    [Fact]
+    public void payloads_without_a_comma_keep_the_legacy_wire_format()
+    {
+        var command = new StartAgents([new Uri("fake://one"), new Uri("fake://two")]);
+
+        System.Text.Encoding.UTF8.GetString(command.Write())
+            .ShouldBe("fake://one/,fake://two/");
+    }
+
+    [Fact]
+    public void reads_a_legacy_comma_joined_payload()
+    {
+        var legacy = System.Text.Encoding.UTF8.GetBytes("fake://one,fake://two,fake://three");
+
+        ((Wolverine.Runtime.Agents.StartAgents)Wolverine.Runtime.Agents.StartAgents.Read(legacy)).AgentUris
+            .ShouldBe([new Uri("fake://one"), new Uri("fake://two"), new Uri("fake://three")]);
+    }
+
+    [Fact]
+    public void an_unreadable_entry_names_itself_and_the_payload()
+    {
+        var garbage = System.Text.Encoding.UTF8.GetBytes("fake://one,%%%not a uri");
+
+        var ex = Should.Throw<FormatException>(() => Wolverine.Runtime.Agents.StartAgents.Read(garbage));
+        ex.Message.ShouldContain("%%%not a uri");
+        ex.Message.ShouldContain("fake://one,%%%not a uri");
+    }
 }

src/Wolverine/Runtime/Agents/StartAgents.cs

@@ -59,6 +59,68 @@ public static int HashOf(Uri[]? uris)
     }
 }
 
+/// <summary>
+///     GH-3733: the wire format for the <c>Uri[]</c> payload every batched agent command carries.
+///
+///     <para>These commands used to join and split on a comma. A comma is a <b>legal</b> URI character —
+///     RFC 3986 lists it as a sub-delim, permitted unescaped in a path segment — and agent URIs embed
+///     operator- and tenant-supplied strings: an event-subscription URI is
+///     <c>event-subscriptions://marten/main/&lt;host&gt;.&lt;db&gt;/&lt;projection&gt;/all/v&lt;n&gt;/&lt;tenant&gt;</c>.
+///     One comma anywhere in one agent URI shattered that agent into fragments, and because the read side
+///     built the whole array in a single projection, the resulting <c>UriFormatException</c> took out the
+///     <b>entire batch</b>, not just the offending entry. For <c>AgentsStarted</c> — the reply that confirms
+///     an agent start — that means the leader gets no confirmation for the whole chunk, which is exactly the
+///     shape of stall reported in GH-3698.</para>
+///
+///     <para>A newline is the delimiter instead: it is not legal unescaped in a URI, and
+///     <see cref="Uri.ToString" /> will never emit a bare one.</para>
+///
+///     <para>The comma nonetheless stays the default on the wire, because it is what every 6.24.0-and-earlier
+///     node both writes and reads, and a rolling upgrade has to keep working in both directions. Only a
+///     payload that actually contains a comma — the case that was already broken — switches to the newline
+///     form, and it announces itself with a leading newline. The two formats can never be confused: a legacy
+///     payload always begins with a URI scheme character.</para>
+/// </summary>
+internal static class AgentUriList
+{
+    public static byte[] Write(Uri[] uris)
+    {
+        var texts = uris.Select(x => x.ToString()).ToArray();
+
+        if (texts.Any(x => x.Contains(',')))
+        {
+            return Encoding.UTF8.GetBytes("\n" + texts.Join("\n"));
+        }
+
+        return Encoding.UTF8.GetBytes(texts.Join(","));
+    }
+
+    public static Uri[] Read(byte[] bytes)
+    {
+        var text = Encoding.UTF8.GetString(bytes);
+        var delimiter = text.StartsWith('\n') ? '\n' : ',';
+
+        var parts = text.Split(delimiter, StringSplitOptions.RemoveEmptyEntries);
+        var uris = new Uri[parts.Length];
+
+        for (var i = 0; i < parts.Length; i++)
+        {
+            // Parsed one at a time so a bad entry names itself. The bare
+            // "Invalid URI: The format of the URI could not be determined." from a LINQ projection over the
+            // whole payload gave no way to tell which agent, or which node, was responsible.
+            if (!Uri.TryCreate(parts[i], UriKind.Absolute, out var uri))
+            {
+                throw new FormatException(
+                    $"Unable to read agent Uri '{parts[i]}' out of an agent command payload. Full payload was '{text}'.");
+            }
+
+            uris[i] = uri;
+        }
+
+        return uris;
+    }
+}
+
 internal record AgentsStarted(Uri[] AgentUris) : IAgentCommand, ISerializable
 {
     public Task<AgentCommands> ExecuteAsync(IWolverineRuntime runtime, CancellationToken cancellationToken)
@@ -73,14 +135,12 @@ public virtual bool Equals(AgentsStarted? other)
 
     public byte[] Write()
     {
-        return Encoding.UTF8.GetBytes(AgentUris.Select(x => x.ToString()).Join(","));
+        return AgentUriList.Write(AgentUris);
     }
 
     public static object Read(byte[] bytes)
     {
-        var uris = Encoding.UTF8.GetString(bytes).Split(',', StringSplitOptions.RemoveEmptyEntries)
-            .Select(x => new Uri(x)).ToArray();
-        return new AgentsStarted(uris);
+        return new AgentsStarted(AgentUriList.Read(bytes));
     }
 }
 
@@ -183,14 +243,12 @@ public override string ToString()
 
     public byte[] Write()
     {
-        return Encoding.UTF8.GetBytes(AgentUris.Select(x => x.ToString()).Join(","));
+        return AgentUriList.Write(AgentUris);
     }
 
     public static object Read(byte[] bytes)
     {
-        var agents = Encoding.UTF8.GetString(bytes).Split(',', StringSplitOptions.RemoveEmptyEntries)
-            .Select(x => new Uri(x)).ToArray();
-        return new StartAgents(agents);
+        return new StartAgents(AgentUriList.Read(bytes));
     }
 }
 
@@ -208,14 +266,12 @@ public virtual bool Equals(AgentsStopped? other)
 
     public byte[] Write()
     {
-        return Encoding.UTF8.GetBytes(AgentUris.Select(x => x.ToString()).Join(","));
+        return AgentUriList.Write(AgentUris);
     }
 
     public static object Read(byte[] bytes)
     {
-        var uris = Encoding.UTF8.GetString(bytes).Split(',', StringSplitOptions.RemoveEmptyEntries)
-            .Select(x => new Uri(x)).ToArray();
-        return new AgentsStopped(uris);
+        return new AgentsStopped(AgentUriList.Read(bytes));
     }
 }
 
@@ -253,13 +309,11 @@ public override string ToString()
 
     public byte[] Write()
     {
-        return Encoding.UTF8.GetBytes(AgentUris.Select(x => x.ToString()).Join(","));
+        return AgentUriList.Write(AgentUris);
     }
 
     public static object Read(byte[] bytes)
     {
-        var agents = Encoding.UTF8.GetString(bytes).Split(',', StringSplitOptions.RemoveEmptyEntries)
-            .Select(x => new Uri(x)).ToArray();
-        return new StopAgents(agents);
+        return new StopAgents(AgentUriList.Read(bytes));
     }
 }