Small cleanup

Author: cschleiden

backend/redis/instance.go

@@ -164,38 +164,6 @@ type instanceState struct {
 	LastSequenceID int64 `json:"last_sequence_id,omitempty"`
 }
 
-func createInstanceP(ctx context.Context, p redis.Pipeliner, instance *core.WorkflowInstance, metadata *metadata.WorkflowMetadata) error {
-	key := instanceKey(instance)
-
-	createdAt := time.Now()
-
-	b, err := json.Marshal(&instanceState{
-		Instance:  instance,
-		State:     core.WorkflowInstanceStateActive,
-		Metadata:  metadata,
-		CreatedAt: createdAt,
-	})
-	if err != nil {
-		return fmt.Errorf("marshaling instance state: %w", err)
-	}
-
-	p.SetNX(ctx, key, string(b), 0)
-
-	// The newly created instance is going to be the active execution
-	if err := setActiveInstanceExecutionP(ctx, p, instance); err != nil {
-		return fmt.Errorf("setting active instance execution: %w", err)
-	}
-
-	p.ZAdd(ctx, instancesByCreation(), redis.Z{
-		Member: instanceSegment(instance),
-		Score:  float64(createdAt.UnixMilli()),
-	})
-
-	p.SAdd(ctx, instancesActive(), instanceSegment(instance))
-
-	return nil
-}
-
 func readInstance(ctx context.Context, rdb redis.UniversalClient, instanceKey string) (*instanceState, error) {
 	p := rdb.Pipeline()
 
@@ -246,14 +214,3 @@ func readActiveInstanceExecution(ctx context.Context, rdb redis.UniversalClient,
 
 	return instance, nil
 }
-
-func setActiveInstanceExecutionP(ctx context.Context, p redis.Pipeliner, instance *core.WorkflowInstance) error {
-	key := activeInstanceExecutionKey(instance.InstanceID)
-
-	b, err := json.Marshal(instance)
-	if err != nil {
-		return fmt.Errorf("marshaling instance: %w", err)
-	}
-
-	return p.Set(ctx, key, string(b), 0).Err()
-}

internal/worker/worker.go

@@ -135,7 +135,9 @@ func (w *Worker[Task, TaskResult]) dispatcher() {
 
 			// Create new context to allow tasks to complete when root context is canceled
 			taskCtx := context.Background()
-			w.handle(taskCtx, t)
+			if err := w.handle(taskCtx, t); err != nil {
+				w.logger.ErrorContext(taskCtx, "error handling task", "error", err)
+			}
 
 			if sem != nil {
 				<-sem

internal/workflow/executor.go

@@ -659,13 +659,11 @@ func (e *executor) handleSideEffectResult(event *history.Event, a *history.SideE
 	return e.workflow.Continue()
 }
 
-func (e *executor) workflowCompleted(result payload.Payload, wfErr error) error {
+func (e *executor) workflowCompleted(result payload.Payload, wfErr error) {
 	eventId := e.workflowState.GetNextScheduleEventID()
 
 	cmd := command.NewCompleteWorkflowCommand(eventId, e.workflowState.Instance(), result, workflowerrors.FromError(wfErr))
 	e.workflowState.AddCommand(cmd)
-
-	return nil
 }
 
 func (e *executor) workflowRestarted(result payload.Payload, continueAsNew *continueasnew.Error) {