Fix Kubernetes worker secret upsert race on concurrent job creation (#16447)

[wtfashwin]

Closes #16447

Problem

KubernetesWorker._upsert_secret reads the API key secret and, on a 404, creates it. When a newly started worker picks up several scheduled flow runs at once, it issues consecutive create_namespaced_job calls that each run _upsert_secret concurrently. They all observe the 404, all attempt to create the secret, and every call after the first fails with a 409 AlreadyExists. That ApiException propagated unhandled and crashed the flow run.

This matches the root-cause analysis in #16447 (a classic check-then-act / TOCTOU race between the read and the create).

Fix

Treat the 409 for what it is — a benign lost create race. On conflict, re-read the secret to obtain the current resourceVersion and replace it with our value, mirroring the existing update path. The secret value is deterministic for a given worker, so the winning value is unchanged either way; the worker no longer crashes.

try:
    secret = await core_client.create_namespaced_secret(namespace=namespace, body=secret)
except ApiException as create_exc:
    if create_exc.status != 409:
        raise
    # A concurrent job submission created the secret between our read and
    # create. Re-read to obtain the current resourceVersion and replace it.
    current_secret = await core_client.read_namespaced_secret(name=name, namespace=namespace)
    current_secret.data = {"value": encoded_value}
    secret = await core_client.replace_namespaced_secret(name=name, namespace=namespace, body=current_secret)

Tests

• Added test_upsert_secret_recovers_from_concurrent_create, which drives the read-404 -> create-409 -> re-read -> replace recovery and asserts the worker's value wins. It fails before the change (the 409 propagates) and passes after.
• Existing secret tests (test_can_store_api_key_in_secret, test_store_api_key_in_existing_secret, and the rest of the secret/api_key suite) still pass.
• ruff check and ruff format --check pass on both changed files.

[desertaxle]

Let's not replace the secret when there's a race between workers. I think we're ok not raising on a 409 in all cases, but if you want to compare values and raise if they don't match for added safety, then I'm cool with that.

[wtfashwin]

Good call — updated to no longer replace the secret on a 409. We now re-read the secret created by the winning worker and leave it in place. For the added safety you mentioned, we compare the existing value against ours and raise only if they unexpectedly differ. Updated the regression test and added one covering the mismatch case.

[wtfashwin]

@desertaxle thanks for the review! I've pushed a change addressing your feedback: we no longer replace the secret on a 409. The losing worker now re-reads the secret created by the winner and leaves it in place, raising only if the concurrently created value unexpectedly differs. Tests updated accordingly. Could you take another look when you get a chance? 🙏

[codspeed-hq]

Merging this PR will not alter performance

✅ 8 untouched benchmarks
⏩ 1 skipped benchmark¹

---

_{Comparing wtfashwin:fix/k8s-upsert-secret-concurrent-create-16447 (c125911) with main (f7d76f2)}

Footnotes

1. 1 benchmark was skipped, so the baseline result was used instead. If it was deleted from the codebase, click here and archive it to remove it from the performance reports. ↩

[wtfashwin]

@desertaxle resolved

[desertaxle]

Please revert these changes to keep the PR scoped only to the changes necessary to solve the linked issue.

[desertaxle]

Please revert these changes to keep the PR scoped only to the changes necessary to solve the linked issue. This change has already been made on main.

[desertaxle]

Please revert these changes to keep the PR scoped only to the changes necessary to solve the linked issue.