Add Committed access for commands

Author: cschleiden

internal/command/command.go

@@ -7,17 +7,17 @@ import (
 
 type CommandState int
 
-//  ┌───────┐
-//  │Pending├
-//  └───────┘
-//      ▼
-// ┌─────────┐
-// │Committed│
-// └─────────┘
-//      ▼
-//   ┌────┐
-//   │Done│
-//   └────┘
+//        ┌───────┐
+// ┌──────┤Pending│ - Command is pending, has just been added
+// │      └───────┘
+// │          ▼
+// │     ┌─────────┐
+// │     │Committed│ - Command has been committed. Its results (e.g., events) have been checkpointed
+// │     └─────────┘
+// │          ▼
+// │       ┌────┐
+// └──────►│Done│ - Command has been marked as done.
+//         └────┘
 const (
 	CommandState_Pending CommandState = iota
 	CommandState_Committed
@@ -27,15 +27,17 @@ const (
 type Command interface {
 	ID() int64
 
+	Type() string
+
+	State() CommandState
+
+	Committed() bool
+
 	Commit(clock clock.Clock) *CommandResult
 
 	// Done marks the command as done. This transitions the state to done and indicates that the result
 	// of this command has been applied.
 	Done()
-
-	State() CommandState
-
-	Type() string
 }
 
 type CommandResult struct {
@@ -64,10 +66,15 @@ func (c *command) ID() int64 {
 	return c.id
 }
 
+func (c *command) Committed() bool {
+	return c.state >= CommandState_Committed
+}
+
 func (c *command) State() CommandState {
 	return c.state
 }
 
+// Done marks the command as done
 func (c *command) Done() {
 	c.state = CommandState_Done
 }

workflow/activity.go

@@ -66,7 +66,7 @@ func executeActivity[TResult any](ctx Context, options ActivityOptions, attempt
 		if c, ok := d.(sync.ChannelInternal[struct{}]); ok {
 			if _, ok := c.ReceiveNonBlocking(); ok {
 				// Workflow has been canceled, check if the activity has already been scheduled, no need to schedule otherwise
-				if cmd.State() != command.CommandState_Committed {
+				if !cmd.Committed() {
 					cmd.Done()
 					wfState.RemoveFuture(scheduleEventID)
 					f.Set(*new(TResult), sync.Canceled)

workflow/subworkflow.go

@@ -71,7 +71,7 @@ func createSubWorkflowInstance[TResult any](ctx sync.Context, options SubWorkflo
 	// Check if the channel is cancelable
 	if c, cancelable := ctx.Done().(sync.CancelChannel); cancelable {
 		c.AddReceiveCallback(func(v struct{}, ok bool) {
-			if cmd.State() == command.CommandState_Committed {
+			if cmd.Committed() {
 				// The command is committed, that means the sub-workflow is already started. Create and add a cancel command
 				// to stop the sub-workflow execution.
 				cancelScheduleEventID := wfState.GetNextScheduleEventID()

workflow/timer.go

@@ -42,7 +42,7 @@ func ScheduleTimer(ctx Context, delay time.Duration) Future[struct{}] {
 		// Register a callback for when it's canceled. The only operation on the `Done` channel
 		// is that it's closed when the context is canceled.
 		c.AddReceiveCallback(func(v struct{}, ok bool) {
-			if timerCmd.State() == command.CommandState_Committed {
+			if timerCmd.Committed() {
 				// If the timer command is already committed, create a cancel command to allow the backend
 				// to clean up the scheduled timer message.
 				cancelScheduleEventID := wfState.GetNextScheduleEventID()