Rename event id

Author: cschleiden

internal/log/log.go

@@ -0,0 +1,8 @@
+package log
+
+type Logger interface {
+	Debug(args ...interface{})
+	Info(args ...interface{})
+	Warning(args ...interface{})
+	Error(args ...interface{})
+}

internal/tester/tester.go

@@ -337,7 +337,6 @@ func (wt *workflowTester) SignalWorkflowInstance(wfi core.WorkflowInstance, name
 		e := history.NewHistoryEvent(
 			wt.clock.Now(),
 			history.EventType_SignalReceived,
-			-1,
 			&history.SignalReceivedAttributes{
 				Name: name,
 				Arg:  arg,
@@ -444,19 +443,19 @@ func (wt *workflowTester) scheduleActivity(wfi core.WorkflowInstance, event hist
 				ne = history.NewHistoryEvent(
 					wt.clock.Now(),
 					history.EventType_ActivityFailed,
-					event.EventID,
 					&history.ActivityFailedAttributes{
 						Reason: activityErr.Error(),
 					},
+					history.ScheduleEventID(event.ScheduleEventID),
 				)
 			} else {
 				ne = history.NewHistoryEvent(
 					wt.clock.Now(),
 					history.EventType_ActivityCompleted,
-					event.EventID,
 					&history.ActivityCompletedAttributes{
 						Result: activityResult,
 					},
+					history.ScheduleEventID(event.ScheduleEventID),
 				)
 			}
 
@@ -551,19 +550,19 @@ func (wt *workflowTester) scheduleSubWorkflow(event core.WorkflowEvent) {
 			he = history.NewHistoryEvent(
 				wt.clock.Now(),
 				history.EventType_SubWorkflowFailed,
-				event.WorkflowInstance.ParentEventID(),
 				&history.SubWorkflowFailedAttributes{
 					Error: workflowErr.Error(),
 				},
+				history.ScheduleEventID(event.WorkflowInstance.ParentEventID()),
 			)
 		} else {
 			he = history.NewHistoryEvent(
 				wt.clock.Now(),
 				history.EventType_SubWorkflowCompleted,
-				event.WorkflowInstance.ParentEventID(),
 				&history.SubWorkflowCompletedAttributes{
 					Result: workflowResult,
 				},
+				history.ScheduleEventID(event.WorkflowInstance.ParentEventID()),
 			)
 		}
 
@@ -585,7 +584,6 @@ func (wt *workflowTester) getInitialEvent(wf workflow.Workflow, args []interface
 	return history.NewHistoryEvent(
 		wt.clock.Now(),
 		history.EventType_WorkflowExecutionStarted,
-		-1,
 		&history.ExecutionStartedAttributes{
 			Name:   name,
 			Inputs: inputs,

internal/worker/activity.go

@@ -143,18 +143,19 @@ func (aw *activityWorker) handleTask(ctx context.Context, task *task.Activity) {
 		event = history.NewHistoryEvent(
 			aw.clock.Now(),
 			history.EventType_ActivityFailed,
-			task.Event.EventID,
 			&history.ActivityFailedAttributes{
 				Reason: err.Error(),
-			})
+			},
+			history.ScheduleEventID(task.Event.ScheduleEventID),
+		)
 	} else {
 		event = history.NewHistoryEvent(
 			aw.clock.Now(),
 			history.EventType_ActivityCompleted,
-			task.Event.EventID,
 			&history.ActivityCompletedAttributes{
 				Result: result,
-			})
+			},
+			history.ScheduleEventID(task.Event.ScheduleEventID))
 	}
 
 	if err := aw.backend.CompleteActivityTask(ctx, task.WorkflowInstance, task.ID, event); err != nil {

internal/workflow/activity.go

@@ -36,14 +36,13 @@ func executeActivity(ctx sync.Context, options ActivityOptions, activity Activit
 	}
 
 	wfState := getWfState(ctx)
-	eventID := wfState.eventID
-	wfState.eventID++
+	scheduleEventID := wfState.getNextScheduleEventID()
 
 	name := fn.Name(activity)
-	cmd := command.NewScheduleActivityTaskCommand(eventID, name, inputs)
+	cmd := command.NewScheduleActivityTaskCommand(scheduleEventID, name, inputs)
 	wfState.addCommand(&cmd)
 
-	wfState.pendingFutures[eventID] = f
+	wfState.pendingFutures[scheduleEventID] = f
 
 	// Handle cancellation
 	if d := ctx.Done(); d != nil {
@@ -57,7 +56,7 @@ func executeActivity(ctx sync.Context, options ActivityOptions, activity Activit
 				}
 
 				wfState.removeCommand(cmd)
-				delete(wfState.pendingFutures, eventID)
+				delete(wfState.pendingFutures, scheduleEventID)
 				f.Set(nil, sync.Canceled)
 			})
 		}

internal/workflow/executor.go

@@ -72,7 +72,7 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) ([]history
 	}
 
 	// Always pad the received events with WorkflowTaskStarted/Finished events to indicate the execution
-	events := []history.Event{history.NewHistoryEvent(e.clock.Now(), history.EventType_WorkflowTaskStarted, -1, &history.WorkflowTaskStartedAttributes{})}
+	events := []history.Event{history.NewHistoryEvent(e.clock.Now(), history.EventType_WorkflowTaskStarted, &history.WorkflowTaskStartedAttributes{})}
 	events = append(events, t.NewEvents...)
 
 	// Execute new events received from the backend
@@ -87,7 +87,7 @@ func (e *executor) ExecuteTask(ctx context.Context, t *task.Workflow) ([]history
 	events = append(events, newCommandEvents...)
 
 	// Execution of this task is finished, add event to history
-	events = append(events, history.NewHistoryEvent(e.clock.Now(), history.EventType_WorkflowTaskFinished, -1, &history.WorkflowTaskFinishedAttributes{}))
+	events = append(events, history.NewHistoryEvent(e.clock.Now(), history.EventType_WorkflowTaskFinished, &history.WorkflowTaskFinishedAttributes{}))
 
 	e.lastEventID = events[len(events)-1].ID
 
@@ -204,7 +204,7 @@ func (e *executor) handleWorkflowTaskStarted(event history.Event, a *history.Wor
 }
 
 func (e *executor) handleActivityScheduled(event history.Event, a *history.ActivityScheduledAttributes) error {
-	c := e.workflowState.removeCommandByEventID(event.EventID)
+	c := e.workflowState.removeCommandByEventID(event.ScheduleEventID)
 	if c != nil {
 		// Ensure the same activity is scheduled again
 		ca := c.Attr.(*command.ScheduleActivityTaskCommandAttr)
@@ -217,52 +217,52 @@ func (e *executor) handleActivityScheduled(event history.Event, a *history.Activ
 }
 
 func (e *executor) handleActivityCompleted(event history.Event, a *history.ActivityCompletedAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.EventID]
+	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
 	if !ok {
 		return nil
 	}
 
-	e.workflowState.removeCommandByEventID(event.EventID)
+	e.workflowState.removeCommandByEventID(event.ScheduleEventID)
 	f.Set(a.Result, nil)
 
 	return e.workflow.Continue(e.workflowCtx)
 }
 
 func (e *executor) handleActivityFailed(event history.Event, a *history.ActivityFailedAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.EventID]
+	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
 	if !ok {
 		return errors.New("no pending future found for activity failed event")
 	}
 
-	e.workflowState.removeCommandByEventID(event.EventID)
+	e.workflowState.removeCommandByEventID(event.ScheduleEventID)
 
 	f.Set(nil, errors.New(a.Reason))
 
 	return e.workflow.Continue(e.workflowCtx)
 }
 
 func (e *executor) handleTimerScheduled(event history.Event, a *history.TimerScheduledAttributes) error {
-	e.workflowState.removeCommandByEventID(event.EventID)
+	e.workflowState.removeCommandByEventID(event.ScheduleEventID)
 
 	return nil
 }
 
 func (e *executor) handleTimerFired(event history.Event, a *history.TimerFiredAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.EventID]
+	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
 	if !ok {
 		// Timer already canceled ignore
 		return nil
 	}
 
-	e.workflowState.removeCommandByEventID(event.EventID)
+	e.workflowState.removeCommandByEventID(event.ScheduleEventID)
 
 	f.Set(nil, nil)
 
 	return e.workflow.Continue(e.workflowCtx)
 }
 
 func (e *executor) handleSubWorkflowScheduled(event history.Event, a *history.SubWorkflowScheduledAttributes) error {
-	c := e.workflowState.removeCommandByEventID(event.EventID)
+	c := e.workflowState.removeCommandByEventID(event.ScheduleEventID)
 	if c != nil {
 		ca := c.Attr.(*command.ScheduleSubWorkflowCommandAttr)
 		if a.Name != ca.Name {
@@ -274,25 +274,25 @@ func (e *executor) handleSubWorkflowScheduled(event history.Event, a *history.Su
 }
 
 func (e *executor) handleSubWorkflowFailed(event history.Event, a *history.SubWorkflowFailedAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.EventID]
+	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
 	if !ok {
 		return errors.New("no pending future found for sub workflow failed event")
 	}
 
-	e.workflowState.removeCommandByEventID(event.EventID)
+	e.workflowState.removeCommandByEventID(event.ScheduleEventID)
 
 	f.Set(nil, errors.New(a.Error))
 
 	return e.workflow.Continue(e.workflowCtx)
 }
 
 func (e *executor) handleSubWorkflowCompleted(event history.Event, a *history.SubWorkflowCompletedAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.EventID]
+	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
 	if !ok {
 		return errors.New("no pending future found for sub workflow completed event")
 	}
 
-	e.workflowState.removeCommandByEventID(event.EventID)
+	e.workflowState.removeCommandByEventID(event.ScheduleEventID)
 
 	f.Set(a.Result, nil)
 
@@ -304,13 +304,13 @@ func (e *executor) handleSignalReceived(event history.Event, a *history.SignalRe
 	sc := e.workflowState.getSignalChannel(a.Name)
 	sc.SendNonblocking(e.workflowCtx, a.Arg)
 
-	e.workflowState.removeCommandByEventID(event.EventID)
+	e.workflowState.removeCommandByEventID(event.ScheduleEventID)
 
 	return e.workflow.Continue(e.workflowCtx)
 }
 
 func (e *executor) handleSideEffectResult(event history.Event, a *history.SideEffectResultAttributes) error {
-	f, ok := e.workflowState.pendingFutures[event.EventID]
+	f, ok := e.workflowState.pendingFutures[event.ScheduleEventID]
 	if !ok {
 		return errors.New("no pending future found for side effect result event")
 	}
@@ -321,8 +321,8 @@ func (e *executor) handleSideEffectResult(event history.Event, a *history.SideEf
 }
 
 func (e *executor) workflowCompleted(result payload.Payload, err error) error {
-	eventId := e.workflowState.eventID
-	e.workflowState.eventID++
+	eventId := e.workflowState.scheduleEventID
+	e.workflowState.scheduleEventID++
 
 	cmd := command.NewCompleteWorkflowCommand(eventId, result, err)
 	e.workflowState.addCommand(&cmd)
@@ -349,11 +349,11 @@ func (e *executor) processCommands(ctx context.Context, t *task.Workflow) ([]his
 			newEvents = append(newEvents, history.NewHistoryEvent(
 				e.clock.Now(),
 				history.EventType_ActivityScheduled,
-				c.ID,
 				&history.ActivityScheduledAttributes{
 					Name:   a.Name,
 					Inputs: a.Inputs,
 				},
+				history.ScheduleEventID(c.ID),
 			))
 
 		case command.CommandType_ScheduleSubWorkflow:
@@ -364,12 +364,12 @@ func (e *executor) processCommands(ctx context.Context, t *task.Workflow) ([]his
 			newEvents = append(newEvents, history.NewHistoryEvent(
 				e.clock.Now(),
 				history.EventType_SubWorkflowScheduled,
-				c.ID,
 				&history.SubWorkflowScheduledAttributes{
 					InstanceID: subWorkflowInstance.GetInstanceID(),
 					Name:       a.Name,
 					Inputs:     a.Inputs,
 				},
+				history.ScheduleEventID(c.ID),
 			))
 
 			// Send message to new workflow instance
@@ -378,11 +378,11 @@ func (e *executor) processCommands(ctx context.Context, t *task.Workflow) ([]his
 				HistoryEvent: history.NewHistoryEvent(
 					e.clock.Now(),
 					history.EventType_WorkflowExecutionStarted,
-					c.ID,
 					&history.ExecutionStartedAttributes{
 						Name:   a.Name,
 						Inputs: a.Inputs,
 					},
+					history.ScheduleEventID(c.ID),
 				),
 			})
 
@@ -391,10 +391,10 @@ func (e *executor) processCommands(ctx context.Context, t *task.Workflow) ([]his
 			newEvents = append(newEvents, history.NewHistoryEvent(
 				e.clock.Now(),
 				history.EventType_SideEffectResult,
-				c.ID,
 				&history.SideEffectResultAttributes{
 					Result: a.Result,
 				},
+				history.ScheduleEventID(c.ID),
 			))
 
 		case command.CommandType_ScheduleTimer:
@@ -403,23 +403,23 @@ func (e *executor) processCommands(ctx context.Context, t *task.Workflow) ([]his
 			newEvents = append(newEvents, history.NewHistoryEvent(
 				e.clock.Now(),
 				history.EventType_TimerScheduled,
-				c.ID,
 				&history.TimerScheduledAttributes{
 					At: a.At,
 				},
+				history.ScheduleEventID(c.ID),
 			))
 
 			// Create timer_fired event which will become visible in the future
 			workflowEvents = append(workflowEvents, core.WorkflowEvent{
 				WorkflowInstance: instance,
-				HistoryEvent: history.NewFutureHistoryEvent(
+				HistoryEvent: history.NewHistoryEvent(
 					e.clock.Now(),
 					history.EventType_TimerFired,
-					c.ID,
 					&history.TimerFiredAttributes{
 						At: a.At,
 					},
-					a.At,
+					history.ScheduleEventID(c.ID),
+					history.VisibleAt(a.At),
 				)},
 			)
 
@@ -429,11 +429,11 @@ func (e *executor) processCommands(ctx context.Context, t *task.Workflow) ([]his
 			newEvents = append(newEvents, history.NewHistoryEvent(
 				e.clock.Now(),
 				history.EventType_WorkflowExecutionFinished,
-				c.ID,
 				&history.ExecutionCompletedAttributes{
 					Result: a.Result,
 					Error:  a.Error,
 				},
+				history.ScheduleEventID(c.ID),
 			))
 
 			if instance.SubWorkflow() {
@@ -445,19 +445,21 @@ func (e *executor) processCommands(ctx context.Context, t *task.Workflow) ([]his
 					historyEvent = history.NewHistoryEvent(
 						e.clock.Now(),
 						history.EventType_SubWorkflowFailed,
-						instance.ParentEventID(), // Ensure the message gets sent back to the parent workflow with the right eventID
 						&history.SubWorkflowFailedAttributes{
 							Error: a.Error,
 						},
+						// Ensure the message gets sent back to the parent workflow with the right eventID
+						history.ScheduleEventID(instance.ParentEventID()),
 					)
 				} else {
 					historyEvent = history.NewHistoryEvent(
 						e.clock.Now(),
 						history.EventType_SubWorkflowCompleted,
-						instance.ParentEventID(), // Ensure the message gets sent back to the parent workflow with the right eventID
 						&history.SubWorkflowCompletedAttributes{
 							Result: a.Result,
 						},
+						// Ensure the message gets sent back to the parent workflow with the right eventID
+						history.ScheduleEventID(instance.ParentEventID()),
 					)
 				}