opensource/coder
3
0
Fork 0
mirror of https://github.com/coder/coder.git synced 2026-09-24 15:04:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore: fix replicasync to function for NATS even if no DERP relay address is set (#27235)

Browse Source 
<!--

If you have used AI to produce some or all of this PR, please ensure you have read our [AI Contribution guidelines](https://coder.com/docs/about/contributing/AI_CONTRIBUTING) before submitting.

-->

Fixes `replicasync` so that it keeps peers even if they don't set a RelayAddress. This allows NATS to function even if you are not running a DERP relay on the primay Coderd instances.

Also renames some replicasync functions to make it clear they are for DERP.
This commit is contained in:
Spike Curtis
2026-07-16 09:42:23 +02:00
committed by GitHub
parent 21d08241e9
commit fa4c7c5700
6 changed files with 27 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
enterprise/coderd/coderd.go
+1 -1
View File
@@ -1024,7 +1024,7 @@ func (api *API) updateEntitlements(ctx context.Context) error {
// Only update DERP mesh if the built-in server is enabled.
if api.Options.DeploymentValues.DERP.Server.Enable {
addresses := make([]string, 0)
for _, replica := range api.replicaManager.Regional() {
for _, replica := range api.replicaManager.DERPReplicasThisRegion() {
// Don't add replicas with an empty relay address.
if replica.RelayAddress == "" {
continue
enterprise/coderd/coderd_test.go
+1 -4
View File
@@ -671,10 +671,7 @@ func TestMultiReplica_NATSPubsubPeers(t *testing.T) {
t.Cleanup(func() { _ = natsB.Close() })
mgr, err := replicasync.New(ctx, logger.Named("replica-b"), db, pgPubsub, &replicasync.Options{
ID: uuid.New(),
// port doesn't matter because we don't have an API up, but replicasync will refuse peers that don't set
// RelayAddress at all.
RelayAddress: "https://127.0.0.1",
ID: uuid.New(),
ClusterHost: "127.0.0.1",
RegionID: 12345,
UpdateInterval: testutil.IntervalFast,
enterprise/coderd/workspaceproxy.go
+1 -1
View File
@@ -723,7 +723,7 @@ func (api *API) workspaceProxyRegister(rw http.ResponseWriter, r *http.Request)
}
// Find sibling regions to respond with for derpmesh.
siblings := api.replicaManager.InRegion(regionID)
siblings := api.replicaManager.DERPReplicasInRegion(regionID)
siblingsRes := make([]codersdk.Replica, 0, len(siblings))
for _, replica := range siblings {
if replica.ID == req.ReplicaID {
enterprise/replicasync/replicasync.go
+15 -12
View File
@@ -260,12 +260,11 @@ func (m *Manager) syncReplicas(ctx context.Context) error {
if replica.ID == m.id {
continue
}
// Don't peer with nodes that have an empty relay address.
if replica.RelayAddress == "" {
m.logger.Debug(ctx, "peer doesn't have an address, skipping",
// legit if they have DERP disabled in that region, so just log for debugging.
m.logger.Debug(ctx, "peer doesn't have an address",
slog.F("replica_hostname", replica.Hostname),
)
continue
}
m.peers = append(m.peers, replica)
}
@@ -279,11 +278,11 @@ func (m *Manager) syncReplicas(ctx context.Context) error {
}
defer client.CloseIdleConnections()
peers := m.Regional()
peers := m.DERPReplicasThisRegion()
errs := make(chan error, len(peers))
for _, peer := range peers {
go func(peer database.Replica) {
err := PingPeerReplica(ctx, client, peer.RelayAddress)
err := DERPPingPeerReplica(ctx, client, peer.RelayAddress)
if err != nil {
errs <- xerrors.Errorf("ping sibling replica %s (%s): %w", peer.Hostname, peer.RelayAddress, err)
m.logger.Warn(ctx, "failed to ping sibling replica, this could happen if the replica has shutdown",
@@ -372,9 +371,9 @@ func (m *Manager) syncReplicas(ctx context.Context) error {
return nil
}
// PingPeerReplica pings a peer replica over it's internal relay address to
// DERPPingPeerReplica pings a peer replica over it's internal relay address to
// ensure it's reachable and alive for health purposes.
func PingPeerReplica(ctx context.Context, client http.Client, relayAddress string) error {
func DERPPingPeerReplica(ctx context.Context, client http.Client, relayAddress string) error {
ra, err := url.Parse(relayAddress)
if err != nil {
return xerrors.Errorf("parse relay address %q: %w", relayAddress, err)
@@ -442,8 +441,8 @@ func (m *Manager) SetSelfNATSPort(port int32) {
// to us. So, we're not going to trigger a synchronous update. We'll just wait for the periodic update ticker.
}
// InRegion returns every replica in the given DERP region excluding itself.
func (m *Manager) InRegion(regionID int32) []database.Replica {
// DERPReplicasInRegion returns every replica in the given DERP region excluding itself.
func (m *Manager) DERPReplicasInRegion(regionID int32) []database.Replica {
m.mutex.Lock()
defer m.mutex.Unlock()
replicas := make([]database.Replica, 0)
@@ -451,14 +450,18 @@ func (m *Manager) InRegion(regionID int32) []database.Replica {
if replica.RegionID != regionID {
continue
}
// Replicas without a relay address cannot be used for DERP.
if replica.RelayAddress == "" {
continue
}
replicas = append(replicas, replica)
}
return replicas
}
// Regional returns all replicas in the same region excluding itself.
func (m *Manager) Regional() []database.Replica {
return m.InRegion(m.regionID())
// DERPReplicasThisRegion returns all replicas in the same region excluding itself.
func (m *Manager) DERPReplicasThisRegion() []database.Replica {
return m.DERPReplicasInRegion(m.regionID())
}
func (m *Manager) regionID() int32 {
enterprise/replicasync/replicasync_test.go
+8 -8
View File
@@ -76,8 +76,8 @@ func TestReplica(t *testing.T) {
require.NoError(t, err)
defer server.Close()
require.Len(t, server.Regional(), 1)
require.Equal(t, peer.ID, server.Regional()[0].ID)
require.Len(t, server.DERPReplicasThisRegion(), 1)
require.Equal(t, peer.ID, server.DERPReplicasThisRegion()[0].ID)
require.Empty(t, server.Self().Error)
_ = server.Close()
})
@@ -122,8 +122,8 @@ func TestReplica(t *testing.T) {
require.NoError(t, err)
defer server.Close()
require.Len(t, server.Regional(), 1)
require.Equal(t, peer.ID, server.Regional()[0].ID)
require.Len(t, server.DERPReplicasThisRegion(), 1)
require.Equal(t, peer.ID, server.DERPReplicasThisRegion()[0].ID)
require.Empty(t, server.Self().Error)
_ = server.Close()
})
@@ -150,8 +150,8 @@ func TestReplica(t *testing.T) {
require.NoError(t, err)
defer server.Close()
require.Len(t, server.Regional(), 1)
require.Equal(t, peer.ID, server.Regional()[0].ID)
require.Len(t, server.DERPReplicasThisRegion(), 1)
require.Equal(t, peer.ID, server.DERPReplicasThisRegion()[0].ID)
require.NotEmpty(t, server.Self().Error)
require.Contains(t, server.Self().Error, "Failed to dial peers")
_ = server.Close()
@@ -182,7 +182,7 @@ func TestReplica(t *testing.T) {
err = pubsub.Publish(replicasync.PubsubEvent, []byte(peer.ID.String()))
require.NoError(t, err)
require.Eventually(t, func() bool {
return len(server.Regional()) == 1
return len(server.DERPReplicasThisRegion()) == 1
}, testutil.WaitShort, testutil.IntervalFast)
_ = server.Close()
})
@@ -204,7 +204,7 @@ func TestReplica(t *testing.T) {
require.NoError(t, err)
defer server.Close()
require.Eventually(t, func() bool {
return len(server.Regional()) == 0
return len(server.DERPReplicasThisRegion()) == 0
}, testutil.WaitShort, testutil.IntervalFast)
})
t.Run("MultipleCallbacks", func(t *testing.T) {
enterprise/wsproxy/wsproxy.go
+1 -1
View File
@@ -600,7 +600,7 @@ func pingReplica(ctx context.Context, client http.Client, replica codersdk.Repli
const attempts = 2
var err error
for i := 0; i < attempts; i++ {
err = replicasync.PingPeerReplica(ctx, client, replica.RelayAddress)
err = replicasync.DERPPingPeerReplica(ctx, client, replica.RelayAddress)
if err == nil {
return nil
}