In PrepareBulk: System.InvalidOperationException: Collection was modified; enumeration operation may not execute.

[RedwoodForest]

Hi,

Using log4stash 3.0.4 we saw this unhandled exception thrown from the library:

System.AggregateException: A Task's exception(s) were not observed either by Waiting on the Task or accessing its Exception property. As a result, the unobserved exception was rethrown by the finalizer thread. ---> System.InvalidOperationException: Collection was modified; enumeration operation may not execute.
   at System.ThrowHelper.ThrowInvalidOperationException(ExceptionResource resource)
   at System.Collections.Generic.List`1.Enumerator.MoveNextRare()
   at log4stash.ElasticClient.RequestFactory.PrepareBulk(IEnumerable`1 bulk)
   at log4stash.ElasticClient.RequestFactory.PrepareRequest(IEnumerable`1 bulk)
   at log4stash.WebElasticClient.<IndexBulkAsync>d__12.MoveNext()
   --- End of inner exception stack trace ---
---> (Inner Exception #0) System.InvalidOperationException: Collection was modified; enumeration operation may not execute.
   at System.ThrowHelper.ThrowInvalidOperationException(ExceptionResource resource)
   at System.Collections.Generic.List`1.Enumerator.MoveNextRare()
   at log4stash.ElasticClient.RequestFactory.PrepareBulk(IEnumerable`1 bulk)
   at log4stash.ElasticClient.RequestFactory.PrepareRequest(IEnumerable`1 bulk)
   at log4stash.WebElasticClient.<IndexBulkAsync>d__12.MoveNext()<---

One plausible cause for this is that LogBulkSet does not prevent ResetBulk from returning a List that is currently being modified by AddEventToBulk because AddEventToBulk (and Count) use a lock on the List instance but ResetBulk uses Interlocked.Exchange. I think this means that a sequence something like this could occur:

1. AddEventToBulk is called and executes as usual down to _bulk.Add(operation).
2. _bulk.Add(operation) starts executing but does not finish immediately (perhaps because it has to allocate more memory for the list)
3. While _bulk.Add(operation) is in progress ResetBulk() is called by ElasticSearchAppender. ResetBulk() returns the List currently being modified by _bulk.Add(operation) from step 2.
4. ElasticSearchAppender calls IndexBulkAsync, which calls _requestFactory.PrepareRequest, which calls PrepareBulk(bulk), which starts enumerating the List.
5. While PrepareBulk is still enumerating the List the _bulk.Add(operation) from step 2 finishes, modifying the list.
6. This results in the InvalidOperationException being thrown.

This is a somewhat convoluted sequence of events, but as far as I can tell it could happen.

If this is indeed the issue then one possible fix is to switch to using a dedicated lock instance and changing ResetBulk() to lock on this dedicated instance instead of using Interlocked.Exchange. That will ensure that any in-progress modifications to the List will be completed before the List is returned by ResetBulk().

I appreciate all the hard work on this library!