streamsync: speed up stream rediscovery and recover faster from local outages

api/service/synchronize/stagedstreamsync/const.go

@@ -52,9 +52,8 @@ const (
 
 	// StreamDiscoveryWatchdogTimeout is the maximum time to wait for enough stream
 	// connections before resetting stream manager runtime state and retrying.
-	// Keep it longer than stream removal cooldown windows so punished nodes are not
-	// unblocked immediately by watchdog recovery.
-	StreamDiscoveryWatchdogTimeout time.Duration = 75 * time.Minute
+	// Intended to stay above streammanager.MaxRemovalCooldownDuration.
+	StreamDiscoveryWatchdogTimeout time.Duration = 30 * time.Minute
 
 	// pivot block distance ranges
 	MinPivotDistanceToHead uint64 = 1024
@@ -98,7 +97,7 @@ type (
 		Concurrency          int // Number of concurrent sync requests
 		MinStreams           int // Minimum number of streams to do sync
 		InitStreams          int // Number of streams requirement for initial bootstrap
-		MaxAdvertiseWaitTime int // maximum time duration between protocol advertisements
+		MaxAdvertiseWaitTime int // max minutes between advertisements in normal mode
 		// stream manager config
 		SmSoftLowCap int
 		SmHardLowCap int

cmd/config/config_migrations_test.go

@@ -389,6 +389,12 @@ func Test_migrateConf(t *testing.T) {
 				hc := defConf
 				hc.Sync.Client = true
 				hc.Sync.Enabled = true
+				// Fixture Sync peer settings are preserved by migration.
+				hc.Sync.Concurrency = 6
+				hc.Sync.MinPeers = 6
+				hc.Sync.InitStreams = 8
+				hc.Sync.DiscSoftLowCap = 8
+				hc.Sync.DiscHardLowCap = 6
 				return hc
 			}(),
 			wantErr: false,

cmd/config/config_test.go

@@ -140,6 +140,13 @@ Version = "1.0.4"
 		t.Errorf("Expected config version: 1.0.4, not %v", config.Version)
 	}
 	config.Version = defConf.Version // Shortcut for testing, value checked above
+	// Fixture Sync peer settings are preserved by migration; MaxAdvertiseWaitTime
+	// uses the current default when absent from the fixture.
+	defConf.Sync.Concurrency = 6
+	defConf.Sync.MinPeers = 6
+	defConf.Sync.InitStreams = 8
+	defConf.Sync.DiscSoftLowCap = 8
+	defConf.Sync.DiscHardLowCap = 6
 	require.Equal(t, config, defConf)
 }
 

cmd/config/default.go

@@ -209,12 +209,12 @@ var (
 		SyncMode:             0,
 		Client:               false,
 		StagedSyncCfg:        defaultStagedSyncConfig,
-		Concurrency:          6,
-		MinPeers:             6,
-		InitStreams:          8,
-		MaxAdvertiseWaitTime: 60, //minutes
-		DiscSoftLowCap:       8,
-		DiscHardLowCap:       6,
+		Concurrency:          4,
+		MinPeers:             4,
+		InitStreams:          5,
+		MaxAdvertiseWaitTime: 15, // minutes
+		DiscSoftLowCap:       5,
+		DiscHardLowCap:       3,
 		DiscHighCap:          128,
 		DiscBatch:            8,
 		TrustedNodes:         []string{},

cmd/config/flags.go

@@ -1982,7 +1982,7 @@ var (
 	}
 	syncMaxAdvertiseWaitTimeFlag = cli.IntFlag{
 		Name:   "sync.max-advertise-wait-time",
-		Usage:  "The max time duration between two advertises for each p2p peer to tell other nodes what protocols it supports",
+		Usage:  "Max minutes between sync protocol advertisements in normal mode",
 		Hidden: true,
 	}
 	syncDiscSoftLowFlag = cli.IntFlag{

internal/configs/harmony/harmony.go

@@ -372,7 +372,7 @@ type SyncConfig struct {
 	Concurrency          int              // concurrency used for stream sync protocol
 	MinPeers             int              // minimum streams to start a sync task.
 	InitStreams          int              // minimum streams in bootstrap to start sync loop.
-	MaxAdvertiseWaitTime int              // maximum time duration between advertisements
+	MaxAdvertiseWaitTime int              // max minutes between advertisements in normal mode
 	DiscSoftLowCap       int              // when number of streams is below this value, spin discover during check
 	DiscHardLowCap       int              // when removing stream, num is below this value, spin discovery immediately
 	DiscHighCap          int              // upper limit of streams in one sync protocol

p2p/stream/common/streammanager/config.go

@@ -16,9 +16,27 @@ const (
 	connectTimeout = 60 * time.Second
 	// MaxReservedStreams is the maximum number of reserved streams
 	MaxReservedStreams = 100
-	// RemovalCooldownDuration defines the cooldown period (in minutes) before a removed stream can reconnect.
-	RemovalCooldownDuration    = 5 * time.Minute
-	MaxRemovalCooldownDuration = 60 * time.Minute
+	// RemovalCooldownDuration defines the cooldown period before a removed stream can reconnect.
+	RemovalCooldownDuration = 5 * time.Minute
+	// MaxRemovalCooldownDuration is the upper bound for removal cooldowns.
+	// Intended to stay below stagedstreamsync.StreamDiscoveryWatchdogTimeout.
+	MaxRemovalCooldownDuration = 15 * time.Minute
+
+	// Mass-disconnect / local-outage detection parameters.
+	massDisconnectWindow   = 45 * time.Second
+	massDisconnectMinCount = 3
+	// localOutageDuration is how long connection-loss removals use soft reconnect.
+	localOutageDuration = 2 * time.Minute
+	// localOutageDiscHoldoff is the delay before rediscovery after a mass disconnect.
+	localOutageDiscHoldoff = 30 * time.Second
+	// localOutageMinInterval is the minimum time between local-outage windows.
+	localOutageMinInterval = 10 * time.Minute
+
+	// streamRegistrationWait is the max wait for async stream registration after
+	// trusted peer NewStream succeeds.
+	streamRegistrationWait = 5 * time.Second
+	// streamRegistrationPoll is the polling interval while waiting for registrations.
+	streamRegistrationPoll = 50 * time.Millisecond
 
 	// setupConcurrency limits concurrent stream setup goroutines
 	setupConcurrency = 16

p2p/stream/common/streammanager/mass_disconnect.go

@@ -0,0 +1,135 @@
+package streammanager
+
+import (
+	"strings"
+	"time"
+)
+
+// disconnectTracker tracks clustered stream removals and local-outage windows.
+type disconnectTracker struct {
+	removalTimes     []time.Time
+	localOutageUntil time.Time
+	lastOutageStart  time.Time
+}
+
+// observeRemoval records a connection-loss removal and reports whether the removal
+// is inside a local-outage window. Call only for connection-loss removals.
+//
+// activeBefore is the number of streams (main+reserved) at the start of this removal.
+func (dt *disconnectTracker) observeRemoval(now time.Time, activeBefore int) (inLocalOutage bool, justEntered bool) {
+	if !dt.localOutageUntil.IsZero() && now.Before(dt.localOutageUntil) {
+		dt.record(now)
+		return true, false
+	}
+
+	dt.prune(now)
+	dt.record(now)
+
+	threshold := massDisconnectThreshold(activeBefore + len(dt.removalTimes) - 1)
+	// activeBefore plus prior removals in the window approximates stream count at
+	// the start of the disconnect wave.
+	if len(dt.removalTimes) < threshold {
+		return false, false
+	}
+
+	if !dt.lastOutageStart.IsZero() && now.Sub(dt.lastOutageStart) < localOutageMinInterval {
+		return false, false
+	}
+
+	dt.localOutageUntil = now.Add(localOutageDuration)
+	dt.lastOutageStart = now
+	return true, true
+}
+
+func (dt *disconnectTracker) inLocalOutage(now time.Time) bool {
+	return !dt.localOutageUntil.IsZero() && now.Before(dt.localOutageUntil)
+}
+
+func (dt *disconnectTracker) record(now time.Time) {
+	dt.removalTimes = append(dt.removalTimes, now)
+}
+
+func (dt *disconnectTracker) prune(now time.Time) {
+	cutoff := now.Add(-massDisconnectWindow)
+	i := 0
+	for i < len(dt.removalTimes) && dt.removalTimes[i].Before(cutoff) {
+		i++
+	}
+	if i > 0 {
+		dt.removalTimes = append([]time.Time(nil), dt.removalTimes[i:]...)
+	}
+}
+
+// massDisconnectThreshold is the number of removals in the window that opens a
+// local-outage window. Smaller stream sets use a lower floor.
+func massDisconnectThreshold(approxActiveAtWaveStart int) int {
+	if approxActiveAtWaveStart <= 0 {
+		return massDisconnectMinCount
+	}
+	if approxActiveAtWaveStart < massDisconnectMinCount {
+		return approxActiveAtWaveStart
+	}
+	half := (approxActiveAtWaveStart + 1) / 2
+	if half < massDisconnectMinCount {
+		return massDisconnectMinCount
+	}
+	return half
+}
+
+// isPunitiveRemovalReason reports removals for invalid peer data or protocol faults.
+func isPunitiveRemovalReason(reason string) bool {
+	r := strings.ToLower(reason)
+	punitiveHints := []string{
+		"too many failures",
+		"nil block",
+		"invalid block",
+		"invalid stream",
+		"protocol error",
+		"critical protocol",
+		"zero hashes",
+		"all zero hashes",
+		"empty blockbytes",
+		"not valid",
+		"unexpected blockbytes",
+		"unmatched number",
+		"unverifiable",
+		"zero bytes block",
+		"expected more hashes",
+	}
+	for _, hint := range punitiveHints {
+		if strings.Contains(r, hint) {
+			return true
+		}
+	}
+	return false
+}
+
+// isConnectionLossReason reports removals for transport or local-network failures.
+// These use soft reconnect while a local-outage window is active.
+func isConnectionLossReason(reason string) bool {
+	if reason == "" {
+		return false
+	}
+	r := strings.ToLower(reason)
+	if isPunitiveRemovalReason(r) {
+		return false
+	}
+	connectionHints := []string{
+		"read msg failed",
+		"remote closed",
+		"progress timeout",
+		"too many recoverable errors",
+		"stream error",
+		"connection reset",
+		"broken pipe",
+		"local network",
+		"network error",
+		"disconnect",
+	}
+	for _, hint := range connectionHints {
+		if strings.Contains(r, hint) {
+			return true
+		}
+	}
+	return false
+}