diff --git a/src/internal/kopia/upload.go b/src/internal/kopia/upload.go index c1e1351e5..6030ec838 100644 --- a/src/internal/kopia/upload.go +++ b/src/internal/kopia/upload.go @@ -1,13 +1,9 @@ package kopia import ( - "bytes" "context" "encoding/base64" - "encoding/binary" "errors" - "io" - "os" "runtime/trace" "strings" "sync" @@ -23,7 +19,6 @@ import ( "github.com/alcionai/corso/src/internal/common/prefixmatcher" "github.com/alcionai/corso/src/internal/common/ptr" - "github.com/alcionai/corso/src/internal/common/readers" "github.com/alcionai/corso/src/internal/data" "github.com/alcionai/corso/src/internal/diagnostics" "github.com/alcionai/corso/src/internal/m365/graph" @@ -37,101 +32,6 @@ import ( const maxInflateTraversalDepth = 500 -var versionSize = readers.VersionFormatSize - -func newBackupStreamReader(version uint32, reader io.ReadCloser) *backupStreamReader { - buf := make([]byte, versionSize) - binary.BigEndian.PutUint32(buf, version) - bufReader := io.NopCloser(bytes.NewReader(buf)) - - return &backupStreamReader{ - readers: []io.ReadCloser{bufReader, reader}, - combined: io.NopCloser(io.MultiReader(bufReader, reader)), - } -} - -// backupStreamReader is a wrapper around the io.Reader that other Corso -// components return when backing up information. It injects a version number at -// the start of the data stream. Future versions of Corso may not need this if -// they use more complex serialization logic as serialization/version injection -// will be handled by other components. -type backupStreamReader struct { - readers []io.ReadCloser - combined io.ReadCloser -} - -func (rw *backupStreamReader) Read(p []byte) (n int, err error) { - if rw.combined == nil { - return 0, os.ErrClosed - } - - return rw.combined.Read(p) -} - -func (rw *backupStreamReader) Close() error { - if rw.combined == nil { - return nil - } - - rw.combined = nil - - var errs *clues.Err - - for _, r := range rw.readers { - err := r.Close() - if err != nil { - errs = clues.Stack(clues.Wrap(err, "closing reader"), errs) - } - } - - return errs.OrNil() -} - -// restoreStreamReader is a wrapper around the io.Reader that kopia returns when -// reading data from an item. It examines and strips off the version number of -// the restored data. Future versions of Corso may not need this if they use -// more complex serialization logic as version checking/deserialization will be -// handled by other components. A reader that returns a version error is no -// longer valid and should not be used once the version error is returned. -type restoreStreamReader struct { - io.ReadCloser - expectedVersion uint32 - readVersion bool -} - -func (rw *restoreStreamReader) checkVersion() error { - versionBuf := make([]byte, versionSize) - - for newlyRead := 0; newlyRead < versionSize; { - n, err := rw.ReadCloser.Read(versionBuf[newlyRead:]) - if err != nil { - return clues.Wrap(err, "reading data format version") - } - - newlyRead += n - } - - version := binary.BigEndian.Uint32(versionBuf) - - if version != rw.expectedVersion { - return clues.New("unexpected data format").With("read_version", version) - } - - return nil -} - -func (rw *restoreStreamReader) Read(p []byte) (n int, err error) { - if !rw.readVersion { - rw.readVersion = true - - if err := rw.checkVersion(); err != nil { - return 0, err - } - } - - return rw.ReadCloser.Read(p) -} - type itemDetails struct { infoer data.ItemInfo repoPath path.Path diff --git a/src/internal/kopia/upload_test.go b/src/internal/kopia/upload_test.go index c88da8af0..168d32617 100644 --- a/src/internal/kopia/upload_test.go +++ b/src/internal/kopia/upload_test.go @@ -14,7 +14,6 @@ import ( "github.com/kopia/kopia/repo/manifest" "github.com/kopia/kopia/snapshot" "github.com/kopia/kopia/snapshot/snapshotfs" - "github.com/pkg/errors" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" @@ -220,135 +219,6 @@ func getDirEntriesForEntry( // --------------- // unit tests // --------------- -type limitedRangeReader struct { - readLen int - io.ReadCloser -} - -func (lrr *limitedRangeReader) Read(p []byte) (int, error) { - if len(p) == 0 { - // Not well specified behavior, defer to underlying reader. - return lrr.ReadCloser.Read(p) - } - - toRead := lrr.readLen - if len(p) < toRead { - toRead = len(p) - } - - return lrr.ReadCloser.Read(p[:toRead]) -} - -type VersionReadersUnitSuite struct { - tester.Suite -} - -func TestVersionReadersUnitSuite(t *testing.T) { - suite.Run(t, &VersionReadersUnitSuite{Suite: tester.NewUnitSuite(t)}) -} - -func (suite *VersionReadersUnitSuite) TestWriteAndRead() { - inputData := []byte("This is some data for the reader to test with") - table := []struct { - name string - readVersion uint32 - writeVersion uint32 - check assert.ErrorAssertionFunc - }{ - { - name: "SameVersionSucceeds", - readVersion: 42, - writeVersion: 42, - check: assert.NoError, - }, - { - name: "DifferentVersionsFail", - readVersion: 7, - writeVersion: 42, - check: assert.Error, - }, - } - - for _, test := range table { - suite.Run(test.name, func() { - t := suite.T() - - baseReader := bytes.NewReader(inputData) - - reversible := &restoreStreamReader{ - expectedVersion: test.readVersion, - ReadCloser: newBackupStreamReader( - test.writeVersion, - io.NopCloser(baseReader)), - } - - defer reversible.Close() - - allData, err := io.ReadAll(reversible) - test.check(t, err, clues.ToCore(err)) - - if err != nil { - return - } - - assert.Equal(t, inputData, allData) - }) - } -} - -func readAllInParts( - t *testing.T, - partLen int, - reader io.ReadCloser, -) ([]byte, int) { - res := []byte{} - read := 0 - tmp := make([]byte, partLen) - - for { - n, err := reader.Read(tmp) - if errors.Is(err, io.EOF) { - break - } - - require.NoError(t, err, clues.ToCore(err)) - - read += n - res = append(res, tmp[:n]...) - } - - return res, read -} - -func (suite *VersionReadersUnitSuite) TestWriteHandlesShortReads() { - t := suite.T() - inputData := []byte("This is some data for the reader to test with") - version := uint32(42) - baseReader := bytes.NewReader(inputData) - versioner := newBackupStreamReader(version, io.NopCloser(baseReader)) - expectedToWrite := len(inputData) + int(versionSize) - - // "Write" all the data. - versionedData, writtenLen := readAllInParts(t, 1, versioner) - assert.Equal(t, expectedToWrite, writtenLen) - - // Read all of the data back. - baseReader = bytes.NewReader(versionedData) - reader := &restoreStreamReader{ - expectedVersion: version, - // Be adversarial and only allow reads of length 1 from the byte reader. - ReadCloser: &limitedRangeReader{ - readLen: 1, - ReadCloser: io.NopCloser(baseReader), - }, - } - readData, readLen := readAllInParts(t, 1, reader) - // This reports the bytes read and returned to the user, excluding the version - // that is stripped off at the start. - assert.Equal(t, len(inputData), readLen) - assert.Equal(t, inputData, readData) -} - type CorsoProgressUnitSuite struct { tester.Suite targetFilePath path.Path