feat: Nexus implementation

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