In-proxy performance improvements

- Don't match client offers that are too close to timing out

- When a client isn't awaiting an answer, return a graceful error and don't
  backoff

psiphon/common/inproxy/api.go

@@ -461,8 +461,11 @@ type ProxyAnswerRequest struct {
 	// SelectedProtocolVersion int32 `cbor:"2,keyasint,omitempty"`
 }
 
-// ProxyAnswerResponse is the acknowledgement for a ProxyAnswerRequest.
+// ProxyAnswerResponse is the acknowledgement for a ProxyAnswerRequest. If
+// NoAwaitingClient is indicated, then the client was no longer awaiting the
+// answer and the proxy should abandon the connection attempt.
 type ProxyAnswerResponse struct {
+	NoAwaitingClient bool `cbor:"1,keyasint,omitempty"`
 }
 
 // ClientRelayedPacketRequest is an API request sent from a client to a

psiphon/common/inproxy/broker.go

@@ -406,6 +406,7 @@ func (b *Broker) SetLimits(
 	matcherOfferLimitEntryCount int,
 	matcherOfferRateLimitQuantity int,
 	matcherOfferRateLimitInterval time.Duration,
+	matcherOfferMinimumDeadline time.Duration,
 	maxCompartmentIDs int,
 	dslRequestRateLimitQuantity int,
 	dslRequestRateLimitInterval time.Duration) {
@@ -417,7 +418,8 @@ func (b *Broker) SetLimits(
 		matcherAnnouncementNonlimitedProxyIDs,
 		matcherOfferLimitEntryCount,
 		matcherOfferRateLimitQuantity,
-		matcherOfferRateLimitInterval)
+		matcherOfferRateLimitInterval,
+		matcherOfferMinimumDeadline)
 
 	b.maxCompartmentIDs.Store(
 		int64(common.ValueOrDefault(maxCompartmentIDs, MaxCompartmentIDs)))
@@ -1210,7 +1212,8 @@ func (b *Broker) handleClientOffer(
 		var limitError *MatcherLimitError
 		limited := std_errors.As(err, &limitError)
 
-		timeout := offerCtx.Err() == context.DeadlineExceeded
+		timeout := offerCtx.Err() == context.DeadlineExceeded ||
+			std_errors.Is(err, errOfferDropped)
 
 		// A no-match response is sent in the case of a timeout awaiting a
 		// match. The faster-failing rate or entry limiting case also results
@@ -1230,6 +1233,13 @@ func (b *Broker) handleClientOffer(
 			// InproxyClientOfferRequestTimeout in tactics, should be configured
 			// so that the broker will timeout first and have an opportunity to
 			// send this response before the client times out.
+			//
+			// In the errOfferDropped case, the matcher dropped the offer due
+			// to age. While this is distinct from a timeout after a
+			// completed match, the same timed_out log field is set. The
+			// cases can be distinguished based on elapsed_time, as the
+			// dropped cases will have an elapsed_time less than
+			// InproxyBrokerClientOfferTimeout.
 			timedOut = true
 		}
 
@@ -1418,6 +1428,18 @@ func (b *Broker) handleProxyAnswer(
 	hasPersonalCompartmentIDs, err := b.matcher.AnnouncementHasPersonalCompartmentIDs(
 		initiatorID, answerRequest.ConnectionID)
 	if err != nil {
+
+		if std_errors.Is(err, errNoPendingAnswer) {
+			// Return a response. This avoids returning a
+			// broker-client-resetting 404 in this case.
+			responsePayload, err := MarshalProxyAnswerResponse(
+				&ProxyAnswerResponse{NoAwaitingClient: true})
+			if err != nil {
+				return nil, errors.Trace(err)
+			}
+			return responsePayload, nil
+		}
+
 		return nil, errors.Trace(err)
 	}
 
@@ -1459,6 +1481,18 @@ func (b *Broker) handleProxyAnswer(
 
 		err = b.matcher.Answer(proxyAnswer)
 		if err != nil {
+
+			if std_errors.Is(err, errNoPendingAnswer) {
+				// Return a response. This avoids returning a
+				// broker-client-resetting 404 in this case.
+				responsePayload, err := MarshalProxyAnswerResponse(
+					&ProxyAnswerResponse{NoAwaitingClient: true})
+				if err != nil {
+					return nil, errors.Trace(err)
+				}
+				return responsePayload, nil
+			}
+
 			return nil, errors.Trace(err)
 		}
 	}

psiphon/common/inproxy/matcher.go

@@ -102,6 +102,7 @@ type Matcher struct {
 	offerLimitEntryCount     int
 	offerRateLimitQuantity   int
 	offerRateLimitInterval   time.Duration
+	offerMinimumDeadline     time.Duration
 
 	matchSignal chan struct{}
 
@@ -124,6 +125,7 @@ type MatcherConfig struct {
 	OfferLimitEntryCount   int
 	OfferRateLimitQuantity int
 	OfferRateLimitInterval time.Duration
+	OfferMinimumDeadline   time.Duration
 
 	// Proxy quality state.
 	ProxyQualityState *ProxyQualityState
@@ -220,6 +222,7 @@ type MatchAnswer struct {
 // MatchMetrics records statistics about the match queue state at the time a
 // match is made.
 type MatchMetrics struct {
+	OfferDeadline          time.Duration
 	OfferMatchIndex        int
 	OfferQueueSize         int
 	AnnouncementMatchIndex int
@@ -233,6 +236,7 @@ func (metrics *MatchMetrics) GetMetrics() common.LogFields {
 		return nil
 	}
 	return common.LogFields{
+		"offer_deadline":           int64(metrics.OfferDeadline / time.Millisecond),
 		"offer_match_index":        metrics.OfferMatchIndex,
 		"offer_queue_size":         metrics.OfferQueueSize,
 		"announcement_match_index": metrics.AnnouncementMatchIndex,
@@ -283,6 +287,11 @@ func (offerEntry *offerEntry) getMatchMetrics() *MatchMetrics {
 type answerInfo struct {
 	announcement *MatchAnnouncement
 	answer       *MatchAnswer
+
+	// offerDropped is sent to Offer's answer channel when the offer has been
+	// dropped by the matcher due to age. This allows Offer to return
+	// immediately on drop and the request handler to log this outcome.
+	offerDropped bool
 }
 
 // pendingAnswer represents an answer that is expected to arrive from a
@@ -333,7 +342,8 @@ func NewMatcher(config *MatcherConfig) *Matcher {
 		config.AnnouncementNonlimitedProxyIDs,
 		config.OfferLimitEntryCount,
 		config.OfferRateLimitQuantity,
-		config.OfferRateLimitInterval)
+		config.OfferRateLimitInterval,
+		config.OfferMinimumDeadline)
 
 	return m
 }
@@ -350,7 +360,8 @@ func (m *Matcher) SetLimits(
 	announcementNonlimitedProxyIDs []ID,
 	offerLimitEntryCount int,
 	offerRateLimitQuantity int,
-	offerRateLimitInterval time.Duration) {
+	offerRateLimitInterval time.Duration,
+	offerMinimumDeadline time.Duration) {
 
 	nonlimitedProxyIDs := make(map[ID]struct{})
 	for _, proxyID := range announcementNonlimitedProxyIDs {
@@ -368,6 +379,7 @@ func (m *Matcher) SetLimits(
 	m.offerLimitEntryCount = offerLimitEntryCount
 	m.offerRateLimitQuantity = offerRateLimitQuantity
 	m.offerRateLimitInterval = offerRateLimitInterval
+	m.offerMinimumDeadline = offerMinimumDeadline
 	m.offerQueueMutex.Unlock()
 }
 
@@ -473,6 +485,8 @@ func (m *Matcher) Announce(
 	return clientOffer, announcementEntry.getMatchMetrics(), nil
 }
 
+var errOfferDropped = std_errors.New("offer dropped")
+
 // Offer enqueues the client offer and blocks until it is matched with a
 // returned announcement or ctx is done. The caller must not mutate the offer
 // or its properties after calling Announce.
@@ -546,6 +560,11 @@ func (m *Matcher) Offer(
 			offerEntry.getMatchMetrics(), errors.TraceNew("no answer")
 	}
 
+	if proxyAnswerInfo.offerDropped {
+		return nil, nil,
+			offerEntry.getMatchMetrics(), errOfferDropped
+	}
+
 	// This is a sanity check and not expected to fail.
 	if !proxyAnswerInfo.answer.ConnectionID.Equal(
 		proxyAnswerInfo.announcement.ConnectionID) {
@@ -559,6 +578,8 @@ func (m *Matcher) Offer(
 		nil
 }
 
+var errNoPendingAnswer = std_errors.New("no pending answer")
+
 // AnnouncementHasPersonalCompartmentIDs looks for a pending answer for an
 // announcement identified by the specified proxy ID and connection ID and
 // returns whether the announcement has personal compartment IDs, indicating
@@ -573,7 +594,7 @@ func (m *Matcher) AnnouncementHasPersonalCompartmentIDs(
 	if !ok {
 		// The input IDs don't correspond to a pending answer, or the client
 		// is no longer awaiting the response.
-		return false, errors.TraceNew("no pending answer")
+		return false, errors.Trace(errNoPendingAnswer)
 	}
 
 	pendingAnswer := pendingAnswerValue.(*pendingAnswer)
@@ -599,7 +620,7 @@ func (m *Matcher) Answer(
 	if !ok {
 		// The input IDs don't correspond to a pending answer, or the client
 		// is no longer awaiting the response.
-		return errors.TraceNew("no pending answer")
+		return errors.Trace(errNoPendingAnswer)
 	}
 
 	m.pendingAnswers.Delete(key)
@@ -677,15 +698,30 @@ func (m *Matcher) matchAllOffers() {
 
 		offerEntry := offer.Value.(*offerEntry)
 
-		// Skip and remove this offer if its deadline has already passed.
-		// There is no signal to the awaiting Offer function, as it will exit
-		// based on the same ctx.
+		// Skip and remove this offer if its deadline has already passed or
+		// the context is canceled. There is no signal to the awaiting Offer
+		// function, as it will exit based on the same ctx.
 
 		if offerEntry.ctx.Err() != nil {
 			m.removeOfferEntry(false, offerEntry)
 			continue
 		}
 
+		offerDeadline, _ := offerEntry.ctx.Deadline()
+		untilOfferDeadline := time.Until(offerDeadline)
+
+		// Drop this offer if it no longer has a sufficient remaining deadline
+		// for the proxy answer phase. This case signals Offer's answerChan
+		// so it can return immediately.
+
+		if m.offerMinimumDeadline > 0 &&
+			untilOfferDeadline < m.offerMinimumDeadline {
+
+			m.removeOfferEntry(false, offerEntry)
+			offerEntry.answerChan <- &answerInfo{offerDropped: true}
+			continue
+		}
+
 		announcementEntry, announcementMatchIndex := m.matchOffer(offerEntry)
 		if announcementEntry == nil {
 			continue
@@ -698,6 +734,7 @@ func (m *Matcher) matchAllOffers() {
 		// were inspected before matching.
 
 		matchMetrics := &MatchMetrics{
+			OfferDeadline:          untilOfferDeadline,
 			OfferMatchIndex:        offerIndex,
 			OfferQueueSize:         m.offerQueue.Len(),
 			AnnouncementMatchIndex: announcementMatchIndex,

psiphon/common/inproxy/matcher_test.go

@@ -47,6 +47,7 @@ func runTestMatcher() error {
 	limitEntryCount := 50
 	rateLimitQuantity := 100
 	rateLimitInterval := 1000 * time.Millisecond
+	minimumDeadline := 1 * time.Hour
 
 	logger := testutils.NewTestLogger()
 
@@ -338,7 +339,7 @@ func runTestMatcher() error {
 
 	m.SetLimits(
 		0, rateLimitQuantity, rateLimitInterval, []ID{},
-		0, rateLimitQuantity, rateLimitInterval)
+		0, rateLimitQuantity, rateLimitInterval, 0)
 
 	time.Sleep(rateLimitInterval)
 
@@ -389,11 +390,29 @@ func runTestMatcher() error {
 		return errors.Tracef("unexpected result: %v", err)
 	}
 
+	// Test: offer dropped due to minimum deadline
+
+	m.SetLimits(
+		0, rateLimitQuantity, rateLimitInterval, []ID{},
+		0, rateLimitQuantity, rateLimitInterval, minimumDeadline)
+
 	time.Sleep(rateLimitInterval)
 
+	go proxyFunc(proxyResultChan, proxyIP, matchProperties, 10*time.Millisecond, nil, true)
+	go clientFunc(clientResultChan, clientIP, matchProperties, 10*time.Millisecond)
+
+	err = <-proxyResultChan
+
+	err = <-clientResultChan
+	if err == nil || !strings.HasSuffix(err.Error(), errOfferDropped.Error()) {
+		return errors.Tracef("unexpected result: %v", err)
+	}
+
 	m.SetLimits(
 		limitEntryCount, rateLimitQuantity, rateLimitInterval, []ID{},
-		limitEntryCount, rateLimitQuantity, rateLimitInterval)
+		limitEntryCount, rateLimitQuantity, rateLimitInterval, 0)
+
+	time.Sleep(rateLimitInterval)
 
 	// Test: basic match
 

psiphon/common/inproxy/proxy.go

@@ -1114,11 +1114,6 @@ func (p *Proxy) proxyOneClient(
 
 	// Trigger back-off if the following WebRTC operations fail to establish a
 	// connections.
-	//
-	// Limitation: the proxy answer request to the broker may fail due to the
-	// non-back-off reasons documented above for the proxy announcment request;
-	// however, these should be unlikely assuming that the broker client is
-	// using a persistent transport connection.
 
 	backOff = true
 
@@ -1183,7 +1178,7 @@ func (p *Proxy) proxyOneClient(
 
 	// Send answer request with SDP or error.
 
-	_, err = brokerClient.ProxyAnswer(
+	answerResponse, err := brokerClient.ProxyAnswer(
 		ctx,
 		&ProxyAnswerRequest{
 			ConnectionID:      announceResponse.ConnectionID,
@@ -1196,6 +1191,11 @@ func (p *Proxy) proxyOneClient(
 			// Prioritize returning any WebRTC error for logging.
 			return backOff, webRTCErr
 		}
+
+		// Don't backoff if the answer request fails due to possible transient
+		// request transport errors.
+
+		backOff = false
 		return backOff, errors.Trace(err)
 	}
 
@@ -1205,6 +1205,22 @@ func (p *Proxy) proxyOneClient(
 		return backOff, webRTCErr
 	}
 
+	// Exit if the client was no longer awaiting the answer. There is no
+	// backoff in this case, and there's no error, as the proxy did not fail
+	// as it's not an unexpected outcome.
+	//
+	// Limitation: it's possible that the announce request responds quickly
+	// and the matched client offer is already close to timing out. The
+	// answer request will also respond quickly. There's an increased chance
+	// of hitting rate limits in this fast turn around scenario. This outcome
+	// is mitigated by InproxyBrokerMatcherOfferMinimumDeadline.
+
+	if answerResponse.NoAwaitingClient {
+
+		backOff = false
+		return backOff, nil
+	}
+
 	// Await the WebRTC connection.
 
 	// We could concurrently dial the destination, to have that network

psiphon/common/parameters/parameters.go

@@ -428,6 +428,7 @@ const (
 	InproxyBrokerMatcherOfferLimitEntryCount           = "InproxyBrokerMatcherOfferLimitEntryCount"
 	InproxyBrokerMatcherOfferRateLimitQuantity         = "InproxyBrokerMatcherOfferRateLimitQuantity"
 	InproxyBrokerMatcherOfferRateLimitInterval         = "InproxyBrokerMatcherOfferRateLimitInterval"
+	InproxyBrokerMatcherOfferMinimumDeadline           = "InproxyBrokerMatcherOfferMinimumDeadline"
 	InproxyBrokerMatcherPrioritizeProxiesProbability   = "InproxyBrokerMatcherPrioritizeProxiesProbability"
 	InproxyBrokerMatcherPrioritizeProxiesFilter        = "InproxyBrokerMatcherPrioritizeProxiesFilter"
 	InproxyBrokerMatcherPrioritizeProxiesMinVersion    = "InproxyBrokerMatcherPrioritizeProxiesMinVersion"
@@ -1061,6 +1062,7 @@ var defaultParameters = map[string]struct {
 	InproxyBrokerMatcherOfferLimitEntryCount:           {value: 10, minimum: 0, flags: serverSideOnly},
 	InproxyBrokerMatcherOfferRateLimitQuantity:         {value: 50, minimum: 0, flags: serverSideOnly},
 	InproxyBrokerMatcherOfferRateLimitInterval:         {value: 1 * time.Minute, minimum: time.Duration(0), flags: serverSideOnly},
+	InproxyBrokerMatcherOfferMinimumDeadline:           {value: 1 * time.Second, minimum: time.Duration(0), flags: serverSideOnly},
 	InproxyBrokerMatcherPrioritizeProxiesProbability:   {value: 1.0, minimum: 0.0, flags: serverSideOnly},
 	InproxyBrokerMatcherPrioritizeProxiesFilter:        {value: KeyStrings{}, flags: serverSideOnly},
 	InproxyBrokerMatcherPrioritizeProxiesMinVersion:    {value: 0, minimum: 0, flags: serverSideOnly},

psiphon/server/meek.go

@@ -1895,6 +1895,7 @@ func (server *MeekServer) inproxyReloadTactics() error {
 		p.Int(parameters.InproxyBrokerMatcherOfferLimitEntryCount),
 		p.Int(parameters.InproxyBrokerMatcherOfferRateLimitQuantity),
 		p.Duration(parameters.InproxyBrokerMatcherOfferRateLimitInterval),
+		p.Duration(parameters.InproxyBrokerMatcherOfferMinimumDeadline),
 		p.Int(parameters.InproxyMaxCompartmentIDListLength),
 		p.Int(parameters.InproxyBrokerDSLRequestRateLimitQuantity),
 		p.Duration(parameters.InproxyBrokerDSLRequestRateLimitInterval))

psiphon/server/pb/psiphond/inproxy_broker.pb.go

@@ -63,6 +63,7 @@ type InproxyBroker struct {
 	TimedOut                      *bool                  `protobuf:"varint,136,opt,name=timed_out,json=timedOut,proto3,oneof" json:"timed_out,omitempty"`
 	MeekServerHttpVersion         *string                `protobuf:"bytes,137,opt,name=meek_server_http_version,json=meekServerHttpVersion,proto3,oneof" json:"meek_server_http_version,omitempty"`
 	PendingAnswersSize            *int64                 `protobuf:"varint,138,opt,name=pending_answers_size,json=pendingAnswersSize,proto3,oneof" json:"pending_answers_size,omitempty"`
+	OfferDeadline                 *int64                 `protobuf:"varint,139,opt,name=offer_deadline,json=offerDeadline,proto3,oneof" json:"offer_deadline,omitempty"`
 	unknownFields                 protoimpl.UnknownFields
 	sizeCache                     protoimpl.SizeCache
 }
@@ -377,11 +378,18 @@ func (x *InproxyBroker) GetPendingAnswersSize() int64 {
 	return 0
 }
 
+func (x *InproxyBroker) GetOfferDeadline() int64 {
+	if x != nil && x.OfferDeadline != nil {
+		return *x.OfferDeadline
+	}
+	return 0
+}
+
 var File_ca_psiphon_psiphond_inproxy_broker_proto protoreflect.FileDescriptor
 
 const file_ca_psiphon_psiphond_inproxy_broker_proto_rawDesc = "" +
 	"\n" +
-	"(ca.psiphon.psiphond/inproxy_broker.proto\x12\x13ca.psiphon.psiphond\x1a%ca.psiphon.psiphond/base_params.proto\"\x93\x16\n" +
+	"(ca.psiphon.psiphond/inproxy_broker.proto\x12\x13ca.psiphon.psiphond\x1a%ca.psiphon.psiphond/base_params.proto\"\xd3\x16\n" +
 	"\rInproxyBroker\x12E\n" +
 	"\vbase_params\x18\x01 \x01(\v2\x1f.ca.psiphon.psiphond.BaseParamsH\x00R\n" +
 	"baseParams\x88\x01\x01\x12=\n" +
@@ -426,7 +434,8 @@ const file_ca_psiphon_psiphond_inproxy_broker_proto_rawDesc = "" +
 	"\x12stored_tactics_tag\x18\x87\x01 \x01(\tH R\x10storedTacticsTag\x88\x01\x01\x12!\n" +
 	"\ttimed_out\x18\x88\x01 \x01(\bH!R\btimedOut\x88\x01\x01\x12=\n" +
 	"\x18meek_server_http_version\x18\x89\x01 \x01(\tH\"R\x15meekServerHttpVersion\x88\x01\x01\x126\n" +
-	"\x14pending_answers_size\x18\x8a\x01 \x01(\x03H#R\x12pendingAnswersSize\x88\x01\x01B\x0e\n" +
+	"\x14pending_answers_size\x18\x8a\x01 \x01(\x03H#R\x12pendingAnswersSize\x88\x01\x01\x12+\n" +
+	"\x0eoffer_deadline\x18\x8b\x01 \x01(\x03H$R\rofferDeadline\x88\x01\x01B\x0e\n" +
 	"\f_base_paramsB\x1b\n" +
 	"\x19_announcement_match_indexB\x1a\n" +
 	"\x18_announcement_queue_sizeB\x0f\n" +
@@ -465,7 +474,8 @@ const file_ca_psiphon_psiphond_inproxy_broker_proto_rawDesc = "" +
 	"\n" +
 	"_timed_outB\x1b\n" +
 	"\x19_meek_server_http_versionB\x17\n" +
-	"\x15_pending_answers_sizeBHZFgithub.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/server/pb/psiphondb\x06proto3"
+	"\x15_pending_answers_sizeB\x11\n" +
+	"\x0f_offer_deadlineBHZFgithub.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/server/pb/psiphondb\x06proto3"
 
 var (
 	file_ca_psiphon_psiphond_inproxy_broker_proto_rawDescOnce sync.Once

psiphon/server/proto/ca.psiphon.psiphond/inproxy_broker.proto

@@ -50,4 +50,5 @@ message InproxyBroker {
     optional bool timed_out = 136;
     optional string meek_server_http_version = 137;
     optional int64 pending_answers_size = 138;
+    optional int64 offer_deadline = 139;
 }