Always check exists on commit error

dmcgowan · dmcgowan · commit 6875d3df3a17 · 2018-09-14T01:26:03.000-07:00
Signed-off-by: Derek McGowan &lt;derek@mcgstyle.net&gt;
diff --git a/content/local/writer.go b/content/local/writer.go
@@ -132,11 +132,11 @@ func (w *writer) Commit(ctx context.Context, size int64, expected digest.Digest,
 	// clean up!!
 	defer os.RemoveAll(w.path)
 
+	if _, err := os.Stat(target); err == nil {
+		// collision with the target file!
+		return errors.Wrapf(errdefs.ErrAlreadyExists, "content %v", dgst)
+	}
 	if err := os.Rename(ingest, target); err != nil {
-		if os.IsExist(err) {
-			// collision with the target file!
-			return errors.Wrapf(errdefs.ErrAlreadyExists, "content %v", dgst)
-		}
 		return err
 	}
 	commitTime := time.Now()
diff --git a/diff/walking/differ.go b/diff/walking/differ.go
@@ -139,7 +139,9 @@ func (s *walkingDiff) Compare(ctx context.Context, lower, upper []mount.Mount, o
 
 			dgst := cw.Digest()
 			if err := cw.Commit(ctx, 0, dgst, commitopts...); err != nil {
-				return errors.Wrap(err, "failed to commit")
+				if !errdefs.IsAlreadyExists(err) {
+					return errors.Wrap(err, "failed to commit")
+				}
 			}
 
 			info, err := s.store.Info(ctx, dgst)
diff --git a/metadata/buckets.go b/metadata/buckets.go
@@ -164,11 +164,11 @@ func getSnapshotterBucket(tx *bolt.Tx, namespace, snapshotter string) *bolt.Buck
 }
 
 func createBlobBucket(tx *bolt.Tx, namespace string, dgst digest.Digest) (*bolt.Bucket, error) {
-	bkt, err := createBucketIfNotExists(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectContent, bucketKeyObjectBlob, []byte(dgst.String()))
+	bkt, err := createBucketIfNotExists(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectContent, bucketKeyObjectBlob)
 	if err != nil {
 		return nil, err
 	}
-	return bkt, nil
+	return bkt.CreateBucket([]byte(dgst.String()))
 }
 
 func getBlobsBucket(tx *bolt.Tx, namespace string) *bolt.Bucket {
diff --git a/metadata/content.go b/metadata/content.go
@@ -592,9 +592,6 @@ func (nw *namespacedWriter) commit(ctx context.Context, tx *bolt.Tx, size int64,
 		}
 		size = nw.desc.Size
 		actual = nw.desc.Digest
-		if getBlobBucket(tx, nw.namespace, actual) != nil {
-			return "", errors.Wrapf(errdefs.ErrAlreadyExists, "content %v", actual)
-		}
 	} else {
 		status, err := nw.w.Status()
 		if err != nil {
@@ -606,18 +603,16 @@ func (nw *namespacedWriter) commit(ctx context.Context, tx *bolt.Tx, size int64,
 		size = status.Offset
 		actual = nw.w.Digest()
 
-		if err := nw.w.Commit(ctx, size, expected); err != nil {
-			if !errdefs.IsAlreadyExists(err) {
-				return "", err
-			}
-			if getBlobBucket(tx, nw.namespace, actual) != nil {
-				return "", errors.Wrapf(errdefs.ErrAlreadyExists, "content %v", actual)
-			}
+		if err := nw.w.Commit(ctx, size, expected); err != nil && !errdefs.IsAlreadyExists(err) {
+			return "", err
 		}
 	}
 
 	bkt, err := createBlobBucket(tx, nw.namespace, actual)
 	if err != nil {
+		if err == bolt.ErrBucketExists {
+			return "", errors.Wrapf(errdefs.ErrAlreadyExists, "content %v", actual)
+		}
 		return "", err
 	}
 
diff --git a/services/content/service.go b/services/content/service.go
@@ -213,7 +213,7 @@ func (s *service) Read(req *api.ReadContentRequest, session api.Content_ReadServ
 	_, err = io.CopyBuffer(
 		&readResponseWriter{session: session},
 		io.NewSectionReader(ra, offset, size), *p)
-	return err
+	return errdefs.ToGRPC(err)
 }
 
 // readResponseWriter is a writer that places the output into ReadContentRequest messages.
@@ -420,7 +420,7 @@ func (s *service) Write(session api.Content_WriteServer) (err error) {
 				// maintain the offset as append only, we just issue the write.
 				n, err := wr.Write(req.Data)
 				if err != nil {
-					return err
+					return errdefs.ToGRPC(err)
 				}
 
 				if n != len(req.Data) {
@@ -438,7 +438,7 @@ func (s *service) Write(session api.Content_WriteServer) (err error) {
 					opts = append(opts, content.WithLabels(req.Labels))
 				}
 				if err := wr.Commit(ctx, total, expected, opts...); err != nil {
-					return err
+					return errdefs.ToGRPC(err)
 				}
 			}
 
diff --git a/task.go b/task.go
@@ -607,8 +607,11 @@ func writeContent(ctx context.Context, store content.Ingester, mediaType, ref st
 	if err != nil {
 		return d, err
 	}
+
 	if err := writer.Commit(ctx, size, "", opts...); err != nil {
-		return d, err
+		if !errdefs.IsAlreadyExists(err) {
+			return d, err
+		}
 	}
 	return v1.Descriptor{
 		MediaType: mediaType,

Original file line number	Diff line number	Diff line change
`@@ -139,7 +139,9 @@ func (s *walkingDiff) Compare(ctx context.Context, lower, upper []mount.Mount, o`
`139`	`139`
`140`	`140`	`dgst := cw.Digest()`
`141`	`141`	`if err := cw.Commit(ctx, 0, dgst, commitopts...); err != nil {`
`142`		`- return errors.Wrap(err, "failed to commit")`
	`142`	`+ if !errdefs.IsAlreadyExists(err) {`
	`143`	`+ return errors.Wrap(err, "failed to commit")`
	`144`	`+ }`
`143`	`145`	`}`
`144`	`146`
`145`	`147`	`info, err := s.store.Info(ctx, dgst)`
Original file line number	Diff line number	Diff line change
`@@ -164,11 +164,11 @@ func getSnapshotterBucket(tx bolt.Tx, namespace, snapshotter string) bolt.Buck`
`164`	`164`	`}`
`165`	`165`
`166`	`166`	`func createBlobBucket(tx bolt.Tx, namespace string, dgst digest.Digest) (bolt.Bucket, error) {`
`167`		`- bkt, err := createBucketIfNotExists(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectContent, bucketKeyObjectBlob, []byte(dgst.String()))`
	`167`	`+ bkt, err := createBucketIfNotExists(tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectContent, bucketKeyObjectBlob)`
`168`	`168`	`if err != nil {`
`169`	`169`	`return nil, err`
`170`	`170`	`}`
`171`		`- return bkt, nil`
	`171`	`+ return bkt.CreateBucket([]byte(dgst.String()))`
`172`	`172`	`}`
`173`	`173`
`174`	`174`	`func getBlobsBucket(tx bolt.Tx, namespace string) bolt.Bucket {`
Original file line number	Diff line number	Diff line change
`@@ -213,7 +213,7 @@ func (s service) Read(req api.ReadContentRequest, session api.Content_ReadServ`
`213`	`213`	`_, err = io.CopyBuffer(`
`214`	`214`	`&readResponseWriter{session: session},`
`215`	`215`	`io.NewSectionReader(ra, offset, size), *p)`
`216`		`- return err`
	`216`	`+ return errdefs.ToGRPC(err)`
`217`	`217`	`}`
`218`	`218`
`219`	`219`	`// readResponseWriter is a writer that places the output into ReadContentRequest messages.`
`@@ -420,7 +420,7 @@ func (s *service) Write(session api.Content_WriteServer) (err error) {`
`420`	`420`	`// maintain the offset as append only, we just issue the write.`
`421`	`421`	`n, err := wr.Write(req.Data)`
`422`	`422`	`if err != nil {`
`423`		`- return err`
	`423`	`+ return errdefs.ToGRPC(err)`
`424`	`424`	`}`
`425`	`425`
`426`	`426`	`if n != len(req.Data) {`
`@@ -438,7 +438,7 @@ func (s *service) Write(session api.Content_WriteServer) (err error) {`
`438`	`438`	`opts = append(opts, content.WithLabels(req.Labels))`
`439`	`439`	`}`
`440`	`440`	`if err := wr.Commit(ctx, total, expected, opts...); err != nil {`
`441`		`- return err`
	`441`	`+ return errdefs.ToGRPC(err)`
`442`	`442`	`}`
`443`	`443`	`}`
`444`	`444`
Original file line number	Diff line number	Diff line change
`@@ -607,8 +607,11 @@ func writeContent(ctx context.Context, store content.Ingester, mediaType, ref st`
`607`	`607`	`if err != nil {`
`608`	`608`	`return d, err`
`609`	`609`	`}`
	`610`	`+`
`610`	`611`	`if err := writer.Commit(ctx, size, "", opts...); err != nil {`
`611`		`- return d, err`
	`612`	`+ if !errdefs.IsAlreadyExists(err) {`
	`613`	`+ return d, err`
	`614`	`+ }`
`612`	`615`	`}`
`613`	`616`	`return v1.Descriptor{`
`614`	`617`	`MediaType: mediaType,`