From 447a0a94527ce3dcee5ea49fb43fc85123d8dcc1 Mon Sep 17 00:00:00 2001
From: Derek McGowan <derek@mcgstyle.net>
Date: Mon, 2 Oct 2017 16:44:57 -0700
Subject: [PATCH] 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      |  1 +
 metadata/snapshot.go     | 67 +++++++++++++++++++++++++++++++++-------
 snapshot/storage/bolt.go |  2 +-
 3 files changed, 58 insertions(+), 12 deletions(-)

diff --git a/metadata/buckets.go b/metadata/buckets.go
index 6097cbf8f..b6e3500e7 100644
--- a/metadata/buckets.go
+++ b/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")
diff --git a/metadata/snapshot.go b/metadata/snapshot.go
index e84f2d8ec..d83c4676d 100644
--- a/metadata/snapshot.go
+++ b/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,8 +379,28 @@ 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 {
-			return err
+
+		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 {
@@ -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 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")
 			}
 		}
-		if bkey == "" {
-			return errors.Wrapf(errdefs.ErrNotFound, "snapshot %v does not exist", key)
+
+		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
 	})
 }
 
diff --git a/snapshot/storage/bolt.go b/snapshot/storage/bolt.go
index 4f0c677ce..3ca3b879c 100644
--- a/snapshot/storage/bolt.go
+++ b/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 != "" {