hey everyone, started getting this weird issue with my Cloud Run Jobs based work pool recently. I have a flow that submits 7 flows (they are actually standalone flows with their own infra, not actually subflows). Each flow trigger its own cloud run job with a unique (prefect generated) name etc. Still I'm getting this GCP error claiming resource already exists. It seems to happen quite randomly. As you can see some of them launch successfully and sometimes all of them do. However as you can imagine it's a big problem when some of them don't. Anyone has an idea of what it can be? When I look at Cloud run I never see the job existing at any time that would justify a duplication. However it's also probably because we have

keep_jobs

set to false so even if it's duplicating some job it's probably also deleting. We're not tracking the flow state or anything like that from python, we just "fire and forget" these flows. This is my code to launch them:

for model in collection_models:
        scenarios = model_to_scenarios.get(model, [])
        if not scenarios:
            logger.warning("No scenarios found for model %s, skipping", model)
            continue

        # Submit task for this model's subflow with model-specific name
        future = launch_model_subflow_task.with_options(
            name=f"launch-{model}",
            retries=2,
            retry_delay_seconds=60,
        ).submit(
            model=model,
            scenarios=scenarios,
            collection_limit_per_brand_model=request.collection_limit_per_brand_model,
        )
        deployment_futures.append(future)

and the definition:

flow_run = await run_deployment(  # type: ignore[misc]
            name=f"process-batch-subflow/{PREFECT_DEPLOYMENT_SUFFIX}-subflow",
            parameters={
                "model": model,
                "collection_scenarios": scenarios,
                "collection_limit_per_brand_model": collection_limit_per_brand_model,
            },
            job_variables={
                "cpu": CPU_CORES,
                "memory": "32G",
            },
            timeout=0,  # Fire-and-forget: don't wait for subflow completion (execution timeout is set on the subflow itself)
            tags=[model],
        )

Not exactly sure how it would work with cloud runs, but at least on local flow creation the name is only applied after the flow is made and begun. Maybe cool-names is making the same name as the flows are getting created? Not sure how much time is between the sub flow submits.

We have been getting the same issue and it is crippling us