External Storage Integration: NexusWorker

[cconstable]

What was changed

• NexusWorker now stores and retrieves payloads.

Why?

• Nexus workers should use external storage.

Checklist

[cconstable]

Missing: NexusClient integration. Adding now.

[Quinn-With-Two-Ns]

Why does this storeOutbound work differently then this one ? Should we just make one common helper ?

[Quinn-With-Two-Ns]

Is there any tests showing external storage working for a Nexus operation? or what PR would that be in?

[Quinn-With-Two-Ns]

Codex flagged these issues as well

I found two actionable issues. Your question about integration tests was
  warranted.

  1. High — Storage failures never produce a retryable Nexus task failure

     ExternalStorageTaskFailure says it is reported as a retryable handler error

     (https://github.com/temporalio/sdk-java/blob/94eaaca69c34a780b00803f1ddac129112ad2107/temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java#L287-L295),
     but the class is unused. Both inbound retrieval and outbound storage errors

     (https://github.com/temporalio/sdk-java/blob/94eaaca69c34a780b00803f1ddac129112ad2107/temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java#L553-L574)
     escape the worker. The generic poll executor only forwards them to the
     uncaught-exception handler; it does not call RespondNexusTaskFailed.

     Recovery therefore waits for task timeout/redelivery. For an outbound
     failure, the user handler has already executed, so redelivery can repeat
     its side effects. Catch storage errors and immediately send a small,
     inline, retryable INTERNAL handler failure—without passing that failure
     through external storage again. Add transient retrieve/store failure
     integration tests.

  2. Medium — External retrieval is excluded from the used-slot lifecycle

     markSlotUsed now happens after retrieveInboundPayloads

     (https://github.com/temporalio/sdk-java/blob/94eaaca69c34a780b00803f1ddac129112ad2107/temporal-sdk/src/main/java/io/temporal/internal/worker/NexusWorker.java#L325-L350).
     During a slow retrieval, used-slot metrics and custom supplier state
     under-report active work. If retrieval fails, the completion callback
     releases the permit as taskComplete, but TrackingSlotSupplier has no slot
     info because it was never marked used

     (https://github.com/temporalio/sdk-java/blob/94eaaca69c34a780b00803f1ddac129112ad2107/temporal-sdk/src/main/java/io/temporal/internal/worker/TrackingSlotSupplier.java#L70-L91).

     Mark the slot used before retrieval, or release failed pre-handler
     retrievals through an explicit never-used/error path.