[OTel]: Bound provider flush waiting without blocking unrelated providers #856

Split from PR #758 — preserve replay completion telemetry, using reference head 0d407e1. The reference contains multiple concerns and outstanding review feedback; use its relevant implementation and tests as a starting point, not as a patch to merge wholesale.

Baseline inspected: main at 2e5135fc6f6c05024458e0be333a8325612abd6c. This issue is based on source and PR-review inspection; no runtime reproduction was executed when filing it. The scenarios below are the implementation and validation work to perform.

Problem and reproduction

Both plugins call the configured provider's force_flush() synchronously at invocation End. A custom provider/processor can ignore a cooperative timeout and hold the handler indefinitely. A single process-wide active-flush slot would also let blocked provider A prevent independent provider B from flushing.

Reproduce with provider A whose force_flush() waits on a test-controlled event and provider B that completes immediately. Finish separate public handler invocations using A and B; B should flush promptly, A's caller should stop waiting within its budget, and subsequent calls using A must not create an accumulating queue of workers.

Design to settle before implementation

Write a short design note describing the blocking path, compare provider-managed/transport timeouts with an SDK caller-wait bound, and explain why the selected mechanism is necessary. PR #758 proposes a daemon worker with a 30-second caller budget matching the provider default; the issue requires the bounded behavior, not unreviewed adoption of that implementation. See the design review.

Required behavior and acceptance criteria

• Bound the caller's total wait (admission plus execution) with one absolute deadline; the reference target is 30 seconds. Tests should inject a short budget and use events/barriers rather than sleep for 30 seconds.
• At most one outstanding flush per provider identity. Provider B progresses while A is blocked, including custom providers with unusual equality/hash behavior.
• A same-provider caller admitted after an earlier attempt starts a fresh flush, so spans queued during that earlier attempt are included. Waiting for someone else's attempt is not proof that this caller's spans were flushed.
• A request that times out before admission creates no queued work. A running attempt keeps its provider slot until it actually finishes, even after its caller stops waiting.
• Invocation Start/End hooks and token cleanup remain on their existing owner thread/context. If a worker is necessary, only provider I/O runs there using the post-cleanup context snapshot.
• Handle successful return, False, fast ordinary errors, fast control exceptions, late errors after timeout, worker-start failure, and a subsequent healthy invocation. Capacity is released on actual completion/start failure, and late errors are logged.
• Preserve ordinary plugin error containment; control failures that finish while the caller is waiting propagate on the lifecycle owner. Do not throw late failures into an already-returned invocation.
• Both OTel views satisfy the behavior through the public handler, with context and logger correlation restored before flush I/O.
• Document the budget, per-provider admission, timeout behavior, and limits in the OTel README.

Code and reference tests

Primary implementation candidate: _flush.py, wired only through _force_flush() in both OTel plugin files.

• test_flush.py
• test_flush_lifecycle_int.py

Maintainer requirement: isolate active attempts by provider.

Independence and exclusions

Implement and merge directly on main using the existing _force_flush() calls. No core change, pending-completion queue, status remapping, or export-retry/finalization framework is needed. Coordinate only this call site with the separate finalization issue.

The bound applies to caller waiting for force_flush, not synchronous processor on_end, provider shutdown, or transport I/O cancellation. The application still owns provider shutdown and transport timeouts. A timeout/False result is not an export acknowledgment. Avoid a general scheduler, unbounded queue, or new user-facing configuration surface unless the design establishes a concrete need.

Validation

From the repository root:

hatch run dev-otel:test
hatch run dev-otel:typecheck

Run hatch fmt --check from packages/aws-durable-execution-sdk-python-otel. Add focused unit tests and public-handler integration coverage for the behavior described above. For OTel changes, also build the core and OTel wheels with hatch build in their package directories and run hatch run test-wheel-otel:test from the root as documented in CONTRIBUTING.md. Use existing test environments; this split does not require testing-library or CI-filter changes. Document any changed existing test expectations and their rationale in the implementing PR.

Related issues in this split

• #852 — Core terminal error snapshot isolation
• #853 — Terminal span status and exception timestamps
• #854 — Replay completion discovery, parentage, and segment identity
• #855 — Export retries and invocation finalization failure isolation