notify remote peer to close session

common/mux/client.go

@@ -322,6 +322,10 @@ func (m *ClientWorker) handleStatusKeep(meta *FrameMetadata, reader *buf.Buffere
 
 	s, found := m.sessionManager.Get(meta.SessionID)
 	if !found {
+		// Notify remote peer to close this session.
+		closingWriter := NewResponseWriter(meta.SessionID, m.link.Writer, protocol.TransferTypeStream)
+		closingWriter.Close()
+
 		return buf.Copy(NewStreamReader(reader), buf.Discard)
 	}
 
@@ -330,6 +334,10 @@ func (m *ClientWorker) handleStatusKeep(meta *FrameMetadata, reader *buf.Buffere
 	if err != nil && buf.IsWriteError(err) {
 		newError("failed to write to downstream. closing session ", s.ID).Base(err).WriteToLog()
 
+		// Notify remote peer to close this session.
+		closingWriter := NewResponseWriter(meta.SessionID, m.link.Writer, protocol.TransferTypeStream)
+		closingWriter.Close()
+
 		drainErr := buf.Copy(rr, buf.Discard)
 		pipe.CloseError(s.input)
 		s.Close()

common/mux/server.go

@@ -159,6 +159,10 @@ func (w *ServerWorker) handleStatusKeep(meta *FrameMetadata, reader *buf.Buffere
 
 	s, found := w.sessionManager.Get(meta.SessionID)
 	if !found {
+		// Notify remote peer to close this session.
+		closingWriter := NewResponseWriter(meta.SessionID, w.link.Writer, protocol.TransferTypeStream)
+		closingWriter.Close()
+
 		return buf.Copy(NewStreamReader(reader), buf.Discard)
 	}
 
@@ -166,7 +170,11 @@ func (w *ServerWorker) handleStatusKeep(meta *FrameMetadata, reader *buf.Buffere
 	err := buf.Copy(rr, s.output)
 
 	if err != nil && buf.IsWriteError(err) {
-		newError("failed to write to downstream writer. closing session ", s.ID).Base(err)
+		newError("failed to write to downstream writer. closing session ", s.ID).Base(err).WriteToLog()
+
+		// Notify remote peer to close this session.
+		closingWriter := NewResponseWriter(meta.SessionID, w.link.Writer, protocol.TransferTypeStream)
+		closingWriter.Close()
 
 		drainErr := buf.Copy(rr, buf.Discard)
 		pipe.CloseError(s.input)