1
package chord
2
 
3
import (
4
	"bytes"
5
	"context"
6
	"fmt"
7
	"net"
8
	"net/http"
9
	"testing"
10
	"time"
11
 
12
	"go.miragespace.co/specter/kv/memory"
13
	"go.miragespace.co/specter/spec/chord"
14
	"go.miragespace.co/specter/spec/mocks"
15
	"go.miragespace.co/specter/spec/protocol"
16
	"go.miragespace.co/specter/spec/rpc"
17
	"go.miragespace.co/specter/spec/transport"
18
	"go.miragespace.co/specter/util/acceptor"
19
 
20
	"github.com/go-chi/chi/v5"
21
	"github.com/stretchr/testify/mock"
22
	"github.com/stretchr/testify/require"
23
	"go.uber.org/zap"
24
	"go.uber.org/zap/zaptest"
25
)
26
 
27
func TestRPC(t *testing.T) {
28
	as := require.New(t)
29
 
30
	logger := zaptest.NewLogger(t, zaptest.WrapOptions(zap.AddCaller()))
31
 
32
	ctx := t.Context()
33
 
34
	t1, t2 := mocks.PipeTransport()
35
 
36
	client1 := rpc.DynamicChordClient(rpc.DisablePooling(ctx), t1)
37
	identity1 := &protocol.Node{
38
		Id:      chord.Random(),
39
		Address: "127.0.0.1:1234",
40
	}
41
	node1 := NewLocalNode(NodeConfig{
42
		BaseLogger:               logger,
43
		Identity:                 identity1,
44
		KVProvider:               memory.WithHashFn(chord.Hash),
45
		FixFingerInterval:        time.Millisecond * 100,
46
		StabilizeInterval:        time.Millisecond * 300,
47
		PredecessorCheckInterval: time.Millisecond * 500,
48
		ChordClient:              client1,
49
		NodesRTT:                 new(mocks.Measurement),
50
	})
51
 
52
	streamRouter1 := transport.NewStreamRouter(logger, t1, nil)
53
	node1.AttachRouter(ctx, streamRouter1)
54
	go streamRouter1.Accept(ctx)
55
 
56
	as.NoError(node1.Create())
57
	defer node1.Leave()
58
 
59
	client2 := rpc.DynamicChordClient(rpc.DisablePooling(ctx), t2)
60
	caller, err := NewRemoteNode(ctx, logger, client2, identity1)
61
	as.NoError(err)
62
 
63
	as.NoError(caller.Ping())
64
 
65
	resp1, err := caller.FindSuccessor(chord.Random())
66
	as.NoError(err)
67
	as.Equal(identity1.GetId(), resp1.ID())
68
 
69
	resp2, err := caller.GetSuccessors()
70
	as.NoError(err)
71
	as.Len(resp2, 1)
72
 
73
	resp3, err := caller.GetPredecessor()
74
	as.NoError(err)
75
	as.NotNil(resp3)
76
	as.Equal(identity1.GetId(), resp3.ID())
77
 
78
	err = caller.Put(ctx, []byte("k"), []byte("v"))
79
	as.NoError(err)
80
 
81
	resp4, err := caller.Get(ctx, []byte("k"))
82
	as.NoError(err)
83
	as.Len(resp4, 1)
84
 
85
	err = caller.Import(ctx, [][]byte{[]byte("y")}, []*protocol.KVTransfer{
86
		{
87
			SimpleValue:    []byte("v"),
88
			PrefixChildren: [][]byte{[]byte("c")},
89
		},
90
	})
91
	as.NoError(err)
92
 
93
	err = caller.Delete(ctx, []byte("y"))
94
	as.NoError(err)
95
 
96
	err = caller.PrefixAppend(ctx, []byte("p"), []byte("c"))
97
	as.NoError(err)
98
 
99
	resp5, err := caller.PrefixList(ctx, []byte("p"))
100
	as.NoError(err)
101
	as.Len(resp5, 1)
102
 
103
	err = caller.PrefixRemove(ctx, []byte("p"), []byte("c"))
104
	as.NoError(err)
105
 
106
	resp6, err := caller.PrefixContains(ctx, []byte("p"), []byte("c"))
107
	as.NoError(err)
108
	as.False(resp6)
109
 
110
	resp7, err := caller.ListKeys(ctx, []byte(""))
111
	as.NoError(err)
112
	as.Greater(len(resp7), 0)
113
 
114
	prevToken, err := caller.Acquire(ctx, []byte("lease"), time.Second)
115
	as.NoError(err)
116
 
117
	newToken, err := caller.Renew(ctx, []byte("lease"), time.Minute, prevToken)
118
	as.NoError(err)
119
 
120
	as.NotEqual(newToken, prevToken)
121
 
122
	err = caller.Release(ctx, []byte("lease"), newToken)
123
	as.NoError(err)
124
 
125
	err = caller.Release(ctx, []byte("lease"), prevToken)
126
	as.Error(err)
127
}
128
 
129
func TestRemoteRPCContext(t *testing.T) {
130
	as := require.New(t)
131
	logger := zaptest.NewLogger(t, zaptest.WrapOptions(zap.AddCaller()))
132
 
133
	e := fmt.Errorf("sup")
134
	key := []byte("key")
135
	val := []byte("val")
136
 
137
	ctx := t.Context()
138
 
139
	m := new(mocks.ChordClient)
140
	m.On("Put", mock.MatchedBy(func(ctx context.Context) bool {
141
		r := rpc.GetContext(ctx)
142
		return r.GetRequestTarget() == protocol.Context_KV_REPLICATION
143
	}), mock.MatchedBy(func(req *protocol.SimpleRequest) bool {
144
		return bytes.Equal(req.GetKey(), key) && bytes.Equal(req.GetValue(), val)
145
	})).Return(nil, e)
146
 
147
	nsTwirp := protocol.NewVNodeServiceServer(m)
148
	ksTwirp := protocol.NewKVServiceServer(m)
149
 
150
	rpcHandler := chi.NewRouter()
151
	rpcHandler.Mount(nsTwirp.PathPrefix(), rpc.ExtractContext(nsTwirp))
152
	rpcHandler.Mount(ksTwirp.PathPrefix(), rpc.ExtractContext(ksTwirp))
153
 
154
	tp := mocks.SelfTransport()
155
 
156
	srv := &http.Server{
157
		BaseContext: func(l net.Listener) context.Context {
158
			return ctx
159
		},
160
		ReadTimeout: rpcTimeout,
161
		Handler:     rpcHandler,
162
	}
163
 
164
	acceptor := acceptor.NewH2Acceptor(nil)
165
	defer acceptor.Close()
166
 
167
	defer srv.Shutdown(ctx)
168
 
169
	go srv.Serve(acceptor)
170
 
171
	streamRouter := transport.NewStreamRouter(logger, tp, nil)
172
	go streamRouter.Accept(ctx)
173
 
174
	// yes, setting up RPC is THAT complicated
175
 
176
	peer := &protocol.Node{
177
		Id:      1234,
178
		Address: "127.0.0.1:1234",
179
	}
180
 
181
	streamRouter.HandleChord(protocol.Stream_RPC, peer, func(delegate *transport.StreamDelegate) {
182
		acceptor.Handle(delegate)
183
	})
184
 
185
	rpcCaller := rpc.DynamicChordClient(ctx, tp)
186
	r, err := NewRemoteNode(ctx, logger, rpcCaller, peer)
187
	as.NoError(err)
188
 
189
	err = r.Put(rpc.WithContext(ctx, &protocol.Context{
190
		RequestTarget: protocol.Context_KV_REPLICATION,
191
	}), key, val)
192
	as.ErrorContains(err, e.Error())
193
 
194
	m.AssertExpectations(t)
195
}