diff --git a/aggregatedpool/handler.go b/aggregatedpool/handler.go index 3afce609..86ccb071 100644 --- a/aggregatedpool/handler.go +++ b/aggregatedpool/handler.go @@ -205,11 +205,14 @@ func (wp *Workflow) handleMessage(msg *internal.Message) error { timerID := wp.env.NewTimer(command.ToDuration(), workflow.TimerOptions{ Summary: command.Summary, }, wp.createCallback(msg.ID, "NewTimer")) + // A non-positive duration is resolved by the SDK inside NewTimer: the callback + // already ran, no timer exists, nothing to cancel. + if timerID == nil { + break + } wp.canceller.Register(msg.ID, func() error { - if timerID != nil { - wp.log.Debug("cancel timer request", zap.String("timerID", timerID.String())) - wp.env.RequestCancelTimer(*timerID) - } + wp.log.Debug("cancel timer request", zap.String("timerID", timerID.String())) + wp.env.RequestCancelTimer(*timerID) return nil }) @@ -530,6 +533,44 @@ func (wp *Workflow) handleMessage(msg *internal.Message) error { return errors.E(op, err) } + case *internal.ExecuteNexusOperation: + wp.log.Debug("nexus operation request", + zap.Uint64("ID", msg.ID), + zap.String("endpoint", command.Endpoint), + zap.String("service", command.Service), + zap.String("operation", command.Operation), + ) + + params := command.NexusOperationParams(msg.Payloads, msg.Header) + + nexusSeq := wp.env.ExecuteNexusOperation( + params, + wp.makeNexusCompletionResponseCallback(msg.ID), + wp.makeNexusStartedRegistryCallback(msg.ID), + ) + + wp.canceller.Register(msg.ID, func() error { + wp.log.Debug("cancel nexus operation request", zap.Int64("seq", nexusSeq)) + wp.env.RequestCancelNexusOperation(nexusSeq) + return nil + }) + + case *internal.GetNexusOperationStarted: + wp.log.Debug("get nexus operation started", zap.Uint64("ID", msg.ID), zap.Uint64("startID", command.ID)) + + // Drop the slot on consume, not on completion: a fast op can complete + // before PHP asks, and discarding early would hang this Listen forever. + wp.nexusStarted.Listen(command.ID, func(token string, err error) { + defer wp.nexusStarted.Discard(command.ID) + // May fire inline when the token was already pushed. + wp.pendingFlush = true + if err != nil { + wp.mq.PushError(msg.ID, temporal.GetDefaultFailureConverter().ErrorToFailure(err), wp.getWorkflowWorkerPid()) + return + } + wp.pushStartEnvelope(msg.ID, NexusStartEnvelope{Async: token != "", Token: token}) + }) + default: return errors.E(op, errors.Str("undefined command")) } @@ -554,7 +595,7 @@ func (wp *Workflow) createLocalActivityCallback(id uint64) bindings.LocalActivit return func(lar *bindings.LocalActivityResultWrapper) { // timer cancel callback can happen inside the loop - if atomic.LoadUint32(&wp.inLoop) == 1 { + if wp.deliverInline() { wp.log.Debug("calling local activity callback IN LOOP", zap.Uint64("ID", id)) callback(lar) return @@ -586,7 +627,7 @@ func (wp *Workflow) createCallback(id uint64, t string) bindings.ResultHandler { return func(result *commonpb.Payloads, err error) { // timer cancel callback can happen inside the loop - if atomic.LoadUint32(&wp.inLoop) == 1 { + if wp.deliverInline() { wp.log.Debug("calling callback IN LOOP", zap.Uint64("ID", id), zap.String("type", t)) callback(result, err) return diff --git a/aggregatedpool/nexus.go b/aggregatedpool/nexus.go new file mode 100644 index 00000000..980e2cf4 --- /dev/null +++ b/aggregatedpool/nexus.go @@ -0,0 +1,474 @@ +package aggregatedpool + +import ( + "context" + "fmt" + "maps" + "net/url" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/roadrunner-server/goridge/v3/pkg/frame" + "github.com/roadrunner-server/pool/payload" + "github.com/temporalio/roadrunner-temporal/v5/api" + "github.com/temporalio/roadrunner-temporal/v5/internal" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + failurepb "go.temporal.io/api/failure/v1" + "go.temporal.io/sdk/converter" + "go.temporal.io/sdk/temporal" + "go.uber.org/zap" + + "github.com/nexus-rpc/sdk-go/nexus" +) + +// Wire contract with PHP — must match FailureConverter::NEXUS_OPERATION_ERROR_TYPE_PREFIX. +const nexusOperationErrorTypePrefix = "nexus.OperationError." + +// Type Go and Java handlers put on a failed operation error. +const nexusOperationErrorType = "OperationError" + +// Timeout for the fire-and-forget CancelNexusOperationMethod RPC. +const nexusCancelMethodTimeout = 5 * time.Second + +// NexusHandler forwards handler-side Nexus Start/Cancel to PHP via the activity pool. +type NexusHandler struct { + codec api.Codec + pool api.Pool + log *zap.Logger + namespace string + // seqID is the wire envelope ID; invocationSeq is the InvocationID seen by + // PHP and used by CancelNexusOperationMethod. Kept separate so wire format + // can evolve without touching cooperative-cancel semantics. + seqID uint64 + invocationSeq uint64 + pldPool *sync.Pool + // inFlight gates CancelNexusOperationMethod emission so we don't race a + // cancel past Start completion. + inFlight sync.Map +} + +func NewNexusHandler(codec api.Codec, pool api.Pool, log *zap.Logger, namespace string) *NexusHandler { + return &NexusHandler{ + codec: codec, + pool: pool, + log: log, + namespace: namespace, + pldPool: &sync.Pool{ + New: func() any { + return new(payload.Payload) + }, + }, + } +} + +type nexusOperation struct { + nexus.UnimplementedOperation[converter.RawValue, converter.RawValue] + name string + serviceName string + taskQueue string + handler *NexusHandler +} + +func (op *nexusOperation) Name() string { + return op.name +} + +func (op *nexusOperation) Start(ctx context.Context, input converter.RawValue, options nexus.StartOperationOptions) (nexus.HandlerStartOperationResult[converter.RawValue], error) { + return op.handler.startOperation(ctx, op.taskQueue, op.serviceName, op.name, input.Payload(), options) +} + +func (op *nexusOperation) Cancel(ctx context.Context, token string, options nexus.CancelOperationOptions) error { + return op.handler.cancelOperation(ctx, op.taskQueue, op.serviceName, op.name, token, options) +} + +// CreateNexusService builds a nexus.Service with pass-through operations. +func (h *NexusHandler) CreateNexusService(taskQueue string, serviceName string, operationNames []string) *nexus.Service { + svc := nexus.NewService(serviceName) + ops := make([]nexus.RegisterableOperation, 0, len(operationNames)) + for _, name := range operationNames { + ops = append(ops, &nexusOperation{ + name: name, + serviceName: serviceName, + taskQueue: taskQueue, + handler: h, + }) + } + svc.MustRegister(ops...) + return svc +} + +func (h *NexusHandler) startOperation( + ctx context.Context, + taskQueue string, + serviceName string, + operationName string, + input *commonpb.Payload, + options nexus.StartOperationOptions, +) (nexus.HandlerStartOperationResult[converter.RawValue], error) { + h.log.Debug("nexus start operation", zap.String("service", serviceName), zap.String("operation", operationName), zap.String(tq, taskQueue)) + + links := nexusLinksToInternal(options.Links) + + invocationID := atomic.AddUint64(&h.invocationSeq, 1) + msg := &internal.Message{ + ID: atomic.AddUint64(&h.seqID, 1), + Command: internal.InvokeNexusOperation{ + Service: serviceName, + Operation: operationName, + Namespace: h.namespace, + TaskQueue: taskQueue, + RequestID: options.RequestID, + Callback: options.CallbackURL, + CallbackHeaders: maps.Clone(options.CallbackHeader), + Headers: maps.Clone(options.Header), + Links: links, + InvocationID: invocationID, + }, + } + + if input != nil { + msg.Payloads = &commonpb.Payloads{Payloads: []*commonpb.Payload{input}} + } + + // Watch ctx cancellation during Start; emit method cancel to PHP. The inFlight-before-done + // ordering only shrinks the stale-observe window; a stale cancel is harmless (best-effort). + h.inFlight.Store(invocationID, struct{}{}) + done := make(chan struct{}) + defer func() { + h.inFlight.Delete(invocationID) + close(done) + }() + go h.watchForMethodCancel(ctx, invocationID, done) + + r, err := h.roundTrip(ctx, taskQueue, msg, "nexus request") + if err != nil { + return nil, err + } + + out := make([]*internal.Message, 0, 1) + if err := h.codec.Decode(r, &out); err != nil { + return nil, newNexusHandlerError(nexus.HandlerErrorTypeInternal, nexus.HandlerErrorRetryBehaviorNonRetryable, "decode nexus response", err) + } + + if len(out) != 1 { + return nil, newNexusHandlerError(nexus.HandlerErrorTypeInternal, nexus.HandlerErrorRetryBehaviorNonRetryable, "invalid nexus worker response", nil) + } + + return h.decodeStartReply(ctx, out[0]) +} + +// roundTrip encodes msg, executes it on the worker pool, and returns the raw +// reply payload. Failures come back as *nexus.HandlerError; what distinguishes +// the request kind in messages ("nexus request" / "nexus cancel request"). +func (h *NexusHandler) roundTrip(ctx context.Context, taskQueue string, msg *internal.Message, what string) (*payload.Payload, error) { + pl := h.getPld() + defer h.putPld(pl) + + if err := h.codec.Encode(&internal.Context{TaskQueue: taskQueue}, pl, msg); err != nil { + // Encoding our own request is a deterministic local bug, not a transient + // fault — don't ask the server to retry it. + return nil, newNexusHandlerError(nexus.HandlerErrorTypeInternal, nexus.HandlerErrorRetryBehaviorNonRetryable, "encode "+what, err) + } + + ch := make(chan struct{}, 1) + result, err := h.pool.Exec(ctx, pl, ch) + if err != nil { + // Pool returned before queueing — typically pool busy / exec rejected; retryable. + return nil, newNexusHandlerError(nexus.HandlerErrorTypeUnavailable, nexus.HandlerErrorRetryBehaviorRetryable, "exec "+what, err) + } + + select { + case pld := <-result: + if pld.Error() != nil { + // Worker-side execution failure: retryable per Nexus spec for INTERNAL. + return nil, newNexusHandlerError(nexus.HandlerErrorTypeInternal, nexus.HandlerErrorRetryBehaviorUnspecified, "nexus worker exec error", pld.Error()) + } + if pld.Payload().Flags&frame.STREAM != 0 { + ch <- struct{}{} + // Streaming worker replies violate the protocol; server-side fault. + return nil, newNexusHandlerError(nexus.HandlerErrorTypeInternal, nexus.HandlerErrorRetryBehaviorNonRetryable, "streaming is not supported", nil) + } + return pld.Payload(), nil + default: + // Pool returned a result channel without a value — should not happen on a + // healthy pool. Treat as transient. + return nil, newNexusHandlerError(nexus.HandlerErrorTypeInternal, nexus.HandlerErrorRetryBehaviorRetryable, "nexus worker empty response", nil) + } +} + +// decodeStartReply maps a PHP→Go reply into the SDK start-result shape. +// Variants: *NexusOperationStarted (sync/async), nil+Failure, nil+nil → HandlerError. +func (h *NexusHandler) decodeStartReply(ctx context.Context, retMsg *internal.Message) (nexus.HandlerStartOperationResult[converter.RawValue], error) { + switch reply := retMsg.Command.(type) { + case *internal.NexusOperationStarted: + forwardNexusLinks(ctx, reply.Links, h.log) + if reply.Async { + return &nexus.HandlerStartOperationResultAsync{ + OperationToken: reply.Token, + }, nil + } + var p *commonpb.Payload + if pls := retMsg.Payloads.GetPayloads(); len(pls) > 0 { + p = pls[0] + } + return &nexus.HandlerStartOperationResultSync[converter.RawValue]{ + Value: converter.NewRawValue(p), + }, nil + case nil: + if retMsg.Failure != nil { + return nil, nexusErrorFromFailure(retMsg.Failure) + } + return nil, newNexusHandlerError( + nexus.HandlerErrorTypeInternal, + nexus.HandlerErrorRetryBehaviorNonRetryable, + "nexus worker reply has neither command nor failure", nil) + default: + return nil, newNexusHandlerError( + nexus.HandlerErrorTypeInternal, + nexus.HandlerErrorRetryBehaviorNonRetryable, + fmt.Sprintf("unexpected nexus reply command %T", retMsg.Command), nil) + } +} + +// nexusLinksToInternal converts SDK links to wire form, dropping URL-less entries. +func nexusLinksToInternal(links []nexus.Link) []internal.NexusLink { + out := make([]internal.NexusLink, 0, len(links)) + for _, l := range links { + if l.URL == nil { + continue + } + out = append(out, internal.NexusLink{URL: l.URL.String(), Type: l.Type}) + } + return out +} + +// nexusLinksFromInternal: drop entries with empty url/type or unparseable URL. +func nexusLinksFromInternal(links []internal.NexusLink, log *zap.Logger) []nexus.Link { + if len(links) == 0 { + return nil + } + out := make([]nexus.Link, 0, len(links)) + for _, l := range links { + if l.URL == "" || l.Type == "" { + continue + } + u, err := url.Parse(l.URL) + if err != nil { + log.Warn("nexus link URL is malformed; skipping", zap.String("url", l.URL), zap.Error(err)) + continue + } + out = append(out, nexus.Link{URL: u, Type: l.Type}) + } + return out +} + +// forwardNexusLinks ships valid links to handler ctx; bare ctx → warn+drop. +func forwardNexusLinks(ctx context.Context, links []internal.NexusLink, log *zap.Logger) { + out := nexusLinksFromInternal(links, log) + if len(out) == 0 { + return + } + if !nexus.IsHandlerContext(ctx) { + log.Warn("nexus handler ctx missing; response links dropped", zap.Int("links", len(out))) + return + } + nexus.AddHandlerLinks(ctx, out...) +} + +// Cause holds the original proto via failureHolder so SDK-Go's ErrorToFailure +// round-trips it back without losing structure (same contract as activity.go). +func nexusErrorFromFailure(f *failurepb.Failure) error { + cause := temporal.GetDefaultFailureConverter().FailureToError(f) + + if nhf := f.GetNexusHandlerFailureInfo(); nhf != nil { + return &nexus.HandlerError{ + Type: nexus.HandlerErrorType(nhf.GetType()), + Message: f.GetMessage(), + RetryBehavior: mapNexusRetryBehavior(nhf.GetRetryBehavior()), + Cause: cause, + } + } + + if f.GetCanceledFailureInfo() != nil { + return &nexus.OperationError{ + State: nexus.OperationStateCanceled, + Message: f.GetMessage(), + Cause: cause, + } + } + + if app := f.GetApplicationFailureInfo(); app != nil { + if app.GetType() == nexusOperationErrorType { + return &nexus.OperationError{ + State: nexus.OperationStateFailed, + Message: f.GetMessage(), + Cause: cause, + } + } + + if t := app.GetType(); strings.HasPrefix(t, nexusOperationErrorTypePrefix) { + stateStr := strings.TrimPrefix(t, nexusOperationErrorTypePrefix) + state := nexus.OperationState(stateStr) + // Spec only allows failed/canceled; coerce anything else. + if state != nexus.OperationStateFailed && state != nexus.OperationStateCanceled { + state = nexus.OperationStateFailed + } + return &nexus.OperationError{ + State: state, + Message: f.GetMessage(), + Cause: cause, + } + } + } + + return &nexus.HandlerError{ + Type: nexus.HandlerErrorTypeInternal, + Message: f.GetMessage(), + Cause: cause, + } +} + +func mapNexusRetryBehavior(b enumspb.NexusHandlerErrorRetryBehavior) nexus.HandlerErrorRetryBehavior { + switch b { + case enumspb.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_RETRYABLE: + return nexus.HandlerErrorRetryBehaviorRetryable + case enumspb.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE: + return nexus.HandlerErrorRetryBehaviorNonRetryable + case enumspb.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_UNSPECIFIED: + return nexus.HandlerErrorRetryBehaviorUnspecified + default: + return nexus.HandlerErrorRetryBehaviorUnspecified + } +} + +func (h *NexusHandler) cancelOperation( + ctx context.Context, + taskQueue string, + serviceName string, + operationName string, + token string, + options nexus.CancelOperationOptions, +) error { + h.log.Debug("nexus cancel operation", zap.String("service", serviceName), zap.String("operation", operationName), zap.String("token", token), zap.String(tq, taskQueue)) + + msg := &internal.Message{ + ID: atomic.AddUint64(&h.seqID, 1), + Command: internal.CancelNexusOperation{ + Service: serviceName, + Operation: operationName, + Namespace: h.namespace, + TaskQueue: taskQueue, + OperationToken: token, + Headers: maps.Clone(options.Header), + }, + } + + r, err := h.roundTrip(ctx, taskQueue, msg, "nexus cancel request") + if err != nil { + return err + } + + return h.decodeCancelReply(r) +} + +// decodeCancelReply maps the PHP cancel reply: always exactly one message; a +// rejected cancel carries a Failure (HandlerException on the PHP side), a +// resolved cancel carries none. +func (h *NexusHandler) decodeCancelReply(r *payload.Payload) error { + out := make([]*internal.Message, 0, 1) + if err := h.codec.Decode(r, &out); err != nil { + return newNexusHandlerError(nexus.HandlerErrorTypeInternal, nexus.HandlerErrorRetryBehaviorNonRetryable, "decode nexus cancel response", err) + } + + if len(out) != 1 { + return newNexusHandlerError(nexus.HandlerErrorTypeInternal, nexus.HandlerErrorRetryBehaviorNonRetryable, "invalid nexus worker cancel response", nil) + } + + if out[0].Failure != nil { + return nexusErrorFromFailure(out[0].Failure) + } + + return nil +} + +// newNexusHandlerError constructs a *nexus.HandlerError with explicit type and +// retry behavior. cause may be nil. The cause string is appended to Message so +// it surfaces via Error() (HandlerError.Error() doesn't print Cause), while the +// original error is preserved for errors.Unwrap / errors.Is. +func newNexusHandlerError(typ nexus.HandlerErrorType, retry nexus.HandlerErrorRetryBehavior, message string, cause error) *nexus.HandlerError { + full := message + if cause != nil { + full = message + ": " + cause.Error() + } + return &nexus.HandlerError{ + Type: typ, + Message: full, + RetryBehavior: retry, + Cause: cause, + } +} + +// watchForMethodCancel: one goroutine per in-flight invocation; emits cancel on ctx.Done. +func (h *NexusHandler) watchForMethodCancel(ctx context.Context, invocationID uint64, done <-chan struct{}) { + select { + case <-ctx.Done(): + if _, ok := h.inFlight.Load(invocationID); !ok { + return + } + h.sendCancelMethod(invocationID, ctx.Err().Error()) + case <-done: + } +} + +// sendCancelMethod is fire-and-forget; failures are logged and swallowed. +func (h *NexusHandler) sendCancelMethod(invocationID uint64, reason string) { + msg := &internal.Message{ + ID: atomic.AddUint64(&h.seqID, 1), + Command: internal.CancelNexusOperationMethod{ + InvocationID: invocationID, + Reason: reason, + }, + } + + pl := h.getPld() + defer h.putPld(pl) + + if err := h.codec.Encode(&internal.Context{}, pl, msg); err != nil { + h.log.Warn("nexus cancel method encode failed", zap.Uint64("invocationID", invocationID), zap.Error(err)) + return + } + + // Original ctx is already canceled; use a fresh one with timeout so we don't + // hang forever if the pool is shutting down. + ctx, cancel := context.WithTimeout(context.Background(), nexusCancelMethodTimeout) + defer cancel() + ch := make(chan struct{}, 1) + result, err := h.pool.Exec(ctx, pl, ch) + if err != nil { + h.log.Warn("nexus cancel method exec failed", zap.Uint64("invocationID", invocationID), zap.Error(err)) + return + } + + select { + case pld := <-result: + if pld != nil && pld.Error() != nil { + h.log.Warn("nexus method cancel delivery failed", zap.Uint64("invocationID", invocationID), zap.Error(pld.Error())) + } + case <-ctx.Done(): + h.log.Warn("nexus method cancel delivery failed", zap.Uint64("invocationID", invocationID), zap.Error(ctx.Err())) + } +} + +func (h *NexusHandler) getPld() *payload.Payload { + return h.pldPool.Get().(*payload.Payload) +} + +func (h *NexusHandler) putPld(pld *payload.Payload) { + pld.Codec = 0 + pld.Context = nil + pld.Body = nil + h.pldPool.Put(pld) +} diff --git a/aggregatedpool/nexus_caller.go b/aggregatedpool/nexus_caller.go new file mode 100644 index 00000000..88169bbc --- /dev/null +++ b/aggregatedpool/nexus_caller.go @@ -0,0 +1,61 @@ +package aggregatedpool + +import ( + commonpb "go.temporal.io/api/common/v1" + "go.temporal.io/sdk/temporal" +) + +// NexusStartEnvelope is the response shape for GetNexusOperationStarted. +// Encoded via the workflow's data converter (default: JSON); PHP decodes via +// EncodedValues::getValue(0, NexusStartEnvelope::class). +type NexusStartEnvelope struct { + Async bool `json:"async"` + Token string `json:"token,omitempty"` +} + +func (wp *Workflow) makeNexusStartedRegistryCallback(startMsgID uint64) func(string, error) { + return func(token string, err error) { + if wp.deliverInline() { + wp.nexusStarted.Push(startMsgID, token, err) + return + } + wp.callbacks = append(wp.callbacks, func() error { + wp.nexusStarted.Push(startMsgID, token, err) + return nil + }) + } +} + +func (wp *Workflow) makeNexusCompletionResponseCallback(startMsgID uint64) func(*commonpb.Payload, error) { + deliver := func(result *commonpb.Payload, err error) { + wp.canceller.Discard(startMsgID) + if err != nil { + wp.mq.PushError(startMsgID, temporal.GetDefaultFailureConverter().ErrorToFailure(err), wp.getWorkflowWorkerPid()) + return + } + payloads := &commonpb.Payloads{} + if result != nil { + payloads.Payloads = []*commonpb.Payload{result} + } + wp.mq.PushResponse(startMsgID, payloads, wp.getWorkflowWorkerPid()) + } + return func(result *commonpb.Payload, err error) { + if wp.deliverInline() { + deliver(result, err) + return + } + wp.callbacks = append(wp.callbacks, func() error { + deliver(result, err) + return nil + }) + } +} + +func (wp *Workflow) pushStartEnvelope(awaitMsgID uint64, envelope NexusStartEnvelope) { + payloads, err := wp.env.GetDataConverter().ToPayloads(envelope) + if err != nil { + wp.mq.PushError(awaitMsgID, temporal.GetDefaultFailureConverter().ErrorToFailure(err), wp.getWorkflowWorkerPid()) + return + } + wp.mq.PushResponse(awaitMsgID, payloads, wp.getWorkflowWorkerPid()) +} diff --git a/aggregatedpool/nexus_caller_test.go b/aggregatedpool/nexus_caller_test.go new file mode 100644 index 00000000..d093ca3b --- /dev/null +++ b/aggregatedpool/nexus_caller_test.go @@ -0,0 +1,211 @@ +package aggregatedpool + +import ( + "encoding/json" + "errors" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/temporalio/roadrunner-temporal/v5/canceller" + "github.com/temporalio/roadrunner-temporal/v5/queue" + "github.com/temporalio/roadrunner-temporal/v5/registry" + commonpb "go.temporal.io/api/common/v1" + "go.temporal.io/sdk/converter" + "go.uber.org/zap" +) + +// ── NexusStartEnvelope JSON wire shape ──────────────────────────────── + +func TestNexusStartEnvelope_AsyncShape(t *testing.T) { + data, err := json.Marshal(NexusStartEnvelope{Async: true, Token: "tok-123"}) + require.NoError(t, err) + assert.JSONEq(t, `{"async":true,"token":"tok-123"}`, string(data)) +} + +func TestNexusStartEnvelope_SyncShape(t *testing.T) { + data, err := json.Marshal(NexusStartEnvelope{Async: false, Token: ""}) + require.NoError(t, err) + assert.JSONEq(t, `{"async":false}`, string(data)) + assert.NotContains(t, string(data), "token") +} + +// Round-trip through the actual data converter PHP receives the envelope +// from — guards against silent breakage if the default converter changes. +func TestNexusStartEnvelope_RoundTripViaTemporalConverter(t *testing.T) { + conv := converter.GetDefaultDataConverter() + payloads, err := conv.ToPayloads(NexusStartEnvelope{Async: true, Token: "round-trip"}) + require.NoError(t, err) + require.Len(t, payloads.Payloads, 1) + + var back NexusStartEnvelope + require.NoError(t, conv.FromPayloads(payloads, &back)) + assert.True(t, back.Async) + assert.Equal(t, "round-trip", back.Token) +} + +// ── Caller-side callbacks fixture ───────────────────────────────── + +// newCallerWorkflow builds the minimum Workflow needed to exercise the +// nexus-caller callbacks. recordingPool.Workers() returns nil → pid 0. +func newCallerWorkflow(t *testing.T) *Workflow { + t.Helper() + return &Workflow{ + log: zap.NewNop(), + mq: queue.NewMessageQueue(func() uint64 { return 0 }), + canceller: new(canceller.Canceller), + nexusStarted: new(registry.NexusStartedRegistry), + pool: &recordingPool{}, + } +} + +// ── makeNexusStartedRegistryCallback ────────────────────────────────── + +func TestMakeNexusStartedRegistryCallback_InLoopPushesImmediately(t *testing.T) { + wp := newCallerWorkflow(t) + atomic.StoreUint32(&wp.inLoop, 1) + + cb := wp.makeNexusStartedRegistryCallback(101) + cb("tok-async", nil) + + assert.Empty(t, wp.callbacks, "in-loop callback must not be deferred") + + var gotToken string + wp.nexusStarted.Listen(101, func(token string, err error) { + gotToken = token + }) + assert.Equal(t, "tok-async", gotToken, "registry must contain the pushed entry") +} + +func TestMakeNexusStartedRegistryCallback_OutOfLoopDefers(t *testing.T) { + wp := newCallerWorkflow(t) + atomic.StoreUint32(&wp.inLoop, 0) + + cb := wp.makeNexusStartedRegistryCallback(202) + cb("tok-deferred", nil) + + require.Len(t, wp.callbacks, 1, "out-of-loop callback must be deferred") + + var fired bool + wp.nexusStarted.Listen(202, func(string, error) { fired = true }) + assert.False(t, fired) + + require.NoError(t, wp.callbacks[0]()) + assert.True(t, fired, "registry listener must fire once the deferred callback runs") +} + +func TestMakeNexusStartedRegistryCallback_ErrorIsForwarded(t *testing.T) { + wp := newCallerWorkflow(t) + atomic.StoreUint32(&wp.inLoop, 1) + + startErr := errors.New("start blew up") + cb := wp.makeNexusStartedRegistryCallback(303) + cb("", startErr) + + var gotErr error + wp.nexusStarted.Listen(303, func(token string, err error) { + gotErr = err + }) + assert.Same(t, startErr, gotErr) +} + +// ── makeNexusCompletionResponseCallback ─────────────────────────────── + +func TestMakeNexusCompletionResponseCallback_InLoopSuccess(t *testing.T) { + wp := newCallerWorkflow(t) + atomic.StoreUint32(&wp.inLoop, 1) + + pl := &commonpb.Payload{ + Metadata: map[string][]byte{"encoding": []byte("json/plain")}, + Data: []byte(`"hello"`), + } + + cb := wp.makeNexusCompletionResponseCallback(404) + cb(pl, nil) + + assert.Empty(t, wp.callbacks, "in-loop callback must not be deferred") + + msgs := wp.mq.Messages() + require.Len(t, msgs, 1) + assert.Equal(t, uint64(404), msgs[0].ID) + assert.Nil(t, msgs[0].Failure, "success path must not push a failure") + require.NotNil(t, msgs[0].Payloads) + require.Len(t, msgs[0].Payloads.Payloads, 1) + assert.Same(t, pl, msgs[0].Payloads.Payloads[0]) +} + +// Sync ops can complete with nil payload — PHP still needs a response with +// a zero-Payloads bag, not no message at all. +func TestMakeNexusCompletionResponseCallback_NilPayloadStillPushesEmptyResponse(t *testing.T) { + wp := newCallerWorkflow(t) + atomic.StoreUint32(&wp.inLoop, 1) + + cb := wp.makeNexusCompletionResponseCallback(505) + cb(nil, nil) + + msgs := wp.mq.Messages() + require.Len(t, msgs, 1) + require.NotNil(t, msgs[0].Payloads) + assert.Empty(t, msgs[0].Payloads.Payloads, "nil result must produce zero-payload Payloads") + assert.Nil(t, msgs[0].Failure) +} + +func TestMakeNexusCompletionResponseCallback_ErrorPath(t *testing.T) { + wp := newCallerWorkflow(t) + atomic.StoreUint32(&wp.inLoop, 1) + + cb := wp.makeNexusCompletionResponseCallback(606) + cb(nil, errors.New("nexus operation failed")) + + msgs := wp.mq.Messages() + require.Len(t, msgs, 1) + assert.Equal(t, uint64(606), msgs[0].ID) + require.NotNil(t, msgs[0].Failure) + assert.Equal(t, "nexus operation failed", msgs[0].Failure.Message) + assert.Nil(t, msgs[0].Payloads) +} + +func TestMakeNexusCompletionResponseCallback_OutOfLoopDefers(t *testing.T) { + wp := newCallerWorkflow(t) + atomic.StoreUint32(&wp.inLoop, 0) + + cb := wp.makeNexusCompletionResponseCallback(707) + cb(&commonpb.Payload{Data: []byte("x")}, nil) + + assert.Empty(t, wp.mq.Messages(), "deferred completion must not produce a message yet") + require.Len(t, wp.callbacks, 1) + + require.NoError(t, wp.callbacks[0]()) + assert.Len(t, wp.mq.Messages(), 1, "deferred callback run produces one queued message") +} + +// A fast op can complete before PHP sends GetNexusOperationStarted. Completion +// must NOT discard the started slot, or the later consume hangs forever. +func TestMakeNexusCompletionResponseCallback_KeepsStartedSlotForLateConsume(t *testing.T) { + wp := newCallerWorkflow(t) + atomic.StoreUint32(&wp.inLoop, 1) + + wp.makeNexusStartedRegistryCallback(808)("tok-late", nil) + wp.makeNexusCompletionResponseCallback(808)(nil, nil) + + var got string + var fired bool + wp.nexusStarted.Listen(808, func(token string, err error) { got, fired = token, true }) + + require.True(t, fired, "start slot must survive completion so a late consume still resolves") + assert.Equal(t, "tok-late", got) +} + +func TestMakeNexusCompletionResponseCallback_DiscardsCancellerSlot(t *testing.T) { + wp := newCallerWorkflow(t) + atomic.StoreUint32(&wp.inLoop, 1) + + var canceled bool + wp.canceller.Register(909, func() error { canceled = true; return nil }) + + wp.makeNexusCompletionResponseCallback(909)(nil, nil) + require.NoError(t, wp.canceller.Cancel(909)) + + assert.False(t, canceled, "completion must discard the canceller slot") +} diff --git a/aggregatedpool/nexus_test.go b/aggregatedpool/nexus_test.go new file mode 100644 index 00000000..e2cf3b36 --- /dev/null +++ b/aggregatedpool/nexus_test.go @@ -0,0 +1,1175 @@ +package aggregatedpool + +import ( + "context" + "errors" + "net/url" + "sync" + "sync/atomic" + "testing" + + "github.com/roadrunner-server/pool/payload" + staticPool "github.com/roadrunner-server/pool/pool/static_pool" + poolWorker "github.com/roadrunner-server/pool/worker" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/temporalio/roadrunner-temporal/v5/internal" + commonpb "go.temporal.io/api/common/v1" + enumspb "go.temporal.io/api/enums/v1" + failurepb "go.temporal.io/api/failure/v1" + "go.temporal.io/sdk/converter" + "go.temporal.io/sdk/temporal" + "go.uber.org/zap" + "google.golang.org/protobuf/proto" + + "github.com/nexus-rpc/sdk-go/nexus" +) + +// ── Service registration ────────────────────────────────────── + +// putPld must clear the payload before returning it to the pool — otherwise +// stale Body/Context bytes leak into the next Encode call. +func TestNexusHandler_PayloadPoolResetsOnPut(t *testing.T) { + handler := NewNexusHandler(nil, nil, zap.NewNop(), "default") + + pld := handler.getPld() + pld.Body = []byte("test") + pld.Context = []byte("ctx") + pld.Codec = 7 + + handler.putPld(pld) + assert.Nil(t, pld.Body) + assert.Nil(t, pld.Context) + assert.Equal(t, uint8(0), pld.Codec) +} + +func TestNexusHandler_CreateNexusService_RegistersOperations(t *testing.T) { + handler := NewNexusHandler(nil, nil, zap.NewNop(), "default") + svc := handler.CreateNexusService("tq", "GreetingService", []string{"greet", "farewell"}) + + require.NotNil(t, svc.Operation("greet")) + require.NotNil(t, svc.Operation("farewell")) + assert.Nil(t, svc.Operation("missing")) +} + +func TestNexusHandler_CreateNexusService_AcceptsEmptyAndNilOperations(t *testing.T) { + handler := NewNexusHandler(nil, nil, zap.NewNop(), "default") + require.NotPanics(t, func() { handler.CreateNexusService("tq", "S", nil) }) + require.NotPanics(t, func() { handler.CreateNexusService("tq", "S", []string{}) }) +} + +func TestNexusHandler_CreateNexusService_IsolatesOperationsBetweenServices(t *testing.T) { + handler := NewNexusHandler(nil, nil, zap.NewNop(), "default") + a := handler.CreateNexusService("tq", "ServiceA", []string{"opA"}) + b := handler.CreateNexusService("tq", "ServiceB", []string{"opB"}) + + assert.NotNil(t, a.Operation("opA")) + assert.Nil(t, a.Operation("opB")) + assert.NotNil(t, b.Operation("opB")) + assert.Nil(t, b.Operation("opA")) +} + +// Compile-time guarantee that nexusOperation satisfies the nexus SDK interfaces. +var ( + _ nexus.RegisterableOperation = (*nexusOperation)(nil) + _ nexus.Operation[converter.RawValue, converter.RawValue] = (*nexusOperation)(nil) +) + +// TaskQueue from CreateNexusService must reach each operation — that's the +// link the dispatch path follows when a task arrives. +func TestNexusOperation_TaskQueuePropagation(t *testing.T) { + log := zap.NewNop() + handler := NewNexusHandler(nil, nil, log, "default") + + taskQueue := "my-special-queue" + svc := handler.CreateNexusService(taskQueue, "Svc", []string{"op1", "op2", "op3"}) + require.NotNil(t, svc) + + for _, opName := range []string{"op1", "op2", "op3"} { + op := svc.Operation(opName) + require.NotNil(t, op) + + concrete, ok := op.(*nexusOperation) + require.True(t, ok, "operation %q is not *nexusOperation", opName) + assert.Equal(t, taskQueue, concrete.taskQueue, "operation %q has wrong task queue", opName) + assert.Equal(t, "Svc", concrete.serviceName, "operation %q has wrong service name", opName) + assert.Equal(t, opName, concrete.name) + } +} + +// ── Mock codec ───────────────────────────────────────────── + +type mockCodec struct { + encodeCalled int32 + decodeCalled int32 + encodeErr error + decodeErr error + encodedCtx *internal.Context + encodedMsg *internal.Message + decodeMsgs []*internal.Message +} + +func (m *mockCodec) Encode(ctx *internal.Context, p *payload.Payload, msgs ...*internal.Message) error { + atomic.AddInt32(&m.encodeCalled, 1) + m.encodedCtx = ctx + if len(msgs) > 0 { + m.encodedMsg = msgs[0] + } + if m.encodeErr != nil { + return m.encodeErr + } + p.Body = []byte("encoded") + return nil +} + +func (m *mockCodec) Decode(p *payload.Payload, msgs *[]*internal.Message) error { + atomic.AddInt32(&m.decodeCalled, 1) + if m.decodeErr != nil { + return m.decodeErr + } + *msgs = append(*msgs, m.decodeMsgs...) + return nil +} + +func (m *mockCodec) DecodeWorkerInfo(_ *payload.Payload, _ *[]*internal.WorkerInfo) error { + return nil +} + +// recordingCodec is a thread-safe codec stub that pushes every Encode'd +// message ID into a channel. Used by concurrency tests. +type recordingCodec struct { + ids chan<- uint64 + stopErr error +} + +func (c *recordingCodec) Encode(_ *internal.Context, _ *payload.Payload, msgs ...*internal.Message) error { + if len(msgs) > 0 { + c.ids <- msgs[0].ID + } + return c.stopErr +} + +func (c *recordingCodec) Decode(_ *payload.Payload, _ *[]*internal.Message) error { return nil } +func (c *recordingCodec) DecodeWorkerInfo(_ *payload.Payload, _ *[]*internal.WorkerInfo) error { + return nil +} + +// ── Encoding behavior tests (no pool needed) ─────────────────────── + +func TestStartOperation_EncodesTaskQueue(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("stop after encode"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + _, err := handler.startOperation( + context.Background(), + "my-task-queue", + "GreetingService", + "greet", + nil, + nexus.StartOperationOptions{}, + ) + + require.Error(t, err) + assert.NotNil(t, codec.encodedCtx, "Context should be set on encode call") + assert.Equal(t, "my-task-queue", codec.encodedCtx.TaskQueue) +} + +func TestStartOperation_EncodesAllFields(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("stop after encode"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + _, _ = handler.startOperation( + context.Background(), + "tq", + "MyService", + "myOp", + nil, + nexus.StartOperationOptions{ + RequestID: "req-123", + CallbackURL: "http://callback.example.com", + Header: nexus.Header{ + "Content-Type": "application/json", + "Authorization": "Bearer xyz", + }, + CallbackHeader: nexus.Header{ + "X-Token": "callback-token", + }, + }, + ) + + require.NotNil(t, codec.encodedMsg) + cmd, ok := codec.encodedMsg.Command.(internal.InvokeNexusOperation) + require.True(t, ok, "Command should be InvokeNexusOperation, got %T", codec.encodedMsg.Command) + + assert.Equal(t, "MyService", cmd.Service) + assert.Equal(t, "myOp", cmd.Operation) + assert.Equal(t, "default", cmd.Namespace) + assert.Equal(t, "tq", cmd.TaskQueue) + assert.Equal(t, "req-123", cmd.RequestID) + assert.Equal(t, "http://callback.example.com", cmd.Callback) + assert.Equal(t, "application/json", cmd.Headers["Content-Type"]) + assert.Equal(t, "Bearer xyz", cmd.Headers["Authorization"]) + assert.Equal(t, "callback-token", cmd.CallbackHeaders["X-Token"]) +} + +func TestStartOperation_EncodesPayload(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("stop after encode"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + input := &commonpb.Payload{ + Data: []byte("hello"), + Metadata: map[string][]byte{"encoding": []byte("json/plain")}, + } + + _, _ = handler.startOperation( + context.Background(), + "tq", + "Svc", + "op", + input, + nexus.StartOperationOptions{}, + ) + + require.NotNil(t, codec.encodedMsg) + require.NotNil(t, codec.encodedMsg.Payloads) + require.Len(t, codec.encodedMsg.Payloads.Payloads, 1) + assert.Equal(t, []byte("hello"), codec.encodedMsg.Payloads.Payloads[0].Data) +} + +func TestStartOperation_NilInputProducesNilPayloads(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("stop after encode"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + _, _ = handler.startOperation( + context.Background(), + "tq", + "Svc", + "op", + nil, + nexus.StartOperationOptions{}, + ) + + require.NotNil(t, codec.encodedMsg) + assert.Nil(t, codec.encodedMsg.Payloads, "Payloads should be nil when input is nil") +} + +func TestStartOperation_EncodeErrorReturnsError(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("boom"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + result, err := handler.startOperation( + context.Background(), + "tq", + "Svc", + "op", + nil, + nexus.StartOperationOptions{}, + ) + + require.Error(t, err) + assert.Nil(t, result) + assert.Contains(t, err.Error(), "boom") +} + +func TestStartOperation_IncrementsSeqID(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("stop"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + _, _ = handler.startOperation(context.Background(), "tq", "S", "o", nil, nexus.StartOperationOptions{}) + firstID := codec.encodedMsg.ID + + _, _ = handler.startOperation(context.Background(), "tq", "S", "o", nil, nexus.StartOperationOptions{}) + secondID := codec.encodedMsg.ID + + assert.Greater(t, secondID, firstID, "seqID should increment between calls") +} + +func TestStartOperation_EncodesCallerLinks(t *testing.T) { + codec := &mockCodec{encodeErr: errors.New("stop")} + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + u1, err := url.Parse("https://caller.example/res/1") + require.NoError(t, err) + u2, err := url.Parse("https://caller.example/res/2") + require.NoError(t, err) + + _, _ = handler.startOperation(context.Background(), "tq", "S", "op", nil, nexus.StartOperationOptions{ + Links: []nexus.Link{ + {URL: u1, Type: "example.one"}, + {URL: u2, Type: "example.two"}, + }, + }) + + require.NotNil(t, codec.encodedMsg) + cmd, ok := codec.encodedMsg.Command.(internal.InvokeNexusOperation) + require.True(t, ok, "Command should be InvokeNexusOperation, got %T", codec.encodedMsg.Command) + require.Len(t, cmd.Links, 2) + assert.Equal(t, "https://caller.example/res/1", cmd.Links[0].URL) + assert.Equal(t, "example.one", cmd.Links[0].Type) + assert.Equal(t, "https://caller.example/res/2", cmd.Links[1].URL) + assert.Equal(t, "example.two", cmd.Links[1].Type) +} + +func TestStartOperation_NoLinksOmitsField(t *testing.T) { + codec := &mockCodec{encodeErr: errors.New("stop")} + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + _, _ = handler.startOperation(context.Background(), "tq", "S", "op", nil, nexus.StartOperationOptions{}) + + require.NotNil(t, codec.encodedMsg) + cmd, ok := codec.encodedMsg.Command.(internal.InvokeNexusOperation) + require.True(t, ok, "Command should be InvokeNexusOperation, got %T", codec.encodedMsg.Command) + assert.Empty(t, cmd.Links, "Links should be empty when options.Links is nil") +} + +// ── Cancel encoding tests ────────────────────────────────────── + +func TestCancelOperation_EncodesTaskQueue(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("stop after encode"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + err := handler.cancelOperation( + context.Background(), + "my-tq", + "Svc", + "op", + "token-xyz", + nexus.CancelOperationOptions{}, + ) + + require.Error(t, err) + require.NotNil(t, codec.encodedCtx) + assert.Equal(t, "my-tq", codec.encodedCtx.TaskQueue) +} + +func TestCancelOperation_EncodesAllFields(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("stop"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + _ = handler.cancelOperation( + context.Background(), + "tq", + "GreetingService", + "greet", + "async-token-123", + nexus.CancelOperationOptions{}, + ) + + require.NotNil(t, codec.encodedMsg) + cmd, ok := codec.encodedMsg.Command.(internal.CancelNexusOperation) + require.True(t, ok, "Command should be CancelNexusOperation, got %T", codec.encodedMsg.Command) + + assert.Equal(t, "GreetingService", cmd.Service) + assert.Equal(t, "greet", cmd.Operation) + assert.Equal(t, "default", cmd.Namespace) + assert.Equal(t, "tq", cmd.TaskQueue) + assert.Equal(t, "async-token-123", cmd.OperationToken) +} + +// TestCancelOperation_ExtractsHeaders verifies the caller's cancel-request +// headers are pulled off nexus.CancelOperationOptions.Header onto the command, +// symmetric with startOperation. The PHP CancelNexusOperation router reads these +// under `headers` and surfaces them on the handler's OperationContext. +func TestCancelOperation_ExtractsHeaders(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("stop"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + _ = handler.cancelOperation( + context.Background(), + "tq", + "GreetingService", + "greet", + "async-token-123", + nexus.CancelOperationOptions{ + Header: nexus.Header{ + "X-Nexus-Trace-Id": "trace-1", + "Authorization": "Bearer xyz", + }, + }, + ) + + require.NotNil(t, codec.encodedMsg) + cmd, ok := codec.encodedMsg.Command.(internal.CancelNexusOperation) + require.True(t, ok, "Command should be CancelNexusOperation, got %T", codec.encodedMsg.Command) + + assert.Equal(t, "trace-1", cmd.Headers["X-Nexus-Trace-Id"]) + assert.Equal(t, "Bearer xyz", cmd.Headers["Authorization"]) +} + +// TestStartOperation_EncodesNamespace verifies the handler's configured +// namespace lands on the InvokeNexusOperation command, where PHP reads it as +// $options['namespace']. Sourced from plugin config, not from internal.Context. +func TestStartOperation_EncodesNamespace(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("stop after encode"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "test-ns") + + _, _ = handler.startOperation( + context.Background(), + "tq", + "MyService", + "myOp", + nil, + nexus.StartOperationOptions{}, + ) + + require.NotNil(t, codec.encodedMsg) + cmd, ok := codec.encodedMsg.Command.(internal.InvokeNexusOperation) + require.True(t, ok, "Command should be InvokeNexusOperation, got %T", codec.encodedMsg.Command) + assert.Equal(t, "test-ns", cmd.Namespace) +} + +// TestCancelOperation_EncodesNamespace is the cancel-side counterpart. +func TestCancelOperation_EncodesNamespace(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("stop"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "test-ns") + + _ = handler.cancelOperation( + context.Background(), + "tq", + "GreetingService", + "greet", + "async-token-123", + nexus.CancelOperationOptions{}, + ) + + require.NotNil(t, codec.encodedMsg) + cmd, ok := codec.encodedMsg.Command.(internal.CancelNexusOperation) + require.True(t, ok, "Command should be CancelNexusOperation, got %T", codec.encodedMsg.Command) + assert.Equal(t, "test-ns", cmd.Namespace) +} + +// ── Cancel reply decoding ─────────────────────────────────────── + +// A PHP-side rejection (e.g. HandlerException NOT_IMPLEMENTED for a manual-token +// operation without an #[OperationCancel] routine) must surface as a handler +// error instead of being swallowed as cancel success. +func TestDecodeCancelReply_FailurePropagatesHandlerError(t *testing.T) { + codec := &mockCodec{ + decodeMsgs: []*internal.Message{ + { + Failure: &failurepb.Failure{ + Message: "cancellation is not supported", + FailureInfo: &failurepb.Failure_NexusHandlerFailureInfo{ + NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{ + Type: "NOT_IMPLEMENTED", + RetryBehavior: enumspb.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE, + }, + }, + }, + }, + }, + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + err := handler.decodeCancelReply(&payload.Payload{}) + + var he *nexus.HandlerError + require.ErrorAs(t, err, &he, "expected *nexus.HandlerError, got %T", err) + assert.Equal(t, nexus.HandlerErrorTypeNotImplemented, he.Type) + assert.Equal(t, nexus.HandlerErrorRetryBehaviorNonRetryable, he.RetryBehavior) + assert.Equal(t, "cancellation is not supported", he.Message) +} + +func TestDecodeCancelReply_NoFailureMeansSuccess(t *testing.T) { + codec := &mockCodec{ + decodeMsgs: []*internal.Message{{}}, + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + assert.NoError(t, handler.decodeCancelReply(&payload.Payload{})) +} + +// PHP always replies with exactly one message; an empty reply is a protocol +// fault, not cancel success. +func TestDecodeCancelReply_EmptyReplyIsProtocolFault(t *testing.T) { + handler := NewNexusHandler(&mockCodec{}, nil, zap.NewNop(), "default") + + err := handler.decodeCancelReply(&payload.Payload{}) + + var he *nexus.HandlerError + require.ErrorAs(t, err, &he, "expected *nexus.HandlerError, got %T", err) + assert.Equal(t, nexus.HandlerErrorTypeInternal, he.Type) + assert.Equal(t, nexus.HandlerErrorRetryBehaviorNonRetryable, he.RetryBehavior) +} + +func TestDecodeCancelReply_DecodeErrorIsInternalNonRetryable(t *testing.T) { + codec := &mockCodec{decodeErr: errors.New("bad frame")} + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + err := handler.decodeCancelReply(&payload.Payload{}) + + var he *nexus.HandlerError + require.ErrorAs(t, err, &he, "expected *nexus.HandlerError, got %T", err) + assert.Equal(t, nexus.HandlerErrorTypeInternal, he.Type) + assert.Equal(t, nexus.HandlerErrorRetryBehaviorNonRetryable, he.RetryBehavior) +} + +func TestCancelOperation_EncodeError(t *testing.T) { + codec := &mockCodec{ + encodeErr: errors.New("encode failed"), + } + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + err := handler.cancelOperation(context.Background(), "tq", "S", "o", "t", nexus.CancelOperationOptions{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "encode failed") +} + +// ── Concurrent seqID test ────────────────────────────────────── + +func TestStartOperation_ConcurrentSeqIDIncrement(t *testing.T) { + const goroutines = 20 + + idCh := make(chan uint64, goroutines) + codec := &recordingCodec{ids: idCh, stopErr: errors.New("stop")} + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + var wg sync.WaitGroup + for i := 0; i < goroutines; i++ { + wg.Add(1) + go func() { + defer wg.Done() + _, _ = handler.startOperation(context.Background(), "tq", "S", "o", nil, nexus.StartOperationOptions{}) + }() + } + wg.Wait() + close(idCh) + + unique := make(map[uint64]struct{}, goroutines) + for id := range idCh { + assert.Greater(t, id, uint64(0)) + unique[id] = struct{}{} + } + assert.Equal(t, goroutines, len(unique), "every concurrent startOperation must get a unique seqID") +} + +// ── Method cancellation tests ────────────────────────────────── + +// InvocationID is the correlation key CancelNexusOperationMethod uses to find +// the in-flight handler; PHP correlates via this field only. +func TestStartOperation_SetsInvocationID(t *testing.T) { + codec := &mockCodec{encodeErr: errors.New("stop")} + handler := NewNexusHandler(codec, nil, zap.NewNop(), "default") + + _, _ = handler.startOperation( + context.Background(), "tq", "S", "op", + nil, nexus.StartOperationOptions{}, + ) + + require.NotNil(t, codec.encodedMsg) + cmd, ok := codec.encodedMsg.Command.(internal.InvokeNexusOperation) + require.True(t, ok) + assert.NotZero(t, cmd.InvocationID) +} + +// ── sendCancelMethod tests (requires a minimal Pool mock) ────── + +type recordingPool struct { + execCalls int32 + lastCtx context.Context + lastCtxErrAtExec error + lastPld *payload.Payload + execCh chan struct{} +} + +func (p *recordingPool) Exec(ctx context.Context, pld *payload.Payload, _ chan struct{}) (chan *staticPool.PExec, error) { + atomic.AddInt32(&p.execCalls, 1) + p.lastCtx = ctx + p.lastCtxErrAtExec = ctx.Err() + p.lastPld = pld + if p.execCh != nil { + <-p.execCh + } + ch := make(chan *staticPool.PExec, 1) + ch <- &staticPool.PExec{} + return ch, nil +} + +func (p *recordingPool) Workers() []*poolWorker.Process { return nil } +func (p *recordingPool) RemoveWorker(context.Context) error { panic("not used") } +func (p *recordingPool) AddWorker() error { panic("not used") } +func (p *recordingPool) QueueSize() uint64 { panic("not used") } +func (p *recordingPool) Reset(context.Context) error { panic("not used") } + +// sendCancelMethod must use a fresh context — the caller's ctx is the one +// that was just canceled, so reusing it would mean the cancel never lands. +func TestSendCancelMethod_EncodesCorrectCommand(t *testing.T) { + codec := &mockCodec{} + pool := &recordingPool{} + handler := NewNexusHandler(codec, pool, zap.NewNop(), "default") + + handler.sendCancelMethod(77, "deadline exceeded") + + require.NotNil(t, codec.encodedMsg) + cmd, ok := codec.encodedMsg.Command.(internal.CancelNexusOperationMethod) + require.True(t, ok, "Command should be CancelNexusOperationMethod, got %T", codec.encodedMsg.Command) + assert.Equal(t, uint64(77), cmd.InvocationID) + assert.Equal(t, "deadline exceeded", cmd.Reason) + + assert.EqualValues(t, 1, atomic.LoadInt32(&pool.execCalls)) + require.NotNil(t, pool.lastCtx) + assert.NoError(t, pool.lastCtxErrAtExec, "sendCancelMethod must use a live ctx at Exec time") +} + +// ctx cancel while a Nexus invocation is in-flight must emit a +// CancelNexusOperationMethod so the PHP-side handler stops promptly. +func TestStartOperation_CtxCancelTriggersMethodCancel(t *testing.T) { + codec := &mockCodec{} + pool := &recordingPool{} + handler := NewNexusHandler(codec, pool, zap.NewNop(), "default") + + handler.inFlight.Store(uint64(5), struct{}{}) + defer handler.inFlight.Delete(uint64(5)) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + + finished := make(chan struct{}) + go func() { + handler.watchForMethodCancel(ctx, 5, done) + close(finished) + }() + + cancel() + <-finished + + require.NotNil(t, codec.encodedMsg) + cmd, ok := codec.encodedMsg.Command.(internal.CancelNexusOperationMethod) + require.True(t, ok, "expected CancelNexusOperationMethod, got %T", codec.encodedMsg.Command) + assert.Equal(t, uint64(5), cmd.InvocationID) + assert.Contains(t, cmd.Reason, "canceled") +} + +func TestStartOperation_DoneClosedSkipsMethodCancel(t *testing.T) { + codec := &mockCodec{} + pool := &recordingPool{} + handler := NewNexusHandler(codec, pool, zap.NewNop(), "default") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan struct{}) + + finished := make(chan struct{}) + go func() { + handler.watchForMethodCancel(ctx, 5, done) + close(finished) + }() + + close(done) + <-finished + + assert.Nil(t, codec.encodedMsg, "no cancel should have been emitted") + assert.EqualValues(t, 0, atomic.LoadInt32(&pool.execCalls)) +} + +// Race guard: ctx cancels AFTER the invocation completed (inFlight already +// cleared). Watcher must swallow the cancel rather than target a gone handler. +func TestStartOperation_CtxCancelAfterCompletionNoop(t *testing.T) { + codec := &mockCodec{} + pool := &recordingPool{} + handler := NewNexusHandler(codec, pool, zap.NewNop(), "default") + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + + finished := make(chan struct{}) + go func() { + handler.watchForMethodCancel(ctx, 777, done) + close(finished) + }() + + cancel() + <-finished + + assert.Nil(t, codec.encodedMsg, "cancel must be swallowed when inFlight entry is absent") +} + +// sendCancelMethod is fire-and-forget — encode failures are logged, never propagated. +func TestSendCancelMethod_EncodeErrorSwallowed(t *testing.T) { + codec := &mockCodec{encodeErr: errors.New("encode boom")} + pool := &recordingPool{} + handler := NewNexusHandler(codec, pool, zap.NewNop(), "default") + + handler.sendCancelMethod(1, "x") + + assert.EqualValues(t, 0, atomic.LoadInt32(&pool.execCalls)) +} + +// ── Failure → Nexus error mapping ────────────────────────────────── + +func TestNexusErrorFromFailure_HandlerFailureInfoPreservesType(t *testing.T) { + f := &failurepb.Failure{ + Message: "payload parsing failed", + FailureInfo: &failurepb.Failure_NexusHandlerFailureInfo{ + NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{ + Type: "BAD_REQUEST", + RetryBehavior: enumspb.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE, + }, + }, + } + + err := nexusErrorFromFailure(f) + + var he *nexus.HandlerError + require.ErrorAs(t, err, &he, "expected *nexus.HandlerError, got %T", err) + assert.Equal(t, nexus.HandlerErrorTypeBadRequest, he.Type) + assert.Equal(t, nexus.HandlerErrorRetryBehaviorNonRetryable, he.RetryBehavior) + assert.Equal(t, "payload parsing failed", he.Message) +} + +func TestNexusErrorFromFailure_RetryBehaviorRetryable(t *testing.T) { + f := &failurepb.Failure{ + Message: "try again", + FailureInfo: &failurepb.Failure_NexusHandlerFailureInfo{ + NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{ + Type: "INTERNAL", + RetryBehavior: enumspb.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_RETRYABLE, + }, + }, + } + + var he *nexus.HandlerError + require.ErrorAs(t, nexusErrorFromFailure(f), &he) + assert.Equal(t, nexus.HandlerErrorTypeInternal, he.Type) + assert.Equal(t, nexus.HandlerErrorRetryBehaviorRetryable, he.RetryBehavior) +} + +func TestNexusErrorFromFailure_RetryBehaviorUnspecifiedDefaults(t *testing.T) { + f := &failurepb.Failure{ + Message: "", + FailureInfo: &failurepb.Failure_NexusHandlerFailureInfo{ + NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{ + Type: "NOT_FOUND", + }, + }, + } + + var he *nexus.HandlerError + require.ErrorAs(t, nexusErrorFromFailure(f), &he) + assert.Equal(t, nexus.HandlerErrorTypeNotFound, he.Type) + assert.Equal(t, nexus.HandlerErrorRetryBehaviorUnspecified, he.RetryBehavior) +} + +func TestNexusErrorFromFailure_AllSpecErrorTypesRoundTrip(t *testing.T) { + cases := []struct { + wire string + want nexus.HandlerErrorType + }{ + {"BAD_REQUEST", nexus.HandlerErrorTypeBadRequest}, + {"UNAUTHENTICATED", nexus.HandlerErrorTypeUnauthenticated}, + {"UNAUTHORIZED", nexus.HandlerErrorTypeUnauthorized}, + {"NOT_FOUND", nexus.HandlerErrorTypeNotFound}, + {"REQUEST_TIMEOUT", nexus.HandlerErrorTypeRequestTimeout}, + {"CONFLICT", nexus.HandlerErrorTypeConflict}, + {"RESOURCE_EXHAUSTED", nexus.HandlerErrorTypeResourceExhausted}, + {"INTERNAL", nexus.HandlerErrorTypeInternal}, + {"NOT_IMPLEMENTED", nexus.HandlerErrorTypeNotImplemented}, + {"UNAVAILABLE", nexus.HandlerErrorTypeUnavailable}, + {"UPSTREAM_TIMEOUT", nexus.HandlerErrorTypeUpstreamTimeout}, + } + + for _, c := range cases { + t.Run(c.wire, func(t *testing.T) { + f := &failurepb.Failure{ + Message: c.wire, + FailureInfo: &failurepb.Failure_NexusHandlerFailureInfo{ + NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{ + Type: c.wire, + }, + }, + } + var he *nexus.HandlerError + require.ErrorAs(t, nexusErrorFromFailure(f), &he) + assert.Equal(t, c.want, he.Type) + }) + } +} + +func TestNexusErrorFromFailure_OperationErrorFailed(t *testing.T) { + f := &failurepb.Failure{ + Message: "user rejected", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{ + Type: "nexus.OperationError.failed", + }, + }, + } + + var oe *nexus.OperationError + require.ErrorAs(t, nexusErrorFromFailure(f), &oe) + assert.Equal(t, nexus.OperationStateFailed, oe.State) + assert.Equal(t, "user rejected", oe.Message) +} + +func TestNexusErrorFromFailure_ApplicationFailureNamedOperationErrorIsFailedState(t *testing.T) { + f := &failurepb.Failure{ + Message: "user rejected", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{ + Type: nexusOperationErrorType, + NonRetryable: true, + }, + }, + } + + var oe *nexus.OperationError + require.ErrorAs(t, nexusErrorFromFailure(f), &oe) + assert.Equal(t, nexus.OperationStateFailed, oe.State) + assert.Equal(t, "user rejected", oe.Message) +} + +func TestNexusErrorFromFailure_CanceledFailureIsCanceledState(t *testing.T) { + f := &failurepb.Failure{ + Message: "user canceled", + FailureInfo: &failurepb.Failure_CanceledFailureInfo{ + CanceledFailureInfo: &failurepb.CanceledFailureInfo{}, + }, + } + + var oe *nexus.OperationError + require.ErrorAs(t, nexusErrorFromFailure(f), &oe) + assert.Equal(t, nexus.OperationStateCanceled, oe.State) + assert.Equal(t, "user canceled", oe.Message) +} + +func TestNexusErrorFromFailure_OperationErrorCanceled(t *testing.T) { + f := &failurepb.Failure{ + Message: "user canceled", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{ + Type: "nexus.OperationError.canceled", + }, + }, + } + + var oe *nexus.OperationError + require.ErrorAs(t, nexusErrorFromFailure(f), &oe) + assert.Equal(t, nexus.OperationStateCanceled, oe.State) + assert.Equal(t, "user canceled", oe.Message) +} + +func TestNexusErrorFromFailure_OperationErrorUnknownStateFallsBackToFailed(t *testing.T) { + f := &failurepb.Failure{ + Message: "weird", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{ + Type: "nexus.OperationError.weird", + }, + }, + } + + var oe *nexus.OperationError + require.ErrorAs(t, nexusErrorFromFailure(f), &oe) + assert.Equal(t, nexus.OperationStateFailed, oe.State, "unknown state must not leak to the wire") +} + +func TestNexusErrorFromFailure_UntaggedApplicationFailureFallsBackToInternal(t *testing.T) { + f := &failurepb.Failure{ + Message: "boom", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{ + Type: "SomeUserType", + }, + }, + } + + var he *nexus.HandlerError + require.ErrorAs(t, nexusErrorFromFailure(f), &he) + assert.Equal(t, nexus.HandlerErrorTypeInternal, he.Type, "unknown failure shape must collapse to Internal") + assert.Equal(t, "boom", he.Message) +} + +func TestNexusErrorFromFailure_NoFailureInfoIsInternal(t *testing.T) { + f := &failurepb.Failure{Message: "bare failure"} + + var he *nexus.HandlerError + require.ErrorAs(t, nexusErrorFromFailure(f), &he) + assert.Equal(t, nexus.HandlerErrorTypeInternal, he.Type) + assert.Equal(t, "bare failure", he.Message) +} + +// ── RetryBehavior enum mapping ────────────────────────────────────── + +func TestMapNexusRetryBehavior_AllValues(t *testing.T) { + cases := []struct { + in enumspb.NexusHandlerErrorRetryBehavior + want nexus.HandlerErrorRetryBehavior + }{ + {enumspb.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_UNSPECIFIED, nexus.HandlerErrorRetryBehaviorUnspecified}, + {enumspb.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_RETRYABLE, nexus.HandlerErrorRetryBehaviorRetryable}, + {enumspb.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE, nexus.HandlerErrorRetryBehaviorNonRetryable}, + } + + for _, c := range cases { + t.Run(c.in.String(), func(t *testing.T) { + got := mapNexusRetryBehavior(c.in) + assert.Equal(t, c.want, got) + }) + } +} + +// ── Failure-cause preservation (round-trip via failureHolder) ── + +func TestNexusErrorFromFailure_HandlerErrorPreservesCauseProto(t *testing.T) { + f := &failurepb.Failure{ + Message: "boom", + StackTrace: "#0 /app/Handler.php(42): run()\n#1 {main}", + FailureInfo: &failurepb.Failure_NexusHandlerFailureInfo{ + NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{ + Type: "INTERNAL", + RetryBehavior: enumspb.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_RETRYABLE, + }, + }, + } + + var he *nexus.HandlerError + require.ErrorAs(t, nexusErrorFromFailure(f), &he) + assert.Equal(t, "boom", he.Message) + + roundTripped := temporal.GetDefaultFailureConverter().ErrorToFailure(he.Cause) + assert.True(t, proto.Equal(f, roundTripped), + "Cause must hold the original proto verbatim;\nwant: %v\ngot: %v", f, roundTripped) +} + +func TestNexusErrorFromFailure_HandlerErrorPreservesNestedCauseProto(t *testing.T) { + inner := &failurepb.Failure{ + Message: "db connection failed", + StackTrace: "at Db->connect()", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{Type: "PDOException"}, + }, + } + outer := &failurepb.Failure{ + Message: "handler failed", + StackTrace: "at EchoService->echo()", + Cause: inner, + FailureInfo: &failurepb.Failure_NexusHandlerFailureInfo{ + NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{Type: "INTERNAL"}, + }, + } + + var he *nexus.HandlerError + require.ErrorAs(t, nexusErrorFromFailure(outer), &he) + roundTripped := temporal.GetDefaultFailureConverter().ErrorToFailure(he.Cause) + assert.True(t, proto.Equal(outer, roundTripped), + "recursive cause chain must survive round-trip;\nwant: %v\ngot: %v", outer, roundTripped) +} + +func TestNexusErrorFromFailure_OperationErrorPreservesCauseProto(t *testing.T) { + innerDetails := &commonpb.Payloads{ + Payloads: []*commonpb.Payload{{ + Metadata: map[string][]byte{"encoding": []byte("json/plain")}, + Data: []byte(`"detail-payload-marker"`), + }}, + } + innerCause := &failurepb.Failure{ + Message: "inner-business-message", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{ + Type: "CustomBusinessType", + Details: innerDetails, + }, + }, + } + outer := &failurepb.Failure{ + Message: "outer-business-error", + StackTrace: "at OrderService->process()", + Cause: innerCause, + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{ + Type: nexusOperationErrorTypePrefix + "failed", + NonRetryable: true, + }, + }, + } + + var oe *nexus.OperationError + require.ErrorAs(t, nexusErrorFromFailure(outer), &oe) + assert.Equal(t, nexus.OperationStateFailed, oe.State) + assert.Equal(t, "outer-business-error", oe.Message) + + roundTripped := temporal.GetDefaultFailureConverter().ErrorToFailure(oe.Cause) + assert.True(t, proto.Equal(outer, roundTripped), + "OperationError.Cause must round-trip with full structure (type, details, recursive cause);\nwant: %v\ngot: %v", + outer, roundTripped) +} + +// ── nexusLinksFromInternal tests ─────────────────────────────── + +func TestNexusLinksFromInternal_EmptyInputReturnsNil(t *testing.T) { + assert.Nil(t, nexusLinksFromInternal(nil, zap.NewNop())) + assert.Nil(t, nexusLinksFromInternal([]internal.NexusLink{}, zap.NewNop())) +} + +func TestNexusLinksFromInternal_DropsEntriesWithEmptyFields(t *testing.T) { + in := []internal.NexusLink{ + {URL: "", Type: "t"}, + {URL: "http://a/", Type: ""}, + {URL: "http://b/", Type: "t"}, + } + out := nexusLinksFromInternal(in, zap.NewNop()) + require.Len(t, out, 1) + assert.Equal(t, "http://b/", out[0].URL.String()) + assert.Equal(t, "t", out[0].Type) +} + +func TestNexusLinksFromInternal_DropsUnparseableURLs(t *testing.T) { + in := []internal.NexusLink{ + {URL: "http://[::bad", Type: "t"}, + {URL: "http://ok/", Type: "t"}, + } + out := nexusLinksFromInternal(in, zap.NewNop()) + require.Len(t, out, 1) + assert.Equal(t, "http://ok/", out[0].URL.String()) +} + +func TestNexusLinksFromInternal_PreservesOrderingAndFields(t *testing.T) { + in := []internal.NexusLink{ + {URL: "http://a/", Type: "x.one"}, + {URL: "http://b/", Type: "x.two"}, + } + out := nexusLinksFromInternal(in, zap.NewNop()) + require.Len(t, out, 2) + assert.Equal(t, "http://a/", out[0].URL.String()) + assert.Equal(t, "x.one", out[0].Type) + assert.Equal(t, "http://b/", out[1].URL.String()) + assert.Equal(t, "x.two", out[1].Type) +} + +// ── decodeStartReply tests ───────────────────────────────────── + +// Sync reply: Command=*NexusOperationStarted{Async:false}, Payloads carries +// the result. Decoder must wrap it as HandlerStartOperationResultSync with +// the payload preserved on the RawValue. +func TestDecodeStartReply_SyncSuccessUnwrapsPayload(t *testing.T) { + handler := NewNexusHandler(&mockCodec{}, nil, zap.NewNop(), "default") + + resultPayload := &commonpb.Payload{ + Data: []byte(`{"ok":true}`), + Metadata: map[string][]byte{"encoding": []byte("json/plain")}, + } + msg := &internal.Message{ + Command: &internal.NexusOperationStarted{ + Async: false, + Links: []internal.NexusLink{{URL: "http://x/y", Type: "t"}}, + }, + Payloads: &commonpb.Payloads{Payloads: []*commonpb.Payload{resultPayload}}, + } + + res, err := handler.decodeStartReply(context.Background(), msg) + require.NoError(t, err) + sync, ok := res.(*nexus.HandlerStartOperationResultSync[converter.RawValue]) + require.True(t, ok, "expected HandlerStartOperationResultSync, got %T", res) + assert.Equal(t, resultPayload, sync.Value.Payload()) +} + +// Sync reply with empty Payloads slice: still returns a sync result with +// nil Value — matches the pre-refactor empty-payload contract. +func TestDecodeStartReply_SyncSuccessEmptyPayloads(t *testing.T) { + handler := NewNexusHandler(&mockCodec{}, nil, zap.NewNop(), "default") + + msg := &internal.Message{ + Command: &internal.NexusOperationStarted{Async: false}, + Payloads: &commonpb.Payloads{}, + } + + res, err := handler.decodeStartReply(context.Background(), msg) + require.NoError(t, err) + sync, ok := res.(*nexus.HandlerStartOperationResultSync[converter.RawValue]) + require.True(t, ok) + assert.Nil(t, sync.Value.Payload()) +} + +// Async reply: Command=*NexusOperationStarted{Async:true, Token}, no Payloads. +// Decoder returns HandlerStartOperationResultAsync with the token preserved. +func TestDecodeStartReply_AsyncSuccessReturnsToken(t *testing.T) { + handler := NewNexusHandler(&mockCodec{}, nil, zap.NewNop(), "default") + + msg := &internal.Message{ + Command: &internal.NexusOperationStarted{ + Async: true, + Token: "tok-1", + Links: []internal.NexusLink{{URL: "http://x/y", Type: "t"}}, + }, + } + + res, err := handler.decodeStartReply(context.Background(), msg) + require.NoError(t, err) + async, ok := res.(*nexus.HandlerStartOperationResultAsync) + require.True(t, ok, "expected HandlerStartOperationResultAsync, got %T", res) + assert.Equal(t, "tok-1", async.OperationToken) +} + +// nil Command + Failure set → existing failurepb mapping path. The mapping +// itself is exercised in ── Failure → Nexus error mapping ── above; here we +// verify routing. +func TestDecodeStartReply_NilCommandWithFailureRoutesToMapping(t *testing.T) { + handler := NewNexusHandler(&mockCodec{}, nil, zap.NewNop(), "default") + + msg := &internal.Message{ + Failure: &failurepb.Failure{ + Message: "boom", + FailureInfo: &failurepb.Failure_NexusHandlerFailureInfo{ + NexusHandlerFailureInfo: &failurepb.NexusHandlerFailureInfo{Type: "INTERNAL"}, + }, + }, + } + + res, err := handler.decodeStartReply(context.Background(), msg) + assert.Nil(t, res) + require.Error(t, err) + var he *nexus.HandlerError + require.ErrorAs(t, err, &he, "expected *nexus.HandlerError, got %T", err) + assert.Equal(t, "boom", he.Message) +} + +// nil Command + nil Failure: malformed reply → HandlerError(Internal). +func TestDecodeStartReply_EmptyReplyIsHandlerError(t *testing.T) { + handler := NewNexusHandler(&mockCodec{}, nil, zap.NewNop(), "default") + + res, err := handler.decodeStartReply(context.Background(), &internal.Message{}) + assert.Nil(t, res) + require.Error(t, err) + var he *nexus.HandlerError + require.ErrorAs(t, err, &he, "expected *nexus.HandlerError, got %T", err) + assert.Equal(t, nexus.HandlerErrorTypeInternal, he.Type) + assert.Contains(t, he.Message, "neither command nor failure") +} + +// Unknown reply command → HandlerError(Internal). Defends against PHP +// emitting a command name Go doesn't recognize. +func TestDecodeStartReply_UnknownCommandIsHandlerError(t *testing.T) { + handler := NewNexusHandler(&mockCodec{}, nil, zap.NewNop(), "default") + + msg := &internal.Message{Command: &internal.CancelNexusOperation{}} + + res, err := handler.decodeStartReply(context.Background(), msg) + assert.Nil(t, res) + require.Error(t, err) + var he *nexus.HandlerError + require.ErrorAs(t, err, &he) + assert.Equal(t, nexus.HandlerErrorTypeInternal, he.Type) + assert.Contains(t, he.Message, "unexpected") +} diff --git a/aggregatedpool/timer_test.go b/aggregatedpool/timer_test.go new file mode 100644 index 00000000..2dc73578 --- /dev/null +++ b/aggregatedpool/timer_test.go @@ -0,0 +1,117 @@ +package aggregatedpool + +import ( + "errors" + "sync" + "testing" + "time" + + "github.com/roadrunner-server/pool/payload" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/temporalio/roadrunner-temporal/v5/canceller" + "github.com/temporalio/roadrunner-temporal/v5/internal" + "github.com/temporalio/roadrunner-temporal/v5/queue" + "github.com/temporalio/roadrunner-temporal/v5/registry" + bindings "go.temporal.io/sdk/internalbindings" + "go.temporal.io/sdk/workflow" + "go.uber.org/zap" +) + +// timerEnv reproduces the sdk-go NewTimer contract: a non-positive duration is +// resolved inside the call (callback invoked inline, nil TimerID returned). +// Everything else is only what getContext/handleMessage touch. +type timerEnv struct { + bindings.WorkflowEnvironment + info workflow.Info +} + +func (e *timerEnv) WorkflowInfo() *workflow.Info { return &e.info } +func (e *timerEnv) Now() time.Time { return time.Unix(0, 0) } +func (e *timerEnv) IsReplaying() bool { return false } + +func (e *timerEnv) NewTimer(d time.Duration, _ workflow.TimerOptions, callback bindings.ResultHandler) *bindings.TimerID { + if d <= 0 { + callback(nil, nil) + return nil + } + + id := bindings.TimerID{} + + return &id +} + +// capturingCodec captures what a flush would send to PHP and stops flushQueue +// before it reaches the pool. +type capturingCodec struct { + err error + encoded []*internal.Message + encCalls int +} + +func (c *capturingCodec) Encode(_ *internal.Context, _ *payload.Payload, msgs ...*internal.Message) error { + c.encCalls++ + c.encoded = append(c.encoded, msgs...) + return c.err +} + +func (c *capturingCodec) Decode(_ *payload.Payload, _ *[]*internal.Message) error { return nil } + +func (c *capturingCodec) DecodeWorkerInfo(_ *payload.Payload, _ *[]*internal.WorkerInfo) error { + return nil +} + +// A zero-duration timer is resolved by the SDK inside NewTimer, so its response +// is queued with no command of its own to carry a flush. Without the drain-loop +// flush nothing is ever sent and PHP hangs until the workflow task times out. +func TestDrainPipeline_ZeroDurationTimerResponseIsFlushed(t *testing.T) { + stop := errors.New("stop before the pool") + codec := &capturingCodec{err: stop} + + wp := &Workflow{ + log: zap.NewNop(), + env: &timerEnv{}, + codec: codec, + pool: &recordingPool{}, + mq: queue.NewMessageQueue(func() uint64 { return 0 }), + canceller: new(canceller.Canceller), + nexusStarted: new(registry.NexusStartedRegistry), + pldPool: &sync.Pool{New: func() any { return new(payload.Payload) }}, + pipeline: []*internal.Message{{ + ID: 42, + Command: &internal.NewTimer{Milliseconds: 0}, + }}, + } + wp.inLoop = 1 + + err := wp.drainPipeline() + + require.ErrorIs(t, err, stop, "the queued response must be flushed after the batch") + require.Len(t, codec.encoded, 1) + assert.Equal(t, uint64(42), codec.encoded[0].ID) +} + +// A real timer keeps its canceller slot and queues nothing synchronously. +func TestDrainPipeline_PositiveDurationTimerQueuesNothing(t *testing.T) { + codec := &capturingCodec{} + + wp := &Workflow{ + log: zap.NewNop(), + env: &timerEnv{}, + codec: codec, + pool: &recordingPool{}, + mq: queue.NewMessageQueue(func() uint64 { return 0 }), + canceller: new(canceller.Canceller), + nexusStarted: new(registry.NexusStartedRegistry), + pldPool: &sync.Pool{New: func() any { return new(payload.Payload) }}, + pipeline: []*internal.Message{{ + ID: 43, + Command: &internal.NewTimer{Milliseconds: 100}, + }}, + } + wp.inLoop = 1 + + require.NoError(t, wp.drainPipeline()) + assert.Zero(t, codec.encCalls, "a scheduled timer must not trigger a flush") + assert.Empty(t, wp.mq.Messages()) +} diff --git a/aggregatedpool/workers.go b/aggregatedpool/workers.go index be78333c..719a2068 100644 --- a/aggregatedpool/workers.go +++ b/aggregatedpool/workers.go @@ -116,7 +116,23 @@ func registerWorkflow(register func(), name, taskQueue string) (err error) { return nil } -func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo, log *zap.Logger, tc temporalClient.Client, interceptors map[string]api.Interceptor, configuredInterceptors []string) ([]worker.Worker, error) { +// registerNexusService converts a panic from the SDK's Nexus registration (invalid +// or duplicate names supplied by the PHP worker at runtime) into a clean init error. +func registerNexusService(register func(), service, taskQueue string) (err error) { + defer func() { + r := recover() + if r == nil { + return + } + + err = errors.E(errors.Op("temporal_register_nexus_service"), errors.Errorf("failed to register nexus service %q on task queue %q: %v", service, taskQueue, r)) + }() + + register() + return nil +} + +func TemporalWorkers(wDef *Workflow, actDef *Activity, nexusHandler *NexusHandler, wi []*internal.WorkerInfo, log *zap.Logger, tc temporalClient.Client, interceptors map[string]api.Interceptor, configuredInterceptors []string) ([]worker.Worker, error) { resolved, err := ResolveInterceptors(interceptors, configuredInterceptors) if err != nil { return nil, err @@ -205,6 +221,22 @@ func TemporalWorkers(wDef *Workflow, actDef *Activity, wi []*internal.WorkerInfo log.Debug("activity registered", zap.String(tq, workerInfo.TaskQueue), zap.Any("workflow name", activity.Name)) } + if nexusHandler != nil && len(wi[i].NexusServices) > 0 { + // Cooperative method-cancel is always wired: every PHP-SDK that + // ships Nexus services also handles CancelNexusOperationMethod + // (both arrived in the same release). + for _, ns := range wi[i].NexusServices { + err := registerNexusService(func() { + wrk.RegisterNexusService(nexusHandler.CreateNexusService(wi[i].TaskQueue, ns.Name, ns.Operations)) + }, ns.Name, wi[i].TaskQueue) + if err != nil { + return nil, err + } + + log.Debug("nexus service registered", zap.String(tq, wi[i].TaskQueue), zap.String("service", ns.Name), zap.Strings("ops", ns.Operations)) + } + } + // add worker to the pool workers = append(workers, wrk) } diff --git a/aggregatedpool/workers_test.go b/aggregatedpool/workers_test.go index b46d45a7..effa878f 100644 --- a/aggregatedpool/workers_test.go +++ b/aggregatedpool/workers_test.go @@ -281,6 +281,24 @@ func TestRegisterWorkflow_NonStringPanic_Handled(t *testing.T) { assert.Contains(t, err.Error(), "42", "should preserve a non-string panic value") } +func TestRegisterNexusService_NoPanic_OK(t *testing.T) { + require.NoError(t, registerNexusService(func() {}, "billing", "my-task-queue")) +} + +func TestRegisterNexusService_DuplicateOperation_ReturnsError(t *testing.T) { + h := &NexusHandler{} + + // CreateNexusService -> svc.MustRegister panics on a duplicate operation name; + // the guard must turn that runtime (PHP-supplied) panic into a clean error. + err := registerNexusService(func() { + h.CreateNexusService("my-task-queue", "billing", []string{"charge", "charge"}) + }, "billing", "my-task-queue") + + require.Error(t, err) + assert.Contains(t, err.Error(), "billing", "should name the offending service") + assert.Contains(t, err.Error(), "my-task-queue", "should name the task queue") +} + func TestTemporalWorkers_MultipleDynamicWorkflows_ReturnsError(t *testing.T) { temporalClient, err := client.NewLazyClient(client.Options{}) require.NoError(t, err) @@ -294,7 +312,7 @@ func TestTemporalWorkers_MultipleDynamicWorkflows_ReturnsError(t *testing.T) { }, }} - _, err = TemporalWorkers(nil, nil, workers, zap.NewNop(), temporalClient, nil, nil) + _, err = TemporalWorkers(nil, nil, nil, workers, zap.NewNop(), temporalClient, nil, nil) require.Error(t, err) assert.Contains(t, err.Error(), "multiple dynamic workflows") assert.Contains(t, err.Error(), "default") diff --git a/aggregatedpool/workflow.go b/aggregatedpool/workflow.go index 3d43d8bd..6e808cc3 100644 --- a/aggregatedpool/workflow.go +++ b/aggregatedpool/workflow.go @@ -62,11 +62,17 @@ type Workflow struct { callbacks []Callback canceller *canceller.Canceller inLoop uint32 + // set when a response was queued from a callback that fired inline during + // dispatch; such a response is flushed once the current batch is drained. + pendingFlush bool // updates updateCompleteCb map[string]func(res *internal.Message) updateValidateCb map[string]func(res *internal.Message) + // caller-side: ExecuteNexusOperation message ID → (token, err) start ack. + nexusStarted *registry.NexusStartedRegistry + log *zap.Logger mh temporalClient.MetricsHandler @@ -102,6 +108,7 @@ func (wp *Workflow) NewWorkflowDefinition() bindings.WorkflowDefinition { updateCompleteCb: make(map[string]func(res *internal.Message)), updateValidateCb: make(map[string]func(res *internal.Message)), updatesQueue: map[string]struct{}{}, + nexusStarted: new(registry.NexusStartedRegistry), // -- updates pool: wp.pool, codec: wp.codec, @@ -313,22 +320,59 @@ func (wp *Workflow) OnWorkflowTaskStarted(t time.Duration) { panic(err) } - for len(wp.pipeline) > 0 { - msg := wp.pipeline[0] - wp.pipeline = wp.pipeline[1:] + err = wp.drainPipeline() + if err != nil { + wp.pipeline = nil + panic(err) + } +} + +// deliverInline reports whether a callback may push its response right now +// instead of deferring it to the next workflow task. A response pushed inline +// during dispatch has no flush of its own, so it is marked as pending here and +// flushed by drainPipeline once the whole batch is handled. +func (wp *Workflow) deliverInline() bool { + if atomic.LoadUint32(&wp.inLoop) != 1 { + return false + } + + wp.pendingFlush = true + + return true +} + +// drainPipeline handles every queued message, then flushes any response that a +// callback queued inline during that dispatch, which in turn may bring new +// messages to handle. +func (wp *Workflow) drainPipeline() error { + for { + for len(wp.pipeline) > 0 { + msg := wp.pipeline[0] + wp.pipeline = wp.pipeline[1:] + + if !msg.IsCommand() { + continue + } - if msg.IsCommand() { if msg.UndefinedResponse() { - wp.pipeline = nil - panic(fmt.Sprintf("undefined response: %s", msg.Command.(*internal.UndefinedResponse).Message)) + return fmt.Errorf("undefined response: %s", msg.Command.(*internal.UndefinedResponse).Message) } - err = wp.handleMessage(msg) + err := wp.handleMessage(msg) + if err != nil { + return err + } } + if !wp.pendingFlush { + return nil + } + + wp.pendingFlush = false + + err := wp.flushQueue() if err != nil { - wp.pipeline = nil - panic(err) + return err } } } diff --git a/go.mod b/go.mod index 081ac4fe..09faa168 100644 --- a/go.mod +++ b/go.mod @@ -32,7 +32,7 @@ require ( github.com/golang/mock v1.7.0-rc.1 // indirect github.com/grpc-ecosystem/grpc-gateway/v2 v2.30.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect - github.com/nexus-rpc/sdk-go v0.6.0 // indirect + github.com/nexus-rpc/sdk-go v0.6.0 github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/prometheus/client_golang v1.24.1 @@ -60,4 +60,10 @@ require ( gopkg.in/yaml.v3 v3.0.1 // indirect ) -require github.com/nexus-rpc/nexus-proto-annotations v0.1.0 // indirect +require ( + github.com/kr/pretty v0.3.1 // indirect + github.com/nexus-rpc/nexus-proto-annotations v0.1.0 // indirect + github.com/rogpeppe/go-internal v1.14.1 // indirect + go.opentelemetry.io/otel/sdk/metric v1.45.0 // indirect + gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect +) diff --git a/go.sum b/go.sum index 0df71d44..bcda05cf 100644 --- a/go.sum +++ b/go.sum @@ -119,6 +119,8 @@ github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxv github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= @@ -138,10 +140,10 @@ github.com/nexus-rpc/nexus-proto-annotations v0.1.0 h1:2fELd+9sqUtNu6Fg//pw8YFsx github.com/nexus-rpc/nexus-proto-annotations v0.1.0/go.mod h1:n3UjF1bPCW8llR8tHvbxJ+27yPWrhpo8w/Yg1IOuY0Y= github.com/nexus-rpc/sdk-go v0.6.0 h1:QRgnP2zTbxEbiyWG/aXH8uSC5LV/Mg1fqb19jb4DBlo= github.com/nexus-rpc/sdk-go v0.6.0/go.mod h1:FHdPfVQwRuJFZFTF0Y2GOAxCrbIBNrcPna9slkGKPYk= -github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWbfPhv4DMiApHyliiK5xCTNVSPiaAs= github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno= github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o= github.com/pborman/uuid v1.2.1/go.mod h1:X/NO0urCmaxf9VXbdlT7C2Yzkj2IKimNn4k+gtPdI/k= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= @@ -187,6 +189,8 @@ github.com/roadrunner-server/pool v1.1.3/go.mod h1:8ceC7NvZKJRciv+KJmcyk5CeDugoe github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ= github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k= github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI= github.com/shirou/gopsutil v3.21.11+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA= github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo= @@ -232,8 +236,7 @@ go.opentelemetry.io/otel/metric v1.45.0 h1:7Eg1uH7CJ5cXv9is6tnBe1FI6rj1nwUdbFypR go.opentelemetry.io/otel/metric v1.45.0/go.mod h1:HAPbm1nd3p1PmFH7v2dR+6BjXxw+Lq4a2+pndMAm08s= go.opentelemetry.io/otel/sdk v1.45.0 h1:4VVSMgQ83dUgW2aoX5f6JgLvHwIvzcuLnF9lUdCSpCw= go.opentelemetry.io/otel/sdk v1.45.0/go.mod h1:Sr40LgXV7DsKMMJMKOhUWOgMWTfAaqvm2kF0g7ilwuA= -go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= -go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/sdk/metric v1.45.0 h1:oVFszMfyj1Am6s24Vtc7wBb8BKLcwepJjNEYILuiE3o= go.opentelemetry.io/otel/trace v1.45.0 h1:l/mP6Uv7oNO7/TblbhpbgMidxhq1uO/rPsikOyVhxag= go.opentelemetry.io/otel/trace v1.45.0/go.mod h1:qoJJA2xNMnxRrdISU/kLtfUH2wNeQbiv+jhs/CxI8bc= go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI= @@ -404,8 +407,8 @@ google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f h1:BLraFXnmrev5lT+xlilqcH8XK9/i0At2xKjWk4p6zsU= gopkg.in/check.v1 v1.0.0-20200227125254-8fa46927fb4f/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/validator.v2 v2.0.0-20200605151824-2b28d334fa05/go.mod h1:o4V0GXN9/CAmCsvJ0oXYZvrZOe7syiDZSN1GWGZTGzc= gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= diff --git a/info.go b/info.go index e4d5592c..0d341d7d 100644 --- a/info.go +++ b/info.go @@ -82,3 +82,16 @@ func ActivitiesInfo(wi []*internal.WorkerInfo) map[string]*internal.ActivityInfo return activitiesInfo } + +func NexusServicesInfo(wi []*internal.WorkerInfo) map[string]*internal.NexusServiceInfo { + nexusInfo := make(map[string]*internal.NexusServiceInfo) + + for i := range wi { + for j := range wi[i].NexusServices { + ns := &wi[i].NexusServices[j] + nexusInfo[ns.Name] = ns + } + } + + return nexusInfo +} diff --git a/internal.go b/internal.go index 353b0611..7239147b 100644 --- a/internal.go +++ b/internal.go @@ -115,7 +115,7 @@ func (p *Plugin) initPool() error { return err } - workers, err := aggregatedpool.TemporalWorkers(wfDef, actDef, wi, p.log, p.temporal.client, p.temporal.interceptors, p.config.Interceptors) + workers, err := aggregatedpool.TemporalWorkers(wfDef, actDef, aggregatedpool.NewNexusHandler(codec, ap, p.log, p.config.Namespace), wi, p.log, p.temporal.client, p.temporal.interceptors, p.config.Interceptors) if err != nil { return err } @@ -134,6 +134,7 @@ func (p *Plugin) initPool() error { p.temporal.activities = ActivitiesInfo(wi) p.temporal.workflows = WorkflowsInfo(wi) + p.temporal.nexusServices = NexusServicesInfo(wi) p.actP = ap p.wfP = wp diff --git a/internal/protocol.go b/internal/protocol.go index 4c90688c..49d4920e 100644 --- a/internal/protocol.go +++ b/internal/protocol.go @@ -49,6 +49,18 @@ const ( cancelCommand = "Cancel" panicCommand = "Panic" + + // Nexus commands: Go → PHP (handler side) + invokeNexusOperationCommand = "InvokeNexusOperation" + cancelNexusOperationCommand = "CancelNexusOperation" + cancelNexusOperationMethodCommand = "CancelNexusOperationMethod" + + // Nexus commands: PHP → Go (caller side from workflow) + executeNexusOperationCommand = "ExecuteNexusOperation" + getNexusOperationStartedCommand = "GetNexusOperationStarted" + + // Nexus reply commands: PHP → Go (handler-side success reply) + nexusOperationStartedCommand = "NexusOperationStarted" ) type TypedSearchAttributeType string @@ -358,6 +370,106 @@ type Panic struct { Message string `json:"message"` } +// NexusLink is the JSON wire form of nexus.Link. +type NexusLink struct { + URL string `json:"url"` + Type string `json:"type"` +} + +// InvokeNexusOperation: Go → PHP, handler-side Nexus task dispatch. +type InvokeNexusOperation struct { + Service string `json:"service"` + Operation string `json:"operation"` + Namespace string `json:"namespace,omitempty"` + TaskQueue string `json:"taskQueue,omitempty"` + RequestID string `json:"requestId"` + Callback string `json:"callback,omitempty"` + CallbackHeaders map[string]string `json:"callbackHeaders,omitempty"` + Headers map[string]string `json:"headers,omitempty"` + Links []NexusLink `json:"links,omitempty"` + // InvocationID correlates with CancelNexusOperationMethod. + InvocationID uint64 `json:"invocationId"` +} + +// CancelNexusOperationMethod cooperatively stops an in-flight handler method +// (distinct from CancelNexusOperation, which targets the business operation). +type CancelNexusOperationMethod struct { + InvocationID uint64 `json:"invocationId"` // matches InvokeNexusOperation.InvocationID + Reason string `json:"reason,omitempty"` +} + +// CancelNexusOperation: Go → PHP, cancel an async business operation by token. +type CancelNexusOperation struct { + Service string `json:"service"` + Operation string `json:"operation"` + Namespace string `json:"namespace,omitempty"` + TaskQueue string `json:"taskQueue,omitempty"` + OperationToken string `json:"operationToken"` + // Raw HTTP-style headers from the caller's cancel request, propagated to the + // handler's OperationContext. Symmetric with ExecuteNexusOperation.NexusHeaders. + Headers map[string]string `json:"headers,omitempty"` +} + +// NexusOperationStarted: PHP→Go reply to InvokeNexusOperation success. +// Async=false → sync (payload in Message.Payloads); Async=true → Token holds the operation token. +type NexusOperationStarted struct { + Async bool `json:"async"` + Token string `json:"token,omitempty"` + Links []NexusLink `json:"links,omitempty"` +} + +// NexusOperationOptions is PHP's "options" DTO. Endpoint/service also appear here +// but are ignored — the top-level ExecuteNexusOperation fields are authoritative. +type NexusOperationOptions struct { + // nanoseconds (PHP DateIntervalType default; matches Go time.Duration encoding). + ScheduleToCloseTimeout time.Duration `json:"scheduleToCloseTimeout,omitempty"` + // Maximum time to wait for the operation to be started by the handler. Requires Temporal Server 1.31.0+. + ScheduleToStartTimeout time.Duration `json:"scheduleToStartTimeout,omitempty"` + // Maximum time an async operation may take to complete after starting. Requires Temporal Server 1.31.0+. + StartToCloseTimeout time.Duration `json:"startToCloseTimeout,omitempty"` + // 0=Unspecified, 1=Abandon, 2=TryCancel, 3=WaitRequested, 4=WaitCompleted. + CancellationType int `json:"cancellationType,omitempty"` + // Single-line summary; the SDK carries it as command UserMetadata. + Summary string `json:"summary,omitempty"` +} + +// GetNexusOperationStarted: PHP → Go, listen-and-wait for the start ack of a +// caller-side Nexus op by its original ExecuteNexusOperation message ID. +type GetNexusOperationStarted struct { + ID uint64 `json:"id"` +} + +// ExecuteNexusOperation: PHP → Go, workflow calling a Nexus operation. +type ExecuteNexusOperation struct { + Endpoint string `json:"endpoint"` + Service string `json:"service"` + Operation string `json:"operation"` + Options NexusOperationOptions `json:"options,omitempty"` + // Raw HTTP-style headers propagated to handler's OperationContext. + // Separate from the Temporal interceptor `Header` (typed payloads). + NexusHeaders map[string]string `json:"nexusHeaders,omitempty"` +} + +// NexusOperationParams builds ExecuteNexusOperationParams (single-payload by spec). +// The Header arg is unused (Nexus headers travel on NexusHeaders); kept for symmetry. +func (cmd ExecuteNexusOperation) NexusOperationParams(payloads *commonpb.Payloads, _ *commonpb.Header) bindings.ExecuteNexusOperationParams { + var input *commonpb.Payload + if pls := payloads.GetPayloads(); len(pls) > 0 { + input = pls[0] + } + + client := bindings.NewNexusClient(cmd.Endpoint, cmd.Service) + options := workflow.NexusOperationOptions{ + ScheduleToCloseTimeout: cmd.Options.ScheduleToCloseTimeout, + ScheduleToStartTimeout: cmd.Options.ScheduleToStartTimeout, + StartToCloseTimeout: cmd.Options.StartToCloseTimeout, + CancellationType: workflow.NexusOperationCancellationType(cmd.Options.CancellationType), + Summary: cmd.Options.Summary, + } + + return bindings.NewExecuteNexusOperationParams(client, cmd.Operation, input, options, cmd.NexusHeaders) +} + // ActivityParams maps activity command to activity params. func (cmd ExecuteActivity) ActivityParams(env bindings.WorkflowEnvironment, payloads *commonpb.Payloads, header *commonpb.Header) bindings.ExecuteActivityParams { params := bindings.ExecuteActivityParams{ @@ -504,6 +616,18 @@ func CommandName(cmd any) (string, error) { return upsertMemo, nil case InvokeUpdate, *InvokeUpdate: return invokeUpdateCommand, nil + case InvokeNexusOperation, *InvokeNexusOperation: + return invokeNexusOperationCommand, nil + case CancelNexusOperation, *CancelNexusOperation: + return cancelNexusOperationCommand, nil + case CancelNexusOperationMethod, *CancelNexusOperationMethod: + return cancelNexusOperationMethodCommand, nil + case ExecuteNexusOperation, *ExecuteNexusOperation: + return executeNexusOperationCommand, nil + case GetNexusOperationStarted, *GetNexusOperationStarted: + return getNexusOperationStartedCommand, nil + case NexusOperationStarted, *NexusOperationStarted: + return nexusOperationStartedCommand, nil default: return "", errors.E(op, errors.Errorf("undefined command type: %s", cmd)) } @@ -597,6 +721,24 @@ func InitCommand(name string) (any, error) { case invokeUpdateCommand: return &InvokeUpdate{}, nil + case invokeNexusOperationCommand: + return &InvokeNexusOperation{}, nil + + case cancelNexusOperationCommand: + return &CancelNexusOperation{}, nil + + case cancelNexusOperationMethodCommand: + return &CancelNexusOperationMethod{}, nil + + case executeNexusOperationCommand: + return &ExecuteNexusOperation{}, nil + + case getNexusOperationStartedCommand: + return &GetNexusOperationStarted{}, nil + + case nexusOperationStartedCommand: + return &NexusOperationStarted{}, nil + default: return nil, errors.E(op, errors.Errorf("undefined command name: %s, possible outdated RoadRunner version", name)) } diff --git a/internal/protocol_test.go b/internal/protocol_test.go index 8b475970..69c36b0a 100644 --- a/internal/protocol_test.go +++ b/internal/protocol_test.go @@ -1,9 +1,12 @@ package internal import ( + "encoding/json" "errors" "testing" + "time" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" commonpb "go.temporal.io/api/common/v1" "go.temporal.io/sdk/converter" @@ -42,3 +45,348 @@ func TestLocalActivityParams_FailureConverterDoesNotPanic(t *testing.T) { _ = params.FailureConverter.ErrorToFailure(errors.New("boom")) }) } +func TestCommandName_Nexus(t *testing.T) { + tests := []struct { + name string + command any + want string + }{ + {"InvokeNexusOperation value", InvokeNexusOperation{}, "InvokeNexusOperation"}, + {"InvokeNexusOperation ptr", &InvokeNexusOperation{}, "InvokeNexusOperation"}, + {"CancelNexusOperation value", CancelNexusOperation{}, "CancelNexusOperation"}, + {"CancelNexusOperation ptr", &CancelNexusOperation{}, "CancelNexusOperation"}, + {"CancelNexusOperationMethod value", CancelNexusOperationMethod{}, "CancelNexusOperationMethod"}, + {"CancelNexusOperationMethod ptr", &CancelNexusOperationMethod{}, "CancelNexusOperationMethod"}, + {"ExecuteNexusOperation value", ExecuteNexusOperation{}, "ExecuteNexusOperation"}, + {"ExecuteNexusOperation ptr", &ExecuteNexusOperation{}, "ExecuteNexusOperation"}, + {"GetNexusOperationStarted value", GetNexusOperationStarted{}, "GetNexusOperationStarted"}, + {"GetNexusOperationStarted ptr", &GetNexusOperationStarted{}, "GetNexusOperationStarted"}, + {"NexusOperationStarted value", NexusOperationStarted{}, "NexusOperationStarted"}, + {"NexusOperationStarted ptr", &NexusOperationStarted{}, "NexusOperationStarted"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := CommandName(tt.command) + require.NoError(t, err) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestInitCommand_Nexus(t *testing.T) { + tests := []struct { + name string + typeName string + wantType any + }{ + {"InvokeNexusOperation", "InvokeNexusOperation", &InvokeNexusOperation{}}, + {"CancelNexusOperation", "CancelNexusOperation", &CancelNexusOperation{}}, + {"CancelNexusOperationMethod", "CancelNexusOperationMethod", &CancelNexusOperationMethod{}}, + {"ExecuteNexusOperation", "ExecuteNexusOperation", &ExecuteNexusOperation{}}, + {"GetNexusOperationStarted", "GetNexusOperationStarted", &GetNexusOperationStarted{}}, + {"NexusOperationStarted", "NexusOperationStarted", &NexusOperationStarted{}}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := InitCommand(tt.typeName) + require.NoError(t, err) + assert.IsType(t, tt.wantType, got) + }) + } +} + +// Wire contract: a single `id` = the original ExecuteNexusOperation message ID. +func TestGetNexusOperationStarted_DecodesPHPWireShape(t *testing.T) { + wire := []byte(`{"id":42}`) + var cmd GetNexusOperationStarted + require.NoError(t, json.Unmarshal(wire, &cmd)) + assert.Equal(t, uint64(42), cmd.ID) +} + +func TestGetNexusOperationStarted_RoundTrip(t *testing.T) { + cmd := GetNexusOperationStarted{ID: 77} + data, err := json.Marshal(cmd) + require.NoError(t, err) + assert.JSONEq(t, `{"id":77}`, string(data)) +} + +// Regression guard: the old polling commands must stay unknown ("undefined command"). +func TestInitCommand_RemovedPollingCommands(t *testing.T) { + for _, removed := range []string{"GetNexusOperationResult", "CancelNexusOperationResult"} { + t.Run(removed, func(t *testing.T) { + got, err := InitCommand(removed) + assert.Error(t, err, "removed command %q must not decode anymore", removed) + assert.Nil(t, got) + }) + } +} + +func TestCancelNexusOperationMethod_JSONRoundTrip(t *testing.T) { + // InvocationID must NOT be omitempty — 0 is a valid id on the wire. + op := CancelNexusOperationMethod{ + InvocationID: 7, + Reason: "deadline", + } + data, err := json.Marshal(op) + require.NoError(t, err) + assert.Contains(t, string(data), `"invocationId":7`) + assert.Contains(t, string(data), `"reason":"deadline"`) + + var back CancelNexusOperationMethod + require.NoError(t, json.Unmarshal(data, &back)) + assert.Equal(t, op, back) +} + +// Sync reply: Async=false, Token omitted. +func TestNexusOperationStarted_SyncWireShape(t *testing.T) { + out, err := json.Marshal(NexusOperationStarted{Async: false}) + require.NoError(t, err) + assert.JSONEq(t, `{"async":false}`, string(out)) + + out, err = json.Marshal(NexusOperationStarted{ + Async: false, + Links: []NexusLink{{URL: "http://x/y", Type: "t"}}, + }) + require.NoError(t, err) + assert.JSONEq(t, `{"async":false,"links":[{"url":"http://x/y","type":"t"}]}`, string(out)) +} + +// Async reply: Async=true, Token populated. +func TestNexusOperationStarted_AsyncWireShape(t *testing.T) { + out, err := json.Marshal(NexusOperationStarted{ + Async: true, + Token: "tok-abc", + }) + require.NoError(t, err) + assert.JSONEq(t, `{"async":true,"token":"tok-abc"}`, string(out)) +} + +// Decodes the typed reply that replaced the legacy _rr_nexus_* markers. +func TestNexusOperationStarted_DecodesPHPWireShape(t *testing.T) { + wire := []byte(`{"async":true,"token":"op-1","links":[{"url":"http://a/b","type":"x.y"}]}`) + var reply NexusOperationStarted + require.NoError(t, json.Unmarshal(wire, &reply)) + assert.True(t, reply.Async) + assert.Equal(t, "op-1", reply.Token) + require.Len(t, reply.Links, 1) + assert.Equal(t, "http://a/b", reply.Links[0].URL) + assert.Equal(t, "x.y", reply.Links[0].Type) +} + +// InvocationID is the cooperative-cancel correlation key; always on the wire. +func TestInvokeNexusOperation_InvocationIDAlwaysPresent(t *testing.T) { + zero, err := json.Marshal(InvokeNexusOperation{Service: "S", Operation: "o"}) + require.NoError(t, err) + assert.Contains(t, string(zero), `"invocationId":0`) + + set, err := json.Marshal(InvokeNexusOperation{Service: "S", Operation: "o", InvocationID: 99}) + require.NoError(t, err) + assert.Contains(t, string(set), `"invocationId":99`) +} + +// endpoint/service inside "options" are ignored even when they mismatch the +// top-level fields, which are authoritative. +func TestExecuteNexusOperation_OptionsEndpointServiceIgnored(t *testing.T) { + wire := []byte(`{ + "endpoint": "top-level-endpoint", + "service": "TopLevelService", + "operation": "echo", + "options": { + "endpoint": "WRONG-ENDPOINT", + "service": "WrongService", + "scheduleToCloseTimeout": 5000000000 + } + }`) + + var op ExecuteNexusOperation + require.NoError(t, json.Unmarshal(wire, &op)) + assert.Equal(t, "top-level-endpoint", op.Endpoint, "top-level endpoint must win") + assert.Equal(t, "TopLevelService", op.Service, "top-level service must win") + assert.Equal(t, 5*time.Second, op.Options.ScheduleToCloseTimeout) +} + +// PHP marshals timeouts as nanoseconds (matches Go time.Duration): 10s ⇒ 10e9 ns. +func TestExecuteNexusOperation_DecodesPHPWireShape(t *testing.T) { + wire := []byte(`{ + "endpoint": "my-nexus-endpoint-name", + "service": "SampleNexusService", + "operation": "echo", + "options": { + "endpoint": "my-nexus-endpoint-name", + "service": "SampleNexusService", + "scheduleToCloseTimeout": 10000000000 + } + }`) + + var op ExecuteNexusOperation + require.NoError(t, json.Unmarshal(wire, &op)) + assert.Equal(t, "my-nexus-endpoint-name", op.Endpoint) + assert.Equal(t, "SampleNexusService", op.Service) + assert.Equal(t, "echo", op.Operation) + assert.Equal(t, 10*time.Second, op.Options.ScheduleToCloseTimeout) +} + +// All-zero options must decode with no timeout enforced (no hard-coded defaults). +func TestExecuteNexusOperation_OmitsZeroOptions(t *testing.T) { + wire := []byte(`{"endpoint":"e","service":"s","operation":"o"}`) + var op ExecuteNexusOperation + require.NoError(t, json.Unmarshal(wire, &op)) + assert.Equal(t, time.Duration(0), op.Options.ScheduleToCloseTimeout) +} + +// cancellationType is PHP's enum int; it must round-trip unchanged (cast later). +func TestNexusOperationOptions_DecodesCancellationType(t *testing.T) { + for name, tc := range map[string]struct { + wire string + want int + }{ + "unspecified missing": {`{}`, 0}, + "unspecified explicit zero": {`{"cancellationType":0}`, 0}, + "abandon": {`{"cancellationType":1}`, 1}, + "try-cancel": {`{"cancellationType":2}`, 2}, + "wait-requested": {`{"cancellationType":3}`, 3}, + "wait-completed": {`{"cancellationType":4}`, 4}, + } { + t.Run(name, func(t *testing.T) { + var opts NexusOperationOptions + require.NoError(t, json.Unmarshal([]byte(tc.wire), &opts)) + assert.Equal(t, tc.want, opts.CancellationType) + }) + } +} + +// omitempty: zero cancellationType stays off the wire. +func TestNexusOperationOptions_OmitsZeroCancellationType(t *testing.T) { + out, err := json.Marshal(NexusOperationOptions{ScheduleToCloseTimeout: time.Second}) + require.NoError(t, err) + assert.NotContains(t, string(out), "cancellationType") +} + +// Top-level `nexusHeaders` (x-nexus-* map) decodes and is forwarded verbatim. +func TestExecuteNexusOperation_DecodesNexusHeaders(t *testing.T) { + wire := []byte(`{ + "endpoint": "e", "service": "s", "operation": "o", + "nexusHeaders": { + "x-nexus-caller-workflow-id": "wf-abc", + "x-nexus-trace-id": "trace-1" + } + }`) + + var op ExecuteNexusOperation + require.NoError(t, json.Unmarshal(wire, &op)) + assert.Equal(t, map[string]string{ + "x-nexus-caller-workflow-id": "wf-abc", + "x-nexus-trace-id": "trace-1", + }, op.NexusHeaders) +} + +// Absent/empty nexusHeaders leaves the map nil and off the wire. +func TestExecuteNexusOperation_OmitsEmptyNexusHeaders(t *testing.T) { + wire := []byte(`{"endpoint":"e","service":"s","operation":"o"}`) + var op ExecuteNexusOperation + require.NoError(t, json.Unmarshal(wire, &op)) + assert.Nil(t, op.NexusHeaders) + + out, err := json.Marshal(ExecuteNexusOperation{Endpoint: "e", Service: "s", Operation: "o"}) + require.NoError(t, err) + assert.NotContains(t, string(out), "nexusHeaders") +} + +// TestNexusOperationOptions_DecodesTimeouts pins the nanosecond decode of the two +// Server-1.31 timeout fields (scheduleToClose is covered separately). +func TestNexusOperationOptions_DecodesTimeouts(t *testing.T) { + wire := []byte(`{"scheduleToStartTimeout":3000000000,"startToCloseTimeout":45000000000}`) + var opts NexusOperationOptions + require.NoError(t, json.Unmarshal(wire, &opts)) + assert.Equal(t, 3*time.Second, opts.ScheduleToStartTimeout) + assert.Equal(t, 45*time.Second, opts.StartToCloseTimeout) +} + +// summary decodes as a plain string (forwarded as command UserMetadata). +func TestNexusOperationOptions_DecodesSummary(t *testing.T) { + wire := []byte(`{"scheduleToCloseTimeout":10000000000,"summary":"charge the card"}`) + var opts NexusOperationOptions + require.NoError(t, json.Unmarshal(wire, &opts)) + assert.Equal(t, "charge the card", opts.Summary) +} + +// omitempty: empty summary stays off the wire. +func TestNexusOperationOptions_OmitsEmptySummary(t *testing.T) { + out, err := json.Marshal(NexusOperationOptions{ScheduleToCloseTimeout: time.Second}) + require.NoError(t, err) + assert.NotContains(t, string(out), "summary") +} + +// TestInvokeNexusOperation_MarshalsTaskQueue pins that the Go→PHP envelope +// carries the handler task queue so the PHP OperationContext is complete. +func TestInvokeNexusOperation_MarshalsTaskQueue(t *testing.T) { + out, err := json.Marshal(InvokeNexusOperation{ + Service: "billing", + Operation: "charge", + TaskQueue: "nexus-tq", + }) + require.NoError(t, err) + assert.Contains(t, string(out), `"taskQueue":"nexus-tq"`) + + out, err = json.Marshal(InvokeNexusOperation{Service: "s", Operation: "o"}) + require.NoError(t, err) + assert.NotContains(t, string(out), "taskQueue", "empty task queue must be omitted") +} + +// TestCancelNexusOperation_MarshalsTaskQueue pins the same for the cancel +// envelope — symmetric with the start path. +func TestCancelNexusOperation_MarshalsTaskQueue(t *testing.T) { + out, err := json.Marshal(CancelNexusOperation{ + Service: "billing", + Operation: "charge", + TaskQueue: "nexus-tq", + }) + require.NoError(t, err) + assert.Contains(t, string(out), `"taskQueue":"nexus-tq"`) + + out, err = json.Marshal(CancelNexusOperation{Service: "s", Operation: "o"}) + require.NoError(t, err) + assert.NotContains(t, string(out), "taskQueue", "empty task queue must be omitted") +} + +// Cancel-request headers are forwarded under `headers` (symmetric with start). +func TestCancelNexusOperation_MarshalsHeaders(t *testing.T) { + out, err := json.Marshal(CancelNexusOperation{ + Service: "s", + Operation: "o", + OperationToken: "tok", + Headers: map[string]string{"x-nexus-trace-id": "trace-1"}, + }) + require.NoError(t, err) + assert.JSONEq(t, `{"service":"s","operation":"o","operationToken":"tok","headers":{"x-nexus-trace-id":"trace-1"}}`, string(out)) +} + +func TestCancelNexusOperation_OmitsEmptyHeaders(t *testing.T) { + out, err := json.Marshal(CancelNexusOperation{Service: "s", Operation: "o", OperationToken: "tok"}) + require.NoError(t, err) + assert.NotContains(t, string(out), "headers") +} + +// Namespace marshals when set, omitted when empty (PHP reads $options['namespace']). +func TestInvokeNexusOperation_MarshalsNamespace(t *testing.T) { + set, err := json.Marshal(InvokeNexusOperation{Service: "S", Operation: "o", Namespace: "my-ns"}) + require.NoError(t, err) + assert.Contains(t, string(set), `"namespace":"my-ns"`) + + zero, err := json.Marshal(InvokeNexusOperation{Service: "S", Operation: "o"}) + require.NoError(t, err) + assert.NotContains(t, string(zero), "namespace") +} + +// Cancel-side counterpart: namespace marshals when set, omitted when empty. +func TestCancelNexusOperation_MarshalsNamespace(t *testing.T) { + set, err := json.Marshal(CancelNexusOperation{Service: "s", Operation: "o", OperationToken: "tok", Namespace: "my-ns"}) + require.NoError(t, err) + assert.Contains(t, string(set), `"namespace":"my-ns"`) + + zero, err := json.Marshal(CancelNexusOperation{Service: "s", Operation: "o", OperationToken: "tok"}) + require.NoError(t, err) + assert.NotContains(t, string(zero), "namespace") +} diff --git a/internal/worker_info.go b/internal/worker_info.go index 27cfb67e..b4679d90 100644 --- a/internal/worker_info.go +++ b/internal/worker_info.go @@ -21,6 +21,8 @@ type WorkerInfo struct { Workflows []WorkflowInfo // Activities provided by the worker. Activities []ActivityInfo + // NexusServices provided by the worker. + NexusServices []NexusServiceInfo `json:"NexusServices,omitempty"` } // WorkflowInfo describes a single worker workflow. @@ -43,3 +45,11 @@ type ActivityInfo struct { // Name describes public activity name. Name string `json:"name"` } + +// NexusServiceInfo describes a single Nexus service registered on the worker. +type NexusServiceInfo struct { + // Name is the Nexus service name. + Name string `json:"name"` + // Operations lists all operation names in this service. + Operations []string `json:"operations"` +} diff --git a/internal/worker_info_test.go b/internal/worker_info_test.go new file mode 100644 index 00000000..dbb77eb8 --- /dev/null +++ b/internal/worker_info_test.go @@ -0,0 +1,64 @@ +package internal + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestWorkerInfo_NexusServicesJSON(t *testing.T) { + jsonPayload := []byte(`{ + "TaskQueue": "test-queue", + "nexusServices": [ + {"name": "GreetingService", "operations": ["greet", "farewell"]}, + {"name": "EchoService", "operations": ["echo"]} + ] + }`) + + var wi WorkerInfo + require.NoError(t, json.Unmarshal(jsonPayload, &wi)) + + assert.Equal(t, "test-queue", wi.TaskQueue) + assert.Len(t, wi.NexusServices, 2) + + assert.Equal(t, "GreetingService", wi.NexusServices[0].Name) + assert.Equal(t, []string{"greet", "farewell"}, wi.NexusServices[0].Operations) + + assert.Equal(t, "EchoService", wi.NexusServices[1].Name) + assert.Equal(t, []string{"echo"}, wi.NexusServices[1].Operations) +} + +func TestWorkerInfo_EmptyNexusServices(t *testing.T) { + jsonPayload := []byte(`{"TaskQueue": "test-queue"}`) + + var wi WorkerInfo + require.NoError(t, json.Unmarshal(jsonPayload, &wi)) + + assert.Equal(t, "test-queue", wi.TaskQueue) + assert.Empty(t, wi.NexusServices) +} + +func TestNexusServiceInfo_JSONMarshal(t *testing.T) { + svc := NexusServiceInfo{ + Name: "GreetingService", + Operations: []string{"greet"}, + } + + data, err := json.Marshal(svc) + require.NoError(t, err) + + assert.JSONEq(t, `{"name":"GreetingService","operations":["greet"]}`, string(data)) +} + +// TestWorkerInfo_NexusServicesJSONLowercase guards Go's case-insensitive +// JSON unmarshal: PHP may ship `nexusServices` (camelCase) and Go must still +// populate the PascalCase-tagged field. +func TestWorkerInfo_NexusServicesJSONLowercase(t *testing.T) { + jsonPayload := []byte(`{"nexusServices":[{"name":"S","operations":["o"]}]}`) + var wi WorkerInfo + require.NoError(t, json.Unmarshal(jsonPayload, &wi)) + require.Len(t, wi.NexusServices, 1) + assert.Equal(t, "S", wi.NexusServices[0].Name) +} diff --git a/plugin.go b/plugin.go index 340b4061..bf339d17 100644 --- a/plugin.go +++ b/plugin.go @@ -56,6 +56,7 @@ type temporal struct { rrWorkflowDef *aggregatedpool.Workflow workflows map[string]*internal.WorkflowInfo activities map[string]*internal.ActivityInfo + nexusServices map[string]*internal.NexusServiceInfo mh tclient.MetricsHandler tallyCloser io.Closer tlsCfg *tls.Config @@ -360,6 +361,7 @@ func (p *Plugin) Reset() error { workers, err := aggregatedpool.TemporalWorkers( p.temporal.rrWorkflowDef, p.temporal.rrActivityDef, + aggregatedpool.NewNexusHandler(p.codec, p.actP, p.log, p.config.Namespace), wi, p.log, p.temporal.client, diff --git a/registry/nexus_started.go b/registry/nexus_started.go new file mode 100644 index 00000000..20010dd9 --- /dev/null +++ b/registry/nexus_started.go @@ -0,0 +1,5 @@ +package registry + +// NexusStartedRegistry maps an ExecuteNexusOperation message ID to the SDK's +// started-callback result `(token, err)`. token != "" for async start, "" for sync. +type NexusStartedRegistry = Registry[string] diff --git a/registry/registry.go b/registry/registry.go index 294fb8d0..7c754282 100644 --- a/registry/registry.go +++ b/registry/registry.go @@ -6,38 +6,57 @@ import ( bindings "go.temporal.io/sdk/internalbindings" ) -// IDRegistry used to gain access to child workflow ids after they become available via callback result. -type IDRegistry struct { +// Registry stores at most one (value, err) entry per uint64 ID and at most one +// listener per ID. Push delivers to a registered listener (if any); Listen +// fires immediately if a Push has already happened for that ID, otherwise it +// waits for the next Push. +type Registry[T any] struct { sync.Mutex ids sync.Map listeners sync.Map } -type Listener func(w bindings.WorkflowExecution, err error) +// ListenerFunc is invoked once per ID with the delivered (value, err) pair. +type ListenerFunc[T any] func(value T, err error) -type entry struct { - w bindings.WorkflowExecution - err error +type entry[T any] struct { + value T + err error } -func (c *IDRegistry) Listen(id uint64, cl Listener) { +func (c *Registry[T]) Listen(id uint64, cl ListenerFunc[T]) { c.listeners.Store(id, cl) val, exist := c.ids.Load(id) if exist { c.Lock() - e := val.(entry) - cl(e.w, e.err) + e := val.(entry[T]) + cl(e.value, e.err) c.Unlock() } } -func (c *IDRegistry) Push(id uint64, w bindings.WorkflowExecution, err error) { - c.ids.Store(id, entry{w: w, err: err}) +func (c *Registry[T]) Push(id uint64, value T, err error) { + c.ids.Store(id, entry[T]{value: value, err: err}) l, exist := c.listeners.Load(id) if exist { c.Lock() - list := l.(Listener) - list(w, err) + list := l.(ListenerFunc[T]) + list(value, err) c.Unlock() } } + +// Discard drops any stored entry and listener for id. Idempotent; safe to call +// when nothing is registered. Owners of an ID call this once they know no +// further Push or Listen for that ID is meaningful — e.g. when the wrapping +// operation has fully completed — to bound memory growth. +func (c *Registry[T]) Discard(id uint64) { + c.ids.Delete(id) + c.listeners.Delete(id) +} + +// IDRegistry used to gain access to child workflow ids after they become available via callback result. +type IDRegistry = Registry[bindings.WorkflowExecution] + +// Listener is the listener type for IDRegistry. +type Listener = ListenerFunc[bindings.WorkflowExecution] diff --git a/registry/registry_test.go b/registry/registry_test.go new file mode 100644 index 00000000..80c0edc4 --- /dev/null +++ b/registry/registry_test.go @@ -0,0 +1,171 @@ +package registry + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestNexusStartedRegistry_PushThenListen(t *testing.T) { + r := &NexusStartedRegistry{} + + r.Push(7, "tok-async", nil) + + var gotToken string + var gotErr error + r.Listen(7, func(token string, err error) { + gotToken = token + gotErr = err + }) + + assert.Equal(t, "tok-async", gotToken) + assert.NoError(t, gotErr) +} + +func TestNexusStartedRegistry_ListenThenPush(t *testing.T) { + r := &NexusStartedRegistry{} + + var gotToken string + var gotErr error + r.Listen(11, func(token string, err error) { + gotToken = token + gotErr = err + }) + assert.Empty(t, gotToken, "listener must not fire before Push") + + r.Push(11, "late-tok", nil) + + assert.Equal(t, "late-tok", gotToken) + assert.NoError(t, gotErr) +} + +// Sync Nexus ops push token="" — listener must still fire so caller resolves cleanly. +func TestNexusStartedRegistry_EmptyTokenSync(t *testing.T) { + r := &NexusStartedRegistry{} + + var fired bool + var gotToken string + r.Listen(3, func(token string, err error) { + fired = true + gotToken = token + }) + r.Push(3, "", nil) + + assert.True(t, fired) + assert.Equal(t, "", gotToken) +} + +func TestNexusStartedRegistry_PushError(t *testing.T) { + r := &NexusStartedRegistry{} + + startErr := errors.New("nexus start failed") + var gotErr error + var gotToken string + r.Listen(42, func(token string, err error) { + gotToken = token + gotErr = err + }) + r.Push(42, "", startErr) + + assert.Equal(t, "", gotToken) + assert.Same(t, startErr, gotErr) +} + +func TestNexusStartedRegistry_PushErrorBeforeListen(t *testing.T) { + r := &NexusStartedRegistry{} + + startErr := errors.New("nexus start failed") + r.Push(42, "", startErr) + + var gotErr error + r.Listen(42, func(token string, err error) { + gotErr = err + }) + + assert.Same(t, startErr, gotErr) +} + +func TestNexusStartedRegistry_DistinctIDs(t *testing.T) { + r := &NexusStartedRegistry{} + + var firedA, firedB bool + r.Listen(1, func(string, error) { firedA = true }) + r.Listen(2, func(string, error) { firedB = true }) + + r.Push(1, "tA", nil) + assert.True(t, firedA) + assert.False(t, firedB, "listener for ID 2 must not fire on Push(1, …)") + + r.Push(2, "tB", nil) + assert.True(t, firedB) +} + +// Repeated Push for the same ID overwrites — late Listen sees only the latest entry. +func TestNexusStartedRegistry_OverwriteEntry(t *testing.T) { + r := &NexusStartedRegistry{} + + r.Push(5, "first", nil) + r.Push(5, "second", nil) + + var gotToken string + r.Listen(5, func(token string, err error) { + gotToken = token + }) + assert.Equal(t, "second", gotToken) +} + +// Second Listen for the same ID wins — Push must reach only the most recent listener. +func TestNexusStartedRegistry_ListenReplacesListener(t *testing.T) { + r := &NexusStartedRegistry{} + + var firedFirst, firedSecond bool + r.Listen(8, func(string, error) { firedFirst = true }) + r.Listen(8, func(string, error) { firedSecond = true }) + + r.Push(8, "tok", nil) + + assert.False(t, firedFirst, "replaced listener must not fire") + assert.True(t, firedSecond) +} + +// After Discard, a late Listen must NOT see the previously-pushed entry. +func TestNexusStartedRegistry_DiscardDropsPushedEntry(t *testing.T) { + r := &NexusStartedRegistry{} + + r.Push(101, "tok", nil) + r.Discard(101) + + var fired bool + r.Listen(101, func(string, error) { fired = true }) + + assert.False(t, fired, "Listen after Discard must not fire from the previous Push") +} + +// After Discard, a subsequent Push must NOT reach a previously registered listener. +func TestNexusStartedRegistry_DiscardDropsListener(t *testing.T) { + r := &NexusStartedRegistry{} + + var fired bool + r.Listen(202, func(string, error) { fired = true }) + r.Discard(202) + r.Push(202, "tok", nil) + + assert.False(t, fired, "Push after Discard must not reach the dropped listener") +} + +// Discard on an unknown ID is a no-op, not a panic. +func TestNexusStartedRegistry_DiscardUnknownIDIsNoop(t *testing.T) { + r := &NexusStartedRegistry{} + + assert.NotPanics(t, func() { r.Discard(9999) }) +} + +// Discard is idempotent — calling twice on the same ID is safe. +func TestNexusStartedRegistry_DiscardIdempotent(t *testing.T) { + r := &NexusStartedRegistry{} + + r.Push(303, "tok", nil) + r.Discard(303) + assert.NotPanics(t, func() { r.Discard(303) }) +} diff --git a/rpc.go b/rpc.go index abaf8fc0..824a919a 100644 --- a/rpc.go +++ b/rpc.go @@ -116,6 +116,17 @@ func (r *rpc) GetWorkflowNames(_ bool, out *[]string) error { return nil } +func (r *rpc) GetNexusServiceNames(_ bool, out *[]string) error { + r.plugin.mu.RLock() + defer r.plugin.mu.RUnlock() + + for k := range r.plugin.temporal.nexusServices { + *out = append(*out, k) + } + + return nil +} + func (r *rpc) ReplayWorkflow(in *protoApi.ReplayRequest, out *protoApi.ReplayResponse) error { r.plugin.log.Debug("replay workflow request", zap.String("run_id", in.GetWorkflowExecution().GetRunId()), diff --git a/tests/env/docker-compose-temporal.yaml b/tests/env/docker-compose-temporal.yaml index f8ca8a95..9800b91b 100644 --- a/tests/env/docker-compose-temporal.yaml +++ b/tests/env/docker-compose-temporal.yaml @@ -54,6 +54,9 @@ services: - POSTGRES_USER=temporal - POSTGRES_PWD=temporal - POSTGRES_SEEDS=postgresql + - DYNAMIC_CONFIG_FILE_PATH=config/dynamicconfig/docker.yaml + volumes: + - ./dynamicconfig:/etc/temporal/config/dynamicconfig ports: - "7233:7233" diff --git a/tests/env/dynamicconfig/docker.yaml b/tests/env/dynamicconfig/docker.yaml index 1cf95977..eedfd4a5 100644 --- a/tests/env/dynamicconfig/docker.yaml +++ b/tests/env/dynamicconfig/docker.yaml @@ -4,3 +4,9 @@ frontend.enableUpdateWorkflowExecution: frontend.enableUpdateWorkflowExecutionAsyncAccepted: - value: true constraints: {} +frontend.ListWorkersEnabled: + - value: true + constraints: {} +frontend.workerHeartbeatsEnabled: + - value: true + constraints: {} diff --git a/tests/general/hp_test.go b/tests/general/hp_test.go index a7b1edca..e2c0c7cb 100644 --- a/tests/general/hp_test.go +++ b/tests/general/hp_test.go @@ -50,6 +50,10 @@ func Test_VerifyRegistrationProto(t *testing.T) { assert.Contains(t, activities, "HeartBeatActivity.doSomething") assert.Contains(t, activities, "SimpleActivity.lower") + + nexusServices := getNexusServices(t) + assert.Empty(t, nexusServices) + stopCh <- struct{}{} wg.Wait() } @@ -893,3 +897,16 @@ func getWorkflows(t *testing.T) []string { return res } + +func getNexusServices(t *testing.T) []string { + conn, err := (&net.Dialer{}).DialContext(t.Context(), "tcp", "127.0.0.1:6001") + assert.NoError(t, err) + c := rpc.NewClientWithCodec(goridgeRpc.NewClientCodec(conn)) + + var res []string + + err = c.Call("temporal.GetNexusServiceNames", true, &res) + assert.NoError(t, err) + + return res +}