feat(dot/sync): improve worker pool

[haikoschol]

The main difference in the worker pool API is that SubmitBatch() does not block until the whole batch has been processed. Instead, it returns an ID which can be used to retrieve the current state of the batch. In addition, Results() returns a channel over which task results are sent as they become available.

The main improvement this brings is increased concurrency, since results can be processed before the whole batch has been completed.

What has not changed is the overall flow of the Strategy interface; getting a new batch of tasks with NextActions() and processing the results with Process().

Changes

• replaced the code in dot/sync/worker_pool.go
• adapted SyncService to the API changes of the new worker pool
• adapted some expectations in tests regarding how often some mocks are called (hopefully without changing the logic being tested)

Tests

go test github.com/ChainSafe/gossamer/dot/sync

Issues

Closes #4198

[CLAassistant]

All committers have signed the CLA.

[haikoschol]

Created as a draft for two reasons:

1. I'd like to run a sync from scratch on Westend and/or Paseo as a regression test for a while.
2. To discuss and possibly address this TODO:

// SubmitBatch accepts a list of tasks and immediately returns a batch ID. The batch ID can be used to query the status
// of the batch using [GetBatchStatus].
// TODO
// If tasks are submitted faster than they are completed, resChan will run full, blocking the calling goroutine.
// Ideally this method would provide backpressure to the caller in that case. The rejected tasks should then stay in
// FullSyncStrategy.requestQueue until the next round. But this would need to be supported in all sync strategies.
func (w *workerPool) SubmitBatch(tasks []Task) (id BatchID, err error) {

[dimartiro]

Why times 10?

[haikoschol]

Just an arbitrary value that "should be enough" while I was testing the branch on the Westend genesis instance. Apart from wanting to do this testing, this TODO and the other one about back pressure are the reasons why this is still a draft.
Do you think it makes sense to use the actual value for number of tasks configured in the sync strategy? And should we add some extra space here or not?

[dimartiro]

Ok got it, we can move it to a constant to self explain the usage.

[haikoschol]

I'd like to try and make this depend on the configuration of the strategy. But if that's too invasive, I'll move it to a constant.

[haikoschol]

I've done this in b6cf4d7. It required extending the Strategy interface and delaying initialization of the worker pool.

[dimartiro]

What if we define result.Task over a generic? so we can skip this casting?

[haikoschol]

I considered making this generic and decided against it, but I can't remember why. Will give it a try.

[haikoschol]

Just tried and the reason I didn't do this is because SyncService is supposed to maintain the worker pool and work with any kind of Strategy. This won't work if the worker pool is a generic type instantiated with *syncTask.

With the current implementation, only the strategies themselves know the type of the tasks processed by the worker pool.

[dimartiro]

Good point, that makes me think that the strategies should be responsible for creating their own workerpool maybe

[haikoschol]

I'm still in favor of having the service manage one pool that is used for all strategies. I think it's easier to manage in terms of concurrency and good separation of concern. Is it worth giving that up to avoid a few casts? The strategies definitely know what concrete type they use as sync tasks.

[dimartiro]

100% agree that the service should be responsible for the working pool, but I'm still trying to find a way to improve the types. We can keep the casts in the meanwhile

[haikoschol]

2. To discuss and possibly address this TODO:

I've addressed the TODO in f4a9ccc. Not sure if this is the right approach, but I think it is better than always adding new tasks, risking a blocked goroutine.