Compare commits

...

13 Commits

Author SHA1 Message Date
Theodor S. Midtlien
8a46353ae2 Add tests 2026-07-21 15:40:39 +02:00
Zoltán Papp
d663da8f82 Merge branch 'main' into refactor/relay-foreign-cache
# Conflicts:
#	shared/relay/client/manager.go
2026-07-21 11:27:09 +02:00
Maycon Santos
82fdfa84b8 [proxy] match Bedrock provider models against the normalized request model (#6773)
## Describe your changes

Native AWS Bedrock requests carry the model in the URL path as a
cross-region inference-profile id (e.g.
`us.anthropic.claude-haiku-4-5`). The request parser normalizes that to
the catalog key (`anthropic.claude-haiku-4-5`) before the router runs,
but the router matched it against the operator's registered provider
models with exact string equality. So a Bedrock provider registered with
the id Bedrock actually uses (`us.anthropic…`) never matched a
normalized request → the request denied with
`llm_policy.model_not_routable` ("no provider configured for model …").
Only a provider registered with the already-stripped catalog id worked,
which is not how Bedrock ids appear.

Fix: introduce a single shared `llm.NormalizeBedrockModel` (the same
ARN/region-prefix/version-suffix stripping the parser already does) and,
in the router's `routeClaimsModel`, normalize a **Bedrock** route's
candidate models before comparing. Now a Bedrock provider registered
with either the raw inference-profile id or the normalized catalog id
matches the request. Non-Bedrock routes keep exact matching.

Surfaced by the new native-Bedrock e2e (`WireBedrock`,
`/model/{id}/invoke`); the old e2e used the Anthropic body shape, which
never normalized either side and so hid this.

The request parser keeps its own identical normalizer for now;
de-duplicating it onto `llm.NormalizeBedrockModel` is a trivial
follow-up.

## Issue ticket number and link

N/A — follow-up to the Agent Network Bedrock support / model-allowlist
work.

## Stack

<!-- branch-stack -->

### Checklist
- [x] Is it a bug fix
- [ ] Is a typo/documentation fix
- [ ] Is a feature enhancement
- [ ] It is a refactor
- [x] Created tests that fail without the change (if possible)
- [x] This change does **not** modify the public API, gRPC protocols,
functionality behavior, CLI / service flags, or introduce a new feature
— **OR** I have discussed it with the NetBird team beforehand (link the
issue / Slack thread in the description). See
[CONTRIBUTING.md](https://github.com/netbirdio/netbird/blob/main/CONTRIBUTING.md#discuss-changes-with-the-netbird-team-first).

> By submitting this pull request, you confirm that you have read and
agree to the terms of the [Contributor License
Agreement](https://github.com/netbirdio/netbird/blob/main/CONTRIBUTOR_LICENSE_AGREEMENT.md).

## Documentation
Select exactly one:

- [ ] I added/updated documentation for this change
- [x] Documentation is **not needed** for this change (explain why)

Internal routing correctness fix; no user-facing surface change.

### Docs PR URL (required if "docs added" is checked)
Paste the PR link from https://github.com/netbirdio/docs here:

https://github.com/netbirdio/docs/pull/__

## Tests

- `proxy/internal/llm`: `NormalizeBedrockModel` unit cases (region
prefixes, version suffixes, ARN).
- `proxy/internal/middleware/builtin/llm_router`: `routeClaimsModel`
matches a Bedrock route registered with the raw `us.anthropic…` id
against a normalized request model; non-Bedrock routes still match
exactly.

Note: full through-tunnel e2e verification of this (the native-Bedrock
`TestProvidersMatrix/bedrock`) also needs the DNS lazy-connection
warm-up (separate PR) to get the client past the proxy-peer gate; they
converge once both land.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **Bug Fixes**
* Improved Amazon Bedrock model matching across ARN formats, regional
prefixes, and version or throughput suffixes.
* Bedrock routes now correctly match equivalent model identifiers even
when requests and route configurations use different formats.
  * Non-Bedrock model matching remains exact.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->
2026-07-21 10:10:12 +02:00
Zoltan Papp
ca80e49aa0 [client] Refresh WireGuard stats in mobile debug bundles (#6814)
The iOS and Android DebugBundle paths built GeneratorDependencies
without setting RefreshStatus, so the bundle's status.txt read the
cached peer state instead of live WireGuard interface stats. When the
periodic health probe had not run yet, connected relayed peers showed
"handshake: -" and "0 B/0 B" even though the interface was passing
traffic.

Wire RefreshStatus to RunHealthProbes on both platforms, matching the
desktop daemon path in client/server/debug.go. The engine reference is
already available in the cc.Engine() block used for client metrics.

## Describe your changes

## Issue ticket number and link

## Stack

<!-- branch-stack -->

### Checklist
- [x] Is it a bug fix
- [ ] Is a typo/documentation fix
- [ ] Is a feature enhancement
- [ ] It is a refactor
- [ ] Created tests that fail without the change (if possible)
- [ ] This change does **not** modify the public API, gRPC protocols,
functionality behavior, CLI / service flags, or introduce a new feature
— **OR** I have discussed it with the NetBird team beforehand (link the
issue / Slack thread in the description). See
[CONTRIBUTING.md](https://github.com/netbirdio/netbird/blob/main/CONTRIBUTING.md#discuss-changes-with-the-netbird-team-first).

> By submitting this pull request, you confirm that you have read and
agree to the terms of the [Contributor License
Agreement](https://github.com/netbirdio/netbird/blob/main/CONTRIBUTOR_LICENSE_AGREEMENT.md).

## Documentation
Select exactly one:

- [ ] I added/updated documentation for this change
- [x] Documentation is **not needed** for this change (explain why)

### Docs PR URL (required if "docs added" is checked)
Paste the PR link from https://github.com/netbirdio/docs here:

https://github.com/netbirdio/docs/pull/__

<!-- codesmith:footer -->
---
<a
href="https://app.blacksmith.sh/netbirdio/codesmith/netbird/pr/6814"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-dark-v2.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-light-v2.svg"><img
alt="View with Codesmith"
src="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-dark-v2.svg"></picture></a>
<a
href="https://backend.blacksmith.sh/track/enable-autofix?expires=1786881440&installation_id=146802194&pr_number=6814&repository=netbirdio%2Fnetbird&return_to=https%3A%2F%2Fgithub.com%2Fnetbirdio%2Fnetbird%2Fpull%2F6814&signature=c07cbcef1028a366312f7743e6479f2fddafd568aac498ab95828a372ac6843c"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-light.svg"><img
alt="Autofix with Codesmith"
src="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-dark.svg"></picture></a>
<sup>Need help on this PR? Tag <code>/codesmith</code> with what you
need. Autofix is disabled.</sup>

<!-- codesmith:autofix:disabled -->
<!-- /codesmith:footer -->

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **Bug Fixes**
* Improved debug bundle generation on Android and iOS by refreshing
connection health status before collecting diagnostic information.
* Ensured debug bundles include more current health-related data for
troubleshooting.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->
2026-07-21 09:25:33 +02:00
Zoltan Papp
51f17bf919 [client] Update wails to v3.0.0-alpha2.117 (#6837)
## Describe your changes

Update wails to v3.0.0-alpha2.117

## Issue ticket number and link

## Stack

<!-- branch-stack -->

### Checklist
- [ ] Is it a bug fix
- [ ] Is a typo/documentation fix
- [x] Is a feature enhancement
- [ ] It is a refactor
- [ ] Created tests that fail without the change (if possible)
- [ ] This change does **not** modify the public API, gRPC protocols,
functionality behavior, CLI / service flags, or introduce a new feature
— **OR** I have discussed it with the NetBird team beforehand (link the
issue / Slack thread in the description). See
[CONTRIBUTING.md](https://github.com/netbirdio/netbird/blob/main/CONTRIBUTING.md#discuss-changes-with-the-netbird-team-first).

> By submitting this pull request, you confirm that you have read and
agree to the terms of the [Contributor License
Agreement](https://github.com/netbirdio/netbird/blob/main/CONTRIBUTOR_LICENSE_AGREEMENT.md).

## Documentation
Select exactly one:

- [ ] I added/updated documentation for this change
- [x] Documentation is **not needed** for this change (explain why)

### Docs PR URL (required if "docs added" is checked)
Paste the PR link from https://github.com/netbirdio/docs here:

https://github.com/netbirdio/docs/pull/__

<!-- codesmith:footer -->
---
<a
href="https://app.blacksmith.sh/netbirdio/codesmith/netbird/pr/6837"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-dark-v2.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-light-v2.svg"><img
alt="View with Codesmith"
src="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-dark-v2.svg"></picture></a>
<a
href="https://backend.blacksmith.sh/track/enable-autofix?expires=1787156447&installation_model_id=427504&pr_number=6837&repository=netbirdio%2Fnetbird&return_to=https%3A%2F%2Fgithub.com%2Fnetbirdio%2Fnetbird%2Fpull%2F6837&signature=600f7cb9ecda9ee75b7fe841ed81149507d00dbf6323c61aa92ac2acdaea2487"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-light.svg"><img
alt="Autofix with Codesmith"
src="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-dark.svg"></picture></a>
<sup>Need help on this PR? Tag <code>/codesmith</code> with what you
need. Autofix is disabled.</sup>

<!-- codesmith:autofix:disabled -->
<!-- /codesmith:footer -->

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **Chores**
  * Updated the application framework dependency to a newer release.
  * Removed an obsolete supporting dependency requirement.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->
2026-07-20 21:12:22 +02:00
Zoltan Papp
cf101c44b4 [relay] Count winning attempt before draining losers
The success path in handleResult left settled at 0, so drainLoser
waited on a result that was already consumed and blocked forever,
leaking a goroutine and the results channel on every successful race.
Increment settled before stop() so drainLoser only waits for actual
started losers.
2026-07-02 12:22:01 +02:00
Zoltan Papp
79b51a79e4 [relay] Release home relay lock before the connection race
OpenConn held relayClientMu for the whole call, including the network-bound
FallbackOpener.Run, which blocked onServerDisconnected/storeClient from taking
the write lock and delayed reconnect by up to the race timeout. Snapshot the
home relay client under the read lock, release it, and run the foreign check
and the race on the snapshot. isForeignServer now takes the client so it uses
the snapshot instead of re-reading the field after the lock is dropped.
2026-07-01 19:22:41 +02:00
Zoltan Papp
a737504ec9 Fix timeout 2026-07-01 17:11:23 +02:00
Zoltan Papp
3b1beb3497 [relay] Return raceOutcome struct from handleResult to avoid nilnil
handleResult returned (net.Conn, error, bool) and produced (nil, nil, false)
on the non-terminal path, which the nilnil linter rejects. Return a
raceOutcome{conn, err, done} struct instead so the not-done case carries no
ambiguous nil value / nil error pair.
2026-07-01 17:05:46 +02:00
Zoltan Papp
49c0aeb6ce [relay] Update relay client tests for the new OpenConn signature
Manager.OpenConn now takes a RelayServer struct and a preferForeign
flag instead of a server address string and a serverIP. Adjust the
manager tests accordingly: bundle addr/IP into RelayServer and set
preferForeign=true for foreign-relay dials, false for home-relay dials.
2026-07-01 17:00:20 +02:00
Zoltan Papp
6774a43eae [relay] Rename ConnRacer to FallbackOpener and align file names with types
Rename the ConnRacer type to FallbackOpener and its constructor to
NewFallbackOpener. Rename fallback.go to fallback_opener.go and
foreign_relays.go to foreign_relays_store.go so file names mirror their
primary struct.
2026-07-01 16:53:42 +02:00
Zoltan Papp
50a29c07ce [relay] Race home and foreign relay for peer connection setup
When a peer advertises a relay different from the home relay, open the
peer connection by racing the home relay and the remote peer's relay in
parallel: start the preferred one (controller prefers home, otherwise the
remote relay), fall back to the other after a delay, use whichever connects
first and drain the loser.

Expose the foreign relay cache as ForeignRelaysStore and the race as
ConnRacer. The manager passes the remote relay as a RelayServer{Addr, IP}.
OpenConn no longer returns the winning server address; the worker derives it
from the returned conn's RemoteAddr to register the close listener, matching
the previous main-branch behavior.
2026-06-30 18:26:13 +02:00
Zoltan Papp
7d8e20030b [relay] Extract foreign relay client cache into a dedicated type
Move the foreign-relay client cache out of Manager into a foreignRelays
type. Concurrent first-time connects to the same server are deduplicated
with singleflight, so the cache mutex is never held during a network
connect (removing the previous stall where a slow connect blocked all map
operations). A per-entry in-use refcount prevents the cleanup loop from
closing a client while a connection is being opened on it.

This drops RelayTrack, its per-track lock and the hand-over-hand locking
between the map lock and the track lock. The exported API is unchanged.
2026-06-29 00:41:47 +02:00
17 changed files with 958 additions and 261 deletions

View File

@@ -247,6 +247,9 @@ func (c *Client) DebugBundle(platformFiles PlatformFiles, anonymize bool) (strin
deps.SyncResponse = resp
if e := cc.Engine(); e != nil {
deps.RefreshStatus = func() {
e.RunHealthProbes(context.Background(), true)
}
if cm := e.GetClientMetrics(); cm != nil {
deps.ClientMetrics = cm
}

View File

@@ -54,19 +54,15 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
w.relaySupportedOnRemotePeer.Store(true)
// the relayManager will return with error in case if the connection has lost with relay server
currentRelayAddress, _, err := w.relayManager.RelayInstanceAddress()
_, _, err := w.relayManager.RelayInstanceAddress()
if err != nil {
w.log.Errorf("failed to handle new offer: %s", err)
return
}
srv := w.preferredRelayServer(currentRelayAddress, remoteOfferAnswer.RelaySrvAddress)
var serverIP netip.Addr
if srv == remoteOfferAnswer.RelaySrvAddress {
serverIP = remoteOfferAnswer.RelaySrvIP
}
relayedConn, err := w.relayManager.OpenConn(w.peerCtx, srv, w.config.Key, serverIP)
preferForeign := !w.isController
remoteRelayServer := relayClient.RelayServer{Addr: remoteOfferAnswer.RelaySrvAddress, IP: remoteOfferAnswer.RelaySrvIP}
relayedConn, err := w.relayManager.OpenConn(w.peerCtx, remoteRelayServer, w.config.Key, preferForeign)
if err != nil {
if errors.Is(err, relayClient.ErrConnAlreadyExists) {
w.log.Debugf("handled offer by reusing existing relay connection")
@@ -80,14 +76,13 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
w.relayedConn = relayedConn
w.relayLock.Unlock()
err = w.relayManager.AddCloseListener(srv, w.onRelayClientDisconnected)
if err != nil {
log.Errorf("failed to add close listener: %s", err)
if err := w.relayManager.AddCloseListener(relayedConn.RemoteAddr().String(), w.onRelayClientDisconnected); err != nil {
w.log.Errorf("failed to add close listener: %s", err)
_ = relayedConn.Close()
return
}
w.log.Debugf("peer conn opened via Relay: %s", srv)
w.log.Debugf("peer conn opened via Relay: %s", relayedConn.RemoteAddr())
go w.conn.onRelayConnectionIsReady(RelayConnInfo{
relayedConn: relayedConn,
rosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
@@ -126,13 +121,6 @@ func (w *WorkerRelay) isRelaySupported(answer *OfferAnswer) bool {
return answer.RelaySrvAddress != ""
}
func (w *WorkerRelay) preferredRelayServer(myRelayAddress, remoteRelayAddress string) string {
if w.isController {
return myRelayAddress
}
return remoteRelayAddress
}
func (w *WorkerRelay) onRelayClientDisconnected() {
go w.conn.onRelayDisconnected()
}

View File

@@ -233,6 +233,9 @@ func (c *Client) DebugBundle(anonymize bool) (string, error) {
deps.SyncResponse = resp
if e := cc.Engine(); e != nil {
deps.RefreshStatus = func() {
e.RunHealthProbes(context.Background(), true)
}
if cm := e.GetClientMetrics(); cm != nil {
deps.ClientMetrics = cm
}

3
go.mod
View File

@@ -113,7 +113,7 @@ require (
github.com/ti-mo/conntrack v0.5.1
github.com/ti-mo/netfilter v0.5.2
github.com/vmihailenco/msgpack/v5 v5.4.1
github.com/wailsapp/wails/v3 v3.0.0-alpha2.111
github.com/wailsapp/wails/v3 v3.0.0-alpha2.117
github.com/yusufpapurcu/wmi v1.2.4
github.com/zcalusic/sysinfo v1.1.3
go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0
@@ -303,7 +303,6 @@ require (
github.com/tklauser/numcpus v0.10.0 // indirect
github.com/vishvananda/netns v0.0.5 // indirect
github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
github.com/wailsapp/wails/webview2 v1.0.27 // indirect
github.com/wlynxg/anet v0.0.5 // indirect
github.com/x448/float16 v0.8.4 // indirect
github.com/zeebo/blake3 v0.2.3 // indirect

6
go.sum
View File

@@ -660,10 +660,8 @@ github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IU
github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
github.com/wailsapp/wails/v3 v3.0.0-alpha2.111 h1:MKx1nOnhnDuEGrRBmtxLOJq1NERwailu2cI4BvzWhi4=
github.com/wailsapp/wails/v3 v3.0.0-alpha2.111/go.mod h1:wrdvmyeCsB/K3YqJDoH8E3MwcN8NXAMnEFaDTW46w60=
github.com/wailsapp/wails/webview2 v1.0.27 h1:wjgAi/I8BBZ7kUGU8um3XF3ILEfzr96Q2Q1G4GPjMns=
github.com/wailsapp/wails/webview2 v1.0.27/go.mod h1:zdM4jcO1IaC61RiJL5F1BzgoqBHFIdacz8gPr5exr0o=
github.com/wailsapp/wails/v3 v3.0.0-alpha2.117 h1:udyjqPG3AIgkod5QDR/WblCkpV8R86BFPSrsWxSyt5Y=
github.com/wailsapp/wails/v3 v3.0.0-alpha2.117/go.mod h1:74WH2FScMsgucZvHHvv7eOefDXCm/CjuIxqhhZgPhKg=
github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU=
github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=

View File

@@ -0,0 +1,38 @@
package llm
import (
"regexp"
"strings"
)
// bedrockRegionPrefixes are the cross-region inference-profile prefixes that
// front a Bedrock model id (e.g. "eu.anthropic.claude-...").
var bedrockRegionPrefixes = []string{"us.", "eu.", "apac.", "global."}
// bedrockVersionSuffix matches the trailing "-vN[:N]" or "-YYYYMMDD-vN[:N]"
// version/throughput suffix of a Bedrock model id.
var bedrockVersionSuffix = regexp.MustCompile(`-(\d{8}-)?v\d+(:\d+)?$`)
// NormalizeBedrockModel strips an ARN wrapper, a cross-region inference-profile
// prefix, and the version/throughput suffix from a Bedrock model id so it
// matches the catalog/pricing key, e.g.
// "eu.anthropic.claude-sonnet-4-5-20250929-v1:0" -> "anthropic.claude-sonnet-4-5"
// and the inference-profile ARN's last segment likewise. It is the single
// source of truth shared by the request parser (which normalizes the request
// model from the URL path) and the router (which normalizes the operator's
// registered Bedrock model ids so both sides compare equal).
func NormalizeBedrockModel(modelID string) string {
m := modelID
if strings.HasPrefix(m, "arn:") {
if i := strings.LastIndex(m, "/"); i >= 0 {
m = m[i+1:]
}
}
for _, p := range bedrockRegionPrefixes {
if strings.HasPrefix(m, p) {
m = m[len(p):]
break
}
}
return bedrockVersionSuffix.ReplaceAllString(m, "")
}

View File

@@ -0,0 +1,23 @@
package llm
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestNormalizeBedrockModel(t *testing.T) {
cases := map[string]string{
"eu.anthropic.claude-sonnet-4-5-20250929-v1:0": "anthropic.claude-sonnet-4-5",
"us.anthropic.claude-haiku-4-5": "anthropic.claude-haiku-4-5",
"us.anthropic.claude-opus-4-8-20250101-v1:0": "anthropic.claude-opus-4-8",
"anthropic.claude-sonnet-4-5-20250929-v1:0": "anthropic.claude-sonnet-4-5",
"meta.llama3-3-70b-instruct-v1:0": "meta.llama3-3-70b-instruct",
"amazon.nova-pro-v1:0": "amazon.nova-pro",
// Inference-profile ARN — model id lives in the last path segment.
"arn:aws:bedrock:eu-central-1:123456789012:inference-profile/eu.anthropic.claude-sonnet-4-5-20250929-v1:0": "anthropic.claude-sonnet-4-5",
}
for in, want := range cases {
require.Equal(t, want, NormalizeBedrockModel(in), "normalize %q", in)
}
}

View File

@@ -0,0 +1,30 @@
package llm_router
import (
"testing"
"github.com/stretchr/testify/assert"
)
// TestRouteClaimsModel_BedrockNormalizesCandidate guards the fix for the native
// Bedrock routing gap: the request model reaches the router already normalized
// (the parser strips the region/inference-profile prefix and version suffix),
// so a provider registered with the raw inference-profile id must still match.
func TestRouteClaimsModel_BedrockNormalizesCandidate(t *testing.T) {
route := ProviderRoute{Bedrock: true, Models: []string{"us.anthropic.claude-haiku-4-5"}}
assert.True(t, routeClaimsModel(route, "anthropic.claude-haiku-4-5"),
"raw region-prefixed Bedrock model must match the normalized request model")
assert.False(t, routeClaimsModel(route, "anthropic.claude-opus-4-8"),
"a model outside the provider's list must not match")
// A provider registered with the already-normalized id also matches.
normalized := ProviderRoute{Bedrock: true, Models: []string{"anthropic.claude-haiku-4-5"}}
assert.True(t, routeClaimsModel(normalized, "anthropic.claude-haiku-4-5"),
"normalized Bedrock model must match")
// Non-Bedrock routes keep exact matching (no prefix stripping).
openai := ProviderRoute{Models: []string{"gpt-4o"}}
assert.True(t, routeClaimsModel(openai, "gpt-4o"), "exact model must match")
assert.False(t, routeClaimsModel(openai, "us.gpt-4o"),
"non-Bedrock routes must not strip a us. prefix")
}

View File

@@ -23,6 +23,7 @@ import (
"golang.org/x/oauth2"
"golang.org/x/oauth2/google"
"github.com/netbirdio/netbird/proxy/internal/llm"
"github.com/netbirdio/netbird/proxy/internal/middleware"
)
@@ -555,6 +556,14 @@ func routeClaimsModel(route ProviderRoute, model string) bool {
if candidate == model {
return true
}
// Bedrock request models reach the router already normalized (the parser
// strips the region / inference-profile prefix and version suffix), but
// the operator may register the raw inference-profile id (e.g.
// "us.anthropic.claude-haiku-4-5"). Normalize the candidate so both sides
// compare equal; otherwise a native Bedrock request denies as not-routable.
if route.Bedrock && llm.NormalizeBedrockModel(candidate) == model {
return true
}
}
return false
}

View File

@@ -0,0 +1,176 @@
package client
import (
"context"
"errors"
"net"
"time"
log "github.com/sirupsen/logrus"
)
const (
raceTotalTimeout = 40 * time.Second
raceFallbackDelay = 10 * time.Second
)
type raceAttempt struct {
conn net.Conn
err error
}
type raceOutcome struct {
conn net.Conn
err error
done bool
}
type connRace struct {
opener *FallbackOpener
peerKey string
remoteRelayServer RelayServer
preferForeign bool
raceCtx context.Context
otherCtx context.Context
cancelPreferred context.CancelFunc
cancelOther context.CancelFunc
results chan raceAttempt
fallbackTimer *time.Timer
otherStarted bool
settled int
lastErr error
}
type FallbackOpener struct {
home *Client
foreignStore *ForeignRelaysStore
fallbackDelay time.Duration
totalTimeout time.Duration
// openFn performs a single attempt. It is overridable in tests; when nil the
// real home/foreign dispatch in open is used.
openFn func(ctx context.Context, peerKey string, remoteRelayServer RelayServer, foreign bool) raceAttempt
}
func NewFallbackOpener(home *Client, foreignStore *ForeignRelaysStore) *FallbackOpener {
return &FallbackOpener{
home: home,
foreignStore: foreignStore,
fallbackDelay: raceFallbackDelay,
totalTimeout: raceTotalTimeout,
}
}
func (r *FallbackOpener) Run(ctx context.Context, peerKey string, remoteRelayServer RelayServer, preferForeign bool) (net.Conn, error) {
raceCtx, cancel := context.WithTimeout(ctx, r.totalTimeout)
defer cancel()
preferredCtx, cancelPreferred := context.WithCancel(raceCtx)
otherCtx, cancelOther := context.WithCancel(raceCtx)
race := &connRace{
opener: r,
peerKey: peerKey,
remoteRelayServer: remoteRelayServer,
preferForeign: preferForeign,
raceCtx: raceCtx,
otherCtx: otherCtx,
cancelPreferred: cancelPreferred,
cancelOther: cancelOther,
results: make(chan raceAttempt, 2),
fallbackTimer: time.NewTimer(r.fallbackDelay),
}
defer race.fallbackTimer.Stop()
go func() {
race.results <- r.open(preferredCtx, peerKey, remoteRelayServer, preferForeign)
}()
for {
select {
case <-race.fallbackTimer.C:
race.startOther()
case res := <-race.results:
if o := race.handleResult(res); o.done {
return o.conn, o.err
}
case <-raceCtx.Done():
return race.onTimeout()
}
}
}
func (c *connRace) startOther() {
if c.otherStarted {
return
}
c.otherStarted = true
c.fallbackTimer.Stop()
go func() {
c.results <- c.opener.open(c.otherCtx, c.peerKey, c.remoteRelayServer, !c.preferForeign)
}()
}
func (c *connRace) handleResult(res raceAttempt) raceOutcome {
if (res.err == nil && res.conn != nil) || errors.Is(res.err, ErrConnAlreadyExists) {
c.settled++
c.stop()
return raceOutcome{conn: res.conn, err: res.err, done: true}
}
c.lastErr = res.err
c.settled++
if !c.otherStarted {
c.startOther()
return raceOutcome{}
}
if c.settled == 2 {
c.cancelPreferred()
c.cancelOther()
return raceOutcome{err: c.lastErr, done: true}
}
return raceOutcome{}
}
func (c *connRace) onTimeout() (net.Conn, error) {
c.stop()
if c.lastErr != nil {
return nil, c.lastErr
}
return nil, c.raceCtx.Err()
}
func (c *connRace) stop() {
c.cancelPreferred()
c.cancelOther()
go c.opener.drainLoser(c.results, c.settled, c.otherStarted)
}
func (r *FallbackOpener) open(ctx context.Context, peerKey string, remoteRelayServer RelayServer, foreign bool) raceAttempt {
if r.openFn != nil {
return r.openFn(ctx, peerKey, remoteRelayServer, foreign)
}
if foreign {
conn, err := r.foreignStore.OpenConn(ctx, peerKey, remoteRelayServer)
return raceAttempt{conn: conn, err: err}
}
conn, err := r.home.OpenConn(ctx, peerKey)
return raceAttempt{conn: conn, err: err}
}
func (r *FallbackOpener) drainLoser(results chan raceAttempt, settled int, otherStarted bool) {
started := 1
if otherStarted {
started = 2
}
for i := settled; i < started; i++ {
res := <-results
if res.conn != nil {
if err := res.conn.Close(); err != nil {
log.Debugf("failed to close losing relay connection: %v", err)
}
}
}
}

View File

@@ -0,0 +1,450 @@
package client
import (
"context"
"errors"
"fmt"
"net"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// The FallbackOpener race is driven by a single goroutine (Run's select loop)
// with worker goroutines that communicate only through the buffered results
// channel and the two cancel contexts. These tests exercise that state machine
// in isolation via an injected openFn, so no relay server or network is needed.
// Timing is scaled down through the fallbackDelay/totalTimeout fields.
// raceFakeConn tracks whether Close was called. Only Close is exercised by the
// race logic (drainLoser closes losers; Run returns the winner untouched).
type raceFakeConn struct {
net.Conn
label string
closed atomic.Bool
}
func (c *raceFakeConn) Close() error {
c.closed.Store(true)
return nil
}
// raceAttemptScript describes how a single scripted attempt behaves.
type raceAttemptScript struct {
delay time.Duration
conn *raceFakeConn // non-nil => the attempt succeeds and returns this conn
err error // returned when conn is nil
// ignoreCtx makes the attempt complete after delay even if its context is
// cancelled. It models an OpenConn that produced a real connection right as
// the race cancelled it - exactly the case drainLoser must clean up.
ignoreCtx bool
}
// fakeOpener replaces FallbackOpener.open. Scripts are keyed by the foreign
// flag, so which script is "preferred" depends on the preferForeign argument
// passed to Run.
type fakeOpener struct {
mu sync.Mutex
scripts map[bool]raceAttemptScript
calls []bool // foreign flag of each open() invocation, in order
}
func (f *fakeOpener) open(ctx context.Context, _ string, _ RelayServer, foreign bool) raceAttempt {
f.mu.Lock()
f.calls = append(f.calls, foreign)
s, ok := f.scripts[foreign]
f.mu.Unlock()
if !ok {
return raceAttempt{err: fmt.Errorf("no script for foreign=%v", foreign)}
}
if s.delay > 0 {
timer := time.NewTimer(s.delay)
defer timer.Stop()
if s.ignoreCtx {
<-timer.C
} else {
select {
case <-timer.C:
case <-ctx.Done():
return raceAttempt{err: ctx.Err()}
}
}
}
if s.conn != nil {
return raceAttempt{conn: s.conn}
}
return raceAttempt{err: s.err}
}
func (f *fakeOpener) callCount() int {
f.mu.Lock()
defer f.mu.Unlock()
return len(f.calls)
}
func (f *fakeOpener) firstCallForeign(t *testing.T) bool {
t.Helper()
f.mu.Lock()
defer f.mu.Unlock()
require.NotEmpty(t, f.calls, "expected at least one open attempt")
return f.calls[0]
}
func newTestOpener(f *fakeOpener, fallbackDelay, totalTimeout time.Duration) *FallbackOpener {
o := NewFallbackOpener(nil, nil)
o.openFn = f.open
o.fallbackDelay = fallbackDelay
o.totalTimeout = totalTimeout
return o
}
const (
// controller prefers the home relay, i.e. preferForeign == false.
preferHome = false
preferForeign = true
)
var errAttempt = errors.New("attempt failed")
// The preferred attempt wins before the fallback timer fires, so the other
// attempt is never started.
func TestFallbackOpener_PreferredWinsImmediately(t *testing.T) {
homeConn := &raceFakeConn{label: "home"}
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {conn: homeConn}, // preferred (home): instant success
true: {delay: 5 * time.Second, conn: foreignConn}, // would never finish in time
}}
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.NoError(t, err)
require.Same(t, homeConn, conn)
assert.Equal(t, 1, f.callCount(), "other attempt must not start when preferred wins first")
assert.False(t, f.firstCallForeign(t), "home must be tried first when preferring home")
assert.False(t, foreignConn.closed.Load())
}
// preferForeign flips which relay is tried first.
func TestFallbackOpener_PreferForeignRoutesForeignFirst(t *testing.T) {
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
true: {conn: foreignConn}, // preferred (foreign): instant success
false: {delay: 5 * time.Second, conn: &raceFakeConn{}},
}}
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferForeign)
require.NoError(t, err)
require.Same(t, foreignConn, conn)
assert.Equal(t, 1, f.callCount())
assert.True(t, f.firstCallForeign(t), "foreign must be tried first when preferring foreign")
}
// ErrConnAlreadyExists counts as success: Run returns it and does not start the
// other attempt.
func TestFallbackOpener_ErrConnAlreadyExistsIsSuccess(t *testing.T) {
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {err: ErrConnAlreadyExists},
true: {delay: 5 * time.Second, conn: &raceFakeConn{}},
}}
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.ErrorIs(t, err, ErrConnAlreadyExists)
assert.Nil(t, conn)
assert.Equal(t, 1, f.callCount(), "other attempt must not start on ErrConnAlreadyExists")
}
// A preferred failure starts the other attempt immediately, without waiting for
// the fallback timer.
func TestFallbackOpener_PreferredFailsStartsOtherBeforeTimer(t *testing.T) {
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {err: errAttempt}, // preferred fails instantly
true: {delay: 5 * time.Millisecond, conn: foreignConn},
}}
fallbackDelay := 500 * time.Millisecond
o := newTestOpener(f, fallbackDelay, 2*time.Second)
start := time.Now()
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
elapsed := time.Since(start)
require.NoError(t, err)
require.Same(t, foreignConn, conn)
assert.Equal(t, 2, f.callCount())
assert.Less(t, elapsed, fallbackDelay/2, "fallback must not wait for the timer after a preferred failure")
}
// When the preferred attempt is slow, the fallback timer starts the other
// attempt and its success wins.
func TestFallbackOpener_TimerStartsOtherWhenPreferredSlow(t *testing.T) {
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {delay: 5 * time.Second}, // preferred hangs until cancelled
true: {delay: 5 * time.Millisecond, conn: foreignConn},
}}
fallbackDelay := 40 * time.Millisecond
o := newTestOpener(f, fallbackDelay, 2*time.Second)
start := time.Now()
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
elapsed := time.Since(start)
require.NoError(t, err)
require.Same(t, foreignConn, conn)
assert.Equal(t, 2, f.callCount())
assert.GreaterOrEqual(t, elapsed, fallbackDelay, "other must not start before the fallback timer fires")
}
// Both attempts fail: Run returns the last error and tries both relays.
func TestFallbackOpener_BothFail(t *testing.T) {
errOther := errors.New("other failed")
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {err: errAttempt},
true: {err: errOther},
}}
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.Error(t, err)
assert.Nil(t, conn)
assert.ErrorIs(t, err, errOther, "the most recent error should be surfaced")
assert.Equal(t, 2, f.callCount())
}
// When both attempts succeed, drainLoser must close the losing connection so it
// is not leaked. Here the preferred attempt wins and the foreign loser - which
// produced a real conn despite being cancelled - is closed.
func TestFallbackOpener_DoubleSuccessClosesLoser(t *testing.T) {
homeConn := &raceFakeConn{label: "home"}
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {delay: 30 * time.Millisecond, conn: homeConn}, // preferred wins
true: {delay: 80 * time.Millisecond, conn: foreignConn, ignoreCtx: true}, // loser yields a conn after cancel
}}
o := newTestOpener(f, 15*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.NoError(t, err)
require.Same(t, homeConn, conn)
assert.Equal(t, 2, f.callCount())
assert.False(t, homeConn.closed.Load(), "the winning connection must not be closed")
require.Eventually(t, foreignConn.closed.Load, time.Second, 5*time.Millisecond,
"the losing connection must be closed by drainLoser")
}
// Winner selection is purely by result arrival order, not by preference: when
// the non-preferred attempt returns first it wins even though home was
// preferred. This is the mechanism behind the split-relay concern - two peers
// racing independently have no shared tie-break, so under adversarial timing
// they can settle on different relays. Documented here as current behavior.
func TestFallbackOpener_FasterOtherWinsDespitePreference(t *testing.T) {
homeConn := &raceFakeConn{label: "home"}
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {delay: 60 * time.Millisecond, conn: homeConn, ignoreCtx: true}, // preferred but slower
true: {delay: 5 * time.Millisecond, conn: foreignConn}, // other is faster
}}
o := newTestOpener(f, 15*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.NoError(t, err)
require.Same(t, foreignConn, conn, "the first successful attempt wins regardless of preference")
require.Eventually(t, homeConn.closed.Load, time.Second, 5*time.Millisecond,
"the slower preferred attempt becomes the loser and is closed")
}
// The whole race is bounded by totalTimeout. With no attempt succeeding or
// failing, Run returns the deadline error.
func TestFallbackOpener_TotalTimeout(t *testing.T) {
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {delay: 5 * time.Second},
true: {delay: 5 * time.Second},
}}
o := newTestOpener(f, 20*time.Millisecond, 80*time.Millisecond)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.Error(t, err)
assert.Nil(t, conn)
assert.ErrorIs(t, err, context.DeadlineExceeded)
}
// Cancelling the caller's context aborts the race promptly with the cancel
// error, even before the fallback timer would fire.
func TestFallbackOpener_ParentContextCanceled(t *testing.T) {
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {delay: 5 * time.Second},
true: {delay: 5 * time.Second},
}}
// fallbackDelay large so the timer never fires; only the parent cancel ends the race.
o := newTestOpener(f, 5*time.Second, 5*time.Second)
ctx, cancel := context.WithCancel(context.Background())
go func() {
time.Sleep(20 * time.Millisecond)
cancel()
}()
start := time.Now()
conn, err := o.Run(ctx, "peer", RelayServer{Addr: "srv"}, preferHome)
require.Error(t, err)
assert.Nil(t, conn)
assert.ErrorIs(t, err, context.Canceled)
assert.Less(t, time.Since(start), time.Second, "must return shortly after the parent context is cancelled")
assert.Equal(t, 1, f.callCount(), "the other attempt must not start")
}
// rendezvous models the relay-level requirement that a relayed connection is
// established only once BOTH peers subscribe to the same relay server. arrive
// records a peer's presence on a relay and returns a channel that closes when
// the second peer arrives, so an attempt can only complete after a real
// rendezvous - the same coupling the production code depends on.
type rendezvous struct {
mu sync.Mutex
arrivals map[string]int
gates map[string]chan struct{}
}
func newRendezvous() *rendezvous {
return &rendezvous{arrivals: map[string]int{}, gates: map[string]chan struct{}{}}
}
func (r *rendezvous) arrive(relay string) <-chan struct{} {
r.mu.Lock()
defer r.mu.Unlock()
g, ok := r.gates[relay]
if !ok {
g = make(chan struct{})
r.gates[relay] = g
}
r.arrivals[relay]++
if r.arrivals[relay] == 2 {
close(g)
}
return g
}
// splitPeer is one peer's view of the two relays. relayFor maps the foreign
// flag to a relay name; postDelay is how long after the rendezvous that peer's
// OpenConn takes to return (its per-relay subscribe latency). Different values
// per peer model the asymmetric timing that triggers finding #1.
type splitPeer struct {
rv *rendezvous
relayFor map[bool]string
postDelay map[string]time.Duration
}
func (p *splitPeer) open(ctx context.Context, _ string, _ RelayServer, foreign bool) raceAttempt {
relay := p.relayFor[foreign]
select {
case <-p.rv.arrive(relay):
case <-ctx.Done():
return raceAttempt{err: ctx.Err()}
}
timer := time.NewTimer(p.postDelay[relay])
defer timer.Stop()
select {
case <-timer.C:
case <-ctx.Done():
return raceAttempt{err: ctx.Err()}
}
return raceAttempt{conn: &raceFakeConn{label: relay}}
}
// TestFallbackOpener_SplitRelaySelection reproduces finding #1: the two peers
// run FallbackOpener.Run independently with no shared tie-break, so the winner
// is chosen purely by local result-arrival order. Under an adversarial - but
// self-consistent - timing profile they settle on DIFFERENT relays.
//
// Both peers prefer relayA (the controller's home). The split needs each peer's
// preferred relayA to be slow enough that both start their fallback (so both
// relays actually rendezvous), and then each peer's fast path to be a different
// relay:
// - peerA: relayA slow (abandoned), relayB fast -> peerA wins relayB
// - peerB: relayA fast (wins), relayB slow -> peerB wins relayA
//
// Each winner then cancels its attempt on the relay the OTHER peer actually
// kept, leaving two half-open relayed connections that were both reported as
// successful. When a deterministic cross-peer tie-break is added to fix this,
// invert the assertion below to require convergence.
func TestFallbackOpener_SplitRelaySelection(t *testing.T) {
const (
relayA = "relayA" // controller's home relay; both peers prefer it
relayB = "relayB" // non-controller's home relay
)
rv := newRendezvous()
peerA := &splitPeer{
rv: rv,
relayFor: map[bool]string{false: relayA, true: relayB}, // home=relayA
postDelay: map[string]time.Duration{
relayA: 500 * time.Millisecond, // preferred but slow -> abandoned
relayB: 10 * time.Millisecond, // fallback is fast -> peerA wins relayB
},
}
peerB := &splitPeer{
rv: rv,
relayFor: map[bool]string{false: relayB, true: relayA}, // home=relayB
postDelay: map[string]time.Duration{
relayA: 80 * time.Millisecond, // preferred, wins - but only after starting fallback
relayB: 500 * time.Millisecond, // fallback (home) is slow -> abandoned
},
}
fallbackDelay := 30 * time.Millisecond
newPeerOpener := func(p *splitPeer) *FallbackOpener {
o := NewFallbackOpener(nil, nil)
o.openFn = p.open
o.fallbackDelay = fallbackDelay
o.totalTimeout = 5 * time.Second
return o
}
oA := newPeerOpener(peerA)
oB := newPeerOpener(peerB)
type result struct {
conn net.Conn
err error
}
var ra, rb result
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
ra.conn, ra.err = oA.Run(context.Background(), "peerB", RelayServer{Addr: relayB}, preferHome)
}()
go func() {
defer wg.Done()
rb.conn, rb.err = oB.Run(context.Background(), "peerA", RelayServer{Addr: relayA}, preferForeign)
}()
wg.Wait()
require.NoError(t, ra.err)
require.NoError(t, rb.err)
aRelay := ra.conn.(*raceFakeConn).label
bRelay := rb.conn.(*raceFakeConn).label
t.Logf("peerA settled on %s, peerB settled on %s", aRelay, bRelay)
assert.Equal(t, aRelay, bRelay,
"peers selected different relays with no cross-peer tie-break")
assert.Equal(t, relayB, aRelay, "peerA abandoned its slow preferred relay and won the fallback")
assert.Equal(t, relayA, bRelay, "peerB won its preferred relay after starting the fallback")
}

View File

@@ -0,0 +1,155 @@
package client
import (
"context"
"net"
"sync"
"time"
log "github.com/sirupsen/logrus"
"golang.org/x/sync/singleflight"
relayAuth "github.com/netbirdio/netbird/shared/relay/auth/hmac"
)
type foreignRelay struct {
client *Client
created time.Time
inUse int
}
type ForeignRelaysStore struct {
mu sync.RWMutex
clients map[string]*foreignRelay
group singleflight.Group
ctx context.Context
tokenStore *relayAuth.TokenStore
peerID string
mtu uint16
transportFallback *transportFallback
onDisconnect func(string)
keepUnusedServerTime time.Duration
}
func NewForeignRelaysStore(ctx context.Context, tokenStore *relayAuth.TokenStore, peerID string, mtu uint16, transportFallback *transportFallback, onDisconnect func(string), keepUnusedServerTime time.Duration) *ForeignRelaysStore {
return &ForeignRelaysStore{
clients: make(map[string]*foreignRelay),
ctx: ctx,
tokenStore: tokenStore,
peerID: peerID,
mtu: mtu,
transportFallback: transportFallback,
onDisconnect: onDisconnect,
keepUnusedServerTime: keepUnusedServerTime,
}
}
func (f *ForeignRelaysStore) OpenConn(ctx context.Context, peerKey string, remoteRelayServer RelayServer) (net.Conn, error) {
fr, err := f.acquire(remoteRelayServer)
if err != nil {
return nil, err
}
defer f.release(fr)
return fr.client.OpenConn(ctx, peerKey)
}
func (f *ForeignRelaysStore) acquire(remoteRelayServer RelayServer) (*foreignRelay, error) {
f.mu.Lock()
if fr, ok := f.clients[remoteRelayServer.Addr]; ok {
fr.inUse++
f.mu.Unlock()
return fr, nil
}
f.mu.Unlock()
v, err, _ := f.group.Do(remoteRelayServer.Addr, func() (any, error) {
f.mu.RLock()
fr, ok := f.clients[remoteRelayServer.Addr]
f.mu.RUnlock()
if ok {
return fr, nil
}
relayClient := NewClientWithServerIP(remoteRelayServer.Addr, remoteRelayServer.IP, f.tokenStore, f.peerID, f.mtu)
relayClient.SetTransportFallback(f.transportFallback)
if err := relayClient.Connect(f.ctx); err != nil {
return nil, err
}
relayClient.SetOnDisconnectListener(f.onDisconnect)
f.mu.Lock()
fr = &foreignRelay{client: relayClient, created: time.Now()}
f.clients[remoteRelayServer.Addr] = fr
f.mu.Unlock()
return fr, nil
})
if err != nil {
return nil, err
}
fr := v.(*foreignRelay)
f.mu.Lock()
if cur, ok := f.clients[remoteRelayServer.Addr]; !ok || cur != fr {
f.mu.Unlock()
return f.acquire(remoteRelayServer)
}
fr.inUse++
f.mu.Unlock()
return fr, nil
}
func (f *ForeignRelaysStore) release(fr *foreignRelay) {
f.mu.Lock()
fr.inUse--
f.mu.Unlock()
}
func (f *ForeignRelaysStore) evict(serverAddress string) {
f.mu.Lock()
defer f.mu.Unlock()
if _, ok := f.clients[serverAddress]; ok {
delete(f.clients, serverAddress)
log.Debugf("evicted disconnected foreign relay client: %s", serverAddress)
}
}
func (f *ForeignRelaysStore) cleanupUnused() {
f.mu.Lock()
defer f.mu.Unlock()
for addr, fr := range f.clients {
if time.Since(fr.created) <= f.keepUnusedServerTime {
continue
}
if fr.inUse > 0 {
continue
}
if fr.client.HasConns() {
continue
}
fr.client.SetOnDisconnectListener(nil)
go func() {
_ = fr.client.Close()
}()
log.Debugf("clean up unused relay server connection: %s", addr)
delete(f.clients, addr)
}
}
func (f *ForeignRelaysStore) states() []RelayConnState {
f.mu.RLock()
clients := make([]*Client, 0, len(f.clients))
for _, fr := range f.clients {
clients = append(clients, fr.client)
}
f.mu.RUnlock()
states := make([]RelayConnState, 0, len(clients))
for _, c := range clients {
states = append(states, relayConnState(c))
}
return states
}

View File

@@ -22,27 +22,6 @@ var (
ErrRelayClientNotConnected = fmt.Errorf("relay client not connected")
)
// RelayTrack hold the relay clients for the foreign relay servers.
// With the mutex can ensure we can open new connection in case the relay connection has been established with
// the relay server.
type RelayTrack struct {
sync.RWMutex
relayClient *Client
err error
created time.Time
// ready is closed once the dial started by openConnVia finishes (relayClient
// or err is set). Callers reusing a track wait on this instead of the track
// lock, so the dial never runs under rt.Lock.
ready chan struct{}
}
func NewRelayTrack() *RelayTrack {
return &RelayTrack{
created: time.Now(),
ready: make(chan struct{}),
}
}
type OnServerCloseListener func()
// ManagerOption configures a Manager at construction time.
@@ -59,6 +38,11 @@ type RelayConnState struct {
Err error
}
type RelayServer struct {
Addr string
IP netip.Addr
}
// WithMaxBackoffInterval caps the exponential backoff between reconnect
// attempts to the home relay. A non-positive value keeps the default.
func WithMaxBackoffInterval(d time.Duration) ManagerOption {
@@ -83,8 +67,7 @@ type Manager struct {
relayClientMu sync.RWMutex
reconnectGuard *Guard
relayClients map[string]*RelayTrack
relayClientsMutex sync.RWMutex
foreign *ForeignRelaysStore
onDisconnectedListeners map[string]*list.List
onReconnectedListenerFn func()
@@ -120,7 +103,6 @@ func NewManager(ctx context.Context, serverURLs []string, peerID string, mtu uin
ConnectionTimeout: defaultConnectionTimeout,
TransportFallback: tf,
},
relayClients: make(map[string]*RelayTrack),
onDisconnectedListeners: make(map[string]*list.List),
cleanupInterval: relayCleanupInterval,
keepUnusedServerTime: keepUnusedServerTime,
@@ -128,6 +110,7 @@ func NewManager(ctx context.Context, serverURLs []string, peerID string, mtu uin
for _, opt := range opts {
opt(m)
}
m.foreign = NewForeignRelaysStore(ctx, tokenStore, peerID, mtu, tf, m.onServerDisconnected, m.keepUnusedServerTime)
m.serverPicker.ServerURLs.Store(serverURLs)
m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval)
return m
@@ -159,40 +142,26 @@ func (m *Manager) Serve() error {
return err
}
// OpenConn opens a connection to the given peer key. If the peer is on the same relay server, the connection will be
// established via the relay server. If the peer is on a different relay server, the manager will establish a new
// connection to the relay server. It returns back with a net.Conn what represent the remote peer connection.
//
// serverIP, when valid and serverAddress is foreign, is used as a dial target if the FQDN-based dial fails.
// Ignored for the local home-server path. TLS verification still uses the FQDN via SNI.
func (m *Manager) OpenConn(ctx context.Context, serverAddress, peerKey string, serverIP netip.Addr) (net.Conn, error) {
func (m *Manager) OpenConn(ctx context.Context, remoteRelayServer RelayServer, peerKey string, preferForeign bool) (net.Conn, error) {
m.relayClientMu.RLock()
defer m.relayClientMu.RUnlock()
relayClient := m.relayClient
m.relayClientMu.RUnlock()
if m.relayClient == nil {
if relayClient == nil {
return nil, ErrRelayClientNotConnected
}
foreign, err := m.isForeignServer(serverAddress)
foreign, err := m.isForeignServer(relayClient, remoteRelayServer.Addr)
if err != nil {
return nil, err
}
var (
netConn net.Conn
)
if !foreign {
log.Debugf("open peer connection via permanent server: %s", peerKey)
netConn, err = m.relayClient.OpenConn(ctx, peerKey)
} else {
log.Debugf("open peer connection via foreign server: %s", serverAddress)
netConn, err = m.openConnVia(ctx, serverAddress, peerKey, serverIP)
}
if err != nil {
return nil, err
return relayClient.OpenConn(ctx, peerKey)
}
return netConn, err
opener := NewFallbackOpener(relayClient, m.foreign)
return opener.Run(ctx, peerKey, remoteRelayServer, preferForeign)
}
// Ready returns true if the home Relay client is connected to the relay server.
@@ -223,7 +192,7 @@ func (m *Manager) AddCloseListener(serverAddress string, onClosedListener OnServ
return ErrRelayClientNotConnected
}
foreign, err := m.isForeignServer(serverAddress)
foreign, err := m.isForeignServer(m.relayClient, serverAddress)
if err != nil {
return err
}
@@ -287,26 +256,7 @@ func (m *Manager) RelayStates() []RelayConnState {
states = append(states, st)
}
// Snapshot the tracks, then query each outside the map lock: a track can be
// held by an in-progress Connect, and blocking on it must not stall other
// relay operations.
m.relayClientsMutex.RLock()
tracks := make([]*RelayTrack, 0, len(m.relayClients))
for _, rt := range m.relayClients {
tracks = append(tracks, rt)
}
m.relayClientsMutex.RUnlock()
// Only connected foreign relays carry state; a failed connect is evicted
// immediately (openConnVia), so there is no error state to surface.
for _, rt := range tracks {
rt.RLock()
rc := rt.relayClient
rt.RUnlock()
if rc != nil {
states = append(states, relayConnState(rc))
}
}
states = append(states, m.foreign.states()...)
return states
}
@@ -327,76 +277,6 @@ func (m *Manager) UpdateToken(token *relayAuth.Token) error {
return m.tokenStore.UpdateToken(token)
}
func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string, serverIP netip.Addr) (net.Conn, error) {
// check if already has a connection to the desired relay server
m.relayClientsMutex.RLock()
rt, ok := m.relayClients[serverAddress]
m.relayClientsMutex.RUnlock()
if ok {
return m.openConnOnTrack(ctx, rt, peerKey)
}
// if not, establish a new connection but check it again (because changed the lock type) before starting the
// connection
m.relayClientsMutex.Lock()
rt, ok = m.relayClients[serverAddress]
if ok {
m.relayClientsMutex.Unlock()
return m.openConnOnTrack(ctx, rt, peerKey)
}
// Publish the track and release the map lock BEFORE dialing, so the dial does
// not run under rt.Lock (which would block RelayStates and the cleanup loop
// for the full dial). Concurrent callers find this track and wait on rt.ready.
rt = NewRelayTrack()
m.relayClients[serverAddress] = rt
m.relayClientsMutex.Unlock()
relayClient := NewClientWithServerIP(serverAddress, serverIP, m.tokenStore, m.peerID, m.mtu)
relayClient.SetTransportFallback(m.transportFallback)
err := relayClient.Connect(m.ctx)
if err != nil {
rt.Lock()
rt.err = err
rt.Unlock()
close(rt.ready)
m.relayClientsMutex.Lock()
delete(m.relayClients, serverAddress)
m.relayClientsMutex.Unlock()
return nil, err
}
// if connection closed then delete the relay client from the list
relayClient.SetOnDisconnectListener(m.onServerDisconnected)
rt.Lock()
rt.relayClient = relayClient
rt.Unlock()
close(rt.ready)
return relayClient.OpenConn(ctx, peerKey)
}
// openConnOnTrack opens a peer connection through an existing relay track,
// waiting for the dial started by another openConnVia call to finish. It waits
// on rt.ready rather than the track lock, so it neither holds nor contends the
// track lock across the dial.
func (m *Manager) openConnOnTrack(ctx context.Context, rt *RelayTrack, peerKey string) (net.Conn, error) {
select {
case <-rt.ready:
case <-ctx.Done():
return nil, ctx.Err()
}
rt.RLock()
defer rt.RUnlock()
if rt.err != nil {
return nil, rt.err
}
if rt.relayClient == nil {
return nil, ErrRelayClientNotConnected
}
return rt.relayClient.OpenConn(ctx, peerKey)
}
func (m *Manager) onServerConnected() {
m.listenerLock.Lock()
defer m.listenerLock.Unlock()
@@ -422,21 +302,12 @@ func (m *Manager) onServerDisconnected(serverAddress string) {
m.relayClientMu.Unlock()
if !isHome {
m.evictForeignRelay(serverAddress)
m.foreign.evict(serverAddress)
}
m.notifyOnDisconnectListeners(serverAddress)
}
func (m *Manager) evictForeignRelay(serverAddress string) {
m.relayClientsMutex.Lock()
defer m.relayClientsMutex.Unlock()
if _, ok := m.relayClients[serverAddress]; ok {
delete(m.relayClients, serverAddress)
log.Debugf("evicted disconnected foreign relay client: %s", serverAddress)
}
}
func (m *Manager) listenGuardEvent(ctx context.Context) {
for {
select {
@@ -459,8 +330,8 @@ func (m *Manager) storeClient(client *Client) {
m.relayClient.SetOnDisconnectListener(m.onServerDisconnected)
}
func (m *Manager) isForeignServer(address string) (bool, error) {
rAddr, err := m.relayClient.ServerInstanceURL()
func (m *Manager) isForeignServer(relayClient *Client, address string) (bool, error) {
rAddr, err := relayClient.ServerInstanceURL()
if err != nil {
return false, fmt.Errorf("relay client not connected")
}
@@ -475,50 +346,11 @@ func (m *Manager) startCleanupLoop() {
case <-m.ctx.Done():
return
case <-ticker.C:
m.cleanUpUnusedRelays()
m.foreign.cleanupUnused()
}
}
}
func (m *Manager) cleanUpUnusedRelays() {
m.relayClientsMutex.Lock()
defer m.relayClientsMutex.Unlock()
for addr, rt := range m.relayClients {
rt.Lock()
// if the connection failed to the server the relay client will be nil
// but the instance will be kept in the relayClients until the next locking
if rt.err != nil {
rt.Unlock()
continue
}
// dial still in progress (openConnVia publishes the track before Connect
// completes and no longer holds rt.Lock during it), nothing to clean up.
if rt.relayClient == nil {
rt.Unlock()
continue
}
if time.Since(rt.created) <= m.keepUnusedServerTime {
rt.Unlock()
continue
}
if rt.relayClient.HasConns() {
rt.Unlock()
continue
}
rt.relayClient.SetOnDisconnectListener(nil)
go func() {
_ = rt.relayClient.Close()
}()
log.Debugf("clean up unused relay server connection: %s", addr)
delete(m.relayClients, addr)
rt.Unlock()
}
}
func (m *Manager) addListener(serverAddress string, onClosedListener OnServerCloseListener) {
m.listenerLock.Lock()
defer m.listenerLock.Unlock()

View File

@@ -2,17 +2,14 @@ package client
import (
"context"
"net/netip"
"testing"
"time"
"github.com/stretchr/testify/require"
)
// TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial drives a real, hanging foreign
// relay dial and asserts cleanUpUnusedRelays does not stall behind it.
// relay dial and asserts the foreign store cleanup does not stall behind it.
func TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial(t *testing.T) {
serverAddr := stallingRelayListener(t)
serverAddr, accepted := stallingRelayListener(t)
mCtx, mCancel := context.WithCancel(context.Background())
t.Cleanup(mCancel)
@@ -22,39 +19,32 @@ func TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial(t *testing.T) {
dialDone := make(chan struct{})
go func() {
defer close(dialDone)
_, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{})
_, _ = m.foreign.OpenConn(mCtx, "peerKey", RelayServer{Addr: serverAddr})
}()
// The track appears in the map once the dial is in flight.
require.Eventually(t, func() bool {
m.relayClientsMutex.RLock()
defer m.relayClientsMutex.RUnlock()
_, ok := m.relayClients[serverAddr]
return ok
}, 5*time.Second, 5*time.Millisecond, "relay dial did not start")
select {
case <-accepted:
case <-time.After(5 * time.Second):
t.Fatal("relay dial did not reach the listener")
}
cleanupDone := make(chan struct{})
go func() {
defer close(cleanupDone)
m.cleanUpUnusedRelays()
m.foreign.cleanupUnused()
}()
select {
case <-cleanupDone:
case <-time.After(2 * time.Second):
t.Fatal("cleanUpUnusedRelays blocked on an in-progress relay dial while holding the relay map lock")
t.Fatal("cleanupUnused blocked on an in-progress relay dial")
}
m.relayClientsMutex.RLock()
_, stillTracked := m.relayClients[serverAddr]
m.relayClientsMutex.RUnlock()
require.True(t, stillTracked, "an in-progress relay dial must not be evicted by cleanup")
// Release the hanging dial so the goroutine can exit cleanly.
mCancel()
select {
case <-dialDone:
case <-time.After(5 * time.Second):
t.Fatal("openConnVia did not return after context cancellation")
t.Fatal("foreign OpenConn did not return after context cancellation")
}
}

View File

@@ -3,7 +3,6 @@ package client
import (
"context"
"net"
"net/netip"
"sync"
"testing"
"time"
@@ -13,13 +12,16 @@ import (
// stallingRelayListener accepts TCP connections and holds them open without ever
// responding, so a relay handshake dialed against it blocks until its context is
// cancelled. It returns the "rel://host:port" URL to dial.
func stallingRelayListener(t *testing.T) string {
// cancelled. accepted is signalled once per incoming connection so a caller can
// wait until a dial has actually reached the listener. It returns the
// "rel://host:port" URL to dial.
func stallingRelayListener(t *testing.T) (string, <-chan struct{}) {
t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
accepted := make(chan struct{}, 1)
var mu sync.Mutex
var conns []net.Conn
go func() {
@@ -31,6 +33,10 @@ func stallingRelayListener(t *testing.T) string {
mu.Lock()
conns = append(conns, c)
mu.Unlock()
select {
case accepted <- struct{}{}:
default:
}
}
}()
t.Cleanup(func() {
@@ -42,14 +48,14 @@ func stallingRelayListener(t *testing.T) string {
mu.Unlock()
})
return "rel://" + ln.Addr().String()
return "rel://" + ln.Addr().String(), accepted
}
// TestRelayStates_DoesNotBlockOnRealHangingDial is a regression test for
// RelayStates() called by a "status -d command" hanging behind an in-progress
// relay dial.
// foreign relay dial.
func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
serverAddr := stallingRelayListener(t)
serverAddr, accepted := stallingRelayListener(t)
mCtx, mCancel := context.WithCancel(context.Background())
t.Cleanup(mCancel)
@@ -59,15 +65,14 @@ func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
dialDone := make(chan struct{})
go func() {
defer close(dialDone)
_, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{})
_, _ = m.foreign.OpenConn(mCtx, "peerKey", RelayServer{Addr: serverAddr})
}()
require.Eventually(t, func() bool {
m.relayClientsMutex.RLock()
defer m.relayClientsMutex.RUnlock()
_, ok := m.relayClients[serverAddr]
return ok
}, 5*time.Second, 5*time.Millisecond, "relay dial did not start")
select {
case <-accepted:
case <-time.After(5 * time.Second):
t.Fatal("relay dial did not reach the listener")
}
done := make(chan []RelayConnState, 1)
go func() {
@@ -86,6 +91,6 @@ func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
select {
case <-dialDone:
case <-time.After(5 * time.Second):
t.Fatal("openConnVia did not return after context cancellation")
t.Fatal("foreign OpenConn did not return after context cancellation")
}
}

View File

@@ -3,7 +3,6 @@ package client
import (
"context"
"io"
"net/netip"
"testing"
"time"
@@ -85,7 +84,7 @@ func TestManager_ForeignRelayServerIP(t *testing.T) {
t.Run("no server IP, dial fails", func(t *testing.T) {
dialCtx, dialCancel := context.WithTimeout(ctx, 5*time.Second)
defer dialCancel()
_, err := mgrAlice.OpenConn(dialCtx, brokenFQDN, "bob", netip.Addr{})
_, err := mgrAlice.OpenConn(dialCtx, RelayServer{Addr: brokenFQDN}, "bob", true)
if err == nil {
t.Fatalf("expected OpenConn to fail without server IP, got success")
}
@@ -95,7 +94,7 @@ func TestManager_ForeignRelayServerIP(t *testing.T) {
// Bob waits for Alice's incoming peer connection on his side.
bobSideCh := make(chan error, 1)
go func() {
conn, err := mgrBob.OpenConn(ctx, bobRealAddr, "alice", netip.Addr{})
conn, err := mgrBob.OpenConn(ctx, RelayServer{Addr: bobRealAddr}, "alice", false)
if err != nil {
bobSideCh <- err
return
@@ -113,7 +112,7 @@ func TestManager_ForeignRelayServerIP(t *testing.T) {
bobSideCh <- nil
}()
aliceConn, err := mgrAlice.OpenConn(ctx, brokenFQDN, "bob", bobAdvertisedIP)
aliceConn, err := mgrAlice.OpenConn(ctx, RelayServer{Addr: brokenFQDN, IP: bobAdvertisedIP}, "bob", true)
if err != nil {
t.Fatalf("alice OpenConn with server IP: %s", err)
}

View File

@@ -3,7 +3,6 @@ package client
import (
"context"
"fmt"
"net/netip"
"testing"
"time"
@@ -106,11 +105,11 @@ func TestForeignConn(t *testing.T) {
if err != nil {
t.Fatalf("failed to get relay address: %s", err)
}
connAliceToBob, err := clientAlice.OpenConn(ctx, bobsSrvAddr, "bob", netip.Addr{})
connAliceToBob, err := clientAlice.OpenConn(ctx, RelayServer{Addr: bobsSrvAddr}, "bob", true)
if err != nil {
t.Fatalf("failed to bind channel: %s", err)
}
connBobToAlice, err := clientBob.OpenConn(ctx, bobsSrvAddr, "alice", netip.Addr{})
connBobToAlice, err := clientBob.OpenConn(ctx, RelayServer{Addr: bobsSrvAddr}, "alice", false)
if err != nil {
t.Fatalf("failed to bind channel: %s", err)
}
@@ -210,7 +209,7 @@ func TestForeginConnClose(t *testing.T) {
if err != nil {
t.Fatalf("failed to serve manager: %s", err)
}
conn, err := mgr.OpenConn(ctx, toURL(srvCfg2)[0], "bob", netip.Addr{})
conn, err := mgr.OpenConn(ctx, RelayServer{Addr: toURL(srvCfg2)[0]}, "bob", true)
if err != nil {
t.Fatalf("failed to bind channel: %s", err)
}
@@ -302,7 +301,7 @@ func TestForeignAutoClose(t *testing.T) {
}
t.Log("open connection to another peer")
if _, err = mgr.OpenConn(ctx, foreignServerURL, "anotherpeer", netip.Addr{}); err == nil {
if _, err = mgr.OpenConn(ctx, RelayServer{Addr: foreignServerURL}, "anotherpeer", true); err == nil {
t.Fatalf("should have failed to open connection to another peer")
}
@@ -372,7 +371,7 @@ func TestAutoReconnect(t *testing.T) {
if err != nil {
t.Errorf("failed to get relay address: %s", err)
}
conn, err := clientAlice.OpenConn(ctx, ra, "bob", netip.Addr{})
conn, err := clientAlice.OpenConn(ctx, RelayServer{Addr: ra}, "bob", false)
if err != nil {
t.Errorf("failed to bind channel: %s", err)
}
@@ -392,7 +391,7 @@ func TestAutoReconnect(t *testing.T) {
}
log.Infof("reopent the connection")
_, err = clientAlice.OpenConn(ctx, ra, "bob", netip.Addr{})
_, err = clientAlice.OpenConn(ctx, RelayServer{Addr: ra}, "bob", false)
if err != nil {
t.Errorf("failed to open channel: %s", err)
}
@@ -454,7 +453,7 @@ func TestNotifierDoubleAdd(t *testing.T) {
t.Fatalf("failed to serve manager: %s", err)
}
conn1, err := clientAlice.OpenConn(ctx, clientAlice.ServerURLs()[0], "bob", netip.Addr{})
conn1, err := clientAlice.OpenConn(ctx, RelayServer{Addr: clientAlice.ServerURLs()[0]}, "bob", false)
if err != nil {
t.Fatalf("failed to bind channel: %s", err)
}