flow/command_key_conflict_test.go at master · goware/flow · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
package flow

import (
	"context"
	"testing"
	"time"

	"github.com/goware/flow/internal/testpg"
)

type crossDecisionArgs struct {
	Value string `json:"value"`
}

func TestCrossDecisionCommandKeyReuseIsAConflict(t *testing.T) {
	t.Parallel()
	for _, test := range []struct {
		name   string
		values [2]string
	}{
		{name: "equivalent", values: [2]string{"same", "same"}},
		{name: "different", values: [2]string{"first", "second"}},
	} {
		t.Run(test.name, func(t *testing.T) {
			database := testpg.Open(t)
			ctx := context.Background()
			if err := Migrate(ctx, database.DB, WithSchema(database.Schema)); err != nil {
				t.Fatal(err)
			}
			root := DefineCommand[None, None]("key_conflict.root."+test.name, 1)
			declarer := DefineCommand[crossDecisionArgs, None]("key_conflict.declarer."+test.name, 1)
			child := DefineCommand[crossDecisionArgs, None]("key_conflict.child."+test.name, 1)
			runtime, err := New(database.DB, WithSchema(database.Schema), WithWorkerConcurrency(1),
				WithNotifications(false), WithPollInterval(5*time.Millisecond))
			if err != nil {
				t.Fatal(err)
			}
			if err := runtime.Register(
				Handle(root, func(_ context.Context, work *Work[None]) (None, error) {
					Enqueue(work, "declarer/1", declarer, crossDecisionArgs{Value: test.values[0]})
					Enqueue(work, "declarer/2", declarer, crossDecisionArgs{Value: test.values[1]})
					return None{}, nil
				}),
				Handle(declarer, func(_ context.Context, work *Work[crossDecisionArgs]) (None, error) {
					Enqueue(work, "shared", child, work.Args)
					return None{}, nil
				}),
				Handle(child, func(context.Context, *Work[crossDecisionArgs]) (None, error) {
					return None{}, nil
				}),
			); err != nil {
				t.Fatal(err)
			}
			cancel, runResult := startRuntime(t, runtime)
			defer stopRuntime(t, cancel, runResult)
			exec, err := root.Enqueue(ctx, runtime, "key-conflict/"+test.name, None{})
			if err != nil {
				t.Fatal(err)
			}
			waitForRunStatus(t, database.Schema, database.DB.Conn, exec.RunID, "failed", 5*time.Second)
			trace, err := Trace(ctx, runtime, exec.RunID)
			if err != nil {
				t.Fatal(err)
			}
			var succeeded, conflicted int
			for _, command := range trace.Commands {
				if command.Name != declarer.Name() {
					continue
				}
				switch {
				case command.Status == CommandStatusSucceeded:
					succeeded++
				case command.Status == CommandStatusFailed && command.Failure != nil && command.Failure.Code == "invalid_decision":
					conflicted++
				}
			}
			if len(trace.Commands) != 4 || succeeded != 1 || conflicted != 1 {
				t.Fatalf("cross-decision trace=%+v", trace.Commands)
			}
		})
	}
}