Fix duplicate resource stopped events during restart

src/Aspire.Hosting/Dcp/DcpExecutor.cs

@@ -1028,6 +1028,7 @@ private async Task CreateResourceReplicasAsync<TDcpResource, TContext>(
                 await _executorEvents.PublishAsync(new OnResourceChangedContext(
                     _shutdownCancellation.Token, resourceType, modelResource,
                     r.DcpResourceName, new ResourceStatus(null, null, null),
+                    PreviousState: null,
                     snapshotBuild)
                 ).ConfigureAwait(false);
             }
@@ -1046,6 +1047,7 @@ await _executorEvents.PublishAsync(new OnResourceChangedContext(
                         cancellationToken, resourceType, modelResource,
                         r.DcpResource.Metadata.Name,
                         new ResourceStatus(KnownResourceStates.NotStarted, null, null),
+                        PreviousState: null,
                         s => s with
                         {
                             State = new ResourceStateSnapshot(KnownResourceStates.NotStarted, null)
@@ -1066,6 +1068,7 @@ await _executorEvents.PublishAsync(new OnResourceChangedContext(
                         cancellationToken, resourceType, modelResource,
                         r.DcpResource.Metadata.Name,
                         new ResourceStatus(KnownResourceStates.NotStarted, null, null),
+                        PreviousState: null,
                         s => s with
                         {
                             State = new ResourceStateSnapshot(KnownResourceStates.NotStarted, null)

src/Aspire.Hosting/Dcp/DcpExecutorEvents.cs

@@ -11,7 +11,7 @@ internal record OnEndpointsAllocatedContext(CancellationToken CancellationToken)
 internal record OnResourceStartingContext(CancellationToken CancellationToken, string ResourceType, IResource Resource, string? DcpResourceName);
 internal record OnConnectionStringAvailableContext(CancellationToken CancellationToken, IResource Resource);
 internal record OnResourcesPreparedContext(CancellationToken CancellationToken);
-internal record OnResourceChangedContext(CancellationToken CancellationToken, string ResourceType, IResource Resource, string DcpResourceName, ResourceStatus Status, Func<CustomResourceSnapshot, CustomResourceSnapshot> UpdateSnapshot);
+internal record OnResourceChangedContext(CancellationToken CancellationToken, string ResourceType, IResource Resource, string DcpResourceName, ResourceStatus Status, string? PreviousState, Func<CustomResourceSnapshot, CustomResourceSnapshot> UpdateSnapshot);
 internal record OnResourceFailedToStartContext(CancellationToken CancellationToken, string ResourceType, IResource Resource, string? DcpResourceName, string? ErrorMessage = null);
 
 internal sealed class DcpExecutorEvents

src/Aspire.Hosting/Dcp/DcpResourceWatcher.cs

@@ -270,6 +270,11 @@ public async ValueTask DisposeAsync()
 
     private async Task ProcessResourceChange<T>(WatchEventType watchEventType, T resource, ConcurrentDictionary<string, T> resourceByName, string resourceKind, Func<T, CustomResourceSnapshot, CustomResourceSnapshot> snapshotFactory) where T : CustomResource, IKubernetesStaticMetadata
     {
+        // Read the DCP state before replacing the cached object. The published snapshot can
+        // already say Waiting or Starting if a stopped handler has requested a restart.
+        var previousState = resourceByName.TryGetValue(resource.Metadata.Name, out var previousResource)
+            ? GetResourceStatus(previousResource).State
+            : null;
         var resourceChange = ProcessResourceChange(resourceByName, watchEventType, resource);
         if (resourceChange != ResourceChangeResult.Ignored)
         {
@@ -349,7 +354,9 @@ private async Task ProcessResourceChange<T>(WatchEventType watchEventType, T res
                         _allLogsFlushed.TryRemove(resource.Metadata.Name, out _);
                     }
 
-                    await _executorEvents.PublishAsync(new OnResourceChangedContext(_shutdownToken, resourceType, appModelResource, resource.Metadata.Name, status, s => snapshotFactory(resource, s))).ConfigureAwait(false);
+                    await _executorEvents.PublishAsync(new OnResourceChangedContext(_shutdownToken, resourceType, appModelResource, resource.Metadata.Name, status,
+                        resourceChange == ResourceChangeResult.Replaced ? null : previousState,
+                        s => snapshotFactory(resource, s))).ConfigureAwait(false);
 
                     if (logsAvailable)
                     {
@@ -953,7 +960,7 @@ private async ValueTask TryRefreshResource(string resourceKind, string resourceN
                 _resourceState.ApplicationModel.TryGetValue(appModelResourceName, out var appModelResource))
             {
                 var status = GetResourceStatus(cr);
-                await _executorEvents.PublishAsync(new OnResourceChangedContext(_shutdownToken, resourceKind, appModelResource, resourceName, status, s =>
+                await _executorEvents.PublishAsync(new OnResourceChangedContext(_shutdownToken, resourceKind, appModelResource, resourceName, status, status.State, s =>
                 {
                     if (cr is Container container)
                     {

src/Aspire.Hosting/Orchestrator/ApplicationOrchestrator.cs

@@ -579,27 +579,19 @@ private async Task OnResourceEndpointsAllocated(ResourceEndpointsAllocatedEvent
 
     private async Task OnResourceChanged(OnResourceChangedContext context)
     {
-        // Get the previous state before updating to detect transitions to stopped states
-        string? previousState = null;
-        if (_notificationService.TryGetCurrentState(context.DcpResourceName, out var previousResourceEvent))
-        {
-            previousState = previousResourceEvent.Snapshot.State?.Text;
-        }
-
         await _notificationService.PublishUpdateAsync(context.Resource, context.DcpResourceName, context.UpdateSnapshot).ConfigureAwait(false);
 
         if (context.ResourceType == KnownResourceTypes.Container)
         {
             await SetChildResourceAsync(context.Resource, context.Status.State, context.Status.StartupTimestamp, context.Status.FinishedTimestamp).ConfigureAwait(false);
         }
 
-        // Check if the resource has transitioned to a terminal/stopped state
-        var currentState = context.Status.State;
-        if (currentState is not null &&
-            KnownResourceStates.TerminalStates.Contains(currentState) &&
-            previousState != currentState &&
-            (previousState is null ||
-            !KnownResourceStates.TerminalStates.Contains(previousState)))
+        // Use the previous DCP state from the context, not the published snapshot. A stopped handler
+        // can restart the resource and change the snapshot to Waiting or Starting while DCP still
+        // reports the same terminal state, which would otherwise fire another stopped event.
+        if (context.Status.State is { } state &&
+            KnownResourceStates.TerminalStates.Contains(state) &&
+            !KnownResourceStates.TerminalStates.Contains(context.PreviousState))
         {
             // Get the current state from notification service after the update
             if (_notificationService.TryGetCurrentState(context.DcpResourceName, out var currentResourceEvent))

tests/Aspire.Hosting.Tests/Dcp/DcpExecutorTests.cs

@@ -2764,6 +2764,119 @@ void AddLogLines(IReadOnlyList<LogLine> batch)
         }
     }
 
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task ResourceWatch_ReportsPreviousStateForChangesRefreshesAndReplacements(bool reportDeletion)
+    {
+        var builder = DistributedApplication.CreateBuilder();
+        builder.AddContainer("database", "image");
+        var kubernetesService = new TestKubernetesService();
+        var changes = Channel.CreateUnbounded<OnResourceChangedContext>();
+        var events = new DcpExecutorEvents();
+        events.Subscribe<OnResourceChangedContext>(context =>
+        {
+            if (context.Resource.Name == "database" && context.Status.State is not null)
+            {
+                changes.Writer.TryWrite(context);
+            }
+
+            return Task.CompletedTask;
+        });
+
+        using var app = builder.Build();
+        await using var executor = CreateAppExecutor(
+            app.Services.GetRequiredService<DistributedApplicationModel>(),
+            kubernetesService: kubernetesService, events: events);
+        await executor.RunApplicationAsync().DefaultTimeout();
+
+        var container = Assert.Single(kubernetesService.CreatedResources.OfType<Container>());
+        await PublishStateAsync(ContainerState.Running, previousState: null);
+        await PublishStateAsync(ContainerState.Exited, ContainerState.Running);
+        await PublishStateAsync(ContainerState.Exited, ContainerState.Exited);
+
+        var endpoint = Endpoint.Create("database-endpoint", "", "database-service");
+        endpoint.Metadata.OwnerReferences = [new V1OwnerReference
+        {
+            ApiVersion = container.ApiVersion,
+            Kind = container.Kind,
+            Name = container.Metadata.Name,
+            Uid = container.Metadata.Uid
+        }];
+        kubernetesService.PushResourceModified(endpoint);
+        await AssertChangeAsync(ContainerState.Exited, ContainerState.Exited);
+
+        await PublishStateAsync(ContainerState.Running, ContainerState.Exited);
+        await PublishStateAsync(ContainerState.Exited, ContainerState.Running);
+
+        if (reportDeletion)
+        {
+            kubernetesService.PushResourceDeleted(container);
+        }
+
+        container.Metadata.Uid = "database-replacement";
+        kubernetesService.PushResourceUnchanged(container, k8s.WatchEventType.Added);
+        await AssertChangeAsync(ContainerState.Exited, previousState: null);
+        await PublishStateAsync(ContainerState.Exited, ContainerState.Exited);
+
+        async Task PublishStateAsync(string state, string? previousState)
+        {
+            container.Status = new ContainerStatus { State = state };
+            kubernetesService.PushResourceModified(container);
+            await AssertChangeAsync(state, previousState);
+        }
+
+        async Task AssertChangeAsync(string state, string? previousState)
+        {
+            var change = await changes.Reader.ReadAsync().AsTask().DefaultTimeout();
+            Assert.Equal(state, change.Status.State);
+            Assert.Equal(previousState, change.PreviousState);
+        }
+    }
+
+    [Fact]
+    public async Task ResourceWatch_PreviousStateIsScopedToReplica()
+    {
+        var builder = DistributedApplication.CreateBuilder();
+        AddExecutableWithPrecomputedReplicas(builder);
+        var kubernetesService = new TestKubernetesService();
+        var changes = Channel.CreateUnbounded<OnResourceChangedContext>();
+        var events = new DcpExecutorEvents();
+        events.Subscribe<OnResourceChangedContext>(context =>
+        {
+            if (context.Resource.Name == "program" && context.Status.State is not null)
+            {
+                changes.Writer.TryWrite(context);
+            }
+
+            return Task.CompletedTask;
+        });
+
+        using var app = builder.Build();
+        await using var executor = CreateAppExecutor(
+            app.Services.GetRequiredService<DistributedApplicationModel>(),
+            kubernetesService: kubernetesService, events: events);
+        await executor.RunApplicationAsync().DefaultTimeout();
+
+        var executables = GetCreatedExecutablesForResource(kubernetesService, "program");
+        Assert.Equal(2, executables.Count);
+        await PublishStateAsync(executables[0], ExecutableState.Finished, previousState: null);
+        await PublishStateAsync(executables[1], ExecutableState.Finished, previousState: null);
+        await PublishStateAsync(executables[0], ExecutableState.Running, ExecutableState.Finished);
+        await PublishStateAsync(executables[1], ExecutableState.Finished, ExecutableState.Finished);
+        await PublishStateAsync(executables[0], ExecutableState.Finished, ExecutableState.Running);
+
+        async Task PublishStateAsync(Executable executable, string state, string? previousState)
+        {
+            executable.Status = new ExecutableStatus { State = state };
+            kubernetesService.PushResourceModified(executable);
+            var change = await changes.Reader.ReadAsync().AsTask().DefaultTimeout();
+            Assert.Equal(executable.Metadata.Name, change.DcpResourceName);
+            Assert.Equal(state, change.Status.State);
+            Assert.Equal(previousState, change.PreviousState);
+        }
+    }
+
     [Fact]
     public async Task ResourceWatch_ResourceWithoutResourceVersionIsAlwaysProcessed()
     {

tests/Aspire.Hosting.Tests/Orchestrator/ApplicationOrchestratorTests.cs

@@ -24,6 +24,54 @@ namespace Aspire.Hosting.Tests.Orchestrator;
 [Trait("Partition", "3")]
 public class ApplicationOrchestratorTests(ITestOutputHelper testOutputHelper)
 {
+    [Theory]
+    [InlineData("Waiting")]
+    [InlineData("Starting")]
+    [InlineData("Running")]
+    public async Task ResourceStoppedEventIsNotRepeatedWhenHandlerChangesSnapshotState(string snapshotState)
+    {
+        using var builder = TestDistributedApplicationBuilder.Create(testOutputHelper);
+        var resource = builder.AddExecutable("resource", "unused", ".");
+        var events = new DcpExecutorEvents();
+        var notifications = ResourceNotificationServiceTestHelpers.Create();
+        var stoppedCount = 0;
+
+        resource.OnResourceStopped(async (_, @event, _) =>
+        {
+            Interlocked.Increment(ref stoppedCount);
+            Assert.Equal(KnownResourceStates.Finished, @event.ResourceEvent.Snapshot.State?.Text);
+            Assert.Equal(0, @event.ResourceEvent.Snapshot.ExitCode);
+            await notifications.PublishUpdateAsync(resource.Resource, resource.Resource.Name,
+                snapshot => snapshot with { State = snapshotState });
+        });
+
+        using var app = builder.Build();
+        var model = app.Services.GetRequiredService<DistributedApplicationModel>();
+        var orchestrator = CreateOrchestrator(model, notificationService: notifications, dcpEvents: events, applicationEventing: builder.Eventing);
+        await orchestrator.RunApplicationAsync();
+
+        await PublishStateAsync(null, null);
+        await PublishStateAsync(KnownResourceStates.Running, null);
+        await PublishStateAsync(KnownResourceStates.Finished, KnownResourceStates.Running);
+        Assert.Equal(1, stoppedCount);
+
+        await PublishStateAsync(KnownResourceStates.Finished, KnownResourceStates.Finished);
+        Assert.Equal(1, stoppedCount);
+
+        await PublishStateAsync(KnownResourceStates.Exited, KnownResourceStates.Finished);
+        await PublishStateAsync(KnownResourceStates.FailedToStart, KnownResourceStates.Exited);
+        Assert.Equal(1, stoppedCount);
+
+        await PublishStateAsync(KnownResourceStates.Running, KnownResourceStates.FailedToStart);
+        await PublishStateAsync(KnownResourceStates.Finished, KnownResourceStates.Running);
+        Assert.Equal(2, stoppedCount);
+
+        Task PublishStateAsync(string? state, string? previousState) => events.PublishAsync(new OnResourceChangedContext(
+            CancellationToken.None, KnownResourceTypes.Executable, resource.Resource, resource.Resource.Name,
+            new ResourceStatus(state, null, null), previousState,
+            snapshot => snapshot with { State = state is not null ? new(state, null) : snapshot.State, ExitCode = state == KnownResourceStates.Finished ? 0 : null }));
+    }
+
     [Fact]
     public async Task ParentPropertySetOnChildResource()
     {
@@ -1311,6 +1359,7 @@ await events.PublishAsync(new OnResourceChangedContext(
             parentContainer.Resource,
             "parent-container-dcp",
             new ResourceStatus(KnownResourceStates.FailedToStart, null, null),
+            PreviousState: null,
             snapshot => snapshot with { State = KnownResourceStates.FailedToStart }));
 
         // Check final states
@@ -1358,6 +1407,7 @@ await events.PublishAsync(new OnResourceChangedContext(
             parentContainer.Resource,
             "parent-container-dcp",
             new ResourceStatus(KnownResourceStates.FailedToStart, null, null),
+            PreviousState: null,
             snapshot => snapshot with { State = KnownResourceStates.FailedToStart }));
 
         // Check final states