Use int64 for event ids

cschleiden · cschleiden · commit bf9134167ac6 · 2022-04-10T19:22:26.000-07:00
diff --git a/backend/mysql/mysql.go b/backend/mysql/mysql.go
@@ -200,7 +200,7 @@ func (b *mysqlBackend) GetWorkflowInstanceState(ctx context.Context, instance *w
 
 func createInstance(ctx context.Context, tx *sql.Tx, wfi *workflow.Instance, ignoreDuplicate bool) error {
 	var parentInstanceID *string
-	var parentEventID *int
+	var parentEventID *int64
 	if wfi.SubWorkflow() {
 		i := wfi.ParentInstanceID
 		parentInstanceID = &i
@@ -291,7 +291,7 @@ func (b *mysqlBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, err
 	var id int
 	var instanceID, executionID string
 	var parentInstanceID *string
-	var parentEventID *int
+	var parentEventID *int64
 	var stickyUntil *time.Time
 	if err := row.Scan(&id, &instanceID, &executionID, &parentInstanceID, &parentEventID, &stickyUntil); err != nil {
 		if err == sql.ErrNoRows {
diff --git a/backend/sqlite/sqlite.go b/backend/sqlite/sqlite.go
@@ -89,7 +89,7 @@ func (sb *sqliteBackend) CreateWorkflowInstance(ctx context.Context, m history.W
 
 func createInstance(ctx context.Context, tx *sql.Tx, wfi *workflow.Instance, ignoreDuplicate bool) error {
 	var parentInstanceID *string
-	var parentEventID *int
+	var parentEventID *int64
 	if wfi.SubWorkflow() {
 		i := wfi.ParentInstanceID
 		parentInstanceID = &i
@@ -257,7 +257,7 @@ func (sb *sqliteBackend) GetWorkflowTask(ctx context.Context) (*task.Workflow, e
 
 	var instanceID, executionID string
 	var parentInstanceID *string
-	var parentEventID *int
+	var parentEventID *int64
 	var stickyUntil *time.Time
 	if err := row.Scan(&instanceID, &executionID, &parentInstanceID, &parentEventID, &stickyUntil); err != nil {
 		if err == sql.ErrNoRows {
diff --git a/internal/command/command.go b/internal/command/command.go
@@ -35,7 +35,7 @@ const (
 type Command struct {
 	State CommandState
 
-	ID int
+	ID int64
 
 	Type CommandType
 
@@ -47,7 +47,7 @@ type ScheduleActivityTaskCommandAttr struct {
 	Inputs []payload.Payload
 }
 
-func NewScheduleActivityTaskCommand(id int, name string, inputs []payload.Payload) Command {
+func NewScheduleActivityTaskCommand(id int64, name string, inputs []payload.Payload) Command {
 	return Command{
 		ID:   id,
 		Type: CommandType_ScheduleActivityTask,
@@ -64,7 +64,7 @@ type ScheduleSubWorkflowCommandAttr struct {
 	Inputs     []payload.Payload
 }
 
-func NewScheduleSubWorkflowCommand(id int, instanceID, name string, inputs []payload.Payload) Command {
+func NewScheduleSubWorkflowCommand(id int64, instanceID, name string, inputs []payload.Payload) Command {
 	if instanceID == "" {
 		instanceID = uuid.New().String()
 	}
@@ -84,7 +84,7 @@ type ScheduleTimerCommandAttr struct {
 	At time.Time
 }
 
-func NewScheduleTimerCommand(id int, at time.Time) Command {
+func NewScheduleTimerCommand(id int64, at time.Time) Command {
 	return Command{
 		ID:   id,
 		Type: CommandType_ScheduleTimer,
@@ -98,7 +98,7 @@ type CancelTimerCommandAttr struct {
 	TimerID int
 }
 
-func NewCancelTimerCommand(id, timerID int) Command {
+func NewCancelTimerCommand(id int64, timerID int) Command {
 	return Command{
 		ID:   id,
 		Type: CommandType_CancelTimer,
@@ -112,7 +112,7 @@ type SideEffectCommandAttr struct {
 	Result payload.Payload
 }
 
-func NewSideEffectCommand(id int, result payload.Payload) Command {
+func NewSideEffectCommand(id int64, result payload.Payload) Command {
 	return Command{
 		ID:   id,
 		Type: CommandType_SideEffect,
@@ -127,7 +127,7 @@ type CompleteWorkflowCommandAttr struct {
 	Error  string
 }
 
-func NewCompleteWorkflowCommand(id int, result payload.Payload, err error) Command {
+func NewCompleteWorkflowCommand(id int64, result payload.Payload, err error) Command {
 	var error string
 	if err != nil {
 		error = err.Error()
diff --git a/internal/core/instance.go b/internal/core/instance.go
@@ -5,7 +5,7 @@ type WorkflowInstance struct {
 	ExecutionID string `json:"execution_id,omitempty"`
 
 	ParentInstanceID string `json:"parent_instance,omitempty"`
-	ParentEventID    int    `json:"parent_event_id,omitempty"`
+	ParentEventID    int64  `json:"parent_event_id,omitempty"`
 }
 
 func NewWorkflowInstance(instanceID, executionID string) *WorkflowInstance {
@@ -15,7 +15,7 @@ func NewWorkflowInstance(instanceID, executionID string) *WorkflowInstance {
 	}
 }
 
-func NewSubWorkflowInstance(instanceID, executionID string, parentInstanceID string, parentEventID int) *WorkflowInstance {
+func NewSubWorkflowInstance(instanceID, executionID string, parentInstanceID string, parentEventID int64) *WorkflowInstance {
 	return &WorkflowInstance{
 		InstanceID:       instanceID,
 		ExecutionID:      executionID,
diff --git a/internal/history/history.go b/internal/history/history.go
@@ -46,28 +46,34 @@ func (et EventType) String() string {
 		return "WorkflowExecutionTerminated"
 	case EventType_WorkflowExecutionCanceled:
 		return "WorkflowExecutionCanceled"
+
 	case EventType_WorkflowTaskStarted:
 		return "WorkflowTaskStarted"
 	case EventType_WorkflowTaskFinished:
 		return "WorkflowTaskFinished"
+
 	case EventType_SubWorkflowScheduled:
 		return "SubWorkflowScheduled"
 	case EventType_SubWorkflowCompleted:
 		return "SubWorkflowCompleted"
 	case EventType_SubWorkflowFailed:
 		return "SubWorkflowFailed"
+
 	case EventType_ActivityScheduled:
 		return "ActivityScheduled"
 	case EventType_ActivityCompleted:
 		return "ActivityCompleted"
 	case EventType_ActivityFailed:
 		return "ActivityFailed"
+
 	case EventType_TimerScheduled:
 		return "TimerScheduled"
 	case EventType_TimerFired:
 		return "TimerFired"
+
 	case EventType_SignalReceived:
 		return "SignalReceived"
+
 	case EventType_SideEffectResult:
 		return "SideEffectResult"
 	default:
@@ -83,10 +89,12 @@ type Event struct {
 
 	Timestamp time.Time
 
+	SequenceID int64
+
 	// ScheduleEventID is used to correlate events belonging together
 	// For example, if an activity is scheduled, ScheduleEventID of the schedule event and the
 	// completion/failure event are the same.
-	ScheduleEventID int
+	ScheduleEventID int64
 
 	// Attributes are event type specific attributes
 	Attributes interface{}
@@ -100,7 +108,7 @@ func (e Event) String() string {
 
 type HistoryEventOption func(e *Event)
 
-func ScheduleEventID(scheduleEventID int) HistoryEventOption {
+func ScheduleEventID(scheduleEventID int64) HistoryEventOption {
 	return func(e *Event) {
 		e.ScheduleEventID = scheduleEventID
 	}
diff --git a/internal/workflowstate/workflowstate.go b/internal/workflowstate/workflowstate.go
@@ -39,9 +39,9 @@ type signalChannel struct {
 
 type WfState struct {
 	instance        *core.WorkflowInstance
-	scheduleEventID int
+	scheduleEventID int64
 	commands        []*command.Command
-	pendingFutures  map[int]DecodingSettable
+	pendingFutures  map[int64]DecodingSettable
 	replaying       bool
 
 	pendingSignals map[string][]payload.Payload
@@ -58,7 +58,7 @@ func NewWorkflowState(instance *core.WorkflowInstance, logger log.Logger, clock
 		instance:        instance,
 		commands:        []*command.Command{},
 		scheduleEventID: 1,
-		pendingFutures:  map[int]DecodingSettable{},
+		pendingFutures:  map[int64]DecodingSettable{},
 
 		pendingSignals: map[string][]payload.Payload{},
 		signalChannels: make(map[string]*signalChannel),
@@ -81,22 +81,22 @@ func WithWorkflowState(ctx sync.Context, wfState *WfState) sync.Context {
 	return sync.WithValue(ctx, workflowCtxKey, wfState)
 }
 
-func (wf *WfState) GetNextScheduleEventID() int {
+func (wf *WfState) GetNextScheduleEventID() int64 {
 	scheduleEventID := wf.scheduleEventID
 	wf.scheduleEventID++
 	return scheduleEventID
 }
 
-func (wf *WfState) TrackFuture(scheduleEventID int, f DecodingSettable) {
+func (wf *WfState) TrackFuture(scheduleEventID int64, f DecodingSettable) {
 	wf.pendingFutures[scheduleEventID] = f
 }
 
-func (wf *WfState) FutureByScheduleEventID(scheduleEventID int) (DecodingSettable, bool) {
+func (wf *WfState) FutureByScheduleEventID(scheduleEventID int64) (DecodingSettable, bool) {
 	f, ok := wf.pendingFutures[scheduleEventID]
 	return f, ok
 }
 
-func (wf *WfState) RemoveFuture(scheduleEventID int) {
+func (wf *WfState) RemoveFuture(scheduleEventID int64) {
 	delete(wf.pendingFutures, scheduleEventID)
 }
 
@@ -108,7 +108,7 @@ func (wf *WfState) AddCommand(cmd *command.Command) {
 	wf.commands = append(wf.commands, cmd)
 }
 
-func (wf *WfState) RemoveCommandByEventID(eventID int) *command.Command {
+func (wf *WfState) RemoveCommandByEventID(eventID int64) *command.Command {
 	for i, c := range wf.commands {
 		if c.ID == eventID {
 			wf.commands = append(wf.commands[:i], wf.commands[i+1:]...)

Original file line number	Diff line number	Diff line change
`@@ -5,7 +5,7 @@ type WorkflowInstance struct {`
`5`	`5`	ExecutionID string `json:"execution_id,omitempty"`
`6`	`6`
`7`	`7`	ParentInstanceID string `json:"parent_instance,omitempty"`
`8`		- ParentEventID int `json:"parent_event_id,omitempty"`
	`8`	+ ParentEventID int64 `json:"parent_event_id,omitempty"`
`9`	`9`	`}`
`10`	`10`
`11`	`11`	`func NewWorkflowInstance(instanceID, executionID string) *WorkflowInstance {`
`@@ -15,7 +15,7 @@ func NewWorkflowInstance(instanceID, executionID string) *WorkflowInstance {`
`15`	`15`	`}`
`16`	`16`	`}`
`17`	`17`
`18`		`-func NewSubWorkflowInstance(instanceID, executionID string, parentInstanceID string, parentEventID int) *WorkflowInstance {`
	`18`	`+func NewSubWorkflowInstance(instanceID, executionID string, parentInstanceID string, parentEventID int64) *WorkflowInstance {`
`19`	`19`	`return &WorkflowInstance{`
`20`	`20`	`InstanceID: instanceID,`
`21`	`21`	`ExecutionID: executionID,`