Require trailing flush in pktline.EachSidebandPacketjv-pack-objects-stream-prep

This changes the behavior of pktline.EachSidebandPacket to expect a
trailing flush packet. The old behavior was to expect the stream to
end on a packet boundary.

This is part of
https://gitlab.com/groups/gitlab-com/gl-infra/-/epics/463, where we
need to be able to signal "end of stream" in a natural way and we
cannot use EOF.

There is only one call site for EachSidebandPacket, and that call site
(PackObjectsHook) only consumes byte streams produced by the same
Gitaly process. In other words, this change is safe in spite of being
incompatible with the old behavior, because it can never be exposed to
the old behavior.

Changelog: other

3 files changed, 29 insertions, 5 deletions

diff --git a/internal/git/pktline/pktline.go b/internal/git/pktline/pktline.go
index 6ba3e24b7..532fba857 100644
--- a/internal/git/pktline/pktline.go
+++ b/internal/git/pktline/pktline.go
@@ -163,13 +163,17 @@ type errNotSideband struct{ pkt string }
 
 func (err *errNotSideband) Error() string { return fmt.Sprintf("invalid sideband packet: %q", err.pkt) }
 
-// EachSidebandPacket iterates over a side-band-64k pktline stream. For
-// each packet, it will call fn with the band ID and the packet. Fn must
-// not retain the packet.
+// EachSidebandPacket iterates over a side-band-64k pktline stream until
+// it reaches a flush packet. For each packet, it will call fn with the
+// band ID and the packet. Fn must not retain the packet.
 func EachSidebandPacket(r io.Reader, fn func(byte, []byte) error) error {
 	scanner := NewScanner(r)
 
 	for scanner.Scan() {
+		if IsFlush(scanner.Bytes()) {
+			return nil
+		}
+
 		data := Data(scanner.Bytes())
 		if len(data) == 0 {
 			return &errNotSideband{scanner.Text()}
@@ -179,7 +183,11 @@ func EachSidebandPacket(r io.Reader, fn func(byte, []byte) error) error {
 		}
 	}
 
-	return scanner.Err()
+	if err := scanner.Err(); err != nil {
+		return err
+	}
+
+	return io.ErrUnexpectedEOF
 }
 
 // SingleBandReader unwraps a flush-terminated sideband-64k stream. It
diff --git a/internal/git/pktline/pkt_line_test.go b/internal/git/pktline/pktline_test.go
index fb68062f2..dd26ab875 100644
--- a/internal/git/pktline/pkt_line_test.go
+++ b/internal/git/pktline/pktline_test.go
@@ -280,16 +280,23 @@ func TestEachSidebandPacket(t *testing.T) {
 	}{
 		{
 			desc: "empty",
+			in:   "0000",
 			out:  map[byte]string{},
 		},
 		{
 			desc:     "empty with failing callback: callback does not run",
+			in:       "0000",
 			out:      map[byte]string{},
 			callback: func(byte, []byte) error { panic("oh no") },
 		},
 		{
 			desc: "valid stream",
-			in:   "0008\x00foo0008\x01bar0008\xfequx0008\xffbaz",
+			in:   "0008\x00foo0008\x01bar0008\xfequx0008\xffbaz0000",
+			out:  map[byte]string{0: "foo", 1: "bar", 254: "qux", 255: "baz"},
+		},
+		{
+			desc: "valid stream trailing garbage",
+			in:   "0008\x00foo0008\x01bar0008\xfequx0008\xffbaz0000 garbage!!",
 			out:  map[byte]string{0: "foo", 1: "bar", 254: "qux", 255: "baz"},
 		},
 		{
@@ -299,6 +306,11 @@ func TestEachSidebandPacket(t *testing.T) {
 			err:      callbackError,
 		},
 		{
+			desc: "valid stream except missing flush",
+			in:   "0008\x00foo0008\x01bar0008\xfequx0008\xffbaz",
+			err:  io.ErrUnexpectedEOF,
+		},
+		{
 			desc: "interrupted stream",
 			in:   "ffff\x10hello world!!",
 			err:  io.ErrUnexpectedEOF,
diff --git a/internal/gitaly/service/hook/pack_objects.go b/internal/gitaly/service/hook/pack_objects.go
index 82c396503..f1be31c1d 100644
--- a/internal/gitaly/service/hook/pack_objects.go
+++ b/internal/gitaly/service/hook/pack_objects.go
@@ -204,6 +204,10 @@ func (s *server) runPackObjects(ctx context.Context, w io.Writer, repo *gitalypb
 		return fmt.Errorf("git-pack-objects: stderr: %q err: %w", stderrBuf.String(), err)
 	}
 
+	if err := pktline.WriteFlush(w); err != nil {
+		return err
+	}
+
 	return nil
 }