Merge pull request #738 from rod-hynes/check-prune

Add aggressive server entry prune check

psiphon/common/parameters/parameters.go

@@ -510,6 +510,13 @@ const (
 	NetworkIDCacheTTL                                  = "NetworkIDCacheTTL"
 	ServerDNSResolverCacheMaxSize                      = "ServerDNSResolverCacheMaxSize"
 	ServerDNSResolverCacheTTL                          = "ServerDNSResolverCacheTTL"
+	CheckServerEntryTagsEnabled                        = "CheckServerEntryTagsEnabled"
+	CheckServerEntryTagsPeriod                         = "CheckServerEntryTagsPeriod"
+	CheckServerEntryTagsRepeatRatio                    = "CheckServerEntryTagsRepeatRatio"
+	CheckServerEntryTagsRepeatMinimum                  = "CheckServerEntryTagsRepeatMinimum"
+	CheckServerEntryTagsMaxSendBytes                   = "CheckServerEntryTagsMaxSendBytes"
+	CheckServerEntryTagsMaxWorkTime                    = "CheckServerEntryTagsMaxWorkTime"
+	ServerEntryPruneDialPortNumberZero                 = "ServerEntryPruneDialPortNumberZero"
 
 	// Retired parameters
 
@@ -1091,6 +1098,14 @@ var defaultParameters = map[string]struct {
 
 	ServerDNSResolverCacheMaxSize: {value: 32, minimum: 0, flags: serverSideOnly},
 	ServerDNSResolverCacheTTL:     {value: 10 * time.Second, minimum: time.Duration(0), flags: serverSideOnly},
+
+	CheckServerEntryTagsEnabled:        {value: true},
+	CheckServerEntryTagsPeriod:         {value: 90 * 24 * time.Hour, minimum: time.Duration(0)},
+	CheckServerEntryTagsRepeatRatio:    {value: 0.10, minimum: 0.0},
+	CheckServerEntryTagsRepeatMinimum:  {value: 1, minimum: 0},
+	CheckServerEntryTagsMaxSendBytes:   {value: 65536, minimum: 1},
+	CheckServerEntryTagsMaxWorkTime:    {value: 60 * time.Second, minimum: time.Duration(0)},
+	ServerEntryPruneDialPortNumberZero: {value: true},
 }
 
 // IsServerSideOnly indicates if the parameter specified by name is used

psiphon/controller.go

@@ -1089,7 +1089,15 @@ loop:
 					controller.stopEstablishing()
 				}
 
-				err := connectedTunnel.Activate(controller.runCtx, controller)
+				// In the case of multi-tunnels, only the first tunnel will send status requests,
+				// including transfer stats (domain bytes), persistent stats, and prune checks.
+				// While transfer stats and persistent stats use a "take out" scheme that would
+				// allow for multiple, concurrent requesters, the prune check does not.
+
+				isStatusReporter := isFirstTunnel
+
+				err := connectedTunnel.Activate(
+					controller.runCtx, controller, isStatusReporter)
 
 				if err != nil {
 					NoticeWarning("failed to activate %s: %v",

psiphon/dataStore.go

@@ -29,6 +29,7 @@ import (
 	"os"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common"
@@ -57,12 +58,14 @@ var (
 	datastoreInproxyCommonCompartmentIDsKey     = []byte("inproxyCommonCompartmentIDs")
 	datastorePersistentStatTypeRemoteServerList = string(datastoreRemoteServerListStatsBucket)
 	datastorePersistentStatTypeFailedTunnel     = string(datastoreFailedTunnelStatsBucket)
+	datastoreCheckServerEntryTagsEndTimeKey     = "checkServerEntryTagsEndTime"
 	datastoreServerEntryFetchGCThreshold        = 10
 
 	datastoreReferenceCountMutex sync.RWMutex
 	datastoreReferenceCount      int64
 	datastoreMutex               sync.RWMutex
 	activeDatastoreDB            *datastoreDB
+	disableCheckServerEntryTags  atomic.Bool
 )
 
 // OpenDataStore opens and initializes the singleton datastore instance.
@@ -595,6 +598,7 @@ type ServerEntryIterator struct {
 	serverEntryIndex             int
 	isTacticsServerEntryIterator bool
 	isTargetServerEntryIterator  bool
+	isPruneServerEntryIterator   bool
 	hasNextTargetServerEntry     bool
 	targetServerEntry            *protocol.ServerEntry
 }
@@ -658,6 +662,23 @@ func NewTacticsServerEntryIterator(config *Config) (*ServerEntryIterator, error)
 	return iterator, nil
 }
 
+func NewPruneServerEntryIterator(config *Config) (*ServerEntryIterator, error) {
+
+	// There is no TargetServerEntry case when pruning.
+
+	iterator := &ServerEntryIterator{
+		config:                     config,
+		isPruneServerEntryIterator: true,
+	}
+
+	err := iterator.reset(true)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	return iterator, nil
+}
+
 // newTargetServerEntryIterator is a helper for initializing the TargetServerEntry case
 func newTargetServerEntryIterator(config *Config, isTactics bool) (bool, *ServerEntryIterator, error) {
 
@@ -765,12 +786,21 @@ func (iterator *ServerEntryIterator) reset(isInitialRound bool) error {
 		serverEntryIDs = make([][]byte, 0)
 		shuffleHead := 0
 
+		// The prune case, isPruneServerEntryIterator, skips all
+		// move-to-front operations and uses a pure random shuffle in order
+		// to uniformly select server entries to prune check. There may be a
+		// benefit to inverting the move and move affinity and potential
+		// replay servers to the _back_ if they're less likely to be pruned;
+		// however, the replay logic here doesn't check the replay TTL and
+		// even potential replay servers might be pruned.
+
 		var affinityServerEntryID []byte
 
 		// In the first round only, move any server affinity candiate to the
 		// very first position.
 
-		if isInitialRound &&
+		if !iterator.isPruneServerEntryIterator &&
+			isInitialRound &&
 			iterator.applyServerAffinity {
 
 			affinityServerEntryID = bucket.get(datastoreAffinityServerEntryIDKey)
@@ -812,7 +842,8 @@ func (iterator *ServerEntryIterator) reset(isInitialRound bool) error {
 
 		p := iterator.config.GetParameters().Get()
 
-		if (isInitialRound || p.WeightedCoinFlip(parameters.ReplayLaterRoundMoveToFrontProbability)) &&
+		if !iterator.isPruneServerEntryIterator &&
+			(isInitialRound || p.WeightedCoinFlip(parameters.ReplayLaterRoundMoveToFrontProbability)) &&
 			p.Int(parameters.ReplayCandidateCount) != 0 {
 
 			networkID := []byte(iterator.config.GetNetworkID())
@@ -1068,7 +1099,11 @@ func (iterator *ServerEntryIterator) Next() (*protocol.ServerEntry, error) {
 
 		// Check filter requirements
 
-		if iterator.isTacticsServerEntryIterator {
+		if iterator.isPruneServerEntryIterator {
+			// No region filter for the prune case.
+			break
+
+		} else if iterator.isTacticsServerEntryIterator {
 
 			// Tactics doesn't filter by egress region.
 			if len(serverEntry.GetSupportedTacticsProtocols()) > 0 {
@@ -1103,23 +1138,28 @@ func MakeCompatibleServerEntry(serverEntry *protocol.ServerEntry) *protocol.Serv
 // PruneServerEntry deletes the server entry, along with associated data,
 // corresponding to the specified server entry tag. Pruning is subject to an
 // age check. In the case of an error, a notice is emitted.
-func PruneServerEntry(config *Config, serverEntryTag string) {
-	err := pruneServerEntry(config, serverEntryTag)
+func PruneServerEntry(config *Config, serverEntryTag string) bool {
+	pruned, err := pruneServerEntry(config, serverEntryTag)
 	if err != nil {
 		NoticeWarning(
 			"PruneServerEntry failed: %s: %s",
 			serverEntryTag, errors.Trace(err))
-		return
+		return false
+	}
+	if pruned {
+		NoticePruneServerEntry(serverEntryTag)
 	}
-	NoticePruneServerEntry(serverEntryTag)
+	return pruned
 }
 
-func pruneServerEntry(config *Config, serverEntryTag string) error {
+func pruneServerEntry(config *Config, serverEntryTag string) (bool, error) {
 
 	minimumAgeForPruning := config.GetParameters().Get().Duration(
 		parameters.ServerEntryMinimumAgeForPruning)
 
-	return datastoreUpdate(func(tx *datastoreTx) error {
+	pruned := false
+
+	err := datastoreUpdate(func(tx *datastoreTx) error {
 
 		serverEntries := tx.bucket(datastoreServerEntriesBucket)
 		serverEntryTags := tx.bucket(datastoreServerEntryTagsBucket)
@@ -1196,8 +1236,12 @@ func pruneServerEntry(config *Config, serverEntryTag string) error {
 			}
 		}
 
+		pruned = true
+
 		return nil
 	})
+
+	return pruned, errors.Trace(err)
 }
 
 // DeleteServerEntry deletes the specified server entry and associated data.
@@ -1592,16 +1636,26 @@ func CountUnreportedPersistentStats() int {
 // set to StateReporting. If the records are successfully reported, clear them
 // with ClearReportedPersistentStats. If the records are not successfully
 // reported, restore them with PutBackUnreportedPersistentStats.
-func TakeOutUnreportedPersistentStats(config *Config) (map[string][][]byte, error) {
+func TakeOutUnreportedPersistentStats(
+	config *Config,
+	adjustMaxSendBytes int) (map[string][][]byte, int, error) {
+
+	// TODO: add a failsafe like disableCheckServerEntryTags, to avoid repeatedly resending
+	// persistent stats in the case of a local error? Also consider just dropping persistent stats
+	// which fail to send due to a network disconnection, rather than invoking
+	// PutBackUnreportedPersistentStats -- especially if it's likely that the server received the
+	// stats and the disconnection occurs just before the request is acknowledged.
 
 	stats := make(map[string][][]byte)
 
 	maxSendBytes := config.GetParameters().Get().Int(
 		parameters.PersistentStatsMaxSendBytes)
 
-	err := datastoreUpdate(func(tx *datastoreTx) error {
+	maxSendBytes -= adjustMaxSendBytes
+
+	sendBytes := 0
 
-		sendBytes := 0
+	err := datastoreUpdate(func(tx *datastoreTx) error {
 
 		for _, statType := range persistentStatTypes {
 
@@ -1653,10 +1707,10 @@ func TakeOutUnreportedPersistentStats(config *Config) (map[string][][]byte, erro
 	})
 
 	if err != nil {
-		return nil, errors.Trace(err)
+		return nil, 0, errors.Trace(err)
 	}
 
-	return stats, nil
+	return stats, sendBytes, nil
 }
 
 // PutBackUnreportedPersistentStats restores a list of persistent
@@ -1752,6 +1806,171 @@ func resetAllPersistentStatsToUnreported() error {
 	return nil
 }
 
+// IsCheckServerEntryTagsDue indicates that a new prune check is due, based on
+// the time of the previous check ending.
+func IsCheckServerEntryTagsDue(config *Config) bool {
+
+	// disableCheckServerEntryTags is a failsafe, enabled in error cases below
+	// and in UpdateCheckServerEntryTagsEndTime to prevent constantly
+	// resending prune check payloads if the scheduling mechanism fails.
+	if disableCheckServerEntryTags.Load() {
+		return false
+	}
+
+	// Whether the next check is due is based on time elapsed since the time
+	// of the previous check ending, with the elapsed time set in tactics.
+	// The previous end time, rather the next due time, is stored, to allow
+	// changes to this tactic to have immediate effect.
+
+	p := config.GetParameters().Get()
+	enabled := p.Bool(parameters.CheckServerEntryTagsEnabled)
+	checkPeriod := p.Duration(parameters.CheckServerEntryTagsPeriod)
+	p.Close()
+
+	if !enabled {
+		return false
+	}
+
+	lastEndTimeValue, err := GetKeyValue(datastoreCheckServerEntryTagsEndTimeKey)
+	if err != nil {
+		NoticeWarning("IsCheckServerEntryTagsDue GetKeyValue failed: %s", errors.Trace(err))
+		disableCheckServerEntryTags.Store(true)
+		return false
+	}
+
+	if lastEndTimeValue == "" {
+		return true
+	}
+
+	lastEndTime, err := time.Parse(time.RFC3339, lastEndTimeValue)
+	if err != nil {
+		NoticeWarning("IsCheckServerEntryTagsDue time.Parse failed: %s", errors.Trace(err))
+		disableCheckServerEntryTags.Store(true)
+		return false
+	}
+
+	return time.Now().After(lastEndTime.Add(checkPeriod))
+}
+
+// UpdateCheckServerEntryTagsEndTime should be called after a prune check is
+// complete. The end time is set, extending the time until the next check,
+// unless there's a sufficiently high ratio of pruned servers from the last
+// check.
+func UpdateCheckServerEntryTagsEndTime(config *Config, checkCount int, pruneCount int) {
+
+	p := config.GetParameters().Get()
+	ratio := p.Float(parameters.CheckServerEntryTagsRepeatRatio)
+	minimum := p.Int(parameters.CheckServerEntryTagsRepeatMinimum)
+	p.Close()
+
+	// When there's a sufficiently high ratio of pruned/checked from
+	// the _previous_ check operation, don't mark the check as ended. This
+	// will result in the next status request performing another check. It's
+	// assumed that the ratio will decrease over the course of repeated
+	// checks as more server entries are pruned, and random selection for
+	// checking will include fewer and fewer invalid server entry tags.
+	//
+	// The rate of repeated checking is also limited by the status request
+	// schedule, where PsiphonAPIStatusRequestPeriodMin/Max defaults to 5-10
+	// minutes.
+
+	if pruneCount >= minimum && ratio > 0 && float64(pruneCount)/float64(checkCount) >= ratio {
+		NoticeInfo("UpdateCheckServerEntryTagsEndTime: %d/%d: repeat", pruneCount, checkCount)
+		return
+	}
+
+	err := SetKeyValue(
+		datastoreCheckServerEntryTagsEndTimeKey,
+		time.Now().Format(time.RFC3339))
+	if err != nil {
+		NoticeWarning("UpdateCheckServerEntryTagsEndTime SetKeyValue failed: %s", errors.Trace(err))
+		disableCheckServerEntryTags.Store(true)
+		return
+	}
+
+	NoticeInfo("UpdateCheckServerEntryTagsEndTime: %d/%d: done", pruneCount, checkCount)
+}
+
+// GetCheckServerEntryTags returns a random selection of server entry tags to
+// be checked for pruning. An empty list is returned if a check is not yet
+// due.
+func GetCheckServerEntryTags(config *Config) ([]string, int, error) {
+
+	if disableCheckServerEntryTags.Load() {
+		return nil, 0, nil
+	}
+
+	if !IsCheckServerEntryTagsDue(config) {
+		return nil, 0, nil
+	}
+
+	// maxSendBytes is intended to limit the request memory overhead and
+	// network size. maxWorkTime ensures that slow devices -- with datastore
+	// operations and JSON unmarshaling particularly slow -- will launch a
+	// request in a timely fashion.
+
+	p := config.GetParameters().Get()
+	maxSendBytes := p.Int(parameters.CheckServerEntryTagsMaxSendBytes)
+	maxWorkTime := p.Duration(parameters.CheckServerEntryTagsMaxWorkTime)
+	minimumAgeForPruning := p.Duration(parameters.ServerEntryMinimumAgeForPruning)
+	p.Close()
+
+	iterator, err := NewPruneServerEntryIterator(config)
+	if err != nil {
+		return nil, 0, errors.Trace(err)
+	}
+
+	var checkTags []string
+	bytes := 0
+	startWork := time.Now()
+
+	for {
+
+		serverEntry, err := iterator.Next()
+		if err != nil {
+			return nil, 0, errors.Trace(err)
+		}
+
+		if serverEntry == nil {
+			break
+		}
+
+		// Skip checking the server entry if PruneServerEntry won't prune it
+		// anyway, due to ServerEntryMinimumAgeForPruning.
+		serverEntryLocalTimestamp, err := time.Parse(time.RFC3339, serverEntry.LocalTimestamp)
+		if err != nil {
+			return nil, 0, errors.Trace(err)
+		}
+		if serverEntryLocalTimestamp.Add(minimumAgeForPruning).After(time.Now()) {
+			continue
+		}
+
+		// Server entries with replay records are not skipped. It's possible that replay records are
+		// retained, due to ReplayRetainFailedProbability, even if the server entry is no longer
+		// valid. Inspecting replay would also require an additional JSON unmarshal of the
+		// DialParameters, in order to check the replay TTL.
+		//
+		// A potential future enhancement could be to add and check a new index that tracks how
+		// recently a server entry connection got as far as completing the SSH handshake, which
+		// verifies the Psiphon server running at that server entry network address. This would
+		// exclude from prune checking all recently known-valid servers regardless of whether they
+		// ultimately pass the liveness test, establish a tunnel, or reach the replay data transfer
+		// targets.
+
+		checkTags = append(checkTags, serverEntry.Tag)
+
+		// Approximate the size of the JSON encoding of the string array,
+		// including quotes and commas.
+		bytes += len(serverEntry.Tag) + 3
+
+		if bytes >= maxSendBytes || (maxWorkTime > 0 && time.Since(startWork) > maxWorkTime) {
+			break
+		}
+	}
+
+	return checkTags, bytes, nil
+}
+
 // CountSLOKs returns the total number of SLOK records.
 func CountSLOKs() int {
 

psiphon/dialParameters.go

@@ -1364,6 +1364,18 @@ func MakeDialParameters(
 		return nil, errors.Trace(err)
 	}
 
+	if dialPortNumber == 0 && p.Bool(parameters.ServerEntryPruneDialPortNumberZero) {
+
+		// Automatically prune any invalid server entry that has produced an
+		// invalid dial port number of 0. This case may arise due to missing
+		// port number fields in server entries. For older clients, this
+		// prune case is enforced in the server's status request
+		// failed_tunnel processing; see server.statusAPIRequestHandler.
+
+		PruneServerEntry(config, serverEntry.IpAddress)
+		return nil, errors.TraceNew("invalid dial port number")
+	}
+
 	dialParams.DialPortNumber = strconv.Itoa(dialPortNumber)
 
 	switch protocol.TunnelProtocolMinusInproxy(dialParams.TunnelProtocol) {

psiphon/server/api.go

@@ -805,6 +805,7 @@ func statusAPIRequestHandler(
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
+
 		for _, remoteServerListStat := range remoteServerListStats {
 
 			for _, name := range remoteServerListStatBackwardsCompatibilityParamNames {
@@ -839,19 +840,18 @@ func statusAPIRequestHandler(
 	// Failed tunnel persistent stats.
 	// Older clients may not submit this data.
 
-	var invalidServerEntryTags map[string]bool
-
-	if statusData["failed_tunnel_stats"] != nil {
+	// Note: no guarantee that PsinetDatabase won't reload between database calls
+	db := support.PsinetDatabase
 
-		// Note: no guarantee that PsinetDatabase won't reload between database calls
-		db := support.PsinetDatabase
+	invalidServerEntryTags := make(map[string]bool)
 
-		invalidServerEntryTags = make(map[string]bool)
+	if statusData["failed_tunnel_stats"] != nil {
 
 		failedTunnelStats, err := getJSONObjectArrayRequestParam(statusData, "failed_tunnel_stats")
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
+
 		for _, failedTunnelStat := range failedTunnelStats {
 
 			err := validateRequestParams(support.Config, failedTunnelStat, failedTunnelStatParams)
@@ -895,7 +895,7 @@ func statusAPIRequestHandler(
 			// historical bugs in various server entry handling implementations. When
 			// missing from a server entry loaded by a client, the port number
 			// evaluates to 0, the zero value, which is not a valid port number even if
-			// were not missing.
+			// it were not missing.
 
 			serverEntryTag, ok := getOptionalStringRequestParam(failedTunnelStat, "server_entry_tag")
 
@@ -924,6 +924,38 @@ func statusAPIRequestHandler(
 		}
 	}
 
+	// Handle the prune check, which is an aggressive server entry prune
+	// operation on top of the opportunistic pruning that is triggered by
+	// failed_tunnel reports.
+
+	if statusData["check_server_entry_tags"] != nil {
+
+		checkServerEntryTags, err := getStringArrayRequestParam(statusData, "check_server_entry_tags")
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+
+		invalidCount := 0
+
+		for _, serverEntryTag := range checkServerEntryTags {
+
+			serverEntryValid := db.IsValidServerEntryTag(serverEntryTag)
+			if !serverEntryValid {
+				invalidServerEntryTags[serverEntryTag] = true
+				invalidCount += 1
+			}
+
+		}
+
+		// Prune metrics will be logged in server_tunnel.
+
+		sshClient.Lock()
+		sshClient.requestCheckServerEntryTags += 1
+		sshClient.checkedServerEntryTags += len(checkServerEntryTags)
+		sshClient.invalidServerEntryTags += invalidCount
+		sshClient.Unlock()
+	}
+
 	for _, logItem := range logQueue {
 		log.LogRawFieldsWithTimestamp(logItem)
 	}

psiphon/server/server_test.go

@@ -620,6 +620,20 @@ func TestPruneServerEntries(t *testing.T) {
 		})
 }
 
+func TestCheckPruneServerEntries(t *testing.T) {
+	runServer(t,
+		&runServerConfig{
+			tunnelProtocol:          "OSSH",
+			requireAuthorization:    true,
+			doTunneledWebRequest:    true,
+			doTunneledNTPRequest:    true,
+			forceLivenessTest:       true,
+			doPruneServerEntries:    true,
+			checkPruneServerEntries: true,
+			doLogHostProvider:       true,
+		})
+}
+
 func TestBurstMonitorAndDestinationBytes(t *testing.T) {
 	runServer(t,
 		&runServerConfig{
@@ -743,6 +757,7 @@ type runServerConfig struct {
 	forceFragmenting         bool
 	forceLivenessTest        bool
 	doPruneServerEntries     bool
+	checkPruneServerEntries  bool
 	doDanglingTCPConn        bool
 	doPacketManipulation     bool
 	doBurstMonitor           bool
@@ -1583,11 +1598,32 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 		}
 
 		if runConfig.doPruneServerEntries {
-			applyParameters[parameters.PsiphonAPIStatusRequestShortPeriodMin] = 1 * time.Millisecond
-			applyParameters[parameters.PsiphonAPIStatusRequestShortPeriodMax] = 1 * time.Millisecond
+			applyParameters[parameters.PsiphonAPIStatusRequestShortPeriodMin] = 1 * time.Second
+			applyParameters[parameters.PsiphonAPIStatusRequestShortPeriodMax] = 1 * time.Second
+
+			if runConfig.checkPruneServerEntries {
+
+				// Set a low MaxSendBytes in order to exercise repeated check
+				// prune requests. Also set a short deadline for the
+				// subsequent status requests, as the default is minutes later.
+
+				applyParameters[parameters.CheckServerEntryTagsRepeatRatio] = 0.0001
+				applyParameters[parameters.CheckServerEntryTagsRepeatMinimum] = 0
+				applyParameters[parameters.CheckServerEntryTagsMaxSendBytes] =
+					(len(pruneServerEntryTestCases) / 2) * 43
+
+				applyParameters[parameters.PsiphonAPIStatusRequestPeriodMin] = 1 * time.Second
+				applyParameters[parameters.PsiphonAPIStatusRequestPeriodMax] = 1 * time.Second
+
+			} else {
+
+				// Force exercising the failed_tunnel prune code path.
+
+				applyParameters[parameters.CheckServerEntryTagsEnabled] = false
+			}
 		}
 
-		err = clientConfig.SetParameters("", true, applyParameters)
+		err = clientConfig.SetParameters("", false, applyParameters)
 		if err != nil {
 			t.Fatalf("SetParameters failed: %s", err)
 		}
@@ -2006,6 +2042,11 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 
 	select {
 	case logFields := <-serverTunnelLog:
+
+		expectCheckServerEntryPruneCount := 0
+		if runConfig.checkPruneServerEntries {
+			expectCheckServerEntryPruneCount = expectedNumPruneNotices
+		}
 		err := checkExpectedServerTunnelLogFields(
 			runConfig,
 			propagationChannelID,
@@ -2024,6 +2065,7 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 			expectLegacyDestinationBytesFields,
 			passthroughAddress,
 			expectMeekHTTPVersion,
+			expectCheckServerEntryPruneCount,
 			inproxyTestConfig,
 			logFields)
 		if err != nil {
@@ -2336,6 +2378,7 @@ func checkExpectedServerTunnelLogFields(
 	expectLegacyDestinationBytesFields bool,
 	expectPassthroughAddress *string,
 	expectMeekHTTPVersion string,
+	expectCheckServerEntryPruneCount int,
 	inproxyTestConfig *inproxyTestConfig,
 	fields map[string]interface{}) error {
 
@@ -3076,6 +3119,35 @@ func checkExpectedServerTunnelLogFields(
 		}
 	}
 
+	for _, name := range []string{
+		"request_check_server_entry_tags",
+		"checked_server_entry_tags",
+		"invalid_server_entry_tags",
+	} {
+		if expectCheckServerEntryPruneCount > 0 && fields[name] == nil {
+			return fmt.Errorf("missing expected field '%s'", name)
+
+		} else if expectCheckServerEntryPruneCount <= 0 && fields[name] != nil {
+			return fmt.Errorf("unexpected field '%s'", name)
+		}
+	}
+	if expectCheckServerEntryPruneCount > 0 {
+		name := "request_check_server_entry_tags"
+		if fields[name].(float64) < 2 {
+			return fmt.Errorf("unexpected field value %s: %v", name, fields[name])
+		}
+		name = "checked_server_entry_tags"
+		if fields[name].(float64) < 1 {
+			return fmt.Errorf("unexpected field value %s: %v", name, fields[name])
+		}
+		// invalid_server_entry_tags may exceed expectCheckServerEntryPruneCount,
+		// due to repeated requests and min prune age.
+		name = "invalid_server_entry_tags"
+		if int(fields[name].(float64)) < expectCheckServerEntryPruneCount {
+			return fmt.Errorf("unexpected field value %s: %v", name, fields[name])
+		}
+	}
+
 	return nil
 }
 
@@ -4192,6 +4264,12 @@ func initializePruneServerEntriesTest(
 		{IPAddress: "192.0.2.13", ExplicitTag: true, LocalTimestamp: oldTimeStamp, PsinetValid: true, ExpectPrune: true, IsEmbedded: true, DialPort0: true},
 	}
 
+	if runConfig.checkPruneServerEntries {
+		// Skip the dial port 0 cases, since the prune check doesn't send the
+		// dial port number in its request.
+		pruneServerEntryTestCases = pruneServerEntryTestCases[0:10]
+	}
+
 	for _, testCase := range pruneServerEntryTestCases {
 
 		dialPort := 4000
@@ -4275,6 +4353,11 @@ func storePruneServerEntriesTest(
 		}
 	}
 
+	if runConfig.checkPruneServerEntries {
+		// The prune check case doesn't create failed_tunnel records.
+		return
+	}
+
 	clientConfig := &psiphon.Config{
 		SponsorId:            "0",
 		PropagationChannelId: "0",
@@ -4295,6 +4378,12 @@ func storePruneServerEntriesTest(
 	applyParameters := make(map[string]interface{})
 	applyParameters[parameters.RecordFailedTunnelPersistentStatsProbability] = 1.0
 
+	// In order to reach the server-side status request failed_tunnel dial
+	// port 0 handling, disable ServerEntryPruneDialPortNumberZero so that
+	// the following MakeDialParameters will ignore the dial port 0 and not
+	// try to immediately prune the server entry.
+	applyParameters[parameters.ServerEntryPruneDialPortNumberZero] = false
+
 	err = clientConfig.SetParameters("", true, applyParameters)
 	if err != nil {
 		t.Fatalf("SetParameters failed: %s", err)

psiphon/server/tunnelServer.go

@@ -1953,6 +1953,9 @@ type sshClient struct {
 	inproxyProxyQualityTracker           *inproxyProxyQualityTracker
 	dnsResolver                          *net.Resolver
 	dnsCache                             *lrucache.Cache
+	requestCheckServerEntryTags          int
+	checkedServerEntryTags               int
+	invalidServerEntryTags               int
 }
 
 type trafficState struct {
@@ -3749,6 +3752,12 @@ func (sshClient *sshClient) logTunnel(additionalMetrics []LogFields) {
 		logFields["relayed_steering_ip"] = sshClient.additionalTransportData.steeringIP
 	}
 
+	if sshClient.requestCheckServerEntryTags > 0 {
+		logFields["request_check_server_entry_tags"] = sshClient.requestCheckServerEntryTags
+		logFields["checked_server_entry_tags"] = sshClient.checkedServerEntryTags
+		logFields["invalid_server_entry_tags"] = sshClient.invalidServerEntryTags
+	}
+
 	// Merge in additional metrics from the optional metrics source
 	for _, metrics := range additionalMetrics {
 		for name, value := range metrics {

psiphon/serverApi.go

@@ -561,7 +561,7 @@ func (serverContext *ServerContext) StatsRegexps() *transferstats.Regexps {
 }
 
 // DoStatusRequest makes a "status" API request to the server, sending session stats.
-func (serverContext *ServerContext) DoStatusRequest(tunnel *Tunnel) error {
+func (serverContext *ServerContext) DoStatusRequest() error {
 
 	params := serverContext.getBaseAPIParameters(
 		baseParametersNoDialParameters, false)
@@ -571,7 +571,7 @@ func (serverContext *ServerContext) DoStatusRequest(tunnel *Tunnel) error {
 
 	statusPayload, statusPayloadInfo, err := makeStatusRequestPayload(
 		serverContext.tunnel.config,
-		tunnel.dialParams.ServerEntry.IpAddress)
+		serverContext.tunnel.dialParams.ServerEntry.IpAddress)
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -616,6 +616,10 @@ func (serverContext *ServerContext) DoStatusRequest(tunnel *Tunnel) error {
 		return errors.Trace(err)
 	}
 
+	// Confirm the payload now that the server response is received. For
+	// persistentStats and transferStats, this clears the reported data as it
+	// is now delivered and doesn't need to be resent.
+
 	confirmStatusRequestPayload(statusPayloadInfo)
 
 	var statusResponse protocol.StatusResponse
@@ -624,8 +628,41 @@ func (serverContext *ServerContext) DoStatusRequest(tunnel *Tunnel) error {
 		return errors.Trace(err)
 	}
 
+	// Prune all server entries flagged by either the failed_tunnel mechanism
+	// or the prune check. Note that server entries that are too new, as
+	// determined by ServerEntryMinimumAgeForPruning, are not pruned, and
+	// this is reflected in pruneCount.
+
+	pruneCount := 0
 	for _, serverEntryTag := range statusResponse.InvalidServerEntryTags {
-		PruneServerEntry(serverContext.tunnel.config, serverEntryTag)
+		if PruneServerEntry(serverContext.tunnel.config, serverEntryTag) {
+			pruneCount++
+		}
+	}
+
+	if pruneCount > 0 {
+		NoticeInfo("Pruned server entries: %d", pruneCount)
+	}
+
+	if statusPayloadInfo.checkServerEntryTagCount > 0 {
+
+		// Schedule the next prune check, now that all pruning is complete. By
+		// design, if the process dies before the end of the prune loop, the
+		// previous due time will be retained.
+		//
+		// UpdateCheckServerEntryTagsEndTime may leave the next prune check
+		// due immediately based on the ratio of server entries checked and
+		// server entries pruned: if many checked server entries were invalid
+		// and pruned, check again and prune more.
+		//
+		// Limitation: the prune count may include failed_tunnel prunes which
+		// aren't in the check count; if this occurs, it will increase the
+		// ratio and make an immediate re-check more likely, which makes sense.
+
+		UpdateCheckServerEntryTagsEndTime(
+			serverContext.tunnel.config,
+			statusPayloadInfo.checkServerEntryTagCount,
+			pruneCount)
 	}
 
 	return nil
@@ -635,9 +672,10 @@ func (serverContext *ServerContext) DoStatusRequest(tunnel *Tunnel) error {
 // either "clear" or "put back" status request payload data depending
 // on whether or not the request succeeded.
 type statusRequestPayloadInfo struct {
-	serverId        string
-	transferStats   *transferstats.AccumulatedStats
-	persistentStats map[string][][]byte
+	serverId                 string
+	transferStats            *transferstats.AccumulatedStats
+	persistentStats          map[string][][]byte
+	checkServerEntryTagCount int
 }
 
 func makeStatusRequestPayload(
@@ -650,10 +688,26 @@ func makeStatusRequestPayload(
 	//
 	// TODO: pack and CBOR encode the status request payload.
 
+	// GetCheckServerEntryTags returns a randomly selected set of server entry
+	// tags to be checked for pruning, or an empty list if a check is not yet
+	// due.
+	//
+	// Both persistentStats and prune check data have a max payload size
+	// allowance, and the allowance for persistentStats is reduced by the
+	// size of the prune check data, if any.
+
+	checkServerEntryTags, tagsSize, err := GetCheckServerEntryTags(config)
+	if err != nil {
+		NoticeWarning(
+			"GetCheckServerEntryTags failed: %s", errors.Trace(err))
+		checkServerEntryTags = nil
+		// Proceed with persistentStats/transferStats only
+	}
+
 	transferStats := transferstats.TakeOutStatsForServer(serverId)
 	hostBytes := transferStats.GetStatsForStatusRequest()
 
-	persistentStats, err := TakeOutUnreportedPersistentStats(config)
+	persistentStats, statsSize, err := TakeOutUnreportedPersistentStats(config, tagsSize)
 	if err != nil {
 		NoticeWarning(
 			"TakeOutUnreportedPersistentStats failed: %s", errors.Trace(err))
@@ -661,13 +715,17 @@ func makeStatusRequestPayload(
 		// Proceed with transferStats only
 	}
 
-	if len(hostBytes) == 0 && len(persistentStats) == 0 {
+	if len(checkServerEntryTags) == 0 && len(hostBytes) == 0 && len(persistentStats) == 0 {
 		// There is no payload to send.
 		return nil, nil, nil
 	}
 
 	payloadInfo := &statusRequestPayloadInfo{
-		serverId, transferStats, persistentStats}
+		serverId,
+		transferStats,
+		persistentStats,
+		len(checkServerEntryTags),
+	}
 
 	payload := make(map[string]interface{})
 
@@ -692,6 +750,8 @@ func makeStatusRequestPayload(
 		payload[persistentStatPayloadNames[statType]] = jsonStats
 	}
 
+	payload["check_server_entry_tags"] = checkServerEntryTags
+
 	jsonPayload, err := json.Marshal(payload)
 	if err != nil {
 
@@ -701,6 +761,10 @@ func makeStatusRequestPayload(
 		return nil, nil, errors.Trace(err)
 	}
 
+	NoticeInfo(
+		"StatusRequestPayload: %d total bytes, %d stats bytes, %d tag bytes",
+		len(jsonPayload), statsSize, tagsSize)
+
 	return jsonPayload, payloadInfo, nil
 }
 

psiphon/tunnel.go

@@ -29,6 +29,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"math"
 	"net"
 	"net/http"
 	"slices"
@@ -89,6 +90,7 @@ type Tunnel struct {
 	mutex                          *sync.Mutex
 	config                         *Config
 	isActivated                    bool
+	isStatusReporter               bool
 	isDiscarded                    bool
 	isClosed                       bool
 	dialParams                     *DialParameters
@@ -178,7 +180,9 @@ func ConnectTunnel(
 // request and starting operateTunnel, the worker that monitors the tunnel
 // and handles periodic management.
 func (tunnel *Tunnel) Activate(
-	ctx context.Context, tunnelOwner TunnelOwner) (retErr error) {
+	ctx context.Context,
+	tunnelOwner TunnelOwner,
+	isStatusReporter bool) (retErr error) {
 
 	// Ensure that, unless the base context is cancelled, any replayed dial
 	// parameters are cleared, no longer to be retried, if the tunnel fails to
@@ -340,6 +344,7 @@ func (tunnel *Tunnel) Activate(
 	}
 
 	tunnel.isActivated = true
+	tunnel.isStatusReporter = isStatusReporter
 	tunnel.serverContext = serverContext
 
 	// establishDuration is the elapsed time between the controller starting tunnel
@@ -1783,16 +1788,31 @@ func (tunnel *Tunnel) operateTunnel(tunnelOwner TunnelOwner) {
 	statsTimer := time.NewTimer(nextStatusRequestPeriod())
 	defer statsTimer.Stop()
 
-	// Schedule an almost-immediate status request to deliver any unreported
-	// persistent stats.
-	unreported := CountUnreportedPersistentStats()
-	if unreported > 0 {
-		NoticeInfo("Unreported persistent stats: %d", unreported)
-		p := tunnel.getCustomParameters()
-		statsTimer.Reset(
-			prng.Period(
-				p.Duration(parameters.PsiphonAPIStatusRequestShortPeriodMin),
-				p.Duration(parameters.PsiphonAPIStatusRequestShortPeriodMax)))
+	// Only one active tunnel should be designated as the status reporter for
+	// sending stats and prune checks. While the stats provide a "take out"
+	// scheme that would allow for multiple, concurrent requesters, the prune
+	// check does not.
+	//
+	// The statsTimer is retained, but set to practically never trigger, in
+	// the !isStatusReporter case to simplify following select statements.
+	if tunnel.isStatusReporter {
+
+		// Schedule an almost-immediate status request to deliver any unreported
+		// persistent stats or perform a server entry prune check.
+		unreported := CountUnreportedPersistentStats()
+		isCheckDue := IsCheckServerEntryTagsDue(tunnel.config)
+		if unreported > 0 || isCheckDue {
+			NoticeInfo(
+				"Unreported persistent stats: %d; server entry check due: %v",
+				unreported, isCheckDue)
+			p := tunnel.getCustomParameters()
+			statsTimer.Reset(
+				prng.Period(
+					p.Duration(parameters.PsiphonAPIStatusRequestShortPeriodMin),
+					p.Duration(parameters.PsiphonAPIStatusRequestShortPeriodMax)))
+		}
+	} else {
+		statsTimer = time.NewTimer(time.Duration(math.MaxInt64))
 	}
 
 	nextSshKeepAlivePeriod := func() time.Duration {
@@ -1922,11 +1942,13 @@ func (tunnel *Tunnel) operateTunnel(tunnelOwner TunnelOwner) {
 			}
 
 		case <-statsTimer.C:
-			select {
-			case signalStatusRequest <- struct{}{}:
-			default:
+			if tunnel.isStatusReporter {
+				select {
+				case signalStatusRequest <- struct{}{}:
+				default:
+				}
+				statsTimer.Reset(nextStatusRequestPeriod())
 			}
-			statsTimer.Reset(nextStatusRequestPeriod())
 
 		case <-sshKeepAliveTimer.C:
 			p := tunnel.getCustomParameters()
@@ -2246,7 +2268,7 @@ func sendStats(tunnel *Tunnel) bool {
 		return true
 	}
 
-	err := tunnel.serverContext.DoStatusRequest(tunnel)
+	err := tunnel.serverContext.DoStatusRequest()
 	if err != nil {
 		NoticeWarning("DoStatusRequest failed for %s: %s",
 			tunnel.dialParams.ServerEntry.GetDiagnosticID(), err)