Fix Pending Request Queue Async Memory Usage Issues

[nathanwoctopusdeploy]

Background

The PendingRequestQueueAsync had to significantly change the way locking worked from the PendingRequestQueue when converted to be async.

This PR fixes some memory leaks identified when testing async Halibut in server.

• AsyncManualResetEvent.WaitAsync should be cancelled if used as part of a Task.WhenAny if passed a cancellation token or no cancellation token should be passed to avoid a memory leak. Not passing a cancellation token if not required is the best scenario as it avoids the creation of a cancellation token source and a Task.WhenAny
• Task.Delay should be cancelled when used in a Task.WhenAny to avoid leaking timers
• Ensure CancellationTokenSource.CreateLinkedCancellationToken is disposed
• PendingRequest used a SemaphoreSlim which was not disposed

This PR also adds a change to delete a PendingRequest from the queue when the code that queued the pending request has stopped waiting (cancelled or timed out). This allows the request to be disposed and also removes a known issues where null may be returned for DequeuNextAsync while there are items waiting to be dequeued and removes an un-necessary poll loop from the Polling Service to get the next request.

Related to https://github.com/OctopusDeploy/Issues/issues/8266

How to review this PR

Quality :heavy_check_mark:

Pre-requisites

• [ ] I have read How we use GitHub Issues for help deciding when and where it's appropriate to make an issue.
• [ ] I have considered informing or consulting the right people, according to the ownership map.
• [ ] I have considered appropriate testing for my change.

[nathanwoctopusdeploy]

[sc-56801]

[shortcut-integration[bot]]

This pull request has been linked to Shortcut Story #56801: Fix Pending Request Queue Async AsyncManualResetEvent leaks and CancellationTokenSourceLeaks.

[LukeButters]

The bug in the queue can be reproduced with:

[Test]
        public async Task leakTheQueue()
        {
            var queue = new PendingRequestQueueBuilder().WithSyncOrAsync(SyncOrAsync.Async).WithEndpoint("poll://foo")
                .WithPollingQueueWaitTimeout(TimeSpan.FromMilliseconds(1))
                .Build();

            var cts = new CancellationTokenSource();
            var tasks = new List<Task>();
            for (int i = 0; i < 2000; i++)
            {
                tasks.Add(TryDequeue(queue, cts.Token));
            }

            await Task.WhenAll(tasks.ToArray());

        }

        static async Task TryDequeue(IPendingRequestQueue queue, CancellationToken cancellationToken)
        {
            for (int i = 0; i < 6520000; i++)
            {
                await queue.DequeueAsync(cancellationToken);
            }
        }