Repository: fxamacker/cbor (go)

Issue:
bug: Decoder returns no error on short read 
### What version of fxamacker/cbor are you using?
	github.com/fxamacker/cbor/v2 v2.4.0


### Does this issue reproduce with the latest release?
  yes

### What OS and CPU architecture are you using (`go env`)?
GOARCH="amd64"
GOHOSTOS="linux"

<details><summary><code>go env</code> Output</summary><br><pre>
$ go env

</pre></details>

### What did you do?
I'm making a stream transport that chunks data into frames and provides  io.Reader and io.Writer interfaces. 
the Read() implementation is nonblocking and returns 0, nil if no data is yet buffered, but no end-of-stream protocol message has arrived, so io.EOF is *not* returned until the remote peer has sent an end-of-stream message.

<!--
If possible, provide steps and/or code to reproduce the problem. 
-->

here is a link to my code and test:
https://github.com/katzenpost/katzenpost/blob/c676eaa9fd1fb5c2db1bd30106b4c97bab79a47c/stream/stream_test.go#L269



### What did you expect to see?
I expected decoder.Decode(fooType) to return an error or correctly read data until a fooType was decoded

### What did you see instead?
no error was returned and fooType was left uninitialized

Pull request:
Retry in decoder if Read() returns 0 bytes read with nil error
Previously, `Decoder` did not properly handle the scenario of `Read` implementations returning 0 bytes read with nil error.

Go [docs](https://pkg.go.dev/io#Reader) for `io.Reader` discourage `Read` implementations from returning 0 bytes with nil error.  And callers should treat this as if nothing happened.

> Implementations of Read are discouraged from returning a zero byte count with a nil error, except when len(p) == 0. Callers should treat a return of 0 and nil as indicating that nothing happened; in particular it does not indicate EOF. 

This PR:
- Retries the call to `Read` if it returns (0, nil).
- Refactors error handling.
- Adds unit tests to reach 100% coverage of stream.go.  :100: :tada: 

Closes #385 - thanks @mixmasala for reporting this issue.

Patch:
diff --git a/.github/workflows/safer-golangci-lint.yml b/.github/workflows/safer-golangci-lint.yml
index 2603524..75c0c37 100644
--- a/.github/workflows/safer-golangci-lint.yml
+++ b/.github/workflows/safer-golangci-lint.yml
@@ -9,37 +9,30 @@
 # This workflow downloads, verifies, and runs golangci-lint in a
 # deterministic, reviewable, and safe manner.
 #
+# To use:
+#   Step 1. Copy this file into [your_github_repo]/.github/workflows/
+#   Step 2. There's no step 2 if you like the default settings.
+#
+# See golangci-lint docs for more info at
+# https://github.com/golangci/golangci-lint
+#
 # 100% of the script for downloading, installing, and running golangci-lint
 # is embedded in this file. The embedded SHA-256 digest is used to verify the
-# downloaded golangci-lint tarball (golangci-lint-1.49.0-linux-amd64.tar.gz).
+# downloaded golangci-lint tarball (golangci-lint-1.xx.x-linux-amd64.tar.gz).
 #
-# The embedded SHA-256 digest matches golangci-lint-1.49.0-checksums.txt at
+# The embedded SHA-256 digest matches golangci-lint-1.xx.x-checksums.txt at
 # https://github.com/golangci/golangci-lint/releases
 #
-# To use:
-#   Step 1. Copy this file into [github_repo]/.github/workflows/
-#   Step 2. There's no step 2 if you like the default settings.
-#
-# Create and use a config file (.golangci.yml) as described in golangci-lint docs.
-#
 # To use a newer version of golangci-lint, change these values:
 #   1. GOLINTERS_VERSION
 #   2. GOLINTERS_TGZ_DGST
 #
-# Release v1.49.0.1 (October 2, 2022) for github.com/fxamacker/cbor
-#   - Allow goimports/gofmt version to use different Go version
-#   - Add GOFMT_VERSION, GOTOOLS_VERSION, etc.
-#
-# Release v1.49.0 (September 18, 2022)
-#   - Bump golangci-lint to 1.49.0
-#   - Bump Go to 1.19 (latest version of 1.19.x because check-latest: true).
-#   - Put Go version in environment variable GO_VERSION.
-#   - Increase timeout to 15m for big projects enabling more linters.
-#   - Use SHA-256 to verify (instead of SHA-384) and mention checksums file.
-#   - Hash of golangci-lint-1.49.0-linux-amd64.tar.gz
-#     - SHA-384: df59267a11317d2763fb6cb454a9b3a6a6d428f4750fcbb8604fb0d289b18a1b3b6cd2bfbf2a2fe976979e97a71fcc36
-#     - SHA-256: 5badc6e9fee2003621efa07e385910d9a88c89b38f6c35aded153193c5125178
-#                This SHA-256 digest matches golangci-lint-1.49.0-checksums.txt at
+# Release v1.51.1 (February 5, 2023)
+#   - Bump golangci-lint to 1.51.1
+#   - Shuffle some comments
+#   - Hash of golangci-lint-1.50.1-linux-amd64.tar.gz
+#     - SHA-256: 17aeb26c76820c22efa0e1838b0ab93e90cfedef43fbfc9a2f33f27eb9e5e070
+#                This SHA-256 digest matches golangci-lint-1.51.1-checksums.txt at
 #                https://github.com/golangci/golangci-lint/releases
 #
 name: linters
@@ -56,70 +49,16 @@ on:
 
 env:
   GO_VERSION: 1.19
-  GOFMT_VERSION: 1.18.6        # version of Go for gofmt and for building goimports
-  GOLINTERS_VERSION: 1.49.0    # version of golangci-lint
-    
+  GOLINTERS_VERSION: 1.51.1
   GOLINTERS_ARCH: linux-amd64
-  GOLINTERS_TGZ_DGST: 5badc6e9fee2003621efa07e385910d9a88c89b38f6c35aded153193c5125178 # golangci-lint 1.49.0
+  GOLINTERS_TGZ_DGST: 17aeb26c76820c22efa0e1838b0ab93e90cfedef43fbfc9a2f33f27eb9e5e070
   GOLINTERS_TIMEOUT: 15m
-  
-  # go tools v0.1.12 (Jan 27, 2022) is commit b3b5c13b291f9653da6f31b95db100a2e26bd186
-  # https://github.com/golang/tools/releases/tag/v0.1.12
-  GOTOOLS_VERSION: b3b5c13b291f9653da6f31b95db100a2e26bd186 # v0.1.12  
-  GOIMPORTS_DGST: 82a00939e3555d5dd7ca7b53bd3e413c4d3de5c32698f05e17f1d62fdc81c07e # go 1.18.6 + go tools v0.1.12
-  GOIMPORTS_FLAGS: -l # -l (filenames) is recommended.  -d (diffs) can be useful.
-  
   OPENSSL_DGST_CMD: openssl dgst -sha256 -r
   CURL_CMD: curl --proto =https --tlsv1.2 --location --silent --show-error --fail
 
 jobs:
-  gofmt:
-    name: goimports
-    runs-on: ubuntu-latest
-    permissions:
-      contents: read
-    steps:
-      - name: Checkout source
-        uses: actions/checkout@v3
-        with:
-          fetch-depth: 1
-
-      - name: Setup Go
-        uses: actions/setup-go@v3
-        with:
-          # Use GOFMT_VERSION (not GO_VERSION) for job goimports
-          go-version: ${{ env.GOFMT_VERSION }}
-          check-latest: false  # false to prevent compiled goimports digest from changing
-          
-      # Run goimports and print file names that didn't pass
-      - name: Run goimports
-        run: |
-          go install golang.org/x/tools/cmd/goimports@${GOTOOLS_VERSION}
-          GOIMPORTS_FULLPATH=$(go env GOPATH)/bin/goimports
-          GOIMPORTS_EXPECTED_DGST="${GOIMPORTS_DGST} *${GOIMPORTS_FULLPATH}"
-          DGST_CMD="${OPENSSL_DGST_CMD} ${GOIMPORTS_FULLPATH}"
-          
-          GOIMPORTS_GOT_DGST=$(${DGST_CMD})
-          echo "${GOIMPORTS_GOT_DGST}"
-               
-          if [ "${GOIMPORTS_GOT_DGST}" != "${GOIMPORTS_EXPECTED_DGST}" ]
-          then
-            echo "Digest of goimports is not equal to expected digest."
-            echo "Expected digest: " "${GOIMPORTS_EXPECTED_DGST}"
-            echo "Got digest:      " "${GOIMPORTS_GOT_DGST}"
-            # exit 1
-          fi
-          
-          GOIMPORTS_OUTPUT=$(${GOIMPORTS_FULLPATH} ${GOIMPORTS_FLAGS} .)
-          if [[ "${GOIMPORTS_OUTPUT}" ]]; then
-            echo "${GOIMPORTS_OUTPUT}"
-            exit 1
-          fi
-
-        shell: bash
-        
-  linters:
-    name: linters
+  main:
+    name: Lint
     runs-on: ubuntu-latest
     permissions:
       contents: read
@@ -156,11 +95,6 @@ jobs:
 
           tar --no-same-owner -xzf "${GOLINTERS_TGZ}" --strip-components 1
           install golangci-lint $(go env GOPATH)/bin
-          go version
-          golangci-lint --version
-          which go
-          which gofmt
-          which golangci-lint
         shell: bash
 
       # Run required linters enabled in .golangci.yml (or default linters if yml doesn't exist)     
diff --git a/decode.go b/decode.go
index bc0e689..343da9e 100644
--- a/decode.go
+++ b/decode.go
@@ -40,17 +40,17 @@ import (
 // To unmarshal CBOR into an empty interface value, Unmarshal uses the
 // following rules:
 //
-//     CBOR booleans decode to bool.
-//     CBOR positive integers decode to uint64.
-//     CBOR negative integers decode to int64 (big.Int if value overflows).
-//     CBOR floating points decode to float64.
-//     CBOR byte strings decode to []byte.
-//     CBOR text strings decode to string.
-//     CBOR arrays decode to []interface{}.
-//     CBOR maps decode to map[interface{}]interface{}.
-//     CBOR null and undefined values decode to nil.
-//     CBOR times (tag 0 and 1) decode to time.Time.
-//     CBOR bignums (tag 2 and 3) decode to big.Int.
+//	CBOR booleans decode to bool.
+//	CBOR positive integers decode to uint64.
+//	CBOR negative integers decode to int64 (big.Int if value overflows).
+//	CBOR floating points decode to float64.
+//	CBOR byte strings decode to []byte.
+//	CBOR text strings decode to string.
+//	CBOR arrays decode to []interface{}.
+//	CBOR maps decode to map[interface{}]interface{}.
+//	CBOR null and undefined values decode to nil.
+//	CBOR times (tag 0 and 1) decode to time.Time.
+//	CBOR bignums (tag 2 and 3) decode to big.Int.
 //
 // To unmarshal a CBOR array into a slice, Unmarshal allocates a new slice
 // if the CBOR array is empty or slice capacity is less than CBOR array length.
@@ -75,9 +75,9 @@ import (
 // To unmarshal a CBOR map into a struct, Unmarshal matches CBOR map keys to the
 // keys in the following priority:
 //
-//     1. "cbor" key in struct field tag,
-//     2. "json" key in struct field tag,
-//     3. struct field name.
+//  1. "cbor" key in struct field tag,
+//  2. "json" key in struct field tag,
+//  3. struct field name.
 //
 // Unmarshal tries an exact match for field name, then a case-insensitive match.
 // Map key-value pairs without corresponding struct fields are ignored.  See
@@ -549,7 +549,16 @@ func (dm *decMode) DecOptions() DecOptions {
 // See the documentation for Unmarshal for details.
 func (dm *decMode) Unmarshal(data []byte, v interface{}) error {
 	d := decoder{data: data, dm: dm}
-	return d.value(v, false)
+
+	// check valid
+	off := d.off          // Save offset before data validation
+	err := d.valid(false) // don't allow any extra data after valid data item.
+	d.off = off           // Restore offset
+	if err != nil {
+		return err
+	}
+
+	return d.value(v)
 }
 
 // Valid checks whether the CBOR data is complete and well-formed.
@@ -570,10 +579,10 @@ type decoder struct {
 }
 
 // value decodes CBOR data item into the value pointed to by v.
-// If CBOR data item is invalid, error is returned and offset isn't changed.
-// If CBOR data item is valid but fails to be decode into v for other reasons,
+// If CBOR data item fails to be decoded into v,
 // error is returned and offset is moved to the next CBOR data item.
-func (d *decoder) value(v interface{}, allowExtraData bool) error {
+// Precondition: d.data contains at least one valid CBOR data item.
+func (d *decoder) value(v interface{}) error {
 	// v can't be nil, non-pointer, or nil pointer value.
 	if v == nil {
 		return &InvalidUnmarshalError{"cbor: Unmarshal(nil)"}
@@ -584,14 +593,6 @@ func (d *decoder) value(v interface{}, allowExtraData bool) error {
 	} else if rv.IsNil() {
 		return &InvalidUnmarshalError{"cbor: Unmarshal(nil " + rv.Type().String() + ")"}
 	}
-
-	off := d.off // Save offset before data validation
-	err := d.valid(allowExtraData)
-	d.off = off // Restore offset
-	if err != nil {
-		return err
-	}
-
 	rv = rv.Elem()
 	return d.parseToValue(rv, getTypeInfo(rv.Type()))
 }
diff --git a/stream.go b/stream.go
index 45a3422..4e8c1e8 100644
--- a/stream.go
+++ b/stream.go
@@ -16,7 +16,6 @@ type Decoder struct {
 	buf       []byte
 	off       int // next read offset in buf
 	bytesRead int
-	readError error
 }
 
 // NewDecoder returns a new decoder that reads and decodes from r using
@@ -27,67 +26,38 @@ func NewDecoder(r io.Reader) *Decoder {
 
 // Decode reads CBOR value and decodes it into the value pointed to by v.
 func (dec *Decoder) Decode(v interface{}) error {
-	if len(dec.buf) == dec.off {
-		if n, err := dec.read(); n == 0 {
-			return err
-		}
-	}
-	for {
-		dec.d.reset(dec.buf[dec.off:])
-		err := dec.d.value(v, true)
-		// Increment dec.off even if err is not nil because
-		// dec.d.off points to the next CBOR data item if current
-		// CBOR data item is valid but failed to be decoded into v.
-		// This allows next CBOR data item to be decoded in next
-		// call to this function.
-		dec.off += dec.d.off
-		dec.bytesRead += dec.d.off
-		if err == nil {
-			return nil
-		}
-		if err != io.ErrUnexpectedEOF {
-			return err
-		}
-		// Need to read more data.
-		if n, err := dec.read(); n == 0 {
-			if err == io.EOF {
-				err = io.ErrUnexpectedEOF
-			}
-			return err
-		}
+	_, err := dec.readNext()
+	if err != nil {
+		// Return validation error or read error.
+		return err
 	}
+
+	dec.d.reset(dec.buf[dec.off:])
+	err = dec.d.value(v)
+
+	// Increment dec.off even if decoding err is not nil because
+	// dec.d.off points to the next CBOR data item if current
+	// CBOR data item is valid but failed to be decoded into v.
+	// This allows next CBOR data item to be decoded in next
+	// call to this function.
+	dec.off += dec.d.off
+	dec.bytesRead += dec.d.off
+
+	return err
 }
 
 // Skip skips to the next CBOR data item (if there is any),
 // otherwise it returns error such as io.EOF, io.UnexpectedEOF, etc.
 func (dec *Decoder) Skip() error {
-	if len(dec.buf) == dec.off {
-		if n, err := dec.read(); n == 0 {
-			return err
-		}
-	}
-	for {
-		dec.d.reset(dec.buf[dec.off:])
-		err := dec.d.valid(true)
-		if err == nil {
-			// Only increment dec.off if current CBOR data item is valid.
-			// If current data item is incomplete (io.ErrUnexpectedEOF),
-			// we want to try again after reading more data.
-			dec.off += dec.d.off
-			dec.bytesRead += dec.d.off
-			return nil
-		}
-		if err != io.ErrUnexpectedEOF {
-			return err
-		}
-		// Need to read more data.
-		if n, err := dec.read(); n == 0 {
-			if err == io.EOF {
-				err = io.ErrUnexpectedEOF
-			}
-			return err
-		}
+	n, err := dec.readNext()
+	if err != nil {
+		// Return validation error or read error.
+		return err
 	}
+
+	dec.off += n
+	dec.bytesRead += n
+	return nil
 }
 
 // NumBytesRead returns the number of bytes read.
@@ -95,11 +65,66 @@ func (dec *Decoder) NumBytesRead() int {
 	return dec.bytesRead
 }
 
-func (dec *Decoder) read() (int, error) {
-	if dec.readError != nil {
-		return 0, dec.readError
+// readNext() reads next CBOR data item from Reader to buffer.
+// It returns the size of next CBOR data item.
+// It also returns validation error or read error if any.
+func (dec *Decoder) readNext() (int, error) {
+	var readErr error
+	var validErr error
+
+	for {
+		// Process any unread data in dec.buf.
+		if dec.off < len(dec.buf) {
+			dec.d.reset(dec.buf[dec.off:])
+			off := dec.off // Save offset before data validation
+			validErr = dec.d.valid(true)
+			dec.off = off // Restore offset
+
+			if validErr == nil {
+				return dec.d.off, nil
+			}
+
+			if validErr != io.ErrUnexpectedEOF {
+				return 0, validErr
+			}
+
+			// Process last read error on io.ErrUnexpectedEOF.
+			if readErr != nil {
+				if readErr == io.EOF {
+					// current CBOR data item is incomplete.
+					return 0, io.ErrUnexpectedEOF
+				}
+				return 0, readErr
+			}
+		}
+
+		// More data is needed and there was no read error.
+		var n int
+		for n == 0 {
+			n, readErr = dec.read()
+			if n == 0 && readErr != nil {
+				// No more data can be read and read error is encountered.
+				// At this point, validErr is either nil or io.ErrUnexpectedEOF.
+				if readErr == io.EOF {
+					if validErr == io.ErrUnexpectedEOF {
+						// current CBOR data item is incomplete.
+						return 0, io.ErrUnexpectedEOF
+					}
+				}
+				return 0, readErr
+			}
+		}
+
+		// At this point, dec.buf contains new data from last read (n > 0).
 	}
+}
 
+// read() reads data from Reader to buffer.
+// It returns number of bytes read and any read error encountered.
+// Postconditions:
+// - dec.buf contains previously unread data and new data.
+// - dec.off is 0.
+func (dec *Decoder) read() (int, error) {
 	// Grow buf if needed.
 	const minRead = 512
 	if cap(dec.buf)-len(dec.buf)+dec.off < minRead {
@@ -116,7 +141,6 @@ func (dec *Decoder) read() (int, error) {
 	// Read from reader and reslice buf.
 	n, err := dec.r.Read(dec.buf[len(dec.buf):cap(dec.buf)])
 	dec.buf = dec.buf[0 : len(dec.buf)+n]
-	dec.readError = err
 	return n, err
 }
 


Test patch:
diff --git a/stream_test.go b/stream_test.go
index 211d79d..bbd74ac 100644
--- a/stream_test.go
+++ b/stream_test.go
@@ -21,37 +21,51 @@ func TestDecoder(t *testing.T) {
 			buf.Write(tc.cborData)
 		}
 	}
-	decoder := NewDecoder(&buf)
-	bytesRead := 0
-	for i := 0; i < 5; i++ {
-		for _, tc := range unmarshalTests {
-			var v interface{}
-			if err := decoder.Decode(&v); err != nil {
-				t.Fatalf("Decode() returned error %v", err)
-			}
-			if tm, ok := tc.emptyInterfaceValue.(time.Time); ok {
-				if vt, ok := v.(time.Time); !ok || !tm.Equal(vt) {
-					t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
+
+	testCases := []struct {
+		name   string
+		reader io.Reader
+	}{
+		{"bytes.Buffer", &buf},
+		{"1 byte reader", newNBytesReader(buf.Bytes(), 1)},
+		{"toggled reader", newToggledReader(buf.Bytes(), 1)},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			decoder := NewDecoder(tc.reader)
+			bytesRead := 0
+			for i := 0; i < 5; i++ {
+				for _, tc := range unmarshalTests {
+					var v interface{}
+					if err := decoder.Decode(&v); err != nil {
+						t.Fatalf("Decode() returned error %v", err)
+					}
+					if tm, ok := tc.emptyInterfaceValue.(time.Time); ok {
+						if vt, ok := v.(time.Time); !ok || !tm.Equal(vt) {
+							t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
+						}
+					} else if !reflect.DeepEqual(v, tc.emptyInterfaceValue) {
+						t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
+					}
+					bytesRead += len(tc.cborData)
+					if decoder.NumBytesRead() != bytesRead {
+						t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+					}
 				}
-			} else if !reflect.DeepEqual(v, tc.emptyInterfaceValue) {
-				t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
 			}
-			bytesRead += len(tc.cborData)
-			if decoder.NumBytesRead() != bytesRead {
-				t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+			for i := 0; i < 2; i++ {
+				// no more data
+				var v interface{}
+				err := decoder.Decode(&v)
+				if v != nil {
+					t.Errorf("Decode() = %v (%T), want nil (no more data)", v, v)
+				}
+				if err != io.EOF {
+					t.Errorf("Decode() returned error %v, want io.EOF (no more data)", err)
+				}
 			}
-		}
-	}
-	for i := 0; i < 2; i++ {
-		// no more data
-		var v interface{}
-		err := decoder.Decode(&v)
-		if v != nil {
-			t.Errorf("Decode() = %v (%T), want nil (no more data)", v, v)
-		}
-		if err != io.EOF {
-			t.Errorf("Decode() returned error %v, want io.EOF (no more data)", err)
-		}
+		})
 	}
 }
 
@@ -64,50 +78,64 @@ func TestDecoderUnmarshalTypeError(t *testing.T) {
 			}
 		}
 	}
-	decoder := NewDecoder(&buf)
-	bytesRead := 0
-	for i := 0; i < 5; i++ {
-		for _, tc := range unmarshalTests {
-			for _, typ := range tc.wrongTypes {
-				v := reflect.New(typ)
-				if err := decoder.Decode(v.Interface()); err == nil {
-					t.Errorf("Decode(0x%x) didn't return an error, want UnmarshalTypeError", tc.cborData)
-				} else if _, ok := err.(*UnmarshalTypeError); !ok {
-					t.Errorf("Decode(0x%x) returned wrong error type %T, want UnmarshalTypeError", tc.cborData, err)
-				}
-				bytesRead += len(tc.cborData)
-				if decoder.NumBytesRead() != bytesRead {
-					t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
-				}
 
-				var vi interface{}
-				if err := decoder.Decode(&vi); err != nil {
-					t.Errorf("Decode() returned error %v", err)
-				}
-				if tm, ok := tc.emptyInterfaceValue.(time.Time); ok {
-					if vt, ok := vi.(time.Time); !ok || !tm.Equal(vt) {
-						t.Errorf("Decode() = %v (%T), want %v (%T)", vi, vi, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
+	testCases := []struct {
+		name   string
+		reader io.Reader
+	}{
+		{"bytes.Buffer", &buf},
+		{"1 byte reader", newNBytesReader(buf.Bytes(), 1)},
+		{"toggled reader", newToggledReader(buf.Bytes(), 1)},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			decoder := NewDecoder(tc.reader)
+			bytesRead := 0
+			for i := 0; i < 5; i++ {
+				for _, tc := range unmarshalTests {
+					for _, typ := range tc.wrongTypes {
+						v := reflect.New(typ)
+						if err := decoder.Decode(v.Interface()); err == nil {
+							t.Errorf("Decode(0x%x) didn't return an error, want UnmarshalTypeError", tc.cborData)
+						} else if _, ok := err.(*UnmarshalTypeError); !ok {
+							t.Errorf("Decode(0x%x) returned wrong error type %T, want UnmarshalTypeError", tc.cborData, err)
+						}
+						bytesRead += len(tc.cborData)
+						if decoder.NumBytesRead() != bytesRead {
+							t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+						}
+
+						var vi interface{}
+						if err := decoder.Decode(&vi); err != nil {
+							t.Errorf("Decode() returned error %v", err)
+						}
+						if tm, ok := tc.emptyInterfaceValue.(time.Time); ok {
+							if vt, ok := vi.(time.Time); !ok || !tm.Equal(vt) {
+								t.Errorf("Decode() = %v (%T), want %v (%T)", vi, vi, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
+							}
+						} else if !reflect.DeepEqual(vi, tc.emptyInterfaceValue) {
+							t.Errorf("Decode() = %v (%T), want %v (%T)", vi, vi, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
+						}
+						bytesRead += len(tc.cborData)
+						if decoder.NumBytesRead() != bytesRead {
+							t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+						}
 					}
-				} else if !reflect.DeepEqual(vi, tc.emptyInterfaceValue) {
-					t.Errorf("Decode() = %v (%T), want %v (%T)", vi, vi, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
 				}
-				bytesRead += len(tc.cborData)
-				if decoder.NumBytesRead() != bytesRead {
-					t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+			}
+			for i := 0; i < 2; i++ {
+				// no more data
+				var v interface{}
+				err := decoder.Decode(&v)
+				if v != nil {
+					t.Errorf("Decode() = %v (%T), want nil (no more data)", v, v)
+				}
+				if err != io.EOF {
+					t.Errorf("Decode() returned error %v, want io.EOF (no more data)", err)
 				}
 			}
-		}
-	}
-	for i := 0; i < 2; i++ {
-		// no more data
-		var v interface{}
-		err := decoder.Decode(&v)
-		if v != nil {
-			t.Errorf("Decode() = %v (%T), want nil (no more data)", v, v)
-		}
-		if err != io.EOF {
-			t.Errorf("Decode() returned error %v, want io.EOF (no more data)", err)
-		}
+		})
 	}
 }
 
@@ -118,36 +146,50 @@ func TestDecoderUnexpectedEOFError(t *testing.T) {
 	}
 	buf.Truncate(buf.Len() - 1)
 
-	decoder := NewDecoder(&buf)
-	bytesRead := 0
-	for i := 0; i < len(unmarshalTests)-1; i++ {
-		tc := unmarshalTests[i]
-		var v interface{}
-		if err := decoder.Decode(&v); err != nil {
-			t.Fatalf("Decode() returned error %v", err)
-		}
-		if tm, ok := tc.emptyInterfaceValue.(time.Time); ok {
-			if vt, ok := v.(time.Time); !ok || !tm.Equal(vt) {
-				t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
-			}
-		} else if !reflect.DeepEqual(v, tc.emptyInterfaceValue) {
-			t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
-		}
-		bytesRead += len(tc.cborData)
-		if decoder.NumBytesRead() != bytesRead {
-			t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
-		}
+	testCases := []struct {
+		name   string
+		reader io.Reader
+	}{
+		{"bytes.Buffer", &buf},
+		{"1 byte reader", newNBytesReader(buf.Bytes(), 1)},
+		{"toggled reader", newToggledReader(buf.Bytes(), 1)},
 	}
-	for i := 0; i < 2; i++ {
-		// truncated data
-		var v interface{}
-		err := decoder.Decode(&v)
-		if v != nil {
-			t.Errorf("Decode() = %v (%T), want nil (no more data)", v, v)
-		}
-		if err != io.ErrUnexpectedEOF {
-			t.Errorf("Decode() returned error %v, want io.UnexpectedEOF (truncated data)", err)
-		}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+
+			decoder := NewDecoder(tc.reader)
+			bytesRead := 0
+			for i := 0; i < len(unmarshalTests)-1; i++ {
+				tc := unmarshalTests[i]
+				var v interface{}
+				if err := decoder.Decode(&v); err != nil {
+					t.Fatalf("Decode() returned error %v", err)
+				}
+				if tm, ok := tc.emptyInterfaceValue.(time.Time); ok {
+					if vt, ok := v.(time.Time); !ok || !tm.Equal(vt) {
+						t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
+					}
+				} else if !reflect.DeepEqual(v, tc.emptyInterfaceValue) {
+					t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
+				}
+				bytesRead += len(tc.cborData)
+				if decoder.NumBytesRead() != bytesRead {
+					t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+				}
+			}
+			for i := 0; i < 2; i++ {
+				// truncated data
+				var v interface{}
+				err := decoder.Decode(&v)
+				if v != nil {
+					t.Errorf("Decode() = %v (%T), want nil (no more data)", v, v)
+				}
+				if err != io.ErrUnexpectedEOF {
+					t.Errorf("Decode() returned error %v, want io.UnexpectedEOF (truncated data)", err)
+				}
+			}
+		})
 	}
 }
 
@@ -160,36 +202,114 @@ func TestDecoderReadError(t *testing.T) {
 
 	readerErr := errors.New("reader error")
 
-	decoder := NewDecoder(NewErrorReader(buf.Bytes(), readerErr))
-	bytesRead := 0
-	for i := 0; i < len(unmarshalTests)-1; i++ {
-		tc := unmarshalTests[i]
-		var v interface{}
-		if err := decoder.Decode(&v); err != nil {
-			t.Fatalf("Decode() returned error %v", err)
-		}
-		if tm, ok := tc.emptyInterfaceValue.(time.Time); ok {
-			if vt, ok := v.(time.Time); !ok || !tm.Equal(vt) {
-				t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
+	testCases := []struct {
+		name   string
+		reader io.Reader
+	}{
+		{"byte reader", newNBytesReaderWithError(buf.Bytes(), 512, readerErr)},
+		{"1 byte reader", newNBytesReaderWithError(buf.Bytes(), 1, readerErr)},
+		{"toggled reader", newToggledReaderWithError(buf.Bytes(), 1, readerErr)},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			decoder := NewDecoder(tc.reader)
+			bytesRead := 0
+			for i := 0; i < len(unmarshalTests)-1; i++ {
+				tc := unmarshalTests[i]
+				var v interface{}
+				if err := decoder.Decode(&v); err != nil {
+					t.Fatalf("Decode() returned error %v", err)
+				}
+				if tm, ok := tc.emptyInterfaceValue.(time.Time); ok {
+					if vt, ok := v.(time.Time); !ok || !tm.Equal(vt) {
+						t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
+					}
+				} else if !reflect.DeepEqual(v, tc.emptyInterfaceValue) {
+					t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
+				}
+				bytesRead += len(tc.cborData)
+				if decoder.NumBytesRead() != bytesRead {
+					t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+				}
 			}
-		} else if !reflect.DeepEqual(v, tc.emptyInterfaceValue) {
-			t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, tc.emptyInterfaceValue, tc.emptyInterfaceValue)
-		}
-		bytesRead += len(tc.cborData)
-		if decoder.NumBytesRead() != bytesRead {
-			t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
-		}
+			for i := 0; i < 2; i++ {
+				// truncated data because Reader returned error
+				var v interface{}
+				err := decoder.Decode(&v)
+				if v != nil {
+					t.Errorf("Decode() = %v (%T), want nil (no more data)", v, v)
+				}
+				if err != readerErr {
+					t.Errorf("Decode() returned error %v, want reader error", err)
+				}
+			}
+		})
 	}
-	for i := 0; i < 2; i++ {
-		// truncated data because Reader returned error
-		var v interface{}
-		err := decoder.Decode(&v)
-		if v != nil {
-			t.Errorf("Decode() = %v (%T), want nil (no more data)", v, v)
-		}
-		if err != readerErr {
-			t.Errorf("Decode() returned error %v, want reader error", err)
-		}
+}
+
+func TestDecoderNoData(t *testing.T) {
+	readerErr := errors.New("reader error")
+
+	testCases := []struct {
+		name    string
+		reader  io.Reader
+		wantErr error
+	}{
+		{"byte.Buffer", new(bytes.Buffer), io.EOF},
+		{"1 byte reader", newNBytesReaderWithError(nil, 0, readerErr), readerErr},
+		{"toggled reader", newToggledReaderWithError(nil, 0, readerErr), readerErr},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			decoder := NewDecoder(tc.reader)
+			for i := 0; i < 2; i++ {
+				var v interface{}
+				err := decoder.Decode(&v)
+				if v != nil {
+					t.Errorf("Decode() = %v (%T), want nil", v, v)
+				}
+				if err != tc.wantErr {
+					t.Errorf("Decode() returned error %v, want error %v", err, tc.wantErr)
+				}
+			}
+		})
+	}
+}
+
+func TestDecoderRecoverableReadError(t *testing.T) {
+	cborData := hexDecode("83010203") // [1,2,3]
+	wantValue := []interface{}{uint64(1), uint64(2), uint64(3)}
+	recoverableReaderErr := errors.New("recoverable reader error")
+
+	decoder := NewDecoder(newRecoverableReader(cborData, 1, recoverableReaderErr))
+
+	var v interface{}
+	err := decoder.Decode(&v)
+	if err != recoverableReaderErr {
+		t.Fatalf("Decode() returned error %v, want error %v", err, recoverableReaderErr)
+	}
+
+	err = decoder.Decode(&v)
+	if err != nil {
+		t.Fatalf("Decode() returned error %v", err)
+	}
+	if !reflect.DeepEqual(v, wantValue) {
+		t.Errorf("Decode() = %v (%T), want %v (%T)", v, v, wantValue, wantValue)
+	}
+	if decoder.NumBytesRead() != len(cborData) {
+		t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), len(cborData))
+	}
+
+	// no more data
+	v = interface{}(nil)
+	err = decoder.Decode(&v)
+	if v != nil {
+		t.Errorf("Decode() = %v (%T), want nil (no more data)", v, v)
+	}
+	if err != io.EOF {
+		t.Errorf("Decode() returned error %v, want io.EOF (no more data)", err)
 	}
 }
 
@@ -219,25 +339,39 @@ func TestDecoderSkip(t *testing.T) {
 			buf.Write(tc.cborData)
 		}
 	}
-	decoder := NewDecoder(&buf)
-	bytesRead := 0
-	for i := 0; i < 5; i++ {
-		for _, tc := range unmarshalTests {
-			if err := decoder.Skip(); err != nil {
-				t.Fatalf("Skip() returned error %v", err)
+
+	testCases := []struct {
+		name   string
+		reader io.Reader
+	}{
+		{"bytes.Buffer", &buf},
+		{"1 byte reader", newNBytesReader(buf.Bytes(), 1)},
+		{"toggled reader", newToggledReader(buf.Bytes(), 1)},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			decoder := NewDecoder(tc.reader)
+			bytesRead := 0
+			for i := 0; i < 5; i++ {
+				for _, tc := range unmarshalTests {
+					if err := decoder.Skip(); err != nil {
+						t.Fatalf("Skip() returned error %v", err)
+					}
+					bytesRead += len(tc.cborData)
+					if decoder.NumBytesRead() != bytesRead {
+						t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+					}
+				}
 			}
-			bytesRead += len(tc.cborData)
-			if decoder.NumBytesRead() != bytesRead {
-				t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+			for i := 0; i < 2; i++ {
+				// no more data
+				err := decoder.Skip()
+				if err != io.EOF {
+					t.Errorf("Skip() returned error %v, want io.EOF (no more data)", err)
+				}
 			}
-		}
-	}
-	for i := 0; i < 2; i++ {
-		// no more data
-		err := decoder.Skip()
-		if err != io.EOF {
-			t.Errorf("Skip() returned error %v, want io.EOF (no more data)", err)
-		}
+		})
 	}
 }
 
@@ -248,26 +382,39 @@ func TestDecoderSkipInvalidDataError(t *testing.T) {
 	}
 	buf.WriteByte(0x3e)
 
-	decoder := NewDecoder(&buf)
-	bytesRead := 0
-	for i := 0; i < len(unmarshalTests); i++ {
-		tc := unmarshalTests[i]
-		if err := decoder.Skip(); err != nil {
-			t.Fatalf("Skip() returned error %v", err)
-		}
-		bytesRead += len(tc.cborData)
-		if decoder.NumBytesRead() != bytesRead {
-			t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
-		}
+	testCases := []struct {
+		name   string
+		reader io.Reader
+	}{
+		{"bytes.Buffer", &buf},
+		{"1 byte reader", newNBytesReader(buf.Bytes(), 1)},
+		{"toggled reader", newToggledReader(buf.Bytes(), 1)},
 	}
-	for i := 0; i < 2; i++ {
-		// last data item is invalid
-		err := decoder.Skip()
-		if err == nil {
-			t.Fatalf("Skip() didn't return error")
-		} else if !strings.Contains(err.Error(), "cbor: invalid additional information") {
-			t.Errorf("Skip() error %q, want \"cbor: invalid additional information\"", err)
-		}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			decoder := NewDecoder(tc.reader)
+			bytesRead := 0
+			for i := 0; i < len(unmarshalTests); i++ {
+				tc := unmarshalTests[i]
+				if err := decoder.Skip(); err != nil {
+					t.Fatalf("Skip() returned error %v", err)
+				}
+				bytesRead += len(tc.cborData)
+				if decoder.NumBytesRead() != bytesRead {
+					t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+				}
+			}
+			for i := 0; i < 2; i++ {
+				// last data item is invalid
+				err := decoder.Skip()
+				if err == nil {
+					t.Fatalf("Skip() didn't return error")
+				} else if !strings.Contains(err.Error(), "cbor: invalid additional information") {
+					t.Errorf("Skip() error %q, want \"cbor: invalid additional information\"", err)
+				}
+			}
+		})
 	}
 }
 
@@ -278,24 +425,37 @@ func TestDecoderSkipUnexpectedEOFError(t *testing.T) {
 	}
 	buf.Truncate(buf.Len() - 1)
 
-	decoder := NewDecoder(&buf)
-	bytesRead := 0
-	for i := 0; i < len(unmarshalTests)-1; i++ {
-		tc := unmarshalTests[i]
-		if err := decoder.Skip(); err != nil {
-			t.Fatalf("Skip() returned error %v", err)
-		}
-		bytesRead += len(tc.cborData)
-		if decoder.NumBytesRead() != bytesRead {
-			t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
-		}
+	testCases := []struct {
+		name   string
+		reader io.Reader
+	}{
+		{"bytes.Buffer", &buf},
+		{"1 byte reader", newNBytesReader(buf.Bytes(), 1)},
+		{"toggled reader", newToggledReader(buf.Bytes(), 1)},
 	}
-	for i := 0; i < 2; i++ {
-		// last data item is invalid
-		err := decoder.Skip()
-		if err != io.ErrUnexpectedEOF {
-			t.Errorf("Skip() returned error %v, want io.ErrUnexpectedEOF (truncated data)", err)
-		}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			decoder := NewDecoder(tc.reader)
+			bytesRead := 0
+			for i := 0; i < len(unmarshalTests)-1; i++ {
+				tc := unmarshalTests[i]
+				if err := decoder.Skip(); err != nil {
+					t.Fatalf("Skip() returned error %v", err)
+				}
+				bytesRead += len(tc.cborData)
+				if decoder.NumBytesRead() != bytesRead {
+					t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+				}
+			}
+			for i := 0; i < 2; i++ {
+				// last data item is invalid
+				err := decoder.Skip()
+				if err != io.ErrUnexpectedEOF {
+					t.Errorf("Skip() returned error %v, want io.ErrUnexpectedEOF (truncated data)", err)
+				}
+			}
+		})
 	}
 }
 
@@ -308,24 +468,89 @@ func TestDecoderSkipReadError(t *testing.T) {
 
 	readerErr := errors.New("reader error")
 
-	decoder := NewDecoder(NewErrorReader(buf.Bytes(), readerErr))
-	bytesRead := 0
-	for i := 0; i < len(unmarshalTests)-1; i++ {
-		tc := unmarshalTests[i]
-		if err := decoder.Skip(); err != nil {
-			t.Fatalf("Skip() returned error %v", err)
-		}
-		bytesRead += len(tc.cborData)
-		if decoder.NumBytesRead() != bytesRead {
-			t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
-		}
+	testCases := []struct {
+		name   string
+		reader io.Reader
+	}{
+		{"byte reader", newNBytesReaderWithError(buf.Bytes(), 512, readerErr)},
+		{"1 byte reader", newNBytesReaderWithError(buf.Bytes(), 1, readerErr)},
+		{"toggled reader", newToggledReaderWithError(buf.Bytes(), 1, readerErr)},
 	}
-	for i := 0; i < 2; i++ {
-		// truncated data because Reader returned error
-		err := decoder.Skip()
-		if err != readerErr {
-			t.Errorf("Skip() returned error %v, want reader error", err)
-		}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			decoder := NewDecoder(tc.reader)
+			bytesRead := 0
+			for i := 0; i < len(unmarshalTests)-1; i++ {
+				tc := unmarshalTests[i]
+				if err := decoder.Skip(); err != nil {
+					t.Fatalf("Skip() returned error %v", err)
+				}
+				bytesRead += len(tc.cborData)
+				if decoder.NumBytesRead() != bytesRead {
+					t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), bytesRead)
+				}
+			}
+			for i := 0; i < 2; i++ {
+				// truncated data because Reader returned error
+				err := decoder.Skip()
+				if err != readerErr {
+					t.Errorf("Skip() returned error %v, want reader error", err)
+				}
+			}
+		})
+	}
+}
+
+func TestDecoderSkipNoData(t *testing.T) {
+	readerErr := errors.New("reader error")
+
+	testCases := []struct {
+		name    string
+		reader  io.Reader
+		wantErr error
+	}{
+		{"byte.Buffer", new(bytes.Buffer), io.EOF},
+		{"1 byte reader", newNBytesReaderWithError(nil, 0, readerErr), readerErr},
+		{"toggled reader", newToggledReaderWithError(nil, 0, readerErr), readerErr},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			decoder := NewDecoder(tc.reader)
+			for i := 0; i < 2; i++ {
+				err := decoder.Skip()
+				if err != tc.wantErr {
+					t.Errorf("Decode() returned error %v, want error %v", err, tc.wantErr)
+				}
+			}
+		})
+	}
+}
+
+func TestDecoderSkipRecoverableReadError(t *testing.T) {
+	cborData := hexDecode("83010203") // [1,2,3]
+	recoverableReaderErr := errors.New("recoverable reader error")
+
+	decoder := NewDecoder(newRecoverableReader(cborData, 1, recoverableReaderErr))
+
+	err := decoder.Skip()
+	if err != recoverableReaderErr {
+		t.Fatalf("Skip() returned error %v, want error %v", err, recoverableReaderErr)
+	}
+
+	err = decoder.Skip()
+	if err != nil {
+		t.Fatalf("Skip() returned error %v", err)
+	}
+	if decoder.NumBytesRead() != len(cborData) {
+		t.Errorf("NumBytesRead() = %v, want %v", decoder.NumBytesRead(), len(cborData))
+	}
+
+	// no more data
+	err = decoder.Skip()
+	if err != io.EOF {
+		t.Errorf("Skip() returned error %v, want io.EOF (no more data)", err)
 	}
 }
 
@@ -663,24 +888,107 @@ func TestNilRawMessageUnmarshalCBORError(t *testing.T) {
 	}
 }
 
-type ErrorReader struct {
-	data []byte
-	off  int
-	err  error
+// nBytesReader reads at most maxBytesPerRead into b.  It also returns error at the last read.
+type nBytesReader struct {
+	data            []byte
+	maxBytesPerRead int
+	off             int
+	err             error
 }
 
-func NewErrorReader(data []byte, err error) *ErrorReader {
-	return &ErrorReader{data: data, err: err}
+func newNBytesReader(data []byte, maxBytesPerRead int) *nBytesReader {
+	return &nBytesReader{
+		data:            append([]byte{}, data...),
+		maxBytesPerRead: maxBytesPerRead,
+		err:             io.EOF,
+	}
+}
+
+func newNBytesReaderWithError(data []byte, maxBytesPerRead int, err error) *nBytesReader {
+	return &nBytesReader{
+		data:            append([]byte{}, data...),
+		maxBytesPerRead: maxBytesPerRead,
+		err:             err,
+	}
 }
 
-func (r *ErrorReader) Read(b []byte) (int, error) {
+func (r *nBytesReader) Read(b []byte) (int, error) {
 	var n int
 	if r.off < len(r.data) {
-		n = copy(b, r.data[r.off:])
+		numOfBytesToRead := len(r.data) - r.off
+		if numOfBytesToRead > r.maxBytesPerRead {
+			numOfBytesToRead = r.maxBytesPerRead
+		}
+		n = copy(b, r.data[r.off:r.off+numOfBytesToRead])
 		r.off += n
 	}
-	if n < len(b) {
+	if r.off == len(r.data) {
 		return n, r.err
 	}
 	return n, nil
 }
+
+// toggledReader returns (0, nil) for every other read to mimic non-blocking read for stream reader.
+type toggledReader struct {
+	nBytesReader
+	toggle bool
+}
+
+func newToggledReader(data []byte, maxBytesPerRead int) *toggledReader {
+	return &toggledReader{
+		nBytesReader: nBytesReader{
+			data:            append([]byte{}, data...),
+			maxBytesPerRead: maxBytesPerRead,
+			err:             io.EOF,
+		},
+		toggle: true, // first read returns (0, nil)
+	}
+}
+
+func newToggledReaderWithError(data []byte, maxBytesPerRead int, err error) *toggledReader {
+	return &toggledReader{
+		nBytesReader: nBytesReader{
+			data:            append([]byte{}, data...),
+			maxBytesPerRead: maxBytesPerRead,
+			err:             err,
+		},
+		toggle: true, // first read returns (0, nil)
+	}
+}
+
+func (r *toggledReader) Read(b []byte) (int, error) {
+	defer func() {
+		r.toggle = !r.toggle
+	}()
+	if r.toggle {
+		return 0, nil
+	}
+	return r.nBytesReader.Read(b)
+}
+
+// recoverableReader returns a recoverable error at first read operation.
+type recoverableReader struct {
+	nBytesReader
+	recoverableErr error
+	first          bool
+}
+
+func newRecoverableReader(data []byte, maxBytesPerRead int, err error) *recoverableReader {
+	return &recoverableReader{
+		nBytesReader: nBytesReader{
+			data:            append([]byte{}, data...),
+			maxBytesPerRead: maxBytesPerRead,
+			err:             io.EOF,
+		},
+		recoverableErr: err,
+		first:          true,
+	}
+}
+
+func (r *recoverableReader) Read(b []byte) (int, error) {
+	if r.first {
+		r.first = false
+		return 0, r.recoverableErr
+	}
+	return r.nBytesReader.Read(b)
+}