Clear decoder buffer

gateway/encoding.go

@@ -247,11 +247,7 @@ func (r *RPCSendBlocks) encodeResponse(e *types.Encoder) {
 	}
 }
 func (r *RPCSendBlocks) decodeResponse(d *types.Decoder) {
-	if n := d.ReadPrefix(); n <= cap(r.Blocks) {
-		r.Blocks = r.Blocks[:n]
-	} else {
-		r.Blocks = make([]types.Block, n)
-	}
+	r.Blocks = make([]types.Block, d.ReadPrefix())
 	for i := range r.Blocks {
 		(*types.V1Block)(&r.Blocks[i]).DecodeFrom(d)
 	}

types/encoding.go

@@ -136,8 +136,15 @@ func (d *Decoder) Read(p []byte) (int, error) {
 		if want > len(d.buf) {
 			want = len(d.buf)
 		}
-		var read int
-		read, d.err = io.ReadFull(&d.lr, d.buf[:want])
+		read, err := io.ReadFull(&d.lr, d.buf[:want])
+		if err != nil {
+			// When the decoder encounters an error, it must clear its buffer and
+			// return empty values for the remaining decodes. Because the decoder uses
+			// sticky errors instead of immediately returning a bad decode can
+			// potentially cause a massive allocation.
+			d.SetErr(err)
+			return n, err
+		}
 		n += copy(p[n:], d.buf[:read])
 	}
 	return n, d.err

types/encoding_test.go

@@ -0,0 +1,46 @@
+package types
+
+import (
+	"io"
+	"testing"
+)
+
+type readErrorer struct {
+	err error
+	r   io.Reader
+}
+
+func (r *readErrorer) Read(p []byte) (n int, err error) {
+	if r.err != nil {
+		return 0, r.err
+	}
+	return r.r.Read(p)
+}
+
+func TestDecoderError(t *testing.T) {
+	r, w := io.Pipe()
+	re := &readErrorer{r: r}
+	enc := NewEncoder(w)
+	d := NewDecoder(io.LimitedReader{R: re, N: 1e6})
+
+	go func() {
+		// writing to the pipe blocks until we read from it
+		enc.WritePrefix(1000)
+		enc.Flush()
+	}()
+
+	// read the value from the encoder
+	n := d.ReadPrefix()
+	if n != 1000 {
+		t.Fatalf("expected 1000, got %d", n)
+	}
+
+	// set the error and try to read again
+	re.err = io.EOF
+
+	// should return 0 since the decoder errored
+	n = d.ReadPrefix()
+	if n != 0 {
+		t.Fatalf("expected 0, got %d", n)
+	}
+}