feat(pool): Use new AsyncPool for the processor and store services

[iambriccardo]

This PR integrates the new AsyncPool in the EnvelopeProcessorService and StoreService.

Please note that currently the integration doesn't leverage the benefits of async, since all the functions called by the pool are blocking. The support for async in the I/O operations of the EnvelopeProcessorService will be coming.

Closes: https://github.com/getsentry/team-ingest/issues/645

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
4	1	3	0

View the top 1 failed test(s) by shortest run time

_integration-test.test_01_basics::test_receive_event

Stack Traces | 65.4s run time

client_login = (<httpx.Client object at 0x7faa689a6930>, <Response [200 OK]>)

    #x1B[0m#x1B[94mdef#x1B[39;49;00m #x1B[92mtest_receive_event#x1B[39;49;00m(client_login):#x1B[90m#x1B[39;49;00m
        event_id = #x1B[94mNone#x1B[39;49;00m#x1B[90m#x1B[39;49;00m
        client, _ = client_login#x1B[90m#x1B[39;49;00m
        #x1B[94mwith#x1B[39;49;00m sentry_sdk.init(dsn=get_sentry_dsn(client)):#x1B[90m#x1B[39;49;00m
            event_id = sentry_sdk.capture_exception(#x1B[96mException#x1B[39;49;00m(#x1B[33m"#x1B[39;49;00m#x1B[33ma failure#x1B[39;49;00m#x1B[33m"#x1B[39;49;00m))#x1B[90m#x1B[39;49;00m
        #x1B[94massert#x1B[39;49;00m event_id #x1B[95mis#x1B[39;49;00m #x1B[95mnot#x1B[39;49;00m #x1B[94mNone#x1B[39;49;00m#x1B[90m#x1B[39;49;00m
>       response = poll_for_response(#x1B[90m#x1B[39;49;00m
            #x1B[33mf#x1B[39;49;00m#x1B[33m"#x1B[39;49;00m#x1B[33m{#x1B[39;49;00mSENTRY_TEST_HOST#x1B[33m}#x1B[39;49;00m#x1B[.../sentry/internal/events/#x1B[39;49;00m#x1B[33m{#x1B[39;49;00mevent_id#x1B[33m}#x1B[39;49;00m#x1B[33m/#x1B[39;49;00m#x1B[33m"#x1B[39;49;00m, client#x1B[90m#x1B[39;49;00m
        )#x1B[90m#x1B[39;49;00m

#x1B[1m#x1B[31m_integration-test/test_01_basics.py#x1B[0m:129: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

request = 'http://localhost:.../internal/events/bfa1bc914f3442bb86b695395287d56a/'
client = <httpx.Client object at 0x7faa689a6930>, validator = None

    #x1B[0m#x1B[94mdef#x1B[39;49;00m #x1B[92mpoll_for_response#x1B[39;49;00m(#x1B[90m#x1B[39;49;00m
        request: #x1B[96mstr#x1B[39;49;00m, client: httpx.Client, validator: Callable = #x1B[94mNone#x1B[39;49;00m#x1B[90m#x1B[39;49;00m
    ) -> httpx.Response:#x1B[90m#x1B[39;49;00m
        #x1B[94mfor#x1B[39;49;00m i #x1B[95min#x1B[39;49;00m #x1B[96mrange#x1B[39;49;00m(TIMEOUT_SECONDS):#x1B[90m#x1B[39;49;00m
            response = client.get(#x1B[90m#x1B[39;49;00m
                request, follow_redirects=#x1B[94mTrue#x1B[39;49;00m, headers={#x1B[33m"#x1B[39;49;00m#x1B[33mReferer#x1B[39;49;00m#x1B[33m"#x1B[39;49;00m: SENTRY_TEST_HOST}#x1B[90m#x1B[39;49;00m
            )#x1B[90m#x1B[39;49;00m
            #x1B[94mif#x1B[39;49;00m response.status_code == #x1B[94m200#x1B[39;49;00m:#x1B[90m#x1B[39;49;00m
                #x1B[94mif#x1B[39;49;00m validator #x1B[95mis#x1B[39;49;00m #x1B[94mNone#x1B[39;49;00m #x1B[95mor#x1B[39;49;00m validator(response.text):#x1B[90m#x1B[39;49;00m
                    #x1B[94mbreak#x1B[39;49;00m#x1B[90m#x1B[39;49;00m
            time.sleep(#x1B[94m1#x1B[39;49;00m)#x1B[90m#x1B[39;49;00m
        #x1B[94melse#x1B[39;49;00m:#x1B[90m#x1B[39;49;00m
>           #x1B[94mraise#x1B[39;49;00m #x1B[96mAssertionError#x1B[39;49;00m(#x1B[90m#x1B[39;49;00m
                #x1B[33m"#x1B[39;49;00m#x1B[33mtimeout waiting for response status code 200 or valid data#x1B[39;49;00m#x1B[33m"#x1B[39;49;00m#x1B[90m#x1B[39;49;00m
            )#x1B[90m#x1B[39;49;00m
#x1B[1m#x1B[31mE           AssertionError: timeout waiting for response status code 200 or valid data#x1B[0m

#x1B[1m#x1B[31m_integration-test/test_01_basics.py#x1B[0m:40: AssertionError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[Dav1dde]

this is probably too high cardinality, maybe we just emit the total size somewhere or the max?

Also we don't have to do it on each poll only when the futures length changes (e.g. on push)