Merge pull request #290 from cschleiden/cleanup

Fix some linting errors

Author: cschleiden

backend/monoprocess/monoprocess.go

@@ -17,7 +17,6 @@ type monoprocessBackend struct {
 
 	workflowSignal chan struct{}
 	activitySignal chan struct{}
-	signalTimeout  time.Duration
 
 	logger *slog.Logger
 }
@@ -106,7 +105,9 @@ func (b *monoprocessBackend) CompleteWorkflowTask(
 		// Note that the worker will be notified even if the timer event gets
 		// cancelled. This is ok, because the poller will simply find no task
 		// and continue.
-		time.AfterFunc(attr.At.Sub(time.Now()), func() { b.notifyWorkflowWorker(context.Background()) })
+		time.AfterFunc(time.Until(attr.At), func() {
+			b.notifyWorkflowWorker(ctx)
+		})
 	}
 
 	b.notifyWorkflowWorker(ctx)

backend/redis/events_future.go

@@ -2,71 +2,13 @@ package redis
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"strconv"
 	"time"
 
-	"github.com/cschleiden/go-workflows/backend/history"
-	"github.com/cschleiden/go-workflows/core"
 	redis "github.com/redis/go-redis/v9"
 )
 
-// Adds an event to be delivered in the future. Not cluster-safe.
-// KEYS[1] - future event zset key
-// KEYS[2] - future event key
-// KEYS[3] - instance payload key
-// ARGV[1] - timestamp/score for set
-// ARGV[2] - Instance segment
-// ARGV[3] - event id
-// ARGV[4] - event data
-// ARGV[5] - event payload
-var addFutureEventCmd = redis.NewScript(`
-	redis.call("ZADD", KEYS[1], ARGV[1], KEYS[2])
-	redis.call("HSET", KEYS[2], "instance", ARGV[2], "id", ARGV[3], "event", ARGV[4])
-	redis.call("HSETNX", KEYS[3], ARGV[3], ARGV[5])
-	return 0
-`)
-
-func addFutureEventP(ctx context.Context, p redis.Pipeliner, instance *core.WorkflowInstance, event *history.Event) error {
-	eventData, err := marshalEventWithoutAttributes(event)
-	if err != nil {
-		return err
-	}
-
-	payloadEventData, err := json.Marshal(event.Attributes)
-	if err != nil {
-		return err
-	}
-
-	return addFutureEventCmd.Run(
-		ctx, p,
-		[]string{futureEventsKey(), futureEventKey(instance, event.ScheduleEventID), payloadKey(instance)},
-		strconv.FormatInt(event.VisibleAt.UnixMilli(), 10),
-		instanceSegment(instance),
-		event.ID,
-		string(eventData),
-		string(payloadEventData),
-	).Err()
-}
-
-// Remove a scheduled future event. Not cluster-safe.
-// KEYS[1] - future event zset key
-// KEYS[2] - future event key
-// KEYS[3] - instance payload key
-var removeFutureEventCmd = redis.NewScript(`
-	redis.call("ZREM", KEYS[1], KEYS[2])
-	local eventID = redis.call("HGET", KEYS[2], "id")
-	redis.call("HDEL", KEYS[3], eventID)
-	return redis.call("DEL", KEYS[2])
-`)
-
-// removeFutureEvent removes a scheduled future event for the given event. Events are associated via their ScheduleEventID
-func removeFutureEventP(ctx context.Context, p redis.Pipeliner, instance *core.WorkflowInstance, event *history.Event) {
-	key := futureEventKey(instance, event.ScheduleEventID)
-	removeFutureEventCmd.Run(ctx, p, []string{futureEventsKey(), key, payloadKey(instance)})
-}
-
 // Find all due future events. For each event:
 // - Look up event data
 // - Add to pending event stream for workflow instance

backend/redis/instance.go

@@ -163,7 +163,9 @@ func createInstanceP(ctx context.Context, p redis.Pipeliner, instance *core.Work
 	p.SetNX(ctx, key, string(b), 0)
 
 	// The newly created instance is going to be the active execution
-	setActiveInstanceExecutionP(ctx, p, instance)
+	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),

backend/redis/redis.go

@@ -59,12 +59,10 @@ func NewRedisBackend(client redis.UniversalClient, opts ...RedisBackendOption) (
 	// them, loads them. This doesn't work when using (transactional) pipelines, so eagerly load them on startup.
 	ctx := context.Background()
 	cmds := map[string]*redis.StringCmd{
-		"addFutureEventCmd":    addFutureEventCmd.Load(ctx, rb.rdb),
-		"futureEventsCmd":      futureEventsCmd.Load(ctx, rb.rdb),
-		"removeFutureEventCmd": removeFutureEventCmd.Load(ctx, rb.rdb),
-		"deleteInstanceCmd":    deleteCmd.Load(ctx, rb.rdb),
-		"expireInstanceCmd":    expireCmd.Load(ctx, rb.rdb),
-		"addPayloadsCmd":       addPayloadsCmd.Load(ctx, rb.rdb),
+		"futureEventsCmd":   futureEventsCmd.Load(ctx, rb.rdb),
+		"deleteInstanceCmd": deleteCmd.Load(ctx, rb.rdb),
+		"expireInstanceCmd": expireCmd.Load(ctx, rb.rdb),
+		"addPayloadsCmd":    addPayloadsCmd.Load(ctx, rb.rdb),
 	}
 	for name, cmd := range cmds {
 		// fmt.Println(name, cmd.Val())

client/client.go

@@ -67,7 +67,9 @@ func (c *Client) CreateWorkflowInstance(ctx context.Context, options WorkflowIns
 	defer span.End()
 
 	for _, propagator := range c.backend.ContextPropagators() {
-		propagator.Inject(ctx, metadata)
+		if err := propagator.Inject(ctx, metadata); err != nil {
+			return nil, fmt.Errorf("injecting context to propagate: %w", err)
+		}
 	}
 
 	startedEvent := history.NewPendingEvent(

internal/args/args_test.go

@@ -67,24 +67,14 @@ func TestInputsToArgs(t *testing.T) {
 				inputs = append(inputs, p)
 			}
 
-			args, addContext, err := InputsToArgs(converter.DefaultConverter, reflect.ValueOf(tt.args.fn), inputs)
+			_, addContext, err := InputsToArgs(converter.DefaultConverter, reflect.ValueOf(tt.args.fn), inputs)
 			if (err != nil) != tt.wantErr {
 				t.Errorf("InputsToArgs() error = %v, wantErr %v", err, tt.wantErr)
 				return
 			}
 			if tt.wantErr {
 				require.EqualError(t, err, tt.err)
 				require.Equal(t, tt.addContext, addContext)
-			} else {
-				if addContext {
-					// Skip the first argument, it will be filled with the context later
-					args = args[1:]
-				}
-
-				argValues := make([]interface{}, 0)
-				for _, arg := range args {
-					argValues = append(argValues, arg.Interface())
-				}
 			}
 		})
 	}

internal/workflow/cache/cache_test.go

@@ -24,7 +24,7 @@ func Test_Cache_StoreAndGet(t *testing.T) {
 	c := NewWorkflowExecutorLRUCache(metrics.NewNoopMetricsClient(), 1, time.Second*10)
 
 	r := wf.NewRegistry()
-	r.RegisterWorkflow(workflowWithActivity)
+	require.NoError(t, r.RegisterWorkflow(workflowWithActivity))
 
 	i := core.NewWorkflowInstance("instanceID", "executionID")
 	e, err := wf.NewExecutor(
@@ -52,7 +52,7 @@ func Test_Cache_StoreAndGet(t *testing.T) {
 	err = c.Store(context.Background(), i2, e2)
 	require.NoError(t, err)
 
-	re, ok, err = c.Get(context.Background(), i)
+	_, ok, err = c.Get(context.Background(), i)
 	require.NoError(t, err)
 	require.False(t, ok)
 }
@@ -66,7 +66,7 @@ func Test_Cache_AutoEviction(t *testing.T) {
 
 	i := core.NewWorkflowInstance("instanceID", "executionID")
 	r := wf.NewRegistry()
-	r.RegisterWorkflow(workflowWithActivity)
+	require.NoError(t, r.RegisterWorkflow(workflowWithActivity))
 	e, err := wf.NewExecutor(
 		slog.Default(), trace.NewNoopTracerProvider().Tracer(backend.TracerName), r,
 		converter.DefaultConverter, []workflow.ContextPropagator{}, &testHistoryProvider{}, i,
@@ -96,7 +96,7 @@ func Test_Cache_Evict(t *testing.T) {
 
 	i := core.NewWorkflowInstance("instanceID", "executionID")
 	r := wf.NewRegistry()
-	r.RegisterWorkflow(workflowWithActivity)
+	require.NoError(t, r.RegisterWorkflow(workflowWithActivity))
 	e, err := wf.NewExecutor(
 		slog.Default(), trace.NewNoopTracerProvider().Tracer(backend.TracerName), r,
 		converter.DefaultConverter, []workflow.ContextPropagator{}, &testHistoryProvider{}, i,

internal/workflow/executor_test.go

@@ -235,10 +235,6 @@ func Test_Executor(t *testing.T) {
 				require.False(t, e.workflow.Completed())
 				require.Len(t, e.workflowState.Commands(), 1)
 
-				h := []*history.Event{}
-				h = append(h, oldTask.NewEvents...)
-				h = append(h, taskResult.Executed...)
-
 				newTask := &backend.WorkflowTask{
 					ID:               "taskID",
 					WorkflowInstance: oldTask.WorkflowInstance,
@@ -399,7 +395,7 @@ func Test_Executor(t *testing.T) {
 					history.NewPendingEvent(time.Now(), history.EventType_ActivityCompleted, &history.ActivityCompletedAttributes{}, history.ScheduleEventID(2)),
 				}, result.Executed[len(result.Executed)-1].SequenceID)
 
-				result, err = e.ExecuteTask(context.Background(), task2)
+				_, err = e.ExecuteTask(context.Background(), task2)
 				require.NoError(t, err)
 				require.Nil(t, e.workflow.err)
 			},
@@ -573,7 +569,7 @@ func Test_Executor(t *testing.T) {
 				// Complete subworkflow
 				swr, _ := converter.DefaultConverter.To(nil)
 				hp.history = append(hp.history, result.Executed...)
-				result, err = e.ExecuteTask(context.Background(), continueTask("instanceID", []*history.Event{
+				_, err = e.ExecuteTask(context.Background(), continueTask("instanceID", []*history.Event{
 					history.NewPendingEvent(time.Now(), history.EventType_SubWorkflowCompleted, &history.SubWorkflowCompletedAttributes{
 						Result: swr,
 					}, history.ScheduleEventID(1)),

internal/workflow/registry_test.go

@@ -144,7 +144,7 @@ func Test_ActivityRegistration(t *testing.T) {
 	err = r.RegisterActivity(reg_activity, WithName("CustomName"))
 	require.NoError(t, err)
 
-	x, err = r.GetActivity("CustomName")
+	_, err = r.GetActivity("CustomName")
 	require.NoError(t, err)
 }
 

internal/workflowerrors/stack.go

@@ -30,5 +30,5 @@ func stack(skip int) string {
 		buf.WriteString(frame.String())
 	}
 
-	return string(buf.Bytes())
+	return buf.String()
 }