diff --git a/api/service/synchronize/stagedstreamsync/const.go b/api/service/synchronize/stagedstreamsync/const.go
index f8b4369203..d79a285a6e 100644
--- a/api/service/synchronize/stagedstreamsync/const.go
+++ b/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
diff --git a/cmd/config/config_migrations_test.go b/cmd/config/config_migrations_test.go
index f07a810d8d..0b0c8dd5a7 100644
--- a/cmd/config/config_migrations_test.go
+++ b/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,
diff --git a/cmd/config/config_test.go b/cmd/config/config_test.go
index 5be64e1e07..a69b988b3f 100644
--- a/cmd/config/config_test.go
+++ b/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)
 }
 
diff --git a/cmd/config/default.go b/cmd/config/default.go
index c8cabd5f0b..f7b3866d9d 100644
--- a/cmd/config/default.go
+++ b/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{},
diff --git a/cmd/config/flags.go b/cmd/config/flags.go
index 414009c2bc..d0a018f18a 100644
--- a/cmd/config/flags.go
+++ b/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{
diff --git a/internal/configs/harmony/harmony.go b/internal/configs/harmony/harmony.go
index 1232703445..f12970b972 100644
--- a/internal/configs/harmony/harmony.go
+++ b/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
diff --git a/p2p/stream/common/streammanager/config.go b/p2p/stream/common/streammanager/config.go
index 480c43353e..93dce50b79 100644
--- a/p2p/stream/common/streammanager/config.go
+++ b/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
diff --git a/p2p/stream/common/streammanager/mass_disconnect.go b/p2p/stream/common/streammanager/mass_disconnect.go
new file mode 100644
index 0000000000..208fdb029d
--- /dev/null
+++ b/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
+}
diff --git a/p2p/stream/common/streammanager/mass_disconnect_test.go b/p2p/stream/common/streammanager/mass_disconnect_test.go
new file mode 100644
index 0000000000..1f3df57a17
--- /dev/null
+++ b/p2p/stream/common/streammanager/mass_disconnect_test.go
@@ -0,0 +1,258 @@
+package streammanager
+
+import (
+	"testing"
+	"time"
+)
+
+func TestMassDisconnectThreshold(t *testing.T) {
+	tests := []struct {
+		active int
+		want   int
+	}{
+		{0, massDisconnectMinCount},
+		{1, 1},
+		{2, 2},
+		{3, massDisconnectMinCount},
+		{4, massDisconnectMinCount},
+		{8, 4},
+		{10, 5},
+	}
+	for _, tt := range tests {
+		if got := massDisconnectThreshold(tt.active); got != tt.want {
+			t.Fatalf("active=%d: got %d want %d", tt.active, got, tt.want)
+		}
+	}
+}
+
+func TestIsConnectionLossReason(t *testing.T) {
+	loss := []string{
+		"force close: remote closed stream",
+		"force close: read msg failed",
+		"force close: progress timeout",
+		"force close: connection reset",
+		"force close: broken pipe",
+	}
+	for _, r := range loss {
+		if !isConnectionLossReason(r) {
+			t.Fatalf("expected connection-loss: %q", r)
+		}
+	}
+	punitive := []string{
+		"force close: too many failures",
+		"force close: nil block hashes",
+		"force close: identifySyncedStreams: critical protocol error",
+		"force close: invalid block is received from stream",
+		"force close: downloadRawBlocks received blockBytes are not valid",
+		"reset", // bare / ambiguous reasons are not treated as connection-loss
+	}
+	for _, r := range punitive {
+		if isConnectionLossReason(r) {
+			t.Fatalf("expected punitive/non-loss: %q", r)
+		}
+	}
+}
+
+func TestDisconnectTracker_MassDetectsLocalOutage(t *testing.T) {
+	var dt disconnectTracker
+	now := time.Now()
+
+	// 8 streams: threshold = 4
+	for i := 0; i < 3; i++ {
+		in, entered := dt.observeRemoval(now.Add(time.Duration(i)*time.Second), 8-i)
+		if in || entered {
+			t.Fatalf("removal %d should not trigger local outage yet", i+1)
+		}
+	}
+	in, entered := dt.observeRemoval(now.Add(4*time.Second), 5)
+	if !in || !entered {
+		t.Fatalf("4th removal should enter local outage, in=%v entered=%v", in, entered)
+	}
+	if !dt.inLocalOutage(now.Add(5 * time.Second)) {
+		t.Fatal("expected active local outage")
+	}
+
+	// Subsequent removal while in outage stays in outage without re-entering.
+	in, entered = dt.observeRemoval(now.Add(6*time.Second), 4)
+	if !in || entered {
+		t.Fatalf("while in outage: in=%v entered=%v", in, entered)
+	}
+}
+
+func TestDisconnectTracker_RateLimitsOutageEntry(t *testing.T) {
+	var dt disconnectTracker
+	now := time.Now()
+
+	// Enter outage.
+	for i := 0; i < 4; i++ {
+		dt.observeRemoval(now.Add(time.Duration(i)*time.Second), 8-i)
+	}
+	if !dt.inLocalOutage(now.Add(4 * time.Second)) {
+		t.Fatal("expected first outage")
+	}
+
+	// Expire outage window but stay inside min interval.
+	dt.localOutageUntil = now
+	dt.removalTimes = nil
+	later := now.Add(localOutageDuration + time.Minute) // still < 10m from lastOutageStart
+	for i := 0; i < 4; i++ {
+		in, entered := dt.observeRemoval(later.Add(time.Duration(i)*time.Second), 8-i)
+		if in || entered {
+			t.Fatalf("rate-limited re-entry should not start outage at removal %d", i+1)
+		}
+	}
+
+	// After min interval, a new wave may enter outage again.
+	dt.removalTimes = nil
+	reopen := now.Add(localOutageMinInterval + time.Second)
+	for i := 0; i < 3; i++ {
+		dt.observeRemoval(reopen.Add(time.Duration(i)*time.Second), 8-i)
+	}
+	in, entered := dt.observeRemoval(reopen.Add(4*time.Second), 5)
+	if !in || !entered {
+		t.Fatalf("after min interval should allow new outage, in=%v entered=%v", in, entered)
+	}
+}
+
+func TestDisconnectTracker_SmallSetLosingAllPeers(t *testing.T) {
+	var dt disconnectTracker
+	now := time.Now()
+
+	in, entered := dt.observeRemoval(now, 2)
+	if in || entered {
+		t.Fatal("first of two should not trigger yet")
+	}
+	in, entered = dt.observeRemoval(now.Add(time.Second), 1)
+	if !in || !entered {
+		t.Fatalf("losing both peers should trigger local outage, in=%v entered=%v", in, entered)
+	}
+}
+
+func TestHandleRemoveStream_MassDisconnectSkipsCriticalCooldownForConnectionLoss(t *testing.T) {
+	sm := newTestStreamManager()
+	sm.pf = newTestPeerFinder(nil, emptyDelayFunc)
+	sm.config.HardLoCap = 2
+	sm.config.SoftLoCap = 2
+	sm.Start()
+	defer sm.Close()
+
+	const lossReason = "force close: remote closed stream"
+
+	for i := 1; i <= 4; i++ {
+		if err := sm.NewStream(newTestStream(makeStreamID(i), testProtoID)); err != nil {
+			t.Fatalf("add stream %d: %v", i, err)
+		}
+	}
+
+	if err := sm.RemoveStream(makeStreamID(1), lossReason, true); err != nil {
+		t.Fatalf("remove 1: %v", err)
+	}
+	info, ok := sm.removedStreams.Get(makeStreamID(1))
+	if !ok {
+		t.Fatal("expected removal info for stream 1")
+	}
+	if info.HasExpired() {
+		t.Fatal("first critical connection-loss should still have cooldown before mass detect")
+	}
+
+	if err := sm.RemoveStream(makeStreamID(2), lossReason, true); err != nil {
+		t.Fatalf("remove 2: %v", err)
+	}
+	if err := sm.RemoveStream(makeStreamID(3), lossReason, true); err != nil {
+		t.Fatalf("remove 3: %v", err)
+	}
+
+	info3, ok := sm.removedStreams.Get(makeStreamID(3))
+	if !ok {
+		t.Fatal("expected removal info for stream 3")
+	}
+	if !info3.HasExpired() {
+		t.Fatal("mass-disconnect connection-loss should allow immediate reconnect")
+	}
+	if !sm.disconnectTracker.inLocalOutage(time.Now()) {
+		t.Fatal("expected local outage after mass disconnect")
+	}
+
+	if err := sm.RemoveStream(makeStreamID(4), lossReason, true); err != nil {
+		t.Fatalf("remove 4: %v", err)
+	}
+	info4, ok := sm.removedStreams.Get(makeStreamID(4))
+	if !ok {
+		t.Fatal("expected removal info for stream 4")
+	}
+	if !info4.HasExpired() {
+		t.Fatal("connection-loss during local outage should allow immediate reconnect")
+	}
+}
+
+func TestHandleRemoveStream_PunitiveKeepsCriticalCooldownDuringOutage(t *testing.T) {
+	sm := newTestStreamManager()
+	sm.pf = newTestPeerFinder(nil, emptyDelayFunc)
+	sm.config.HardLoCap = 2
+	sm.config.SoftLoCap = 2
+	sm.Start()
+	defer sm.Close()
+
+	const lossReason = "force close: remote closed stream"
+	const badReason = "force close: identifySyncedStreams: critical protocol error"
+
+	for i := 1; i <= 5; i++ {
+		if err := sm.NewStream(newTestStream(makeStreamID(i), testProtoID)); err != nil {
+			t.Fatalf("add stream %d: %v", i, err)
+		}
+	}
+
+	// Trigger local outage with connection-loss removals.
+	for i := 1; i <= 3; i++ {
+		if err := sm.RemoveStream(makeStreamID(i), lossReason, true); err != nil {
+			t.Fatalf("loss remove %d: %v", i, err)
+		}
+	}
+	if !sm.disconnectTracker.inLocalOutage(time.Now()) {
+		t.Fatal("expected local outage")
+	}
+
+	if err := sm.RemoveStream(makeStreamID(4), badReason, true); err != nil {
+		t.Fatalf("punitive remove: %v", err)
+	}
+	info, ok := sm.removedStreams.Get(makeStreamID(4))
+	if !ok {
+		t.Fatal("expected removal info for punitive peer")
+	}
+	if info.HasExpired() {
+		t.Fatal("punitive removal during local outage must keep critical cooldown")
+	}
+}
+
+func TestHandleRemoveStream_PunitiveDoesNotTriggerMassDisconnect(t *testing.T) {
+	sm := newTestStreamManager()
+	sm.pf = newTestPeerFinder(nil, emptyDelayFunc)
+	sm.config.HardLoCap = 2
+	sm.config.SoftLoCap = 2
+	sm.Start()
+	defer sm.Close()
+
+	const badReason = "force close: nil block hashes"
+	for i := 1; i <= 4; i++ {
+		if err := sm.NewStream(newTestStream(makeStreamID(i), testProtoID)); err != nil {
+			t.Fatalf("add stream %d: %v", i, err)
+		}
+	}
+	for i := 1; i <= 4; i++ {
+		if err := sm.RemoveStream(makeStreamID(i), badReason, true); err != nil {
+			t.Fatalf("remove %d: %v", i, err)
+		}
+	}
+	if sm.disconnectTracker.inLocalOutage(time.Now()) {
+		t.Fatal("punitive removals must not open a local-outage window")
+	}
+	for i := 1; i <= 4; i++ {
+		info, ok := sm.removedStreams.Get(makeStreamID(i))
+		if !ok {
+			t.Fatalf("missing removal info %d", i)
+		}
+		if info.HasExpired() {
+			t.Fatalf("punitive peer %d should still be on critical cooldown", i)
+		}
+	}
+}
diff --git a/p2p/stream/common/streammanager/streammanager.go b/p2p/stream/common/streammanager/streammanager.go
index 69303a118d..9a572fd35a 100644
--- a/p2p/stream/common/streammanager/streammanager.go
+++ b/p2p/stream/common/streammanager/streammanager.go
@@ -88,6 +88,9 @@ type streamManager struct {
 	numTrustedStreamsMain     int64 // Count of trusted streams in main list
 	numTrustedStreamsReserved int64 // Count of trusted streams in reserved list
 
+	// disconnectTracker tracks clustered stream removals and local-outage windows.
+	disconnectTracker disconnectTracker
+
 	// callback for when enough streams are found
 	enoughStreamsCallback func()
 }
@@ -135,7 +138,7 @@ func (rm *RemovalInfo) HasExpired() bool {
 	rm.mu.RLock()
 	defer rm.mu.RUnlock()
 
-	return time.Now().After(rm.expireAt)
+	return !time.Now().Before(rm.expireAt)
 }
 
 // BumpCount increases the removal count.
@@ -154,6 +157,18 @@ func (rm *RemovalInfo) ResetCount() {
 	rm.count = 0
 }
 
+// MarkRemovedForLocalOutage records a connection-loss removal during a local-outage
+// window. The peer may reconnect immediately and the removal count is reset.
+func (rm *RemovalInfo) MarkRemovedForLocalOutage() {
+	rm.mu.Lock()
+	defer rm.mu.Unlock()
+
+	now := time.Now()
+	rm.removedAt = now
+	rm.expireAt = now
+	rm.count = 0
+}
+
 // SetEnoughStreamsCallback sets the callback function to be called when enough streams are found
 func (sm *streamManager) SetEnoughStreamsCallback(callback func()) {
 	sm.enoughStreamsCallback = callback
@@ -595,104 +610,68 @@ func (sm *streamManager) addStreamFromReserved(count int) (int, error) {
 }
 
 func (sm *streamManager) handleRemoveStream(id sttypes.StreamID, reason string, criticalErr bool) error {
-	// Check which set contains the stream - only delete from the one that has it
 	st, inMain := sm.streams.get(id)
+	inReserved := false
 	if !inMain {
-		// Try reserved streams
-		st, inReserved := sm.reservedStreams.get(id)
+		st, inReserved = sm.reservedStreams.get(id)
 		if !inReserved {
 			return ErrStreamAlreadyRemoved
 		}
-		// Stream is in reserved list
-		if criticalErr {
-			streamCriticalErrorCounterVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Inc()
-		}
-
-		// Check if this is a trusted stream and handle accordingly
-		_, isTrusted := sm.trustedStreams.Get(id)
-		if isTrusted {
-			if criticalErr {
-				// Trusted streams with critical errors are not removed (protected)
-				sm.logger.Info().
-					Str("protocolID", string(sm.myProtoID)).
-					Uint32("shardID", uint32(sm.myProtoSpec.ShardID)).
-					Int("NumStreams", sm.streams.size()).
-					Int("NumTrustedStreamsMain", int(atomic.LoadInt64(&sm.numTrustedStreamsMain))).
-					Int("NumTrustedStreamsReserved", int(atomic.LoadInt64(&sm.numTrustedStreamsReserved))).
-					Interface("StreamID", id).
-					Bool("trusted", true).
-					Str("reason", reason).
-					Bool("criticalErr", criticalErr).
-					Msg("[StreamManager] trusted peer got critical error but not removed")
-				return nil
-			}
-			// Trusted stream with non-critical error: remove from trustedStreams map and update counters
-			sm.trustedStreams.Delete(id)
-			atomic.AddInt64(&sm.numTrustedStreamsReserved, -1)
-		}
+	}
 
-		sm.reservedStreams.deleteStream(st)
+	now := time.Now()
+	connectionLoss := isConnectionLossReason(reason)
+	_, isTrusted := sm.trustedStreams.Get(id)
+	// Trusted streams with critical errors stay registered unless the removal is a
+	// connection loss during an active local-outage window.
+	if isTrusted && criticalErr && !(sm.disconnectTracker.inLocalOutage(now) && connectionLoss) {
+		streamCriticalErrorCounterVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Inc()
 		sm.logger.Info().
 			Str("protocolID", string(sm.myProtoID)).
 			Uint32("shardID", uint32(sm.myProtoSpec.ShardID)).
 			Int("NumStreams", sm.streams.size()).
-			Int("NumReservedStreams", sm.reservedStreams.size()).
 			Int("NumTrustedStreamsMain", int(atomic.LoadInt64(&sm.numTrustedStreamsMain))).
 			Int("NumTrustedStreamsReserved", int(atomic.LoadInt64(&sm.numTrustedStreamsReserved))).
 			Interface("StreamID", id).
+			Bool("trusted", true).
+			Bool("reserved", !inMain).
 			Str("reason", reason).
 			Bool("criticalErr", criticalErr).
-			Bool("trusted", isTrusted).
-			Msg("[StreamManager] removed stream from reserved streams list")
+			Msg("[StreamManager] trusted peer got critical error but not removed")
+		return nil
+	}
 
-		info, exist := sm.removedStreams.Get(id)
-		if !exist {
-			info = &RemovalInfo{count: 0}
-			sm.removedStreams.Set(id, info)
+	activeBefore := sm.streams.size() + sm.reservedStreams.size()
+	inLocalOutage := sm.disconnectTracker.inLocalOutage(now)
+	justEntered := false
+	if connectionLoss {
+		inLocalOutage, justEntered = sm.disconnectTracker.observeRemoval(now, activeBefore)
+		if justEntered {
+			sm.onLocalOutageDetected(activeBefore, reason)
 		}
-		info.MarkAsRemoved(criticalErr)
-
-		sm.removeStreamFeed.Send(EvtStreamRemoved{id})
-		removedStreamsCounterVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Inc()
-		streamRemovalReasonCounterVec.With(prometheus.Labels{"reason": reason, "critical": strconv.FormatBool(criticalErr)}).Inc()
-		numStreamsGaugeVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Set(float64(sm.streams.size()))
-		numReservedStreamsGaugeVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Set(float64(sm.reservedStreams.size()))
-		numTrustedPeerStreamsGaugeVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Set(float64(atomic.LoadInt64(&sm.numTrustedStreamsMain)))
-		numReservedTrustedPeerStreamsGaugeVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Set(float64(atomic.LoadInt64(&sm.numTrustedStreamsReserved)))
-
-		sm.tryToReplaceRemovedStream()
-		return nil
 	}
 
-	// Stream is in main list
-	if criticalErr {
+	// Soft reconnect applies to connection-loss removals during a local-outage window.
+	softReconnect := inLocalOutage && connectionLoss
+	effectiveCritical := criticalErr && !softReconnect
+	if effectiveCritical {
 		streamCriticalErrorCounterVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Inc()
 	}
 
-	// Check if this is a trusted stream and handle accordingly
-	_, isTrusted := sm.trustedStreams.Get(id)
 	if isTrusted {
-		if criticalErr {
-			// Trusted streams with critical errors are not removed (protected)
-			sm.logger.Info().
-				Str("protocolID", string(sm.myProtoID)).
-				Uint32("shardID", uint32(sm.myProtoSpec.ShardID)).
-				Int("NumStreams", sm.streams.size()).
-				Int("NumTrustedStreamsMain", int(atomic.LoadInt64(&sm.numTrustedStreamsMain))).
-				Int("NumTrustedStreamsReserved", int(atomic.LoadInt64(&sm.numTrustedStreamsReserved))).
-				Interface("StreamID", id).
-				Bool("trusted", true).
-				Str("reason", reason).
-				Bool("criticalErr", criticalErr).
-				Msg("[StreamManager] trusted peer got critical error but not removed")
-			return nil
-		}
-		// Trusted stream with non-critical error: remove from trustedStreams map and update counters
 		sm.trustedStreams.Delete(id)
-		atomic.AddInt64(&sm.numTrustedStreamsMain, -1)
+		if inMain {
+			atomic.AddInt64(&sm.numTrustedStreamsMain, -1)
+		} else {
+			atomic.AddInt64(&sm.numTrustedStreamsReserved, -1)
+		}
 	}
 
-	sm.streams.deleteStream(st)
+	if inMain {
+		sm.streams.deleteStream(st)
+	} else {
+		sm.reservedStreams.deleteStream(st)
+	}
 
 	sm.logger.Info().
 		Str("protocolID", string(sm.myProtoID)).
@@ -704,29 +683,71 @@ func (sm *streamManager) handleRemoveStream(id sttypes.StreamID, reason string,
 		Interface("StreamID", id).
 		Str("reason", reason).
 		Bool("criticalErr", criticalErr).
+		Bool("effectiveCritical", effectiveCritical).
+		Bool("connectionLoss", connectionLoss).
+		Bool("softReconnect", softReconnect).
+		Bool("localOutage", inLocalOutage).
 		Bool("trusted", isTrusted).
-		Msg("[StreamManager] removed stream from main streams list")
+		Bool("reserved", !inMain).
+		Msg("[StreamManager] removed stream")
+
+	sm.recordStreamRemoval(id, effectiveCritical, softReconnect, reason)
+	sm.tryToReplaceRemovedStream()
+	return nil
+}
 
+func (sm *streamManager) recordStreamRemoval(id sttypes.StreamID, effectiveCritical, softReconnect bool, reason string) {
 	info, exist := sm.removedStreams.Get(id)
 	if !exist {
 		info = &RemovalInfo{count: 0}
 		sm.removedStreams.Set(id, info)
 	}
-	info.MarkAsRemoved(criticalErr)
+	if softReconnect {
+		info.MarkRemovedForLocalOutage()
+	} else {
+		info.MarkAsRemoved(effectiveCritical)
+	}
 
-	// try to replace removed streams from reserved list
 	sm.removeStreamFeed.Send(EvtStreamRemoved{id})
-
 	removedStreamsCounterVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Inc()
-	streamRemovalReasonCounterVec.With(prometheus.Labels{"reason": reason, "critical": strconv.FormatBool(criticalErr)}).Inc()
+	streamRemovalReasonCounterVec.With(prometheus.Labels{
+		"reason":   reason,
+		"critical": strconv.FormatBool(effectiveCritical),
+	}).Inc()
 	numStreamsGaugeVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Set(float64(sm.streams.size()))
 	numReservedStreamsGaugeVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Set(float64(sm.reservedStreams.size()))
 	numTrustedPeerStreamsGaugeVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Set(float64(atomic.LoadInt64(&sm.numTrustedStreamsMain)))
 	numReservedTrustedPeerStreamsGaugeVec.With(prometheus.Labels{"topic": string(sm.myProtoID)}).Set(float64(atomic.LoadInt64(&sm.numTrustedStreamsReserved)))
+}
 
-	sm.tryToReplaceRemovedStream()
+func (sm *streamManager) onLocalOutageDetected(activeBefore int, reason string) {
+	sm.logger.Warn().
+		Int("activeBefore", activeBefore).
+		Int("removalsInWindow", len(sm.disconnectTracker.removalTimes)).
+		Dur("outageDuration", localOutageDuration).
+		Dur("discHoldoff", localOutageDiscHoldoff).
+		Dur("minInterval", localOutageMinInterval).
+		Str("lastReason", reason).
+		Msg("[StreamManager] mass disconnect detected; local-outage window started")
 
-	return nil
+	sm.coolDownCache.Reset()
+
+	sm.coolDown.Set()
+	go func() {
+		timer := time.NewTimer(localOutageDiscHoldoff)
+		defer timer.Stop()
+		select {
+		case <-timer.C:
+			sm.coolDown.UnSet()
+			select {
+			case sm.discCh <- discTask{}:
+			default:
+			}
+			sm.logger.Info().Msg("[StreamManager] local outage discovery holdoff ended; rediscovery triggered")
+		case <-sm.ctx.Done():
+			sm.coolDown.UnSet()
+		}
+	}()
 }
 
 func (sm *streamManager) tryToReplaceRemovedStream() error {
@@ -756,8 +777,7 @@ func (sm *streamManager) handleResetForWatchdog() error {
 	mainStreams := sm.streams.size()
 	reservedStreams := sm.reservedStreams.size()
 
-	// Keep active connections intact. Clear cooldown/blocking state so peer
-	// discovery can retry candidates again.
+	// Preserve active connections. Clear cooldown and discovery-blocking state.
 	sm.removedStreams.Clear()
 	sm.coolDownCache.Reset()
 	sm.coolDown.UnSet()
@@ -1082,23 +1102,33 @@ func (sm *streamManager) discoverAndSetupStream(discCtx context.Context) (int, e
 				Int("NumTrustedStreamsReserved", int(atomic.LoadInt64(&sm.numTrustedStreamsReserved))).
 				Msg("[discoverAndSetupStream] processing trusted peers for bootstrap stream setup")
 
-			// Setup trusted streams - this function handles batching and waiting
+			streamsBefore := sm.streams.size()
 			successCount := sm.setupTrustedStreams(discCtx, trustedPeers, trustedMinPeers)
 			connectedTrustedStreams = successCount
 
+			if successCount > 0 {
+				expected := streamsBefore + successCount
+				if expected > sm.config.HardLoCap {
+					expected = sm.config.HardLoCap
+				}
+				sm.waitForStreamRegistrations(discCtx, expected)
+			}
+
 			sm.logger.Info().
 				Str("protocolID", string(sm.myProtoID)).
 				Uint32("shardID", uint32(sm.myProtoSpec.ShardID)).
 				Int("successCount", successCount).
 				Int("trustedMinPeers", trustedMinPeers).
+				Int("registeredStreams", sm.streams.size()).
 				Int("NumTrustedStreamsMain", int(atomic.LoadInt64(&sm.numTrustedStreamsMain))).
 				Int("NumTrustedStreamsReserved", int(atomic.LoadInt64(&sm.numTrustedStreamsReserved))).
-				Msg("[discoverAndSetupStream] completed trusted peer stream setup, proceeding to discover other peers")
+				Msg("[discoverAndSetupStream] completed trusted peer stream setup")
 		}
 	}
 
-	if sm.streams.size()+connectedTrustedStreams >= sm.config.HardLoCap {
-		return connectedTrustedStreams, nil
+	// Skip DHT when enough compatible streams are already registered.
+	if sm.hardHaveEnoughStream() {
+		return sm.streams.size(), nil
 	}
 
 	peers, err := sm.discover(discCtx)
@@ -1141,6 +1171,40 @@ func (sm *streamManager) discoverAndSetupStream(discCtx context.Context) (int, e
 	return connectedTrustedStreams + connecting, nil
 }
 
+// waitForStreamRegistrations waits until target compatible streams are registered,
+// HardLoCap is met, or the wait budget/context expires.
+func (sm *streamManager) waitForStreamRegistrations(ctx context.Context, target int) {
+	if target <= 0 || sm.hardHaveEnoughStream() {
+		return
+	}
+	if sm.streams.numStreamsWithMinProtoSpec(sm.myProtoSpec) >= target {
+		return
+	}
+
+	waitCtx, cancel := context.WithTimeout(ctx, streamRegistrationWait)
+	defer cancel()
+	ticker := time.NewTicker(streamRegistrationPoll)
+	defer ticker.Stop()
+
+	for {
+		if sm.hardHaveEnoughStream() {
+			return
+		}
+		if sm.streams.numStreamsWithMinProtoSpec(sm.myProtoSpec) >= target {
+			return
+		}
+		select {
+		case <-waitCtx.Done():
+			sm.logger.Debug().
+				Int("target", target).
+				Int("registered", sm.streams.numStreamsWithMinProtoSpec(sm.myProtoSpec)).
+				Msg("[StreamManager] timed out waiting for stream registrations; continuing with DHT if needed")
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
 func (sm *streamManager) discover(ctx context.Context) (<-chan libp2p_peer.AddrInfo, error) {
 	numStreams := sm.streams.size()
 
diff --git a/p2p/stream/common/streammanager/streammanager_test.go b/p2p/stream/common/streammanager/streammanager_test.go
index 4882deb1e4..4156c050da 100644
--- a/p2p/stream/common/streammanager/streammanager_test.go
+++ b/p2p/stream/common/streammanager/streammanager_test.go
@@ -1,6 +1,7 @@
 package streammanager
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strings"
@@ -284,6 +285,63 @@ func TestStreamSet_numStreamsWithMinProtoID(t *testing.T) {
 	}
 }
 
+func TestWaitForStreamRegistrations(t *testing.T) {
+	sm := newTestStreamManager()
+	sm.config.HardLoCap = 3
+
+	done := make(chan struct{})
+	go func() {
+		sm.waitForStreamRegistrations(context.Background(), 2)
+		close(done)
+	}()
+
+	time.Sleep(20 * time.Millisecond)
+	select {
+	case <-done:
+		t.Fatal("wait returned before streams were registered")
+	default:
+	}
+
+	sm.streams.addStream(newTestStream(makeStreamID(1), testProtoID))
+	sm.streams.addStream(newTestStream(makeStreamID(2), testProtoID))
+
+	select {
+	case <-done:
+	case <-time.After(defTestWait):
+		t.Fatal("timed out waiting for registration wait to complete")
+	}
+}
+
+func TestDiscoverSkipsDHTOnlyAfterRegisteredHardLoCap(t *testing.T) {
+	sm := newTestStreamManager()
+	sm.config.HardLoCap = 2
+	sm.config.DiscBatch = 4
+	// Empty peer finder: connecting stays 0 when DHT runs.
+	sm.pf = newTestPeerFinder(nil, emptyDelayFunc)
+
+	discovered, err := sm.discoverAndSetupStream(context.Background())
+	if err != nil {
+		t.Fatalf("discover: %v", err)
+	}
+	if discovered != 0 {
+		t.Fatalf("expected DHT path with no peers below HardLoCap, got discovered=%d", discovered)
+	}
+	if sm.hardHaveEnoughStream() {
+		t.Fatal("expected HardLoCap unmet before registrations")
+	}
+
+	sm.streams.addStream(newTestStream(makeStreamID(101), testProtoID))
+	sm.streams.addStream(newTestStream(makeStreamID(102), testProtoID))
+
+	discovered, err = sm.discoverAndSetupStream(context.Background())
+	if err != nil {
+		t.Fatalf("discover after hard cap: %v", err)
+	}
+	if discovered != sm.streams.size() {
+		t.Fatalf("expected skip to return registered stream count %d, got %d", sm.streams.size(), discovered)
+	}
+}
+
 func assertError(got, exp error) error {
 	if (got == nil) != (exp == nil) {
 		return fmt.Errorf("unexpected error: %v / %v", got, exp)
diff --git a/p2p/stream/protocols/sync/const.go b/p2p/stream/protocols/sync/const.go
index 6f117a8a43..11bc432fe1 100644
--- a/p2p/stream/protocols/sync/const.go
+++ b/p2p/stream/protocols/sync/const.go
@@ -96,8 +96,9 @@ const (
 	MaxBackoffStartup           = 15 * time.Second  // 15 seconds max backoff
 
 	// Advertisement loop timing constants
-	MinSleepTimeNormal  = 30 * time.Second // Minimum sleep time in normal mode
-	MaxSleepTimeNormal  = 60 * time.Minute // Maximum sleep time in normal mode
+	MinSleepTimeNormal = 30 * time.Second // Minimum sleep time in normal mode
+	// MaxSleepTimeNormal is used when Sync.MaxAdvertiseWaitTime is unset or non-positive.
+	MaxSleepTimeNormal  = 15 * time.Minute
 	MinSleepTimeStartup = 10 * time.Second // Minimum sleep time in startup mode
 	MaxSleepTimeStartup = 2 * time.Minute  // Maximum sleep time in startup mode
 
@@ -111,10 +112,10 @@ const (
 	// DHT Request Limits - How many peers to request from DHT
 	// These should be higher than target limits because DHT may return invalid peers
 	// Based on stream sync configuration and realistic peer discovery ratios:
-	// - Mainnet: Request 20, expect ~8 valid (40% success rate)
+	// - Mainnet: Request 20, expect ~5 valid (matches InitStreams)
 	// - Testnet: Request 8, expect ~2 valid (25% success rate)
 	// - Devnet: Request 12, expect ~4 valid (33% success rate)
-	DHTRequestLimitMainnet   = 20 // Request 20, expect ~8 valid
+	DHTRequestLimitMainnet   = 20 // Request 20, expect ~5 valid
 	DHTRequestLimitTestnet   = 10 // Request 10, expect ~3 valid
 	DHTRequestLimitPangaea   = 10 // Request 10, expect ~3 valid
 	DHTRequestLimitPartner   = 10 // Request 10, expect ~3 valid
@@ -125,12 +126,12 @@ const (
 	// Target Valid Peer Counts - How many valid peers we want to find
 	// These are the actual peer counts we aim for after filtering
 	// Based on stream sync configuration requirements:
-	// - Mainnet: InitStreams=8, DiscSoftLowCap=8, DiscHardLowCap=6
+	// - Mainnet: InitStreams=5, DiscSoftLowCap=5, DiscHardLowCap=3
 	// - Testnet: InitStreams=3, DiscSoftLowCap=3, DiscHardLowCap=3
 	// - Localnet: InitStreams=4, DiscSoftLowCap=4, DiscHardLowCap=4
 	// - Partner: InitStreams=3, DiscSoftLowCap=3, DiscHardLowCap=3
 	// - Else: InitStreams=4, DiscSoftLowCap=4, DiscHardLowCap=4
-	TargetValidPeersMainnet   = 8 // Target 8 valid peers (matches InitStreams)
+	TargetValidPeersMainnet   = 5 // Target 5 valid peers (matches InitStreams)
 	TargetValidPeersTestnet   = 3 // Target 3 valid peers (matches InitStreams)
 	TargetValidPeersPangaea   = 3 // Target 3 valid peers (testnet-like)
 	TargetValidPeersPartner   = 3 // Target 3 valid peers (matches InitStreams)
diff --git a/p2p/stream/protocols/sync/protocol.go b/p2p/stream/protocols/sync/protocol.go
index 6aa39eb134..19ffa313b5 100644
--- a/p2p/stream/protocols/sync/protocol.go
+++ b/p2p/stream/protocols/sync/protocol.go
@@ -80,7 +80,7 @@ type (
 		Explorer   bool
 		EpochChain bool
 
-		MaxAdvertiseWaitTime int
+		MaxAdvertiseWaitTime int // max minutes between advertisements in normal mode
 		// stream manager config
 		SmSoftLowCap int
 		SmHardLowCap int
@@ -268,7 +268,7 @@ func (p *Protocol) advertiseLoop() {
 			maxSleepTime = MaxSleepTimeStartup
 		} else {
 			minSleepTime = MinSleepTimeNormal
-			maxSleepTime = MaxSleepTimeNormal
+			maxSleepTime = p.maxAdvertiseSleep()
 		}
 
 		sleep := p.advertise()
@@ -305,6 +305,15 @@ func (p *Protocol) advertiseLoop() {
 	}
 }
 
+// maxAdvertiseSleep returns the normal-mode max sleep between advertise cycles.
+// Uses Sync.MaxAdvertiseWaitTime when set; otherwise MaxSleepTimeNormal.
+func (p *Protocol) maxAdvertiseSleep() time.Duration {
+	if p.config.MaxAdvertiseWaitTime > 0 {
+		return time.Duration(p.config.MaxAdvertiseWaitTime) * time.Minute
+	}
+	return MaxSleepTimeNormal
+}
+
 // isValidPeer checks if a discovered peer is valid for our use case
 // TODO: Implement more sophisticated validation logic
 func (p *Protocol) isValidPeer(peer libp2p_peer.AddrInfo) bool {
diff --git a/p2p/stream/protocols/sync/protocol_test.go b/p2p/stream/protocols/sync/protocol_test.go
index 886432ebb6..29d57502be 100644
--- a/p2p/stream/protocols/sync/protocol_test.go
+++ b/p2p/stream/protocols/sync/protocol_test.go
@@ -16,6 +16,21 @@ import (
 	nodeconfig "github.com/harmony-one/harmony/internal/configs/node"
 )
 
+func TestProtocol_MaxAdvertiseSleep(t *testing.T) {
+	p := &Protocol{
+		config: Config{},
+	}
+	if got := p.maxAdvertiseSleep(); got != MaxSleepTimeNormal {
+		t.Fatalf("fallback sleep: got %v want %v", got, MaxSleepTimeNormal)
+	}
+
+	p.config.MaxAdvertiseWaitTime = 7
+	want := 7 * time.Minute
+	if got := p.maxAdvertiseSleep(); got != want {
+		t.Fatalf("configured sleep: got %v want %v", got, want)
+	}
+}
+
 func TestProtocol_Match(t *testing.T) {
 	tests := []struct {
 		targetID protocol.ID