diff --git a/snapshot/manifest.go b/snapshot/manifest.go index ac9b59e10..58962e0a8 100644 --- a/snapshot/manifest.go +++ b/snapshot/manifest.go @@ -82,6 +82,15 @@ type DirEntry struct { DirSummary *fs.DirectorySummary `json:"summ,omitempty"` } +// DirManifest represents serialized contents of a directory. +// The entries are sorted lexicographically and summary only refers to properties of +// entries, so directory with the same contents always serializes to exactly the same JSON. +type DirManifest struct { + StreamType string `json:"stream"` // legacy + Entries []*DirEntry `json:"entries"` + Summary *fs.DirectorySummary `json:"summary"` +} + // RootObjectID returns the ID of a root object. func (m *Manifest) RootObjectID() object.ID { if m.RootEntry != nil { diff --git a/snapshot/snapshotfs/dir_reader.go b/snapshot/snapshotfs/dir_reader.go index 71204647b..7bb53e568 100644 --- a/snapshot/snapshotfs/dir_reader.go +++ b/snapshot/snapshotfs/dir_reader.go @@ -1,37 +1,28 @@ package snapshotfs import ( - "bufio" + "encoding/json" "io" + "github.com/pkg/errors" + "github.com/kopia/kopia/fs" - "github.com/kopia/kopia/internal/jsonstream" "github.com/kopia/kopia/snapshot" ) var directoryStreamType = "kopia:directory" -// readDirEntries reads all the Entry from the specified reader. +// readDirEntries reads all directory entries from the specified reader. func readDirEntries(r io.Reader) ([]*snapshot.DirEntry, *fs.DirectorySummary, error) { - var summ fs.DirectorySummary - psr, err := jsonstream.NewReader(bufio.NewReader(r), directoryStreamType, &summ) - if err != nil { - return nil, nil, err - } - var entries []*snapshot.DirEntry - for { - e := &snapshot.DirEntry{} - err := psr.Read(e) - if err == io.EOF { - break - } + var dir snapshot.DirManifest - if err != nil { - return nil, nil, err - } - - entries = append(entries, e) + if err := json.NewDecoder(r).Decode(&dir); err != nil { + return nil, nil, errors.Wrap(err, "unable to parse directory object") } - return entries, &summ, nil + if dir.StreamType != directoryStreamType { + return nil, nil, errors.Errorf("invalid directory stream type") + } + + return dir.Entries, dir.Summary, nil } diff --git a/snapshot/snapshotfs/dir_writer.go b/snapshot/snapshotfs/dir_writer.go deleted file mode 100644 index 83edd4db4..000000000 --- a/snapshot/snapshotfs/dir_writer.go +++ /dev/null @@ -1,37 +0,0 @@ -package snapshotfs - -import ( - "io" - - "github.com/kopia/kopia/fs" - "github.com/kopia/kopia/internal/jsonstream" - "github.com/kopia/kopia/snapshot" -) - -// dirWriter writes a stream of directory entries. -type dirWriter struct { - w *jsonstream.Writer -} - -// WriteEntry writes the specified entry to the output. -func (dw *dirWriter) WriteEntry(e *snapshot.DirEntry) error { - if err := e.ObjectID.Validate(); err != nil { - panic("invalid object ID: " + err.Error()) - } - - return dw.w.Write(e) -} - -// Finalize writes the trailing data to the output stream. -func (dw *dirWriter) Finalize(summ *fs.DirectorySummary) error { - return dw.w.FinalizeWithSummary(summ) -} - -// newDirWriter returns new directoryWriter for with the specified output. -func newDirWriter(w io.Writer) *dirWriter { - dw := &dirWriter{ - w: jsonstream.NewWriter(w, directoryStreamType), - } - - return dw -} diff --git a/snapshot/snapshotfs/upload.go b/snapshot/snapshotfs/upload.go index 85c6026e3..047098f1a 100644 --- a/snapshot/snapshotfs/upload.go +++ b/snapshot/snapshotfs/upload.go @@ -3,6 +3,7 @@ package snapshotfs import ( "bytes" "context" + "encoding/json" "hash/fnv" "io" "os" @@ -285,7 +286,7 @@ type entryResult struct { de *snapshot.DirEntry } -func (u *Uploader) processSubdirectories(ctx context.Context, relativePath string, entries fs.Entries, previousEntries []fs.Entries, dw *dirWriter, summ *fs.DirectorySummary) error { +func (u *Uploader) processSubdirectories(ctx context.Context, relativePath string, entries fs.Entries, previousEntries []fs.Entries, dirManifest *snapshot.DirManifest, summ *fs.DirectorySummary) error { return u.foreachEntryUnlessCancelled(relativePath, entries, func(entry fs.Entry, entryRelativePath string) error { dir, ok := entry.(fs.Directory) if !ok { @@ -324,10 +325,7 @@ func (u *Uploader) processSubdirectories(ctx context.Context, relativePath strin } de.DirSummary = &subdirsumm - if err := dw.WriteEntry(de); err != nil { - return errors.Wrap(err, "unable to write dir entry") - } - + dirManifest.Entries = append(dirManifest.Entries, de) return nil }) } @@ -510,7 +508,7 @@ func (u *Uploader) launchWorkItems(workItems []*uploadWorkItem, wg *sync.WaitGro } } -func (u *Uploader) processUploadWorkItems(workItems []*uploadWorkItem, dw *dirWriter) error { +func (u *Uploader) processUploadWorkItems(workItems []*uploadWorkItem, dirManifest *snapshot.DirManifest) error { var wg sync.WaitGroup u.launchWorkItems(workItems, &wg) @@ -531,9 +529,7 @@ func (u *Uploader) processUploadWorkItems(workItems []*uploadWorkItem, dw *dirWr return errors.Errorf("unable to process %q: %s", it.entryRelativePath, result.err) } - if err := dw.WriteEntry(result.de); err != nil { - return errors.Wrap(err, "unable to write directory entry") - } + dirManifest.Entries = append(dirManifest.Entries, result.de) } // wait for workers, this is technically not needed, but let's make sure we don't leak goroutines @@ -609,15 +605,11 @@ func uploadDirInternal( summ.MaxModTime = directory.ModTime() } - writer := u.repo.Objects.NewWriter(ctx, object.WriterOptions{ - Description: "DIR:" + dirRelativePath, - Prefix: "k", - }) + dirManifest := &snapshot.DirManifest{ + StreamType: directoryStreamType, + } - dw := newDirWriter(writer) - defer writer.Close() //nolint:errcheck - - if err := u.processSubdirectories(ctx, dirRelativePath, entries, prevEntries, dw, &summ); err != nil && err != errCancelled { + if err := u.processSubdirectories(ctx, dirRelativePath, entries, prevEntries, dirManifest, &summ); err != nil && err != errCancelled { return "", fs.DirectorySummary{}, err } u.prepareProgress(dirRelativePath, entries) @@ -626,11 +618,18 @@ func uploadDirInternal( if workItemErr != nil && workItemErr != errCancelled { return "", fs.DirectorySummary{}, workItemErr } - if err := u.processUploadWorkItems(workItems, dw); err != nil && err != errCancelled { + if err := u.processUploadWorkItems(workItems, dirManifest); err != nil && err != errCancelled { return "", fs.DirectorySummary{}, err } - if err := dw.Finalize(&summ); err != nil { - return "", fs.DirectorySummary{}, errors.Wrap(err, "unable to finalize directory") + dirManifest.Summary = &summ + + writer := u.repo.Objects.NewWriter(ctx, object.WriterOptions{ + Description: "DIR:" + dirRelativePath, + Prefix: "k", + }) + + if err := json.NewEncoder(writer).Encode(&dirManifest); err != nil { + return "", fs.DirectorySummary{}, errors.Wrap(err, "unable to encode directory JSON") } oid, err := writer.Result()