Getting corrupted workflow when sending workflow with high concurrency

[hulucc]

The list view saying its open but actually completed.

It can be easily reproduced by send 2000 workflow to cadence server concurrencyly

package main

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/google/uuid"
	"go.uber.org/cadence/activity"
	"go.uber.org/cadence/worker"
	"go.uber.org/yarpc/transport/tchannel"

	// "go.uber.org/yarpc/api/transport"
	"go.uber.org/cadence"
	"go.uber.org/cadence/.gen/go/cadence/workflowserviceclient"
	"go.uber.org/cadence/client"
	"go.uber.org/cadence/workflow"
	"go.uber.org/yarpc"
)

var cc client.Client

func init() {
	workflow.Register(SimpleWorkflow)
	workflow.Register(CronWorkflow)
	activity.Register(SimpleActivity)
}

func SimpleWorkflow(ctx workflow.Context) error {
	ao := workflow.ActivityOptions{
		ScheduleToStartTimeout: time.Minute,
		ScheduleToCloseTimeout: time.Minute,
		StartToCloseTimeout:    time.Minute,
		WaitForCancellation:    false,
		RetryPolicy: &cadence.RetryPolicy{
			InitialInterval:    time.Second,
			BackoffCoefficient: 2,
			MaximumInterval:    time.Minute,
			MaximumAttempts:    3,
			// ExpirationInterval: time.Hour,
			NonRetriableErrorReasons: []string{
				"cadenceInternal:Panic",
				"cadenceInternal:Generic",
			},
		},
	}
	ctx = workflow.WithActivityOptions(ctx, ao)
	err := workflow.ExecuteActivity(ctx, SimpleActivity).Get(ctx, nil)
	return err
}

func CronWorkflow(ctx workflow.Context) error {
	ao := workflow.ActivityOptions{
		ScheduleToStartTimeout: time.Minute,
		ScheduleToCloseTimeout: time.Minute,
		StartToCloseTimeout:    time.Minute,
		WaitForCancellation:    false,
		RetryPolicy: &cadence.RetryPolicy{
			InitialInterval:    time.Second,
			BackoffCoefficient: 2,
			MaximumInterval:    time.Minute,
			MaximumAttempts:    3,
			// ExpirationInterval: time.Hour,
			NonRetriableErrorReasons: []string{
				"cadenceInternal:Panic",
				"cadenceInternal:Generic",
			},
		},
	}
	ctx = workflow.WithActivityOptions(ctx, ao)
	workflow.ExecuteActivity(ctx, SimpleActivity).Get(ctx, nil)
	workflow.Sleep(ctx, time.Second*10)
	return workflow.NewContinueAsNewError(ctx, CronWorkflow)

}

func SimpleActivity(ctx context.Context) error {
	time.Sleep(time.Second * 1)
	println("done")
	return nil
}

func main() {
	hostPort := "cadence-frontend.cadence:7933"
	clientName := "simple-worker"
	domain := "play"
	taskList := "simple-worker"
	cadenceService := "cadence-frontend"
	ch, err := tchannel.NewChannelTransport(tchannel.ServiceName(clientName))
	if err != nil {
		panic(err)
	}
	dispatcher := yarpc.NewDispatcher(yarpc.Config{
		Name: clientName,
		Outbounds: yarpc.Outbounds{
			cadenceService: {Unary: ch.NewSingleOutbound(hostPort)},
		},
	})
	err = dispatcher.Start()
	if err != nil {
		panic(err)
	}

	service := workflowserviceclient.New(dispatcher.ClientConfig(cadenceService))
	worker := worker.New(service, domain, taskList, worker.Options{})
	cc = client.NewClient(service, domain, &client.Options{})

	retry := &cadence.RetryPolicy{
		InitialInterval:    time.Second,
		BackoffCoefficient: 2,
		MaximumInterval:    time.Minute,
		ExpirationInterval: time.Minute * 10,
		MaximumAttempts:    10,
		NonRetriableErrorReasons: []string{
			"cadenceInternal:Panic",
			"cadenceInternal:Generic",
		},
	}
	_ = retry

	start := func() {
		s := time.Now()
		wf, err := cc.StartWorkflow(context.Background(), client.StartWorkflowOptions{
			ID:                           uuid.New().String(),
			TaskList:                     taskList,
			ExecutionStartToCloseTimeout: time.Hour,
		}, SimpleWorkflow)
		if err != nil {
			panic(err)
		}
		fmt.Printf("duration: %+v, start: %v, end: %v, id: %s\n", time.Since(s).Seconds(), s.Format("15:04:05"), time.Now().Format("15:04:05"), wf.ID)
	}
	cron := func() {
		_, err := cc.StartWorkflow(context.Background(), client.StartWorkflowOptions{
			ID:                           "CronWorkflow",
			TaskList:                     taskList,
			ExecutionStartToCloseTimeout: time.Hour,
			CronSchedule:                 "* * * * *",
		}, CronWorkflow)
		if err != nil {
			panic(err)
		}
	}

	if len(os.Args) > 1 {
		switch os.Args[1] {
		case "start":
			for i := 0; i < 2000; i++ {
				go start()
			}
			select {}
		case "cron":
			cron()
			select {}
		default:
			fmt.Printf("%+v\n", "role: start/cron")
		}
	} else {
		err = worker.Run()
		if err != nil {
			panic(err)
		}
	}
}

[hulucc]

It's caused by system.enableVisibilitySampling. You can disable visibility sampling and use standalone db with visibility, or use elasticsearch to avoid this problem.

[zhangkyou]

The default value of system.enableVisibilitySampling in dynamicconfig is FALSE. I didn't write this config key in my dynamicconfig file, but I still have the problem.