Refactor the Update type to a state machine #4297

mmcshane · 2023-05-07T22:11:45Z

What changed?

Write update.Update as a protocol state machine that encapsulates all message processing, validation, event writing, and metric emission related to an update.
Make explicit the set of "Provisional" state transitions wherein the Update has recieved and reacted to a protocol message but cannot make externally observable changes like returning update outcomes to RPC callers until it knows that changes have been written to stable storage.
The effect package is introduced as a utility to buffer and later apply or cancel a set of effects.
With more responsibility in the Update type, UpdateRegistry is simplified.
UpdateRegistry can restore its state (mainly in-flight Accepted updates) from MutableState.
Update message validation is moved into the update package.
UpdateRegistry differentiates between having incomplete updates (which prevents a workflow from closing) and having messages to send (which may trigger the creation of a new workflow task)
Removed a chunk of processing from workflowTaskHandler that is now encapsulated in the Update state machine, allowing task handling to be more decoupled from update protocol semantics.

Why?
This is a step in the direction of being able to process messages one at a time rather than in a batch. Individual message processing will be required to respect protocols (like update) that need message processing that is ordered with respect to other commands. Additionally we lay the groundwork for async updates and outcome polling.

How did you test it?

Lots of new unit tests for the update package
Corrections to functional tests as needed (/tests/ directory here)
Sync update tests in the features repo

Potential risks
I'm not thrilled about the size of this PR but there were a number of things that could not be teased apart. Changes external to the update package were minimized

Is hotfix candidate?
no.

mmcshane · 2023-05-08T00:51:43Z

tests/update_workflow_test.go

 			} else {
-				s.NoError(err)
+				require.NoError(t, err)


This block here is the loop body of a table-driven test within a Suite. Using s.Error or s.Contains here is wrong because it fails the parent test rather than the current test as we're iterating through the testcase data.

mmcshane · 2023-05-08T00:54:33Z

tests/update_workflow_test.go

+			// 3rd attempt UpdateWorkflowExecution call has timed out but the
+			// update is still running
+			updRequestMsg := task.Messages[0]
+			s.EqualValues(9, updRequestMsg.GetEventId())


When the update was cleared/deleted at the end of the RPC handler function (with a defer), this was correct. However as part of supporting async updates, we will need to let Updates exist in the Registry longer than the RPC handler function is on the stack. Thus this modification to the test to expect that the update has not gone away at this point.

mmcshane · 2023-05-08T00:58:44Z

internal/effect/controller.go

+
+type Controller interface {
+	OnAfterCommit(func(context.Context))
+	OnAfterRollback(func(context.Context))


passing context here not so much for timeout/cancellation as for the possibility that there might be useful values attached - specifically otel spans. I also considered including a ControllerID in the callback so that these funcs could verify that the caller is actually the right caller. But these callbacks need to be infallible so it felt like there wasn't any point to that.

mmcshane · 2023-05-08T01:00:48Z

service/history/api/updateworkflow/api.go

For encapsulation, I'd prefer for this Invoke function to be in the update package. Maybe in the future - not needed for this PR

mmcshane · 2023-05-08T01:03:51Z

service/history/workflow/update/registry.go

+	RegistryStore interface {
+		GetAcceptedWorkflowExecutionUpdateIDs(context.Context) []string
+		GetUpdateInfo(ctx context.Context, updateID string) (*persistencespb.UpdateInfo, bool)
+		GetUpdateOutcome(ctx context.Context, updateID string) (*updatepb.Outcome, error)


I split the Registry storage needs and the Event storage needs since they're used separately. In practice, they're both MutableState.

mmcshane · 2023-05-08T01:07:27Z

service/history/workflow/update/registry.go

 )

-var _ Registry = (*RegistryImpl)(nil)
+//revive:disable:unexported-return I *want* it to be unexported


I don't want external packages to be able to create their own option implementations. I get that there's a hack where it's still possible, but this makes it harder and sends a message.

How about moving to an internal package instead?

mmcshane · 2023-05-08T01:11:22Z

service/history/workflow/update/registry.go

+		Find(ctx context.Context, protocolInstanceID string) (*Update, bool)
+
+		// CreateOutgoingMessages polls each registered Update for outbound
+		// messages and returns them.
 		CreateOutgoingMessages(startedEventID int64) ([]*protocolpb.Message, error)


startedEventID probably shouldn't be here - it breaks layering - but it's fine for now.

mmcshane · 2023-05-08T01:18:51Z

service/history/workflow/update/update.go

+// effectively gives us update request deduplication by update ID. If the Update
+// is in stateAdmitted then it builds a protocolpb.Message that will be sent on
+// ensuing calls to PollOutboundMessages until the update is accepted.
+func (u *Update) onRequestMsg(


The pattern for all the on<Type>Msg functions:

assert expected state(s)

validate message

update msg counter

write event(s) if applicable

transition to next provisional state

register OnAfterCommit to transition from Provisional to non-Provisional state and set Future(s) if applicable

register OnAfterRollback to undo transition into provisional

mmcshane · 2023-05-08T01:22:22Z

service/history/workflow/update/util.go

+	// lazyOutcome adapts a func to the future.Future[*updatepb.Outcome] interface
+	lazyOutcome func(context.Context) (*updatepb.Outcome, error)
+
+	instrumentation struct {


Wouldn't be the worst idea to make this type and its fields exported in common (remove the member functions) and attach an instance to every context.Context. Would save us a lot of boilerplate moving these three types around to various places.

mmcshane · 2023-05-08T01:26:02Z

service/history/workflowTaskHandler.go

+	); err != nil {
+		return handler.failWorkflow(enumspb.WORKFLOW_TASK_FAILED_CAUSE_BAD_UPDATE_WORKFLOW_EXECUTION_MESSAGE, err)
+	}
+


next 4 lines probably belong in a protocol utility package. TODO.

mmcshane · 2023-05-08T01:37:13Z

service/history/workflowTaskHandlerCallbacks.go

+			effects.Cancel(ctx)
+		}
+		effects.Apply(ctx)
+	}()


This seems like the safest thing but the error handling is weird between here and workflowTaskHandler so I've also added additional calls to effects.Cancel in some key places - namely, if command/message handling fails and if the MutableState write fails.

Why? To cancel effects earlier? I think this PR comment needs to be converted into code comment.

service/history/workflow/update/update.go

mmcshane · 2023-05-09T14:08:18Z

For reference, if we accept this refactoring

this is what in-flight update limiting looks like
this is what async updates look like

mindaugasrukas · 2023-05-09T17:58:10Z

internal/effect/buffer.go

+// Buffer holds a set of effect and rollback functions that can be invoked as a
+// batch with a defined order. Once either Apply or Cancel is called, all
+// buffered effects are cleared. This type is not threadsafe. The zero-value of
+// a Buffer is a valid state.
+type Buffer struct {
+	effects []func(context.Context)
+	cancels []func(context.Context)
+}


As noted that this type is not threadsafe is this not an issue?

Just means that it's up to the caller to make sure that there's no concurrent usage. Pretty common for Go types.

mindaugasrukas · 2023-05-09T18:00:28Z

internal/effect/controller.go

+	OnAfterCommit(func(context.Context))
+	OnAfterRollback(func(context.Context))


nit: how about creating a separate type for func(context.Context)?

I want for values to be able to implement this interface without referencing this package. If I use type CommitCallback func(context.Context) or similar then an implementer will have to import this pkg.

mindaugasrukas · 2023-05-09T18:16:45Z

service/history/api/updateworkflow/api.go

-	upd, duplicate, removeFn := weCtx.GetContext().UpdateRegistry().Add(req.GetRequest().GetRequest())
-	if removeFn != nil {
-		defer removeFn()
+	updateID := req.GetRequest().GetRequest().GetMeta().GetUpdateId()


🤯 .GetRequest().GetRequest()

Another question is there a possible NPE(nil-pointer-exception)?

Request.Request is because the pattern used is to wrap a workflowservice request with a historyservice request. This is pretty common in our system. There's no NPE concern because there's upstream validation (in Frontend service) and because the Go codegen for protobuf types includes a nil check in the Get<Thing>() function definitions.

This is pretty common in our system is a wrong argument IMO ;)
And from the user's perspective, it looks like a bug having Request.Request.
IMO it should be something like this:

req.GetWorkflowServiceRequest() req.GetHistoryServiceRequest()

It is matter of preferences, but WorkflowService.Request is a type. Including type name in field name is not good. Everyone just used to the fact that in both matching and history service, inner request is original FE request. GetOriginRequest? GetFrontendRequest? GetUserRequest? Just GetRequest!

Something smells here. You won't make me believe that having .GetRequest().GetRequest() is the correct approach.
Shouldn't WorkflowService be able to return UpdateId without knowing the internals of the n-th level dependency?

What I meant is that wrapping an api request in a history request is common

Shouldn't WorkflowService be able to return UpdateId without knowing the internals of the n-th level dependency?

It absolutely should. Law of Demeter and all that. Problem is that we're dealing with generated code here and Protobuf doesn't have any way to specify inline embedding of messages so you end up with GetParent().GetChild().GetFoo(). If in a proto message you could specify something like "all the field from that message plus these extra fields" (similar to struct embedding in Go) then we could specify these messages that way and we wouldn't have this problem. But that capability doesn't exist.

I mean if you want to get down to it, I don't like the "Get" part of these method names or the fact that proto doesn't capitalize the last 'd' in UpdateId. Working with these generated types is always jarring.

Well, I am happy with Id. I think we should change all ID in our go code base to Id.

service/history/workflow/update/registry.go

mindaugasrukas · 2023-05-09T18:36:19Z

service/history/workflow/update/registry.go

-		sync.RWMutex
-		updates map[string]*Update
-		store   Storage
+		mu              sync.RWMutex


nit: what is the reasoning for naming the sync.RWMutex?

I consider it an antipattern and agree with uber-go/guide#127

mindaugasrukas · 2023-05-09T18:37:47Z

service/history/workflow/update/registry.go

 )

-var _ Registry = (*RegistryImpl)(nil)
+//revive:disable:unexported-return I *want* it to be unexported


How about moving to an internal package instead?

mindaugasrukas · 2023-05-09T19:11:24Z

service/history/workflowTaskHandlerCallbacks.go

+			weContext.UpdateRegistry(ctx),
+			&effects,


nit: the function newWorkflowTaskHandler has too many arguments.

I don't disagree. But that's going to be out of scope for this PR

I'm not asking to fix this in this PR; I'm just pointing here as you added more arguments.

https://www.brainyquote.com/quotes/alan_perlis_177279 😄

This is genius. I haven't heard it!

service/history/workflow/update/registry.go

alexshtin · 2023-05-09T05:11:52Z

service/history/workflow/util.go

+	}
+}
+
+type MutableStateWithEffects struct {


Why here? Why not mutable_state_with_effects.go ?

40% lazy; 60% it's just a quick utility class to bring together two types so it seemed like util.go was reasonable. I don't subscribe to one-type == one-file.

alexshtin · 2023-05-10T00:04:02Z

service/history/api/updateworkflow/api.go

-	upd, duplicate, removeFn := weCtx.GetContext().UpdateRegistry().Add(req.GetRequest().GetRequest())
-	if removeFn != nil {
-		defer removeFn()
+	updateID := req.GetRequest().GetRequest().GetMeta().GetUpdateId()


It is matter of preferences, but WorkflowService.Request is a type. Including type name in field name is not good. Everyone just used to the fact that in both matching and history service, inner request is original FE request. GetOriginRequest? GetFrontendRequest? GetUserRequest? Just GetRequest!

service/history/workflow/update/update.go

alexshtin · 2023-05-10T01:00:28Z

service/history/workflow/update/update.go

-		outcome               *future.FutureImpl[*updatepb.Outcome]
-		pendingOutcome        *updatepb.Outcome
+		// accessed only while holding workflow lock
+		id              string


id field in Update struct reads as UpdateID. I believe MessageID is a better name here.

This is the UpdateID though. MessageID is built just-in-time when sending the protocolpb.Message

service/history/workflow/update/registry.go

service/history/workflow/update/update.go

alexshtin · 2023-05-10T01:45:15Z

service/history/workflow/update/util.go

+func invalidArgf(tmpl string, args ...any) error {
+	return serviceerror.NewInvalidArgument(fmt.Sprintf(tmpl, args...))
+}


I am not sure about helpers like this. If we use the everywhere and build up a culture of using it, yes.

There should be one of these in common/something for each of the types in serviceerror. But that's a different PR

the moment you move it out it becomes not that attractive, because it needs to be prefixed with package name. I whould add serviceerror.NewInvalidArgumentf for every error in serviceerror package.

alexshtin · 2023-05-10T01:53:33Z

service/history/workflowTaskHandlerCallbacks.go

+			effects.Cancel(ctx)
+		}
+		effects.Apply(ctx)
+	}()


Why? To cancel effects earlier? I think this PR comment needs to be converted into code comment.

alexshtin · 2023-05-10T01:54:13Z

service/history/workflowTaskHandlerCallbacks.go

@@ -468,8 +476,6 @@ func (handler *workflowTaskHandlerCallbacksImpl) handleWorkflowTaskCompleted(
 		activityNotStartedCancelled bool
 		newMutableState             workflow.MutableState
 	)
-	// hasPendingUpdates indicates if there are more pending updates (excluding those which are accepted/rejected by this workflow task).
-	hasPendingUpdates := weContext.UpdateRegistry().HasPending(request.GetMessages())


I am super happy see this getting removed.

Ha. If only it didn't break 🤦

mmcshane · 2023-05-10T12:01:22Z

service/history/workflowTaskHandler.go

 	switch command.GetCommandType() {
 	case enumspb.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK:
 		return handler.handleCommandScheduleActivity(ctx, command.GetScheduleActivityTaskCommandAttributes())

 	case enumspb.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION:
+		if err := earlyDeliverMessages(ctx); err != nil {
+			return nil, err
+		}


@alexshtin this what gets us around the problem discovered last night - early deliver messages in this one case so that update completion is visible to the command handler.

mmcshane · 2023-05-10T12:16:26Z

tests/update_workflow_test.go

- 12 WorkflowExecutionCompleted`, events)
+ 12 WorkflowExecutionUpdateAccepted
+ 13 WorkflowExecutionUpdateCompleted
+ 14 WorkflowExecutionCompleted`, events)


Old expectation was based on a bug - we shouldn't be able to complete executions with incomplete updates so either we see UpdateAccepted+UpdateCompleted before the WorkflowExecutionCompleted or the history is invalid.

alexshtin · 2023-05-10T17:37:27Z

service/history/api/updateworkflow/api.go

-	upd, duplicate, removeFn := weCtx.GetContext().UpdateRegistry().Add(req.GetRequest().GetRequest())
-	if removeFn != nil {
-		defer removeFn()
+	updateID := req.GetRequest().GetRequest().GetMeta().GetUpdateId()


Well, I am happy with Id. I think we should change all ID in our go code base to Id.

alexshtin · 2023-05-10T17:42:23Z

service/history/workflow/update/util.go

+func invalidArgf(tmpl string, args ...any) error {
+	return serviceerror.NewInvalidArgument(fmt.Sprintf(tmpl, args...))
+}


the moment you move it out it becomes not that attractive, because it needs to be prefixed with package name. I whould add serviceerror.NewInvalidArgumentf for every error in serviceerror package.

alexshtin · 2023-05-10T17:45:57Z

service/history/workflowTaskHandlerCallbacks.go

+			weContext.UpdateRegistry(ctx),
+			&effects,


This is genius. I haven't heard it!

alexshtin · 2023-05-10T17:55:51Z

This is a very good PR. I learnt few things reviewing it.

This is a step in the direction of being able to process messages one at a time rather than in a batch. Individual message processing will be required to respect protocols (like update) that need message processing that is ordered with respect to other commands. Additionally we lay the groundwork for async updates and outcome polling. A summary of what changed: - Write update.Update as a protocol state machine that encapsulates all message processing, validation, event writing, and metric emission related to an update. - Make explicit the set of "Provisional" state transitions wherein the Update has recieved and reacted to a protocol message but cannot make externally observable changes like returning update outcomes to RPC callers until it knows that changes have been written to stable storage. - The effect package is introduced as a utility to buffer and later apply or cancel a set of effects. - With more responsibility in the Update type, UpdateRegistry is simplified. - UpdateRegistry can restore its state (mainly in-flight Accepted updates) from MutableState. - Update message validation is moved into the update package. - UpdateRegistry differentiates between having incomplete updates (which prevents a workflow from closing) and having messages to send (which may trigger the creation of a new workflow task) - Removed a chunk of processing from workflowTaskHandler that is now encapsulated in the Update state machine, allowing task handling to be more decoupled from update protocol semantics.

Return true for found, false for not found.

Closing a workflow with incomplete updates causes an error but it may be the case that the messages that complete the incomplete updates are in the same WFTCompleted message as the CompleteWorkflowExecution command. In such cases becuase we currently deliver all messages _after_ all commands, the fact that the update completed is not known at the time we attempt to handle the CompleteWorkflowExecution command and thus the WFTCompletion fails. Here we add (temporarily) the ability to early-deliver messages for this one command type. Soon the messages will be sequenced in with the commands so that this problem goes away.

Improves ergonomics of the state type and adds state masking to avoid allocating temporary slices for every call.

This reverts commit 1c7e895.

mmcshane requested a review from a team as a code owner May 7, 2023 22:11

mmcshane requested review from alexshtin, yycptt and yiminc May 7, 2023 23:45

mmcshane commented May 8, 2023

View reviewed changes

mmcshane commented May 9, 2023

View reviewed changes

service/history/workflow/update/update.go Outdated Show resolved Hide resolved

mindaugasrukas reviewed May 9, 2023

View reviewed changes

alexshtin reviewed May 10, 2023

View reviewed changes

mmcshane commented May 10, 2023

View reviewed changes

Quinn-With-Two-Ns mentioned this pull request May 10, 2023

Add support for update to the Java test server temporalio/sdk-java#1762

Merged

alexshtin approved these changes May 10, 2023

View reviewed changes

mmcshane enabled auto-merge (squash) May 10, 2023 18:00

Matt McShane added 11 commits May 10, 2023 14:13

Flipped the semantics of FindOrCreate bool return

ae2e78f

Return true for found, false for not found.

Comment typo: tis -> this

fb7a397

testing longer test timeout

6bc71c4

move update.state into its own file

02e4fb0

Can use a shared lock for Len()

036c8a0

Rename RegistryStore to UpdateStore

7a5765b

Rename CreateOutgoingMessages -> ReadOutgoingMessages

aeb2faa

Avoid allocations for update.state checks

a7b449b

Improves ergonomics of the state type and adds state masking to avoid allocating temporary slices for every call.

Comment on the belt-and-suspenders error handling

4a5bd93

mmcshane merged commit 1c7e895 into temporalio:master May 10, 2023

mmcshane deleted the mpm/split-update-registry-state branch May 10, 2023 19:15

wxing1292 added a commit that referenced this pull request May 11, 2023

Revert "Refactor the Update type to a state machine (#4297)"

6cf952f

This reverts commit 1c7e895.

		OnAfterCommit(func(context.Context))
		OnAfterRollback(func(context.Context))

Refactor the Update type to a state machine #4297

Refactor the Update type to a state machine #4297

Conversation

mmcshane commented May 7, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mmcshane May 8, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mmcshane commented May 9, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mmcshane May 9, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mmcshane May 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mmcshane May 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alexshtin commented May 10, 2023 • edited Loading

mmcshane May 8, 2023 •

edited

Loading

mmcshane May 9, 2023 •

edited

Loading

mmcshane May 10, 2023 •

edited

Loading

mmcshane May 10, 2023 •

edited

Loading

alexshtin commented May 10, 2023 •

edited

Loading