fix(urgent): implement relayset
Lee Yunjin committed
Apr 9, 2026 at 10:26 UTC
e3b073388e60ab933783403a1b013392a7a028d9
2 files changed
+308
-2
portal/discovery/relayset.go
+304
-1
@@ -1,6 +1,7 @@
1
package discovery
2
3
import (
4
+ "context"
5
"errors"
6
"net/http"
7
"reflect"
@@ -59,8 +60,12 @@ type RelaySet struct {
60
lastStatusReachable map[string]bool
61
lastStatusSummary RelaySummary
62
haveLastStatus bool
63
+ selfRelayKey string
64
+ selfRelayURL string
65
}
66
67
+const defaultDiscoveryRecoveryFailures = 3
68
+
69
func NewRelaySet() *RelaySet {
70
return &RelaySet{
71
relayKeysByURL: make(map[string]string),
@@ -69,6 +74,75 @@ func NewRelaySet() *RelaySet {
74
}
75
}
76
77
+func (s *RelaySet) isSelfRelayURLLocked(relayURL string) bool {
78
+ if s == nil {
79
+ return false
80
+ }
81
+ relayURL = strings.TrimSpace(relayURL)
82
+ return relayURL != "" && s.selfRelayURL != "" && relayURL == s.selfRelayURL
83
+}
84
+
85
+func (s *RelaySet) isSelfRelayDescriptorLocked(desc types.RelayDescriptor) bool {
86
+ if s == nil {
87
+ return false
88
+ }
89
+ if s.selfRelayKey != "" {
90
+ if relayKey := desc.Key(); relayKey != "" && relayKey == s.selfRelayKey {
91
+ return true
92
+ }
93
+ }
94
+ return s.isSelfRelayURLLocked(desc.APIHTTPSAddr)
95
+}
96
+
97
+func (s *RelaySet) pruneSelfRelayLocked() {
98
+ if s == nil {
99
+ return
100
+ }
101
+ if s.selfRelayURL != "" {
102
+ filtered := s.knownRelayURLs[:0]
103
+ for _, relayURL := range s.knownRelayURLs {
104
+ if s.isSelfRelayURLLocked(relayURL) {
105
+ continue
106
+ }
107
+ filtered = append(filtered, relayURL)
108
+ }
109
+ s.knownRelayURLs = filtered
110
+ delete(s.localByURL, s.selfRelayURL)
111
+ delete(s.relayKeysByURL, s.selfRelayURL)
112
+ }
113
+ if s.selfRelayKey != "" {
114
+ if view, ok := s.relays[s.selfRelayKey]; ok {
115
+ delete(s.localByURL, view.Descriptor.APIHTTPSAddr)
116
+ delete(s.relayKeysByURL, view.Descriptor.APIHTTPSAddr)
117
+ }
118
+ delete(s.relays, s.selfRelayKey)
119
+ }
120
+}
121
+
122
+// SetSelfRelay configures the relay set with the local relay identity so it
123
+// can skip self-references when evaluating discovery hints.
124
+func (s *RelaySet) SetSelfRelay(identity types.Identity, relayURL string) error {
125
+ if s == nil {
126
+ return nil
127
+ }
128
+ relayURL = strings.TrimSpace(relayURL)
129
+ if relayURL != "" {
130
+ normalized, err := utils.NormalizeRelayURL(relayURL)
131
+ if err != nil {
132
+ return err
133
+ }
134
+ relayURL = normalized
135
+ }
136
+
137
+ s.mu.Lock()
138
+ defer s.mu.Unlock()
139
+ s.selfRelayKey = identity.Key()
140
+ s.selfRelayURL = relayURL
141
+ s.pruneSelfRelayLocked()
142
+ s.logStatusChange()
143
+ return nil
144
+}
145
+
146
func (s *RelaySet) trackedRelayURLs() []string {
147
if s == nil {
148
return nil
@@ -256,6 +330,27 @@ func (s *RelaySet) BootstrapDescriptors() []types.RelayDescriptor {
330
return out
331
}
332
333
+func (s *RelaySet) ActiveRelayDescriptors() []types.RelayDescriptor {
334
+ advertised := s.AdvertisedDescriptors()
335
+ if len(advertised) == 0 {
336
+ return nil
337
+ }
338
+ s.mu.RLock()
339
+ defer s.mu.RUnlock()
340
+
341
+ filtered := advertised[:0]
342
+ for _, desc := range advertised {
343
+ if s.isSelfRelayDescriptorLocked(desc) {
344
+ continue
345
+ }
346
+ filtered = append(filtered, desc)
347
+ }
348
+ if len(filtered) == 0 {
349
+ return nil
350
+ }
351
+ return append([]types.RelayDescriptor(nil), filtered...)
352
+}
353
+
354
func (s *RelaySet) BanRelayURL(relayURL, reason string) bool {
355
if s == nil {
356
return false
@@ -486,6 +581,107 @@ func (s *RelaySet) ReplaceKnownRelayURLs(relayURLs []string) {
581
s.knownRelayURLs = append([]string(nil), filtered...)
582
}
583
584
+func (s *RelaySet) SetBootstrapRelayURLs(relayURLs []string) {
585
+ if s == nil {
586
+ return
587
+ }
588
+ normalized := make([]string, 0, len(relayURLs))
589
+ seen := make(map[string]struct{}, len(relayURLs))
590
+ for _, relayURL := range relayURLs {
591
+ relayURL = strings.TrimSpace(relayURL)
592
+ if relayURL == "" {
593
+ continue
594
+ }
595
+ parsed, err := utils.NormalizeRelayURL(relayURL)
596
+ if err != nil {
597
+ log.Warn().
598
+ Err(err).
599
+ Str("relay_url", relayURL).
600
+ Msg("skip invalid bootstrap relay url")
601
+ continue
602
+ }
603
+ if _, ok := seen[parsed]; ok {
604
+ continue
605
+ }
606
+ seen[parsed] = struct{}{}
607
+ normalized = append(normalized, parsed)
608
+ }
609
+
610
+ s.mu.Lock()
611
+ defer s.mu.Unlock()
612
+
613
+ filtered := normalized[:0]
614
+ for _, relayURL := range normalized {
615
+ if s.isSelfRelayURLLocked(relayURL) {
616
+ continue
617
+ }
618
+ filtered = append(filtered, relayURL)
619
+ state := s.localByURL[relayURL]
620
+ state.Bootstrap = true
621
+ s.localByURL[relayURL] = state
622
+ }
623
+ normalized = filtered
624
+
625
+ current := make(map[string]struct{}, len(normalized))
626
+ for _, relayURL := range normalized {
627
+ current[relayURL] = struct{}{}
628
+ }
629
+
630
+ existing := make(map[string]struct{}, len(s.knownRelayURLs))
631
+ for _, relayURL := range s.knownRelayURLs {
632
+ existing[relayURL] = struct{}{}
633
+ if _, ok := current[relayURL]; !ok {
634
+ state := s.localByURL[relayURL]
635
+ if state.Bootstrap {
636
+ state.Bootstrap = false
637
+ s.localByURL[relayURL] = state
638
+ }
639
+ }
640
+ }
641
+ s.knownRelayURLs = append([]string(nil), normalized...)
642
+ s.logStatusChange()
643
+}
644
+
645
+func (s *RelaySet) mergeKnownRelayURLs(relayURLs []string) error {
646
+ if s == nil || len(relayURLs) == 0 {
647
+ return nil
648
+ }
649
+ normalized, err := utils.NormalizeRelayURLs(relayURLs...)
650
+ if err != nil {
651
+ return err
652
+ }
653
+ if len(normalized) == 0 {
654
+ return nil
655
+ }
656
+
657
+ s.mu.Lock()
658
+ defer s.mu.Unlock()
659
+
660
+ existing := make(map[string]struct{}, len(s.knownRelayURLs))
661
+ for _, relayURL := range s.knownRelayURLs {
662
+ existing[relayURL] = struct{}{}
663
+ }
664
+
665
+ changed := false
666
+ for _, relayURL := range normalized {
667
+ if s.isSelfRelayURLLocked(relayURL) {
668
+ continue
669
+ }
670
+ if _, ok := existing[relayURL]; ok {
671
+ continue
672
+ }
673
+ existing[relayURL] = struct{}{}
674
+ s.knownRelayURLs = append(s.knownRelayURLs, relayURL)
675
+ state := s.localByURL[relayURL]
676
+ s.localByURL[relayURL] = state
677
+ changed = true
678
+ }
679
+ if changed {
680
+ s.logStatusChange()
681
+ }
682
+ return nil
683
+}
684
+
685
func (s *RelaySet) registerDescriptor(desc types.RelayDescriptor, now time.Time) (string, bool, bool, error) {
686
if s == nil {
687
return "", false, false, nil
@@ -583,6 +779,9 @@ func (s *RelaySet) applyDiscoveryDescriptors(targetIdentity types.Identity, targ
779
return false, 0, err
780
}
781
for _, relayDescriptor := range relayDescriptors {
782
+ if s.isSelfRelayDescriptorLocked(relayDescriptor) {
783
+ continue
784
+ }
785
if err := apply(relayDescriptor, false, true); err != nil {
786
return false, 0, err
787
}
@@ -641,6 +840,31 @@ func (s *RelaySet) ApplyOverlayRelayDiscoveryResponse(targetIdentity types.Ident
840
return relayDiscoveryURLs(selfDescriptor, filteredRelayDescriptors), relaySetChanged, addedRelayCount, warnErr, nil
841
}
842
843
+func (s *RelaySet) ApplyRelayDiscoveryResponseSimple(targetIdentity types.Identity, targetURL string, resp types.DiscoveryResponse, now time.Time) error {
844
+ relayURLs, _, _, warnErr, err := s.ApplyRelayDiscoveryResponse(targetIdentity, targetURL, resp, now)
845
+ if warnErr != nil {
846
+ log.Warn().
847
+ Err(warnErr).
848
+ Str("relay", targetURL).
849
+ Msg("relay discovery response completed with warnings")
850
+ }
851
+ if err != nil {
852
+ return err
853
+ }
854
+
855
+ targetRelayURL := strings.TrimSpace(resp.Self.APIHTTPSAddr)
856
+ if targetRelayURL == "" && len(relayURLs) > 0 {
857
+ targetRelayURL = relayURLs[0]
858
+ }
859
+ if targetRelayURL == "" {
860
+ return nil
861
+ }
862
+ if err := s.mergeKnownRelayURLs([]string{targetRelayURL}); err != nil {
863
+ return err
864
+ }
865
+ return nil
866
+}
867
+
868
func (s *RelaySet) RegisterBootstrapRelayURLs(inputs []string) ([]string, error) {
869
if s == nil || len(inputs) == 0 {
870
return nil, nil
@@ -686,7 +910,82 @@ func (s *RelaySet) RegisterBootstrapRelayURLs(inputs []string) ([]string, error)
910
return added, nil
911
}
912
689
-func (s *RelaySet) RecordDiscoveryFailure(identity types.Identity, relayURL string, err error, recoveryFailures int, now time.Time) (expired bool, expireReason string, consecutiveFailures int) {
913
+func (s *RelaySet) refreshBootstrapDiscovery(ctx context.Context, rootCAPEM []byte) error {
914
+ bootstraps := s.BootstrapDescriptors()
915
+ for _, bootstrap := range bootstraps {
916
+ resp, err := DiscoverRelayDiscovery(ctx, bootstrap.APIHTTPSAddr, rootCAPEM, nil)
917
+ if err != nil {
918
+ if ctx.Err() != nil {
919
+ return nil
920
+ }
921
+ s.RecordBootstrapDiscoveryFailure(bootstrap.APIHTTPSAddr, err, time.Now().UTC())
922
+ continue
923
+ }
924
+
925
+ now := time.Now().UTC()
926
+ relayURLs, _, _, warnErr, applyErr := s.ApplyRelayDiscoveryResponse(bootstrap.Identity, bootstrap.APIHTTPSAddr, resp, now)
927
+ if warnErr != nil {
928
+ log.Warn().
929
+ Err(warnErr).
930
+ Str("relay", bootstrap.APIHTTPSAddr).
931
+ Msg("bootstrap relay discovery completed with warnings")
932
+ }
933
+ if applyErr != nil {
934
+ s.MarkRelayFailure(bootstrap.APIHTTPSAddr, now)
935
+ log.Warn().
936
+ Err(applyErr).
937
+ Str("relay", bootstrap.APIHTTPSAddr).
938
+ Msg("bootstrap relay discovery failed")
939
+ continue
940
+ }
941
+ if len(relayURLs) == 0 {
942
+ continue
943
+ }
944
+ if err := s.mergeKnownRelayURLs(relayURLs); err != nil {
945
+ log.Warn().
946
+ Err(err).
947
+ Str("relay", bootstrap.APIHTTPSAddr).
948
+ Msg("merge discovered relay urls")
949
+ }
950
+ }
951
+ return nil
952
+}
953
+
954
+func (s *RelaySet) RunLoop(ctx context.Context, rootCAPEM []byte, syncRuntime func() error) error {
955
+ if s == nil {
956
+ <-ctx.Done()
957
+ return nil
958
+ }
959
+
960
+ ticker := time.NewTicker(types.DiscoveryPollInterval)
961
+ defer ticker.Stop()
962
+
963
+ for {
964
+ if err := s.refreshBootstrapDiscovery(ctx, rootCAPEM); err != nil {
965
+ return err
966
+ }
967
+ if ctx.Err() != nil {
968
+ return nil
969
+ }
970
+ if syncRuntime != nil {
971
+ if err := syncRuntime(); err != nil {
972
+ return err
973
+ }
974
+ }
975
+
976
+ select {
977
+ case <-ctx.Done():
978
+ return nil
979
+ case <-ticker.C:
980
+ }
981
+ }
982
+}
983
+
984
+func (s *RelaySet) RecordDiscoveryFailure(identity types.Identity, relayURL string, err error) (expired bool, expireReason string, consecutiveFailures int) {
985
+ return s.RecordDiscoveryFailureWithRecovery(identity, relayURL, err, defaultDiscoveryRecoveryFailures, time.Now().UTC())
986
+}
987
+
988
+func (s *RelaySet) RecordDiscoveryFailureWithRecovery(identity types.Identity, relayURL string, err error, recoveryFailures int, now time.Time) (expired bool, expireReason string, consecutiveFailures int) {
989
if s == nil {
990
return false, "", 0
991
}
@@ -700,6 +999,8 @@ func (s *RelaySet) RecordDiscoveryFailure(identity types.Identity, relayURL stri
999
}
1000
if now.IsZero() {
1001
now = time.Now().UTC()
1002
+ } else {
1003
+ now = now.UTC()
1004
}
1005
1006
s.mu.Lock()
@@ -718,6 +1019,7 @@ func (s *RelaySet) RecordDiscoveryFailure(identity types.Identity, relayURL stri
1019
s.logStatusChange()
1020
if !localState.Expired && localState.ConsecutiveFailures >= recoveryFailures {
1021
state := s.localByURL[view.Descriptor.APIHTTPSAddr]
1022
+ state.Advertised = false
1023
state.Expired = true
1024
s.localByURL[view.Descriptor.APIHTTPSAddr] = state
1025
s.logStatusChange()
@@ -730,6 +1032,7 @@ func (s *RelaySet) RecordDiscoveryFailure(identity types.Identity, relayURL stri
1032
apiErr.StatusCode == http.StatusNotFound ||
1033
apiErr.StatusCode == http.StatusGone) {
1034
state := s.localByURL[view.Descriptor.APIHTTPSAddr]
1035
+ state.Advertised = false
1036
state.Expired = true
1037
s.localByURL[view.Descriptor.APIHTTPSAddr] = state
1038
s.logStatusChange()
portal/server.go
+4
-1
@@ -218,6 +218,9 @@ func NewServer(cfg ServerConfig) (*Server, error) {
218
}
219
if cfg.DiscoveryEnabled {
220
set := discovery.NewRelaySet()
221
+ if err := set.SetSelfRelay(identity, selfRelayURL); err != nil {
222
+ return nil, fmt.Errorf("set self relay: %w", err)
223
+ }
224
if _, err := set.RegisterBootstrapRelayURLs(cfg.Bootstraps); err != nil {
225
return nil, err
226
}
@@ -779,7 +782,7 @@ func (s *Server) runRelayDiscoveryLoop(ctx context.Context) error {
782
}
783
}
784
782
- expired, expireReason, consecutiveFailures := s.relaySet.RecordDiscoveryFailure(relay.Identity, relay.APIHTTPSAddr, failureErr, defaultWGRecoveryFailures, time.Now().UTC())
785
+ expired, expireReason, consecutiveFailures := s.relaySet.RecordDiscoveryFailureWithRecovery(relay.Identity, relay.APIHTTPSAddr, failureErr, defaultWGRecoveryFailures, time.Now().UTC())
786
if expired {
787
if syncErr := s.overlay.Sync(s.identity.Key(), s.relaySet.Snapshot()); syncErr != nil && failureErr == nil {
788
failureErr = syncErr