build: depend on qbit-object-storage 0.1.0-RC.1 (Central), not a SNAPSHOT

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

docs/screenshots/README.md

@@ -0,0 +1,7 @@
+# Backups home and record screenshots
+
+Taken 2026-10-12 against `BackupsDemoServer` (src/test/.../demo, memory backend, real engine history) and the Next UI 1.0.0-RC.14 jar.
+Light and dark at 390, 600, 768, 1024, 1440, 1920 px (home) and 390, 1440 px (records).
+
+- `.` : this branch (foundation base): home, plan, run, failed run, lists.
+- `with-restore-verify/` : a scratch merge of PRs #5 (verify), #6 (restore), #7 (prune/schedule) onto this branch, so verifications, a restore run and retention classes exist. Not part of this PR.

pom.xml

@@ -86,6 +86,35 @@
          <optional>true</optional>
       </dependency>
 
+      <!-- Optional: the production BackupStorage adapter. Apps that install this qbit -->
+      <!-- with object storage already have it; others use their own BackupStorage.   -->
+      <dependency>
+         <groupId>com.kingsrook.qbits</groupId>
+         <artifactId>qbit-object-storage</artifactId>
+         <version>0.1.0-RC.1</version>
+         <optional>true</optional>
+      </dependency>
+
+      <dependency>
+         <groupId>org.testcontainers</groupId>
+         <artifactId>testcontainers</artifactId>
+         <version>2.0.3</version>
+         <scope>test</scope>
+      </dependency>
+
+      <!-- Demo server only (src/test/.../demo): never shipped. -->
+      <dependency>
+         <groupId>com.kingsrook.qqq</groupId>
+         <artifactId>qqq-middleware-javalin</artifactId>
+         <scope>test</scope>
+      </dependency>
+      <dependency>
+         <groupId>com.kingsrook.qqq</groupId>
+         <artifactId>qqq-frontend-next</artifactId>
+         <version>1.0.0-RC.14</version>
+         <scope>test</scope>
+      </dependency>
+
       <dependency>
          <groupId>org.junit.jupiter</groupId>
          <artifactId>junit-jupiter</artifactId>
@@ -101,6 +130,17 @@
 
    <build>
       <plugins>
+         <plugin>
+            <groupId>org.apache.maven.plugins</groupId>
+            <artifactId>maven-surefire-plugin</artifactId>
+            <configuration>
+               <!-- the MinIO test container default credentials, resolved through ${env.X} refs like production -->
+               <environmentVariables>
+                  <STORAGE_TEST_ACCESS_KEY>minioadmin</STORAGE_TEST_ACCESS_KEY>
+                  <STORAGE_TEST_SECRET_KEY>minioadmin</STORAGE_TEST_SECRET_KEY>
+               </environmentVariables>
+            </configuration>
+         </plugin>
          <plugin>
             <groupId>org.jacoco</groupId>
             <artifactId>jacoco-maven-plugin</artifactId>

src/main/java/com/kingsrook/qbits/backups/BackupsQBitConfig.java

@@ -45,6 +45,7 @@ public class BackupsQBitConfig implements QBitConfig
    private String appName  = "backups";
    private String appLabel = "Backups";
    private String parentAppName;
+   private String tablePathPrefix;
    private String schedulerName;
    private String instanceName;
 
@@ -179,6 +180,28 @@ public class BackupsQBitConfig implements QBitConfig
 
 
 
+   /*******************************************************************************
+    ** Getter for tablePathPrefix (the front of the links the home makes to tables and
+    ** records, "/app" for the Next dashboard, which routes /app/{table}/{id}).
+    *******************************************************************************/
+   public String getTablePathPrefix()
+   {
+      return (tablePathPrefix == null ? "/app" : tablePathPrefix);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for tablePathPrefix
+    *******************************************************************************/
+   public BackupsQBitConfig withTablePathPrefix(String tablePathPrefix)
+   {
+      this.tablePathPrefix = tablePathPrefix;
+      return (this);
+   }
+
+
+
    /*******************************************************************************
     ** Getter for schedulerName (the QQQ scheduler used for plan jobs and the SLA checker)
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/backups/BackupsQBitProducer.java

@@ -149,12 +149,29 @@ public class BackupsQBitProducer implements QBitMetaDataProducer<BackupsQBitConf
          process.setSourceQBitName(qBit.getName());
          output.add(process);
       }
+      for(var process : com.kingsrook.qbits.backups.processes.ExportTableProcess.exportProcesses(qInstance))
+      {
+         process.setSourceQBitName(qBit.getName());
+         output.add(process);
+      }
       for(var widget : com.kingsrook.qbits.backups.app.BackupsApp.widgets())
       {
          output.add(widget);
       }
-      output.add(com.kingsrook.qbits.backups.app.BackupsApp.app(qBitConfig, tables));
+      for(var widget : com.kingsrook.qbits.backups.model.BackupsTables.widgets())
+      {
+         output.add(widget);
+      }
+      com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData app = com.kingsrook.qbits.backups.app.BackupsApp.app(qBitConfig, tables);
+      output.add(app);
       output.addSelfToInstance(qInstance);
+
+      // A parent app only lists the apps that are its children, so attach ourselves (for example Admin, then Backups).
+      com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData parent = qBitConfig.getParentAppName() == null ? null : qInstance.getApp(qBitConfig.getParentAppName());
+      if(parent != null && (parent.getChildren() == null || parent.getChildren().stream().noneMatch(child -> app.getName().equals(child.getName()))))
+      {
+         parent.withChild(app);
+      }
       return (new com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerMultiOutput());
    }
 

src/main/java/com/kingsrook/qbits/backups/app/BackupsApp.java

@@ -29,7 +29,6 @@ import com.kingsrook.qqq.backend.core.model.dashboard.widgets.WidgetType;
 import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
 import com.kingsrook.qqq.backend.core.model.metadata.dashboard.QWidgetMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData;
-import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppSection;
 import com.kingsrook.qqq.backend.core.model.metadata.layout.QIcon;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
 
@@ -62,7 +61,7 @@ public final class BackupsApp
    public static List<QWidgetMetaData> widgets()
    {
       return (List.of(
-         new QWidgetMetaData().withName(HOME_WIDGET).withLabel("Backups").withType(WidgetType.HTML.getType())
+         new QWidgetMetaData().withName(HOME_WIDGET).withLabel("Overview").withType(WidgetType.HTML.getType())
             .withCodeReference(new QCodeReference(HomeRenderer.class)).withGridColumns(12).withIsCard(false).withPermissionRules(BackupsTables.permissionRules()),
          new QWidgetMetaData().withName(HEALTH_CARD_WIDGET).withLabel("Backups").withType(WidgetType.HTML.getType())
             .withCodeReference(new QCodeReference(HealthCardRenderer.class)).withGridColumns(12).withIsCard(true).withPermissionRules(BackupsTables.permissionRules())));
@@ -80,13 +79,7 @@ public final class BackupsApp
          .withLabel(config.getAppLabel())
          .withIcon(new QIcon("backup"))
          .withPermissionRules(BackupsTables.permissionRules())
-         .withWidgets(List.of(HOME_WIDGET))
-         .withSection(new QAppSection().withName("backups").withLabel("Backups").withIcon(new QIcon("backup"))
-            .withTables(List.of(BackupsTables.PLAN, BackupsTables.RUN)))
-         .withSection(new QAppSection().withName("recovery").withLabel("Restore and verify").withIcon(new QIcon("restore"))
-            .withTables(List.of(BackupsTables.RESTORE_RUN, BackupsTables.VERIFICATION)))
-         .withSection(new QAppSection().withName("files").withLabel("Files").withIcon(new QIcon("attach_file"))
-            .withTables(List.of(BackupsTables.FILE_OBJECT)));
+         .withWidgets(List.of(HOME_WIDGET));
       if(config.getParentAppName() != null)
       {
          app.setParentAppName(config.getParentAppName());
@@ -143,10 +136,6 @@ public final class BackupsApp
    static String pathPrefix()
    {
       var config = com.kingsrook.qbits.backups.BackupsRuntime.getConfig();
-      if(config == null)
-      {
-         return ("/backups");
-      }
-      return (config.getParentAppName() != null ? "/" + config.getParentAppName() + "/" + config.getAppName() : "/" + config.getAppName());
+      return (config == null ? "/app" : config.getTablePathPrefix());
    }
 }

src/main/java/com/kingsrook/qbits/backups/app/BackupsHomeData.java

@@ -159,12 +159,53 @@ public class BackupsHomeData
 
 
 
+   /*******************************************************************************
+    ** The status to show for a plan: the stored slaStatus when it is a problem state,
+    ** otherwise LATE when the age of the last success is past the SLA (the SLA
+    ** checker may not have run yet), otherwise the stored status (UNKNOWN when none).
+    *******************************************************************************/
+   public String effectiveStatus(QRecord plan)
+   {
+      String stored = plan.getValueString("slaStatus");
+      if(stored != null && ATTENTION.contains(stored))
+      {
+         return (stored);
+      }
+      Long    age = ageHours(plan);
+      Integer sla = plan.getValueInteger("slaHours");
+      if(age != null && sla != null && age > sla)
+      {
+         return ("LATE");
+      }
+      return (stored == null ? "UNKNOWN" : stored);
+   }
+
+
+
    /*******************************************************************************
     ** Plans that need attention.
     *******************************************************************************/
    public List<QRecord> plansNeedingAttention()
    {
-      return (plans.stream().filter(p -> Boolean.TRUE.equals(p.getValueBoolean("isActive")) && ATTENTION.contains(p.getValueString("slaStatus"))).toList());
+      return (plans.stream().filter(p -> Boolean.TRUE.equals(p.getValueBoolean("isActive")) && ATTENTION.contains(effectiveStatus(p))).toList());
+   }
+
+
+
+   /*******************************************************************************
+    ** The last run of a plan (from the recent runs), or null.
+    *******************************************************************************/
+   public QRecord lastRun(QRecord plan)
+   {
+      Integer lastRunId = plan.getValueInteger("lastRunId");
+      for(QRecord run : recentRuns)
+      {
+         if(lastRunId != null && lastRunId.equals(run.getValueInteger("id")))
+         {
+            return (run);
+         }
+      }
+      return (null);
    }
 
 
@@ -206,7 +247,7 @@ public class BackupsHomeData
             Integer runId = verification.getValueInteger("backupRunId");
             for(QRecord run : recentRuns)
             {
-               if(runId != null && runId.equals(run.getValueInteger("id")) && planId.equals(run.getValueInteger("backupPlanId")))
+               if(runId != null && planId != null && runId.equals(run.getValueInteger("id")) && planId.equals(run.getValueInteger("backupPlanId")))
                {
                   return (verification);
                }
@@ -224,6 +265,10 @@ public class BackupsHomeData
    public List<QRecord> successfulRuns(Integer planId)
    {
       List<QRecord> rs = new ArrayList<>();
+      if(planId == null)
+      {
+         return (rs);
+      }
       for(QRecord run : recentRuns)
       {
          if(planId.equals(run.getValueInteger("backupPlanId")) && run.getValueString("status") != null && run.getValueString("status").startsWith("SUCCEEDED"))
@@ -242,6 +287,10 @@ public class BackupsHomeData
    public Map<String, Integer> retentionCounts(Integer planId)
    {
       Map<String, Integer> counts = new LinkedHashMap<>();
+      if(planId == null)
+      {
+         return (counts);
+      }
       for(QRecord run : recentRuns)
       {
          if(planId.equals(run.getValueInteger("backupPlanId")) && run.getValueString("status") != null && run.getValueString("status").startsWith("SUCCEEDED")

src/main/java/com/kingsrook/qbits/backups/app/BackupsHomeView.java

@@ -27,21 +27,30 @@ import com.kingsrook.qqq.backend.core.model.data.QRecord;
 
 
 /*******************************************************************************
- ** Renders the Backups home: one card per concern, each with an honest empty
- ** state.  Cards stack to one column on narrow screens, long names wrap, and
- ** there are no tables and no inner horizontal scroll (definition lists and
- ** lists only), so it reads at every width in light and dark mode (colors are
- ** inherited, not set).
+ ** Renders the Backups home: a plan status row per plan across the full width,
+ ** then cards for what needs attention, the latest test restore, recent activity,
+ ** the size trend and retention.  Every card has an honest empty state.  Layout
+ ** is flow and grid only (no tables, no fixed widths, no inner scrolling): plan
+ ** rows wrap their cells and the cards stack to one column on narrow screens,
+ ** long names wrap, and colors are inherited (only the status dot is colored), so
+ ** it reads at every width in light and dark mode.
  *******************************************************************************/
 public final class BackupsHomeView
 {
    private static final DateTimeFormatter WHEN = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm 'UTC'").withZone(ZoneOffset.UTC);
 
-   private static final String GRID  = "display:grid;grid-template-columns:repeat(auto-fit,minmax(min(100%,22rem),1fr));gap:1rem;";
+   private static final String ROOT  = "display:grid;gap:1rem;min-width:0;";
+   private static final String GRID  = "display:grid;grid-template-columns:repeat(auto-fit,minmax(min(100%,24rem),1fr));gap:1rem;align-items:start;min-width:0;";
    private static final String CARD  = "border:1px solid rgba(127,127,127,.3);border-radius:8px;padding:1rem;min-width:0;overflow-wrap:anywhere;";
+   private static final String ROW   = "border:1px solid rgba(127,127,127,.25);border-radius:6px;padding:.75rem;margin:0 0 .75rem;min-width:0;";
+   private static final String CELLS = "display:grid;grid-template-columns:repeat(auto-fit,minmax(9.5rem,1fr));gap:.6rem 1rem;margin:.6rem 0 0;font-size:.9rem;";
    private static final String LIST  = "display:grid;grid-template-columns:minmax(7rem,auto) 1fr;gap:.25rem .75rem;margin:0 0 .75rem;font-size:.9rem;";
-   private static final String ITEMS = "margin:0 0 .75rem;padding-left:1.25rem;font-size:.9rem;";
+   private static final String ITEMS = "margin:0 0 .75rem;padding-left:1.25rem;list-style:disc;font-size:.9rem;display:grid;gap:.35rem;";
    private static final String MUTED = "opacity:.75;";
+   private static final String LINK  = "color:inherit;text-decoration:underline;text-underline-offset:2px;";
+   private static final String CHIP  = "display:inline-flex;align-items:center;gap:.4rem;padding:.05rem .6rem;border-radius:999px;border:1px solid rgba(127,127,127,.4);font-size:.8rem;white-space:nowrap;";
+
+   private static final int TREND_BARS = 30;
 
 
 
@@ -59,14 +68,15 @@ public final class BackupsHomeView
     *******************************************************************************/
    public static String render(BackupsHomeData data, String pathPrefix)
    {
-      StringBuilder html = new StringBuilder("<div class=\"backups-home\" style=\"" + GRID + "\">");
+      StringBuilder html = new StringBuilder("<div class=\"backups-home\" style=\"" + ROOT + "\">");
       planStatus(html, data, pathPrefix);
+      html.append("<div style=\"").append(GRID).append("\">");
       needsAttention(html, data, pathPrefix);
       latestTestRestore(html, data, pathPrefix);
-      trend(html, data);
       recentActivity(html, data, pathPrefix);
+      trend(html, data);
       retention(html, data);
-      return (html.append("</div>").toString());
+      return (html.append("</div></div>").toString());
    }
 
 
@@ -82,13 +92,13 @@ public final class BackupsHomeView
          case "ATTENTION" -> "Needs attention: see the Backups home.";
          default -> "All backups are within their SLA.";
       };
-      return ("<div style=\"" + CARD + "\"><strong>Backups</strong><p>" + esc(text) + "</p><a href=\"" + esc(pathPrefix) + "\">Open the Backups home</a></div>");
+      return ("<div style=\"" + CARD + "\"><strong>Backups</strong><p>" + esc(text) + "</p><a style=\"" + LINK + "\" href=\"" + esc(pathPrefix) + "\">Open the Backups home</a></div>");
    }
 
 
 
    /*******************************************************************************
-    ** Plan status: last successful backup, age against the SLA, last run, verification.
+    ** Plan status: one row per plan, the cells wrap to the width.
     *******************************************************************************/
    private static void planStatus(StringBuilder html, BackupsHomeData data, String prefix)
    {
@@ -99,24 +109,29 @@ public final class BackupsHomeView
       }
       for(QRecord plan : data.plans())
       {
-         Long   age = data.ageHours(plan);
-         Integer sla = plan.getValueInteger("slaHours");
-         html.append("<dl style=\"").append(LIST).append("\">");
-         html.append("<dt><strong>Plan</strong></dt><dd><a href=\"").append(esc(prefix + "/" + BackupsTables.PLAN + "/" + plan.getValue("id"))).append("\">")
-            .append(esc(plan.getValueString("name"))).append("</a>").append(Boolean.TRUE.equals(plan.getValueBoolean("isActive")) ? "" : " (not active)").append("</dd>");
-         row(html, "Status", label(plan.getValueString("slaStatus")));
-         row(html, "Last backup", plan.getValueInstant("lastSuccessDate") == null ? "No successful backup yet" : WHEN.format(plan.getValueInstant("lastSuccessDate")));
-         if(age != null && sla != null)
+         Long    age    = data.ageHours(plan);
+         Integer sla    = plan.getValueInteger("slaHours");
+         String  status = data.effectiveStatus(plan);
+         QRecord last   = data.lastRun(plan);
+
+         html.append("<div style=\"").append(ROW).append("\"><div style=\"display:flex;flex-wrap:wrap;align-items:center;gap:.5rem .75rem;\">");
+         html.append("<strong>").append(recordLink(prefix, BackupsTables.PLAN, plan.getValue("id"), plan.getValueString("name"))).append("</strong>");
+         html.append(chip(status));
+         if(!Boolean.TRUE.equals(plan.getValueBoolean("isActive")))
          {
-            row(html, "Age against SLA", age + " h of " + sla + " h" + (age > sla ? " - late" : ""));
+            html.append("<span style=\"").append(MUTED).append("\">Not active</span>");
          }
-         row(html, "Last verified", plan.getValueInstant("lastVerifiedDate") == null ? "Not verified yet" : WHEN.format(plan.getValueInstant("lastVerifiedDate")));
-         row(html, "Schedule", plan.getValueString("cronExpression") == null ? "Manual only" : plan.getValueString("cronExpression"));
+         html.append("</div><div style=\"").append(CELLS).append("\">");
+         cell(html, "Last backup", plan.getValueInstant("lastSuccessDate") == null ? "No successful backup yet" : WHEN.format(plan.getValueInstant("lastSuccessDate")));
+         cell(html, "Age against SLA", age != null && sla != null ? age + " h of " + sla + " h" : (sla == null ? "No SLA set" : "Not run yet"));
+         cell(html, "Last run", last == null ? "None yet" : label(last.getValueString("status")) + (last.getValueInstant("startDate") == null ? "" : ", " + WHEN.format(last.getValueInstant("startDate"))));
+         cell(html, "Schedule", plan.getValueString("cronExpression") == null ? "Manual only" : plan.getValueString("cronExpression"));
+         cell(html, "Last verified", plan.getValueInstant("lastVerifiedDate") == null ? "Not verified yet" : WHEN.format(plan.getValueInstant("lastVerifiedDate")));
          if(plan.getValueString("objectLockMode") != null && !"NONE".equals(plan.getValueString("objectLockMode")))
          {
-            row(html, "Object lock", plan.getValueString("objectLockMode") + ", " + plan.getValueInteger("objectLockDays") + " days");
+            cell(html, "Object lock", plan.getValueString("objectLockMode") + (plan.getValueInteger("objectLockDays") == null ? "" : ", " + plan.getValueInteger("objectLockDays") + " days"));
          }
-         html.append("</dl>");
+         html.append("</div></div>");
       }
       link(html, prefix, BackupsTables.PLAN, "Backup plans");
       close(html);
@@ -144,13 +159,13 @@ public final class BackupsHomeView
          html.append("<ul style=\"").append(ITEMS).append("\">");
          for(QRecord plan : plans)
          {
-            html.append("<li><a href=\"").append(esc(prefix + "/" + BackupsTables.PLAN + "/" + plan.getValue("id"))).append("\">").append(esc(plan.getValueString("name")))
-               .append("</a>: ").append(esc(label(plan.getValueString("slaStatus")))).append("</li>");
+            html.append("<li>").append(recordLink(prefix, BackupsTables.PLAN, plan.getValue("id"), plan.getValueString("name")))
+               .append(": ").append(esc(label(data.effectiveStatus(plan)))).append("</li>");
          }
          for(QRecord run : data.failedRuns())
          {
-            html.append("<li><a href=\"").append(esc(prefix + "/" + BackupsTables.RUN + "/" + run.getValue("id"))).append("\">Run ").append(esc(run.getValueString("runKey")))
-               .append("</a> ").append(esc(label(run.getValueString("status")))).append("</li>");
+            html.append("<li>Run ").append(recordLink(prefix, BackupsTables.RUN, run.getValue("id"), run.getValueString("runKey")))
+               .append(": ").append(esc(label(run.getValueString("status")))).append("</li>");
          }
          html.append("</ul>");
       }
@@ -178,7 +193,8 @@ public final class BackupsHomeView
          html.append("<dl style=\"").append(LIST).append("\">");
          row(html, "Plan", plan.getValueString("name"));
          row(html, "Result", label(v.getValueString("status")));
-         row(html, "Compared", v.getValueInteger("tablesChecked") + " tables, " + v.getValueLong("rowsChecked") + " rows, " + v.getValueInteger("mismatches") + " mismatches");
+         row(html, "Compared", number(v.getValueLong("rowsChecked")) + " rows in " + number(toLong(v.getValueInteger("tablesChecked"))) + " tables, "
+            + number(toLong(v.getValueInteger("mismatches"))) + " mismatches");
          row(html, "Scratch target", v.getValueString("scratchTarget") == null ? "Not recorded" : v.getValueString("scratchTarget"));
          html.append("</dl>");
       }
@@ -193,7 +209,8 @@ public final class BackupsHomeView
 
 
    /*******************************************************************************
-    ** Size and duration trend per plan, as numbers next to each other.
+    ** Size trend per plan: bars for the stored size of each successful run (oldest
+    ** left), then the numbers.
     *******************************************************************************/
    private static void trend(StringBuilder html, BackupsHomeData data)
    {
@@ -209,13 +226,14 @@ public final class BackupsHomeView
          any = true;
          QRecord first = runs.get(0);
          QRecord last  = runs.get(runs.size() - 1);
+         html.append("<div style=\"margin:0 0 .75rem;\"><strong style=\"font-size:.9rem;\">").append(esc(plan.getValueString("name"))).append("</strong>");
+         bars(html, runs);
          html.append("<dl style=\"").append(LIST).append("\">");
-         row(html, "Plan", plan.getValueString("name"));
          row(html, "Successful runs", String.valueOf(runs.size()));
-         row(html, "Stored (first, latest)", bytes(first.getValueLong("bytesStored")) + ", " + bytes(last.getValueLong("bytesStored")));
-         row(html, "Rows (first, latest)", number(first.getValueLong("rowsTotal")) + ", " + number(last.getValueLong("rowsTotal")));
-         row(html, "Duration (latest)", last.getValueLong("durationMs") == null ? "Not recorded" : last.getValueLong("durationMs") / 1000 + " s");
-         html.append("</dl>");
+         row(html, "Stored", bytes(first.getValueLong("bytesStored")) + " at first, " + bytes(last.getValueLong("bytesStored")) + " latest");
+         row(html, "Rows", number(first.getValueLong("rowsTotal")) + " at first, " + number(last.getValueLong("rowsTotal")) + " latest");
+         row(html, "Duration", last.getValueLong("durationMs") == null ? "Not recorded" : duration(last.getValueLong("durationMs")) + " latest");
+         html.append("</dl></div>");
       }
       if(!any)
       {
@@ -227,7 +245,31 @@ public final class BackupsHomeView
 
 
    /*******************************************************************************
-    ** Recent activity: the last runs and restores.
+    ** A row of bars, one per run, tall in proportion to the stored bytes.
+    *******************************************************************************/
+   private static void bars(StringBuilder html, List<QRecord> runs)
+   {
+      List<QRecord> shown = runs.size() > TREND_BARS ? runs.subList(runs.size() - TREND_BARS, runs.size()) : runs;
+      long          max   = 1;
+      for(QRecord run : shown)
+      {
+         max = Math.max(max, run.getValueLong("bytesStored") == null ? 0 : run.getValueLong("bytesStored"));
+      }
+      html.append("<div role=\"img\" aria-label=\"Stored size of the last ").append(shown.size()).append(" successful runs\" style=\"display:flex;align-items:flex-end;gap:2px;height:2.5rem;margin:.4rem 0 .6rem;\">");
+      for(QRecord run : shown)
+      {
+         long   value = run.getValueLong("bytesStored") == null ? 0 : run.getValueLong("bytesStored");
+         long   pct   = Math.max(4, Math.round(100.0 * value / max));
+         String title = (run.getValueInstant("startDate") == null ? "" : WHEN.format(run.getValueInstant("startDate")) + ": ") + bytes(run.getValueLong("bytesStored"));
+         html.append("<span title=\"").append(esc(title)).append("\" style=\"flex:1 1 0;min-width:2px;max-width:1.5rem;height:").append(pct).append("%;background:currentColor;opacity:.55;border-radius:2px 2px 0 0;\"></span>");
+      }
+      html.append("</div>");
+   }
+
+
+
+   /*******************************************************************************
+    ** Recent activity: the last runs, restores and verifications.
     *******************************************************************************/
    private static void recentActivity(StringBuilder html, BackupsHomeData data, String prefix)
    {
@@ -242,13 +284,13 @@ public final class BackupsHomeView
          data.recentRuns().stream().limit(6).forEach(run ->
          {
             Instant start = run.getValueInstant("startDate");
-            html.append("<li>Backup <a href=\"").append(esc(prefix + "/" + BackupsTables.RUN + "/" + run.getValue("id"))).append("\">").append(esc(run.getValueString("runKey")))
-               .append("</a>: ").append(esc(label(run.getValueString("status")))).append(start == null ? "" : ", " + WHEN.format(start)).append("</li>");
+            html.append("<li>Backup ").append(recordLink(prefix, BackupsTables.RUN, run.getValue("id"), run.getValueString("runKey")))
+               .append(": ").append(esc(label(run.getValueString("status")))).append(start == null ? "" : ", " + esc(WHEN.format(start))).append("</li>");
          });
-         data.restores().stream().limit(2).forEach(restore -> html.append("<li>Restore <a href=\"").append(esc(prefix + "/" + BackupsTables.RESTORE_RUN + "/" + restore.getValue("id")))
-            .append("\">#").append(esc(String.valueOf(restore.getValue("id")))).append("</a>: ").append(esc(label(restore.getValueString("status")))).append("</li>"));
-         data.verifications().stream().limit(2).forEach(v -> html.append("<li>Verify <a href=\"").append(esc(prefix + "/" + BackupsTables.VERIFICATION + "/" + v.getValue("id")))
-            .append("\">").append(esc(label(v.getValueString("level")))).append("</a>: ").append(esc(label(v.getValueString("status")))).append("</li>"));
+         data.restores().stream().limit(2).forEach(restore -> html.append("<li>Restore ").append(recordLink(prefix, BackupsTables.RESTORE_RUN, restore.getValue("id"), "#" + restore.getValue("id")))
+            .append(": ").append(esc(label(restore.getValueString("status")))).append("</li>"));
+         data.verifications().stream().limit(2).forEach(v -> html.append("<li>Verify ").append(recordLink(prefix, BackupsTables.VERIFICATION, v.getValue("id"), label(v.getValueString("level"))))
+            .append(": ").append(esc(label(v.getValueString("status")))).append("</li>"));
          html.append("</ul>");
       }
       link(html, prefix, BackupsTables.RUN, "All runs");
@@ -274,9 +316,10 @@ public final class BackupsHomeView
          any = true;
          html.append("<dl style=\"").append(LIST).append("\">");
          row(html, "Plan", plan.getValueString("name"));
-         row(html, "Kept", counts.entrySet().stream().map(e -> e.getValue() + " " + label(e.getKey()).toLowerCase()).reduce((a, b) -> a + ", " + b).orElse(""));
-         List<QRecord> runs = data.successfulRuns(plan.getValueInteger("id"));
-         Instant oldest = runs.isEmpty() ? null : runs.get(0).getValueInstant("startDate");
+         row(html, "Kept", counts.entrySet().stream().map(e -> e.getValue() + " " + ("UNCLASSIFIED".equals(e.getKey()) ? "not classified yet (no prune has run)" : label(e.getKey()).toLowerCase()))
+            .reduce((a, b) -> a + ", " + b).orElse(""));
+         List<QRecord> runs   = data.successfulRuns(plan.getValueInteger("id"));
+         Instant       oldest = runs.isEmpty() ? null : runs.get(0).getValueInstant("startDate");
          row(html, "Oldest restorable", oldest == null ? "Not known" : WHEN.format(oldest));
          html.append("</dl>");
       }
@@ -290,17 +333,17 @@ public final class BackupsHomeView
 
 
    /*******************************************************************************
-    **
+    ** Open a card.
     *******************************************************************************/
    private static void open(StringBuilder html, String title)
    {
-      html.append("<section style=\"").append(CARD).append("\"><h3 style=\"margin:0 0 .75rem;font-size:1rem;\">").append(esc(title)).append("</h3>");
+      html.append("<section style=\"").append(CARD).append("\"><h3 style=\"margin:0 0 .75rem;font-size:1rem;font-weight:600;\">").append(esc(title)).append("</h3>");
    }
 
 
 
    /*******************************************************************************
-    **
+    ** Close a card.
     *******************************************************************************/
    private static void close(StringBuilder html)
    {
@@ -310,17 +353,17 @@ public final class BackupsHomeView
 
 
    /*******************************************************************************
-    **
+    ** A muted empty-state line.
     *******************************************************************************/
    private static void empty(StringBuilder html, String text)
    {
-      html.append("<p style=\"").append(MUTED).append("margin:0 0 .75rem;\">").append(esc(text)).append("</p>");
+      html.append("<p style=\"").append(MUTED).append("margin:0 0 .75rem;font-size:.9rem;\">").append(esc(text)).append("</p>");
    }
 
 
 
    /*******************************************************************************
-    **
+    ** A name and value pair in a definition list.
     *******************************************************************************/
    private static void row(StringBuilder html, String name, String value)
    {
@@ -330,17 +373,53 @@ public final class BackupsHomeView
 
 
    /*******************************************************************************
-    **
+    ** A small label over its value (a plan row cell).
     *******************************************************************************/
-   private static void link(StringBuilder html, String prefix, String table, String text)
+   private static void cell(StringBuilder html, String name, String value)
    {
-      html.append("<a href=\"").append(esc(prefix + "/" + table)).append("\">").append(esc(text)).append("</a>");
+      html.append("<div style=\"min-width:0;\"><div style=\"").append(MUTED).append("font-size:.75rem;\">").append(esc(name)).append("</div><div>").append(esc(value)).append("</div></div>");
    }
 
 
 
    /*******************************************************************************
-    ** SOME_STATUS to Some status.
+    ** A status chip: a colored dot (the only color) and the status label.
+    *******************************************************************************/
+   private static String chip(String status)
+   {
+      String color = switch(status == null ? "UNKNOWN" : status)
+      {
+         case "OK" -> "#16a34a";
+         case "LATE", "UNVERIFIED", "UNKNOWN" -> "#d97706";
+         default -> "#dc2626";
+      };
+      return ("<span style=\"" + CHIP + "\"><span aria-hidden=\"true\" style=\"width:.55rem;height:.55rem;border-radius:50%;background:" + color + ";display:inline-block;\"></span>" + esc(label(status)) + "</span>");
+   }
+
+
+
+   /*******************************************************************************
+    ** A link to the table list at the end of a card.
+    *******************************************************************************/
+   private static void link(StringBuilder html, String prefix, String table, String text)
+   {
+      html.append("<a style=\"").append(LINK).append("font-size:.9rem;\" href=\"").append(esc(prefix + "/" + table)).append("\">").append(esc(text)).append("</a>");
+   }
+
+
+
+   /*******************************************************************************
+    ** A link to a record.
+    *******************************************************************************/
+   private static String recordLink(String prefix, String table, Object id, String text)
+   {
+      return ("<a style=\"" + LINK + "\" href=\"" + esc(prefix + "/" + table + "/" + id) + "\">" + esc(text == null ? String.valueOf(id) : text) + "</a>");
+   }
+
+
+
+   /*******************************************************************************
+    ** SOME_STATUS to Some status (OK stays OK).
     *******************************************************************************/
    static String label(String status)
    {
@@ -348,6 +427,10 @@ public final class BackupsHomeView
       {
          return ("Unknown");
       }
+      if("OK".equals(status))
+      {
+         return ("OK");
+      }
       String s = status.replace('_', ' ').toLowerCase();
       return (Character.toUpperCase(s.charAt(0)) + s.substring(1));
    }
@@ -355,7 +438,32 @@ public final class BackupsHomeView
 
 
    /*******************************************************************************
-    **
+    ** An Integer as a Long, null-safe.
+    *******************************************************************************/
+   private static Long toLong(Integer value)
+   {
+      return (value == null ? null : value.longValue());
+   }
+
+
+
+   /*******************************************************************************
+    ** Milliseconds as a short duration.
+    *******************************************************************************/
+   static String duration(long millis)
+   {
+      if(millis < 1000)
+      {
+         return (millis + " ms");
+      }
+      long seconds = millis / 1000;
+      return (seconds < 60 ? seconds + " s" : seconds / 60 + " min " + seconds % 60 + " s");
+   }
+
+
+
+   /*******************************************************************************
+    ** Bytes as a short size.
     *******************************************************************************/
    static String bytes(Long value)
    {
@@ -363,7 +471,7 @@ public final class BackupsHomeView
       {
          return ("Not recorded");
       }
-      double v = value;
+      double   v     = value;
       String[] units = { "B", "KB", "MB", "GB", "TB" };
       int      i     = 0;
       while(v >= 1024 && i < units.length - 1)
@@ -377,7 +485,7 @@ public final class BackupsHomeView
 
 
    /*******************************************************************************
-    **
+    ** A number with separators.
     *******************************************************************************/
    static String number(Long value)
    {
@@ -387,7 +495,7 @@ public final class BackupsHomeView
 
 
    /*******************************************************************************
-    **
+    ** HTML-escape.
     *******************************************************************************/
    static String esc(String s)
    {

src/main/java/com/kingsrook/qbits/backups/codec/TableArtifactWriter.java

@@ -22,9 +22,11 @@ import java.io.IOException;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
+import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.function.Consumer;
 import java.util.zip.GZIPOutputStream;
 import com.kingsrook.qbits.backups.storage.BackupStorage;
 import com.kingsrook.qbits.backups.storage.WriteOptions;
@@ -65,6 +67,10 @@ public class TableArtifactWriter implements Closeable
    private String              firstKeyOfTable;
    private String              lastKeyOfTable;
 
+   private Consumer<PartResult> partListener;
+   private String               objectLockMode;
+   private Instant              objectLockUntil;
+
 
 
    /***************************************************************************
@@ -122,6 +128,63 @@ public class TableArtifactWriter implements Closeable
 
 
 
+   /*******************************************************************************
+    ** Be told about each part as it is committed to storage (so the catalog records
+    ** a part the moment it is complete, which is what a resume starts from).
+    *******************************************************************************/
+   public TableArtifactWriter withPartListener(Consumer<PartResult> partListener)
+   {
+      this.partListener = partListener;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Apply an object lock to every part written.
+    *******************************************************************************/
+   public TableArtifactWriter withObjectLock(String mode, Instant until)
+   {
+      this.objectLockMode = mode;
+      this.objectLockUntil = until;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Continue after parts that were already written (resume): numbering carries on
+    ** after them and the table totals include them.  The caller then feeds the rows
+    ** hash with seedRowsHash.
+    *******************************************************************************/
+   public void resumeFrom(List<PartResult> priorParts)
+   {
+      if(priorParts.isEmpty())
+      {
+         return;
+      }
+      parts.addAll(priorParts);
+      partNumber = priorParts.get(priorParts.size() - 1).partNumber();
+      for(PartResult part : priorParts)
+      {
+         totalRows += part.rows();
+      }
+      firstKeyOfTable = priorParts.get(0).firstKey();
+      lastKeyOfTable = priorParts.get(priorParts.size() - 1).lastKey();
+   }
+
+
+
+   /*******************************************************************************
+    ** Feed already-written canonical JSONL bytes into the rows hash (resume).
+    *******************************************************************************/
+   public void seedRowsHash(byte[] bytes, int offset, int length)
+   {
+      rowsDigest.update(bytes, offset, length);
+   }
+
+
+
    /*******************************************************************************
     ** Write one record.
     *******************************************************************************/
@@ -245,6 +308,10 @@ public class TableArtifactWriter implements Closeable
       WriteOptions options = WriteOptions.create().withContentType("application/octet-stream")
          .withMetadata("qqq-backup-run", runKey)
          .withMetadata("qqq-backup-table", table.getName());
+      if(objectLockMode != null)
+      {
+         options.withObjectLock(objectLockMode, objectLockUntil);
+      }
       OutputStream toStorage = storage.openWrite(current.key, options);
       current.stored = new HashingOutputStream(toStorage);
       OutputStream transformed = transform.wrap(current.stored);
@@ -266,9 +333,14 @@ public class TableArtifactWriter implements Closeable
    private void closePart() throws IOException
    {
       current.raw.close();
-      parts.add(new PartResult(partNumber, current.key, current.rows, current.raw.getCount(), current.stored.getCount(),
+      PartResult result = new PartResult(partNumber, current.key, current.rows, current.raw.getCount(), current.stored.getCount(),
          current.raw.getSha256(), current.stored.getSha256(), current.firstKey, current.lastKey, transform.encrypts(),
-         csv ? "CSV" : "JSONL", gzip ? "GZIP" : "NONE"));
+         csv ? "CSV" : "JSONL", gzip ? "GZIP" : "NONE");
+      parts.add(result);
       current = null;
+      if(partListener != null)
+      {
+         partListener.accept(result);
+      }
    }
 }

src/main/java/com/kingsrook/qbits/backups/files/FileCopier.java

@@ -57,6 +57,8 @@ public class FileCopier
    private final FileStoreRegistry               registry;
    private final boolean                         referenceOnly;
    private final String                          runKey;
+   private String                                objectLockMode;
+   private java.time.Instant                     objectLockUntil;
 
 
 
@@ -119,6 +121,19 @@ public class FileCopier
 
 
 
+   /*******************************************************************************
+    ** Lock every object this copier writes (a server-side copy cannot set a lock,
+    ** so with a lock the bytes are streamed instead).
+    *******************************************************************************/
+   public FileCopier withObjectLock(String mode, java.time.Instant until)
+   {
+      this.objectLockMode = mode;
+      this.objectLockUntil = until;
+      return (this);
+   }
+
+
+
    /*******************************************************************************
     ** The content-addressed key for a hash: prefix/ab/cd/sha256.
     *******************************************************************************/
@@ -177,7 +192,7 @@ public class FileCopier
     *******************************************************************************/
    private String copy(BackupStorage source, FileReference ref, String key) throws IOException
    {
-      if(source == destination && !transform.encrypts())
+      if(source == destination && !transform.encrypts() && objectLockMode == null)
       {
          destination.copy(ref.key(), key);
          return (null);
@@ -185,6 +200,10 @@ public class FileCopier
 
       WriteOptions options = WriteOptions.create().withContentType(ref.contentType() == null ? "application/octet-stream" : ref.contentType())
          .withMetadata("qqq-backup-run", runKey);
+      if(objectLockMode != null)
+      {
+         options.withObjectLock(objectLockMode, objectLockUntil);
+      }
       try(InputStream in = source.openRead(ref.key());
           HashingOutputStream plain = new HashingOutputStream(transform.wrap(destination.openWrite(key, options))))
       {

src/main/java/com/kingsrook/qbits/backups/model/BackupsTables.java

@@ -23,8 +23,20 @@ import java.util.List;
 import java.util.Map;
 import com.kingsrook.qbits.backups.BackupsQBitConfig;
 import com.kingsrook.qbits.backups.permissions.BackupsPermissionChecker;
+import com.kingsrook.qbits.backups.plan.BackupPlanTableCustomizer;
+import com.kingsrook.qbits.backups.schedule.BackupPlanSyncToScheduledJobProcess;
+import com.kingsrook.qbits.backups.schedule.BackupPlanVerifySyncToScheduledJobProcess;
+import com.kingsrook.qqq.backend.core.actions.customizers.MultiCustomizer;
+import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReferenceWithProperties;
+import com.kingsrook.qqq.backend.core.scheduler.processes.BaseSyncToScheduledJobTableCustomizer;
 import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.actions.dashboard.widgets.ChildRecordListRenderer;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
+import com.kingsrook.qqq.backend.core.model.metadata.dashboard.QWidgetMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QFieldSection;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.Tier;
 import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
 import com.kingsrook.qqq.backend.core.model.metadata.joins.JoinOn;
 import com.kingsrook.qqq.backend.core.model.metadata.joins.JoinType;
@@ -63,7 +75,7 @@ public final class BackupsTables
    public static final List<String> PROCESS_WRITTEN = List.of(RUN, ARTIFACT, VERIFICATION, RESTORE_RUN, RESTORE_RUN_TABLE, FILE_OBJECT, ID_MAP);
 
    /** The tables shown in the nav (sub-tables appear as relations on their parents). */
-   public static final List<String> NAV_TABLES = List.of(PLAN, RUN, RESTORE_RUN, VERIFICATION, FILE_OBJECT);
+   public static final List<String> NAV_TABLES = List.of(PLAN, RUN, RESTORE_RUN, VERIFICATION);
 
    public static final List<String> ALL = List.of(PLAN, PLAN_SCOPE, RUN, ARTIFACT, VERIFICATION, RESTORE_RUN, RESTORE_RUN_TABLE, FILE_OBJECT, ID_MAP);
 
@@ -91,14 +103,14 @@ public final class BackupsTables
       ENUMS.put("lock_mode", List.of("NONE", "GOVERNANCE", "COMPLIANCE"));
       ENUMS.put("sla_status", List.of("OK", "LATE", "FAILING", "MISSING_IN_STORAGE", "UNVERIFIED", "ABANDONED", "OWNER_INVALID", "UNKNOWN"));
       ENUMS.put("run_trigger", List.of("SCHEDULED", "MANUAL", "PRE_RESTORE", "RESUME", "IMPORTED"));
-      ENUMS.put("run_status", List.of("QUEUED", "RUNNING", "SUCCEEDED", "SUCCEEDED_WITH_WARNINGS", "FAILED", "TIMED_OUT", "CANCELLED", "ABANDONED"));
+      ENUMS.put("run_status", List.of("QUEUED", "RUNNING", "SUCCEEDED", "SUCCEEDED_WITH_WARNINGS", "FAILED", "TIMED_OUT", "CANCELLING", "CANCELLED", "ABANDONED"));
       ENUMS.put("retention_class", List.of("DAILY", "WEEKLY", "MONTHLY", "YEARLY", "NONE"));
       ENUMS.put("verification_status", List.of("UNVERIFIED", "STORAGE_OK", "CONTENT_OK", "TEST_RESTORE_OK", "FAILED"));
       ENUMS.put("artifact_type", List.of("TABLE_DATA", "TABLE_BLOBS", "FILE_OBJECTS_INDEX", "NATIVE_DUMP", "MANIFEST", "CHECKSUMS"));
       ENUMS.put("artifact_status", List.of("WRITING", "COMPLETE", "FAILED"));
       ENUMS.put("verify_level", List.of("STORAGE", "CONTENT", "TEST_RESTORE"));
       ENUMS.put("verify_result", List.of("PASSED", "PASSED_WITH_WARNINGS", "FAILED", "RUNNING", "ERROR"));
-      ENUMS.put("restore_status", List.of("PREVIEWED", "QUEUED", "RUNNING", "SUCCEEDED", "SUCCEEDED_WITH_WARNINGS", "FAILED", "CANCELLED"));
+      ENUMS.put("restore_status", List.of("PREVIEWED", "QUEUED", "RUNNING", "SUCCEEDED", "SUCCEEDED_WITH_WARNINGS", "FAILED", "CANCELLING", "CANCELLED"));
       ENUMS.put("restore_mode", List.of("INSERT_NEW", "UPSERT", "REPLACE"));
       ENUMS.put("id_strategy", List.of("PRESERVE", "REMAP_NEW_IDS"));
       ENUMS.put("schema_status", List.of("MATCH", "COMPATIBLE", "MISMATCH"));
@@ -203,7 +215,111 @@ public final class BackupsTables
    public static List<QTableMetaData> tables(BackupsQBitConfig config)
    {
       return (List.of(plan(config), planScope(config), run(config), artifact(config), verification(config),
-         restoreRun(config), restoreRunTable(config), fileObject(config), idMap(config)));
+         restoreRun(config), restoreRunTable(config), fileObject(config), idMap(config)).stream().map(BackupsTables::withSections).toList());
+   }
+
+
+
+   /*******************************************************************************
+    ** Group a table's fields into record-screen sections (the identity and status
+    ** first, then one section per concern, then the system fields), so a record reads
+    ** as a few short groups instead of one long list.  Fields not named here land in
+    ** the last section.
+    *******************************************************************************/
+   static QTableMetaData withSections(QTableMetaData table)
+   {
+      List<Object[]> layout = switch(table.getName())
+      {
+         case PLAN -> List.of(
+            new Object[] { "plan", "Plan", "backup", Tier.T1, "name", "description", "planKey", "isActive", "scopeType", "slaStatus", "lastSuccessDate", "lastVerifiedDate" },
+            new Object[] { "schedule", "Schedule and SLA", "schedule", Tier.T2, "cronExpression", "cronTimeZoneId", "scheduledJobId", "slaHours", "verifySlaDays", "verifyCronExpression", "verifyLevel" },
+            new Object[] { "where", "Scope and storage", "storage", Tier.T2, "includeRoles", "storageBucketAlias", "pruneBucketAlias", "prefix", "critical", "consistency", "useReadReplica", "incrementalMode", "nativeDump" },
+            new Object[] { "format", "Format and safety", "shield", Tier.T2, "format", "compression", "encryptionMode", "encryptionKeyRef", "sensitiveFieldPolicy", "heavyFieldPolicy", "fileObjectPolicy", "fileMissingPolicy" },
+            new Object[] { "retention", "Retention and lock", "lock_clock", Tier.T2, "retainDaily", "retainWeekly", "retainMonthly", "retainYearly", "retainMinimum", "objectLockMode", "objectLockDays" },
+            new Object[] { "system", "Limits and history", "history", Tier.T3, "maxDurationMinutes", "pageSize", "targetPartBytes", "ownerUserId", "lastRunId", "lastSuccessRunId", "lastVerifiedRunId", "id", "createDate", "modifyDate" });
+         case RUN -> List.of(
+            new Object[] { "run", "Run", "history", Tier.T1, "runKey", "backupPlanId", "status", "trigger", "startDate", "endDate", "durationMs", "errorMessage" },
+            new Object[] { "volume", "Rows and size", "data_usage", Tier.T2, "tablesPlanned", "tablesDone", "rowsTotal", "bytesRaw", "bytesStored", "consistencyAchieved", "warningCount" },
+            new Object[] { "files", "Files", "attach_file", Tier.T2, "filesReferenced", "filesCopied", "filesDeduplicated", "filesMissing", "fileBytesCopied", "fileBytesReferenced" },
+            new Object[] { "storage", "Storage and integrity", "storage", Tier.T2, "storageBucketAlias", "runPrefix", "manifestKey", "manifestSha256", "snapshotId", "schemaFingerprint", "securityScope", "encryptionKeyRef", "encryptionKeyFingerprint", "verificationStatus" },
+            new Object[] { "retention", "Retention", "lock_clock", Tier.T2, "retentionClass", "lockedUntil", "prunedDate", "prunedByUserId" },
+            new Object[] { "system", "System", "settings", Tier.T3, "requestedByUserId", "heartbeatDate", "qqqVersion", "appVersion", "instanceName", "errorDetail", "id", "createDate", "modifyDate" });
+         case VERIFICATION -> List.of(
+            new Object[] { "result", "Result", "verified", Tier.T1, "backupRunId", "level", "status", "tablesChecked", "rowsChecked", "mismatches", "scratchTarget", "startDate", "endDate" },
+            new Object[] { "system", "Detail", "settings", Tier.T3, "trigger", "requestedByUserId", "detailJson", "id", "createDate", "modifyDate" });
+         case RESTORE_RUN -> List.of(
+            new Object[] { "restore", "Restore", "restore", Tier.T1, "sourceBackupRunId", "status", "mode", "idStrategy", "matchOn", "targetInstanceName", "startDate", "endDate", "errorMessage" },
+            new Object[] { "rows", "Rows", "table_rows", Tier.T2, "rowsInserted", "rowsUpdated", "rowsUnchanged", "rowsDeleted", "rowsSkipped", "rowsErrored", "sequencesResynced" },
+            new Object[] { "files", "Files", "attach_file", Tier.T2, "filesRestored", "filesAlreadyPresent", "filesConflicting", "filesMissingFromBackup", "fileBytesRestored" },
+            new Object[] { "system", "Safety and system", "settings", Tier.T3, "preRestoreBackupRunId", "scopeSummary", "automationsSuppressed", "esbSuppressed", "confirmationText", "requestedByUserId", "id", "createDate", "modifyDate" });
+         default -> List.of();
+      };
+      if(layout.isEmpty())
+      {
+         return (table);
+      }
+      java.util.Set<String> placed = new java.util.HashSet<>();
+      for(int i = 0; i < layout.size(); i++)
+      {
+         Object[]     spec   = layout.get(i);
+         List<String> fields = new ArrayList<>();
+         for(int f = 4; f < spec.length; f++)
+         {
+            fields.add((String) spec[f]);
+         }
+         if(i == layout.size() - 1)
+         {
+            table.getFields().keySet().stream().filter(name -> !placed.contains(name) && !fields.contains(name)).forEach(fields::add);
+         }
+         placed.addAll(fields);
+         table.withSection(new QFieldSection((String) spec[0], (String) spec[1], new QIcon((String) spec[2]), (Tier) spec[3], fields));
+      }
+      for(ChildList childList : CHILD_LISTS)
+      {
+         if(childList.parent().equals(table.getName()))
+         {
+            table.withSection(new QFieldSection(childList.key(), new QIcon(childList.icon()), Tier.T2).withLabel(childList.label()).withWidgetName(childList.joinName()));
+         }
+      }
+      return (table);
+   }
+
+
+
+   /** A child list on a parent record: its join, the columns shown, and the order. */
+   record ChildList(String parent, String joinName, String key, String label, String icon, int maxRows, boolean canAdd, String orderField, List<String> columns)
+   {
+   }
+
+   /** The relations shown on the record screens (sub-tables are child lists here, not in the nav). */
+   static final List<ChildList> CHILD_LISTS = List.of(
+      new ChildList(PLAN, "backupsPlanScopeJoinPlan", "scopeRules", "Scope rules", "rule", 50, true, "id", List.of("action", "selectorType", "selectorValue", "includeAssociatedChildren", "fieldExclusions")),
+      new ChildList(PLAN, "backupsRunJoinPlan", "runs", "Runs", "history", 10, false, "id", List.of("runKey", "status", "startDate", "durationMs", "rowsTotal", "bytesStored")),
+      new ChildList(RUN, "backupsArtifactJoinRun", "artifacts", "Artifacts", "inventory_2", 100, false, "id", List.of("tableName", "artifactType", "partNumber", "rowCount", "bytesStored", "status")),
+      new ChildList(RUN, "backupsVerificationJoinRun", "verifications", "Verifications", "verified", 20, false, "id", List.of("level", "status", "tablesChecked", "rowsChecked", "mismatches", "startDate")),
+      new ChildList(RUN, "backupsRestoreRunJoinRun", "restores", "Restores", "restore", 20, false, "id", List.of("status", "mode", "startDate", "rowsInserted", "rowsUpdated", "rowsErrored")),
+      new ChildList(RESTORE_RUN, "backupsRestoreRunTableJoinRestoreRun", "restoreTables", "Tables restored", "table_rows", 100, false, "id",
+         List.of("tableName", "targetTableName", "rowsInserted", "rowsUpdated", "rowsUnchanged", "rowsErrored", "schemaStatus")));
+
+
+
+   /*******************************************************************************
+    ** The child-list widgets for the record screens (see CHILD_LISTS).  Each shows a
+    ** few readable columns, newest first, and never scrolls sideways.
+    *******************************************************************************/
+   public static List<QWidgetMetaData> widgets()
+   {
+      Map<String, QJoinMetaData> byName = new LinkedHashMap<>();
+      joins().forEach(join -> byName.put(join.getName(), join));
+      List<QWidgetMetaData> rs = new ArrayList<>();
+      for(ChildList childList : CHILD_LISTS)
+      {
+         rs.add(ChildRecordListRenderer.widgetMetaDataBuilder(byName.get(childList.joinName())).withLabel(childList.label()).withMaxRows(childList.maxRows())
+            .withCanAddChildRecord(childList.canAdd()).withOnlyIncludeFieldNames(childList.columns())
+            .withOrderBys(List.of(new QFilterOrderBy(childList.orderField(), !PLAN.equals(childList.parent()) || childList.canAdd())))
+            .getWidgetMetaData().withPermissionRules(permissionRules()));
+      }
+      return (rs);
    }
 
 
@@ -286,6 +402,34 @@ public final class BackupsTables
     ** backups_plan: what, how, where, when and how long (spec 4.1).
     *******************************************************************************/
    static QTableMetaData plan(BackupsQBitConfig config)
+   {
+      QTableMetaData table = planBase(config);
+
+      ////////////////////////////////////////////////////////////////////////////////////////
+      // save validation (spec 4.1, 11.1), and the plan -> scheduled job sync when the qbit  //
+      // has a scheduler (spec 9.1): one job per active plan, and one for the verify cron    //
+      ////////////////////////////////////////////////////////////////////////////////////////
+      QCodeReference validation = new QCodeReference(BackupPlanTableCustomizer.class);
+      table.withCustomizer(TableCustomizers.PRE_INSERT_RECORD, validation);
+      table.withCustomizer(TableCustomizers.PRE_UPDATE_RECORD, validation);
+      if(StringUtils.hasContent(config.getSchedulerName()))
+      {
+         QCodeReferenceWithProperties sync = MultiCustomizer.of(
+            BaseSyncToScheduledJobTableCustomizer.makeCodeReference(table, new BackupPlanSyncToScheduledJobProcess()),
+            BaseSyncToScheduledJobTableCustomizer.makeCodeReference(table, new BackupPlanVerifySyncToScheduledJobProcess()));
+         table.withCustomizer(TableCustomizers.POST_INSERT_RECORD, sync);
+         table.withCustomizer(TableCustomizers.POST_UPDATE_RECORD, sync);
+         table.withCustomizer(TableCustomizers.POST_DELETE_RECORD, sync);
+      }
+      return (table);
+   }
+
+
+
+   /*******************************************************************************
+    ** The plan table before the customizers.
+    *******************************************************************************/
+   private static QTableMetaData planBase(BackupsQBitConfig config)
    {
       return (base(config, PLAN, "Backup Plan", "backup")
          .withRecordLabelFormatAndFields("%s", "name")
@@ -369,8 +513,6 @@ public final class BackupsTables
       return (base(config, RUN, "Backup Run", "history")
          .withRecordLabelFormatAndFields("%s", "runKey")
          .withUniqueKey(new UniqueKey("runKey"))
-         .withAssociation(new Association().withName("artifacts").withAssociatedTableName(ARTIFACT).withJoinName("backupsArtifactJoinRun"))
-         .withAssociation(new Association().withName("verifications").withAssociatedTableName(VERIFICATION).withJoinName("backupsVerificationJoinRun"))
          .withField(ro("backupPlanId", QFieldType.INTEGER).withPossibleValueSourceName(PLAN))
          .withField(ro("runKey", QFieldType.STRING))
          .withField(roEnum("trigger", "run_trigger"))
@@ -476,7 +618,6 @@ public final class BackupsTables
    {
       return (base(config, RESTORE_RUN, "Backup Restore", "restore")
          .withRecordLabelFormatAndFields("Restore %s", "id")
-         .withAssociation(new Association().withName("tables").withAssociatedTableName(RESTORE_RUN_TABLE).withJoinName("backupsRestoreRunTableJoinRestoreRun"))
          .withField(ro("sourceBackupRunId", QFieldType.INTEGER).withPossibleValueSourceName(RUN))
          .withField(ro("requestedByUserId", QFieldType.STRING))
          .withField(ro("startDate", QFieldType.DATE_TIME))

src/main/java/com/kingsrook/qbits/backups/plan/BackupPlanTableCustomizer.java

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.plan;
+
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import com.kingsrook.qbits.backups.model.BackupsTables;
+import com.kingsrook.qbits.backups.run.BackupCatalog;
+import com.kingsrook.qbits.backups.run.BackupRunFactory;
+import com.kingsrook.qbits.backups.scope.PlanSettings;
+import com.kingsrook.qbits.backups.scope.ScopeResolver;
+import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.AbstractActionInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage;
+import com.kingsrook.qqq.backend.core.model.statusmessages.QWarningMessage;
+
+
+/*******************************************************************************
+ ** Plan save validation (pre insert and pre update of backups_plan).  See
+ ** BackupPlanValidator for the rules.  An update that touches none of the plan
+ ** configuration (the processes write the last run and SLA fields) is not
+ ** validated again.
+ **
+ ** The saving user becomes the plan owner when none is given.
+ *******************************************************************************/
+public class BackupPlanTableCustomizer implements TableCustomizerInterface
+{
+   private static final List<String> CONFIGURATION_FIELDS = List.of("planKey", "isActive", "scopeType", "includeRoles", "storageBucketAlias", "pruneBucketAlias",
+      "format", "compression", "encryptionMode", "encryptionKeyRef", "sensitiveFieldPolicy", "incrementalMode", "nativeDump", "cronExpression", "cronTimeZoneId",
+      "verifyCronExpression", "retainDaily", "retainWeekly", "retainMonthly", "retainYearly", "retainMinimum", "objectLockMode", "objectLockDays",
+      "slaHours", "pageSize", "ownerUserId");
+
+
+
+   /*******************************************************************************
+    ** {@inheritDoc}
+    *******************************************************************************/
+   @Override
+   public List<QRecord> preInsertOrUpdate(AbstractActionInput input, List<QRecord> records, boolean isPreview, Optional<List<QRecord>> oldRecordList) throws QException
+   {
+      Map<Serializable, QRecord> oldById = oldRecordList.isPresent() ? oldRecordListToMap("id", oldRecordList).orElse(Map.of()) : Map.of();
+      BackupCatalog              catalog = new BackupCatalog();
+
+      for(QRecord record : records)
+      {
+         QRecord old = oldRecordList.isPresent() ? oldById.get(record.getValue("id")) : null;
+         if(old != null && !touchesConfiguration(record))
+         {
+            continue;
+         }
+
+         QRecord merged = new QRecord();
+         if(old != null)
+         {
+            merged.getValues().putAll(old.getValues());
+         }
+         merged.getValues().putAll(record.getValues());
+
+         if(merged.getValue("ownerUserId") == null && Boolean.TRUE.equals(merged.getValueBoolean("isActive")))
+         {
+            QSession session = QContext.getQSession();
+            String   owner   = session == null || session.getUser() == null ? null : session.getUser().getIdReference();
+            if(owner != null)
+            {
+               record.setValue("ownerUserId", owner);
+               merged.setValue("ownerUserId", owner);
+            }
+         }
+
+         boolean hasRun = old != null && !catalog.where(BackupsTables.RUN, "backupPlanId", old.getValueInteger("id")).isEmpty();
+         BackupPlanValidator.Result result = BackupPlanValidator.validate(PlanSettings.of(merged), old == null ? null : old.getValueString("planKey"), hasRun, scopeRules(record, old));
+         for(String error : result.errors())
+         {
+            record.addError(new BadInputStatusMessage(error));
+         }
+         for(String warning : result.warnings())
+         {
+            record.addWarning(new QWarningMessage(warning));
+         }
+      }
+      return (records);
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static boolean touchesConfiguration(QRecord record)
+   {
+      for(String field : CONFIGURATION_FIELDS)
+      {
+         if(record.getValues().containsKey(field))
+         {
+            return (true);
+         }
+      }
+      return (false);
+   }
+
+
+
+   /*******************************************************************************
+    ** The scope rules to resolve: the ones sent with the plan (associated rows), or
+    ** the saved ones.
+    *******************************************************************************/
+   private static List<ScopeResolver.ScopeRule> scopeRules(QRecord record, QRecord old) throws QException
+   {
+      List<QRecord> sent = record.getAssociatedRecords() == null ? null : record.getAssociatedRecords().get("scopes");
+      if(sent != null)
+      {
+         List<ScopeResolver.ScopeRule> rules = new ArrayList<>();
+         for(QRecord row : sent)
+         {
+            rules.add(new ScopeResolver.ScopeRule(!"EXCLUDE".equals(row.getValueString("action")), row.getValueString("selectorType"), row.getValueString("selectorValue"),
+               row.getValueString("filterJson"), !Boolean.FALSE.equals(row.getValueBoolean("includeAssociatedChildren")), row.getValueString("fieldExclusions")));
+         }
+         return (rules);
+      }
+      return (old == null ? List.of() : BackupRunFactory.scopeRules(old.getValueInteger("id")));
+   }
+}

src/main/java/com/kingsrook/qbits/backups/plan/BackupPlanValidator.java

@@ -0,0 +1,354 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.plan;
+
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.regex.Pattern;
+import com.kingsrook.qbits.backups.BackupsRuntime;
+import com.kingsrook.qbits.backups.permissions.BackupsPermissionChecker;
+import com.kingsrook.qbits.backups.permissions.BackupsPermissions;
+import com.kingsrook.qbits.backups.scope.PlanSettings;
+import com.kingsrook.qbits.backups.scope.ScopeResolver;
+import com.kingsrook.qbits.backups.storage.BackupStorage;
+import com.kingsrook.qqq.backend.core.actions.permissions.PermissionsHelper;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.actions.permissions.TablePermissionSubType;
+
+
+/*******************************************************************************
+ ** The rules a backup plan must satisfy when it is saved or activated (spec 4.1,
+ ** 6.2, 11.1, 11.3): the refusals are clear messages, not run-time failures.
+ **
+ ** - the saving user has backups.admin and may read every table the plan resolves
+ **   to (a scheduled run is full-fidelity, so whoever activates the plan must be
+ **   allowed to read all of it)
+ ** - an object lock mode is checked against what the storage bucket can do
+ ** - INCLUDE_ENCRYPTED_ONLY needs encryption
+ ** - formats, compressions and options that are not built yet are refused
+ ** - the plan key is a slug, and cannot change after the first run (it is in every
+ **   object key)
+ *******************************************************************************/
+public final class BackupPlanValidator
+{
+   private static final Pattern PLAN_KEY = Pattern.compile("[a-z0-9][a-z0-9_-]{0,59}");
+
+   private static final Set<String> FORMATS      = Set.of("JSONL", "CSV", "JSONL_AND_CSV");
+   private static final Set<String> COMPRESSIONS = Set.of("GZIP", "NONE");
+   private static final Set<String> LOCK_MODES   = Set.of("NONE", "GOVERNANCE", "COMPLIANCE");
+
+   /** Replace with the real check when client-side encryption ships (spec 11.2). */
+   private static volatile boolean encryptionAvailable = false;
+
+
+
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private BackupPlanValidator()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** Tell the validator that client-side encryption is available (the encryption
+    ** work calls this when it installs a cipher).
+    *******************************************************************************/
+   public static void setEncryptionAvailable(boolean available)
+   {
+      encryptionAvailable = available;
+   }
+
+
+
+   /*******************************************************************************
+    ** The outcome of a validation.
+    *******************************************************************************/
+   public record Result(List<String> errors, List<String> warnings)
+   {
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      public boolean ok()
+      {
+         return (errors.isEmpty());
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Validate the complete settings of a plan.
+    **
+    ** @param settings the plan values as they will be after the save
+    ** @param oldPlanKey the plan key before the save, null for a new plan
+    ** @param hasRun whether the plan already has a run
+    ** @param rules the scope rules of the plan
+    *******************************************************************************/
+   public static Result validate(PlanSettings settings, String oldPlanKey, boolean hasRun, List<ScopeResolver.ScopeRule> rules)
+   {
+      List<String> errors   = new ArrayList<>();
+      List<String> warnings = new ArrayList<>();
+
+      validateKey(settings, oldPlanKey, hasRun, errors);
+      validateFormats(settings, errors);
+      validateEncryption(settings, errors);
+      validateStorage(settings, errors, warnings);
+      validateSchedule(settings, errors);
+      validateNumbers(settings, errors);
+
+      if(settings.bool("isActive", true))
+      {
+         validateOwner(settings, rules, errors);
+      }
+
+      return (new Result(errors, warnings));
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static void validateKey(PlanSettings settings, String oldPlanKey, boolean hasRun, List<String> errors)
+   {
+      String planKey = settings.string("planKey", null);
+      if(planKey == null || !PLAN_KEY.matcher(planKey).matches())
+      {
+         errors.add("The plan key must be lowercase letters, digits, dash or underscore (up to 60 characters): it is part of every object key.");
+      }
+      if(oldPlanKey != null && planKey != null && !oldPlanKey.equals(planKey) && hasRun)
+      {
+         errors.add("The plan key cannot change after the plan has run: it is part of the object keys of its backups. Create a new plan instead.");
+      }
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static void validateFormats(PlanSettings settings, List<String> errors)
+   {
+      String format = settings.string("format", "JSONL");
+      if("PARQUET".equals(format))
+      {
+         errors.add("The PARQUET format is not available yet. Use JSONL, CSV or JSONL_AND_CSV.");
+      }
+      else if(!FORMATS.contains(format))
+      {
+         errors.add("Unknown format " + format + ". Use JSONL, CSV or JSONL_AND_CSV.");
+      }
+
+      String compression = settings.string("compression", "GZIP");
+      if("ZSTD".equals(compression))
+      {
+         errors.add("ZSTD compression is not available yet. Use GZIP or NONE.");
+      }
+      else if(!COMPRESSIONS.contains(compression))
+      {
+         errors.add("Unknown compression " + compression + ". Use GZIP or NONE.");
+      }
+
+      if("CHANGED_SINCE_LAST".equals(settings.string("incrementalMode", "FULL")))
+      {
+         errors.add("Incremental backups are not available yet. Use FULL.");
+      }
+      if(!"NONE".equals(settings.string("nativeDump", "NONE")))
+      {
+         errors.add("Native database dumps are not available yet. Use NONE.");
+      }
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static void validateEncryption(PlanSettings settings, List<String> errors)
+   {
+      String mode   = settings.string("encryptionMode", "NONE");
+      String policy = settings.string("sensitiveFieldPolicy", "EXCLUDE");
+      if("CLIENT_AES256_GCM".equals(mode))
+      {
+         if(!encryptionAvailable)
+         {
+            errors.add("Client-side encryption is not available in this version. Use NONE (and keep the sensitive field policy at EXCLUDE).");
+         }
+         else if(settings.string("encryptionKeyRef", null) == null)
+         {
+            errors.add("Client-side encryption needs an encryption key reference (the name of the key, never the key itself).");
+         }
+      }
+      if("INCLUDE_ENCRYPTED_ONLY".equals(policy) && !"CLIENT_AES256_GCM".equals(mode))
+      {
+         errors.add("The sensitive field policy INCLUDE_ENCRYPTED_ONLY requires client-side encryption (CLIENT_AES256_GCM).");
+      }
+      if("INCLUDE".equals(policy) && !BackupsPermissionChecker.has(BackupsPermissions.ADMIN))
+      {
+         errors.add("Including sensitive fields in the clear needs the backups.admin permission.");
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** The destination: the bucket exists, and an object lock is something the bucket can do.
+    *******************************************************************************/
+   private static void validateStorage(PlanSettings settings, List<String> errors, List<String> warnings)
+   {
+      String alias = settings.string("storageBucketAlias", null);
+      if(alias == null)
+      {
+         errors.add("Choose a storage bucket for the backups.");
+         return;
+      }
+
+      BackupStorage storage = null;
+      try
+      {
+         storage = BackupsRuntime.storageFor(alias);
+      }
+      catch(RuntimeException e)
+      {
+         errors.add("The storage bucket " + alias + " cannot be used: " + e.getMessage());
+      }
+
+      String lockMode = settings.string("objectLockMode", "NONE");
+      if(!LOCK_MODES.contains(lockMode))
+      {
+         errors.add("Unknown object lock mode " + lockMode + ". Use NONE, GOVERNANCE or COMPLIANCE.");
+      }
+      else if(!"NONE".equals(lockMode))
+      {
+         Integer days = settings.integer("objectLockDays", null);
+         if(days == null || days < 1)
+         {
+            errors.add("Object lock needs a number of days to keep each object locked (objectLockDays).");
+         }
+         if(storage != null && !storage.capabilities().contains(BackupStorage.StorageCapability.OBJECT_LOCK))
+         {
+            errors.add("Object lock " + lockMode + " was asked for, but the storage bucket " + alias + " does not support object lock (its capabilities: "
+               + storage.capabilities() + "). Choose NONE, or a bucket whose Test connection found object lock supported.");
+         }
+      }
+
+      String pruneAlias = settings.string("pruneBucketAlias", null);
+      if(pruneAlias == null)
+      {
+         warnings.add("No prune bucket is set: Prune will use the backup bucket, so one key can both write and delete backups. Give the plan a separate delete-capable alias.");
+      }
+      else
+      {
+         try
+         {
+            BackupsRuntime.storageFor(pruneAlias);
+         }
+         catch(RuntimeException e)
+         {
+            errors.add("The prune bucket " + pruneAlias + " cannot be used: " + e.getMessage());
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static void validateSchedule(PlanSettings settings, List<String> errors)
+   {
+      for(String field : List.of("cronExpression", "verifyCronExpression"))
+      {
+         String cron = settings.string(field, null);
+         if(cron != null && !org.quartz.CronExpression.isValidExpression(cron))
+         {
+            errors.add("The " + field + " is not a valid cron expression: " + cron);
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static void validateNumbers(PlanSettings settings, List<String> errors)
+   {
+      for(String field : List.of("retainDaily", "retainWeekly", "retainMonthly", "retainYearly", "retainMinimum"))
+      {
+         Integer value = settings.integer(field, 0);
+         if(value < 0)
+         {
+            errors.add(field + " cannot be negative.");
+         }
+      }
+      if(settings.integer("slaHours", 26) < 1)
+      {
+         errors.add("slaHours must be at least 1.");
+      }
+      if(settings.integer("pageSize", 5000) < 1)
+      {
+         errors.add("pageSize must be at least 1.");
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Saving or activating a plan: the saving user must have backups.admin and read
+    ** access to every table the plan resolves to (spec 11.1).
+    *******************************************************************************/
+   private static void validateOwner(PlanSettings settings, List<ScopeResolver.ScopeRule> rules, List<String> errors)
+   {
+      if(!BackupsPermissionChecker.has(BackupsPermissions.ADMIN))
+      {
+         errors.add("Saving an active plan needs the backups.admin permission: the plan runs on its own with full access.");
+         return;
+      }
+
+      ScopeResolver.ResolvedScope scope;
+      try
+      {
+         scope = new ScopeResolver(QContext.getQInstance()).resolve(settings, rules);
+      }
+      catch(RuntimeException e)
+      {
+         errors.add("The plan scope cannot be resolved: " + e.getMessage());
+         return;
+      }
+
+      List<String> unreadable = new ArrayList<>();
+      for(ScopeResolver.ResolvedTable table : scope.tables())
+      {
+         if(!PermissionsHelper.hasTablePermission(new QueryInput(table.name()), table.name(), TablePermissionSubType.READ))
+         {
+            unreadable.add(table.name());
+         }
+      }
+      if(!unreadable.isEmpty())
+      {
+         errors.add("The plan covers tables that its owner is not allowed to read, so it could not back them up in full: "
+            + String.join(", ", unreadable.stream().limit(10).toList()) + (unreadable.size() > 10 ? " and " + (unreadable.size() - 10) + " more" : "")
+            + ". Remove them from the scope, or save the plan as a user who can read them.");
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/backups/plan/DestinationProbe.java

@@ -0,0 +1,260 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.plan;
+
+
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import com.kingsrook.qbits.backups.BackupsRuntime;
+import com.kingsrook.qbits.backups.prune.BackupPruner;
+import com.kingsrook.qbits.backups.storage.BackupStorage;
+import com.kingsrook.qbits.backups.storage.StoredObjectInfo;
+import com.kingsrook.qbits.backups.storage.WriteOptions;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+
+
+/*******************************************************************************
+ ** Test destination (spec 9): proves the plan destination works before the first
+ ** night depends on it.  Writes a small probe object, reads it back, lists it, and
+ ** deletes it through the prune alias (so the delete credential is exercised too).
+ **
+ ** On a bucket that is object-locked the probe is head only: a probe object could
+ ** not be deleted again, so nothing is written; the prefix is listed, and the
+ ** newest manifest is looked at with a head.  A capability report always comes with
+ ** it.
+ *******************************************************************************/
+public final class DestinationProbe
+{
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private DestinationProbe()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** One thing that was checked.
+    *******************************************************************************/
+   public record Check(String name, boolean ok, String detail)
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** The whole report.
+    *******************************************************************************/
+   public record Report(boolean headOnly, List<Check> checks, List<String> capabilities, List<String> warnings)
+   {
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      public boolean ok()
+      {
+         return (checks.stream().allMatch(Check::ok));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Probe the destination of a plan.
+    *******************************************************************************/
+   public static Report probe(QRecord plan)
+   {
+      List<Check>  checks   = new ArrayList<>();
+      List<String> warnings = new ArrayList<>();
+
+      String alias = plan.getValueString("storageBucketAlias");
+      BackupStorage storage;
+      try
+      {
+         storage = BackupsRuntime.storageFor(alias);
+      }
+      catch(RuntimeException e)
+      {
+         checks.add(new Check("Open the backup bucket " + alias, false, e.getMessage()));
+         return (new Report(false, checks, List.of(), warnings));
+      }
+      checks.add(new Check("Open the backup bucket " + alias, true, "found"));
+
+      BackupStorage deleteSide = storage;
+      String        pruneAlias = plan.getValueString("pruneBucketAlias");
+      if(pruneAlias == null || pruneAlias.isBlank())
+      {
+         warnings.add("No prune bucket is set, so one key both writes and deletes backups.");
+      }
+      else
+      {
+         try
+         {
+            deleteSide = BackupsRuntime.storageFor(pruneAlias);
+            checks.add(new Check("Open the prune bucket " + pruneAlias, true, "found"));
+         }
+         catch(RuntimeException e)
+         {
+            checks.add(new Check("Open the prune bucket " + pruneAlias, false, e.getMessage()));
+         }
+      }
+
+      List<String> capabilities = new ArrayList<>();
+      for(BackupStorage.StorageCapability capability : BackupStorage.StorageCapability.values())
+      {
+         capabilities.add(capability.name() + ": " + (storage.capabilities().contains(capability) ? "supported" : "not supported"));
+      }
+
+      String lockMode = plan.getValueString("objectLockMode");
+      boolean locked  = lockMode != null && !"NONE".equals(lockMode);
+      if(locked && !storage.capabilities().contains(BackupStorage.StorageCapability.OBJECT_LOCK))
+      {
+         checks.add(new Check("Object lock " + lockMode, false, "the plan asks for object lock, but this bucket does not support it"));
+      }
+
+      String prefix = BackupPruner.planPrefix(plan);
+      if(locked)
+      {
+         headOnly(plan, storage, prefix, checks);
+      }
+      else
+      {
+         writeReadListDelete(storage, deleteSide, prefix + ".qqq-probe/", checks);
+      }
+      return (new Report(locked, checks, capabilities, warnings));
+   }
+
+
+
+   /*******************************************************************************
+    ** The probe for a locked bucket: list the prefix, and head the newest manifest.
+    *******************************************************************************/
+   private static void headOnly(QRecord plan, BackupStorage storage, String prefix, List<Check> checks)
+   {
+      checks.add(attempt("Write and delete a probe object", () -> "skipped: object lock is on, a probe object could not be deleted again"));
+      checks.add(attempt("List the plan prefix", () ->
+      {
+         List<StoredObjectInfo> objects = storage.list(prefix);
+         return (objects.size() + " objects under " + prefix);
+      }));
+
+      Integer lastSuccess = plan.getValueInteger("lastSuccessRunId");
+      if(lastSuccess == null)
+      {
+         checks.add(new Check("Head the newest manifest", true, "there is no successful run yet"));
+         return;
+      }
+      checks.add(attempt("Head the newest manifest", () ->
+      {
+         QRecord run = new com.kingsrook.qbits.backups.run.BackupCatalog().get(com.kingsrook.qbits.backups.model.BackupsTables.RUN, lastSuccess).orElseThrow();
+         String  key = run.getValueString("manifestKey");
+         if(key == null || storage.head(key).isEmpty())
+         {
+            throw (new IllegalStateException("the manifest of the newest successful run is not in the bucket"));
+         }
+         return ("found " + key);
+      }));
+   }
+
+
+
+   /*******************************************************************************
+    ** Write, read back, list and delete a probe object.
+    *******************************************************************************/
+   private static void writeReadListDelete(BackupStorage storage, BackupStorage deleteSide, String probePrefix, List<Check> checks)
+   {
+      String key     = probePrefix + UUID.randomUUID();
+      byte[] content = ("qqq backups probe " + UUID.randomUUID()).getBytes(StandardCharsets.UTF_8);
+
+      Check write = attempt("Write a probe object", () ->
+      {
+         try(OutputStream out = storage.openWrite(key, WriteOptions.create().withContentType("text/plain")))
+         {
+            out.write(content);
+         }
+         return ("wrote " + content.length + " bytes");
+      });
+      checks.add(write);
+      if(!write.ok())
+      {
+         return;
+      }
+
+      checks.add(attempt("Read it back", () ->
+      {
+         try(InputStream in = storage.openRead(key))
+         {
+            if(!java.util.Arrays.equals(content, in.readAllBytes()))
+            {
+               throw (new IllegalStateException("the bytes read back are not the bytes written"));
+            }
+         }
+         return ("the same bytes came back");
+      }));
+
+      checks.add(attempt("List it", () ->
+      {
+         if(storage.list(probePrefix).stream().noneMatch(o -> o.key().equals(key)))
+         {
+            throw (new IllegalStateException("the probe object is not in the listing"));
+         }
+         return ("found in the listing");
+      }));
+
+      checks.add(attempt("Delete it", () ->
+      {
+         BackupStorage.DeleteResult result = deleteSide.deletePermanent(key);
+         if(result != BackupStorage.DeleteResult.DELETED)
+         {
+            throw (new IllegalStateException("the delete returned " + result));
+         }
+         return ("deleted");
+      }));
+   }
+
+
+
+   /***************************************************************************
+    **
+    ***************************************************************************/
+   @FunctionalInterface
+   private interface Attempt
+   {
+      String run() throws Exception;
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static Check attempt(String name, Attempt attempt)
+   {
+      try
+      {
+         return (new Check(name, true, attempt.run()));
+      }
+      catch(Exception e)
+      {
+         return (new Check(name, false, e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage()));
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/backups/processes/BackupsProcesses.java

@@ -25,9 +25,8 @@ import java.util.Map;
 import com.kingsrook.qbits.backups.BackupsRuntime;
 import com.kingsrook.qbits.backups.model.BackupsTables;
 import com.kingsrook.qbits.backups.permissions.BackupsPermissions;
-import com.kingsrook.qbits.backups.run.BackupEngine;
+import com.kingsrook.qbits.backups.run.BackupRunCoordinator;
 import com.kingsrook.qbits.backups.run.BackupRunFactory;
-import com.kingsrook.qbits.backups.run.BackupRunRequest;
 import com.kingsrook.qbits.backups.run.BackupRunResult;
 import com.kingsrook.qbits.backups.scope.ScopeResolver;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
@@ -76,6 +75,7 @@ public final class BackupsProcesses
       processes.addAll(VerifyProcesses.processes());
       processes.addAll(RestoreProcesses.processes());
       processes.addAll(CatalogProcesses.processes());
+      processes.addAll(LifecycleProcesses.processes());
       return (processes);
    }
 
@@ -212,6 +212,7 @@ public final class BackupsProcesses
       private final ProcessSummaryLine warnings  = line(Status.WARNING, "backup finished with warnings", "backups finished with warnings",
          "backup finished with warnings", "backups finished with warnings");
       private final ProcessSummaryLine failed    = line(Status.ERROR, "backup failed", "backups failed", "backup failed", "backups failed");
+      private final ProcessSummaryLine skipped   = line(Status.WARNING, "backup did not start", "backups did not start", "backup did not start", "backups did not start");
       private final ProcessSummaryLine rows      = line(Status.INFO, "row was written", "rows were written", "row was written", "rows were written");
       private final ProcessSummaryLine files     = line(Status.INFO, "referenced file was copied or already stored", "referenced files were copied or already stored",
          "referenced file was copied or already stored", "referenced files were copied or already stored");
@@ -227,13 +228,17 @@ public final class BackupsProcesses
       {
          for(QRecord plan : input.getRecords())
          {
-            BackupRunRequest request = BackupRunFactory.forPlan(plan, "MANUAL");
-            BackupRunResult  result  = new BackupEngine().execute(request);
+            BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "MANUAL");
             results.add(result);
             switch(result.getStatus())
             {
                case "SUCCEEDED" -> succeeded.incrementCountAndAddPrimaryKey(result.getRunId());
                case "SUCCEEDED_WITH_WARNINGS" -> warnings.incrementCountAndAddPrimaryKey(result.getRunId());
+               case BackupRunCoordinator.SKIPPED_CONCURRENT, BackupRunCoordinator.QUEUED ->
+               {
+                  skipped.incrementCountAndAddPrimaryKey(result.getRunId());
+                  skipped.setMessageSuffix(result.getErrorMessage() == null ? null : ": " + result.getErrorMessage());
+               }
                default ->
                {
                   failed.incrementCountAndAddPrimaryKey(result.getRunId());
@@ -258,6 +263,7 @@ public final class BackupsProcesses
          succeeded.addSelfToListIfAnyCount(lines);
          warnings.addSelfToListIfAnyCount(lines);
          failed.addSelfToListIfAnyCount(lines);
+         skipped.addSelfToListIfAnyCount(lines);
          rows.addSelfToListIfAnyCount(lines);
          files.addSelfToListIfAnyCount(lines);
          return (lines);

src/main/java/com/kingsrook/qbits/backups/processes/CancelRunProcess.java

@@ -0,0 +1,246 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.processes;
+
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.backups.BackupsRuntime;
+import com.kingsrook.qbits.backups.model.BackupsTables;
+import com.kingsrook.qbits.backups.permissions.BackupsPermissions;
+import com.kingsrook.qbits.backups.run.BackupCatalog;
+import com.kingsrook.qbits.backups.run.BackupRunCoordinator;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.Status;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractLoadStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractTransformStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ProcessSummaryProviderInterface;
+import static com.kingsrook.qbits.backups.processes.BackupsProcesses.line;
+
+
+/*******************************************************************************
+ ** Cancel (spec 9): a record action on a running backup run, and on a running
+ ** restore run.  The cancel is cooperative: it sets the run row to CANCELLING, and
+ ** the run, which looks at its own row at every page, stops at the next page and
+ ** ends CANCELLED.  A queued run is cancelled at once, and a run that has lost its
+ ** heartbeat (nothing is running it) is closed straight away.
+ *******************************************************************************/
+public final class CancelRunProcess
+{
+   public static final String NAME         = "backupCancelRun";
+   public static final String NAME_RESTORE = "backupCancelRestoreRun";
+
+
+
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private CancelRunProcess()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** The two processes.
+    *******************************************************************************/
+   public static List<QProcessMetaData> processes()
+   {
+      return (List.of(
+         BackupsProcesses.permission(BackupsProcesses.etl(NAME, "Cancel", BackupsTables.RUN, Transform.class, Load.class,
+            "This asks the selected runs to stop. A run stops at its next page."), BackupsPermissions.RUN),
+         BackupsProcesses.permission(BackupsProcesses.etl(NAME_RESTORE, "Cancel", BackupsTables.RESTORE_RUN, Transform.class, Load.class,
+            "This asks the selected restores to stop. A restore stops at its next page."), BackupsPermissions.RESTORE)));
+   }
+
+
+
+   /*******************************************************************************
+    ** What should happen to a run, by its row.
+    *******************************************************************************/
+   enum Action
+   {
+      ASK_TO_STOP, CANCEL_NOW, CLOSE_STALE, NOT_CANCELLABLE
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   static Action actionFor(QRecord run)
+   {
+      String status = run.getValueString("status");
+      if("QUEUED".equals(status))
+      {
+         return (Action.CANCEL_NOW);
+      }
+      if("RUNNING".equals(status))
+      {
+         Instant heartbeat = run.getValueInstant("heartbeatDate");
+         boolean live = heartbeat != null && heartbeat.isAfter(BackupsRuntime.getClock().instant().minus(BackupRunCoordinator.HEARTBEAT_STALE));
+         return (live ? Action.ASK_TO_STOP : Action.CLOSE_STALE);
+      }
+      return (Action.NOT_CANCELLABLE);
+   }
+
+
+
+   /*******************************************************************************
+    ** Review.
+    *******************************************************************************/
+   public static class Transform extends AbstractTransformStep implements ProcessSummaryProviderInterface
+   {
+      private final ProcessSummaryLine stop   = line(Status.OK, "running run will be asked to stop at its next page", "running runs will be asked to stop at their next page",
+         "running run was asked to stop at its next page", "running runs were asked to stop at their next page");
+      private final ProcessSummaryLine queued = line(Status.OK, "queued run will be cancelled", "queued runs will be cancelled", "queued run was cancelled", "queued runs were cancelled");
+      private final ProcessSummaryLine stale  = line(Status.WARNING, "run has lost its heartbeat (nothing is running it) and will be closed as cancelled",
+         "runs have lost their heartbeat (nothing is running them) and will be closed as cancelled", "run had lost its heartbeat and was closed as cancelled",
+         "runs had lost their heartbeat and were closed as cancelled");
+      private final ProcessSummaryLine no     = line(Status.ERROR, "run cannot be cancelled: it is not running", "runs cannot be cancelled: they are not running",
+         "run could not be cancelled: it is not running", "runs could not be cancelled: they are not running");
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         for(QRecord run : input.getRecords())
+         {
+            switch(actionFor(run))
+            {
+               case ASK_TO_STOP -> stop.incrementCountAndAddPrimaryKey(run.getValue("id"));
+               case CANCEL_NOW -> queued.incrementCountAndAddPrimaryKey(run.getValue("id"));
+               case CLOSE_STALE -> stale.incrementCountAndAddPrimaryKey(run.getValue("id"));
+               default -> no.incrementCountAndAddPrimaryKey(run.getValue("id"));
+            }
+            output.getRecords().add(run);
+         }
+      }
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         stop.addSelfToListIfAnyCount(lines);
+         queued.addSelfToListIfAnyCount(lines);
+         stale.addSelfToListIfAnyCount(lines);
+         no.addSelfToListIfAnyCount(lines);
+         return (lines);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Result.
+    *******************************************************************************/
+   public static class Load extends AbstractLoadStep implements ProcessSummaryProviderInterface
+   {
+      private final ProcessSummaryLine stop   = line(Status.OK, "running run was asked to stop at its next page", "running runs were asked to stop at their next page",
+         "running run was asked to stop at its next page", "running runs were asked to stop at their next page");
+      private final ProcessSummaryLine queued = line(Status.OK, "queued run was cancelled", "queued runs were cancelled", "queued run was cancelled", "queued runs were cancelled");
+      private final ProcessSummaryLine stale  = line(Status.WARNING, "run that had lost its heartbeat was closed as cancelled", "runs that had lost their heartbeat were closed as cancelled",
+         "run that had lost its heartbeat was closed as cancelled", "runs that had lost their heartbeat were closed as cancelled");
+      private final ProcessSummaryLine no     = line(Status.ERROR, "run was not cancelled: it is not running", "runs were not cancelled: they are not running",
+         "run was not cancelled: it is not running", "runs were not cancelled: they are not running");
+
+
+
+      /*******************************************************************************
+       ** Cancel each run.
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         BackupCatalog catalog = new BackupCatalog();
+         String        table   = input.getTableName();
+         for(QRecord selected : input.getRecords())
+         {
+            QRecord run = catalog.get(table, selected.getValueInteger("id")).orElse(selected);
+            switch(actionFor(run))
+            {
+               case ASK_TO_STOP ->
+               {
+                  catalog.update(table, run.getValueInteger("id"), Map.of("status", "CANCELLING"));
+                  stop.incrementCountAndAddPrimaryKey(run.getValue("id"));
+               }
+               case CANCEL_NOW ->
+               {
+                  catalog.update(table, run.getValueInteger("id"), closed("Cancelled before it started."));
+                  queued.incrementCountAndAddPrimaryKey(run.getValue("id"));
+               }
+               case CLOSE_STALE ->
+               {
+                  catalog.update(table, run.getValueInteger("id"), closed("Cancelled: the run had stopped sending heartbeats."));
+                  stale.incrementCountAndAddPrimaryKey(run.getValue("id"));
+               }
+               default -> no.incrementCountAndAddPrimaryKey(run.getValue("id"));
+            }
+            output.getRecords().add(run);
+         }
+      }
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      private static Map<String, Serializable> closed(String message)
+      {
+         Map<String, Serializable> values = new LinkedHashMap<>();
+         values.put("status", "CANCELLED");
+         values.put("endDate", BackupsRuntime.getClock().instant());
+         values.put("errorMessage", message);
+         return (values);
+      }
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         stop.addSelfToListIfAnyCount(lines);
+         queued.addSelfToListIfAnyCount(lines);
+         stale.addSelfToListIfAnyCount(lines);
+         no.addSelfToListIfAnyCount(lines);
+         return (lines);
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/backups/processes/ExportTableProcess.java

@@ -0,0 +1,407 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.processes;
+
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.backups.BackupsQBitConfig;
+import com.kingsrook.qbits.backups.BackupsRuntime;
+import com.kingsrook.qbits.backups.model.BackupsTables;
+import com.kingsrook.qbits.backups.permissions.BackupsPermissionChecker;
+import com.kingsrook.qbits.backups.permissions.BackupsPermissions;
+import com.kingsrook.qbits.backups.run.BackupCatalog;
+import com.kingsrook.qbits.backups.run.BackupRunCoordinator;
+import com.kingsrook.qbits.backups.run.BackupRunRequest;
+import com.kingsrook.qbits.backups.run.BackupRunResult;
+import com.kingsrook.qbits.backups.run.TableReader;
+import com.kingsrook.qbits.backups.scope.PlanSettings;
+import com.kingsrook.qbits.backups.scope.ScopeResolver;
+import com.kingsrook.qbits.backups.scope.TableRole;
+import com.kingsrook.qbits.backups.scope.TableRoles;
+import com.kingsrook.qqq.backend.core.actions.permissions.PermissionsHelper;
+import com.kingsrook.qqq.backend.core.actions.QBackendTransaction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.Status;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.actions.permissions.TablePermissionSubType;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractLoadStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractTransformStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ProcessSummaryProviderInterface;
+import static com.kingsrook.qbits.backups.processes.BackupsProcesses.line;
+
+
+/*******************************************************************************
+ ** Back up these records (spec 9, 11.3): a table and bulk action on any table the
+ ** user can read.  It writes an ad-hoc backups_run (no plan) that restores like any
+ ** other, with referenced files copied the same way.
+ **
+ ** - The records are whatever the table action selected (the chosen records or the
+ **   current filter), read as the user, so record locks apply and the manifest says
+ **   securityScope USER.
+ ** - Sensitive fields are left out unless the user has backups.admin.
+ ** - The destination is the bucket given in the process input (storageBucketAlias),
+ **   else the bucket of the newest active plan.
+ **
+ ** The records are held in memory until the export runs (an export of selected
+ ** records is bounded by what a person selected); a whole table is what a plan is for.
+ **
+ ** The process is added to each table that can be backed up, named
+ ** table.backupExportTable (see exportProcesses).
+ *******************************************************************************/
+public final class ExportTableProcess
+{
+   public static final String NAME_SUFFIX = "backupExportTable";
+
+
+
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private ExportTableProcess()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** The export process for one table.
+    *******************************************************************************/
+   public static QProcessMetaData processFor(String tableName)
+   {
+      return (BackupsProcesses.permission(BackupsProcesses.etl(tableName + "." + NAME_SUFFIX, "Back up these records", tableName, Transform.class, Load.class,
+         "This is a preview of what will be backed up. Nothing is written until you continue."), BackupsPermissions.RUN));
+   }
+
+
+
+   /*******************************************************************************
+    ** The export process of every table of the instance that can be backed up (not the
+    ** qbit own tables, and not secrets or sessions).
+    *******************************************************************************/
+   public static List<QProcessMetaData> exportProcesses(QInstance qInstance)
+   {
+      List<QProcessMetaData> processes = new ArrayList<>();
+      for(QTableMetaData table : qInstance.getTables().values())
+      {
+         if(table.getName().startsWith(BackupsQBitConfig.TABLE_NAME_PREFIX))
+         {
+            continue;
+         }
+         TableRole role = TableRoles.roleOf(qInstance, table);
+         if(role == TableRole.DATA || role == TableRole.AUDIT || role == TableRole.DERIVED)
+         {
+            processes.add(processFor(table.getName()));
+         }
+      }
+      return (processes);
+   }
+
+
+
+   /*******************************************************************************
+    ** The sensitive field policy for the user.
+    *******************************************************************************/
+   static String sensitivePolicy()
+   {
+      return (BackupsPermissionChecker.has(BackupsPermissions.ADMIN) ? "INCLUDE" : "EXCLUDE");
+   }
+
+
+
+   /*******************************************************************************
+    ** The bucket the export goes to, or null when there is none.
+    *******************************************************************************/
+   static String bucketAlias(RunBackendStepInput input) throws QException
+   {
+      String given = input.getValueString("storageBucketAlias");
+      if(given != null && !given.isBlank())
+      {
+         return (given);
+      }
+      List<QRecord> plans = new BackupCatalog().query(BackupsTables.PLAN, new com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter(
+         new com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria("isActive", com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator.EQUALS, true))
+         .withOrderBy(new com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy("id", false)));
+      return (plans.isEmpty() ? null : plans.get(0).getValueString("storageBucketAlias"));
+   }
+
+
+
+   /*******************************************************************************
+    ** Review: what is selected and what will be left out.
+    *******************************************************************************/
+   public static class Transform extends AbstractTransformStep implements ProcessSummaryProviderInterface
+   {
+      private final ProcessSummaryLine records   = line(Status.OK, "record will be backed up", "records will be backed up", "record was backed up", "records were backed up");
+      private final ProcessSummaryLine sensitive = line(Status.INFO, "field is sensitive and will be left out of the backup (including it needs backups.admin)",
+         "fields are sensitive and will be left out of the backup (including them needs backups.admin)", "field was left out as sensitive", "fields were left out as sensitive");
+      private final ProcessSummaryLine included  = line(Status.WARNING, "field is sensitive and will be written in the clear, because you have backups.admin",
+         "fields are sensitive and will be written in the clear, because you have backups.admin", "field was written in the clear", "fields were written in the clear");
+      private final ProcessSummaryLine noBucket  = line(Status.ERROR, "export has no destination: there is no active backup plan to take its bucket from",
+         "exports have no destination: there is no active backup plan to take their bucket from", "export had no destination", "exports had no destination");
+      private final ProcessSummaryLine denied    = line(Status.ERROR, "table cannot be read by you, so it cannot be backed up", "tables cannot be read by you, so they cannot be backed up",
+         "table could not be read by you", "tables could not be read by you");
+      private boolean described = false;
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         String table = input.getTableName();
+         if(!described)
+         {
+            described = true;
+            if(!PermissionsHelper.hasTablePermission(input, table, TablePermissionSubType.READ))
+            {
+               denied.incrementCount();
+            }
+            if(bucketAlias(input) == null)
+            {
+               noBucket.incrementCount();
+            }
+            QTableMetaData meta  = QContext.getQInstance().getTable(table);
+            int            count = 0;
+            for(QFieldMetaData field : meta.getFields().values())
+            {
+               if(!field.getName().equals(meta.getPrimaryKeyField()) && (field.getIsSensitive() || field.getType() == QFieldType.PASSWORD))
+               {
+                  count++;
+               }
+            }
+            (BackupsPermissionChecker.has(BackupsPermissions.ADMIN) ? included : sensitive).incrementCount(count);
+         }
+         if(denied.getCount() > 0)
+         {
+            return;
+         }
+         records.incrementCount(input.getRecords().size());
+         output.getRecords().addAll(input.getRecords());
+      }
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         records.addSelfToListIfAnyCount(lines);
+         sensitive.addSelfToListIfAnyCount(lines);
+         included.addSelfToListIfAnyCount(lines);
+         noBucket.addSelfToListIfAnyCount(lines);
+         denied.addSelfToListIfAnyCount(lines);
+         return (lines);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Result: collect the selected records, then run the export through the engine when
+    ** the Result lines are asked for (after the last page, before post run).
+    *******************************************************************************/
+   public static class Load extends AbstractLoadStep implements ProcessSummaryProviderInterface
+   {
+      private final ProcessSummaryLine succeeded = line(Status.OK, "export succeeded", "exports succeeded", "export succeeded", "exports succeeded");
+      private final ProcessSummaryLine warnings  = line(Status.WARNING, "export finished with warnings", "exports finished with warnings", "export finished with warnings", "exports finished with warnings");
+      private final ProcessSummaryLine failed    = line(Status.ERROR, "export failed", "exports failed", "export failed", "exports failed");
+      private final ProcessSummaryLine rows      = line(Status.INFO, "record was written", "records were written", "record was written", "records were written");
+      private final ProcessSummaryLine files     = line(Status.INFO, "referenced file was copied or already stored", "referenced files were copied or already stored",
+         "referenced file was copied or already stored", "referenced files were copied or already stored");
+
+      private final List<QRecord> buffered = new ArrayList<>();
+
+      private String  tableName;
+      private String  bucket;
+      private boolean exported = false;
+
+
+
+      /*******************************************************************************
+       ** Remember what is being exported.
+       *******************************************************************************/
+      @Override
+      public void preRun(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         super.preRun(input, output);
+         tableName = input.getTableName();
+         bucket = bucketAlias(input);
+      }
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         buffered.addAll(input.getRecords());
+         output.getRecords().addAll(input.getRecords());
+      }
+
+
+
+      /*******************************************************************************
+       ** All records are in (the summary is asked for once the last page is loaded, and
+       ** before post run): back them up, once.
+       *******************************************************************************/
+      private void export() throws QException
+      {
+         if(exported)
+         {
+            return;
+         }
+         exported = true;
+         if(bucket == null)
+         {
+            failed.incrementCount();
+            failed.setMessageSuffix(": there is no active backup plan to take a bucket from");
+            return;
+         }
+         BackupRunResult result = new BackupRunCoordinator().runAdHoc(request(tableName, bucket, buffered));
+         switch(result.getStatus())
+         {
+            case "SUCCEEDED" -> succeeded.incrementCountAndAddPrimaryKey(result.getRunId());
+            case "SUCCEEDED_WITH_WARNINGS" -> warnings.incrementCountAndAddPrimaryKey(result.getRunId());
+            default ->
+            {
+               failed.incrementCountAndAddPrimaryKey(result.getRunId());
+               failed.setMessageSuffix(result.getErrorMessage() == null ? null : ": " + result.getErrorMessage());
+            }
+         }
+         rows.incrementCount((int) Math.min(Integer.MAX_VALUE, result.getRows()));
+         files.incrementCount(result.getFilesCopied() + result.getFilesDeduplicated());
+      }
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         try
+         {
+            export();
+         }
+         catch(QException e)
+         {
+            failed.incrementCount();
+            failed.setMessageSuffix(": " + e.getMessage());
+         }
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         succeeded.addSelfToListIfAnyCount(lines);
+         warnings.addSelfToListIfAnyCount(lines);
+         failed.addSelfToListIfAnyCount(lines);
+         rows.addSelfToListIfAnyCount(lines);
+         files.addSelfToListIfAnyCount(lines);
+         return (lines);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** The ad-hoc request for the selected records of a table.
+    *******************************************************************************/
+   public static BackupRunRequest request(String tableName, String bucketAlias, List<QRecord> selected)
+   {
+      QInstance      qInstance = QContext.getQInstance();
+      QTableMetaData table     = qInstance.getTable(tableName);
+      String         pk        = table.getPrimaryKeyField();
+      QSession       session   = QContext.getQSession();
+      String         user      = session == null || session.getUser() == null ? "system" : session.getUser().getIdReference();
+      BackupsQBitConfig config = BackupsRuntime.getConfig();
+
+      Map<String, QRecord> unique = new LinkedHashMap<>();
+      for(QRecord record : selected)
+      {
+         unique.put(String.valueOf(record.getValue(pk)), record);
+      }
+      List<QRecord> inPkOrder = new ArrayList<>(unique.values());
+      inPkOrder.sort(Comparator.comparing(r -> (Comparable) r.getValue(pk)));
+
+      PlanSettings settings = PlanSettings.adHoc().with("planKey", "adhoc").with("storageBucketAlias", bucketAlias).with("sensitiveFieldPolicy", sensitivePolicy());
+      return (new BackupRunRequest()
+         .withSettings(settings)
+         .withScope(new ScopeResolver.ResolvedScope(List.of(new ScopeResolver.ResolvedTable(tableName, TableRoles.roleOf(qInstance, table), null, List.of())), List.of(), List.of()))
+         .withTrigger("MANUAL")
+         .withRequestedBy(user)
+         .withSecurityScope("USER")
+         .withBucketAlias(bucketAlias)
+         .withStorage(BackupsRuntime.storageFor(bucketAlias))
+         .withSourceResolver(BackupsRuntime::storageFor)
+         .withInstanceName(config.getInstanceName() != null ? config.getInstanceName() : "qqq")
+         .withPlanName("Export of " + table.getLabel())
+         .withReader(new BufferedReader(inPkOrder))
+         .withClock(BackupsRuntime.getClock()));
+   }
+
+
+
+   /*******************************************************************************
+    ** Serves the selected records to the engine, in primary key order.
+    *******************************************************************************/
+   static class BufferedReader implements TableReader
+   {
+      private final List<QRecord> records;
+
+
+
+      /*******************************************************************************
+       ** Constructor.
+       *******************************************************************************/
+      BufferedReader(List<QRecord> records)
+      {
+         this.records = records;
+      }
+
+
+
+      /*******************************************************************************
+       ** {@inheritDoc}
+       *******************************************************************************/
+      @Override
+      public void read(QTableMetaData table, String filterJson, Serializable startAfterKey, int pageSize, QBackendTransaction transaction, java.util.function.Consumer<QRecord> sink)
+      {
+         records.forEach(sink);
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/backups/processes/LifecycleProcesses.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.processes;
+
+
+import java.util.ArrayList;
+import java.util.List;
+import com.kingsrook.qbits.backups.BackupsQBitConfig;
+import com.kingsrook.qbits.backups.BackupsRuntime;
+import com.kingsrook.qbits.backups.schedule.BackupPlanSyncToScheduledJobProcess;
+import com.kingsrook.qbits.backups.schedule.BackupPlanVerifySyncToScheduledJobProcess;
+import com.kingsrook.qbits.backups.schedule.BackupScheduledRunProcess;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+import com.kingsrook.qqq.backend.core.utils.StringUtils;
+
+
+/*******************************************************************************
+ ** The processes of the run lifecycle (spec 9): scheduled runs, cancel, resume,
+ ** test destination and prune, and the plan to scheduled job sync.  Added to the
+ ** qbit by BackupsProcesses.processes().  The per-table export process
+ ** (exportProcesses) needs the instance tables, so the producer adds it.
+ *******************************************************************************/
+public final class LifecycleProcesses
+{
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private LifecycleProcesses()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** The processes.
+    *******************************************************************************/
+   public static List<QProcessMetaData> processes()
+   {
+      List<QProcessMetaData> processes = new ArrayList<>();
+      processes.add(BackupScheduledRunProcess.processMetaData());
+      processes.addAll(CancelRunProcess.processes());
+      processes.add(ResumeRunProcess.process());
+      processes.add(TestDestinationProcess.process());
+      processes.addAll(PruneProcess.processes());
+
+      BackupsQBitConfig config = BackupsRuntime.getConfig();
+      if(config != null && StringUtils.hasContent(config.getSchedulerName()))
+      {
+         try
+         {
+            processes.add(new BackupPlanSyncToScheduledJobProcess().produce(null));
+            processes.add(new BackupPlanVerifySyncToScheduledJobProcess().produce(null));
+         }
+         catch(QException e)
+         {
+            throw (new IllegalStateException("Could not define the plan schedule sync processes: " + e.getMessage(), e));
+         }
+      }
+      return (processes);
+   }
+}

src/main/java/com/kingsrook/qbits/backups/processes/PruneProcess.java

@@ -0,0 +1,336 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.processes;
+
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import com.kingsrook.qbits.backups.BackupsQBitConfig;
+import com.kingsrook.qbits.backups.BackupsRuntime;
+import com.kingsrook.qbits.backups.model.BackupsTables;
+import com.kingsrook.qbits.backups.permissions.BackupsPermissions;
+import com.kingsrook.qbits.backups.prune.BackupPruner;
+import com.kingsrook.qbits.backups.prune.PruneCalculator;
+import com.kingsrook.qbits.backups.run.BackupCatalog;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.Status;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.scheduleing.QScheduleMetaData;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractLoadStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractTransformStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ProcessSummaryProviderInterface;
+import static com.kingsrook.qbits.backups.processes.BackupsProcesses.line;
+
+
+/*******************************************************************************
+ ** Prune per retention (spec 9, 10.3).  A record action on backups_plan, and a
+ ** table action on backups_run (which prunes the plans of the selected runs); the
+ ** plan one also runs daily when the qbit has a scheduler.
+ **
+ ** Review shows exactly what Result will do: the runs to delete, the runs kept by
+ ** class, the runs kept because an object is still locked (with the date), the
+ ** debris prefixes, the content-addressed objects to sweep, and the bytes freed.
+ ** Review writes nothing.
+ *******************************************************************************/
+public final class PruneProcess
+{
+   public static final String NAME      = "backupPrune";
+   public static final String NAME_RUNS = "backupPruneRuns";
+
+   private static final long MIB = 1024L * 1024L;
+
+
+
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private PruneProcess()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** The two processes.
+    *******************************************************************************/
+   public static List<QProcessMetaData> processes()
+   {
+      QProcessMetaData onPlans = BackupsProcesses.permission(BackupsProcesses.etl(NAME, "Prune per retention", BackupsTables.PLAN, PlanTransform.class, Load.class,
+         "This is a preview of what Prune will delete. Nothing is deleted until you continue."), BackupsPermissions.ADMIN);
+
+      BackupsQBitConfig config = BackupsRuntime.getConfig();
+      if(config != null && config.getSchedulerName() != null && !config.getSchedulerName().isBlank())
+      {
+         onPlans.withSchedule(new QScheduleMetaData().withSchedulerName(config.getSchedulerName()).withRepeatSeconds(24 * 3600).withInitialDelaySeconds(3600));
+      }
+
+      QProcessMetaData onRuns = BackupsProcesses.permission(BackupsProcesses.etl(NAME_RUNS, "Prune per retention", BackupsTables.RUN, RunTransform.class, Load.class,
+         "This is a preview of what Prune will delete for the plans of the selected runs. Nothing is deleted until you continue."), BackupsPermissions.ADMIN);
+      return (List.of(onPlans, onRuns));
+   }
+
+
+
+   /*******************************************************************************
+    ** The review lines, shared by the two transforms.
+    *******************************************************************************/
+   abstract static class BaseTransform extends AbstractTransformStep implements ProcessSummaryProviderInterface
+   {
+      private final ProcessSummaryLine plans    = line(Status.OK, "plan will be pruned", "plans will be pruned", "plan was pruned", "plans were pruned");
+      private final ProcessSummaryLine delete   = line(Status.WARNING, "run will be deleted", "runs will be deleted", "run was deleted", "runs were deleted");
+      private final ProcessSummaryLine daily    = line(Status.OK, "run is kept as a daily backup", "runs are kept as daily backups", "run was kept as a daily backup", "runs were kept as daily backups");
+      private final ProcessSummaryLine weekly   = line(Status.OK, "run is kept as a weekly backup", "runs are kept as weekly backups", "run was kept as a weekly backup", "runs were kept as weekly backups");
+      private final ProcessSummaryLine monthly  = line(Status.OK, "run is kept as a monthly backup", "runs are kept as monthly backups", "run was kept as a monthly backup", "runs were kept as monthly backups");
+      private final ProcessSummaryLine yearly   = line(Status.OK, "run is kept as a yearly backup", "runs are kept as yearly backups", "run was kept as a yearly backup", "runs were kept as yearly backups");
+      private final ProcessSummaryLine guarded  = line(Status.OK, "run is kept by a guard (the minimum, the newest verified run, or a recent restore)",
+         "runs are kept by a guard (the minimum, the newest verified run, or a recent restore)", "run was kept by a guard", "runs were kept by a guard");
+      private final ProcessSummaryLine locked   = line(Status.WARNING, "run is kept: its objects are locked", "runs are kept: their objects are locked",
+         "run was kept: its objects are locked", "runs were kept: their objects are locked");
+      private final ProcessSummaryLine debris   = line(Status.INFO, "unfinished backup (older than 7 days) will be deleted", "unfinished backups (older than 7 days) will be deleted",
+         "unfinished backup was deleted", "unfinished backups were deleted");
+      private final ProcessSummaryLine sweep    = line(Status.INFO, "stored file that no retained backup references will be deleted", "stored files that no retained backup references will be deleted",
+         "stored file was deleted", "stored files were deleted");
+      private final ProcessSummaryLine sweepLocked = line(Status.WARNING, "stored file is kept: it is locked", "stored files are kept: they are locked", "stored file was kept: it is locked", "stored files were kept: they are locked");
+      private final ProcessSummaryLine deferred = line(Status.WARNING, "plan's file store sweep is skipped", "plans' file store sweeps are skipped", "plan's file store sweep was skipped", "plans' file store sweeps were skipped");
+      private final ProcessSummaryLine freed    = line(Status.INFO, "MiB of storage will be freed", "MiB of storage will be freed", "MiB of storage was freed", "MiB of storage was freed");
+      private final ProcessSummaryLine failed   = line(Status.ERROR, "plan cannot be pruned", "plans cannot be pruned", "plan could not be pruned", "plans could not be pruned");
+
+      private final Set<Integer> seenPlans = new HashSet<>();
+      private       long         bytes     = 0;
+      private       Instant      latestLock;
+
+
+
+      /*******************************************************************************
+       ** The plan records this step looks at.
+       *******************************************************************************/
+      abstract List<QRecord> plansFor(RunBackendStepInput input) throws QException;
+
+
+
+      /*******************************************************************************
+       ** Plan each selected plan, writing nothing.
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         for(QRecord plan : plansFor(input))
+         {
+            if(!seenPlans.add(plan.getValueInteger("id")))
+            {
+               continue;
+            }
+            try
+            {
+               BackupPruner.PrunePlan prunePlan = new BackupPruner().plan(plan);
+               plans.incrementCountAndAddPrimaryKey(plan.getValue("id"));
+               delete.incrementCount(prunePlan.toDelete().size());
+               var kept = prunePlan.keptByClass();
+               daily.incrementCount(kept.getOrDefault(PruneCalculator.RetentionClass.DAILY, 0));
+               weekly.incrementCount(kept.getOrDefault(PruneCalculator.RetentionClass.WEEKLY, 0));
+               monthly.incrementCount(kept.getOrDefault(PruneCalculator.RetentionClass.MONTHLY, 0));
+               yearly.incrementCount(kept.getOrDefault(PruneCalculator.RetentionClass.YEARLY, 0));
+               guarded.incrementCount(kept.getOrDefault(PruneCalculator.RetentionClass.NONE, 0));
+               locked.incrementCount(prunePlan.lockedNotKept().size());
+               for(BackupPruner.RunPlan run : prunePlan.lockedNotKept())
+               {
+                  latestLock = latestLock == null || run.lockedUntil().isAfter(latestLock) ? run.lockedUntil() : latestLock;
+               }
+               debris.incrementCount(prunePlan.debris().stream().filter(d -> d.lockedUntil() == null).toList().size());
+               sweep.incrementCount(prunePlan.sweep().deletable().size());
+               sweepLocked.incrementCount(prunePlan.sweep().locked().size());
+               if(prunePlan.sweep().deferredReason() != null)
+               {
+                  deferred.incrementCount();
+                  deferred.setMessageSuffix(": " + prunePlan.sweep().deferredReason());
+               }
+               bytes += prunePlan.bytesToFree();
+               output.getRecords().add(plan);
+            }
+            catch(Exception e)
+            {
+               failed.incrementCountAndAddPrimaryKey(plan.getValue("id"));
+               failed.setMessageSuffix(": " + e.getMessage());
+            }
+         }
+      }
+
+
+
+      /*******************************************************************************
+       ** The lines.
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         if(latestLock != null)
+         {
+            locked.setMessageSuffix(" (the latest lock ends " + latestLock + ")");
+         }
+         freed.setCount((int) Math.min(Integer.MAX_VALUE, (bytes + MIB - 1) / MIB));
+
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         plans.addSelfToListIfAnyCount(lines);
+         delete.addSelfToListIfAnyCount(lines);
+         daily.addSelfToListIfAnyCount(lines);
+         weekly.addSelfToListIfAnyCount(lines);
+         monthly.addSelfToListIfAnyCount(lines);
+         yearly.addSelfToListIfAnyCount(lines);
+         guarded.addSelfToListIfAnyCount(lines);
+         locked.addSelfToListIfAnyCount(lines);
+         debris.addSelfToListIfAnyCount(lines);
+         sweep.addSelfToListIfAnyCount(lines);
+         sweepLocked.addSelfToListIfAnyCount(lines);
+         deferred.addSelfToListIfAnyCount(lines);
+         freed.addSelfToListIfAnyCount(lines);
+         failed.addSelfToListIfAnyCount(lines);
+         return (lines);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Review for plans.
+    *******************************************************************************/
+   public static class PlanTransform extends BaseTransform
+   {
+      /*******************************************************************************
+       ** {@inheritDoc}
+       *******************************************************************************/
+      @Override
+      List<QRecord> plansFor(RunBackendStepInput input)
+      {
+         return (input.getRecords());
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Review for runs: the plans of the selected runs.
+    *******************************************************************************/
+   public static class RunTransform extends BaseTransform
+   {
+      /*******************************************************************************
+       ** {@inheritDoc}
+       *******************************************************************************/
+      @Override
+      List<QRecord> plansFor(RunBackendStepInput input) throws QException
+      {
+         List<QRecord> plans = new ArrayList<>();
+         BackupCatalog catalog = new BackupCatalog();
+         for(QRecord run : input.getRecords())
+         {
+            if(run.getValueInteger("backupPlanId") != null)
+            {
+               catalog.get(BackupsTables.PLAN, run.getValueInteger("backupPlanId")).ifPresent(plans::add);
+            }
+         }
+         return (plans);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Result: do it, and report what was done.
+    *******************************************************************************/
+   public static class Load extends AbstractLoadStep implements ProcessSummaryProviderInterface
+   {
+      private final ProcessSummaryLine plans     = line(Status.OK, "plan will be pruned", "plans will be pruned", "plan was pruned", "plans were pruned");
+      private final ProcessSummaryLine deleted   = line(Status.OK, "run will be deleted", "runs will be deleted", "run was deleted", "runs were deleted");
+      private final ProcessSummaryLine locked    = line(Status.WARNING, "run was kept: locked until a later date", "runs were kept: locked until a later date",
+         "run was kept: locked until a later date", "runs were kept: locked until a later date");
+      private final ProcessSummaryLine debris    = line(Status.OK, "unfinished backup will be deleted", "unfinished backups will be deleted", "unfinished backup was deleted", "unfinished backups were deleted");
+      private final ProcessSummaryLine files     = line(Status.OK, "stored file will be deleted", "stored files will be deleted", "stored file was deleted", "stored files were deleted");
+      private final ProcessSummaryLine filesKept = line(Status.WARNING, "stored file was kept: it is locked", "stored files were kept: they are locked", "stored file was kept: it is locked", "stored files were kept: they are locked");
+      private final ProcessSummaryLine notes     = line(Status.INFO, "note", "notes", "note", "notes");
+      private final ProcessSummaryLine freed     = line(Status.INFO, "MiB of storage was freed", "MiB of storage was freed", "MiB of storage was freed", "MiB of storage was freed");
+      private final ProcessSummaryLine failed    = line(Status.ERROR, "plan could not be pruned", "plans could not be pruned", "plan could not be pruned", "plans could not be pruned");
+      private       long               bytes     = 0;
+      private final List<String>       messages  = new ArrayList<>();
+
+
+
+      /*******************************************************************************
+       ** Prune each plan.
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         com.kingsrook.qqq.backend.core.model.session.QSession session = com.kingsrook.qqq.backend.core.context.QContext.getQSession();
+         String user = session == null || session.getUser() == null ? "system:scheduler" : session.getUser().getIdReference();
+         for(QRecord plan : input.getRecords())
+         {
+            try
+            {
+               BackupPruner.Outcome outcome = new BackupPruner().prune(plan, user);
+               plans.incrementCountAndAddPrimaryKey(plan.getValue("id"));
+               deleted.incrementCount(outcome.runsDeleted());
+               locked.incrementCount(outcome.runsLocked());
+               debris.incrementCount(outcome.debrisPrefixesDeleted());
+               files.incrementCount(outcome.filesSwept());
+               filesKept.incrementCount(outcome.filesLocked());
+               bytes += outcome.bytesFreed();
+               messages.addAll(outcome.messages());
+               output.getRecords().add(plan);
+            }
+            catch(Exception e)
+            {
+               failed.incrementCountAndAddPrimaryKey(plan.getValue("id"));
+               failed.setMessageSuffix(": " + e.getMessage());
+            }
+         }
+      }
+
+
+
+      /*******************************************************************************
+       ** The Result lines: what was deleted, and what was kept because it is locked.
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         freed.setCount((int) Math.min(Integer.MAX_VALUE, (bytes + MIB - 1) / MIB));
+         notes.setCount(messages.size());
+         if(!messages.isEmpty())
+         {
+            notes.setMessageSuffix(": " + String.join("; ", messages.stream().limit(5).toList()) + (messages.size() > 5 ? "; and " + (messages.size() - 5) + " more" : ""));
+         }
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         plans.addSelfToListIfAnyCount(lines);
+         deleted.addSelfToListIfAnyCount(lines);
+         locked.addSelfToListIfAnyCount(lines);
+         debris.addSelfToListIfAnyCount(lines);
+         files.addSelfToListIfAnyCount(lines);
+         filesKept.addSelfToListIfAnyCount(lines);
+         freed.addSelfToListIfAnyCount(lines);
+         notes.addSelfToListIfAnyCount(lines);
+         failed.addSelfToListIfAnyCount(lines);
+         return (lines);
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/backups/processes/ResumeRunProcess.java

@@ -0,0 +1,200 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.processes;
+
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.backups.model.BackupsTables;
+import com.kingsrook.qbits.backups.permissions.BackupsPermissions;
+import com.kingsrook.qbits.backups.run.BackupCatalog;
+import com.kingsrook.qbits.backups.run.BackupResumePlanner;
+import com.kingsrook.qbits.backups.run.BackupRunCoordinator;
+import com.kingsrook.qbits.backups.run.BackupRunResult;
+import com.kingsrook.qbits.backups.run.ResumeState;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.Status;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractLoadStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractTransformStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ProcessSummaryProviderInterface;
+import static com.kingsrook.qbits.backups.processes.BackupsProcesses.line;
+
+
+/*******************************************************************************
+ ** Resume (spec 9): a record action on a failed or timed out run that read each
+ ** table on its own.  The same run carries on: the tables that finished are kept,
+ ** and the interrupted table continues from the last key of its last complete
+ ** part.  A run that held one snapshot cannot be resumed (a new read would not be
+ ** the same instant), and says so; Review explains why and the way out is a fresh
+ ** run of the plan.
+ *******************************************************************************/
+public final class ResumeRunProcess
+{
+   public static final String NAME = "backupResumeRun";
+
+
+
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private ResumeRunProcess()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** The process.
+    *******************************************************************************/
+   public static QProcessMetaData process()
+   {
+      return (BackupsProcesses.permission(BackupsProcesses.etl(NAME, "Resume", BackupsTables.RUN, Transform.class, Load.class,
+         "This is a preview of where each selected run will carry on. Nothing is written until you continue."), BackupsPermissions.RUN));
+   }
+
+
+
+   /*******************************************************************************
+    ** Review: what is already done, and where the interrupted table resumes.
+    *******************************************************************************/
+   public static class Transform extends AbstractTransformStep implements ProcessSummaryProviderInterface
+   {
+      private final ProcessSummaryLine runs     = line(Status.OK, "run will be resumed", "runs will be resumed", "run was resumed", "runs were resumed");
+      private final ProcessSummaryLine done     = line(Status.INFO, "table is complete and will be kept as it is", "tables are complete and will be kept as they are",
+         "table was complete and was kept", "tables were complete and were kept");
+      private final ProcessSummaryLine carryOn  = line(Status.INFO, "table will carry on from its last complete part", "tables will carry on from their last complete part",
+         "table carried on from its last complete part", "tables carried on from their last complete part");
+      private final ProcessSummaryLine cannot   = line(Status.ERROR, "run cannot be resumed", "runs cannot be resumed", "run could not be resumed", "runs could not be resumed");
+      private final List<String>       resumeAt = new ArrayList<>();
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         BackupCatalog catalog = new BackupCatalog();
+         for(QRecord run : input.getRecords())
+         {
+            QRecord plan = run.getValueInteger("backupPlanId") == null ? null : catalog.get(BackupsTables.PLAN, run.getValueInteger("backupPlanId")).orElse(null);
+            String  why  = BackupResumePlanner.whyNotResumable(run, plan);
+            if(why != null)
+            {
+               cannot.incrementCountAndAddPrimaryKey(run.getValue("id"));
+               cannot.setMessageSuffix(": " + why);
+               continue;
+            }
+            ResumeState state = BackupResumePlanner.plan(run);
+            runs.incrementCountAndAddPrimaryKey(run.getValue("id"));
+            for(Map.Entry<String, ResumeState.TableState> entry : state.tables().entrySet())
+            {
+               if(entry.getValue().complete())
+               {
+                  done.incrementCount();
+               }
+               else
+               {
+                  carryOn.incrementCount();
+                  resumeAt.add(entry.getKey() + " after key " + entry.getValue().lastKey());
+               }
+            }
+            output.getRecords().add(run);
+         }
+      }
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         if(!resumeAt.isEmpty())
+         {
+            carryOn.setMessageSuffix(": " + String.join(", ", resumeAt.stream().limit(5).toList()));
+         }
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         runs.addSelfToListIfAnyCount(lines);
+         done.addSelfToListIfAnyCount(lines);
+         carryOn.addSelfToListIfAnyCount(lines);
+         cannot.addSelfToListIfAnyCount(lines);
+         return (lines);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Result: run the engine again on each run.
+    *******************************************************************************/
+   public static class Load extends AbstractLoadStep implements ProcessSummaryProviderInterface
+   {
+      private final ProcessSummaryLine succeeded = line(Status.OK, "run will succeed", "runs succeeded", "run succeeded after being resumed", "runs succeeded after being resumed");
+      private final ProcessSummaryLine warnings  = line(Status.WARNING, "run finished with warnings", "runs finished with warnings", "run finished with warnings", "runs finished with warnings");
+      private final ProcessSummaryLine failed    = line(Status.ERROR, "run failed again", "runs failed again", "run failed again", "runs failed again");
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         for(QRecord run : input.getRecords())
+         {
+            BackupRunResult result = new BackupRunCoordinator().resume(run);
+            switch(result.getStatus())
+            {
+               case "SUCCEEDED" -> succeeded.incrementCountAndAddPrimaryKey(run.getValue("id"));
+               case "SUCCEEDED_WITH_WARNINGS" -> warnings.incrementCountAndAddPrimaryKey(run.getValue("id"));
+               default ->
+               {
+                  failed.incrementCountAndAddPrimaryKey(run.getValue("id"));
+                  failed.setMessageSuffix(result.getErrorMessage() == null ? null : ": " + result.getErrorMessage());
+               }
+            }
+            output.getRecords().add(run);
+         }
+      }
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         succeeded.addSelfToListIfAnyCount(lines);
+         warnings.addSelfToListIfAnyCount(lines);
+         failed.addSelfToListIfAnyCount(lines);
+         return (lines);
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/backups/processes/TestDestinationProcess.java

@@ -0,0 +1,173 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.processes;
+
+
+import java.util.ArrayList;
+import java.util.List;
+import com.kingsrook.qbits.backups.model.BackupsTables;
+import com.kingsrook.qbits.backups.permissions.BackupsPermissions;
+import com.kingsrook.qbits.backups.plan.DestinationProbe;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.Status;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractLoadStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractTransformStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ProcessSummaryProviderInterface;
+import static com.kingsrook.qbits.backups.processes.BackupsProcesses.line;
+
+
+/*******************************************************************************
+ ** Test destination (spec 9): a record action on backups_plan.  Result shows the
+ ** write, read, list and delete of a probe object (head only on a locked bucket),
+ ** and the capability report of the bucket.  See DestinationProbe.
+ *******************************************************************************/
+public final class TestDestinationProcess
+{
+   public static final String NAME = "backupTestDestination";
+
+
+
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private TestDestinationProcess()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** The process.
+    *******************************************************************************/
+   public static QProcessMetaData process()
+   {
+      return (BackupsProcesses.permission(BackupsProcesses.etl(NAME, "Test destination", BackupsTables.PLAN, Transform.class, Load.class,
+         "This will write a small probe object to the destination, read it back, list it and delete it. On a locked bucket it only reads."), BackupsPermissions.ADMIN));
+   }
+
+
+
+   /*******************************************************************************
+    ** Review: nothing is written yet.
+    *******************************************************************************/
+   public static class Transform extends AbstractTransformStep implements ProcessSummaryProviderInterface
+   {
+      private final ProcessSummaryLine plans = line(Status.OK, "plan destination will be tested", "plan destinations will be tested", "plan destination was tested", "plan destinations were tested");
+      private final ProcessSummaryLine locked = line(Status.INFO, "destination is object locked: the test will only read (no probe object is written)",
+         "destinations are object locked: the tests will only read (no probe object is written)", "destination was object locked: the test only read", "destinations were object locked: the tests only read");
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         for(QRecord plan : input.getRecords())
+         {
+            plans.incrementCountAndAddPrimaryKey(plan.getValue("id"));
+            String mode = plan.getValueString("objectLockMode");
+            if(mode != null && !"NONE".equals(mode))
+            {
+               locked.incrementCount();
+            }
+            output.getRecords().add(plan);
+         }
+      }
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         plans.addSelfToListIfAnyCount(lines);
+         locked.addSelfToListIfAnyCount(lines);
+         return (lines);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Result: the checks and the capability report.
+    *******************************************************************************/
+   public static class Load extends AbstractLoadStep implements ProcessSummaryProviderInterface
+   {
+      private final List<ProcessSummaryLineInterface> lines = new ArrayList<>();
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         for(QRecord plan : input.getRecords())
+         {
+            DestinationProbe.Report report = DestinationProbe.probe(plan);
+            String                  name   = plan.getValueString("name");
+            for(DestinationProbe.Check check : report.checks())
+            {
+               ProcessSummaryLine line = line(check.ok() ? Status.OK : Status.ERROR, check.name() + " succeeded: " + check.detail(), check.name() + " succeeded: " + check.detail(),
+                  check.name() + " succeeded: " + check.detail(), check.name() + " succeeded: " + check.detail());
+               if(!check.ok())
+               {
+                  line = line(Status.ERROR, check.name() + " failed: " + check.detail(), check.name() + " failed: " + check.detail(),
+                     check.name() + " failed: " + check.detail(), check.name() + " failed: " + check.detail());
+               }
+               line.setCount(1);
+               line.setMessageSuffix(" (" + name + ")");
+               lines.add(line);
+            }
+            ProcessSummaryLine capabilities = line(Status.INFO, "Capabilities: " + String.join(", ", report.capabilities()), "Capabilities: " + String.join(", ", report.capabilities()),
+               "Capabilities: " + String.join(", ", report.capabilities()), "Capabilities: " + String.join(", ", report.capabilities()));
+            capabilities.setCount(1);
+            lines.add(capabilities);
+            for(String warning : report.warnings())
+            {
+               ProcessSummaryLine w = line(Status.WARNING, warning, warning, warning, warning);
+               w.setCount(1);
+               lines.add(w);
+            }
+            output.getRecords().add(plan);
+         }
+      }
+
+
+
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean isForResultScreen)
+      {
+         return (new ArrayList<>(lines));
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/backups/prune/BackupPruner.java

@@ -0,0 +1,771 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.prune;
+
+
+import java.io.BufferedReader;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.zip.GZIPInputStream;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.kingsrook.qbits.backups.BackupsQBitConfig;
+import com.kingsrook.qbits.backups.BackupsRuntime;
+import com.kingsrook.qbits.backups.model.BackupsTables;
+import com.kingsrook.qbits.backups.run.BackupCatalog;
+import com.kingsrook.qbits.backups.run.BackupRunCoordinator;
+import com.kingsrook.qbits.backups.storage.BackupStorage;
+import com.kingsrook.qbits.backups.storage.StoredObjectInfo;
+import com.kingsrook.qqq.backend.core.actions.audits.AuditAction;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
+
+
+/*******************************************************************************
+ ** Prune per retention (spec 10.3, 6.4, 8.5): plans what to delete without writing
+ ** anything (the Review step), and carries it out (the Result step).
+ **
+ ** - Runs are classified (backups_run.retentionClass) by the GFS PruneCalculator,
+ **   and the runs that are not kept have every object deleted through
+ **   BackupStorage.deletePermanent on the plan prune alias (pruneBucketAlias; the
+ **   backup alias when none is set).  The run row is kept as history, with
+ **   prunedDate set.
+ ** - A run with an object that is still locked is not touched: it is reported as
+ **   "kept: locked until ..." and tried again at the next prune.
+ ** - Prefixes of the plan with no manifest, whose newest object is older than 7
+ **   days, are debris and are always deleted.
+ ** - The content-addressed file store is mark-and-sweep: the file indexes of the
+ **   retained runs give each backups_file_object its referenceCount, and objects
+ **   nothing retained references (and that are not locked) are deleted, with
+ **   deletedDate set.  Nothing is swept while a backup is running or when any index
+ **   cannot be read: no deletion on uncertainty.
+ *******************************************************************************/
+public class BackupPruner
+{
+   private static final QLogger       LOG    = QLogger.getLogger(BackupPruner.class);
+   private static final ObjectMapper  MAPPER = new ObjectMapper();
+
+   private final BackupCatalog catalog = new BackupCatalog();
+
+
+
+   /*******************************************************************************
+    ** One run, with what it holds in storage.
+    *******************************************************************************/
+   public record RunPlan(Integer runId, String runKey, String runPrefix, PruneCalculator.Decision decision, List<StoredObjectInfo> objects, long bytes, Instant lockedUntil)
+   {
+      /*******************************************************************************
+       ** Whether the run is to be deleted (not kept, and nothing locked).
+       *******************************************************************************/
+      public boolean toDelete()
+      {
+         return (!decision.keep() && lockedUntil == null);
+      }
+
+
+
+      /*******************************************************************************
+       ** Whether the run is not kept by retention but cannot be deleted yet.
+       *******************************************************************************/
+      public boolean lockedNotKept()
+      {
+         return (!decision.keep() && lockedUntil != null);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** A prefix with no manifest, older than 7 days.
+    *******************************************************************************/
+   public record DebrisPlan(String prefix, Integer runId, List<StoredObjectInfo> objects, long bytes, Instant lockedUntil)
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** One content-addressed object that nothing retained references.
+    *******************************************************************************/
+   public record SweepCandidate(Integer fileObjectId, String sha256, String objectKey, long size, Instant lockedUntil)
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** The mark-and-sweep of the file store.
+    **
+    ** @param deferredReason why nothing will be swept, or null
+    ** @param referenceCounts the new referenceCount of each live file object, by id
+    ** @param candidates the unreferenced objects
+    *******************************************************************************/
+   public record SweepPlan(String deferredReason, Map<Integer, Integer> referenceCounts, List<SweepCandidate> candidates)
+   {
+      /*******************************************************************************
+       ** Candidates that will be deleted (not locked).
+       *******************************************************************************/
+      public List<SweepCandidate> deletable()
+      {
+         return (candidates.stream().filter(c -> c.lockedUntil() == null).toList());
+      }
+
+
+
+      /*******************************************************************************
+       ** Candidates still locked.
+       *******************************************************************************/
+      public List<SweepCandidate> locked()
+      {
+         return (candidates.stream().filter(c -> c.lockedUntil() != null).toList());
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** The whole plan for one backup plan.
+    *******************************************************************************/
+   public record PrunePlan(Integer planId, String planKey, List<RunPlan> runs, List<DebrisPlan> debris, SweepPlan sweep)
+   {
+      /*******************************************************************************
+       ** Runs to delete.
+       *******************************************************************************/
+      public List<RunPlan> toDelete()
+      {
+         return (runs.stream().filter(RunPlan::toDelete).toList());
+      }
+
+
+
+      /*******************************************************************************
+       ** Runs not kept by retention but locked.
+       *******************************************************************************/
+      public List<RunPlan> lockedNotKept()
+      {
+         return (runs.stream().filter(RunPlan::lockedNotKept).toList());
+      }
+
+
+
+      /*******************************************************************************
+       ** The kept runs by class.
+       *******************************************************************************/
+      public Map<PruneCalculator.RetentionClass, Integer> keptByClass()
+      {
+         return (PruneCalculator.keptByClass(runs.stream().map(RunPlan::decision).toList()));
+      }
+
+
+
+      /*******************************************************************************
+       ** How many runs are kept only because a guard says so (minimum, verified, restore).
+       *******************************************************************************/
+      public long keptByGuard()
+      {
+         return (runs.stream().filter(r -> r.decision().keep() && !r.decision().guards().isEmpty()).count());
+      }
+
+
+
+      /*******************************************************************************
+       ** Bytes that will be freed.
+       *******************************************************************************/
+      public long bytesToFree()
+      {
+         return (toDelete().stream().mapToLong(RunPlan::bytes).sum()
+            + debris.stream().filter(d -> d.lockedUntil() == null).mapToLong(DebrisPlan::bytes).sum()
+            + sweep.deletable().stream().mapToLong(SweepCandidate::size).sum());
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** What a prune did.
+    *******************************************************************************/
+   public record Outcome(int runsDeleted, int runsLocked, int debrisPrefixesDeleted, int debrisLocked, int filesSwept, int filesLocked, long bytesFreed, List<String> messages)
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** Work out what a prune would do, writing nothing.
+    *******************************************************************************/
+   public PrunePlan plan(QRecord plan) throws QException
+   {
+      Instant       now        = BackupsRuntime.getClock().instant();
+      Integer       planId     = plan.getValueInteger("id");
+      String        planKey    = plan.getValueString("planKey");
+      BackupStorage deleteSide = BackupsRuntime.storageFor(pruneAlias(plan));
+
+      List<QRecord> allRuns = catalog.where(BackupsTables.RUN, "backupPlanId", planId);
+
+      /////////////////
+      // GFS classes //
+      /////////////////
+      Set<Integer> protectedByRestore = restoreProtected(now);
+      List<PruneCalculator.RunInfo> infos = new ArrayList<>();
+      Map<Integer, QRecord> usable = new LinkedHashMap<>();
+      for(QRecord run : allRuns)
+      {
+         if(run.getValue("prunedDate") == null && isSuccessful(run))
+         {
+            usable.put(run.getValueInteger("id"), run);
+            infos.add(new PruneCalculator.RunInfo(run.getValueInteger("id"), startOf(run), isVerified(run), protectedByRestore.contains(run.getValueInteger("id"))));
+         }
+      }
+      ZoneId zone = zoneOf(plan);
+      List<PruneCalculator.Decision> decisions = PruneCalculator.calculate(infos, new PruneCalculator.Policy(intOf(plan, "retainDaily", 7),
+         intOf(plan, "retainWeekly", 4), intOf(plan, "retainMonthly", 12), intOf(plan, "retainYearly", 0), intOf(plan, "retainMinimum", 3)), zone);
+
+      List<RunPlan> runPlans = new ArrayList<>();
+      for(PruneCalculator.Decision decision : decisions)
+      {
+         QRecord                run     = usable.get(decision.id());
+         List<StoredObjectInfo> objects = decision.keep() ? List.of() : deleteSide.list(run.getValueString("runPrefix"));
+         runPlans.add(new RunPlan(decision.id(), run.getValueString("runKey"), run.getValueString("runPrefix"), decision, objects,
+            objects.stream().mapToLong(StoredObjectInfo::size).sum(), latestLock(objects, now)));
+      }
+
+      Set<Integer> deleting = new HashSet<>();
+      runPlans.stream().filter(RunPlan::toDelete).forEach(r -> deleting.add(r.runId()));
+
+      return (new PrunePlan(planId, planKey, runPlans, debris(plan, allRuns, deleteSide, now), sweepPlan(plan, deleting, deleteSide, now)));
+   }
+
+
+
+   /*******************************************************************************
+    ** Carry out a prune.
+    *******************************************************************************/
+   public Outcome prune(QRecord plan, String prunedBy) throws QException
+   {
+      Instant       now        = BackupsRuntime.getClock().instant();
+      BackupStorage deleteSide = BackupsRuntime.storageFor(pruneAlias(plan));
+      PrunePlan     prunePlan  = plan(plan);
+      List<String>  messages   = new ArrayList<>();
+
+      int  runsDeleted = 0;
+      int  runsLocked  = 0;
+      long bytesFreed  = 0;
+
+      ///////////////////////////////////////////////////////////////////////
+      // classify every run (the label tells why it is, or was, kept)      //
+      ///////////////////////////////////////////////////////////////////////
+      for(RunPlan run : prunePlan.runs())
+      {
+         Map<String, Serializable> values = new LinkedHashMap<>();
+         values.put("retentionClass", run.decision().keep() ? run.decision().retentionClass().name() : "NONE");
+         if(run.lockedUntil() != null)
+         {
+            values.put("lockedUntil", run.lockedUntil());
+         }
+         catalog.update(BackupsTables.RUN, run.runId(), values);
+      }
+
+      for(RunPlan run : prunePlan.runs())
+      {
+         if(run.decision().keep())
+         {
+            continue;
+         }
+         if(run.lockedUntil() != null)
+         {
+            runsLocked++;
+            messages.add("Run " + run.runKey() + " kept: locked until " + run.lockedUntil());
+            continue;
+         }
+
+         Instant lockedLate = deleteAll(deleteSide, run.objects(), now);
+         if(lockedLate != null)
+         {
+            runsLocked++;
+            messages.add("Run " + run.runKey() + " kept: locked until " + lockedLate);
+            continue;
+         }
+         Map<String, Serializable> done = new LinkedHashMap<>();
+         done.put("prunedDate", now);
+         done.put("prunedByUserId", prunedBy);
+         done.put("retentionClass", "NONE");
+         catalog.update(BackupsTables.RUN, run.runId(), done);
+         runsDeleted++;
+         bytesFreed += run.bytes();
+      }
+
+      //////////////////////////////////
+      // debris is always deleted     //
+      //////////////////////////////////
+      int debrisDeleted = 0;
+      int debrisLocked  = 0;
+      for(DebrisPlan debris : prunePlan.debris())
+      {
+         if(debris.lockedUntil() != null)
+         {
+            debrisLocked++;
+            messages.add("Debris " + debris.prefix() + " kept: locked until " + debris.lockedUntil());
+            continue;
+         }
+         Instant lockedLate = deleteAll(deleteSide, debris.objects(), now);
+         if(lockedLate != null)
+         {
+            debrisLocked++;
+            messages.add("Debris " + debris.prefix() + " kept: locked until " + lockedLate);
+            continue;
+         }
+         debrisDeleted++;
+         bytesFreed += debris.bytes();
+         if(debris.runId() != null)
+         {
+            catalog.update(BackupsTables.RUN, debris.runId(), Map.of("prunedDate", now, "prunedByUserId", prunedBy));
+         }
+      }
+
+      ////////////////////////////////////////////////////////////////////////
+      // the file store: planned again, after the runs went, then swept      //
+      ////////////////////////////////////////////////////////////////////////
+      SweepPlan sweep      = sweepPlan(plan, Set.of(), deleteSide, now);
+      int       filesSwept = 0;
+      int       filesLocked = 0;
+      for(Map.Entry<Integer, Integer> entry : sweep.referenceCounts().entrySet())
+      {
+         catalog.update(BackupsTables.FILE_OBJECT, entry.getKey(), Map.of("referenceCount", entry.getValue()));
+      }
+      if(sweep.deferredReason() != null)
+      {
+         messages.add("File store sweep skipped: " + sweep.deferredReason());
+      }
+      for(SweepCandidate candidate : sweep.candidates())
+      {
+         BackupStorage.DeleteResult result = candidate.lockedUntil() != null ? BackupStorage.DeleteResult.LOCKED : deleteSide.deletePermanent(candidate.objectKey());
+         if(result == BackupStorage.DeleteResult.LOCKED)
+         {
+            filesLocked++;
+            Instant until = candidate.lockedUntil() != null ? candidate.lockedUntil() : deleteSide.head(candidate.objectKey()).map(StoredObjectInfo::lockedUntil).orElse(null);
+            if(until != null)
+            {
+               catalog.update(BackupsTables.FILE_OBJECT, candidate.fileObjectId(), Map.of("lockedUntil", until));
+            }
+            messages.add("File " + candidate.sha256().substring(0, Math.min(12, candidate.sha256().length())) + " kept: locked until " + until);
+         }
+         else
+         {
+            filesSwept++;
+            bytesFreed += candidate.size();
+            catalog.update(BackupsTables.FILE_OBJECT, candidate.fileObjectId(), Map.of("deletedDate", now, "referenceCount", 0));
+         }
+      }
+
+      audit(plan, prunedBy, runsDeleted, debrisDeleted, filesSwept, bytesFreed);
+      return (new Outcome(runsDeleted, runsLocked, debrisDeleted, debrisLocked, filesSwept, filesLocked, bytesFreed, messages));
+   }
+
+
+
+   /*******************************************************************************
+    ** Delete objects, the manifest last (so a half-deleted run is not mistaken for
+    ** a usable one).  Returns the latest lock date when something turned out to be
+    ** locked after all, else null.
+    *******************************************************************************/
+   private Instant deleteAll(BackupStorage storage, List<StoredObjectInfo> objects, Instant now)
+   {
+      List<StoredObjectInfo> ordered = new ArrayList<>(objects);
+      ordered.sort(Comparator.comparing((StoredObjectInfo o) -> o.key().endsWith("/manifest.json")).thenComparing(StoredObjectInfo::key));
+      Instant locked = null;
+      for(StoredObjectInfo object : ordered)
+      {
+         BackupStorage.DeleteResult result = storage.deletePermanent(object.key());
+         if(result == BackupStorage.DeleteResult.LOCKED)
+         {
+            Instant until = storage.head(object.key()).map(StoredObjectInfo::lockedUntil).orElse(now);
+            if(locked == null || (until != null && until.isAfter(locked)))
+            {
+               locked = until == null ? now : until;
+            }
+         }
+      }
+      return (locked);
+   }
+
+
+
+   /*******************************************************************************
+    ** Prefixes of the plan that never got a manifest and are old enough.
+    *******************************************************************************/
+   private List<DebrisPlan> debris(QRecord plan, List<QRecord> allRuns, BackupStorage deleteSide, Instant now) throws QException
+   {
+      String prefix = planPrefix(plan);
+      Map<String, List<StoredObjectInfo>> groups = new LinkedHashMap<>();
+      for(StoredObjectInfo object : deleteSide.list(prefix))
+      {
+         String[] parts = object.key().substring(prefix.length()).split("/", 5);
+         if(parts.length < 5)
+         {
+            continue;
+         }
+         groups.computeIfAbsent(prefix + parts[0] + "/" + parts[1] + "/" + parts[2] + "/" + parts[3] + "/", k -> new ArrayList<>()).add(object);
+      }
+
+      Map<String, QRecord> runsByPrefix = new LinkedHashMap<>();
+      for(QRecord run : allRuns)
+      {
+         if(run.getValueString("runPrefix") != null)
+         {
+            runsByPrefix.put(run.getValueString("runPrefix"), run);
+         }
+      }
+
+      List<DebrisPlan> debris = new ArrayList<>();
+      for(Map.Entry<String, List<StoredObjectInfo>> group : groups.entrySet())
+      {
+         List<StoredObjectInfo> objects = group.getValue();
+         if(objects.stream().anyMatch(o -> o.key().equals(group.getKey() + "manifest.json")))
+         {
+            continue;
+         }
+         QRecord run = runsByPrefix.get(group.getKey());
+         if(run != null && (List.of("RUNNING", "QUEUED", "CANCELLING").contains(run.getValueString("status")) && isLive(run, now)))
+         {
+            continue;
+         }
+         Instant newest = objects.stream().map(StoredObjectInfo::lastModified).max(Comparator.naturalOrder()).orElse(now);
+         if(newest.isAfter(now.minus(Duration.ofDays(PruneCalculator.DEBRIS_DAYS))))
+         {
+            continue;
+         }
+         debris.add(new DebrisPlan(group.getKey(), run == null ? null : run.getValueInteger("id"), objects, objects.stream().mapToLong(StoredObjectInfo::size).sum(), latestLock(objects, now)));
+      }
+      return (debris);
+   }
+
+
+
+   /*******************************************************************************
+    ** Mark and sweep the content-addressed file store for the plan bucket.
+    **
+    ** @param alsoDeleting runs about to be deleted: they do not count as retained
+    *******************************************************************************/
+   private SweepPlan sweepPlan(QRecord plan, Set<Integer> alsoDeleting, BackupStorage deleteSide, Instant now) throws QException
+   {
+      String        alias       = plan.getValueString("storageBucketAlias");
+      BackupStorage readSide    = BackupsRuntime.storageFor(alias);
+      Map<String, Integer> counts = new LinkedHashMap<>();
+
+      ///////////////////////////////////////////////////////////////////////////
+      // mark: every file the file indexes of retained runs list, counted once   //
+      // per run                                                                 //
+      ///////////////////////////////////////////////////////////////////////////
+      List<QRecord> allRuns = catalog.query(BackupsTables.RUN, new QQueryFilter(new QFilterCriteria("storageBucketAlias", QCriteriaOperator.EQUALS, alias)));
+      Map<Integer, QRecord> byId = new LinkedHashMap<>();
+      for(QRecord run : allRuns)
+      {
+         byId.put(run.getValueInteger("id"), run);
+      }
+      String unreadable = null;
+      for(QRecord run : allRuns)
+      {
+         if(run.getValue("prunedDate") != null || !isSuccessful(run) || alsoDeleting.contains(run.getValueInteger("id")))
+         {
+            continue;
+         }
+         try
+         {
+            for(String sha : filesIndexed(run, readSide))
+            {
+               counts.merge(sha, 1, Integer::sum);
+            }
+         }
+         catch(Exception e)
+         {
+            unreadable = "the file index of run " + run.getValueString("runKey") + " could not be read (" + e.getMessage() + ")";
+            break;
+         }
+      }
+
+      List<QRecord> fileObjects = catalog.query(BackupsTables.FILE_OBJECT, new QQueryFilter(new QFilterCriteria("storageBucketAlias", QCriteriaOperator.EQUALS, alias)));
+      Map<Integer, Integer> newCounts = new LinkedHashMap<>();
+      for(QRecord file : fileObjects)
+      {
+         if(file.getValue("deletedDate") == null)
+         {
+            Integer count = counts.getOrDefault(file.getValueString("sha256"), 0);
+            if(!count.equals(file.getValueInteger("referenceCount")))
+            {
+               newCounts.put(file.getValueInteger("id"), count);
+            }
+         }
+      }
+
+      String deferred = unreadable;
+      if(deferred == null)
+      {
+         boolean running = catalog.liveRuns(now.minus(BackupRunCoordinator.HEARTBEAT_STALE)).stream().anyMatch(r -> alias.equals(r.getValueString("storageBucketAlias")))
+            || !catalog.queuedRuns().isEmpty();
+         if(running)
+         {
+            deferred = "a backup is running or queued, and its files are not in any file index yet";
+         }
+      }
+      if(unreadable != null)
+      {
+         newCounts.clear();
+      }
+      if(deferred != null)
+      {
+         return (new SweepPlan(deferred, newCounts, List.of()));
+      }
+
+      /////////////////////////////////////////////////////////////////////////////////////////////
+      // sweep: live objects nothing retained references; not those a recently failed (and so     //
+      // perhaps resumable) run registered, and not locked ones, which are reported               //
+      /////////////////////////////////////////////////////////////////////////////////////////////
+      List<SweepCandidate> candidates = new ArrayList<>();
+      for(QRecord file : fileObjects)
+      {
+         if(file.getValue("deletedDate") != null || counts.containsKey(file.getValueString("sha256")))
+         {
+            continue;
+         }
+         QRecord registeredBy = byId.get(file.getValueInteger("lastReferencedRunId"));
+         if(registeredBy != null && registeredBy.getValue("prunedDate") == null && !isSuccessful(registeredBy) && startOf(registeredBy).isAfter(now.minus(Duration.ofDays(PruneCalculator.DEBRIS_DAYS))))
+         {
+            continue;
+         }
+         Instant lockedUntil = file.getValueInstant("lockedUntil");
+         if(lockedUntil == null || !lockedUntil.isAfter(now))
+         {
+            lockedUntil = deleteSide.head(file.getValueString("objectKey")).map(StoredObjectInfo::lockedUntil).filter(until -> until != null && until.isAfter(now)).orElse(null);
+         }
+         candidates.add(new SweepCandidate(file.getValueInteger("id"), file.getValueString("sha256"), file.getValueString("objectKey"),
+            file.getValueLong("sizeBytes") == null ? 0 : file.getValueLong("sizeBytes"), lockedUntil));
+      }
+      return (new SweepPlan(null, newCounts, candidates));
+   }
+
+
+
+   /*******************************************************************************
+    ** The hashes a run file index lists as stored (copied or deduplicated).
+    *******************************************************************************/
+   private Set<String> filesIndexed(QRecord run, BackupStorage readSide) throws Exception
+   {
+      Set<String> hashes = new LinkedHashSet<>();
+      for(QRecord artifact : catalog.query(BackupsTables.ARTIFACT, new QQueryFilter(
+         new QFilterCriteria("backupRunId", QCriteriaOperator.EQUALS, run.getValueInteger("id")),
+         new QFilterCriteria("artifactType", QCriteriaOperator.EQUALS, "FILE_OBJECTS_INDEX"))))
+      {
+         String key = run.getValueString("runPrefix") + artifact.getValueString("objectKey");
+         try(InputStream raw = readSide.openRead(key);
+             InputStream in = "GZIP".equals(artifact.getValueString("compression")) ? new GZIPInputStream(raw) : raw;
+             BufferedReader reader = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8)))
+         {
+            String line;
+            while((line = reader.readLine()) != null)
+            {
+               if(line.isBlank())
+               {
+                  continue;
+               }
+               JsonNode node   = MAPPER.readTree(line);
+               String   status = node.path("status").asText("");
+               String   sha    = node.path("sha256").asText(null);
+               if(sha != null && ("COPIED".equals(status) || "DEDUPLICATED".equals(status)))
+               {
+                  hashes.add(sha);
+               }
+            }
+         }
+      }
+      return (hashes);
+   }
+
+
+
+   /*******************************************************************************
+    ** Runs a restore used in the last 30 days, as its source or its safety backup.
+    *******************************************************************************/
+   private Set<Integer> restoreProtected(Instant now) throws QException
+   {
+      Instant      cutoff = now.minus(Duration.ofDays(PruneCalculator.RESTORE_PROTECTION_DAYS));
+      Set<Integer> ids    = new HashSet<>();
+      for(QRecord restore : catalog.query(BackupsTables.RESTORE_RUN, new QQueryFilter()))
+      {
+         Instant when = restore.getValueInstant("startDate") != null ? restore.getValueInstant("startDate") : restore.getValueInstant("createDate");
+         if(when == null || when.isAfter(cutoff))
+         {
+            if(restore.getValueInteger("sourceBackupRunId") != null)
+            {
+               ids.add(restore.getValueInteger("sourceBackupRunId"));
+            }
+            if(restore.getValueInteger("preRestoreBackupRunId") != null)
+            {
+               ids.add(restore.getValueInteger("preRestoreBackupRunId"));
+            }
+         }
+      }
+      return (ids);
+   }
+
+
+
+   /*******************************************************************************
+    ** One summary audit per prune.  Auditing must never make a prune fail.
+    *******************************************************************************/
+   private void audit(QRecord plan, String prunedBy, int runs, int debris, int files, long bytes)
+   {
+      try
+      {
+         AuditAction.execute(BackupsTables.PLAN, plan.getValueInteger("id"), Map.of(), "Prune by " + prunedBy + ": " + runs + " runs, " + debris
+            + " debris prefixes and " + files + " stored files deleted (" + bytes + " bytes)");
+      }
+      catch(Exception e)
+      {
+         LOG.debug("Could not audit a prune", e, logPair("plan", plan.getValueString("planKey")));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static Instant latestLock(List<StoredObjectInfo> objects, Instant now)
+   {
+      return (objects.stream().map(StoredObjectInfo::lockedUntil).filter(until -> until != null && until.isAfter(now)).max(Comparator.naturalOrder()).orElse(null));
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static boolean isSuccessful(QRecord run)
+   {
+      return (("SUCCEEDED".equals(run.getValueString("status")) || "SUCCEEDED_WITH_WARNINGS".equals(run.getValueString("status"))) && run.getValueString("manifestKey") != null);
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static boolean isVerified(QRecord run)
+   {
+      return (List.of("STORAGE_OK", "CONTENT_OK", "TEST_RESTORE_OK").contains(run.getValueString("verificationStatus")));
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static boolean isLive(QRecord run, Instant now)
+   {
+      Instant heartbeat = run.getValueInstant("heartbeatDate");
+      return (heartbeat != null && heartbeat.isAfter(now.minus(BackupRunCoordinator.HEARTBEAT_STALE)));
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static Instant startOf(QRecord run)
+   {
+      return (run.getValueInstant("startDate") != null ? run.getValueInstant("startDate") : run.getValueInstant("createDate"));
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static int intOf(QRecord plan, String field, int defaultValue)
+   {
+      return (plan.getValueInteger(field) == null ? defaultValue : plan.getValueInteger(field));
+   }
+
+
+
+   /*******************************************************************************
+    ** The alias Prune deletes through: the prune alias, else the backup alias.
+    *******************************************************************************/
+   public static String pruneAlias(QRecord plan)
+   {
+      String alias = plan.getValueString("pruneBucketAlias");
+      return (alias == null || alias.isBlank() ? plan.getValueString("storageBucketAlias") : alias);
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static ZoneId zoneOf(QRecord plan)
+   {
+      try
+      {
+         String id = plan.getValueString("cronTimeZoneId");
+         return (id == null || id.isBlank() ? ZoneId.of("UTC") : ZoneId.of(id));
+      }
+      catch(Exception e)
+      {
+         return (ZoneId.of("UTC"));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** The prefix every run of the plan is under, ending in a slash.
+    *******************************************************************************/
+   public static String planPrefix(QRecord plan)
+   {
+      BackupsQBitConfig config   = BackupsRuntime.getConfig();
+      String            instance = config != null && config.getInstanceName() != null ? config.getInstanceName() : "qqq";
+      String            prefix   = plan.getValueString("prefix") == null || plan.getValueString("prefix").isBlank() ? "qqq-backups" : plan.getValueString("prefix");
+      String            trimmed  = prefix;
+      while(trimmed.startsWith("/"))
+      {
+         trimmed = trimmed.substring(1);
+      }
+      while(trimmed.endsWith("/"))
+      {
+         trimmed = trimmed.substring(0, trimmed.length() - 1);
+      }
+      return (trimmed + "/" + instance + "/" + plan.getValueString("planKey") + "/");
+   }
+}

src/main/java/com/kingsrook/qbits/backups/prune/PruneCalculator.java

@@ -0,0 +1,221 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.prune;
+
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.ZoneId;
+import java.time.temporal.IsoFields;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.EnumMap;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+
+/*******************************************************************************
+ ** The grandfather-father-son keep-set (spec 10.3).  Pure: it works on run
+ ** summaries and a clock, so tests can use a synthetic one.
+ **
+ ** The rules, over the successful, not yet pruned runs, newest first:
+ ** - DAILY: the newest run of each of the last retainDaily calendar days that have
+ **   a run
+ ** - WEEKLY: the newest run of each of the last retainWeekly ISO weeks that have one
+ ** - MONTHLY: the newest run of each of the last retainMonthly months that have one
+ ** - YEARLY: the newest run of each of the last retainYearly years that have one
+ **   ("last N" counts the periods that have a run, so a quiet week does not use up
+ **   the allowance)
+ ** - the newest retainMinimum runs are always kept
+ ** - the newest run that passed any verification is always kept
+ ** - a run a restore used in the last 30 days (as its source, or its pre-restore
+ **   safety backup) is never deleted
+ **
+ ** A run kept by several rules is labelled with the longest lived one (YEARLY, then
+ ** MONTHLY, WEEKLY, DAILY); a run kept only by a guard (minimum, verified, restore)
+ ** is labelled NONE and carries the reasons.  A run that is not kept is labelled NONE.
+ *******************************************************************************/
+public final class PruneCalculator
+{
+   public static final int RESTORE_PROTECTION_DAYS = 30;
+
+   /** Prefixes with no manifest older than this are debris and are always deleted. */
+   public static final int DEBRIS_DAYS = 7;
+
+
+
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private PruneCalculator()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** The retention classes, longest lived first.
+    *******************************************************************************/
+   public enum RetentionClass
+   {
+      YEARLY, MONTHLY, WEEKLY, DAILY, NONE
+   }
+
+
+
+   /*******************************************************************************
+    ** Why a run is kept, besides the class.
+    *******************************************************************************/
+   public enum Guard
+   {
+      MINIMUM, NEWEST_VERIFIED, RESTORE_REFERENCE
+   }
+
+
+
+   /*******************************************************************************
+    ** What the calculator needs to know about one run.
+    **
+    ** @param id the run id
+    ** @param startDate when the run started
+    ** @param verified whether any verification passed for the run
+    ** @param usedByRecentRestore whether a restore in the last 30 days used the run
+    *******************************************************************************/
+   public record RunInfo(Integer id, Instant startDate, boolean verified, boolean usedByRecentRestore)
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** The retention numbers of a plan.
+    *******************************************************************************/
+   public record Policy(int daily, int weekly, int monthly, int yearly, int minimum)
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** The decision for one run.
+    *******************************************************************************/
+   public record Decision(Integer id, boolean keep, RetentionClass retentionClass, Set<Guard> guards)
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** Decide every run.  The result keeps the input order of runs newest first.
+    *******************************************************************************/
+   public static List<Decision> calculate(List<RunInfo> runs, Policy policy, ZoneId zone)
+   {
+      List<RunInfo> sorted = new ArrayList<>(runs);
+      sorted.sort(Comparator.comparing(RunInfo::startDate).reversed().thenComparing(RunInfo::id, Comparator.reverseOrder()));
+
+      Map<Integer, RetentionClass> best   = new LinkedHashMap<>();
+      Map<Integer, Set<Guard>>     guards = new LinkedHashMap<>();
+      for(RunInfo run : sorted)
+      {
+         best.put(run.id(), RetentionClass.NONE);
+         guards.put(run.id(), new LinkedHashSet<>());
+      }
+
+      mark(sorted, policy.daily(), RetentionClass.DAILY, run -> run.startDate().atZone(zone).toLocalDate().toString(), best);
+      mark(sorted, policy.weekly(), RetentionClass.WEEKLY, run -> isoWeek(run.startDate().atZone(zone).toLocalDate()), best);
+      mark(sorted, policy.monthly(), RetentionClass.MONTHLY, run -> run.startDate().atZone(zone).getYear() + "-" + run.startDate().atZone(zone).getMonthValue(), best);
+      mark(sorted, policy.yearly(), RetentionClass.YEARLY, run -> String.valueOf(run.startDate().atZone(zone).getYear()), best);
+
+      for(int i = 0; i < Math.min(policy.minimum(), sorted.size()); i++)
+      {
+         guards.get(sorted.get(i).id()).add(Guard.MINIMUM);
+      }
+      sorted.stream().filter(RunInfo::verified).findFirst().ifPresent(run -> guards.get(run.id()).add(Guard.NEWEST_VERIFIED));
+      sorted.stream().filter(RunInfo::usedByRecentRestore).forEach(run -> guards.get(run.id()).add(Guard.RESTORE_REFERENCE));
+
+      List<Decision> decisions = new ArrayList<>();
+      for(RunInfo run : sorted)
+      {
+         RetentionClass retentionClass = best.get(run.id());
+         boolean        keep           = retentionClass != RetentionClass.NONE || !guards.get(run.id()).isEmpty();
+         decisions.add(new Decision(run.id(), keep, retentionClass, guards.get(run.id())));
+      }
+      return (decisions);
+   }
+
+
+
+   /*******************************************************************************
+    ** Count the kept runs by class (NONE counts the guard-kept ones).
+    *******************************************************************************/
+   public static Map<RetentionClass, Integer> keptByClass(List<Decision> decisions)
+   {
+      Map<RetentionClass, Integer> counts = new EnumMap<>(RetentionClass.class);
+      for(Decision decision : decisions)
+      {
+         if(decision.keep())
+         {
+            counts.merge(decision.retentionClass(), 1, Integer::sum);
+         }
+      }
+      return (counts);
+   }
+
+
+
+   /*******************************************************************************
+    ** Keep the newest run of each of the first `limit` distinct periods.
+    *******************************************************************************/
+   private static void mark(List<RunInfo> newestFirst, int limit, RetentionClass retentionClass, java.util.function.Function<RunInfo, String> periodOf, Map<Integer, RetentionClass> best)
+   {
+      if(limit <= 0)
+      {
+         return;
+      }
+      Set<String> seen = new LinkedHashSet<>();
+      for(RunInfo run : newestFirst)
+      {
+         String period = periodOf.apply(run);
+         if(seen.contains(period))
+         {
+            continue;
+         }
+         if(seen.size() >= limit)
+         {
+            return;
+         }
+         seen.add(period);
+         if(retentionClass.ordinal() < best.get(run.id()).ordinal())
+         {
+            best.put(run.id(), retentionClass);
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** The ISO week of a date, as week-based-year-week.
+    *******************************************************************************/
+   private static String isoWeek(LocalDate date)
+   {
+      return (date.get(IsoFields.WEEK_BASED_YEAR) + "-W" + date.get(IsoFields.WEEK_OF_WEEK_BASED_YEAR));
+   }
+}

src/main/java/com/kingsrook/qbits/backups/run/BackupCatalog.java

@@ -26,10 +26,12 @@ import java.util.Map;
 import java.util.Optional;
 import com.kingsrook.qbits.backups.files.FileCopier;
 import com.kingsrook.qbits.backups.model.BackupsTables;
+import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
 import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
 import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
 import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
 import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
@@ -235,6 +237,72 @@ public class BackupCatalog
 
 
 
+   /*******************************************************************************
+    ** Every run (any plan, or an ad-hoc export) that is RUNNING with a heartbeat
+    ** newer than the cutoff: the live leases.
+    *******************************************************************************/
+   public List<QRecord> liveRuns(Instant heartbeatCutoff) throws QException
+   {
+      List<QRecord> live = new ArrayList<>();
+      for(QRecord run : query(BackupsTables.RUN, new QQueryFilter(new QFilterCriteria("status", QCriteriaOperator.IN, List.of("RUNNING", "CANCELLING")))))
+      {
+         Instant heartbeat = run.getValueInstant("heartbeatDate");
+         if(heartbeat != null && heartbeat.isAfter(heartbeatCutoff))
+         {
+            live.add(run);
+         }
+      }
+      return (live);
+   }
+
+
+
+   /*******************************************************************************
+    ** The QUEUED runs, oldest first.
+    *******************************************************************************/
+   public List<QRecord> queuedRuns() throws QException
+   {
+      return (query(BackupsTables.RUN, new QQueryFilter(new QFilterCriteria("status", QCriteriaOperator.EQUALS, "QUEUED")).withOrderBy(new QFilterOrderBy("id", true))));
+   }
+
+
+
+   /*******************************************************************************
+    ** Whether somebody asked for a run to be cancelled (its row says CANCELLING).
+    ** The engine checks this every page.
+    *******************************************************************************/
+   public boolean isCancelRequested(Integer runId)
+   {
+      try
+      {
+         return (get(BackupsTables.RUN, runId).map(r -> "CANCELLING".equals(r.getValueString("status"))).orElse(false));
+      }
+      catch(QException e)
+      {
+         return (false);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Delete catalog rows by id (used only for artifact rows of a part that is
+    ** being rewritten).
+    *******************************************************************************/
+   public void deleteByIds(String table, List<Integer> ids) throws QException
+   {
+      if(ids.isEmpty())
+      {
+         return;
+      }
+      DeleteInput input = new DeleteInput(table);
+      input.setPrimaryKeys(new ArrayList<>(ids));
+      input.withPermissionBypass("backups catalog");
+      new DeleteAction().execute(input);
+   }
+
+
+
    /*******************************************************************************
     ** The runs that are RUNNING for a plan, with a heartbeat newer than the cutoff.
     *******************************************************************************/

src/main/java/com/kingsrook/qbits/backups/run/BackupEngine.java

@@ -53,6 +53,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
 import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.utils.ValueUtils;
 import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
 
 
@@ -90,10 +91,10 @@ public class BackupEngine
    {
       BackupRunResult result  = new BackupRunResult();
       Instant         started = request.getClock().instant();
-      String          runKey  = RUN_KEY_FORMAT.format(started) + "-" + UUID.randomUUID().toString().substring(0, 4);
+      String          runKey  = request.getResume() != null ? request.getResume().runKey() : RUN_KEY_FORMAT.format(started) + "-" + UUID.randomUUID().toString().substring(0, 4);
       String          planPrefix = request.getSettings().string("prefix", "qqq-backups");
       String          instancePrefix = trimSlashes(planPrefix) + "/" + request.getInstanceName() + "/";
-      String          runPrefix  = instancePrefix + request.getSettings().planKey() + "/" + DATE_PATH.format(started) + "/" + runKey + "/";
+      String          runPrefix  = request.getResume() != null ? request.getResume().runPrefix() : instancePrefix + request.getSettings().planKey() + "/" + DATE_PATH.format(started) + "/" + runKey + "/";
       result.setRunKey(runKey);
       result.setRunPrefix(runPrefix);
 
@@ -116,7 +117,24 @@ public class BackupEngine
          runValues.put("encryptionKeyRef", request.getEncryptionKeyRef());
          runValues.put("encryptionKeyFingerprint", request.getEncryptionKeyFingerprint());
          runValues.put("verificationStatus", "UNVERIFIED");
-         result.setRunId(catalog.insert(BackupsTables.RUN, runValues));
+         if(request.getResume() != null)
+         {
+            ///////////////////////////////////////////////////////////////////////////////
+            // a resume carries on the original attempt: keep its start, clear its error //
+            ///////////////////////////////////////////////////////////////////////////////
+            runValues.remove("startDate");
+            runValues.put("endDate", null);
+            runValues.put("errorMessage", null);
+         }
+         if(request.getExistingRunId() != null)
+         {
+            catalog.update(BackupsTables.RUN, request.getExistingRunId(), runValues);
+            result.setRunId(request.getExistingRunId());
+         }
+         else
+         {
+            result.setRunId(catalog.insert(BackupsTables.RUN, runValues));
+         }
       }
       catch(Exception e)
       {
@@ -170,10 +188,32 @@ public class BackupEngine
          throw (new IllegalStateException("The sensitive field policy INCLUDE_ENCRYPTED_ONLY requires client-side encryption."));
       }
       String consistency = consistencyFor(settings.string("consistency", "SNAPSHOT_PREFERRED"), result);
+      catalog.update(BackupsTables.RUN, result.getRunId(), Map.of("consistencyAchieved", result.getConsistencyAchieved()));
+
+      //////////////////////////////////////////////////////////////////////////////////////
+      // object lock is applied at write time, to every object of the run (spec 10.3)     //
+      //////////////////////////////////////////////////////////////////////////////////////
+      String  lockMode    = settings.string("objectLockMode", "NONE");
+      Instant retainUntil = null;
+      if(!"NONE".equals(lockMode))
+      {
+         if(!storage.capabilities().contains(BackupStorage.StorageCapability.OBJECT_LOCK))
+         {
+            throw (new IllegalStateException("The plan asks for object lock " + lockMode + ", but the storage bucket does not support object lock."));
+         }
+         retainUntil = started.plus(java.time.Duration.ofDays(Math.max(1, settings.integer("objectLockDays", 1))));
+      }
+      else
+      {
+         lockMode = null;
+      }
+      final String  fileLockMode    = lockMode;
+      final Instant fileRetainUntil = retainUntil;
 
       Function<String, BackupStorage> sourceResolver = request.getSourceResolver() != null ? request.getSourceResolver() : alias -> storage;
       FileCopier copier = new FileCopier(sourceResolver, request.getBucketAlias(), storage, instancePrefix + "_files/sha256/", request.getTransform(),
-         catalog.fileStoreRegistry(request.getBucketAlias(), result.getRunId(), request.getEncryptionKeyFingerprint()), !copyFiles, result.getRunKey());
+         catalog.fileStoreRegistry(request.getBucketAlias(), result.getRunId(), request.getEncryptionKeyFingerprint()), !copyFiles, result.getRunKey())
+         .withObjectLock(lockMode, retainUntil);
 
       List<BackupManifest.TableEntry> manifestTables = new ArrayList<>();
       List<String>                    checksums      = new ArrayList<>();
@@ -200,65 +240,131 @@ public class BackupEngine
             List<BackupManifest.ExcludedField> excluded = excludedFields(table, resolved, sensitivePolicy, settings.string("heavyFieldPolicy", "INLINE"));
             List<String> excludedNames = excluded.stream().map(BackupManifest.ExcludedField::name).toList();
 
+            ResumeState.TableState prior = request.getResume() == null ? null : request.getResume().table(table.getName());
+
             List<TableArtifactWriter> writers = new ArrayList<>();
             String tablePrefix = runPrefix + "tables/" + table.getName() + "/";
-            for(String f : "JSONL_AND_CSV".equals(format) ? List.of("JSONL", "CSV") : List.of("CSV".equals(format) ? "CSV" : "JSONL"))
-            {
-               writers.add(new TableArtifactWriter(storage, tablePrefix, table, excludedNames, f, compression, settings.targetPartBytes(), MAX_ROWS_PER_PART,
-                  request.getTransform(), result.getRunKey()));
-            }
+            Map<TableArtifactWriter, Integer> lastArtifactIds = new LinkedHashMap<>();
 
             List<FileReference> references = new ArrayList<>();
             long[]              counter    = { 0 };
             Instant             deadline   = started.plusMillis(maxMillis);
-            Throwable[]         writeError = { null };
+            String              rowsHash   = null;
+            List<TableArtifactWriter.PartResult> allResults = new ArrayList<>();
+            boolean wantsReferences = copyFiles || "REFERENCE_ONLY".equals(settings.fileObjectPolicy());
 
-            request.getReader().read(table, resolved.filterJson(), null, settings.pageSize(), null, record ->
+            if(prior != null && prior.complete())
             {
-               try
+               ////////////////////////////////////////////////////////////////////////////////
+               // resume: this table finished before the failure.  Keep its parts, and only  //
+               // walk its rows again for the file references (the index is rebuilt)         //
+               ////////////////////////////////////////////////////////////////////////////////
+               if(wantsReferences)
                {
-                  for(TableArtifactWriter writer : writers)
+                  collectReferences(request, table, resolved.filterJson(), null, references);
+               }
+               allResults.addAll(prior.parts());
+               rowsHash = prior.rowsHash();
+               counter[0] = prior.rows();
+            }
+            else
+            {
+               for(String f : "JSONL_AND_CSV".equals(format) ? List.of("JSONL", "CSV") : List.of("CSV".equals(format) ? "CSV" : "JSONL"))
+               {
+                  TableArtifactWriter writer = new TableArtifactWriter(storage, tablePrefix, table, excludedNames, f, compression, settings.targetPartBytes(), MAX_ROWS_PER_PART,
+                     request.getTransform(), result.getRunKey()).withObjectLock(fileLockMode, fileRetainUntil);
+                  writer.withPartListener(part ->
                   {
-                     writer.write(record);
-                  }
+                     try
+                     {
+                        lastArtifactIds.put(writer, catalog.insertArtifact(result.getRunId(), "TABLE_DATA", table.getName(), part.partNumber(), part.key().substring(runPrefix.length()),
+                           part.format(), part.compression(), part.encrypted(), part.rows(), part.bytesRaw(), part.bytesStored(), part.sha256Stored(), part.sha256Raw(), null,
+                           part.firstKey(), part.lastKey(), TableFingerprints.of(table, backendType)));
+                     }
+                     catch(Exception e)
+                     {
+                        throw (new IllegalStateException("Could not record a part of " + table.getName() + ": " + e.getMessage(), e));
+                     }
+                  });
+                  writers.add(writer);
                }
-               catch(IOException e)
-               {
-                  throw (new IllegalStateException("Could not write a row of " + table.getName() + ": " + e.getMessage(), e));
-               }
-               if(copyFiles || "REFERENCE_ONLY".equals(settings.fileObjectPolicy()))
-               {
-                  for(FileReferenceProvider provider : request.getFileProviders())
-                  {
-                     references.addAll(provider.referencesIn(table, record));
-                  }
-               }
-               counter[0]++;
-               if(counter[0] % settings.pageSize() == 0)
-               {
-                  heartbeat(request, result, deadline, table.getName(), counter[0]);
-               }
-            });
 
-            String       rowsHash = null;
+               Serializable startAfter = null;
+               if(prior != null && !prior.parts().isEmpty())
+               {
+                  boolean canContinue = writers.size() == 1 && "JSONL".equals(format) && !request.getTransform().encrypts();
+                  if(canContinue)
+                  {
+                     TableArtifactWriter writer = writers.get(0);
+                     writer.resumeFrom(prior.parts());
+                     seedRowsHash(storage, runPrefix, prior.parts(), writer);
+                     lastArtifactIds.put(writer, prior.artifactIds().get(prior.artifactIds().size() - 1));
+                     startAfter = ValueUtils.getValueAsFieldType(table.getField(table.getPrimaryKeyField()).getType(), prior.lastKey());
+                     counter[0] = prior.rows();
+                     if(wantsReferences)
+                     {
+                        collectReferences(request, table, resolved.filterJson(), prior.lastKey(), references);
+                     }
+                  }
+                  else
+                  {
+                     /////////////////////////////////////////////////////////////////////////
+                     // CSV, a second format or encryption cannot rebuild the rows hash from //
+                     // the parts: start this table over (its part objects are overwritten)  //
+                     /////////////////////////////////////////////////////////////////////////
+                     catalog.deleteByIds(BackupsTables.ARTIFACT, prior.artifactIds());
+                  }
+               }
+
+               request.getReader().read(table, resolved.filterJson(), startAfter, settings.pageSize(), null, record ->
+               {
+                  try
+                  {
+                     for(TableArtifactWriter writer : writers)
+                     {
+                        writer.write(record);
+                     }
+                  }
+                  catch(IOException e)
+                  {
+                     throw (new IllegalStateException("Could not write a row of " + table.getName() + ": " + e.getMessage(), e));
+                  }
+                  if(wantsReferences)
+                  {
+                     for(FileReferenceProvider provider : request.getFileProviders())
+                     {
+                        references.addAll(provider.referencesIn(table, record));
+                     }
+                  }
+                  counter[0]++;
+                  if(counter[0] % settings.pageSize() == 0)
+                  {
+                     heartbeat(request, result, deadline, table.getName(), counter[0]);
+                  }
+               });
+
+               for(TableArtifactWriter writer : writers)
+               {
+                  allResults.addAll(writer.finish());
+                  rowsHash = writer.rowsHash();
+                  Integer lastId = lastArtifactIds.get(writer);
+                  if(lastId != null)
+                  {
+                     catalog.update(BackupsTables.ARTIFACT, lastId, Map.of("rowsHash", rowsHash));
+                  }
+               }
+            }
+
             List<BackupManifest.Part> parts = new ArrayList<>();
             String fingerprint = TableFingerprints.of(table, backendType);
             fingerprints.add(fingerprint);
-            for(TableArtifactWriter writer : writers)
+            for(TableArtifactWriter.PartResult part : allResults)
             {
-               List<TableArtifactWriter.PartResult> results = writer.finish();
-               rowsHash = writer.rowsHash();
-               for(TableArtifactWriter.PartResult part : results)
-               {
-                  String relative = part.key().substring(runPrefix.length());
-                  parts.add(new BackupManifest.Part(relative, part.rows(), part.bytesRaw(), part.bytesStored(), part.sha256Stored(), part.sha256Raw(), part.format(), part.compression(), part.encrypted()));
-                  checksums.add(part.sha256Stored() + "  " + relative);
-                  bytesStored += part.bytesStored();
-                  bytesRaw += part.bytesRaw();
-                  catalog.insertArtifact(result.getRunId(), "TABLE_DATA", table.getName(), part.partNumber(), relative, part.format(), part.compression(),
-                     part.encrypted(), part.rows(), part.bytesRaw(), part.bytesStored(), part.sha256Stored(), part.sha256Raw(),
-                     part.partNumber() == results.size() ? rowsHash : null, part.firstKey(), part.lastKey(), fingerprint);
-               }
+               String relative = part.key().substring(runPrefix.length());
+               parts.add(new BackupManifest.Part(relative, part.rows(), part.bytesRaw(), part.bytesStored(), part.sha256Stored(), part.sha256Raw(), part.format(), part.compression(), part.encrypted()));
+               checksums.add(part.sha256Stored() + "  " + relative);
+               bytesStored += part.bytesStored();
+               bytesRaw += part.bytesRaw();
             }
             rows += counter[0];
 
@@ -266,7 +372,7 @@ public class BackupEngine
             {
                if(indexWriter == null)
                {
-                  indexWriter = new FileIndexWriter(storage, runPrefix + "files/index-00001.jsonl.gz", result.getRunKey());
+                  indexWriter = new FileIndexWriter(storage, runPrefix + "files/index-00001.jsonl.gz", result.getRunKey(), fileLockMode, fileRetainUntil);
                }
                FileCopier.IndexEntry entry = copier.process(ref);
                indexWriter.write(entry);
@@ -289,8 +395,8 @@ public class BackupEngine
             }
 
             Map<String, String> keyRange = new LinkedHashMap<>();
-            keyRange.put("min", writers.get(0).getFirstKey());
-            keyRange.put("max", writers.get(0).getLastKey());
+            keyRange.put("min", allResults.isEmpty() ? null : allResults.get(0).firstKey());
+            keyRange.put("max", allResults.isEmpty() ? null : allResults.get(allResults.size() - 1).lastKey());
             manifestTables.add(new BackupManifest.TableEntry(table.getName(), resolved.role().name(), table.getBackendName(), table.getPrimaryKeyField(),
                fingerprint, TableFingerprints.descriptor(table, backendType), resolved.filterJson(), excluded, counter[0], rowsHash, keyRange, parts, consistency));
             done++;
@@ -332,7 +438,7 @@ public class BackupEngine
       // checksums, then the manifest LAST (the commit mark) //
       /////////////////////////////////////////////////////////
       byte[] checksumBytes = (String.join("\n", checksums) + "\n").getBytes(StandardCharsets.UTF_8);
-      putBytes(storage, runPrefix + "checksums.sha256", checksumBytes, result.getRunKey());
+      putBytes(storage, runPrefix + "checksums.sha256", checksumBytes, result.getRunKey(), lockMode, retainUntil);
       catalog.insertArtifact(result.getRunId(), "CHECKSUMS", null, 1, "checksums.sha256", "TEXT", "NONE", false, checksums.size(), checksumBytes.length,
          checksumBytes.length, HashingOutputStream.sha256Hex(checksumBytes), HashingOutputStream.sha256Hex(checksumBytes), null, null, null, null);
 
@@ -352,7 +458,7 @@ public class BackupEngine
          new BackupManifest.Totals(manifestTables.size(), rows, bytesStored));
       byte[] manifestBytes = manifest.toJsonBytes();
       String manifestKey   = runPrefix + "manifest.json";
-      putBytes(storage, manifestKey, manifestBytes, result.getRunKey());
+      putBytes(storage, manifestKey, manifestBytes, result.getRunKey(), lockMode, retainUntil);
       if(storage.head(manifestKey).isEmpty())
       {
          throw (new IllegalStateException("The manifest was written but is not visible in the bucket"));
@@ -384,6 +490,7 @@ public class BackupEngine
       extra.put("filesMissing", missing);
       extra.put("fileBytesCopied", fileBytes);
       extra.put("fileBytesReferenced", fileBytesRef);
+      extra.put("lockedUntil", retainUntil);
       extra.put("tablesDone", manifestTables.size());
       extra.put("rowsTotal", rows);
       extra.put("bytesRaw", bytesRaw);
@@ -601,9 +708,14 @@ public class BackupEngine
    /*******************************************************************************
     **
     *******************************************************************************/
-   private static void putBytes(BackupStorage storage, String key, byte[] bytes, String runKey) throws IOException
+   private static void putBytes(BackupStorage storage, String key, byte[] bytes, String runKey, String lockMode, Instant retainUntil) throws IOException
    {
-      try(OutputStream out = storage.openWrite(key, WriteOptions.create().withMetadata("qqq-backup-run", runKey)))
+      WriteOptions options = WriteOptions.create().withMetadata("qqq-backup-run", runKey);
+      if(lockMode != null)
+      {
+         options.withObjectLock(lockMode, retainUntil);
+      }
+      try(OutputStream out = storage.openWrite(key, options))
       {
          out.write(bytes);
       }
@@ -641,6 +753,82 @@ public class BackupEngine
 
 
 
+   /*******************************************************************************
+    ** Walk a table's rows only for the file references they hold (resume: rows that
+    ** an earlier attempt already wrote to parts).  Stops after stopAfterKey (inclusive)
+    ** when given.
+    *******************************************************************************/
+   private void collectReferences(BackupRunRequest request, QTableMetaData table, String filterJson, String stopAfterKey, List<FileReference> references) throws Exception
+   {
+      String pk = table.getPrimaryKeyField();
+      try
+      {
+         request.getReader().read(table, filterJson, null, request.getSettings().pageSize(), null, record ->
+         {
+            if(stopAfterKey != null && !stopAfterKey.equals(String.valueOf(record.getValue(pk))) && keyAfter(record.getValue(pk), stopAfterKey))
+            {
+               throw (new StopReadingException());
+            }
+            for(FileReferenceProvider provider : request.getFileProviders())
+            {
+               references.addAll(provider.referencesIn(table, record));
+            }
+         });
+      }
+      catch(StopReadingException e)
+      {
+         // done: we are past the last key of the written parts
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Whether a primary key value sorts after a key given as text.
+    *******************************************************************************/
+   @SuppressWarnings({ "unchecked", "rawtypes" })
+   private static boolean keyAfter(Serializable value, String lastKeyText)
+   {
+      if(value instanceof Number number)
+      {
+         return (new java.math.BigDecimal(number.toString()).compareTo(new java.math.BigDecimal(lastKeyText)) > 0);
+      }
+      return (String.valueOf(value).compareTo(lastKeyText) > 0);
+   }
+
+
+
+   /*******************************************************************************
+    ** Feed the rows hash of a resumed table from the parts already in storage (JSONL,
+    ** gzip or none, not encrypted: the decoded bytes are the canonical lines).
+    *******************************************************************************/
+   private static void seedRowsHash(BackupStorage storage, String runPrefix, List<TableArtifactWriter.PartResult> parts, TableArtifactWriter writer) throws IOException
+   {
+      byte[] buffer = new byte[65536];
+      for(TableArtifactWriter.PartResult part : parts)
+      {
+         try(java.io.InputStream raw = storage.openRead(part.key()); java.io.InputStream in = "GZIP".equals(part.compression()) ? new java.util.zip.GZIPInputStream(raw) : raw)
+         {
+            int n;
+            while((n = in.read(buffer)) >= 0)
+            {
+               writer.seedRowsHash(buffer, 0, n);
+            }
+         }
+      }
+   }
+
+
+
+   /***************************************************************************
+    ** Thrown from a row sink to stop a reference-only walk.
+    ***************************************************************************/
+   static class StopReadingException extends RuntimeException
+   {
+   }
+
+
+
    /***************************************************************************
     **
     ***************************************************************************/
@@ -680,9 +868,14 @@ public class BackupEngine
       /*******************************************************************************
        ** Constructor.
        *******************************************************************************/
-      FileIndexWriter(BackupStorage storage, String key, String runKey) throws IOException
+      FileIndexWriter(BackupStorage storage, String key, String runKey, String lockMode, Instant retainUntil) throws IOException
       {
-         stored = new HashingOutputStream(storage.openWrite(key, WriteOptions.create().withMetadata("qqq-backup-run", runKey)));
+         WriteOptions options = WriteOptions.create().withMetadata("qqq-backup-run", runKey);
+         if(lockMode != null)
+         {
+            options.withObjectLock(lockMode, retainUntil);
+         }
+         stored = new HashingOutputStream(storage.openWrite(key, options));
          raw = new HashingOutputStream(new GZIPOutputStream(stored));
       }
 

src/main/java/com/kingsrook/qbits/backups/run/BackupResumePlanner.java

@@ -0,0 +1,127 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.run;
+
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.backups.codec.TableArtifactWriter;
+import com.kingsrook.qbits.backups.model.BackupsTables;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterOrderBy;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+
+
+/*******************************************************************************
+ ** Works out whether a failed run can be resumed, and from where (spec 9, Resume).
+ **
+ ** Only a run that read each table on its own (PER_TABLE, or BEST_EFFORT while no
+ ** backend gives a snapshot) can resume: a run that held one snapshot cannot carry
+ ** on across a new one, and says so.  A resume continues the SAME run (same run key
+ ** and prefix): finished tables are kept, and the interrupted table carries on from
+ ** the last key of its last complete part (the catalog records each part the moment
+ ** it is complete).
+ *******************************************************************************/
+public final class BackupResumePlanner
+{
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private BackupResumePlanner()
+   {
+   }
+
+
+
+   /*******************************************************************************
+    ** Why the run cannot be resumed, or null when it can.
+    *******************************************************************************/
+   public static String whyNotResumable(QRecord run, QRecord plan)
+   {
+      String status = run.getValueString("status");
+      if(!List.of("FAILED", "TIMED_OUT", "ABANDONED").contains(status))
+      {
+         return ("Only a failed, timed out or abandoned run can be resumed (this one is " + status + ").");
+      }
+      if(run.getValueInteger("backupPlanId") == null || plan == null)
+      {
+         return ("Only a run of a plan can be resumed. Start a new export instead.");
+      }
+      if(run.getValueString("manifestKey") != null)
+      {
+         return ("This run already finished: it has a manifest.");
+      }
+      String achieved = run.getValueString("consistencyAchieved");
+      String requested = plan.getValueString("consistency");
+      boolean snapshot = "INSTANCE_SNAPSHOT".equals(achieved) || "BACKEND_SNAPSHOT".equals(achieved) || (achieved == null && "SNAPSHOT_REQUIRED".equals(requested));
+      if(snapshot)
+      {
+         return ("This run read from one consistent snapshot, which cannot be carried on after it failed. Run the plan again to take a new backup.");
+      }
+      return (null);
+   }
+
+
+
+   /*******************************************************************************
+    ** What the run already wrote.
+    *******************************************************************************/
+   public static ResumeState plan(QRecord run) throws QException
+   {
+      BackupCatalog catalog = new BackupCatalog();
+      List<QRecord> artifacts = catalog.query(BackupsTables.ARTIFACT, new QQueryFilter(
+         new QFilterCriteria("backupRunId", QCriteriaOperator.EQUALS, run.getValueInteger("id")),
+         new QFilterCriteria("artifactType", QCriteriaOperator.EQUALS, "TABLE_DATA"))
+         .withOrderBy(new QFilterOrderBy("tableName", true)).withOrderBy(new QFilterOrderBy("partNumber", true)).withOrderBy(new QFilterOrderBy("id", true)));
+
+      String runPrefix = run.getValueString("runPrefix");
+      Map<String, List<QRecord>> byTable = new LinkedHashMap<>();
+      for(QRecord artifact : artifacts)
+      {
+         byTable.computeIfAbsent(artifact.getValueString("tableName"), k -> new ArrayList<>()).add(artifact);
+      }
+
+      Map<String, ResumeState.TableState> tables = new LinkedHashMap<>();
+      for(Map.Entry<String, List<QRecord>> entry : byTable.entrySet())
+      {
+         List<TableArtifactWriter.PartResult> parts = new ArrayList<>();
+         List<Integer>                        ids   = new ArrayList<>();
+         long                                 rows  = 0;
+         String                               hash  = null;
+         for(QRecord a : entry.getValue())
+         {
+            parts.add(new TableArtifactWriter.PartResult(a.getValueInteger("partNumber"), runPrefix + a.getValueString("objectKey"), a.getValueLong("rowCount"),
+               a.getValueLong("bytesRaw"), a.getValueLong("bytesStored"), a.getValueString("sha256Raw"), a.getValueString("sha256Stored"),
+               a.getValueString("firstKey"), a.getValueString("lastKey"), Boolean.TRUE.equals(a.getValueBoolean("encrypted")), a.getValueString("format"),
+               a.getValueString("compression")));
+            ids.add(a.getValueInteger("id"));
+            if(a.getValueString("format") != null && a.getValueString("format").equals(entry.getValue().get(0).getValueString("format")))
+            {
+               rows += a.getValueLong("rowCount");
+            }
+            hash = a.getValueString("rowsHash");
+         }
+         tables.put(entry.getKey(), new ResumeState.TableState(hash != null, parts, hash, rows, ids));
+      }
+      return (new ResumeState(run.getValueInteger("id"), run.getValueString("runKey"), runPrefix, tables));
+   }
+}

src/main/java/com/kingsrook/qbits/backups/run/BackupRunCoordinator.java

@@ -0,0 +1,417 @@
+/*
+ * Copyright 2026 Kingsrook, LLC
+ *
+ * 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 com.kingsrook.qbits.backups.run;
+
+
+import java.io.Serializable;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import com.kingsrook.qbits.backups.BackupsQBitConfig;
+import com.kingsrook.qbits.backups.BackupsRuntime;
+import com.kingsrook.qbits.backups.alerts.BackupAlertSink;
+import com.kingsrook.qbits.backups.model.BackupsTables;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.logging.QLogger;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
+
+
+/*******************************************************************************
+ ** Decides whether a run may start, and starts it (spec 9.1): one run per plan at
+ ** a time, and at most maxConcurrentRuns across the instance.
+ **
+ ** The lease is the run row itself.  A run is inserted QUEUED, and flipped to
+ ** RUNNING only when no other run of the plan has a live heartbeat (newer than
+ ** HEARTBEAT_STALE) and the instance has a free slot.  The check and the flip happen
+ ** under one lock in this JVM; because the lock cannot span pods, the flip is
+ ** followed by a second look, and when two runs of the same plan are RUNNING the
+ ** one with the higher id gives way.  A second trigger while one is running is not
+ ** silent: it leaves a CANCELLED run row saying SKIPPED_CONCURRENT and raises a
+ ** warning.  When the instance is full the run stays QUEUED and is started, oldest
+ ** first, as slots free up (drainQueue runs after every managed run).
+ **
+ ** The scheduled entry point and the Run backup now action both come through here.
+ *******************************************************************************/
+public class BackupRunCoordinator
+{
+   private static final QLogger LOG = QLogger.getLogger(BackupRunCoordinator.class);
+
+   /** A RUNNING run whose heartbeat is older than this is dead (spec 9.1). */
+   public static final Duration HEARTBEAT_STALE = Duration.ofMinutes(10);
+
+   public static final String SKIPPED_CONCURRENT = "SKIPPED_CONCURRENT";
+   public static final String QUEUED             = "QUEUED";
+
+   private static final Object LEASE_LOCK = new Object();
+
+   private final BackupCatalog catalog = new BackupCatalog();
+
+
+
+   /*******************************************************************************
+    ** Run a plan.  The result status is the run status, or SKIPPED_CONCURRENT (a
+    ** run of the plan is already live) or QUEUED (the instance is full; the run
+    ** starts when a slot frees up).
+    *******************************************************************************/
+   public BackupRunResult runPlan(QRecord plan, String trigger) throws QException
+   {
+      return (runPlan(plan, trigger, null));
+   }
+
+
+
+   /*******************************************************************************
+    ** Run a plan, letting the caller adjust the request first (swap the row reader,
+    ** add a file reference provider, ...).
+    *******************************************************************************/
+   public BackupRunResult runPlan(QRecord plan, String trigger, java.util.function.Consumer<BackupRunRequest> customizer) throws QException
+   {
+      BackupRunRequest request = BackupRunFactory.forPlan(plan, trigger);
+      if(customizer != null)
+      {
+         customizer.accept(request);
+      }
+
+      String refusal = encryptionRefusal(plan, request);
+      if(refusal != null)
+      {
+         BackupRunResult result = new BackupRunResult();
+         result.setStatus("FAILED");
+         result.setErrorMessage(refusal);
+         return (result);
+      }
+
+      Claim claim = claim(plan.getValueInteger("id"), null, trigger, request.getRequestedBy(), true);
+      try
+      {
+         if(claim.outcome != Outcome.STARTED)
+         {
+            return (claim.toResult());
+         }
+         return (execute(request, claim.runId));
+      }
+      finally
+      {
+         if(claim.outcome == Outcome.STARTED)
+         {
+            drainQueue();
+         }
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Run an ad-hoc export (no plan).  It counts as a live run for the instance
+    ** limit but, being built from records in hand, it is never queued.
+    *******************************************************************************/
+   public BackupRunResult runAdHoc(BackupRunRequest request) throws QException
+   {
+      Claim claim = claim(null, null, request.getTrigger(), request.getRequestedBy(), false);
+      try
+      {
+         return (execute(request, claim.runId));
+      }
+      finally
+      {
+         drainQueue();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Resume a failed run of a plan: the same run carries on.  Refused (with a
+    ** message) when it cannot be resumed, or when a run of the plan is live.
+    *******************************************************************************/
+   public BackupRunResult resume(QRecord run) throws QException
+   {
+      QRecord plan = run.getValueInteger("backupPlanId") == null ? null : catalog.get(BackupsTables.PLAN, run.getValueInteger("backupPlanId")).orElse(null);
+      String  why  = BackupResumePlanner.whyNotResumable(run, plan);
+      if(why != null)
+      {
+         BackupRunResult refused = new BackupRunResult();
+         refused.setRunId(run.getValueInteger("id"));
+         refused.setStatus("FAILED");
+         refused.setErrorMessage(why);
+         return (refused);
+      }
+
+      BackupRunRequest request = BackupRunFactory.forPlan(plan, "MANUAL");
+      Claim claim = claim(plan.getValueInteger("id"), run.getValueInteger("id"), run.getValueString("trigger"), request.getRequestedBy(), false);
+      if(claim.outcome != Outcome.STARTED)
+      {
+         BackupRunResult refused = claim.toResult();
+         refused.setRunId(run.getValueInteger("id"));
+         refused.setStatus("FAILED");
+         refused.setErrorMessage("Cannot resume now: " + claim.message);
+         return (refused);
+      }
+      try
+      {
+         request.withResume(BackupResumePlanner.plan(run));
+         return (execute(request, claim.runId));
+      }
+      finally
+      {
+         drainQueue();
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Start every queued run that now has a slot, oldest first.
+    *******************************************************************************/
+   public List<BackupRunResult> drainQueue() throws QException
+   {
+      List<BackupRunResult> results = new ArrayList<>();
+      while(true)
+      {
+         QRecord next = nextStartable();
+         if(next == null)
+         {
+            return (results);
+         }
+         Integer planId = next.getValueInteger("backupPlanId");
+         QRecord plan   = planId == null ? null : catalog.get(BackupsTables.PLAN, planId).orElse(null);
+         if(plan == null)
+         {
+            catalog.update(BackupsTables.RUN, next.getValueInteger("id"), Map.of("status", "CANCELLED", "endDate", now(), "errorMessage", "The plan of this queued run no longer exists."));
+            continue;
+         }
+         BackupRunRequest request = BackupRunFactory.forPlan(plan, next.getValueString("trigger") == null ? "MANUAL" : next.getValueString("trigger"));
+         results.add(execute(request, next.getValueInteger("id")));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** The oldest QUEUED run that may start now (the plan has no live run and the
+    ** instance has a free slot); it is claimed (RUNNING) before it is returned.
+    *******************************************************************************/
+   private QRecord nextStartable() throws QException
+   {
+      synchronized(LEASE_LOCK)
+      {
+         List<QRecord> live = catalog.liveRuns(cutoff());
+         for(QRecord queued : catalog.queuedRuns())
+         {
+            if(live.size() >= maxConcurrentRuns())
+            {
+               return (null);
+            }
+            Integer planId = queued.getValueInteger("backupPlanId");
+            if(planId != null && live.stream().anyMatch(r -> planId.equals(r.getValueInteger("backupPlanId"))))
+            {
+               continue;
+            }
+            catalog.update(BackupsTables.RUN, queued.getValueInteger("id"), Map.of("status", "RUNNING", "heartbeatDate", now()));
+            return (queued);
+         }
+         return (null);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Run the engine on a claimed run row, then return its result.
+    *******************************************************************************/
+   private BackupRunResult execute(BackupRunRequest request, Integer runId)
+   {
+      request.withExistingRunId(runId).withCancelRequested(() -> catalog.isCancelRequested(runId));
+      return (new BackupEngine().execute(request));
+   }
+
+
+
+   /*******************************************************************************
+    ** The lease: insert (or reuse) the run row and decide.
+    *******************************************************************************/
+   private Claim claim(Integer planId, Integer existingRunId, String trigger, String requestedBy, boolean queueIfFull) throws QException
+   {
+      synchronized(LEASE_LOCK)
+      {
+         Instant       now  = now();
+         List<QRecord> live = catalog.liveRuns(cutoff());
+
+         Integer runId = existingRunId;
+         if(runId == null)
+         {
+            Map<String, Serializable> values = new LinkedHashMap<>();
+            values.put("backupPlanId", planId);
+            values.put("runKey", "queued-" + UUID.randomUUID().toString().substring(0, 8));
+            values.put("trigger", trigger);
+            values.put("requestedByUserId", requestedBy);
+            values.put("status", QUEUED);
+            values.put("startDate", now);
+            runId = catalog.insert(BackupsTables.RUN, values);
+         }
+         final Integer claimedId = runId;
+
+         if(planId != null && live.stream().anyMatch(r -> planId.equals(r.getValueInteger("backupPlanId")) && !claimedId.equals(r.getValueInteger("id"))))
+         {
+            return (skip(planId, runId, existingRunId == null, "a run of this plan is already running"));
+         }
+
+         if(planId != null && live.size() >= maxConcurrentRuns() && queueIfFull)
+         {
+            LOG.info("A backup run is queued because the instance is at its concurrent run limit", logPair("runId", runId));
+            return (new Claim(Outcome.QUEUED, runId, "the instance is running its maximum of " + maxConcurrentRuns() + " backups; this one is queued"));
+         }
+         if(planId != null && live.size() >= maxConcurrentRuns() && !queueIfFull)
+         {
+            return (new Claim(Outcome.QUEUED, runId, "the instance is running its maximum of " + maxConcurrentRuns() + " backups"));
+         }
+
+         catalog.update(BackupsTables.RUN, runId, Map.of("status", "RUNNING", "heartbeatDate", now));
+
+         /////////////////////////////////////////////////////////////////////////////
+         // a second look: another pod may have claimed the same plan in the meantime //
+         /////////////////////////////////////////////////////////////////////////////
+         if(planId != null)
+         {
+            for(QRecord other : catalog.liveRuns(cutoff()))
+            {
+               if(planId.equals(other.getValueInteger("backupPlanId")) && other.getValueInteger("id") < runId && !other.getValueInteger("id").equals(existingRunId))
+               {
+                  return (skip(planId, runId, existingRunId == null, "a run of this plan started at the same moment"));
+               }
+            }
+         }
+         return (new Claim(Outcome.STARTED, runId, null));
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Record that a trigger was skipped because the plan is already running: a
+    ** visible run row, and a warning.
+    *******************************************************************************/
+   private Claim skip(Integer planId, Integer runId, boolean ownRow, String why) throws QException
+   {
+      String message = SKIPPED_CONCURRENT + ": " + why;
+      if(ownRow)
+      {
+         Map<String, Serializable> values = new LinkedHashMap<>();
+         values.put("status", "CANCELLED");
+         values.put("endDate", now());
+         values.put("errorMessage", message);
+         values.put("warningCount", 1);
+         catalog.update(BackupsTables.RUN, runId, values);
+      }
+      String planKey = String.valueOf(planId);
+      try
+      {
+         planKey = catalog.get(BackupsTables.PLAN, planId).map(p -> p.getValueString("planKey")).orElse(planKey);
+      }
+      catch(QException e)
+      {
+         LOG.debug("Could not look up a plan key", e);
+      }
+      LOG.warn("A backup trigger was skipped: " + why, logPair("planId", planId));
+      BackupsRuntime.getAlertSink().raise("backups.run.skipped_concurrent", "backups.run.skipped_concurrent:" + planKey, BackupAlertSink.Severity.WARNING,
+         "A backup was skipped because the plan is already running", Map.of("plan", planKey));
+      return (new Claim(Outcome.SKIPPED, runId, message));
+   }
+
+
+
+   /*******************************************************************************
+    ** A plan that asks for client-side encryption must not silently produce plain
+    ** files: refuse unless the request carries a cipher.
+    *******************************************************************************/
+   private static String encryptionRefusal(QRecord plan, BackupRunRequest request)
+   {
+      String mode = plan.getValueString("encryptionMode");
+      if(mode != null && !"NONE".equals(mode) && !request.getTransform().encrypts())
+      {
+         return ("The plan asks for " + mode + " encryption, but no cipher is available in this version. Nothing was written.");
+      }
+      return (null);
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static int maxConcurrentRuns()
+   {
+      BackupsQBitConfig config = BackupsRuntime.getConfig();
+      Integer           max    = config == null ? null : config.getMaxConcurrentRuns();
+      return (max == null || max < 1 ? 1 : max);
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static Instant now()
+   {
+      return (BackupsRuntime.getClock().instant());
+   }
+
+
+
+   /*******************************************************************************
+    **
+    *******************************************************************************/
+   private static Instant cutoff()
+   {
+      return (now().minus(HEARTBEAT_STALE));
+   }
+
+
+
+   /***************************************************************************
+    **
+    ***************************************************************************/
+   private enum Outcome
+   {
+      STARTED, SKIPPED, QUEUED
+   }
+
+
+
+   /***************************************************************************
+    **
+    ***************************************************************************/
+   private record Claim(Outcome outcome, Integer runId, String message)
+   {
+      /*******************************************************************************
+       **
+       *******************************************************************************/
+      BackupRunResult toResult()
+      {
+         BackupRunResult result = new BackupRunResult();
+         result.setRunId(runId);
+         result.setStatus(outcome == Outcome.SKIPPED ? SKIPPED_CONCURRENT : QUEUED);
+         result.setErrorMessage(message);
+         return (result);
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/backups/run/BackupRunRequest.java

@@ -52,6 +52,8 @@ public class BackupRunRequest
    private Clock                      clock           = Clock.systemUTC();
    private BooleanSupplier            cancelRequested = () -> false;
    private java.util.function.Function<String, BackupStorage> sourceResolver;
+   private Integer                    existingRunId;
+   private ResumeState                resume;
 
 
 
@@ -410,4 +412,47 @@ public class BackupRunRequest
       this.sourceResolver = sourceResolver;
       return (this);
    }
+
+
+
+   /*******************************************************************************
+    ** Getter for existingRunId: a run row the coordinator already created (QUEUED,
+    ** then claimed RUNNING); the engine then updates it instead of inserting one.
+    *******************************************************************************/
+   public Integer getExistingRunId()
+   {
+      return (existingRunId);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for existingRunId
+    *******************************************************************************/
+   public BackupRunRequest withExistingRunId(Integer existingRunId)
+   {
+      this.existingRunId = existingRunId;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** Getter for resume: what an earlier, failed attempt of this run already wrote.
+    *******************************************************************************/
+   public ResumeState getResume()
+   {
+      return (resume);
+   }
+
+
+
+   /*******************************************************************************
+    ** Fluent setter for resume
+    *******************************************************************************/
+   public BackupRunRequest withResume(ResumeState resume)
+   {
+      this.resume = resume;
+      return (this);
+   }
 }