ResourceNotificationService.WatchAsync is not thread safe

[afscrome]

If multiple consumers call ResourceNotificationService.WatchAsync in parallel, it looks like some of the subscriptions get lost.

Context in which I encountered this - I've been hitting a intermittent deadlocks (roughly 1 in 5 starts) with WaitForDependencies since I added my own listener to WatchAsync in which resources get stuck in the Waiting state indefinitely. WaitForDependenciesRunningHook makes it's call to WatchAsync during Task.Run, and I believe that can get delayed long enough to conflict with my own WatchAsync call in a different IDistributedApplicationLifecycleHook.

https://github.com/davidfowl/WaitForDependenciesAspire/blob/22e32d218943bd3b15e85c2f04e6415acff6054f/WaitForDependencies.Aspire.Hosting/WaitForDependenciesExtensions.cs#L146

I suspect the root problem is due to the following not being thread safe:

aspire/src/Aspire.Hosting/ApplicationModel/ResourceNotificationService.cs

Line 128 in 0f4cc84

OnResourceUpdated += WriteToChannel;

Repro Code:

int watchers = 0;
int executions = 0;
async Task Watch()
{
    await Task.Yield();
    var asyncEnumerable = notificationService.WatchAsync(default);
    Interlocked.Increment(ref watchers);
    await foreach (var item in asyncEnumerable)
    {
        Interlocked.Increment(ref executions);
        break;
    }
}

var tasks = Enumerable.Range(0, 100).Select(x => Watch()).ToArray();
await Task.Delay(TimeSpan.FromSeconds(1));
await notificationService.PublishUpdateAsync(resource, state => state with { Properties = state.Properties.Add(new("A", "value")) });
await Task.Delay(TimeSpan.FromSeconds(2));

Console.WriteLine($"Watchers: {watchers}");
Console.WriteLine($"Executions: {executions}");

class CustomResource(string name) : Resource(name),
    IResourceWithEnvironment,
    IResourceWithConnectionString,
    IResourceWithEndpoints
{
    public ReferenceExpression ConnectionStringExpression =>
        ReferenceExpression.Create($"CustomConnectionString");
}

Expected Results:

Watchers: 100
Executions: 100

Actual Results

Watchers: 100
Executions: 75

(Execution number varies from run to run, but is always below 100.)

[mitchdenny]

Putting this on the backlog for now. It isn't clear whether this is an issue with WaitForDependencies or soemthing inside the AppModel. When we get around to doing WaitForDependencies within the main tree itself then we can watch out for this.

[afscrome]

@mitchdenny the repro code above doesn’t use WaitForDependencies, only the app model. I actually derived the repo from one of the existing unit tests for ResourceNotificationService.

The repo spawns 100 concurrent tasks that all call notificationService.WatchAsync. When PublishUpdateAsync is called, only some of the subscribers get updates.

[davidfowl]

Fixed in #5515