Merge pull request #763 from rod-hynes/inproxy-enhancements

In-proxy proxy enhancements

MobileLibrary/Android/PsiphonTunnel/PsiphonTunnel.java

@@ -135,12 +135,14 @@ public class PsiphonTunnel {
          * Called when tunnel-core reports proxy usage statistics.
          * By default onInproxyProxyActivity is disabled. Enable it by setting
          * EmitInproxyProxyActivity to true in the Psiphon config.
+         * @param announcing Number of new clients the proxy is accepting.
          * @param connectingClients Number of clients connecting to the proxy.
          * @param connectedClients Number of clients currently connected to the proxy.
          * @param bytesUp  Bytes uploaded through the proxy since the last report.
          * @param bytesDown Bytes downloaded through the proxy since the last report.
          */
-        default void onInproxyProxyActivity(int connectingClients, int connectedClients,long bytesUp, long bytesDown) {}
+        default void onInproxyProxyActivity(
+            int announcing, int connectingClients, int connectedClients,long bytesUp, long bytesDown) {}
         /**
          * Called when tunnel-core reports connected server region information.
          * @param region The server region received.
@@ -923,6 +925,7 @@ public class PsiphonTunnel {
             } else if (noticeType.equals("InproxyProxyActivity")) {
                 JSONObject data = notice.getJSONObject("data");
                 mHostService.onInproxyProxyActivity(
+                        data.getInt("announcing"),
                         data.getInt("connectingClients"),
                         data.getInt("connectedClients"),
                         data.getLong("bytesUp"),

MobileLibrary/iOS/PsiphonTunnel/PsiphonTunnel/PsiphonTunnel.h

@@ -312,12 +312,14 @@ followed by a tunnel-core shutdown.
  Called when tunnel-core reports in-proxy usage statistics
  By default onInproxyProxyActivity is disabled. Enable it by setting
  EmitInproxyProxyActivity to true in the Psiphon config.
+ @param announcing Number of new clients the proxy is accepting.
  @param connectingClients Number of clients connecting to the proxy.
  @param connectedClients Number of clients currently connected to the proxy.
  @param bytesUp Bytes uploaded through the proxy since the last report.
  @param bytesDown Bytes downloaded through the proxy since the last report.
  */
-- (void)onInproxyProxyActivity:(int)connectingClients
+- (void)onInproxyProxyActivity:(int)announcing
+              connectingClients:(int)connectingClients
               connectedClients:(int)connectedClients
                        bytesUp:(long)bytesUp
                      bytesDown:(long)bytesDown;

MobileLibrary/iOS/PsiphonTunnel/PsiphonTunnel/PsiphonTunnel.m

@@ -1191,17 +1191,24 @@ typedef NS_ERROR_ENUM(PsiphonTunnelErrorDomain, PsiphonTunnelErrorCode) {
         }
     }
     else if ([noticeType isEqualToString:@"InproxyProxyActivity"]) {
+        id announcing = [notice valueForKeyPath:@"data.announcing"];
         id connectingClients = [notice valueForKeyPath:@"data.connectingClients"];
         id connectedClients = [notice valueForKeyPath:@"data.connectedClients"];
         id bytesUp = [notice valueForKeyPath:@"data.bytesUp"];
         id bytesDown = [notice valueForKeyPath:@"data.bytesDown"];
-        if (![connectingClients isKindOfClass:[NSNumber class]] || ![connectedClients isKindOfClass:[NSNumber class]] || ![bytesUp isKindOfClass:[NSNumber class]] || ![bytesDown isKindOfClass:[NSNumber class]]) {
+        if (![announcing isKindOfClass:[NSNumber class]] ||
+            ![connectingClients isKindOfClass:[NSNumber class]] ||
+            ![connectedClients isKindOfClass:[NSNumber class]] ||
+            ![bytesUp isKindOfClass:[NSNumber class]] ||
+            ![bytesDown isKindOfClass:[NSNumber class]]) {
             [self logMessage:[NSString stringWithFormat: @"InproxyProxyActivity notice has invalid data types: %@", noticeJSON]];
             return;
         }
-        if ([self.tunneledAppDelegate respondsToSelector:@selector(onInproxyProxyActivity:connectedClients:bytesUp:bytesDown:)]) {
+        if ([self.tunneledAppDelegate respondsToSelector:@selector(onInproxyProxyActivity:connectingClients:connectedClients:bytesUp:bytesDown:)]) {
             dispatch_sync(self->callbackQueue, ^{
-                [self.tunneledAppDelegate onInproxyProxyActivity:[connectingClients intValue] connectedClients:[connectedClients intValue] bytesUp:[bytesUp longValue] bytesDown:[bytesDown longValue]];
+                [self.tunneledAppDelegate onInproxyProxyActivity: [announcing intValue]
+                    connectingClients:[connectingClients intValue] connectedClients:[connectedClients intValue]
+                    bytesUp:[bytesUp longValue] bytesDown:[bytesDown longValue]];
             });
         }
     }

psiphon/common/inproxy/inproxy_test.go

@@ -556,12 +556,14 @@ func runTestInproxy(doMustUpgrade bool) error {
 			LimitUpstreamBytesPerSecond:   bytesToSend / targetElapsedSeconds,
 			LimitDownstreamBytesPerSecond: bytesToSend / targetElapsedSeconds,
 
-			ActivityUpdater: func(connectingClients int32, connectedClients int32,
+			ActivityUpdater: func(
+				announcing int32,
+				connectingClients int32, connectedClients int32,
 				bytesUp int64, bytesDown int64, bytesDuration time.Duration) {
 
-				fmt.Printf("[%s][%s] ACTIVITY: %d connecting, %d connected, %d up, %d down\n",
+				fmt.Printf("[%s][%s] ACTIVITY: %d announcing, %d connecting, %d connected, %d up, %d down\n",
 					time.Now().UTC().Format(time.RFC3339), name,
-					connectingClients, connectedClients, bytesUp, bytesDown)
+					announcing, connectingClients, connectedClients, bytesUp, bytesDown)
 			},
 
 			MustUpgrade: func() {

psiphon/common/inproxy/proxy.go

@@ -50,11 +50,13 @@ type Proxy struct {
 	bytesDown         atomic.Int64
 	peakBytesUp       atomic.Int64
 	peakBytesDown     atomic.Int64
-	connectingClients int32
-	connectedClients  int32
+	announcing        atomic.Int32
+	connectingClients atomic.Int32
+	connectedClients  atomic.Int32
 
 	config                *ProxyConfig
 	activityUpdateWrapper *activityUpdateWrapper
+	lastAnnouncing        int32
 	lastConnectingClients int32
 	lastConnectedClients  int32
 
@@ -65,6 +67,10 @@ type Proxy struct {
 	nextAnnounceMutex        sync.Mutex
 	nextAnnounceBrokerClient *BrokerClient
 	nextAnnounceNotBefore    time.Time
+
+	useReducedSettings bool
+	reducedStartMinute int
+	reducedEndMinute   int
 }
 
 // TODO: add PublicNetworkAddress/ListenNetworkAddress to facilitate manually
@@ -144,17 +150,49 @@ type ProxyConfig struct {
 	// for a single client. When 0, there is no limit.
 	LimitDownstreamBytesPerSecond int
 
+	// ReducedStartTime specifies the local time of day (HH:MM, 24-hour, UTC)
+	// at which reduced client settings begin.
+	ReducedStartTime string
+
+	// ReducedEndTime specifies the local time of day (HH:MM, 24-hour, UTC) at
+	// which reduced client settings end.
+	ReducedEndTime string
+
+	// ReducedMaxClients specifies the maximum number of clients that are
+	// allowed to connect to the proxy during the reduced time range.
+	//
+	// Clients connected when the reduced settings begin will not be
+	// disconnected.
+	ReducedMaxClients int
+
+	// ReducedLimitUpstreamBytesPerSecond limits the upstream data transfer
+	// rate for a single client during the reduced time range. When 0,
+	// LimitUpstreamBytesPerSecond is the limit.
+	//
+	// Rates for clients already connected when the reduced settings begin or
+	// end will not change.
+	ReducedLimitUpstreamBytesPerSecond int
+
+	// ReducedLimitDownstreamBytesPerSecond limits the downstream data
+	// transfer rate for a single client during the reduced time range. When
+	// 0, LimitDownstreamBytesPerSecond is the limit.
+	//
+	// Rates for clients already connected when the reduced settings begin or
+	// end will not change.
+	ReducedLimitDownstreamBytesPerSecond int
+
 	// ActivityUpdater specifies an ActivityUpdater for activity associated
 	// with this proxy.
 	ActivityUpdater ActivityUpdater
 }
 
-// ActivityUpdater is a callback that is invoked when clients connect and
-// disconnect and periodically with data transfer updates (unless idle). This
-// callback may be used to update an activity UI. This callback should post
-// this data to another thread or handler and return immediately and not
-// block on UI updates.
+// ActivityUpdater is a callback that is invoked when the proxy announces
+// availability, when clients connect and disconnect, and periodically with
+// data transfer updates (unless idle). This callback may be used to update
+// an activity UI. This callback should post this data to another thread or
+// handler and return immediately and not block on UI updates.
 type ActivityUpdater func(
+	announcing int32,
 	connectingClients int32,
 	connectedClients int32,
 	bytesUp int64,
@@ -172,6 +210,34 @@ func NewProxy(config *ProxyConfig) (*Proxy, error) {
 		config: config,
 	}
 
+	if config.ReducedStartTime != "" ||
+		config.ReducedEndTime != "" ||
+		config.ReducedMaxClients > 0 {
+
+		startMinute, err := common.ParseTimeOfDayMinutes(config.ReducedStartTime)
+		if err != nil {
+			return nil, errors.Tracef("invalid ReducedStartTime: %v", err)
+		}
+
+		endMinute, err := common.ParseTimeOfDayMinutes(config.ReducedEndTime)
+		if err != nil {
+			return nil, errors.Tracef("invalid ReducedEndTime: %v", err)
+		}
+
+		if startMinute == endMinute {
+			return nil, errors.TraceNew("invalid ReducedStartTime/ReducedEndTime")
+		}
+
+		if config.ReducedMaxClients <= 0 ||
+			config.ReducedMaxClients > config.MaxClients {
+			return nil, errors.TraceNew("invalid ReducedMaxClients")
+		}
+
+		p.useReducedSettings = true
+		p.reducedStartMinute = startMinute
+		p.reducedEndMinute = endMinute
+	}
+
 	p.activityUpdateWrapper = &activityUpdateWrapper{p: p}
 
 	return p, nil
@@ -206,12 +272,39 @@ func (p *Proxy) Run(ctx context.Context) {
 
 	proxyWaitGroup := new(sync.WaitGroup)
 
+	// Capture activity updates every second, which is the required frequency
+	// for PeakUp/DownstreamBytesPerSecond. This is also a reasonable
+	// frequency for invoking the ActivityUpdater and updating UI widgets.
+
+	proxyWaitGroup.Add(1)
+	go func() {
+		defer proxyWaitGroup.Done()
+
+		p.lastAnnouncing = 0
+		p.lastConnectingClients = 0
+		p.lastConnectedClients = 0
+
+		activityUpdatePeriod := 1 * time.Second
+		ticker := time.NewTicker(activityUpdatePeriod)
+		defer ticker.Stop()
+
+	loop:
+		for {
+			select {
+			case <-ticker.C:
+				p.activityUpdate(activityUpdatePeriod)
+			case <-ctx.Done():
+				break loop
+			}
+		}
+	}()
+
 	// Launch the first proxy worker, passing a signal to be triggered once
 	// the very first announcement round trip is complete. The first round
 	// trip is awaited so that:
 	//
 	// - The first announce response will arrive with any new tactics,
-	//   which may be applied before launching additions workers.
+	//   which may be applied before launching additional workers.
 	//
 	// - The first worker gets no announcement delay and is also guaranteed to
 	//   be the shared session establisher. Since the announcement delays are
@@ -228,7 +321,7 @@ func (p *Proxy) Run(ctx context.Context) {
 	proxyWaitGroup.Add(1)
 	go func() {
 		defer proxyWaitGroup.Done()
-		p.proxyClients(ctx, signalFirstAnnounceDone)
+		p.proxyClients(ctx, signalFirstAnnounceDone, false)
 	}()
 
 	select {
@@ -240,80 +333,55 @@ func (p *Proxy) Run(ctx context.Context) {
 	// Launch the remaining workers.
 
 	for i := 0; i < p.config.MaxClients-1; i++ {
-		proxyWaitGroup.Add(1)
-		go func() {
-			defer proxyWaitGroup.Done()
-			p.proxyClients(ctx, nil)
-		}()
-	}
 
-	// Capture activity updates every second, which is the required frequency
-	// for PeakUp/DownstreamBytesPerSecond. This is also a reasonable
-	// frequency for invoking the ActivityUpdater and updating UI widgets.
-
-	p.lastConnectingClients = 0
-	p.lastConnectedClients = 0
+		// When reduced settings are in effect, a subset of workers will pause
+		// during the reduced time period. Since ReducedMaxClients > 0 the
+		// first proxy worker is never paused.
+		workerNum := i + 1
+		reducedPause := p.useReducedSettings &&
+			workerNum >= p.config.ReducedMaxClients
 
-	activityUpdatePeriod := 1 * time.Second
-	ticker := time.NewTicker(activityUpdatePeriod)
-	defer ticker.Stop()
-
-loop:
-	for {
-		select {
-		case <-ticker.C:
-			p.activityUpdate(activityUpdatePeriod)
-		case <-ctx.Done():
-			break loop
-		}
+		proxyWaitGroup.Add(1)
+		go func(reducedPause bool) {
+			defer proxyWaitGroup.Done()
+			p.proxyClients(ctx, nil, reducedPause)
+		}(reducedPause)
 	}
 
 	proxyWaitGroup.Wait()
 }
 
-// getAnnounceDelayParameters is a helper that fetches the proxy announcement
-// delay parameters from the current broker client.
-//
-// getAnnounceDelayParameters is used to configure a delay when
-// proxyOneClient fails. As having no broker clients is a possible
-// proxyOneClient failure case, GetBrokerClient errors are ignored here and
-// defaults used in that case.
-func (p *Proxy) getAnnounceDelayParameters() (time.Duration, time.Duration, float64) {
-	brokerClient, err := p.config.GetBrokerClient()
-	if err != nil {
-		return proxyAnnounceDelay, proxyAnnounceMaxBackoffDelay, proxyAnnounceDelayJitter
-	}
-	brokerCoordinator := brokerClient.GetBrokerDialCoordinator()
-	return common.ValueOrDefault(brokerCoordinator.AnnounceDelay(), proxyAnnounceDelay),
-		common.ValueOrDefault(brokerCoordinator.AnnounceMaxBackoffDelay(), proxyAnnounceMaxBackoffDelay),
-		common.ValueOrDefault(brokerCoordinator.AnnounceDelayJitter(), proxyAnnounceDelayJitter)
-
-}
-
 func (p *Proxy) activityUpdate(period time.Duration) {
 
-	connectingClients := atomic.LoadInt32(&p.connectingClients)
-	connectedClients := atomic.LoadInt32(&p.connectedClients)
+	// Concurrency: activityUpdate is called by only the single goroutine
+	// created in Run.
+
+	announcing := p.announcing.Load()
+	connectingClients := p.connectingClients.Load()
+	connectedClients := p.connectedClients.Load()
 	bytesUp := p.bytesUp.Swap(0)
 	bytesDown := p.bytesDown.Swap(0)
 
 	greaterThanSwapInt64(&p.peakBytesUp, bytesUp)
 	greaterThanSwapInt64(&p.peakBytesDown, bytesDown)
 
-	clientsChanged := connectingClients != p.lastConnectingClients ||
+	stateChanged := announcing != p.lastAnnouncing ||
+		connectingClients != p.lastConnectingClients ||
 		connectedClients != p.lastConnectedClients
 
+	p.lastAnnouncing = announcing
 	p.lastConnectingClients = connectingClients
 	p.lastConnectedClients = connectedClients
 
-	if !clientsChanged &&
+	if !stateChanged &&
 		bytesUp == 0 &&
 		bytesDown == 0 {
-		// Skip the activity callback on idle bytes or no change in client counts.
+		// Skip the activity callback on idle bytes or no change in worker state.
 		return
 	}
 
 	p.config.ActivityUpdater(
+		announcing,
 		connectingClients,
 		connectedClients,
 		bytesUp,
@@ -333,8 +401,93 @@ func greaterThanSwapInt64(addr *atomic.Int64, new int64) bool {
 	return false
 }
 
+func (p *Proxy) isReducedUntil() (int, time.Time) {
+	if !p.useReducedSettings {
+		return p.config.MaxClients, time.Time{}
+	}
+
+	now := time.Now().UTC()
+	minute := now.Hour()*60 + now.Minute()
+
+	isReduced := false
+	if p.reducedStartMinute < p.reducedEndMinute {
+		isReduced = minute >= p.reducedStartMinute && minute < p.reducedEndMinute
+	} else {
+		isReduced = minute >= p.reducedStartMinute || minute < p.reducedEndMinute
+	}
+
+	if !isReduced {
+		return p.config.MaxClients, time.Time{}
+	}
+
+	endHour := p.reducedEndMinute / 60
+	endMinute := p.reducedEndMinute % 60
+	endTime := time.Date(
+		now.Year(),
+		now.Month(),
+		now.Day(),
+		endHour,
+		endMinute,
+		0,
+		0,
+		now.Location(),
+	)
+	if !endTime.After(now) {
+		endTime = endTime.AddDate(0, 0, 1)
+	}
+	return p.config.ReducedMaxClients, endTime
+}
+
+func (p *Proxy) getLimits() (int, common.RateLimits) {
+
+	rateLimits := common.RateLimits{
+		ReadBytesPerSecond:  int64(p.config.LimitUpstreamBytesPerSecond),
+		WriteBytesPerSecond: int64(p.config.LimitDownstreamBytesPerSecond),
+	}
+
+	maxClients, reducedUntil := p.isReducedUntil()
+	if !reducedUntil.IsZero() {
+
+		upstream := p.config.ReducedLimitUpstreamBytesPerSecond
+		if upstream == 0 {
+			upstream = p.config.LimitUpstreamBytesPerSecond
+		}
+
+		downstream := p.config.ReducedLimitDownstreamBytesPerSecond
+		if downstream == 0 {
+			downstream = p.config.LimitDownstreamBytesPerSecond
+		}
+
+		rateLimits = common.RateLimits{
+			ReadBytesPerSecond:  int64(upstream),
+			WriteBytesPerSecond: int64(downstream),
+		}
+	}
+
+	return maxClients, rateLimits
+}
+
+// getAnnounceDelayParameters is a helper that fetches the proxy announcement
+// delay parameters from the current broker client.
+//
+// getAnnounceDelayParameters is used to configure a delay when
+// proxyOneClient fails. As having no broker clients is a possible
+// proxyOneClient failure case, GetBrokerClient errors are ignored here and
+// defaults used in that case.
+func (p *Proxy) getAnnounceDelayParameters() (time.Duration, time.Duration, float64) {
+	brokerClient, err := p.config.GetBrokerClient()
+	if err != nil {
+		return proxyAnnounceDelay, proxyAnnounceMaxBackoffDelay, proxyAnnounceDelayJitter
+	}
+	brokerCoordinator := brokerClient.GetBrokerDialCoordinator()
+	return common.ValueOrDefault(brokerCoordinator.AnnounceDelay(), proxyAnnounceDelay),
+		common.ValueOrDefault(brokerCoordinator.AnnounceMaxBackoffDelay(), proxyAnnounceMaxBackoffDelay),
+		common.ValueOrDefault(brokerCoordinator.AnnounceDelayJitter(), proxyAnnounceDelayJitter)
+
+}
+
 func (p *Proxy) proxyClients(
-	ctx context.Context, signalAnnounceDone func()) {
+	ctx context.Context, signalAnnounceDone func(), reducedPause bool) {
 
 	// Proxy one client, repeating until ctx is done.
 	//
@@ -381,6 +534,31 @@ func (p *Proxy) proxyClients(
 			break
 		}
 
+		// Pause designated workers during the reduced time range. In-flight
+		// announces are not interrupted and connected clients are not
+		// disconnected, so there is a gradual transition into reduced mode.
+
+		if reducedPause {
+			_, reducedUntil := p.isReducedUntil()
+			if !reducedUntil.IsZero() {
+
+				pauseDuration := time.Until(reducedUntil)
+				p.config.Logger.WithTraceFields(common.LogFields{
+					"duration": pauseDuration.String(),
+				}).Info("pause worker")
+
+				timer := time.NewTimer(pauseDuration)
+				select {
+				case <-timer.C:
+				case <-ctx.Done():
+				}
+				timer.Stop()
+				if ctx.Err() != nil {
+					break
+				}
+			}
+		}
+
 		if time.Since(startLogSampleTime) >= proxyAnnounceLogSamplePeriod {
 			logAnnounceCount = proxyAnnounceLogSampleSize
 			logErrorsCount = proxyAnnounceLogSampleSize
@@ -590,6 +768,8 @@ func (p *Proxy) proxyOneClient(
 	// for tactics.
 	checkTactics := signalAnnounceDone != nil
 
+	maxClients, rateLimits := p.getLimits()
+
 	// Get the base Psiphon API parameters and additional proxy metrics,
 	// including performance information, which is sent to the broker in the
 	// proxy announcment.
@@ -601,7 +781,7 @@ func (p *Proxy) proxyOneClient(
 	// with the original network ID.
 
 	metrics, tacticsNetworkID, compressTactics, err := p.getMetrics(
-		checkTactics, brokerCoordinator, webRTCCoordinator)
+		checkTactics, brokerCoordinator, webRTCCoordinator, maxClients, rateLimits)
 	if err != nil {
 		return backOff, errors.Trace(err)
 	}
@@ -652,6 +832,9 @@ func (p *Proxy) proxyOneClient(
 	//
 	// ProxyAnnounce applies an additional request timeout to facilitate
 	// long-polling.
+
+	p.announcing.Add(1)
+
 	announceStartTime := time.Now()
 	personalCompartmentIDs := brokerCoordinator.PersonalCompartmentIDs()
 	announceResponse, err := brokerClient.ProxyAnnounce(
@@ -668,6 +851,9 @@ func (p *Proxy) proxyOneClient(
 			"elapsedTime": time.Since(announceStartTime).String(),
 		}).Info("announcement request")
 	}
+
+	p.announcing.Add(-1)
+
 	if err != nil {
 		return backOff, errors.Trace(err)
 	}
@@ -743,11 +929,11 @@ func (p *Proxy) proxyOneClient(
 
 	// For activity updates, indicate that a client connection is now underway.
 
-	atomic.AddInt32(&p.connectingClients, 1)
+	p.connectingClients.Add(1)
 	connected := false
 	defer func() {
 		if !connected {
-			atomic.AddInt32(&p.connectingClients, -1)
+			p.connectingClients.Add(-1)
 		}
 	}()
 
@@ -881,10 +1067,10 @@ func (p *Proxy) proxyOneClient(
 	// For activity updates, indicate that a client connection is established.
 
 	connected = true
-	atomic.AddInt32(&p.connectingClients, -1)
-	atomic.AddInt32(&p.connectedClients, 1)
+	p.connectingClients.Add(-1)
+	p.connectedClients.Add(1)
 	defer func() {
-		atomic.AddInt32(&p.connectedClients, -1)
+		p.connectedClients.Add(-1)
 	}()
 
 	// Throttle the relay connection.
@@ -895,14 +1081,15 @@ func (p *Proxy) proxyOneClient(
 	// generated by dividing the limit by MaxClients. This approach favors
 	// performance stability: each client gets the same throttling limits
 	// regardless of how many other clients are connected.
+	//
+	// Rate limits are applied only when a client connection is established;
+	// connected clients retain their initial limits even when reduced time
+	// starts or ends.
 
 	destinationConn = common.NewThrottledConn(
 		destinationConn,
 		announceResponse.NetworkProtocol.IsStream(),
-		common.RateLimits{
-			ReadBytesPerSecond:  int64(p.config.LimitUpstreamBytesPerSecond),
-			WriteBytesPerSecond: int64(p.config.LimitDownstreamBytesPerSecond),
-		})
+		rateLimits)
 
 	// Hook up bytes transferred counting for activity updates.
 
@@ -1012,7 +1199,9 @@ func (p *Proxy) proxyOneClient(
 func (p *Proxy) getMetrics(
 	includeTacticsParameters bool,
 	brokerCoordinator BrokerDialCoordinator,
-	webRTCCoordinator WebRTCDialCoordinator) (
+	webRTCCoordinator WebRTCDialCoordinator,
+	maxClients int,
+	rateLimits common.RateLimits) (
 	*ProxyMetrics, string, bool, error) {
 
 	// tacticsNetworkID records the exact network ID that corresponds to the
@@ -1040,11 +1229,11 @@ func (p *Proxy) getMetrics(
 		ProtocolVersion:               LatestProtocolVersion,
 		NATType:                       webRTCCoordinator.NATType(),
 		PortMappingTypes:              webRTCCoordinator.PortMappingTypes(),
-		MaxClients:                    int32(p.config.MaxClients),
-		ConnectingClients:             atomic.LoadInt32(&p.connectingClients),
-		ConnectedClients:              atomic.LoadInt32(&p.connectedClients),
-		LimitUpstreamBytesPerSecond:   int64(p.config.LimitUpstreamBytesPerSecond),
-		LimitDownstreamBytesPerSecond: int64(p.config.LimitDownstreamBytesPerSecond),
+		MaxClients:                    int32(maxClients),
+		ConnectingClients:             p.connectingClients.Load(),
+		ConnectedClients:              p.connectedClients.Load(),
+		LimitUpstreamBytesPerSecond:   rateLimits.ReadBytesPerSecond,
+		LimitDownstreamBytesPerSecond: rateLimits.WriteBytesPerSecond,
 		PeakUpstreamBytesPerSecond:    p.peakBytesUp.Load(),
 		PeakDownstreamBytesPerSecond:  p.peakBytesDown.Load(),
 	}, tacticsNetworkID, compressTactics, nil

psiphon/common/inproxy/reduced_test.go

@@ -0,0 +1,111 @@
+/*
+ * Copyright (c) 2026, 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 inproxy
+
+import (
+	"testing"
+	"time"
+
+	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/errors"
+)
+
+func TestReduced(t *testing.T) {
+	err := runTestReduced()
+	if err != nil {
+		t.Error(errors.Trace(err).Error())
+	}
+}
+
+func runTestReduced() error {
+
+	now := time.Now().UTC()
+	minuteOfDay := now.Hour()*60 + now.Minute()
+
+	addMinutes := func(minute, delta int) int {
+		m := (minute + delta) % (24 * 60)
+		if m < 0 {
+			m += 24 * 60
+		}
+		return m
+	}
+
+	// Test: inside reduced period
+
+	start := addMinutes(minuteOfDay, -60)
+	end := addMinutes(minuteOfDay, 60)
+
+	config := &ProxyConfig{
+		MaxClients:                           10,
+		ReducedMaxClients:                    5,
+		LimitUpstreamBytesPerSecond:          100,
+		LimitDownstreamBytesPerSecond:        200,
+		ReducedLimitUpstreamBytesPerSecond:   10,
+		ReducedLimitDownstreamBytesPerSecond: 20,
+	}
+
+	config.ReducedStartTime = time.Unix(int64(start*60), 0).UTC().Format("15:04")
+	config.ReducedEndTime = time.Unix(int64(end*60), 0).UTC().Format("15:04")
+
+	p, err := NewProxy(config)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	maxClients1, until := p.isReducedUntil()
+	maxClients2, limits := p.getLimits()
+
+	if maxClients1 != 5 || maxClients2 != 5 {
+		return errors.TraceNew("unexpected maxClients")
+	}
+	if until.IsZero() || time.Until(until) <= 0 {
+		return errors.TraceNew("unexpected until")
+	}
+	if limits.ReadBytesPerSecond != 10 || limits.WriteBytesPerSecond != 20 {
+		return errors.TraceNew("unexpected rate limits")
+	}
+
+	// Test: outside reduced period
+
+	start = addMinutes(minuteOfDay, 60)
+	end = addMinutes(minuteOfDay, 120)
+
+	config.ReducedStartTime = time.Unix(int64(start*60), 0).UTC().Format("15:04")
+	config.ReducedEndTime = time.Unix(int64(end*60), 0).UTC().Format("15:04")
+
+	p, err = NewProxy(config)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	maxClients1, until = p.isReducedUntil()
+	maxClients2, limits = p.getLimits()
+
+	if maxClients1 != 10 || maxClients2 != 10 {
+		return errors.TraceNew("unexpected maxClients")
+	}
+	if !until.IsZero() {
+		return errors.TraceNew("unexpected until")
+	}
+	if limits.ReadBytesPerSecond != 100 || limits.WriteBytesPerSecond != 200 {
+		return errors.TraceNew("unexpected rate limits")
+	}
+
+	return nil
+}

psiphon/common/utils.go

@@ -133,6 +133,16 @@ func TruncateTimestampToHour(timestamp string) string {
 	return t.Truncate(1 * time.Hour).Format(time.RFC3339)
 }
 
+// ParseTimeOfDayMinutes parses a time of day in HH:MM 24-hour format and
+// returns the number of minutes since midnight.
+func ParseTimeOfDayMinutes(value string) (int, error) {
+	t, err := time.Parse("15:04", value)
+	if err != nil {
+		return 0, errors.Trace(err)
+	}
+	return t.Hour()*60 + t.Minute(), nil
+}
+
 const (
 	CompressionNone = int32(0)
 	CompressionZlib = int32(1)

psiphon/config.go

@@ -665,6 +665,38 @@ type Config struct {
 	// transfer rate limit for each proxied client. When 0, there is no limit.
 	InproxyLimitDownstreamBytesPerSecond int `json:",omitempty"`
 
+	// InproxyReducedStartTime specifies the local time of day(HH:MM, 24-hour,
+	// UTC) at which reduced in-proxy settings begin.
+	InproxyReducedStartTime string `json:",omitempty"`
+
+	// InproxyReducedEndTime specifies the local time of day (HH:MM, 24-hour,
+	// UTC) at which reduced in-proxy settings end.
+	InproxyReducedEndTime string `json:",omitempty"`
+
+	// InproxyReducedMaxClients specifies the maximum number of in-proxy
+	// clients to be proxied concurrently during the reduced time range.
+	// When set, must be > 0 and <= InproxyMaxClients.
+	//
+	// Clients connected when the reduced settings begin will not be
+	// disconnected, so InproxyReducedMaxClients is a soft limit.
+	InproxyReducedMaxClients int `json:",omitempty"`
+
+	// InproxyReducedLimitUpstreamBytesPerSecond specifies the upstream byte
+	// transfer rate limit for each proxied client during the reduced time
+	// range. When 0, InproxyLimitUpstreamBytesPerSecond is the limit.
+	//
+	// Rates for clients already connected when the reduced settings begin or
+	// end will not change.
+	InproxyReducedLimitUpstreamBytesPerSecond int `json:",omitempty"`
+
+	// InproxyReducedLimitDownstreamBytesPerSecond specifies the downstream byte
+	// transfer rate limit for each proxied client during the reduced time
+	// range. When 0, InproxyLimitDownstreamBytesPerSecond is the limit.
+	//
+	// Rates for clients already connected when the reduced settings begin or
+	// end will not change.
+	InproxyReducedLimitDownstreamBytesPerSecond int `json:",omitempty"`
+
 	// 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
@@ -1490,8 +1522,46 @@ func (config *Config) Commit(migrateFromLegacyFields bool) error {
 		return errors.TraceNew("invalid ObfuscatedSSHAlgorithms")
 	}
 
-	if config.InproxyEnableProxy && config.InproxyMaxClients <= 0 {
-		return errors.TraceNew("invalid InproxyMaxClients")
+	if config.InproxyEnableProxy {
+
+		if config.InproxyMaxClients <= 0 {
+			return errors.TraceNew("invalid InproxyMaxClients")
+		}
+
+		if config.InproxyReducedStartTime != "" ||
+			config.InproxyReducedEndTime != "" ||
+			config.InproxyReducedMaxClients > 0 {
+
+			startMinute, err := common.ParseTimeOfDayMinutes(config.InproxyReducedStartTime)
+			if err != nil {
+				return errors.Tracef("invalid InproxyReducedStartTime: %v", err)
+			}
+
+			endMinute, err := common.ParseTimeOfDayMinutes(config.InproxyReducedEndTime)
+			if err != nil {
+				return errors.Tracef("invalid InproxyReducedEndTime: %v", err)
+			}
+
+			// Reduced all day is not a valid configuration.
+			if startMinute == endMinute {
+				return errors.TraceNew("invalid InproxyReducedStartTime/InproxyReducedEndTime")
+			}
+
+			if config.InproxyReducedMaxClients <= 0 ||
+				config.InproxyReducedMaxClients > config.InproxyMaxClients {
+				return errors.TraceNew("invalid InproxyReducedMaxClients")
+			}
+
+			// InproxyReducedLimitUpstream/DownstreamBytesPerSecond don't necessarily
+			// need to be less than InproxyLimitUpstream/DownstreamBytesPerSecond.
+
+			if config.InproxyReducedLimitUpstreamBytesPerSecond == 0 {
+				config.InproxyReducedLimitUpstreamBytesPerSecond = config.InproxyLimitUpstreamBytesPerSecond
+			}
+			if config.InproxyReducedLimitDownstreamBytesPerSecond == 0 {
+				config.InproxyReducedLimitDownstreamBytesPerSecond = config.InproxyLimitDownstreamBytesPerSecond
+			}
+		}
 	}
 
 	if !config.DisableTunnels &&

psiphon/controller.go

@@ -3224,10 +3224,12 @@ func (controller *Controller) runInproxyProxy() {
 	debugLogging := controller.config.InproxyEnableWebRTCDebugLogging
 
 	var lastActivityNotice time.Time
+	var lastAnnouncing int32
 	var lastActivityConnectingClients, lastActivityConnectedClients int32
 	var lastActivityConnectingClientsTotal, lastActivityConnectedClientsTotal int32
 	var activityTotalBytesUp, activityTotalBytesDown int64
 	activityUpdater := func(
+		announcing int32,
 		connectingClients int32,
 		connectedClients int32,
 		bytesUp int64,
@@ -3244,13 +3246,15 @@ func (controller *Controller) runInproxyProxy() {
 		// activity display.
 
 		if controller.config.EmitInproxyProxyActivity &&
-			(bytesUp > 0 || bytesDown > 0) ||
-			connectingClients != lastActivityConnectingClients ||
-			connectedClients != lastActivityConnectedClients {
+			(bytesUp > 0 || bytesDown > 0 ||
+				announcing != lastAnnouncing ||
+				connectingClients != lastActivityConnectingClients ||
+				connectedClients != lastActivityConnectedClients) {
 
 			NoticeInproxyProxyActivity(
-				connectingClients, connectedClients, bytesUp, bytesDown)
+				announcing, connectingClients, connectedClients, bytesUp, bytesDown)
 
+			lastAnnouncing = announcing
 			lastActivityConnectingClients = connectingClients
 			lastActivityConnectedClients = connectedClients
 		}
@@ -3262,13 +3266,17 @@ func (controller *Controller) runInproxyProxy() {
 		// transferred since starting; in addition to the current number of
 		// connecting and connected clients, whenever that changes. This
 		// notice is for diagnostics.
+		//
+		// Changes in announcing count are frequent and don't trigger
+		// InproxyProxyTotalActivity; the current announcing count is
+		// recorded as a snapshot.
 
 		if lastActivityNotice.Add(activityNoticePeriod).Before(time.Now()) ||
 			connectingClients != lastActivityConnectingClientsTotal ||
 			connectedClients != lastActivityConnectedClientsTotal {
 
 			NoticeInproxyProxyTotalActivity(
-				connectingClients, connectedClients,
+				announcing, connectingClients, connectedClients,
 				activityTotalBytesUp, activityTotalBytesDown)
 			lastActivityNotice = time.Now()
 
@@ -3278,19 +3286,24 @@ func (controller *Controller) runInproxyProxy() {
 	}
 
 	config := &inproxy.ProxyConfig{
-		Logger:                        NoticeCommonLogger(debugLogging),
-		EnableWebRTCDebugLogging:      debugLogging,
-		WaitForNetworkConnectivity:    controller.inproxyWaitForNetworkConnectivity,
-		GetCurrentNetworkContext:      controller.getCurrentNetworkContext,
-		GetBrokerClient:               controller.inproxyGetProxyBrokerClient,
-		GetBaseAPIParameters:          controller.inproxyGetProxyAPIParameters,
-		MakeWebRTCDialCoordinator:     controller.inproxyMakeProxyWebRTCDialCoordinator,
-		HandleTacticsPayload:          controller.inproxyHandleProxyTacticsPayload,
-		MaxClients:                    controller.config.InproxyMaxClients,
-		LimitUpstreamBytesPerSecond:   controller.config.InproxyLimitUpstreamBytesPerSecond,
-		LimitDownstreamBytesPerSecond: controller.config.InproxyLimitDownstreamBytesPerSecond,
-		MustUpgrade:                   controller.config.OnInproxyMustUpgrade,
-		ActivityUpdater:               activityUpdater,
+		Logger:                               NoticeCommonLogger(debugLogging),
+		EnableWebRTCDebugLogging:             debugLogging,
+		WaitForNetworkConnectivity:           controller.inproxyWaitForNetworkConnectivity,
+		GetCurrentNetworkContext:             controller.getCurrentNetworkContext,
+		GetBrokerClient:                      controller.inproxyGetProxyBrokerClient,
+		GetBaseAPIParameters:                 controller.inproxyGetProxyAPIParameters,
+		MakeWebRTCDialCoordinator:            controller.inproxyMakeProxyWebRTCDialCoordinator,
+		HandleTacticsPayload:                 controller.inproxyHandleProxyTacticsPayload,
+		MaxClients:                           controller.config.InproxyMaxClients,
+		LimitUpstreamBytesPerSecond:          controller.config.InproxyLimitUpstreamBytesPerSecond,
+		LimitDownstreamBytesPerSecond:        controller.config.InproxyLimitDownstreamBytesPerSecond,
+		ReducedStartTime:                     controller.config.InproxyReducedStartTime,
+		ReducedEndTime:                       controller.config.InproxyReducedEndTime,
+		ReducedMaxClients:                    controller.config.InproxyReducedMaxClients,
+		ReducedLimitUpstreamBytesPerSecond:   controller.config.InproxyReducedLimitUpstreamBytesPerSecond,
+		ReducedLimitDownstreamBytesPerSecond: controller.config.InproxyReducedLimitDownstreamBytesPerSecond,
+		MustUpgrade:                          controller.config.OnInproxyMustUpgrade,
+		ActivityUpdater:                      activityUpdater,
 	}
 
 	proxy, err := inproxy.NewProxy(config)
@@ -3306,7 +3319,7 @@ func (controller *Controller) runInproxyProxy() {
 
 	// Emit one last NoticeInproxyProxyTotalActivity with the final byte counts.
 	NoticeInproxyProxyTotalActivity(
-		lastActivityConnectingClients, lastActivityConnectedClients,
+		lastAnnouncing, lastActivityConnectingClients, lastActivityConnectedClients,
 		activityTotalBytesUp, activityTotalBytesDown)
 
 	NoticeInfo("inproxy proxy: stopped")

psiphon/notice.go

@@ -1151,6 +1151,7 @@ func NoticeInproxyMustUpgrade() {
 // with EmitInproxyProxyActivity for functionality such as traffic display;
 // and this frequent notice is not intended to be included with feedback.
 func NoticeInproxyProxyActivity(
+	announcing int32,
 	connectingClients int32,
 	connectedClients int32,
 	bytesUp int64,
@@ -1158,6 +1159,7 @@ func NoticeInproxyProxyActivity(
 
 	singletonNoticeLogger.outputNotice(
 		"InproxyProxyActivity", noticeIsNotDiagnostic,
+		"announcing", announcing,
 		"connectingClients", connectingClients,
 		"connectedClients", connectedClients,
 		"bytesUp", bytesUp,
@@ -1168,6 +1170,7 @@ func NoticeInproxyProxyActivity(
 // transferred in total up to this point; in addition to current connection
 // status. This is a diagnostic notice.
 func NoticeInproxyProxyTotalActivity(
+	announcing int32,
 	connectingClients int32,
 	connectedClients int32,
 	totalBytesUp int64,
@@ -1175,6 +1178,7 @@ func NoticeInproxyProxyTotalActivity(
 
 	singletonNoticeLogger.outputNotice(
 		"InproxyProxyTotalActivity", noticeIsDiagnostic,
+		"announcing", announcing,
 		"connectingClients", connectingClients,
 		"connectedClients", connectedClients,
 		"totalBytesUp", totalBytesUp,