1
package chord
2
 
3
import (
4
	"bytes"
5
	"context"
6
	"fmt"
7
	"net/http"
8
	"net/http/httptest"
9
	"sort"
10
	"strings"
11
	"testing"
12
	"time"
13
 
14
	"go.miragespace.co/specter/kv/memory"
15
	"go.miragespace.co/specter/spec/chord"
16
	"go.miragespace.co/specter/spec/mocks"
17
	"go.miragespace.co/specter/spec/protocol"
18
	"go.miragespace.co/specter/spec/rtt"
19
	"go.miragespace.co/specter/util/testcond"
20
 
21
	"github.com/stretchr/testify/mock"
22
	"github.com/stretchr/testify/require"
23
	"go.uber.org/goleak"
24
	"go.uber.org/zap"
25
	"go.uber.org/zap/zaptest"
26
)
27
 
28
const (
29
	defaultInterval = time.Millisecond * 1
30
	waitInterval    = defaultInterval * 10
31
)
32
 
33
func devConfig(t *testing.T, as *require.Assertions) NodeConfig {
34
	logger := zaptest.NewLogger(t, zaptest.WrapOptions(zap.AddCaller()))
35
	iden := &protocol.Node{
36
		Id: chord.Random(),
37
	}
38
	m := new(mocks.Measurement)
39
	m.On("Snapshot", mock.Anything, mock.Anything).Return(&rtt.Statistics{})
40
 
41
	as.True(true) // placeholder
42
 
43
	return NodeConfig{
44
		BaseLogger:               logger,
45
		ChordClient:              new(mocks.ChordClient),
46
		Identity:                 iden,
47
		KVProvider:               memory.WithHashFn(chord.Hash),
48
		StabilizeInterval:        defaultInterval * 3,
49
		FixFingerInterval:        defaultInterval * 5,
50
		PredecessorCheckInterval: defaultInterval * 7,
51
		NodesRTT:                 m,
52
	}
53
}
54
 
55
func waitRing(as *require.Assertions, node *LocalNode) {
56
	as.NoError(testcond.WaitForCondition(func() bool {
57
		ring := node.ringTrace()
58
		if !strings.HasSuffix(ring, "error") && ring != "unstable" && node.getPredecessor() != nil {
59
			return true
60
		}
61
		return false
62
	}, waitInterval, time.Second*5))
63
}
64
 
65
// it looks like a race condition in macos runner but it is impossible to be a race condition
66
// -- famous last words
67
func waitRingLong(as *require.Assertions, nodes []*LocalNode) {
68
	as.NoError(testcond.WaitForCondition(func() bool {
69
		for _, node := range nodes {
70
			if node.getPredecessor() == nil {
71
				return false
72
			}
73
		}
74
		return true
75
	}, waitInterval, time.Second*5))
76
}
77
 
78
func makeRing(t *testing.T, as *require.Assertions, num int) ([]*LocalNode, func()) {
79
	nodes := make([]*LocalNode, num)
80
	for i := range num {
81
		node := NewLocalNode(devConfig(t, as))
82
		nodes[i] = node
83
	}
84
 
85
	nodes[0].Create()
86
	for i := 1; i < num; i++ {
87
		as.NoError(nodes[i].Join(nodes[0]))
88
		<-time.After(waitInterval)
89
	}
90
 
91
	// wait until the ring is mostly stablized before we ring check
92
	waitRing(as, nodes[0])
93
	waitRingLong(as, nodes)
94
 
95
	RingCheck(t, as, nodes, true)
96
 
97
	return nodes, func() {
98
		for i := range num {
99
			nodes[i].Leave()
100
		}
101
	}
102
}
103
 
104
// should not be called after any of the nodes stopped
105
func RingCheck(t *testing.T, as *require.Assertions, nodes []*LocalNode, counter bool) {
106
	if len(nodes) == 0 {
107
		return
108
	}
109
	for _, node := range nodes {
110
		as.NotNil(node.getPredecessor(), "node %d has nil predecessor", node.ID())
111
		as.NotNil(node.getSuccessor())
112
	}
113
 
114
	t.Logf("Ring: %s\n", nodes[0].ringTrace())
115
 
116
	if len(nodes) == 1 {
117
		as.Equal(nodes[0].ID(), nodes[0].getPredecessor().ID())
118
		as.Equal(nodes[0].ID(), nodes[0].getSuccessor().ID())
119
		return
120
	}
121
	sort.SliceStable(nodes, func(i, j int) bool {
122
		return nodes[i].ID() < nodes[j].ID()
123
	})
124
	if counter {
125
		// counter clockwise
126
		for i := 0; i < len(nodes)-1; i++ {
127
			as.Equal(nodes[i].ID(), nodes[i+1].getPredecessor().ID())
128
		}
129
		as.Equal(nodes[len(nodes)-1].ID(), nodes[0].getPredecessor().ID())
130
	}
131
	// clockwise
132
	for i := 0; i < len(nodes)-1; i++ {
133
		as.Equal(nodes[i+1].ID(), nodes[i].getSuccessor().ID())
134
	}
135
	as.Equal(nodes[0].ID(), nodes[len(nodes)-1].getSuccessor().ID())
136
}
137
 
138
func TestMain(m *testing.M) {
139
	goleak.VerifyTestMain(m)
140
}
141
 
142
func TestCreate(t *testing.T) {
143
	as := require.New(t)
144
 
145
	n1 := NewLocalNode(devConfig(t, as))
146
	n1.Create()
147
 
148
	<-time.After(waitInterval)
149
 
150
	n1.Leave()
151
 
152
	<-time.After(waitInterval)
153
 
154
	RingCheck(t, as, []*LocalNode{n1}, true)
155
}
156
 
157
func TestJoin(t *testing.T) {
158
	as := require.New(t)
159
 
160
	n2 := NewLocalNode(devConfig(t, as))
161
	n2.Create()
162
	defer n2.Leave()
163
 
164
	n1 := NewLocalNode(devConfig(t, as))
165
	as.NoError(n1.Join(n2))
166
	defer n1.Leave()
167
 
168
	waitRing(as, n2)
169
	waitRingLong(as, []*LocalNode{n1, n2})
170
 
171
	RingCheck(t, as, []*LocalNode{
172
		n1,
173
		n2,
174
	}, true)
175
}
176
 
177
func TestRandomNodes(t *testing.T) {
178
	as := require.New(t)
179
 
180
	num := 8
181
	nodes, done := makeRing(t, as, num)
182
	defer done()
183
 
184
	for i := range num {
185
		nodes[i].fingers[1].computeView(func(node chord.VNode) {
186
			as.Equal(nodes[i].getSuccessor().ID(), node.ID())
187
		})
188
		fmt.Printf("%d: %v\n---\n", nodes[i].ID(), nodes[i].fingerTrace())
189
	}
190
}
191
 
192
func TestLotsOfNodes(t *testing.T) {
193
	if testing.Short() {
194
		t.Skip("skipping many nodes in short mode")
195
	}
196
	as := require.New(t)
197
 
198
	num := 64
199
	nodes, done := makeRing(t, as, num)
200
	defer done()
201
 
202
	for i := range num {
203
		nodes[i].fingers[1].computeView(func(node chord.VNode) {
204
			as.Equal(nodes[i].getSuccessor().ID(), node.ID())
205
		})
206
		fmt.Printf("%d: %v\n---\n", nodes[i].ID(), nodes[i].fingerTrace())
207
	}
208
}
209
 
210
func TestStatsSummaryHandler(t *testing.T) {
211
	as := require.New(t)
212
 
213
	node := NewLocalNode(devConfig(t, as))
214
	as.NoError(node.Create())
215
	defer node.Leave()
216
 
217
	testKey := "helloworld"
218
	node.kv.Put(context.Background(), []byte(testKey), []byte("bye"))
219
 
220
	rr := httptest.NewRecorder()
221
	handler := http.HandlerFunc(statsHandler([]*LocalNode{node}))
222
 
223
	req, err := http.NewRequest("GET", "/", nil)
224
	as.NoError(err)
225
	handler.ServeHTTP(rr, req)
226
 
227
	resp := rr.Result()
228
	defer resp.Body.Close()
229
 
230
	as.Equal(http.StatusOK, resp.StatusCode)
231
 
232
	var body bytes.Buffer
233
	body.ReadFrom(resp.Body)
234
 
235
	as.Contains(body.String(), "Active")
236
	as.Contains(body.String(), testKey)
237
}
238
 
239
func TestStatsKeyHandler(t *testing.T) {
240
	as := require.New(t)
241
 
242
	node := NewLocalNode(devConfig(t, as))
243
	as.NoError(node.Create())
244
	defer node.Leave()
245
 
246
	testKey := "helloworld"
247
	node.kv.Put(context.Background(), []byte(testKey), []byte("hello"))
248
 
249
	rr := httptest.NewRecorder()
250
	handler := http.HandlerFunc(statsHandler([]*LocalNode{node}))
251
 
252
	req, err := http.NewRequest("GET", fmt.Sprintf("/?key=%s", testKey), nil)
253
	as.NoError(err)
254
	handler.ServeHTTP(rr, req)
255
 
256
	resp := rr.Result()
257
	defer resp.Body.Close()
258
 
259
	as.Equal(http.StatusOK, resp.StatusCode)
260
 
261
	var body bytes.Buffer
262
	body.ReadFrom(resp.Body)
263
 
264
	as.Equal("hello", body.String())
265
}