modules/kvs-watch: Add watcher refcount

Add a ref count to watcher objects, as a few corner cases
could result in a use after free.  In particular, a watcher could
be used in a lookup_continuation(), but already be destroyed
from the namespace.

src/modules/kvs-watch/kvs-watch.c

@@ -55,6 +55,7 @@ struct watcher {
     struct namespace *ns;       // back pointer for removal
     void *prev;                 // previous watch for KVS_WATCH_FULL
     int prev_len;               // previous watch len for KVS_WATCH_FULL
+    int refcount;               // refcount on watcher
 };
 
 /* Current KVS root.
@@ -94,7 +95,6 @@ struct watch_ctx {
     zhash_t *namespaces;        // hash of monitored namespaces
 };
 
-
 static void watcher_destroy (struct watcher *w)
 {
     if (w) {
@@ -113,6 +113,21 @@ static void watcher_destroy (struct watcher *w)
     }
 }
 
+static void watcher_incref (struct watcher *w)
+{
+    if (w)
+        w->refcount++;
+}
+
+static void watcher_decref (struct watcher *w)
+{
+    if (w) {
+        w->refcount--;
+        if (w->refcount == 0)
+            watcher_destroy (w);
+    }
+}
+
 static struct watcher *watcher_create (const flux_msg_t *msg, const char *key,
                                        int flags)
 {
@@ -131,6 +146,7 @@ static struct watcher *watcher_create (const flux_msg_t *msg, const char *key,
         goto error_nomem;
     w->flags = flags;
     w->rootseq = -1;
+    w->refcount = 1;
     return w;
 error_nomem:
     errno = ENOMEM;
@@ -226,6 +242,13 @@ static struct namespace *namespace_create (struct watch_ctx *ctx,
     return NULL;
 }
 
+static void namespace_watchers_remove (struct namespace *ns,
+                                       struct watcher *w)
+{
+    if (w->refcount == 1)
+        zlist_remove (ns->watchers, w);
+}
+
 /* Verify that (userid, rolemask) credential is authorized to access 'ns'.
  * The instance owner or namespace owner are permitted (return 0).
  * All others are denied (return -1, errno == EPERM).
@@ -351,6 +374,7 @@ static int handle_lookup_response (flux_future_t *f, struct watcher *w)
     if (!w->mute) {
         if (flux_respond_error (h, w->request, errno, NULL) < 0)
             flux_log_error (h, "%s: flux_respond_error", __FUNCTION__);
+        w->mute = true;
     }
     return -1;
 }
@@ -371,10 +395,11 @@ static void lookup_continuation (flux_future_t *f, void *arg)
         if (rc < 0)
             goto destroy_watcher;
     }
+    watcher_decref (w);
     return;
 destroy_watcher:
-    zlist_remove (ns->watchers, w);
-    watcher_destroy (w);
+    namespace_watchers_remove (ns, w);
+    watcher_decref (w);
     if (zlist_size (ns->watchers) == 0)
         zhash_delete (ns->ctx->namespaces, ns->name);
 }
@@ -523,17 +548,19 @@ static void watcher_respond (struct namespace *ns, struct watcher *w)
             flux_future_destroy (f);
             goto error_respond;
         }
+        watcher_incref (w);
         w->rootseq = ns->commit->rootseq;
     }
     return;
 error_respond:
     if (!w->mute) {
         if (flux_respond_error (ns->ctx->h, w->request, errno, NULL) < 0)
             flux_log_error (ns->ctx->h, "%s: flux_respond_error", __FUNCTION__);
+        w->mute = true;
     }
 error:
-    zlist_remove (ns->watchers, w);
-    watcher_destroy (w);
+    namespace_watchers_remove (ns, w);
+    watcher_decref (w);
     if (zlist_size (ns->watchers) == 0)
         zhash_delete (ns->ctx->namespaces, ns->name);
 }
@@ -822,8 +849,9 @@ static void getroot_cb (flux_t *h, flux_msg_handler_t *mh,
     if (!(w = watcher_create (msg, NULL, flags)))
         goto error;
     w->ns = ns;
+    /* watcher_create inits refcount to 1, which is used on this append */
     if (zlist_append (ns->watchers, w) < 0) {
-        watcher_destroy (w);
+        watcher_decref (w);
         errno = ENOMEM;
         goto error;
     }
@@ -866,8 +894,9 @@ static void lookup_cb (flux_t *h, flux_msg_handler_t *mh,
     if (!(w = watcher_create (msg, key_suffix ? key_suffix : key, flags)))
         goto error;
     w->ns = ns;
+    /* watcher_create inits refcount to 1, which is used on this append */
     if (zlist_append (ns->watchers, w) < 0) {
-        watcher_destroy (w);
+        watcher_decref (w);
         errno = ENOMEM;
         goto error;
     }