Temporal fails to create workflow with the same ID as previous one when Scylla is used as persistence

[dbarbashov]

Expected Behavior

Have Temporal up and running and have its persistence configured to work with Scylla. When child-workflow sample is ran twice in a row, child workflow finishes twice.

Actual Behavior

Temporal child-workflow sample couldn't be executed twice in a row. Child workflow only finishes once and is never created second time. Seems like Temporal gets into some loop as it consumes more than 200% cpu (according to Macs' Activity Monitor). Web UI becomes unresponsive.

Also there is following error in temporal logs:

{
  "level":"error",
  "ts":"2022-03-30T17:20:42.225Z",
  "msg":"Persistent store operation Failure",
  "shard-id":1,
  "address":"127.0.0.1:7234",
  "wf-namespace-id":"1e16cd9f-097d-4e81-8969-05fce5865111",
  "wf-id":"ABC-SIMPLE-CHILD-WORKFLOW-ID",
  "wf-run-id":"e4b9f903-85dc-4b1f-ab83-77e3e321775c",
  "store-operation":"create-wf-execution",
  "error":"Encounter shard ownership lost, request range ID: 55456, actual range ID: 0",
  "logging-call-at":"transaction_impl.go:344",
  "stacktrace":"go.temporal.io/server/common/log.(*zapLogger).Error\n\t/temporal/common/log/zap_logger.go:142\ngo.temporal.io/server/service/history/workflow.createWorkflowExecutionWithRetry\n\t/temporal/service/history/workflow/transaction_impl.go:344\ngo.temporal.io/server/service/history/workflow.(*ContextImpl).CreateWorkflowExecution\n\t/temporal/service/history/workflow/context.go:343\ngo.temporal.io/server/service/history.(*historyEngineImpl).StartWorkflowExecution\n\t/temporal/service/history/historyEngine.go:580\ngo.temporal.io/server/service/history.(*Handler).StartWorkflowExecution\n\t/temporal/service/history/handler.go:527\ngo.temporal.io/server/api/historyservice/v1._HistoryService_StartWorkflowExecution_Handler.func1\n\t/temporal/api/historyservice/v1/service.pb.go:931\ngo.temporal.io/server/common/rpc/interceptor.(*RateLimitInterceptor).Intercept\n\t/temporal/common/rpc/interceptor/rate_limit.go:84\ngoogle.golang.org/grpc.chainUnaryInterceptors.func1.1\n\t/go/pkg/mod/google.golang.org/grpc@v1.44.0/server.go:1116\ngo.temporal.io/server/common/rpc/interceptor.(*TelemetryInterceptor).Intercept\n\t/temporal/common/rpc/interceptor/telemetry.go:108\ngoogle.golang.org/grpc.chainUnaryInterceptors.func1.1\n\t/go/pkg/mod/google.golang.org/grpc@v1.44.0/server.go:1119\ngo.temporal.io/server/common/metrics.NewServerMetricsTrailerPropagatorInterceptor.func1\n\t/temporal/common/metrics/grpc.go:113\ngoogle.golang.org/grpc.chainUnaryInterceptors.func1.1\n\t/go/pkg/mod/google.golang.org/grpc@v1.44.0/server.go:1119\ngo.temporal.io/server/common/metrics.NewServerMetricsContextInjectorInterceptor.func1\n\t/temporal/common/metrics/grpc.go:66\ngoogle.golang.org/grpc.chainUnaryInterceptors.func1.1\n\t/go/pkg/mod/google.golang.org/grpc@v1.44.0/server.go:1119\ngo.temporal.io/server/common/rpc.ServiceErrorInterceptor\n\t/temporal/common/rpc/grpc.go:131\ngoogle.golang.org/grpc.chainUnaryInterceptors.func1.1\n\t/go/pkg/mod/google.golang.org/grpc@v1.44.0/server.go:1119\ngoogle.golang.org/grpc.chainUnaryInterceptors.func1\n\t/go/pkg/mod/google.golang.org/grpc@v1.44.0/server.go:1121\ngo.temporal.io/server/api/historyservice/v1._HistoryService_StartWorkflowExecution_Handler\n\t/temporal/api/historyservice/v1/service.pb.go:933\ngoogle.golang.org/grpc.(*Server).processUnaryRPC\n\t/go/pkg/mod/google.golang.org/grpc@v1.44.0/server.go:1282\ngoogle.golang.org/grpc.(*Server).handleStream\n\t/go/pkg/mod/google.golang.org/grpc@v1.44.0/server.go:1616\ngoogle.golang.org/grpc.(*Server).serveStreams.func1.2\n\t/go/pkg/mod/google.golang.org/grpc@v1.44.0/server.go:921"
}

Steps to Reproduce the Problem

1. Clone reproduction repo
2. Follow Steps to reproduce section

Specifications

• Version: 1.15.2
• Platform: Mac OS Monterey 12.2.1 (M1)

[v0id3r]

Some investigation:
We have

temporal/service/history/historyEngine.go

Line 504 in 6d90f65

func (e *historyEngineImpl) StartWorkflowExecution(

It tries to create brand new WF:

createMode := persistence.CreateWorkflowModeBrandNew
	prevRunID := ""
	prevLastWriteVersion := int64(0)
	err = weContext.CreateWorkflowExecution(
		now,
		createMode,
		prevRunID,
		prevLastWriteVersion,
		mutableState,
		newWorkflow,
		newWorkflowEventsSeq,
	)

and on specific error continues with reuse:

if t, ok := err.(*persistence.CurrentWorkflowConditionFailedError); ok {
...
createMode = persistence.CreateWorkflowModeWorkflowIDReuse
			prevRunID = t.RunID
			prevLastWriteVersion = t.LastWriteVersion
			if err = e.applyWorkflowIDReusePolicyHelper(
				t.RequestID,
				prevRunID,
				t.State,
				t.Status,
				execution,
				startRequest.StartRequest.GetWorkflowIdReusePolicy(),
			); err != nil {
				return nil, err
			}
			err = weContext.CreateWorkflowExecution(
				now,
				createMode,
				prevRunID,
				prevLastWriteVersion,
				mutableState,
				newWorkflow,
				newWorkflowEventsSeq,
			)
...
}

On the first call to weContext.CreateWorkflowExecution we have an error, but the one returned is shard ownership lost, as

temporal/common/persistence/cassandra/errors.go

Line 59 in 6d90f65

func convertErrors(

does sorting by priority and shard ownership lost has priority 0:

errors = sortErrors(errors)
return errors[0]

Why do we have shard ownership lost errors at all?
Looks like conflictIter.MapScan(conflictRecord) has records with type = 0 (for some reason) when batch fails.
In this case call to

temporal/common/persistence/cassandra/errors.go

Line 114 in 6d90f65

func extractErrors(

succeeds at

temporal/common/persistence/cassandra/errors.go

Line 124 in 6d90f65

if err := extractShardOwnershipLostError(

and appends this kind of error.

As a PoC fix worked:

• return a combined error (with the help of e.g. multierr)

return multierr.Combine(errors...)

• assert error type with errors.As

var t *persistence.CurrentWorkflowConditionFailedError
if !errors.As(err, &t) {
    return nil, err
}

In this case, everything works fine.

But I am still worried about error extraction logic and what side effects this change might introduce.
One is that any direct error type assertion should be refactored to use errors.As, another - including shard ownership lost is wrong and some other place might do wrong branching if it detects this type is also present.
Do we need to assert iterator at all in case an error is found in the record itself?

[v0id3r]

@wxing1292 Looks like you might know better (found your commits touching mentioned parts). Would be great if you can help us!

[wxing1292]

Why do we have shard ownership lost errors at all?

cassandra persistence layer will issue a batch of queries, asserting shard ownership & create mutable states & insert tasks

As a PoC fix worked:

return a combined error (with the help of e.g. multierr)
return multierr.Combine(errors...)
assert error type with errors.As
var t *persistence.CurrentWorkflowConditionFailedError
if !errors.As(err, &t) {
return nil, err
}
...
But I am still worried about error extraction logic and what side effects this change might introduce.
One is that any direct error type assertion should be refactored to use errors.As

the reason not to wrap the error ( and parse later ) but to filter & return the most relevant error is to reduce complexity:
if seeing shard ownership lost error, then simply give up the shard
if seeing current workflow condition error, then current run ID mismatch is encountered
...
plus, persistence layer is responsible of parsing the result ( and error ) returned from DB

another - including shard ownership lost is wrong and some other place might do wrong branching if it detects this type is also present.

what do you mean? why returning shard ownership error is wrong?

[wxing1292]

currently scylla is not supported by temporal, but if you want to check the compatibility ( using our cassandra driver ), then try this

[v0id3r]

currently scylla is not supported by temporal, but if you want to check the compatibility ( using our cassandra driver ), then try this

I found the cause finally.
Batch result rows/iterator has slightly different behaviour:

• Cassandra: returns only 2 rows with values

- Scylla: returns 5 rows (as we have 5 queries in batch) with corresponding ones (2 rows) containing values and others - not, all columns nulls.

Not sure if there are conditions when Cassandra might also return nullable rows.

All error extraction helpers rely on type column comparing with predefined consts/enum:

temporal/common/persistence/cassandra/errors.go

Lines 175 to 182 in 306eda8

	rowType, ok := conflictRecord["type"].(int)
	if !ok {
	// this case should not happen, maybe panic?
	return nil
	}
	if rowType != rowTypeShard {
	return nil
	}

By-default scanner uses zero-value for nullable columns and defaults our type to 0 (rowTypeShard).
https://github.com/gocql/gocql/blob/a3457c7b31fe9738d2df4891fc3d6c554711b586/marshal.go#L280-L284

And we have false-erroring about ownership lost (as range_id is also unmarshalled from null to 0 integer) on all-null rows:

temporal/common/persistence/cassandra/errors.go

Lines 184 to 185 in 306eda8

	actualRangeID := conflictRecord["range_id"].(int64)
	if actualRangeID != requestRangeID {

I successfully tested it working on both engines if we scan type column as nullable (*int) and check if it is actually set:

// helper to initialize dest for scanning
// it is a feature of gocql to reuse keys in map during scanning already
func newConflictRecord() map[string]interface{}{ {
	typ := new(int)
	return map[string]interface{}{
		"type": &typ,
	}
} 
// use nullable type column
rowType, ok := conflictRecord["type"].(*int)
if !ok || rowType == nil { ... }

Will it be possible to accept such a small change if I prepare PR and verify tests are all fine?
We don't have any other issues with Scylla and it would help not to maintain fork just for this small fix.

[wxing1292]

Will it be possible to accept such a small change if I prepare PR and verify tests are all fine?

plz send out the PR, thank you for the contribution :)

cc @yiminc

[v0id3r]

Will it be possible to accept such a small change if I prepare PR and verify tests are all fine?

plz send out the PR, thank you for the contribution :)

cc @yiminc

Will do, thank you for assistance!

[PenguinToast]

We're also running Temporal on top of ScyllaDB, so I went ahead and made a PR since I didn't see one submitted yet for this issue (#3027). Happy to close mine if there's already an existing PR out for this.

[tdeebswihart]

@PenguinToast and @v0id3r: are either of you still running Temporal on top of ScyllaDB? If so I'd like to reach out and talk to y'all about it