feat(spooler): Add ability to partition the buffer into multiple ones

[iambriccardo]

This PR allows the buffer to be partitioned across several partitions. For simplicity, partitioning is achieved at the highest level, meaning that Relay will run x independent partitions of the same EnvelopeBufferService each with its own database file (if the database config is used).

[jjbayer]

Nice!

[jjbayer]

nit: I would still call this field envelope_buffer, the use does not really care whether it partitions internally or not.

[jjbayer]

I would change the interface of PartitionedEnvelopeBuffer to behave roughly like ObservableEnvelopeBuffer behaved before. I.e. give it a send(..) and a has_capacity() method, but not expose internals about partitioning.

[iambriccardo]

I thought about this when designing the API, but it conflicted with how we access the buffer in the queue_envelope. I wanted to keep that logic like it is right now.

[loewenheim]

You can do this with collect, though it's probably a matter of taste which is better.

Suggested change

	let mut envelope_buffers = Vec::with_capacity(partitions.get() as usize);
	for partition_id in 0..partitions.get() {
	let envelope_buffer = EnvelopeBufferService::new(
	partition_id,
	config.clone(),
	memory_stat.clone(),
	global_config_rx.clone(),
	Services {
	envelopes_tx: envelopes_tx.clone(),
	project_cache_handle: project_cache_handle.clone(),
	outcome_aggregator: outcome_aggregator.clone(),
	test_store: test_store.clone(),
	},
	)
	.map(|b| b.start_in(runner));
	if let Some(envelope_buffer) = envelope_buffer {
	envelope_buffers.push(envelope_buffer);
	}
	}
	Self {
	buffers: Arc::new(envelope_buffers),
	}
	}
	let envelope_buffers = (0..partitions.get())
	.filter_map(|partition_id| {
	EnvelopeBufferService::new(
	partition_id,
	config.clone(),
	memory_stat.clone(),
	global_config_rx.clone(),
	Services {
	envelopes_tx: envelopes_tx.clone(),
	project_cache_handle: project_cache_handle.clone(),
	outcome_aggregator: outcome_aggregator.clone(),
	test_store: test_store.clone(),
	},
	)
	})
	.map(|b| b.start_in(runner))
	.collect();