Improve Stream Closure Logs, Remove advertise logs (#4895)

* add reason for stream closure

* fix p2p stream tests, add reason to test stream closure

* adjust log level of advertise messages to debug for cleaner output

* remove logs for stream removal cool down

Author: GheisMohammadi

p2p/stream/common/requestmanager/interface_test.go

@@ -107,7 +107,7 @@ func (st *testStream) ProtoSpec() (sttypes.ProtoSpec, error) {
 	return sttypes.ProtoIDToProtoSpec(testProtoID)
 }
 
-func (st *testStream) Close() error {
+func (st *testStream) Close(reason string) error {
 	return nil
 }
 

p2p/stream/common/streammanager/interface_test.go

@@ -83,7 +83,7 @@ func (st *testStream) ResetFailedTimes() {
 	return
 }
 
-func (st *testStream) Close() error {
+func (st *testStream) Close(reason string) error {
 	if st.closed {
 		return errors.New("already closed")
 	}

p2p/stream/common/streammanager/streammanager.go

@@ -212,7 +212,7 @@ func (sm *streamManager) loop() {
 // NewStream handles a new stream from stream handler protocol
 func (sm *streamManager) NewStream(stream sttypes.Stream) error {
 	if err := sm.sanityCheckStream(stream); err != nil {
-		stream.Close()
+		stream.Close("stream sanity check failed")
 		return errors.Wrap(err, "stream sanity check failed")
 	}
 	task := addStreamTask{

p2p/stream/protocols/sync/protocol.go

@@ -200,7 +200,7 @@ func (p *Protocol) HandleStream(raw libp2p_network.Stream) {
 	st := p.wrapStream(raw)
 	if err := p.sm.NewStream(st); err != nil {
 		// Possibly we have reach the hard limit of the stream
-		if !errors.Is(err, streammanager.ErrStreamAlreadyExist) {
+		if !errors.Is(err, streammanager.ErrStreamAlreadyExist) && !errors.Is(err, streammanager.ErrStreamRemovalNotExpired) {
 			p.logger.Warn().Err(err).Str("stream ID", string(st.ID())).
 				Msg("failed to add new stream")
 		}
@@ -281,7 +281,7 @@ func (p *Protocol) advertise() time.Duration {
 
 			if err == nil {
 				newPeersDiscovered = true
-				p.logger.Info().
+				p.logger.Debug().
 					Str("protocol", string(pid)).
 					Dur("elapsed(sec)", time.Duration(elapsed.Seconds())).
 					Int("retry", retries).
@@ -319,7 +319,7 @@ func (p *Protocol) advertise() time.Duration {
 		}
 
 		if err != nil {
-			p.logger.Error().Err(err).
+			p.logger.Debug().Err(err).
 				Str("protocol", string(pid)).
 				Msg("Advertise failed after retries")
 			continue
@@ -379,7 +379,7 @@ func (p *Protocol) RemoveStream(stID sttypes.StreamID, reason string) {
 	st, exist := p.sm.GetStreamByID(stID)
 	if exist && st != nil {
 		//TODO: log this incident with reason
-		st.Close()
+		st.Close(reason)
 		p.logger.Info().
 			Str("stream ID", string(stID)).
 			Str("reason", reason).
@@ -397,7 +397,7 @@ func (p *Protocol) StreamFailed(stID sttypes.StreamID, reason string) {
 			Str("reason", reason).
 			Msg("stream failed")
 		if st.Failures() >= MaxStreamFailures {
-			st.Close()
+			st.Close("too many failures")
 			p.logger.Warn().
 				Str("stream ID", string(st.ID())).
 				Str("reason", "too many failures").

p2p/stream/protocols/sync/stream.go

@@ -74,7 +74,7 @@ func (st *syncStream) readMsgLoop() {
 		default:
 			msg, err := st.readMsg()
 			if err != nil {
-				if err := st.Close(); err != nil {
+				if err := st.Close("read msg failed"); err != nil {
 					st.logger.Err(err).Msg("failed to close sync stream")
 				}
 				return
@@ -125,7 +125,7 @@ func (st *syncStream) handleReqLoop() {
 			if err != nil {
 				st.logger.Info().Err(err).Str("request", req.String()).
 					Msg("handle request error. Closing stream")
-				if err := st.Close(); err != nil {
+				if err := st.Close("handle request error"); err != nil {
 					st.logger.Err(err).Msg("failed to close sync stream")
 				}
 				return
@@ -150,13 +150,13 @@ func (st *syncStream) handleRespLoop() {
 }
 
 // Close stops the stream handling and closes the underlying stream
-func (st *syncStream) Close() error {
+func (st *syncStream) Close(reason string) error {
 	notClosed := atomic.CompareAndSwapUint32(&st.closeStat, 0, 1)
 	if !notClosed {
 		// Already closed by another goroutine. Directly return
 		return nil
 	}
-	if err := st.protocol.sm.RemoveStream(st.ID(), "force close"); err != nil {
+	if err := st.protocol.sm.RemoveStream(st.ID(), "force close: "+reason); err != nil {
 		st.logger.Err(err).Str("stream ID", string(st.ID())).
 			Msg("failed to remove sync stream on close")
 	}

p2p/stream/types/stream.go

@@ -20,7 +20,7 @@ type Stream interface {
 	ProtoSpec() (ProtoSpec, error)
 	WriteBytes([]byte) error
 	ReadBytes() ([]byte, error)
-	Close() error
+	Close(reason string) error
 	CloseOnExit() error
 	Failures() int32
 	AddFailedTimes(faultRecoveryThreshold time.Duration)