feat!(extstore): add new concurrency controls for external storage.

[cconstable]

What was changed

ExternalStorage now has a concurrency field with two limits:

• max_driver_operations caps requests in flight for a single client or worker. Each client and each worker gets its own limit.
• max_operations_per_message caps requests for a single message (workflow activation, etc). Each message gets its own budget.

Both limits are cooperative. Drivers receive a limiter on the store and retrieve contexts and must wrap each request in permit. Requests made outside a permit are not counted, and a warning is logged when a driver completes without taking one.

async with context.limiter.permit(payload):
    # store/retrieve logic

Removed the max_workflow_task_external_storage_concurrency Worker argument. The payload visitor now walks a message's payload sites max_operations_per_message at a time. The bundled S3 driver takes permits.

Why?

The previous control bounded payload visits rather than driver calls, so setting it to 3 did not limit external storage requests to 3. It also only applied to workflow tasks, leaving client, activity, and Nexus paths unbounded.

max_driver_operations is scoped to a client or worker so that ExternalStorage stays configuration. Holding the counter there meant two clients built from one config silently shared a limit.

Checklist

Added tests

[jmaeagle99]

Discussed elsewhere. I think visiting should not be unbounded and should pass in ExternalStorageConcurrency.max_operations_per_message

[jmaeagle99]

I think this cache is per ExternalStorage instance. And each ExternalStorage instance gets a fresh new copy every time a new DataConverter gets created. So it's likely that while this works for a given instance of ExternalStorage / DataConverter, it is probably not worker-wide or client-wide; each time with_context is called, a new DataConverter is created.

[jmaeagle99]

This is tied to the event loop that the DataConverter is on and forces DataConverter initialization to be on an event loop; unsure the latter was the case already.

[jmaeagle99]

I was not expecting a semaphore to be shared out from DataConverter. I think it might be easier to have a client constructor one itself directly from the DataConverter.external_storage.concurrency values and hold onto it. It might need to be created per asyncio event-loop if clients can be reused across event loops.