From 4833835e755bae956aa3eeceb5f8438ee3716b7f Mon Sep 17 00:00:00 2001
From: Albert Chu <chu11@llnl.gov>
Date: Wed, 17 Apr 2019 14:32:40 -0700
Subject: [PATCH] cmd/flux: Fix unlink corner cases with links

Fix corner cases in kvs unlink in regards to links.

If the link pointed to an invalid target, namespace, or
had an infinite cycle, the unlink_safety_check() function
would fail and not allow the user to remove the link.

Instead, unlink should always allow links to be removed.
Re-work unlink_saftety_check() to ensure links are always
removable.

Add regression tests.

Fixes #2129
---
 src/cmd/flux-kvs.c | 47 +++++++++++++++++++++++++++++++++++++---------
 t/t1000-kvs.t      | 19 +++++++++++++++++++
 2 files changed, 57 insertions(+), 9 deletions(-)

diff --git a/src/cmd/flux-kvs.c b/src/cmd/flux-kvs.c
index 1170c231569c..e1e236e7d5d3 100644
--- a/src/cmd/flux-kvs.c
+++ b/src/cmd/flux-kvs.c
@@ -930,34 +930,63 @@ int cmd_put (optparse_t *p, int argc, char **argv)
     return (0);
 }
 
+static int unlink_check_dir_empty (flux_t *h, const char *key, const char *ns)
+{
+    flux_future_t *f;
+    const flux_kvsdir_t *dir = NULL;
+    int rc = -1;
+
+    if (!(f = flux_kvs_lookup (h, ns, FLUX_KVS_READDIR, key)))
+        goto done;
+    if (flux_kvs_lookup_get_dir (f, &dir) < 0)
+        goto done;
+    if (flux_kvsdir_get_size (dir) > 0) {
+        errno = ENOTEMPTY;
+        goto done;
+    }
+    rc = 0;
+done:
+    flux_future_destroy (f);
+    return rc;
+}
+
 /* Some checks prior to unlinking key:
  * - fail if key does not exist (ENOENT) and fopt not specified or
  *   other fatal lookup error
  * - fail if key is a non-empty directory (ENOTEMPTY) and -R was not specified
+ * - if key is a link, a val, or a valref, we can always remove it.
  */
 static int unlink_safety_check (flux_t *h, const char *key, const char *ns,
                                 bool Ropt, bool fopt, bool *unlinkable)
 {
     flux_future_t *f;
-    const flux_kvsdir_t *dir = NULL;
+    const char *json_str;
+    json_t *treeobj = NULL;
     int rc = -1;
 
-    *unlinkable = false;
-
-    if (!(f = flux_kvs_lookup (h, ns, FLUX_KVS_READDIR, key)))
+    if (!(f = flux_kvs_lookup (h, ns, FLUX_KVS_TREEOBJ, key)))
         goto done;
-    if (flux_kvs_lookup_get_dir (f, &dir) < 0) {
+    if (flux_kvs_lookup_get_treeobj (f, &json_str) < 0) {
         if (errno == ENOENT && fopt)
             goto out;
-        if (errno != ENOTDIR)
-            goto done;
+        goto done;
     }
-    else if (!Ropt) {
-        if (flux_kvsdir_get_size (dir) > 0) {
+    if (!(treeobj = treeobj_decode (json_str)))
+        log_err_exit ("%s: metadata decode error", key);
+    if (treeobj_is_dir (treeobj)) {
+        if (!Ropt && treeobj_get_count (treeobj) > 0) {
             errno = ENOTEMPTY;
             goto done;
         }
     }
+    else if (treeobj_is_dirref (treeobj)) {
+        if (!Ropt) {
+            /* have to do another lookup to resolve this dirref */
+            if (unlink_check_dir_empty (h, key, ns) < 0)
+                goto done;
+        }
+    }
+    /* else treeobj is symlink, val, or valref */
     *unlinkable = true;
 out:
     rc = 0;
diff --git a/t/t1000-kvs.t b/t/t1000-kvs.t
index c1f6c8c4d7cd..b751a040a2f5 100755
--- a/t/t1000-kvs.t
+++ b/t/t1000-kvs.t
@@ -389,6 +389,25 @@ test_expect_success 'kvs: empty directory remains after key removed' '
         flux kvs unlink $DIR.a &&
 	test_empty_directory $DIR
 '
+test_expect_success 'kvs: unlink works on link that points to invalid target' '
+	flux kvs unlink -Rf $DIR &&
+        flux kvs link invalid $DIR.link &&
+        flux kvs unlink $DIR.link
+'
+test_expect_success 'kvs: unlink works on link that points to invalid namespace' '
+	flux kvs unlink -Rf $DIR &&
+        flux kvs put $DIR.foo=1 &&
+        flux kvs link --target-namespace=invalid $DIR.foo $DIR.link &&
+        flux kvs unlink $DIR.link
+'
+test_expect_success 'kvs: unlink works on link with infinite cycle' '
+	flux kvs unlink -Rf $DIR &&
+        flux kvs link $DIR.a $DIR.b &&
+        flux kvs link $DIR.b $DIR.a &&
+        flux kvs unlink $DIR.a &&
+        flux kvs unlink $DIR.b &&
+	test_empty_directory $DIR
+'
 
 #
 # empty string corner case tests