fix(rule): reimplement Automated Rule activation using Quartz jobs

src/main/docker/include/cryostat.jfc

@@ -35,6 +35,22 @@
    0 ms
   </setting>
  </event>
+ <event name="io.cryostat.rules.RuleService.RuleActivationEvent">
+  <setting name="enabled">
+   true
+  </setting>
+  <setting name="threshold">
+   0 ms
+  </setting>
+ </event>
+ <event name="io.cryostat.rules.RuleService.RecordingCleanupEvent">
+  <setting name="enabled">
+   true
+  </setting>
+  <setting name="threshold">
+   0 ms
+  </setting>
+ </event>
  <event name="jdk.ThreadAllocationStatistics">
   <setting name="enabled">
    true

src/main/java/io/cryostat/recordings/RecordingHelper.java

@@ -293,11 +293,19 @@ private List<ActiveRecording> listActiveRecordingsImpl(Target target) {
             List<IRecordingDescriptor> descriptors =
                     connectionManager.executeConnectedTask(
                             target, conn -> conn.getService().getAvailableRecordings());
+
+            var remoteIds =
+                    new HashSet<>(descriptors.stream().map(IRecordingDescriptor::getId).toList());
+
             boolean updated = false;
             var it = target.activeRecordings.iterator();
             while (it.hasNext()) {
                 var r = it.next();
-                if (!previousIds.contains(r.remoteId)) {
+                if (!remoteIds.contains(r.remoteId)) {
+                    logger.warnv(
+                            "Orphaned recording detected: id={0} remoteId={1} name={2} on target"
+                                    + " {3}, removing from database",
+                            r.id, r.remoteId, r.name, target.connectUrl);
                     r.delete();
                     it.remove();
                     updated |= true;
@@ -469,101 +477,137 @@ private ActiveRecording startRecordingImpl(
         }
         getActiveRecording(lockedTarget, r -> r.name.equals(recordingName))
                 .ifPresent(r -> this.deleteRecording(r).await().indefinitely());
-        var desc =
-                connectionManager.executeConnectedTask(
-                        lockedTarget,
-                        conn -> {
-                            RecordingOptionsBuilder optionsBuilder =
-                                    recordingOptionsBuilderFactory
-                                            .create(lockedTarget)
-                                            .name(recordingName);
-                            if (options.duration().isPresent()) {
-                                optionsBuilder =
-                                        optionsBuilder.duration(
-                                                TimeUnit.SECONDS.toMillis(
-                                                        options.duration().get()));
-                            }
-                            if (options.toDisk().isPresent()) {
-                                optionsBuilder = optionsBuilder.toDisk(options.toDisk().get());
-                            }
-                            if (options.maxAge().isPresent()) {
-                                optionsBuilder = optionsBuilder.maxAge(options.maxAge().get());
-                            }
-                            if (options.maxSize().isPresent()) {
-                                optionsBuilder = optionsBuilder.maxSize(options.maxSize().get());
-                            }
-                            IConstrainedMap<String> recordingOptions = optionsBuilder.build();
-
-                            switch (template.getType()) {
-                                case PRESET:
-                                    return conn.getService()
-                                            .start(
-                                                    recordingOptions,
-                                                    presetTemplateService
-                                                            .getXml(
-                                                                    template.getName(),
-                                                                    TemplateType.CUSTOM)
-                                                            .orElseThrow());
-                                case CUSTOM:
-                                    return conn.getService()
-                                            .start(
-                                                    recordingOptions,
-                                                    customTemplateService
-                                                            .getXml(
-                                                                    template.getName(),
-                                                                    TemplateType.CUSTOM)
-                                                            .orElseThrow());
-                                case TARGET:
-                                    return conn.getService().start(recordingOptions, template);
-                                default:
-                                    throw new IllegalStateException(
-                                            "Unknown template type: " + template.getType());
-                            }
-                        });
+
+        IRecordingDescriptor desc;
+        try {
+            desc =
+                    connectionManager.executeConnectedTask(
+                            lockedTarget,
+                            conn -> {
+                                RecordingOptionsBuilder optionsBuilder =
+                                        recordingOptionsBuilderFactory
+                                                .create(lockedTarget)
+                                                .name(recordingName);
+                                if (options.duration().isPresent()) {
+                                    optionsBuilder =
+                                            optionsBuilder.duration(
+                                                    TimeUnit.SECONDS.toMillis(
+                                                            options.duration().get()));
+                                }
+                                if (options.toDisk().isPresent()) {
+                                    optionsBuilder = optionsBuilder.toDisk(options.toDisk().get());
+                                }
+                                if (options.maxAge().isPresent()) {
+                                    optionsBuilder = optionsBuilder.maxAge(options.maxAge().get());
+                                }
+                                if (options.maxSize().isPresent()) {
+                                    optionsBuilder =
+                                            optionsBuilder.maxSize(options.maxSize().get());
+                                }
+                                IConstrainedMap<String> recordingOptions = optionsBuilder.build();
+
+                                switch (template.getType()) {
+                                    case PRESET:
+                                        return conn.getService()
+                                                .start(
+                                                        recordingOptions,
+                                                        presetTemplateService
+                                                                .getXml(
+                                                                        template.getName(),
+                                                                        TemplateType.CUSTOM)
+                                                                .orElseThrow());
+                                    case CUSTOM:
+                                        return conn.getService()
+                                                .start(
+                                                        recordingOptions,
+                                                        customTemplateService
+                                                                .getXml(
+                                                                        template.getName(),
+                                                                        TemplateType.CUSTOM)
+                                                                .orElseThrow());
+                                    case TARGET:
+                                        return conn.getService().start(recordingOptions, template);
+                                    default:
+                                        throw new IllegalStateException(
+                                                "Unknown template type: " + template.getType());
+                                }
+                            });
+        } catch (Exception e) {
+            logger.errorv(
+                    e,
+                    "Failed to start remote recording \"{0}\" on target {1}",
+                    recordingName,
+                    lockedTarget.connectUrl);
+            throw e;
+        }
 
         Map<String, String> labels = new HashMap<>(rawLabels);
         labels.put("template.name", template.getName());
         labels.put("template.type", template.getType().toString());
         Metadata meta = new Metadata(labels);
 
-        ActiveRecording recording = ActiveRecording.from(lockedTarget, desc, meta, options);
-
-        Optional<ActiveRecording> existingOpt =
-                ActiveRecording.<ActiveRecording>find(
-                                "target.id = ?1 and remoteId = ?2",
-                                lockedTarget.id,
-                                recording.remoteId)
-                        .firstResultOptional();
-
-        if (existingOpt.isPresent()) {
-            ActiveRecording existingRecording = existingOpt.get();
-            logger.infov(
-                    "Found existing recording id={0} remoteId={1} name={2},"
-                            + " merging state from startRecording",
-                    existingRecording.id, existingRecording.remoteId, existingRecording.name);
-
-            existingRecording.name = recording.name;
-            existingRecording.state = recording.state;
-            existingRecording.duration = recording.duration;
-            existingRecording.startTime = recording.startTime;
-            existingRecording.archiveOnStop = recording.archiveOnStop;
-            existingRecording.continuous = recording.continuous;
-            existingRecording.toDisk = recording.toDisk;
-            existingRecording.maxSize = recording.maxSize;
-            existingRecording.maxAge = recording.maxAge;
-            existingRecording.external = recording.external;
-
-            var mergedLabels = new HashMap<>(existingRecording.metadata.labels());
-            mergedLabels.putAll(recording.metadata.labels());
-            existingRecording.metadata = new Metadata(mergedLabels);
-
-            existingRecording.persist();
-            recording = existingRecording;
-        } else {
-            recording.persist();
-        }
+        ActiveRecording recording;
+        try {
+            recording = ActiveRecording.from(lockedTarget, desc, meta, options);
+
+            Optional<ActiveRecording> existingOpt =
+                    ActiveRecording.<ActiveRecording>find(
+                                    "target.id = ?1 and remoteId = ?2",
+                                    lockedTarget.id,
+                                    recording.remoteId)
+                            .firstResultOptional();
+
+            if (existingOpt.isPresent()) {
+                ActiveRecording existingRecording = existingOpt.get();
+                logger.infov(
+                        "Found existing recording id={0} remoteId={1} name={2},"
+                                + " merging state from startRecording",
+                        existingRecording.id, existingRecording.remoteId, existingRecording.name);
+
+                existingRecording.name = recording.name;
+                existingRecording.state = recording.state;
+                existingRecording.duration = recording.duration;
+                existingRecording.startTime = recording.startTime;
+                existingRecording.archiveOnStop = recording.archiveOnStop;
+                existingRecording.continuous = recording.continuous;
+                existingRecording.toDisk = recording.toDisk;
+                existingRecording.maxSize = recording.maxSize;
+                existingRecording.maxAge = recording.maxAge;
+                existingRecording.external = recording.external;
+
+                var mergedLabels = new HashMap<>(existingRecording.metadata.labels());
+                mergedLabels.putAll(recording.metadata.labels());
+                existingRecording.metadata = new Metadata(mergedLabels);
+
+                existingRecording.persist();
+                recording = existingRecording;
+            } else {
+                recording.persist();
+            }
 
-        lockedTarget.activeRecordings.add(recording);
+            lockedTarget.activeRecordings.add(recording);
+        } catch (Exception e) {
+            logger.errorv(
+                    e,
+                    "Failed to persist recording \"{0}\" to database, attempting to stop remote"
+                            + " recording",
+                    recordingName);
+            try {
+                connectionManager.executeConnectedTask(
+                        lockedTarget,
+                        conn -> {
+                            conn.getService().close(desc);
+                            return null;
+                        });
+            } catch (Exception cleanupEx) {
+                logger.warnv(
+                        cleanupEx,
+                        "Failed to cleanup remote recording \"{0}\" after database persist"
+                                + " failure",
+                        recordingName);
+            }
+            throw e;
+        }
 
         if (!recording.continuous) {
             JobKey key =

src/main/java/io/cryostat/rules/RecordingCleanupJob.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright The Cryostat Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.cryostat.rules;
+
+import jakarta.inject.Inject;
+import org.jboss.logging.Logger;
+import org.quartz.DisallowConcurrentExecution;
+import org.quartz.Job;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+
+/**
+ * Quartz job that delegates to RuleService for fault-tolerant recording cleanup.
+ *
+ * <p>This job is scheduled when a recording needs to be cleaned up (stopped). It delegates to
+ * RuleService.cleanupRecording() which handles all retry logic, timeouts, and backpressure via
+ * SmallRye Fault Tolerance annotations.
+ *
+ * <p>The job is one-time only and unscheduled after execution (success or failure).
+ */
+@DisallowConcurrentExecution
+public class RecordingCleanupJob implements Job {
+
+    @Inject Logger logger;
+    @Inject RuleService ruleService;
+
+    @Override
+    public void execute(JobExecutionContext ctx) throws JobExecutionException {
+        String ruleName = ctx.getMergedJobDataMap().getString("ruleName");
+        String jvmId = ctx.getMergedJobDataMap().getString("jvmId");
+        String recordingName = ctx.getMergedJobDataMap().getString("recordingName");
+
+        logger.debugv(
+                "Executing cleanup job: rule={0} jvmId={1} recording={2}",
+                ruleName, jvmId, recordingName);
+
+        try {
+            ruleService.cleanupRecording(ruleName, jvmId, recordingName).await().indefinitely();
+            logger.debugv("Recording cleanup completed: rule={0} jvmId={1}", ruleName, jvmId);
+            ctx.getScheduler().unscheduleJob(ctx.getTrigger().getKey());
+        } catch (Exception e) {
+            logger.errorv(
+                    e,
+                    "Recording cleanup failed after all retries: rule={0} jvmId={1}",
+                    ruleName,
+                    jvmId);
+            JobExecutionException ex = new JobExecutionException(e);
+            ex.setUnscheduleFiringTrigger(true);
+            throw ex;
+        }
+    }
+}

src/main/java/io/cryostat/rules/RuleActivationJob.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright The Cryostat Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.cryostat.rules;
+
+import jakarta.inject.Inject;
+import org.jboss.logging.Logger;
+import org.quartz.DisallowConcurrentExecution;
+import org.quartz.Job;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+
+/**
+ * Quartz job that delegates to RuleService for fault-tolerant rule activation.
+ *
+ * <p>This job is scheduled when a rule needs to be activated on a target. It delegates to
+ * RuleService.activateRule() which handles all retry logic, timeouts, and backpressure via SmallRye
+ * Fault Tolerance annotations.
+ *
+ * <p>The job is one-time only and unscheduled after execution (success or failure).
+ */
+@DisallowConcurrentExecution
+public class RuleActivationJob implements Job {
+
+    @Inject Logger logger;
+    @Inject RuleService ruleService;
+
+    @Override
+    public void execute(JobExecutionContext ctx) throws JobExecutionException {
+        String ruleName = ctx.getMergedJobDataMap().getString("ruleName");
+        String jvmId = ctx.getMergedJobDataMap().getString("jvmId");
+
+        logger.debugv("Executing rule activation job: rule={0} jvmId={1}", ruleName, jvmId);
+
+        try {
+            ruleService.activateRule(ruleName, jvmId).await().indefinitely();
+            logger.debugv("Rule activation completed: rule={0} jvmId={1}", ruleName, jvmId);
+            ctx.getScheduler().unscheduleJob(ctx.getTrigger().getKey());
+        } catch (Exception e) {
+            logger.errorv(
+                    e,
+                    "Rule activation failed after all retries: rule={0} jvmId={1}",
+                    ruleName,
+                    jvmId);
+            JobExecutionException ex = new JobExecutionException(e);
+            ex.setUnscheduleFiringTrigger(true);
+            throw ex;
+        }
+    }
+}