plumbing: fix sideband demux on flush

Canonical Git implementation terminates the sideband stream with a
flush. If the demux scanner is not terminated, it can hang the
connection. Returns io.EOF when encountering a flush pkt.

https://github.com/git/git/blob/master/upload-pack.c#L460

2 files changed, 28 insertions, 3 deletions

diff --git a/plumbing/protocol/packp/sideband/demux.go b/plumbing/protocol/packp/sideband/demux.go
index 0116f96..01d95a3 100644
--- a/plumbing/protocol/packp/sideband/demux.go
+++ b/plumbing/protocol/packp/sideband/demux.go
@@ -114,7 +114,7 @@ func (d *Demuxer) nextPackData() ([]byte, error) {
 
 	size := len(content)
 	if size == 0 {
-		return nil, nil
+		return nil, io.EOF
 	} else if size > d.max {
 		return nil, ErrMaxPackedExceeded
 	}
diff --git a/plumbing/protocol/packp/sideband/demux_test.go b/plumbing/protocol/packp/sideband/demux_test.go
index 8f23353..1ba3ad9 100644
--- a/plumbing/protocol/packp/sideband/demux_test.go
+++ b/plumbing/protocol/packp/sideband/demux_test.go
@@ -105,8 +105,34 @@ func (s *SidebandSuite) TestDecodeWithProgress(c *C) {
 	c.Assert(progress, DeepEquals, []byte{'F', 'O', 'O', '\n'})
 }
 
-func (s *SidebandSuite) TestDecodeWithUnknownChannel(c *C) {
+func (s *SidebandSuite) TestDecodeFlushEOF(c *C) {
+	expected := []byte("abcdefghijklmnopqrstuvwxyz")
+
+	input := bytes.NewBuffer(nil)
+	e := pktline.NewEncoder(input)
+	e.Encode(PackData.WithPayload(expected[0:8]))
+	e.Encode(ProgressMessage.WithPayload([]byte{'F', 'O', 'O', '\n'}))
+	e.Encode(PackData.WithPayload(expected[8:16]))
+	e.Encode(PackData.WithPayload(expected[16:26]))
+	e.Flush()
+	e.Encode(PackData.WithPayload([]byte("bar\n")))
+
+	output := bytes.NewBuffer(nil)
+	content := bytes.NewBuffer(nil)
+	d := NewDemuxer(Sideband64k, input)
+	d.Progress = output
+
+	n, err := content.ReadFrom(d)
+	c.Assert(err, IsNil)
+	c.Assert(n, Equals, int64(26))
+	c.Assert(content.Bytes(), DeepEquals, expected)
 
+	progress, err := io.ReadAll(output)
+	c.Assert(err, IsNil)
+	c.Assert(progress, DeepEquals, []byte{'F', 'O', 'O', '\n'})
+}
+
+func (s *SidebandSuite) TestDecodeWithUnknownChannel(c *C) {
 	buf := bytes.NewBuffer(nil)
 	e := pktline.NewEncoder(buf)
 	e.Encode([]byte{'4', 'F', 'O', 'O', '\n'})
@@ -150,5 +176,4 @@ func (s *SidebandSuite) TestDecodeErrMaxPacked(c *C) {
 	n, err := io.ReadFull(d, content)
 	c.Assert(err, Equals, ErrMaxPackedExceeded)
 	c.Assert(n, Equals, 0)
-
 }