@Marvin # Prefect Concurrency Lease Renewal Failure - Need Expert Advice ## Current Setup We have a Prefect workflow with the following architecture: ### Flow Hierarchy

Parent Flow (run_multiple_operations_flow)
  └─> Batch Flow (process_operations_batch_flow) - timeout_seconds=36000
       └─> 100x Subflow (process_single_operation_flow)
            └─> Task (execute_operation_task) ← Tag-based concurrency applied here

### Concurrency Configuration ***Task Definition:***

python
@task(
    _name_="Execute Operation Task",
    _retries_=0,
    _tags_=["my_operation_tag"],  _# Tag-based concurrency control_
)
_async_ _def_ execute_operation_task(_config_: _dict_, _item_id_: _str_) -> _dict_:
    result = await perform_long_running_operation(config, item_id)  _# Takes ~5 minutes_
    return result

***Prefect Concurrency Limit:***

bash
$ prefect concurrency-limit inspect my_operation_tag
Tag: my_operation_tag
Concurrency Limit: 1

***Batch Flow Code:***

python
@flow(
    _name_="Process Operations Batch",
    _log_prints_=True,
    _timeout_seconds_=36000,  _# 10 hours_
)
_async_ _def_ process_operations_batch_flow(
    _items_: list[_dict_],
    _run_name_: _str_ | None = None,
):
    _# Create 100 subflow coroutines_
    tasks = []
    for item in items:  _# 100 items_
        task = process_single_operation_flow(_item_=item, _run_name_=run_name)
        tasks.append(task)

    _# Wait for all 100 to complete_
    gathered_results = await asyncio.gather(*tasks, _return_exceptions_=True)

    return results

## The Error After running for ***~6 minutes***, we get:

Concurrency lease renewal failed - slots are no longer reserved.
Terminating execution to prevent over-allocation.

Crash detected! Execution was cancelled by the runtime environment.

Traceback:
  File "/usr/local/lib/python3.13/site-packages/prefect/concurrency/_leases.py", line 135, in amaintain_concurrency_lease
    yield
  ...
  asyncio.exceptions.CancelledError: Cancelled via cancel scope

The above exception was the direct cause of:
  prefect._internal.concurrency.cancellation.CancelledError

Finished in state Crashed('Execution was cancelled by the runtime environment.')

## Questions 1. ***Root Cause***: With tag-based concurrency on the task (not using a context manager), does the concurrency lease have a default TTL that's causing it to expire after ~6 minutes? 2. ***Tag-Based Concurrency Lease Duration***: When using

tags=["igxe_web_registration"]

on the task decorator, can we configure the lease duration? Or is there a hardcoded default? 3. ***Proper Solution***: What's the correct way to handle this scenario where: - We want to run 100 registrations sequentially (concurrency=1) - Each registration takes ~5 minutes - Total runtime = ~500 minutes (8+ hours) - We need the concurrency lease to stay valid throughout 4. ***Context Manager vs Tags***: Should we switch from tag-based concurrency to using

async with concurrency("igxe_web_registration", occupy=1, lease_duration=timedelta(hours=2))

inside the task? Does that solve the lease renewal issue? 5. ***Parent Flow Timeout***: We added

timeout_seconds=36000

to the batch flow. Does that help with lease renewal, or is that a separate concern? 6. ***Best Practice***: For long-running workflows with many sequential tasks (100 tasks, concurrency=1, 8+ hours total), what's the recommended Prefect pattern? Should we batch process (10 at a time) to give natural lease renewal points? ## What We Need Clear guidance on: - Whether tag-based concurrency has lease duration limits - How to properly configure lease duration for our use case - Whether we need to refactor from tags to context managers - Best practices for long-running sequential workflows in Prefect Thank you!

✅ thought for 430.8 seconds

@Marvin Our System Architecture We have Prefect workflows that orchestrate browser automation tasks. At the service layer, we use: - State machines that coordinate multi-step processes - Selenium/Appium for browser automation (inherently blocking - ~33 calls per workflow) - Django ORM for persistence (synchronous) - Only 2-3 true async I/O operations (HTTP requests, email polling) Our Prefect structure follows this pattern: # Parent Flow: Orchestrates batch processing @flow(name="Process Batch") async def process_batch_flow(items: list[dict]) -> list[Result]: # Submit 100 subflows concurrently subflows = [] for item in items: subflow = process_single_item_flow(item=item) subflows.append(subflow) # Wait for all subflows to complete results = await asyncio.gather(*subflows, return_exceptions=True) return results # Child Flow: Processes single item @flow(name="Process Single Item") async def process_single_item_flow(item: dict) -> Result: # Calls the actual work task result = await execute_work_task(item) return result # Task: Does the actual work (blocking operations) @task( name="Execute Work", tags=["work_concurrency"], # Tag-based concurrency limit = 1 ) async def execute_work_task(item: dict) -> Result: # Calls async state machine wrapper result = await _execute_state_machine(item) return result async def _execute_state_machine(item: dict) -> Result: # 33+ blocking Selenium calls inside here browser.navigate() # Blocking browser.fill_form() # Blocking browser.click() # Blocking # ... etc The Problem We're experiencing Prefect concurrency lease renewal failures: Concurrency lease renewal failed - slots are no longer reserved. asyncio.exceptions.CancelledError: Cancelled via cancel scope After investigation, we identified the root cause: Event loop blocking from Selenium operations prevents Prefect's lease renewal coroutine from running. Our Conundrum 1. 95% of our code is blocking (Selenium/Appium/Django ORM), so async provides minimal benefit 2. BUT: At the Prefect orchestration layer, asyncio.gather() allows us to submit 100 subflows concurrently without blocking 3. If we make everything sync, we're concerned that this pattern would block: @flow(name="Process Batch") def process_batch_flow(items: list[dict]) -> list[Result]: results = [] for item in items: result = process_single_item_flow(item=item) # ← Does this block? results.append(result) return results Our Questions 1. Does calling a sync subflow block the parent flow? If we convert everything to sync, will the loop above wait for each subflow to complete before submitting the next, thereby negating Prefect's tag-based concurrency control? 2. What's the minimal async approach? Should we: - Option A: Keep async flows for non-blocking submission, but make tasks synchronous? - Option B: Convert everything to sync and use .submit() on tasks/flows? - Option C: Use task mapping instead of subflows? 3. How does Prefect handle sync task submission? When we have: @flow() def batch_flow(items): futures = [] for item in items: future = work_task.submit(item=item) # ← Non-blocking? futures.append(future) results = [f.result() for f in futures] 3. Does .submit() on a sync task return immediately (like async), allowing concurrent execution via Prefect's scheduler? 4. What's the recommended pattern for our use case? Given that we have: - Blocking operations (Selenium/Django) - Low concurrency (1-5 simultaneous executions) - Need to orchestrate 100+ sequential tasks per batch - Tag-based concurrency limits already configured Should we refactor to full sync or keep async at the orchestration layer only? --- Context: We prefer a clean architecture over hybrid approaches. If async is only providing orchestration benefits but complicating the service layer, we'd rather refactor fully to sync - but only if it doesn't break concurrent submission behavior.

Your message was too long, here's your message at the allowed limit: Our System Architecture We have Prefect workflows that orchestrate browser automation tasks At the service layer we use - State machines that coordinate multi-step processes - Selenium/Appium for browser automation (inherently blocking - ~33 calls per workflow) - Django ORM for persistence (synchronous) - Only 2-3 true async I/O operations (HTTP requests email polling) Our Prefect structure follows this pattern # Parent Flow Orchestrates batch processing @flow(name= Process Batch ) async def process_batch_flow(items list[dict]) -> list[Result] # Submit 100 subflows concurrently subflows = [] for item in items subflow = process_single_item_flow(item=item) subflows append(subflow) # Wait for all subflows to complete results = await asyncio gather(*subflows return_exceptions=True) return results # Child Flow Processes single item @flow(name= Process Single Item ) async def process_single_item_flow(item dict) -> Result # Calls the actual work task result = await execute_work_task(item) return result # Task Does the actual work (blocking operations) @task( name= Execute Work tags=[ work_concurrency ] # Tag-based concurrency limit = 1 ) async def execute_work_task(item dict) -> Result # Calls async state machine wrapper result = await _execute_state_machine(item) return result async def _execute_state_machine(item dict) -> Result # 33+ blocking Selenium calls inside here browser navigate() # Blocking browser fill_form() # Blocking browser click() # Blocking # etc The Problem We're experiencing Prefect concurrency lease renewal failures Concurrency lease renewal failed - slots are no longer reserved asyncio exceptions CancelledError Cancelled via cancel scope After investigation we identified the root cause Event loop blocking from Selenium operations prevents Prefect's lease renewal coroutine from running Our Conundrum 1 95% of our code is blocking (Selenium/Appium/Django ORM) so async provides minimal benefit 2 BUT At the Prefect orchestration layer asyncio gather() allows us to submit 100 subflows concurrently without blocking 3 If we make everything sync we're concerned that this pattern would block @flow(name= Process Batch ) def process_batch_flow(items list[dict]) -> list[Result] results = [] for item in items result = process_single_item_flow(item=item) # ← Does this block? results append(result) return results Our Questions 1 Does calling a sync subflow block the parent flow? If we convert everything to sync will the loop above wait for each subflow to complete before submitting the next thereby negating Prefect's tag-based concurrency control? 2 What's the minimal async approach? Should we - Option A Keep async flows for non-blocking submission but make tasks synchronous? - Option B Convert everything to sync and use submit() on tasks/flows? - Option C Use task mapping instead of subflows? 3 How does Prefect handle sync task submission? When we have @flow() def batch_flow(items) futures = [] for item in items future = work_task submit(item=item) # ← Non-blocking? futures append(future) results = [f result() for f in futures] 3 Does submit() on a sync task return immediately (like async) allowing concurrent execution via Prefect's scheduler? 4 What's the recommended pattern for our use case? Given that we have - Blocking operations (Selenium/Django) - Low concurrency (1-5 simultaneous executions) - Need to orchestrate 100+ sequential tasks per batch - Tag-based concurrency limits already configured Should we

@Marvin Our System Architecture We have Prefect workflows orchestrating browser automation tasks. At the service layer: - State machines coordinating multi-step processes - Selenium/Appium for browser automation (blocking - ~33 calls per workflow) - Django ORM (synchronous) - Only 2-3 true async I/O operations (HTTP requests, email polling) Current structure: @flow(name="Process Batch") async def process_batch_flow(items: list[dict]) -> list[Result]: subflows = [process_single_item_flow(item=item) for item in items] results = await asyncio.gather(*subflows, return_exceptions=True) return results @flow(name="Process Single Item") async def process_single_item_flow(item: dict) -> Result: result = await execute_work_task(item) return result @task(tags=["work_concurrency"]) # Tag-based concurrency limit = 1 async def execute_work_task(item: dict) -> Result: result = await _execute_state_machine(item) return result async def _execute_state_machine(item: dict) -> Result: browser.navigate() # Blocking browser.fill_form() # Blocking # ... 33+ blocking calls The Problem Experiencing Prefect concurrency lease renewal failures: Concurrency lease renewal failed - slots are no longer reserved. asyncio.exceptions.CancelledError: Cancelled via cancel scope Root cause: Event loop blocking from Selenium prevents Prefect's lease renewal coroutine from running. Our Conundrum 1. 95% of our code is blocking, so async provides minimal benefit 2. BUT: asyncio.gather() lets us submit 100 subflows concurrently without blocking 3. If we make everything sync, will this pattern block? @flow() def process_batch_flow(items): results = [] for item in items: result = process_single_item_flow(item) # Does this block? results.append(result) return results Our Questions 1. Does calling a sync subflow block the parent flow? Will the loop wait for each subflow to complete before submitting the next, negating tag-based concurrency control? 2. What's the minimal async approach? Should we keep async flows for non-blocking submission but make tasks synchronous, or convert everything to sync and use .submit()? 3. How does Prefect handle sync task submission? @flow() def batch_flow(items): futures = [work_task.submit(item=i) for i in items] # Non-blocking? results = [f.result() for f in futures] 3. Does .submit() on sync tasks return immediately, allowing concurrent execution via Prefect's scheduler? 4. Recommended pattern for our use case? Given blocking operations, low concurrency (1-5), 100+ tasks per batch, and tag-based concurrency limits already configured, should we

✅ thought for 338.4 seconds

@Marvin I have a different question but dont feel like opening a new thread so i'll just ask here. I'm refactoring a large Prefect workflow system and need guidance on the proper architectural pattern for separating business logic from Prefect orchestration. Current Architecture: We have a service layer with pure Python business logic, and Prefect tasks that are thin wrappers around these services: # Service Layer (Pure Python - No Prefect) class RegistrationService: def execute_registration(self, user_data, config): context = self._create_context(user_data) clients = self._create_clients(config) try: result = self._run_workflow(context, clients) # 200+ lines self._create_audit_record(result) # DB write return result finally: self._cleanup_clients(clients) class PersistenceService: def update_user_status(self, user_id, success): user = User.objects.get(id=user_id) user.completed = success user.save() # Prefect Task Layer (Thin wrappers) @task(retries=0) def execute_registration_task(user_data, config): service = RegistrationService() return service.execute_registration(user_data, config) @task(retries=1) # <-- DB write, so retry=1 def update_user_status_task(user_id, success): service = PersistenceService() service.update_user_status(user_id, success) # Flow @flow def process_registration_flow(user_data): success = execute_registration_task(user_data, config) update_user_status_task(user_data.id, success) # Separate task? return success The Problem: We're wrapping almost every service method in a @task decorator. This pattern is repeated throughout the codebase, but I'm unsure when it's necessary vs unnecessary overhead. Key Questions: 1. When should service methods be wrapped in @task vs called directly? - Database writes like update_user_status() - should these be tasks for retry logic? - Or should they just be called directly from the parent task? 2. Persistence operations specifically: - Notice RegistrationService calls _create_audit_record() directly (DB write inside service) - But update_user_status() is wrapped as a separate task - This is inconsistent - should ALL persistence be tasks, or ALL be in service methods? 3. What are we trading off? - Task wrappers give: retries, observability in UI, granular error attribution - But add: indirection, more files to navigate, unclear transaction boundaries 4. Performance/overhead: - Does wrapping every operation in a task add significant overhead? - How granular should tasks be? Example scenarios: Simple DB update - task or direct call? # Option A: Task wrapper (current) update_status_task(id, status) # Option B: Direct in parent task @task def parent_task(): result = do_work() User.objects.filter(id=id).update(status=status) External API call - task or direct call? # Option A: Task wrapper @task(retries=3) def call_api_task(data): return requests.post(url, json=data) # Option B: In service class Service: def process(self): response = requests.post(url, json=data) What's the Prefect-recommended pattern for service layer separation, and when should operations be wrapped in @task vs called directly? I want observability and retries, but want to avoid over-engineering.

✅ thought for 53.0 seconds