Inproxy enhancements

- New personal pairing mode broker rendezvous algorithm

- Custom client offer timeout for personal pairing

- For common pairing mode, potentially invoke broker fail over when no match
  is found

- Allow overriding long-polling request timeouts with fronting provider
  limits

- Log irregular_tunnel in broker deobfuscation cases

- Additional port mapper logging; truncate verbose log

- Fix has_private_IP metric incorrectly reporting true

psiphon/common/inproxy/broker.go

@@ -85,9 +85,11 @@ type Broker struct {
 	commonCompartmentsMutex sync.Mutex
 	commonCompartments      *consistent.Consistent
 
-	proxyAnnounceTimeout    int64
-	clientOfferTimeout      int64
-	pendingServerReportsTTL int64
+	proxyAnnounceTimeout       int64
+	clientOfferTimeout         int64
+	clientOfferPersonalTimeout int64
+	pendingServerReportsTTL    int64
+	maxRequestTimeouts         atomic.Value
 
 	maxCompartmentIDs int64
 }
@@ -162,9 +164,10 @@ type BrokerConfig struct {
 	ServerEntrySignaturePublicKey string
 
 	// These timeout parameters may be used to override defaults.
-	ProxyAnnounceTimeout    time.Duration
-	ClientOfferTimeout      time.Duration
-	PendingServerReportsTTL time.Duration
+	ProxyAnnounceTimeout       time.Duration
+	ClientOfferTimeout         time.Duration
+	ClientOfferPersonalTimeout time.Duration
+	PendingServerReportsTTL    time.Duration
 
 	// Announcement queue limit configuration.
 	MatcherAnnouncementLimitEntryCount    int
@@ -219,9 +222,10 @@ func NewBroker(config *BrokerConfig) (*Broker, error) {
 			OfferRateLimitInterval:         config.MatcherOfferRateLimitInterval,
 		}),
 
-		proxyAnnounceTimeout:    int64(config.ProxyAnnounceTimeout),
-		clientOfferTimeout:      int64(config.ClientOfferTimeout),
-		pendingServerReportsTTL: int64(config.PendingServerReportsTTL),
+		proxyAnnounceTimeout:       int64(config.ProxyAnnounceTimeout),
+		clientOfferTimeout:         int64(config.ClientOfferTimeout),
+		clientOfferPersonalTimeout: int64(config.ClientOfferPersonalTimeout),
+		pendingServerReportsTTL:    int64(config.PendingServerReportsTTL),
 
 		maxCompartmentIDs: int64(common.ValueOrDefault(config.MaxCompartmentIDs, MaxCompartmentIDs)),
 	}
@@ -273,11 +277,15 @@ func (b *Broker) SetCommonCompartmentIDs(commonCompartmentIDs []ID) error {
 func (b *Broker) SetTimeouts(
 	proxyAnnounceTimeout time.Duration,
 	clientOfferTimeout time.Duration,
-	pendingServerReportsTTL time.Duration) {
+	clientOfferPersonalTimeout time.Duration,
+	pendingServerReportsTTL time.Duration,
+	maxRequestTimeouts map[string]time.Duration) {
 
 	atomic.StoreInt64(&b.proxyAnnounceTimeout, int64(proxyAnnounceTimeout))
 	atomic.StoreInt64(&b.clientOfferTimeout, int64(clientOfferTimeout))
+	atomic.StoreInt64(&b.clientOfferPersonalTimeout, int64(clientOfferPersonalTimeout))
 	atomic.StoreInt64(&b.pendingServerReportsTTL, int64(pendingServerReportsTTL))
+	b.maxRequestTimeouts.Store(maxRequestTimeouts)
 }
 
 // SetLimits sets new queue limit values, replacing the previous
@@ -413,7 +421,7 @@ func (b *Broker) HandleSessionPacket(
 
 	// HandlePacket returns both a packet and an error in the expired session
 	// reset token case. Log the error here, clear it, and return the
-	// packetto be relayed back to the broker client.
+	// packet to be relayed back to the broker client.
 
 	outPacket, err := b.responderSessions.HandlePacket(
 		inPacket, handleUnwrappedRequest)
@@ -599,6 +607,11 @@ func (b *Broker) handleProxyAnnounce(
 	timeout := common.ValueOrDefault(
 		time.Duration(atomic.LoadInt64(&b.proxyAnnounceTimeout)),
 		brokerProxyAnnounceTimeout)
+
+	// Adjust the timeout to respect any shorter maximum request timeouts for
+	// the fronting provider.
+	timeout = b.adjustRequestTimeout(logFields, timeout)
+
 	announceCtx, cancelFunc := context.WithTimeout(ctx, timeout)
 	defer cancelFunc()
 	extendTransportTimeout(timeout)
@@ -852,9 +865,21 @@ func (b *Broker) handleClientOffer(
 	// Enqueue the client offer and await a proxy matching and subsequent
 	// proxy answer.
 
-	timeout := common.ValueOrDefault(
-		time.Duration(atomic.LoadInt64(&b.clientOfferTimeout)),
-		brokerClientOfferTimeout)
+	// The Client Offer timeout may be configured with a shorter value in
+	// personal pairing mode, to facilitate a faster no-match result and
+	// resulting broker rotation.
+	var timeout time.Duration
+	if len(offerRequest.PersonalCompartmentIDs) > 0 {
+		timeout = time.Duration(atomic.LoadInt64(&b.clientOfferPersonalTimeout))
+	} else {
+		timeout = time.Duration(atomic.LoadInt64(&b.clientOfferTimeout))
+	}
+	timeout = common.ValueOrDefault(timeout, brokerClientOfferTimeout)
+
+	// Adjust the timeout to respect any shorter maximum request timeouts for
+	// the fronting provider.
+	timeout = b.adjustRequestTimeout(logFields, timeout)
+
 	offerCtx, cancelFunc := context.WithTimeout(ctx, timeout)
 	defer cancelFunc()
 	extendTransportTimeout(timeout)
@@ -1242,6 +1267,33 @@ func (b *Broker) handleClientRelayedPacket(
 	return responsePayload, nil
 }
 
+func (b *Broker) adjustRequestTimeout(
+	logFields common.LogFields, timeout time.Duration) time.Duration {
+
+	// Adjust long-polling request timeouts to respect any maximum request
+	// timeout supported by the provider fronting the request.
+	//
+	// Limitation: the client is trusted to provide the correct fronting
+	// provider ID.
+
+	maxRequestTimeouts, ok := b.maxRequestTimeouts.Load().(map[string]time.Duration)
+	if !ok || maxRequestTimeouts == nil {
+		return timeout
+	}
+
+	frontingProviderID, ok := logFields["fronting_provider_id"].(string)
+	if !ok {
+		return timeout
+	}
+
+	maxRequestTimeout, ok := maxRequestTimeouts[frontingProviderID]
+	if !ok || maxRequestTimeout <= 0 || timeout <= maxRequestTimeout {
+		return timeout
+	}
+
+	return maxRequestTimeout
+}
+
 type pendingServerReport struct {
 	serverID     string
 	serverReport *BrokerServerReport

psiphon/common/inproxy/brokerClient.go

@@ -115,7 +115,7 @@ func (b *BrokerClient) ProxyAnnounce(
 		b.coordinator.AnnounceRequestTimeout(),
 		proxyAnnounceRequestTimeout)
 
-	responsePayload, err := b.roundTrip(
+	responsePayload, _, err := b.roundTrip(
 		ctx, requestDelay, requestTimeout, requestPayload)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -132,18 +132,26 @@ func (b *BrokerClient) ProxyAnnounce(
 // ClientOffer sends a ClientOffer request and returns the response.
 func (b *BrokerClient) ClientOffer(
 	ctx context.Context,
-	request *ClientOfferRequest) (*ClientOfferResponse, error) {
+	request *ClientOfferRequest,
+	hasPersonalCompartmentIDs bool) (*ClientOfferResponse, error) {
 
 	requestPayload, err := MarshalClientOfferRequest(request)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
 
+	var offerRequestTimeout time.Duration
+	if hasPersonalCompartmentIDs {
+		offerRequestTimeout = b.coordinator.OfferRequestPersonalTimeout()
+	} else {
+		offerRequestTimeout = b.coordinator.OfferRequestTimeout()
+	}
+
 	requestTimeout := common.ValueOrDefault(
-		b.coordinator.OfferRequestTimeout(),
+		offerRequestTimeout,
 		clientOfferRequestTimeout)
 
-	responsePayload, err := b.roundTrip(
+	responsePayload, roundTripper, err := b.roundTrip(
 		ctx, 0, requestTimeout, requestPayload)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -154,6 +162,15 @@ func (b *BrokerClient) ClientOffer(
 		return nil, errors.Trace(err)
 	}
 
+	if response.NoMatch {
+
+		// Signal the no match event, which may trigger broker rotation. As
+		// with BrokerClientRoundTripperSucceeded/Failed callbacks, the
+		// RoundTripper used is passed in to ensure the correct broker client
+		// is reset.
+		b.coordinator.BrokerClientNoMatch(roundTripper)
+	}
+
 	return response, nil
 }
 
@@ -171,7 +188,7 @@ func (b *BrokerClient) ProxyAnswer(
 		b.coordinator.AnswerRequestTimeout(),
 		proxyAnswerRequestTimeout)
 
-	responsePayload, err := b.roundTrip(
+	responsePayload, _, err := b.roundTrip(
 		ctx, 0, requestTimeout, requestPayload)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -200,7 +217,7 @@ func (b *BrokerClient) ClientRelayedPacket(
 		b.coordinator.RelayedPacketRequestTimeout(),
 		clientRelayedPacketRequestTimeout)
 
-	responsePayload, err := b.roundTrip(
+	responsePayload, _, err := b.roundTrip(
 		ctx, 0, requestTimeout, requestPayload)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -218,14 +235,14 @@ func (b *BrokerClient) roundTrip(
 	ctx context.Context,
 	requestDelay time.Duration,
 	requestTimeout time.Duration,
-	request []byte) ([]byte, error) {
+	request []byte) ([]byte, RoundTripper, error) {
 
 	// The round tripper may need to establish a transport-level connection;
 	// or this may already be established.
 
 	roundTripper, err := b.coordinator.BrokerClientRoundTripper()
 	if err != nil {
-		return nil, errors.Trace(err)
+		return nil, nil, errors.Trace(err)
 	}
 
 	// InitiatorSessions.RoundTrip may make serveral round trips with
@@ -291,10 +308,10 @@ func (b *BrokerClient) roundTrip(
 			b.coordinator.BrokerClientRoundTripperFailed(roundTripper)
 		}
 
-		return nil, errors.Trace(err)
+		return nil, nil, errors.Trace(err)
 	}
 
 	b.coordinator.BrokerClientRoundTripperSucceeded(roundTripper)
 
-	return response, nil
+	return response, roundTripper, nil
 }

psiphon/common/inproxy/client.go

@@ -370,8 +370,8 @@ func dialClientWebRTCConn(
 	// WebRTCDialCoordinator.PortMappingTypes may be populated via
 	// newWebRTCConnWithOffer.
 
-	// ClientOffer applies BrokerDialCoordinator.OfferRequestTimeout as the
-	// request timeout.
+	// ClientOffer applies BrokerDialCoordinator.OfferRequestTimeout or
+	// OfferRequestPersonalTimeout as the request timeout.
 	offerResponse, err := config.BrokerClient.ClientOffer(
 		ctx,
 		&ClientOfferRequest{
@@ -391,7 +391,8 @@ func dialClientWebRTCConn(
 			PackedDestinationServerEntry: config.PackedDestinationServerEntry,
 			NetworkProtocol:              config.DialNetworkProtocol,
 			DestinationAddress:           config.DialAddress,
-		})
+		},
+		hasPersonalCompartmentIDs)
 	if err != nil {
 		return nil, false, errors.Trace(err)
 	}

psiphon/common/inproxy/coordinator.go

@@ -162,12 +162,21 @@ type BrokerDialCoordinator interface {
 	// after closing its network resources.
 	BrokerClientRoundTripperFailed(roundTripper RoundTripper)
 
+	// BrokerClientNoMatch is called after a Client Offer fails due to no
+	// match. This signal may be used to rotate to a new broker in order to
+	// find a match. In personal pairing mode, clients should rotate on no
+	// match, as the corresponding proxy may be announcing only on another
+	// broker. In common pairing mode, clients may rotate, in case common
+	// proxies are not well balanced across brokers.
+	BrokerClientNoMatch(roundTripper RoundTripper)
+
 	SessionHandshakeRoundTripTimeout() time.Duration
 	AnnounceRequestTimeout() time.Duration
 	AnnounceDelay() time.Duration
 	AnnounceDelayJitter() float64
 	AnswerRequestTimeout() time.Duration
 	OfferRequestTimeout() time.Duration
+	OfferRequestPersonalTimeout() time.Duration
 	OfferRetryDelay() time.Duration
 	OfferRetryJitter() float64
 	RelayedPacketRequestTimeout() time.Duration

psiphon/common/inproxy/coordinator_test.go

@@ -45,12 +45,14 @@ type testBrokerDialCoordinator struct {
 	brokerClientRoundTripper          RoundTripper
 	brokerClientRoundTripperSucceeded func(RoundTripper)
 	brokerClientRoundTripperFailed    func(RoundTripper)
+	brokerClientNoMatch               func(RoundTripper)
 	sessionHandshakeRoundTripTimeout  time.Duration
 	announceRequestTimeout            time.Duration
 	announceDelay                     time.Duration
 	announceDelayJitter               float64
 	answerRequestTimeout              time.Duration
 	offerRequestTimeout               time.Duration
+	offerRequestPersonalTimeout       time.Duration
 	offerRetryDelay                   time.Duration
 	offerRetryJitter                  float64
 	relayedPacketRequestTimeout       time.Duration
@@ -116,6 +118,12 @@ func (t *testBrokerDialCoordinator) BrokerClientRoundTripperFailed(roundTripper
 	t.brokerClientRoundTripperFailed(roundTripper)
 }
 
+func (t *testBrokerDialCoordinator) BrokerClientNoMatch(roundTripper RoundTripper) {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+	t.brokerClientNoMatch(roundTripper)
+}
+
 func (t *testBrokerDialCoordinator) SessionHandshakeRoundTripTimeout() time.Duration {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
@@ -152,6 +160,12 @@ func (t *testBrokerDialCoordinator) OfferRequestTimeout() time.Duration {
 	return t.offerRequestTimeout
 }
 
+func (t *testBrokerDialCoordinator) OfferRequestPersonalTimeout() time.Duration {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+	return t.offerRequestPersonalTimeout
+}
+
 func (t *testBrokerDialCoordinator) OfferRetryDelay() time.Duration {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()

psiphon/common/inproxy/inproxy_test.go

@@ -102,6 +102,7 @@ func runTestInproxy(doMustUpgrade bool) error {
 	roundTripperSucceded := func(RoundTripper) { atomic.AddInt32(&roundTripperSucceededCount, 1) }
 	roundTripperFailedCount := int32(0)
 	roundTripperFailed := func(RoundTripper) { atomic.AddInt32(&roundTripperFailedCount, 1) }
+	noMatch := func(RoundTripper) {}
 
 	var receivedProxyMustUpgrade chan struct{}
 	var receivedClientMustUpgrade chan struct{}
@@ -667,6 +668,7 @@ func runTestInproxy(doMustUpgrade bool) error {
 				brokerListener.Addr().String(), "client"),
 			brokerClientRoundTripperSucceeded: roundTripperSucceded,
 			brokerClientRoundTripperFailed:    roundTripperFailed,
+			brokerClientNoMatch:               noMatch,
 		}
 
 		webRTCCoordinator := &testWebRTCDialCoordinator{

psiphon/common/inproxy/obfuscation.go

@@ -206,6 +206,27 @@ func obfuscateSessionPacket(
 	return obfuscatedPacket, nil
 }
 
+// DeobfuscationAnomoly is an error type that is returned when an anomalous
+// condition is encountered while deobfuscating a session packet. This may
+// include malformed packets; packets obfuscated without knowledge of the
+// correct obfuscation secret; replay of valid packets; etc.
+//
+// On the server side, Broker.HandleSessionPacket already specifies that
+// anti-probing mechanisms should be applied on any error return; the
+// DeobfuscationAnomoly error type enables further error filtering before
+// logging an irregular tunnel event.
+type DeobfuscationAnomoly struct {
+	err error
+}
+
+func NewDeobfuscationAnomoly(err error) *DeobfuscationAnomoly {
+	return &DeobfuscationAnomoly{err: err}
+}
+
+func (e DeobfuscationAnomoly) Error() string {
+	return e.err.Error()
+}
+
 // deobfuscateSessionPacket deobfuscates a session packet obfuscated with
 // obfuscateSessionPacket and the same deobfuscateSessionPacket.
 //
@@ -234,14 +255,16 @@ func deobfuscateSessionPacket(
 
 	if len(obfuscatedPacket) < obfuscationSessionPacketNonceSize {
 		imitateDeobfuscateSessionPacketDuration(replayHistory)
-		return nil, errors.TraceNew("invalid nonce")
+		return nil, NewDeobfuscationAnomoly(
+			errors.TraceNew("invalid nonce"))
 	}
 
 	nonce := obfuscatedPacket[:obfuscationSessionPacketNonceSize]
 
 	if replayHistory != nil && replayHistory.Lookup(nonce) {
 		imitateDeobfuscateSessionPacketDuration(nil)
-		return nil, errors.TraceNew("replayed nonce")
+		return nil, NewDeobfuscationAnomoly(
+			errors.TraceNew("replayed nonce"))
 	}
 
 	// As an AEAD, AES-GCM authenticates that the sender used the expected
@@ -272,17 +295,20 @@ func deobfuscateSessionPacket(
 	if replayHistory != nil {
 		timestamp, n = binary.Varint(plaintext[offset:])
 		if timestamp == 0 && n <= 0 {
-			return nil, errors.TraceNew("invalid timestamp")
+			return nil, NewDeobfuscationAnomoly(
+				errors.TraceNew("invalid timestamp"))
 		}
 		offset += n
 	}
 	paddingSize, n := binary.Uvarint(plaintext[offset:])
 	if n < 1 {
-		return nil, errors.TraceNew("invalid padding size")
+		return nil, NewDeobfuscationAnomoly(
+			errors.TraceNew("invalid padding size"))
 	}
 	offset += n
 	if len(plaintext[offset:]) < int(paddingSize) {
-		return nil, errors.TraceNew("invalid padding")
+		return nil, NewDeobfuscationAnomoly(
+			errors.TraceNew("invalid padding"))
 	}
 	offset += int(paddingSize)
 
@@ -294,10 +320,12 @@ func deobfuscateSessionPacket(
 
 		now := time.Now().Unix()
 		if timestamp+antiReplayTimeFactorPeriodSeconds/2 < now {
-			return nil, errors.TraceNew("timestamp behind")
+			return nil, NewDeobfuscationAnomoly(
+				errors.TraceNew("timestamp behind"))
 		}
 		if timestamp-antiReplayTimeFactorPeriodSeconds/2 > now {
-			return nil, errors.TraceNew("timestamp ahead")
+			return nil, NewDeobfuscationAnomoly(
+				errors.TraceNew("timestamp ahead"))
 		}
 
 		// Now that it's validated, add this packet to the replay history. The

psiphon/common/inproxy/portmapper.go

@@ -24,6 +24,7 @@ package inproxy
 import (
 	"context"
 	"fmt"
+	"strings"
 	"sync"
 
 	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common"
@@ -62,6 +63,7 @@ type portMapper struct {
 	havePortMappingOnce sync.Once
 	portMappingAddress  chan string
 	client              *portmapper.Client
+	portMappingLogger   func(format string, args ...any)
 }
 
 // newPortMapper initializes a new port mapper, configured to map to the
@@ -72,11 +74,13 @@ func newPortMapper(
 	localPort int) *portMapper {
 
 	portMappingLogger := func(format string, args ...any) {
-		logger.WithTrace().Info("port mapping: " + fmt.Sprintf(format, args))
+		logger.WithTrace().Info(
+			"port mapping: " + formatPortMappingLog(format, args...))
 	}
 
 	p := &portMapper{
 		portMappingAddress: make(chan string, 1),
+		portMappingLogger:  portMappingLogger,
 	}
 
 	// This code assumes assumes tailscale NewClient call does only
@@ -94,6 +98,7 @@ func newPortMapper(
 			if ok {
 				// With sync.Once and a buffer size of 1, this send won't block.
 				p.portMappingAddress <- address.String()
+				portMappingLogger("address obtained")
 			} else {
 
 				// This is not an expected case; there should be a port
@@ -116,6 +121,7 @@ func newPortMapper(
 
 // start initiates the port mapping attempt.
 func (p *portMapper) start() {
+	p.portMappingLogger("started")
 	_, _ = p.client.GetCachedMappingOrStartCreatingOne()
 }
 
@@ -127,7 +133,18 @@ func (p *portMapper) portMappingExternalAddress() <-chan string {
 
 // close releases the port mapping
 func (p *portMapper) close() error {
-	return errors.Trace(p.client.Close())
+	err := p.client.Close()
+	p.portMappingLogger("closed")
+	return errors.Trace(err)
+}
+
+func formatPortMappingLog(format string, args ...any) string {
+	truncatePrefix := "[v1] UPnP reply"
+	if strings.HasPrefix(format, truncatePrefix) {
+		// Omit packet portion of this log, but still log the event
+		return truncatePrefix
+	}
+	return fmt.Sprintf(format, args...)
 }
 
 // probePortMapping discovers and reports which port mapping protocols are
@@ -144,7 +161,8 @@ func probePortMapping(
 	logger common.Logger) (PortMappingTypes, error) {
 
 	portMappingLogger := func(format string, args ...any) {
-		logger.WithTrace().Info("port mapping probe: " + fmt.Sprintf(format, args))
+		logger.WithTrace().Info(
+			"port mapping probe: " + formatPortMappingLog(format, args...))
 	}
 
 	client := portmapper.NewClient(portMappingLogger, nil, nil, nil)

psiphon/common/inproxy/webrtc.go

@@ -1687,7 +1687,7 @@ func processSDPAddresses(
 
 	candidateTypes := map[ICECandidateType]bool{}
 	hasIPv6 := false
-	hasPrivateIP := true
+	hasPrivateIP := false
 	filteredCandidateReasons := make(map[string]int)
 
 	var portMappingICECandidates []sdp.Attribute

psiphon/common/parameters/keyStrings.go

@@ -1,37 +0,0 @@
-/*
- * Copyright (c) 2023, Psiphon Inc.
- * All rights reserved.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- */
-
-package parameters
-
-// KeyStrings represents a set of key/strings pairs.
-type KeyStrings map[string][]string
-
-// Validates that the keys and values are well formed.
-func (keyStrings KeyStrings) Validate() error {
-	// Always succeeds because KeyStrings is generic and does not impose any
-	// restrictions on keys/values. Consider imposing limits like maximum
-	// map/array/string sizes.
-	return nil
-}
-
-func (p ParametersAccessor) KeyStrings(name, key string) []string {
-	value := KeyStrings{}
-	p.snapshot.getValue(name, &value)
-	return value[key]
-}

psiphon/common/parameters/keyValues.go

@@ -21,6 +21,7 @@ package parameters
 
 import (
 	"encoding/json"
+	"time"
 
 	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/errors"
 )
@@ -39,3 +40,28 @@ func (keyValues KeyValues) Validate() error {
 	}
 	return nil
 }
+
+// KeyStrings represents a set of key/strings pairs.
+type KeyStrings map[string][]string
+
+// Validates that the keys and values are well formed.
+func (keyStrings KeyStrings) Validate() error {
+	// Always succeeds because KeyStrings is generic and does not impose any
+	// restrictions on keys/values. Consider imposing limits like maximum
+	// map/array/string sizes.
+	return nil
+}
+
+// KeyDurations represents a set of key/duration pairs.
+type KeyDurations map[string]string
+
+// Validates that the keys and durations are well formed.
+func (keyDurations KeyDurations) Validate() error {
+	for _, value := range keyDurations {
+		_, err := time.ParseDuration(value)
+		if err != nil {
+			return errors.Trace(err)
+		}
+	}
+	return nil
+}

psiphon/common/parameters/parameters.go

@@ -399,6 +399,7 @@ const (
 	InproxyBrokerMatcherOfferRateLimitInterval         = "InproxyBrokerMatcherOfferRateLimitInterval"
 	InproxyBrokerProxyAnnounceTimeout                  = "InproxyBrokerProxyAnnounceTimeout"
 	InproxyBrokerClientOfferTimeout                    = "InproxyBrokerClientOfferTimeout"
+	InproxyBrokerClientOfferPersonalTimeout            = "InproxyBrokerClientOfferPersonalTimeout"
 	InproxyBrokerPendingServerRequestsTTL              = "InproxyBrokerPendingServerRequestsTTL"
 	InproxySessionHandshakeRoundTripTimeout            = "InproxySessionHandshakeRoundTripTimeout"
 	InproxyProxyAnnounceRequestTimeout                 = "InproxyProxyAnnounceRequestTimeout"
@@ -406,6 +407,7 @@ const (
 	InproxyProxyAnnounceDelayJitter                    = "InproxyProxyAnnounceDelayJitter"
 	InproxyProxyAnswerRequestTimeout                   = "InproxyProxyAnswerRequestTimeout"
 	InproxyClientOfferRequestTimeout                   = "InproxyClientOfferRequestTimeout"
+	InproxyClientOfferRequestPersonalTimeout           = "InproxyClientOfferRequestPersonalTimeout"
 	InproxyClientOfferRetryDelay                       = "InproxyClientOfferRetryDelay"
 	InproxyClientOfferRetryJitter                      = "InproxyClientOfferRetryJitter"
 	InproxyClientRelayedPacketRequestTimeout           = "InproxyCloientRelayedPacketRequestTimeout"
@@ -444,6 +446,10 @@ const (
 	InproxyPersonalPairingConnectionWorkerPoolSize     = "InproxyPersonalPairingConnectionWorkerPoolSize"
 	InproxyClientDialRateLimitQuantity                 = "InproxyClientDialRateLimitQuantity"
 	InproxyClientDialRateLimitInterval                 = "InproxyClientDialRateLimitInterval"
+	InproxyClientNoMatchFailoverProbability            = "InproxyClientNoMatchFailoverProbability"
+	InproxyClientNoMatchFailoverPersonalProbability    = "InproxyClientNoMatchFailoverPersonalProbability"
+	InproxyFrontingProviderClientMaxRequestTimeouts    = "InproxyFrontingProviderClientMaxRequestTimeouts"
+	InproxyFrontingProviderServerMaxRequestTimeouts    = "InproxyFrontingProviderServerMaxRequestTimeouts"
 
 	// Retired parameters
 
@@ -901,6 +907,7 @@ var defaultParameters = map[string]struct {
 	InproxyBrokerMatcherOfferRateLimitInterval:         {value: 1 * time.Minute, minimum: time.Duration(0), flags: serverSideOnly},
 	InproxyBrokerProxyAnnounceTimeout:                  {value: 2 * time.Minute, minimum: time.Duration(0), flags: serverSideOnly},
 	InproxyBrokerClientOfferTimeout:                    {value: 10 * time.Second, minimum: time.Duration(0), flags: serverSideOnly},
+	InproxyBrokerClientOfferPersonalTimeout:            {value: 5 * time.Second, minimum: time.Duration(0), flags: serverSideOnly},
 	InproxyBrokerPendingServerRequestsTTL:              {value: 60 * time.Second, minimum: time.Duration(0), flags: serverSideOnly},
 	InproxySessionHandshakeRoundTripTimeout:            {value: 10 * time.Second, minimum: time.Duration(0), flags: useNetworkLatencyMultiplier},
 	InproxyProxyAnnounceRequestTimeout:                 {value: 2*time.Minute + 10*time.Second, minimum: time.Duration(0)},
@@ -908,6 +915,7 @@ var defaultParameters = map[string]struct {
 	InproxyProxyAnnounceDelayJitter:                    {value: 0.5, minimum: 0.0},
 	InproxyProxyAnswerRequestTimeout:                   {value: 10*time.Second + 10*time.Second, minimum: time.Duration(0)},
 	InproxyClientOfferRequestTimeout:                   {value: 10*time.Second + 10*time.Second, minimum: time.Duration(0)},
+	InproxyClientOfferRequestPersonalTimeout:           {value: 5*time.Second + 10*time.Second, minimum: time.Duration(0)},
 	InproxyClientOfferRetryDelay:                       {value: 100 * time.Millisecond, minimum: time.Duration(0)},
 	InproxyClientOfferRetryJitter:                      {value: 0.5, minimum: 0.0},
 	InproxyClientRelayedPacketRequestTimeout:           {value: 10 * time.Second, minimum: time.Duration(0)},
@@ -946,6 +954,10 @@ var defaultParameters = map[string]struct {
 	InproxyPersonalPairingConnectionWorkerPoolSize:     {value: 2, minimum: 1},
 	InproxyClientDialRateLimitQuantity:                 {value: 10, minimum: 0},
 	InproxyClientDialRateLimitInterval:                 {value: 1 * time.Minute, minimum: time.Duration(0)},
+	InproxyClientNoMatchFailoverProbability:            {value: 0.5, minimum: 0.0},
+	InproxyClientNoMatchFailoverPersonalProbability:    {value: 1.0, minimum: 0.0},
+	InproxyFrontingProviderClientMaxRequestTimeouts:    {value: KeyDurations{}},
+	InproxyFrontingProviderServerMaxRequestTimeouts:    {value: KeyDurations{}, flags: serverSideOnly},
 }
 
 // IsServerSideOnly indicates if the parameter specified by name is used
@@ -1325,6 +1337,14 @@ func (p *Parameters) Set(
 					}
 					return nil, errors.Trace(err)
 				}
+			case KeyDurations:
+				err := v.Validate()
+				if err != nil {
+					if skipOnError {
+						continue
+					}
+					return nil, errors.Trace(err)
+				}
 			case *BPFProgramSpec:
 				if v != nil {
 					err := v.Validate()
@@ -1906,6 +1926,30 @@ func (p ParametersAccessor) KeyValues(name string) KeyValues {
 	return value
 }
 
+// KeyStrings returns a KeyStrings parameter value.
+func (p ParametersAccessor) KeyStrings(name, key string) []string {
+	value := KeyStrings{}
+	p.snapshot.getValue(name, &value)
+	return value[key]
+}
+
+// KeyDurations returns a KeyDurations parameter value, with string durations
+// converted to time.Duration.
+func (p ParametersAccessor) KeyDurations(name string) map[string]time.Duration {
+	value := KeyDurations{}
+	p.snapshot.getValue(name, &value)
+	durations := make(map[string]time.Duration)
+	for key, duration := range value {
+		d, err := time.ParseDuration(duration)
+		if err != nil {
+			// Skip invalid duration. Not expected with validation.
+			continue
+		}
+		durations[key] = d
+	}
+	return durations
+}
+
 // BPFProgram returns an assembled BPF program corresponding to a
 // BPFProgramSpec parameter value. Returns nil in the case of any empty
 // program.

psiphon/common/parameters/parameters_test.go

@@ -128,6 +128,23 @@ func TestGetDefaultParameters(t *testing.T) {
 			if !reflect.DeepEqual(v, g) {
 				t.Fatalf("KeyValues returned %+v expected %+v", g, v)
 			}
+		case KeyStrings:
+			for key, strings := range v {
+				g := p.Get().KeyStrings(name, key)
+				if !reflect.DeepEqual(strings, g) {
+					t.Fatalf("KeyStrings returned %+v expected %+v", g, strings)
+				}
+			}
+		case KeyDurations:
+			g := p.Get().KeyDurations(name)
+			durations := make(map[string]time.Duration)
+			for key, duration := range v {
+				d, _ := time.ParseDuration(duration)
+				durations[key] = d
+			}
+			if !reflect.DeepEqual(durations, g) {
+				t.Fatalf("KeyDurations returned %+v expected %+v", g, durations)
+			}
 		case *BPFProgramSpec:
 			ok, name, rawInstructions := p.Get().BPFProgram(name)
 			if v != nil || ok || name != "" || rawInstructions != nil {
@@ -189,13 +206,6 @@ func TestGetDefaultParameters(t *testing.T) {
 			if !reflect.DeepEqual(v, g) {
 				t.Fatalf("ConjureTransports returned %+v expected %+v", g, v)
 			}
-		case KeyStrings:
-			for key, strings := range v {
-				g := p.Get().KeyStrings(name, key)
-				if !reflect.DeepEqual(strings, g) {
-					t.Fatalf("KeyStrings returned %+v expected %+v", g, strings)
-				}
-			}
 		case InproxyBrokerSpecsValue:
 			g := p.Get().InproxyBrokerSpecs(name)
 			if !reflect.DeepEqual(v, g) {

psiphon/config.go

@@ -644,44 +644,22 @@ type Config struct {
 	// transfer rate limit for each proxied client. When 0, there is no limit.
 	InproxyLimitDownstreamBytesPerSecond int
 
-	// InproxyProxyPersonalCompartmentIDs specifies the personal compartment
-	// IDs used by an in-proxy proxy. Personal compartment IDs are
+	// InproxyProxyPersonalCompartmentID specifies the personal compartment
+	// ID used by an in-proxy proxy. Personal compartment IDs are
 	// distributed from proxy operators to client users out-of-band and
 	// provide a mechanism to allow only certain clients to use a proxy.
-	//
-	// Limitation: currently, at most 1 personal compartment may be specified.
-	// See InproxyClientPersonalCompartmentIDs comment for additional
-	// personal pairing limitations.
-	InproxyProxyPersonalCompartmentIDs []string
+	InproxyProxyPersonalCompartmentID string
 
-	// InproxyClientPersonalCompartmentIDs specifies the personal compartment
-	// IDs used by an in-proxy client. Personal compartment IDs are
+	// InproxyClientPersonalCompartmentID specifies the personal compartment
+	// ID used by an in-proxy client. Personal compartment IDs are
 	// distributed from proxy operators to client users out-of-band and
 	// provide a mechanism to ensure a client uses only a certain proxy for
 	// all tunnels connections.
 	//
-	// When InproxyClientPersonalCompartmentIDs is set, the client will use
+	// When an InproxyClientPersonalCompartmentID is set, the client will use
 	// only in-proxy protocols, ensuring that all connections go through the
-	// proxy or proxies with the same personal compartment IDs.
-	//
-	// Limitations:
-	//
-	// While fully functional, the personal pairing mode has a number of
-	// limitations that make the current implementation less suitable for
-	// large scale deployment.
-	//
-	// Since the mode requires an in-proxy connection to a proxy, announcing
-	// with the corresponding personal compartment ID, not only must that
-	// proxy be available, but also a broker, and both the client and proxy
-	// must rendezvous at the same broker.
-	//
-	// In personal mode, clients and proxies use a simplistic approach to
-	// rendezvous: always select the first broker spec. This works, but is
-	// not robust in terms of load balancing, and fails if the first broker
-	// is unreachable or overloaded. Non-personal in-proxy dials can simply
-	// use any available broker.
-	//
-	InproxyClientPersonalCompartmentIDs []string
+	// proxy or proxies with the same personal compartment ID.
+	InproxyClientPersonalCompartmentID string
 
 	// InproxyPersonalPairingConnectionWorkerPoolSize specifies the value for
 	// ConnectionWorkerPoolSize in personal pairing mode. If omitted or when
@@ -1013,6 +991,7 @@ type Config struct {
 	InproxyProxyAnnounceDelayJitter                        *float64
 	InproxyProxyAnswerRequestTimeoutMilliseconds           *int
 	InproxyClientOfferRequestTimeoutMilliseconds           *int
+	InproxyClientOfferRequestPersonalTimeoutMilliseconds   *int
 	InproxyClientOfferRetryDelayMilliseconds               *int
 	InproxyClientOfferRetryJitter                          *float64
 	InproxyClientRelayedPacketRequestTimeoutMilliseconds   *int
@@ -1048,6 +1027,9 @@ type Config struct {
 	InproxyProxyTotalActivityNoticePeriodMilliseconds      *int
 	InproxyClientDialRateLimitQuantity                     *int
 	InproxyClientDialRateLimitIntervalMilliseconds         *int
+	InproxyClientNoMatchFailoverProbability                *float64
+	InproxyClientNoMatchFailoverPersonalProbability        *float64
+	InproxyFrontingProviderClientMaxRequestTimeouts        map[string]string
 
 	InproxySkipAwaitFullyConnected  bool
 	InproxyEnableWebRTCDebugLogging bool
@@ -1421,18 +1403,14 @@ func (config *Config) Commit(migrateFromLegacyFields bool) error {
 	if !config.DisableTunnels &&
 		config.InproxyEnableProxy &&
 		!GetAllowOverlappingPersonalCompartmentIDs() &&
-		common.ContainsAny(
-			config.InproxyProxyPersonalCompartmentIDs,
-			config.InproxyClientPersonalCompartmentIDs) {
+		len(config.InproxyProxyPersonalCompartmentID) > 0 &&
+		config.InproxyProxyPersonalCompartmentID ==
+			config.InproxyClientPersonalCompartmentID {
 
 		// Don't allow an in-proxy client and proxy run in the same app to match.
 		return errors.TraceNew("invalid overlapping personal compartment IDs")
 	}
 
-	if len(config.InproxyProxyPersonalCompartmentIDs) > 1 {
-		return errors.TraceNew("invalid proxy personal compartment ID count")
-	}
-
 	// This constraint is expected by logic in Controller.runTunnels().
 
 	if config.PacketTunnelTunFileDescriptor > 0 && config.TunnelPoolSize != 1 {
@@ -1808,9 +1786,9 @@ func (config *Config) SetSignalComponentFailure(signalComponentFailure func()) {
 
 // IsInproxyPersonalPairingMode indicates that the client is in in-proxy
 // personal pairing mode, where connections are made only through in-proxy
-// proxies with corresponding personal compartment IDs.
+// proxies with the corresponding personal compartment ID.
 func (config *Config) IsInproxyPersonalPairingMode() bool {
-	return len(config.InproxyClientPersonalCompartmentIDs) > 0
+	return len(config.InproxyClientPersonalCompartmentID) > 0
 }
 
 // OnInproxyMustUpgrade is invoked when the in-proxy broker returns the
@@ -2517,6 +2495,10 @@ func (config *Config) makeConfigParameters() map[string]interface{} {
 		applyParameters[parameters.InproxyClientOfferRequestTimeout] = fmt.Sprintf("%dms", *config.InproxyClientOfferRequestTimeoutMilliseconds)
 	}
 
+	if config.InproxyClientOfferRequestPersonalTimeoutMilliseconds != nil {
+		applyParameters[parameters.InproxyClientOfferRequestPersonalTimeout] = fmt.Sprintf("%dms", *config.InproxyClientOfferRequestPersonalTimeoutMilliseconds)
+	}
+
 	if config.InproxyClientOfferRetryDelayMilliseconds != nil {
 		applyParameters[parameters.InproxyClientOfferRetryDelay] = fmt.Sprintf("%dms", *config.InproxyClientOfferRetryDelayMilliseconds)
 	}
@@ -2657,6 +2639,18 @@ func (config *Config) makeConfigParameters() map[string]interface{} {
 		applyParameters[parameters.InproxyClientDialRateLimitInterval] = fmt.Sprintf("%dms", *config.InproxyClientDialRateLimitIntervalMilliseconds)
 	}
 
+	if config.InproxyClientNoMatchFailoverProbability != nil {
+		applyParameters[parameters.InproxyClientNoMatchFailoverProbability] = *config.InproxyClientNoMatchFailoverProbability
+	}
+
+	if config.InproxyClientNoMatchFailoverPersonalProbability != nil {
+		applyParameters[parameters.InproxyClientNoMatchFailoverPersonalProbability] = *config.InproxyClientNoMatchFailoverPersonalProbability
+	}
+
+	if config.InproxyFrontingProviderClientMaxRequestTimeouts != nil {
+		applyParameters[parameters.InproxyFrontingProviderClientMaxRequestTimeouts] = config.InproxyFrontingProviderClientMaxRequestTimeouts
+	}
+
 	// When adding new config dial parameters that may override tactics, also
 	// update setDialParametersHash.
 
@@ -3324,38 +3318,6 @@ func (config *Config) setDialParametersHash() {
 		hash.Write([]byte("InproxyMaxCompartmentIDListLength"))
 		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyMaxCompartmentIDListLength))
 	}
-	if config.InproxyProxyAnnounceRequestTimeoutMilliseconds != nil {
-		hash.Write([]byte("InproxyProxyAnnounceRequestTimeoutMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyProxyAnnounceRequestTimeoutMilliseconds))
-	}
-	if config.InproxyProxyAnnounceDelayMilliseconds != nil {
-		hash.Write([]byte("InproxyProxyAnnounceDelayMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyProxyAnnounceDelayMilliseconds))
-	}
-	if config.InproxyProxyAnnounceDelayJitter != nil {
-		hash.Write([]byte("InproxyProxyAnnounceDelayJitter"))
-		binary.Write(hash, binary.LittleEndian, *config.InproxyProxyAnnounceDelayJitter)
-	}
-	if config.InproxyProxyAnswerRequestTimeoutMilliseconds != nil {
-		hash.Write([]byte("InproxyProxyAnswerRequestTimeoutMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyProxyAnswerRequestTimeoutMilliseconds))
-	}
-	if config.InproxyClientOfferRequestTimeoutMilliseconds != nil {
-		hash.Write([]byte("InproxyClientOfferRequestTimeoutMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyClientOfferRequestTimeoutMilliseconds))
-	}
-	if config.InproxyClientOfferRetryDelayMilliseconds != nil {
-		hash.Write([]byte("InproxyClientOfferRetryDelayMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyClientOfferRetryDelayMilliseconds))
-	}
-	if config.InproxyClientOfferRetryJitter != nil {
-		hash.Write([]byte("InproxyClientOfferRetryJitter"))
-		binary.Write(hash, binary.LittleEndian, *config.InproxyClientOfferRetryJitter)
-	}
-	if config.InproxyClientRelayedPacketRequestTimeoutMilliseconds != nil {
-		hash.Write([]byte("InproxyClientRelayedPacketRequestTimeoutMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyClientRelayedPacketRequestTimeoutMilliseconds))
-	}
 	if config.InproxyDTLSRandomizationProbability != nil {
 		hash.Write([]byte("InproxyDTLSRandomizationProbability"))
 		binary.Write(hash, binary.LittleEndian, *config.InproxyDTLSRandomizationProbability)
@@ -3444,34 +3406,6 @@ func (config *Config) setDialParametersHash() {
 		hash.Write([]byte("InproxyClientDisableIPv6ICECandidates"))
 		binary.Write(hash, binary.LittleEndian, *config.InproxyClientDisableIPv6ICECandidates)
 	}
-	if config.InproxyProxyDiscoverNATTimeoutMilliseconds != nil {
-		hash.Write([]byte("InproxyProxyDiscoverNATTimeoutMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyProxyDiscoverNATTimeoutMilliseconds))
-	}
-	if config.InproxyClientDiscoverNATTimeoutMilliseconds != nil {
-		hash.Write([]byte("InproxyClientDiscoverNATTimeoutMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyClientDiscoverNATTimeoutMilliseconds))
-	}
-	if config.InproxyWebRTCAnswerTimeoutMilliseconds != nil {
-		hash.Write([]byte("InproxyWebRTCAnswerTimeoutMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyWebRTCAnswerTimeoutMilliseconds))
-	}
-	if config.InproxyProxyWebRTCAwaitDataChannelTimeoutMilliseconds != nil {
-		hash.Write([]byte("InproxyProxyWebRTCAwaitDataChannelTimeoutMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyProxyWebRTCAwaitDataChannelTimeoutMilliseconds))
-	}
-	if config.InproxyClientWebRTCAwaitDataChannelTimeoutMilliseconds != nil {
-		hash.Write([]byte("InproxyClientWebRTCAwaitDataChannelTimeoutMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyClientWebRTCAwaitDataChannelTimeoutMilliseconds))
-	}
-	if config.InproxyProxyDestinationDialTimeoutMilliseconds != nil {
-		hash.Write([]byte("InproxyProxyDestinationDialTimeoutMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyProxyDestinationDialTimeoutMilliseconds))
-	}
-	if config.InproxyPsiphonAPIRequestTimeoutMilliseconds != nil {
-		hash.Write([]byte("InproxyPsiphonAPIRequestTimeoutMilliseconds"))
-		binary.Write(hash, binary.LittleEndian, int64(*config.InproxyPsiphonAPIRequestTimeoutMilliseconds))
-	}
 
 	config.dialParametersHash = hash.Sum(nil)
 }

psiphon/dataStore.go

@@ -2193,17 +2193,26 @@ func SetNetworkReplayParameters[R any](networkID, replayID string, replayParams
 	return setBucketValue(datastoreNetworkReplayParametersBucket, key, data)
 }
 
-// ShuffleAndGetNetworkReplayParameters takes a list of candidate objects and
-// selects one. The candidates are considered in random order. The first
-// candidate with a valid replay record is returned, along with its replay
-// parameters. The caller provides isValidReplay which should indicate if
-// replay parameters remain valid; the caller should check for expiry and
-// changes to the underlhying tactics. When no valid replay parameters are
-// found, ShuffleAndGetNetworkReplayParameters returns a candidate and nil
-// replay parameters.
-func ShuffleAndGetNetworkReplayParameters[C, R any](
+// SelectCandidateWithNetworkReplayParameters takes a list of candidate
+// objects and selects one. The candidates are considered in the specified
+// order. The first candidate with a valid replay record is returned, along
+// with its replay parameters.
+//
+// The caller provides isValidReplay which should indicate if replay
+// parameters remain valid; the caller should check for expiry and changes to
+// the underlhying tactics.
+//
+// When no candidates with valid replay parameters are found,
+// SelectCandidateWithNetworkReplayParameters returns the first candidate and
+// nil replay parameters.
+//
+// When selectFirstCandidate is specified,
+// SelectCandidateWithNetworkReplayParameters will check for valid replay
+// parameters for the first candidate only, and then select the first
+// candidate.
+func SelectCandidateWithNetworkReplayParameters[C, R any](
 	networkID string,
-	replayEnabled bool,
+	selectFirstCandidate bool,
 	candidates []*C,
 	getReplayID func(*C) string,
 	isValidReplay func(*C, *R) bool) (*C, *R, error) {
@@ -2212,25 +2221,14 @@ func ShuffleAndGetNetworkReplayParameters[C, R any](
 		return nil, nil, errors.TraceNew("no candidates")
 	}
 
-	// Don't shuffle or otherwise mutate the candidates slice, which may be a
-	// tactics parameter.
-	permutedIndexes := prng.Perm(len(candidates))
-
-	candidate := candidates[permutedIndexes[0]]
+	candidate := candidates[0]
 	var replay *R
 
-	if !replayEnabled {
-		// If replay is disabled, return the first post-shuffle candidate with
-		// nil replay parameters.
-		return candidate, replay, nil
-	}
-
 	err := datastoreUpdate(func(tx *datastoreTx) error {
 
 		bucket := tx.bucket(datastoreNetworkReplayParametersBucket)
 
-		for _, i := range permutedIndexes {
-			c := candidates[i]
+		for _, c := range candidates {
 			key := makeNetworkReplayParametersKey[R](networkID, getReplayID(c))
 			value := bucket.get(key)
 			if value == nil {
@@ -2243,9 +2241,9 @@ func ShuffleAndGetNetworkReplayParameters[C, R any](
 				// Delete the record. This avoids continually checking it.
 				// Note that the deletes performed here won't prune records
 				// for old candidates which are no longer passed in to
-				// ShuffleAndGetNetworkReplayParameters.
+				// SelectCandidateWithNetworkReplayParameters.
 				NoticeWarning(
-					"ShuffleAndGetNetworkReplayParameters: unmarshal failed: %s",
+					"SelectCandidateWithNetworkReplayParameters: unmarshal failed: %s",
 					errors.Trace(err))
 				_ = bucket.delete(key)
 				continue
@@ -2254,6 +2252,8 @@ func ShuffleAndGetNetworkReplayParameters[C, R any](
 				candidate = c
 				replay = r
 				return nil
+			} else if selectFirstCandidate {
+				return nil
 			} else {
 
 				// Delete the record if it's no longer valid due to expiry or

psiphon/dialParameters.go

@@ -1653,15 +1653,22 @@ func (dialParams *DialParameters) GetInproxyMetrics() common.LogFields {
 		return inproxyMetrics
 	}
 
-	for _, metrics := range []common.LogFields{
-		dialParams.inproxyBrokerDialParameters.GetMetrics(),
-		dialParams.InproxySTUNDialParameters.GetMetrics(),
-		dialParams.InproxyWebRTCDialParameters.GetMetrics(),
-	} {
-		for name, value := range metrics {
-			inproxyMetrics[name] = value
-		}
+	inproxyMetrics.Add(dialParams.inproxyBrokerDialParameters.GetMetrics())
+	inproxyMetrics.Add(dialParams.InproxySTUNDialParameters.GetMetrics())
+	inproxyMetrics.Add(dialParams.InproxyWebRTCDialParameters.GetMetrics())
+
+	return inproxyMetrics
+}
+
+func (dialParams *DialParameters) GetInproxyBrokerMetrics() common.LogFields {
+	inproxyMetrics := common.LogFields{}
+
+	if !dialParams.inproxyDialInitialized {
+		return inproxyMetrics
 	}
+
+	inproxyMetrics.Add(dialParams.inproxyBrokerDialParameters.GetBrokerMetrics())
+
 	return inproxyMetrics
 }
 

psiphon/inproxy.go

@@ -75,6 +75,7 @@ type InproxyBrokerClientManager struct {
 	isProxy bool
 
 	mutex                sync.Mutex
+	brokerSelectCount    int
 	networkID            string
 	brokerClientInstance *InproxyBrokerClientInstance
 }
@@ -101,6 +102,7 @@ func NewInproxyBrokerClientManager(
 // called when tactics have changed, which triggers a broker client reset in
 // order to apply potentially changed parameters.
 func (b *InproxyBrokerClientManager) TacticsApplied() error {
+
 	b.mutex.Lock()
 	defer b.mutex.Unlock()
 
@@ -113,7 +115,7 @@ func (b *InproxyBrokerClientManager) TacticsApplied() error {
 	// TODO: as a future future enhancement, don't reset when the tactics
 	// brokerSpecs.Hash() is unchanged?
 
-	return errors.Trace(b.reset())
+	return errors.Trace(b.reset(resetBrokerClientReasonTacticsApplied))
 }
 
 // GetBrokerClient returns the current, shared broker client and its
@@ -128,7 +130,7 @@ func (b *InproxyBrokerClientManager) GetBrokerClient(
 	defer b.mutex.Unlock()
 
 	if b.brokerClientInstance == nil || b.networkID != networkID {
-		err := b.reset()
+		err := b.reset(resetBrokerClientReasonInit)
 		if err != nil {
 			return nil, nil, errors.Trace(err)
 		}
@@ -155,10 +157,53 @@ func (b *InproxyBrokerClientManager) resetBrokerClientOnRoundTripperFailed(
 		return nil
 	}
 
-	return errors.Trace(b.reset())
+	return errors.Trace(b.reset(resetBrokerClientReasonRoundTripperFailed))
+}
+
+func (b *InproxyBrokerClientManager) resetBrokerClientOnNoMatch(
+	brokerClientInstance *InproxyBrokerClientInstance) error {
+
+	// Ignore the no match callback for proxies. For personal pairing, the
+	// broker rotation scheme has clients moving brokers to find relatively
+	// static proxies. For common pairing, we want to achieve balanced supply
+	// across brokers.
+	//
+	// Currently, inproxy.BrokerDialCoordinator.BrokerClientNoMatch is only
+	// wired up for clients, but this check ensures it'll still be ignored in
+	// case that changes.
+	if b.isProxy {
+		return nil
+	}
+
+	if b.brokerClientInstance != brokerClientInstance {
+		// See comment for same logic in resetBrokerClientOnRoundTripperFailed.
+		return nil
+	}
+
+	p := b.config.GetParameters().Get()
+	defer p.Close()
+
+	probability := parameters.InproxyClientNoMatchFailoverProbability
+	if b.config.IsInproxyPersonalPairingMode() {
+		probability = parameters.InproxyClientNoMatchFailoverPersonalProbability
+	}
+	if !p.WeightedCoinFlip(probability) {
+		return nil
+	}
+
+	return errors.Trace(b.reset(resetBrokerClientReasonRoundNoMatch))
 }
 
-func (b *InproxyBrokerClientManager) reset() error {
+type resetBrokerClientReason int
+
+const (
+	resetBrokerClientReasonInit resetBrokerClientReason = iota + 1
+	resetBrokerClientReasonTacticsApplied
+	resetBrokerClientReasonRoundTripperFailed
+	resetBrokerClientReasonRoundNoMatch
+)
+
+func (b *InproxyBrokerClientManager) reset(reason resetBrokerClientReason) error {
 
 	// Assumes b.mutex lock is held.
 
@@ -173,6 +218,20 @@ func (b *InproxyBrokerClientManager) reset() error {
 		b.brokerClientInstance.Close()
 	}
 
+	// b.brokerSelectCount tracks the number of broker resets and is used to
+	// iterate over the brokers in a deterministic rotation when running in
+	// personal pairing mode.
+
+	switch reason {
+	case resetBrokerClientReasonInit,
+		resetBrokerClientReasonTacticsApplied:
+		b.brokerSelectCount = 0
+
+	case resetBrokerClientReasonRoundTripperFailed,
+		resetBrokerClientReasonRoundNoMatch:
+		b.brokerSelectCount += 1
+	}
+
 	// Any existing broker client is removed, even if
 	// NewInproxyBrokerClientInstance fails. This ensures, for example, that
 	// an existing broker client is removed when its spec is no longer
@@ -183,7 +242,12 @@ func (b *InproxyBrokerClientManager) reset() error {
 	networkID := b.config.GetNetworkID()
 
 	brokerClientInstance, err := NewInproxyBrokerClientInstance(
-		b.config, b, networkID, b.isProxy)
+		b.config,
+		b,
+		networkID,
+		b.isProxy,
+		b.brokerSelectCount,
+		reason == resetBrokerClientReasonRoundNoMatch)
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -218,6 +282,7 @@ type InproxyBrokerClientInstance struct {
 	announceDelayJitter           float64
 	answerRequestTimeout          time.Duration
 	offerRequestTimeout           time.Duration
+	offerRequestPersonalTimeout   time.Duration
 	offerRetryDelay               time.Duration
 	offerRetryJitter              float64
 	relayedPacketRequestTimeout   time.Duration
@@ -236,7 +301,9 @@ func NewInproxyBrokerClientInstance(
 	config *Config,
 	brokerClientManager *InproxyBrokerClientManager,
 	networkID string,
-	isProxy bool) (*InproxyBrokerClientInstance, error) {
+	isProxy bool,
+	brokerSelectCount int,
+	resetReasonNoMatch bool) (*InproxyBrokerClientInstance, error) {
 
 	p := config.GetParameters().Get()
 	defer p.Close()
@@ -251,6 +318,9 @@ func NewInproxyBrokerClientInstance(
 	if !isProxy && len(commonCompartmentIDs) == 0 && len(personalCompartmentIDs) == 0 {
 		return nil, errors.TraceNew("no compartment IDs")
 	}
+	if len(personalCompartmentIDs) > 1 {
+		return nil, errors.TraceNew("unexpected multiple personal compartment IDs")
+	}
 
 	// Select the broker to use, optionally favoring brokers with replay data.
 	// In the InproxyBrokerSpecs calls, the first non-empty tactics parameter
@@ -290,45 +360,109 @@ func NewInproxyBrokerClientInstance(
 		return nil, errors.TraceNew("no broker specs")
 	}
 
-	// To ensure personal compartment ID client/proxy rendezvous at same
-	// broker, simply pick the first configured broker.
+	// Select a broker.
+
+	// In common pairing mode, the available brokers are shuffled before
+	// selection, for random load balancing. Brokers with available dial
+	// parameter replay data are preferred. When rotating brokers due to a no
+	// match, the available replay data is ignored to increase the chance of
+	// selecting a different broker.
+	//
+	// In personal pairing mode, arrange for the proxy and client to
+	// rendezvous at the same broker by shuffling based on the shared
+	// personal compartment ID. Both the client and proxy will select the
+	// same initial broker, and fail over to other brokers in the same order.
+	// By design, clients will move between brokers aggressively, rotating on
+	// no-match responses and applying a shorter client offer timeout; while
+	// proxies will remain in place in order to be found. Since rendezvous
+	// depends on the ordering, each broker is selected in shuffle order;
+	// dial parameter replay data is used when available but not considered
+	// in selection ordering. The brokerSelectCount input is used to
+	// progressively index into the list of shuffled brokers.
+	//
+	// Potential future enhancements:
 	//
-	// Limitations: there's no failover or load balancing for the personal
-	// compartment ID case; and this logic assumes that the broker spec
-	// tactics are the same for the client and proxy.
+	// - Use brokerSelectCount in the common pairing case as well, to ensure
+	//   that a no-match reset always selects a different broker; but, unlike
+	//   the personal pairing logic, still prefer brokers with replay rather
+	//   than following a strict shuffle order.
+	//
+	// - The common pairing no match broker rotation is intended to partially
+	//   mitigate poor common proxy load balancing that can leave a broker
+	//   with little proxy supply. A more robust mitigation would be to make
+	//   proxies distribute announcements across multiple or even all brokers.
+
+	personalPairing := len(personalCompartmentIDs) > 0
 
-	if len(personalCompartmentIDs) > 0 {
-		brokerSpecs = brokerSpecs[:1]
+	// In the following cases, don't shuffle or otherwise mutate the original
+	// broker spec slice, as it is a tactics parameter.
+
+	if personalPairing {
+
+		if len(personalCompartmentIDs[0]) < prng.SEED_LENGTH {
+			// Both inproxy.ID and prng.SEED_LENGTH are 32 bytes.
+			return nil, errors.TraceNew("unexpected ID length")
+		}
+
+		seed := prng.Seed(personalCompartmentIDs[0][0:prng.SEED_LENGTH])
+		PRNG := prng.NewPRNGWithSeed(&seed)
+
+		permutedIndexes := PRNG.Perm(len(brokerSpecs))
+		selectedIndex := permutedIndexes[brokerSelectCount%len(permutedIndexes)]
+		brokerSpecs = brokerSpecs[selectedIndex : selectedIndex+1]
+
+	} else {
+
+		permutedIndexes := prng.Perm(len(brokerSpecs))
+		shuffledBrokerSpecs := make(parameters.InproxyBrokerSpecsValue, len(brokerSpecs))
+		for i, index := range permutedIndexes {
+			shuffledBrokerSpecs[i] = brokerSpecs[index]
+		}
+		brokerSpecs = shuffledBrokerSpecs
 	}
 
-	now := time.Now()
+	selectFirstCandidate := resetReasonNoMatch || personalPairing
 
-	// Prefer a broker with replay data.
+	// Replay broker dial parameters.
+
+	// In selectFirstCandidate cases, SelectCandidateWithNetworkReplayParameters
+	// will always select the first candidate, returning corresponding replay
+	// data when available. Otherwise, SelectCandidateWithNetworkReplayParameters
+	// iterates over the shuffled candidates and returns the first with replay data.
+
+	var brokerSpec *parameters.InproxyBrokerSpec
+	var brokerDialParams *InproxyBrokerDialParameters
 
 	// Replay is disabled when the TTL, InproxyReplayBrokerDialParametersTTL,
 	// is 0.
+	now := time.Now()
 	ttl := p.Duration(parameters.InproxyReplayBrokerDialParametersTTL)
 
 	replayEnabled := ttl > 0 &&
 		!config.DisableReplay &&
 		prng.FlipWeightedCoin(p.Float(parameters.InproxyReplayBrokerDialParametersProbability))
 
-	brokerSpec, brokerDialParams, err :=
-		ShuffleAndGetNetworkReplayParameters[parameters.InproxyBrokerSpec, InproxyBrokerDialParameters](
-			networkID,
-			replayEnabled,
-			brokerSpecs,
-			func(spec *parameters.InproxyBrokerSpec) string { return spec.BrokerPublicKey },
-			func(spec *parameters.InproxyBrokerSpec, dialParams *InproxyBrokerDialParameters) bool {
-				return dialParams.LastUsedTimestamp.After(now.Add(-ttl)) &&
-					bytes.Equal(dialParams.LastUsedBrokerSpecHash, hashBrokerSpec(spec))
-			})
-	if err != nil {
-		NoticeWarning("ShuffleAndGetNetworkReplayParameters failed: %v", errors.Trace(err))
+	if replayEnabled {
+		brokerSpec, brokerDialParams, err =
+			SelectCandidateWithNetworkReplayParameters[parameters.InproxyBrokerSpec, InproxyBrokerDialParameters](
+				networkID,
+				selectFirstCandidate,
+				brokerSpecs,
+				func(spec *parameters.InproxyBrokerSpec) string { return spec.BrokerPublicKey },
+				func(spec *parameters.InproxyBrokerSpec, dialParams *InproxyBrokerDialParameters) bool {
+					return dialParams.LastUsedTimestamp.After(now.Add(-ttl)) &&
+						bytes.Equal(dialParams.LastUsedBrokerSpecHash, hashBrokerSpec(spec))
+				})
+		if err != nil {
+			NoticeWarning("SelectCandidateWithNetworkReplayParameters failed: %v", errors.Trace(err))
+			// Continue without replay
+		}
+	}
 
-		// When there's an error, try to continue, using a random broker spec
-		// and no replay dial parameters.
-		brokerSpec = brokerSpecs[prng.Intn(len(brokerSpecs)-1)]
+	// Select the first broker in the shuffle when replay is not enabled or in
+	// case SelectCandidateWithNetworkReplayParameters fails.
+	if brokerSpec == nil {
+		brokerSpec = brokerSpecs[0]
 	}
 
 	// Generate new broker dial parameters if not replaying. Later, isReplay
@@ -407,6 +541,7 @@ func NewInproxyBrokerClientInstance(
 		announceDelayJitter:           p.Float(parameters.InproxyProxyAnnounceDelayJitter),
 		answerRequestTimeout:          p.Duration(parameters.InproxyProxyAnswerRequestTimeout),
 		offerRequestTimeout:           p.Duration(parameters.InproxyClientOfferRequestTimeout),
+		offerRequestPersonalTimeout:   p.Duration(parameters.InproxyClientOfferRequestPersonalTimeout),
 		offerRetryDelay:               p.Duration(parameters.InproxyClientOfferRetryDelay),
 		offerRetryJitter:              p.Float(parameters.InproxyClientOfferRetryJitter),
 		relayedPacketRequestTimeout:   p.Duration(parameters.InproxyClientRelayedPacketRequestTimeout),
@@ -414,6 +549,22 @@ func NewInproxyBrokerClientInstance(
 		replayUpdateFrequency:         p.Duration(parameters.InproxyReplayBrokerUpdateFrequency),
 	}
 
+	// Adjust long-polling request timeouts to respect any maximum request
+	// timeout supported by the provider fronting the request.
+	maxRequestTimeout, ok := p.KeyDurations(
+		parameters.InproxyFrontingProviderClientMaxRequestTimeouts)[brokerDialParams.FrontingProviderID]
+	if ok && maxRequestTimeout > 0 {
+		if b.announceRequestTimeout > maxRequestTimeout {
+			b.announceRequestTimeout = maxRequestTimeout
+		}
+		if b.offerRequestTimeout > maxRequestTimeout {
+			b.offerRequestTimeout = maxRequestTimeout
+		}
+		if b.offerRequestPersonalTimeout > maxRequestTimeout {
+			b.offerRequestPersonalTimeout = maxRequestTimeout
+		}
+	}
+
 	// Initialize broker client. This will start with a fresh broker session.
 	//
 	// When resetBrokerClientOnRoundTripperFailed is invoked due to a failure
@@ -462,9 +613,29 @@ func prepareCompartmentIDs(
 
 	maxCompartmentIDListLength := p.Int(parameters.InproxyMaxCompartmentIDListLength)
 
-	configPersonalCompartmentIDs := config.InproxyProxyPersonalCompartmentIDs
-	if !isProxy {
-		configPersonalCompartmentIDs = config.InproxyClientPersonalCompartmentIDs
+	// Personal compartment ID limitations:
+	//
+	// The broker API messages, ProxyAnnounceRequest and ClientOfferRequest,
+	// support lists of personal compartment IDs. However, both the proxy and
+	// the client are currently limited to specifying at most one personal
+	// compartment ID due to the following limitations:
+	//
+	// - On the broker side, the matcher queue implementation supports at most
+	//   one proxy personal compartment ID. See inproxy/Matcher.Announce. The
+	//   broker currently enforces that at most one personal compartment ID
+	//   may be specified per ProxyAnnounceRequest.
+	//
+	// - On the proxy/client side, the personal pairing rendezvous logic --
+	//   which aims for proxies and clients to select the same initial broker
+	//   and same order of failover to other brokers -- uses a shuffle that
+	//   assumes both the proxy and client use the same single, personal
+	//   compartment ID
+
+	var configPersonalCompartmentIDs []string
+	if isProxy && len(config.InproxyProxyPersonalCompartmentID) > 0 {
+		configPersonalCompartmentIDs = []string{config.InproxyProxyPersonalCompartmentID}
+	} else if !isProxy && len(config.InproxyClientPersonalCompartmentID) > 0 {
+		configPersonalCompartmentIDs = []string{config.InproxyClientPersonalCompartmentID}
 	}
 	personalCompartmentIDs, err := inproxy.IDsFromStrings(configPersonalCompartmentIDs)
 	if err != nil {
@@ -698,6 +869,27 @@ func (b *InproxyBrokerClientInstance) BrokerClientRoundTripperFailed(roundTrippe
 	}
 }
 
+// Implements the inproxy.BrokerDialCoordinator interface.
+func (b *InproxyBrokerClientInstance) BrokerClientNoMatch(roundTripper inproxy.RoundTripper) {
+	b.mutex.Lock()
+	defer b.mutex.Unlock()
+
+	if rt, ok := roundTripper.(*InproxyBrokerRoundTripper); !ok || rt != b.roundTripper {
+		// See roundTripper check comment in BrokerClientRoundTripperFailed.
+		NoticeError("BrokerClientNoMatch: roundTripper instance mismatch")
+		return
+	}
+
+	// Any persistent replay dial parameters are retained and not deleted,
+	// since the broker client successfully transacted with the broker.
+
+	err := b.brokerClientManager.resetBrokerClientOnNoMatch(b)
+	if err != nil {
+		NoticeWarning("reset broker client failed: %v", errors.Trace(err))
+		// Continue with old broker client instance.
+	}
+}
+
 // Implements the inproxy.BrokerDialCoordinator interface.
 func (b *InproxyBrokerClientInstance) AnnounceRequestTimeout() time.Duration {
 	return b.announceRequestTimeout
@@ -728,6 +920,11 @@ func (b *InproxyBrokerClientInstance) OfferRequestTimeout() time.Duration {
 	return b.offerRequestTimeout
 }
 
+// Implements the inproxy.BrokerDialCoordinator interface.
+func (b *InproxyBrokerClientInstance) OfferRequestPersonalTimeout() time.Duration {
+	return b.offerRequestPersonalTimeout
+}
+
 // Implements the inproxy.BrokerDialCoordinator interface.
 func (b *InproxyBrokerClientInstance) OfferRetryDelay() time.Duration {
 	return b.offerRetryDelay
@@ -1075,11 +1272,26 @@ func (brokerDialParams *InproxyBrokerDialParameters) prepareDialConfigs(
 	return nil
 }
 
+// GetBrokerMetrics returns  dial parameter log fields to be reported to a
+// broker.
+func (brokerDialParams *InproxyBrokerDialParameters) GetBrokerMetrics() common.LogFields {
+
+	logFields := common.LogFields{}
+
+	// TODO: add additional broker fronting dial parameters to be logged by
+	// the broker -- as successful parameters might not otherwise by logged
+	// via server_tunnel if the subsequent WebRTC dials fail.
+
+	logFields["fronting_provider_id"] = brokerDialParams.FrontingProviderID
+
+	return logFields
+}
+
 // GetMetrics implements the common.MetricsSource interface and returns log
 // fields detailing the broker dial parameters.
 func (brokerDialParams *InproxyBrokerDialParameters) GetMetrics() common.LogFields {
 
-	logFields := make(common.LogFields)
+	logFields := common.LogFields{}
 
 	logFields["inproxy_broker_transport"] = brokerDialParams.BrokerTransport
 

psiphon/inproxy_test.go

@@ -71,9 +71,8 @@ func runInproxyBrokerDialParametersTest() error {
 	networkID := "NETWORK1"
 	addressRegex := `[a-z0-9]{5,10}\.example\.org`
 	commonCompartmentID, _ := inproxy.MakeID()
-	personalCompartmentID, _ := inproxy.MakeID()
 	commonCompartmentIDs := []string{commonCompartmentID.String()}
-	personalCompartmentIDs := []string{personalCompartmentID.String()}
+	personalCompartmentID, _ := inproxy.MakeID()
 	privateKey, _ := inproxy.GenerateSessionPrivateKey()
 	publicKey, _ := privateKey.GetPublicKey()
 	obfuscationSecret, _ := inproxy.GenerateRootObfuscationSecret()
@@ -269,8 +268,8 @@ func runInproxyBrokerDialParametersTest() error {
 
 	// Test: no common compartment IDs sent when personal ID is set
 
-	config.InproxyClientPersonalCompartmentIDs = personalCompartmentIDs
-	config.InproxyProxyPersonalCompartmentIDs = personalCompartmentIDs
+	config.InproxyClientPersonalCompartmentID = personalCompartmentID.String()
+	config.InproxyProxyPersonalCompartmentID = personalCompartmentID.String()
 
 	manager = NewInproxyBrokerClientManager(config, isProxy)
 

psiphon/server/api.go

@@ -970,7 +970,8 @@ func getTacticsAPIParameterLogFieldFormatter() common.APIParameterLogFieldFormat
 var inproxyBrokerRequestParams = append(
 	append(
 		[]requestParamSpec{
-			{"session_id", isHexDigits, 0}},
+			{"session_id", isHexDigits, 0},
+			{"fronting_provider_id", isAnyString, requestParamOptional}},
 		tacticsParams...),
 	baseParams...)
 

psiphon/server/meek.go

@@ -1840,7 +1840,9 @@ func (server *MeekServer) inproxyReloadTactics() error {
 	server.inproxyBroker.SetTimeouts(
 		p.Duration(parameters.InproxyBrokerProxyAnnounceTimeout),
 		p.Duration(parameters.InproxyBrokerClientOfferTimeout),
-		p.Duration(parameters.InproxyBrokerPendingServerRequestsTTL))
+		p.Duration(parameters.InproxyBrokerClientOfferPersonalTimeout),
+		p.Duration(parameters.InproxyBrokerPendingServerRequestsTTL),
+		p.KeyDurations(parameters.InproxyFrontingProviderServerMaxRequestTimeouts))
 
 	nonlimitedProxyIDs, err := inproxy.IDsFromStrings(
 		p.Strings(parameters.InproxyBrokerMatcherAnnouncementNonlimitedProxyIDs))
@@ -1980,6 +1982,19 @@ func (server *MeekServer) inproxyBrokerHandler(
 		geoIPData,
 		packet)
 	if err != nil {
+
+		var deobfuscationAnomoly *inproxy.DeobfuscationAnomoly
+		isAnomolous := std_errors.As(err, &deobfuscationAnomoly)
+		if isAnomolous {
+			logIrregularTunnel(
+				server.support,
+				server.listenerTunnelProtocol,
+				server.listenerPort,
+				clientIP,
+				errors.Trace(err),
+				nil)
+		}
+
 		return errors.Trace(err)
 	}
 

psiphon/server/server_test.go

@@ -1048,6 +1048,21 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 			case serverTunnelLog <- logFields:
 			default:
 			}
+		case "inproxy_broker":
+			// Check that broker receives the correct fronting provider ID.
+			//
+			// TODO: check inproxy_broker logs received when expected and
+			// check more fields
+			event, ok := logFields["broker_event"].(string)
+			if !ok {
+				t.Errorf("missing inproxy_broker.broker_event")
+			}
+			if event == "client_offer" || event == "proxy_announce" {
+				fronting_provider_id, ok := logFields["fronting_provider_id"].(string)
+				if !ok || fronting_provider_id != inproxyTestConfig.brokerFrontingProviderID {
+					t.Errorf("unexpected inproxy_broker.fronting_provider_id")
+				}
+			}
 		}
 	})
 
@@ -1327,8 +1342,8 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 			psiphon.SetAllowOverlappingPersonalCompartmentIDs(true)
 			defer psiphon.SetAllowOverlappingPersonalCompartmentIDs(false)
 
-			clientConfig.InproxyClientPersonalCompartmentIDs = []string{inproxyTestConfig.personalCompartmentID}
-			clientConfig.InproxyProxyPersonalCompartmentIDs = []string{inproxyTestConfig.personalCompartmentID}
+			clientConfig.InproxyClientPersonalCompartmentID = inproxyTestConfig.personalCompartmentID
+			clientConfig.InproxyProxyPersonalCompartmentID = inproxyTestConfig.personalCompartmentID
 		}
 
 		// Simulate a CDN adding required HTTP headers by injecting them at
@@ -3612,6 +3627,18 @@ func generateInproxyTestConfig(
 		clientBrokerSpecsJSON = "[]"
 	}
 
+	maxRequestTimeoutsJSON := ""
+	if prng.FlipCoin() {
+		maxRequestTimeoutsJSONFormat := `
+            "InproxyFrontingProviderClientMaxRequestTimeouts": {"%s": "10s"},
+            "InproxyFrontingProviderServerMaxRequestTimeouts": {"%s": "5s"},
+        `
+		maxRequestTimeoutsJSON = fmt.Sprintf(
+			maxRequestTimeoutsJSONFormat,
+			brokerFrontingProviderID,
+			brokerFrontingProviderID)
+	}
+
 	tacticsParametersJSONFormat := `
             "InproxyAllowProxy": true,
             "InproxyAllowClient": true,
@@ -3626,6 +3653,7 @@ func generateInproxyTestConfig(
             "InproxyDisableSTUN": true,
             "InproxyDisablePortMapping": true,
             "InproxyDisableIPv6ICECandidates": true,
+            %s
     `
 
 	tacticsParametersJSON := fmt.Sprintf(
@@ -3636,7 +3664,8 @@ func generateInproxyTestConfig(
 		proxyBrokerSpecsJSON,
 		clientBrokerSpecsJSON,
 		commonCompartmentIDStr,
-		commonCompartmentIDStr)
+		commonCompartmentIDStr,
+		maxRequestTimeoutsJSON)
 
 	config := &inproxyTestConfig{
 		tacticsParametersJSON:               tacticsParametersJSON,

psiphon/tunnel.go

@@ -1547,13 +1547,11 @@ func dialInproxy(
 	// Unlike the proxy broker case, clients already actively fetch tactics
 	// during tunnel estalishment, so tactics.SetTacticsAPIParameters are not
 	// sent to the broker and no tactics are returned by the broker.
-	//
-	// TODO: include broker fronting dial parameters to be logged by the
-	// broker -- as successful parameters might not otherwise by logged via
-	// server_tunnel if the subsequent WebRTC dials fail.
 	params := getBaseAPIParameters(
 		baseParametersNoDialParameters, true, config, nil)
 
+	common.LogFields(params).Add(dialParams.GetInproxyBrokerMetrics())
+
 	// The debugLogging flag is passed to both NoticeCommonLogger and to the
 	// inproxy package as well; skipping debug logs in the inproxy package,
 	// before calling into the notice logger, avoids unnecessary allocations