test(integration): edge removal stops flow; node removal preserves stream

tests/VolcanicCore.Integration.Tests/Scenarios/EdgeRemovalTests.cs

@@ -0,0 +1,74 @@
+using System.Text;
+using NATS.Client.Core;
+using NATS.Client.JetStream;
+using NATS.Client.JetStream.Models;
+using VolcanicCore.Integration.Tests.Fixtures;
+using Xunit;
+
+namespace VolcanicCore.Integration.Tests.Scenarios;
+
+public class EdgeRemovalTests : IAsyncLifetime
+{
+    private readonly RealNatsFixture _fx = new();
+    public Task InitializeAsync() => _fx.InitializeAsync();
+    public Task DisposeAsync() => _fx.DisposeAsync();
+
+    [Fact]
+    public async Task Removing_source_stops_flow_but_preserves_past_messages()
+    {
+        await _fx.Js.CreateStreamAsync(new StreamConfig
+        {
+            Name = "A",
+            Subjects = new List<string> { "a.>" },
+            Storage = StreamConfigStorage.File,
+        });
+        await _fx.Js.CreateStreamAsync(new StreamConfig
+        {
+            Name = "B",
+            Subjects = new List<string> { "b.>" },
+            Storage = StreamConfigStorage.File,
+            Sources = new List<StreamSource>
+            {
+                new()
+                {
+                    Name = "A",
+                    SubjectTransforms = new List<SubjectTransform>
+                    {
+                        new() { Src = "a.>", Dest = "b.>" }
+                    }
+                }
+            }
+        });
+
+        for (var i = 0; i < 10; i++)
+            await _fx.Js.PublishAsync($"a.{i}", Encoding.UTF8.GetBytes($"m{i}"));
+
+        // Wait for the first batch to propagate
+        var streamB = await _fx.Js.GetStreamAsync("B");
+        for (var i = 0; i < 30; i++)
+        {
+            await streamB.RefreshAsync();
+            if (streamB.Info.State.Messages >= 10) break;
+            await Task.Delay(200);
+        }
+        await streamB.RefreshAsync();
+        Assert.Equal(10u, streamB.Info.State.Messages);
+
+        // Now update B to remove its source
+        await _fx.Js.UpdateStreamAsync(new StreamConfig
+        {
+            Name = "B",
+            Subjects = new List<string> { "b.>" },
+            Storage = StreamConfigStorage.File,
+            Sources = new List<StreamSource>(),
+        });
+
+        // Publish 10 more
+        for (var i = 10; i < 20; i++)
+            await _fx.Js.PublishAsync($"a.{i}", Encoding.UTF8.GetBytes($"m{i}"));
+
+        await Task.Delay(1500);
+        await streamB.RefreshAsync();
+        Assert.Equal(10u, streamB.Info.State.Messages); // still just the first 10
+    }
+}

tests/VolcanicCore.Integration.Tests/Scenarios/NodeRemovalPreservesStreamTests.cs

@@ -0,0 +1,38 @@
+using System.Text;
+using NATS.Client.JetStream;
+using NATS.Client.JetStream.Models;
+using VolcanicCore.Integration.Tests.Fixtures;
+using Xunit;
+
+namespace VolcanicCore.Integration.Tests.Scenarios;
+
+public class NodeRemovalPreservesStreamTests : IAsyncLifetime
+{
+    private readonly RealNatsFixture _fx = new();
+    public Task InitializeAsync() => _fx.InitializeAsync();
+    public Task DisposeAsync() => _fx.DisposeAsync();
+
+    [Fact]
+    public async Task Stream_survives_when_the_graph_forgets_about_it()
+    {
+        await _fx.Js.CreateStreamAsync(new StreamConfig
+        {
+            Name = "SURVIVOR",
+            Subjects = new List<string> { "survivor.>" },
+            Storage = StreamConfigStorage.File,
+        });
+        for (var i = 0; i < 5; i++)
+            await _fx.Js.PublishAsync($"survivor.msg.{i}", Encoding.UTF8.GetBytes($"m{i}"));
+
+        // The reconciler "forgets" about the node (SQLite deletion has no effect on NATS).
+        // Simulate a re-adoption on a fresh backend by opening a new context — the stream
+        // remains discoverable.
+        var names = new List<string>();
+        await foreach (var s in _fx.Js.ListStreamsAsync()) names.Add(s.Info.Config.Name);
+        Assert.Contains("SURVIVOR", names);
+
+        var stream = await _fx.Js.GetStreamAsync("SURVIVOR");
+        await stream.RefreshAsync();
+        Assert.Equal(5u, stream.Info.State.Messages);
+    }
+}