1
package client
2
 
3
import (
4
	"context"
5
	"fmt"
6
	"sort"
7
	"time"
8
 
9
	"go.miragespace.co/specter/spec/chord"
10
	"go.miragespace.co/specter/spec/protocol"
11
	"go.miragespace.co/specter/spec/rpc"
12
	"go.miragespace.co/specter/spec/rtt"
13
	"go.miragespace.co/specter/spec/tun"
14
 
15
	"github.com/avast/retry-go/v5"
16
	"go.uber.org/zap"
17
)
18
 
19
func (c *Client) bootstrap(ctx context.Context, apex string) error {
20
	c.Logger.Info("Bootstraping connection to specter server", zap.String("addr", apex))
21
	return c.openRPC(ctx, &protocol.Node{
22
		Address: apex,
23
	})
24
}
25
 
26
func (c *Client) openRPC(ctx context.Context, node *protocol.Node) error {
27
	if _, ok := c.connections.Load(node.GetAddress()); ok {
28
		return nil
29
	}
30
 
31
	callCtx, cancel := context.WithTimeout(ctx, connectTimeout)
32
	defer cancel()
33
 
34
	resp, err := c.ping(callCtx, node)
35
	if err != nil {
36
		return err
37
	}
38
 
39
	identity := resp.GetNode()
40
	c.Logger.Info("Connected to specter server", zap.String("addr", identity.GetAddress()))
41
	c.connections.Store(identity.GetAddress(), identity)
42
 
43
	return nil
44
}
45
 
46
func retryRPC[V any](c *Client, ctx context.Context, fn func(node *protocol.Node) (V, error)) (resp V, err error) {
47
	candidates := c.getConnectedNodes()
48
	retrier := retry.New(
49
		retry.Context(ctx),
50
		retry.Attempts(2),
51
		retry.LastErrorOnly(true),
52
		retry.Delay(time.Millisecond*500),
53
		retry.RetryIf(chord.ErrorIsRetryable),
54
	)
55
	err = retrier.Do(func() error {
56
		var (
57
			candidate *protocol.Node
58
			rpcError  error
59
		)
60
		if len(candidates) > 0 {
61
			candidate, candidates = candidates[0], candidates[1:]
62
		}
63
		if candidate == nil {
64
			return fmt.Errorf("no rpc candidates available")
65
		}
66
		resp, rpcError = fn(candidate)
67
		return chord.ErrorMapper(rpcError)
68
	})
69
	return
70
}
71
 
72
func (c *Client) getConnectedNodes() (nodes []*protocol.Node) {
73
	c.connections.Range(func(_ string, node *protocol.Node) bool {
74
		if len(nodes) < tun.NumRedundantLinks {
75
			nodes = append(nodes, node)
76
		}
77
		return true
78
	})
79
 
80
	// fast path exit if we don't have rtt enabled
81
	if c.Recorder == nil {
82
		return
83
	}
84
 
85
	// sort routes based on rtt to different gateways, so hostname/1 and rpc calls
86
	// always resolves to the gateway with the lowest rtt to the client
87
	rttLookup := make(map[string]time.Duration)
88
	for _, n := range nodes {
89
		m := c.Recorder.Snapshot(rtt.MakeMeasurementKey(n), time.Second*10)
90
		if m == nil {
91
			continue
92
		}
93
		rttLookup[rtt.MakeMeasurementKey(n)] = m.Average
94
	}
95
	sort.SliceStable(nodes, func(i, j int) bool {
96
		l, lOK := rttLookup[rtt.MakeMeasurementKey(nodes[i])]
97
		r, rOK := rttLookup[rtt.MakeMeasurementKey(nodes[j])]
98
		if lOK && !rOK {
99
			return true
100
		}
101
		if !lOK && rOK {
102
			return false
103
		}
104
		return l < r
105
	})
106
 
107
	c.Logger.Debug("rtt information", zap.String("table", fmt.Sprint(rttLookup)))
108
 
109
	return nodes
110
}
111
 
112
func (c *Client) getAliveNodes(ctx context.Context) (alive []*protocol.Node, dead int) {
113
	alive = make([]*protocol.Node, 0)
114
	c.connections.Range(func(addr string, node *protocol.Node) bool {
115
		func() {
116
			callCtx, cancel := context.WithTimeout(ctx, connectTimeout)
117
			defer cancel()
118
 
119
			_, err := c.ping(callCtx, node)
120
			if err != nil {
121
				c.connections.Delete(addr)
122
				dead++
123
			} else {
124
				alive = append(alive, node)
125
			}
126
		}()
127
		return true
128
	})
129
	return
130
}
131
 
132
func (c *Client) periodicReconnection(ctx context.Context) {
133
	defer c.closeWg.Done()
134
 
135
	ctx, cancel := context.WithCancel(ctx)
136
	defer cancel()
137
	go func() {
138
		select {
139
		case <-c.closeCh:
140
			cancel()
141
		case <-ctx.Done():
142
		}
143
	}()
144
 
145
	runReconnectLoop(ctx, nil, func(ctx context.Context) (time.Duration, error) {
146
		c.reconcileConnections(ctx)
147
		return 0, nil
148
	})
149
}
150
 
151
// runReconnectLoop serializes maintenance and preserves the full client's ticker
152
// cadence. A positive retry delay takes precedence over ticks and wakeups until
153
// it fires, avoiding duplicate attempts during a complete outage.
154
func runReconnectLoop(ctx context.Context, wake <-chan struct{}, reconcile func(context.Context) (time.Duration, error)) error {
155
	ticker := time.NewTicker(checkInterval)
156
	defer ticker.Stop()
157
	retry := time.NewTimer(checkInterval)
158
	retry.Stop()
159
	defer retry.Stop()
160
	var retryC <-chan time.Time
161
	for {
162
		select {
163
		case <-ctx.Done():
164
			return ctx.Err()
165
		case <-ticker.C:
166
			if retryC != nil {
167
				continue
168
			}
169
		case <-wake:
170
			if retryC != nil {
171
				continue
172
			}
173
		case <-retryC:
174
			retryC = nil
175
		}
176
		if ctx.Err() != nil {
177
			return ctx.Err()
178
		}
179
		delay, err := reconcile(ctx)
180
		if err != nil {
181
			return err
182
		}
183
		if delay > 0 {
184
			retry.Reset(delay)
185
			retryC = retry.C
186
		}
187
	}
188
}
189
 
190
func (c *Client) reconcileConnections(ctx context.Context) {
191
	prev, failed := c.getAliveNodes(ctx)
192
	if failed > 0 {
193
		c.Logger.Info("Some connections have failed, opening more connections to specter server", zap.Int("dead", failed))
194
	}
195
	if err := c.maintainConnections(ctx); err != nil {
196
		// A candidate lookup can fail while existing gateways still work. It
197
		// must not prevent retrying previously failed tunnel publication.
198
		c.Logger.Warn("Failed to maintain gateway connections", zap.Error(err))
199
	}
200
	now, _ := c.getAliveNodes(ctx)
201
	changed := failed > 0 || endpointSignature(prev) != endpointSignature(now)
202
 
203
	c.syncMu.Lock()
204
	defer c.syncMu.Unlock()
205
	if ctx.Err() != nil {
206
		return
207
	}
208
	c.syncStateMu.RLock()
209
	retryPending := c.lastSync.pendingPublication() && !time.Now().Before(c.nextSync)
210
	c.syncStateMu.RUnlock()
211
	if changed {
212
		clear(c.publication)
213
	}
214
	if changed || retryPending {
215
		c.syncConfigTunnels(ctx, false)
216
	}
217
}
218
 
219
func (c *Client) maintainConnections(ctx context.Context) error {
220
	if c.connections.Len() == 0 {
221
		c.Logger.Info("No connected nodes, re-bootstrapping using apex")
222
		c.configMu.RLock()
223
		apex := c.ClientConfig.Configuration.Apex
224
		c.configMu.RUnlock()
225
		if err := c.bootstrap(ctx, apex); err != nil {
226
			return fmt.Errorf("rebootstrapping connection to specter: %w", err)
227
		}
228
	}
229
 
230
	callCtx, cancel := context.WithTimeout(ctx, rpcTimeout)
231
	defer cancel()
232
 
233
	nodes, err := c.requestCandidates(callCtx)
234
	if err != nil {
235
		return err
236
	}
237
	c.Logger.Debug("Candidates for RPC connections", zap.Int("num", len(nodes)))
238
 
239
	for _, node := range nodes {
240
		if c.connections.Len() >= tun.NumRedundantLinks {
241
			return nil
242
		}
243
		if err := c.openRPC(ctx, node); err != nil {
244
			return fmt.Errorf("connecting to specter server: %w", err)
245
		}
246
	}
247
 
248
	return nil
249
}
250
 
251
func (c *Client) ping(ctx context.Context, node *protocol.Node) (*protocol.ClientPingResponse, error) {
252
	return c.tunnelClient.Ping(rpc.WithNode(ctx, node), &protocol.ClientPingRequest{})
253
}
254
 
255
func (c *Client) requestCandidates(ctx context.Context) ([]*protocol.Node, error) {
256
	resp, err := retryRPC(c, ctx, func(node *protocol.Node) (*protocol.GetNodesResponse, error) {
257
		ctx = rpc.WithNode(ctx, node)
258
		return c.tunnelClient.GetNodes(ctx, &protocol.GetNodesRequest{})
259
	})
260
	if err != nil {
261
		return nil, err
262
	}
263
	return resp.GetNodes(), nil
264
}