more logs

Signed-off-by: Valery Piashchynski <piashchynski.valery@gmail.com>

aggregatedpool/handler.go

@@ -170,7 +170,7 @@ func (wp *Workflow) handleMessage(msg *internal.Message) error {
 			func() (*commonpb.Payloads, error) {
 				return msg.Payloads, nil
 			},
-			wp.createContinuableCallback(msg.ID),
+			wp.createContinuableCallback(msg.ID, "SideEffect"),
 		)
 
 	case *internal.CompleteWorkflow:
@@ -303,22 +303,23 @@ 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 {
-			wp.log.Debug("calling callback IN LOOP", zap.Uint64("ID", id))
+			wp.log.Debug("calling callback IN LOOP", zap.Uint64("ID", id), zap.String("type", t))
 			callback(result, err)
 			return
 		}
 
 		wp.callbacks = append(wp.callbacks, func() error {
-			wp.log.Debug("appending callback", zap.Uint64("ID", id))
+			wp.log.Debug("appending callback", zap.Uint64("ID", id), zap.String("type", t))
 			callback(result, err)
 			return nil
 		})
 	}
 }
 
 // callback to be called inside the queue processing, adds new messages at the end of the queue
-func (wp *Workflow) createContinuableCallback(id uint64) bindings.ResultHandler {
+func (wp *Workflow) createContinuableCallback(id uint64, t string) bindings.ResultHandler {
 	callback := func(result *commonpb.Payloads, err error) {
+		wp.log.Debug("executing continuable callback", zap.Uint64("ID", id), zap.String("type", t))
 		wp.canceller.Discard(id)
 
 		if err != nil {

aggregatedpool/workflow.go

@@ -29,6 +29,13 @@ import (
 
 type Callback func() error
 
+// seqID is global sequence ID
+var seqID uint64 //nolint:gochecknoglobals
+
+func seq() uint64 {
+	return atomic.AddUint64(&seqID, 1)
+}
+
 type Workflow struct {
 	codec common.Codec
 	pool  common.Pool
@@ -38,7 +45,6 @@ type Workflow struct {
 	mq        *queue.MessageQueue
 	ids       *registry.IDRegistry
 	seqID     uint64
-	sID       func() uint64
 	runID     string
 	pipeline  []*internal.Message
 	callbacks []Callback
@@ -49,10 +55,9 @@ type Workflow struct {
 	mh  temporalClient.MetricsHandler
 }
 
-func NewWorkflowDefinition(codec common.Codec, pool common.Pool, log *zap.Logger, seqID func() uint64) *Workflow {
+func NewWorkflowDefinition(codec common.Codec, pool common.Pool, log *zap.Logger) *Workflow {
 	return &Workflow{
 		log:   log,
-		sID:   seqID,
 		codec: codec,
 		pool:  pool,
 	}
@@ -65,7 +70,6 @@ func (wp *Workflow) NewWorkflowDefinition() bindings.WorkflowDefinition {
 		pool:  wp.pool,
 		codec: wp.codec,
 		log:   wp.log,
-		sID:   wp.sID,
 	}
 }
 
@@ -81,7 +85,7 @@ func (wp *Workflow) Execute(env bindings.WorkflowEnvironment, header *commonpb.H
 	wp.canceller = new(canceller.Canceller)
 
 	// sequenceID shared for all pool workflows
-	wp.mq = queue.NewMessageQueue(wp.sID)
+	wp.mq = queue.NewMessageQueue(seq)
 	wp.ids = new(registry.IDRegistry)
 
 	env.RegisterCancelHandler(wp.handleCancel)

plugin.go

@@ -10,7 +10,6 @@ import (
 	"strconv"
 	"strings"
 	"sync"
-	"sync/atomic"
 	"time"
 
 	prom "github.com/prometheus/client_golang/prometheus"
@@ -85,7 +84,6 @@ type Plugin struct {
 	events   chan events.Event
 	stopCh   chan struct{}
 
-	seqID   uint64
 	workers []worker.Worker
 }
 
@@ -454,12 +452,6 @@ func (p *Plugin) RPC() any {
 	return &rpc{srv: p, client: p.client}
 }
 
-func (p *Plugin) SedID() uint64 {
-	p.log.Debug("sequenceID", zap.Uint64("before", atomic.LoadUint64(&p.seqID)))
-	defer p.log.Debug("sequenceID", zap.Uint64("after", atomic.LoadUint64(&p.seqID)+1))
-	return atomic.AddUint64(&p.seqID, 1)
-}
-
 func (p *Plugin) MetricsCollector() []prom.Collector {
 	// p - implements Exporter interface (workers)
 	// other - request duration and count
@@ -507,7 +499,7 @@ func (p *Plugin) initPool() error {
 		return err
 	}
 
-	p.rrWorkflowDef = aggregatedpool.NewWorkflowDefinition(p.codec, wp, p.log, p.SedID)
+	p.rrWorkflowDef = aggregatedpool.NewWorkflowDefinition(p.codec, wp, p.log)
 
 	// get worker information
 	wi, err := WorkerInfo(p.codec, wp, p.rrVersion)