Cleanup future canceled timer messages for mysql backend

Author: cschleiden

backend/mysql/events.go

@@ -8,7 +8,7 @@ import (
 	"github.com/cschleiden/go-workflows/internal/history"
 )
 
-func insertNewEvents(ctx context.Context, tx *sql.Tx, instanceID string, newEvents []history.Event) error {
+func insertPendingEvents(ctx context.Context, tx *sql.Tx, instanceID string, newEvents []history.Event) error {
 	return insertEvents(ctx, tx, "pending_events", instanceID, newEvents)
 }
 
@@ -51,3 +51,14 @@ func insertEvents(ctx context.Context, tx *sql.Tx, tableName string, instanceID
 
 	return nil
 }
+
+func removeFutureEvent(ctx context.Context, tx *sql.Tx, instanceID string, scheduleEventID int64) error {
+	_, err := tx.ExecContext(
+		ctx,
+		"DELETE FROM `pending_events` WHERE instance_id = ? AND schedule_event_id = ? AND visible_at IS NOT NULL",
+		instanceID,
+		scheduleEventID,
+	)
+
+	return err
+}

backend/mysql/mysql.go

@@ -73,7 +73,7 @@ func (b *mysqlBackend) CreateWorkflowInstance(ctx context.Context, m history.Wor
 	}
 
 	// Initial history is empty, store only new events
-	if err := insertNewEvents(ctx, tx, m.WorkflowInstance.InstanceID, []history.Event{m.HistoryEvent}); err != nil {
+	if err := insertPendingEvents(ctx, tx, m.WorkflowInstance.InstanceID, []history.Event{m.HistoryEvent}); err != nil {
 		return fmt.Errorf("inserting new event: %w", err)
 	}
 
@@ -110,7 +110,7 @@ func (b *mysqlBackend) CancelWorkflowInstance(ctx context.Context, instance *wor
 		return err
 	}
 
-	if err := insertNewEvents(ctx, tx, instanceID, []history.Event{*event}); err != nil {
+	if err := insertPendingEvents(ctx, tx, instanceID, []history.Event{*event}); err != nil {
 		return fmt.Errorf("inserting cancellation event: %w", err)
 	}
 
@@ -252,7 +252,7 @@ func (b *mysqlBackend) SignalWorkflow(ctx context.Context, instanceID string, ev
 		return backend.ErrInstanceNotFound
 	}
 
-	if err := insertNewEvents(ctx, tx, instanceID, []history.Event{event}); err != nil {
+	if err := insertPendingEvents(ctx, tx, instanceID, []history.Event{event}); err != nil {
 		return fmt.Errorf("inserting signal event: %w", err)
 	}
 
@@ -475,6 +475,15 @@ func (b *mysqlBackend) CompleteWorkflowTask(
 		}
 	}
 
+	for _, event := range executedEvents {
+		switch event.Type {
+		case history.EventType_TimerCanceled:
+			if err := removeFutureEvent(ctx, tx, instance.InstanceID, event.ScheduleEventID); err != nil {
+				return fmt.Errorf("removing future event: %w", err)
+			}
+		}
+	}
+
 	// Insert new workflow events
 	groupedEvents := make(map[*workflow.Instance][]history.Event)
 	for _, m := range workflowEvents {
@@ -493,7 +502,7 @@ func (b *mysqlBackend) CompleteWorkflowTask(
 			}
 		}
 
-		if err := insertNewEvents(ctx, tx, targetInstance.InstanceID, events); err != nil {
+		if err := insertPendingEvents(ctx, tx, targetInstance.InstanceID, events); err != nil {
 			return fmt.Errorf("inserting messages: %w", err)
 		}
 	}
@@ -631,7 +640,7 @@ func (b *mysqlBackend) CompleteActivityTask(ctx context.Context, instance *workf
 	}
 
 	// Insert new event generated during this workflow execution
-	if err := insertNewEvents(ctx, tx, instance.InstanceID, []history.Event{event}); err != nil {
+	if err := insertPendingEvents(ctx, tx, instance.InstanceID, []history.Event{event}); err != nil {
 		return fmt.Errorf("inserting new events for completed activity: %w", err)
 	}
 

backend/mysql/mysql_test.go

@@ -100,6 +100,51 @@ func TestMySqlBackendE2E(t *testing.T) {
 var _ test.TestBackend = (*mysqlBackend)(nil)
 
 func (mb *mysqlBackend) GetFutureEvents(ctx context.Context) ([]history.Event, error) {
-	// TODO: TESTING: Implement
-	return nil, nil
+	tx, err := mb.db.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	// There is no index on `visible_at`, but this is okay for test only usage.
+	futureEvents, err := tx.QueryContext(
+		ctx,
+		"SELECT event_id, sequence_id, instance_id, event_type, timestamp, schedule_event_id, attributes, visible_at FROM `pending_events` WHERE visible_at IS NOT NULL",
+	)
+	if err != nil {
+		return nil, fmt.Errorf("getting history: %w", err)
+	}
+
+	f := make([]history.Event, 0)
+
+	for futureEvents.Next() {
+		var instanceID string
+		var attributes []byte
+
+		fe := history.Event{}
+
+		if err := futureEvents.Scan(
+			&fe.ID,
+			&fe.SequenceID,
+			&instanceID,
+			&fe.Type,
+			&fe.Timestamp,
+			&fe.ScheduleEventID,
+			&attributes,
+			&fe.VisibleAt,
+		); err != nil {
+			return nil, fmt.Errorf("scanning event: %w", err)
+		}
+
+		a, err := history.DeserializeAttributes(fe.Type, attributes)
+		if err != nil {
+			return nil, fmt.Errorf("deserializing attributes: %w", err)
+		}
+
+		fe.Attributes = a
+
+		f = append(f, fe)
+	}
+
+	return f, nil
 }

backend/mysql/schema.sql

@@ -28,7 +28,7 @@ CREATE TABLE IF NOT EXISTS `pending_events` (
   `visible_at` DATETIME NULL,
 
   INDEX `idx_pending_events_instance_id` (`instance_id`),
-  INDEX `idx_pending_events_instance_id_visible_at` (`instance_id`, `visible_at`)
+  INDEX `idx_pending_events_instance_id_visible_at_schedule_event_id` (`instance_id`, `visible_at`, `schedule_event_id`)
 );