@samitouri / QOSAMI-X402Tunnel-1 / commits / 34b52bd9

refactor: replace maps with Snapshot for immutable state management

Kim committed May 15, 2026 at 15:22 UTC 34b52bd92195dadd90cde7724addeaf42c2dc2d1
21 files changed +640 -363
cmd/portal-tunnel/agent/control.go
+4 -4
@@ -9,7 +9,7 @@ import (
9 "strings"
10 "time"
11
12 - portalauth "github.com/gosuda/portal-tunnel/v2/portal/auth"
12 + "github.com/gosuda/portal-tunnel/v2/portal/auth"
13 "github.com/gosuda/portal-tunnel/v2/types"
14 "github.com/gosuda/portal-tunnel/v2/utils"
15 )
@@ -30,7 +30,7 @@ type endpoint struct {
30 type controlHandler struct {
31 manager *manager
32 token string
33 - auth *portalauth.WalletAuthenticator
33 + auth *auth.WalletAuthenticator
34 shutdown func()
35 }
36
@@ -272,9 +272,9 @@ func agentAuthURI(r *http.Request, endpointPath string) string {
272
273 func writeAgentWalletAuthError(w http.ResponseWriter, err error) {
274 switch {
275 - case errors.Is(err, portalauth.ErrWalletAuthUnauthorized):
275 + case errors.Is(err, auth.ErrWalletAuthUnauthorized):
276 utils.WriteAPIError(w, http.StatusForbidden, types.APIErrorCodeUnauthorized, err.Error())
277 - case errors.Is(err, portalauth.ErrWalletAuthChallengeNotFound), errors.Is(err, portalauth.ErrWalletAuthChallengeExpired), errors.Is(err, portalauth.ErrWalletAuthInvalidSignature):
277 + case errors.Is(err, auth.ErrWalletAuthChallengeNotFound), errors.Is(err, auth.ErrWalletAuthChallengeExpired), errors.Is(err, auth.ErrWalletAuthInvalidSignature):
278 utils.WriteAPIError(w, http.StatusUnauthorized, types.APIErrorCodeUnauthorized, err.Error())
279 default:
280 utils.WriteAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidRequest, err.Error())
cmd/portal-tunnel/agent/manager.go
+3 -2
@@ -411,6 +411,7 @@ func (m *manager) ApplyConfig(cfg Config) error {
411
412 func (m *manager) Snapshot() types.AgentStatusResponse {
413 m.mu.RLock()
414 + configPath := m.cfg.sourcePath
415 tunnels := make([]*managedTunnel, 0, len(m.tunnels))
416 for _, tunnel := range m.tunnels {
417 tunnels = append(tunnels, tunnel)
@@ -426,7 +427,7 @@ func (m *manager) Snapshot() types.AgentStatusResponse {
427 })
428
429 return types.AgentStatusResponse{
429 - ConfigPath: m.cfg.sourcePath,
430 + ConfigPath: configPath,
431 ControlAddr: m.controlAddr,
432 Tunnels: statuses,
433 }
@@ -521,8 +522,8 @@ func (t *managedTunnel) SetMultiHop(relayURLs []string) error {
522
523 func (t *managedTunnel) UpdateSettings(updateMetadata, updateMaxActiveRelays bool) error {
524 t.mu.RLock()
524 - cfg := t.cfg
525 exposure := t.exposure
526 + cfg := t.cfg
527 t.mu.RUnlock()
528 if exposure == nil {
529 return nil
cmd/portal-tunnel/agent/run.go
+2 -2
@@ -13,7 +13,7 @@ import (
13
14 "github.com/rs/zerolog/log"
15
16 - portalauth "github.com/gosuda/portal-tunnel/v2/portal/auth"
16 + "github.com/gosuda/portal-tunnel/v2/portal/auth"
17 "github.com/gosuda/portal-tunnel/v2/utils"
18 )
19
@@ -31,7 +31,7 @@ func Run(ctx context.Context, cfg Config) error {
31 defer cancel()
32
33 manager := newManager(cfg, "")
34 - walletAuth, err := portalauth.NewWalletAuthenticator(portalauth.WalletAuthConfig{
34 + walletAuth, err := auth.NewWalletAuthenticator(auth.WalletAuthConfig{
35 AllowedAddresses: cfg.Agent.AllowedWallets,
36 AllowAnyAddress: len(cfg.Agent.AllowedWallets) == 0,
37 Statement: "Sign in to Portal agent",
cmd/relay-server/admin.go
+15 -10
@@ -8,7 +8,8 @@ import (
8 "strings"
9 "time"
10
11 - portalauth "github.com/gosuda/portal-tunnel/v2/portal/auth"
11 + "github.com/gosuda/portal-tunnel/v2/portal"
12 + "github.com/gosuda/portal-tunnel/v2/portal/auth"
13 "github.com/gosuda/portal-tunnel/v2/portal/identity"
14 "github.com/gosuda/portal-tunnel/v2/portal/policy"
15 "github.com/gosuda/portal-tunnel/v2/types"
@@ -21,7 +22,7 @@ const (
22 adminBodyLimit = 1 << 16
23 )
24
24 -func loadAdminState(path string, runtime *policy.Runtime) (persistedAdminState, error) {
25 +func loadAdminState(path string, server *portal.Server) (persistedAdminState, error) {
26 path = strings.TrimSpace(path)
27 if path == "" {
28 return persistedAdminState{}, nil
@@ -31,7 +32,7 @@ func loadAdminState(path string, runtime *policy.Runtime) (persistedAdminState,
32 if _, err := utils.ReadJSONFileIfExists(path, &payload); err != nil {
33 return persistedAdminState{}, err
34 }
34 - if err := payload.apply(runtime); err != nil {
35 + if err := payload.apply(server); err != nil {
36 return persistedAdminState{}, err
37 }
38 return payload, nil
@@ -140,14 +141,14 @@ func (f *Frontend) serveAdmin(w http.ResponseWriter, r *http.Request) {
141 })
142 case types.PathAdminUDP:
143 f.handlePortSettings(w, r, invalidRequestBody, runtime,
143 - runtime.SetUDPPolicy,
144 + f.server.SetUDPPolicy,
145 func() any {
146 return types.AdminUDPSettingsResponse{Enabled: runtime.IsUDPEnabled(), MaxLeases: runtime.UDPMaxLeases()}
147 },
148 )
149 case types.PathAdminTCPPort:
150 f.handlePortSettings(w, r, invalidRequestBody, runtime,
150 - runtime.SetTCPPortPolicy,
151 + f.server.SetTCPPortPolicy,
152 func() any {
153 return types.AdminTCPPortSettingsResponse{Enabled: runtime.IsTCPPortEnabled(), MaxLeases: runtime.TCPPortMaxLeases()}
154 },
@@ -385,9 +386,9 @@ func adminAuthURI(r *http.Request, endpointPath string) string {
386
387 func writeWalletAuthError(w http.ResponseWriter, err error) {
388 switch {
388 - case errors.Is(err, portalauth.ErrWalletAuthUnauthorized):
389 + case errors.Is(err, auth.ErrWalletAuthUnauthorized):
390 utils.WriteAPIError(w, http.StatusForbidden, types.APIErrorCodeUnauthorized, err.Error())
390 - case errors.Is(err, portalauth.ErrWalletAuthChallengeNotFound), errors.Is(err, portalauth.ErrWalletAuthChallengeExpired), errors.Is(err, portalauth.ErrWalletAuthInvalidSignature):
391 + case errors.Is(err, auth.ErrWalletAuthChallengeNotFound), errors.Is(err, auth.ErrWalletAuthChallengeExpired), errors.Is(err, auth.ErrWalletAuthInvalidSignature):
392 utils.WriteAPIError(w, http.StatusUnauthorized, types.APIErrorCodeUnauthorized, err.Error())
393 default:
394 utils.WriteAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidRequest, err.Error())
@@ -450,7 +451,11 @@ func applyOptionalPolicy(enabled *bool, maxLeases *int, getEnabled func() bool,
451 set(e, m)
452 }
453
453 -func (s persistedAdminState) apply(runtime *policy.Runtime) error {
454 +func (s persistedAdminState) apply(server *portal.Server) error {
455 + if server == nil {
456 + return nil
457 + }
458 + runtime := server.PolicyRuntime()
459 if runtime == nil {
460 return nil
461 }
@@ -466,7 +471,7 @@ func (s persistedAdminState) apply(runtime *policy.Runtime) error {
471 runtime.SetBannedIdentityKeys(identity.NormalizeIdentityKeys(s.BannedIdentityKeys))
472 runtime.IPFilter().SetBannedIPs(s.BannedIPs)
473 runtime.BPSManager().SetIdentityBPSLimits(identity.NormalizeIdentityKeyBPS(s.IdentityBPS))
469 - applyOptionalPolicy(s.UDPEnabled, s.UDPMaxLeases, runtime.IsUDPEnabled, runtime.UDPMaxLeases, runtime.SetUDPPolicy)
470 - applyOptionalPolicy(s.TCPPortEnabled, s.TCPPortMaxLeases, runtime.IsTCPPortEnabled, runtime.TCPPortMaxLeases, runtime.SetTCPPortPolicy)
474 + applyOptionalPolicy(s.UDPEnabled, s.UDPMaxLeases, runtime.IsUDPEnabled, runtime.UDPMaxLeases, server.SetUDPPolicy)
475 + applyOptionalPolicy(s.TCPPortEnabled, s.TCPPortMaxLeases, runtime.IsTCPPortEnabled, runtime.TCPPortMaxLeases, server.SetTCPPortPolicy)
476 return nil
477 }
cmd/relay-server/frontend.go
+4 -4
@@ -19,7 +19,7 @@ import (
19
20 "github.com/gosuda/portal-tunnel/v2/cmd/portal-tunnel/installer"
21 "github.com/gosuda/portal-tunnel/v2/portal"
22 - portalauth "github.com/gosuda/portal-tunnel/v2/portal/auth"
22 + "github.com/gosuda/portal-tunnel/v2/portal/auth"
23 "github.com/gosuda/portal-tunnel/v2/portal/identity"
24 "github.com/gosuda/portal-tunnel/v2/types"
25 "github.com/gosuda/portal-tunnel/v2/utils"
@@ -36,7 +36,7 @@ var embeddedDistFS embed.FS
36 type Frontend struct {
37 distFS readDirFileFS
38 server *portal.Server
39 - auth *portalauth.WalletAuthenticator
39 + auth *auth.WalletAuthenticator
40 adminSettingsPath string
41 thumbnails *thumbnailService
42
@@ -57,13 +57,13 @@ func NewFrontend(server *portal.Server, identityPath string, defaultLandingPageE
57 if adminSettingsPath == "" {
58 return nil, errors.New("frontend requires identity path")
59 }
60 - state, err := loadAdminState(adminSettingsPath, runtime)
60 + state, err := loadAdminState(adminSettingsPath, server)
61 if err != nil {
62 return nil, err
63 }
64 relayIdentity := server.RelayIdentity()
65 allowedWallets := append([]string{relayIdentity.Address}, adminWallets...)
66 - authenticator, err := portalauth.NewWalletAuthenticator(portalauth.WalletAuthConfig{
66 + authenticator, err := auth.NewWalletAuthenticator(auth.WalletAuthConfig{
67 AllowedAddresses: allowedWallets,
68 Statement: "Sign in to Portal relay admin",
69 })
portal/acme/acme.go
+12 -10
@@ -72,8 +72,7 @@ type Manager struct {
72 stopOnce sync.Once
73 dnssecLogOnce sync.Once
74 ensLogOnce sync.Once
75 - ensStatusMu sync.RWMutex
76 - ensStatus types.ENSStatus
75 + ensStatus *utils.Snapshot[types.ENSStatus]
76 trackedMu sync.Mutex
77 echMu sync.Mutex
78 echRecords map[string]HTTPSRecord
@@ -185,14 +184,14 @@ func NewManager(cfg Config) (*Manager, error) {
184 return &Manager{
185 cfg: cfg,
186 stopCh: make(chan struct{}),
188 - ensStatus: newENSStatus(cfg, nil),
187 + ensStatus: utils.NewSnapshot(newENSStatus(cfg, nil)),
188 }, nil
189 }
190
191 manager := &Manager{
192 cfg: cfg,
193 stopCh: make(chan struct{}),
195 - ensStatus: newENSStatus(cfg, nil),
194 + ensStatus: utils.NewSnapshot(newENSStatus(cfg, nil)),
195 }
196
197 acmeDNS, err := NewDNSProvider(cfg.DNSProvider, cfg)
@@ -224,9 +223,10 @@ func (m *Manager) ENSStatus() types.ENSStatus {
223 if m == nil {
224 return types.ENSStatus{}
225 }
227 - m.ensStatusMu.RLock()
228 - status := m.ensStatus
229 - m.ensStatusMu.RUnlock()
226 + status := types.ENSStatus{}
227 + if m.ensStatus != nil {
228 + status = m.ensStatus.Load()
229 + }
230 if status.Provider == "" && m.dns != nil {
231 status.Provider = m.dns.Name()
232 }
@@ -246,9 +246,11 @@ func (m *Manager) setENSStatus(state, record, message string, syncErr error) {
246 status.LastError = syncErr.Error()
247 }
248
249 - m.ensStatusMu.Lock()
250 - m.ensStatus = status
251 - m.ensStatusMu.Unlock()
249 + if m.ensStatus == nil {
250 + m.ensStatus = utils.NewSnapshot(status)
251 + return
252 + }
253 + m.ensStatus.Store(status)
254 }
255
256 func ensDNSSECVerified(state string) bool {
portal/acme/cloudflare/provider.go
+14 -14
@@ -7,7 +7,6 @@ import (
7 "net/http"
8 "net/url"
9 "strings"
10 - "sync"
10
11 "github.com/go-acme/lego/v4/challenge"
12 "github.com/go-acme/lego/v4/providers/dns/cloudflare"
@@ -22,8 +21,7 @@ const (
21 type Provider struct {
22 token string
23
25 - zoneMu sync.RWMutex
26 - zones map[string]string
24 + zones *utils.Snapshot[map[string]string]
25 }
26
27 type apiError struct {
@@ -80,7 +78,10 @@ type dnssecResult struct {
78 }
79
80 func New(token string) *Provider {
83 - return &Provider{token: strings.TrimSpace(token)}
81 + return &Provider{
82 + token: strings.TrimSpace(token),
83 + zones: utils.NewSnapshot(map[string]string{}, utils.CloneMap[string, string]),
84 + }
85 }
86
87 func (p *Provider) Name() string {
@@ -366,14 +367,12 @@ func (p *Provider) findZoneID(ctx context.Context, domain string) (string, error
367 domain = utils.NormalizeHostname(domain)
368 candidates := utils.DomainCandidates(domain)
369
369 - p.zoneMu.RLock()
370 + zones := p.zones.Load()
371 for _, candidate := range candidates {
371 - if zoneID := p.zones[candidate]; zoneID != "" {
372 - p.zoneMu.RUnlock()
372 + if zoneID := zones[candidate]; zoneID != "" {
373 return zoneID, nil
374 }
375 }
376 - p.zoneMu.RUnlock()
376
377 for _, candidate := range candidates {
378 zones, err := listZones(ctx, p.token, candidate)
@@ -386,12 +385,13 @@ func (p *Provider) findZoneID(ctx context.Context, domain string) (string, error
385 if zoneID == "" {
386 continue
387 }
389 - p.zoneMu.Lock()
390 - if p.zones == nil {
391 - p.zones = make(map[string]string)
392 - }
393 - p.zones[utils.NormalizeHostname(z.Name)] = zoneID
394 - p.zoneMu.Unlock()
388 + zoneName := utils.NormalizeHostname(z.Name)
389 + p.zones.UpdateCopy(func(zones *map[string]string) {
390 + if *zones == nil {
391 + *zones = make(map[string]string)
392 + }
393 + (*zones)[zoneName] = zoneID
394 + })
395 return zoneID, nil
396 }
397 }
portal/acme/gcloud/provider.go
+17 -20
@@ -7,7 +7,6 @@ import (
7 "net/http"
8 "strconv"
9 "strings"
10 - "sync"
10 "time"
11
12 "cloud.google.com/go/compute/metadata"
@@ -32,9 +31,8 @@ type Config struct {
31 }
32
33 type Provider struct {
35 - cfg Config
36 - zoneMu sync.RWMutex
37 - zones map[string]string
34 + cfg Config
35 + zones *utils.Snapshot[map[string]string]
36 }
37
38 type runtimeConfig struct {
@@ -49,6 +47,7 @@ func New(cfg Config) *Provider {
47 ProjectID: strings.TrimSpace(cfg.ProjectID),
48 ManagedZone: strings.TrimSpace(cfg.ManagedZone),
49 },
50 + zones: utils.NewSnapshot(map[string]string{}, utils.CloneMap[string, string]),
51 }
52 }
53
@@ -436,14 +435,12 @@ func (p *Provider) findManagedZone(ctx context.Context, service *dns.Service, pr
435 domain = utils.NormalizeHostname(domain)
436 candidates := utils.DomainCandidates(domain)
437
439 - p.zoneMu.RLock()
438 + zones := p.zones.Load()
439 for _, candidate := range candidates {
441 - if zoneName := p.zones[candidate]; zoneName != "" {
442 - p.zoneMu.RUnlock()
440 + if zoneName := zones[candidate]; zoneName != "" {
441 return &dns.ManagedZone{Name: zoneName, DnsName: fqdn(candidate)}, nil
442 }
443 }
446 - p.zoneMu.RUnlock()
444
445 if explicit = strings.TrimSpace(explicit); explicit != "" {
446 zone, err := service.ManagedZones.Get(projectID, explicit).Context(ctx).Do()
@@ -454,12 +451,12 @@ func (p *Provider) findManagedZone(ctx context.Context, service *dns.Service, pr
451 return nil, err
452 }
453 if zoneDomain := utils.NormalizeHostname(zone.DnsName); zoneDomain != "" && zone.Name != "" {
457 - p.zoneMu.Lock()
458 - if p.zones == nil {
459 - p.zones = make(map[string]string)
460 - }
461 - p.zones[zoneDomain] = zone.Name
462 - p.zoneMu.Unlock()
454 + p.zones.UpdateCopy(func(zones *map[string]string) {
455 + if *zones == nil {
456 + *zones = make(map[string]string)
457 + }
458 + (*zones)[zoneDomain] = zone.Name
459 + })
460 }
461 return zone, nil
462 }
@@ -474,12 +471,12 @@ func (p *Provider) findManagedZone(ctx context.Context, service *dns.Service, pr
471 continue
472 }
473 if zone.Name != "" {
477 - p.zoneMu.Lock()
478 - if p.zones == nil {
479 - p.zones = make(map[string]string)
480 - }
481 - p.zones[candidate] = zone.Name
482 - p.zoneMu.Unlock()
474 + p.zones.UpdateCopy(func(zones *map[string]string) {
475 + if *zones == nil {
476 + *zones = make(map[string]string)
477 + }
478 + (*zones)[candidate] = zone.Name
479 + })
480 }
481 return zone, nil
482 }
portal/acme/route53/provider.go
+13 -16
@@ -6,7 +6,6 @@ import (
6 "fmt"
7 "strconv"
8 "strings"
9 - "sync"
9 "time"
10
11 "github.com/aws/aws-sdk-go-v2/aws"
@@ -35,9 +34,8 @@ type Config struct {
34 }
35
36 type Provider struct {
38 - cfg Config
39 - zoneMu sync.RWMutex
40 - zones map[string]string
37 + cfg Config
38 + zones *utils.Snapshot[map[string]string]
39 }
40
41 func New(cfg Config) *Provider {
@@ -50,6 +48,7 @@ func New(cfg Config) *Provider {
48 HostedZoneID: normalizeZoneID(cfg.HostedZoneID),
49 KMSKeyARN: strings.TrimSpace(cfg.KMSKeyARN),
50 },
51 + zones: utils.NewSnapshot(map[string]string{}, utils.CloneMap[string, string]),
52 }
53 }
54
@@ -371,14 +370,12 @@ func (p *Provider) findHostedZoneID(ctx context.Context, client *awsroute53.Clie
370 return "", fmt.Errorf("invalid base domain for hosted zone lookup: %q", domain)
371 }
372
374 - p.zoneMu.RLock()
373 + zones := p.zones.Load()
374 for _, candidate := range candidates {
376 - if zoneID := p.zones[candidate]; zoneID != "" {
377 - p.zoneMu.RUnlock()
375 + if zoneID := zones[candidate]; zoneID != "" {
376 return zoneID, nil
377 }
378 }
381 - p.zoneMu.RUnlock()
379
380 zonesByName := make(map[string]string)
381 paginator := awsroute53.NewListHostedZonesPaginator(client, &awsroute53.ListHostedZonesInput{})
@@ -401,14 +398,14 @@ func (p *Provider) findHostedZoneID(ctx context.Context, client *awsroute53.Clie
398 }
399
400 if len(zonesByName) > 0 {
404 - p.zoneMu.Lock()
405 - if p.zones == nil {
406 - p.zones = make(map[string]string)
407 - }
408 - for zoneName, zoneID := range zonesByName {
409 - p.zones[zoneName] = zoneID
410 - }
411 - p.zoneMu.Unlock()
401 + p.zones.UpdateCopy(func(zones *map[string]string) {
402 + if *zones == nil {
403 + *zones = make(map[string]string)
404 + }
405 + for zoneName, zoneID := range zonesByName {
406 + (*zones)[zoneName] = zoneID
407 + }
408 + })
409 }
410
411 for _, candidate := range candidates {
portal/acme/vultr/provider.go
+13 -14
@@ -6,7 +6,6 @@ import (
6 "fmt"
7 "strconv"
8 "strings"
9 - "sync"
9
10 "github.com/go-acme/lego/v4/challenge"
11 legovultr "github.com/go-acme/lego/v4/providers/dns/vultr"
@@ -21,12 +20,14 @@ const defaultRecordTTL = 60
20 type Provider struct {
21 apiKey string
22
24 - zoneMu sync.RWMutex
25 - zones map[string]string
23 + zones *utils.Snapshot[map[string]string]
24 }
25
26 func New(apiKey string) *Provider {
29 - return &Provider{apiKey: strings.TrimSpace(apiKey)}
27 + return &Provider{
28 + apiKey: strings.TrimSpace(apiKey),
29 + zones: utils.NewSnapshot(map[string]string{}, utils.CloneMap[string, string]),
30 + }
31 }
32
33 func (p *Provider) Name() string {
@@ -290,14 +291,12 @@ func (p *Provider) findZone(ctx context.Context, client *govultr.Client, domain
291 domain = utils.NormalizeHostname(domain)
292 candidates := utils.DomainCandidates(domain)
293
293 - p.zoneMu.RLock()
294 + zones := p.zones.Load()
295 for _, candidate := range candidates {
295 - if zone := p.zones[candidate]; zone != "" {
296 - p.zoneMu.RUnlock()
296 + if zone := zones[candidate]; zone != "" {
297 return zone, nil
298 }
299 }
300 - p.zoneMu.RUnlock()
300
301 listOptions := &govultr.ListOptions{PerPage: 100}
302 for {
@@ -311,12 +310,12 @@ func (p *Provider) findZone(ctx context.Context, client *govultr.Client, domain
310 if zone != candidate {
311 continue
312 }
314 - p.zoneMu.Lock()
315 - if p.zones == nil {
316 - p.zones = make(map[string]string)
317 - }
318 - p.zones[candidate] = zone
319 - p.zoneMu.Unlock()
313 + p.zones.UpdateCopy(func(zones *map[string]string) {
314 + if *zones == nil {
315 + *zones = make(map[string]string)
316 + }
317 + (*zones)[candidate] = zone
318 + })
319 return zone, nil
320 }
321 }
portal/api_server.go
+11 -9
@@ -104,13 +104,13 @@ func (s *Server) apiHandler(base *http.ServeMux, keylessSignerHandler http.Handl
104 case types.PathSDKConnect:
105 s.handleConnect(w, r)
106 case types.PathDiscovery:
107 - if !s.cfg.DiscoveryEnabled {
107 + if !s.config().DiscoveryEnabled {
108 base.ServeHTTP(w, r)
109 return
110 }
111 s.handleRelayDiscovery(w, r)
112 case types.PathDiscoveryAnnounce:
113 - if !s.cfg.DiscoveryEnabled {
113 + if !s.config().DiscoveryEnabled {
114 base.ServeHTTP(w, r)
115 return
116 }
@@ -212,7 +212,8 @@ func (s *Server) handleRelayDiscoveryAnnounce(w http.ResponseWriter, r *http.Req
212 fmt.Sprintf("self-announce rejected: host %q is local-only", host))
213 return
214 }
215 - if selfURL, err := utils.NormalizeRelayURL(s.cfg.PortalURL); err == nil && desc.APIHTTPSAddr == selfURL {
215 + cfg := s.config()
216 + if selfURL, err := utils.NormalizeRelayURL(cfg.PortalURL); err == nil && desc.APIHTTPSAddr == selfURL {
217 utils.WriteAPIError(w, http.StatusBadRequest, types.APIErrorCodeInvalidRequest,
218 fmt.Sprintf("self-announce rejected: %q matches receiving relay url", desc.APIHTTPSAddr))
219 return
@@ -300,7 +301,7 @@ func (s *Server) handleRegister(w http.ResponseWriter, r *http.Request) {
301 writeAPIErrorResponse(w, err)
302 return
303 }
303 - s.registry.promoteECHDNS(record, s.acmeManager, s.cfg.SNIPort)
304 + s.registry.promoteECHDNS(record, s.acmeManager, s.config().SNIPort)
305
306 utils.WriteAPIData(w, http.StatusCreated, resp)
307 }
@@ -338,11 +339,11 @@ func (s *Server) handleRegisterChallenge(w http.ResponseWriter, r *http.Request)
339 utils.WriteAPIError(w, http.StatusServiceUnavailable, types.APIErrorCodeFeatureUnavailable, errFeatureUnavailable.Error())
340 return
341 }
341 - if req.UDPEnabled && (!s.cfg.UDPEnabled || s.group != nil && s.quicBackhaul == nil) {
342 + if req.UDPEnabled && !s.supportsUDP() {
343 utils.WriteAPIError(w, http.StatusServiceUnavailable, types.APIErrorCodeFeatureUnavailable, errFeatureUnavailable.Error())
344 return
345 }
345 - if req.TCPEnabled && !s.cfg.TCPEnabled {
346 + if req.TCPEnabled && !s.supportsTCP() {
347 utils.WriteAPIError(w, http.StatusServiceUnavailable, types.APIErrorCodeFeatureUnavailable, errFeatureUnavailable.Error())
348 return
349 }
@@ -429,7 +430,8 @@ func (s *Server) handleHop(w http.ResponseWriter, r *http.Request) {
430 utils.InvalidRequestError(err).Write(w)
431 return
432 }
432 - if route.RelayURL != s.cfg.PortalURL {
433 + cfg := s.config()
434 + if route.RelayURL != cfg.PortalURL {
435 utils.WriteAPIError(w, http.StatusForbidden, types.APIErrorCodeUnauthorized, "hop route relay url does not match receiving relay")
436 return
437 }
@@ -484,7 +486,7 @@ func (s *Server) handleHop(w http.ResponseWriter, r *http.Request) {
486 writeAPIErrorResponse(w, err)
487 return
488 }
487 - s.registry.promoteECHDNS(record, s.acmeManager, s.cfg.SNIPort)
489 + s.registry.promoteECHDNS(record, s.acmeManager, cfg.SNIPort)
490 var accessToken string
491 if record.isPublicEntry() {
492 accessToken, err = s.registry.issueLeaseAccessToken(record, now)
@@ -495,7 +497,7 @@ func (s *Server) handleHop(w http.ResponseWriter, r *http.Request) {
497 }
498 utils.WriteAPIData(w, http.StatusOK, types.HopRouteResponse{
499 AccessToken: accessToken,
498 - SNIPort: s.cfg.SNIPort,
500 + SNIPort: cfg.SNIPort,
501 })
502 }
503
portal/policy/approver.go
+83 -42
@@ -2,7 +2,9 @@ package policy
2
3 import (
4 "fmt"
5 - "sync"
5 + "maps"
6 +
7 + "github.com/gosuda/portal-tunnel/v2/utils"
8 )
9
10 type Mode string
@@ -13,91 +15,126 @@ const (
15 )
16
17 type Approver struct {
18 + state *utils.Snapshot[approvalState]
19 +}
20 +
21 +type approvalState struct {
22 approvedKeys map[string]struct{}
23 deniedKeys map[string]struct{}
24 approvalMode Mode
19 - mu sync.RWMutex
25 }
26
22 -func NewApprover() *Approver {
23 - return &Approver{
27 +func newApprovalState() approvalState {
28 + return approvalState{
29 approvalMode: ModeAuto,
30 approvedKeys: make(map[string]struct{}),
31 deniedKeys: make(map[string]struct{}),
32 }
33 }
34
35 +func (state approvalState) snapshot() approvalState {
36 + state.approvedKeys = maps.Clone(state.approvedKeys)
37 + state.deniedKeys = maps.Clone(state.deniedKeys)
38 + return state
39 +}
40 +
41 +func NewApprover() *Approver {
42 + return &Approver{
43 + state: utils.NewSnapshot(newApprovalState(), approvalState.snapshot),
44 + }
45 +}
46 +
47 +func (a *Approver) current() approvalState {
48 + if a == nil || a.state == nil {
49 + return newApprovalState()
50 + }
51 + return a.state.Load()
52 +}
53 +
54 func (a *Approver) Mode() Mode {
31 - a.mu.RLock()
32 - defer a.mu.RUnlock()
33 - return a.approvalMode
55 + return a.current().approvalMode
56 }
57
58 func (a *Approver) SetMode(mode Mode) error {
59 if mode != ModeAuto && mode != ModeManual {
60 return fmt.Errorf("invalid approval mode: %q", mode)
61 }
40 - a.mu.Lock()
41 - defer a.mu.Unlock()
42 - a.approvalMode = mode
62 + if a == nil || a.state == nil {
63 + return nil
64 + }
65 + a.state.UpdateCopy(func(state *approvalState) {
66 + state.approvalMode = mode
67 + })
68 return nil
69 }
70
71 func (a *Approver) IsApproved(key string) bool {
47 - a.mu.RLock()
48 - defer a.mu.RUnlock()
49 - _, ok := a.approvedKeys[key]
72 + _, ok := a.current().approvedKeys[key]
73 return ok
74 }
75
76 func (a *Approver) Approve(key string) {
54 - a.mu.Lock()
55 - defer a.mu.Unlock()
56 - a.approvedKeys[key] = struct{}{}
57 - delete(a.deniedKeys, key)
77 + if a == nil || a.state == nil {
78 + return
79 + }
80 + a.state.UpdateCopy(func(state *approvalState) {
81 + if state.approvedKeys == nil {
82 + state.approvedKeys = make(map[string]struct{})
83 + }
84 + state.approvedKeys[key] = struct{}{}
85 + delete(state.deniedKeys, key)
86 + })
87 }
88
89 func (a *Approver) Revoke(key string) {
61 - a.mu.Lock()
62 - defer a.mu.Unlock()
63 - delete(a.approvedKeys, key)
90 + if a == nil || a.state == nil {
91 + return
92 + }
93 + a.state.UpdateCopy(func(state *approvalState) {
94 + delete(state.approvedKeys, key)
95 + })
96 }
97
98 func (a *Approver) ApprovedKeys() []string {
67 - a.mu.RLock()
68 - defer a.mu.RUnlock()
69 - out := make([]string, 0, len(a.approvedKeys))
70 - for key := range a.approvedKeys {
99 + approvedKeys := a.current().approvedKeys
100 + out := make([]string, 0, len(approvedKeys))
101 + for key := range approvedKeys {
102 out = append(out, key)
103 }
104 return out
105 }
106
107 func (a *Approver) IsDenied(key string) bool {
77 - a.mu.RLock()
78 - defer a.mu.RUnlock()
79 - _, ok := a.deniedKeys[key]
108 + _, ok := a.current().deniedKeys[key]
109 return ok
110 }
111
112 func (a *Approver) Deny(key string) {
84 - a.mu.Lock()
85 - defer a.mu.Unlock()
86 - a.deniedKeys[key] = struct{}{}
87 - delete(a.approvedKeys, key)
113 + if a == nil || a.state == nil {
114 + return
115 + }
116 + a.state.UpdateCopy(func(state *approvalState) {
117 + if state.deniedKeys == nil {
118 + state.deniedKeys = make(map[string]struct{})
119 + }
120 + state.deniedKeys[key] = struct{}{}
121 + delete(state.approvedKeys, key)
122 + })
123 }
124
125 func (a *Approver) Undeny(key string) {
91 - a.mu.Lock()
92 - defer a.mu.Unlock()
93 - delete(a.deniedKeys, key)
126 + if a == nil || a.state == nil {
127 + return
128 + }
129 + a.state.UpdateCopy(func(state *approvalState) {
130 + delete(state.deniedKeys, key)
131 + })
132 }
133
134 func (a *Approver) DeniedKeys() []string {
97 - a.mu.RLock()
98 - defer a.mu.RUnlock()
99 - out := make([]string, 0, len(a.deniedKeys))
100 - for key := range a.deniedKeys {
135 + deniedKeys := a.current().deniedKeys
136 + out := make([]string, 0, len(deniedKeys))
137 + for key := range deniedKeys {
138 out = append(out, key)
139 }
140 return out
@@ -125,8 +162,12 @@ func (a *Approver) SetDecisions(approvedKeys, deniedKeys []string) {
162 denied[key] = struct{}{}
163 }
164
128 - a.mu.Lock()
129 - a.approvedKeys = approved
130 - a.deniedKeys = denied
131 - a.mu.Unlock()
165 + if a.state == nil {
166 + return
167 + }
168 + a.state.Update(func(state approvalState) approvalState {
169 + state.approvedKeys = approved
170 + state.deniedKeys = denied
171 + return state
172 + })
173 }
portal/policy/bps_manager.go
+42 -23
@@ -1,20 +1,21 @@
1 package policy
2
3 import (
4 - "maps"
4 "sync"
5 "time"
6 +
7 + "github.com/gosuda/portal-tunnel/v2/utils"
8 )
9
10 type BPSManager struct {
10 - identityBPS map[string]int64
11 + identityBPS *utils.Snapshot[map[string]int64]
12 identityLimiters map[string]*bpsLimiter
13 mu sync.RWMutex
14 }
15
16 func NewBPSManager() *BPSManager {
17 return &BPSManager{
17 - identityBPS: make(map[string]int64),
18 + identityBPS: utils.NewSnapshot(map[string]int64{}, utils.CloneMap[string, int64]),
19 identityLimiters: make(map[string]*bpsLimiter),
20 }
21 }
@@ -23,10 +24,10 @@ func (m *BPSManager) IdentityBPS(key string) int64 {
24 if m == nil || key == "" {
25 return 0
26 }
26 -
27 - m.mu.RLock()
28 - defer m.mu.RUnlock()
29 - return m.identityBPS[key]
27 + if m.identityBPS == nil {
28 + return 0
29 + }
30 + return m.identityBPS.Load()[key]
31 }
32
33 func (m *BPSManager) SetIdentityBPS(key string, bps int64) {
@@ -34,14 +35,24 @@ func (m *BPSManager) SetIdentityBPS(key string, bps int64) {
35 return
36 }
37
37 - m.mu.Lock()
38 - defer m.mu.Unlock()
38 + if m.identityBPS == nil {
39 + return
40 + }
41 if bps <= 0 {
40 - delete(m.identityBPS, key)
42 + m.identityBPS.UpdateCopy(func(limits *map[string]int64) {
43 + delete(*limits, key)
44 + })
45 + m.mu.Lock()
46 delete(m.identityLimiters, key)
47 + m.mu.Unlock()
48 return
49 }
44 - m.identityBPS[key] = bps
50 + m.identityBPS.UpdateCopy(func(limits *map[string]int64) {
51 + if *limits == nil {
52 + *limits = make(map[string]int64)
53 + }
54 + (*limits)[key] = bps
55 + })
56 }
57
58 func (m *BPSManager) DeleteIdentityBPS(key string) {
@@ -49,10 +60,14 @@ func (m *BPSManager) DeleteIdentityBPS(key string) {
60 return
61 }
62
63 + if m.identityBPS != nil {
64 + m.identityBPS.UpdateCopy(func(limits *map[string]int64) {
65 + delete(*limits, key)
66 + })
67 + }
68 m.mu.Lock()
53 - defer m.mu.Unlock()
54 - delete(m.identityBPS, key)
69 delete(m.identityLimiters, key)
70 + m.mu.Unlock()
71 }
72
73 func (m *BPSManager) IdentityBPSLimits() map[string]int64 {
@@ -60,12 +75,10 @@ func (m *BPSManager) IdentityBPSLimits() map[string]int64 {
75 return nil
76 }
77
63 - m.mu.RLock()
64 - defer m.mu.RUnlock()
65 -
66 - out := make(map[string]int64, len(m.identityBPS))
67 - maps.Copy(out, m.identityBPS)
68 - return out
78 + if m.identityBPS == nil {
79 + return nil
80 + }
81 + return m.identityBPS.Load()
82 }
83
84 func (m *BPSManager) SetIdentityBPSLimits(limits map[string]int64) {
@@ -81,8 +94,10 @@ func (m *BPSManager) SetIdentityBPSLimits(limits map[string]int64) {
94 next[key] = bps
95 }
96
97 + if m.identityBPS != nil {
98 + m.identityBPS.Store(next)
99 + }
100 m.mu.Lock()
85 - m.identityBPS = next
101 m.identityLimiters = make(map[string]*bpsLimiter)
102 m.mu.Unlock()
103 }
@@ -107,18 +122,22 @@ func (m *BPSManager) ThrottleIdentityBPS(key string, maxBytes int) int {
122 }
123
124 func (m *BPSManager) identityLimiter(key string) (int64, *bpsLimiter) {
125 + bps := m.IdentityBPS(key)
126 + if bps <= 0 {
127 + return 0, nil
128 + }
129 +
130 m.mu.RLock()
111 - bps := m.identityBPS[key]
131 limiter := m.identityLimiters[key]
132 m.mu.RUnlock()
114 - if bps <= 0 || limiter != nil {
133 + if limiter != nil {
134 return bps, limiter
135 }
136
137 m.mu.Lock()
138 defer m.mu.Unlock()
139
121 - bps = m.identityBPS[key]
140 + bps = m.IdentityBPS(key)
141 if bps <= 0 {
142 return 0, nil
143 }
portal/policy/ip_filter.go
+37 -19
@@ -4,10 +4,12 @@ import (
4 "slices"
5 "strings"
6 "sync"
7 +
8 + "github.com/gosuda/portal-tunnel/v2/utils"
9 )
10
11 type IPFilter struct {
10 - bannedIPs map[string]struct{}
12 + bannedIPs *utils.Snapshot[map[string]struct{}]
13 identityToIP map[string]string
14 ipToIdentities map[string][]string
15 mu sync.RWMutex
@@ -15,52 +17,68 @@ type IPFilter struct {
17
18 func NewIPFilter() *IPFilter {
19 return &IPFilter{
18 - bannedIPs: make(map[string]struct{}),
20 + bannedIPs: utils.NewSnapshot(map[string]struct{}{}, utils.CloneMap[string, struct{}]),
21 identityToIP: make(map[string]string),
22 ipToIdentities: make(map[string][]string),
23 }
24 }
25
26 func (f *IPFilter) BanIP(ip string) {
25 - f.mu.Lock()
26 - defer f.mu.Unlock()
27 - f.bannedIPs[strings.TrimSpace(ip)] = struct{}{}
27 + if f == nil || f.bannedIPs == nil {
28 + return
29 + }
30 + ip = strings.TrimSpace(ip)
31 + f.bannedIPs.UpdateCopy(func(ips *map[string]struct{}) {
32 + if *ips == nil {
33 + *ips = make(map[string]struct{})
34 + }
35 + (*ips)[ip] = struct{}{}
36 + })
37 }
38
39 func (f *IPFilter) UnbanIP(ip string) {
31 - f.mu.Lock()
32 - defer f.mu.Unlock()
33 - delete(f.bannedIPs, strings.TrimSpace(ip))
40 + if f == nil || f.bannedIPs == nil {
41 + return
42 + }
43 + ip = strings.TrimSpace(ip)
44 + f.bannedIPs.UpdateCopy(func(ips *map[string]struct{}) {
45 + delete(*ips, ip)
46 + })
47 }
48
49 func (f *IPFilter) IsIPBanned(ip string) bool {
37 - f.mu.RLock()
38 - defer f.mu.RUnlock()
39 - _, ok := f.bannedIPs[strings.TrimSpace(ip)]
50 + if f == nil || f.bannedIPs == nil {
51 + return false
52 + }
53 + _, ok := f.bannedIPs.Load()[strings.TrimSpace(ip)]
54 return ok
55 }
56
57 func (f *IPFilter) BannedIPs() []string {
44 - f.mu.RLock()
45 - defer f.mu.RUnlock()
46 - out := make([]string, 0, len(f.bannedIPs))
47 - for ip := range f.bannedIPs {
58 + if f == nil || f.bannedIPs == nil {
59 + return nil
60 + }
61 + ips := f.bannedIPs.Load()
62 + out := make([]string, 0, len(ips))
63 + for ip := range ips {
64 out = append(out, ip)
65 }
66 return out
67 }
68
69 func (f *IPFilter) SetBannedIPs(ips []string) {
54 - f.mu.Lock()
55 - defer f.mu.Unlock()
56 - f.bannedIPs = make(map[string]struct{}, len(ips))
70 + if f == nil || f.bannedIPs == nil {
71 + return
72 + }
73 + bannedIPs := make(map[string]struct{}, len(ips))
74 for _, ip := range ips {
75 ip = strings.TrimSpace(ip)
76 if ip == "" {
77 continue
78 }
62 - f.bannedIPs[ip] = struct{}{}
79 + bannedIPs[ip] = struct{}{}
80 }
81 + f.bannedIPs.Store(bannedIPs)
82 }
83
84 func (f *IPFilter) RegisterIdentityIP(key, ip string) {
portal/policy/proxy_trust.go
+5 -6
@@ -44,12 +44,11 @@ func (r *Runtime) ExtractClientIP(req *http.Request) string {
44 return ""
45 }
46
47 - r.mu.RLock()
48 - trustProxyHeaders := r.trustProxyHeaders
49 - trustedProxyCIDRs := append([]*net.IPNet(nil), r.trustedProxyCIDRs...)
50 - r.mu.RUnlock()
51 -
52 - if trustProxyHeaders && isTrustedProxyRemoteAddr(req.RemoteAddr, trustedProxyCIDRs) {
47 + cfg := runtimeConfig{}
48 + if r.config != nil {
49 + cfg = r.config.Load()
50 + }
51 + if cfg.trustProxyHeaders && isTrustedProxyRemoteAddr(req.RemoteAddr, cfg.trustedProxyCIDRs) {
52 if xff := req.Header.Get("X-Forwarded-For"); xff != "" {
53 if before, _, ok := strings.Cut(xff, ","); ok {
54 if ip := normalizeClientIPCandidate(before); ip != "" {
portal/policy/runtime.go
+80 -55
@@ -3,7 +3,6 @@ package policy
3 import (
4 "fmt"
5 "net"
6 - "sync"
6
7 "github.com/gosuda/portal-tunnel/v2/utils"
8 )
@@ -21,27 +20,37 @@ func (p *PortPolicy) Set(enabled bool, maxLeases int) {
20 p.maxLeases = maxLeases
21 }
22
23 +type runtimeConfig struct {
24 + udp PortPolicy
25 + tcpPort PortPolicy
26 + trustProxyHeaders bool
27 + trustedProxyCIDRs []*net.IPNet
28 +}
29 +
30 +func (cfg runtimeConfig) snapshot() runtimeConfig {
31 + cfg.trustedProxyCIDRs = utils.CloneSlice(cfg.trustedProxyCIDRs)
32 + return cfg
33 +}
34 +
35 type Runtime struct {
36 approver *Approver
37 bpsManager *BPSManager
38 ipFilter *IPFilter
28 - bannedIdentityKeys map[string]struct{}
29 - udp PortPolicy
30 - tcpPort PortPolicy
31 - trustProxyHeaders bool
32 - trustedProxyCIDRs []*net.IPNet
33 - mu sync.RWMutex
39 + config *utils.Snapshot[runtimeConfig]
40 + bannedIdentityKeys *utils.Snapshot[map[string]struct{}]
41 }
42
43 func NewRuntime(udpEnabled, tcpPortEnabled bool, trustProxyHeaders bool, rawTrustedProxyCIDRs string) (*Runtime, error) {
44 runtime := &Runtime{
38 - approver: NewApprover(),
39 - bpsManager: NewBPSManager(),
40 - ipFilter: NewIPFilter(),
41 - bannedIdentityKeys: make(map[string]struct{}),
45 + approver: NewApprover(),
46 + bpsManager: NewBPSManager(),
47 + ipFilter: NewIPFilter(),
48 + config: utils.NewSnapshot(runtimeConfig{
49 + udp: PortPolicy{enabled: udpEnabled},
50 + tcpPort: PortPolicy{enabled: tcpPortEnabled},
51 + }, runtimeConfig.snapshot),
52 + bannedIdentityKeys: utils.NewSnapshot(map[string]struct{}{}, utils.CloneMap[string, struct{}]),
53 }
43 - runtime.udp.Set(udpEnabled, 0)
44 - runtime.tcpPort.Set(tcpPortEnabled, 0)
54 if err := runtime.SetProxyTrust(trustProxyHeaders, rawTrustedProxyCIDRs); err != nil {
55 return nil, err
56 }
@@ -61,44 +70,50 @@ func (r *Runtime) BPSManager() *BPSManager {
70 }
71
72 func (r *Runtime) BanIdentity(key string) {
64 - if key == "" {
73 + if r == nil || r.bannedIdentityKeys == nil || key == "" {
74 return
75 }
67 - r.mu.Lock()
68 - defer r.mu.Unlock()
69 - r.bannedIdentityKeys[key] = struct{}{}
76 + r.bannedIdentityKeys.UpdateCopy(func(keys *map[string]struct{}) {
77 + if *keys == nil {
78 + *keys = make(map[string]struct{})
79 + }
80 + (*keys)[key] = struct{}{}
81 + })
82 }
83
84 func (r *Runtime) UnbanIdentity(key string) {
73 - if key == "" {
85 + if r == nil || r.bannedIdentityKeys == nil || key == "" {
86 return
87 }
76 - r.mu.Lock()
77 - defer r.mu.Unlock()
78 - delete(r.bannedIdentityKeys, key)
88 + r.bannedIdentityKeys.UpdateCopy(func(keys *map[string]struct{}) {
89 + delete(*keys, key)
90 + })
91 }
92
93 func (r *Runtime) IsIdentityBanned(key string) bool {
82 - if key == "" {
94 + if r == nil || r.bannedIdentityKeys == nil || key == "" {
95 return false
96 }
85 - r.mu.RLock()
86 - defer r.mu.RUnlock()
87 - _, ok := r.bannedIdentityKeys[key]
97 + _, ok := r.bannedIdentityKeys.Load()[key]
98 return ok
99 }
100
101 func (r *Runtime) BannedIdentityKeys() []string {
92 - r.mu.RLock()
93 - defer r.mu.RUnlock()
94 - out := make([]string, 0, len(r.bannedIdentityKeys))
95 - for key := range r.bannedIdentityKeys {
102 + if r == nil || r.bannedIdentityKeys == nil {
103 + return nil
104 + }
105 + keys := r.bannedIdentityKeys.Load()
106 + out := make([]string, 0, len(keys))
107 + for key := range keys {
108 out = append(out, key)
109 }
110 return out
111 }
112
113 func (r *Runtime) SetBannedIdentityKeys(keys []string) {
114 + if r == nil || r.bannedIdentityKeys == nil {
115 + return
116 + }
117 bannedIdentityKeys := make(map[string]struct{}, len(keys))
118 for _, key := range keys {
119 if key == "" {
@@ -107,9 +122,7 @@ func (r *Runtime) SetBannedIdentityKeys(keys []string) {
122 bannedIdentityKeys[key] = struct{}{}
123 }
124
110 - r.mu.Lock()
111 - r.bannedIdentityKeys = bannedIdentityKeys
112 - r.mu.Unlock()
125 + r.bannedIdentityKeys.Store(bannedIdentityKeys)
126 }
127
128 func (r *Runtime) EffectiveApproval(key string) bool {
@@ -137,27 +150,35 @@ func (r *Runtime) IsIdentityRoutable(key string) bool {
150 }
151
152 func (r *Runtime) SetUDPPolicy(enabled bool, maxLeases int) {
140 - r.mu.Lock()
141 - r.udp.Set(enabled, maxLeases)
142 - r.mu.Unlock()
153 + if r == nil || r.config == nil {
154 + return
155 + }
156 + r.config.UpdateCopy(func(cfg *runtimeConfig) {
157 + cfg.udp.Set(enabled, maxLeases)
158 + })
159 }
160
161 func (r *Runtime) IsUDPEnabled() bool {
146 - r.mu.RLock()
147 - defer r.mu.RUnlock()
148 - return r.udp.IsEnabled()
162 + if r == nil || r.config == nil {
163 + return false
164 + }
165 + return r.config.Load().udp.IsEnabled()
166 }
167
168 func (r *Runtime) UDPMaxLeases() int {
152 - r.mu.RLock()
153 - defer r.mu.RUnlock()
154 - return r.udp.MaxLeases()
169 + if r == nil || r.config == nil {
170 + return 0
171 + }
172 + return r.config.Load().udp.MaxLeases()
173 }
174
175 func (r *Runtime) SetTCPPortPolicy(enabled bool, maxLeases int) {
158 - r.mu.Lock()
159 - r.tcpPort.Set(enabled, maxLeases)
160 - r.mu.Unlock()
176 + if r == nil || r.config == nil {
177 + return
178 + }
179 + r.config.UpdateCopy(func(cfg *runtimeConfig) {
180 + cfg.tcpPort.Set(enabled, maxLeases)
181 + })
182 }
183
184 func (r *Runtime) SetProxyTrust(trustProxyHeaders bool, rawTrustedProxyCIDRs string) error {
@@ -168,24 +189,28 @@ func (r *Runtime) SetProxyTrust(trustProxyHeaders bool, rawTrustedProxyCIDRs str
189 if err != nil {
190 return fmt.Errorf("parse trusted proxy cidrs: %w", err)
191 }
171 - copied := append([]*net.IPNet(nil), trustedProxyCIDRs...)
172 - r.mu.Lock()
173 - r.trustProxyHeaders = trustProxyHeaders
174 - r.trustedProxyCIDRs = copied
175 - r.mu.Unlock()
192 + if r.config == nil {
193 + r.config = utils.NewSnapshot(runtimeConfig{}, runtimeConfig.snapshot)
194 + }
195 + r.config.UpdateCopy(func(cfg *runtimeConfig) {
196 + cfg.trustProxyHeaders = trustProxyHeaders
197 + cfg.trustedProxyCIDRs = append([]*net.IPNet(nil), trustedProxyCIDRs...)
198 + })
199 return nil
200 }
201
202 func (r *Runtime) IsTCPPortEnabled() bool {
180 - r.mu.RLock()
181 - defer r.mu.RUnlock()
182 - return r.tcpPort.IsEnabled()
203 + if r == nil || r.config == nil {
204 + return false
205 + }
206 + return r.config.Load().tcpPort.IsEnabled()
207 }
208
209 func (r *Runtime) TCPPortMaxLeases() int {
186 - r.mu.RLock()
187 - defer r.mu.RUnlock()
188 - return r.tcpPort.MaxLeases()
210 + if r == nil || r.config == nil {
211 + return 0
212 + }
213 + return r.config.Load().tcpPort.MaxLeases()
214 }
215
216 func (r *Runtime) ForgetIdentity(key string) {
portal/server.go
+78 -24
@@ -103,17 +103,26 @@ func normalizeServerConfig(cfg ServerConfig) (ServerConfig, error) {
103 }
104 }
105
106 - cfg.UDPEnabled = cfg.UDPEnabled && hasPortRange
107 - cfg.TCPEnabled = cfg.TCPEnabled && hasPortRange
106 + cfg.UDPEnabled = cfg.UDPEnabled && cfg.hasLeasePortRange()
107 + cfg.TCPEnabled = cfg.TCPEnabled && cfg.hasLeasePortRange()
108 return cfg, nil
109 }
110
111 +func (cfg ServerConfig) snapshot() ServerConfig {
112 + cfg.Bootstraps = utils.CloneSlice(cfg.Bootstraps)
113 + return cfg
114 +}
115 +
116 +func (cfg ServerConfig) hasLeasePortRange() bool {
117 + return cfg.MinPort > 0 && cfg.MaxPort > 0 && cfg.MinPort <= 65535 && cfg.MaxPort <= 65535 && cfg.MinPort <= cfg.MaxPort
118 +}
119 +
120 type Server struct {
121 cancel context.CancelFunc
122 group *errgroup.Group
123 shutdownOnce sync.Once
124
116 - cfg ServerConfig
125 + cfg *utils.Snapshot[ServerConfig]
126 identity types.RelayIdentity
127 authority identity.Authority
128 acmeManager *acme.Manager
@@ -162,7 +171,7 @@ func NewServer(cfg ServerConfig) (*Server, error) {
171 }
172
173 server := &Server{
165 - cfg: cfg,
174 + cfg: utils.NewSnapshot(cfg, ServerConfig.snapshot),
175 identity: relayIdentity,
176 authority: relayAuthority,
177 registry: registry,
@@ -173,10 +182,52 @@ func NewServer(cfg ServerConfig) (*Server, error) {
182 return server, nil
183 }
184
185 +func (s *Server) config() ServerConfig {
186 + return s.cfg.Load()
187 +}
188 +
189 +func (s *Server) SetUDPPolicy(enabled bool, maxLeases int) {
190 + if enabled && !s.config().hasLeasePortRange() {
191 + enabled = false
192 + }
193 + if runtime := s.PolicyRuntime(); runtime != nil {
194 + runtime.SetUDPPolicy(enabled, maxLeases)
195 + }
196 + s.cfg.UpdateCopy(func(cfg *ServerConfig) {
197 + cfg.UDPEnabled = enabled
198 + })
199 +}
200 +
201 +func (s *Server) SetTCPPortPolicy(enabled bool, maxLeases int) {
202 + if enabled && !s.config().hasLeasePortRange() {
203 + enabled = false
204 + }
205 + if runtime := s.PolicyRuntime(); runtime != nil {
206 + runtime.SetTCPPortPolicy(enabled, maxLeases)
207 + }
208 + s.cfg.UpdateCopy(func(cfg *ServerConfig) {
209 + cfg.TCPEnabled = enabled
210 + })
211 +}
212 +
213 +func (s *Server) supportsUDP() bool {
214 + runtime := s.PolicyRuntime()
215 + if runtime == nil || !runtime.IsUDPEnabled() {
216 + return false
217 + }
218 + return s.group == nil || s.quicBackhaul != nil
219 +}
220 +
221 +func (s *Server) supportsTCP() bool {
222 + runtime := s.PolicyRuntime()
223 + return runtime != nil && runtime.IsTCPPortEnabled()
224 +}
225 +
226 func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
227 if s.group != nil {
228 return errors.New("server already started")
229 }
230 + cfg := s.config()
231 apiTLS, acmeManager, err := s.prepareAPITLS(ctx)
232 if err != nil {
233 return err
@@ -222,11 +273,11 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
273 }()
274 var listenConfig net.ListenConfig
275
225 - apiListener, err = listenConfig.Listen(serverCtx, "tcp", s.cfg.APIListenAddr)
276 + apiListener, err = listenConfig.Listen(serverCtx, "tcp", cfg.APIListenAddr)
277 if err != nil {
278 return fmt.Errorf("listen api: %w", err)
279 }
229 - sniListener, err = listenConfig.Listen(serverCtx, "tcp", s.cfg.SNIListenAddr)
280 + sniListener, err = listenConfig.Listen(serverCtx, "tcp", cfg.SNIListenAddr)
281 if err != nil {
282 return fmt.Errorf("listen sni: %w", err)
283 }
@@ -236,8 +287,8 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
287 if err != nil {
288 return err
289 }
239 - if s.cfg.PProfEnabled {
240 - pprofListener, err = listenConfig.Listen(serverCtx, "tcp", s.cfg.PProfListenAddr)
290 + if cfg.PProfEnabled {
291 + pprofListener, err = listenConfig.Listen(serverCtx, "tcp", cfg.PProfListenAddr)
292 if err != nil {
293 return fmt.Errorf("listen pprof: %w", err)
294 }
@@ -259,7 +310,7 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
310 return err
311 }
312 }
262 - if s.cfg.UDPEnabled {
313 + if cfg.UDPEnabled {
314 quicBackhaul, err = s.newQUICBackhaulListener(apiTLS)
315 if err != nil {
316 log.Warn().Err(err).Msg("quic backhaul listener disabled")
@@ -292,7 +343,7 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
343 group.Go(s.runQUICBackhaulListener)
344 }
345 group.Go(func() error { return s.runRegistryJanitor(groupCtx, 5*time.Second) })
295 - if s.cfg.DiscoveryEnabled {
346 + if cfg.DiscoveryEnabled {
347 group.Go(func() error { return s.runRelayDiscoveryLoop(groupCtx) })
348 }
349 s.acmeManager.Start(serverCtx)
@@ -307,14 +358,14 @@ func (s *Server) Start(ctx context.Context, apiMux *http.ServeMux) error {
358 Str("api_addr", utils.HostPortOrLoopback(s.apiListener.Addr().String())).
359 Str("sni_addr", s.sniListener.Addr().String()).
360 Str("root_host", s.identity.Name).
310 - Str("acme_dns_provider", s.cfg.ACME.DNSProvider).
311 - Int("min_port", s.cfg.MinPort).
312 - Int("max_port", s.cfg.MaxPort).
313 - Bool("discovery_enabled", s.cfg.DiscoveryEnabled).
361 + Str("acme_dns_provider", cfg.ACME.DNSProvider).
362 + Int("min_port", cfg.MinPort).
363 + Int("max_port", cfg.MaxPort).
364 + Bool("discovery_enabled", cfg.DiscoveryEnabled).
365 Bool("wireguard_enabled", s.overlay != nil).
366 Bool("multihop_enabled", s.overlay != nil).
367 Bool("udp_enabled", s.quicBackhaul != nil).
317 - Bool("tcp_enabled", s.cfg.TCPEnabled).
368 + Bool("tcp_enabled", s.supportsTCP()).
369 Bool("api_ech_enabled", len(apiTLS.EncryptedClientHelloKeys) > 0).
370 Bool("pprof_enabled", s.pprofServer != nil)
371 if s.pprofListener != nil {
@@ -350,7 +401,7 @@ func (s *Server) PortalURL() string {
401 if s == nil {
402 return ""
403 }
353 - return s.cfg.PortalURL
404 + return s.config().PortalURL
405 }
406
407 func (s *Server) PublicLeases() []types.Lease {
@@ -420,7 +471,8 @@ func (s *Server) Shutdown(ctx context.Context) error {
471 }
472
473 func (s *Server) prepareAPITLS(ctx context.Context) (keyless.TLSMaterialConfig, *acme.Manager, error) {
423 - acmeCfg := s.cfg.ACME
474 + cfg := s.config()
475 + acmeCfg := cfg.ACME
476 if baseDomain := utils.NormalizeHostname(acmeCfg.BaseDomain); baseDomain != "" && baseDomain != s.identity.Name {
477 return keyless.TLSMaterialConfig{}, nil, fmt.Errorf("acme base domain %q does not match portal root host %q", acmeCfg.BaseDomain, s.identity.Name)
478 }
@@ -461,7 +513,7 @@ func (s *Server) prepareAPITLS(ctx context.Context) (keyless.TLSMaterialConfig,
513 }
514 if len(echKeys) > 0 {
515 apiTLS.EncryptedClientHelloKeys = echKeys
464 - if err := manager.SyncECHConfig(ctx, s.identity.Name, echConfigList, s.cfg.SNIPort); err != nil {
516 + if err := manager.SyncECHConfig(ctx, s.identity.Name, echConfigList, cfg.SNIPort); err != nil {
517 log.Warn().
518 Err(err).
519 Str("hostname", s.identity.Name).
@@ -627,7 +679,7 @@ func (s *Server) newQUICBackhaulListener(apiTLS keyless.TLSMaterialConfig) (*qui
679 if err != nil {
680 return nil, fmt.Errorf("parse quic backhaul tls keypair: %w", err)
681 }
630 - return transport.ListenQUICBackhaul(s.cfg.SNIListenAddr, tlsCert)
682 + return transport.ListenQUICBackhaul(s.config().SNIListenAddr, tlsCert)
683 }
684
685 func (s *Server) runQUICBackhaulListener() error {
@@ -686,17 +738,18 @@ func (s *Server) handleQUICBackhaulConn(conn *quic.Conn) {
738 }
739
740 func (s *Server) startOverlay() (*overlay.Overlay, error) {
741 + cfg := s.config()
742 peerMux := http.NewServeMux()
743 peerMux.HandleFunc(types.PathRoot, s.handleRoot)
744 peerMux.HandleFunc(types.PathHealthz, s.handleHealthz)
692 - if s.cfg.DiscoveryEnabled {
745 + if cfg.DiscoveryEnabled {
746 peerMux.HandleFunc(types.PathDiscovery, s.handleRelayDiscovery)
747 }
748
749 ov, err := overlay.NewOverlay(overlay.Config{
750 PrivateKey: s.identity.WireGuardPrivateKey,
751 PublicKey: s.identity.WireGuardPublicKey,
699 - ListenPort: s.cfg.WireGuardPort,
752 + ListenPort: cfg.WireGuardPort,
753 }, peerMux, nil)
754 if err != nil {
755 return nil, fmt.Errorf("start wireguard overlay: %w", err)
@@ -770,6 +823,7 @@ func (s *Server) newSelfDescriptor(now time.Time) (types.RelayDescriptor, error)
823 } else {
824 now = now.UTC()
825 }
826 + cfg := s.config()
827
828 var wireGuardPublicKey string
829 var wireGuardPort int
@@ -786,12 +840,12 @@ func (s *Server) newSelfDescriptor(now time.Time) (types.RelayDescriptor, error)
840 Version: types.DiscoveryVersion,
841 IssuedAt: now,
842 ExpiresAt: now.Add(discovery.DiscoveryDescriptorTTL),
789 - APIHTTPSAddr: s.cfg.PortalURL,
843 + APIHTTPSAddr: cfg.PortalURL,
844 WireGuardPublicKey: wireGuardPublicKey,
845 WireGuardPort: wireGuardPort,
846 SupportsOverlay: supportsOverlay,
793 - SupportsUDP: s.cfg.UDPEnabled && s.quicBackhaul != nil,
794 - SupportsTCP: s.cfg.TCPEnabled,
847 + SupportsUDP: s.supportsUDP(),
848 + SupportsTCP: s.supportsTCP(),
849 ActiveConnections: s.proxy.activeConnectionCount(),
850 TCPBPS: s.proxy.currentTCPBPS(now),
851 }, s.authority)
portal/server_test.go
+6 -6
@@ -292,8 +292,8 @@ func TestRegisterLeaseIncludesSNIPortForPublicIngress(t *testing.T) {
292 record.Close()
293 })
294
295 - if resp.SNIPort != server.cfg.SNIPort {
296 - t.Fatalf("RegisterResponse.SNIPort = %d, want %d", resp.SNIPort, server.cfg.SNIPort)
295 + if resp.SNIPort != server.config().SNIPort {
296 + t.Fatalf("RegisterResponse.SNIPort = %d, want %d", resp.SNIPort, server.config().SNIPort)
297 }
298 }
299
@@ -411,7 +411,7 @@ func TestRegisterLeaseBuildsUDPEnabledRuntime(t *testing.T) {
411 if err != nil {
412 t.Fatalf("NewServer() error = %v", err)
413 }
414 - server.registry.policy.SetUDPPolicy(true, 0)
414 + server.SetUDPPolicy(true, 0)
415
416 record, resp, err := server.registry.Register(types.RegisterChallengeRequest{
417 Identity: types.Identity{
@@ -436,8 +436,8 @@ func TestRegisterLeaseBuildsUDPEnabledRuntime(t *testing.T) {
436 if got := record.datagram.UDPPort(); got < 40000 || got > 40009 {
437 t.Fatalf("UDPPort() = %d, want port within %d-%d", got, 40000, 40009)
438 }
439 - if resp.SNIPort != server.cfg.SNIPort {
440 - t.Fatalf("RegisterResponse.SNIPort = %d, want %d", resp.SNIPort, server.cfg.SNIPort)
439 + if resp.SNIPort != server.config().SNIPort {
440 + t.Fatalf("RegisterResponse.SNIPort = %d, want %d", resp.SNIPort, server.config().SNIPort)
441 }
442 if resp.UDPAddr == "" {
443 t.Fatal("RegisterResponse.UDPAddr = empty, want public udp address")
@@ -476,7 +476,7 @@ func TestServerStartHidesDiscoveryRoutesWhenDisabled(t *testing.T) {
476 if resp.StatusCode != http.StatusNotFound {
477 t.Fatalf("GET relay discovery status = %d, want %d", resp.StatusCode, http.StatusNotFound)
478 }
479 - if server.cfg.DiscoveryEnabled {
479 + if server.config().DiscoveryEnabled {
480 t.Fatal("cfg.DiscoveryEnabled = true, want false without configured discovery service")
481 }
482 }
sdk/expose.go
+58 -71
@@ -27,8 +27,7 @@ type Exposure struct {
27 cancel context.CancelFunc
28 done <-chan struct{}
29
30 - cfgMu sync.RWMutex
31 - cfg ExposeConfig
30 + cfg *utils.Snapshot[ExposeConfig]
31
32 accepted chan net.Conn
33 datagrams chan types.DatagramFrame
@@ -59,32 +58,14 @@ type ExposeConfig struct {
58 Metadata types.LeaseMetadata
59 }
60
62 -func (cfg ExposeConfig) clone() ExposeConfig {
63 - cfg.RelayURLs = append([]string(nil), cfg.RelayURLs...)
61 +func (cfg ExposeConfig) snapshot() ExposeConfig {
62 + cfg.RelayURLs = utils.CloneSlice(cfg.RelayURLs)
63 cfg.Identity = cfg.Identity.Copy()
65 - cfg.MultiHop = append([]string(nil), cfg.MultiHop...)
64 + cfg.MultiHop = utils.CloneSlice(cfg.MultiHop)
65 cfg.Metadata = cfg.Metadata.Copy()
66 return cfg
67 }
68
70 -func (e *Exposure) config() ExposeConfig {
71 - if e == nil {
72 - return ExposeConfig{}
73 - }
74 - e.cfgMu.RLock()
75 - defer e.cfgMu.RUnlock()
76 - return e.cfg.clone()
77 -}
78 -
79 -func (e *Exposure) metadata() types.LeaseMetadata {
80 - if e == nil {
81 - return types.LeaseMetadata{}
82 - }
83 - e.cfgMu.RLock()
84 - defer e.cfgMu.RUnlock()
85 - return e.cfg.Metadata.Copy()
86 -}
87 -
69 // Expose creates relay listeners for the selected relay pool and exposes a
70 // dynamic listener hub for accepting traffic from all of them.
71 func Expose(ctx context.Context, cfg ExposeConfig) (*Exposure, error) {
@@ -160,7 +141,7 @@ func Expose(ctx context.Context, cfg ExposeConfig) (*Exposure, error) {
141 return nil, fmt.Errorf("invalid --udp-addr value %q: %w", cfg.UDPAddr, err)
142 }
143 }
163 - runtimeCfg := cfg.clone()
144 + runtimeCfg := cfg.snapshot()
145 runtimeCfg.RelayURLs = append([]string(nil), explicitRelayURLs...)
146 runtimeCfg.Identity = listenerIdentity.Copy()
147 runtimeCfg.TargetAddr = targetAddr
@@ -172,7 +153,7 @@ func Expose(ctx context.Context, cfg ExposeConfig) (*Exposure, error) {
153 exposure := &Exposure{
154 cancel: cancel,
155 done: exposureCtx.Done(),
175 - cfg: runtimeCfg.clone(),
156 + cfg: utils.NewSnapshot(runtimeCfg, ExposeConfig.snapshot),
157 accepted: make(chan net.Conn, max(initialRouteCapacity(listenerRelayURLs, cfg.MultiHopDepth)*defaultReadyTarget*2, 1)),
158 datagrams: make(chan types.DatagramFrame, max(initialRouteCapacity(listenerRelayURLs, cfg.MultiHopDepth)*32, 1)),
159 relaySet: discovery.NewRelaySet(relaySetURLs),
@@ -220,11 +201,11 @@ func (e *Exposure) AddRelay(relayURL string) error {
201 return errors.New("exposure relay set is not initialized")
202 }
203
223 - e.cfgMu.Lock()
224 - if !slices.Contains(e.cfg.RelayURLs, relayURL) {
225 - e.cfg.RelayURLs = append(append([]string(nil), e.cfg.RelayURLs...), relayURL)
226 - }
227 - e.cfgMu.Unlock()
204 + e.cfg.UpdateCopy(func(cfg *ExposeConfig) {
205 + if !slices.Contains(cfg.RelayURLs, relayURL) {
206 + cfg.RelayURLs = append(cfg.RelayURLs, relayURL)
207 + }
208 + })
209
210 e.relaySet.AllowRelayURL(relayURL)
211 e.relaySet.AddBootstrapRelayURL(relayURL)
@@ -245,19 +226,21 @@ func (e *Exposure) RemoveRelay(relayURL string) error {
226 return errors.New("exposure relay set is not initialized")
227 }
228
248 - e.cfgMu.Lock()
249 - if slices.Contains(e.cfg.MultiHop, relayURL) {
250 - e.cfgMu.Unlock()
251 - return errors.New("relay is part of the multi-hop route; clear multi-hop first")
252 - }
253 - nextRelays := make([]string, 0, len(e.cfg.RelayURLs))
254 - for _, existing := range e.cfg.RelayURLs {
255 - if existing != relayURL {
256 - nextRelays = append(nextRelays, existing)
229 + if _, ok := e.cfg.UpdateIf(func(cfg ExposeConfig) (ExposeConfig, bool) {
230 + if slices.Contains(cfg.MultiHop, relayURL) {
231 + return cfg, false
232 }
233 + nextRelays := cfg.RelayURLs[:0]
234 + for _, existing := range cfg.RelayURLs {
235 + if existing != relayURL {
236 + nextRelays = append(nextRelays, existing)
237 + }
238 + }
239 + cfg.RelayURLs = nextRelays
240 + return cfg, true
241 + }); !ok {
242 + return errors.New("relay is part of the multi-hop route; clear multi-hop first")
243 }
259 - e.cfg.RelayURLs = nextRelays
260 - e.cfgMu.Unlock()
244
245 e.relaySet.DeactivateRelayURL(relayURL)
246 e.relaySet.RemoveBootstrapRelayURL(relayURL)
@@ -279,7 +262,7 @@ func (e *Exposure) SetMultiHop(relayURLs []string) error {
262 if len(multiHop) == 1 {
263 return errors.New("multi-hop requires at least entry and exit relay urls")
264 }
282 - cfg := e.config()
265 + cfg := e.Config()
266 if len(multiHop) > 0 && (cfg.UDPEnabled || cfg.TCPEnabled) {
267 return errors.New("multi-hop currently supports only the default SNI TLS stream transport")
268 }
@@ -295,10 +278,10 @@ func (e *Exposure) SetMultiHop(relayURLs []string) error {
278 e.relaySet.AddBootstrapRelayURL(relayURL)
279 }
280
298 - e.cfgMu.Lock()
299 - e.cfg.MultiHop = append([]string(nil), multiHop...)
300 - e.cfg.MultiHopDepth = 0
301 - e.cfgMu.Unlock()
281 + e.cfg.UpdateCopy(func(cfg *ExposeConfig) {
282 + cfg.MultiHop = append([]string(nil), multiHop...)
283 + cfg.MultiHopDepth = 0
284 + })
285 return e.reconcileRelayListeners(false)
286 }
287
@@ -307,9 +290,9 @@ func (e *Exposure) UpdateMetadata(metadata types.LeaseMetadata) error {
290 return net.ErrClosed
291 }
292
310 - e.cfgMu.Lock()
311 - e.cfg.Metadata = metadata.Copy()
312 - e.cfgMu.Unlock()
293 + e.cfg.UpdateCopy(func(cfg *ExposeConfig) {
294 + cfg.Metadata = metadata.Copy()
295 + })
296 return nil
297 }
298
@@ -321,10 +304,13 @@ func (e *Exposure) UpdateMaxActiveRelays(maxActiveRelays int) error {
304 return net.ErrClosed
305 }
306
324 - e.cfgMu.Lock()
325 - changed := e.cfg.MaxActiveRelays != maxActiveRelays
326 - e.cfg.MaxActiveRelays = maxActiveRelays
327 - e.cfgMu.Unlock()
307 + _, changed := e.cfg.UpdateIf(func(cfg ExposeConfig) (ExposeConfig, bool) {
308 + if cfg.MaxActiveRelays == maxActiveRelays {
309 + return cfg, false
310 + }
311 + cfg.MaxActiveRelays = maxActiveRelays
312 + return cfg, true
313 + })
314 if !changed {
315 return nil
316 }
@@ -359,7 +345,7 @@ func (e *Exposure) closed() bool {
345 }
346
347 func (e *Exposure) Addr() net.Addr {
362 - identity := e.config().Identity
348 + identity := e.Config().Identity
349 if identity.Address == "" {
350 return exposureAddr("portal:exposure")
351 }
@@ -372,11 +358,14 @@ func (a exposureAddr) Network() string { return "portal" }
358 func (a exposureAddr) String() string { return string(a) }
359
360 func (e *Exposure) Config() ExposeConfig {
375 - return e.config()
361 + if e == nil || e.cfg == nil {
362 + return ExposeConfig{}
363 + }
364 + return e.cfg.Load()
365 }
366
367 func (e *Exposure) Snapshot() types.AgentTunnelStatus {
379 - cfg := e.config()
368 + cfg := e.Config()
369 e.mu.RLock()
370 listeners := make([]*listener, 0, len(e.relayListeners))
371 for _, listener := range e.relayListeners {
@@ -458,7 +447,7 @@ func (e *Exposure) Snapshot() types.AgentTunnelStatus {
447 }
448
449 func (e *Exposure) AcceptDatagram() (types.DatagramFrame, error) {
461 - if !e.config().UDPEnabled {
450 + if !e.Config().UDPEnabled {
451 return types.DatagramFrame{}, net.ErrClosed
452 }
453
@@ -471,7 +460,7 @@ func (e *Exposure) AcceptDatagram() (types.DatagramFrame, error) {
460 }
461
462 func (e *Exposure) SendDatagram(frame types.DatagramFrame) error {
474 - if !e.config().UDPEnabled {
463 + if !e.Config().UDPEnabled {
464 return net.ErrClosed
465 }
466
@@ -485,7 +474,7 @@ func (e *Exposure) SendDatagram(frame types.DatagramFrame) error {
474 }
475
476 func (e *Exposure) WaitDatagramReady(ctx context.Context) ([]string, error) {
488 - if !e.config().UDPEnabled {
477 + if !e.Config().UDPEnabled {
478 return nil, errors.New("exposure does not have udp enabled")
479 }
480
@@ -686,7 +675,7 @@ func (e *Exposure) reconcileRelayListeners(failOnError bool) error {
675 var multiHop []string
676 var listenerRelayURLs []string
677
689 - cfg := e.config()
678 + cfg := e.Config()
679 e.mu.Lock()
680 multiHop = append([]string(nil), cfg.MultiHop...)
681 if len(multiHop) > 0 {
@@ -761,7 +750,7 @@ func (e *Exposure) reconcileRelayListeners(failOnError bool) error {
750 TCPEnabled: cfg.TCPEnabled,
751 BanMITM: cfg.BanMITM,
752 Metadata: func() types.LeaseMetadata {
764 - return e.metadata()
753 + return e.Config().Metadata
754 },
755 MultiHop: listenerMultiHop,
756 RetryCount: retryCount,
@@ -856,17 +845,15 @@ func (e *Exposure) runListenerAcceptLoop(listener *listener) {
845 }
846
847 removedExplicit := false
859 - e.cfgMu.Lock()
860 - next := e.cfg.RelayURLs[:0]
861 - for _, existing := range e.cfg.RelayURLs {
862 - if existing == relayURL {
863 - removedExplicit = true
864 - continue
865 - }
866 - next = append(next, existing)
848 + if e.cfg != nil {
849 + _, removedExplicit = e.cfg.UpdateIf(func(cfg ExposeConfig) (ExposeConfig, bool) {
850 + if !slices.Contains(cfg.RelayURLs, relayURL) {
851 + return cfg, false
852 + }
853 + cfg.RelayURLs = utils.RemoveRelayURL(cfg.RelayURLs, relayURL)
854 + return cfg, true
855 + })
856 }
868 - e.cfg.RelayURLs = next
869 - e.cfgMu.Unlock()
857
858 if removedExplicit && e.relaySet != nil {
859 e.relaySet.DeactivateRelayURL(relayURL)
sdk/expose_test.go
+13 -12
@@ -6,6 +6,7 @@ import (
6
7 "github.com/gosuda/portal-tunnel/v2/portal/discovery"
8 "github.com/gosuda/portal-tunnel/v2/types"
9 + "github.com/gosuda/portal-tunnel/v2/utils"
10 )
11
12 func mustRelaySet(t *testing.T, relayURLs ...string) *discovery.RelaySet {
@@ -15,7 +16,7 @@ func mustRelaySet(t *testing.T, relayURLs ...string) *discovery.RelaySet {
16
17 func TestExposureConfigSnapshotsDoNotShareMutableState(t *testing.T) {
18 exposure := &Exposure{
18 - cfg: ExposeConfig{
19 + cfg: utils.NewSnapshot(ExposeConfig{
20 RelayURLs: []string{"https://relay-a.example"},
21 Identity: types.Identity{
22 Name: "svc",
@@ -24,7 +25,7 @@ func TestExposureConfigSnapshotsDoNotShareMutableState(t *testing.T) {
25 Metadata: types.LeaseMetadata{
26 Tags: []string{"initial"},
27 },
27 - },
28 + }, ExposeConfig.snapshot),
29 }
30
31 snapshot := exposure.Config()
@@ -39,15 +40,15 @@ func TestExposureConfigSnapshotsDoNotShareMutableState(t *testing.T) {
40 t.Fatalf("Metadata.Tags[0] = %q, want original tag", got)
41 }
42
42 - exposure.cfgMu.Lock()
43 - exposure.cfg.MaxActiveRelays = 2
44 - exposure.cfg.Metadata = types.LeaseMetadata{Tags: []string{"updated"}}
45 - exposure.cfgMu.Unlock()
43 + exposure.cfg.UpdateCopy(func(cfg *ExposeConfig) {
44 + cfg.MaxActiveRelays = 2
45 + cfg.Metadata = types.LeaseMetadata{Tags: []string{"updated"}}
46 + })
47
47 - metadata := exposure.metadata()
48 + metadata := exposure.Config().Metadata
49 metadata.Tags[0] = "mutated"
49 - if got := exposure.metadata().Tags[0]; got != "updated" {
50 - t.Fatalf("MetadataSnapshot().Tags[0] = %q, want updated", got)
50 + if got := exposure.Config().Metadata.Tags[0]; got != "updated" {
51 + t.Fatalf("Metadata.Tags[0] = %q, want updated", got)
52 }
53 if got := exposure.Config().MaxActiveRelays; got != 2 {
54 t.Fatalf("MaxActiveRelays = %d, want 2", got)
@@ -70,7 +71,7 @@ func TestExposureReconcileRemovesBannedRelayFromActiveSet(t *testing.T) {
71 }
72
73 exposure := &Exposure{
73 - cfg: ExposeConfig{RelayURLs: []string{relayA, relayB}},
74 + cfg: utils.NewSnapshot(ExposeConfig{RelayURLs: []string{relayA, relayB}}, ExposeConfig.snapshot),
75 relaySet: mustRelaySet(t, relayA, relayB),
76 relayListeners: make(map[string]*listener, 2),
77 }
@@ -126,7 +127,7 @@ func TestExposureReconcileRemovesStaleListener(t *testing.T) {
127
128 relayAClosed := make(chan struct{})
129 exposure := &Exposure{
129 - cfg: ExposeConfig{RelayURLs: []string{relayA, relayB}},
130 + cfg: utils.NewSnapshot(ExposeConfig{RelayURLs: []string{relayA, relayB}}, ExposeConfig.snapshot),
131 relaySet: mustRelaySet(t, relayA, relayB),
132 relayListeners: make(map[string]*listener, 2),
133 }
@@ -178,7 +179,7 @@ func TestExposureRemoveRelayStopsRunningListener(t *testing.T) {
179
180 relayAClosed := make(chan struct{})
181 exposure := &Exposure{
181 - cfg: ExposeConfig{RelayURLs: []string{relayA}},
182 + cfg: utils.NewSnapshot(ExposeConfig{RelayURLs: []string{relayA}}, ExposeConfig.snapshot),
183 relaySet: mustRelaySet(t, relayA),
184 relayListeners: make(map[string]*listener, 1),
185 }
utils/snapshot.go new
+130
@@ -0,0 +1,130 @@
1 +package utils
2 +
3 +import (
4 + "maps"
5 + "slices"
6 + "sync/atomic"
7 +)
8 +
9 +// Snapshot stores an immutable value snapshot for lock-free reads.
10 +// Use a snapshot function when T contains mutable maps, slices, or pointers.
11 +// Do not copy a Snapshot after first use.
12 +type Snapshot[T any] struct {
13 + value atomic.Pointer[T]
14 + snapshot func(T) T
15 +}
16 +
17 +func NewSnapshot[T any](initial T, snapshot ...func(T) T) *Snapshot[T] {
18 + s := &Snapshot[T]{}
19 + if len(snapshot) > 0 {
20 + s.snapshot = snapshot[0]
21 + }
22 + s.Store(initial)
23 + return s
24 +}
25 +
26 +func (s *Snapshot[T]) Load() T {
27 + if s == nil {
28 + var zero T
29 + return zero
30 + }
31 + value := s.value.Load()
32 + if value == nil {
33 + var zero T
34 + return zero
35 + }
36 + return s.snapshotValue(*value)
37 +}
38 +
39 +func (s *Snapshot[T]) Store(value T) {
40 + if s == nil {
41 + return
42 + }
43 + value = s.snapshotValue(value)
44 + s.value.Store(&value)
45 +}
46 +
47 +func (s *Snapshot[T]) snapshotValue(value T) T {
48 + if s != nil && s.snapshot != nil {
49 + return s.snapshot(value)
50 + }
51 + return value
52 +}
53 +
54 +// Update applies a copy-on-write update and stores the resulting snapshot.
55 +// The update function may be called more than once under contention, so it
56 +// must not perform side effects.
57 +func (s *Snapshot[T]) Update(update func(T) T) T {
58 + if s == nil {
59 + var zero T
60 + return zero
61 + }
62 + for {
63 + currentPtr := s.value.Load()
64 + var current T
65 + if currentPtr != nil {
66 + current = s.snapshotValue(*currentPtr)
67 + }
68 +
69 + next := update(current)
70 + next = s.snapshotValue(next)
71 + if s.value.CompareAndSwap(currentPtr, &next) {
72 + return s.snapshotValue(next)
73 + }
74 + }
75 +}
76 +
77 +// UpdateIf stores the returned snapshot only when update returns true.
78 +// The update function may be called more than once under contention, so it
79 +// must not perform side effects.
80 +func (s *Snapshot[T]) UpdateIf(update func(T) (T, bool)) (T, bool) {
81 + if s == nil {
82 + var zero T
83 + return zero, false
84 + }
85 + for {
86 + currentPtr := s.value.Load()
87 + var current T
88 + if currentPtr != nil {
89 + current = s.snapshotValue(*currentPtr)
90 + }
91 +
92 + next, ok := update(current)
93 + if !ok {
94 + return current, false
95 + }
96 + next = s.snapshotValue(next)
97 + if s.value.CompareAndSwap(currentPtr, &next) {
98 + return s.snapshotValue(next), true
99 + }
100 + }
101 +}
102 +
103 +// UpdateCopy copies the current snapshot, applies a local mutation to the copy,
104 +// and stores the copy. The update function may be called more than once under
105 +// contention, so it must not perform side effects.
106 +func (s *Snapshot[T]) UpdateCopy(update func(*T)) T {
107 + return s.Update(func(current T) T {
108 + next := current
109 + if update != nil {
110 + update(&next)
111 + }
112 + return next
113 + })
114 +}
115 +
116 +func CloneSlice[T any](values []T) []T {
117 + return slices.Clone(values)
118 +}
119 +
120 +func CloneMap[K comparable, V any](values map[K]V) map[K]V {
121 + return maps.Clone(values)
122 +}
123 +
124 +func ClonePtr[T any](value *T) *T {
125 + if value == nil {
126 + return nil
127 + }
128 + next := *value
129 + return &next
130 +}