ev-node/pkg/sync/syncer_status_test.go at 571f381f9896d2d3200d8b00024aff67ead6bae8 · stringsbuilder/ev-node · 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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
package sync

import (
	"errors"
	"sync"
	"sync/atomic"
	"testing"

	"github.com/stretchr/testify/require"
)

func TestSyncerStatusStartOnce(t *testing.T) {
	t.Parallel()

	specs := map[string]struct {
		run func(*testing.T, *SyncerStatus)
	}{
		"concurrent_start_only_runs_once": {
			run: func(t *testing.T, status *SyncerStatus) {
				t.Helper()

				var calls atomic.Int32
				started := make(chan struct{})
				release := make(chan struct{})
				var wg sync.WaitGroup

				for range 8 {
					wg.Go(func() {
						_, err := status.startOnce(func() error {
							if calls.Add(1) == 1 {
								close(started)
							}
							<-release
							return nil
						})
						require.NoError(t, err)
					})
				}

				<-started
				close(release)
				wg.Wait()

				require.Equal(t, int32(1), calls.Load())
				require.True(t, status.isStarted())
			},
		},
		"failed_start_can_retry": {
			run: func(t *testing.T, status *SyncerStatus) {
				t.Helper()

				var calls atomic.Int32
				errBoom := errors.New("boom")

				startedNow, err := status.startOnce(func() error {
					calls.Add(1)
					return errBoom
				})
				require.ErrorIs(t, err, errBoom)
				require.False(t, startedNow)
				require.False(t, status.isStarted())

				startedNow, err = status.startOnce(func() error {
					calls.Add(1)
					return nil
				})
				require.NoError(t, err)
				require.True(t, startedNow)
				require.True(t, status.isStarted())
				require.Equal(t, int32(2), calls.Load())
			},
		},
	}

	for name, spec := range specs {
		t.Run(name, func(t *testing.T) {
			t.Parallel()
			spec.run(t, &SyncerStatus{})
		})
	}
}

func TestSyncerStatusStopIfStarted(t *testing.T) {
	t.Parallel()

	specs := map[string]struct {
		started bool
		wantErr bool
	}{
		"no_op_when_not_started": {
			started: false,
			wantErr: false,
		},
		"stop_clears_started": {
			started: true,
			wantErr: false,
		},
	}

	for name, spec := range specs {
		t.Run(name, func(t *testing.T) {
			t.Parallel()

			status := &SyncerStatus{started: spec.started}
			var stopCalls atomic.Int32

			err := status.stopIfStarted(func() error {
				stopCalls.Add(1)
				return nil
			})

			if spec.wantErr {
				require.Error(t, err)
			} else {
				require.NoError(t, err)
			}

			if spec.started {
				require.Equal(t, int32(1), stopCalls.Load())
			} else {
				require.Zero(t, stopCalls.Load())
			}
			require.False(t, status.isStarted())
		})
	}
}