1
package client
2
 
3
import (
4
	"context"
5
	"errors"
6
	"os"
7
	"testing"
8
 
9
	"go.miragespace.co/specter/spec/chord"
10
	"go.miragespace.co/specter/spec/mocks"
11
	"go.miragespace.co/specter/spec/protocol"
12
	"go.miragespace.co/specter/spec/rpc"
13
 
14
	"github.com/stretchr/testify/mock"
15
	"github.com/stretchr/testify/require"
16
	"github.com/zhangyunhao116/skipmap"
17
	"go.uber.org/zap/zaptest"
18
)
19
 
20
func TestPublishPreferenceRTT(t *testing.T) {
21
	as := require.New(t)
22
	logger := zaptest.NewLogger(t)
23
 
24
	file, err := os.CreateTemp("", "client")
25
	as.NoError(err)
26
	defer os.Remove(file.Name())
27
 
28
	ctx := t.Context()
29
 
30
	token := &protocol.ClientToken{
31
		Token: []byte("test"),
32
	}
33
	cl := &protocol.Node{
34
		Id: chord.Random(),
35
	}
36
 
37
	der, cert, key := makeCertificate(as, logger, cl, token, nil)
38
	cfg := &Config{
39
		path:        file.Name(),
40
		router:      skipmap.NewString[route](),
41
		Apex:        testApex,
42
		Certificate: cert,
43
		PrivKey:     key,
44
		Tunnels: []Tunnel{
45
			{
46
				Target: "tcp://127.0.0.1:2345",
47
			},
48
		},
49
	}
50
	as.NoError(cfg.validate())
51
 
52
	m := func(s *mocks.TunnelService, t1 *mocks.MemoryTransport, publishCall *mock.Call) {
53
		defaultNoHostnames(s)
54
		transportHelper(t1, der)
55
	}
56
 
57
	client, _, assertion := setupClient(t, as, ctx, logger, nil, cfg, nil, m, true, 1)
58
	defer assertion()
59
	defer client.Close()
60
 
61
	client.Start(ctx)
62
}
63
 
64
func TestReconciliationRecoversFromCompleteOutage(t *testing.T) {
65
	c, tunnelClient := newOutcomeTestClient(t, []Tunnel{{
66
		Hostname: "configured",
67
		Target:   "http://localhost:8080",
68
	}})
69
	original := c.getConnectedNodes()
70
	tunnelClient.TunnelService.On("RegisteredHostnames", mock.Anything, mock.Anything).
71
		Return(&protocol.RegisteredHostnamesResponse{}, nil).Once()
72
	tunnelClient.TunnelService.On("PublishTunnel", mock.Anything, mock.Anything).
73
		Return(&protocol.PublishTunnelResponse{Published: original}, nil).Once()
74
	require.Empty(t, c.SyncConfigTunnels(t.Context()).Error)
75
	saved, err := os.ReadFile(c.Configuration.path)
76
	require.NoError(t, err)
77
 
78
	gateway := mock.MatchedBy(func(ctx context.Context) bool {
79
		return rpc.GetNode(ctx).GetAddress() == original[0].GetAddress()
80
	})
81
	apex := mock.MatchedBy(func(ctx context.Context) bool {
82
		return rpc.GetNode(ctx).GetAddress() == c.Configuration.Apex
83
	})
84
	tunnelClient.TunnelService.On("Ping", gateway, mock.Anything).
85
		Return(nil, errors.New("gateway disconnected")).Once()
86
	tunnelClient.TunnelService.On("Ping", apex, mock.Anything).
87
		Return(nil, errors.New("apex temporarily unavailable")).Once()
88
 
89
	// Losing every gateway clears prior publication acknowledgements even when
90
	// the first bootstrap fails. Recovery remains pending for a later pass.
91
	c.reconcileConnections(t.Context())
92
	require.Empty(t, c.getConnectedNodes())
93
	require.True(t, c.getStatus().Pending)
94
	require.False(t, c.getStatus().Synchronization.Tunnels[0].Published)
95
	tunnelClient.TunnelService.AssertNumberOfCalls(t, "PublishTunnel", 1)
96
 
97
	peer := &protocol.Node{
98
		Id:      2,
99
		Address: "peer.example.com",
100
	}
101
	recovered := []*protocol.Node{original[0], peer}
102
	tunnelClient.TunnelService.On("Ping", apex, mock.Anything).
103
		Return(&protocol.ClientPingResponse{Node: original[0]}, nil).Once()
104
	tunnelClient.TunnelService.On("Ping", gateway, mock.Anything).
105
		Return(&protocol.ClientPingResponse{Node: original[0]}, nil).Once()
106
	tunnelClient.TunnelService.On("Ping", mock.MatchedBy(func(ctx context.Context) bool {
107
		return rpc.GetNode(ctx).GetAddress() == peer.GetAddress()
108
	}), mock.Anything).Return(&protocol.ClientPingResponse{Node: peer}, nil).Twice()
109
	tunnelClient.TunnelService.On("GetNodes", gateway, mock.Anything).
110
		Return(&protocol.GetNodesResponse{Nodes: recovered}, nil).Once()
111
	tunnelClient.TunnelService.On("PublishTunnel", mock.Anything, mock.MatchedBy(func(req *protocol.PublishTunnelRequest) bool {
112
		return req.GetHostname() == "configured" && endpointSignature(req.GetServers()) == endpointSignature(recovered)
113
	})).Return(&protocol.PublishTunnelResponse{Published: recovered}, nil).Once()
114
 
115
	// One pass bootstraps, discovers another peer, and republishes without a
116
	// separate worker or a delay between those steps.
117
	c.reconcileConnections(t.Context())
118
	require.Equal(t, endpointSignature(recovered), endpointSignature(c.getConnectedNodes()))
119
	status := c.getStatus()
120
	require.False(t, status.Pending)
121
	require.Empty(t, status.Synchronization.Error)
122
	require.Equal(t, 2, status.Synchronization.Tunnels[0].PublishedEndpoints)
123
	onDisk, err := os.ReadFile(c.Configuration.path)
124
	require.NoError(t, err)
125
	require.Equal(t, saved, onDisk, "reconnection must not rewrite the configuration")
126
}