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

@@ -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;

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