From 5eaf95052dc6d9edc9d4e9637a5201f7162f46ef Mon Sep 17 00:00:00 2001 From: ashmrtn <3891298+ashmrtn@users.noreply.github.com> Date: Fri, 29 Sep 2023 14:26:15 -0700 Subject: [PATCH] Remove old serialization format code (#4410) Remove the now unused serialization format code that lived in the kopia package --- #### Does this PR need a docs update or release note? - [ ] :white_check_mark: Yes, it's included - [ ] :clock1: Yes, but in a later PR - [x] :no_entry: No #### Type of change - [ ] :sunflower: Feature - [ ] :bug: Bugfix - [ ] :world_map: Documentation - [ ] :robot: Supportability/Tests - [ ] :computer: CI/Deployment - [x] :broom: Tech Debt/Cleanup #### Issue(s) * closes #4328 #### Test Plan - [x] :muscle: Manual - [ ] :zap: Unit test - [ ] :green_heart: E2E --- src/internal/kopia/upload.go | 100 ----------------------- src/internal/kopia/upload_test.go | 130 ------------------------------ 2 files changed, 230 deletions(-) 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