Add children bucket to back reference snapshots

Adds back links from parent to children in order to prevent
deletion of a referenced snapshot in a namespace.
Avoid removing snapshot during metadata delete to
prevent shared namespaces from being mistakenly deleted.

Signed-off-by: Derek McGowan <derek@mcgstyle.net>

metadata/buckets.go

@@ -45,6 +45,7 @@ var (
 	bucketKeyRuntime     = []byte("runtime")
 	bucketKeyName        = []byte("name")
 	bucketKeyParent      = []byte("parent")
+	bucketKeyChildren    = []byte("children")
 	bucketKeyOptions     = []byte("options")
 	bucketKeySpec        = []byte("spec")
 	bucketKeySnapshotKey = []byte("snapshotKey")

metadata/snapshot.go

@@ -283,10 +283,18 @@ func (s *snapshotter) createSnapshot(ctx context.Context, key, parent string, re
 		if parent != "" {
 			pbkt := bkt.Bucket([]byte(parent))
 			if pbkt == nil {
-				return errors.Wrapf(errdefs.ErrNotFound, "snapshot %v does not exist", parent)
+				return errors.Wrapf(errdefs.ErrNotFound, "parent snapshot %v does not exist", parent)
 			}
 			bparent = string(pbkt.Get(bucketKeyName))
 
+			cbkt, err := pbkt.CreateBucketIfNotExists(bucketKeyChildren)
+			if err != nil {
+				return err
+			}
+			if err := cbkt.Put([]byte(key), nil); err != nil {
+				return err
+			}
+
 			if err := bbkt.Put(bucketKeyParent, []byte(parent)); err != nil {
 				return err
 			}
@@ -360,7 +368,6 @@ func (s *snapshotter) Commit(ctx context.Context, name, key string, opts ...snap
 		}
 
 		bkey := string(obkt.Get(bucketKeyName))
-		parent := string(obkt.Get(bucketKeyParent))
 
 		sid, err := bkt.NextSequence()
 		if err != nil {
@@ -372,9 +379,29 @@ func (s *snapshotter) Commit(ctx context.Context, name, key string, opts ...snap
 		if err := bbkt.Put(bucketKeyName, []byte(nameKey)); err != nil {
 			return err
 		}
-		if err := bbkt.Put(bucketKeyParent, []byte(parent)); err != nil {
+
+		parent := obkt.Get(bucketKeyParent)
+		if len(parent) > 0 {
+			pbkt := bkt.Bucket(parent)
+			if pbkt == nil {
+				return errors.Wrapf(errdefs.ErrNotFound, "parent snapshot %v does not exist", string(parent))
+			}
+
+			cbkt, err := pbkt.CreateBucketIfNotExists(bucketKeyChildren)
+			if err != nil {
 				return err
 			}
+			if err := cbkt.Delete([]byte(key)); err != nil {
+				return err
+			}
+			if err := cbkt.Put([]byte(name), nil); err != nil {
+				return err
+			}
+
+			if err := bbkt.Put(bucketKeyParent, parent); err != nil {
+				return err
+			}
+		}
 		ts := time.Now().UTC()
 		if err := boltutil.WriteTimestamps(bbkt, ts, ts); err != nil {
 			return err
@@ -400,23 +427,41 @@ func (s *snapshotter) Remove(ctx context.Context, key string) error {
 	}
 
 	return update(ctx, s.db, func(tx *bolt.Tx) error {
-		var bkey string
+		var sbkt *bolt.Bucket
 		bkt := getSnapshotterBucket(tx, ns, s.name)
 		if bkt != nil {
-			sbkt := bkt.Bucket([]byte(key))
-			if sbkt != nil {
-				bkey = string(sbkt.Get(bucketKeyName))
+			sbkt = bkt.Bucket([]byte(key))
 		}
-		}
-		if bkey == "" {
+		if sbkt == nil {
 			return errors.Wrapf(errdefs.ErrNotFound, "snapshot %v does not exist", key)
 		}
 
+		cbkt := sbkt.Bucket(bucketKeyChildren)
+		if cbkt != nil {
+			if child, _ := cbkt.Cursor().First(); child != nil {
+				return errors.Wrap(errdefs.ErrFailedPrecondition, "cannot remove snapshot with child")
+			}
+		}
+
+		parent := sbkt.Get(bucketKeyParent)
+		if len(parent) > 0 {
+			pbkt := bkt.Bucket(parent)
+			if pbkt == nil {
+				return errors.Wrapf(errdefs.ErrNotFound, "parent snapshot %v does not exist", string(parent))
+			}
+			cbkt := pbkt.Bucket(bucketKeyChildren)
+			if cbkt != nil {
+				if err := cbkt.Delete([]byte(key)); err != nil {
+					return errors.Wrap(err, "failed to remove child link")
+				}
+			}
+		}
+
 		if err := bkt.DeleteBucket([]byte(key)); err != nil {
 			return err
 		}
 
-		return s.Snapshotter.Remove(ctx, bkey)
+		return nil
 	})
 }
 

snapshot/storage/bolt.go

@@ -305,7 +305,7 @@ func Remove(ctx context.Context, key string) (string, snapshot.Kind, error) {
 		if pbkt != nil {
 			k, _ := pbkt.Cursor().Seek(parentPrefixKey(id))
 			if getParentPrefix(k) == id {
-				return errors.Errorf("cannot remove snapshot with child")
+				return errors.Wrap(errdefs.ErrFailedPrecondition, "cannot remove snapshot with child")
 			}
 
 			if si.Parent != "" {