From fa4c7c5700023ec0a13008cce0ec71ed89d92bf8 Mon Sep 17 00:00:00 2001 From: Spike Curtis Date: Thu, 16 Jul 2026 09:42:23 +0200 Subject: [PATCH] chore: fix replicasync to function for NATS even if no DERP relay address is set (#27235) 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. --- enterprise/coderd/coderd.go | 2 +- enterprise/coderd/coderd_test.go | 5 +--- enterprise/coderd/workspaceproxy.go | 2 +- enterprise/replicasync/replicasync.go | 27 ++++++++++++---------- enterprise/replicasync/replicasync_test.go | 16 ++++++------- enterprise/wsproxy/wsproxy.go | 2 +- 6 files changed, 27 insertions(+), 27 deletions(-) diff --git a/enterprise/coderd/coderd.go b/enterprise/coderd/coderd.go index 7d71e516a9..faf204227d 100644 --- a/enterprise/coderd/coderd.go +++ b/enterprise/coderd/coderd.go @@ -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 diff --git a/enterprise/coderd/coderd_test.go b/enterprise/coderd/coderd_test.go index 32292b211a..9582fe2b29 100644 --- a/enterprise/coderd/coderd_test.go +++ b/enterprise/coderd/coderd_test.go @@ -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, diff --git a/enterprise/coderd/workspaceproxy.go b/enterprise/coderd/workspaceproxy.go index c70d02e9cf..d5440c5edb 100644 --- a/enterprise/coderd/workspaceproxy.go +++ b/enterprise/coderd/workspaceproxy.go @@ -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 { diff --git a/enterprise/replicasync/replicasync.go b/enterprise/replicasync/replicasync.go index f1acececfe..d4d1aacb72 100644 --- a/enterprise/replicasync/replicasync.go +++ b/enterprise/replicasync/replicasync.go @@ -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 { diff --git a/enterprise/replicasync/replicasync_test.go b/enterprise/replicasync/replicasync_test.go index 6b2fc517d2..d9e7d1cc05 100644 --- a/enterprise/replicasync/replicasync_test.go +++ b/enterprise/replicasync/replicasync_test.go @@ -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) { diff --git a/enterprise/wsproxy/wsproxy.go b/enterprise/wsproxy/wsproxy.go index 402ee53d5e..0c8f69f30f 100644 --- a/enterprise/wsproxy/wsproxy.go +++ b/enterprise/wsproxy/wsproxy.go @@ -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 }