Merge remote-tracking branch 'origin/develop' into feature/job-progress

# Conflicts:
#	CHANGELOG.md
#	README.md
#	pom.xml
#	src/main/java/com/kingsrook/qbits/backups/processes/BackupsProcesses.java

CHANGELOG.md

@@ -5,6 +5,9 @@ All notable changes to this project are documented here. The format follows Keep
 ## [0.1.0] - Unreleased
 
 ### Added
+- Always-durable backup, resume, scheduled backup, restore and verify processes with conservative `NONE` idempotency and stable backup identities on explicit job retry.
+- Weighted backup phases, table/file counts, stored bytes, safe per-table/artifact logs, warnings, failures and resume notices through QQQ `JobProgress`.
+- Cooperative job cancellation during backup pages, media copying, restore pages and verification scans; local media-reference spooling keeps phase separation from growing heap usage across tables.
 - Complete host-neutral in-app help bundle with live references and a help coverage gate. Hosts can override entries with a HOST bundle.
 - Guided empty states: help bundle `help/qbit-backups.yaml` (app and all nine tables, with "How rows get created") applied by `BackupsHelp`, and a Setup card on the Backups home (plan, scope rules, storage, schedule, verification, prune bucket).
 - "Create default plan" table action on Backup plans: adds "Nightly full" for a storage bucket alias; insert-only.
@@ -12,6 +15,8 @@ All notable changes to this project are documented here. The format follows Keep
 - Repository scaffold: `BackupsQBitConfig` and `BackupsQBitProducer` (table prefix `backups_`, permissions `backups.*`), build, CI and repository standard files.
 
 ### Changed
+- Target QQQ `4.1.0-RC.8` (job progress).
+- Persist controlled backup failure messages instead of untrusted provider exception text.
 - Pin QQQ to released 4.1.0-RC.7 and replace the temporary help loader with core HelpBundles registration.
 - Saving a plan with Use read replica on is refused (it was never read); the not-yet-available settings (read replica, native dump, incremental, PARQUET, ZSTD) say so in their field help; scheduledJobId is documented as unused.
 - Build against the released QQQ BOM 4.1.0-RC.5; replace reflective alerts binding with core QAlerts and real sink integration tests.

README.md

@@ -10,7 +10,7 @@
 ## Requirements
 
 - Java 21.
-- QQQ `4.1.0-RC.7` or later.
+- QQQ `4.1.0-RC.8` or later.
 
 ## Install
 
@@ -57,6 +57,26 @@ checks empty-state, widget and possible-value-source coverage. Run it with
 
 Always use the Maven wrapper (Maven 3.9.11); Maven 3.10 cannot read the parent POM.
 
+## Durable jobs and progress
+
+Run backup now, Resume, scheduled backups, both Restore actions, and Verify backup
+are always durable jobs. Hosts must configure QQQ durable jobs and its worker.
+Review remains a preview; execution runs in the background and reports through
+the job page and jobs tray.
+
+Backups report Resolve scope, Back up tables, Copy media files, and Write catalog
+and manifest. Separate verification jobs report a single Verify phase.
+The job includes table/file counts, stored bytes, per-table
+rows and duration, artifact names, warnings and failures. Logs exclude row values,
+source file keys, storage credentials and endpoints. Provider errors use controlled
+failure messages; detailed exceptions stay in server diagnostics.
+
+These processes use `NONE` idempotency: a worker interruption requires review before
+retry because object writes and catalog updates are not atomic. Explicitly retrying
+a backup job reuses its run and resumes completed parts; it does not create another
+backup. Snapshot runs that cannot resume must be started as a new backup. See
+[the progress and retry contract](docs/DESIGN.md#job-progress-implementation) for limits and cancellation.
+
 ## License
 
 Apache License 2.0. See [LICENSE](LICENSE) and [NOTICE](NOTICE).

docs/DESIGN.md

@@ -1,3 +1,18 @@
 # QBit Backups: design
 
 The design for this qbit is tracked in [QRun-IO/qqq#1091](https://github.com/QRun-IO/qqq/pull/1091). This repository implements it; the implementation is in progress.
+
+
+## Job progress implementation
+
+Backup, resume, scheduled backup, restore, and verify execution use always-durable
+jobs with `NONE` idempotency against the QQQ job-progress API targeted for RC.8.
+Backup phases are `prepare`, `tables`, `media`, and `upload`, weighted 2, 75, 18,
+and 5. Verification runs separately with a single `verify` phase.
+
+Progress reports table/file counts, stored bytes, and safe artifact names without
+row data, source keys, credentials, or endpoints. Cancellation is cooperative at
+page and media-copy boundaries. Media references spool locally between phases.
+Object writes and catalog updates are not atomic, so worker interruptions require
+review before retry. Explicit backup retries reuse the run and resume completed
+parts; snapshot runs that cannot resume require a new backup.

pom.xml

@@ -194,14 +194,14 @@
    <profiles>
       <!-- Built against qqq 4.1 (QBitMetaDataProducer). Until qbit-build-parent  -->
       <!-- imports the released 4.1 BOM, this profile is active by default and     -->
-      <!-- imports qqq-bom-pom 4.1.0-RC.7 ahead of the parent's 4.0.0 BOM.         -->
+      <!-- imports qqq-bom-pom 4.1.0-RC.8 ahead of the parent's 4.0.0 BOM.         -->
       <profile>
          <id>qqq-snapshot</id>
          <activation>
             <activeByDefault>true</activeByDefault>
          </activation>
          <properties>
-            <qqq.snapshot.version>4.1.0-RC.7</qqq.snapshot.version>
+            <qqq.snapshot.version>4.1.0-RC.8</qqq.snapshot.version>
          </properties>
          <dependencyManagement>
             <dependencies>

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

@@ -30,6 +30,11 @@ import com.kingsrook.qbits.backups.run.BackupRunFactory;
 import com.kingsrook.qbits.backups.run.BackupRunResult;
 import com.kingsrook.qbits.backups.scope.ScopeResolver;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.jobs.DurableProcessMetaData;
+import com.kingsrook.qqq.backend.core.jobs.model.JobIdempotency;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException;
+import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
 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;
@@ -71,7 +76,7 @@ public final class BackupsProcesses
    {
       List<QProcessMetaData> processes = new ArrayList<>();
       processes.add(permission(etl(RUN_PLAN, "Run backup now", BackupsTables.PLAN, RunPlanTransform.class, RunPlanLoad.class,
-         "This is a preview of what the backup will include. Nothing is written until you continue."), BackupsPermissions.RUN));
+         "This is a preview of what the backup will include. Nothing is written until you continue.").withSupplementalMetaData(durability()), BackupsPermissions.RUN));
       processes.add(DefaultPlanProcess.process());
       processes.addAll(VerifyProcesses.processes());
       processes.addAll(RestoreProcesses.processes());
@@ -82,6 +87,40 @@ public final class BackupsProcesses
 
 
 
+   /*******************************************************************************
+    ** External writes cannot commit with the job chunk. Require review before retry
+    ** instead of claiming that replay is naturally idempotent.
+    *******************************************************************************/
+   public static DurableProcessMetaData durability()
+   {
+      return (new DurableProcessMetaData().withEligible(true).withAlwaysDurable(true)
+         .withChunkSize(1).withIdempotency(JobIdempotency.NONE));
+   }
+
+
+
+   /*******************************************************************************
+    ** A durable worker must see failures/cancellation instead of a successful ETL
+    ** summary. Inline callers retain the existing result-screen behavior.
+    *******************************************************************************/
+   public static void requireSuccessfulJob(BackupRunResult result) throws QException
+   {
+      if(JobChunkScope.current().isEmpty())
+      {
+         return;
+      }
+      if("CANCELLED".equals(result.getStatus()))
+      {
+         throw (new JobInterruptedException(JobInterruptedException.Reason.CANCEL));
+      }
+      if(!result.isUsable())
+      {
+         throw (new QException(result.getErrorMessage() == null ? "Backup did not complete" : result.getErrorMessage()));
+      }
+   }
+
+
+
    /*******************************************************************************
     ** Record the permission a process needs.
     *******************************************************************************/
@@ -229,8 +268,9 @@ public final class BackupsProcesses
       {
          for(QRecord plan : input.getRecords())
          {
-            BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "MANUAL");
+            BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "MANUAL", JobProgress.of(input), null);
             results.add(result);
+            requireSuccessfulJob(result);
             switch(result.getStatus())
             {
                case "SUCCEEDED" -> succeeded.incrementCountAndAddPrimaryKey(result.getRunId());

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

@@ -33,6 +33,8 @@ import com.kingsrook.qbits.backups.restore.RestoreRequest;
 import com.kingsrook.qbits.backups.restore.RestoreSummaries;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
 import com.kingsrook.qqq.backend.core.exceptions.QUserFacingException;
+import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
+import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
 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;
@@ -107,7 +109,7 @@ public final class RestoreProcesses
       process.setLabel(label);
       process.setTableName(table);
       process.setPermissionRules(BackupsTables.permissionRules());
-      return (process);
+      return (process.withSupplementalMetaData(BackupsProcesses.durability()));
    }
 
 
@@ -241,13 +243,38 @@ public final class RestoreProcesses
       @Override
       public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
       {
+         JobProgress progress = JobProgress.of(input);
+         progress.phases(JobPhase.of("restore", "Restore backups", 100)).phase("restore");
+         Map<Integer, Set<String>> sources = sources(input.getRecords());
+         progress.total(sources.size(), "backups").done(0);
+         int done = 0;
          RestoreEngine engine = new RestoreEngine();
-         for(Map.Entry<Integer, Set<String>> source : sources(input.getRecords()).entrySet())
+         try
          {
-            RestoreRequest request = RestoreRequest.fromInput(input, source.getKey(), new ArrayList<>(source.getValue()));
-            outcomes.add(engine.execute(request));
+            for(Map.Entry<Integer, Set<String>> source : sources.entrySet())
+            {
+               progress.message("Restoring backup " + source.getKey());
+               RestoreRequest request = RestoreRequest.fromInput(input, source.getKey(), new ArrayList<>(source.getValue()));
+               outcomes.add(engine.execute(request));
+               progress.info("Restored backup " + source.getKey());
+               progress.done(++done);
+            }
+            progress.getSnapshot().getPhases().getFirst().setStatus(JobPhase.Status.DONE);
+            output.getRecords().addAll(input.getRecords());
+         }
+         catch(QException | RuntimeException e)
+         {
+            progress.getSnapshot().getPhases().getFirst().setStatus(JobPhase.Status.FAILED);
+            if(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.find(e) == null)
+            {
+               progress.error("Restore failed (" + e.getClass().getSimpleName() + ")");
+            }
+            throw (e);
+         }
+         finally
+         {
+            progress.flush();
          }
-         output.getRecords().addAll(input.getRecords());
       }
 
 

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

@@ -70,7 +70,7 @@ public final class ResumeRunProcess
    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));
+         "This is a preview of where each selected run will carry on. Nothing is written until you continue.").withSupplementalMetaData(BackupsProcesses.durability()), BackupsPermissions.RUN));
    }
 
 
@@ -167,7 +167,8 @@ public final class ResumeRunProcess
       {
          for(QRecord run : input.getRecords())
          {
-            BackupRunResult result = new BackupRunCoordinator().resume(run);
+            BackupRunResult result = new BackupRunCoordinator().resume(run, com.kingsrook.qqq.backend.core.jobs.progress.JobProgress.of(input));
+            BackupsProcesses.requireSuccessfulJob(result);
             switch(result.getStatus())
             {
                case "SUCCEEDED" -> succeeded.incrementCountAndAddPrimaryKey(run.getValue("id"));

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

@@ -30,6 +30,9 @@ import com.kingsrook.qbits.backups.verify.VerificationOutcome;
 import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
 import com.kingsrook.qqq.backend.core.context.QContext;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
+import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
+import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
 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;
@@ -85,7 +88,7 @@ public final class VerifyProcesses
    {
       List<QProcessMetaData> processes = new ArrayList<>();
       processes.add(BackupsProcesses.permission(BackupsProcesses.etl(VERIFY, "Verify backup", BackupsTables.RUN, VerifyTransform.class, VerifyLoad.class,
-         "This checks that the backup's objects are all in storage, then reads every byte back, decrypts and decodes it, and compares counts and hashes with the manifest. Nothing is changed in the backup."),
+         "This checks that the backup's objects are all in storage, then reads every byte back, decrypts and decodes it, and compares counts and hashes with the manifest. Nothing is changed in the backup.").withSupplementalMetaData(BackupsProcesses.durability()),
          BackupsPermissions.RUN));
       processes.add(checkSlaProcess(BackupsRuntime.getConfig()));
       return (processes);
@@ -220,12 +223,31 @@ public final class VerifyProcesses
       @Override
       public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
       {
+         JobProgress progress = JobProgress.of(input);
+         progress.phases(JobPhase.of("verify", "Verify", 100)).phase("verify").total(input.getRecords().size(), "backups").done(0);
+         int done = 0;
          String level = input.getValueString(FIELD_VERIFY_LEVEL) == null ? BackupVerifier.CONTENT : input.getValueString(FIELD_VERIFY_LEVEL);
          String user  = QContext.getQSession() == null || QContext.getQSession().getUser() == null ? "system" : QContext.getQSession().getUser().getIdReference();
          for(QRecord run : input.getRecords())
          {
+            JobChunkScope.checkBetweenPages();
+            progress.message("Verifying backup " + run.getValueInteger("id"));
             List<VerificationOutcome> outcomes = new BackupVerifier().verifyThrough(run.getValueInteger("id"), level, "MANUAL", user);
             VerificationOutcome       last     = outcomes.get(outcomes.size() - 1);
+            for(VerificationOutcome outcome : outcomes)
+            {
+               progress.info("Verification " + outcome.level() + ": " + outcome.status());
+            }
+            if(!last.passed())
+            {
+               progress.getSnapshot().getPhases().getFirst().setStatus(JobPhase.Status.FAILED);
+               progress.error("Backup verification " + last.status());
+               if(JobChunkScope.current().isPresent())
+               {
+                  throw (new QException("Backup verification " + last.status()));
+               }
+            }
+            progress.done(++done);
             switch(last.status())
             {
                case "PASSED" -> passed.incrementCountAndAddPrimaryKey(run.getValue("id"));
@@ -248,6 +270,8 @@ public final class VerifyProcesses
             }
             output.getRecords().add(run);
          }
+         progress.getSnapshot().getPhases().getFirst().setStatus(failed.getCount() > 0 || errored.getCount() > 0 ? JobPhase.Status.FAILED : JobPhase.Status.DONE);
+         progress.flush();
       }
 
 

src/main/java/com/kingsrook/qbits/backups/restore/RestoreEngine.java

@@ -49,6 +49,8 @@ import com.kingsrook.qqq.backend.core.actions.audits.AuditAction;
 import com.kingsrook.qqq.backend.core.actions.tables.DeleteAction;
 import com.kingsrook.qqq.backend.core.context.QContext;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException;
 import com.kingsrook.qqq.backend.core.logging.QLogger;
 import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
 import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
@@ -373,6 +375,15 @@ public class RestoreEngine
       }
       catch(QException | RuntimeException e)
       {
+         JobInterruptedException interrupted = JobInterruptedException.find(e);
+         if(interrupted != null)
+         {
+            if(interrupted.getReason() == JobInterruptedException.Reason.CANCEL)
+            {
+               catalog.update(BackupsTables.RESTORE_RUN, restoreRunId, Map.of("status", "CANCELLED", "endDate", BackupsRuntime.getClock().instant()));
+            }
+            throw (interrupted);
+         }
          LOG.warn("Restore failed", e, logPair("restoreRunId", restoreRunId));
          String message = e.getClass().getSimpleName() + ": " + e.getMessage();
          try
@@ -447,6 +458,7 @@ public class RestoreEngine
       }
 
       BackupRunRequest backup = new BackupRunRequest()
+         .withProgress(new com.kingsrook.qqq.backend.core.jobs.progress.JobProgress((com.kingsrook.qqq.backend.core.actions.async.AsyncJobCallback) null))
          .withSettings(settings)
          .withScope(new ScopeResolver(QContext.getQInstance()).resolve(settings, rules))
          .withTrigger("PRE_RESTORE")
@@ -507,10 +519,11 @@ public class RestoreEngine
     ** index.  A conflicting file (different content at the target) is only replaced
     ** by a REPLACE restore.
     *******************************************************************************/
-   private void restoreFiles(RestoreRequest request, Plan plan, RestorePass pass, RestoreOutcome outcome, Integer restoreRunId)
+   private void restoreFiles(RestoreRequest request, Plan plan, RestorePass pass, RestoreOutcome outcome, Integer restoreRunId) throws QException
    {
       for(RestorePass.FileWork work : pass.getFileWork())
       {
+         JobChunkScope.checkBetweenPages();
          if(work.conflict() && !RestoreRequest.MODE_REPLACE.equals(request.getMode()))
          {
             continue;

src/main/java/com/kingsrook/qbits/backups/restore/RestorePass.java

@@ -46,6 +46,7 @@ import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
 import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
 import com.kingsrook.qqq.backend.core.context.QContext;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
 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;
@@ -200,6 +201,7 @@ public class RestorePass
 
       for(String name : order)
       {
+         JobChunkScope.checkBetweenPages();
          restoreTable(name);
       }
       retryPass();
@@ -566,6 +568,7 @@ public class RestorePass
     *******************************************************************************/
    private void processPage(TableContext c, List<QRecord> page) throws QException
    {
+      JobChunkScope.checkBetweenPages();
       RestoreOutcome.TableOutcome o = c.outcome;
       o.addRows(page.size());
 

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

@@ -17,16 +17,21 @@
 package com.kingsrook.qbits.backups.run;
 
 
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
 import java.io.IOException;
 import java.io.OutputStream;
 import java.io.Serializable;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.time.Instant;
 import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.UUID;
 import java.util.function.Function;
@@ -47,6 +52,11 @@ import com.kingsrook.qbits.backups.scope.ScopeResolver;
 import com.kingsrook.qbits.backups.storage.BackupStorage;
 import com.kingsrook.qbits.backups.storage.WriteOptions;
 import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException;
+import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
+import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
 import com.kingsrook.qqq.backend.core.logging.QLogger;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
@@ -85,13 +95,19 @@ public class BackupEngine
 
    /*******************************************************************************
     ** Run a backup.  Never throws for a backup failure: the result carries the
-    ** status and message (and the run row is FAILED).
+    ** status and message (and the run row is FAILED). Cooperative job interrupts
+    ** propagate as QException so the durable worker handles cancellation/lease loss.
     *******************************************************************************/
-   public BackupRunResult execute(BackupRunRequest request)
+   public BackupRunResult execute(BackupRunRequest request) throws QException
    {
+      JobProgress progress = request.getProgress();
+      if(!"prepare".equals(progress.getSnapshot().getCurrentPhase()))
+      {
+         prepare(progress);
+      }
       BackupRunResult result  = new BackupRunResult();
       Instant         started = request.getClock().instant();
-      String          runKey  = request.getResume() != null ? request.getResume().runKey() : RUN_KEY_FORMAT.format(started) + "-" + UUID.randomUUID().toString().substring(0, 4);
+      String          runKey  = request.getResume() != null ? request.getResume().runKey() : request.getRunKey() != null ? request.getRunKey() : 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  = request.getResume() != null ? request.getResume().runPrefix() : instancePrefix + request.getSettings().planKey() + "/" + DATE_PATH.format(started) + "/" + runKey + "/";
@@ -139,27 +155,52 @@ public class BackupEngine
       catch(Exception e)
       {
          result.setStatus("FAILED");
-         result.setErrorMessage("Could not create the run record: " + e.getMessage());
+         result.setErrorMessage("Could not create the run record (" + e.getClass().getSimpleName() + ")");
+         failProgress(progress, "Could not create the run record (" + e.getClass().getSimpleName() + ")");
          return (result);
       }
 
       try
       {
+         checkJob();
+         if(request.getResume() != null)
+         {
+            long complete = request.getResume().tables().values().stream().filter(ResumeState.TableState::complete).count();
+            progress.info("Resuming run " + result.getRunId() + ": " + complete + " tables already done");
+         }
          doRun(request, result, runPrefix, instancePrefix, started);
+         completePhase(progress, JobPhase.Status.DONE);
       }
       catch(CancelledException e)
       {
          finish(request, result, "CANCELLED", "Cancelled", started);
+         progress.info("Backup cancelled");
       }
       catch(TimedOutException e)
       {
          finish(request, result, "TIMED_OUT", e.getMessage(), started);
+         failProgress(progress, e.getMessage());
       }
       catch(Exception e)
       {
+         JobInterruptedException interrupted = JobInterruptedException.find(e);
+         if(interrupted != null)
+         {
+            if(interrupted.getReason() != JobInterruptedException.Reason.LEASE_LOST)
+            {
+               finish(request, result, interrupted.getReason() == JobInterruptedException.Reason.CANCEL ? "CANCELLED" : "ABANDONED", "Job interrupted", started);
+            }
+            throw (interrupted);
+         }
          LOG.warn("Backup run failed", e, logPair("runKey", runKey));
+         // Provider and reader messages can contain credentials or row values.
+         failProgress(progress, scrub(e));
          finish(request, result, "FAILED", scrub(e), started);
       }
+      finally
+      {
+         progress.flush();
+      }
       return (result);
    }
 
@@ -170,6 +211,7 @@ public class BackupEngine
     *******************************************************************************/
    private void doRun(BackupRunRequest request, BackupRunResult result, String runPrefix, String instancePrefix, Instant started) throws Exception
    {
+      JobProgress      progress = request.getProgress();
       QInstance        qInstance = QContext.getQInstance();
       BackupStorage    storage   = request.getStorage();
       var              settings  = request.getSettings();
@@ -188,6 +230,7 @@ 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);
+      result.getWarnings().forEach(progress::warn);
       catalog.update(BackupsTables.RUN, result.getRunId(), Map.of("consistencyAchieved", result.getConsistencyAchieved()));
 
       //////////////////////////////////////////////////////////////////////////////////////
@@ -230,10 +273,15 @@ public class BackupEngine
       long                            fileBytes      = 0;
       long                            fileBytesRef   = 0;
 
-      try
+      progress.phase("tables").total(request.getScope().tables().size(), "tables").done(0).bytesDone(0);
+      try(PendingFiles pendingFiles = new PendingFiles())
       {
          for(ScopeResolver.ResolvedTable resolved : request.getScope().tables())
          {
+            checkJob();
+            long tableStarted = System.nanoTime();
+            long tableBytesBefore = bytesStored;
+            progress.message("Backing up " + resolved.name() + " (" + (done + 1) + " of " + request.getScope().tables().size() + ")");
             QTableMetaData table = qInstance.getTable(resolved.name());
             String         backendType = qInstance.getBackend(table.getBackendName()) == null ? null : qInstance.getBackend(table.getBackendName()).getBackendType();
 
@@ -242,6 +290,7 @@ public class BackupEngine
 
             ResumeState.TableState prior = request.getResume() == null ? null : request.getResume().table(table.getName());
 
+            long[] tableWritten = { 0 };
             List<TableArtifactWriter> writers = new ArrayList<>();
             String tablePrefix = runPrefix + "tables/" + table.getName() + "/";
             Map<TableArtifactWriter, Integer> lastArtifactIds = new LinkedHashMap<>();
@@ -277,6 +326,9 @@ public class BackupEngine
                   {
                      try
                      {
+                        tableWritten[0] += part.bytesStored();
+                        progress.bytesDone(tableBytesBefore + tableWritten[0]);
+                        progress.info("Wrote " + part.key().substring(runPrefix.length()));
                         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)));
@@ -296,6 +348,8 @@ public class BackupEngine
                   if(canContinue)
                   {
                      TableArtifactWriter writer = writers.get(0);
+                     tableWritten[0] = prior.parts().stream().mapToLong(TableArtifactWriter.PartResult::bytesStored).sum();
+                     progress.bytesDone(tableBytesBefore + tableWritten[0]);
                      writer.resumeFrom(prior.parts());
                      seedRowsHash(storage, runPrefix, prior.parts(), writer);
                      lastArtifactIds.put(writer, prior.artifactIds().get(prior.artifactIds().size() - 1));
@@ -368,8 +422,31 @@ public class BackupEngine
             }
             rows += counter[0];
 
-            for(FileReference ref : references)
+            pendingFiles.write(references);
+
+            Map<String, String> keyRange = new LinkedHashMap<>();
+            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.CORE_ALGORITHM, TableFingerprints.descriptor(table, backendType), resolved.filterJson(), excluded, counter[0], rowsHash, keyRange, parts, consistency));
+            done++;
+            catalogProgress(request, result, done, rows, bytesRaw, bytesStored);
+            progress.done(done).bytesDone(bytesStored);
+            progress.info(String.format(Locale.US, "%s: %,d rows, %.1f MB in %.1f s", table.getName(), counter[0],
+               (bytesStored - tableBytesBefore) / 1_000_000.0, (System.nanoTime() - tableStarted) / 1_000_000_000.0));
+            if(request.getCancelRequested().getAsBoolean())
             {
+               throw (new CancelledException());
+            }
+         }
+         progress.phase("media").total(pendingFiles.count, "files").done(0).bytesDone(bytesStored);
+         try(BufferedReader pending = pendingFiles.openReader())
+         {
+            String line;
+            while((line = pending.readLine()) != null)
+            {
+               FileReference ref = MAPPER.readValue(line, FileReference.class);
+               checkJob();
                if(indexWriter == null)
                {
                   indexWriter = new FileIndexWriter(storage, runPrefix + "files/index-00001.jsonl.gz", result.getRunKey(), fileLockMode, fileRetainUntil);
@@ -388,24 +465,16 @@ public class BackupEngine
                   case FileCopier.MISSING -> missing++;
                   default -> fileBytesRef += Math.max(0, entry.size());
                }
+               progress.done(referenced).bytesDone(bytesStored + fileBytes)
+                  .message("Media: " + copied + " copied, " + deduplicated + " already stored (" + referenced + " of " + pendingFiles.count + ")");
                if(request.getCancelRequested().getAsBoolean())
                {
                   throw (new CancelledException());
                }
             }
-
-            Map<String, String> keyRange = new LinkedHashMap<>();
-            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.CORE_ALGORITHM, TableFingerprints.descriptor(table, backendType), resolved.filterJson(), excluded, counter[0], rowsHash, keyRange, parts, consistency));
-            done++;
-            catalogProgress(result, done, rows, bytesRaw, bytesStored);
-            if(request.getCancelRequested().getAsBoolean())
-            {
-               throw (new CancelledException());
-            }
          }
+
+         progress.info("Media: " + referenced + " files, " + copied + " copied, " + deduplicated + " already stored, " + missing + " missing");
       }
       finally
       {
@@ -415,10 +484,13 @@ public class BackupEngine
          }
       }
 
+      checkJob();
+      progress.phase("upload").message("Writing catalog and manifest");
       List<String> indexKeys = new ArrayList<>();
       if(indexWriter != null)
       {
          indexKeys.add("files/index-00001.jsonl.gz");
+         progress.info("Wrote files/index-00001.jsonl.gz");
          checksums.add(indexWriter.sha256Stored() + "  files/index-00001.jsonl.gz");
          catalog.insertArtifact(result.getRunId(), "FILE_OBJECTS_INDEX", null, 1, "files/index-00001.jsonl.gz", "JSONL", "GZIP", request.getTransform().encrypts(),
             indexWriter.rows(), indexWriter.bytesRaw(), indexWriter.bytesStored(), indexWriter.sha256Stored(), indexWriter.sha256Raw(), null, null, null, null);
@@ -431,7 +503,9 @@ public class BackupEngine
          {
             throw (new IllegalStateException(missing + " referenced files are missing at the source"));
          }
-         result.getWarnings().add(missing + " referenced files were missing at the source");
+         String warning = missing + " referenced files were missing at the source";
+         result.getWarnings().add(warning);
+         progress.warn(warning);
       }
 
       /////////////////////////////////////////////////////////
@@ -439,6 +513,7 @@ public class BackupEngine
       /////////////////////////////////////////////////////////
       byte[] checksumBytes = (String.join("\n", checksums) + "\n").getBytes(StandardCharsets.UTF_8);
       putBytes(storage, runPrefix + "checksums.sha256", checksumBytes, result.getRunKey(), lockMode, retainUntil);
+      progress.info("Wrote checksums.sha256");
       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);
 
@@ -463,6 +538,7 @@ public class BackupEngine
       {
          throw (new IllegalStateException("The manifest was written but is not visible in the bucket"));
       }
+      progress.info("Wrote manifest.json");
       catalog.insertArtifact(result.getRunId(), "MANIFEST", null, 1, "manifest.json", "JSON", "NONE", false, manifestTables.size(), manifestBytes.length,
          manifestBytes.length, HashingOutputStream.sha256Hex(manifestBytes), HashingOutputStream.sha256Hex(manifestBytes), null, null, null, null);
 
@@ -501,6 +577,58 @@ public class BackupEngine
 
 
 
+   /*******************************************************************************
+    ** Declare the work before scope resolution. No byte estimate requires a scan.
+    *******************************************************************************/
+   public static void prepare(JobProgress progress)
+   {
+      progress.phases(JobPhase.of("prepare", "Resolve scope", 2), JobPhase.of("tables", "Back up tables", 75),
+         JobPhase.of("media", "Copy media files", 18), JobPhase.of("upload", "Write catalog and manifest", 5));
+      progress.phase("prepare").message("Resolving scope");
+   }
+
+
+
+   /*******************************************************************************
+    ** Complete or fail the current phase; the core has no phase-finish method.
+    *******************************************************************************/
+   private static void completePhase(JobProgress progress, JobPhase.Status status)
+   {
+      progress.getSnapshot().getPhases().stream().filter(p -> p.getStatus() == JobPhase.Status.RUNNING).forEach(p -> p.setStatus(status));
+      progress.flush();
+   }
+
+
+
+   /*******************************************************************************
+    ** Log only a controlled failure description, never provider or row text.
+    *******************************************************************************/
+   private static void failProgress(JobProgress progress, String message)
+   {
+      completePhase(progress, JobPhase.Status.FAILED);
+      progress.error(message);
+   }
+
+
+
+   /*******************************************************************************
+    ** Bridge checked job interrupts through the row consumer, preserving the cause
+    ** so execute can propagate it back to the durable worker.
+    *******************************************************************************/
+   private static void checkJob()
+   {
+      try
+      {
+         JobChunkScope.checkBetweenPages();
+      }
+      catch(QException e)
+      {
+         throw (new IllegalStateException(e));
+      }
+   }
+
+
+
    /*******************************************************************************
     ** Work out the consistency achieved.  Until the core consistent-read hook ships
     ** (QQQ core PR feature/consistent-read-transactions, next 4.1.x RC), no backend
@@ -563,6 +691,7 @@ public class BackupEngine
     *******************************************************************************/
    private void heartbeat(BackupRunRequest request, BackupRunResult result, Instant deadline, String tableName, long rowsSoFar)
    {
+      checkJob();
       Instant now = request.getClock().instant();
       if(request.getCancelRequested().getAsBoolean())
       {
@@ -579,6 +708,7 @@ public class BackupEngine
       catch(Exception e)
       {
          LOG.warn("Could not record a heartbeat", e);
+         request.getProgress().warn("Could not record a heartbeat");
       }
    }
 
@@ -587,7 +717,7 @@ public class BackupEngine
    /*******************************************************************************
     **
     *******************************************************************************/
-   private void catalogProgress(BackupRunResult result, int done, long rows, long bytesRaw, long bytesStored)
+   private void catalogProgress(BackupRunRequest request, BackupRunResult result, int done, long rows, long bytesRaw, long bytesStored)
    {
       try
       {
@@ -601,6 +731,7 @@ public class BackupEngine
       catch(Exception e)
       {
          LOG.warn("Could not record progress", e);
+         request.getProgress().warn("Could not record progress");
       }
    }
 
@@ -618,6 +749,7 @@ public class BackupEngine
       catch(Exception e)
       {
          LOG.warn("Could not finish the run record", e);
+         request.getProgress().warn("Could not finish the run record");
          result.setStatus(status);
          result.setErrorMessage(message);
       }
@@ -694,13 +826,12 @@ public class BackupEngine
 
 
    /*******************************************************************************
-    ** A message safe to store: the exception type and message, never row values (the
-    ** engine never puts values in messages), truncated.
+    ** Provider exceptions may contain credentials or row values. Persist only the
+    ** failure type; detailed diagnostics stay in the server log.
     *******************************************************************************/
    private static String scrub(Exception e)
    {
-      String message = e.getClass().getSimpleName() + ": " + e.getMessage();
-      return (message.length() > 2000 ? message.substring(0, 2000) : message);
+      return ("Backup failed (" + e.getClass().getSimpleName() + "); see server logs for details.");
    }
 
 
@@ -769,6 +900,11 @@ public class BackupEngine
             {
                throw (new StopReadingException());
             }
+            checkJob();
+            if(request.getCancelRequested().getAsBoolean())
+            {
+               throw (new CancelledException());
+            }
             for(FileReferenceProvider provider : request.getFileProviders())
             {
                references.addAll(provider.referencesIn(table, record));
@@ -854,6 +990,75 @@ public class BackupEngine
 
 
 
+   /***************************************************************************
+    ** Spool references locally between table export and media copy. Keeping the
+    ** whole run's references in heap would make memory grow with every table.
+    ** The temporary file is removed on success, failure and cancellation.
+    ***************************************************************************/
+   private static class PendingFiles implements AutoCloseable
+   {
+      private final Path path;
+      private final BufferedWriter writer;
+      private long count;
+
+
+
+      /*******************************************************************************
+       ** Create a private temporary spool, outside the stored backup artifacts.
+       *******************************************************************************/
+      PendingFiles() throws IOException
+      {
+         path = Files.createTempFile("qqq-backup-references-", ".jsonl");
+         writer = Files.newBufferedWriter(path, StandardCharsets.UTF_8);
+      }
+
+
+
+      /*******************************************************************************
+       ** Append this table's references without retaining earlier tables in memory.
+       *******************************************************************************/
+      void write(List<FileReference> references) throws IOException
+      {
+         for(FileReference reference : references)
+         {
+            writer.write(MAPPER.writeValueAsString(reference));
+            writer.newLine();
+            count++;
+         }
+      }
+
+
+
+      /*******************************************************************************
+       ** Finish the spool and stream it back during the media phase.
+       *******************************************************************************/
+      BufferedReader openReader() throws IOException
+      {
+         writer.close();
+         return (Files.newBufferedReader(path, StandardCharsets.UTF_8));
+      }
+
+
+
+      /*******************************************************************************
+       ** Always release the local references, including on exceptions.
+       *******************************************************************************/
+      @Override
+      public void close() throws IOException
+      {
+         try
+         {
+            writer.close();
+         }
+         finally
+         {
+            Files.deleteIfExists(path);
+         }
+      }
+   }
+
+
+
    /***************************************************************************
     ** The per-run file index, written as it goes: one JSON line per reference.
     ***************************************************************************/

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

@@ -30,6 +30,8 @@ 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.jobs.plans.JobChunkScope;
+import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
 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;
@@ -85,7 +87,28 @@ public class BackupRunCoordinator
     *******************************************************************************/
    public BackupRunResult runPlan(QRecord plan, String trigger, java.util.function.Consumer<BackupRunRequest> customizer) throws QException
    {
-      BackupRunRequest request = BackupRunFactory.forPlan(plan, trigger);
+      return (runPlan(plan, trigger, JobProgress.current(), customizer));
+   }
+
+
+
+   /*******************************************************************************
+    ** Share the step reporter and reuse a durable job's run on an explicit retry.
+    ** NONE idempotency prevents automatic replay of non-atomic external writes.
+    *******************************************************************************/
+   public BackupRunResult runPlan(QRecord plan, String trigger, JobProgress progress, java.util.function.Consumer<BackupRunRequest> customizer) throws QException
+   {
+      BackupEngine.prepare(progress);
+      BackupRunRequest request;
+      try
+      {
+         request = BackupRunFactory.forPlan(plan, trigger).withProgress(progress);
+      }
+      catch(QException | RuntimeException e)
+      {
+         progress.error("Could not resolve backup scope (" + e.getClass().getSimpleName() + ")");
+         throw (new QException("Could not resolve backup scope", e));
+      }
       if(customizer != null)
       {
          customizer.accept(request);
@@ -97,21 +120,72 @@ public class BackupRunCoordinator
          BackupRunResult result = new BackupRunResult();
          result.setStatus("FAILED");
          result.setErrorMessage(refusal);
+         progress.error(refusal);
          return (result);
       }
 
-      Claim claim = claim(plan.getValueInteger("id"), null, trigger, request.getRequestedBy(), true);
+      JobChunkScope.ChunkIdentity identity = JobChunkScope.current().map(JobChunkScope::getIdentity).orElse(null);
+      Integer existingRunId = null;
+      if(identity != null)
+      {
+         String key = "job-" + identity.jobId() + "-plan-" + plan.getValueInteger("id");
+         request.withRunKey(key);
+         List<QRecord> previous = catalog.where(BackupsTables.RUN, "runKey", key);
+         if(!previous.isEmpty())
+         {
+            QRecord run = previous.get(0);
+            existingRunId = run.getValueInteger("id");
+            if(run.getValueString("manifestKey") != null)
+            {
+               progress.info("Run " + existingRunId + " already has a manifest; nothing was rewritten. Use Verify backup to recheck it.");
+               BackupRunResult result = new BackupRunResult();
+               result.setRunId(existingRunId);
+               result.setRunKey(key);
+               result.setRunPrefix(run.getValueString("runPrefix"));
+               result.setManifestKey(run.getValueString("manifestKey"));
+               result.setStatus(run.getValueString("status"));
+               return (result);
+            }
+            if(run.getValueString("runPrefix") != null)
+            {
+               // Only the same durable job can recover its interrupted RUNNING row.
+               if("RUNNING".equals(run.getValueString("status")))
+               {
+                  catalog.update(BackupsTables.RUN, existingRunId, Map.of("status", "ABANDONED"));
+                  run.setValue("status", "ABANDONED");
+               }
+               String why = BackupResumePlanner.whyNotResumable(run, plan);
+               if(why != null)
+               {
+                  progress.error(why);
+                  throw (new QException(why));
+               }
+               request.withResume(BackupResumePlanner.plan(run));
+            }
+         }
+      }
+      Claim claim = claim(plan.getValueInteger("id"), existingRunId, trigger, request.getRequestedBy(), identity == null, request.getRunKey());
       try
       {
          if(claim.outcome != Outcome.STARTED)
          {
+            progress.warn(claim.message);
+            if(identity != null)
+            {
+               if(claim.outcome == Outcome.QUEUED)
+               {
+                  catalog.update(BackupsTables.RUN, claim.runId, Map.of("status", "FAILED", "endDate", now(), "errorMessage", claim.message));
+               }
+               progress.error("Backup could not start: " + claim.message);
+               throw (new QException("Backup could not start: " + claim.message));
+            }
             return (claim.toResult());
          }
          return (execute(request, claim.runId));
       }
       finally
       {
-         if(claim.outcome == Outcome.STARTED)
+         if(claim.outcome == Outcome.STARTED && JobChunkScope.current().isEmpty())
          {
             drainQueue();
          }
@@ -133,7 +207,10 @@ public class BackupRunCoordinator
       }
       finally
       {
-         drainQueue();
+         if(JobChunkScope.current().isEmpty())
+         {
+            drainQueue();
+         }
       }
    }
 
@@ -144,6 +221,16 @@ public class BackupRunCoordinator
     ** message) when it cannot be resumed, or when a run of the plan is live.
     *******************************************************************************/
    public BackupRunResult resume(QRecord run) throws QException
+   {
+      return (resume(run, JobProgress.current()));
+   }
+
+
+
+   /*******************************************************************************
+    ** Resume using the caller's reporter (including an inline async callback).
+    *******************************************************************************/
+   public BackupRunResult resume(QRecord run, JobProgress progress) throws QException
    {
       QRecord plan = run.getValueInteger("backupPlanId") == null ? null : catalog.get(BackupsTables.PLAN, run.getValueInteger("backupPlanId")).orElse(null);
       String  why  = BackupResumePlanner.whyNotResumable(run, plan);
@@ -156,7 +243,7 @@ public class BackupRunCoordinator
          return (refused);
       }
 
-      BackupRunRequest request = BackupRunFactory.forPlan(plan, "MANUAL");
+      BackupRunRequest request = BackupRunFactory.forPlan(plan, "MANUAL").withProgress(progress);
       Claim claim = claim(plan.getValueInteger("id"), run.getValueInteger("id"), run.getValueString("trigger"), request.getRequestedBy(), false);
       if(claim.outcome != Outcome.STARTED)
       {
@@ -173,7 +260,10 @@ public class BackupRunCoordinator
       }
       finally
       {
-         drainQueue();
+         if(JobChunkScope.current().isEmpty())
+         {
+            drainQueue();
+         }
       }
    }
 
@@ -238,7 +328,7 @@ public class BackupRunCoordinator
    /*******************************************************************************
     ** Run the engine on a claimed run row, then return its result.
     *******************************************************************************/
-   private BackupRunResult execute(BackupRunRequest request, Integer runId)
+   private BackupRunResult execute(BackupRunRequest request, Integer runId) throws QException
    {
       request.withExistingRunId(runId).withCancelRequested(() -> catalog.isCancelRequested(runId));
       return (new BackupEngine().execute(request));
@@ -250,6 +340,16 @@ public class BackupRunCoordinator
     ** 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
+   {
+      return (claim(planId, existingRunId, trigger, requestedBy, queueIfFull, null));
+   }
+
+
+
+   /*******************************************************************************
+    ** Persist the job identity in the initial claim, before any external writes.
+    *******************************************************************************/
+   private Claim claim(Integer planId, Integer existingRunId, String trigger, String requestedBy, boolean queueIfFull, String runKey) throws QException
    {
       synchronized(LEASE_LOCK)
       {
@@ -261,7 +361,7 @@ public class BackupRunCoordinator
          {
             Map<String, Serializable> values = new LinkedHashMap<>();
             values.put("backupPlanId", planId);
-            values.put("runKey", "queued-" + UUID.randomUUID().toString().substring(0, 8));
+            values.put("runKey", runKey != null ? runKey : "queued-" + UUID.randomUUID().toString().substring(0, 8));
             values.put("trigger", trigger);
             values.put("requestedByUserId", requestedBy);
             values.put("status", QUEUED);

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

@@ -26,6 +26,7 @@ import com.kingsrook.qbits.backups.files.FileReferenceProvider;
 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.jobs.progress.JobProgress;
 
 
 /*******************************************************************************
@@ -54,6 +55,50 @@ public class BackupRunRequest
    private java.util.function.Function<String, BackupStorage> sourceResolver;
    private Integer                    existingRunId;
    private ResumeState                resume;
+   private JobProgress                progress = JobProgress.current();
+   private String                     runKey;
+
+
+
+   /*******************************************************************************
+    ** The reporter shared by the process, coordinator and engine.
+    *******************************************************************************/
+   public JobProgress getProgress()
+   {
+      return (progress);
+   }
+
+
+
+   /*******************************************************************************
+    ** Supply the process reporter, including its inline async callback.
+    *******************************************************************************/
+   public BackupRunRequest withProgress(JobProgress progress)
+   {
+      this.progress = progress;
+      return (this);
+   }
+
+
+
+   /*******************************************************************************
+    ** A stable identity for an explicitly retried durable job.
+    *******************************************************************************/
+   public String getRunKey()
+   {
+      return (runKey);
+   }
+
+
+
+   /*******************************************************************************
+    ** Set the durable job's backup identity before creating any artifacts.
+    *******************************************************************************/
+   public BackupRunRequest withRunKey(String runKey)
+   {
+      this.runKey = runKey;
+      return (this);
+   }
 
 
 

src/main/java/com/kingsrook/qbits/backups/schedule/BackupScheduledRunProcess.java

@@ -80,7 +80,7 @@ public class BackupScheduledRunProcess implements BackendStep
             .withInputData(new QFunctionInputMetaData().withRecordListMetaData(new QRecordListMetaData().withTableName(BackupsTables.PLAN)))
             .withCode(new QCodeReference(BackupScheduledRunProcess.class)));
       process.setPermissionRules(BackupsTables.permissionRules());
-      return (BackupsProcesses.permission(process, BackupsPermissions.RUN));
+      return (BackupsProcesses.permission(process.withSupplementalMetaData(BackupsProcesses.durability()), BackupsPermissions.RUN));
    }
 
 
@@ -147,11 +147,12 @@ public class BackupScheduledRunProcess implements BackendStep
       }
       BackupsRuntime.getAlertSink().resolve("backups.plan.owner-invalid", "backups.plan.owner-invalid:" + plan.getValueString("planKey"));
 
-      BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "SCHEDULED");
+      BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "SCHEDULED", com.kingsrook.qqq.backend.core.jobs.progress.JobProgress.of(input), null);
       output.addValue("status", result.getStatus());
       output.addValue("runId", result.getRunId());
       if(!"SUCCEEDED".equals(result.getStatus()) && !"SUCCEEDED_WITH_WARNINGS".equals(result.getStatus()))
       {
+         BackupsProcesses.requireSuccessfulJob(result);
          LOG.info("A scheduled backup did not succeed", logPair("plan", plan.getValueString("planKey")), logPair("status", result.getStatus()), logPair("message", result.getErrorMessage()));
       }
    }

src/main/java/com/kingsrook/qbits/backups/verify/BackupVerifier.java

@@ -47,6 +47,8 @@ import com.kingsrook.qbits.backups.storage.BackupStorageException;
 import com.kingsrook.qbits.backups.storage.StoredObjectInfo;
 import com.kingsrook.qqq.backend.core.context.QContext;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException;
 import com.kingsrook.qqq.backend.core.logging.QLogger;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
@@ -153,6 +155,10 @@ public class BackupVerifier
          }
          status = findings.status();
       }
+      catch(JobInterruptedException e)
+      {
+         throw (e);
+      }
       catch(VerifyException | BackupStorageException | QException e)
       {
          status = "ERROR";
@@ -188,7 +194,7 @@ public class BackupVerifier
    /*******************************************************************************
     ** STORAGE level: head every object, including every file-store object.
     *******************************************************************************/
-   void storageLevel(RunBundle bundle, VerifyFindings findings)
+   void storageLevel(RunBundle bundle, VerifyFindings findings) throws QException
    {
       BackupManifest manifest = bundle.manifest();
       BackupStorage  storage  = bundle.storage();
@@ -199,6 +205,7 @@ public class BackupVerifier
 
       for(BackupManifest.TableEntry table : manifest.tables())
       {
+         JobChunkScope.checkBetweenPages();
          for(BackupManifest.Part part : table.parts() == null ? List.<BackupManifest.Part>of() : table.parts())
          {
             findings.objectsChecked++;
@@ -231,6 +238,7 @@ public class BackupVerifier
       int missingFiles = 0;
       for(RunBundle.FileIndexEntry entry : bundle.distinctStoredFiles(findings))
       {
+         JobChunkScope.checkBetweenPages();
          findings.objectsChecked++;
          findings.filesChecked++;
          Optional<StoredObjectInfo> head = storage.head(entry.casKey());
@@ -262,6 +270,7 @@ public class BackupVerifier
 
       for(BackupManifest.TableEntry entry : manifest.tables())
       {
+         JobChunkScope.checkBetweenPages();
          QTableMetaData table = qInstance.getTable(entry.name());
          JsonlRowCodec  codec = null;
          boolean        roundTrip = false;
@@ -291,7 +300,7 @@ public class BackupVerifier
    /*******************************************************************************
     ** All parts of one table.
     *******************************************************************************/
-   private void verifyTable(RunBundle bundle, BackupManifest.TableEntry entry, JsonlRowCodec codec, boolean roundTrip, VerifyFindings findings)
+   private void verifyTable(RunBundle bundle, BackupManifest.TableEntry entry, JsonlRowCodec codec, boolean roundTrip, VerifyFindings findings) throws QException
    {
       MessageDigest rowsDigest = HashingOutputStream.newSha256();
       long          rows       = 0;
@@ -339,6 +348,11 @@ public class BackupVerifier
          }
          catch(QException e)
          {
+            JobInterruptedException interrupted = JobInterruptedException.find(e);
+            if(interrupted != null)
+            {
+               throw (interrupted);
+            }
             findings.mismatch("The part " + part.key() + " could not be read: " + e.getMessage());
          }
       }
@@ -451,6 +465,7 @@ public class BackupVerifier
 
       for(RunBundle.FileIndexEntry entry : bundle.distinctStoredFiles(findings))
       {
+         JobChunkScope.checkBetweenPages();
          QRecord fileObject = catalog.where(BackupsTables.FILE_OBJECT, "sha256", entry.sha256()).stream().filter(r -> r.getValue("deletedDate") == null).findFirst().orElse(null);
          if(fileObject != null && "PASSED".equals(fileObject.getValueString("lastVerifyStatus"))
             && fileObject.getValueInstant("lastVerifiedDate") != null && fileObject.getValueInstant("lastVerifiedDate").isAfter(cutoff))

src/main/java/com/kingsrook/qbits/backups/verify/PartScanner.java

@@ -29,6 +29,7 @@ import java.util.zip.GZIPInputStream;
 import com.kingsrook.qbits.backups.codec.HashingOutputStream;
 import com.kingsrook.qbits.backups.manifest.BackupManifest;
 import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
 
 
 /*******************************************************************************
@@ -80,6 +81,7 @@ final class PartScanner
     *******************************************************************************/
    static Scan scan(RunBundle bundle, BackupManifest.Part part, LineConsumer consumer) throws IOException, QException
    {
+      JobChunkScope.checkBetweenPages();
       MessageDigest storedDigest = HashingOutputStream.newSha256();
       MessageDigest rawDigest    = HashingOutputStream.newSha256();
       long          lines        = 0;
@@ -102,6 +104,10 @@ final class PartScanner
                String         line;
                while((line = reader.readLine()) != null)
                {
+                  if(lines % 1000 == 0)
+                  {
+                     JobChunkScope.checkBetweenPages();
+                  }
                   lines++;
                   consumer.accept(line);
                }

src/test/java/com/kingsrook/qbits/backups/BackupsQBitProducerTest.java

@@ -79,4 +79,26 @@ class BackupsQBitProducerTest
       assertThat(BackupsQBitConfig.PERMISSION_BASE_NAME).isEqualTo("backups");
    }
 
+
+
+   /*******************************************************************************
+    ** Long operations always enqueue, with no unsafe automatic external replay.
+    *******************************************************************************/
+   @Test
+   void testDurableProcesses() throws Exception
+   {
+      BackupsTestUtils.init();
+      for(String name : java.util.List.of("backupRunPlan", "backupRunScheduledPlan", "backupResumeRun", "backupRestore", "backupRestoreTable", "backupVerify"))
+      {
+         var process = com.kingsrook.qqq.backend.core.context.QContext.getQInstance().getProcess(name);
+         assertThat(process).as(name).isNotNull();
+         var durability = com.kingsrook.qqq.backend.core.jobs.DurableProcessMetaData.of(process);
+         assertThat(durability).as(name).isNotNull();
+         assertThat(durability.getAlwaysDurable()).isTrue();
+         assertThat(durability.getEligible()).isTrue();
+         assertThat(durability.getIdempotency()).isEqualTo(com.kingsrook.qqq.backend.core.jobs.model.JobIdempotency.NONE);
+         assertThat(durability.getChunkSize()).isEqualTo(1);
+      }
+   }
+
 }

src/test/java/com/kingsrook/qbits/backups/restore/RestoreSafetyTest.java

@@ -205,4 +205,24 @@ class RestoreSafetyTest
          List.of(new com.kingsrook.qqq.backend.core.model.data.QRecord().withValue("id", 99)), request, null)).hasMessageContaining("backups.restore");
    }
 
+
+
+   /*******************************************************************************
+    ** Job-page cancellation reaches the restore worker and its catalog status.
+    *******************************************************************************/
+   @Test
+   void testDurableCancellation() throws Exception
+   {
+      BackupRunResult run = RestoreTestSupport.backup(Map.of());
+      RestoreRequest request = RestoreTestSupport.request(run, RestoreRequest.MODE_UPSERT).withPreRestoreBackup(false);
+      try(var scope = com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.open(null, () ->
+      {
+         throw (new com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.Reason.CANCEL));
+      }))
+      {
+         assertThatThrownBy(() -> new RestoreEngine().execute(request)).isInstanceOf(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.class);
+      }
+      assertThat(new BackupCatalog().where(BackupsTables.RESTORE_RUN, "status", "CANCELLED")).hasSize(1);
+   }
+
 }

src/test/java/com/kingsrook/qbits/backups/run/BackupEngineTest.java

@@ -20,6 +20,7 @@ package com.kingsrook.qbits.backups.run;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.zip.GZIPInputStream;
@@ -28,11 +29,18 @@ import com.kingsrook.qbits.backups.manifest.BackupManifest;
 import com.kingsrook.qbits.backups.model.BackupsTables;
 import com.kingsrook.qbits.backups.scope.PlanSettings;
 import com.kingsrook.qbits.backups.scope.ScopeResolver;
+import com.kingsrook.qqq.backend.core.actions.async.AsyncJobCallback;
 import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope;
+import com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException;
+import com.kingsrook.qqq.backend.core.jobs.progress.JobLogLevel;
+import com.kingsrook.qqq.backend.core.jobs.progress.JobPhase;
+import com.kingsrook.qqq.backend.core.jobs.progress.JobProgress;
 import com.kingsrook.qqq.backend.core.model.data.QRecord;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 
 /*******************************************************************************
@@ -187,7 +195,9 @@ class BackupEngineTest
       assertThat(gunzip(BackupsTestUtils.BACKUPS.getBytes(refOnly.getRunPrefix() + "files/index-00001.jsonl.gz"))).contains("REFERENCE_ONLY");
 
       BackupsTestUtils.MEDIA.removeIgnoringLock("media/photo-1.jpg");
-      BackupRunResult missing = run(new BackupCatalog().where(BackupsTables.PLAN, "planKey", "nightly").get(0));
+      RecordingProgress progress = new RecordingProgress();
+      BackupRunResult missing = new BackupEngine().execute(BackupRunFactory.forPlan(new BackupCatalog().where(BackupsTables.PLAN, "planKey", "nightly").get(0), "MANUAL").withProgress(progress));
+      assertThat(progress.logs).anyMatch(l -> l.startsWith("WARN:") && l.contains("missing"));
       assertThat(missing.getFilesMissing()).isGreaterThan(0);
       assertThat(missing.getWarnings()).anyMatch(w -> w.contains("missing"));
       assertThat(missing.getStatus()).isEqualTo("SUCCEEDED_WITH_WARNINGS");
@@ -208,7 +218,7 @@ class BackupEngineTest
          BackupRunResult result = run(plan(Map.of()));
          assertThat(result.getStatus()).isEqualTo("FAILED");
          assertThat(result.isUsable()).isFalse();
-         assertThat(result.getErrorMessage()).contains("injected write failure");
+         assertThat(result.getErrorMessage()).contains("Backup failed");
          assertThat(BackupsTestUtils.BACKUPS.list(result.getRunPrefix() + "manifest.json")).isEmpty();
          assertThat(new BackupCatalog().get(BackupsTables.RUN, result.getRunId()).orElseThrow().getValueString("status")).isEqualTo("FAILED");
       }
@@ -281,4 +291,215 @@ class BackupEngineTest
       assertThat(scope.tables()).isEmpty();
       assertThat(scope.excluded()).extracting(ScopeResolver.ExcludedTable::reason).containsExactly("BACKUPS");
    }
+
+
+   /*******************************************************************************
+    ** A real plan reports work in order, relative artifact names and no row data.
+    *******************************************************************************/
+   @Test
+   void testJobProgress() throws Exception
+   {
+      RecordingProgress progress = new RecordingProgress();
+      BackupRunResult result = new BackupRunCoordinator().runPlan(plan(Map.of("prefix", "private-prefix")), "MANUAL", progress, null);
+      assertThat(result.isUsable()).isTrue();
+      assertThat(progress.phases).containsExactly("prepare", "tables", "media", "upload");
+      assertThat(progress.getSnapshot().getPhases()).extracting(JobPhase::getWeight).containsExactly(2.0, 75.0, 18.0, 5.0);
+      assertThat(progress.getSnapshot().getPhases()).extracting(JobPhase::getStatus)
+         .containsExactly(JobPhase.Status.DONE, JobPhase.Status.DONE, JobPhase.Status.DONE, JobPhase.Status.DONE);
+      assertThat(progress.counts).contains("tables:0/2 tables", "tables:1/2 tables", "tables:2/2 tables", "media:0/4 files", "media:4/4 files");
+      assertThat(progress.bytes).isSorted();
+      assertThat(progress.bytes.getLast()).isGreaterThan(0);
+      assertThat(progress.getSnapshot().getBytesTotal()).isNull();
+      assertThat(progress.messages).contains("Backing up author (1 of 2)", "Backing up blogPost (2 of 2)");
+      assertThat(progress.logs).anyMatch(l -> l.matches("INFO:author: 3 rows, .* MB in .* s"));
+      assertThat(progress.logs).anyMatch(l -> l.matches("INFO:blogPost: 7 rows, .* MB in .* s"));
+      assertThat(progress.logs).contains("INFO:Media: 4 files, 3 copied, 1 already stored, 0 missing", "INFO:Wrote checksums.sha256", "INFO:Wrote manifest.json", "INFO:Wrote files/index-00001.jsonl.gz");
+      assertThat(progress.logs).anyMatch(l -> l.startsWith("INFO:Wrote tables/blogPost/"));
+      for(String warning : result.getWarnings())
+      {
+         assertThat(progress.logs).contains("WARN:" + warning);
+      }
+      assertThat(String.join("\n", progress.logs)).doesNotContain("private-prefix", "do-not-leak", "Body of", "media/photo-");
+   }
+
+
+
+   /*******************************************************************************
+    ** Repeated backups report deduplicated media and complete all four phases.
+    *******************************************************************************/
+   @Test
+   void testProgressWithDeduplication() throws Exception
+   {
+      QRecord plan = plan(Map.of());
+      run(plan);
+      RecordingProgress progress = new RecordingProgress();
+      BackupRunResult result = new BackupEngine().execute(BackupRunFactory.forPlan(plan, "MANUAL").withProgress(progress));
+      assertThat(result.isUsable()).isTrue();
+      assertThat(progress.phases).containsExactly("prepare", "tables", "media", "upload");
+      assertThat(progress.getSnapshot().getPhases()).allMatch(p -> p.getStatus() == JobPhase.Status.DONE);
+      assertThat(progress.logs).contains("INFO:Media: 4 files, 0 copied, 4 already stored, 0 missing");
+   }
+
+
+
+   /*******************************************************************************
+    ** Arbitrary provider failures must not put credentials or row values in jobs.
+    *******************************************************************************/
+   @Test
+   void testProgressFailureDoesNotLeakProviderText() throws Exception
+   {
+      RecordingProgress progress = new RecordingProgress();
+      BackupRunRequest request = BackupRunFactory.forPlan(plan(Map.of()), "MANUAL").withProgress(progress)
+         .withReader((table, filter, start, size, transaction, sink) ->
+         {
+            throw (new com.kingsrook.qqq.backend.core.exceptions.QException("https://storage-user:storage-password@host/?secret=storage-secret accessKey=storage-access row=private-row"));
+         });
+      BackupRunResult result = new BackupEngine().execute(request);
+      assertThat(result.getStatus()).isEqualTo("FAILED");
+      assertThat(progress.logs).contains("ERROR:" + result.getErrorMessage());
+      assertThat(progress.getSnapshot().getPhases()).anyMatch(p -> p.getStatus() == JobPhase.Status.FAILED);
+      assertThat(String.join("\n", progress.logs) + result.getErrorMessage()).doesNotContain("storage-user", "storage-password", "storage-secret", "storage-access", "private-row", "https://");
+   }
+
+
+
+   /*******************************************************************************
+    ** The job page's cancel check must escape the engine to the durable worker.
+    *******************************************************************************/
+   @Test
+   void testJobCancellationBetweenPages() throws Exception
+   {
+      BackupRunRequest request = BackupRunFactory.forPlan(plan(Map.of("pageSize", 1)), "MANUAL");
+      int[] checks = { 0 };
+      try(JobChunkScope ignored = JobChunkScope.open(null, () ->
+      {
+         if(++checks[0] == 3)
+         {
+            throw (new JobInterruptedException(JobInterruptedException.Reason.CANCEL));
+         }
+      }))
+      {
+         assertThatThrownBy(() -> new BackupEngine().execute(request)).isInstanceOf(JobInterruptedException.class);
+      }
+      assertThat(new BackupCatalog().where(BackupsTables.RUN, "status", "CANCELLED")).hasSize(1);
+   }
+
+
+
+   /*******************************************************************************
+    ** Storage providers can include endpoint user-info and raw keys in exceptions.
+    *******************************************************************************/
+   @Test
+   void testStorageFailureDoesNotLeakCredentials() throws Exception
+   {
+      RecordingProgress progress = new RecordingProgress();
+      var storage = new com.kingsrook.qbits.backups.storage.InMemoryBackupStorage()
+      {
+         /*******************************************************************************
+          ** Deliberately unsafe provider diagnostics must stay out of job output.
+          *******************************************************************************/
+         @Override
+         public java.io.OutputStream openWrite(String key, com.kingsrook.qbits.backups.storage.WriteOptions options)
+         {
+            throw (new com.kingsrook.qbits.backups.storage.BackupStorageException(com.kingsrook.qbits.backups.storage.BackupStorageException.Reason.ACCESS_DENIED,
+               "https://endpoint-user:endpoint-password@storage accessKey=access-secret secretKey=secret-secret"));
+         }
+      };
+      BackupRunResult result = new BackupEngine().execute(BackupRunFactory.forPlan(plan(Map.of()), "MANUAL").withProgress(progress).withStorage(storage));
+      assertThat(result.getStatus()).isEqualTo("FAILED");
+      assertThat(progress.logs).contains("ERROR:" + result.getErrorMessage());
+      assertThat(String.join("\n", progress.logs) + result.getErrorMessage()).doesNotContain("endpoint-user", "endpoint-password", "access-secret", "secret-secret", "https://");
+   }
+
+
+
+   /*******************************************************************************
+    ** Failed backups must fail the durable chunk, not just add a summary line.
+    *******************************************************************************/
+   @Test
+   void testDurableLoadPropagatesFailure() throws Exception
+   {
+      var input = new com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput();
+      input.setRecords(List.of(plan(Map.of())));
+      BackupsTestUtils.BACKUPS.withFailWritesFor(key -> key.contains("/tables/"));
+      try(JobChunkScope scope = JobChunkScope.open(null).withProgress(new RecordingProgress()))
+      {
+         assertThatThrownBy(() -> new com.kingsrook.qbits.backups.processes.BackupsProcesses.RunPlanLoad().runOnePage(input,
+            new com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput()))
+            .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QException.class).hasMessageContaining("Backup failed");
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** Record values at call time: JobProgress's live snapshot is mutable.
+    *******************************************************************************/
+   static class RecordingProgress extends JobProgress
+   {
+      final List<String> phases = new ArrayList<>();
+      final List<String> counts = new ArrayList<>();
+      final List<Long> bytes = new ArrayList<>();
+      final List<String> logs = new ArrayList<>();
+      final List<String> messages = new ArrayList<>();
+
+      /*******************************************************************************
+       ** Use the core's safe standalone destination.
+       *******************************************************************************/
+      RecordingProgress()
+      {
+         super((AsyncJobCallback) null);
+      }
+
+      /*******************************************************************************
+       ** Record transitions.
+       *******************************************************************************/
+      @Override
+      public JobProgress phase(String name)
+      {
+         phases.add(name);
+         return (super.phase(name));
+      }
+
+      /*******************************************************************************
+       ** Record counts and their units.
+       *******************************************************************************/
+      @Override
+      public JobProgress done(long done)
+      {
+         counts.add(getSnapshot().getCurrentPhase() + ":" + done + "/" + getSnapshot().getTotal() + " " + getSnapshot().getUnitLabel());
+         return (super.done(done));
+      }
+
+      /*******************************************************************************
+       ** Record byte counts.
+       *******************************************************************************/
+      @Override
+      public JobProgress bytesDone(long done)
+      {
+         bytes.add(done);
+         return (super.bytesDone(done));
+      }
+
+      /*******************************************************************************
+       ** Capture text before core redaction as well as exercising core publication.
+       *******************************************************************************/
+      @Override
+      public JobProgress log(JobLogLevel level, String message)
+      {
+         logs.add(level + ":" + message);
+         return (super.log(level, message));
+      }
+
+      /*******************************************************************************
+       ** Record live messages.
+       *******************************************************************************/
+      @Override
+      public JobProgress message(String message)
+      {
+         messages.add(message);
+         return (super.message(message));
+      }
+   }
+
 }

src/test/java/com/kingsrook/qbits/backups/run/BackupRunCoordinatorTest.java

@@ -24,10 +24,10 @@ import java.util.List;
 import java.util.Map;
 import com.kingsrook.qbits.backups.BackupsRuntime;
 import com.kingsrook.qbits.backups.BackupsTestUtils;
-import com.kingsrook.qbits.backups.LifecycleTestSupport;
 import com.kingsrook.qbits.backups.LifecycleTestSupport.EtlSummaries;
 import com.kingsrook.qbits.backups.LifecycleTestSupport.MutableClock;
 import com.kingsrook.qbits.backups.LifecycleTestSupport.RecordingAlertSink;
+import com.kingsrook.qbits.backups.LifecycleTestSupport;
 import com.kingsrook.qbits.backups.manifest.BackupManifest;
 import com.kingsrook.qbits.backups.model.BackupsTables;
 import com.kingsrook.qbits.backups.processes.CancelRunProcess;
@@ -296,7 +296,7 @@ class BackupRunCoordinatorTest
       QRecord         plan   = LifecycleTestSupport.plan(perTableTinyParts);
       BackupRunResult failed = new BackupRunCoordinator().runPlan(plan, "MANUAL", request -> request.withReader(failingAfter(request.getReader(), "blogPost", 4)));
       assertThat(failed.getStatus()).isEqualTo("FAILED");
-      assertThat(failed.getErrorMessage()).contains("injected failure in blogPost");
+      assertThat(failed.getErrorMessage()).contains("Backup failed");
       assertThat(BackupsTestUtils.BACKUPS.list(failed.getRunPrefix() + "manifest.json")).isEmpty();
 
       QRecord failedRow = catalog.get(BackupsTables.RUN, failed.getRunId()).orElseThrow();
@@ -384,4 +384,59 @@ class BackupRunCoordinatorTest
       assertThat(manifest.table("blogPost").parts()).hasSize(7);
       assertThat(catalog.where(BackupsTables.ARTIFACT, "backupRunId", failed.getRunId()).stream().filter(a -> "blogPost".equals(a.getValueString("tableName")))).hasSize(7);
    }
+
+
+   /*******************************************************************************
+    ** Explicit retry of a durable chunk must reuse its backup and completed parts.
+    *******************************************************************************/
+   @Test
+   void testDurableRetryResumesSameRun() throws Exception
+   {
+      QRecord plan = LifecycleTestSupport.plan(Map.of());
+      BackupEngineTest.RecordingProgress progress = new BackupEngineTest.RecordingProgress();
+      var identity = new com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.ChunkIdentity(42, 8, 0, "backupRunPlan");
+      try(var scope = com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.open(null, null, identity).withProgress(progress))
+      {
+         BackupsTestUtils.BACKUPS.withFailWritesFor(key -> key.contains("/tables/blogPost/"));
+         BackupRunResult failed = new BackupRunCoordinator().runPlan(plan, "MANUAL");
+         assertThat(failed.getStatus()).isEqualTo("FAILED");
+         BackupsTestUtils.BACKUPS.withFailWritesFor(key -> false);
+         // A worker death can leave RUNNING rather than FAILED; identity, not age,
+         // ties this recovery to the same durable job.
+         catalog.update(BackupsTables.RUN, failed.getRunId(), Map.of("status", "RUNNING"));
+         BackupRunResult resumed = new BackupRunCoordinator().runPlan(plan, "MANUAL");
+         assertThat(resumed.isUsable()).isTrue();
+         assertThat(resumed.getRunId()).isEqualTo(failed.getRunId());
+         assertThat(resumed.getRunPrefix()).isEqualTo(failed.getRunPrefix());
+         assertThat(progress.logs).contains("INFO:Resuming run " + failed.getRunId() + ": 1 tables already done");
+         assertThat(catalog.where(BackupsTables.RUN, "backupPlanId", plan.getValueInteger("id"))).hasSize(1);
+         int artifacts = catalog.where(BackupsTables.ARTIFACT, "backupRunId", failed.getRunId()).size();
+         BackupRunResult again = new BackupRunCoordinator().runPlan(plan, "MANUAL");
+         assertThat(again.getRunId()).isEqualTo(failed.getRunId());
+         assertThat(catalog.where(BackupsTables.ARTIFACT, "backupRunId", failed.getRunId())).hasSize(artifacts);
+      }
+   }
+
+
+
+   /*******************************************************************************
+    ** A durable job refused for capacity must not escape into the inline queue.
+    *******************************************************************************/
+   @Test
+   void testDurableCapacityRefusalDoesNotQueue() throws Exception
+   {
+      QRecord occupied = LifecycleTestSupport.plan(Map.of());
+      liveRun(occupied.getValueInteger("id"), "RUNNING", clock.instant());
+      QRecord requested = LifecycleTestSupport.plan(Map.of("planKey", "second", "name", "Second"));
+      var identity = new com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.ChunkIdentity(43, 9, 0, "backupRunPlan");
+      try(var scope = com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.open(null, null, identity))
+      {
+         org.assertj.core.api.Assertions.assertThatThrownBy(() -> new BackupRunCoordinator().runPlan(requested, "MANUAL"))
+            .isInstanceOf(com.kingsrook.qqq.backend.core.exceptions.QException.class).hasMessageContaining("could not start");
+      }
+      assertThat(catalog.queuedRuns()).isEmpty();
+      assertThat(catalog.where(BackupsTables.RUN, "runKey", "job-43-plan-" + requested.getValueInteger("id")))
+         .singleElement().satisfies(run -> assertThat(run.getValueString("status")).isEqualTo("FAILED"));
+   }
+
 }

src/test/java/com/kingsrook/qbits/backups/verify/BackupVerifierTest.java

@@ -591,4 +591,23 @@ class BackupVerifierTest
       }
       return (bytes.toByteArray());
    }
+
+
+   /*******************************************************************************
+    ** A verification must propagate a job interrupt, not report it as corruption.
+    *******************************************************************************/
+   @Test
+   void testDurableCancellation() throws Exception
+   {
+      BackupRunResult run = backup(plan(Map.of()));
+      try(var scope = com.kingsrook.qqq.backend.core.jobs.plans.JobChunkScope.open(null, () ->
+      {
+         throw (new com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.Reason.CANCEL));
+      }))
+      {
+         org.assertj.core.api.Assertions.assertThatThrownBy(() -> new BackupVerifier().verify(run.getRunId(), BackupVerifier.CONTENT, "MANUAL", "test"))
+            .isInstanceOf(com.kingsrook.qqq.backend.core.jobs.plans.JobInterruptedException.class);
+      }
+   }
+
 }