Make bloom gateway tasks cancelable (and other fixes)

[chaudum]

Why we need it:

A bloom gateway request calling FilterChunkRefs that timed out caused a deadlock of the workers, because when the request context is cancelled, the request returned, but the task continued to process until the block querier tried to write to the results channel. That writing was blocked, because the request that was supposed to read from the results channel to gather the result was gone.

How is is solved:

The PR introduces two changes:

1. The request context is also passed when creating a new Task, which allows to check the cancellation of a request also in the task.
2. The channels of the Task are replaced by a "pipe channel", which consists of a sender channel and a receiver channel. The pipe reads from the sender channel until it is closed, and forwards the received item to the receiver channel, but only if the internal context of the pipe isn't closed.
3. In the worker, when dequeuing and processing the tests, the context of tasks is checked for error, to shortcut operations and return errors early.

---

Further smaller changes/fixes:

1. Fix a potential index out of bounds panic in the code that removes filtered out chunk refs from th original request.
2. Make sure that the DequeueMany call on the task queue returns successive tasks based on their enqueue order.
3. Do not remove cancelled task from task list once they have been partitioned by fingerprint ranged.

[vlad-diachenko]

looks brilliant 💎

[salvacorts]

❤️

[salvacorts]

nit: I'd rather use Send and Receive

[chaudum]

During testing I found another issue with this PR:

2024-01-19 15:08:03.363		/src/enterprise-logs/vendor/github.com/grafana/dskit/services/basic_service.go:119 +0x105
2024-01-19 15:08:03.363	created by github.com/grafana/dskit/services.(*BasicService).StartAsync.func1 in goroutine 772
2024-01-19 15:08:03.363		/src/enterprise-logs/vendor/github.com/grafana/dskit/services/basic_service.go:190 +0x1b4
2024-01-19 15:08:03.363	github.com/grafana/dskit/services.(*BasicService).main(0xc00214cc80)
2024-01-19 15:08:03.363		/src/enterprise-logs/vendor/github.com/grafana/loki/pkg/bloomgateway/worker.go:238 +0x1666
2024-01-19 15:08:03.363	github.com/grafana/loki/pkg/bloomgateway.(*worker).running(0xc000a42700, {0x37eadf8, 0xc0001437c0})
2024-01-19 15:08:03.363		/src/enterprise-logs/vendor/github.com/grafana/loki/pkg/bloomgateway/multiplexing.go:158
2024-01-19 15:08:03.363	github.com/grafana/loki/pkg/bloomgateway.Task.Close(...)
2024-01-19 15:08:03.363		/src/enterprise-logs/vendor/go.uber.org/atomic/bool.go:66
2024-01-19 15:08:03.363	go.uber.org/atomic.(*Bool).CompareAndSwap(...)
2024-01-19 15:08:03.363	goroutine 780 [running]:
2024-01-19 15:08:03.363	
2024-01-19 15:08:03.363	[signal SIGSEGV: segmentation violation code=0x1 addr=0x0 pc=0x22bc6c6]
2024-01-19 15:08:03.363	panic: runtime error: invalid memory address or nil pointer dereference

Not quite sure yet what the nil pointer dereference is caused by.

[owen-d]

skimmed, lgtm

[chaudum]

Going to open a few smaller PRs with individual fixes.

[chaudum]

Superseded by #11792