Compare commits

..
Author SHA1 Message Date
james.maesandClaude Sonnet 5.5 4022fa1eee test(verify): do not depend on which mismatch is reported first for a flipped byte
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 18:57:50 -05:00
james.maesandClaude Sonnet 5.5 6017f3ed2d test(sla): save the plan with a bucket that exists, then remove it from the config
Plan save now validates the bucket alias (lifecycle), so the SLA test creates the
plan while the bucket is configured and removes it afterwards.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 18:55:21 -05:00
james.maesandClaude Sonnet 5.5 01b60708f8 feat: run lifecycle - plan schedule, run lease, cancel, resume, GFS prune, object lock, plan validation
- BackupPlanSyncToScheduledJobProcess (+ verify variant) and the plan table wiring
- BackupRunCoordinator: one run per plan (live-heartbeat lease, QUEUED to RUNNING,
  SKIPPED_CONCURRENT warning), maxConcurrentRuns queueing, scheduled entry point
  (backupRunScheduledPlan) as the system session with plan-owner validation
- plan save validation (admin and read on resolved tables, object lock vs bucket
  capabilities, encryption rules, unsupported formats, planKey immutable after a run)
- PruneCalculator (GFS) and BackupPruner (retention classes, deletePermanent on the
  prune alias, kept: locked until, debris, file store mark-and-sweep)
- object lock applied at write time in the engine, parts catalogued as they complete
- Back up these records, Cancel, Resume (PER_TABLE, from the last complete part),
  Test destination and Prune processes with Review and Result lines

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-12 18:54:24 -05:00
james.maes 06c87afa20 Merge pull request #6 from QRun-IO/feature/restore
feat(restore): restore, catalog import and download (spec 9.3-9.7)
2026-10-12 18:53:53 -05:00
35 changed files with 7094 additions and 66 deletions

No files matched your search

@@ -149,6 +149,11 @@ 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);
@@ -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);
}
}
}
@@ -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))))
{
@@ -23,6 +23,13 @@ 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.model.metadata.fields.QFieldMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
@@ -91,14 +98,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"));
@@ -286,6 +293,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")
@@ -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")));
}
}
@@ -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.");
}
}
}
@@ -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()));
}
}
}
@@ -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);
@@ -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);
}
}
}
@@ -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);
}
}
}
@@ -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);
}
}
@@ -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);
}
}
}
@@ -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);
}
}
}
@@ -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));
}
}
}
@@ -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") + "/");
}
}
@@ -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));
}
}
@@ -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.
*******************************************************************************/
@@ -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));
}
@@ -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));
}
}
@@ -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);
}
}
}
@@ -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);
}
}
@@ -0,0 +1,67 @@
/*
* 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.List;
import java.util.Map;
import com.kingsrook.qbits.backups.codec.TableArtifactWriter;
/*******************************************************************************
** What a failed run already wrote, so the engine can carry on instead of starting
** over (spec 9, Resume): the same run key and prefix, every table that finished
** kept as it is, and the interrupted table continuing from the last key of its
** last complete part.
**
** @param runId the run row being resumed
** @param runKey the run key of the original attempt
** @param runPrefix the object prefix of the original attempt
** @param tables the state of each table that has any catalogued part
*******************************************************************************/
public record ResumeState(Integer runId, String runKey, String runPrefix, Map<String, TableState> tables)
{
/*******************************************************************************
** The state of one table.
**
** @param complete the table finished (its last part carries the rows hash)
** @param parts the complete parts, in order
** @param rowsHash the rows hash, only when complete
** @param rows the rows in the complete parts
** @param artifactIds the catalog artifact rows of the parts
*******************************************************************************/
public record TableState(boolean complete, List<TableArtifactWriter.PartResult> parts, String rowsHash, long rows, List<Integer> artifactIds)
{
/*******************************************************************************
** The last key written, or null when there are no parts.
*******************************************************************************/
public String lastKey()
{
return (parts.isEmpty() ? null : parts.get(parts.size() - 1).lastKey());
}
}
/*******************************************************************************
** The state of one table, or null when nothing of it was written.
*******************************************************************************/
public TableState table(String name)
{
return (tables.get(name));
}
}
@@ -0,0 +1,193 @@
/*
* 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.schedule;
import com.kingsrook.qbits.backups.BackupsQBitConfig;
import com.kingsrook.qbits.backups.BackupsRuntime;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
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.processes.QProcessMetaData;
import com.kingsrook.qqq.backend.core.model.scheduledjobs.ScheduledJob;
import com.kingsrook.qqq.backend.core.scheduler.processes.AbstractRecordSyncToScheduledJobProcess;
import com.kingsrook.qqq.backend.core.utils.StringUtils;
/*******************************************************************************
** Keeps one scheduledJob per active plan in step with the plan cron fields (spec
** 9.1), the scheduled reports pattern: foreignKeyType backups_plan, the plan id as
** the foreign key value, and a job that runs the scheduled backup entry point
** (backupRunScheduledPlan) for that plan. The plan stays the single editable
** source; the job shows up in the standard scheduled jobs screens.
**
** A plan with no cron expression is manual-only and has no job. A plan that is
** made inactive (or loses its cron) has its existing job switched off. Deleting a
** plan removes its job (the table customizer does that).
*******************************************************************************/
public class BackupPlanSyncToScheduledJobProcess extends AbstractRecordSyncToScheduledJobProcess
{
public static final String FOREIGN_KEY_TYPE = BackupsTables.PLAN;
/*******************************************************************************
** Whether the plan should have a live job.
*******************************************************************************/
protected boolean wantsLiveJob(QRecord plan)
{
return (Boolean.TRUE.equals(plan.getValueBoolean("isActive")) && StringUtils.hasContent(cronOf(plan)));
}
/*******************************************************************************
** The cron expression this job follows.
*******************************************************************************/
protected String cronOf(QRecord plan)
{
return (plan.getValueString("cronExpression"));
}
/*******************************************************************************
** Skip plans that need no job; switch off the job of a plan that no longer wants one.
*******************************************************************************/
@Override
public QRecord populateRecordToStore(RunBackendStepInput runBackendStepInput, QRecord destinationRecord, QRecord sourceRecord) throws QException
{
boolean exists = destinationRecord != null && destinationRecord.getValue("id") != null;
if(!wantsLiveJob(sourceRecord))
{
if(!exists)
{
return (null);
}
ScheduledJob off = new ScheduledJob(destinationRecord);
off.setIsActive(false);
return (off.toQRecord());
}
return (super.populateRecordToStore(runBackendStepInput, destinationRecord, sourceRecord));
}
/*******************************************************************************
** The cron, the time zone, the scheduler, and a readable label.
*******************************************************************************/
@Override
protected ScheduledJob customizeScheduledJob(ScheduledJob scheduledJob, QRecord sourceRecord) throws QException
{
scheduledJob.setCronExpression(cronOf(sourceRecord));
scheduledJob.setCronTimeZoneId(StringUtils.hasContent(sourceRecord.getValueString("cronTimeZoneId")) ? sourceRecord.getValueString("cronTimeZoneId") : "UTC");
scheduledJob.setIsActive(true);
//////////////////////////////////////////////////////////////////////////////////
// the base class keys the job by its source table; the verify job needs its own //
//////////////////////////////////////////////////////////////////////////////////
scheduledJob.setForeignKeyType(getScheduledJobForeignKeyType());
scheduledJob.setLabel(label(sourceRecord));
scheduledJob.setDescription(label(sourceRecord) + " (runs the " + getProcessNameScheduledJobParameter() + " process for backup plan " + sourceRecord.getValueInteger("id") + ")");
BackupsQBitConfig config = BackupsRuntime.getConfig();
if(!StringUtils.hasContent(scheduledJob.getSchedulerName()) && config != null && StringUtils.hasContent(config.getSchedulerName()))
{
scheduledJob.setSchedulerName(config.getSchedulerName());
}
return (scheduledJob);
}
/*******************************************************************************
**
*******************************************************************************/
protected String label(QRecord plan)
{
return ("Backup " + plan.getValueString("name"));
}
/*******************************************************************************
** The process metadata (the scheduled-job sync process).
*******************************************************************************/
@Override
public QProcessMetaData produce(QInstance qInstance) throws com.kingsrook.qqq.backend.core.exceptions.QException
{
QProcessMetaData process = super.produce(qInstance);
process.setPermissionRules(BackupsTables.permissionRules());
return (process);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
protected String getScheduledJobForeignKeyType()
{
return (FOREIGN_KEY_TYPE);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
protected String getRecordForeignKeyFieldName()
{
return ("id");
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
protected String getRecordForeignKeyPossibleValueSourceName()
{
return (BackupsTables.PLAN);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
protected String getSourceTableName()
{
return (BackupsTables.PLAN);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
protected String getProcessNameScheduledJobParameter()
{
return (BackupScheduledRunProcess.NAME);
}
}
@@ -0,0 +1,81 @@
/*
* 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.schedule;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
/*******************************************************************************
** The second scheduled job of a plan (spec 9.1): the one that follows
** verifyCronExpression and runs the scheduled verification of the plan's newest
** successful run. It has its own foreign key type (backups_plan_verify) so the
** two jobs of one plan do not collide.
**
** The process it runs, backupVerifyPlan, is registered by the verify work (spec
** 10.1); this class only owns the schedule.
*******************************************************************************/
public class BackupPlanVerifySyncToScheduledJobProcess extends BackupPlanSyncToScheduledJobProcess
{
public static final String FOREIGN_KEY_TYPE = BackupsTables.PLAN + "_verify";
public static final String VERIFY_PROCESS_NAME = "backupVerifyPlan";
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
protected String cronOf(QRecord plan)
{
return (plan.getValueString("verifyCronExpression"));
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
protected String label(QRecord plan)
{
return ("Verify backup " + plan.getValueString("name"));
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
protected String getScheduledJobForeignKeyType()
{
return (FOREIGN_KEY_TYPE);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
protected String getProcessNameScheduledJobParameter()
{
return (VERIFY_PROCESS_NAME);
}
}
@@ -0,0 +1,158 @@
/*
* 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.schedule;
import java.util.List;
import java.util.Map;
import java.util.function.Predicate;
import com.kingsrook.qbits.backups.BackupsRuntime;
import com.kingsrook.qbits.backups.alerts.BackupAlertSink;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.permissions.BackupsPermissions;
import com.kingsrook.qbits.backups.processes.BackupsProcesses;
import com.kingsrook.qbits.backups.run.BackupCatalog;
import com.kingsrook.qbits.backups.run.BackupRunCoordinator;
import com.kingsrook.qbits.backups.run.BackupRunResult;
import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.exceptions.QUserFacingException;
import com.kingsrook.qqq.backend.core.logging.QLogger;
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.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
import com.kingsrook.qqq.backend.core.model.metadata.layout.QIcon;
import com.kingsrook.qqq.backend.core.model.metadata.processes.QBackendStepMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.processes.QFunctionInputMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.processes.QRecordListMetaData;
import static com.kingsrook.qqq.backend.core.logging.LogUtils.logPair;
/*******************************************************************************
** The scheduled entry point (spec 9.1, 11.1): the process a plan scheduled job
** runs. The scheduler gives it the plan record and its own system session; the
** run is a full-instance, system-session backup (securityScope FULL, requested by
** system:scheduler), which is why the plan owner is validated first: a plan whose
** owner has gone, or lost backups.admin, is not run silently. It raises
** backups.plan.owner_invalid, marks the plan OWNER_INVALID, and stops.
**
** Everything goes through the run coordinator, so a trigger that finds the plan
** already running is skipped visibly, and a full instance queues the run.
*******************************************************************************/
public class BackupScheduledRunProcess implements BackendStep
{
private static final QLogger LOG = QLogger.getLogger(BackupScheduledRunProcess.class);
public static final String NAME = "backupRunScheduledPlan";
private static volatile Predicate<String> ownerHasAdmin = userId -> true;
/*******************************************************************************
** The process metadata: not in the nav, a record process on backups_plan.
*******************************************************************************/
public static QProcessMetaData processMetaData()
{
QProcessMetaData process = new QProcessMetaData()
.withName(NAME)
.withLabel("Run scheduled backup")
.withTableName(BackupsTables.PLAN)
.withIcon(new QIcon().withName("backup"))
.withStep(new QBackendStepMetaData()
.withName("execute")
.withInputData(new QFunctionInputMetaData().withRecordListMetaData(new QRecordListMetaData().withTableName(BackupsTables.PLAN)))
.withCode(new QCodeReference(BackupScheduledRunProcess.class)));
process.setPermissionRules(BackupsTables.permissionRules());
return (BackupsProcesses.permission(process, BackupsPermissions.RUN));
}
/*******************************************************************************
** How the qbit asks whether a plan owner still exists and still holds
** backups.admin. The default accepts any owner id; an application with a user
** directory installs the real check here.
*******************************************************************************/
public static void setOwnerHasAdminCheck(Predicate<String> check)
{
ownerHasAdmin = check == null ? userId -> true : check;
}
/*******************************************************************************
** Why a plan may not run on its own now, or null when it may.
*******************************************************************************/
public static String ownerProblem(QRecord plan)
{
String owner = plan.getValueString("ownerUserId");
if(owner == null || owner.isBlank())
{
return ("the plan has no owner");
}
if(!ownerHasAdmin.test(owner))
{
return ("the plan owner " + owner + " no longer exists or no longer has the backups.admin permission");
}
return (null);
}
/*******************************************************************************
** Run the plan.
*******************************************************************************/
@Override
public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException
{
List<QRecord> records = input.getRecords();
if(records == null || records.isEmpty())
{
throw (new QUserFacingException("No backup plan was selected or found."));
}
BackupCatalog catalog = new BackupCatalog();
QRecord plan = catalog.get(BackupsTables.PLAN, records.get(0).getValueInteger("id")).orElseThrow(() -> new QUserFacingException("The backup plan no longer exists."));
if(!Boolean.TRUE.equals(plan.getValueBoolean("isActive")))
{
LOG.info("A scheduled backup fired for a plan that is not active; nothing was run", logPair("plan", plan.getValueString("planKey")));
output.addValue("status", "INACTIVE");
return;
}
String problem = ownerProblem(plan);
if(problem != null)
{
catalog.update(BackupsTables.PLAN, plan.getValueInteger("id"), Map.of("slaStatus", "OWNER_INVALID"));
BackupsRuntime.getAlertSink().raise("backups.plan.owner_invalid", "backups.plan.owner_invalid:" + plan.getValueString("planKey"), BackupAlertSink.Severity.WARNING,
"A scheduled backup did not run: " + problem, Map.of("plan", plan.getValueString("planKey")));
throw (new QUserFacingException("The scheduled backup did not run: " + problem + "."));
}
BackupsRuntime.getAlertSink().resolve("backups.plan.owner_invalid", "backups.plan.owner_invalid:" + plan.getValueString("planKey"));
BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "SCHEDULED");
output.addValue("status", result.getStatus());
output.addValue("runId", result.getRunId());
if(!"SUCCEEDED".equals(result.getStatus()) && !"SUCCEEDED_WITH_WARNINGS".equals(result.getStatus()))
{
LOG.info("A scheduled backup did not succeed", logPair("plan", plan.getValueString("planKey")), logPair("status", result.getStatus()), logPair("message", result.getErrorMessage()));
}
}
}
@@ -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;
import java.io.Serializable;
import java.time.Clock;
import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import com.kingsrook.qbits.backups.alerts.BackupAlertSink;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.run.BackupCatalog;
import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction;
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.RunProcessInput;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessOutput;
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 com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.StreamedETLWithFrontendProcess;
/*******************************************************************************
** Helpers for the lifecycle tests: a synthetic clock that the test moves, an alert
** sink that remembers what was raised, and plan creation.
*******************************************************************************/
public final class LifecycleTestSupport
{
/*******************************************************************************
** Utility class.
*******************************************************************************/
private LifecycleTestSupport()
{
}
/*******************************************************************************
** A clock the test sets.
*******************************************************************************/
public static class MutableClock extends Clock
{
private Instant now;
/*******************************************************************************
** Constructor.
*******************************************************************************/
public MutableClock(Instant start)
{
this.now = start;
}
/*******************************************************************************
** Move to an instant.
*******************************************************************************/
public void set(Instant instant)
{
this.now = instant;
}
/*******************************************************************************
** Move forward.
*******************************************************************************/
public void advance(Duration duration)
{
this.now = now.plus(duration);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public ZoneId getZone()
{
return (ZoneOffset.UTC);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public Clock withZone(ZoneId zone)
{
return (this);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public Instant instant()
{
return (now);
}
}
/*******************************************************************************
** Remembers every alert raised and resolved.
*******************************************************************************/
public static class RecordingAlertSink implements BackupAlertSink
{
public final List<String> raised = new ArrayList<>();
public final List<String> resolved = new ArrayList<>();
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public void raise(String type, String dedupeKey, Severity severity, String title, Map<String, String> detail)
{
raised.add(type + "|" + dedupeKey);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public void resolve(String type, String dedupeKey)
{
resolved.add(type + "|" + dedupeKey);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public void heartbeat(String monitorKey, long expectedEverySeconds, long graceSeconds)
{
}
}
/*******************************************************************************
** Install a synthetic clock on the runtime and the memory buckets.
*******************************************************************************/
public static MutableClock installClock(Instant start)
{
MutableClock clock = new MutableClock(start);
BackupsRuntime.setClock(clock);
BackupsTestUtils.BACKUPS.withClock(clock);
BackupsTestUtils.MEDIA.withClock(clock);
return (clock);
}
/*******************************************************************************
** Put the system clock back.
*******************************************************************************/
public static void restoreClock()
{
BackupsRuntime.setClock(Clock.systemUTC());
BackupsTestUtils.BACKUPS.withClock(Clock.systemUTC());
BackupsTestUtils.MEDIA.withClock(Clock.systemUTC());
}
/*******************************************************************************
** Insert a plan (through the table, so save validation runs) and return it.
*******************************************************************************/
public static QRecord plan(Map<String, Serializable> overrides) throws Exception
{
Map<String, Serializable> values = new LinkedHashMap<>(Map.of("name", "Nightly", "planKey", "nightly", "isActive", true,
"scopeType", "INSTANCE", "storageBucketAlias", "backups", "slaHours", 26));
values.putAll(overrides);
Integer id = new BackupCatalog().insert(BackupsTables.PLAN, values);
return (new BackupCatalog().get(BackupsTables.PLAN, id).orElseThrow());
}
/*******************************************************************************
** The two summaries of a streamed ETL run: what Review showed, and what Result said.
*******************************************************************************/
public record EtlSummaries(List<ProcessSummaryLine> review, List<ProcessSummaryLine> result, RunProcessOutput output)
{
/*******************************************************************************
** The text of a summary, count first: "3 runs will be deleted".
*******************************************************************************/
public static List<String> texts(List<ProcessSummaryLine> lines, boolean past)
{
List<String> texts = new ArrayList<>();
for(ProcessSummaryLine line : lines)
{
texts.add(line.getCount() + " " + (past ? (line.getCount() == 1 ? line.getSingularPastMessage() : line.getPluralPastMessage())
: (line.getCount() == 1 ? line.getSingularFutureMessage() : line.getPluralFutureMessage())) + (line.getMessageSuffix() == null ? "" : line.getMessageSuffix()));
}
return (texts);
}
}
/*******************************************************************************
** Run a streamed ETL process of the qbit over the records of a table matching a
** field, through Review and then Result.
*******************************************************************************/
@SuppressWarnings("unchecked")
public static EtlSummaries runEtl(String processName, String field, List<? extends Serializable> values) throws QException
{
RunProcessInput input = new RunProcessInput();
input.setProcessName(processName);
input.addValue(StreamedETLWithFrontendProcess.FIELD_DEFAULT_QUERY_FILTER, new QQueryFilter(new QFilterCriteria(field, QCriteriaOperator.IN, values)));
///////////////////////////////////////////////////////////////////////////////////
// the processes validate every record in the first call (full validation is on), //
// so Review is ready after it; the second call executes and gives the Result //
///////////////////////////////////////////////////////////////////////////////////
RunProcessOutput output = new RunProcessAction().execute(input);
List<ProcessSummaryLine> review = (List<ProcessSummaryLine>) output.getValues().get(StreamedETLWithFrontendProcess.FIELD_VALIDATION_SUMMARY);
input.addValue("processUUID", output.getProcessUUID());
input.setStartAfterStep("review");
output = new RunProcessAction().execute(input);
List<ProcessSummaryLine> result = (List<ProcessSummaryLine>) output.getValues().get(StreamedETLWithFrontendProcess.FIELD_PROCESS_SUMMARY);
return (new EtlSummaries(review, result, output));
}
}
@@ -0,0 +1,279 @@
/*
* 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.EnumSet;
import java.util.LinkedHashMap;
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.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.InMemoryBackupStorage;
import com.kingsrook.qqq.backend.core.actions.permissions.PermissionsHelper;
import com.kingsrook.qqq.backend.core.actions.permissions.TablePermissionSubType;
import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.permissions.PermissionLevel;
import com.kingsrook.qqq.backend.core.model.metadata.permissions.QPermissionRules;
import com.kingsrook.qqq.backend.core.model.session.QSession;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/*******************************************************************************
** What a plan may be saved as (spec 4.1, 6.2, 11.1, 11.3): the refusals are clear
** messages at save time, not failures on the first night.
*******************************************************************************/
class BackupPlanValidationTest
{
private final BackupCatalog catalog = new BackupCatalog();
/*******************************************************************************
**
*******************************************************************************/
@BeforeEach
void setUp() throws Exception
{
BackupsTestUtils.init();
BackupsTestUtils.loadBlog(3);
BackupValidatorReset.reset();
}
/*******************************************************************************
** Save a plan through the table, returning the record (with its errors) so the
** test can read what the save said.
*******************************************************************************/
private QRecord save(Map<String, Serializable> overrides) throws QException
{
Map<String, Serializable> values = new LinkedHashMap<>(Map.of("name", "P" + System.nanoTime(), "planKey", "p" + System.nanoTime(), "isActive", true, "scopeType", "INSTANCE", "storageBucketAlias", "backups"));
values.putAll(overrides);
QRecord record = new QRecord();
record.setValues(values);
InsertInput input = new InsertInput(BackupsTables.PLAN);
input.setRecords(List.of(record));
input.withPermissionBypass("test");
return (new InsertAction().execute(input).getRecords().get(0));
}
/*******************************************************************************
** Update one plan through the table, returning the record with its errors.
*******************************************************************************/
private QRecord update(Integer id, String field, Serializable value) throws QException
{
UpdateInput input = new UpdateInput(BackupsTables.PLAN);
input.setRecords(List.of(new QRecord().withValue("id", id).withValue(field, value)));
input.withPermissionBypass("test");
return (new UpdateAction().execute(input).getRecords().get(0));
}
/*******************************************************************************
**
*******************************************************************************/
private static List<String> errors(QRecord record)
{
return (record.getErrors().stream().map(e -> e.getMessage()).toList());
}
/*******************************************************************************
** A plain plan saves, with a warning that no prune bucket is set (one key can
** write and delete).
*******************************************************************************/
@Test
void testAGoodPlanSaves() throws Exception
{
QRecord saved = save(Map.of());
assertThat(errors(saved)).isEmpty();
assertThat(saved.getValueInteger("id")).isNotNull();
assertThat(saved.getWarnings()).anyMatch(w -> w.getMessage().contains("No prune bucket"));
}
/*******************************************************************************
** An object lock the bucket cannot do is refused, with what the bucket can do; a
** lock needs a number of days; and a lock on a capable bucket is fine.
*******************************************************************************/
@Test
void testObjectLockIsCheckedAgainstTheBucketCapabilities() throws Exception
{
InMemoryBackupStorage noLock = new InMemoryBackupStorage().withCapabilities(EnumSet.of(BackupStorage.StorageCapability.MULTIPART));
BackupsRuntime.getConfig().withStorageResolver(alias -> "garage".equals(alias) ? noLock : "backups".equals(alias) ? BackupsTestUtils.BACKUPS : BackupsTestUtils.MEDIA);
assertThat(errors(save(Map.of("storageBucketAlias", "garage", "objectLockMode", "GOVERNANCE", "objectLockDays", 14))))
.singleElement().asString().contains("does not support object lock").contains("garage").contains("MULTIPART");
assertThat(errors(save(Map.of("objectLockMode", "COMPLIANCE")))).singleElement().asString().contains("number of days");
assertThat(errors(save(Map.of("objectLockMode", "GOVERNANCE", "objectLockDays", 14)))).isEmpty();
assertThat(errors(save(Map.of("storageBucketAlias", "garage")))).as("no lock asked, none needed").isEmpty();
assertThat(errors(save(Map.of("objectLockMode", "SOMETIMES")))).anyMatch(m -> m.contains("Unknown object lock mode"));
}
/*******************************************************************************
** An unknown bucket or prune bucket is refused.
*******************************************************************************/
@Test
void testUnknownBucketsAreRefused() throws Exception
{
assertThat(errors(save(Map.of("storageBucketAlias", "nowhere")))).singleElement().asString().contains("nowhere").contains("cannot be used");
assertThat(errors(save(Map.of("pruneBucketAlias", "nowhere")))).singleElement().asString().contains("prune bucket nowhere");
}
/*******************************************************************************
** Including sensitive fields only encrypted needs encryption; client-side
** encryption is refused until a cipher is installed, and then needs a key name.
*******************************************************************************/
@Test
void testEncryptionRules() throws Exception
{
assertThat(errors(save(Map.of("sensitiveFieldPolicy", "INCLUDE_ENCRYPTED_ONLY")))).singleElement().asString().contains("requires client-side encryption");
assertThat(errors(save(Map.of("encryptionMode", "CLIENT_AES256_GCM", "encryptionKeyRef", "kek")))).singleElement().asString().contains("not available in this version");
assertThat(errors(save(Map.of("sensitiveFieldPolicy", "INCLUDE_ENCRYPTED_ONLY", "encryptionMode", "CLIENT_AES256_GCM", "encryptionKeyRef", "kek")))).hasSize(1);
BackupPlanValidator.setEncryptionAvailable(true);
assertThat(errors(save(Map.of("sensitiveFieldPolicy", "INCLUDE_ENCRYPTED_ONLY", "encryptionMode", "CLIENT_AES256_GCM", "encryptionKeyRef", "kek")))).isEmpty();
assertThat(errors(save(Map.of("encryptionMode", "CLIENT_AES256_GCM")))).singleElement().asString().contains("encryption key reference");
}
/*******************************************************************************
** Formats, compressions and options that are not built are refused with a clear
** message that names what to use instead.
*******************************************************************************/
@Test
void testUnsupportedFormatsAreRefusedClearly() throws Exception
{
assertThat(errors(save(Map.of("format", "PARQUET")))).singleElement().asString().isEqualTo("The PARQUET format is not available yet. Use JSONL, CSV or JSONL_AND_CSV.");
assertThat(errors(save(Map.of("compression", "ZSTD")))).singleElement().asString().isEqualTo("ZSTD compression is not available yet. Use GZIP or NONE.");
assertThat(errors(save(Map.of("incrementalMode", "CHANGED_SINCE_LAST")))).singleElement().asString().contains("Incremental backups are not available yet");
assertThat(errors(save(Map.of("nativeDump", "PG_DUMP_CUSTOM")))).singleElement().asString().contains("Native database dumps are not available yet");
assertThat(errors(save(Map.of("format", "JSONL_AND_CSV", "compression", "NONE")))).isEmpty();
}
/*******************************************************************************
** Cron expressions, retention numbers and plan keys.
*******************************************************************************/
@Test
void testScheduleRetentionAndKeyRules() throws Exception
{
assertThat(errors(save(Map.of("cronExpression", "not a cron")))).singleElement().asString().contains("cronExpression is not a valid cron expression");
assertThat(errors(save(Map.of("verifyCronExpression", "0 0 4 ? * SUN")))).isEmpty();
assertThat(errors(save(Map.of("retainDaily", -1)))).singleElement().asString().isEqualTo("retainDaily cannot be negative.");
assertThat(errors(save(Map.of("planKey", "Has Spaces")))).singleElement().asString().contains("plan key must be lowercase");
}
/*******************************************************************************
** The plan key is part of every object key, so it is fixed once the plan has run.
*******************************************************************************/
@Test
void testPlanKeyIsImmutableAfterTheFirstRun() throws Exception
{
QRecord plan = save(Map.of("planKey", "nightly"));
QRecord renamed = update(plan.getValueInteger("id"), "planKey", "renamed");
assertThat(errors(renamed)).as("before the first run it can change").isEmpty();
assertThat(new BackupRunCoordinator().runPlan(catalog.get(BackupsTables.PLAN, plan.getValueInteger("id")).orElseThrow(), "MANUAL").isUsable()).isTrue();
QRecord again = update(plan.getValueInteger("id"), "planKey", "nightly-2");
assertThat(errors(again)).singleElement().asString().contains("cannot change after the plan has run");
QRecord fine = update(plan.getValueInteger("id"), "slaHours", 30);
assertThat(errors(fine)).isEmpty();
assertThat(catalog.get(BackupsTables.PLAN, plan.getValueInteger("id")).orElseThrow().getValueString("planKey")).isEqualTo("renamed");
}
/*******************************************************************************
** Saving an active plan needs backups.admin; an inactive plan is only a draft.
*******************************************************************************/
@Test
void testActivePlanNeedsTheAdminPermission() throws Exception
{
BackupsRuntime.getConfig().withOpenWhenAnonymous(false);
QContext.setQSession(new QSession().withPermission("backups.view"));
assertThat(errors(save(Map.of()))).singleElement().asString().contains("needs the backups.admin permission");
assertThat(errors(save(Map.of("isActive", false)))).as("a draft").isEmpty();
QContext.setQSession(new QSession().withPermission("backups.admin"));
assertThat(errors(save(Map.of()))).isEmpty();
}
/*******************************************************************************
** The saving user must be able to read every table the plan covers; the message
** lists the ones they cannot.
*******************************************************************************/
@Test
void testOwnerMustReadEveryTable() throws Exception
{
BackupsRuntime.getConfig().withOpenWhenAnonymous(false);
QContext.getQInstance().getTable("author").withPermissionRules(new QPermissionRules().withLevel(PermissionLevel.READ_WRITE_PERMISSIONS));
String authorRead = PermissionsHelper.getTablePermissionName("author", TablePermissionSubType.READ);
QContext.setQSession(new QSession().withPermission("backups.admin"));
assertThat(errors(save(Map.of()))).singleElement().asString().contains("not allowed to read").contains("author").doesNotContain("blogPost");
QContext.setQSession(new QSession().withPermissions("backups.admin", authorRead));
assertThat(errors(save(Map.of()))).isEmpty();
}
/*******************************************************************************
** The saving user becomes the owner when none is given, and is never overwritten.
*******************************************************************************/
@Test
void testSavingUserBecomesOwner() throws Exception
{
QContext.setQSession(new QSession().withUser(new com.kingsrook.qqq.backend.core.model.session.QUser().withIdReference("jmaes")));
QRecord mine = save(Map.of());
assertThat(mine.getValueString("ownerUserId")).isEqualTo("jmaes");
QRecord given = save(Map.of("ownerUserId", "someone"));
assertThat(given.getValueString("ownerUserId")).isEqualTo("someone");
}
}
@@ -0,0 +1,46 @@
/*
* 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 com.kingsrook.qbits.backups.schedule.BackupScheduledRunProcess;
/*******************************************************************************
** Puts the static hooks of the plan and schedule code back to their defaults, so
** a test that moves one does not leak it into the next.
*******************************************************************************/
public final class BackupValidatorReset
{
/*******************************************************************************
** Utility class.
*******************************************************************************/
private BackupValidatorReset()
{
}
/*******************************************************************************
** Reset.
*******************************************************************************/
public static void reset()
{
BackupPlanValidator.setEncryptionAvailable(false);
BackupScheduledRunProcess.setOwnerHasAdminCheck(null);
}
}
@@ -0,0 +1,285 @@
/*
* 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.ByteArrayInputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.time.Instant;
import java.util.List;
import java.util.Map;
import java.util.zip.GZIPInputStream;
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.manifest.BackupManifest;
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.run.BackupRunResult;
import com.kingsrook.qqq.backend.core.context.QContext;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.session.QSession;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/*******************************************************************************
** Back up these records, Test destination and Prune per retention, driven the way
** the UI drives them: through the streamed ETL process, Review first, then Result.
*******************************************************************************/
class LifecycleProcessesTest
{
private static final Instant DAY_1 = Instant.parse("2026-10-01T02:00:00Z");
private final BackupCatalog catalog = new BackupCatalog();
private MutableClock clock;
/*******************************************************************************
**
*******************************************************************************/
@BeforeEach
void setUp() throws Exception
{
BackupsTestUtils.init();
BackupsTestUtils.loadBlog(7);
clock = LifecycleTestSupport.installClock(DAY_1);
}
/*******************************************************************************
**
*******************************************************************************/
@AfterEach
void tearDown()
{
LifecycleTestSupport.restoreClock();
}
/*******************************************************************************
**
*******************************************************************************/
private static String gunzip(byte[] bytes) throws IOException
{
return (new String(new GZIPInputStream(new ByteArrayInputStream(bytes)).readAllBytes(), StandardCharsets.UTF_8));
}
/*******************************************************************************
** Every table that can be backed up gets the action; the qbit own tables, secrets
** and sessions do not.
*******************************************************************************/
@Test
void testExportActionIsOnEveryBackupTable()
{
assertThat(QContext.getQInstance().getProcess("blogPost.backupExportTable")).isNotNull();
assertThat(QContext.getQInstance().getProcess("author.backupExportTable")).isNotNull();
assertThat(QContext.getQInstance().getProcess("backups_run.backupExportTable")).isNull();
assertThat(QContext.getQInstance().getProcess("userSession.backupExportTable")).isNull();
assertThat(BackupsRuntime.processPermissions()).containsEntry("blogPost.backupExportTable", "backups.run");
}
/*******************************************************************************
** Back up these records: an ad-hoc run (no plan) of just the chosen records, with
** their files copied, restorable like any other, and by a user without
** backups.admin the sensitive field is left out.
*******************************************************************************/
@Test
void testExportSelectedRecordsLeavesSensitiveFieldsOutForNonAdmins() throws Exception
{
LifecycleTestSupport.plan(Map.of());
BackupsRuntime.getConfig().withOpenWhenAnonymous(false);
QContext.setQSession(new QSession().withPermission("backups.run"));
EtlSummaries summaries = LifecycleTestSupport.runEtl("blogPost.backupExportTable", "id", List.of(1, 2, 3));
assertThat(EtlSummaries.texts(summaries.review(), false)).contains("3 records will be backed up")
.anyMatch(t -> t.startsWith("1 field is sensitive and will be left out"));
assertThat(EtlSummaries.texts(summaries.result(), true)).contains("1 export finished with warnings", "3 records were written", "2 referenced files were copied or already stored");
QRecord run = catalog.query(BackupsTables.RUN, new com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter()).stream()
.filter(r -> r.getValue("backupPlanId") == null).findFirst().orElseThrow();
assertThat(run.getValueString("status")).isIn("SUCCEEDED", "SUCCEEDED_WITH_WARNINGS");
assertThat(run.getValueString("securityScope")).isEqualTo("USER");
assertThat(run.getValueString("runPrefix")).contains("/adhoc/");
assertThat(run.getValueLong("rowsTotal")).isEqualTo(3L);
assertThat(run.getValueInteger("filesReferenced")).as("posts 1 and 3 have a file").isEqualTo(2);
BackupManifest manifest = BackupManifest.read(BackupsTestUtils.BACKUPS.getBytes(run.getValueString("manifestKey")));
assertThat(manifest.tables()).extracting(BackupManifest.TableEntry::name).containsExactly("blogPost");
assertThat(manifest.table("blogPost").excludedFields()).extracting(BackupManifest.ExcludedField::name).containsExactly("secretNote");
String text = gunzip(BackupsTestUtils.BACKUPS.getBytes(run.getValueString("runPrefix") + manifest.table("blogPost").parts().get(0).key()));
assertThat(text.split("\n")).hasSize(3);
assertThat(text).doesNotContain("do-not-leak");
}
/*******************************************************************************
** With backups.admin the sensitive field is included, and Review says so.
*******************************************************************************/
@Test
void testExportIncludesSensitiveFieldsForAdmins() throws Exception
{
LifecycleTestSupport.plan(Map.of());
EtlSummaries summaries = LifecycleTestSupport.runEtl("blogPost.backupExportTable", "id", List.of(2));
assertThat(EtlSummaries.texts(summaries.review(), false)).anyMatch(t -> t.startsWith("1 field is sensitive and will be written in the clear"));
QRecord run = catalog.query(BackupsTables.RUN, new com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter()).stream()
.filter(r -> r.getValue("backupPlanId") == null).findFirst().orElseThrow();
BackupManifest manifest = BackupManifest.read(BackupsTestUtils.BACKUPS.getBytes(run.getValueString("manifestKey")));
assertThat(gunzip(BackupsTestUtils.BACKUPS.getBytes(run.getValueString("runPrefix") + manifest.table("blogPost").parts().get(0).key()))).contains("do-not-leak-2");
}
/*******************************************************************************
** With no active plan there is no bucket to write to: Review says so, and nothing runs.
*******************************************************************************/
@Test
void testExportNeedsADestination() throws Exception
{
EtlSummaries summaries = LifecycleTestSupport.runEtl("author.backupExportTable", "id", List.of(1));
assertThat(EtlSummaries.texts(summaries.review(), false)).anyMatch(t -> t.contains("export has no destination"));
assertThat(EtlSummaries.texts(summaries.result(), true)).anyMatch(t -> t.contains("export failed"));
assertThat(BackupsTestUtils.BACKUPS.list("")).isEmpty();
}
/*******************************************************************************
** Test destination: write, read, list and delete a probe, leaving nothing behind.
*******************************************************************************/
@Test
void testDestinationProbeWritesReadsListsAndDeletes() throws Exception
{
QRecord plan = LifecycleTestSupport.plan(Map.of());
EtlSummaries summaries = LifecycleTestSupport.runEtl(TestDestinationProcess.NAME, "id", List.of(plan.getValueInteger("id")));
List<String> result = EtlSummaries.texts(summaries.result(), true);
assertThat(result).anyMatch(t -> t.contains("Write a probe object succeeded"))
.anyMatch(t -> t.contains("Read it back succeeded"))
.anyMatch(t -> t.contains("List it succeeded"))
.anyMatch(t -> t.contains("Delete it succeeded"))
.anyMatch(t -> t.contains("Capabilities:") && t.contains("OBJECT_LOCK: supported"))
.anyMatch(t -> t.contains("No prune bucket is set"));
assertThat(BackupsTestUtils.BACKUPS.list("")).as("the probe is deleted again").isEmpty();
}
/*******************************************************************************
** A failing destination shows as an error line, not a thrown exception.
*******************************************************************************/
@Test
void testDestinationProbeReportsFailures() throws Exception
{
QRecord plan = LifecycleTestSupport.plan(Map.of());
BackupsTestUtils.BACKUPS.withFailWritesFor(key -> true);
EtlSummaries summaries = LifecycleTestSupport.runEtl(TestDestinationProcess.NAME, "id", List.of(plan.getValueInteger("id")));
assertThat(summaries.result().stream().filter(l -> l.getStatus() == com.kingsrook.qqq.backend.core.model.actions.processes.Status.ERROR)).hasSize(1);
assertThat(EtlSummaries.texts(summaries.result(), true)).anyMatch(t -> t.contains("Write a probe object failed") && t.contains("injected write failure"));
}
/*******************************************************************************
** On a locked bucket nothing is written (a probe object could not be deleted);
** the prefix is listed and the newest manifest is looked at with a head.
*******************************************************************************/
@Test
void testDestinationProbeIsHeadOnlyOnALockedBucket() throws Exception
{
QRecord plan = LifecycleTestSupport.plan(Map.of("objectLockMode", "GOVERNANCE", "objectLockDays", 3));
BackupRunResult run = new BackupRunCoordinator().runPlan(plan, "MANUAL");
assertThat(run.isUsable()).isTrue();
int objectsBefore = BackupsTestUtils.BACKUPS.list("").size();
EtlSummaries summaries = LifecycleTestSupport.runEtl(TestDestinationProcess.NAME, "id", List.of(plan.getValueInteger("id")));
assertThat(EtlSummaries.texts(summaries.review(), false)).anyMatch(t -> t.contains("destination is object locked: the test will only read"));
assertThat(EtlSummaries.texts(summaries.result(), true)).anyMatch(t -> t.contains("Write and delete a probe object succeeded: skipped"))
.anyMatch(t -> t.contains("List the plan prefix succeeded") && t.contains("objects under"))
.anyMatch(t -> t.contains("Head the newest manifest succeeded: found "));
assertThat(BackupsTestUtils.BACKUPS.list("")).hasSize(objectsBefore);
}
/*******************************************************************************
** Prune through the UI path: Review shows the runs to delete, the runs kept by
** class, the files to sweep and the space freed; Result does it.
*******************************************************************************/
@Test
void testPruneReviewAndResult() throws Exception
{
QRecord plan = LifecycleTestSupport.plan(Map.of("retainDaily", 2, "retainWeekly", 0, "retainMonthly", 0, "retainYearly", 0, "retainMinimum", 0));
BackupRunResult r1 = null;
for(int day = 1; day <= 4; day++)
{
clock.set(DAY_1.plus(Duration.ofDays(day - 1)));
if(day == 3)
{
new com.kingsrook.qqq.backend.core.actions.tables.DeleteAction().execute(new com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput("blogPost")
.withPrimaryKeys(new java.util.ArrayList<>(List.of(1, 2, 3, 4, 5, 6, 7))));
}
BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "SCHEDULED");
r1 = r1 == null ? result : r1;
}
EtlSummaries summaries = LifecycleTestSupport.runEtl(PruneProcess.NAME, "id", List.of(plan.getValueInteger("id")));
assertThat(EtlSummaries.texts(summaries.review(), false)).contains("1 plan will be pruned", "2 runs will be deleted", "2 runs are kept as daily backups",
"3 stored files that no retained backup references will be deleted");
assertThat(EtlSummaries.texts(summaries.review(), false)).anyMatch(t -> t.endsWith("MiB of storage will be freed"));
assertThat(EtlSummaries.texts(summaries.result(), true)).contains("1 plan was pruned", "2 runs were deleted", "3 stored files were deleted");
assertThat(BackupsTestUtils.BACKUPS.list(r1.getRunPrefix())).isEmpty();
}
/*******************************************************************************
** The run table action prunes the plans of the selected runs, once each.
*******************************************************************************/
@Test
void testPruneRunsActionPrunesTheirPlans() throws Exception
{
QRecord plan = LifecycleTestSupport.plan(Map.of("retainDaily", 1, "retainWeekly", 0, "retainMonthly", 0, "retainYearly", 0, "retainMinimum", 0));
BackupRunResult first = null;
BackupRunResult second = null;
for(int day = 1; day <= 3; day++)
{
clock.set(DAY_1.plus(Duration.ofDays(day - 1)));
BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "SCHEDULED");
first = first == null ? result : first;
second = day == 2 ? result : second;
}
EtlSummaries summaries = LifecycleTestSupport.runEtl(PruneProcess.NAME_RUNS, "id", List.of(first.getRunId(), second.getRunId()));
assertThat(EtlSummaries.texts(summaries.review(), false)).contains("1 plan will be pruned", "2 runs will be deleted");
assertThat(EtlSummaries.texts(summaries.result(), true)).contains("2 runs were deleted");
}
}
@@ -177,7 +177,8 @@ class VerifyProcessesTest
List<ProcessSummaryLineInterface> summary = lines(output, StreamedETLWithFrontendProcess.FIELD_PROCESS_SUMMARY);
ProcessSummaryLine failed = (ProcessSummaryLine) summary.stream().filter(l -> ((ProcessSummaryLine) l).getStatus() == Status.ERROR).findFirst().orElseThrow();
assertThat(failed.getCount()).isEqualTo(1);
assertThat(failed.getMessageSuffix()).contains("corrupted");
// which mismatch is reported first depends on where the flipped byte lands in the gzip stream
assertThat(failed.getMessageSuffix()).matches("(?s).*(corrupted|cannot be parsed|checksum|could not be read).*");
assertThat(VerifyTestSupport.run(result.getRunId()).getValueString("verificationStatus")).isEqualTo("FAILED");
}
@@ -0,0 +1,543 @@
/*
* 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.InputStream;
import java.io.OutputStream;
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.Optional;
import java.util.Set;
import com.kingsrook.qbits.backups.BackupsRuntime;
import com.kingsrook.qbits.backups.BackupsTestUtils;
import com.kingsrook.qbits.backups.LifecycleTestSupport;
import com.kingsrook.qbits.backups.LifecycleTestSupport.MutableClock;
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.run.BackupRunResult;
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.actions.tables.DeleteAction;
import com.kingsrook.qqq.backend.core.model.actions.tables.delete.DeleteInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/*******************************************************************************
** Prune end to end on the memory backend with a synthetic clock: real runs are
** written day by day, then pruned. Covers the GFS result on real objects, the
** file-store mark-and-sweep, object lock (kept: locked until), debris, the
** restore and verification guards, the prune alias, and Review writing nothing.
*******************************************************************************/
class BackupPrunerTest
{
private static final Instant DAY_1 = Instant.parse("2026-10-01T02:00:00Z");
private MutableClock clock;
private final BackupCatalog catalog = new BackupCatalog();
/*******************************************************************************
**
*******************************************************************************/
@BeforeEach
void setUp() throws Exception
{
BackupsTestUtils.init();
BackupsTestUtils.loadBlog(7);
clock = LifecycleTestSupport.installClock(DAY_1);
}
/*******************************************************************************
**
*******************************************************************************/
@AfterEach
void tearDown()
{
LifecycleTestSupport.restoreClock();
}
/*******************************************************************************
** A plan that keeps `daily` days and nothing else.
*******************************************************************************/
private QRecord plan(int daily, Map<String, Serializable> more) throws Exception
{
Map<String, Serializable> values = new LinkedHashMap<>(Map.of("retainDaily", daily, "retainWeekly", 0, "retainMonthly", 0, "retainYearly", 0, "retainMinimum", 0));
values.putAll(more);
return (LifecycleTestSupport.plan(values));
}
/*******************************************************************************
** Run the plan on a given day.
*******************************************************************************/
private BackupRunResult runOnDay(QRecord plan, int day) throws Exception
{
clock.set(DAY_1.plus(Duration.ofDays(day - 1)));
BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "SCHEDULED");
assertThat(result.isUsable()).as("run on day " + day + ": " + result.getErrorMessage()).isTrue();
return (result);
}
/*******************************************************************************
**
*******************************************************************************/
private void deleteAllPosts() throws Exception
{
new DeleteAction().execute(new DeleteInput("blogPost").withPrimaryKeys(new ArrayList<>(List.of(1, 2, 3, 4, 5, 6, 7))));
}
/*******************************************************************************
**
*******************************************************************************/
private QRecord run(Integer id) throws Exception
{
return (catalog.get(BackupsTables.RUN, id).orElseThrow());
}
/*******************************************************************************
** Four nightly runs, keep two days: the two oldest go, their objects go, their
** rows stay as history, and the files only they referenced are swept. Review
** shows the same numbers and writes nothing.
*******************************************************************************/
@Test
void testPruneDeletesOldRunsAndSweepsTheFileStore() throws Exception
{
QRecord plan = plan(2, Map.of());
BackupRunResult r1 = runOnDay(plan, 1);
BackupRunResult r2 = runOnDay(plan, 2);
deleteAllPosts();
BackupRunResult r3 = runOnDay(plan, 3);
BackupRunResult r4 = runOnDay(plan, 4);
clock.set(DAY_1.plus(Duration.ofDays(4)));
List<QRecord> files = catalog.query(BackupsTables.FILE_OBJECT, new QQueryFilter());
assertThat(files).hasSize(3);
////////////
// review //
////////////
BackupPruner.PrunePlan review = new BackupPruner().plan(plan);
assertThat(review.toDelete()).extracting(BackupPruner.RunPlan::runId).containsExactlyInAnyOrder(r1.getRunId(), r2.getRunId());
assertThat(review.keptByClass()).containsEntry(PruneCalculator.RetentionClass.DAILY, 2);
assertThat(review.lockedNotKept()).isEmpty();
assertThat(review.sweep().deferredReason()).isNull();
assertThat(review.sweep().deletable()).hasSize(3);
assertThat(review.bytesToFree()).isPositive();
assertThat(run(r1.getRunId()).getValue("prunedDate")).as("review writes nothing").isNull();
assertThat(BackupsTestUtils.BACKUPS.list(r1.getRunPrefix())).isNotEmpty();
assertThat(run(r1.getRunId()).getValue("retentionClass")).isNull();
/////////////
// execute //
/////////////
BackupPruner.Outcome outcome = new BackupPruner().prune(plan, "tester");
assertThat(outcome.runsDeleted()).isEqualTo(2);
assertThat(outcome.runsLocked()).isZero();
assertThat(outcome.filesSwept()).isEqualTo(3);
assertThat(outcome.bytesFreed()).isEqualTo(review.bytesToFree());
for(BackupRunResult gone : List.of(r1, r2))
{
assertThat(BackupsTestUtils.BACKUPS.list(gone.getRunPrefix())).isEmpty();
QRecord row = run(gone.getRunId());
assertThat(row.getValue("prunedDate")).isNotNull();
assertThat(row.getValueString("prunedByUserId")).isEqualTo("tester");
assertThat(row.getValueString("retentionClass")).isEqualTo("NONE");
}
for(BackupRunResult kept : List.of(r3, r4))
{
assertThat(BackupsTestUtils.BACKUPS.getBytes(kept.getManifestKey())).isNotNull();
assertThat(run(kept.getRunId()).getValue("prunedDate")).isNull();
assertThat(run(kept.getRunId()).getValueString("retentionClass")).isEqualTo("DAILY");
}
for(QRecord file : catalog.query(BackupsTables.FILE_OBJECT, new QQueryFilter()))
{
assertThat(file.getValue("deletedDate")).isNotNull();
assertThat(BackupsTestUtils.BACKUPS.getBytes(file.getValueString("objectKey"))).isNull();
}
///////////////////////////////////////////////////////////////////////////
// pruning again finds nothing to do (the rows stay, but they are history) //
///////////////////////////////////////////////////////////////////////////
BackupPruner.Outcome again = new BackupPruner().prune(plan, "tester");
assertThat(again.runsDeleted()).isZero();
assertThat(again.filesSwept()).isZero();
}
/*******************************************************************************
** A file a retained run still references is not swept, and its referenceCount is
** the number of retained runs that list it.
*******************************************************************************/
@Test
void testFilesOfRetainedRunsAreKeptAndCounted() throws Exception
{
QRecord plan = plan(2, Map.of());
runOnDay(plan, 1);
runOnDay(plan, 2);
BackupRunResult r3 = runOnDay(plan, 3);
clock.set(DAY_1.plus(Duration.ofDays(3)));
BackupPruner.Outcome outcome = new BackupPruner().prune(plan, "tester");
assertThat(outcome.runsDeleted()).isEqualTo(1);
assertThat(outcome.filesSwept()).isZero();
List<QRecord> files = catalog.query(BackupsTables.FILE_OBJECT, new QQueryFilter());
assertThat(files).hasSize(3);
for(QRecord file : files)
{
assertThat(file.getValue("deletedDate")).isNull();
assertThat(file.getValueInteger("referenceCount")).as("referenced by the two retained runs").isEqualTo(2);
assertThat(BackupsTestUtils.BACKUPS.getBytes(file.getValueString("objectKey"))).isNotNull();
}
assertThat(r3.getFilesReferenced()).isEqualTo(4);
}
/*******************************************************************************
** Object lock is applied to every object at write time; a run whose objects are
** still locked is reported "kept: locked until" and not touched; once the lock
** has passed, the next prune deletes it.
*******************************************************************************/
@Test
void testLockedRunIsKeptUntilTheLockPasses() throws Exception
{
QRecord plan = plan(1, Map.of("objectLockMode", "GOVERNANCE", "objectLockDays", 3));
BackupRunResult r1 = runOnDay(plan, 1);
BackupRunResult r2 = runOnDay(plan, 2);
Instant r1LockedUntil = DAY_1.plus(Duration.ofDays(3));
List<StoredObjectInfo> r1Objects = BackupsTestUtils.BACKUPS.list(r1.getRunPrefix());
assertThat(r1Objects).isNotEmpty().allSatisfy(o -> assertThat(o.lockedUntil()).as(o.key()).isEqualTo(r1LockedUntil));
assertThat(run(r1.getRunId()).getValueInstant("lockedUntil")).isEqualTo(r1LockedUntil);
for(QRecord file : catalog.query(BackupsTables.FILE_OBJECT, new QQueryFilter()))
{
assertThat(BackupsTestUtils.BACKUPS.head(file.getValueString("objectKey")).orElseThrow().lockedUntil()).as("file store objects are locked too").isEqualTo(r1LockedUntil);
}
clock.set(DAY_1.plus(Duration.ofDays(2)).plusSeconds(3600));
BackupPruner.PrunePlan review = new BackupPruner().plan(plan);
assertThat(review.toDelete()).isEmpty();
assertThat(review.lockedNotKept()).extracting(BackupPruner.RunPlan::runId).containsExactly(r1.getRunId());
assertThat(review.lockedNotKept().get(0).lockedUntil()).isEqualTo(r1LockedUntil);
BackupPruner.Outcome outcome = new BackupPruner().prune(plan, "tester");
assertThat(outcome.runsDeleted()).isZero();
assertThat(outcome.runsLocked()).isEqualTo(1);
assertThat(outcome.messages()).anyMatch(m -> m.contains("kept: locked until " + r1LockedUntil));
assertThat(BackupsTestUtils.BACKUPS.list(r1.getRunPrefix())).hasSize(r1Objects.size());
assertThat(run(r1.getRunId()).getValue("prunedDate")).isNull();
clock.set(DAY_1.plus(Duration.ofDays(4)));
BackupPruner.Outcome later = new BackupPruner().prune(plan, "tester");
assertThat(later.runsDeleted()).isEqualTo(1);
assertThat(BackupsTestUtils.BACKUPS.list(r1.getRunPrefix())).isEmpty();
assertThat(run(r1.getRunId()).getValue("prunedDate")).isNotNull();
assertThat(BackupsTestUtils.BACKUPS.getBytes(r2.getManifestKey())).isNotNull();
}
/*******************************************************************************
** A content-addressed object nobody references but that is still locked is kept
** and reported, with its lock date recorded on its row.
*******************************************************************************/
@Test
void testLockedUnreferencedFileObjectIsKept() throws Exception
{
QRecord plan = plan(1, Map.of());
runOnDay(plan, 1);
deleteAllPosts();
clock.set(DAY_1.plus(Duration.ofDays(1)));
String sha = "ab".repeat(32);
String key = "qqq-backups/testapp/_files/sha256/ab/ab/" + sha;
Instant until = DAY_1.plus(Duration.ofDays(10));
try(OutputStream out = BackupsTestUtils.BACKUPS.openWrite(key, WriteOptions.create().withObjectLock("GOVERNANCE", until)))
{
out.write("old file".getBytes());
}
Integer fileId = catalog.insert(BackupsTables.FILE_OBJECT, Map.of("sha256", sha, "sizeBytes", 8L, "storageBucketAlias", "backups", "objectKey", key, "referenceCount", 0));
BackupPruner.PrunePlan review = new BackupPruner().plan(plan);
assertThat(review.sweep().locked()).hasSize(1).allSatisfy(c -> assertThat(c.lockedUntil()).isEqualTo(until));
BackupPruner.Outcome outcome = new BackupPruner().prune(plan, "tester");
assertThat(outcome.filesLocked()).isEqualTo(1);
assertThat(outcome.messages()).anyMatch(m -> m.contains("kept: locked until " + until));
assertThat(BackupsTestUtils.BACKUPS.getBytes(key)).isNotNull();
QRecord row = catalog.get(BackupsTables.FILE_OBJECT, fileId).orElseThrow();
assertThat(row.getValue("deletedDate")).isNull();
assertThat(row.getValueInstant("lockedUntil")).isEqualTo(until);
clock.set(until.plusSeconds(1));
assertThat(new BackupPruner().prune(plan, "tester").filesSwept()).isEqualTo(1);
assertThat(BackupsTestUtils.BACKUPS.getBytes(key)).isNull();
}
/*******************************************************************************
** The prefix of a run that never got a manifest is debris once its newest object
** is older than 7 days, and not before.
*******************************************************************************/
@Test
void testDebrisOlderThanSevenDaysIsAlwaysDeleted() throws Exception
{
QRecord plan = plan(7, Map.of());
BackupsTestUtils.BACKUPS.withFailWritesFor(key -> key.contains("/tables/blogPost/"));
BackupRunResult failed = new BackupRunCoordinator().runPlan(plan, "SCHEDULED");
BackupsTestUtils.BACKUPS.withFailWritesFor(null);
assertThat(failed.getStatus()).isEqualTo("FAILED");
assertThat(BackupsTestUtils.BACKUPS.list(failed.getRunPrefix())).isNotEmpty();
BackupRunResult good = runOnDay(plan, 6);
clock.set(DAY_1.plus(Duration.ofDays(6)));
assertThat(new BackupPruner().plan(plan).debris()).as("6 days old is not debris yet").isEmpty();
clock.set(DAY_1.plus(Duration.ofDays(8)));
BackupPruner.PrunePlan review = new BackupPruner().plan(plan);
assertThat(review.debris()).hasSize(1).allSatisfy(d -> assertThat(d.prefix()).isEqualTo(failed.getRunPrefix()));
BackupPruner.Outcome outcome = new BackupPruner().prune(plan, "tester");
assertThat(outcome.debrisPrefixesDeleted()).isEqualTo(1);
assertThat(BackupsTestUtils.BACKUPS.list(failed.getRunPrefix())).isEmpty();
assertThat(run(failed.getRunId()).getValue("prunedDate")).isNotNull();
assertThat(BackupsTestUtils.BACKUPS.getBytes(good.getManifestKey())).as("a good run is never debris").isNotNull();
}
/*******************************************************************************
** A run a restore used in the last 30 days is never deleted; one an older restore
** used is. The newest run that passed verification is always kept.
*******************************************************************************/
@Test
void testRestoreAndVerificationGuards() throws Exception
{
QRecord plan = plan(1, Map.of());
BackupRunResult r1 = runOnDay(plan, 1);
BackupRunResult r2 = runOnDay(plan, 2);
BackupRunResult r3 = runOnDay(plan, 3);
BackupRunResult r4 = runOnDay(plan, 4);
clock.set(DAY_1.plus(Duration.ofDays(40)));
catalog.insert(BackupsTables.RESTORE_RUN, Map.of("sourceBackupRunId", r1.getRunId(), "status", "SUCCEEDED", "startDate", clock.instant().minus(Duration.ofDays(29))));
catalog.insert(BackupsTables.RESTORE_RUN, Map.of("sourceBackupRunId", r2.getRunId(), "status", "SUCCEEDED", "startDate", clock.instant().minus(Duration.ofDays(31))));
catalog.update(BackupsTables.RUN, r3.getRunId(), Map.of("verificationStatus", "CONTENT_OK"));
BackupPruner.PrunePlan review = new BackupPruner().plan(plan);
assertThat(review.toDelete()).extracting(BackupPruner.RunPlan::runId).containsExactly(r2.getRunId());
assertThat(review.keptByGuard()).isEqualTo(2);
new BackupPruner().prune(plan, "tester");
assertThat(BackupsTestUtils.BACKUPS.list(r1.getRunPrefix())).as("used by a restore 29 days ago").isNotEmpty();
assertThat(BackupsTestUtils.BACKUPS.list(r2.getRunPrefix())).as("used by a restore 31 days ago").isEmpty();
assertThat(BackupsTestUtils.BACKUPS.list(r3.getRunPrefix())).as("newest verified").isNotEmpty();
assertThat(BackupsTestUtils.BACKUPS.list(r4.getRunPrefix())).isNotEmpty();
}
/*******************************************************************************
** Nothing in the file store is swept while a backup is running: its files are not
** in any index yet.
*******************************************************************************/
@Test
void testSweepIsDeferredWhileARunIsLive() throws Exception
{
QRecord plan = plan(1, Map.of());
runOnDay(plan, 1);
deleteAllPosts();
runOnDay(plan, 2);
clock.set(DAY_1.plus(Duration.ofDays(2)));
Integer otherPlan = catalog.insert(BackupsTables.PLAN, Map.of("name", "Other", "planKey", "other", "isActive", false, "scopeType", "INSTANCE", "storageBucketAlias", "backups"));
catalog.insert(BackupsTables.RUN, Map.of("backupPlanId", otherPlan, "runKey", "live-one", "status", "RUNNING", "heartbeatDate", clock.instant().minusSeconds(30),
"storageBucketAlias", "backups", "startDate", clock.instant().minusSeconds(60)));
BackupPruner.Outcome outcome = new BackupPruner().prune(plan, "tester");
assertThat(outcome.runsDeleted()).isEqualTo(1);
assertThat(outcome.filesSwept()).isZero();
assertThat(outcome.messages()).anyMatch(m -> m.contains("sweep skipped") && m.contains("running"));
assertThat(catalog.query(BackupsTables.FILE_OBJECT, new QQueryFilter())).allSatisfy(f -> assertThat(f.getValue("deletedDate")).isNull());
}
/*******************************************************************************
** Deletes go through the prune alias, not the backup alias.
*******************************************************************************/
@Test
void testDeletesGoThroughThePruneAlias() throws Exception
{
CountingStorage pruneSide = new CountingStorage(BackupsTestUtils.BACKUPS);
CountingStorage writeSide = new CountingStorage(BackupsTestUtils.BACKUPS);
BackupsRuntime.getConfig().withStorageResolver(alias -> switch(alias)
{
case "backups" -> writeSide;
case "backupsPrune" -> pruneSide;
case "media" -> BackupsTestUtils.MEDIA;
default -> null;
});
QRecord plan = plan(1, Map.of("pruneBucketAlias", "backupsPrune"));
runOnDay(plan, 1);
runOnDay(plan, 2);
clock.set(DAY_1.plus(Duration.ofDays(2)));
BackupPruner.Outcome outcome = new BackupPruner().prune(plan, "tester");
assertThat(outcome.runsDeleted()).isEqualTo(1);
assertThat(pruneSide.deletes).isPositive();
assertThat(writeSide.deletes).isZero();
}
/*******************************************************************************
** A storage that delegates, and counts permanent deletes.
*******************************************************************************/
static class CountingStorage implements BackupStorage
{
private final BackupStorage delegate;
int deletes = 0;
/*******************************************************************************
** Constructor.
*******************************************************************************/
CountingStorage(BackupStorage delegate)
{
this.delegate = delegate;
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public OutputStream openWrite(String key, WriteOptions options)
{
return (delegate.openWrite(key, options));
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public InputStream openRead(String key)
{
return (delegate.openRead(key));
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public Optional<StoredObjectInfo> head(String key)
{
return (delegate.head(key));
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public List<StoredObjectInfo> list(String prefix)
{
return (delegate.list(prefix));
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public void copy(String sourceKey, String destinationKey)
{
delegate.copy(sourceKey, destinationKey);
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public DeleteResult deletePermanent(String key)
{
deletes++;
return (delegate.deletePermanent(key));
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public String presignGet(String key, Duration ttl, String downloadName)
{
return (delegate.presignGet(key, ttl, downloadName));
}
/*******************************************************************************
** {@inheritDoc}
*******************************************************************************/
@Override
public Set<StorageCapability> capabilities()
{
return (delegate.capabilities());
}
}
}
@@ -0,0 +1,226 @@
/*
* 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.ZoneId;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import com.kingsrook.qbits.backups.prune.PruneCalculator.Decision;
import com.kingsrook.qbits.backups.prune.PruneCalculator.Guard;
import com.kingsrook.qbits.backups.prune.PruneCalculator.Policy;
import com.kingsrook.qbits.backups.prune.PruneCalculator.RetentionClass;
import com.kingsrook.qbits.backups.prune.PruneCalculator.RunInfo;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/*******************************************************************************
** GFS keep-set fixtures with dates written out, so each expectation can be read
** off a calendar.
*******************************************************************************/
class PruneCalculatorTest
{
private static final ZoneId UTC = ZoneId.of("UTC");
/*******************************************************************************
**
*******************************************************************************/
private static RunInfo run(int id, String instant)
{
return (new RunInfo(id, Instant.parse(instant), false, false));
}
/*******************************************************************************
**
*******************************************************************************/
private static Set<Integer> kept(List<Decision> decisions)
{
return (decisions.stream().filter(Decision::keep).map(Decision::id).collect(Collectors.toSet()));
}
/*******************************************************************************
** One run a night for ten nights, keep 7 daily: the seven newest stay.
*******************************************************************************/
@Test
void testDailyKeepsTheNewestDays()
{
List<RunInfo> runs = new ArrayList<>();
for(int day = 1; day <= 10; day++)
{
runs.add(run(day, String.format("2026-10-%02dT02:00:00Z", day)));
}
List<Decision> decisions = PruneCalculator.calculate(runs, new Policy(7, 0, 0, 0, 0), UTC);
assertThat(kept(decisions)).containsExactlyInAnyOrder(4, 5, 6, 7, 8, 9, 10);
assertThat(decisions.stream().filter(Decision::keep).map(Decision::retentionClass).distinct()).containsExactly(RetentionClass.DAILY);
}
/*******************************************************************************
** Two runs in one calendar day count as one day; the newest of the day is the one kept.
*******************************************************************************/
@Test
void testOnePerCalendarDay()
{
List<RunInfo> runs = List.of(
run(1, "2026-10-10T01:00:00Z"), run(2, "2026-10-10T13:00:00Z"),
run(3, "2026-10-11T01:00:00Z"), run(4, "2026-10-11T23:59:00Z"),
run(5, "2026-10-12T02:00:00Z"));
assertThat(kept(PruneCalculator.calculate(runs, new Policy(2, 0, 0, 0, 0), UTC))).containsExactlyInAnyOrder(4, 5);
assertThat(kept(PruneCalculator.calculate(runs, new Policy(3, 0, 0, 0, 0), UTC))).containsExactlyInAnyOrder(2, 4, 5);
}
/*******************************************************************************
** The calendar day is the plan time zone day, not the UTC day.
*******************************************************************************/
@Test
void testDayBoundaryFollowsTheTimeZone()
{
List<RunInfo> runs = List.of(run(1, "2026-10-11T23:30:00Z"), run(2, "2026-10-12T00:30:00Z"));
assertThat(kept(PruneCalculator.calculate(runs, new Policy(1, 0, 0, 0, 0), UTC))).containsExactly(2);
///////////////////////////////////////////////////////////////////////////////////////////
// in Chicago (UTC-5 in October) both are the evening of the 11th: one day, newest wins //
///////////////////////////////////////////////////////////////////////////////////////////
assertThat(kept(PruneCalculator.calculate(runs, new Policy(1, 0, 0, 0, 0), ZoneId.of("America/Chicago")))).containsExactly(2);
List<RunInfo> three = List.of(run(1, "2026-10-11T04:30:00Z"), run(2, "2026-10-11T06:30:00Z"), run(3, "2026-10-12T02:00:00Z"));
assertThat(kept(PruneCalculator.calculate(three, new Policy(2, 0, 0, 0, 0), ZoneId.of("America/Chicago")))).as("1 is the 10th; 2 and 3 are the 11th, where 3 is newest").containsExactlyInAnyOrder(1, 3);
assertThat(kept(PruneCalculator.calculate(three, new Policy(2, 0, 0, 0, 0), UTC))).containsExactlyInAnyOrder(2, 3);
}
/*******************************************************************************
** Weekly, monthly and yearly keep the newest of the last N periods that have a
** run, and the label is the longest lived rule that keeps the run.
*******************************************************************************/
@Test
void testWeeklyMonthlyYearlyAndLabels()
{
List<RunInfo> runs = new ArrayList<>();
int id = 0;
////////////////////////////////////////////////////////////////////////
// a run on the 1st, 8th, 15th, 22nd and 29th of Sep, Oct and Nov 2026 //
////////////////////////////////////////////////////////////////////////
for(int month : new int[] { 9, 10, 11 })
{
for(int day : new int[] { 1, 8, 15, 22, 29 })
{
runs.add(run(++id, String.format("2026-%02d-%02dT02:00:00Z", month, day)));
}
}
runs.add(run(100, "2025-12-31T02:00:00Z"));
runs.add(run(101, "2024-06-30T02:00:00Z"));
List<Decision> decisions = PruneCalculator.calculate(runs, new Policy(2, 3, 2, 3, 0), UTC);
Map<Integer, Decision> byId = decisions.stream().collect(Collectors.toMap(Decision::id, d -> d));
/////////////////////////////////////////////////////////////////////////
// newest two days: Nov 29 and Nov 22 (daily); but Nov 29 is also the //
// newest of November (monthly) and the newest of its week and its year //
/////////////////////////////////////////////////////////////////////////
assertThat(byId.get(15).retentionClass()).isEqualTo(RetentionClass.YEARLY);
assertThat(byId.get(14).retentionClass()).isEqualTo(RetentionClass.WEEKLY);
assertThat(byId.get(13).retentionClass()).isEqualTo(RetentionClass.WEEKLY);
/////////////////////////////////////////////////////////////////////////
// monthly keeps the newest of Nov and Oct; the Oct one is also a week //
/////////////////////////////////////////////////////////////////////////
assertThat(byId.get(10).retentionClass()).isEqualTo(RetentionClass.MONTHLY);
assertThat(byId.get(12).keep()).isFalse();
assertThat(byId.get(5).keep()).as("September is outside the last two months").isFalse();
/////////////////////////////////////////////////////////////////////////
// yearly: the newest of 2026, 2025 and 2024 //
/////////////////////////////////////////////////////////////////////////
assertThat(byId.get(100).retentionClass()).isEqualTo(RetentionClass.YEARLY);
assertThat(byId.get(101).retentionClass()).isEqualTo(RetentionClass.YEARLY);
assertThat(PruneCalculator.keptByClass(decisions).get(RetentionClass.YEARLY)).isEqualTo(3);
}
/*******************************************************************************
** ISO weeks cross the new year: 29 Dec 2026 to 3 Jan 2027 is one ISO week (2026-W53).
*******************************************************************************/
@Test
void testIsoWeekAcrossTheNewYear()
{
List<RunInfo> runs = List.of(run(1, "2026-12-29T02:00:00Z"), run(2, "2027-01-03T02:00:00Z"), run(3, "2027-01-04T02:00:00Z"));
List<Decision> decisions = PruneCalculator.calculate(runs, new Policy(0, 2, 0, 0, 0), UTC);
assertThat(kept(decisions)).as("2 and 1 share an ISO week, so 3 and 2 are the last two weeks").containsExactlyInAnyOrder(2, 3);
}
/*******************************************************************************
** The minimum keeps the newest runs even when no rule would.
*******************************************************************************/
@Test
void testMinimum()
{
List<RunInfo> runs = List.of(run(1, "2026-01-01T02:00:00Z"), run(2, "2026-02-01T02:00:00Z"), run(3, "2026-03-01T02:00:00Z"), run(4, "2026-04-01T02:00:00Z"));
List<Decision> decisions = PruneCalculator.calculate(runs, new Policy(0, 0, 0, 0, 3), UTC);
assertThat(kept(decisions)).containsExactlyInAnyOrder(2, 3, 4);
assertThat(decisions.stream().filter(d -> d.id() == 4).findFirst().orElseThrow().guards()).containsExactly(Guard.MINIMUM);
assertThat(decisions.stream().filter(d -> d.id() == 4).findFirst().orElseThrow().retentionClass()).isEqualTo(RetentionClass.NONE);
}
/*******************************************************************************
** The newest verified run is always kept, and so is a run a recent restore used.
*******************************************************************************/
@Test
void testVerifiedAndRestoreGuards()
{
List<RunInfo> runs = List.of(
new RunInfo(1, Instant.parse("2026-01-01T02:00:00Z"), true, false),
new RunInfo(2, Instant.parse("2026-01-02T02:00:00Z"), true, false),
new RunInfo(3, Instant.parse("2026-01-03T02:00:00Z"), false, true),
new RunInfo(4, Instant.parse("2026-01-04T02:00:00Z"), false, false),
new RunInfo(5, Instant.parse("2026-01-05T02:00:00Z"), false, false));
List<Decision> decisions = PruneCalculator.calculate(runs, new Policy(1, 0, 0, 0, 0), UTC);
assertThat(kept(decisions)).containsExactlyInAnyOrder(5, 3, 2);
assertThat(decisions.stream().filter(d -> d.id() == 2).findFirst().orElseThrow().guards()).containsExactly(Guard.NEWEST_VERIFIED);
assertThat(decisions.stream().filter(d -> d.id() == 3).findFirst().orElseThrow().guards()).containsExactly(Guard.RESTORE_REFERENCE);
}
/*******************************************************************************
** Nothing to decide, and a zero policy with no runs of note.
*******************************************************************************/
@Test
void testEmptyAndZeroPolicy()
{
assertThat(PruneCalculator.calculate(List.of(), new Policy(7, 4, 12, 0, 3), UTC)).isEmpty();
List<Decision> decisions = PruneCalculator.calculate(List.of(run(1, "2026-01-01T02:00:00Z")), new Policy(0, 0, 0, 0, 0), UTC);
assertThat(kept(decisions)).isEmpty();
assertThat(PruneCalculator.keptByClass(decisions)).isEmpty();
}
}
@@ -0,0 +1,387 @@
/*
* 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.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.manifest.BackupManifest;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.processes.CancelRunProcess;
import com.kingsrook.qbits.backups.processes.ResumeRunProcess;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.assertj.core.api.Assertions.assertThat;
/*******************************************************************************
** The run lease (one run per plan, an instance limit with queueing), cancel, and
** resume after an injected failure.
*******************************************************************************/
class BackupRunCoordinatorTest
{
private static final Instant START = Instant.parse("2026-10-12T02:00:00Z");
private MutableClock clock;
private RecordingAlertSink alerts;
private final BackupCatalog catalog = new BackupCatalog();
/*******************************************************************************
**
*******************************************************************************/
@BeforeEach
void setUp() throws Exception
{
BackupsTestUtils.init();
BackupsTestUtils.loadBlog(7);
clock = LifecycleTestSupport.installClock(START);
alerts = new RecordingAlertSink();
BackupsRuntime.setAlertSink(alerts);
}
/*******************************************************************************
**
*******************************************************************************/
@AfterEach
void tearDown()
{
LifecycleTestSupport.restoreClock();
}
/*******************************************************************************
** A run row that says another run of the plan is in flight.
*******************************************************************************/
private Integer liveRun(Integer planId, String status, Instant heartbeat) throws Exception
{
Map<String, Serializable> values = new java.util.LinkedHashMap<>();
values.put("backupPlanId", planId);
values.put("runKey", "live-" + System.nanoTime());
values.put("status", status);
values.put("startDate", heartbeat);
values.put("heartbeatDate", heartbeat);
values.put("storageBucketAlias", "backups");
return (catalog.insert(BackupsTables.RUN, values));
}
/*******************************************************************************
** A trigger that finds the plan already running is skipped, visibly: a run row
** that says so, and a warning. Nothing is written to storage.
*******************************************************************************/
@Test
void testSecondTriggerIsSkippedNotSilent() throws Exception
{
QRecord plan = LifecycleTestSupport.plan(Map.of());
Integer live = liveRun(plan.getValueInteger("id"), "RUNNING", clock.instant().minusSeconds(120));
BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "SCHEDULED");
assertThat(result.getStatus()).isEqualTo(BackupRunCoordinator.SKIPPED_CONCURRENT);
assertThat(result.getErrorMessage()).contains("SKIPPED_CONCURRENT").contains("already running");
QRecord row = catalog.get(BackupsTables.RUN, result.getRunId()).orElseThrow();
assertThat(row.getValueString("status")).isEqualTo("CANCELLED");
assertThat(row.getValueString("errorMessage")).startsWith("SKIPPED_CONCURRENT");
assertThat(row.getValueInteger("warningCount")).isEqualTo(1);
assertThat(alerts.raised).containsExactly("backups.run.skipped_concurrent|backups.run.skipped_concurrent:nightly");
assertThat(BackupsTestUtils.BACKUPS.list("")).isEmpty();
assertThat(catalog.get(BackupsTables.RUN, live).orElseThrow().getValueString("status")).as("the live run is left alone").isEqualTo("RUNNING");
}
/*******************************************************************************
** A run that stopped sending heartbeats (a killed pod) does not hold the lease.
*******************************************************************************/
@Test
void testStaleHeartbeatDoesNotBlock() throws Exception
{
QRecord plan = LifecycleTestSupport.plan(Map.of());
liveRun(plan.getValueInteger("id"), "RUNNING", clock.instant().minus(Duration.ofMinutes(11)));
BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "SCHEDULED");
assertThat(result.isUsable()).as(result.getErrorMessage()).isTrue();
QRecord row = catalog.get(BackupsTables.RUN, result.getRunId()).orElseThrow();
assertThat(row.getValueString("trigger")).isEqualTo("SCHEDULED");
assertThat(row.getValueString("requestedByUserId")).isEqualTo("system:scheduler");
assertThat(row.getValueString("securityScope")).isEqualTo("FULL");
}
/*******************************************************************************
** At the instance limit a run waits, QUEUED; when the live run is gone it starts,
** oldest first, and the plan lease still applies to queued runs.
*******************************************************************************/
@Test
void testInstanceLimitQueuesAndDrains() throws Exception
{
QRecord first = LifecycleTestSupport.plan(Map.of());
QRecord second = LifecycleTestSupport.plan(Map.of("name", "Second", "planKey", "second"));
Integer live = liveRun(first.getValueInteger("id"), "RUNNING", clock.instant().minusSeconds(30));
BackupRunResult queued = new BackupRunCoordinator().runPlan(second, "MANUAL");
assertThat(queued.getStatus()).isEqualTo(BackupRunCoordinator.QUEUED);
assertThat(catalog.get(BackupsTables.RUN, queued.getRunId()).orElseThrow().getValueString("status")).isEqualTo("QUEUED");
assertThat(BackupsTestUtils.BACKUPS.list("")).isEmpty();
assertThat(new BackupRunCoordinator().drainQueue()).as("no slot yet").isEmpty();
catalog.update(BackupsTables.RUN, live, Map.of("status", "SUCCEEDED"));
List<BackupRunResult> drained = new BackupRunCoordinator().drainQueue();
assertThat(drained).hasSize(1);
assertThat(drained.get(0).getRunId()).isEqualTo(queued.getRunId());
assertThat(drained.get(0).isUsable()).as(drained.get(0).getErrorMessage()).isTrue();
QRecord row = catalog.get(BackupsTables.RUN, queued.getRunId()).orElseThrow();
assertThat(row.getValueString("trigger")).isEqualTo("MANUAL");
assertThat(row.getValueString("runKey")).doesNotStartWith("queued-");
}
/*******************************************************************************
** A raised limit lets two plans run side by side.
*******************************************************************************/
@Test
void testMaxConcurrentRunsRaised() throws Exception
{
BackupsRuntime.getConfig().withMaxConcurrentRuns(2);
QRecord first = LifecycleTestSupport.plan(Map.of());
QRecord second = LifecycleTestSupport.plan(Map.of("name", "Second", "planKey", "second"));
liveRun(first.getValueInteger("id"), "RUNNING", clock.instant().minusSeconds(30));
assertThat(new BackupRunCoordinator().runPlan(second, "MANUAL").isUsable()).isTrue();
}
/*******************************************************************************
** Cancel is cooperative: the run row says CANCELLING, and the run stops at its
** next page, ends CANCELLED, leaves no manifest, and raises no failure alert.
*******************************************************************************/
@Test
void testCancelStopsAtTheNextPage() throws Exception
{
QRecord plan = LifecycleTestSupport.plan(Map.of("pageSize", 2));
int[] seen = { 0 };
BackupRunResult result = new BackupRunCoordinator().runPlan(plan, "MANUAL", request ->
{
TableReader keyset = request.getReader();
request.withReader((table, filter, startAfter, pageSize, transaction, sink) -> keyset.read(table, filter, startAfter, pageSize, transaction, record ->
{
sink.accept(record);
if("blogPost".equals(table.getName()) && ++seen[0] == 3)
{
try
{
QRecord running = catalog.where(BackupsTables.RUN, "backupPlanId", plan.getValueInteger("id")).get(0);
catalog.update(BackupsTables.RUN, running.getValueInteger("id"), Map.of("status", "CANCELLING"));
}
catch(Exception e)
{
throw (new IllegalStateException(e));
}
}
}));
});
assertThat(result.getStatus()).isEqualTo("CANCELLED");
assertThat(seen[0]).as("stopped at the next page, not at the end").isBetween(3, 5);
assertThat(result.isUsable()).isFalse();
assertThat(BackupsTestUtils.BACKUPS.list(result.getRunPrefix() + "manifest.json")).isEmpty();
QRecord row = catalog.get(BackupsTables.RUN, result.getRunId()).orElseThrow();
assertThat(row.getValueString("status")).isEqualTo("CANCELLED");
assertThat(row.getValue("endDate")).isNotNull();
assertThat(alerts.raised).isEmpty();
}
/*******************************************************************************
** The Cancel action: a live run is asked to stop, a queued run is cancelled at
** once, a run with no heartbeat is closed, and a finished run cannot be cancelled.
*******************************************************************************/
@Test
void testCancelProcess() throws Exception
{
QRecord plan = LifecycleTestSupport.plan(Map.of());
Integer live = liveRun(plan.getValueInteger("id"), "RUNNING", clock.instant().minusSeconds(10));
Integer queued = liveRun(plan.getValueInteger("id"), "QUEUED", clock.instant());
Integer stale = liveRun(plan.getValueInteger("id"), "RUNNING", clock.instant().minus(Duration.ofHours(1)));
Integer done = liveRun(plan.getValueInteger("id"), "SUCCEEDED", clock.instant());
EtlSummaries summaries = LifecycleTestSupport.runEtl(CancelRunProcess.NAME, "id", List.of(live, queued, stale, done));
assertThat(EtlSummaries.texts(summaries.review(), false)).contains("1 running run will be asked to stop at its next page", "1 queued run will be cancelled")
.anyMatch(t -> t.startsWith("1 run has lost its heartbeat")).anyMatch(t -> t.startsWith("1 run cannot be cancelled"));
assertThat(EtlSummaries.texts(summaries.result(), true)).contains("1 running run was asked to stop at its next page", "1 queued run was cancelled");
assertThat(catalog.get(BackupsTables.RUN, live).orElseThrow().getValueString("status")).isEqualTo("CANCELLING");
assertThat(catalog.get(BackupsTables.RUN, queued).orElseThrow().getValueString("status")).isEqualTo("CANCELLED");
assertThat(catalog.get(BackupsTables.RUN, stale).orElseThrow().getValueString("status")).isEqualTo("CANCELLED");
assertThat(catalog.get(BackupsTables.RUN, done).orElseThrow().getValueString("status")).isEqualTo("SUCCEEDED");
assertThat(catalog.isCancelRequested(live)).isTrue();
}
/*******************************************************************************
** The row reader that fails once, after some rows of one table.
*******************************************************************************/
private static TableReader failingAfter(TableReader delegate, String tableName, int rows)
{
int[] seen = { 0 };
return ((table, filter, startAfter, pageSize, transaction, sink) -> delegate.read(table, filter, startAfter, pageSize, transaction, record ->
{
if(tableName.equals(table.getName()) && ++seen[0] > rows)
{
throw (new IllegalStateException("injected failure in " + tableName));
}
sink.accept(record);
}));
}
/*******************************************************************************
** A run that fails in the middle of a table resumes: the finished table is kept,
** the interrupted table carries on after the last key of its last complete part,
** and the result is the same backup a clean run makes (same rows hash, same
** parts, same file index).
*******************************************************************************/
@Test
void testResumeAfterInjectedFailure() throws Exception
{
Map<String, Serializable> perTableTinyParts = Map.of("consistency", "PER_TABLE", "pageSize", 2, "targetPartBytes", 1L);
/////////////////////////////////////////////////////
// a clean run, for the numbers a resume must match //
/////////////////////////////////////////////////////
QRecord clean = LifecycleTestSupport.plan(Map.of("name", "Clean", "planKey", "clean", "consistency", "PER_TABLE", "pageSize", 2, "targetPartBytes", 1L));
BackupRunResult cleanResult = new BackupRunCoordinator().runPlan(clean, "MANUAL");
assertThat(cleanResult.isUsable()).as(cleanResult.getErrorMessage()).isTrue();
BackupManifest cleanManifest = BackupManifest.read(BackupsTestUtils.BACKUPS.getBytes(cleanResult.getManifestKey()));
///////////////////////////////////////////////////////////////
// the run that fails on the 5th of the 7 blogPost rows //
///////////////////////////////////////////////////////////////
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(BackupsTestUtils.BACKUPS.list(failed.getRunPrefix() + "manifest.json")).isEmpty();
QRecord failedRow = catalog.get(BackupsTables.RUN, failed.getRunId()).orElseThrow();
assertThat(failedRow.getValueString("consistencyAchieved")).isEqualTo("PER_TABLE");
ResumeState state = BackupResumePlanner.plan(failedRow);
assertThat(state.table("author").complete()).isTrue();
assertThat(state.table("blogPost").complete()).isFalse();
assertThat(state.table("blogPost").parts()).as("a part per row: 4 rows were written, the 5th failed before it was").hasSize(4);
assertThat(state.table("blogPost").lastKey()).isEqualTo("4");
assertThat(state.table("blogPost").rows()).isEqualTo(4);
byte[] authorPart = BackupsTestUtils.BACKUPS.getBytes(failed.getRunPrefix() + "tables/author/part-00001.jsonl.gz");
////////////
// review //
////////////
EtlSummaries summaries = LifecycleTestSupport.runEtl(ResumeRunProcess.NAME, "id", List.of(failed.getRunId()));
assertThat(EtlSummaries.texts(summaries.review(), false)).contains("1 run will be resumed", "1 table is complete and will be kept as it is")
.anyMatch(t -> t.startsWith("1 table will carry on from its last complete part") && t.contains("blogPost after key 4"));
assertThat(EtlSummaries.texts(summaries.result(), true)).contains("1 run succeeded after being resumed");
QRecord resumedRow = catalog.get(BackupsTables.RUN, failed.getRunId()).orElseThrow();
assertThat(resumedRow.getValueString("status")).isEqualTo("SUCCEEDED");
assertThat(resumedRow.getValueString("runKey")).isEqualTo(failed.getRunKey());
assertThat(resumedRow.getValueString("errorMessage")).isNull();
assertThat(resumedRow.getValueInstant("startDate")).as("the original start is kept").isEqualTo(failedRow.getValueInstant("startDate"));
assertThat(BackupsTestUtils.BACKUPS.getBytes(failed.getRunPrefix() + "tables/author/part-00001.jsonl.gz")).as("a finished table is not rewritten").isEqualTo(authorPart);
BackupManifest manifest = BackupManifest.read(BackupsTestUtils.BACKUPS.getBytes(resumedRow.getValueString("manifestKey")));
assertThat(manifest.table("blogPost").rowCount()).isEqualTo(7);
assertThat(manifest.table("blogPost").parts()).hasSize(7);
assertThat(manifest.table("blogPost").rowsHash()).isEqualTo(cleanManifest.table("blogPost").rowsHash());
assertThat(manifest.table("author").rowsHash()).isEqualTo(cleanManifest.table("author").rowsHash());
assertThat(manifest.table("blogPost").parts()).extracting(BackupManifest.Part::sha256Raw).isEqualTo(cleanManifest.table("blogPost").parts().stream().map(BackupManifest.Part::sha256Raw).toList());
assertThat(manifest.files().objects()).isEqualTo(cleanManifest.files().objects());
assertThat(BackupsTestUtils.BACKUPS.getBytes(failed.getRunPrefix() + "files/index-00001.jsonl.gz")).isNotNull();
List<QRecord> artifacts = catalog.where(BackupsTables.ARTIFACT, "backupRunId", failed.getRunId());
assertThat(artifacts.stream().filter(a -> "blogPost".equals(a.getValueString("tableName")) && a.getValue("rowsHash") != null)).as("only the last part carries the rows hash").hasSize(1);
assertThat(artifacts.stream().filter(a -> "TABLE_DATA".equals(a.getValueString("artifactType")))).hasSize(3 + 7);
assertThat(resumedRow.getValueLong("rowsTotal")).isEqualTo(10L);
}
/*******************************************************************************
** A run that is not failed, or that read from one snapshot, cannot be resumed,
** and says why.
*******************************************************************************/
@Test
void testResumeIsRefusedWhenItCannotCarryOn() throws Exception
{
QRecord ok = LifecycleTestSupport.plan(Map.of());
BackupRunResult good = new BackupRunCoordinator().runPlan(ok, "MANUAL");
BackupRunResult refused = new BackupRunCoordinator().resume(catalog.get(BackupsTables.RUN, good.getRunId()).orElseThrow());
assertThat(refused.getStatus()).isEqualTo("FAILED");
assertThat(refused.getErrorMessage()).contains("Only a failed, timed out or abandoned run can be resumed");
QRecord snapshot = LifecycleTestSupport.plan(Map.of("name", "Snap", "planKey", "snap", "consistency", "SNAPSHOT_REQUIRED"));
BackupRunResult failed = new BackupRunCoordinator().runPlan(snapshot, "MANUAL");
assertThat(failed.getStatus()).isEqualTo("FAILED");
BackupRunResult snapshotRefused = new BackupRunCoordinator().resume(catalog.get(BackupsTables.RUN, failed.getRunId()).orElseThrow());
assertThat(snapshotRefused.getErrorMessage()).contains("one consistent snapshot").contains("Run the plan again");
EtlSummaries summaries = LifecycleTestSupport.runEtl(ResumeRunProcess.NAME, "id", List.of(failed.getRunId()));
assertThat(EtlSummaries.texts(summaries.review(), false)).anyMatch(t -> t.startsWith("1 run cannot be resumed") && t.contains("one consistent snapshot"));
}
/*******************************************************************************
** A table written as CSV cannot rebuild its rows hash from its parts, so the
** interrupted table starts over, and the result is still right.
*******************************************************************************/
@Test
void testResumeRestartsAnInterruptedCsvTable() throws Exception
{
QRecord plan = LifecycleTestSupport.plan(Map.of("consistency", "PER_TABLE", "pageSize", 2, "targetPartBytes", 1L, "format", "CSV"));
BackupRunResult failed = new BackupRunCoordinator().runPlan(plan, "MANUAL", request -> request.withReader(failingAfter(request.getReader(), "blogPost", 4)));
assertThat(failed.getStatus()).isEqualTo("FAILED");
BackupRunResult resumed = new BackupRunCoordinator().resume(catalog.get(BackupsTables.RUN, failed.getRunId()).orElseThrow());
assertThat(resumed.isUsable()).as(resumed.getErrorMessage()).isTrue();
BackupManifest manifest = BackupManifest.read(BackupsTestUtils.BACKUPS.getBytes(resumed.getManifestKey()));
assertThat(manifest.table("blogPost").rowCount()).isEqualTo(7);
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);
}
}
@@ -0,0 +1,323 @@
/*
* 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.schedule;
import java.io.Serializable;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import com.kingsrook.qbits.backups.BackupsQBitConfig;
import com.kingsrook.qbits.backups.BackupsQBitProducer;
import com.kingsrook.qbits.backups.BackupsRuntime;
import com.kingsrook.qbits.backups.BackupsTestUtils;
import com.kingsrook.qbits.backups.LifecycleTestSupport.RecordingAlertSink;
import com.kingsrook.qbits.backups.model.BackupsTables;
import com.kingsrook.qbits.backups.plan.BackupValidatorReset;
import com.kingsrook.qbits.backups.run.BackupCatalog;
import com.kingsrook.qqq.backend.core.actions.processes.QProcessCallbackFactory;
import com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction;
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.context.QContext;
import com.kingsrook.qqq.backend.core.exceptions.QException;
import com.kingsrook.qqq.backend.core.instances.QInstanceValidator;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput;
import com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessOutput;
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.QueryInput;
import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
import com.kingsrook.qqq.backend.core.model.common.TimeZonePossibleValueSourceMetaDataProvider;
import com.kingsrook.qqq.backend.core.model.data.QRecord;
import com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType;
import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData;
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.scheduleing.simple.SimpleSchedulerMetaData;
import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
import com.kingsrook.qqq.backend.core.model.scheduledjobs.ScheduledJob;
import com.kingsrook.qqq.backend.core.model.scheduledjobs.ScheduledJobsMetaDataProvider;
import com.kingsrook.qqq.backend.core.model.session.QSession;
import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule;
import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore;
import com.kingsrook.qqq.backend.core.scheduler.QScheduleManager;
import org.junit.jupiter.api.AfterEach;
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;
/*******************************************************************************
** The plan schedule (spec 9.1): one scheduledJob per active plan, one more for the
** verify cron, kept in step as the plan changes, and the scheduled entry point that
** runs the engine as the system session after checking the plan owner.
*******************************************************************************/
class BackupPlanScheduleTest
{
private static final String BACKEND = "memory";
private final BackupCatalog catalog = new BackupCatalog();
private RecordingAlertSink alerts;
/*******************************************************************************
** An instance with the scheduled jobs tables, a scheduler, and the qbit configured
** with that scheduler.
*******************************************************************************/
@BeforeEach
void setUp() throws Exception
{
BackupsRuntime.reset();
MemoryRecordStore.getInstance().reset();
BackupsTestUtils.BACKUPS.clear();
BackupValidatorReset.reset();
QInstance qInstance = new QInstance();
qInstance.withInstanceDefaultAuthentication(new QAuthenticationMetaData().withType(QAuthenticationType.FULLY_ANONYMOUS).withName("anon"));
qInstance.addBackend(new QBackendMetaData().withName(BACKEND).withBackendType(MemoryBackendModule.class));
qInstance.addScheduler(new SimpleSchedulerMetaData().withName("simple"));
qInstance.addTable(new QTableMetaData().withName("author").withBackendName(BACKEND).withPrimaryKeyField("id")
.withField(new QFieldMetaData("id", QFieldType.INTEGER))
.withField(new QFieldMetaData("name", QFieldType.STRING)));
new ScheduledJobsMetaDataProvider().defineAll(qInstance, BACKEND, null);
qInstance.addPossibleValueSource(new TimeZonePossibleValueSourceMetaDataProvider().produce());
BackupsQBitConfig config = new BackupsQBitConfig().withBackendName(BACKEND).withInstanceName("testapp").withSchedulerName("simple")
.withStorageResolver(alias -> "backups".equals(alias) ? BackupsTestUtils.BACKUPS : null).withOpenWhenAnonymous(true);
new BackupsQBitProducer().withQBitConfig(config).withNamespace("backups").produce(qInstance);
new QInstanceValidator().validate(qInstance);
QContext.init(qInstance, new QSession());
QScheduleManager.initInstance(qInstance, QSystemUserSession::new);
alerts = new RecordingAlertSink();
BackupsRuntime.setAlertSink(alerts);
InsertInput authors = new InsertInput("author");
authors.setRecords(List.of(new QRecord().withValue("id", 1).withValue("name", "Ada")));
new InsertAction().execute(authors);
}
/*******************************************************************************
**
*******************************************************************************/
@AfterEach
void tearDown()
{
try
{
QScheduleManager.getInstance().stop();
QScheduleManager.getInstance().unInit();
}
catch(IllegalStateException e)
{
// not initialized
}
}
/*******************************************************************************
**
*******************************************************************************/
private Integer insertPlan(Map<String, Serializable> overrides) throws QException
{
Map<String, Serializable> values = new LinkedHashMap<>(Map.of("name", "Nightly", "planKey", "nightly", "isActive", true, "scopeType", "INSTANCE", "storageBucketAlias", "backups"));
values.putAll(overrides);
return (catalog.insert(BackupsTables.PLAN, values));
}
/*******************************************************************************
**
*******************************************************************************/
private List<ScheduledJob> jobs() throws QException
{
return (new QueryAction().execute(new QueryInput(ScheduledJob.TABLE_NAME).withIncludeAssociations(true)).getRecordEntities(ScheduledJob.class));
}
/*******************************************************************************
**
*******************************************************************************/
private static String param(ScheduledJob job, String key)
{
return (job.getJobParameters().stream().filter(p -> p.getKey().equals(key)).findFirst().map(p -> p.getValue()).orElse(null));
}
/*******************************************************************************
** An active plan with a cron gets one job; its verify cron gets a second; both are
** keyed to the plan, point at the right process, and follow the plan as it changes.
*******************************************************************************/
@Test
void testPlansOwnTheirScheduledJobs() throws Exception
{
Integer planId = insertPlan(Map.of("cronExpression", "0 0 2 * * ?", "cronTimeZoneId", "America/Chicago", "verifyCronExpression", "0 0 4 ? * SUN"));
List<ScheduledJob> jobs = jobs();
assertThat(jobs).hasSize(2);
ScheduledJob run = jobs.stream().filter(j -> BackupPlanSyncToScheduledJobProcess.FOREIGN_KEY_TYPE.equals(j.getForeignKeyType())).findFirst().orElseThrow();
ScheduledJob verify = jobs.stream().filter(j -> BackupPlanVerifySyncToScheduledJobProcess.FOREIGN_KEY_TYPE.equals(j.getForeignKeyType())).findFirst().orElseThrow();
assertThat(run.getForeignKeyType()).isEqualTo("backups_plan");
assertThat(run.getForeignKeyValue()).isEqualTo(String.valueOf(planId));
assertThat(run.getCronExpression()).isEqualTo("0 0 2 * * ?");
assertThat(run.getCronTimeZoneId()).isEqualTo("America/Chicago");
assertThat(run.getIsActive()).isTrue();
assertThat(run.getSchedulerName()).isEqualTo("simple");
assertThat(param(run, "processName")).isEqualTo(BackupScheduledRunProcess.NAME);
assertThat(param(run, "recordId")).isEqualTo(String.valueOf(planId));
assertThat(verify.getForeignKeyType()).isEqualTo("backups_plan_verify");
assertThat(verify.getForeignKeyValue()).isEqualTo(String.valueOf(planId));
assertThat(verify.getCronExpression()).isEqualTo("0 0 4 ? * SUN");
assertThat(param(verify, "processName")).isEqualTo(BackupPlanVerifySyncToScheduledJobProcess.VERIFY_PROCESS_NAME);
///////////////////////////////////
// a new cron follows the plan //
///////////////////////////////////
UpdateInput update = new UpdateInput(BackupsTables.PLAN);
update.setRecords(List.of(new QRecord().withValue("id", planId).withValue("cronExpression", "0 30 3 * * ?")));
update.withPermissionBypass("test");
new UpdateAction().execute(update);
assertThat(jobs()).hasSize(2).anyMatch(j -> "0 30 3 * * ?".equals(j.getCronExpression()));
////////////////////////////////////////////////////////////////
// an inactive plan keeps its jobs but switches them off //
////////////////////////////////////////////////////////////////
UpdateInput off = new UpdateInput(BackupsTables.PLAN);
off.setRecords(List.of(new QRecord().withValue("id", planId).withValue("isActive", false)));
off.withPermissionBypass("test");
new UpdateAction().execute(off);
assertThat(jobs()).hasSize(2).allSatisfy(j -> assertThat(j.getIsActive()).isFalse());
UpdateInput on = new UpdateInput(BackupsTables.PLAN);
on.setRecords(List.of(new QRecord().withValue("id", planId).withValue("isActive", true)));
on.withPermissionBypass("test");
new UpdateAction().execute(on);
assertThat(jobs()).hasSize(2).allSatisfy(j -> assertThat(j.getIsActive()).isTrue());
///////////////////////////////////////////////
// deleting the plan deletes its jobs //
///////////////////////////////////////////////
DeleteInput delete = new DeleteInput(BackupsTables.PLAN);
delete.setPrimaryKeys(new java.util.ArrayList<>(List.of(planId)));
delete.withPermissionBypass("test");
new DeleteAction().execute(delete);
assertThat(jobs()).isEmpty();
}
/*******************************************************************************
** A plan with no cron is manual only: no job. A verify cron alone makes only the
** verify job.
*******************************************************************************/
@Test
void testManualOnlyPlansHaveNoJob() throws Exception
{
insertPlan(Map.of());
assertThat(jobs()).isEmpty();
insertPlan(Map.of("name", "Verify only", "planKey", "verify-only", "verifyCronExpression", "0 0 4 ? * SUN"));
assertThat(jobs()).singleElement().satisfies(j -> assertThat(j.getForeignKeyType()).isEqualTo("backups_plan_verify"));
insertPlan(Map.of("name", "Draft", "planKey", "draft", "isActive", false, "cronExpression", "0 0 2 * * ?"));
assertThat(jobs()).as("an inactive plan gets no new job").hasSize(1);
}
/*******************************************************************************
** Run the scheduled entry point for a plan, as the scheduler would.
*******************************************************************************/
private RunProcessOutput runScheduled(Integer planId) throws QException
{
RunProcessInput input = new RunProcessInput();
input.setProcessName(BackupScheduledRunProcess.NAME);
input.setCallback(QProcessCallbackFactory.forPrimaryKey("id", planId));
input.setFrontendStepBehavior(RunProcessInput.FrontendStepBehavior.SKIP);
return (new RunProcessAction().execute(input));
}
/*******************************************************************************
** The scheduled entry point runs the engine as the system scheduler session: a
** full-instance run, requested by system:scheduler, triggered SCHEDULED.
*******************************************************************************/
@Test
void testScheduledEntryPointRunsAsTheSystem() throws Exception
{
Integer planId = insertPlan(Map.of("cronExpression", "0 0 2 * * ?", "cronTimeZoneId", "UTC", "ownerUserId", "jmaes"));
RunProcessOutput output = runScheduled(planId);
assertThat(output.getException()).isEmpty();
assertThat(output.getValues().get("status")).isIn("SUCCEEDED", "SUCCEEDED_WITH_WARNINGS");
QRecord run = catalog.where(BackupsTables.RUN, "backupPlanId", planId).get(0);
assertThat(run.getValueString("trigger")).isEqualTo("SCHEDULED");
assertThat(run.getValueString("requestedByUserId")).isEqualTo("system:scheduler");
assertThat(run.getValueString("securityScope")).isEqualTo("FULL");
assertThat(run.getValueString("manifestKey")).isNotNull();
assertThat(BackupsTestUtils.BACKUPS.getBytes(run.getValueString("manifestKey"))).isNotNull();
assertThat(catalog.get(BackupsTables.PLAN, planId).orElseThrow().getValueInteger("lastSuccessRunId")).isEqualTo(run.getValueInteger("id"));
}
/*******************************************************************************
** A plan whose owner is gone does not run silently: the plan says OWNER_INVALID,
** a warning is raised, and nothing is written. An inactive plan does nothing.
*******************************************************************************/
@Test
void testScheduledRunRefusesAnInvalidOwner() throws Exception
{
Integer planId = insertPlan(Map.of("cronExpression", "0 0 2 * * ?", "cronTimeZoneId", "UTC", "ownerUserId", "departed"));
BackupScheduledRunProcess.setOwnerHasAdminCheck(owner -> !"departed".equals(owner));
assertThatThrownBy(() -> runScheduled(planId)).hasMessageContaining("did not run").hasMessageContaining("departed");
assertThat(catalog.where(BackupsTables.RUN, "backupPlanId", planId)).isEmpty();
assertThat(BackupsTestUtils.BACKUPS.list("")).isEmpty();
assertThat(catalog.get(BackupsTables.PLAN, planId).orElseThrow().getValueString("slaStatus")).isEqualTo("OWNER_INVALID");
assertThat(alerts.raised).containsExactly("backups.plan.owner_invalid|backups.plan.owner_invalid:nightly");
BackupScheduledRunProcess.setOwnerHasAdminCheck(null);
Integer noOwner = insertPlan(Map.of("name", "Ownerless", "planKey", "ownerless", "isActive", true));
assertThatThrownBy(() -> runScheduled(noOwner)).hasMessageContaining("the plan has no owner");
UpdateInput off = new UpdateInput(BackupsTables.PLAN);
off.setRecords(List.of(new QRecord().withValue("id", planId).withValue("isActive", false)));
off.withPermissionBypass("test");
new UpdateAction().execute(off);
RunProcessOutput inactive = runScheduled(planId);
assertThat(inactive.getValues().get("status")).isEqualTo("INACTIVE");
assertThat(catalog.where(BackupsTables.RUN, "backupPlanId", planId)).isEmpty();
}
}
@@ -365,7 +365,11 @@ class SlaCheckerTest
{
plan(Map.of());
plan(Map.of("name", "Off", "planKey", "off", "isActive", false));
// the bucket existed when the plan was saved (plan save validates it), and was removed from the config later
var originalResolver = com.kingsrook.qbits.backups.BackupsRuntime.getConfig().getStorageResolver();
com.kingsrook.qbits.backups.BackupsRuntime.getConfig().withStorageResolver(alias -> "no-such-bucket".equals(alias) ? com.kingsrook.qbits.backups.BackupsTestUtils.BACKUPS : originalResolver.apply(alias));
plan(Map.of("name", "Bad", "planKey", "bad", "storageBucketAlias", "no-such-bucket"));
com.kingsrook.qbits.backups.BackupsRuntime.getConfig().withStorageResolver(originalResolver);
clock.advance(Duration.ofHours(30));
List<SlaChecker.PlanResult> results = new SlaChecker().checkAll();