Concurrent writes and deletes can lead to inconsistent state

# Part 1: The problem
## Summary

The `WritePoints` and `DeleteMeasurement` paths have insufficient locking, and concurrent calls can cause a situation where the recorded type of the data in the `MeasurementFields` does not match the data on disk.

## Mechanism

Consider this sequence of events:

- Write 1 of `cpu value=1.0` starts.
- Write 1 creates a measurement fields entry with type `float`.
- `DeleteMeasurement` is called on `cpu`, removing that field type.
- Write 2 of `cpu value=1i` starts.
- Write 2 creates a measurement fields entry with type `integer`.
- Write 1 writes points to the `Engine`.
- Write 2 writes points to the `Engine`.

## Repro

In the appendix (at the end) is a patch that when applied will cause this bug to trigger. The patch forces the scheduling of multiple writes so that the above situation happens. It does this with a `SWAP` channel, where one of the goroutines initially starts blocked on reading it. Every time a send and receive happens, which goroutine is running swaps.

## Extra concerns

I believe these delete races are rather ingrained in the code base. The `Shard` has no synchronization on any of its delete methods (`DeleteSeriesRange`, `DeleteSeriesRangeWithPredicate`, or `DeleteMeasurement`). Indeed, each of those methods calls directly into the `Engine`, which all funnel into `DeleteSeriesRangeWithPredicate` and finally `deleteSeriesRange` in the case of `tsm1`, which contains no locking. While it is composed of many pieces (a `FileStore`, `Cache`, `WAL`, etc.) that do lock, there is no locking *between* them, causing the analysis to be very difficult. Some of these problems are understood and pointed out in the comments:

```
// Note: this is inherently racy if writes are occurring to the same measurement/series are
// being removed. A write could occur and exist in the cache at this point, but we
// would delete it from the index.
```

Given all these concerns, I think that in order to fully fix this, we need to have the `Engine`/`Shard` exclude deletes for series that are currently being written to. In general, I believe this will be easier if the updates/validation with the `SeriesFile` and `MeasurementFields` to move from the `Shard` to the `Engine`, as that is the place that must handle the deletes: the `Shard` is not privvy to what is being deleted.

---
# Part 2: What to do

## Possible solutions

| | Action | Positives | Negatives |
|-|-|-|-|
|1| Big honkin' write lock on delete |

• Simple, obviously correct
• Only expensive during deletes

• Limits scalability of deletes
• Any delete would require stopping all writes
|
|2| Lock pool on measurements |
• Granular locking
• Can do tricks like [sharded mutexes](https://github.com/jonhoo/drwmutex) if necessary
|
• Can be allocation heavy
• Punishing the common case
• Lock ordering issues
• The pool could hurt scalability

|3| Delayed lock pool on measurements (think write barrier with gc) |
• Same as lock pool on measurements
• Defers needing to acquire the locks until a delete call is made
|
• Can be allocation heavy
• Lock ordering issues
• Keeping track of writes that have not acquired locks can hurt scalability
• Degraded performance while deletes are happening

## Discussion on solutions

### 1. Write lock on delete

Right now we grab an `RLock` for most of `WritePoints`. If we grab a `Lock` during `deleteSeriesRange`, then the race will be fixed: you can't delete the created fields until after the points have been written. This means only one delete at a time, and no writes while deletes happen. Additionally, any incoming writes will be delayed while the `Lock` call is waiting for any `RLock`s to finish. Untenable, but presented so that the problem can be better understood.

### 2. Lock pool on measurements

Instead of having a single `RLock` per `WritePoints`, assume that most writes and deletes operate on different sets of measurements and assign one `RWMutex` to each measurement. During a write, we would acquire each measurements `RLock` in sorted order. Similarly, in the `deleteSeriesRange`, we have access to all of the series keys that we will possibly be deleting, and so we can sort and acquire the appropriate `Lock`s. Unfortunately, there is not a bounded set of measurements that we can know upfront, which means some sort of pool. In order to keep track of the locks, the pool would either use some form of weak pointer (hello dark arts), or just reference counting with explicit replaces (probably with deferred cleanup to avoid reallocating a mutex constantly). Making this scalable and cheap will be a challenge, but there exist scalable (cpu sharded) mutexes, etc., that should avoid contention in the read case.

Additionally, the assumption about measurements being typically distinct may not be true. But, since both writes and deletes have access to a full series key, we could make the locks as granular as needed.

### 3. Delayed lock pool on measurements (think write barrier with gc)

One of main challenges with solution number 2 is the cheap lock pool and forcing writes to pay for locks when none should be required most of the time. Instead, we can do a strategy akin to the write barrier in the garbage collector. We could keep track of the number of concurrent writes and deletes in flight at a time. When a write comes in, if there are no concurrent deletes, it can proceed without acquiring any mutexes. Otherwise, it would grab mutexes as described above. When a delete comes in, it would mark that there was a concurrent delete, wait for any concurrent writes that have not grabbed mutexes, and then proceed as described above. This stategy can be implemented with two counters and a condition variable, and should have lower overhead when there are no deletes, which is hopefully the common case. If these counters and condition variable cause too much contention, I have some wacky ideas.

# Part 3: Conclusion

## What does it all mean?

I dunno, I felt like it needed to have some sort of closing.

---
# Part 4: Appendix

###### Repro patch
```
diff --git a/tsdb/shard.go b/tsdb/shard.go
index 6e5827c7a..0cadf6964 100644
--- a/tsdb/shard.go
+++ b/tsdb/shard.go
@@ -31,6 +31,8 @@ import (
"go.uber.org/zap"
)

+var SWAP = make(chan struct{})
+
const (
statWriteReq = "writeReq"
statWriteReqOK = "writeReqOk"
@@ -512,6 +514,9 @@ func (s *Shard) WritePoints(points []models.Point) error {
return err
}

+ SWAP <- struct{}{}
+ <-SWAP
+
// Write to the engine.
if err := engine.WritePoints(points); err != nil {
atomic.AddInt64(&s.stats.WritePointsErr, int64(len(points)))
diff --git a/tsdb/shard_test.go b/tsdb/shard_test.go
index a70554350..e77ed74f6 100644
--- a/tsdb/shard_test.go
+++ b/tsdb/shard_test.go
@@ -528,12 +528,6 @@ func TestShard_WritePoints_FieldConflictConcurrentQuery(t *testing.T) {
time.Unix(int64(i), 0),
))
} else {
- points = append(points, models.MustNewPoint(
- "cpu",
- models.NewTags(map[string]string{"host": "server"}),
- map[string]interface{}{"value": int64(1)},
- time.Unix(int64(i), 0),
- ))
}
}

@@ -543,6 +537,10 @@ func TestShard_WritePoints_FieldConflictConcurrentQuery(t *testing.T) {
}

sh.WritePoints(points)
+
+ tsdb.SWAP <- struct{}{}
+ <-tsdb.SWAP
+
m := &influxql.Measurement{Name: "cpu"}
iter, err := sh.CreateIterator(context.Background(), m, query.IteratorOptions{
Expr: influxql.MustParseExpr(`value`),
@@ -589,15 +587,11 @@ func TestShard_WritePoints_FieldConflictConcurrentQuery(t *testing.T) {
time.Unix(int64(i), 0),
))
} else {
- points = append(points, models.MustNewPoint(
- "cpu",
- models.NewTags(map[string]string{"host": "server"}),
- map[string]interface{}{"value": 1.0},
- time.Unix(int64(i), 0),
- ))
}
}
for i := 0; i < 500; i++ {
+ <-tsdb.SWAP
+
if err := sh.DeleteMeasurement([]byte("cpu")); err != nil {
errC <- err
}
@@ -630,6 +624,8 @@ func TestShard_WritePoints_FieldConflictConcurrentQuery(t *testing.T) {
}
iter.Close()
}
+
+ tsdb.SWAP <- struct{}{}
}
errC <- nil
}()
```

Start with tsdb/shard.go, especially Shard.WritePoints and the delete paths, then inspect the concurrency repro in tsdb/shard_test.go and the Engine delete flow described in the issue. Compare the proposed locking strategies and establish how synchronization should cover MeasurementFields and on-disk writes. Done means concurrent writes and deletes no longer leave inconsistent field types, with a regression test covering the repro.

Written by the indexing model from the issue text.