| 1050 | } |
| 1051 | |
| 1052 | func TestServiceRunSerializesConcurrentCalls(t *testing.T) { |
| 1053 | t.Parallel() |
| 1054 | |
| 1055 | lock := &stubLock{ |
| 1056 | tryAcquireFn: func() (time.Time, bool, error) { |
| 1057 | return time.Time{}, true, nil |
| 1058 | }, |
| 1059 | } |
| 1060 | service := NewService(withLock(lock)) |
| 1061 | |
| 1062 | var active atomic.Int32 |
| 1063 | var maxActive atomic.Int32 |
| 1064 | started := make(chan struct{}, 2) |
| 1065 | releaseFirst := make(chan struct{}) |
| 1066 | errCh := make(chan error, 2) |
| 1067 | var first sync.Once |
| 1068 | |
| 1069 | spawner := func(context.Context, string, string, string, time.Time) error { |
| 1070 | current := active.Add(1) |
| 1071 | for { |
| 1072 | previous := maxActive.Load() |
| 1073 | if current <= previous || maxActive.CompareAndSwap(previous, current) { |
| 1074 | break |
| 1075 | } |
| 1076 | } |
| 1077 | |
| 1078 | started <- struct{}{} |
| 1079 | shouldBlock := false |
| 1080 | first.Do(func() { shouldBlock = true }) |
| 1081 | if shouldBlock { |
| 1082 | <-releaseFirst |
| 1083 | } |
| 1084 | |
| 1085 | active.Add(-1) |
| 1086 | return nil |
| 1087 | } |
| 1088 | |
| 1089 | go func() { |
| 1090 | errCh <- service.Run(testutil.Context(t), spawner, "") |
| 1091 | }() |
| 1092 | <-started |
| 1093 | |
| 1094 | go func() { |
| 1095 | errCh <- service.Run(testutil.Context(t), spawner, "") |
| 1096 | }() |
| 1097 | |
| 1098 | select { |
| 1099 | case <-started: |
| 1100 | t.Fatal("second spawner started before first run finished") |
| 1101 | case <-time.After(150 * time.Millisecond): |
| 1102 | } |
| 1103 | |
| 1104 | close(releaseFirst) |
| 1105 | <-started |
| 1106 | |
| 1107 | for range 2 { |
| 1108 | if err := <-errCh; err != nil { |
| 1109 | t.Fatalf("Run() error = %v", err) |