1
package client
2
 
3
import (
4
	"errors"
5
	"testing"
6
	"time"
7
 
8
	"go.miragespace.co/specter/spec/mocks"
9
	"go.miragespace.co/specter/spec/protocol"
10
	"go.miragespace.co/specter/spec/rtt"
11
 
12
	"github.com/stretchr/testify/mock"
13
	"github.com/stretchr/testify/require"
14
)
15
 
16
func TestPublicationRetryInvalidatesAcknowledgementAfterGatewayReordering(t *testing.T) {
17
	c, tunnelClient := newOutcomeTestClient(t, []Tunnel{
18
		{Hostname: "first", Target: "http://localhost:8080"},
19
		{Hostname: "pending", Target: "http://localhost:9090"},
20
	})
21
	a := c.getConnectedNodes()[0]
22
	b := &protocol.Node{Id: 2, Address: "second-gateway.example.com"}
23
	c.connections.Store(b.GetAddress(), b)
24
	ab := []*protocol.Node{a, b}
25
	ba := []*protocol.Node{b, a}
26
 
27
	aRTT := &rtt.Statistics{Average: time.Millisecond}
28
	bRTT := &rtt.Statistics{Average: 2 * time.Millisecond}
29
	recorder := new(mocks.Measurement)
30
	recorder.On("Snapshot", rtt.MakeMeasurementKey(a), mock.Anything).Return(aRTT)
31
	recorder.On("Snapshot", rtt.MakeMeasurementKey(b), mock.Anything).Return(bRTT)
32
	c.Recorder = recorder
33
	t.Cleanup(func() { recorder.AssertExpectations(t) })
34
 
35
	tunnelClient.TunnelService.On("RegisteredHostnames", mock.Anything, mock.Anything).
36
		Return(&protocol.RegisteredHostnamesResponse{}, nil).Once()
37
	firstAt := func(nodes []*protocol.Node) any {
38
		return mock.MatchedBy(func(req *protocol.PublishTunnelRequest) bool {
39
			return req.GetHostname() == "first" && len(req.GetServers()) == 2 &&
40
				req.GetServers()[0].GetId() == nodes[0].GetId() &&
41
				req.GetServers()[1].GetId() == nodes[1].GetId()
42
		})
43
	}
44
	tunnelClient.TunnelService.On("PublishTunnel", mock.Anything, firstAt(ab)).
45
		Return(&protocol.PublishTunnelResponse{Published: ab}, nil).Once()
46
	tunnelClient.TunnelService.On("PublishTunnel", mock.Anything, firstAt(ba)).
47
		Return(&protocol.PublishTunnelResponse{Published: []*protocol.Node{b}}, nil).Once()
48
	tunnelClient.TunnelService.On("PublishTunnel", mock.Anything, firstAt(ab)).
49
		Return(&protocol.PublishTunnelResponse{Published: ab}, nil).Once()
50
	pending := mock.MatchedBy(func(req *protocol.PublishTunnelRequest) bool {
51
		return req.GetHostname() == "pending"
52
	})
53
	tunnelClient.TunnelService.On("PublishTunnel", mock.Anything, pending).
54
		Return(nil, errors.New("temporary publication failure")).Twice()
55
	tunnelClient.TunnelService.On("PublishTunnel", mock.Anything, pending).
56
		Return(&protocol.PublishTunnelResponse{Published: ab}, nil).Once()
57
 
58
	initial := c.SyncConfigTunnels(t.Context())
59
	require.True(t, initial.Tunnels[0].Published)
60
	require.Empty(t, initial.Tunnels[0].Error)
61
	require.NotEmpty(t, initial.Tunnels[1].Error)
62
	require.True(t, c.getStatus().Pending)
63
 
64
	// A partial attempt in the opposite order can overwrite numbered routing
65
	// slots, so the original successful acknowledgement is no longer valid.
66
	aRTT.Average, bRTT.Average = 2*time.Millisecond, time.Millisecond
67
	c.syncConfigTunnels(t.Context(), false)
68
	partial := c.getStatus()
69
	require.True(t, partial.Pending)
70
	require.Contains(t, partial.Synchronization.Tunnels[0].Error, "published 1 of 2")
71
	require.Equal(t, 1, partial.Synchronization.Tunnels[0].PublishedEndpoints)
72
 
73
	// Returning to the original order must repair those slots with a fresh RPC.
74
	aRTT.Average, bRTT.Average = time.Millisecond, 2*time.Millisecond
75
	c.syncConfigTunnels(t.Context(), false)
76
	status := c.getStatus()
77
	require.False(t, status.Pending)
78
	require.Nil(t, status.RetryAt)
79
	require.Empty(t, status.Synchronization.Error)
80
	for _, tunnel := range status.Synchronization.Tunnels {
81
		require.True(t, tunnel.Published)
82
		require.Equal(t, 2, tunnel.PublishedEndpoints)
83
		require.Empty(t, tunnel.Error)
84
	}
85
}