job-manager: avoid posting redundant and failing dependency events

doc/man7/flux-jobtap-plugins.rst

@@ -230,8 +230,10 @@ job.state.depend
   a job. Jobs remain in ``DEPEND`` state until all dependencies are removed
   via ``flux_jobtap_dependency_remove()``. A dependency may be used only once.
   A second ``flux_jobtap_dependency_add()`` call with the same description
-  returns ``EEXIST``, even if the dependency was removed. This enables
-  idempotent operation during job-manager or plugin restart.
+  succeeds with no effect, even if the dependency was removed. Likewise,
+  removing a dependency that was already removed succeeds with no effect.
+  This enables idempotent operation during job-manager or plugin restart.
+  Removing a description that was never added fails with ``ENOENT``.
 
   .. note::
     See :ref:`event_processing` for important information about when state

src/modules/job-manager/job.c

@@ -587,6 +587,29 @@ bool job_event_is_queued (struct job *job, const char *name)
     return false;
 }
 
+bool job_dependency_is_queued (struct job *job,
+                               const char *name,
+                               const char *description)
+{
+    size_t index;
+    json_t *wrap;
+    const char *entry_name;
+    const char *s;
+
+    json_array_foreach (job->event_queue, index, wrap) {
+        if (json_unpack (wrap,
+                         "{s:{s:s s:{s:s}}}",
+                         "entry",
+                           "name", &entry_name,
+                           "context",
+                             "description", &s) == 0
+            && streq (entry_name, name)
+            && streq (s, description))
+            return true;
+    }
+    return false;
+}
+
 const char *job_event_queue_print (struct job *job, char *buf, int size)
 {
     size_t index;

src/modules/job-manager/job.h

@@ -128,6 +128,13 @@ int job_event_enqueue (struct job *job, int flags, json_t *entry);
 int job_event_dequeue (struct job *job, int *flagsp, json_t **entryp);
 int job_event_peek (struct job *job, int *flagsp, json_t **entryp);
 bool job_event_is_queued (struct job *job, const char *name);
+
+/* Return true if a dependency event 'name' (dependency-add or
+ * dependency-remove) for 'description' is queued.
+ */
+bool job_dependency_is_queued (struct job *job,
+                               const char *name,
+                               const char *description);
 const char *job_event_queue_print (struct job *job, char *buf, int size);
 
 /*  Validate updates as valid RFC 21 jobspec-update event context.

src/modules/job-manager/jobtap.c

@@ -2130,10 +2130,31 @@ static int jobtap_emit_dependency_event (struct jobtap *jobtap,
         errno = EINVAL;
         return -1;
     }
-    /* Avoid duplicate dependency-add events on restart
+    /* A description may be used only once, so adding a description that
+     * is used or queued for add is a no-op. This makes add idempotent on
+     * restart or plugin reload.
      */
-    if (add && grudgeset_contains (job->dependencies, description))
-        return 0;
+    if (add) {
+        if (grudgeset_used (job->dependencies, description)
+            || job_dependency_is_queued (job, "dependency-add", description))
+            return 0;
+    }
+    /* Removing a dependency that is removed or queued for removal is a
+     * no-op. Removing a description that was never added fails. Check for
+     * a queued add, since an add posted earlier in the same callback is not
+     * applied yet.
+     */
+    else {
+        if (job_dependency_is_queued (job, "dependency-remove", description)
+            || (grudgeset_used (job->dependencies, description)
+                && !grudgeset_contains (job->dependencies, description)))
+            return 0;
+        if (!grudgeset_used (job->dependencies, description)
+            && !job_dependency_is_queued (job, "dependency-add", description)) {
+            errno = ENOENT;
+            return -1;
+        }
+    }
     return event_job_post_pack (jobtap->ctx->event,
                                 job,
                                 event,

src/modules/job-manager/jobtap.h

@@ -103,14 +103,15 @@ int flux_jobtap_reject_job (flux_plugin_t *p,
  *
  *  This function is only valid from the job.state.depend callback.
  *
- *  A job dependency may only be added to a job once.
+ *  A job dependency may only be added to a job once. Adding a
+ *   description that was already used succeeds with no effect, even if
+ *   the dependency was removed.
  *
  *  The description may not be empty and may be at most 256 characters
  *   long.
  *
  *  Returns 0 on success, or -1 on error with errno set:
  *   - ENOENT: job 'id' not found
- *   - EEXIST: dependency 'description' has already been used
  *   - EINVAL: invalid argument
  *
  */
@@ -122,8 +123,11 @@ int flux_jobtap_dependency_add (flux_plugin_t *p,
  *   Once all outstanding job dependencies have been removed, then the job
  *   will proceed out of the DEPEND state.
  *
+ *  Removing a dependency that was already removed succeeds with no
+ *   effect.
+ *
  *  Returns 0 on success, -1 on failure with errno set:
- *   ENOENT: job 'id' not found or 'description' is not a current dependency
+ *   ENOENT: job 'id' not found or 'description' was never added
  *   EINVAL: invalid argument
  *
  */

t/job-manager/plugins/dependency-test.c

@@ -109,7 +109,9 @@ static int dependency_test_cb (flux_plugin_t *p,
                         name);
         return -1;
     }
-    if (remove) {
+    /* Remove the dependency 'remove' times, to test repeated removal.
+     */
+    for (int i = 0; i < remove; i++) {
         if (flux_jobtap_dependency_remove (p, id, name) < 0)
             return flux_jobtap_reject_job (p, args,
                                            "dependency_remove: %s",
@@ -128,16 +130,18 @@ static int depend_cb (flux_plugin_t *p,
                       void *data)
 {
     const char *description = NULL;
+    const char *absent = NULL;
     flux_jobid_t id;
 
     if (flux_plugin_arg_unpack (args,
                                 FLUX_PLUGIN_ARG_IN,
-                                "{s:I s:{s:{s?{s?s}}}}",
+                                "{s:I s:{s:{s?{s?s s?s}}}}",
                                 "id", &id,
                                 "jobspec",
                                 "attributes",
                                 "system",
-                                "dependency-test", &description) < 0) {
+                                "dependency-test", &description,
+                                "dependency-test-absent", &absent) < 0) {
         flux_jobtap_raise_exception (p, FLUX_JOBTAP_CURRENT_JOB,
                                      "dependency-test", 0,
                                      "failed to unpack dependency-test args");
@@ -153,6 +157,20 @@ static int depend_cb (flux_plugin_t *p,
             return -1;
         }
     }
+    /* Removing a dependency that was never added must fail with ENOENT.
+     */
+    if (absent) {
+        if (flux_jobtap_dependency_remove (p, id, absent) == 0
+            || errno != ENOENT) {
+            flux_jobtap_raise_exception (p,
+                                         FLUX_JOBTAP_CURRENT_JOB,
+                                         "dependency-test", 0,
+                                         "dependency_remove (%s) did not"
+                                         " fail with ENOENT",
+                                         absent);
+            return -1;
+        }
+    }
     return 0;
 }
 

t/t2270-job-dependencies.t

@@ -165,6 +165,46 @@ test_expect_success 'dependency add/remove in callback does not incorrectly rele
 	flux python dep-remove.py ${id} foo &&
 	flux job wait-event -vt 15 ${id} clean
 '
+dependency_event_count() {
+	flux job eventlog -f json $1 \
+		| jq -s "map(select(.name == \"$2\"
+			and .context.description == \"$3\")) | length"
+}
+test_expect_success 'repeated dependency remove succeeds without an event' '
+	id=$(flux submit \
+		--dependency=test:a \
+		--dependency=test:b \
+		hostname) &&
+	flux job wait-event -t 15 -m description=b ${id} dependency-add &&
+	flux python dep-remove.py ${id} a &&
+	flux python dep-remove.py ${id} a &&
+	test $(dependency_event_count ${id} dependency-remove a) -eq 1
+'
+test_expect_success 'removing a dependency never added fails with ENOENT' '
+	test_must_fail flux python dep-remove.py ${id} notadded 2>notadded.err &&
+	grep "Errno 2" notadded.err &&
+	test $(dependency_event_count ${id} dependency-remove notadded) -eq 0 &&
+	flux python dep-remove.py ${id} b &&
+	flux job wait-event -vt 15 ${id} clean
+'
+test_expect_success 'repeated dependency remove in a callback posts one event' '
+	id=$(flux submit \
+		--dependency=test:bar?remove=2 \
+		--dependency=test:foo \
+		hostname) &&
+	flux job wait-event -t 15 -m description=bar ${id} dependency-remove &&
+	test $(dependency_event_count ${id} dependency-remove bar) -eq 1 &&
+	test_must_fail sh -c "flux dmesg -H | grep \"${id}: error processing\"" &&
+	flux python dep-remove.py ${id} foo &&
+	flux job wait-event -vt 15 ${id} clean
+'
+test_expect_success 'removing a dependency never added in a callback fails' '
+	id=$(flux submit \
+		--setattr=system.dependency-test-absent=notadded \
+		hostname) &&
+	flux job wait-event -vt 15 ${id} clean &&
+	test_must_fail flux job wait-event -t 0.1 ${id} exception
+'
 test_expect_success 'restart: start job with 2 dependencies to test restart' '
 	jobid=$(flux submit \
 		--dependency=test:foo \
@@ -202,6 +242,9 @@ test_expect_success 'restart: reload dependency test plugin' '
 test_expect_success 'restart: job.dependency.test called for jobs on load' '
 	flux python dep-check.py ${jobid} foo
 '
+test_expect_success 'restart: reload does not re-add a removed dependency' '
+	test $(dependency_event_count ${jobid} dependency-add bar) -eq 1
+'
 test_expect_success 'restart: subsequent removal of dependency releases job' '
 	flux python dep-remove.py ${jobid} foo &&
 	flux jobs -o {id}:{dependencies} &&